diff --git a/scripts/validate_claude_headless.sh b/scripts/validate_claude_headless.sh new file mode 100755 index 0000000..ff46c7d --- /dev/null +++ b/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" diff --git a/src/handler/config.py b/src/handler/config.py index 7f5718d..eaebd14 100644 --- a/src/handler/config.py +++ b/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. diff --git a/src/handler/control/headless.py b/src/handler/control/headless.py new file mode 100644 index 0000000..3582810 --- /dev/null +++ b/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//``; 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 diff --git a/src/handler/db/repository.py b/src/handler/db/repository.py index c737be0..ecf4dc0 100644 --- a/src/handler/db/repository.py +++ b/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) diff --git a/src/handler/db/tables.py b/src/handler/db/tables.py index 3e62f04..843839c 100644 --- a/src/handler/db/tables.py +++ b/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---" + 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 ``.jsonl`` + its sidecar +# dir from ``~/.claude/projects//``. 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 diff --git a/src/handler/migrations/versions/0008_headless_runs.py b/src/handler/migrations/versions/0008_headless_runs.py new file mode 100644 index 0000000..eaf86be --- /dev/null +++ b/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") diff --git a/tests/fixtures/fake_claude.py b/tests/fixtures/fake_claude.py new file mode 100755 index 0000000..126fe63 --- /dev/null +++ b/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//`` 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()) diff --git a/tests/test_headless_parser.py b/tests/test_headless_parser.py new file mode 100644 index 0000000..0b5863b --- /dev/null +++ b/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) diff --git a/tests/test_repository_runs.py b/tests/test_repository_runs.py new file mode 100644 index 0000000..3616320 --- /dev/null +++ b/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