fix(gateway): the flush-time fence compares whole seconds

The flush payload's ``ts`` is ``int(time.time())`` while ``sessions.started_at`` is a REAL, so
``started_at > not_after`` rejected a row minted later in the same second as the flush — the
common "message arrives, session minted, SIGTERM" shape lost recoverability (file preserved, not
misrouted). Compare ``floor(started_at)`` against the whole-second ``ts`` so same-second rows
are adopted and only rows from a later second are refused.

Spotted by the round-2 gate on the salvage stack.

gateway/session_recovery.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 
 import logging
 import json
+import math
 import threading
 from dataclasses import replace
 from datetime import datetime
@@ -221,7 +222,9 @@ class SessionRecoveryMixin:
         if not isinstance(row, dict) or not row.get("id"):
             return None
         started_at = row.get("started_at")
-        if not_after is not None and started_at is not None and float(started_at) > float(not_after):
+        # ``ts`` is ``int(time.time())`` while ``started_at`` is a REAL, so compare whole seconds:
+        # a row minted in the same second as the flush is still a valid origin.
+        if not_after is not None and started_at is not None and math.floor(float(started_at)) > int(not_after):
             return None
         return str(row["id"]), db
 

tests/gateway/test_session_recovery.py

@@ -58,3 +58,7 @@ def test_resolve_db_fallback_rejects_row_started_after_flush(tmp_path):
     store = _store(tmp_path, db)
     assert store.resolve_session_id_for_key(key, not_after=1700000000) is None
     assert store.resolve_session_id_for_key(key, not_after=1700000100) == ("late-row", db)
+    # The flush ``ts`` is a whole second while ``started_at`` is a REAL: a row minted later in the
+    # SAME second as the flush is still a valid origin, not a post-flush row.
+    same_second = _FakeGatewayDB({key: {"id": "same-sec", "started_at": 1700000000.818}})
+    assert _store(tmp_path / "b", same_second).resolve_session_id_for_key(key, not_after=1700000000) == ("same-sec", same_second)