fix(gateway): recover shutdown-flushed messages

[Gabrielnkl]

Summary

Follow-up to #75536: make shutdown-flush recovery reliable on the current gateway implementation.

The original session-resolution issue identified in #75536 is still present on current main, but the gateway has since changed and FIFO overflow events are now also persisted through the shutdown-flush path.

This PR rebases the fix onto the current recovery path and adds two correctness fixes discovered while testing the current implementation:

• resolve real MessageEvent payloads through session_key → session_id
• replay flushed events according to their persisted sequence instead of filename order
• isolate malformed/failed spool entries so one bad file does not strand later valid messages

Root cause

A real gateway MessageEvent does not contain session_id.

During shutdown serialization, the event therefore produces a payload shaped like:

{
  "session_key": "...",
  "data": {
    "text": "..."
  }
}

The recovery path previously required data["session_id"].

As a result:

gateway receives message
        ↓
message is durably flushed
        ↓
gateway restarts
        ↓
recovery finds no session_id
        ↓
message is skipped
        ↓
spool file remains
        ↓
message is never delivered

The existing persisted session_key is sufficient to recover the session through SessionStore.peek_session_id().

Current-main overflow interaction

Since #100972, FIFO overflow events are also written to the shutdown spool.

Those events use the same recovery format, so they are subject to the same session-resolution problem.

This PR adds regression coverage for both normal pending-message flushes and overflow flushes using the real MessageEvent shape.

Ordered replay

Shutdown-flush payloads can contain a persisted seq representing arrival order.

The recovery path previously iterated spool files by filename. UUID-based filenames do not guarantee arrival order.

For example:

pending-aaa.json → seq=1 → "second"
pending-zzz.json → seq=0 → "first"

Recovery must produce:

first
second

rather than filename order.

The recovery path now sorts by:

(ts, seq, filename)

Legacy payloads without these fields retain deterministic fallback ordering.

Failure isolation

Recovery should not allow one bad spool entry to prevent unrelated valid messages from being recovered.

The new behavior is:

• malformed JSON → warn and continue
• malformed payload → warn and continue
• unresolved session → keep that file and continue
• append failure → keep that file and continue
• valid later files → still recovered

A failed recovery never deletes its spool file.

Tests

Added regression coverage for:

• real MessageEvent payloads without session_id
• session-key → session-id recovery
• overflow shutdown-flush recovery
• unresolved session keys preserving their spool files
• adversarial sequence/filename ordering
• malformed spool files not preventing subsequent valid recovery

Focused validation:

tests/gateway/test_shutdown_flush.py

Current branch:

13+ tests passing

The full gateway suite was also exercised; unrelated pre-existing failures remain outside the shutdown-flush changes.

Relationship to #75536

#75536 identified and implemented the original session_key → session_id recovery fix.

This PR carries that fix forward onto the current gateway architecture and adds:

1. current-main startup wiring
2. overflow-path regression coverage
3. sequence-based replay ordering
4. per-file recovery failure isolation

Credit to #75536 for the original diagnosis and core session-resolution approach.

Related: #72680, #99882, #100972.