# AgentNet v0.1.45 Collaboration and Messaging Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Turn the existing communication kernel into a user-facing collaboration surface with authorized recipient resolution, messages, rooms, structured tasks, handoffs, cancellations, and response obligations.

**Architecture:** Add a general immutable `CollaborationScope` service above current human authority and exact harness identities. Use the existing non-enumerating directory to resolve plain-language recipient queries only inside the caller’s active scopes. Reuse `ConversationService`, `RoomService`, `MailboxService`, and `ResponseObligationService`; add no parallel message or receipt state machine.

**Tech Stack:** Python, Pydantic strict models, SQLite/PostgreSQL through `StoreBackend`, existing signed HTTP and local canonical bindings, pytest/Hypothesis.

## Global Constraints

- `CollaborationScope` is an authorization boundary, not a credential, identity, filesystem permission, model grant, secret grant, or execution grant.
- A verified local human principal owns the scope. Each member stores exact `(authority_kind, authority_id, harness_id)`: `principal` for a local human or `guest` for a host-local guest. Guest membership never maps to a sponsor principal or home-domain authority.
- Scope action/resource sets are canonical sorted tuples. Member digests bind kind, authority ID, and harness; unknown, stale, revoked, mismatched, or colliding identity state fails closed.
- Direct recipient resolution returns exact harness IDs only from active authorized scope. No global list, email-equality merge, wildcard authority expansion, sibling fallback, or principal/guest fallback.
- Room membership snapshots resolve to exact harness recipients at the authorized sequence; message delivery never dynamically broadens later.
- Transport acceptance, durable custody, recipient acknowledgement, processing, and external effect remain distinct facts.
- Requests requiring an answer create a durable `ResponseObligation` bound to one responsible exact harness.
- Downward assignments auto-accept only under active, scoped `may_assign`; upward/lateral/cross-domain/out-of-scope assignments remain `pending_human`.
- Affected IDs: `AUTH-001..010`, `ORG-001..006`, `COM-001..011`, `AVL-001..007`, `UX-001..004`, `SEC-001..006`.

---

### Task 1: Implement immutable collaboration scopes

**Files:**
- Create: `src/agentnet/authorization/collaboration_scope.py`
- Create: `src/agentnet/authorization/collaboration_scope_service.py`
- Modify: `src/agentnet/core/app.py:144-698`
- Modify: `src/agentnet/storage/collaboration_scope_schema.py` (created by the lifecycle plan’s schema-v7 task)
- Test: `tests/authorization/test_collaboration_scope.py`
- Test: `tests/authorization/test_collaboration_scope_service.py`

**Interfaces:**
- Consumes: schema-v7 collaboration tables, `StoreBackend`, `VerifiedActor`, `IssuanceAuthority`, policy revision, domain revocation epoch, and exact harness records.
- Produces:
```python
class CollaborationScope(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)
    schema_version: Literal["agentnet.collaboration-scope.v1"]
    scope_id: str
    domain_id: str
    owner_principal_id: str
    member_harness_ids: tuple[str, ...]
    allowed_actions: tuple[str, ...]
    allowed_resource_prefixes: tuple[str, ...]
    policy_revision: int
    domain_revocation_epoch: int
    expires_at: int | None
    revision: int
    state: Literal["active", "revoked", "expired"]

class CollaborationScopeService:
    def issue(self, *, actor: VerifiedActor, proposal: CollaborationScopeProposal, authority: IssuanceAuthority) -> CollaborationScope: ...
    def get_for_actor(self, *, actor: VerifiedActor, scope_id: str) -> CollaborationScope: ...
    def require(self, *, actor: VerifiedActor, action: str, resource: str, target_harness_ids: tuple[str, ...]) -> CollaborationScope: ...
    def revoke(self, *, actor: VerifiedActor, scope_id: str, expected_revision: int, reason: str, authority: IssuanceAuthority) -> CollaborationScope: ...
```

- [ ] **Step 1: Write failing model and authorization tests**

```python
def test_scope_requires_sorted_unique_members_actions_and_resources():
    with pytest.raises(ValidationError):
        CollaborationScope(member_harness_ids=("h2", "h1", "h1"), **VALID_SCOPE)


def test_scope_does_not_transfer_owner_data_authority(service, manager_actor, subordinate_harness):
    issue_message_scope(service, manager_actor, subordinate_harness)
    with pytest.raises(AuthorizationError):
        service.require(
            actor=actor_for(subordinate_harness),
            action="data.read",
            resource="principal:manager/private",
            target_harness_ids=(),
        )

```

- [ ] **Step 2: Verify tests fail**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/authorization/test_collaboration_scope.py tests/authorization/test_collaboration_scope_service.py
```
Expected: FAIL because the models and service are absent.

- [ ] **Step 3: Implement canonical validation and atomic issuance**

```python
ALLOWED_COLLABORATION_ACTIONS = frozenset({
    "message.send", "message.read", "message.acknowledge",
    "room.create", "room.read", "room.send", "room.member.add",
    "task.propose", "task.accept", "task.handoff", "task.cancel",
    "obligation.create", "obligation.respond",
    "artifact.send", "artifact.download",
})
```

Issue the scope and append audit in one transaction after rechecking current actor, authority decision, member status, policy revision, domain epoch, expiry, and scope digest. Revocation increments revision and prevents all new visibility/mutation immediately; queued ciphertext remains inaccessible to revoked endpoints.

- [ ] **Step 4: Verify service behavior and races**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/authorization/test_collaboration_scope.py tests/authorization/test_collaboration_scope_service.py
```
Expected: PASS including concurrent issue/revoke, expiry, stale revision, wrong domain, sibling harness, and policy-epoch cases.

- [ ] **Step 5: Commit**

```bash
git add src/agentnet/authorization/collaboration_scope.py src/agentnet/authorization/collaboration_scope_service.py src/agentnet/core/app.py tests/authorization/test_collaboration_scope.py tests/authorization/test_collaboration_scope_service.py
git commit -m "feat(auth): add collaboration scopes"
```

### Task 2: Add authorized plain-language recipient resolution

**Files:**
- Create: `src/agentnet/discovery/recipient_resolver.py`
- Modify: `src/agentnet/discovery/directory.py:21-215`
- Modify: `src/agentnet/core/app.py`
- Test: `tests/discovery/test_recipient_resolver.py`
- Test: `tests/discovery/test_non_enumerating_directory.py`

**Interfaces:**
- Consumes: `CollaborationScopeService`, `DirectoryService`, exact current harness status, and caller `VerifiedActor`.
- Produces:
```python
class ResolvedEndpoint(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)
    harness_id: str
    display_name: str
    harness_kind: str
    availability: Literal["online", "offline", "unknown"]
    scope_id: str

class AuthorizedRecipientResolver:
    def resolve(self, *, actor: VerifiedActor, query: str) -> tuple[ResolvedEndpoint, ...]: ...
```

- [ ] **Step 1: Write failing resolution tests**

```python
def test_resolves_only_visible_exact_endpoint(resolver, actor):
    assert resolver.resolve(actor=actor, query="the enrolled server") == (
        ResolvedEndpoint(
            harness_id="server-harness",
            display_name="The enrolled server",
            harness_kind="server",
            availability="online",
            scope_id="scope-1",
        ),
    )


def test_ambiguous_sibling_names_fail_without_listing(resolver, actor):
    with pytest.raises(ConflictError, match="recipient is ambiguous") as failure:
        resolver.resolve(actor=actor, query="Sergey's Pi")
    assert "harness-" not in str(failure.value)
```

- [ ] **Step 2: Verify tests fail**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/discovery/test_recipient_resolver.py
```
Expected: FAIL because the resolver does not exist.

- [ ] **Step 3: Implement bounded normalized matching**

```python
def normalize_query(value: str) -> str:
    normalized = " ".join(value.casefold().split())
    if not 1 <= len(normalized) <= 256:
        raise ValidationError("recipient query is outside the supported profile")
    return normalized
```

Fetch only directory rows visible to the actor's explicit principal-or-guest authority kind and ID, intersect with active collaboration-scope members, then match exact display name, approved alias, or exact harness kind alias. Return at most 20 records. Every candidate rechecks current domain, principal or host-local guest, exact harness, credential epoch/validity, and revocation state. Zero and ambiguous results use non-enumerating errors; never return hidden candidates.

- [ ] **Step 4: Verify resolver and disclosure tests**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/discovery/test_recipient_resolver.py tests/discovery/test_non_enumerating_directory.py
```
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add src/agentnet/discovery/recipient_resolver.py src/agentnet/discovery/directory.py src/agentnet/core/app.py tests/discovery
git commit -m "feat(discovery): resolve authorized exact recipients"
```

### Task 3: Bind communication operations to collaboration scope

**Files:**
- Modify: `src/agentnet/core/app.py:1253-1681`
- Modify: `src/agentnet/messaging/conversation.py:53-1083`
- Modify: `src/agentnet/rooms/service.py`
- Modify: `src/agentnet/mailbox/service.py:105-1105`
- Modify: `src/agentnet/organization/assignment.py`
- Test: `tests/integration/test_collaboration_scope_messaging.py`
- Test: `tests/messaging/test_conversation_semantics.py`
- Test: `tests/rooms/test_room_authority.py`

**Interfaces:**
- Consumes: exact resolved harness IDs and `CollaborationScopeService.require`.
- Produces existing Core methods with an added internal `scope_id` binding in the immutable event/audit context; public local tools still accept understandable recipient selections resolved before the call.

- [ ] **Step 1: Write failing end-to-end authorization tests**

```python
def test_message_event_binds_scope_and_exact_recipient(core, sender, target):
    result = core.send_message(
        actor=sender,
        recipients=(target.harness_id,),
        payload={"text": "Hello"},
        idempotency_key="message-scope-0001",
    )
    event = load_event(result["event_id"])
    assert event.recipient_harness_ids == (target.harness_id,)
    assert event.authorization_context["collaboration_scope_id"] == "scope-1"


def test_revoked_scope_blocks_queued_message_read(core, target, queued_event):
    revoke_scope("scope-1")
    with pytest.raises(AuthorizationError):
        core.mailbox(actor=target, after_cursor=0, limit=25)
```

- [ ] **Step 2: Verify tests fail**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/integration/test_collaboration_scope_messaging.py
```
Expected: FAIL because current message authorization does not bind the new scope.

- [ ] **Step 3: Add scope checks at semantic owners**

```python
scope = self.collaboration_scopes.require(
    actor=actor,
    action="message.send",
    resource=f"conversation:{conversation_id or 'direct'}",
    target_harness_ids=tuple(recipients),
)
```

Pass `scope.scope_id`, `scope.revision`, `scope.policy_revision`, and exact member snapshot into event construction and audit context. Recheck scope on mailbox visibility and acknowledgement. Do not modify `DeliveryFact` meanings.

- [ ] **Step 4: Verify messages, rooms, tasks, and obligations**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/integration/test_collaboration_scope_messaging.py tests/messaging tests/rooms tests/organization tests/bindings/test_response_obligation_tools.py
```
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add src/agentnet/core/app.py src/agentnet/messaging src/agentnet/rooms src/agentnet/mailbox src/agentnet/organization tests/integration/test_collaboration_scope_messaging.py tests/messaging tests/rooms tests/organization
git commit -m "feat(collaboration): authorize communication by scope"
```

### Task 4: Add ergonomic canonical collaboration tools

**Files:**
- Modify: `src/agentnet/bindings/tools.py`
- Modify: `src/agentnet/bindings/pi_extension.ts`
- Modify: `src/agentnet/bindings/remote_manager.py`
- Test: `tests/bindings/test_collaboration_tools.py`

**Interfaces:**
- Consumes: recipient resolver and current message/conversation/room/obligation methods.
- Produces user-facing tools with strict names:
```text
agentnet_recipient_resolve
agentnet_send
agentnet_inbox
agentnet_inbox_acknowledge
agentnet_conversation_create
agentnet_conversation_action
agentnet_conversation_thread
agentnet_room_create
agentnet_room_member_add
agentnet_room_get
agentnet_room_send
agentnet_obligation_inbox
agentnet_obligation_list
agentnet_obligation_get
agentnet_obligation_transition
agentnet_obligation_cancel
```

- [ ] **Step 1: Write failing tool behavior tests**

```python
def test_agentnet_send_resolves_display_name_before_sending(dispatcher):
    result = dispatcher.dispatch(
        CanonicalToolRequest(
            method="agentnet.send",
            arguments={"recipient_query": "the enrolled server", "payload": {"text": "hello"}, "idempotency_key": "send-display-0001"},
        )
    )
    assert result["recipient_harness_ids"] == ["server-harness"]
```

The final arguments model must accept exactly one of `recipient_query` or `recipients`; both/neither fail validation.

- [ ] **Step 2: Verify tests fail**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/bindings/test_collaboration_tools.py
```
Expected: FAIL because friendly recipient resolution is absent.

- [ ] **Step 3: Implement strict one-of resolution**

```python
class SendArguments(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)
    recipient_query: str | None = Field(default=None, min_length=1, max_length=256)
    recipients: tuple[str, ...] | None = Field(default=None, min_length=1, max_length=1000)
    payload: dict[str, Any]
    idempotency_key: str = Field(min_length=16, max_length=256)
    classification: Classification = Classification.C1_INTERNAL

    @model_validator(mode="after")
    def exact_recipient_form(self) -> "SendArguments":
        if (self.recipient_query is None) == (self.recipients is None):
            raise ValueError("exactly one recipient form is required")
        return self
```

Resolve and freeze exact recipients before authorizing/sending. Return display names and honest custody states, never message content from unrelated inbox rows.

- [ ] **Step 4: Verify canonical tools**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/bindings/test_collaboration_tools.py tests/bindings/test_canonical_tool_parity.py tests/bindings/test_remote_manager.py
```
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add src/agentnet/bindings tests/bindings/test_collaboration_tools.py
git commit -m "feat(tools): add ergonomic collaboration operations"
```

### Task 5: Implement obligation-driven background wake-up

**Files:**
- Modify: `src/agentnet/supervisor/integration.py`
- Modify: `src/agentnet/supervisor/live_gate.py`
- Modify: `src/agentnet/messaging/obligation.py`
- Test: `tests/integration/test_obligation_background_wakeup.py`
- Test: `tests/supervisor/test_live_delivery_watch.py`

**Interfaces:**
- Consumes: exact obligation `responsible_harness_id`, endpoint runtime state, and current scope.
- Produces: background wake only for active exact responsible endpoint; ordinary information remains quietly queued.

- [ ] **Step 1: Write failing wake-up tests**

```python
def test_request_wakes_only_responsible_endpoint(host, obligation, responsible, sibling):
    host.reconcile_once()
    assert wake_count(responsible.harness_id) == 1
    assert wake_count(sibling.harness_id) == 0


def test_ordinary_message_does_not_wake_endpoint(host, ordinary_event, target):
    host.reconcile_once()
    assert wake_count(target.harness_id) == 0
    assert attention_state(target.harness_id).unread_information == 1
```

- [ ] **Step 2: Verify tests fail**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/integration/test_obligation_background_wakeup.py
```
Expected: FAIL because host-level exact obligation wake is not composed.

- [ ] **Step 3: Implement content-free wake selection**

```python
def should_wake(item: MailboxItem, endpoint: EndpointBinding) -> bool:
    return (
        item.response_obligation is not None
        and item.response_obligation.responsible_harness_id == endpoint.harness_id
        and item.response_obligation.state in {"open", "in_progress"}
    )
```

Recheck current scope/credential before launching. Store only encrypted content-free counters and obligation IDs in the supervisor queue. Never inject a foreground turn.

- [ ] **Step 4: Verify background semantics**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/integration/test_obligation_background_wakeup.py tests/supervisor/test_live_delivery_watch.py tests/messaging/test_response_obligation.py
```
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add src/agentnet/supervisor src/agentnet/messaging/obligation.py tests/integration/test_obligation_background_wakeup.py tests/supervisor/test_live_delivery_watch.py
git commit -m "feat(supervisor): wake exact obligated endpoint"
```