1588 lines
75 KiB
Python
1588 lines
75 KiB
Python
# /// 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_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_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"))
|
|
ctx = int(prior.get("context") or 0)
|
|
tot = {"in": 0, "out": 0, "cw": 0, "cr": 0}
|
|
prev = ctx
|
|
for i in range(turns):
|
|
if i == 0:
|
|
mode = d.get("cache", "hit" if resume else "cold")
|
|
cr = {"hit": ctx, "partial": ctx // 2, "miss": 0, "cold": 0}[mode]
|
|
cw = (ctx - cr) + 1000
|
|
else:
|
|
cr, cw = prev, 300
|
|
inp, out = 2, 40
|
|
u = {"input_tokens": inp, "cache_creation_input_tokens": cw, "cache_read_input_tokens": cr,
|
|
"cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": cw},
|
|
"output_tokens": 1, "service_tier": "standard"}
|
|
mid = "msg_stub_" + uuid.uuid4().hex[:12]
|
|
for block in ("thinking", "text"): # two events per message id, interim output_tokens
|
|
emit({"type": "assistant", "session_id": session,
|
|
"message": {"id": mid, "model": model.replace("[1m]", ""), "role": "assistant",
|
|
"content": [{"type": block}], "usage": u}})
|
|
emit({"type": "rate_limit_event", "rate_limit_info": {"status": "allowed"}})
|
|
tot["in"] += inp; tot["out"] += out; tot["cw"] += cw; tot["cr"] += cr
|
|
prev = inp + cw + cr
|
|
if i == 0 and "hang" in d:
|
|
time.sleep(600 if d["hang"] == "1" else float(d["hang"]))
|
|
if i == 0 and "die" in d:
|
|
os._exit(int(d.get("code", "3")))
|
|
if "sleep" in d:
|
|
time.sleep(float(d["sleep"]))
|
|
cost = (tot["in"] * PRICE["in"] + tot["out"] * PRICE["out"] + tot["cw"] * PRICE["cw"]
|
|
+ tot["cr"] * PRICE["cr"])
|
|
mu = json.loads(json.dumps(prior.get("modelUsage") or {}))
|
|
m = mu.setdefault(used, {"inputTokens": 0, "outputTokens": 0, "cacheReadInputTokens": 0,
|
|
"cacheCreationInputTokens": 0, "webSearchRequests": 0, "costUSD": 0.0,
|
|
"contextWindow": 1000000})
|
|
m["inputTokens"] += tot["in"]; m["outputTokens"] += tot["out"]
|
|
m["cacheReadInputTokens"] += tot["cr"]; m["cacheCreationInputTokens"] += tot["cw"]
|
|
m["costUSD"] += cost
|
|
total = float(prior.get("cost") or 0.0) + cost
|
|
usage = {"input_tokens": tot["in"], "cache_creation_input_tokens": tot["cw"],
|
|
"cache_read_input_tokens": tot["cr"],
|
|
"output_tokens": tot["out"] + (7 if "mismatch" in d else 0),
|
|
"cache_creation": {"ephemeral_1h_input_tokens": tot["cw"], "ephemeral_5m_input_tokens": 0}}
|
|
denials = ([{"tool_name": "Bash", "tool_use_id": "toolu_stub", "tool_input": {"command": "rm -rf build"}}]
|
|
if "deny" in d else [])
|
|
is_error = "error" in d
|
|
save(os.path.join(SESS, session + ".json"), {"modelUsage": mu, "cost": total, "context": prev})
|
|
emit({"type": "result", "subtype": "error_during_execution" if is_error else "success",
|
|
"is_error": is_error, "num_turns": turns, "total_cost_usd": total, "usage": usage,
|
|
"modelUsage": mu, "permission_denials": denials, "session_id": session, "result": "READY"})
|
|
return 1 if is_error else int(d.get("exit", "0"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|
|
'''
|
|
|
|
|
|
def stub_interpreter() -> str:
|
|
base = getattr(sys, "_base_executable", None)
|
|
return base if base and os.path.isfile(base) else sys.executable
|
|
|
|
|
|
def make_stream_stub(d: str) -> str:
|
|
"""Generate the stream stub under `d` (never committed). Windows: a `.cmd` wrapper that runs
|
|
the base interpreter with -I; POSIX: a sh wrapper."""
|
|
sd = os.path.join(d, "stub")
|
|
os.makedirs(sd, exist_ok=True)
|
|
py = os.path.join(sd, "stub.py")
|
|
with open(py, "w", newline="\n") as f:
|
|
f.write(STUB_PY % {"version": STUB_VERSION, "price": PRICE})
|
|
if IS_WINDOWS:
|
|
p = os.path.join(sd, "claude.cmd")
|
|
with open(p, "w", newline="\r\n") as f:
|
|
f.write('@echo off\n"%s" -I "%s" %%*\n' % (stub_interpreter(), py))
|
|
else:
|
|
p = os.path.join(sd, "claude")
|
|
with open(p, "w", newline="\n") as f:
|
|
f.write('#!/bin/sh\nexec "%s" -I "%s" "$@"\n' % (stub_interpreter(), py))
|
|
os.chmod(p, 0o755)
|
|
return p
|
|
|
|
|
|
def stub_calls(stub: str) -> list[dict]:
|
|
d = os.path.join(os.path.dirname(stub), "calls")
|
|
out = []
|
|
for n in sorted(os.listdir(d)) if os.path.isdir(d) else []:
|
|
if n.endswith(".json"):
|
|
with open(os.path.join(d, n)) as f:
|
|
out.append(json.load(f))
|
|
return out
|
|
|
|
|
|
def stub_call_for(stub: str, job_prompt_marker: str) -> dict | None:
|
|
for c in stub_calls(stub):
|
|
if job_prompt_marker in c.get("prompt", ""):
|
|
return c
|
|
return None
|
|
|
|
|
|
def price(tokens: dict) -> float:
|
|
return (tokens["input"] * PRICE["in"] + tokens["output"] * PRICE["out"]
|
|
+ tokens["cache_creation"] * PRICE["cw"] + tokens["cache_read"] * PRICE["cr"])
|
|
|
|
|
|
def visible_windows() -> list[tuple]:
|
|
"""Visible top-level windows: [(hwnd, pid, class, title)] (Windows only)."""
|
|
import ctypes
|
|
import ctypes.wintypes as wt
|
|
u32 = ctypes.WinDLL("user32", use_last_error=True)
|
|
out = []
|
|
proto = ctypes.WINFUNCTYPE(wt.BOOL, wt.HWND, wt.LPARAM)
|
|
|
|
def cb(hwnd, _):
|
|
if u32.IsWindowVisible(hwnd):
|
|
pid = wt.DWORD()
|
|
u32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
|
|
cls = ctypes.create_unicode_buffer(256)
|
|
u32.GetClassNameW(hwnd, cls, 256)
|
|
ttl = ctypes.create_unicode_buffer(256)
|
|
u32.GetWindowTextW(hwnd, ttl, 256)
|
|
out.append((int(hwnd or 0), pid.value, cls.value, ttl.value))
|
|
return True
|
|
u32.EnumWindows(proto(cb), 0)
|
|
return out
|
|
|
|
|
|
def wait_for(pred, timeout=30.0, step=0.1):
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
v = pred()
|
|
if v:
|
|
return v
|
|
time.sleep(step)
|
|
return pred()
|
|
|
|
|
|
STATE_DIRS: list[str] = [] # every launch test's state dir, for the module-end process check
|
|
|
|
|
|
class LaunchBase(Base):
|
|
"""One run per test: a git repo as cwd, a stream stub as --binary, the real detached spawn."""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
for name, val in (("WAIT_POLL_S", 0.2),):
|
|
p = mock.patch.object(forks, name, val)
|
|
p.start()
|
|
self.addCleanup(p.stop)
|
|
p = mock.patch.object(forks, "USER_SETTINGS_PATH", os.path.join(self.tmp, "home", "settings.json"))
|
|
p.start()
|
|
self.addCleanup(p.stop)
|
|
self.cwd = os.path.join(self.tmp, "proj")
|
|
os.makedirs(self.cwd)
|
|
self.git("init", "-q")
|
|
self.commit_file("README.md", "hello\n", "initial")
|
|
os.makedirs(os.path.join(self.tmp, "reports"))
|
|
self.run_log = os.path.join(self.tmp, "reports", "2026-10-16-12-t-run.md")
|
|
self.state = forks.state_dir_for(self.run_log)
|
|
STATE_DIRS.append(self.state)
|
|
self.stub = make_stream_stub(self.tmp)
|
|
self.addCleanup(self.cleanup_processes)
|
|
self.nbrief = 0
|
|
|
|
# --- helpers ---
|
|
def git(self, *args):
|
|
r = subprocess.run(["git", "-c", "user.name=t", "-c", "user.email=t@t", *args], cwd=self.cwd,
|
|
capture_output=True, timeout=60)
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
return r.stdout.decode().strip()
|
|
|
|
def commit_file(self, name, text, msg):
|
|
with open(os.path.join(self.cwd, name), "w", newline="\n") as f:
|
|
f.write(text)
|
|
self.git("add", name)
|
|
self.git("commit", "-q", "-m", msg)
|
|
|
|
def setup_run(self, *extra, orphan_after=600, stall=900, cap=2):
|
|
code, out, _ = call(["init", "--run", self.run_log, "--model", TEST_MODEL, "--cwd", self.cwd,
|
|
"--binary", self.stub, "--orphan-after", str(orphan_after),
|
|
"--stall-after-s", str(stall), "--parallel-cap", str(cap), *extra])
|
|
self.assertEqual(code, 0, out)
|
|
return out
|
|
|
|
def brief(self, *directives, text="Do the thing."):
|
|
self.nbrief += 1
|
|
p = os.path.join(self.tmp, "brief-%d.md" % self.nbrief)
|
|
marker = "BRIEF-MARKER-%d" % self.nbrief
|
|
with open(p, "w", newline="\n") as f:
|
|
f.write("%s\n%s\n" % (marker, text))
|
|
if directives:
|
|
f.write("@stub %s\n" % " ".join(directives))
|
|
return p
|
|
|
|
def cmd(self, *argv, code=0):
|
|
c, out, raw = call([argv[0], "--run", self.run_log, *argv[1:]])
|
|
self.assert_one_json_line(raw)
|
|
if code is not None:
|
|
self.assertEqual(c, code, out)
|
|
return out
|
|
|
|
def train(self, base="B1", *directives):
|
|
out = self.cmd("train", "--base", base, "--brief", self.brief(*directives))
|
|
self.assertEqual(set(out), {"job", "base", "session_id"})
|
|
return out
|
|
|
|
def fork(self, *directives, base="B1", block="1.1", extra=(), code=0):
|
|
return self.cmd("fork", "--base", base, "--block", block, "--brief", self.brief(*directives),
|
|
*extra, code=code)
|
|
|
|
def wait(self, *jobs, timeout=60, any_=False):
|
|
argv = ["wait", "--run", self.run_log, "--timeout", str(timeout)]
|
|
argv += list(jobs) if jobs else ["--all"]
|
|
if any_:
|
|
argv.append("--any")
|
|
code, out, raw = call(argv)
|
|
self.assert_one_json_line(raw)
|
|
return code, out
|
|
|
|
def finish(self, job, timeout=60):
|
|
"""Wait for one job and return its result (from `done`)."""
|
|
code, out = self.wait(job, timeout=timeout)
|
|
self.assertIn(code, (0, 4), out)
|
|
got = [r for r in out["done"] if r["job"] == job]
|
|
self.assertEqual(len(got), 1, out)
|
|
return got[0]
|
|
|
|
def reg(self):
|
|
return forks.read_registry(self.state)
|
|
|
|
def runtime(self, job):
|
|
return forks.read_json(forks.job_file(self.state, job)) or {}
|
|
|
|
def ledger(self):
|
|
p = forks.ledger_path(self.state)
|
|
if not os.path.exists(p):
|
|
return []
|
|
with open(p, "rb") as f:
|
|
return [json.loads(x) for x in f.read().decode("utf-8").splitlines() if x.strip()]
|
|
|
|
def until_running(self, job, need_child=False):
|
|
"""Wait until the job's claude runs and the stub (started by a `fork_marked` brief) has
|
|
recorded itself, with its grandchild when `need_child`."""
|
|
def ready():
|
|
rt = self.runtime(job)
|
|
if not rt.get("claude"):
|
|
return None
|
|
c = stub_call_for(self.stub, "RUN-%s" % job)
|
|
if not c or (need_child and not c.get("child_pid")):
|
|
return None
|
|
return rt
|
|
rt = wait_for(ready, timeout=30)
|
|
self.assertTrue(rt, "job %s never got its claude running" % job)
|
|
return rt
|
|
|
|
def assert_tree_dead(self, recs, timeout=15):
|
|
wait_for(lambda: not any(forks.record_alive(r) for r in recs), timeout=timeout)
|
|
alive = [r for r in recs if forks.record_alive(r)]
|
|
self.assertEqual(alive, [], "processes still alive: %r" % alive)
|
|
|
|
def cleanup_processes(self):
|
|
reg = None
|
|
with contextlib.suppress(Exception):
|
|
reg = forks.read_registry(self.state)
|
|
for job, e in ((reg or {}).get("jobs") or {}).items():
|
|
rt = self.runtime(job)
|
|
recs = [r for r in (rt.get("supervisor"), e.get("supervisor"), rt.get("claude")) if r]
|
|
if recs:
|
|
forks.kill_tree(recs[0], extra=recs[1:])
|
|
|
|
def fork_marked(self, *directives, **kw):
|
|
"""A fork whose brief carries `RUN-<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_stream_state(self):
|
|
ss = forks.StreamState()
|
|
u1 = {"input_tokens": 2, "cache_creation_input_tokens": 10, "cache_read_input_tokens": 100}
|
|
u2 = {"input_tokens": 2, "cache_creation_input_tokens": 5, "cache_read_input_tokens": 112}
|
|
for ev in ({"type": "system", "subtype": "init", "session_id": "s1", "tools": ["Read", "Bash"],
|
|
"permissionMode": "dontAsk", "apiKeySource": "none",
|
|
"plugins": [{"name": "spire", "version": "0.2.2"}], "agents": ["claude"]},
|
|
{"type": "assistant", "message": {"id": "m1", "usage": u1}},
|
|
{"type": "assistant", "message": {"id": "m1", "usage": u1}},
|
|
{"type": "system", "subtype": "informational", "level": "warning",
|
|
"content": "Model \"x\" is restricted by your organization's settings."},
|
|
{"type": "assistant", "message": {"id": "m2", "usage": u2}},
|
|
{"type": "result", "num_turns": 2}):
|
|
ss.feed(ev, 5.0)
|
|
self.assertEqual((ss.session_id, ss.turns), ("s1", 2))
|
|
self.assertEqual(ss.first_usage, u1)
|
|
self.assertEqual(forks.context_tokens(ss.last_usage), 119)
|
|
self.assertIn("is restricted", ss.refused)
|
|
rt = forks.runtime_fingerprint(ss.init)
|
|
self.assertEqual(rt["plugins"], ["spire@0.2.2"])
|
|
other = dict(rt, plugins=["spire@0.3.0"], apiKeySource="oauth")
|
|
self.assertEqual(forks.runtime_drift(other, rt, {}), ["apiKeySource", "plugins"])
|
|
self.assertEqual(forks.runtime_drift(rt, None, {"permission_mode": "default"}), ["permissionMode"])
|
|
|
|
|
|
class LaunchTests(LaunchBase):
|
|
def test_train_fork_continue_end_to_end(self):
|
|
self.setup_run()
|
|
t = self.train("B1", "turns=4")
|
|
self.assertEqual((t["job"], t["base"]), ("j0001", "B1"))
|
|
r = self.finish("j0001")
|
|
self.assertEqual(r["status"], "ok", r)
|
|
self.assertEqual(set(r), {"job", "kind", "base", "block", "phase", "status", "flags", "fingerprint",
|
|
"session", "parent_session", "models_used", "tokens", "cost_delta_usd",
|
|
"cache", "base_state", "wall_s", "turns", "denials", "note", "raw",
|
|
"paths", "reason", "model_requested"})
|
|
self.assertEqual(r["kind"], "train")
|
|
self.assertIsNone(r["cache"])
|
|
self.assertEqual(r["session"], t["session_id"])
|
|
self.assertIsNone(r["parent_session"])
|
|
self.assertEqual(r["models_used"], [TEST_MODEL])
|
|
self.assertEqual(r["model_requested"], TEST_MODEL)
|
|
self.assertEqual(r["fingerprint"], "ok")
|
|
self.assertEqual(r["turns"], 4)
|
|
self.assertEqual(r["flags"], [])
|
|
self.assertIsNone(r["reason"])
|
|
self.assertAlmostEqual(r["cost_delta_usd"], price(r["tokens"]), places=6)
|
|
self.assertEqual(r["raw"], "raw/j0001.jsonl")
|
|
self.assertTrue(os.path.isfile(os.path.join(self.state, r["raw"])))
|
|
base = self.reg()["bases"]["B1"]
|
|
self.assertEqual(base["status"], "active")
|
|
self.assertEqual(base["session"], t["session_id"])
|
|
self.assertEqual(base["base_commit"], self.git("rev-parse", "HEAD"))
|
|
# base context = input + creation + read of the last assistant message: 2 + 300 + (prev)
|
|
self.assertEqual(base["context_tokens"], 2 + 1000 + 3 * (2 + 300) + 0)
|
|
self.assertIsNotNone(base["last_hit"])
|
|
self.assertAlmostEqual(base["training_cost_usd"], r["cost_delta_usd"], places=6)
|
|
# the command line: absolute stored binary, full fingerprint, pre-assigned session id
|
|
call0 = stub_calls(self.stub)[0]
|
|
argv = call0["argv"]
|
|
self.assertEqual(argv[argv.index("--session-id") + 1], t["session_id"])
|
|
self.assertEqual(argv[argv.index("--model") + 1], TEST_MODEL)
|
|
self.assertEqual(argv[argv.index("--permission-mode") + 1], "dontAsk")
|
|
self.assertIn("--strict-mcp-config", argv)
|
|
self.assertEqual(argv[argv.index("--append-system-prompt-file") + 1], forks.preamble_path())
|
|
self.assertNotIn("--resume", argv)
|
|
self.assertEqual(self.reg()["jobs"]["j0001"]["argv"][0], forks.norm_path(self.stub))
|
|
self.assertEqual(os.path.normcase(call0["cwd"]), os.path.normcase(self.cwd))
|
|
self.assertEqual(call0["env"]["CLAUDE_CODE_EXECPATH"], forks.norm_path(self.stub))
|
|
self.assertEqual(call0["env"]["SPIRE_FORKS_CHILD"], "1")
|
|
self.assertEqual(call0["env"]["CLAUDE_CODE_PROMPT_CACHE_TTL"], "1h")
|
|
self.assertIsNone(call0["env"]["VIRTUAL_ENV"])
|
|
self.assertIn("READY", call0["prompt"]) # templates/train.md frame
|
|
# --- fork, after a commit on the trunk ---
|
|
self.commit_file("src.py", "x = 1\n", "add src after training")
|
|
os.makedirs(os.path.join(self.state, "notes"), exist_ok=True)
|
|
with open(os.path.join(self.state, "notes", "1.1.md"), "w") as f:
|
|
f.write("note\n")
|
|
f_out = self.fork("turns=2", extra=("--phase", "1", "--paths", "src.py", "docs/a.md"))
|
|
self.assertEqual(set(f_out), {"job", "base", "base_state", "warm_until", "rebase_due",
|
|
"retrain_due", "reasons"})
|
|
self.assertEqual(f_out["base_state"], "warm")
|
|
self.assertFalse(f_out["rebase_due"])
|
|
self.assertFalse(f_out["retrain_due"])
|
|
self.assertEqual(f_out["reasons"], [])
|
|
fj = f_out["job"]
|
|
r2 = self.finish(fj)
|
|
self.assertEqual(r2["status"], "ok", r2)
|
|
self.assertEqual(r2["kind"], "fork")
|
|
self.assertEqual(r2["cache"], "hit")
|
|
self.assertEqual(r2["phase"], 1)
|
|
self.assertEqual(r2["base_state"], "warm")
|
|
self.assertEqual(r2["flags"], [])
|
|
self.assertEqual(r2["fingerprint"], "ok")
|
|
self.assertEqual(r2["parent_session"], t["session_id"])
|
|
self.assertNotEqual(r2["session"], t["session_id"])
|
|
self.assertEqual(r2["paths"], ["src.py", "docs/a.md"])
|
|
self.assertEqual(r2["note"], "notes/1.1.md")
|
|
self.assertEqual(r2["models_used"], [TEST_MODEL])
|
|
# first request reads the base context, the second the base context plus the first turn
|
|
self.assertEqual(r2["tokens"]["cache_read"], 2 * base["context_tokens"] + 1002)
|
|
self.assertAlmostEqual(r2["cost_delta_usd"], price(r2["tokens"]), places=6)
|
|
fcall = [c for c in stub_calls(self.stub) if c["session"] == r2["session"]][0]
|
|
a = fcall["argv"]
|
|
self.assertEqual(a[a.index("--resume") + 1], t["session_id"])
|
|
self.assertIn("--fork-session", a)
|
|
self.assertEqual(a[a.index("--session-id") + 1], r2["session"])
|
|
self.assertIn("--append-system-prompt-file", a) # flags passed again
|
|
with open(os.path.join(self.state, "briefs", fj + ".md"), encoding="utf-8") as fh:
|
|
brief_txt = fh.read()
|
|
self.assertIn("add src after training", brief_txt) # delta header: git log
|
|
self.assertIn("1 file changed", brief_txt) # delta header: shortstat
|
|
self.assertIn("stale", brief_txt) # R13 sentence
|
|
self.assertIn(os.path.join(self.state, "notes", "1.1.md"), brief_txt)
|
|
# --- continue the fork: cost delta against the fork's own last total ---
|
|
c_out = self.cmd("continue", "--job", fj, "--brief", self.brief("turns=1"))
|
|
self.assertEqual(set(c_out), {"job"})
|
|
r3 = self.finish(c_out["job"])
|
|
self.assertEqual(r3["status"], "ok", r3)
|
|
self.assertEqual(r3["kind"], "continue")
|
|
self.assertEqual(r3["session"], r2["session"])
|
|
self.assertEqual(r3["block"], "1.1")
|
|
self.assertAlmostEqual(r3["cost_delta_usd"], price(r3["tokens"]), places=6)
|
|
self.assertEqual(r3["cache"], "hit")
|
|
ccall = [c for c in stub_calls(self.stub) if c["session"] == r2["session"] and c is not fcall
|
|
and "--fork-session" not in c["argv"]][0]
|
|
self.assertEqual(ccall["argv"][ccall["argv"].index("--resume") + 1], r2["session"])
|
|
self.assertNotIn("--session-id", ccall["argv"])
|
|
# ledger lines match the results
|
|
led = {x["job"]: x for x in self.ledger()}
|
|
self.assertEqual(sorted(led), ["j0001", fj, c_out["job"]])
|
|
for res in (r, r2, r3):
|
|
line = led[res["job"]]
|
|
for k in ("status", "flags", "tokens", "cost_delta_usd", "cache", "models_used", "session",
|
|
"parent_session", "fingerprint", "turns", "reason", "note"):
|
|
self.assertEqual(line[k], res[k], k)
|
|
self.assertEqual(line["binary"], STUB_VERSION)
|
|
self.assertIsNotNone(line["cost_cumulative_usd"])
|
|
# list shows every base and job; nothing running
|
|
lst = self.cmd("list")
|
|
self.assertEqual([b["name"] for b in lst["bases"]], ["B1"])
|
|
self.assertEqual(lst["bases"][0]["base_state"], "warm")
|
|
self.assertEqual(len(lst["jobs"]), 3)
|
|
self.assertEqual(lst["running"], [])
|
|
self.assertIsNotNone(self.reg()["run"]["last_orchestrator_call_at"])
|
|
# refusals on continue
|
|
self.cmd("continue", "--job", "j0001", "--brief", self.brief(), code=2) # not a fork
|
|
self.cmd("continue", "--job", "j0999", "--brief", self.brief(), code=2) # unknown
|
|
self.cmd("train", "--base", "B1", "--brief", self.brief(), code=2) # base exists
|
|
|
|
def test_cache_verdicts_and_cold_start(self):
|
|
self.setup_run()
|
|
self.train("B1")
|
|
self.finish("j0001")
|
|
a = self.fork("cache=partial", extra=("--max-budget-usd", "2.5"))["job"]
|
|
b = self.fork("cache=miss")["job"]
|
|
argv_a = self.reg()["jobs"][a]["argv"]
|
|
self.assertEqual(argv_a[argv_a.index("--max-budget-usd") + 1], "2.5")
|
|
self.assertNotIn("--max-budget-usd", self.reg()["jobs"][b]["argv"])
|
|
code, out = self.wait(a, b)
|
|
self.assertEqual(code, 0, out)
|
|
res = {x["job"]: x for x in out["done"]}
|
|
self.assertEqual(res[a]["cache"], "partial")
|
|
self.assertEqual(res[b]["cache"], "miss")
|
|
for j in (a, b):
|
|
self.assertIn("cold_start", res[j]["flags"])
|
|
self.assertEqual(res[j]["status"], "ok")
|
|
self.assertEqual(res[j]["base_state"], "warm")
|
|
# a cold base: hit, but cold at launch -> cold_start
|
|
with forks.RegistryLock(self.state):
|
|
reg = forks.read_registry(self.state)
|
|
reg["bases"]["B1"]["last_hit"] = forks._iso(time.time() - 4000)
|
|
forks.write_registry(self.state, reg)
|
|
out = self.fork()
|
|
self.assertEqual(out["base_state"], "cold")
|
|
r = self.finish(out["job"])
|
|
self.assertEqual(r["cache"], "hit")
|
|
self.assertEqual(r["base_state"], "cold")
|
|
self.assertIn("cold_start", r["flags"])
|
|
# the hit refreshed last_hit: the base is warm again
|
|
self.assertEqual(self.cmd("list")["bases"][0]["base_state"], "warm")
|
|
|
|
def test_failures_flags_and_drift(self):
|
|
self.setup_run()
|
|
self.train("B1")
|
|
self.finish("j0001")
|
|
cases = [("die=1",), ("error",), ("restricted=1",), ("model=claude-other-9",), ("deny",),
|
|
("mismatch",), ("plugin=0.3.0",)]
|
|
results = {}
|
|
for i in range(0, len(cases), 2):
|
|
jobs = [self.fork(*c)["job"] for c in cases[i:i + 2]]
|
|
code, out = self.wait(*jobs)
|
|
for x in out["done"]:
|
|
results[x["job"]] = x
|
|
r = [results["j%04d" % (i + 2)] for i in range(len(cases))]
|
|
die, err, restricted, substituted, deny, mismatch, plugin = r
|
|
self.assertEqual((die["status"], die["reason"]), ("failed", "exit_code"))
|
|
self.assertEqual(die["turns"], 1)
|
|
self.assertIsNone(die["cost_delta_usd"])
|
|
self.assertEqual((err["status"], err["reason"]), ("failed", "is_error"))
|
|
self.assertEqual((restricted["status"], restricted["reason"]), ("failed", "model_refused"))
|
|
self.assertEqual((substituted["status"], substituted["reason"]), ("failed", "model_refused"))
|
|
self.assertEqual(substituted["models_used"], ["claude-other-9"])
|
|
self.assertEqual(deny["status"], "ok")
|
|
self.assertIn("needs_attention", deny["flags"])
|
|
self.assertEqual(deny["denials"], 1)
|
|
self.assertEqual(mismatch["status"], "ok")
|
|
self.assertIn("usage_mismatch", mismatch["flags"])
|
|
self.assertEqual(plugin["fingerprint"], "drift:plugins")
|
|
self.assertNotIn("cold_start", plugin["flags"])
|
|
# model refused while training: the base fails
|
|
t = self.train("B2", "restricted=1")
|
|
r = self.finish(t["job"])
|
|
self.assertEqual((r["status"], r["reason"]), ("failed", "model_refused"))
|
|
self.assertEqual(self.reg()["bases"]["B2"]["status"], "failed")
|
|
self.fork(base="B2", code=2) # not active
|
|
# binary drift: the stored binary changed size and mtime since training
|
|
with open(self.stub, "a", newline="\r\n" if IS_WINDOWS else "\n") as f:
|
|
f.write("rem drift\n" if IS_WINDOWS else "# drift\n")
|
|
r = self.finish(self.fork()["job"])
|
|
self.assertEqual(r["status"], "ok")
|
|
self.assertEqual(r["fingerprint"], "drift:binary")
|
|
|
|
def test_timeout_and_stalled(self):
|
|
self.setup_run(stall=1)
|
|
self.train("B1")
|
|
self.finish("j0001")
|
|
slow = self.fork("silent=2.5")["job"]
|
|
hang = self.fork_marked("hang=1", extra=("--max-minutes", "0.06"))["job"]
|
|
rt = self.until_running(hang)
|
|
tree = forks.process_tree(rt["supervisor"])
|
|
self.assertGreaterEqual(len(tree), 3)
|
|
code, out = self.wait(slow, hang, timeout=60)
|
|
self.assertEqual(code, 4, out)
|
|
res = {x["job"]: x for x in out["done"]}
|
|
self.assertEqual(res[slow]["status"], "ok")
|
|
self.assertIn("stalled", res[slow]["flags"])
|
|
self.assertEqual(res[hang]["status"], "timeout")
|
|
self.assertIsNone(res[hang]["reason"])
|
|
self.assertIn("stalled", res[hang]["flags"])
|
|
self.assert_tree_dead(tree) # claude tree killed; the supervisor exited
|
|
|
|
def test_kill_tree_and_no_visible_window(self):
|
|
self.setup_run()
|
|
self.train("B1")
|
|
self.finish("j0001")
|
|
job = self.fork_marked("hang=1", "child")["job"]
|
|
rt = self.until_running(job, need_child=True)
|
|
sup = rt["supervisor"]
|
|
tree = forks.process_tree(sup)
|
|
pids = {p["pid"] for p in tree}
|
|
stub = stub_call_for(self.stub, "RUN-%s" % job)
|
|
self.assertIn(sup["pid"], pids)
|
|
self.assertIn(rt["claude"]["pid"], pids)
|
|
self.assertIn(stub["pid"], pids)
|
|
self.assertIn(stub["child_pid"], pids)
|
|
if IS_WINDOWS:
|
|
# spawned through CIM: the supervisor's parent is the WMI provider host, not us
|
|
table = {pid: (ppid, name) for pid, ppid, name in forks.process_table()}
|
|
parent = table.get(table[sup["pid"]][0], (None, None))[1]
|
|
self.assertEqual((parent or "").lower(), "wmiprvse.exe")
|
|
self.assertIn("python.exe", [p["name"].lower() for p in tree])
|
|
self.assertIn("cmd.exe", [p["name"].lower() for p in tree])
|
|
all_windows = visible_windows()
|
|
self.assertGreater(len(all_windows), 0, "window enumeration sees nothing (no desktop?)")
|
|
for _ in range(3):
|
|
mine = [w for w in visible_windows() if w[1] in pids]
|
|
self.assertEqual(mine, [], "visible window of the job's process tree")
|
|
time.sleep(0.2)
|
|
out = self.cmd("kill", "--job", job)
|
|
self.assertTrue(out["killed"])
|
|
self.assertTrue(out["ok"], out)
|
|
self.assertEqual(out["survivors"], [])
|
|
self.assertEqual(out["status"], "killed")
|
|
snap_pids = {p["pid"] for p in out["snapshot"]}
|
|
for pid in (sup["pid"], rt["claude"]["pid"], stub["pid"], stub["child_pid"]):
|
|
self.assertIn(pid, snap_pids)
|
|
self.assert_tree_dead(tree, timeout=1)
|
|
r = self.finish(job)
|
|
self.assertEqual((r["status"], r["reason"]), ("killed", "kill"))
|
|
self.assertEqual(r["turns"], 1)
|
|
self.assertEqual([x["status"] for x in self.ledger() if x["job"] == job], ["killed"])
|
|
again = self.cmd("kill", "--job", job)
|
|
self.assertFalse(again["killed"])
|
|
self.cmd("kill", "--job", "j0999", code=2)
|
|
|
|
def test_lost(self):
|
|
self.setup_run()
|
|
self.train("B1")
|
|
self.finish("j0001")
|
|
job = self.fork_marked("hang=1")["job"]
|
|
rt = self.until_running(job)
|
|
tree = forks.process_tree(rt["supervisor"])
|
|
self.assertTrue(forks.terminate(rt["supervisor"]))
|
|
self.assertTrue(wait_for(lambda: not forks.record_alive(rt["supervisor"]), 10))
|
|
lst = self.cmd("list") # heartbeat is fresh: not lost yet
|
|
self.assertEqual(lst["running"], [job])
|
|
later = time.time() + forks.LOST_AFTER_S + 30
|
|
with mock.patch.object(forks, "_now", lambda: later):
|
|
lst = self.cmd("list")
|
|
self.assertEqual(lst["lost_now"], [job])
|
|
self.assertEqual(lst["running"], [])
|
|
self.assert_tree_dead(tree) # its orphaned claude tree went with it
|
|
code, out = self.wait(job)
|
|
self.assertEqual(code, 4)
|
|
self.assertEqual((out["done"][0]["status"], out["done"][0]["reason"]), ("lost", "supervisor_dead"))
|
|
self.assertEqual([x["status"] for x in self.ledger() if x["job"] == job], ["lost"])
|
|
|
|
def test_orphan_watchdog(self):
|
|
self.setup_run() # orphan_after 600 s
|
|
self.train("B1")
|
|
self.finish("j0001")
|
|
job = self.fork_marked("hang=1", "child")["job"]
|
|
rt = self.until_running(job, need_child=True)
|
|
tree = forks.process_tree(rt["claude"])
|
|
self.assertGreaterEqual(len(tree), 2)
|
|
self.assertEqual(self.reg()["jobs"][job]["status"], "running")
|
|
# injected clock: the last orchestrator call is recorded as 2000 s ago
|
|
past = time.time() - 2000
|
|
with mock.patch.object(forks, "_now", lambda: past):
|
|
forks.touch_orchestrator(self.state)
|
|
# no orchestrator call from here on: poll the registry directly (no touch)
|
|
final = wait_for(lambda: self.reg()["jobs"][job]["status"] not in forks.RUNNING, timeout=30, step=0.2)
|
|
self.assertTrue(final)
|
|
e = self.reg()["jobs"][job]
|
|
self.assertEqual((e["status"], e["reason"]), ("killed", "orphaned"))
|
|
self.assertIn("orphaned", e["flags"])
|
|
self.assert_tree_dead(tree)
|
|
self.assertTrue(wait_for(lambda: not forks.record_alive(rt["supervisor"]), 15))
|
|
self.assertEqual([x["flags"] for x in self.ledger() if x["job"] == job], [e["flags"]])
|
|
|
|
def test_expire_kills_running_jobs(self):
|
|
self.setup_run()
|
|
self.train("B1")
|
|
self.finish("j0001")
|
|
job = self.fork_marked("hang=1")["job"]
|
|
rt = self.until_running(job)
|
|
tree = forks.process_tree(rt["supervisor"])
|
|
out = self.cmd("expire", "--base", "B1")
|
|
self.assertEqual(out["retired"], ["B1"])
|
|
self.assertEqual([k["job"] for k in out["killed"]], [job])
|
|
self.assertTrue(out["ok"])
|
|
self.assert_tree_dead(tree)
|
|
r = self.finish(job)
|
|
self.assertEqual((r["status"], r["reason"]), ("killed", "expire"))
|
|
self.assertIn("expired", r["flags"])
|
|
self.assertEqual(self.reg()["bases"]["B1"]["status"], "retired")
|
|
self.fork(code=2) # retired: refused without --force
|
|
r = self.finish(self.fork(extra=("--force",))["job"])
|
|
self.assertEqual(r["status"], "ok")
|
|
self.assertEqual(self.cmd("expire", "--all")["retired"], [])
|
|
|
|
def test_wait_exit_codes(self):
|
|
self.setup_run()
|
|
self.train("B1")
|
|
self.assertEqual(self.wait()[0], 0) # settles the training job
|
|
a = self.fork("die=1")["job"]
|
|
b = self.fork("sleep=2", "turns=5")["job"]
|
|
wait_for(lambda: self.reg()["jobs"][a]["status"] == "failed", timeout=30)
|
|
code, out = self.wait(timeout=1) # mixed: one failed, one running
|
|
self.assertEqual(code, 3, out)
|
|
self.assertEqual([x["job"] for x in out["done"]], [a])
|
|
self.assertEqual(out["done"][0]["status"], "failed")
|
|
self.assertEqual([x["job"] for x in out["running"]], [b])
|
|
self.assertEqual(set(out["running"][0]), {"job", "block", "elapsed_s", "last_event_s_ago", "turns"})
|
|
self.assertEqual(out["pings"], [])
|
|
code, out = self.wait(timeout=60) # condition met, one failed -> 4
|
|
self.assertEqual(code, 4, out)
|
|
self.assertEqual([x["job"] for x in out["done"]], [b]) # a is not returned twice
|
|
self.assertEqual(out["done"][0]["status"], "ok")
|
|
code, out = self.wait(timeout=5) # nothing left
|
|
self.assertEqual((code, out["done"], out["running"]), (0, [], []))
|
|
self.assertEqual(self.wait(b)[0], 0)
|
|
self.assertEqual(self.wait(a)[0], 4)
|
|
# --any
|
|
c = self.fork("die=1")["job"]
|
|
d = self.fork("sleep=1", "turns=3")["job"]
|
|
code, out = self.wait(timeout=60, any_=True)
|
|
self.assertEqual(code, 4, out)
|
|
self.assertEqual([x["job"] for x in out["done"]], [c])
|
|
code, out = self.wait(timeout=60)
|
|
self.assertEqual(code, 0, out)
|
|
self.assertEqual([x["job"] for x in out["done"]], [d])
|
|
# usage errors
|
|
for argv in (["wait", "--run", self.run_log, "--all", "--timeout", "571"],
|
|
["wait", "--run", self.run_log],
|
|
["wait", "--run", self.run_log, "j0999"],
|
|
["wait", "--run", self.run_log, "--all", "j0001"]):
|
|
self.assertEqual(call(argv)[0], 2, argv)
|
|
|
|
def test_spawn_failed(self):
|
|
self.setup_run()
|
|
failed = {"ok": False, "method": forks.SPAWN_METHOD, "detail": "Win32_Process.Create returned 9"}
|
|
with mock.patch.object(forks, "spawn_supervisor", lambda *a: failed):
|
|
t = self.train("B1")
|
|
code, out = self.wait(t["job"], timeout=5)
|
|
self.assertEqual(code, 4, out)
|
|
r = out["done"][0]
|
|
self.assertEqual((r["status"], r["reason"]), ("failed", "spawn_failed"))
|
|
self.assertEqual(r["tokens"], {k: None for k in forks.TOKEN_KEYS})
|
|
self.assertEqual(self.reg()["bases"]["B1"]["status"], "failed")
|
|
self.assertEqual(self.reg()["jobs"][t["job"]]["reason_detail"], failed["detail"])
|
|
self.assertEqual([x["reason"] for x in self.ledger()], ["spawn_failed"])
|
|
self.assertEqual(stub_calls(self.stub), [])
|
|
|
|
def test_refusals(self):
|
|
# launch-field overrides on fork and continue (refused before anything is read)
|
|
values = {"model": "claude-x-1", "effort": "low", "permission_mode": "default", "tools": "Read",
|
|
"allowed_tools": "Read", "disallowed_tools": "Bash", "agent": "a", "cwd": self.tmp,
|
|
"binary": self.stub}
|
|
for name in forks.LAUNCH_FIELDS:
|
|
flag = "--" + name.replace("_", "-")
|
|
for argv in (["fork", "--base", "B1", "--block", "1", "--brief", "b.md"],
|
|
["continue", "--job", "j0001", "--brief", "b.md"]):
|
|
with self.subTest(field=name, cmd=argv[0]):
|
|
out = self.cmd(*argv, flag, values[name], code=2)
|
|
self.assertEqual(out["fields"], [name])
|
|
self.assertFalse(os.path.exists(self.state))
|
|
self.setup_run()
|
|
# N2: parallel needs --paths
|
|
out = self.cmd("fork", "--base", "B1", "--block", "1", "--brief", self.brief(), "--mode",
|
|
"parallel", code=2)
|
|
self.assertIn("--paths", out["error"])
|
|
self.fork(base="B9", code=2) # unknown base
|
|
self.train("B1")
|
|
self.finish("j0001")
|
|
# parallel cap (2): two hanging parallel forks, a third launch is refused
|
|
j1 = self.fork_marked("hang=1", extra=("--mode", "parallel", "--paths", "a.py"))["job"]
|
|
j2 = self.fork_marked("hang=1", extra=("--mode", "parallel", "--paths", "b.py", "c.py"))["job"]
|
|
out = self.fork(code=2)
|
|
self.assertEqual(out["error"], "parallel cap reached")
|
|
self.assertEqual(out["running"], [j1, j2])
|
|
out = self.cmd("continue", "--job", j1, "--brief", self.brief(), code=2) # still running
|
|
self.assertIn("running", out["error"])
|
|
# N3: launching refused inside a fork; list and wait stay allowed
|
|
os.environ[forks.CHILD_MARKER] = "1"
|
|
try:
|
|
for argv in (["train", "--base", "B3", "--brief", self.brief()],
|
|
["fork", "--base", "B1", "--block", "2", "--brief", self.brief()],
|
|
["continue", "--job", j1, "--brief", self.brief()]):
|
|
out = self.cmd(*argv, code=2)
|
|
self.assertIn("inside a fork", out["error"])
|
|
self.assertEqual(self.cmd("list")["running"], [j1, j2])
|
|
code, out = self.wait(j1, timeout=1)
|
|
self.assertEqual(code, 3)
|
|
finally:
|
|
os.environ.pop(forks.CHILD_MARKER, None)
|
|
for j in (j1, j2):
|
|
self.until_running(j)
|
|
self.assertTrue(self.cmd("kill", "--job", j)["ok"])
|
|
code, out = self.wait(j1, j2)
|
|
self.assertEqual(code, 4)
|
|
self.assertEqual({x["job"]: x["paths"] for x in out["done"]}, {j1: ["a.py"], j2: ["b.py", "c.py"]})
|
|
# missing binary: the stored path is gone
|
|
moved = self.stub + ".moved"
|
|
os.replace(self.stub, moved)
|
|
try:
|
|
out = self.fork(code=2)
|
|
self.assertEqual(out["reason"], "binary_missing")
|
|
out = self.cmd("train", "--base", "B4", "--brief", self.brief(), code=2)
|
|
self.assertEqual(out["reason"], "binary_missing")
|
|
finally:
|
|
os.replace(moved, self.stub)
|
|
self.cmd("fork", "--base", "B1", "--block", "1", "--brief", os.path.join(self.tmp, "nope.md"), code=2)
|
|
|
|
|
|
def tearDownModule():
|
|
"""No supervisor or stub of this suite may survive it (checked by command line, never by name)."""
|
|
if not IS_WINDOWS or not STATE_DIRS:
|
|
return
|
|
ps = ("$p = 'forks' + '-test-'; Get-CimInstance Win32_Process | "
|
|
"Where-Object { $_.CommandLine -and $_.CommandLine.Contains($p) } | "
|
|
"ForEach-Object { '{0}|{1}' -f $_.ProcessId, $_.CommandLine }")
|
|
r = subprocess.run(["powershell", "-NoProfile", "-NonInteractive", "-Command", ps],
|
|
capture_output=True, timeout=120, creationflags=0x08000000)
|
|
lines = [x for x in r.stdout.decode("utf-8", "replace").splitlines() if x.strip()]
|
|
for x in lines:
|
|
with contextlib.suppress(ValueError, OSError):
|
|
forks.terminate(forks.pid_record(int(x.split("|", 1)[0])))
|
|
if lines:
|
|
raise AssertionError("leftover test processes (killed now): %s" % lines)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|