fix(gateway): make failed-turn retry guidance side-effect safe

gateway/run_turn.py

@@ -1522,13 +1522,31 @@ class GatewayTurnMixin:
     _FAILED_TURN_NOTICE = (
         "Your request was not processed. Send it again if you still want me to carry it out."
     )
+    _PARTIAL_FAILED_TURN_NOTICE = (
+        "This turn did not complete. Some actions may already have run; verify their effects "
+        "before resending."
+    )
 
-    def _hmwa_add_failed_turn_notice(self, response):
+    def _hmwa_add_failed_turn_notice(self, response, notice=None):
         """Make failed-turn delivery explicit without replacing the provider-specific guidance."""
+        notice = notice or self._FAILED_TURN_NOTICE
         response = str(response or "").strip()
-        if self._FAILED_TURN_NOTICE in response:
+        if notice in response:
             return response
-        return f"{response}\n\n{self._FAILED_TURN_NOTICE}" if response else self._FAILED_TURN_NOTICE
+        return f"{response}\n\n{notice}" if response else notice
+
+    def _hmwa_failed_turn_notice(self, agent_result):
+        """Choose retry guidance without assuming completed tool effects can be repeated safely."""
+        messages = agent_result.get("messages", []) or []
+        history_offset = agent_result.get("history_offset", 0)
+        turn_messages = messages[history_offset:]
+        if any(
+            message.get("role") == "tool"
+            or (message.get("role") == "assistant" and message.get("tool_calls"))
+            for message in turn_messages
+        ):
+            return self._PARTIAL_FAILED_TURN_NOTICE
+        return self._FAILED_TURN_NOTICE
 
     def _hmwa_classify_turn_failure(self, agent_result, history, session_entry):
         """Classify a finished turn for transcript persistence. Returns
@@ -1636,6 +1654,7 @@ class GatewayTurnMixin:
     async def _hmwa_persist_turn_transcript(
         self, *, event, source, session_entry, session_key, agent_result, agent_messages,
         prepared, response, agent_failed_early, hidden_reasoning_incomplete, is_context_overflow_failure,
+        failed_turn_notice,
     ):
         """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
@@ -1687,7 +1706,7 @@ class GatewayTurnMixin:
                 # 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},
+                    {"role": "assistant", "content": failed_turn_notice, "timestamp": ts},
                 )
             else:
                 # Only the NEW messages: history_offset (what the agent saw), not len(history), which
@@ -1797,7 +1816,7 @@ class GatewayTurnMixin:
                     session_entry.session_id,
                     {
                         "role": "assistant",
-                        "content": self._FAILED_TURN_NOTICE,
+                        "content": self._PARTIAL_FAILED_TURN_NOTICE,
                         "timestamp": time.time(),
                     },
                 )
@@ -1826,13 +1845,15 @@ class GatewayTurnMixin:
             if len(prepared.history) > 50:
                 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."
+                    "compress the conversation, or /reset to start fresh.",
+                    self._PARTIAL_FAILED_TURN_NOTICE,
                 )
             elif status_code == 400:
                 status_hint = " The request was rejected by the API."
         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."
+            "Use /reset to start a fresh session if needed.",
+            self._PARTIAL_FAILED_TURN_NOTICE,
         )
 
     def _hmwa_discard_stale_result(self, source, _quick_key, run_generation):
@@ -2036,8 +2057,9 @@ class GatewayTurnMixin:
             agent_failed_early, hidden_reasoning_incomplete, is_context_overflow_failure = (
                 self._hmwa_classify_turn_failure(agent_result, history, session_entry)
             )
+            failed_turn_notice = self._hmwa_failed_turn_notice(agent_result)
             if agent_failed_early and not is_context_overflow_failure:
-                response = self._hmwa_add_failed_turn_notice(response)
+                response = self._hmwa_add_failed_turn_notice(response, failed_turn_notice)
             response, session_entry = await self._hmwa_compression_exhaustion_reset(
                 agent_result, response, session_entry, session_key, source,
             )
@@ -2047,6 +2069,7 @@ class GatewayTurnMixin:
                 response=response, agent_failed_early=agent_failed_early,
                 hidden_reasoning_incomplete=hidden_reasoning_incomplete,
                 is_context_overflow_failure=is_context_overflow_failure,
+                failed_turn_notice=failed_turn_notice,
             )
             return await self._hmwa_deliver_turn_response(
                 event, source, session_entry, session_key, run_generation,

tests/gateway/test_42039_duplicate_user_message.py

@@ -171,6 +171,51 @@ async def test_agent_failed_early_skip_db_when_agent_has_session_db(
     assert replay[-1]["content"] == "unrelated question"
 
 
+@pytest.mark.asyncio
+async def test_failed_turn_with_tool_activity_does_not_recommend_blind_retry(
+    monkeypatch, tmp_path
+):
+    runner = _bootstrap(monkeypatch, tmp_path)
+    runner._run_agent = AsyncMock(
+        return_value={
+            "failed": True,
+            "final_response": "API call failed after 3 retries: 500 Internal Server Error",
+            "error": "500 Internal Server Error",
+            "messages": [
+                {"role": "user", "content": "reset the password"},
+                {
+                    "role": "assistant",
+                    "content": "",
+                    "tool_calls": [
+                        {
+                            "id": "call-1",
+                            "function": {"name": "reset_password", "arguments": "{}"},
+                        }
+                    ],
+                },
+                {"role": "tool", "tool_call_id": "call-1", "content": "Password reset"},
+            ],
+            "history_offset": 0,
+            "last_prompt_tokens": 0,
+        }
+    )
+
+    response = await runner._handle_message_with_agent(
+        _event(), _source(), "agent:main:telegram:group:-1001:12345", 1
+    )
+
+    assistant_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") == "assistant"
+    ]
+    assert len(assistant_rows) == 1
+    assert assistant_rows[0]["content"] == runner._PARTIAL_FAILED_TURN_NOTICE
+    assert runner._PARTIAL_FAILED_TURN_NOTICE in response
+    assert "not processed" not in response
+    assert "Send it again" not in response
+
+
 # ── Test 2: agent_failed_early with no _session_db → skip_db not True ─
 
 

tests/gateway/test_failure_writer_ownership.py

@@ -93,7 +93,7 @@ def test_failure_owner_follows_only_live_lineage_markers(tmp_path):
             store._transcript_reroutes.clear()
             assert store.has_input_owner(sid, owner) is owned, location
             before = db.message_count()
-            await runner._hmwa_agent_error_reply(
+            reply = await runner._hmwa_agent_error_reply(
                 RuntimeError("controlled post-compaction failure"),
                 MessageEvent(text="same", source=source, message_id=pid),
                 source, entry, entry.session_key, prepared,
@@ -104,7 +104,10 @@ def test_failure_owner_follows_only_live_lineage_markers(tmp_path):
             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"]
+            assert "Some actions may already have run" in live_messages[-1]["content"]
+            assert "not processed" not in live_messages[-1]["content"]
+            assert "Some actions may already have run" in reply
+            assert "not processed" not in reply
             if not owned:
                 persisted_user = live_messages[-2]
                 assert persisted_user["content"] == prepared.persist_user_message

tests/gateway/test_incomplete_gateway_turns.py

@@ -123,7 +123,7 @@ def _make_event() -> MessageEvent:
 
 
 @pytest.mark.asyncio
-async def test_incomplete_codex_turn_stays_out_of_slack_transcript(monkeypatch, tmp_path):
+async def test_incomplete_codex_turn_closes_transcript_without_slack_delivery(monkeypatch, tmp_path):
     adapter = CaptureSlackAdapter()
     runner = _make_runner(adapter)
 
@@ -148,8 +148,11 @@ async def test_incomplete_codex_turn_stays_out_of_slack_transcript(monkeypatch,
         call.args[1]["role"]
         for call in runner.session_store.append_to_transcript.call_args_list
     ]
-    assert transcript_roles == ["session_meta", "user"]
+    assert transcript_roles == ["session_meta", "user", "assistant"]
     assert runner.session_store.append_to_transcript.call_args_list[1].args[1]["content"] == "hello"
+    boundary = runner.session_store.append_to_transcript.call_args_list[2].args[1]["content"]
+    assert "not processed" in boundary
+    assert "remained incomplete" not in boundary
     assert adapter.processing_hooks == [
         ("start", "m-1"),
         ("complete", "m-1", ProcessingOutcome.SUCCESS),