From bc60646ff902c63143983c64e43aafb5d53ea94b Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Fri, 24 Jul 2026 13:56:37 -0700 Subject: [PATCH] fix(#894): fold the truncation generation into the /history flight key MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The r8 joined-flight window, server half (Patrick-approved scope expansion): the #884 single-flight key was (ws_id, limit), so a /history dispatched AFTER a rewind/retry could join a flight whose load_messages ran BEFORE the truncation committed — the joined pre-rewind payload reads as fresh truth client-side (the client's dispatch stamp is current; the staleness is the flight's transaction point, visible only server-side) and reopened the over-rewind window through the server seam. Reachable single-user (rewind clicked during a truncated-resync fetch) and multi-viewer (any concurrent pane's /history). ChatSession gains _history_generation, bumped in _persist_truncation — the shared rewind/retry chokepoint — BEFORE the storage write (the in-memory tail is already trimmed by both callers; a spuriously fresh flight is harmless, a wrongly-joined one is not). The flight key becomes (ws_id, limit, generation): post-truncation dispatches can never join pre-truncation flights, and the client-side clearUiEpoch (prior commit) covers the converse (pre-rewind dispatches never CLEAR a post-rewind latch). Cold workstreams key at generation 0 and the first post-load truncation bumps, so cold flights cannot straddle a rewind either. Unit test mirrors the #884 coalescing determinism scheme: the owner parks in load_messages under generation 0, the mid-flight bump simulates the truncation commit, and the post-bump request must MISS the held flight (load_calls -> 2, no coalesced record). Negative-controlled: reverting the key to (ws_id, limit) fails the test. --- tests/test_workstream_endpoints.py | 49 ++++++++++++++++++++++++++++++ turnstone/core/session.py | 13 ++++++++ turnstone/core/session_routes.py | 18 +++++++++-- 3 files changed, 77 insertions(+), 3 deletions(-) diff --git a/tests/test_workstream_endpoints.py b/tests/test_workstream_endpoints.py index 41d66fec..4a5a99bd 100644 --- a/tests/test_workstream_endpoints.py +++ b/tests/test_workstream_endpoints.py @@ -1659,6 +1659,55 @@ class TestHistoryCoalescing: assert contents == ["hello", "hi there"] assert gated.load_calls == 1 + def test_history_rewind_mid_flight_starts_fresh_flight( + self, _inject_storage: Any, caplog: Any + ) -> None: + """#894: the flight key folds in the ws's truncation generation, so + a request dispatched AFTER a rewind/retry can never join a flight + whose ``load_messages`` ran before it. A joined pre-rewind payload + reads as fresh truth client-side (the client's dispatch stamp is + current — the staleness is the flight's transaction point, which + only the server can see) and reopened the coordinator's + over-rewind window through this seam. + + Choreography mirrors the join test's positive edges: the owner + parks in ``load_messages`` under generation 0; the mid-flight + rewind commit is simulated by bumping ``_history_generation`` + (``ChatSession._persist_truncation``'s bump, the shared + rewind/retry chokepoint); the second request must MISS the held + flight (``load_calls`` -> 2, no ``ws.history.coalesced`` record) + and draw its own reconstruction.""" + gated, _tenant_calls, app, url = self._scaffold(_inject_storage, "ws-flight-gen") + # The scaffold's MagicMock ws: pin the generation to a real int so + # the key is deterministic, as the live session attribute is. + mock_ws = app.state.workstreams.get("ws-flight-gen") + mock_ws._history_generation = 0 + + async def drive() -> tuple[httpx.Response, httpx.Response]: + caplog.set_level(logging.DEBUG, logger="turnstone.core.session_routes") + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://t") as c: + t1 = asyncio.create_task(c.get(url)) + await _until(lambda: gated.load_calls == 1) + # The rewind commits mid-flight: the truncation chokepoint + # bumps the generation. + mock_ws._history_generation = 1 + t2 = asyncio.create_task(c.get(url)) + # Positive edge for the MISS: t2's own load_messages entry + # bumps the counter before the gate can park it. + await _until(lambda: gated.load_calls == 2) + gated.gate.set() + r1, r2 = await asyncio.gather(t1, t2) + return r1, r2 + + r1, r2 = asyncio.run(drive()) + assert r1.status_code == 200 + assert r2.status_code == 200 + # Two reconstructions, zero joins: the post-rewind request never + # shared the pre-rewind flight. + assert gated.load_calls == 2 + assert not any("ws.history.coalesced ws=" in rec.getMessage() for rec in caplog.records) + def test_failed_shared_draw_not_fanned_out_to_joiner( self, _inject_storage: Any, caplog: Any ) -> None: diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 6fc3535d..b2af5a32 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -1661,6 +1661,9 @@ class ChatSession: self._persona_memory: bool self._apply_persona_snapshot(persona_snapshot) self._title_generated = False + # Monotonic truncation counter — folded into the /history + # single-flight key (see _persist_truncation). + self._history_generation = 0 self._read_files: set[str] = set() # Session-monotonic run counter for sub-agent id minting (see # ``_run_agent``): the parent call id alone can repeat across runs (a @@ -6962,6 +6965,16 @@ class ChatSession: does; a later resume rehydrates ``[summary] + [surviving tail]`` and the two reconcile. """ + # History generation (#894/#884 seam): every truncation bumps the + # counter the /history single-flight folds into its flight key, so + # a request dispatched AFTER a rewind/retry can never join a flight + # whose load_messages ran BEFORE it (a joined pre-rewind payload + # rendered as fresh truth on the coordinator and reopened the + # over-rewind window the #894 client latch closes). Bumped before + # the storage write on purpose: the in-memory tail is already + # trimmed by both callers, and a spuriously fresh flight is + # harmless while a wrongly-joined one is not. + self._history_generation += 1 if removed_count <= 0: return # The caller already truncated self.messages (rewind/retry both trim diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index aba89d75..6838d50e 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -3466,7 +3466,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], asyncio.Task[_HistoryFlightResult]] = {} + flights: dict[tuple[str, int, int], asyncio.Task[_HistoryFlightResult]] = {} async def history(request: Request) -> Response: if cfg.permission_gate is not None: @@ -3547,7 +3547,19 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler: # existence, limit) and must stay above it: a request may only # join a flight after its own gates passed. Everything below is # the caller-independent reconstruction, shared via ``flights``. - key = (ws_id, limit) + # The ws's truncation generation joins the key (#894): a request + # dispatched after a rewind/retry must never join a flight whose + # load_messages ran before it — the joined pre-rewind payload + # reads as fresh truth client-side (the dispatch stamp is + # current) and reopened the over-rewind window. Cold workstream: + # generation 0; the first post-load truncation bumps to 1, so a + # cold flight can never be joined across a rewind either. + live_gen = ( + getattr(live_session, "_history_generation", 0) + if live_session is not None + else 0 + ) + key = (ws_id, limit, live_gen) task = flights.get(key) joined = task is not None if task is None: @@ -3595,7 +3607,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], + key: tuple[str, int, int], mgr: SessionManager, storage: Any, app_state: Any,