fix(gateway): close failed turns before future replay

gateway/run_turn.py

@@ -1519,6 +1519,17 @@ class GatewayTurnMixin:
         except Exception as e:
             logger.debug("Watch queue drain error: %s", e)
 
+    _FAILED_TURN_NOTICE = (
+        "Your request was not processed. Send it again if you still want me to carry it out."
+    )
+
+    def _hmwa_add_failed_turn_notice(self, response):
+        """Make failed-turn delivery explicit without replacing the provider-specific guidance."""
+        response = str(response or "").strip()
+        if self._FAILED_TURN_NOTICE in response:
+            return response
+        return f"{response}\n\n{self._FAILED_TURN_NOTICE}" if response else self._FAILED_TURN_NOTICE
+
     def _hmwa_classify_turn_failure(self, agent_result, history, session_entry):
         """Classify a finished turn for transcript persistence. Returns
         ``(agent_failed_early, hidden_reasoning_incomplete, is_context_overflow_failure)``.
@@ -1602,11 +1613,10 @@ class GatewayTurnMixin:
     @staticmethod
     def _hmwa_user_transcript_entry(event, prepared, ts):
         """Transcript row for the inbound user turn (clean text + event time when captured)."""
-        # Transient failure (429/timeout/5xx): persist only the user message so the next message can load a
-        # transcript that reflects what was said. Skip the assistant error text since it's a
-        # gateway-generated hint, not model output. Hidden- reasoning-only incomplete turns follow the same
-        # persistence rule so peer-agent channels don't ingest them as completed assistant turns. (#7100,
-        # #51628)
+        # Transient failure (429/timeout/5xx): persist the user message so the next message can load a
+        # transcript that reflects what was said. The caller pairs it with a stable assistant safety
+        # boundary rather than the provider error text. Hidden-reasoning-only incomplete turns follow the
+        # same persistence rule so peer-agent channels don't ingest provider details. (#7100, #51628)
         _user_entry = {
             "role": "user",
             "content": (
@@ -1627,8 +1637,8 @@ class GatewayTurnMixin:
         self, *, event, source, session_entry, session_key, agent_result, agent_messages,
         prepared, response, agent_failed_early, hidden_reasoning_incomplete, is_context_overflow_failure,
     ):
-        """Persist this turn to the transcript (session_meta on first turn, user-only on transient
-        failure, nothing on context overflow), update last_prompt_tokens, and re-baseline the
+        """Persist this turn to the transcript (session_meta on first turn, closed failed turn on
+        transient failure, nothing on context overflow), update last_prompt_tokens, and re-baseline the
         cached agent's message count."""
         from gateway.run import _resolve_gateway_model
         ts = time.time()  # Unix epoch float — consistent with DB storage
@@ -1661,8 +1671,8 @@ class GatewayTurnMixin:
                     "timestamp": ts,
                 })
             if agent_failed_early or hidden_reasoning_incomplete:
-                # Transient failure / hidden-reasoning incomplete: persist only the user message (the
-                # assistant error text is a gateway hint, not model output). Dedupe on platform
+                # Transient failure / hidden-reasoning incomplete: persist the user message without
+                # the provider error text (a gateway hint, not model output). Dedupe on platform
                 # message_id (Telegram retries after transient failures).
                 if event.message_id and await store.has_platform_message_id(sid, str(event.message_id)):
                     logger.info(
@@ -1671,6 +1681,14 @@ class GatewayTurnMixin:
                     )
                 else:
                     await store.append_to_transcript(sid, _user_row, skip_db=agent_persisted)
+                # Close the failed turn with a durable assistant boundary. Leaving a user-only tail
+                # lets alternation repair merge this request into an unrelated future message and
+                # can replay stale side effects. Persist only the stable safety statement, not raw
+                # provider details; this row is gateway-owned and was not written by the agent.
+                await store.append_to_transcript(
+                    sid,
+                    {"role": "assistant", "content": self._FAILED_TURN_NOTICE, "timestamp": ts},
+                )
             else:
                 # Only the NEW messages: history_offset (what the agent saw), not len(history), which
                 # counts session_meta entries stripped before the agent saw them.
@@ -1775,6 +1793,14 @@ class GatewayTurnMixin:
                     await self.async_session_store.append_to_transcript(
                         session_entry.session_id, self._hmwa_user_transcript_entry(event, prepared, time.time()),
                     )
+                await self.async_session_store.append_to_transcript(
+                    session_entry.session_id,
+                    {
+                        "role": "assistant",
+                        "content": self._FAILED_TURN_NOTICE,
+                        "timestamp": time.time(),
+                    },
+                )
         except Exception:
             logger.debug("Failed to persist inbound user message after agent exception", exc_info=True)
         # Never expose raw exception types/messages to end users (info-leakage risk).
@@ -1798,13 +1824,13 @@ class GatewayTurnMixin:
         elif status_code in {400, 500}:
             # 400/500 on a large session: context overflow / payload too large.
             if len(prepared.history) > 50:
-                return (
+                return self._hmwa_add_failed_turn_notice(
                     "⚠️ Session too large for the model's context window.\nUse /compact to "
                     "compress the conversation, or /reset to start fresh."
                 )
             elif status_code == 400:
                 status_hint = " The request was rejected by the API."
-        return (
+        return self._hmwa_add_failed_turn_notice(
             f"Sorry, I encountered an unexpected error.{status_hint}\n"
             "Try again or use /reset to start a fresh session."
         )
@@ -2010,6 +2036,8 @@ class GatewayTurnMixin:
             agent_failed_early, hidden_reasoning_incomplete, is_context_overflow_failure = (
                 self._hmwa_classify_turn_failure(agent_result, history, session_entry)
             )
+            if agent_failed_early and not is_context_overflow_failure:
+                response = self._hmwa_add_failed_turn_notice(response)
             response, session_entry = await self._hmwa_compression_exhaustion_reset(
                 agent_result, response, session_entry, session_key, source,
             )

tests/gateway/test_42039_duplicate_user_message.py

@@ -137,7 +137,7 @@ async def test_agent_failed_early_skip_db_when_agent_has_session_db(
     runner._run_agent = AsyncMock(
         return_value={
             "failed": True,
-            "final_response": None,
+            "final_response": "API call failed after 3 retries: 429 Too Many Requests",
             "error": "429 Too Many Requests — rate limit exceeded",
             "messages": [],
             "history_offset": 0,
@@ -145,13 +145,30 @@ async def test_agent_failed_early_skip_db_when_agent_has_session_db(
         }
     )
 
-    await runner._handle_message_with_agent(
+    response = await runner._handle_message_with_agent(
         _event(), _source(), "agent:main:telegram:group:-1001:12345", 1
     )
 
     _assert_user_call_has_skip_db(
         runner.session_store.append_to_transcript.call_args_list, True
     )
+    assert "not processed" in response
+
+    transcript_rows = [
+        call.args[1]
+        for call in runner.session_store.append_to_transcript.call_args_list
+        if len(call.args) >= 2 and call.args[1].get("role") in {"user", "assistant"}
+    ]
+    assert [row["role"] for row in transcript_rows] == ["user", "assistant"]
+    assert "not processed" in transcript_rows[-1]["content"]
+
+    # The next unrelated input remains its own turn instead of alternation repair
+    # merging the failed mutating request into it.
+    from agent.agent_runtime_helpers import repair_message_sequence
+
+    replay = [*transcript_rows, {"role": "user", "content": "unrelated question"}]
+    assert repair_message_sequence(None, replay) == 0
+    assert replay[-1]["content"] == "unrelated question"
 
 
 # ── Test 2: agent_failed_early with no _session_db → skip_db not True ─

tests/gateway/test_failure_writer_ownership.py

@@ -98,12 +98,17 @@ def test_failure_owner_follows_only_live_lineage_markers(tmp_path):
                 MessageEvent(text="same", source=source, message_id=pid),
                 source, entry, entry.session_key, prepared,
             )
-            assert db.message_count() == before + (not owned), location
+            # Exception fallback adds the missing user only when unowned, then always
+            # closes the failed turn with a durable assistant safety boundary.
+            assert db.message_count() == before + (not owned) + 1, location
             assert store.has_input_owner(sid, owner), location
+            live_messages = db.get_messages(child)
+            assert live_messages[-1]["role"] == "assistant"
+            assert "not processed" in live_messages[-1]["content"]
             if not owned:
-                latest = db.get_messages(child)[-1]
-                assert latest["content"] == prepared.persist_user_message
-                assert latest["display_metadata"]["gateway_input_owner"] == owner
+                persisted_user = live_messages[-2]
+                assert persisted_user["content"] == prepared.persist_user_message
+                assert persisted_user["display_metadata"]["gateway_input_owner"] == owner
         db.close()
 
     asyncio.run(check())