From d36ac2fa32c3e996417b18613ffc97e7fe0ca3b8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gabor=20K=C3=B6rber?= Date: Fri, 16 Oct 2026 12:29:36 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20forks:=20helper=20skeleton,=20state?= =?UTF-8?q?,=20where=20and=20init=20(block=201.1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../2026-10-16-08-headless-forks-run.md | 30 + plugins/spire/skills/forks/forks.py | 1124 +++++++++++++++++ plugins/spire/skills/forks/preamble.md | 9 + .../spire/skills/forks/tests/test_forks.py | 673 ++++++++++ 4 files changed, 1836 insertions(+) create mode 100644 plugins/spire/skills/forks/forks.py create mode 100644 plugins/spire/skills/forks/preamble.md create mode 100644 plugins/spire/skills/forks/tests/test_forks.py 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 14fdc85..e581d22 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 @@ -63,3 +63,33 @@ Intake 2026-10-16 12:18 +0200, Athanor `claude-opus-5-5[1m]`. Start commit `83f6 | Block | Owner (model id) | Commit | Tokens (if known) | |---|---|---|---| + +### Block 1.1 — Skeleton, state, where, init + +- Model: `claude-opus-5-5[1m]` +- Start / end: Fri Oct 16 12:20:48 +0200 2026 / Fri Oct 16 12:28:49 +0200 2026 (`date`; tests and gates finished by the end time, the note and commit followed) +- Done: New `plugins/spire/skills/forks/forks.py` (plan §2 PEP 723 header, stdlib only, LF): argparse dispatch for all 13 commands; exit codes 0/2/3/4 as constants; stdout = exactly one JSON object per call (argparse errors, `--help` and uncaught exceptions included); path normalisation (relative, `G:\`, `G:/`, `/g/…`, `/c/…`; POSIX `abspath`); state dir `-forks` with `.gitignore` (`raw/`, `*.lock`), `mcp-empty.json`, `jobs/ briefs/ notes/ results/ logs/ raw/`; `registry.json` `"schema": 1`; `RegistryLock` (create-exclusive, stale after 30 s, token-checked release); atomic temp-then-`os.replace` with `PermissionError` retries (40 × 50 ms) for writers and readers; `ledger_append` (caller holds the lock); `_now()` / `_iso()`; `pid_record` / `pid_alive` / `proc_ctime` (Windows `OpenProcess` + `GetProcessTimes`, POSIX `/proc//stat`); `git()` always `git --no-optional-locks`; scrub constants (`SCRUB_DROP`, `SCRUB_DROP_PATTERNS`, `SCRUB_KEEP`, `SCRUB_PINNED` + `CLAUDE_CODE_EXECPATH`) and `scrubbed_env(binary)`; `supervisor_interpreter()` (C6); `SPAWN_METHOD` (`cim-hidden` / `setsid`); `where`; `init` (all launch fields and defaults, fingerprint sha256 of canonical JSON, config extras incl. `orphan_after` 1200 and `pings` on, idempotency, C8 admission, all validation before any write). Launching and Phase 1 commands return `{"error":"not implemented","block":"1.2"}`, Phase 2 ones `{"phase":2}`, exit 2. New `preamble.md` (digest content, short). New `tests/test_forks.py` (PEP 723 + `unittest`, imports `forks.py` by path, 52 tests: path forms, state-dir derivation, scrub, lock contention in-process and across processes, stale-lock recovery by `at`, by mtime and under an injected `_now`, safe release after takeover, replace while a reader holds the target (retried, `attempts > 1` asserted on Windows), registry schema, ledger, pid + ctime, `init` defaults / idempotency / change before and after a fake base / bad model ids / `plan` / unknown mode and effort / missing binary / agent, C8 exact / `[1m]` strip / family alias / no list / unlisted → exit 2 naming the files with settings untouched, `where` via `uv run` with a stub and with the host binary, error-case stdout invariant via `uv run`, unknown command, not implemented, `--help`, internal error, N3 guard). Stub binary generated at test runtime in `tempfile` (`.cmd` on Windows); `USER_SETTINGS_PATH` monkeypatched to a temp file, so no test reads `~/.claude`. +- Gates: + - G-unit `uv run plugins/spire/skills/forks/tests/test_forks.py` → exit 0, `Ran 52 tests in 3.173s` / `OK`. + - G-where `uv run plugins/spire/skills/forks/forks.py where` → exit 0; stdout 1 line, a JSON object, `ok: true` (binary `CLAUDE_CODE_EXECPATH` 2.1.299, Python 3.13.1, uv 0.5.13, git 2.34.1.windows.1, `spawn: cim-hidden`). uv's `Reading inline script metadata…` line goes to stderr. +- Deviations from plan (digest): + - Argument sets for the 1.2 commands are already declared in argparse from the digest (`train --base --brief [--block --phase]`, `fork --base --block --brief [--phase --mode --paths --max-minutes --max-budget-usd --force]`, `continue --job --brief [--max-minutes]`, `wait [jobs] --all --any --timeout`, `expire --base|--all`, `kill --job`, `_supervise --job`). 1.2 may change them. + - The N3 guard is already in the dispatch for `train`, `fork`, `continue`, `rebase` (exit 2 before "not implemented" when `SPIRE_FORKS_CHILD` is set); `tick` left for Phase 2. + - `--help` returns `{"help": …, "commands": […]}` on stdout with exit 0 (keeps the invariant). Unknown extra arguments are a usage error (exit 2), except on the Phase 2 stubs, which tolerate them until Phase 2. + - Uncaught exceptions: exit 2, `{"error":"internal error","type","detail"}`, traceback on stderr (the digest defines no code for it). + - `init` takes more flags than the digest lists: `--ttl-s`, `--stall-after-s`, and one per threshold (`--rebase-commits`, `--rebase-lines`, `--rebase-delta-tokens`, `--max-base-tokens`, `--max-rebases`). Thresholds sit nested under `config.thresholds`. Config also carries `binary_resolved_by` and `model_admission` (how, not what; ignored by the change comparison). + - `where` takes an optional `--binary` (tests use a stub) and also reports `python.supervisor_interpreter`, `platform`, `scrub.drop_patterns`; on failure it carries both `reason` and `error`. + - `preamble.md` adds two lines beyond the digest: "nobody can answer a question mid-run" and "never push, rebase or reset" (mirrors the deny list). +- Flagged for review: + - `--effort` is restricted to `low|medium|high|xhigh|max` and `--permission-mode` to `default|acceptEdits|dontAsk|bypassPermissions|auto` (`plan` refused). Both sets are assumed, not checked against `claude --help`. + - Fingerprint launch dict: `model, effort, permission_mode, tools, allowed_tools, disallowed_tools, mcp {strict, config path}, append_system_prompt {path, sha256}, agent {name, path, sha256} | null, cwd, binary {path, version, size, mtime}, env.pinned` (only the two TTL/autoupdater vars; `SPIRE_FORKS_CHILD` and `CLAUDE_CODE_EXECPATH` are added at launch by `scrubbed_env`, outside the hash). The mcp config path is per run, so the hash differs between runs by construction. + - `--allowed-tools` / `--disallowed-tools` are repeatable lists and each value is also split on commas outside parentheses (`Bash(a,b)` stays one rule). + - Stale-lock breaking re-reads the lock and removes it only if unchanged; two breakers racing still have a narrow window. Release removes the lock only if its token is still ours. Lock wait is 45 s (longer than the 30 s stale age). + - C8: an unreadable or invalid settings file is skipped with a stderr note; `availableModels: []` counts as a list that admits nothing. + - Host env has `ANTHROPIC_BASE_URL`; it is on neither list, so it is kept for the child and listed under `scrub.unknown` by `where` (the P0 probes kept it too and auth worked). The coordinator may want it on a list. + - `pid_alive` treats `ERROR_ACCESS_DENIED` from `OpenProcess` as alive with an unverifiable creation time. + - Git Bash paths whose first segment is not a single letter (`/tmp/x`) are treated as rooted on the cwd's drive, not mapped to Git's install dir. + - The model regex is the digest's; it admits ids like `claude-sonnet` with no version. +- 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: see the follow-up hash commit below. diff --git a/plugins/spire/skills/forks/forks.py b/plugins/spire/skills/forks/forks.py new file mode 100644 index 0000000..bfef6de --- /dev/null +++ b/plugins/spire/skills/forks/forks.py @@ -0,0 +1,1124 @@ +# /// script +# requires-python = ">=3.10" +# dependencies = [] +# /// +"""spire:forks helper: trains one headless implementer per build and forks it per block. + +Run as `uv run /forks.py [args]`. Every call prints exactly one JSON +object on stdout (newline-terminated); diagnostics go to stderr. Exit codes: 0 ok, 2 usage +error or refused, 3 `wait` condition not met at timeout, 4 `wait` condition met and a job +in it did not end `ok`. + +Standard library only. State lives next to the run log: `-forks/`. +""" +from __future__ import annotations + +import argparse +import contextlib +import datetime as _dt +import hashlib +import json +import ntpath +import os +import platform +import posixpath +import re +import shutil +import subprocess +import sys +import time +import traceback +import uuid + +# -------------------------------------------------------------------------------------- +# Constants +# -------------------------------------------------------------------------------------- + +SCHEMA = 1 + +EXIT_OK = 0 +EXIT_USAGE = 2 +EXIT_WAIT_TIMEOUT = 3 +EXIT_JOB_FAILED = 4 + +IS_WINDOWS = os.name == "nt" + +# How the detached supervisor is created (C6, P0.1 rung 3 hidden; POSIX: new session). +SPAWN_METHOD = "cim-hidden" if IS_WINDOWS else "setsid" +CREATE_NO_WINDOW = 0x08000000 +_NOWIN = CREATE_NO_WINDOW if IS_WINDOWS else 0 + +# Registry file semantics (C3a). +LOCK_STALE_S = 30.0 +LOCK_WAIT_S = 45.0 # longer than LOCK_STALE_S, so a waiter can break a stale lock +LOCK_POLL_S = 0.05 +RETRY_TRIES = 40 # PermissionError retries for replace and read (>= 20) +RETRY_SLEEP_S = 0.05 + +MODEL_RE = re.compile(r"^claude-[a-z0-9-]+(\[1m\])?$") +ONE_M_SUFFIX = "[1m]" +EFFORTS = ("low", "medium", "high", "xhigh", "max") +PERMISSION_MODES = ("default", "acceptEdits", "dontAsk", "bypassPermissions", "auto") +REFUSED_PERMISSION_MODES = ("plan",) + +DEFAULT_TOOLS = "Bash,Read,Edit,Write,Glob,Grep,Skill" +DEFAULT_DISALLOWED = ["Bash(git push:*)", "Bash(git rebase:*)", "Bash(git reset --hard:*)", + "Bash(claude:*)"] + +CONFIG_DEFAULTS = { + "parallel_cap": 2, + "orphan_after": 1200, # N1, seconds; > 2 x 570 + "pings": "on", # P0.6 passed; used in Phase 2 + "ttl_s": 3600, + "max_minutes": 90, + "stall_after_s": 900, +} +THRESHOLD_DEFAULTS = { + "rebase_commits": 6, + "rebase_lines": 1500, + "rebase_delta_tokens": 12000, + "max_base_tokens": 250000, + "max_rebases": 4, +} + +# Env pinned into every launch's fingerprint (design 3.3). +ENV_PINNED_LAUNCH = {"CLAUDE_CODE_PROMPT_CACHE_TTL": "1h", "DISABLE_AUTOUPDATER": "1"} + +# --- Env scrub: one constant (design 4 + plan 2 + P0.0) --- +SCRUB_DROP = frozenset({ + # the twelve Desktop vars (design 4, research E1) + "CLAUDE_CODE_MESSAGING_SOCKET", "CLAUDE_CODE_MESSAGING_TOKEN", "CLAUDE_CODE_HOST_SESSION_ID", + "CLAUDE_CODE_ENTRYPOINT", "CLAUDE_CODE_DESKTOP_APP_VERSION", + "CLAUDE_CODE_SDK_HAS_HOST_AUTH_REFRESH", "CLAUDE_CODE_TERMINAL_MCP_TOOLS", + "CLAUDE_CODE_REPORT_FINDINGS", "CLAUDE_CODE_SESSION_ID", "CLAUDE_AGENT_SDK_VERSION", + "CLAUDE_CODE_ENABLE_ASK_USER_QUESTION_TOOL", "CLAUDE_CODE_ENABLE_SDK_FILE_CHECKPOINTING", + # fingerprint and auth path + "CLAUDE_EFFORT", "ANTHROPIC_MODEL", "ANTHROPIC_SMALL_FAST_MODEL", "CLAUDE_CODE_SUBAGENT_MODEL", + "ENABLE_PROMPT_CACHING_1H", "FORCE_PROMPT_CACHING_5M", "CLAUDE_CODE_SUBAGENT_PROMPT_CACHE_TTL", + "ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", "CLAUDE_CODE_USE_BEDROCK", "CLAUDE_CODE_USE_VERTEX", + # uv (plan 2, P0.0) + "VIRTUAL_ENV", +}) +SCRUB_DROP_PATTERNS = (re.compile(r"^ANTHROPIC_DEFAULT_.+_MODEL$"), re.compile(r"^UV_")) +SCRUB_KEEP = frozenset({ + "CLAUDECODE", "CLAUDE_CODE_CHILD_SESSION", "AI_AGENT", "CLAUDE_PID", "CLAUDE_CODE_GIT_BASH_PATH", + "PATH", "HOME", "USERPROFILE", +}) +SCRUB_PINNED = {**ENV_PINNED_LAUNCH, "SPIRE_FORKS_CHILD": "1"} +SCRUB_PINNED_BINARY = "CLAUDE_CODE_EXECPATH" # set to the stored absolute binary path +CHILD_MARKER = "SPIRE_FORKS_CHILD" # N3 +SCRUB_WATCH_PREFIXES = ("CLAUDE_", "ANTHROPIC_") # `where` lists unknown vars with these + +# C8: the user settings file. Module-level so tests can point it at a temp file. +USER_SETTINGS_PATH: str | None = None + +PHASE2_COMMANDS = ("rebase", "usage", "tick") +CHILD_REFUSED = ("train", "fork", "continue", "rebase") # N3 (tick pings: Phase 2) + +# -------------------------------------------------------------------------------------- +# Errors, output, clock +# -------------------------------------------------------------------------------------- + + +class ForksError(Exception): + """A refusal or usage error: exit `code`, payload printed as the stdout JSON object.""" + + def __init__(self, error: str, code: int = EXIT_USAGE, **fields): + super().__init__(error) + self.code = code + self.payload = {"error": error, **fields} + + +class _HelpRequested(Exception): + def __init__(self, text: str): + super().__init__("help") + self.text = text + + +def _emit(obj: dict) -> None: + sys.stdout.write(json.dumps(obj) + "\n") + sys.stdout.flush() + + +def _diag(msg: str) -> None: + sys.stderr.write("forks: " + msg + "\n") + sys.stderr.flush() + + +def _now() -> float: + """The only clock (epoch seconds). Tests monkeypatch it.""" + return time.time() + + +def _iso(t: float | None = None) -> str: + """ISO 8601 with offset, from `_now()` unless given.""" + if t is None: + t = _now() + return _dt.datetime.fromtimestamp(t).astimezone().isoformat(timespec="seconds") + + +def _sha256_bytes(b: bytes) -> str: + return hashlib.sha256(b).hexdigest() + + +def _sha256_file(path: str) -> str: + with open(path, "rb") as f: + return _sha256_bytes(f.read()) + + +def canonical_json(obj) -> str: + return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + + +def fingerprint_hash(launch: dict) -> str: + return _sha256_bytes(canonical_json(launch).encode("utf-8")) + + +# -------------------------------------------------------------------------------------- +# Paths +# -------------------------------------------------------------------------------------- + +_GITBASH_DRIVE = re.compile(r"^/([A-Za-z])(?=/|$)") + + +def norm_path(p: str, cwd: str | None = None, windows: bool | None = None) -> str: + """Normalise relative, `G:\\…`, `G:/…` and Git Bash `/g/…` forms. + + Windows: absolute path with backslashes and an upper-case drive letter. + POSIX: `os.path.abspath` semantics (relative to `cwd`). + """ + if windows is None: + windows = IS_WINDOWS + if not isinstance(p, str) or not p.strip(): + raise ForksError("empty path") + p = p.strip() + if not windows: + base = cwd if cwd is not None else os.getcwd() + return posixpath.normpath(posixpath.join(base, p)) + m = _GITBASH_DRIVE.match(p) + if m: + p = m.group(1).upper() + ":" + (p[2:] or "/") + p = p.replace("/", "\\") + drive, rest = ntpath.splitdrive(p) + if not (drive and rest.startswith("\\")): + base = norm_path(cwd if cwd is not None else os.getcwd(), windows=True) + if not drive and rest.startswith("\\"): + p = ntpath.splitdrive(base)[0] + rest # rooted, no drive: cwd's drive + else: + p = ntpath.join(base, p) + p = ntpath.normpath(p) + drive, rest = ntpath.splitdrive(p) + if not drive or not rest.startswith("\\"): + raise ForksError("cannot normalise path", path=p) + if len(drive) == 2 and drive[1] == ":": + p = drive.upper() + rest + return p + + +def state_dir_for(run: str) -> str: + """`reports/-run.md` -> `reports/-run-forks` (a sibling of the run log).""" + run_n = norm_path(run) + if not run_n.lower().endswith(".md"): + raise ForksError("--run must name the run log .md file", run=run_n) + return run_n[:-3] + "-forks" + + +STATE_SUBDIRS = ("jobs", "briefs", "notes", "results", "logs", "raw") +STATE_GITIGNORE = "raw/\n*.lock\n" +MCP_EMPTY = {"mcpServers": {}} + + +def registry_path(state: str) -> str: + return os.path.join(state, "registry.json") + + +def lock_path(state: str) -> str: + return os.path.join(state, "registry.lock") + + +def ledger_path(state: str) -> str: + return os.path.join(state, "ledger.jsonl") + + +def mcp_empty_path(state: str) -> str: + return os.path.join(state, "mcp-empty.json") + + +def ensure_state(state: str) -> None: + """Create the state dir and its fixed files. Idempotent.""" + os.makedirs(state, exist_ok=True) + for d in STATE_SUBDIRS: + os.makedirs(os.path.join(state, d), exist_ok=True) + gi = os.path.join(state, ".gitignore") + if not os.path.exists(gi): + atomic_write_bytes(gi, STATE_GITIGNORE.encode("utf-8")) + mcp = mcp_empty_path(state) + if not os.path.exists(mcp): + atomic_write_bytes(mcp, (json.dumps(MCP_EMPTY) + "\n").encode("utf-8")) + + +# -------------------------------------------------------------------------------------- +# Atomic files, retries, the registry lock (C3) +# -------------------------------------------------------------------------------------- + + +def _retry(fn, tries: int = RETRY_TRIES, sleep_s: float = RETRY_SLEEP_S): + """Call fn(); retry PermissionError (Windows sharing violations). Returns (value, attempts).""" + for attempt in range(1, tries + 1): + try: + return fn(), attempt + except PermissionError: + if attempt == tries: + raise + time.sleep(sleep_s) + raise AssertionError("unreachable") + + +def atomic_write_bytes(path: str, data: bytes) -> int: + """Write to a temp file in the same dir, then `os.replace` with retries. Returns attempts.""" + d = os.path.dirname(path) or "." + tmp = os.path.join(d, ".%s.%d.%s.tmp" % (os.path.basename(path), os.getpid(), uuid.uuid4().hex[:8])) + with open(tmp, "wb") as f: + f.write(data) + f.flush() + os.fsync(f.fileno()) + try: + _, attempts = _retry(lambda: os.replace(tmp, path)) + except BaseException: + with contextlib.suppress(OSError): + os.remove(tmp) + raise + return attempts + + +def write_json_atomic(path: str, obj) -> int: + return atomic_write_bytes(path, (json.dumps(obj, indent=1, sort_keys=True) + "\n").encode("utf-8")) + + +def read_bytes_retry(path: str) -> bytes: + def _read(): + with open(path, "rb") as f: + return f.read() + return _retry(_read)[0] + + +def read_json(path: str): + """Read a JSON file with PermissionError retries. Returns None if it does not exist.""" + try: + data = read_bytes_retry(path) + except FileNotFoundError: + return None + try: + return json.loads(data.decode("utf-8")) + except ValueError as e: + raise ForksError("corrupt state file", path=path, detail=str(e)) + + +class RegistryLock: + """`registry.lock`: create-exclusive; a lock older than LOCK_STALE_S may be broken.""" + + def __init__(self, state: str, timeout: float = LOCK_WAIT_S): + self.path = lock_path(state) + self.timeout = timeout + self.token = uuid.uuid4().hex + self.held = False + self.broke_stale = False + + def _content(self) -> bytes: + return json.dumps({"pid": os.getpid(), "ctime": proc_ctime(os.getpid()), + "at": _now(), "token": self.token}).encode("utf-8") + + def _try_create(self) -> bool: + flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_BINARY", 0) + try: + fd = os.open(self.path, flags) + except FileExistsError: + return False + except PermissionError: # Windows: the old lock is being deleted + return False + try: + os.write(fd, self._content()) + finally: + os.close(fd) + return True + + def _read_lock(self): + """(raw bytes, age in seconds) of the current lock, or (None, None) if gone.""" + try: + raw = read_bytes_retry(self.path) + mtime = os.path.getmtime(self.path) + except (FileNotFoundError, PermissionError): + return None, None + at = None + try: + at = float(json.loads(raw.decode("utf-8")).get("at")) + except (ValueError, TypeError, AttributeError): + pass + if at is None: + at = mtime + return raw, _now() - at + + def acquire(self) -> "RegistryLock": + deadline = time.monotonic() + self.timeout + while True: + if self._try_create(): + self.held = True + return self + raw, age = self._read_lock() + if raw is not None and age is not None and age > LOCK_STALE_S: + # Break it only if it is still the same lock we judged stale. + again, _ = self._read_lock() + if again == raw: + with contextlib.suppress(FileNotFoundError, PermissionError): + os.remove(self.path) + self.broke_stale = True + _diag("broke stale registry lock (age %.1f s)" % age) + continue + if time.monotonic() >= deadline: + raise ForksError("registry locked", lock=self.path, waited_s=self.timeout) + time.sleep(LOCK_POLL_S) + + def release(self) -> None: + if not self.held: + return + self.held = False + try: + raw = read_bytes_retry(self.path) + except FileNotFoundError: + return + try: + mine = json.loads(raw.decode("utf-8")).get("token") == self.token + except (ValueError, AttributeError): + mine = False + if mine: + with contextlib.suppress(FileNotFoundError): + _retry(lambda: os.remove(self.path)) + else: + _diag("registry lock was taken over while held; not removing it") + + def __enter__(self): + return self.acquire() + + def __exit__(self, *exc): + self.release() + return False + + +def new_registry() -> dict: + return {"schema": SCHEMA, "config": None, "bases": {}, "jobs": {}, "run": {"created_at": _iso()}} + + +def read_registry(state: str) -> dict | None: + reg = read_json(registry_path(state)) + if reg is None: + return None + if not isinstance(reg, dict) or reg.get("schema") != SCHEMA: + raise ForksError("unsupported registry schema", path=registry_path(state), + schema=(reg or {}).get("schema") if isinstance(reg, dict) else None, + expected=SCHEMA) + return reg + + +def write_registry(state: str, reg: dict) -> int: + """Caller must hold RegistryLock(state).""" + reg["schema"] = SCHEMA + return write_json_atomic(registry_path(state), reg) + + +def ledger_append(state: str, line: dict) -> None: + """Append one JSON line. Caller must hold RegistryLock(state) (C3c).""" + data = (json.dumps(line, sort_keys=True) + "\n").encode("utf-8") + + def _append(): + with open(ledger_path(state), "ab") as f: + f.write(data) + _retry(_append) + + +# -------------------------------------------------------------------------------------- +# Processes: pid + creation time (A19), subprocess helpers, git (C7) +# -------------------------------------------------------------------------------------- + +if IS_WINDOWS: + import ctypes + import ctypes.wintypes as _wt + + _k32 = ctypes.WinDLL("kernel32", use_last_error=True) + _k32.OpenProcess.argtypes = [_wt.DWORD, _wt.BOOL, _wt.DWORD] + _k32.OpenProcess.restype = _wt.HANDLE + _k32.GetProcessTimes.argtypes = [_wt.HANDLE] + [ctypes.POINTER(_wt.FILETIME)] * 4 + _k32.GetProcessTimes.restype = _wt.BOOL + _k32.GetExitCodeProcess.argtypes = [_wt.HANDLE, ctypes.POINTER(_wt.DWORD)] + _k32.GetExitCodeProcess.restype = _wt.BOOL + _k32.CloseHandle.argtypes = [_wt.HANDLE] + _k32.CloseHandle.restype = _wt.BOOL + _PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 + _STILL_ACTIVE = 259 + _ERROR_ACCESS_DENIED = 5 + + +def proc_ctime(pid: int) -> int | None: + """Process creation time: Windows FILETIME ticks; POSIX /proc start time; else None.""" + if IS_WINDOWS: + h = _k32.OpenProcess(_PROCESS_QUERY_LIMITED_INFORMATION, False, int(pid)) + if not h: + return None + try: + c, e, k, u = _wt.FILETIME(), _wt.FILETIME(), _wt.FILETIME(), _wt.FILETIME() + if not _k32.GetProcessTimes(h, ctypes.byref(c), ctypes.byref(e), ctypes.byref(k), + ctypes.byref(u)): + return None + return (c.dwHighDateTime << 32) | c.dwLowDateTime + finally: + _k32.CloseHandle(h) + try: + with open("/proc/%d/stat" % int(pid), "rb") as f: + stat = f.read().decode("ascii", "replace") + return int(stat[stat.rindex(")") + 2:].split()[19]) + except (OSError, ValueError, IndexError): + return None + + +def pid_record(pid: int) -> dict: + """{"pid", "ctime"}: the pair every stored pid is compared by.""" + return {"pid": int(pid), "ctime": proc_ctime(pid)} + + +def pid_alive(pid: int, ctime: int | None = None) -> bool: + """True if `pid` runs and (when both are known) its creation time equals `ctime`.""" + if pid is None or int(pid) <= 0: + return False + pid = int(pid) + if IS_WINDOWS: + h = _k32.OpenProcess(_PROCESS_QUERY_LIMITED_INFORMATION, False, pid) + if not h: + # Exists but not openable: alive, creation time unverifiable. + return ctypes.get_last_error() == _ERROR_ACCESS_DENIED + try: + code = _wt.DWORD() + if not _k32.GetExitCodeProcess(h, ctypes.byref(code)) or code.value != _STILL_ACTIVE: + return False + finally: + _k32.CloseHandle(h) + else: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + pass + if ctime is not None: + now_ct = proc_ctime(pid) + if now_ct is not None and now_ct != ctime: + return False + return True + + +def record_alive(rec: dict | None) -> bool: + return bool(rec) and pid_alive(rec.get("pid"), rec.get("ctime")) + + +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, + timeout=timeout, creationflags=_NOWIN) + + +def git(args, cwd=None, timeout=60) -> subprocess.CompletedProcess: + """Every helper git call goes through here: `git --no-optional-locks …` (C7).""" + return _run(["git", "--no-optional-locks", *args], cwd=cwd, timeout=timeout) + + +def git_toplevel(cwd: str | None = None) -> str | None: + try: + r = git(["rev-parse", "--show-toplevel"], cwd=cwd) + except (OSError, subprocess.SubprocessError): + return None + if r.returncode != 0: + return None + out = r.stdout.decode("utf-8", "replace").strip() + return norm_path(out) if out else None + + +# -------------------------------------------------------------------------------------- +# Env scrub +# -------------------------------------------------------------------------------------- + + +def scrub_class(name: str) -> str | None: + """'drop', 'pinned', 'keep' or None (not on any list).""" + n = name.upper() + if n in SCRUB_DROP or any(p.match(n) for p in SCRUB_DROP_PATTERNS): + return "drop" + if n in SCRUB_PINNED or n == SCRUB_PINNED_BINARY: + return "pinned" + if n in SCRUB_KEEP: + return "keep" + return None + + +def _venv_prefixes(environ) -> list[str]: + out = [] + venv = environ.get("VIRTUAL_ENV") + if venv: + out.append(venv) + if os.path.normcase(os.path.abspath(sys.prefix)) != os.path.normcase(os.path.abspath(sys.base_prefix)): + out.append(sys.prefix) + return [os.path.normcase(os.path.normpath(os.path.abspath(x))) for x in out] + + +def _under(path: str, prefixes) -> bool: + try: + p = os.path.normcase(os.path.normpath(os.path.abspath(path))) + except ValueError: + return False + return any(p == px or p.startswith(px.rstrip("\\/") + os.sep) for px in prefixes) + + +def scrubbed_env(binary: str, environ=None) -> dict: + """The child env for `claude`: host env minus the drop list, venv PATH entries removed, + pinned vars on top, `CLAUDE_CODE_EXECPATH` = the stored absolute binary path.""" + if environ is None: + environ = os.environ + prefixes = _venv_prefixes(environ) + env = {} + for k, v in environ.items(): + cls = scrub_class(k) + if cls in ("drop", "pinned"): + continue + if k.upper() == "PATH": + v = os.pathsep.join(e for e in v.split(os.pathsep) if e and not _under(e, prefixes)) + env[k] = v + env.update(SCRUB_PINNED) + env[SCRUB_PINNED_BINARY] = binary + return env + + +def scrub_unknown(environ=None) -> list[str]: + """Host CLAUDE_* / ANTHROPIC_* vars on neither list (kept, but worth a look).""" + if environ is None: + environ = os.environ + return sorted(k for k in environ + if k.upper().startswith(SCRUB_WATCH_PREFIXES) and scrub_class(k) is None) + + +def scrub_report(environ=None) -> dict: + return { + "drop": sorted(SCRUB_DROP), + "drop_patterns": [p.pattern for p in SCRUB_DROP_PATTERNS], + "drop_path_entries": "under VIRTUAL_ENV, or under sys.prefix when it differs from sys.base_prefix", + "keep": sorted(SCRUB_KEEP), + "keep_rule": "everything not dropped is kept", + "pinned": dict(SCRUB_PINNED, **{SCRUB_PINNED_BINARY: ""}), + "unknown": scrub_unknown(environ), + } + + +# -------------------------------------------------------------------------------------- +# Binary, uv, python +# -------------------------------------------------------------------------------------- + + +def resolve_binary(explicit: str | None = None, environ=None) -> tuple[str | None, str | None]: + """(absolute path, how): `--binary`, else $CLAUDE_CODE_EXECPATH, else PATH `claude`.""" + if environ is None: + environ = os.environ + if explicit: + return norm_path(explicit), "--binary" + ep = environ.get("CLAUDE_CODE_EXECPATH") + if ep: + return norm_path(ep), "CLAUDE_CODE_EXECPATH" + w = shutil.which("claude") + if w: + return norm_path(w), "PATH" + return None, None + + +def _binary_version(path: str) -> str | None: + try: + r = _run([path, "--version"], timeout=60) + except (OSError, subprocess.SubprocessError): + return None + toks = r.stdout.decode("utf-8", "replace").split() + return toks[0] if r.returncode == 0 and toks else None + + +def binary_info(path: str) -> dict: + """{"path", "version", "size", "mtime"} of an existing binary; raises ForksError otherwise.""" + if not path or not os.path.isfile(path): + raise ForksError("claude binary not found", binary=path) + st = os.stat(path) + version = _binary_version(path) + if not version: + raise ForksError("claude binary did not report a version", binary=path) + return {"path": path, "version": version, "size": st.st_size, "mtime": st.st_mtime} + + +def _uv_report(environ=None) -> dict: + if environ is None: + environ = os.environ + out = {"path": None, "version": None} + w = shutil.which("uv") + if w: + out["path"] = norm_path(w) + try: + r = _run([w, "--version"], timeout=30) + toks = r.stdout.decode("utf-8", "replace").split() + out["version"] = toks[1] if len(toks) > 1 else (toks[0] if toks else None) + except (OSError, subprocess.SubprocessError) as e: + out["error"] = str(e) + if environ.get("SCRIPTS_UV"): + out["SCRIPTS_UV"] = environ["SCRIPTS_UV"] + return out + + +def _git_report() -> dict: + out = {"path": None, "version": None} + w = shutil.which("git") + if w: + out["path"] = norm_path(w) + try: + r = git(["--version"], timeout=30) + toks = r.stdout.decode("utf-8", "replace").split() + out["version"] = toks[2] if len(toks) > 2 else None + except (OSError, subprocess.SubprocessError) as e: + out["error"] = str(e) + return out + + +def script_path() -> str: + return norm_path(os.path.abspath(__file__)) + + +def preamble_path() -> str: + return norm_path(os.path.join(os.path.dirname(os.path.abspath(__file__)), "preamble.md")) + + +def supervisor_interpreter() -> str: + """C6: the base interpreter when `sys.executable` is a venv launcher.""" + base = getattr(sys, "_base_executable", None) + return base if base and base != sys.executable else sys.executable + + +# -------------------------------------------------------------------------------------- +# C8: availableModels admission +# -------------------------------------------------------------------------------------- + + +def user_settings_path() -> str: + if USER_SETTINGS_PATH: + return USER_SETTINGS_PATH + return os.path.join(os.path.expanduser("~"), ".claude", "settings.json") + + +def settings_files(cwd: str) -> list[str]: + return [user_settings_path(), + os.path.join(cwd, ".claude", "settings.json"), + os.path.join(cwd, ".claude", "settings.local.json")] + + +def model_family(model: str) -> str | None: + toks = model.replace(ONE_M_SUFFIX, "").split("-") + return toks[1] if len(toks) > 1 else None + + +def model_admission(model: str, cwd: str) -> dict: + """Apply the P0.5 rule. Returns {"admitted", "lists": [{"file", "availableModels"}], "via"}.""" + lists = [] + for f in settings_files(cwd): + try: + data = read_json(f) + except ForksError as e: + _diag("ignoring unreadable settings file %s (%s)" % (f, e.payload.get("detail"))) + continue + except OSError: + continue + if isinstance(data, dict) and isinstance(data.get("availableModels"), list): + lists.append({"file": f, "availableModels": [str(x) for x in data["availableModels"]]}) + if not lists: + return {"admitted": True, "lists": [], "via": "no availableModels list"} + candidates = [("exact id", model)] + if model.endswith(ONE_M_SUFFIX): + candidates.append(("id without [1m]", model[: -len(ONE_M_SUFFIX)])) + fam = model_family(model) + if fam: + candidates.append(("family alias", fam)) + for via, cand in candidates: + for entry in lists: + if cand in entry["availableModels"]: + return {"admitted": True, "lists": lists, "via": "%s %r in %s" % (via, cand, entry["file"])} + return {"admitted": False, "lists": lists, "via": None} + + +# -------------------------------------------------------------------------------------- +# Argument parsing (argparse that never prints to stdout or exits on its own) +# -------------------------------------------------------------------------------------- + + +class _Parser(argparse.ArgumentParser): + def error(self, message): + raise ForksError("usage", detail=message, usage=self.format_usage().strip()) + + def print_help(self, file=None): + raise _HelpRequested(self.format_help()) + + def exit(self, status=0, message=None): + if message: + _diag(message.strip()) + raise ForksError("usage", detail=(message or "").strip() or "argument parser exit") + + +def _split_top_level_commas(s: str) -> list[str]: + """Split on commas outside parentheses: `Bash(a,b),Read` -> [`Bash(a,b)`, `Read`].""" + out, depth, cur = [], 0, [] + for ch in s: + if ch == "(": + depth += 1 + elif ch == ")": + depth = max(0, depth - 1) + if ch == "," and depth == 0: + out.append("".join(cur).strip()) + cur = [] + else: + cur.append(ch) + out.append("".join(cur).strip()) + return [x for x in out if x] + + +def _tool_list(values) -> list[str] | None: + if values is None: + return None + out = [] + for v in values: + out.extend(_split_top_level_commas(v)) + return out + + +def _pos_int(s: str) -> int: + try: + v = int(s) + except ValueError: + raise argparse.ArgumentTypeError("not an integer: %r" % s) + if 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") + + +def build_parser() -> _Parser: + p = _Parser(prog="forks.py", description="spire:forks helper (one JSON object on stdout).") + sub = p.add_subparsers(dest="command", metavar="command", parser_class=_Parser) + sub.required = True + + def cmd(name, help_text): + sp = (sub.add_parser(name, help=help_text, description=help_text) if help_text + else sub.add_parser(name)) + if name != "where": + sp.add_argument("--run", required=True, help="path of the run log .md") + return sp + + w = cmd("where", "Preflight report: script, python, uv, binary, git, spawn method, env scrub.") + w.add_argument("--binary", help="claude binary to report instead of the default resolution") + + i = cmd("init", "Create the state dir and the launch config (idempotent).") + i.add_argument("--model", required=True, help="full model id, e.g. claude-opus-5-5[1m]") + i.add_argument("--effort", default="high", choices=EFFORTS) + i.add_argument("--permission-mode", default="dontAsk") + i.add_argument("--tools", default=DEFAULT_TOOLS, help="comma-separated tool names") + i.add_argument("--allowed-tools", nargs="+", action="extend", default=None, + help="allow rules (default: the tools)") + i.add_argument("--disallowed-tools", nargs="+", action="extend", default=None, + help="deny rules (default: git push/rebase/reset --hard, claude)") + i.add_argument("--agent", help="project agent name (/.claude/agents/.md)") + i.add_argument("--cwd", help="working dir of base and forks (default: git toplevel)") + i.add_argument("--binary", help="claude binary (default: $CLAUDE_CODE_EXECPATH, else PATH)") + i.add_argument("--parallel-cap", type=_pos_int, default=CONFIG_DEFAULTS["parallel_cap"]) + i.add_argument("--orphan-after", type=_pos_int, default=CONFIG_DEFAULTS["orphan_after"]) + i.add_argument("--pings", choices=("on", "off"), default=CONFIG_DEFAULTS["pings"]) + i.add_argument("--ttl-s", type=_pos_int, default=CONFIG_DEFAULTS["ttl_s"]) + i.add_argument("--max-minutes", type=_pos_int, default=CONFIG_DEFAULTS["max_minutes"]) + i.add_argument("--stall-after-s", type=_pos_int, default=CONFIG_DEFAULTS["stall_after_s"]) + for k, v in THRESHOLD_DEFAULTS.items(): + i.add_argument("--" + k.replace("_", "-"), type=_pos_int, default=v) + + t = cmd("train", "Train a new base (headless session).") + t.add_argument("--base", required=True) + t.add_argument("--brief", required=True) + t.add_argument("--block") + t.add_argument("--phase") + + f = cmd("fork", "Fork a base for one block.") + f.add_argument("--base", required=True) + f.add_argument("--block", required=True) + f.add_argument("--phase") + 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") + + 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) + + 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) + + cmd("list", "Show every base and job.") + + e = cmd("expire", "Retire a base (or all); running jobs of it are killed.") + eg = e.add_mutually_exclusive_group(required=True) + eg.add_argument("--base") + eg.add_argument("--all", action="store_true") + + k = cmd("kill", "Kill a job's supervisor and claude process tree.") + k.add_argument("--job", required=True) + + s = cmd("_supervise", None) # internal + s.add_argument("--job", required=True) + + for name, text in (("rebase", "Rebase a base onto the current state (Phase 2)."), + ("usage", "Aggregate the ledger (Phase 2)."), + ("tick", "One keep-warm pass (Phase 2).")): + cmd(name, text) # arguments arrive with Phase 2; until then extra args are tolerated + return p + + +# -------------------------------------------------------------------------------------- +# Commands +# -------------------------------------------------------------------------------------- + + +def cmd_where(args) -> tuple[dict, int]: + reasons = [] + py_ok = sys.version_info >= (3, 10) + if not py_ok: + reasons.append("python < 3.10") + py = {"version": platform.python_version(), "executable": sys.executable, + "base_executable": getattr(sys, "_base_executable", sys.executable), + "supervisor_interpreter": supervisor_interpreter(), "ok": py_ok} + path, how = resolve_binary(args.binary) + binary = {"path": path, "how": how} + if path is None: + reasons.append("no claude binary (no --binary, no $CLAUDE_CODE_EXECPATH, none on PATH)") + else: + try: + binary.update(binary_info(path)) + except ForksError as e: + binary["error"] = e.payload["error"] + reasons.append("%s: %s" % (e.payload["error"], path)) + out = { + "script": script_path(), + "python": py, + "uv": _uv_report(), + "binary": binary, + "git": _git_report(), + "spawn": SPAWN_METHOD, + "scrub": scrub_report(), + "platform": sys.platform, + "ok": not reasons, + } + if reasons: + out["reason"] = "; ".join(reasons) + out["error"] = out["reason"] + return out, EXIT_USAGE + return out, EXIT_OK + + +def _resolve_cwd(arg: str | None) -> str: + if arg: + cwd = norm_path(arg) + else: + cwd = git_toplevel() + if not cwd: + raise ForksError("no --cwd given and the current directory is not inside a git repo") + if not os.path.isdir(cwd): + raise ForksError("--cwd is not a directory", cwd=cwd) + return cwd + + +def build_init_config(args, state: str) -> dict: + """Validate the launch fields and build the config block. Writes nothing.""" + model = args.model + if not MODEL_RE.match(model or ""): + raise ForksError("model must be a full id (aliases are refused)", model=model, + pattern=MODEL_RE.pattern) + mode = args.permission_mode + if mode in REFUSED_PERMISSION_MODES: + raise ForksError("permission mode refused", permission_mode=mode, + detail="plan mode swaps models; use dontAsk") + if mode not in PERMISSION_MODES: + raise ForksError("unknown permission mode", permission_mode=mode, allowed=list(PERMISSION_MODES)) + tools = _tool_list([args.tools]) + if not tools: + raise ForksError("--tools is empty") + allowed = _tool_list(args.allowed_tools) + if allowed is None: + allowed = list(tools) + disallowed = _tool_list(args.disallowed_tools) + if disallowed is None: + disallowed = list(DEFAULT_DISALLOWED) + + cwd = _resolve_cwd(args.cwd) + + adm = model_admission(model, cwd) + if not adm["admitted"]: + files = [x["file"] for x in adm["lists"]] + raise ForksError( + "model not admitted by availableModels", model=model, files=files, + message=("Model %s is not admitted by the availableModels list(s) in: %s. The user must " + "add the id (or its family alias) to one of these files; forks never writes " + "settings." % (model, ", ".join(files)))) + + bpath, how = resolve_binary(args.binary) + if bpath is None: + raise ForksError("no claude binary (no --binary, no $CLAUDE_CODE_EXECPATH, none on PATH)") + binfo = binary_info(bpath) + + pre = preamble_path() + if not os.path.isfile(pre): + raise ForksError("preamble.md missing next to forks.py", path=pre) + + agent = None + if args.agent: + name = args.agent + if not re.match(r"^[A-Za-z0-9_.:-]+$", name) or name in (".", ".."): + raise ForksError("bad agent name", agent=name) + apath = os.path.join(cwd, ".claude", "agents", name + ".md") + if not os.path.isfile(apath): + raise ForksError("agent file not found", agent=name, path=apath) + agent = {"name": name, "path": apath, "sha256": _sha256_file(apath)} + + launch = { + "model": model, + "effort": args.effort, + "permission_mode": mode, + "tools": tools, + "allowed_tools": allowed, + "disallowed_tools": disallowed, + "mcp": {"strict": True, "config": mcp_empty_path(state)}, + "append_system_prompt": {"path": pre, "sha256": _sha256_file(pre)}, + "agent": agent, + "cwd": cwd, + "binary": binfo, + "env": {"pinned": dict(ENV_PINNED_LAUNCH)}, + } + cfg = { + "launch": launch, + "fingerprint": fingerprint_hash(launch), + "binary_resolved_by": how, + "parallel_cap": args.parallel_cap, + "orphan_after": args.orphan_after, + "pings": args.pings, + "ttl_s": args.ttl_s, + "max_minutes": args.max_minutes, + "stall_after_s": args.stall_after_s, + "thresholds": {k: getattr(args, k) for k in THRESHOLD_DEFAULTS}, + "model_admission": adm["via"], + } + return cfg + + +def _diff_fields(a, b, prefix="") -> list[str]: + if isinstance(a, dict) and isinstance(b, dict): + out = [] + for k in sorted(set(a) | set(b)): + out += _diff_fields(a.get(k), b.get(k), prefix + k + ".") + return out + return [] if a == b else [prefix.rstrip(".")] + + +def cmd_init(args) -> tuple[dict, int]: + state = state_dir_for(args.run) + cfg = build_init_config(args, state) # all validation before any write + ensure_state(state) + with RegistryLock(state): + reg = read_registry(state) + created = reg is None + if created: + reg = new_registry() + old = reg.get("config") + changed_fields = [] if old is None else _diff_fields(old, cfg) + # `model_admission` and `binary_resolved_by` describe how, not what: not a change. + changed_fields = [f for f in changed_fields + if f not in ("model_admission", "binary_resolved_by")] + if old is not None and changed_fields and reg.get("bases"): + raise ForksError("init values differ and a base exists; train a new run instead", + fields=changed_fields, bases=sorted(reg["bases"]), state=state) + if old is None or changed_fields: + reg["config"] = cfg + reg.setdefault("run", {})["init_at"] = _iso() + write_registry(state, reg) + else: + cfg = old + return {"ok": True, "state": state, "registry": registry_path(state), "created": created, + "changed": bool(old is not None and changed_fields), "changed_fields": changed_fields, + "config": cfg}, EXIT_OK + + +def _not_implemented(block=None, phase=None): + def f(args): + out = {"error": "not implemented", "command": args.command} + if block: + out["block"] = block + if phase: + out["phase"] = phase + return out, EXIT_USAGE + return f + + +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"), + "rebase": _not_implemented(phase=2), + "usage": _not_implemented(phase=2), + "tick": _not_implemented(phase=2), +} + + +def run(argv) -> tuple[dict, int]: + parser = build_parser() + try: + args, unknown = parser.parse_known_args(argv) + except _HelpRequested as h: + return {"help": h.text, "commands": [c for c in COMMANDS if not c.startswith("_")]}, EXIT_OK + if unknown and args.command not in PHASE2_COMMANDS: + parser.error("unrecognized arguments: %s" % " ".join(unknown)) + if args.command in CHILD_REFUSED and os.environ.get(CHILD_MARKER): + raise ForksError("launching commands are refused inside a fork (%s is set)" % CHILD_MARKER, + command=args.command) + return HANDLERS[args.command](args) + + +def main(argv=None) -> int: + if argv is None: + argv = sys.argv[1:] + try: + out, code = run(argv) + except ForksError as e: + out, code = e.payload, e.code + except Exception as e: # never break the stdout invariant + traceback.print_exc(file=sys.stderr) + out, code = {"error": "internal error", "type": type(e).__name__, "detail": str(e)}, EXIT_USAGE + _emit(out) + return code + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/plugins/spire/skills/forks/preamble.md b/plugins/spire/skills/forks/preamble.md new file mode 100644 index 0000000..1f607e2 --- /dev/null +++ b/plugins/spire/skills/forks/preamble.md @@ -0,0 +1,9 @@ +# Headless implementer (spire:forks) + +You run headless under `claude -p`. You are trained once on a build, then forked once per block; any session of yours may later be resumed or forked again. Nobody can answer a question mid-run: decide, act, and record what you decided. + +- Do the work yourself. Never dispatch agents, and never start `claude` or the forks helper (`forks.py`). +- Never use `run_in_background` on Bash. Run every command in the foreground. +- Stage and commit by explicit paths only, and only when your brief says this fork commits. Never push, rebase or reset. +- Write gate and test output to `.txt` files under the state dir's `logs/` folder (the brief names the state dir). +- When you finish a block, write the handoff note to the path the brief names, in the run-log handoff-note format (`plugins/spire/skills/implementation/references/run-log.md` in the Spire plugin), and stamp your exact model id. diff --git a/plugins/spire/skills/forks/tests/test_forks.py b/plugins/spire/skills/forks/tests/test_forks.py new file mode 100644 index 0000000..56d2b12 --- /dev/null +++ b/plugins/spire/skills/forks/tests/test_forks.py @@ -0,0 +1,673 @@ +# /// script +# requires-python = ">=3.10" +# dependencies = [] +# /// +"""Unit tests for the spire:forks helper. + +Run from the repo root: `uv run plugins/spire/skills/forks/tests/test_forks.py`. +Imports forks.py by path. Everything generated (stub binary, state dirs, settings files) +lives under `tempfile`; nothing touches ~/.claude or the repo's own state. +""" +from __future__ import annotations + +import contextlib +import importlib.util +import io +import json +import os +import shutil +import subprocess +import sys +import tempfile +import threading +import time +import unittest +from unittest import mock + +HERE = os.path.dirname(os.path.abspath(__file__)) +FORKS_PY = os.path.join(os.path.dirname(HERE), "forks.py") + + +def _load_forks(): + spec = importlib.util.spec_from_file_location("forks", FORKS_PY) + mod = importlib.util.module_from_spec(spec) + sys.modules["forks"] = mod + spec.loader.exec_module(mod) + return mod + + +forks = _load_forks() +IS_WINDOWS = os.name == "nt" +STUB_VERSION = "2.1.999" + + +def make_stub_binary(d: str, version: str = STUB_VERSION) -> str: + """A fake `claude` that answers --version. Generated at runtime (A18), never committed.""" + if IS_WINDOWS: + p = os.path.join(d, "claude-stub.cmd") + with open(p, "w", newline="\r\n") as f: + f.write("@echo off\necho %s (Claude Code stub)\n" % version) + else: + p = os.path.join(d, "claude-stub") + with open(p, "w", newline="\n") as f: + f.write("#!/bin/sh\necho '%s (Claude Code stub)'\n" % version) + os.chmod(p, 0o755) + return p + + +def call(argv): + """Run forks.main in-process. Returns (exit code, parsed JSON, raw stdout).""" + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + code = forks.main(list(argv)) + raw = out.getvalue() + return code, json.loads(raw), raw + + +def child_env(): + env = dict(os.environ) + env.pop(forks.CHILD_MARKER, None) + return env + + +class Base(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="forks-test-") + self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True) + env_patch = mock.patch.dict(os.environ) + env_patch.start() + self.addCleanup(env_patch.stop) + os.environ.pop(forks.CHILD_MARKER, None) + + def assert_one_json_line(self, raw: str): + self.assertTrue(raw.endswith("\n"), repr(raw[-40:])) + self.assertEqual(raw.count("\n"), 1, raw) + obj = json.loads(raw) + self.assertIsInstance(obj, dict) + return obj + + +# -------------------------------------------------------------------------------------- +class PathTests(Base): + def test_windows_forms(self): + cwd = "G:\\dev\\proj" + cases = { + "docs\\a.md": "G:\\dev\\proj\\docs\\a.md", + "docs/a.md": "G:\\dev\\proj\\docs\\a.md", + "G:\\x\\y.md": "G:\\x\\y.md", + "G:/x/y.md": "G:\\x\\y.md", + "g:/x/y.md": "G:\\x\\y.md", + "/g/x/y.md": "G:\\x\\y.md", + "/c/Users/me": "C:\\Users\\me", + "/c": "C:\\", + "/c/": "C:\\", + "../other/./z.md": "G:\\dev\\other\\z.md", + "\\rooted\\a": "G:\\rooted\\a", + "/tmp/x": "G:\\tmp\\x", # not a drive form: rooted on the cwd's drive + " G:/x/y.md ": "G:\\x\\y.md", + } + for src, want in cases.items(): + with self.subTest(src=src): + self.assertEqual(forks.norm_path(src, cwd=cwd, windows=True), want) + + def test_windows_cwd_in_gitbash_form(self): + self.assertEqual(forks.norm_path("a.md", cwd="/g/dev/proj", windows=True), "G:\\dev\\proj\\a.md") + + def test_posix_forms(self): + cwd = "/home/u/proj" + self.assertEqual(forks.norm_path("a/b.md", cwd=cwd, windows=False), "/home/u/proj/a/b.md") + self.assertEqual(forks.norm_path("/abs/x.md", cwd=cwd, windows=False), "/abs/x.md") + self.assertEqual(forks.norm_path("../z", cwd=cwd, windows=False), "/home/u/z") + + def test_native_relative(self): + self.assertEqual(os.path.normcase(forks.norm_path("x/y.md")), + os.path.normcase(os.path.abspath(os.path.join("x", "y.md")))) + + def test_empty_refused(self): + with self.assertRaises(forks.ForksError): + forks.norm_path(" ") + + @unittest.skipUnless(IS_WINDOWS, "Windows path forms") + def test_state_dir_windows_forms(self): + want = "G:\\w\\reports\\2026-10-16-08-x-run-forks" + for src in ("G:\\w\\reports\\2026-10-16-08-x-run.md", "G:/w/reports/2026-10-16-08-x-run.md", + "/g/w/reports/2026-10-16-08-x-run.md"): + with self.subTest(src=src): + self.assertEqual(forks.state_dir_for(src), want) + + def test_state_dir_relative(self): + got = forks.state_dir_for(os.path.join("reports", "2026-10-16-08-x-run.md")) + self.assertEqual(os.path.normcase(got), + os.path.normcase(os.path.abspath(os.path.join("reports", "2026-10-16-08-x-run-forks")))) + + def test_state_dir_needs_md(self): + with self.assertRaises(forks.ForksError) as cm: + forks.state_dir_for(os.path.join(self.tmp, "run.txt")) + self.assertEqual(cm.exception.code, 2) + + +# -------------------------------------------------------------------------------------- +class ScrubTests(Base): + def test_scrubbed_env(self): + venv = os.path.join(self.tmp, "venv") + keep_dir = os.path.join(self.tmp, "bin") + host = { + "CLAUDE_CODE_MESSAGING_SOCKET": "s", "CLAUDE_CODE_SESSION_ID": "x", "CLAUDE_EFFORT": "high", + "ANTHROPIC_API_KEY": "k", "ANTHROPIC_DEFAULT_OPUS_MODEL": "m", "UV_CACHE_DIR": "c", + "VIRTUAL_ENV": venv, "CLAUDECODE": "1", "CLAUDE_CODE_CHILD_SESSION": "1", "AI_AGENT": "a", + "CLAUDE_PID": "1", "ANTHROPIC_BASE_URL": "u", "HOME": "h", + "CLAUDE_CODE_PROMPT_CACHE_TTL": "5m", "CLAUDE_CODE_EXECPATH": "old", + "PATH": os.pathsep.join([os.path.join(venv, "Scripts"), keep_dir]), + } + env = forks.scrubbed_env("BIN", environ=host) + for k in ("CLAUDE_CODE_MESSAGING_SOCKET", "CLAUDE_CODE_SESSION_ID", "CLAUDE_EFFORT", + "ANTHROPIC_API_KEY", "ANTHROPIC_DEFAULT_OPUS_MODEL", "UV_CACHE_DIR", "VIRTUAL_ENV"): + self.assertNotIn(k, env) + for k in ("CLAUDECODE", "CLAUDE_CODE_CHILD_SESSION", "AI_AGENT", "CLAUDE_PID", + "ANTHROPIC_BASE_URL", "HOME"): + self.assertEqual(env[k], host[k]) + self.assertEqual(env["CLAUDE_CODE_PROMPT_CACHE_TTL"], "1h") + self.assertEqual(env["DISABLE_AUTOUPDATER"], "1") + self.assertEqual(env["SPIRE_FORKS_CHILD"], "1") + self.assertEqual(env["CLAUDE_CODE_EXECPATH"], "BIN") + self.assertEqual(env["PATH"], keep_dir) + + def test_unknown(self): + host = {"CLAUDECODE": "1", "CLAUDE_EFFORT": "x", "CLAUDE_CODE_NEW_THING": "1", + "ANTHROPIC_BASE_URL": "u", "OTHER": "1", "CLAUDE_CODE_EXECPATH": "p"} + self.assertEqual(forks.scrub_unknown(host), ["ANTHROPIC_BASE_URL", "CLAUDE_CODE_NEW_THING"]) + + +# -------------------------------------------------------------------------------------- +class LockTests(Base): + def setUp(self): + super().setUp() + self.state = os.path.join(self.tmp, "st") + os.makedirs(self.state) + + def test_contention_times_out(self): + with forks.RegistryLock(self.state): + t0 = time.monotonic() + with self.assertRaises(forks.ForksError) as cm: + forks.RegistryLock(self.state, timeout=0.3).acquire() + self.assertGreaterEqual(time.monotonic() - t0, 0.25) + self.assertEqual(cm.exception.payload["error"], "registry locked") + with forks.RegistryLock(self.state, timeout=0.3): + pass + self.assertFalse(os.path.exists(forks.lock_path(self.state))) + + def test_waiter_gets_lock_after_release(self): + l1 = forks.RegistryLock(self.state).acquire() + threading.Timer(0.2, l1.release).start() + t0 = time.monotonic() + with forks.RegistryLock(self.state, timeout=5) as l2: + self.assertGreaterEqual(time.monotonic() - t0, 0.15) + self.assertFalse(l2.broke_stale) + + def test_contention_across_processes(self): + code = ("import importlib.util,sys;" + "s=importlib.util.spec_from_file_location('forks',sys.argv[1]);m=importlib.util.module_from_spec(s);" + "s.loader.exec_module(m)\n" + "try:\n m.RegistryLock(sys.argv[2],timeout=0.3).acquire(); sys.exit(0)\n" + "except m.ForksError: sys.exit(7)\n") + with forks.RegistryLock(self.state): + r = subprocess.run([sys.executable, "-c", code, FORKS_PY, self.state], capture_output=True) + self.assertEqual(r.returncode, 7, r.stderr) + + def test_stale_lock_broken(self): + with open(forks.lock_path(self.state), "w") as f: + json.dump({"pid": 1, "at": time.time() - (forks.LOCK_STALE_S + 1), "token": "old"}, f) + with forks.RegistryLock(self.state, timeout=2) as lk: + self.assertTrue(lk.broke_stale) + with open(forks.lock_path(self.state)) as f: + self.assertEqual(json.load(f)["token"], lk.token) + + def test_garbage_stale_lock_broken_by_mtime(self): + p = forks.lock_path(self.state) + with open(p, "w") as f: + f.write("") + old = time.time() - (forks.LOCK_STALE_S + 5) + os.utime(p, (old, old)) + with forks.RegistryLock(self.state, timeout=2) as lk: + self.assertTrue(lk.broke_stale) + + def test_fresh_lock_not_broken(self): + with open(forks.lock_path(self.state), "w") as f: + json.dump({"pid": 1, "at": time.time() - 5, "token": "other"}, f) + with self.assertRaises(forks.ForksError): + forks.RegistryLock(self.state, timeout=0.3).acquire() + + def test_stale_under_injected_clock_and_safe_release(self): + l1 = forks.RegistryLock(self.state).acquire() + real = time.time() + with mock.patch.object(forks, "_now", lambda: real + forks.LOCK_STALE_S + 1): + l2 = forks.RegistryLock(self.state, timeout=2).acquire() + self.assertTrue(l2.broke_stale) + l1.release() # must not remove l2's lock + self.assertTrue(os.path.exists(forks.lock_path(self.state))) + l2.release() + self.assertFalse(os.path.exists(forks.lock_path(self.state))) + + +# -------------------------------------------------------------------------------------- +class AtomicTests(Base): + def test_replace_while_reader_holds_target(self): + target = os.path.join(self.tmp, "registry.json") + forks.write_json_atomic(target, {"v": 1}) + opened, done = threading.Event(), threading.Event() + + def reader(): + with open(target, "rb") as f: + opened.set() + time.sleep(0.4) + f.read() + done.set() + th = threading.Thread(target=reader) + th.start() + opened.wait(2) + attempts = forks.write_json_atomic(target, {"v": 2}) + th.join() + self.assertEqual(forks.read_json(target), {"v": 2}) + if IS_WINDOWS: + self.assertGreater(attempts, 1, "replace should have hit a sharing violation and retried") + self.assertEqual([n for n in os.listdir(self.tmp) if n.endswith(".tmp")], []) + + def test_read_json_missing_is_none(self): + self.assertIsNone(forks.read_json(os.path.join(self.tmp, "nope.json"))) + + def test_registry_roundtrip_and_schema(self): + st = os.path.join(self.tmp, "st") + forks.ensure_state(st) + with forks.RegistryLock(st): + forks.write_registry(st, forks.new_registry()) + reg = forks.read_registry(st) + self.assertEqual(reg["schema"], 1) + with open(forks.registry_path(st), "rb") as f: + self.assertNotIn(b"\r\n", f.read()) + forks.write_json_atomic(forks.registry_path(st), {"schema": 99}) + with self.assertRaises(forks.ForksError): + forks.read_registry(st) + + def test_ledger_append(self): + st = os.path.join(self.tmp, "st") + forks.ensure_state(st) + with forks.RegistryLock(st): + forks.ledger_append(st, {"job": "j0001"}) + forks.ledger_append(st, {"job": "j0002"}) + with open(forks.ledger_path(st), "rb") as f: + lines = f.read().decode().splitlines() + self.assertEqual([json.loads(x)["job"] for x in lines], ["j0001", "j0002"]) + + +# -------------------------------------------------------------------------------------- +class PidTests(Base): + def test_own_pid(self): + rec = forks.pid_record(os.getpid()) + self.assertTrue(forks.record_alive(rec)) + if rec["ctime"] is not None: + self.assertFalse(forks.pid_alive(os.getpid(), rec["ctime"] + 1)) + + def test_dead_pid(self): + p = subprocess.Popen([sys.executable, "-c", "pass"]) + rec = forks.pid_record(p.pid) + p.wait() + self.assertFalse(forks.record_alive(rec)) + self.assertFalse(forks.pid_alive(0)) + + def test_supervisor_interpreter(self): + self.assertTrue(os.path.isfile(forks.supervisor_interpreter())) + + +# -------------------------------------------------------------------------------------- +class InitTests(Base): + def setUp(self): + super().setUp() + self.cwd = os.path.join(self.tmp, "proj") + os.makedirs(os.path.join(self.cwd, ".claude", "agents")) + os.makedirs(os.path.join(self.tmp, "reports")) + self.run_log = os.path.join(self.tmp, "reports", "2026-10-16-08-t-run.md") + self.state = forks.state_dir_for(self.run_log) + self.user_settings = os.path.join(self.tmp, "home", ".claude", "settings.json") + p = mock.patch.object(forks, "USER_SETTINGS_PATH", self.user_settings) + p.start() + self.addCleanup(p.stop) + self.stub = make_stub_binary(self.tmp) + + def init(self, *extra, model="claude-opus-5-5[1m]"): + argv = ["init", "--run", self.run_log, "--model", model, "--cwd", self.cwd, + "--binary", self.stub, *extra] + return call(argv) + + def write_settings(self, path, models): + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w") as f: + json.dump({"availableModels": models}, f) + + def add_fake_base(self): + with forks.RegistryLock(self.state): + reg = forks.read_registry(self.state) + reg["bases"]["B1"] = {"status": "active", "session": "00000000-0000-0000-0000-000000000000"} + forks.write_registry(self.state, reg) + + def reg_bytes(self): + with open(forks.registry_path(self.state), "rb") as f: + return f.read() + + # --- basic --- + def test_init_creates_state_and_config(self): + code, out, raw = self.init() + self.assertEqual(code, 0, out) + self.assert_one_json_line(raw) + self.assertTrue(out["created"]) + st = self.state + for d in forks.STATE_SUBDIRS: + self.assertTrue(os.path.isdir(os.path.join(st, d)), d) + with open(os.path.join(st, ".gitignore"), "rb") as f: + self.assertEqual(f.read(), b"raw/\n*.lock\n") + self.assertEqual(forks.read_json(forks.mcp_empty_path(st)), {"mcpServers": {}}) + self.assertFalse(os.path.exists(forks.lock_path(st))) + reg = forks.read_registry(st) + self.assertEqual(reg["schema"], 1) + self.assertEqual(reg["bases"], {}) + self.assertEqual(reg["jobs"], {}) + cfg = reg["config"] + la = cfg["launch"] + self.assertEqual(la["model"], "claude-opus-5-5[1m]") + self.assertEqual(la["effort"], "high") + self.assertEqual(la["permission_mode"], "dontAsk") + self.assertEqual(la["tools"], forks.DEFAULT_TOOLS.split(",")) + self.assertEqual(la["allowed_tools"], la["tools"]) + self.assertEqual(la["disallowed_tools"], forks.DEFAULT_DISALLOWED) + self.assertEqual(la["mcp"], {"strict": True, "config": forks.mcp_empty_path(st)}) + self.assertEqual(la["cwd"], forks.norm_path(self.cwd)) + self.assertIsNone(la["agent"]) + self.assertEqual(la["env"]["pinned"], {"CLAUDE_CODE_PROMPT_CACHE_TTL": "1h", "DISABLE_AUTOUPDATER": "1"}) + b = la["binary"] + self.assertEqual(b["path"], forks.norm_path(self.stub)) + self.assertTrue(os.path.isabs(b["path"])) + self.assertEqual(b["version"], STUB_VERSION) + self.assertEqual(b["size"], os.path.getsize(self.stub)) + pre = la["append_system_prompt"] + self.assertEqual(pre["path"], forks.preamble_path()) + self.assertEqual(pre["sha256"], forks._sha256_file(forks.preamble_path())) + self.assertEqual(cfg["fingerprint"], forks.fingerprint_hash(la)) + self.assertEqual(len(cfg["fingerprint"]), 64) + self.assertEqual(cfg["parallel_cap"], 2) + self.assertEqual(cfg["orphan_after"], 1200) + self.assertEqual(cfg["pings"], "on") + self.assertEqual(cfg["ttl_s"], 3600) + self.assertEqual(cfg["max_minutes"], 90) + self.assertEqual(cfg["stall_after_s"], 900) + self.assertEqual(cfg["thresholds"], forks.THRESHOLD_DEFAULTS) + self.assertEqual(out["config"], cfg) + + def test_fingerprint_canonical(self): + a = {"b": [1, 2], "a": {"y": 1, "x": None}} + b = {"a": {"x": None, "y": 1}, "b": [1, 2]} + self.assertEqual(forks.fingerprint_hash(a), forks.fingerprint_hash(b)) + self.assertEqual(forks.canonical_json(a), '{"a":{"x":null,"y":1},"b":[1,2]}') + + def test_list_flags(self): + code, out, _ = self.init("--tools", "Read,Bash", "--allowed-tools", "Read", "Bash(git status:*)", + "--disallowed-tools", "Bash(rm:*),Bash(a,b)") + self.assertEqual(code, 0, out) + la = out["config"]["launch"] + self.assertEqual(la["tools"], ["Read", "Bash"]) + self.assertEqual(la["allowed_tools"], ["Read", "Bash(git status:*)"]) + self.assertEqual(la["disallowed_tools"], ["Bash(rm:*)", "Bash(a,b)"]) + + # --- idempotency --- + def test_idempotent(self): + self.assertEqual(self.init()[0], 0) + before = self.reg_bytes() + code, out, _ = self.init() + self.assertEqual(code, 0, out) + self.assertFalse(out["created"]) + self.assertFalse(out["changed"]) + self.assertEqual(self.reg_bytes(), before) + + def test_changed_values_before_base(self): + _, first, _ = self.init() + code, out, _ = self.init("--effort", "medium", "--orphan-after", "1500") + self.assertEqual(code, 0, out) + self.assertTrue(out["changed"]) + self.assertIn("launch.effort", out["changed_fields"]) + self.assertIn("orphan_after", out["changed_fields"]) + self.assertIn("fingerprint", out["changed_fields"]) + cfg = forks.read_registry(self.state)["config"] + self.assertEqual(cfg["launch"]["effort"], "medium") + self.assertEqual(cfg["orphan_after"], 1500) + self.assertNotEqual(cfg["fingerprint"], first["config"]["fingerprint"]) + + def test_changed_values_after_base_refused(self): + self.assertEqual(self.init()[0], 0) + self.add_fake_base() + before = self.reg_bytes() + code, out, raw = self.init("--effort", "medium") + self.assertEqual(code, 2) + self.assert_one_json_line(raw) + self.assertIn("base exists", out["error"]) + self.assertEqual(out["bases"], ["B1"]) + self.assertIn("launch.effort", out["fields"]) + self.assertEqual(self.reg_bytes(), before) + self.assertFalse(os.path.exists(forks.lock_path(self.state))) + # identical values stay fine with a base + code, out, _ = self.init() + self.assertEqual(code, 0, out) + self.assertFalse(out["changed"]) + + # --- refusals --- + def assert_refused_no_state(self, *extra, model="claude-opus-5-5[1m]"): + code, out, raw = self.init(*extra, model=model) + self.assertEqual(code, 2, out) + self.assert_one_json_line(raw) + self.assertIn("error", out) + self.assertFalse(os.path.exists(self.state), "a refused init must not write") + return out + + def test_bad_model_ids(self): + for m in ("opus", "sonnet", "claude-opus-5-5[200k]", "Claude-Opus-5-5", "claude-opus 5"): + with self.subTest(model=m): + self.assert_refused_no_state(model=m) + + def test_plan_mode_refused(self): + out = self.assert_refused_no_state("--permission-mode", "plan") + self.assertEqual(out["error"], "permission mode refused") + + def test_unknown_mode_and_effort_refused(self): + self.assert_refused_no_state("--permission-mode", "yolo") + self.assert_refused_no_state("--effort", "turbo") + + def test_missing_binary_refused(self): + out = self.assert_refused_no_state("--binary", os.path.join(self.tmp, "nope.exe")) + self.assertEqual(out["error"], "claude binary not found") + + def test_agent(self): + self.assert_refused_no_state("--agent", "ghost") + self.assert_refused_no_state("--agent", "../x") + ap = os.path.join(self.cwd, ".claude", "agents", "impl.md") + with open(ap, "w") as f: + f.write("---\nname: impl\n---\nbody\n") + code, out, _ = self.init("--agent", "impl") + self.assertEqual(code, 0, out) + ag = out["config"]["launch"]["agent"] + self.assertEqual(ag["name"], "impl") + self.assertEqual(ag["sha256"], forks._sha256_file(ap)) + + # --- C8 admission --- + def proj(self): + return os.path.join(self.cwd, ".claude", "settings.json") + + def local(self): + return os.path.join(self.cwd, ".claude", "settings.local.json") + + def test_admission_no_list_admits(self): + self.assertTrue(forks.model_admission("claude-whatever-9", self.cwd)["admitted"]) + + def test_admission_exact(self): + self.write_settings(self.user_settings, ["claude-opus-5-5[1m]"]) + self.assertTrue(forks.model_admission("claude-opus-5-5[1m]", self.cwd)["admitted"]) + # the bare id is not the [1m] entry and `opus` is not listed + self.assertFalse(forks.model_admission("claude-opus-5-5", self.cwd)["admitted"]) + + def test_admission_strips_1m(self): + self.write_settings(self.proj(), ["claude-opus-5-5"]) + adm = forks.model_admission("claude-opus-5-5[1m]", self.cwd) + self.assertTrue(adm["admitted"]) + self.assertIn("without [1m]", adm["via"]) + + def test_admission_family_alias(self): + self.write_settings(self.local(), ["sonnet"]) + adm = forks.model_admission("claude-sonnet-5-5", self.cwd) + self.assertTrue(adm["admitted"]) + self.assertIn("family alias", adm["via"]) + + def test_admission_any_list_admits(self): + self.write_settings(self.user_settings, ["claude-opus-5-5"]) + self.write_settings(self.local(), ["haiku"]) + self.assertTrue(forks.model_admission("claude-haiku-5-5", self.cwd)["admitted"]) + + def test_admission_unlisted_refused_by_init(self): + self.write_settings(self.user_settings, ["claude-opus-5-5"]) + self.write_settings(self.local(), ["opus"]) + with open(self.local(), "rb") as f: + local_before = f.read() + out = self.assert_refused_no_state(model="claude-sonnet-5-5") + self.assertEqual(out["error"], "model not admitted by availableModels") + self.assertEqual(out["model"], "claude-sonnet-5-5") + self.assertEqual(sorted(out["files"]), sorted([self.user_settings, self.local()])) + self.assertIn("must add", out["message"]) + self.assertIn(self.local(), out["message"]) + with open(self.local(), "rb") as f: + self.assertEqual(f.read(), local_before, "settings must never be written") + self.assertFalse(os.path.exists(self.proj())) + + def test_admission_admitted_by_init(self): + self.write_settings(self.proj(), ["opus", "sonnet"]) + code, out, _ = self.init(model="claude-sonnet-5-5") + self.assertEqual(code, 0, out) + self.assertIn("family alias", out["config"]["model_admission"]) + + +# -------------------------------------------------------------------------------------- +class CliTests(Base): + def test_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(): + 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.assertFalse(os.path.exists(run_log[:-3] + "-forks")) + + def test_unknown_command_and_missing_run(self): + for argv in (["bogus"], [], ["list"], ["init", "--run", "x.md"], + ["list", "--run", "x.md", "--bogus"]): + with self.subTest(argv=argv): + code, out, raw = call(argv) + self.assertEqual(code, 2) + self.assert_one_json_line(raw) + self.assertEqual(out["error"], "usage") + + def test_help_is_json(self): + code, out, raw = call(["--help"]) + self.assertEqual(code, 0) + self.assert_one_json_line(raw) + self.assertIn("where", out["commands"]) + self.assertNotIn("_supervise", out["commands"]) + code, out, _ = call(["init", "--help"]) + self.assertEqual(code, 0) + self.assertIn("--model", out["help"]) + + def test_child_refuses_launching(self): + os.environ[forks.CHILD_MARKER] = "1" + run_log = os.path.join(self.tmp, "r-run.md") + code, out, _ = call(["fork", "--run", run_log, "--base", "B1", "--block", "1", "--brief", "b"]) + 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 + stub = make_stub_binary(self.tmp) + self.assertEqual(call(["where", "--binary", stub])[0], 0) + + def test_internal_error_keeps_invariant(self): + def boom(args): + raise RuntimeError("kaboom") + with mock.patch.dict(forks.HANDLERS, {"list": boom}): + code, out, raw = call(["list", "--run", os.path.join(self.tmp, "r-run.md")]) + self.assertEqual(code, 2) + self.assert_one_json_line(raw) + self.assertEqual(out["error"], "internal error") + + def test_where_in_process_missing_binary(self): + code, out, raw = call(["where", "--binary", os.path.join(self.tmp, "nope.exe")]) + self.assertEqual(code, 2) + self.assert_one_json_line(raw) + self.assertFalse(out["ok"]) + self.assertIn("reason", out) + self.assertIn("error", out) + + +# -------------------------------------------------------------------------------------- +@unittest.skipUnless(shutil.which("uv"), "uv not on PATH") +class SubprocessTests(Base): + def run_uv(self, *argv): + return subprocess.run(["uv", "run", FORKS_PY, *argv], cwd=self.tmp, env=child_env(), + capture_output=True, timeout=180) + + def test_where_via_uv_stub(self): + stub = make_stub_binary(self.tmp) + r = self.run_uv("where", "--binary", stub) + self.assertEqual(r.returncode, 0, r.stderr) + out = self.assert_one_json_line(r.stdout.decode("utf-8")) + self.assertTrue(out["ok"]) + self.assertEqual(out["spawn"], "cim-hidden" if IS_WINDOWS else "setsid") + self.assertEqual(out["binary"]["version"], STUB_VERSION) + self.assertEqual(out["binary"]["how"], "--binary") + self.assertEqual(os.path.normcase(out["script"]), os.path.normcase(os.path.abspath(FORKS_PY))) + for k in ("python", "uv", "git", "scrub"): + self.assertIn(k, out) + for k in ("drop", "keep", "pinned", "unknown"): + self.assertIn(k, out["scrub"]) + self.assertIn("base_executable", out["python"]) + self.assertIsNotNone(out["git"]["path"]) + + def test_where_via_uv_host_binary(self): + if not (os.environ.get("CLAUDE_CODE_EXECPATH") or shutil.which("claude")): + self.skipTest("no claude binary on this host") + r = self.run_uv("where") + self.assertEqual(r.returncode, 0, r.stderr) + self.assertTrue(self.assert_one_json_line(r.stdout.decode("utf-8"))["ok"]) + + def test_error_case_stdout_invariant(self): + run_log = os.path.join(self.tmp, "x-run.md") + r = self.run_uv("init", "--run", run_log, "--model", "opus", "--cwd", self.tmp) + self.assertEqual(r.returncode, 2, r.stderr) + out = self.assert_one_json_line(r.stdout.decode("utf-8")) + self.assertIn("error", out) + self.assertFalse(os.path.exists(os.path.join(self.tmp, "x-run-forks"))) + + def test_unknown_command_exit_2(self): + r = self.run_uv("frobnicate") + self.assertEqual(r.returncode, 2, r.stderr) + self.assertEqual(self.assert_one_json_line(r.stdout.decode("utf-8"))["error"], "usage") + + +if __name__ == "__main__": + unittest.main(verbosity=2)