From 7cdc372d0b6c6ff6deac396d60d8a93da71e0c5c Mon Sep 17 00:00:00 2001 From: happy5318 Date: Sun, 20 Sep 2026 22:15:07 +0800 Subject: [PATCH 01/10] fix(models): suppress recovery append when transcript has already advanced past the pending turn (#6366) Stale-cancel recovery in _apply_core_sync_or_error_marker (line ~3497+) only inspects the transcript tail (`session.messages[-1]`) when checking whether the durable transcript already contains the pending turn. When the pending turn's user row is older than the tail, the tail is a *newer* settled assistant row that can never match the pending **user** checkpoint, and the recovery path falls through into the append branch. The cancel/recovery tail then appends, after the valid final answer, a recovered user row, a `_partial` clone of the old journal, and a generic no-response error that survives reload in the durable transcript. The fix adds `_transcript_already_advanced_past_pending()` which scans the messages for any strictly-matching user row (content + timestamp + source + attachments) and, when one is found, looks for a later settled assistant row (no `_partial` / `_error` / `_recovered`). When that is the case, the helper returns True and the non-completed-journal branch clears only the stale pending fields and returns without appending any rows. Strict match is required so that a *new* user turn that happens to repeat the same prompt text (e.g. user re-sends the same question) is not misclassified as a stale advance. The regression guard is the existing `test_repeated_pending_prompt_uses_current_checkpoint_identity` which now passes alongside the new helper unit cases. The helper is a pure read; running recovery twice therefore produces the same outcome, satisfying the maintainer's idempotence requirement. Validation: 11/11 new tests pass (covers the strict-match distinction, the partial/error/recovered skip, and idempotence); 124/124 related stale-pending / sidecar-repair / recovery tests pass; py_compile clean. --- api/models.py | 95 +++++++ tests/test_issue6366_stale_cancel_recovery.py | 237 ++++++++++++++++++ 2 files changed, 332 insertions(+) create mode 100644 tests/test_issue6366_stale_cancel_recovery.py diff --git a/api/models.py b/api/models.py index 84a91b9c..de21e3d9 100644 --- a/api/models.py +++ b/api/models.py @@ -2438,6 +2438,78 @@ def _latest_user_matches_pending_text(messages, pending_text): return False +def _transcript_already_advanced_past_pending(session) -> bool: + """#6366: a stale pending user message whose content already appears + in the durable transcript as a user row, followed by a newer settled + assistant row, means the transcript has already advanced past this + pending turn. Returning True here lets the recovery path skip the + recovered user / ``_partial`` clone / journal replay / generic + no-response error that would otherwise append after the valid + final answer. The tail-only check at line ~3485 (``session.messages[-1]``) + misses this case: when the pending turn's user row is older than + the transcript tail, the tail is a *newer* assistant row that + can never match the pending **user** checkpoint, and the recovery + path falls through into the append branch. + + Strict match: uses ``_message_matches_pending_checkpoint`` (content + + timestamp + source + attachments) rather than the looser text- + only matcher, so a *new* user turn that happens to repeat the + same prompt text (e.g. user re-sends the same question) is NOT + misclassified as a stale advance. That distinction is the + regression guard for the repeated-prompt identity cases covered + by ``test_repeated_pending_prompt_uses_current_checkpoint_identity``. + + Idempotent: this is a pure read that does not mutate the session. + Running recovery twice therefore produces the same outcome. + """ + pending_text = getattr(session, 'pending_user_message', None) + if not pending_text: + return False + pending_started_at = getattr(session, 'pending_started_at', None) + pending_source = getattr(session, 'pending_user_source', None) + pending_attachments = getattr(session, 'pending_attachments', None) + messages = getattr(session, 'messages', None) + if not isinstance(messages, list): + return False + for idx, message in enumerate(messages): + if not isinstance(message, dict) or message.get('role') != 'user': + continue + if not _message_matches_pending_checkpoint( + message, + pending_text, + pending_started_at, + pending_source, + pending_attachments, + ): + continue + # Found a strictly-matching user row at ``idx``. Walk + # forward and look for a settled assistant row (not + # ``_partial``, not ``_error``, not the recovery marker). + # Any such row means the transcript has already advanced + # past this pending turn into a newer settled + # user/assistant boundary, and recovery must not append + # duplicate rows after it. + for later in messages[idx + 1:]: + if not isinstance(later, dict): + continue + if later.get('role') != 'assistant': + continue + if later.get('_partial') or later.get('_error'): + continue + if later.get('_recovered'): + # A previously-recovered assistant row (from an + # earlier recovery pass that may have appended the + # very row we are trying to avoid) is not a + # settled answer either. Skip and keep looking. + continue + return True + # Matched user row but no later settled assistant row → + # the transcript has not advanced past this pending turn. + # Recovery is the legitimate path. + return False + return False + + def _partial_message_signature(message: dict) -> tuple: """Return a stable identity for partial assistant markers recovered on load.""" if not isinstance(message, dict): @@ -3515,6 +3587,29 @@ def _apply_core_sync_or_error_marker( ) return True if not _tail_user_already_checkpointed: + # #6366 re-gate: when the durable transcript has already + # advanced past this pending turn into a newer settled + # user/assistant boundary, the recovery path must NOT + # append a recovered user row + ``_partial`` clone + + # journal replay + generic no-response error after the + # valid final answer. The tail-only check above misses + # that case (the tail is a newer assistant row that can + # never match the pending user checkpoint). Clear only + # the stale pending fields and return; the transcript is + # already correct and durable. + if _transcript_already_advanced_past_pending(session): + session.active_stream_id = None + session.pending_user_message = None + session.pending_attachments = [] + session.pending_started_at = None + session.pending_user_source = None + session.save(touch_updated_at=touch_updated_at) + logger.info( + "Session %s: cleared stale pending state for stream %s — transcript already advanced past this turn", + sid, + _stream_id, + ) + return True _append_recovered_pending_turn(session, timestamp=_recovered_ts) else: recovered = { diff --git a/tests/test_issue6366_stale_cancel_recovery.py b/tests/test_issue6366_stale_cancel_recovery.py new file mode 100644 index 00000000..30db9c40 --- /dev/null +++ b/tests/test_issue6366_stale_cancel_recovery.py @@ -0,0 +1,237 @@ +"""Tests for #6366: stale-cancel recovery must not append duplicate +rows when the durable transcript has already advanced past the +pending turn. + +Bug shape: the recovery path at ``api/models.py::_apply_core_sync_or_error_marker`` +checks only ``session.messages[-1]`` against the pending user +checkpoint. When a stale pending turn's user row is older than the +transcript tail, the tail is a *newer* settled assistant row that +can never match the pending **user** checkpoint, and the recovery +path falls through into the append branch. It then appends a +recovered user row, a ``_partial`` clone of the old journal, and a +generic ``No response from provider`` error — all after the valid +final answer. The duplicates survive reload. + +Fix: add a transcript-advance detector that scans the durable +messages for a matching user row at any index and, when found, +checks for a later settled assistant row. If one exists, the +recovery path must clear the stale pending fields without +appending any rows. The helper is a pure read so recovery is +idempotent: running it twice produces the same outcome. +""" + +import pytest + +import api.models as models +from api.models import ( + _transcript_already_advanced_past_pending, +) + + +class _FakeSession: + """Minimal stand-in for api.models.Session — only the fields the + new helper reads.""" + + def __init__(self, pending=None, messages=None): + self.pending_user_message = pending + self.messages = list(messages or []) + + +# ── _transcript_already_advanced_past_pending unit cases ─────────────── + + +def test_advances_past_pending_when_user_matched_and_settled_assistant_follows(): + """Canonical bug shape: a stale pending turn's user row sits at + an older index, and a newer settled assistant row follows. + Recovery must not append duplicate rows after the assistant. + Strict match: the existing user row must match the pending + checkpoint (content + timestamp + source + attachments) so + that a *new* user turn repeating the same text is not + misclassified as a stale advance.""" + s = _FakeSession( + pending="hello world", + messages=[ + {"role": "user", "content": "hello world", "timestamp": 222, "_source": "webui", "attachments": [{"name": "a.png"}]}, + {"role": "assistant", "content": "answer", "timestamp": 223}, + ], + ) + # Plumb the full checkpoint identity so the strict match hits. + s.pending_started_at = 222 + s.pending_user_source = "webui" + s.pending_attachments = [{"name": "a.png"}] + assert _transcript_already_advanced_past_pending(s) is True + + +def test_advances_past_pending_ignores_partial_clone_after_user(): + """A ``_partial`` clone after the matching user row does NOT count + as a settled advance — the canonical bug shape is the partial + clone followed by a generic no-response error, both of which + the previous recovery path appended. Only a non-partial, + non-error assistant row represents a real advance.""" + s = _FakeSession( + pending="hello world", + messages=[ + {"role": "user", "content": "hello world", "timestamp": 1}, + {"role": "assistant", "content": "answer", "timestamp": 2, "_partial": True}, + {"role": "assistant", "content": "no response", "timestamp": 3, "_error": True}, + ], + ) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_advances_past_pending_ignores_recovered_assistant_row(): + """A previously-recovered assistant row is not a settled answer + either; the recovery path may have produced it on an earlier + pass. Skip it and keep looking forward.""" + s = _FakeSession( + pending="hello world", + messages=[ + {"role": "user", "content": "hello world", "timestamp": 1}, + {"role": "assistant", "content": "answer", "timestamp": 2, "_recovered": True}, + ], + ) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_does_not_advance_when_only_user_row_matches(): + """The matching user row exists but no later assistant row has + been settled. Recovery is the legitimate path; the helper + returns False.""" + s = _FakeSession( + pending="hello world", + messages=[ + {"role": "user", "content": "hello world", "timestamp": 1}, + ], + ) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_does_not_advance_when_no_user_row_matches(): + """No transcript row carries the pending text. The bug shape + here is the older non-matching rows + a stale pending. The + recovery path is legitimate; the helper returns False.""" + s = _FakeSession( + pending="hello world", + messages=[ + {"role": "user", "content": "other text", "timestamp": 1}, + {"role": "assistant", "content": "answer", "timestamp": 2}, + ], + ) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_does_not_advance_when_no_pending_text(): + """An empty pending text means there is nothing to compare + against. The helper returns False without scanning the + transcript.""" + s = _FakeSession( + pending=None, + messages=[ + {"role": "assistant", "content": "answer", "timestamp": 2}, + ], + ) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_does_not_advance_when_no_messages(): + """An empty transcript is the cold-start case; recovery is the + legitimate path. The helper returns False.""" + s = _FakeSession(pending="hello world", messages=[]) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_advances_past_pending_with_unrelated_intervening_rows(): + """A matching user row followed by an unrelated tool row and + THEN a settled assistant row counts as a real advance. The + helper scans past non-assistant roles correctly. Strict + match on the user row.""" + s = _FakeSession( + pending="hello world", + messages=[ + {"role": "user", "content": "hello world", "timestamp": 222, "_source": "webui", "attachments": []}, + {"role": "tool", "content": "tool output"}, + {"role": "assistant", "content": "answer", "timestamp": 223}, + ], + ) + s.pending_started_at = 222 + s.pending_user_source = "webui" + s.pending_attachments = [] + assert _transcript_already_advanced_past_pending(s) is True + + +def test_helper_is_pure_read_idempotent(): + """Idempotence: running the helper twice on the same session + state returns the same result. The helper does not mutate the + session, so recovery is automatically safe to run twice.""" + s = _FakeSession( + pending="hello world", + messages=[ + {"role": "user", "content": "hello world", "timestamp": 222, "_source": "webui", "attachments": []}, + {"role": "assistant", "content": "answer", "timestamp": 223}, + ], + ) + s.pending_started_at = 222 + s.pending_user_source = "webui" + s.pending_attachments = [] + first = _transcript_already_advanced_past_pending(s) + second = _transcript_already_advanced_past_pending(s) + assert first == second is True + # And the session state was not mutated. + assert s.pending_user_message == "hello world" + assert len(s.messages) == 2 + + + +def test_does_not_advance_when_repeated_prompt_has_different_checkpoint(): + """#6366 regression guard for repeated-prompt identity: a *new* + user turn that happens to repeat the same prompt text as an + older transcript row is NOT a stale advance. The strict match + uses timestamp + source + attachments to tell a stale duplicate + from a legitimate re-send.""" + s = _FakeSession( + pending="repeat this", + messages=[ + # Older user row, same text but different identity. + { + "role": "user", + "content": "repeat this", + "timestamp": 111, + "_source": "webui", + "attachments": [{"name": "old.png"}], + }, + {"role": "assistant", "content": "Earlier answer"}, + ], + ) + # Pending checkpoint is a fresh turn: different timestamp and + # different attachments. + s.pending_started_at = 222 + s.pending_user_source = "webui" + s.pending_attachments = [{"name": "current.png"}] + # Strict match fails (timestamp + attachments differ), so the + # helper returns False and recovery is the legitimate path. + assert _transcript_already_advanced_past_pending(s) is False + + +def test_advances_past_pending_with_exact_checkpoint_match(): + """The strict-match path: a transcript user row that exactly + matches the pending checkpoint (content + timestamp + + source + attachments) followed by a settled assistant row is + the canonical #6366 bug shape and must suppress the recovery + append.""" + s = _FakeSession( + pending="hello world", + messages=[ + { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_source": "webui", + "attachments": [{"name": "a.png"}], + }, + {"role": "assistant", "content": "answer", "timestamp": 223}, + ], + ) + s.pending_started_at = 222 + s.pending_user_source = "webui" + s.pending_attachments = [{"name": "a.png"}] + assert _transcript_already_advanced_past_pending(s) is True From 1aa5d778a69bd2427079692054da6e9bd60b1950 Mon Sep 17 00:00:00 2001 From: happy5318 Date: Mon, 21 Sep 2026 13:47:50 +0800 Subject: [PATCH 02/10] chore(tests): drop unused `pytest` and `api.models as models` imports ruff F401 flagged both imports; the test references ``_transcript_already_advanced_past_pending`` directly via the ``from api.models import (...)`` form below. Dropped the unused ``import pytest`` and the aliased module import. Files: tests/test_issue6366_stale_cancel_recovery.py:23,25 Model: MiniMax-M3 --- tests/test_issue6366_stale_cancel_recovery.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/tests/test_issue6366_stale_cancel_recovery.py b/tests/test_issue6366_stale_cancel_recovery.py index 30db9c40..1a2d3df9 100644 --- a/tests/test_issue6366_stale_cancel_recovery.py +++ b/tests/test_issue6366_stale_cancel_recovery.py @@ -20,9 +20,6 @@ appending any rows. The helper is a pure read so recovery is idempotent: running it twice produces the same outcome. """ -import pytest - -import api.models as models from api.models import ( _transcript_already_advanced_past_pending, ) From 54f7d97d57c6ce8abd83a4011db15401042ce09c Mon Sep 17 00:00:00 2001 From: happy5318 Date: Mon, 21 Sep 2026 22:41:55 +0800 Subject: [PATCH 03/10] fix(models): bind stale-cancel suppression to turn identity and final answers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The #6366 suppression was too broad and could destroy data in two ways, both verified with full `_apply_core_sync_or_error_marker()` probes: 1. A repeated prompt could be silently discarded. Checkpoint matching truncates timestamps to integer seconds and ignored `_active_turn_token`, so a prior identical prompt from a *different* stream could match the new pending turn — and the suppression branch then cleared the only durable copy of the prompt, with no recovered row and no error marker. Identity is now bound to the pending stream's exact active-turn token (`_transcript_user_row_is_pending_turn`). Integer-second timestamp equality never overrides a token mismatch, and a row that carries no token cannot be proven to be this turn. Legacy sessions with no token on either side fall back to the strict content + timestamp + source + attachments checkpoint. 2. Interrupted tool-running turns were treated as completed. Any unflagged assistant row counted as settled, including empty rows, tool-call-only rows, interim-progress rows and compaction rows, so `:3600` bypassed journal replay and the interruption marker. Completion now reuses the established final-answer semantics (`_pending_turn_has_final_assistant_answer` -> `_assistant_message_has_final_visible_text`) inside the matched user-turn boundary, rejecting tool-only, empty, interim, journal- recovered and compaction tails. The suppression is deliberately asymmetric: a false negative leaves a visible cosmetic duplicate the user can ignore, while a false positive silently destroys a prompt or a response. Whenever identity or completion cannot be positively proven, recovery runs instead of suppression. Tests: the 11 predicate unit cases are re-derived against the new identity/completion helpers, plus full-recovery-path cases that exercise `_apply_core_sync_or_error_marker()` itself — distinct same-second repeated turns (token-bearing and legacy) proving the prompt survives, and empty / tool-call-only / interim / compaction tails proving journal replay and the interruption marker still run — plus a positive control for the suppression branch itself. --- api/models.py | 179 ++-- tests/test_issue6366_stale_cancel_recovery.py | 785 ++++++++++++++++-- 2 files changed, 853 insertions(+), 111 deletions(-) diff --git a/api/models.py b/api/models.py index de21e3d9..ee1ae53f 100644 --- a/api/models.py +++ b/api/models.py @@ -2438,26 +2438,136 @@ def _latest_user_matches_pending_text(messages, pending_text): return False +def _pending_active_turn_token(session): + """Return the exact WebUI active-turn token for the pending turn. + + The token is ``":"`` — the same value the + eager user-message checkpoint and the agent-result merge stamp onto the + materialized user row (``stamp_message_source``). It is unforgeable + per-turn identity: two distinct streams that submit the same prompt + inside the same second produce different tokens. + """ + from api.process_event_utils import build_active_turn_token + + return build_active_turn_token( + getattr(session, 'active_stream_id', None), + getattr(session, 'pending_started_at', None), + ) + + +def _transcript_user_row_is_pending_turn(message, session, pending_token) -> bool: + """Return True only when a transcript user row provably IS the pending turn. + + Identity is bound to the active-turn token, never to integer-second + timestamp equality: two different streams that send the same prompt + within one second truncate to the same ``int(timestamp)``, so the + checkpoint matcher alone would accept the *other* stream's row as the + pending turn. A token mismatch therefore always loses, and a row that + carries no token at all cannot be proven to be this turn — which means + the caller must recover, not suppress. + """ + if not isinstance(message, dict) or message.get('role') != 'user': + return False + row_token = message.get('_active_turn_token') + if pending_token: + if not row_token: + # Ambiguous legacy identity: the pending turn has a resolvable + # token but this row predates token stamping. We cannot prove it + # is the same turn, so recovery must run. + return False + return row_token == pending_token + if row_token: + # The row belongs to a different, token-bearing turn. Never let a + # same-second timestamp collision override that mismatch. + return False + # Neither side carries a token (legacy/dead-stream pending state): fall + # back to the strict content + timestamp + source + attachments + # checkpoint, which is the only identity signal available. + return _message_matches_pending_checkpoint( + message, + getattr(session, 'pending_user_message', None), + getattr(session, 'pending_started_at', None), + getattr(session, 'pending_user_source', None), + getattr(session, 'pending_attachments', None), + ) + + +def _pending_turn_has_final_assistant_answer(messages, start_idx: int) -> bool: + """Return True when the matched user turn ends in a genuine final answer. + + Reuses the established final-answer semantics + (``_assistant_message_has_final_visible_text``) instead of "an assistant + row exists". An assistant row only settles the turn when it carries real + visible answer text: empty rows, tool-call-only rows, interim + ``_partial`` progress rows, rows recovered from the run journal and + compaction reference cards are all rejected, so a turn that was + interrupted mid-tool-execution is never mistaken for a completed one. + + The scan stops at the next real user row — a final answer that belongs + to a *later* turn must not settle this pending turn. + """ + from api.streaming import ( + _assistant_message_has_final_visible_text, + _is_synthetic_control_message, + ) + + for later in messages[start_idx + 1:]: + if not isinstance(later, dict): + continue + if _is_synthetic_control_message(later): + continue + role = later.get('role') + if role == 'user': + if is_context_compression_marker(later): + # Synthetic compaction cards are not a user-turn boundary. + continue + return False + if role != 'assistant': + # Tool rows never prove a final answer. + continue + if ( + later.get('_partial') + or later.get('_error') + or later.get('_recovered') + or later.get('_recovered_from_run_journal') + ): + continue + if is_context_compression_marker(later): + continue + if _assistant_message_has_final_visible_text(later): + return True + return False + + def _transcript_already_advanced_past_pending(session) -> bool: - """#6366: a stale pending user message whose content already appears - in the durable transcript as a user row, followed by a newer settled - assistant row, means the transcript has already advanced past this - pending turn. Returning True here lets the recovery path skip the - recovered user / ``_partial`` clone / journal replay / generic - no-response error that would otherwise append after the valid + """#6366: a stale pending user message whose own user row already appears + in the durable transcript, followed by a genuine final assistant answer + inside that same user-turn boundary, means the transcript has already + advanced past this pending turn. Returning True here lets the recovery + path skip the recovered user / ``_partial`` clone / journal replay / + generic no-response error that would otherwise append after the valid final answer. The tail-only check at line ~3485 (``session.messages[-1]``) misses this case: when the pending turn's user row is older than the transcript tail, the tail is a *newer* assistant row that can never match the pending **user** checkpoint, and the recovery path falls through into the append branch. - Strict match: uses ``_message_matches_pending_checkpoint`` (content - + timestamp + source + attachments) rather than the looser text- - only matcher, so a *new* user turn that happens to repeat the - same prompt text (e.g. user re-sends the same question) is NOT - misclassified as a stale advance. That distinction is the - regression guard for the repeated-prompt identity cases covered - by ``test_repeated_pending_prompt_uses_current_checkpoint_identity``. + Identity is bound to the pending stream's exact active-turn token + (``_transcript_user_row_is_pending_turn``), so a repeated prompt from a + different stream — even one submitted in the same integer second — + can never be matched against this pending turn and silently dropped. + + Completion requires a genuine final visible assistant answer + (``_pending_turn_has_final_assistant_answer``): empty, tool-call-only, + interim ``_partial``, journal-recovered and compaction tails all keep + recovery active, because that is precisely the interrupted-turn shape + the recovery path exists for. + + The suppression is therefore deliberately conservative: whenever the + turn's identity or its completion cannot be positively proven, this + helper returns False and the prompt is recovered instead of discarded. + A false negative leaves a visible cosmetic duplicate; a false positive + silently destroys a prompt or a response. Idempotent: this is a pure read that does not mutate the session. Running recovery twice therefore produces the same outcome. @@ -2465,48 +2575,17 @@ def _transcript_already_advanced_past_pending(session) -> bool: pending_text = getattr(session, 'pending_user_message', None) if not pending_text: return False - pending_started_at = getattr(session, 'pending_started_at', None) - pending_source = getattr(session, 'pending_user_source', None) - pending_attachments = getattr(session, 'pending_attachments', None) messages = getattr(session, 'messages', None) if not isinstance(messages, list): return False + pending_token = _pending_active_turn_token(session) for idx, message in enumerate(messages): - if not isinstance(message, dict) or message.get('role') != 'user': + if not _transcript_user_row_is_pending_turn(message, session, pending_token): continue - if not _message_matches_pending_checkpoint( - message, - pending_text, - pending_started_at, - pending_source, - pending_attachments, - ): - continue - # Found a strictly-matching user row at ``idx``. Walk - # forward and look for a settled assistant row (not - # ``_partial``, not ``_error``, not the recovery marker). - # Any such row means the transcript has already advanced - # past this pending turn into a newer settled - # user/assistant boundary, and recovery must not append - # duplicate rows after it. - for later in messages[idx + 1:]: - if not isinstance(later, dict): - continue - if later.get('role') != 'assistant': - continue - if later.get('_partial') or later.get('_error'): - continue - if later.get('_recovered'): - # A previously-recovered assistant row (from an - # earlier recovery pass that may have appended the - # very row we are trying to avoid) is not a - # settled answer either. Skip and keep looking. - continue - return True - # Matched user row but no later settled assistant row → - # the transcript has not advanced past this pending turn. - # Recovery is the legitimate path. - return False + # Found the pending turn's own user row at ``idx``. Only a genuine + # final answer inside that turn's boundary proves the transcript + # advanced past it. + return _pending_turn_has_final_assistant_answer(messages, idx) return False diff --git a/tests/test_issue6366_stale_cancel_recovery.py b/tests/test_issue6366_stale_cancel_recovery.py index 1a2d3df9..6b35c47c 100644 --- a/tests/test_issue6366_stale_cancel_recovery.py +++ b/tests/test_issue6366_stale_cancel_recovery.py @@ -1,8 +1,9 @@ """Tests for #6366: stale-cancel recovery must not append duplicate rows when the durable transcript has already advanced past the -pending turn. +pending turn — and must never silently discard a distinct repeated +prompt or an interrupted tool-running turn. -Bug shape: the recovery path at ``api/models.py::_apply_core_sync_or_error_marker`` +Bug shape: the recovery path at ``api.models.py::_apply_core_sync_or_error_marker`` checks only ``session.messages[-1]`` against the pending user checkpoint. When a stale pending turn's user row is older than the transcript tail, the tail is a *newer* settled assistant row that @@ -13,25 +14,282 @@ generic ``No response from provider`` error — all after the valid final answer. The duplicates survive reload. Fix: add a transcript-advance detector that scans the durable -messages for a matching user row at any index and, when found, -checks for a later settled assistant row. If one exists, the -recovery path must clear the stale pending fields without -appending any rows. The helper is a pure read so recovery is -idempotent: running it twice produces the same outcome. +messages for the pending turn's own user row at any index and, when +found, checks for a genuine final assistant answer inside that turn's +boundary. If one exists, the recovery path must clear the stale +pending fields without appending any rows. The helper is a pure read +so recovery is idempotent: running it twice produces the same +outcome. + +Two guard rails keep the suppression from destroying data: + +* **Identity** is bound to the pending stream's exact active-turn + token, never to integer-second timestamp equality. Two different + streams that send the same prompt within one second truncate to the + same second, so the checkpoint matcher alone would match the other + stream's row and the branch would then clear the only durable copy + of the prompt. A token mismatch — or a token-less legacy row — + always loses, so an unprovable identity recovers instead of + suppressing. + +* **Completion** reuses the established final-answer semantics + (``_assistant_message_has_final_visible_text``) rather than "an + assistant row exists". Empty, tool-call-only, interim ``_partial`` + and compaction tails all keep recovery active, so a turn that was + interrupted mid-tool-execution is replayed and marked instead of + being treated as already advanced. + +Because a false negative leaves a visible cosmetic duplicate while a +false positive silently destroys a prompt, the suppression defaults to +recovering whenever identity or completion cannot be proven. The +full-recovery cases below exercise ``_apply_core_sync_or_error_marker`` +itself, because neither guard rail is reachable through the predicate +in isolation — the suppression decision only takes effect on the real +recovery path. """ +import json + +import pytest + +import api.models as models from api.models import ( + Session, + _apply_core_sync_or_error_marker, + _pending_turn_has_final_assistant_answer, _transcript_already_advanced_past_pending, + _transcript_user_row_is_pending_turn, ) +from api.run_journal import append_run_event + + +@pytest.fixture(autouse=True) +def _isolate_session_state(tmp_path, monkeypatch): + session_dir = tmp_path / "sessions" + session_dir.mkdir() + monkeypatch.setattr(models, "SESSION_DIR", session_dir) + monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") + models.SESSIONS.clear() + yield + models.SESSIONS.clear() class _FakeSession: """Minimal stand-in for api.models.Session — only the fields the - new helper reads.""" + new helpers read.""" - def __init__(self, pending=None, messages=None): + def __init__( + self, + pending=None, + messages=None, + *, + stream_id=None, + started_at=None, + source=None, + attachments=None, + ): self.pending_user_message = pending self.messages = list(messages or []) + self.active_stream_id = stream_id + self.pending_started_at = started_at + self.pending_user_source = source + self.pending_attachments = attachments + + +def _token(stream_id, started_at): + return f"{stream_id}:{float(started_at):.17g}" + + +# ── _transcript_user_row_is_pending_turn identity cases ───────────────── + + +def test_user_row_matches_only_with_identical_active_turn_token(): + """Identity is unforgeable: the row's ``_active_turn_token`` must equal + the token derived from the pending stream id + started_at.""" + s = _FakeSession( + pending="hello world", + stream_id="stream-a", + started_at=222.5, + ) + own_row = { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_active_turn_token": "stream-a:222.5", + } + other_row = { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_active_turn_token": "stream-b:222.5", + } + assert _transcript_user_row_is_pending_turn(own_row, s, _token("stream-a", 222.5)) is True + assert _transcript_user_row_is_pending_turn(other_row, s, _token("stream-a", 222.5)) is False + + +def test_user_row_without_token_never_matches_a_token_bearing_pending_turn(): + """A legacy row that predates token stamping cannot be proven to be the + pending turn, so the caller must recover rather than suppress.""" + s = _FakeSession( + pending="hello world", + stream_id="stream-a", + started_at=222.5, + ) + legacy_row = { + "role": "user", + "content": "hello world", + "timestamp": 222, + } + assert _transcript_user_row_is_pending_turn(legacy_row, s, _token("stream-a", 222.5)) is False + + +def test_token_bearing_row_never_matches_a_token_less_pending_turn(): + """The mirror case: a token-bearing row belongs to a real turn, so it + must not be claimed by a pending turn that has no resolvable token.""" + s = _FakeSession( + pending="hello world", + stream_id=None, + started_at=None, + ) + token_row = { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_active_turn_token": "stream-a:222.5", + } + assert _transcript_user_row_is_pending_turn(token_row, s, None) is False + + +def test_legacy_pending_turn_falls_back_to_strict_checkpoint_match(): + """With no token on either side the strict content + timestamp + source + + attachments checkpoint is the only identity signal, and it still + distinguishes a same-second repeated prompt from the pending turn.""" + s = _FakeSession( + pending="hello world", + stream_id=None, + started_at=222, + source="webui", + attachments=[{"name": "a.png"}], + ) + same_turn_row = { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_source": "webui", + "attachments": [{"name": "a.png"}], + } + other_turn_row = { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_source": "webui", + "attachments": [{"name": "old.png"}], + } + assert _transcript_user_row_is_pending_turn(same_turn_row, s, None) is True + assert _transcript_user_row_is_pending_turn(other_turn_row, s, None) is False + + +# ── _pending_turn_has_final_assistant_answer completion cases ─────────── + + +def test_final_answer_requires_visible_answer_text(): + assert _pending_turn_has_final_assistant_answer( + [ + {"role": "user", "content": "q"}, + {"role": "assistant", "content": "Here is the final answer."}, + ], + 0, + ) is True + + +def test_final_answer_rejects_empty_assistant_row(): + assert _pending_turn_has_final_assistant_answer( + [ + {"role": "user", "content": "q"}, + {"role": "assistant", "content": ""}, + ], + 0, + ) is False + + +def test_final_answer_rejects_tool_call_only_row(): + assert _pending_turn_has_final_assistant_answer( + [ + {"role": "user", "content": "q"}, + { + "role": "assistant", + "content": "I found a source and will inspect it.", + "tool_calls": [{"id": "call_1"}], + }, + {"role": "tool", "tool_call_id": "call_1", "content": "result"}, + ], + 0, + ) is False + + +def test_final_answer_rejects_interim_partial_row(): + assert _pending_turn_has_final_assistant_answer( + [ + {"role": "user", "content": "q"}, + { + "role": "assistant", + "content": "Still working on it…", + "_partial": True, + }, + ], + 0, + ) is False + + +def test_final_answer_rejects_compaction_marker_row(): + assert _pending_turn_has_final_assistant_answer( + [ + {"role": "user", "content": "q"}, + { + "role": "assistant", + "content": "[CONTEXT COMPACTION — REFERENCE ONLY] summary", + }, + ], + 0, + ) is False + + +def test_final_answer_accepts_answer_after_compaction_marker(): + assert _pending_turn_has_final_assistant_answer( + [ + {"role": "user", "content": "q"}, + { + "role": "assistant", + "content": "[CONTEXT COMPACTION — REFERENCE ONLY] summary", + }, + {"role": "assistant", "content": "Here is the final answer."}, + ], + 0, + ) is True + + +def test_final_answer_does_not_leak_across_a_later_user_turn(): + """A final answer belonging to a *later* turn must not settle this one.""" + assert _pending_turn_has_final_assistant_answer( + [ + {"role": "user", "content": "q"}, + {"role": "user", "content": "later question"}, + {"role": "assistant", "content": "Here is the final answer."}, + ], + 0, + ) is False + + +def test_final_answer_tolerates_tool_rows_before_the_answer(): + assert _pending_turn_has_final_assistant_answer( + [ + {"role": "user", "content": "q"}, + {"role": "assistant", "content": "", "tool_calls": [{"id": "call_1"}]}, + {"role": "tool", "tool_call_id": "call_1", "content": "result"}, + {"role": "assistant", "content": "Here is the final answer."}, + ], + 0, + ) is True # ── _transcript_already_advanced_past_pending unit cases ─────────────── @@ -41,37 +299,48 @@ def test_advances_past_pending_when_user_matched_and_settled_assistant_follows() """Canonical bug shape: a stale pending turn's user row sits at an older index, and a newer settled assistant row follows. Recovery must not append duplicate rows after the assistant. - Strict match: the existing user row must match the pending - checkpoint (content + timestamp + source + attachments) so - that a *new* user turn repeating the same text is not - misclassified as a stale advance.""" + Identity comes from the pending stream's own active-turn token.""" s = _FakeSession( pending="hello world", messages=[ - {"role": "user", "content": "hello world", "timestamp": 222, "_source": "webui", "attachments": [{"name": "a.png"}]}, + { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_source": "webui", + "attachments": [{"name": "a.png"}], + "_active_turn_token": "stream-a:222.5", + }, {"role": "assistant", "content": "answer", "timestamp": 223}, ], + stream_id="stream-a", + started_at=222.5, + source="webui", + attachments=[{"name": "a.png"}], ) - # Plumb the full checkpoint identity so the strict match hits. - s.pending_started_at = 222 - s.pending_user_source = "webui" - s.pending_attachments = [{"name": "a.png"}] assert _transcript_already_advanced_past_pending(s) is True def test_advances_past_pending_ignores_partial_clone_after_user(): - """A ``_partial`` clone after the matching user row does NOT count + """A ``_partial`` clone after the pending turn's user row does NOT count as a settled advance — the canonical bug shape is the partial clone followed by a generic no-response error, both of which - the previous recovery path appended. Only a non-partial, - non-error assistant row represents a real advance.""" + the previous recovery path appended. Only a genuine final visible + answer represents a real advance.""" s = _FakeSession( pending="hello world", messages=[ - {"role": "user", "content": "hello world", "timestamp": 1}, + { + "role": "user", + "content": "hello world", + "timestamp": 1, + "_active_turn_token": "stream-a:1.5", + }, {"role": "assistant", "content": "answer", "timestamp": 2, "_partial": True}, {"role": "assistant", "content": "no response", "timestamp": 3, "_error": True}, ], + stream_id="stream-a", + started_at=1.5, ) assert _transcript_already_advanced_past_pending(s) is False @@ -83,36 +352,57 @@ def test_advances_past_pending_ignores_recovered_assistant_row(): s = _FakeSession( pending="hello world", messages=[ - {"role": "user", "content": "hello world", "timestamp": 1}, + { + "role": "user", + "content": "hello world", + "timestamp": 1, + "_active_turn_token": "stream-a:1.5", + }, {"role": "assistant", "content": "answer", "timestamp": 2, "_recovered": True}, ], + stream_id="stream-a", + started_at=1.5, ) assert _transcript_already_advanced_past_pending(s) is False def test_does_not_advance_when_only_user_row_matches(): - """The matching user row exists but no later assistant row has + """The pending turn's user row exists but no final assistant answer has been settled. Recovery is the legitimate path; the helper returns False.""" s = _FakeSession( pending="hello world", messages=[ - {"role": "user", "content": "hello world", "timestamp": 1}, + { + "role": "user", + "content": "hello world", + "timestamp": 1, + "_active_turn_token": "stream-a:1.5", + }, ], + stream_id="stream-a", + started_at=1.5, ) assert _transcript_already_advanced_past_pending(s) is False def test_does_not_advance_when_no_user_row_matches(): - """No transcript row carries the pending text. The bug shape + """No transcript row carries the pending turn's token. The bug shape here is the older non-matching rows + a stale pending. The recovery path is legitimate; the helper returns False.""" s = _FakeSession( pending="hello world", messages=[ - {"role": "user", "content": "other text", "timestamp": 1}, + { + "role": "user", + "content": "hello world", + "timestamp": 1, + "_active_turn_token": "stream-z:1.5", + }, {"role": "assistant", "content": "answer", "timestamp": 2}, ], + stream_id="stream-a", + started_at=1.5, ) assert _transcript_already_advanced_past_pending(s) is False @@ -138,21 +428,28 @@ def test_does_not_advance_when_no_messages(): def test_advances_past_pending_with_unrelated_intervening_rows(): - """A matching user row followed by an unrelated tool row and - THEN a settled assistant row counts as a real advance. The - helper scans past non-assistant roles correctly. Strict - match on the user row.""" + """The pending turn's user row followed by a tool row and THEN a final + answer counts as a real advance. The helper scans past non-assistant + roles correctly.""" s = _FakeSession( pending="hello world", messages=[ - {"role": "user", "content": "hello world", "timestamp": 222, "_source": "webui", "attachments": []}, + { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_source": "webui", + "attachments": [], + "_active_turn_token": "stream-a:222.5", + }, {"role": "tool", "content": "tool output"}, {"role": "assistant", "content": "answer", "timestamp": 223}, ], + stream_id="stream-a", + started_at=222.5, + source="webui", + attachments=[], ) - s.pending_started_at = 222 - s.pending_user_source = "webui" - s.pending_attachments = [] assert _transcript_already_advanced_past_pending(s) is True @@ -163,13 +460,21 @@ def test_helper_is_pure_read_idempotent(): s = _FakeSession( pending="hello world", messages=[ - {"role": "user", "content": "hello world", "timestamp": 222, "_source": "webui", "attachments": []}, + { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_source": "webui", + "attachments": [], + "_active_turn_token": "stream-a:222.5", + }, {"role": "assistant", "content": "answer", "timestamp": 223}, ], + stream_id="stream-a", + started_at=222.5, + source="webui", + attachments=[], ) - s.pending_started_at = 222 - s.pending_user_source = "webui" - s.pending_attachments = [] first = _transcript_already_advanced_past_pending(s) second = _transcript_already_advanced_past_pending(s) assert first == second is True @@ -178,43 +483,49 @@ def test_helper_is_pure_read_idempotent(): assert len(s.messages) == 2 - -def test_does_not_advance_when_repeated_prompt_has_different_checkpoint(): +def test_does_not_advance_when_repeated_prompt_has_different_token(): """#6366 regression guard for repeated-prompt identity: a *new* user turn that happens to repeat the same prompt text as an - older transcript row is NOT a stale advance. The strict match - uses timestamp + source + attachments to tell a stale duplicate - from a legitimate re-send.""" + older transcript row is NOT a stale advance. The active-turn + token tells a stale duplicate from a legitimate re-send, even + when both truncate to the same integer second.""" s = _FakeSession( pending="repeat this", messages=[ - # Older user row, same text but different identity. + # Older user row from a DIFFERENT stream, same text and the + # same integer second. { "role": "user", "content": "repeat this", "timestamp": 111, "_source": "webui", "attachments": [{"name": "old.png"}], + "_active_turn_token": "stream-old:111.5", }, {"role": "assistant", "content": "Earlier answer"}, + # The current turn's own row — pending, no answer yet. + { + "role": "user", + "content": "repeat this", + "timestamp": 111, + "_source": "webui", + "attachments": [{"name": "current.png"}], + "_active_turn_token": "stream-a:111.5", + }, ], + stream_id="stream-a", + started_at=111.5, + source="webui", + attachments=[{"name": "current.png"}], ) - # Pending checkpoint is a fresh turn: different timestamp and - # different attachments. - s.pending_started_at = 222 - s.pending_user_source = "webui" - s.pending_attachments = [{"name": "current.png"}] - # Strict match fails (timestamp + attachments differ), so the - # helper returns False and recovery is the legitimate path. assert _transcript_already_advanced_past_pending(s) is False -def test_advances_past_pending_with_exact_checkpoint_match(): - """The strict-match path: a transcript user row that exactly - matches the pending checkpoint (content + timestamp + - source + attachments) followed by a settled assistant row is - the canonical #6366 bug shape and must suppress the recovery - append.""" +def test_advances_past_pending_with_exact_token_match(): + """The strict-match path: a transcript user row that carries the + pending stream's own active-turn token, followed by a genuine final + answer, is the canonical #6366 bug shape and must suppress the + recovery append.""" s = _FakeSession( pending="hello world", messages=[ @@ -224,11 +535,363 @@ def test_advances_past_pending_with_exact_checkpoint_match(): "timestamp": 222, "_source": "webui", "attachments": [{"name": "a.png"}], + "_active_turn_token": "stream-a:222.5", }, {"role": "assistant", "content": "answer", "timestamp": 223}, ], + stream_id="stream-a", + started_at=222.5, + source="webui", + attachments=[{"name": "a.png"}], ) - s.pending_started_at = 222 - s.pending_user_source = "webui" - s.pending_attachments = [{"name": "a.png"}] assert _transcript_already_advanced_past_pending(s) is True + + +def test_does_not_advance_when_final_answer_belongs_to_a_later_turn(): + """A final answer that sits after a newer user row belongs to that + newer turn, so it cannot prove the pending turn advanced.""" + s = _FakeSession( + pending="hello world", + messages=[ + { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_active_turn_token": "stream-a:222.5", + }, + {"role": "user", "content": "a different question", "timestamp": 224}, + {"role": "assistant", "content": "answer", "timestamp": 225}, + ], + stream_id="stream-a", + started_at=222.5, + ) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_does_not_advance_when_tool_execution_was_interrupted(): + """A turn interrupted mid-tool-execution ends with a tool-call-only + assistant row and a tool result — not a final answer. Recovery must + still replay the journal and mark the interruption.""" + s = _FakeSession( + pending="run the migration", + messages=[ + { + "role": "user", + "content": "run the migration", + "timestamp": 222, + "_active_turn_token": "stream-a:222.5", + }, + { + "role": "assistant", + "content": "I will run the migration now.", + "tool_calls": [{"id": "call_1"}], + }, + {"role": "tool", "tool_call_id": "call_1", "content": "partial output"}, + ], + stream_id="stream-a", + started_at=222.5, + ) + assert _transcript_already_advanced_past_pending(s) is False + + +# ── full _apply_core_sync_or_error_marker recovery cases ──────────────── +# +# Neither guard rail above is reachable through the predicate in +# isolation: the suppression decision only takes effect where +# ``_apply_core_sync_or_error_marker`` calls it. These probes drive the +# real repair path end to end. + + +def _write_journal(session_id, stream_id, events): + for event_name, payload in events: + append_run_event(session_id, stream_id, event_name, payload) + + +def test_full_recovery_keeps_distinct_same_second_repeated_prompt(tmp_path): + """Two distinct streams send the same prompt inside one integer second. + + The earlier stream's turn is fully settled in the durable transcript. + The later stream is the pending one and was cancelled before it ever + checkpointed a user row. The repair must recover the pending prompt + instead of matching it against the earlier stream's row — which the + integer-second-truncating checkpoint matcher happily accepts — and + silently clearing the only durable copy of it. + """ + session_id = "issue6366_same_second_repeat" + stale_stream = "stream-stale" + pending_stream = "stream-pending" + # Both turns started inside the same integer second. + shared_second = 1_700_000_000 + + session = Session( + session_id=session_id, + title="Same-second repeated prompt", + messages=[ + { + "role": "user", + "content": "run the migration", + "timestamp": shared_second + 0.1, + "_active_turn_token": f"{stale_stream}:{shared_second + 0.1:.17g}", + }, + {"role": "assistant", "content": "Migration finished."}, + ], + context_messages=[ + {"role": "user", "content": "run the migration"}, + {"role": "assistant", "content": "Migration finished."}, + ], + pending_user_message="run the migration", + pending_started_at=shared_second + 0.6, + active_stream_id=pending_stream, + ) + _write_journal( + session_id, + pending_stream, + [("token", {"text": "Checking the migration status."})], + ) + + assert _apply_core_sync_or_error_marker( + session, + tmp_path / "missing-core-transcript.json", + stream_id_for_recheck=pending_stream, + ) is True + + # The pending prompt survives — recovery materialized it as a recovered + # user row instead of dropping it against the other stream's checkpoint. + recovered_users = [ + message + for message in session.messages + if message.get("_recovered") and message.get("content") == "run the migration" + ] + assert len(recovered_users) == 1 + # The journal's visible partial work was replayed, not suppressed. + assert any( + message.get("_recovered_from_run_journal") + and message.get("content") == "Checking the migration status." + for message in session.messages + ) + # And the interruption is marked so the user can see what happened. + assert any( + message.get("_error") and message.get("type") == "interrupted" + for message in session.messages + ) + # The earlier settled turn was left untouched. + assert session.messages[1]["content"] == "Migration finished." + + +def test_full_recovery_keeps_legacy_same_second_repeated_prompt(tmp_path): + """Legacy transcript with no tokens on either side: two identical + prompts inside one second must still be told apart, otherwise the + second prompt is matched against the first's checkpoint and dropped.""" + session_id = "issue6366_legacy_same_second_repeat" + stream_id = "stream-legacy-pending" + shared_second = 1_700_000_100 + + session = Session( + session_id=session_id, + title="Legacy same-second repeated prompt", + messages=[ + # The other stream's row: same text, same integer second, and + # distinguishable only by its different attachments. + { + "role": "user", + "content": "retry the upload", + "timestamp": shared_second, + "_source": "webui", + "attachments": [{"name": "first.png"}], + }, + {"role": "assistant", "content": "Upload done."}, + ], + context_messages=[ + {"role": "user", "content": "retry the upload"}, + {"role": "assistant", "content": "Upload done."}, + ], + pending_user_message="retry the upload", + pending_started_at=shared_second, + pending_user_source="webui", + pending_attachments=[{"name": "second.png"}], + active_stream_id=stream_id, + ) + _write_journal(session_id, stream_id, [("token", {"text": "Re-running upload."})]) + + assert _apply_core_sync_or_error_marker( + session, + tmp_path / "missing-core-transcript.json", + stream_id_for_recheck=stream_id, + ) is True + + recovered_users = [ + message + for message in session.messages + if message.get("_recovered") and message.get("content") == "retry the upload" + ] + assert len(recovered_users) == 1 + assert recovered_users[0].get("attachments") == [{"name": "second.png"}] + assert any( + message.get("_recovered_from_run_journal") + and message.get("content") == "Re-running upload." + for message in session.messages + ) + + +@pytest.mark.parametrize( + "tail_rows", + [ + pytest.param( + [{"role": "assistant", "content": ""}], + id="empty-assistant-tail", + ), + pytest.param( + [ + { + "role": "assistant", + "content": "I will inspect the file now.", + "tool_calls": [{"id": "call_1"}], + }, + {"role": "tool", "tool_call_id": "call_1", "content": "partial output"}, + ], + id="tool-call-only-tail", + ), + pytest.param( + [ + { + "role": "assistant", + "content": "Still working on the first step…", + "_partial": True, + }, + ], + id="interim-progress-tail", + ), + pytest.param( + [ + { + "role": "assistant", + "content": "[CONTEXT COMPACTION — REFERENCE ONLY] summary", + }, + ], + id="compaction-tail", + ), + ], +) +def test_full_recovery_runs_journal_replay_for_non_final_tails(tmp_path, tail_rows): + """An interrupted turn whose tail is not a genuine final answer must NOT + be treated as completed: journal replay has to run and the turn has to + be marked as interrupted. + """ + session_id = f"issue6366_tail_{tail_rows[-1].get('_partial', 'x')}" + session_id = "issue6366_non_final_tail" + stream_id = "stream-interrupted-tools" + started_at = 1_700_000_500.5 + + session = Session( + session_id=session_id, + title="Interrupted tool-running turn", + messages=[ + { + "role": "user", + "content": "inspect the failing test", + "timestamp": int(started_at), + "_active_turn_token": f"{stream_id}:{started_at:.17g}", + }, + *tail_rows, + ], + context_messages=[ + {"role": "user", "content": "inspect the failing test"}, + ], + pending_user_message="inspect the failing test", + pending_started_at=started_at, + active_stream_id=stream_id, + ) + _write_journal( + session_id, + stream_id, + [ + ("reasoning", {"text": "Reading the assertion first."}), + ("token", {"text": "The assertion never ran."}), + ( + "tool", + { + "name": "terminal", + "preview": "pytest tests/test_x.py", + "args": {"command": "pytest tests/test_x.py"}, + }, + ), + ], + ) + before = json.dumps(session.messages, ensure_ascii=False) + + assert _apply_core_sync_or_error_marker( + session, + tmp_path / "missing-core-transcript.json", + stream_id_for_recheck=stream_id, + ) is True + + # Journal replay ran: the visible partial output is back in the transcript. + assert any( + message.get("_recovered_from_run_journal") + and message.get("content") == "The assertion never ran." + for message in session.messages + ) + # The interrupted turn is marked, so the user sees what happened. + assert any( + message.get("_error") + and message.get("type") == "interrupted" + and "partial output above was recovered" + in str(message.get("content", "")) + for message in session.messages + ) + # Recovery actually appended rows rather than clearing pending state only. + assert json.dumps(session.messages, ensure_ascii=False) != before + assert session.messages[0]["content"] == "inspect the failing test" + + +def test_full_recovery_suppresses_duplicates_only_for_genuine_final_answer(tmp_path): + """Positive control for the suppression branch: when the pending turn's + own user row is followed by a genuine final answer, the repair clears + the stale pending state and appends nothing. + """ + session_id = "issue6366_suppression_control" + stream_id = "stream-stale-cancelled" + started_at = 1_700_000_900.25 + + session = Session( + session_id=session_id, + title="Already-answered pending turn", + messages=[ + { + "role": "user", + "content": "summarise the diff", + "timestamp": int(started_at), + "_active_turn_token": f"{stream_id}:{started_at:.17g}", + }, + {"role": "assistant", "content": "The diff adds a guard."}, + ], + context_messages=[ + {"role": "user", "content": "summarise the diff"}, + {"role": "assistant", "content": "The diff adds a guard."}, + ], + pending_user_message="summarise the diff", + pending_started_at=started_at, + active_stream_id=stream_id, + ) + _write_journal(session_id, stream_id, [("token", {"text": "stale partial text"})]) + before = json.dumps(session.messages, ensure_ascii=False) + + assert _apply_core_sync_or_error_marker( + session, + tmp_path / "missing-core-transcript.json", + stream_id_for_recheck=stream_id, + ) is True + + # Nothing was appended after the valid final answer. + assert json.dumps(session.messages, ensure_ascii=False) == before + assert not any( + message.get("_recovered") or message.get("_recovered_from_run_journal") + for message in session.messages + ) + assert not any(message.get("_error") for message in session.messages) + # The stale pending state was cleared. + assert session.pending_user_message is None + assert session.active_stream_id is None + assert session.pending_started_at is None + assert session.pending_user_source is None + assert session.pending_attachments == [] From 2fbb64bf1c5a50c957e2427d83acb5480446eb50 Mon Sep 17 00:00:00 2001 From: happy5318 Date: Wed, 23 Sep 2026 15:49:43 +0800 Subject: [PATCH 04/10] fix(models): require durable terminal evidence before suppressing recovery (re-gate 9/22) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Re-review #7682 (nesquena-hermes 2026-09-22T19:18:00Z, CHANGES_REQUESTED). `_transcript_already_advanced_past_pending` suppressed recovery on a transcript-only heuristic: the first assistant row with final visible text inside the pending turn's boundary. An interrupted multi-segment turn shaped like user -> assistant "Let me check the logs first." (unflagged interim prose) -> tool -> tool ... (stream dies here) passed the predicate as "finished" even though the interim sentence is not a final answer and the stream was cut off mid-tool. `_apply_core_sync_or_error_marker` then cleared pending state and returned early: the run journal was never replayed and no interruption marker was added — a half-finished turn rendered as complete, journaled tool output never appearing. Fix: a genuine final answer inside the turn is now NECESSARY BUT NOT SUFFICIENT. The suppression also requires durable same-stream terminal evidence — `_run_journal_terminal_state(session, active_stream_id) == 'completed'` (a `done`/`stream_end` event). Without it, fall through to the normal journal-recovery path so the partial output is replayed and the interruption is marked. Regression tests (red/green verified — both fail with the gate removed): - test_does_not_advance_when_unflagged_interim_prose_precedes_tools - test_full_recovery_runs_journal_replay_for_unflagged_interim_prose (drives `_apply_core_sync_or_error_marker`, asserts journal replay + interruption marker present) - test_does_advance_when_journal_marked_done_after_interim_prose (positive) Existing `_FakeSession` fixtures updated to carry a session_id and write a `done` journal event so the durable-evidence gate is satisfied for the settled-turn tests. Verification: test_issue6366 35 passed; anchor_scene + cancelled_turn neighbours 91 passed. Red/green proven on both regressions. Co-Authored-By: Claude Opus 4.8 (1M context) --- api/models.py | 39 ++- tests/test_issue6366_stale_cancel_recovery.py | 245 +++++++++++++++++- 2 files changed, 269 insertions(+), 15 deletions(-) diff --git a/api/models.py b/api/models.py index ee1ae53f..1e05e019 100644 --- a/api/models.py +++ b/api/models.py @@ -2563,11 +2563,23 @@ def _transcript_already_advanced_past_pending(session) -> bool: recovery active, because that is precisely the interrupted-turn shape the recovery path exists for. + The transcript heuristic alone is not durable: an unflagged interim + prose row ("Let me check the logs first.") followed by tool rows that + were never followed by a final answer in the transcript passes the + predicate as a "completed" turn, but the stream may have died mid- + tool and the run journal was never marked done. 9/22 re-gate: also + require durable same-stream terminal evidence — the run journal + must report ``completed`` (a ``done`` or ``stream_end`` event) for + the pending stream id. Without it, fall through to the normal + journal-recovery path so the partial output is replayed and the + interruption is marked. + The suppression is therefore deliberately conservative: whenever the - turn's identity or its completion cannot be positively proven, this - helper returns False and the prompt is recovered instead of discarded. - A false negative leaves a visible cosmetic duplicate; a false positive - silently destroys a prompt or a response. + turn's identity, its completion, or its durable terminal evidence + cannot be positively proven, this helper returns False and the + prompt is recovered instead of discarded. A false negative leaves a + visible cosmetic duplicate; a false positive silently destroys a + prompt or a response. Idempotent: this is a pure read that does not mutate the session. Running recovery twice therefore produces the same outcome. @@ -2582,10 +2594,21 @@ def _transcript_already_advanced_past_pending(session) -> bool: for idx, message in enumerate(messages): if not _transcript_user_row_is_pending_turn(message, session, pending_token): continue - # Found the pending turn's own user row at ``idx``. Only a genuine - # final answer inside that turn's boundary proves the transcript - # advanced past it. - return _pending_turn_has_final_assistant_answer(messages, idx) + # Found the pending turn's own user row at ``idx``. A genuine + # final answer inside that turn's boundary is necessary but not + # sufficient: the run journal must also have a same-stream + # terminal ``completed`` event, otherwise unflagged interim + # prose ("Let me check the logs first.") followed by tool rows + # can be mistaken for a finished turn. + if not _pending_turn_has_final_assistant_answer(messages, idx): + continue + stream_id = getattr(session, 'active_stream_id', None) + if _run_journal_terminal_state(session, stream_id) != 'completed': + # No durable same-stream terminal evidence — the turn may + # have died mid-tool. Fall through to the journal-recovery + # path instead of suppressing it. + return False + return True return False diff --git a/tests/test_issue6366_stale_cancel_recovery.py b/tests/test_issue6366_stale_cancel_recovery.py index 6b35c47c..b84957cf 100644 --- a/tests/test_issue6366_stale_cancel_recovery.py +++ b/tests/test_issue6366_stale_cancel_recovery.py @@ -87,7 +87,9 @@ class _FakeSession: started_at=None, source=None, attachments=None, + session_id="fake-session-6366", ): + self.session_id = session_id self.pending_user_message = pending self.messages = list(messages or []) self.active_stream_id = stream_id @@ -299,7 +301,9 @@ def test_advances_past_pending_when_user_matched_and_settled_assistant_follows() """Canonical bug shape: a stale pending turn's user row sits at an older index, and a newer settled assistant row follows. Recovery must not append duplicate rows after the assistant. - Identity comes from the pending stream's own active-turn token.""" + Identity comes from the pending stream's own active-turn token; + the journal is marked ``done`` (terminal_state ``completed``) so + the suppression gate is satisfied.""" s = _FakeSession( pending="hello world", messages=[ @@ -318,6 +322,7 @@ def test_advances_past_pending_when_user_matched_and_settled_assistant_follows() source="webui", attachments=[{"name": "a.png"}], ) + _write_journal(s.session_id, "stream-a", [("done", {})]) assert _transcript_already_advanced_past_pending(s) is True @@ -430,7 +435,8 @@ def test_does_not_advance_when_no_messages(): def test_advances_past_pending_with_unrelated_intervening_rows(): """The pending turn's user row followed by a tool row and THEN a final answer counts as a real advance. The helper scans past non-assistant - roles correctly.""" + roles correctly, and the journal is marked ``done`` so the durable + terminal evidence gate is satisfied.""" s = _FakeSession( pending="hello world", messages=[ @@ -450,13 +456,16 @@ def test_advances_past_pending_with_unrelated_intervening_rows(): source="webui", attachments=[], ) + _write_journal(s.session_id, "stream-a", [("done", {})]) assert _transcript_already_advanced_past_pending(s) is True def test_helper_is_pure_read_idempotent(): """Idempotence: running the helper twice on the same session state returns the same result. The helper does not mutate the - session, so recovery is automatically safe to run twice.""" + session, so recovery is automatically safe to run twice. Journal + is marked ``done`` so the durable terminal evidence gate passes + on both invocations.""" s = _FakeSession( pending="hello world", messages=[ @@ -475,9 +484,11 @@ def test_helper_is_pure_read_idempotent(): source="webui", attachments=[], ) + _write_journal(s.session_id, "stream-a", [("done", {})]) first = _transcript_already_advanced_past_pending(s) second = _transcript_already_advanced_past_pending(s) - assert first == second is True + assert first is True + assert second is True # And the session state was not mutated. assert s.pending_user_message == "hello world" assert len(s.messages) == 2 @@ -544,6 +555,7 @@ def test_advances_past_pending_with_exact_token_match(): source="webui", attachments=[{"name": "a.png"}], ) + _write_journal(s.session_id, "stream-a", [("done", {})]) assert _transcript_already_advanced_past_pending(s) is True @@ -594,6 +606,92 @@ def test_does_not_advance_when_tool_execution_was_interrupted(): assert _transcript_already_advanced_past_pending(s) is False +def test_does_not_advance_when_unflagged_interim_prose_precedes_tools(): + """9/22 re-gate regression: an unflagged interim prose row + ("Let me check the logs first.") followed by tool rows passes the + transcript predicate (``_pending_turn_has_final_assistant_answer`` + returns True at the first visible assistant row), but the run + journal was never marked ``done`` — the stream may have died mid- + tool. Without durable same-stream terminal evidence the suppression + gate must stay closed, so ``_transcript_already_advanced_past_pending`` + returns False and the recovery path is free to run.""" + s = _FakeSession( + pending="check the failing test", + messages=[ + { + "role": "user", + "content": "check the failing test", + "timestamp": 222, + "_active_turn_token": "stream-a:222.5", + }, + # The bug shape: an unflagged assistant row with visible + # text and no tool_calls. The transcript heuristic accepts + # this as a final answer (it carries visible text and has + # no _partial / _error / _recovered marker), but the row + # is actually pre-final interim prose that the repository + # journals as ``interim_assistant`` while the model + # decides what tool to call. The tool_calls that + # produced the following tool rows live on a *later* row + # in the original stream — the journal has not yet + # recorded them because the WebUI process died. + { + "role": "assistant", + "content": "Let me check the logs first.", + }, + # The stream then ran tools before dying. The transcript + # therefore has tool rows after the unflagged prose. + {"role": "tool", "tool_call_id": "call_1", "content": "log tail"}, + {"role": "tool", "tool_call_id": "call_2", "content": "log grep"}, + ], + stream_id="stream-a", + started_at=222.5, + ) + # The journal has tokens and an interim_assistant checkpoint, but + # no ``done`` / ``stream_end`` — the run is still mid-tool from + # the journal's point of view. + _write_journal( + s.session_id, "stream-a", + [ + ("token", {"text": "Let me check the logs first."}), + ("interim_assistant", {"text": "Let me check the logs first."}), + ("tool", {"name": "terminal", "tid": "call_1", "preview": "tail -n 50"}), + ("tool_complete", {"name": "terminal", "tid": "call_1", "preview": "log tail"}), + ("tool", {"name": "terminal", "tid": "call_2", "preview": "grep ERROR"}), + ], + ) + assert _transcript_already_advanced_past_pending(s) is False + + +def test_does_advance_when_journal_marked_done_after_interim_prose(): + """Companion positive control: when the run journal IS marked + ``done`` (terminal_state ``completed``) for the same stream, the + suppression gate opens even though the transcript carries the + unflagged interim prose + tool shape. Without the journal evidence + the predicate would still hold the gate closed, so this case pins + the boundary between the two new branches.""" + s = _FakeSession( + pending="check the failing test", + messages=[ + { + "role": "user", + "content": "check the failing test", + "timestamp": 222, + "_active_turn_token": "stream-a:222.5", + }, + { + "role": "assistant", + "content": "Let me check the logs first.", + }, + {"role": "tool", "tool_call_id": "call_1", "content": "log tail"}, + {"role": "assistant", "content": "The logs show a stale lock.", "timestamp": 223}, + ], + stream_id="stream-a", + started_at=222.5, + ) + _write_journal(s.session_id, "stream-a", [("done", {})]) + assert _transcript_already_advanced_past_pending(s) is True + + # ── full _apply_core_sync_or_error_marker recovery cases ──────────────── # # Neither guard rail above is reachable through the predicate in @@ -846,8 +944,9 @@ def test_full_recovery_runs_journal_replay_for_non_final_tails(tmp_path, tail_ro def test_full_recovery_suppresses_duplicates_only_for_genuine_final_answer(tmp_path): """Positive control for the suppression branch: when the pending turn's - own user row is followed by a genuine final answer, the repair clears - the stale pending state and appends nothing. + own user row is followed by a genuine final answer AND the run journal + has the same-stream ``done`` event (durable terminal evidence), the + repair clears the stale pending state and appends nothing. """ session_id = "issue6366_suppression_control" stream_id = "stream-stale-cancelled" @@ -873,7 +972,12 @@ def test_full_recovery_suppresses_duplicates_only_for_genuine_final_answer(tmp_p pending_started_at=started_at, active_stream_id=stream_id, ) - _write_journal(session_id, stream_id, [("token", {"text": "stale partial text"})]) + # Same-stream durable terminal evidence — the run journal was marked + # ``done`` (terminal_state ``completed``) so the suppression gate + # opens and the transcript-only advance is accepted as terminal. + # No visible token events are written so the line-3648 path's + # ``_append_journaled_partial_output`` has nothing to append. + _write_journal(session_id, stream_id, [("done", {})]) before = json.dumps(session.messages, ensure_ascii=False) assert _apply_core_sync_or_error_marker( @@ -895,3 +999,130 @@ def test_full_recovery_suppresses_duplicates_only_for_genuine_final_answer(tmp_p assert session.pending_started_at is None assert session.pending_user_source is None assert session.pending_attachments == [] + + +def test_full_recovery_runs_journal_replay_for_unflagged_interim_prose(tmp_path): + """9/22 re-gate regression, end-to-end: the durable transcript + already advanced past the pending turn into an unflagged interim + prose row followed by tool rows, but the run journal was never + marked ``done``. Before the fix, + ``_transcript_already_advanced_past_pending`` returned True at the + first visible assistant row, ``_apply_core_sync_or_error_marker`` + cleared the pending state, and the user saw a half-finished turn + with the journal's partial output and the interruption marker + both suppressed. The fix requires durable same-stream terminal + evidence (``done`` / ``stream_end`` → ``completed``) before + suppressing recovery, so the run journal is replayed and the + ``interrupted`` marker is appended for this unflagged-prose + shape. + """ + session_id = "issue6366_unflagged_interim_prose" + stream_id = "stream-unflagged-interim" + started_at = 1_700_001_200.75 + + session = Session( + session_id=session_id, + title="Unflagged interim prose followed by tool rows", + messages=[ + { + "role": "user", + "content": "check the failing test", + "timestamp": int(started_at), + "_active_turn_token": f"{stream_id}:{started_at:.17g}", + }, + # The bug shape: an unflagged assistant row that the + # transcript heuristic accepts as a "final answer" because + # it carries visible text and no _partial / _error / + # _recovered / _recovered_from_run_journal hint. The + # stream had just started talking about what it was about + # to do, then ran tools. The tool_calls that produced the + # following tool rows live on a later row in the original + # stream — the WebUI process died before the tool-call + # assistant row was committed, so the transcript currently + # shows just the pre-final prose followed by tool result + # rows. + { + "role": "assistant", + "content": "Let me check the logs first.", + }, + {"role": "tool", "tool_call_id": "call_1", "content": "log tail"}, + {"role": "tool", "tool_call_id": "call_2", "content": "log grep"}, + ], + context_messages=[ + {"role": "user", "content": "check the failing test"}, + ], + pending_user_message="check the failing test", + pending_started_at=started_at, + active_stream_id=stream_id, + ) + # The run journal has visible progress — the unflagged prose was + # emitted as a token, the run was checkpointed as interim_assistant, + # and one of the tool rows finished. Crucially there is no + # ``done`` / ``stream_end`` event: the stream died mid-tool, so + # the journal has no durable terminal evidence. + _write_journal( + session_id, stream_id, + [ + ("reasoning", {"text": "Reading the assertion first."}), + ("token", {"text": "Let me check the logs first."}), + ("interim_assistant", {"text": "checkpoint"}), + ( + "tool", + { + "name": "terminal", + "tid": "call_1", + "preview": "tail -n 50", + "args": {"command": "tail -n 50"}, + }, + ), + ( + "tool_complete", + { + "name": "terminal", + "tid": "call_1", + "preview": "log tail", + "duration": 0.05, + }, + ), + ], + ) + before = json.dumps(session.messages, ensure_ascii=False) + + assert _apply_core_sync_or_error_marker( + session, + tmp_path / "missing-core-transcript.json", + stream_id_for_recheck=stream_id, + ) is True + + # Recovery actually appended rows rather than clearing pending + # state only. + assert json.dumps(session.messages, ensure_ascii=False) != before + # The journal's visible partial output is back in the transcript: + # the "Let me check the logs first." prose that was emitted over + # SSE before the WebUI process died is now visible again. + assert any( + message.get("_recovered_from_run_journal") + and "Let me check the logs first." in str(message.get("content", "")) + for message in session.messages + ) + # And the interrupted turn is marked, so the user can see the + # stream was cut off rather than presented with a half-finished + # turn that looks complete. + interrupted_marker = next( + ( + message + for message in session.messages + if message.get("_error") and message.get("type") == "interrupted" + ), + None, + ) + assert interrupted_marker is not None + assert "partial output above was recovered" in str( + interrupted_marker.get("content", ""), + ) + # Pending state was cleared (recovery always clears it) and no + # duplicate user row was appended ahead of the journal replay — + # the original user row is still the first message. + assert session.messages[0]["content"] == "check the failing test" + assert session.pending_user_message is None + assert session.active_stream_id is None From a066c25338d48a5c866ef29b460eb0227d115004 Mon Sep 17 00:00:00 2001 From: Rodrigo Gomes da Silva Date: Tue, 15 Sep 2026 03:26:17 -0300 Subject: [PATCH 05/10] fix(sessions): stop save() from reading and parsing the whole sidecar to count it The #1558 backup safeguard in Session.save() compares the on-disk message count with the incoming one and writes .json.bak only when the save would shrink the array. Sound -- but it obtained the on-disk count by json.loads(self.path.read_text()), the entire file, on EVERY save, including the 99.9% of saves that grow the conversation and never back anything up. Measured in production (Mac docker stack), in the running container: sessions/fa3bca34a0c6.json 203,439,398 bytes / 266,940 messages read_text 17,453 ms json.loads 2,924 ms TOTAL to obtain one integer: 20,377 ms <- paid on every save That was the 39,357 ms /api/session request stuck at t6_after_json_write, after the read-side re-parse had already been fixed. It scales linearly with the file, and the file grows for as long as the conversation does. The decision below needs ONE integer. save() already writes that integer into the metadata prefix, before `messages`, as `message_count` -- exactly so readers can have it without parsing the body, which is how load_metadata_only() and the sidebar freshness check already work. So save() now reads THAT, through the same bounded 64 KiB prefix reader, and falls back to the full read + parse whenever the prefix does not carry a usable count: a legacy pre-#5854 sidecar whose count serialized after the scenes, a file with no top-level `messages` key at all, a corrupt or truncated prefix, metadata alone that overflows the budget, or an unreadable path. Fail-open is the contract -- never "assume no shrink". The full text is still read for one thing only: the .bak body, when a shrink is actually detected. An in-memory stat identity was tried first and is NOT sufficient, so it is gone. The first revision of this branch stamped the (inode, size, mtime_ns) of the bytes it wrote and trusted the remembered count while that tuple matched, on the reasoning that "stat says nothing touched it, so the array is the one I wrote". Review (nesquena-hermes, 2026-09-21) reproduced the hole: a stat tuple is not a content identity. A same-length in-place rewrite inside one mtime tick keeps inode, size AND mtime_ns -- ext4 stamps mtime from a coarse clock, so two writes in the same tick share one mtime_ns -- and the next shrinking save then reads a stale count, concludes it is growing, overwrites the longer transcript and writes no .bak. That is the #1558 data-loss shape the safeguard exists to prevent, reachable on this branch. A count read from the file's own bytes cannot be fooled that way: any rewrite of the bytes rewrites the count. So _disk_identity() is removed along with the identity fields on Session, the stamp after the write, and the seeding in load(). _prefix_message_count() is the replacement: bounded prefix -> the number the writer itself recorded. One trap survives the redesign and is pinned. load() de-duplicates adjacent partials and immediately saves the shorter transcript (#2592), and that save must produce a .bak because it shrinks the array on purpose. The count that decides it is the one the writer stamped on the file -- the ON-DISK length it wrote -- not the post-collapse length of the object. Read the object's own length instead and the #2592 backup disappears silently. session_recovery._msg_count() pays the same cost at boot and is deliberately left alone. Its "torn file -> -1" contract is what makes recovery restore a .bak over a truncated live file, and no bounded read can prove a file is not truncated. A boot pays it once; a save paid it every time. Tests: 9 in tests/test_save_count_without_full_parse.py. The review's reproduction is the last one -- a same-inode, same-byte-length, same-mtime_ns in-place rewrite of 2 -> 5 messages, then a save of 3, which must leave the 5-message array recoverable. It fails against the removed stat-identity design (re-verified by restoring that revision: no .bak is written and the 5-message transcript is lost) and passes here. Mutation-checked the same way: recording the post-collapse length fails exactly the #2592 test. 783 tests pass across the 41 files that touch this surface (save/load, metadata-only, collapse, .bak recovery, sidecar reconciliation); ruff is clean on the diff. --- api/models.py | 84 +++++- tests/test_save_count_without_full_parse.py | 318 ++++++++++++++++++++ 2 files changed, 396 insertions(+), 6 deletions(-) create mode 100644 tests/test_save_count_without_full_parse.py diff --git a/api/models.py b/api/models.py index dd775c68..18566cd9 100644 --- a/api/models.py +++ b/api/models.py @@ -1537,12 +1537,43 @@ class Session: # their .bak get restored automatically. try: if self.path.exists(): - existing_text = self.path.read_text(encoding='utf-8') - try: - existing = json.loads(existing_text) - existing_msg_count = len(existing.get('messages') or []) - except (json.JSONDecodeError, ValueError): - existing_msg_count = -1 # corrupt → always back up + # The on-disk count, without reading the body. + # + # The decision below is a function of ONE integer -- how many + # messages the file on disk holds -- and save() already writes + # that integer into the metadata prefix, before `messages`, as + # `message_count` (see METADATA_FIELDS above; load_metadata_only + # and the sidebar freshness check read it the same way). So read + # THAT through a bounded 64 KiB prefix instead of the whole file. + # Measured before this: a 203,439,398-byte sidecar cost 20,377 ms + # (17,453 in read_text, 2,924 in json.loads) to yield one integer, + # on EVERY save -- including the grow-saves that never back + # anything up. The prefix read is O(64 KiB), and because the count + # is part of the bytes on disk it travels with any rewrite of them. + # + # An in-memory "I wrote this, stat says nothing changed" cache is + # NOT sufficient here, and was removed after review: (inode, size, + # mtime_ns) is not a content identity. A same-length in-place + # rewrite inside one mtime tick keeps all three fields -- ext4 + # stamps mtime from a coarse clock, so two writes in the same tick + # share one mtime_ns -- and a stale cached count then reads a real + # shrink as a growth and skips the #1558 backup. The prefix count + # cannot be fooled that way. + # + # Every unknown falls through to the full read + parse below: a + # legacy (pre-#5854) sidecar whose count is not in the prefix, a + # corrupt or truncated prefix, a file with no top-level `messages` + # key at all, or metadata alone that overflows the budget. + # Fail-open is the contract -- never "assume no shrink". + existing_text = None + existing_msg_count = _prefix_message_count(self.path) + if existing_msg_count is None: + existing_text = self.path.read_text(encoding='utf-8') + try: + existing = json.loads(existing_text) + existing_msg_count = len(existing.get('messages') or []) + except (json.JSONDecodeError, ValueError): + existing_msg_count = -1 # corrupt → always back up incoming_msg_count = len(self.messages or []) if ( existing_msg_count > 0 @@ -1560,6 +1591,10 @@ class Session: return if existing_msg_count > incoming_msg_count: bak_path = self.path.with_suffix('.json.bak') + if existing_text is None: + # The .bak body is the one thing that needs the full text, + # and a shrink is the one time it is needed. + existing_text = self.path.read_text(encoding='utf-8') # SHOULD-FIX #2 (Opus): atomic write via tmp+replace, # mirroring the main save() pattern below. Prevents a # torn .bak from a crash mid-write or a concurrent @@ -4239,6 +4274,43 @@ def _sidecar_stat_signature(path): int(st.st_size), int(getattr(st, 'st_ctime_ns', int(st.st_ctime * 1_000_000_000)))) +def _prefix_message_count(path): + """The sidecar's own persisted ``message_count``, from a bounded prefix. + + The #1558 shrink check in save() needs exactly one integer -- how many + messages the file on disk holds -- and the writer puts that integer in the + metadata prefix, before ``messages``, precisely so readers can have it + without parsing the body (see METADATA_FIELDS). Reading the prefix costs + O(64 KiB) against O(file size) for ``read_text`` + ``json.loads`` (measured + 2026-09-15: 20,377 ms for one integer on a 203,439,398-byte sidecar). + + Content-derived on purpose. A stat identity cannot stand in for it: an + in-place rewrite of the same length inside one mtime tick keeps inode, size + *and* mtime_ns, so a cached count would read a real shrink as a growth and + skip the backup. The count is part of the bytes, so any rewrite of them + rewrites it too. + + Returns None -- meaning "the caller must fall back to the full read + + parse" -- for every shape that does not carry a usable count: a legacy + pre-#5854 sidecar (scenes serialized before the count), a file with no + top-level ``messages`` key at all, a corrupt or truncated prefix, an + unreadable path, or metadata alone that overflows the budget. + """ + try: + raw = _read_metadata_json_prefix(path) + except (OSError, ValueError): + return None + if not raw: + return None + try: + parsed = json.loads(raw) + except (json.JSONDecodeError, ValueError): + return None + if not isinstance(parsed, dict): + return None + return _parse_nonnegative_int(parsed.get('message_count')) + + def _legacy_sidecar_facts_get(sid): """Return cached authoritative facts for a LEGACY sidecar, or None (#5854). diff --git a/tests/test_save_count_without_full_parse.py b/tests/test_save_count_without_full_parse.py new file mode 100644 index 00000000..cacde78d --- /dev/null +++ b/tests/test_save_count_without_full_parse.py @@ -0,0 +1,318 @@ +"""A save must not read and parse the whole sidecar just to count its messages. + +The #1558 backup safeguard in `Session.save()` compares the on-disk message +count with the incoming one, and writes `.json.bak` only when the save +would SHRINK the array. Sound. But it obtained the on-disk count by +`json.loads(self.path.read_text())` -- the entire file -- on EVERY save, +including the 99.9% of saves that grow the conversation and never back +anything up. `session_recovery._msg_count()` did the same on every boot, for +every live file that has a `.bak`. + +Measured in production on 2026-09-15, in the running container: + + sessions/fa3bca34a0c6.json 203,439,398 bytes / 266,940 messages + read_text 17,453 ms + json.loads 2,924 ms + TOTAL to obtain one integer: 20,377 ms <- paid on every save + +That was the 39,357 ms `/api/session` request stuck at stage +`t6_after_json_write`, after the read-side re-parse (#4633 recurrence) had +already been fixed. The cost scales linearly with the file, and the file grows +for as long as the conversation does. + +What the check needs is ONE integer -- how many messages the file on disk +holds -- and `save()` already writes that integer into the metadata prefix, +before `messages`, as `message_count`, exactly so readers can have it without +parsing the body (that is how `load_metadata_only()` and the sidebar freshness +check work). So `save()` reads that, through the same bounded 64 KiB prefix +reader, and falls back to the full read + parse whenever the prefix does not +carry a usable count. The `.bak` body is the one thing that still needs the +full text, and a shrink is the one time it is needed. + +The count is read from the FILE, not cached in the object, and that is the +point. An earlier revision of this PR trusted an in-memory (inode, size, +mtime_ns) signature: "this object wrote the file and stat says nothing changed, +so its length is already known". Review (nesquena-hermes, 2026-09-21) showed +that is not a content identity -- a same-length in-place rewrite inside one +mtime tick keeps all three fields, because ext4 stamps mtime from a coarse +clock, so a stale cached count reads a real shrink as a growth and skips the +#1558 backup. That is the test at the bottom of this file. A count read from +the bytes cannot be fooled that way: any rewrite of the bytes rewrites it. + +Fail-open is the contract these tests pin -- whenever the prefix has no usable +count (a legacy pre-#5854 sidecar, a file with no top-level `messages` key, a +corrupt or truncated prefix, metadata that overflows the budget), the behaviour +must be exactly today's full parse, never "assume no shrink". + +`session_recovery._msg_count()` pays the same cost at boot and is left alone +on purpose: its "torn file -> -1" contract is what makes recovery restore a +`.bak` over a truncated live file, and no bounded read can prove a file is +not truncated. +""" +import json +import os +import pathlib +from collections import OrderedDict + +import pytest + +import api.models as M + + +@pytest.fixture +def session_store(tmp_path, monkeypatch): + sdir = tmp_path / "sessions" + sdir.mkdir(parents=True, exist_ok=True) + monkeypatch.setattr(M, "SESSION_DIR", sdir) + monkeypatch.setattr(M, "SESSIONS", OrderedDict()) + return sdir + + +def _msgs(n): + return [{"role": "user" if i % 2 == 0 else "assistant", "content": f"m{i}"} for i in range(n)] + + +def _make(session_store, sid, n): + s = M.Session(session_id=sid, title="T", workspace=str(session_store.parent), + model="glm", messages=_msgs(n)) + s.save() + return s + + +def _spy_full_reads(monkeypatch, target): + """Count `Path.read_text()` calls on ONE path. + + `read_text` is the whole-file read; the metadata-prefix reader uses `open()` + with a byte budget and never goes through it. So this counter is exactly + "how many times was this file read in full", which is the cost being removed. + """ + calls = {"n": 0} + real = pathlib.Path.read_text + target = pathlib.Path(target) + + def counting(self, *a, **k): + if pathlib.Path(self) == target: + calls["n"] += 1 + return real(self, *a, **k) + + monkeypatch.setattr(type(pathlib.Path()), "read_text", counting) + return calls + + +# ── save(): the grow path must be free of full reads ──────────────────────── + +def test_grow_save_does_not_read_the_existing_file_in_full(session_store, monkeypatch): + """The common case. Appending messages must not read the previous file + in full -- this object wrote that file and it is untouched, so the on-disk + count is already known.""" + s = _make(session_store, "g1", 5) + calls = _spy_full_reads(monkeypatch, s.path) + + s.messages = _msgs(7) + s.save() + + assert calls["n"] == 0, f"grow-save read the existing sidecar in full {calls['n']}x" + assert not s.path.with_suffix(".json.bak").exists(), "a grow-save must not produce a backup" + assert len(M.Session.load("g1").messages) == 7 + + +def test_same_size_save_does_not_read_the_existing_file_in_full(session_store, monkeypatch): + """Metadata-only saves (title, flags, stream state) keep the array as-is. + They are the most frequent save of all and must be just as cheap.""" + s = _make(session_store, "g2", 5) + calls = _spy_full_reads(monkeypatch, s.path) + + s.title = "renamed" + s.save() + + assert calls["n"] == 0 + assert not s.path.with_suffix(".json.bak").exists() + + +# ── save(): the safeguards must be untouched ──────────────────────────────── + +def test_shrink_save_still_writes_bak_with_the_pre_shrink_content(session_store, monkeypatch): + """#1558 must keep working. A shrink still produces a `.bak` holding the + pre-shrink array -- and reads the file in full exactly once, for that body. + Not twice: knowing the count must not be followed by a redundant parse.""" + s = _make(session_store, "s1", 5) + calls = _spy_full_reads(monkeypatch, s.path) + + s.messages = _msgs(3) + s.save() + + reads_by_save = calls["n"] # taken before load() below, which reads the file too + bak = s.path.with_suffix(".json.bak") + assert bak.exists(), "a shrinking save must leave a recoverable backup" + assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 5 + assert len(M.Session.load("s1").messages) == 3 + assert reads_by_save == 1, f"shrink-save read the sidecar in full {reads_by_save}x; the .bak body needs exactly one" + + +def test_empty_active_snapshot_is_still_refused(session_store, monkeypatch): + """The other guard in the same block: an empty array with a live stream or + pending prompt must NOT overwrite a populated file (the #1558 data-loss + shape). This object was neither loaded nor saved by this process, so it has + no signature to trust and refusing may cost today's single full read -- + never more, and it must still refuse.""" + _make(session_store, "r1", 5) + empty = M.Session(session_id="r1", title="T", workspace=str(session_store.parent), + model="glm", messages=[], active_stream_id="a" * 32, + pending_user_message="still typing") + calls = _spy_full_reads(monkeypatch, empty.path) + + empty.save() + + reads_by_save = calls["n"] # taken before load() below, which reads the file too + assert len(M.Session.load("r1").messages) == 5, "the populated file must survive" + assert reads_by_save <= 1, f"refusing read the file {reads_by_save}x; an unknown identity costs at most one" + + +# ── save(): fail-open whenever the prefix carries no usable count ─────────── + +def _write_legacy_without_count(session_store, sid, n): + """A pre-#5854 sidecar: no persisted `message_count` at all.""" + doc = {"session_id": sid, "title": "Legacy", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "messages": _msgs(n)} + p = session_store / f"{sid}.json" + p.write_text(json.dumps(doc, indent=2), encoding="utf-8") + return p + + +def test_legacy_sidecar_without_message_count_still_backs_up_on_shrink(session_store): + """The prefix has no count to give, so this must fall back to the full + parse and keep the safeguard. Silently treating "no count" as "no shrink" + would reopen #1558.""" + _write_legacy_without_count(session_store, "l1", 5) + s = M.Session(session_id="l1", title="Legacy", workspace=str(session_store.parent), + model="glm", messages=_msgs(3)) + + s.save() + + bak = s.path.with_suffix(".json.bak") + assert bak.exists(), "fail-open: a legacy file must still be backed up on shrink" + assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 5 + + +def test_file_changed_on_disk_since_last_save_falls_back_and_still_backs_up(session_store): + """An externally grown file. This process saved 3 messages; something else + then rewrote the file with 6 (an external appender, another process, a + restore). The next save with 4 is a real shrink relative to DISK, and the + count that governs it is the one the file now carries (6), not the 3 this + process remembers: the `.bak` must hold the 6-message array.""" + s = _make(session_store, "x1", 3) + external = {"session_id": "x1", "title": "T", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "message_count": 6, "messages": _msgs(6)} + s.path.write_text(json.dumps(external, indent=2), encoding="utf-8") + + s.messages = _msgs(4) + s.save() + + bak = s.path.with_suffix(".json.bak") + assert bak.exists(), "an externally grown file must still be backed up on shrink" + assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 6 + + +def test_file_changed_on_disk_without_a_count_still_backs_up_on_shrink(session_store): + """The same external rewrite, but the replacement carries no + `message_count` (an older writer, a hand-edited file). No prefix count -> + the full parse runs and the 6-message array is still backed up.""" + s = _make(session_store, "x2", 3) + external = {"session_id": "x2", "title": "T", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "messages": _msgs(6)} + s.path.write_text(json.dumps(external, indent=2), encoding="utf-8") + + s.messages = _msgs(4) + s.save() + + bak = s.path.with_suffix(".json.bak") + assert bak.exists(), "fail-open: a count-less external rewrite must still be backed up" + assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 6 + + +def _tool_partial(ts=123): + """The exact shape the #2592 collapse recognises (copied from its test).""" + return {"role": "assistant", "content": "", "_partial": True, "timestamp": ts, + "reasoning": "same reasoning", + "_partial_tool_calls": [{"name": "execute_code", + "args": {"code": "raise RuntimeError('boom')"}, + "done": True, "is_error": True, "duration": 3.87}]} + + +def test_collapse_self_heal_on_load_still_backs_up_the_pre_collapse_array(session_store): + """#2592: load() de-duplicates adjacent partials and immediately saves the + shorter transcript, and that save must produce a `.bak` because it shrinks + the array on purpose. The count that decides it comes from the file's own + prefix, which the writer stamped with the length it wrote (5) -- the + ON-DISK length, not the post-collapse length of the object (3). Read the + object's own length instead and this backup silently disappears, so the + fixture carries the count a real save() would have written.""" + doc = {"session_id": "h1", "title": "T", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "message_count": 5, + "messages": [{"role": "user", "content": "run this"}, + _tool_partial(), _tool_partial(), _tool_partial(), + {"role": "assistant", "content": "**Task cancelled.**", "_error": True}]} + (session_store / "h1.json").write_text(json.dumps(doc), encoding="utf-8") + + loaded = M.Session.load("h1") + + assert sum(1 for m in loaded.messages if m.get("_partial")) == 1, "the collapse must have fired" + assert len(loaded.messages) == 3 + bak = (session_store / "h1.json").with_suffix(".json.bak") + assert bak.exists(), "the self-heal shrink must leave the pre-collapse array recoverable" + assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 5 + + +# ── the collision a stat identity cannot see (review, 2026-09-21) ─────────── + +def test_same_length_in_place_rewrite_inside_one_mtime_tick_still_backs_up(session_store): + """A same-length, same-mtime, same-inode rewrite must not hide a shrink. + + The removed revision of this PR trusted an in-memory (inode, size, + mtime_ns) identity: "this object wrote the file, stat says nothing changed, + so the on-disk length is the one I remember". A stat tuple is not a content + identity. Another writer replaces the sidecar IN PLACE with a longer + transcript that occupies the same number of bytes, inside one mtime tick, + so all three fields still match -- and the next shrinking save then reads + its stale count as "growing", overwrites the longer transcript and writes + no `.bak`. That is the #1558 data-loss shape the safeguard exists to + prevent, reachable on the branch, which is why the identity cache is gone. + + The count now comes from the file's own metadata prefix, which any rewrite + of the bytes has to carry, so the shrink is seen for what it is. + """ + p = session_store / "i1.json" + base = {"session_id": "i1", "title": "T", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0} + + def _render(n): + return json.dumps({**base, "message_count": n, "messages": _msgs(n)}, + indent=2, ensure_ascii=False) + + width = max(len(_render(2)), len(_render(5))) + short, grown = _render(2).ljust(width), _render(5).ljust(width) + assert len(short) == len(grown) == width + p.write_text(short, encoding="utf-8") + + loaded = M.Session.load("i1") # the only version this object knows + before = p.stat() + + p.write_text(grown, encoding="utf-8") # same inode, same byte length... + os.utime(p, ns=(before.st_atime_ns, before.st_mtime_ns)) + after = p.stat() + assert (after.st_ino, after.st_size, after.st_mtime_ns) == \ + (before.st_ino, before.st_size, before.st_mtime_ns), \ + "premise: the rewrite must be invisible to (inode, size, mtime_ns)" + assert len(loaded.messages) == 2, "premise: this object remembers 2 messages" + + loaded.messages = _msgs(3) + loaded.save() + + bak = p.with_suffix(".json.bak") + assert bak.exists(), "the 5-message transcript must be recoverable" + assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 5 + assert len(M.Session.load("i1").messages) == 3 From 41efd6c15ddec38ed38989d1261813fa8f5df0d8 Mon Sep 17 00:00:00 2001 From: Rodrigo Gomes da Silva Date: Wed, 23 Sep 2026 06:34:33 -0300 Subject: [PATCH 06/10] fix(recovery): derive sidecar count from message rows --- api/session_recovery.py | 9 +++- .../test_session_db_sidecar_reconciliation.py | 45 +++++++++++++++++++ 2 files changed, 53 insertions(+), 1 deletion(-) diff --git a/api/session_recovery.py b/api/session_recovery.py index d2d46cb2..d9b09479 100644 --- a/api/session_recovery.py +++ b/api/session_recovery.py @@ -601,7 +601,14 @@ def _state_db_row_to_sidecar(row: dict) -> dict: 'session_id': row.get('id'), 'title': row.get('title') or 'Recovered WebUI Session', 'workspace': workspace_value if isinstance(workspace_value, str) else '', - 'message_count': row.get('message_count') if isinstance(row.get('message_count'), int) else len(messages), + # state.db is canonical for the recovered message rows. Its denormalized + # sessions.message_count is absent from older schemas (read above as 0) + # and can be stale after recovery, so never copy it into a sidecar whose + # `messages` array we just materialized. Session.save() uses this field + # to decide whether an overwrite shrinks the transcript; it must match + # the payload exactly or the #1558 backup / empty-snapshot guards can be + # bypassed. + 'message_count': len(messages if isinstance(messages, list) else []), 'worktree_path': row.get('worktree_path') or None, 'worktree_branch': row.get('worktree_branch') or None, 'worktree_repo_root': row.get('worktree_repo_root') or None, diff --git a/tests/test_session_db_sidecar_reconciliation.py b/tests/test_session_db_sidecar_reconciliation.py index 8b00892b..ba27e574 100644 --- a/tests/test_session_db_sidecar_reconciliation.py +++ b/tests/test_session_db_sidecar_reconciliation.py @@ -45,9 +45,54 @@ def test_recover_missing_sidecars_from_state_db_materializes_webui_row(tmp_path) assert data["parent_session_id"] == "parent-1" assert data["source_tag"] == "webui" assert data["session_source"] == "webui" + assert data["message_count"] == 2 assert [m["content"] for m in data["messages"]] == ["message 1", "message 2"] +def test_recovered_sidecar_derives_message_count_from_materialized_rows(tmp_path, monkeypatch): + """Recovery must not copy a stale state.db summary into a live sidecar. + + `Session.save()` relies on sidecar `message_count` for both #1558 guards. + A recovery row whose denormalized DB count says zero but whose canonical + message rows contain five messages must therefore refuse an empty active + snapshot and preserve a five-message backup on a later shrink. + """ + import api.models as models + + sid = _make_state_db(tmp_path / "state.db", sid="recovered_stale_count", messages=5) + with sqlite3.connect(tmp_path / "state.db") as conn: + conn.execute("UPDATE sessions SET message_count = 0 WHERE id = ?", (sid,)) + + assert recover_missing_sidecars_from_state_db(tmp_path, tmp_path / "state.db")["materialized"] == 1 + sidecar = tmp_path / f"{sid}.json" + assert json.loads(sidecar.read_text(encoding="utf-8"))["message_count"] == 5 + + monkeypatch.setattr(models, "SESSION_DIR", tmp_path) + monkeypatch.setattr(models, "SESSION_INDEX_FILE", tmp_path / "_index.json") + active_empty = models.Session( + session_id=sid, + title="Recovered from DB", + workspace=str(tmp_path), + model="openai/gpt-5", + messages=[], + active_stream_id="a" * 32, + pending_user_message="still typing", + ) + active_empty.save() + assert len(json.loads(sidecar.read_text(encoding="utf-8"))["messages"]) == 5 + + shrinking = models.Session( + session_id=sid, + title="Recovered from DB", + workspace=str(tmp_path), + model="openai/gpt-5", + messages=[{"role": "user", "content": f"replacement {i}"} for i in range(3)], + ) + shrinking.save() + backup = sidecar.with_suffix(".json.bak") + assert len(json.loads(backup.read_text(encoding="utf-8"))["messages"]) == 5 + + def test_recover_missing_sidecars_from_state_db_skips_deleted_webui_tombstone(tmp_path, monkeypatch): import api.models as _m monkeypatch.setattr(_m, "SESSION_DIR", tmp_path) From 2054587673421fc77f301756118aa0e7fb41a556 Mon Sep 17 00:00:00 2001 From: Rodrigo Gomes da Silva Date: Wed, 23 Sep 2026 10:46:46 -0300 Subject: [PATCH 07/10] fix(sessions): gate the prefix count on a writer marker A persisted message_count is only authoritative when the write that produced the messages array also vouched for it. Sidecars written by an older writer (recovery materialization copied the denormalized state.db count over real rows) could carry a count stale against the array beside it, and the prefix shortcut then reads a real shrink as a growth and skips the #1558 backup. save() now stamps _mc_v alongside message_count, and _prefix_message_count refuses any count without the current marker: those files get the full parse, exactly as on the legacy path. One save re-marks the file, so the fast path resumes for steady state. Regression covers the unmarked stale-count shape and the re-marking transition; mutation-checked by removing the gate (the unmarked fixture then skips the backup, as the review's reproduction predicted). --- api/models.py | 49 ++++++++++++--- tests/test_save_count_without_full_parse.py | 70 ++++++++++++++++++++- 2 files changed, 111 insertions(+), 8 deletions(-) diff --git a/api/models.py b/api/models.py index 18566cd9..f340c7d8 100644 --- a/api/models.py +++ b/api/models.py @@ -1107,6 +1107,14 @@ _METADATA_PREFIX_MAX_BYTES = 1024 * 1024 #: even when the stop key sits at 2 KB. _METADATA_PREFIX_FIRST_STAGE_BYTES = 64 * 1024 +# Marks a sidecar as written by the CURRENT writer contract, where the +# persisted `message_count` equals len(messages) by construction (both keys +# land in the same atomic write). save()'s bounded-prefix shrink check trusts +# a persisted count ONLY when this marker sits next to it in the file; see +# _prefix_message_count(). Bump the value whenever the writer contract changes +# meaning, so counts from an older contract fall back to the full parse. +_MESSAGE_COUNT_MARKER = 1 + def _read_metadata_json_prefix(path, max_prefix_bytes=_METADATA_PREFIX_MAX_BYTES): """Read only the metadata portion before the large arrays. @@ -1505,6 +1513,16 @@ class Session: # legacy-format reader that stops at a scene key still finds the count. # The full anchor_activity_scenes bodies serialize AFTER messages. meta['message_count'] = len(self.messages or []) + # _mc_v marks this file as written by the current writer contract, + # where `message_count` equals len(messages) by construction and both + # keys land in the same atomic write. save()'s shrink guard takes the + # bounded-prefix shortcut ONLY for marked files; an unmarked count + # (a legacy pre-#5854 sidecar, a sidecar materialized by an older + # recovery writer, any foreign writer) gets the full parse, so a stale + # count from outside this contract can never read a real shrink as a + # growth and skip the #1558 backup. One save re-marks the file, so the + # fast path still covers steady state. + meta['_mc_v'] = _MESSAGE_COUNT_MARKER meta['anchor_scene_index'] = _anchor_scene_index_from_records(self.anchor_activity_scenes) # Keep the in-memory fingerprint aligned with what we just persisted, so a # later metadata-only reload of THIS object (or any fingerprint reader) @@ -1517,7 +1535,7 @@ class Session: meta['anchor_activity_scenes'] = self.anchor_activity_scenes if isinstance(self.anchor_activity_scenes, dict) else {} # Fields not in METADATA_FIELDS (e.g. last_usage) go at the end. Exclude # the keys we placed explicitly above so they aren't emitted twice. - _placed = {'message_count', 'anchor_scene_index', 'messages', 'tool_calls', 'anchor_activity_scenes'} + _placed = {'message_count', '_mc_v', 'anchor_scene_index', 'messages', 'tool_calls', 'anchor_activity_scenes'} extra = {k: v for k, v in self.__dict__.items() if k not in METADATA_FIELDS and k not in _placed and not k.startswith('_')} @@ -1562,8 +1580,11 @@ class Session: # # Every unknown falls through to the full read + parse below: a # legacy (pre-#5854) sidecar whose count is not in the prefix, a - # corrupt or truncated prefix, a file with no top-level `messages` - # key at all, or metadata alone that overflows the budget. + # count written without the current writer's _mc_v marker (an + # older writer's count can be stale relative to the messages + # array next to it), a corrupt or truncated prefix, a file with + # no top-level `messages` key at all, or metadata alone that + # overflows the budget. # Fail-open is the contract -- never "assume no shrink". existing_text = None existing_msg_count = _prefix_message_count(self.path) @@ -4274,6 +4295,9 @@ def _sidecar_stat_signature(path): int(st.st_size), int(getattr(st, 'st_ctime_ns', int(st.st_ctime * 1_000_000_000)))) +_MESSAGE_COUNT_MARKER = 1 + + def _prefix_message_count(path): """The sidecar's own persisted ``message_count``, from a bounded prefix. @@ -4290,11 +4314,20 @@ def _prefix_message_count(path): skip the backup. The count is part of the bytes, so any rewrite of them rewrites it too. + Gated on the writer marker (``_mc_v``, review 2026-09-22): a count is only + trusted when the SAME write that produced the messages array also vouched + for the count. Anything unmarked -- a legacy pre-#5854 sidecar, a sidecar + materialized by an older recovery writer whose denormalized count could be + stale against its rows, any foreign writer -- returns None and the caller + falls back to the full read + parse. After one save by the current writer + the file carries the marker and the fast path resumes. + Returns None -- meaning "the caller must fall back to the full read + - parse" -- for every shape that does not carry a usable count: a legacy - pre-#5854 sidecar (scenes serialized before the count), a file with no - top-level ``messages`` key at all, a corrupt or truncated prefix, an - unreadable path, or metadata alone that overflows the budget. + parse" -- for every shape that does not carry a usable MARKED count: an + unmarked or wrong-marker file, a legacy sidecar whose count is not in the + prefix, a file with no top-level ``messages`` key at all, a corrupt or + truncated prefix, an unreadable path, or metadata alone that overflows the + budget. """ try: raw = _read_metadata_json_prefix(path) @@ -4308,6 +4341,8 @@ def _prefix_message_count(path): return None if not isinstance(parsed, dict): return None + if parsed.get('_mc_v') != _MESSAGE_COUNT_MARKER: + return None return _parse_nonnegative_int(parsed.get('message_count')) diff --git a/tests/test_save_count_without_full_parse.py b/tests/test_save_count_without_full_parse.py index cacde78d..543ace41 100644 --- a/tests/test_save_count_without_full_parse.py +++ b/tests/test_save_count_without_full_parse.py @@ -233,6 +233,72 @@ def test_file_changed_on_disk_without_a_count_still_backs_up_on_shrink(session_s assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 6 +def test_unmarked_stale_count_from_a_foreign_writer_still_backs_up_on_shrink(session_store): + """A persisted count with no `_mc_v` marker is not trusted. + + The exact shape review (nesquena-hermes, 2026-09-22) pinned: a sidecar + materialized by an OLDER writer can carry a count stale against the + messages array beside it (the recovery writer used to copy a denormalized + state.db count over real rows). Here the foreign file says 2 but really + holds 6; a save of 4 must not read that 2 as "the save grows the file" and + skip the #1558 backup. The marker gate makes the prefix reader refuse the + unmarked count, the full parse sees the real array, and the 6 messages + land in the `.bak`. Remove the gate in `_prefix_message_count` and this + test fails with no backup written. + """ + s = _make(session_store, "x3", 3) + foreign = {"session_id": "x3", "title": "T", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "message_count": 2, "messages": _msgs(6)} # stale count, NO _mc_v + s.path.write_text(json.dumps(foreign, indent=2), encoding="utf-8") + + s.messages = _msgs(4) + s.save() + + bak = s.path.with_suffix(".json.bak") + assert bak.exists(), "an unmarked stale count must not hide the shrink" + assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 6 + assert len(M.Session.load("x3").messages) == 4 + + +def test_wrong_marker_version_still_backs_up_on_shrink(session_store): + """A count marked by a DIFFERENT writer contract is equally untrusted.""" + s = _make(session_store, "x4", 3) + foreign = {"session_id": "x4", "title": "T", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "_mc_v": M._MESSAGE_COUNT_MARKER + 1, + "message_count": 2, "messages": _msgs(6)} + s.path.write_text(json.dumps(foreign, indent=2), encoding="utf-8") + + s.messages = _msgs(4) + s.save() + + bak = s.path.with_suffix(".json.bak") + assert bak.exists(), "a count from another writer version must not hide the shrink" + assert len(json.loads(bak.read_text(encoding="utf-8"))["messages"]) == 6 + + +def test_first_save_remarks_the_file_and_the_fast_path_resumes(session_store, monkeypatch): + """The gate costs one full parse, once. After a save by the current writer + the file carries `_mc_v`, and the cheap prefix path is back in service.""" + s = _make(session_store, "x5", 3) + foreign = {"session_id": "x5", "title": "T", "workspace": str(session_store.parent), + "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "message_count": 2, "messages": _msgs(6)} # unmarked foreign file + s.path.write_text(json.dumps(foreign, indent=2), encoding="utf-8") + + s.messages = _msgs(6) + s.save() # equal-count save; unmarked prefix count ignored, full parse, no shrink + + assert json.loads(s.path.read_text(encoding="utf-8"))["_mc_v"] == M._MESSAGE_COUNT_MARKER + + calls = _spy_full_reads(monkeypatch, s.path) + s.messages = _msgs(8) + s.save() + assert calls["n"] == 0, "a marked file must take the bounded-prefix path again" + assert not s.path.with_suffix(".json.bak").exists() + + def _tool_partial(ts=123): """The exact shape the #2592 collapse recognises (copied from its test).""" return {"role": "assistant", "content": "", "_partial": True, "timestamp": ts, @@ -252,6 +318,7 @@ def test_collapse_self_heal_on_load_still_backs_up_the_pre_collapse_array(sessio fixture carries the count a real save() would have written.""" doc = {"session_id": "h1", "title": "T", "workspace": str(session_store.parent), "model": "glm", "created_at": 1.0, "updated_at": 2.0, + "_mc_v": M._MESSAGE_COUNT_MARKER, "message_count": 5, "messages": [{"role": "user", "content": "run this"}, _tool_partial(), _tool_partial(), _tool_partial(), @@ -290,7 +357,8 @@ def test_same_length_in_place_rewrite_inside_one_mtime_tick_still_backs_up(sessi "model": "glm", "created_at": 1.0, "updated_at": 2.0} def _render(n): - return json.dumps({**base, "message_count": n, "messages": _msgs(n)}, + return json.dumps({**base, "_mc_v": M._MESSAGE_COUNT_MARKER, + "message_count": n, "messages": _msgs(n)}, indent=2, ensure_ascii=False) width = max(len(_render(2)), len(_render(5))) From 6af1a7e7186b8c4c528ae1864c1efd3fc9314c71 Mon Sep 17 00:00:00 2001 From: happy5318 <5318happy@users.noreply.github.com> Date: Thu, 24 Sep 2026 00:45:41 +0800 Subject: [PATCH 08/10] fix(models): derive turn-completion from turn-journal for advanced-past-pending predicate (re-gate 2026-09-23) The suppression predicate _transcript_already_advanced_past_pending() required a run-journal terminal 'completed' state, but its caller in _apply_core_sync_or_error_marker already returns early on that same condition. The predicate branch was therefore unreachable dead code: it could never suppress anything, and stale-cancel recovery could still append a duplicate recovered user row, a journal clone and an interruption marker for a turn that actually finished. Derive the completion evidence from the crash-safe turn journal instead: _turn_journal_records_completion() reports True when the exact-stream 'completed' event is the latest terminal state for the pending stream id. The run journal's terminal write is not durable evidence for this path (it is consumed by the caller's early return), so the turn journal is the only still-observable source. A cancelled/crashed turn (interrupted) keeps recovery active, and completion for another stream of the same session never settles the pending one. Also record the same completion event on the Gateway success writeback path, so Gateway-backed runs have the equivalent completion persistence instead of leaving the run journal's lost terminal write unrecoverable. Regression coverage: - test_turn_journal_helper_reports_completed_for_matching_stream - test_turn_journal_helper_rejects_interrupted_turn - test_advances_past_pending_from_turn_journal_without_run_journal_terminal - test_full_recovery_suppresses_duplicates_on_turn_journal_completion - test_full_recovery_appends_rows_on_turn_journal_completion_loss - test_gateway_chat_worker_records_turn_journal_completion Reverting the predicate gate turns the two new CORE regressions red; reverting the Gateway write turns the Gateway regression red. --- api/gateway_chat.py | 16 + api/models.py | 81 ++++- tests/test_issue6366_stale_cancel_recovery.py | 300 +++++++++++++++++- tests/test_webui_gateway_chat_backend.py | 68 ++++ 4 files changed, 452 insertions(+), 13 deletions(-) diff --git a/api/gateway_chat.py b/api/gateway_chat.py index 35ced6fd..3c4f8bb8 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -38,6 +38,7 @@ from api.config import ( from api.helpers import _redact_text, redact_session_data from api.models import clear_process_wakeup_pause, get_session, merge_session_messages_append_only from api.run_journal import RunJournalWriter, bound_run_journal_snapshot_args +from api.turn_journal import append_turn_journal_event_for_stream logger = logging.getLogger(__name__) @@ -1389,6 +1390,21 @@ def _run_gateway_chat_streaming( if cancel_event.is_set(): _restore_cancelled_success_writeback() return + # #6366 re-gate: record the durable same-stream completion + # event in the crash-safe turn journal. The run journal's + # terminal state is only reached on its own ``stream_end`` + # write path, so a Gateway run whose terminal write is lost + # would otherwise leave no completion evidence at all and + # stale-cancel recovery would re-append a duplicate + # recovered row after the valid final answer. + try: + append_turn_journal_event_for_stream( + session_id, + stream_id, + {"event": "completed", "created_at": time.time()}, + ) + except Exception: + logger.debug("Failed to append completed turn journal event", exc_info=True) success_writeback_committed = True try: from api.goals import evaluate_goal_after_turn, has_active_goal diff --git a/api/models.py b/api/models.py index 1e05e019..1eff0e60 100644 --- a/api/models.py +++ b/api/models.py @@ -2567,12 +2567,18 @@ def _transcript_already_advanced_past_pending(session) -> bool: prose row ("Let me check the logs first.") followed by tool rows that were never followed by a final answer in the transcript passes the predicate as a "completed" turn, but the stream may have died mid- - tool and the run journal was never marked done. 9/22 re-gate: also - require durable same-stream terminal evidence — the run journal - must report ``completed`` (a ``done`` or ``stream_end`` event) for - the pending stream id. Without it, fall through to the normal - journal-recovery path so the partial output is replayed and the - interruption is marked. + tool and the run journal was never marked done. 9/23 re-gate: also + require durable same-stream terminal evidence. The run journal + cannot supply that evidence on this path — the caller in + ``_apply_core_sync_or_error_marker`` already returns early whenever + the run journal reports ``completed`` for the stream, so the branch + that reaches this predicate can never observe that state (which is + why the previous run-journal gate was unreachable dead code). The + evidence therefore has to come from the turn journal, which records + the exact-stream ``completed`` event for the pending turn + (``_turn_journal_records_completion``). Without it, fall through to + the normal journal-recovery path so the partial output is replayed + and the interruption is marked. The suppression is therefore deliberately conservative: whenever the turn's identity, its completion, or its durable terminal evidence @@ -2596,14 +2602,19 @@ def _transcript_already_advanced_past_pending(session) -> bool: continue # Found the pending turn's own user row at ``idx``. A genuine # final answer inside that turn's boundary is necessary but not - # sufficient: the run journal must also have a same-stream - # terminal ``completed`` event, otherwise unflagged interim - # prose ("Let me check the logs first.") followed by tool rows - # can be mistaken for a finished turn. + # sufficient: durable same-stream terminal evidence must also + # exist, otherwise unflagged interim prose + # ("Let me check the logs first.") followed by tool rows can be + # mistaken for a finished turn. if not _pending_turn_has_final_assistant_answer(messages, idx): continue stream_id = getattr(session, 'active_stream_id', None) - if _run_journal_terminal_state(session, stream_id) != 'completed': + # The turn journal — not the run journal — carries the + # completion evidence that is still observable here: the caller + # (``_apply_core_sync_or_error_marker``) already consumed the + # run-journal ``completed`` state on its own early return, so a + # run-journal gate at this depth can never fire. + if not _turn_journal_records_completion(session, stream_id): # No durable same-stream terminal evidence — the turn may # have died mid-tool. Fall through to the journal-recovery # path instead of suppressing it. @@ -2813,6 +2824,54 @@ def _run_journal_terminal_state(session, stream_id: str | None) -> str | None: return str(terminal.get('terminal_state') or '') or None +def _turn_journal_records_completion(session, stream_id: str | None) -> bool: + """Return True when the crash-safe turn journal proves the turn for + ``stream_id`` reached its terminal ``completed`` state. + + The turn journal is written per stream/turn id by the exact-stream + workers and by :mod:`api.routes` on submission, so a ``completed`` + event for the *pending stream id* is completion evidence that + survives a run-journal write failure: the run journal only reaches + ``done`` / ``stream_end`` on its own write path, and a process that + dies between the turn journal write and the run-journal terminal + append leaves the run journal without terminal state while the turn + journal still records that the turn actually finished. + + Bound strictly to ``stream_id`` — a ``completed`` turn belonging to + another stream of the same session never settles this one — and to + the latest turn recorded for that stream, so a turn that was + cancelled or crashed (``interrupted``) keeps recovery active even + when an earlier turn of the same stream completed normally. + """ + if not stream_id: + return False + try: + from api.turn_journal import ( + derive_turn_journal_states, + read_turn_journal, + ) + journal = read_turn_journal(session.session_id) + states, _ = derive_turn_journal_states(journal.get('events') or []) + except Exception: + return False + latest: tuple[float, str, dict] | None = None + for turn_id, event in states.items(): + if str(event.get('stream_id') or '') != str(stream_id): + continue + try: + created_at = float(event.get('created_at') or 0) + except (TypeError, ValueError): + created_at = 0.0 + # ``>=`` on both elements keeps the winner deterministic when two + # turns of the same stream share a ``created_at``. + candidate = (created_at, turn_id, event) + if latest is None or candidate[:2] >= latest[:2]: + latest = candidate + if latest is None: + return False + return latest[2].get('event') == 'completed' + + def _recoverable_unsaved_gateway_terminal_error( session, stream_id: str | None, diff --git a/tests/test_issue6366_stale_cancel_recovery.py b/tests/test_issue6366_stale_cancel_recovery.py index b84957cf..a76b3d73 100644 --- a/tests/test_issue6366_stale_cancel_recovery.py +++ b/tests/test_issue6366_stale_cancel_recovery.py @@ -54,13 +54,17 @@ import pytest import api.models as models from api.models import ( + SESSIONS, Session, _apply_core_sync_or_error_marker, _pending_turn_has_final_assistant_answer, + _run_journal_terminal_state, _transcript_already_advanced_past_pending, _transcript_user_row_is_pending_turn, + _turn_journal_records_completion, ) from api.run_journal import append_run_event +from api.turn_journal import append_turn_journal_event_for_stream @pytest.fixture(autouse=True) @@ -69,9 +73,9 @@ def _isolate_session_state(tmp_path, monkeypatch): session_dir.mkdir() monkeypatch.setattr(models, "SESSION_DIR", session_dir) monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") - models.SESSIONS.clear() + SESSIONS.clear() yield - models.SESSIONS.clear() + SESSIONS.clear() class _FakeSession: @@ -323,6 +327,11 @@ def test_advances_past_pending_when_user_matched_and_settled_assistant_follows() attachments=[{"name": "a.png"}], ) _write_journal(s.session_id, "stream-a", [("done", {})]) + _write_turn_journal( + s.session_id, + "stream-a", + [("submitted", {}), ("completed", {})], + ) assert _transcript_already_advanced_past_pending(s) is True @@ -457,6 +466,11 @@ def test_advances_past_pending_with_unrelated_intervening_rows(): attachments=[], ) _write_journal(s.session_id, "stream-a", [("done", {})]) + _write_turn_journal( + s.session_id, + "stream-a", + [("submitted", {}), ("completed", {})], + ) assert _transcript_already_advanced_past_pending(s) is True @@ -485,6 +499,11 @@ def test_helper_is_pure_read_idempotent(): attachments=[], ) _write_journal(s.session_id, "stream-a", [("done", {})]) + _write_turn_journal( + s.session_id, + "stream-a", + [("submitted", {}), ("completed", {})], + ) first = _transcript_already_advanced_past_pending(s) second = _transcript_already_advanced_past_pending(s) assert first is True @@ -556,6 +575,11 @@ def test_advances_past_pending_with_exact_token_match(): attachments=[{"name": "a.png"}], ) _write_journal(s.session_id, "stream-a", [("done", {})]) + _write_turn_journal( + s.session_id, + "stream-a", + [("submitted", {}), ("completed", {})], + ) assert _transcript_already_advanced_past_pending(s) is True @@ -689,9 +713,281 @@ def test_does_advance_when_journal_marked_done_after_interim_prose(): started_at=222.5, ) _write_journal(s.session_id, "stream-a", [("done", {})]) + _write_turn_journal( + s.session_id, + "stream-a", + [("submitted", {}), ("completed", {})], + ) assert _transcript_already_advanced_past_pending(s) is True +# ── turn-journal completion evidence (9/23 re-gate) ─────────────────── +# +# The suppression predicate used to require a *run-journal* terminal +# ``completed`` state. Its caller in ``_apply_core_sync_or_error_marker`` +# already returns early on that same condition, so by the time the +# predicate ran the run-journal branch could never be true and the +# predicate could never suppress anything. The completion evidence +# therefore has to come from the turn journal — the crash-safe journal +# that records the exact-stream ``completed`` event for the pending turn. + + +_TURN_JOURNAL_CLOCK = {"tick": 0} + + +def _write_turn_journal(session_id, stream_id, events): + """Append events to the turn journal for ``stream_id``, reusing the + stream's turn id exactly like ``append_turn_journal_event_for_stream`` + does in production. + + ``created_at`` is taken from a module-level monotonic clock so events + appended across several calls — and across turn ids — keep a strict + ordering; a per-call sequence would hand the same timestamp to events + of different turns, leaving the derived "latest turn per stream" + ambiguous. + """ + written = [] + for event_name, payload in events: + _TURN_JOURNAL_CLOCK["tick"] += 1 + written.append( + append_turn_journal_event_for_stream( + session_id, + stream_id, + { + "event": event_name, + "created_at": 1_700_000_000.0 + _TURN_JOURNAL_CLOCK["tick"], + **payload, + }, + ) + ) + return written + + +def test_turn_journal_helper_reports_completed_for_matching_stream(): + """The helper reads the turn journal and reports completion only for + the stream whose ``completed`` event was recorded.""" + session_id = "issue6366_turn_journal_helper" + _write_turn_journal( + session_id, + "stream-a", + [("submitted", {}), ("completed", {})], + ) + probe = _FakeSession( + pending="hello world", + messages=[], + stream_id="stream-a", + started_at=222.5, + session_id=session_id, + ) + assert _turn_journal_records_completion(probe, "stream-a") is True + assert _turn_journal_records_completion(probe, "stream-b") is False + + +def test_turn_journal_helper_rejects_interrupted_turn(): + """An ``interrupted`` turn is not a completed turn, even when the + journal also carries an earlier ``completed`` event for another + turn of the same stream.""" + session_id = "issue6366_turn_journal_interrupted" + _write_turn_journal( + session_id, + "stream-a", + [("submitted", {}), ("completed", {})], + ) + _write_turn_journal( + session_id, + "stream-a", + [("submitted", {}), ("interrupted", {"reason": "cancelled"})], + ) + probe = _FakeSession( + pending="hello world", + messages=[], + stream_id="stream-a", + started_at=222.5, + session_id=session_id, + ) + assert _turn_journal_records_completion(probe, "stream-a") is False + + +def test_advances_past_pending_from_turn_journal_without_run_journal_terminal(): + """9/23 re-gate root case: the run journal never wrote a terminal + event (the stream died before ``done``/``stream_end`` reached it), + but the turn journal recorded the exact-stream ``completed`` event. + + The transcript alone proves the turn produced a genuine final + answer, and the turn journal proves the turn reached its terminal + state, so the recovery path must suppress the duplicate recovered + user row + journal clone + interruption marker. + """ + session_id = "issue6366_turn_journal_completed" + s = _FakeSession( + pending="hello world", + messages=[ + { + "role": "user", + "content": "hello world", + "timestamp": 222, + "_source": "webui", + "attachments": [{"name": "a.png"}], + "_active_turn_token": "stream-a:222.5", + }, + {"role": "assistant", "content": "answer", "timestamp": 223}, + ], + stream_id="stream-a", + started_at=222.5, + source="webui", + attachments=[{"name": "a.png"}], + session_id=session_id, + ) + # The run journal holds visible progress but NO terminal event. + _write_journal( + session_id, + "stream-a", + [ + ("token", {"text": "answer"}), + ], + ) + # The turn journal has the exact-stream completion event. + _write_turn_journal( + session_id, + "stream-a", + [("submitted", {}), ("completed", {})], + ) + assert _run_journal_terminal_state(s, "stream-a") != "completed" + assert _turn_journal_records_completion(s, "stream-a") is True + assert _transcript_already_advanced_past_pending(s) is True + + +def test_full_recovery_suppresses_duplicates_on_turn_journal_completion(tmp_path): + """Full-path 9/23 re-gate regression: the run journal never wrote a + terminal state, but the turn journal recorded the exact-stream + ``completed`` event. Recovery must clear the stale pending state + without appending the recovered user row, the journal clone, or the + interruption marker after the valid final answer.""" + session_id = "issue6366_full_recovery_turn_journal_completed" + stream_id = "stream-turn-journal-completed" + started_at = 1_700_002_100.5 + + session = Session( + session_id=session_id, + title="Completed per turn journal, silent run journal", + messages=[ + { + "role": "user", + "content": "summarise the diff", + "timestamp": int(started_at), + "_source": "webui", + "attachments": [], + "_active_turn_token": f"{stream_id}:{started_at:.17g}", + }, + {"role": "assistant", "content": "The diff adds a guard."}, + ], + context_messages=[ + {"role": "user", "content": "summarise the diff"}, + {"role": "assistant", "content": "The diff adds a guard."}, + ], + pending_user_message="summarise the diff", + pending_started_at=started_at, + pending_user_source="webui", + pending_attachments=[], + active_stream_id=stream_id, + ) + # The run journal has visible progress but no terminal event, so the + # line-3671 early return cannot fire and the suppression predicate + # is now the only gate that can keep the transcript clean. + _write_journal( + session_id, + stream_id, + [("token", {"text": "The diff adds a guard."})], + ) + # The turn journal recorded the exact-stream terminal completion. + _write_turn_journal( + session_id, + stream_id, + [("submitted", {}), ("completed", {})], + ) + before = json.dumps(session.messages, ensure_ascii=False) + + assert _apply_core_sync_or_error_marker( + session, + tmp_path / "missing-core-transcript.json", + stream_id_for_recheck=stream_id, + ) is True + + # Nothing was appended after the valid final answer. + assert json.dumps(session.messages, ensure_ascii=False) == before + assert not any( + message.get("_recovered") or message.get("_recovered_from_run_journal") + for message in session.messages + ) + assert not any(message.get("_error") for message in session.messages) + # The stale pending state was cleared. + assert session.pending_user_message is None + assert session.active_stream_id is None + assert session.pending_started_at is None + assert session.pending_user_source is None + assert session.pending_attachments == [] + + +def test_full_recovery_appends_rows_on_turn_journal_completion_loss(tmp_path): + """Mirror control: when neither the run journal nor the turn journal + has terminal completion evidence, the same transcript shape must + still recover — the suppression must not widen to cover a turn whose + completion can no longer be proven.""" + session_id = "issue6366_full_recovery_no_turn_journal_completion" + stream_id = "stream-no-turn-journal-completion" + started_at = 1_700_002_200.5 + + session = Session( + session_id=session_id, + title="Silent run journal, no turn journal completion", + messages=[ + { + "role": "user", + "content": "summarise the diff", + "timestamp": int(started_at), + "_source": "webui", + "attachments": [], + "_active_turn_token": f"{stream_id}:{started_at:.17g}", + }, + {"role": "assistant", "content": "The diff adds a guard."}, + ], + context_messages=[ + {"role": "user", "content": "summarise the diff"}, + ], + pending_user_message="summarise the diff", + pending_started_at=started_at, + pending_user_source="webui", + pending_attachments=[], + active_stream_id=stream_id, + ) + # Neither journal carries terminal completion evidence. + _write_journal( + session_id, + stream_id, + [("token", {"text": "The diff adds a guard."})], + ) + before = json.dumps(session.messages, ensure_ascii=False) + + assert _apply_core_sync_or_error_marker( + session, + tmp_path / "missing-core-transcript.json", + stream_id_for_recheck=stream_id, + ) is True + + # Recovery ran: the journal replay is visible again and the turn is + # marked interrupted instead of being silently declared finished. + assert json.dumps(session.messages, ensure_ascii=False) != before + assert any( + message.get("_recovered_from_run_journal") + and "The diff adds a guard." in str(message.get("content", "")) + for message in session.messages + ) + assert any( + message.get("_error") and message.get("type") == "interrupted" + for message in session.messages + ) + + # ── full _apply_core_sync_or_error_marker recovery cases ──────────────── # # Neither guard rail above is reachable through the predicate in diff --git a/tests/test_webui_gateway_chat_backend.py b/tests/test_webui_gateway_chat_backend.py index 0e7d2830..eab194d8 100644 --- a/tests/test_webui_gateway_chat_backend.py +++ b/tests/test_webui_gateway_chat_backend.py @@ -24,6 +24,7 @@ from api.gateway_chat import ( webui_chat_backend_mode, webui_gateway_chat_enabled, ) +from api.turn_journal import derive_turn_journal_states, read_turn_journal def test_gateway_chat_backend_is_default_off_for_truthy_values(): @@ -396,6 +397,73 @@ def test_gateway_chat_worker_translates_sse_and_persists_session(tmp_path, monke assert all(len(item) == 3 and item[2] for item in events) +def test_gateway_chat_worker_records_turn_journal_completion(tmp_path, monkeypatch): + """#6366 re-gate: a successful Gateway run must record durable + same-stream completion evidence in the crash-safe turn journal. + + The stale-cancel recovery predicate derives its completion evidence + from the turn journal, so a Gateway run that persisted its final + answer but lost the run journal's terminal write would otherwise + have no completion evidence at all — and recovery would re-append a + duplicate recovered row after the valid final answer. + """ + session_dir = tmp_path / "sessions" + session_dir.mkdir() + monkeypatch.setattr(models, "SESSION_DIR", session_dir) + monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json") + monkeypatch.setattr(models, "SESSIONS", OrderedDict()) + + class FakeResponse: + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def __iter__(self): + yield b'data: {"choices":[{"delta":{"content":"hello"}}]}\n\n' + yield b"data: [DONE]\n\n" + + monkeypatch.setenv("HERMES_WEBUI_GATEWAY_BASE_URL", "http://gateway.local") + monkeypatch.setenv("HERMES_WEBUI_GATEWAY_API_KEY", "secret-token") + monkeypatch.setattr( + gateway_chat.urllib.request, + "urlopen", + lambda req, timeout=0: FakeResponse(), + ) + + s = new_session() + stream_id = "stream-gateway-turn-journal-completed" + s.active_stream_id = stream_id + s.pending_user_message = "Say hello" + s.pending_attachments = [] + s.pending_started_at = 456 + s.save() + channel = create_stream_channel() + STREAMS[stream_id] = channel + + gateway_chat._run_gateway_chat_streaming( + s.session_id, + "Say hello", + "test-model", + str(tmp_path), + stream_id, + [], + ) + + journal = read_turn_journal(s.session_id, session_dir=session_dir) + states, _ = derive_turn_journal_states(journal.get("events") or []) + stream_events = [ + event + for event in states.values() + if str(event.get("stream_id") or "") == stream_id + ] + assert stream_events, "expected turn journal events for the gateway stream" + assert any( + event.get("event") == "completed" for event in stream_events + ), "the gateway success writeback must record a completed turn journal event" + + def test_gateway_chat_worker_classifies_terminal_provider_error_without_text(tmp_path, monkeypatch): """Gateway terminal errors must survive an empty assistant stream.""" from unittest.mock import MagicMock From 4f88aa1026668a9346f9d5b274d03772404fad59 Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Wed, 23 Sep 2026 18:36:16 +0000 Subject: [PATCH 09/10] docs(changelog): #7682 --- CHANGELOG.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 41bb356e..c2004fe7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,12 @@ ### Fixed +- **A cancelled or recovered turn no longer duplicates an answer that was already saved (#6366).** + When a completed assistant turn had been persisted and a later cancel/recovery path ran for the + same pending turn, recovery could append a duplicate user turn plus a `_partial` clone of the + journal. Recovery now stops once the transcript has already advanced past the pending turn. + (#7682 by @happy5318) + - **The live stream reports which model actually served the turn.** A new additive `runtime_model` SSE event carries the model (and provider, when known) that the Agent reported while producing output, separately from the model that was requested. It is journaled for replay, is exposed as From 19693918fabbc9c26cf9a80094f8eec8d0dc4cae Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Wed, 23 Sep 2026 18:36:16 +0000 Subject: [PATCH 10/10] docs(changelog): #7578 --- CHANGELOG.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 41bb356e..e8e2ec80 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,12 @@ ### Fixed +- **Saving a very large session no longer reads and parses the whole file just to count messages.** + The #1558 backup safeguard in `Session.save()` needs the on-disk message count; it obtained it by + loading the entire sidecar. On a real 203 MB / 266,940-message session that made every save + expensive. The count is now taken without a full parse, and the shrink-backup behaviour is + unchanged. (#7578 by @rodrigogs) + - **The live stream reports which model actually served the turn.** A new additive `runtime_model` SSE event carries the model (and provider, when known) that the Agent reported while producing output, separately from the model that was requested. It is journaled for replay, is exposed as