test/network/test_adapter_edges.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
"""Adapter validate_send / validate_create / fold edge cases.
Each adapter (consulting, conversation, discussion, workflow) declares
its own protocol shape. The hub gates substantive sends through
``adapter.validate_send`` under a per-channel lock; rejection raises
``ProtocolError`` and the WAL must remain untouched. Hydrate re-folds
the WAL through ``adapter.fold`` so determinism is required (replaying
the same envelope sequence must produce the same state).
These tests exercise the rejection paths and fold determinism that
the existing per-adapter integration tests don't cover directly.
"""
import tempfile
import pytest
from ag2 import Agent
from ag2.knowledge import DiskKnowledgeStore, MemoryKnowledgeStore
from ag2.network import (
EV_TEXT,
Envelope,
Hub,
ProtocolError,
)
from ag2.network.adapters.consulting import ConsultingAdapter, ConsultingState
from ag2.network.adapters.conversation import ConversationAdapter
from ag2.network.adapters.discussion import DiscussionAdapter, DiscussionState
from ag2.network.channel import (
ChannelMetadata,
ChannelState,
Participant,
ParticipantRole,
)
from ag2.network.envelope import EV_CHANNEL_OPENED
from ._helpers import ScriptedConfig
def _agent(name: str) -> Agent:
return Agent(name=name, config=ScriptedConfig())
def _make_metadata(
*,
manifest,
creator: str,
participants: list[tuple[str, ParticipantRole]],
knobs: dict | None = None,
) -> ChannelMetadata:
parts = [
Participant(agent_id=aid, role=role, order=i, joined_at="2026-01-01T00:00:00+00:00")
for i, (aid, role) in enumerate(participants)
]
return ChannelMetadata(
channel_id="s1",
manifest=manifest,
creator_id=creator,
participants=parts,
state=ChannelState.ACTIVE,
created_at="2026-01-01T00:00:00+00:00",
knobs=knobs or {},
)
class TestConsultingAdapter:
"""ConsultingAdapter unit-level edge cases."""
def test_validate_create_rejects_three_participants(self) -> None:
adapter = ConsultingAdapter()
meta = _make_metadata(
manifest=adapter.manifest,
creator="alice",
participants=[
("alice", ParticipantRole.INITIATOR),
("bob", ParticipantRole.RESPONDENT),
("carol", ParticipantRole.PARTICIPANT),
],
)
with pytest.raises(ProtocolError, match="exactly 2"):
adapter.validate_create(meta)
def test_validate_create_rejects_missing_respondent(self) -> None:
adapter = ConsultingAdapter()
meta = _make_metadata(
manifest=adapter.manifest,
creator="alice",
participants=[
("alice", ParticipantRole.INITIATOR),
("bob", ParticipantRole.PARTICIPANT),
],
)
with pytest.raises(ProtocolError, match="respondent"):
adapter.validate_create(meta)
def test_validate_send_rejects_respondent_first(self) -> None:
"""Initiator must send the first substantive envelope."""
adapter = ConsultingAdapter()
meta = _make_metadata(
manifest=adapter.manifest,
creator="alice",
participants=[
("alice", ParticipantRole.INITIATOR),
("bob", ParticipantRole.RESPONDENT),
],
)
state = adapter.initial_state(meta)
envelope = Envelope(
channel_id="s1",
sender_id="bob",
audience=["alice"],
event_type=EV_TEXT,
event_data={"text": "first?"},
)
with pytest.raises(ProtocolError, match="initiator"):
adapter.validate_send(meta, envelope, state)
def test_validate_send_rejects_after_both_replied(self) -> None:
adapter = ConsultingAdapter()
meta = _make_metadata(
manifest=adapter.manifest,
creator="alice",
participants=[
("alice", ParticipantRole.INITIATOR),
("bob", ParticipantRole.RESPONDENT),
],
)
state = ConsultingState(initiator_sent=True, respondent_replied=True)
envelope = Envelope(
channel_id="s1",
sender_id="alice",
audience=["bob"],
event_type=EV_TEXT,
event_data={"text": "third send"},
)
with pytest.raises(ProtocolError, match="already complete"):
adapter.validate_send(meta, envelope, state)
def test_fold_is_deterministic_under_replay(self) -> None:
"""Replaying the same WAL through initial_state + fold reproduces state."""
adapter = ConsultingAdapter()
meta = _make_metadata(
manifest=adapter.manifest,
creator="alice",
participants=[
("alice", ParticipantRole.INITIATOR),
("bob", ParticipantRole.RESPONDENT),
],
)
envelopes = [
Envelope(
channel_id="s1",
sender_id="alice",
audience=["bob"],
event_type=EV_TEXT,
event_data={"text": "q"},
),
Envelope(
channel_id="s1",
sender_id="bob",
audience=["alice"],
event_type=EV_TEXT,
event_data={"text": "a"},
),
]
# Stamp envelope_ids so fold can record them.
envelopes[0].envelope_id = "env-1"
envelopes[1].envelope_id = "env-2"
s1 = adapter.initial_state(meta)
for env in envelopes:
s1 = adapter.fold(env, s1)
s2 = adapter.initial_state(meta)
for env in envelopes:
s2 = adapter.fold(env, s2)
assert s1 == s2
def test_on_accepted_returns_closed_after_both_replied(self) -> None:
adapter = ConsultingAdapter()
meta = _make_metadata(
manifest=adapter.manifest,
creator="alice",
participants=[
("alice", ParticipantRole.INITIATOR),
("bob", ParticipantRole.RESPONDENT),
],
)
state = ConsultingState(initiator_sent=True, respondent_replied=True)
envelope = Envelope(
channel_id="s1",
sender_id="bob",
audience=["alice"],
event_type=EV_TEXT,
event_data={"text": "reply"},
)
result = adapter.on_accepted(meta, envelope, state)
assert result.next_state == ChannelState.CLOSED
assert result.auto_close_reason == "consulting_complete"
class TestConversationAdapter:
def test_validate_send_rejects_non_participant(self) -> None:
adapter = ConversationAdapter()
meta = _make_metadata(
manifest=adapter.manifest,
creator="alice",
participants=[
("alice", ParticipantRole.INITIATOR),
("bob", ParticipantRole.RESPONDENT),
],
)
envelope = Envelope(
channel_id="s1",
sender_id="eve",
audience=["alice"],
event_type=EV_TEXT,
event_data={"text": "x"},
)
with pytest.raises(ProtocolError, match="only accepts sends from participants"):
adapter.validate_send(meta, envelope, adapter.initial_state(meta))
def test_fold_advances_turn_count_and_speaker(self) -> None:
adapter = ConversationAdapter()
meta = _make_metadata(
manifest=adapter.manifest,
creator="alice",
participants=[
("alice", ParticipantRole.INITIATOR),
("bob", ParticipantRole.RESPONDENT),
],
)
state = adapter.initial_state(meta)
envelopes = [
Envelope(
channel_id="s1", sender_id="alice", audience=["bob"], event_type=EV_TEXT, event_data={"text": "1"}
),
Envelope(
channel_id="s1", sender_id="bob", audience=["alice"], event_type=EV_TEXT, event_data={"text": "2"}
),
Envelope(
channel_id="s1", sender_id="alice", audience=["bob"], event_type=EV_TEXT, event_data={"text": "3"}
),
]
for i, env in enumerate(envelopes, 1):
env.envelope_id = f"e{i}"
state = adapter.fold(env, state)
assert state.turn_count == 3
assert state.last_speaker_id == "alice"
assert state.last_envelope_id == "e3"
def test_fold_ignores_protocol_envelopes(self) -> None:
adapter = ConversationAdapter()
meta = _make_metadata(
manifest=adapter.manifest,
creator="alice",
participants=[
("alice", ParticipantRole.INITIATOR),
("bob", ParticipantRole.RESPONDENT),
],
)
state = adapter.initial_state(meta)
opened = Envelope(
channel_id="s1",
sender_id="alice",
audience=["alice", "bob"],
event_type=EV_CHANNEL_OPENED,
event_data={"channel_id": "s1"},
)
opened.envelope_id = "p1"
new = adapter.fold(opened, state)
assert new == state # untouched
class TestDiscussionAdapter:
def test_validate_create_rejects_unknown_ordering(self) -> None:
adapter = DiscussionAdapter()
meta = _make_metadata(
manifest=adapter.manifest,
creator="alice",
participants=[
("alice", ParticipantRole.INITIATOR),
("bob", ParticipantRole.PARTICIPANT),
],
knobs={"ordering": "weighted"}, # not supported
)
with pytest.raises(ProtocolError, match="ordering"):
adapter.validate_create(meta)
def test_validate_create_rejects_solo(self) -> None:
adapter = DiscussionAdapter()
meta = _make_metadata(
manifest=adapter.manifest,
creator="alice",
participants=[("alice", ParticipantRole.INITIATOR)],
)
with pytest.raises(ProtocolError, match="at least 2"):
adapter.validate_create(meta)
def test_validate_send_rejects_out_of_turn(self) -> None:
adapter = DiscussionAdapter()
meta = _make_metadata(
manifest=adapter.manifest,
creator="alice",
participants=[
("alice", ParticipantRole.INITIATOR),
("bob", ParticipantRole.PARTICIPANT),
("carol", ParticipantRole.PARTICIPANT),
],
)
state = adapter.initial_state(meta)
# alice is expected_next_speaker; bob trying to speak is out-of-turn
envelope = Envelope(
channel_id="s1",
sender_id="bob",
audience=None,
event_type=EV_TEXT,
event_data={"text": "jump in"},
)
with pytest.raises(ProtocolError, match="expects 'alice' to speak"):
adapter.validate_send(meta, envelope, state)
def test_round_robin_rotation_wraps_after_n_turns(self) -> None:
"""After N turns where N == participant count, expected_next_speaker
cycles back to the initiator."""
adapter = DiscussionAdapter()
meta = _make_metadata(
manifest=adapter.manifest,
creator="alice",
participants=[
("alice", ParticipantRole.INITIATOR),
("bob", ParticipantRole.PARTICIPANT),
("carol", ParticipantRole.PARTICIPANT),
],
)
state = adapter.initial_state(meta)
speakers = ["alice", "bob", "carol"]
for i, speaker in enumerate(speakers, 1):
assert state.expected_next_speaker == speaker
env = Envelope(
channel_id="s1",
sender_id=speaker,
audience=None,
event_type=EV_TEXT,
event_data={"text": f"turn-{i}"},
)
env.envelope_id = f"e{i}"
state = adapter.fold(env, state)
# After carol, rotation wraps back to alice.
assert state.expected_next_speaker == "alice"
assert state.turn_count == 3
def test_fold_with_unknown_sender_does_not_crash(self) -> None:
"""Sender not in participant_order — fold returns unchanged state.
validate_send normally gates this, but fold must be defensive
because hydrate replays the WAL deterministically and any
in-memory ParticipantOrder change between create and hydrate
could leave a stale envelope.
"""
adapter = DiscussionAdapter()
meta = _make_metadata(
manifest=adapter.manifest,
creator="alice",
participants=[
("alice", ParticipantRole.INITIATOR),
("bob", ParticipantRole.PARTICIPANT),
],
)
state = adapter.initial_state(meta)
ghost_env = Envelope(
channel_id="s1",
sender_id="ghost",
audience=None,
event_type=EV_TEXT,
event_data={"text": "x"},
)
ghost_env.envelope_id = "g1"
new = adapter.fold(ghost_env, state)
assert new == state
@pytest.mark.asyncio
async def test_validate_send_rejection_does_not_append_to_wal() -> None:
"""A rejected send must NOT touch the WAL.
post_envelope holds the WAL lock and runs validate_send BEFORE
appending. A rejected envelope must leave the WAL exactly as it was.
"""
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub.register(_agent("alice"))
bob = await hub.register(_agent("bob"))
await hub.register(_agent("carol"))
channel = await alice.open(type="discussion", target=["bob", "carol"])
pre_wal = await hub.read_wal(channel.channel_id)
pre_count = len(pre_wal)
# Bob tries to speak out-of-turn (alice is initiator, alice goes first).
envelope = Envelope(
channel_id=channel.channel_id,
sender_id=bob.agent_id,
audience=None,
event_type=EV_TEXT,
event_data={"text": "out of turn"},
)
with pytest.raises(ProtocolError):
await bob.send_envelope(envelope)
post_wal = await hub.read_wal(channel.channel_id)
assert len(post_wal) == pre_count
# And no envelope from bob with the rejected text.
assert all(e.event_data.get("text") != "out of turn" for e in post_wal)
await hub.close()
@pytest.mark.asyncio
async def test_hydrate_refolds_discussion_state_deterministically() -> None:
"""Close hub, reopen → refolded DiscussionState matches what fold
would produce live."""
with tempfile.TemporaryDirectory() as tmp:
store = DiskKnowledgeStore(tmp)
hub1 = await Hub.open(store, ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub1.register(_agent("alice"))
bob = await hub1.register(_agent("bob"))
carol = await hub1.register(_agent("carol"))
channel = await alice.open(type="discussion", target=["bob", "carol"])
sid = channel.channel_id
await channel.send("a-1", audience=None)
# bob's default handler does NOT auto-respond (ScriptedConfig
# returns ""), so we send manually for each speaker in turn.
await bob._hub_client.post_envelope(
Envelope(
channel_id=sid,
sender_id=bob.agent_id,
audience=None,
event_type=EV_TEXT,
event_data={"text": "b-1"},
)
)
await carol._hub_client.post_envelope(
Envelope(
channel_id=sid,
sender_id=carol.agent_id,
audience=None,
event_type=EV_TEXT,
event_data={"text": "c-1"},
)
)
live_state: DiscussionState = hub1._adapter_states[sid]
live_summary = (
live_state.expected_next_speaker,
live_state.last_speaker_id,
live_state.turn_count,
live_state.participant_order,
)
await hub1.close()
hub2 = await Hub.open(DiskKnowledgeStore(tmp), ttl_sweep_interval=0, expectation_sweep_interval=0)
rehydrated: DiscussionState = hub2._adapter_states[sid]
rehydrated_summary = (
rehydrated.expected_next_speaker,
rehydrated.last_speaker_id,
rehydrated.turn_count,
rehydrated.participant_order,
)
assert live_summary == rehydrated_summary
await hub2.close()