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

🤖 forks: warmth, pings and tick (block 2.1)

Browse Source 
TTL inference per base and ttl_downgrade; warmth with init --warm-margin-s; ping rule (activity window, break-even caps 19/11, pings on/off) with C3(d) in-flight dedupe under the lock; pings returned once through wait's pings list, partial refreshes, miss/failed flag ping_missed; tick in every supervisor (60 s), in wait (start + 30 s) and as tick --run (N3-refused). M16: job-file and running-info progress counter renamed to messages. M8: visible-window check widened (console/Terminal diff, positive control, skip without desktop). Billed-error test (stub directive billed).
This commit is contained in:
g4borg
2026-10-16 16:57:23 +02:00
parent 571a3d6006
commit fff43a059a
2 changed files with 847 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/spire/skills/forks/forks.py
+289 -33
View File
@@ -122,9 +122,20 @@ SCRUB_WATCH_PREFIXES = ("CLAUDE_", "ANTHROPIC_") # `where` lists unknown vars
# 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")
# N3 (tick pings: Phase 2). A fork may only run where, list, usage and wait.
CHILD_REFUSED = ("train", "fork", "continue", "rebase", "kill", "expire", "init")
PHASE2_COMMANDS = ("rebase", "usage")
# N3: a fork may only run where, list, usage and wait (`tick` launches pings).
CHILD_REFUSED = ("train", "fork", "continue", "rebase", "kill", "expire", "init", "tick")
# Warmth and keep-warm (design 5, block 2.1). Module-level so tests can shorten them.
TICK_SUP_S = 60.0 # a supervisor runs one tick pass this often
TICK_WAIT_S = 30.0 # `wait` runs one tick pass at its start and then this often
ACTIVITY_WINDOW_S = 600.0 # ping only while a fork/continue ran or `wait` was called this recently
PING_PROMPT = "Reply OK"
PING_MAX_S = 300.0 # wall-time limit of one ping (a single turn)
TTL_1H, TTL_5M = 3600, 300
# Price ratios against the base input price, in hundredths (integers: floor(1.9 / 0.1) is 18 in floats).
PRICE_RATIOS = {"read": 10, "write_1h": 200, "write_5m": 125}
REAL_KINDS = ("train", "fork", "continue", "rebase") # every kind but `ping`
# 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)
@@ -1243,17 +1254,46 @@ def cache_verdict(first_usage: dict | None, base_context) -> str | None:
return "partial" if cr > 0 else "miss"
def infer_ttl(tokens: dict | None, prev, default=None):
"""Design 5: the TTL a base's cache entries live for, from its most recent call's tokens.
1h written -> 3600; only 5m written -> 300; otherwise (read only, or nothing) the previous
known TTL, and with none known `default` (the configured ttl_s)."""
if not tokens:
return prev if prev is not None else default
if int(tokens.get("cache_creation_1h") or 0) > 0:
return TTL_1H
if int(tokens.get("cache_creation_5m") or 0) > 0:
return TTL_5M
return prev if prev is not None else default
def wrote_only_5m(tokens: dict | None) -> bool:
"""`ttl_downgrade`: the call wrote 5-minute cache entries and no 1-hour ones."""
return bool(tokens) and int(tokens.get("cache_creation_5m") or 0) > 0 \
and int(tokens.get("cache_creation_1h") or 0) == 0
def break_even(ttl) -> int:
"""floor((write - read) / read): pings pay while fewer are needed (19 at 1 h, 11 at 5 min)."""
write = PRICE_RATIOS["write_1h"] if float(ttl) > TTL_5M else PRICE_RATIOS["write_5m"]
return (write - PRICE_RATIOS["read"]) // PRICE_RATIOS["read"]
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"])
"""Design 5: ttl = the base's inferred TTL, else the configured ttl_s; warm_until = last_hit +
ttl; margin = init's warm_margin_s if set, else max(120, ttl/10)."""
ttl = float((base or {}).get("ttl_s") or cfg.get("ttl_s") or CONFIG_DEFAULTS["ttl_s"])
wm = cfg.get("warm_margin_s")
margin = float(wm) if wm is not None else max(120.0, ttl / 10.0)
out = {"ttl_s": int(ttl) if ttl.is_integer() else ttl,
"margin_s": int(margin) if margin.is_integer() else margin}
last = _ts((base or {}).get("last_hit"))
if last is None:
return {"base_state": "cold", "warm_until": None}
return {"base_state": "cold", "warm_until": None, **out}
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)}
return {"base_state": state, "warm_until": _iso(wu), **out}
# --------------------------------------------------------------------------------------
@@ -1383,7 +1423,7 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint:
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
if kind in ("fork", "continue", "ping") 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
@@ -1394,6 +1434,14 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint:
if kind in ("fork", "continue") and (base_state == "cold" or (cache not in (None, "hit")
and fingerprint == "ok")):
flags.append("cold_start")
if u and wrote_only_5m(u["tokens"]):
flags.append("ttl_downgrade")
# A ping refreshes the base unless it missed (P0.6: a `partial` ping still kept the base warm;
# it re-created about the shared system prefix). A miss or a failed ping leaves last_hit alone.
ping_refreshed = (kind == "ping" and status == "ok" and cache in ("hit", "partial")
and bool(ss and ss.first_at))
if kind == "ping" and not ping_refreshed:
flags.append("ping_missed")
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
@@ -1435,15 +1483,22 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint:
holder["session_model_usage_cum"] = u["model_usage"]
if base is not None:
# Design 5: only a request that reads exactly the base prefix refreshes it: a fork's first
# request (and the training's end, which writes it). A continue reads the fork's own
# longer prefix, so it leaves the base's last_hit alone (I3).
# request, a ping that did not miss (and the training's end, which writes it). A continue
# reads the fork's own longer prefix, so it leaves the base's last_hit alone (I3).
hit_t = None
if ss and kind == "train":
hit_t = ss.last_assistant_at
elif ss and kind == "fork":
hit_t = ss.first_at
elif ping_refreshed:
hit_t = ss.first_at
if hit_t and (_ts(base.get("last_hit")) or 0) < hit_t:
base["last_hit"] = _iso(hit_t)
if u: # design 5: TTL from the most recent call on this base (any kind)
base["ttl_s"] = infer_ttl(u["tokens"], base.get("ttl_s"))
base["ttl_from"] = entry["job"]
if kind == "ping" and base.get("ping_in_flight") == entry["job"]:
base["ping_in_flight"] = None # C3(d): cleared when the ping finishes, any status
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,
@@ -1492,15 +1547,19 @@ def in_child() -> bool:
return bool(os.environ.get(CHILD_MARKER))
def stamp_orchestrator(reg: dict, t: float | None = None) -> None:
"""N1: set `run.last_orchestrator_call_at` in `reg` (caller holds the lock and writes it).
def stamp_orchestrator(reg: dict, t: float | None = None, wait: bool = False) -> None:
"""N1: set `run.last_orchestrator_call_at` in `reg` (caller holds the lock and writes it);
`wait=True` also sets `run.last_wait_at` (the ping rule's activity condition, design 5).
A fork's own calls never stamp, so a fork cannot keep its own watchdog alive (M4)."""
if in_child():
return
reg.setdefault("run", {})["last_orchestrator_call_at"] = _iso(t)
run = reg.setdefault("run", {})
run["last_orchestrator_call_at"] = _iso(t)
if wait:
run["last_wait_at"] = run["last_orchestrator_call_at"]
def touch_orchestrator(state: str) -> None:
def touch_orchestrator(state: str, wait: bool = False) -> None:
"""N1: record that the orchestrator called us (wait, fork, continue, train, list)."""
if in_child():
return
@@ -1508,7 +1567,7 @@ def touch_orchestrator(state: str) -> None:
reg = read_registry(state)
if reg is None:
return
stamp_orchestrator(reg)
stamp_orchestrator(reg, wait=wait)
write_registry(state, reg)
@@ -1686,7 +1745,10 @@ def build_parser() -> _Parser:
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("--ttl-s", type=_pos_int, default=CONFIG_DEFAULTS["ttl_s"],
help="TTL assumed until a base's calls show one (default 3600)")
i.add_argument("--warm-margin-s", type=_pos_int, default=None,
help="seconds before warm_until a base turns `expiring` (default max(120, ttl/10))")
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():
@@ -1745,9 +1807,10 @@ def build_parser() -> _Parser:
s = cmd("_supervise", None) # internal
s.add_argument("--job", required=True)
cmd("tick", "One keep-warm pass: ping every base the rule picks (design 5).")
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).")):
("usage", "Aggregate the ledger (Phase 2).")):
cmd(name, text) # arguments arrive with Phase 2; until then extra args are tolerated
return p
@@ -1887,6 +1950,7 @@ def build_init_config(args, state: str) -> dict:
"orphan_after": args.orphan_after,
"pings": args.pings,
"ttl_s": args.ttl_s,
"warm_margin_s": args.warm_margin_s,
"max_minutes": args.max_minutes,
"stall_after_s": args.stall_after_s,
"thresholds": {k: getattr(args, k) for k in THRESHOLD_DEFAULTS},
@@ -2249,6 +2313,173 @@ def cmd_continue(args) -> tuple[dict, int]:
return {"job": job}, EXIT_OK
# --- keep warm: the ping rule and the tick (design 5, C3d) ----------------------------
def run_for_state(state: str) -> str:
"""The run log path of a state dir (inverse of `state_dir_for`)."""
return state[: -len("-forks")] + ".md"
def launch_record(base: dict | None, cfg: dict) -> dict:
"""The launch fields a launch on `base` uses: the base's own record when it has one, else the
run config (every base has the init config's fields until bases carry their own record)."""
return (base or {}).get("launch") or cfg["launch"]
def _job_num(job) -> int:
m = re.fullmatch(r"j(\d+)", str(job or ""))
return int(m.group(1)) if m else -1
def pings_since_real_call(reg: dict, name: str) -> int:
"""Pings of base `name` launched after its last real (non-ping) call. Job ids are monotonic
per run, so "after" is by job number (launch times have whole seconds only)."""
jobs = reg.get("jobs", {})
last_real = max((_job_num(j) for j, e in jobs.items()
if e.get("base") == name and e.get("kind") in REAL_KINDS), default=-1)
return sum(1 for j, e in jobs.items()
if e.get("base") == name and e.get("kind") == "ping" and _job_num(j) > last_real)
def ping_in_flight(reg: dict, base: dict | None) -> str | None:
"""The base's in-flight ping job (C3d), or None. A mark whose job is no longer running (or
gone) does not count: it clears itself on the next marking."""
j = (base or {}).get("ping_in_flight")
if j and (reg.get("jobs", {}).get(j) or {}).get("status") in RUNNING:
return j
return None
def run_active(reg: dict, now: float) -> bool:
"""Design 5's activity condition: a fork/continue job ran, or `wait` was called, in the last
ACTIVITY_WINDOW_S (the Athanor is in its turn)."""
lw = _ts((reg.get("run") or {}).get("last_wait_at"))
if lw is not None and now - lw <= ACTIVITY_WINDOW_S:
return True
for e in reg.get("jobs", {}).values():
if e.get("kind") not in ("fork", "continue"):
continue
if e.get("status") in RUNNING:
return True
end = _ts(e.get("ended_at"))
if end is not None and now - end <= ACTIVITY_WINDOW_S:
return True
return False
def ping_decision(reg: dict, name: str, now: float | None = None) -> dict:
"""The ping rule for one base: pings on, base active, state expiring, no ping in flight, the
run active in the last 10 min, and fewer pings since its last real call than break_even."""
now = _now() if now is None else now
cfg = reg.get("config") or {}
base = reg["bases"][name]
w = base_warmth(base, cfg, now)
n = pings_since_real_call(reg, name)
be = break_even(w["ttl_s"])
inflight = ping_in_flight(reg, base)
if cfg.get("pings", CONFIG_DEFAULTS["pings"]) != "on":
why = "pings_off"
elif base.get("status") != "active" or not base.get("session"):
why = "status:%s" % base.get("status")
elif w["base_state"] != "expiring":
why = "state:%s" % w["base_state"]
elif inflight:
why = "in_flight:%s" % inflight
elif not run_active(reg, now):
why = "no_activity"
elif n >= be:
why = "cap:%d/%d" % (n, be)
else:
why = None
return {"base": name, "ping": why is None, "why": why, "base_state": w["base_state"],
"warm_until": w["warm_until"], "ttl_s": w["ttl_s"], "pings_since_real_call": n,
"break_even": be, "ping_in_flight": inflight}
def tick_pass(state: str) -> dict:
"""One pass of the ping rule over every base of the run. The decision and the `ping_in_flight`
mark are made under the registry lock (C3d), so concurrent tickers launch one ping per base.
The binary check runs before the lock for the bases that look eligible. Pings go through the
same hidden spawn as any job. Returns {"pinged": [job...], "bases": [decision...]}."""
reg = read_registry(state)
if not reg or not reg.get("config"):
return {"pinged": [], "bases": []}
now = _now()
decisions = [ping_decision(reg, n, now) for n in sorted(reg.get("bases", {}))]
pre = [d["base"] for d in decisions if d["ping"]]
if not pre:
return {"pinged": [], "bases": decisions}
checks = {}
for n in pre:
base = reg["bases"][n]
la = launch_record(base, reg["config"])
try:
checks[n] = _check_binary(base.get("binary") or la["binary"], la)
except ForksError as e:
checks[n] = e
launches = []
with RegistryLock(state):
reg = read_registry(state)
cfg = reg["config"]
now = _now()
decisions = []
for n in sorted(reg["bases"]):
d = ping_decision(reg, n, now)
if d["ping"] and n not in checks:
d.update(ping=False, why="eligible_since_this_tick_began") # the next tick takes it
elif d["ping"] and isinstance(checks[n], ForksError):
d.update(ping=False, why=checks[n].payload.get("reason") or checks[n].payload["error"])
if d["ping"]:
base = reg["bases"][n]
la = launch_record(base, cfg)
binfo, drift = checks[n]
job = _next_job_id(reg)
session = str(uuid.uuid4())
argv = claude_argv(la) + ["--resume", base["session"], "--fork-session",
"--session-id", session, "--max-turns", "1"]
reg["jobs"][job] = _new_entry(
job, "ping", base=n, session=session, parent_session=base["session"],
model_requested=la["model"], cwd=la["cwd"], argv=argv, max_s=PING_MAX_S,
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=d["base_state"],
warm_until_at_launch=d["warm_until"], launch_drift=drift,
binary_version=binfo["version"])
base["ping_in_flight"] = job
base["ping_in_flight_at"] = _iso(now)
d["job"] = job
launches.append((job, la))
decisions.append(d)
if launches:
write_registry(state, reg)
run = run_for_state(state)
for job, la in launches:
_start_job(state, run, job, PING_PROMPT, la)
return {"pinged": [j for j, _ in launches], "bases": decisions}
def safe_tick(state: str, where: str) -> dict | None:
"""A tick pass for `wait` and the supervisor: never raises (diagnostics on stderr)."""
try:
out = tick_pass(state)
except Exception as ex: # a keep-warm failure must not break `wait` or a supervisor
_diag("%s tick failed: %s: %s" % (where, type(ex).__name__, ex))
return None
if out["pinged"]:
_diag("%s tick: pinged %s" % (where, ", ".join("%s (%s)" % (d["job"], d["base"])
for d in out["bases"] if d.get("job"))))
return out
def cmd_tick(args) -> tuple[dict, int]:
"""`tick --run R`: one pass by hand (N3-refused in a fork)."""
state, _ = _load_run(args)
reap_lost(state)
out = tick_pass(state)
return {"state": state, **out}, EXIT_OK
# --- the supervisor -------------------------------------------------------------------
@@ -2304,15 +2535,17 @@ class Supervisor:
self.mu = threading.Lock()
self.flags: list[str] = []
self.last_output = _now()
# `messages` (M16): a progress counter of distinct assistant message ids; the result's and
# the ledger's `turns` is `result.num_turns`.
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,
"started_at": _iso(), "heartbeat_at": _iso(), "messages": 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["messages"] = 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()
@@ -2379,7 +2612,8 @@ class Supervisor:
hit_written = e["kind"] != "fork" # only a fork's first request refreshes (I3)
clock = SupClock(t0)
last_orphan_check = 0.0
last_turns = 0
last_messages = 0
last_tick = t0
while p.poll() is None:
now = _now()
skipped = clock.tick(now)
@@ -2390,7 +2624,7 @@ class Supervisor:
self.flags.append("clock_jump")
self.write_rt()
with self.mu:
refused, first_at, turns = self.ss.refused, self.ss.first_at, self.ss.turns
refused, first_at, messages = self.ss.refused, self.ss.first_at, self.ss.turns
quiet = now - self.last_output
if refused:
kill_tree(crec, alive_until=time.time())
@@ -2419,9 +2653,13 @@ class Supervisor:
detail = "no orchestrator call for %.0f s (orphan_after %.0f s)" % (
clock.orphan_age(now, last), orphan_s)
break
if now - self._last_write >= HEARTBEAT_S or (turns != last_turns and now - self._last_write >= 1.0):
last_turns = turns
if now - self._last_write >= HEARTBEAT_S or (messages != last_messages
and now - self._last_write >= 1.0):
last_messages = messages
self.write_rt()
if now - last_tick >= TICK_SUP_S: # keep warm (design 5): one pass a minute
last_tick = now
safe_tick(self.state, "supervisor %s" % self.job)
time.sleep(SUP_TICK_S)
exit_code = p.wait()
# leftovers of claude (backgrounded commands); claude exited just now
@@ -2476,7 +2714,7 @@ def _running_info(state: str, job: str, e: dict, now: float) -> dict:
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)}
"messages": int(rt.get("messages") or 0)}
def cmd_wait(args) -> tuple[dict, int]:
@@ -2490,12 +2728,17 @@ def cmd_wait(args) -> tuple[dict, int]:
unknown = [j for j in args.jobs if j not in reg["jobs"]]
if unknown:
raise ForksError("unknown job", jobs=unknown)
touch_orchestrator(state)
pings = [j for j in args.jobs if reg["jobs"][j].get("kind") == "ping"]
if pings:
raise ForksError("pings are not waited on; their results arrive in `pings`", jobs=pings)
touch_orchestrator(state, wait=True)
deadline = time.monotonic() + args.timeout
ticks = not in_child() # N3: a fork's `wait` never launches pings
next_tick = time.monotonic() # one tick pass at the start, then every TICK_WAIT_S
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"))
j for j, x in r["jobs"].items() if not x.get("settled_at") and x.get("kind") != "ping")
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
@@ -2503,6 +2746,9 @@ def cmd_wait(args) -> tuple[dict, int]:
while True:
reap_lost(state)
if ticks and time.monotonic() >= next_tick:
next_tick = time.monotonic() + TICK_WAIT_S
safe_tick(state, "wait")
members, finished, running, met = evaluate(read_registry(state))
left = deadline - time.monotonic()
if met or left <= 0:
@@ -2523,12 +2769,22 @@ def cmd_wait(args) -> tuple[dict, int]:
if met:
for j in finished:
reg["jobs"][j]["settled_at"] = _iso(now)
stamp_orchestrator(reg, now)
# Pings never come back through `done` and never count toward the exit code: each
# finished ping result is returned once, in `pings`.
ping_results = []
for j, e in sorted(reg["jobs"].items()):
if e.get("kind") != "ping" or e.get("status") not in FINAL or e.get("delivered_at"):
continue
res = read_json(result_file(state, j))
if res is not None:
ping_results.append(res)
e["delivered_at"] = _iso(now)
stamp_orchestrator(reg, now, wait=True)
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": []}
"pings": ping_results}
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)
@@ -2539,7 +2795,7 @@ LIST_JOB_KEYS = ("job", "kind", "base", "block", "phase", "mode", "paths", "stat
"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")
"created_at", "trained_at", "ttl_from", "ping_in_flight")
def cmd_list(args) -> tuple[dict, int]:
@@ -2561,7 +2817,7 @@ def cmd_list(args) -> tuple[dict, int]:
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")})
row.update({k: info[k] for k in ("elapsed_s", "last_event_s_ago", "messages")})
running.append(j)
else:
row["turns"] = e.get("turns")
@@ -2612,7 +2868,7 @@ HANDLERS = {
"_supervise": cmd_supervise,
"rebase": _not_implemented(phase=2),
"usage": _not_implemented(phase=2),
"tick": _not_implemented(phase=2),
"tick": cmd_tick,
}
plugins/spire/skills/forks/tests/test_forks.py
+558 -13
View File
@@ -759,7 +759,7 @@ class InitTests(Base):
class CliTests(Base):
def test_phase2_not_implemented(self):
run_log = os.path.join(self.tmp, "r-run.md")
for argv in (("rebase",), ("usage", "--phase", "1"), ("tick",)):
for argv in (("rebase",), ("usage", "--phase", "1")): # `tick` is implemented (2.1)
with self.subTest(cmd=argv[0]):
code, out, raw = call([argv[0], "--run", run_log, *argv[1:]])
self.assertEqual(code, 2)
@@ -777,7 +777,7 @@ class CliTests(Base):
("fork", "--base", "B1", "--block", "1.1", "--brief", brief),
("continue", "--job", "j0001", "--brief", brief),
("wait", "--all"), ("list",), ("expire", "--all"), ("kill", "--job", "j0001"),
("_supervise", "--job", "j0001"))
("_supervise", "--job", "j0001"), ("tick",))
for argv in cases:
with self.subTest(cmd=argv[0]):
code, out, raw = call([argv[0], "--run", run_log, *argv[1:]])
@@ -1028,10 +1028,13 @@ def main():
if "silent" in d:
time.sleep(float(d["silent"]))
turns = int(d.get("turns", "3"))
if opt(a, "--max-turns"):
turns = min(turns, int(opt(a, "--max-turns")))
api = d.get("apierr") # api_error_status of the result, e.g. 429
is_error = "error" in d or api is not None
if is_error:
if is_error and "billed" not in d:
turns = 0 # rejected before the model answered: nothing billed
five = d.get("ttl") == "5m" # ttl=5m: cache writes land in ephemeral_5m
ctx = int(prior.get("context") or 0)
tot = {"in": 0, "out": 0, "cw": 0, "cr": 0}
prev = ctx
@@ -1044,7 +1047,8 @@ def main():
cr, cw = prev, 300
inp, out = 2, 40
u = {"input_tokens": inp, "cache_creation_input_tokens": cw, "cache_read_input_tokens": cr,
"cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": cw},
"cache_creation": {"ephemeral_5m_input_tokens": cw if five else 0,
"ephemeral_1h_input_tokens": 0 if five else cw},
"output_tokens": 1, "service_tier": "standard"}
mid = "msg_stub_" + uuid.uuid4().hex[:12]
for block in ("thinking", "text"): # two events per message id, interim output_tokens
@@ -1073,7 +1077,8 @@ def main():
usage = {"input_tokens": tot["in"], "cache_creation_input_tokens": tot["cw"],
"cache_read_input_tokens": tot["cr"],
"output_tokens": tot["out"] + (7 if "mismatch" in d else 0),
"cache_creation": {"ephemeral_1h_input_tokens": tot["cw"], "ephemeral_5m_input_tokens": 0}}
"cache_creation": {"ephemeral_1h_input_tokens": 0 if five else tot["cw"],
"ephemeral_5m_input_tokens": tot["cw"] if five else 0}}
denials = ([{"tool_name": "Bash", "tool_use_id": "toolu_stub", "tool_input": {"command": "rm -rf build"}}]
if "deny" in d else [])
errtext = d.get("errtext", "").replace("_", " ")
@@ -1161,6 +1166,15 @@ def visible_windows() -> list[tuple]:
return out
# P0.1's detector: a visible console window (conhost) or Windows Terminal window. The latter belongs
# to WindowsTerminal.exe, not to the process tree that asked for it, so a tree-scoped check misses it.
CONSOLE_CLASSES = ("ConsoleWindowClass", "CASCADIA_HOSTING_WINDOW_CLASS")
def console_windows() -> list[tuple]:
return [w for w in visible_windows() if w[2] in CONSOLE_CLASSES]
def wait_for(pred, timeout=30.0, step=0.1):
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
@@ -1433,6 +1447,306 @@ class RuleTests(Base):
self.assertEqual(forks.runtime_drift(rt, None, {"permission_mode": "default"}), ["permissionMode"])
class WarmthTests(Base):
"""Design 5 under an injected clock (block 2.1): warmth boundaries, TTL inference, the ping rule."""
T0 = 1_800_000_000.0 # the base's last hit
E1H = T0 + 3600 - 100 # expiring at 1 h (margin 360)
E5M = T0 + 300 - 60 # expiring at 5 min (margin 120)
def make_reg(self, now, *, pings="on", margin=None, base_ttl=None, status="active", wait_ago=60.0):
reg = forks.new_registry()
reg["config"] = {"launch": {}, "ttl_s": 3600, "pings": pings, "warm_margin_s": margin}
reg["bases"]["B1"] = {"name": "B1", "status": status, "session": "s-b1",
"last_hit": forks._iso(self.T0), "ttl_s": base_ttl, "ping_in_flight": None}
reg["jobs"]["j0001"] = forks._new_entry("j0001", "train", base="B1", status="ok")
if wait_ago is not None:
reg["run"]["last_wait_at"] = forks._iso(now - wait_ago)
return reg
def add_job(self, reg, kind, base="B1", **kw):
j = forks._next_job_id(reg)
reg["jobs"][j] = forks._new_entry(j, kind, base=base, **kw)
return j
def decide(self, reg, now):
return forks.ping_decision(reg, "B1", now)
def test_boundaries_1h_and_5m(self):
T0 = self.T0
b = {"last_hit": forks._iso(T0)}
def st(base, cfg, t):
return forks.base_warmth(base, cfg, t)["base_state"]
cfg = {"ttl_s": 3600}
for t, want in ((T0 + 3239, "warm"), (T0 + 3240, "expiring"), (T0 + 3599, "expiring"),
(T0 + 3600, "cold")):
with self.subTest(ttl=3600, t=t - T0):
self.assertEqual(st(b, cfg, t), want)
w = forks.base_warmth(b, cfg, T0)
self.assertEqual((w["ttl_s"], w["margin_s"], forks._ts(w["warm_until"])), (3600, 360, T0 + 3600))
# 5 min: inferred on the base (config still 3600), and configured
for base, c in (({**b, "ttl_s": 300}, cfg), (b, {"ttl_s": 300})):
for t, want in ((T0 + 179, "warm"), (T0 + 180, "expiring"), (T0 + 299, "expiring"),
(T0 + 300, "cold")):
with self.subTest(ttl=300, base_ttl=base.get("ttl_s"), t=t - T0):
self.assertEqual(st(base, c, t), want)
w = forks.base_warmth(base, c, T0)
self.assertEqual((w["ttl_s"], w["margin_s"], forks._ts(w["warm_until"])), (300, 120, T0 + 300))
# init --warm-margin-s overrides the formula, at either TTL
m = {"ttl_s": 3600, "warm_margin_s": 900}
for t, want in ((T0 + 2699, "warm"), (T0 + 2700, "expiring"), (T0 + 3600, "cold")):
with self.subTest(margin=900, t=t - T0):
self.assertEqual(st(b, m, t), want)
self.assertEqual(forks.base_warmth(b, m, T0)["margin_s"], 900)
for t, want in ((T0, "expiring"), (T0 + 299, "expiring"), (T0 + 300, "cold")):
with self.subTest(margin=900, base_ttl=300, t=t - T0):
self.assertEqual(st({**b, "ttl_s": 300}, m, t), want)
self.assertEqual(st({}, cfg, T0), "cold")
def test_infer_ttl_and_downgrade(self):
def t(h=0, m=0, r=0):
return {"cache_creation_1h": h, "cache_creation_5m": m, "cache_read": r}
f = forks.infer_ttl
self.assertEqual(f(t(h=100, r=5), None), 3600) # 1h-only
self.assertEqual(f(t(h=100), 300), 3600)
self.assertEqual(f(t(m=100, r=5), 3600), 300) # 5m-only
self.assertEqual(f(t(h=1, m=1), 300), 3600) # both written: 1 h
self.assertEqual(f(t(r=500), 3600), 3600) # read-only keeps the known TTL
self.assertEqual(f(t(r=500), 300), 300)
self.assertIsNone(f(t(r=500), None)) # none known: the config's
self.assertEqual(f(t(r=500), None, 3600), 3600)
self.assertEqual(f(t(), 300), 300)
self.assertEqual(f(None, 300), 300)
self.assertTrue(forks.wrote_only_5m(t(m=100)))
self.assertFalse(forks.wrote_only_5m(t(h=1, m=1)))
self.assertFalse(forks.wrote_only_5m(t(h=100)))
self.assertFalse(forks.wrote_only_5m(t(r=5)))
self.assertFalse(forks.wrote_only_5m(None))
# warmth with no inferred TTL falls back to the configured one
b = {"last_hit": forks._iso(self.T0), "ttl_s": None}
self.assertEqual(forks.base_warmth(b, {"ttl_s": 300}, self.T0)["ttl_s"], 300)
def test_break_even(self):
self.assertEqual(forks.break_even(3600), 19)
self.assertEqual(forks.break_even(300), 11)
def test_ping_rule_states_status_and_pings_off(self):
now = self.E1H
d = self.decide(self.make_reg(now), now)
self.assertTrue(d["ping"], d)
self.assertIsNone(d["why"])
self.assertEqual((d["base_state"], d["break_even"], d["pings_since_real_call"]), ("expiring", 19, 0))
for t, why in ((self.T0 + 100, "state:warm"), (self.T0 + 3601, "state:cold")):
d = self.decide(self.make_reg(t), t)
self.assertEqual((d["ping"], d["why"]), (False, why))
for status in ("training", "retired", "superseded", "failed"):
d = self.decide(self.make_reg(now, status=status), now)
self.assertEqual((d["ping"], d["why"]), (False, "status:%s" % status))
d = self.decide(self.make_reg(now, pings="off"), now)
self.assertEqual((d["ping"], d["why"]), (False, "pings_off"))
# 5 min TTL inferred on the base, margin from init
d = self.decide(self.make_reg(self.E5M, base_ttl=300), self.E5M)
self.assertEqual((d["ping"], d["ttl_s"], d["break_even"]), (True, 300, 11))
t = self.T0 + 100
self.assertTrue(self.decide(self.make_reg(t, margin=3600), t)["ping"])
def test_ping_rule_activity_window(self):
now = self.E1H
self.assertTrue(self.decide(self.make_reg(now, wait_ago=599), now)["ping"])
for ago in (601, None):
d = self.decide(self.make_reg(now, wait_ago=ago), now)
self.assertEqual((d["ping"], d["why"]), (False, "no_activity"))
reg = self.make_reg(now, wait_ago=None)
self.add_job(reg, "train", base="B2", status="running") # not a fork/continue
self.add_job(reg, "ping", base="B2", status="running")
self.assertEqual(self.decide(reg, now)["why"], "no_activity")
for kind in ("fork", "continue"):
with self.subTest(kind=kind):
reg = self.make_reg(now, wait_ago=None)
j = self.add_job(reg, kind, base="B2", status="running")
self.assertTrue(self.decide(reg, now)["ping"]) # running now
reg["jobs"][j].update(status="ok", ended_at=forks._iso(now - 590))
self.assertTrue(self.decide(reg, now)["ping"]) # ended 590 s ago
reg["jobs"][j]["ended_at"] = forks._iso(now - 610)
self.assertEqual(self.decide(reg, now)["why"], "no_activity")
def test_ping_cap_19_and_11(self):
for now, base_ttl, cap in ((self.E1H, None, 19), (self.E5M, 300, 11)):
with self.subTest(cap=cap):
reg = self.make_reg(now, base_ttl=base_ttl)
for _ in range(cap - 1):
self.add_job(reg, "ping", status="ok")
d = self.decide(reg, now)
self.assertEqual((d["ping"], d["pings_since_real_call"], d["break_even"]), (True, cap - 1, cap))
self.add_job(reg, "ping", status="failed") # failed pings count too
d = self.decide(reg, now)
self.assertEqual((d["ping"], d["why"]), (False, "cap:%d/%d" % (cap, cap)))
self.add_job(reg, "ping", base="B2", status="ok") # another base's pings do not
self.assertEqual(self.decide(reg, now)["pings_since_real_call"], cap)
# a real call on the base starts the count again
self.add_job(reg, "fork", status="ok", ended_at=forks._iso(now - 5000))
d = self.decide(reg, now)
self.assertEqual((d["ping"], d["pings_since_real_call"]), (True, 0))
self.add_job(reg, "ping", status="ok")
self.assertEqual(self.decide(reg, now)["pings_since_real_call"], 1)
def test_ping_in_flight(self):
now = self.E1H
reg = self.make_reg(now)
j = self.add_job(reg, "ping", status="running")
reg["bases"]["B1"]["ping_in_flight"] = j
d = self.decide(reg, now)
self.assertEqual((d["ping"], d["why"], d["ping_in_flight"]), (False, "in_flight:%s" % j, j))
reg["jobs"][j]["status"] = "lost" # a stale mark (its job ended) does not block
d = self.decide(reg, now)
self.assertEqual((d["ping"], d["ping_in_flight"]), (True, None))
def test_pings_not_in_cap(self):
reg = self.make_reg(self.E1H)
self.add_job(reg, "ping", status="running")
self.assertEqual(forks._running_capped(reg), [])
f = self.add_job(reg, "fork", status="running")
self.assertEqual(forks._running_capped(reg), [f])
class TickTests(Base):
"""`tick_pass` on a real state dir and lock; the spawn and the binary check are stubbed, so no
process is started (C3d dedupe, the `tick` command, N3)."""
def setUp(self):
super().setUp()
os.makedirs(os.path.join(self.tmp, "reports"))
self.run_log = os.path.join(self.tmp, "reports", "2026-10-16-16-t-run.md")
self.state = forks.state_dir_for(self.run_log)
forks.ensure_state(self.state)
now = time.time()
self.stub = make_stub_binary(self.tmp)
self.binfo = {"path": self.stub, "version": STUB_VERSION, "size": 0, "mtime": 0}
launch = {"model": TEST_MODEL, "effort": "high", "permission_mode": "dontAsk", "tools": ["Read"],
"allowed_tools": ["Read"], "disallowed_tools": [],
"mcp": {"strict": True, "config": forks.mcp_empty_path(self.state)},
"append_system_prompt": {"path": forks.preamble_path(), "sha256": "x"}, "agent": None,
"cwd": self.tmp, "binary": self.binfo, "env": {"pinned": {}}}
reg = forks.new_registry()
reg["config"] = {"launch": launch, "ttl_s": 3600, "pings": "on", "warm_margin_s": None,
"max_minutes": 90}
reg["bases"]["B1"] = {"name": "B1", "status": "active", "session": "s-b1",
"last_hit": forks._iso(now - 3500), "cost_cum_usd": 1.25,
"model_usage_cum": {TEST_MODEL: {"inputTokens": 1}}, "context_tokens": 1000}
reg["jobs"]["j0001"] = forks._new_entry("j0001", "train", base="B1", status="ok")
reg["run"]["last_wait_at"] = forks._iso(now - 60)
with forks.RegistryLock(self.state):
forks.write_registry(self.state, reg)
p = subprocess.Popen([sys.executable, "-c", "pass"])
dead = forks.pid_record(p.pid)
p.wait()
self.spawned = []
def spawn(state, run, job):
self.spawned.append((run, job))
return {"ok": True, "method": "test", "pid": dead["pid"], "ctime": dead["ctime"]}
for name, fn in (("spawn_supervisor", spawn), ("_check_binary", lambda ref, la: (self.binfo, []))):
pt = mock.patch.object(forks, name, fn)
pt.start()
self.addCleanup(pt.stop)
def reg(self):
return forks.read_registry(self.state)
def test_two_tickers_one_ping(self):
"""C3(d): two tickers that both judged the base eligible (outside the lock) launch one ping;
the second skips the base as in flight."""
barrier = threading.Barrier(2, timeout=10)
def check(ref, la):
barrier.wait() # both are past the unlocked first look
return self.binfo, []
results, errors = [], []
def ticker():
try:
results.append(forks.tick_pass(self.state))
except Exception as e: # surfaced below
errors.append(e)
with mock.patch.object(forks, "_check_binary", check):
ths = [threading.Thread(target=ticker) for _ in range(2)]
for th in ths:
th.start()
for th in ths:
th.join(30)
self.assertEqual(errors, [])
self.assertEqual(len(results), 2)
pinged = [j for r in results for j in r["pinged"]]
self.assertEqual(len(pinged), 1, results)
job = pinged[0]
loser = [r for r in results if not r["pinged"]][0]
self.assertEqual(loser["bases"][0]["why"], "in_flight:%s" % job)
reg = self.reg()
self.assertEqual([j for j, e in reg["jobs"].items() if e["kind"] == "ping"], [job])
self.assertEqual(reg["bases"]["B1"]["ping_in_flight"], job)
self.assertEqual(self.spawned, [(forks.norm_path(self.run_log), job)])
e = reg["jobs"][job]
self.assertEqual((e["status"], e["base"], e["parent_session"], e["base_state_at_launch"]),
("running", "B1", "s-b1", "expiring"))
self.assertEqual((e["cost_before"], e["base_context"], e["max_s"]), (1.25, 1000, forks.PING_MAX_S))
a = e["argv"]
self.assertEqual(a[a.index("--resume") + 1], "s-b1")
self.assertIn("--fork-session", a)
self.assertEqual(a[a.index("--session-id") + 1], e["session"])
self.assertEqual(a[a.index("--max-turns") + 1], "1")
self.assertEqual(a[a.index("--model") + 1], TEST_MODEL)
with open(os.path.join(self.state, "briefs", job + ".md"), encoding="utf-8") as f:
self.assertEqual(f.read(), forks.PING_PROMPT)
self.assertEqual(forks._running_capped(reg), []) # pings are not in the cap
# a third ticker: still in flight
out = forks.tick_pass(self.state)
self.assertEqual((out["pinged"], out["bases"][0]["why"]), ([], "in_flight:%s" % job))
# the ping ends (any status): the mark clears, the result says it missed
res = forks.finalize_job(self.state, job, ss=None, hint="killed", reason="kill")
self.assertIn("ping_missed", res["flags"])
self.assertEqual(res["kind"], "ping")
self.assertIsNone(self.reg()["bases"]["B1"]["ping_in_flight"])
out = forks.tick_pass(self.state)
self.assertEqual(len(out["pinged"]), 1)
self.assertEqual(out["bases"][0]["pings_since_real_call"], 1) # counted before this one
def test_tick_command_and_n3(self):
code, out, raw = call(["tick", "--run", self.run_log])
self.assertEqual(code, 0, out)
self.assert_one_json_line(raw)
self.assertEqual(len(out["pinged"]), 1)
self.assertEqual(out["state"], self.state)
self.assertEqual(out["bases"][0]["job"], out["pinged"][0])
os.environ[forks.CHILD_MARKER] = "1"
code, out, raw = call(["tick", "--run", self.run_log])
self.assertEqual(code, 2, out)
self.assert_one_json_line(raw)
self.assertIn("inside a fork", out["error"])
self.assertEqual(len([e for e in self.reg()["jobs"].values() if e["kind"] == "ping"]), 1)
def test_tick_nothing_to_do_takes_no_lock_and_binary_missing(self):
with forks.RegistryLock(self.state):
reg = self.reg()
reg["config"]["pings"] = "off"
forks.write_registry(self.state, reg)
with forks.RegistryLock(self.state): # held: a pass with nothing to ping must not wait
out = forks.tick_pass(self.state)
self.assertEqual((out["pinged"], out["bases"][0]["why"]), ([], "pings_off"))
with forks.RegistryLock(self.state):
reg = self.reg()
reg["config"]["pings"] = "on"
forks.write_registry(self.state, reg)
def missing(ref, la):
raise forks.ForksError("claude binary missing", reason="binary_missing")
with mock.patch.object(forks, "_check_binary", missing):
out = forks.tick_pass(self.state)
self.assertEqual((out["pinged"], out["bases"][0]["why"]), ([], "binary_missing"))
self.assertEqual(self.spawned, [])
self.assertEqual(list(self.reg()["jobs"]), ["j0001"])
class LaunchTests(LaunchBase):
def test_train_fork_continue_end_to_end(self):
self.setup_run()
@@ -1731,7 +2045,9 @@ class LaunchTests(LaunchBase):
self.assertIn("stalled", res[hang]["flags"])
self.assert_tree_dead(tree) # claude tree killed; the supervisor exited
def test_kill_tree_and_no_visible_window(self):
def test_kill_tree(self):
"""The visible-window part moved to `test_no_visible_window` (M8), so that a host without a
desktop skips it instead of failing the kill test."""
self.setup_run()
self.train("B1")
self.finish("j0001")
@@ -1752,12 +2068,6 @@ class LaunchTests(LaunchBase):
self.assertEqual((parent or "").lower(), "wmiprvse.exe")
self.assertIn("python.exe", [p["name"].lower() for p in tree])
self.assertIn("cmd.exe", [p["name"].lower() for p in tree])
all_windows = visible_windows()
self.assertGreater(len(all_windows), 0, "window enumeration sees nothing (no desktop?)")
for _ in range(3):
mine = [w for w in visible_windows() if w[1] in pids]
self.assertEqual(mine, [], "visible window of the job's process tree")
time.sleep(0.2)
out = self.cmd("kill", "--job", job)
self.assertTrue(out["killed"])
self.assertTrue(out["ok"], out)
@@ -1853,7 +2163,8 @@ class LaunchTests(LaunchBase):
self.assertEqual([x["job"] for x in out["done"]], [a])
self.assertEqual(out["done"][0]["status"], "failed")
self.assertEqual([x["job"] for x in out["running"]], [b])
self.assertEqual(set(out["running"][0]), {"job", "block", "elapsed_s", "last_event_s_ago", "turns"})
# M16: the progress counter is `messages` (distinct assistant message ids), not `turns`
self.assertEqual(set(out["running"][0]), {"job", "block", "elapsed_s", "last_event_s_ago", "messages"})
self.assertEqual(out["pings"], [])
code, out = self.wait(timeout=60) # condition met, one failed -> 4
self.assertEqual(code, 4, out)
@@ -1954,6 +2265,240 @@ class LaunchTests(LaunchBase):
os.replace(moved, self.stub)
self.cmd("fork", "--base", "B1", "--block", "1", "--brief", os.path.join(self.tmp, "nope.md"), code=2)
# --- block 2.1 -----------------------------------------------------------------------
@unittest.skipUnless(IS_WINDOWS, "visible windows are a Windows check")
def test_no_visible_window(self):
"""M8: while a job runs, no visible window belongs to its process tree, and no visible
console or Windows Terminal window appears anywhere on the desktop (a Terminal window
belongs to WindowsTerminal.exe, not to the tree). A positive control, a deliberately visible
console (minimised, not activated) closed at once, shows the check sees one. Skipped, not
failed, when there is no desktop."""
if not visible_windows():
self.skipTest("no desktop: window enumeration sees no visible window")
# positive control
before = {w[0] for w in console_windows()}
marker = "forks-test-poscontrol-%d" % os.getpid() # matched by the module-end sweep
si = subprocess.STARTUPINFO()
si.dwFlags |= subprocess.STARTF_USESHOWWINDOW
si.wShowWindow = 7 # SW_SHOWMINNOACTIVE
ctl = subprocess.Popen(
[stub_interpreter(), "-I", "-c", "import ctypes, time; "
"ctypes.windll.kernel32.SetConsoleTitleW(%r); time.sleep(120)" % marker],
creationflags=subprocess.CREATE_NEW_CONSOLE, startupinfo=si)
ctl_rec = forks.pid_record(ctl.pid)
try:
seen = wait_for(lambda: [w for w in console_windows() if w[0] not in before or marker in w[3]],
timeout=15)
finally:
forks.kill_tree(ctl_rec, alive_until=time.time())
with contextlib.suppress(subprocess.TimeoutExpired):
ctl.wait(10)
self.assertTrue(seen, "positive control: a deliberately visible console was not detected")
hwnds = {w[0] for w in seen}
self.assertTrue(wait_for(lambda: not [w for w in console_windows() if w[0] in hwnds], timeout=15),
"positive control: its console window outlived the process")
# the job
self.setup_run()
self.train("B1")
self.finish("j0001")
before = {w[0] for w in console_windows()}
job = self.fork_marked("hang=1", "child")["job"]
rt = self.until_running(job, need_child=True)
try:
for _ in range(5):
pids = {p["pid"] for p in forks.process_tree(rt["supervisor"])}
self.assertIn(rt["claude"]["pid"], pids)
mine = [w for w in visible_windows() if w[1] in pids]
self.assertEqual(mine, [], "visible window of the job's process tree")
new = [w for w in console_windows() if w[0] not in before]
self.assertEqual(new, [], "a console or Terminal window appeared while the job ran")
time.sleep(0.2)
finally:
self.assertTrue(self.cmd("kill", "--job", job)["ok"])
def test_error_after_requested_model_billed(self):
"""The call billed the requested model, then its result is an error: `failed` with an error
reason (`is_error`, or the API-error reason), never `model_refused` (stub `billed`)."""
self.setup_run(cap=4)
self.train("B1")
self.finish("j0001")
a = self.fork("error", "billed", "turns=2")["job"]
b = self.fork("apierr=500", "billed", "turns=2")["job"]
code, out = self.wait(a, b)
self.assertEqual(code, 4, out)
res = {x["job"]: x for x in out["done"]}
for job, reason in ((a, "is_error"), (b, "api_error")):
with self.subTest(reason=reason):
r = res[job]
self.assertEqual((r["status"], r["reason"]), ("failed", reason), r)
self.assertEqual(r["models_used"], [TEST_MODEL])
self.assertEqual(r["tokens"]["output"], 2 * 40)
self.assertGreater(r["cost_delta_usd"], 0)
self.assertAlmostEqual(r["cost_delta_usd"], price(r["tokens"]), places=6)
def set_registry(self, fn):
with forks.RegistryLock(self.state):
reg = forks.read_registry(self.state)
fn(reg)
forks.write_registry(self.state, reg)
def set_last_hit(self, iso):
self.set_registry(lambda r: r["bases"]["B1"].__setitem__("last_hit", iso))
def set_pings(self, on):
self.set_registry(lambda r: r["config"].__setitem__("pings", "on" if on else "off"))
def test_ttl_inference_downgrade_and_cold_start(self):
"""Design 5 / C5: a call that wrote only 5-minute entries carries `ttl_downgrade` and sets the
base's TTL to 300 s; warmth, and so `cold_start`, follow the inferred TTL; a later 1 h write
brings it back to 3600 s."""
self.setup_run()
self.train("B1", "ttl=5m")
r = self.finish("j0001")
self.assertEqual(r["status"], "ok", r)
self.assertIn("ttl_downgrade", r["flags"])
self.assertEqual(r["tokens"]["cache_creation_1h"], 0)
self.assertGreater(r["tokens"]["cache_creation_5m"], 0)
b = self.reg()["bases"]["B1"]
self.assertEqual((b["ttl_s"], b["ttl_from"]), (300, "j0001"))
row = self.cmd("list")["bases"][0]
self.assertEqual((row["ttl_s"], row["margin_s"], row["ttl_from"]), (300, 120, "j0001"))
# 400 s after the last hit: warm under the configured 3600 s, cold under the inferred 300 s
self.set_last_hit(forks._iso(time.time() - 400))
self.assertEqual(self.cmd("list")["bases"][0]["base_state"], "cold")
out = self.fork("turns=1")
self.assertEqual(out["base_state"], "cold")
r2 = self.finish(out["job"])
self.assertEqual((r2["status"], r2["cache"], r2["base_state"]), ("ok", "hit", "cold"), r2)
self.assertIn("cold_start", r2["flags"])
self.assertNotIn("ttl_downgrade", r2["flags"])
b = self.reg()["bases"]["B1"]
self.assertEqual((b["ttl_s"], b["ttl_from"]), (3600, out["job"]))
row = self.cmd("list")["bases"][0]
self.assertEqual((row["base_state"], row["ttl_s"]), ("warm", 3600))
def ping_jobs(self):
return sorted(j for j, e in self.reg()["jobs"].items() if e["kind"] == "ping")
def collect_ping(self, job):
"""Wait (pings off) until the ping ended; return it from `wait`'s `pings` list."""
self.assertTrue(wait_for(lambda: self.reg()["jobs"][job]["status"] in forks.FINAL, timeout=60),
self.reg()["jobs"][job])
code, out = self.wait(timeout=5)
self.assertEqual((code, out["done"]), (0, []), out)
self.assertEqual([x["job"] for x in out["pings"]], [job], out)
return out["pings"][0]
def test_ping_end_to_end_through_wait(self):
"""A stubbed ping: launched by `wait`'s tick, returned once through `wait`'s `pings` (never
`done`); a `partial` ping refreshes last_hit, a missed or failed one does not and carries
`ping_missed`; the in-flight mark clears; each ping has its ledger line."""
self.setup_run("--warm-margin-s", "3600") # expiring from the last hit on
self.train("B1")
self.finish("j0001")
base_session = self.reg()["bases"]["B1"]["session"]
old = forks._iso(time.time() - 100)
self.set_last_hit(old)
# 1. partial, launched by wait's tick (the wait itself is the activity)
with mock.patch.object(forks, "PING_PROMPT", "Reply OK\n@stub cache=partial"):
code, out = self.wait(timeout=5)
self.assertEqual((code, out["done"], out["pings"]), (0, [], []), out)
pings = self.ping_jobs()
self.assertEqual(len(pings), 1, self.reg()["jobs"])
p1 = pings[0]
self.assertEqual(self.reg()["bases"]["B1"]["ping_in_flight"], p1)
self.set_pings(False) # no further pings from the waits below
r1 = self.collect_ping(p1)
self.assertEqual((r1["kind"], r1["status"], r1["cache"], r1["base"], r1["base_state"]),
("ping", "ok", "partial", "B1", "expiring"), r1)
self.assertNotIn("ping_missed", r1["flags"])
self.assertEqual((r1["parent_session"], r1["turns"], r1["block"]), (base_session, 1, None))
self.assertEqual(r1["models_used"], [TEST_MODEL])
b = self.reg()["bases"]["B1"]
self.assertGreater(forks._ts(b["last_hit"]), forks._ts(old)) # partial refreshes
self.assertIsNone(b["ping_in_flight"])
self.assertEqual(self.wait(timeout=5)[1]["pings"], []) # returned once
c = stub_call_for(self.stub, "Reply OK")
a = c["argv"]
self.assertEqual(a[a.index("--resume") + 1], base_session)
self.assertIn("--fork-session", a)
self.assertEqual(a[a.index("--session-id") + 1], r1["session"])
self.assertEqual(a[a.index("--max-turns") + 1], "1")
self.assertEqual(c["env"]["SPIRE_FORKS_CHILD"], "1")
# 2. a miss, launched by `tick` by hand: last_hit stays, flag ping_missed
self.set_last_hit(old)
self.set_pings(True)
with mock.patch.object(forks, "PING_PROMPT", "Reply OK\n@stub cache=miss"):
out = self.cmd("tick")
self.assertEqual(len(out["pinged"]), 1, out)
p2 = out["pinged"][0]
d = [x for x in out["bases"] if x["base"] == "B1"][0]
self.assertEqual((d["pings_since_real_call"], d["break_even"], d["job"]), (1, 19, p2))
self.set_pings(False)
r2 = self.collect_ping(p2)
self.assertEqual((r2["status"], r2["cache"]), ("ok", "miss"), r2)
self.assertIn("ping_missed", r2["flags"])
self.assertEqual(self.reg()["bases"]["B1"]["last_hit"], old)
# 3. a failed ping: last_hit stays, flag ping_missed
self.set_pings(True)
with mock.patch.object(forks, "PING_PROMPT", "Reply OK\n@stub error"):
p3 = self.cmd("tick")["pinged"][0]
self.set_pings(False)
r3 = self.collect_ping(p3)
self.assertEqual((r3["status"], r3["reason"]), ("failed", "is_error"), r3)
self.assertIn("ping_missed", r3["flags"])
self.assertEqual(self.reg()["bases"]["B1"]["last_hit"], old)
self.assertIsNone(self.reg()["bases"]["B1"]["ping_in_flight"])
led = [x for x in self.ledger() if x["kind"] == "ping"]
self.assertEqual([x["job"] for x in led], [p1, p2, p3])
self.assertEqual([x["status"] for x in led], ["ok", "ok", "failed"])
def test_tick_runs_in_wait_and_supervisor(self):
"""The tick runs inside `wait` (at its start, then every TICK_WAIT_S; never in a fork) and
inside every supervisor (every TICK_SUP_S)."""
self.setup_run()
self.train("B1")
self.finish("j0001")
calls = []
def rec(state):
calls.append(state)
return {"pinged": [], "bases": []}
with mock.patch.object(forks, "TICK_WAIT_S", 0.5), mock.patch.object(forks, "tick_pass", rec):
job = self.fork("sleep=1", "turns=3")["job"]
t0 = time.monotonic()
self.finish(job)
took = time.monotonic() - t0
self.assertGreaterEqual(took, 1.5)
self.assertGreaterEqual(len(calls), 3, (len(calls), took))
self.assertEqual(set(calls), {self.state})
calls.clear()
os.environ[forks.CHILD_MARKER] = "1"
try:
with mock.patch.object(forks, "tick_pass", rec):
self.wait(job, timeout=1)
finally:
os.environ.pop(forks.CHILD_MARKER, None)
self.assertEqual(calls, [])
# a supervisor, run in-process
p = subprocess.Popen([sys.executable, "-c", "pass"])
dead = forks.pid_record(p.pid)
p.wait()
fake = {"ok": True, "method": "test", "pid": dead["pid"], "ctime": dead["ctime"]}
with mock.patch.object(forks, "spawn_supervisor", lambda *a: fake):
job = self.fork("sleep=1", "turns=3")["job"]
sup = forks.Supervisor(self.state, job)
sup.rt["supervisor"] = dead
with mock.patch.dict(os.environ), mock.patch.object(forks, "TICK_SUP_S", 0.5), \
mock.patch.object(forks, "tick_pass", rec):
out, code = sup.run()
self.assertEqual(code, 0, out)
self.assertEqual(self.reg()["jobs"][job]["status"], "ok")
self.assertGreaterEqual(len(calls), 3, len(calls))
self.assertIn("messages", self.runtime(job)) # M16: the job file's counter
self.assertNotIn("turns", self.runtime(job))
def tearDownModule():
"""No supervisor or stub of this suite may survive it (checked by command line, never by name)."""