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

@@ -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,

tests/gateway/test_shutdown_spool_limits.py

@@ -32,3 +32,23 @@ def test_recovery_payload_memory_does_not_scale_with_backlog(
     # 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"))