test/network/test_at_least_once.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
"""At-least-once delivery: per-recipient inbox cursor advance via
``ReceiptFrame``, causation-id dedup index + ``find_envelope_by_causation``,
and replay of unacked envelopes on reconnect with
``HelloFrame.since_envelope_id``.
Two test patterns coexist here:
* HubClient-level: register through ``HubClient.register``; the default
in-process notify path auto-acks each delivered envelope so the
cursor naturally tracks the WAL head. Used for end-to-end coverage
of cursor / dedup / replay over the normal flow.
* Raw-link: register through ``Hub.register`` and bind a bare
``LocalLinkClient``. No auto-ack runs, so tests can drive
``ReceiptFrame`` traffic by hand. Used for the low-level cursor
invariants (monotonicity, malformed receipts, NACK behaviour).
"""
import asyncio
import json
from unittest import mock
import pytest
from ag2 import Agent
from ag2.knowledge import MemoryKnowledgeStore
from ag2.network import (
EV_CHANNEL_INVITE,
EV_CHANNEL_INVITE_ACK,
EV_TEXT,
Envelope,
HelloFrame,
Hub,
HubClient,
LocalLink,
NotifyFrame,
Passport,
ReceiptFrame,
RequestFrame,
Resume,
WelcomeFrame,
)
from ag2.network.ids import make_id
from ._helpers import ScriptedConfig, wait_for_text_count
def _post_frame(envelope: Envelope) -> RequestFrame:
"""Build a control-plane ``post_envelope`` request — the wire path a
raw client uses to post an envelope now that ``SendFrame`` is retired."""
return RequestFrame(request_id=make_id(), op="post_envelope", params={"envelope": envelope.to_dict()})
def _agent(name: str, *replies: str) -> Agent:
return Agent(name=name, config=ScriptedConfig(*replies))
async def _new_hub() -> Hub:
return await Hub.open(
MemoryKnowledgeStore(),
ttl_sweep_interval=0,
expectation_sweep_interval=0,
)
async def _drain_frame(link, timeout: float = 1.0):
"""Read one frame off a ``LocalLinkClient`` with a timeout."""
async def _first():
async for frame in link.frames():
return frame
raise AssertionError("link closed before a frame arrived")
return await asyncio.wait_for(_first(), timeout)
async def _raw_bob(hub: Hub, name: str = "bob"):
"""Register an agent directly via ``Hub.register`` and bind a bare
``LocalLinkClient`` to it. The returned link can be driven with
arbitrary frames without the default auto-ack interfering."""
passport = await hub.register_identity(Passport(name=name), Resume())
link = LocalLink(hub)
raw_client = link.client()
hub.bind_endpoint(raw_client.endpoint_id, passport.agent_id)
return passport, raw_client
async def _consume_invite_and_ack(raw_client, agent_id: str, *, timeout: float = 1.0) -> str:
"""Walk inbound frames until ``EV_CHANNEL_INVITE`` lands, post the
matching ``EV_CHANNEL_INVITE_ACK`` back via a ``post_envelope`` request,
return the channel id the invite was for. Helper for raw-bob test setups
that bypass the HubClient/AgentClient default handler."""
async def _scan() -> str:
async for frame in raw_client.frames():
if not isinstance(frame, NotifyFrame):
continue
env = frame.envelope
if env.event_type != EV_CHANNEL_INVITE:
continue
ack = Envelope(
channel_id=env.channel_id,
sender_id=agent_id,
audience=None,
event_type=EV_CHANNEL_INVITE_ACK,
event_data={"channel_id": env.channel_id},
causation_id=env.envelope_id,
)
await raw_client.send_frame(_post_frame(ack))
return env.channel_id
raise AssertionError("link closed before any invite arrived")
return await asyncio.wait_for(_scan(), timeout)
async def _await_cursor(hub, agent_id, channel_id, predicate, *, timeout=5.0):
"""Poll the per-channel inbox cursor until ``predicate(cursor)`` holds.
Acks travel an async path (notify handler -> ReceiptFrame -> hub), so a
fixed sleep races with delivery on a slow runner; poll the cursor instead
of guessing a delay.
"""
deadline = asyncio.get_event_loop().time() + timeout
while asyncio.get_event_loop().time() < deadline:
if predicate(hub.inbox_cursor(agent_id, channel_id)):
return
await asyncio.sleep(0.01)
class TestReceiptCursorAdvance:
@pytest.mark.asyncio
async def test_auto_ack_advances_cursor_to_wal_head(self) -> None:
"""End-to-end: a HubClient-served recipient auto-acks every
delivery, so the cursor follows the WAL head."""
hub = await _new_hub()
alice = await hub.register(_agent("alice"))
bob = await hub.register(_agent("bob"))
try:
channel = await alice.open(type="discussion", target=["bob"])
await channel.send("hello bob")
await wait_for_text_count(hub, channel.channel_id, 1)
wal = await hub.read_wal(channel.channel_id)
text_env = next(e for e in wal if e.event_type == EV_TEXT)
await _await_cursor(hub, bob.agent_id, channel.channel_id, lambda c: c == text_env.envelope_id)
assert hub.inbox_cursor(bob.agent_id, channel.channel_id) == text_env.envelope_id
cursor_blob = await hub._store.read(f"/agents/{bob.agent_id}/inbox.cursors.json")
assert json.loads(cursor_blob)[channel.channel_id] == text_env.envelope_id
finally:
await hub.close()
@pytest.mark.asyncio
async def test_cursor_tracks_wal_head_when_clock_does_not_advance(self) -> None:
"""Regression: envelope ids must sort in WAL-append order even when
the wall clock does not advance between mints — coarse-resolution
timers (Windows ~15 ms) or many envelopes inside one tick. The hub
mints through a strictly-monotonic source, so the cursor follows the
WAL head instead of landing on whichever same-tick envelope drew the
largest random suffix. Frozen here (the worst case of a coarse clock)
so the assertion would fail ~always without strict monotonicity."""
with mock.patch("time.time_ns", return_value=1_700_000_000_000_000_000):
hub = await _new_hub()
alice = await hub.register(_agent("alice"))
bob = await hub.register(_agent("bob"))
try:
# conversation channel: no turn-taking, so one sender may
# post many envelopes back-to-back (all within the frozen tick).
channel = await alice.open(type="conversation", target=["bob"])
for i in range(20):
await channel.send(f"msg {i}")
await wait_for_text_count(hub, channel.channel_id, 20)
wal = await hub.read_wal(channel.channel_id)
env_ids = [e.envelope_id for e in wal]
assert env_ids == sorted(env_ids) # sort order == append order
assert len(set(env_ids)) == len(env_ids) # all unique
last_text = [e for e in wal if e.event_type == EV_TEXT][-1]
await _await_cursor(hub, bob.agent_id, channel.channel_id, lambda c: c == last_text.envelope_id)
assert hub.inbox_cursor(bob.agent_id, channel.channel_id) == last_text.envelope_id
finally:
await hub.close()
@pytest.mark.asyncio
async def test_ack_is_monotonic_under_out_of_order_receipts(self) -> None:
"""Acks for older envelope ids must not rewind the cursor."""
hub = await _new_hub()
alice = await hub.register(_agent("alice"))
bob_passport, bob_raw = await _raw_bob(hub)
try:
channel_open = asyncio.create_task(alice.open(type="discussion", target=["bob"]))
channel_id = await _consume_invite_and_ack(bob_raw, bob_passport.agent_id)
channel = await channel_open
assert channel.channel_id == channel_id
await channel.send("first text")
# Wait for bob to receive a frame for the text envelope.
async for frame in bob_raw.frames():
if isinstance(frame, NotifyFrame) and frame.envelope.event_type == EV_TEXT:
break
wal = await hub.read_wal(channel_id)
text_envs = [e for e in wal if e.event_type == EV_TEXT]
assert len(text_envs) == 1
text_env = text_envs[0]
opened_env = next(e for e in wal if e.event_type != EV_TEXT and e.sender_id == alice.agent_id)
# text_env was stamped after opened_env so it's strictly later.
assert text_env.envelope_id > opened_env.envelope_id
# Ack the newer (text) first.
await bob_raw.send_frame(
ReceiptFrame(
envelope_id=text_env.envelope_id,
status="ack",
recipient_id=bob_passport.agent_id,
channel_id=channel_id,
)
)
# Then ack an older envelope. Cursor must NOT rewind.
await bob_raw.send_frame(
ReceiptFrame(
envelope_id=opened_env.envelope_id,
status="ack",
recipient_id=bob_passport.agent_id,
channel_id=channel_id,
)
)
await _await_cursor(hub, bob_passport.agent_id, channel_id, lambda c: c == text_env.envelope_id)
assert hub.inbox_cursor(bob_passport.agent_id, channel_id) == text_env.envelope_id
finally:
await bob_raw.close()
await hub.close()
@pytest.mark.asyncio
async def test_nack_leaves_cursor_unchanged(self) -> None:
"""A NACK does not advance the cursor — without an ack later,
the envelope will be replayed on the next reconnect."""
hub = await _new_hub()
alice = await hub.register(_agent("alice"))
bob_passport, bob_raw = await _raw_bob(hub)
try:
channel_open = asyncio.create_task(alice.open(type="discussion", target=["bob"]))
await _consume_invite_and_ack(bob_raw, bob_passport.agent_id)
channel = await channel_open
await channel.send("hello")
async for frame in bob_raw.frames():
if isinstance(frame, NotifyFrame) and frame.envelope.event_type == EV_TEXT:
text_env = frame.envelope
break
cursor_before = hub.inbox_cursor(bob_passport.agent_id, channel.channel_id)
await bob_raw.send_frame(
ReceiptFrame(
envelope_id=text_env.envelope_id,
status="nack",
recipient_id=bob_passport.agent_id,
channel_id=channel.channel_id,
reason="processing failed",
)
)
await asyncio.sleep(0.05)
cursor_after = hub.inbox_cursor(bob_passport.agent_id, channel.channel_id)
assert cursor_after == cursor_before
finally:
await bob_raw.close()
await hub.close()
@pytest.mark.asyncio
async def test_receipt_without_recipient_id_is_ignored(self) -> None:
"""A receipt missing ``recipient_id`` cannot be attributed and
must not affect any cursor — defensive against legacy or
malformed clients."""
hub = await _new_hub()
alice = await hub.register(_agent("alice"))
bob_passport, bob_raw = await _raw_bob(hub)
try:
channel_open = asyncio.create_task(alice.open(type="discussion", target=["bob"]))
await _consume_invite_and_ack(bob_raw, bob_passport.agent_id)
channel = await channel_open
await channel.send("hello")
async for frame in bob_raw.frames():
if isinstance(frame, NotifyFrame) and frame.envelope.event_type == EV_TEXT:
text_env = frame.envelope
break
cursor_before = hub.inbox_cursor(bob_passport.agent_id, channel.channel_id)
await bob_raw.send_frame(
ReceiptFrame(envelope_id=text_env.envelope_id, status="ack") # recipient_id="" default
)
await asyncio.sleep(0.05)
cursor_after = hub.inbox_cursor(bob_passport.agent_id, channel.channel_id)
assert cursor_after == cursor_before
finally:
await bob_raw.close()
await hub.close()
@pytest.mark.asyncio
async def test_receipt_without_channel_id_is_ignored(self) -> None:
"""An ack the hub can't attribute to a channel cannot advance any
per-channel cursor — receipts must name their channel."""
hub = await _new_hub()
alice = await hub.register(_agent("alice"))
bob_passport, bob_raw = await _raw_bob(hub)
try:
channel_open = asyncio.create_task(alice.open(type="discussion", target=["bob"]))
await _consume_invite_and_ack(bob_raw, bob_passport.agent_id)
channel = await channel_open
await channel.send("hello")
async for frame in bob_raw.frames():
if isinstance(frame, NotifyFrame) and frame.envelope.event_type == EV_TEXT:
text_env = frame.envelope
break
cursor_before = hub.inbox_cursor(bob_passport.agent_id, channel.channel_id)
await bob_raw.send_frame(
# recipient_id set, channel_id omitted (default "").
ReceiptFrame(
envelope_id=text_env.envelope_id,
status="ack",
recipient_id=bob_passport.agent_id,
)
)
await asyncio.sleep(0.05)
cursor_after = hub.inbox_cursor(bob_passport.agent_id, channel.channel_id)
assert cursor_after == cursor_before == ""
finally:
await bob_raw.close()
await hub.close()
class TestCausationIndex:
@pytest.mark.asyncio
async def test_find_returns_prior_envelope_for_same_causation_id(self) -> None:
hub = await _new_hub()
alice = await hub.register(_agent("alice"))
await hub.register(_agent("bob"))
try:
channel = await alice.open(type="discussion", target=["bob"])
await channel.send("hello", causation_id="logical-retry-1")
await wait_for_text_count(hub, channel.channel_id, 1)
found = await hub.find_envelope_by_causation(
channel.channel_id,
sender_id=alice.agent_id,
causation_id="logical-retry-1",
)
assert found is not None
assert found.event_data.get("text") == "hello"
assert found.sender_id == alice.agent_id
finally:
await hub.close()
@pytest.mark.asyncio
async def test_find_returns_none_when_causation_id_unseen(self) -> None:
hub = await _new_hub()
alice = await hub.register(_agent("alice"))
await hub.register(_agent("bob"))
try:
channel = await alice.open(type="discussion", target=["bob"])
await channel.send("hello")
await wait_for_text_count(hub, channel.channel_id, 1)
found = await hub.find_envelope_by_causation(
channel.channel_id,
sender_id=alice.agent_id,
causation_id="never-existed",
)
assert found is None
finally:
await hub.close()
@pytest.mark.asyncio
async def test_envelope_with_null_causation_is_not_indexed(self) -> None:
"""Envelopes posted without a causation_id (the default) must
not consume an index slot under any falsy key."""
hub = await _new_hub()
alice = await hub.register(_agent("alice"))
await hub.register(_agent("bob"))
try:
channel = await alice.open(type="discussion", target=["bob"])
await channel.send("hello") # No causation_id supplied.
await wait_for_text_count(hub, channel.channel_id, 1)
wal = await hub.read_wal(channel.channel_id)
text_env = next(e for e in wal if e.event_type == EV_TEXT)
assert text_env.causation_id is None
# No "" / None / falsy key should be present for alice in this channel.
assert (channel.channel_id, alice.agent_id, "") not in hub._causation_index
for (cid, sid, _key), env_id in hub._causation_index.items():
if cid == channel.channel_id and sid == alice.agent_id:
# The only matches must be from the alice-sender protocol
# envelopes (invite carries no causation; opened carries none).
assert env_id != text_env.envelope_id
finally:
await hub.close()
@pytest.mark.asyncio
async def test_terminal_channel_prunes_index_entries(self) -> None:
hub = await _new_hub()
alice = await hub.register(_agent("alice"))
await hub.register(_agent("bob"))
try:
channel = await alice.open(type="discussion", target=["bob"])
await channel.send("hello", causation_id="will-be-pruned")
await wait_for_text_count(hub, channel.channel_id, 1)
assert (channel.channel_id, alice.agent_id, "will-be-pruned") in hub._causation_index
await hub.close_channel(channel.channel_id, reason="test")
assert not any(k[0] == channel.channel_id for k in hub._causation_index)
found = await hub.find_envelope_by_causation(
channel.channel_id,
sender_id=alice.agent_id,
causation_id="will-be-pruned",
)
assert found is None
finally:
await hub.close()
@pytest.mark.asyncio
async def test_hydrate_rebuilds_index_from_wal(self) -> None:
"""Restarting the hub against the same store re-populates the
causation index from each active channel's persisted WAL."""
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub.register(_agent("alice"))
await hub.register(_agent("bob"))
# Use two separate channels so alice can post two causation-keyed
# texts back-to-back without round-robin interference.
chan_a = await alice.open(type="discussion", target=["bob"])
chan_b = await alice.open(type="discussion", target=["bob"])
await chan_a.send("first", causation_id="c-first")
await chan_b.send("second", causation_id="c-second")
await wait_for_text_count(hub, chan_a.channel_id, 1)
await wait_for_text_count(hub, chan_b.channel_id, 1)
chan_a_id = chan_a.channel_id
chan_b_id = chan_b.channel_id
alice_id = alice.agent_id
await hub.close()
# Re-open against the same store: fresh memory, identical disk.
hub2 = await Hub.open(store, ttl_sweep_interval=0, expectation_sweep_interval=0)
try:
assert (chan_a_id, alice_id, "c-first") in hub2._causation_index
assert (chan_b_id, alice_id, "c-second") in hub2._causation_index
found = await hub2.find_envelope_by_causation(chan_b_id, sender_id=alice_id, causation_id="c-second")
assert found is not None
assert found.event_data.get("text") == "second"
finally:
await hub2.close()
class TestReplayOnReconnect:
@pytest.mark.asyncio
async def test_replay_redelivers_unacked_envelopes_past_since(self) -> None:
"""While bob is disconnected, alice posts envelopes targeted at
him via ``hub.post_envelope`` (using ``ag2.task.*`` event types
which bypass adapter round-robin). Reconnecting bob with
``since_envelope_id`` set to his last seen id replays only the
envelopes posted past that mark, in WAL order."""
hub = await _new_hub()
link = LocalLink(hub)
alice_hc = HubClient(link, hub=hub)
bob_hc = HubClient(link, hub=hub)
alice = await alice_hc.register(_agent("alice"), Passport(name="alice"), Resume())
bob = await bob_hc.register(_agent("bob"), Passport(name="bob"), Resume())
try:
channel = await alice.open(type="discussion", target=["bob"])
await channel.send("seen-before-disconnect")
await wait_for_text_count(hub, channel.channel_id, 1)
await asyncio.sleep(0.05)
wal = await hub.read_wal(channel.channel_id)
seen_env = next(e for e in wal if e.event_type == EV_TEXT)
await bob_hc.close()
# Post two task-progress envelopes addressed to bob while he
# is offline. Task events bypass adapter ``validate_send``
# so they don't need round-robin compliance — they land in
# the WAL and dispatch to bob's (now-missing) endpoint.
missed_1 = Envelope(
channel_id=channel.channel_id,
sender_id=alice.agent_id,
audience=[bob.agent_id],
event_type="ag2.task.progress",
event_data={"step": "missed-1"},
)
await hub.post_envelope(missed_1)
missed_2 = Envelope(
channel_id=channel.channel_id,
sender_id=alice.agent_id,
audience=[bob.agent_id],
event_type="ag2.task.progress",
event_data={"step": "missed-2"},
)
await hub.post_envelope(missed_2)
raw_link = LocalLink(hub)
raw_client = raw_link.client()
try:
await raw_client.send_frame(HelloFrame(name="bob", since_envelope_id=seen_env.envelope_id))
welcome = await _drain_frame(raw_client)
assert isinstance(welcome, WelcomeFrame)
replayed: list[Envelope] = []
for _ in range(2):
frame = await _drain_frame(raw_client, timeout=1.0)
assert isinstance(frame, NotifyFrame)
assert frame.recipient_id == bob.agent_id
replayed.append(frame.envelope)
steps = [e.event_data.get("step") for e in replayed]
assert steps == ["missed-1", "missed-2"]
finally:
await raw_client.close()
finally:
await alice_hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_replay_skips_terminal_channels(self) -> None:
"""A channel that closed while the recipient was offline is
pruned from replay scope — replay only walks active channels."""
hub = await _new_hub()
link = LocalLink(hub)
alice_hc = HubClient(link, hub=hub)
bob_hc = HubClient(link, hub=hub)
alice = await alice_hc.register(_agent("alice"), Passport(name="alice"), Resume())
await bob_hc.register(_agent("bob"), Passport(name="bob"), Resume())
try:
channel = await alice.open(type="discussion", target=["bob"])
await channel.send("seen")
await wait_for_text_count(hub, channel.channel_id, 1)
await asyncio.sleep(0.05)
wal = await hub.read_wal(channel.channel_id)
seen_env = next(e for e in wal if e.event_type == EV_TEXT)
await bob_hc.close()
await hub.close_channel(channel.channel_id, reason="offline test")
raw_link = LocalLink(hub)
raw_client = raw_link.client()
try:
await raw_client.send_frame(HelloFrame(name="bob", since_envelope_id=seen_env.envelope_id))
welcome = await _drain_frame(raw_client)
assert isinstance(welcome, WelcomeFrame)
with pytest.raises(TimeoutError):
await _drain_frame(raw_client, timeout=0.25)
finally:
await raw_client.close()
finally:
await alice_hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_replay_skips_envelopes_at_or_below_cursor(self) -> None:
"""When the persisted cursor is already past an envelope, it
is not replayed even when ``since_envelope_id`` is empty."""
hub = await _new_hub()
link = LocalLink(hub)
alice_hc = HubClient(link, hub=hub)
bob_hc = HubClient(link, hub=hub)
alice = await alice_hc.register(_agent("alice"), Passport(name="alice"), Resume())
bob = await bob_hc.register(_agent("bob"), Passport(name="bob"), Resume())
try:
channel = await alice.open(type="discussion", target=["bob"])
await channel.send("first")
await wait_for_text_count(hub, channel.channel_id, 1)
wal = await hub.read_wal(channel.channel_id)
text_env = next(e for e in wal if e.event_type == EV_TEXT)
# Wait until bob has acked through the WAL head (not merely acked
# *something*), so the reconnect below has nothing left to replay.
await _await_cursor(hub, bob.agent_id, channel.channel_id, lambda c: c == text_env.envelope_id)
cursor = hub.inbox_cursor(bob.agent_id, channel.channel_id)
assert cursor == text_env.envelope_id
await bob_hc.close()
raw_link = LocalLink(hub)
raw_client = raw_link.client()
try:
await raw_client.send_frame(HelloFrame(name="bob", since_envelope_id=""))
welcome = await _drain_frame(raw_client)
assert isinstance(welcome, WelcomeFrame)
# Cursor wins over the empty since_envelope_id, so no
# replay frames arrive.
with pytest.raises(TimeoutError):
await _drain_frame(raw_client, timeout=0.25)
finally:
await raw_client.close()
finally:
await alice_hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_reconnect_without_since_does_not_replay(self) -> None:
"""A plain reconnect (``since_envelope_id=None``) gets a Welcome
but no historic frames — replay is opt-in."""
hub = await _new_hub()
link = LocalLink(hub)
alice_hc = HubClient(link, hub=hub)
bob_hc = HubClient(link, hub=hub)
alice = await alice_hc.register(_agent("alice"), Passport(name="alice"), Resume())
await bob_hc.register(_agent("bob"), Passport(name="bob"), Resume())
try:
channel = await alice.open(type="discussion", target=["bob"])
await channel.send("miscible")
await wait_for_text_count(hub, channel.channel_id, 1)
await bob_hc.close()
raw_link = LocalLink(hub)
raw_client = raw_link.client()
try:
await raw_client.send_frame(HelloFrame(name="bob")) # no since
welcome = await _drain_frame(raw_client)
assert isinstance(welcome, WelcomeFrame)
with pytest.raises(TimeoutError):
await _drain_frame(raw_client, timeout=0.25)
finally:
await raw_client.close()
finally:
await alice_hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_ack_in_one_channel_does_not_suppress_replay_in_another(self) -> None:
"""The delivery cursor is per (recipient, channel). Acking a
newer envelope in one channel must not skip an older unacked
envelope in a different channel on reconnect — the failure mode
a single global high-water cursor would cause."""
hub = await _new_hub()
link = LocalLink(hub)
alice_hc = HubClient(link, hub=hub)
bob_hc = HubClient(link, hub=hub)
alice = await alice_hc.register(_agent("alice"), Passport(name="alice"), Resume())
bob = await bob_hc.register(_agent("bob"), Passport(name="bob"), Resume())
try:
chan_a = await alice.open(type="discussion", target=["bob"])
chan_b = await alice.open(type="discussion", target=["bob"])
await asyncio.sleep(0.05) # bob auto-acks both opened envelopes
await bob_hc.close()
await asyncio.sleep(0.05)
# Channel B's envelope is stamped first (lower id) and stays
# unacked; channel A's is stamped later (higher id) and gets
# acked. A single global cursor at channel A's id would
# wrongly suppress channel B's replay.
missed_b = Envelope(
channel_id=chan_b.channel_id,
sender_id=alice.agent_id,
audience=[bob.agent_id],
event_type="ag2.task.progress",
event_data={"step": "missed-b"},
)
await hub.post_envelope(missed_b)
missed_a = Envelope(
channel_id=chan_a.channel_id,
sender_id=alice.agent_id,
audience=[bob.agent_id],
event_type="ag2.task.progress",
event_data={"step": "missed-a"},
)
missed_a_id = await hub.post_envelope(missed_a)
# Reconnect once; ack ONLY channel A's envelope.
raw_link = LocalLink(hub)
raw1 = raw_link.client()
try:
await raw1.send_frame(HelloFrame(name="bob", since_envelope_id=""))
welcome = await _drain_frame(raw1)
assert isinstance(welcome, WelcomeFrame)
await raw1.send_frame(
ReceiptFrame(
envelope_id=missed_a_id,
status="ack",
recipient_id=bob.agent_id,
channel_id=chan_a.channel_id,
)
)
await asyncio.sleep(0.05)
finally:
await raw1.close()
# Reconnect again: channel B's envelope must still replay
# (its cursor never advanced); channel A's must not.
raw2 = raw_link.client()
try:
await raw2.send_frame(HelloFrame(name="bob", since_envelope_id=""))
welcome = await _drain_frame(raw2)
assert isinstance(welcome, WelcomeFrame)
steps: list[str | None] = []
while True:
try:
frame = await _drain_frame(raw2, timeout=0.5)
except TimeoutError:
break
if isinstance(frame, NotifyFrame) and frame.envelope.event_type == "ag2.task.progress":
steps.append(frame.envelope.event_data.get("step"))
assert "missed-b" in steps, f"channel B replay was suppressed; saw {steps}"
assert "missed-a" not in steps, f"channel A re-replayed after ack; saw {steps}"
finally:
await raw2.close()
finally:
await alice_hc.close()
await hub.close()
class TestCursorPersistence:
@pytest.mark.asyncio
async def test_hydrate_restores_cursor(self) -> None:
store = MemoryKnowledgeStore()
hub = await Hub.open(store, ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub.register(_agent("alice"))
bob = await hub.register(_agent("bob"))
channel = await alice.open(type="discussion", target=["bob"])
await channel.send("hello")
await wait_for_text_count(hub, channel.channel_id, 1)
bob_id = bob.agent_id
channel_id = channel.channel_id
wal = await hub.read_wal(channel_id)
text_env = next(e for e in wal if e.event_type == EV_TEXT)
await _await_cursor(hub, bob_id, channel_id, lambda c: c == text_env.envelope_id)
expected_cursor = hub.inbox_cursor(bob_id, channel_id)
assert expected_cursor == text_env.envelope_id
await hub.close()
hub2 = await Hub.open(store, ttl_sweep_interval=0, expectation_sweep_interval=0)
try:
assert hub2.inbox_cursor(bob_id, channel_id) == expected_cursor
finally:
await hub2.close()
@pytest.mark.asyncio
async def test_unregister_clears_cursor_and_disk_file(self) -> None:
hub = await _new_hub()
alice = await hub.register(_agent("alice"))
bob = await hub.register(_agent("bob"))
try:
channel = await alice.open(type="discussion", target=["bob"])
await channel.send("hello")
await wait_for_text_count(hub, channel.channel_id, 1)
bob_id = bob.agent_id
await _await_cursor(hub, bob_id, channel.channel_id, lambda c: c != "")
assert bob_id in hub._inbox_cursors
await bob.unregister()
assert bob_id not in hub._inbox_cursors
assert await hub._store.read(f"/agents/{bob_id}/inbox.cursors.json") is None
finally:
await hub.close()
class TestRedeliveryIdempotency:
@pytest.mark.asyncio
async def test_redelivered_trigger_does_not_double_post(self) -> None:
"""At-least-once redelivery of a trigger the agent already
answered must not re-run the LLM or post a second reply. The
default handler short-circuits on the causation index. Uses a
conversation channel (free-form turns) so only the dedup guard —
not turn ordering — can prevent the second reply."""
hub = await _new_hub()
alice = await hub.register(_agent("alice"))
bob = await hub.register(_agent("bob", "reply-A", "reply-B"))
try:
channel = await alice.open(type="conversation", target=["bob"])
await channel.send("question", audience=[bob.agent_id])
# alice's question + bob's first reply.
wal = await wait_for_text_count(hub, channel.channel_id, 2)
trigger = next(e for e in wal if e.event_type == EV_TEXT and e.sender_id == alice.agent_id)
bob_first = [e for e in wal if e.event_type == EV_TEXT and e.sender_id == bob.agent_id]
assert len(bob_first) == 1
assert bob_first[0].event_data.get("text") == "reply-A"
# Redeliver the exact same trigger envelope.
await bob.receive(trigger)
await asyncio.sleep(0.1)
wal_after = await hub.read_wal(channel.channel_id)
bob_after = [e for e in wal_after if e.event_type == EV_TEXT and e.sender_id == bob.agent_id]
assert len(bob_after) == 1, f"redelivery double-posted: {[e.event_data for e in bob_after]}"
assert bob_after[0].event_data.get("text") == "reply-A"
finally:
await hub.close()