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

@@ -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)

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

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 <session_id>`. 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)

agentd/executors/codex.py

@@ -0,0 +1,23 @@
+"""Codex CLI: `codex exec --json ... -o last.md -` (stdin prompt), resume with `codex exec resume <thread>`.
+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)

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))

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)<li[^>]*>", "- ", h or "")
+    h = re.sub(r"(?i)<br\s*/?>|</p>|</div>|</li>|</pre>|</h\d>|</tr>", "\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 = "<p>%s</p>" % html.escape(text.strip(), quote=False).replace("\n", "<br>")
+    if pre:
+        out += "<pre>%s</pre>" % 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 <pre>."""
+    return text if text.lstrip().startswith("<") else "<pre>%s</pre>" % 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})

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)

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)

agentd/transports/warpgate_ssh.py

@@ -0,0 +1,140 @@
+"""Windows terminal behind Warpgate, reached through the server-access relay helper scripts:
+  win10-ps.sh  <script>          run PowerShell (sent as -EncodedCommand)
+  win10-put.sh <local> <remote>  upload a file (gzip+base64 chunks, SHA-256 verified)
+Runs are launched as one-shot scheduled tasks (schtasks) so they outlive the SSH session.
+Note: the helper scripts currently hard-code `--target win10`; `target` is checked against that."""
+import base64
+import hashlib
+import os
+import re
+import subprocess
+import tempfile
+
+from . import TransportError
+
+RUNNER = r"""$ErrorActionPreference = 'Continue'
+[Console]::OutputEncoding = [Text.Encoding]::UTF8
+$OutputEncoding = [Text.Encoding]::UTF8
+Set-Location {cwd}
+Set-Content -Path '{d}\pid.txt' -Value $PID
+Set-Content -Path '{d}\started.txt' -Value (Get-Date).ToString('s')
+$prompt = Get-Content -Raw -Encoding UTF8 '{d}\prompt.md'
+$prompt | {command} 2> '{d}\stderr.txt' | ForEach-Object {{ Add-Content -Path '{d}\{stdout}' -Value $_ -Encoding UTF8 }}
+Set-Content -Path '{d}\exit.txt' -Value ("exit=" + $LASTEXITCODE + " at " + (Get-Date).ToString('s'))
+"""
+
+
+def _b64_payload(out, marker):
+    m = re.search(marker + r"=([A-Za-z0-9+/=]*)", out)
+    if not m:
+        return None
+    return base64.b64decode(m.group(1)).decode("utf-8", "replace") if m.group(1) else ""
+
+
+class WarpgateSshTransport:
+    os = "windows"
+
+    def __init__(self, cfg, terminal):
+        self.cfg, self.terminal = cfg, terminal
+        t = cfg.terminals[terminal]
+        self.target = t.get("target", terminal)
+        helpers = cfg.helpers_dir(terminal)
+        self.ps_script, self.put_script = str(helpers / "win10-ps.sh"), str(helpers / "win10-put.sh")
+        self.repo = t["repo"]
+
+    def default_run_root(self):
+        return self.join(self.repo, "runtime", "agentd")
+
+    def join(self, *parts):
+        return "\\".join(p.rstrip("\\") for p in parts)
+
+    def quote(self, s):
+        return "'" + str(s).replace("'", "''") + "'"
+
+    # ---------------------------------------------------------------- relay primitives
+    def ps(self, script, timeout=150):
+        env = dict(os.environ, WIN10_TIMEOUT=str(timeout))
+        try:
+            out = subprocess.run([self.ps_script, script], capture_output=True, text=True, timeout=timeout + 60, env=env)
+        except subprocess.TimeoutExpired:
+            raise TransportError("relay timeout after %ss" % timeout) from None
+        if out.returncode != 0 and not out.stdout.strip():
+            raise TransportError("relay exit %s: %s" % (out.returncode, out.stderr.strip()[-300:]))
+        return out.stdout.replace("\r", "")
+
+    def put(self, local, remote):
+        sha = hashlib.sha256(open(local, "rb").read()).hexdigest()
+        out = subprocess.run([self.put_script, local, remote], capture_output=True, text=True, timeout=600)
+        if ("remote sha256 " + sha) not in out.stdout:
+            raise TransportError("upload of %s failed: %s" % (remote, (out.stdout + out.stderr).strip()[-300:]))
+
+    # ---------------------------------------------------------------- file ops
+    def mkdir(self, path, clean=False):
+        q = self.quote(path)
+        script = "New-Item -ItemType Directory -Force -Path %s | Out-Null\n" % q
+        if clean:
+            script += "Get-ChildItem -LiteralPath %s -File | Remove-Item -Force\n" % q
+        self.ps(script + "'MKDIR=ok'")
+
+    def write_text(self, path, text):
+        fd, tmp = tempfile.mkstemp(prefix="agentd-", suffix=".txt")
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8", newline="") as fh:
+                fh.write(text)
+            self.put(tmp, path)
+        finally:
+            os.unlink(tmp)
+
+    def read_text(self, path):
+        q = self.quote(path)
+        out = self.ps("if (Test-Path -LiteralPath %s) { 'B64=' + [Convert]::ToBase64String([IO.File]::ReadAllBytes(%s)) }"
+                      " else { 'MISSING' }" % (q, q), timeout=200)
+        if "MISSING" in out and "B64=" not in out:
+            return None
+        text = _b64_payload(out, "B64")
+        if text is None:
+            raise TransportError("read %s: unexpected relay output" % path)
+        return text.lstrip("")
+
+    def grep_first(self, path, needle):
+        q = self.quote(path)
+        out = self.ps("$m = if (Test-Path -LiteralPath %s) { Select-String -LiteralPath %s -SimpleMatch -Pattern %s | Select-Object -First 1 }\n"
+                      "'B64=' + $(if ($m) { [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($m.Line)) } else { '' })"
+                      % (q, q, self.quote(needle)))
+        return _b64_payload(out, "B64") or None
+
+    def tail(self, path, n):
+        q = self.quote(path)
+        out = self.ps("$t = if (Test-Path -LiteralPath %s) { (Get-Content -LiteralPath %s -Encoding UTF8 -Tail %d) -join \"`n\" } else { '' }\n"
+                      "'B64=' + [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes([string]$t))" % (q, q, n))
+        return _b64_payload(out, "B64") or ""
+
+    # ---------------------------------------------------------------- processes
+    def runner(self, run_dir, cwd, command, stdout_name):
+        return "run.ps1", RUNNER.format(cwd=self.quote(cwd), d=run_dir.replace("'", "''"), command=command, stdout=stdout_name)
+
+    def launch(self, task, script_path, run_dir):
+        out = self.ps("schtasks /Delete /F /TN '%(t)s' 2>$null | Out-Null\n"
+                      "$tr = 'powershell.exe -NoProfile -NonInteractive -ExecutionPolicy Bypass -WindowStyle Hidden -File \"%(s)s\"'\n"
+                      "schtasks /Create /F /TN '%(t)s' /SC ONCE /ST 00:00 /TR $tr | Out-Null\n"
+                      "schtasks /Run /TN '%(t)s' | Out-Null\n"
+                      "'LAUNCHED=' + $LASTEXITCODE" % dict(t=task, s=script_path))
+        if "LAUNCHED=0" not in out:
+            raise TransportError("schtasks launch of %s failed: %s" % (task, out.strip()[-300:]))
+
+    def kill(self, task, run_dir):
+        self.ps("$p = Join-Path %s 'pid.txt'\n"
+                "if (Test-Path $p) { taskkill /T /F /PID (Get-Content $p | Select-Object -First 1) 2>&1 | Out-Null }\n"
+                "schtasks /End /TN '%s' 2>&1 | Out-Null\n"
+                "schtasks /Delete /F /TN '%s' 2>&1 | Out-Null\n'KILLED=ok'" % (self.quote(run_dir), task, task))
+
+    def forget(self, task):
+        self.ps("schtasks /Delete /F /TN '%s' 2>&1 | Out-Null; 'FORGOT=ok'" % task)
+
+    def process_count(self, name):
+        try:
+            out = self.ps("'COUNT=' + @(Get-Process -Name %s -ErrorAction SilentlyContinue).Count" % self.quote(name))
+        except TransportError:
+            return None
+        m = re.search(r"COUNT=(\d+)", out)
+        return int(m.group(1)) if m else None