jinghui/hermes-webui
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: bind gateway terminal recovery to run identity

Browse Source
This commit is contained in:
Frank Song committed 2026-07-31 17:31:12 +08:00
1 parent ded12e48c5
commit dadffe0e02
3 files changed
+580 -142
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

api/models.py
+246 -99
View File
@@ -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
api/run_journal.py
+23 -5
View File
@@ -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),
tests/test_session_sidecar_repair.py
+311 -38
View File
@@ -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,