feat(delegation): add delegate_tool_reply explicit delivery channel f…

agent/agent_runtime_helpers.py

@@ -80,6 +80,7 @@ def _ra():
 AGENT_RUNTIME_POST_HOOK_TOOL_NAMES = frozenset({
     "todo_list", "session_search", "memory", "clarify", "read_terminal", "desktop_preview",
     "drive_preview", "annotate_preview", "read_window_below", "setup_mcp", "gui_tour", "delegate_task",
+    "delegate_tool_reply",
 })
 
 _TRAJECTORY_SYSTEM_PROMPT = (

agent/inline_tool_executors.py

@@ -197,6 +197,11 @@ def _desktop_preview(agent, args: dict, ctx: InlineToolContext) -> Any:
         ("server", "server", ""), ("action", "action", "install"), ("reason", "reason", ""),
     ),
     "delegate_task": lambda agent, args, ctx: agent._dispatch_delegate_task(args),
+    # Registry dispatch has no executing-agent reference; both tool paths use this table.
+    "delegate_tool_reply": _tool(
+        "tools.delegate_tool_reply", "delegate_tool_reply", ("content", "content", ""),
+        parent_agent=lambda agent, ctx: agent,
+    ),
 }
 
 # ``invoke_tool`` (concurrent path) consults the memory manager right after these three

agent/tool_dispatch_helpers.py

@@ -25,7 +25,7 @@
 logger = logging.getLogger(__name__)
 
 # Interactive / user-facing tools never run concurrently: any of these in a batch is a barrier.
-_NEVER_PARALLEL_TOOLS = frozenset({"clarify", "manage_connections"})
+_NEVER_PARALLEL_TOOLS = frozenset({"clarify", "manage_connections", "delegate_tool_reply"})
 
 # Read-only tools with no shared mutable session state.
 _PARALLEL_SAFE_TOOLS = frozenset({

model_tools.py

@@ -324,6 +324,9 @@ def _select_tool_names(enabled_toolsets: Optional[List[str]], disabled_toolsets:
     else:
         from toolsets import get_all_toolsets
         for ts_name in get_all_toolsets():
+            # Internal delivery is explicitly granted by child construction, never by "all tools".
+            if ts_name == "delegation_reply":
+                continue
             tools.update(resolve_toolset(ts_name))
     # Disabled toolsets are always subtracted LAST, so a tool in a disabled
     # toolset is stripped even when a composite (hermes-cli) re-enables it.

tests/agent/test_run_agent.py

@@ -2492,6 +2492,7 @@ class TestAgentRuntimePostHookOwnershipSync:
         ("setup_mcp", {"server": "linear", "action": "install"}),
         ("gui_tour", {"action": "stop"}),
         ("delegate_task", {"goal": "Check the child path"}),
+        ("delegate_tool_reply", {"content": "Explicit child result"}),
     )
 
     @pytest.mark.parametrize(("tool_name", "tool_args"), _CASES)

tests/tools/test_delegate.py

@@ -1283,7 +1283,6 @@ def test_same_provider_shares_parent_pool(self):
 
     # --- Custom-endpoint identity resolution (issue #7833) ---
 
-
     @patch(
         "tools.delegate_tool._load_config",
         return_value={"inherit_mcp_toolsets": False},
@@ -1309,7 +1308,7 @@ def test_build_child_agent_strict_intersection_when_opted_out(self, mock_cfg):
 
         self.assertEqual(
             MockAgent.call_args[1]["enabled_toolsets"],
-            ["web", "browser"],
+            ["web", "browser", "delegation_reply"],
         )
 
 

tests/tools/test_delegate_output_schema.py

@@ -16,6 +16,8 @@
 import threading
 from unittest.mock import MagicMock, patch
 
+import pytest
+
 from tools.delegate_tool import (
     DELEGATE_TASK_SCHEMA,
     _run_single_child,
@@ -167,13 +169,19 @@ class _StubChild:
     def __init__(self, responses):
         self.responses = list(responses)
         self.calls: list = []
+        self._delegate_reply_chunks: list[str] = []
 
     def get_activity_summary(self):
         return {"api_call_count": 1, "max_iterations": 5, "current_tool": None}
 
     def run_conversation(self, user_message, task_id=None, **_kwargs):
         self.calls.append(user_message)
-        text = self.responses.pop(0)
+        response = self.responses.pop(0)
+        if isinstance(response, tuple):
+            text, reply_chunks = response
+            self._delegate_reply_chunks.extend(reply_chunks)
+        else:
+            text = response
         return {
             "final_response": text,
             "completed": True,
@@ -198,6 +206,67 @@ def _run(child):
 
 
 class TestRunSingleChildSchemaValidation:
+    def test_retry_without_explicit_call_uses_only_corrected_final_response(self):
+        child = _StubChild([
+            ("cleanup", ["not json"]),
+            ('{"city": "Oslo"}', []),
+        ])
+        child._delegate_output_schema = ADDRESS_SCHEMA
+        entry = _run(child)
+        assert entry["schema_valid"] is True
+        assert entry["summary"] == '{"city": "Oslo"}'
+        assert child._delegate_reply_chunks == []
+
+    def test_multiple_delivery_chunks_are_validated_as_one_document(self):
+        child = _StubChild([("cleanup", ['{"city":', '"Rome"}'])])
+        child._delegate_output_schema = ADDRESS_SCHEMA
+        entry = _run(child)
+        assert entry["schema_valid"] is True
+        assert json.loads(entry["summary"]) == {"city": "Rome"}
+        assert len(child.calls) == 1
+
+    @pytest.mark.parametrize("correction", ["", "   "])
+    def test_empty_explicit_retry_is_not_replaced_by_cleanup(self, correction):
+        child = _StubChild([
+            ("cleanup", ["rejected delivery"]),
+            ('{"city": "not the deliverable"}', [correction]),
+        ])
+        child._delegate_output_schema = ADDRESS_SCHEMA
+        entry = _run(child)
+        assert entry["schema_valid"] is False
+        assert entry["status"] == "failed"
+        assert entry["schema_retries"] == 1
+        assert child._delegate_reply_chunks == [correction]
+        assert "not the deliverable" not in entry["summary"]
+
+    def test_explicit_delivery_is_validated_before_trailing_prose(self):
+        child = _StubChild(
+            [("cleanup complete", ['{"city": "Rome"}'])]
+        )
+        child._delegate_output_schema = ADDRESS_SCHEMA
+
+        entry = _run(child)
+
+        assert entry["schema_valid"] is True
+        assert entry["summary"] == '{"city": "Rome"}'
+        assert len(child.calls) == 1
+
+    def test_schema_retry_replaces_rejected_delivery_attempt(self):
+        child = _StubChild(
+            [
+                ("first cleanup", ["not json"]),
+                ("retry cleanup", ['{"city": "Oslo"}']),
+            ]
+        )
+        child._delegate_output_schema = ADDRESS_SCHEMA
+
+        entry = _run(child)
+
+        assert entry["schema_valid"] is True
+        assert entry["schema_retries"] == 1
+        assert entry["summary"] == '{"city": "Oslo"}'
+        assert "not json" not in entry["summary"]
+
     def test_valid_first_try_no_retry(self):
         child = _StubChild(['{"city": "Berlin"}'])
         child._delegate_output_schema = ADDRESS_SCHEMA