From 7cdc372d0b6c6ff6deac396d60d8a93da71e0c5c Mon Sep 17 00:00:00 2001 From: happy5318 Date: Sun, 20 Sep 2026 22:15:07 +0800 Subject: [PATCH 1/6] 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 2/6] 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 3/6] 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 4/6] 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 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 5/6] 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 6/6] 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