From 60f6dc07a2707055140b0c2105291275d296a248 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Fri, 24 Jul 2026 14:30:21 -0700 Subject: [PATCH] fix(#894): drop the unreachable epoch guard; abort-Set; bump-after-delete; producer pins MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review round 9 (4 minor bug, 4 quality, 1 perf nit; security zero). - The r8 clearUiEpoch guard was UNREACHABLE (r9 bug find): clear_ui always dispatches immediately after bumping, so a stale-epoch dispatch is also a stale-seq dispatch and the currency gate discards it before it can paint or clear — the client half of the joined- flight fix was already carried by seq, and the server generation key is the sole load-bearing layer. Machinery removed (decl, bump, capture, conditional clear, section-9 pins); the latch-clear comment now states the two-layer accounting. - destroy()'s abort handle becomes a Set: a newest-wins single slot, nulled by the newer dispatch's finally, left an OLDER overlapping fetch unabortable — the destroyed closure pinned for the bound's remainder. Pinned. - _history_generation now bumps AFTER delete_messages_after: flights rebuild from storage, so old-generation-reads-post-delete is the harmless spuriously-fresh direction while new-generation-reads- pre-delete would be wrongly joinable; the count/floor error paths correctly leave it unbumped. Two-arm producer pin in test_rewind_retry (persisted-rows bump on rewind AND retry; in-memory-only error path must NOT bump) — the flight test's mock can no longer mask a deleted bump. - The harness load_calls increment takes a lock (to_thread workers genuinely overlap under delay_load; a lost update false-fails G7). - G7's viewer B is now a background authenticated GET (a raw request enters load_messages identically; the second browser bought no proof); stale two-tuple key comments and the coalescing matrix line updated; the _send_in_page enumeration dropped for prose. 250 pins green; G1/G6/G7 re-run READY. --- scripts/recovery_e2e.py | 44 +++++++----- tests/_sse_recovery_server.py | 9 ++- tests/test_coordinator_page.py | 51 +++----------- tests/test_rewind_retry.py | 63 +++++++++++++++++ tests/test_workstream_endpoints.py | 5 +- .../console/static/coordinator/coordinator.js | 67 ++++++++----------- turnstone/core/session.py | 23 ++++--- turnstone/core/session_routes.py | 9 ++- 8 files changed, 158 insertions(+), 113 deletions(-) diff --git a/scripts/recovery_e2e.py b/scripts/recovery_e2e.py index 59939a9a..4d62be25 100644 --- a/scripts/recovery_e2e.py +++ b/scripts/recovery_e2e.py @@ -158,8 +158,9 @@ turn), posts 1, ``history_requests`` grew. Stamps ``RECOVERY-READY-COORDORPHANREWIND-posts1-rows1-orphan0``. Scenario G7 (coord-joined-flight, #894 r8): the joined-flight window. -Two browsers on one ws; ``delay_load`` parks B's pre-rewind /history -flight open INSIDE ``load_messages`` (the flight layer — the fault +One browser plus a background GET ("viewer B") on one ws; +``delay_load`` parks B's pre-rewind /history flight open INSIDE +``load_messages`` (the flight layer — the fault layer's ``delay_history`` cannot overlap flights); A rewinds mid-hold. A's clear_ui refetch must MISS the held flight — the server folds the truncation generation into the #884 flight key — proven by @@ -319,6 +320,7 @@ import socket import struct import subprocess import sys +import threading import time import urllib.request from pathlib import Path @@ -1978,7 +1980,8 @@ def _poll_until(pred: Any, timeout: float, interval: float = 0.1) -> bool: def _send_in_page(cdp: CDP, message: str) -> None: """POST /send from inside the page via the pane's own authFetch (cookie auth, node-proxy base) — the one shared shape for scenarios that drive a - turn mid-flight (E1/E4/G3). A raw POST emits no live user row, so the sent + turn mid-flight (the shared shape for every scenario that drives a + turn outside the composer). A raw POST emits no live user row, so the sent turn appears only via the next /history render.""" cdp.evaluate( "window.authFetch('/v1/api/workstreams/' + " @@ -3245,9 +3248,10 @@ def run_coord_joined_flight(chrome: str) -> str: Choreography: page A paints three rows; ``delay_load`` then holds every reconstruction open INSIDE ``load_messages`` (the flight layer — ``delay_history`` sleeps in the fault layer, before the - route, where flights never overlap); page B (a second browser on - the SAME ws) navigates, parking a pre-rewind flight; A rewinds - mid-hold — its clear_ui refetch must MISS B's held flight + route, where flights never overlap); a background authenticated GET + ("viewer B" — a raw request enters ``load_messages`` identically, + without a second browser's cost) parks a pre-rewind flight; A + rewinds mid-hold — its clear_ui refetch must MISS B's held flight (``load_calls`` grows by TWO: a joined request never enters ``load_messages`` — the e2e twin of the unit test's proof) and render the POST-rewind single row once the holds release. Pre-fix @@ -3255,11 +3259,8 @@ def run_coord_joined_flight(chrome: str) -> str: truth.""" node, ws_id = _seed_three_completed_turns("browser-coord-joined-flight") profile_a = Path(_scratch()) / "chrome-coord-joined-a" - profile_b = Path(_scratch()) / "chrome-coord-joined-b" proc_a, cdp_port_a = _launch_chrome(chrome, profile_a) - proc_b = None cdp_a: CDP | None = None - cdp_b: CDP | None = None try: cdp_a = CDP(_page_ws_url(cdp_port_a)) url = f"{node.base_url}/coord-recovery?ws_id={ws_id}&scenario=coord-joined-flight" @@ -3271,12 +3272,23 @@ def run_coord_joined_flight(chrome: str) -> str: load_baseline = node.load_calls # Hold every reconstruction open INSIDE load_messages (the flight # layer — delay_history sleeps in the fault layer, before the - # route, where flights never overlap), then park B's pre-rewind - # flight under the hold. + # route, where flights never overlap), then park the pre-rewind + # flight under the hold. The parked "viewer B" is a plain + # authenticated GET on a background thread: a raw request enters + # load_messages identically, and a second headless browser added + # ~300 MB + seconds of launch for no additional proof. node.delay_load(3000) - proc_b, cdp_port_b = _launch_chrome(chrome, profile_b) - cdp_b = CDP(_page_ws_url(cdp_port_b)) - _set_cookie_and_navigate(cdp_b, node.base_url, node.token, url) + + def _b_get() -> None: + req = urllib.request.Request( + f"{node.base_url}/v1/api/workstreams/{ws_id}/history", + headers={"Cookie": f"turnstone_auth_server={node.token}"}, + ) + with contextlib.suppress(Exception): + urllib.request.urlopen(req, timeout=30).read() + + b_thread = threading.Thread(target=_b_get, daemon=True) + b_thread.start() if not _poll_until(lambda: node.load_calls == load_baseline + 1, 15, 0.05): raise AssertionError("coord-joined-flight: B's flight never entered load_messages") # A rewinds mid-hold: second row -> 2 turns -> server keeps one @@ -3304,11 +3316,7 @@ def run_coord_joined_flight(chrome: str) -> str: finally: if cdp_a is not None: cdp_a.close() - if cdp_b is not None: - cdp_b.close() _kill(proc_a) - if proc_b is not None: - _kill(proc_b) node.stop() diff --git a/tests/_sse_recovery_server.py b/tests/_sse_recovery_server.py index 15f6fa92..d4e01e59 100644 --- a/tests/_sse_recovery_server.py +++ b/tests/_sse_recovery_server.py @@ -152,11 +152,18 @@ class RecoveryServer: # singleton's load_messages; restored in stop(). self._load_delay_ms = 0 self._load_calls = 0 + # load_messages runs on asyncio.to_thread WORKERS, and delay_load + # exists precisely to overlap two of them — unlike the + # single-writer HTTP counters, this one has genuine concurrent + # writers, so the increment takes a lock (a lost update would + # false-FAIL G7's load_delta === 2, or mask a third load). + self._load_calls_lock = threading.Lock() _storage_obj = get_storage() self._orig_load_messages = _storage_obj.load_messages def _delayed_load(*a: Any, **k: Any) -> Any: - self._load_calls += 1 + with self._load_calls_lock: + self._load_calls += 1 result = self._orig_load_messages(*a, **k) # Sleep AFTER the load: the held flight must hold the data it # actually read (its transaction point), so a flight parked diff --git a/tests/test_coordinator_page.py b/tests/test_coordinator_page.py index aa016ccf..0313959c 100644 --- a/tests/test_coordinator_page.py +++ b/tests/test_coordinator_page.py @@ -608,7 +608,11 @@ def test_coordinator_history_stale_latch_contract(): the r6 critical; never plain busy — the r5 critical), the optimistic busySource flavor, and stream-OPENness. The latch-clear sits below every skip point, so a skipped render - can never reopen the affordances over a stale DOM. + can never reopen the affordances over a stale DOM. (Joined + pre-rewind server flights are closed SERVER-side — the #884 + flight key folds in the truncation generation; an r8 + client-epoch guard here was unreachable, being redundant with + the seq gate, and was removed in r9.) """ from pathlib import Path @@ -939,45 +943,12 @@ 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). - assert "activeHistCtrl.abort()" in _strip_comments(destroy_slice), ( - "destroy() must abort any in-flight /history — the 15s bound " - "alone pins the destroyed closure until it fires." - ) - - # 9. Rewind-freshness epoch (r8): the #884 single-flight can hand a - # joiner a pre-rewind payload; only a dispatch that post-dates - # the latest clear_ui may CLEAR the latch. Producer: the bump - # sits in the clear_ui case before the latch set; the capture - # sits beside seq, before the await; the clear is conditional. - assert body.count("clearUiEpoch++;") == 1, ( - "clearUiEpoch must bump in exactly one place (clear_ui arrival)." - ) - cu_case = body.index('case "clear_ui"') - bump = body.index("clearUiEpoch++;") - assert cu_case < bump < set_site, ( - "the epoch bump must sit inside the clear_ui case BEFORE the " - "latch set, so the pre-rewind flight window is stamped closed " - "before any dispatch can capture the old epoch." - ) - assert body.count("const epoch = clearUiEpoch;") == 1, ( - "refetchHistory must capture the epoch exactly once." - ) - assert body.index("const epoch = clearUiEpoch;", fetch_start) < awt, ( - "the epoch capture must sit BEFORE the await, beside seq." - ) - assert "if (epoch === clearUiEpoch) {" in fetch_code, ( - "the latch clear must be epoch-conditional — an unconditional " - "clear lets a joined pre-rewind #884 flight reopen the " - "over-rewind window through the server seam." - ) - epoch_gate = body.index("if (epoch === clearUiEpoch) {", fetch_start) - assert wipe < epoch_gate, ( - "the epoch condition guards the CLEAR (below the wipe), not the " - "render — a pre-rewind payload may paint (stale-but-real) but " - "must not clear the latch." - ) - assert body.index("historyStale = false;", fetch_start) > epoch_gate, ( - "the latch clear must sit inside the epoch-conditional block." + 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 " + "newest-wins single slot left older overlapping fetches " + "unabortable); the 15s bound alone pins the destroyed closure " + "until it fires." ) diff --git a/tests/test_rewind_retry.py b/tests/test_rewind_retry.py index 8659a983..2a123343 100644 --- a/tests/test_rewind_retry.py +++ b/tests/test_rewind_retry.py @@ -422,3 +422,66 @@ class TestRewindDBSync: assert len(db_msgs) == 2 assert db_msgs[0]["content"] == "Hello" assert db_msgs[1]["content"] == "Hi!" + + +def test_truncation_bumps_history_generation(tmp_db) -> None: + """#894: the /history single-flight keys on _history_generation, so a + truncation that DELETED storage rows must bump it — and a truncation + that could not delete (storage count unavailable/empty — the + error-path early returns) must NOT, because flights rebuild from + storage and an unbumped generation on error is the safe direction (a + stale-keyed flight reading post-delete rows is spuriously fresh; a + new-keyed flight reading pre-delete rows would be wrongly joinable). + The endpoint-level flight test mocks the counter, so this is the + PRODUCER pin: delete the bump in _persist_truncation and the first + arm fails while everything else stays green.""" + from turnstone.core.storage import get_storage + + 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", + ) + _populate_simple(session) + for role, content in ( + ("user", "Hello"), + ("assistant", "Hi there!"), + ("user", "How are you?"), + ("assistant", "I'm fine."), + ): + storage.save_message("ws-gen-pin", role, content) + + g0 = session._history_generation + assert session.rewind(1) > 0 + assert session._history_generation == g0 + 1, ( + "a storage-deleting rewind must bump the history generation" + ) + + # retry: re-persist the tail the rewind removed, then drop the last + # assistant turn. + storage.save_message("ws-gen-pin", "user", "How are you?") + storage.save_message("ws-gen-pin", "assistant", "I'm fine.") + _populate_simple(session) + g1 = session._history_generation + assert session.retry() is not None + assert session._history_generation == g1 + 1, ( + "a storage-deleting retry must bump the history generation" + ) + + # Error-path arm: an in-memory-only session (no persisted rows — the + # count<=0 early return skips the delete) must NOT bump. + bare = _make_session(tmp_db) + _populate_simple(bare) + g2 = bare._history_generation + assert bare.rewind(1) > 0 + assert bare._history_generation == g2, ( + "a truncation that could not delete storage rows must leave the " + "generation unbumped (fail-safe direction)" + ) diff --git a/tests/test_workstream_endpoints.py b/tests/test_workstream_endpoints.py index 4829bc35..2ea60576 100644 --- a/tests/test_workstream_endpoints.py +++ b/tests/test_workstream_endpoints.py @@ -1548,8 +1548,9 @@ class TestHistoryCoalescing: """Single-flight coalescing of concurrent ``/history`` requests (#884). The matrix these tests assert: join-vs-miss × gates-per-request × - failed-vs-clean shared draw × key isolation × no cross-flight - caching × owner cancellation. Driven through ``httpx.ASGITransport`` + failed-vs-clean shared draw × key isolation (ws, limit, AND the + #894 truncation generation) × no cross-flight caching × owner + cancellation. Driven through ``httpx.ASGITransport`` on a private loop because ``TestClient`` cannot hold two requests in flight at once. diff --git a/turnstone/console/static/coordinator/coordinator.js b/turnstone/console/static/coordinator/coordinator.js index dec5a55c..cd9f380b 100644 --- a/turnstone/console/static/coordinator/coordinator.js +++ b/turnstone/console/static/coordinator/coordinator.js @@ -677,26 +677,13 @@ function createCoordinatorPane(root, wsId, opts) { // an older snapshot landing late can neither double-render nor clear // the staleness latch over a newer truth. let refetchSeq = 0; - // Rewind-freshness epoch (r8): bumped at every clear_ui ARRIVAL. A - // refetch dispatch captures it, and only a render whose dispatch - // post-dates the latest clear_ui may CLEAR the staleness latch. The - // server's #884 /history single-flight can hand a joiner a payload - // whose load_messages ran BEFORE the rewind committed (the flight - // key is (ws_id, limit); joining is invisible to the client) — that - // pre-rewind payload may still PAINT (stale-but-real posture, the - // affordance gate holds) but must not clear the latch, or the exact - // over-rewind window this latch exists to close reopens through the - // server seam. The un-cleared latch arms the retry, whose fresh - // dispatch starts a NEW flight (the old one popped at settle) and - // heals with post-rewind truth. - let clearUiEpoch = 0; - // The newest in-flight /history's AbortController — destroy() aborts it + // EVERY in-flight /history AbortController — destroy() aborts them all // so a slow fetch cannot pin the destroyed pane's closure for the // bound's full 15s (the same dead-not-inert ruling destroy() applies - // to staleRetryTimer). Overlapping dispatches: each finally releases - // only its OWN handle, so the newest in-flight controller stays - // reachable for the teardown abort. - let activeHistCtrl = null; + // to staleRetryTimer). A Set, not a single slot (r9): overlapping + // dispatches settle in any order, and a newest-wins slot nulled by + // the newer dispatch's finally left the OLDER fetch unabortable. + const histCtrls = new Set(); // call_ids of tool calls whose results are still ARRIVING ON THE LIVE // STREAM — the render-time gate's tool-phase liveness signal. Fed // ONLY by live SSE events (tool_pending / tool_info add, tool_result @@ -3744,7 +3731,6 @@ function createCoordinatorPane(root, wsId, opts) { // decl). The cosmetic [data-busy] grey-out for this window stays a // deferred parity item (#890 PR ruling) — the handler-side // ``busy || historyStale`` gate carries the correctness. - clearUiEpoch++; historyStale = true; refetchHistory() .then(() => { @@ -5853,7 +5839,6 @@ function createCoordinatorPane(root, wsId, opts) { // mid-render. The seq stamp makes overlapping dispatches resolve // last-dispatch-wins at the render-time gate. const seq = ++refetchSeq; - const epoch = clearUiEpoch; refetchesInFlight++; // Bound the await (r7): the counter and both heals gate on this // fetch settling. A /history that is accepted and never answered @@ -5865,7 +5850,7 @@ function createCoordinatorPane(root, wsId, opts) { // retry on later organic edges against a fresh attempt. const histCtrl = typeof AbortController === "function" ? new AbortController() : null; - activeHistCtrl = histCtrl; + if (histCtrl) histCtrls.add(histCtrl); const histTimer = histCtrl ? setTimeout(() => histCtrl.abort(), 15000) : null; @@ -5879,7 +5864,7 @@ function createCoordinatorPane(root, wsId, opts) { hist = null; } finally { if (histTimer) clearTimeout(histTimer); - if (activeHistCtrl === histCtrl) activeHistCtrl = null; + if (histCtrl) histCtrls.delete(histCtrl); refetchesInFlight--; } // A FAILED fetch keeps the pane intact: the wipe + tracking resets @@ -6000,18 +5985,20 @@ function createCoordinatorPane(root, wsId, opts) { // latch set (which is what arms the retry/backstop), while a mid-render // throw doesn't re-close a pane whose replaceChildren already // committed (a partially painted FRESH transcript at worst - // UNDER-counts, the safe direction). The ONLY latch-clear site — - // and epoch-conditional (r8): a dispatch that PREDATES the latest - // clear_ui may have joined a pre-rewind #884 server flight; its - // payload may paint (stale-but-real) but must not clear the latch - // (see clearUiEpoch's decl) — the surviving latch arms the retry, - // whose fresh dispatch heals with post-rewind truth. - if (epoch === clearUiEpoch) { - historyStale = false; - if (staleRetryTimer) { - clearTimeout(staleRetryTimer); - staleRetryTimer = null; - } + // UNDER-counts, the safe direction). The ONLY latch-clear site. + // Freshness layers (r9 accounting): a dispatch that PREDATES the + // latest clear_ui never reaches here at all — clear_ui always + // dispatches immediately after arriving, so a stale-epoch dispatch + // is also a stale-SEQ dispatch and the currency gate above discards + // it (an r8 epoch guard here was unreachable and was removed). The + // remaining freshness hazard — a CURRENT dispatch joining a + // pre-rewind server flight — is closed server-side: the #884 flight + // key folds in the ws's truncation generation. + historyStale = false; + if (staleRetryTimer) { + clearTimeout(staleRetryTimer); + staleRetryTimer = null; + } toolRows.clear(); activeBatch = null; @@ -6341,17 +6328,17 @@ function createCoordinatorPane(root, wsId, opts) { clearTimeout(staleRetryTimer); staleRetryTimer = null; } - // Abort any in-flight /history for the same reason: the 15s bound + // Abort every in-flight /history for the same reason: the 15s bound // alone would keep the detached pane's closure alive until it fired - // (the settled fetch's render then discards via the !hist path). - if (activeHistCtrl) { + // (the settled fetches' renders then discard via the !hist path). + histCtrls.forEach((c) => { try { - activeHistCtrl.abort(); + c.abort(); } catch (_) { /* noop */ } - activeHistCtrl = null; - } + }); + histCtrls.clear(); [ cancelTimeoutId, forceTimeoutId, diff --git a/turnstone/core/session.py b/turnstone/core/session.py index b2af5a32..0541e3ff 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -6965,16 +6965,6 @@ 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 @@ -7000,6 +6990,19 @@ class ChatSession: # summarized prefix — skip rather than risk it; resume reconciles. return delete_messages_after(self._ws_id, max(floor, total - removed_count)) + # 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 AFTER + # the storage delete: flights rebuild from storage, so a flight + # keyed with the OLD generation that reads post-delete rows is the + # harmless spuriously-fresh direction, while a NEW-generation + # flight reading pre-delete rows would be the wrongly-joined one — + # and the early-return error paths above (count/floor unavailable, + # delete skipped) correctly leave the generation unbumped. + self._history_generation += 1 def rewind(self, n: int) -> int: """Drop the last *n* complete turns from the conversation. diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index 5e818b49..7e7b4934 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -3438,7 +3438,8 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler: inline on the event loop). Restart-herd coalescing (issue #884): concurrent requests for the - same ``(ws_id, limit)`` share ONE reconstruction (single-flight) — + same ``(ws_id, limit, history_generation)`` share ONE reconstruction + (single-flight) — after a node restart every open pane resyncs via REST ``/history`` inside the same jitter window, and each un-coalesced request repeats the full ``load_messages`` → decoration → projection pipeline @@ -3457,7 +3458,11 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler: cfg: per-kind policy bundle. """ - # In-flight reconstructions, keyed ``(ws_id, limit)``. Holds ONLY + # In-flight reconstructions, keyed ``(ws_id, limit, + # history_generation)`` — the third component isolates flights across + # rewind/retry truncations (see ChatSession._persist_truncation), so + # a post-truncation request can never join a pre-truncation + # reconstruction. Holds ONLY # live flights — each task pops its own key in a ``finally`` before # completing, so a later request can never read a completed (stale) # result: this map is a single-flight, not a cache. Scoped to this