feat(db,control): dormant headless-runner schema, stream parser, and fixtures (phase 1)

Groundwork for replacing tmux-TUI agent runs with worker-owned
'claude -p --output-format stream-json' subprocesses (Postgres as the
single source of truth; no shared files between workers):

- migration 0008: workers (heartbeat registry), agent_runs (one row per
  headless invocation), agent_events (persisted stream-json event log),
  session_archives (tar.gz'd claude session per agent for cross-worker
  --resume); agents gains session_id/worker_id, commands gains
  target_worker; 'crashed' joins the agent status vocabulary
- control/headless.py: munged-path helper, argv builders, tolerant
  stream parser, archive/materialize round-trip, RunSupervisor + launch
  (nothing calls it yet - runner config still defaults to tmux)
- repository: run/event/archive/worker accessors; claim_next_command
  learns target_worker pinning and type exclusion for full slots
- tests/fixtures/fake_claude.py: scripted stream-json stand-in binary
- scripts/validate_claude_headless.sh: manual real-binary validation
  checklist (resume-on-clean-HOME linchpin, hook behavior under -p)

No behavior change; suite 245 -> 270 green.

scripts/validate_claude_headless.sh

@@ -0,0 +1,138 @@
+#!/usr/bin/env bash
+# Phase-1 validation of the real `claude` binary's headless behavior (plan §0).
+# Run MANUALLY inside the control container (needs real claude + credentials); findings
+# go in the phase-1 PR description. Never wired into CI — this probes the real binary.
+#
+#   CLAUDE_BIN=claude ./scripts/validate_claude_headless.sh /tmp/headless-validation
+#
+# Each check prints PASS/FAIL/INFO; the script keeps going so one failure doesn't hide
+# the rest. Items map to the plan's "must validate" list:
+#   1. --verbose requirement with -p --output-format stream-json
+#   2. exact stream-json event shapes for this CLI version
+#   3. --resume from a transcript materialized onto a clean $HOME  (THE LINCHPIN)
+#   4. PreToolUse deny behavior under -p
+#   5. Stop-hook decision:block behavior under -p
+#   6. -p exit-code semantics + credential env var name
+
+set -uo pipefail
+
+CLAUDE_BIN="${CLAUDE_BIN:-claude}"
+WORK="${1:-/tmp/headless-validation}"
+SID="$(python3 -c 'import uuid; print(uuid.uuid4())')"
+
+mkdir -p "$WORK/repo" "$WORK/home-a" "$WORK/home-b"
+cd "$WORK/repo"
+
+note() { printf '\n=== %s ===\n' "$*"; }
+result() { printf -- '--> %s\n' "$*"; }
+
+note "0. version"
+"$CLAUDE_BIN" --version
+
+note "1. does -p --output-format stream-json work WITHOUT --verbose?"
+if HOME="$WORK/home-a" "$CLAUDE_BIN" -p --output-format stream-json \
+    --session-id "$SID" -- 'Reply with the single word: ping' >"$WORK/no-verbose.out" 2>&1; then
+  result "PASS without --verbose (flag optional — keep passing it anyway)"
+else
+  result "FAIL without --verbose (required, as the runner assumes): $(tail -1 "$WORK/no-verbose.out")"
+fi
+
+note "2. event shapes (fresh run, --verbose)"
+SID2="$(python3 -c 'import uuid; print(uuid.uuid4())')"
+HOME="$WORK/home-a" "$CLAUDE_BIN" -p --verbose --output-format stream-json \
+  --session-id "$SID2" -- 'Reply with the single word: pong' \
+  | tee "$WORK/stream.jsonl" \
+  | python3 -c '
+import json, sys
+for line in sys.stdin:
+    line = line.strip()
+    if not line: continue
+    try:
+        e = json.loads(line)
+        keys = ",".join(sorted(e.keys()))
+        print(f"  type={e.get(\"type\")}/{e.get(\"subtype\")} keys=[{keys}]")
+    except Exception:
+        print(f"  UNPARSEABLE: {line[:100]}")
+'
+result "exit code: $? — full stream saved to $WORK/stream.jsonl"
+
+note "3. cross-worker resume: materialize transcript onto a clean HOME (LINCHPIN)"
+MUNGED="$(python3 -c "import sys; print(sys.argv[1].replace('/', '-').replace('.', '-'))" "$WORK/repo")"
+SRC="$WORK/home-a/.claude/projects/$MUNGED"
+DST="$WORK/home-b/.claude/projects/$MUNGED"
+mkdir -p "$DST"
+if cp -r "$SRC/$SID2.jsonl" "$DST/" 2>/dev/null; then
+  [ -d "$SRC/$SID2" ] && cp -r "$SRC/$SID2" "$DST/"
+  if HOME="$WORK/home-b" "$CLAUDE_BIN" -p --verbose --output-format stream-json \
+      --resume "$SID2" -- 'What word did you reply with before? Answer with just that word.' \
+      >"$WORK/resume.jsonl" 2>"$WORK/resume.err"; then
+    if grep -q 'pong' "$WORK/resume.jsonl"; then
+      result "PASS: resume on clean HOME retained context (found 'pong')"
+    else
+      result "PARTIAL: resume ran but context unclear — inspect $WORK/resume.jsonl"
+    fi
+  else
+    result "FAIL: resume on clean HOME errored — $(tail -1 "$WORK/resume.err") (fallback re-injection will be load-bearing)"
+  fi
+else
+  result "FAIL: no transcript found at $SRC — munge algorithm wrong for this version?"
+fi
+
+note "4. PreToolUse deny under -p"
+mkdir -p .claude
+cat > .claude/settings.json <<'EOF'
+{
+  "hooks": {
+    "PreToolUse": [
+      {
+        "matcher": "Bash",
+        "hooks": [
+          {"type": "command",
+           "command": "echo '{\"hookSpecificOutput\":{\"hookEventName\":\"PreToolUse\",\"permissionDecision\":\"deny\",\"permissionDecisionReason\":\"validation: always deny\"}}'"}
+        ]
+      }
+    ]
+  },
+  "permissions": {"defaultMode": "acceptEdits", "allow": ["Bash(echo *)"]}
+}
+EOF
+HOME="$WORK/home-a" "$CLAUDE_BIN" -p --verbose --output-format stream-json \
+  --settings .claude/settings.json \
+  -- 'Run the bash command `echo hello` and tell me its output.' \
+  >"$WORK/deny.jsonl" 2>&1
+result "exit=$? — grep the stream for the deny reason:"
+grep -o 'validation: always deny' "$WORK/deny.jsonl" | head -1 \
+  && result "PASS: deny reason surfaced in stream" \
+  || result "INSPECT $WORK/deny.jsonl: deny reason not found"
+
+note "5. Stop hook decision:block under -p"
+cat > .claude/settings.json <<'EOF'
+{
+  "hooks": {
+    "Stop": [
+      {"hooks": [
+        {"type": "command",
+         "command": "if [ -f /tmp/.hv-stop-once ]; then echo '{}'; else touch /tmp/.hv-stop-once; echo '{\"decision\":\"block\",\"reason\":\"validation: say BLOCKED-ONCE then finish\"}'; fi"}
+      ]}
+    ]
+  }
+}
+EOF
+rm -f /tmp/.hv-stop-once
+HOME="$WORK/home-a" "$CLAUDE_BIN" -p --verbose --output-format stream-json \
+  --settings .claude/settings.json -- 'Reply with the word: initial' \
+  >"$WORK/stopblock.jsonl" 2>&1
+result "exit=$? — expect a second turn mentioning BLOCKED-ONCE:"
+grep -o 'BLOCKED-ONCE' "$WORK/stopblock.jsonl" | head -1 \
+  && result "PASS: Stop block re-prompted under -p" \
+  || result "INSPECT $WORK/stopblock.jsonl: no evidence of re-prompt"
+rm -f /tmp/.hv-stop-once
+
+note "6. exit codes"
+HOME="$WORK/home-a" "$CLAUDE_BIN" -p --verbose --output-format stream-json \
+  --resume "$(python3 -c 'import uuid; print(uuid.uuid4())')" -- 'x' \
+  >"$WORK/badresume.out" 2>&1
+result "resume of nonexistent session exit=$? (runner treats nonzero-before-assistant as resume failure)"
+result "INFO: check credential env name with: $CLAUDE_BIN setup-token --help (expect CLAUDE_CODE_OAUTH_TOKEN)"
+
+note "done — artifacts in $WORK"

src/handler/config.py

@@ -51,6 +51,28 @@ class Settings(BaseSettings):
     forge_bin: str = "forge"
     git_bin: str = "git"
 
+    # ---- Headless runner (claude -p --output-format stream-json). ``runner`` selects the
+    # launch path: "tmux" (legacy interactive session, the default until the headless path
+    # is validated) or "headless" (worker-owned subprocess streaming events to the DB).
+    runner: str = "tmux"
+    # How many concurrent claude runs one worker container supervises; commands that would
+    # start a run are left queued (for another worker) while all slots are busy.
+    max_concurrent_runs: int = 4
+    # Heartbeats older than this many seconds mark a worker dead; the reaper flips its
+    # running runs (and their agents) to ``crashed``.
+    worker_stale_after: float = 60.0
+    # Per-run spend cap passed as ``--max-budget-usd``. 0 disables the flag.
+    run_budget_usd: float = 0.0
+    # Refuse to upload a claude session archive larger than this (a runaway sidecar dir
+    # shouldn't balloon the DB); the run still works, only cross-worker resume degrades.
+    session_archive_max_bytes: int = 32 * 1024 * 1024
+    # settings.json ``permissions.defaultMode`` for headless runs. ``-p`` auto-denies
+    # anything that would prompt interactively, so this plus the allowlist below is what
+    # lets normal work proceed; the PreToolUse/Stop hooks stay the hard gate.
+    headless_permission_mode: str = "acceptEdits"
+    # Comma-separated permission allow rules added to generated settings for headless runs.
+    headless_allowed_tools: str = "Bash(git *),Bash(mise *)"
+
     # The pinned `forge` version (README 3.6 / Phase 2: pin, never float on @latest).
     # When set, spawn verifies the injected forge matches and records a mismatch; when
     # empty the check is skipped. Operators align this with what their base image installs.

src/handler/control/headless.py

@@ -0,0 +1,385 @@
+"""The headless runner: worker-owned ``claude -p`` subprocesses streaming to the DB.
+
+This is the tmux replacement seam for agent *runs* (tmux stays only for the interactive
+``/login`` flow). Each invocation is ``claude -p --output-format stream-json`` with a
+pre-assigned ``--session-id``; a :class:`RunSupervisor` owns the child process, appends
+every stdout JSON line to ``agent_events`` as it arrives, derives ``agents.last_output``
+from the latest assistant text (so the existing UI keeps working), and reconciles the
+agent's status from the *process* — exit code and EOF are positive liveness, replacing
+the old pane scraping that could neither see a dead process nor populate the log.
+
+Resume continuity without shared files: claude persists its session under
+``~/.claude/projects/<munged-cwd>/``; the supervisor tars that into ``session_archives``
+(periodically and at exit), and whichever worker later claims a resume materializes the
+archive at the same munged path before running ``claude -p --resume``. Workers therefore
+need the same ``projects_root`` layout — a deployment invariant — but share nothing.
+
+``launch()`` is the function ``spawn``/``worker`` call and tests mock (via the
+``claude_bin`` setting pointing at a fake, same pattern as the tmux fakes).
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import os
+import subprocess
+import tarfile
+import tempfile
+import threading
+import time
+import uuid
+from pathlib import Path
+
+from ..config import get_settings
+from ..db import repository as repo
+from ..db.engine import connection
+
+# Stream types we recognize from ``--output-format stream-json``; anything else (or an
+# unparseable line) is stored as-is so no output is ever dropped. ``worker`` is our own:
+# runner-generated notices (crashes, archive failures, degraded resumes).
+_KNOWN_TYPES = ("system", "assistant", "user", "result", "hook")
+
+# How long a SIGTERM'd claude gets to die before SIGKILL.
+_TERM_GRACE_SECONDS = 10.0
+
+
+def munged_project_dir(working_dir: str) -> str:
+    """The directory name claude uses for a cwd under ``~/.claude/projects/``.
+
+    Claude munges the absolute path by mapping ``/`` and ``.`` to ``-`` (verified against
+    real session state: ``/root/handler`` → ``-root-handler``,
+    ``/root/Talos/.claude/worktrees/x`` → ``-root-Talos--claude-worktrees-x``).
+    """
+    return working_dir.replace("/", "-").replace(".", "-")
+
+
+def session_dir(working_dir: str) -> Path:
+    """Where claude persists sessions for ``working_dir`` (under the *current* $HOME)."""
+    return Path(os.path.expanduser("~")) / ".claude" / "projects" / munged_project_dir(working_dir)
+
+
+def build_spawn_argv(task: str, settings_path: str, session_id: str) -> list[str]:
+    """The headless spawn invocation. ``--verbose`` is required with stream-json in
+    print mode; ``--session-id`` pre-assigns the UUID so the session is addressable
+    (and archivable) from the first event."""
+    s = get_settings()
+    argv = [
+        s.claude_bin, "-p", "--verbose",
+        "--output-format", "stream-json",
+        "--session-id", session_id,
+        "--settings", settings_path,
+    ]
+    if s.run_budget_usd > 0:
+        argv += ["--max-budget-usd", str(s.run_budget_usd)]
+    argv += ["--", task]
+    return argv
+
+
+def build_resume_argv(session_id: str, answer: str, settings_path: str) -> list[str]:
+    """The headless resume invocation — a brand-new process continuing ``session_id``."""
+    s = get_settings()
+    argv = [
+        s.claude_bin, "-p", "--verbose",
+        "--output-format", "stream-json",
+        "--resume", session_id,
+        "--settings", settings_path,
+    ]
+    if s.run_budget_usd > 0:
+        argv += ["--max-budget-usd", str(s.run_budget_usd)]
+    argv += ["--", answer]
+    return argv
+
+
+def parse_stream_line(line: str) -> tuple[str, dict]:
+    """One stdout line → ``(event_type, payload)``. Never raises: malformed JSON or an
+    unrecognized shape comes back as ``("raw", {"line": ...})`` so the event log keeps
+    everything the process said, even across CLI format drift."""
+    text = line.strip()
+    if not text:
+        return "raw", {"line": line}
+    try:
+        payload = json.loads(text)
+    except (ValueError, TypeError):
+        return "raw", {"line": line}
+    if not isinstance(payload, dict):
+        return "raw", {"line": line}
+    etype = payload.get("type")
+    if not isinstance(etype, str) or not etype:
+        return "raw", payload
+    if etype not in _KNOWN_TYPES:
+        # Future/unknown top-level types still store under their own name — the UI
+        # ignores what it doesn't know, but nothing is lost.
+        return etype, payload
+    return etype, payload
+
+
+def assistant_text(payload: dict) -> str | None:
+    """The concatenated text blocks of an ``assistant`` event, or None when it carries
+    none (e.g. a pure tool_use turn). Feeds ``agents.last_output``."""
+    message = payload.get("message")
+    if not isinstance(message, dict):
+        return None
+    content = message.get("content")
+    if isinstance(content, str):
+        return content or None
+    if not isinstance(content, list):
+        return None
+    parts = [
+        block.get("text", "")
+        for block in content
+        if isinstance(block, dict) and block.get("type") == "text"
+    ]
+    text = "\n".join(p for p in parts if p)
+    return text or None
+
+
+def archive_session(working_dir: str, session_id: str, max_bytes: int | None = None) -> bytes | None:
+    """Tar.gz the session transcript + sidecar dir, or None when nothing exists yet or
+    the result would exceed ``max_bytes`` (the caller records a worker event; the run
+    itself is unaffected — only cross-worker resume degrades)."""
+    base = session_dir(working_dir)
+    jsonl = base / f"{session_id}.jsonl"
+    sidecar = base / session_id
+    if not jsonl.exists() and not sidecar.exists():
+        return None
+    buf = io.BytesIO()
+    with tarfile.open(fileobj=buf, mode="w:gz") as tar:
+        if jsonl.exists():
+            tar.add(jsonl, arcname=jsonl.name)
+        if sidecar.is_dir():
+            tar.add(sidecar, arcname=sidecar.name)
+    data = buf.getvalue()
+    limit = max_bytes if max_bytes is not None else get_settings().session_archive_max_bytes
+    if limit and len(data) > limit:
+        return None
+    return data
+
+
+def materialize_session(working_dir: str, archive: bytes) -> None:
+    """Unpack a session archive where claude will look for it on ``--resume``.
+
+    ``filter="data"`` rejects path traversal and special members — the archive came from
+    our own DB, but a defense-in-depth default costs nothing.
+    """
+    base = session_dir(working_dir)
+    base.mkdir(parents=True, exist_ok=True)
+    with tarfile.open(fileobj=io.BytesIO(archive), mode="r:gz") as tar:
+        tar.extractall(base, filter="data")
+
+
+class RunSupervisor:
+    """Owns one headless claude subprocess for its whole life.
+
+    A reader thread pumps stdout lines into ``agent_events``; the supervisor thread
+    watches the process, polls ``cancel_requested`` (cross-worker kill), uploads the
+    session archive periodically, and on exit reconciles run + agent status. Threads are
+    daemons: if the whole worker dies, the reaper — not us — settles the record.
+    """
+
+    def __init__(
+        self,
+        agent: dict,
+        run: dict,
+        argv: list[str],
+        cwd: str,
+        env: dict[str, str],
+        *,
+        cancel_poll: float = 5.0,
+        archive_interval: float = 60.0,
+        on_exit=None,
+    ) -> None:
+        self.agent = agent
+        self.run = run
+        self.argv = argv
+        self.cwd = cwd
+        self.env = env
+        self.cancel_poll = cancel_poll
+        self.archive_interval = archive_interval
+        self.on_exit = on_exit  # worker's slot-release callback
+        self._seq = 0
+        self._result_payload: dict | None = None
+        self._canceled = False
+        self.thread: threading.Thread | None = None
+
+    def start(self) -> None:
+        self.thread = threading.Thread(
+            target=self._supervise, name=f"run-{self.run['id']}", daemon=True
+        )
+        self.thread.start()
+
+    # ---------------------------------------------------------------- internals
+
+    def _insert_event(self, etype: str, payload: dict) -> None:
+        self._seq += 1
+        with connection() as conn:
+            repo.insert_agent_event(
+                conn,
+                self.agent["id"],
+                self.run["id"],
+                seq=self._seq,
+                type=etype,
+                payload=payload,
+                session_id=self.run["session_id"],
+            )
+
+    def _pump_stdout(self, stream) -> None:
+        for line in stream:
+            etype, payload = parse_stream_line(line)
+            try:
+                self._insert_event(etype, payload)
+                if etype == "result":
+                    self._result_payload = payload
+                elif etype == "assistant":
+                    text = assistant_text(payload)
+                    if text:
+                        with connection() as conn:
+                            repo.update_agent_output(conn, self.agent["id"], text)
+            except Exception:  # noqa: BLE001 - a DB hiccup must not sever the pipe
+                continue
+
+    def _upload_archive(self) -> None:
+        try:
+            data = archive_session(self.cwd, self.run["session_id"])
+            if data is None:
+                return
+            with connection() as conn:
+                repo.upsert_session_archive(
+                    conn, self.agent["id"], self.run["session_id"], data
+                )
+        except Exception as exc:  # noqa: BLE001 - archiving is best-effort
+            try:
+                self._insert_event(
+                    "worker", {"notice": "session archive failed", "error": str(exc)}
+                )
+            except Exception:  # noqa: BLE001
+                pass
+
+    def _terminate(self, proc: subprocess.Popen) -> None:
+        self._canceled = True
+        proc.terminate()
+        try:
+            proc.wait(timeout=_TERM_GRACE_SECONDS)
+        except subprocess.TimeoutExpired:
+            proc.kill()
+
+    def _supervise(self) -> None:
+        stderr_file = tempfile.TemporaryFile()
+        try:
+            proc = subprocess.Popen(
+                self.argv,
+                cwd=self.cwd,
+                env={**os.environ, **self.env},
+                stdin=subprocess.DEVNULL,
+                stdout=subprocess.PIPE,
+                stderr=stderr_file,
+                text=True,
+            )
+        except OSError as exc:
+            stderr_file.close()
+            self._settle(exit_code=None, stderr_tail=f"failed to launch claude: {exc}")
+            return
+        reader = threading.Thread(
+            target=self._pump_stdout, args=(proc.stdout,), daemon=True
+        )
+        reader.start()
+        last_archive = time.monotonic()
+        next_cancel_check = time.monotonic() + self.cancel_poll
+        while proc.poll() is None:
+            time.sleep(0.2)
+            now = time.monotonic()
+            if now >= next_cancel_check:
+                next_cancel_check = now + self.cancel_poll
+                with connection() as conn:
+                    if repo.get_cancel_requested(conn, self.run["id"]):
+                        self._terminate(proc)
+                        break
+            if self.archive_interval > 0 and now - last_archive >= self.archive_interval:
+                self._upload_archive()
+                last_archive = now
+        proc.wait()
+        reader.join(timeout=30.0)
+        stderr_file.seek(0)
+        stderr_tail = stderr_file.read()[-4000:].decode("utf-8", "replace")
+        stderr_file.close()
+        self._settle(exit_code=proc.returncode, stderr_tail=stderr_tail)
+
+    def _settle(self, exit_code: int | None, stderr_tail: str) -> None:
+        """Reconcile run + agent status once the process is gone, then final-archive."""
+        result = self._result_payload
+        clean = (
+            exit_code == 0
+            and result is not None
+            and not result.get("is_error", False)
+        )
+        if self._canceled:
+            run_status = "canceled"
+        elif clean:
+            run_status = "completed"
+        else:
+            run_status = "failed"
+        try:
+            with connection() as conn:
+                finished = repo.finish_run(
+                    conn, self.run["id"], run_status, exit_code=exit_code, result=result
+                )
+                # Hooks are the authority on agent status — they ran inside the run and
+                # may have set paused_for_input/blocked/done already. Only an agent still
+                # marked ``working`` needs the process's verdict.
+                agent = repo.get_agent_by_id(conn, self.agent["id"])
+                if finished and agent is not None and agent["status"] == "working":
+                    if self._canceled:
+                        repo.set_agent_status(conn, self.agent["id"], "done")
+                    elif clean:
+                        repo.set_agent_status(conn, self.agent["id"], "done")
+                    else:
+                        repo.set_agent_status(conn, self.agent["id"], "blocked")
+            if not clean and not self._canceled:
+                detail = {"notice": "run failed", "exit_code": exit_code}
+                if stderr_tail.strip():
+                    detail["stderr_tail"] = stderr_tail
+                self._insert_event("worker", detail)
+        except Exception:  # noqa: BLE001 - never let bookkeeping raise out of the thread
+            pass
+        self._upload_archive()
+        if self.on_exit is not None:
+            try:
+                self.on_exit(self)
+            except Exception:  # noqa: BLE001
+                pass
+
+
+def launch(
+    agent: dict,
+    *,
+    kind: str,
+    prompt: str,
+    settings_path: str,
+    env: dict[str, str],
+    worker_id: str,
+    on_exit=None,
+) -> dict:
+    """Start a headless run for ``agent`` and return its ``agent_runs`` row.
+
+    ``kind`` is ``spawn`` (fresh session, new UUID) or ``resume`` (materialize the stored
+    archive, continue the agent's existing session). Fire-and-forget from the caller's
+    perspective — the returned run row is already ``running`` and a daemon supervisor
+    owns the process from here.
+    """
+    working_dir = agent["working_dir"]
+    if kind == "spawn":
+        session_id = str(uuid.uuid4())
+        argv = build_spawn_argv(prompt, settings_path, session_id)
+    else:
+        session_id = agent.get("session_id")
+        if not session_id:
+            raise ValueError(f"agent '{agent['name']}' has no session to resume")
+        argv = build_resume_argv(session_id, prompt, settings_path)
+    with connection() as conn:
+        run = repo.create_run(conn, agent["id"], session_id, worker_id, kind)
+        repo.set_agent_session(conn, agent["id"], session_id, worker_id)
+    supervisor = RunSupervisor(
+        agent, run, argv, cwd=working_dir, env=env, on_exit=on_exit
+    )
+    supervisor.start()
+    return run

src/handler/db/repository.py

@@ -22,6 +22,8 @@ from typing import Any
 from sqlalchemy import Connection, select
 
 from .tables import (
+    agent_events,
+    agent_runs,
     agents,
     approvals,
     checkmarks,
@@ -30,7 +32,9 @@ from .tables import (
     log_entries,
     projects,
     schedules,
+    session_archives,
     shared_context,
+    workers,
 )
 from .upsert import upsert_checkmark
 
@@ -373,6 +377,10 @@ def _purge_agent_dependents(conn: Connection, agent_ids: list[int]) -> None:
     conn.execute(approvals.delete().where(approvals.c.approved_by_agent_id.in_(agent_ids)))
     conn.execute(checkmarks.delete().where(checkmarks.c.agent_id.in_(agent_ids)))
     conn.execute(log_entries.delete().where(log_entries.c.agent_id.in_(agent_ids)))
+    # Headless-runner rows: events reference runs, so they go first.
+    conn.execute(agent_events.delete().where(agent_events.c.agent_id.in_(agent_ids)))
+    conn.execute(agent_runs.delete().where(agent_runs.c.agent_id.in_(agent_ids)))
+    conn.execute(session_archives.delete().where(session_archives.c.agent_id.in_(agent_ids)))
 
 
 def delete_project(conn: Connection, project_id: str) -> bool:
@@ -424,8 +432,13 @@ def enqueue_command(
     agent_name: str | None = None,
     payload: dict | None = None,
     requested_by: str | None = None,
+    target_worker: str | None = None,
 ) -> dict:
-    """Insert a ``queued`` control command for the worker to pick up. Returns the row."""
+    """Insert a ``queued`` control command for the worker to pick up. Returns the row.
+
+    ``target_worker`` pins the command to one worker id (used by login_submit, which must
+    reach the container holding the live login tmux session); null = any worker.
+    """
     result = conn.execute(
         commands.insert().values(
             project_id=project_id,
@@ -434,6 +447,7 @@ def enqueue_command(
             payload=payload,
             status="queued",
             requested_by=requested_by,
+            target_worker=target_worker,
             created_at=_now(),
         )
     )
@@ -457,20 +471,34 @@ def list_commands(
     return [dict(r._mapping) for r in rows]
 
 
-def claim_next_command(conn: Connection, worker_id: str) -> dict | None:
+def claim_next_command(
+    conn: Connection,
+    worker_id: str,
+    types_excluded: tuple[str, ...] = (),
+) -> dict | None:
     """Atomically claim the oldest queued command, flipping it to ``running``.
 
     Postgres uses ``FOR UPDATE SKIP LOCKED`` so multiple workers never grab the same row;
     on SQLite (single writer per transaction) the guarded ``WHERE status='queued'`` update
     plus a rowcount check is enough. Returns the claimed row, or ``None`` when the queue is
     empty or another worker won the race.
+
+    A command whose ``target_worker`` is set is only claimable by that worker (the login
+    flow's two steps must land on the same container). ``types_excluded`` lets a worker
+    with no free run slots skip claiming commands that would launch a new run, leaving
+    them for a less-loaded worker.
     """
     sel = (
         select(commands.c.id)
-        .where(commands.c.status == "queued")
+        .where(
+            commands.c.status == "queued",
+            (commands.c.target_worker.is_(None)) | (commands.c.target_worker == worker_id),
+        )
         .order_by(commands.c.id.asc())
         .limit(1)
     )
+    if types_excluded:
+        sel = sel.where(commands.c.type.not_in(types_excluded))
     if conn.dialect.name == "postgresql":
         sel = sel.with_for_update(skip_locked=True)
     row = conn.execute(sel).first()
@@ -659,3 +687,190 @@ def mark_schedule_run(
             last_run_at=last_run_at, next_run_at=next_run_at, last_command_id=last_command_id
         )
     )
+
+
+# ------------------------------------------------- headless runner (workers/runs/events)
+
+
+def upsert_worker_heartbeat(
+    conn: Connection,
+    worker_id: str,
+    *,
+    hostname: str | None = None,
+    pid: int | None = None,
+    max_runs: int | None = None,
+    active_runs: int | None = None,
+) -> None:
+    """Register a worker or refresh its heartbeat — the reaper's liveness signal."""
+    now = _now()
+    result = conn.execute(
+        workers.update()
+        .where(workers.c.id == worker_id)
+        .values(
+            hostname=hostname, pid=pid, max_runs=max_runs,
+            active_runs=active_runs, heartbeat_at=now,
+        )
+    )
+    if result.rowcount == 0:
+        conn.execute(
+            workers.insert().values(
+                id=worker_id, hostname=hostname, pid=pid, max_runs=max_runs,
+                active_runs=active_runs, started_at=now, heartbeat_at=now,
+            )
+        )
+
+
+def list_stale_workers(conn: Connection, cutoff: datetime) -> list[dict]:
+    """Workers whose heartbeat predates ``cutoff`` — reaper input."""
+    rows = conn.execute(select(workers).where(workers.c.heartbeat_at < cutoff)).all()
+    return [dict(r._mapping) for r in rows]
+
+
+def create_run(
+    conn: Connection, agent_id: int, session_id: str, worker_id: str, kind: str
+) -> dict:
+    """Open an ``agent_runs`` row for a launching headless invocation."""
+    result = conn.execute(
+        agent_runs.insert().values(
+            agent_id=agent_id,
+            session_id=session_id,
+            worker_id=worker_id,
+            kind=kind,
+            status="running",
+            cancel_requested=False,
+            started_at=_now(),
+        )
+    )
+    return get_run(conn, result.inserted_primary_key[0])
+
+
+def get_run(conn: Connection, run_id: int) -> dict | None:
+    row = conn.execute(select(agent_runs).where(agent_runs.c.id == run_id)).first()
+    return _row_to_dict(row)
+
+
+def finish_run(
+    conn: Connection,
+    run_id: int,
+    status: str,
+    exit_code: int | None = None,
+    result: dict | None = None,
+) -> bool:
+    """Close a run — only if still ``running``, so a reaper's ``crashed`` verdict and the
+    supervisor's own exit reconciliation can race without the loser clobbering the winner."""
+    res = conn.execute(
+        agent_runs.update()
+        .where(agent_runs.c.id == run_id, agent_runs.c.status == "running")
+        .values(status=status, exit_code=exit_code, result=result, finished_at=_now())
+    )
+    return res.rowcount > 0
+
+
+def request_run_cancel(conn: Connection, run_id: int) -> bool:
+    """Flag a running run for cancellation; its owning supervisor polls and SIGTERMs."""
+    result = conn.execute(
+        agent_runs.update()
+        .where(agent_runs.c.id == run_id, agent_runs.c.status == "running")
+        .values(cancel_requested=True)
+    )
+    return result.rowcount > 0
+
+
+def get_cancel_requested(conn: Connection, run_id: int) -> bool:
+    row = conn.execute(
+        select(agent_runs.c.cancel_requested).where(agent_runs.c.id == run_id)
+    ).first()
+    return bool(row[0]) if row is not None else False
+
+
+def list_running_runs(conn: Connection, worker_id: str | None = None) -> list[dict]:
+    """Runs currently ``running`` — all of them, or one worker's (the reaper's sweep)."""
+    stmt = select(agent_runs).where(agent_runs.c.status == "running")
+    if worker_id is not None:
+        stmt = stmt.where(agent_runs.c.worker_id == worker_id)
+    rows = conn.execute(stmt.order_by(agent_runs.c.id)).all()
+    return [dict(r._mapping) for r in rows]
+
+
+def get_latest_run(conn: Connection, agent_id: int) -> dict | None:
+    row = conn.execute(
+        select(agent_runs)
+        .where(agent_runs.c.agent_id == agent_id)
+        .order_by(agent_runs.c.id.desc())
+        .limit(1)
+    ).first()
+    return _row_to_dict(row)
+
+
+def set_agent_session(
+    conn: Connection, agent_id: int, session_id: str, worker_id: str
+) -> None:
+    """Record the claude session UUID + supervising worker on the agent row."""
+    conn.execute(
+        agents.update()
+        .where(agents.c.id == agent_id)
+        .values(session_id=session_id, worker_id=worker_id)
+    )
+
+
+def insert_agent_event(
+    conn: Connection,
+    agent_id: int,
+    run_id: int,
+    seq: int,
+    type: str,
+    payload: dict | None = None,
+    session_id: str | None = None,
+) -> int:
+    result = conn.execute(
+        agent_events.insert().values(
+            agent_id=agent_id,
+            run_id=run_id,
+            session_id=session_id,
+            seq=seq,
+            type=type,
+            payload=payload,
+            created_at=_now(),
+        )
+    )
+    return result.inserted_primary_key[0]
+
+
+def list_agent_events(
+    conn: Connection, agent_id: int, after_id: int = 0, limit: int = 200
+) -> list[dict]:
+    """Events for an agent in insertion order, cursor-paged by row id (the UI polls with
+    ``after_id`` = the last id it has, so each poll returns only what's new)."""
+    rows = conn.execute(
+        select(agent_events)
+        .where(agent_events.c.agent_id == agent_id, agent_events.c.id > after_id)
+        .order_by(agent_events.c.id.asc())
+        .limit(limit)
+    ).all()
+    return [dict(r._mapping) for r in rows]
+
+
+def upsert_session_archive(
+    conn: Connection, agent_id: int, session_id: str, archive: bytes
+) -> None:
+    """Store (replace) the agent's latest claude session archive."""
+    now = _now()
+    result = conn.execute(
+        session_archives.update()
+        .where(session_archives.c.agent_id == agent_id)
+        .values(session_id=session_id, archive=archive, bytes=len(archive), updated_at=now)
+    )
+    if result.rowcount == 0:
+        conn.execute(
+            session_archives.insert().values(
+                agent_id=agent_id, session_id=session_id, archive=archive,
+                bytes=len(archive), updated_at=now,
+            )
+        )
+
+
+def get_session_archive(conn: Connection, agent_id: int) -> dict | None:
+    row = conn.execute(
+        select(session_archives).where(session_archives.c.agent_id == agent_id)
+    ).first()
+    return _row_to_dict(row)

src/handler/db/tables.py

@@ -14,6 +14,7 @@ from sqlalchemy import (
     Column,
     ForeignKey,
     Index,
+    LargeBinary,
     MetaData,
     String,
     Table,
@@ -27,7 +28,9 @@ metadata = MetaData()
 
 # Status vocabularies kept as free TEXT (README uses plain strings, not PG enums, so
 # both dialects match). CheckConstraints make the allowed sets explicit and portable.
-AGENT_STATUSES = ("working", "paused_for_input", "blocked", "done")
+# ``crashed`` is reserved for the reaper: it marks an agent whose owning worker went
+# silent mid-run — never a normal exit, which reconciles to done/blocked instead.
+AGENT_STATUSES = ("working", "paused_for_input", "blocked", "done", "crashed")
 GATE_STATUSES = ("pass", "fail", "unknown")
 CI_STATUSES = ("not_applicable", "pending", "pass", "fail")
 VISIBILITIES = ("project", "global")
@@ -55,6 +58,10 @@ COMMAND_TYPES = (
 COMMAND_STATUSES = ("queued", "running", "done", "failed")
 # Forge families a host can belong to (drives per-host token env conventions).
 FORGE_TYPES = ("github", "gitlab", "gitea", "forgejo", "bitbucket")
+# One agent_runs row per headless ``claude -p`` invocation (spawn or resume).
+# ``canceled`` = an operator kill; ``crashed`` = the reaper found the owning worker dead.
+RUN_KINDS = ("spawn", "resume")
+RUN_STATUSES = ("running", "completed", "failed", "crashed", "canceled")
 
 
 def _in(column: str, values: tuple[str, ...]) -> str:
@@ -87,8 +94,14 @@ agents = Table(
     # A periodic snapshot of the agent's live tmux pane tail (last ~40 lines), refreshed by
     # the control worker's poll loop. The tmux socket lives only in the control container,
     # so this DB column is how the API/UI see what a running — or wedged — agent is doing.
+    # Headless runs reuse it, derived from the latest assistant text instead of a pane tail.
     Column("last_output", String),
     Column("output_at", PortableTimestamp),
+    # Headless runner (null on legacy tmux agents — the rollout discriminator): the claude
+    # session UUID pre-assigned at spawn (stable across resumes), and the worker currently
+    # (or last) supervising a run for this agent.
+    Column("session_id", String),
+    Column("worker_id", String),
     Column("created_at", PortableTimestamp, nullable=False, server_default=func.now()),
     UniqueConstraint("project_id", "name", name="uq_agents_project_name"),
     CheckConstraint(_in("status", AGENT_STATUSES), name="ck_agents_status"),
@@ -192,6 +205,9 @@ commands = Table(
     Column("error", String),
     Column("requested_by", String),  # actor label, e.g. "operator:web"
     Column("claimed_by", String),  # worker id that claimed the command
+    # Pins a command to one worker (null = any). login_submit must run on the worker that
+    # ran login_start — the live tmux login session exists only in that container.
+    Column("target_worker", String),
     Column("created_at", PortableTimestamp, nullable=False, server_default=func.now()),
     Column("claimed_at", PortableTimestamp),
     Column("finished_at", PortableTimestamp),
@@ -227,6 +243,79 @@ forge_hosts = Table(
     CheckConstraint(_in("forge_type", FORGE_TYPES), name="ck_forge_hosts_type"),
 )
 
+# Worker registry + heartbeat (headless runner). Each worker container upserts its row
+# every loop pass; the reaper marks a worker's running runs (and their agents) ``crashed``
+# when ``heartbeat_at`` goes stale — the positive-liveness replacement for tmux scraping.
+workers = Table(
+    "workers",
+    metadata,
+    Column("id", String, primary_key=True),  # "worker-<host>-<pid>-<rand>"
+    Column("hostname", String),
+    Column("pid", BigInteger),
+    Column("max_runs", BigInteger),
+    Column("active_runs", BigInteger),
+    Column("started_at", PortableTimestamp, nullable=False, server_default=func.now()),
+    Column("heartbeat_at", PortableTimestamp, nullable=False, server_default=func.now()),
+)
+
+# One row per headless ``claude -p`` invocation. The spawn/resume *command* finishes at
+# launch (fire-and-forget, matching tmux semantics); the run row is what tracks the
+# process's actual life — status, exit code, and the final result event.
+agent_runs = Table(
+    "agent_runs",
+    metadata,
+    Column("id", PortableBigInt, primary_key=True, autoincrement=True),
+    Column("agent_id", BigInteger, ForeignKey("agents.id"), nullable=False),
+    Column("session_id", String, nullable=False),
+    Column("worker_id", String, nullable=False),
+    Column("kind", String, nullable=False),
+    Column("status", String, nullable=False, server_default="running"),
+    # Cross-worker kill: any worker/API sets this; the owning supervisor polls it and
+    # SIGTERMs its own child — nobody signals a process they don't own.
+    Column("cancel_requested", Boolean, nullable=False, server_default="0"),
+    Column("exit_code", BigInteger),
+    Column("result", PortableJSON),  # the stream's result event (cost/turns/is_error/text)
+    Column("started_at", PortableTimestamp, nullable=False, server_default=func.now()),
+    Column("finished_at", PortableTimestamp),
+    CheckConstraint(_in("kind", RUN_KINDS), name="ck_agent_runs_kind"),
+    CheckConstraint(_in("status", RUN_STATUSES), name="ck_agent_runs_status"),
+    Index("ix_agent_runs_status_worker", "status", "worker_id"),
+)
+
+# The persisted event stream — one row per stream-json stdout line of a run, in order.
+# This is what the UI's log/event panel reads; ``type`` mirrors the stream's top-level
+# type (system/assistant/user/result), plus ``hook`` (--include-hook-events), ``worker``
+# (runner-generated notices: crash marks, archive failures, fallback resumes) and ``raw``
+# (an unparseable line, stored verbatim — the parser never drops data).
+agent_events = Table(
+    "agent_events",
+    metadata,
+    Column("id", PortableBigInt, primary_key=True, autoincrement=True),
+    Column("agent_id", BigInteger, ForeignKey("agents.id"), nullable=False),
+    Column("run_id", BigInteger, ForeignKey("agent_runs.id"), nullable=False),
+    Column("session_id", String),
+    Column("seq", BigInteger, nullable=False),  # per-run stdout line counter
+    Column("type", String, nullable=False),
+    Column("payload", PortableJSON),
+    Column("created_at", PortableTimestamp, nullable=False, server_default=func.now()),
+    Index("ix_agent_events_agent_id", "agent_id", "id"),
+)
+
+# Latest claude session archive per agent — the tar.gz of ``<sid>.jsonl`` + its sidecar
+# dir from ``~/.claude/projects/<munged-cwd>/``. Uploaded by the supervising worker
+# (periodically and at run end) and materialized by whichever worker claims the next
+# resume, so ``--resume`` works cross-worker with no shared filesystem (README: DB is the
+# single source of truth; observed sizes are KBs-to-low-MBs).
+session_archives = Table(
+    "session_archives",
+    metadata,
+    Column("agent_id", BigInteger, ForeignKey("agents.id"), primary_key=True),
+    Column("session_id", String, nullable=False),
+    Column("archive", LargeBinary, nullable=False),
+    Column("bytes", BigInteger, nullable=False),
+    Column("updated_at", PortableTimestamp, nullable=False, server_default=func.now()),
+)
+
 # Recurring agent spawns. The worker checks for due rows on every loop pass and enqueues
 # an ordinary ``spawn`` command per firing (so scheduled runs show up in the Activity
 # audit trail like any other control action). Agent names must be unique per project, so

src/handler/migrations/versions/0008_headless_runs.py

@@ -0,0 +1,128 @@
+"""headless runner: workers, runs, events, session archives
+
+Revision ID: 0008_headless_runs
+Revises: 0007_agent_pane_output
+Create Date: 2026-07-21
+
+Dormant schema for the headless ``claude -p --output-format stream-json`` runner
+(nothing writes these until the runner is enabled):
+
+- ``workers`` — worker registry + heartbeat, the reaper's liveness input.
+- ``agent_runs`` — one row per headless invocation (spawn or resume), tracking the
+  process's real life: status, exit code, cancel flag, final result event.
+- ``agent_events`` — the persisted stream-json event log the UI reads.
+- ``session_archives`` — latest claude session tar.gz per agent, so ``--resume`` works
+  from any worker with no shared filesystem.
+- ``agents`` gains ``session_id`` (null = legacy tmux agent, the rollout discriminator)
+  and ``worker_id``; ``commands`` gains ``target_worker`` (pins login_submit to the
+  worker holding the live login session).
+- ``crashed`` joins the agent status vocabulary (reaper-set); the CHECK constraint
+  swaps go through ``batch_alter_table`` so SQLite recreates while Postgres alters in
+  place (same pattern as 0006's command type).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+from handler.db.types import PortableBigInt, PortableJSON, PortableTimestamp
+
+revision: str = "0008_headless_runs"
+down_revision: str | None = "0007_agent_pane_output"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+OLD_AGENT_STATUSES = "'working', 'paused_for_input', 'blocked', 'done'"
+NEW_AGENT_STATUSES = "'working', 'paused_for_input', 'blocked', 'done', 'crashed'"
+RUN_KINDS = "'spawn', 'resume'"
+RUN_STATUSES = "'running', 'completed', 'failed', 'crashed', 'canceled'"
+
+
+def upgrade() -> None:
+    op.create_table(
+        "workers",
+        sa.Column("id", sa.String(), primary_key=True),
+        sa.Column("hostname", sa.String()),
+        sa.Column("pid", sa.BigInteger()),
+        sa.Column("max_runs", sa.BigInteger()),
+        sa.Column("active_runs", sa.BigInteger()),
+        sa.Column("started_at", PortableTimestamp, nullable=False, server_default=sa.func.now()),
+        sa.Column("heartbeat_at", PortableTimestamp, nullable=False, server_default=sa.func.now()),
+    )
+    op.create_table(
+        "agent_runs",
+        sa.Column("id", PortableBigInt, primary_key=True, autoincrement=True),
+        sa.Column("agent_id", sa.BigInteger(), sa.ForeignKey("agents.id"), nullable=False),
+        sa.Column("session_id", sa.String(), nullable=False),
+        sa.Column("worker_id", sa.String(), nullable=False),
+        sa.Column("kind", sa.String(), nullable=False),
+        sa.Column("status", sa.String(), nullable=False, server_default="running"),
+        sa.Column("cancel_requested", sa.Boolean(), nullable=False, server_default="0"),
+        sa.Column("exit_code", sa.BigInteger()),
+        sa.Column("result", PortableJSON),
+        sa.Column("started_at", PortableTimestamp, nullable=False, server_default=sa.func.now()),
+        sa.Column("finished_at", PortableTimestamp),
+        sa.CheckConstraint(f"kind IN ({RUN_KINDS})", name="ck_agent_runs_kind"),
+        sa.CheckConstraint(f"status IN ({RUN_STATUSES})", name="ck_agent_runs_status"),
+    )
+    op.create_index(
+        "ix_agent_runs_status_worker", "agent_runs", ["status", "worker_id"]
+    )
+    op.create_table(
+        "agent_events",
+        sa.Column("id", PortableBigInt, primary_key=True, autoincrement=True),
+        sa.Column("agent_id", sa.BigInteger(), sa.ForeignKey("agents.id"), nullable=False),
+        sa.Column("run_id", sa.BigInteger(), sa.ForeignKey("agent_runs.id"), nullable=False),
+        sa.Column("session_id", sa.String()),
+        sa.Column("seq", sa.BigInteger(), nullable=False),
+        sa.Column("type", sa.String(), nullable=False),
+        sa.Column("payload", PortableJSON),
+        sa.Column("created_at", PortableTimestamp, nullable=False, server_default=sa.func.now()),
+    )
+    op.create_index("ix_agent_events_agent_id", "agent_events", ["agent_id", "id"])
+    op.create_table(
+        "session_archives",
+        sa.Column("agent_id", sa.BigInteger(), sa.ForeignKey("agents.id"), primary_key=True),
+        sa.Column("session_id", sa.String(), nullable=False),
+        sa.Column("archive", sa.LargeBinary(), nullable=False),
+        sa.Column("bytes", sa.BigInteger(), nullable=False),
+        sa.Column("updated_at", PortableTimestamp, nullable=False, server_default=sa.func.now()),
+    )
+    op.add_column("agents", sa.Column("session_id", sa.String()))
+    op.add_column("agents", sa.Column("worker_id", sa.String()))
+    op.add_column("commands", sa.Column("target_worker", sa.String()))
+    with op.batch_alter_table("agents", schema=None) as batch_op:
+        batch_op.drop_constraint("ck_agents_status", type_="check")
+        batch_op.create_check_constraint(
+            "ck_agents_status", f"status IN ({NEW_AGENT_STATUSES})"
+        )
+    with op.batch_alter_table("checkmarks", schema=None) as batch_op:
+        batch_op.drop_constraint("ck_checkmarks_status", type_="check")
+        batch_op.create_check_constraint(
+            "ck_checkmarks_status", f"status IN ({NEW_AGENT_STATUSES})"
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("checkmarks", schema=None) as batch_op:
+        batch_op.drop_constraint("ck_checkmarks_status", type_="check")
+        batch_op.create_check_constraint(
+            "ck_checkmarks_status", f"status IN ({OLD_AGENT_STATUSES})"
+        )
+    with op.batch_alter_table("agents", schema=None) as batch_op:
+        batch_op.drop_constraint("ck_agents_status", type_="check")
+        batch_op.create_check_constraint(
+            "ck_agents_status", f"status IN ({OLD_AGENT_STATUSES})"
+        )
+    op.drop_column("commands", "target_worker")
+    op.drop_column("agents", "worker_id")
+    op.drop_column("agents", "session_id")
+    op.drop_table("session_archives")
+    op.drop_index("ix_agent_events_agent_id", table_name="agent_events")
+    op.drop_table("agent_events")
+    op.drop_index("ix_agent_runs_status_worker", table_name="agent_runs")
+    op.drop_table("agent_runs")
+    op.drop_table("workers")

tests/fixtures/fake_claude.py

@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+"""A stand-in ``claude`` binary for headless-runner tests.
+
+Wired in via the existing ``claude_bin`` setting (the same seam the tmux fakes used).
+Parses the real headless argv, emits a scripted ``--output-format stream-json`` stream on
+stdout, and writes a genuine session transcript + sidecar under
+``$HOME/.claude/projects/<munged-cwd>/`` so archive/materialize/resume paths exercise the
+real filesystem layout. Behavior is selected with ``FAKE_CLAUDE_MODE``:
+
+- ``success`` (default): init + assistant + result, exit 0.
+- ``error``: init + assistant + one garbage line, then exit 2 with no result event.
+- ``hang``: init, then sleep forever (the kill/cancel/reaper tests SIGTERM it).
+- ``slow``: like success with a pause between events (concurrency tests).
+- ``resume-fail``: a ``--resume`` invocation exits 1 before any assistant event
+  (exercises the context re-injection fallback).
+
+On ``--resume`` (outside resume-fail) the transcript materialized by the worker MUST
+already exist at the expected path — missing means cross-worker materialization broke,
+so the fake fails loudly, exit 3.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import signal
+import sys
+import time
+from pathlib import Path
+
+
+def _parse_argv(argv: list[str]) -> dict:
+    opts = {
+        "print": False,
+        "verbose": False,
+        "output_format": None,
+        "session_id": None,
+        "resume": None,
+        "settings": None,
+        "budget": None,
+        "prompt": None,
+    }
+    i = 0
+    while i < len(argv):
+        arg = argv[i]
+        if arg == "-p":
+            opts["print"] = True
+        elif arg == "--verbose":
+            opts["verbose"] = True
+        elif arg == "--output-format":
+            i += 1
+            opts["output_format"] = argv[i]
+        elif arg == "--session-id":
+            i += 1
+            opts["session_id"] = argv[i]
+        elif arg in ("--resume", "-r"):
+            i += 1
+            opts["resume"] = argv[i]
+        elif arg == "--settings":
+            i += 1
+            opts["settings"] = argv[i]
+        elif arg == "--max-budget-usd":
+            i += 1
+            opts["budget"] = argv[i]
+        elif arg == "--":
+            opts["prompt"] = " ".join(argv[i + 1 :])
+            break
+        else:
+            opts["prompt"] = arg
+        i += 1
+    return opts
+
+
+def _munged(cwd: str) -> str:
+    return cwd.replace("/", "-").replace(".", "-")
+
+
+def _emit(event: dict) -> None:
+    sys.stdout.write(json.dumps(event) + "\n")
+    sys.stdout.flush()
+
+
+def _write_transcript(session_id: str, prompt: str) -> None:
+    base = Path(os.path.expanduser("~")) / ".claude" / "projects" / _munged(os.getcwd())
+    base.mkdir(parents=True, exist_ok=True)
+    jsonl = base / f"{session_id}.jsonl"
+    with jsonl.open("a") as fh:
+        fh.write(json.dumps({"type": "user", "prompt": prompt}) + "\n")
+        fh.write(json.dumps({"type": "assistant", "text": f"handled: {prompt}"}) + "\n")
+    sidecar = base / session_id / "tool-results"
+    sidecar.mkdir(parents=True, exist_ok=True)
+    (sidecar / "result-1.txt").write_text("fake tool output\n")
+
+
+def main() -> int:
+    signal.signal(signal.SIGTERM, signal.SIG_DFL)
+    mode = os.environ.get("FAKE_CLAUDE_MODE", "success")
+    opts = _parse_argv(sys.argv[1:])
+    if not opts["print"] or opts["output_format"] != "stream-json":
+        sys.stderr.write("fake_claude: expected -p --output-format stream-json\n")
+        return 64
+
+    session_id = opts["session_id"] or opts["resume"] or "fake-session"
+    prompt = opts["prompt"] or ""
+
+    if mode == "resume-fail" and opts["resume"]:
+        sys.stderr.write("fake_claude: no conversation found to resume\n")
+        return 1
+
+    if opts["resume"]:
+        base = Path(os.path.expanduser("~")) / ".claude" / "projects" / _munged(os.getcwd())
+        if not (base / f"{session_id}.jsonl").exists():
+            sys.stderr.write(f"fake_claude: transcript missing at {base}\n")
+            return 3
+
+    _emit(
+        {
+            "type": "system",
+            "subtype": "init",
+            "session_id": session_id,
+            "cwd": os.getcwd(),
+            "tools": ["Bash", "Read", "Edit"],
+        }
+    )
+    if mode == "hang":
+        time.sleep(3600)
+        return 0
+    if mode == "slow":
+        time.sleep(float(os.environ.get("FAKE_CLAUDE_SLOW_SECONDS", "1.0")))
+
+    _emit(
+        {
+            "type": "assistant",
+            "session_id": session_id,
+            "message": {
+                "role": "assistant",
+                "content": [{"type": "text", "text": f"working on: {prompt}"}],
+            },
+        }
+    )
+    _write_transcript(session_id, prompt)
+
+    if mode == "error":
+        sys.stdout.write("this is not json\n")
+        sys.stdout.flush()
+        return 2
+
+    _emit(
+        {
+            "type": "result",
+            "subtype": "success",
+            "session_id": session_id,
+            "is_error": False,
+            "num_turns": 1,
+            "total_cost_usd": 0.01,
+            "result": f"done: {prompt}",
+        }
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_headless_parser.py

@@ -0,0 +1,178 @@
+"""Pure-helper tests for the headless runner: stream parsing, path munging, argv
+construction, and the session archive round-trip. No subprocess is launched here —
+the process-level tests live in test_headless_run.py (phase 2)."""
+
+from __future__ import annotations
+
+import io
+import tarfile
+
+from handler.control import headless
+
+
+# ------------------------------------------------------------------ parse_stream_line
+
+
+def test_parse_valid_event_types():
+    for etype in ("system", "assistant", "user", "result", "hook"):
+        parsed_type, payload = headless.parse_stream_line(f'{{"type": "{etype}", "x": 1}}\n')
+        assert parsed_type == etype
+        assert payload == {"type": etype, "x": 1}
+
+
+def test_parse_unknown_type_keeps_its_name():
+    parsed_type, payload = headless.parse_stream_line('{"type": "telemetry", "n": 2}')
+    assert parsed_type == "telemetry"
+    assert payload["n"] == 2
+
+
+def test_parse_garbage_becomes_raw():
+    parsed_type, payload = headless.parse_stream_line("this is not json\n")
+    assert parsed_type == "raw"
+    assert payload == {"line": "this is not json\n"}
+
+
+def test_parse_non_dict_json_becomes_raw():
+    parsed_type, payload = headless.parse_stream_line('["a", "b"]')
+    assert parsed_type == "raw"
+
+
+def test_parse_missing_type_becomes_raw():
+    parsed_type, payload = headless.parse_stream_line('{"message": "no type field"}')
+    assert parsed_type == "raw"
+    assert payload == {"message": "no type field"}
+
+
+def test_parse_blank_line_becomes_raw():
+    parsed_type, _ = headless.parse_stream_line("   \n")
+    assert parsed_type == "raw"
+
+
+# -------------------------------------------------------------------- assistant_text
+
+
+def test_assistant_text_joins_text_blocks():
+    payload = {
+        "type": "assistant",
+        "message": {
+            "content": [
+                {"type": "text", "text": "first"},
+                {"type": "tool_use", "name": "Bash", "input": {}},
+                {"type": "text", "text": "second"},
+            ]
+        },
+    }
+    assert headless.assistant_text(payload) == "first\nsecond"
+
+
+def test_assistant_text_none_for_pure_tool_use():
+    payload = {
+        "type": "assistant",
+        "message": {"content": [{"type": "tool_use", "name": "Bash", "input": {}}]},
+    }
+    assert headless.assistant_text(payload) is None
+
+
+def test_assistant_text_string_content():
+    assert headless.assistant_text({"message": {"content": "plain"}}) == "plain"
+
+
+def test_assistant_text_missing_message():
+    assert headless.assistant_text({"type": "assistant"}) is None
+
+
+# ---------------------------------------------------------------- munged_project_dir
+
+
+def test_munge_matches_recorded_real_examples():
+    # Recorded from a real ~/.claude/projects/ (see plan): '/' and '.' both map to '-'.
+    assert headless.munged_project_dir("/root/handler") == "-root-handler"
+    assert (
+        headless.munged_project_dir("/root/Talos/.claude/worktrees/mise-tooling")
+        == "-root-Talos--claude-worktrees-mise-tooling"
+    )
+
+
+# ------------------------------------------------------------------- argv builders
+
+
+def test_spawn_argv_shape(env):
+    argv = headless.build_spawn_argv("do the task", "/wd/.claude/settings.json", "sid-1")
+    assert argv[0] == "claude"
+    assert "-p" in argv and "--verbose" in argv
+    assert argv[argv.index("--output-format") + 1] == "stream-json"
+    assert argv[argv.index("--session-id") + 1] == "sid-1"
+    assert argv[argv.index("--settings") + 1] == "/wd/.claude/settings.json"
+    assert "--max-budget-usd" not in argv  # default budget is 0 = off
+    assert argv[-2:] == ["--", "do the task"]
+    assert "--resume" not in argv
+
+
+def test_resume_argv_shape(env):
+    argv = headless.build_resume_argv("sid-2", "the answer", "/wd/.claude/settings.json")
+    assert argv[argv.index("--resume") + 1] == "sid-2"
+    assert argv[-2:] == ["--", "the answer"]
+    assert "--session-id" not in argv
+
+
+def test_spawn_argv_includes_budget_when_set(env, monkeypatch):
+    monkeypatch.setenv("RUN_BUDGET_USD", "2.5")
+    from handler import config
+
+    config.get_settings.cache_clear()
+    argv = headless.build_spawn_argv("t", "/s.json", "sid")
+    assert argv[argv.index("--max-budget-usd") + 1] == "2.5"
+    config.get_settings.cache_clear()
+
+
+# ------------------------------------------------------------- archive round-trip
+
+
+def _write_fake_session(home, working_dir: str, session_id: str) -> None:
+    base = home / ".claude" / "projects" / headless.munged_project_dir(working_dir)
+    base.mkdir(parents=True)
+    (base / f"{session_id}.jsonl").write_text('{"type": "user", "prompt": "hi"}\n')
+    sidecar = base / session_id / "tool-results"
+    sidecar.mkdir(parents=True)
+    (sidecar / "r1.txt").write_text("tool output")
+
+
+def test_archive_and_materialize_round_trip(env, tmp_path, monkeypatch):
+    working_dir = "/projects/demo"
+    _write_fake_session(tmp_path, working_dir, "sid-rt")
+
+    data = headless.archive_session(working_dir, "sid-rt")
+    assert data is not None
+
+    # Materialize onto a *different* worker: a fresh HOME with no session state.
+    other_home = tmp_path / "other-worker"
+    other_home.mkdir()
+    monkeypatch.setenv("HOME", str(other_home))
+    headless.materialize_session(working_dir, data)
+
+    base = other_home / ".claude" / "projects" / headless.munged_project_dir(working_dir)
+    assert (base / "sid-rt.jsonl").read_text() == '{"type": "user", "prompt": "hi"}\n'
+    assert (base / "sid-rt" / "tool-results" / "r1.txt").read_text() == "tool output"
+
+
+def test_archive_none_when_no_session(env):
+    assert headless.archive_session("/projects/none", "missing-sid") is None
+
+
+def test_archive_refuses_oversize(env, tmp_path):
+    working_dir = "/projects/big"
+    _write_fake_session(tmp_path, working_dir, "sid-big")
+    assert headless.archive_session(working_dir, "sid-big", max_bytes=10) is None
+
+
+def test_archive_contains_only_session_members(env, tmp_path):
+    working_dir = "/projects/demo2"
+    _write_fake_session(tmp_path, working_dir, "sid-a")
+    # A sibling session must not leak into sid-a's archive.
+    base = tmp_path / ".claude" / "projects" / headless.munged_project_dir(working_dir)
+    (base / "sid-other.jsonl").write_text("{}\n")
+
+    data = headless.archive_session(working_dir, "sid-a")
+    with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") as tar:
+        names = tar.getnames()
+    assert all(n == "sid-a.jsonl" or n.startswith("sid-a") for n in names)

tests/test_repository_runs.py

@@ -0,0 +1,174 @@
+"""Repository coverage for the headless-runner tables (workers / agent_runs /
+agent_events / session_archives) and the new claim filters. The ``env`` fixture runs the
+real ``alembic upgrade head``, so migration 0008 itself is under test here too."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from handler.db import repository as repo
+
+
+def _agent(conn, name="a1"):
+    repo.create_project(conn, "p", "/projects/p")
+    return repo.create_agent(conn, "p", name, "/projects/p")
+
+
+# ------------------------------------------------------------------------ agent_runs
+
+
+def test_run_lifecycle(conn):
+    agent = _agent(conn)
+    run = repo.create_run(conn, agent["id"], "sid-1", "worker-a", "spawn")
+    assert run["status"] == "running"
+    assert run["kind"] == "spawn"
+    assert run["cancel_requested"] is False
+
+    assert repo.finish_run(conn, run["id"], "completed", exit_code=0, result={"ok": True})
+    done = repo.get_run(conn, run["id"])
+    assert done["status"] == "completed"
+    assert done["exit_code"] == 0
+    assert done["result"] == {"ok": True}
+    assert done["finished_at"] is not None
+
+
+def test_finish_run_only_once(conn):
+    """The supervisor's verdict and a racing reaper can't clobber each other."""
+    agent = _agent(conn)
+    run = repo.create_run(conn, agent["id"], "sid", "w", "spawn")
+    assert repo.finish_run(conn, run["id"], "crashed") is True
+    assert repo.finish_run(conn, run["id"], "completed", exit_code=0) is False
+    assert repo.get_run(conn, run["id"])["status"] == "crashed"
+
+
+def test_cancel_request_roundtrip(conn):
+    agent = _agent(conn)
+    run = repo.create_run(conn, agent["id"], "sid", "w", "resume")
+    assert repo.get_cancel_requested(conn, run["id"]) is False
+    assert repo.request_run_cancel(conn, run["id"]) is True
+    assert repo.get_cancel_requested(conn, run["id"]) is True
+    # A finished run can't be re-flagged.
+    repo.finish_run(conn, run["id"], "canceled")
+    assert repo.request_run_cancel(conn, run["id"]) is False
+
+
+def test_list_running_runs_scoped_by_worker(conn):
+    agent = _agent(conn)
+    r1 = repo.create_run(conn, agent["id"], "s1", "worker-a", "spawn")
+    r2 = repo.create_run(conn, agent["id"], "s2", "worker-b", "spawn")
+    repo.finish_run(conn, r1["id"], "completed")
+    running = repo.list_running_runs(conn)
+    assert [r["id"] for r in running] == [r2["id"]]
+    assert repo.list_running_runs(conn, worker_id="worker-a") == []
+    assert [r["id"] for r in repo.list_running_runs(conn, worker_id="worker-b")] == [r2["id"]]
+
+
+def test_latest_run_and_agent_session(conn):
+    agent = _agent(conn)
+    repo.create_run(conn, agent["id"], "s1", "w", "spawn")
+    latest = repo.create_run(conn, agent["id"], "s1", "w", "resume")
+    assert repo.get_latest_run(conn, agent["id"])["id"] == latest["id"]
+
+    repo.set_agent_session(conn, agent["id"], "s1", "worker-a")
+    updated = repo.get_agent_by_id(conn, agent["id"])
+    assert updated["session_id"] == "s1"
+    assert updated["worker_id"] == "worker-a"
+
+
+def test_agent_status_crashed_allowed(conn):
+    """Migration 0008 widened ck_agents_status — 'crashed' must insert cleanly."""
+    agent = _agent(conn)
+    repo.set_agent_status(conn, agent["id"], "crashed")
+    assert repo.get_agent_by_id(conn, agent["id"])["status"] == "crashed"
+
+
+# ---------------------------------------------------------------------- agent_events
+
+
+def test_events_cursor_pagination(conn):
+    agent = _agent(conn)
+    run = repo.create_run(conn, agent["id"], "sid", "w", "spawn")
+    for seq in range(1, 4):
+        repo.insert_agent_event(
+            conn, agent["id"], run["id"], seq=seq, type="assistant",
+            payload={"n": seq}, session_id="sid",
+        )
+    first = repo.list_agent_events(conn, agent["id"], limit=2)
+    assert [e["payload"]["n"] for e in first] == [1, 2]
+    rest = repo.list_agent_events(conn, agent["id"], after_id=first[-1]["id"])
+    assert [e["payload"]["n"] for e in rest] == [3]
+    assert repo.list_agent_events(conn, agent["id"], after_id=rest[-1]["id"]) == []
+
+
+# ------------------------------------------------------------------ session_archives
+
+
+def test_session_archive_upsert_replaces(conn):
+    agent = _agent(conn)
+    repo.upsert_session_archive(conn, agent["id"], "s1", b"v1")
+    repo.upsert_session_archive(conn, agent["id"], "s2", b"v2-longer")
+    row = repo.get_session_archive(conn, agent["id"])
+    assert row["session_id"] == "s2"
+    assert bytes(row["archive"]) == b"v2-longer"
+    assert row["bytes"] == len(b"v2-longer")
+
+
+def test_session_archive_missing(conn):
+    agent = _agent(conn)
+    assert repo.get_session_archive(conn, agent["id"]) is None
+
+
+# ------------------------------------------------------------------------- workers
+
+
+def test_worker_heartbeat_upsert_and_staleness(conn):
+    repo.upsert_worker_heartbeat(conn, "worker-a", hostname="h1", pid=42, max_runs=4)
+    repo.upsert_worker_heartbeat(conn, "worker-a", hostname="h1", pid=42, active_runs=2)
+
+    future = datetime.now(UTC) + timedelta(seconds=1)
+    stale = repo.list_stale_workers(conn, cutoff=future)
+    assert [w["id"] for w in stale] == ["worker-a"]
+    assert stale[0]["active_runs"] == 2
+
+    past = datetime.now(UTC) - timedelta(minutes=5)
+    assert repo.list_stale_workers(conn, cutoff=past) == []
+
+
+# ------------------------------------------------------------------- claim filters
+
+
+def test_claim_respects_target_worker(conn):
+    repo.create_project(conn, "p", "/projects/p")
+    pinned = repo.enqueue_command(conn, "login_submit", payload={"code": "x"},
+                                  target_worker="worker-b")
+    # worker-a can't see worker-b's pinned command...
+    assert repo.claim_next_command(conn, "worker-a") is None
+    # ...but worker-b claims it.
+    claimed = repo.claim_next_command(conn, "worker-b")
+    assert claimed["id"] == pinned["id"]
+    assert claimed["status"] == "running"
+
+
+def test_claim_excludes_types_when_slots_full(conn):
+    repo.create_project(conn, "p", "/projects/p")
+    spawn_cmd = repo.enqueue_command(conn, "spawn", project_id="p", agent_name="a")
+    sync_cmd = repo.enqueue_command(conn, "sync", project_id="p")
+    # With run-starting types excluded, the older spawn is skipped for the sync.
+    claimed = repo.claim_next_command(
+        conn, "w", types_excluded=("spawn", "resume", "mise_init")
+    )
+    assert claimed["id"] == sync_cmd["id"]
+    # The spawn stays queued for a worker with a free slot.
+    assert repo.claim_next_command(conn, "w2")["id"] == spawn_cmd["id"]
+
+
+def test_delete_agent_cascades_headless_rows(conn):
+    agent = _agent(conn)
+    run = repo.create_run(conn, agent["id"], "sid", "w", "spawn")
+    repo.insert_agent_event(conn, agent["id"], run["id"], seq=1, type="system", payload={})
+    repo.upsert_session_archive(conn, agent["id"], "sid", b"data")
+
+    assert repo.delete_agent(conn, "p", agent["name"]) is True
+    assert repo.get_latest_run(conn, agent["id"]) is None
+    assert repo.list_agent_events(conn, agent["id"]) == []
+    assert repo.get_session_archive(conn, agent["id"]) is None