From d544f537222cda5d0d291e8ba6a6f429ed1aa3b3 Mon Sep 17 00:00:00 2001 From: Bolin Chen Date: Sat, 5 Sep 2026 19:22:04 +0000 Subject: [PATCH] fix(history): pair a frozen save snapshot with its prefix boundary The rewind boundary transaction freezes a window snapshot on the event loop and then awaits the save in a worker thread, where _save_slot_to_history read the LIVE _disk_older_count. A trim at the window cap in that gap credits the trimmed rows to the frozen prefix while the snapshot still holds them at its head, so the rewrite emitted them twice -- a silently duplicated transcript, reachable on origin/main today. The endpoint now passes the boundary it observed in the same synchronous stretch as the snapshot. The save refuses on drift (nothing written, the retryable 503 the transaction already contracts for) and the commit re-adopts that boundary, because the commit puts the pre-trim window prefix back on the live window. The durable position base (_disk_older_durable_count) is re-adopted with it: the trim advances the two together, and a row counted as having left the window front while it is back inside it refuses a valid absolute cursor or repeats rows. The post-write witnesses are gated on the routing for the same reason: they describe the file just written but live on the slot, which the loop can rebind mid-write. Stamping then cleared a _pending_rewrite the new transcript still owed and claimed its unsaved rows as persisted, which a later flush believed. Every witness left at its pre-save value is the conservative reading of that witness. Split out of the edit-resend port (#8431 carries the endpoint); both defects live in the shared save and are reachable on rewind without it. --- docs/system-specs/modules/history.md | 54 +++++++ src/kiro_crew/dashboard/chat_persistence.py | 131 +++++++++++---- src/kiro_crew/dashboard/chat_rewind.py | 67 ++++++-- test/test_dashboard_chat_rewind.py | 168 +++++++++++++++++++- 4 files changed, 377 insertions(+), 43 deletions(-) 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.