Address ManagerWorkers review feedback from oracle#211

pyagentspec/src/pyagentspec/adapters/langgraph/_langgraphconverter.py

@@ -38,9 +38,9 @@
 from pyagentspec.adapters.langgraph._managerworkers import (
     _MANAGER_NODE_KEY,
     _append_workers_roster,
+    _make_manager_router,
     _make_worker_delegation_tool,
     _patch_with_manager_workers_execution_span,
-    _route_manager_to_worker_or_end,
     _safe_node_name,
     _wrap_worker_for_subgraph,
 )
@@ -1085,13 +1085,13 @@ def _manager_workers_convert_to_langgraph(
 
         Topology::
 
-                            ┌─ delegate_to_w1 ─→ worker_1 ─┐
-            START → manager ┤                              ├→ manager (loop)
-                            └─ delegate_to_w2 ─→ worker_2 ─┘
+                            ┌─ __delegate_to__w1 ─→ worker_1 ─┐
+            START → manager ┤                                 ├→ manager (loop)
+                            └─ __delegate_to__w2 ─→ worker_2 ─┘
                                     │
                                     └─ no tool_call ─→ END
 
-        The manager is a react-agent holding one synthetic ``delegate_to_<worker>``
+        The manager is a react-agent holding one synthetic ``__delegate_to__<worker>``
         tool per worker. A conditional edge routes each delegation to its worker,
         which runs in an isolated message context and answers with a ``ToolMessage``
         matched to the pending delegation id. Workers are converted recursively and
@@ -1160,7 +1160,7 @@ def _manager_workers_convert_to_langgraph(
         builder.add_edge(langgraph_graph.START, _MANAGER_NODE_KEY)
         builder.add_conditional_edges(
             _MANAGER_NODE_KEY,
-            _route_manager_to_worker_or_end,
+            _make_manager_router(worker_node_names),
             # The path map covers every worker plus END, so langgraph can validate
             # the routing statically.
             {node_name: node_name for node_name in worker_node_names}

pyagentspec/src/pyagentspec/adapters/langgraph/_managerworkers.py

@@ -7,13 +7,13 @@
 """Helpers for compiling a ``ManagerWorkers`` into LangGraph, orchestrated by
 ``AgentSpecToLangGraphConverter._manager_workers_convert_to_langgraph``.
 
-The delegation protocol is visible on purpose: ``delegate_to_<worker>`` calls stream
-like any other tool call. Consumers that would rather not render them can filter on
-:func:`is_delegation_tool_name`.
+The delegation protocol is visible on purpose: ``__delegate_to__<worker>`` calls
+stream like any other tool call. Consumers that would rather not render them can
+filter on :func:`is_delegation_tool_name`.
 """
 
 import re
-from typing import Any, Dict, List, Tuple
+from typing import Any, Dict, Iterable, List, Tuple
 
 from pyagentspec.adapters.langgraph._execution_span import patch_with_execution_span
 from pyagentspec.adapters.langgraph._types import CompiledStateGraph, langgraph_graph
@@ -31,9 +31,11 @@
 # Cannot collide with a normalized worker node name, which is always [a-z0-9_].
 _MANAGER_NODE_KEY = "__manager__"
 
-#: Prefix of the synthetic ``delegate_to_<worker>`` tool names the manager's LLM uses
-#: to address a worker. Public so consumers can recognize the protocol.
-DELEGATE_TOOL_PREFIX = "delegate_to_"
+#: Prefix of the synthetic ``__delegate_to__<worker>`` tool names the manager's LLM
+#: uses to address a worker. The dunder prefix, like the delegation keys below, keeps
+#: it from colliding with a real tool named ``delegate_to_<something>``. Public so
+#: consumers can recognize the protocol.
+DELEGATE_TOOL_PREFIX = "__delegate_to__"
 
 # Keys of the per-delegation ``Send`` payload: the task to run, and the tool_call_id
 # the worker's reply must answer. Routing per delegation (instead of off shared state)
@@ -45,7 +47,7 @@
 
 
 def is_delegation_tool_name(name: Any) -> bool:
-    """True for the synthetic ``delegate_to_<worker>`` tool names a manager emits."""
+    """True for the synthetic ``__delegate_to__<worker>`` tool names a manager emits."""
     return isinstance(name, str) and name.startswith(DELEGATE_TOOL_PREFIX)
 
 
@@ -57,7 +59,7 @@ def _normalize_identifier(s: str) -> str:
 def _safe_node_name(name: str, fallback_id: str) -> str:
     """Normalize a worker name into a LangGraph node identifier.
 
-    The LLM has to emit ``delegate_to_<node_name>`` reliably as a tool name, so node
+    The LLM has to emit ``__delegate_to__<node_name>`` reliably as a tool name, so node
     names stay ASCII identifiers. Falls back to the normalized component id when the
     name slugifies to nothing.
     """
@@ -87,12 +89,12 @@ def _append_workers_roster(system_prompt: str, entries: List[Tuple[str, str]]) -
 
 
 def _make_worker_delegation_tool(worker_node_name: str) -> Any:
-    """Build the ``delegate_to_<worker>`` tool the manager's LLM emits to route to a
-    worker.
+    """Build the ``__delegate_to__<worker>`` tool the manager's LLM emits to route to
+    a worker.
 
     Executing the tool is only how the call escapes the react subgraph: its body
     surfaces the subgraph messages to the parent with ``Command(graph=PARENT)`` and no
-    ``goto``. Routing stays in :func:`_route_manager_to_worker_or_end`; a ``goto`` here
+    ``goto``. Routing stays in the edge built by :func:`_make_manager_router`; a ``goto`` here
     would collapse several same-turn delegations into one parent Command and leave the
     other ``tool_call_id``s unanswered.
     """
@@ -123,33 +125,45 @@ def _delegate(
     return _delegate
 
 
-def _route_manager_to_worker_or_end(state: Dict[str, Any]) -> Any:
-    """Route the parent graph off the manager's last AIMessage: one ``Send`` per
-    ``delegate_to_<worker>`` tool call, or ``END`` when it emitted none.
+def _make_manager_router(worker_node_names: Iterable[str]) -> Any:
+    """Build the conditional edge routing the parent graph off the manager's last
+    AIMessage: one ``Send`` per ``__delegate_to__<worker>`` tool call, or ``END``
+    when it emitted none.
 
     Every delegation gets its own ``Send``, so each tool_call_id is answered
     independently; an unanswered one breaks the manager's next-turn tool-call/result
-    sequence. Plain tool calls already ran inside the react loop.
+    sequence. Plain tool calls already ran inside the react loop; that includes a
+    real tool whose name merely starts with the prefix, which is why a suffix that
+    is not a worker node is not routed.
     """
-    from langgraph.types import Send
-
-    messages = state.get("messages") or []
-    last = messages[-1] if messages else None
-    sends = []
-    for tool_call in getattr(last, "tool_calls", None) or []:
-        name = tool_call.get("name")
-        if is_delegation_tool_name(name):
+    known_workers = frozenset(worker_node_names)
+
+    def _route_manager_to_worker_or_end(state: Dict[str, Any]) -> Any:
+        from langgraph.types import Send
+
+        messages = state.get("messages") or []
+        last = messages[-1] if messages else None
+        sends = []
+        for tool_call in getattr(last, "tool_calls", None) or []:
+            name = tool_call.get("name")
+            if not is_delegation_tool_name(name):
+                continue
+            worker_node_name = name[len(DELEGATE_TOOL_PREFIX) :]
+            if worker_node_name not in known_workers:
+                continue
             args = tool_call.get("args") or {}
             sends.append(
                 Send(
-                    name[len(DELEGATE_TOOL_PREFIX) :],
+                    worker_node_name,
                     {
                         _DELEGATE_TASK_KEY: args.get("task") or "",
                         _DELEGATE_CALL_ID_KEY: tool_call.get("id") or "",
                     },
                 )
             )
-    return sends or langgraph_graph.END
+        return sends or langgraph_graph.END
+
+    return _route_manager_to_worker_or_end
 
 
 def _worker_input(state: Dict[str, Any]) -> Dict[str, Any]:

pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py

@@ -493,6 +493,17 @@ def __init__(
         super().__init__(node)
         if not isinstance(self.node, AgentSpecAgentNode):
             raise TypeError("AgentNodeExecutor can only be initialized with AgentNode")
+        if isinstance(self.node.agent, AgentSpecManagerWorkers):
+            # The hierarchical graph runs over MessagesState, which cannot carry a
+            # structured_response outward: the manager's final message is the only
+            # result, so anything but a single string output cannot be honored.
+            # Raising here fails at conversion time rather than mid-run.
+            outputs = self.node.outputs or []
+            if outputs and (len(outputs) != 1 or outputs[0].type != "string"):
+                raise NotImplementedError(
+                    "A ManagerWorkers flow step supports a single string output; "
+                    f"node `{self.node.name}` declares {[o.title for o in outputs]}."
+                )
         self.tool_registry = tool_registry
         self.checkpointer = checkpointer
         self.converted_components = converted_components
@@ -607,13 +618,9 @@ def _format_agent_result(self, result: Dict[str, Any]) -> ExecuteOutput:
         if isinstance(self.node.agent, AgentSpecManagerWorkers):
             # The hierarchical graph runs over MessagesState, which cannot carry a
             # structured_response outward: the manager's final message is the result.
-            outputs = self.node.outputs
-            if len(outputs) != 1 or outputs[0].type != "string":
-                raise NotImplementedError(
-                    "A ManagerWorkers flow step supports a single string output; "
-                    f"node `{self.node.name}` declares {[o.title for o in outputs]}."
-                )
-            return {outputs[0].title: result["messages"][-1].content}, NodeExecutionDetails()
+            # __init__ already rejected any shape but a single string output.
+            node_outputs = self.node.outputs or []
+            return {node_outputs[0].title: result["messages"][-1].content}, NodeExecutionDetails()
 
         outputs = extract_outputs_from_invoke_result(result, self.node.outputs or [])
         return outputs, NodeExecutionDetails()

pyagentspec/src/pyagentspec/managerworkers.py

@@ -67,10 +67,11 @@ class ManagerWorkers(AgenticComponent):
     )
 
     def _get_inferred_inputs(self) -> List[Property]:
-        # The group manager drives the conversation and is the component whose prompt
-        # the runtime renders, so the group accepts exactly the inputs the manager
-        # accepts. The hasattr guard matches Flow._get_inferred_inputs: validators can
-        # run this against a partially-constructed model with no group_manager yet.
+        # Per the language spec, the inputs of a ManagerWorkers are the inputs of its
+        # group manager (same name and type): the manager drives the conversation and
+        # is the component whose prompt the runtime renders. The hasattr guard matches
+        # Flow._get_inferred_inputs: validators can run this against a
+        # partially-constructed model with no group_manager yet.
         return (self.group_manager.inputs or []) if hasattr(self, "group_manager") else []
 
     def _get_inferred_outputs(self) -> List[Property]:
@@ -91,3 +92,26 @@ def _validate_group_manager_is_not_included_as_a_worker(self) -> Self:
         if any(self.group_manager is agent for agent in self.workers):
             raise ValueError("Group manager cannot be a worker.")
         return self
+
+    @model_validator_with_error_accumulation
+    def _validate_ios_match_group_manager_ios(self) -> Self:
+        # Per the language spec, the I/Os of a ManagerWorkers must be the I/Os of its
+        # group manager, same name and type. The base ComponentWithIO validators
+        # already enforce matching titles; enforce matching types here.
+        if not hasattr(self, "group_manager"):
+            return self
+        for kind, own_properties, manager_properties in (
+            ("input", self.inputs or [], self.group_manager.inputs or []),
+            ("output", self.outputs or [], self.group_manager.outputs or []),
+        ):
+            manager_type_by_title = {p.title: p.type for p in manager_properties}
+            for own_property in own_properties:
+                manager_type = manager_type_by_title.get(own_property.title)
+                if manager_type is not None and own_property.type != manager_type:
+                    raise ValueError(
+                        f"The {kind}s of a `ManagerWorkers` must match the {kind}s of its "
+                        f"group manager (same name and type), but {kind} "
+                        f"`{own_property.title}` has type `{own_property.type}` while the "
+                        f"group manager declares `{manager_type}`."
+                    )
+        return self

pyagentspec/tests/adapters/conftest.py

@@ -6,8 +6,6 @@
 
 import os
 import ssl
-from contextlib import ExitStack
-from importlib import import_module
 from pathlib import Path
 from typing import Any
 from unittest.mock import patch
@@ -114,34 +112,6 @@ def _skip(*_args, **_kwargs):
             p.stop()
 
 
-def _resolve(dotted: str) -> Any:
-    module_path, _, attr = dotted.rpartition(".")
-    module_path, _, cls_name = module_path.rpartition(".")
-    return getattr(getattr(import_module(module_path), cls_name), attr)
-
-
-# Captured at conftest import, before any session fixture starts patching, so these
-# are the genuine constructors rather than a skip stub.
-_REAL_LLM_INITS = {dotted: _resolve(dotted) for dotted in LLM_MOCKED_METHODS}
-
-
-@pytest.fixture
-def allow_llm_config_construction():
-    """
-    Opt out of the SKIP_LLM_TESTS=1 construction guard for one test, restoring the
-    real LLM config constructors. Only request this from a test that stubs the model
-    and never reaches an endpoint; such tests should run offline instead of skipping.
-    """
-    if not should_skip_llm_test():
-        # Nothing patched the constructors, so there is nothing to restore.
-        yield
-        return
-    with ExitStack() as stack:
-        for dotted, real in _REAL_LLM_INITS.items():
-            stack.enter_context(patch(dotted, new=real))
-        yield
-
-
 @pytest.fixture(scope="package")
 def json_server(json_server_port: int):
     api_server = Path(__file__).parent / "api_server.py"