dirigence/reliquary
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
4515dbbebb59d382abb950552909899b300f8f9e
reliquary/plugins/spire/skills/forks/tests/test_forks.py
T
g4borg 7b9b7488de 🐛 forks: Phase 1 review fixes (I1-I3, M1-M5, M9) 
- I1: model_refused only on the restriction warning or a real substitution;
  API errors get usage_limit / overloaded / prompt_too_long / api_error
- I2: a dead root claims only children created during the job's lifetime
- I3: continue no longer advances the base's last_hit
- M1: registry status before the ledger line; ledger dedupes by job
- M2: stale lock = dead holder or > 300 s; atomic rename-based break;
  deadline and sleep on a failed break
- M3: supervisor cuts clock jumps out of max_minutes, re-arms the orphan
  watchdog, flags clock_jump
- M4: kill/expire/init refused in a fork; a fork never stamps the orchestrator
- M5: --permission-mode choices from claude 2.1.299
- M9: ANTHROPIC_BASE_URL kept explicitly; where warns on a non-default value
2026-10-16 16:26:59 +02:00

1977 lines
96 KiB
Python
Raw Blame History

# /// 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"}
# M9: ANTHROPIC_BASE_URL is on the keep list now, so it is no longer "unknown"
self.assertEqual(forks.scrub_unknown(host), ["CLAUDE_CODE_NEW_THING"])
self.assertEqual(forks.scrub_class("ANTHROPIC_BASE_URL"), "keep")
# --------------------------------------------------------------------------------------
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 dead_holder(self) -> dict:
p = subprocess.Popen([sys.executable, "-c", "pass"])
rec = forks.pid_record(p.pid)
p.wait()
return rec
def write_lock(self, holder: dict, at: float, token: str):
with open(forks.lock_path(self.state), "w") as f:
json.dump({"pid": holder["pid"], "ctime": holder["ctime"], "at": at, "token": token}, f)
def test_dead_holder_lock_broken_at_once(self):
# M2: a lock whose holder (pid + creation time) is dead is stale even when it is fresh
self.write_lock(self.dead_holder(), time.time(), "old")
t0 = time.monotonic()
with forks.RegistryLock(self.state, timeout=2) as lk:
self.assertTrue(lk.broke_stale)
self.assertLess(time.monotonic() - t0, 1.5)
with open(forks.lock_path(self.state)) as f:
self.assertEqual(json.load(f)["token"], lk.token)
self.assertEqual([n for n in os.listdir(self.state) if n.endswith(".lock")], [])
def test_abandoned_lock_broken_by_age(self):
me = forks.pid_record(os.getpid()) # a live holder: only the age rule applies
self.write_lock(me, time.time() - (forks.LOCK_ABANDONED_S + 1), "old")
with forks.RegistryLock(self.state, timeout=2) as lk:
self.assertTrue(lk.broke_stale)
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_ABANDONED_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):
# a live holder's lock is never broken before LOCK_ABANDONED_S (garbage content neither)
self.write_lock(forks.pid_record(os.getpid()), time.time() - 60, "other")
with self.assertRaises(forks.ForksError):
forks.RegistryLock(self.state, timeout=0.3).acquire()
with open(forks.lock_path(self.state), "w") as f:
f.write("")
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_ABANDONED_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)))
def test_unbreakable_stale_lock_respects_deadline(self):
"""M2: a stale lock that cannot be removed or renamed (AV, sync client) must not spin:
the call ends with its "registry locked" error once the deadline passes."""
self.write_lock(self.dead_holder(), time.time() - 1000, "old") # stale by holder and by age
lp = os.path.normcase(forks.lock_path(self.state))
real_remove, real_rename = os.remove, os.rename
def no_remove(p, *a, **k):
if os.path.normcase(str(p)) == lp:
raise PermissionError(13, "held by another program", p)
return real_remove(p, *a, **k)
def no_rename(src, dst, *a, **k):
if os.path.normcase(str(src)) == lp:
raise PermissionError(13, "held by another program", src)
return real_rename(src, dst, *a, **k)
box = {}
def attempt():
t0 = time.monotonic()
try:
forks.RegistryLock(self.state, timeout=0.5).acquire()
box["result"] = "acquired"
except forks.ForksError as e:
box["result"] = e.payload["error"]
box["took"] = time.monotonic() - t0
with mock.patch.object(os, "remove", no_remove), mock.patch.object(os, "rename", no_rename):
th = threading.Thread(target=attempt, daemon=True)
th.start()
th.join(5)
alive = th.is_alive()
self.assertFalse(alive, "acquire spins on a lock it cannot break (hot loop, no deadline)")
self.assertEqual(box["result"], "registry locked")
self.assertLess(box["took"], 2.0)
def test_fresh_lock_taken_during_break_survives(self):
"""M2: between a waiter's judgement (stale) and its break, another waiter breaks the stale
lock and takes a fresh one. The first waiter must not delete the fresh lock."""
self.write_lock(self.dead_holder(), time.time(), "stale")
lp = os.path.normcase(forks.lock_path(self.state))
fresh = {"pid": os.getpid(), "ctime": forks.proc_ctime(os.getpid()), "at": time.time(),
"token": "fresh"}
real_remove, real_rename = os.remove, os.rename
swapped = []
def swap_in_fresh():
if not swapped:
swapped.append(1)
with open(forks.lock_path(self.state), "w") as f:
json.dump(dict(fresh, at=time.time()), f)
def hook_remove(p, *a, **k):
if os.path.normcase(str(p)) == lp:
swap_in_fresh()
return real_remove(p, *a, **k)
def hook_rename(src, dst, *a, **k):
if os.path.normcase(str(src)) == lp:
swap_in_fresh()
return real_rename(src, dst, *a, **k)
with mock.patch.object(os, "remove", hook_remove), mock.patch.object(os, "rename", hook_rename):
with self.assertRaises(forks.ForksError):
forks.RegistryLock(self.state, timeout=0.5).acquire()
self.assertEqual(swapped, [1], "the break was never attempted")
with open(forks.lock_path(self.state)) as f:
self.assertEqual(json.load(f)["token"], "fresh")
self.assertEqual(sorted(n for n in os.listdir(self.state) if n.endswith(".lock")),
["registry.lock"])
# --------------------------------------------------------------------------------------
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"])
def test_ledger_append_dedupes_by_job(self):
"""M1: a second finaliser (kill racing the supervisor) never writes a second line."""
st = os.path.join(self.tmp, "st")
forks.ensure_state(st)
with forks.RegistryLock(st):
self.assertTrue(forks.ledger_append(st, {"job": "j0001", "status": "ok"}))
self.assertFalse(forks.ledger_append(st, {"job": "j0001", "status": "killed"}))
self.assertTrue(forks.ledger_append(st, {"job": "j0002", "status": "ok"}))
# a torn last line (writer killed mid-append) does not swallow the next line
with open(forks.ledger_path(st), "ab") as f:
f.write(b'{"job": "j0003", "sta')
with forks.RegistryLock(st):
self.assertTrue(forks.ledger_append(st, {"job": "j0004", "status": "ok"}))
self.assertFalse(forks.ledger_append(st, {"job": "j0002", "status": "killed"}))
with open(forks.ledger_path(st), "rb") as f:
lines = f.read().decode().splitlines()
parsed = []
for x in lines:
with contextlib.suppress(ValueError):
parsed.append(json.loads(x))
self.assertEqual([(p["job"], p["status"]) for p in parsed],
[("j0001", "ok"), ("j0002", "ok"), ("j0004", "ok")])
# --------------------------------------------------------------------------------------
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 ProcessTreeTests(Base):
"""I2: walking a dead root's tree on a fake process table (nothing real is touched or killed)."""
ROOT, A, A_KID, B, C, C_KID, OTHER = 900001, 900011, 900012, 900021, 900031, 900032, 900041
def setUp(self):
super().setUp()
self.t0 = time.time() - 5000 # the job (its root) started here
self.hb = self.t0 + 500 # last heartbeat: the job was known alive until here
e2c = forks.epoch_to_ctime
if e2c(self.t0) is None:
self.skipTest("no epoch -> creation-time conversion on this host")
self.root_ct = e2c(self.t0)
self.ctimes = {
self.A: e2c(self.t0 + 600), # (a) reused pid's child: created after the last heartbeat
self.A_KID: e2c(self.t0 + 601),
self.B: e2c(self.t0 - 100), # (b) created before the job started
self.C: e2c(self.t0 + 100), # (c) the job's own child, created while it ran
self.C_KID: e2c(self.t0 + 700), # its child (parent verified alive): ours
self.OTHER: e2c(self.t0 + 50),
}
self.table = [(self.A, self.ROOT, "a.exe"), (self.A_KID, self.A, "a-kid.exe"),
(self.B, self.ROOT, "dwm.exe"), (self.C, self.ROOT, "bash.exe"),
(self.C_KID, self.C, "sleep.exe"), (self.OTHER, 4, "other.exe")]
self.live = set(self.ctimes)
for name, fn in (("process_table", lambda: list(self.table)),
("proc_ctime", lambda pid: self.ctimes.get(int(pid)) if int(pid) in self.live else None),
("pid_alive", self.fake_alive)):
p = mock.patch.object(forks, name, fn)
p.start()
self.addCleanup(p.stop)
def fake_alive(self, pid, ctime=None):
pid = int(pid)
return pid in self.live and (ctime is None or self.ctimes.get(pid) == ctime)
def rec(self):
return {"pid": self.ROOT, "ctime": self.root_ct}
def test_dead_root_claims_only_children_from_its_lifetime(self):
got = {p["pid"] for p in forks.process_tree(self.rec(), alive_until=self.hb)}
self.assertEqual(got, {self.C, self.C_KID})
def test_dead_root_without_bounds_claims_nothing(self):
self.assertEqual(forks.process_tree(self.rec()), [])
self.assertEqual(forks.process_tree({"pid": self.ROOT, "ctime": None}, alive_until=self.hb), [])
def test_live_root_unchanged(self):
self.live.add(self.ROOT)
self.ctimes[self.ROOT] = self.root_ct
got = {p["pid"] for p in forks.process_tree(self.rec(), alive_until=None)}
self.assertEqual(got, {self.ROOT, self.A, self.A_KID, self.C, self.C_KID}) # B is older
def test_reap_lost_bounds_by_heartbeat(self):
st = os.path.join(self.tmp, "st")
forks.ensure_state(st)
old = time.time() - 1000
reg = forks.new_registry()
reg["config"] = {"launch": {}, "ttl_s": 3600}
dead_sup = {"pid": 900091, "ctime": 1}
reg["jobs"]["j0001"] = forks._new_entry("j0001", "fork", status="running", supervisor=dead_sup,
launched_at=forks._iso(old - 10))
with forks.RegistryLock(st):
forks.write_registry(st, reg)
forks.write_json_atomic(forks.job_file(st, "j0001"), {
"supervisor": dead_sup, "claude": self.rec(), "started_at": forks._iso(old - 10),
"heartbeat_at": forks._iso(old)})
seen = []
with mock.patch.object(forks, "kill_tree", lambda rec, extra=(), alive_until=None:
seen.append((rec, alive_until)) or {"snapshot": [], "survivors": [],
"taskkill_rc": None}):
self.assertEqual(forks.reap_lost(st), ["j0001"])
self.assertEqual(seen, [(self.rec(), forks._ts(forks._iso(old)))])
# --------------------------------------------------------------------------------------
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_permission_modes_match_2_1_299(self):
"""M5: `claude --help` 2.1.299 offers acceptEdits, auto, bypassPermissions, manual, dontAsk,
plan. `default` is not one of them; `plan` is refused."""
out = self.assert_refused_no_state("--permission-mode", "default")
self.assertEqual(out["error"], "unknown permission mode")
self.assertEqual(sorted(out["allowed"]),
sorted(["acceptEdits", "auto", "bypassPermissions", "manual", "dontAsk"]))
for mode in ("acceptEdits", "auto", "bypassPermissions", "manual", "dontAsk"):
with self.subTest(mode=mode):
code, out, _ = self.init("--permission-mode", mode)
self.assertEqual(code, 0, out)
self.assertEqual(out["config"]["launch"]["permission_mode"], mode)
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_phase2_not_implemented(self):
run_log = os.path.join(self.tmp, "r-run.md")
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")
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):
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.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)
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_warns_on_base_url(self):
"""M9: `where` warns (exit 0) when ANTHROPIC_BASE_URL is set to anything but the default."""
stub = make_stub_binary(self.tmp)
os.environ.pop("ANTHROPIC_BASE_URL", None)
for value in (None, "https://api.anthropic.com", "https://api.anthropic.com/"):
with self.subTest(value=value):
if value is not None:
os.environ["ANTHROPIC_BASE_URL"] = value
code, out, raw = call(["where", "--binary", stub])
self.assertEqual(code, 0, out)
self.assert_one_json_line(raw)
self.assertEqual(out["warnings"], [])
os.environ["ANTHROPIC_BASE_URL"] = "https://llm-gateway.example.internal/anthropic"
code, out, raw = call(["where", "--binary", stub])
self.assertEqual(code, 0, out)
self.assert_one_json_line(raw)
self.assertTrue(out["ok"])
self.assertEqual(len(out["warnings"]), 1, out["warnings"])
self.assertIn("ANTHROPIC_BASE_URL", out["warnings"][0])
self.assertIn("llm-gateway.example.internal", out["warnings"][0])
def test_child_refuses_kill_expire_init(self):
"""M4: a fork may not run kill, expire or init."""
os.environ[forks.CHILD_MARKER] = "1"
stub = make_stub_binary(self.tmp)
run_log = os.path.join(self.tmp, "r-run.md")
for argv in (["kill", "--run", run_log, "--job", "j0001"],
["expire", "--run", run_log, "--all"],
["init", "--run", run_log, "--model", "claude-opus-5-5", "--cwd", self.tmp,
"--binary", stub]):
with self.subTest(cmd=argv[0]):
code, out, raw = call(argv)
self.assertEqual(code, 2, out)
self.assert_one_json_line(raw)
self.assertIn("inside a fork", out["error"])
self.assertEqual(out["command"], argv[0])
self.assertFalse(os.path.exists(run_log[:-3] + "-forks"))
def test_child_does_not_stamp_orchestrator(self):
"""M4: list and wait from inside a fork leave run.last_orchestrator_call_at alone."""
os.makedirs(os.path.join(self.tmp, "reports"))
run_log = os.path.join(self.tmp, "reports", "r-run.md")
state = forks.state_dir_for(run_log)
forks.ensure_state(state)
old = forks._iso(time.time() - 5000)
reg = forks.new_registry()
reg["config"] = {"launch": {}, "ttl_s": 3600}
reg["run"]["last_orchestrator_call_at"] = old
with forks.RegistryLock(state):
forks.write_registry(state, reg)
os.environ[forks.CHILD_MARKER] = "1"
self.assertEqual(call(["list", "--run", run_log])[0], 0)
self.assertEqual(call(["wait", "--run", run_log, "--all", "--timeout", "1"])[0], 0)
self.assertEqual(forks.read_registry(state)["run"]["last_orchestrator_call_at"], old)
os.environ.pop(forks.CHILD_MARKER)
self.assertEqual(call(["list", "--run", run_log])[0], 0) # the orchestrator does stamp
self.assertNotEqual(forks.read_registry(state)["run"]["last_orchestrator_call_at"], old)
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")
# --------------------------------------------------------------------------------------
# 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"))
api = d.get("apierr") # api_error_status of the result, e.g. 429
is_error = "error" in d or api is not None
if is_error:
turns = 0 # rejected before the model answered: nothing billed
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 [])
errtext = d.get("errtext", "").replace("_", " ")
if is_error and not errtext:
errtext = ("API Error: " + api) if api else "stub error"
save(os.path.join(SESS, session + ".json"), {"modelUsage": mu, "cost": total, "context": prev})
emit({"type": "result", "subtype": "error_during_execution" if "error" in d else "success",
"is_error": is_error, "api_error_status": int(api) if api else None, "num_turns": turns,
"total_cost_usd": total, "usage": usage, "modelUsage": mu, "permission_denials": denials,
"session_id": session, "result": errtext if is_error else "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-<job>` 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_api_error_reason(self):
"""I1: API errors get their own reason, from api_error_status or the result text."""
f = forks.api_error_reason
self.assertIsNone(f(None))
self.assertIsNone(f({"is_error": False, "result": "we talked about the rate limit"}))
self.assertIsNone(f({"is_error": True, "subtype": "error_during_execution", "result": "boom"}))
cases = [
({"api_error_status": 429, "result": "x"}, "usage_limit"),
({"api_error_status": "429", "result": "x"}, "usage_limit"),
({"result": "Claude AI usage limit reached|1760000000"}, "usage_limit"),
({"result": "API Error: 429 rate_limit_error"}, "usage_limit"),
({"api_error_status": 529, "result": "x"}, "overloaded"),
({"result": 'API Error: 529 {"type":"error","error":{"type":"overloaded_error"}}'}, "overloaded"),
({"api_error_status": 400, "result": "Prompt is too long"}, "prompt_too_long"),
({"api_error_status": 500, "result": "Internal server error"}, "api_error"),
({"result": "API Error: 502 bad gateway"}, "api_error"),
]
for res, want in cases:
with self.subTest(res=res):
got = f(dict(res, is_error=True))
self.assertIsNotNone(got)
self.assertEqual(got[0], want)
self.assertIn("429", f({"is_error": True, "api_error_status": 429, "result": "x"})[1])
def test_sup_clock_jump(self):
"""M3: a suspend between supervisor ticks is not run time and re-arms the orphan watchdog."""
c = forks.SupClock(1000.0, tick_s=0.25, jump_s=60.0)
for t in (1000.25, 1000.5, 1000.75, 1030.75): # a 30 s stall is not a jump
self.assertIsNone(c.tick(t))
self.assertAlmostEqual(c.elapsed(1030.75), 30.75)
self.assertAlmostEqual(c.tick(8230.75), 7199.75) # 2 h suspend
self.assertAlmostEqual(c.elapsed(8231.0), 31.25) # 30.75 + one tick of the gap + 0.25
last = 990.0 # the last orchestrator call, before it
self.assertAlmostEqual(c.orphan_age(8231.0, last), 0.25) # re-armed at the jump
self.assertGreater(c.orphan_age(8230.75 + 1201, last), 1200) # one more orphan_after
self.assertAlmostEqual(c.orphan_age(9000.0, 8900.0), 100.0) # later calls count as usual
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_api_errors_are_not_model_refused(self):
"""I1: an error before the model answers (nothing billed, models_used []) is not a refusal;
API errors get their own reason; a real substitution still reads model_refused."""
self.setup_run(cap=8)
self.train("B1")
self.finish("j0001")
cases = [(("error",), "is_error", None),
(("apierr=429",), "usage_limit", "429"),
(("error", "errtext=Claude_AI_usage_limit_reached"), "usage_limit", "usage limit"),
(("apierr=529",), "overloaded", "529"),
(("apierr=400", "errtext=Prompt_is_too_long"), "prompt_too_long", "400"),
(("apierr=500",), "api_error", "500"),
(("model=claude-other-9",), "model_refused", "claude-other-9")]
jobs = [self.fork(*c)["job"] for c, _, _ in cases]
code, out = self.wait(*jobs)
self.assertEqual(code, 4, out)
res = {x["job"]: x for x in out["done"]}
reg = self.reg()
for job, (directives, reason, detail) in zip(jobs, cases):
with self.subTest(directives=directives):
r = res[job]
self.assertEqual((r["status"], r["reason"]), ("failed", reason), r)
if reason != "model_refused":
self.assertEqual(r["models_used"], [])
self.assertEqual(r["cost_delta_usd"], 0.0)
if detail:
self.assertIn(detail, reg["jobs"][job]["reason_detail"])
def test_continue_leaves_base_last_hit(self):
"""I3: a continue reads the fork's own longer prefix; only a fork refreshes the base."""
self.setup_run()
self.train("B1")
self.finish("j0001")
fj = self.fork("turns=1")["job"]
self.assertEqual(self.finish(fj)["status"], "ok")
old = forks._iso(time.time() - 1000)
with forks.RegistryLock(self.state):
reg = forks.read_registry(self.state)
reg["bases"]["B1"]["last_hit"] = old
forks.write_registry(self.state, reg)
r = self.finish(self.cmd("continue", "--job", fj, "--brief", self.brief("turns=2"))["job"])
self.assertEqual((r["status"], r["kind"]), ("ok", "continue"))
self.assertEqual(self.reg()["bases"]["B1"]["last_hit"], old)
r = self.finish(self.fork("turns=1")["job"]) # a fork still refreshes it
self.assertEqual(r["status"], "ok")
self.assertGreater(forks._ts(self.reg()["bases"]["B1"]["last_hit"]), forks._ts(old))
def test_clock_jump_not_orphaned_or_timed_out(self):
"""M3: a supervisor (run in-process under an injected clock) sees the clock jump 2 h mid-run,
as after a host suspend. Neither orphan_after (600 s) nor max_minutes (1) may fire; the
result carries `clock_jump` and the jump is logged."""
self.setup_run() # orphan_after 600 s
self.train("B1")
self.finish("j0001")
p = subprocess.Popen([sys.executable, "-c", "pass"])
dead = forks.pid_record(p.pid) # recorded as the supervisor: never us
p.wait()
fake = {"ok": True, "method": "test", "pid": dead["pid"], "ctime": dead["ctime"]}
with mock.patch.object(forks, "spawn_supervisor", lambda *a: fake):
job = self.fork("sleep=1", "turns=3", extra=("--max-minutes", "1"))["job"]
sup = forks.Supervisor(self.state, job)
sup.rt["supervisor"] = dead
offset = [0.0]
real = time.time
jump = threading.Timer(1.0, offset.__setitem__, args=(0, 7200.0))
err = io.StringIO()
try:
with mock.patch.dict(os.environ), mock.patch.object(forks, "_now", lambda: real() + offset[0]), \
contextlib.redirect_stderr(err):
jump.start()
out, code = sup.run()
finally:
jump.cancel()
self.assertEqual(offset[0], 7200.0, "the clock never jumped during the run")
self.assertEqual(code, 0, out)
e = self.reg()["jobs"][job]
self.assertEqual((e["status"], e["reason"]), ("ok", None), e)
self.assertIn("clock_jump", e["flags"])
self.assertIn("clock_jump", forks.read_json(forks.result_file(self.state, job))["flags"])
self.assertIn("clock jump", err.getvalue())
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)
Reference in New Issue View Git Blame Copy Permalink