feat(control): flag-gated headless runner with cross-worker resume (phase 2)

Wires the phase-1 headless machinery behind runner=headless (default
stays tmux; legacy agents, session_id null, keep the tmux paths):

- spawn: branches tmux vs headless.launch; extracts _agent_env (shared
  with resume - a headless resume is a new process needing identity/
  credential env); headless spawns require a task (no idle-REPL mode),
  enforced at spawn and as a 400 in the API
- resume: headless path materializes the session archive from the DB
  onto whichever worker claimed the command, then claude -p --resume;
  falls back to a fresh session with DB-re-injected context (visible
  worker event) when no transcript survives anywhere; refuses while a
  run is live. Undeliverable resumes now raise -> command FAILED,
  fixing silent input loss on both runners
- kill: headless path flags cancel_requested; the owning supervisor
  SIGTERMs its own child (cross-worker safe)
- worker: stable per-container ids, DB-driven run slots (full workers
  skip claiming spawn/resume/mise_init, leaving them for less-loaded
  workers), credsync refresh in the main loop
- settings_gen: permissions block (defaultMode + allowlist) for
  headless runs - -p auto-denies anything that would prompt; hooks
  remain the hard gate
- credsync + migration 0009 (runtime_secrets): login publishes the
  Fernet-encrypted claude credential bundle; every worker materializes
  it (merge-safe for local trust state); login_submit pinned to the
  login_start worker via commands.target_worker

Suite 270 -> 290 green, including the cross-worker resume linchpin
(clean-HOME materialize + --resume against the fake binary).

tests/test_credsync.py

@@ -0,0 +1,110 @@
+"""Credential distribution through runtime_secrets: the login completes on one worker,
+every other worker materializes the encrypted bundle from the DB — no shared files."""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+
+import pytest
+from cryptography.fernet import Fernet
+
+from handler.control import credsync
+from handler.db import repository as repo
+from handler.db.engine import get_engine
+
+
+@pytest.fixture
+def secret_env(env, monkeypatch):
+    from handler import config
+
+    monkeypatch.setenv("HANDLER_SECRET_KEY", Fernet.generate_key().decode())
+    config.get_settings.cache_clear()
+    credsync._state.__init__()  # fresh sync cursor per test
+    yield env
+    config.get_settings.cache_clear()
+
+
+def _write_local_credentials(home: Path) -> None:
+    (home / ".claude").mkdir(parents=True, exist_ok=True)
+    (home / ".claude.json").write_text(
+        json.dumps({"oauthAccount": {"email": "op@example.com"}, "theme": "light"})
+    )
+    (home / ".claude" / ".credentials.json").write_text('{"token": "secret-oauth-token"}')
+
+
+def test_upload_stores_encrypted_bundle(secret_env, tmp_path):
+    _write_local_credentials(tmp_path)
+    assert credsync.upload() is True
+    with get_engine().begin() as conn:
+        row = repo.get_runtime_secret(conn, credsync.SECRET_KEY)
+    assert row is not None
+    # Ciphertext at rest — the raw token must not appear in the DB value.
+    assert "secret-oauth-token" not in row["value_enc"]
+
+
+def test_refresh_materializes_on_fresh_worker(secret_env, tmp_path, monkeypatch):
+    _write_local_credentials(tmp_path)
+    assert credsync.upload() is True
+
+    other_home = tmp_path / "worker-b"
+    other_home.mkdir()
+    monkeypatch.setenv("HOME", str(other_home))
+    credsync._state.__init__()  # worker B's process state
+
+    assert credsync.refresh() == "materialized"
+    creds = json.loads((other_home / ".claude" / ".credentials.json").read_text())
+    assert creds["token"] == "secret-oauth-token"
+    data = json.loads((other_home / ".claude.json").read_text())
+    assert data["oauthAccount"]["email"] == "op@example.com"
+    # A second pass is a no-op — nothing changed anywhere.
+    assert credsync.refresh() is None
+
+
+def test_materialize_merges_claude_json_preserving_local_state(secret_env, tmp_path, monkeypatch):
+    _write_local_credentials(tmp_path)
+    credsync.upload()
+
+    other_home = tmp_path / "worker-c"
+    (other_home / ".claude").mkdir(parents=True)
+    (other_home / ".claude.json").write_text(
+        json.dumps(
+            {
+                "hasCompletedOnboarding": True,
+                "theme": "dark",
+                "projects": {"/projects/p/a": {"hasTrustDialogAccepted": True}},
+            }
+        )
+    )
+    monkeypatch.setenv("HOME", str(other_home))
+    credsync._state.__init__()
+
+    assert credsync.refresh() == "materialized"
+    data = json.loads((other_home / ".claude.json").read_text())
+    # Account arrived...
+    assert data["oauthAccount"]["email"] == "op@example.com"
+    # ...but this worker's own onboarding/trust state (claude_config's writes) survived.
+    assert data["theme"] == "dark"
+    assert data["projects"]["/projects/p/a"]["hasTrustDialogAccepted"] is True
+
+
+def test_refresh_uploads_local_change(secret_env, tmp_path):
+    _write_local_credentials(tmp_path)
+    assert credsync.refresh() == "uploaded"  # bootstrap: local creds, empty DB
+    # A token refresh on disk (mtime/size change) re-publishes.
+    os.utime(tmp_path / ".claude" / ".credentials.json", ns=(1, 1))
+    assert credsync.refresh() == "uploaded"
+
+
+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

tests/test_headless_run.py

@@ -0,0 +1,272 @@
+"""End-to-end headless runner tests against the fake ``claude`` binary.
+
+Real subprocesses, real threads, real SQLite: ``headless.launch`` starts
+``tests/fixtures/fake_claude.py`` (selected via the ``claude_bin`` setting, the same
+seam the tmux fakes used), the supervisor streams its stdout into ``agent_events``, and
+the tests assert on what landed in the DB — exactly what the API/UI will read."""
+
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+
+import pytest
+
+from handler.control import headless, settings_gen, spawn
+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)
+    monkeypatch.setenv("RUNNER", "headless")
+    config.get_settings.cache_clear()
+    yield env
+    config.get_settings.cache_clear()
+
+
+def _make_agent(tmp_path, name="h1"):
+    working_dir = tmp_path / "projects" / "p" / name
+    working_dir.mkdir(parents=True)
+    with get_engine().begin() as conn:
+        repo.create_project(conn, "p", str(tmp_path / "projects" / "p"))
+        agent = repo.create_agent(conn, "p", name, str(working_dir))
+    return agent
+
+
+def _wait_for(predicate, timeout=20.0):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        result = predicate()
+        if result:
+            return result
+        time.sleep(0.1)
+    return None
+
+
+def _finished_run(run_id):
+    def check():
+        with get_engine().begin() as conn:
+            run = repo.get_run(conn, run_id)
+        return run if run["status"] != "running" else None
+
+    return check
+
+
+def test_spawn_run_streams_events_and_completes(headless_env, tmp_path):
+    agent = _make_agent(tmp_path)
+    run = headless.launch(
+        agent, kind="spawn", prompt="build the thing",
+        settings_path=str(tmp_path / "settings.json"), env={}, worker_id="w1",
+    )
+    finished = _wait_for(_finished_run(run["id"]))
+    assert finished is not None, "run never finished"
+    assert finished["status"] == "completed"
+    assert finished["exit_code"] == 0
+    assert finished["result"]["is_error"] is False
+
+    with get_engine().begin() as conn:
+        events = repo.list_agent_events(conn, agent["id"])
+        updated = repo.get_agent_by_id(conn, agent["id"])
+        archive = repo.get_session_archive(conn, agent["id"])
+
+    types = [e["type"] for e in events]
+    assert types == ["system", "assistant", "result"]
+    assert [e["seq"] for e in events] == [1, 2, 3]
+    # last_output is now derived from assistant text — the log the UI shows is real.
+    assert updated["last_output"] == "working on: build the thing"
+    assert updated["status"] == "done"
+    assert updated["session_id"] == run["session_id"]
+    assert updated["worker_id"] == "w1"
+    # The session archive was uploaded at exit for cross-worker resume.
+    assert archive is not None
+    assert archive["session_id"] == run["session_id"]
+
+
+def test_failed_run_marks_blocked_with_worker_event(headless_env, tmp_path, monkeypatch):
+    monkeypatch.setenv("FAKE_CLAUDE_MODE", "error")
+    agent = _make_agent(tmp_path, "h-err")
+    run = headless.launch(
+        agent, kind="spawn", prompt="boom",
+        settings_path=str(tmp_path / "s.json"), env={}, worker_id="w1",
+    )
+    finished = _wait_for(_finished_run(run["id"]))
+    assert finished["status"] == "failed"
+    assert finished["exit_code"] == 2
+
+    with get_engine().begin() as conn:
+        events = repo.list_agent_events(conn, agent["id"])
+        updated = repo.get_agent_by_id(conn, agent["id"])
+    types = [e["type"] for e in events]
+    # The unparseable stdout line is preserved verbatim as a raw event, and the runner
+    # records why the run failed as a worker event.
+    assert "raw" in types
+    raw = next(e for e in events if e["type"] == "raw")
+    assert "this is not json" in raw["payload"]["line"]
+    worker_ev = next(e for e in events if e["type"] == "worker")
+    assert worker_ev["payload"]["exit_code"] == 2
+    assert updated["status"] == "blocked"
+
+
+def test_hook_written_status_survives_reconciliation(headless_env, tmp_path, monkeypatch):
+    """Hooks are the status authority: if one set paused_for_input during the run, the
+    supervisor's exit pass must not overwrite it with done."""
+    monkeypatch.setenv("FAKE_CLAUDE_MODE", "slow")
+    monkeypatch.setenv("FAKE_CLAUDE_SLOW_SECONDS", "1.5")
+    agent = _make_agent(tmp_path, "h-hook")
+    run = headless.launch(
+        agent, kind="spawn", prompt="ask me something",
+        settings_path=str(tmp_path / "s.json"), env={}, worker_id="w1",
+    )
+    # Simulate a hook (inside the run) recording an open question.
+    with get_engine().begin() as conn:
+        repo.set_agent_status(conn, agent["id"], "paused_for_input")
+    finished = _wait_for(_finished_run(run["id"]))
+    assert finished["status"] == "completed"
+    with get_engine().begin() as conn:
+        assert repo.get_agent_by_id(conn, agent["id"])["status"] == "paused_for_input"
+
+
+def test_cancel_terminates_hanging_run(headless_env, tmp_path, monkeypatch):
+    monkeypatch.setenv("FAKE_CLAUDE_MODE", "hang")
+    agent = _make_agent(tmp_path, "h-hang")
+    run = headless.launch(
+        agent, kind="spawn", prompt="hang forever",
+        settings_path=str(tmp_path / "s.json"), env={}, worker_id="w1",
+    )
+    # Give the supervisor a moment to start the process, then flag the cancel the same
+    # way a cross-worker kill would.
+    _wait_for(lambda: _events_count(agent["id"]) >= 1)
+    with get_engine().begin() as conn:
+        assert repo.request_run_cancel(conn, run["id"]) is True
+    finished = _wait_for(_finished_run(run["id"]), timeout=30.0)
+    assert finished is not None, "cancel never terminated the run"
+    assert finished["status"] == "canceled"
+
+
+def _events_count(agent_id):
+    with get_engine().begin() as conn:
+        return len(repo.list_agent_events(conn, agent_id))
+
+
+def test_kill_headless_agent_requests_cancel(headless_env, tmp_path, monkeypatch):
+    monkeypatch.setenv("FAKE_CLAUDE_MODE", "hang")
+    agent = _make_agent(tmp_path, "h-kill")
+    run = headless.launch(
+        agent, kind="spawn", prompt="hang",
+        settings_path=str(tmp_path / "s.json"), env={}, worker_id="w1",
+    )
+    _wait_for(lambda: _events_count(agent["id"]) >= 1)
+    spawn.kill("p", "h-kill")
+    finished = _wait_for(_finished_run(run["id"]), timeout=30.0)
+    assert finished["status"] == "canceled"
+    with get_engine().begin() as conn:
+        assert repo.get_agent_by_id(conn, agent["id"])["status"] == "done"
+
+
+def test_cross_worker_resume_materializes_archive(headless_env, tmp_path, monkeypatch):
+    """The linchpin: worker B resumes a session it never ran, from the DB archive alone."""
+    agent = _make_agent(tmp_path, "h-resume")
+    run = headless.launch(
+        agent, kind="spawn", prompt="first pass",
+        settings_path=str(tmp_path / "s.json"), env={}, worker_id="worker-a",
+    )
+    assert _wait_for(_finished_run(run["id"]))["status"] == "completed"
+
+    # "Worker B": a clean HOME with no local claude state at all.
+    other_home = tmp_path / "worker-b-home"
+    other_home.mkdir()
+    monkeypatch.setenv("HOME", str(other_home))
+
+    with get_engine().begin() as conn:
+        agent = repo.get_agent_by_id(conn, agent["id"])  # refetch: has session_id now
+    ok, detail = spawn.resume(agent, "the operator's answer", worker_id="worker-b")
+    assert ok, detail
+
+    with get_engine().begin() as conn:
+        resumed = repo.get_latest_run(conn, agent["id"])
+    assert resumed["kind"] == "resume"
+    assert resumed["worker_id"] == "worker-b"
+    finished = _wait_for(_finished_run(resumed["id"]))
+    # fake_claude exits 3 if the transcript was NOT materialized where claude looks.
+    assert finished["status"] == "completed", f"exit={finished['exit_code']}"
+    assert finished["session_id"] == run["session_id"]  # same session, continued
+
+
+def test_resume_without_transcript_reinjects_context(headless_env, tmp_path, monkeypatch):
+    """Owning worker died before its first archive: resume degrades to a fresh session
+    with DB-rebuilt context, visibly marked as such."""
+    agent = _make_agent(tmp_path, "h-fallback")
+    with get_engine().begin() as conn:
+        repo.set_agent_session(conn, agent["id"], "lost-session-uuid", "worker-dead")
+        repo.upsert_checkmark_row(
+            conn, agent["id"], status="paused_for_input",
+            where_it_stopped="mid-refactor", open_question="which db?",
+        )
+        agent = repo.get_agent_by_id(conn, agent["id"])
+
+    ok, detail = spawn.resume(agent, "use postgres", worker_id="worker-b")
+    assert ok
+    assert "re-injected" in detail
+
+    with get_engine().begin() as conn:
+        new_run = repo.get_latest_run(conn, agent["id"])
+        events = repo.list_agent_events(conn, agent["id"])
+    assert new_run["kind"] == "spawn"  # genuinely new session
+    assert new_run["session_id"] != "lost-session-uuid"
+    notice = next(e for e in events if e["type"] == "worker")
+    assert notice["payload"]["previous_session_id"] == "lost-session-uuid"
+    finished = _wait_for(_finished_run(new_run["id"]))
+    assert finished["status"] == "completed"
+    with get_engine().begin() as conn:
+        assistant = [
+            e for e in repo.list_agent_events(conn, agent["id"]) if e["type"] == "assistant"
+        ]
+    # The re-injected prompt (with the operator's answer) reached the fresh claude.
+    assert any("use postgres" in json.dumps(e["payload"]) for e in assistant)
+
+
+def test_resume_refused_while_run_live(headless_env, tmp_path, monkeypatch):
+    monkeypatch.setenv("FAKE_CLAUDE_MODE", "hang")
+    agent = _make_agent(tmp_path, "h-busy")
+    run = headless.launch(
+        agent, kind="spawn", prompt="hang",
+        settings_path=str(tmp_path / "s.json"), env={}, worker_id="w1",
+    )
+    with get_engine().begin() as conn:
+        agent = repo.get_agent_by_id(conn, agent["id"])
+    ok, detail = spawn.resume(agent, "answer", worker_id="w1")
+    assert not ok
+    assert "live run" in detail
+    with get_engine().begin() as conn:
+        repo.request_run_cancel(conn, run["id"])
+    _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)
+    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(
+        "/projects", json={"id": "p2", "root_dir": "/tmp/p2"}, headers=auth
+    )
+    resp = client.post(
+        "/projects/p2/agents/spawn", json={"name": "idle"}, headers=auth
+    )
+    assert resp.status_code == 400
+    assert "task is required" in resp.json()["detail"]

tests/test_worker.py

@@ -65,8 +65,8 @@ def test_resume_command_feeds_answer_and_sets_working(env, monkeypatch):
     _seed_project("api")
     seen = {}
 
-    def fake_resume(agent, ans):
-        seen.update(name=agent["name"], ans=ans)
+    def fake_resume(agent, ans, worker_id=None):
+        seen.update(name=agent["name"], ans=ans, worker_id=worker_id)
         return True, "ok"
 
     monkeypatch.setattr(spawn, "resume", fake_resume)
@@ -74,11 +74,29 @@ def test_resume_command_feeds_answer_and_sets_working(env, monkeypatch):
 
     worker.drain("w")
     assert _get(cmd["id"])["status"] == "done"
-    assert seen == {"name": "api", "ans": "Postgres"}
+    assert seen == {"name": "api", "ans": "Postgres", "worker_id": "w"}
     with get_engine().begin() as conn:
         assert repo.get_agent_by_name(conn, "p", "api")["status"] == "working"
 
 
+def test_resume_command_fails_loudly_when_undeliverable(env, monkeypatch):
+    """An undeliverable answer must surface as a FAILED command — never a silent 'done'
+    (the original bug: send-keys into a dead tmux pane reported success)."""
+    _seed_project("api")
+    monkeypatch.setattr(
+        spawn, "resume", lambda agent, ans, worker_id=None: (False, "no live session")
+    )
+    cmd = _enqueue(type="resume", project_id="p", agent_name="api", payload={"answer": "x"})
+
+    worker.drain("w")
+    failed = _get(cmd["id"])
+    assert failed["status"] == "failed"
+    assert "no live session" in failed["error"]
+    with get_engine().begin() as conn:
+        # The agent must NOT be flipped to working when nothing was delivered.
+        assert repo.get_agent_by_name(conn, "p", "api")["status"] != "working"
+
+
 def test_approve_command_records_operator_verdict_with_head_sha(env, fake_gitops):
     _seed_project("senior")
     cmd = _enqueue(

tests/test_worker_concurrency.py

@@ -0,0 +1,81 @@
+"""Slot-aware command claiming: a worker at max_concurrent_runs must leave run-starting
+commands queued (for a less-loaded worker) while still processing everything else. Slot
+accounting is DB-driven — this worker's ``running`` agent_runs rows — so it needs no
+in-memory registry and is exercised here without real subprocesses."""
+
+from __future__ import annotations
+
+import pytest
+
+from handler.control import spawn, worker
+from handler.db import repository as repo
+from handler.db.engine import get_engine
+
+
+@pytest.fixture
+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
+    config.get_settings.cache_clear()
+
+
+def _seed(conn_count_running_for=None):
+    with get_engine().begin() as conn:
+        repo.create_project(conn, "p", "/tmp/p")
+        agent = repo.create_agent(conn, "p", "a", "/tmp/p/a")
+        return agent
+
+
+def _running_run(agent_id, worker_id):
+    with get_engine().begin() as conn:
+        return repo.create_run(conn, agent_id, f"sid-{worker_id}", worker_id, "spawn")
+
+
+def test_full_worker_skips_run_commands_but_processes_others(headless_env, monkeypatch):
+    agent = _seed()
+    _running_run(agent["id"], "w-full")
+    _running_run(agent["id"], "w-full")  # 2 running == MAX_CONCURRENT_RUNS
+
+    spawned = {}
+    monkeypatch.setattr(
+        spawn, "spawn",
+        lambda project_id, name, **kw: spawned.update(name=name, **kw)
+        or {"id": 1, "name": name, "working_dir": "/tmp/p/x", "forge_note": None},
+    )
+    with get_engine().begin() as conn:
+        spawn_cmd = repo.enqueue_command(conn, "spawn", project_id="p", agent_name="x")
+        kill_cmd = repo.enqueue_command(conn, "kill", project_id="p", agent_name="a")
+    monkeypatch.setattr(spawn, "kill", lambda p, n: None)
+
+    # The full worker processes the kill but leaves the spawn queued.
+    assert worker.drain("w-full") == 1
+    with get_engine().begin() as conn:
+        assert repo.get_command(conn, kill_cmd["id"])["status"] == "done"
+        assert repo.get_command(conn, spawn_cmd["id"])["status"] == "queued"
+    assert spawned == {}
+
+    # A worker with free slots picks the spawn up.
+    assert worker.drain("w-free") == 1
+    with get_engine().begin() as conn:
+        assert repo.get_command(conn, spawn_cmd["id"])["status"] == "done"
+    assert spawned["name"] == "x"
+    assert spawned["worker_id"] == "w-free"
+
+
+def test_slot_frees_when_run_finishes(headless_env, monkeypatch):
+    agent = _seed()
+    run1 = _running_run(agent["id"], "w1")
+    _running_run(agent["id"], "w1")
+    assert worker._full_slot_exclusions("w1") == worker._RUN_COMMANDS
+
+    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") == ()