diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py
index f9bbefaf..050f0cf7 100644
--- a/nanobot/agent/loop.py
+++ b/nanobot/agent/loop.py
@@ -1810,12 +1810,16 @@ class AgentLoop:
         on_stream_end: Callable[..., Awaitable[None]] | None = None,
         ephemeral: bool = False,
         tools: ToolRegistry | None = None,
+        persist_user_message: bool = True,
     ) -> OutboundMessage | None:
         """Process a message directly and return the outbound payload."""
         await self._connect_mcp()
+        metadata: dict[str, Any] = {}
+        if not persist_user_message:
+            metadata[turn_continuation.SKIP_USER_PERSIST_META] = True
         msg = InboundMessage(
             channel=channel, sender_id="user", chat_id=chat_id,
-            content=content, media=media or [],
+            content=content, media=media or [], metadata=metadata,
         )
         # Share the dispatch lock so direct calls serialize with bus turns.
         lock = self._session_locks.setdefault(session_key, asyncio.Lock())
diff --git a/nanobot/api/server.py b/nanobot/api/server.py
index 0fd35a97..be3f58c9 100644
--- a/nanobot/api/server.py
+++ b/nanobot/api/server.py
@@ -340,6 +340,7 @@ async def handle_chat_completions(request: web.Request) -> web.Response:
                             session_key=session_key,
                             channel="api",
                             chat_id=API_CHAT_ID,
+                            persist_user_message=False,
                         ),
                         timeout=timeout_s,
                     )
diff --git a/nanobot/session/turn_continuation.py b/nanobot/session/turn_continuation.py
index 6b2a5bac..2cb5e868 100644
--- a/nanobot/session/turn_continuation.py
+++ b/nanobot/session/turn_continuation.py
@@ -22,6 +22,7 @@ INTERNAL_CONTINUATION_META = "_internal_continuation"
 INTERNAL_CONTINUATION_KIND_META = "_internal_continuation_kind"
 INTERNAL_CONTINUATION_PENDING_META = "_internal_continuation_pending"
 INTERNAL_CONTINUATION_RUN_STARTED_AT_META = "_internal_continuation_run_started_at"
+SKIP_USER_PERSIST_META = "_skip_user_persist"
 
 _GOAL_CONTINUATION_KIND = "sustained_goal"
 _GOAL_CONTINUATION_SENDER = "system:continuation"
@@ -59,6 +60,8 @@ def internal_continuation_run_started_at(metadata: Mapping[str, Any] | None) ->
 
 def should_persist_user_message(metadata: Mapping[str, Any] | None) -> bool:
     """Return whether this inbound message should be persisted as user input."""
+    if metadata and metadata.get(SKIP_USER_PERSIST_META) is True:
+        return False
     return not internal_continuation_inbound(metadata)
 
 
diff --git a/tests/test_openai_api.py b/tests/test_openai_api.py
index ba75435a..5ba6541f 100644
--- a/tests/test_openai_api.py
+++ b/tests/test_openai_api.py
@@ -400,6 +400,32 @@ async def test_empty_response_retry_then_success(aiohttp_client) -> None:
     assert call_count == 2
 
 
+@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
+@pytest.mark.asyncio
+async def test_empty_response_retry_does_not_duplicate_user_turn(aiohttp_client) -> None:
+    persist_flags = []
+
+    async def record(content, session_key="", channel="", chat_id="", **kwargs):
+        persist_flags.append(kwargs.get("persist_user_message", True))
+        return "" if len(persist_flags) == 1 else "recovered response"
+
+    agent = MagicMock()
+    agent.process_direct = record
+    agent._connect_mcp = AsyncMock()
+    agent.close_mcp = AsyncMock()
+    agent._last_usage = {}
+
+    app = create_app(agent, model_name="m")
+    client = await aiohttp_client(app)
+    resp = await client.post(
+        "/v1/chat/completions",
+        json={"messages": [{"role": "user", "content": "hello"}]},
+    )
+    assert resp.status == 200
+    # first call persists the user turn; the retry must not persist it again
+    assert persist_flags == [True, False]
+
+
 @pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
 @pytest.mark.asyncio
 async def test_empty_response_falls_back(aiohttp_client) -> None: