test/network/test_packet_routing.py
Rama main · réplica de lectura del repositorio original.
# Copyright (c) 2026, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
#
# SPDX-License-Identifier: Apache-2.0
"""Unit tests for packet routing resolution.
Pins the contract that ``WorkflowAdapter._resolve_routing`` walks
agent local-stream events in emission order and applies first-emit-
wins across both static (``ToolCalled`` graph rule) and dynamic
(``Handoff`` typed return) routing.
"""
import json
from ag2.events import DataInput, ToolCallEvent, ToolResult, ToolResultEvent
from ag2.network import (
AgentTarget,
Finish,
Handoff,
TerminateTarget,
ToolCalled,
Transition,
TransitionGraph,
)
from ag2.network.adapters.workflow import _resolve_routing
def _call(name: str, *, call_id: str = "id1", reason: str = "") -> ToolCallEvent:
args = json.dumps({"reason": reason}) if reason else "{}"
return ToolCallEvent(id=call_id, name=name, arguments=args)
def _result(parent_id: str, *, name: str | None = None, value=None) -> ToolResultEvent:
"""Build a ToolResultEvent. ``value`` becomes the ``result.parts[0]``
payload — a Handoff goes into a DataInput, a string into TextInput
via Input.ensure_input."""
return ToolResultEvent(
parent_id=parent_id,
name=name,
result=ToolResult.ensure_result(value if value is not None else "ok"),
)
def _graph(*tools: str) -> TransitionGraph:
"""Build a minimal graph with ToolCalled rules for each tool name."""
transitions = [Transition(when=ToolCalled(t), then=AgentTarget(f"agent_{t}")) for t in tools]
return TransitionGraph(
initial_speaker="agent_initial",
transitions=transitions,
default_target=TerminateTarget(reason="default"),
)
class TestResolveRouting:
def test_no_events_returns_text(self) -> None:
graph = _graph("delegate_a")
routing = _resolve_routing([], graph, name_to_id={})
assert routing == {"kind": "text"}
def test_no_graph_returns_text(self) -> None:
events = [_call("delegate_a")]
routing = _resolve_routing(events, None, name_to_id={})
assert routing == {"kind": "text"}
def test_single_static_tool_call_routes(self) -> None:
"""A single ToolCallEvent matching a ToolCalled rule sets
routing.tool; routing.target left unset (fold's select_next
resolves it)."""
graph = _graph("delegate_a")
events = [_call("delegate_a", call_id="c1", reason="go")]
routing = _resolve_routing(events, graph, name_to_id={})
assert routing == {
"kind": "handoff",
"tool": "delegate_a",
"reason": "go",
}
def test_unmatched_tool_call_returns_text(self) -> None:
"""Tool whose name doesn't match any ToolCalled rule falls
through to text routing."""
graph = _graph("delegate_a")
events = [_call("some_other_tool", call_id="c1")]
routing = _resolve_routing(events, graph, name_to_id={})
assert routing == {"kind": "text"}
def test_two_static_tools_first_emit_wins(self) -> None:
"""Two parallel ToolCallEvents both matching ToolCalled rules:
the FIRST in emission order wins routing."""
graph = _graph("delegate_a", "delegate_b")
events = [
_call("delegate_a", call_id="c1", reason="first"),
_call("delegate_b", call_id="c2", reason="second"),
]
routing = _resolve_routing(events, graph, name_to_id={})
assert routing["tool"] == "delegate_a"
assert routing["reason"] == "first"
def test_dynamic_handoff_to_an_exact_agent_id_is_not_shadowed_by_a_name(self) -> None:
graph = _graph("delegate_a")
call = _call("smart_route", call_id="c1")
result = _result("c1", name="smart_route", value=Handoff(target="id-bob", reason="picked"))
routing = _resolve_routing(
[call, result],
graph,
name_to_id={"bob": "id-bob", "id-bob": "id-mallory"},
)
assert routing["target"] == "id-bob"
def test_dynamic_handoff_resolves_target(self) -> None:
"""A ToolResultEvent carrying a Handoff dataclass produces
dynamic routing with the resolved target id."""
graph = _graph("delegate_a")
call = _call("smart_route", call_id="c1")
result = _result("c1", name="smart_route", value=Handoff(target="researcher", reason="picked"))
events = [call, result]
routing = _resolve_routing(
events,
graph,
name_to_id={"researcher": "agent-id-researcher"},
)
assert routing == {
"kind": "handoff",
"tool": "smart_route",
"reason": "picked",
"target": "agent-id-researcher",
}
def test_dynamic_handoff_unresolved_name_falls_through_to_name(self) -> None:
"""If the Handoff's target isn't in name_to_id, the name itself
is used (caller's responsibility to handle UnknownHandoffTarget)."""
graph = _graph("delegate_a")
events = [
_call("smart_route", call_id="c1"),
_result("c1", name="smart_route", value=Handoff(target="unknown_name")),
]
routing = _resolve_routing(events, graph, name_to_id={})
assert routing["target"] == "unknown_name"
def test_dynamic_overrides_static_for_same_tool(self) -> None:
"""A tool registered via ToolCalled AND returning Handoff:
the dynamic Handoff wins for that tool's call."""
graph = _graph("delegate_a")
events = [
_call("delegate_a", call_id="c1"),
_result("c1", name="delegate_a", value=Handoff(target="explicit_target")),
]
routing = _resolve_routing(
events,
graph,
name_to_id={"explicit_target": "agent-id-explicit"},
)
# Dynamic Handoff present → routing carries target.
assert routing["target"] == "agent-id-explicit"
def test_dynamic_first_static_second_first_wins(self) -> None:
"""First emission wins regardless of dynamic vs static."""
graph = _graph("delegate_b")
events = [
_call("smart_route", call_id="c1"), # dynamic, first
_result("c1", name="smart_route", value=Handoff(target="alice")),
_call("delegate_b", call_id="c2"), # static, second
]
routing = _resolve_routing(events, graph, name_to_id={"alice": "agent-id-alice"})
assert routing["tool"] == "smart_route"
assert routing["target"] == "agent-id-alice"
def test_static_first_dynamic_second_first_wins(self) -> None:
"""First emission wins regardless of dynamic vs static (reversed)."""
graph = _graph("delegate_a")
events = [
_call("delegate_a", call_id="c1"), # static, first
_call("smart_route", call_id="c2"), # dynamic, second
_result("c2", name="smart_route", value=Handoff(target="bob")),
]
routing = _resolve_routing(events, graph, name_to_id={"bob": "agent-id-bob"})
assert routing["tool"] == "delegate_a"
# No target field for static (fold's select_next resolves).
assert "target" not in routing
def test_string_result_not_handoff(self) -> None:
"""A tool that returns a plain string produces no dynamic
routing — only Handoff instances trigger that path."""
graph = _graph("delegate_a")
events = [
_call("delegate_a", call_id="c1"),
_result("c1", name="delegate_a", value="just a string"),
]
routing = _resolve_routing(events, graph, name_to_id={})
# Static routing still applies (tool name matches).
assert routing["tool"] == "delegate_a"
assert "target" not in routing
def test_data_input_non_handoff_ignored(self) -> None:
"""A DataInput whose data isn't a Handoff is not routing-relevant."""
graph = _graph("delegate_a")
result = ToolResultEvent(
parent_id="c1",
name="other_tool",
result=ToolResult.ensure_result(DataInput(data={"some": "data"})),
)
events = [_call("other_tool", call_id="c1"), result]
routing = _resolve_routing(events, graph, name_to_id={})
# No matching rule, no Handoff → text routing.
assert routing == {"kind": "text"}
def test_results_without_calls_ignored(self) -> None:
"""Orphan ToolResultEvent (no matching call) doesn't drive routing."""
graph = _graph("delegate_a")
events = [_result("orphan", value=Handoff(target="alice"))]
routing = _resolve_routing(events, graph, name_to_id={"alice": "agent-id-alice"})
# No calls at all → text routing.
assert routing == {"kind": "text"}
def test_dynamic_finish_terminates(self) -> None:
"""A ToolResultEvent carrying a Finish dataclass produces
routing with ``kind: "finish"`` so fold can close the channel."""
graph = _graph("delegate_a")
events = [
_call("finish", call_id="c1"),
_result("c1", name="finish", value=Finish(summary="all done", reason="done")),
]
routing = _resolve_routing(events, graph, name_to_id={})
assert routing == {
"kind": "finish",
"tool": "finish",
"reason": "done",
"summary": "all done",
}
def test_finish_default_reason_propagates(self) -> None:
"""Finish() with no args still produces a valid termination routing."""
graph = _graph("delegate_a")
events = [
_call("finish", call_id="c1"),
_result("c1", name="finish", value=Finish()),
]
routing = _resolve_routing(events, graph, name_to_id={})
assert routing == {
"kind": "finish",
"tool": "finish",
"reason": "finished",
"summary": "",
}
def test_finish_beats_handoff_in_same_emission(self) -> None:
"""If a single tool somehow returns Finish, it beats a later
Handoff from another call — first-emitted-wins extends to
Finish."""
graph = _graph("delegate_a")
events = [
_call("finish", call_id="c1"),
_result("c1", name="finish", value=Finish(reason="early")),
_call("delegate_a", call_id="c2"),
_result("c2", name="delegate_a", value=Handoff(target="alice")),
]
routing = _resolve_routing(events, graph, name_to_id={"alice": "agent-id-alice"})
assert routing["kind"] == "finish"
assert routing["tool"] == "finish"
def test_handoff_before_finish_still_first_wins(self) -> None:
"""First emission still wins — a Handoff that fires before a
later Finish takes precedence."""
graph = _graph("delegate_a")
events = [
_call("delegate_a", call_id="c1"),
_result("c1", name="delegate_a", value=Handoff(target="alice")),
_call("finish", call_id="c2"),
_result("c2", name="finish", value=Finish(reason="late")),
]
routing = _resolve_routing(events, graph, name_to_id={"alice": "agent-id-alice"})
assert routing["kind"] == "handoff"
assert routing["target"] == "agent-id-alice"