# ADR 0002 — Implementation Tasks

Spec: `docs/adr/0002-notify-pull-receive-path.md` (§8b transport boundary, §8c clarifications #1-13, §8d lifecycle model, §8e protocol completion; 16 acceptance criteria).
Index: `docs/repo-index.md` (ast-grep symbol map).

Rules:
- Each task = one PR-sized unit, individually takable, lands on `feat/adr-0002-notify-pull`.
- DoD = code + tests + docs touched + acceptance criteria it satisfies.
- Tasks with same dependency tier may be taken in any order; never skip a dependency.
- "Legacy path" = non-Pi runtimes (`sidecar.mjs`, `echo-peer.mjs`) — out of scope, keep working (ADR 8c #3).

---

## T1 — Transport contract reshape

**Dep:** none. **Tier:** 1

- `src/transports/contract.mjs`: required surface = `initMailbox, sendMessage, listInbox, readMessage, watchInbox, replyTo, formatMessage, hasAmq`. `drainInbox`/`waitForMessages` drop off required list but stay exported on the transport object (8c #12 additive-safe; sidecar/echo-peer keep calling them).
- `src/transports/amq-client.mjs` + `.d.mts`: `sendMessage`/`replyTo` gain `priority`; `listInbox` gains `box ('new'|'cur'), from, kind, priority, limit, offset`; `readMessage` promoted required; `drainInbox`/`waitForMessages` retained as compat exports.
- `tests/transport-contract.test.mjs`: assert new required set, old methods still present.
- **ADR:** §8b contract, 8c #12. **Criteria:** 2, 3, 9 (CLI halves).
- **DoD:** `npm test` green; sidecar/echo-peer still import cleanly.

## T2 — watchInbox adapter (continuous stream over one-shot watch)

**Dep:** T1. **Tier:** 1

- `amq-client.mjs`: `watchInbox({root, me, signal?, poll?}): AsyncIterable<WatchEvent>`.
- Internal: spawn `amq watch --json --timeout 0`; parse `new_message`/`existing` → typed `WatchEvent` (messages = re-listed envelopes, 8c #7); respawn after every event (exit 0 = normal); exit 4 (timeout no-op) → respawn no backoff; exit 3 (missing mailbox) → surface error once, stop; signal → SIGTERM child; debounce 1s when re-list yields no new ids, extend after N cycles (default 5 → 5s).
- `--poll` passthrough for network FS.
- **Tests (new, fake watch-child script):** one event then exit 0 → stream yields + respawns; backlog `existing`; exit 4 no backoff; exit 3 stops + surfaces; abort kills child (no zombie); debounce extends on no-new-id cycles.
- **ADR:** §1, §5, §8b, 8c #7. **Criteria:** 12.
- **DoD:** zombie-free (pid observation test), no spin with empty inbox.

## T3 — Shared per-(root,me) state + claim/release protocol

**Dep:** none. **Tier:** 1

- New `src/state/` module: `~/.amq-bridge/state/<sha256(root)>-<me>.json`, schema v3 `{version, owner{pid,startedAt,sessionKey}|null, processedIds (cap 500), activeId|null, migrated0002}`. `mkdir -p` on first claim; tmp+rename atomic write (sibling tmp, same FS).
- Claim/release exactly per §8e: O_EXCL lock `<...>.owner.lock` carrying identity; EEXIST → dead-pid unlink+retry, live-pid wait ≤2s → secondary; write-then-unlink order; conditional owner clear (`owner.sessionKey == self`); lock scope = ownership transitions + migration flag only; appends lock-free via single-writer invariant.
- `sha256(root)` hex filenames (never lossy char-map — M4).
- **Tests:** concurrent claim → exactly one owner; stale lock (dead pid) → reclaim; cross-session takeover resets activeId; same-sessionKey resume keeps activeId; processedIds cap; corrupt file → backup + recreate; kill -9 between lock and state write → next claimer recovers.
- **ADR:** §8d, §8e. **Criteria:** 6, 7, 13, 14, 15.
- **DoD:** protocol tests green incl. race + crash windows.

## T4 — Upgrade bootstrap

**Dep:** T3. **Tier:** 1

- On first claim: union legacy per-session `injected.json` ids for `(root,me)` into shared `processedIds`; legacy `.amq-bridge/migrated-0002` marker present → set `migrated0002`, skip note.
- After migration: legacy `inbox.json`/`injected.json` removed (or ignored; files die with T6 buffer removal).
- **Tests:** old-version file layout → first attach: no re-inject of pre-upgrade injected ids; migration note exactly once.
- **ADR:** §8e #2. **Criteria:** 10, 16.
- **DoD:** upgrade simulation test green.

## T5 — Injection policy + scheduler (startInboxLoop rewrite)

**Dep:** T2, T3. **Tier:** 2

- `pi-extension.ts` `startInboxLoop`/`scheduleInboxLoop` rewrite: consume `watchInbox` stream; inject **envelopes only** (no bodies; strip `details.messages` bodies); `triggerTurn` only when batch has actionable (kind in {question, decision, review_request, review_response, answer, todo}) not in `processedIds`; one-active (oldest → `activeId`, rest queued context); dequeue on reply/resolve (re-list → next active); **persist processedIds before triggerTurn** (8c #4, 8e #5).
- Urgent-wake trust gate (8c #2 vocabulary): trusted = source `attach|manual` or live presence; untrusted = `handshake|message|discover|no-source`. Urgent wakes only from trusted.
- Delete: `drainIntoBuffer`, `readInboxBuffer`/`writeInboxBuffer`/`appendDedupeMessages`, `readInjectedIds`/`writeInjectedIds`/`unseenActionableMessages`/`markInjected`, `ambiguousReplyText`, `waitForMessages` receive path.
- **Tests:** unit for actionability/trust classification; e2e envelope-only injection, no-turn for status-only, urgent trusted wake, handshake no-wake, dequeue ≤2s.
- **ADR:** §2-4, §7, 8c #2/#4. **Criteria:** 1, 4, 5, 8.
- **DoD:** no body ever injected; one turn per batch.

## T6 — Tools + commands surface

**Dep:** T1, T3. **Tier:** 2

- New tools: `amq_bridge_read <id>` (readMessage, marks read, re-readable), `amq_bridge_resolve <id>` (readMessage + processed record + dequeue; no watch event).
- `amq_bridge_inbox`: `list --new` envelopes; `--all` → `--cur`; newest-first display (8c #5); optional `--limit`; optional lazy `--preview`.
- `amq_bridge_reply`: `messageId` always required; missing → pending envelope list error (never bare CLI error, never latest-reply).
- `amq_bridge_send`: `priority` param.
- `amq_bridge_status`: + pending count, activeId.
- Commands: `/amq-bridge read <id>`, `resolve <id>`; `inbox` envelope view; `send` default kind `status`→`question` + `--priority` (8 §8); `reply` id-required; completions updated.
- **ADR:** §3, §8, 8c #5. **Criteria:** 3, 9.
- **DoD:** tool smoke updated to envelopes; reply-without-id errors with list.

## T7 — Lifecycle wiring (attach/connect/detach/restore/shutdown)

**Dep:** T3, T5. **Tier:** 2

- Shared `startWatch()` helper used by `attach` AND `connect` (8c #10): owner claim (only when watcher will run — TUI+attached, 8d #3) → migration check (T4) → watchInbox stream → backlog inject.
- Watcher key `${root}:${me}`; one watcher per session; attach/connect with new self aborts old `(root,*)` watchers of this session (8d #4).
- `session_start` restore: claim through `startWatch()`; non-TUI = no claim, no watch.
- `detach` + `session_shutdown`: release owner (conditional clear), keep processedIds, reset activeId; attachments cleanup owner-only; session_shutdown aborts by `(root,me)` key + releases marker (8c #13, 8d #2).
- Secondary sessions read-only: send/reply/resolve blocked with explicit error; inbox/read/status/attach/detach allowed (8d #5).
- Exit-3 from watch: notify once, stop, release claim (8d #8).
- Migration marker/state at `~/.amq-bridge/state/` (survives detach `rmSync`).
- **Tests:** detach→re-attach single inject; owner kill-9 → takeover by next claim; concurrent attach; non-TUI no-claim; secondary blocked; restore-while-attached migration.
- **ADR:** §8c #1/#9/#10/#13, §8d, §8e. **Criteria:** 6, 7, 13, 14, 15.
- **DoD:** lifecycle e2e green (concurrent claim, kill-9 takeover, upgrade restore).

## T8 — Migration cleanup + doc alignment

**Dep:** T5. **Tier:** 3

- Delete remaining `inbox.json`/`injected.json` path helpers (inboxPath/injectedPath) once T5/T6 land.
- ADR text cleanup: stale §2/§7/§8-main/What-dies/Open-questions wording (later-section-wins → single coherent doc).
- `integrations/pi-skill/SKILL.md`: "auto-polls inbox" → watch-driven; document read/resolve/priority (8c #9).
- `docs/message-kinds.md`: align kind/priority claims with implementation (8c #8).
- `formatMessage`: envelope-only rendering (8c #8).
- **ADR:** 8c #5/#8/#9. **Criteria:** 10, 11.
- **DoD:** grep confirms no `inbox.json`/`injected.json`/`waitForMessages` in receive path; docs coherent.

## T9 — Smoke + e2e migration

**Dep:** T6, T7. **Tier:** 3

- 4 smoke harnesses (`pi-extension-smoke`, `-tool-smoke`, `-pty-smoke`, `-multipeer-smoke`): stop reading `.amq-bridge/sessions/<n>/inbox.json`; assert envelopes via tool output + `amq_bridge_read` bodies.
- e2e scripts (`run-cross-cwd-e2e.sh`, `run-real-pi-e2e.sh`): PING/ACK assertions via read tool (subject/from envelopes).
- New lifecycle e2e: concurrent claim, owner kill-9, upgrade boundary (criteria 14-16).
- **ADR:** criteria 1-16. **Criteria:** all.
- **DoD:** all smokes + e2e green on CI-equivalent run.

## T10 — Validation pass

**Dep:** T8, T9. **Tier:** 4

- Full `npm test` + all smokes + both e2e.
- Independent reviewer pass on the concrete diff vs ADR §8b-§8e (transport boundary, protocol, lifecycle).
- Dogfood: two real Pi sessions, attach a↔b, bombard, detach/re-attach, kill -9.
- Close any residual: watch-child orphan, resume/fork first-bite (accepted, record outcome).
- **ADR:** §8e #5 residuals. **Criteria:** 1-16.
- **DoD:** reviewer sign-off; dogfood notes appended to decision-log.

---

## Dependency graph

```
T1 ──► T2 ──► T5 ──► T8 ──► T10
                ▲      ▲
T3 ──► T4 ──────┘      │
  └────► T6 ──► T7 ────┘
```

Parallel lanes: (T1,T2,T3,T4) → (T5,T6,T7) → (T8,T9) → T10.

## Quick reference — what dies vs what stays

| Dies | Stays |
|---|---|
| `waitForMessages` poll loop | `sendMessage`, `replyTo`, `listInbox` (expanded) |
| `drainInbox` ingestion | `drainInbox`/`waitForMessages` as compat exports (non-Pi) |
| `inbox.json` / `injected.json` | `amq-bridge-state` roster (ADR 0001) |
| reply-latest + ambiguous path | explicit-id reply + resolve |
| bodies in injection | envelope-only injection |
| cwd-scoped watcher key/markers | `(root:me)` watcher + root-hashed shared state |