From 7b9b7488debbe10764d721425cc48656ab5f0dbc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gabor=20K=C3=B6rber?= Date: Fri, 16 Oct 2026 16:26:59 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20forks:=20Phase=201=20review=20fi?= =?UTF-8?q?xes=20(I1-I3,=20M1-M5,=20M9)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- plugins/spire/skills/forks/forks.py | 369 +++++++++++++--- .../spire/skills/forks/tests/test_forks.py | 409 +++++++++++++++++- 2 files changed, 706 insertions(+), 72 deletions(-) diff --git a/plugins/spire/skills/forks/forks.py b/plugins/spire/skills/forks/forks.py index c8e07d1..184455d 100644 --- a/plugins/spire/skills/forks/forks.py +++ b/plugins/spire/skills/forks/forks.py @@ -51,9 +51,13 @@ SPAWN_METHOD = "cim-hidden" if IS_WINDOWS else "setsid" CREATE_NO_WINDOW = 0x08000000 _NOWIN = CREATE_NO_WINDOW if IS_WINDOWS else 0 -# Registry file semantics (C3a). -LOCK_STALE_S = 30.0 -LOCK_WAIT_S = 45.0 # longer than LOCK_STALE_S, so a waiter can break a stale lock +# Registry file semantics (C3a). A lock is stale when its holder (pid + creation time) is dead, +# or when it is older than LOCK_ABANDONED_S. Every critical section is a read-modify-write of a few +# 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 RETRY_TRIES = 40 # PermissionError retries for replace and read (>= 20) RETRY_SLEEP_S = 0.05 @@ -61,7 +65,8 @@ RETRY_SLEEP_S = 0.05 MODEL_RE = re.compile(r"^claude-[a-z0-9-]+(\[1m\])?$") ONE_M_SUFFIX = "[1m]" EFFORTS = ("low", "medium", "high", "xhigh", "max") -PERMISSION_MODES = ("default", "acceptEdits", "dontAsk", "bypassPermissions", "auto") +# `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",) 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({ "CLAUDECODE", "CLAUDE_CODE_CHILD_SESSION", "AI_AGENT", "CLAUDE_PID", "CLAUDE_CODE_GIT_BASH_PATH", "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_BINARY = "CLAUDE_CODE_EXECPATH" # set to the stored absolute binary path 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 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. HEARTBEAT_S = 10.0 # the supervisor writes heartbeat_at at least this often (<= 30 s) LOST_AFTER_S = 120.0 # heartbeat older than this and a dead supervisor -> `lost` SUP_TICK_S = 0.25 # supervisor main-loop period +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) WAIT_POLL_S = 2.0 # `wait` poll period WAIT_DEFAULT_S = 540 @@ -362,7 +371,10 @@ def read_json(path: str): 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): self.path = lock_path(state) @@ -390,20 +402,66 @@ class RegistryLock: return True 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: raw = read_bytes_retry(self.path) mtime = os.path.getmtime(self.path) except (FileNotFoundError, PermissionError): - return None, None - at = None + return None, None, None + holder, at = None, None 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): pass if at is None: 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": deadline = time.monotonic() + self.timeout @@ -411,16 +469,12 @@ class RegistryLock: if self._try_create(): self.held = True return self - raw, age = self._read_lock() - if raw is not None and age is not None and age > LOCK_STALE_S: - # Break it only if it is still the same lock we judged stale. - again, _ = self._read_lock() - if again == raw: - with contextlib.suppress(FileNotFoundError, PermissionError): - os.remove(self.path) - self.broke_stale = True - _diag("broke stale registry lock (age %.1f s)" % age) - continue + raw, age, holder = self._read_lock() + if raw is not None and self._is_stale(holder, age) and self._break(raw): + self.broke_stale = True + _diag("broke stale registry lock (age %.1f s, holder pid %s)" + % (age, (holder or {}).get("pid"))) + continue # the path is free: try to create at once if time.monotonic() >= deadline: raise ForksError("registry locked", lock=self.path, waited_s=self.timeout) 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) -def ledger_append(state: str, line: dict) -> None: - """Append one JSON line. Caller must hold RegistryLock(state) (C3c).""" - data = (json.dumps(line, sort_keys=True) + "\n").encode("utf-8") +def _ledger_jobs(state: str) -> tuple[set, bool]: + """(job ids that already have a ledger line, whether the file ends without a newline).""" + 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(): with open(ledger_path(state), "ab") as f: f.write(data) _retry(_append) + return True # -------------------------------------------------------------------------------------- @@ -545,6 +622,24 @@ def proc_ctime(pid: int) -> int | 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: """{"pid", "ctime"}: the pair every stored pid is compared by.""" return {"pid": int(pid), "ctime": proc_ctime(pid)} @@ -619,10 +714,19 @@ def process_table() -> list[tuple[int, int, str]]: 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 - a child only if it was created at or after its parent (guards against pid reuse). Descendants - of a dead root are still found: their parent pid stays recorded.""" + a child only if it was created at or after its parent (guards against pid reuse). + + 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"): return [] kids, names = {}, {} @@ -631,20 +735,33 @@ def process_tree(rec: dict | None) -> list[dict]: if pid != ppid: kids.setdefault(ppid, []).append(pid) root = int(rec["pid"]) + root_ct = rec.get("ctime") out = [] if record_alive(rec): - out.append({"pid": root, "ctime": rec.get("ctime"), "ppid": None, "name": names.get(root)}) - stack, seen = [(root, rec.get("ctime"))], {root} + out.append({"pid": root, "ctime": root_ct, "ppid": None, "name": names.get(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: ppid, pct = stack.pop() - now_ct = proc_ctime(ppid) - if pct is not None and now_ct is not None and now_ct != pct: - continue # this pid now belongs to another process: its children are not ours + is_dead_root = ppid == root and dead_bounds is not None + if not is_dead_root: + now_ct = proc_ctime(ppid) + if pct is not None and now_ct is not None and now_ct != pct: + continue # this pid now belongs to another process: its children are not ours for c in kids.get(ppid, []): if c in seen: continue ct = proc_ctime(c) - if pct is not None and ct is not None and ct < pct: + 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 seen.add(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" -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 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.""" roots = [r for r in [rec, *extra] if r and r.get("pid")] snap, seen = [], set() for r in roots: - for p in process_tree(r): + for p in process_tree(r, alive_until): key = (p["pid"], p["ctime"]) if key not in seen: 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} +_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: """C5: first assistant event's cache_read against the base's context tokens.""" 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 [] if hint == "exited": 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" detail = ss.refused or "models_used %s, requested %s" % (models_used, requested) elif res is None: status = "failed" reason = "exit_code" if exit_code else "no_result" detail = "exit %s, no result event" % exit_code + elif api: + status, (reason, detail) = "failed", api elif res.get("is_error"): status, reason, detail = "failed", "is_error", str(res.get("subtype")) 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_model_usage_cum"] = u["model_usage"] if base is not None: - hit_t = (ss.last_assistant_at if kind == "train" else ss.first_at) if ss else None + # 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: base["last_hit"] = _iso(hit_t) 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, detail=None, exit_code=None, wall_s=None, base_commit=None) -> dict | None: - """Write `results/.json`, the ledger line and the registry update, exactly once, under the - lock (C3c). Returns the result, or None when the job was already final.""" + """Write `results/.json`, the registry update and the ledger line, exactly once, under the + 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): reg = read_registry(state) 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, wall_s, base_commit, _now()) write_json_atomic(result_file(state, job), result) - ledger_append(state, ledger) write_registry(state, reg) + ledger_append(state, ledger) return result def bump_last_hit(state: str, base_name: str, t: float) -> None: - """Record a hit on the base prefix (first assistant event of a fork/continue) as it happens.""" + """Record a hit on the base prefix (first assistant event of a fork) as it happens.""" with RegistryLock(state): reg = read_registry(state) 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) +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: """N1: record that the orchestrator called us (wait, fork, continue, train, list).""" + if in_child(): + return with RegistryLock(state): reg = read_registry(state) if reg is None: return - reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso() + stamp_orchestrator(reg) write_registry(state, reg) @@ -1335,6 +1519,13 @@ def _job_runtime(state: str, job: str) -> dict: 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]: """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, @@ -1353,7 +1544,7 @@ def reap_lost(state: str) -> list[str]: if sup and record_alive(sup): continue 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")) res = finalize_job(state, job, ss=StreamState.from_file(raw_file(state, job)), hint="lost", 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") cl = rt.get("claude") 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 []) if by == "expire": flags.append("expired") @@ -1584,6 +1775,12 @@ def cmd_where(args) -> tuple[dict, int]: except ForksError as e: binary["error"] = e.payload["error"] 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 = { "script": script_path(), "python": py, @@ -1593,6 +1790,7 @@ def cmd_where(args) -> tuple[dict, int]: "spawn": SPAWN_METHOD, "scrub": scrub_report(), "platform": sys.platform, + "warnings": warnings, "ok": not reasons, } if reasons: @@ -1625,7 +1823,8 @@ def build_init_config(args, state: str) -> dict: raise ForksError("permission mode refused", permission_mode=mode, detail="plan mode swaps models; use dontAsk") if mode not in PERMISSION_MODES: - raise ForksError("unknown permission mode", permission_mode=mode, allowed=list(PERMISSION_MODES)) + 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]) if not tools: raise ForksError("--tools is empty") @@ -1905,7 +2104,7 @@ def cmd_train(args) -> tuple[dict, int]: head = _git_head(la["cwd"]) with RegistryLock(state): reg = read_registry(state) - reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso() + stamp_orchestrator(reg) old = reg["bases"].get(name) if old and old.get("status") != "failed": raise ForksError("base exists", base=name, status=old.get("status")) @@ -1954,7 +2153,7 @@ def cmd_fork(args) -> tuple[dict, int]: reap_lost(state) with RegistryLock(state): reg = read_registry(state) - reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso() + stamp_orchestrator(reg) base = reg["bases"][args.base] if base.get("status") != "active" and not args.force: raise ForksError("base is not active (use --force to fork it anyway)", base=args.base, @@ -2023,7 +2222,7 @@ def cmd_continue(args) -> tuple[dict, int]: reap_lost(state) with RegistryLock(state): reg = read_registry(state) - reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso() + stamp_orchestrator(reg) fj = reg["jobs"][args.job] if 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 ------------------------------------------------------------------- +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: """`_supervise`: runs one `claude` call, parses its stream, watches it, writes the result. The only writer of `jobs/.json` (C3b).""" @@ -2141,20 +2376,28 @@ class Supervisor: stall_s = float(self.cfg.get("stall_after_s") or CONFIG_DEFAULTS["stall_after_s"]) orphan_s = float(self.cfg.get("orphan_after") or CONFIG_DEFAULTS["orphan_after"]) hint, reason, detail, flags_extra = "exited", None, None, [] - hit_written = e["kind"] == "train" + hit_written = e["kind"] != "fork" # only a fork's first request refreshes (I3) + clock = SupClock(t0) last_orphan_check = 0.0 last_turns = 0 while p.poll() is None: 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: refused, first_at, turns = self.ss.refused, self.ss.first_at, self.ss.turns quiet = now - self.last_output if refused: - kill_tree(crec) + kill_tree(crec, alive_until=time.time()) break # decided as model_refused from the stream - if now - t0 > max_s: - kill_tree(crec) - hint, detail = "timeout", "wall time over max_minutes (%.0f s)" % max_s + if clock.elapsed(now) > max_s: + kill_tree(crec, alive_until=time.time()) + hint, detail = "timeout", "run time over max_minutes (%.0f s)" % max_s break if quiet > stall_s and "stalled" not in self.flags: self.flags.append("stalled") @@ -2170,17 +2413,19 @@ class Supervisor: last = _ts(((reg or {}).get("run") or {}).get("last_orchestrator_call_at")) except (ForksError, OSError): last = None - if last is not None and now - last > orphan_s: - kill_tree(crec) + if last is not None and clock.orphan_age(now, last) > orphan_s: + kill_tree(crec, alive_until=time.time()) 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 if now - self._last_write >= HEARTBEAT_S or (turns != last_turns and now - self._last_write >= 1.0): last_turns = turns self.write_rt() time.sleep(SUP_TICK_S) exit_code = p.wait() - sweep = kill_tree(crec) # leftovers of claude (backgrounded commands) + # leftovers of claude (backgrounded commands); claude exited just now + sweep = kill_tree(crec, alive_until=time.time()) if sweep["snapshot"]: _diag("swept %d leftover process(es) of claude: %s" % ( 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): rt = _job_runtime(state, args.job) 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)), hint="failed", reason="supervisor_error", detail="%s: %s" % (type(ex).__name__, ex)) raise @@ -2278,7 +2523,7 @@ def cmd_wait(args) -> tuple[dict, int]: if met: for j in finished: reg["jobs"][j]["settled_at"] = _iso(now) - reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso(now) + stamp_orchestrator(reg, now) write_registry(state, reg) statuses = {j: reg["jobs"][j].get("status") for j in finished} entries = {j: dict(reg["jobs"][j]) for j in running} @@ -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 if unknown and args.command not in PHASE2_COMMANDS: parser.error("unrecognized arguments: %s" % " ".join(unknown)) - if args.command in CHILD_REFUSED and os.environ.get(CHILD_MARKER): - raise ForksError("launching commands are refused inside a fork (%s is set)" % CHILD_MARKER, - command=args.command) + if args.command in CHILD_REFUSED and in_child(): + raise ForksError("`%s` is refused inside a fork (%s is set); a fork may only run where, list, " + "usage and wait" % (args.command, CHILD_MARKER), command=args.command) return HANDLERS[args.command](args) diff --git a/plugins/spire/skills/forks/tests/test_forks.py b/plugins/spire/skills/forks/tests/test_forks.py index eee6b28..5fbebc2 100644 --- a/plugins/spire/skills/forks/tests/test_forks.py +++ b/plugins/spire/skills/forks/tests/test_forks.py @@ -175,7 +175,9 @@ class ScrubTests(Base): 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"]) + # 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) 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: - 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: self.assertTrue(lk.broke_stale) + self.assertLess(time.monotonic() - t0, 1.5) with open(forks.lock_path(self.state)) as f: self.assertEqual(json.load(f)["token"], lk.token) + self.assertEqual([n for n in os.listdir(self.state) if n.endswith(".lock")], []) + + def test_abandoned_lock_broken_by_age(self): + me = forks.pid_record(os.getpid()) # a live holder: only the age rule applies + self.write_lock(me, time.time() - (forks.LOCK_ABANDONED_S + 1), "old") + with forks.RegistryLock(self.state, timeout=2) as lk: + self.assertTrue(lk.broke_stale) def test_garbage_stale_lock_broken_by_mtime(self): p = forks.lock_path(self.state) with open(p, "w") as f: f.write("") - old = time.time() - (forks.LOCK_STALE_S + 5) + old = time.time() - (forks.LOCK_ABANDONED_S + 5) os.utime(p, (old, old)) with forks.RegistryLock(self.state, timeout=2) as lk: self.assertTrue(lk.broke_stale) def test_fresh_lock_not_broken(self): + # a live holder's lock is never broken before LOCK_ABANDONED_S (garbage content neither) + self.write_lock(forks.pid_record(os.getpid()), time.time() - 60, "other") + with self.assertRaises(forks.ForksError): + forks.RegistryLock(self.state, timeout=0.3).acquire() with open(forks.lock_path(self.state), "w") as f: - json.dump({"pid": 1, "at": time.time() - 5, "token": "other"}, f) + f.write("") with self.assertRaises(forks.ForksError): forks.RegistryLock(self.state, timeout=0.3).acquire() def test_stale_under_injected_clock_and_safe_release(self): l1 = forks.RegistryLock(self.state).acquire() real = time.time() - with mock.patch.object(forks, "_now", lambda: real + forks.LOCK_STALE_S + 1): + with mock.patch.object(forks, "_now", lambda: real + forks.LOCK_ABANDONED_S + 1): l2 = forks.RegistryLock(self.state, timeout=2).acquire() self.assertTrue(l2.broke_stale) l1.release() # must not remove l2's lock @@ -248,6 +273,75 @@ class LockTests(Base): l2.release() self.assertFalse(os.path.exists(forks.lock_path(self.state))) + def test_unbreakable_stale_lock_respects_deadline(self): + """M2: a stale lock that cannot be removed or renamed (AV, sync client) must not spin: + the call ends with its "registry locked" error once the deadline passes.""" + self.write_lock(self.dead_holder(), time.time() - 1000, "old") # stale by holder and by age + lp = os.path.normcase(forks.lock_path(self.state)) + real_remove, real_rename = os.remove, os.rename + + def no_remove(p, *a, **k): + if os.path.normcase(str(p)) == lp: + raise PermissionError(13, "held by another program", p) + return real_remove(p, *a, **k) + + def no_rename(src, dst, *a, **k): + if os.path.normcase(str(src)) == lp: + raise PermissionError(13, "held by another program", src) + return real_rename(src, dst, *a, **k) + box = {} + + def attempt(): + t0 = time.monotonic() + try: + forks.RegistryLock(self.state, timeout=0.5).acquire() + box["result"] = "acquired" + except forks.ForksError as e: + box["result"] = e.payload["error"] + box["took"] = time.monotonic() - t0 + with mock.patch.object(os, "remove", no_remove), mock.patch.object(os, "rename", no_rename): + th = threading.Thread(target=attempt, daemon=True) + th.start() + th.join(5) + alive = th.is_alive() + self.assertFalse(alive, "acquire spins on a lock it cannot break (hot loop, no deadline)") + self.assertEqual(box["result"], "registry locked") + self.assertLess(box["took"], 2.0) + + def test_fresh_lock_taken_during_break_survives(self): + """M2: between a waiter's judgement (stale) and its break, another waiter breaks the stale + lock and takes a fresh one. The first waiter must not delete the fresh lock.""" + self.write_lock(self.dead_holder(), time.time(), "stale") + lp = os.path.normcase(forks.lock_path(self.state)) + fresh = {"pid": os.getpid(), "ctime": forks.proc_ctime(os.getpid()), "at": time.time(), + "token": "fresh"} + real_remove, real_rename = os.remove, os.rename + swapped = [] + + def swap_in_fresh(): + if not swapped: + swapped.append(1) + with open(forks.lock_path(self.state), "w") as f: + json.dump(dict(fresh, at=time.time()), f) + + def hook_remove(p, *a, **k): + if os.path.normcase(str(p)) == lp: + swap_in_fresh() + return real_remove(p, *a, **k) + + def hook_rename(src, dst, *a, **k): + if os.path.normcase(str(src)) == lp: + swap_in_fresh() + return real_rename(src, dst, *a, **k) + with mock.patch.object(os, "remove", hook_remove), mock.patch.object(os, "rename", hook_rename): + with self.assertRaises(forks.ForksError): + forks.RegistryLock(self.state, timeout=0.5).acquire() + self.assertEqual(swapped, [1], "the break was never attempted") + with open(forks.lock_path(self.state)) as f: + self.assertEqual(json.load(f)["token"], "fresh") + self.assertEqual(sorted(n for n in os.listdir(self.state) if n.endswith(".lock")), + ["registry.lock"]) + # -------------------------------------------------------------------------------------- class AtomicTests(Base): @@ -298,6 +392,29 @@ class AtomicTests(Base): lines = f.read().decode().splitlines() self.assertEqual([json.loads(x)["job"] for x in lines], ["j0001", "j0002"]) + def test_ledger_append_dedupes_by_job(self): + """M1: a second finaliser (kill racing the supervisor) never writes a second line.""" + st = os.path.join(self.tmp, "st") + forks.ensure_state(st) + with forks.RegistryLock(st): + self.assertTrue(forks.ledger_append(st, {"job": "j0001", "status": "ok"})) + self.assertFalse(forks.ledger_append(st, {"job": "j0001", "status": "killed"})) + self.assertTrue(forks.ledger_append(st, {"job": "j0002", "status": "ok"})) + # a torn last line (writer killed mid-append) does not swallow the next line + with open(forks.ledger_path(st), "ab") as f: + f.write(b'{"job": "j0003", "sta') + with forks.RegistryLock(st): + self.assertTrue(forks.ledger_append(st, {"job": "j0004", "status": "ok"})) + self.assertFalse(forks.ledger_append(st, {"job": "j0002", "status": "killed"})) + with open(forks.ledger_path(st), "rb") as f: + lines = f.read().decode().splitlines() + parsed = [] + for x in lines: + with contextlib.suppress(ValueError): + parsed.append(json.loads(x)) + self.assertEqual([(p["job"], p["status"]) for p in parsed], + [("j0001", "ok"), ("j0002", "ok"), ("j0004", "ok")]) + # -------------------------------------------------------------------------------------- class PidTests(Base): @@ -318,6 +435,82 @@ class PidTests(Base): self.assertTrue(os.path.isfile(forks.supervisor_interpreter())) +# -------------------------------------------------------------------------------------- +class ProcessTreeTests(Base): + """I2: walking a dead root's tree on a fake process table (nothing real is touched or killed).""" + + ROOT, A, A_KID, B, C, C_KID, OTHER = 900001, 900011, 900012, 900021, 900031, 900032, 900041 + + def setUp(self): + super().setUp() + self.t0 = time.time() - 5000 # the job (its root) started here + self.hb = self.t0 + 500 # last heartbeat: the job was known alive until here + e2c = forks.epoch_to_ctime + if e2c(self.t0) is None: + self.skipTest("no epoch -> creation-time conversion on this host") + self.root_ct = e2c(self.t0) + self.ctimes = { + self.A: e2c(self.t0 + 600), # (a) reused pid's child: created after the last heartbeat + self.A_KID: e2c(self.t0 + 601), + self.B: e2c(self.t0 - 100), # (b) created before the job started + self.C: e2c(self.t0 + 100), # (c) the job's own child, created while it ran + self.C_KID: e2c(self.t0 + 700), # its child (parent verified alive): ours + self.OTHER: e2c(self.t0 + 50), + } + self.table = [(self.A, self.ROOT, "a.exe"), (self.A_KID, self.A, "a-kid.exe"), + (self.B, self.ROOT, "dwm.exe"), (self.C, self.ROOT, "bash.exe"), + (self.C_KID, self.C, "sleep.exe"), (self.OTHER, 4, "other.exe")] + self.live = set(self.ctimes) + for name, fn in (("process_table", lambda: list(self.table)), + ("proc_ctime", lambda pid: self.ctimes.get(int(pid)) if int(pid) in self.live else None), + ("pid_alive", self.fake_alive)): + p = mock.patch.object(forks, name, fn) + p.start() + self.addCleanup(p.stop) + + def fake_alive(self, pid, ctime=None): + pid = int(pid) + return pid in self.live and (ctime is None or self.ctimes.get(pid) == ctime) + + def rec(self): + return {"pid": self.ROOT, "ctime": self.root_ct} + + def test_dead_root_claims_only_children_from_its_lifetime(self): + got = {p["pid"] for p in forks.process_tree(self.rec(), alive_until=self.hb)} + self.assertEqual(got, {self.C, self.C_KID}) + + def test_dead_root_without_bounds_claims_nothing(self): + self.assertEqual(forks.process_tree(self.rec()), []) + self.assertEqual(forks.process_tree({"pid": self.ROOT, "ctime": None}, alive_until=self.hb), []) + + def test_live_root_unchanged(self): + self.live.add(self.ROOT) + self.ctimes[self.ROOT] = self.root_ct + got = {p["pid"] for p in forks.process_tree(self.rec(), alive_until=None)} + self.assertEqual(got, {self.ROOT, self.A, self.A_KID, self.C, self.C_KID}) # B is older + + def test_reap_lost_bounds_by_heartbeat(self): + st = os.path.join(self.tmp, "st") + forks.ensure_state(st) + old = time.time() - 1000 + reg = forks.new_registry() + reg["config"] = {"launch": {}, "ttl_s": 3600} + dead_sup = {"pid": 900091, "ctime": 1} + reg["jobs"]["j0001"] = forks._new_entry("j0001", "fork", status="running", supervisor=dead_sup, + launched_at=forks._iso(old - 10)) + with forks.RegistryLock(st): + forks.write_registry(st, reg) + forks.write_json_atomic(forks.job_file(st, "j0001"), { + "supervisor": dead_sup, "claude": self.rec(), "started_at": forks._iso(old - 10), + "heartbeat_at": forks._iso(old)}) + seen = [] + with mock.patch.object(forks, "kill_tree", lambda rec, extra=(), alive_until=None: + seen.append((rec, alive_until)) or {"snapshot": [], "survivors": [], + "taskkill_rc": None}): + self.assertEqual(forks.reap_lost(st), ["j0001"]) + self.assertEqual(seen, [(self.rec(), forks._ts(forks._iso(old)))]) + + # -------------------------------------------------------------------------------------- class InitTests(Base): def setUp(self): @@ -478,6 +671,19 @@ class InitTests(Base): self.assert_refused_no_state("--permission-mode", "yolo") self.assert_refused_no_state("--effort", "turbo") + def test_permission_modes_match_2_1_299(self): + """M5: `claude --help` 2.1.299 offers acceptEdits, auto, bypassPermissions, manual, dontAsk, + plan. `default` is not one of them; `plan` is refused.""" + out = self.assert_refused_no_state("--permission-mode", "default") + self.assertEqual(out["error"], "unknown permission mode") + self.assertEqual(sorted(out["allowed"]), + sorted(["acceptEdits", "auto", "bypassPermissions", "manual", "dontAsk"])) + for mode in ("acceptEdits", "auto", "bypassPermissions", "manual", "dontAsk"): + with self.subTest(mode=mode): + code, out, _ = self.init("--permission-mode", mode) + self.assertEqual(code, 0, out) + self.assertEqual(out["config"]["launch"]["permission_mode"], mode) + def test_missing_binary_refused(self): out = self.assert_refused_no_state("--binary", os.path.join(self.tmp, "nope.exe")) self.assertEqual(out["error"], "claude binary not found") @@ -620,6 +826,64 @@ class CliTests(Base): self.assert_one_json_line(raw) self.assertEqual(out["error"], "internal error") + def test_where_warns_on_base_url(self): + """M9: `where` warns (exit 0) when ANTHROPIC_BASE_URL is set to anything but the default.""" + stub = make_stub_binary(self.tmp) + os.environ.pop("ANTHROPIC_BASE_URL", None) + for value in (None, "https://api.anthropic.com", "https://api.anthropic.com/"): + with self.subTest(value=value): + if value is not None: + os.environ["ANTHROPIC_BASE_URL"] = value + code, out, raw = call(["where", "--binary", stub]) + self.assertEqual(code, 0, out) + self.assert_one_json_line(raw) + self.assertEqual(out["warnings"], []) + os.environ["ANTHROPIC_BASE_URL"] = "https://llm-gateway.example.internal/anthropic" + code, out, raw = call(["where", "--binary", stub]) + self.assertEqual(code, 0, out) + self.assert_one_json_line(raw) + self.assertTrue(out["ok"]) + self.assertEqual(len(out["warnings"]), 1, out["warnings"]) + self.assertIn("ANTHROPIC_BASE_URL", out["warnings"][0]) + self.assertIn("llm-gateway.example.internal", out["warnings"][0]) + + def test_child_refuses_kill_expire_init(self): + """M4: a fork may not run kill, expire or init.""" + os.environ[forks.CHILD_MARKER] = "1" + stub = make_stub_binary(self.tmp) + run_log = os.path.join(self.tmp, "r-run.md") + for argv in (["kill", "--run", run_log, "--job", "j0001"], + ["expire", "--run", run_log, "--all"], + ["init", "--run", run_log, "--model", "claude-opus-5-5", "--cwd", self.tmp, + "--binary", stub]): + with self.subTest(cmd=argv[0]): + code, out, raw = call(argv) + self.assertEqual(code, 2, out) + self.assert_one_json_line(raw) + self.assertIn("inside a fork", out["error"]) + self.assertEqual(out["command"], argv[0]) + self.assertFalse(os.path.exists(run_log[:-3] + "-forks")) + + def test_child_does_not_stamp_orchestrator(self): + """M4: list and wait from inside a fork leave run.last_orchestrator_call_at alone.""" + os.makedirs(os.path.join(self.tmp, "reports")) + run_log = os.path.join(self.tmp, "reports", "r-run.md") + state = forks.state_dir_for(run_log) + forks.ensure_state(state) + old = forks._iso(time.time() - 5000) + reg = forks.new_registry() + reg["config"] = {"launch": {}, "ttl_s": 3600} + reg["run"]["last_orchestrator_call_at"] = old + with forks.RegistryLock(state): + forks.write_registry(state, reg) + os.environ[forks.CHILD_MARKER] = "1" + self.assertEqual(call(["list", "--run", run_log])[0], 0) + self.assertEqual(call(["wait", "--run", run_log, "--all", "--timeout", "1"])[0], 0) + self.assertEqual(forks.read_registry(state)["run"]["last_orchestrator_call_at"], old) + os.environ.pop(forks.CHILD_MARKER) + self.assertEqual(call(["list", "--run", run_log])[0], 0) # the orchestrator does stamp + self.assertNotEqual(forks.read_registry(state)["run"]["last_orchestrator_call_at"], old) + def test_where_in_process_missing_binary(self): code, out, raw = call(["where", "--binary", os.path.join(self.tmp, "nope.exe")]) self.assertEqual(code, 2) @@ -764,6 +1028,10 @@ def main(): if "silent" in d: time.sleep(float(d["silent"])) turns = int(d.get("turns", "3")) + api = d.get("apierr") # api_error_status of the result, e.g. 429 + is_error = "error" in d or api is not None + if is_error: + turns = 0 # rejected before the model answered: nothing billed ctx = int(prior.get("context") or 0) tot = {"in": 0, "out": 0, "cw": 0, "cr": 0} prev = ctx @@ -808,11 +1076,14 @@ def main(): "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 + errtext = d.get("errtext", "").replace("_", " ") + if is_error and not errtext: + errtext = ("API Error: " + api) if api else "stub error" save(os.path.join(SESS, session + ".json"), {"modelUsage": mu, "cost": total, "context": prev}) - emit({"type": "result", "subtype": "error_during_execution" if 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"}) + emit({"type": "result", "subtype": "error_during_execution" if "error" in d else "success", + "is_error": is_error, "api_error_status": int(api) if api else None, "num_turns": turns, + "total_cost_usd": total, "usage": usage, "modelUsage": mu, "permission_denials": denials, + "session_id": session, "result": errtext if is_error else "READY"}) return 1 if is_error else int(d.get("exit", "0")) @@ -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({}, cfg, t0)["base_state"], "cold") + def test_api_error_reason(self): + """I1: API errors get their own reason, from api_error_status or the result text.""" + f = forks.api_error_reason + self.assertIsNone(f(None)) + self.assertIsNone(f({"is_error": False, "result": "we talked about the rate limit"})) + self.assertIsNone(f({"is_error": True, "subtype": "error_during_execution", "result": "boom"})) + cases = [ + ({"api_error_status": 429, "result": "x"}, "usage_limit"), + ({"api_error_status": "429", "result": "x"}, "usage_limit"), + ({"result": "Claude AI usage limit reached|1760000000"}, "usage_limit"), + ({"result": "API Error: 429 rate_limit_error"}, "usage_limit"), + ({"api_error_status": 529, "result": "x"}, "overloaded"), + ({"result": 'API Error: 529 {"type":"error","error":{"type":"overloaded_error"}}'}, "overloaded"), + ({"api_error_status": 400, "result": "Prompt is too long"}, "prompt_too_long"), + ({"api_error_status": 500, "result": "Internal server error"}, "api_error"), + ({"result": "API Error: 502 bad gateway"}, "api_error"), + ] + for res, want in cases: + with self.subTest(res=res): + got = f(dict(res, is_error=True)) + self.assertIsNotNone(got) + self.assertEqual(got[0], want) + self.assertIn("429", f({"is_error": True, "api_error_status": 429, "result": "x"})[1]) + + def test_sup_clock_jump(self): + """M3: a suspend between supervisor ticks is not run time and re-arms the orphan watchdog.""" + c = forks.SupClock(1000.0, tick_s=0.25, jump_s=60.0) + for t in (1000.25, 1000.5, 1000.75, 1030.75): # a 30 s stall is not a jump + self.assertIsNone(c.tick(t)) + self.assertAlmostEqual(c.elapsed(1030.75), 30.75) + self.assertAlmostEqual(c.tick(8230.75), 7199.75) # 2 h suspend + self.assertAlmostEqual(c.elapsed(8231.0), 31.25) # 30.75 + one tick of the gap + 0.25 + last = 990.0 # the last orchestrator call, before it + self.assertAlmostEqual(c.orphan_age(8231.0, last), 0.25) # re-armed at the jump + self.assertGreater(c.orphan_age(8230.75 + 1201, last), 1200) # one more orphan_after + self.assertAlmostEqual(c.orphan_age(9000.0, 8900.0), 100.0) # later calls count as usual + def test_stream_state(self): ss = forks.StreamState() u1 = {"input_tokens": 2, "cache_creation_input_tokens": 10, "cache_read_input_tokens": 100} @@ -1323,6 +1631,87 @@ class LaunchTests(LaunchBase): self.assertEqual(r["status"], "ok") self.assertEqual(r["fingerprint"], "drift:binary") + def test_api_errors_are_not_model_refused(self): + """I1: an error before the model answers (nothing billed, models_used []) is not a refusal; + API errors get their own reason; a real substitution still reads model_refused.""" + self.setup_run(cap=8) + self.train("B1") + self.finish("j0001") + cases = [(("error",), "is_error", None), + (("apierr=429",), "usage_limit", "429"), + (("error", "errtext=Claude_AI_usage_limit_reached"), "usage_limit", "usage limit"), + (("apierr=529",), "overloaded", "529"), + (("apierr=400", "errtext=Prompt_is_too_long"), "prompt_too_long", "400"), + (("apierr=500",), "api_error", "500"), + (("model=claude-other-9",), "model_refused", "claude-other-9")] + jobs = [self.fork(*c)["job"] for c, _, _ in cases] + code, out = self.wait(*jobs) + self.assertEqual(code, 4, out) + res = {x["job"]: x for x in out["done"]} + reg = self.reg() + for job, (directives, reason, detail) in zip(jobs, cases): + with self.subTest(directives=directives): + r = res[job] + self.assertEqual((r["status"], r["reason"]), ("failed", reason), r) + if reason != "model_refused": + self.assertEqual(r["models_used"], []) + self.assertEqual(r["cost_delta_usd"], 0.0) + if detail: + self.assertIn(detail, reg["jobs"][job]["reason_detail"]) + + def test_continue_leaves_base_last_hit(self): + """I3: a continue reads the fork's own longer prefix; only a fork refreshes the base.""" + self.setup_run() + self.train("B1") + self.finish("j0001") + fj = self.fork("turns=1")["job"] + self.assertEqual(self.finish(fj)["status"], "ok") + old = forks._iso(time.time() - 1000) + with forks.RegistryLock(self.state): + reg = forks.read_registry(self.state) + reg["bases"]["B1"]["last_hit"] = old + forks.write_registry(self.state, reg) + r = self.finish(self.cmd("continue", "--job", fj, "--brief", self.brief("turns=2"))["job"]) + self.assertEqual((r["status"], r["kind"]), ("ok", "continue")) + self.assertEqual(self.reg()["bases"]["B1"]["last_hit"], old) + r = self.finish(self.fork("turns=1")["job"]) # a fork still refreshes it + self.assertEqual(r["status"], "ok") + self.assertGreater(forks._ts(self.reg()["bases"]["B1"]["last_hit"]), forks._ts(old)) + + def test_clock_jump_not_orphaned_or_timed_out(self): + """M3: a supervisor (run in-process under an injected clock) sees the clock jump 2 h mid-run, + as after a host suspend. Neither orphan_after (600 s) nor max_minutes (1) may fire; the + result carries `clock_jump` and the jump is logged.""" + self.setup_run() # orphan_after 600 s + self.train("B1") + self.finish("j0001") + p = subprocess.Popen([sys.executable, "-c", "pass"]) + dead = forks.pid_record(p.pid) # recorded as the supervisor: never us + p.wait() + fake = {"ok": True, "method": "test", "pid": dead["pid"], "ctime": dead["ctime"]} + with mock.patch.object(forks, "spawn_supervisor", lambda *a: fake): + job = self.fork("sleep=1", "turns=3", extra=("--max-minutes", "1"))["job"] + sup = forks.Supervisor(self.state, job) + sup.rt["supervisor"] = dead + offset = [0.0] + real = time.time + jump = threading.Timer(1.0, offset.__setitem__, args=(0, 7200.0)) + err = io.StringIO() + try: + with mock.patch.dict(os.environ), mock.patch.object(forks, "_now", lambda: real() + offset[0]), \ + contextlib.redirect_stderr(err): + jump.start() + out, code = sup.run() + finally: + jump.cancel() + self.assertEqual(offset[0], 7200.0, "the clock never jumped during the run") + self.assertEqual(code, 0, out) + e = self.reg()["jobs"][job] + self.assertEqual((e["status"], e["reason"]), ("ok", None), e) + self.assertIn("clock_jump", e["flags"]) + self.assertIn("clock_jump", forks.read_json(forks.result_file(self.state, job))["flags"]) + self.assertIn("clock jump", err.getvalue()) + def test_timeout_and_stalled(self): self.setup_run(stall=1) self.train("B1")