test/providers/test_network_cross_process_smoke.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
"""Cross-process network smoke tests with real, mixed-provider LLMs.
A single ``Hub`` runs ``serve_ws`` on a loopback port. Every agent is
its own node: a ``HubClient(WsLink(url))`` with **no in-process hub
reference**, so every control-plane call (register, open a channel,
send, discovery, task ops) crosses the wire as a ``RequestFrame`` RPC
and every delivery arrives as a ``NotifyFrame`` — the genuinely
distributed path the cross-process work added. The agents are backed by
different providers in the same network (Anthropic ``claude-haiku-4-5``,
OpenAI ``gpt-5.4-mini``, Gemini ``gemini-3.6-flash``), mirroring the
common real-world setting where one hub coordinates heterogeneous
agents.
Each adapter family is exercised end to end through the public
``HubClient`` / ``AgentClient`` API — the initiator opens the channel
and sends; the default notify handler folds adapter state locally and
drives each turn over the wire:
* ``consulting`` — 1-question/1-reply auto-close (Anthropic ↔ OpenAI).
* ``conversation`` — free two-party exchange (Anthropic ↔ Gemini).
* ``discussion`` — three-way round-robin, one provider each.
* ``workflow`` — a handoff routed across providers via a transition graph.
* reconnect-by-name — a node drops mid-turn before acking; a fresh node
attaches under the same name and the unacked delivery replays.
* ``RemoteAgentProxy`` — a tenant federation proxy (LLM-backed) answers
on behalf of a ``kind="remote_agent"`` participant.
Each test skips unless the API keys for the providers it uses are
present (loaded from ``.env`` at repo root).
"""
import asyncio
import contextlib
import os
from collections.abc import AsyncGenerator
from pathlib import Path
import pytest
pytest.importorskip("websockets")
pytest.importorskip("anthropic")
pytest.importorskip("openai")
pytest.importorskip("google.genai")
from ag2 import Agent
from ag2.config import AnthropicConfig, GeminiConfig, OpenAIConfig
from ag2.knowledge import MemoryKnowledgeStore
from ag2.network import (
EV_PACKET,
EV_TEXT,
AuthBlock,
Envelope,
Handoff,
Hub,
HubClient,
Passport,
Resume,
WsLink,
serve_ws,
)
from ag2.network.adapters.consulting import CONSULTING_TYPE
from ag2.network.adapters.conversation import CONVERSATION_TYPE
from ag2.network.adapters.discussion import DISCUSSION_TYPE, ORDERING_ROUND_ROBIN
from ag2.network.adapters.workflow import WORKFLOW_TYPE
from ag2.network.envelope import EV_CHANNEL_INVITE, EV_CHANNEL_INVITE_ACK
from ag2.network.transitions import (
AgentTarget,
FromSpeaker,
RevertToInitiatorTarget,
TerminateTarget,
ToolCalled,
Transition,
TransitionGraph,
)
try:
from dotenv import load_dotenv
_REPO_ROOT = Path(__file__).resolve().parents[3]
load_dotenv(_REPO_ROOT / ".env")
except ImportError:
pass
ANTHROPIC_MODEL = "claude-haiku-4-5"
OPENAI_MODEL = "gpt-5.4-mini"
GEMINI_MODEL = "gemini-3.6-flash"
def _require(*names: str) -> dict[str, str]:
"""Return the named API keys, or skip if any are missing."""
values = {name: os.getenv(name) for name in names}
missing = [name for name, value in values.items() if not value]
if missing:
pytest.skip(f"cross-process smoke needs {', '.join(missing)}")
return {name: value for name, value in values.items() if value}
def _anthropic(keys: dict[str, str]) -> AnthropicConfig:
return AnthropicConfig(model=ANTHROPIC_MODEL, api_key=keys["ANTHROPIC_API_KEY"])
def _openai(keys: dict[str, str]) -> OpenAIConfig:
return OpenAIConfig(model=OPENAI_MODEL, api_key=keys["OPENAI_API_KEY"], temperature=0)
def _gemini(keys: dict[str, str]) -> GeminiConfig:
return GeminiConfig(model=GEMINI_MODEL, api_key=keys["GEMINI_API_KEY"], temperature=0)
@contextlib.asynccontextmanager
async def _network() -> AsyncGenerator[tuple[Hub, str, list[HubClient]]]:
"""Start a hub on a loopback ``serve_ws`` port.
Yields ``(hub, ws_url, clients)``. The hub object is held by the
test for read-only assertions (``read_wal`` / ``adapter_state``);
the agents only ever reach it over ``ws_url``. ``clients`` is the
cleanup registry — append every ``HubClient`` so it is closed on
exit.
"""
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0, expectation_sweep_interval=0)
clients: list[HubClient] = []
try:
async with serve_ws(hub, "127.0.0.1", 0) as server:
url = f"ws://127.0.0.1:{server.sockets[0].getsockname()[1]}"
yield hub, url, clients
finally:
for hc in clients:
with contextlib.suppress(Exception):
await hc.close()
await hub.close()
async def _join(
url: str,
clients: list[HubClient],
agent: Agent,
*,
resume: Resume | None = None,
):
"""Connect a fresh node (its own ``WsLink`` / ``HubClient``) and
register ``agent`` over the wire. Returns the ``AgentClient``."""
hc = HubClient(WsLink(url))
clients.append(hc)
return await hc.register(agent, Passport(name=agent.name), resume or Resume())
async def _wait_for_text_count(hub: Hub, channel_id: str, expected: int, *, timeout: float = 120.0) -> int:
deadline = asyncio.get_event_loop().time() + timeout
while asyncio.get_event_loop().time() < deadline:
wal = await hub.read_wal(channel_id)
count = sum(1 for e in wal if e.event_type == EV_TEXT)
if count >= expected:
return count
await asyncio.sleep(0.2)
return sum(1 for e in (await hub.read_wal(channel_id)) if e.event_type == EV_TEXT)
@pytest.mark.anthropic
@pytest.mark.openai
@pytest.mark.asyncio
async def test_consulting_anthropic_to_openai_over_wire() -> None:
"""Consulting 1Q1R across processes and providers: alice (Anthropic)
opens a consulting channel with bob (OpenAI), asks one question, and
bob's notify handler answers over the wire before the adapter
auto-closes."""
keys = _require("ANTHROPIC_API_KEY", "OPENAI_API_KEY")
async with _network() as (hub, url, clients):
alice = await _join(
url,
clients,
Agent(name="alice", prompt="You are alice, a coordinator.", config=_anthropic(keys)),
resume=Resume(summary="coordinator"),
)
await _join(
url,
clients,
Agent(
name="bob",
prompt="You are an arithmetic specialist. Reply with just the integer result, no words.",
config=_openai(keys),
),
resume=Resume(claimed_capabilities=["math"], summary="math specialist"),
)
channel = await alice.open(type=CONSULTING_TYPE, target=["bob"], intent="ask bob to compute a small product")
await channel.send("What is 12 times 11? Reply with just the integer.")
count = await _wait_for_text_count(hub, channel.channel_id, 2, timeout=90.0)
assert count == 2, f"expected alice's question + bob's reply, got {count} text envelopes"
wal = await hub.read_wal(channel.channel_id)
texts = [e for e in wal if e.event_type == EV_TEXT]
assert "132" in texts[1].event_data.get("text", ""), (
f"bob (OpenAI) should answer 132 for 12*11; got {texts[1].event_data.get('text')!r}"
)
@pytest.mark.anthropic
@pytest.mark.gemini
@pytest.mark.asyncio
async def test_conversation_anthropic_with_gemini_over_wire() -> None:
"""Free two-party conversation across providers: alice (Anthropic)
and bob (Gemini) exchange at least one turn each over the wire. The
channel is closed explicitly once both have spoken (a free
conversation has no auto-terminator)."""
keys = _require("ANTHROPIC_API_KEY", "GEMINI_API_KEY")
async with _network() as (hub, url, clients):
alice = await _join(
url,
clients,
Agent(
name="alice",
prompt="You are alice. Keep replies to one short, friendly sentence.",
config=_anthropic(keys),
),
)
await _join(
url,
clients,
Agent(
name="bob",
prompt="You are bob. Keep replies to one short, friendly sentence.",
config=_gemini(keys),
),
)
channel = await alice.open(type=CONVERSATION_TYPE, target=["bob"], intent="quick hello")
await channel.send("Hi Bob — in one sentence, what's your favorite kind of weather?")
count = await _wait_for_text_count(hub, channel.channel_id, 2, timeout=90.0)
assert count >= 2, f"expected at least alice + bob turns, got {count}"
wal = await hub.read_wal(channel.channel_id)
speakers = {e.sender_id for e in wal if e.event_type == EV_TEXT}
assert len(speakers) == 2, f"both providers should contribute a turn; speakers={speakers}"
await channel.close(reason="smoke_done")
@pytest.mark.anthropic
@pytest.mark.openai
@pytest.mark.gemini
@pytest.mark.asyncio
async def test_discussion_three_providers_round_robin_over_wire() -> None:
"""Three nodes, one provider each (Anthropic, OpenAI, Gemini), take
round-robin turns in one discussion over the wire. The adapter gates
turns by ``can_send``, so the speaking order holds regardless of how
concurrent deliveries interleave across connections."""
keys = _require("ANTHROPIC_API_KEY", "OPENAI_API_KEY", "GEMINI_API_KEY")
async with _network() as (hub, url, clients):
prompt = (
"You are {name} in a 3-way discussion. When it is your turn, reply "
"with exactly one short opinion (one sentence) as plain text. Do "
"not ask questions and do not call any tools."
)
alice = await _join(
url, clients, Agent(name="alice", prompt=prompt.format(name="alice"), config=_anthropic(keys))
)
bob = await _join(url, clients, Agent(name="bob", prompt=prompt.format(name="bob"), config=_openai(keys)))
carol = await _join(url, clients, Agent(name="carol", prompt=prompt.format(name="carol"), config=_gemini(keys)))
channel = await alice.open(
type=DISCUSSION_TYPE,
target=["bob", "carol"],
knobs={"ordering": ORDERING_ROUND_ROBIN},
intent="mixed-provider debate",
)
await channel.send("Quick debate: should new services default to typed languages?")
count = await _wait_for_text_count(hub, channel.channel_id, 3, timeout=180.0)
assert count >= 3, f"expected one turn per provider, got {count}"
wal = await hub.read_wal(channel.channel_id)
speakers = [e.sender_id for e in wal if e.event_type == EV_TEXT][:3]
assert speakers == [alice.agent_id, bob.agent_id, carol.agent_id], (
f"round-robin order broken across providers/processes; got {speakers}"
)
@pytest.mark.anthropic
@pytest.mark.openai
@pytest.mark.asyncio
async def test_workflow_handoff_across_providers_over_wire() -> None:
"""Workflow handoff over the wire: a human seeds the graph; triage
(Anthropic) routes the question to eng (OpenAI) via a handoff tool;
the transition graph then reverts to the initiator. Verifies the
routing ``EV_PACKET`` and the receiving provider's notify handler
both work across processes."""
keys = _require("ANTHROPIC_API_KEY", "OPENAI_API_KEY")
async with _network() as (hub, url, clients):
triage = await _join(
url,
clients,
Agent(
name="triage",
prompt=(
"You are the triage coordinator. When the user asks an engineering "
"question, call the transfer_to_eng tool with a one-line reason. Do "
"not answer it yourself."
),
config=_anthropic(keys),
),
resume=Resume(claimed_capabilities=["triage"]),
)
eng = await _join(
url,
clients,
Agent(
name="eng",
prompt="You are a senior engineer. Answer concisely in one sentence.",
config=_openai(keys),
),
resume=Resume(claimed_capabilities=["engineering"]),
)
user_hc = HubClient(WsLink(url))
clients.append(user_hc)
user = await user_hc.register_human(Passport(name="user"))
eng_agent_id = eng.agent_id
@triage.agent.tool
async def transfer_to_eng(reason: str = "") -> Handoff:
"""Transfer the conversation to the engineering specialist."""
return Handoff(target=eng_agent_id, reason=reason)
graph = TransitionGraph(
initial_speaker=user.agent_id,
transitions=[
Transition(when=FromSpeaker(user.agent_id), then=AgentTarget(triage.agent_id)),
Transition(when=ToolCalled("transfer_to_eng"), then=AgentTarget(eng.agent_id)),
Transition(when=FromSpeaker(eng.agent_id), then=RevertToInitiatorTarget()),
],
default_target=TerminateTarget(reason="triage_done"),
max_turns=8,
)
channel = await triage.open(
type=WORKFLOW_TYPE,
target=[user.agent_id, eng.agent_id],
knobs={"graph": graph.to_dict()},
intent="triage routes deep questions to engineering",
)
await user.send(
channel.channel_id,
"What's the practical difference between processes and threads in Python?",
)
# triage hands off to eng (an EV_PACKET with routing.kind == handoff).
async def _handoff_seen() -> bool:
wal = await hub.read_wal(channel.channel_id)
return any(
e.event_type == EV_PACKET and (e.event_data.get("routing", {}) or {}).get("kind") == "handoff"
for e in wal
)
deadline = asyncio.get_event_loop().time() + 90.0
handed_off = False
while asyncio.get_event_loop().time() < deadline:
if await _handoff_seen():
handed_off = True
break
await asyncio.sleep(0.3)
assert handed_off, "triage (Anthropic) did not hand off to eng within 90s"
# eng (OpenAI) replies via its notify handler, then the graph
# reverts to the initiator (triage).
deadline = asyncio.get_event_loop().time() + 90.0
eng_replied = False
while asyncio.get_event_loop().time() < deadline:
state = hub.adapter_state(channel.channel_id)
if state is not None and getattr(state, "last_speaker_id", None) == eng.agent_id:
eng_replied = True
break
await asyncio.sleep(0.2)
assert eng_replied, "eng (OpenAI) did not reply to triage's (Anthropic) handoff within 90s"
@pytest.mark.anthropic
@pytest.mark.openai
@pytest.mark.asyncio
async def test_reconnect_by_name_replays_unacked_turn_over_wire() -> None:
"""A node drops mid-turn before acking, then a fresh node attaches
under the same name and the hub replays the unacked delivery.
bob's first connection receives alice's seed but its connection is
closed while the (slow, real-LLM) turn is still in flight — so the
seed is never acked and bob never replies. A second ``bob`` node
``attach``es over the wire with ``since_envelope_id=""``; the hub
replays the unacked seed past bob's cursor and the fresh node
answers."""
keys = _require("ANTHROPIC_API_KEY", "OPENAI_API_KEY")
async with _network() as (hub, url, clients):
alice = await _join(url, clients, Agent(name="alice", prompt="You are alice.", config=_anthropic(keys)))
# First bob node: connects + auto-acks the invite, but we drop its
# connection right after the seed is dispatched, before the
# in-flight LLM turn can post a reply or ack the delivery.
bob_hc1 = HubClient(WsLink(url))
clients.append(bob_hc1)
bob1 = await bob_hc1.register(
Agent(
name="bob",
prompt="You are bob. Reply with just the single integer answer.",
config=_openai(keys),
),
Passport(name="bob"),
Resume(),
)
bob_id = bob1.agent_id
channel = await alice.open(type=DISCUSSION_TYPE, target=["bob"], knobs={"ordering": ORDERING_ROUND_ROBIN})
await channel.send("Bob, what is 7 plus 8? Reply with just the integer.")
# Let the seed reach bob1 and start his (multi-second) turn, then
# crash the connection before it can ack.
await asyncio.sleep(0.3)
await bob_hc1.close()
# Only alice's seed is in the WAL; bob never replied.
before = await hub.read_wal(channel.channel_id)
assert sum(1 for e in before if e.event_type == EV_TEXT) == 1, "bob should not have replied before reconnect"
# Fresh bob node attaches under the same name and asks for replay.
bob_hc2 = HubClient(WsLink(url))
clients.append(bob_hc2)
bob2 = await bob_hc2.attach(
Agent(
name="bob",
prompt="You are bob. Reply with just the single integer answer.",
config=_openai(keys),
),
name="bob",
passport=Passport(name="bob"),
resume=Resume(),
since_envelope_id="",
)
assert bob2.agent_id == bob_id, "attach should re-bind the same identity"
count = await _wait_for_text_count(hub, channel.channel_id, 2, timeout=90.0)
assert count == 2, f"replay should let the fresh bob answer; got {count} text envelopes"
wal = await hub.read_wal(channel.channel_id)
bob_texts = [e for e in wal if e.event_type == EV_TEXT and e.sender_id == bob_id]
assert len(bob_texts) == 1, f"bob should reply exactly once after replay, got {len(bob_texts)}"
assert "15" in bob_texts[0].event_data.get("text", ""), (
f"bob (OpenAI) should answer 15 for 7+8; got {bob_texts[0].event_data.get('text')!r}"
)
class _LLMRemoteProxy:
"""Tenant ``RemoteAgentProxy`` standing in for a federated peer.
Bridges envelopes the hub addresses to a ``kind="remote_agent"``
passport: it acks invites on the remote's behalf, runs a real
(Gemini) agent on inbound text, and reposts the reply into the
channel via ``Hub.post_envelope`` — the same shape an A2A or gRPC
bridge would take, with a local LLM standing in for the remote
endpoint so the federation seam is exercised end to end.
"""
scheme = "rpc"
def __init__(self, hub: Hub, agent: Agent, agent_id: str) -> None:
self._hub = hub
self._agent = agent
self._agent_id = agent_id
self.dispatched = 0
async def dispatch(self, envelope: Envelope, recipient: Passport) -> None:
if envelope.event_type == EV_CHANNEL_INVITE:
await self._hub.post_envelope(
Envelope(
channel_id=envelope.channel_id,
sender_id=self._agent_id,
audience=None,
event_type=EV_CHANNEL_INVITE_ACK,
event_data={"channel_id": envelope.channel_id},
causation_id=envelope.envelope_id,
)
)
return
if envelope.event_type != EV_TEXT:
return
reply = await self._agent.ask(envelope.event_data.get("text", ""))
await self._hub.post_envelope(
Envelope(
channel_id=envelope.channel_id,
sender_id=self._agent_id,
audience=[envelope.sender_id],
event_type=EV_TEXT,
event_data={"text": reply.body or ""},
causation_id=envelope.envelope_id,
)
)
self.dispatched += 1
async def close(self) -> None:
return None
@pytest.mark.anthropic
@pytest.mark.gemini
@pytest.mark.asyncio
async def test_remote_agent_proxy_federation_over_wire() -> None:
"""The ``RemoteAgentProxy`` federation seam: alice (Anthropic, a
remote ``WsLink`` node) consults a ``kind="remote_agent"`` peer whose
passport routes to a tenant proxy. The proxy answers with a real
Gemini turn and reposts it; alice sees the reply over the wire as a
normal consulting answer."""
keys = _require("ANTHROPIC_API_KEY", "GEMINI_API_KEY")
async with _network() as (hub, url, clients):
# Register the remote-agent identity + its proxy directly on the
# hub (this is the federation operator's side of the seam).
echo = await hub.register_identity(
Passport(name="echo", kind="remote_agent", auth=AuthBlock(scheme="rpc", claim={})),
Resume(claimed_capabilities=["math"]),
)
proxy = _LLMRemoteProxy(
hub,
Agent(
name="echo-llm",
prompt="You are an arithmetic specialist. Reply with just the integer result.",
config=_gemini(keys),
),
echo.agent_id,
)
hub.register_remote_proxy(proxy)
alice = await _join(url, clients, Agent(name="alice", prompt="You are alice.", config=_anthropic(keys)))
channel = await alice.open(
type=CONSULTING_TYPE, target=["echo"], intent="delegate a computation to a remote peer"
)
await channel.send("What is 9 times 13? Reply with just the integer.")
count = await _wait_for_text_count(hub, channel.channel_id, 2, timeout=90.0)
assert count == 2, f"expected alice's question + the remote's reply, got {count}"
assert proxy.dispatched >= 1, "proxy.dispatch should have fired for the remote recipient"
wal = await hub.read_wal(channel.channel_id)
remote_texts = [e for e in wal if e.event_type == EV_TEXT and e.sender_id == echo.agent_id]
assert remote_texts and "117" in remote_texts[0].event_data.get("text", ""), (
f"remote peer (Gemini via proxy) should answer 117 for 9*13; got "
f"{[e.event_data.get('text') for e in remote_texts]}"
)