Phase 12.3 — CollaborationChannel: real-time pub/sub workspace for cooperating agents

[web3guru888]

Phase 12.3 — CollaborationChannel: real-time shared workspace for cooperating agents

Parent Phase: Phase 12 — Multi-Agent Collaboration
Depends on: Phase 12.1 AgentRegistry (#352), Phase 12.2 NegotiationEngine (#355)
Next sub-phase: Phase 12.4 ConsensusVoting (planned)

---

Motivation

Once the NegotiationEngine assigns a task to a winning agent, that agent may need to collaborate in real-time with peers: sharing intermediate results, broadcasting state snapshots, and requesting specialised help mid-execution. CollaborationChannel provides a pub/sub workspace scoped to a single collaboration session, sitting between the negotiation layer and the consensus/coalition layers.

---

Data model

class MessageType(Enum):
    STATE_SNAPSHOT   = auto()   # agent broadcasts its current internal state
    PARTIAL_RESULT   = auto()   # intermediate computation result
    HELP_REQUEST     = auto()   # agent requests assistance from peers
    HELP_RESPONSE    = auto()   # peer responds to a help request
    SYNC_BARRIER     = auto()   # all-or-nothing synchronisation point
    HEARTBEAT        = auto()   # keep-alive / liveness probe

class ChannelStatus(Enum):
    OPEN    = auto()   # accepting posts and subscribers
    CLOSED  = auto()   # no new posts; existing subscribers drain
    EXPIRED = auto()   # TTL elapsed; channel evicted

@dataclass(frozen=True)
class ChannelMessage:
    message_id:   str
    channel_id:   str
    sender_did:   str
    message_type: MessageType
    payload:      Any
    reply_to:     str | None = None
    timestamp:    float = field(default_factory=time.time)

@dataclass(frozen=True)
class ChannelConfig:
    max_subscribers:    int   = 32
    max_history:        int   = 256
    ttl_seconds:        float = 600.0
    heartbeat_interval: float = 10.0
    require_membership: bool  = True

@dataclass(frozen=True)
class ChannelInfo:
    channel_id:    str
    owner_did:     str
    topic:         str
    status:        ChannelStatus
    member_dids:   frozenset[str]
    message_count: int
    created_at:    float
    last_activity: float

---

Protocols

class CollaborationChannel(Protocol):
    async def post(self, sender_did, message_type, payload, reply_to=None) -> ChannelMessage: ...
    async def subscribe(self, subscriber_did) -> AsyncIterator[ChannelMessage]: ...
    async def add_member(self, agent_did) -> None: ...
    async def remove_member(self, agent_did) -> None: ...
    async def close(self) -> None: ...
    async def info(self) -> ChannelInfo: ...

class ChannelManager(Protocol):
    async def create_channel(self, owner_did, topic, initial_members, config=None) -> CollaborationChannel: ...
    async def get_channel(self, channel_id) -> CollaborationChannel: ...
    async def list_channels(self, agent_did) -> list[ChannelInfo]: ...
    async def close_channel(self, channel_id) -> None: ...
    async def start(self) -> None: ...
    async def stop(self) -> None: ...

---

InMemoryCollaborationChannel

• Per-subscriber asyncio.Queue(maxsize=512) — slow subscribers drop rather than block
• Ring-buffer deque(maxlen=256) for history replay on late join
• Membership guard (require_membership=True) — non-members raise PermissionError
• _heartbeat_loop asyncio task — posts HEARTBEAT every heartbeat_interval seconds
• Channel auto-expires after ttl_seconds of inactivity (status → EXPIRED)

---

InMemoryChannelManager

• create_channel() — allocates UUID, starts _heartbeat_loop
• get_channel() / list_channels() — lookup by ID or DID membership
• close_channel() — removes from registry and closes channel
• _evict_loop() — background task cleaning CLOSED/EXPIRED channels every 60 s

---

CognitiveCycle integration

async def _open_collaboration_channel(
    self, task_id, winner_did, peer_dids
) -> None:
    ch = await self._channel_manager.create_channel(
        owner_did       = self.agent_did,
        topic           = f"task:{task_id}",
        initial_members = [winner_did] + peer_dids,
    )
    self._active_channels[task_id] = ch
    await self._federated_task_router.dispatch(
        task_id, winner_did, collaboration_channel_id=ch._channel_id
    )

---

Prometheus metrics (5)

Metric	Type	Labels
asi_collab_channels_total	Counter	—
asi_collab_channels_evicted_total	Counter	—
asi_collab_messages_total	Counter	channel, type
asi_collab_active_subscribers	Gauge	channel
asi_collab_queue_drop_total	Counter	channel

---

Test targets (12)

1. test_post_and_receive — single subscriber receives posted message
2. test_late_join_history_replay — history deque replayed on subscribe
3. test_require_membership_guard — non-member post raises PermissionError
4. test_max_subscribers_cap — 33rd subscriber raises RuntimeError
5. test_slow_subscriber_drop — full queue drops without blocking sender
6. test_heartbeat_posted — heartbeat message delivered after interval
7. test_ttl_expiry — channel transitions to EXPIRED after idle TTL
8. test_close_drains_subscribers — close() terminates subscribe generators
9. test_reply_threading — reply_to field chains messages correctly
10. test_create_and_get_channel — manager round-trip via channel_id
11. test_list_channels_by_member — list_channels() filters by DID membership
12. test_evict_loop_removes_expired — eviction task cleans up expired channels

---

Labels: enhancement, phase-12, multi-agent, collaboration

[web3guru888]

Implementation Notes

subscribe() skeleton — async generator with history replay

async def subscribe(
    self,
    subscriber_did: str,
) -> AsyncIterator[ChannelMessage]:
    if len(self._queues) >= self._cfg.max_subscribers:
        raise RuntimeError(
            f"Channel {self._channel_id} reached subscriber cap "
            f"({self._cfg.max_subscribers})"
        )
    q: asyncio.Queue[ChannelMessage] = asyncio.Queue(maxsize=512)

    # ── replay ring-buffer history for late joiners ──────────────────────
    for old_msg in self._history:
        q.put_nowait(old_msg)          # buffer already bounded; won't block

    self._queues[subscriber_did] = q
    COLLAB_ACTIVE_SUBSCRIBERS.labels(channel=self._channel_id).inc()
    try:
        while self._status == ChannelStatus.OPEN:
            try:
                msg = await asyncio.wait_for(
                    q.get(), timeout=self._cfg.heartbeat_interval
                )
                yield msg
            except asyncio.TimeoutError:
                continue               # heartbeat window; check status again
    finally:
        del self._queues[subscriber_did]
        COLLAB_ACTIVE_SUBSCRIBERS.labels(channel=self._channel_id).dec()

Design note: asyncio.wait_for with a timeout gives the loop a chance to re-check self._status even when the queue is idle. This ensures close() eventually terminates all subscribers without needing an explicit sentinel value.

---

post() slow-subscriber drop skeleton

for q in self._queues.values():
    try:
        q.put_nowait(msg)
    except asyncio.QueueFull:
        # slow subscriber: instrument and drop rather than block the sender
        COLLAB_QUEUE_DROP_TOTAL.labels(channel=self._channel_id).inc()

---

_heartbeat_loop() skeleton

async def _heartbeat_loop(self) -> None:
    while self._status == ChannelStatus.OPEN:
        await asyncio.sleep(self._cfg.heartbeat_interval)
        now = time.time()
        if now - self._last_activity > self._cfg.ttl_seconds:
            self._status = ChannelStatus.EXPIRED
            COLLAB_CHANNELS_EVICTED_TOTAL.inc()
            break
        # post heartbeat — bypasses membership check (internal sender)
        msg = ChannelMessage(
            message_id   = str(uuid.uuid4()),
            channel_id   = self._channel_id,
            sender_did   = self._owner_did,
            message_type = MessageType.HEARTBEAT,
            payload      = {"ts": now},
        )
        self._history.append(msg)
        self._msg_count += 1
        self._last_activity = now
        for q in self._queues.values():
            try:
                q.put_nowait(msg)
            except asyncio.QueueFull:
                COLLAB_QUEUE_DROP_TOTAL.labels(channel=self._channel_id).inc()

---

mypy compatibility table

Symbol	--strict	Notes
ChannelMessage	✅	payload: Any intentional
ChannelConfig	✅	all primitives
ChannelInfo	✅	frozenset[str] typed
InMemoryCollaborationChannel.subscribe	✅	returns AsyncIterator[ChannelMessage]
InMemoryChannelManager	✅	—

---

Test isolation skeleton

import asyncio
import pytest
from asi_build.multi_agent.collaboration_channel import (
    InMemoryCollaborationChannel,
    InMemoryChannelManager,
    ChannelConfig,
    MessageType,
    ChannelStatus,
    build_channel_manager,
)

@pytest.fixture
def cfg() -> ChannelConfig:
    return ChannelConfig(
        max_subscribers    = 4,
        max_history        = 8,
        ttl_seconds        = 300.0,
        heartbeat_interval = 30.0,
        require_membership = True,
    )

@pytest.mark.asyncio
async def test_post_and_receive(cfg):
    ch = InMemoryCollaborationChannel("ch-1", "did:peer:owner", "test-topic", cfg)
    await ch.add_member("did:peer:alice")

    received = []
    async def consumer():
        async for msg in ch.subscribe("did:peer:alice"):
            received.append(msg)
            break   # take one and stop

    t = asyncio.create_task(consumer())
    await ch.post("did:peer:alice", MessageType.PARTIAL_RESULT, {"v": 42})
    await asyncio.sleep(0.05)
    await ch.close()
    await t

    assert len(received) == 1
    assert received[0].payload == {"v": 42}
    assert received[0].sender_did == "did:peer:alice"

[web3guru888]

Phase 12.4 filed — #361

Phase 12.4 ConsensusVoting is now tracked in #361.

CollaborationChannel → ConsensusVoting integration:
When a ballot closes (PASSED / FAILED / EXPIRED / CANCELLED), ConsensusVoting publishes a msg_type = "CONSENSUS_RESULT" event to the CollaborationChannel workspace keyed by ballot_id. Any subscriber — including CognitiveCycle._ratify_high_stakes_goal() — receives the verdict immediately.

Phase 12 sub-phase tracker

Sub-phase	Component	Issue	Status
12.1	AgentRegistry	#352	✅
12.2	NegotiationEngine	#355	✅
12.3	CollaborationChannel	#358	✅
12.4	ConsensusVoting	#361	🟡
12.5	CoalitionFormation	—	⏳

Discussions: Show & Tell #362 · Q&A #363