fix(#894): fold the truncation generation into the /history flight key

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

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

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

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,