From 61c239648aa34dc6a6a326005dc26312220fcc0c Mon Sep 17 00:00:00 2001 From: lukemaj Date: Sun, 27 Sep 2026 22:11:52 +0200 Subject: [PATCH 1/3] wip: rebase #132 controller and interrupt fixes onto 0.40.0 --- GLOSSARY.md | 2 +- RUNNER.md | 20 +- runner/controller.py | 180 ++++++++++++----- runner/core.py | 98 ++++----- runner/t3exec.py | 254 ++++++++++++++++++++---- tests/test_issue132.py | 436 +++++++++++++++++++++++++++++++++++++++++ tests/test_t3_only.py | 12 +- 7 files changed, 858 insertions(+), 144 deletions(-) create mode 100644 tests/test_issue132.py diff --git a/GLOSSARY.md b/GLOSSARY.md index a3ea56c..eeadeb8 100644 --- a/GLOSSARY.md +++ b/GLOSSARY.md @@ -14,7 +14,7 @@ Canonical terms for Model Router. One term per concept; avoid the synonyms. | Route | One dispatchable choice: a T3 provider instance, model, and effort, with its pool and family. A Prism preference with no policy route runs as `t3::@`. | model (alone) | | Pool | A subscription allowance a route draws from: Zen free, Go, xAI, Codex, Claude. Its meter is one T3 provider's usage windows (Go is one OpenCode account meter); Zen free has none. An exhaustion error rests the whole pool. | provider, account | | Signal class | How provider evidence is classified: exhausted, overloaded, stalled, context, hard. | error type | -| Stall | A T3 turn with no streaming tokens and no running tool call past about a minute (`T3_SILENCE_SECS`), confirmed by a fresh snapshot read. | hang | +| Stall | A T3 turn with no streaming tokens and no running tool call past its driver's measured window (`T3_SILENCE_SECS_BY_DRIVER`) of awake time, confirmed by a fresh snapshot read. | hang | | Assumed reset | The reset of a limit error that names none: one 5-hour window, flagged as assumed rather than provider-reported. | guessed reset | | Context window | A route's input context in tokens as policy data; a context-overflow turn moves to a larger one in its lane. | context size | | Retry | The automatic re-attempt of a failed worker turn: first in the same worker thread, then fresh on the Retry route. Its stage and Prism role keep the internal key `correction` so the snapshot contract stays stable. Prism can switch it off (`enabled: false`), which drops both rungs. | correction (in user-facing text) | diff --git a/RUNNER.md b/RUNNER.md index 35b45cb..cae768b 100644 --- a/RUNNER.md +++ b/RUNNER.md @@ -204,10 +204,15 @@ Liveness is activity-based: a turn is healthy while assistant tokens stream (`tool.started` until `tool.completed`/`tool.denied`, `task.started` until `task.completed`), so a long test run is never a stall. A tool held behind an unresolved approval or user-input request is not running. Silence with -no running tool past about a minute (`T3_SILENCE_SECS`, 60) is probed at -once with a fresh snapshot read; explicit provider errors act at once. A -message posted onto an existing thread is watched only through the turn it -starts. A stalled worker turn is interrupted and confirmed idle before any +no running tool past the thread driver's window is probed at once with a +fresh snapshot read; explicit provider errors act at once. The windows come +from the largest healthy gaps measured per driver, times 1.25: Codex 300 s, +OpenCode 210 s, Claude 90 s, anything else 300 s +(`T3_SILENCE_SECS_BY_DRIVER`). Time the host spends asleep never counts as +silence. Every watched turn records its longest silence and time to first +token as a `t3_turn_stream` event. A message posted onto an existing thread +is watched only through the turn it starts. A stalled worker turn is +interrupted, then polled until T3 reports the turn settled, before any route move. After a T3 server restart with `continueThreadsAfterServerUpdate` on, interrupted Claude, Codex and OpenCode turns continue on the same thread; Grok turns are re-sent by the router. @@ -384,7 +389,12 @@ to the database, events, or logs; worker text, proof output, and error evidence are redacted before they are persisted or forwarded. States: `pending -> running <-> question_pending -> succeeded | failed | -cancelled`, with `cancelling` while the controller and proof group stop. +cancelled`, with `cancelling` while the controller, proof group and saved +T3 turns stop. When T3 cannot be reached, a saved turn whose last known +state is completed, errored or interrupted needs no T3 call; otherwise the +reason names T3 unreachability. The controller keeps stepping while the job +is `pending`, `running` or `question_pending`; a step that stops with the +job still active blocks it as `controller_exit_unhandled: `. `blocked` always has a reason. `recover` re-evaluates only its own blocks (a dead controller, a pending launch, step budgets, unresolved proof ownership); any other block stays until the planner answers or resubmits. diff --git a/runner/controller.py b/runner/controller.py index 415979c..d3fa6ff 100644 --- a/runner/controller.py +++ b/runner/controller.py @@ -23,6 +23,14 @@ from . import adapters, core, policy, store, t3exec, t3snapshot MAX_LOOP_STEPS = 12 +# The controller keeps stepping while the job has one of these statuses. +# ``question_pending`` is active: the next step asks the planner in its T3 +# thread, and a question left without a reply blocks the job +# (``planner_question_pending``), which is the "waiting on the planner" stop. +ACTIVE_STEP_STATUSES = ("pending", "running", "question_pending") +# Step actions that mean "stop here"; with the job still active they block +# it as ``controller_exit_unhandled`` instead of orphaning it (#132). +STOP_STEP_ACTIONS = ("blocked", "cancelled", "noop-terminal") # Lease token of the controller process that owns this job. Every durable # controller write checks it under the write lock; tests and helpers that @@ -341,8 +349,13 @@ def _t3_thread_for(job: dict, slot: str) -> dict | None: def _save_t3_thread(state_dir, request_id: str, slot: str, thread_id: str, route: str | None = None, created: bool | None = None, - turn_started: bool | None = None) -> None: - """Record a T3 child thread id for adoption by later controllers.""" + turn_started: bool | None = None, + turn_state: str | None = None) -> None: + """Record a T3 child thread id for adoption by later controllers. + + ``turn_state`` is the last known state of the thread's latest turn + (``running``, or a terminal ``completed``/``error``/``interrupted``): + cancel reads it when T3 is unreachable (#132).""" con = store.connect(state_dir) try: con.execute("BEGIN IMMEDIATE") @@ -363,12 +376,17 @@ def _save_t3_thread(state_dir, request_id: str, slot: str, threads = {} previous = threads.get(slot) if isinstance(threads.get(slot), dict) else {} record = dict(previous) - record.update({"thread_id": thread_id, "route": route, - "updated_at": core._utcnow()}) + record.update({"thread_id": thread_id, "updated_at": core._utcnow()}) + if route is not None or "route" not in record: + record["route"] = route if created is not None: record["created"] = bool(created) if turn_started is not None: record["turn_started"] = bool(turn_started) + if turn_started: + record["turn_state"] = "running" + if turn_state is not None: + record["turn_state"] = turn_state threads[slot] = record st["t3_threads"] = threads con.execute("UPDATE jobs SET controller_state=?, updated_at=? WHERE request_id=?", @@ -387,6 +405,46 @@ def _save_t3_thread(state_dir, request_id: str, slot: str, con.close() +def _note_t3_turn(state_dir, request_id: str, slot: str, thread_id: str | None, + route: str | None, outcome: dict) -> None: + """Record a watched turn's end: its last known state on the saved + thread, and its stream statistics (max silence, time to first token) + in the ledger. Best effort: evidence never blocks the job.""" + state = outcome.get("state") + if not thread_id: + return + try: + _save_t3_thread(state_dir, request_id, slot, thread_id, route, + turn_state=state if state in t3exec.TURN_TERMINAL_STATES + else "running") + except Exception: + pass + stream = outcome.get("stream") + if isinstance(stream, dict): + try: + core._record_t3_interrupt_event( + state_dir, request_id, "t3_turn_stream", + {"slot": str(slot)[:48], "route": str(route or "")[:64], + "state": str(state or "")[:16], + **{k: stream.get(k) for k in ("max_silence", "first_token_secs", + "window", "slept")}}) + except Exception: + pass + + +def _t3_post_and_watch(state_dir, request_id: str, slot: str, client, + thread_id: str, text: str, route: str | None = None, + **kwargs) -> dict: + """``t3exec.post_and_watch`` on a saved thread, keeping its last known + turn state current (running before the post, the outcome after).""" + _save_t3_thread(state_dir, request_id, slot, thread_id, route, + turn_state="running") + outcome = t3exec.post_and_watch(client, thread_id, text, route=route, + **kwargs) + _note_t3_turn(state_dir, request_id, slot, thread_id, route, outcome) + return outcome + + def _check_t3_start_lease(state_dir, request_id: str) -> None: """Fence a turn start against a cancellation committed after the step guard.""" con = store.connect(state_dir) @@ -446,9 +504,9 @@ def _t3_repair_envelope(state_dir, request_id: str, *, thread_id: str, "only: exactly one JSON object whose action is " "planner_question, implementation or completion, and nothing else.") try: - outcome = t3exec.post_and_watch(client, thread_id, prompt, - role="dispatch", - watch_kwargs=_t3_watch_kwargs()) + outcome = _t3_post_and_watch(state_dir, request_id, "dispatch", client, + thread_id, prompt, role="dispatch", + watch_kwargs=_t3_watch_kwargs()) except t3exec.T3Error as e: _mark_blocked(state_dir, request_id, f"t3_unavailable: {e}") return None @@ -533,6 +591,8 @@ def _t3_run_turn(state_dir, request_id: str, job: dict, *, slot: str, except Exception: pass outcome["slot"] = slot + _note_t3_turn(state_dir, request_id, slot, outcome.get("thread_id"), route, + outcome) if outcome.get("state") == "unknown": # A failed reconciliation read is recoverable. Do not turn it into a # sticky dispatch or worker failure while the remote state is unknown. @@ -565,9 +625,10 @@ def _t3_dispatch_outcome(state_dir, request_id: str, prompt: str, # thread, then accept whatever that watch reports. job = core.get_job(state_dir, request_id) try: - second = t3exec.post_and_watch(client, thread_id, prompt, route=route, - role="dispatch", - watch_kwargs=_t3_watch_kwargs()) + second = _t3_post_and_watch(state_dir, request_id, "dispatch", + client, thread_id, prompt, route=route, + role="dispatch", + watch_kwargs=_t3_watch_kwargs()) except t3exec.T3Error as e: return _t3_block(state_dir, request_id, f"t3_unavailable: {e}") second["thread_id"] = thread_id @@ -755,8 +816,9 @@ def _resume_luna_via_t3(state_dir, request_id: str, message: str, route = saved.get("route") or _dispatcher_route(job) try: client = _t3_client_for_job(job, t3_client) - outcome = t3exec.post_and_watch(client, thread_id, message, role="dispatch", - watch_kwargs=_t3_watch_kwargs()) + outcome = _t3_post_and_watch(state_dir, request_id, "dispatch", client, + thread_id, message, role="dispatch", + watch_kwargs=_t3_watch_kwargs()) except t3exec.T3Error as e: return _t3_block(state_dir, request_id, f"t3_unavailable: {e}") try: @@ -852,34 +914,37 @@ def _run_t3_worker_turn(state_dir, request_id: str, job: dict, # The server cut the turn (restart): continue once on the same # thread like the dispatcher path, then accept the second watch. try: - second = t3exec.post_and_watch(client, thread_id, prompt, - route=route, role=role, - watch_kwargs=_t3_watch_kwargs()) + second = _t3_post_and_watch(state_dir, request_id, slot, client, + thread_id, prompt, route=route, role=role, + watch_kwargs=_t3_watch_kwargs()) second["thread_id"] = thread_id outcome = second except t3exec.T3Error as e: return _t3_block(state_dir, request_id, f"t3_unavailable: {e}") if outcome.get("state") == "stalled": # Never start a second writer behind a possibly live server-side - # turn: interrupt first, then confirm idle before the shared - # stall path may move routes. - try: - client.dispatch({"type": "thread.turn.interrupt", - "commandId": f"cmd-{request_id}-seq{seq}-{secrets.token_hex(4)}", - "threadId": thread_id, - "createdAt": t3exec._utcnow_iso()}) - except Exception: - pass - try: - snap = client.thread_snapshot(thread_id) - ev = t3exec.evaluate_turn(snap, now=time.time(), - silence_secs=t3exec.T3_SILENCE_SECS) - outcome["idle_confirmed"] = ev.get("state") in ( - "completed", "error", "interrupted") - if ev.get("state") == "completed" and not outcome.get("assistant_text"): - outcome["assistant_text"] = t3exec.latest_assistant_text(snap) - except Exception: - outcome["idle_confirmed"] = False + # turn: interrupt first, then poll until the turn settles (T3 + # applies the interrupt asynchronously) before the shared stall + # path may move routes. + settle = t3exec.interrupt_and_settle( + client, thread_id, + command={"type": "thread.turn.interrupt", + "commandId": f"cmd-{request_id}-seq{seq}-{secrets.token_hex(4)}", + "threadId": thread_id, + "createdAt": t3exec._utcnow_iso()}) + outcome["idle_confirmed"] = settle["settled"] + if settle["settled"]: + try: + _save_t3_thread(state_dir, request_id, slot, thread_id, route, + turn_state=(settle["state"] + if settle["state"] in t3exec.TURN_TERMINAL_STATES + else "interrupted")) + except core.LeaseLostError: + raise + except Exception: + pass + if settle["state"] == "completed" and not outcome.get("assistant_text"): + outcome["assistant_text"] = t3exec.latest_assistant_text(settle["snapshot"]) full, rc = _t3_worker_full(outcome, thread_id) return _finish_worker_turn(state_dir, request_id, job, route, seq, artifact, full, thread_id, rc, @@ -2924,15 +2989,17 @@ def _review_block(state_dir, request_id: str, reason: str) -> dict: return _t3_block(state_dir, request_id, reason[:500]) -def _review_verdict_repair(client, thread_id: str, text: str) -> tuple[dict | None, str]: +def _review_verdict_repair(state_dir, request_id: str, slot: str, client, + thread_id: str, text: str) -> tuple[dict | None, str]: """Ask the review thread once for the verdict JSON only.""" prompt = ("VERDICT REPAIR: your last reply had no parsable verdict " f"({_review_verdict_error(text)}). Reply with exactly one JSON " 'object and nothing else: {"verdict": "approve" | "request_changes", ' '"findings": "..."}.') try: - outcome = t3exec.post_and_watch(client, thread_id, prompt, role="review", - watch_kwargs=_t3_watch_kwargs()) + outcome = _t3_post_and_watch(state_dir, request_id, slot, client, thread_id, + prompt, role="review", + watch_kwargs=_t3_watch_kwargs()) except t3exec.T3Error as e: return None, f"repair post failed: {e}" reply = outcome.get("assistant_text") or "" @@ -3013,8 +3080,9 @@ def _run_review(state_dir, request_id: str, job: dict, head: str, if outcome.get("state") == "interrupted": # The server cut the turn (restart): continue once on the same thread. try: - second = t3exec.post_and_watch(client, thread_id, prompt, route=route, - role="review", watch_kwargs=_t3_watch_kwargs()) + second = _t3_post_and_watch(state_dir, request_id, slot, client, + thread_id, prompt, route=route, + role="review", watch_kwargs=_t3_watch_kwargs()) except t3exec.T3Error as e: return _t3_block(state_dir, request_id, f"t3_unavailable: {e}") second["thread_id"] = thread_id @@ -3025,7 +3093,8 @@ def _run_review(state_dir, request_id: str, job: dict, head: str, return _review_block(state_dir, request_id, f"review_failed: {detail}") verdict = parse_review_verdict(text) if verdict is None: - verdict, why = _review_verdict_repair(client, thread_id, text) + verdict, why = _review_verdict_repair(state_dir, request_id, slot, + client, thread_id, text) if verdict is None: return _review_block(state_dir, request_id, f"review_missing_verdict: {why}") if _workspace_state(workspace) != baseline.get("state"): @@ -4306,12 +4375,11 @@ def run_controller_process(state_dir: str, request_id: str, token: str) -> int: _LEASE["token"] = token # Bounded durable loop: every turn runs as a T3 thread and every # decision is persisted before its side effect. - # An exhausted escalation asks the planner in its thread on the next - # step, so the planner wakes without a manual recover. - continuing = ("dispatched", "question-answered-resumed", - "implementation-resumed", "completion-refused-resumed", - "transferred_to_go", "route_switched", - "stalled_retry", "recovery-exhausted-question") + # Continuation follows the job's status, never a list of step actions + # (#132): the loop runs while the job is active and stops once it is + # terminal, blocked (including waiting on the planner) or cancelling. + # A step that reports a stop while the job is still active blocks it + # with a named reason, so the planner is always woken. for _ in range(MAX_LOOP_STEPS): try: total = _count_step(state_dir, request_id) @@ -4342,7 +4410,25 @@ def run_controller_process(state_dir: str, request_id: str, token: str) -> int: _advertise(state_dir, request_id, token) except Exception: pass - if (res or {}).get("action") not in continuing: + try: + job = core.get_job(state_dir, request_id) + if job["status"] not in ACTIVE_STEP_STATUSES or job.get("cancel_requested"): + break + action = (res or {}).get("action") + if not action or action in STOP_STEP_ACTIONS: + why = (res or {}).get("reason") + _mark_blocked(state_dir, request_id, + f"controller_exit_unhandled: {action}" + + (f" ({why})" if why else "")) + break + except core.LeaseLostError: + return 0 + except Exception as e: # noqa: BLE001 - persisted, never silent + try: + _mark_blocked(state_dir, request_id, + f"controller_error: {type(e).__name__}: {str(e)[:300]}") + except Exception: + pass break else: try: diff --git a/runner/core.py b/runner/core.py index 883e0f7..2411d53 100644 --- a/runner/core.py +++ b/runner/core.py @@ -1805,8 +1805,9 @@ def cancel(state_dir, request_id: str) -> dict: # children confirm dead. children_dead = _drain_owned_children(state_dir, request_id, owner) proof_dead = _drain_proof_group(state_dir, request_id) - remote_settled = (_interrupt_saved_t3_threads(state_dir, request_id, dict(job)) - if children_dead and proof_dead else False) + remote_settled, remote_detail = ( + _interrupt_saved_t3_threads(state_dir, request_id, dict(job)) + if children_dead and proof_dead else (False, "")) all_dead = bool(children_dead and proof_dead and remote_settled) # Best-effort stop of the legacy owned worker PID when ownership proven. @@ -1821,71 +1822,71 @@ def cancel(state_dir, request_id: str) -> dict: _finalize_stopped(state_dir, request_id, all_dead, remote_settled, local_settled=bool(children_dead and proof_dead), - controller_fenced=children_dead) + controller_fenced=children_dead, + remote_detail=remote_detail) return get_job(state_dir, request_id) -def _interrupt_saved_t3_threads(state_dir, request_id: str, job: dict) -> bool: - """Interrupt and confirm every saved T3 child turn is settled.""" +def _interrupt_saved_t3_threads(state_dir, request_id: str, job: dict) -> tuple[bool, str]: + """Interrupt every saved T3 child turn and confirm it settled. + + Returns ``(settled, detail)``; ``detail`` names why a turn is not + confirmed stopped. When T3 cannot be reached, a turn whose last known + state (recorded by the controller) is terminal needs no T3 call (#132). + """ try: state = json.loads(job.get("controller_state") or "{}") threads = state.get("t3_threads") if isinstance(state, dict) else {} except (ValueError, TypeError): threads = {} if not isinstance(threads, dict) or not threads: - return True + return True, "" try: client = t3exec.client_for_job(job) - except t3exec.T3Error: - _record_t3_interrupt_event( - state_dir, request_id, "t3_turn_interrupt_failed", - {"error": "T3 client unavailable"}) - return False - settled = True + unreachable = "" + except t3exec.T3Error as e: + client = None + unreachable = f"T3 unreachable: {e}"[:300] + settled, detail = True, "" for slot, record in threads.items(): thread_id = record.get("thread_id") if isinstance(record, dict) else None if not thread_id: continue + last_known = record.get("turn_state") try: + if client is None: + raise t3exec.T3Error(unreachable) snapshot = client.thread_snapshot(thread_id) - thread = t3exec.snapshot_thread(snapshot) - if not thread or "latestTurn" not in thread: - settled = False - continue - latest = thread.get("latestTurn") - state = latest.get("state") if isinstance(latest, dict) else None - if latest is None or state in ("completed", "error", "interrupted"): + if t3exec.turn_settled(snapshot): continue - result = client.dispatch(t3exec.turn_interrupt_command(thread_id)) - confirmed = client.thread_snapshot(thread_id) - confirmed_thread = t3exec.snapshot_thread(confirmed) - if not confirmed_thread or "latestTurn" not in confirmed_thread: - settled = False - continue - confirmed_latest = confirmed_thread.get("latestTurn") - confirmed_state = (confirmed_latest.get("state") - if isinstance(confirmed_latest, dict) else None) - if (confirmed_latest is not None - and confirmed_state not in ("completed", "error", "interrupted")): - settled = False + res = t3exec.interrupt_and_settle(client, thread_id) _record_t3_interrupt_event( state_dir, request_id, "t3_turn_interrupt", {"slot": slot, "thread_id": thread_id, - "result": result if isinstance(result, dict) else {}, - "settled": confirmed_latest is None or confirmed_state in ( - "completed", "error", "interrupted"), - "state": confirmed_state}) + "result": res.get("result") if isinstance(res.get("result"), dict) else {}, + "settled": res["settled"], "state": res["state"], + **({"error": res["error"]} if res.get("error") else {})}) + if not res["settled"]: + settled = False + detail = detail or ("T3 turn not confirmed stopped after interrupt" + + (f": {res['error']}" if res.get("error") else "")) except t3exec.T3NotFoundError: # After the controller exits, a typed 404 means this child was # never observed remotely. Connection errors are not absence. continue except Exception as e: # best effort; local cancellation still proceeds + if last_known in t3exec.TURN_TERMINAL_STATES: + continue settled = False + reason = (f"{e}" if isinstance(e, t3exec.T3Error) else + f"{type(e).__name__}: {e}")[:300] + detail = detail or (reason if reason.startswith("T3 unreachable") + else f"T3 unreachable: {reason}") _record_t3_interrupt_event( state_dir, request_id, "t3_turn_interrupt_failed", - {"slot": slot, "thread_id": thread_id, - "error": str(e)[:300]}) - return settled + {"slot": slot, "thread_id": thread_id, "error": reason, + "last_known_state": last_known}) + return settled, detail def _record_t3_interrupt_event(state_dir, request_id: str, kind: str, @@ -1907,7 +1908,8 @@ def _record_t3_interrupt_event(state_dir, request_id: str, kind: str, def _finalize_stopped(state_dir, request_id: str, all_dead: bool, remote_settled: bool = True, local_settled: bool = True, - controller_fenced: bool = True) -> None: + controller_fenced: bool = True, + remote_detail: str = "") -> None: """Finish a cancel or timeout drain in one guarded statement. The outcome follows the intent stored at commit time (cancel=1 wins @@ -1923,12 +1925,15 @@ def _finalize_stopped(state_dir, request_id: str, all_dead: bool, status = ("cancelling" if not controller_fenced or (not remote_settled and local_settled) else "blocked") + # Name what is actually still live: a local process group, or + # a T3 turn that could not be confirmed stopped (#132). + what = ("live child process group remains" if not local_settled + else remote_detail or "T3 turn not confirmed stopped") cur = con.execute( "UPDATE jobs SET status=?, block_reason=CASE WHEN cancel_requested=2" - " THEN 'timeout_pending: live child process group remains'" - " ELSE 'cancellation_pending: live child process group remains' END," + " THEN 'timeout_pending: ' || ? ELSE 'cancellation_pending: ' || ? END," " updated_at=? WHERE request_id=? AND status NOT IN ('succeeded','failed','cancelled')", - (status, now, request_id)) + (status, what, what, now, request_id)) if cur.rowcount: _event(con, request_id, "blocked", {"reason": "stop_pending"}) con.execute("COMMIT") @@ -2120,13 +2125,14 @@ def recover_one(state_dir, request_id: str) -> dict: con.close() children_dead = _drain_owned_children(state_dir, request_id, dict(job)) proof_dead = _drain_proof_group(state_dir, request_id) - remote_settled = (_interrupt_saved_t3_threads( - state_dir, request_id, dict(job)) - if children_dead and proof_dead else False) + remote_settled, remote_detail = ( + _interrupt_saved_t3_threads(state_dir, request_id, dict(job)) + if children_dead and proof_dead else (False, "")) all_dead = bool(children_dead and proof_dead and remote_settled) _finalize_stopped(state_dir, request_id, all_dead, remote_settled, local_settled=bool(children_dead and proof_dead), - controller_fenced=children_dead) + controller_fenced=children_dead, + remote_detail=remote_detail) fin = get_job(state_dir, request_id) return {"request_id": request_id, "action": "timeout" if fin["error_class"] == "timeout" else "cancelled", diff --git a/runner/t3exec.py b/runner/t3exec.py index 1fdebad..35cafef 100644 --- a/runner/t3exec.py +++ b/runner/t3exec.py @@ -20,13 +20,14 @@ thread. Every child's first message names the job and links the planner thread. -Liveness is activity-based, never a fixed silence window: a turn is -healthy while assistant tokens stream (message ``updatedAt`` advances) -or a ``tool.*`` / ``task.*`` call is still open. Silence with no running tool past -``T3_SILENCE_SECS`` (≈ a minute) is flagged and probed immediately -with a fresh snapshot read; explicit provider errors act at once. A -long test run holds a running tool activity, so it is never mistaken -for a stall. +Liveness is activity-based: a turn is healthy while assistant tokens +stream (message ``updatedAt`` advances) or a ``tool.*`` / ``task.*`` call +is still open. Silence with no running tool past the driver's measured +window (:data:`T3_SILENCE_SECS_BY_DRIVER`) is flagged and probed +immediately with a fresh snapshot read; explicit provider errors act at +once. A long test run holds a running tool activity, so it is never +mistaken for a stall. Time the host spent asleep never counts as +silence (#132). Restart: after a T3 server restart with ``continueThreadsAfterServerUpdate`` on, interrupted Claude, Codex and @@ -70,8 +71,32 @@ T3_PLANNER_HARNESS = "t3" # Activity-based liveness: silence with no running tool is flagged past -# this window (about a minute) and probed immediately with a fresh read. -T3_SILENCE_SECS = 60.0 +# the driver's window and probed immediately with a fresh read (#132). +# Interim until T3 exposes its stream clock (toolboxmd/t3code#55): each +# window is the largest healthy gap observed for that driver, times 1.25, +# rounded up to 30 s. Gaps come from completed T3 turns +# (``projection_thread_messages`` updatedAt and activity createdAt, open +# tools excluded, host sleep excluded via ``pmset -g log``), 2026-09-21 +# to 2026-09-27, plus the earlier issue evidence: +# - codex: 232 s (gpt-6-astra context compaction, n=102 turns); max-effort +# reasoning silent 181 s (#73). 232 * 1.25 -> 300 s. +# - opencode: 96 s in T3 (Muse on Go, n=10); healthy sessions up to 161 s +# in OpenCode's own database (#33). 161 * 1.25 -> 210 s. +# - claudeAgent: 54 s before the first token (n=545). 54 * 1.25 -> 90 s. +# Drivers without measurements (grok, unknown) take the largest window. +# Replace these with per-route thresholds from T3's per-turn stream +# statistics once t3code#55 lands; ``silence_window`` is the only reader. +T3_SILENCE_SECS_BY_DRIVER = {"codex": 300.0, "opencode": 210.0, + "claudeAgent": 90.0} +T3_SILENCE_SECS = max(T3_SILENCE_SECS_BY_DRIVER.values()) +# A wall-clock step this much larger than the monotonic step between two +# polls means the host slept (macOS and Linux monotonic clocks stop +# during suspend); that time is not silence. +T3_SLEEP_TOLERANCE_SECS = 2.0 +# After an interrupt, poll this often until the turn settles, at most +# this long (T3 applies interrupts asynchronously: 58 ms on #132 job -2). +T3_SETTLE_POLL_SECS = 0.5 +T3_SETTLE_MAX_SECS = 15.0 # Poll interval while watching a T3 turn. T3_POLL_SECS = 2.0 # Recovery must not watch an already-saved child forever when its turn was @@ -718,18 +743,113 @@ def error_evidence(snapshot: dict, turn_id: str | None = None) -> str | None: return None +def snapshot_driver(snapshot: dict) -> str | None: + """T3 driver of the thread's model selection (``codex``, ``opencode``...).""" + thread = snapshot_thread(snapshot) + selection = thread.get("modelSelection") + instance = selection.get("instanceId") if isinstance(selection, dict) else None + if not isinstance(instance, str) or not instance: + session = thread.get("session") + instance = session.get("providerInstanceId") if isinstance(session, dict) else None + if not isinstance(instance, str) or not instance: + return None + return next((d for d in T3_EFFORT_OPTION if instance.startswith(d)), instance) + + +def silence_window(driver: str | None) -> float: + """Silence (no running tool) after which a turn of ``driver`` is flagged.""" + return T3_SILENCE_SECS_BY_DRIVER.get(driver or "", T3_SILENCE_SECS) + + +def awake_secs(start: float, end: float, sleeps=()) -> float: + """Seconds in ``[start, end]`` the host was awake (``sleeps``: wall intervals).""" + asleep = sum(max(0.0, min(end, b) - max(start, a)) for a, b in sleeps) + return max(0.0, end - start - asleep) + + +TURN_TERMINAL_STATES = ("completed", "error", "interrupted") + + +def turn_settled(snapshot: dict) -> bool: + """True when the thread has no live turn: the session names no active + turn, and the latest turn is completed, errored or interrupted, there + is none, or the session is stopped.""" + thread = snapshot_thread(snapshot) + if not thread or "latestTurn" not in thread: + return False + session = thread.get("session") + if isinstance(session, dict) and session.get("activeTurnId"): + # The session still runs a turn, whatever latestTurn says (live T3 + # clears activeTurnId once a turn ends). + return False + latest = thread.get("latestTurn") + if latest is None: + return True + if isinstance(latest, dict) and latest.get("state") in TURN_TERMINAL_STATES: + return True + return isinstance(session, dict) and session.get("status") == "stopped" + + + +def interrupt_and_settle(client: "T3Client", thread_id: str, *, + command: dict | None = None, + poll_secs: float = T3_SETTLE_POLL_SECS, + max_secs: float = T3_SETTLE_MAX_SECS, + now_fn=None, sleep_fn=None) -> dict: + """Interrupt the thread's turn and poll until it settles (#132). + + T3 applies ``thread.turn.interrupt`` asynchronously, so one read right + after the dispatch can still show the live turn. Returns ``settled``, + the latest turn ``state``, the last ``snapshot`` read (or None), and + ``error`` when the dispatch or a read failed. + """ + now_fn = now_fn or time.monotonic + sleep_fn = sleep_fn or time.sleep + out: dict = {"settled": False, "state": None, "snapshot": None} + try: + out["result"] = client.dispatch(command or turn_interrupt_command(thread_id)) + except Exception as e: # noqa: BLE001 - reported, then the reads decide + out["error"] = f"interrupt dispatch failed: {e}"[:300] + deadline = now_fn() + max_secs + while True: + try: + snapshot = client.thread_snapshot(thread_id) + except T3Error as e: + out["error"] = f"t3 snapshot failed: {e}"[:300] + return out + latest = snapshot_thread(snapshot).get("latestTurn") + out.update(snapshot=snapshot, + state=latest.get("state") if isinstance(latest, dict) else None) + if turn_settled(snapshot): + out["settled"] = True + return out + if now_fn() >= deadline: + return out + sleep_fn(poll_secs) + + def evaluate_turn(snapshot: dict, *, now: float, last_known_activity: float | None = None, - silence_secs: float = T3_SILENCE_SECS) -> dict: + silence_secs: float | None = None, + sleeps=()) -> dict: """Activity-based liveness of one T3 turn. Returns ``state`` ``running`` (tokens streaming or a tool running), ``completed`` (with ``assistant_text``), ``error`` (with ``reason``), ``interrupted`` (server restart cut the turn), or ``stalled`` - (silence past ``silence_secs`` with no running tool, with - ``silence`` age and ``last_part`` evidence). ``last_activity_ts`` + (silence past the window with no running tool, with ``silence`` age + and ``last_part`` evidence). ``silence_secs`` defaults to the + thread's driver window; ``sleeps`` lists wall-clock intervals the + host slept, which never count as silence. ``last_activity_ts`` carries the newest observed activity for the caller's next poll. + + Seam for toolboxmd/t3code#55: once T3 serves ``lastStreamAt`` and + ``openTool``, they replace the anchor and the tool check below, and + ``silence_window`` reads per-route thresholds from T3's per-turn + stream statistics. """ + if silence_secs is None: + silence_secs = silence_window(snapshot_driver(snapshot)) thread = snapshot_thread(snapshot) latest = thread.get("latestTurn") latest_state = latest.get("state") if isinstance(latest, dict) else None @@ -784,9 +904,10 @@ def evaluate_turn(snapshot: dict, *, now: float, # Nothing observed yet: the turn just started, not a stall. return {"state": "running", "reason": "awaiting first activity", "last_activity_ts": None} - silence = max(0.0, now - anchor) + silence = awake_secs(anchor, now, sleeps) if silence >= silence_secs: return {"state": "stalled", "silence": silence, + "window": silence_secs, "slept": max(0.0, now - anchor) - silence, "last_part": _last_part_kind(snapshot), "last_activity_ts": anchor} return {"state": "running", "reason": "within silence window", @@ -835,10 +956,30 @@ def _session_error_since(snapshot: dict, since: float) -> str | None: else "provider turn error" +def first_token_secs(snapshot: dict) -> float | None: + """Seconds from the latest turn's start to its first non-user message + or activity, or None before any.""" + thread = snapshot_thread(snapshot) + latest = thread.get("latestTurn") + if not isinstance(latest, dict): + return None + turn = latest.get("turnId") + start = _parse_ts(latest.get("startedAt")) or _parse_ts(latest.get("requestedAt")) + if start is None: + return None + firsts = [_parse_ts(m.get("createdAt")) for m in (thread.get("messages") or []) + if isinstance(m, dict) and m.get("role") != "user" + and m.get("turnId") == turn] + firsts += [_parse_ts(a.get("createdAt")) for a in (thread.get("activities") or []) + if isinstance(a, dict) and a.get("turnId") == turn] + firsts = [t for t in firsts if t is not None and t >= start] + return round(min(firsts) - start, 3) if firsts else None + + def watch_turn(client: T3Client, thread_id: str, *, - now_fn=None, sleep_fn=None, + now_fn=None, sleep_fn=None, mono_fn=None, poll_secs: float = T3_POLL_SECS, - silence_secs: float = T3_SILENCE_SECS, + silence_secs: float | None = None, timeout_secs: float | None = None, on_snapshot=None, prior_turn_id: str | None = None, @@ -853,22 +994,51 @@ def watch_turn(client: T3Client, thread_id: str, *, long as the turn stays active. Explicit provider errors return at once. """ + # Host sleep shows as a wall-clock step larger than the monotonic step + # (#132: a 99 s clamshell sleep read as 105.7 s of silence). An injected + # wall clock without a monotonic one disables the detection. + if mono_fn is None: + mono_fn = time.monotonic if now_fn is None else now_fn now_fn = now_fn or time.time sleep_fn = sleep_fn or time.sleep started = now_fn() deadline = (started + timeout_secs) if timeout_secs else None last_activity: float | None = None probation: dict | None = None + sleeps: list[tuple[float, float]] = [] + clocks = (started, mono_fn()) + stats = {"max_silence": 0.0} + + def _done(ev: dict, snapshot: dict | None = None) -> dict: + stream = dict(stats, slept=round(sum(b - a for a, b in sleeps), 3)) + stream["max_silence"] = round(stream["max_silence"], 3) + if snapshot is not None: + stream["window"] = (silence_secs if silence_secs is not None + else silence_window(snapshot_driver(snapshot))) + stream["first_token_secs"] = first_token_secs(snapshot) + ev["stream"] = stream + return ev + + def _observe(ev: dict) -> dict: + if isinstance(ev.get("silence"), (int, float)): + stats["max_silence"] = max(stats["max_silence"], float(ev["silence"])) + return ev + while True: now = now_fn() + mono = mono_fn() + jump = (now - clocks[0]) - (mono - clocks[1]) + if jump > T3_SLEEP_TOLERANCE_SECS: + sleeps.append((now - jump, now)) + clocks = (now, mono) if deadline is not None and now >= deadline: - return {"state": "timeout", "reason": "watch timed out", - "last_activity_ts": last_activity} + return _done({"state": "timeout", "reason": "watch timed out", + "last_activity_ts": last_activity}) try: snapshot = client.thread_snapshot(thread_id) except T3Error as e: - return {"state": "error", "reason": f"t3 snapshot failed: {e}", - "signal": None, "last_activity_ts": last_activity} + return _done({"state": "error", "reason": f"t3 snapshot failed: {e}", + "signal": None, "last_activity_ts": last_activity}) if on_snapshot is not None: try: on_snapshot(snapshot) @@ -890,22 +1060,26 @@ def watch_turn(client: T3Client, thread_id: str, *, # once. A turn that never starts goes silent like any other. refused = _session_error_since(snapshot, started - 2.0) if refused is not None: - return {"state": "error", "reason": refused, - "signal": classify_provider_error(refused), - "last_activity_ts": last_activity} - if now - started >= silence_secs: - return {"state": "stalled", "silence": now - started, - "last_part": "turn never started", "probed": True, - "last_activity_ts": last_activity} + return _done({"state": "error", "reason": refused, + "signal": classify_provider_error(refused), + "last_activity_ts": last_activity}) + waited = awake_secs(started, now, sleeps) + window = (silence_secs if silence_secs is not None + else silence_window(snapshot_driver(snapshot))) + if waited >= window: + return _done({"state": "stalled", "silence": waited, + "window": window, + "last_part": "turn never started", "probed": True, + "last_activity_ts": last_activity}) sleep_fn(poll_secs) continue - ev = evaluate_turn(snapshot, now=now, - last_known_activity=last_activity, - silence_secs=silence_secs) + ev = _observe(evaluate_turn(snapshot, now=now, + last_known_activity=last_activity, + silence_secs=silence_secs, sleeps=sleeps)) last_activity = ev.get("last_activity_ts", last_activity) state = ev.get("state") if state in ("completed", "error", "interrupted"): - return ev + return _done(ev, snapshot) if state == "stalled": if probation is None: # Flagged: probe immediately with a fresh read before it counts. @@ -913,27 +1087,28 @@ def watch_turn(client: T3Client, thread_id: str, *, try: snapshot = client.thread_snapshot(thread_id) except T3Error as e: - return {"state": "error", - "reason": f"t3 probe failed: {e}", - "signal": None, "last_activity_ts": last_activity} + return _done({"state": "error", + "reason": f"t3 probe failed: {e}", + "signal": None, "last_activity_ts": last_activity}) if on_snapshot is not None: try: on_snapshot(snapshot) except Exception: pass - second = evaluate_turn(snapshot, now=now_fn(), - last_known_activity=last_activity, - silence_secs=silence_secs) + second = _observe(evaluate_turn(snapshot, now=now_fn(), + last_known_activity=last_activity, + silence_secs=silence_secs, + sleeps=sleeps)) last_activity = second.get("last_activity_ts", last_activity) if second.get("state") == "running": probation = None ev = second else: second["probed"] = True - return second + return _done(second, snapshot) else: ev["probed"] = True - return ev + return _done(ev, snapshot) else: probation = None sleep_fn(poll_secs) @@ -947,8 +1122,7 @@ def post_restart_action(harness_name: str | None, snapshot: dict) -> dict: turn); Grok turns are re-sent by the router with the same message. A still-running turn is simply watched on. """ - ev_state = evaluate_turn(snapshot, now=time.time(), - silence_secs=T3_SILENCE_SECS).get("state") + ev_state = evaluate_turn(snapshot, now=time.time()).get("state") if ev_state not in ("interrupted", "error", "stalled"): return {"action": "continue", "reason": f"turn {ev_state} after restart"} if (harness_name or "") == "grok": diff --git a/tests/test_issue132.py b/tests/test_issue132.py new file mode 100644 index 0000000..f6b4a16 --- /dev/null +++ b/tests/test_issue132.py @@ -0,0 +1,436 @@ +"""Prism jobs never go silent (toolboxmd/model-router#132). + +Deterministic: every T3 call runs against the in-memory fake T3 client. +""" +import json +import os +import subprocess +import sys +import tempfile +import unittest +from datetime import timedelta +from pathlib import Path +from unittest import mock + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from runner import controller, core, store, t3exec, t3snapshot # noqa: E402 +from tests.fakes import (NOW, FakeT3Client, isolate_t3_env, snap, # noqa: E402 + use_fake_t3) + + +def setUpModule(): + isolate_t3_env() + + +APPROVE = {"verdict": "approve", "findings": ""} +CHANGES = {"verdict": "request_changes", "findings": "secret leaks into proof.log"} +PLANNER = "planner-t3" + + +def git(ws, *args): + subprocess.run(["git", *args], cwd=str(ws), check=True, capture_output=True) + + +def events(sd, rid, kind): + con = store.connect(sd) + try: + rows = con.execute("SELECT payload_json FROM events WHERE request_id=? AND kind=?" + " ORDER BY id", (rid, kind)).fetchall() + finally: + con.close() + return [json.loads(r["payload_json"] or "{}") for r in rows] + + +def update_job(sd, rid, **cols): + con = store.connect(sd) + try: + sets = ", ".join(f"{k}=?" for k in cols) + con.execute(f"UPDATE jobs SET {sets} WHERE request_id=?", (*cols.values(), rid)) + con.commit() + finally: + con.close() + + +def set_state(sd, rid, **fields): + job = core.get_job(sd, rid) + st = json.loads(job.get("controller_state") or "{}") + st.update(fields) + update_job(sd, rid, controller_state=json.dumps(st, sort_keys=True), status="running") + + +class SettlesLater(FakeT3Client): + """Applies an interrupt only after ``delay_reads`` more snapshot reads, + like T3 applying ``thread.turn.interrupt`` asynchronously.""" + + def __init__(self, *a, delay_reads=3, **k): + super().__init__(*a, **k) + self.delay_reads = delay_reads + self.pending = {} + + def dispatch(self, command): + self.commands.append(dict(command)) + if command.get("type") == "thread.turn.interrupt": + self.pending[command["threadId"]] = self.delay_reads + return {"sequence": len(self.commands)} + self.commands.pop() + return super().dispatch(command) + + def thread_snapshot(self, thread_id): + left = self.pending.get(thread_id) + if left is not None: + if left <= 0: + self.pending.pop(thread_id) + self.scripts[thread_id] = snap(thread_id, state="interrupted") + else: + self.pending[thread_id] = left - 1 + self.reads[thread_id] = self.reads.get(thread_id, 0) + 1 + return snap(thread_id, state="running") + return super().thread_snapshot(thread_id) + + +class ControllerLoop(unittest.TestCase): + """Continuation follows the job's status, never an action allowlist.""" + + def setUp(self): + t3snapshot.reset() + self.addCleanup(t3snapshot.reset) + tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.addCleanup(tmp.cleanup) + base = Path(tmp.name) + self.sd = str(base / "state") + self.ws = base / "ws" + self.ws.mkdir() + git(self.ws, "init") + git(self.ws, "config", "user.email", "test@example.test") + git(self.ws, "config", "user.name", "Test") + (self.ws / "a.txt").write_text("one\n") + git(self.ws, "add", "-A") + git(self.ws, "commit", "-m", "one") + self.addCleanup(self._release_lock) + + def _release_lock(self): + fd = controller._LOCK.get("fd") + if fd is not None: + os.close(fd) + controller._LOCK["fd"] = None + controller._LEASE["token"] = None + + def job_at_completion(self, rid, replies): + core.submit(self.sd, rid, {"goal": "x", "proof": "true"}, str(self.ws), "p", + planner_t3_thread=PLANNER) + fake = use_fake_t3(self, self.sd, rid, replies=list(replies)) + job = core.get_job(self.sd, rid) + full = {"assistant_text": "did stuff", "usage": None, "native_ids": {}, + "finish": "stop", "actual_model": None} + controller._write_turn_report(self.sd, rid, job, 1, "muse-spark-xhigh-free", + full, "ses") + set_state(self.sd, rid, seq=1, last_action_name="completion", + last_action={"action": "completion", "output": "DONE"}) + update_job(self.sd, rid, owner_token="tok") + return fake + + def test_review_changes_run_correction_rereview_and_approve_in_one_run(self): + fake = self.job_at_completion( + "rc", [CHANGES, "fixed the leak", + {"action": "completion", "output": "DONE AGAIN"}, APPROVE]) + first_head = core._workspace_head(str(self.ws)) + post = fake.post_message + + def worker_commits(thread_id, text, **kw): + # The correction worker fixes and commits: a new head to review. + if "worker seq" in text: + (self.ws / "a.txt").write_text("two\n") + git(self.ws, "commit", "-am", "fix") + return post(thread_id, text, **kw) + + fake.post_message = worker_commits + self.assertEqual(controller.run_controller_process(self.sd, "rc", "tok"), 0) + job = core.get_job(self.sd, "rc") + self.assertEqual(job["status"], "succeeded", job.get("block_reason")) + new_head = core._workspace_head(str(self.ws)) + self.assertNotEqual(first_head, new_head) + verdicts = events(self.sd, "rc", "review_verdict") + self.assertEqual([(v["round"], v["verdict"], v["reviewed_sha"]) for v in verdicts], + [(1, "request_changes", first_head), (2, "approve", new_head)]) + # One controller run did it all: launched once, exited once. + self.assertEqual(len(events(self.sd, "rc", "controller_exited")), 1) + # Every watched turn left its stream statistics in the ledger. + stream = events(self.sd, "rc", "t3_turn_stream") + self.assertTrue(stream) + self.assertTrue(all("max_silence" in e and "first_token_secs" in e + and "window" in e for e in stream), stream) + + def test_step_stop_with_active_job_blocks_with_named_reason(self): + self.job_at_completion("un", []) + with mock.patch.object(controller, "step", + return_value={"action": "blocked", "reason": "odd"}): + controller.run_controller_process(self.sd, "un", "tok") + job = core.get_job(self.sd, "un") + self.assertEqual(job["status"], "blocked") + self.assertEqual(job["block_reason"], "controller_exit_unhandled: blocked (odd)") + + def test_unlisted_action_with_running_job_keeps_stepping(self): + self.job_at_completion("go", []) + calls = [] + + def fake_step(sd, rid, token=None): + calls.append(1) + if len(calls) == 3: + controller._mark_blocked(sd, rid, "planner decides") + return {"action": "some-new-action"} + + with mock.patch.object(controller, "step", side_effect=fake_step): + controller.run_controller_process(self.sd, "go", "tok") + self.assertEqual(len(calls), 3) + self.assertEqual(core.get_job(self.sd, "go")["block_reason"], "planner decides") + + +class InterruptSettles(unittest.TestCase): + """After an interrupt, poll until the turn settles.""" + + def test_polls_until_the_delayed_interrupt_lands(self): + fake = SettlesLater(planner=PLANNER, delay_reads=3) + fake.scripts["w"] = snap("w", state="running") + naps = [] + res = t3exec.interrupt_and_settle(fake, "w", sleep_fn=naps.append) + self.assertTrue(res["settled"]) + self.assertEqual(res["state"], "interrupted") + self.assertEqual(naps, [t3exec.T3_SETTLE_POLL_SECS] * 3) + + def test_stopped_session_without_active_turn_counts_as_settled(self): + stopped = snap("w", state="running", session_status="stopped") + stopped["thread"]["session"]["activeTurnId"] = None + self.assertTrue(t3exec.turn_settled(stopped)) + self.assertFalse(t3exec.turn_settled(snap("w", state="running"))) + + def test_no_latest_turn_with_an_active_session_turn_is_not_settled(self): + live = snap("w", state="running") + live["thread"]["latestTurn"] = None + self.assertFalse(t3exec.turn_settled(live)) + live["thread"]["session"]["activeTurnId"] = None + self.assertTrue(t3exec.turn_settled(live)) + + def test_active_session_turn_outranks_a_terminal_latest_turn(self): + newer = snap("w", state="completed", text="old") + newer["thread"]["session"]["activeTurnId"] = "t2" + self.assertFalse(t3exec.turn_settled(newer)) + + def test_cancel_interrupts_when_session_runs_a_newer_turn(self): + tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.addCleanup(tmp.cleanup) + sd = str(Path(tmp.name) / "state") + ws = Path(tmp.name) / "ws" + ws.mkdir() + core.submit(sd, "c3", {"goal": "t"}, str(ws), "s", planner_t3_thread=PLANNER) + update_job(sd, "c3", controller_state=json.dumps({"t3_threads": {"dispatch": { + "thread_id": "sub.planner-t3.d", "route": "luna/max", "created": True, + "turn_started": True, "turn_state": "completed"}}})) + fake = SettlesLater(planner=PLANNER, delay_reads=1) + newer = snap("sub.planner-t3.d", state="completed", text="old") + newer["thread"]["session"]["activeTurnId"] = "t2" + fake.scripts["sub.planner-t3.d"] = newer + with mock.patch.object(t3exec, "client_for_job", return_value=fake), \ + mock.patch.object(t3exec.time, "sleep"): + job = core.cancel(sd, "c3") + self.assertEqual(job["status"], "cancelled") + self.assertEqual(sum(c.get("type") == "thread.turn.interrupt" + for c in fake.commands), 1) + + def test_cancel_interrupts_a_live_turn_missing_from_latest_turn(self): + tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.addCleanup(tmp.cleanup) + sd = str(Path(tmp.name) / "state") + ws = Path(tmp.name) / "ws" + ws.mkdir() + core.submit(sd, "c2", {"goal": "t"}, str(ws), "s", planner_t3_thread=PLANNER) + update_job(sd, "c2", controller_state=json.dumps({"t3_threads": {"dispatch": { + "thread_id": "sub.planner-t3.d", "route": "luna/max", "created": True, + "turn_started": True, "turn_state": "running"}}})) + fake = SettlesLater(planner=PLANNER, delay_reads=1) + live = snap("sub.planner-t3.d", state="running") + live["thread"]["latestTurn"] = None + fake.scripts["sub.planner-t3.d"] = live + with mock.patch.object(t3exec, "client_for_job", return_value=fake), \ + mock.patch.object(t3exec.time, "sleep"): + job = core.cancel(sd, "c2") + self.assertEqual(job["status"], "cancelled") + self.assertEqual(sum(c.get("type") == "thread.turn.interrupt" + for c in fake.commands), 1) + + def test_gives_up_after_the_cap(self): + fake = SettlesLater(planner=PLANNER, delay_reads=10 ** 6) + clock = [0.0] + + def nap(secs): + clock[0] += secs + + res = t3exec.interrupt_and_settle(fake, "w", sleep_fn=nap, + now_fn=lambda: clock[0]) + self.assertFalse(res["settled"]) + self.assertGreaterEqual(clock[0], t3exec.T3_SETTLE_MAX_SECS) + + def test_worker_stall_path_confirms_idle_after_delayed_interrupt(self): + tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.addCleanup(tmp.cleanup) + sd = str(Path(tmp.name) / "state") + ws = Path(tmp.name) / "ws" + ws.mkdir() + core.submit(sd, "st", {"goal": "t"}, str(ws), "s", planner_t3_thread=PLANNER) + fake = SettlesLater(planner=PLANNER, delay_reads=3) + job = core.get_job(sd, "st") + seen = {} + + def capture(state_dir, request_id, job, route, seq, artifact, full, + thread_id, rc, source=None): + seen.update(full) + return {"action": "captured"} + + stalled = {"state": "stalled", "silence": 400.0, "last_part": "message:assistant", + "probed": True, "thread_id": "sub.planner-t3.w1"} + with mock.patch.object(controller, "_t3_run_turn", return_value=stalled), \ + mock.patch.object(controller, "_finish_worker_turn", side_effect=capture), \ + mock.patch.object(t3exec.time, "sleep"): + fake.scripts["sub.planner-t3.w1"] = snap("sub.planner-t3.w1", state="running") + controller._run_t3_worker_turn(sd, "st", job, str(ws), "muse-spark-xhigh-free", + "p", 1, "r", t3_client=fake) + self.assertEqual(seen.get("signal"), "stalled") + # A single read right after the interrupt still saw the live turn; + # polling waits for T3 to apply it, so the stall may move routes. + self.assertTrue(seen.get("idle_confirmed"), seen) + + +class Liveness(unittest.TestCase): + """Measured per-driver windows; host sleep is never silence.""" + + def running(self, instance, age): + s = snap("w", state="running", text="thinking", age_secs=age, + requested_age_secs=age + 5) + s["thread"]["modelSelection"] = {"instanceId": instance, "model": "m"} + return s + + def test_window_follows_the_thread_driver(self): + now = NOW.timestamp() + for instance, window in (("codex", 300.0), ("opencode", 210.0), + ("claudeAgent", 90.0), ("grok", 300.0)): + quiet = t3exec.evaluate_turn(self.running(instance, window - 5), now=now) + self.assertEqual(quiet["state"], "running", instance) + dead = t3exec.evaluate_turn(self.running(instance, window + 5), now=now) + self.assertEqual(dead["state"], "stalled", instance) + self.assertEqual(dead["window"], window) + + def test_old_sixty_second_window_no_longer_stalls_opencode(self): + # #132 job -2: 105.7 s of wall-clock silence on an OpenCode worker. + ev = t3exec.evaluate_turn(self.running("opencode", 105.7), now=NOW.timestamp()) + self.assertEqual(ev["state"], "running") + + def test_host_sleep_is_not_silence(self): + # The last token 4 s before a 99 s sleep: 103 s of wall clock, 4 s awake. + wall = [NOW.timestamp()] + mono = [1000.0] + fake = FakeT3Client(planner=PLANNER) + s = self.running("claudeAgent", 4) + fake.scripts["w"] = s + reads = [] + + def nap(secs): + reads.append(secs) + if len(reads) == 1: + wall[0] += 99.0 + secs # the host sleeps: only wall time moves + mono[0] += secs + elif len(reads) >= 3: + fake.scripts["w"] = snap("w", state="completed", text="done") + else: + wall[0] += secs + mono[0] += secs + + out = t3exec.watch_turn(fake, "w", now_fn=lambda: wall[0], + mono_fn=lambda: mono[0], sleep_fn=nap) + self.assertEqual(out["state"], "completed", out) + self.assertGreaterEqual(out["stream"]["slept"], 99.0) + self.assertLess(out["stream"]["max_silence"], 20.0) + + def test_awake_silence_past_window_still_stalls(self): + wall = [NOW.timestamp()] + fake = FakeT3Client(planner=PLANNER) + fake.scripts["w"] = self.running("claudeAgent", 0) + + def nap(secs): + wall[0] += 50.0 + + out = t3exec.watch_turn(fake, "w", now_fn=lambda: wall[0], + mono_fn=lambda: wall[0], sleep_fn=nap) + self.assertEqual(out["state"], "stalled") + self.assertGreaterEqual(out["silence"], 90.0) + self.assertEqual(out["stream"]["slept"], 0.0) + + def test_first_token_secs(self): + s = snap("w", state="completed", text="hi", age_secs=0, requested_age_secs=7) + self.assertAlmostEqual(t3exec.first_token_secs(s), 7.0, places=2) + + def test_awake_secs_subtracts_only_overlap(self): + self.assertEqual(t3exec.awake_secs(0, 100, [(50, 149)]), 50) + self.assertEqual(t3exec.awake_secs(0, 100, [(-50, -10)]), 100) + + +class CancelWithT3Unreachable(unittest.TestCase): + def setUp(self): + tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.addCleanup(tmp.cleanup) + self.sd = str(Path(tmp.name) / "state") + ws = Path(tmp.name) / "ws" + ws.mkdir() + core.submit(self.sd, "c1", {"goal": "t"}, str(ws), "s", planner_t3_thread=PLANNER) + + def saved(self, **turn_states): + threads = {slot: {"thread_id": f"sub.planner-t3.{slot}", "route": "luna/max", + "created": True, "turn_started": True, "turn_state": state} + for slot, state in turn_states.items()} + update_job(self.sd, "c1", controller_state=json.dumps({"t3_threads": threads})) + + def cancel_unreachable(self): + with mock.patch.object(t3exec, "client_for_job", + side_effect=t3exec.T3Error("no T3 bearer token")): + return core.cancel(self.sd, "c1") + + def test_all_turns_terminal_finalizes_cancelled_and_frees_workspace(self): + self.saved(dispatch="completed", impl_1="interrupted", review_1="error") + job = self.cancel_unreachable() + self.assertEqual(job["status"], "cancelled") + self.assertNotIn(job["status"], store.ACTIVE_WORKSPACE_STATUSES) + + def test_live_turn_names_t3_unreachability(self): + self.saved(dispatch="completed", impl_1="running") + job = self.cancel_unreachable() + self.assertEqual(job["status"], "cancelling") + self.assertTrue(job["block_reason"].startswith( + "cancellation_pending: T3 unreachable: no T3 bearer token"), job["block_reason"]) + self.assertNotIn("process group", job["block_reason"]) + failed = events(self.sd, "c1", "t3_turn_interrupt_failed") + self.assertEqual([(e["slot"], e["last_known_state"]) for e in failed], + [("impl_1", "running")]) + + +class TurnStateRecord(unittest.TestCase): + def test_saved_thread_keeps_route_when_a_note_has_none(self): + tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.addCleanup(tmp.cleanup) + sd = str(Path(tmp.name) / "state") + ws = Path(tmp.name) / "ws" + ws.mkdir() + core.submit(sd, "n1", {"goal": "t"}, str(ws), "s", planner_t3_thread=PLANNER) + controller._save_t3_thread(sd, "n1", "dispatch", "sub.x", "luna/max", + created=True, turn_started=True) + rec = controller._t3_thread_for(core.get_job(sd, "n1"), "dispatch") + self.assertEqual(rec["turn_state"], "running") + controller._note_t3_turn(sd, "n1", "dispatch", "sub.x", None, + {"state": "completed", + "stream": {"max_silence": 3.0, "first_token_secs": 1.0, + "window": 300.0, "slept": 0.0}}) + rec = controller._t3_thread_for(core.get_job(sd, "n1"), "dispatch") + self.assertEqual((rec["route"], rec["turn_state"]), ("luna/max", "completed")) + self.assertEqual(events(sd, "n1", "t3_turn_stream")[0]["max_silence"], 3.0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_t3_only.py b/tests/test_t3_only.py index 18e216b..ebb9d02 100644 --- a/tests/test_t3_only.py +++ b/tests/test_t3_only.py @@ -214,10 +214,12 @@ def test_exhausted_route_is_skipped_by_later_jobs(self): self.assertEqual(policy.next_capacity_route("glm-5.1-go", set()), (None, None)) -def _continuing(): - import inspect - src = inspect.getsource(controller.run_controller_process) - return src[src.index("continuing = ("):src.index(")", src.index("continuing = ("))] +def _loop_continues(sd, request_id, res): + """The controller loop's own continuation rule (status-based, #132).""" + job = core.get_job(sd, request_id) + return (job["status"] in controller.ACTIVE_STEP_STATUSES + and not job.get("cancel_requested") + and res.get("action") not in controller.STOP_STEP_ACTIONS) class EscalationThroughT3(Base): @@ -245,7 +247,7 @@ def test_failures_end_in_one_planner_question(self): if res.get("action") == "recovery-exhausted-question": # The controller loop continues into the next step, which # asks the planner in its thread; no manual recover. - self.assertIn('"recovery-exhausted-question"', _continuing()) + self.assertTrue(_loop_continues(self.sd, "esc", res)) controller.step(self.sd, "esc") break job = core.get_job(self.sd, "esc") From 78e131e754e7a6baa3da573668c73213006a9604 Mon Sep 17 00:00:00 2001 From: lukemaj Date: Mon, 28 Sep 2026 20:18:36 +0200 Subject: [PATCH 2/3] wip: consume T3 staleness and preserve stale worker evidence --- runner/controller.py | 61 +++++++++++-- runner/t3exec.py | 86 +++++++++++++------ runner/t3salvage.py | 124 +++++++++++++++++++++++++++ tests/fakes.py | 8 ++ tests/test_issue132.py | 189 ++++++++++++++++++++++++++++++++++++++++- 5 files changed, 438 insertions(+), 30 deletions(-) create mode 100644 runner/t3salvage.py diff --git a/runner/controller.py b/runner/controller.py index d3fa6ff..110a907 100644 --- a/runner/controller.py +++ b/runner/controller.py @@ -20,7 +20,7 @@ import time from pathlib import Path -from . import adapters, core, policy, store, t3exec, t3snapshot +from . import adapters, core, policy, store, t3exec, t3salvage, t3snapshot MAX_LOOP_STEPS = 12 # The controller keeps stepping while the job has one of these statuses. @@ -406,13 +406,15 @@ def _save_t3_thread(state_dir, request_id: str, slot: str, def _note_t3_turn(state_dir, request_id: str, slot: str, thread_id: str | None, - route: str | None, outcome: dict) -> None: + route: str | None, outcome: dict, client=None) -> None: """Record a watched turn's end: its last known state on the saved thread, and its stream statistics (max silence, time to first token) in the ledger. Best effort: evidence never blocks the job.""" state = outcome.get("state") if not thread_id: return + if state == "stalled": + _record_t3_stale(state_dir, request_id, slot, thread_id, route, outcome, client) try: _save_t3_thread(state_dir, request_id, slot, thread_id, route, turn_state=state if state in t3exec.TURN_TERMINAL_STATES @@ -432,6 +434,45 @@ def _note_t3_turn(state_dir, request_id: str, slot: str, thread_id: str | None, pass +def _record_t3_stale(state_dir, request_id: str, slot: str, thread_id: str, + route: str | None, outcome: dict, client) -> None: + """Persist partial work and tell the planner before any interrupt.""" + job = core.get_job(state_dir, request_id) + directory = store.job_dir_for(store.ensure_state_dir(state_dir), request_id) / ( + f"stale-{secrets.token_hex(8)}") + evidence = outcome.get("liveness") or {} + rec = {"slot": slot, "thread_id": thread_id, "route": route, + "path": str(directory / "report.json"), + **{k: evidence.get(k) for k in ("silenceMs", "thresholdMs", "thresholdSource", + "reason", "staleSince")}} + if isinstance(rec["reason"], str): + rec["reason"] = adapters.redact_text(rec["reason"]) + try: + rec["salvage"] = t3salvage.capture( + outcome.get("snapshot") or {}, outcome.get("assistant_text") or "", + job["workspace"], directory) + except (OSError, ValueError) as e: + rec["salvage_error"] = adapters.redact_text(str(e)) + rec["salvage"] = {"text": adapters.redact_text(outcome.get("assistant_text") or "")} + directory.mkdir(parents=True, exist_ok=True, mode=0o700) + store.secure_write_text(directory / "report.json", json.dumps(rec, indent=2)) + previous = _load_controller_state(job).get("stale_events") or [] + _set_phase(state_dir, request_id, stale_events=[*previous, rec]) + core._record_t3_interrupt_event( + state_dir, request_id, "t3_turn_stale", + {k: v for k, v in rec.items() if k != "salvage"}) + outcome["salvage_path"] = rec["path"] + message = (f"Job {request_id}: stale turn on {thread_id} ({route}).\n" + f"silenceMs={rec['silenceMs']}; thresholdMs={rec['thresholdMs']}; " + f"thresholdSource={rec['thresholdSource']}; reason={rec['reason']}\n" + "The controller is handling this stale turn; partial evidence follows.\n" + + t3salvage.report([rec])) + posted, detail = _post_planner_text_via_t3( + state_dir, request_id, message, "t3_stale_post", ["t3", "stale-post", thread_id], client) + core._record_t3_interrupt_event(state_dir, request_id, "t3_stale_notification", + {"thread_id": thread_id, "posted": posted, "detail": detail}) + + def _t3_post_and_watch(state_dir, request_id: str, slot: str, client, thread_id: str, text: str, route: str | None = None, **kwargs) -> dict: @@ -441,7 +482,7 @@ def _t3_post_and_watch(state_dir, request_id: str, slot: str, client, turn_state="running") outcome = t3exec.post_and_watch(client, thread_id, text, route=route, **kwargs) - _note_t3_turn(state_dir, request_id, slot, thread_id, route, outcome) + _note_t3_turn(state_dir, request_id, slot, thread_id, route, outcome, client) return outcome @@ -592,7 +633,7 @@ def _t3_run_turn(state_dir, request_id: str, job: dict, *, slot: str, pass outcome["slot"] = slot _note_t3_turn(state_dir, request_id, slot, outcome.get("thread_id"), route, - outcome) + outcome, client) if outcome.get("state") == "unknown": # A failed reconciliation read is recoverable. Do not turn it into a # sticky dispatch or worker failure while the remote state is unknown. @@ -863,7 +904,9 @@ def _t3_worker_full(outcome: dict, thread_id: str) -> tuple[dict, int]: "signal": "stalled", "signal_evidence": {"silence": outcome.get("silence"), "last_part": outcome.get("last_part"), - "probed": outcome.get("probed", True)}, + "probed": outcome.get("probed", True), + **(outcome.get("liveness") or {})}, + "salvage_path": outcome.get("salvage_path"), "error": outcome.get("reason") or "t3 turn stalled", "t3_thread_id": thread_id, "idle_confirmed": bool(outcome.get("idle_confirmed", False))}, 1 @@ -1171,6 +1214,7 @@ def _deliver_terminal_report_via_t3(state_dir, request_id: str, "attempts": attempts + 1} text = _terminal_report_text(request_id, status, pr_url, reason, job.get("handoff_summary")) + text += t3salvage.report(_load_controller_state(job).get("stale_events") or []) posted, detail = _post_planner_text_via_t3( state_dir, request_id, text, T3_TERMINAL_POST_KIND, ["t3", "terminal-post", event_id], t3_client) @@ -1664,6 +1708,12 @@ def _handle_stalled(state_dir, request_id: str, job: dict, seq: int, route: str, _stall_reset_evidence(evidence)) moved["report"] = stall_report return moved + if evidence.get("stale") is True and evidence.get("thresholdSource") != "router-fallback": + # T3 has already confirmed the stale turn. Move to the next route + # without the interim fallback's same-route probation retries. + moved = _move_after_signal(state_dir, request_id, route, "stalled", evidence) + moved["report"] = stall_report + return moved spec = policy.SIGNAL_CLASSES["stalled"] now = time.time() seen = _stalled_retry_state(state_dir, request_id, route) @@ -2344,6 +2394,7 @@ class with its timestamps, the harness signal where one exists, and "proof_started_at": proof_started_at, "proof_ended_at": proof_ended_at, "proof_log": str(proof_log), "diff": str(turn_dir / "diff.patch"), "worker_text": str(turn_dir / "worker.txt"), "worker_summary": worker_text[:1500], + "salvage_path": full.get("salvage_path"), "blockers": adapters.redact_nested(blockers_raw), "tokens": full.get("usage"), "native_ids": full.get("native_ids"), "longest_silence_secs": full.get("longest_silence_secs"), diff --git a/runner/t3exec.py b/runner/t3exec.py index 35cafef..fa72e26 100644 --- a/runner/t3exec.py +++ b/runner/t3exec.py @@ -20,14 +20,11 @@ thread. Every child's first message names the job and links the planner thread. -Liveness is activity-based: a turn is healthy while assistant tokens -stream (message ``updatedAt`` advances) or a ``tool.*`` / ``task.*`` call -is still open. Silence with no running tool past the driver's measured -window (:data:`T3_SILENCE_SECS_BY_DRIVER`) is flagged and probed -immediately with a fresh snapshot read; explicit provider errors act at -once. A long test run holds a running tool activity, so it is never -mistaken for a stall. Time the host spent asleep never counts as -silence (#132). +T3 owns liveness: its Prism liveness route is polled every second and +``stale`` is consumed directly, without a Router timeout or confirmation. +Only unavailable liveness falls back to activity-based driver windows and +a confirming snapshot. Time the host spent asleep never counts as silence +(#132); T3 handles it for its decision, Router for the fallback. Restart: after a T3 server restart with ``continueThreadsAfterServerUpdate`` on, interrupted Claude, Codex and @@ -72,7 +69,7 @@ # Activity-based liveness: silence with no running tool is flagged past # the driver's window and probed immediately with a fresh read (#132). -# Interim until T3 exposes its stream clock (toolboxmd/t3code#55): each +# Fallback for a missing stale field or unavailable liveness route: each # window is the largest healthy gap observed for that driver, times 1.25, # rounded up to 30 s. Gaps come from completed T3 turns # (``projection_thread_messages`` updatedAt and activity createdAt, open @@ -84,8 +81,7 @@ # in OpenCode's own database (#33). 161 * 1.25 -> 210 s. # - claudeAgent: 54 s before the first token (n=545). 54 * 1.25 -> 90 s. # Drivers without measurements (grok, unknown) take the largest window. -# Replace these with per-route thresholds from T3's per-turn stream -# statistics once t3code#55 lands; ``silence_window`` is the only reader. +# T3's stale decision (toolboxmd/chromeria#62) supersedes these windows. T3_SILENCE_SECS_BY_DRIVER = {"codex": 300.0, "opencode": 210.0, "claudeAgent": 90.0} T3_SILENCE_SECS = max(T3_SILENCE_SECS_BY_DRIVER.values()) @@ -98,7 +94,7 @@ T3_SETTLE_POLL_SECS = 0.5 T3_SETTLE_MAX_SECS = 15.0 # Poll interval while watching a T3 turn. -T3_POLL_SECS = 2.0 +T3_POLL_SECS = 1.0 # Recovery must not watch an already-saved child forever when its turn was # never started or the server lost the turn state. T3_RECOVERY_WATCH_SECS = 120.0 @@ -512,6 +508,11 @@ def prism_snapshot(self, project_id: str | None = None) -> dict: path += "?projectId=" + urllib.parse.quote(project_id, safe="") return self._request("GET", path) + def prism_liveness(self, thread_id: str) -> dict: + """T3 owns the stale decision, including thresholds and host sleep.""" + tid = urllib.parse.quote(validate_thread_id(thread_id), safe="") + return self._request("GET", f"/api/prism/liveness?threadId={tid}") + def create_child(self, child_id: str, parent_thread_id: str, project_id: str, title: str, route: str, role: str = "implementation") -> dict: @@ -831,7 +832,7 @@ def interrupt_and_settle(client: "T3Client", thread_id: str, *, def evaluate_turn(snapshot: dict, *, now: float, last_known_activity: float | None = None, silence_secs: float | None = None, - sleeps=()) -> dict: + sleeps=(), liveness: dict | None = None) -> dict: """Activity-based liveness of one T3 turn. Returns ``state`` ``running`` (tokens streaming or a tool running), @@ -843,10 +844,8 @@ def evaluate_turn(snapshot: dict, *, now: float, host slept, which never count as silence. ``last_activity_ts`` carries the newest observed activity for the caller's next poll. - Seam for toolboxmd/t3code#55: once T3 serves ``lastStreamAt`` and - ``openTool``, they replace the anchor and the tool check below, and - ``silence_window`` reads per-route thresholds from T3's per-turn - stream statistics. + A boolean ``liveness.stale`` from T3 overrides local silence checks. + The driver window is only a fallback for unavailable liveness. """ if silence_secs is None: silence_secs = silence_window(snapshot_driver(snapshot)) @@ -881,6 +880,13 @@ def evaluate_turn(snapshot: dict, *, now: float, return {"state": "interrupted", "last_activity_ts": last_activity_ts(snapshot) or last_known_activity} + if isinstance(liveness, dict) and isinstance(liveness.get("stale"), bool): + silence_ms = liveness.get("silenceMs") + return {"state": "stalled" if liveness["stale"] else "running", + "reason": liveness.get("reason"), "liveness": liveness, + "silence": silence_ms / 1000 if isinstance(silence_ms, (int, float)) else None, + "last_activity_ts": last_activity_ts(snapshot) or last_known_activity} + # Silence runs from the newest of: any message or activity, the turn's # own request/start (so a resumed thread's older history never counts # against a fresh turn, and a turn stuck before its first token still @@ -986,9 +992,9 @@ def watch_turn(client: T3Client, thread_id: str, *, await_new_turn: bool = False) -> dict: """Poll a T3 turn to its end. Returns the terminal evaluation. - A stalled read is probed immediately with a fresh snapshot before it - counts: a turn that produced activity between the two reads stays - running. ``timeout_secs`` bounds only the watch (tests and operator + T3's boolean stale decision is final. Only fallback silence is probed + with a fresh snapshot: a turn that produced activity between those + reads stays running. ``timeout_secs`` bounds only the watch (tests and operator tools); production watches carry no elapsed deadline (no per-turn deadline since #88): pass None to wait as long as the turn stays active. Explicit provider errors return at @@ -1015,7 +1021,18 @@ def _done(ev: dict, snapshot: dict | None = None) -> dict: if snapshot is not None: stream["window"] = (silence_secs if silence_secs is not None else silence_window(snapshot_driver(snapshot))) + if isinstance(ev.get("liveness"), dict): + threshold = ev["liveness"].get("thresholdMs") + stream["window"] = threshold / 1000 if isinstance(threshold, (int, float)) else None stream["first_token_secs"] = first_token_secs(snapshot) + if ev.get("state") == "stalled": + ev["snapshot"] = snapshot + ev["assistant_text"] = latest_assistant_text(snapshot, latest_turn_id(snapshot)) + if ev.get("state") == "stalled" and "liveness" not in ev: + ev["liveness"] = { + "stale": True, "silenceMs": ev.get("silence", 0) * 1000, + "thresholdMs": ev.get("window", T3_SILENCE_SECS) * 1000, + "thresholdSource": "router-fallback", "reason": fallback_reason} ev["stream"] = stream return ev @@ -1024,6 +1041,16 @@ def _observe(ev: dict) -> dict: stats["max_silence"] = max(stats["max_silence"], float(ev["silence"])) return ev + def _liveness() -> tuple[dict | None, str]: + try: + live = client.prism_liveness(thread_id) + except T3Error as e: + return None, f"liveness unavailable: {e}" + if isinstance(live, dict) and isinstance(live.get("stale"), bool): + return live, "" + return None, "liveness response lacks a boolean stale field" + + fallback_reason = "liveness unavailable" while True: now = now_fn() mono = mono_fn() @@ -1044,6 +1071,7 @@ def _observe(ev: dict) -> dict: on_snapshot(snapshot) except Exception: pass + liveness, fallback_reason = _liveness() if await_new_turn and _awaiting_new_turn(snapshot, prior_turn_id): latest = snapshot_thread(snapshot).get("latestTurn") if isinstance(latest, dict) and latest.get("state") == "running": @@ -1066,21 +1094,30 @@ def _observe(ev: dict) -> dict: waited = awake_secs(started, now, sleeps) window = (silence_secs if silence_secs is not None else silence_window(snapshot_driver(snapshot))) - if waited >= window: + if liveness is not None: + if liveness["stale"]: + return _done({"state": "stalled", "liveness": liveness, + "reason": liveness.get("reason"), + "last_activity_ts": last_activity}, snapshot) + elif waited >= window: return _done({"state": "stalled", "silence": waited, "window": window, "last_part": "turn never started", "probed": True, - "last_activity_ts": last_activity}) + "last_activity_ts": last_activity}, snapshot) sleep_fn(poll_secs) continue ev = _observe(evaluate_turn(snapshot, now=now, last_known_activity=last_activity, - silence_secs=silence_secs, sleeps=sleeps)) + silence_secs=silence_secs, sleeps=sleeps, + liveness=liveness)) last_activity = ev.get("last_activity_ts", last_activity) state = ev.get("state") if state in ("completed", "error", "interrupted"): return _done(ev, snapshot) if state == "stalled": + if liveness is not None: + # T3 already confirmed staleness. No Router timer or probe. + return _done(ev, snapshot) if probation is None: # Flagged: probe immediately with a fresh read before it counts. probation = ev @@ -1095,10 +1132,11 @@ def _observe(ev: dict) -> dict: on_snapshot(snapshot) except Exception: pass + liveness, fallback_reason = _liveness() second = _observe(evaluate_turn(snapshot, now=now_fn(), last_known_activity=last_activity, silence_secs=silence_secs, - sleeps=sleeps)) + sleeps=sleeps, liveness=liveness)) last_activity = second.get("last_activity_ts", last_activity) if second.get("state") == "running": probation = None diff --git a/runner/t3salvage.py b/runner/t3salvage.py new file mode 100644 index 0000000..f98ab67 --- /dev/null +++ b/runner/t3salvage.py @@ -0,0 +1,124 @@ +"""Preserve evidence named by a stale T3 worker before interrupting it.""" +from __future__ import annotations + +import json +import os +from pathlib import Path +import re +import stat +import tempfile + +from . import adapters, store, t3exec + + +TEXT_LIMIT = 64 * 1024 +ARTIFACT_LIMIT = 16 +# Explicit file references, including Markdown links, backticks and $TMPDIR. +_PATH = re.compile(r"(?:https?://|\$\{TMPDIR\}|\$TMPDIR|/|\./|[\w.-]+/)[^\s`\"'<>|;()]+") +_TEXT_SUFFIXES = (".txt", ".log", ".md", ".json", ".patch", ".diff") + + +def _references(text: str) -> list[str]: + # Preserve spaces and bare filenames when explicitly quoted or linked. + quoted = re.findall(r"`([^`\n]+)`|\[[^\]]*\]\(?\)", text) + refs = [a or b for a, b in quoted + if Path(a or b).suffix.lower() in _TEXT_SUFFIXES] + refs += [m.group().rstrip(".,:]}\\") for m in _PATH.finditer(text)] + return list(dict.fromkeys(refs)) + + +def capture(snapshot: dict, text: str, workspace: str, directory: Path) -> dict: + """Retain redacted text and bounded named text artifacts in the job store. + + Never execute worker text or fetch URLs. Only regular text files inside + the workspace or host temporary directories are read. Keep a reference + and explicit reason for every artifact that cannot be copied. + """ + thread = t3exec.snapshot_thread(snapshot) + turn = t3exec.latest_turn_id(snapshot) + messages = [m.get("text", "") for m in thread.get("messages", []) + if isinstance(m, dict) and m.get("role") == "assistant" + and (not turn or m.get("turnId") == turn) + and isinstance(m.get("text"), str)] + text = "\n\n".join(messages) or text + directory.mkdir(parents=True, exist_ok=True, mode=0o700) + text = adapters.redact_text(text) + text_path = directory / "worker.txt" + store.secure_write_text(text_path, text) + # Tool output can name an artifact before the assistant mentions it. + activities = [a for a in thread.get("activities", []) + if isinstance(a, dict) and (not turn or a.get("turnId") == turn)] + sources = text + "\n" + json.dumps(activities, ensure_ascii=False) + refs = _references(sources) + roots = [Path(workspace).resolve(), Path(tempfile.gettempdir()).resolve(), Path("/tmp").resolve()] + artifacts = [] + copied = 0 + for ref in refs: + # URL paths are references, not local files to fetch. + if "://" in ref or ref.startswith("//"): + artifacts.append({"source": adapters.redact_text(ref), "note": "URL reference only"}) + continue + expanded = ref.replace("${TMPDIR}", tempfile.gettempdir()).replace("$TMPDIR", tempfile.gettempdir()) + path = Path(expanded) + if not path.is_absolute(): + path = roots[0] / path + item = {"source": adapters.redact_text(ref)} + try: + path = path.resolve() + if not any(path.is_relative_to(root) for root in roots): + item["note"] = "outside workspace and temporary directories; reference only" + elif path.suffix.lower() not in _TEXT_SUFFIXES: + item["note"] = "not a text artifact; reference only" + elif copied >= ARTIFACT_LIMIT: + item["note"] = "artifact copy limit reached; reference only" + else: + fd = os.open(path, os.O_RDONLY | os.O_NONBLOCK) + with os.fdopen(fd, "rb") as source: + if not stat.S_ISREG(os.fstat(source.fileno()).st_mode): + raise ValueError("not a regular file") + data = source.read(TEXT_LIMIT + 1) + if b"\x00" in data: + raise ValueError("not a text file") + contents = data[:TEXT_LIMIT].decode("utf-8", errors="replace") + if path.suffix.lower() == ".json": + try: + contents = json.dumps(adapters.redact_nested(json.loads(contents)), indent=2) + except ValueError: + pass + contents = adapters.redact_text(contents) + saved = directory / f"artifact-{copied + 1}.txt" + store.secure_write_text(saved, contents) + item.update(saved=str(saved), excerpt=contents[:4000], + truncated=len(data) > TEXT_LIMIT or len(contents) > 4000) + copied += 1 + except (OSError, ValueError) as e: + item["note"] = adapters.redact_text(f"unavailable: {e}") + artifacts.append(item) + return {"worker_text": str(text_path), "text": text[:8000], + "text_truncated": len(text) > 8000, "artifacts": artifacts} + + +def report(records: list[dict]) -> str: + """Evidence carried across retries into the eventual terminal report.""" + if not records: + return "" + lines = ["", "SALVAGED STALE WORKER EVIDENCE (partial, not a completion verdict):"] + for rec in records: + lines += [f"Thread: {rec.get('thread_id')} | Route: {rec.get('route')}", + f"Evidence: {rec.get('path')}"] + salvage = rec.get("salvage") or {} + if salvage.get("text"): + lines.append(salvage["text"]) + if salvage.get("text_truncated"): + lines.append(f"[Text excerpt; full text: {salvage['worker_text']}]") + for item in salvage.get("artifacts", []): + lines.append(f"Artifact: {item['source']}") + if item.get("saved"): + lines += [f"Saved: {item['saved']}", item["excerpt"]] + if item.get("truncated"): + lines.append("[Artifact excerpt; see saved copy]") + else: + lines.append(item.get("note", "reference only")) + if rec.get("salvage_error"): + lines.append(f"Salvage incomplete: {rec['salvage_error']}") + return "\n".join(lines) diff --git a/tests/fakes.py b/tests/fakes.py index dc6c9c4..f39fffa 100644 --- a/tests/fakes.py +++ b/tests/fakes.py @@ -121,6 +121,8 @@ def __init__(self, planner=PLANNER, project=PROJECT): # server without the endpoint, answered with HTTP 404). self.prism = default_prism() self.prism_reads = [] + self.liveness = {} + self.liveness_reads = [] def dispatch(self, command): self.commands.append(copy.deepcopy(command)) @@ -184,6 +186,12 @@ def prism_snapshot(self, project_id=None): raise t3exec.T3Error("T3 GET /api/prism/snapshot failed: HTTP 404") return copy.deepcopy(self.prism) + def prism_liveness(self, thread_id): + self.liveness_reads.append(thread_id) + if isinstance(self.liveness, Exception): + raise self.liveness + return copy.deepcopy(self.liveness) + def complete(self, thread_id, text): self.scripts[thread_id] = snap(thread_id, state="completed", text=text) diff --git a/tests/test_issue132.py b/tests/test_issue132.py index f6b4a16..541ed31 100644 --- a/tests/test_issue132.py +++ b/tests/test_issue132.py @@ -14,7 +14,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1])) -from runner import controller, core, store, t3exec, t3snapshot # noqa: E402 +from runner import controller, core, store, t3exec, t3salvage, t3snapshot # noqa: E402 from tests.fakes import (NOW, FakeT3Client, isolate_t3_env, snap, # noqa: E402 use_fake_t3) @@ -373,6 +373,193 @@ def test_awake_secs_subtracts_only_overlap(self): self.assertEqual(t3exec.awake_secs(0, 100, [(-50, -10)]), 100) +class StreamLiveness(unittest.TestCase): + running = Liveness.running + def watch(self, liveness, age=1): + fake = FakeT3Client(planner=PLANNER) + fake.liveness = liveness + fake.scripts["w"] = self.running("codex", age) + clock = [NOW.timestamp()] + naps = [] + + def nap(seconds): + naps.append(seconds) + clock[0] += seconds + if len(naps) == 3: + fake.scripts["w"] = snap("w", state="completed", text="done") + + result = t3exec.watch_turn(fake, "w", now_fn=lambda: clock[0], sleep_fn=nap) + return result, fake, naps + + def test_stale_true_returns_in_one_poll_without_router_confirmation(self): + evidence = {"stale": True, "staleSince": NOW.isoformat(), + "silenceMs": 20000, "thresholdMs": 18000, + "thresholdSource": "measured", "reason": "stream silent"} + result, fake, naps = self.watch(evidence) + self.assertEqual(result["state"], "stalled") + self.assertEqual(result["liveness"], evidence) + self.assertEqual(fake.liveness_reads, ["w"]) + self.assertEqual(fake.reads["w"], 1) + self.assertEqual(naps, []) + + def test_stale_false_never_uses_router_silence_window(self): + result, fake, naps = self.watch({"stale": False, "silenceMs": 3600000}, age=3600) + self.assertEqual(result["state"], "completed") + self.assertEqual(naps, [1.0, 1.0, 1.0]) + self.assertGreaterEqual(len(fake.liveness_reads), 3) + + def test_missing_field_and_unreachable_route_use_driver_fallback(self): + for liveness in ({"silenceMs": 3600000}, + t3exec.T3Error("offline", unreachable=True)): + with self.subTest(liveness=liveness): + result, fake, naps = self.watch(liveness, age=301) + self.assertEqual(result["state"], "stalled") + self.assertEqual(result["window"], 300.0) + self.assertEqual(result["liveness"]["thresholdMs"], 300000) + self.assertEqual(result["liveness"]["thresholdSource"], "router-fallback") + self.assertTrue(result["probed"]) + self.assertEqual(naps, []) + + def test_client_gets_liveness_for_the_worker_thread(self): + client = t3exec.T3Client("http://127.0.0.1:9", "fake") + with mock.patch.object(client, "_request", return_value={"stale": False}) as request: + self.assertEqual(client.prism_liveness("sub.p.w"), {"stale": False}) + request.assert_called_once_with("GET", "/api/prism/liveness?threadId=sub.p.w") + + def test_stale_false_while_waiting_for_new_turn_does_not_use_local_deadline(self): + fake = FakeT3Client(planner=PLANNER) + fake.liveness = {"stale": False} + fake.scripts["w"] = snap("w", state="completed", text="old", turn="old") + clock = [NOW.timestamp()] + naps = [] + + def nap(seconds): + clock[0] += 3600 + naps.append(seconds) + if len(naps) == 2: + fake.scripts["w"] = snap("w", state="completed", text="new", turn="new") + + result = t3exec.watch_turn(fake, "w", now_fn=lambda: clock[0], sleep_fn=nap, + prior_turn_id="old", await_new_turn=True) + self.assertEqual(result["assistant_text"], "new") + self.assertEqual(naps, [1.0, 1.0]) + + def test_route_healthy_at_fallback_confirmation_cancels_stall(self): + fake = FakeT3Client(planner=PLANNER) + fake.scripts["w"] = self.running("codex", 3600) + with mock.patch.object(fake, "prism_liveness", side_effect=[{}, {"stale": False}, {"stale": False}]): + result = t3exec.watch_turn(fake, "w", now_fn=lambda: NOW.timestamp(), + sleep_fn=lambda _: fake.complete("w", "done")) + self.assertEqual(result["state"], "completed") + + +class StaleEvidence(unittest.TestCase): + def setUp(self): + tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.addCleanup(tmp.cleanup) + self.sd = str(Path(tmp.name) / "state") + self.ws = Path(tmp.name) / "ws" + self.ws.mkdir() + core.submit(self.sd, "salvage", {"goal": "t"}, str(self.ws), "s", + planner_t3_thread=PLANNER) + self.thread = "sub.planner-t3.worker" + self.route = "muse-spark-xhigh-free" + set_state(self.sd, "salvage", t3_threads={"impl_1": { + "thread_id": self.thread, "route": self.route, + "created": True, "turn_started": True, "turn_state": "running"}}) + self.artifact = self.ws / "review.log" + self.artifact.write_text("request_changes: seven major findings\napi_key=secret-value\n") + self.fake = FakeT3Client(planner=PLANNER) + self.fake.liveness = {"stale": True, "staleSince": NOW.isoformat(), + "silenceMs": 23000, "thresholdMs": 18000, + "thresholdSource": "measured", "reason": "no stream events"} + self.fake.scripts[self.thread] = snap( + self.thread, state="running", text=f"Review found problems: `{self.artifact}`") + + def test_salvage_ledger_and_planner_post_precede_interrupt_and_survive_route_move(self): + dispatch = self.fake.dispatch + interrupts = [] + + def check_before_interrupt(command): + if command["type"] == "thread.turn.interrupt": + records = controller._load_controller_state(core.get_job(self.sd, "salvage"))["stale_events"] + saved = Path(records[0]["salvage"]["artifacts"][0]["saved"]) + self.assertIn("seven major findings", saved.read_text()) + self.assertNotIn("secret-value", saved.read_text()) + self.assertEqual(self.fake.posts[0][0], PLANNER) + self.assertIn("seven major findings", self.fake.posts[0][1]) + self.artifact.unlink() # The worker's temp artifact can now disappear. + interrupts.append(command) + return dispatch(command) + + self.fake.dispatch = check_before_interrupt + with mock.patch.object(controller, "_move_after_signal", return_value={"action": "route_switched"}) as move: + result = controller._run_t3_worker_turn( + self.sd, "salvage", core.get_job(self.sd, "salvage"), str(self.ws), + self.route, "work", 1, "test", t3_client=self.fake) + self.assertEqual(result["action"], "route_switched") + self.assertEqual(len(interrupts), 1) + self.assertEqual(self.fake.liveness_reads, [self.thread]) + self.assertEqual(move.call_args.args[3], "stalled") + self.assertTrue(Path(result["report"]["salvage_path"]).exists()) + evidence = events(self.sd, "salvage", "t3_turn_stale") + self.assertEqual(len(evidence), 1) + for field in ("silenceMs", "thresholdMs", "thresholdSource", "reason"): + self.assertEqual(evidence[0][field], self.fake.liveness[field]) + self.assertTrue(events(self.sd, "salvage", "t3_stale_notification")[0]["posted"]) + # Subsequent work must not replace salvaged findings in the eventual report. + controller._set_phase(self.sd, "salvage", implementation_output="later worker completed") + update_job(self.sd, "salvage", status="blocked", block_reason="need judgment") + report = controller.deliver_terminal_report(self.sd, "salvage", t3_client=self.fake) + self.assertEqual(report["action"], "reported") + terminal = self.fake.posts[-1][1] + self.assertIn("seven major findings", terminal) + self.assertIn(str(self.artifact), terminal) + self.assertIn("Review found problems", terminal) + self.assertNotIn("secret-value", terminal) + + def test_failed_notification_is_recorded_and_evidence_still_reaches_terminal_report(self): + with mock.patch.object(self.fake, "post_message", side_effect=t3exec.T3Error("offline")): + outcome = t3exec.watch_turn(self.fake, self.thread) + controller._note_t3_turn(self.sd, "salvage", "impl_1", self.thread, + self.route, outcome, self.fake) + self.assertFalse(events(self.sd, "salvage", "t3_stale_notification")[0]["posted"]) + update_job(self.sd, "salvage", status="failed", block_reason="offline") + controller.deliver_terminal_report(self.sd, "salvage", t3_client=self.fake) + self.assertIn("seven major findings", self.fake.posts[-1][1]) + + def test_every_stale_event_notifies_and_success_report_keeps_all_findings(self): + for number in (1, 2): + self.fake.scripts[self.thread] = snap(self.thread, text=f"finding {number}") + outcome = t3exec.watch_turn(self.fake, self.thread) + controller._note_t3_turn(self.sd, "salvage", f"impl_{number}", self.thread, + self.route, outcome, self.fake) + self.assertEqual([tid for tid, _ in self.fake.posts], [PLANNER, PLANNER]) + self.assertEqual(len(events(self.sd, "salvage", "t3_turn_stale")), 2) + update_job(self.sd, "salvage", status="succeeded") + controller.deliver_terminal_report(self.sd, "salvage", t3_client=self.fake) + self.assertIn("finding 1", self.fake.posts[-1][1]) + self.assertIn("finding 2", self.fake.posts[-1][1]) + + def test_artifact_in_tool_output_is_preserved_and_unsafe_reference_is_not_read(self): + snapshot = snap(self.thread, text="Review is written.", activities=[{ + "turnId": "t1", "payload": {"output": f"Saved `{self.artifact}`"}}]) + result = t3salvage.capture(snapshot, "", str(self.ws), self.ws / "saved") + self.assertIn("seven major findings", result["artifacts"][0]["excerpt"]) + self.assertNotIn("secret-value", json.dumps(result)) + snapshot = snap(self.thread, text="Reference `/etc/hosts.txt`") + result = t3salvage.capture(snapshot, "", str(self.ws), self.ws / "saved2") + self.assertIn("reference only", result["artifacts"][0]["note"]) + + def test_bare_artifact_and_spaced_markdown_path_are_copied(self): + (self.ws / "review notes.md").write_text("Important review findings.") + snapshot = snap(self.thread, text="See `review.log` and [notes](review notes.md).") + result = t3salvage.capture(snapshot, "", str(self.ws), self.ws / "saved") + self.assertEqual(len(result["artifacts"]), 2) + self.assertIn("seven major findings", result["artifacts"][0]["excerpt"]) + self.assertIn("Important review findings", result["artifacts"][1]["excerpt"]) + + class CancelWithT3Unreachable(unittest.TestCase): def setUp(self): tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) From 0ab66fb7bf52f828cc65788740c2c9fd2eff4192 Mon Sep 17 00:00:00 2001 From: lukemaj Date: Mon, 28 Sep 2026 20:21:58 +0200 Subject: [PATCH 3/3] Finish #132 stream liveness integration; prepare 0.41.0 --- .claude-plugin/plugin.json | 2 +- .codex-plugin/plugin.json | 2 +- .grok-plugin/plugin.json | 2 +- CHANGELOG.md | 8 +++++++ GLOSSARY.md | 2 +- RUNNER.md | 44 +++++++++++++++++++++++++------------- VERSION | 2 +- runner/controller.py | 3 ++- runner/t3exec.py | 12 +++++------ tests/test_issue106.py | 6 ++++++ tests/test_issue132.py | 15 +++++++++++++ 11 files changed, 71 insertions(+), 27 deletions(-) diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json index 127fbc8..17c46bc 100644 --- a/.claude-plugin/plugin.json +++ b/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "model-router", - "version": "0.40.0", + "version": "0.41.0", "description": "Route authorized software work through one durable, capacity-aware subscription policy.", "author": { "name": "toolboxmd", diff --git a/.codex-plugin/plugin.json b/.codex-plugin/plugin.json index 1e28f73..d16c04d 100644 --- a/.codex-plugin/plugin.json +++ b/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "model-router", - "version": "0.40.0", + "version": "0.41.0", "description": "Route authorized software work through one durable, capacity-aware subscription policy.", "author": { "name": "toolboxmd", diff --git a/.grok-plugin/plugin.json b/.grok-plugin/plugin.json index 127fbc8..17c46bc 100644 --- a/.grok-plugin/plugin.json +++ b/.grok-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "model-router", - "version": "0.40.0", + "version": "0.41.0", "description": "Route authorized software work through one durable, capacity-aware subscription policy.", "author": { "name": "toolboxmd", diff --git a/CHANGELOG.md b/CHANGELOG.md index 6fbfd0a..12a9686 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog +## [0.41.0] - 2026-09-28 + +### Added + +- Consume T3 stream staleness, preserve worker findings, and keep Prism jobs advancing (#132) +- Poll T3 liveness every second, honor its stale decision, retain sleep-aware driver windows only as fallback, and notify the planner with salvaged text and artifacts before interruption. +- Continue after review changes and recovered envelopes, wait for interrupts to settle, and finalize cancelled jobs with known finished turns when T3 is unreachable. + ## [0.40.0] - 2026-09-28 ### Changed diff --git a/GLOSSARY.md b/GLOSSARY.md index eeadeb8..0af0d4a 100644 --- a/GLOSSARY.md +++ b/GLOSSARY.md @@ -14,7 +14,7 @@ Canonical terms for Model Router. One term per concept; avoid the synonyms. | Route | One dispatchable choice: a T3 provider instance, model, and effort, with its pool and family. A Prism preference with no policy route runs as `t3::@`. | model (alone) | | Pool | A subscription allowance a route draws from: Zen free, Go, xAI, Codex, Claude. Its meter is one T3 provider's usage windows (Go is one OpenCode account meter); Zen free has none. An exhaustion error rests the whole pool. | provider, account | | Signal class | How provider evidence is classified: exhausted, overloaded, stalled, context, hard. | error type | -| Stall | A T3 turn with no streaming tokens and no running tool call past its driver's measured window (`T3_SILENCE_SECS_BY_DRIVER`) of awake time, confirmed by a fresh snapshot read. | hang | +| Stall | A turn T3 marks `stale` through its Prism liveness route. If that decision is unavailable, Router falls back to the driver's measured silence window of awake time and a confirming snapshot. | hang | | Assumed reset | The reset of a limit error that names none: one 5-hour window, flagged as assumed rather than provider-reported. | guessed reset | | Context window | A route's input context in tokens as policy data; a context-overflow turn moves to a larger one in its lane. | context size | | Retry | The automatic re-attempt of a failed worker turn: first in the same worker thread, then fresh on the Retry route. Its stage and Prism role keep the internal key `correction` so the snapshot contract stays stable. Prism can switch it off (`enabled: false`), which drops both rungs. | correction (in user-facing text) | diff --git a/RUNNER.md b/RUNNER.md index cae768b..df610fa 100644 --- a/RUNNER.md +++ b/RUNNER.md @@ -199,21 +199,35 @@ Grok, `effort` for Claude, `variant` for OpenCode). The thread ids are saved before the watch, so a restarted controller adopts a live turn instead of starting a second writer. -Liveness is activity-based: a turn is healthy while assistant tokens stream -(the message's `updatedAt` advances) or a tool or task call is open -(`tool.started` until `tool.completed`/`tool.denied`, `task.started` until -`task.completed`), so a long test run is never a stall. A tool held behind -an unresolved approval or user-input request is not running. Silence with -no running tool past the thread driver's window is probed at once with a -fresh snapshot read; explicit provider errors act at once. The windows come -from the largest healthy gaps measured per driver, times 1.25: Codex 300 s, -OpenCode 210 s, Claude 90 s, anything else 300 s -(`T3_SILENCE_SECS_BY_DRIVER`). Time the host spends asleep never counts as -silence. Every watched turn records its longest silence and time to first -token as a `t3_turn_stream` event. A message posted onto an existing thread -is watched only through the turn it starts. A stalled worker turn is -interrupted, then polled until T3 reports the turn settled, before any -route move. After a T3 server restart with `continueThreadsAfterServerUpdate` +Router polls `GET /api/prism/liveness?threadId=` every second. T3 owns the +stale decision, including stream clocks, running tools, thresholds and host +sleep. `stale: true` enters the stalled path immediately; `stale: false` +keeps the turn alive regardless of elapsed silence. Router adds no timeout +or confirming probe to that decision. + +Only when the route is unavailable or lacks a boolean `stale` field does +Router use its interim activity-based check: streaming message updates or +an open tool keep the turn alive, and silence past the driver's measured +window is confirmed with a fresh snapshot. The fallback windows are Codex +300 s, OpenCode 210 s, Claude 90 s, and other drivers 300 s. Host sleep does +not count as silence. Every watched turn records its longest observed +silence and time to first token as a `t3_turn_stream` event. + +Before interrupting a stale turn, Router saves the current turn's assistant +text and named artifacts in the job's `stale-*` directory, records a +`t3_turn_stale` event with `silenceMs`, `thresholdMs`, `thresholdSource` and +`reason`, and posts the evidence to `planner_t3_thread`. A failed post is +recorded as `t3_stale_notification` with `posted: false`. The evidence also +travels in the eventual terminal report, including after a successful retry. +Salvage copies up to 16 named regular text artifacts (`.txt`, `.log`, `.md`, +`.json`, `.patch`, `.diff`) from the workspace or host temporary directories, +up to 64 KiB each, with credential redaction. Other references and read errors +remain explicit. Reports include text and artifact excerpts with saved paths. + +A message posted onto an existing thread is watched only through the turn +it starts. A stale worker is interrupted and polled until T3 reports it +settled before moving to the next route. The fallback retains its bounded +same-route retries. After a T3 server restart with `continueThreadsAfterServerUpdate` on, interrupted Claude, Codex and OpenCode turns continue on the same thread; Grok turns are re-sent by the router. diff --git a/VERSION b/VERSION index 9b0025a..72a8a63 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.40.0 +0.41.0 diff --git a/runner/controller.py b/runner/controller.py index 110a907..1f473a0 100644 --- a/runner/controller.py +++ b/runner/controller.py @@ -409,7 +409,8 @@ def _note_t3_turn(state_dir, request_id: str, slot: str, thread_id: str | None, route: str | None, outcome: dict, client=None) -> None: """Record a watched turn's end: its last known state on the saved thread, and its stream statistics (max silence, time to first token) - in the ledger. Best effort: evidence never blocks the job.""" + in the ledger. Save stale evidence before interruption; the other + turn notes are best effort.""" state = outcome.get("state") if not thread_id: return diff --git a/runner/t3exec.py b/runner/t3exec.py index fa72e26..6c95da2 100644 --- a/runner/t3exec.py +++ b/runner/t3exec.py @@ -1073,6 +1073,11 @@ def _liveness() -> tuple[dict | None, str]: pass liveness, fallback_reason = _liveness() if await_new_turn and _awaiting_new_turn(snapshot, prior_turn_id): + if liveness is not None and liveness["stale"]: + # A queued message cannot keep its stale predecessor alive. + return _done({"state": "stalled", "liveness": liveness, + "reason": liveness.get("reason"), + "last_activity_ts": last_activity}, snapshot) latest = snapshot_thread(snapshot).get("latestTurn") if isinstance(latest, dict) and latest.get("state") == "running": # The thread is still busy with an earlier turn (a planner @@ -1094,12 +1099,7 @@ def _liveness() -> tuple[dict | None, str]: waited = awake_secs(started, now, sleeps) window = (silence_secs if silence_secs is not None else silence_window(snapshot_driver(snapshot))) - if liveness is not None: - if liveness["stale"]: - return _done({"state": "stalled", "liveness": liveness, - "reason": liveness.get("reason"), - "last_activity_ts": last_activity}, snapshot) - elif waited >= window: + if liveness is None and waited >= window: return _done({"state": "stalled", "silence": waited, "window": window, "last_part": "turn never started", "probed": True, diff --git a/tests/test_issue106.py b/tests/test_issue106.py index e398c8d..6f0ffa9 100644 --- a/tests/test_issue106.py +++ b/tests/test_issue106.py @@ -922,6 +922,9 @@ def do_GET(self): self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() + if self.path == "/api/prism/liveness?threadId=worker-1": + self.wfile.write(b'{"stale":false,"silenceMs":900000}') + return payload = {"snapshotSequence": 3, "thread": {"id": "planner-1", "projectId": "proj-1", "messages": [], @@ -952,6 +955,9 @@ def do_GET(self): self.assertEqual(seen["get_path"], "/api/orchestration/threads/planner-1") self.assertEqual(snap_["thread"]["projectId"], "proj-1") + self.assertEqual(client.prism_liveness("worker-1"), + {"stale": False, "silenceMs": 900000}) + self.assertEqual(seen["get_path"], "/api/prism/liveness?threadId=worker-1") bad = t3exec.T3Client(url, "wrong") with self.assertRaises(t3exec.T3Error): bad.thread_snapshot("planner-1") diff --git a/tests/test_issue132.py b/tests/test_issue132.py index 541ed31..909c195 100644 --- a/tests/test_issue132.py +++ b/tests/test_issue132.py @@ -452,6 +452,21 @@ def test_route_healthy_at_fallback_confirmation_cancels_stall(self): sleep_fn=lambda _: fake.complete("w", "done")) self.assertEqual(result["state"], "completed") + def test_stale_prior_turn_does_not_hide_behind_a_queued_message(self): + fake = FakeT3Client(planner=PLANNER) + fake.liveness = {"stale": True, "reason": "provider stopped"} + fake.scripts["w"] = snap("w", state="running", turn="prior") + naps = [] + + def nap(seconds): + naps.append(seconds) + fake.scripts["w"] = snap("w", state="completed", turn="new") + + result = t3exec.watch_turn(fake, "w", now_fn=lambda: NOW.timestamp(), + sleep_fn=nap, prior_turn_id="prior", await_new_turn=True) + self.assertEqual(result["state"], "stalled") + self.assertEqual(naps, []) + class StaleEvidence(unittest.TestCase): def setUp(self):