From 1517e4dca81d84486ca8915d2f00c4987f6207ad Mon Sep 17 00:00:00 2001 From: Wyatt Date: Tue, 21 Jul 2026 23:20:39 -0400 Subject: [PATCH] feat!: headless is the only runner - delete the tmux run path (phase 4) Agent runs are now always worker-owned 'claude -p' subprocesses; tmux survives only for the interactive /login flow. - deleted: worker.capture_agent_output/_pane_tail + the capture loop arm (the empty-/log bug's home), spawn's tmux launch/_claude_command, the tmux resume/kill branches (the silent-send-keys bug's home), tmux.session_name/list_sessions, the CLI attach subcommand, the 'runner' setting - spawn: task is now a hard requirement (headless has no idle REPL) - enforced in spawn (SpawnError) and the API (400); onboarding seeding dropped (-p skips the trust dialog) - resume: single headless path; pre-headless agent rows (no session_id) degrade to the context-re-injection fresh run - settings_gen: permissions allowlist is always emitted - credsync: change-triggered uploads key on .claude/.credentials.json only (claude touches ~/.claude.json every run - keying on it would ping-pong uploads between workers); logins still publish explicitly - cli list: liveness from agent_runs in the DB, not tmux - tests: spawn/kill/resume re-pointed at the fake_launch seam (conftest); integration test now drives API -> worker -> real fake claude subprocess -> events endpoint; README documents the headless model + multi-worker deployment invariants Suite 295 green; frontend unchanged since phase 3. --- README.md | 51 ++++++++++--- src/handler/api/routes/agents.py | 9 +-- src/handler/config.py | 7 +- src/handler/control/cli.py | 37 +++------ src/handler/control/credsync.py | 29 ++++--- src/handler/control/settings_gen.py | 25 +++--- src/handler/control/spawn.py | 113 +++++++++------------------- src/handler/control/tmux.py | 27 +------ src/handler/control/worker.py | 74 ++++-------------- tests/conftest.py | 37 +++++++-- tests/test_control_spawn.py | 102 ++++++++++++++++++------- tests/test_control_spawn_forge.py | 28 +++---- tests/test_credsync.py | 7 -- tests/test_headless_run.py | 8 +- tests/test_integration_web_spawn.py | 93 +++++++++++++++++++---- tests/test_worker.py | 39 ---------- tests/test_worker_concurrency.py | 5 -- 17 files changed, 337 insertions(+), 354 deletions(-) diff --git a/README.md b/README.md index 51d3321..6ef589d 100644 --- a/README.md +++ b/README.md @@ -13,10 +13,11 @@ git remote, and your own network exposure. > control layer, HTTP API, database, migrations, and verification/approval hooks are > implemented and tested (106 tests, SQLite). Phase 2 adds credential resolution + > injection, role-based forge-workflow skills, a hard approval gate, and a CI-status -> poller. Live end-to-end agent spawning against a real `claude` binary + tmux is stubbed -> behind mockable seams (`tmux`, `verify`, `forge`, `gitops`, `spawn.resume`) and wired -> but not yet exercised against production binaries. See [`docs/PLAN.md`](docs/PLAN.md) -> for the full design and roadmap. +> poller. Agent runs are headless (`claude -p --output-format stream-json`, supervised +> by the worker, events persisted to the DB); the run/kill/resume paths are exercised +> end-to-end against a scripted fake claude binary, with a manual validation script +> (`scripts/validate_claude_headless.sh`) for the real one. See +> [`docs/PLAN.md`](docs/PLAN.md) for the full design and roadmap. --- @@ -45,19 +46,23 @@ the control layer and API are disposable compute that can restart or scale out f ``` writes reads (+ answer backfill) ┌──────────────────┐ ┌──────────────┐ ┌──────────────────┐ - │ control layer │───────▶│ database │◀───────│ HTTP API │ + │ worker(s) │───────▶│ database │◀───────│ HTTP API │ │ (CLI + hooks) │ │ PG / SQLite │ │ (FastAPI) │ └──────────────────┘ └──────────────┘ └──────────────────┘ │ ▲ ▲ │ spawns │ Stop / PreToolUse / Notification hooks │ curl, UI, any client - ▼ │ write checkmark + log rows │ (bearer token) - tmux + claude binary (one working dir / worktree per agent) + ▼ │ + streamed run events, checkmark, log │ (bearer token) + claude -p --output-format stream-json (one working dir / worktree per agent) ``` -- **Control layer** (`handler.control`) — the only writer. Spawns/lists/attaches/kills - agents as `tmux` sessions running the `claude` binary, one working directory or git - worktree per agent, namespaced `project__agent`. Stateless; every write goes straight - to the database. +- **Control layer / workers** (`handler.control`) — the only writer. Runs each agent as + a **headless** `claude -p --output-format stream-json` subprocess (one working + directory or git worktree per agent), streams every stdout event into the database as + it happens, and reconciles agent status from the process itself (exit code + EOF — + positive liveness, no screen scraping). Stateless: repo state is pulled from git when + a task is claimed, claude session transcripts are archived to / materialized from the + DB for cross-worker `--resume`, and the claude login credential bundle is distributed + encrypted through the DB. tmux survives only to drive the interactive `/login` flow. - **Hooks** (`handler.hooks`) — run inside each agent via a generated `settings.json`. They write the checkpoint/log rows and enforce the test and push gates. - **API** (`handler.api`) — a thin, read-mostly HTTP layer over the same database (the @@ -77,10 +82,32 @@ The data model is defined once (SQLAlchemy Core) and renders correctly on both: Portable column types bridge the two, and the checkmark upsert uses native `INSERT … ON CONFLICT DO UPDATE` on both dialects. Migrations are Alembic, dual-dialect. +### Scaling workers horizontally + +Multiple worker containers can drain the same command queue concurrently (Postgres +`FOR UPDATE SKIP LOCKED`); each supervises up to `MAX_CONCURRENT_RUNS` claude processes +and skips claiming run-starting commands while full, leaving them for a less-loaded +worker. Workers heartbeat into the DB; if one dies mid-run, any surviving worker's +reaper marks its runs (and their agents) `crashed` — visible in the UI with the last +output preserved — and the operator resumes explicitly on whichever worker picks it up. + +Deployment invariants for multi-worker: + +- **No shared filesystems.** Git carries repo state (workers clone/pull on claim); + claude session transcripts live in `session_archives`; login credentials are + Fernet-encrypted into `runtime_secrets` and materialized by every worker. +- **Identical `PROJECTS_ROOT` on every worker** — claude keys its session storage to + the absolute working-dir path, so cross-worker `--resume` needs the same layout. +- **The same `HANDLER_SECRET_KEY` on every worker** (and the API) — without it, the + credential bundle can't be distributed and only the worker that ran `/login` can run + agents. +- The two-step web login is automatically pinned to one worker + (`commands.target_worker`), so it works unchanged with a fleet. + ## Requirements - Python 3.11+ -- `git` and `tmux` (for live spawning) +- `git` (for live spawning) and `tmux` (only for the web `/login` flow) - A `claude` binary, authenticated (for live spawning) - `mise` in each managed project, with a `.mise.toml` defining at least a `test` task - Postgres (default) — or nothing but a file path for the SQLite fallback diff --git a/src/handler/api/routes/agents.py b/src/handler/api/routes/agents.py index 80b9683..0a0a46e 100644 --- a/src/handler/api/routes/agents.py +++ b/src/handler/api/routes/agents.py @@ -12,7 +12,6 @@ from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy import Connection from sqlalchemy.exc import IntegrityError -from ...config import get_settings from ...db import repository as repo from ..deps import db_conn, require_admin, require_auth from ..schemas import ( @@ -79,10 +78,10 @@ def enqueue_spawn(project: str, body: SpawnIn, conn: Connection = Depends(db_con status.HTTP_409_CONFLICT, detail=f"agent '{body.name}' already exists in project '{project}'", ) - if get_settings().runner == "headless" and not body.task: - # A tmux agent can idle at the REPL awaiting input; a headless `claude -p` run - # with no prompt exits immediately. Reject here (400) instead of letting the - # command fail asynchronously in the worker. + if not body.task: + # A headless `claude -p` run with no prompt exits immediately having done + # nothing. Reject here (400) instead of letting the command fail asynchronously + # in the worker. raise HTTPException( status.HTTP_400_BAD_REQUEST, detail="a task is required: the headless runner has no idle-REPL mode", diff --git a/src/handler/config.py b/src/handler/config.py index bb618a7..1684dc4 100644 --- a/src/handler/config.py +++ b/src/handler/config.py @@ -51,10 +51,9 @@ 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" + # ---- Headless runner (claude -p --output-format stream-json): worker-owned + # subprocesses streaming events to the DB. Agent runs are always headless; tmux + # remains only for the interactive /login flow. # 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 diff --git a/src/handler/control/cli.py b/src/handler/control/cli.py index be9d20a..2c7285c 100644 --- a/src/handler/control/cli.py +++ b/src/handler/control/cli.py @@ -1,10 +1,11 @@ """``handler`` CLI — the control layer's write side. -Spawn/list/attach/kill manage agent processes. Phase 2 adds the forge-workflow control -commands: ``approve``/``reject`` (the senior agent records its verdict, which the deploy -gate checks), ``poll-ci`` (backfill CI verdicts), and ``forge-init`` (write the role -skills into a managed repo). The DB is the source of truth for what agents exist; tmux is -cross-checked for liveness. All commands are project-namespaced. +Spawn/list/kill manage agent runs (headless ``claude -p`` processes supervised by the +worker). Phase 2 adds the forge-workflow control commands: ``approve``/``reject`` (the +senior agent records its verdict, which the deploy gate checks), ``poll-ci`` (backfill +CI verdicts), and ``forge-init`` (write the role skills into a managed repo). The DB is +the single source of truth: what agents exist AND whether their runs are live both come +from it. All commands are project-namespaced. """ from __future__ import annotations @@ -15,7 +16,7 @@ import sys from ..db import repository as repo from ..db.engine import connection -from . import poller, reposync, skills_gen, spawn, tmux, worker +from . import poller, reposync, skills_gen, spawn, worker def _cmd_spawn(args: argparse.Namespace) -> int: @@ -37,38 +38,27 @@ def _cmd_spawn(args: argparse.Namespace) -> int: print(f" role: {args.role}") if agent.get("forge_note"): print(f" warning: {agent['forge_note']}", file=sys.stderr) - print(f" tmux session: {tmux.session_name(args.project, args.name)}") return 0 def _cmd_list(args: argparse.Namespace) -> int: - live = set(tmux.list_sessions()) with connection() as conn: + live = {run["agent_id"] for run in repo.list_running_runs(conn)} if args.project: projects = [args.project] if repo.get_project(conn, args.project) else [] else: projects = [p["id"] for p in repo.list_projects(conn)] for project_id in projects: for agent in repo.list_agents(conn, project_id): - session = tmux.session_name(project_id, agent["name"]) - alive = "live" if session in live else "-" + alive = "live" if agent["id"] in live else "-" role = agent.get("role") or "-" + worker_id = agent.get("worker_id") or "-" print( - f"{project_id}/{agent['name']}\t{role}\t{agent['status']}\t{alive}\t{session}" + f"{project_id}/{agent['name']}\t{role}\t{agent['status']}\t{alive}\t{worker_id}" ) return 0 -def _cmd_attach(args: argparse.Namespace) -> int: - session = tmux.session_name(args.project, args.name) - if not tmux.has_session(session): - print(f"error: no live session '{session}'", file=sys.stderr) - return 1 - # Replace this process with an interactive tmux attach. - os.execvp("tmux", ["tmux", "attach", "-t", session]) - return 0 # pragma: no cover - execvp does not return - - def _cmd_kill(args: argparse.Namespace) -> int: try: spawn.kill(args.project, args.name) @@ -233,11 +223,6 @@ def build_parser() -> argparse.ArgumentParser: p_list.add_argument("--project", help="limit to one project") p_list.set_defaults(func=_cmd_list) - p_attach = sub.add_parser("attach", help="attach to an agent's tmux session") - p_attach.add_argument("--project", required=True) - p_attach.add_argument("--name", required=True) - p_attach.set_defaults(func=_cmd_attach) - p_kill = sub.add_parser("kill", help="kill an agent's session") p_kill.add_argument("--project", required=True) p_kill.add_argument("--name", required=True) diff --git a/src/handler/control/credsync.py b/src/handler/control/credsync.py index f253b41..f76363f 100644 --- a/src/handler/control/credsync.py +++ b/src/handler/control/credsync.py @@ -43,15 +43,19 @@ def _credential_files() -> dict[str, str]: def fingerprint() -> tuple: - """(path, mtime_ns, size) of the on-disk credential files — cheap change detection.""" - fp = [] - for path in sorted(_credential_files().values()): - try: - st = os.stat(path) - fp.append((path, st.st_mtime_ns, st.st_size)) - except OSError: - continue - return tuple(fp) + """(path, mtime_ns, size) of the OAuth token file — cheap change detection. + + Deliberately only ``.claude/.credentials.json``: claude touches ``~/.claude.json`` + on every run (project entries, UI state), and treating those as "new credentials" + would ping-pong uploads between workers forever. A login that only rewrites + ``.claude.json`` is still published — the login flow calls :func:`upload` directly. + """ + path = _credential_files()[".claude/.credentials.json"] + try: + st = os.stat(path) + except OSError: + return () + return ((path, st.st_mtime_ns, st.st_size),) def upload() -> bool: @@ -165,10 +169,3 @@ def refresh() -> str | None: _state.seen_updated_at = stored["updated_at"] if stored else None return "uploaded" return None - - -def note_local_write() -> None: - """Record that this process just changed local credentials deliberately (e.g. the - claude_config onboarding merge at spawn), so refresh() doesn't misread the mtime - bump as a new login and ping-pong uploads between workers.""" - _state.last_fingerprint = fingerprint() diff --git a/src/handler/control/settings_gen.py b/src/handler/control/settings_gen.py index f8c1ea1..6e89d02 100644 --- a/src/handler/control/settings_gen.py +++ b/src/handler/control/settings_gen.py @@ -21,7 +21,7 @@ def _hook_command(event: str) -> str: return f"{sys.executable} -m handler.hooks {event}" -def build_settings(headless: bool = False) -> dict: +def build_settings() -> dict: settings = { "hooks": { "Stop": [{"hooks": [{"type": "command", "command": _hook_command("stop")}]}], @@ -39,24 +39,23 @@ def build_settings(headless: bool = False) -> dict: ], } } - if headless: - # ``claude -p`` never prompts — anything that would ask for permission is - # auto-denied. The allowlist is therefore what lets normal work (git, mise, the - # project's own tooling) proceed; the PreToolUse/Stop hooks above remain the hard - # gate either way, since a hook deny overrides any allow. - s = get_settings() - settings["permissions"] = { - "defaultMode": s.headless_permission_mode, - "allow": s.headless_allowed_tools_list, - } + # ``claude -p`` never prompts — anything that would ask for permission is + # auto-denied. The allowlist is therefore what lets normal work (git, mise, the + # project's own tooling) proceed; the PreToolUse/Stop hooks above remain the hard + # gate either way, since a hook deny overrides any allow. + s = get_settings() + settings["permissions"] = { + "defaultMode": s.headless_permission_mode, + "allow": s.headless_allowed_tools_list, + } return settings -def write_settings(working_dir: str, headless: bool = False) -> str: +def write_settings(working_dir: str) -> str: """Write ``.claude/settings.json`` under the agent's working dir; return its path.""" claude_dir = os.path.join(working_dir, ".claude") os.makedirs(claude_dir, exist_ok=True) path = os.path.join(claude_dir, "settings.json") with open(path, "w") as fh: - json.dump(build_settings(headless=headless), fh, indent=2) + json.dump(build_settings(), fh, indent=2) return path diff --git a/src/handler/control/spawn.py b/src/handler/control/spawn.py index 008fd7c..561c11c 100644 --- a/src/handler/control/spawn.py +++ b/src/handler/control/spawn.py @@ -15,16 +15,13 @@ from ..config import get_settings from ..db import repository as repo from ..db.engine import connection from . import ( - claude_config, credentials, - credsync, forge, gitops, headless, mise, reposync, settings_gen, - tmux, worktree, ) @@ -47,18 +44,6 @@ def require_test_task(working_dir: str) -> None: ) -def _claude_command(task: str | None, settings_path: str) -> str: - claude = get_settings().claude_bin - argv = [claude, "--settings", settings_path] - if task: - argv.append(_shell_quote(task)) - return " ".join(argv) - - -def _shell_quote(value: str) -> str: - return "'" + value.replace("'", "'\\''") + "'" - - def _install_git_credentials( working_dir: str, git_remote: str | None, conn=None ) -> None: @@ -97,10 +82,10 @@ def spawn( test gate. ``worker_id`` identifies the calling worker container (headless runs record it on the run row; the CLI defaults to a pid-scoped id). """ - if get_settings().runner == "headless" and not task: - # A tmux agent without a task idles at the REPL waiting for input; ``claude -p`` - # has no such mode — an empty prompt would exit immediately having done nothing. - raise SpawnError("a headless agent requires a task (the runner is 'headless')") + if not task: + # ``claude -p`` has no idle-REPL mode — an empty prompt would exit immediately + # having done nothing, so a task is a hard requirement. + raise SpawnError("an agent requires a task (headless claude has no idle mode)") sync_note = None with connection() as conn: project = repo.get_project(conn, project_id) @@ -154,8 +139,7 @@ def spawn( role=role, ) - headless_run = get_settings().runner == "headless" - settings_path = settings_gen.write_settings(working_dir, headless=headless_run) + settings_path = settings_gen.write_settings(working_dir) env = _agent_env(project, agent, token, role=role, mise_init=mise_init) # Verify the pinned forge version, if one is configured. Non-fatal: a version drift @@ -163,25 +147,14 @@ def spawn( # touches forge and the base image is the real pin (README 3.6, Phase 2). forge_note = _check_forge_version(working_dir) - # Mark Claude Code onboarding complete + trust the working dir before launching. The - # tmux path needs both (no human at the TTY to answer the theme/trust screens); - # ``-p`` skips the trust dialog but still reads onboarding state, so keep it for both. - claude_config.ensure_onboarded(working_dir) - credsync.note_local_write() - - if headless_run: - headless.launch( - agent, - kind="spawn", - prompt=task, - settings_path=settings_path, - env=env, - worker_id=worker_id or f"cli-{os.getpid()}", - ) - else: - session = tmux.session_name(project_id, name) - command = _claude_command(task, settings_path) - tmux.new_session(session, cwd=working_dir, command=command, env=env) + headless.launch( + agent, + kind="spawn", + prompt=task, + settings_path=settings_path, + env=env, + worker_id=worker_id or f"cli-{os.getpid()}", + ) agent = {**agent, "forge_note": forge_note, "sync_note": sync_note} return agent @@ -234,51 +207,35 @@ def _check_forge_version(working_dir: str) -> str | None: def kill(project_id: str, name: str) -> None: + """Stop an agent: flag its running run for cancel and mark the row done. + + The owning worker's supervisor polls the cancel flag and SIGTERMs its own child + (cross-worker safe — nobody signals a process they don't own). No running run means + the process is already gone; the status update is all that's left to do. + """ with connection() as conn: agent = repo.get_agent_by_name(conn, project_id, name) if agent is None: raise SpawnError(f"agent '{name}' not found in project '{project_id}'") - if agent.get("session_id"): - # Headless agent: flag the running run for cancel; the owning worker's - # supervisor polls the flag and SIGTERMs its own child (cross-worker safe — - # nobody signals a process they don't own). No running run = already dead. - run = repo.get_latest_run(conn, agent["id"]) - if run is not None and run["status"] == "running": - repo.request_run_cancel(conn, run["id"]) - repo.set_agent_status(conn, agent["id"], "done") - return - session = tmux.session_name(project_id, name) - if tmux.has_session(session): - tmux.kill_session(session) + run = repo.get_latest_run(conn, agent["id"]) + if run is not None and run["status"] == "running": + repo.request_run_cancel(conn, run["id"]) repo.set_agent_status(conn, agent["id"], "done") def resume(agent: dict, answer: str, worker_id: str | None = None) -> tuple[bool, str]: - """Feed an operator's answer back to an agent. + """Feed an operator's answer back to an agent as a new ``claude -p --resume`` run. - The seam the API's ``/resume`` route calls (and the one tests mock). Legacy tmux - agents (``session_id`` null) get the answer typed into their live session; headless - agents get a brand-new ``claude -p --resume`` run on this worker, with the session - transcript materialized from the DB archive first so any worker can serve the resume. - """ - if agent.get("session_id"): - return _resume_headless(agent, answer, worker_id or f"cli-{os.getpid()}") - session = tmux.session_name(agent["project_id"], agent["name"]) - if not tmux.has_session(session): - return False, f"no live session '{session}' to resume" - tmux.send_keys(session, answer) - return True, f"answer delivered to session '{session}'" - - -def _resume_headless(agent: dict, answer: str, worker_id: str) -> tuple[bool, str]: - """Launch a ``--resume`` run for a headless agent, materializing its session first. - - Refuses while a run is still live (two concurrent processes on one session would - corrupt it). When neither this worker nor the DB has the transcript — the owning - worker died before its first archive — falls back to a *fresh* session whose prompt - re-injects context from the DB (checkmark + open question + answer), recorded as a - ``worker`` event so the UI shows the degraded continuity. + The seam the API's ``/resume`` route calls (and the one tests mock). The session + transcript is materialized from the DB archive first, so ANY worker can serve the + resume. Refuses while a run is still live (two concurrent processes on one session + would corrupt it). When no transcript survives anywhere — the owning worker died + before its first archive, or the row predates the headless runner — falls back to a + *fresh* session whose prompt re-injects context from the DB (checkmark + open + question + answer), recorded as a ``worker`` event so the UI shows the degraded + continuity. """ + worker_id = worker_id or f"cli-{os.getpid()}" with connection() as conn: run = repo.get_latest_run(conn, agent["id"]) if run is not None and run["status"] == "running": @@ -289,7 +246,7 @@ def _resume_headless(agent: dict, answer: str, worker_id: str) -> tuple[bool, st return False, f"project '{agent['project_id']}' not registered" working_dir = agent["working_dir"] - settings_path = settings_gen.write_settings(working_dir, headless=True) + settings_path = settings_gen.write_settings(working_dir) try: token = None with connection() as conn: @@ -298,6 +255,10 @@ def _resume_headless(agent: dict, answer: str, worker_id: str) -> tuple[bool, st return False, str(exc) env = _agent_env(project, agent, token) + if not agent.get("session_id"): + # Pre-headless agent row (or a spawn that never launched): nothing to --resume. + return _resume_reinjected(agent, answer, settings_path, env, worker_id) + transcript = headless.session_dir(working_dir) / f"{agent['session_id']}.jsonl" if archive is not None: try: diff --git a/src/handler/control/tmux.py b/src/handler/control/tmux.py index a29fb9d..94bd49f 100644 --- a/src/handler/control/tmux.py +++ b/src/handler/control/tmux.py @@ -1,7 +1,8 @@ -"""Thin tmux wrapper — the single mock seam for spawning. +"""Thin tmux wrapper — now used ONLY by the interactive ``/login`` flow. -Every tmux/claude invocation goes through these functions so tests can substitute a -fake and never touch a real tmux server or ``claude`` binary. +Agent runs are headless (``control.headless``); the one thing that still genuinely +needs a TTY is driving claude's ``/login`` OAuth screens. Everything here goes through +subprocess so the login tests can substitute a fake and never touch a real tmux server. """ from __future__ import annotations @@ -11,14 +12,6 @@ import subprocess from ..config import get_settings -def session_name(project_id: str, agent_name: str) -> str: - """``project__agent`` with tmux-illegal characters sanitized (README 3.4).""" - safe = f"{project_id}__{agent_name}" - for ch in (".", ":", " "): - safe = safe.replace(ch, "-") - return safe - - def new_session( name: str, cwd: str, @@ -57,18 +50,6 @@ def has_session(name: str) -> bool: return result.returncode == 0 -def list_sessions() -> list[str]: - tmux = get_settings().tmux_bin - result = subprocess.run( - [tmux, "list-sessions", "-F", "#{session_name}"], - capture_output=True, - text=True, - ) - if result.returncode != 0: - return [] - return [line for line in result.stdout.splitlines() if line] - - def kill_session(name: str) -> None: tmux = get_settings().tmux_bin subprocess.run([tmux, "kill-session", "-t", name], check=True) diff --git a/src/handler/control/worker.py b/src/handler/control/worker.py index 36c2b36..7368b99 100644 --- a/src/handler/control/worker.py +++ b/src/handler/control/worker.py @@ -1,11 +1,13 @@ -"""The control-container worker: executes commands the API enqueues. +"""The control/worker container: executes commands the API enqueues and supervises +headless claude runs. -The API (in its own container) has no ``git``/``tmux``/``claude`` and does not own the -tmux sessions, so it cannot run control actions directly. Instead it writes a ``queued`` -row to the ``commands`` table; this worker — running in the control container — claims each -row, dispatches it to the *same* control functions the CLI uses (``spawn``/``poller``/ -``skills_gen``/``repo.record_approval``), and writes the result or error back. It also runs -the periodic CI sweep, subsuming the old ``poll-ci --watch`` loop. +The API (in its own container) has no ``git``/``claude``, so it cannot run control +actions directly. Instead it writes a ``queued`` row to the ``commands`` table; any +worker claims each row (multi-worker safe — ``FOR UPDATE SKIP LOCKED`` + slot-aware +claim filters), dispatches it to the *same* control functions the CLI uses (``spawn``/ +``poller``/``skills_gen``/``repo.record_approval``), and writes the result or error +back. It also heartbeats + reaps dead workers' runs, syncs claude credentials, and runs +the periodic CI sweep. Every command runs in isolation: one bad command is recorded as ``failed`` and never stops the loop. ``execute_command`` is the pure dispatch seam (given a claimed command dict, @@ -23,7 +25,7 @@ from datetime import UTC, datetime, timedelta from ..config import get_settings from ..db import repository as repo from ..db.engine import connection -from . import credsync, gitops, login, poller, reposync, skills_gen, spawn, tmux +from . import credsync, gitops, login, poller, reposync, skills_gen, spawn # Command types that launch a claude run and therefore need a free slot on this worker. # A worker with all slots busy leaves these queued for a less-loaded worker to claim. @@ -394,43 +396,6 @@ def fire_due_schedules(now: datetime | None = None) -> int: return fired -# How many trailing pane lines to snapshot — enough to show the current screen (a menu, a -# prompt, the tail of the last command) without bloating the row. -_PANE_TAIL_LINES = 40 - - -def _pane_tail(pane: str, lines: int = _PANE_TAIL_LINES) -> str: - """The last ``lines`` of a captured pane, trailing blank lines trimmed so an idle - screen doesn't store as a wall of whitespace.""" - rows = (pane or "").splitlines() - while rows and not rows[-1].strip(): - rows.pop() - return "\n".join(rows[-lines:]) - - -def capture_agent_output() -> int: - """Snapshot each working agent's live tmux pane tail into the DB. - - The tmux socket lives only in the control container, so this is the one channel the - API/UI have onto what a running — or wedged — agent is actually doing: an agent stuck - on claude's first-run theme picker surfaces as that screen instead of a misleading - green 'working'. A missing session is skipped (its process is gone). Returns the count - updated. - """ - with connection() as conn: - working = repo.list_agents_by_status(conn, "working") - updated = 0 - for agent in working: - session = tmux.session_name(agent["project_id"], agent["name"]) - if not tmux.has_session(session): - continue - tail = _pane_tail(tmux.capture_pane(session)) - with connection() as conn: - repo.update_agent_output(conn, agent["id"], tail) - updated += 1 - return updated - - def drain(worker_id: str, limit: int | None = None) -> int: """Claim and run queued commands until the queue is empty (or ``limit`` reached). @@ -454,15 +419,12 @@ def drain(worker_id: str, limit: int | None = None) -> int: def _full_slot_exclusions(worker_id: str) -> tuple[str, ...]: """Command types this worker must not claim right now. - Headless runs are supervised in-process, so a worker at ``max_concurrent_runs`` skips + Runs are supervised in-process, so a worker at ``max_concurrent_runs`` skips claiming run-starting commands — they stay queued for a worker with a free slot. Slot accounting is DB-driven (this worker's ``running`` runs), so it needs no in-memory registry and survives restarts (a fresh process gets a fresh worker id; stale rows - belong to the old id and are the reaper's problem). Tmux runs are fire-and-forget and - never consume a slot. + belong to the old id and are the reaper's problem). """ - if get_settings().runner != "headless": - return () with connection() as conn: active = len(repo.list_running_runs(conn, worker_id=worker_id)) if active >= get_settings().max_concurrent_runs: @@ -474,20 +436,18 @@ def run( worker_id: str | None = None, poll_interval: float = 2.0, ci_interval: float = 30.0, - capture_interval: float = 2.0, credsync_interval: float = 30.0, reap_interval: float = 15.0, iterations: int | None = None, ) -> None: - """The control-container main loop: drain the command queue, snapshot live agent - output, sync claude credentials, heartbeat + reap dead workers, and sweep CI. + """The control-container main loop: drain the command queue, sync claude + credentials, heartbeat + reap dead workers, and sweep CI. ``iterations`` bounds the loop for tests; production runs unbounded. Sleeps ``poll_interval`` only when a pass found no commands, so bursts drain promptly. """ worker_id = worker_id or make_worker_id() last_ci = 0.0 - last_capture = 0.0 last_credsync = 0.0 last_reap = 0.0 count = 0 @@ -508,12 +468,6 @@ def run( pass did_work = drain(worker_id) > 0 now = time.monotonic() - if capture_interval > 0 and now - last_capture >= capture_interval: - try: - capture_agent_output() - except Exception: # noqa: BLE001 - a capture hiccup must not kill the worker - pass - last_capture = now if credsync_interval > 0 and (last_credsync == 0.0 or now - last_credsync >= credsync_interval): # First pass runs immediately: a fresh worker container must materialize the # claude credentials before it claims its first spawn. diff --git a/tests/conftest.py b/tests/conftest.py index 5adb9d7..d931253 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,7 +1,8 @@ """Shared fixtures. Everything runs on a fresh SQLite file per test, materialized via a *real* ``alembic upgrade head`` — so the migration path itself is under test, not -just ``create_all``. No live claude/tmux/mise is ever touched: the three seams -(``control.tmux``, ``hooks.verify``, ``control.spawn.resume``) are faked. +just ``create_all``. No live claude/tmux/mise is ever touched: the seams +(``control.headless.launch`` for runs, ``control.tmux`` for the login flow, +``hooks.verify``, ``control.spawn.resume``) are faked. """ from __future__ import annotations @@ -111,20 +112,44 @@ def fake_tmux(monkeypatch): def send_enter(name): calls["send_enter"].append({"name": name}) - def list_sessions(): - return list(live) - monkeypatch.setattr(tmux, "new_session", new_session) monkeypatch.setattr(tmux, "has_session", has_session) monkeypatch.setattr(tmux, "kill_session", kill_session) monkeypatch.setattr(tmux, "send_keys", send_keys) monkeypatch.setattr(tmux, "send_text", send_text) monkeypatch.setattr(tmux, "send_enter", send_enter) - monkeypatch.setattr(tmux, "list_sessions", list_sessions) return {"calls": calls, "live": live} +@pytest.fixture +def fake_launch(monkeypatch): + """Record ``headless.launch`` calls instead of spawning a claude subprocess. + + Mirrors the real launch's DB side effects (run row + agent session/worker) so kill/ + resume logic downstream of a fake spawn behaves like production, minus the process. + """ + from handler.control import headless + from handler.db import repository as repo + from handler.db.engine import connection + + calls: list[dict] = [] + + def launch(agent, *, kind, prompt, settings_path, env, worker_id, on_exit=None): + session_id = agent.get("session_id") if kind == "resume" else f"fake-sid-{len(calls) + 1}" + 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) + calls.append( + {"agent": agent, "kind": kind, "prompt": prompt, "settings_path": settings_path, + "env": env, "worker_id": worker_id, "run": run} + ) + return run + + monkeypatch.setattr(headless, "launch", launch) + return calls + + @pytest.fixture def fake_gitops(monkeypatch): """Fake the git seam: record config/add/commit, return a controllable branch/sha.""" diff --git a/tests/test_control_spawn.py b/tests/test_control_spawn.py index fb34d92..5fb48ef 100644 --- a/tests/test_control_spawn.py +++ b/tests/test_control_spawn.py @@ -1,4 +1,7 @@ -"""Control-layer spawn: the hard test-task gate, settings generation, identity env.""" +"""Control-layer spawn: the hard test-task gate, settings generation, identity env. + +Spawns go through the ``fake_launch`` seam (conftest) — the headless analogue of the old +fake tmux: it records the launch and mirrors its DB side effects, no subprocess.""" from __future__ import annotations @@ -24,36 +27,48 @@ def _write_mise(root, with_test=True): (root / ".mise.toml").write_text(body) -def test_spawn_refuses_without_test_task(env, fake_tmux): +def test_spawn_refuses_without_test_task(env, fake_launch): root = env["tmp"] / "proj" _write_mise(root, with_test=False) _register_project(root) with pytest.raises(spawn.SpawnError, match="no \\[tasks.test\\]"): - spawn.spawn("proj", "api") - assert fake_tmux["calls"]["new_session"] == [] + spawn.spawn("proj", "api", task="do it") + assert fake_launch == [] -def test_spawn_refuses_without_mise_file(env, fake_tmux): +def test_spawn_refuses_without_mise_file(env, fake_launch): root = env["tmp"] / "proj" root.mkdir(parents=True, exist_ok=True) _register_project(root) with pytest.raises(spawn.SpawnError, match="no mise config"): + spawn.spawn("proj", "api", task="do it") + + +def test_spawn_refuses_without_task(env, fake_launch): + root = env["tmp"] / "proj" + _write_mise(root) + _register_project(root) + with pytest.raises(spawn.SpawnError, match="requires a task"): spawn.spawn("proj", "api") + # Fail-fast: no orphaned agent row behind the refused spawn. + with get_engine().begin() as conn: + assert repo.get_agent_by_name(conn, "proj", "api") is None + assert fake_launch == [] -def test_spawn_accepts_dotless_mise_toml(env, fake_tmux): +def test_spawn_accepts_dotless_mise_toml(env, fake_launch): # mise also reads `mise.toml` (no leading dot); the gate must honor it too. root = env["tmp"] / "proj" root.mkdir(parents=True, exist_ok=True) (root / "mise.toml").write_text("[tasks.test]\nrun = 'pytest'\n") _register_project(root) - agent = spawn.spawn("proj", "api") + agent = spawn.spawn("proj", "api", task="do it") with get_engine().begin() as conn: assert repo.get_agent_by_name(conn, "proj", "api")["id"] == agent["id"] -def test_spawn_creates_agent_settings_and_session(env, fake_tmux): +def test_spawn_creates_agent_settings_and_run(env, fake_launch): root = env["tmp"] / "proj" _write_mise(root, with_test=True) _register_project(root) @@ -64,66 +79,99 @@ def test_spawn_creates_agent_settings_and_session(env, fake_tmux): with get_engine().begin() as conn: assert repo.get_agent_by_name(conn, "proj", "api")["id"] == agent["id"] - # settings.json wires all four hook events. + # settings.json wires all four hook events AND the headless permission allowlist + # (claude -p auto-denies anything that would prompt; the allowlist is what lets + # normal work proceed — the hooks stay the hard gate). settings = json.loads((root / ".claude" / "settings.json").read_text()) assert set(settings["hooks"]) == {"Stop", "SessionEnd", "PreToolUse", "Notification"} pre = settings["hooks"]["PreToolUse"][0] assert pre["matcher"] == "AskUserQuestion|Bash" assert "handler.hooks pre_tool_use" in pre["hooks"][0]["command"] + assert settings["permissions"]["defaultMode"] == "acceptEdits" + assert "Bash(git *)" in settings["permissions"]["allow"] - # tmux session named project__agent, with identity + DATABASE_URL in env. - call = fake_tmux["calls"]["new_session"][0] - assert call["name"] == "proj__api" + # A headless run launched with identity + DATABASE_URL in env and the task as prompt. + call = fake_launch[0] + assert call["kind"] == "spawn" + assert call["prompt"] == "build the thing" assert call["env"]["HANDLER_PROJECT_ID"] == "proj" assert call["env"]["HANDLER_AGENT_NAME"] == "api" assert call["env"]["HANDLER_AGENT_ID"] == str(agent["id"]) assert call["env"]["DATABASE_URL"] == env["url"] + # The run row + session id landed on the agent. + with get_engine().begin() as conn: + row = repo.get_agent_by_name(conn, "proj", "api") + assert row["session_id"] == call["run"]["session_id"] + assert repo.get_latest_run(conn, row["id"])["kind"] == "spawn" -def test_spawn_mise_init_skips_test_gate_and_marks_env(env, fake_tmux): +def test_spawn_mise_init_skips_test_gate_and_marks_env(env, fake_launch): # A repo with no .mise.toml at all: the normal gate would refuse, but the mise-init # bootstrap agent must launch anyway (creating that file is its whole job). root = env["tmp"] / "proj" root.mkdir(parents=True, exist_ok=True) _register_project(root) - agent = spawn.spawn("proj", "mise-init", require_tests=False, mise_init=True) + agent = spawn.spawn( + "proj", "mise-init", task="write the mise config", require_tests=False, mise_init=True + ) with get_engine().begin() as conn: assert repo.get_agent_by_name(conn, "proj", "mise-init")["id"] == agent["id"] - # The launched session carries HANDLER_MISE_INIT so its hooks enforce commit + push. - call = fake_tmux["calls"]["new_session"][0] - assert call["env"]["HANDLER_MISE_INIT"] == "1" + # The launched run carries HANDLER_MISE_INIT so its hooks enforce commit + push. + assert fake_launch[0]["env"]["HANDLER_MISE_INIT"] == "1" -def test_spawn_still_gates_without_mise_init_flag(env, fake_tmux): +def test_spawn_still_gates_without_mise_init_flag(env, fake_launch): root = env["tmp"] / "proj" root.mkdir(parents=True, exist_ok=True) _register_project(root) # require_tests defaults on, so a normal spawn against a mise-less repo still refuses. with pytest.raises(spawn.SpawnError, match="no mise config"): - spawn.spawn("proj", "api") - assert fake_tmux["calls"]["new_session"] == [] + spawn.spawn("proj", "api", task="do it") + assert fake_launch == [] -def test_kill_sets_done_and_kills_session(env, fake_tmux): +def test_kill_cancels_run_and_sets_done(env, fake_launch): root = env["tmp"] / "proj" _write_mise(root, with_test=True) _register_project(root) - spawn.spawn("proj", "api") + spawn.spawn("proj", "api", task="do it") spawn.kill("proj", "api") - assert "proj__api" in fake_tmux["calls"]["kill_session"] with get_engine().begin() as conn: - assert repo.get_agent_by_name(conn, "proj", "api")["status"] == "done" + agent = repo.get_agent_by_name(conn, "proj", "api") + assert agent["status"] == "done" + # The running run was flagged; the owning supervisor terminates its own child. + assert repo.get_latest_run(conn, agent["id"])["cancel_requested"] is True -def test_resume_sends_answer_to_live_session(env, fake_tmux): +def test_resume_reinjects_when_no_transcript(env, fake_launch): + """A resume with no archive and no local transcript degrades to a fresh run whose + prompt carries the operator's answer (context re-injection).""" root = env["tmp"] / "proj" _write_mise(root, with_test=True) _register_project(root) - agent = spawn.spawn("proj", "api") + spawn.spawn("proj", "api", task="do it") + with get_engine().begin() as conn: + agent = repo.get_agent_by_name(conn, "proj", "api") + repo.finish_run(conn, repo.get_latest_run(conn, agent["id"])["id"], "completed") ok, detail = spawn.resume(agent, "use Postgres") assert ok is True - assert fake_tmux["calls"]["send_keys"][0] == {"name": "proj__api", "keys": "use Postgres"} + assert "re-injected" in detail + assert fake_launch[-1]["kind"] == "spawn" + assert "use Postgres" in fake_launch[-1]["prompt"] + + +def test_resume_refused_while_run_live(env, fake_launch): + root = env["tmp"] / "proj" + _write_mise(root, with_test=True) + _register_project(root) + spawn.spawn("proj", "api", task="do it") # fake run stays 'running' + with get_engine().begin() as conn: + agent = repo.get_agent_by_name(conn, "proj", "api") + + ok, detail = spawn.resume(agent, "answer") + assert ok is False + assert "live run" in detail diff --git a/tests/test_control_spawn_forge.py b/tests/test_control_spawn_forge.py index b50404c..8d29ff3 100644 --- a/tests/test_control_spawn_forge.py +++ b/tests/test_control_spawn_forge.py @@ -19,15 +19,15 @@ def _register(root, **kw): repo.create_project(conn, "proj", str(root), **kw) -def test_spawn_injects_credentials_and_installs_helper(env, fake_tmux, fake_gitops, monkeypatch): +def test_spawn_injects_credentials_and_installs_helper(env, fake_launch, fake_gitops, monkeypatch): monkeypatch.setenv("PROJ_TOKEN", "s3cret") root = env["tmp"] / "proj" _write_mise(root) _register(root, git_remote="https://github.com/me/proj.git", credential_ref="env:PROJ_TOKEN") - spawn.spawn("proj", "junior", role="junior") + spawn.spawn("proj", "junior", role="junior", task="do it") - call = fake_tmux["calls"]["new_session"][0] + call = fake_launch[0] # Token injected under the generic + host-specific names, never the raw ref stored. assert call["env"]["FORGE_TOKEN"] == "s3cret" assert call["env"]["GITHUB_TOKEN"] == "s3cret" @@ -38,43 +38,43 @@ def test_spawn_injects_credentials_and_installs_helper(env, fake_tmux, fake_gito assert "$FORGE_TOKEN" in helper[0]["value"] -def test_spawn_ssh_remote_installs_no_https_helper(env, fake_tmux, fake_gitops, monkeypatch): +def test_spawn_ssh_remote_installs_no_https_helper(env, fake_launch, fake_gitops, monkeypatch): monkeypatch.setenv("PROJ_TOKEN", "s3cret") root = env["tmp"] / "proj" _write_mise(root) _register(root, git_remote="git@github.com:me/proj.git", credential_ref="env:PROJ_TOKEN") - spawn.spawn("proj", "junior", role="junior") + spawn.spawn("proj", "junior", role="junior", task="do it") # ssh remote -> token still injected, but no HTTPS credential helper installed. - assert fake_tmux["calls"]["new_session"][0]["env"]["GITHUB_TOKEN"] == "s3cret" + assert fake_launch[0]["env"]["GITHUB_TOKEN"] == "s3cret" assert fake_gitops["config"] == [] -def test_spawn_fails_fast_on_broken_credential_ref(env, fake_tmux, fake_gitops, monkeypatch): +def test_spawn_fails_fast_on_broken_credential_ref(env, fake_launch, fake_gitops, monkeypatch): monkeypatch.delenv("ABSENT_TOKEN", raising=False) root = env["tmp"] / "proj" _write_mise(root) _register(root, credential_ref="env:ABSENT_TOKEN") with pytest.raises(spawn.SpawnError, match="not set"): - spawn.spawn("proj", "junior", role="junior") + spawn.spawn("proj", "junior", role="junior", task="do it") # No agent row and no session left behind by the failed spawn. with get_engine().begin() as conn: assert repo.get_agent_by_name(conn, "proj", "junior") is None - assert fake_tmux["calls"]["new_session"] == [] + assert fake_launch == [] -def test_spawn_without_credential_ref_injects_no_token(env, fake_tmux, fake_gitops): +def test_spawn_without_credential_ref_injects_no_token(env, fake_launch, fake_gitops): root = env["tmp"] / "proj" _write_mise(root) _register(root) - spawn.spawn("proj", "api") - call = fake_tmux["calls"]["new_session"][0] + spawn.spawn("proj", "api", task="do it") + call = fake_launch[0] assert "FORGE_TOKEN" not in call["env"] # No token -> no credential helper installed. assert fake_gitops["config"] == [] -def test_spawn_reports_forge_version_mismatch(env, fake_tmux, fake_gitops, fake_forge, monkeypatch): +def test_spawn_reports_forge_version_mismatch(env, fake_launch, fake_gitops, fake_forge, monkeypatch): monkeypatch.setenv("FORGE_VERSION", "9.9.9") from handler import config from handler.db import engine @@ -88,5 +88,5 @@ def test_spawn_reports_forge_version_mismatch(env, fake_tmux, fake_gitops, fake_ fake_forge["version_ok"] = False fake_forge["version_out"] = "forge 1.2.3" - agent = spawn.spawn("proj", "api") + agent = spawn.spawn("proj", "api", task="do it") assert "9.9.9" in agent["forge_note"] diff --git a/tests/test_credsync.py b/tests/test_credsync.py index cb748c4..f7bb8bb 100644 --- a/tests/test_credsync.py +++ b/tests/test_credsync.py @@ -101,10 +101,3 @@ def test_disabled_without_secret_key(env, tmp_path): _write_local_credentials(tmp_path) assert credsync.upload() is False assert credsync.refresh() is None - - -def test_note_local_write_suppresses_upload(secret_env, tmp_path): - _write_local_credentials(tmp_path) - credsync.note_local_write() - # The deliberate local write (e.g. ensure_onboarded at spawn) is not re-published. - assert credsync.refresh() is None diff --git a/tests/test_headless_run.py b/tests/test_headless_run.py index ad9600c..e413ccf 100644 --- a/tests/test_headless_run.py +++ b/tests/test_headless_run.py @@ -26,7 +26,6 @@ def headless_env(env, monkeypatch): from handler import config monkeypatch.setenv("CLAUDE_BIN", FAKE_CLAUDE) - monkeypatch.setenv("RUNNER", "headless") config.get_settings.cache_clear() yield env config.get_settings.cache_clear() @@ -250,16 +249,13 @@ def test_resume_refused_while_run_live(headless_env, tmp_path, monkeypatch): _wait_for(_finished_run(run["id"]), timeout=30.0) -def test_headless_settings_include_permissions(headless_env, tmp_path): - path = settings_gen.write_settings(str(tmp_path / "wd"), headless=True) +def test_settings_include_permissions_and_hooks(headless_env, tmp_path): + path = settings_gen.write_settings(str(tmp_path / "wd")) data = json.loads(Path(path).read_text()) assert data["permissions"]["defaultMode"] == "acceptEdits" assert "Bash(git *)" in data["permissions"]["allow"] assert "hooks" in data # the hard gate is untouched - tmux_path = settings_gen.write_settings(str(tmp_path / "wd2")) - assert "permissions" not in json.loads(Path(tmux_path).read_text()) - def test_api_rejects_empty_task_headless_spawn(headless_env, client, auth): client.post( diff --git a/tests/test_integration_web_spawn.py b/tests/test_integration_web_spawn.py index 06ff58b..01b7dae 100644 --- a/tests/test_integration_web_spawn.py +++ b/tests/test_integration_web_spawn.py @@ -1,13 +1,32 @@ """End-to-end web management: the dashboard's HTTP calls -> command queue -> worker -> -real ``spawn.spawn`` -> tmux seam. Proves the full container-split flow works with only the -tmux/claude boundary faked, not the control layer itself.""" +real ``spawn.spawn`` -> a real headless subprocess (the fake claude binary). Proves the +full container-split flow works with only the claude binary faked, not the control +layer: events stream into the DB, the run reconciles, kill cancels.""" from __future__ import annotations +import time +from pathlib import Path + +import pytest + from handler.control import worker from handler.db import repository as repo from handler.db.engine import get_engine +REPO_ROOT = Path(__file__).resolve().parents[1] +FAKE_CLAUDE = str(REPO_ROOT / "tests" / "fixtures" / "fake_claude.py") + + +@pytest.fixture +def headless_env(env, monkeypatch): + from handler import config + + monkeypatch.setenv("CLAUDE_BIN", FAKE_CLAUDE) + config.get_settings.cache_clear() + yield env + config.get_settings.cache_clear() + def _spawnable_project(root): root.mkdir(parents=True, exist_ok=True) @@ -16,10 +35,21 @@ def _spawnable_project(root): repo.create_project(conn, "proj", str(root)) -def test_spawn_via_api_then_worker_creates_agent_and_session(client, auth, env, fake_tmux): - _spawnable_project(env["tmp"] / "proj") +def _wait(predicate, timeout=20.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + result = predicate() + if result: + return result + time.sleep(0.1) + return None - # 1. The dashboard enqueues a spawn (202 + a queued command). + +def test_spawn_via_api_then_worker_runs_headless_claude(client, auth, headless_env): + _spawnable_project(headless_env["tmp"] / "proj") + + # 1. The dashboard enqueues a spawn (202 + a queued command). A task is mandatory — + # headless claude has no idle-REPL mode. r = client.post( "/projects/proj/agents/spawn", json={"name": "api", "task": "build the thing"}, @@ -32,29 +62,62 @@ def test_spawn_via_api_then_worker_creates_agent_and_session(client, auth, env, # No agent yet — the worker hasn't run. assert client.get("/projects/proj/agents", headers=auth).json() == [] - # 2. The control worker drains the queue (runs the real spawn.spawn). + # 2. The control worker drains the queue (real spawn.spawn -> real subprocess). assert worker.drain("test-worker") == 1 - # 3. The command is done and the agent + tmux session now exist. + # 3. The command finished at launch (fire-and-forget)... got = client.get(f"/commands/{command_id}", headers=auth).json() assert got["status"] == "done" assert got["result"]["name"] == "api" - agents = client.get("/projects/proj/agents", headers=auth).json() - assert [a["name"] for a in agents] == ["api"] - assert fake_tmux["calls"]["new_session"][0]["name"] == "proj__api" + # ...and the run's whole life shows up via the API: events stream in, the agent + # reconciles to done, last_output is the assistant's text. + def finished(): + agents = client.get("/projects/proj/agents", headers=auth).json() + return agents if agents and agents[0]["status"] == "done" else None + + agents = _wait(finished) + assert agents is not None, "run never reconciled to done" + agent = agents[0] + assert agent["name"] == "api" + assert agent["session_id"] + assert agent["worker_id"] == "test-worker" + assert agent["last_output"] == "working on: build the thing" + + events = client.get("/projects/proj/agents/api/events", headers=auth).json() + assert [e["type"] for e in events] == ["system", "assistant", "result"] -def test_kill_via_api_then_worker(client, auth, env, fake_tmux): - _spawnable_project(env["tmp"] / "proj") - client.post("/projects/proj/agents/spawn", json={"name": "api"}, headers=auth) +def test_spawn_without_task_is_rejected(client, auth, headless_env): + _spawnable_project(headless_env["tmp"] / "proj") + r = client.post("/projects/proj/agents/spawn", json={"name": "api"}, headers=auth) + assert r.status_code == 400 + assert "task is required" in r.json()["detail"] + + +def test_kill_via_api_then_worker(client, auth, headless_env, monkeypatch): + monkeypatch.setenv("FAKE_CLAUDE_MODE", "hang") + _spawnable_project(headless_env["tmp"] / "proj") + client.post( + "/projects/proj/agents/spawn", json={"name": "api", "task": "hang"}, headers=auth + ) worker.drain("w") + # The hanging run is live; kill flags it and the supervisor SIGTERMs its child. r = client.post("/projects/proj/agents/api/kill", headers=auth) assert r.status_code == 202 worker.drain("w") assert client.get(f"/commands/{r.json()['id']}", headers=auth).json()["status"] == "done" - assert "proj__api" in fake_tmux["calls"]["kill_session"] with get_engine().begin() as conn: - assert repo.get_agent_by_name(conn, "proj", "api")["status"] == "done" + agent = repo.get_agent_by_name(conn, "proj", "api") + assert agent["status"] == "done" + + def canceled(): + with get_engine().begin() as conn: + run = repo.get_latest_run(conn, agent["id"]) + return run if run["status"] != "running" else None + + run = _wait(canceled, timeout=30.0) + assert run is not None, "kill never terminated the hanging run" + assert run["status"] == "canceled" diff --git a/tests/test_worker.py b/tests/test_worker.py index 6ac847f..135570d 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -229,45 +229,6 @@ def test_bad_command_is_recorded_failed_not_raised(env): assert "agent name" in failed["error"] -def test_capture_agent_output_snapshots_working_agents(env, monkeypatch): - with get_engine().begin() as conn: - repo.create_project(conn, "p", "/tmp/p") - agent = repo.create_agent(conn, "p", "api", "/tmp/p/api", status="working") - - monkeypatch.setattr(worker.tmux, "has_session", lambda name: True) - monkeypatch.setattr( - worker.tmux, "capture_pane", lambda name, escapes=False: "boot\nTheme picker\n\n\n" - ) - - assert worker.capture_agent_output() == 1 - with get_engine().begin() as conn: - row = repo.get_agent_by_id(conn, agent["id"]) - # The tail is stored with trailing blank lines trimmed. - assert row["last_output"] == "boot\nTheme picker" - assert row["output_at"] is not None - - -def test_capture_agent_output_skips_dead_sessions_and_nonworking(env, monkeypatch): - with get_engine().begin() as conn: - repo.create_project(conn, "p", "/tmp/p") - repo.create_agent(conn, "p", "gone", "/tmp/p/gone", status="working") - done = repo.create_agent(conn, "p", "done", "/tmp/p/done", status="done") - - captured = [] - monkeypatch.setattr(worker.tmux, "has_session", lambda name: False) - monkeypatch.setattr( - worker.tmux, - "capture_pane", - lambda name, escapes=False: captured.append(name) or "x", - ) - - # The working agent's session is dead (skipped); the done agent isn't queried at all. - assert worker.capture_agent_output() == 0 - assert captured == [] - with get_engine().begin() as conn: - assert repo.get_agent_by_id(conn, done["id"])["last_output"] is None - - def test_drain_processes_multiple_then_stops(env, monkeypatch): _seed_project() monkeypatch.setattr(poller, "sweep", lambda project_id=None: {"checked": 0}) diff --git a/tests/test_worker_concurrency.py b/tests/test_worker_concurrency.py index b96fe9d..9e9d0e2 100644 --- a/tests/test_worker_concurrency.py +++ b/tests/test_worker_concurrency.py @@ -16,7 +16,6 @@ from handler.db.engine import get_engine def headless_env(env, monkeypatch): from handler import config - monkeypatch.setenv("RUNNER", "headless") monkeypatch.setenv("MAX_CONCURRENT_RUNS", "2") config.get_settings.cache_clear() yield env @@ -75,7 +74,3 @@ def test_slot_frees_when_run_finishes(headless_env, monkeypatch): with get_engine().begin() as conn: repo.finish_run(conn, run1["id"], "completed", exit_code=0) assert worker._full_slot_exclusions("w1") == () - - -def test_tmux_runner_never_excludes(env): - assert worker._full_slot_exclusions("w") == ()