fix(gateway): recover resume_pending sessions instead of sending a blank turn

A session interrupted by a gateway restart is flagged resume_pending and
auto-continued on startup via _schedule_resume_pending_sessions(), which
dispatches an empty-text internal MessageEvent. The recovery system note
that should fill that empty turn is gated, in _run_agent(), on
_interruption_is_fresh — the age of the LAST PERSISTED TRANSCRIPT ROW.

For an active thread returned to after >1h of silence, that transcript
clock is stale even though the interruption (last_resume_marked_at) is
seconds old. The gate evaluates False, the note is not prepended, and the
model receives a genuinely blank user turn — replying with confused
'that message came through blank' noise.

Fix (two parts, both default-on, behavior unchanged for healthy turns):

1. resume_pending freshness now also considers last_resume_marked_at (the
   restart watchdog's own stamp). The branch fires when EITHER the
   transcript clock OR the resume mark is fresh, so the startup scheduler's
   freshness decision and the per-turn injection agree.

2. Empty-turn safety net: if the user turn is still blank after all
   injections AND the session is resume_pending, backfill a recovery note
   so a blank turn can never reach the model. Scoped to resume_pending so
   ordinary empty turns (e.g. uncaptioned image) are untouched.

Adds 3 regression tests; the two core ones fail on the pre-fix logic.

gateway/run.py

@@ -17334,10 +17334,30 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     _resume_entry = self.session_store._entries.get(session_key)
                 except Exception:
                     _resume_entry = None
+
+            # resume_pending freshness uses a SECOND signal in addition to the
+            # transcript clock above.  The restart watchdog stamps the session
+            # with ``last_resume_marked_at`` at interrupt time — that is the
+            # correct "when were we interrupted" signal.  The transcript clock
+            # (_interruption_is_fresh) can be far older: an active thread you
+            # return to may have its last persisted row hours back, even though
+            # the interruption itself just happened.  Gating resume_pending on
+            # the transcript clock alone makes the recovery note silently drop,
+            # and because the startup auto-resume turn carries empty text
+            # (_schedule_resume_pending_sessions), the model then receives a
+            # blank user message and replies with confused "the message came
+            # through blank" noise.  Treat the marker as fresh when
+            # EITHER signal is fresh so the two freshness checks agree.
+            _resume_mark_is_fresh = False
+            if _resume_entry is not None and getattr(_resume_entry, "resume_pending", False):
+                _resume_mark_is_fresh = _is_fresh_gateway_interruption(
+                    getattr(_resume_entry, "last_resume_marked_at", None),
+                    window_secs=_freshness_window,
+                )
             _is_resume_pending = bool(
                 _resume_entry is not None
                 and getattr(_resume_entry, "resume_pending", False)
-                and _interruption_is_fresh
+                and (_interruption_is_fresh or _resume_mark_is_fresh)
             )
             _has_fresh_tool_tail = bool(
                 agent_history
@@ -17399,6 +17419,29 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 if _srn:
                     message = _srn + "\n\n" + message
 
+            # Safety net: a startup auto-resume event carries empty
+            # text and relies on the resume_pending branch above to supply the
+            # recovery note.  If that branch did not fire for any reason (e.g.
+            # both freshness signals disagreed, or the marker was cleared
+            # between scheduling and dispatch) we must NOT hand the model a
+            # blank user turn — it responds with confused "the message came
+            # through blank" noise.  Restricted to resume_pending sessions so
+            # legitimately empty user turns (e.g. an image with no caption,
+            # wrapped as native content below) are untouched.
+            if (
+                isinstance(message, str)
+                and not message.strip()
+                and _resume_entry is not None
+                and getattr(_resume_entry, "resume_pending", False)
+            ):
+                message = (
+                    "[System note: The previous turn in this session was "
+                    "interrupted by a gateway restart. Review the conversation "
+                    "history above, finish any unfinished work, and summarize "
+                    "what was accomplished, then wait for the user's next "
+                    "message.]"
+                )
+
             _approval_session_key = session_key or ""
             _approval_session_token = set_current_session_key(_approval_session_key)
             register_gateway_notify(_approval_session_key, _approval_notify_sync)

tests/gateway/test_restart_resume_pending.py

@@ -133,10 +133,16 @@ def _simulate_note_injection(
     )
 
     message = user_message
+    resume_mark_is_fresh = False
+    if resume_entry is not None and getattr(resume_entry, "resume_pending", False):
+        resume_mark_is_fresh = _is_fresh_gateway_interruption(
+            getattr(resume_entry, "last_resume_marked_at", None),
+            window_secs=window,
+        )
     is_resume_pending = bool(
         resume_entry is not None
         and getattr(resume_entry, "resume_pending", False)
-        and interruption_is_fresh
+        and (interruption_is_fresh or resume_mark_is_fresh)
     )
     has_fresh_tool_tail = bool(
         agent_history
@@ -180,6 +186,22 @@ def _simulate_note_injection(
             "below FIRST. Do NOT re-execute old tool calls from the history.]\n\n"
             + message
         )
+
+    # Empty-turn safety net: mirrors gateway/run.py — a blank
+    # auto-resume turn on a resume_pending session must never reach the model.
+    if (
+        isinstance(message, str)
+        and not message.strip()
+        and resume_entry is not None
+        and getattr(resume_entry, "resume_pending", False)
+    ):
+        message = (
+            "[System note: The previous turn in this session was "
+            "interrupted by a gateway restart. Review the conversation "
+            "history above, finish any unfinished work, and summarize "
+            "what was accomplished, then wait for the user's next "
+            "message.]"
+        )
     return message
 
 
@@ -533,6 +555,71 @@ class TestResumePendingSystemNote:
         )
         assert result == "start a new task"
 
+    def test_fresh_resume_mark_fires_despite_stale_transcript(self):
+        """Regression: the recovery note must fire when the restart
+        watchdog just stamped the session, even if the last persisted
+        transcript row is far older than the freshness window.
+
+        This is the exact gap that produced the blank-turn symptom: an
+        active thread returned to after >1h of silence has a stale
+        transcript clock, but the interruption itself (last_resume_marked_at)
+        is seconds old. The two freshness signals must agree.
+        """
+        entry = self._pending_entry()
+        entry.last_resume_marked_at = datetime.now()  # interrupted just now
+
+        history = [
+            {"role": "assistant", "content": "older context",
+             "timestamp": time.time() - 3600},  # transcript clock stale
+        ]
+        result = _simulate_note_injection(
+            history=history,
+            user_message="continue",
+            resume_entry=entry,
+            window_secs=1800,
+        )
+        assert "[System note:" in result
+        assert "gateway restart" in result
+
+    def test_empty_resume_turn_never_reaches_model_blank(self):
+        """Regression: a blank auto-resume turn on a resume_pending
+        session must be backfilled with a recovery note, never sent empty.
+
+        _schedule_resume_pending_sessions dispatches an empty-text internal
+        event. If the resume_pending branch did not fire, the safety net
+        must still produce non-blank text so the model does not reply with
+        confused 'the message came through blank' noise.
+        """
+        entry = self._pending_entry()
+        # Force the resume_pending branch to miss by making BOTH signals stale,
+        # so only the empty-turn safety net can save us.
+        entry.last_resume_marked_at = datetime.now() - timedelta(hours=2)
+        history = [
+            {"role": "assistant", "content": "old", "timestamp": time.time() - 7200},
+        ]
+        result = _simulate_note_injection(
+            history=history,
+            user_message="",  # the empty auto-resume event text
+            resume_entry=entry,
+            window_secs=1800,
+        )
+        assert result.strip(), "blank turn must never reach the model"
+        assert "[System note:" in result
+
+    def test_empty_turn_guard_only_applies_to_resume_pending(self):
+        """The empty-turn backfill must NOT fire for ordinary sessions —
+        a legitimately empty user turn (e.g. an uncaptioned image) on a
+        non-resume_pending session is left untouched.
+        """
+        result = _simulate_note_injection(
+            history=[
+                {"role": "assistant", "content": "hi", "timestamp": time.time()},
+            ],
+            user_message="",
+            resume_entry=None,
+        )
+        assert result == ""
+
     def test_fresh_tool_tail_preserves_auto_continue_note(self):
         history = [
             {"role": "assistant", "content": None, "tool_calls": [