Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
abhay-codes07 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ab929a0
fix(inner): drop the prompt future from _pending on every ACP turn exit
abhay-codes07 Aug 7, 2026
e8b55f4
fix(inner): apply the same _pending cleanup to the goose and qwen exe…
abhay-codes07 Aug 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 54 additions & 45 deletions omnigent/inner/acp_executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
136 changes: 73 additions & 63 deletions omnigent/inner/goose_executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1184,75 +1184,85 @@ 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 goose 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"goose process error: {exc}", retryable=True)
# The stdout reader pops ``req_id`` only when it matches a RESPONSE.
# On timeout, or when the reader resolves ``fut`` with an exception
# (EOF / reader error) without a match, nothing removes it, so
# ``self._pending`` would grow one stale entry per silent/failed turn
# on a long-lived session. Drop it on every exit (mirrors ``_rpc``).
try:
while True:
remaining = deadline - loop.time()
if remaining <= 0:
yield ExecutorError(
message="Timeout waiting for goose 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"goose 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", {})
update_type = update.get("sessionUpdate", "")

if update_type == _UPDATE_AGENT_MESSAGE_CHUNK:
content = update.get("content", {})
text = content.get("text", "") if isinstance(content, dict) else ""
if text:
accumulated_text.append(text)
yield TextChunk(text=text)
elif update_type == _UPDATE_USAGE:
size = update.get("size")
if isinstance(size, int) and size > 0:
self._context_window = size
elif update_type == _UPDATE_TOOL_CALL:
logger.debug("goose tool_call: %s", update.get("title", "tool_call"))
elif update_type == _UPDATE_TOOL_CALL_UPDATE:
pass

elif notification.get("id") is not None and notification.get("method"):
# Server-initiated request (session/request_permission): routes
# through policy + elicitation. Blocks while the human decides.
await self._respond_to_agent_request(notification)
# Surface any fs ToolCall events the handler buffered so the
# I/O shows in history.
while self._fs_events:
yield self._fs_events.pop(0)

# Inbound message = progress; reset the idle deadline (after the
# approval block so a slow approval doesn't time out).
deadline = loop.time() + _PROMPT_TIMEOUT_SECONDS
method = notification.get("method", "")
params = notification.get("params", {})

if method == _CLIENT_NOTIFICATION_SESSION_UPDATE:
update = params.get("update", {})
update_type = update.get("sessionUpdate", "")

if update_type == _UPDATE_AGENT_MESSAGE_CHUNK:
content = update.get("content", {})
text = content.get("text", "") if isinstance(content, dict) else ""
if text:
accumulated_text.append(text)
yield TextChunk(text=text)
elif update_type == _UPDATE_USAGE:
size = update.get("size")
if isinstance(size, int) and size > 0:
self._context_window = size
elif update_type == _UPDATE_TOOL_CALL:
logger.debug("goose tool_call: %s", update.get("title", "tool_call"))
elif update_type == _UPDATE_TOOL_CALL_UPDATE:
pass

elif notification.get("id") is not None and notification.get("method"):
# Server-initiated request (session/request_permission): routes
# through policy + elicitation. Blocks while the human decides.
await self._respond_to_agent_request(notification)
# Surface any fs ToolCall events the handler buffered so the
# I/O shows in history.
while self._fs_events:
yield self._fs_events.pop(0)

# Inbound message = progress; reset the idle deadline (after the
# approval block so a slow approval doesn't time out).
deadline = loop.time() + _PROMPT_TIMEOUT_SECONDS
finally:
self._pending.pop(req_id, None)

async def close_session(self, session_key: str) -> None:
"""Close a named session (no-op; the ACP session is per-process)."""
Expand Down
Loading
Loading