feat(sdk/subagent): add confirmation policy (#2345)

Author: VascoSch92

openhands-sdk/openhands/sdk/subagent/schema.py

@@ -4,14 +4,18 @@
 
 import re
 from pathlib import Path
-from typing import Any, Final
+from typing import TYPE_CHECKING, Any, Final
 
 import frontmatter
 from pydantic import BaseModel, Field
 
 from openhands.sdk.hooks.config import HookConfig
 
 
+if TYPE_CHECKING:
+    from openhands.sdk.security.confirmation_policy import ConfirmationPolicyBase
+
+
 KNOWN_FIELDS: Final[set[str]] = {
     "name",
     "description",
@@ -22,6 +26,13 @@
     "max_iteration_per_run",
     "hooks",
     "profile_store_dir",
+    "permission_mode",
+}
+
+_VALID_PERMISSION_MODES: Final[set[str]] = {
+    "always_confirm",
+    "never_confirm",
+    "confirm_risky",
 }
 
 
@@ -79,6 +90,20 @@ def _extract_examples(description: str) -> list[str]:
     return [m.strip() for m in matches if m.strip()]
 
 
+def _extract_permission_mode(fm: dict[str, object]) -> str | None:
+    """Extract permission_mode from frontmatter, defaulting to None (inherit parent)."""
+    raw = fm.get("permission_mode")
+    if raw is None:
+        return None
+    value = str(raw).strip().lower()
+    if value not in _VALID_PERMISSION_MODES:
+        raise ValueError(
+            f"Invalid permission_mode '{raw}'. "
+            f"Must be one of: {', '.join(sorted(_VALID_PERMISSION_MODES))}"
+        )
+    return value
+
+
 def _extract_max_iteration_per_run(fm: dict[str, object]) -> int | None:
     """Extract max iterations per run from frontmatter file."""
     max_iter_raw = fm.get("max_iteration_per_run")
@@ -130,6 +155,13 @@ class AgentDefinition(BaseModel):
     hooks: HookConfig | None = Field(
         default=None, description="Hook configuration for this agent"
     )
+    permission_mode: str | None = Field(
+        default=None,
+        description="How the subagent handles permissions. "
+        "None inherits the parent policy, 'always_confirm' requires "
+        "confirmation for every action, 'never_confirm' skips all confirmations, "
+        "'confirm_risky' only confirms actions above a risk threshold.",
+    )
     max_iteration_per_run: int | None = Field(
         default=None,
         description="Maximum iterations per run. "
@@ -145,6 +177,34 @@ class AgentDefinition(BaseModel):
         default_factory=dict, description="Additional metadata from frontmatter"
     )
 
+    def get_confirmation_policy(self) -> ConfirmationPolicyBase | None:
+        """Convert permission_mode to a ConfirmationPolicyBase instance.
+
+        Returns None when permission_mode is None (inherit parent policy).
+        """
+        if self.permission_mode is None:
+            return None
+
+        match self.permission_mode:
+            case "always_confirm":
+                from openhands.sdk.security.confirmation_policy import AlwaysConfirm
+
+                return AlwaysConfirm()
+            case "never_confirm":
+                from openhands.sdk.security.confirmation_policy import NeverConfirm
+
+                return NeverConfirm()
+            case "confirm_risky":
+                from openhands.sdk.security.confirmation_policy import ConfirmRisky
+
+                return ConfirmRisky()
+            case _:
+                # Should never reach here due to validation
+                # in _extract_permission_mode()
+                raise AssertionError(
+                    f"Unexpected permission_mode: {self.permission_mode}"
+                )
+
     @classmethod
     def load(cls, agent_path: Path) -> AgentDefinition:
         """Load an agent definition from a Markdown file.
@@ -156,6 +216,8 @@ def load(cls, agent_path: Path) -> AgentDefinition:
         - skills (optional): Comma-separated skill names or list of skill names
         - model (optional): Model profile to use (default: 'inherit')
         - color (optional): Display color
+        - permission_mode (optional): How the subagent handles permissions
+          ('always_confirm', 'never_confirm', 'confirm_risky'). None inherits parent.
         - max_iterations_per_run: Max iteration per run
         - hooks (optional): List of applicable hooks
 
@@ -180,6 +242,7 @@ def load(cls, agent_path: Path) -> AgentDefinition:
         color: str | None = _extract_color(fm)
         tools: list[str] = _extract_tools(fm)
         skills: list[str] = _extract_skills(fm)
+        permission_mode: str | None = _extract_permission_mode(fm)
         max_iteration_per_run: int | None = _extract_max_iteration_per_run(fm)
         profile_store_dir: str | None = _extract_profile_store_dir(fm)
         hooks: HookConfig | None = _extract_hooks(fm)
@@ -197,6 +260,7 @@ def load(cls, agent_path: Path) -> AgentDefinition:
             color=color,
             tools=tools,
             skills=skills,
+            permission_mode=permission_mode,
             max_iteration_per_run=max_iteration_per_run,
             hooks=hooks,
             profile_store_dir=profile_store_dir,

openhands-tools/openhands/tools/delegate/__init__.py

@@ -5,11 +5,12 @@
     DelegateObservation,
     DelegateTool,
 )
-from openhands.tools.delegate.impl import DelegateExecutor
+from openhands.tools.delegate.impl import ConfirmationHandler, DelegateExecutor
 from openhands.tools.delegate.visualizer import DelegationVisualizer
 
 
 __all__ = [
+    "ConfirmationHandler",
     "DelegateAction",
     "DelegateObservation",
     "DelegateExecutor",

openhands-tools/openhands/tools/delegate/definition.py

@@ -18,6 +18,7 @@
 
 if TYPE_CHECKING:
     from openhands.sdk.conversation.state import ConversationState
+    from openhands.tools.delegate.impl import ConfirmationHandler
 
 
 PROMPT_DIR = pathlib.Path(__file__).parent / "templates"
@@ -67,12 +68,18 @@ def create(
         cls,
         conv_state: "ConversationState",
         max_children: int = 5,
+        confirmation_handler: "ConfirmationHandler | None" = None,
     ) -> Sequence["DelegateTool"]:
         """Initialize DelegateTool with a DelegateExecutor.
 
         Args:
             conv_state: Conversation state (used to get workspace location)
             max_children: Maximum number of concurrent sub-agents (default: 5)
+            confirmation_handler: Optional callback invoked when a sub-agent's
+                confirmation policy requires user approval.  Receives
+                `(agent_id, pending_actions)` and must return `True` to
+                approve or `False` to reject.  When `None`, pending actions
+                are auto-approved.
 
         Returns:
             List containing a single delegate tool definition
@@ -95,7 +102,10 @@ def create(
 
         # Initialize the executor without parent conversation
         # (will be set on first call)
-        executor = DelegateExecutor(max_children=max_children)
+        executor = DelegateExecutor(
+            max_children=max_children,
+            confirmation_handler=confirmation_handler,
+        )
 
         # Initialize the parent Tool with the executor
         return [

openhands-tools/openhands/tools/delegate/impl.py

@@ -1,21 +1,31 @@
 """Implementation of delegate tool executor."""
 
 import threading
+from collections.abc import Callable
 from typing import TYPE_CHECKING
 
 from openhands.sdk.conversation.impl.local_conversation import LocalConversation
 from openhands.sdk.conversation.response_utils import get_agent_final_response
+from openhands.sdk.conversation.state import (
+    ConversationExecutionStatus,
+    ConversationState,
+)
 from openhands.sdk.logger import get_logger
 from openhands.sdk.subagent import get_agent_factory
 from openhands.sdk.tool.tool import ToolExecutor
 from openhands.tools.delegate.definition import DelegateObservation
 
 
 if TYPE_CHECKING:
+    from openhands.sdk.event import ActionEvent
     from openhands.tools.delegate.definition import DelegateAction
 
 logger = get_logger(__name__)
 
+# Called when a sub-agent hits WAITING_FOR_CONFIRMATION.
+# Receives (agent_id, pending_actions) and returns True to approve, False to reject.
+ConfirmationHandler = Callable[[str, list["ActionEvent"]], bool]
+
 
 class DelegateExecutor(ToolExecutor):
     """Executor for delegation operations.
@@ -25,11 +35,16 @@ class DelegateExecutor(ToolExecutor):
     - Delegating tasks to sub-agents and waiting for results (blocking)
     """
 
-    def __init__(self, max_children: int = 5):
+    def __init__(
+        self,
+        max_children: int = 5,
+        confirmation_handler: ConfirmationHandler | None = None,
+    ):
         self._parent_conversation: LocalConversation | None = None
         # Map from user-friendly identifier to conversation
         self._sub_agents: dict[str, LocalConversation] = {}
         self._max_children: int = max_children
+        self._confirmation_handler = confirmation_handler
 
     @property
     def parent_conversation(self) -> LocalConversation:
@@ -79,6 +94,27 @@ def _resolve_agent_type(self, action: "DelegateAction", index: int) -> str:
             return "default"
         return action.agent_types[index].strip() or "default"
 
+    def _run_until_finished(
+        self, agent_id: str, conversation: LocalConversation
+    ) -> None:
+        """Run a sub-agent conversation to completion, handling confirmations."""
+        conversation.run()
+        while (
+            conversation.state.execution_status
+            == ConversationExecutionStatus.WAITING_FOR_CONFIRMATION
+        ):
+            pending = ConversationState.get_unmatched_actions(conversation.state.events)
+            if not pending:
+                break
+
+            if self._confirmation_handler is None or self._confirmation_handler(
+                agent_id, pending
+            ):
+                conversation.run()
+            else:
+                conversation.reject_pending_actions("User rejected the actions")
+                conversation.run()
+
     def _spawn_agents(self, action: "DelegateAction") -> DelegateObservation:
         """Spawn sub-agents with optional agent types."""
         if not action.ids:
@@ -159,6 +195,16 @@ def _spawn_agents(self, action: "DelegateAction") -> DelegateObservation:
 
                 sub_conversation = LocalConversation(**conv_kwargs)
 
+                # Apply permission_mode: explicit mode from definition,
+                # or inherit the parent's policy when None.
+                confirmation_policy = factory.definition.get_confirmation_policy()
+                if confirmation_policy is None:
+                    sub_conversation.set_confirmation_policy(
+                        parent_conversation.state.confirmation_policy
+                    )
+                else:
+                    sub_conversation.set_confirmation_policy(confirmation_policy)
+
                 self._sub_agents[agent_id] = sub_conversation
 
                 # Log what type of agent was created
@@ -242,11 +288,9 @@ def run_task(
                 """Run a single task on a sub-agent."""
                 try:
                     logger.info(f"Sub-agent {agent_id} starting task: {task[:100]}...")
-                    # Pass raw parent_name - visualizer handles formatting
                     conversation.send_message(task, sender=parent_name)
-                    conversation.run()
+                    self._run_until_finished(agent_id, conversation)
 
-                    # Extract the final response using get_agent_final_response
                     final_response = get_agent_final_response(conversation.state.events)
                     if final_response:
                         results[agent_id] = final_response

openhands-tools/openhands/tools/task/definition.py

@@ -29,6 +29,7 @@
 if TYPE_CHECKING:
     from openhands.sdk.conversation.state import ConversationState
     from openhands.tools.task.impl import TaskExecutor
+    from openhands.tools.task.manager import ConfirmationHandler
 
 
 class TaskAction(Action):
@@ -182,11 +183,16 @@ class TaskToolSet(ToolDefinition[TaskAction, TaskObservation]):
     def create(
         cls,
         conv_state: "ConversationState",  # noqa: ARG003
+        confirmation_handler: "ConfirmationHandler | None" = None,
     ) -> list[ToolDefinition]:
         """Create the task tool.
 
         Args:
             conv_state: Conversation state for workspace info.
+            confirmation_handler: Optional callback invoked when a sub-agent's
+                confirmation policy requires user approval.  Receives
+                `(task_id, pending_actions)` and must return `True` to
+                approve or `False` to reject.
 
         Returns:
             List containing a single TaskTool.
@@ -199,7 +205,7 @@ def create(
             agent_types_info=agent_types_info
         )
 
-        manager = TaskManager()
+        manager = TaskManager(confirmation_handler=confirmation_handler)
         task_executor = TaskExecutor(manager=manager)
 
         tools: list[ToolDefinition] = []