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] 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