From b438a8989e1c94121134b389bb3e5cbbe160ce72 Mon Sep 17 00:00:00 2001 From: briandevans <252620095+briandevans@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:48:30 -0700 Subject: [PATCH 01/18] fix(gateway): replay cap-dropped transcript spool files in drop order on restart MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit recover_pending_to_db walks sorted(flush_dir.glob("*.json")) to replay spool files left behind by a restart. Spool files are named pending-.json by _write_payload, so that sort is effectively random: a transcript recovered after a crash comes back scrambled. The damage is permanent rather than cosmetic. SessionDB restores a conversation with "ORDER BY id" — AUTOINCREMENT insertion order, never timestamp — precisely so a non-monotonic clock cannot sort an assistant tool_calls row after its tool response, "breaking tool-call/response adjacency and triggering an HTTP 400 on replay". Replaying the spool in filename order writes exactly that inversion into the row ids, so the session errors out on the next turn. The payloads already carry the ordering fields: spool_dropped_transcript_ message stamps every one with "ts" and a monotonic "seq". The sibling consumer of the same spool in this file, drain_transcript_spool, already sorts on them via sorted(entries, key=lambda e: e[:3]); only the restart path ignored them. Order on (ts, seq, filename) so the live drain and the cross-restart drain agree. Ordering by the payload rather than by a new file-naming scheme also recovers spool files that were written before this change, which are the ones an affected user already has on disk. Payloads that cannot be parsed keep their previous treatment: they sort last, are handed back unparsed, and are re-read inside the loop so the existing handler reports and preserves them unchanged. (cherry picked from commit 0e46604abc001a798f7f8567aedcaf19923f72df) (cherry picked from commit c043f0116e55b87b6e631592b12abbf60496aec8) --- gateway/shutdown_flush.py | 64 +++++++++++++++++++++++++++++++++++++-- 1 file changed, 61 insertions(+), 3 deletions(-) diff --git a/gateway/shutdown_flush.py b/gateway/shutdown_flush.py index 74af44744f5a6..5613ac2e10eb2 100644 --- a/gateway/shutdown_flush.py +++ b/gateway/shutdown_flush.py @@ -206,6 +206,61 @@ def _serialise_value(value: Any) -> Optional[dict]: return {"text": str(value)} +def _sort_number(value: Any) -> float: + """Coerce a spool ordering field to a float; unusable values sort first. + + Ordering fields are read back from JSON on disk and may be missing or + corrupt. A sort key that mixes ``str`` and ``int`` raises ``TypeError`` + and would abort the entire recovery pass, so anything non-numeric is + normalised to ``0.0``. + """ + if isinstance(value, bool) or not isinstance(value, (int, float)): + return 0.0 + return float(value) + + +def _order_flush_files(paths) -> list[tuple[Path, Optional[Dict[str, Any]]]]: + """Order recovery payloads by drop order — ``(ts, seq, filename)``. + + Recovery used to walk ``sorted(glob("*.json"))``, but spool files are + named ``pending-.json`` (see :func:`_write_payload`), so filename + order is effectively random. ``SessionDB`` restores a conversation by + AUTOINCREMENT id — true insertion order, never timestamp — so replaying + in filename order permanently scrambles the recovered transcript and can + separate an assistant tool call from its result. + + The payloads already carry ``ts`` and ``seq`` + (:func:`spool_dropped_transcript_message`), so this mirrors + :func:`drain_transcript_spool`'s ``sorted(entries, key=lambda e: e[:3])`` + and the live drain and the cross-restart drain agree on replay order. + + Returns ``(path, payload)`` pairs in replay order. A file whose payload + cannot be parsed sorts last, by name, and is returned with ``None`` so + the caller reports it through its own error handling. + """ + entries = [] + for path in paths: + try: + payload = json.loads(path.read_text(encoding="utf-8-sig")) + except Exception: + payload = None + if not isinstance(payload, dict): + entries.append(((1, 0.0, 0.0, path.name), path, None)) + continue + entries.append(( + ( + 0, + _sort_number(payload.get("ts")), + _sort_number(payload.get("seq")), + path.name, + ), + path, + payload, + )) + entries.sort(key=lambda entry: entry[0]) + return [(path, payload) for _key, path, payload in entries] + + def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int: """Replay flush-dir ``*.json`` files via ``SessionDB.append_message``, deleting each on success. @@ -216,7 +271,7 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int: branch. A returned ``db`` routes the append to the profile store owning the key (multiplexed gateways); ``None`` falls back to ``session_db``. Returns the number of messages recovered. """ - flush_files = sorted(_get_flush_dir().glob("*.json")) + flush_files = _order_flush_files(_get_flush_dir().glob("*.json")) if not flush_files: return 0 own_db = session_db is None @@ -225,12 +280,15 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int: session_db = acquire() recovered = 0 try: - for path in flush_files: + for path, payload in flush_files: # One unparseable payload or rejected append must only skip THIS file: the file is # never unlinked, so aborting the pass would re-poison every later boot. # utf-8-sig: our BOM-tolerant read fix for flush files. try: - payload = json.loads(path.read_text(encoding="utf-8-sig")) + if payload is None: + # Unparseable on the ordering pass: re-read so the failure is raised and + # reported here exactly as it was before. + payload = json.loads(path.read_text(encoding="utf-8-sig")) # Agent-history snapshots use a different schema (reason + # messages list) and are meant for manual operator recovery, # not automatic DB insertion. Skip them silently. From 9f77071be1d229fe18ae3f1b38ba7b0e95e6f4bf Mon Sep 17 00:00:00 2001 From: briandevans <252620095+briandevans@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:49:00 -0700 Subject: [PATCH 02/18] fix(gateway): preserve structured transcript fields when recovering spooled messages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The cross-restart replay of a cap-dropped transcript message forwarded only four fields to append_message — session_id, role, content, timestamp — and then unlinked the spool file, making the loss permanent. Everything else on the message was discarded: tool_calls, tool_call_id, tool_name, the reasoning columns, codex items, platform_message_id, observed, and the api_content sidecar. append_message already accepts all of them, and the payload already carries them, because spool_dropped_transcript_message writes the full transcript message dict. Two of the dropped fields are load-bearing rather than decorative. Losing tool_calls/tool_call_id orphans a tool result from the call it answers, which is the adjacency SessionDB's "ORDER BY id" exists to protect. Losing api_content contradicts the requirement stated at the live writer, that the sidecar "must survive any gateway-side persistence path or the next turn's replay diverges at this row" — and this is such a path. The same spool is drained two ways: by drain_transcript_spool during live operation, which replays through SessionStore._append_transcript_message and keeps every field, and by this function after a restart. Same files, same payloads, different fidelity. Mirror _append_transcript_message field for field so the outcome no longer depends on whether the gateway happened to restart, including its role gate on the assistant-only reasoning columns and its message_id fallback for platform_message_id. Fields are whitelisted explicitly rather than splatted from the message dict: the payload is arbitrary JSON from disk, and an unexpected key would raise TypeError and abort the whole recovery pass. content is now passed through as-is instead of being coerced to "", since an assistant tool-call row legitimately has no content. (cherry picked from commit 99ef8fbb41fca1b47873188ec93d64476fa59902) (cherry picked from commit 27ee00d618c973304c9e64bffccaeeaf1b54fbaf) --- gateway/shutdown_flush.py | 49 ++++++++++++++++++++++++++++++++++++--- 1 file changed, 46 insertions(+), 3 deletions(-) diff --git a/gateway/shutdown_flush.py b/gateway/shutdown_flush.py index 5613ac2e10eb2..a71234a532d3d 100644 --- a/gateway/shutdown_flush.py +++ b/gateway/shutdown_flush.py @@ -261,6 +261,51 @@ def _order_flush_files(paths) -> list[tuple[Path, Optional[Dict[str, Any]]]]: return [(path, payload) for _key, path, payload in entries] +def _transcript_append_kwargs( + session_id: str, + message: Dict[str, Any], + payload: Dict[str, Any], +) -> Dict[str, Any]: + """Build ``SessionDB.append_message`` kwargs for a spooled transcript row. + + Mirrors ``SessionTranscriptMixin._append_transcript_message`` in + ``gateway/session_transcript.py`` field for field, so a message replayed after a + restart lands as the same row the live drain would have written. The + fields are listed explicitly rather than splatted from *message*: the + spool payload is arbitrary JSON from disk and an unexpected key would + raise ``TypeError`` and abort the recovery pass. + + ``timestamp`` keeps the payload-level ``ts`` fallback, which is the only + clock available when the message itself was spooled without one. + """ + from agent.turn_context import extract_api_content_sidecar + from gateway.session_transcript import _ASSISTANT_ONLY_KEYS + + # Reasoning columns are assistant-only in the live writer; copying them + # onto another role would fabricate rows the gateway never produces. + is_assistant = message.get("role") == "assistant" + return { + "session_id": session_id, + "role": message.get("role", "unknown"), + "content": message.get("content"), + "tool_name": message.get("tool_name"), + "tool_calls": message.get("tool_calls"), + "tool_call_id": message.get("tool_call_id"), + **{k: message.get(k) if is_assistant else None for k in _ASSISTANT_ONLY_KEYS}, + "platform_message_id": ( + message.get("platform_message_id") or message.get("message_id") + ), + "observed": bool(message.get("observed")), + "timestamp": message.get("timestamp") or payload.get("ts"), + # The api_content sidecar is the exact bytes sent to the API for this + # row; the live writer requires it to survive every persistence path + # or the next replay diverges at this row. + "api_content": extract_api_content_sidecar(message), + "display_kind": message.get("display_kind"), + "display_metadata": message.get("display_metadata"), + } + + def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int: """Replay flush-dir ``*.json`` files via ``SessionDB.append_message``, deleting each on success. @@ -323,9 +368,7 @@ def _recover_one_payload(session_db, path: Path, payload: Dict[str, Any], *, logger.warning("Cannot recover structurally invalid transcript spool " "file %s; preserved for manual inspection", path) return False - session_db.append_message(session_id=spooled_sid, role=message.get("role", "unknown"), - content=message.get("content") or "", - timestamp=message.get("timestamp") or payload.get("ts")) + session_db.append_message(**_transcript_append_kwargs(spooled_sid, message, payload)) return True session_key, data = payload.get("session_key", ""), payload.get("data", {}) text = data.get("text", "") From 70488ccff9b11cff7022e3cfb47eafb3f2d465f6 Mon Sep 17 00:00:00 2001 From: briandevans <252620095+briandevans@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:49:29 -0700 Subject: [PATCH 03/18] fix(gateway): stop a session's spool replay after the first failed append MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ordering the spool files fixes the happy path only. If append_message fails partway through — the DB is still unhealthy, which is the situation that produced the spool in the first place — the loop logs the failure, keeps that file for a later retry, and then carries on and writes the messages that come after it. Those later messages land now; the failed one lands on some future start. Because SessionDB orders a conversation by AUTOINCREMENT id, the retried message then gets a HIGHER row id than the messages it originally preceded, and the inversion this pass just prevented is written to disk anyway — this time permanently, since both files are gone. drain_transcript_spool, the live drain of the same spool, already states the rule: "On the first replay failure the drain stops and remaining files are kept for the next attempt (the DB is likely still unhealthy)." Apply it here too. The block is per-session rather than global. Replay order is only defined within a session, and this function drains every session's spool in one pass, so a single unhealthy session must not strand the others' messages on disk. Non-transcript pending payloads are unaffected. (cherry picked from commit 5ea5104b6f4f896884508a36c11eb81010ccaefa) (cherry picked from commit a9591541728fa3bd76ef5a349dd55f066a039243) --- gateway/shutdown_flush.py | 22 +++++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/gateway/shutdown_flush.py b/gateway/shutdown_flush.py index a71234a532d3d..19a01dcdcb994 100644 --- a/gateway/shutdown_flush.py +++ b/gateway/shutdown_flush.py @@ -324,6 +324,9 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int: from hermes_state_registry import acquire session_db = acquire() recovered = 0 + # Sessions whose spool replay already failed this pass. Ordering is a per-session property, so + # one unhealthy session must not hold back the others. + blocked_sessions: set = set() try: for path, payload in flush_files: # One unparseable payload or rejected append must only skip THIS file: the file is @@ -340,7 +343,8 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int: if payload.get("reason") == "shutdown-with-unpersisted-agent-history": continue if _recover_one_payload(session_db, path, payload, - session_resolver=session_resolver): + session_resolver=session_resolver, + blocked_sessions=blocked_sessions): recovered += 1 path.unlink(missing_ok=True) except Exception as exc: @@ -356,7 +360,7 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int: def _recover_one_payload(session_db, path: Path, payload: Dict[str, Any], *, - session_resolver=None) -> bool: + session_resolver=None, blocked_sessions: Optional[set] = None) -> bool: """Append one flush payload to ``session_db``; False (file kept) when structurally invalid.""" # Cap-dropped transcript payloads carry the full message dict keyed by session_id — replay directly # (#78182). This handles spool files that were never drained before a restart. @@ -368,7 +372,19 @@ def _recover_one_payload(session_db, path: Path, payload: Dict[str, Any], *, logger.warning("Cannot recover structurally invalid transcript spool " "file %s; preserved for manual inspection", path) return False - session_db.append_message(**_transcript_append_kwargs(spooled_sid, message, payload)) + if blocked_sessions is not None and spooled_sid in blocked_sessions: + # An older message for this session could not be replayed. Writing this one now would give + # it a lower row id than the message it follows, permanently inverting the transcript, so + # leave it for the next start. + return False + try: + session_db.append_message(**_transcript_append_kwargs(spooled_sid, message, payload)) + except Exception: + # Same contract as drain_transcript_spool: stop this session's replay on the first failure + # and keep the remaining spool files for the next attempt. + if blocked_sessions is not None: + blocked_sessions.add(spooled_sid) + raise return True session_key, data = payload.get("session_key", ""), payload.get("data", {}) text = data.get("text", "") From be3f1c8cec7d0b9c5b16b0ba6161d651b7fb06a1 Mon Sep 17 00:00:00 2001 From: briandevans <252620095+briandevans@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:50:01 -0700 Subject: [PATCH 04/18] test(gateway): cover spool replay ordering across a restart MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Names the spool files so that filename order is the exact reverse of drop order, which is what uuid4 names produce on average, and asserts the replay comes back in drop order. A second case gives three payloads the same one-second ts so only the monotonic seq can separate them. Both fail on the previous sorted(glob()) implementation. A third case pins the pre-existing treatment of a corrupt payload — reported by the loop's own handler and left on disk — so the new ordering pass cannot silently swallow one. Lives in its own file rather than tests/gateway/test_shutdown_flush.py to keep the restart-recovery cases together with the spool fixtures they need. (cherry picked from commit 555d8719d0854b038a7663e96e4ac477842f345e) (cherry picked from commit c372b25d103436a373c9da14c26e71426afae908) --- tests/gateway/test_shutdown_flush_recovery.py | 131 ++++++++++++++++++ 1 file changed, 131 insertions(+) create mode 100644 tests/gateway/test_shutdown_flush_recovery.py diff --git a/tests/gateway/test_shutdown_flush_recovery.py b/tests/gateway/test_shutdown_flush_recovery.py new file mode 100644 index 0000000000000..f824bce35547a --- /dev/null +++ b/tests/gateway/test_shutdown_flush_recovery.py @@ -0,0 +1,131 @@ +"""Cross-restart recovery of cap-dropped transcript spool files (#78182). + +``recover_pending_to_db`` is the restart-time consumer of the same spool +``drain_transcript_spool`` drains during live operation. These tests pin the +properties the live drain already guarantees for that spool. +""" + +import json +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +from gateway.shutdown_flush import ( + TRANSCRIPT_CAP_DROP_REASON, + recover_pending_to_db, +) + + +@pytest.fixture +def flush_dir(tmp_path, monkeypatch): + """A temp spool directory wired into the module under test.""" + directory = tmp_path / "pending_messages" + directory.mkdir(parents=True, exist_ok=True) + monkeypatch.setattr( + "gateway.shutdown_flush._get_flush_dir", lambda: directory + ) + return directory + + +def _write_spool( + flush_dir: Path, + name: str, + session_id: str, + message: dict, + *, + ts: int, + seq: int, +) -> Path: + """Write one cap-drop spool payload under an explicit file name. + + Production names these ``pending-.json``; the tests choose the + names so that filename order and drop order can be made to disagree. + """ + path = flush_dir / name + path.write_text( + json.dumps( + { + "session_key": session_id, + "reason": TRANSCRIPT_CAP_DROP_REASON, + "ts": ts, + "seq": seq, + "data": {"session_id": session_id, "message": message}, + } + ), + encoding="utf-8", + ) + return path + + +def _contents(mock_db) -> list: + return [c.kwargs["content"] for c in mock_db.append_message.call_args_list] + + +class TestSpoolReplayOrder: + """Spool files must replay in drop order, not in file-name order.""" + + def test_replays_in_drop_order_when_names_disagree(self, flush_dir): + # Drop order is first -> second -> third; the uuid4-style names sort + # in exactly the opposite direction, which is what a real spool does + # on average. + _write_spool( + flush_dir, "pending-ccc.json", "sess-1", + {"role": "user", "content": "first"}, ts=100, seq=0, + ) + _write_spool( + flush_dir, "pending-bbb.json", "sess-1", + {"role": "assistant", "content": "second"}, ts=101, seq=1, + ) + _write_spool( + flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "third"}, ts=102, seq=2, + ) + + mock_db = MagicMock() + assert recover_pending_to_db(mock_db) == 3 + + # SessionDB restores by AUTOINCREMENT id, so append order IS the + # order the user will see after recovery. + assert _contents(mock_db) == ["first", "second", "third"] + + def test_seq_breaks_ties_within_the_same_second(self, flush_dir): + # ts has one-second resolution, so a burst of cap drops shares a ts + # and only ``seq`` can order them. + _write_spool( + flush_dir, "pending-zzz.json", "sess-1", + {"role": "user", "content": "first"}, ts=100, seq=0, + ) + _write_spool( + flush_dir, "pending-mmm.json", "sess-1", + {"role": "user", "content": "second"}, ts=100, seq=1, + ) + _write_spool( + flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "third"}, ts=100, seq=2, + ) + + mock_db = MagicMock() + assert recover_pending_to_db(mock_db) == 3 + assert _contents(mock_db) == ["first", "second", "third"] + + def test_unparseable_payload_is_reported_and_preserved( + self, flush_dir, caplog + ): + """A corrupt file must not break ordering or the recovery pass.""" + broken = flush_dir / "pending-aaa.json" + broken.write_text("{not json", encoding="utf-8") + _write_spool( + flush_dir, "pending-zzz.json", "sess-1", + {"role": "user", "content": "survivor"}, ts=100, seq=0, + ) + + mock_db = MagicMock() + with caplog.at_level("WARNING"): + assert recover_pending_to_db(mock_db) == 1 + + assert _contents(mock_db) == ["survivor"] + # The corrupt file is kept for the operator, and the failure is still + # reported by the loop's own handler. + assert broken.exists() + assert "Failed to recover pending message" in caplog.text From 19e53d5a070ea878314a558f7c30c38d788cd2b6 Mon Sep 17 00:00:00 2001 From: briandevans <252620095+briandevans@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:50:19 -0700 Subject: [PATCH 05/18] test(gateway): cover field fidelity and failure handling in spool recovery Round-trips an assistant tool-call row carrying every field the live writer persists and asserts each one reaches append_message: tool_calls, tool_call_id, tool_name, the reasoning and codex columns, platform_message_id, observed, timestamp, and the api_content sidecar. Also pins content=None passing through uncoerced, since an assistant tool-call row has no content, and the role gate that keeps the assistant-only reasoning columns off a user row. A separate case covers a failed replay: with the first of two messages for one session rejected, neither is written and both spool files stay on disk, while a third message for an unrelated session still recovers. All but the two behaviour-preservation cases fail on the previous implementation. (cherry picked from commit 869acc61272b835ae35c7e4f65efe25df8699ae6) (cherry picked from commit e02d9fcd6414e549f2eb7791f4c7388d6e07bacb) --- tests/gateway/test_shutdown_flush_recovery.py | 149 ++++++++++++++++++ 1 file changed, 149 insertions(+) diff --git a/tests/gateway/test_shutdown_flush_recovery.py b/tests/gateway/test_shutdown_flush_recovery.py index f824bce35547a..58f7322851461 100644 --- a/tests/gateway/test_shutdown_flush_recovery.py +++ b/tests/gateway/test_shutdown_flush_recovery.py @@ -129,3 +129,152 @@ def test_unparseable_payload_is_reported_and_preserved( # reported by the loop's own handler. assert broken.exists() assert "Failed to recover pending message" in caplog.text + + +class TestSpoolReplayFidelity: + """The full transcript message must survive the restart round trip.""" + + def test_structured_fields_are_preserved(self, flush_dir): + tool_calls = [ + { + "id": "call-1", + "type": "function", + "function": {"name": "send_payment", "arguments": "{}"}, + } + ] + message = { + "role": "assistant", + "content": None, + "tool_calls": tool_calls, + "reasoning": "deliberating", + "reasoning_content": "chain", + "reasoning_details": [{"type": "text"}], + "codex_reasoning_items": [{"id": "r1"}], + "codex_message_items": [{"id": "m1"}], + "platform_message_id": "tg-42", + "observed": True, + "timestamp": 12345, + "api_content": "exact bytes sent to the API", + } + _write_spool( + flush_dir, "pending-aaa.json", "sess-1", message, ts=100, seq=0, + ) + + mock_db = MagicMock() + assert recover_pending_to_db(mock_db) == 1 + + kwargs = mock_db.append_message.call_args.kwargs + assert kwargs["session_id"] == "sess-1" + assert kwargs["role"] == "assistant" + # An assistant tool-call row legitimately has no content; forcing it + # to "" would rewrite the message. + assert kwargs["content"] is None + assert kwargs["tool_calls"] == tool_calls + assert kwargs["reasoning"] == "deliberating" + assert kwargs["reasoning_content"] == "chain" + assert kwargs["reasoning_details"] == [{"type": "text"}] + assert kwargs["codex_reasoning_items"] == [{"id": "r1"}] + assert kwargs["codex_message_items"] == [{"id": "m1"}] + assert kwargs["platform_message_id"] == "tg-42" + assert kwargs["observed"] is True + assert kwargs["timestamp"] == 12345 + assert kwargs["api_content"] == "exact bytes sent to the API" + + def test_tool_result_keeps_its_call_id_and_name(self, flush_dir): + """Losing tool_call_id orphans the result and breaks replay.""" + _write_spool( + flush_dir, "pending-aaa.json", "sess-1", + { + "role": "tool", + "content": "receipt-1", + "tool_call_id": "call-1", + "tool_name": "send_payment", + }, + ts=100, seq=0, + ) + + mock_db = MagicMock() + assert recover_pending_to_db(mock_db) == 1 + + kwargs = mock_db.append_message.call_args.kwargs + assert kwargs["tool_call_id"] == "call-1" + assert kwargs["tool_name"] == "send_payment" + + def test_reasoning_is_not_fabricated_for_non_assistant_roles( + self, flush_dir + ): + """Mirrors the live writer, which gates reasoning on role.""" + _write_spool( + flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "hi", "reasoning": "leaked"}, + ts=100, seq=0, + ) + + mock_db = MagicMock() + assert recover_pending_to_db(mock_db) == 1 + assert mock_db.append_message.call_args.kwargs["reasoning"] is None + + def test_message_id_is_used_when_platform_message_id_is_absent( + self, flush_dir + ): + _write_spool( + flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "hi", "message_id": "tg-7"}, + ts=100, seq=0, + ) + + mock_db = MagicMock() + assert recover_pending_to_db(mock_db) == 1 + assert ( + mock_db.append_message.call_args.kwargs["platform_message_id"] + == "tg-7" + ) + + def test_payload_ts_is_the_timestamp_fallback(self, flush_dir): + _write_spool( + flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "hi"}, ts=999, seq=0, + ) + + mock_db = MagicMock() + assert recover_pending_to_db(mock_db) == 1 + assert mock_db.append_message.call_args.kwargs["timestamp"] == 999 + + +class TestSpoolReplayFailure: + """A failed replay must not let newer messages overtake older ones.""" + + def test_failure_blocks_later_messages_for_that_session_only( + self, flush_dir + ): + first = _write_spool( + flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "first"}, ts=100, seq=0, + ) + second = _write_spool( + flush_dir, "pending-bbb.json", "sess-1", + {"role": "user", "content": "second"}, ts=101, seq=1, + ) + other = _write_spool( + flush_dir, "pending-ccc.json", "sess-2", + {"role": "user", "content": "other-session"}, ts=102, seq=2, + ) + + mock_db = MagicMock() + + def append_message(**kwargs): + if kwargs["content"] == "first": + raise RuntimeError("controlled database outage") + return 1 + + mock_db.append_message.side_effect = append_message + + assert recover_pending_to_db(mock_db) == 1 + + # "second" must NOT be written: it would land with a lower row id + # than "first" once "first" is retried on a later start. + assert _contents(mock_db) == ["first", "other-session"] + assert first.exists() + assert second.exists() + # A different session is unaffected by sess-1's outage. + assert not other.exists() From c918adf3c6af990504d2e10b3dd7030ab65fc01f Mon Sep 17 00:00:00 2001 From: briandevans <252620095+briandevans@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:59:43 -0700 Subject: [PATCH 06/18] fix(gateway): fall back to the payload clock only when a timestamp is absent _transcript_append_kwargs chose the spooled message's timestamp with `message.get("timestamp") or payload.get("ts")`, carried over from the code this PR replaces. Truthiness is the wrong test: epoch 0 is a valid timestamp and would be silently rewritten to the payload's ts, which is the fidelity loss this PR exists to remove. Fall back only when the field is genuinely absent. (cherry picked from commit 62206b0bcd71ea22c3a383f05fe12dd9242ab4a1) (cherry picked from commit a7b6828ee97f0e876fc033c74f3026aea779e9fd) --- gateway/shutdown_flush.py | 7 ++++++- tests/gateway/test_shutdown_flush_recovery.py | 13 +++++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/gateway/shutdown_flush.py b/gateway/shutdown_flush.py index 19a01dcdcb994..e5c3313852c66 100644 --- a/gateway/shutdown_flush.py +++ b/gateway/shutdown_flush.py @@ -284,6 +284,11 @@ def _transcript_append_kwargs( # Reasoning columns are assistant-only in the live writer; copying them # onto another role would fabricate rows the gateway never produces. is_assistant = message.get("role") == "assistant" + # Only a *missing* timestamp falls back to the payload clock. A truthiness + # test would rewrite epoch 0, which is a valid timestamp. + timestamp = message.get("timestamp") + if timestamp is None: + timestamp = payload.get("ts") return { "session_id": session_id, "role": message.get("role", "unknown"), @@ -296,7 +301,7 @@ def _transcript_append_kwargs( message.get("platform_message_id") or message.get("message_id") ), "observed": bool(message.get("observed")), - "timestamp": message.get("timestamp") or payload.get("ts"), + "timestamp": timestamp, # The api_content sidecar is the exact bytes sent to the API for this # row; the live writer requires it to survive every persistence path # or the next replay diverges at this row. diff --git a/tests/gateway/test_shutdown_flush_recovery.py b/tests/gateway/test_shutdown_flush_recovery.py index 58f7322851461..4b3d984f6b63d 100644 --- a/tests/gateway/test_shutdown_flush_recovery.py +++ b/tests/gateway/test_shutdown_flush_recovery.py @@ -230,6 +230,19 @@ def test_message_id_is_used_when_platform_message_id_is_absent( == "tg-7" ) + def test_epoch_zero_timestamp_is_not_replaced_by_the_fallback( + self, flush_dir + ): + """0 is a valid timestamp; only a missing one may fall back.""" + _write_spool( + flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "hi", "timestamp": 0}, ts=999, seq=0, + ) + + mock_db = MagicMock() + assert recover_pending_to_db(mock_db) == 1 + assert mock_db.append_message.call_args.kwargs["timestamp"] == 0 + def test_payload_ts_is_the_timestamp_fallback(self, flush_dir): _write_spool( flush_dir, "pending-aaa.json", "sess-1", From a2c0e193a2eac2ab60cdd2304886ed57cf80a9b4 Mon Sep 17 00:00:00 2001 From: John Paul Soliva Date: Sat, 26 Sep 2026 19:18:18 +0900 Subject: [PATCH 07/18] fix(gateway): the live and restart transcript writers share one row builder Review follow-ups (teknium1): - One kwargs builder, two callers. transcript_append_kwargs() in gateway/session_transcript.py builds the append_message kwargs, and both _append_transcript_message and the spool's restart replay call it, so the restart writer cannot lag the live writer again (display_kind and display_metadata were added after #84785 was written). fallback_ts applies only to a message with no timestamp; the live writer passes none, so its rows are byte-identical. - A failed replay says what it held back: one INFO line per pass with the file count per blocked session, so a spool dir that never empties has a reason in the log. - seq is per process and ts has one-second resolution; the sort key now says so where it reads seq. - Tests fold to three invariants: drop order beats name order (seq breaks a same-second tie), a replayed row is the row the live writer writes, and a failed replay holds back that session's later messages only. The two that already passed on main (a corrupt file is kept; the payload-clock fallback) are dropped; the fallback is asserted inside the fidelity test. (cherry picked from commit e2bacb06b3311b00aa354ffce7d53fc15ba44d04) --- gateway/session_transcript.py | 52 +-- gateway/shutdown_flush.py | 72 +--- tests/gateway/test_shutdown_flush_recovery.py | 310 +++++------------- 3 files changed, 130 insertions(+), 304 deletions(-) diff --git a/gateway/session_transcript.py b/gateway/session_transcript.py index 1d3c9f5e05610..b26ed850f4fe0 100644 --- a/gateway/session_transcript.py +++ b/gateway/session_transcript.py @@ -43,6 +43,36 @@ def _spool_dropped(session_id: str, message: Dict[str, Any]): ) +def transcript_append_kwargs(session_id: str, message: Dict[str, Any], *, fallback_ts: Any = None) -> Dict[str, Any]: + """``SessionDB.append_message`` kwargs for one transcript row. The live writer and the restart + replay of the transcript spool (gateway/shutdown_flush.py) both build rows here, so a replayed + message lands as the row the live drain would have written. Fields are listed rather than + splatted: a spooled message is arbitrary JSON from disk, and an unexpected key would raise and + abort the replay. *fallback_ts* applies only to a message with no timestamp at all, because a + truthiness test would rewrite epoch 0.""" + is_assistant = message.get("role") == "assistant" + timestamp = message.get("timestamp") + return { + "session_id": session_id, + "role": message.get("role", "unknown"), + "content": message.get("content"), + "tool_name": message.get("tool_name"), + "tool_calls": message.get("tool_calls"), + "tool_call_id": message.get("tool_call_id"), + **{k: message.get(k) if is_assistant else None for k in _ASSISTANT_ONLY_KEYS}, + "platform_message_id": message.get("platform_message_id") or message.get("message_id"), + "observed": bool(message.get("observed")), + "timestamp": fallback_ts if timestamp is None else timestamp, + # Exact bytes sent to the API (prompt-cache-stable replay); must survive every + # persistence path or the next replay diverges. + "api_content": extract_api_content_sidecar(message), + # Presentation typing ("internal_notification" for self-injected async-delegation/background + # notification turns, #82888). DB-only; stripped from provider-bound payloads. + "display_kind": message.get("display_kind"), + "display_metadata": message.get("display_metadata"), + } + + class SessionTranscriptMixin: """SessionStore transcript I/O: SQLite append with a per-session retry queue, compression-reroute following, FTS corruption recovery, rewrite/rewind/load.""" @@ -370,27 +400,7 @@ def _append_transcript_message(self, session_id: str, message: Dict[str, Any]) - # into the ambient store. raise RuntimeError( f"no owning session store for {session_id}; deferring transcript write") - is_assistant = message.get("role") == "assistant" - _db.append_message( - session_id=session_id, - role=message.get("role", "unknown"), - content=message.get("content"), - tool_name=message.get("tool_name"), - tool_calls=message.get("tool_calls"), - tool_call_id=message.get("tool_call_id"), - **{k: message.get(k) if is_assistant else None for k in _ASSISTANT_ONLY_KEYS}, - platform_message_id=(message.get("platform_message_id") or message.get("message_id")), - observed=bool(message.get("observed")), - timestamp=message.get("timestamp"), - # Exact bytes sent to the API (prompt-cache-stable replay); must survive every - # persistence path or the next replay diverges. - api_content=extract_api_content_sidecar(message), - # Presentation typing (e.g. "internal_notification"); DB-only. - # "internal_notification" for self-injected async-delegation/background notification turns, - # #82888). DB-only; stripped from provider-bound payloads. - display_kind=message.get("display_kind"), - display_metadata=message.get("display_metadata"), - ) + _db.append_message(**transcript_append_kwargs(session_id, message)) @staticmethod def _is_fts_corruption_error(exc: Exception) -> bool: diff --git a/gateway/shutdown_flush.py b/gateway/shutdown_flush.py index e5c3313852c66..37c5b63afeaf8 100644 --- a/gateway/shutdown_flush.py +++ b/gateway/shutdown_flush.py @@ -251,6 +251,9 @@ def _order_flush_files(paths) -> list[tuple[Path, Optional[Dict[str, Any]]]]: ( 0, _sort_number(payload.get("ts")), + # seq is per process (_TRANSCRIPT_SPOOL_SEQ) and ts has one-second resolution, so + # files two processes spool in the same second can interleave; within one process + # the order is exact. _sort_number(payload.get("seq")), path.name, ), @@ -261,56 +264,6 @@ def _order_flush_files(paths) -> list[tuple[Path, Optional[Dict[str, Any]]]]: return [(path, payload) for _key, path, payload in entries] -def _transcript_append_kwargs( - session_id: str, - message: Dict[str, Any], - payload: Dict[str, Any], -) -> Dict[str, Any]: - """Build ``SessionDB.append_message`` kwargs for a spooled transcript row. - - Mirrors ``SessionTranscriptMixin._append_transcript_message`` in - ``gateway/session_transcript.py`` field for field, so a message replayed after a - restart lands as the same row the live drain would have written. The - fields are listed explicitly rather than splatted from *message*: the - spool payload is arbitrary JSON from disk and an unexpected key would - raise ``TypeError`` and abort the recovery pass. - - ``timestamp`` keeps the payload-level ``ts`` fallback, which is the only - clock available when the message itself was spooled without one. - """ - from agent.turn_context import extract_api_content_sidecar - from gateway.session_transcript import _ASSISTANT_ONLY_KEYS - - # Reasoning columns are assistant-only in the live writer; copying them - # onto another role would fabricate rows the gateway never produces. - is_assistant = message.get("role") == "assistant" - # Only a *missing* timestamp falls back to the payload clock. A truthiness - # test would rewrite epoch 0, which is a valid timestamp. - timestamp = message.get("timestamp") - if timestamp is None: - timestamp = payload.get("ts") - return { - "session_id": session_id, - "role": message.get("role", "unknown"), - "content": message.get("content"), - "tool_name": message.get("tool_name"), - "tool_calls": message.get("tool_calls"), - "tool_call_id": message.get("tool_call_id"), - **{k: message.get(k) if is_assistant else None for k in _ASSISTANT_ONLY_KEYS}, - "platform_message_id": ( - message.get("platform_message_id") or message.get("message_id") - ), - "observed": bool(message.get("observed")), - "timestamp": timestamp, - # The api_content sidecar is the exact bytes sent to the API for this - # row; the live writer requires it to survive every persistence path - # or the next replay diverges at this row. - "api_content": extract_api_content_sidecar(message), - "display_kind": message.get("display_kind"), - "display_metadata": message.get("display_metadata"), - } - - def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int: """Replay flush-dir ``*.json`` files via ``SessionDB.append_message``, deleting each on success. @@ -329,9 +282,9 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int: from hermes_state_registry import acquire session_db = acquire() recovered = 0 - # Sessions whose spool replay already failed this pass. Ordering is a per-session property, so - # one unhealthy session must not hold back the others. - blocked_sessions: set = set() + # Sessions whose spool replay already failed this pass -> files held back for them. Ordering is a + # per-session property, so one unhealthy session must not hold back the others. + blocked_sessions: Dict[str, int] = {} try: for path, payload in flush_files: # One unparseable payload or rejected append must only skip THIS file: the file is @@ -361,11 +314,16 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int: release_or_close(session_db) if recovered: logger.info("Recovered %d pending message(s) from shutdown flush", recovered) + if blocked_sessions: + # Otherwise a spool dir that never empties gives an operator no reason why. + logger.info("Held back %d spooled transcript file(s) for the next start after a failed replay: %s", + sum(blocked_sessions.values()), + ", ".join(f"{sid} ({count})" for sid, count in blocked_sessions.items())) return recovered def _recover_one_payload(session_db, path: Path, payload: Dict[str, Any], *, - session_resolver=None, blocked_sessions: Optional[set] = None) -> bool: + session_resolver=None, blocked_sessions: Optional[Dict[str, int]] = None) -> bool: """Append one flush payload to ``session_db``; False (file kept) when structurally invalid.""" # Cap-dropped transcript payloads carry the full message dict keyed by session_id — replay directly # (#78182). This handles spool files that were never drained before a restart. @@ -381,14 +339,16 @@ def _recover_one_payload(session_db, path: Path, payload: Dict[str, Any], *, # An older message for this session could not be replayed. Writing this one now would give # it a lower row id than the message it follows, permanently inverting the transcript, so # leave it for the next start. + blocked_sessions[spooled_sid] += 1 return False try: - session_db.append_message(**_transcript_append_kwargs(spooled_sid, message, payload)) + from gateway.session_transcript import transcript_append_kwargs + session_db.append_message(**transcript_append_kwargs(spooled_sid, message, fallback_ts=payload.get("ts"))) except Exception: # Same contract as drain_transcript_spool: stop this session's replay on the first failure # and keep the remaining spool files for the next attempt. if blocked_sessions is not None: - blocked_sessions.add(spooled_sid) + blocked_sessions[spooled_sid] = 1 raise return True session_key, data = payload.get("session_key", ""), payload.get("data", {}) diff --git a/tests/gateway/test_shutdown_flush_recovery.py b/tests/gateway/test_shutdown_flush_recovery.py index 4b3d984f6b63d..dd358bb96418b 100644 --- a/tests/gateway/test_shutdown_flush_recovery.py +++ b/tests/gateway/test_shutdown_flush_recovery.py @@ -62,232 +62,88 @@ def _contents(mock_db) -> list: return [c.kwargs["content"] for c in mock_db.append_message.call_args_list] -class TestSpoolReplayOrder: - """Spool files must replay in drop order, not in file-name order.""" - - def test_replays_in_drop_order_when_names_disagree(self, flush_dir): - # Drop order is first -> second -> third; the uuid4-style names sort - # in exactly the opposite direction, which is what a real spool does - # on average. - _write_spool( - flush_dir, "pending-ccc.json", "sess-1", - {"role": "user", "content": "first"}, ts=100, seq=0, - ) - _write_spool( - flush_dir, "pending-bbb.json", "sess-1", - {"role": "assistant", "content": "second"}, ts=101, seq=1, - ) - _write_spool( - flush_dir, "pending-aaa.json", "sess-1", - {"role": "user", "content": "third"}, ts=102, seq=2, - ) - - mock_db = MagicMock() - assert recover_pending_to_db(mock_db) == 3 - - # SessionDB restores by AUTOINCREMENT id, so append order IS the - # order the user will see after recovery. - assert _contents(mock_db) == ["first", "second", "third"] - - def test_seq_breaks_ties_within_the_same_second(self, flush_dir): - # ts has one-second resolution, so a burst of cap drops shares a ts - # and only ``seq`` can order them. - _write_spool( - flush_dir, "pending-zzz.json", "sess-1", - {"role": "user", "content": "first"}, ts=100, seq=0, - ) - _write_spool( - flush_dir, "pending-mmm.json", "sess-1", - {"role": "user", "content": "second"}, ts=100, seq=1, - ) - _write_spool( - flush_dir, "pending-aaa.json", "sess-1", - {"role": "user", "content": "third"}, ts=100, seq=2, - ) - - mock_db = MagicMock() - assert recover_pending_to_db(mock_db) == 3 - assert _contents(mock_db) == ["first", "second", "third"] - - def test_unparseable_payload_is_reported_and_preserved( - self, flush_dir, caplog - ): - """A corrupt file must not break ordering or the recovery pass.""" - broken = flush_dir / "pending-aaa.json" - broken.write_text("{not json", encoding="utf-8") - _write_spool( - flush_dir, "pending-zzz.json", "sess-1", - {"role": "user", "content": "survivor"}, ts=100, seq=0, - ) - - mock_db = MagicMock() - with caplog.at_level("WARNING"): - assert recover_pending_to_db(mock_db) == 1 - - assert _contents(mock_db) == ["survivor"] - # The corrupt file is kept for the operator, and the failure is still - # reported by the loop's own handler. - assert broken.exists() - assert "Failed to recover pending message" in caplog.text - - -class TestSpoolReplayFidelity: - """The full transcript message must survive the restart round trip.""" - - def test_structured_fields_are_preserved(self, flush_dir): - tool_calls = [ - { - "id": "call-1", - "type": "function", - "function": {"name": "send_payment", "arguments": "{}"}, - } - ] - message = { - "role": "assistant", - "content": None, - "tool_calls": tool_calls, - "reasoning": "deliberating", - "reasoning_content": "chain", - "reasoning_details": [{"type": "text"}], - "codex_reasoning_items": [{"id": "r1"}], - "codex_message_items": [{"id": "m1"}], - "platform_message_id": "tg-42", - "observed": True, - "timestamp": 12345, - "api_content": "exact bytes sent to the API", - } - _write_spool( - flush_dir, "pending-aaa.json", "sess-1", message, ts=100, seq=0, - ) - - mock_db = MagicMock() - assert recover_pending_to_db(mock_db) == 1 - - kwargs = mock_db.append_message.call_args.kwargs - assert kwargs["session_id"] == "sess-1" - assert kwargs["role"] == "assistant" - # An assistant tool-call row legitimately has no content; forcing it - # to "" would rewrite the message. - assert kwargs["content"] is None - assert kwargs["tool_calls"] == tool_calls - assert kwargs["reasoning"] == "deliberating" - assert kwargs["reasoning_content"] == "chain" - assert kwargs["reasoning_details"] == [{"type": "text"}] - assert kwargs["codex_reasoning_items"] == [{"id": "r1"}] - assert kwargs["codex_message_items"] == [{"id": "m1"}] - assert kwargs["platform_message_id"] == "tg-42" - assert kwargs["observed"] is True - assert kwargs["timestamp"] == 12345 - assert kwargs["api_content"] == "exact bytes sent to the API" - - def test_tool_result_keeps_its_call_id_and_name(self, flush_dir): - """Losing tool_call_id orphans the result and breaks replay.""" - _write_spool( - flush_dir, "pending-aaa.json", "sess-1", - { - "role": "tool", - "content": "receipt-1", - "tool_call_id": "call-1", - "tool_name": "send_payment", - }, - ts=100, seq=0, - ) - - mock_db = MagicMock() - assert recover_pending_to_db(mock_db) == 1 - - kwargs = mock_db.append_message.call_args.kwargs - assert kwargs["tool_call_id"] == "call-1" - assert kwargs["tool_name"] == "send_payment" - - def test_reasoning_is_not_fabricated_for_non_assistant_roles( - self, flush_dir - ): - """Mirrors the live writer, which gates reasoning on role.""" - _write_spool( - flush_dir, "pending-aaa.json", "sess-1", - {"role": "user", "content": "hi", "reasoning": "leaked"}, - ts=100, seq=0, - ) - - mock_db = MagicMock() - assert recover_pending_to_db(mock_db) == 1 - assert mock_db.append_message.call_args.kwargs["reasoning"] is None - - def test_message_id_is_used_when_platform_message_id_is_absent( - self, flush_dir - ): - _write_spool( - flush_dir, "pending-aaa.json", "sess-1", - {"role": "user", "content": "hi", "message_id": "tg-7"}, - ts=100, seq=0, - ) - - mock_db = MagicMock() - assert recover_pending_to_db(mock_db) == 1 - assert ( - mock_db.append_message.call_args.kwargs["platform_message_id"] - == "tg-7" - ) - - def test_epoch_zero_timestamp_is_not_replaced_by_the_fallback( - self, flush_dir - ): - """0 is a valid timestamp; only a missing one may fall back.""" - _write_spool( - flush_dir, "pending-aaa.json", "sess-1", - {"role": "user", "content": "hi", "timestamp": 0}, ts=999, seq=0, - ) - - mock_db = MagicMock() - assert recover_pending_to_db(mock_db) == 1 - assert mock_db.append_message.call_args.kwargs["timestamp"] == 0 - - def test_payload_ts_is_the_timestamp_fallback(self, flush_dir): - _write_spool( - flush_dir, "pending-aaa.json", "sess-1", - {"role": "user", "content": "hi"}, ts=999, seq=0, - ) - - mock_db = MagicMock() - assert recover_pending_to_db(mock_db) == 1 - assert mock_db.append_message.call_args.kwargs["timestamp"] == 999 - - -class TestSpoolReplayFailure: - """A failed replay must not let newer messages overtake older ones.""" - - def test_failure_blocks_later_messages_for_that_session_only( - self, flush_dir - ): - first = _write_spool( - flush_dir, "pending-aaa.json", "sess-1", - {"role": "user", "content": "first"}, ts=100, seq=0, - ) - second = _write_spool( - flush_dir, "pending-bbb.json", "sess-1", - {"role": "user", "content": "second"}, ts=101, seq=1, - ) - other = _write_spool( - flush_dir, "pending-ccc.json", "sess-2", - {"role": "user", "content": "other-session"}, ts=102, seq=2, - ) - - mock_db = MagicMock() - - def append_message(**kwargs): - if kwargs["content"] == "first": - raise RuntimeError("controlled database outage") - return 1 - - mock_db.append_message.side_effect = append_message - +def test_replays_in_drop_order_not_file_name_order(flush_dir): + """SessionDB restores by AUTOINCREMENT id, so append order IS the order the user sees + after recovery. Production names are ``pending-.json``; these names sort opposite + to drop order, and a burst inside one second shares ``ts``, so ``seq`` breaks the tie.""" + _write_spool(flush_dir, "pending-zzz.json", "sess-1", + {"role": "user", "content": "first"}, ts=100, seq=0) + _write_spool(flush_dir, "pending-mmm.json", "sess-1", + {"role": "assistant", "content": "second"}, ts=100, seq=1) + _write_spool(flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "third"}, ts=101, seq=2) + + mock_db = MagicMock() + assert recover_pending_to_db(mock_db) == 3 + assert _contents(mock_db) == ["first", "second", "third"] + + +def test_a_replayed_row_is_the_row_the_live_writer_writes(flush_dir): + """Every field the live drain persists survives the restart round trip: losing + tool_call_id orphans a tool result, losing api_content makes the next replay diverge, + and reasoning stays assistant-only. A missing timestamp falls back to the payload + clock, but epoch 0 is a real timestamp.""" + tool_calls = [{"id": "call-1", "type": "function", + "function": {"name": "send_payment", "arguments": "{}"}}] + _write_spool(flush_dir, "pending-ccc.json", "sess-1", { + "role": "assistant", "content": None, "tool_calls": tool_calls, + "reasoning": "deliberating", "reasoning_content": "chain", + "reasoning_details": [{"type": "text"}], "codex_reasoning_items": [{"id": "r1"}], + "codex_message_items": [{"id": "m1"}], "platform_message_id": "tg-42", + "observed": True, "timestamp": 0, "api_content": "exact bytes sent to the API", + "display_kind": "internal_notification", + }, ts=100, seq=0) + _write_spool(flush_dir, "pending-bbb.json", "sess-1", { + "role": "tool", "content": "receipt-1", "tool_call_id": "call-1", "tool_name": "send_payment", + }, ts=100, seq=1) + _write_spool(flush_dir, "pending-aaa.json", "sess-1", { + "role": "user", "content": "hi", "reasoning": "leaked", "message_id": "tg-7", + }, ts=999, seq=2) + + mock_db = MagicMock() + assert recover_pending_to_db(mock_db) == 3 + assistant, tool, user = (c.kwargs for c in mock_db.append_message.call_args_list) + + assert assistant["content"] is None # a tool-call row legitimately has no content + assert assistant["tool_calls"] == tool_calls + assert (assistant["reasoning"], assistant["reasoning_content"]) == ("deliberating", "chain") + assert assistant["reasoning_details"] == [{"type": "text"}] + assert assistant["codex_reasoning_items"] == [{"id": "r1"}] + assert assistant["codex_message_items"] == [{"id": "m1"}] + assert (assistant["platform_message_id"], assistant["observed"]) == ("tg-42", True) + assert assistant["api_content"] == "exact bytes sent to the API" + assert assistant["display_kind"] == "internal_notification" + assert assistant["timestamp"] == 0 + assert (tool["tool_call_id"], tool["tool_name"]) == ("call-1", "send_payment") + assert user["reasoning"] is None + assert user["platform_message_id"] == "tg-7" + assert user["timestamp"] == 999 + + +def test_a_failed_replay_holds_back_that_sessions_later_messages_only(flush_dir, caplog): + """Writing "second" after "first" failed would give it a lower row id than "first" once + "first" is retried on a later start: the inversion lands on disk for good. Another + session is unaffected, and the pass says what it held back.""" + first = _write_spool(flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "first"}, ts=100, seq=0) + second = _write_spool(flush_dir, "pending-bbb.json", "sess-1", + {"role": "user", "content": "second"}, ts=101, seq=1) + other = _write_spool(flush_dir, "pending-ccc.json", "sess-2", + {"role": "user", "content": "other-session"}, ts=102, seq=2) + + mock_db = MagicMock() + + def append_message(**kwargs): + if kwargs["content"] == "first": + raise RuntimeError("controlled database outage") + return 1 + + mock_db.append_message.side_effect = append_message + + with caplog.at_level("INFO", logger="gateway.shutdown_flush"): assert recover_pending_to_db(mock_db) == 1 - # "second" must NOT be written: it would land with a lower row id - # than "first" once "first" is retried on a later start. - assert _contents(mock_db) == ["first", "other-session"] - assert first.exists() - assert second.exists() - # A different session is unaffected by sess-1's outage. - assert not other.exists() + assert _contents(mock_db) == ["first", "other-session"] + assert first.exists() and second.exists() + assert not other.exists() + assert "Held back 2 spooled transcript file(s)" in caplog.text and "sess-1 (2)" in caplog.text From 0bd036ff9d453f8dd07fc8c43b2d981ef41ef7e8 Mon Sep 17 00:00:00 2001 From: John Paul Soliva Date: Sun, 27 Sep 2026 00:22:51 +0900 Subject: [PATCH 08/18] fix(gateway): an out-of-range spool ts or seq no longer aborts recovery _order_flush_files runs before recover_pending_to_db's per-file error handling, so float(10**400) raising OverflowError stopped every file from being recovered. Unusable ordering values (too large for a float, NaN, Infinity) now sort first like any other non-numeric value. (cherry picked from commit 50c9b97954ae85f3acee96352a9f1a883ba75688) --- gateway/shutdown_flush.py | 11 +++++++++-- tests/gateway/test_shutdown_flush_recovery.py | 13 +++++++++++++ 2 files changed, 22 insertions(+), 2 deletions(-) diff --git a/gateway/shutdown_flush.py b/gateway/shutdown_flush.py index 37c5b63afeaf8..a1c873ae62430 100644 --- a/gateway/shutdown_flush.py +++ b/gateway/shutdown_flush.py @@ -15,6 +15,7 @@ import itertools import json import logging +import math import os import time import uuid @@ -212,11 +213,17 @@ def _sort_number(value: Any) -> float: Ordering fields are read back from JSON on disk and may be missing or corrupt. A sort key that mixes ``str`` and ``int`` raises ``TypeError`` and would abort the entire recovery pass, so anything non-numeric is - normalised to ``0.0``. + normalised to ``0.0``. So is a JSON integer too large for a float + (``float(10**400)`` raises ``OverflowError``) and a NaN/Infinity literal + (NaN compares false both ways and would scramble the sort). """ if isinstance(value, bool) or not isinstance(value, (int, float)): return 0.0 - return float(value) + try: + number = float(value) + except OverflowError: + return 0.0 + return number if math.isfinite(number) else 0.0 def _order_flush_files(paths) -> list[tuple[Path, Optional[Dict[str, Any]]]]: diff --git a/tests/gateway/test_shutdown_flush_recovery.py b/tests/gateway/test_shutdown_flush_recovery.py index dd358bb96418b..0776cb382734d 100644 --- a/tests/gateway/test_shutdown_flush_recovery.py +++ b/tests/gateway/test_shutdown_flush_recovery.py @@ -78,6 +78,19 @@ def test_replays_in_drop_order_not_file_name_order(flush_dir): assert _contents(mock_db) == ["first", "second", "third"] +def test_an_out_of_range_ordering_field_does_not_abort_recovery(flush_dir): + """Ordering runs before the per-file error handling, so one file whose ``ts`` is a JSON + integer no float can hold must not stop every other file from being recovered.""" + _write_spool(flush_dir, "pending-bad.json", "sess-1", + {"role": "user", "content": "huge ts"}, ts=10**400, seq=0) + _write_spool(flush_dir, "pending-ok.json", "sess-2", + {"role": "user", "content": "valid"}, ts=100, seq=1) + + mock_db = MagicMock() + assert recover_pending_to_db(mock_db) == 2 + assert sorted(_contents(mock_db)) == ["huge ts", "valid"] + + def test_a_replayed_row_is_the_row_the_live_writer_writes(flush_dir): """Every field the live drain persists survives the restart round trip: losing tool_call_id orphans a tool result, losing api_content makes the next replay diverge, From e2154ae53656dce2bfb56cef44dae789672ce957 Mon Sep 17 00:00:00 2001 From: unsupportedpastels Date: Tue, 6 Oct 2026 14:32:37 +0000 Subject: [PATCH 09/18] fix(gateway): catch only read and decode errors when ordering spool files The code-health gate (BLE001) rejects the blind except added to _order_flush_files. A file that cannot be read or parsed still sorts last with a None payload, so malformed-file isolation is unchanged. --- gateway/shutdown_flush.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/gateway/shutdown_flush.py b/gateway/shutdown_flush.py index a1c873ae62430..114ac480ecd1f 100644 --- a/gateway/shutdown_flush.py +++ b/gateway/shutdown_flush.py @@ -249,7 +249,8 @@ def _order_flush_files(paths) -> list[tuple[Path, Optional[Dict[str, Any]]]]: for path in paths: try: payload = json.loads(path.read_text(encoding="utf-8-sig")) - except Exception: + # OSError: unreadable; ValueError: bad JSON or bytes; RecursionError: nesting too deep. + except (OSError, ValueError, RecursionError): payload = None if not isinstance(payload, dict): entries.append(((1, 0.0, 0.0, path.name), path, None)) From 36f27f287f002065903a1ad673f75eaf1b9c7813 Mon Sep 17 00:00:00 2001 From: unsupportedpastels Date: Tue, 6 Oct 2026 15:43:13 +0000 Subject: [PATCH 10/18] fix(gateway): drain boot-held-back spool files before the session's next live write recover_pending_to_db held back a session's later spool files after a failed replay, but never told the running gateway. The live writer drains the spool only for sessions in SessionStore._spooled_drop_sessions, which only this process's own spooling filled, so the session's next live row landed ahead of the held-back files until the next restart. recover_pending_to_db now adds the sessions it held back to a caller-supplied set, and _recover_pending_flushes passes the store's own drain set (exposed as spooled_drop_sessions()), so the next live write replays them first. Found in review by @BearHuddleston. --- gateway/run.py | 6 ++-- gateway/session_transcript.py | 5 +++ gateway/shutdown_flush.py | 8 ++++- tests/gateway/test_shutdown_flush_recovery.py | 33 +++++++++++++++++++ 4 files changed, 48 insertions(+), 4 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index f59aa5dec796e..591473ce736eb 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1662,8 +1662,8 @@ def _recover_pending_flushes(runner) -> int: from gateway.shutdown_flush import recover_pending_to_db from hermes_constants import get_hermes_home, reset_hermes_home_override, set_hermes_home_override - resolver = runner.session_store.resolve_session_id_for_key - recovered = recover_pending_to_db(session_resolver=resolver) + resolver, held_back = runner.session_store.resolve_session_id_for_key, runner.session_store.spooled_drop_sessions() + recovered = recover_pending_to_db(session_resolver=resolver, held_back=held_back) if not getattr(runner.config, "multiplex_profiles", False): return recovered launch_home = Path(get_hermes_home()).resolve() @@ -1672,7 +1672,7 @@ def _recover_pending_flushes(runner) -> int: continue token = set_hermes_home_override(str(home)) try: - recovered += recover_pending_to_db(session_resolver=resolver) + recovered += recover_pending_to_db(session_resolver=resolver, held_back=held_back) except Exception: # one profile's unreadable spool must not strand the others' logger.warning("Pending-message recovery failed for profile %s", name, exc_info=True) finally: diff --git a/gateway/session_transcript.py b/gateway/session_transcript.py index b26ed850f4fe0..05ce8dde709b3 100644 --- a/gateway/session_transcript.py +++ b/gateway/session_transcript.py @@ -392,6 +392,11 @@ def _drain_spooled_drops(self, session_id: str) -> None: except Exception as exc: logger.warning("Failed to drain transcript spool for %s: %s", session_id, exc) + def spooled_drop_sessions(self) -> set: + """The live set of sessions whose next transcript write drains the spool first. Boot recovery + adds the sessions it held back, so their next live row cannot land ahead of those files.""" + return self._lazy("_spooled_drop_sessions", set) + def _append_transcript_message(self, session_id: str, message: Dict[str, Any]) -> None: """Write one transcript row. Caller handles retry queuing.""" _db = self._db_for_session_id(session_id) diff --git a/gateway/shutdown_flush.py b/gateway/shutdown_flush.py index 114ac480ecd1f..23509d393cb5a 100644 --- a/gateway/shutdown_flush.py +++ b/gateway/shutdown_flush.py @@ -272,7 +272,8 @@ def _order_flush_files(paths) -> list[tuple[Path, Optional[Dict[str, Any]]]]: return [(path, payload) for _key, path, payload in entries] -def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int: +def recover_pending_to_db(session_db=None, *, session_resolver=None, + held_back: Optional[set] = None) -> int: """Replay flush-dir ``*.json`` files via ``SessionDB.append_message``, deleting each on success. ``session_db=None`` opens (and afterwards releases) the shared default ``state.db``. @@ -281,6 +282,9 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int: ``MessageEvent`` objects carry no ``session_id``, so without it every recovery lands in the skip branch. A returned ``db`` routes the append to the profile store owning the key (multiplexed gateways); ``None`` falls back to ``session_db``. Returns the number of messages recovered. + ``held_back`` (optional set) receives the session ids whose spool files this pass held back + after a failed replay, so the caller can have the live writer drain them before that session's + next row. """ flush_files = _order_flush_files(_get_flush_dir().glob("*.json")) if not flush_files: @@ -327,6 +331,8 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int: logger.info("Held back %d spooled transcript file(s) for the next start after a failed replay: %s", sum(blocked_sessions.values()), ", ".join(f"{sid} ({count})" for sid, count in blocked_sessions.items())) + if held_back is not None: + held_back.update(blocked_sessions) return recovered diff --git a/tests/gateway/test_shutdown_flush_recovery.py b/tests/gateway/test_shutdown_flush_recovery.py index 0776cb382734d..35c33f102bb9f 100644 --- a/tests/gateway/test_shutdown_flush_recovery.py +++ b/tests/gateway/test_shutdown_flush_recovery.py @@ -6,7 +6,10 @@ """ import json +import threading +import time from pathlib import Path +from types import SimpleNamespace from unittest.mock import MagicMock import pytest @@ -160,3 +163,33 @@ def append_message(**kwargs): assert first.exists() and second.exists() assert not other.exists() assert "Held back 2 spooled transcript file(s)" in caplog.text and "sess-1 (2)" in caplog.text + + +def test_held_back_files_drain_before_that_sessions_next_live_write(flush_dir, monkeypatch): + """The live writer drains a session's spool only when the store knows it has one. If boot + recovery does not say which sessions it held back, the next live row lands ahead of them.""" + import hermes_state_registry + from gateway.run import _recover_pending_flushes + from gateway.session import SessionStore + + _write_spool(flush_dir, "pending-bbb.json", "sess-1", + {"role": "user", "content": "old0"}, ts=100, seq=0) + _write_spool(flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "old1"}, ts=100, seq=1) + mock_db = MagicMock() + mock_db.append_message.side_effect = [RuntimeError("still locked at boot"), 1, 1, 1] + monkeypatch.setattr(hermes_state_registry, "acquire", lambda: mock_db) + monkeypatch.setattr(hermes_state_registry, "release_or_close", lambda _db: None) + store = object.__new__(SessionStore) + store._db = mock_db + store._transcript_retry_lock = threading.Lock() + store._dirty_transcripts = {} + store._transcript_append_failures = {} + store._fts_rebuild_last_attempt_at = time.monotonic() + + runner = SimpleNamespace(config=SimpleNamespace(multiplex_profiles=False), session_store=store) + assert _recover_pending_flushes(runner) == 0 + store.append_to_transcript("sess-1", {"role": "user", "content": "new-live"}) + + assert _contents(mock_db) == ["old0", "old0", "old1", "new-live"] + assert not list(flush_dir.glob("*.json")) From d4d5a8ed0305719b49fa25cfd89b56b49a5c7480 Mon Sep 17 00:00:00 2001 From: unsupportedpastels Date: Tue, 6 Oct 2026 16:23:22 +0000 Subject: [PATCH 11/18] fix(gateway): bound decoded payload memory during restart recovery Retain only sort keys and paths; re-read payloads one at a time for replay. Regression with 6 MiB of spool data fails on the previous head at 8.1 MiB peak allocations and passes with the metadata-only index. Review finding by @keeltrace. --- gateway/shutdown_flush.py | 20 +++++------- tests/gateway/test_shutdown_spool_limits.py | 34 +++++++++++++++++++++ 2 files changed, 42 insertions(+), 12 deletions(-) create mode 100644 tests/gateway/test_shutdown_spool_limits.py diff --git a/gateway/shutdown_flush.py b/gateway/shutdown_flush.py index 23509d393cb5a..c95e273ad089b 100644 --- a/gateway/shutdown_flush.py +++ b/gateway/shutdown_flush.py @@ -226,7 +226,7 @@ def _sort_number(value: Any) -> float: return number if math.isfinite(number) else 0.0 -def _order_flush_files(paths) -> list[tuple[Path, Optional[Dict[str, Any]]]]: +def _order_flush_files(paths) -> list[Path]: """Order recovery payloads by drop order — ``(ts, seq, filename)``. Recovery used to walk ``sorted(glob("*.json"))``, but spool files are @@ -241,9 +241,9 @@ def _order_flush_files(paths) -> list[tuple[Path, Optional[Dict[str, Any]]]]: :func:`drain_transcript_spool`'s ``sorted(entries, key=lambda e: e[:3])`` and the live drain and the cross-restart drain agree on replay order. - Returns ``(path, payload)`` pairs in replay order. A file whose payload - cannot be parsed sorts last, by name, and is returned with ``None`` so - the caller reports it through its own error handling. + Returns paths in replay order, retaining only ordering metadata, not decoded + payloads. Unparseable files sort last by name. The caller re-reads each + payload for replay and reports failures through its own error handling. """ entries = [] for path in paths: @@ -253,7 +253,7 @@ def _order_flush_files(paths) -> list[tuple[Path, Optional[Dict[str, Any]]]]: except (OSError, ValueError, RecursionError): payload = None if not isinstance(payload, dict): - entries.append(((1, 0.0, 0.0, path.name), path, None)) + entries.append(((1, 0.0, 0.0, path.name), path)) continue entries.append(( ( @@ -266,10 +266,9 @@ def _order_flush_files(paths) -> list[tuple[Path, Optional[Dict[str, Any]]]]: path.name, ), path, - payload, )) entries.sort(key=lambda entry: entry[0]) - return [(path, payload) for _key, path, payload in entries] + return [path for _key, path in entries] def recover_pending_to_db(session_db=None, *, session_resolver=None, @@ -298,15 +297,12 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None, # per-session property, so one unhealthy session must not hold back the others. blocked_sessions: Dict[str, int] = {} try: - for path, payload in flush_files: + for path in flush_files: # One unparseable payload or rejected append must only skip THIS file: the file is # never unlinked, so aborting the pass would re-poison every later boot. # utf-8-sig: our BOM-tolerant read fix for flush files. try: - if payload is None: - # Unparseable on the ordering pass: re-read so the failure is raised and - # reported here exactly as it was before. - payload = json.loads(path.read_text(encoding="utf-8-sig")) + payload = json.loads(path.read_text(encoding="utf-8-sig")) # Agent-history snapshots use a different schema (reason + # messages list) and are meant for manual operator recovery, # not automatic DB insertion. Skip them silently. diff --git a/tests/gateway/test_shutdown_spool_limits.py b/tests/gateway/test_shutdown_spool_limits.py new file mode 100644 index 0000000000000..75a6e1a2977eb --- /dev/null +++ b/tests/gateway/test_shutdown_spool_limits.py @@ -0,0 +1,34 @@ +"""Recovery must not retain every decoded spool payload while sorting.""" + +import tracemalloc +from pathlib import Path + +import pytest + +from gateway import shutdown_flush + + +def test_recovery_payload_memory_does_not_scale_with_backlog( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(shutdown_flush, "_get_flush_dir", lambda: tmp_path) + content = "x" * (256 * 1024) + for seq in range(24): + shutdown_flush.spool_dropped_transcript_message( + "sess", {"role": "user", "content": content, "timestamp": seq}, + ) + + class Sink: + def append_message(self, **kwargs: object) -> None: + assert kwargs["content"] == content + + tracemalloc.start() + try: + assert shutdown_flush.recover_pending_to_db(Sink()) == 24 + _, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + # Six MiB of payloads on disk; allow ample room for one decoded payload, + # its source text and the small ordering index, but not the entire backlog. + assert peak < 4 * 1024 * 1024, f"recovery retained {peak} bytes" + assert not list(tmp_path.glob("*.json")) From 4d7d3c06a26183640d4cbd895960abe4a82135de Mon Sep 17 00:00:00 2001 From: unsupportedpastels Date: Tue, 6 Oct 2026 16:24:05 +0000 Subject: [PATCH 12/18] fix(gateway): order shutdown slot heads before same-second overflow Treat an absent sequence as -1 during restart sorting, before the overflow tail starting at zero. This also repairs ordering for existing spool files, without requiring a new writer format. Frozen-clock adversarial-filename regression reproduces B,A,C before the fix and A,B,C afterwards. Review finding by @keeltrace. --- gateway/shutdown_flush.py | 4 +++- tests/gateway/test_shutdown_spool_limits.py | 20 ++++++++++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/gateway/shutdown_flush.py b/gateway/shutdown_flush.py index c95e273ad089b..4c2719754797c 100644 --- a/gateway/shutdown_flush.py +++ b/gateway/shutdown_flush.py @@ -262,7 +262,9 @@ def _order_flush_files(paths) -> list[Path]: # seq is per process (_TRANSCRIPT_SPOOL_SEQ) and ts has one-second resolution, so # files two processes spool in the same second can interleave; within one process # the order is exact. - _sort_number(payload.get("seq")), + # Slot heads have no seq; overflow tails start at zero. Interpret legacy + # heads as -1 too, so files already on disk recover before their tail. + _sort_number(payload.get("seq", -1)), path.name, ), path, diff --git a/tests/gateway/test_shutdown_spool_limits.py b/tests/gateway/test_shutdown_spool_limits.py index 75a6e1a2977eb..125407d577418 100644 --- a/tests/gateway/test_shutdown_spool_limits.py +++ b/tests/gateway/test_shutdown_spool_limits.py @@ -32,3 +32,23 @@ def append_message(self, **kwargs: object) -> None: # its source text and the small ordering index, but not the entire backlog. assert peak < 4 * 1024 * 1024, f"recovery retained {peak} bytes" assert not list(tmp_path.glob("*.json")) + + +def test_same_second_shutdown_slot_precedes_overflow( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + from types import SimpleNamespace + from unittest.mock import MagicMock + + monkeypatch.setattr(shutdown_flush, "_get_flush_dir", lambda: tmp_path) + monkeypatch.setattr(shutdown_flush.time, "time", lambda: 100) + names = iter(["zzz", "aaa", "bbb"]) + monkeypatch.setattr(shutdown_flush.uuid, "uuid4", lambda: SimpleNamespace(hex=next(names))) + assert shutdown_flush.flush_pending_to_file({"session": {"text": "A", "session_id": "sess"}}) == 1 + assert shutdown_flush.flush_overflow_to_file({"session": [ + {"text": "B", "session_id": "sess"}, {"text": "C", "session_id": "sess"}, + ]}) == 2 + db = MagicMock() + assert shutdown_flush.recover_pending_to_db(db) == 3 + assert [call.kwargs["content"] for call in db.append_message.call_args_list] == ["A", "B", "C"] + assert not list(tmp_path.glob("*.json")) From 02938cb9d520d9c495022b72e39d1ae4bf3d10af Mon Sep 17 00:00:00 2001 From: unsupportedpastels Date: Tue, 6 Oct 2026 16:59:40 +0000 Subject: [PATCH 13/18] fix(gateway): keep live transcript rows queued while older spool files fail to drain The live writer replayed a session's spool before each write but ignored a failed replay and wrote the new row anyway, so a second failure (boot hold-back, then the pre-write drain) gave the live row a lower row id than the older spooled rows. _drain_spooled_drops now reports files left on disk, and the write takes the existing retry path instead, leaving the row queued in memory behind them. Review finding by @ehz0ah. --- gateway/session_transcript.py | 19 ++++++---- tests/gateway/test_shutdown_flush_recovery.py | 35 +++++++++++++++++++ 2 files changed, 47 insertions(+), 7 deletions(-) diff --git a/gateway/session_transcript.py b/gateway/session_transcript.py index 05ce8dde709b3..2355a90348359 100644 --- a/gateway/session_transcript.py +++ b/gateway/session_transcript.py @@ -274,10 +274,12 @@ def _ack_head() -> bool: # DB write outside the retry lock so other sessions can append. while True: - # Spooled backlog (cap eviction or a stalled session) is older than ``msg``: replay it - # first so recovery keeps transcript order; a still-dead DB just fails both. - self._drain_spooled_drops(session_id) try: + # Spooled backlog (cap eviction, a stalled session, a boot hold-back) is older than + # ``msg``: replay it first. While any of it stays on disk ``msg`` stays queued, since + # writing it now would give it a lower row id than that backlog for good. + if self._drain_spooled_drops(session_id): + raise RuntimeError(f"older spooled transcript rows for {session_id} still pending") self._append_transcript_message(session_id, msg) except Exception as exc: from hermes_state import StateDbCorruptError, StateDbReplacedError @@ -371,12 +373,13 @@ def _spool_stalled_backlog(self, session_id: str, queue_session_id: str) -> int: self._dirty_transcripts.pop(queue_session_id, None) return spooled - def _drain_spooled_drops(self, session_id: str) -> None: - """Replay cap-dropped spooled transcript messages after DB recovery. Best-effort: replay - failures keep the spool files for the next successful flush; nothing here may raise.""" + def _drain_spooled_drops(self, session_id: str) -> bool: + """Replay cap-dropped spooled transcript messages after DB recovery; True while some of them + are still on disk. Best-effort: replay failures keep the spool files for the next successful + flush; nothing here may raise.""" spooled_sessions = getattr(self, "_spooled_drop_sessions", None) if not spooled_sessions or session_id not in spooled_sessions: - return + return False try: from gateway.shutdown_flush import drain_transcript_spool # Inside an outage the append that follows logs/escalates the same failure; the @@ -389,8 +392,10 @@ def _drain_spooled_drops(self, session_id: str) -> None: ) if not remaining: spooled_sessions.discard(session_id) + return bool(remaining) except Exception as exc: logger.warning("Failed to drain transcript spool for %s: %s", session_id, exc) + return False def spooled_drop_sessions(self) -> set: """The live set of sessions whose next transcript write drains the spool first. Boot recovery diff --git a/tests/gateway/test_shutdown_flush_recovery.py b/tests/gateway/test_shutdown_flush_recovery.py index 35c33f102bb9f..8cbf5f03c71e2 100644 --- a/tests/gateway/test_shutdown_flush_recovery.py +++ b/tests/gateway/test_shutdown_flush_recovery.py @@ -193,3 +193,38 @@ def test_held_back_files_drain_before_that_sessions_next_live_write(flush_dir, m assert _contents(mock_db) == ["old0", "old0", "old1", "new-live"] assert not list(flush_dir.glob("*.json")) + + +def test_live_write_waits_while_held_back_files_still_fail_to_drain(flush_dir, monkeypatch): + """If the pre-write drain fails again, the live row must stay queued in memory: writing it + while the older files are still on disk gives it a lower row id than they will get.""" + import hermes_state_registry + from gateway.run import _recover_pending_flushes + from gateway.session import SessionStore + + _write_spool(flush_dir, "pending-bbb.json", "sess-1", + {"role": "user", "content": "old0"}, ts=100, seq=0) + _write_spool(flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "old1"}, ts=100, seq=1) + mock_db = MagicMock() + mock_db.append_message.side_effect = [ + RuntimeError("still locked at boot"), RuntimeError("still locked at drain"), 1, 1, 1, 1] + monkeypatch.setattr(hermes_state_registry, "acquire", lambda: mock_db) + monkeypatch.setattr(hermes_state_registry, "release_or_close", lambda _db: None) + store = object.__new__(SessionStore) + store._db = mock_db + store._transcript_retry_lock = threading.Lock() + store._dirty_transcripts = {} + store._transcript_append_failures = {} + store._fts_rebuild_last_attempt_at = time.monotonic() + + runner = SimpleNamespace(config=SimpleNamespace(multiplex_profiles=False), session_store=store) + assert _recover_pending_flushes(runner) == 0 + store.append_to_transcript("sess-1", {"role": "user", "content": "new-live"}) + assert _contents(mock_db) == ["old0", "old0"] + assert len(list(flush_dir.glob("*.json"))) == 2 + + store.append_to_transcript("sess-1", {"role": "user", "content": "new-live-2"}) + assert _contents(mock_db) == ["old0", "old0", "old0", "old1", "new-live", "new-live-2"] + assert not list(flush_dir.glob("*.json")) + assert not store._dirty_transcripts From 7a54b7f710f35f15e2a6a5ddd09acbab2b1599df Mon Sep 17 00:00:00 2001 From: unsupportedpastels Date: Tue, 6 Oct 2026 17:03:24 +0000 Subject: [PATCH 14/18] fix(gateway): replay the pending spool before boot resume turns and queued inbound Boot recovery ran in main() only after runner.start() returned, by which point start() had already run the auto-resume turns and drained the inbound queued during startup. Those turns write live transcript rows, and a fresh store does not know which sessions have spooled rows, so the previous run's spool landed after them. _start_finish_wiring now runs _recover_pending_flushes just before _schedule_resume_pending_sessions(), and failures log at WARNING instead of being swallowed silently. Review finding by @dokterdok. --- gateway/run.py | 7 ----- gateway/run_startup.py | 9 +++++- tests/gateway/test_shutdown_flush_recovery.py | 30 +++++++++++++++++++ 3 files changed, 38 insertions(+), 8 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index 591473ce736eb..657c04fd93bd4 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -5968,13 +5968,6 @@ def _start_keepalive() -> None: if not success: _shutdown_gateway_health_export(runner) return False - - def _recover_pending() -> None: - recovered = _recover_pending_flushes(runner) - if recovered: - logger.info("Recovered %d pending message(s) from shutdown flush", recovered) - - _best_effort(_recover_pending) if runner.should_exit_cleanly: _shutdown_gateway_health_export(runner) if runner.exit_reason: diff --git a/gateway/run_startup.py b/gateway/run_startup.py index a20ba923f33cd..fbcf3863888d9 100644 --- a/gateway/run_startup.py +++ b/gateway/run_startup.py @@ -1507,7 +1507,8 @@ async def _start_post_connect_services(self, connected_count: int) -> None: async def _start_finish_wiring(self, connected_count: int) -> None: """Post-connect wiring: services, boot notifications, startup restore, recovered watchers.""" - from gateway.run import _planned_restart_notification_pending, _restart_notification_pending + from gateway.run import ( + _planned_restart_notification_pending, _recover_pending_flushes, _restart_notification_pending) await self._start_post_connect_services(connected_count) # Let fresh adapters settle before lifecycle sends (helps Discord thread deliveries). if connected_count > 0: @@ -1523,6 +1524,12 @@ async def _start_finish_wiring(self, connected_count: int) -> None: await self._await_startup_boot_sends( planned_restart_notification_pending=_planned_restart_notification_pending(), ) + # Replay the previous run's pending spool before resume turns and queued inbound write live + # rows: only this pass tells the store which sessions have spooled rows, and a live row + # written first would get a lower row id than them for good. + with _log_suppressed(logging.WARNING, "Pending-message recovery failed: %s"): + if recovered := _recover_pending_flushes(self): + logger.info("Recovered %d pending message(s) from shutdown flush", recovered) # Auto-resume restart-interrupted sessions (ledger-answered ones were cleared above); a failed # auto-resume stays visible on the next user message. self._schedule_resume_pending_sessions() diff --git a/tests/gateway/test_shutdown_flush_recovery.py b/tests/gateway/test_shutdown_flush_recovery.py index 8cbf5f03c71e2..e8ce49f06b0ca 100644 --- a/tests/gateway/test_shutdown_flush_recovery.py +++ b/tests/gateway/test_shutdown_flush_recovery.py @@ -228,3 +228,33 @@ def test_live_write_waits_while_held_back_files_still_fail_to_drain(flush_dir, m assert _contents(mock_db) == ["old0", "old0", "old0", "old1", "new-live", "new-live-2"] assert not list(flush_dir.glob("*.json")) assert not store._dirty_transcripts + + +def test_boot_recovery_runs_before_resume_turns_and_queued_inbound(monkeypatch): + """Resume turns and queued inbound write live rows. The store knows nothing about the previous + run's spool until recovery has run, so a live row written first lands ahead of it for good.""" + import asyncio + + import gateway.run as gateway_run + + order = [] + monkeypatch.setattr(gateway_run, "_recover_pending_flushes", + lambda runner: order.append("recover") or 0) + monkeypatch.setattr(gateway_run, "_restart_notification_pending", lambda: False) + monkeypatch.setattr(gateway_run, "_planned_restart_notification_pending", lambda: False) + + async def noop(*_args, **_kwargs): + return None + + async def finish_startup_restore(): + order.append("drain inbound") + + runner = SimpleNamespace( + _start_post_connect_services=noop, _await_startup_boot_sends=noop, + _schedule_resume_pending_sessions=lambda: order.append("resume"), + _finish_startup_restore=finish_startup_restore, + _send_session_db_warning_notifications=noop, _spawn_supervised=lambda *a, **k: None, + ) + asyncio.run(gateway_run.GatewayRunner._start_finish_wiring(runner, 0)) + + assert order == ["recover", "resume", "drain inbound"] From 376d2e134eb02478a4aedb8e0b710ecbc3bc87b5 Mon Sep 17 00:00:00 2001 From: unsupportedpastels Date: Tue, 6 Oct 2026 17:10:26 +0000 Subject: [PATCH 15/18] fix(gateway): keep the spool replay error classifiable when it blocks a live write Gating the live write on a pending spool raised a generic RuntimeError, hiding the replay's real failure from the FTS-rebuild, replaced-database and compression handlers, so a repairable FTS fault became a permanent stall. _drain_spooled_drops now returns the replay's own exception, the caller raises it, and after a successful FTS rebuild the loop drains the older spool again before writing the live row. Found in independent review. --- gateway/session_transcript.py | 40 +++++++++++++------ tests/gateway/test_shutdown_flush_recovery.py | 39 ++++++++++++++++++ 2 files changed, 66 insertions(+), 13 deletions(-) diff --git a/gateway/session_transcript.py b/gateway/session_transcript.py index 2355a90348359..809b24a97e19c 100644 --- a/gateway/session_transcript.py +++ b/gateway/session_transcript.py @@ -274,12 +274,15 @@ def _ack_head() -> bool: # DB write outside the retry lock so other sessions can append. while True: + spool_exc = None try: # Spooled backlog (cap eviction, a stalled session, a boot hold-back) is older than # ``msg``: replay it first. While any of it stays on disk ``msg`` stays queued, since - # writing it now would give it a lower row id than that backlog for good. - if self._drain_spooled_drops(session_id): - raise RuntimeError(f"older spooled transcript rows for {session_id} still pending") + # writing it now would give it a lower row id than that backlog for good. The replay's + # own error is raised so the repair/divert handling below still classifies it. + spool_exc = self._drain_spooled_drops(session_id) + if spool_exc is not None: + raise spool_exc self._append_transcript_message(session_id, msg) except Exception as exc: from hermes_state import StateDbCorruptError, StateDbReplacedError @@ -321,6 +324,8 @@ def _ack_head() -> bool: "no unique live child; not retrying", session_id) return if self._is_fts_corruption_error(exc) and self._rebuild_fts_once(): + if spool_exc is not None: + continue # repaired: drain the older spool again before ``msg`` try: self._append_transcript_message(session_id, msg) except Exception as retry_exc: @@ -373,13 +378,22 @@ def _spool_stalled_backlog(self, session_id: str, queue_session_id: str) -> int: self._dirty_transcripts.pop(queue_session_id, None) return spooled - def _drain_spooled_drops(self, session_id: str) -> bool: - """Replay cap-dropped spooled transcript messages after DB recovery; True while some of them - are still on disk. Best-effort: replay failures keep the spool files for the next successful - flush; nothing here may raise.""" + def _drain_spooled_drops(self, session_id: str) -> Exception | None: + """Replay cap-dropped spooled transcript messages after DB recovery; return the replay + failure while some of them are still on disk, else None. Best-effort: replay failures keep + the spool files for the next successful flush; nothing here may raise.""" spooled_sessions = getattr(self, "_spooled_drop_sessions", None) if not spooled_sessions or session_id not in spooled_sessions: - return False + return None + failures: list[Exception] = [] + + def replay(message: Dict[str, Any]) -> None: + try: + self._append_transcript_message(session_id, message) + except Exception as exc: + failures.append(exc) + raise + try: from gateway.shutdown_flush import drain_transcript_spool # Inside an outage the append that follows logs/escalates the same failure; the @@ -387,15 +401,15 @@ def _drain_spooled_drops(self, session_id: str) -> bool: with self._transcript_retry_lock: known_failing = bool(self._transcript_append_failures.get(session_id)) _replayed, remaining = drain_transcript_spool( - session_id, lambda message: self._append_transcript_message(session_id, message), - db_known_failing=known_failing, - ) + session_id, replay, db_known_failing=known_failing) if not remaining: spooled_sessions.discard(session_id) - return bool(remaining) + return None + return failures[-1] if failures else RuntimeError( + f"older spooled transcript rows for {session_id} still pending") except Exception as exc: logger.warning("Failed to drain transcript spool for %s: %s", session_id, exc) - return False + return None def spooled_drop_sessions(self) -> set: """The live set of sessions whose next transcript write drains the spool first. Boot recovery diff --git a/tests/gateway/test_shutdown_flush_recovery.py b/tests/gateway/test_shutdown_flush_recovery.py index e8ce49f06b0ca..e6f7004ce0aae 100644 --- a/tests/gateway/test_shutdown_flush_recovery.py +++ b/tests/gateway/test_shutdown_flush_recovery.py @@ -230,6 +230,45 @@ def test_live_write_waits_while_held_back_files_still_fail_to_drain(flush_dir, m assert not store._dirty_transcripts +def test_repairable_spool_replay_failure_still_reaches_fts_rebuild(flush_dir): + """A spool replay failure must keep its own type, so the FTS rebuild still runs. After the + repair the older spooled row is written before the live one.""" + from gateway.session import SessionStore + + _write_spool(flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "old"}, ts=100, seq=0) + + class RepairableDb: + def __init__(self): + self.broken, self.rows, self.repairs = True, [], 0 + + def append_message(self, **kwargs): + if self.broken: + raise RuntimeError("no such table: messages_fts") + self.rows.append(kwargs["content"]) + + def rebuild_fts(self): + self.repairs += 1 + self.broken = False + return 1 + + db = RepairableDb() + store = object.__new__(SessionStore) + store._db = db + store._transcript_retry_lock = threading.Lock() + store._dirty_transcripts = {} + store._transcript_append_failures = {} + store._fts_rebuild_last_attempt_at = None + store.spooled_drop_sessions().add("sess-1") + + store.append_to_transcript("sess-1", {"role": "user", "content": "live"}) + + assert db.repairs == 1 + assert db.rows == ["old", "live"] + assert not list(flush_dir.glob("*.json")) + assert not store._dirty_transcripts + + def test_boot_recovery_runs_before_resume_turns_and_queued_inbound(monkeypatch): """Resume turns and queued inbound write live rows. The store knows nothing about the previous run's spool until recovery has run, so a live row written first lands ahead of it for good.""" From ab8d36830c3643f57163254f122c3963f684a6f5 Mon Sep 17 00:00:00 2001 From: unsupportedpastels Date: Tue, 6 Oct 2026 17:56:27 +0000 Subject: [PATCH 16/18] fix(gateway): stop one malformed spooled row from stalling its session transcript_append_kwargs passed tool_call_id, tool_name, role, reasoning and platform_message_id to sqlite as they were, so a spooled dict or list in one of them failed to bind on every replay. With live writes now waiting behind a failing spool, that one row stopped the whole session persisting, across restarts too. Those fields are now made bindable, and a row the database still rejects on its own values is moved to *.json.bad and logged once instead of holding the session back, both at boot and in the live drain. Also from the same review: the live drain and the restart pass share one sort key (_spool_sort_key), so a string or NaN ts no longer raises in the live drain; boot recovery returns the sessions it held back and the store marks them under its lock (mark_spooled_drop_sessions) instead of handing out its set; _recover_pending_flushes moves out of gateway/run.py as shutdown_flush.recover_gateway_pending; the transcript_append_kwargs import sits outside the try; blocked_sessions is required; the agent-history reason is a constant; the recovery tests share one store fixture and compare replayed rows with the live writer's. Review findings by @kshitijk4poor. --- gateway/run.py | 29 --- gateway/run_startup.py | 7 +- gateway/session_transcript.py | 38 ++- gateway/shutdown_flush.py | 185 +++++++++----- tests/gateway/test_profile_spool_recovery.py | 5 +- tests/gateway/test_shutdown_flush_recovery.py | 236 ++++++++++++------ 6 files changed, 315 insertions(+), 185 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index 657c04fd93bd4..a98d8297b7dfa 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1651,35 +1651,6 @@ def _multiplex_profile_homes(config: object) -> list[tuple[str, "Path"]]: return list(profiles_to_serve(multiplex=True)) -def _recover_pending_flushes(runner) -> int: - """Replay every ``pending_messages`` spool this gateway owns into state.db; return the count. - - ``_get_flush_dir`` follows the active HERMES_HOME, so a routed turn on a multiplexed gateway spools - its stalled transcript backlog under ``profiles//`` and the runtime drain forgets it on - restart. After the launch home, replay each served profile inside its own home so the default - store ``recover_pending_to_db`` opens is that profile's state.db (#123584). - """ - from gateway.shutdown_flush import recover_pending_to_db - from hermes_constants import get_hermes_home, reset_hermes_home_override, set_hermes_home_override - - resolver, held_back = runner.session_store.resolve_session_id_for_key, runner.session_store.spooled_drop_sessions() - recovered = recover_pending_to_db(session_resolver=resolver, held_back=held_back) - if not getattr(runner.config, "multiplex_profiles", False): - return recovered - launch_home = Path(get_hermes_home()).resolve() - for name, home in _multiplex_profile_homes(runner.config): - if Path(home).resolve() == launch_home or not (Path(home) / "pending_messages").is_dir(): - continue - token = set_hermes_home_override(str(home)) - try: - recovered += recover_pending_to_db(session_resolver=resolver, held_back=held_back) - except Exception: # one profile's unreadable spool must not strand the others' - logger.warning("Pending-message recovery failed for profile %s", name, exc_info=True) - finally: - reset_hermes_home_override(token) - return recovered - - def _cron_tick_profile_homes(config: object) -> list[tuple[str, "Path"]]: """Profile homes the in-process ticker visits: the served set PLUS the process-active profile: ``profiles_to_serve`` lists default + every live named profile, but a ``--profile diff --git a/gateway/run_startup.py b/gateway/run_startup.py index fbcf3863888d9..94b941728881a 100644 --- a/gateway/run_startup.py +++ b/gateway/run_startup.py @@ -1507,8 +1507,8 @@ async def _start_post_connect_services(self, connected_count: int) -> None: async def _start_finish_wiring(self, connected_count: int) -> None: """Post-connect wiring: services, boot notifications, startup restore, recovered watchers.""" - from gateway.run import ( - _planned_restart_notification_pending, _recover_pending_flushes, _restart_notification_pending) + from gateway.run import _planned_restart_notification_pending, _restart_notification_pending + from gateway.shutdown_flush import recover_gateway_pending await self._start_post_connect_services(connected_count) # Let fresh adapters settle before lifecycle sends (helps Discord thread deliveries). if connected_count > 0: @@ -1528,8 +1528,7 @@ async def _start_finish_wiring(self, connected_count: int) -> None: # rows: only this pass tells the store which sessions have spooled rows, and a live row # written first would get a lower row id than them for good. with _log_suppressed(logging.WARNING, "Pending-message recovery failed: %s"): - if recovered := _recover_pending_flushes(self): - logger.info("Recovered %d pending message(s) from shutdown flush", recovered) + recover_gateway_pending(self) # Auto-resume restart-interrupted sessions (ledger-answered ones were cleared above); a failed # auto-resume stays visible on the next user message. self._schedule_resume_pending_sessions() diff --git a/gateway/session_transcript.py b/gateway/session_transcript.py index 809b24a97e19c..da8e114e28f47 100644 --- a/gateway/session_transcript.py +++ b/gateway/session_transcript.py @@ -5,6 +5,7 @@ from __future__ import annotations import contextlib +import json import logging import threading import time @@ -41,6 +42,19 @@ def _spool_dropped(session_id: str, message: Dict[str, Any]): "reasoning", "reasoning_content", "reasoning_details", "codex_reasoning_items", "codex_message_items", ) +# Message keys bound straight to a TEXT column. A spooled message is JSON from disk, and a dict or list +# here makes sqlite refuse the whole row. +_BOUND_TEXT_KEYS = ("role", "tool_name", "tool_call_id", "reasoning", "reasoning_content", "platform_message_id") + + +def _bindable_text(value: Any) -> Any: + """*value* when sqlite can bind it as is, else its JSON text (``str`` if that fails too).""" + if value is None or isinstance(value, (str, int, float)): + return value + try: + return json.dumps(value, default=str) + except (TypeError, ValueError): + return str(value) def transcript_append_kwargs(session_id: str, message: Dict[str, Any], *, fallback_ts: Any = None) -> Dict[str, Any]: @@ -49,10 +63,11 @@ def transcript_append_kwargs(session_id: str, message: Dict[str, Any], *, fallba message lands as the row the live drain would have written. Fields are listed rather than splatted: a spooled message is arbitrary JSON from disk, and an unexpected key would raise and abort the replay. *fallback_ts* applies only to a message with no timestamp at all, because a - truthiness test would rewrite epoch 0.""" + truthiness test would rewrite epoch 0. Values bound straight to a TEXT column are made bindable + (:func:`_bindable_text`), so one malformed field cannot make the row unwritable.""" is_assistant = message.get("role") == "assistant" timestamp = message.get("timestamp") - return { + kwargs = { "session_id": session_id, "role": message.get("role", "unknown"), "content": message.get("content"), @@ -71,6 +86,9 @@ def transcript_append_kwargs(session_id: str, message: Dict[str, Any], *, fallba "display_kind": message.get("display_kind"), "display_metadata": message.get("display_metadata"), } + for key in _BOUND_TEXT_KEYS: + kwargs[key] = _bindable_text(kwargs[key]) + return kwargs class SessionTranscriptMixin: @@ -371,8 +389,8 @@ def _spool_stalled_backlog(self, session_id: str, queue_session_id: str) -> int: break spooled += 1 if spooled: - self._lazy("_spooled_drop_sessions", set).add(session_id) with self._transcript_retry_lock: + self._lazy("_spooled_drop_sessions", set).add(session_id) del pending[:spooled] if not pending: self._dirty_transcripts.pop(queue_session_id, None) @@ -403,7 +421,8 @@ def replay(message: Dict[str, Any]) -> None: _replayed, remaining = drain_transcript_spool( session_id, replay, db_known_failing=known_failing) if not remaining: - spooled_sessions.discard(session_id) + with self._transcript_retry_lock: + spooled_sessions.discard(session_id) return None return failures[-1] if failures else RuntimeError( f"older spooled transcript rows for {session_id} still pending") @@ -411,10 +430,13 @@ def replay(message: Dict[str, Any]) -> None: logger.warning("Failed to drain transcript spool for %s: %s", session_id, exc) return None - def spooled_drop_sessions(self) -> set: - """The live set of sessions whose next transcript write drains the spool first. Boot recovery - adds the sessions it held back, so their next live row cannot land ahead of those files.""" - return self._lazy("_spooled_drop_sessions", set) + def mark_spooled_drop_sessions(self, session_ids) -> None: + """Have each session's next transcript write drain the on-disk spool first. Boot recovery marks + the sessions it held back, so their next live row cannot land ahead of those files.""" + if not session_ids: + return + with self._transcript_retry_lock: + self._lazy("_spooled_drop_sessions", set).update(session_ids) def _append_transcript_message(self, session_id: str, message: Dict[str, Any]) -> None: """Write one transcript row. Caller handles retry queuing.""" diff --git a/gateway/shutdown_flush.py b/gateway/shutdown_flush.py index 4c2719754797c..10836a2b2bd1b 100644 --- a/gateway/shutdown_flush.py +++ b/gateway/shutdown_flush.py @@ -4,7 +4,7 @@ ``agent._session_messages`` are the only surviving copies; shutdown ``.clear()`` would drop them. All hooks write atomic JSON payloads under ``/pending_messages/``: ``flush_pending_to_file`` / ``flush_overflow_to_file`` (queue head / FIFO tail, before clear), -``recover_pending_to_db`` (after ``runner.start()``; replays via ``SessionDB.append_message``, +``recover_pending_to_db`` (at startup, before resume turns; replays via ``SessionDB.append_message``, deletes each file on success), ``flush_agent_history_to_file`` (DB flush raised), ``spool_dropped_transcript_message`` / ``drain_transcript_spool``. """ @@ -17,6 +17,7 @@ import logging import math import os +import sqlite3 import time import uuid from pathlib import Path @@ -28,6 +29,10 @@ # operation. Payloads carry the full transcript message dict for verbatim replay. # See #78182. TRANSCRIPT_CAP_DROP_REASON = "transcript_cap_drop" +# Reason tag for agent-history snapshots: kept for manual operator recovery, never replayed. +AGENT_HISTORY_REASON = "shutdown-with-unpersisted-agent-history" +# Suffix for a spool file whose row the database rejected; no replay glob matches it. +QUARANTINE_SUFFIX = ".bad" # Monotonic tiebreaker so same-second spool files replay in drop order. _TRANSCRIPT_SPOOL_SEQ = itertools.count() @@ -137,9 +142,11 @@ def drain_transcript_spool(session_id: str, replay, *, db_known_failing: bool = """Replay cap-dropped transcript messages spooled for *session_id*; return ``(replayed, remaining)``. ``replay(message_dict)`` runs per message in drop order; a spool file is deleted only after its replay succeeds. The first failure stops the drain (the DB is likely still - unhealthy) and keeps the rest for retry. With ``db_known_failing`` (the caller's last write - already failed and is being logged/escalated) a replay failure is expected and logs at DEBUG, - so a stalled session does not add one WARNING per append on top of its ERROR (#114266). + unhealthy) and keeps the rest for retry, unless the DB rejected that row's own values + (:func:`is_row_rejection`): it is quarantined and the drain goes on. With ``db_known_failing`` + (the caller's last write already failed and is being logged/escalated) a replay failure is + expected and logs at DEBUG, so a stalled session does not add one WARNING per append on top of + its ERROR (#114266). """ try: candidates = list(_get_flush_dir().glob("pending-*.json")) @@ -163,12 +170,15 @@ def drain_transcript_spool(session_id: str, replay, *, db_known_failing: bool = logger.warning("Removing structurally invalid transcript spool file %s", path) path.unlink(missing_ok=True) continue - entries.append((payload.get("ts", 0), payload.get("seq", 0), path.name, path, message)) - ordered, replayed, remaining = sorted(entries, key=lambda e: e[:3]), 0, 0 - for idx, (_ts, _seq, _name, path, message) in enumerate(ordered): + entries.append((_spool_sort_key(payload, path.name), path, message)) + ordered, replayed, remaining = sorted(entries, key=lambda e: e[0]), 0, 0 + for idx, (_key, path, message) in enumerate(ordered): try: replay(message) except Exception as exc: + if is_row_rejection(exc): # no retry can write it, so it must not hold the session back + _quarantine_spool_file(path, session_id, exc) + continue (logger.debug if db_known_failing else logger.warning)( "Replay of spooled transcript message %s for %s failed; " "keeping spool file for retry: %s", path, session_id, exc) @@ -226,24 +236,48 @@ def _sort_number(value: Any) -> float: return number if math.isfinite(number) else 0.0 +def _spool_sort_key(payload: Dict[str, Any], name: str) -> tuple: + """Drop-order key ``(ts, seq, filename)`` for one spool payload. The live drain and the restart + pass both sort on it, so they replay a session's files in the same order.""" + # seq is per process (_TRANSCRIPT_SPOOL_SEQ) and ts has one-second resolution, so files two + # processes spool in the same second can interleave; within one process the order is exact. + # Slot heads have no seq and overflow tails start at zero, so a missing seq sorts as -1. + return _sort_number(payload.get("ts")), _sort_number(payload.get("seq", -1)), name + + +def is_row_rejection(exc: BaseException) -> bool: + """True when the database refused this row's own values, so no retry can succeed: a value sqlite + cannot bind, or a malformed field that fails before the write. Lock, I/O, corruption and routing + failures stay retryable, and so do constraint failures, which can be session-wide.""" + if isinstance(exc, sqlite3.InterfaceError): + return True + if isinstance(exc, sqlite3.ProgrammingError): + return "binding parameter" in str(exc).lower() + return isinstance(exc, (TypeError, ValueError, AttributeError)) + + +def _quarantine_spool_file(path: Path, session_id: str, exc: BaseException) -> None: + """Move a spool file whose row the database rejected out of every replay, so it cannot hold its + session's later rows back on every write and every restart. It stays on disk for manual recovery.""" + target = path.with_name(path.name + QUARANTINE_SUFFIX) + try: + path.replace(target) + except OSError as move_exc: + logger.error("Spooled transcript row for %s in %s was rejected by the database (%s) and could " + "not be quarantined: %s", session_id, path, exc, move_exc) + return + logger.warning("Spooled transcript row for %s was rejected by the database (%s); moved it to %s " + "for manual recovery and continuing with the session's later rows", + session_id, exc, target) + + def _order_flush_files(paths) -> list[Path]: - """Order recovery payloads by drop order — ``(ts, seq, filename)``. - - Recovery used to walk ``sorted(glob("*.json"))``, but spool files are - named ``pending-.json`` (see :func:`_write_payload`), so filename - order is effectively random. ``SessionDB`` restores a conversation by - AUTOINCREMENT id — true insertion order, never timestamp — so replaying - in filename order permanently scrambles the recovered transcript and can - separate an assistant tool call from its result. - - The payloads already carry ``ts`` and ``seq`` - (:func:`spool_dropped_transcript_message`), so this mirrors - :func:`drain_transcript_spool`'s ``sorted(entries, key=lambda e: e[:3])`` - and the live drain and the cross-restart drain agree on replay order. - - Returns paths in replay order, retaining only ordering metadata, not decoded - payloads. Unparseable files sort last by name. The caller re-reads each - payload for replay and reports failures through its own error handling. + """Order recovery payloads by :func:`_spool_sort_key`, as :func:`drain_transcript_spool` does. + ``SessionDB`` restores a conversation by AUTOINCREMENT id and spool files have random names, so + any other order permanently scrambles the recovered transcript. + + Keeps only the key and path, not decoded payloads; the caller re-reads each file to replay it. + Unparseable files sort last by name. """ entries = [] for path in paths: @@ -252,29 +286,21 @@ def _order_flush_files(paths) -> list[Path]: # OSError: unreadable; ValueError: bad JSON or bytes; RecursionError: nesting too deep. except (OSError, ValueError, RecursionError): payload = None - if not isinstance(payload, dict): + if isinstance(payload, dict): + entries.append(((0, *_spool_sort_key(payload, path.name)), path)) + else: entries.append(((1, 0.0, 0.0, path.name), path)) - continue - entries.append(( - ( - 0, - _sort_number(payload.get("ts")), - # seq is per process (_TRANSCRIPT_SPOOL_SEQ) and ts has one-second resolution, so - # files two processes spool in the same second can interleave; within one process - # the order is exact. - # Slot heads have no seq; overflow tails start at zero. Interpret legacy - # heads as -1 too, so files already on disk recover before their tail. - _sort_number(payload.get("seq", -1)), - path.name, - ), - path, - )) entries.sort(key=lambda entry: entry[0]) return [path for _key, path in entries] -def recover_pending_to_db(session_db=None, *, session_resolver=None, - held_back: Optional[set] = None) -> int: +def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int: + """Replay flush-dir ``*.json`` files into state.db; return the number of messages recovered. + See :func:`recover_pending_spool`, which also reports the sessions it held back.""" + return recover_pending_spool(session_db, session_resolver=session_resolver)[0] + + +def recover_pending_spool(session_db=None, *, session_resolver=None) -> tuple[int, set[str]]: """Replay flush-dir ``*.json`` files via ``SessionDB.append_message``, deleting each on success. ``session_db=None`` opens (and afterwards releases) the shared default ``state.db``. @@ -282,14 +308,13 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None, ``SessionStore.resolve_session_id_for_key``) is required for real flush files: adapter ``MessageEvent`` objects carry no ``session_id``, so without it every recovery lands in the skip branch. A returned ``db`` routes the append to the profile store owning the key (multiplexed - gateways); ``None`` falls back to ``session_db``. Returns the number of messages recovered. - ``held_back`` (optional set) receives the session ids whose spool files this pass held back - after a failed replay, so the caller can have the live writer drain them before that session's - next row. + gateways); ``None`` falls back to ``session_db``. Returns ``(recovered, held_back)``: the number of + messages recovered, and the session ids whose spool files this pass held back after a failed + replay, so the live writer can drain them before that session's next row. """ flush_files = _order_flush_files(_get_flush_dir().glob("*.json")) if not flush_files: - return 0 + return 0, set() own_db = session_db is None if own_db: from hermes_state_registry import acquire @@ -302,19 +327,19 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None, for path in flush_files: # One unparseable payload or rejected append must only skip THIS file: the file is # never unlinked, so aborting the pass would re-poison every later boot. - # utf-8-sig: our BOM-tolerant read fix for flush files. try: payload = json.loads(path.read_text(encoding="utf-8-sig")) # Agent-history snapshots use a different schema (reason + # messages list) and are meant for manual operator recovery, # not automatic DB insertion. Skip them silently. - if payload.get("reason") == "shutdown-with-unpersisted-agent-history": + if payload.get("reason") == AGENT_HISTORY_REASON: continue if _recover_one_payload(session_db, path, payload, session_resolver=session_resolver, blocked_sessions=blocked_sessions): recovered += 1 path.unlink(missing_ok=True) + # health: allow BLE001 -- per-file boundary moved unchanged from recover_pending_to_db; a traceback per locked-DB file would only add noise except Exception as exc: logger.warning("Failed to recover pending message from %s: %s", path, exc) finally: @@ -329,38 +354,38 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None, logger.info("Held back %d spooled transcript file(s) for the next start after a failed replay: %s", sum(blocked_sessions.values()), ", ".join(f"{sid} ({count})" for sid, count in blocked_sessions.items())) - if held_back is not None: - held_back.update(blocked_sessions) - return recovered + return recovered, set(blocked_sessions) def _recover_one_payload(session_db, path: Path, payload: Dict[str, Any], *, - session_resolver=None, blocked_sessions: Optional[Dict[str, int]] = None) -> bool: - """Append one flush payload to ``session_db``; False (file kept) when structurally invalid.""" - # Cap-dropped transcript payloads carry the full message dict keyed by session_id — replay directly - # (#78182). This handles spool files that were never drained before a restart. + session_resolver=None, blocked_sessions: Dict[str, int]) -> bool: + """Append one flush payload to ``session_db``; False when it was not replayed (the file is kept, + or quarantined when the database rejected the row itself).""" + # Cap-dropped transcript payloads carry the full message dict keyed by session_id: replay it + # directly (#78182). This handles spool files that were never drained before a restart. if payload.get("reason") == TRANSCRIPT_CAP_DROP_REASON: - # Cap-dropped payloads carry the full message dict keyed by session_id — replay directly. + from gateway.session_transcript import transcript_append_kwargs data = payload.get("data", {}) or {} spooled_sid, message = data.get("session_id", ""), data.get("message") if not spooled_sid or not isinstance(message, dict): logger.warning("Cannot recover structurally invalid transcript spool " "file %s; preserved for manual inspection", path) return False - if blocked_sessions is not None and spooled_sid in blocked_sessions: + if spooled_sid in blocked_sessions: # An older message for this session could not be replayed. Writing this one now would give # it a lower row id than the message it follows, permanently inverting the transcript, so # leave it for the next start. blocked_sessions[spooled_sid] += 1 return False try: - from gateway.session_transcript import transcript_append_kwargs session_db.append_message(**transcript_append_kwargs(spooled_sid, message, fallback_ts=payload.get("ts"))) - except Exception: + except Exception as exc: + if is_row_rejection(exc): # no retry can write it, so it must not hold the session back + _quarantine_spool_file(path, spooled_sid, exc) + return False # Same contract as drain_transcript_spool: stop this session's replay on the first failure # and keep the remaining spool files for the next attempt. - if blocked_sessions is not None: - blocked_sessions[spooled_sid] = 1 + blocked_sessions[spooled_sid] = 1 raise return True session_key, data = payload.get("session_key", ""), payload.get("data", {}) @@ -393,6 +418,40 @@ def _recover_one_payload(session_db, path: Path, payload: Dict[str, Any], *, return True +def recover_gateway_pending(runner) -> int: + """Replay every ``pending_messages`` spool this gateway owns into state.db; return the count. + + ``_get_flush_dir`` follows the active HERMES_HOME, so a routed turn on a multiplexed gateway spools + its stalled transcript backlog under ``profiles//`` and the runtime drain forgets it on + restart. After the launch home, replay each served profile inside its own home so the default + store ``recover_pending_spool`` opens is that profile's state.db (#123584). The sessions a pass + holds back are marked on the session store, so their next live row drains the spool first. + """ + from gateway.run import _multiplex_profile_homes + from hermes_constants import get_hermes_home, reset_hermes_home_override, set_hermes_home_override + + store = runner.session_store + recovered, held_back = recover_pending_spool(session_resolver=store.resolve_session_id_for_key) + store.mark_spooled_drop_sessions(held_back) + if not getattr(runner.config, "multiplex_profiles", False): + return recovered + launch_home = Path(get_hermes_home()).resolve() + for name, home in _multiplex_profile_homes(runner.config): + if Path(home).resolve() == launch_home or not (Path(home) / "pending_messages").is_dir(): + continue + token = set_hermes_home_override(str(home)) + try: + count, held_back = recover_pending_spool(session_resolver=store.resolve_session_id_for_key) + except Exception: # one profile's unreadable spool must not strand the others' + logger.warning("Pending-message recovery failed for profile %s", name, exc_info=True) + continue + finally: + reset_hermes_home_override(token) + recovered += count + store.mark_spooled_drop_sessions(held_back) + return recovered + + def flush_agent_history_to_file(session_id: Optional[str], history: list) -> None: """Best-effort dump of an agent's in-memory transcript before teardown. Used when ``_flush_messages_to_session_db`` raises (e.g. FTS/SQLite corruption): the transcript is written @@ -410,7 +469,7 @@ def flush_agent_history_to_file(session_id: Optional[str], history: list) -> Non except Exception: continue _write_payload(flush_dir, { - "reason": "shutdown-with-unpersisted-agent-history", "issue": "#72680", + "reason": AGENT_HISTORY_REASON, "issue": "#72680", "session_id": session_id, "count": len(snapshot), "messages": snapshot, }) logger.warning("Preserved %d in-memory message(s) for session %s " diff --git a/tests/gateway/test_profile_spool_recovery.py b/tests/gateway/test_profile_spool_recovery.py index 0afa8d6d583fb..de0da4e4702e4 100644 --- a/tests/gateway/test_profile_spool_recovery.py +++ b/tests/gateway/test_profile_spool_recovery.py @@ -16,9 +16,8 @@ import agent.secret_scope as ss from gateway.config import GatewayConfig from gateway.platforms.base import Platform, SessionSource -from gateway.run import _recover_pending_flushes from gateway.session import SessionEntry, SessionStore -from gateway.shutdown_flush import spool_dropped_transcript_message +from gateway.shutdown_flush import recover_gateway_pending, spool_dropped_transcript_message from hermes_constants import get_hermes_home, reset_hermes_home_override, set_hermes_home_override @@ -73,7 +72,7 @@ def test_boot_recovery_replays_a_routed_profiles_spooled_backlog(multiplex_homes _spool_under(root, default_sid, "default while locked") _spool_under(profile, work_sid, "work while locked") - recovered = _recover_pending_flushes(SimpleNamespace(config=store.config, session_store=store)) + recovered = recover_gateway_pending(SimpleNamespace(config=store.config, session_store=store)) assert recovered == 2 assert [m["content"] for m in default_db.get_messages(default_sid)] == ["default while locked"] diff --git a/tests/gateway/test_shutdown_flush_recovery.py b/tests/gateway/test_shutdown_flush_recovery.py index e6f7004ce0aae..23d58aefcd3b6 100644 --- a/tests/gateway/test_shutdown_flush_recovery.py +++ b/tests/gateway/test_shutdown_flush_recovery.py @@ -6,16 +6,21 @@ """ import json -import threading -import time +import math +import sqlite3 from pathlib import Path from types import SimpleNamespace -from unittest.mock import MagicMock +from typing import Any +from unittest.mock import MagicMock, patch import pytest from gateway.shutdown_flush import ( + QUARANTINE_SUFFIX, TRANSCRIPT_CAP_DROP_REASON, + _order_flush_files, + drain_transcript_spool, + recover_gateway_pending, recover_pending_to_db, ) @@ -31,14 +36,48 @@ def flush_dir(tmp_path, monkeypatch): return directory +@pytest.fixture +def make_store(tmp_path): + """Build a real ``SessionStore`` whose transcript writes go to *db*. It goes through the + constructor, so new store state does not break these tests.""" + from gateway.config import GatewayConfig + from gateway.session import SessionStore + + def build(db): + with patch("gateway.session.SessionStore._ensure_loaded"): + store = SessionStore(sessions_dir=tmp_path / "sessions", config=GatewayConfig()) + store._loaded = True + store._db = db + return store + + return build + + +@pytest.fixture +def boot_db(monkeypatch): + """Make boot recovery's default state.db the given handle.""" + import hermes_state_registry + + def use(db): + monkeypatch.setattr(hermes_state_registry, "acquire", lambda: db) + monkeypatch.setattr(hermes_state_registry, "release_or_close", lambda _db: None) + + return use + + +def _boot(store) -> int: + runner = SimpleNamespace(config=SimpleNamespace(multiplex_profiles=False), session_store=store) + return recover_gateway_pending(runner) + + def _write_spool( flush_dir: Path, name: str, session_id: str, message: dict, *, - ts: int, - seq: int, + ts: Any, + seq: Any, ) -> Path: """Write one cap-drop spool payload under an explicit file name. @@ -94,46 +133,60 @@ def test_an_out_of_range_ordering_field_does_not_abort_recovery(flush_dir): assert sorted(_contents(mock_db)) == ["huge ts", "valid"] -def test_a_replayed_row_is_the_row_the_live_writer_writes(flush_dir): - """Every field the live drain persists survives the restart round trip: losing - tool_call_id orphans a tool result, losing api_content makes the next replay diverge, - and reasoning stays assistant-only. A missing timestamp falls back to the payload - clock, but epoch 0 is a real timestamp.""" +def test_live_drain_and_restart_pass_order_malformed_files_alike(flush_dir): + """Every live write waits on the live drain, so a ``ts`` it cannot compare must not raise there, + and both passes must replay a session's files in the same order.""" + _write_spool(flush_dir, "pending-a.json", "sess-1", + {"role": "user", "content": "string ts"}, ts="later", seq=0) + _write_spool(flush_dir, "pending-b.json", "sess-1", + {"role": "user", "content": "nan ts"}, ts=math.nan, seq=1) + _write_spool(flush_dir, "pending-c.json", "sess-1", + {"role": "user", "content": "ts 5"}, ts=5, seq=2) + + restart = [json.loads(path.read_text())["data"]["message"]["content"] + for path in _order_flush_files(flush_dir.glob("*.json"))] + live = [] + assert drain_transcript_spool("sess-1", lambda message: live.append(message["content"])) == (3, 0) + + assert live == restart == ["string ts", "nan ts", "ts 5"] + + +def test_a_replayed_row_is_the_row_the_live_writer_writes(flush_dir, make_store): + """A replayed spool row must be the row the live writer would have written for the same + message: losing tool_call_id orphans a tool result, losing api_content makes the next replay + diverge. Only a message with no timestamp differs: it takes the payload clock, and epoch 0 is + a real timestamp.""" tool_calls = [{"id": "call-1", "type": "function", "function": {"name": "send_payment", "arguments": "{}"}}] - _write_spool(flush_dir, "pending-ccc.json", "sess-1", { + assistant = { "role": "assistant", "content": None, "tool_calls": tool_calls, "reasoning": "deliberating", "reasoning_content": "chain", "reasoning_details": [{"type": "text"}], "codex_reasoning_items": [{"id": "r1"}], "codex_message_items": [{"id": "m1"}], "platform_message_id": "tg-42", "observed": True, "timestamp": 0, "api_content": "exact bytes sent to the API", "display_kind": "internal_notification", - }, ts=100, seq=0) - _write_spool(flush_dir, "pending-bbb.json", "sess-1", { - "role": "tool", "content": "receipt-1", "tool_call_id": "call-1", "tool_name": "send_payment", - }, ts=100, seq=1) - _write_spool(flush_dir, "pending-aaa.json", "sess-1", { - "role": "user", "content": "hi", "reasoning": "leaked", "message_id": "tg-7", - }, ts=999, seq=2) - - mock_db = MagicMock() - assert recover_pending_to_db(mock_db) == 3 - assistant, tool, user = (c.kwargs for c in mock_db.append_message.call_args_list) - - assert assistant["content"] is None # a tool-call row legitimately has no content - assert assistant["tool_calls"] == tool_calls - assert (assistant["reasoning"], assistant["reasoning_content"]) == ("deliberating", "chain") - assert assistant["reasoning_details"] == [{"type": "text"}] - assert assistant["codex_reasoning_items"] == [{"id": "r1"}] - assert assistant["codex_message_items"] == [{"id": "m1"}] - assert (assistant["platform_message_id"], assistant["observed"]) == ("tg-42", True) - assert assistant["api_content"] == "exact bytes sent to the API" - assert assistant["display_kind"] == "internal_notification" - assert assistant["timestamp"] == 0 - assert (tool["tool_call_id"], tool["tool_name"]) == ("call-1", "send_payment") - assert user["reasoning"] is None - assert user["platform_message_id"] == "tg-7" - assert user["timestamp"] == 999 + } + tool = {"role": "tool", "content": "receipt-1", "tool_call_id": "call-1", "tool_name": "send_payment"} + user = {"role": "user", "content": "hi", "reasoning": "leaked", "message_id": "tg-7"} + _write_spool(flush_dir, "pending-ccc.json", "sess-1", assistant, ts=100, seq=0) + _write_spool(flush_dir, "pending-bbb.json", "sess-1", tool, ts=100, seq=1) + _write_spool(flush_dir, "pending-aaa.json", "sess-1", user, ts=999, seq=2) + + recovery_db, live_db = MagicMock(), MagicMock() + assert recover_pending_to_db(recovery_db) == 3 + store = make_store(live_db) + for message in (assistant, tool, user): + store._append_transcript_message("sess-1", message) + + replayed = [c.kwargs for c in recovery_db.append_message.call_args_list] + written = [c.kwargs for c in live_db.append_message.call_args_list] + assert [row.pop("timestamp") for row in replayed] == [0, 100, 999] + assert [row.pop("timestamp") for row in written] == [0, None, None] + assert replayed == written + assert replayed[0]["tool_calls"] == tool_calls + assert replayed[0]["api_content"] == "exact bytes sent to the API" + assert (replayed[1]["tool_call_id"], replayed[1]["tool_name"]) == ("call-1", "send_payment") + assert (replayed[2]["reasoning"], replayed[2]["platform_message_id"]) == (None, "tg-7") def test_a_failed_replay_holds_back_that_sessions_later_messages_only(flush_dir, caplog): @@ -165,43 +218,28 @@ def append_message(**kwargs): assert "Held back 2 spooled transcript file(s)" in caplog.text and "sess-1 (2)" in caplog.text -def test_held_back_files_drain_before_that_sessions_next_live_write(flush_dir, monkeypatch): +def test_held_back_files_drain_before_that_sessions_next_live_write(flush_dir, make_store, boot_db): """The live writer drains a session's spool only when the store knows it has one. If boot recovery does not say which sessions it held back, the next live row lands ahead of them.""" - import hermes_state_registry - from gateway.run import _recover_pending_flushes - from gateway.session import SessionStore - _write_spool(flush_dir, "pending-bbb.json", "sess-1", {"role": "user", "content": "old0"}, ts=100, seq=0) _write_spool(flush_dir, "pending-aaa.json", "sess-1", {"role": "user", "content": "old1"}, ts=100, seq=1) mock_db = MagicMock() mock_db.append_message.side_effect = [RuntimeError("still locked at boot"), 1, 1, 1] - monkeypatch.setattr(hermes_state_registry, "acquire", lambda: mock_db) - monkeypatch.setattr(hermes_state_registry, "release_or_close", lambda _db: None) - store = object.__new__(SessionStore) - store._db = mock_db - store._transcript_retry_lock = threading.Lock() - store._dirty_transcripts = {} - store._transcript_append_failures = {} - store._fts_rebuild_last_attempt_at = time.monotonic() + boot_db(mock_db) + store = make_store(mock_db) - runner = SimpleNamespace(config=SimpleNamespace(multiplex_profiles=False), session_store=store) - assert _recover_pending_flushes(runner) == 0 + assert _boot(store) == 0 store.append_to_transcript("sess-1", {"role": "user", "content": "new-live"}) assert _contents(mock_db) == ["old0", "old0", "old1", "new-live"] assert not list(flush_dir.glob("*.json")) -def test_live_write_waits_while_held_back_files_still_fail_to_drain(flush_dir, monkeypatch): +def test_live_write_waits_while_held_back_files_still_fail_to_drain(flush_dir, make_store, boot_db): """If the pre-write drain fails again, the live row must stay queued in memory: writing it while the older files are still on disk gives it a lower row id than they will get.""" - import hermes_state_registry - from gateway.run import _recover_pending_flushes - from gateway.session import SessionStore - _write_spool(flush_dir, "pending-bbb.json", "sess-1", {"role": "user", "content": "old0"}, ts=100, seq=0) _write_spool(flush_dir, "pending-aaa.json", "sess-1", @@ -209,17 +247,10 @@ def test_live_write_waits_while_held_back_files_still_fail_to_drain(flush_dir, m mock_db = MagicMock() mock_db.append_message.side_effect = [ RuntimeError("still locked at boot"), RuntimeError("still locked at drain"), 1, 1, 1, 1] - monkeypatch.setattr(hermes_state_registry, "acquire", lambda: mock_db) - monkeypatch.setattr(hermes_state_registry, "release_or_close", lambda _db: None) - store = object.__new__(SessionStore) - store._db = mock_db - store._transcript_retry_lock = threading.Lock() - store._dirty_transcripts = {} - store._transcript_append_failures = {} - store._fts_rebuild_last_attempt_at = time.monotonic() + boot_db(mock_db) + store = make_store(mock_db) - runner = SimpleNamespace(config=SimpleNamespace(multiplex_profiles=False), session_store=store) - assert _recover_pending_flushes(runner) == 0 + assert _boot(store) == 0 store.append_to_transcript("sess-1", {"role": "user", "content": "new-live"}) assert _contents(mock_db) == ["old0", "old0"] assert len(list(flush_dir.glob("*.json"))) == 2 @@ -230,11 +261,65 @@ def test_live_write_waits_while_held_back_files_still_fail_to_drain(flush_dir, m assert not store._dirty_transcripts -def test_repairable_spool_replay_failure_still_reaches_fts_rebuild(flush_dir): +def test_a_poisoned_spool_row_does_not_stop_later_live_rows(flush_dir, tmp_path, make_store, boot_db): + """A spooled field sqlite cannot bind used to fail on every replay. With live writes waiting + behind the spool, the session then stopped persisting for good, across restarts too.""" + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "state.db") + try: + db.create_session("sess-1", source="telegram") + _write_spool(flush_dir, "pending-bbb.json", "sess-1", + {"role": "tool", "content": "result", "tool_call_id": {"bad": 1}}, ts=100, seq=0) + _write_spool(flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "older"}, ts=100, seq=1) + boot_db(db) + store = make_store(db) + + assert _boot(store) == 2 + store.append_to_transcript("sess-1", {"role": "user", "content": "live"}) + + assert [m["content"] for m in db.get_messages("sess-1")] == ["result", "older", "live"] + assert not list(flush_dir.iterdir()) + assert not store._dirty_transcripts + finally: + db.close() + + +def test_a_row_the_database_rejects_is_quarantined_not_retried(flush_dir, make_store, boot_db): + """No retry can write a row the database refuses on its own values, so at boot and in the + live drain it is moved aside instead of holding back the session's later rows.""" + def append_message(**kwargs): + if kwargs["content"].startswith("poison"): + raise sqlite3.InterfaceError("Error binding parameter 3: unsupported type") + return 1 + + mock_db = MagicMock() + mock_db.append_message.side_effect = append_message + _write_spool(flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "poison at boot"}, ts=100, seq=0) + _write_spool(flush_dir, "pending-bbb.json", "sess-1", + {"role": "user", "content": "after"}, ts=100, seq=1) + boot_db(mock_db) + store = make_store(mock_db) + assert _boot(store) == 1 + + _write_spool(flush_dir, "pending-ccc.json", "sess-1", + {"role": "user", "content": "poison live"}, ts=200, seq=2) + _write_spool(flush_dir, "pending-ddd.json", "sess-1", + {"role": "user", "content": "after live"}, ts=200, seq=3) + store.mark_spooled_drop_sessions({"sess-1"}) + store.append_to_transcript("sess-1", {"role": "user", "content": "live"}) + + assert _contents(mock_db) == ["poison at boot", "after", "poison live", "after live", "live"] + assert sorted(p.name for p in flush_dir.iterdir()) == [ + f"pending-aaa.json{QUARANTINE_SUFFIX}", f"pending-ccc.json{QUARANTINE_SUFFIX}"] + assert not store._dirty_transcripts + + +def test_repairable_spool_replay_failure_still_reaches_fts_rebuild(flush_dir, make_store): """A spool replay failure must keep its own type, so the FTS rebuild still runs. After the repair the older spooled row is written before the live one.""" - from gateway.session import SessionStore - _write_spool(flush_dir, "pending-aaa.json", "sess-1", {"role": "user", "content": "old"}, ts=100, seq=0) @@ -253,13 +338,8 @@ def rebuild_fts(self): return 1 db = RepairableDb() - store = object.__new__(SessionStore) - store._db = db - store._transcript_retry_lock = threading.Lock() - store._dirty_transcripts = {} - store._transcript_append_failures = {} - store._fts_rebuild_last_attempt_at = None - store.spooled_drop_sessions().add("sess-1") + store = make_store(db) + store.mark_spooled_drop_sessions({"sess-1"}) store.append_to_transcript("sess-1", {"role": "user", "content": "live"}) @@ -277,7 +357,7 @@ def test_boot_recovery_runs_before_resume_turns_and_queued_inbound(monkeypatch): import gateway.run as gateway_run order = [] - monkeypatch.setattr(gateway_run, "_recover_pending_flushes", + monkeypatch.setattr("gateway.shutdown_flush.recover_gateway_pending", lambda runner: order.append("recover") or 0) monkeypatch.setattr(gateway_run, "_restart_notification_pending", lambda: False) monkeypatch.setattr(gateway_run, "_planned_restart_notification_pending", lambda: False) From 575b4dd70d20a25a1fd34bc4795d09145b36f229 Mon Sep 17 00:00:00 2001 From: unsupportedpastels Date: Tue, 6 Oct 2026 18:13:31 +0000 Subject: [PATCH 17/18] fix(gateway): quarantine only rows sqlite cannot bind is_row_rejection treated every sqlite3.InterfaceError as a bad row, but SessionDB retries InterfaceError('no more rows available') as WAL contention and re-raises it once its patience runs out, so a valid row could be moved out of replay for good. Among sqlite errors only a 'binding parameter' message now counts; OverflowError does too. _bindable_text also passed ints outside sqlite's 64-bit range, which raise OverflowError even for a TEXT column and stalled the session the same way a dict did; they are now stored as text. --- gateway/session_transcript.py | 6 ++- gateway/shutdown_flush.py | 16 ++++--- tests/gateway/test_shutdown_flush_recovery.py | 43 ++++++++++++++++++- 3 files changed, 54 insertions(+), 11 deletions(-) diff --git a/gateway/session_transcript.py b/gateway/session_transcript.py index da8e114e28f47..9d33ba85b1cdc 100644 --- a/gateway/session_transcript.py +++ b/gateway/session_transcript.py @@ -48,8 +48,10 @@ def _spool_dropped(session_id: str, message: Dict[str, Any]): def _bindable_text(value: Any) -> Any: - """*value* when sqlite can bind it as is, else its JSON text (``str`` if that fails too).""" - if value is None or isinstance(value, (str, int, float)): + """*value* when sqlite can bind it as is, else its JSON text (``str`` if that fails too). An int + outside sqlite's signed 64-bit range raises OverflowError even for a TEXT column.""" + if value is None or isinstance(value, (str, float)) or ( + isinstance(value, int) and -2**63 <= value < 2**63): return value try: return json.dumps(value, default=str) diff --git a/gateway/shutdown_flush.py b/gateway/shutdown_flush.py index 10836a2b2bd1b..c30b50b29a6a5 100644 --- a/gateway/shutdown_flush.py +++ b/gateway/shutdown_flush.py @@ -247,13 +247,15 @@ def _spool_sort_key(payload: Dict[str, Any], name: str) -> tuple: def is_row_rejection(exc: BaseException) -> bool: """True when the database refused this row's own values, so no retry can succeed: a value sqlite - cannot bind, or a malformed field that fails before the write. Lock, I/O, corruption and routing - failures stay retryable, and so do constraint failures, which can be session-wide.""" - if isinstance(exc, sqlite3.InterfaceError): - return True - if isinstance(exc, sqlite3.ProgrammingError): - return "binding parameter" in str(exc).lower() - return isinstance(exc, (TypeError, ValueError, AttributeError)) + cannot bind (an unsupported type, or an int outside 64 bits), or a malformed field that fails + before the write. Lock, I/O, corruption and routing failures stay retryable, and so do constraint + failures, which can be session-wide. Only a binding message counts among sqlite errors: + ``InterfaceError("no more rows available")`` is WAL contention that ``SessionDB`` retries and + re-raises unchanged once its patience runs out.""" + if isinstance(exc, sqlite3.Error): + return (isinstance(exc, (sqlite3.InterfaceError, sqlite3.ProgrammingError)) + and "binding parameter" in str(exc).lower()) + return isinstance(exc, (TypeError, ValueError, AttributeError, OverflowError)) def _quarantine_spool_file(path: Path, session_id: str, exc: BaseException) -> None: diff --git a/tests/gateway/test_shutdown_flush_recovery.py b/tests/gateway/test_shutdown_flush_recovery.py index 23d58aefcd3b6..70b1faabe0c79 100644 --- a/tests/gateway/test_shutdown_flush_recovery.py +++ b/tests/gateway/test_shutdown_flush_recovery.py @@ -20,6 +20,7 @@ TRANSCRIPT_CAP_DROP_REASON, _order_flush_files, drain_transcript_spool, + is_row_rejection, recover_gateway_pending, recover_pending_to_db, ) @@ -273,13 +274,16 @@ def test_a_poisoned_spool_row_does_not_stop_later_live_rows(flush_dir, tmp_path, {"role": "tool", "content": "result", "tool_call_id": {"bad": 1}}, ts=100, seq=0) _write_spool(flush_dir, "pending-aaa.json", "sess-1", {"role": "user", "content": "older"}, ts=100, seq=1) + # sqlite cannot bind an int past 64 bits, even into a TEXT column. + _write_spool(flush_dir, "pending-ccc.json", "sess-1", + {"role": "tool", "content": "big id", "tool_call_id": 2**100}, ts=100, seq=2) boot_db(db) store = make_store(db) - assert _boot(store) == 2 + assert _boot(store) == 3 store.append_to_transcript("sess-1", {"role": "user", "content": "live"}) - assert [m["content"] for m in db.get_messages("sess-1")] == ["result", "older", "live"] + assert [m["content"] for m in db.get_messages("sess-1")] == ["result", "older", "big id", "live"] assert not list(flush_dir.iterdir()) assert not store._dirty_transcripts finally: @@ -317,6 +321,41 @@ def append_message(**kwargs): assert not store._dirty_transcripts +def test_only_errors_about_the_rows_own_values_are_rejections(): + """A rejection is quarantined for good, so it must be an error no retry can clear.""" + assert is_row_rejection(sqlite3.ProgrammingError("Error binding parameter 4: type 'dict' is not supported")) + assert is_row_rejection(sqlite3.InterfaceError("Error binding parameter 3 - probably unsupported type")) + assert is_row_rejection(OverflowError("Python int too large to convert to SQLite INTEGER")) + assert not is_row_rejection(sqlite3.InterfaceError("no more rows available")) + assert not is_row_rejection(sqlite3.OperationalError("database is locked")) + assert not is_row_rejection(sqlite3.IntegrityError("FOREIGN KEY constraint failed")) + assert not is_row_rejection(sqlite3.DatabaseError("database disk image is malformed")) + + +def test_transient_no_more_rows_error_keeps_the_row_for_retry(flush_dir, make_store, boot_db): + """SessionDB retries 'no more rows available' as WAL contention and re-raises it unchanged once + its patience runs out. The row itself is valid, so at boot and in the live drain it must stay on + disk and be written once the contention clears, not be quarantined.""" + transient = sqlite3.InterfaceError("no more rows available") + path = _write_spool(flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "old"}, ts=100, seq=0) + mock_db = MagicMock() + mock_db.append_message.side_effect = [transient, transient, 1, 1, 1] + boot_db(mock_db) + store = make_store(mock_db) + + assert _boot(store) == 0 + assert path.exists() + store.append_to_transcript("sess-1", {"role": "user", "content": "live"}) + assert _contents(mock_db) == ["old", "old"] + assert path.exists() + + store.append_to_transcript("sess-1", {"role": "user", "content": "live-2"}) + assert _contents(mock_db) == ["old", "old", "old", "live", "live-2"] + assert not list(flush_dir.iterdir()) + assert not store._dirty_transcripts + + def test_repairable_spool_replay_failure_still_reaches_fts_rebuild(flush_dir, make_store): """A spool replay failure must keep its own type, so the FTS rebuild still runs. After the repair the older spooled row is written before the live one.""" From 0b9659c7887a4589ab33412c7a076ccd0324c42a Mon Sep 17 00:00:00 2001 From: unsupportedpastels Date: Tue, 6 Oct 2026 18:33:14 +0000 Subject: [PATCH 18/18] fix(gateway): never replay a written spool row whose file cannot be deleted When a spooled row was written but its file could not be deleted, the OSError aborted the drain: the files after it were left for later, so the live row landed ahead of them, and because the file stayed, the next live write replayed and wrote the same row again, once per write. The boot pass had the same repeat. _remove_replayed now logs the failure, remembers the path so neither replay pass in this process touches it again, and lets replay go on with the later files. A restart still writes that row once more, since nothing on disk marks it; the log says to delete the file. Live rows are not held back for this: an unwritable spool directory stays fail-open. Review finding by Greptile. --- gateway/shutdown_flush.py | 23 +++++++- tests/gateway/test_shutdown_flush_recovery.py | 53 +++++++++++++++++++ 2 files changed, 74 insertions(+), 2 deletions(-) diff --git a/gateway/shutdown_flush.py b/gateway/shutdown_flush.py index c30b50b29a6a5..e395fc5bd9979 100644 --- a/gateway/shutdown_flush.py +++ b/gateway/shutdown_flush.py @@ -35,6 +35,9 @@ QUARANTINE_SUFFIX = ".bad" # Monotonic tiebreaker so same-second spool files replay in drop order. _TRANSCRIPT_SPOOL_SEQ = itertools.count() +# Spool files whose row was written but which could not be deleted. Replaying one again would write +# its row twice, so both replay passes skip it for the rest of this process. +_REPLAYED_UNREMOVABLE: set[Path] = set() def _get_flush_dir(): @@ -155,6 +158,8 @@ def drain_transcript_spool(session_id: str, replay, *, db_known_failing: bool = return 0, 0 entries = [] for path in candidates: + if path in _REPLAYED_UNREMOVABLE: + continue try: payload = json.loads(path.read_text(encoding="utf-8-sig")) except Exception: @@ -184,7 +189,7 @@ def drain_transcript_spool(session_id: str, replay, *, db_known_failing: bool = "keeping spool file for retry: %s", path, session_id, exc) remaining = len(ordered) - idx break - path.unlink(missing_ok=True) + _remove_replayed(path, session_id) replayed += 1 if replayed: logger.info("Replayed %d spooled transcript message(s) for %s after DB recovery", replayed, @@ -258,6 +263,18 @@ def is_row_rejection(exc: BaseException) -> bool: return isinstance(exc, (TypeError, ValueError, AttributeError, OverflowError)) +def _remove_replayed(path: Path, session_id: str) -> None: + """Delete a spool file whose row was just written. If it cannot be deleted, this process never + replays it again, and replay goes on with the files after it, which are still older than any + live row. Nothing on disk marks it, so a restart writes the row a second time.""" + try: + path.unlink(missing_ok=True) + except OSError as exc: + _REPLAYED_UNREMOVABLE.add(path) + logger.error("Spooled transcript row for %s was written, but %s could not be deleted (%s); " + "delete it by hand, or the next start writes that row again", session_id, path, exc) + + def _quarantine_spool_file(path: Path, session_id: str, exc: BaseException) -> None: """Move a spool file whose row the database rejected out of every replay, so it cannot hold its session's later rows back on every write and every restart. It stays on disk for manual recovery.""" @@ -327,6 +344,8 @@ def recover_pending_spool(session_db=None, *, session_resolver=None) -> tuple[in blocked_sessions: Dict[str, int] = {} try: for path in flush_files: + if path in _REPLAYED_UNREMOVABLE: + continue # One unparseable payload or rejected append must only skip THIS file: the file is # never unlinked, so aborting the pass would re-poison every later boot. try: @@ -340,7 +359,7 @@ def recover_pending_spool(session_db=None, *, session_resolver=None) -> tuple[in session_resolver=session_resolver, blocked_sessions=blocked_sessions): recovered += 1 - path.unlink(missing_ok=True) + _remove_replayed(path, payload.get("session_key", "")) # health: allow BLE001 -- per-file boundary moved unchanged from recover_pending_to_db; a traceback per locked-DB file would only add noise except Exception as exc: logger.warning("Failed to recover pending message from %s: %s", path, exc) diff --git a/tests/gateway/test_shutdown_flush_recovery.py b/tests/gateway/test_shutdown_flush_recovery.py index 70b1faabe0c79..08f906f5af6d1 100644 --- a/tests/gateway/test_shutdown_flush_recovery.py +++ b/tests/gateway/test_shutdown_flush_recovery.py @@ -34,6 +34,7 @@ def flush_dir(tmp_path, monkeypatch): monkeypatch.setattr( "gateway.shutdown_flush._get_flush_dir", lambda: directory ) + monkeypatch.setattr("gateway.shutdown_flush._REPLAYED_UNREMOVABLE", set()) return directory @@ -105,6 +106,18 @@ def _contents(mock_db) -> list: return [c.kwargs["content"] for c in mock_db.append_message.call_args_list] +def _fail_unlink_for(monkeypatch, name: str) -> None: + """Make deleting the spool file *name* fail, as on a directory that stopped being writable.""" + real_unlink = Path.unlink + + def unlink(self, missing_ok=False): + if self.name == name: + raise PermissionError(13, "Permission denied", str(self)) + return real_unlink(self, missing_ok=missing_ok) + + monkeypatch.setattr(Path, "unlink", unlink) + + def test_replays_in_drop_order_not_file_name_order(flush_dir): """SessionDB restores by AUTOINCREMENT id, so append order IS the order the user sees after recovery. Production names are ``pending-.json``; these names sort opposite @@ -356,6 +369,46 @@ def test_transient_no_more_rows_error_keeps_the_row_for_retry(flush_dir, make_st assert not store._dirty_transcripts +def test_a_replayed_file_that_cannot_be_deleted_is_not_replayed_again(flush_dir, make_store, monkeypatch): + """Its row is already written, so replaying it before the next live write duplicates that row. + The files after it are still older than the live row, so they must land first.""" + _write_spool(flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "old0"}, ts=100, seq=0) + second = _write_spool(flush_dir, "pending-bbb.json", "sess-1", + {"role": "user", "content": "old1"}, ts=100, seq=1) + _fail_unlink_for(monkeypatch, "pending-aaa.json") + mock_db = MagicMock() + store = make_store(mock_db) + store.mark_spooled_drop_sessions({"sess-1"}) + + store.append_to_transcript("sess-1", {"role": "user", "content": "live"}) + store.mark_spooled_drop_sessions({"sess-1"}) + store.append_to_transcript("sess-1", {"role": "user", "content": "live-2"}) + + assert _contents(mock_db) == ["old0", "old1", "live", "live-2"] + assert not second.exists() + assert not store._dirty_transcripts + + +def test_boot_recovery_counts_and_never_repeats_a_file_it_could_not_delete( + flush_dir, make_store, boot_db, monkeypatch): + """Boot recovery wrote the row, so it is recovered; the live drain must not write it again.""" + _write_spool(flush_dir, "pending-aaa.json", "sess-1", + {"role": "user", "content": "old0"}, ts=100, seq=0) + _write_spool(flush_dir, "pending-bbb.json", "sess-1", + {"role": "user", "content": "old1"}, ts=100, seq=1) + _fail_unlink_for(monkeypatch, "pending-aaa.json") + mock_db = MagicMock() + boot_db(mock_db) + store = make_store(mock_db) + + assert _boot(store) == 2 + store.mark_spooled_drop_sessions({"sess-1"}) + store.append_to_transcript("sess-1", {"role": "user", "content": "live"}) + + assert _contents(mock_db) == ["old0", "old1", "live"] + + def test_repairable_spool_replay_failure_still_reaches_fts_rebuild(flush_dir, make_store): """A spool replay failure must keep its own type, so the FTS rebuild still runs. After the repair the older spooled row is written before the live one."""