refactor(adapters/langgraph): align ManagerWorkers code and tests with codebase style

pyagentspec/src/pyagentspec/adapters/langgraph/_execution_span.py

@@ -7,13 +7,11 @@
 """Wrap a compiled graph's ``stream``/``astream`` in an Agent Spec execution span.
 
 Agent, Flow and ManagerWorkers graphs all need the same wrapper: open a span, emit a
-Start event carrying the invocation inputs, replay the chunks the underlying stream
-yields while remembering the last state chunk, then emit an End event built from that
-final state. Only the span class and the two event payloads differ, so they come in
-as factories.
-
-``invoke``/``ainvoke`` need no patch of their own; they go through ``stream``/
-``astream`` internally.
+Start event carrying the invocation inputs, yield the chunks the underlying stream
+produces while remembering the last state chunk, then emit an End event built from
+that final state. Only the span class and the two event payloads differ, so they come
+in as factories. ``invoke``/``ainvoke`` need no patch; they use ``stream``/``astream``
+internally.
 """
 
 from typing import Any, AsyncGenerator, Callable, Dict, Generator

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

@@ -1092,21 +1092,15 @@ def _manager_workers_convert_to_langgraph(
                                     └─ no tool_call ─→ END
 
         The manager is a react-agent holding one synthetic ``delegate_to_<worker>``
-        tool per worker. The parent graph's conditional edge inspects its last
-        AIMessage to pick the next node, and the worker node runs in an isolated
-        message context and answers with a ``ToolMessage`` matched to the pending
-        delegation id.
-
-        Workers are converted recursively and wired in as subgraph nodes, so
-        ``astream_events`` still exposes the parent/child boundary
-        (``subgraph=True``) for tracing and SSE streaming. Workers that are
-        themselves ``ManagerWorkers`` compose through ``self.convert(...)``.
+        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
+        wired in as subgraph nodes, so ``astream_events`` still exposes the
+        parent/child boundary (``subgraph=True``) for tracing and streaming.
         """
         if not isinstance(mw.group_manager, AgentSpecAgent):
-            # The manager has to decide which worker to delegate to, so it needs a
-            # chat-LLM that emits tool_calls. Only Agent (and its SpecializedAgent
-            # subclass) has that shape; a Flow, Swarm or nested ManagerWorkers gives
-            # us nothing to route on.
+            # Delegation is routed off the manager's tool_calls, so the manager needs
+            # a chat-LLM; a Flow, Swarm or nested ManagerWorkers gives nothing to route on.
             raise NotImplementedError(
                 f"ManagerWorkers.group_manager must be an Agent for LangGraph "
                 f"conversion; got {type(mw.group_manager).__name__}."
@@ -1137,9 +1131,8 @@ def _manager_workers_convert_to_langgraph(
             [(node_name, worker.description or "") for node_name, worker in named_workers],
         )
 
-        # The delegation tools do execute inside the react loop: their body returns a
-        # Command(graph=PARENT), which is how the call escapes the react subgraph so
-        # the conditional edge below can route on it.
+        # The delegation tools execute inside the react loop: their Command(graph=PARENT)
+        # is how the call escapes the subgraph so the conditional edge below can route on it.
         manager_graph = self._create_react_agent_with_given_info(
             name=manager_agent.name,
             system_prompt=rendered_prompt,
@@ -1258,7 +1251,9 @@ def _create_react_agent_with_given_info(
             make_span=lambda: AgentSpecAgentExecutionSpan(
                 name=f"AgentExecution[{agent.name}]", agent=agent
             ),
-            make_start_event=lambda inputs: AgentSpecAgentExecutionStart(agent=agent, inputs=inputs),
+            make_start_event=lambda inputs: AgentSpecAgentExecutionStart(
+                agent=agent, inputs=inputs
+            ),
             make_end_event=lambda result: AgentSpecAgentExecutionEnd(
                 agent=agent,
                 outputs=extract_outputs_from_invoke_result(result, agent.outputs or []),

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

@@ -4,19 +4,15 @@
 # (LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0) or Universal Permissive License
 # (UPL) 1.0 (LICENSE-UPL or https://oss.oracle.com/licenses/upl), at your option.
 
-"""Helpers for compiling a ``ManagerWorkers`` into LangGraph.
+"""Helpers for compiling a ``ManagerWorkers`` into LangGraph, orchestrated by
+``AgentSpecToLangGraphConverter._manager_workers_convert_to_langgraph``.
 
-``AgentSpecToLangGraphConverter._manager_workers_convert_to_langgraph`` orchestrates
-these; they live here to keep the converter module from growing further.
-
-Nothing hides the routing protocol. ``delegate_to_<worker>`` calls stream like any
-other tool call, because which worker got which task is usually the most useful thing
-a run reports. Consumers that would rather not render it can filter on
+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 functools import lru_cache
 from typing import Any, Dict, List, Tuple
 
 from pyagentspec.adapters.langgraph._execution_span import patch_with_execution_span
@@ -35,13 +31,13 @@
 # Cannot collide with a normalized worker node name, which is always [a-z0-9_].
 _MANAGER_NODE_KEY = "__manager__"
 
-#: Prefix the manager's LLM uses to address a delegation tool, suffixed with the
-#: normalized worker node name. Public so consumers can recognize the protocol.
+#: 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_"
 
-# Carried on the per-delegation ``Send`` payload so a worker run knows its task and
-# which ``tool_call_id`` its reply must answer. Routing per delegation instead of off
-# shared state lets one manager turn delegate to several workers at once.
+# 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)
+# lets one manager turn delegate to several workers at once.
 _DELEGATE_TASK_KEY = "__delegate_task__"
 _DELEGATE_CALL_ID_KEY = "__delegate_tool_call_id__"
 
@@ -61,45 +57,21 @@ 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 sees ``delegate_to_<node_name>`` as a tool name and has to emit it
-    reliably, so node names stay ASCII identifiers. Falls back to the component id,
-    normalized the same way, when the name slugifies to nothing.
+    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.
     """
     return _normalize_identifier(name) or _normalize_identifier(fallback_id) or "worker"
 
 
-def _tc_get(tool_call: Any, key: str) -> Any:
-    """Read ``key`` off a tool call, which langchain emits as a dict or an object
-    depending on the message source."""
-    if isinstance(tool_call, dict):
-        return tool_call.get(key)
-    return getattr(tool_call, key, None)
-
-
 def _messages_of(state: Any) -> List[Any]:
     """Read ``messages`` off a state, which langgraph injects as a dict or an object."""
     if isinstance(state, dict):
         return list(state.get("messages") or [])
     return list(getattr(state, "messages", []) or [])
 
 
-def _surface_to_parent_command(state: Any) -> Any:
-    """Break out of the manager's react loop, projecting the subgraph's messages onto
-    the parent state (including the AIMessage carrying the triggering tool call).
-
-    Carries no ``goto``: routing is the parent graph's job. The ``add_messages``
-    reducer dedupes by id, so re-surfacing existing messages is a no-op. Modelled on
-    ``langgraph_swarm.create_handoff_tool``.
-    """
-    from langgraph.types import Command
-
-    return Command(graph=Command.PARENT, update={"messages": _messages_of(state)})
-
-
-def _append_workers_roster(
-    system_prompt: str,
-    entries: List[Tuple[str, str]],
-) -> str:
+def _append_workers_roster(system_prompt: str, entries: List[Tuple[str, str]]) -> str:
     """Append an ``Available workers:`` block listing ``- <name>: <description>``.
 
     Descriptions are flattened to one line each, since the LLM routes off the block's
@@ -114,19 +86,15 @@ def _append_workers_roster(
     return f"{system_prompt}\n\n{roster}" if system_prompt else roster
 
 
-@lru_cache(maxsize=256)
 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.
 
-    The body carries no ``goto``. Routing fans out one ``Send`` per delegation in
-    :func:`_route_manager_to_worker_or_end`; a ``goto`` here would collapse several
-    same-turn delegations into one parent Command and leave the other
-    ``tool_call_id``s unanswered.
-
-    Memoized because the tool depends only on the node name and holds no per-graph
-    state. Without it every compile re-runs the ``@tool`` decorator, which costs a
-    ``get_type_hints`` pass and a pydantic args-schema build.
+    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
+    would collapse several same-turn delegations into one parent Command and leave the
+    other ``tool_call_id``s unanswered.
     """
     from typing import Annotated
 
@@ -136,22 +104,21 @@ def _make_worker_delegation_tool(worker_node_name: str) -> Any:
 
     tool_name = f"{DELEGATE_TOOL_PREFIX}{worker_node_name}"
     description = (
-        f"Delegate a task to the {worker_node_name} worker and receive "
-        f"its response. Use this when the task fits the worker's "
-        f"described capability."
+        f"Delegate a task to the {worker_node_name} worker and receive its response. "
+        f"Use this when the task fits the worker's described capability."
     )
 
     @tool(tool_name, description=description)
     def _delegate(
         task: str,
         state: Annotated[Any, InjectedState],
         tool_call_id: Annotated[str, InjectedToolCallId],
-    ) -> Command:
-        # Declared for the LLM-facing schema but unused here: executing is only how the
-        # call escapes the react subgraph, and the routing edge recovers both off the
-        # surfaced AIMessage's tool_calls.
+    ) -> Any:
+        # task and tool_call_id are declared for the LLM-facing schema; the routing
+        # edge recovers both off the surfaced AIMessage's tool_calls. The
+        # add_messages reducer dedupes by id, so re-surfacing messages is a no-op.
         del task, tool_call_id
-        return _surface_to_parent_command(state)
+        return Command(graph=Command.PARENT, update={"messages": _messages_of(state)})
 
     return _delegate
 
@@ -160,42 +127,33 @@ 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.
 
-    Every delegation gets its own ``Send`` carrying the task and ``tool_call_id``, so
-    each 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.
+    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.
     """
     from langgraph.types import Send
 
     messages = state.get("messages") or []
-    if not messages:
-        return langgraph_graph.END
-    last = messages[-1]
-    tool_calls = getattr(last, "tool_calls", None) or []
+    last = messages[-1] if messages else None
     sends = []
-    for tc in tool_calls:
-        name = _tc_get(tc, "name")
+    for tool_call in getattr(last, "tool_calls", None) or []:
+        name = tool_call.get("name")
         if is_delegation_tool_name(name):
-            args = _tc_get(tc, "args") or {}
+            args = tool_call.get("args") or {}
             sends.append(
                 Send(
                     name[len(DELEGATE_TOOL_PREFIX) :],
                     {
                         _DELEGATE_TASK_KEY: args.get("task") or "",
-                        _DELEGATE_CALL_ID_KEY: _tc_get(tc, "id") or "",
+                        _DELEGATE_CALL_ID_KEY: tool_call.get("id") or "",
                     },
                 )
             )
     return sends or langgraph_graph.END
 
 
 def _worker_input(state: Dict[str, Any]) -> Dict[str, Any]:
-    """The single-message context a worker run starts from.
-
-    Passes no explicit config, so the worker inherits this node's ambient run config.
-    Its ``checkpoint_ns`` (``<worker_node>:<task_id>``) streams the worker's token
-    events under the worker node, and the per-superstep namespace keeps repeated
-    delegations isolated without a fresh thread_id.
-    """
+    """The single-message context a worker run starts from."""
     from langchain_core.messages import HumanMessage
 
     return {"messages": [HumanMessage(content=state.get(_DELEGATE_TASK_KEY) or "")]}
@@ -214,51 +172,34 @@ def _worker_reply(state: Dict[str, Any], result: Any) -> Dict[str, Any]:
     }
 
 
-class _WorkerSubgraphNode:
-    """Runs one worker subgraph as a node of the ManagerWorkers parent graph.
-
-    Hierarchical rather than shared-state like a Swarm: workers never see each other's
-    messages, and each run is handed only the manager's chosen task. The worker's last
-    message comes back as the ToolMessage content, so the manager's react loop sees a
-    well-formed tool response on its next turn.
-
-    A class rather than a closure, so the long-lived node holds only the graph instead
-    of keeping a whole factory frame alive.
-    """
-
-    __slots__ = ("_graph",)
-
-    def __init__(self, worker_graph: CompiledStateGraph[Any, Any, Any]) -> None:
-        self._graph = worker_graph
-
-    def run(self, state: Dict[str, Any]) -> Dict[str, Any]:
-        return _worker_reply(state, self._graph.invoke(_worker_input(state)))
-
-    async def arun(self, state: Dict[str, Any]) -> Dict[str, Any]:
-        return _worker_reply(state, await self._graph.ainvoke(_worker_input(state)))
-
-
 def _wrap_worker_for_subgraph(
     worker_graph: CompiledStateGraph[Any, Any, Any],
     worker_node_name: str,
 ) -> Any:
-    """Wrap a worker subgraph as a node exposing sync and async entrypoints.
+    """Wrap a worker subgraph as a node of the ManagerWorkers parent graph.
 
-    LangGraph picks between them depending on whether the parent graph was invoked
-    via ``invoke`` or ``ainvoke``.
+    Hierarchical rather than shared-state like a Swarm: each run is handed only the
+    manager's chosen task, and the worker's answer comes back as a ToolMessage so the
+    manager's react loop sees a well-formed tool response on its next turn. The worker
+    is invoked with no explicit config and inherits this node's ambient run config,
+    which streams its token events under the worker node's checkpoint namespace.
     """
     from pyagentspec.adapters.langgraph._types import RunnableLambda
 
-    node = _WorkerSubgraphNode(worker_graph)
-    return RunnableLambda(func=node.run, afunc=node.arun, name=f"worker:{worker_node_name}")
+    def run(state: Dict[str, Any]) -> Dict[str, Any]:
+        return _worker_reply(state, worker_graph.invoke(_worker_input(state)))
+
+    async def arun(state: Dict[str, Any]) -> Dict[str, Any]:
+        return _worker_reply(state, await worker_graph.ainvoke(_worker_input(state)))
+
+    return RunnableLambda(func=run, afunc=arun, name=f"worker:{worker_node_name}")
 
 
 def _patch_with_manager_workers_execution_span(
     compiled_graph: CompiledStateGraph[Any, Any, Any],
     mw: AgentSpecManagerWorkers,
 ) -> None:
-    """Wrap ``stream``/``astream`` so each run emits a ``ManagerWorkersExecutionSpan``,
-    using the same patcher as the Agent and Flow graphs."""
+    """Wrap ``stream``/``astream`` so each run emits a ``ManagerWorkersExecutionSpan``."""
     patch_with_execution_span(
         compiled_graph,
         make_span=lambda: AgentSpecManagerWorkersExecutionSpan(

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

@@ -543,20 +543,15 @@ def _create_manager_workers_with_given_input_values(
 
         The graph runs over ``MessagesState``, which can't carry structured inputs
         inward to the group manager, so the node inputs are rendered into its
-        ``system_prompt`` and the satisfied ports dropped from both the manager and
-        the component. That keeps declared and inferred ports equal for the
-        downstream span re-validation, and the graph runs on messages alone.
-
-        Cached by rendered prompt, the same key
-        :meth:`_create_react_agent_with_given_input_values` uses.
+        ``system_prompt`` and the satisfied ports dropped. Cached by rendered prompt,
+        the same key :meth:`_create_react_agent_with_given_input_values` uses.
         """
         from pyagentspec.adapters.langgraph._langgraphconverter import AgentSpecToLangGraphConverter
 
         converter = AgentSpecToLangGraphConverter()
         entry_agent = component.group_manager
         if not isinstance(entry_agent, AgentSpecAgent):
-            # Not routable. Nothing to render or cache, and the converter owns the
-            # error message for this case.
+            # Nothing to render or cache; the converter owns the error for this case.
             return converter._manager_workers_convert_to_langgraph(
                 component, **self._conversion_kwargs()
             )
@@ -609,6 +604,17 @@ def _format_agent_result(self, result: Dict[str, Any]) -> ExecuteOutput:
             ]
             return {}, NodeExecutionDetails(generated_messages=generated_messages)
 
+        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()
+
         outputs = extract_outputs_from_invoke_result(result, self.node.outputs or [])
         return outputs, NodeExecutionDetails()