fix(#894): cold flights key on None; typed generation access; abort-Set producer pins

Review round 10 (1 minor bug; 2 major + 2 small quality — the majors
both pins-that-cannot-fail).

- The flight key's cold fallback was the literal 0, which collides
  with a live session's generation 0: an eviction/close landing inside
  a held flight's window let a post-truncation request rejoin a
  generation-0 pre-truncation flight.  Cold/detached workstreams now
  key on None (rewinds need a live session, so two cold flights are
  always mutually safe; a rehydrated session restarting at 0 can never
  share the manager slot with its evicted predecessor — documented
  at-site).  The read is TYPED (live_session.session._history_generation)
  so mypy carries the shape a getattr chain hid — and the typed access
  immediately surfaced an unfaithful SimpleNamespace mock in the
  reasoning-rehydration tests (no .session attr), now made faithful.
- Abort-Set producer pins: histCtrls.add exactly once and BEFORE the
  await, delete exactly once and in the finally — without them the
  destroy() consumer sweep was satisfiable by an always-empty Set.
- _make_session gains ws_id; the generation producer pin uses it.
- _coord_stick_latch: G2/G5's inline single-failure prologues RULED
  deliberate at-site (their baselines/phase timings interleave into
  the prologue; a per-divergence flag would obscure the choreography).
- Stray trailing whitespace stripped.

250 pins green; G2/G5/G7 re-run READY.

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):

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 "

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"),

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)

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;

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,