2404 lines
99 KiB
Python
2404 lines
99 KiB
Python
# /// script
|
|
# requires-python = ">=3.10"
|
|
# dependencies = []
|
|
# ///
|
|
"""spire:forks helper: trains one headless implementer per build and forks it per block.
|
|
|
|
Run as `uv run <abs path>/forks.py <command> [args]`. Every call prints exactly one JSON
|
|
object on stdout (newline-terminated); diagnostics go to stderr. Exit codes: 0 ok, 2 usage
|
|
error or refused, 3 `wait` condition not met at timeout, 4 `wait` condition met and a job
|
|
in it did not end `ok`.
|
|
|
|
Standard library only. State lives next to the run log: `<run log without .md>-forks/`.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import contextlib
|
|
import datetime as _dt
|
|
import hashlib
|
|
import json
|
|
import ntpath
|
|
import os
|
|
import platform
|
|
import posixpath
|
|
import re
|
|
import shutil
|
|
import signal
|
|
import string
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import traceback
|
|
import uuid
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
# Constants
|
|
# --------------------------------------------------------------------------------------
|
|
|
|
SCHEMA = 1
|
|
|
|
EXIT_OK = 0
|
|
EXIT_USAGE = 2
|
|
EXIT_WAIT_TIMEOUT = 3
|
|
EXIT_JOB_FAILED = 4
|
|
|
|
IS_WINDOWS = os.name == "nt"
|
|
|
|
# How the detached supervisor is created (C6, P0.1 rung 3 hidden; POSIX: new session).
|
|
SPAWN_METHOD = "cim-hidden" if IS_WINDOWS else "setsid"
|
|
CREATE_NO_WINDOW = 0x08000000
|
|
_NOWIN = CREATE_NO_WINDOW if IS_WINDOWS else 0
|
|
|
|
# Registry file semantics (C3a).
|
|
LOCK_STALE_S = 30.0
|
|
LOCK_WAIT_S = 45.0 # longer than LOCK_STALE_S, so a waiter can break a stale lock
|
|
LOCK_POLL_S = 0.05
|
|
RETRY_TRIES = 40 # PermissionError retries for replace and read (>= 20)
|
|
RETRY_SLEEP_S = 0.05
|
|
|
|
MODEL_RE = re.compile(r"^claude-[a-z0-9-]+(\[1m\])?$")
|
|
ONE_M_SUFFIX = "[1m]"
|
|
EFFORTS = ("low", "medium", "high", "xhigh", "max")
|
|
PERMISSION_MODES = ("default", "acceptEdits", "dontAsk", "bypassPermissions", "auto")
|
|
REFUSED_PERMISSION_MODES = ("plan",)
|
|
|
|
DEFAULT_TOOLS = "Bash,Read,Edit,Write,Glob,Grep,Skill"
|
|
DEFAULT_DISALLOWED = ["Bash(git push:*)", "Bash(git rebase:*)", "Bash(git reset --hard:*)",
|
|
"Bash(claude:*)"]
|
|
|
|
CONFIG_DEFAULTS = {
|
|
"parallel_cap": 2,
|
|
"orphan_after": 1200, # N1, seconds; > 2 x 570
|
|
"pings": "on", # P0.6 passed; used in Phase 2
|
|
"ttl_s": 3600,
|
|
"max_minutes": 90,
|
|
"stall_after_s": 900,
|
|
}
|
|
THRESHOLD_DEFAULTS = {
|
|
"rebase_commits": 6,
|
|
"rebase_lines": 1500,
|
|
"rebase_delta_tokens": 12000,
|
|
"max_base_tokens": 250000,
|
|
"max_rebases": 4,
|
|
}
|
|
|
|
# Env pinned into every launch's fingerprint (design 3.3).
|
|
ENV_PINNED_LAUNCH = {"CLAUDE_CODE_PROMPT_CACHE_TTL": "1h", "DISABLE_AUTOUPDATER": "1"}
|
|
|
|
# --- Env scrub: one constant (design 4 + plan 2 + P0.0) ---
|
|
SCRUB_DROP = frozenset({
|
|
# the twelve Desktop vars (design 4, research E1)
|
|
"CLAUDE_CODE_MESSAGING_SOCKET", "CLAUDE_CODE_MESSAGING_TOKEN", "CLAUDE_CODE_HOST_SESSION_ID",
|
|
"CLAUDE_CODE_ENTRYPOINT", "CLAUDE_CODE_DESKTOP_APP_VERSION",
|
|
"CLAUDE_CODE_SDK_HAS_HOST_AUTH_REFRESH", "CLAUDE_CODE_TERMINAL_MCP_TOOLS",
|
|
"CLAUDE_CODE_REPORT_FINDINGS", "CLAUDE_CODE_SESSION_ID", "CLAUDE_AGENT_SDK_VERSION",
|
|
"CLAUDE_CODE_ENABLE_ASK_USER_QUESTION_TOOL", "CLAUDE_CODE_ENABLE_SDK_FILE_CHECKPOINTING",
|
|
# fingerprint and auth path
|
|
"CLAUDE_EFFORT", "ANTHROPIC_MODEL", "ANTHROPIC_SMALL_FAST_MODEL", "CLAUDE_CODE_SUBAGENT_MODEL",
|
|
"ENABLE_PROMPT_CACHING_1H", "FORCE_PROMPT_CACHING_5M", "CLAUDE_CODE_SUBAGENT_PROMPT_CACHE_TTL",
|
|
"ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", "CLAUDE_CODE_USE_BEDROCK", "CLAUDE_CODE_USE_VERTEX",
|
|
# uv (plan 2, P0.0)
|
|
"VIRTUAL_ENV",
|
|
})
|
|
SCRUB_DROP_PATTERNS = (re.compile(r"^ANTHROPIC_DEFAULT_.+_MODEL$"), re.compile(r"^UV_"))
|
|
SCRUB_KEEP = frozenset({
|
|
"CLAUDECODE", "CLAUDE_CODE_CHILD_SESSION", "AI_AGENT", "CLAUDE_PID", "CLAUDE_CODE_GIT_BASH_PATH",
|
|
"PATH", "HOME", "USERPROFILE",
|
|
})
|
|
SCRUB_PINNED = {**ENV_PINNED_LAUNCH, "SPIRE_FORKS_CHILD": "1"}
|
|
SCRUB_PINNED_BINARY = "CLAUDE_CODE_EXECPATH" # set to the stored absolute binary path
|
|
CHILD_MARKER = "SPIRE_FORKS_CHILD" # N3
|
|
SCRUB_WATCH_PREFIXES = ("CLAUDE_", "ANTHROPIC_") # `where` lists unknown vars with these
|
|
|
|
# C8: the user settings file. Module-level so tests can point it at a temp file.
|
|
USER_SETTINGS_PATH: str | None = None
|
|
|
|
PHASE2_COMMANDS = ("rebase", "usage", "tick")
|
|
CHILD_REFUSED = ("train", "fork", "continue", "rebase") # N3 (tick pings: Phase 2)
|
|
|
|
# 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
|
|
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
|
|
WAIT_MAX_S = 570
|
|
KILL_VERIFY_S = 10.0 # how long `kill` keeps sweeping before it reports survivors
|
|
READER_JOIN_S = 10.0 # after claude exits, how long the supervisor waits for the stream reader
|
|
CACHE_HIT_RATIO = 0.9 # C5
|
|
RUNNING = ("created", "running")
|
|
FINAL = ("ok", "failed", "timeout", "killed", "lost")
|
|
LAUNCH_FIELDS = ("model", "effort", "permission_mode", "tools", "allowed_tools", "disallowed_tools",
|
|
"agent", "cwd", "binary") # fixed at init; fork/continue refuse them
|
|
MU_FIELDS = ("inputTokens", "outputTokens", "cacheReadInputTokens", "cacheCreationInputTokens")
|
|
TOKEN_KEYS = ("input", "output", "cache_creation", "cache_creation_1h", "cache_creation_5m", "cache_read")
|
|
TEMPLATES_DIR = "templates"
|
|
|
|
# CIM hidden spawn (P0.1 rung 3 + ShowWindow=0). Command line and cwd arrive through env vars.
|
|
POWERSHELL_CIM = (
|
|
"$si = New-CimInstance -ClassName Win32_ProcessStartup -ClientOnly -Property @{ShowWindow=[uint16]0}; "
|
|
"$r = Invoke-CimMethod -ClassName Win32_Process -MethodName Create -Arguments "
|
|
"@{CommandLine=$env:SPIRE_FORKS_CMD; CurrentDirectory=$env:SPIRE_FORKS_CWD; ProcessStartupInformation=$si}; "
|
|
"\"{0} {1}\" -f $r.ReturnValue, $r.ProcessId")
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
# Errors, output, clock
|
|
# --------------------------------------------------------------------------------------
|
|
|
|
|
|
class ForksError(Exception):
|
|
"""A refusal or usage error: exit `code`, payload printed as the stdout JSON object."""
|
|
|
|
def __init__(self, error: str, code: int = EXIT_USAGE, **fields):
|
|
super().__init__(error)
|
|
self.code = code
|
|
self.payload = {"error": error, **fields}
|
|
|
|
|
|
class _HelpRequested(Exception):
|
|
def __init__(self, text: str):
|
|
super().__init__("help")
|
|
self.text = text
|
|
|
|
|
|
def _emit(obj: dict) -> None:
|
|
sys.stdout.write(json.dumps(obj) + "\n")
|
|
sys.stdout.flush()
|
|
|
|
|
|
def _diag(msg: str) -> None:
|
|
sys.stderr.write("forks: " + msg + "\n")
|
|
sys.stderr.flush()
|
|
|
|
|
|
def _now() -> float:
|
|
"""The only clock (epoch seconds). Tests monkeypatch it."""
|
|
return time.time()
|
|
|
|
|
|
def _iso(t: float | None = None) -> str:
|
|
"""ISO 8601 with offset, from `_now()` unless given."""
|
|
if t is None:
|
|
t = _now()
|
|
return _dt.datetime.fromtimestamp(t).astimezone().isoformat(timespec="seconds")
|
|
|
|
|
|
def _ts(iso) -> float | None:
|
|
"""Epoch seconds of an ISO string written by `_iso`, or None."""
|
|
if not iso or not isinstance(iso, str):
|
|
return None
|
|
try:
|
|
return _dt.datetime.fromisoformat(iso).timestamp()
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _uniq(items) -> list:
|
|
out = []
|
|
for x in items:
|
|
if x not in out:
|
|
out.append(x)
|
|
return out
|
|
|
|
|
|
def _sha256_bytes(b: bytes) -> str:
|
|
return hashlib.sha256(b).hexdigest()
|
|
|
|
|
|
def _sha256_file(path: str) -> str:
|
|
with open(path, "rb") as f:
|
|
return _sha256_bytes(f.read())
|
|
|
|
|
|
def canonical_json(obj) -> str:
|
|
return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
|
|
|
|
|
|
def fingerprint_hash(launch: dict) -> str:
|
|
return _sha256_bytes(canonical_json(launch).encode("utf-8"))
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
# Paths
|
|
# --------------------------------------------------------------------------------------
|
|
|
|
_GITBASH_DRIVE = re.compile(r"^/([A-Za-z])(?=/|$)")
|
|
|
|
|
|
def norm_path(p: str, cwd: str | None = None, windows: bool | None = None) -> str:
|
|
"""Normalise relative, `G:\\…`, `G:/…` and Git Bash `/g/…` forms.
|
|
|
|
Windows: absolute path with backslashes and an upper-case drive letter.
|
|
POSIX: `os.path.abspath` semantics (relative to `cwd`).
|
|
"""
|
|
if windows is None:
|
|
windows = IS_WINDOWS
|
|
if not isinstance(p, str) or not p.strip():
|
|
raise ForksError("empty path")
|
|
p = p.strip()
|
|
if not windows:
|
|
base = cwd if cwd is not None else os.getcwd()
|
|
return posixpath.normpath(posixpath.join(base, p))
|
|
m = _GITBASH_DRIVE.match(p)
|
|
if m:
|
|
p = m.group(1).upper() + ":" + (p[2:] or "/")
|
|
p = p.replace("/", "\\")
|
|
drive, rest = ntpath.splitdrive(p)
|
|
if not (drive and rest.startswith("\\")):
|
|
base = norm_path(cwd if cwd is not None else os.getcwd(), windows=True)
|
|
if not drive and rest.startswith("\\"):
|
|
p = ntpath.splitdrive(base)[0] + rest # rooted, no drive: cwd's drive
|
|
else:
|
|
p = ntpath.join(base, p)
|
|
p = ntpath.normpath(p)
|
|
drive, rest = ntpath.splitdrive(p)
|
|
if not drive or not rest.startswith("\\"):
|
|
raise ForksError("cannot normalise path", path=p)
|
|
if len(drive) == 2 and drive[1] == ":":
|
|
p = drive.upper() + rest
|
|
return p
|
|
|
|
|
|
def state_dir_for(run: str) -> str:
|
|
"""`reports/<x>-run.md` -> `reports/<x>-run-forks` (a sibling of the run log)."""
|
|
run_n = norm_path(run)
|
|
if not run_n.lower().endswith(".md"):
|
|
raise ForksError("--run must name the run log .md file", run=run_n)
|
|
return run_n[:-3] + "-forks"
|
|
|
|
|
|
STATE_SUBDIRS = ("jobs", "briefs", "notes", "results", "logs", "raw")
|
|
STATE_GITIGNORE = "raw/\n*.lock\n"
|
|
MCP_EMPTY = {"mcpServers": {}}
|
|
|
|
|
|
def registry_path(state: str) -> str:
|
|
return os.path.join(state, "registry.json")
|
|
|
|
|
|
def lock_path(state: str) -> str:
|
|
return os.path.join(state, "registry.lock")
|
|
|
|
|
|
def ledger_path(state: str) -> str:
|
|
return os.path.join(state, "ledger.jsonl")
|
|
|
|
|
|
def mcp_empty_path(state: str) -> str:
|
|
return os.path.join(state, "mcp-empty.json")
|
|
|
|
|
|
def ensure_state(state: str) -> None:
|
|
"""Create the state dir and its fixed files. Idempotent."""
|
|
os.makedirs(state, exist_ok=True)
|
|
for d in STATE_SUBDIRS:
|
|
os.makedirs(os.path.join(state, d), exist_ok=True)
|
|
gi = os.path.join(state, ".gitignore")
|
|
if not os.path.exists(gi):
|
|
atomic_write_bytes(gi, STATE_GITIGNORE.encode("utf-8"))
|
|
mcp = mcp_empty_path(state)
|
|
if not os.path.exists(mcp):
|
|
atomic_write_bytes(mcp, (json.dumps(MCP_EMPTY) + "\n").encode("utf-8"))
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
# Atomic files, retries, the registry lock (C3)
|
|
# --------------------------------------------------------------------------------------
|
|
|
|
|
|
def _retry(fn, tries: int = RETRY_TRIES, sleep_s: float = RETRY_SLEEP_S):
|
|
"""Call fn(); retry PermissionError (Windows sharing violations). Returns (value, attempts)."""
|
|
for attempt in range(1, tries + 1):
|
|
try:
|
|
return fn(), attempt
|
|
except PermissionError:
|
|
if attempt == tries:
|
|
raise
|
|
time.sleep(sleep_s)
|
|
raise AssertionError("unreachable")
|
|
|
|
|
|
def atomic_write_bytes(path: str, data: bytes) -> int:
|
|
"""Write to a temp file in the same dir, then `os.replace` with retries. Returns attempts."""
|
|
d = os.path.dirname(path) or "."
|
|
tmp = os.path.join(d, ".%s.%d.%s.tmp" % (os.path.basename(path), os.getpid(), uuid.uuid4().hex[:8]))
|
|
with open(tmp, "wb") as f:
|
|
f.write(data)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
try:
|
|
_, attempts = _retry(lambda: os.replace(tmp, path))
|
|
except BaseException:
|
|
with contextlib.suppress(OSError):
|
|
os.remove(tmp)
|
|
raise
|
|
return attempts
|
|
|
|
|
|
def write_json_atomic(path: str, obj) -> int:
|
|
return atomic_write_bytes(path, (json.dumps(obj, indent=1, sort_keys=True) + "\n").encode("utf-8"))
|
|
|
|
|
|
def read_bytes_retry(path: str) -> bytes:
|
|
def _read():
|
|
with open(path, "rb") as f:
|
|
return f.read()
|
|
return _retry(_read)[0]
|
|
|
|
|
|
def read_json(path: str):
|
|
"""Read a JSON file with PermissionError retries. Returns None if it does not exist."""
|
|
try:
|
|
data = read_bytes_retry(path)
|
|
except FileNotFoundError:
|
|
return None
|
|
try:
|
|
return json.loads(data.decode("utf-8"))
|
|
except ValueError as e:
|
|
raise ForksError("corrupt state file", path=path, detail=str(e))
|
|
|
|
|
|
class RegistryLock:
|
|
"""`registry.lock`: create-exclusive; a lock older than LOCK_STALE_S may be broken."""
|
|
|
|
def __init__(self, state: str, timeout: float = LOCK_WAIT_S):
|
|
self.path = lock_path(state)
|
|
self.timeout = timeout
|
|
self.token = uuid.uuid4().hex
|
|
self.held = False
|
|
self.broke_stale = False
|
|
|
|
def _content(self) -> bytes:
|
|
return json.dumps({"pid": os.getpid(), "ctime": proc_ctime(os.getpid()),
|
|
"at": _now(), "token": self.token}).encode("utf-8")
|
|
|
|
def _try_create(self) -> bool:
|
|
flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_BINARY", 0)
|
|
try:
|
|
fd = os.open(self.path, flags)
|
|
except FileExistsError:
|
|
return False
|
|
except PermissionError: # Windows: the old lock is being deleted
|
|
return False
|
|
try:
|
|
os.write(fd, self._content())
|
|
finally:
|
|
os.close(fd)
|
|
return True
|
|
|
|
def _read_lock(self):
|
|
"""(raw bytes, age in seconds) of the current lock, or (None, None) if gone."""
|
|
try:
|
|
raw = read_bytes_retry(self.path)
|
|
mtime = os.path.getmtime(self.path)
|
|
except (FileNotFoundError, PermissionError):
|
|
return None, None
|
|
at = None
|
|
try:
|
|
at = float(json.loads(raw.decode("utf-8")).get("at"))
|
|
except (ValueError, TypeError, AttributeError):
|
|
pass
|
|
if at is None:
|
|
at = mtime
|
|
return raw, _now() - at
|
|
|
|
def acquire(self) -> "RegistryLock":
|
|
deadline = time.monotonic() + self.timeout
|
|
while True:
|
|
if self._try_create():
|
|
self.held = True
|
|
return self
|
|
raw, age = self._read_lock()
|
|
if raw is not None and age is not None and age > LOCK_STALE_S:
|
|
# Break it only if it is still the same lock we judged stale.
|
|
again, _ = self._read_lock()
|
|
if again == raw:
|
|
with contextlib.suppress(FileNotFoundError, PermissionError):
|
|
os.remove(self.path)
|
|
self.broke_stale = True
|
|
_diag("broke stale registry lock (age %.1f s)" % age)
|
|
continue
|
|
if time.monotonic() >= deadline:
|
|
raise ForksError("registry locked", lock=self.path, waited_s=self.timeout)
|
|
time.sleep(LOCK_POLL_S)
|
|
|
|
def release(self) -> None:
|
|
if not self.held:
|
|
return
|
|
self.held = False
|
|
try:
|
|
raw = read_bytes_retry(self.path)
|
|
except FileNotFoundError:
|
|
return
|
|
try:
|
|
mine = json.loads(raw.decode("utf-8")).get("token") == self.token
|
|
except (ValueError, AttributeError):
|
|
mine = False
|
|
if mine:
|
|
with contextlib.suppress(FileNotFoundError):
|
|
_retry(lambda: os.remove(self.path))
|
|
else:
|
|
_diag("registry lock was taken over while held; not removing it")
|
|
|
|
def __enter__(self):
|
|
return self.acquire()
|
|
|
|
def __exit__(self, *exc):
|
|
self.release()
|
|
return False
|
|
|
|
|
|
def new_registry() -> dict:
|
|
return {"schema": SCHEMA, "config": None, "bases": {}, "jobs": {}, "run": {"created_at": _iso()}}
|
|
|
|
|
|
def read_registry(state: str) -> dict | None:
|
|
reg = read_json(registry_path(state))
|
|
if reg is None:
|
|
return None
|
|
if not isinstance(reg, dict) or reg.get("schema") != SCHEMA:
|
|
raise ForksError("unsupported registry schema", path=registry_path(state),
|
|
schema=(reg or {}).get("schema") if isinstance(reg, dict) else None,
|
|
expected=SCHEMA)
|
|
return reg
|
|
|
|
|
|
def write_registry(state: str, reg: dict) -> int:
|
|
"""Caller must hold RegistryLock(state)."""
|
|
reg["schema"] = SCHEMA
|
|
return write_json_atomic(registry_path(state), reg)
|
|
|
|
|
|
def ledger_append(state: str, line: dict) -> None:
|
|
"""Append one JSON line. Caller must hold RegistryLock(state) (C3c)."""
|
|
data = (json.dumps(line, sort_keys=True) + "\n").encode("utf-8")
|
|
|
|
def _append():
|
|
with open(ledger_path(state), "ab") as f:
|
|
f.write(data)
|
|
_retry(_append)
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
# Processes: pid + creation time (A19), subprocess helpers, git (C7)
|
|
# --------------------------------------------------------------------------------------
|
|
|
|
if IS_WINDOWS:
|
|
import ctypes
|
|
import ctypes.wintypes as _wt
|
|
|
|
_k32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
|
_k32.OpenProcess.argtypes = [_wt.DWORD, _wt.BOOL, _wt.DWORD]
|
|
_k32.OpenProcess.restype = _wt.HANDLE
|
|
_k32.GetProcessTimes.argtypes = [_wt.HANDLE] + [ctypes.POINTER(_wt.FILETIME)] * 4
|
|
_k32.GetProcessTimes.restype = _wt.BOOL
|
|
_k32.GetExitCodeProcess.argtypes = [_wt.HANDLE, ctypes.POINTER(_wt.DWORD)]
|
|
_k32.GetExitCodeProcess.restype = _wt.BOOL
|
|
_k32.CloseHandle.argtypes = [_wt.HANDLE]
|
|
_k32.CloseHandle.restype = _wt.BOOL
|
|
_PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
|
|
_PROCESS_TERMINATE = 0x0001
|
|
_STILL_ACTIVE = 259
|
|
_ERROR_ACCESS_DENIED = 5
|
|
_TH32CS_SNAPPROCESS = 0x00000002
|
|
_INVALID_HANDLE = _wt.HANDLE(-1).value
|
|
|
|
class _PE32(ctypes.Structure):
|
|
_fields_ = [("dwSize", _wt.DWORD), ("cntUsage", _wt.DWORD), ("th32ProcessID", _wt.DWORD),
|
|
("th32DefaultHeapID", ctypes.c_size_t), ("th32ModuleID", _wt.DWORD),
|
|
("cntThreads", _wt.DWORD), ("th32ParentProcessID", _wt.DWORD),
|
|
("pcPriClassBase", _wt.LONG), ("dwFlags", _wt.DWORD),
|
|
("szExeFile", _wt.WCHAR * 260)]
|
|
|
|
_k32.CreateToolhelp32Snapshot.argtypes = [_wt.DWORD, _wt.DWORD]
|
|
_k32.CreateToolhelp32Snapshot.restype = _wt.HANDLE
|
|
_k32.Process32FirstW.argtypes = [_wt.HANDLE, ctypes.POINTER(_PE32)]
|
|
_k32.Process32FirstW.restype = _wt.BOOL
|
|
_k32.Process32NextW.argtypes = [_wt.HANDLE, ctypes.POINTER(_PE32)]
|
|
_k32.Process32NextW.restype = _wt.BOOL
|
|
_k32.TerminateProcess.argtypes = [_wt.HANDLE, _wt.UINT]
|
|
_k32.TerminateProcess.restype = _wt.BOOL
|
|
|
|
|
|
def proc_ctime(pid: int) -> int | None:
|
|
"""Process creation time: Windows FILETIME ticks; POSIX /proc start time; else None."""
|
|
if IS_WINDOWS:
|
|
h = _k32.OpenProcess(_PROCESS_QUERY_LIMITED_INFORMATION, False, int(pid))
|
|
if not h:
|
|
return None
|
|
try:
|
|
c, e, k, u = _wt.FILETIME(), _wt.FILETIME(), _wt.FILETIME(), _wt.FILETIME()
|
|
if not _k32.GetProcessTimes(h, ctypes.byref(c), ctypes.byref(e), ctypes.byref(k),
|
|
ctypes.byref(u)):
|
|
return None
|
|
return (c.dwHighDateTime << 32) | c.dwLowDateTime
|
|
finally:
|
|
_k32.CloseHandle(h)
|
|
try:
|
|
with open("/proc/%d/stat" % int(pid), "rb") as f:
|
|
stat = f.read().decode("ascii", "replace")
|
|
return int(stat[stat.rindex(")") + 2:].split()[19])
|
|
except (OSError, ValueError, IndexError):
|
|
return None
|
|
|
|
|
|
def pid_record(pid: int) -> dict:
|
|
"""{"pid", "ctime"}: the pair every stored pid is compared by."""
|
|
return {"pid": int(pid), "ctime": proc_ctime(pid)}
|
|
|
|
|
|
def pid_alive(pid: int, ctime: int | None = None) -> bool:
|
|
"""True if `pid` runs and (when both are known) its creation time equals `ctime`."""
|
|
if pid is None or int(pid) <= 0:
|
|
return False
|
|
pid = int(pid)
|
|
if IS_WINDOWS:
|
|
h = _k32.OpenProcess(_PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
|
|
if not h:
|
|
# Exists but not openable: alive, creation time unverifiable.
|
|
return ctypes.get_last_error() == _ERROR_ACCESS_DENIED
|
|
try:
|
|
code = _wt.DWORD()
|
|
if not _k32.GetExitCodeProcess(h, ctypes.byref(code)) or code.value != _STILL_ACTIVE:
|
|
return False
|
|
finally:
|
|
_k32.CloseHandle(h)
|
|
else:
|
|
try:
|
|
os.kill(pid, 0)
|
|
except ProcessLookupError:
|
|
return False
|
|
except PermissionError:
|
|
pass
|
|
if ctime is not None:
|
|
now_ct = proc_ctime(pid)
|
|
if now_ct is not None and now_ct != ctime:
|
|
return False
|
|
return True
|
|
|
|
|
|
def record_alive(rec: dict | None) -> bool:
|
|
return bool(rec) and pid_alive(rec.get("pid"), rec.get("ctime"))
|
|
|
|
|
|
def process_table() -> list[tuple[int, int, str]]:
|
|
"""[(pid, ppid, name)] of every process. Windows: one Toolhelp32 snapshot (fast, no PowerShell)."""
|
|
out = []
|
|
if IS_WINDOWS:
|
|
snap = _k32.CreateToolhelp32Snapshot(_TH32CS_SNAPPROCESS, 0)
|
|
if not snap or snap == _INVALID_HANDLE:
|
|
return out
|
|
try:
|
|
e = _PE32()
|
|
e.dwSize = ctypes.sizeof(_PE32)
|
|
ok = _k32.Process32FirstW(snap, ctypes.byref(e))
|
|
while ok:
|
|
out.append((int(e.th32ProcessID), int(e.th32ParentProcessID), e.szExeFile))
|
|
ok = _k32.Process32NextW(snap, ctypes.byref(e))
|
|
finally:
|
|
_k32.CloseHandle(snap)
|
|
return out
|
|
try:
|
|
names = os.listdir("/proc")
|
|
except OSError:
|
|
return out
|
|
for n in names:
|
|
if not n.isdigit():
|
|
continue
|
|
try:
|
|
with open("/proc/%s/stat" % n, "rb") as f:
|
|
st = f.read().decode("ascii", "replace")
|
|
comm = st[st.index("(") + 1: st.rindex(")")]
|
|
ppid = int(st[st.rindex(")") + 2:].split()[1])
|
|
out.append((int(n), ppid, comm))
|
|
except (OSError, ValueError, IndexError):
|
|
continue
|
|
return out
|
|
|
|
|
|
def process_tree(rec: dict | 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."""
|
|
if not rec or not rec.get("pid"):
|
|
return []
|
|
kids, names = {}, {}
|
|
for pid, ppid, name in process_table():
|
|
names[pid] = name
|
|
if pid != ppid:
|
|
kids.setdefault(ppid, []).append(pid)
|
|
root = int(rec["pid"])
|
|
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}
|
|
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
|
|
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:
|
|
continue
|
|
seen.add(c)
|
|
out.append({"pid": c, "ctime": ct, "ppid": ppid, "name": names.get(c)})
|
|
stack.append((c, ct))
|
|
return out
|
|
|
|
|
|
def terminate(rec: dict) -> bool:
|
|
"""Kill one process identified by pid + creation time. True if it is gone or was signalled."""
|
|
if not record_alive(rec):
|
|
return True
|
|
pid = int(rec["pid"])
|
|
if IS_WINDOWS:
|
|
h = _k32.OpenProcess(_PROCESS_TERMINATE | _PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
|
|
if not h:
|
|
return False
|
|
try:
|
|
return bool(_k32.TerminateProcess(h, 1))
|
|
finally:
|
|
_k32.CloseHandle(h)
|
|
try:
|
|
os.kill(pid, signal.SIGKILL)
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def _taskkill() -> str:
|
|
p = os.path.join(os.environ.get("SystemRoot", r"C:\Windows"), "System32", "taskkill.exe")
|
|
return p if os.path.isfile(p) else "taskkill"
|
|
|
|
|
|
def kill_tree(rec: dict | None, extra=()) -> 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.
|
|
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):
|
|
key = (p["pid"], p["ctime"])
|
|
if key not in seen:
|
|
seen.add(key)
|
|
snap.append(p)
|
|
rc = None
|
|
for r in roots:
|
|
if not record_alive(r):
|
|
continue
|
|
if IS_WINDOWS:
|
|
try:
|
|
rc = _run([_taskkill(), "/T", "/F", "/PID", str(int(r["pid"]))], timeout=60).returncode
|
|
except (OSError, subprocess.SubprocessError) as e:
|
|
_diag("taskkill failed: %s" % e)
|
|
else:
|
|
with contextlib.suppress(OSError):
|
|
os.killpg(int(r["pid"]), signal.SIGKILL)
|
|
break # one tree kill of the first live root; the sweep covers the rest
|
|
deadline = time.monotonic() + KILL_VERIFY_S
|
|
while True:
|
|
survivors = [p for p in snap if record_alive(p)]
|
|
if not survivors or time.monotonic() >= deadline:
|
|
break
|
|
for p in survivors:
|
|
terminate(p)
|
|
time.sleep(0.1)
|
|
return {"snapshot": snap, "taskkill_rc": rc, "survivors": survivors}
|
|
|
|
|
|
def _run(args, cwd=None, timeout=60, env=None) -> subprocess.CompletedProcess:
|
|
"""Run a short helper process: stdout captured (never leaks to ours), no window."""
|
|
return subprocess.run(args, cwd=cwd, env=env, stdin=subprocess.DEVNULL, capture_output=True,
|
|
timeout=timeout, creationflags=_NOWIN)
|
|
|
|
|
|
def git(args, cwd=None, timeout=60) -> subprocess.CompletedProcess:
|
|
"""Every helper git call goes through here: `git --no-optional-locks …` (C7)."""
|
|
return _run(["git", "--no-optional-locks", *args], cwd=cwd, timeout=timeout)
|
|
|
|
|
|
def git_toplevel(cwd: str | None = None) -> str | None:
|
|
try:
|
|
r = git(["rev-parse", "--show-toplevel"], cwd=cwd)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return None
|
|
if r.returncode != 0:
|
|
return None
|
|
out = r.stdout.decode("utf-8", "replace").strip()
|
|
return norm_path(out) if out else None
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
# Env scrub
|
|
# --------------------------------------------------------------------------------------
|
|
|
|
|
|
def scrub_class(name: str) -> str | None:
|
|
"""'drop', 'pinned', 'keep' or None (not on any list)."""
|
|
n = name.upper()
|
|
if n in SCRUB_DROP or any(p.match(n) for p in SCRUB_DROP_PATTERNS):
|
|
return "drop"
|
|
if n in SCRUB_PINNED or n == SCRUB_PINNED_BINARY:
|
|
return "pinned"
|
|
if n in SCRUB_KEEP:
|
|
return "keep"
|
|
return None
|
|
|
|
|
|
def _venv_prefixes(environ) -> list[str]:
|
|
out = []
|
|
venv = environ.get("VIRTUAL_ENV")
|
|
if venv:
|
|
out.append(venv)
|
|
if os.path.normcase(os.path.abspath(sys.prefix)) != os.path.normcase(os.path.abspath(sys.base_prefix)):
|
|
out.append(sys.prefix)
|
|
return [os.path.normcase(os.path.normpath(os.path.abspath(x))) for x in out]
|
|
|
|
|
|
def _under(path: str, prefixes) -> bool:
|
|
try:
|
|
p = os.path.normcase(os.path.normpath(os.path.abspath(path)))
|
|
except ValueError:
|
|
return False
|
|
return any(p == px or p.startswith(px.rstrip("\\/") + os.sep) for px in prefixes)
|
|
|
|
|
|
def scrubbed_env(binary: str, environ=None) -> dict:
|
|
"""The child env for `claude`: host env minus the drop list, venv PATH entries removed,
|
|
pinned vars on top, `CLAUDE_CODE_EXECPATH` = the stored absolute binary path."""
|
|
if environ is None:
|
|
environ = os.environ
|
|
prefixes = _venv_prefixes(environ)
|
|
env = {}
|
|
for k, v in environ.items():
|
|
cls = scrub_class(k)
|
|
if cls in ("drop", "pinned"):
|
|
continue
|
|
if k.upper() == "PATH":
|
|
v = os.pathsep.join(e for e in v.split(os.pathsep) if e and not _under(e, prefixes))
|
|
env[k] = v
|
|
env.update(SCRUB_PINNED)
|
|
env[SCRUB_PINNED_BINARY] = binary
|
|
return env
|
|
|
|
|
|
def scrub_unknown(environ=None) -> list[str]:
|
|
"""Host CLAUDE_* / ANTHROPIC_* vars on neither list (kept, but worth a look)."""
|
|
if environ is None:
|
|
environ = os.environ
|
|
return sorted(k for k in environ
|
|
if k.upper().startswith(SCRUB_WATCH_PREFIXES) and scrub_class(k) is None)
|
|
|
|
|
|
def scrub_report(environ=None) -> dict:
|
|
return {
|
|
"drop": sorted(SCRUB_DROP),
|
|
"drop_patterns": [p.pattern for p in SCRUB_DROP_PATTERNS],
|
|
"drop_path_entries": "under VIRTUAL_ENV, or under sys.prefix when it differs from sys.base_prefix",
|
|
"keep": sorted(SCRUB_KEEP),
|
|
"keep_rule": "everything not dropped is kept",
|
|
"pinned": dict(SCRUB_PINNED, **{SCRUB_PINNED_BINARY: "<stored binary path>"}),
|
|
"unknown": scrub_unknown(environ),
|
|
}
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
# Binary, uv, python
|
|
# --------------------------------------------------------------------------------------
|
|
|
|
|
|
def resolve_binary(explicit: str | None = None, environ=None) -> tuple[str | None, str | None]:
|
|
"""(absolute path, how): `--binary`, else $CLAUDE_CODE_EXECPATH, else PATH `claude`."""
|
|
if environ is None:
|
|
environ = os.environ
|
|
if explicit:
|
|
return norm_path(explicit), "--binary"
|
|
ep = environ.get("CLAUDE_CODE_EXECPATH")
|
|
if ep:
|
|
return norm_path(ep), "CLAUDE_CODE_EXECPATH"
|
|
w = shutil.which("claude")
|
|
if w:
|
|
return norm_path(w), "PATH"
|
|
return None, None
|
|
|
|
|
|
def _binary_version(path: str) -> str | None:
|
|
try:
|
|
r = _run([path, "--version"], timeout=60)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return None
|
|
toks = r.stdout.decode("utf-8", "replace").split()
|
|
return toks[0] if r.returncode == 0 and toks else None
|
|
|
|
|
|
def binary_info(path: str) -> dict:
|
|
"""{"path", "version", "size", "mtime"} of an existing binary; raises ForksError otherwise."""
|
|
if not path or not os.path.isfile(path):
|
|
raise ForksError("claude binary not found", binary=path)
|
|
st = os.stat(path)
|
|
version = _binary_version(path)
|
|
if not version:
|
|
raise ForksError("claude binary did not report a version", binary=path)
|
|
return {"path": path, "version": version, "size": st.st_size, "mtime": st.st_mtime}
|
|
|
|
|
|
def _uv_report(environ=None) -> dict:
|
|
if environ is None:
|
|
environ = os.environ
|
|
out = {"path": None, "version": None}
|
|
w = shutil.which("uv")
|
|
if w:
|
|
out["path"] = norm_path(w)
|
|
try:
|
|
r = _run([w, "--version"], timeout=30)
|
|
toks = r.stdout.decode("utf-8", "replace").split()
|
|
out["version"] = toks[1] if len(toks) > 1 else (toks[0] if toks else None)
|
|
except (OSError, subprocess.SubprocessError) as e:
|
|
out["error"] = str(e)
|
|
if environ.get("SCRIPTS_UV"):
|
|
out["SCRIPTS_UV"] = environ["SCRIPTS_UV"]
|
|
return out
|
|
|
|
|
|
def _git_report() -> dict:
|
|
out = {"path": None, "version": None}
|
|
w = shutil.which("git")
|
|
if w:
|
|
out["path"] = norm_path(w)
|
|
try:
|
|
r = git(["--version"], timeout=30)
|
|
toks = r.stdout.decode("utf-8", "replace").split()
|
|
out["version"] = toks[2] if len(toks) > 2 else None
|
|
except (OSError, subprocess.SubprocessError) as e:
|
|
out["error"] = str(e)
|
|
return out
|
|
|
|
|
|
def script_path() -> str:
|
|
return norm_path(os.path.abspath(__file__))
|
|
|
|
|
|
def preamble_path() -> str:
|
|
return norm_path(os.path.join(os.path.dirname(os.path.abspath(__file__)), "preamble.md"))
|
|
|
|
|
|
def supervisor_interpreter() -> str:
|
|
"""C6: the base interpreter when `sys.executable` is a venv launcher."""
|
|
base = getattr(sys, "_base_executable", None)
|
|
return base if base and base != sys.executable else sys.executable
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
# C8: availableModels admission
|
|
# --------------------------------------------------------------------------------------
|
|
|
|
|
|
def user_settings_path() -> str:
|
|
if USER_SETTINGS_PATH:
|
|
return USER_SETTINGS_PATH
|
|
return os.path.join(os.path.expanduser("~"), ".claude", "settings.json")
|
|
|
|
|
|
def settings_files(cwd: str) -> list[str]:
|
|
return [user_settings_path(),
|
|
os.path.join(cwd, ".claude", "settings.json"),
|
|
os.path.join(cwd, ".claude", "settings.local.json")]
|
|
|
|
|
|
def model_family(model: str) -> str | None:
|
|
toks = model.replace(ONE_M_SUFFIX, "").split("-")
|
|
return toks[1] if len(toks) > 1 else None
|
|
|
|
|
|
def model_admission(model: str, cwd: str) -> dict:
|
|
"""Apply the P0.5 rule. Returns {"admitted", "lists": [{"file", "availableModels"}], "via"}."""
|
|
lists = []
|
|
for f in settings_files(cwd):
|
|
try:
|
|
data = read_json(f)
|
|
except ForksError as e:
|
|
_diag("ignoring unreadable settings file %s (%s)" % (f, e.payload.get("detail")))
|
|
continue
|
|
except OSError:
|
|
continue
|
|
if isinstance(data, dict) and isinstance(data.get("availableModels"), list):
|
|
lists.append({"file": f, "availableModels": [str(x) for x in data["availableModels"]]})
|
|
if not lists:
|
|
return {"admitted": True, "lists": [], "via": "no availableModels list"}
|
|
candidates = [("exact id", model)]
|
|
if model.endswith(ONE_M_SUFFIX):
|
|
candidates.append(("id without [1m]", model[: -len(ONE_M_SUFFIX)]))
|
|
fam = model_family(model)
|
|
if fam:
|
|
candidates.append(("family alias", fam))
|
|
for via, cand in candidates:
|
|
for entry in lists:
|
|
if cand in entry["availableModels"]:
|
|
return {"admitted": True, "lists": lists, "via": "%s %r in %s" % (via, cand, entry["file"])}
|
|
return {"admitted": False, "lists": lists, "via": None}
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
# Stream parsing, usage rule (P0.3), cache verdict (C5), warmth (C1)
|
|
# --------------------------------------------------------------------------------------
|
|
|
|
|
|
class StreamState:
|
|
"""What one `claude -p --output-format stream-json --verbose` stream told us so far."""
|
|
|
|
def __init__(self):
|
|
self.init = None
|
|
self.session_id = None
|
|
self.msg_ids: list = []
|
|
self._ids: set = set()
|
|
self.first_usage = None # usage of the first assistant event (cache verdict)
|
|
self.first_at = None # its arrival time (last_hit)
|
|
self.last_usage = None # usage of the last assistant event (base context)
|
|
self.last_assistant_at = None
|
|
self.last_event_at = None
|
|
self.result = None
|
|
self.refused = None # content of a model-restricted warning
|
|
self.events = 0
|
|
|
|
def feed(self, ev: dict, t: float | None) -> None:
|
|
if not isinstance(ev, dict):
|
|
return
|
|
self.events += 1
|
|
if t is not None:
|
|
self.last_event_at = t
|
|
ty = ev.get("type")
|
|
if ty == "system":
|
|
st = ev.get("subtype")
|
|
if st == "init" and self.init is None:
|
|
self.init = ev
|
|
self.session_id = ev.get("session_id") or self.session_id
|
|
elif (st == "informational" and ev.get("level") == "warning"
|
|
and "is restricted" in str(ev.get("content", ""))):
|
|
self.refused = str(ev.get("content"))
|
|
elif ty == "assistant":
|
|
m = ev.get("message") or {}
|
|
mid = m.get("id")
|
|
if mid not in self._ids:
|
|
self._ids.add(mid)
|
|
self.msg_ids.append(mid)
|
|
u = m.get("usage") or {}
|
|
if self.first_usage is None:
|
|
self.first_usage, self.first_at = u, t
|
|
self.last_usage, self.last_assistant_at = u, t
|
|
elif ty == "result":
|
|
self.result = ev
|
|
self.session_id = self.session_id or ev.get("session_id")
|
|
|
|
@property
|
|
def turns(self) -> int:
|
|
return len(self.msg_ids)
|
|
|
|
@classmethod
|
|
def from_file(cls, path: str) -> "StreamState":
|
|
"""Re-parse a raw stream file (event times unknown)."""
|
|
ss = cls()
|
|
try:
|
|
data = read_bytes_retry(path)
|
|
except OSError:
|
|
return ss
|
|
for line in data.splitlines():
|
|
try:
|
|
ss.feed(json.loads(line.decode("utf-8", "replace")), None)
|
|
except ValueError:
|
|
continue
|
|
return ss
|
|
|
|
|
|
def context_tokens(usage: dict | None) -> int:
|
|
"""Prompt-side tokens of one request: input + cache creation + cache read."""
|
|
if not usage:
|
|
return 0
|
|
return int(usage.get("input_tokens") or 0) + int(usage.get("cache_creation_input_tokens") or 0) \
|
|
+ int(usage.get("cache_read_input_tokens") or 0)
|
|
|
|
|
|
def runtime_fingerprint(init: dict | None) -> dict | None:
|
|
"""The runtime fields of a `system/init` event that must match the base's."""
|
|
if not init:
|
|
return None
|
|
tools = sorted(str(x) for x in (init.get("tools") or []))
|
|
plugins = sorted("%s@%s" % (p.get("name"), p.get("version") or "")
|
|
for p in (init.get("plugins") or []) if isinstance(p, dict))
|
|
agents = sorted(str(a) if not isinstance(a, dict) else str(a.get("name")) for a in (init.get("agents") or []))
|
|
return {"tools": _sha256_bytes(canonical_json(tools).encode("utf-8")),
|
|
"permissionMode": init.get("permissionMode"), "plugins": plugins, "agents": agents,
|
|
"apiKeySource": init.get("apiKeySource")}
|
|
|
|
|
|
def runtime_drift(rt: dict | None, ref: dict | None, launch: dict) -> list[str]:
|
|
if not rt:
|
|
return []
|
|
out = []
|
|
if ref:
|
|
out += [k for k in sorted(rt) if rt.get(k) != ref.get(k)]
|
|
elif rt.get("permissionMode") not in (None, launch.get("permission_mode")):
|
|
out.append("permissionMode")
|
|
return out
|
|
|
|
|
|
def usage_from_result(result: dict, before_mu: dict, cost_before: float) -> dict:
|
|
"""P0.3: the four columns and the 1h/5m split from `result.usage`; `models_used` = modelUsage
|
|
keys that grew against the session total before the call; cost delta against the session's
|
|
total before the call; `mismatch` when `result.usage` differs from the modelUsage delta."""
|
|
usage = result.get("usage") or {}
|
|
cc = usage.get("cache_creation") or {}
|
|
tokens = {
|
|
"input": int(usage.get("input_tokens") or 0),
|
|
"output": int(usage.get("output_tokens") or 0),
|
|
"cache_creation": int(usage.get("cache_creation_input_tokens") or 0),
|
|
"cache_creation_1h": int(cc.get("ephemeral_1h_input_tokens") or 0),
|
|
"cache_creation_5m": int(cc.get("ephemeral_5m_input_tokens") or 0),
|
|
"cache_read": int(usage.get("cache_read_input_tokens") or 0),
|
|
}
|
|
after = result.get("modelUsage") or {}
|
|
before_mu = before_mu or {}
|
|
delta = {}
|
|
for m, v in after.items():
|
|
b = before_mu.get(m) or {}
|
|
delta[m] = {f: int((v or {}).get(f) or 0) - int(b.get(f) or 0) for f in MU_FIELDS}
|
|
models_used = sorted(m for m, d in delta.items() if any(x > 0 for x in d.values()))
|
|
sums = {f: sum(d[f] for d in delta.values()) for f in MU_FIELDS}
|
|
mismatch = (sums["inputTokens"], sums["outputTokens"], sums["cacheCreationInputTokens"],
|
|
sums["cacheReadInputTokens"]) != (tokens["input"], tokens["output"],
|
|
tokens["cache_creation"], tokens["cache_read"])
|
|
total = result.get("total_cost_usd")
|
|
cost_delta = None if total is None else round(float(total) - float(cost_before or 0.0), 6)
|
|
return {"tokens": tokens, "models_used": models_used, "mismatch": mismatch,
|
|
"cost_delta": cost_delta, "total": total, "model_usage": after}
|
|
|
|
|
|
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:
|
|
return None
|
|
cr = int(first_usage.get("cache_read_input_tokens") or 0)
|
|
if base_context and cr >= CACHE_HIT_RATIO * float(base_context):
|
|
return "hit"
|
|
return "partial" if cr > 0 else "miss"
|
|
|
|
|
|
def base_warmth(base: dict, cfg: dict, now: float | None = None) -> dict:
|
|
"""C1 (Phase 1): warm_until = last_hit + ttl_s; margin = max(120, ttl_s/10)."""
|
|
ttl = float(cfg.get("ttl_s") or CONFIG_DEFAULTS["ttl_s"])
|
|
last = _ts((base or {}).get("last_hit"))
|
|
if last is None:
|
|
return {"base_state": "cold", "warm_until": None}
|
|
now = _now() if now is None else now
|
|
wu = last + ttl
|
|
margin = max(120.0, ttl / 10.0)
|
|
state = "warm" if now < wu - margin else ("expiring" if now < wu else "cold")
|
|
return {"base_state": state, "warm_until": _iso(wu)}
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
# Jobs: paths, templates, spawning, finalising (C3b, C3c)
|
|
# --------------------------------------------------------------------------------------
|
|
|
|
|
|
def job_file(state: str, job: str) -> str:
|
|
return os.path.join(state, "jobs", job + ".json")
|
|
|
|
|
|
def result_file(state: str, job: str) -> str:
|
|
return os.path.join(state, "results", job + ".json")
|
|
|
|
|
|
def raw_file(state: str, job: str, ext: str = ".jsonl") -> str:
|
|
return os.path.join(state, "raw", job + ext)
|
|
|
|
|
|
def _safe_name(s: str) -> str:
|
|
return re.sub(r"[^A-Za-z0-9._-]", "_", str(s))
|
|
|
|
|
|
def note_rel(block) -> str | None:
|
|
return None if block in (None, "") else "notes/%s.md" % _safe_name(block)
|
|
|
|
|
|
def _phase(v):
|
|
if v is None:
|
|
return None
|
|
return int(v) if re.fullmatch(r"\d+", str(v)) else str(v)
|
|
|
|
|
|
def render_template(name: str, values: dict) -> str:
|
|
path = os.path.join(os.path.dirname(os.path.abspath(__file__)), TEMPLATES_DIR, name)
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
text = f.read()
|
|
return string.Template(text).safe_substitute(values)
|
|
|
|
|
|
def _powershell() -> str:
|
|
w = shutil.which("powershell")
|
|
if w:
|
|
return w
|
|
return os.path.join(os.environ.get("SystemRoot", r"C:\Windows"), "System32", "WindowsPowerShell",
|
|
"v1.0", "powershell.exe")
|
|
|
|
|
|
def spawn_supervisor(state: str, run: str, job: str) -> dict:
|
|
"""Start `_supervise` detached (C6). Windows: CIM `Win32_Process.Create` with a hidden window,
|
|
through a `powershell` that itself has CREATE_NO_WINDOW. POSIX: a new session.
|
|
Returns {"ok", "pid", "ctime", "method"} or {"ok": False, "detail"}."""
|
|
argv = [supervisor_interpreter(), "-I", script_path(), "_supervise", "--run", run, "--job", job]
|
|
if IS_WINDOWS:
|
|
env = dict(os.environ)
|
|
env["SPIRE_FORKS_CMD"] = subprocess.list2cmdline(argv)
|
|
env["SPIRE_FORKS_CWD"] = state
|
|
try:
|
|
r = _run([_powershell(), "-NoProfile", "-NonInteractive", "-Command", POWERSHELL_CIM],
|
|
env=env, timeout=120)
|
|
except (OSError, subprocess.SubprocessError) as e:
|
|
return {"ok": False, "method": SPAWN_METHOD, "detail": "powershell: %s" % e}
|
|
toks = r.stdout.decode("utf-8", "replace").split()
|
|
try:
|
|
rv, pid = int(toks[-2]), int(toks[-1])
|
|
except (ValueError, IndexError):
|
|
return {"ok": False, "method": SPAWN_METHOD,
|
|
"detail": "unparsable CIM output: %r / %r" % (r.stdout[-200:], r.stderr[-400:])}
|
|
if rv != 0:
|
|
return {"ok": False, "method": SPAWN_METHOD, "detail": "Win32_Process.Create returned %d" % rv}
|
|
return {"ok": True, "method": SPAWN_METHOD, "pid": pid, "ctime": proc_ctime(pid)}
|
|
log = open(raw_file(state, job, ".sup.log"), "ab")
|
|
try:
|
|
p = subprocess.Popen(argv, cwd=state, stdin=subprocess.DEVNULL, stdout=log, stderr=log,
|
|
start_new_session=True, close_fds=True)
|
|
except OSError as e:
|
|
return {"ok": False, "method": SPAWN_METHOD, "detail": str(e)}
|
|
finally:
|
|
log.close()
|
|
return {"ok": True, "method": SPAWN_METHOD, "pid": p.pid, "ctime": proc_ctime(p.pid)}
|
|
|
|
|
|
def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint: str, flags, reason,
|
|
detail, exit_code, wall_s, base_commit, now: float):
|
|
"""Build the result, the ledger line, and apply the registry updates for one finished job.
|
|
`hint`: "exited" (decide ok/failed from the stream), or a final status given by the caller."""
|
|
cfg = reg["config"]
|
|
launch = cfg["launch"]
|
|
kind = entry["kind"]
|
|
base = reg["bases"].get(entry.get("base")) if entry.get("base") else None
|
|
res = ss.result if ss else None
|
|
u = usage_from_result(res, entry.get("model_usage_before") or {}, entry.get("cost_before") or 0.0) \
|
|
if res else None
|
|
flags = list(flags or [])
|
|
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)):
|
|
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 res.get("is_error"):
|
|
status, reason, detail = "failed", "is_error", str(res.get("subtype"))
|
|
elif exit_code:
|
|
status, reason, detail = "failed", "exit_code", "exit %s" % exit_code
|
|
else:
|
|
status, reason, detail = "ok", None, None
|
|
kr = entry.get("kill_requested")
|
|
if kr and status != "ok":
|
|
status, reason, detail = "killed", kr.get("by"), "kill requested"
|
|
if kr.get("by") == "expire":
|
|
flags.append("expired")
|
|
else:
|
|
status = hint
|
|
if u and u["mismatch"]:
|
|
flags.append("usage_mismatch")
|
|
denials = len((res or {}).get("permission_denials") or [])
|
|
if denials:
|
|
flags.append("needs_attention")
|
|
cache = cache_verdict(ss.first_usage if ss else None, entry.get("base_context")) \
|
|
if kind in ("fork", "continue") else None
|
|
rt = runtime_fingerprint(ss.init) if ss else None
|
|
drift = list(entry.get("launch_drift") or [])
|
|
ref = (base or {}).get("runtime_fingerprint") if kind != "train" else None
|
|
drift += runtime_drift(rt, ref, launch)
|
|
drift = sorted(set(drift))
|
|
fingerprint = "ok" if not drift else "drift:" + ",".join(drift)
|
|
base_state = entry.get("base_state_at_launch")
|
|
if kind in ("fork", "continue") and (base_state == "cold" or (cache not in (None, "hit")
|
|
and fingerprint == "ok")):
|
|
flags.append("cold_start")
|
|
flags = _uniq(flags)
|
|
turns = int(res.get("num_turns") or 0) if res else (ss.turns if ss else 0)
|
|
note = note_rel(entry.get("block")) if kind != "train" else None
|
|
if note and not os.path.isfile(os.path.join(state, note)):
|
|
note = None
|
|
session = (ss.session_id if ss and ss.session_id else None) or entry.get("session")
|
|
tokens = u["tokens"] if u else {k: None for k in TOKEN_KEYS}
|
|
ended = _iso(now)
|
|
result = {
|
|
"job": entry["job"], "kind": kind, "base": entry.get("base"), "block": entry.get("block"),
|
|
"phase": entry.get("phase"), "status": status, "flags": flags, "fingerprint": fingerprint,
|
|
"session": session, "parent_session": entry.get("parent_session"), "models_used": models_used,
|
|
"tokens": tokens, "cost_delta_usd": u["cost_delta"] if u else None, "cache": cache,
|
|
"base_state": base_state, "wall_s": None if wall_s is None else round(float(wall_s), 1),
|
|
"turns": turns, "denials": denials, "note": note, "raw": "raw/%s.jsonl" % entry["job"],
|
|
"paths": entry.get("paths"), "reason": reason, "model_requested": entry.get("model_requested"),
|
|
}
|
|
ledger = {
|
|
"job": entry["job"], "kind": kind, "phase": entry.get("phase"), "block": entry.get("block"),
|
|
"base": entry.get("base"), "session": session, "parent_session": entry.get("parent_session"),
|
|
"started": entry.get("started_at") or entry.get("launched_at") or entry.get("created_at"),
|
|
"ended": ended, "wall_s": result["wall_s"], "binary": entry.get("binary_version"),
|
|
"model_requested": entry.get("model_requested"), "models_used": models_used, "tokens": tokens,
|
|
"cost_delta_usd": result["cost_delta_usd"], "cost_cumulative_usd": u["total"] if u else None,
|
|
"turns": turns, "cache": cache, "base_state": base_state, "fingerprint": fingerprint,
|
|
"denials": denials, "status": status, "flags": flags, "reason": reason, "note": note,
|
|
}
|
|
# registry updates
|
|
entry.update({"status": status, "flags": flags, "reason": reason, "reason_detail": detail,
|
|
"ended_at": ended, "session": session, "turns": turns, "exit_code": exit_code,
|
|
"fingerprint": fingerprint, "cache": cache})
|
|
if ss and ss.first_at:
|
|
entry["first_event_at"] = _iso(ss.first_at)
|
|
if u:
|
|
entry["cost_cum_usd"], entry["model_usage_cum"] = u["total"], u["model_usage"]
|
|
holder = entry if kind == "fork" else reg["jobs"].get(entry.get("continues") or "")
|
|
if kind in ("fork", "continue") and holder is not None:
|
|
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
|
|
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"]:
|
|
if status == "ok" and base.get("status") == "training":
|
|
base.update({"status": "active", "session": session, "base_commit": base_commit,
|
|
"context_tokens": context_tokens(ss.last_usage if ss else None),
|
|
"training_cost_usd": u["cost_delta"] if u else None,
|
|
"cost_cum_usd": u["total"] if u else None,
|
|
"model_usage_cum": u["model_usage"] if u else {},
|
|
"runtime_fingerprint": rt, "trained_at": ended})
|
|
elif base.get("status") == "training":
|
|
base["status"] = "failed"
|
|
return result, ledger
|
|
|
|
|
|
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/<job>.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."""
|
|
with RegistryLock(state):
|
|
reg = read_registry(state)
|
|
entry = (reg or {}).get("jobs", {}).get(job)
|
|
if not entry or entry.get("status") not in RUNNING:
|
|
return None
|
|
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)
|
|
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."""
|
|
with RegistryLock(state):
|
|
reg = read_registry(state)
|
|
base = (reg or {}).get("bases", {}).get(base_name)
|
|
if base is None or (_ts(base.get("last_hit")) or 0) >= t:
|
|
return
|
|
base["last_hit"] = _iso(t)
|
|
write_registry(state, reg)
|
|
|
|
|
|
def touch_orchestrator(state: str) -> None:
|
|
"""N1: record that the orchestrator called us (wait, fork, continue, train, list)."""
|
|
with RegistryLock(state):
|
|
reg = read_registry(state)
|
|
if reg is None:
|
|
return
|
|
reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso()
|
|
write_registry(state, reg)
|
|
|
|
|
|
def _job_runtime(state: str, job: str) -> dict:
|
|
try:
|
|
return read_json(job_file(state, job)) or {}
|
|
except ForksError:
|
|
return {}
|
|
|
|
|
|
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,
|
|
is killed with it."""
|
|
reg = read_registry(state)
|
|
lost = []
|
|
now = _now()
|
|
for job, e in sorted((reg or {}).get("jobs", {}).items()):
|
|
if e.get("status") not in RUNNING:
|
|
continue
|
|
rt = _job_runtime(state, job)
|
|
hb = _ts(rt.get("heartbeat_at")) or _ts(e.get("launched_at")) or _ts(e.get("created_at"))
|
|
if hb is not None and now - hb <= LOST_AFTER_S:
|
|
continue
|
|
sup = rt.get("supervisor") or e.get("supervisor")
|
|
if sup and record_alive(sup):
|
|
continue
|
|
if rt.get("claude"):
|
|
kill_tree(rt["claude"])
|
|
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",
|
|
detail="heartbeat %s, supervisor %s dead" % (rt.get("heartbeat_at"), sup),
|
|
wall_s=(hb - start) if (hb and start) else None)
|
|
if res is not None:
|
|
lost.append(job)
|
|
return lost
|
|
|
|
|
|
def kill_job(state: str, job: str, by: str) -> dict:
|
|
"""`kill` (by="kill") and `expire` (by="expire"): snapshot the supervisor's tree, taskkill it,
|
|
sweep, verify, finalise as `killed`."""
|
|
with RegistryLock(state):
|
|
reg = read_registry(state)
|
|
entry = reg["jobs"].get(job)
|
|
if entry is None:
|
|
raise ForksError("unknown job", job=job)
|
|
if entry.get("status") not in RUNNING:
|
|
return {"job": job, "killed": False, "status": entry.get("status"), "ok": True}
|
|
entry["kill_requested"] = {"by": by, "at": _iso()}
|
|
write_registry(state, reg)
|
|
rt = _job_runtime(state, job)
|
|
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 [])
|
|
flags = list(rt.get("flags") or [])
|
|
if by == "expire":
|
|
flags.append("expired")
|
|
start = _ts(rt.get("started_at")) or _ts(entry.get("launched_at"))
|
|
res = finalize_job(state, job, ss=StreamState.from_file(raw_file(state, job)), hint="killed",
|
|
flags=flags, reason=by, detail="killed by `%s`" % by,
|
|
wall_s=(_now() - start) if start else None)
|
|
if res is not None and not sup_alive and rt:
|
|
rt.update({"status": "killed", "ended_at": _iso()})
|
|
with contextlib.suppress(OSError):
|
|
write_json_atomic(job_file(state, job), rt)
|
|
final = read_registry(state)["jobs"][job].get("status")
|
|
return {"job": job, "killed": True, "status": final, "ok": not k["survivors"],
|
|
"snapshot": k["snapshot"], "survivors": k["survivors"], "taskkill_rc": k["taskkill_rc"]}
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
# Argument parsing (argparse that never prints to stdout or exits on its own)
|
|
# --------------------------------------------------------------------------------------
|
|
|
|
|
|
class _Parser(argparse.ArgumentParser):
|
|
def error(self, message):
|
|
raise ForksError("usage", detail=message, usage=self.format_usage().strip())
|
|
|
|
def print_help(self, file=None):
|
|
raise _HelpRequested(self.format_help())
|
|
|
|
def exit(self, status=0, message=None):
|
|
if message:
|
|
_diag(message.strip())
|
|
raise ForksError("usage", detail=(message or "").strip() or "argument parser exit")
|
|
|
|
|
|
def _split_top_level_commas(s: str) -> list[str]:
|
|
"""Split on commas outside parentheses: `Bash(a,b),Read` -> [`Bash(a,b)`, `Read`]."""
|
|
out, depth, cur = [], 0, []
|
|
for ch in s:
|
|
if ch == "(":
|
|
depth += 1
|
|
elif ch == ")":
|
|
depth = max(0, depth - 1)
|
|
if ch == "," and depth == 0:
|
|
out.append("".join(cur).strip())
|
|
cur = []
|
|
else:
|
|
cur.append(ch)
|
|
out.append("".join(cur).strip())
|
|
return [x for x in out if x]
|
|
|
|
|
|
def _tool_list(values) -> list[str] | None:
|
|
if values is None:
|
|
return None
|
|
out = []
|
|
for v in values:
|
|
out.extend(_split_top_level_commas(v))
|
|
return out
|
|
|
|
|
|
def _pos_int(s: str) -> int:
|
|
try:
|
|
v = int(s)
|
|
except ValueError:
|
|
raise argparse.ArgumentTypeError("not an integer: %r" % s)
|
|
if v <= 0:
|
|
raise argparse.ArgumentTypeError("must be > 0: %r" % s)
|
|
return v
|
|
|
|
|
|
def _pos_float(s: str) -> float:
|
|
try:
|
|
v = float(s)
|
|
except ValueError:
|
|
raise argparse.ArgumentTypeError("not a number: %r" % s)
|
|
if not v > 0:
|
|
raise argparse.ArgumentTypeError("must be > 0: %r" % s)
|
|
return v
|
|
|
|
|
|
COMMANDS = ("where", "init", "train", "fork", "continue", "wait", "list", "expire", "kill",
|
|
"rebase", "usage", "tick", "_supervise")
|
|
|
|
|
|
def build_parser() -> _Parser:
|
|
p = _Parser(prog="forks.py", description="spire:forks helper (one JSON object on stdout).")
|
|
sub = p.add_subparsers(dest="command", metavar="command", parser_class=_Parser)
|
|
sub.required = True
|
|
|
|
def cmd(name, help_text):
|
|
sp = (sub.add_parser(name, help=help_text, description=help_text) if help_text
|
|
else sub.add_parser(name))
|
|
if name != "where":
|
|
sp.add_argument("--run", required=True, help="path of the run log .md")
|
|
return sp
|
|
|
|
w = cmd("where", "Preflight report: script, python, uv, binary, git, spawn method, env scrub.")
|
|
w.add_argument("--binary", help="claude binary to report instead of the default resolution")
|
|
|
|
i = cmd("init", "Create the state dir and the launch config (idempotent).")
|
|
i.add_argument("--model", required=True, help="full model id, e.g. claude-opus-5-5[1m]")
|
|
i.add_argument("--effort", default="high", choices=EFFORTS)
|
|
i.add_argument("--permission-mode", default="dontAsk")
|
|
i.add_argument("--tools", default=DEFAULT_TOOLS, help="comma-separated tool names")
|
|
i.add_argument("--allowed-tools", nargs="+", action="extend", default=None,
|
|
help="allow rules (default: the tools)")
|
|
i.add_argument("--disallowed-tools", nargs="+", action="extend", default=None,
|
|
help="deny rules (default: git push/rebase/reset --hard, claude)")
|
|
i.add_argument("--agent", help="project agent name (<cwd>/.claude/agents/<name>.md)")
|
|
i.add_argument("--cwd", help="working dir of base and forks (default: git toplevel)")
|
|
i.add_argument("--binary", help="claude binary (default: $CLAUDE_CODE_EXECPATH, else PATH)")
|
|
i.add_argument("--parallel-cap", type=_pos_int, default=CONFIG_DEFAULTS["parallel_cap"])
|
|
i.add_argument("--orphan-after", type=_pos_int, default=CONFIG_DEFAULTS["orphan_after"])
|
|
i.add_argument("--pings", choices=("on", "off"), default=CONFIG_DEFAULTS["pings"])
|
|
i.add_argument("--ttl-s", type=_pos_int, default=CONFIG_DEFAULTS["ttl_s"])
|
|
i.add_argument("--max-minutes", type=_pos_int, default=CONFIG_DEFAULTS["max_minutes"])
|
|
i.add_argument("--stall-after-s", type=_pos_int, default=CONFIG_DEFAULTS["stall_after_s"])
|
|
for k, v in THRESHOLD_DEFAULTS.items():
|
|
i.add_argument("--" + k.replace("_", "-"), type=_pos_int, default=v)
|
|
|
|
t = cmd("train", "Train a new base (headless session).")
|
|
t.add_argument("--base", required=True)
|
|
t.add_argument("--brief", required=True)
|
|
t.add_argument("--block")
|
|
t.add_argument("--phase")
|
|
|
|
def launch_overrides(sp):
|
|
# Accepted only so that they can be refused with a clear message (plan 4.3).
|
|
g = sp.add_argument_group("refused launch fields (fixed at init; read from the base record)")
|
|
for name in LAUNCH_FIELDS:
|
|
flag = "--" + name.replace("_", "-")
|
|
if name in ("allowed_tools", "disallowed_tools"):
|
|
g.add_argument(flag, nargs="+", default=None, help=argparse.SUPPRESS)
|
|
else:
|
|
g.add_argument(flag, default=None, help=argparse.SUPPRESS)
|
|
|
|
f = cmd("fork", "Fork a base for one block.")
|
|
f.add_argument("--base", required=True)
|
|
f.add_argument("--block", required=True)
|
|
f.add_argument("--phase")
|
|
f.add_argument("--brief", required=True)
|
|
f.add_argument("--mode", choices=("sequential", "parallel"), default="sequential")
|
|
f.add_argument("--paths", nargs="+", action="extend", default=None)
|
|
f.add_argument("--max-minutes", type=_pos_float, help="wall-time limit (default: init's max_minutes)")
|
|
f.add_argument("--max-budget-usd", type=_pos_float)
|
|
f.add_argument("--force", action="store_true", help="fork a base that is not active")
|
|
launch_overrides(f)
|
|
|
|
c = cmd("continue", "One more turn in a fork (resume without --fork-session).")
|
|
c.add_argument("--job", required=True)
|
|
c.add_argument("--brief", required=True)
|
|
c.add_argument("--max-minutes", type=_pos_float)
|
|
launch_overrides(c)
|
|
|
|
wt = cmd("wait", "Block until jobs finish or the timeout (default 540, max 570).")
|
|
wt.add_argument("jobs", nargs="*")
|
|
wt.add_argument("--all", action="store_true", help="every job not yet settled by a wait")
|
|
wt.add_argument("--any", action="store_true", help="return when any one of them has finished")
|
|
wt.add_argument("--timeout", type=_pos_int, default=WAIT_DEFAULT_S)
|
|
|
|
cmd("list", "Show every base and job.")
|
|
|
|
e = cmd("expire", "Retire a base (or all); running jobs of it are killed.")
|
|
eg = e.add_mutually_exclusive_group(required=True)
|
|
eg.add_argument("--base")
|
|
eg.add_argument("--all", action="store_true")
|
|
|
|
k = cmd("kill", "Kill a job's supervisor and claude process tree.")
|
|
k.add_argument("--job", required=True)
|
|
|
|
s = cmd("_supervise", None) # internal
|
|
s.add_argument("--job", required=True)
|
|
|
|
for name, text in (("rebase", "Rebase a base onto the current state (Phase 2)."),
|
|
("usage", "Aggregate the ledger (Phase 2)."),
|
|
("tick", "One keep-warm pass (Phase 2).")):
|
|
cmd(name, text) # arguments arrive with Phase 2; until then extra args are tolerated
|
|
return p
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
# Commands
|
|
# --------------------------------------------------------------------------------------
|
|
|
|
|
|
def cmd_where(args) -> tuple[dict, int]:
|
|
reasons = []
|
|
py_ok = sys.version_info >= (3, 10)
|
|
if not py_ok:
|
|
reasons.append("python < 3.10")
|
|
py = {"version": platform.python_version(), "executable": sys.executable,
|
|
"base_executable": getattr(sys, "_base_executable", sys.executable),
|
|
"supervisor_interpreter": supervisor_interpreter(), "ok": py_ok}
|
|
path, how = resolve_binary(args.binary)
|
|
binary = {"path": path, "how": how}
|
|
if path is None:
|
|
reasons.append("no claude binary (no --binary, no $CLAUDE_CODE_EXECPATH, none on PATH)")
|
|
else:
|
|
try:
|
|
binary.update(binary_info(path))
|
|
except ForksError as e:
|
|
binary["error"] = e.payload["error"]
|
|
reasons.append("%s: %s" % (e.payload["error"], path))
|
|
out = {
|
|
"script": script_path(),
|
|
"python": py,
|
|
"uv": _uv_report(),
|
|
"binary": binary,
|
|
"git": _git_report(),
|
|
"spawn": SPAWN_METHOD,
|
|
"scrub": scrub_report(),
|
|
"platform": sys.platform,
|
|
"ok": not reasons,
|
|
}
|
|
if reasons:
|
|
out["reason"] = "; ".join(reasons)
|
|
out["error"] = out["reason"]
|
|
return out, EXIT_USAGE
|
|
return out, EXIT_OK
|
|
|
|
|
|
def _resolve_cwd(arg: str | None) -> str:
|
|
if arg:
|
|
cwd = norm_path(arg)
|
|
else:
|
|
cwd = git_toplevel()
|
|
if not cwd:
|
|
raise ForksError("no --cwd given and the current directory is not inside a git repo")
|
|
if not os.path.isdir(cwd):
|
|
raise ForksError("--cwd is not a directory", cwd=cwd)
|
|
return cwd
|
|
|
|
|
|
def build_init_config(args, state: str) -> dict:
|
|
"""Validate the launch fields and build the config block. Writes nothing."""
|
|
model = args.model
|
|
if not MODEL_RE.match(model or ""):
|
|
raise ForksError("model must be a full id (aliases are refused)", model=model,
|
|
pattern=MODEL_RE.pattern)
|
|
mode = args.permission_mode
|
|
if mode in REFUSED_PERMISSION_MODES:
|
|
raise ForksError("permission mode refused", permission_mode=mode,
|
|
detail="plan mode swaps models; use dontAsk")
|
|
if mode not in PERMISSION_MODES:
|
|
raise ForksError("unknown permission mode", permission_mode=mode, allowed=list(PERMISSION_MODES))
|
|
tools = _tool_list([args.tools])
|
|
if not tools:
|
|
raise ForksError("--tools is empty")
|
|
allowed = _tool_list(args.allowed_tools)
|
|
if allowed is None:
|
|
allowed = list(tools)
|
|
disallowed = _tool_list(args.disallowed_tools)
|
|
if disallowed is None:
|
|
disallowed = list(DEFAULT_DISALLOWED)
|
|
|
|
cwd = _resolve_cwd(args.cwd)
|
|
|
|
adm = model_admission(model, cwd)
|
|
if not adm["admitted"]:
|
|
files = [x["file"] for x in adm["lists"]]
|
|
raise ForksError(
|
|
"model not admitted by availableModels", model=model, files=files,
|
|
message=("Model %s is not admitted by the availableModels list(s) in: %s. The user must "
|
|
"add the id (or its family alias) to one of these files; forks never writes "
|
|
"settings." % (model, ", ".join(files))))
|
|
|
|
bpath, how = resolve_binary(args.binary)
|
|
if bpath is None:
|
|
raise ForksError("no claude binary (no --binary, no $CLAUDE_CODE_EXECPATH, none on PATH)")
|
|
binfo = binary_info(bpath)
|
|
|
|
pre = preamble_path()
|
|
if not os.path.isfile(pre):
|
|
raise ForksError("preamble.md missing next to forks.py", path=pre)
|
|
|
|
agent = None
|
|
if args.agent:
|
|
name = args.agent
|
|
if not re.match(r"^[A-Za-z0-9_.:-]+$", name) or name in (".", ".."):
|
|
raise ForksError("bad agent name", agent=name)
|
|
apath = os.path.join(cwd, ".claude", "agents", name + ".md")
|
|
if not os.path.isfile(apath):
|
|
raise ForksError("agent file not found", agent=name, path=apath)
|
|
agent = {"name": name, "path": apath, "sha256": _sha256_file(apath)}
|
|
|
|
launch = {
|
|
"model": model,
|
|
"effort": args.effort,
|
|
"permission_mode": mode,
|
|
"tools": tools,
|
|
"allowed_tools": allowed,
|
|
"disallowed_tools": disallowed,
|
|
"mcp": {"strict": True, "config": mcp_empty_path(state)},
|
|
"append_system_prompt": {"path": pre, "sha256": _sha256_file(pre)},
|
|
"agent": agent,
|
|
"cwd": cwd,
|
|
"binary": binfo,
|
|
"env": {"pinned": dict(ENV_PINNED_LAUNCH)},
|
|
}
|
|
cfg = {
|
|
"launch": launch,
|
|
"fingerprint": fingerprint_hash(launch),
|
|
"binary_resolved_by": how,
|
|
"parallel_cap": args.parallel_cap,
|
|
"orphan_after": args.orphan_after,
|
|
"pings": args.pings,
|
|
"ttl_s": args.ttl_s,
|
|
"max_minutes": args.max_minutes,
|
|
"stall_after_s": args.stall_after_s,
|
|
"thresholds": {k: getattr(args, k) for k in THRESHOLD_DEFAULTS},
|
|
"model_admission": adm["via"],
|
|
}
|
|
return cfg
|
|
|
|
|
|
def _diff_fields(a, b, prefix="") -> list[str]:
|
|
if isinstance(a, dict) and isinstance(b, dict):
|
|
out = []
|
|
for k in sorted(set(a) | set(b)):
|
|
out += _diff_fields(a.get(k), b.get(k), prefix + k + ".")
|
|
return out
|
|
return [] if a == b else [prefix.rstrip(".")]
|
|
|
|
|
|
def cmd_init(args) -> tuple[dict, int]:
|
|
state = state_dir_for(args.run)
|
|
cfg = build_init_config(args, state) # all validation before any write
|
|
ensure_state(state)
|
|
with RegistryLock(state):
|
|
reg = read_registry(state)
|
|
created = reg is None
|
|
if created:
|
|
reg = new_registry()
|
|
old = reg.get("config")
|
|
changed_fields = [] if old is None else _diff_fields(old, cfg)
|
|
# `model_admission` and `binary_resolved_by` describe how, not what: not a change.
|
|
changed_fields = [f for f in changed_fields
|
|
if f not in ("model_admission", "binary_resolved_by")]
|
|
if old is not None and changed_fields and reg.get("bases"):
|
|
raise ForksError("init values differ and a base exists; train a new run instead",
|
|
fields=changed_fields, bases=sorted(reg["bases"]), state=state)
|
|
if old is None or changed_fields:
|
|
reg["config"] = cfg
|
|
reg.setdefault("run", {})["init_at"] = _iso()
|
|
write_registry(state, reg)
|
|
else:
|
|
cfg = old
|
|
return {"ok": True, "state": state, "registry": registry_path(state), "created": created,
|
|
"changed": bool(old is not None and changed_fields), "changed_fields": changed_fields,
|
|
"config": cfg}, EXIT_OK
|
|
|
|
|
|
def _not_implemented(block=None, phase=None):
|
|
def f(args):
|
|
out = {"error": "not implemented", "command": args.command}
|
|
if block:
|
|
out["block"] = block
|
|
if phase:
|
|
out["phase"] = phase
|
|
return out, EXIT_USAGE
|
|
return f
|
|
|
|
|
|
# --- launching: train, fork, continue -------------------------------------------------
|
|
|
|
|
|
def _load_run(args) -> tuple[str, dict]:
|
|
state = state_dir_for(args.run)
|
|
reg = read_registry(state) if os.path.isdir(state) else None
|
|
if not reg or not reg.get("config"):
|
|
raise ForksError("no launch config for this run; call init first", state=state)
|
|
return state, reg
|
|
|
|
|
|
def _refuse_overrides(args) -> None:
|
|
given = [n for n in LAUNCH_FIELDS if getattr(args, n, None) is not None]
|
|
if given:
|
|
raise ForksError("launch fields are fixed at init; %s reads them from the base record and "
|
|
"cannot override them (a different value needs a new base)" % args.command,
|
|
fields=given, command=args.command)
|
|
|
|
|
|
def _read_brief(path: str) -> tuple[str, str]:
|
|
p = norm_path(path)
|
|
try:
|
|
with open(p, "r", encoding="utf-8") as f:
|
|
return p, f.read()
|
|
except OSError as e:
|
|
raise ForksError("cannot read brief", brief=p, detail=str(e))
|
|
|
|
|
|
def _check_binary(ref: dict, launch: dict) -> tuple[dict, list[str]]:
|
|
"""Launch-time binary re-check (version, size, mtime) against `ref`; also the hashed files."""
|
|
path = launch["binary"]["path"]
|
|
if not os.path.isfile(path):
|
|
raise ForksError("claude binary missing (the stored path is gone); the base cannot be "
|
|
"launched with another binary", binary=path, reason="binary_missing")
|
|
info = binary_info(path)
|
|
drift = [] if all(info.get(k) == (ref or {}).get(k) for k in ("version", "size", "mtime")) else ["binary"]
|
|
asp = launch["append_system_prompt"]
|
|
try:
|
|
if _sha256_file(asp["path"]) != asp["sha256"]:
|
|
drift.append("append_system_prompt")
|
|
except OSError:
|
|
drift.append("append_system_prompt")
|
|
ag = launch.get("agent")
|
|
if ag:
|
|
try:
|
|
if _sha256_file(ag["path"]) != ag["sha256"]:
|
|
drift.append("agent")
|
|
except OSError:
|
|
drift.append("agent")
|
|
return info, drift
|
|
|
|
|
|
def claude_argv(launch: dict) -> list[str]:
|
|
"""The fingerprinted command line (passed in full on every call; flags are not inherited)."""
|
|
a = [launch["binary"]["path"], "-p", "--output-format", "stream-json", "--verbose",
|
|
"--model", launch["model"], "--effort", launch["effort"],
|
|
"--permission-mode", launch["permission_mode"], "--tools", ",".join(launch["tools"])]
|
|
if launch.get("allowed_tools"):
|
|
a += ["--allowedTools", ",".join(launch["allowed_tools"])]
|
|
if launch.get("disallowed_tools"):
|
|
a += ["--disallowedTools", ",".join(launch["disallowed_tools"])]
|
|
a += ["--strict-mcp-config", "--mcp-config", launch["mcp"]["config"],
|
|
"--append-system-prompt-file", launch["append_system_prompt"]["path"]]
|
|
if launch.get("agent"):
|
|
a += ["--agent", launch["agent"]["name"]]
|
|
return a
|
|
|
|
|
|
def _git_head(cwd: str) -> str | None:
|
|
try:
|
|
r = git(["rev-parse", "HEAD"], cwd=cwd)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return None
|
|
out = r.stdout.decode("utf-8", "replace").strip()
|
|
return out if r.returncode == 0 and out else None
|
|
|
|
|
|
def delta_header(cwd: str, base_commit: str | None) -> str:
|
|
"""Commits since the base's training commit plus the working tree's shortstat against it."""
|
|
if not base_commit:
|
|
return ("No base commit is recorded (the base's cwd was not a git repository, or git failed "
|
|
"at training). Treat everything on disk as possibly changed since training.")
|
|
parts = []
|
|
try:
|
|
lg = git(["log", "--oneline", "%s..HEAD" % base_commit], cwd=cwd)
|
|
st = git(["diff", "--shortstat", base_commit], cwd=cwd)
|
|
log_txt = lg.stdout.decode("utf-8", "replace").strip() if lg.returncode == 0 else \
|
|
"(git log failed: %s)" % lg.stderr.decode("utf-8", "replace").strip()
|
|
st_txt = st.stdout.decode("utf-8", "replace").strip() if st.returncode == 0 else \
|
|
"(git diff failed: %s)" % st.stderr.decode("utf-8", "replace").strip()
|
|
except (OSError, subprocess.SubprocessError) as e:
|
|
log_txt = st_txt = "(git failed: %s)" % e
|
|
parts.append("Base commit: `%s`." % base_commit)
|
|
parts.append("Commits since then (`git log --oneline %s..HEAD`):\n\n```\n%s\n```"
|
|
% (base_commit[:12], log_txt or "(none)"))
|
|
parts.append("Working tree against the base commit (`git diff --shortstat %s`):\n\n```\n%s\n```"
|
|
% (base_commit[:12], st_txt or "(no changes)"))
|
|
return "\n\n".join(parts)
|
|
|
|
|
|
def _next_job_id(reg: dict) -> str:
|
|
n = 0
|
|
for j in reg.get("jobs", {}):
|
|
m = re.fullmatch(r"j(\d+)", j)
|
|
if m:
|
|
n = max(n, int(m.group(1)))
|
|
return "j%04d" % (n + 1)
|
|
|
|
|
|
def _running_capped(reg: dict) -> list[str]:
|
|
return sorted(j for j, e in reg.get("jobs", {}).items()
|
|
if e.get("kind") in ("fork", "continue") and e.get("status") in RUNNING)
|
|
|
|
|
|
def _new_entry(job: str, kind: str, **kw) -> dict:
|
|
e = {"job": job, "kind": kind, "base": None, "block": None, "phase": None, "mode": None,
|
|
"paths": None, "status": "created", "flags": [], "reason": None, "session": None,
|
|
"parent_session": None, "model_requested": None, "created_at": _iso(), "launched_at": None,
|
|
"ended_at": None, "supervisor": None, "cwd": None, "argv": None,
|
|
"brief": "briefs/%s.md" % job, "raw": "raw/%s.jsonl" % job, "max_s": None,
|
|
"cost_before": 0.0, "model_usage_before": {}, "base_context": None,
|
|
"base_state_at_launch": None, "launch_drift": [], "binary_version": None,
|
|
"head_at_launch": None, "continues": None, "delivered_at": None, "settled_at": None}
|
|
e.update(kw)
|
|
return e
|
|
|
|
|
|
def _start_job(state: str, run: str, job: str, prompt: str, launch: dict) -> dict:
|
|
"""Write the brief and the child env, spawn the supervisor, record it. A failed spawn ends the
|
|
job as `failed`, reason `spawn_failed`."""
|
|
atomic_write_bytes(os.path.join(state, "briefs", job + ".md"), prompt.encode("utf-8"))
|
|
write_json_atomic(raw_file(state, job, ".env.json"), scrubbed_env(launch["binary"]["path"]))
|
|
sp = spawn_supervisor(state, run, job)
|
|
if not sp.get("ok"):
|
|
finalize_job(state, job, ss=None, hint="failed", reason="spawn_failed", detail=sp.get("detail"))
|
|
return sp
|
|
with RegistryLock(state):
|
|
reg = read_registry(state)
|
|
e = reg["jobs"][job]
|
|
e["supervisor"] = {"pid": sp["pid"], "ctime": sp.get("ctime")}
|
|
e["launched_at"] = _iso()
|
|
if e.get("status") == "created":
|
|
e["status"] = "running"
|
|
write_registry(state, reg)
|
|
return sp
|
|
|
|
|
|
def cmd_train(args) -> tuple[dict, int]:
|
|
state, reg = _load_run(args)
|
|
name = args.base
|
|
if not re.fullmatch(r"[A-Za-z0-9_.-]+", name or ""):
|
|
raise ForksError("bad base name", base=name)
|
|
cfg = reg["config"]
|
|
la = cfg["launch"]
|
|
brief_path, brief = _read_brief(args.brief)
|
|
binfo, drift = _check_binary(la["binary"], la)
|
|
session = str(uuid.uuid4())
|
|
argv = claude_argv(la) + ["--session-id", session]
|
|
head = _git_head(la["cwd"])
|
|
with RegistryLock(state):
|
|
reg = read_registry(state)
|
|
reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso()
|
|
old = reg["bases"].get(name)
|
|
if old and old.get("status") != "failed":
|
|
raise ForksError("base exists", base=name, status=old.get("status"))
|
|
job = _next_job_id(reg)
|
|
reg["bases"][name] = {"name": name, "status": "training", "session": session, "cwd": la["cwd"],
|
|
"fingerprint": cfg["fingerprint"], "binary": binfo, "created_at": _iso(),
|
|
"train_job": job, "phase": _phase(args.phase), "last_hit": None,
|
|
"base_commit": None, "context_tokens": None, "forks": []}
|
|
reg["jobs"][job] = _new_entry(
|
|
job, "train", base=name, block=args.block, phase=_phase(args.phase), session=session,
|
|
model_requested=la["model"], cwd=la["cwd"], argv=argv, max_s=float(cfg["max_minutes"]) * 60,
|
|
launch_drift=drift, binary_version=binfo["version"], head_at_launch=head,
|
|
brief_source=brief_path)
|
|
write_registry(state, reg)
|
|
prompt = render_template("train.md", {"base": name, "run": norm_path(args.run), "state_dir": state,
|
|
"notes_dir": os.path.join(state, "notes"),
|
|
"logs_dir": os.path.join(state, "logs"), "brief": brief})
|
|
_start_job(state, norm_path(args.run), job, prompt, la)
|
|
return {"job": job, "base": name, "session_id": session}, EXIT_OK
|
|
|
|
|
|
def cmd_fork(args) -> tuple[dict, int]:
|
|
_refuse_overrides(args)
|
|
if args.mode == "parallel" and not args.paths:
|
|
raise ForksError("--mode parallel needs --paths (the files this fork may touch)", mode="parallel")
|
|
state, reg = _load_run(args)
|
|
cfg = reg["config"]
|
|
la = cfg["launch"]
|
|
base = reg["bases"].get(args.base)
|
|
if base is None:
|
|
raise ForksError("unknown base", base=args.base, bases=sorted(reg["bases"]))
|
|
if base.get("status") != "active" and not args.force:
|
|
raise ForksError("base is not active (use --force to fork it anyway)", base=args.base,
|
|
status=base.get("status"))
|
|
if not base.get("session"):
|
|
raise ForksError("base has no session", base=args.base, status=base.get("status"))
|
|
brief_path, brief = _read_brief(args.brief)
|
|
binfo, drift = _check_binary(base.get("binary") or la["binary"], la)
|
|
delta = delta_header(la["cwd"], base.get("base_commit"))
|
|
head = _git_head(la["cwd"])
|
|
session = str(uuid.uuid4())
|
|
argv = claude_argv(la) + ["--resume", base["session"], "--fork-session", "--session-id", session]
|
|
if args.max_budget_usd:
|
|
argv += ["--max-budget-usd", "%g" % args.max_budget_usd]
|
|
paths = list(args.paths) if args.paths else None
|
|
reap_lost(state)
|
|
with RegistryLock(state):
|
|
reg = read_registry(state)
|
|
reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso()
|
|
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,
|
|
status=base.get("status"))
|
|
running = _running_capped(reg)
|
|
if len(running) >= int(cfg["parallel_cap"]):
|
|
raise ForksError("parallel cap reached", parallel_cap=cfg["parallel_cap"], running=running)
|
|
job = _next_job_id(reg)
|
|
warmth = base_warmth(base, cfg)
|
|
reg["jobs"][job] = _new_entry(
|
|
job, "fork", base=args.base, block=args.block, phase=_phase(args.phase), mode=args.mode,
|
|
paths=paths, session=session, parent_session=base["session"], model_requested=la["model"],
|
|
cwd=la["cwd"], argv=argv,
|
|
max_s=float(args.max_minutes if args.max_minutes else cfg["max_minutes"]) * 60,
|
|
cost_before=float(base.get("cost_cum_usd") or 0.0),
|
|
model_usage_before=base.get("model_usage_cum") or {},
|
|
base_context=base.get("context_tokens"), base_state_at_launch=warmth["base_state"],
|
|
warm_until_at_launch=warmth["warm_until"], launch_drift=drift,
|
|
binary_version=binfo["version"], head_at_launch=head, brief_source=brief_path,
|
|
forced=bool(args.force and base.get("status") != "active"))
|
|
base.setdefault("forks", []).append(job)
|
|
write_registry(state, reg)
|
|
if args.mode == "parallel":
|
|
commit_rule = ("This fork runs in parallel with other forks: do NOT commit and do not stage. "
|
|
"Touch only these paths: %s. The orchestrator commits after its gate."
|
|
% ", ".join("`%s`" % p for p in paths))
|
|
else:
|
|
commit_rule = ("This fork commits: after your gate passes, stage and commit your block by "
|
|
"explicit paths only (never `git add -A` or `.`).")
|
|
if paths:
|
|
commit_rule += " Expected paths: %s." % ", ".join("`%s`" % p for p in paths)
|
|
prompt = render_template("fork.md", {
|
|
"base": args.base, "run": norm_path(args.run), "state_dir": state, "block": args.block,
|
|
"phase": "n/a" if args.phase is None else str(args.phase), "mode": args.mode, "delta": delta,
|
|
"commit_rule": commit_rule, "note_path": os.path.join(state, *note_rel(args.block).split("/")),
|
|
"logs_dir": os.path.join(state, "logs"), "brief": brief})
|
|
_start_job(state, norm_path(args.run), job, prompt, la)
|
|
return {"job": job, "base": args.base, "base_state": warmth["base_state"],
|
|
"warm_until": warmth["warm_until"], "rebase_due": False, "retrain_due": False,
|
|
"reasons": []}, EXIT_OK
|
|
|
|
|
|
def cmd_continue(args) -> tuple[dict, int]:
|
|
_refuse_overrides(args)
|
|
state, reg = _load_run(args)
|
|
cfg = reg["config"]
|
|
la = cfg["launch"]
|
|
fj = reg["jobs"].get(args.job)
|
|
if fj is None:
|
|
raise ForksError("unknown job", job=args.job)
|
|
if fj.get("kind") != "fork":
|
|
raise ForksError("continue takes a fork job", job=args.job, kind=fj.get("kind"))
|
|
|
|
def busy(r):
|
|
return [j for j, e in r["jobs"].items() if e.get("status") in RUNNING
|
|
and (j == args.job or (e.get("kind") == "continue" and e.get("continues") == args.job))]
|
|
if busy(reg):
|
|
raise ForksError("the fork is still running", job=args.job, running=busy(reg))
|
|
if not fj.get("session"):
|
|
raise ForksError("the fork has no session", job=args.job)
|
|
base = reg["bases"].get(fj.get("base")) or {}
|
|
brief_path, brief = _read_brief(args.brief)
|
|
binfo, drift = _check_binary(base.get("binary") or la["binary"], la)
|
|
head = _git_head(la["cwd"])
|
|
argv = claude_argv(la) + ["--resume", fj["session"]]
|
|
reap_lost(state)
|
|
with RegistryLock(state):
|
|
reg = read_registry(state)
|
|
reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso()
|
|
fj = reg["jobs"][args.job]
|
|
if busy(reg):
|
|
raise ForksError("the fork is still running", job=args.job, running=busy(reg))
|
|
running = _running_capped(reg)
|
|
if len(running) >= int(cfg["parallel_cap"]):
|
|
raise ForksError("parallel cap reached", parallel_cap=cfg["parallel_cap"], running=running)
|
|
base = reg["bases"].get(fj.get("base")) or {}
|
|
cum = fj.get("session_cost_cum_usd")
|
|
job = _next_job_id(reg)
|
|
warmth = base_warmth(base, cfg)
|
|
reg["jobs"][job] = _new_entry(
|
|
job, "continue", base=fj.get("base"), block=fj.get("block"), phase=fj.get("phase"),
|
|
mode=fj.get("mode"), paths=fj.get("paths"), session=fj["session"],
|
|
parent_session=fj.get("parent_session"), model_requested=la["model"], cwd=la["cwd"],
|
|
argv=argv, max_s=float(args.max_minutes if args.max_minutes else cfg["max_minutes"]) * 60,
|
|
cost_before=float(cum if cum is not None else (base.get("cost_cum_usd") or 0.0)),
|
|
model_usage_before=(fj.get("session_model_usage_cum") if cum is not None
|
|
else base.get("model_usage_cum")) or {},
|
|
base_context=base.get("context_tokens"), base_state_at_launch=warmth["base_state"],
|
|
launch_drift=drift, binary_version=binfo["version"], head_at_launch=head,
|
|
continues=args.job, brief_source=brief_path)
|
|
write_registry(state, reg)
|
|
_start_job(state, norm_path(args.run), job, brief, la)
|
|
return {"job": job}, EXIT_OK
|
|
|
|
|
|
# --- the supervisor -------------------------------------------------------------------
|
|
|
|
|
|
class Supervisor:
|
|
"""`_supervise`: runs one `claude` call, parses its stream, watches it, writes the result.
|
|
The only writer of `jobs/<job>.json` (C3b)."""
|
|
|
|
def __init__(self, state: str, job: str):
|
|
self.state, self.job = state, job
|
|
reg = read_registry(state)
|
|
if not reg or job not in reg.get("jobs", {}):
|
|
raise ForksError("unknown job", job=job, state=state)
|
|
self.cfg = reg["config"]
|
|
self.entry = reg["jobs"][job]
|
|
self.env = read_json(raw_file(state, job, ".env.json"))
|
|
self.ss = StreamState()
|
|
self.mu = threading.Lock()
|
|
self.flags: list[str] = []
|
|
self.last_output = _now()
|
|
self.rt = {"job": job, "status": "running", "supervisor": pid_record(os.getpid()), "claude": None,
|
|
"started_at": _iso(), "heartbeat_at": _iso(), "turns": 0, "last_event_at": None,
|
|
"flags": []}
|
|
self._last_write = 0.0
|
|
|
|
def write_rt(self, **kw) -> None:
|
|
with self.mu:
|
|
self.rt.update(kw)
|
|
self.rt["turns"] = self.ss.turns
|
|
self.rt["last_event_at"] = _iso(self.ss.last_event_at) if self.ss.last_event_at else None
|
|
self.rt["flags"] = list(self.flags)
|
|
self.rt["heartbeat_at"] = _iso()
|
|
snap = dict(self.rt)
|
|
write_json_atomic(job_file(self.state, self.job), snap)
|
|
self._last_write = _now()
|
|
|
|
def _reader(self, stream, raw) -> None:
|
|
for line in iter(stream.readline, b""):
|
|
t = _now()
|
|
raw.write(line)
|
|
raw.flush()
|
|
try:
|
|
ev = json.loads(line.decode("utf-8", "replace"))
|
|
except ValueError:
|
|
ev = None
|
|
with self.mu:
|
|
self.last_output = t
|
|
if ev is not None:
|
|
self.ss.feed(ev, t)
|
|
|
|
def _feeder(self, stdin, data: bytes) -> None:
|
|
with contextlib.suppress(OSError, ValueError):
|
|
stdin.write(data)
|
|
with contextlib.suppress(OSError, ValueError):
|
|
stdin.close()
|
|
|
|
def run(self) -> tuple[dict, int]:
|
|
e = self.entry
|
|
if e.get("status") not in RUNNING or e.get("kill_requested"):
|
|
return {"job": self.job, "skipped": "job is %s" % e.get("status")}, EXIT_OK
|
|
if self.env is None:
|
|
finalize_job(self.state, self.job, ss=None, hint="failed", reason="supervisor_error",
|
|
detail="missing env file")
|
|
return {"error": "missing env file"}, EXIT_USAGE
|
|
os.environ.clear() # git and taskkill lookups use the child's PATH
|
|
os.environ.update(self.env)
|
|
self.write_rt()
|
|
with open(os.path.join(self.state, "briefs", self.job + ".md"), "rb") as f:
|
|
prompt = f.read()
|
|
t0 = _now()
|
|
raw = open(raw_file(self.state, self.job), "wb")
|
|
err = open(raw_file(self.state, self.job, ".err"), "wb")
|
|
try:
|
|
p = subprocess.Popen(e["argv"], cwd=e["cwd"], env=self.env, stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE, stderr=err, creationflags=_NOWIN,
|
|
start_new_session=not IS_WINDOWS)
|
|
except OSError as ex:
|
|
raw.close()
|
|
err.close()
|
|
finalize_job(self.state, self.job, ss=self.ss, hint="failed", reason="launch_error",
|
|
detail=str(ex), wall_s=0)
|
|
self.write_rt(status="failed")
|
|
return {"error": "launch_error", "detail": str(ex)}, EXIT_USAGE
|
|
crec = pid_record(p.pid)
|
|
self.write_rt(claude=crec)
|
|
reader = threading.Thread(target=self._reader, args=(p.stdout, raw), daemon=True)
|
|
reader.start()
|
|
threading.Thread(target=self._feeder, args=(p.stdin, prompt), daemon=True).start()
|
|
max_s = float(e.get("max_s") or float(self.cfg["max_minutes"]) * 60)
|
|
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"
|
|
last_orphan_check = 0.0
|
|
last_turns = 0
|
|
while p.poll() is None:
|
|
now = _now()
|
|
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)
|
|
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
|
|
break
|
|
if quiet > stall_s and "stalled" not in self.flags:
|
|
self.flags.append("stalled")
|
|
self.write_rt()
|
|
if not hit_written and first_at:
|
|
hit_written = True
|
|
with contextlib.suppress(ForksError, OSError):
|
|
bump_last_hit(self.state, e["base"], first_at)
|
|
if now - last_orphan_check >= ORPHAN_CHECK_S:
|
|
last_orphan_check = now
|
|
try:
|
|
reg = read_registry(self.state)
|
|
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)
|
|
hint, reason, flags_extra = "killed", "orphaned", ["orphaned"]
|
|
detail = "no orchestrator call for %.0f s (orphan_after %.0f s)" % (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)
|
|
if sweep["snapshot"]:
|
|
_diag("swept %d leftover process(es) of claude: %s" % (
|
|
len(sweep["snapshot"]), [(x["pid"], x["name"]) for x in sweep["snapshot"]]))
|
|
reader.join(READER_JOIN_S)
|
|
with contextlib.suppress(OSError):
|
|
raw.close()
|
|
with contextlib.suppress(OSError):
|
|
err.close()
|
|
wall = _now() - t0
|
|
base_commit = None
|
|
if e["kind"] == "train":
|
|
base_commit = _git_head(e["cwd"]) or e.get("head_at_launch")
|
|
res = finalize_job(self.state, self.job, ss=self.ss, hint=hint, flags=self.flags + flags_extra,
|
|
reason=reason, detail=detail, exit_code=exit_code, wall_s=wall,
|
|
base_commit=base_commit)
|
|
final = res["status"] if res else (read_registry(self.state)["jobs"][self.job].get("status"))
|
|
self.write_rt(status=final, ended_at=_iso(), exit_code=exit_code)
|
|
return {"job": self.job, "status": final}, EXIT_OK
|
|
|
|
|
|
def cmd_supervise(args) -> tuple[dict, int]:
|
|
state = state_dir_for(args.run)
|
|
if not os.path.isdir(os.path.join(state, "raw")) or not re.fullmatch(r"j\d+", args.job):
|
|
raise ForksError("no such job state", state=state, job=args.job)
|
|
log = open(raw_file(state, args.job, ".sup.log"), "a", encoding="utf-8", buffering=1)
|
|
sys.stdout = sys.stderr = log
|
|
_diag("supervisor %d for %s starting (%s)" % (os.getpid(), args.job, _iso()))
|
|
try:
|
|
return Supervisor(state, args.job).run()
|
|
except Exception as ex:
|
|
traceback.print_exc(file=log)
|
|
with contextlib.suppress(Exception):
|
|
rt = _job_runtime(state, args.job)
|
|
if rt.get("claude"):
|
|
kill_tree(rt["claude"])
|
|
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
|
|
|
|
|
|
# --- wait, list, kill, expire -----------------------------------------------------------
|
|
|
|
|
|
def _running_info(state: str, job: str, e: dict, now: float) -> dict:
|
|
rt = _job_runtime(state, job)
|
|
start = _ts(rt.get("started_at")) or _ts(e.get("launched_at")) or _ts(e.get("created_at"))
|
|
last = _ts(rt.get("last_event_at"))
|
|
return {"job": job, "block": e.get("block"),
|
|
"elapsed_s": None if start is None else round(now - start, 1),
|
|
"last_event_s_ago": None if last is None else round(now - last, 1),
|
|
"turns": int(rt.get("turns") or 0)}
|
|
|
|
|
|
def cmd_wait(args) -> tuple[dict, int]:
|
|
state, reg = _load_run(args)
|
|
if args.timeout > WAIT_MAX_S:
|
|
raise ForksError("--timeout over the maximum", timeout=args.timeout, max=WAIT_MAX_S)
|
|
if args.jobs and args.all:
|
|
raise ForksError("name jobs or --all, not both")
|
|
if not args.jobs and not args.all:
|
|
raise ForksError("name jobs or --all")
|
|
unknown = [j for j in args.jobs if j not in reg["jobs"]]
|
|
if unknown:
|
|
raise ForksError("unknown job", jobs=unknown)
|
|
touch_orchestrator(state)
|
|
deadline = time.monotonic() + args.timeout
|
|
|
|
def evaluate(r):
|
|
members = list(args.jobs) if args.jobs else sorted(
|
|
j for j, x in r["jobs"].items() if not x.get("settled_at"))
|
|
finished = [j for j in members if r["jobs"][j].get("status") in FINAL]
|
|
running = [j for j in members if j not in finished]
|
|
met = bool(finished) or not members if args.any else not running
|
|
return members, finished, running, met
|
|
|
|
while True:
|
|
reap_lost(state)
|
|
members, finished, running, met = evaluate(read_registry(state))
|
|
left = deadline - time.monotonic()
|
|
if met or left <= 0:
|
|
break
|
|
time.sleep(min(WAIT_POLL_S, left))
|
|
with RegistryLock(state):
|
|
reg = read_registry(state)
|
|
members, finished, running, met = evaluate(reg)
|
|
done, now = [], _now()
|
|
for j in finished:
|
|
e = reg["jobs"][j]
|
|
if e.get("delivered_at"):
|
|
continue
|
|
res = read_json(result_file(state, j))
|
|
if res is not None:
|
|
done.append(res)
|
|
e["delivered_at"] = _iso(now)
|
|
if met:
|
|
for j in finished:
|
|
reg["jobs"][j]["settled_at"] = _iso(now)
|
|
reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso(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}
|
|
out = {"done": done, "running": [_running_info(state, j, entries[j], now) for j in running],
|
|
"pings": []}
|
|
if not met:
|
|
return out, EXIT_WAIT_TIMEOUT
|
|
return out, (EXIT_OK if all(s == "ok" for s in statuses.values()) else EXIT_JOB_FAILED)
|
|
|
|
|
|
LIST_JOB_KEYS = ("job", "kind", "base", "block", "phase", "mode", "paths", "status", "flags", "reason",
|
|
"reason_detail", "session", "parent_session", "continues", "created_at", "launched_at",
|
|
"ended_at", "fingerprint", "cache", "delivered_at", "settled_at")
|
|
LIST_BASE_KEYS = ("name", "status", "session", "cwd", "base_commit", "context_tokens",
|
|
"training_cost_usd", "cost_cum_usd", "last_hit", "train_job", "forks", "phase",
|
|
"created_at", "trained_at")
|
|
|
|
|
|
def cmd_list(args) -> tuple[dict, int]:
|
|
state, _ = _load_run(args)
|
|
touch_orchestrator(state)
|
|
lost = reap_lost(state)
|
|
reg = read_registry(state)
|
|
now = _now()
|
|
cfg = reg["config"]
|
|
bases = []
|
|
for name, b in sorted(reg["bases"].items()):
|
|
row = {k: b.get(k) for k in LIST_BASE_KEYS}
|
|
row["name"] = name
|
|
row.update(base_warmth(b, cfg, now))
|
|
row.update({"rebase_due": False, "retrain_due": False, "reasons": []})
|
|
bases.append(row)
|
|
jobs, running = [], []
|
|
for j, e in sorted(reg["jobs"].items()):
|
|
row = {k: e.get(k) for k in LIST_JOB_KEYS}
|
|
if e.get("status") in RUNNING:
|
|
info = _running_info(state, j, e, now)
|
|
row.update({k: info[k] for k in ("elapsed_s", "last_event_s_ago", "turns")})
|
|
running.append(j)
|
|
else:
|
|
row["turns"] = e.get("turns")
|
|
jobs.append(row)
|
|
return {"state": state, "bases": bases, "jobs": jobs, "running": running, "lost_now": lost}, EXIT_OK
|
|
|
|
|
|
def cmd_kill(args) -> tuple[dict, int]:
|
|
state, reg = _load_run(args)
|
|
if args.job not in reg["jobs"]:
|
|
raise ForksError("unknown job", job=args.job)
|
|
return kill_job(state, args.job, "kill"), EXIT_OK
|
|
|
|
|
|
def cmd_expire(args) -> tuple[dict, int]:
|
|
state, reg = _load_run(args)
|
|
names = sorted(reg["bases"]) if args.all else [args.base]
|
|
unknown = [n for n in names if n not in reg["bases"]]
|
|
if unknown:
|
|
raise ForksError("unknown base", bases=unknown)
|
|
retired = []
|
|
with RegistryLock(state):
|
|
reg = read_registry(state)
|
|
for n in names:
|
|
b = reg["bases"][n]
|
|
if b.get("status") not in ("retired", "superseded"):
|
|
b["status"] = "retired"
|
|
b["retired_at"] = _iso()
|
|
retired.append(n)
|
|
write_registry(state, reg)
|
|
targets = sorted(j for j, e in reg["jobs"].items()
|
|
if e.get("base") in names and e.get("status") in RUNNING)
|
|
killed = [kill_job(state, j, "expire") for j in targets]
|
|
return {"retired": retired, "bases": names, "killed": killed,
|
|
"ok": all(k.get("ok", True) for k in killed)}, EXIT_OK
|
|
|
|
|
|
HANDLERS = {
|
|
"where": cmd_where,
|
|
"init": cmd_init,
|
|
"train": cmd_train,
|
|
"fork": cmd_fork,
|
|
"continue": cmd_continue,
|
|
"wait": cmd_wait,
|
|
"list": cmd_list,
|
|
"expire": cmd_expire,
|
|
"kill": cmd_kill,
|
|
"_supervise": cmd_supervise,
|
|
"rebase": _not_implemented(phase=2),
|
|
"usage": _not_implemented(phase=2),
|
|
"tick": _not_implemented(phase=2),
|
|
}
|
|
|
|
|
|
def run(argv) -> tuple[dict, int]:
|
|
parser = build_parser()
|
|
try:
|
|
args, unknown = parser.parse_known_args(argv)
|
|
except _HelpRequested as h:
|
|
return {"help": h.text, "commands": [c for c in COMMANDS if not c.startswith("_")]}, EXIT_OK
|
|
if unknown and args.command not in PHASE2_COMMANDS:
|
|
parser.error("unrecognized arguments: %s" % " ".join(unknown))
|
|
if args.command in CHILD_REFUSED and os.environ.get(CHILD_MARKER):
|
|
raise ForksError("launching commands are refused inside a fork (%s is set)" % CHILD_MARKER,
|
|
command=args.command)
|
|
return HANDLERS[args.command](args)
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
if argv is None:
|
|
argv = sys.argv[1:]
|
|
try:
|
|
out, code = run(argv)
|
|
except ForksError as e:
|
|
out, code = e.payload, e.code
|
|
except Exception as e: # never break the stdout invariant
|
|
traceback.print_exc(file=sys.stderr)
|
|
out, code = {"error": "internal error", "type": type(e).__name__, "detail": str(e)}, EXIT_USAGE
|
|
_emit(out)
|
|
return code
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|