test/network/test_human_client.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
"""``HumanClient`` integration tests.
Covers registration, push + pull surfaces, channel participation across
adapter types, and the rejection paths that keep the agent / human entry
points distinct.
"""
import asyncio
import pytest
from ag2 import Agent
from ag2.knowledge import MemoryKnowledgeStore
from ag2.network import (
EV_TEXT,
Channel,
Hub,
HubClient,
HumanClient,
LocalLink,
Passport,
Resume,
)
from ag2.network.channel import ChannelState
from ag2.testing import TestConfig
from ._helpers import wait_for_text_count
def _agent(name: str, *replies: str) -> Agent:
return Agent(name=name, config=TestConfig(*replies))
@pytest.mark.asyncio
async def test_register_human_stamps_kind_and_id() -> None:
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
hc = HubClient(link, hub=hub)
human = await hc.register_human(
Passport(name="reviewer-1"),
resume=Resume(summary="approves payouts"),
)
assert isinstance(human, HumanClient)
assert human.passport.kind == "human"
assert human.passport.effective_kind == "human"
assert human.agent_id # stamped
assert human.resume.summary == "approves payouts"
await hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_register_human_with_default_resume() -> None:
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
hc = HubClient(link, hub=hub)
human = await hc.register_human(Passport(name="dave"))
assert human.resume == Resume()
await hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_register_rejects_human_kind() -> None:
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
hc = HubClient(link, hub=hub)
with pytest.raises(ValueError, match="register_human"):
await hc.register(
_agent("bot"),
Passport(name="bot", kind="human"),
Resume(),
)
await hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_register_human_rejects_non_human_kind() -> None:
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
hc = HubClient(link, hub=hub)
with pytest.raises(ValueError, match="kind='human'"):
await hc.register_human(Passport(name="x", kind="agent"))
await hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_list_agents_filter_by_kind() -> None:
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
hc = HubClient(link, hub=hub)
await hub.register(_agent("alice"))
await hc.register_human(Passport(name="reviewer"))
agents = await hub.list_agents(kind="agent")
humans = await hub.list_agents(kind="human")
everyone = await hub.list_agents()
assert {p.name for p in agents} == {"alice"}
assert {p.name for p in humans} == {"reviewer"}
assert {p.name for p in everyone} == {"alice", "reviewer"}
await hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_consulting_human_responds_to_agent() -> None:
"""An agent opens consulting to a human; the human's reply closes the channel."""
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
human_hc = HubClient(link, hub=hub)
alice = await hub.register(_agent("alice"))
human = await human_hc.register_human(Passport(name="reviewer"))
# Human reacts to the inbound prompt by replying once.
received: list[str] = []
async def respond(envelope):
if envelope.event_type != EV_TEXT:
return
text = envelope.event_data.get("text", "")
received.append(text)
# Reply via the human's send().
await human.send(envelope.channel_id, "approved", audience=[envelope.sender_id])
human.on_envelope(respond)
channel = await alice.open(type="consulting", target="reviewer")
assert channel.state == ChannelState.ACTIVE
await channel.send("Please review payout #42", audience=[human.agent_id])
# Wait for the channel to close (consulting auto-closes after the respondent's reply).
close_env = await alice.wait_for_channel_event(
channel_id=channel.channel_id,
predicate=lambda e: e.event_type == "ag2.channel.closed",
timeout=5.0,
)
assert close_env is not None
assert received == ["Please review payout #42"]
await human_hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_pull_surface_returns_envelopes_in_order() -> None:
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
human_hc = HubClient(link, hub=hub)
alice = await hub.register(_agent("alice"))
human = await human_hc.register_human(Passport(name="reviewer"))
channel = await alice.open(type="conversation", target="reviewer")
# Drain channel-protocol envelopes first so the test waits on substantive.
await human.next_envelope(timeout=2.0) # invite or opened, application detail
# Send a substantive envelope from alice.
await channel.send("hello, reviewer")
envelope = await human.next_envelope(
predicate=lambda e: e.event_type == EV_TEXT,
timeout=2.0,
)
assert envelope.event_data["text"] == "hello, reviewer"
await human_hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_envelopes_iterator_streams_until_disconnect() -> None:
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
human_hc = HubClient(link, hub=hub)
alice = await hub.register(_agent("alice"))
human = await human_hc.register_human(Passport(name="reviewer"))
channel = await alice.open(type="conversation", target="reviewer")
await channel.send("one")
await channel.send("two")
seen: list[str] = []
async def consume() -> None:
async for envelope in human.envelopes():
if envelope.event_type == EV_TEXT:
seen.append(envelope.event_data["text"])
if len(seen) == 2:
await human.disconnect()
return
await asyncio.wait_for(consume(), timeout=5.0)
assert seen == ["one", "two"]
await human_hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_callback_exception_does_not_break_dispatch() -> None:
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
human_hc = HubClient(link, hub=hub)
alice = await hub.register(_agent("alice"))
human = await human_hc.register_human(Passport(name="reviewer"))
good_calls = 0
async def bad(envelope):
raise RuntimeError("boom")
async def good(envelope):
nonlocal good_calls
if envelope.event_type == EV_TEXT:
good_calls += 1
human.on_envelope(bad)
human.on_envelope(good)
channel = await alice.open(type="conversation", target="reviewer")
await channel.send("first")
await channel.send("second")
# Pull queue is still populated even though the push callback raised.
text_envs = []
while len(text_envs) < 2:
env = await human.next_envelope(timeout=2.0)
if env.event_type == EV_TEXT:
text_envs.append(env)
assert good_calls == 2
assert [e.event_data["text"] for e in text_envs] == ["first", "second"]
await human_hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_human_initiates_consulting() -> None:
"""Human opens consulting against an agent; agent's notify handler replies."""
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
human_hc = HubClient(link, hub=hub)
human = await human_hc.register_human(Passport(name="ops"))
await hub.register(_agent("bob", "bob's reply"))
channel = await human.open(type="consulting", target="bob")
assert isinstance(channel, Channel)
assert channel.state == ChannelState.ACTIVE
await channel.send("status check", audience=[await _id_for(human_hc, "bob")])
# Wait for bob's substantive reply on the human's pull queue.
reply = await human.next_envelope(
predicate=lambda e: e.event_type == EV_TEXT and e.sender_id != human.agent_id,
timeout=5.0,
)
assert reply.event_data["text"] == "bob's reply"
await human_hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_discussion_round_robin_with_human() -> None:
"""A human participates in a round-robin discussion alongside agents.
The human's reply callback gates on adapter state — the human only
posts when round-robin makes them the expected next speaker. This is
the realistic UI pattern: "show me whose turn it is, let the user
type only when it's theirs."
"""
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
human_hc = HubClient(link, hub=hub)
alice = await hub.register(_agent("alice", "alice says hi"))
await hub.register(_agent("bob", "bob says hi"))
human = await human_hc.register_human(Passport(name="reviewer"))
posted = {"done": False}
async def respond(envelope):
if envelope.event_type != EV_TEXT or envelope.sender_id == human.agent_id:
return
if posted["done"]:
return
state = await human_hc.adapter_state(envelope.channel_id)
if state is None or getattr(state, "expected_next_speaker", None) != human.agent_id:
return
posted["done"] = True
await human.send(envelope.channel_id, "reviewer notes")
human.on_envelope(respond)
channel = await alice.open(
type="discussion",
target=["bob", "reviewer"],
knobs={"ordering": "round_robin"},
)
await channel.send("kickoff")
# 3 substantive texts: kickoff (alice) + bob + human reviewer.
await wait_for_text_count(hub, channel.channel_id, expected=3, timeout=5.0)
wal = await hub.read_wal(channel.channel_id)
text_senders = [e.sender_id for e in wal if e.event_type == EV_TEXT]
assert human.agent_id in text_senders
# Round-robin order should have human after bob, both after alice's kickoff.
text_only = [e for e in wal if e.event_type == EV_TEXT]
assert text_only[0].sender_id == alice.agent_id
assert text_only[2].sender_id == human.agent_id
await human_hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_receive_chunk_is_noop() -> None:
"""Streaming chunks to a HumanClient don't raise."""
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
hc = HubClient(link, hub=hub)
human = await hc.register_human(Passport(name="reviewer"))
# Direct method call — chunks are an LLM-streaming concept; humans
# see only completed envelopes in V1.
await human.receive_chunk(object(), channel_id="x", parent_envelope_id="y")
await hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_post_envelope_after_disconnect_raises() -> None:
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
hc = HubClient(link, hub=hub)
human = await hc.register_human(Passport(name="x"))
await human.disconnect()
with pytest.raises(RuntimeError, match="disconnected"):
await human.send("c", "hi")
await hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_disconnect_wakes_blocked_next_envelope() -> None:
"""A consumer parked on next_envelope must wake when disconnect fires.
Regression for the HumanClient hang where disconnect() flipped the flag
but did not unblock pending Queue.get() calls.
"""
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
hc = HubClient(link, hub=hub)
human = await hc.register_human(Passport(name="x"))
async def disconnect_soon() -> None:
await asyncio.sleep(0.05)
await human.disconnect()
asyncio.create_task(disconnect_soon())
with pytest.raises(RuntimeError, match="disconnected"):
await asyncio.wait_for(human.next_envelope(timeout=5.0), timeout=2.0)
await hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_disconnect_wakes_blocked_envelopes_iterator() -> None:
"""An async-for over envelopes() must terminate when disconnect fires."""
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
hc = HubClient(link, hub=hub)
human = await hc.register_human(Passport(name="x"))
seen: list[object] = []
async def consume() -> None:
async for envelope in human.envelopes():
seen.append(envelope)
consumer = asyncio.create_task(consume())
await asyncio.sleep(0.05) # let consumer park on get()
await human.disconnect()
await asyncio.wait_for(consumer, timeout=2.0)
# Iterator exited cleanly; no envelopes ever arrived.
assert seen == []
await hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_disconnect_wakes_concurrent_pull_consumers() -> None:
"""Multiple consumers parked on the pull queue all wake on disconnect.
The sentinel-resender pattern means the first awaiter that observes
the disconnect re-enqueues the sentinel for siblings.
"""
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
hc = HubClient(link, hub=hub)
human = await hc.register_human(Passport(name="x"))
waiters = [asyncio.create_task(human.next_envelope(timeout=5.0)) for _ in range(3)]
await asyncio.sleep(0.05)
await human.disconnect()
results = await asyncio.gather(*waiters, return_exceptions=True)
for result in results:
assert isinstance(result, RuntimeError)
assert "disconnected" in str(result)
await hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_wait_for_channel_event_wakes_on_disconnect() -> None:
"""Per-channel waiters also wake when disconnect fires."""
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
hc = HubClient(link, hub=hub)
human = await hc.register_human(Passport(name="x"))
# Pre-create a per-channel inbox so the waiter parks on it.
human.ensure_channel_inbox("ch-pending")
async def disconnect_soon() -> None:
await asyncio.sleep(0.05)
await human.disconnect()
asyncio.create_task(disconnect_soon())
with pytest.raises(RuntimeError, match="disconnected"):
await human.wait_for_channel_event(
channel_id="ch-pending",
predicate=lambda _e: True,
timeout=5.0,
)
await hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_disconnect_idempotent() -> None:
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0)
link = LocalLink(hub)
hc = HubClient(link, hub=hub)
human = await hc.register_human(Passport(name="x"))
await human.disconnect()
await human.disconnect() # second call must not enqueue another sentinel
# The single sentinel is consumed and re-enqueued by next_envelope; if
# a second was inserted, two waiters would consume two sentinels
# without re-enqueue contention. We assert that the inbox has exactly
# one item (the sentinel) by reading it once and confirming a second
# read still finds one (the re-enqueue).
with pytest.raises(RuntimeError):
await human.next_envelope(timeout=1.0)
with pytest.raises(RuntimeError):
await human.next_envelope(timeout=1.0)
await hc.close()
await hub.close()
# ── Helpers ─────────────────────────────────────────────────────────────────
async def _id_for(hub_client: HubClient, name: str) -> str:
passport = await hub_client.get_agent(name)
assert passport.agent_id is not None
return passport.agent_id