diff --git a/plugins/spire/skills/forks/forks.py b/plugins/spire/skills/forks/forks.py index 184455d..460639b 100644 --- a/plugins/spire/skills/forks/forks.py +++ b/plugins/spire/skills/forks/forks.py @@ -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, } diff --git a/plugins/spire/skills/forks/tests/test_forks.py b/plugins/spire/skills/forks/tests/test_forks.py index 5fbebc2..a3e91af 100644 --- a/plugins/spire/skills/forks/tests/test_forks.py +++ b/plugins/spire/skills/forks/tests/test_forks.py @@ -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)."""