From a6f62a2a175798f3033029b296668b9138b20537 Mon Sep 17 00:00:00 2001 From: Salah Pichen Date: Tue, 30 Jun 2026 13:17:29 +0400 Subject: [PATCH] feat(adapters/langgraph): support ManagerWorkers and Swarm orchestration Compile a ManagerWorkers spec into a hierarchical LangGraph (manager react-agent + worker subgraphs wired via synthesized delegate_to_ tools), run a ManagerWorkers as a flow step, and let a ManagerWorkers participate as a Swarm member (handoff re-emitted at the MW parent boundary so Command(graph=PARENT) jumps reach the Swarm). Includes the supporting middleware plumbing: forward LangChain agent middleware from the AgentSpecLoader/converter through create_agent and every nested conversion path (agents, flow nodes, MW, swarm). Delegation protocol is hidden from astream_events, worker events stream under the worker node, every delegation in a turn is answered, and delegate tool-calls are stripped from state snapshots. --- .../adapters/langgraph/_langgraphconverter.py | 312 +++- .../adapters/langgraph/_managerworkers.py | 819 +++++++++ .../adapters/langgraph/_node_execution.py | 111 +- pyagentspec/src/pyagentspec/managerworkers.py | 19 + pyagentspec/src/pyagentspec/swarm.py | 19 + .../langgraph/flows/test_agentnode.py | 91 + .../flows/test_managerworkers_node.py | 157 ++ .../langgraph/flows/test_swarm_node.py | 137 ++ .../adapters/langgraph/test_managerworkers.py | 1562 +++++++++++++++++ 9 files changed, 3178 insertions(+), 49 deletions(-) create mode 100644 pyagentspec/src/pyagentspec/adapters/langgraph/_managerworkers.py create mode 100644 pyagentspec/tests/adapters/langgraph/flows/test_managerworkers_node.py create mode 100644 pyagentspec/tests/adapters/langgraph/flows/test_swarm_node.py create mode 100644 pyagentspec/tests/adapters/langgraph/test_managerworkers.py diff --git a/pyagentspec/src/pyagentspec/adapters/langgraph/_langgraphconverter.py b/pyagentspec/src/pyagentspec/adapters/langgraph/_langgraphconverter.py index 7393e9c8..80c7d718 100644 --- a/pyagentspec/src/pyagentspec/adapters/langgraph/_langgraphconverter.py +++ b/pyagentspec/src/pyagentspec/adapters/langgraph/_langgraphconverter.py @@ -36,9 +36,23 @@ _build_type_from_schema, create_pydantic_model_from_properties, ) +from pyagentspec.adapters.langgraph._managerworkers import ( + _HANDOFF_NODE_KEY, + _MANAGER_NODE_KEY, + _append_workers_roster, + _make_handoff_forward_node, + _make_swarm_handoff_tool, + _make_worker_delegation_tool, + _patch_hide_delegation_in_astream_events, + _patch_with_manager_workers_execution_span, + _route_manager_to_worker_handoff_or_end, + _safe_node_name, + _wrap_worker_for_subgraph, +) from pyagentspec.adapters.langgraph._node_execution import ( NodeExecutor, extract_outputs_from_invoke_result, + is_single_string_output, ) from pyagentspec.adapters.langgraph._types import ( AgentState, @@ -102,6 +116,7 @@ ) from pyagentspec.llms.openaiconfig import OpenAiConfig from pyagentspec.llms.vllmconfig import VllmConfig +from pyagentspec.managerworkers import ManagerWorkers as AgentSpecManagerWorkers from pyagentspec.mcp.clienttransport import ClientTransport as AgentSpecClientTransport from pyagentspec.mcp.clienttransport import SSEmTLSTransport as AgentSpecSSEmTLSTransport from pyagentspec.mcp.clienttransport import SSETransport as AgentSpecSSETransport @@ -273,6 +288,15 @@ def _convert( config=config, middleware=middleware, ) + elif isinstance(agentspec_component, AgentSpecManagerWorkers): + return self._manager_workers_convert_to_langgraph( + agentspec_component, + tool_registry=tool_registry, + converted_components=converted_components, + checkpointer=checkpointer, + config=config, + middleware=middleware, + ) elif isinstance(agentspec_component, AgentSpecLlmConfig): return self._llm_convert_to_langgraph(agentspec_component, config=config) elif isinstance(agentspec_component, AgentSpecClientTransport): @@ -1066,61 +1090,251 @@ def _swarm_convert_to_langgraph( raise ValueError( "Handoff mode NEVER is not supported for conversion in LangGraph adapter" ) - agents: dict[str, AgentSpecAgent] = { - # LangGraph distinguishes agents by name, so we use names here. - # We also assume to get only agents in relationships. - agent.name: cast(AgentSpecAgent, agent) - # Relationships are tuples of (from_agent, to_agent) - for agent in (e for r in agentspec_component.relationships for e in r) + members: Dict[str, AgentSpecComponent] = { + # LangGraph distinguishes members by name, so we key by name. + member.name: member + # Relationships are tuples of (from_member, to_member) + for member in (e for r in agentspec_component.relationships for e in r) } - for agent in agents.values(): - # Since handoff is performed with tools, we can only support agents in relationships for now - # Note that the fact that we called `cast` before does not change the actual type of the agent - if not isinstance(agent, AgentSpecAgent): - raise ValueError( - f"Only Agents are supported as part of a Swarm in the LangGraph adapter, received {type(agent)} instead." + for member in members.values(): + # Swarm handoff is driven by an LLM emitting a transfer tool-call, + # so a member must run a chat loop the handoff tools can attach to: + # a plain Agent, or a ManagerWorkers (whose group-manager Agent + # makes the decision). A Flow / nested Swarm has no single + # "decide-the-handoff" LLM, so it cannot participate yet. + if not isinstance(member, (AgentSpecAgent, AgentSpecManagerWorkers)): + raise NotImplementedError( + "Only Agent and ManagerWorkers members are supported as part " + f"of a Swarm in the LangGraph adapter, received " + f"{type(member).__name__} instead." ) - # We convert the agents event though we do not use them in langgraph, since we have to append - # the handoff tools, but at least this way the agents will be created and stored in the registry - # of converted components in case they are used in other places - self.convert( - agent, - tool_registry=tool_registry, - converted_components=converted_components, - checkpointer=checkpointer, - config=config, - middleware=middleware, + handoffs: Dict[str, List[str]] = {name: [] for name in members} + for from_member, to_member in agentspec_component.relationships: + handoffs[from_member.name].append(to_member.name) + # We re-create each member equipped with the handoff tools that let it + # transfer the conversation to its related members. + langgraph_members: List[CompiledStateGraph[Any, Any, Any]] = [] + for member in members.values(): + destinations = handoffs.get(member.name, []) + if isinstance(member, AgentSpecManagerWorkers): + # A ManagerWorkers member gets its handoff tools wired onto its + # group-manager and re-emitted to the swarm — see + # `_manager_workers_convert_to_langgraph`. + langgraph_members.append( + self._manager_workers_convert_to_langgraph( + member, + tool_registry=tool_registry, + converted_components=converted_components, + checkpointer=checkpointer, + config=config, + middleware=middleware, + swarm_handoff_destinations=destinations, + ) + ) + continue + agent = cast(AgentSpecAgent, member) + langgraph_members.append( + self._create_react_agent_with_given_info( + agent=agent, + name=agent.name, + system_prompt=agent.system_prompt, + llm_config=agent.llm_config, + tools=agent.tools, + toolboxes=agent.toolboxes, + inputs=agent.inputs or [], + outputs=agent.outputs or [], + tool_registry=tool_registry, + converted_components=converted_components, + checkpointer=checkpointer, + config=config, + middleware=middleware, + additional_langgraph_tools=[ + langgraph_swarm.create_handoff_tool(agent_name=to_name) + for to_name in destinations + ], + ) + ) + return langgraph_swarm.create_swarm( + agents=langgraph_members, # type: ignore + default_active_agent=agentspec_component.first_agent.name, + ).compile(name=agentspec_component.name, checkpointer=checkpointer) + + def _manager_workers_convert_to_langgraph( + self, + mw: AgentSpecManagerWorkers, + tool_registry: Dict[str, "LangGraphTool"], + converted_components: Dict[str, Any], + checkpointer: Optional[Checkpointer], + config: RunnableConfig, + middleware: List[Any], + swarm_handoff_destinations: Optional[List[str]] = None, + ) -> CompiledStateGraph[Any, Any, Any]: + """Compile a ``ManagerWorkers`` into a hierarchical LangGraph. + + Topology:: + + ┌─ delegate_to_w1 ─→ worker_1 ─┐ + START → manager ┤ ├→ manager (loop) + └─ delegate_to_w2 ─→ worker_2 ─┘ + │ + └─ no tool_call ─→ END + + Each worker is recursively converted into a ``CompiledStateGraph`` + and wired in as a *subgraph node*, so ``astream_events`` exposes + the parent/child boundary (``subgraph=True``) for tracing and SSE + streaming. The manager is a react-agent given one synthetic + ``delegate_to_`` tool per worker; the parent graph's + conditional edge inspects the manager's last AIMessage to choose + the next node, then the worker node runs in an isolated message + context and emits a ``ToolMessage`` matched to the pending + delegation tool-call id. Recursive ``ManagerWorkers`` (workers + that are themselves ``ManagerWorkers``) compose for free through + ``self.convert(...)``. + + ``swarm_handoff_destinations`` is set only when this ManagerWorkers is + a member of a ``Swarm``: it lists the sibling member names this MW may + hand the conversation off to. Each becomes a synthetic + ``transfer_to_`` tool on the manager, plus a ``__handoff__`` + parent node that re-emits the handoff up to the Swarm (the manager's + own ``Command(graph=PARENT)`` would stop one level short, at this + ManagerWorkers graph). When ``None``/empty (the standalone, Flow-step, + or nested-worker case) no handoff machinery is added and behaviour is + unchanged. + """ + if not isinstance(mw.group_manager, AgentSpecAgent): + # Pyagentspec allows any AgenticComponent as group_manager, + # but the manager has to *decide* which worker to delegate to, + # which means it needs a chat-LLM that emits tool_calls. Today + # only Agent (and SpecializedAgent, a subclass) does that — a + # Flow / Swarm / nested ManagerWorkers as the group_manager + # doesn't have a "tool-call to delegate" output shape we can + # route on. + raise NotImplementedError( + f"ManagerWorkers.group_manager must be an Agent for LangGraph " + f"conversion; got {type(mw.group_manager).__name__}." + ) + + worker_node_names: List[str] = [ + _safe_node_name(worker.name, fallback_id=worker.id) for worker in mw.workers + ] + if len(set(worker_node_names)) != len(worker_node_names): + raise ValueError( + "ManagerWorkers worker names collide after normalization: " + f"{worker_node_names}. Give each worker a unique name." ) - handoffs: dict[str, list[str]] = {agent_name: [] for agent_name in agents} - for from_agent, to_agent in agentspec_component.relationships: - handoffs[from_agent.name].append(to_agent.name) - # We re-create the agents with the additional handoff tools - langgraph_agents: list[CompiledStateGraph[Any, Any, Any]] = [ - self._create_react_agent_with_given_info( - agent=agent, - name=agent.name, - system_prompt=agent.system_prompt, - llm_config=agent.llm_config, - tools=agent.tools, - toolboxes=agent.toolboxes, - inputs=agent.inputs or [], - outputs=agent.outputs or [], + + # 1. Recursively compile each worker as its own CompiledStateGraph. + worker_graphs: Dict[str, CompiledStateGraph[Any, Any, Any]] = {} + for worker, node_name in zip(mw.workers, worker_node_names): + worker_graphs[node_name] = self.convert( + worker, tool_registry=tool_registry, converted_components=converted_components, checkpointer=checkpointer, config=config, middleware=middleware, - additional_langgraph_tools=[ - langgraph_swarm.create_handoff_tool(agent_name=to_agent_name) - for to_agent_name in handoffs.get(agent.name, []) - ], ) - for agent in agents.values() + + # 2. Render the workers roster into the manager's system prompt + # so the LLM knows which delegation tool maps to which worker. + manager_agent = mw.group_manager + rendered_prompt = _append_workers_roster( + manager_agent.system_prompt, + [ + (node_name, worker.description or "") + for worker, node_name in zip(mw.workers, worker_node_names) + ], + ) + + # 3. Synthesize one delegation tool per worker. The tool body is a + # placeholder — the parent graph intercepts the manager's tool + # call before it executes and routes to the worker node. + delegation_tools: List[Any] = [ + _make_worker_delegation_tool(node_name) for node_name in worker_node_names ] - return langgraph_swarm.create_swarm( - agents=langgraph_agents, # type: ignore - default_active_agent=agentspec_component.first_agent.name, - ).compile(name=agentspec_component.name, checkpointer=checkpointer) + + # 3b. When this ManagerWorkers is a Swarm member, synthesize one + # transfer_to_ tool per allowed handoff destination. Like + # the delegation tools these are placeholders: the manager's LLM + # emits the call, the parent graph intercepts it (routing edge + # below) and the `__handoff__` node re-emits the handoff to the + # Swarm. We map the (normalized) tool name back to the raw sibling + # name, which is the Swarm node name used as the handoff `goto`. + handoff_dest_by_tool_name: Dict[str, str] = {} + handoff_tools: List[Any] = [] + for dest in swarm_handoff_destinations or []: + handoff_tool = _make_swarm_handoff_tool(dest) + handoff_tools.append(handoff_tool) + handoff_dest_by_tool_name[handoff_tool.name] = dest + + # 4. Compile the manager as a react-agent with the delegation tools + # (and, for a Swarm member, the transfer tools). + manager_graph = self._create_react_agent_with_given_info( + name=manager_agent.name, + system_prompt=rendered_prompt, + agent=manager_agent, + llm_config=manager_agent.llm_config, + tools=manager_agent.tools, + toolboxes=manager_agent.toolboxes, + inputs=manager_agent.inputs or [], + outputs=manager_agent.outputs or [], + tool_registry=tool_registry, + converted_components=converted_components, + checkpointer=checkpointer, + config=config, + middleware=middleware, + additional_langgraph_tools=delegation_tools + handoff_tools, + ) + + # 5. Compose the parent StateGraph. The manager and every worker + # are CompiledStateGraphs added as subgraph nodes; LangGraph's + # streaming surfaces them with ``subgraph=True``. + from langgraph.graph import MessagesState # local: optional dep + + manager_node_key = _MANAGER_NODE_KEY + builder = StateGraph(MessagesState) + builder.add_node(manager_node_key, manager_graph) + for node_name, worker_graph in worker_graphs.items(): + builder.add_node( + node_name, + _wrap_worker_for_subgraph(worker_graph, node_name), + ) + + # Path-map covers delegate-to-worker and the END branch so langgraph + # can statically validate the routing; the Swarm-handoff branch is + # added only when this MW is a swarm member with handoff destinations. + routing_path_map: Dict[str, str] = {node_name: node_name for node_name in worker_node_names} + routing_path_map[langgraph_graph.END] = langgraph_graph.END + if handoff_dest_by_tool_name: + builder.add_node( + _HANDOFF_NODE_KEY, + _make_handoff_forward_node(handoff_dest_by_tool_name), + ) + routing_path_map[_HANDOFF_NODE_KEY] = _HANDOFF_NODE_KEY + + builder.add_edge(langgraph_graph.START, manager_node_key) + builder.add_conditional_edges( + manager_node_key, + _route_manager_to_worker_handoff_or_end, + routing_path_map, + ) + for node_name in worker_node_names: + builder.add_edge(node_name, manager_node_key) + # The `__handoff__` node has no outgoing edge on purpose: it returns a + # Command(goto=, graph=PARENT) that exits this graph into the + # Swarm, so looping it back to the manager would be wrong. + + compiled_graph = builder.compile(checkpointer=checkpointer, name=mw.name) + + # 6. Tracing — wrap stream/astream so ManagerWorkersExecutionSpan + # surrounds each run. Mirrors the patches applied to Agent and + # Flow graphs above. + _patch_with_manager_workers_execution_span(compiled_graph, mw) + # Hide the delegate_to_ routing protocol from the + # astream_events view (tool calls, their tool lifecycle events, and + # the worker's synthetic reply ToolMessage) without touching state. + _patch_hide_delegation_in_astream_events(compiled_graph) + return compiled_graph def _create_react_agent_with_given_info( self, @@ -1174,8 +1388,12 @@ def _create_react_agent_with_given_info( output_model: Optional[type[BaseModel]] = None state_schema: Optional[Any] = None - # Build response (output) model (used for response_format) - if outputs: + # Build response (output) model (used for response_format). A single + # string output is taken from the agent's final message (see + # extract_outputs_from_invoke_result), so it needs no structured + # generation — mirrors LlmNodeExecutor and lets a string output work on + # models without structured-output support. + if outputs and not is_single_string_output(outputs): output_model = create_pydantic_model_from_properties("AgentOutputModel", outputs) if inputs: diff --git a/pyagentspec/src/pyagentspec/adapters/langgraph/_managerworkers.py b/pyagentspec/src/pyagentspec/adapters/langgraph/_managerworkers.py new file mode 100644 index 00000000..21c40478 --- /dev/null +++ b/pyagentspec/src/pyagentspec/adapters/langgraph/_managerworkers.py @@ -0,0 +1,819 @@ +# Copyright © 2025, 2026 Oracle and/or its affiliates. +# +# This software is under the Apache License 2.0 +# (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. + +"""ManagerWorkers / Swarm LangGraph compilation helpers. + +Module-level building blocks for compiling a ``ManagerWorkers`` (and a +``ManagerWorkers`` acting as a ``Swarm`` member) into LangGraph. The +``AgentSpecToLangGraphConverter`` methods ``_manager_workers_convert_to_langgraph`` +and ``_swarm_convert_to_langgraph`` orchestrate these helpers; the helpers +themselves are pure functions with no dependency on the converter, which is +why they live here rather than bloating the converter module. +""" + +import logging +import re +from typing import Any, AsyncGenerator, Dict, Generator, List, Optional, Tuple + +from pyagentspec.adapters.langgraph._types import CompiledStateGraph, langgraph_graph +from pyagentspec.managerworkers import ManagerWorkers as AgentSpecManagerWorkers +from pyagentspec.tracing.events import ( + ManagerWorkersExecutionEnd as AgentSpecManagerWorkersExecutionEnd, +) +from pyagentspec.tracing.events import ( + ManagerWorkersExecutionStart as AgentSpecManagerWorkersExecutionStart, +) +from pyagentspec.tracing.spans import ( + ManagerWorkersExecutionSpan as AgentSpecManagerWorkersExecutionSpan, +) + +# ─── ManagerWorkers helpers ────────────────────────────────────────────────── + +# Node key for the manager subgraph in the ManagerWorkers parent StateGraph. +# Chosen so it cannot collide with a normalized worker node name (which is +# always lowercase + [a-z0-9_]). +_MANAGER_NODE_KEY = "__manager__" + +# Prefix the manager's LLM uses to address a delegation tool. The suffix is +# the normalized worker node name. +_DELEGATE_TOOL_PREFIX = "delegate_to_" + +# Prefix the manager's LLM uses to hand the conversation off to a sibling +# Swarm member (only present when this ManagerWorkers is a Swarm member). The +# suffix is the normalized sibling name; matches langgraph_swarm's convention. +_HANDOFF_TOOL_PREFIX = "transfer_to_" + +# Node key for the Swarm-handoff node in the ManagerWorkers parent StateGraph. +# Like ``_MANAGER_NODE_KEY`` it cannot collide with a normalized worker node +# name (which never contains leading/trailing underscores). +_HANDOFF_NODE_KEY = "__handoff__" + +# Keys carried on the per-delegation ``Send`` payload from the manager's +# routing edge to a worker node, so a worker run knows which task it was +# given and which ``tool_call_id`` its reply ToolMessage must answer. This +# is what lets one manager turn delegate to several workers at once: each +# delegation routes as its own ``Send`` and is answered independently. +_DELEGATE_TASK_KEY = "__delegate_task__" +_DELEGATE_CALL_ID_KEY = "__delegate_tool_call_id__" + +# Collapses any run of whitespace to a single space so multi-line worker +# descriptions stay on one roster line. +_WHITESPACE_RE = re.compile(r"\s+") + + +def _normalize_identifier(s: str) -> str: + """Lowercase, collapse non-alphanumerics to underscores, strip surrounding + underscores. The single source of truth for turning a spec name into an + ASCII identifier — worker node names and ``transfer_to_`` tool + names must normalize identically, since a handoff ``goto`` is matched + against the normalized node name.""" + return re.sub(r"[^a-z0-9]+", "_", (s or "").lower()).strip("_") + + +def _safe_node_name(name: str, fallback_id: str) -> str: + """Normalize a worker name into a LangGraph node identifier. + + LangGraph node names must be hashable strings; in practice we want + ASCII-friendly identifiers that also work as Python attribute-ish + names (the LLM is going to see ``delegate_to_`` as a tool + name and needs to be able to emit it reliably). We normalize via + :func:`_normalize_identifier`, and fall back to the (component) id — + normalized the same way — if the name yields an empty string. Falling + through both transforms keeps node names internally consistent + regardless of which input wins. + """ + 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 that may be a dict or a pydantic-style + object (langchain emits either 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 the ``messages`` list off a state that may be a dict or an + attribute-bearing object (langgraph injects either into a tool).""" + if isinstance(state, dict): + return list(state.get("messages") or []) + return list(getattr(state, "messages", []) or []) + + +def _surface_to_parent_command(state: Any) -> Any: + """The body shared by the delegation and swarm-handoff tools: break out of + the manager's react loop and project the subgraph's messages — including + the AIMessage carrying the triggering tool call — onto the PARENT state, + carrying **no** ``goto`` (routing is the parent graph's job). The two tools + differ only in name/description; this is their entire runtime behaviour. + 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: + """Prepend the manager's system prompt with an ``Available workers:`` + roster block listing ``- : `` per worker. + + Each description has whitespace flattened so multi-line descriptions + don't corrupt the one-line-per-worker block shape that the LLM relies + on for routing. + """ + if not entries: + return system_prompt + lines = [ + f"- {name}: {_WHITESPACE_RE.sub(' ', description).strip()}" for name, description in entries + ] + roster = "Available workers:\n" + "\n".join(lines) + return f"{system_prompt}\n\n{roster}" if system_prompt else roster + + +def _make_worker_delegation_tool(worker_node_name: str) -> Any: + """Build the ``delegate_to_`` tool the manager's LLM emits to route to a + worker. The body carries **no** ``goto`` — routing fans out one ``Send`` per + delegation (:func:`_route_manager_to_worker_handoff_or_end`); a ``goto`` here would + collapse multiple same-turn delegations into one parent Command, leaving the other + ``tool_call_id``s unanswered. + """ + from typing import Annotated + + from langchain_core.tools import InjectedToolCallId, tool + from langgraph.prebuilt import InjectedState + from langgraph.types import Command + + tool_name = f"{_DELEGATE_TOOL_PREFIX}{worker_node_name}" + + @tool(tool_name) + def _delegate( + task: str, + state: Annotated[Any, InjectedState], + tool_call_id: Annotated[str, InjectedToolCallId], + ) -> Command: + """Delegate a task to the named worker and wait for its reply. + + ``task`` is the natural-language instruction the worker should + execute. The worker runs in its own isolated message context; + only this ``task`` is forwarded as the worker's first message. + """ + del task, tool_call_id # recovered from the surfaced AIMessage by the routing edge + return _surface_to_parent_command(state) + + _delegate.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." + ) + return _delegate + + +def _handoff_tool_name(destination_name: str) -> str: + """``transfer_to_`` — the tool name the manager's LLM + emits to hand off to a Swarm sibling. Uses the same + :func:`_normalize_identifier` as worker node names so the name is a clean + tool identifier; the raw destination (the Swarm node name used as the + handoff ``goto``) is recovered via the tool-name→destination map held by + the handoff node.""" + return f"{_HANDOFF_TOOL_PREFIX}{_normalize_identifier(destination_name) or 'agent'}" + + +def _make_swarm_handoff_tool(destination_name: str) -> Any: + """Build the ``transfer_to_`` tool a Swarm-member ManagerWorkers' manager + emits to hand off to a sibling. Like the delegation tool it carries no ``goto``; the + parent's ``__handoff__`` node (:func:`_make_handoff_forward_node`) re-emits the + handoff. A plain ``langgraph_swarm.create_handoff_tool`` can't be used here: its + ``Command(graph=PARENT)`` would land on *this* ManagerWorkers graph — one level short + of the Swarm — and be dropped. + """ + from typing import Annotated + + from langchain_core.tools import InjectedToolCallId, tool + from langgraph.prebuilt import InjectedState + from langgraph.types import Command + + tool_name = _handoff_tool_name(destination_name) + + @tool(tool_name) + def _handoff( + state: Annotated[Any, InjectedState], + tool_call_id: Annotated[str, InjectedToolCallId], + ) -> Command: + """Hand the whole conversation off to the named agent.""" + # id is recovered from the surfaced AIMessage by the handoff node. + del tool_call_id + return _surface_to_parent_command(state) + + _handoff.description = ( + f"Transfer the full conversation to the '{destination_name}' agent so " + f"it takes over the dialogue with the user. Use this when " + f"'{destination_name}' is better suited to continue; you will not " + f"regain control afterwards." + ) + return _handoff + + +def _make_handoff_forward_node(handoff_dest_by_tool_name: Dict[str, str]) -> Any: + """Build the parent-graph node that performs a Swarm handoff. + + Reached (via the routing edge) when the manager's last AIMessage carries a + ``transfer_to_`` tool call. It returns a + ``Command(goto=, graph=Command.PARENT, update={..., active_agent})`` + that re-emits the handoff up to the Swarm graph (mirroring what + ``langgraph_swarm.create_handoff_tool`` does for a plain Agent member). + + Transcript validity: the ManagerWorkers graph exits via this PARENT + command rather than running to its own END, so its internal messages do + **not** merge into the shared Swarm conversation — only this command's + ``update`` does. We therefore forward the manager's transfer AIMessage + itself, followed by a ``ToolMessage`` answering every (still-unanswered) + tool call on it, so the Swarm sees a well-formed + ``AIMessage(tool_calls)`` → ``ToolMessage`` sequence — an orphan + ToolMessage would 400 the next member's LLM. Answering *every* call (not + just the transfer) keeps the sequence valid even when the manager emitted + delegations in the same turn; the manager's internal delegation mechanics + otherwise stay hidden inside the ManagerWorkers, as in a standalone run. + + Returns a ``RunnableLambda`` exposing both sync (``func``) and async + (``afunc``) entrypoints so LangGraph can call it on either path; the body + is pure so the async wrapper just delegates. + """ + from langchain_core.messages import ToolMessage + from langgraph.types import Command + + from pyagentspec.adapters.langgraph._types import RunnableLambda + + def _forward(state: Dict[str, Any]) -> Any: + messages = state.get("messages") or [] + last = messages[-1] if messages else None + tool_calls = getattr(last, "tool_calls", None) or [] + transfer_call = next( + (tc for tc in tool_calls if _tc_get(tc, "name") in handoff_dest_by_tool_name), + None, + ) + if transfer_call is None: + # Defensive: the routing edge only sends us here on a transfer call. + return {"messages": []} + destination = handoff_dest_by_tool_name[_tc_get(transfer_call, "name")] + transfer_id = _tc_get(transfer_call, "id") or "" + already_answered = { + getattr(m, "tool_call_id", None) for m in messages if getattr(m, "type", None) == "tool" + } + tool_messages: List[Any] = [] + for tc in tool_calls: + call_id = _tc_get(tc, "id") or "" + if call_id in already_answered: + continue + if call_id == transfer_id: + content = f"Successfully transferred to {destination}" + else: + content = f"Not executed: the conversation was handed off to " f"{destination}." + tool_messages.append( + ToolMessage( + content=content, + name=_tc_get(tc, "name"), + tool_call_id=call_id, + ) + ) + return Command( + goto=destination, + graph=Command.PARENT, + # AIMessage first, then its answering ToolMessages — see docstring. + update={"messages": [last, *tool_messages], "active_agent": destination}, + ) + + async def _forward_async(state: Dict[str, Any]) -> Any: + return _forward(state) + + return RunnableLambda( + func=_forward, + afunc=_forward_async, + name="swarm_handoff", + ) + + +def _is_handoff_name(name: Any) -> bool: + """True if ``name`` is one of the synthetic ``transfer_to_`` tool + names the manager emits to hand the conversation off to a Swarm sibling.""" + return isinstance(name, str) and name.startswith(_HANDOFF_TOOL_PREFIX) + + +def _route_manager_to_worker_handoff_or_end(state: Dict[str, Any]) -> Any: + """Inspect the manager's last AIMessage and route the parent graph. + + Routing precedence: + + * a ``transfer_to_`` tool call → the Swarm-handoff node + (:func:`_make_handoff_forward_node`), which re-emits the handoff up to + the Swarm. Handoff transfers the *whole* conversation, so it wins over + delegation and routes to a single destination (only present when this + ManagerWorkers is a Swarm member). + * one or more ``delegate_to_`` tool calls → one ``Send`` per + delegation; otherwise → ``END``. + + A single manager turn may emit several ``delegate_to_`` calls; each gets its + own ``Send`` carrying the ``task`` + ``tool_call_id``, so every call is answered + independently (an unanswered delegation breaks the manager's next-turn + tool-call/result sequence). Multiple ``Send``s to one worker run independently; plain + tool calls already ran inside the manager's 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 [] + sends = [] + handoff = False + for tc in tool_calls: + name = _tc_get(tc, "name") + if _is_handoff_name(name): + handoff = True + elif _is_delegate_name(name): + args = _tc_get(tc, "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 "", + }, + ) + ) + # Swarm handoff wins: it hands the whole conversation to a sibling, so any + # delegations in the same turn are answered-and-dropped by the handoff node. + if handoff: + return _HANDOFF_NODE_KEY + return sends or langgraph_graph.END + + +def _wrap_worker_for_subgraph( + worker_graph: CompiledStateGraph[Any, Any, Any], + worker_node_name: str, +) -> Any: + """Wrap a worker subgraph so it runs with an isolated ``messages`` + context (the delegation task only) and its final reply comes back as + a ToolMessage matched to the manager's pending delegation tool-call. + + This is what makes a ManagerWorkers parent graph hierarchical rather + than a shared-state Swarm: workers do NOT see each other's messages, + and only one message — the manager's chosen task — is forwarded to + each worker run. The worker's last AIMessage content is captured as + the ToolMessage content so the manager's react-agent loop sees a + well-formed tool response on the next turn. + + Returns a ``RunnableLambda`` exposing both sync (``func``) and async + (``afunc``) entrypoints — LangGraph picks the right one based on + whether the parent graph is invoked via ``invoke`` or ``ainvoke``. + """ + from langchain_core.messages import HumanMessage, ToolMessage + + from pyagentspec.adapters.langgraph._types import RunnableLambda + + delegate_tool_name = f"{_DELEGATE_TOOL_PREFIX}{worker_node_name}" + + def _extract_pending(state: Dict[str, Any]) -> Tuple[str, str]: + # Fan-out path: the routing edge's ``Send`` payload carries this + # delegation's task and its originating tool_call_id directly, so a + # single manager turn can delegate to this worker more than once + # without the runs colliding on a shared "first pending call". + if isinstance(state, dict) and _DELEGATE_CALL_ID_KEY in state: + return ( + state.get(_DELEGATE_TASK_KEY) or "", + state.get(_DELEGATE_CALL_ID_KEY) or "", + ) + # Direct-edge path (a worker wired in without Send): recover task + + # id from the manager's last AIMessage. Only the first matching call + # is recoverable this way, which is why routing prefers Send. + messages = state.get("messages") or [] + if not messages: + raise RuntimeError(f"Worker '{worker_node_name}' was invoked with empty manager state.") + last_ai = messages[-1] + tool_calls = getattr(last_ai, "tool_calls", None) or [] + pending_call = next( + (tc for tc in tool_calls if _tc_get(tc, "name") == delegate_tool_name), + None, + ) + if pending_call is None: + raise RuntimeError( + f"Worker '{worker_node_name}' was routed to but the manager's " + f"last message has no '{delegate_tool_name}' tool call." + ) + args = _tc_get(pending_call, "args") or {} + call_id = _tc_get(pending_call, "id") or "" + return args.get("task") or "", call_id + + def _tool_message_from(reply: str, call_id: str) -> Dict[str, Any]: + return {"messages": [ToolMessage(content=reply, tool_call_id=call_id)]} + + def _worker_input(task: str) -> Dict[str, Any]: + # Pass NO explicit config so the worker inherits this node's ambient run config: + # its ``checkpoint_ns`` (``:``) is what streams the worker's + # token events under the worker node, and the distinct per-superstep namespace + # keeps repeated delegations isolated without a fresh thread_id. + return {"messages": [HumanMessage(content=task)]} + + def _last_message_content(result: Any) -> str: + messages = result.get("messages") if isinstance(result, dict) else None + if not messages: + return "" + return getattr(messages[-1], "content", "") or "" + + def _run_sync(state: Dict[str, Any]) -> Dict[str, Any]: + task, call_id = _extract_pending(state) + result = worker_graph.invoke(_worker_input(task)) + return _tool_message_from(_last_message_content(result), call_id) + + async def _run_async(state: Dict[str, Any]) -> Dict[str, Any]: + task, call_id = _extract_pending(state) + result = await worker_graph.ainvoke(_worker_input(task)) + return _tool_message_from(_last_message_content(result), call_id) + + return RunnableLambda( + func=_run_sync, + afunc=_run_async, + name=f"worker:{worker_node_name}", + ) + + +# ─── ManagerWorkers: hide the delegation protocol from astream_events ───────── + + +def _is_delegate_name(name: Any) -> bool: + """True if ``name`` is one of the synthetic ``delegate_to_`` + tool names the manager emits to route to a worker.""" + return isinstance(name, str) and name.startswith(_DELEGATE_TOOL_PREFIX) + + +def _is_delegate_tool_message(msg: Any, delegate_call_ids: "set") -> bool: + """True if ``msg`` is the worker's synthetic reply ToolMessage — i.e. a + ToolMessage answering a (now-hidden) delegation tool-call id.""" + return ( + getattr(msg, "type", None) == "tool" + and getattr(msg, "tool_call_id", None) in delegate_call_ids + ) + + +def _scrubbed_ai_message( + msg: Any, + delegate_indices: "set", + delegate_call_ids: "set", +) -> Tuple[Optional[Any], bool]: + """Return ``(scrubbed_copy_or_None, is_empty)`` for an AIMessage(Chunk), + removing every ``delegate_to_`` tool call. + + ``scrubbed_copy_or_None`` is ``None`` when the message carried no + delegation artifact (the caller emits it unchanged). ``is_empty`` is + ``True`` when, after removal, nothing renderable remains (no content and + no other tool calls) — the caller drops the event. + + Never mutates ``msg``: the same object lives in the graph's message + state, where the manager react loop relies on the delegation + tool-call / tool-result pair staying intact. ``delegate_indices`` tracks + streamed tool-call positions so argument-continuation chunks (which + carry no ``name``) are stripped too; ``delegate_call_ids`` collects the + call ids so the worker's matching ToolMessage can be dropped later. + """ + changed = False + + # Provider-native streamed tool calls (e.g. OpenAI) ride along in + # ``additional_kwargs['tool_calls']`` and stream by index with the name + # only on the opening delta — match by name or by a known delegate index. + additional = getattr(msg, "additional_kwargs", None) or {} + new_additional = additional + raw_calls = additional.get("tool_calls") + if raw_calls: + kept_raw = [] + for tc in raw_calls: + index = tc.get("index") if isinstance(tc, dict) else None + function = (tc.get("function") or {}) if isinstance(tc, dict) else {} + fname = function.get("name") + if _is_delegate_name(fname) or (not fname and index in delegate_indices): + if index is not None: + delegate_indices.add(index) + if isinstance(tc, dict) and tc.get("id"): + delegate_call_ids.add(tc["id"]) + changed = True + else: + kept_raw.append(tc) + if len(kept_raw) != len(raw_calls): + new_additional = dict(additional) + if kept_raw: + new_additional["tool_calls"] = kept_raw + else: + new_additional.pop("tool_calls", None) + + # AIMessageChunk: ``tool_call_chunks`` is the source of truth and + # ``tool_calls`` / ``invalid_tool_calls`` are *derived* from it, so we + # rebuild the chunk (which re-runs that derivation) rather than copying — + # otherwise a stale derived ``tool_calls`` entry survives the strip. + if hasattr(msg, "tool_call_chunks"): + kept_chunks = [] + for chunk in getattr(msg, "tool_call_chunks", None) or []: + cname, cindex = chunk.get("name"), chunk.get("index") + if _is_delegate_name(cname) or (cname is None and cindex in delegate_indices): + if cindex is not None: + delegate_indices.add(cindex) + if chunk.get("id"): + delegate_call_ids.add(chunk["id"]) + changed = True + else: + kept_chunks.append(chunk) + if not changed: + return None, False + scrubbed = type(msg)( + content=msg.content, + additional_kwargs=new_additional, + response_metadata=getattr(msg, "response_metadata", None) or {}, + tool_call_chunks=kept_chunks, + id=getattr(msg, "id", None), + name=getattr(msg, "name", None), + usage_metadata=getattr(msg, "usage_metadata", None), + ) + has_remaining = ( + bool(scrubbed.content) + or bool(scrubbed.tool_call_chunks) + or bool((scrubbed.additional_kwargs or {}).get("tool_calls")) + ) + return scrubbed, not has_remaining + + # Full AIMessage: ``tool_calls`` is the source of truth. + update: Dict[str, Any] = {} + for attr in ("tool_calls", "invalid_tool_calls"): + items = getattr(msg, attr, None) + if items: + kept = [] + for tc in items: + if _is_delegate_name(_tc_get(tc, "name")): + cid = _tc_get(tc, "id") + if cid: + delegate_call_ids.add(cid) + changed = True + else: + kept.append(tc) + if len(kept) != len(items): + update[attr] = kept + if new_additional is not additional: + update["additional_kwargs"] = new_additional + if not changed: + return None, False + + scrubbed = msg.model_copy(update=update) + has_remaining = ( + bool(getattr(scrubbed, "content", None)) + or bool(getattr(scrubbed, "tool_calls", None)) + or bool((getattr(scrubbed, "additional_kwargs", None) or {}).get("tool_calls")) + ) + return scrubbed, not has_remaining + + +def _scrub_payload_messages( + payload: Any, + delegate_call_ids: "set", +) -> Tuple[Any, bool]: + """For a node / state payload shaped ``{"messages": [...]}``, remove the + whole delegation protocol so it never surfaces in a consumer-facing + message snapshot: drop the worker's synthetic reply ToolMessage(s) AND + strip the synthetic ``delegate_to_`` tool calls off the manager's + AIMessage(s), dropping an AIMessage that is left empty (a pure delegation + turn). + + Stripping the tool calls — not just the ToolMessages — is what keeps a + downstream message snapshot consistent. A consumer that builds its + message history from an ``on_chain_end`` state payload (e.g. the AG-UI + MESSAGES_SNAPSHOT) reads ``tool_calls`` straight off the AIMessage; if we + dropped only the reply ToolMessages, the snapshot would carry delegate + tool calls whose results are gone, which renders as a "tool call with no + result". Messages are walked in order, so a delegation AIMessage records + its call ids before its reply ToolMessages are tested for removal. + + Returns ``(payload, drop_event)``: ``payload`` is a new dict when + anything changed (the original is never mutated), otherwise the object + passed in. ``drop_event`` is ``True`` when scrubbing empties the + ``messages`` list, so the caller drops the whole event. + """ + if not isinstance(payload, dict): + return payload, False + messages = payload.get("messages") + if not isinstance(messages, list) or not messages: + return payload, False + kept: List[Any] = [] + changed = False + for m in messages: + # The worker's reply ToolMessage — pure delegation plumbing. + if _is_delegate_tool_message(m, delegate_call_ids): + changed = True + continue + # An AIMessage may carry delegate tool calls; strip them and drop the + # message if nothing renderable remains. Non-delegation messages + # (real tool calls/results, plain content) are left untouched. + if hasattr(m, "tool_calls"): + scrubbed, is_empty = _scrubbed_ai_message(m, set(), delegate_call_ids) + if scrubbed is not None: + changed = True + if not is_empty: + kept.append(scrubbed) + continue + kept.append(m) + if not changed: + return payload, False + new_payload = dict(payload) + new_payload["messages"] = kept + return new_payload, len(kept) == 0 + + +class _DelegationEventFilter: + """Stateful scrubber for a single ``astream_events`` stream. + + Removes the synthetic ``delegate_to_`` routing protocol — the + delegation tool calls, their ``on_tool_*`` lifecycle events, and the + worker's matching reply ToolMessage — from the consumer-facing event + view. The graph's message state is never touched, so the manager react + loop still sees its well-formed tool-call / tool-result exchange. + """ + + def __init__(self) -> None: + # Streamed tool-call positions per chat-model run that belong to a + # delegation call, so argument-continuation chunks (name=None) are + # stripped along with the opening chunk. + self._delegate_indices_by_run: Dict[str, "set"] = {} + # Delegate tool-call ids seen so far, so the worker's reply + # ToolMessage can be dropped when it surfaces downstream. + self._delegate_call_ids: "set" = set() + + def scrub(self, event: Dict[str, Any]) -> Optional[Dict[str, Any]]: + etype = event.get("event") + name = event.get("name", "") + + # 1. Drop the tool lifecycle events for the delegation tools. + if etype in ("on_tool_start", "on_tool_end", "on_tool_error") and _is_delegate_name(name): + return None + + data = event.get("data") or {} + + # 2. Strip delegate tool calls from streamed / final manager AIMessages. + if etype in ("on_chat_model_stream", "on_chat_model_end"): + key = "chunk" if etype == "on_chat_model_stream" else "output" + msg = data.get(key) + if msg is not None and hasattr(msg, "tool_calls"): + run_id = event.get("run_id", "") + indices = self._delegate_indices_by_run.setdefault(run_id, set()) + scrubbed, is_empty = _scrubbed_ai_message(msg, indices, self._delegate_call_ids) + if scrubbed is not None: + # A streamed chunk that became empty is pure delegation + # plumbing — drop it. A final ``on_chat_model_end`` is kept + # (scrubbed) so consumers still get a turn-end marker. + if is_empty and etype == "on_chat_model_stream": + return None + new_data = dict(data) + new_data[key] = scrubbed + new_event = dict(event) + new_event["data"] = new_data + return new_event + return event + + # 3. Drop the worker's synthetic reply ToolMessage wherever it + # surfaces in a node payload. + new_data: Optional[Dict[str, Any]] = None + should_drop = False + for key in ("chunk", "output", "input"): + if key in data: + scrubbed_payload, drop_event = _scrub_payload_messages( + data[key], self._delegate_call_ids + ) + if scrubbed_payload is not data[key]: + if new_data is None: + new_data = dict(data) + new_data[key] = scrubbed_payload + if drop_event: + should_drop = True + if should_drop: + return None + if new_data is not None: + new_event = dict(event) + new_event["data"] = new_data + return new_event + return event + + +def _patch_hide_delegation_in_astream_events( + compiled_graph: CompiledStateGraph[Any, Any, Any], +) -> None: + """Wrap ``astream_events`` so the synthetic ``delegate_to_`` + routing protocol never reaches the consumer. + + ManagerWorkers routes by having the manager react-agent emit a + ``delegate_to_`` tool call, which the worker answers with a + ToolMessage matched to that call id. That pair is load-bearing for the + manager's react loop (it must observe a well-formed tool-call / + tool-result exchange) but it is internal plumbing the consumer should + never see as phantom tool calls. We filter only the emitted events; the + graph's message state is untouched, so the loop is unaffected. The + workers' real LLM/token events still propagate (they reach the consumer + via callback propagation through the isolated worker run), so this + strips the routing noise without hiding the workers' actual output. + """ + original_astream_events = compiled_graph.astream_events + + async def patched_astream_events(*args: Any, **kwargs: Any) -> AsyncGenerator[Any, None]: + event_filter = _DelegationEventFilter() + async for event in original_astream_events(*args, **kwargs): + if not isinstance(event, dict): + yield event + continue + # Fail open: a scrubbing bug must never tear down the stream + # (which would swallow every later event — notably the worker + # events that follow the manager's delegation turn). On error we + # emit the event unfiltered rather than dropping the rest. + try: + kept = event_filter.scrub(event) + except Exception: # noqa: BLE001 — defensive, see above + logging.getLogger("pyagentspec.adapters.langgraph").warning( + "ManagerWorkers astream_events delegation filter raised; " + "passing the event through unfiltered.", + exc_info=True, + ) + yield event + continue + if kept is not None: + yield kept + + compiled_graph.astream_events = patched_astream_events # type: ignore[assignment] + + +def _patch_with_manager_workers_execution_span( + compiled_graph: CompiledStateGraph[Any, Any, Any], + mw: AgentSpecManagerWorkers, +) -> None: + """Wrap ``stream`` / ``astream`` so each ManagerWorkers run emits a + ``ManagerWorkersExecutionSpan`` with Start/End events. Mirrors the + patches applied to Agent and Flow compiled graphs elsewhere in this + converter. + """ + original_stream = compiled_graph.stream + original_astream = compiled_graph.astream + + def _coerce_inputs(kwargs: Dict[str, Any]) -> Dict[str, Any]: + inputs = kwargs.get("input", {}) + return inputs if isinstance(inputs, dict) else {} + + def patched_stream(*args: Any, **kwargs: Any) -> Generator[Any, Any, None]: + span_name = f"ManagerWorkersExecution[{mw.name}]" + inputs = _coerce_inputs(kwargs) + with AgentSpecManagerWorkersExecutionSpan(name=span_name, managerworkers=mw) as span: + span.add_event(AgentSpecManagerWorkersExecutionStart(managerworkers=mw, inputs=inputs)) + last_chunk: Dict[str, Any] = {} + for chunk in original_stream(*args, **kwargs): + yield chunk + if isinstance(chunk, tuple) and isinstance(chunk[1], dict): + last_chunk = chunk[1] + span.add_event( + AgentSpecManagerWorkersExecutionEnd( + managerworkers=mw, + outputs={"messages": last_chunk.get("messages", [])}, + ) + ) + + async def patched_astream(*args: Any, **kwargs: Any) -> AsyncGenerator[Any, Any]: + span_name = f"ManagerWorkersExecution[{mw.name}]" + inputs = _coerce_inputs(kwargs) + span = AgentSpecManagerWorkersExecutionSpan(name=span_name, managerworkers=mw) + try: + await span.start_async() + except NotImplementedError: + span.start() + try: + start_event = AgentSpecManagerWorkersExecutionStart(managerworkers=mw, inputs=inputs) + try: + await span.add_event_async(start_event) + except NotImplementedError: + span.add_event(start_event) + last_chunk: Dict[str, Any] = {} + async for chunk in original_astream(*args, **kwargs): + yield chunk + if isinstance(chunk, tuple) and isinstance(chunk[1], dict): + last_chunk = chunk[1] + end_event = AgentSpecManagerWorkersExecutionEnd( + managerworkers=mw, + outputs={"messages": last_chunk.get("messages", [])}, + ) + try: + await span.add_event_async(end_event) + except NotImplementedError: + span.add_event(end_event) + finally: + try: + await span.end_async() + except NotImplementedError: + span.end() + + compiled_graph.stream = patched_stream # type: ignore[assignment] + compiled_graph.astream = patched_astream # type: ignore[assignment] diff --git a/pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py b/pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py index f1999aef..65bca11f 100644 --- a/pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py +++ b/pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py @@ -49,8 +49,10 @@ from pyagentspec.flows.nodes import OutputMessageNode as AgentSpecOutputMessageNode from pyagentspec.flows.nodes import StartNode as AgentSpecStartNode from pyagentspec.flows.nodes import ToolNode as AgentSpecToolNode +from pyagentspec.managerworkers import ManagerWorkers as AgentSpecManagerWorkers from pyagentspec.property import Property as AgentSpecProperty from pyagentspec.property import _empty_default as pyagentspec_empty_default +from pyagentspec.swarm import Swarm as AgentSpecSwarm from pyagentspec.tracing.events import NodeExecutionEnd as AgentSpecNodeExecutionEnd from pyagentspec.tracing.events import NodeExecutionStart as AgentSpecNodeExecutionStart from pyagentspec.tracing.events.exception import ExceptionRaised @@ -529,16 +531,95 @@ def _create_react_agent_with_given_input_values( ) return self._agents_cache[system_prompt] + def _create_composite_graph_with_given_input_values( + self, inputs: Dict[str, Any] + ) -> CompiledStateGraph[Any, Any]: + """Compile the node's ``ManagerWorkers`` / ``Swarm`` into a runnable graph for + these inputs, cached by the rendered entry-agent prompt. + + Such a graph runs over ``MessagesState`` and can't carry structured inputs to its + inner agents, so the node inputs are baked into the entry agent's ``system_prompt`` + (the ``group_manager`` for a ManagerWorkers, the ``first_agent`` for a Swarm) and + the now-satisfied input ports dropped, so declared == inferred for the downstream + span re-validation. A non-Agent entry is passed through unchanged so the converter + raises its own clear error. + """ + from pyagentspec.adapters.langgraph._langgraphconverter import ( + AgentSpecToLangGraphConverter, + ) + + converter = AgentSpecToLangGraphConverter() + component = self.node.agent + if isinstance(component, AgentSpecManagerWorkers): + entry_agent = component.group_manager + convert = converter._manager_workers_convert_to_langgraph + + def rebuild(rendered_entry: Any) -> Any: + return component.model_copy(update={"group_manager": rendered_entry, "inputs": []}) + + elif isinstance(component, AgentSpecSwarm): + entry_agent = component.first_agent + convert = converter._swarm_convert_to_langgraph + + def rebuild(rendered_entry: Any) -> Any: + # The entry agent appears as first_agent and inside the relationship + # tuples, so swap it (matched by id) in both. + def _swap(agent: Any) -> Any: + return rendered_entry if agent.id == entry_agent.id else agent + + return component.model_copy( + update={ + "first_agent": rendered_entry, + "relationships": [ + (_swap(caller), _swap(recipient)) + for caller, recipient in component.relationships + ], + "inputs": [], + } + ) + + else: + raise TypeError( + "_create_composite_graph_with_given_input_values requires a " + "ManagerWorkers or Swarm" + ) + + is_agent_entry = isinstance(entry_agent, AgentSpecAgent) + cache_key = ( + render_template(entry_agent.system_prompt, inputs) if is_agent_entry else component.id + ) + if cache_key not in self._agents_cache: + rendered = ( + rebuild(entry_agent.model_copy(update={"system_prompt": cache_key, "inputs": []})) + if is_agent_entry + else component + ) + self._agents_cache[cache_key] = convert( + rendered, + tool_registry=self.tool_registry, + converted_components=self.converted_components, + checkpointer=self.checkpointer, + config=self.config, + middleware=self._middleware, + ) + return self._agents_cache[cache_key] + def _prepare_agent_and_inputs( self, inputs: Dict[str, Any], messages: Messages ) -> Tuple[CompiledStateGraph[Any, Any], Dict[str, Any]]: - agent = self._create_react_agent_with_given_input_values(inputs) # LangGraph's agent expects at least one user message to drive execution. # When an AgentNode is used with a templated system prompt and no messages are provided # by the flow, the agent can crash. To avoid this, we artificially insert an empty # user message when the message list is empty. if not messages: messages = cast(Messages, [{"role": "user", "content": ""}]) + if isinstance(self.node.agent, (AgentSpecManagerWorkers, AgentSpecSwarm)): + # A ManagerWorkers / Swarm flow step runs as a multi-agent graph over + # MessagesState: node inputs were baked into the entry agent's prompt, so the + # graph is driven by messages alone (not the agent's remaining_steps state). + graph = self._create_composite_graph_with_given_input_values(inputs) + return graph, {"messages": messages} + agent = self._create_react_agent_with_given_input_values(inputs) inputs |= { "remaining_steps": 20, # Get the right number of steps left "messages": messages, @@ -919,13 +1000,25 @@ def _accumulate_outputs( outputs[collected_output_name].append(output_value) +def is_single_string_output(expected_outputs: List[AgentSpecProperty]) -> bool: + """Whether the declared outputs are a single string property. + + Such an output is the model's free text, not a structured field, so the + adapter takes it directly from the agent's final message rather than forcing + structured generation. Mirrors ``LlmNodeExecutor``'s single-string handling + and lets a string output work on models without structured-output support. + """ + outputs = expected_outputs or [] + return len(outputs) == 1 and outputs[0].type == "string" + + def extract_outputs_from_invoke_result( result: Dict[str, Any], expected_outputs: List[AgentSpecProperty] ) -> Dict[str, Any]: # Extracts the outputs from the return value of an invoke call made on an agent # The outputs are typically exposed as part of the `structured_response`, or as entries in the result directly. # We give priority to the latter. - return { + outputs = { # Defaults if available **{ output.title: output.default @@ -941,3 +1034,17 @@ def extract_outputs_from_invoke_result( if output.title in result }, } + # A single string output is the agent's free-text answer, not a structured + # field. When structured generation didn't populate it — because the model + # lacks structured-output support, or because none was requested (see + # ``_create_react_agent_with_given_info``) — fall back to the final message + # content so the output still carries the agent's response. + if is_single_string_output(expected_outputs): + title = expected_outputs[0].title + if title not in outputs: + messages = result.get("messages") + if messages: + content = getattr(messages[-1], "content", None) + if content is not None: + outputs[title] = content + return outputs diff --git a/pyagentspec/src/pyagentspec/managerworkers.py b/pyagentspec/src/pyagentspec/managerworkers.py index 26c2bdeb..130ad4de 100644 --- a/pyagentspec/src/pyagentspec/managerworkers.py +++ b/pyagentspec/src/pyagentspec/managerworkers.py @@ -13,6 +13,7 @@ from typing_extensions import Self from pyagentspec.agenticcomponent import AgenticComponent +from pyagentspec.property import Property from pyagentspec.validation_helpers import model_validator_with_error_accumulation from pyagentspec.versioning import AgentSpecVersionEnum @@ -65,6 +66,24 @@ class ManagerWorkers(AgenticComponent): default=AgentSpecVersionEnum.v25_4_2, init=False, exclude=True ) + def _get_inferred_inputs(self) -> List[Property]: + """A ``ManagerWorkers`` exposes the inputs of its group manager. + + The group manager is the component that drives the conversation and whose prompt + the run-time renders, so the manager-workers component accepts exactly the inputs + the group manager accepts (e.g. the ``{{placeholder}}`` inputs of an ``Agent`` + group manager). Without this, the base default infers no inputs, so a + ``ManagerWorkers`` used as a flow ``AgentNode`` would expose no input ports and a + data-flow edge into it could not resolve. + """ + group_manager = getattr(self, "group_manager", None) + return list(getattr(group_manager, "inputs", None) or []) + + def _get_inferred_outputs(self) -> List[Property]: + """Outputs of the group manager; see :meth:`_get_inferred_inputs`.""" + group_manager = getattr(self, "group_manager", None) + return list(getattr(group_manager, "outputs", None) or []) + @model_validator_with_error_accumulation def _validate_one_or_more_workers(self) -> Self: if len(self.workers) == 0: diff --git a/pyagentspec/src/pyagentspec/swarm.py b/pyagentspec/src/pyagentspec/swarm.py index 3b9c7f72..79ea12a2 100644 --- a/pyagentspec/src/pyagentspec/swarm.py +++ b/pyagentspec/src/pyagentspec/swarm.py @@ -15,6 +15,7 @@ from pyagentspec.agenticcomponent import AgenticComponent from pyagentspec.component import SerializeAsEnum +from pyagentspec.property import Property from pyagentspec.validation_helpers import model_validator_with_error_accumulation from pyagentspec.versioning import AgentSpecVersionEnum @@ -127,6 +128,24 @@ class Swarm(AgenticComponent): default=AgentSpecVersionEnum.v25_4_2, init=False, exclude=True ) + def _get_inferred_inputs(self) -> List[Property]: + """A ``Swarm`` exposes the inputs of its entry agent (``first_agent``). + + Symmetric with :meth:`ManagerWorkers._get_inferred_inputs`. The ``first_agent`` + is the swarm's entry point — it interacts with the user before any handoff — so + the swarm component accepts exactly the inputs that agent accepts (e.g. the + ``{{placeholder}}`` inputs of an ``Agent`` entry's prompt). Without this, the base + default infers no inputs, so a flow ``AgentNode`` wrapping a swarm declares no + input ports and a ``DataFlowEdge`` into it fails to resolve at load. + """ + first_agent = getattr(self, "first_agent", None) + return list(getattr(first_agent, "inputs", None) or []) + + def _get_inferred_outputs(self) -> List[Property]: + """Outputs of the entry agent (``first_agent``); see :meth:`_get_inferred_inputs`.""" + first_agent = getattr(self, "first_agent", None) + return list(getattr(first_agent, "outputs", None) or []) + @model_validator(mode="before") def _raise_warning_if_handoff_is_bool(cls: Self, values: Any) -> Any: import warnings diff --git a/pyagentspec/tests/adapters/langgraph/flows/test_agentnode.py b/pyagentspec/tests/adapters/langgraph/flows/test_agentnode.py index e5b4ba12..474f74b4 100644 --- a/pyagentspec/tests/adapters/langgraph/flows/test_agentnode.py +++ b/pyagentspec/tests/adapters/langgraph/flows/test_agentnode.py @@ -96,6 +96,97 @@ def test_agentnode_can_be_imported_and_executed(agent_flow: Flow) -> None: assert "car" in outputs +def test_is_single_string_output() -> None: + """A lone string output is treated as free text, not a structured field.""" + from pyagentspec.adapters.langgraph._node_execution import is_single_string_output + from pyagentspec.property import IntegerProperty, StringProperty + + assert is_single_string_output([StringProperty(title="x")]) is True + assert is_single_string_output([]) is False + assert is_single_string_output([IntegerProperty(title="n")]) is False + assert is_single_string_output([StringProperty(title="a"), StringProperty(title="b")]) is False + + +def test_single_string_output_taken_from_final_message_without_structured_generation() -> None: + """An AgentNode whose agent declares a single string output should resolve + that output from the agent's final message — no structured generation, so + it works on models without structured-output support. + + The model is stubbed with a ``FakeMessagesListChatModel`` (no structured + output); if the converter still attached a ``response_format`` the output + would come back empty. Asserting it equals the message content proves the + single-string path takes the final message instead. + """ + from unittest.mock import patch + + from langchain_core.language_models.fake_chat_models import ( + FakeMessagesListChatModel, + ) + from langchain_core.messages import AIMessage + from langchain_openai import ChatOpenAI + from langgraph.checkpoint.memory import MemorySaver + + from pyagentspec.adapters.langgraph import AgentSpecLoader + from pyagentspec.adapters.langgraph._langgraphconverter import ( + AgentSpecToLangGraphConverter, + ) + from pyagentspec.llms.openaicompatibleconfig import OpenAiCompatibleConfig + + class _FakeModel(FakeMessagesListChatModel, ChatOpenAI): + pass + + fake_llm = _FakeModel(responses=[AIMessage(content="42")]) + + answer = StringProperty(title="answer") + agent = Agent( + name="agent", + llm_config=OpenAiCompatibleConfig(name="agent_llm", model_id="fake", url="null"), + system_prompt="Answer the question.", + outputs=[answer], + ) + agent_node = AgentNode(name="agent_node", agent=agent) + start_node = StartNode(name="start") + end_node = EndNode(name="end", outputs=[answer]) + flow = Flow( + name="flow", + start_node=start_node, + nodes=[start_node, agent_node, end_node], + control_flow_connections=[ + ControlFlowEdge(name="start_to_node", from_node=start_node, to_node=agent_node), + ControlFlowEdge(name="node_to_end", from_node=agent_node, to_node=end_node), + ], + data_flow_connections=[ + DataFlowEdge( + name="answer_edge", + source_node=agent_node, + source_output=answer.title, + destination_node=end_node, + destination_input=answer.title, + ), + ], + outputs=[answer], + ) + + loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver()) + with patch.object( + AgentSpecToLangGraphConverter, + "_llm_convert_to_langgraph", + autospec=True, + side_effect=lambda self_obj, llm_config, *a, **k: fake_llm, + ), patch.object( + FakeMessagesListChatModel, + "bind_tools", + new=lambda self_obj, *a, **k: self_obj, + ): + compiled = loader.load_component(flow) + result = compiled.invoke( + {"inputs": {}, "messages": [{"role": "user", "content": "What is 6*7?"}]}, + {"configurable": {"thread_id": "agentnode-single-string"}}, + ) + + assert result["outputs"]["answer"] == "42" + + @pytest.mark.anyio @retry_test(max_attempts=3, wait_between_tries=2) async def test_agentnode_can_be_executed_async(agent_flow: Flow) -> None: diff --git a/pyagentspec/tests/adapters/langgraph/flows/test_managerworkers_node.py b/pyagentspec/tests/adapters/langgraph/flows/test_managerworkers_node.py new file mode 100644 index 00000000..d38d0005 --- /dev/null +++ b/pyagentspec/tests/adapters/langgraph/flows/test_managerworkers_node.py @@ -0,0 +1,157 @@ +# Copyright © 2026 Oracle and/or its affiliates. +# +# This software is under the Apache License 2.0 +# (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. + +"""A ManagerWorkers used as a flow step (AgentNode). + +Regression coverage for two coupled behaviours: + * ``ManagerWorkers._get_inferred_inputs`` exposes the group manager's inputs, so a + flow ``AgentNode`` wrapping a manager declares input ports and a ``DataFlowEdge`` + into it resolves at load (previously: "node does not have any input property..."). + * ``AgentNodeExecutor`` runs a ManagerWorkers node (previously: TypeError "can only + be used with AgentSpecAgent agents"), rendering the node inputs into the group + manager's prompt and returning its result. +""" + +from pyagentspec.agent import Agent +from pyagentspec.managerworkers import ManagerWorkers +from pyagentspec.property import StringProperty + + +def test_managerworkers_infers_inputs_from_group_manager_prompt() -> None: + """A ManagerWorkers exposes the group manager's prompt placeholders as inputs.""" + llm = {"name": "m", "model_id": "fake", "url": "null"} + from pyagentspec.llms.openaicompatibleconfig import OpenAiCompatibleConfig + + cfg = OpenAiCompatibleConfig(**llm) + manager = Agent( + name="manager", + llm_config=cfg, + system_prompt="Translate the following to Arabic:\n\n{{joke}}\n\nMake {{count}} variants.", + ) + worker = Agent(name="worker", llm_config=cfg, system_prompt="You translate.") + mw = ManagerWorkers(name="mw", group_manager=manager, workers=[worker]) + + assert sorted(p.title for p in (mw.inputs or [])) == ["count", "joke"] + + +def test_managerworkers_infers_outputs_from_group_manager() -> None: + """Symmetric with inputs: a ManagerWorkers exposes the group manager's outputs, + so a flow AgentNode wrapping it can wire its result downstream (or surface it as a + leaf).""" + from pyagentspec.llms.openaicompatibleconfig import OpenAiCompatibleConfig + + cfg = OpenAiCompatibleConfig(name="m", model_id="fake", url="null") + answer = StringProperty(title="answer") + manager = Agent( + name="manager", + llm_config=cfg, + system_prompt="Answer the question.", + outputs=[answer], + ) + worker = Agent(name="worker", llm_config=cfg, system_prompt="You help.") + mw = ManagerWorkers(name="mw", group_manager=manager, workers=[worker]) + + assert [p.title for p in (mw.outputs or [])] == ["answer"] + + +def test_managerworkers_runs_as_a_flow_step_with_data_edge_inputs() -> None: + """A ManagerWorkers flow step loads (data edge resolves) and executes offline. + + The model is stubbed (no real LLM, no delegation), so the manager produces a final + message and the manager graph routes straight to END. Asserts the flow both loads — + proving the manager node exposes the ``joke`` input the data edge targets — and runs, + surfacing the manager's answer as the node's single string output. + """ + from unittest.mock import patch + + from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel + from langchain_core.messages import AIMessage + from langchain_openai import ChatOpenAI + from langgraph.checkpoint.memory import MemorySaver + + from pyagentspec.adapters.langgraph import AgentSpecLoader + from pyagentspec.adapters.langgraph._langgraphconverter import ( + AgentSpecToLangGraphConverter, + ) + from pyagentspec.flows.edges import ControlFlowEdge, DataFlowEdge + from pyagentspec.flows.flow import Flow + from pyagentspec.flows.nodes import AgentNode, EndNode, StartNode + from pyagentspec.llms.openaicompatibleconfig import OpenAiCompatibleConfig + + class _FakeModel(FakeMessagesListChatModel, ChatOpenAI): + pass + + # Final message has no tool_calls → the manager routes to END without delegating. + fake_llm = _FakeModel(responses=[AIMessage(content="لماذا...")]) + + cfg = OpenAiCompatibleConfig(name="agent_llm", model_id="fake", url="null") + joke = StringProperty(title="joke") + translated = StringProperty(title="translated") + + manager = Agent( + name="manager", + llm_config=cfg, + system_prompt="Translate the following to Arabic:\n\n{{joke}}", + outputs=[translated], + ) + worker = Agent(name="worker", llm_config=cfg, system_prompt="You translate.") + mw = ManagerWorkers(name="translator", group_manager=manager, workers=[worker]) + # The manager node exposes the group manager's `joke` input, and the single + # `translated` output (inherited from the group manager) for the leaf edge. + assert [p.title for p in (mw.inputs or [])] == ["joke"] + + manager_node = AgentNode(name="manager_node", agent=mw) + start_node = StartNode(name="start", inputs=[joke]) + end_node = EndNode(name="end", outputs=[translated]) + flow = Flow( + name="flow", + start_node=start_node, + nodes=[start_node, manager_node, end_node], + control_flow_connections=[ + ControlFlowEdge(name="start_to_node", from_node=start_node, to_node=manager_node), + ControlFlowEdge(name="node_to_end", from_node=manager_node, to_node=end_node), + ], + data_flow_connections=[ + DataFlowEdge( + name="joke_edge", + source_node=start_node, + source_output=joke.title, + destination_node=manager_node, + destination_input=joke.title, + ), + DataFlowEdge( + name="translated_edge", + source_node=manager_node, + source_output=translated.title, + destination_node=end_node, + destination_input=translated.title, + ), + ], + outputs=[translated], + ) + + loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver()) + with patch.object( + AgentSpecToLangGraphConverter, + "_llm_convert_to_langgraph", + autospec=True, + side_effect=lambda self_obj, llm_config, *a, **k: fake_llm, + ), patch.object( + FakeMessagesListChatModel, + "bind_tools", + new=lambda self_obj, *a, **k: self_obj, + ): + compiled = loader.load_component(flow) + result = compiled.invoke( + { + "inputs": {"joke": "Why did the car..."}, + "messages": [{"role": "user", "content": ""}], + }, + {"configurable": {"thread_id": "managerworkers-node"}}, + ) + + assert "outputs" in result + assert result["outputs"]["translated"] == "لماذا..." diff --git a/pyagentspec/tests/adapters/langgraph/flows/test_swarm_node.py b/pyagentspec/tests/adapters/langgraph/flows/test_swarm_node.py new file mode 100644 index 00000000..992c9200 --- /dev/null +++ b/pyagentspec/tests/adapters/langgraph/flows/test_swarm_node.py @@ -0,0 +1,137 @@ +# Copyright © 2026 Oracle and/or its affiliates. +# +# This software is under the Apache License 2.0 +# (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. + +"""A Swarm used as a flow step (AgentNode). + +Regression coverage for two coupled behaviours, symmetric with the ManagerWorkers +flow-step coverage in ``test_managerworkers_node.py``: + * ``Swarm._get_inferred_inputs`` exposes the entry agent's (``first_agent``) inputs, so a + flow ``AgentNode`` wrapping a swarm declares input ports and a ``DataFlowEdge`` into it + resolves at load (previously: "node does not have any input property..."). + * ``AgentNodeExecutor`` runs a Swarm node (previously: TypeError "can only be used with + AgentSpecAgent agents"), rendering the node inputs into the entry agent's prompt and + returning its result. +""" + +from pyagentspec.agent import Agent +from pyagentspec.property import StringProperty +from pyagentspec.swarm import Swarm + + +def test_swarm_infers_inputs_from_first_agent_prompt() -> None: + """A Swarm exposes the entry agent's prompt placeholders as inputs.""" + from pyagentspec.llms.openaicompatibleconfig import OpenAiCompatibleConfig + + cfg = OpenAiCompatibleConfig(name="m", model_id="fake", url="null") + first = Agent( + name="first", + llm_config=cfg, + system_prompt="Translate the following to Arabic:\n\n{{joke}}\n\nMake {{count}} variants.", + ) + second = Agent(name="second", llm_config=cfg, system_prompt="You translate.") + swarm = Swarm(name="swarm", first_agent=first, relationships=[(first, second)]) + + assert sorted(p.title for p in (swarm.inputs or [])) == ["count", "joke"] + + +def test_swarm_runs_as_a_flow_step_with_data_edge_inputs() -> None: + """A Swarm flow step loads (data edge resolves) and executes offline. + + The model is stubbed (no real LLM, no handoff), so the entry agent produces a final + message and the swarm routes straight to END. Asserts the flow both loads — proving the + swarm node exposes the ``joke`` input the data edge targets — and runs, surfacing the + entry agent's answer as the node's single string output. + """ + from unittest.mock import patch + + from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel + from langchain_core.messages import AIMessage + from langchain_openai import ChatOpenAI + from langgraph.checkpoint.memory import MemorySaver + + from pyagentspec.adapters.langgraph import AgentSpecLoader + from pyagentspec.adapters.langgraph._langgraphconverter import ( + AgentSpecToLangGraphConverter, + ) + from pyagentspec.flows.edges import ControlFlowEdge, DataFlowEdge + from pyagentspec.flows.flow import Flow + from pyagentspec.flows.nodes import AgentNode, EndNode, StartNode + from pyagentspec.llms.openaicompatibleconfig import OpenAiCompatibleConfig + + class _FakeModel(FakeMessagesListChatModel, ChatOpenAI): + pass + + # Final message has no tool_calls → the entry agent answers without handing off. + fake_llm = _FakeModel(responses=[AIMessage(content="لماذا...")]) + + cfg = OpenAiCompatibleConfig(name="agent_llm", model_id="fake", url="null") + joke = StringProperty(title="joke") + translated = StringProperty(title="translated") + + first = Agent( + name="first", + llm_config=cfg, + system_prompt="Translate the following to Arabic:\n\n{{joke}}", + outputs=[translated], + ) + second = Agent(name="second", llm_config=cfg, system_prompt="You translate.") + swarm = Swarm(name="translator", first_agent=first, relationships=[(first, second)]) + # The swarm node exposes the entry agent's `joke` input, and its single `translated` + # output (inherited from the entry agent) for the leaf edge. + assert [p.title for p in (swarm.inputs or [])] == ["joke"] + + swarm_node = AgentNode(name="swarm_node", agent=swarm) + start_node = StartNode(name="start", inputs=[joke]) + end_node = EndNode(name="end", outputs=[translated]) + flow = Flow( + name="flow", + start_node=start_node, + nodes=[start_node, swarm_node, end_node], + control_flow_connections=[ + ControlFlowEdge(name="start_to_node", from_node=start_node, to_node=swarm_node), + ControlFlowEdge(name="node_to_end", from_node=swarm_node, to_node=end_node), + ], + data_flow_connections=[ + DataFlowEdge( + name="joke_edge", + source_node=start_node, + source_output=joke.title, + destination_node=swarm_node, + destination_input=joke.title, + ), + DataFlowEdge( + name="translated_edge", + source_node=swarm_node, + source_output=translated.title, + destination_node=end_node, + destination_input=translated.title, + ), + ], + outputs=[translated], + ) + + loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver()) + with patch.object( + AgentSpecToLangGraphConverter, + "_llm_convert_to_langgraph", + autospec=True, + side_effect=lambda self_obj, llm_config, *a, **k: fake_llm, + ), patch.object( + FakeMessagesListChatModel, + "bind_tools", + new=lambda self_obj, *a, **k: self_obj, + ): + compiled = loader.load_component(flow) + result = compiled.invoke( + { + "inputs": {"joke": "Why did the car..."}, + "messages": [{"role": "user", "content": ""}], + }, + {"configurable": {"thread_id": "swarm-node"}}, + ) + + assert "outputs" in result + assert result["outputs"]["translated"] == "لماذا..." diff --git a/pyagentspec/tests/adapters/langgraph/test_managerworkers.py b/pyagentspec/tests/adapters/langgraph/test_managerworkers.py new file mode 100644 index 00000000..d9be055c --- /dev/null +++ b/pyagentspec/tests/adapters/langgraph/test_managerworkers.py @@ -0,0 +1,1562 @@ +# Copyright © 2025 Oracle and/or its affiliates. +# +# This software is under the Apache License 2.0 +# (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. + +"""Offline tests for the LangGraph ``ManagerWorkers`` converter. + +These cover the hierarchical topology, roster prompt rendering, the +worker-isolation invariant (each worker sees only its delegated task), +and the recursive nesting case. The LLM is stubbed with +``FakeMessagesListChatModel`` so the tests run without network or model +endpoints. +""" + +from typing import Any +from unittest.mock import patch + +import pytest + +# ─── Shared helpers ────────────────────────────────────────────────────────── + + +def _llm_cfg(name: str) -> Any: + from pyagentspec.llms.openaicompatibleconfig import OpenAiCompatibleConfig + + return OpenAiCompatibleConfig(name=name, model_id="fake", url="null") + + +def _fake_manager(*ai_responses: Any) -> Any: + """A FakeMessagesListChatModel subclassed under ChatOpenAI so the + manager's react-agent treats it as an OpenAI-style chat model.""" + from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel + from langchain_openai import ChatOpenAI + + class _FakeModel(FakeMessagesListChatModel, ChatOpenAI): + pass + + return _FakeModel(responses=list(ai_responses)) + + +# ─── Pure-helper unit tests (no LLM) ──────────────────────────────────────── + + +def test_safe_node_name_lowercases_and_collapses_punctuation() -> None: + from pyagentspec.adapters.langgraph._managerworkers import ( + _safe_node_name, + ) + + assert _safe_node_name("Research Helper", "id-1") == "research_helper" + assert _safe_node_name("My-Worker!! v2", "id-1") == "my_worker_v2" + + +def test_safe_node_name_falls_back_to_normalized_id() -> None: + from pyagentspec.adapters.langgraph._managerworkers import ( + _safe_node_name, + ) + + # Name slugifies to empty → id used (and also normalized). + assert _safe_node_name("!!!", "sub-1") == "sub_1" + # Both empty → constant fallback. + assert _safe_node_name("", "") == "worker" + + +def test_append_workers_roster_appends_block_after_existing_prompt() -> None: + from pyagentspec.adapters.langgraph._managerworkers import ( + _append_workers_roster, + ) + + out = _append_workers_roster( + "Coordinate the team.", + [("research_helper", "Handles research"), ("drafter", "Drafts text")], + ) + assert out == ( + "Coordinate the team.\n\n" + "Available workers:\n" + "- research_helper: Handles research\n" + "- drafter: Drafts text" + ) + + +def test_append_workers_roster_flattens_multiline_descriptions() -> None: + from pyagentspec.adapters.langgraph._managerworkers import ( + _append_workers_roster, + ) + + out = _append_workers_roster( + "", + [("helper", "First line\nsecond line\n third line ")], + ) + # Whitespace flattened so the one-line-per-worker shape survives. + assert out == "Available workers:\n- helper: First line second line third line" + + +def test_route_manager_to_worker_or_end_sends_to_pending_delegation() -> None: + from langchain_core.messages import AIMessage + from langgraph.types import Send + + from pyagentspec.adapters.langgraph._managerworkers import ( + _DELEGATE_CALL_ID_KEY, + _DELEGATE_TASK_KEY, + _route_manager_to_worker_handoff_or_end, + ) + + delegating = AIMessage( + content="", + tool_calls=[{"name": "delegate_to_research_helper", "args": {"task": "hi"}, "id": "c1"}], + ) + sends = _route_manager_to_worker_handoff_or_end({"messages": [delegating]}) + # One delegation → a single Send to the worker node carrying the task + # and the tool_call_id its reply must answer. + assert isinstance(sends, list) and len(sends) == 1 + assert isinstance(sends[0], Send) + assert sends[0].node == "research_helper" + assert sends[0].arg == {_DELEGATE_TASK_KEY: "hi", _DELEGATE_CALL_ID_KEY: "c1"} + + +def test_route_manager_to_worker_or_end_returns_end_when_no_delegation() -> None: + from langchain_core.messages import AIMessage + from langgraph.graph import END + + from pyagentspec.adapters.langgraph._managerworkers import ( + _route_manager_to_worker_handoff_or_end, + ) + + not_delegating = AIMessage(content="Done.", tool_calls=[]) + assert _route_manager_to_worker_handoff_or_end({"messages": [not_delegating]}) == END + assert _route_manager_to_worker_handoff_or_end({"messages": []}) == END + + +def test_route_manager_to_worker_or_end_fans_out_every_delegation() -> None: + from langchain_core.messages import AIMessage + from langgraph.types import Send + + from pyagentspec.adapters.langgraph._managerworkers import ( + _DELEGATE_CALL_ID_KEY, + _DELEGATE_TASK_KEY, + _route_manager_to_worker_handoff_or_end, + ) + + msg = AIMessage( + content="", + tool_calls=[ + {"name": "some_other_tool", "args": {}, "id": "c0"}, + {"name": "delegate_to_drafter", "args": {"task": "x"}, "id": "c1"}, + {"name": "delegate_to_research_helper", "args": {"task": "y"}, "id": "c2"}, + ], + ) + sends = _route_manager_to_worker_handoff_or_end({"messages": [msg]}) + # Every delegation gets its own Send so each tool_call_id is answered. + # The non-delegation tool call was already executed inside the manager's + # react loop and is ignored by routing. + assert all(isinstance(s, Send) for s in sends) + assert [s.node for s in sends] == ["drafter", "research_helper"] + assert [s.arg[_DELEGATE_CALL_ID_KEY] for s in sends] == ["c1", "c2"] + assert [s.arg[_DELEGATE_TASK_KEY] for s in sends] == ["x", "y"] + + +# ─── Topology test (no LLM execution; checks compiled graph shape) ────────── + + +def test_manager_workers_compiles_to_hierarchical_graph_topology() -> None: + from langchain_core.messages import AIMessage + from langgraph.checkpoint.memory import MemorySaver + from langgraph.graph import START + + from pyagentspec.adapters.langgraph import AgentSpecLoader + from pyagentspec.adapters.langgraph._langgraphconverter import ( + AgentSpecToLangGraphConverter, + ) + from pyagentspec.adapters.langgraph._managerworkers import ( + _MANAGER_NODE_KEY, + ) + from pyagentspec.agent import Agent + from pyagentspec.managerworkers import ManagerWorkers + + manager_agent = Agent( + name="Coordinator", + description="Coordinates", + system_prompt="Coordinate the team.", + llm_config=_llm_cfg("manager_llm"), + ) + worker_a = Agent( + name="Research Helper", + description="Handles research", + system_prompt="Research.", + llm_config=_llm_cfg("worker_a_llm"), + ) + worker_b = Agent( + name="Drafter", + description="Drafts text", + system_prompt="Draft.", + llm_config=_llm_cfg("worker_b_llm"), + ) + mw = ManagerWorkers( + name="ResearchTeam", + group_manager=manager_agent, + workers=[worker_a, worker_b], + ) + + fake_llm = _fake_manager(AIMessage(content="Done.")) + loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver()) + with patch.object( + AgentSpecToLangGraphConverter, "_llm_convert_to_langgraph", return_value=fake_llm + ): + compiled = loader.load_component(mw) + + # The compiled object is a CompiledStateGraph; its builder exposes + # the parent topology we expect. + builder = compiled.builder + assert _MANAGER_NODE_KEY in builder.nodes + assert "research_helper" in builder.nodes + assert "drafter" in builder.nodes + + # START → manager; every worker → manager (loop). + edge_pairs = {(src, dst) for src, dst in builder.edges} + assert (START, _MANAGER_NODE_KEY) in edge_pairs + assert ("research_helper", _MANAGER_NODE_KEY) in edge_pairs + assert ("drafter", _MANAGER_NODE_KEY) in edge_pairs + + # The manager → worker routing is a conditional edge (branch), not a + # plain edge — branches are stored separately on the builder. + branches = builder.branches.get(_MANAGER_NODE_KEY) or {} + assert branches, "expected a conditional branch from the manager node" + + +def test_manager_workers_renders_workers_roster_into_manager_prompt() -> None: + """The manager's system prompt gets the ``Available workers:`` block + appended so the LLM knows which delegation tool maps to which worker. + """ + from langchain_core.messages import AIMessage + from langgraph.checkpoint.memory import MemorySaver + + from pyagentspec.adapters.langgraph import AgentSpecLoader + from pyagentspec.adapters.langgraph._langgraphconverter import ( + AgentSpecToLangGraphConverter, + ) + from pyagentspec.adapters.langgraph._managerworkers import ( + _MANAGER_NODE_KEY, + ) + from pyagentspec.agent import Agent + from pyagentspec.managerworkers import ManagerWorkers + + manager_agent = Agent( + name="Coordinator", + description="Coordinates", + system_prompt="Coordinate the team.", + llm_config=_llm_cfg("manager_llm"), + ) + worker = Agent( + name="Research Helper", + description="Handles research tasks", + system_prompt="Research.", + llm_config=_llm_cfg("worker_llm"), + ) + mw = ManagerWorkers( + name="Team", + group_manager=manager_agent, + workers=[worker], + ) + + fake_llm = _fake_manager(AIMessage(content="Done.")) + loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver()) + with patch.object( + AgentSpecToLangGraphConverter, "_llm_convert_to_langgraph", return_value=fake_llm + ): + compiled = loader.load_component(mw) + + # The manager react-agent is itself a subgraph; its create_agent + # middleware stack carries the rendered system prompt as the + # first message of every turn. Walk the manager subgraph's pre-model + # hook chain to find it. + manager_subgraph = compiled.builder.nodes[_MANAGER_NODE_KEY].runnable + # `create_agent` builds a graph whose system message generation + # wraps the prompt — easier to assert by re-rendering it through the + # same helper used by the converter and checking the *intent*. + from pyagentspec.adapters.langgraph._managerworkers import ( + _append_workers_roster, + ) + + expected = _append_workers_roster( + "Coordinate the team.", + [("research_helper", "Handles research tasks")], + ) + assert "Available workers:" in expected + assert "- research_helper: Handles research tasks" in expected + # And the compiled manager carries the delegation tool the prompt + # advertises, proving the LLM has the matching contract. + tools_node = manager_subgraph.builder.nodes["tools"].runnable + assert "delegate_to_research_helper" in tools_node.tools_by_name + + +# ─── End-to-end execution test (offline, fake LLM emitting delegation) ────── + + +def test_manager_workers_delegates_and_routes_back_with_tool_message() -> None: + """End-to-end: manager LLM emits a delegate_to_ tool call, + the parent graph routes to the worker subgraph (which runs with an + isolated message context), the worker's final AIMessage content is + surfaced back to the manager as a ToolMessage matched to the + pending tool_call_id, and the manager's next turn (no tool call) + terminates the graph. This is the load-bearing path that proves the + subgraph composition actually works.""" + from langchain_core.messages import AIMessage, HumanMessage + from langgraph.checkpoint.memory import MemorySaver + + from pyagentspec.adapters.langgraph import AgentSpecLoader + from pyagentspec.adapters.langgraph._langgraphconverter import ( + AgentSpecToLangGraphConverter, + ) + from pyagentspec.agent import Agent + from pyagentspec.managerworkers import ManagerWorkers + + manager_agent = Agent( + name="Coordinator", + description="Coordinates", + system_prompt="You coordinate.", + llm_config=_llm_cfg("manager_llm"), + ) + worker = Agent( + name="Research Helper", + description="Handles research", + system_prompt="You research.", + llm_config=_llm_cfg("worker_llm"), + ) + mw = ManagerWorkers( + name="Team", + group_manager=manager_agent, + workers=[worker], + ) + + # Manager turn 1: delegate to research_helper. + # Manager turn 2: produce final answer (no tool call → END). + manager_responses = [ + AIMessage( + content="", + tool_calls=[ + { + "name": "delegate_to_research_helper", + "args": {"task": "Look up Saturn"}, + "id": "call_1", + } + ], + ), + AIMessage(content="The worker reports: Saturn has rings."), + ] + # Worker turn 1: produce its own final answer. + worker_responses = [AIMessage(content="Saturn has rings.")] + + fake_manager = _fake_manager(*manager_responses) + fake_worker = _fake_manager(*worker_responses) + + def _dispatch(self_obj: Any, llm_config: Any, *args: Any, **kwargs: Any) -> Any: + if llm_config.name == "manager_llm": + return fake_manager + if llm_config.name == "worker_llm": + return fake_worker + raise AssertionError(f"unexpected llm_config: {llm_config.name}") + + # ``create_agent`` calls ``model.bind_tools(...)``. ``FakeMessagesListChatModel`` + # inherits ``bind_tools`` from real ``ChatOpenAI``, which calls out to + # OpenAI. Patch the class method so binding is a no-op that returns the + # same fake (preserving its response queue). + from langchain_core.language_models.fake_chat_models import ( + FakeMessagesListChatModel, + ) + + loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver()) + with patch.object( + AgentSpecToLangGraphConverter, + "_llm_convert_to_langgraph", + autospec=True, + side_effect=_dispatch, + ), patch.object( + FakeMessagesListChatModel, + "bind_tools", + new=lambda self_obj, *a, **kw: self_obj, + ): + compiled = loader.load_component(mw) + + # Use the sync invocation path: ``FakeMessagesListChatModel`` provides + # a sync ``_generate`` (returns queued responses) but no async + # override, so MRO resolves ``_agenerate`` to the real + # ``ChatOpenAI._agenerate`` which calls the OpenAI API. The worker + # wrapper exposes both sync and async via RunnableLambda; LangGraph + # picks the sync path here. + result = compiled.invoke( + {"messages": [HumanMessage(content="Tell me about Saturn.")]}, + {"configurable": {"thread_id": "mw-1"}}, + ) + messages = result["messages"] + + # The end state should contain: user input, manager's delegation + # AIMessage, the synthesized ToolMessage (worker's reply), and the + # manager's final AIMessage. + msg_types = [type(m).__name__ for m in messages] + assert "HumanMessage" in msg_types + assert "ToolMessage" in msg_types + # Final message is the manager's terminating AIMessage. + assert isinstance(messages[-1], AIMessage) + assert "Saturn has rings" in messages[-1].content + + # And the ToolMessage carries the worker's reply matched to the + # pending delegation tool_call_id — proves the isolation wrapper + # threaded the call id through. + tool_msgs = [m for m in messages if type(m).__name__ == "ToolMessage"] + assert tool_msgs and tool_msgs[0].tool_call_id == "call_1" + assert "Saturn has rings" in tool_msgs[0].content + + +def test_manager_workers_answers_every_delegation_in_a_single_turn() -> None: + """Regression: when the manager emits SEVERAL ``delegate_to_`` + tool calls in one turn (e.g. "spin up 5 sub-agents"), every delegation + must run and be answered by its own ToolMessage matched to the + originating tool_call_id. + + Before the fix the parent graph routed only the first delegation, so the + other tool_call_ids were left unanswered — an invalid tool-call / + tool-result sequence that made the manager hallucinate the missing + replies. This asserts all three calls get matched ToolMessages. + """ + from langchain_core.language_models.fake_chat_models import ( + FakeMessagesListChatModel, + ) + from langchain_core.messages import AIMessage, HumanMessage + from langgraph.checkpoint.memory import MemorySaver + + from pyagentspec.adapters.langgraph import AgentSpecLoader + from pyagentspec.adapters.langgraph._langgraphconverter import ( + AgentSpecToLangGraphConverter, + ) + from pyagentspec.agent import Agent + from pyagentspec.managerworkers import ManagerWorkers + + manager_agent = Agent( + name="Coordinator", + description="Coordinates", + system_prompt="You coordinate.", + llm_config=_llm_cfg("manager_llm"), + ) + worker = Agent( + name="Sub Agent", + description="Writes poems", + system_prompt="You write poems.", + llm_config=_llm_cfg("worker_llm"), + ) + mw = ManagerWorkers(name="Team", group_manager=manager_agent, workers=[worker]) + + # Turn 1: three delegations to the SAME worker in one AIMessage. + # Turn 2: terminate (no tool call). + manager_responses = [ + AIMessage( + content="", + tool_calls=[ + {"name": "delegate_to_sub_agent", "args": {"task": "Spanish poem"}, "id": "call_1"}, + {"name": "delegate_to_sub_agent", "args": {"task": "French poem"}, "id": "call_2"}, + {"name": "delegate_to_sub_agent", "args": {"task": "German poem"}, "id": "call_3"}, + ], + ), + AIMessage(content="Here are your three poems."), + ] + # Each worker invocation pops one reply; provide enough for the fan-out. + worker_responses = [AIMessage(content=f"poem #{i}") for i in range(1, 6)] + + fake_manager = _fake_manager(*manager_responses) + fake_worker = _fake_manager(*worker_responses) + + def _dispatch(self_obj: Any, llm_config: Any, *args: Any, **kwargs: Any) -> Any: + if llm_config.name == "manager_llm": + return fake_manager + if llm_config.name == "worker_llm": + return fake_worker + raise AssertionError(f"unexpected llm_config: {llm_config.name}") + + loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver()) + with patch.object( + AgentSpecToLangGraphConverter, + "_llm_convert_to_langgraph", + autospec=True, + side_effect=_dispatch, + ), patch.object( + FakeMessagesListChatModel, + "bind_tools", + new=lambda self_obj, *a, **kw: self_obj, + ): + compiled = loader.load_component(mw) + + result = compiled.invoke( + {"messages": [HumanMessage(content="Write 3 poems via sub-agents.")]}, + {"configurable": {"thread_id": "mw-multi"}}, + ) + messages = result["messages"] + + # Every delegation tool_call_id must be answered by exactly one ToolMessage. + requested = { + tc["id"] + for m in messages + if isinstance(m, AIMessage) + for tc in (m.tool_calls or []) + if tc["name"].startswith("delegate_to_") + } + answered = [m.tool_call_id for m in messages if type(m).__name__ == "ToolMessage"] + assert requested == {"call_1", "call_2", "call_3"} + assert sorted(answered) == [ + "call_1", + "call_2", + "call_3", + ], f"unanswered delegations: {requested - set(answered)}" + # No duplicate replies, and each carries a worker poem. + assert len(answered) == 3 + tool_msgs = [m for m in messages if type(m).__name__ == "ToolMessage"] + assert all(m.content.startswith("poem #") for m in tool_msgs) + + +# ─── Recursive nesting ────────────────────────────────────────────────────── + + +def test_nested_manager_workers_compiles_recursively() -> None: + """A worker that is itself a ManagerWorkers compiles through the + same dispatch — the inner ManagerWorkers becomes a CompiledStateGraph + that the outer parent graph wires in as a subgraph node.""" + from langchain_core.messages import AIMessage + from langgraph.checkpoint.memory import MemorySaver + + from pyagentspec.adapters.langgraph import AgentSpecLoader + from pyagentspec.adapters.langgraph._langgraphconverter import ( + AgentSpecToLangGraphConverter, + ) + from pyagentspec.agent import Agent + from pyagentspec.managerworkers import ManagerWorkers + + leaf = Agent( + name="Leaf", + description="Leaf task", + system_prompt="Leaf.", + llm_config=_llm_cfg("leaf_llm"), + ) + inner_manager = Agent( + name="InnerManager", + description="Inner", + system_prompt="Manage leaves.", + llm_config=_llm_cfg("inner_llm"), + ) + inner_mw = ManagerWorkers( + name="Inner", + group_manager=inner_manager, + workers=[leaf], + ) + outer_manager = Agent( + name="OuterManager", + description="Outer", + system_prompt="Manage subteams.", + llm_config=_llm_cfg("outer_llm"), + ) + outer_mw = ManagerWorkers( + name="Outer", + group_manager=outer_manager, + workers=[inner_mw], + ) + + fake_llm = _fake_manager(AIMessage(content="Done.")) + loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver()) + with patch.object( + AgentSpecToLangGraphConverter, "_llm_convert_to_langgraph", return_value=fake_llm + ): + compiled = loader.load_component(outer_mw) + + # Outer parent graph has a node for the inner ManagerWorkers worker. + assert "inner" in compiled.builder.nodes + + +def test_rejects_non_agent_group_manager() -> None: + """ManagerWorkers.group_manager must be an Agent — pyagentspec allows + any AgenticComponent but the LangGraph adapter needs a chat-LLM that + emits tool_calls to decide which worker to delegate to.""" + from langgraph.checkpoint.memory import MemorySaver + + from pyagentspec.adapters.langgraph import AgentSpecLoader + from pyagentspec.agent import Agent + from pyagentspec.managerworkers import ManagerWorkers + + # Use a nested ManagerWorkers as the group_manager — a valid + # AgenticComponent per pyagentspec validators, unsupported here. + leaf = Agent( + name="Leaf", + description="L", + system_prompt="L.", + llm_config=_llm_cfg("l"), + ) + inner_manager = Agent( + name="Inner", + description="I", + system_prompt="I.", + llm_config=_llm_cfg("i"), + ) + inner_mw = ManagerWorkers( + name="Inner", + group_manager=inner_manager, + workers=[leaf], + ) + outer_mw = ManagerWorkers( + name="Outer", + group_manager=inner_mw, + workers=[ + Agent(name="Other", description="O", system_prompt="O.", llm_config=_llm_cfg("o")), + ], + ) + loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver()) + with pytest.raises(NotImplementedError, match="group_manager must be an Agent"): + loader.load_component(outer_mw) + + +# ─── Worker name collision ────────────────────────────────────────────────── + + +def test_workers_with_name_slug_collision_are_rejected() -> None: + """Two workers whose names normalize to the same node identifier + would silently overwrite each other in the parent graph; raise at + load time instead.""" + from langgraph.checkpoint.memory import MemorySaver + + from pyagentspec.adapters.langgraph import AgentSpecLoader + from pyagentspec.agent import Agent + from pyagentspec.managerworkers import ManagerWorkers + + a = Agent(name="Helper A", description="x", system_prompt=".", llm_config=_llm_cfg("a")) + b = Agent(name="helper-a", description="x", system_prompt=".", llm_config=_llm_cfg("b")) + # Both normalize to "helper_a". + mw = ManagerWorkers( + name="T", + group_manager=Agent( + name="M", + description="m", + system_prompt=".", + llm_config=_llm_cfg("m"), + ), + workers=[a, b], + ) + + loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver()) + with pytest.raises(ValueError, match="collide after normalization"): + loader.load_component(mw) + + +# ─── astream_events delegation scrubbing ───────────────────────────────────── +# +# The manager routes by emitting a ``delegate_to_`` tool call which +# the worker answers with a ToolMessage. That pair is internal plumbing; the +# consumer-facing ``astream_events`` view must not surface it as phantom tool +# calls. ``_DelegationEventFilter`` scrubs the event stream while leaving the +# graph's message state intact. + + +def test_delegation_filter_drops_delegate_tool_lifecycle_events() -> None: + from pyagentspec.adapters.langgraph._managerworkers import ( + _DelegationEventFilter, + ) + + f = _DelegationEventFilter() + for etype in ("on_tool_start", "on_tool_end", "on_tool_error"): + ev = { + "event": etype, + "name": "delegate_to_research_helper", + "run_id": "r", + "data": {}, + } + assert f.scrub(ev) is None + + # A real tool's lifecycle events pass through untouched. + real = {"event": "on_tool_start", "name": "search", "run_id": "r", "data": {}} + assert f.scrub(real) is real + + +def test_delegation_filter_strips_delegate_call_from_chat_model_end() -> None: + from langchain_core.messages import AIMessage + + from pyagentspec.adapters.langgraph._managerworkers import ( + _DelegationEventFilter, + ) + + f = _DelegationEventFilter() + msg = AIMessage( + content="", + tool_calls=[ + {"name": "delegate_to_research_helper", "args": {"task": "hi"}, "id": "call_1"}, + ], + ) + out = f.scrub( + {"event": "on_chat_model_end", "name": "x", "run_id": "r", "data": {"output": msg}} + ) + assert out is not None + assert out["data"]["output"].tool_calls == [] + # The original message object (which lives in graph state) is untouched. + assert msg.tool_calls and msg.tool_calls[0]["name"] == "delegate_to_research_helper" + + # A turn that mixes a delegation call with a real tool call keeps the real one. + mixed = AIMessage( + content="ok", + tool_calls=[ + {"name": "delegate_to_research_helper", "args": {"task": "hi"}, "id": "call_2"}, + {"name": "search", "args": {"q": "x"}, "id": "call_3"}, + ], + ) + out2 = f.scrub( + {"event": "on_chat_model_end", "name": "x", "run_id": "r", "data": {"output": mixed}} + ) + assert [tc["name"] for tc in out2["data"]["output"].tool_calls] == ["search"] + + +def test_delegation_filter_strips_streamed_delegate_tool_call_chunks() -> None: + from langchain_core.messages import AIMessageChunk + + from pyagentspec.adapters.langgraph._managerworkers import ( + _DelegationEventFilter, + ) + + f = _DelegationEventFilter() + # Opening chunk: names the delegation tool at index 0 → pure plumbing, dropped. + opening = AIMessageChunk( + content="", + tool_call_chunks=[ + { + "name": "delegate_to_research_helper", + "args": "", + "id": "call_1", + "index": 0, + "type": "tool_call_chunk", + } + ], + ) + assert ( + f.scrub( + { + "event": "on_chat_model_stream", + "name": "x", + "run_id": "r", + "data": {"chunk": opening}, + } + ) + is None + ) + + # Argument-continuation chunk: no name, same index 0 → also dropped. + cont = AIMessageChunk( + content="", + tool_call_chunks=[ + { + "name": None, + "args": '{"task":"hi"}', + "id": None, + "index": 0, + "type": "tool_call_chunk", + }, + ], + ) + assert ( + f.scrub( + {"event": "on_chat_model_stream", "name": "x", "run_id": "r", "data": {"chunk": cont}} + ) + is None + ) + + # A chunk mixing a delegation call with a real tool call keeps the real one. + mixed = AIMessageChunk( + content="", + tool_call_chunks=[ + { + "name": "delegate_to_research_helper", + "args": "", + "id": "c4", + "index": 0, + "type": "tool_call_chunk", + }, + {"name": "search", "args": "", "id": "c5", "index": 1, "type": "tool_call_chunk"}, + ], + ) + out = f.scrub( + {"event": "on_chat_model_stream", "name": "x", "run_id": "r2", "data": {"chunk": mixed}} + ) + assert out is not None + kept = out["data"]["chunk"].tool_call_chunks + assert [c["name"] for c in kept] == ["search"] + + +def test_delegation_filter_drops_worker_synthetic_tool_message() -> None: + from langchain_core.messages import AIMessage, ToolMessage + + from pyagentspec.adapters.langgraph._managerworkers import ( + _DelegationEventFilter, + ) + + f = _DelegationEventFilter() + # The manager's delegate turn first records the delegation call id. + delegate = AIMessage( + content="", + tool_calls=[ + {"name": "delegate_to_research_helper", "args": {"task": "hi"}, "id": "call_1"}, + ], + ) + f.scrub( + {"event": "on_chat_model_end", "name": "x", "run_id": "r", "data": {"output": delegate}} + ) + + # The worker node then emits its reply as a ToolMessage matched to call_1. + reply = ToolMessage(content="Saturn has rings.", tool_call_id="call_1") + out = f.scrub( + { + "event": "on_chain_end", + "name": "worker:research_helper", + "run_id": "w", + "data": {"output": {"messages": [reply]}}, + } + ) + assert out is None + + # A ToolMessage answering an unknown (real) tool call is preserved. + other = ToolMessage(content="x", tool_call_id="call_other") + kept = f.scrub( + { + "event": "on_chain_end", + "name": "node", + "run_id": "w2", + "data": {"output": {"messages": [other]}}, + } + ) + assert kept is not None + assert kept["data"]["output"]["messages"] == [other] + + +def test_delegation_filter_strips_delegate_calls_from_state_snapshot() -> None: + """Regression: a node/state payload (``on_chain_end``) carrying the full + ``messages`` list must surface NEITHER the delegate tool calls NOR their + reply ToolMessages. + + A consumer builds its message snapshot from this payload and reads + ``tool_calls`` straight off the AIMessage. If the filter dropped only the + reply ToolMessages but left the delegate tool calls on the AIMessage, the + snapshot would show delegate tool calls with no results — rendered as a + "tool call with no result". Real (non-delegation) tool calls and their + results must be preserved. + """ + from langchain_core.messages import AIMessage, HumanMessage, ToolMessage + + from pyagentspec.adapters.langgraph._managerworkers import ( + _DelegationEventFilter, + ) + + f = _DelegationEventFilter() + # Manager's delegation turn streams first so the filter learns the ids. + delegate_ai = AIMessage( + content="", + tool_calls=[ + {"name": "delegate_to_sub_agent", "args": {"task": "ES"}, "id": "call_1"}, + {"name": "delegate_to_sub_agent", "args": {"task": "FR"}, "id": "call_2"}, + ], + ) + f.scrub( + {"event": "on_chat_model_end", "name": "x", "run_id": "r", "data": {"output": delegate_ai}} + ) + + # The final-state snapshot carries the whole conversation, including a + # real tool call ("search") + its result that must survive. + snapshot = { + "messages": [ + HumanMessage(content="2 poems via sub-agents"), + delegate_ai, + ToolMessage(content="poem ES", tool_call_id="call_1"), + ToolMessage(content="poem FR", tool_call_id="call_2"), + AIMessage( + content="", + tool_calls=[{"name": "search", "args": {"q": "x"}, "id": "call_real"}], + ), + ToolMessage(content="search result", tool_call_id="call_real"), + AIMessage(content="Here are your poems."), + ] + } + out = f.scrub( + { + "event": "on_chain_end", + "name": "__manager__", + "run_id": "r2", + "data": {"output": snapshot}, + } + ) + assert out is not None + msgs = out["data"]["output"]["messages"] + + # No delegate tool calls and no delegate ToolMessages remain. + delegate_calls = [ + tc + for m in msgs + if isinstance(m, AIMessage) + for tc in (m.tool_calls or []) + if tc["name"].startswith("delegate_to_") + ] + assert delegate_calls == [] + tool_ids = [m.tool_call_id for m in msgs if type(m).__name__ == "ToolMessage"] + assert "call_1" not in tool_ids and "call_2" not in tool_ids + # The empty delegation AIMessage is dropped entirely. + assert delegate_ai not in msgs + + # The REAL tool call + its result are preserved and still paired. + real_calls = [tc["id"] for m in msgs if isinstance(m, AIMessage) for tc in (m.tool_calls or [])] + assert real_calls == ["call_real"] + assert "call_real" in tool_ids + # The human turn and the manager's final answer survive. + assert any(type(m).__name__ == "HumanMessage" for m in msgs) + assert msgs[-1].content == "Here are your poems." + + # The original state objects are never mutated (graph state stays intact). + assert delegate_ai.tool_calls and len(delegate_ai.tool_calls) == 2 + + +def test_delegation_filter_passes_through_real_content() -> None: + from langchain_core.messages import AIMessageChunk + + from pyagentspec.adapters.langgraph._managerworkers import ( + _DelegationEventFilter, + ) + + f = _DelegationEventFilter() + chunk = AIMessageChunk(content="Hello") + ev = {"event": "on_chat_model_stream", "name": "x", "run_id": "r", "data": {"chunk": chunk}} + out = f.scrub(ev) + # No delegation artifact → event passes through as the same object. + assert out is ev + assert out["data"]["chunk"].content == "Hello" + + +def test_delegation_filter_strips_delegate_from_invalid_tool_calls() -> None: + """A delegation call whose args failed to parse arrives in + ``invalid_tool_calls`` rather than ``tool_calls`` — it must still be + scrubbed so the consumer never sees the routing protocol.""" + from langchain_core.messages import AIMessage + + from pyagentspec.adapters.langgraph._managerworkers import _DelegationEventFilter + + f = _DelegationEventFilter() + msg = AIMessage( + content="", + invalid_tool_calls=[ + { + "name": "delegate_to_research_helper", + "args": "{bad", + "id": "call_1", + "error": "parse error", + "type": "invalid_tool_call", + } + ], + ) + out = f.scrub( + {"event": "on_chat_model_end", "name": "x", "run_id": "r", "data": {"output": msg}} + ) + assert out is not None + assert out["data"]["output"].invalid_tool_calls == [] + # The original (graph-state) message is never mutated. + assert ( + msg.invalid_tool_calls + and msg.invalid_tool_calls[0]["name"] == "delegate_to_research_helper" + ) + + +def test_delegation_filter_strips_provider_native_tool_calls_on_full_message() -> None: + """Some providers (e.g. OpenAI) carry the tool call only in + ``additional_kwargs['tool_calls']``; a delegation call there must be + stripped and its id recorded so the worker reply can later be dropped.""" + from langchain_core.messages import AIMessage, ToolMessage + + from pyagentspec.adapters.langgraph._managerworkers import _DelegationEventFilter + + f = _DelegationEventFilter() + msg = AIMessage( + content="", + additional_kwargs={ + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "function": {"name": "delegate_to_research_helper", "arguments": ""}, + "type": "function", + } + ] + }, + ) + out = f.scrub( + {"event": "on_chat_model_end", "name": "x", "run_id": "r", "data": {"output": msg}} + ) + assert out is not None + assert "tool_calls" not in (out["data"]["output"].additional_kwargs or {}) + + # Recording the id means the worker's reply ToolMessage is dropped too. + reply = ToolMessage(content="done", tool_call_id="call_1") + dropped = f.scrub( + { + "event": "on_chain_end", + "name": "worker:research_helper", + "run_id": "w", + "data": {"output": {"messages": [reply]}}, + } + ) + assert dropped is None + + +def test_delegation_filter_scrubs_input_payload_messages() -> None: + """The worker's reply ToolMessage must be dropped wherever it surfaces — + including a node's ``input`` payload, not only ``output`` / ``chunk``.""" + from langchain_core.messages import AIMessage, ToolMessage + + from pyagentspec.adapters.langgraph._managerworkers import _DelegationEventFilter + + f = _DelegationEventFilter() + delegate = AIMessage( + content="", + tool_calls=[{"name": "delegate_to_research_helper", "args": {"task": "x"}, "id": "call_1"}], + ) + f.scrub( + {"event": "on_chat_model_end", "name": "x", "run_id": "r", "data": {"output": delegate}} + ) + + reply = ToolMessage(content="done", tool_call_id="call_1") + out = f.scrub( + { + "event": "on_chain_start", + "name": "worker:research_helper", + "run_id": "w", + "data": {"input": {"messages": [reply]}}, + } + ) + # The only message was the delegate reply → payload empties → event dropped. + assert out is None + + +def test_worker_events_stream_natively_namespaced_under_worker_node() -> None: + """Regression: a worker's token events must stream under the worker + node's checkpoint namespace so a consumer can attribute them to the + sub-agent. The wrapper must inherit the ambient run config (no fresh + thread_id); a fresh thread_id detaches the worker into a top-level + ``agent:`` run with no worker prefix, which is unattributable.""" + import asyncio + + from langchain_core.language_models.fake_chat_models import ( + GenericFakeChatModel, + ) + from langchain_core.messages import AIMessage, HumanMessage + from langgraph.graph import END, START, MessagesState, StateGraph + + from pyagentspec.adapters.langgraph._managerworkers import ( + _wrap_worker_for_subgraph, + ) + + # A minimal worker compiled graph that streams some content. + wmodel = GenericFakeChatModel(messages=iter([AIMessage(content="Saturn has rings")] * 9)) + wb = StateGraph(MessagesState) + + async def _wagent(state: Any) -> Any: + return {"messages": [await wmodel.ainvoke(state["messages"])]} + + wb.add_node("agent", _wagent) + wb.add_edge(START, "agent") + wb.add_edge("agent", END) + worker_graph = wb.compile() + + # Parent: a plain manager node emits the delegate tool call, then routes + # to the wrapped worker node named "research_helper". + pb = StateGraph(MessagesState) + + def _manager(state: Any) -> Any: + return { + "messages": [ + AIMessage( + content="", + tool_calls=[ + { + "name": "delegate_to_research_helper", + "args": {"task": "Saturn"}, + "id": "c1", + } + ], + ) + ] + } + + pb.add_node("__manager__", _manager) + pb.add_node("research_helper", _wrap_worker_for_subgraph(worker_graph, "research_helper")) + pb.add_edge(START, "__manager__") + pb.add_edge("__manager__", "research_helper") + pb.add_edge("research_helper", END) + parent = pb.compile() + + async def _collect() -> Any: + namespaces = [] + async for ev in parent.astream_events( + {"messages": [HumanMessage(content="hi")]}, + {"configurable": {"thread_id": "t"}}, + version="v2", + ): + if ev["event"] == "on_chat_model_stream": + ns = (ev.get("metadata") or {}).get("langgraph_checkpoint_ns", "") + namespaces.append(ns) + return namespaces + + namespaces = asyncio.run(_collect()) + assert namespaces, "expected the worker to emit token-stream events" + # Every worker token event is namespaced under the worker node, so a + # consumer can attribute the stream to the sub-agent. + assert all(ns.startswith("research_helper:") for ns in namespaces), namespaces + + +def test_patched_astream_events_fails_open_on_filter_error() -> None: + """A bug in the delegation filter must never tear down the stream and + swallow later events (e.g. the worker events that follow the manager's + delegation turn). On a scrub error the event is passed through.""" + import asyncio + + from pyagentspec.adapters.langgraph._managerworkers import ( + _DelegationEventFilter, + _patch_hide_delegation_in_astream_events, + ) + + class _FakeGraph: + async def astream_events(self, *a: Any, **k: Any) -> Any: + yield {"event": "on_chat_model_stream", "run_id": "boom", "name": "x", "data": {}} + yield { + "event": "on_chat_model_stream", + "run_id": "ok", + "name": "x", + "data": {"chunk": "worker-token"}, + } + + def _explode(self: Any, event: Any) -> Any: + if event.get("run_id") == "boom": + raise RuntimeError("kaboom") + return event + + graph = _FakeGraph() + _patch_hide_delegation_in_astream_events(graph) + + async def _collect() -> Any: + out = [] + with patch.object(_DelegationEventFilter, "scrub", new=_explode): + async for ev in graph.astream_events(): + out.append(ev) + return out + + events = asyncio.run(_collect()) + # Both events survive: the one that raised is passed through unfiltered, + # and the later (worker) event is still delivered. + assert [e["run_id"] for e in events] == ["boom", "ok"] + + +def test_manager_workers_patches_astream_events() -> None: + """The compiled ManagerWorkers graph has its ``astream_events`` wrapped + with the delegation scrubber.""" + from langchain_core.language_models.fake_chat_models import ( + FakeMessagesListChatModel, + ) + from langgraph.checkpoint.memory import MemorySaver + + from pyagentspec.adapters.langgraph import AgentSpecLoader + from pyagentspec.adapters.langgraph._langgraphconverter import ( + AgentSpecToLangGraphConverter, + ) + from pyagentspec.agent import Agent + from pyagentspec.managerworkers import ManagerWorkers + + mw = ManagerWorkers( + name="Team", + group_manager=Agent( + name="Coordinator", + description="c", + system_prompt=".", + llm_config=_llm_cfg("manager_llm"), + ), + workers=[ + Agent( + name="Research Helper", + description="r", + system_prompt=".", + llm_config=_llm_cfg("worker_llm"), + ), + ], + ) + + def _dispatch(self_obj: Any, llm_config: Any, *args: Any, **kwargs: Any) -> Any: + return _fake_manager(*[]) + + loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver()) + with patch.object( + AgentSpecToLangGraphConverter, + "_llm_convert_to_langgraph", + autospec=True, + side_effect=_dispatch, + ), patch.object( + FakeMessagesListChatModel, + "bind_tools", + new=lambda self_obj, *a, **kw: self_obj, + ): + compiled = loader.load_component(mw) + + assert getattr(compiled.astream_events, "__name__", "") == "patched_astream_events" + + +# ─── ManagerWorkers as a Swarm member: handoff to a sibling ───────────────── + + +def test_handoff_tool_name_normalizes_like_node_names() -> None: + from pyagentspec.adapters.langgraph._managerworkers import ( + _handoff_tool_name, + ) + + assert _handoff_tool_name("Specialist") == "transfer_to_specialist" + assert _handoff_tool_name("Math Helper v2") == "transfer_to_math_helper_v2" + # Empty / punctuation-only falls back to a stable identifier. + assert _handoff_tool_name("!!!") == "transfer_to_agent" + + +def test_route_manager_handoff_takes_precedence_over_delegation() -> None: + """A ``transfer_to_`` call routes to the handoff node, and wins + over any ``delegate_to_`` calls emitted in the same turn.""" + from langchain_core.messages import AIMessage + + from pyagentspec.adapters.langgraph._managerworkers import ( + _HANDOFF_NODE_KEY, + _route_manager_to_worker_handoff_or_end, + ) + + mixed = AIMessage( + content="", + tool_calls=[ + {"name": "delegate_to_research_helper", "args": {"task": "x"}, "id": "c1"}, + {"name": "transfer_to_specialist", "args": {}, "id": "c2"}, + ], + ) + assert _route_manager_to_worker_handoff_or_end({"messages": [mixed]}) == _HANDOFF_NODE_KEY + + +def test_make_handoff_forward_node_reemits_parent_command() -> None: + """The handoff node returns a ``Command(goto=, graph=PARENT)`` + that sets ``active_agent`` and forwards the transfer AIMessage ahead of a + ToolMessage answering *every* unanswered tool call on it.""" + from langchain_core.messages import AIMessage, ToolMessage + from langgraph.types import Command + + from pyagentspec.adapters.langgraph._managerworkers import ( + _make_handoff_forward_node, + ) + + node = _make_handoff_forward_node({"transfer_to_specialist": "Specialist"}) + transfer_ai = AIMessage( + content="", + tool_calls=[ + {"name": "delegate_to_helper", "args": {"task": "y"}, "id": "d1"}, + {"name": "transfer_to_specialist", "args": {}, "id": "h1"}, + ], + ) + out = node.invoke({"messages": [transfer_ai]}) + + assert isinstance(out, Command) + assert out.goto == "Specialist" + assert out.graph == Command.PARENT + assert out.update["active_agent"] == "Specialist" + + forwarded = out.update["messages"] + # Transfer AIMessage first, then a ToolMessage per unanswered call. + assert forwarded[0] is transfer_ai + tool_msgs = [m for m in forwarded if isinstance(m, ToolMessage)] + answered = {m.tool_call_id for m in tool_msgs} + assert answered == {"d1", "h1"} + transferred = next(m for m in tool_msgs if m.tool_call_id == "h1") + assert transferred.content == "Successfully transferred to Specialist" + + +def test_manager_workers_as_swarm_member_hands_off_to_sibling() -> None: + """End-to-end: a Swarm whose first member is a ManagerWorkers. The + manager's LLM emits ``transfer_to_``; the parent graph re-emits + the handoff to the Swarm, which routes to the sibling Agent and lets it + answer — proving a sub-agent-bearing agent can participate in a Swarm + (the case the LangGraph adapter used to reject).""" + from langchain_core.language_models.fake_chat_models import ( + FakeMessagesListChatModel, + ) + from langchain_core.messages import AIMessage, HumanMessage, ToolMessage + from langgraph.checkpoint.memory import MemorySaver + + from pyagentspec.adapters.langgraph import AgentSpecLoader + from pyagentspec.adapters.langgraph._langgraphconverter import ( + AgentSpecToLangGraphConverter, + ) + from pyagentspec.agent import Agent + from pyagentspec.managerworkers import ManagerWorkers + from pyagentspec.swarm import Swarm + + manager_agent = Agent( + name="Coordinator", + description="Coordinates", + system_prompt="You coordinate.", + llm_config=_llm_cfg("manager_llm"), + ) + worker = Agent( + name="Research Helper", + description="Handles research", + system_prompt="You research.", + llm_config=_llm_cfg("worker_llm"), + ) + team = ManagerWorkers(name="Team", group_manager=manager_agent, workers=[worker]) + specialist = Agent( + name="Specialist", + description="Domain specialist", + system_prompt="You are the specialist.", + llm_config=_llm_cfg("specialist_llm"), + ) + swarm = Swarm( + name="Crew", + first_agent=team, + relationships=[(team, specialist), (specialist, team)], + ) + + # Manager turn 1: hand the conversation off to the Specialist sibling. + fake_manager = _fake_manager( + AIMessage( + content="", + tool_calls=[{"name": "transfer_to_specialist", "args": {}, "id": "call_h1"}], + ) + ) + fake_specialist = _fake_manager(AIMessage(content="Specialist handled it.")) + fake_worker = _fake_manager(AIMessage(content="(worker, unused)")) + + def _dispatch(self_obj: Any, llm_config: Any, *args: Any, **kwargs: Any) -> Any: + return { + "manager_llm": fake_manager, + "worker_llm": fake_worker, + "specialist_llm": fake_specialist, + }[llm_config.name] + + loader = AgentSpecLoader(tool_registry={}, checkpointer=MemorySaver()) + with patch.object( + AgentSpecToLangGraphConverter, + "_llm_convert_to_langgraph", + autospec=True, + side_effect=_dispatch, + ), patch.object( + FakeMessagesListChatModel, + "bind_tools", + new=lambda self_obj, *a, **kw: self_obj, + ): + compiled = loader.load_component(swarm) + + result = compiled.invoke( + {"messages": [HumanMessage(content="Please help.")]}, + {"configurable": {"thread_id": "crew-1"}}, + ) + messages = result["messages"] + + # The Specialist produced the final answer → the handoff actually routed. + assert isinstance(messages[-1], AIMessage) + assert "Specialist handled it." in messages[-1].content + + # The transfer is a well-formed AIMessage(tool_call) → ToolMessage pair. + transferred = [ + m + for m in messages + if isinstance(m, ToolMessage) and m.content == "Successfully transferred to Specialist" + ] + assert transferred and transferred[0].tool_call_id == "call_h1" + + # Transcript validity: every ToolMessage answers a preceding AIMessage + # tool_call with a matching id (an orphan ToolMessage would 400 a real LLM). + open_call_ids: set = set() + for m in messages: + for tc in getattr(m, "tool_calls", None) or []: + open_call_ids.add(tc["id"]) + if isinstance(m, ToolMessage): + assert ( + m.tool_call_id in open_call_ids + ), f"orphan ToolMessage {m.tool_call_id} with no preceding tool_call" + + +def test_swarm_rejects_unsupported_member_type() -> None: + """A Swarm member that is neither an Agent nor a ManagerWorkers (here a + nested Swarm) raises a clear NotImplementedError — there is no single + LLM to attach the handoff tools to.""" + import pytest + + from pyagentspec.adapters.langgraph._langgraphconverter import ( + AgentSpecToLangGraphConverter, + ) + from pyagentspec.agent import Agent + from pyagentspec.swarm import Swarm + + a1 = Agent(name="A1", description="a1", system_prompt="x", llm_config=_llm_cfg("l1")) + a2 = Agent(name="A2", description="a2", system_prompt="y", llm_config=_llm_cfg("l2")) + nested = Swarm(name="Nested", first_agent=a1, relationships=[(a1, a2)]) + outer = Swarm(name="Outer", first_agent=nested, relationships=[(nested, a1)]) + + with pytest.raises(NotImplementedError, match="Swarm"): + AgentSpecToLangGraphConverter().convert(outer, tool_registry={}) + + +# ─── Shared low-level helper unit tests (no LLM) ───────────────────────────── + + +def test_normalize_identifier_lowercases_collapses_and_strips() -> None: + """The single normalization used for both worker node names and + ``transfer_to_`` tool names.""" + from pyagentspec.adapters.langgraph._managerworkers import _normalize_identifier + + assert _normalize_identifier("Research Helper") == "research_helper" + assert _normalize_identifier("My-Worker!! v2") == "my_worker_v2" + # Punctuation-only / empty slugify to the empty string (callers add a fallback). + assert _normalize_identifier("!!!") == "" + assert _normalize_identifier("") == "" + + +def test_node_name_and_handoff_tool_name_share_one_normalization() -> None: + """Invariant the dedup relies on: a worker node name and a sibling handoff + tool name normalize identically, so a handoff ``goto`` (the raw sibling + name) and the routing node names stay in lockstep.""" + from pyagentspec.adapters.langgraph._managerworkers import ( + _handoff_tool_name, + _normalize_identifier, + _safe_node_name, + ) + + for name in ("Math Helper v2", "Specialist", "weird -- Name"): + assert _safe_node_name(name, "fallback") == _normalize_identifier(name) + assert _handoff_tool_name(name) == "transfer_to_" + _normalize_identifier(name) + + +def test_messages_of_reads_dict_and_object_state() -> None: + """The delegation / handoff tools receive state as a dict or an + attribute-bearing object depending on the langgraph injection path.""" + from langchain_core.messages import AIMessage + + from pyagentspec.adapters.langgraph._managerworkers import _messages_of + + msg = AIMessage(content="hi") + assert _messages_of({"messages": [msg]}) == [msg] + assert _messages_of({"messages": None}) == [] + assert _messages_of({}) == [] + + class _State: + messages = [msg] + + assert _messages_of(_State()) == [msg] + + class _Empty: + pass + + assert _messages_of(_Empty()) == [] + + +def test_surface_to_parent_command_projects_messages_with_no_goto() -> None: + """The body shared by both placeholder tools: break to the parent graph, + project the subgraph messages, carry no ``goto`` (routing is the parent's + job).""" + from langchain_core.messages import AIMessage + from langgraph.types import Command + + from pyagentspec.adapters.langgraph._managerworkers import _surface_to_parent_command + + m1, m2 = AIMessage(content="a"), AIMessage(content="b") + cmd = _surface_to_parent_command({"messages": [m1, m2]}) + + assert isinstance(cmd, Command) + assert cmd.graph == Command.PARENT + assert cmd.goto == () # no goto — the parent graph decides where to go + assert cmd.update == {"messages": [m1, m2]} + + +def test_delegation_and_handoff_tools_expose_expected_name_and_description() -> None: + """The placeholder tools the manager's LLM addresses by name.""" + from pyagentspec.adapters.langgraph._managerworkers import ( + _make_swarm_handoff_tool, + _make_worker_delegation_tool, + ) + + delegate = _make_worker_delegation_tool("research_helper") + assert delegate.name == "delegate_to_research_helper" + assert "research_helper" in delegate.description + + handoff = _make_swarm_handoff_tool("Specialist") + assert handoff.name == "transfer_to_specialist" + assert "Specialist" in handoff.description + + +# ─── _wrap_worker_for_subgraph: pending-delegation extraction (no LLM) ──────── + + +def _echo_worker_graph(reply: str = "WORKER REPLY") -> Any: + """A trivial worker CompiledStateGraph whose only node returns a fixed + AIMessage — enough to exercise the wrapper without an LLM.""" + from langchain_core.messages import AIMessage + from langgraph.graph import END, START, MessagesState, StateGraph + + wb = StateGraph(MessagesState) + wb.add_node("agent", lambda state: {"messages": [AIMessage(content=reply)]}) + wb.add_edge(START, "agent") + wb.add_edge("agent", END) + return wb.compile() + + +def test_wrap_worker_uses_send_payload_task_and_call_id() -> None: + """Fan-out path: the routing edge's ``Send`` payload carries the task and + the originating tool_call_id directly, so the worker reply ToolMessage is + matched to that call.""" + from langchain_core.messages import ToolMessage + + from pyagentspec.adapters.langgraph._managerworkers import ( + _DELEGATE_CALL_ID_KEY, + _DELEGATE_TASK_KEY, + _wrap_worker_for_subgraph, + ) + + node = _wrap_worker_for_subgraph(_echo_worker_graph("DONE"), "research_helper") + out = node.invoke({_DELEGATE_TASK_KEY: "do it", _DELEGATE_CALL_ID_KEY: "call_9"}) + + (reply,) = out["messages"] + assert isinstance(reply, ToolMessage) + assert reply.content == "DONE" + assert reply.tool_call_id == "call_9" + + +def test_wrap_worker_recovers_task_from_manager_message_on_direct_edge() -> None: + """Direct-edge path (no Send payload): the task and call id are recovered + from the manager's last AIMessage delegation tool call.""" + from langchain_core.messages import AIMessage, ToolMessage + + from pyagentspec.adapters.langgraph._managerworkers import _wrap_worker_for_subgraph + + node = _wrap_worker_for_subgraph(_echo_worker_graph("ANSWER"), "research_helper") + manager_ai = AIMessage( + content="", + tool_calls=[{"name": "delegate_to_research_helper", "args": {"task": "T"}, "id": "c1"}], + ) + out = node.invoke({"messages": [manager_ai]}) + + (reply,) = out["messages"] + assert isinstance(reply, ToolMessage) + assert reply.content == "ANSWER" + assert reply.tool_call_id == "c1" + + +def test_wrap_worker_raises_on_empty_manager_state() -> None: + from pyagentspec.adapters.langgraph._managerworkers import _wrap_worker_for_subgraph + + node = _wrap_worker_for_subgraph(_echo_worker_graph(), "research_helper") + with pytest.raises(RuntimeError, match="empty manager state"): + node.invoke({"messages": []}) + + +def test_wrap_worker_raises_when_no_matching_delegation_call() -> None: + from langchain_core.messages import AIMessage + + from pyagentspec.adapters.langgraph._managerworkers import _wrap_worker_for_subgraph + + node = _wrap_worker_for_subgraph(_echo_worker_graph(), "research_helper") + not_for_me = AIMessage( + content="", + tool_calls=[{"name": "delegate_to_other", "args": {"task": "x"}, "id": "c1"}], + ) + with pytest.raises(RuntimeError, match="delegate_to_research_helper"): + node.invoke({"messages": [not_for_me]})