# AgentNet v0.1.45 Artifact Delivery 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:** Let authorized exact endpoints send and download files through AgentNet without exposing bytes before quarantine, integrity, scanner, policy, audit, and current-authorization gates succeed.

**Architecture:** Preserve the existing `ArtifactService` lifecycle and add a backend-neutral transfer coordinator above it. Use a maintained ClamAV daemon adapter to create signed scanner attestations; the deterministic local prefilter remains deny/indeterminate only and never authorizes release. Send-file orchestration reserves quota, streams bytes to quarantine, records provenance, obtains a fresh scan, releases immutably, then sends only a released artifact binding to exact recipients.

**Tech Stack:** Python, `StoreBackend`, encrypted filesystem object store, ClamAV daemon Unix socket/TCP loopback adapter, P-256 scanner attestations, existing mailbox/conversation services, pytest.

## Global Constraints

- Files are not ordinary message bytes. Event payloads contain only typed released-artifact bindings.
- Authorization and byte/quota reservation precede upload. Exact digest and size are verified before manifest promotion.
- Quarantine bytes are immutable and encrypted. Object existence, preview, plaintext, key material, and deduplication facts are not disclosed before release authorization.
- The local prefilter may deny but can never authorize. Release requires a fresh trusted attestation from the configured maintained scanner profile.
- Every download rechecks current actor, collaboration scope, artifact state, policy revision, scanner freshness, retention/legal hold, and exact audience harness.
- Download capabilities are bounded and single-use. Local writes use an owner-approved destination, no-follow creation, temporary sibling file, fsync, digest verification, and atomic rename.
- Symlink roots, path traversal, device files, FIFOs, sockets, directories, oversized files, changing source files, stale scanners, malicious files, secrets, archives, and executables fail closed according to policy.
- No `file sent` result is returned until immutable release and exact-recipient event custody exist.
- Affected IDs: `COM-002`, `COM-004..011`, `FILE-001..006`, `AVL-001..007`, `SEC-001..007`, `OPS-001..006`.

---

### Task 1: Make artifact lifecycle backend-neutral and schema-v7 exact

**Files:**
- Modify: `src/agentnet/artifacts/service.py:269-2260`
- Modify: `src/agentnet/storage/artifact_transfer_schema.py` (created by the lifecycle plan’s schema-v7 task)
- Modify: `src/agentnet/storage/postgres_catalog.py`
- Test: `tests/artifacts/test_backend_parity.py`
- Test: `tests/artifacts/test_staged_artifact.py`

**Interfaces:**
- Consumes: `StoreBackend`, existing artifact lifecycle/quota tables, schema-v7 `artifact_transfers` and `artifact_transfer_recipients`.
- Produces: `ArtifactService(store: StoreBackend, ...)` and identical SQLite/PostgreSQL behavior.

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

```python
@pytest.mark.parametrize("backend", ["sqlite", "postgresql"])
def test_reserve_upload_scan_release_download_is_backend_neutral(backend, artifact_fixture):
    service = artifact_fixture.service(backend)
    released = artifact_fixture.release_text(service, b"hello")
    assert released["state"] == "released"
    assert service.consume_download(released["token"], actor=artifact_fixture.recipient) == b"hello"


def test_unknown_schema_or_catalog_fails_before_artifact_access(tampered_store):
    with pytest.raises(GateBlocked, match="schema"):
        open_artifact_service(tampered_store)
```

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

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/artifacts/test_backend_parity.py
```
Expected: SQLite remains the regression baseline; the PostgreSQL parameter fails because `ArtifactService` is typed/composed around `SQLiteStore` and catalog coverage is incomplete.

- [ ] **Step 3: Replace concrete-store assumptions**

```python
class ArtifactService:
    def __init__(
        self,
        store: StoreBackend,
        objects: FilesystemArtifactStore,
        *,
        enabled: bool,
        trusted_scanner_keys: Mapping[tuple[str, int], bytes],
        scanner_policy: ScannerTrustPolicy,
        local_prefilter: ArtifactScanner,
        operations_policy: OperationsPolicy,
        outage_gate: OutageGate,
        provenance: ProvenanceService,
        clock: Callable[[], int] | None = None,
    ) -> None: ...
```

Use qmark SQL through the existing PostgreSQL adapter. Normalize integrity errors through the repository’s established `sqlite3.IntegrityError` compatibility seam. Extend exact v7 catalog checks without loosening v1–v6 checks.

- [ ] **Step 4: Run artifact and conditional PostgreSQL suites**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/artifacts/test_backend_parity.py tests/artifacts/test_staged_artifact.py tests/storage/test_postgres_migrations.py
```
Expected: SQLite PASS; PostgreSQL PASS when the dedicated mutation-authorized DSN is configured, otherwise the PostgreSQL parameter emits its named prerequisite skip.

- [ ] **Step 5: Commit**

```bash
git add src/agentnet/artifacts/service.py src/agentnet/storage/postgres_catalog.py tests/artifacts
git commit -m "refactor(artifacts): support verified storage backends"
```

### Task 2: Integrate a maintained scanner provider

**Files:**
- Create: `src/agentnet/artifacts/clamav.py`
- Create: `src/agentnet/supervisor/scanner_worker.py`
- Modify: `src/agentnet/artifacts/scanner.py`
- Modify: `src/agentnet/operations/server_setup.py`
- Modify: `deploy/compose.production.json`
- Modify: `deploy/.env.production.example`
- Modify: `docs/BUILD_VS_REUSE.md`
- Test: `tests/artifacts/test_clamav_scanner.py`
- Test: `tests/supervisor/test_scanner_worker.py`

**Interfaces:**
- Consumes: ClamAV daemon `INSTREAM` protocol over a configured Unix socket or loopback endpoint, product-owned scanner P-256 key, current engine/signature version, and `ScannerTrustPolicy`.
- Produces:
```python
class ClamAVScanner:
    def scan(self, *, artifact_id: str, plaintext_digest: str, content: bytes, issued_at: int, expires_at: int) -> ArtifactScanAttestationV1: ...

class ScannerWorker:
    def process_once(self, *, limit: int = 25) -> tuple[str, ...]: ...
```

- [ ] **Step 1: Write failing protocol and fail-closed tests**

```python
def test_clean_clamav_result_produces_signed_attestation(fake_clamd, scanner):
    fake_clamd.reply(b"stream: OK\x00")
    attestation = scanner.scan(
        artifact_id="artifact-1",
        plaintext_digest=sha256(b"hello").hexdigest(),
        content=b"hello",
        issued_at=100,
        expires_at=160,
    )
    assert attestation.result == "clean"
    assert attestation.scanner_profile.startswith("clamav:")


@pytest.mark.parametrize("failure", ["timeout", "malformed", "unknown", "stale-db", "oversize"])
def test_scanner_failure_never_releases(failure, scanner_fixture):
    scanner_fixture.fail_with(failure)
    assert scanner_fixture.process_once() == ()
    assert scanner_fixture.artifact_state == "quarantined"
```

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

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/artifacts/test_clamav_scanner.py tests/supervisor/test_scanner_worker.py
```
Expected: FAIL because the provider and worker are absent.

- [ ] **Step 3: Implement bounded ClamAV protocol and signed evidence**

```python
class ClamAVScanner:
    def __init__(self, endpoint: ScannerEndpoint, key: P256KeyPair, *, timeout_seconds: float = 30.0, max_bytes: int = MAX_ARTIFACT_BYTES) -> None:
        self.endpoint = endpoint
        self.key = key
        self.timeout_seconds = timeout_seconds
        self.max_bytes = max_bytes
```

Use `zINSTREAM\0`, 4-byte network-order chunk lengths, bounded 64 KiB chunks, zero terminator, bounded response, and exact `OK`/`FOUND` parsing. Treat all other responses as indeterminate. Bind scanner ID, key epoch, engine version, signature version/time, artifact ID, plaintext digest, policy revision, issue/expiry, and result in the signed fields.

- [ ] **Step 4: Add setup/readiness gates**

Server setup resolves the scanner endpoint and package-owned worker configuration itself. Startup fails file capability closed unless ClamAV readiness, signature freshness, scanner key custody, and Core trust configuration match. Messaging without files remains available.

Run:
```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/artifacts/test_clamav_scanner.py tests/supervisor/test_scanner_worker.py tests/operations/test_server_setup.py -k scanner
```
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add src/agentnet/artifacts src/agentnet/supervisor/scanner_worker.py src/agentnet/operations/server_setup.py deploy docs/BUILD_VS_REUSE.md tests/artifacts tests/supervisor/test_scanner_worker.py
git commit -m "feat(artifacts): add maintained scanner worker"
```

### Task 3: Implement atomic send-file orchestration

**Files:**
- Create: `src/agentnet/artifacts/transfer.py`
- Modify: `src/agentnet/core/app.py`
- Test: `tests/artifacts/test_transfer_service.py`
- Test: `tests/integration/test_file_send.py`

**Interfaces:**
- Consumes: `ArtifactService`, `ScannerWorker`, `CollaborationScopeService`, `ConversationService`, exact recipients, and `StoreBackend`.
- Produces:
```python
class ArtifactTransferService:
    def send_file(
        self,
        *,
        actor: VerifiedActor,
        recipients: tuple[str, ...],
        source: Path,
        media_type: str,
        classification: Classification,
        idempotency_key: str,
    ) -> dict[str, Any]: ...
    def status(self, *, actor: VerifiedActor, transfer_id: str) -> dict[str, Any]: ...
```

- [ ] **Step 1: Write failing send and crash-recovery tests**

```python
def test_file_sent_only_after_release_and_recipient_custody(transfer, sender, recipient, source_file):
    result = transfer.send_file(
        actor=sender,
        recipients=(recipient.harness_id,),
        source=source_file,
        media_type="text/plain",
        classification=Classification.C1_INTERNAL,
        idempotency_key="file-send-0000001",
    )
    assert result["state"] == "recipient_custody_recorded"
    assert released_binding(result["artifact_id"]).plaintext_digest == sha256(source_file.read_bytes()).hexdigest()


@pytest.mark.parametrize("phase", ["after_reserve", "after_upload", "after_manifest", "after_scan", "after_release", "after_event"])
def test_send_reconciles_without_duplicate_event(transfer_fixture, phase):
    transfer_fixture.crash_after(phase)
    first = transfer_fixture.attempt()
    second = transfer_fixture.retry_same_key()
    assert second["event_id"] == first.recorded_event_id
    assert transfer_fixture.event_count == 1
```

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

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/artifacts/test_transfer_service.py tests/integration/test_file_send.py
```
Expected: FAIL because `ArtifactTransferService` is absent.

- [ ] **Step 3: Implement source-safe staged orchestration**

```python
class TransferState(str, Enum):
    RESERVED = "reserved"
    QUARANTINED = "quarantined"
    MANIFESTED = "manifested"
    SCANNED = "scanned"
    RELEASED = "released"
    EVENT_COMMITTED = "event_committed"
    RECIPIENT_CUSTODY_RECORDED = "recipient_custody_recorded"
    REJECTED = "rejected"
```

Open source with no-follow flags; require a regular owner-readable file; record stat before/after bounded streaming; reject mutation. Persist each transition with request digest, expected prior state/revision, artifact ID, event ID, and audit hash. Retry resumes from the durable state and never repeats an already committed event.

- [ ] **Step 4: Verify transfer behavior**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/artifacts/test_transfer_service.py tests/integration/test_file_send.py tests/artifacts/test_staged_artifact.py
```
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add src/agentnet/artifacts/transfer.py src/agentnet/core/app.py tests/artifacts/test_transfer_service.py tests/integration/test_file_send.py
git commit -m "feat(artifacts): send released files atomically"
```

### Task 4: Implement safe bounded downloads

**Files:**
- Modify: `src/agentnet/artifacts/transfer.py`
- Create: `src/agentnet/artifacts/local_destination.py`
- Test: `tests/artifacts/test_safe_download.py`

**Interfaces:**
- Consumes: released artifact binding, current collaboration scope, `ArtifactService.issue_download_capability`, and exact audience endpoint.
- Produces:
```python
class SafeDownloadDestination:
    def write(self, *, destination: Path, content: bytes, expected_digest: str) -> Path: ...

class ArtifactTransferService:
    def download_file(self, *, actor: VerifiedActor, artifact_id: str, destination: Path, idempotency_key: str) -> dict[str, Any]: ...
```

- [ ] **Step 1: Write failing path and capability tests**

```python
@pytest.mark.parametrize("kind", ["symlink-root", "symlink-target", "directory", "fifo", "device", "existing-file"])
def test_unsafe_destination_is_rejected(kind, download_fixture):
    destination = download_fixture.destination(kind)
    with pytest.raises(ValidationError):
        download_fixture.download(destination)


def test_download_capability_is_exact_agent_single_use(download_fixture):
    token = download_fixture.issue()
    with pytest.raises(AuthorizationError):
        download_fixture.consume(token, actor=download_fixture.sibling)
    assert download_fixture.consume(token, actor=download_fixture.recipient) == b"hello"
    with pytest.raises(ConflictError):
        download_fixture.consume(token, actor=download_fixture.recipient)
```

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

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/artifacts/test_safe_download.py
```
Expected: FAIL because safe destination orchestration is absent.

- [ ] **Step 3: Implement no-follow atomic destination writes**

```python
def write(self, *, destination: Path, content: bytes, expected_digest: str) -> Path:
    parent = require_owner_private_real_directory(destination.parent)
    temporary = parent / f".{destination.name}.{secrets.token_hex(8)}.agentnet"
    fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
    # write, fsync, close, verify digest, then os.replace only if destination is absent
```

Recheck destination immediately before rename. On any error, remove only the exact temporary inode created by this call. Never overwrite an existing destination.

- [ ] **Step 4: Verify download behavior**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/artifacts/test_safe_download.py tests/artifacts/test_staged_artifact.py
```
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add src/agentnet/artifacts/transfer.py src/agentnet/artifacts/local_destination.py tests/artifacts/test_safe_download.py
git commit -m "feat(artifacts): download files safely"
```

### Task 5: Expose file tools and prove end-to-end transfer

**Files:**
- Modify: `src/agentnet/bindings/tools.py`
- Modify: `src/agentnet/bindings/pi_extension.ts`
- Modify: `src/agentnet/bindings/remote_manager.py`
- Create: `tests/integration/test_file_tools_e2e.py`
- Modify: `npm/scripts/check-packed-package.mjs`

**Interfaces:**
- Consumes: `ArtifactTransferService`.
- Produces canonical `agentnet.file.send`, `agentnet.file.status`, and `agentnet.file.download` in every adapter.

- [ ] **Step 1: Write failing canonical tool scenario**

```python
def test_send_then_download_file_from_exact_recipient(dispatchers, tmp_path):
    source = tmp_path / "source.txt"
    source.write_text("hello", encoding="utf-8")
    sent = dispatchers.sender.call("agentnet.file.send", {
        "recipient_query": "the enrolled server",
        "source_path": str(source),
        "media_type": "text/plain",
        "classification": "C1",
        "idempotency_key": "file-tool-send-001",
    })
    destination = tmp_path / "received.txt"
    downloaded = dispatchers.recipient.call("agentnet.file.download", {
        "artifact_id": sent["artifact_id"],
        "destination_path": str(destination),
        "idempotency_key": "file-tool-download-001",
    })
    assert downloaded["digest"] == sha256(b"hello").hexdigest()
    assert destination.read_bytes() == b"hello"
```

- [ ] **Step 2: Verify the scenario fails**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/integration/test_file_tools_e2e.py
```
Expected: FAIL because file tools are not yet connected to transfer service.

- [ ] **Step 3: Connect strict dispatch and status output**

Return only transfer/artifact/event IDs, exact custody/release states, digest, size, media type, and destination path. Do not return quarantine paths, object keys, scanner keys, download tokens, or other recipients’ state.

- [ ] **Step 4: Run source and packed E2E**

```bash
UV_CACHE_DIR=/tmp/uv-cache uv run pytest -q tests/integration/test_file_tools_e2e.py
npm run check:packed
```
Expected: PASS with byte-identical source/destination and no residual download token or temporary file.

- [ ] **Step 5: Commit**

```bash
git add src/agentnet/bindings tests/integration/test_file_tools_e2e.py npm/scripts/check-packed-package.mjs
git commit -m "feat(tools): expose safe file transfer"
```