From dadffe0e02e9527c1724c43b08163e30f4c4fbd6 Mon Sep 17 00:00:00 2001 From: Frank Song Date: Fri, 31 Jul 2026 17:31:12 +0800 Subject: [PATCH] fix: bind gateway terminal recovery to run identity --- api/models.py | 345 ++++++++++++++++++-------- api/run_journal.py | 28 ++- tests/test_session_sidecar_repair.py | 349 ++++++++++++++++++++++++--- 3 files changed, 580 insertions(+), 142 deletions(-) diff --git a/api/models.py b/api/models.py index 3fe1e228..dd3b5b33 100644 --- a/api/models.py +++ b/api/models.py @@ -2477,32 +2477,57 @@ def _run_journal_has_visible_output(session, stream_id: str | None) -> bool: return False +def _run_journal_event_owns_run( + event, + session_id: str, + stream_id: str | None, +) -> bool: + if not isinstance(event, dict) or not stream_id: + return False + seq = event.get('seq') + if isinstance(seq, bool) or not isinstance(seq, int) or seq < 1: + return False + return ( + event.get('session_id') == session_id + and event.get('run_id') == stream_id + and event.get('event_id') == f"{stream_id}:{seq}" + ) + + def _run_journal_terminal_state(session, stream_id: str | None) -> str | None: if not stream_id: return None try: - from api.run_journal import latest_run_summary - summary = latest_run_summary(session.session_id, stream_id) + from api.run_journal import ( + read_run_events, + select_authoritative_terminal_event, + ) + journal = read_run_events(session.session_id, stream_id) + terminal = select_authoritative_terminal_event(journal.get('events') or []) except Exception: return None - if not summary.get('terminal'): + if ( + not _run_journal_event_owns_run( + terminal, session.session_id, stream_id, + ) + or terminal.get('terminal') is not True + ): return None - return str(summary.get('terminal_state') or '') or None + return str(terminal.get('terminal_state') or '') or None -def _materialize_unsaved_gateway_terminal_error(session, stream_id: str | None) -> bool: - """Recover a gateway terminal error whose session save was not durable. - - Gateway terminal-error events carry the redacted in-memory transcript so a - failed sidecar save does not turn a specific provider error into a generic - restart marker. Only an exact persisted-session match is authoritative; - every other marker state is treated as non-durable and requires the - embedded terminal error to be validated and materialized locally. - """ +def _recoverable_unsaved_gateway_terminal_error( + session, + stream_id: str | None, +) -> dict | None: + """Return one validated current-turn terminal error from the run journal.""" if not stream_id: - return False + return None try: - from api.run_journal import read_run_events + from api.run_journal import ( + read_run_events, + select_authoritative_terminal_event, + ) journal = read_run_events(session.session_id, stream_id) except Exception: logger.debug( @@ -2511,54 +2536,162 @@ def _materialize_unsaved_gateway_terminal_error(session, stream_id: str | None) stream_id, exc_info=True, ) + return None + + event = select_authoritative_terminal_event(journal.get('events') or []) + if ( + not isinstance(event, dict) + or event.get('event') != 'apperror' + or event.get('type') != 'apperror' + or event.get('terminal') is not True + ): + return None + if ( + not _run_journal_event_owns_run( + event, session.session_id, stream_id, + ) + ): + return None + expected_event_id = event['event_id'] + + payload = event.get('payload') + if not isinstance(payload, dict) or payload.get('session_id') != session.session_id: + return None + embedded_session = payload.get('session') + if ( + not isinstance(embedded_session, dict) + or embedded_session.get('session_id') != session.session_id + ): + return None + persisted_id = payload.get('terminal_session_persisted_session_id') + if ( + payload.get('terminal_session_persisted') is True + and persisted_id == session.session_id + ): + return None + + embedded_messages = embedded_session.get('messages') + if not isinstance(embedded_messages, list): + return None + current_user_idx = next( + ( + idx + for idx in range(len(embedded_messages) - 1, -1, -1) + if isinstance(embedded_messages[idx], dict) + and embedded_messages[idx].get('role') == 'user' + ), + None, + ) + if current_user_idx is None: + return None + candidate = next( + ( + embedded_messages[idx] + for idx in range(len(embedded_messages) - 1, current_user_idx, -1) + if isinstance(embedded_messages[idx], dict) + and embedded_messages[idx].get('role') == 'assistant' + ), + None, + ) + if ( + not isinstance(candidate, dict) + or candidate.get('_error') is not True + or not isinstance(candidate.get('content'), str) + or not candidate.get('content').strip() + ): + return None + return { + 'event_id': expected_event_id, + 'stream_id': stream_id, + 'message': dict(candidate), + } + + +def _pending_recovery_turn_start(session) -> int | None: + pending_text = getattr(session, 'pending_user_message', None) + if not pending_text: + return None + for idx in range(len(session.messages or []) - 1, -1, -1): + message = session.messages[idx] + if _message_matches_pending_checkpoint( + message, + pending_text, + session.pending_started_at, + session.pending_user_source, + session.pending_attachments, + ) or _message_matches_pending_text(message, pending_text): + return idx + return None + + +def _materialize_unsaved_gateway_terminal_error( + session, + stream_id: str | None, + recovery: dict | None = None, +) -> bool: + """Place the validated current-turn gateway error at the transcript tail.""" + recovery = recovery or _recoverable_unsaved_gateway_terminal_error( + session, stream_id, + ) + if not isinstance(recovery, dict): + return False + event_id = recovery.get('event_id') + candidate = recovery.get('message') + if not event_id or not isinstance(candidate, dict): return False - for event in journal.get('events') or []: - if not isinstance(event, dict) or event.get('event') != 'apperror': - continue - payload = event.get('payload') - if not isinstance(payload, dict): - continue - embedded_session = payload.get('session') - if not isinstance(embedded_session, dict): - continue - payload_session_id = payload.get('session_id') - embedded_session_id = embedded_session.get('session_id') - if payload_session_id is not None and str(payload_session_id) != session.session_id: - continue - if embedded_session_id is not None and str(embedded_session_id) != session.session_id: - continue - persisted_id = payload.get('terminal_session_persisted_session_id') + for existing in session.messages or []: if ( - payload.get('terminal_session_persisted') is True - and str(persisted_id or '') == session.session_id + isinstance(existing, dict) + and existing.get('_recovered_event_id') == event_id ): - return False - embedded_messages = embedded_session.get('messages') - if not isinstance(embedded_messages, list): - continue - for candidate in embedded_messages: - if not isinstance(candidate, dict) or candidate.get('role') != 'assistant': - continue - if candidate.get('_error') is not True: - continue - content = candidate.get('content') - if not isinstance(content, str) or not content.strip(): - continue - if any( - isinstance(existing, dict) - and existing.get('role') == 'assistant' - and existing.get('_error') is True - and existing.get('content') == content - for existing in session.messages or [] - ): - return True - recovered = dict(candidate) - recovered['_recovered_from_run_journal'] = True - recovered['_recovered_stream_id'] = stream_id - session.messages.append(recovered) return True - return False + + turn_start = _pending_recovery_turn_start(session) + if turn_start is not None: + for existing in reversed((session.messages or [])[turn_start + 1:]): + if not isinstance(existing, dict): + continue + existing_stream = existing.get('_recovered_stream_id') + if existing_stream and existing_stream != stream_id: + continue + if ( + existing.get('role') == 'assistant' + and existing.get('_error') is True + and existing.get('content') == candidate.get('content') + ): + existing['_recovered_from_run_journal'] = True + existing['_recovered_stream_id'] = stream_id + existing['_recovered_event_id'] = event_id + return True + + recovered = dict(candidate) + recovered['_recovered_from_run_journal'] = True + recovered['_recovered_stream_id'] = stream_id + recovered['_recovered_event_id'] = event_id + session.messages.append(recovered) + return True + + +def _recover_journaled_output_and_terminal_error( + session, + stream_id: str | None, + *, + dedupe_existing: bool = False, + terminal_recovery: dict | None = None, +) -> tuple[bool, bool]: + """Recover readable activity first, then append its authoritative terminal error.""" + recovered_output = _append_journaled_partial_output( + session, + stream_id, + dedupe_existing=dedupe_existing, + ) + terminal_error_recovered = _materialize_unsaved_gateway_terminal_error( + session, + stream_id, + terminal_recovery, + ) + return recovered_output, terminal_error_recovered def _journal_is_still_arriving(session, stream_id: str | None) -> bool: @@ -2907,11 +3040,11 @@ def _append_journaled_partial_output( # onto the marker: `_journal_retry_stream_id`, `_journal_retry_attempts`, # `_journal_retry_first_seen_ts`. # * Every `get_session()` call that returns the full session checks the -# latest assistant marker; if the flag is set it re-runs -# `_append_journaled_partial_output` with `dedupe_existing=True`. On -# success the marker is promoted in place to the recovered-output -# wording, the journaled rows are reordered to sit above the marker, -# and all retry meta is stripped. If the journal is still missing or +# latest assistant marker; if the flag is set it re-runs journaled output +# and terminal-error recovery with `dedupe_existing=True`. On success, +# journaled rows move above the marker. A specific gateway terminal error +# replaces the marker; otherwise the marker is promoted to recovered-output +# wording and its retry meta is stripped. If the journal is still missing or # zero-byte, the retry is a no-op and does not consume attempt budget. # Terminal/non-useful journals consume attempt budget and can demote # immediately at the max-attempt cap. @@ -3050,7 +3183,7 @@ def _retry_journal_recovery_in_place( ) -> bool: """Re-attempt run-journal recovery for the most recent pending marker. - Returns True if the marker was promoted to the recovered-output wording. + Returns True if journal output or a specific terminal error resolved the marker. Never raises — caller is best-effort. """ try: @@ -3104,29 +3237,39 @@ def _retry_journal_recovery_in_place( exc_info=True, ) return False - tail_len_before = len(session.messages) - ok = _append_journaled_partial_output( - session, stream_id, dedupe_existing=True, + recovered_output, terminal_error_recovered = ( + _recover_journaled_output_and_terminal_error( + session, + stream_id, + dedupe_existing=True, + ) ) - if ok: - msg['content'] = _INTERRUPTED_RECOVERED_WORDING - _strip_journal_retry_meta(msg) + if recovered_output or terminal_error_recovered: + if not terminal_error_recovered: + msg['content'] = _INTERRUPTED_RECOVERED_WORDING + _strip_journal_retry_meta(msg) # The journaled rows were appended at the end of messages; - # only the rows past the previous tail count as "newly - # journaled" and need to move above the marker. - _ = tail_len_before # informational; helper below scans + # move them above the marker before either retaining its + # interrupted wording or replacing it with a specific terminal + # error from that same stream. _reorder_journal_tail_above_marker(session, idx) + if terminal_error_recovered: + session.messages = [ + message + for message in session.messages + if message is not msg + ] try: session.save(touch_updated_at=False) except Exception: logger.debug( - "save() failed while promoting marker for session %s", + "save() failed while applying lazy journal recovery for session %s", getattr(session, 'session_id', '?'), exc_info=True, ) logger.info( - "Session %s: lazy journal-recovery promoted marker for " - "stream %s after %d attempts", + "Session %s: lazy journal-recovery applied stream %s " + "after %d attempts", getattr(session, 'session_id', '?'), stream_id, attempts, @@ -3203,6 +3346,10 @@ def _apply_core_sync_or_error_marker( return False if require_stream_dead and session.active_stream_id in _active_stream_ids(): return False + _stream_id = stream_id_for_recheck or session.active_stream_id + _terminal_recovery = _recoverable_unsaved_gateway_terminal_error( + session, _stream_id, + ) # When messages is already non-empty, do not overwrite history from any core # transcript. The pending user turn may still be the only durable copy of a @@ -3223,7 +3370,6 @@ def _apply_core_sync_or_error_marker( session.messages[-1], session.pending_user_message, ) - _stream_id = stream_id_for_recheck or session.active_stream_id _pending_started_at = session.pending_started_at if _run_journal_terminal_state(session, _stream_id) == 'completed': if not (_already_checkpointed or _latest_user_matches_pending_text(session.messages, session.pending_user_message)): @@ -3260,12 +3406,12 @@ def _apply_core_sync_or_error_marker( if session.pending_attachments: recovered['attachments'] = list(session.pending_attachments) _append_recovered_turn_to_context(session, recovered) - terminal_error_recovered = _materialize_unsaved_gateway_terminal_error( - session, _stream_id, - ) - recovered_output = _append_journaled_partial_output( - session, - _stream_id, + recovered_output, terminal_error_recovered = ( + _recover_journaled_output_and_terminal_error( + session, + _stream_id, + terminal_recovery=_terminal_recovery, + ) ) session.active_stream_id = None session.pending_user_message = None @@ -3294,7 +3440,6 @@ def _apply_core_sync_or_error_marker( core = json.load(f) core_messages = core.get('messages', []) if core_messages: - _stream_id = stream_id_for_recheck or session.active_stream_id session.messages = core_messages session.tool_calls = core.get('tool_calls', []) for field in ('input_tokens', 'output_tokens', 'estimated_cost'): @@ -3318,16 +3463,19 @@ def _apply_core_sync_or_error_marker( if ( _pending_text and not _tail_user_already_checkpointed - and _run_journal_has_visible_output(session, _stream_id) + and ( + _run_journal_has_visible_output(session, _stream_id) + or _terminal_recovery is not None + ) ): _append_recovered_pending_turn(session, timestamp=_recovered_ts) - terminal_error_recovered = _materialize_unsaved_gateway_terminal_error( - session, _stream_id, - ) - recovered_output = _append_journaled_partial_output( - session, - _stream_id, - dedupe_existing=True, + recovered_output, terminal_error_recovered = ( + _recover_journaled_output_and_terminal_error( + session, + _stream_id, + dedupe_existing=True, + terminal_recovery=_terminal_recovery, + ) ) _pending_started_at = session.pending_started_at session.active_stream_id = None @@ -3372,14 +3520,13 @@ def _apply_core_sync_or_error_marker( if isinstance(session.pending_started_at, (int, float)) and session.pending_started_at > 0: _recovered_ts = int(session.pending_started_at) _append_recovered_pending_turn(session, timestamp=_recovered_ts) - terminal_error_recovered = _materialize_unsaved_gateway_terminal_error( - session, stream_id_for_recheck or session.active_stream_id, + recovered_output, terminal_error_recovered = ( + _recover_journaled_output_and_terminal_error( + session, + _stream_id, + terminal_recovery=_terminal_recovery, + ) ) - recovered_output = _append_journaled_partial_output( - session, - stream_id_for_recheck or session.active_stream_id, - ) - _stream_id = stream_id_for_recheck or session.active_stream_id _pending_started_at = session.pending_started_at session.active_stream_id = None session.pending_user_message = None diff --git a/api/run_journal.py b/api/run_journal.py index cabdacfa..ab8d0a7e 100644 --- a/api/run_journal.py +++ b/api/run_journal.py @@ -484,14 +484,32 @@ def read_run_events( } +def select_authoritative_terminal_event(events: Iterable[dict]) -> dict | None: + """Return the terminal event that owns the run's settled outcome. + + ``stream_end`` is transport closure, so a preceding semantic terminal event + (done, cancel, or error) remains authoritative. Among semantic terminal + events, the latest journal row wins. + """ + terminal_events = [ + event + for event in events + if isinstance(event, dict) and event.get("terminal") + ] + return next( + ( + event + for event in reversed(terminal_events) + if event.get("event") != "stream_end" + ), + terminal_events[-1] if terminal_events else None, + ) + + def _summary_from_events(session_id: str, run_id: str, events: Iterable[dict]) -> dict: ordered = [event for event in events if isinstance(event, dict)] last = ordered[-1] if ordered else None - terminal_events = [event for event in ordered if event.get("terminal")] - terminal = next( - (event for event in reversed(terminal_events) if event.get("event") != "stream_end"), - terminal_events[-1] if terminal_events else None, - ) + terminal = select_authoritative_terminal_event(ordered) status = terminal.get("terminal_state") if terminal else ("running" if ordered else "unknown") return { "session_id": str(session_id), diff --git a/tests/test_session_sidecar_repair.py b/tests/test_session_sidecar_repair.py index 04930576..e8bb6a6d 100644 --- a/tests/test_session_sidecar_repair.py +++ b/tests/test_session_sidecar_repair.py @@ -21,7 +21,7 @@ from api.models import ( import api.config as config import api.streaming as streaming import api.profiles as profiles -from api.run_journal import append_run_event +from api.run_journal import RunJournalWriter, append_run_event # ── Fixtures ──────────────────────────────────────────────────────────────── @@ -118,6 +118,66 @@ def _register_active_run(stream_id): config.register_active_run(stream_id, session_id="stale_sid", phase="running") +def _journal_gateway_terminal_save_failure( + monkeypatch, + stale_session, + *, + live_messages, + partial_text="Current partial output.", + terminal_error="gateway exploded", +): + """Compose the real gateway producer payload and append it through the real journal.""" + import api.gateway_chat as gateway_chat + + stream_id = stale_session.active_stream_id + live_session = Session( + session_id=stale_session.session_id, + title=stale_session.title, + workspace=stale_session.workspace, + model=stale_session.model, + model_provider=stale_session.model_provider, + messages=[dict(message) for message in live_messages], + context_messages=[dict(message) for message in live_messages], + ) + live_session.pending_user_message = stale_session.pending_user_message + live_session.pending_started_at = stale_session.pending_started_at + live_session.pending_attachments = list(stale_session.pending_attachments) + live_session.pending_user_source = stale_session.pending_user_source + live_session.active_stream_id = stream_id + + def fail_save(*_args, **_kwargs): + raise OSError("forced gateway terminal error save failure") + + monkeypatch.setattr(live_session, "save", fail_save) + monkeypatch.setattr(gateway_chat, "get_session", lambda _sid: live_session) + monkeypatch.setitem(config.STREAM_PARTIAL_TEXT, stream_id, partial_text) + monkeypatch.setitem(config.STREAM_REASONING_TEXT, stream_id, "") + monkeypatch.setitem(config.STREAM_LIVE_TOOL_CALLS, stream_id, []) + + payload = gateway_chat._settle_gateway_terminal_error( + live_session.session_id, + stream_id, + live_session.workspace, + live_session.model, + live_session.model_provider, + terminal_error, + ) + assert payload["terminal_session_persisted"] is False + + writer = RunJournalWriter(live_session.session_id, stream_id) + if partial_text: + writer.append_sse_event("token", {"text": partial_text}) + terminal_event = writer.append_sse_event("apperror", payload) + terminal_message = next( + message + for message in reversed(payload["session"]["messages"]) + if isinstance(message, dict) + and message.get("role") == "assistant" + and message.get("_error") is True + ) + return payload, terminal_event, terminal_message + + class TestRepairStalePendingNoDeadlock: """_repair_stale_pending uses non-blocking lock acquire so callers that already hold the per-session lock (retry_last, undo_last, cancel_stream) @@ -763,51 +823,264 @@ class TestNonEmptyMessagesPendingCleared: assert "partial output above was recovered" in error_msgs[0]["content"] assert "no agent output was recovered" not in error_msgs[0]["content"] - def test_gateway_terminal_error_transcript_survives_unsaved_recovery( - self, hermes_home, monkeypatch, + @pytest.mark.parametrize("sidecar_shape", ["non_empty", "core", "empty"]) + def test_gateway_terminal_error_cold_recovery_keeps_turn_identity_and_order( + self, hermes_home, monkeypatch, tmp_path, sidecar_shape, ): - """An unsaved gateway terminal error must recover its specific error, - rather than replacing it with the generic restart marker.""" - s = _make_session(messages=[{"role": "user", "content": "existing turn"}]) - s.pending_user_message = "Gateway request" - s.pending_started_at = time.time() - 120 - s.active_stream_id = "gateway_terminal_error_stream" - s.save() - - terminal_message = { + """The exact producer payload must cold-recover after partial output in every branch.""" + sid = f"gateway_terminal_cold_{sidecar_shape}" + stream_id = f"{sid}_stream" + repeated_error = { "role": "assistant", - "content": "**Gateway error:** gateway exploded", - "timestamp": int(time.time()), + "content": "**Error:** gateway exploded", + "timestamp": int(time.time()) - 300, "_error": True, } - append_run_event( - s.session_id, - s.active_stream_id, - "apperror", - { - "session_id": s.session_id, - "terminal_session_persisted": False, - "session": { - "session_id": s.session_id, - "messages": [ - {"role": "user", "content": "Gateway request"}, - terminal_message, - ], - }, - }, + history = [ + {"role": "user", "content": "Earlier gateway request"}, + repeated_error, + ] if sidecar_shape != "empty" else [] + stale = _make_session( + session_id=sid, + workspace=str(tmp_path), + messages=history if sidecar_shape == "non_empty" else [], + ) + stale.pending_user_message = "Current gateway request" + stale.pending_started_at = time.time() - 120 + stale.active_stream_id = stream_id + stale.save() + if sidecar_shape == "core": + _write_core_transcript(hermes_home, sid, history) + + _payload, terminal_event, terminal_message = _journal_gateway_terminal_save_failure( + monkeypatch, + stale, + live_messages=history, + ) + models.SESSIONS.pop(sid, None) + + recovered = models.get_session(sid) + current_errors = [ + message for message in recovered.messages + if message.get("_recovered_event_id") == terminal_event["event_id"] + ] + assert [message["content"] for message in current_errors] == [ + terminal_message["content"] + ] + current_error_idx = recovered.messages.index(current_errors[0]) + current_user_idx = max( + idx for idx, message in enumerate(recovered.messages) + if message.get("role") == "user" + and message.get("content") == "Current gateway request" + ) + partial_idx = next( + idx for idx, message in enumerate(recovered.messages) + if message.get("_recovered_stream_id") == stream_id + and message.get("content") == "Current partial output." + ) + assert current_user_idx < partial_idx < current_error_idx + assert current_error_idx == len(recovered.messages) - 1 + + same_text_errors = [ + message for message in recovered.messages + if message.get("_error") is True + and message.get("content") == terminal_message["content"] + ] + expected_count = 1 if sidecar_shape == "empty" else 2 + assert len(same_text_errors) == expected_count + + @pytest.mark.parametrize( + "malformed_field", + [ + "session_id", + "run_id", + "seq", + "event_id", + "payload_session_id", + "embedded_session_id", + "later_done_event_id", + ], + ) + def test_gateway_terminal_error_rejects_foreign_or_malformed_identity( + self, hermes_home, monkeypatch, tmp_path, malformed_field, + ): + """A journal path is not proof that the event envelope or payload owns the turn.""" + sid = f"gateway_terminal_bad_{malformed_field}" + stream_id = f"{sid}_stream" + stale = _make_session( + session_id=sid, + workspace=str(tmp_path), + messages=[ + {"role": "user", "content": "Earlier request"}, + {"role": "assistant", "content": "Earlier answer"}, + ], + ) + stale.pending_user_message = "Current gateway request" + stale.pending_started_at = time.time() - 120 + stale.active_stream_id = stream_id + stale.save() + + _payload, _terminal_event, terminal_message = _journal_gateway_terminal_save_failure( + monkeypatch, + stale, + live_messages=stale.messages, + ) + if malformed_field == "later_done_event_id": + RunJournalWriter(sid, stream_id).append_sse_event( + "done", {"session_id": sid}, + ) + journal_path = ( + models.SESSION_DIR + / "_run_journal" + / sid + / f"{stream_id}.jsonl" + ) + events = [ + json.loads(line) + for line in journal_path.read_text(encoding="utf-8").splitlines() + ] + terminal = events[-1] + if malformed_field == "session_id": + terminal["session_id"] = "foreign_session" + elif malformed_field == "run_id": + terminal["run_id"] = "foreign_run" + elif malformed_field == "seq": + terminal["seq"] = str(terminal["seq"]) + elif malformed_field == "event_id": + terminal["event_id"] = "foreign_run:999" + elif malformed_field == "payload_session_id": + terminal["payload"].pop("session_id", None) + elif malformed_field == "embedded_session_id": + terminal["payload"]["session"].pop("session_id", None) + else: + terminal["event_id"] = "foreign_run:999" + journal_path.write_text( + "".join( + json.dumps(event, ensure_ascii=False, separators=(",", ":")) + "\n" + for event in events + ), + encoding="utf-8", + ) + models.SESSIONS.pop(sid, None) + + recovered = models.get_session(sid) + assert terminal_message["content"] not in [ + message.get("content") for message in recovered.messages + ] + assert any( + message.get("type") == "interrupted" + for message in recovered.messages ) - core_path = hermes_home / "sessions" / f"session_{s.session_id}.json" - result = _apply_core_sync_or_error_marker( - s, - core_path, - stream_id_for_recheck=s.active_stream_id, + @pytest.mark.parametrize( + ("later_event_name", "expected_error"), + [ + ("apperror", "**Error:** later gateway failure"), + ("done", None), + ("cancel", None), + ("stream_end", "**Error:** gateway exploded"), + ], + ) + def test_gateway_terminal_error_uses_authoritative_latest_terminal_event( + self, hermes_home, monkeypatch, tmp_path, later_event_name, expected_error, + ): + """Later terminal evidence supersedes an older error, except transport-only stream_end.""" + sid = f"gateway_terminal_latest_{later_event_name}" + stream_id = f"{sid}_stream" + stale = _make_session( + session_id=sid, + workspace=str(tmp_path), + messages=[ + {"role": "user", "content": "Earlier request"}, + {"role": "assistant", "content": "Earlier answer"}, + ], + ) + stale.pending_user_message = "Current gateway request" + stale.pending_started_at = time.time() - 120 + stale.active_stream_id = stream_id + stale.save() + payload, first_terminal, first_message = _journal_gateway_terminal_save_failure( + monkeypatch, + stale, + live_messages=stale.messages, ) - assert result is True - error_msgs = [m for m in s.messages if m.get("_error")] - assert [m["content"] for m in error_msgs] == [terminal_message["content"]] - assert all("Response interrupted" not in m["content"] for m in error_msgs) + writer = RunJournalWriter(sid, stream_id) + later_payload = {"session_id": sid} + later_terminal = None + if later_event_name == "apperror": + later_payload = json.loads(json.dumps(payload)) + later_message = next( + message + for message in reversed(later_payload["session"]["messages"]) + if message.get("_error") is True + ) + later_message["content"] = expected_error + later_terminal = writer.append_sse_event("apperror", later_payload) + else: + writer.append_sse_event(later_event_name, later_payload) + models.SESSIONS.pop(sid, None) + + recovered = models.get_session(sid) + recovered_error_contents = [ + message.get("content") + for message in recovered.messages + if message.get("_error") is True + and message.get("type") != "interrupted" + ] + if expected_error is None: + assert first_message["content"] not in recovered_error_contents + else: + assert recovered_error_contents == [expected_error] + expected_event_id = ( + later_terminal["event_id"] + if later_terminal is not None + else first_terminal["event_id"] + ) + assert recovered.messages[-1].get("_recovered_event_id") == expected_event_id + + def test_late_gateway_terminal_journal_replaces_pending_retry_marker( + self, hermes_home, monkeypatch, tmp_path, + ): + """A journal that appears after first repair must still recover its specific error.""" + sid = "gateway_terminal_late_journal" + stream_id = f"{sid}_stream" + stale = _make_session( + session_id=sid, + workspace=str(tmp_path), + messages=[ + {"role": "user", "content": "Earlier request"}, + {"role": "assistant", "content": "Earlier answer"}, + ], + ) + stale.pending_user_message = "Current gateway request" + stale.pending_started_at = time.time() - 120 + stale.active_stream_id = stream_id + stale.save() + models.SESSIONS.pop(sid, None) + + first_recovery = models.get_session(sid) + assert any( + message.get("_pending_journal_recovery") is True + for message in first_recovery.messages + ) + _payload, terminal_event, terminal_message = _journal_gateway_terminal_save_failure( + monkeypatch, + stale, + live_messages=stale.messages, + ) + + recovered = models.get_session(sid) + assert all( + message.get("_pending_journal_recovery") is not True + for message in recovered.messages + ) + assert all( + message.get("type") != "interrupted" + for message in recovered.messages + ) + assert recovered.messages[-1].get("_recovered_event_id") == terminal_event["event_id"] + assert recovered.messages[-1]["content"] == terminal_message["content"] def test_journal_recovery_restores_reasoning_only_as_display_metadata( self, hermes_home, monkeypatch,