diff --git a/scripts/recovery_e2e.py b/scripts/recovery_e2e.py index 4d62be25..2257f53f 100644 --- a/scripts/recovery_e2e.py +++ b/scripts/recovery_e2e.py @@ -2747,7 +2747,16 @@ def _coord_stick_latch(cdp: CDP, node: Any, tag: str) -> None: clear_ui refetch AND its one bounded 2s retry, so only an organic idle-edge heal can clear it. Extracted so G4's premise (latch stuck exactly as in G3) is enforced by construction, the same rationale - _seed_three_completed_turns documents for the E family.""" + _seed_three_completed_turns documents for the E family. + + RULED (r10): G2/G5 deliberately keep their single-failure prologues + inline rather than adopting this helper — their baseline captures + and phase timings interleave INTO the prologue steps (G2 snapshots + history_requests before the click; G5 hides the instant the fail + budget drains), so a parameterized version would need a flag per + divergence and obscure the choreography it exists to clarify. The + helper serves the two double-failure scenarios whose premise must + match exactly.""" if not _poll_until(lambda: cdp.evaluate(_COORD_ROWS_JS) == 3, 20, 0.2): raise AssertionError(f"{tag}: three user rows never rendered") if not _poll_until(lambda: cdp.evaluate("window.__esOpens") >= 1, 10, 0.05): diff --git a/tests/test_coordinator_page.py b/tests/test_coordinator_page.py index 0313959c..599b4c5d 100644 --- a/tests/test_coordinator_page.py +++ b/tests/test_coordinator_page.py @@ -943,6 +943,21 @@ def test_coordinator_history_stale_latch_contract(): ) # destroy() must abort the in-flight fetch (dead-not-inert, the # staleRetryTimer ruling applied to the r7 bound). + # Producer pins first — the destroy() consumer sweep below is + # satisfiable by an always-empty Set without them. + assert body.count("histCtrls.add(histCtrl)") == 1, ( + "every dispatch must register its controller in the abort Set." + ) + assert body.count("histCtrls.delete(histCtrl)") == 1, ( + "the fetch finally must release its own controller — without the " + "delete the Set grows for the life of the pane." + ) + assert body.index("histCtrls.add(histCtrl)", fetch_start) < awt, ( + "the controller must be registered BEFORE the await." + ) + assert fin < body.index("histCtrls.delete(histCtrl)", fetch_start), ( + "the controller release must sit in the fetch finally." + ) destroy_code = _strip_comments(destroy_slice) assert "histCtrls.forEach" in destroy_code and ".abort()" in destroy_code, ( "destroy() must abort EVERY in-flight /history (a Set — a " diff --git a/tests/test_rewind_retry.py b/tests/test_rewind_retry.py index 2a123343..02d49823 100644 --- a/tests/test_rewind_retry.py +++ b/tests/test_rewind_retry.py @@ -77,7 +77,7 @@ class NullUI: pass -def _make_session(tmp_db) -> ChatSession: +def _make_session(tmp_db, ws_id: str | None = None) -> ChatSession: return ChatSession( client=MagicMock(), model="test-model", @@ -86,6 +86,7 @@ def _make_session(tmp_db) -> ChatSession: temperature=0.5, max_tokens=4096, tool_timeout=30, + ws_id=ws_id, ) @@ -439,16 +440,7 @@ def test_truncation_bumps_history_generation(tmp_db) -> None: storage = get_storage() storage.register_workstream("ws-gen-pin", kind="interactive", user_id="test-user") - session = ChatSession( - client=MagicMock(), - model="test-model", - ui=NullUI(), - instructions="", - temperature=0.5, - max_tokens=4096, - tool_timeout=30, - ws_id="ws-gen-pin", - ) + session = _make_session(tmp_db, ws_id="ws-gen-pin") _populate_simple(session) for role, content in ( ("user", "Hello"), diff --git a/tests/test_workstream_endpoints.py b/tests/test_workstream_endpoints.py index 2ea60576..16ce1704 100644 --- a/tests/test_workstream_endpoints.py +++ b/tests/test_workstream_endpoints.py @@ -2401,6 +2401,9 @@ class TestHistoryReasoningRehydration: provider_data = json.dumps([{"type": "thinking", "thinking": "hidden", "signature": "s"}]) _inject_storage.save_message(ws_id, "assistant", "Answer.", provider_data=provider_data) live_session = SimpleNamespace( + # The real Workstream carries .session (ChatSession | None); + # the flight key's typed generation read requires the shape. + session=None, id=ws_id, _registry=SimpleNamespace( get_config=lambda alias: SimpleNamespace(surface_persisted_reasoning=False) diff --git a/turnstone/console/static/coordinator/coordinator.js b/turnstone/console/static/coordinator/coordinator.js index cd9f380b..acee1b55 100644 --- a/turnstone/console/static/coordinator/coordinator.js +++ b/turnstone/console/static/coordinator/coordinator.js @@ -5998,7 +5998,7 @@ function createCoordinatorPane(root, wsId, opts) { if (staleRetryTimer) { clearTimeout(staleRetryTimer); staleRetryTimer = null; - + } toolRows.clear(); activeBatch = null; diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index 7e7b4934..3f0e1b8b 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -3471,7 +3471,7 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler: # isolation of every other lifted verb. Touched only from the event # loop thread — no lock needed; the ``limit`` component is required # (a limit=10 caller must not receive a limit=500 payload). - flights: dict[tuple[str, int, int], asyncio.Task[_HistoryFlightResult]] = {} + flights: dict[tuple[str, int, int | None], asyncio.Task[_HistoryFlightResult]] = {} async def history(request: Request) -> Response: if cfg.permission_gate is not None: @@ -3562,10 +3562,17 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler: # mgr.get returns the Workstream WRAPPER — the counter lives on # its ChatSession (the G7 harness caught a direct getattr # silently defaulting to 0 forever, which re-enabled joining). - live_gen = ( - getattr(getattr(live_session, "session", None), "_history_generation", 0) - if live_session is not None - else 0 + # Typed access, not getattr chains, so mypy carries the shape. + # A cold/detached workstream keys on None, NEVER 0: an eviction + # or close landing inside a held flight's window would otherwise + # let a post-truncation request join a generation-0 live flight + # (rewinds need a live session, so two COLD flights are always + # mutually safe — and a rehydrated session restarting at 0 can + # never share the manager slot with its evicted predecessor). + live_gen: int | None = ( + live_session.session._history_generation + if live_session is not None and live_session.session is not None + else None ) key = (ws_id, limit, live_gen) task = flights.get(key) @@ -3615,7 +3622,7 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler: return JSONResponse({"ws_id": ws_id, "messages": messages, "cursor": cursor}) async def _run_flight( - key: tuple[str, int, int], + key: tuple[str, int, int | None], mgr: SessionManager, storage: Any, app_state: Any,