From 009a25289af68ea71adc6f41d47e348d0da6cd16 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gabor=20K=C3=B6rber?= Date: Fri, 16 Oct 2026 15:49:32 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=A4=96=20forks:=20launch,=20supervise,=20?= =?UTF-8?q?wait,=20kill=20(block=201.2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../2026-10-16-08-headless-forks-run.md | 38 + plugins/spire/skills/forks/forks.py | 1309 ++++++++++++++++- plugins/spire/skills/forks/templates/fork.md | 19 + plugins/spire/skills/forks/templates/train.md | 11 + .../spire/skills/forks/tests/test_forks.py | 950 +++++++++++- 5 files changed, 2294 insertions(+), 33 deletions(-) create mode 100644 plugins/spire/skills/forks/templates/fork.md create mode 100644 plugins/spire/skills/forks/templates/train.md diff --git a/docs/workpad/reports/2026-10-16-08-headless-forks-run.md b/docs/workpad/reports/2026-10-16-08-headless-forks-run.md index 5854ddc..f6f2580 100644 --- a/docs/workpad/reports/2026-10-16-08-headless-forks-run.md +++ b/docs/workpad/reports/2026-10-16-08-headless-forks-run.md @@ -96,3 +96,41 @@ Gate after 1.1 (Athanor, 12:29): G-unit `Ran 52 tests … OK`, exit 0; G-where e - Left for later: everything the digest assigns to 1.2 (templates, launch, supervise, results, ledger lines, `wait`/`list`/`expire`/`kill`, no-visible-window test, `run.last_orchestrator_call_at`); Phase 2 commands. - Expected broken: `train`, `fork`, `continue`, `wait`, `list`, `expire`, `kill`, `_supervise` exit 2 "not implemented" (block 1.2); `rebase`, `usage`, `tick` (Phase 2). - Commits: `d36ac2f` (code, tests, preamble, this note), then a one-line follow-up filling in this hash. + +### Block 1.2 — Launch and supervise + +- Model: `claude-opus-5-5[1m]` +- Start / end: Fri Oct 16 12:31:01 WEDT 2026 / Fri Oct 16 15:48:45 WEDT 2026 (`date`; the host clock read 12:31 at my first call and 15:42 at the first manual run, so the span includes a jump like the one Phase 0 saw; the gates ran 15:47:38–15:48:30) +- Done: `forks.py` gains `train`, `fork`, `continue`, `wait`, `list`, `expire`, `kill`, `_supervise`. Launch writes the job (registry index, status `created`) under the lock, `briefs/.md` (template + brief) and `raw/.env.json` (scrubbed env with `CLAUDE_CODE_EXECPATH` = stored binary and `SPIRE_FORKS_CHILD=1`), then spawns ` -I forks.py _supervise --run --job` through `powershell` (`CREATE_NO_WINDOW`) → CIM `Win32_Process.Create` with `ShowWindow=0`, command line and cwd passed in `SPIRE_FORKS_CMD`/`SPIRE_FORKS_CWD`; `ReturnValue` ≠ 0 ends the job `failed`/`spawn_failed`. The supervisor redirects its stdio to `raw/.sup.log`, starts `claude` with exactly the env file and `CREATE_NO_WINDOW` (stored absolute binary path, full fingerprinted command line every call, `--session-id` on train and fork, `--resume … --fork-session` on fork, `--resume` only on continue, `--max-budget-usd` passed through), feeds the prompt on stdin, parses the stream (`StreamState`: init, message ids, first/last assistant usage, `system/informational` restriction warning, result), and is the only writer of `jobs/.json` (heartbeat every 10 s). Watchdogs: `max_minutes` → kill tree, `timeout`; `stall_after_s` of no stdout → flag `stalled`; N1 orphan check every 1 s → kill tree, `killed` + `orphaned`; restriction warning → kill tree, `failed`/`model_refused`. Finalising (`finalize_job`, once, under the lock): `results/.json` (plan §4.2 fields + `paths`, `reason`, `model_requested`), ledger line (design §6 + `status`, `flags`, `reason`), registry update (status, cumulative cost and `modelUsage` for the session, base `last_hit`; on train: base `active` with `base_commit`, context tokens, training cost, cumulative `modelUsage`, runtime fingerprint). Usage rule, `usage_mismatch`, cache verdict, `cold_start`, C1 warmth, fingerprint `ok|drift:` (launch: `binary`, `append_system_prompt`, `agent`; runtime: `tools`, `permissionMode`, `plugins`, `agents`, `apiKeySource` against the base's init). `wait` per C4, `list` (all bases with warmth, all jobs, running info), `lost` reaper (heartbeat > 120 s and supervisor pid + ctime dead), `kill` (Toolhelp32 tree snapshot with ctimes → `taskkill /T /F` → sweep with `TerminateProcess` by pid + ctime → verify up to 10 s), `expire` (retire, kill running jobs of the base, flag `expired`). Refusals: N2, N3, cap, launch-field overrides on fork/continue (each field accepted by the parser only to be refused), base not active without `--force`, continue of a running or non-fork job, missing binary (`reason: binary_missing`), base name taken. New `templates/train.md` and `templates/fork.md` (stale-snapshot sentence, delta header from `git --no-optional-locks log --oneline ..HEAD` and `diff --shortstat `, commit rule per mode, note path, logs dir). `preamble.md` unchanged. Tests: 17 new (69 total): a generated stream stub `claude` (`stub/claude.cmd` → base interpreter `-I stub.py`, behaviour per call from `@stub …` lines in the brief, sessions kept beside it so resume/fork report cumulative `modelUsage` and `total_cost_usd`), real CIM launches with the stub as `--binary`, `RuleTests` on the real P0.2 seq-1 numbers, and a module-end check that no process whose command line contains the tests' temp prefix survives. +- Gates: + - G-unit `uv run plugins/spire/skills/forks/tests/test_forks.py` → exit 0, `Ran 69 tests in 52.001s` / `OK` (suite wall 52 s; two earlier full runs at 68 tests: 51.7 s each, OK). + - G-where `uv run plugins/spire/skills/forks/forks.py where` → exit 0, stdout one line, a JSON object, `ok: true`, `spawn: cim-hidden`, binary 2.1.299. + - Leftover processes after the gates: none. Checked by command line (`forks-test-`, `_supervise`, my manual scratch run `b12-manual`) via `Get-CimInstance Win32_Process`, not by name; also asserted by the suite's `tearDownModule`. + - Manual check (scratch run in the session scratchpad, stub binary): the supervisor's parent is `WmiPrvSE.exe`; its tree is python → conhost, cmd (claude) → conhost, python stub → python grandchild; `kill` snapshot 6 pids, `taskkill_rc` 0, no survivors. +- Deviations from plan: + - `fork --max-minutes` and `continue --max-minutes` take a positive float (the tests use 0.06 min); `init --max-minutes` stays an integer. + - `fork`/`continue` also run the `lost` reaper before the cap check (the digest names `wait` and `list`), so a dead supervisor cannot block the cap. + - The `lost` reaper kills the job's leftover `claude` tree (from the job file) before marking it `lost`; the digest only says to mark it. + - On the restriction warning the supervisor kills `claude` at once instead of letting the substituted model finish (saves the spend; the result still reads `failed`/`model_refused`). + - After `claude` exits, the supervisor sweeps any descendants of it that are still alive (the "backgrounded `sleep 240`" trap); logged in `sup.log`. + - `train` takes no launch-field overrides (the design's table allows them; the plan's invariant and the digest do not); they are usage errors there. +- Flagged for review (semantics the digest left open): + - `wait --all` = every job not yet *settled*; a job is settled when it was in the condition of a `wait` that returned 0 or 4. `done` holds only results not delivered before. So the mixed case gives 3 (failed result in `done`), then 4 when the rest finish (the failed job stays in the condition without being returned again), then 0 on an empty set. Explicit job lists use the same delivery marks. `--any`: met when any member has finished (or the set is empty); exit 4 if any finished member is not `ok`. `wait` with neither jobs nor `--all` is a usage error; `--timeout` > 570 is refused (exit 2). + - `reason` vocabulary: `model_refused`, `is_error`, `exit_code`, `no_result`, `spawn_failed`, `launch_error`, `supervisor_error` (failed); `kill`, `expire`, `orphaned` (killed); `supervisor_dead` (lost). `timeout` has `reason: null` (the digest: null unless failed/killed/lost). Details go to the registry's `reason_detail` (shown by `list`), not into the result or ledger. + - Precedence when deciding a finished call: model refused > no result (`exit_code` if non-zero, else `no_result`) > `is_error` > non-zero exit. A model is "refused" when the warning appeared or the requested id (with `[1m]`) is not among the `modelUsage` keys that grew; extra models do not count. + - Without a `result` event (killed, lost, died): `tokens` are all `null`, `cost_delta_usd` `null`, `models_used` `[]`, `turns` = distinct assistant message ids seen; cache verdict still from the first assistant event if one arrived. + - `last_hit` for a train is the time of its **last** assistant event (the full prefix is written there); for fork/continue the first, written to the base as soon as it arrives. A continue refreshes its base's `last_hit`. + - `base_state` is `null` for train jobs; `cold_start` is only computed for fork/continue (a train is always cold). + - `continue` sends the brief as given (no template; there is no `templates/continue.md`). Its `parent_session` is the fork's parent (the base session); its cost and `modelUsage` "before" are the fork session's last cumulative totals (updated by every continue), falling back to the base's when the fork has none (a killed fork: the delta then includes the killed fork's spend). + - `note` = `notes/.md` (block sanitised) if that file exists when the job ends, else `null`; `fork.md` tells the fork that path. Train jobs have `note: null`. + - A failed training leaves the base `failed`; `train` may reuse a name only when that base is `failed`. + - `kill` of a job that is no longer running exits 0 with `killed: false` and its status; survivors after the 10 s sweep give `ok: false` (exit still 0). + - A `kill` that lands while the supervisor holds the registry lock to finalise leaves a stale lock (broken after 30 s) and, in a narrow window, a duplicate ledger line. `kill` marks `kill_requested` first; a supervisor that finalises after that records `killed` unless the call ended `ok`. + - `wait` updates `last_orchestrator_call_at` at its start and end only (a 570 s wait stays under the 1200 s `orphan_after`). + - Stall is measured on any stdout line, not only parsed JSON events. + - `process_tree` follows parent pids through dead processes (so orphans of a dead `claude` are found) but stops at a node whose pid now belongs to a process with another creation time; this reuse guard is not exercised by a test. + - `spawn_failed` shows up in `wait`, not in the launch return (the launch return is C1's fixed shape). +- Not tested: the POSIX path (`setsid` spawn, process-group kill, `/proc` tree) — written, never run; `--agent` drift (only the code path, no test); a real `claude` (no test spends tokens; Phase 1.3 smoke); supervisor survival across the end of the orchestrator's turn (1.3). +- Left for later: warmth inference, pings, `rebase`, `rebase_due`/`retrain_due` reasons, `usage`, `tick` (Phase 2); `SKILL.md` (Phase 3). +- Expected broken: `rebase`, `usage`, `tick` exit 2 "not implemented" (Phase 2). +- Commits: see the follow-up line below. diff --git a/plugins/spire/skills/forks/forks.py b/plugins/spire/skills/forks/forks.py index bfef6de..c8e07d1 100644 --- a/plugins/spire/skills/forks/forks.py +++ b/plugins/spire/skills/forks/forks.py @@ -24,8 +24,11 @@ import platform import posixpath import re import shutil +import signal +import string import subprocess import sys +import threading import time import traceback import uuid @@ -115,6 +118,32 @@ USER_SETTINGS_PATH: str | None = None PHASE2_COMMANDS = ("rebase", "usage", "tick") CHILD_REFUSED = ("train", "fork", "continue", "rebase") # N3 (tick pings: Phase 2) +# Supervision and jobs (1.2). Module-level so tests can shorten the in-process ones. +HEARTBEAT_S = 10.0 # the supervisor writes heartbeat_at at least this often (<= 30 s) +LOST_AFTER_S = 120.0 # heartbeat older than this and a dead supervisor -> `lost` +SUP_TICK_S = 0.25 # supervisor main-loop period +ORPHAN_CHECK_S = 1.0 # how often the supervisor reads run.last_orchestrator_call_at (N1) +WAIT_POLL_S = 2.0 # `wait` poll period +WAIT_DEFAULT_S = 540 +WAIT_MAX_S = 570 +KILL_VERIFY_S = 10.0 # how long `kill` keeps sweeping before it reports survivors +READER_JOIN_S = 10.0 # after claude exits, how long the supervisor waits for the stream reader +CACHE_HIT_RATIO = 0.9 # C5 +RUNNING = ("created", "running") +FINAL = ("ok", "failed", "timeout", "killed", "lost") +LAUNCH_FIELDS = ("model", "effort", "permission_mode", "tools", "allowed_tools", "disallowed_tools", + "agent", "cwd", "binary") # fixed at init; fork/continue refuse them +MU_FIELDS = ("inputTokens", "outputTokens", "cacheReadInputTokens", "cacheCreationInputTokens") +TOKEN_KEYS = ("input", "output", "cache_creation", "cache_creation_1h", "cache_creation_5m", "cache_read") +TEMPLATES_DIR = "templates" + +# CIM hidden spawn (P0.1 rung 3 + ShowWindow=0). Command line and cwd arrive through env vars. +POWERSHELL_CIM = ( + "$si = New-CimInstance -ClassName Win32_ProcessStartup -ClientOnly -Property @{ShowWindow=[uint16]0}; " + "$r = Invoke-CimMethod -ClassName Win32_Process -MethodName Create -Arguments " + "@{CommandLine=$env:SPIRE_FORKS_CMD; CurrentDirectory=$env:SPIRE_FORKS_CWD; ProcessStartupInformation=$si}; " + "\"{0} {1}\" -f $r.ReturnValue, $r.ProcessId") + # -------------------------------------------------------------------------------------- # Errors, output, clock # -------------------------------------------------------------------------------------- @@ -157,6 +186,24 @@ def _iso(t: float | None = None) -> str: return _dt.datetime.fromtimestamp(t).astimezone().isoformat(timespec="seconds") +def _ts(iso) -> float | None: + """Epoch seconds of an ISO string written by `_iso`, or None.""" + if not iso or not isinstance(iso, str): + return None + try: + return _dt.datetime.fromisoformat(iso).timestamp() + except ValueError: + return None + + +def _uniq(items) -> list: + out = [] + for x in items: + if x not in out: + out.append(x) + return out + + def _sha256_bytes(b: bytes) -> str: return hashlib.sha256(b).hexdigest() @@ -453,8 +500,27 @@ if IS_WINDOWS: _k32.CloseHandle.argtypes = [_wt.HANDLE] _k32.CloseHandle.restype = _wt.BOOL _PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 + _PROCESS_TERMINATE = 0x0001 _STILL_ACTIVE = 259 _ERROR_ACCESS_DENIED = 5 + _TH32CS_SNAPPROCESS = 0x00000002 + _INVALID_HANDLE = _wt.HANDLE(-1).value + + class _PE32(ctypes.Structure): + _fields_ = [("dwSize", _wt.DWORD), ("cntUsage", _wt.DWORD), ("th32ProcessID", _wt.DWORD), + ("th32DefaultHeapID", ctypes.c_size_t), ("th32ModuleID", _wt.DWORD), + ("cntThreads", _wt.DWORD), ("th32ParentProcessID", _wt.DWORD), + ("pcPriClassBase", _wt.LONG), ("dwFlags", _wt.DWORD), + ("szExeFile", _wt.WCHAR * 260)] + + _k32.CreateToolhelp32Snapshot.argtypes = [_wt.DWORD, _wt.DWORD] + _k32.CreateToolhelp32Snapshot.restype = _wt.HANDLE + _k32.Process32FirstW.argtypes = [_wt.HANDLE, ctypes.POINTER(_PE32)] + _k32.Process32FirstW.restype = _wt.BOOL + _k32.Process32NextW.argtypes = [_wt.HANDLE, ctypes.POINTER(_PE32)] + _k32.Process32NextW.restype = _wt.BOOL + _k32.TerminateProcess.argtypes = [_wt.HANDLE, _wt.UINT] + _k32.TerminateProcess.restype = _wt.BOOL def proc_ctime(pid: int) -> int | None: @@ -518,6 +584,135 @@ def record_alive(rec: dict | None) -> bool: return bool(rec) and pid_alive(rec.get("pid"), rec.get("ctime")) +def process_table() -> list[tuple[int, int, str]]: + """[(pid, ppid, name)] of every process. Windows: one Toolhelp32 snapshot (fast, no PowerShell).""" + out = [] + if IS_WINDOWS: + snap = _k32.CreateToolhelp32Snapshot(_TH32CS_SNAPPROCESS, 0) + if not snap or snap == _INVALID_HANDLE: + return out + try: + e = _PE32() + e.dwSize = ctypes.sizeof(_PE32) + ok = _k32.Process32FirstW(snap, ctypes.byref(e)) + while ok: + out.append((int(e.th32ProcessID), int(e.th32ParentProcessID), e.szExeFile)) + ok = _k32.Process32NextW(snap, ctypes.byref(e)) + finally: + _k32.CloseHandle(snap) + return out + try: + names = os.listdir("/proc") + except OSError: + return out + for n in names: + if not n.isdigit(): + continue + try: + with open("/proc/%s/stat" % n, "rb") as f: + st = f.read().decode("ascii", "replace") + comm = st[st.index("(") + 1: st.rindex(")")] + ppid = int(st[st.rindex(")") + 2:].split()[1]) + out.append((int(n), ppid, comm)) + except (OSError, ValueError, IndexError): + continue + return out + + +def process_tree(rec: dict | None) -> list[dict]: + """`rec` (if alive) and all its descendants, each {pid, ctime, ppid, name}. A process counts as + a child only if it was created at or after its parent (guards against pid reuse). Descendants + of a dead root are still found: their parent pid stays recorded.""" + if not rec or not rec.get("pid"): + return [] + kids, names = {}, {} + for pid, ppid, name in process_table(): + names[pid] = name + if pid != ppid: + kids.setdefault(ppid, []).append(pid) + root = int(rec["pid"]) + out = [] + if record_alive(rec): + out.append({"pid": root, "ctime": rec.get("ctime"), "ppid": None, "name": names.get(root)}) + stack, seen = [(root, rec.get("ctime"))], {root} + while stack: + ppid, pct = stack.pop() + now_ct = proc_ctime(ppid) + if pct is not None and now_ct is not None and now_ct != pct: + continue # this pid now belongs to another process: its children are not ours + for c in kids.get(ppid, []): + if c in seen: + continue + ct = proc_ctime(c) + if pct is not None and ct is not None and ct < pct: + continue + seen.add(c) + out.append({"pid": c, "ctime": ct, "ppid": ppid, "name": names.get(c)}) + stack.append((c, ct)) + return out + + +def terminate(rec: dict) -> bool: + """Kill one process identified by pid + creation time. True if it is gone or was signalled.""" + if not record_alive(rec): + return True + pid = int(rec["pid"]) + if IS_WINDOWS: + h = _k32.OpenProcess(_PROCESS_TERMINATE | _PROCESS_QUERY_LIMITED_INFORMATION, False, pid) + if not h: + return False + try: + return bool(_k32.TerminateProcess(h, 1)) + finally: + _k32.CloseHandle(h) + try: + os.kill(pid, signal.SIGKILL) + return True + except OSError: + return False + + +def _taskkill() -> str: + p = os.path.join(os.environ.get("SystemRoot", r"C:\Windows"), "System32", "taskkill.exe") + return p if os.path.isfile(p) else "taskkill" + + +def kill_tree(rec: dict | None, extra=()) -> dict: + """Snapshot the tree of `rec` (plus the trees of `extra` records), kill it (`taskkill /T /F` on + Windows, the process group on POSIX), sweep every snapshot pid still alive, and verify. + Returns {"snapshot", "taskkill_rc", "survivors"}; pids are always checked with their ctime.""" + roots = [r for r in [rec, *extra] if r and r.get("pid")] + snap, seen = [], set() + for r in roots: + for p in process_tree(r): + key = (p["pid"], p["ctime"]) + if key not in seen: + seen.add(key) + snap.append(p) + rc = None + for r in roots: + if not record_alive(r): + continue + if IS_WINDOWS: + try: + rc = _run([_taskkill(), "/T", "/F", "/PID", str(int(r["pid"]))], timeout=60).returncode + except (OSError, subprocess.SubprocessError) as e: + _diag("taskkill failed: %s" % e) + else: + with contextlib.suppress(OSError): + os.killpg(int(r["pid"]), signal.SIGKILL) + break # one tree kill of the first live root; the sweep covers the rest + deadline = time.monotonic() + KILL_VERIFY_S + while True: + survivors = [p for p in snap if record_alive(p)] + if not survivors or time.monotonic() >= deadline: + break + for p in survivors: + terminate(p) + time.sleep(0.1) + return {"snapshot": snap, "taskkill_rc": rc, "survivors": survivors} + + def _run(args, cwd=None, timeout=60, env=None) -> subprocess.CompletedProcess: """Run a short helper process: stdout captured (never leaks to ours), no window.""" return subprocess.run(args, cwd=cwd, env=env, stdin=subprocess.DEVNULL, capture_output=True, @@ -750,6 +945,458 @@ def model_admission(model: str, cwd: str) -> dict: return {"admitted": False, "lists": lists, "via": None} +# -------------------------------------------------------------------------------------- +# Stream parsing, usage rule (P0.3), cache verdict (C5), warmth (C1) +# -------------------------------------------------------------------------------------- + + +class StreamState: + """What one `claude -p --output-format stream-json --verbose` stream told us so far.""" + + def __init__(self): + self.init = None + self.session_id = None + self.msg_ids: list = [] + self._ids: set = set() + self.first_usage = None # usage of the first assistant event (cache verdict) + self.first_at = None # its arrival time (last_hit) + self.last_usage = None # usage of the last assistant event (base context) + self.last_assistant_at = None + self.last_event_at = None + self.result = None + self.refused = None # content of a model-restricted warning + self.events = 0 + + def feed(self, ev: dict, t: float | None) -> None: + if not isinstance(ev, dict): + return + self.events += 1 + if t is not None: + self.last_event_at = t + ty = ev.get("type") + if ty == "system": + st = ev.get("subtype") + if st == "init" and self.init is None: + self.init = ev + self.session_id = ev.get("session_id") or self.session_id + elif (st == "informational" and ev.get("level") == "warning" + and "is restricted" in str(ev.get("content", ""))): + self.refused = str(ev.get("content")) + elif ty == "assistant": + m = ev.get("message") or {} + mid = m.get("id") + if mid not in self._ids: + self._ids.add(mid) + self.msg_ids.append(mid) + u = m.get("usage") or {} + if self.first_usage is None: + self.first_usage, self.first_at = u, t + self.last_usage, self.last_assistant_at = u, t + elif ty == "result": + self.result = ev + self.session_id = self.session_id or ev.get("session_id") + + @property + def turns(self) -> int: + return len(self.msg_ids) + + @classmethod + def from_file(cls, path: str) -> "StreamState": + """Re-parse a raw stream file (event times unknown).""" + ss = cls() + try: + data = read_bytes_retry(path) + except OSError: + return ss + for line in data.splitlines(): + try: + ss.feed(json.loads(line.decode("utf-8", "replace")), None) + except ValueError: + continue + return ss + + +def context_tokens(usage: dict | None) -> int: + """Prompt-side tokens of one request: input + cache creation + cache read.""" + if not usage: + return 0 + return int(usage.get("input_tokens") or 0) + int(usage.get("cache_creation_input_tokens") or 0) \ + + int(usage.get("cache_read_input_tokens") or 0) + + +def runtime_fingerprint(init: dict | None) -> dict | None: + """The runtime fields of a `system/init` event that must match the base's.""" + if not init: + return None + tools = sorted(str(x) for x in (init.get("tools") or [])) + plugins = sorted("%s@%s" % (p.get("name"), p.get("version") or "") + for p in (init.get("plugins") or []) if isinstance(p, dict)) + agents = sorted(str(a) if not isinstance(a, dict) else str(a.get("name")) for a in (init.get("agents") or [])) + return {"tools": _sha256_bytes(canonical_json(tools).encode("utf-8")), + "permissionMode": init.get("permissionMode"), "plugins": plugins, "agents": agents, + "apiKeySource": init.get("apiKeySource")} + + +def runtime_drift(rt: dict | None, ref: dict | None, launch: dict) -> list[str]: + if not rt: + return [] + out = [] + if ref: + out += [k for k in sorted(rt) if rt.get(k) != ref.get(k)] + elif rt.get("permissionMode") not in (None, launch.get("permission_mode")): + out.append("permissionMode") + return out + + +def usage_from_result(result: dict, before_mu: dict, cost_before: float) -> dict: + """P0.3: the four columns and the 1h/5m split from `result.usage`; `models_used` = modelUsage + keys that grew against the session total before the call; cost delta against the session's + total before the call; `mismatch` when `result.usage` differs from the modelUsage delta.""" + usage = result.get("usage") or {} + cc = usage.get("cache_creation") or {} + tokens = { + "input": int(usage.get("input_tokens") or 0), + "output": int(usage.get("output_tokens") or 0), + "cache_creation": int(usage.get("cache_creation_input_tokens") or 0), + "cache_creation_1h": int(cc.get("ephemeral_1h_input_tokens") or 0), + "cache_creation_5m": int(cc.get("ephemeral_5m_input_tokens") or 0), + "cache_read": int(usage.get("cache_read_input_tokens") or 0), + } + after = result.get("modelUsage") or {} + before_mu = before_mu or {} + delta = {} + for m, v in after.items(): + b = before_mu.get(m) or {} + delta[m] = {f: int((v or {}).get(f) or 0) - int(b.get(f) or 0) for f in MU_FIELDS} + models_used = sorted(m for m, d in delta.items() if any(x > 0 for x in d.values())) + sums = {f: sum(d[f] for d in delta.values()) for f in MU_FIELDS} + mismatch = (sums["inputTokens"], sums["outputTokens"], sums["cacheCreationInputTokens"], + sums["cacheReadInputTokens"]) != (tokens["input"], tokens["output"], + tokens["cache_creation"], tokens["cache_read"]) + total = result.get("total_cost_usd") + cost_delta = None if total is None else round(float(total) - float(cost_before or 0.0), 6) + return {"tokens": tokens, "models_used": models_used, "mismatch": mismatch, + "cost_delta": cost_delta, "total": total, "model_usage": after} + + +def cache_verdict(first_usage: dict | None, base_context) -> str | None: + """C5: first assistant event's cache_read against the base's context tokens.""" + if first_usage is None: + return None + cr = int(first_usage.get("cache_read_input_tokens") or 0) + if base_context and cr >= CACHE_HIT_RATIO * float(base_context): + return "hit" + return "partial" if cr > 0 else "miss" + + +def base_warmth(base: dict, cfg: dict, now: float | None = None) -> dict: + """C1 (Phase 1): warm_until = last_hit + ttl_s; margin = max(120, ttl_s/10).""" + ttl = float(cfg.get("ttl_s") or CONFIG_DEFAULTS["ttl_s"]) + last = _ts((base or {}).get("last_hit")) + if last is None: + return {"base_state": "cold", "warm_until": None} + now = _now() if now is None else now + wu = last + ttl + margin = max(120.0, ttl / 10.0) + state = "warm" if now < wu - margin else ("expiring" if now < wu else "cold") + return {"base_state": state, "warm_until": _iso(wu)} + + +# -------------------------------------------------------------------------------------- +# Jobs: paths, templates, spawning, finalising (C3b, C3c) +# -------------------------------------------------------------------------------------- + + +def job_file(state: str, job: str) -> str: + return os.path.join(state, "jobs", job + ".json") + + +def result_file(state: str, job: str) -> str: + return os.path.join(state, "results", job + ".json") + + +def raw_file(state: str, job: str, ext: str = ".jsonl") -> str: + return os.path.join(state, "raw", job + ext) + + +def _safe_name(s: str) -> str: + return re.sub(r"[^A-Za-z0-9._-]", "_", str(s)) + + +def note_rel(block) -> str | None: + return None if block in (None, "") else "notes/%s.md" % _safe_name(block) + + +def _phase(v): + if v is None: + return None + return int(v) if re.fullmatch(r"\d+", str(v)) else str(v) + + +def render_template(name: str, values: dict) -> str: + path = os.path.join(os.path.dirname(os.path.abspath(__file__)), TEMPLATES_DIR, name) + with open(path, "r", encoding="utf-8") as f: + text = f.read() + return string.Template(text).safe_substitute(values) + + +def _powershell() -> str: + w = shutil.which("powershell") + if w: + return w + return os.path.join(os.environ.get("SystemRoot", r"C:\Windows"), "System32", "WindowsPowerShell", + "v1.0", "powershell.exe") + + +def spawn_supervisor(state: str, run: str, job: str) -> dict: + """Start `_supervise` detached (C6). Windows: CIM `Win32_Process.Create` with a hidden window, + through a `powershell` that itself has CREATE_NO_WINDOW. POSIX: a new session. + Returns {"ok", "pid", "ctime", "method"} or {"ok": False, "detail"}.""" + argv = [supervisor_interpreter(), "-I", script_path(), "_supervise", "--run", run, "--job", job] + if IS_WINDOWS: + env = dict(os.environ) + env["SPIRE_FORKS_CMD"] = subprocess.list2cmdline(argv) + env["SPIRE_FORKS_CWD"] = state + try: + r = _run([_powershell(), "-NoProfile", "-NonInteractive", "-Command", POWERSHELL_CIM], + env=env, timeout=120) + except (OSError, subprocess.SubprocessError) as e: + return {"ok": False, "method": SPAWN_METHOD, "detail": "powershell: %s" % e} + toks = r.stdout.decode("utf-8", "replace").split() + try: + rv, pid = int(toks[-2]), int(toks[-1]) + except (ValueError, IndexError): + return {"ok": False, "method": SPAWN_METHOD, + "detail": "unparsable CIM output: %r / %r" % (r.stdout[-200:], r.stderr[-400:])} + if rv != 0: + return {"ok": False, "method": SPAWN_METHOD, "detail": "Win32_Process.Create returned %d" % rv} + return {"ok": True, "method": SPAWN_METHOD, "pid": pid, "ctime": proc_ctime(pid)} + log = open(raw_file(state, job, ".sup.log"), "ab") + try: + p = subprocess.Popen(argv, cwd=state, stdin=subprocess.DEVNULL, stdout=log, stderr=log, + start_new_session=True, close_fds=True) + except OSError as e: + return {"ok": False, "method": SPAWN_METHOD, "detail": str(e)} + finally: + log.close() + return {"ok": True, "method": SPAWN_METHOD, "pid": p.pid, "ctime": proc_ctime(p.pid)} + + +def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint: str, flags, reason, + detail, exit_code, wall_s, base_commit, now: float): + """Build the result, the ledger line, and apply the registry updates for one finished job. + `hint`: "exited" (decide ok/failed from the stream), or a final status given by the caller.""" + cfg = reg["config"] + launch = cfg["launch"] + kind = entry["kind"] + base = reg["bases"].get(entry.get("base")) if entry.get("base") else None + res = ss.result if ss else None + u = usage_from_result(res, entry.get("model_usage_before") or {}, entry.get("cost_before") or 0.0) \ + if res else None + flags = list(flags or []) + models_used = u["models_used"] if u else [] + if hint == "exited": + requested = entry.get("model_requested") + if ss and (ss.refused or (res and requested not in models_used)): + status, reason = "failed", "model_refused" + detail = ss.refused or "models_used %s, requested %s" % (models_used, requested) + elif res is None: + status = "failed" + reason = "exit_code" if exit_code else "no_result" + detail = "exit %s, no result event" % exit_code + elif res.get("is_error"): + status, reason, detail = "failed", "is_error", str(res.get("subtype")) + elif exit_code: + status, reason, detail = "failed", "exit_code", "exit %s" % exit_code + else: + status, reason, detail = "ok", None, None + kr = entry.get("kill_requested") + if kr and status != "ok": + status, reason, detail = "killed", kr.get("by"), "kill requested" + if kr.get("by") == "expire": + flags.append("expired") + else: + status = hint + if u and u["mismatch"]: + flags.append("usage_mismatch") + denials = len((res or {}).get("permission_denials") or []) + if denials: + flags.append("needs_attention") + cache = cache_verdict(ss.first_usage if ss else None, entry.get("base_context")) \ + if kind in ("fork", "continue") else None + rt = runtime_fingerprint(ss.init) if ss else None + drift = list(entry.get("launch_drift") or []) + ref = (base or {}).get("runtime_fingerprint") if kind != "train" else None + drift += runtime_drift(rt, ref, launch) + drift = sorted(set(drift)) + fingerprint = "ok" if not drift else "drift:" + ",".join(drift) + base_state = entry.get("base_state_at_launch") + if kind in ("fork", "continue") and (base_state == "cold" or (cache not in (None, "hit") + and fingerprint == "ok")): + flags.append("cold_start") + flags = _uniq(flags) + turns = int(res.get("num_turns") or 0) if res else (ss.turns if ss else 0) + note = note_rel(entry.get("block")) if kind != "train" else None + if note and not os.path.isfile(os.path.join(state, note)): + note = None + session = (ss.session_id if ss and ss.session_id else None) or entry.get("session") + tokens = u["tokens"] if u else {k: None for k in TOKEN_KEYS} + ended = _iso(now) + result = { + "job": entry["job"], "kind": kind, "base": entry.get("base"), "block": entry.get("block"), + "phase": entry.get("phase"), "status": status, "flags": flags, "fingerprint": fingerprint, + "session": session, "parent_session": entry.get("parent_session"), "models_used": models_used, + "tokens": tokens, "cost_delta_usd": u["cost_delta"] if u else None, "cache": cache, + "base_state": base_state, "wall_s": None if wall_s is None else round(float(wall_s), 1), + "turns": turns, "denials": denials, "note": note, "raw": "raw/%s.jsonl" % entry["job"], + "paths": entry.get("paths"), "reason": reason, "model_requested": entry.get("model_requested"), + } + ledger = { + "job": entry["job"], "kind": kind, "phase": entry.get("phase"), "block": entry.get("block"), + "base": entry.get("base"), "session": session, "parent_session": entry.get("parent_session"), + "started": entry.get("started_at") or entry.get("launched_at") or entry.get("created_at"), + "ended": ended, "wall_s": result["wall_s"], "binary": entry.get("binary_version"), + "model_requested": entry.get("model_requested"), "models_used": models_used, "tokens": tokens, + "cost_delta_usd": result["cost_delta_usd"], "cost_cumulative_usd": u["total"] if u else None, + "turns": turns, "cache": cache, "base_state": base_state, "fingerprint": fingerprint, + "denials": denials, "status": status, "flags": flags, "reason": reason, "note": note, + } + # registry updates + entry.update({"status": status, "flags": flags, "reason": reason, "reason_detail": detail, + "ended_at": ended, "session": session, "turns": turns, "exit_code": exit_code, + "fingerprint": fingerprint, "cache": cache}) + if ss and ss.first_at: + entry["first_event_at"] = _iso(ss.first_at) + if u: + entry["cost_cum_usd"], entry["model_usage_cum"] = u["total"], u["model_usage"] + holder = entry if kind == "fork" else reg["jobs"].get(entry.get("continues") or "") + if kind in ("fork", "continue") and holder is not None: + holder["session_cost_cum_usd"] = u["total"] + holder["session_model_usage_cum"] = u["model_usage"] + if base is not None: + hit_t = (ss.last_assistant_at if kind == "train" else ss.first_at) if ss else None + if hit_t and (_ts(base.get("last_hit")) or 0) < hit_t: + base["last_hit"] = _iso(hit_t) + if kind == "train" and base.get("train_job") == entry["job"]: + if status == "ok" and base.get("status") == "training": + base.update({"status": "active", "session": session, "base_commit": base_commit, + "context_tokens": context_tokens(ss.last_usage if ss else None), + "training_cost_usd": u["cost_delta"] if u else None, + "cost_cum_usd": u["total"] if u else None, + "model_usage_cum": u["model_usage"] if u else {}, + "runtime_fingerprint": rt, "trained_at": ended}) + elif base.get("status") == "training": + base["status"] = "failed" + return result, ledger + + +def finalize_job(state: str, job: str, *, ss: StreamState | None, hint: str, flags=(), reason=None, + detail=None, exit_code=None, wall_s=None, base_commit=None) -> dict | None: + """Write `results/.json`, the ledger line and the registry update, exactly once, under the + lock (C3c). Returns the result, or None when the job was already final.""" + with RegistryLock(state): + reg = read_registry(state) + entry = (reg or {}).get("jobs", {}).get(job) + if not entry or entry.get("status") not in RUNNING: + return None + result, ledger = _conclude(state, reg, entry, ss, hint, flags, reason, detail, exit_code, + wall_s, base_commit, _now()) + write_json_atomic(result_file(state, job), result) + ledger_append(state, ledger) + write_registry(state, reg) + return result + + +def bump_last_hit(state: str, base_name: str, t: float) -> None: + """Record a hit on the base prefix (first assistant event of a fork/continue) as it happens.""" + with RegistryLock(state): + reg = read_registry(state) + base = (reg or {}).get("bases", {}).get(base_name) + if base is None or (_ts(base.get("last_hit")) or 0) >= t: + return + base["last_hit"] = _iso(t) + write_registry(state, reg) + + +def touch_orchestrator(state: str) -> None: + """N1: record that the orchestrator called us (wait, fork, continue, train, list).""" + with RegistryLock(state): + reg = read_registry(state) + if reg is None: + return + reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso() + write_registry(state, reg) + + +def _job_runtime(state: str, job: str) -> dict: + try: + return read_json(job_file(state, job)) or {} + except ForksError: + return {} + + +def reap_lost(state: str) -> list[str]: + """Decision 7 / A19: a running job whose heartbeat is older than LOST_AFTER_S and whose + supervisor (pid + creation time) is dead becomes `lost`. Its claude tree, if any is left, + is killed with it.""" + reg = read_registry(state) + lost = [] + now = _now() + for job, e in sorted((reg or {}).get("jobs", {}).items()): + if e.get("status") not in RUNNING: + continue + rt = _job_runtime(state, job) + hb = _ts(rt.get("heartbeat_at")) or _ts(e.get("launched_at")) or _ts(e.get("created_at")) + if hb is not None and now - hb <= LOST_AFTER_S: + continue + sup = rt.get("supervisor") or e.get("supervisor") + if sup and record_alive(sup): + continue + if rt.get("claude"): + kill_tree(rt["claude"]) + start = _ts(rt.get("started_at")) or _ts(e.get("launched_at")) + res = finalize_job(state, job, ss=StreamState.from_file(raw_file(state, job)), hint="lost", + flags=rt.get("flags") or [], reason="supervisor_dead", + detail="heartbeat %s, supervisor %s dead" % (rt.get("heartbeat_at"), sup), + wall_s=(hb - start) if (hb and start) else None) + if res is not None: + lost.append(job) + return lost + + +def kill_job(state: str, job: str, by: str) -> dict: + """`kill` (by="kill") and `expire` (by="expire"): snapshot the supervisor's tree, taskkill it, + sweep, verify, finalise as `killed`.""" + with RegistryLock(state): + reg = read_registry(state) + entry = reg["jobs"].get(job) + if entry is None: + raise ForksError("unknown job", job=job) + if entry.get("status") not in RUNNING: + return {"job": job, "killed": False, "status": entry.get("status"), "ok": True} + entry["kill_requested"] = {"by": by, "at": _iso()} + write_registry(state, reg) + rt = _job_runtime(state, job) + sup = rt.get("supervisor") or entry.get("supervisor") + cl = rt.get("claude") + sup_alive = record_alive(sup) + k = kill_tree(sup, extra=[cl] if cl else []) + flags = list(rt.get("flags") or []) + if by == "expire": + flags.append("expired") + start = _ts(rt.get("started_at")) or _ts(entry.get("launched_at")) + res = finalize_job(state, job, ss=StreamState.from_file(raw_file(state, job)), hint="killed", + flags=flags, reason=by, detail="killed by `%s`" % by, + wall_s=(_now() - start) if start else None) + if res is not None and not sup_alive and rt: + rt.update({"status": "killed", "ended_at": _iso()}) + with contextlib.suppress(OSError): + write_json_atomic(job_file(state, job), rt) + final = read_registry(state)["jobs"][job].get("status") + return {"job": job, "killed": True, "status": final, "ok": not k["survivors"], + "snapshot": k["snapshot"], "survivors": k["survivors"], "taskkill_rc": k["taskkill_rc"]} + + # -------------------------------------------------------------------------------------- # Argument parsing (argparse that never prints to stdout or exits on its own) # -------------------------------------------------------------------------------------- @@ -804,6 +1451,16 @@ def _pos_int(s: str) -> int: return v +def _pos_float(s: str) -> float: + try: + v = float(s) + except ValueError: + raise argparse.ArgumentTypeError("not a number: %r" % s) + if not v > 0: + raise argparse.ArgumentTypeError("must be > 0: %r" % s) + return v + + COMMANDS = ("where", "init", "train", "fork", "continue", "wait", "list", "expire", "kill", "rebase", "usage", "tick", "_supervise") @@ -850,6 +1507,16 @@ def build_parser() -> _Parser: t.add_argument("--block") t.add_argument("--phase") + def launch_overrides(sp): + # Accepted only so that they can be refused with a clear message (plan 4.3). + g = sp.add_argument_group("refused launch fields (fixed at init; read from the base record)") + for name in LAUNCH_FIELDS: + flag = "--" + name.replace("_", "-") + if name in ("allowed_tools", "disallowed_tools"): + g.add_argument(flag, nargs="+", default=None, help=argparse.SUPPRESS) + else: + g.add_argument(flag, default=None, help=argparse.SUPPRESS) + f = cmd("fork", "Fork a base for one block.") f.add_argument("--base", required=True) f.add_argument("--block", required=True) @@ -857,20 +1524,22 @@ def build_parser() -> _Parser: f.add_argument("--brief", required=True) f.add_argument("--mode", choices=("sequential", "parallel"), default="sequential") f.add_argument("--paths", nargs="+", action="extend", default=None) - f.add_argument("--max-minutes", type=_pos_int) - f.add_argument("--max-budget-usd", type=float) - f.add_argument("--force", action="store_true") + f.add_argument("--max-minutes", type=_pos_float, help="wall-time limit (default: init's max_minutes)") + f.add_argument("--max-budget-usd", type=_pos_float) + f.add_argument("--force", action="store_true", help="fork a base that is not active") + launch_overrides(f) c = cmd("continue", "One more turn in a fork (resume without --fork-session).") c.add_argument("--job", required=True) c.add_argument("--brief", required=True) - c.add_argument("--max-minutes", type=_pos_int) + c.add_argument("--max-minutes", type=_pos_float) + launch_overrides(c) wt = cmd("wait", "Block until jobs finish or the timeout (default 540, max 570).") wt.add_argument("jobs", nargs="*") - wt.add_argument("--all", action="store_true") - wt.add_argument("--any", action="store_true") - wt.add_argument("--timeout", type=_pos_int, default=540) + wt.add_argument("--all", action="store_true", help="every job not yet settled by a wait") + wt.add_argument("--any", action="store_true", help="return when any one of them has finished") + wt.add_argument("--timeout", type=_pos_int, default=WAIT_DEFAULT_S) cmd("list", "Show every base and job.") @@ -1075,17 +1744,627 @@ def _not_implemented(block=None, phase=None): return f +# --- launching: train, fork, continue ------------------------------------------------- + + +def _load_run(args) -> tuple[str, dict]: + state = state_dir_for(args.run) + reg = read_registry(state) if os.path.isdir(state) else None + if not reg or not reg.get("config"): + raise ForksError("no launch config for this run; call init first", state=state) + return state, reg + + +def _refuse_overrides(args) -> None: + given = [n for n in LAUNCH_FIELDS if getattr(args, n, None) is not None] + if given: + raise ForksError("launch fields are fixed at init; %s reads them from the base record and " + "cannot override them (a different value needs a new base)" % args.command, + fields=given, command=args.command) + + +def _read_brief(path: str) -> tuple[str, str]: + p = norm_path(path) + try: + with open(p, "r", encoding="utf-8") as f: + return p, f.read() + except OSError as e: + raise ForksError("cannot read brief", brief=p, detail=str(e)) + + +def _check_binary(ref: dict, launch: dict) -> tuple[dict, list[str]]: + """Launch-time binary re-check (version, size, mtime) against `ref`; also the hashed files.""" + path = launch["binary"]["path"] + if not os.path.isfile(path): + raise ForksError("claude binary missing (the stored path is gone); the base cannot be " + "launched with another binary", binary=path, reason="binary_missing") + info = binary_info(path) + drift = [] if all(info.get(k) == (ref or {}).get(k) for k in ("version", "size", "mtime")) else ["binary"] + asp = launch["append_system_prompt"] + try: + if _sha256_file(asp["path"]) != asp["sha256"]: + drift.append("append_system_prompt") + except OSError: + drift.append("append_system_prompt") + ag = launch.get("agent") + if ag: + try: + if _sha256_file(ag["path"]) != ag["sha256"]: + drift.append("agent") + except OSError: + drift.append("agent") + return info, drift + + +def claude_argv(launch: dict) -> list[str]: + """The fingerprinted command line (passed in full on every call; flags are not inherited).""" + a = [launch["binary"]["path"], "-p", "--output-format", "stream-json", "--verbose", + "--model", launch["model"], "--effort", launch["effort"], + "--permission-mode", launch["permission_mode"], "--tools", ",".join(launch["tools"])] + if launch.get("allowed_tools"): + a += ["--allowedTools", ",".join(launch["allowed_tools"])] + if launch.get("disallowed_tools"): + a += ["--disallowedTools", ",".join(launch["disallowed_tools"])] + a += ["--strict-mcp-config", "--mcp-config", launch["mcp"]["config"], + "--append-system-prompt-file", launch["append_system_prompt"]["path"]] + if launch.get("agent"): + a += ["--agent", launch["agent"]["name"]] + return a + + +def _git_head(cwd: str) -> str | None: + try: + r = git(["rev-parse", "HEAD"], cwd=cwd) + except (OSError, subprocess.SubprocessError): + return None + out = r.stdout.decode("utf-8", "replace").strip() + return out if r.returncode == 0 and out else None + + +def delta_header(cwd: str, base_commit: str | None) -> str: + """Commits since the base's training commit plus the working tree's shortstat against it.""" + if not base_commit: + return ("No base commit is recorded (the base's cwd was not a git repository, or git failed " + "at training). Treat everything on disk as possibly changed since training.") + parts = [] + try: + lg = git(["log", "--oneline", "%s..HEAD" % base_commit], cwd=cwd) + st = git(["diff", "--shortstat", base_commit], cwd=cwd) + log_txt = lg.stdout.decode("utf-8", "replace").strip() if lg.returncode == 0 else \ + "(git log failed: %s)" % lg.stderr.decode("utf-8", "replace").strip() + st_txt = st.stdout.decode("utf-8", "replace").strip() if st.returncode == 0 else \ + "(git diff failed: %s)" % st.stderr.decode("utf-8", "replace").strip() + except (OSError, subprocess.SubprocessError) as e: + log_txt = st_txt = "(git failed: %s)" % e + parts.append("Base commit: `%s`." % base_commit) + parts.append("Commits since then (`git log --oneline %s..HEAD`):\n\n```\n%s\n```" + % (base_commit[:12], log_txt or "(none)")) + parts.append("Working tree against the base commit (`git diff --shortstat %s`):\n\n```\n%s\n```" + % (base_commit[:12], st_txt or "(no changes)")) + return "\n\n".join(parts) + + +def _next_job_id(reg: dict) -> str: + n = 0 + for j in reg.get("jobs", {}): + m = re.fullmatch(r"j(\d+)", j) + if m: + n = max(n, int(m.group(1))) + return "j%04d" % (n + 1) + + +def _running_capped(reg: dict) -> list[str]: + return sorted(j for j, e in reg.get("jobs", {}).items() + if e.get("kind") in ("fork", "continue") and e.get("status") in RUNNING) + + +def _new_entry(job: str, kind: str, **kw) -> dict: + e = {"job": job, "kind": kind, "base": None, "block": None, "phase": None, "mode": None, + "paths": None, "status": "created", "flags": [], "reason": None, "session": None, + "parent_session": None, "model_requested": None, "created_at": _iso(), "launched_at": None, + "ended_at": None, "supervisor": None, "cwd": None, "argv": None, + "brief": "briefs/%s.md" % job, "raw": "raw/%s.jsonl" % job, "max_s": None, + "cost_before": 0.0, "model_usage_before": {}, "base_context": None, + "base_state_at_launch": None, "launch_drift": [], "binary_version": None, + "head_at_launch": None, "continues": None, "delivered_at": None, "settled_at": None} + e.update(kw) + return e + + +def _start_job(state: str, run: str, job: str, prompt: str, launch: dict) -> dict: + """Write the brief and the child env, spawn the supervisor, record it. A failed spawn ends the + job as `failed`, reason `spawn_failed`.""" + atomic_write_bytes(os.path.join(state, "briefs", job + ".md"), prompt.encode("utf-8")) + write_json_atomic(raw_file(state, job, ".env.json"), scrubbed_env(launch["binary"]["path"])) + sp = spawn_supervisor(state, run, job) + if not sp.get("ok"): + finalize_job(state, job, ss=None, hint="failed", reason="spawn_failed", detail=sp.get("detail")) + return sp + with RegistryLock(state): + reg = read_registry(state) + e = reg["jobs"][job] + e["supervisor"] = {"pid": sp["pid"], "ctime": sp.get("ctime")} + e["launched_at"] = _iso() + if e.get("status") == "created": + e["status"] = "running" + write_registry(state, reg) + return sp + + +def cmd_train(args) -> tuple[dict, int]: + state, reg = _load_run(args) + name = args.base + if not re.fullmatch(r"[A-Za-z0-9_.-]+", name or ""): + raise ForksError("bad base name", base=name) + cfg = reg["config"] + la = cfg["launch"] + brief_path, brief = _read_brief(args.brief) + binfo, drift = _check_binary(la["binary"], la) + session = str(uuid.uuid4()) + argv = claude_argv(la) + ["--session-id", session] + head = _git_head(la["cwd"]) + with RegistryLock(state): + reg = read_registry(state) + reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso() + old = reg["bases"].get(name) + if old and old.get("status") != "failed": + raise ForksError("base exists", base=name, status=old.get("status")) + job = _next_job_id(reg) + reg["bases"][name] = {"name": name, "status": "training", "session": session, "cwd": la["cwd"], + "fingerprint": cfg["fingerprint"], "binary": binfo, "created_at": _iso(), + "train_job": job, "phase": _phase(args.phase), "last_hit": None, + "base_commit": None, "context_tokens": None, "forks": []} + reg["jobs"][job] = _new_entry( + job, "train", base=name, block=args.block, phase=_phase(args.phase), session=session, + model_requested=la["model"], cwd=la["cwd"], argv=argv, max_s=float(cfg["max_minutes"]) * 60, + launch_drift=drift, binary_version=binfo["version"], head_at_launch=head, + brief_source=brief_path) + write_registry(state, reg) + prompt = render_template("train.md", {"base": name, "run": norm_path(args.run), "state_dir": state, + "notes_dir": os.path.join(state, "notes"), + "logs_dir": os.path.join(state, "logs"), "brief": brief}) + _start_job(state, norm_path(args.run), job, prompt, la) + return {"job": job, "base": name, "session_id": session}, EXIT_OK + + +def cmd_fork(args) -> tuple[dict, int]: + _refuse_overrides(args) + if args.mode == "parallel" and not args.paths: + raise ForksError("--mode parallel needs --paths (the files this fork may touch)", mode="parallel") + state, reg = _load_run(args) + cfg = reg["config"] + la = cfg["launch"] + base = reg["bases"].get(args.base) + if base is None: + raise ForksError("unknown base", base=args.base, bases=sorted(reg["bases"])) + if base.get("status") != "active" and not args.force: + raise ForksError("base is not active (use --force to fork it anyway)", base=args.base, + status=base.get("status")) + if not base.get("session"): + raise ForksError("base has no session", base=args.base, status=base.get("status")) + brief_path, brief = _read_brief(args.brief) + binfo, drift = _check_binary(base.get("binary") or la["binary"], la) + delta = delta_header(la["cwd"], base.get("base_commit")) + head = _git_head(la["cwd"]) + session = str(uuid.uuid4()) + argv = claude_argv(la) + ["--resume", base["session"], "--fork-session", "--session-id", session] + if args.max_budget_usd: + argv += ["--max-budget-usd", "%g" % args.max_budget_usd] + paths = list(args.paths) if args.paths else None + reap_lost(state) + with RegistryLock(state): + reg = read_registry(state) + reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso() + base = reg["bases"][args.base] + if base.get("status") != "active" and not args.force: + raise ForksError("base is not active (use --force to fork it anyway)", base=args.base, + status=base.get("status")) + running = _running_capped(reg) + if len(running) >= int(cfg["parallel_cap"]): + raise ForksError("parallel cap reached", parallel_cap=cfg["parallel_cap"], running=running) + job = _next_job_id(reg) + warmth = base_warmth(base, cfg) + reg["jobs"][job] = _new_entry( + job, "fork", base=args.base, block=args.block, phase=_phase(args.phase), mode=args.mode, + paths=paths, session=session, parent_session=base["session"], model_requested=la["model"], + cwd=la["cwd"], argv=argv, + max_s=float(args.max_minutes if args.max_minutes else cfg["max_minutes"]) * 60, + cost_before=float(base.get("cost_cum_usd") or 0.0), + model_usage_before=base.get("model_usage_cum") or {}, + base_context=base.get("context_tokens"), base_state_at_launch=warmth["base_state"], + warm_until_at_launch=warmth["warm_until"], launch_drift=drift, + binary_version=binfo["version"], head_at_launch=head, brief_source=brief_path, + forced=bool(args.force and base.get("status") != "active")) + base.setdefault("forks", []).append(job) + write_registry(state, reg) + if args.mode == "parallel": + commit_rule = ("This fork runs in parallel with other forks: do NOT commit and do not stage. " + "Touch only these paths: %s. The orchestrator commits after its gate." + % ", ".join("`%s`" % p for p in paths)) + else: + commit_rule = ("This fork commits: after your gate passes, stage and commit your block by " + "explicit paths only (never `git add -A` or `.`).") + if paths: + commit_rule += " Expected paths: %s." % ", ".join("`%s`" % p for p in paths) + prompt = render_template("fork.md", { + "base": args.base, "run": norm_path(args.run), "state_dir": state, "block": args.block, + "phase": "n/a" if args.phase is None else str(args.phase), "mode": args.mode, "delta": delta, + "commit_rule": commit_rule, "note_path": os.path.join(state, *note_rel(args.block).split("/")), + "logs_dir": os.path.join(state, "logs"), "brief": brief}) + _start_job(state, norm_path(args.run), job, prompt, la) + return {"job": job, "base": args.base, "base_state": warmth["base_state"], + "warm_until": warmth["warm_until"], "rebase_due": False, "retrain_due": False, + "reasons": []}, EXIT_OK + + +def cmd_continue(args) -> tuple[dict, int]: + _refuse_overrides(args) + state, reg = _load_run(args) + cfg = reg["config"] + la = cfg["launch"] + fj = reg["jobs"].get(args.job) + if fj is None: + raise ForksError("unknown job", job=args.job) + if fj.get("kind") != "fork": + raise ForksError("continue takes a fork job", job=args.job, kind=fj.get("kind")) + + def busy(r): + return [j for j, e in r["jobs"].items() if e.get("status") in RUNNING + and (j == args.job or (e.get("kind") == "continue" and e.get("continues") == args.job))] + if busy(reg): + raise ForksError("the fork is still running", job=args.job, running=busy(reg)) + if not fj.get("session"): + raise ForksError("the fork has no session", job=args.job) + base = reg["bases"].get(fj.get("base")) or {} + brief_path, brief = _read_brief(args.brief) + binfo, drift = _check_binary(base.get("binary") or la["binary"], la) + head = _git_head(la["cwd"]) + argv = claude_argv(la) + ["--resume", fj["session"]] + reap_lost(state) + with RegistryLock(state): + reg = read_registry(state) + reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso() + fj = reg["jobs"][args.job] + if busy(reg): + raise ForksError("the fork is still running", job=args.job, running=busy(reg)) + running = _running_capped(reg) + if len(running) >= int(cfg["parallel_cap"]): + raise ForksError("parallel cap reached", parallel_cap=cfg["parallel_cap"], running=running) + base = reg["bases"].get(fj.get("base")) or {} + cum = fj.get("session_cost_cum_usd") + job = _next_job_id(reg) + warmth = base_warmth(base, cfg) + reg["jobs"][job] = _new_entry( + job, "continue", base=fj.get("base"), block=fj.get("block"), phase=fj.get("phase"), + mode=fj.get("mode"), paths=fj.get("paths"), session=fj["session"], + parent_session=fj.get("parent_session"), model_requested=la["model"], cwd=la["cwd"], + argv=argv, max_s=float(args.max_minutes if args.max_minutes else cfg["max_minutes"]) * 60, + cost_before=float(cum if cum is not None else (base.get("cost_cum_usd") or 0.0)), + model_usage_before=(fj.get("session_model_usage_cum") if cum is not None + else base.get("model_usage_cum")) or {}, + base_context=base.get("context_tokens"), base_state_at_launch=warmth["base_state"], + launch_drift=drift, binary_version=binfo["version"], head_at_launch=head, + continues=args.job, brief_source=brief_path) + write_registry(state, reg) + _start_job(state, norm_path(args.run), job, brief, la) + return {"job": job}, EXIT_OK + + +# --- the supervisor ------------------------------------------------------------------- + + +class Supervisor: + """`_supervise`: runs one `claude` call, parses its stream, watches it, writes the result. + The only writer of `jobs/.json` (C3b).""" + + def __init__(self, state: str, job: str): + self.state, self.job = state, job + reg = read_registry(state) + if not reg or job not in reg.get("jobs", {}): + raise ForksError("unknown job", job=job, state=state) + self.cfg = reg["config"] + self.entry = reg["jobs"][job] + self.env = read_json(raw_file(state, job, ".env.json")) + self.ss = StreamState() + self.mu = threading.Lock() + self.flags: list[str] = [] + self.last_output = _now() + self.rt = {"job": job, "status": "running", "supervisor": pid_record(os.getpid()), "claude": None, + "started_at": _iso(), "heartbeat_at": _iso(), "turns": 0, "last_event_at": None, + "flags": []} + self._last_write = 0.0 + + def write_rt(self, **kw) -> None: + with self.mu: + self.rt.update(kw) + self.rt["turns"] = self.ss.turns + self.rt["last_event_at"] = _iso(self.ss.last_event_at) if self.ss.last_event_at else None + self.rt["flags"] = list(self.flags) + self.rt["heartbeat_at"] = _iso() + snap = dict(self.rt) + write_json_atomic(job_file(self.state, self.job), snap) + self._last_write = _now() + + def _reader(self, stream, raw) -> None: + for line in iter(stream.readline, b""): + t = _now() + raw.write(line) + raw.flush() + try: + ev = json.loads(line.decode("utf-8", "replace")) + except ValueError: + ev = None + with self.mu: + self.last_output = t + if ev is not None: + self.ss.feed(ev, t) + + def _feeder(self, stdin, data: bytes) -> None: + with contextlib.suppress(OSError, ValueError): + stdin.write(data) + with contextlib.suppress(OSError, ValueError): + stdin.close() + + def run(self) -> tuple[dict, int]: + e = self.entry + if e.get("status") not in RUNNING or e.get("kill_requested"): + return {"job": self.job, "skipped": "job is %s" % e.get("status")}, EXIT_OK + if self.env is None: + finalize_job(self.state, self.job, ss=None, hint="failed", reason="supervisor_error", + detail="missing env file") + return {"error": "missing env file"}, EXIT_USAGE + os.environ.clear() # git and taskkill lookups use the child's PATH + os.environ.update(self.env) + self.write_rt() + with open(os.path.join(self.state, "briefs", self.job + ".md"), "rb") as f: + prompt = f.read() + t0 = _now() + raw = open(raw_file(self.state, self.job), "wb") + err = open(raw_file(self.state, self.job, ".err"), "wb") + try: + p = subprocess.Popen(e["argv"], cwd=e["cwd"], env=self.env, stdin=subprocess.PIPE, + stdout=subprocess.PIPE, stderr=err, creationflags=_NOWIN, + start_new_session=not IS_WINDOWS) + except OSError as ex: + raw.close() + err.close() + finalize_job(self.state, self.job, ss=self.ss, hint="failed", reason="launch_error", + detail=str(ex), wall_s=0) + self.write_rt(status="failed") + return {"error": "launch_error", "detail": str(ex)}, EXIT_USAGE + crec = pid_record(p.pid) + self.write_rt(claude=crec) + reader = threading.Thread(target=self._reader, args=(p.stdout, raw), daemon=True) + reader.start() + threading.Thread(target=self._feeder, args=(p.stdin, prompt), daemon=True).start() + max_s = float(e.get("max_s") or float(self.cfg["max_minutes"]) * 60) + stall_s = float(self.cfg.get("stall_after_s") or CONFIG_DEFAULTS["stall_after_s"]) + orphan_s = float(self.cfg.get("orphan_after") or CONFIG_DEFAULTS["orphan_after"]) + hint, reason, detail, flags_extra = "exited", None, None, [] + hit_written = e["kind"] == "train" + last_orphan_check = 0.0 + last_turns = 0 + while p.poll() is None: + now = _now() + with self.mu: + refused, first_at, turns = self.ss.refused, self.ss.first_at, self.ss.turns + quiet = now - self.last_output + if refused: + kill_tree(crec) + break # decided as model_refused from the stream + if now - t0 > max_s: + kill_tree(crec) + hint, detail = "timeout", "wall time over max_minutes (%.0f s)" % max_s + break + if quiet > stall_s and "stalled" not in self.flags: + self.flags.append("stalled") + self.write_rt() + if not hit_written and first_at: + hit_written = True + with contextlib.suppress(ForksError, OSError): + bump_last_hit(self.state, e["base"], first_at) + if now - last_orphan_check >= ORPHAN_CHECK_S: + last_orphan_check = now + try: + reg = read_registry(self.state) + last = _ts(((reg or {}).get("run") or {}).get("last_orchestrator_call_at")) + except (ForksError, OSError): + last = None + if last is not None and now - last > orphan_s: + kill_tree(crec) + hint, reason, flags_extra = "killed", "orphaned", ["orphaned"] + detail = "no orchestrator call for %.0f s (orphan_after %.0f s)" % (now - last, orphan_s) + break + if now - self._last_write >= HEARTBEAT_S or (turns != last_turns and now - self._last_write >= 1.0): + last_turns = turns + self.write_rt() + time.sleep(SUP_TICK_S) + exit_code = p.wait() + sweep = kill_tree(crec) # leftovers of claude (backgrounded commands) + if sweep["snapshot"]: + _diag("swept %d leftover process(es) of claude: %s" % ( + len(sweep["snapshot"]), [(x["pid"], x["name"]) for x in sweep["snapshot"]])) + reader.join(READER_JOIN_S) + with contextlib.suppress(OSError): + raw.close() + with contextlib.suppress(OSError): + err.close() + wall = _now() - t0 + base_commit = None + if e["kind"] == "train": + base_commit = _git_head(e["cwd"]) or e.get("head_at_launch") + res = finalize_job(self.state, self.job, ss=self.ss, hint=hint, flags=self.flags + flags_extra, + reason=reason, detail=detail, exit_code=exit_code, wall_s=wall, + base_commit=base_commit) + final = res["status"] if res else (read_registry(self.state)["jobs"][self.job].get("status")) + self.write_rt(status=final, ended_at=_iso(), exit_code=exit_code) + return {"job": self.job, "status": final}, EXIT_OK + + +def cmd_supervise(args) -> tuple[dict, int]: + state = state_dir_for(args.run) + if not os.path.isdir(os.path.join(state, "raw")) or not re.fullmatch(r"j\d+", args.job): + raise ForksError("no such job state", state=state, job=args.job) + log = open(raw_file(state, args.job, ".sup.log"), "a", encoding="utf-8", buffering=1) + sys.stdout = sys.stderr = log + _diag("supervisor %d for %s starting (%s)" % (os.getpid(), args.job, _iso())) + try: + return Supervisor(state, args.job).run() + except Exception as ex: + traceback.print_exc(file=log) + with contextlib.suppress(Exception): + rt = _job_runtime(state, args.job) + if rt.get("claude"): + kill_tree(rt["claude"]) + finalize_job(state, args.job, ss=StreamState.from_file(raw_file(state, args.job)), + hint="failed", reason="supervisor_error", detail="%s: %s" % (type(ex).__name__, ex)) + raise + + +# --- wait, list, kill, expire ----------------------------------------------------------- + + +def _running_info(state: str, job: str, e: dict, now: float) -> dict: + rt = _job_runtime(state, job) + start = _ts(rt.get("started_at")) or _ts(e.get("launched_at")) or _ts(e.get("created_at")) + last = _ts(rt.get("last_event_at")) + return {"job": job, "block": e.get("block"), + "elapsed_s": None if start is None else round(now - start, 1), + "last_event_s_ago": None if last is None else round(now - last, 1), + "turns": int(rt.get("turns") or 0)} + + +def cmd_wait(args) -> tuple[dict, int]: + state, reg = _load_run(args) + if args.timeout > WAIT_MAX_S: + raise ForksError("--timeout over the maximum", timeout=args.timeout, max=WAIT_MAX_S) + if args.jobs and args.all: + raise ForksError("name jobs or --all, not both") + if not args.jobs and not args.all: + raise ForksError("name jobs or --all") + unknown = [j for j in args.jobs if j not in reg["jobs"]] + if unknown: + raise ForksError("unknown job", jobs=unknown) + touch_orchestrator(state) + deadline = time.monotonic() + args.timeout + + def evaluate(r): + members = list(args.jobs) if args.jobs else sorted( + j for j, x in r["jobs"].items() if not x.get("settled_at")) + finished = [j for j in members if r["jobs"][j].get("status") in FINAL] + running = [j for j in members if j not in finished] + met = bool(finished) or not members if args.any else not running + return members, finished, running, met + + while True: + reap_lost(state) + members, finished, running, met = evaluate(read_registry(state)) + left = deadline - time.monotonic() + if met or left <= 0: + break + time.sleep(min(WAIT_POLL_S, left)) + with RegistryLock(state): + reg = read_registry(state) + members, finished, running, met = evaluate(reg) + done, now = [], _now() + for j in finished: + e = reg["jobs"][j] + if e.get("delivered_at"): + continue + res = read_json(result_file(state, j)) + if res is not None: + done.append(res) + e["delivered_at"] = _iso(now) + if met: + for j in finished: + reg["jobs"][j]["settled_at"] = _iso(now) + reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso(now) + write_registry(state, reg) + statuses = {j: reg["jobs"][j].get("status") for j in finished} + entries = {j: dict(reg["jobs"][j]) for j in running} + out = {"done": done, "running": [_running_info(state, j, entries[j], now) for j in running], + "pings": []} + if not met: + return out, EXIT_WAIT_TIMEOUT + return out, (EXIT_OK if all(s == "ok" for s in statuses.values()) else EXIT_JOB_FAILED) + + +LIST_JOB_KEYS = ("job", "kind", "base", "block", "phase", "mode", "paths", "status", "flags", "reason", + "reason_detail", "session", "parent_session", "continues", "created_at", "launched_at", + "ended_at", "fingerprint", "cache", "delivered_at", "settled_at") +LIST_BASE_KEYS = ("name", "status", "session", "cwd", "base_commit", "context_tokens", + "training_cost_usd", "cost_cum_usd", "last_hit", "train_job", "forks", "phase", + "created_at", "trained_at") + + +def cmd_list(args) -> tuple[dict, int]: + state, _ = _load_run(args) + touch_orchestrator(state) + lost = reap_lost(state) + reg = read_registry(state) + now = _now() + cfg = reg["config"] + bases = [] + for name, b in sorted(reg["bases"].items()): + row = {k: b.get(k) for k in LIST_BASE_KEYS} + row["name"] = name + row.update(base_warmth(b, cfg, now)) + row.update({"rebase_due": False, "retrain_due": False, "reasons": []}) + bases.append(row) + jobs, running = [], [] + for j, e in sorted(reg["jobs"].items()): + row = {k: e.get(k) for k in LIST_JOB_KEYS} + if e.get("status") in RUNNING: + info = _running_info(state, j, e, now) + row.update({k: info[k] for k in ("elapsed_s", "last_event_s_ago", "turns")}) + running.append(j) + else: + row["turns"] = e.get("turns") + jobs.append(row) + return {"state": state, "bases": bases, "jobs": jobs, "running": running, "lost_now": lost}, EXIT_OK + + +def cmd_kill(args) -> tuple[dict, int]: + state, reg = _load_run(args) + if args.job not in reg["jobs"]: + raise ForksError("unknown job", job=args.job) + return kill_job(state, args.job, "kill"), EXIT_OK + + +def cmd_expire(args) -> tuple[dict, int]: + state, reg = _load_run(args) + names = sorted(reg["bases"]) if args.all else [args.base] + unknown = [n for n in names if n not in reg["bases"]] + if unknown: + raise ForksError("unknown base", bases=unknown) + retired = [] + with RegistryLock(state): + reg = read_registry(state) + for n in names: + b = reg["bases"][n] + if b.get("status") not in ("retired", "superseded"): + b["status"] = "retired" + b["retired_at"] = _iso() + retired.append(n) + write_registry(state, reg) + targets = sorted(j for j, e in reg["jobs"].items() + if e.get("base") in names and e.get("status") in RUNNING) + killed = [kill_job(state, j, "expire") for j in targets] + return {"retired": retired, "bases": names, "killed": killed, + "ok": all(k.get("ok", True) for k in killed)}, EXIT_OK + + HANDLERS = { "where": cmd_where, "init": cmd_init, - "train": _not_implemented(block="1.2"), - "fork": _not_implemented(block="1.2"), - "continue": _not_implemented(block="1.2"), - "wait": _not_implemented(block="1.2"), - "list": _not_implemented(block="1.2"), - "expire": _not_implemented(block="1.2"), - "kill": _not_implemented(block="1.2"), - "_supervise": _not_implemented(block="1.2"), + "train": cmd_train, + "fork": cmd_fork, + "continue": cmd_continue, + "wait": cmd_wait, + "list": cmd_list, + "expire": cmd_expire, + "kill": cmd_kill, + "_supervise": cmd_supervise, "rebase": _not_implemented(phase=2), "usage": _not_implemented(phase=2), "tick": _not_implemented(phase=2), diff --git a/plugins/spire/skills/forks/templates/fork.md b/plugins/spire/skills/forks/templates/fork.md new file mode 100644 index 0000000..8d99ea7 --- /dev/null +++ b/plugins/spire/skills/forks/templates/fork.md @@ -0,0 +1,19 @@ +# Block ${block} (spire:forks) + +You are a fork of base `${base}` (run log `${run}`), now working on block `${block}` (phase ${phase}, mode ${mode}). + +Your system prompt was snapshotted when the base was trained: the git status, the CLAUDE.md files and the environment it shows are **stale**. The delta header below lists what changed in the repository since then. Trust it and the files on disk over the snapshot and over what you remember from training; re-read every file before you change it. + +## Delta since training + +${delta} + +## Rules for this fork + +- ${commit_rule} +- Write your handoff note to `${note_path}`. +- Write gate and test output to `.txt` files under `${logs_dir}`. + +## Brief + +${brief} diff --git a/plugins/spire/skills/forks/templates/train.md b/plugins/spire/skills/forks/templates/train.md new file mode 100644 index 0000000..2f9c137 --- /dev/null +++ b/plugins/spire/skills/forks/templates/train.md @@ -0,0 +1,11 @@ +# Training (spire:forks) + +You are base `${base}` of the build whose run log is `${run}`. This session is trained once and then forked, once per block; every fork starts from what you know at the end of this session. Build an accurate picture of the code the brief below points you at: read it, trace the seams, note the invariants and the traps. + +- Do not change code in this session. If the brief asks for trunk notes, write them to the path it names; that is the only file you write. +- The helper's state dir is `${state_dir}`. Forks write handoff notes under `${notes_dir}` and gate and test output under `${logs_dir}`. +- Finish with one line `READY`, then a short summary of what you read and anything the brief asked for that you could not do. + +## Brief + +${brief} diff --git a/plugins/spire/skills/forks/tests/test_forks.py b/plugins/spire/skills/forks/tests/test_forks.py index 56d2b12..eee6b28 100644 --- a/plugins/spire/skills/forks/tests/test_forks.py +++ b/plugins/spire/skills/forks/tests/test_forks.py @@ -551,29 +551,34 @@ class InitTests(Base): # -------------------------------------------------------------------------------------- class CliTests(Base): - def test_not_implemented(self): + def test_phase2_not_implemented(self): run_log = os.path.join(self.tmp, "r-run.md") - cases = { - ("train", "--base", "B1", "--brief", "b.md"): {"block": "1.2"}, - ("fork", "--base", "B1", "--block", "1.1", "--brief", "b.md"): {"block": "1.2"}, - ("continue", "--job", "j0001", "--brief", "b.md"): {"block": "1.2"}, - ("wait", "--all"): {"block": "1.2"}, - ("list",): {"block": "1.2"}, - ("expire", "--all"): {"block": "1.2"}, - ("kill", "--job", "j0001"): {"block": "1.2"}, - ("_supervise", "--job", "j0001"): {"block": "1.2"}, - ("rebase",): {"phase": 2}, - ("usage", "--phase", "1"): {"phase": 2}, - ("tick",): {"phase": 2}, - } - for argv, want in cases.items(): + for argv in (("rebase",), ("usage", "--phase", "1"), ("tick",)): with self.subTest(cmd=argv[0]): code, out, raw = call([argv[0], "--run", run_log, *argv[1:]]) self.assertEqual(code, 2) self.assert_one_json_line(raw) self.assertEqual(out["error"], "not implemented") - for k, v in want.items(): - self.assertEqual(out[k], v) + self.assertEqual(out["phase"], 2) + self.assertFalse(os.path.exists(run_log[:-3] + "-forks")) + + def test_commands_need_init(self): + run_log = os.path.join(self.tmp, "r-run.md") + brief = os.path.join(self.tmp, "b.md") + with open(brief, "w") as f: + f.write("x\n") + cases = (("train", "--base", "B1", "--brief", brief), + ("fork", "--base", "B1", "--block", "1.1", "--brief", brief), + ("continue", "--job", "j0001", "--brief", brief), + ("wait", "--all"), ("list",), ("expire", "--all"), ("kill", "--job", "j0001"), + ("_supervise", "--job", "j0001")) + for argv in cases: + with self.subTest(cmd=argv[0]): + code, out, raw = call([argv[0], "--run", run_log, *argv[1:]]) + self.assertEqual(code, 2, out) + self.assert_one_json_line(raw) + self.assertIn("error", out) + self.assertNotEqual(out["error"], "not implemented") self.assertFalse(os.path.exists(run_log[:-3] + "-forks")) def test_unknown_command_and_missing_run(self): @@ -602,7 +607,7 @@ class CliTests(Base): self.assertEqual(code, 2) self.assertIn("inside a fork", out["error"]) code, out, _ = call(["list", "--run", run_log]) - self.assertEqual(out["error"], "not implemented") # not refused as a launch + self.assertNotIn("inside a fork", out["error"]) # not refused as a launch (no init yet) stub = make_stub_binary(self.tmp) self.assertEqual(call(["where", "--binary", stub])[0], 0) @@ -669,5 +674,914 @@ class SubprocessTests(Base): self.assertEqual(self.assert_one_json_line(r.stdout.decode("utf-8"))["error"], "usage") +# -------------------------------------------------------------------------------------- +# Launch and supervise (block 1.2): a stream-json stub `claude`, the real detached spawn. +# -------------------------------------------------------------------------------------- + +TEST_MODEL = "claude-opus-5-5[1m]" +PRICE = {"in": 5e-6, "out": 25e-6, "cw": 10e-6, "cr": 0.5e-6} + +# The stub `claude`. Behaviour is chosen per call by `@stub key[=value] ...` lines in the prompt +# (the brief reaches the stub on stdin). Sessions are kept next to the stub so that resume and +# fork see the cumulative modelUsage / total_cost_usd of the session, like the real CLI. +STUB_PY = r''' +import json, os, subprocess, sys, time, uuid +HERE = os.path.dirname(os.path.abspath(__file__)) +SESS = os.path.join(HERE, "sessions") +CALLS = os.path.join(HERE, "calls") +VERSION = "%(version)s" +PRICE = %(price)r +ENV_KEYS = ("CLAUDE_CODE_EXECPATH", "SPIRE_FORKS_CHILD", "CLAUDE_CODE_PROMPT_CACHE_TTL", + "DISABLE_AUTOUPDATER", "VIRTUAL_ENV", "CLAUDE_EFFORT") + + +def emit(o): + sys.stdout.write(json.dumps(o) + "\n") + sys.stdout.flush() + + +def opt(a, name): + if name in a and a.index(name) + 1 < len(a): + return a[a.index(name) + 1] + return None + + +def load(sid): + try: + with open(os.path.join(SESS, sid + ".json")) as f: + return json.load(f) + except OSError: + return {"modelUsage": {}, "cost": 0.0, "context": 0} + + +def save(path, obj): + os.makedirs(os.path.dirname(path), exist_ok=True) + tmp = path + ".%%d.tmp" %% os.getpid() + with open(tmp, "w") as f: + json.dump(obj, f) + os.replace(tmp, path) + + +def main(): + a = sys.argv[1:] + if "--version" in a: + print(VERSION + " (Claude Code stub)") + return 0 + prompt = sys.stdin.read() + d = {} + for line in prompt.splitlines(): + s = line.strip() + if s.startswith("@stub "): + for tok in s[6:].split(): + k, _, v = tok.partition("=") + d[k] = v if v else "1" + model = opt(a, "--model") or "claude-stub" + resume, fork, sid = opt(a, "--resume"), "--fork-session" in a, opt(a, "--session-id") + session = resume if (resume and not fork) else (sid or str(uuid.uuid4())) + prior = load(resume) if resume else {"modelUsage": {}, "cost": 0.0, "context": 0} + rec = {"argv": a, "cwd": os.getcwd(), "pid": os.getpid(), "session": session, "directives": d, + "env": {k: os.environ.get(k) for k in ENV_KEYS}, "prompt": prompt} + if "child" in d: + c = subprocess.Popen([sys.executable, "-I", "-c", "import time; time.sleep(600)"], + stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + rec["child_pid"] = c.pid + save(os.path.join(CALLS, "%%d.json" %% os.getpid()), rec) + tools = [t for t in (opt(a, "--tools") or "").split(",") if t] + emit({"type": "system", "subtype": "init", "cwd": os.getcwd(), "session_id": session, "model": model, + "permissionMode": opt(a, "--permission-mode"), "tools": tools, "mcp_servers": [], + "apiKeySource": d.get("apikey", "none"), "claude_code_version": VERSION, + "plugins": [{"name": "spire", "path": "x", "source": "spire@reliquary", + "version": d.get("plugin", "0.2.2")}], + "agents": ["claude", "general-purpose"]}) + used = model + if "restricted" in d: + used = "claude-opus-5-5" if d["restricted"] == "1" else d["restricted"] + emit({"type": "system", "subtype": "informational", "level": "warning", "session_id": session, + "content": "Model \"%%s\" is restricted by your organization's settings. Using %%s instead." + %% (model, used)}) + if "model" in d: + used = d["model"] # silent substitution + if "silent" in d: + time.sleep(float(d["silent"])) + turns = int(d.get("turns", "3")) + ctx = int(prior.get("context") or 0) + tot = {"in": 0, "out": 0, "cw": 0, "cr": 0} + prev = ctx + for i in range(turns): + if i == 0: + mode = d.get("cache", "hit" if resume else "cold") + cr = {"hit": ctx, "partial": ctx // 2, "miss": 0, "cold": 0}[mode] + cw = (ctx - cr) + 1000 + else: + cr, cw = prev, 300 + inp, out = 2, 40 + u = {"input_tokens": inp, "cache_creation_input_tokens": cw, "cache_read_input_tokens": cr, + "cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": cw}, + "output_tokens": 1, "service_tier": "standard"} + mid = "msg_stub_" + uuid.uuid4().hex[:12] + for block in ("thinking", "text"): # two events per message id, interim output_tokens + emit({"type": "assistant", "session_id": session, + "message": {"id": mid, "model": model.replace("[1m]", ""), "role": "assistant", + "content": [{"type": block}], "usage": u}}) + emit({"type": "rate_limit_event", "rate_limit_info": {"status": "allowed"}}) + tot["in"] += inp; tot["out"] += out; tot["cw"] += cw; tot["cr"] += cr + prev = inp + cw + cr + if i == 0 and "hang" in d: + time.sleep(600 if d["hang"] == "1" else float(d["hang"])) + if i == 0 and "die" in d: + os._exit(int(d.get("code", "3"))) + if "sleep" in d: + time.sleep(float(d["sleep"])) + cost = (tot["in"] * PRICE["in"] + tot["out"] * PRICE["out"] + tot["cw"] * PRICE["cw"] + + tot["cr"] * PRICE["cr"]) + mu = json.loads(json.dumps(prior.get("modelUsage") or {})) + m = mu.setdefault(used, {"inputTokens": 0, "outputTokens": 0, "cacheReadInputTokens": 0, + "cacheCreationInputTokens": 0, "webSearchRequests": 0, "costUSD": 0.0, + "contextWindow": 1000000}) + m["inputTokens"] += tot["in"]; m["outputTokens"] += tot["out"] + m["cacheReadInputTokens"] += tot["cr"]; m["cacheCreationInputTokens"] += tot["cw"] + m["costUSD"] += cost + total = float(prior.get("cost") or 0.0) + cost + usage = {"input_tokens": tot["in"], "cache_creation_input_tokens": tot["cw"], + "cache_read_input_tokens": tot["cr"], + "output_tokens": tot["out"] + (7 if "mismatch" in d else 0), + "cache_creation": {"ephemeral_1h_input_tokens": tot["cw"], "ephemeral_5m_input_tokens": 0}} + denials = ([{"tool_name": "Bash", "tool_use_id": "toolu_stub", "tool_input": {"command": "rm -rf build"}}] + if "deny" in d else []) + is_error = "error" in d + save(os.path.join(SESS, session + ".json"), {"modelUsage": mu, "cost": total, "context": prev}) + emit({"type": "result", "subtype": "error_during_execution" if is_error else "success", + "is_error": is_error, "num_turns": turns, "total_cost_usd": total, "usage": usage, + "modelUsage": mu, "permission_denials": denials, "session_id": session, "result": "READY"}) + return 1 if is_error else int(d.get("exit", "0")) + + +if __name__ == "__main__": + sys.exit(main()) +''' + + +def stub_interpreter() -> str: + base = getattr(sys, "_base_executable", None) + return base if base and os.path.isfile(base) else sys.executable + + +def make_stream_stub(d: str) -> str: + """Generate the stream stub under `d` (never committed). Windows: a `.cmd` wrapper that runs + the base interpreter with -I; POSIX: a sh wrapper.""" + sd = os.path.join(d, "stub") + os.makedirs(sd, exist_ok=True) + py = os.path.join(sd, "stub.py") + with open(py, "w", newline="\n") as f: + f.write(STUB_PY % {"version": STUB_VERSION, "price": PRICE}) + if IS_WINDOWS: + p = os.path.join(sd, "claude.cmd") + with open(p, "w", newline="\r\n") as f: + f.write('@echo off\n"%s" -I "%s" %%*\n' % (stub_interpreter(), py)) + else: + p = os.path.join(sd, "claude") + with open(p, "w", newline="\n") as f: + f.write('#!/bin/sh\nexec "%s" -I "%s" "$@"\n' % (stub_interpreter(), py)) + os.chmod(p, 0o755) + return p + + +def stub_calls(stub: str) -> list[dict]: + d = os.path.join(os.path.dirname(stub), "calls") + out = [] + for n in sorted(os.listdir(d)) if os.path.isdir(d) else []: + if n.endswith(".json"): + with open(os.path.join(d, n)) as f: + out.append(json.load(f)) + return out + + +def stub_call_for(stub: str, job_prompt_marker: str) -> dict | None: + for c in stub_calls(stub): + if job_prompt_marker in c.get("prompt", ""): + return c + return None + + +def price(tokens: dict) -> float: + return (tokens["input"] * PRICE["in"] + tokens["output"] * PRICE["out"] + + tokens["cache_creation"] * PRICE["cw"] + tokens["cache_read"] * PRICE["cr"]) + + +def visible_windows() -> list[tuple]: + """Visible top-level windows: [(hwnd, pid, class, title)] (Windows only).""" + import ctypes + import ctypes.wintypes as wt + u32 = ctypes.WinDLL("user32", use_last_error=True) + out = [] + proto = ctypes.WINFUNCTYPE(wt.BOOL, wt.HWND, wt.LPARAM) + + def cb(hwnd, _): + if u32.IsWindowVisible(hwnd): + pid = wt.DWORD() + u32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid)) + cls = ctypes.create_unicode_buffer(256) + u32.GetClassNameW(hwnd, cls, 256) + ttl = ctypes.create_unicode_buffer(256) + u32.GetWindowTextW(hwnd, ttl, 256) + out.append((int(hwnd or 0), pid.value, cls.value, ttl.value)) + return True + u32.EnumWindows(proto(cb), 0) + return out + + +def wait_for(pred, timeout=30.0, step=0.1): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + v = pred() + if v: + return v + time.sleep(step) + return pred() + + +STATE_DIRS: list[str] = [] # every launch test's state dir, for the module-end process check + + +class LaunchBase(Base): + """One run per test: a git repo as cwd, a stream stub as --binary, the real detached spawn.""" + + def setUp(self): + super().setUp() + for name, val in (("WAIT_POLL_S", 0.2),): + p = mock.patch.object(forks, name, val) + p.start() + self.addCleanup(p.stop) + p = mock.patch.object(forks, "USER_SETTINGS_PATH", os.path.join(self.tmp, "home", "settings.json")) + p.start() + self.addCleanup(p.stop) + self.cwd = os.path.join(self.tmp, "proj") + os.makedirs(self.cwd) + self.git("init", "-q") + self.commit_file("README.md", "hello\n", "initial") + os.makedirs(os.path.join(self.tmp, "reports")) + self.run_log = os.path.join(self.tmp, "reports", "2026-10-16-12-t-run.md") + self.state = forks.state_dir_for(self.run_log) + STATE_DIRS.append(self.state) + self.stub = make_stream_stub(self.tmp) + self.addCleanup(self.cleanup_processes) + self.nbrief = 0 + + # --- helpers --- + def git(self, *args): + r = subprocess.run(["git", "-c", "user.name=t", "-c", "user.email=t@t", *args], cwd=self.cwd, + capture_output=True, timeout=60) + self.assertEqual(r.returncode, 0, r.stderr) + return r.stdout.decode().strip() + + def commit_file(self, name, text, msg): + with open(os.path.join(self.cwd, name), "w", newline="\n") as f: + f.write(text) + self.git("add", name) + self.git("commit", "-q", "-m", msg) + + def setup_run(self, *extra, orphan_after=600, stall=900, cap=2): + code, out, _ = call(["init", "--run", self.run_log, "--model", TEST_MODEL, "--cwd", self.cwd, + "--binary", self.stub, "--orphan-after", str(orphan_after), + "--stall-after-s", str(stall), "--parallel-cap", str(cap), *extra]) + self.assertEqual(code, 0, out) + return out + + def brief(self, *directives, text="Do the thing."): + self.nbrief += 1 + p = os.path.join(self.tmp, "brief-%d.md" % self.nbrief) + marker = "BRIEF-MARKER-%d" % self.nbrief + with open(p, "w", newline="\n") as f: + f.write("%s\n%s\n" % (marker, text)) + if directives: + f.write("@stub %s\n" % " ".join(directives)) + return p + + def cmd(self, *argv, code=0): + c, out, raw = call([argv[0], "--run", self.run_log, *argv[1:]]) + self.assert_one_json_line(raw) + if code is not None: + self.assertEqual(c, code, out) + return out + + def train(self, base="B1", *directives): + out = self.cmd("train", "--base", base, "--brief", self.brief(*directives)) + self.assertEqual(set(out), {"job", "base", "session_id"}) + return out + + def fork(self, *directives, base="B1", block="1.1", extra=(), code=0): + return self.cmd("fork", "--base", base, "--block", block, "--brief", self.brief(*directives), + *extra, code=code) + + def wait(self, *jobs, timeout=60, any_=False): + argv = ["wait", "--run", self.run_log, "--timeout", str(timeout)] + argv += list(jobs) if jobs else ["--all"] + if any_: + argv.append("--any") + code, out, raw = call(argv) + self.assert_one_json_line(raw) + return code, out + + def finish(self, job, timeout=60): + """Wait for one job and return its result (from `done`).""" + code, out = self.wait(job, timeout=timeout) + self.assertIn(code, (0, 4), out) + got = [r for r in out["done"] if r["job"] == job] + self.assertEqual(len(got), 1, out) + return got[0] + + def reg(self): + return forks.read_registry(self.state) + + def runtime(self, job): + return forks.read_json(forks.job_file(self.state, job)) or {} + + def ledger(self): + p = forks.ledger_path(self.state) + if not os.path.exists(p): + return [] + with open(p, "rb") as f: + return [json.loads(x) for x in f.read().decode("utf-8").splitlines() if x.strip()] + + def until_running(self, job, need_child=False): + """Wait until the job's claude runs and the stub (started by a `fork_marked` brief) has + recorded itself, with its grandchild when `need_child`.""" + def ready(): + rt = self.runtime(job) + if not rt.get("claude"): + return None + c = stub_call_for(self.stub, "RUN-%s" % job) + if not c or (need_child and not c.get("child_pid")): + return None + return rt + rt = wait_for(ready, timeout=30) + self.assertTrue(rt, "job %s never got its claude running" % job) + return rt + + def assert_tree_dead(self, recs, timeout=15): + wait_for(lambda: not any(forks.record_alive(r) for r in recs), timeout=timeout) + alive = [r for r in recs if forks.record_alive(r)] + self.assertEqual(alive, [], "processes still alive: %r" % alive) + + def cleanup_processes(self): + reg = None + with contextlib.suppress(Exception): + reg = forks.read_registry(self.state) + for job, e in ((reg or {}).get("jobs") or {}).items(): + rt = self.runtime(job) + recs = [r for r in (rt.get("supervisor"), e.get("supervisor"), rt.get("claude")) if r] + if recs: + forks.kill_tree(recs[0], extra=recs[1:]) + + def fork_marked(self, *directives, **kw): + """A fork whose brief carries `RUN-` so the stub call can be matched to the job.""" + nxt = forks._next_job_id(self.reg()) + p = self.brief(*directives, text="RUN-%s" % nxt) + out = self.cmd("fork", "--base", kw.get("base", "B1"), "--block", kw.get("block", "1.1"), + "--brief", p, *kw.get("extra", ()), code=kw.get("code", 0)) + if kw.get("code", 0) == 0: + self.assertEqual(out["job"], nxt) + return out + + +class RuleTests(Base): + """Pure functions: the P0.3 usage rule (real numbers from the P0.2 seq-1 fork), C5, C1.""" + + BASE_MU = {"claude-opus-5-5": {"inputTokens": 14, "outputTokens": 7250, "cacheReadInputTokens": 361690, + "cacheCreationInputTokens": 176553, "costUSD": 1.629818}} + FORK_RESULT = { + "usage": {"input_tokens": 14, "cache_creation_input_tokens": 3113, "cache_read_input_tokens": 1288193, + "output_tokens": 1448, + "cache_creation": {"ephemeral_1h_input_tokens": 3113, "ephemeral_5m_input_tokens": 0}}, + "modelUsage": {"claude-opus-5-5": {"inputTokens": 28, "outputTokens": 8698, + "cacheReadInputTokens": 1649883, + "cacheCreationInputTokens": 179666, "costUSD": 1.9413766}}, + "total_cost_usd": 1.9413766} + + def test_usage_rule_real_fork(self): + u = forks.usage_from_result(self.FORK_RESULT, self.BASE_MU, 1.629818) + self.assertEqual(u["tokens"], {"input": 14, "output": 1448, "cache_creation": 3113, + "cache_creation_1h": 3113, "cache_creation_5m": 0, + "cache_read": 1288193}) + self.assertEqual(u["models_used"], ["claude-opus-5-5"]) + self.assertFalse(u["mismatch"]) + self.assertAlmostEqual(u["cost_delta"], 0.311559, places=6) + + def test_usage_rule_mismatch_and_new_model(self): + res = json.loads(json.dumps(self.FORK_RESULT)) + res["usage"]["output_tokens"] += 1 + self.assertTrue(forks.usage_from_result(res, self.BASE_MU, 1.629818)["mismatch"]) + # a model that did not grow is not "used"; a train has no prior totals + res = json.loads(json.dumps(self.FORK_RESULT)) + res["modelUsage"]["claude-haiku-5-5"] = {"inputTokens": 0, "outputTokens": 0} + self.assertEqual(forks.usage_from_result(res, self.BASE_MU, 0)["models_used"], ["claude-opus-5-5"]) + self.assertTrue(forks.usage_from_result(self.FORK_RESULT, {}, 0)["mismatch"]) + + def test_cache_verdict(self): + u = lambda cr: {"cache_read_input_tokens": cr} + self.assertEqual(forks.cache_verdict(u(182202), 182204), "hit") + self.assertEqual(forks.cache_verdict(u(900), 1000), "hit") + self.assertEqual(forks.cache_verdict(u(899), 1000), "partial") + self.assertEqual(forks.cache_verdict(u(7469), 92877), "partial") # P0.6 base B + self.assertEqual(forks.cache_verdict(u(0), 1000), "miss") + self.assertIsNone(forks.cache_verdict(None, 1000)) + + def test_warmth(self): + cfg = {"ttl_s": 3600} + t0 = 1_800_000_000.0 + base = {"last_hit": forks._iso(t0)} + self.assertEqual(forks.base_warmth(base, cfg, t0 + 10)["base_state"], "warm") + self.assertEqual(forks.base_warmth(base, cfg, t0 + 3600 - 361)["base_state"], "warm") + self.assertEqual(forks.base_warmth(base, cfg, t0 + 3600 - 359)["base_state"], "expiring") + self.assertEqual(forks.base_warmth(base, cfg, t0 + 3601)["base_state"], "cold") + self.assertEqual(forks._ts(forks.base_warmth(base, cfg, t0)["warm_until"]), t0 + 3600) + short = {"ttl_s": 300} + self.assertEqual(forks.base_warmth(base, short, t0 + 300 - 121)["base_state"], "warm") + self.assertEqual(forks.base_warmth(base, short, t0 + 300 - 119)["base_state"], "expiring") + self.assertEqual(forks.base_warmth({}, cfg, t0)["base_state"], "cold") + + def test_stream_state(self): + ss = forks.StreamState() + u1 = {"input_tokens": 2, "cache_creation_input_tokens": 10, "cache_read_input_tokens": 100} + u2 = {"input_tokens": 2, "cache_creation_input_tokens": 5, "cache_read_input_tokens": 112} + for ev in ({"type": "system", "subtype": "init", "session_id": "s1", "tools": ["Read", "Bash"], + "permissionMode": "dontAsk", "apiKeySource": "none", + "plugins": [{"name": "spire", "version": "0.2.2"}], "agents": ["claude"]}, + {"type": "assistant", "message": {"id": "m1", "usage": u1}}, + {"type": "assistant", "message": {"id": "m1", "usage": u1}}, + {"type": "system", "subtype": "informational", "level": "warning", + "content": "Model \"x\" is restricted by your organization's settings."}, + {"type": "assistant", "message": {"id": "m2", "usage": u2}}, + {"type": "result", "num_turns": 2}): + ss.feed(ev, 5.0) + self.assertEqual((ss.session_id, ss.turns), ("s1", 2)) + self.assertEqual(ss.first_usage, u1) + self.assertEqual(forks.context_tokens(ss.last_usage), 119) + self.assertIn("is restricted", ss.refused) + rt = forks.runtime_fingerprint(ss.init) + self.assertEqual(rt["plugins"], ["spire@0.2.2"]) + other = dict(rt, plugins=["spire@0.3.0"], apiKeySource="oauth") + self.assertEqual(forks.runtime_drift(other, rt, {}), ["apiKeySource", "plugins"]) + self.assertEqual(forks.runtime_drift(rt, None, {"permission_mode": "default"}), ["permissionMode"]) + + +class LaunchTests(LaunchBase): + def test_train_fork_continue_end_to_end(self): + self.setup_run() + t = self.train("B1", "turns=4") + self.assertEqual((t["job"], t["base"]), ("j0001", "B1")) + r = self.finish("j0001") + self.assertEqual(r["status"], "ok", r) + self.assertEqual(set(r), {"job", "kind", "base", "block", "phase", "status", "flags", "fingerprint", + "session", "parent_session", "models_used", "tokens", "cost_delta_usd", + "cache", "base_state", "wall_s", "turns", "denials", "note", "raw", + "paths", "reason", "model_requested"}) + self.assertEqual(r["kind"], "train") + self.assertIsNone(r["cache"]) + self.assertEqual(r["session"], t["session_id"]) + self.assertIsNone(r["parent_session"]) + self.assertEqual(r["models_used"], [TEST_MODEL]) + self.assertEqual(r["model_requested"], TEST_MODEL) + self.assertEqual(r["fingerprint"], "ok") + self.assertEqual(r["turns"], 4) + self.assertEqual(r["flags"], []) + self.assertIsNone(r["reason"]) + self.assertAlmostEqual(r["cost_delta_usd"], price(r["tokens"]), places=6) + self.assertEqual(r["raw"], "raw/j0001.jsonl") + self.assertTrue(os.path.isfile(os.path.join(self.state, r["raw"]))) + base = self.reg()["bases"]["B1"] + self.assertEqual(base["status"], "active") + self.assertEqual(base["session"], t["session_id"]) + self.assertEqual(base["base_commit"], self.git("rev-parse", "HEAD")) + # base context = input + creation + read of the last assistant message: 2 + 300 + (prev) + self.assertEqual(base["context_tokens"], 2 + 1000 + 3 * (2 + 300) + 0) + self.assertIsNotNone(base["last_hit"]) + self.assertAlmostEqual(base["training_cost_usd"], r["cost_delta_usd"], places=6) + # the command line: absolute stored binary, full fingerprint, pre-assigned session id + call0 = stub_calls(self.stub)[0] + argv = call0["argv"] + self.assertEqual(argv[argv.index("--session-id") + 1], t["session_id"]) + self.assertEqual(argv[argv.index("--model") + 1], TEST_MODEL) + self.assertEqual(argv[argv.index("--permission-mode") + 1], "dontAsk") + self.assertIn("--strict-mcp-config", argv) + self.assertEqual(argv[argv.index("--append-system-prompt-file") + 1], forks.preamble_path()) + self.assertNotIn("--resume", argv) + self.assertEqual(self.reg()["jobs"]["j0001"]["argv"][0], forks.norm_path(self.stub)) + self.assertEqual(os.path.normcase(call0["cwd"]), os.path.normcase(self.cwd)) + self.assertEqual(call0["env"]["CLAUDE_CODE_EXECPATH"], forks.norm_path(self.stub)) + self.assertEqual(call0["env"]["SPIRE_FORKS_CHILD"], "1") + self.assertEqual(call0["env"]["CLAUDE_CODE_PROMPT_CACHE_TTL"], "1h") + self.assertIsNone(call0["env"]["VIRTUAL_ENV"]) + self.assertIn("READY", call0["prompt"]) # templates/train.md frame + # --- fork, after a commit on the trunk --- + self.commit_file("src.py", "x = 1\n", "add src after training") + os.makedirs(os.path.join(self.state, "notes"), exist_ok=True) + with open(os.path.join(self.state, "notes", "1.1.md"), "w") as f: + f.write("note\n") + f_out = self.fork("turns=2", extra=("--phase", "1", "--paths", "src.py", "docs/a.md")) + self.assertEqual(set(f_out), {"job", "base", "base_state", "warm_until", "rebase_due", + "retrain_due", "reasons"}) + self.assertEqual(f_out["base_state"], "warm") + self.assertFalse(f_out["rebase_due"]) + self.assertFalse(f_out["retrain_due"]) + self.assertEqual(f_out["reasons"], []) + fj = f_out["job"] + r2 = self.finish(fj) + self.assertEqual(r2["status"], "ok", r2) + self.assertEqual(r2["kind"], "fork") + self.assertEqual(r2["cache"], "hit") + self.assertEqual(r2["phase"], 1) + self.assertEqual(r2["base_state"], "warm") + self.assertEqual(r2["flags"], []) + self.assertEqual(r2["fingerprint"], "ok") + self.assertEqual(r2["parent_session"], t["session_id"]) + self.assertNotEqual(r2["session"], t["session_id"]) + self.assertEqual(r2["paths"], ["src.py", "docs/a.md"]) + self.assertEqual(r2["note"], "notes/1.1.md") + self.assertEqual(r2["models_used"], [TEST_MODEL]) + # first request reads the base context, the second the base context plus the first turn + self.assertEqual(r2["tokens"]["cache_read"], 2 * base["context_tokens"] + 1002) + self.assertAlmostEqual(r2["cost_delta_usd"], price(r2["tokens"]), places=6) + fcall = [c for c in stub_calls(self.stub) if c["session"] == r2["session"]][0] + a = fcall["argv"] + self.assertEqual(a[a.index("--resume") + 1], t["session_id"]) + self.assertIn("--fork-session", a) + self.assertEqual(a[a.index("--session-id") + 1], r2["session"]) + self.assertIn("--append-system-prompt-file", a) # flags passed again + with open(os.path.join(self.state, "briefs", fj + ".md"), encoding="utf-8") as fh: + brief_txt = fh.read() + self.assertIn("add src after training", brief_txt) # delta header: git log + self.assertIn("1 file changed", brief_txt) # delta header: shortstat + self.assertIn("stale", brief_txt) # R13 sentence + self.assertIn(os.path.join(self.state, "notes", "1.1.md"), brief_txt) + # --- continue the fork: cost delta against the fork's own last total --- + c_out = self.cmd("continue", "--job", fj, "--brief", self.brief("turns=1")) + self.assertEqual(set(c_out), {"job"}) + r3 = self.finish(c_out["job"]) + self.assertEqual(r3["status"], "ok", r3) + self.assertEqual(r3["kind"], "continue") + self.assertEqual(r3["session"], r2["session"]) + self.assertEqual(r3["block"], "1.1") + self.assertAlmostEqual(r3["cost_delta_usd"], price(r3["tokens"]), places=6) + self.assertEqual(r3["cache"], "hit") + ccall = [c for c in stub_calls(self.stub) if c["session"] == r2["session"] and c is not fcall + and "--fork-session" not in c["argv"]][0] + self.assertEqual(ccall["argv"][ccall["argv"].index("--resume") + 1], r2["session"]) + self.assertNotIn("--session-id", ccall["argv"]) + # ledger lines match the results + led = {x["job"]: x for x in self.ledger()} + self.assertEqual(sorted(led), ["j0001", fj, c_out["job"]]) + for res in (r, r2, r3): + line = led[res["job"]] + for k in ("status", "flags", "tokens", "cost_delta_usd", "cache", "models_used", "session", + "parent_session", "fingerprint", "turns", "reason", "note"): + self.assertEqual(line[k], res[k], k) + self.assertEqual(line["binary"], STUB_VERSION) + self.assertIsNotNone(line["cost_cumulative_usd"]) + # list shows every base and job; nothing running + lst = self.cmd("list") + self.assertEqual([b["name"] for b in lst["bases"]], ["B1"]) + self.assertEqual(lst["bases"][0]["base_state"], "warm") + self.assertEqual(len(lst["jobs"]), 3) + self.assertEqual(lst["running"], []) + self.assertIsNotNone(self.reg()["run"]["last_orchestrator_call_at"]) + # refusals on continue + self.cmd("continue", "--job", "j0001", "--brief", self.brief(), code=2) # not a fork + self.cmd("continue", "--job", "j0999", "--brief", self.brief(), code=2) # unknown + self.cmd("train", "--base", "B1", "--brief", self.brief(), code=2) # base exists + + def test_cache_verdicts_and_cold_start(self): + self.setup_run() + self.train("B1") + self.finish("j0001") + a = self.fork("cache=partial", extra=("--max-budget-usd", "2.5"))["job"] + b = self.fork("cache=miss")["job"] + argv_a = self.reg()["jobs"][a]["argv"] + self.assertEqual(argv_a[argv_a.index("--max-budget-usd") + 1], "2.5") + self.assertNotIn("--max-budget-usd", self.reg()["jobs"][b]["argv"]) + code, out = self.wait(a, b) + self.assertEqual(code, 0, out) + res = {x["job"]: x for x in out["done"]} + self.assertEqual(res[a]["cache"], "partial") + self.assertEqual(res[b]["cache"], "miss") + for j in (a, b): + self.assertIn("cold_start", res[j]["flags"]) + self.assertEqual(res[j]["status"], "ok") + self.assertEqual(res[j]["base_state"], "warm") + # a cold base: hit, but cold at launch -> cold_start + with forks.RegistryLock(self.state): + reg = forks.read_registry(self.state) + reg["bases"]["B1"]["last_hit"] = forks._iso(time.time() - 4000) + forks.write_registry(self.state, reg) + out = self.fork() + self.assertEqual(out["base_state"], "cold") + r = self.finish(out["job"]) + self.assertEqual(r["cache"], "hit") + self.assertEqual(r["base_state"], "cold") + self.assertIn("cold_start", r["flags"]) + # the hit refreshed last_hit: the base is warm again + self.assertEqual(self.cmd("list")["bases"][0]["base_state"], "warm") + + def test_failures_flags_and_drift(self): + self.setup_run() + self.train("B1") + self.finish("j0001") + cases = [("die=1",), ("error",), ("restricted=1",), ("model=claude-other-9",), ("deny",), + ("mismatch",), ("plugin=0.3.0",)] + results = {} + for i in range(0, len(cases), 2): + jobs = [self.fork(*c)["job"] for c in cases[i:i + 2]] + code, out = self.wait(*jobs) + for x in out["done"]: + results[x["job"]] = x + r = [results["j%04d" % (i + 2)] for i in range(len(cases))] + die, err, restricted, substituted, deny, mismatch, plugin = r + self.assertEqual((die["status"], die["reason"]), ("failed", "exit_code")) + self.assertEqual(die["turns"], 1) + self.assertIsNone(die["cost_delta_usd"]) + self.assertEqual((err["status"], err["reason"]), ("failed", "is_error")) + self.assertEqual((restricted["status"], restricted["reason"]), ("failed", "model_refused")) + self.assertEqual((substituted["status"], substituted["reason"]), ("failed", "model_refused")) + self.assertEqual(substituted["models_used"], ["claude-other-9"]) + self.assertEqual(deny["status"], "ok") + self.assertIn("needs_attention", deny["flags"]) + self.assertEqual(deny["denials"], 1) + self.assertEqual(mismatch["status"], "ok") + self.assertIn("usage_mismatch", mismatch["flags"]) + self.assertEqual(plugin["fingerprint"], "drift:plugins") + self.assertNotIn("cold_start", plugin["flags"]) + # model refused while training: the base fails + t = self.train("B2", "restricted=1") + r = self.finish(t["job"]) + self.assertEqual((r["status"], r["reason"]), ("failed", "model_refused")) + self.assertEqual(self.reg()["bases"]["B2"]["status"], "failed") + self.fork(base="B2", code=2) # not active + # binary drift: the stored binary changed size and mtime since training + with open(self.stub, "a", newline="\r\n" if IS_WINDOWS else "\n") as f: + f.write("rem drift\n" if IS_WINDOWS else "# drift\n") + r = self.finish(self.fork()["job"]) + self.assertEqual(r["status"], "ok") + self.assertEqual(r["fingerprint"], "drift:binary") + + def test_timeout_and_stalled(self): + self.setup_run(stall=1) + self.train("B1") + self.finish("j0001") + slow = self.fork("silent=2.5")["job"] + hang = self.fork_marked("hang=1", extra=("--max-minutes", "0.06"))["job"] + rt = self.until_running(hang) + tree = forks.process_tree(rt["supervisor"]) + self.assertGreaterEqual(len(tree), 3) + code, out = self.wait(slow, hang, timeout=60) + self.assertEqual(code, 4, out) + res = {x["job"]: x for x in out["done"]} + self.assertEqual(res[slow]["status"], "ok") + self.assertIn("stalled", res[slow]["flags"]) + self.assertEqual(res[hang]["status"], "timeout") + self.assertIsNone(res[hang]["reason"]) + self.assertIn("stalled", res[hang]["flags"]) + self.assert_tree_dead(tree) # claude tree killed; the supervisor exited + + def test_kill_tree_and_no_visible_window(self): + self.setup_run() + self.train("B1") + self.finish("j0001") + job = self.fork_marked("hang=1", "child")["job"] + rt = self.until_running(job, need_child=True) + sup = rt["supervisor"] + tree = forks.process_tree(sup) + pids = {p["pid"] for p in tree} + stub = stub_call_for(self.stub, "RUN-%s" % job) + self.assertIn(sup["pid"], pids) + self.assertIn(rt["claude"]["pid"], pids) + self.assertIn(stub["pid"], pids) + self.assertIn(stub["child_pid"], pids) + if IS_WINDOWS: + # spawned through CIM: the supervisor's parent is the WMI provider host, not us + table = {pid: (ppid, name) for pid, ppid, name in forks.process_table()} + parent = table.get(table[sup["pid"]][0], (None, None))[1] + self.assertEqual((parent or "").lower(), "wmiprvse.exe") + self.assertIn("python.exe", [p["name"].lower() for p in tree]) + self.assertIn("cmd.exe", [p["name"].lower() for p in tree]) + all_windows = visible_windows() + self.assertGreater(len(all_windows), 0, "window enumeration sees nothing (no desktop?)") + for _ in range(3): + mine = [w for w in visible_windows() if w[1] in pids] + self.assertEqual(mine, [], "visible window of the job's process tree") + time.sleep(0.2) + out = self.cmd("kill", "--job", job) + self.assertTrue(out["killed"]) + self.assertTrue(out["ok"], out) + self.assertEqual(out["survivors"], []) + self.assertEqual(out["status"], "killed") + snap_pids = {p["pid"] for p in out["snapshot"]} + for pid in (sup["pid"], rt["claude"]["pid"], stub["pid"], stub["child_pid"]): + self.assertIn(pid, snap_pids) + self.assert_tree_dead(tree, timeout=1) + r = self.finish(job) + self.assertEqual((r["status"], r["reason"]), ("killed", "kill")) + self.assertEqual(r["turns"], 1) + self.assertEqual([x["status"] for x in self.ledger() if x["job"] == job], ["killed"]) + again = self.cmd("kill", "--job", job) + self.assertFalse(again["killed"]) + self.cmd("kill", "--job", "j0999", code=2) + + def test_lost(self): + self.setup_run() + self.train("B1") + self.finish("j0001") + job = self.fork_marked("hang=1")["job"] + rt = self.until_running(job) + tree = forks.process_tree(rt["supervisor"]) + self.assertTrue(forks.terminate(rt["supervisor"])) + self.assertTrue(wait_for(lambda: not forks.record_alive(rt["supervisor"]), 10)) + lst = self.cmd("list") # heartbeat is fresh: not lost yet + self.assertEqual(lst["running"], [job]) + later = time.time() + forks.LOST_AFTER_S + 30 + with mock.patch.object(forks, "_now", lambda: later): + lst = self.cmd("list") + self.assertEqual(lst["lost_now"], [job]) + self.assertEqual(lst["running"], []) + self.assert_tree_dead(tree) # its orphaned claude tree went with it + code, out = self.wait(job) + self.assertEqual(code, 4) + self.assertEqual((out["done"][0]["status"], out["done"][0]["reason"]), ("lost", "supervisor_dead")) + self.assertEqual([x["status"] for x in self.ledger() if x["job"] == job], ["lost"]) + + def test_orphan_watchdog(self): + self.setup_run() # orphan_after 600 s + self.train("B1") + self.finish("j0001") + job = self.fork_marked("hang=1", "child")["job"] + rt = self.until_running(job, need_child=True) + tree = forks.process_tree(rt["claude"]) + self.assertGreaterEqual(len(tree), 2) + self.assertEqual(self.reg()["jobs"][job]["status"], "running") + # injected clock: the last orchestrator call is recorded as 2000 s ago + past = time.time() - 2000 + with mock.patch.object(forks, "_now", lambda: past): + forks.touch_orchestrator(self.state) + # no orchestrator call from here on: poll the registry directly (no touch) + final = wait_for(lambda: self.reg()["jobs"][job]["status"] not in forks.RUNNING, timeout=30, step=0.2) + self.assertTrue(final) + e = self.reg()["jobs"][job] + self.assertEqual((e["status"], e["reason"]), ("killed", "orphaned")) + self.assertIn("orphaned", e["flags"]) + self.assert_tree_dead(tree) + self.assertTrue(wait_for(lambda: not forks.record_alive(rt["supervisor"]), 15)) + self.assertEqual([x["flags"] for x in self.ledger() if x["job"] == job], [e["flags"]]) + + def test_expire_kills_running_jobs(self): + self.setup_run() + self.train("B1") + self.finish("j0001") + job = self.fork_marked("hang=1")["job"] + rt = self.until_running(job) + tree = forks.process_tree(rt["supervisor"]) + out = self.cmd("expire", "--base", "B1") + self.assertEqual(out["retired"], ["B1"]) + self.assertEqual([k["job"] for k in out["killed"]], [job]) + self.assertTrue(out["ok"]) + self.assert_tree_dead(tree) + r = self.finish(job) + self.assertEqual((r["status"], r["reason"]), ("killed", "expire")) + self.assertIn("expired", r["flags"]) + self.assertEqual(self.reg()["bases"]["B1"]["status"], "retired") + self.fork(code=2) # retired: refused without --force + r = self.finish(self.fork(extra=("--force",))["job"]) + self.assertEqual(r["status"], "ok") + self.assertEqual(self.cmd("expire", "--all")["retired"], []) + + def test_wait_exit_codes(self): + self.setup_run() + self.train("B1") + self.assertEqual(self.wait()[0], 0) # settles the training job + a = self.fork("die=1")["job"] + b = self.fork("sleep=2", "turns=5")["job"] + wait_for(lambda: self.reg()["jobs"][a]["status"] == "failed", timeout=30) + code, out = self.wait(timeout=1) # mixed: one failed, one running + self.assertEqual(code, 3, out) + self.assertEqual([x["job"] for x in out["done"]], [a]) + self.assertEqual(out["done"][0]["status"], "failed") + self.assertEqual([x["job"] for x in out["running"]], [b]) + self.assertEqual(set(out["running"][0]), {"job", "block", "elapsed_s", "last_event_s_ago", "turns"}) + self.assertEqual(out["pings"], []) + code, out = self.wait(timeout=60) # condition met, one failed -> 4 + self.assertEqual(code, 4, out) + self.assertEqual([x["job"] for x in out["done"]], [b]) # a is not returned twice + self.assertEqual(out["done"][0]["status"], "ok") + code, out = self.wait(timeout=5) # nothing left + self.assertEqual((code, out["done"], out["running"]), (0, [], [])) + self.assertEqual(self.wait(b)[0], 0) + self.assertEqual(self.wait(a)[0], 4) + # --any + c = self.fork("die=1")["job"] + d = self.fork("sleep=1", "turns=3")["job"] + code, out = self.wait(timeout=60, any_=True) + self.assertEqual(code, 4, out) + self.assertEqual([x["job"] for x in out["done"]], [c]) + code, out = self.wait(timeout=60) + self.assertEqual(code, 0, out) + self.assertEqual([x["job"] for x in out["done"]], [d]) + # usage errors + for argv in (["wait", "--run", self.run_log, "--all", "--timeout", "571"], + ["wait", "--run", self.run_log], + ["wait", "--run", self.run_log, "j0999"], + ["wait", "--run", self.run_log, "--all", "j0001"]): + self.assertEqual(call(argv)[0], 2, argv) + + def test_spawn_failed(self): + self.setup_run() + failed = {"ok": False, "method": forks.SPAWN_METHOD, "detail": "Win32_Process.Create returned 9"} + with mock.patch.object(forks, "spawn_supervisor", lambda *a: failed): + t = self.train("B1") + code, out = self.wait(t["job"], timeout=5) + self.assertEqual(code, 4, out) + r = out["done"][0] + self.assertEqual((r["status"], r["reason"]), ("failed", "spawn_failed")) + self.assertEqual(r["tokens"], {k: None for k in forks.TOKEN_KEYS}) + self.assertEqual(self.reg()["bases"]["B1"]["status"], "failed") + self.assertEqual(self.reg()["jobs"][t["job"]]["reason_detail"], failed["detail"]) + self.assertEqual([x["reason"] for x in self.ledger()], ["spawn_failed"]) + self.assertEqual(stub_calls(self.stub), []) + + def test_refusals(self): + # launch-field overrides on fork and continue (refused before anything is read) + values = {"model": "claude-x-1", "effort": "low", "permission_mode": "default", "tools": "Read", + "allowed_tools": "Read", "disallowed_tools": "Bash", "agent": "a", "cwd": self.tmp, + "binary": self.stub} + for name in forks.LAUNCH_FIELDS: + flag = "--" + name.replace("_", "-") + for argv in (["fork", "--base", "B1", "--block", "1", "--brief", "b.md"], + ["continue", "--job", "j0001", "--brief", "b.md"]): + with self.subTest(field=name, cmd=argv[0]): + out = self.cmd(*argv, flag, values[name], code=2) + self.assertEqual(out["fields"], [name]) + self.assertFalse(os.path.exists(self.state)) + self.setup_run() + # N2: parallel needs --paths + out = self.cmd("fork", "--base", "B1", "--block", "1", "--brief", self.brief(), "--mode", + "parallel", code=2) + self.assertIn("--paths", out["error"]) + self.fork(base="B9", code=2) # unknown base + self.train("B1") + self.finish("j0001") + # parallel cap (2): two hanging parallel forks, a third launch is refused + j1 = self.fork_marked("hang=1", extra=("--mode", "parallel", "--paths", "a.py"))["job"] + j2 = self.fork_marked("hang=1", extra=("--mode", "parallel", "--paths", "b.py", "c.py"))["job"] + out = self.fork(code=2) + self.assertEqual(out["error"], "parallel cap reached") + self.assertEqual(out["running"], [j1, j2]) + out = self.cmd("continue", "--job", j1, "--brief", self.brief(), code=2) # still running + self.assertIn("running", out["error"]) + # N3: launching refused inside a fork; list and wait stay allowed + os.environ[forks.CHILD_MARKER] = "1" + try: + for argv in (["train", "--base", "B3", "--brief", self.brief()], + ["fork", "--base", "B1", "--block", "2", "--brief", self.brief()], + ["continue", "--job", j1, "--brief", self.brief()]): + out = self.cmd(*argv, code=2) + self.assertIn("inside a fork", out["error"]) + self.assertEqual(self.cmd("list")["running"], [j1, j2]) + code, out = self.wait(j1, timeout=1) + self.assertEqual(code, 3) + finally: + os.environ.pop(forks.CHILD_MARKER, None) + for j in (j1, j2): + self.until_running(j) + self.assertTrue(self.cmd("kill", "--job", j)["ok"]) + code, out = self.wait(j1, j2) + self.assertEqual(code, 4) + self.assertEqual({x["job"]: x["paths"] for x in out["done"]}, {j1: ["a.py"], j2: ["b.py", "c.py"]}) + # missing binary: the stored path is gone + moved = self.stub + ".moved" + os.replace(self.stub, moved) + try: + out = self.fork(code=2) + self.assertEqual(out["reason"], "binary_missing") + out = self.cmd("train", "--base", "B4", "--brief", self.brief(), code=2) + self.assertEqual(out["reason"], "binary_missing") + finally: + os.replace(moved, self.stub) + self.cmd("fork", "--base", "B1", "--block", "1", "--brief", os.path.join(self.tmp, "nope.md"), code=2) + + +def tearDownModule(): + """No supervisor or stub of this suite may survive it (checked by command line, never by name).""" + if not IS_WINDOWS or not STATE_DIRS: + return + ps = ("$p = 'forks' + '-test-'; Get-CimInstance Win32_Process | " + "Where-Object { $_.CommandLine -and $_.CommandLine.Contains($p) } | " + "ForEach-Object { '{0}|{1}' -f $_.ProcessId, $_.CommandLine }") + r = subprocess.run(["powershell", "-NoProfile", "-NonInteractive", "-Command", ps], + capture_output=True, timeout=120, creationflags=0x08000000) + lines = [x for x in r.stdout.decode("utf-8", "replace").splitlines() if x.strip()] + for x in lines: + with contextlib.suppress(ValueError, OSError): + forks.terminate(forks.pid_record(int(x.split("|", 1)[0]))) + if lines: + raise AssertionError("leftover test processes (killed now): %s" % lines) + + if __name__ == "__main__": unittest.main(verbosity=2)