fix: prevent duplicate update prompt spam in gateway watcher (#8343)

The _watch_update_progress() poll loop never deleted .update_prompt.json
after forwarding the prompt to the user, causing the same prompt to be
re-sent every poll cycle (2s). Two fixes:

1. Delete .update_prompt.json after forwarding — the update process only
   polls for .update_response, it doesn't need the prompt file to persist.
2. Guard re-sends with _update_prompt_pending check — belt-and-suspenders
   to prevent duplicates even under race conditions.

Add regression test asserting the prompt is sent exactly once.

gateway/run.py

@@ -6668,8 +6668,12 @@ class GatewayRunner:
             if buffer.strip() and (loop.time() - last_stream_time) >= stream_interval:
                 await _flush_buffer()
 
-            # Check for prompts
+            # Check for prompts — only forward if we haven't already sent
-            if prompt_path.exists() and session_key:
+            # one that's still awaiting a response.  Without this guard the
+            # watcher would re-read the same .update_prompt.json every poll
+            # cycle and spam the user with duplicate prompt messages.
+            if (prompt_path.exists() and session_key
+                    and not self._update_prompt_pending.get(session_key)):
                 try:
                     prompt_data = json.loads(prompt_path.read_text())
                     prompt_text = prompt_data.get("prompt", "")
@@ -6701,6 +6705,11 @@ class GatewayRunner:
                                 f"or type your answer directly."
                             )
                         self._update_prompt_pending[session_key] = True
+                        # Remove the prompt file so it isn't re-read on the
+                        # next poll cycle.  The update process only needs
+                        # .update_response to continue — it doesn't re-check
+                        # .update_prompt.json while waiting.
+                        prompt_path.unlink(missing_ok=True)
                         logger.info("Forwarded update prompt to %s: %s", session_key, prompt_text[:80])
                 except (json.JSONDecodeError, OSError) as e:
                     logger.debug("Failed to read update prompt: %s", e)

tests/gateway/test_update_streaming.py

@@ -403,6 +403,56 @@ class TestWatchUpdateProgress:
 
         # Should not crash; legacy notification handles this case
 
+    @pytest.mark.asyncio
+    async def test_prompt_forwarded_only_once(self, tmp_path):
+        """Regression: prompt must not be re-sent on every poll cycle.
+
+        Before the fix, the watcher never deleted .update_prompt.json after
+        forwarding, causing the same prompt to be sent every poll_interval.
+        """
+        runner = _make_runner()
+        hermes_home = tmp_path / "hermes"
+        hermes_home.mkdir()
+
+        pending = {"platform": "telegram", "chat_id": "111", "user_id": "222",
+                   "session_key": "agent:main:telegram:dm:111"}
+        (hermes_home / ".update_pending.json").write_text(json.dumps(pending))
+        (hermes_home / ".update_output.txt").write_text("")
+
+        mock_adapter = AsyncMock()
+        runner.adapters = {Platform.TELEGRAM: mock_adapter}
+
+        # Write the prompt file up front (before the watcher starts).
+        # The watcher should forward it exactly once, then delete it.
+        prompt = {"prompt": "Would you like to configure new options now? Y/n",
+                  "default": "n", "id": "dup-test"}
+        (hermes_home / ".update_prompt.json").write_text(json.dumps(prompt))
+
+        async def finish_after_polls():
+            # Wait long enough for multiple poll cycles to occur, then
+            # simulate a response + completion.
+            await asyncio.sleep(1.0)
+            (hermes_home / ".update_response").write_text("n")
+            await asyncio.sleep(0.3)
+            (hermes_home / ".update_exit_code").write_text("0")
+
+        with patch("gateway.run._hermes_home", hermes_home):
+            task = asyncio.create_task(finish_after_polls())
+            await runner._watch_update_progress(
+                poll_interval=0.1,
+                stream_interval=0.2,
+                timeout=10.0,
+            )
+            await task
+
+        # Count how many times the prompt text was sent
+        all_sent = [str(c) for c in mock_adapter.send.call_args_list]
+        prompt_sends = [s for s in all_sent if "configure new options" in s]
+        assert len(prompt_sends) == 1, (
+            f"Prompt was sent {len(prompt_sends)} times (expected 1). "
+            f"All sends: {all_sent}"
+        )
+
 
 # ---------------------------------------------------------------------------
 # Message interception for update prompts