fix(gateway): resolve session_key in shutdown-flush recovery

[spfcraze]

What does this PR do?

Fixes the #72680 pending-message recovery never recovering anything: every real shutdown-flush file is silently skipped at startup.

The mechanism was built so a message arriving while the gateway is draining ("⏳ queued for the next turn after it comes back") is written to <hermes_home>/pending_messages/ and re-ingested on the next boot. But the data path is broken end-to-end:

1. _serialise_value captures only text from adapter _pending_messages values — they are MessageEvent objects, which have no session_id attribute (that lives on event.source, which isn't serialised).
2. recover_pending_to_db therefore always lands in the "no session_id in flush file" branch: warning logged, message skipped, file preserved — on every boot, forever.

The existing tests masked it by hand-writing session_id into flush payloads that real MessageEvents never produce.

The fix adds an optional session_resolver parameter to recover_pending_to_db and wires it at the startup call site to SessionStore.peek_session_id (the existing lock-held session_key → session_id accessor). Files with genuinely unresolvable keys keep the current preserve-and-warn behavior.

Related Issue

No GitHub issue — discovered via code review and reproduced live (see below). Happy to file one first if preferred.

Changes Made

• gateway/shutdown_flush.py: recover_pending_to_db gains session_resolver; when session_id is absent from the payload it tries session_resolver(session_key) before falling back to the preserve-and-warn skip. Docstring documents why real flush files lack session_id.
• gateway/run.py: the startup recovery call passes session_resolver=runner.session_store.peek_session_id.
• tests/gateway/test_shutdown_flush.py: three regression tests — a text-only (real-MessageEvent-shaped) payload is recovered via the resolver and the file deleted; the same payload without a resolver is skipped with the file preserved; and an integration test through the real startup boundary — a mapping persisted by one SessionStore, reloaded by a fresh one, recovered via peek_session_id (the exact run.py wiring, no mocks).

How to Test

Reproduction (against pre-fix code, real MessageEvent through the real serialiser):

payload = _serialise_value(message_event)          # -> {"text": "are you there?"}  (no session_id)
# flush file written with that payload, then on next boot:
recover_pending_to_db(db)                          # pre-fix:  0 recovered, file preserved (message silently lost)
recover_pending_to_db(db, session_resolver=store.peek_session_id)
                                                   # post-fix: 1 recovered, correct row, file cleaned up

Focused validation completed:

1. bash scripts/run_tests.sh tests/gateway/test_shutdown_flush.py — 8/8 pass.
2. Sabotage check: reverting the fix makes both the resolver test and the integration test fail (6 pass); restoring returns to 8/8.
3. Related suites: test_gateway_shutdown.py test_13121_shutdown_inflight_transcript_flush.py test_pending_drain_no_recursion.py test_bounded_adapter_teardown.py + the flush file — 22/22 pass.
4. Full repo-wide suite, bash scripts/run_tests.sh (branch): 22,920 pass / 98 fail. Baseline on clean main (this branch's parent), same machine: 22,917 pass / 99 fail — zero branch-only failures (environment-dependent lanes only). Nothing in the shutdown/flush surface fails.
5. uvx --from ruff==0.15.10 ruff check gateway/shutdown_flush.py gateway/run.py tests/gateway/test_shutdown_flush.py — clean.
6. git diff --check — clean.
7. Adversarial cases executed: resolver returning None (unknown key) falls back to preserve-and-warn; payloads that DO carry session_id still take the direct path; resolver raising is contained (falls back, no crash).

The full repo-wide suite was run locally (item 4) with zero branch-only failures vs clean main; GitHub CI remains the final confirmation environment.

Logs

Sabotage verification output:

# fix stashed (pre-fix behavior):
=== Summary: 1 files, 6 tests passed, 2 failed ===   (resolver + integration tests)
# fix restored:
=== Summary: 1 files, 8 tests passed, 0 failed ===

[teknium1]

Thanks for tracing this through the real adapter-pending representation. The premise is confirmed on current main: gateway/shutdown_flush.py:228-242 only accepts data.session_id, while gateway/platforms/base.py:2051-2063 defines MessageEvent without that field. The proposed fallback at gateway/shutdown_flush.py:239-243 and startup wiring at gateway/run.py:25449-25451 fit the existing lock-held SessionStore.peek_session_id() accessor (gateway/session.py:2905-2919).

Problems

• tests/gateway/test_shutdown_flush.py:154-158 mocks the resolver and DB, so it does not exercise persisted routing reload through the real SessionStore startup boundary.

Suggested changes

• Add a temp-HERMES_HOME integration test that persists a session-key mapping, constructs a fresh SessionStore, and recovers a text-only payload via peek_session_id.

Automated hermes-sweeper review.

[teknium1]

This proves the callback fallback, but not the startup integration. Please add a temp-HERMES_HOME test using a fresh real SessionStore whose persisted routing entry is reloaded through peek_session_id, then verify recovery appends to that session ID.

[spfcraze]

Addressed in the latest push (b1db79117):

Integration test added — test_recover_via_real_session_store_routing_reload exercises the real startup boundary with no mocks: boot 1 persists a session-key mapping via SessionStore.get_or_create_session; boot 2 constructs a fresh SessionStore and peek_session_id reloads the mapping from disk; a text-only payload (the real MessageEvent shape) is then recovered through recover_pending_to_db(db, session_resolver=store2.peek_session_id) — the exact wiring installed at run.py:25449. Asserts the transcript row lands in the target session and the flush file is deleted.

Sabotage check: reverting the fix fails both the resolver unit test and the integration test (6 pass / 2 fail); restored, 8/8 in the file. ruff + diff-check clean.