"""Tests for gateway/shutdown_flush.py — pending message durability (#72680).""" import json import time from pathlib import Path from unittest.mock import MagicMock import pytest from gateway.shutdown_flush import ( flush_overflow_to_file, flush_pending_to_file, recover_pending_to_db, ) def _make_flush_dir(tmp_path: Path) -> Path: """Create a temp flush dir and monkeypatch _get_flush_dir to use it.""" flush_dir = tmp_path / "pending_messages" flush_dir.mkdir(parents=True, exist_ok=True) return flush_dir def test_flush_writes_string_pending_to_file(tmp_path, monkeypatch): flush_dir = _make_flush_dir(tmp_path) monkeypatch.setattr( "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir ) pending = {"agent:main:telegram:supergroup:123": "hello world"} count = flush_pending_to_file(pending, reason="shutdown") assert count == 1 files = list(flush_dir.glob("*.json")) assert len(files) == 1 payload = json.loads(files[0].read_text(encoding="utf-8")) assert payload["session_key"] == "agent:main:telegram:supergroup:123" assert payload["reason"] == "shutdown" assert payload["data"]["text"] == "hello world" assert ":" not in files[0].name assert "telegram" not in files[0].name def test_flush_writes_message_event_to_file(tmp_path, monkeypatch): flush_dir = _make_flush_dir(tmp_path) monkeypatch.setattr( "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir ) event = MagicMock() event.text = "user message" event.session_id = "20260728_120000_abc" event.platform = "telegram" event.sender_id = "456" event.sender_name = "Alice" event.reply_to = None event.media = None event.raw_event = None count = flush_pending_to_file({"session_key_1": event}, reason="adapter_shutdown") assert count == 1 files = list(flush_dir.glob("*.json")) assert len(files) == 1 payload = json.loads(files[0].read_text(encoding="utf-8")) assert payload["data"]["text"] == "user message" assert payload["data"]["session_id"] == "20260728_120000_abc" def test_recover_inserts_via_append_message_and_deletes_file(tmp_path, monkeypatch): flush_dir = _make_flush_dir(tmp_path) monkeypatch.setattr( "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir ) ts = int(time.time()) # Write a flush file with session_id payload = { "session_key": "agent:main:telegram:supergroup:123", "reason": "shutdown", "ts": ts, "data": { "text": "lost message", "session_id": "20260728_120000_abc", }, } flush_file = flush_dir / "test_session_123.json" flush_file.write_text(json.dumps(payload), encoding="utf-8") mock_db = MagicMock() count = recover_pending_to_db(mock_db) assert count == 1 mock_db.append_message.assert_called_once_with( session_id="20260728_120000_abc", role="user", content="lost message", timestamp=ts, ) assert not flush_file.exists() def test_recover_payload_without_session_id_uses_resolver_and_deletes_file( tmp_path, monkeypatch ): """A str-pending payload carries session_key but no session_id; a resolver maps key→id so the message replays and the file is removed.""" flush_dir = _make_flush_dir(tmp_path) monkeypatch.setattr("gateway.shutdown_flush._get_flush_dir", lambda: flush_dir) ts = int(time.time()) payload = { "session_key": "agent:main:whatsapp:dm:15551234567", "reason": "shutdown", "ts": ts, "data": {"text": "lost message"}, } flush_file = flush_dir / "no_sid.json" flush_file.write_text(json.dumps(payload), encoding="utf-8") mock_db = MagicMock() routed_db = MagicMock() # the store owning the key, as the resolver reports it resolver = MagicMock(return_value=("sid-resolved", routed_db)) count = recover_pending_to_db(mock_db, session_resolver=resolver) assert count == 1 resolver.assert_called_once_with("agent:main:whatsapp:dm:15551234567", not_after=ts) routed_db.append_message.assert_called_once_with( session_id="sid-resolved", role="user", content="lost message", timestamp=ts ) # The resolver's db is authoritative: the owned default store never sees a resolved payload. mock_db.append_message.assert_not_called() assert not flush_file.exists() # A resolver that names an id but no store (fail-closed profile home) preserves the file # instead of appending to the owned root store. flush_file.write_text(json.dumps(payload), encoding="utf-8") resolver = MagicMock(return_value=("sid-resolved", None)) assert recover_pending_to_db(mock_db, session_resolver=resolver) == 0 mock_db.append_message.assert_not_called() assert flush_file.exists() def test_recover_closes_owned_db_when_unexpected_exception_escapes( tmp_path, monkeypatch ): """Owned SessionDB must close even when recovery is interrupted.""" flush_dir = _make_flush_dir(tmp_path) monkeypatch.setattr( "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir ) (flush_dir / "pending.json").write_text( json.dumps( { "session_key": "agent:main:telegram:123", "data": {"text": "message", "session_id": "sid"}, } ), encoding="utf-8", ) class InterruptingDB: released = False def append_message(self, **_kwargs): raise KeyboardInterrupt db = InterruptingDB() monkeypatch.setattr("hermes_state_registry.acquire", lambda: db) monkeypatch.setattr( "hermes_state_registry.release_or_close", lambda _: setattr(db, "released", True) ) with pytest.raises(KeyboardInterrupt): recover_pending_to_db() assert db.released is True def _write_flush_file(flush_dir: Path, name: str, session_id: str, text: str) -> Path: """Write one well-formed pending-message flush file.""" path = flush_dir / name path.write_text( json.dumps( { "session_key": "agent:main:telegram:supergroup:123", "reason": "shutdown", "ts": 1700000000, "data": {"text": text, "session_id": session_id}, } ), encoding="utf-8", ) return path def test_recover_skips_failing_payload_and_continues(tmp_path, monkeypatch): """One unrecoverable file must not abort recovery of the others. Recovery walks ``sorted(glob("*.json"))``, so a file that raises an ordinary exception used to propagate out of the whole pass. Because that file is never unlinked it would then re-poison every subsequent boot, stranding a different subset of messages each time. """ flush_dir = _make_flush_dir(tmp_path) monkeypatch.setattr( "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir ) bad = _write_flush_file(flush_dir, "pending-a.json", "sid-bad", "first") good = _write_flush_file(flush_dir, "pending-b.json", "sid-good", "second") class PartiallyFailingDB: def __init__(self): self.appended = [] def append_message(self, **kwargs): if kwargs["session_id"] == "sid-bad": raise RuntimeError("session is closed") self.appended.append(kwargs["session_id"]) db = PartiallyFailingDB() assert recover_pending_to_db(db) == 1 assert db.appended == ["sid-good"] # The good file is consumed; the bad one is preserved for a retry. assert not good.exists() assert bad.exists() def test_get_flush_dir_uses_get_hermes_home(tmp_path, monkeypatch): """Flush dir must use get_hermes_home(), not hardcoded Path.home().""" import gateway.shutdown_flush as mod captured = {} def fake_get_hermes_home(): captured["called"] = True return tmp_path monkeypatch.setattr( "hermes_constants.get_hermes_home", fake_get_hermes_home ) result = mod._get_flush_dir() assert captured.get("called") is True assert result == tmp_path / "pending_messages" # ── FIFO overflow tail durability (#99882) ───────────────────────────── def _overflow_event(text: str, session_id: str = "20260901_120000_fifo"): event = MagicMock() event.text = text event.session_id = session_id event.platform = "telegram" event.sender_id = "1572286605" event.sender_name = "tester" event.reply_to = None event.media = None event.raw_event = None return event def test_flush_overflow_writes_one_payload_per_event_in_arrival_order(tmp_path, monkeypatch): """The FIFO tail (queued_events) must survive shutdown like the slot does. Each overflow entry is its own recover_pending_to_db-compatible payload, with ``seq`` recording arrival order inside the session. """ flush_dir = _make_flush_dir(tmp_path) monkeypatch.setattr("gateway.shutdown_flush._get_flush_dir", lambda: flush_dir) count = flush_overflow_to_file( { "agent:main:telegram:dm:1": [ _overflow_event("follow-up B"), _overflow_event("follow-up C"), ], "agent:main:telegram:dm:2": [], "": [_overflow_event("keyless — skipped")], }, reason="shutdown", ) assert count == 2 payloads = sorted( (json.loads(f.read_text(encoding="utf-8")) for f in flush_dir.glob("*.json")), key=lambda p: p["seq"], ) assert [p["data"]["text"] for p in payloads] == ["follow-up B", "follow-up C"] assert {p["session_key"] for p in payloads} == {"agent:main:telegram:dm:1"} assert all(p["reason"] == "shutdown" for p in payloads) def test_flushed_overflow_is_replayed_by_recover_pending_to_db(tmp_path, monkeypatch): """Round-trip: overflow payloads use the slot-flush shape, so the existing startup recovery inserts them as user rows without any new reader.""" flush_dir = _make_flush_dir(tmp_path) monkeypatch.setattr("gateway.shutdown_flush._get_flush_dir", lambda: flush_dir) flush_overflow_to_file({"agent:main:telegram:dm:1": [_overflow_event("orphan-1")]}) db = MagicMock() recovered = recover_pending_to_db(session_db=db) assert recovered == 1 db.append_message.assert_called_once() kwargs = db.append_message.call_args.kwargs assert kwargs["session_id"] == "20260901_120000_fifo" assert kwargs["role"] == "user" assert kwargs["content"] == "orphan-1" assert list(flush_dir.glob("*.json")) == [] def test_flush_overflow_noop_on_empty(): assert flush_overflow_to_file({}) == 0 assert flush_overflow_to_file({"k": []}) == 0 def test_drain_transcript_spool_skips_parseable_non_dict_payload(tmp_path, monkeypatch): """A scalar/list JSON spool file must not abort the drain; the healthy payload still replays.""" from gateway.shutdown_flush import drain_transcript_spool, spool_dropped_transcript_message flush_dir = _make_flush_dir(tmp_path) monkeypatch.setattr("gateway.shutdown_flush._get_flush_dir", lambda: flush_dir) (flush_dir / "pending-00-scalar.json").write_text("42", encoding="utf-8") (flush_dir / "pending-01-list.json").write_text("[1, 2]", encoding="utf-8") spool_dropped_transcript_message("sess-1", {"role": "user", "content": "hi"}) replayed = [] assert drain_transcript_spool("sess-1", replayed.append) == (1, 0) assert replayed == [{"role": "user", "content": "hi"}]