diff --git a/docs/system-specs/modules/history.md b/docs/system-specs/modules/history.md index 199450b6298..399b3f51457 100644 --- a/docs/system-specs/modules/history.md +++ b/docs/system-specs/modules/history.md @@ -223,6 +223,60 @@ no longer destroy older turns. consistent snapshot: it reads `_disk_older_count`, snapshots `list(slot.messages)`, and re-checks `_disk_older_count` (bounded retry) so a concurrent trim cannot interleave with the read-serialize-write. +- **Explicit-snapshot pairing (`expected_disk_older_count`)**: a caller that + freezes its own `messages` snapshot on the loop and then awaits the save cannot + use that retry — the snapshot is already frozen, and the counter the worker + reads belongs to a later moment. A trim at the window cap in that gap credits + the trimmed rows to `_disk_older_count`, so the write emits them twice: once in + the frozen prefix it now claims, once at the head of the still-frozen snapshot. + Such a caller passes the counter it observed in the SAME synchronous stretch as + the snapshot; the save refuses on drift (returns `False`, writes nothing) and + the caller answers its retryable refusal. The rewind boundary transaction does + this and re-adopts the same boundary at its commit, since the commit puts the + pre-trim window prefix back, together with `_disk_older_durable_count`, which + the trim advances beside the boundary — leaving either advanced counts a row as + having left the window front while it is back inside it. A trim landing after + the worker read the boundary cannot be refused (the correct file is already + written), so both are corrected at the commit instead. Neither is stamped by + the save, so the pre-await values are the file's truth in every interleaving. + Any other caller that freezes a snapshot across an await owes the same pairing; + `save_slot_off_loop` does not forward the parameter yet, so a boundary + transaction routed through it still reads the live counter in the worker. +- **`_disk_window_len` is deliberately left possibly SHORT after such a trim, and + the direction is the whole argument.** The save stamps it *absolutely*, so a + trim landing BEFORE the stamp has its decrement erased while one landing after + it does not — and the commit cannot distinguish the two without the count the + save actually wrote, which is not `len(snapshot)` either (a note row authorized + elsewhere is filtered out of the write, so the snapshot can be longer than the + file's window region). Over-claiming is the harmful direction: a later trim then + credits rows to the frozen prefix that the file does not hold, and the next save + re-emits window rows. Under-claiming costs no rows — it under-credits the prefix, + warns about rows that are in fact on disk, and drops the following save onto a + whole-file re-read, while the foreign-append merge below preserves the on-disk + window line the memory window has dropped. Making it exact wants the save to + publish its whole witness set as ONE routing-keyed record, which is also what + the stamping race above wants. `_frozen_prefix_cache`, the trim's last casualty, + needs nothing: the trim sets it to `None`, which only costs the next save a + re-read. +- **Witness stamping is routing-gated**: the post-write bookkeeping + (`_pending_rewrite`, `_disk_window_len`, `_disk_meta_*`, `_frozen_prefix_cache`) + describes the file this save wrote, but it lives on the live slot, which the + event loop can rebind mid-write. The write stays correct (it lands on the + transcript authorized before it), so the save re-confirms + `slot_history_key(slot)` against the key it wrote and SKIPS the stamping when + they differ — stamping would clear a `_pending_rewrite` the new transcript still + owes and claim its unsaved rows as persisted. Every witness left at its pre-save + value is the conservative reading, so the next save re-reads the prefix, + re-takes the archive-safe path, and re-observes the file. The + `ConversationLog` cache invalidation is keyed on the file that WAS written and + stays unconditional. Everything the stamp needs (the post-write `stat`, the + carried-forward `created_at`) is computed BEFORE the re-check so the stamped + region is assignments only — a save runs in a worker thread, and a syscall + inside that region is the realistic point at which the loop gets to rebind + under a half-applied stamp. Full atomicity against the loop is not reachable + from the thread (`slot._lock` is an asyncio lock, and once the rebind path has + recomputed these for its own transcript no undo is right); it wants the five + fields collapsed into one assignable record carrying the key it describes. - **Cross-process lock (`_locked`)**: `_save_slot_to_history` holds the session's cross-process `_locked` (the SAME lock `append` / `append_off_loop` / rotate / rewrite / metadata edits take) across its metadata read, frozen-prefix read, diff --git a/src/kiro_crew/dashboard/chat_persistence.py b/src/kiro_crew/dashboard/chat_persistence.py index 16a9dd2c471..64a5a550b3b 100644 --- a/src/kiro_crew/dashboard/chat_persistence.py +++ b/src/kiro_crew/dashboard/chat_persistence.py @@ -2535,6 +2535,7 @@ def _save_slot_to_history( force: bool = False, rewrite: bool = False, expected_history_key: str | None = None, + expected_disk_older_count: int | None = None, rows_only: bool = False, ) -> bool: """Persist slot messages to JSONL history (append-safe). @@ -2594,10 +2595,25 @@ def _save_slot_to_history( and persists on a later flush, and after the pop no flush ever visits that slot again. - Returns ``False`` only when the delete-won guard aborted the save because - the session was permanently deleted while this save awaited the lock — the - in-memory window was NOT persisted and must not be treated as durable. - Every other completion (including the benign no-op skips) returns ``True``. + ``expected_disk_older_count`` pairs an explicit *messages* snapshot with the + ``slot._disk_older_count`` the caller observed in the SAME synchronous stretch + it froze that snapshot in. A snapshot is internally consistent by + construction, but the frozen-prefix boundary it must be written against is + not: a concurrent ``append`` at the window cap trims the front and credits + the trimmed rows to ``_disk_older_count``, so a snapshot frozen before that + trim, written against the counter read after it, emits those rows twice — + once in the frozen prefix and once at the head of the snapshot. Supplying the + paired count makes that drift refuse the save (``False``, nothing written) + instead of committing a duplicated transcript; a caller that reads the live + counter itself cannot detect the drift at all. Ignored without *messages*, + where the bounded retry below already takes both halves together. + + Returns ``False`` when the delete-won guard aborted the save because the + session was permanently deleted while this save awaited the lock, when + ``expected_history_key`` no longer matches the slot's routing, or when + ``expected_disk_older_count`` drifted — the in-memory window was NOT + persisted and must not be treated as durable. Every other completion + (including the benign no-op skips) returns ``True``. """ if not state.conversation_log: return True @@ -2615,10 +2631,25 @@ def _save_slot_to_history( # snapshot the window, then confirm _disk_older_count is unchanged; a small # bounded retry closes the race without locks (slot._lock is an asyncio.Lock # and so cannot be acquired from this thread). An explicit snapshot is - # already consistent by construction. + # internally consistent by construction, but its PAIRING with the frozen + # prefix boundary is not -- see ``expected_disk_older_count`` above. if messages is not None: window = list(messages) disk_older = slot._disk_older_count + if expected_disk_older_count is not None and disk_older != expected_disk_older_count: + # The window hit the cap and trimmed while this save was in flight, + # so the trimmed rows are now credited to the frozen prefix AND + # still present at the head of the frozen snapshot. Writing would + # duplicate them. Refuse like the guards below: nothing written, and + # the caller (which holds the retryable-503 contract) re-decides + # against the state that actually exists. + logger.warning( + "Slot %s save refused: frozen prefix moved from %d to %d during the write", + slot.key, + expected_disk_older_count, + disk_older, + ) + return False else: for _ in range(_FLUSH_SNAPSHOT_RETRIES): disk_older = slot._disk_older_count @@ -3354,45 +3385,83 @@ def _refresh_under_lock(meta: dict) -> bool: logger.debug( "could not restore pre-close mtime for %s", history_key, exc_info=True ) - # A rewrite (archive-safe) save succeeded → clear the pending-rewrite - # flag so later saves return to the cheap default path. - if rewrite: - slot._pending_rewrite = False - # Record how many window messages are now on disk so memory trimming - # can safely fold leading window messages into the frozen prefix. - slot._disk_window_len = len(window) - # Record the disk identity this save just wrote (carried forward - # from ``existing_meta`` when present), so the delete-won guard can - # recognize a file recreated by another writer after a permanent - # delete on the NEXT save. - slot._disk_meta_created_at = str(meta_line.get("created_at") or "") - # A committed save is a direct observation of the file this slot - # writes — even when the carried-forward metadata is legacy and - # has no ``created_at`` for the identity string above. - slot._disk_meta_observed = True - # Record the post-write mtime in the frozen-prefix cache (even when - # there is no frozen prefix, ``disk_older == 0``). The cache doubles - # as the "did another process touch this file since we last wrote - # it?" signal: a matching mtime on the next save proves THIS slot was - # the last writer, so the frozen prefix is reusable and no NEW + # The witnesses below all describe THIS file. They live on the live + # slot, so they may only be stamped while the slot still routes to + # the transcript this save wrote. The event loop can rebind the slot + # (a cron injection re-linking it) after the routing snapshot above + # and while this worker writes: the write itself stays correct (it + # lands on the authorized transcript), but stamping would then + # describe the OLD file on a slot that now writes the NEW one — + # clearing ``_pending_rewrite`` the new transcript still owes, + # over-claiming ``_disk_window_len`` rows as persisted, and handing + # the delete-won guard another file's identity. Skipping leaves every + # witness at its pre-save value, which is the conservative side of + # each one: the next save re-reads the prefix, re-takes the + # archive-safe path, and re-observes the file. The cache + # invalidations after this block are keyed on the file that WAS + # written, so they stay unconditional. + # Everything the stamping needs is computed BEFORE the routing + # re-check, so the stamped region is assignments only: this runs in a + # worker thread, and a syscall between the check and the last + # assignment is the realistic point at which the event loop gets to + # rebind the slot underneath a half-applied stamp. It cannot be made + # atomic against the loop from here (``slot._lock`` is an asyncio lock + # and no undo is right once the rebind path has recomputed these for + # its own transcript) -- collapsing the five fields into one + # assignable record carrying the key it describes is the real fix, and + # belongs with that record rather than here. + # + # The frozen-prefix cache records the post-write mtime (even when + # there is no frozen prefix, ``disk_older == 0``). It doubles as the + # "did another process touch this file since we last wrote it?" + # signal: a matching mtime on the next save proves THIS slot was the + # last writer, so the frozen prefix is reusable and no NEW # cross-process append can have landed — letting the foreign-append - # scan take the O(window) fast path instead of re-reading the - # whole file. The foreign lines this save just preserved are cached + # scan take the O(window) fast path instead of re-reading the whole + # file. The foreign lines this save just preserved are cached # alongside so the fast path re-emits them verbatim: they now live in # the on-disk window region (after the frozen prefix), and because # ``disk_older`` is unchanged a bare frozen+window rebuild on the next # save would otherwise silently delete them. + _post_write_cache: tuple[float, int, int, str, list[str]] | None try: _st = path.stat() - slot._frozen_prefix_cache = ( + except OSError: + _post_write_cache = None + else: + _post_write_cache = ( _st.st_mtime, _st.st_size, disk_older, frozen_prefix, foreign_lines, ) - except OSError: - slot._frozen_prefix_cache = None + # The disk identity this save just wrote (carried forward from + # ``existing_meta`` when present), so the delete-won guard can + # recognize a file recreated by another writer after a permanent + # delete on the NEXT save. + _post_write_created_at = str(meta_line.get("created_at") or "") + if slot_history_key(slot) == history_key: + # A rewrite (archive-safe) save succeeded → clear the pending-rewrite + # flag so later saves return to the cheap default path. + if rewrite: + slot._pending_rewrite = False + # How many window messages are now on disk, so memory trimming can + # safely fold leading window messages into the frozen prefix. + slot._disk_window_len = len(window) + slot._disk_meta_created_at = _post_write_created_at + # A committed save is a direct observation of the file this slot + # writes — even when the carried-forward metadata is legacy and + # has no ``created_at`` for the identity string above. + slot._disk_meta_observed = True + slot._frozen_prefix_cache = _post_write_cache + else: + logger.warning( + "Slot %s was rebound from %s while its save was in flight; " + "leaving the persistence witnesses at their pre-save values", + slot.key, + history_key, + ) state.conversation_log._invalidate_cache(history_key) state.conversation_log.note_tab_id(history_key, tab_id) return True diff --git a/src/kiro_crew/dashboard/chat_rewind.py b/src/kiro_crew/dashboard/chat_rewind.py index 0c22815111f..1af0bc6e873 100644 --- a/src/kiro_crew/dashboard/chat_rewind.py +++ b/src/kiro_crew/dashboard/chat_rewind.py @@ -266,6 +266,21 @@ async def api_chat_slot_rewind(request: web.Request) -> web.Response: redacted_content, _ = redact_credentials(redacted_content) prospective_slot.append("user", redacted_content, "msg msg-u") msgs_snapshot = list(prospective_slot.messages) + # The frozen-prefix boundary this snapshot must be written against. An + # ``append`` at the window cap credits trimmed rows to this counter, so a + # save that read it AFTER such a trim would emit the trimmed rows twice -- + # once in the frozen prefix, once at the head of the snapshot. Captured + # here, with no await between it and the snapshot above, so the pair is + # exact; the save refuses on drift and this endpoint answers its retryable + # 503. + pre_await_disk_older_count = slot._disk_older_count + # The trim advances the durable POSITION base beside the disk boundary, + # and the two must move together or absolute positions + # (``session_control.read_messages``) disagree with the window: a row + # counted as having left the front while it is still IN the window either + # refuses a valid cursor (``since < base``) or repeats rows. The save has + # no contract on this one, so it travels only to the commit. + pre_await_disk_older_durable_count = slot._disk_older_durable_count retired_question_ids = [ question_id for question_id in slot._question_pending @@ -399,17 +414,46 @@ def _commit_live_state() -> None: slot.invalidate_source_links() slot._dirty = True slot._resumed_count = 0 - # Deliberately NOT copied from ``prospective_slot``: the - # persistence witnesses (``_pending_rewrite``, ``_disk_*``, - # ``_frozen_prefix_cache``). The save above ran on the LIVE - # slot and stamped them with the post-rewrite truth - # (``_pending_rewrite`` cleared, disk window/meta/mtime cache - # matching the truncated file); the prospective copies are the - # PRE-save values. Restoring those would re-arm - # ``_pending_rewrite`` -- making the next flush repeat the - # destructive rewrite and discard any cross-process append - # (workflow/cron) that landed in between -- and would move the - # monotone ``_disk_tail_ts`` floor backwards. + # The frozen-prefix boundary the file was just written against. + # A cap-trim landing after the save read this counter credits + # rows to the prefix that the line above puts BACK in the live + # window (the prospective list was frozen pre-trim), leaving the + # slot claiming one row in two places -- the next default save + # would then emit it twice. The save wrote + # ``prefix(pre_await) + snapshot`` and stamped + # ``_disk_window_len`` to match, so adopting the same boundary is + # what makes the three agree. The durable position base moves with + # it, for the same reason and on the same rows -- leaving it + # advanced would count a row as having left the front while it is + # back in the window. Both are no-ops when nothing trimmed. + slot._disk_older_count = pre_await_disk_older_count + slot._disk_older_durable_count = pre_await_disk_older_durable_count + # ``_disk_window_len`` is deliberately NOT corrected here, and the + # direction is the whole argument. The save stamps it absolutely, so a + # trim BEFORE the stamp has its decrement erased and a trim AFTER it + # does not -- the commit cannot tell the two apart without the count + # the save actually wrote (which is not ``len(msgs_snapshot)``: a note + # row authorized elsewhere is filtered out of the write). Guessing + # risks over-claiming, which makes a later trim credit rows to the + # frozen prefix that are not in it -- the duplication this transaction + # exists to prevent. Leaving it possibly SHORT is the safe direction and + # costs no rows: a short count under-credits the prefix, and the + # foreign-append merge preserves an on-disk window line the memory + # window has dropped. Making it exact wants the save to publish its + # whole witness set as one routing-keyed record; see history.md. + # ``_frozen_prefix_cache``, the trim's last casualty, needs nothing -- + # the trim sets it to None, which only costs the next save a re-read. + # + # Deliberately NOT copied from ``prospective_slot``: the remaining + # persistence witnesses (``_pending_rewrite``, ``_disk_meta_*``, + # ``_frozen_prefix_cache``). The save above ran on the LIVE slot + # and stamped them with the post-rewrite truth + # (``_pending_rewrite`` cleared, disk meta/mtime cache matching the + # truncated file); the prospective copies are the PRE-save values. + # Restoring those would re-arm ``_pending_rewrite`` -- making the + # next flush repeat the destructive rewrite and discard any + # cross-process append (workflow/cron) that landed in between -- + # and would move the monotone ``_disk_tail_ts`` floor backwards. if slot._pending: slot.event.set() else: @@ -454,6 +498,7 @@ def _commit_live_state() -> None: slot, msgs_snapshot, expected_history_key=expected_history_key, + expected_disk_older_count=pre_await_disk_older_count, ) ) try: diff --git a/test/test_dashboard_chat_rewind.py b/test/test_dashboard_chat_rewind.py index 4a8143b3939..2c477f76b8b 100644 --- a/test/test_dashboard_chat_rewind.py +++ b/test/test_dashboard_chat_rewind.py @@ -11,6 +11,8 @@ from aiohttp.test_utils import TestClient, TestServer from chat_test_helpers import _make_app, _make_state +from kiro_crew.dashboard import chat_persistence + @pytest.fixture(autouse=True) def _mock_run_chat(monkeypatch): @@ -298,7 +300,9 @@ async def test_rewind_commit_keeps_the_post_save_persistence_witnesses( slot._disk_tail_ts = "2026-05-21T15:00:00Z" state.sessions._session_map.get = MagicMock(return_value="") - def _save_stamps_witnesses(_state, saved_slot, msgs, *, expected_history_key): + def _save_stamps_witnesses( + _state, saved_slot, msgs, *, expected_history_key, expected_disk_older_count + ): # Emulate the real save's post-write bookkeeping on the live slot. saved_slot._pending_rewrite = False saved_slot._disk_window_len = len(msgs) @@ -326,6 +330,168 @@ def _save_stamps_witnesses(_state, saved_slot, msgs, *, expected_history_key): if slot.task: slot.task.cancel() + @pytest.mark.asyncio + async def test_rewind_pairs_the_snapshot_with_the_frozen_prefix_boundary( + self, tmp_path, monkeypatch + ): + """The frozen window must be written against the boundary it was frozen at. + + ``_disk_older_count`` is where the frozen prefix ends, and an ``append`` + at the window cap moves it. Reading it in the save worker instead of + pairing it with the snapshot writes the trimmed rows twice -- once in the + prefix, once at the head of the snapshot. This pins the wiring (the + endpoint hands the save the PRE-await boundary and re-adopts it at the + commit); the two tests below drive the save's own refusal for real. + """ + state = _make_state(tmp_path) + slot = _populate_slot(state) + state.sessions._session_map.get = MagicMock(return_value="") + seen: dict[str, object] = {} + + async def _moves_the_boundary(key, **kwargs): + # Stands in for a cap-trim landing inside the awaited boundary, which + # advances the disk boundary and the durable position base together. + slot._disk_older_count = 7 + slot._disk_older_durable_count = 7 + return True + + state.sessions.discard_conversation = AsyncMock(side_effect=_moves_the_boundary) + + def _record_pairing( + _state, saved_slot, msgs, *, expected_history_key, expected_disk_older_count + ): + seen["boundary"] = expected_disk_older_count + return True + + monkeypatch.setattr( + "kiro_crew.dashboard.chat_rewind._save_slot_to_history", _record_pairing + ) + + app = _make_app(state) + async with TestClient(TestServer(app)) as client: + resp = await client.post( + "/api/chat/slots/src/rewind", + json={"at_message_index": 0, "content": "edited first question"}, + ) + assert resp.status == 200 + + # The PRE-await boundary, not the one the worker would have read. + assert seen["boundary"] == 0 + assert slot._disk_older_count == 0 # the commit re-adopts it + assert slot._disk_older_durable_count == 0 # and the durable base with it + if slot.task: + slot.task.cancel() + + @pytest.mark.asyncio + async def test_rewind_refuses_when_a_cap_trim_moves_the_frozen_prefix( + self, tmp_path, monkeypatch + ): + """A trim that re-credits rows to the frozen prefix must refuse the save. + + Drives the refusal through the REAL save. The file written is + ``frozen_prefix + snapshot`` and the prefix boundary is + ``_disk_older_count``; an ``append`` at the cap trims the front and + credits the trimmed rows to that counter, so a save reading the counter + AFTER the trim writes those rows twice -- once in the prefix it now + claims, once at the head of the still-frozen snapshot. The paired count + makes that drift refuse instead: nothing written, retryable 503, and the + transcript on disk is untouched. + """ + monkeypatch.setattr("kiro_crew.dashboard.state._MAX_SLOT_MESSAGES", 4) + state = _make_state(tmp_path) + slot = _populate_slot(state) + state.sessions._session_map.get = MagicMock(return_value="") + # The rows must be ON DISK for a trim to credit them to the frozen + # prefix: ``append`` only counts the persisted portion of the evicted + # slice. + await asyncio.to_thread(state.flush_slot_now, slot) + assert slot._disk_window_len == 4 + assert slot._disk_older_count == 0 + persisted_before = [ + (m["role"], m["content"]) for m in state.conversation_log.read_messages("dashboard:src") + ] + assert [content for _role, content in persisted_before] == [ + "first question", + "first answer", + "second question", + "second answer", + ] + + async def _flush_with_cap_arrival(): + # Runs after the native discard and BEFORE the history rewrite, so + # the boundary moves while the snapshot is already frozen. + slot.append("assistant", "workflow result", "msg msg-a") + assert slot._disk_older_count == 1 + + state.sessions.discard_conversation = AsyncMock(return_value=True) + state.sessions.aflush = AsyncMock(side_effect=_flush_with_cap_arrival) + + app = _make_app(state) + async with TestClient(TestServer(app)) as client: + resp = await client.post( + "/api/chat/slots/src/rewind", + json={"at_message_index": 2, "content": "edited second question"}, + ) + assert resp.status == 503 + assert (await resp.json())["code"] == "rewind_save_failed" + + # A bare live-counter read would write prefix ["first question"] plus a + # snapshot that still starts with it. + assert [ + (m["role"], m["content"]) for m in state.conversation_log.read_messages("dashboard:src") + ] == persisted_before + if slot.task: + slot.task.cancel() + + @pytest.mark.asyncio + async def test_rewind_leaves_the_witnesses_alone_when_a_rebind_wins_the_write(self, tmp_path): + """A rebind landing inside the write must not stamp the live witnesses. + + The write itself is correct -- the save's own routing pin was checked + before it and the bytes land on the authorized transcript. What must not + follow is the post-write bookkeeping: those witnesses describe the file + just written, and the slot now writes a DIFFERENT one. Stamping would + clear a ``_pending_rewrite`` the new transcript still owes and claim its + unsaved rows as persisted, which a later flush then believes. + """ + state = _make_state(tmp_path) + slot = _populate_slot(state) + state.sessions._session_map.get = MagicMock(return_value="") + original_messages = list(slot.messages) + slot._pending_rewrite = True # a rewrite is owed on the CURRENT transcript + slot._disk_window_len = 0 + slot._disk_meta_observed = False + + real_atomic_write = chat_persistence.atomic_write + + def _rebinding_atomic_write(path, payload, **kwargs): + # The last thing before the witness stamping: the slot moves to + # another transcript with the authorized bytes already on their way + # to disk. + slot.linked_session_key = "slack:9876543210.999" + return real_atomic_write(path, payload, **kwargs) + + state.sessions.discard_conversation = AsyncMock(return_value=True) + + with patch.object(chat_persistence, "atomic_write", _rebinding_atomic_write): + app = _make_app(state) + async with TestClient(TestServer(app)) as client: + resp = await client.post( + "/api/chat/slots/src/rewind", + json={"at_message_index": 0, "content": "edited first question"}, + ) + assert resp.status == 503 + assert (await resp.json())["code"] == "rewind_slot_rebound" + + assert slot.messages == original_messages + # Untouched, so the next flush of the NEW transcript still archives + # before it rewrites and still treats its window as unpersisted. + assert slot._pending_rewrite is True + assert slot._disk_window_len == 0 + assert slot._disk_meta_observed is False + if slot.task: + slot.task.cancel() + @pytest.mark.asyncio async def test_rewind_app_cannot_reach_a_channel_linked_session(self, tmp_path): """An app-owned slot with a channel link must not rewind through it.