fix(inner): drop the prompt future from _pending on every ACP-family turn exit (acp/goose/qwen)

omnigent/inner/acp_executor.py

@@ -1108,57 +1108,66 @@ async def run_turn(
         deadline = loop.time() + _PROMPT_TIMEOUT_SECONDS
         accumulated_text: list[str] = []
 
-        while True:
-            remaining = deadline - loop.time()
-            if remaining <= 0:
-                yield ExecutorError(message="Timeout waiting for ACP response", retryable=True)
-                return
-
-            # Complete only once the future is resolved AND the queue is drained,
-            # so trailing chunks aren't truncated.
-            if fut.done() and self._queue.empty():
-                try:
-                    response = fut.result()
-                except Exception as exc:  # noqa: BLE001
-                    self._session_id = None
-                    self._system_prompt_sent = False
-                    yield ExecutorError(message=f"ACP process error: {exc}", retryable=True)
+        # The stdout reader pops ``req_id`` only when it matches a RESPONSE
+        # (see ``_read_stdout``). On the timeout path below, or when the reader
+        # resolves ``fut`` with an exception (EOF / reader error) without a
+        # match, nothing removes ``req_id`` — so ``self._pending`` would grow one
+        # stale entry per silent/failed turn on a long-lived ACP session. Drop
+        # it in ``finally`` on every exit (mirrors ``_rpc``'s timeout cleanup).
+        try:
+            while True:
+                remaining = deadline - loop.time()
+                if remaining <= 0:
+                    yield ExecutorError(message="Timeout waiting for ACP response", retryable=True)
                     return
-                if "error" in response:
-                    error_msg = response["error"].get("message", "Unknown ACP error")
-                    if "Session not found" in error_msg:
+
+                # Complete only once the future is resolved AND the queue is drained,
+                # so trailing chunks aren't truncated.
+                if fut.done() and self._queue.empty():
+                    try:
+                        response = fut.result()
+                    except Exception as exc:  # noqa: BLE001
                         self._session_id = None
                         self._system_prompt_sent = False
-                    yield ExecutorError(message=error_msg, retryable=True)
+                        yield ExecutorError(message=f"ACP process error: {exc}", retryable=True)
+                        return
+                    if "error" in response:
+                        error_msg = response["error"].get("message", "Unknown ACP error")
+                        if "Session not found" in error_msg:
+                            self._session_id = None
+                            self._system_prompt_sent = False
+                        yield ExecutorError(message=error_msg, retryable=True)
+                        return
+                    result = response.get("result", {}) if isinstance(response, dict) else {}
+                    usage = self._usage_from_result(result) if isinstance(result, dict) else None
+                    yield TurnComplete(response="".join(accumulated_text), usage=usage)
                     return
-                result = response.get("result", {}) if isinstance(response, dict) else {}
-                usage = self._usage_from_result(result) if isinstance(result, dict) else None
-                yield TurnComplete(response="".join(accumulated_text), usage=usage)
-                return
 
-            try:
-                notification = await asyncio.wait_for(
-                    self._queue.get(), timeout=min(remaining, 2.0)
-                )
-            except asyncio.TimeoutError:
-                continue
+                try:
+                    notification = await asyncio.wait_for(
+                        self._queue.get(), timeout=min(remaining, 2.0)
+                    )
+                except asyncio.TimeoutError:
+                    continue
 
-            method = notification.get("method", "")
-            params = notification.get("params", {})
-
-            if method == _CLIENT_NOTIFICATION_SESSION_UPDATE:
-                update = params.get("update", {})
-                for event in self._handle_session_update(update):
-                    if isinstance(event, TextChunk):
-                        accumulated_text.append(event.text)
-                    yield event
-            elif notification.get("id") is not None and notification.get("method"):
-                # Server-initiated request (session/request_permission / fs/*):
-                # routes through policy + elicitation. Blocks while the human decides.
-                await self._respond_to_agent_request(notification)
-
-            # Inbound message = progress; reset the idle deadline.
-            deadline = loop.time() + _PROMPT_TIMEOUT_SECONDS
+                method = notification.get("method", "")
+                params = notification.get("params", {})
+
+                if method == _CLIENT_NOTIFICATION_SESSION_UPDATE:
+                    update = params.get("update", {})
+                    for event in self._handle_session_update(update):
+                        if isinstance(event, TextChunk):
+                            accumulated_text.append(event.text)
+                        yield event
+                elif notification.get("id") is not None and notification.get("method"):
+                    # Server-initiated request (session/request_permission / fs/*):
+                    # routes through policy + elicitation. Blocks while the human decides.
+                    await self._respond_to_agent_request(notification)
+
+                # Inbound message = progress; reset the idle deadline.
+                deadline = loop.time() + _PROMPT_TIMEOUT_SECONDS
+        finally:
+            self._pending.pop(req_id, None)
 
     async def interrupt_session(self, session_key: str) -> bool:  # noqa: ARG002 — one ACP session per process
         """Abort the running turn via the ACP ``session/cancel`` notification.

tests/inner/test_acp_executor.py

@@ -25,6 +25,7 @@
 from omnigent.inner._acp_omnigent_mcp import OmnigentAcpMcp, _to_acp_mcp_servers
 from omnigent.inner.acp_executor import AcpAgentConfig, AcpExecutor
 from omnigent.inner.executor import (
+    ExecutorError,
     ReasoningChunk,
     TextChunk,
     ToolCallComplete,
@@ -569,3 +570,51 @@ async def deny(tool_name: str, tool_input: dict) -> bool:
 
     # Turn still completes even though the tool was rejected.
     assert any(isinstance(e, TurnComplete) for e in events)
+
+
+# ---------------------------------------------------------------------------
+# run_turn cleans up its pending prompt future on every exit path
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_run_turn_timeout_does_not_leak_pending_future(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A turn that times out must not leave its request future in ``_pending``.
+
+    The stdout reader pops ``req_id`` only when it matches a response. On the
+    timeout path (and the EOF/reader-error path) nothing removes it, so before
+    the ``finally`` cleanup ``_pending`` grew one stale entry per silent/failed
+    turn on a long-lived ACP session. Drive a turn whose response never arrives
+    and assert the map is empty afterward.
+    """
+    from types import SimpleNamespace
+
+    import omnigent.inner.acp_executor as acp_mod
+
+    # Time out the prompt loop near-instantly instead of after the 300s default.
+    monkeypatch.setattr(acp_mod, "_PROMPT_TIMEOUT_SECONDS", 0.05)
+
+    ex = AcpExecutor(AcpAgentConfig(command="x"))
+
+    async def _noop(*_a: object, **_k: object) -> None:
+        return None
+
+    async def _sess() -> str:
+        return "sess-1"
+
+    # No real subprocess / handshake; nothing ever resolves the prompt future,
+    # so run_turn falls through to the timeout branch.
+    monkeypatch.setattr(ex, "_start_process", _noop)
+    monkeypatch.setattr(ex, "_ensure_initialized", _noop)
+    monkeypatch.setattr(ex, "_ensure_session", _sess)
+    monkeypatch.setattr(ex, "_send", _noop)
+    ex._proc = SimpleNamespace(returncode=None)  # type: ignore[assignment]
+
+    events = [event async for event in ex.run_turn([{"role": "user", "content": "hi"}], [], "sys")]
+
+    # It surfaced the timeout...
+    assert any(isinstance(e, ExecutorError) for e in events)
+    # ...and, the point of the fix, left no stale future behind.
+    assert ex._pending == {}, "run_turn must drop its prompt future from _pending on timeout"