diff --git a/api/gateway_chat.py b/api/gateway_chat.py index c66175d5..0dfa7227 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__) @@ -1404,6 +1405,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 0322a97b..ca1da72d 100644 --- a/api/models.py +++ b/api/models.py @@ -2442,6 +2442,191 @@ 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 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. + + 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 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/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 + 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. + """ + pending_text = getattr(session, 'pending_user_message', None) + if not pending_text: + return False + 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 _transcript_user_row_is_pending_turn(message, session, pending_token): + 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: 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) + # 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. + return False + return True + 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): @@ -2643,6 +2828,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, @@ -3519,6 +3752,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..a76b3d73 --- /dev/null +++ b/tests/test_issue6366_stale_cancel_recovery.py @@ -0,0 +1,1424 @@ +"""Tests for #6366: stale-cancel recovery must not append duplicate +rows when the durable transcript has already advanced past the +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`` +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 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 ( + 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) +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") + SESSIONS.clear() + yield + SESSIONS.clear() + + +class _FakeSession: + """Minimal stand-in for api.models.Session — only the fields the + new helpers read.""" + + def __init__( + self, + pending=None, + messages=None, + *, + stream_id=None, + 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 + 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 ─────────────── + + +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; + the journal is marked ``done`` (terminal_state ``completed``) so + the suppression gate is satisfied.""" + 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"}], + ) + _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 + + +def test_advances_past_pending_ignores_partial_clone_after_user(): + """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 genuine final visible + answer represents a real advance.""" + s = _FakeSession( + pending="hello world", + messages=[ + { + "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 + + +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, + "_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 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, + "_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 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": "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 + + +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(): + """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, and the journal is marked ``done`` so the durable + terminal evidence gate is satisfied.""" + s = _FakeSession( + pending="hello world", + messages=[ + { + "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=[], + ) + _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 + + +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. Journal + is marked ``done`` so the durable terminal evidence gate passes + on both invocations.""" + s = _FakeSession( + pending="hello world", + messages=[ + { + "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=[], + ) + _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 + assert 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_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 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 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"}], + ) + assert _transcript_already_advanced_past_pending(s) is False + + +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=[ + { + "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"}], + ) + _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 + + +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 + + +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", {})]) + _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 +# 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 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" + 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, + ) + # 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( + 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_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 diff --git a/tests/test_webui_gateway_chat_backend.py b/tests/test_webui_gateway_chat_backend.py index 226fee6c..88b78f6b 100644 --- a/tests/test_webui_gateway_chat_backend.py +++ b/tests/test_webui_gateway_chat_backend.py @@ -25,6 +25,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(): @@ -397,6 +398,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