From 9535122c9972c57e17129ef97f45d1697a91a8f7 Mon Sep 17 00:00:00 2001 From: "jj.wang" Date: Wed, 21 Oct 2026 07:04:32 +0000 Subject: [PATCH] Plane client, transports and executor adapters Plane: paginated issue/comment/state/label/member access, one client per terminal identity. Transports: local (setsid-detached runner) and warpgate-ssh (win10-ps.sh / win10-put.sh relay, schtasks launch, taskkill on timeout). Executors share a runner shape (prompt.md, exit.txt marker) with codex, claude-code and shell adapters for last message and thread id. --- agentd/executors/__init__.py | 16 ++++ agentd/executors/base.py | 121 ++++++++++++++++++++++++++ agentd/executors/claude_code.py | 38 ++++++++ agentd/executors/codex.py | 23 +++++ agentd/executors/shell.py | 19 ++++ agentd/plane.py | 131 ++++++++++++++++++++++++++++ agentd/transports/__init__.py | 24 +++++ agentd/transports/local.py | 78 +++++++++++++++++ agentd/transports/warpgate_ssh.py | 140 ++++++++++++++++++++++++++++++ 9 files changed, 590 insertions(+) create mode 100644 agentd/executors/__init__.py create mode 100644 agentd/executors/base.py create mode 100644 agentd/executors/claude_code.py create mode 100644 agentd/executors/codex.py create mode 100644 agentd/executors/shell.py create mode 100644 agentd/plane.py create mode 100644 agentd/transports/__init__.py create mode 100644 agentd/transports/local.py create mode 100644 agentd/transports/warpgate_ssh.py diff --git a/agentd/executors/__init__.py b/agentd/executors/__init__.py new file mode 100644 index 0000000..6395c6d --- /dev/null +++ b/agentd/executors/__init__.py @@ -0,0 +1,16 @@ +"""Executor adapters. `make_executor` picks the adapter class by the executor's `adapter` (default: its name).""" +from .base import ExecResult, Executor, Run # noqa: F401 + + +def make_executor(cfg, terminal, name, transport): + ecfg = cfg.executors[name] + adapter = ecfg.get("adapter", name) + if adapter == "codex": + from .codex import CodexExecutor as cls + elif adapter == "claude-code": + from .claude_code import ClaudeCodeExecutor as cls + elif adapter == "shell": + from .shell import ShellExecutor as cls + else: + raise ValueError("no executor adapter %s" % adapter) + return cls(name, ecfg, terminal, cfg.terminals[terminal], transport) diff --git a/agentd/executors/base.py b/agentd/executors/base.py new file mode 100644 index 0000000..ca9dc9b --- /dev/null +++ b/agentd/executors/base.py @@ -0,0 +1,121 @@ +"""Executor contract (docs/executor-contract.md). + +agentd is not resident: a run outlives the process that started it, so `start` returns a JSON-serialisable +handle that is stored in the state file, and `poll`/`collect`/`kill` work from that handle alone.""" +import re +from dataclasses import dataclass, field +from typing import List, Optional + + +@dataclass +class Run: + run_id: str + repo: str + prompt: str + model: Optional[str] = None + effort: Optional[str] = None + tier: Optional[str] = None + thread_id: Optional[str] = None # set => resume this thread + script: Optional[str] = None # shell executor + + +@dataclass +class ExecResult: + last_message: str + exit_code: Optional[int] + thread_id: Optional[str] + log_paths: List[str] = field(default_factory=list) + tail: str = "" # last frame: stderr tail / last event, for failure notes + + +class Executor: + """Abstract adapter. `resume` says whether the executor can continue a thread.""" + resume = False + + def __init__(self, name, cfg, terminal, tcfg, transport): + self.name, self.cfg, self.terminal, self.tcfg, self.transport = name, cfg, terminal, tcfg, transport + + def start(self, run): # -> handle (dict) + raise NotImplementedError + + def poll(self, handle): # -> "running" | "finished" + raise NotImplementedError + + def collect(self, handle): # -> ExecResult + raise NotImplementedError + + def kill(self, handle): + raise NotImplementedError + + +def fill(template, params): + """Replace {name} placeholders that exist in params; leave anything else untouched.""" + return re.sub(r"\{(\w+)\}", lambda m: str(params[m.group(1)]) if m.group(1) in params else m.group(0), template) + + +class RunnerExecutor(Executor): + """Common shape for CLI executors: write prompt.md + a runner script into a run directory on the terminal, + launch it detached through the transport, and treat `exit.txt` as the completion marker. + Subclasses define `stdout_name` and `collect`.""" + stdout_name = "stdout.txt" + + @property + def resume(self): + return isinstance(self.cfg.get("resume"), str) and bool(self.cfg["resume"]) + + def run_root(self): + return self.tcfg.get("run_root") or self.transport.default_run_root() + + def params(self, run, run_dir): + tier = fill(self.cfg.get("tier_arg", ""), {"tier": run.tier}) if run.tier else "" + return {"repo": run.repo, "model": run.model or "", "effort": run.effort or "", "tier": tier, + "out": self.transport.join(run_dir, "last.md"), "thread": run.thread_id or "", + "perm": self.cfg.get("permission_mode", "default"), "run_dir": run_dir, + "script": run.script or self.cfg.get("script", "")} + + def command(self, run, run_dir): + template = self.cfg["resume"] if run.thread_id and self.resume else self.cfg["run"] + cmd = " ".join(fill(template, self.params(run, run_dir)).split()) + program, _, rest = cmd.partition(" ") + binary = self.tcfg.get("bins", {}).get(program) + if binary: + program = self.transport.quote(binary) + if self.transport.os == "windows" and (binary or program.startswith("'")): + program = "& " + program + return (program + " " + rest).strip() + + def start(self, run): + t = self.transport + run_dir = t.join(self.run_root(), run.run_id) + t.mkdir(run_dir, clean=True) + t.write_text(t.join(run_dir, "prompt.md"), run.prompt) + script_name, script = t.runner(run_dir, run.repo, self.command(run, run_dir), self.stdout_name) + script_path = t.join(run_dir, script_name) + t.write_text(script_path, script) + task = "agentd-" + run.run_id + t.launch(task, script_path, run_dir) + return {"run_dir": run_dir, "task": task, "thread_id": run.thread_id} + + def poll(self, handle): + text = self.transport.read_text(self.transport.join(handle["run_dir"], "exit.txt")) + return "finished" if text and "exit=" in text else "running" + + def exit_code(self, handle): + text = self.transport.read_text(self.transport.join(handle["run_dir"], "exit.txt")) or "" + m = re.search(r"exit=(-?\d+)", text) + return int(m.group(1)) if m else None + + def kill(self, handle): + self.transport.kill(handle["task"], handle["run_dir"]) + + def logs(self, handle, *names): + return [self.transport.join(handle["run_dir"], n) for n in names] + + def stderr_tail(self, handle, n=5): + return self.transport.tail(self.transport.join(handle["run_dir"], "stderr.txt"), n) + + def forget(self, handle): + try: + self.transport.forget(handle["task"]) + except Exception: + pass diff --git a/agentd/executors/claude_code.py b/agentd/executors/claude_code.py new file mode 100644 index 0000000..b4ec19c --- /dev/null +++ b/agentd/executors/claude_code.py @@ -0,0 +1,38 @@ +"""Claude Code CLI: `claude -p --output-format json` with the prompt on stdin and cwd = repo; +resume with `--resume `. The JSON result carries `result` (last message) and `session_id`.""" +import json + +from .base import ExecResult, RunnerExecutor + + +def parse_output(text): + """Return the result object from `--output-format json` output (whole text or its last JSON line).""" + text = (text or "").strip() + candidates = [text] + list(reversed(text.splitlines())) + for c in candidates: + try: + obj = json.loads(c) + except ValueError: + continue + if isinstance(obj, dict): + return obj + if isinstance(obj, list): # stream-json style array: take the result event + res = [o for o in obj if isinstance(o, dict) and o.get("type") == "result"] + if res: + return res[-1] + return {} + + +class ClaudeCodeExecutor(RunnerExecutor): + stdout_name = "out.json" + + def collect(self, handle): + t, d = self.transport, handle["run_dir"] + obj = parse_output(t.read_text(t.join(d, "out.json"))) + err = bool(obj.get("is_error")) or obj.get("subtype", "success") != "success" + last = "" if err else (obj.get("result") or "") + tail = self.stderr_tail(handle) + if err: + tail = ("%s %s" % (obj.get("subtype", ""), str(obj.get("result", ""))[:400]) + "\n" + tail).strip() + return ExecResult(last.strip(), self.exit_code(handle), obj.get("session_id") or handle.get("thread_id"), + self.logs(handle, "out.json", "stderr.txt"), tail) diff --git a/agentd/executors/codex.py b/agentd/executors/codex.py new file mode 100644 index 0000000..2a43962 --- /dev/null +++ b/agentd/executors/codex.py @@ -0,0 +1,23 @@ +"""Codex CLI: `codex exec --json ... -o last.md -` (stdin prompt), resume with `codex exec resume `. +Thread id comes from the `thread.started` event in events.jsonl; the last message from last.md.""" +import json + +from .base import ExecResult, RunnerExecutor + + +class CodexExecutor(RunnerExecutor): + stdout_name = "events.jsonl" + + def collect(self, handle): + t, d = self.transport, handle["run_dir"] + last = t.read_text(t.join(d, "last.md")) or "" + thread = handle.get("thread_id") + line = t.grep_first(t.join(d, "events.jsonl"), "thread.started") + if line: + try: + thread = json.loads(line).get("thread_id") or thread + except ValueError: + pass + tail = (self.stderr_tail(handle) + "\n" + t.tail(t.join(d, "events.jsonl"), 1)[-600:]).strip() + return ExecResult(last.strip(), self.exit_code(handle), thread, + self.logs(handle, "events.jsonl", "stderr.txt", "last.md"), tail) diff --git a/agentd/executors/shell.py b/agentd/executors/shell.py new file mode 100644 index 0000000..d8eca52 --- /dev/null +++ b/agentd/executors/shell.py @@ -0,0 +1,19 @@ +"""Shell/script executor: runs `{script}` with the prompt on stdin. The last message is `last.md` in the run +directory if the script wrote it (its path is passed as {out}), else the tail of stdout. No threads.""" +from .base import ExecResult, RunnerExecutor + + +class ShellExecutor(RunnerExecutor): + stdout_name = "stdout.txt" + + @property + def resume(self): + return False + + def collect(self, handle): + t, d = self.transport, handle["run_dir"] + last = t.read_text(t.join(d, "last.md")) + if last is None: + last = (t.read_text(t.join(d, "stdout.txt")) or "")[-6000:] + return ExecResult(last.strip(), self.exit_code(handle), None, + self.logs(handle, "stdout.txt", "stderr.txt"), self.stderr_tail(handle)) diff --git a/agentd/plane.py b/agentd/plane.py new file mode 100644 index 0000000..cc19146 --- /dev/null +++ b/agentd/plane.py @@ -0,0 +1,131 @@ +"""Minimal Plane REST client (API v1, header X-API-Key) plus HTML<->text helpers. + +One client per identity: agentd posts with the token of the terminal that executes the run.""" +import html +import json +import re +import urllib.error +import urllib.request + + +class PlaneError(Exception): + pass + + +def html_to_text(h): + h = re.sub(r"(?i)]*>", "- ", h or "") + h = re.sub(r"(?i)|

|||||", "\n", h) + text = html.unescape(re.sub(r"<[^>]+>", "", h)) + text = re.sub(r"[ \t]+\n", "\n", text) + return re.sub(r"\n{3,}", "\n\n", text).strip() + + +def text_to_html(text, pre=None): + out = "

%s

" % html.escape(text.strip(), quote=False).replace("\n", "
") + if pre: + out += "
%s
" % html.escape(pre, quote=False) + return out + + +def brief_to_html(text): + """Work-item description from a brief file: raw HTML passes through, anything else goes in 
."""
+    return text if text.lstrip().startswith("<") else "
%s
" % html.escape(text, quote=False)
+
+
+class Plane:
+    def __init__(self, base_url, workspace, token, timeout=60):
+        if not token:
+            raise PlaneError("no Plane token for this identity (is the vault key injected?)")
+        self.root = "%s/api/v1/workspaces/%s/" % (base_url.rstrip("/"), workspace)
+        self._token = token
+        self.timeout = timeout
+
+    def __repr__(self):
+        return "Plane(%s)" % self.root
+
+    def request(self, method, path, body=None):
+        url = path if path.startswith("http") else self.root + path
+        data = json.dumps(body).encode() if body is not None else None
+        req = urllib.request.Request(url, data=data, method=method)
+        req.add_header("X-API-Key", self._token)
+        req.add_header("Content-Type", "application/json")
+        try:
+            with urllib.request.urlopen(req, timeout=self.timeout) as resp:
+                raw = resp.read().decode()
+        except urllib.error.HTTPError as exc:
+            detail = exc.read().decode(errors="replace")[:300]
+            raise PlaneError("%s %s -> HTTP %s %s" % (method, path, exc.code, detail)) from None
+        except urllib.error.URLError as exc:
+            raise PlaneError("%s %s -> %s" % (method, path, exc.reason)) from None
+        return json.loads(raw) if raw else None
+
+    def paged(self, path):
+        """Follow Plane cursor pagination and return every result."""
+        out, cursor = [], None
+        while True:
+            sep = "&" if "?" in path else "?"
+            r = self.request("GET", path + sep + "per_page=100" + ("&cursor=" + cursor if cursor else ""))
+            if isinstance(r, list):
+                return r
+            out += r.get("results", [])
+            if not r.get("next_page_results") or not r.get("next_cursor"):
+                return out
+            cursor = r["next_cursor"]
+
+    # ---------------------------------------------------------------- lookups (name -> id)
+    def states(self, project):
+        return {s["name"]: s["id"] for s in self.paged("projects/%s/states/" % project)}
+
+    def labels(self, project):
+        return {l["name"]: l["id"] for l in self.paged("projects/%s/labels/" % project)}
+
+    def members(self):
+        m = self.request("GET", "members/")
+        if isinstance(m, dict):
+            m = m.get("results", [])
+        return {x["display_name"]: x["id"] for x in m if x.get("display_name")}
+
+    def modules(self, project):
+        return {m["name"]: m["id"] for m in self.paged("projects/%s/modules/" % project)}
+
+    # ---------------------------------------------------------------- work items
+    def list_issues(self, project):
+        return self.paged("projects/%s/issues/" % project)
+
+    def get_issue(self, project, issue):
+        return self.request("GET", "projects/%s/issues/%s/" % (project, issue))
+
+    def find_issue(self, project, seq):
+        for i in self.list_issues(project):
+            if i.get("sequence_id") == seq:
+                return i
+        raise PlaneError("no work item with sequence %s" % seq)
+
+    def create_issue(self, project, body):
+        return self.request("POST", "projects/%s/issues/" % project, body)
+
+    def patch_issue(self, project, issue, **fields):
+        return self.request("PATCH", "projects/%s/issues/%s/" % (project, issue), fields)
+
+    def set_state(self, project, issue, state_id, assignee_ids=None):
+        fields = {"state": state_id}
+        if assignee_ids is not None:
+            fields["assignees"] = assignee_ids
+        return self.patch_issue(project, issue, **fields)
+
+    def set_assignees(self, project, issue, assignee_ids):
+        return self.patch_issue(project, issue, assignees=assignee_ids)
+
+    def add_to_module(self, project, module, issue):
+        return self.request("POST", "projects/%s/modules/%s/module-issues/" % (project, module), {"issues": [issue]})
+
+    def create_module(self, project, name):
+        return self.request("POST", "projects/%s/modules/" % project, {"name": name})
+
+    # ---------------------------------------------------------------- comments
+    def list_comments(self, project, issue):
+        cs = self.paged("projects/%s/issues/%s/comments/" % (project, issue))
+        return sorted(cs, key=lambda c: c.get("created_at", ""))
+
+    def post_comment(self, project, issue, comment_html):
+        return self.request("POST", "projects/%s/issues/%s/comments/" % (project, issue), {"comment_html": comment_html})
diff --git a/agentd/transports/__init__.py b/agentd/transports/__init__.py
new file mode 100644
index 0000000..908257a
--- /dev/null
+++ b/agentd/transports/__init__.py
@@ -0,0 +1,24 @@
+"""Transports: how agentd reaches a terminal's machine (files, detached launch, kill).
+
+Interface (duck-typed, see docs/executor-contract.md):
+  os, join(*parts), quote(s), default_run_root()
+  mkdir(path, clean=False), write_text(path, text), read_text(path) -> str|None,
+  grep_first(path, needle) -> str|None, tail(path, n) -> str
+  runner(run_dir, cwd, command, stdout_name) -> (script_name, script_text)
+  launch(task, script_path, run_dir), kill(task, run_dir), forget(task)
+  process_count(name) -> int|None (None = unreachable)"""
+
+
+class TransportError(Exception):
+    pass
+
+
+def make_transport(cfg, terminal):
+    kind = cfg.terminals[terminal]["transport"]
+    if kind == "local":
+        from .local import LocalTransport
+        return LocalTransport(cfg, terminal)
+    if kind == "warpgate-ssh":
+        from .warpgate_ssh import WarpgateSshTransport
+        return WarpgateSshTransport(cfg, terminal)
+    raise TransportError("unknown transport %s" % kind)
diff --git a/agentd/transports/local.py b/agentd/transports/local.py
new file mode 100644
index 0000000..9009abd
--- /dev/null
+++ b/agentd/transports/local.py
@@ -0,0 +1,78 @@
+"""Local Linux transport: plain files plus a detached `bash run.sh` in its own session (setsid)."""
+import os
+import signal
+import subprocess
+from pathlib import Path
+
+RUNNER = """#!/usr/bin/env bash
+cd {cwd} || {{ echo "exit=97 at $(date -Is) (cd failed)" > {d}/exit.txt; exit 97; }}
+echo $$ > {d}/pid.txt
+date -Is > {d}/started.txt
+(
+{command}
+) < {d}/prompt.md > {d}/{stdout} 2> {d}/stderr.txt
+echo "exit=$? at $(date -Is)" > {d}/exit.txt
+"""
+
+
+class LocalTransport:
+    os = "linux"
+
+    def __init__(self, cfg, terminal):
+        self.cfg, self.terminal = cfg, terminal
+
+    def default_run_root(self):
+        return str(self.cfg.state_dir / "runs")
+
+    def join(self, *parts):
+        return os.path.join(*parts)
+
+    def quote(self, s):
+        return "'" + str(s).replace("'", "'\\''") + "'"
+
+    def mkdir(self, path, clean=False):
+        p = Path(path)
+        p.mkdir(parents=True, exist_ok=True)
+        if clean:
+            for f in p.iterdir():
+                if f.is_file():
+                    f.unlink()
+
+    def write_text(self, path, text):
+        Path(path).write_text(text, encoding="utf-8")
+
+    def read_text(self, path):
+        p = Path(path)
+        return p.read_text(encoding="utf-8", errors="replace") if p.exists() else None
+
+    def grep_first(self, path, needle):
+        text = self.read_text(path) or ""
+        return next((line for line in text.splitlines() if needle in line), None)
+
+    def tail(self, path, n):
+        text = self.read_text(path) or ""
+        return "\n".join(text.splitlines()[-n:])
+
+    def runner(self, run_dir, cwd, command, stdout_name):
+        return "run.sh", RUNNER.format(cwd=self.quote(cwd), d=self.quote(run_dir), command=command, stdout=stdout_name)
+
+    def launch(self, task, script_path, run_dir):
+        # `setsid -f` double-forks: the runner leads its own session (pgid == pid, see kill) and is reparented
+        # to init, so a long-running `tick --loop` never accumulates zombies.
+        subprocess.run(["setsid", "-f", "bash", script_path], cwd=run_dir, check=True, close_fds=True,
+                       stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+
+    def kill(self, task, run_dir):
+        pid = (self.read_text(os.path.join(run_dir, "pid.txt")) or "").strip()
+        if pid.isdigit():
+            try:
+                os.killpg(int(pid), signal.SIGTERM)  # runner is a session leader: pgid == pid
+            except ProcessLookupError:
+                pass
+
+    def forget(self, task):
+        pass
+
+    def process_count(self, name):
+        out = subprocess.run(["pgrep", "-c", "-x", name], capture_output=True, text=True)
+        return int(out.stdout.strip() or 0)
diff --git a/agentd/transports/warpgate_ssh.py b/agentd/transports/warpgate_ssh.py
new file mode 100644
index 0000000..e1fedcb
--- /dev/null
+++ b/agentd/transports/warpgate_ssh.py
@@ -0,0 +1,140 @@
+"""Windows terminal behind Warpgate, reached through the server-access relay helper scripts:
+  win10-ps.sh