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