dirigence/reliquary
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

🐛 forks: Phase 1 review fixes (I1-I3, M1-M5, M9)

Browse Source 
- 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
This commit is contained in:
g4borg
2026-10-16 16:26:59 +02:00
parent 20b049e400
commit 7b9b7488de
2 changed files with 706 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/spire/skills/forks/forks.py
+303 -58
View File
@@ -51,9 +51,13 @@ SPAWN_METHOD = "cim-hidden" if IS_WINDOWS else "setsid"
CREATE_NO_WINDOW = 0x08000000 CREATE_NO_WINDOW = 0x08000000
_NOWIN = CREATE_NO_WINDOW if IS_WINDOWS else 0 _NOWIN = CREATE_NO_WINDOW if IS_WINDOWS else 0
# Registry file semantics (C3a). # Registry file semantics (C3a). A lock is stale when its holder (pid + creation time) is dead,
LOCK_STALE_S = 30.0 # or when it is older than LOCK_ABANDONED_S. Every critical section is a read-modify-write of a few
LOCK_WAIT_S = 45.0 # longer than LOCK_STALE_S, so a waiter can break a stale lock # small JSON files: milliseconds, at worst a few seconds of PermissionError retries (40 x 50 ms
# per file). 300 s is two orders of magnitude beyond that, so only a holder that cannot be judged
# (garbage content, an unverifiable pid) and is really gone ever reaches it.
LOCK_ABANDONED_S = 300.0
LOCK_WAIT_S = 45.0 # a dead holder's lock is broken at once; a live holder is waited for
LOCK_POLL_S = 0.05 LOCK_POLL_S = 0.05
RETRY_TRIES = 40 # PermissionError retries for replace and read (>= 20) RETRY_TRIES = 40 # PermissionError retries for replace and read (>= 20)
RETRY_SLEEP_S = 0.05 RETRY_SLEEP_S = 0.05
@@ -61,7 +65,8 @@ RETRY_SLEEP_S = 0.05
MODEL_RE = re.compile(r"^claude-[a-z0-9-]+(\[1m\])?$") MODEL_RE = re.compile(r"^claude-[a-z0-9-]+(\[1m\])?$")
ONE_M_SUFFIX = "[1m]" ONE_M_SUFFIX = "[1m]"
EFFORTS = ("low", "medium", "high", "xhigh", "max") EFFORTS = ("low", "medium", "high", "xhigh", "max")
PERMISSION_MODES = ("default", "acceptEdits", "dontAsk", "bypassPermissions", "auto") # `claude --help` 2.1.299 (checked by the Phase 1 review); `plan` is in the list but refused.
PERMISSION_MODES = ("acceptEdits", "auto", "bypassPermissions", "manual", "dontAsk", "plan")
REFUSED_PERMISSION_MODES = ("plan",) REFUSED_PERMISSION_MODES = ("plan",)
DEFAULT_TOOLS = "Bash,Read,Edit,Write,Glob,Grep,Skill" DEFAULT_TOOLS = "Bash,Read,Edit,Write,Glob,Grep,Skill"
@@ -106,7 +111,9 @@ SCRUB_DROP_PATTERNS = (re.compile(r"^ANTHROPIC_DEFAULT_.+_MODEL$"), re.compile(r
SCRUB_KEEP = frozenset({ SCRUB_KEEP = frozenset({
"CLAUDECODE", "CLAUDE_CODE_CHILD_SESSION", "AI_AGENT", "CLAUDE_PID", "CLAUDE_CODE_GIT_BASH_PATH", "CLAUDECODE", "CLAUDE_CODE_CHILD_SESSION", "AI_AGENT", "CLAUDE_PID", "CLAUDE_CODE_GIT_BASH_PATH",
"PATH", "HOME", "USERPROFILE", "PATH", "HOME", "USERPROFILE",
"ANTHROPIC_BASE_URL", # kept (gateways need it); `where` warns when it is not the default
}) })
DEFAULT_BASE_URL = "https://api.anthropic.com"
SCRUB_PINNED = {**ENV_PINNED_LAUNCH, "SPIRE_FORKS_CHILD": "1"} SCRUB_PINNED = {**ENV_PINNED_LAUNCH, "SPIRE_FORKS_CHILD": "1"}
SCRUB_PINNED_BINARY = "CLAUDE_CODE_EXECPATH" # set to the stored absolute binary path SCRUB_PINNED_BINARY = "CLAUDE_CODE_EXECPATH" # set to the stored absolute binary path
CHILD_MARKER = "SPIRE_FORKS_CHILD" # N3 CHILD_MARKER = "SPIRE_FORKS_CHILD" # N3
@@ -116,12 +123,14 @@ SCRUB_WATCH_PREFIXES = ("CLAUDE_", "ANTHROPIC_") # `where` lists unknown vars
USER_SETTINGS_PATH: str | None = None USER_SETTINGS_PATH: str | None = None
PHASE2_COMMANDS = ("rebase", "usage", "tick") PHASE2_COMMANDS = ("rebase", "usage", "tick")
CHILD_REFUSED = ("train", "fork", "continue", "rebase") # N3 (tick pings: Phase 2) # N3 (tick pings: Phase 2). A fork may only run where, list, usage and wait.
CHILD_REFUSED = ("train", "fork", "continue", "rebase", "kill", "expire", "init")
# Supervision and jobs (1.2). Module-level so tests can shorten the in-process ones. # Supervision and jobs (1.2). Module-level so tests can shorten the in-process ones.
HEARTBEAT_S = 10.0 # the supervisor writes heartbeat_at at least this often (<= 30 s) HEARTBEAT_S = 10.0 # the supervisor writes heartbeat_at at least this often (<= 30 s)
LOST_AFTER_S = 120.0 # heartbeat older than this and a dead supervisor -> `lost` LOST_AFTER_S = 120.0 # heartbeat older than this and a dead supervisor -> `lost`
SUP_TICK_S = 0.25 # supervisor main-loop period SUP_TICK_S = 0.25 # supervisor main-loop period
CLOCK_JUMP_S = 60.0 # a tick gap above this is a host suspend / clock jump (M3), not run time
ORPHAN_CHECK_S = 1.0 # how often the supervisor reads run.last_orchestrator_call_at (N1) ORPHAN_CHECK_S = 1.0 # how often the supervisor reads run.last_orchestrator_call_at (N1)
WAIT_POLL_S = 2.0 # `wait` poll period WAIT_POLL_S = 2.0 # `wait` poll period
WAIT_DEFAULT_S = 540 WAIT_DEFAULT_S = 540
@@ -362,7 +371,10 @@ def read_json(path: str):
class RegistryLock: class RegistryLock:
"""`registry.lock`: create-exclusive; a lock older than LOCK_STALE_S may be broken.""" """`registry.lock`: create-exclusive. A waiter breaks a lock only when it is stale (its holder's
pid + creation time is dead, or it is older than LOCK_ABANDONED_S), and only atomically: it
renames the lock aside and keeps it only if the renamed content is the one it judged stale;
otherwise it puts it back. So a waiter never deletes a fresh lock."""
def __init__(self, state: str, timeout: float = LOCK_WAIT_S): def __init__(self, state: str, timeout: float = LOCK_WAIT_S):
self.path = lock_path(state) self.path = lock_path(state)
@@ -390,20 +402,66 @@ class RegistryLock:
return True return True
def _read_lock(self): def _read_lock(self):
"""(raw bytes, age in seconds) of the current lock, or (None, None) if gone.""" """(raw bytes, age in seconds, holder dict or None) of the current lock, or
(None, None, None) if it is gone or unreadable right now."""
try: try:
raw = read_bytes_retry(self.path) raw = read_bytes_retry(self.path)
mtime = os.path.getmtime(self.path) mtime = os.path.getmtime(self.path)
except (FileNotFoundError, PermissionError): except (FileNotFoundError, PermissionError):
return None, None return None, None, None
at = None holder, at = None, None
try: try:
at = float(json.loads(raw.decode("utf-8")).get("at")) holder = json.loads(raw.decode("utf-8"))
at = float(holder.get("at"))
except (ValueError, TypeError, AttributeError): except (ValueError, TypeError, AttributeError):
pass pass
if at is None: if at is None:
at = mtime at = mtime
return raw, _now() - at return raw, _now() - at, holder if isinstance(holder, dict) else None
@staticmethod
def _is_stale(holder, age) -> bool:
if age is not None and age > LOCK_ABANDONED_S:
return True
if holder and holder.get("pid"):
try:
return not pid_alive(int(holder["pid"]), holder.get("ctime"))
except (TypeError, ValueError):
return False
return False
def _restore(self, aside: str) -> bool:
"""Put a lock renamed aside back, but never over a lock that appeared meanwhile."""
try:
if IS_WINDOWS:
_retry(lambda: os.rename(aside, self.path)) # Windows rename never overwrites
else:
os.link(aside, self.path) # fails if the path exists
os.remove(aside)
return True
except OSError:
return False
def _break(self, raw: bytes) -> bool:
"""Remove the lock atomically if it still is `raw` (the content judged stale)."""
aside = os.path.join(os.path.dirname(self.path),
"registry.%s.broken.lock" % uuid.uuid4().hex[:12])
try:
os.rename(self.path, aside) # atomic: we now own whatever lock was there
except OSError: # gone already, or busy: judge again later
return False
try:
got = read_bytes_retry(aside)
except OSError:
got = None
if got == raw:
with contextlib.suppress(OSError):
_retry(lambda: os.remove(aside))
return True
if not self._restore(aside):
_diag("registry lock changed while being broken and could not be put back (left at %s)"
% aside)
return False
def acquire(self) -> "RegistryLock": def acquire(self) -> "RegistryLock":
deadline = time.monotonic() + self.timeout deadline = time.monotonic() + self.timeout
@@ -411,16 +469,12 @@ class RegistryLock:
if self._try_create(): if self._try_create():
self.held = True self.held = True
return self return self
raw, age = self._read_lock() raw, age, holder = self._read_lock()
if raw is not None and age is not None and age > LOCK_STALE_S: if raw is not None and self._is_stale(holder, age) and self._break(raw):
# Break it only if it is still the same lock we judged stale.
again, _ = self._read_lock()
if again == raw:
with contextlib.suppress(FileNotFoundError, PermissionError):
os.remove(self.path)
self.broke_stale = True self.broke_stale = True
_diag("broke stale registry lock (age %.1f s)" % age) _diag("broke stale registry lock (age %.1f s, holder pid %s)"
continue % (age, (holder or {}).get("pid")))
continue # the path is free: try to create at once
if time.monotonic() >= deadline: if time.monotonic() >= deadline:
raise ForksError("registry locked", lock=self.path, waited_s=self.timeout) raise ForksError("registry locked", lock=self.path, waited_s=self.timeout)
time.sleep(LOCK_POLL_S) time.sleep(LOCK_POLL_S)
@@ -472,14 +526,37 @@ def write_registry(state: str, reg: dict) -> int:
return write_json_atomic(registry_path(state), reg) return write_json_atomic(registry_path(state), reg)
def ledger_append(state: str, line: dict) -> None: def _ledger_jobs(state: str) -> tuple[set, bool]:
"""Append one JSON line. Caller must hold RegistryLock(state) (C3c).""" """(job ids that already have a ledger line, whether the file ends without a newline)."""
data = (json.dumps(line, sort_keys=True) + "\n").encode("utf-8") try:
data = read_bytes_retry(ledger_path(state))
except FileNotFoundError:
return set(), False
jobs = set()
for ln in data.splitlines():
try:
obj = json.loads(ln.decode("utf-8", "replace"))
except ValueError:
continue
if isinstance(obj, dict) and obj.get("job"):
jobs.add(obj["job"])
return jobs, bool(data) and not data.endswith(b"\n")
def ledger_append(state: str, line: dict) -> bool:
"""Append one JSON line, at most one per job (M1: skipped if the job already has one).
Caller must hold RegistryLock(state) (C3c). Returns True if a line was written."""
have, torn = _ledger_jobs(state)
if line.get("job") and line["job"] in have:
_diag("ledger already has a line for %s; not appending another" % line["job"])
return False
data = (("\n" if torn else "") + json.dumps(line, sort_keys=True) + "\n").encode("utf-8")
def _append(): def _append():
with open(ledger_path(state), "ab") as f: with open(ledger_path(state), "ab") as f:
f.write(data) f.write(data)
_retry(_append) _retry(_append)
return True
# -------------------------------------------------------------------------------------- # --------------------------------------------------------------------------------------
@@ -545,6 +622,24 @@ def proc_ctime(pid: int) -> int | None:
return None return None
_FILETIME_EPOCH = 116444736000000000 # 1970-01-01 in FILETIME ticks (100 ns since 1601)
def epoch_to_ctime(t: float | None) -> int | None:
"""Epoch seconds in the units of `proc_ctime` (Windows FILETIME ticks; POSIX clock ticks since
boot), or None when the conversion is unknown here."""
if t is None:
return None
if IS_WINDOWS:
return int(float(t) * 10_000_000) + _FILETIME_EPOCH
try:
with open("/proc/stat", "rb") as f:
btime = next(int(ln.split()[1]) for ln in f.read().splitlines() if ln.startswith(b"btime "))
return int((float(t) - btime) * os.sysconf("SC_CLK_TCK"))
except (OSError, ValueError, StopIteration, IndexError, AttributeError):
return None
def pid_record(pid: int) -> dict: def pid_record(pid: int) -> dict:
"""{"pid", "ctime"}: the pair every stored pid is compared by.""" """{"pid", "ctime"}: the pair every stored pid is compared by."""
return {"pid": int(pid), "ctime": proc_ctime(pid)} return {"pid": int(pid), "ctime": proc_ctime(pid)}
@@ -619,10 +714,19 @@ def process_table() -> list[tuple[int, int, str]]:
return out return out
def process_tree(rec: dict | None) -> list[dict]: ALIVE_UNTIL_SLACK_S = 1.0 # job-file times are ISO with whole seconds: allow the truncated part
def process_tree(rec: dict | None, alive_until: float | None = None) -> list[dict]:
"""`rec` (if alive) and all its descendants, each {pid, ctime, ppid, name}. A process counts as """`rec` (if alive) and all its descendants, each {pid, ctime, ppid, name}. A process counts as
a child only if it was created at or after its parent (guards against pid reuse). Descendants a child only if it was created at or after its parent (guards against pid reuse).
of a dead root are still found: their parent pid stays recorded."""
Dead root (I2): its pid may have been reused by a later process whose own children now carry
the same stale parent pid. Its direct children are therefore taken only if they were created
after the root's recorded creation time AND before `alive_until` (epoch seconds: the last
time the job was known alive, e.g. the job file's `heartbeat_at` / `ended_at`). Without a
recorded creation time or `alive_until`, a dead root claims nothing. Below a claimed (live,
verified) child the normal rule applies."""
if not rec or not rec.get("pid"): if not rec or not rec.get("pid"):
return [] return []
kids, names = {}, {} kids, names = {}, {}
@@ -631,12 +735,21 @@ def process_tree(rec: dict | None) -> list[dict]:
if pid != ppid: if pid != ppid:
kids.setdefault(ppid, []).append(pid) kids.setdefault(ppid, []).append(pid)
root = int(rec["pid"]) root = int(rec["pid"])
root_ct = rec.get("ctime")
out = [] out = []
if record_alive(rec): if record_alive(rec):
out.append({"pid": root, "ctime": rec.get("ctime"), "ppid": None, "name": names.get(root)}) out.append({"pid": root, "ctime": root_ct, "ppid": None, "name": names.get(root)})
stack, seen = [(root, rec.get("ctime"))], {root} dead_bounds = None
else:
upper = epoch_to_ctime(None if alive_until is None else alive_until + ALIVE_UNTIL_SLACK_S)
if root_ct is None or upper is None:
return []
dead_bounds = (int(root_ct), upper)
stack, seen = [(root, root_ct)], {root}
while stack: while stack:
ppid, pct = stack.pop() ppid, pct = stack.pop()
is_dead_root = ppid == root and dead_bounds is not None
if not is_dead_root:
now_ct = proc_ctime(ppid) now_ct = proc_ctime(ppid)
if pct is not None and now_ct is not None and now_ct != pct: if pct is not None and now_ct is not None and now_ct != pct:
continue # this pid now belongs to another process: its children are not ours continue # this pid now belongs to another process: its children are not ours
@@ -644,7 +757,11 @@ def process_tree(rec: dict | None) -> list[dict]:
if c in seen: if c in seen:
continue continue
ct = proc_ctime(c) ct = proc_ctime(c)
if pct is not None and ct is not None and ct < pct: if is_dead_root:
lo, hi = dead_bounds
if ct is None or not (lo < ct <= hi):
continue # created before the job, or after it was last known alive: not ours
elif pct is not None and ct is not None and ct < pct:
continue continue
seen.add(c) seen.add(c)
out.append({"pid": c, "ctime": ct, "ppid": ppid, "name": names.get(c)}) out.append({"pid": c, "ctime": ct, "ppid": ppid, "name": names.get(c)})
@@ -677,14 +794,15 @@ def _taskkill() -> str:
return p if os.path.isfile(p) else "taskkill" return p if os.path.isfile(p) else "taskkill"
def kill_tree(rec: dict | None, extra=()) -> dict: def kill_tree(rec: dict | None, extra=(), alive_until: float | None = None) -> dict:
"""Snapshot the tree of `rec` (plus the trees of `extra` records), kill it (`taskkill /T /F` on """Snapshot the tree of `rec` (plus the trees of `extra` records), kill it (`taskkill /T /F` on
Windows, the process group on POSIX), sweep every snapshot pid still alive, and verify. Windows, the process group on POSIX), sweep every snapshot pid still alive, and verify.
`alive_until` bounds the descendants of a dead root (see `process_tree`).
Returns {"snapshot", "taskkill_rc", "survivors"}; pids are always checked with their ctime.""" Returns {"snapshot", "taskkill_rc", "survivors"}; pids are always checked with their ctime."""
roots = [r for r in [rec, *extra] if r and r.get("pid")] roots = [r for r in [rec, *extra] if r and r.get("pid")]
snap, seen = [], set() snap, seen = [], set()
for r in roots: for r in roots:
for p in process_tree(r): for p in process_tree(r, alive_until):
key = (p["pid"], p["ctime"]) key = (p["pid"], p["ctime"])
if key not in seen: if key not in seen:
seen.add(key) seen.add(key)
@@ -1079,6 +1197,42 @@ def usage_from_result(result: dict, before_mu: dict, cost_before: float) -> dict
"cost_delta": cost_delta, "total": total, "model_usage": after} "cost_delta": cost_delta, "total": total, "model_usage": after}
_API_ERROR_TEXT = ( # checked in this order, after the status code
("prompt_too_long", re.compile(r"prompt is too long|prompt too long|prompt_too_long", re.I)),
("usage_limit", re.compile(r"usage limit|rate[ _-]?limit|hit your limit|api error:?\s*429", re.I)),
("overloaded", re.compile(r"overloaded|api error:?\s*529", re.I)),
("api_error", re.compile(r"api error", re.I)),
)
def api_error_reason(res: dict | None) -> tuple[str, str] | None:
"""(reason, detail) for a result that carries an API error, else None (I1). `reason` is
`usage_limit` (429 / rate or usage limit), `overloaded` (529), `prompt_too_long`, or
`api_error` (any other API error status); `detail` holds the status and a short text."""
if not res:
return None
status = res.get("api_error_status")
try:
status = int(status) if status not in (None, "") else None
except (TypeError, ValueError):
status = None
if not res.get("is_error") and status is None:
return None
parts = [res.get("result"), res.get("error"), res.get("errors")]
text = " | ".join(p if isinstance(p, str) else json.dumps(p) for p in parts if p)
detail = "api_error_status %s: %s" % (status, text[:200] if text else "(no text)")
if status == 429:
return "usage_limit", detail
if status == 529:
return "overloaded", detail
for reason, rx in _API_ERROR_TEXT:
if rx.search(text):
return reason, detail
if status is not None:
return "api_error", detail
return None
def cache_verdict(first_usage: dict | None, base_context) -> str | None: def cache_verdict(first_usage: dict | None, base_context) -> str | None:
"""C5: first assistant event's cache_read against the base's context tokens.""" """C5: first assistant event's cache_read against the base's context tokens."""
if first_usage is None: if first_usage is None:
@@ -1197,13 +1351,19 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint:
models_used = u["models_used"] if u else [] models_used = u["models_used"] if u else []
if hint == "exited": if hint == "exited":
requested = entry.get("model_requested") requested = entry.get("model_requested")
if ss and (ss.refused or (res and requested not in models_used)): # Refused = the restriction warning, or a substitution: some other model billed and the
# requested one did not. An empty delta (an error before the model answered) is not one.
substituted = bool(models_used) and requested not in models_used
api = api_error_reason(res)
if ss and (ss.refused or (res and substituted)):
status, reason = "failed", "model_refused" status, reason = "failed", "model_refused"
detail = ss.refused or "models_used %s, requested %s" % (models_used, requested) detail = ss.refused or "models_used %s, requested %s" % (models_used, requested)
elif res is None: elif res is None:
status = "failed" status = "failed"
reason = "exit_code" if exit_code else "no_result" reason = "exit_code" if exit_code else "no_result"
detail = "exit %s, no result event" % exit_code detail = "exit %s, no result event" % exit_code
elif api:
status, (reason, detail) = "failed", api
elif res.get("is_error"): elif res.get("is_error"):
status, reason, detail = "failed", "is_error", str(res.get("subtype")) status, reason, detail = "failed", "is_error", str(res.get("subtype"))
elif exit_code: elif exit_code:
@@ -1274,7 +1434,14 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint:
holder["session_cost_cum_usd"] = u["total"] holder["session_cost_cum_usd"] = u["total"]
holder["session_model_usage_cum"] = u["model_usage"] holder["session_model_usage_cum"] = u["model_usage"]
if base is not None: if base is not None:
hit_t = (ss.last_assistant_at if kind == "train" else ss.first_at) if ss else None # Design 5: only a request that reads exactly the base prefix refreshes it: a fork's first
# request (and the training's end, which writes it). A continue reads the fork's own
# longer prefix, so it leaves the base's last_hit alone (I3).
hit_t = None
if ss and kind == "train":
hit_t = ss.last_assistant_at
elif ss and kind == "fork":
hit_t = ss.first_at
if hit_t and (_ts(base.get("last_hit")) or 0) < hit_t: if hit_t and (_ts(base.get("last_hit")) or 0) < hit_t:
base["last_hit"] = _iso(hit_t) base["last_hit"] = _iso(hit_t)
if kind == "train" and base.get("train_job") == entry["job"]: if kind == "train" and base.get("train_job") == entry["job"]:
@@ -1292,8 +1459,10 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint:
def finalize_job(state: str, job: str, *, ss: StreamState | None, hint: str, flags=(), reason=None, def finalize_job(state: str, job: str, *, ss: StreamState | None, hint: str, flags=(), reason=None,
detail=None, exit_code=None, wall_s=None, base_commit=None) -> dict | None: detail=None, exit_code=None, wall_s=None, base_commit=None) -> dict | None:
"""Write `results/<job>.json`, the ledger line and the registry update, exactly once, under the """Write `results/<job>.json`, the registry update and the ledger line, exactly once, under the
lock (C3c). Returns the result, or None when the job was already final.""" lock (C3c). The registry status is written first: it is the once-only guard, so a finaliser
killed after it leaves no second finaliser to run (M1); the ledger append also skips a job
that already has a line. Returns the result, or None when the job was already final."""
with RegistryLock(state): with RegistryLock(state):
reg = read_registry(state) reg = read_registry(state)
entry = (reg or {}).get("jobs", {}).get(job) entry = (reg or {}).get("jobs", {}).get(job)
@@ -1302,13 +1471,13 @@ def finalize_job(state: str, job: str, *, ss: StreamState | None, hint: str, fla
result, ledger = _conclude(state, reg, entry, ss, hint, flags, reason, detail, exit_code, result, ledger = _conclude(state, reg, entry, ss, hint, flags, reason, detail, exit_code,
wall_s, base_commit, _now()) wall_s, base_commit, _now())
write_json_atomic(result_file(state, job), result) write_json_atomic(result_file(state, job), result)
ledger_append(state, ledger)
write_registry(state, reg) write_registry(state, reg)
ledger_append(state, ledger)
return result return result
def bump_last_hit(state: str, base_name: str, t: float) -> None: def bump_last_hit(state: str, base_name: str, t: float) -> None:
"""Record a hit on the base prefix (first assistant event of a fork/continue) as it happens.""" """Record a hit on the base prefix (first assistant event of a fork) as it happens."""
with RegistryLock(state): with RegistryLock(state):
reg = read_registry(state) reg = read_registry(state)
base = (reg or {}).get("bases", {}).get(base_name) base = (reg or {}).get("bases", {}).get(base_name)
@@ -1318,13 +1487,28 @@ def bump_last_hit(state: str, base_name: str, t: float) -> None:
write_registry(state, reg) write_registry(state, reg)
def in_child() -> bool:
"""N3: this call comes from inside a fork (`SPIRE_FORKS_CHILD` is set)."""
return bool(os.environ.get(CHILD_MARKER))
def stamp_orchestrator(reg: dict, t: float | None = None) -> None:
"""N1: set `run.last_orchestrator_call_at` in `reg` (caller holds the lock and writes it).
A fork's own calls never stamp, so a fork cannot keep its own watchdog alive (M4)."""
if in_child():
return
reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso(t)
def touch_orchestrator(state: str) -> None: def touch_orchestrator(state: str) -> None:
"""N1: record that the orchestrator called us (wait, fork, continue, train, list).""" """N1: record that the orchestrator called us (wait, fork, continue, train, list)."""
if in_child():
return
with RegistryLock(state): with RegistryLock(state):
reg = read_registry(state) reg = read_registry(state)
if reg is None: if reg is None:
return return
reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso() stamp_orchestrator(reg)
write_registry(state, reg) write_registry(state, reg)
@@ -1335,6 +1519,13 @@ def _job_runtime(state: str, job: str) -> dict:
return {} return {}
def last_alive_at(rt: dict) -> float | None:
"""The last time a job's supervisor was known alive: the later of its job file's
`heartbeat_at` and `ended_at` (bounds a dead root's descendants, I2)."""
ts = [t for t in (_ts(rt.get("heartbeat_at")), _ts(rt.get("ended_at"))) if t is not None]
return max(ts) if ts else None
def reap_lost(state: str) -> list[str]: def reap_lost(state: str) -> list[str]:
"""Decision 7 / A19: a running job whose heartbeat is older than LOST_AFTER_S and whose """Decision 7 / A19: a running job whose heartbeat is older than LOST_AFTER_S and whose
supervisor (pid + creation time) is dead becomes `lost`. Its claude tree, if any is left, supervisor (pid + creation time) is dead becomes `lost`. Its claude tree, if any is left,
@@ -1353,7 +1544,7 @@ def reap_lost(state: str) -> list[str]:
if sup and record_alive(sup): if sup and record_alive(sup):
continue continue
if rt.get("claude"): if rt.get("claude"):
kill_tree(rt["claude"]) kill_tree(rt["claude"], alive_until=last_alive_at(rt))
start = _ts(rt.get("started_at")) or _ts(e.get("launched_at")) start = _ts(rt.get("started_at")) or _ts(e.get("launched_at"))
res = finalize_job(state, job, ss=StreamState.from_file(raw_file(state, job)), hint="lost", res = finalize_job(state, job, ss=StreamState.from_file(raw_file(state, job)), hint="lost",
flags=rt.get("flags") or [], reason="supervisor_dead", flags=rt.get("flags") or [], reason="supervisor_dead",
@@ -1380,7 +1571,7 @@ def kill_job(state: str, job: str, by: str) -> dict:
sup = rt.get("supervisor") or entry.get("supervisor") sup = rt.get("supervisor") or entry.get("supervisor")
cl = rt.get("claude") cl = rt.get("claude")
sup_alive = record_alive(sup) sup_alive = record_alive(sup)
k = kill_tree(sup, extra=[cl] if cl else []) k = kill_tree(sup, extra=[cl] if cl else [], alive_until=last_alive_at(rt))
flags = list(rt.get("flags") or []) flags = list(rt.get("flags") or [])
if by == "expire": if by == "expire":
flags.append("expired") flags.append("expired")
@@ -1584,6 +1775,12 @@ def cmd_where(args) -> tuple[dict, int]:
except ForksError as e: except ForksError as e:
binary["error"] = e.payload["error"] binary["error"] = e.payload["error"]
reasons.append("%s: %s" % (e.payload["error"], path)) reasons.append("%s: %s" % (e.payload["error"], path))
warnings = []
base_url = os.environ.get("ANTHROPIC_BASE_URL")
if base_url is not None and base_url.strip().rstrip("/").lower() != DEFAULT_BASE_URL:
warnings.append("ANTHROPIC_BASE_URL is %r, not the default %s; it is passed to every "
"base and fork, so they authenticate and bill through that endpoint"
% (base_url, DEFAULT_BASE_URL))
out = { out = {
"script": script_path(), "script": script_path(),
"python": py, "python": py,
@@ -1593,6 +1790,7 @@ def cmd_where(args) -> tuple[dict, int]:
"spawn": SPAWN_METHOD, "spawn": SPAWN_METHOD,
"scrub": scrub_report(), "scrub": scrub_report(),
"platform": sys.platform, "platform": sys.platform,
"warnings": warnings,
"ok": not reasons, "ok": not reasons,
} }
if reasons: if reasons:
@@ -1625,7 +1823,8 @@ def build_init_config(args, state: str) -> dict:
raise ForksError("permission mode refused", permission_mode=mode, raise ForksError("permission mode refused", permission_mode=mode,
detail="plan mode swaps models; use dontAsk") detail="plan mode swaps models; use dontAsk")
if mode not in PERMISSION_MODES: if mode not in PERMISSION_MODES:
raise ForksError("unknown permission mode", permission_mode=mode, allowed=list(PERMISSION_MODES)) raise ForksError("unknown permission mode", permission_mode=mode,
allowed=[m for m in PERMISSION_MODES if m not in REFUSED_PERMISSION_MODES])
tools = _tool_list([args.tools]) tools = _tool_list([args.tools])
if not tools: if not tools:
raise ForksError("--tools is empty") raise ForksError("--tools is empty")
@@ -1905,7 +2104,7 @@ def cmd_train(args) -> tuple[dict, int]:
head = _git_head(la["cwd"]) head = _git_head(la["cwd"])
with RegistryLock(state): with RegistryLock(state):
reg = read_registry(state) reg = read_registry(state)
reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso() stamp_orchestrator(reg)
old = reg["bases"].get(name) old = reg["bases"].get(name)
if old and old.get("status") != "failed": if old and old.get("status") != "failed":
raise ForksError("base exists", base=name, status=old.get("status")) raise ForksError("base exists", base=name, status=old.get("status"))
@@ -1954,7 +2153,7 @@ def cmd_fork(args) -> tuple[dict, int]:
reap_lost(state) reap_lost(state)
with RegistryLock(state): with RegistryLock(state):
reg = read_registry(state) reg = read_registry(state)
reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso() stamp_orchestrator(reg)
base = reg["bases"][args.base] base = reg["bases"][args.base]
if base.get("status") != "active" and not args.force: if base.get("status") != "active" and not args.force:
raise ForksError("base is not active (use --force to fork it anyway)", base=args.base, raise ForksError("base is not active (use --force to fork it anyway)", base=args.base,
@@ -2023,7 +2222,7 @@ def cmd_continue(args) -> tuple[dict, int]:
reap_lost(state) reap_lost(state)
with RegistryLock(state): with RegistryLock(state):
reg = read_registry(state) reg = read_registry(state)
reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso() stamp_orchestrator(reg)
fj = reg["jobs"][args.job] fj = reg["jobs"][args.job]
if busy(reg): if busy(reg):
raise ForksError("the fork is still running", job=args.job, running=busy(reg)) raise ForksError("the fork is still running", job=args.job, running=busy(reg))
@@ -2053,6 +2252,42 @@ def cmd_continue(args) -> tuple[dict, int]:
# --- the supervisor ------------------------------------------------------------------- # --- the supervisor -------------------------------------------------------------------
class SupClock:
"""The supervisor's run time with host suspends cut out (M3). Each loop tick reports the
wall clock; a gap above `jump_s` between ticks (the tick is ~SUP_TICK_S) is a suspend or a
clock jump: the gap minus one tick is left out of `elapsed`, and the orphan watchdog is
re-armed at the jump (one more `orphan_after` from now)."""
def __init__(self, t0: float, tick_s: float | None = None, jump_s: float | None = None):
self.t0 = self.prev = t0
self.tick_s = SUP_TICK_S if tick_s is None else tick_s
self.jump_s = CLOCK_JUMP_S if jump_s is None else jump_s
self.excluded_s = 0.0
self.rearmed_at: float | None = None
self.jumps: list[float] = []
def tick(self, now: float) -> float | None:
"""Record a tick; returns the excluded seconds if this tick follows a jump, else None."""
gap, self.prev = now - self.prev, now
if gap <= self.jump_s:
return None
skipped = gap - self.tick_s
self.excluded_s += skipped
self.rearmed_at = now
self.jumps.append(skipped)
return skipped
def elapsed(self, now: float) -> float:
"""Seconds of run time since t0 (what `max_minutes` is measured against)."""
return now - self.t0 - self.excluded_s
def orphan_age(self, now: float, last_call: float) -> float:
"""Seconds the orphan watchdog counts: since the later of the last orchestrator call and
the last re-arm."""
ref = last_call if self.rearmed_at is None else max(last_call, self.rearmed_at)
return now - ref
class Supervisor: class Supervisor:
"""`_supervise`: runs one `claude` call, parses its stream, watches it, writes the result. """`_supervise`: runs one `claude` call, parses its stream, watches it, writes the result.
The only writer of `jobs/<job>.json` (C3b).""" The only writer of `jobs/<job>.json` (C3b)."""
@@ -2141,20 +2376,28 @@ class Supervisor:
stall_s = float(self.cfg.get("stall_after_s") or CONFIG_DEFAULTS["stall_after_s"]) stall_s = float(self.cfg.get("stall_after_s") or CONFIG_DEFAULTS["stall_after_s"])
orphan_s = float(self.cfg.get("orphan_after") or CONFIG_DEFAULTS["orphan_after"]) orphan_s = float(self.cfg.get("orphan_after") or CONFIG_DEFAULTS["orphan_after"])
hint, reason, detail, flags_extra = "exited", None, None, [] hint, reason, detail, flags_extra = "exited", None, None, []
hit_written = e["kind"] == "train" hit_written = e["kind"] != "fork" # only a fork's first request refreshes (I3)
clock = SupClock(t0)
last_orphan_check = 0.0 last_orphan_check = 0.0
last_turns = 0 last_turns = 0
while p.poll() is None: while p.poll() is None:
now = _now() now = _now()
skipped = clock.tick(now)
if skipped is not None:
_diag("clock jump: %.0f s between supervisor ticks at %s (host suspend?); not counted "
"toward max_minutes, orphan watchdog re-armed" % (skipped, _iso(now)))
if "clock_jump" not in self.flags:
self.flags.append("clock_jump")
self.write_rt()
with self.mu: with self.mu:
refused, first_at, turns = self.ss.refused, self.ss.first_at, self.ss.turns refused, first_at, turns = self.ss.refused, self.ss.first_at, self.ss.turns
quiet = now - self.last_output quiet = now - self.last_output
if refused: if refused:
kill_tree(crec) kill_tree(crec, alive_until=time.time())
break # decided as model_refused from the stream break # decided as model_refused from the stream
if now - t0 > max_s: if clock.elapsed(now) > max_s:
kill_tree(crec) kill_tree(crec, alive_until=time.time())
hint, detail = "timeout", "wall time over max_minutes (%.0f s)" % max_s hint, detail = "timeout", "run time over max_minutes (%.0f s)" % max_s
break break
if quiet > stall_s and "stalled" not in self.flags: if quiet > stall_s and "stalled" not in self.flags:
self.flags.append("stalled") self.flags.append("stalled")
@@ -2170,17 +2413,19 @@ class Supervisor:
last = _ts(((reg or {}).get("run") or {}).get("last_orchestrator_call_at")) last = _ts(((reg or {}).get("run") or {}).get("last_orchestrator_call_at"))
except (ForksError, OSError): except (ForksError, OSError):
last = None last = None
if last is not None and now - last > orphan_s: if last is not None and clock.orphan_age(now, last) > orphan_s:
kill_tree(crec) kill_tree(crec, alive_until=time.time())
hint, reason, flags_extra = "killed", "orphaned", ["orphaned"] hint, reason, flags_extra = "killed", "orphaned", ["orphaned"]
detail = "no orchestrator call for %.0f s (orphan_after %.0f s)" % (now - last, orphan_s) detail = "no orchestrator call for %.0f s (orphan_after %.0f s)" % (
clock.orphan_age(now, last), orphan_s)
break break
if now - self._last_write >= HEARTBEAT_S or (turns != last_turns and now - self._last_write >= 1.0): if now - self._last_write >= HEARTBEAT_S or (turns != last_turns and now - self._last_write >= 1.0):
last_turns = turns last_turns = turns
self.write_rt() self.write_rt()
time.sleep(SUP_TICK_S) time.sleep(SUP_TICK_S)
exit_code = p.wait() exit_code = p.wait()
sweep = kill_tree(crec) # leftovers of claude (backgrounded commands) # leftovers of claude (backgrounded commands); claude exited just now
sweep = kill_tree(crec, alive_until=time.time())
if sweep["snapshot"]: if sweep["snapshot"]:
_diag("swept %d leftover process(es) of claude: %s" % ( _diag("swept %d leftover process(es) of claude: %s" % (
len(sweep["snapshot"]), [(x["pid"], x["name"]) for x in sweep["snapshot"]])) len(sweep["snapshot"]), [(x["pid"], x["name"]) for x in sweep["snapshot"]]))
@@ -2215,7 +2460,7 @@ def cmd_supervise(args) -> tuple[dict, int]:
with contextlib.suppress(Exception): with contextlib.suppress(Exception):
rt = _job_runtime(state, args.job) rt = _job_runtime(state, args.job)
if rt.get("claude"): if rt.get("claude"):
kill_tree(rt["claude"]) kill_tree(rt["claude"], alive_until=time.time())
finalize_job(state, args.job, ss=StreamState.from_file(raw_file(state, args.job)), finalize_job(state, args.job, ss=StreamState.from_file(raw_file(state, args.job)),
hint="failed", reason="supervisor_error", detail="%s: %s" % (type(ex).__name__, ex)) hint="failed", reason="supervisor_error", detail="%s: %s" % (type(ex).__name__, ex))
raise raise
@@ -2278,7 +2523,7 @@ def cmd_wait(args) -> tuple[dict, int]:
if met: if met:
for j in finished: for j in finished:
reg["jobs"][j]["settled_at"] = _iso(now) reg["jobs"][j]["settled_at"] = _iso(now)
reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso(now) stamp_orchestrator(reg, now)
write_registry(state, reg) write_registry(state, reg)
statuses = {j: reg["jobs"][j].get("status") for j in finished} statuses = {j: reg["jobs"][j].get("status") for j in finished}
entries = {j: dict(reg["jobs"][j]) for j in running} entries = {j: dict(reg["jobs"][j]) for j in running}
@@ -2379,9 +2624,9 @@ def run(argv) -> tuple[dict, int]:
return {"help": h.text, "commands": [c for c in COMMANDS if not c.startswith("_")]}, EXIT_OK return {"help": h.text, "commands": [c for c in COMMANDS if not c.startswith("_")]}, EXIT_OK
if unknown and args.command not in PHASE2_COMMANDS: if unknown and args.command not in PHASE2_COMMANDS:
parser.error("unrecognized arguments: %s" % " ".join(unknown)) parser.error("unrecognized arguments: %s" % " ".join(unknown))
if args.command in CHILD_REFUSED and os.environ.get(CHILD_MARKER): if args.command in CHILD_REFUSED and in_child():
raise ForksError("launching commands are refused inside a fork (%s is set)" % CHILD_MARKER, raise ForksError("`%s` is refused inside a fork (%s is set); a fork may only run where, list, "
command=args.command) "usage and wait" % (args.command, CHILD_MARKER), command=args.command)
return HANDLERS[args.command](args) return HANDLERS[args.command](args)
plugins/spire/skills/forks/tests/test_forks.py
+399 -10
View File
@@ -175,7 +175,9 @@ class ScrubTests(Base):
def test_unknown(self): def test_unknown(self):
host = {"CLAUDECODE": "1", "CLAUDE_EFFORT": "x", "CLAUDE_CODE_NEW_THING": "1", host = {"CLAUDECODE": "1", "CLAUDE_EFFORT": "x", "CLAUDE_CODE_NEW_THING": "1",
"ANTHROPIC_BASE_URL": "u", "OTHER": "1", "CLAUDE_CODE_EXECPATH": "p"} "ANTHROPIC_BASE_URL": "u", "OTHER": "1", "CLAUDE_CODE_EXECPATH": "p"}
self.assertEqual(forks.scrub_unknown(host), ["ANTHROPIC_BASE_URL", "CLAUDE_CODE_NEW_THING"]) # 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")
# -------------------------------------------------------------------------------------- # --------------------------------------------------------------------------------------
@@ -214,33 +216,56 @@ class LockTests(Base):
r = subprocess.run([sys.executable, "-c", code, FORKS_PY, self.state], capture_output=True) r = subprocess.run([sys.executable, "-c", code, FORKS_PY, self.state], capture_output=True)
self.assertEqual(r.returncode, 7, r.stderr) self.assertEqual(r.returncode, 7, r.stderr)
def test_stale_lock_broken(self): 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: 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) 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: with forks.RegistryLock(self.state, timeout=2) as lk:
self.assertTrue(lk.broke_stale) self.assertTrue(lk.broke_stale)
self.assertLess(time.monotonic() - t0, 1.5)
with open(forks.lock_path(self.state)) as f: with open(forks.lock_path(self.state)) as f:
self.assertEqual(json.load(f)["token"], lk.token) 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): def test_garbage_stale_lock_broken_by_mtime(self):
p = forks.lock_path(self.state) p = forks.lock_path(self.state)
with open(p, "w") as f: with open(p, "w") as f:
f.write("") f.write("")
old = time.time() - (forks.LOCK_STALE_S + 5) old = time.time() - (forks.LOCK_ABANDONED_S + 5)
os.utime(p, (old, old)) os.utime(p, (old, old))
with forks.RegistryLock(self.state, timeout=2) as lk: with forks.RegistryLock(self.state, timeout=2) as lk:
self.assertTrue(lk.broke_stale) self.assertTrue(lk.broke_stale)
def test_fresh_lock_not_broken(self): 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: with open(forks.lock_path(self.state), "w") as f:
json.dump({"pid": 1, "at": time.time() - 5, "token": "other"}, f) f.write("")
with self.assertRaises(forks.ForksError): with self.assertRaises(forks.ForksError):
forks.RegistryLock(self.state, timeout=0.3).acquire() forks.RegistryLock(self.state, timeout=0.3).acquire()
def test_stale_under_injected_clock_and_safe_release(self): def test_stale_under_injected_clock_and_safe_release(self):
l1 = forks.RegistryLock(self.state).acquire() l1 = forks.RegistryLock(self.state).acquire()
real = time.time() real = time.time()
with mock.patch.object(forks, "_now", lambda: real + forks.LOCK_STALE_S + 1): with mock.patch.object(forks, "_now", lambda: real + forks.LOCK_ABANDONED_S + 1):
l2 = forks.RegistryLock(self.state, timeout=2).acquire() l2 = forks.RegistryLock(self.state, timeout=2).acquire()
self.assertTrue(l2.broke_stale) self.assertTrue(l2.broke_stale)
l1.release() # must not remove l2's lock l1.release() # must not remove l2's lock
@@ -248,6 +273,75 @@ class LockTests(Base):
l2.release() l2.release()
self.assertFalse(os.path.exists(forks.lock_path(self.state))) 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): class AtomicTests(Base):
@@ -298,6 +392,29 @@ class AtomicTests(Base):
lines = f.read().decode().splitlines() lines = f.read().decode().splitlines()
self.assertEqual([json.loads(x)["job"] for x in lines], ["j0001", "j0002"]) 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): class PidTests(Base):
@@ -318,6 +435,82 @@ class PidTests(Base):
self.assertTrue(os.path.isfile(forks.supervisor_interpreter())) 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): class InitTests(Base):
def setUp(self): def setUp(self):
@@ -478,6 +671,19 @@ class InitTests(Base):
self.assert_refused_no_state("--permission-mode", "yolo") self.assert_refused_no_state("--permission-mode", "yolo")
self.assert_refused_no_state("--effort", "turbo") 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): def test_missing_binary_refused(self):
out = self.assert_refused_no_state("--binary", os.path.join(self.tmp, "nope.exe")) out = self.assert_refused_no_state("--binary", os.path.join(self.tmp, "nope.exe"))
self.assertEqual(out["error"], "claude binary not found") self.assertEqual(out["error"], "claude binary not found")
@@ -620,6 +826,64 @@ class CliTests(Base):
self.assert_one_json_line(raw) self.assert_one_json_line(raw)
self.assertEqual(out["error"], "internal error") 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): def test_where_in_process_missing_binary(self):
code, out, raw = call(["where", "--binary", os.path.join(self.tmp, "nope.exe")]) code, out, raw = call(["where", "--binary", os.path.join(self.tmp, "nope.exe")])
self.assertEqual(code, 2) self.assertEqual(code, 2)
@@ -764,6 +1028,10 @@ def main():
if "silent" in d: if "silent" in d:
time.sleep(float(d["silent"])) time.sleep(float(d["silent"]))
turns = int(d.get("turns", "3")) 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) ctx = int(prior.get("context") or 0)
tot = {"in": 0, "out": 0, "cw": 0, "cr": 0} tot = {"in": 0, "out": 0, "cw": 0, "cr": 0}
prev = ctx prev = ctx
@@ -808,11 +1076,14 @@ def main():
"cache_creation": {"ephemeral_1h_input_tokens": tot["cw"], "ephemeral_5m_input_tokens": 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"}}] denials = ([{"tool_name": "Bash", "tool_use_id": "toolu_stub", "tool_input": {"command": "rm -rf build"}}]
if "deny" in d else []) if "deny" in d else [])
is_error = "error" in d 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}) 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", emit({"type": "result", "subtype": "error_during_execution" if "error" in d else "success",
"is_error": is_error, "num_turns": turns, "total_cost_usd": total, "usage": usage, "is_error": is_error, "api_error_status": int(api) if api else None, "num_turns": turns,
"modelUsage": mu, "permission_denials": denials, "session_id": session, "result": "READY"}) "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")) return 1 if is_error else int(d.get("exit", "0"))
@@ -1100,6 +1371,43 @@ class RuleTests(Base):
self.assertEqual(forks.base_warmth(base, short, t0 + 300 - 119)["base_state"], "expiring") self.assertEqual(forks.base_warmth(base, short, t0 + 300 - 119)["base_state"], "expiring")
self.assertEqual(forks.base_warmth({}, cfg, t0)["base_state"], "cold") 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): def test_stream_state(self):
ss = forks.StreamState() ss = forks.StreamState()
u1 = {"input_tokens": 2, "cache_creation_input_tokens": 10, "cache_read_input_tokens": 100} u1 = {"input_tokens": 2, "cache_creation_input_tokens": 10, "cache_read_input_tokens": 100}
@@ -1323,6 +1631,87 @@ class LaunchTests(LaunchBase):
self.assertEqual(r["status"], "ok") self.assertEqual(r["status"], "ok")
self.assertEqual(r["fingerprint"], "drift:binary") 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): def test_timeout_and_stalled(self):
self.setup_run(stall=1) self.setup_run(stall=1)
self.train("B1") self.train("B1")