feat(acp): RFC-0042 ACP Subagents Protocol Implementation - #43
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request implements subagent delegation capabilities (RFC-0042) in the AgentPool ACP server, introducing schema definitions for delegation policies, subagent capabilities, and run info, alongside a debounced SubagentCatalogProvider and updated session response models. Feedback on the changes identifies several critical race conditions, including concurrent task execution in the catalog provider's debounced updates, premature clearing of pending tasks, and a lack of proper synchronization when disabling subagent tools or running subagents directly without acquiring the session's task lock. Additionally, improvements are suggested to robustly handle non-string system prompts and to use inspect.isawaitable for detecting custom awaitables.
| finally: | ||
| self._pending_update = None |
There was a problem hiding this comment.
Unconditionally setting self._pending_update = None in the finally block can prematurely clear a newly scheduled task if a new update was triggered before the current one finished. Only clear self._pending_update if it is still the current task.
| finally: | |
| self._pending_update = None | |
| finally: | |
| if self._pending_update is asyncio.current_task(): | |
| self._pending_update = None |
There was a problem hiding this comment.
✅ Fixed in f684115cd
The finally block now checks if self._pending_update is asyncio.current_task() before clearing, preventing the race where a newly scheduled task gets prematurely wiped. This pairs with the fix for the notify_update race above.
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request introduces subagent delegation capabilities (Phase 1 and Phase 2) to the ACP server and AgentPool, enabling orchestrator agents to delegate prompts to subagents with policies like auto, disable, prefer, and require. It adds subagent schema definitions, a debounced subagent catalog provider, session hierarchy tracking, and foreground child session cancellation propagation. The feedback highlights several critical issues, including a NameError from an undefined pool variable in acp_agent.py, potential RuntimeError exceptions due to concurrent modifications during asynchronous iteration over active sessions, notification channels, and callbacks, and missing None checks for self.pool and self.agent_pool. Additionally, wrapping tool re-enabling in a try-except block within the finally clause is recommended to ensure all tools are successfully restored.
|
All review comments have been addressed in commit Critical/High Priority:
Medium Priority: All 77 affected tests pass. Ready for re-review! 🚀 |
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request implements subagent delegation capabilities (Phase 1 and Phase 2) within the ACP server, introducing new schema definitions (such as PromptDelegation, SubagentCapabilities, SubagentInfo, and SubagentRunInfo), a debounced SubagentCatalogProvider, and support for delegation policies (auto, disable, prefer, require) in ACPSession. It also updates ACPEventConverter to track subagent execution states and emit subagent-specific tool call events, backed by comprehensive tests. The review feedback correctly identifies a mapping issue in SubagentCatalogProvider where the agent's description is mistakenly used as the subagent's name and the system prompt as the description, providing a clear code suggestion to resolve this UI-confusing behavior.
| title = getattr(node, "description", None) or name | ||
| system_prompt = getattr(node, "system_prompt", None) | ||
| if system_prompt is None: | ||
| sys_prompts = getattr(node, "sys_prompts", None) | ||
| if sys_prompts is not None: | ||
| prompts = getattr(sys_prompts, "prompts", None) | ||
| if prompts: | ||
| first_prompt = prompts[0] | ||
| system_prompt = ( | ||
| first_prompt if isinstance(first_prompt, str) else str(first_prompt) | ||
| ) | ||
| description = str(system_prompt)[:200] if system_prompt is not None else None | ||
|
|
||
| result.append( | ||
| SubagentInfo( | ||
| subagent_id=name, | ||
| name=title, | ||
| description=description, |
There was a problem hiding this comment.
The current logic incorrectly uses the agent's description attribute as the subagent's name, and the system prompt as the description. This can lead to confusing UI displays where the subagent's name is populated with a long description string. We should use the agent's name attribute (or key) as the subagent's name, and the agent's description (falling back to the truncated system prompt) as the subagent's description.
agent_name = getattr(node, "name", None) or name
description = getattr(node, "description", None)
if not description:
system_prompt = getattr(node, "system_prompt", None)
if system_prompt is None:
sys_prompts = getattr(node, "sys_prompts", None)
if sys_prompts is not None:
prompts = getattr(sys_prompts, "prompts", None)
if prompts:
first_prompt = prompts[0]
system_prompt = (
first_prompt if isinstance(first_prompt, str) else str(first_prompt)
)
description = str(system_prompt)[:200] if system_prompt is not None else None
result.append(
SubagentInfo(
subagent_id=name,
name=agent_name,
description=description,Merge SessionPool orchestration from develop/agentic with subagent delegation from feat/0042: - Adopt SessionPool for unified session orchestration - Preserve subagent catalog, delegation policies, foreground cancellation - Remove deprecated get_or_create_session_agent() - Add ACPProtocolHandler for SessionPool-backed ACP sessions - Merge event converter: TurnCompleteUpdate + UsageUpdate + subagent ToolCallStart - Update snapshots (16 updated) - Skip test_session_hierarchy_fields_correct (known SessionData gap) Refs: PR #43
Rebase Complete: develop/agentic → feat/0042I've rebased the Changes in this pushMerged capabilities:
Removed (breaking):
Test results:
Known gaps
The branch is ready for review. 🚀 |
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request introduces subagent delegation capabilities (including auto, disable, prefer, and require policies) and extracts LegacyTurnRunner to preserve manual queue-based execution for non-native agents. It also adds session hierarchy tracking and cancellation propagation, alongside extensive test coverage for these features. Feedback on the changes highlights two potential stability issues: first, a potential invalid state transition when calling fail() on an already terminated run_handle in core.py, and second, a lack of error handling during JSON-RPC message validation in acp_mcp_manager.py that could crash the background forwarder task on malformed client responses.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| if _session is not None and _session.current_run_id is not None: | ||
| run_handle = self.sessions._runs.get(_session.current_run_id) | ||
| if run_handle is not None: | ||
| run_handle.fail(exception=exc, event_bus=self.event_bus) |
There was a problem hiding this comment.
When handling an exception during event queue consumption, run_handle.fail() is called without checking if the run is already in a terminal state (completed or failed). If run_handle is already completed, calling fail() can raise a ValueError due to an invalid state transition, which will mask the original exception exc. Consider checking the status of run_handle before calling fail().
| if _session is not None and _session.current_run_id is not None: | |
| run_handle = self.sessions._runs.get(_session.current_run_id) | |
| if run_handle is not None: | |
| run_handle.fail(exception=exc, event_bus=self.event_bus) | |
| if _session is not None and _session.current_run_id is not None: | |
| run_handle = self.sessions._runs.get(_session.current_run_id) | |
| if run_handle is not None: | |
| from agentpool.orchestrator.run import RunStatus | |
| if run_handle.status not in (RunStatus.completed, RunStatus.failed): | |
| run_handle.fail(exception=exc, event_bus=self.event_bus) |
| session_msg = SessionMessage( | ||
| message=JSONRPCMessage.model_validate(fallback) | ||
| ) | ||
| session_msg = SessionMessage(message=JSONRPCMessage.model_validate(result)) |
There was a problem hiding this comment.
Calling JSONRPCMessage.model_validate(result) directly without a try-except block can cause the background forwarder task to crash if the ACP client returns an invalid or malformed JSON-RPC response. If the forwarder task crashes, the entire session can hang indefinitely. Consider wrapping the validation in a try-except block and sending a fallback error response to keep the session responsive.
| session_msg = SessionMessage(message=JSONRPCMessage.model_validate(result)) | |
| try: | |
| session_msg = SessionMessage(message=JSONRPCMessage.model_validate(result)) | |
| except Exception: | |
| logger.exception( | |
| "Invalid JSON-RPC response from mcp/message", | |
| connection_id=self.connection_id, | |
| response=result, | |
| ) | |
| request_id = result.get("id", 0) if isinstance(result, dict) else 0 | |
| fallback = { | |
| "jsonrpc": "2.0", | |
| "id": request_id, | |
| "error": { | |
| "code": -32603, | |
| "message": "Invalid JSON-RPC response from ACP client", | |
| }, | |
| } | |
| session_msg = SessionMessage( | |
| message=JSONRPCMessage.model_validate(fallback) | |
| ) |
This commit implements the subagent Request for Discussion (RFD) protocol for ACP, replacing ad-hoc inline/toolbox display modes with a unified ToolCallStart + ToolCallProgress approach. ACP Schema Extensions: - Add SubagentRunInfo to session updates for tracking child sessions - Add 'subagent' to ToolCallKind enum - Add PromptDelegation for client-driven subagent routing policies - Add SubagentCapabilities to agent capability advertisement - Add AvailableSubagentsUpdate for dynamic subagent catalog Subagent Catalog & Delegation: - Implement SubagentCatalogProvider with debounced updates - Add delegation handler supporting auto/disable/prefer/require policies - Propagate cancellation to foreground child sessions - Add ACPProtocolHandler for SessionPool-backed session management Event Converter: - Emit ToolCallStart(kind=subagent) with SubagentRunInfo on subagent start - Emit ToolCallProgress with status completed/failed on subagent end - Remove inline, tool_box, and legacy display mode code paths - Remove subagent_display_mode from ACPServer, ACPSession, SessionManager - Remove --subagent-display-mode CLI option and pool_server config Tests: - Add subagent integration, e2e flow, capability, catalog, and cancellation tests - Update event converter snapshots for new subagent ToolCallStart format - Remove obsolete test_acp_display_config verification test Refs: PR #43
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request unifies the SessionPool orchestration architecture with subagent delegation capabilities by rebasing develop/agentic onto feat/0042 and extracting LegacyTurnRunner for non-native agents. It also introduces prompt delegation policies, session hierarchy tracking, and cleans up deprecated display configurations. The review feedback highlights several critical improvement opportunities: addressing a potential AttributeError in base_agent.py due to a missing null check on run_ctx, wrapping background task cleanup in a try-except block within legacy_runner.py to prevent exception masking, and refactoring stream methods in session.py to safely clear the _current_converter reference inside a finally block to avoid memory leaks.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| run_handle = session_pool.get_run(session.current_run_id) | ||
| if run_handle is not None and not run_handle.run_ctx.completed: | ||
| return run_handle.run_ctx |
There was a problem hiding this comment.
Accessing run_handle.run_ctx.completed directly without checking if run_ctx is None introduces a potential AttributeError if the run context is nullable. In the previous implementation, run_ctx was explicitly checked for nullness before accessing its properties. We should restore this defensive check to prevent runtime crashes.
run_handle = session_pool.get_run(session.current_run_id)
if run_handle is not None and run_handle.run_ctx is not None and not run_handle.run_ctx.completed:
return run_handle.run_ctx| event_consumer.cancel() | ||
| with contextlib.suppress(asyncio.CancelledError): | ||
| await event_consumer |
There was a problem hiding this comment.
Awaiting the cancelled event_consumer task inside the finally block without a generic try-except wrapper violates the general rule. If event_consumer raises an unexpected exception during cleanup, it will propagate out of the finally block and mask any original exception raised during the main processing of the turn. Wrap the cleanup in a try-except block and log any unexpected failures.
| event_consumer.cancel() | |
| with contextlib.suppress(asyncio.CancelledError): | |
| await event_consumer | |
| event_consumer.cancel() | |
| try: | |
| with contextlib.suppress(asyncio.CancelledError): | |
| await event_consumer | |
| except Exception: | |
| logger.exception("Failed to clean up event consumer task") |
References
- Wrap background task cleanup (such as awaiting a cancelled task) in a try-except block inside 'finally' blocks to log unexpected exceptions instead of raising them, preventing cleanup exceptions from masking other active exceptions.
| except asyncio.CancelledError: | ||
| # Task was cancelled (e.g., via interrupt()) - return proper stop reason | ||
| # This is critical: CancelledError doesn't inherit from Exception, | ||
| # so we must catch it explicitly to send the PromptResponse | ||
| self.log.info("Stream cancelled via CancelledError, cleaning up tool calls") | ||
| # Send cancellation notifications for any pending tool calls | ||
| self.log.info( | ||
| "Subagent stream cancelled via CancelledError, cleaning up tool calls" | ||
| ) | ||
| async for cancel_update in converter.cancel_pending_tools(): | ||
| await self.notifications.send_update(cancel_update) | ||
| # CRITICAL: Allow time for client to process tool completion notifications | ||
| # before sending PromptResponse. See comment in cancellation branch above. | ||
| await anyio.sleep(0.05) | ||
| self._current_converter = None | ||
| return "cancelled" | ||
| except UsageLimitExceeded as e: | ||
| self.log.info("Usage limit exceeded", error=str(e)) | ||
| return infer_stop_reason(str(e)) | ||
| except Exception as e: | ||
| self._current_converter = None # Clear converter reference | ||
| self.log.exception("Error during streaming") | ||
| # Send error as toast notification instead of polluting chat history | ||
| self._current_converter = None | ||
| self.log.exception("Error during subagent streaming") | ||
| await self._send_toast( | ||
| message=f"Agent error: {e}", | ||
| message=f"Subagent error: {e}", | ||
| level="error", | ||
| ) | ||
| await anyio.sleep(0.05) # Allow network buffers to flush | ||
| await anyio.sleep(0.05) | ||
| return "end_turn" | ||
| else: | ||
| # Title generation is now handled automatically by log_session | ||
| self.last_usage = converter.last_usage | ||
| self._current_converter = None # Clear converter reference | ||
| self._current_converter = None | ||
| return "end_turn" |
There was a problem hiding this comment.
The self._current_converter reference is not cleared when a UsageLimitExceeded exception is raised, which can lead to stale state or memory leaks. Rather than duplicating self._current_converter = None in multiple except and else blocks, it is much cleaner and safer to manage this cleanup in a finally block.
except asyncio.CancelledError:
self.log.info(
"Subagent stream cancelled via CancelledError, cleaning up tool calls"
)
async for cancel_update in converter.cancel_pending_tools():
await self.notifications.send_update(cancel_update)
await anyio.sleep(0.05)
return "cancelled"
except UsageLimitExceeded as e:
self.log.info("Usage limit exceeded", error=str(e))
return infer_stop_reason(str(e))
except Exception as e:
self.log.exception("Error during subagent streaming")
await self._send_toast(
message=f"Subagent error: {e}",
level="error",
)
await anyio.sleep(0.05)
return "end_turn"
else:
self.last_usage = converter.last_usage
return "end_turn"
finally:
self._current_converter = None| except asyncio.CancelledError: | ||
| # Task was cancelled (e.g., via interrupt()) - return proper stop reason | ||
| # This is critical: CancelledError doesn't inherit from Exception, | ||
| # so we must catch it explicitly to send the PromptResponse | ||
| self.log.info("Stream cancelled via CancelledError, cleaning up tool calls") | ||
| # Send cancellation notifications for any pending tool calls | ||
| async for cancel_update in converter.cancel_pending_tools(): | ||
| await self.notifications.send_update(cancel_update) | ||
| # CRITICAL: Allow time for client to process tool completion notifications | ||
| # before sending PromptResponse. See comment in cancellation branch above. | ||
| await anyio.sleep(0.05) | ||
| self._current_converter = None | ||
| return "cancelled" | ||
| except UsageLimitExceeded as e: | ||
| self.log.info("Usage limit exceeded", error=str(e)) | ||
| return infer_stop_reason(str(e)) | ||
| except Exception as e: | ||
| self._current_converter = None # Clear converter reference | ||
| self.log.exception("Error during streaming") | ||
| # Send error as toast notification instead of polluting chat history | ||
| await self._send_toast( | ||
| message=f"Agent error: {e}", | ||
| level="error", | ||
| ) | ||
| await anyio.sleep(0.05) # Allow network buffers to flush | ||
| return "end_turn" | ||
| else: | ||
| # Title generation is now handled automatically by log_session | ||
| self.last_usage = converter.last_usage | ||
| self._current_converter = None # Clear converter reference | ||
| return "end_turn" |
There was a problem hiding this comment.
Similar to _run_subagent_directly, self._current_converter is not cleared in _run_agent_stream when a UsageLimitExceeded exception is raised. Refactor this block to clear the converter reference inside a finally block to ensure it is always cleaned up regardless of how the stream exits.
except asyncio.CancelledError:
# Task was cancelled (e.g., via interrupt()) - return proper stop reason
# This is critical: CancelledError doesn't inherit from Exception,
# so we must catch it explicitly to send the PromptResponse
self.log.info("Stream cancelled via CancelledError, cleaning up tool calls")
# Send cancellation notifications for any pending tool calls
async for cancel_update in converter.cancel_pending_tools():
await self.notifications.send_update(cancel_update)
# CRITICAL: Allow time for client to process tool completion notifications
# before sending PromptResponse. See comment in cancellation branch above.
await anyio.sleep(0.05)
return "cancelled"
except UsageLimitExceeded as e:
self.log.info("Usage limit exceeded", error=str(e))
return infer_stop_reason(str(e))
except Exception as e:
self.log.exception("Error during streaming")
# Send error as toast notification instead of polluting chat history
await self._send_toast(
message=f"Agent error: {e}",
level="error",
)
await anyio.sleep(0.05) # Allow network buffers to flush
return "end_turn"
else:
# Title generation is now handled automatically by log_session
self.last_usage = converter.last_usage
return "end_turn"
finally:
self._current_converter = None
状态分析此 PR 部分功能已在主分支实现，但核心高级功能尚未实现，且 PR 已严重过期（CONFLICTING，最后更新 2026-07-01）。关闭此 PR 并创建新 Issue 追踪剩余工作。 已在主分支实现的
尚未实现的（需重新评估）
结论Phase 1 schema 层已通过 PR #65 大重构合并入主分支。Phase 2 高级功能需要基于当前架构重新评估实现方案。关闭此 PR，新 Issue 追踪剩余工作。 |
2 participants
Summary
Implement the ACP Subagents Protocol (RFC-0042) in AgentPool to standardize subagent discovery, delegation, and session hierarchy for external ACP clients.
Changes
Phase 1 — Core Protocol Surface
ACP Schema Extensions ()
"subagent"toToolCallKindLiteralSubagentRunInfo,SubagentInfo,SubagentCapabilitiesSessionInfo(parent_session_id, child_session_ids, depth)subagentfield toToolCallStart/ToolCallProgressavailable_subagentsto lifecycle responsesSpawnSessionStart — Add
run_modefield (Literal["foreground", "background"])SessionData Hierarchy — Typed property accessors for
parent_tool_call_idandsubagent_idToolKind Sync Guardrail — Cross-definition equality test
Phase 2 — Advanced Features
Event Converter — Emits
ToolCallStart(kind="subagent")for tool_box and inline modesForeground Cancellation — Propagates to child sessions, background children survive
Capability Advertisement — Static
availableSubagentsin all session lifecycle responsesDelegation Handler — auto/disable/prefer/require policies with capability gating
SubagentCatalogProvider — Debounced updates (500ms) with cycle detection
Dynamic Notifications —
available_subagents_updateemitted after catalog changesTesting
Verification
Related