# AgentNet v0.1.45 Exact-Agent Adapter 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:** Make AgentNet tools available in ordinary OMP, Pi, Claude, Codex, and Antigravity conversations while guaranteeing that only the exact addressed enrolled endpoint can receive or process its work.

**Architecture:** Replace the interactive `manager-run` workaround with a package-owned host supervisor that manages multiple isolated endpoint runtimes. Each endpoint gets its own process-bound actor, capability root, binding descriptor, adapter generation, mailbox cursor, queue namespace, and background worker. Pi and OMP use the direct extension binding; Claude, Codex, and Antigravity use the canonical MCP binding.

**Tech Stack:** Python asyncio/process supervision, Unix IPC and Windows named pipes, MCP 1.28.1, Pi TypeScript extension, harness launch specs, Pydantic, pytest.

## Global Constraints

- One enrolled agent instance equals one durable endpoint. A process restart does not change its harness ID; creating a new profile creates a new enrollment.
- OMP and Pi may share extension implementation code, but never identity files, mailbox cursors, adapter generations, state roots, or capability roots.
- The host supervisor may compose multiple endpoints but holds no positive human authority of its own.
- Endpoint activation uses the current enrolled credential and measured local process; config or payload claims cannot select a different actor.
- Revocation, expiry, credential rotation, generation mismatch, PID reuse, stale descriptor, symlinked root, or ambiguous profile blocks the endpoint immediately.
- Background work never enters, steals, or mutates the foreground conversation. Passive indication is content-free only.
- A request-bearing obligation may wake only the responsible exact endpoint’s background worker.
- All adapters expose the same canonical operation names and strict argument/result schemas.
- Affected IDs: `ARC-001..005`, `ID-006..009`, `AUTH-001..004`, `COM-001..011`, `AVL-002..007`, `UX-001..006`, `SEC-003..006`, `OPS-001..003`.

---

### Task 1: Define exact endpoint binding and capability-root custody

**Files:**
- Create: `src/agentnet/bindings/endpoint.py`
- Modify: `src/agentnet/bindings/composition.py:53-203`
- Modify: `src/agentnet/core/capabilities.py`
- Test: `tests/bindings/test_endpoint_binding.py`
- Test: `tests/bindings/test_local_binding_composition.py`

**Interfaces:**
- Consumes: schema-v7 `endpoint_lifecycle` rows and current `CredentialBinding`.
- Produces:
```python
@dataclass(frozen=True, slots=True)
class EndpointBinding:
    domain_id: str
    principal_id: str
    harness_id: str
    harness_kind: str
    credential_id: str
    credential_epoch: int
    adapter_generation: int
    mailbox_cursor: int
    profile_key: str
    capability_root_path: Path
    process_measurement: str

class EndpointBindingRepository:
    def load_current(self, *, domain_id: str, harness_id: str) -> EndpointBinding: ...
    def rotate_generation(self, *, actor: VerifiedActor, expected_generation: int, process_measurement: str) -> EndpointBinding: ...
```

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

```python
def test_two_pi_profiles_receive_distinct_capability_roots(repository, pi_actor_a, pi_actor_b):
    a = repository.load_current(domain_id=pi_actor_a.domain_id, harness_id=pi_actor_a.harness_id)
    b = repository.load_current(domain_id=pi_actor_b.domain_id, harness_id=pi_actor_b.harness_id)
    assert a.capability_root_path != b.capability_root_path
    assert read_capability_digest(a) != read_capability_digest(b)


def test_binding_rejects_stale_generation(repository, endpoint_actor):
    old = repository.load_current(domain_id=endpoint_actor.domain_id, harness_id=endpoint_actor.harness_id)
    repository.rotate_generation(actor=endpoint_actor, expected_generation=old.adapter_generation, process_measurement="pid:22:start:900")
    with pytest.raises(AuthenticationError, match="generation"):
        authenticate_old_descriptor(old)
```

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

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/bindings/test_endpoint_binding.py
```
Expected: FAIL because endpoint-bound roots and repository do not exist.

- [ ] **Step 3: Implement owner-private endpoint derivation**

```python
def endpoint_root(base: Path, binding: EndpointBinding) -> Path:
    opaque = sha256(
        f"{binding.domain_id}\0{binding.harness_id}\0{binding.adapter_generation}".encode()
    ).hexdigest()
    root = base / "endpoints" / opaque
    require_owner_private_real_directory(root)
    return root
```

Store only the root digest centrally. The capability bytes remain in an owner-only real file or sealed descriptor. Extend `LocalBindingService` to compare exact domain, harness, credential ID/epoch, generation, process measurement, and current revocation before every request.

- [ ] **Step 4: Run binding security tests**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/bindings/test_endpoint_binding.py tests/bindings/test_local_binding_composition.py tests/security/test_ipc_capability.py
```
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add src/agentnet/bindings/endpoint.py src/agentnet/bindings/composition.py src/agentnet/security/capabilities.py tests/bindings/test_endpoint_binding.py tests/bindings/test_local_binding_composition.py
git commit -m "feat(bindings): isolate exact agent endpoints"
```

### Task 2: Implement the multi-endpoint host supervisor

**Files:**
- Create: `src/agentnet/supervisor/host.py`
- Modify: `src/agentnet/supervisor/daemon.py:42-367`
- Modify: `src/agentnet/supervisor/service.py`
- Modify: `src/agentnet/supervisor/integration.py`
- Modify: `src/agentnet/supervisor/queue.py`
- Test: `tests/supervisor/test_host_endpoints.py`
- Test: `tests/supervisor/test_live_delivery_watch.py`

**Interfaces:**
- Consumes: `EndpointBindingRepository`, `EndpointLifecycleService`, `build_launch_spec`, `DeviceSupervisor`, and `LocalQueue`.
- Produces:
```python
class HostEndpointSupervisor:
    def activate(self, binding: EndpointBinding) -> EndpointRuntimeStatus: ...
    def deactivate(self, harness_id: str, *, reason: str) -> EndpointRuntimeStatus: ...
    def reconcile_once(self) -> tuple[EndpointRuntimeStatus, ...]: ...
    def status(self) -> tuple[EndpointRuntimeStatus, ...]: ...
```

- [ ] **Step 1: Write failing sibling and offline tests**

```python
def test_only_exact_target_worker_dequeues(host, endpoint_a, endpoint_b, targeted_event):
    host.activate(endpoint_a)
    host.activate(endpoint_b)
    enqueue_for(targeted_event, recipient=endpoint_a.harness_id)
    host.reconcile_once()
    assert processed_by(endpoint_a, targeted_event.event_id)
    assert not observed_by(endpoint_b, targeted_event.event_id)


def test_offline_target_stays_exclusively_queued(host, endpoint_a, endpoint_b, targeted_event):
    host.activate(endpoint_b)
    enqueue_for(targeted_event, recipient=endpoint_a.harness_id)
    host.reconcile_once()
    assert queue_owner(targeted_event.event_id) == endpoint_a.harness_id
    assert not processed_by(endpoint_b, targeted_event.event_id)
```

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

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/supervisor/test_host_endpoints.py
```
Expected: FAIL because the host supervisor is absent.

- [ ] **Step 3: Implement endpoint-keyed runtimes**

```python
@dataclass(slots=True)
class _EndpointRuntime:
    binding: EndpointBinding
    supervisor: DeviceSupervisor
    queue: LocalQueue
    worker: CleanWorkerLauncher
    process: subprocess.Popen[bytes] | None = None

class HostEndpointSupervisor:
    def __init__(self, repository: EndpointBindingRepository) -> None:
        self._repository = repository
        self._runtimes: dict[str, _EndpointRuntime] = {}
```

Index every queue and runtime by exact harness ID. Before dequeue and acknowledgement, reload the current binding and compare generation/credential/revocation. On mismatch, close that runtime and retain the event for its exact endpoint; never offer it to another runtime.

- [ ] **Step 4: Verify restart, response-loss, and cleanup behavior**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/supervisor/test_host_endpoints.py tests/supervisor/test_live_delivery_watch.py tests/supervisor/test_background_queue.py tests/adapters/test_subprocess_lifecycle.py
```
Expected: PASS with no child process, socket, or capability file left after the fixture closes.

- [ ] **Step 5: Commit**

```bash
git add src/agentnet/supervisor tests/supervisor
git commit -m "feat(supervisor): manage isolated endpoint runtimes"
```

### Task 3: Extend the canonical local tool contract

**Files:**
- Modify: `src/agentnet/bindings/tools.py:23-58,202-498`
- Modify: `src/agentnet/bindings/mcp.py`
- Modify: `src/agentnet/bindings/ipc.py`
- Modify: `src/agentnet/bindings/remote_manager.py:346-680`
- Test: `tests/bindings/test_canonical_tool_parity.py`
- Test: `tests/bindings/test_remote_manager.py`

**Interfaces:**
- Consumes from later feature plans: `recipient.resolve`, `file.send`, `file.status`, and `file.download` methods on `BoundCore`.
- Produces these exact canonical names in addition to all current operations:
```python
CanonicalToolName = Literal[
    # existing names retained
    "agentnet.recipient.resolve",
    "agentnet.file.send",
    "agentnet.file.status",
    "agentnet.file.download",
]
```

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

```python
EXPECTED_V0145_TOOLS = {
    *CANONICAL_TOOL_NAMES,
    "agentnet.recipient.resolve",
    "agentnet.file.send",
    "agentnet.file.status",
    "agentnet.file.download",
}

@pytest.mark.parametrize("binding", ["direct_ipc", "mcp", "remote"])
def test_each_binding_exposes_exact_v0145_surface(binding):
    assert registered_tool_names(binding) == EXPECTED_V0145_TOOLS
```

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

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/bindings/test_canonical_tool_parity.py
```
Expected: FAIL with the four missing operations.

- [ ] **Step 3: Add strict arguments and dispatch only**

```python
class RecipientResolveArguments(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)
    query: str = Field(min_length=1, max_length=256)

class FileSendArguments(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)
    recipients: tuple[str, ...] = Field(min_length=1, max_length=1000)
    source_path: str = Field(min_length=1, max_length=4096)
    media_type: str = Field(min_length=3, max_length=255)
    classification: Classification = Classification.C1_INTERNAL
    idempotency_key: str = Field(min_length=16, max_length=256)
```

The dispatcher calls `actor_provider()` on every invocation. It must not accept an identity path, harness ID override, or credential in arguments. Until the collaboration/file plans implement `BoundCore`, tests use a strict fake with the final signatures.

- [ ] **Step 4: Verify binding parity**

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

- [ ] **Step 5: Commit**

```bash
git add src/agentnet/bindings tests/bindings/test_canonical_tool_parity.py tests/bindings/test_remote_manager.py
git commit -m "feat(bindings): expose v0.1.45 canonical tools"
```

### Task 4: Register the tools in OMP, Pi, Claude, Codex, and Antigravity

**Files:**
- Modify: `src/agentnet/bindings/pi_extension.ts:159-516`
- Modify: `src/agentnet/adapters/specs.py:18-251`
- Modify: `src/agentnet/adapters/catalog.py`
- Create: `src/agentnet/adapters/omp.py`
- Modify: `src/agentnet/adapters/pi.py`
- Modify: `src/agentnet/adapters/claude.py`
- Modify: `src/agentnet/adapters/codex.py`
- Modify: `src/agentnet/adapters/antigravity.py`
- Modify: `src/agentnet/bindings/mcp_bootstrap.py`
- Modify: `src/agentnet/bindings/windows_mcp_bootstrap.py`
- Test: `tests/adapters/test_all_harnesses.py`
- Test: `tests/adapters/test_launch_specs.py`
- Test: `tests/bindings/test_pi_extension_tools.py`

**Interfaces:**
- Consumes: exact canonical tool list and endpoint-specific local binding descriptor.
- Produces: identical semantic operations in each supported harness, with adapter-specific presentation only.

- [ ] **Step 1: Write failing five-environment surface tests**

```python
@pytest.mark.parametrize("harness", ["omp", "pi", "claude", "codex", "antigravity"])
def test_normal_session_registers_exact_tool_surface(harness, installed_package):
    session = installed_package.launch_existing_conversation(harness)
    assert session.agentnet_tools == EXPECTED_V0145_TOOLS
    assert session.foreground_turns_injected == 0
```

For OMP, first inspect its supported package/extension discovery. If it natively consumes the Pi package manifest, use the same TypeScript implementation under a distinct endpoint profile; do not invent a second loader.

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

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/adapters/test_all_harnesses.py tests/adapters/test_launch_specs.py tests/bindings/test_pi_extension_tools.py
```
Expected: FAIL because ordinary sessions do not yet expose the full surface.

- [ ] **Step 3: Extend TypeScript and MCP registrations**

```typescript
type CanonicalMethod =
  | ExistingCanonicalMethod
  | "agentnet.recipient.resolve"
  | "agentnet.file.send"
  | "agentnet.file.status"
  | "agentnet.file.download";
```

Every tool invocation travels over the endpoint’s sealed descriptor/socket. No adapter reads a global `~/.agentnet/identity.json` fallback. If the harness cannot refresh tools in place, return `restart_required` through the lifecycle service and wait for user action.

- [ ] **Step 4: Run adapter contract and installed-session smoke tests**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/adapters/test_all_harnesses.py tests/adapters/test_launch_specs.py tests/bindings/test_pi_extension_tools.py tests/adapters/test_installed_live_inference.py
```
Expected: contract tests PASS; installed live inference is evidence only when exact installed credentials are supplied and otherwise remains an explicit skip.

- [ ] **Step 5: Commit**

```bash
git add src/agentnet/adapters src/agentnet/bindings tests/adapters tests/bindings/test_pi_extension_tools.py
git commit -m "feat(adapters): register AgentNet in ordinary sessions"
```

### Task 5: Smoke-test exact-agent routing with sibling endpoints

**Files:**
- Create: `scripts/ci/exact_endpoint_routing_e2e.py`
- Create: `tests/integration/test_exact_endpoint_routing.py`
- Modify: `package.json`
- Modify: `npm/scripts/check-packed-package.mjs`

**Interfaces:**
- Consumes: host supervisor and all adapter registrations.
- Produces: one deterministic multi-endpoint smoke result with exact event ID, target harness ID, processing harness ID, sibling observation count, and offline queue owner.

- [ ] **Step 1: Implement the integration scenario**

```python
def scenario(core, host, endpoints):
    target, sibling_pi, sibling_omp, sibling_claude = endpoints
    event = send_exact(core, recipient=target.harness_id, payload={"question": "reply with 7"})
    host.reconcile_once()
    assert processed_harnesses(event.event_id) == [target.harness_id]

    host.deactivate(target.harness_id, reason="offline-test")
    queued = send_exact(core, recipient=target.harness_id, payload={"question": "reply with 8"})
    host.reconcile_once()
    assert mailbox_owner(queued.event_id) == target.harness_id
    assert processed_harnesses(queued.event_id) == []
```

- [ ] **Step 2: Run the smoke scenario from source**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/integration/test_exact_endpoint_routing.py
UV_CACHE_DIR=/tmp/uv-cache uv run python scripts/ci/exact_endpoint_routing_e2e.py
```
Expected: PASS and JSON reporting `sibling_reactions: 0`.

- [ ] **Step 3: Run it from an isolated packed npm install**

```bash
npm run check:packed
```
Expected: PASS; the package tree remains unchanged and the temporary install leaves no endpoint process or capability root.

- [ ] **Step 4: Commit**

```bash
git add scripts/ci/exact_endpoint_routing_e2e.py tests/integration/test_exact_endpoint_routing.py package.json npm/scripts/check-packed-package.mjs
git commit -m "test(adapters): prove exact endpoint routing"
```