fix(daemon,cli): PL-004 Phase A R1 — topology-backed validation + default nudge + handoff-and-complete + whoami + SSE alias + CLI tests

Closes adjudication revise-required from delivery-orch-lead@openrig-velocity
per qitem-20260503021500-49b0db82 + handoff packet
substrate/rigs/openrig-velocity/state/delivery-orch/pl004-phase-a-revision-driver-handoff-20260503.md.
Foundation: openrig d8d8ee9 (Phase A R0). Branch: phase-a-coordination-primitive-system (worktree-scoped; main NOT touched).

PRIOR REVIEW INPUTS (all four corroborate the R1 deliverable list):
- Codex independent review (revise-required): qitem-20260503020120-ef9b04db / artifact pl004-phase-a-independent-codex-review-20260503.md
- Claude independent review (clean-with-followups): qitem-20260503020117-0efbff1d / artifact pl004-phase-a-independent-claude-review-20260503.md
- Empirical QA (proof-corroborated-with-caveats): pl004-phase-a-empirical-qa-20260503.md
- Model-diverse blocker packet (blocked-pending-adjudication): coordination-primitive-system-phase-a-d8d8ee9-model-diverse-review-blocker-20260503.md
- Adjudication: orch-lead@openrig-velocity-claude → revise-required.

R1 SCOPE DELIVERED (4 ratified Phase A gaps):

1. Topology-backed validateRig wired in startup.ts (closes Codex finding 1; closes Claude FU2-rebound):
   - packages/daemon/src/startup.ts:131 now constructs QueueRepository with `validateRig: topologyValidateRig` that splits `<member>@<rig>` and rejects rigs absent from rigRepo.findRigsByName(). Bare ids (no @) also rejected (no canonical rig binding).
   - packages/daemon/src/domain/queue-repository.ts: `create()` and `handoff()` already called the hook; new `handoffAndComplete()` also enforces it.
   - Route + repo + CLI rejection coverage: queue-routes.test.ts adds 4 R1 tests (unknown-rig 400 on create / on handoff / on handoff-and-complete + accept-known-rig); queue-repository.test.ts already had the hook test.

2. Default nudge wiring + persistence (closes Codex finding 2):
   - QueueRepository now takes optional `transport: QueueNudgeTransport` (minimal interface; concrete impl is SessionTransport). After successful create / handoff / handoff-and-complete commits, fires `transport.send(destination, "Queue handoff: <qitemId> - check your queue.", { verify: true })` and persists last_nudge_attempt + last_nudge_result via existing recordNudgeAttempt(). recordNudgeAttempt() is no longer dead code.
   - Opt-out: `nudge: false` in the input skips the call entirely. CLI exposes `--no-nudge` on create / handoff / handoff-and-complete. Default behavior is nudging.
   - InboxHandler.absorb() passes `nudge: false` when creating the absorbed qitem (the inbox surface already had its own delivery via the original drop; second nudge on absorb would be duplicative).
   - Nudge failure is recorded as `failed:<reason>` in last_nudge_result; the underlying queue mutation is NOT unwound (operator can retry the wake separately).
   - Post-construction transport attachment: new `attachTransport()` setter lets startup.ts wire the transport AFTER both QueueRepository and SessionTransport are constructed (SessionTransport depends on agentActivityStore which depends on eventBus, so it's later in the dep graph).
   - 5 R1 tests in queue-repository.test.ts: nudge default + opt-out + failure-recording + handoff nudges + attachTransport post-hoc wiring.

3. Missing route/CLI surfaces (closes Codex finding 3):
   - POST /api/queue/:qitemId/handoff-and-complete (new route) — variant of handoff() that closes source as state=done (terminal) instead of state=handed-off (intermediate). Same atomic close+create + chain_of_record + default-nudge contract. Mirrored on the CLI as `rig queue handoff-and-complete <qitemId> --from --to [--body --note --priority --tier --tags --no-nudge --json]`.
   - GET /api/queue/whoami (new route; mounted before /:qitemId so literal path wins) — returns {session, asDestination:{pending,inProgress,blocked,recent[]}, asSource:{total}} for the queried session. Read-only. CLI: `rig queue whoami --session <s> [--recent-limit N --json]`.
   - GET /api/stream/sse and GET /api/queue/sse — Phase A contract route names per IMPL § Routes (which named /api/stream/sse). Mounted as a SECOND alias alongside the existing /watch handler in both stream.ts and queue.ts (same handler function dual-mounted; either path emits the identical event stream). Documented as "compat alias" in code comments. No behavior change for existing /watch consumers.
   - 3 R1 tests in queue-routes.test.ts (whoami GET success + missing-session 400 + handoff-and-complete success + handoff-and-complete missing-fromSession). 1 R1 test in stream-routes.test.ts (sse mounted alongside watch). 2 R1 tests in queue-repository.test.ts for whoami + handoff-and-complete repository methods.

4. CLI tests for queue + stream (closes Codex finding 3 + Claude FU8):
   - NEW packages/cli/test/queue.test.ts (9 cases): registration check + create POST shape + --no-nudge passthrough + update done-without-closure_reason error rendering with validReasons + handoff-and-complete POST shape + whoami query params + unknown-rig 400 surfacing + handoff --no-nudge + list filter params.
   - NEW packages/cli/test/stream.test.ts (8 cases): registration check + emit POST shape + --id idempotent passthrough + list filter + show + archive + non-2xx exit on emit + 404 exit on show.
   - Pattern mirrors compact-plan.test.ts: vi.mock daemon-lifecycle, inject clientFactory with stubbed HTTP client. No real daemon. No DB. No network.

ASYNC CASCADE (mechanical):
   - QueueRepository.create() / handoff() / new handoffAndComplete() are now async (they `await` the post-commit nudge call). Routes + InboxHandler.absorb() updated to await. Test files updated:
     * queue-repository.test.ts: every it() callback async; every repo.create()/repo.handoff() awaited; .toThrow() patterns refactored to .rejects.toThrow() where appropriate.
     * inbox-handler.test.ts: same async-cascade for inbox.absorb() callers.
     * queue-routes.test.ts: route handler at /inbox/:inboxId/absorb now awaits InboxHandler.absorb().
   - InboxHandler.absorb() is now async; the queueRepo.create() call inside still happens BEFORE the inbox UPDATE transaction (Claude FU1 atomicity concern from prior review remains as a separate followup; in-scope fix would be to wrap both in a single SAVEPOINT-nested transaction).

RE-VERIFICATION COMMANDS (all green; node v22.22.1):

Focused (under explicit temp-rooted OPENRIG_HOME + OPENRIG_DB):
- daemon: `OPENRIG_HOME=/tmp/pl004-r1-final-d OPENRIG_DB=/tmp/pl004-r1-final-d/openrig.sqlite npx vitest run test/queue-repository.test.ts test/queue-routes.test.ts test/coordination-integration.test.ts test/inbox-handler.test.ts test/hot-potato-enforcer.test.ts test/stream-routes.test.ts test/stream-store.test.ts test/outbox-handler.test.ts` → 89/89 PASS (68 baseline + 21 R1).
- cli: `OPENRIG_HOME=/tmp/pl004-r1-final-c OPENRIG_DB=/tmp/pl004-r1-final-c/openrig.sqlite npx vitest run test/queue.test.ts test/stream.test.ts` → 17/17 PASS (NEW; no prior CLI tests for queue/stream).
- daemon tsc + build: exit 0; cli tsc + build: exit 0.

Compiled-output runtime checks:
- node import dist/domain/queue-repository.js → QueueRepository: function; attachTransport: function; handoffAndComplete: function; whoami: function.
- node import dist/index.js → queue subs: claim,create,fallback,handoff,handoff-and-complete,inbox-absorb,inbox-deny,inbox-drop,inbox-pending,list,outbox-list,outbox-record,overdue,show,transitions,unclaim,update,whoami; stream subs: archive,emit,list,show.

Full per-package suites (NO env override; honest baseline):
- daemon: 2330/2330 PASS / 0 FAIL.
- cli: 757/761 PASS / 4 FAIL named pre-existing flakes (chatroom wait timing, config-store env pollution, two preflight host-port). NO new regressions; matches prior PL-004 review baseline.

DB ISOLATION (per dispatch boundary):
- All test runs used explicit temp-rooted OPENRIG_HOME + OPENRIG_DB under `/tmp/pl004-r1-*`.
- No daemon launched. No production DB opened. The d8d8ee9 prior incident (5 additive tables in production DB) remains inert per orch decision; this commit does not interact with production state.

BOUNDARIES PRESERVED:
- Worktree-only changes; main checkout `/Users/wrandom/code/projects/openrig` HEAD unchanged at ab822ee throughout. No merge / rebase / cherry-pick / commit on main. Verified pre and post.
- 12 files changed (10 modified + 2 new), all in packages/daemon/{src,test} and packages/cli/{src/commands,test}.
- M2/M3/M4 surfaces (Restore-Packet vertical) NOT touched.
- Velocity prior-art .mjs unchanged.
- No host daemon mutation. No host credential mutation. No live rig topology mutation. No baseline VM mutation.
- No production DB cleanup attempted (per dispatch boundary).
- No standing Codex auto-review or approval-heavy loop (per safety stop).
- Normal posture approvals_reviewer=user honored throughout.

HANDOFF RECOMMENDATION: ready-for-qa-proof.

Driver-side focused verification passed (89 + 17 = 106 focused tests under explicit temp DB isolation + compiled-output runtime checks). QA-side sustained proof loop (cross-loop dogfood from disposable VM + Velocity-pair cross-rig validation + nudge persistence under real SessionTransport) belongs to velocity-qa per the velocity SOP correction (driver fast + implementation + focused verify; QA owns sustained VM/dogfood/E2E/runtime proof loops).

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>

packages/cli/src/commands/queue.ts

@@ -58,6 +58,7 @@ export function queueCommand(depsOverride?: QueueDeps): Command {
     .option("--tags <tags>", "Comma-separated tags")
     .option("--expires-at <iso>", "ISO timestamp at which the qitem expires")
     .option("--id <qitemId>", "Idempotent qitem_id (skip if not provided)")
+    .option("--no-nudge", "Suppress the default destination nudge (cold-queue)")
     .option("--json", "JSON output for agents")
     .action(async (opts: {
       source: string;
@@ -68,6 +69,7 @@ export function queueCommand(depsOverride?: QueueDeps): Command {
       tags?: string;
       expiresAt?: string;
       id?: string;
+      nudge?: boolean;
       json?: boolean;
     }) => {
       const deps = getDeps();
@@ -82,6 +84,7 @@ export function queueCommand(depsOverride?: QueueDeps): Command {
           tier: opts.tier,
           tags,
           expiresAt: opts.expiresAt,
+          nudge: opts.nudge,
         });
         printResult(opts.json ?? false, res.data, res.status);
       });
@@ -159,6 +162,7 @@ export function queueCommand(depsOverride?: QueueDeps): Command {
     .option("--priority <priority>", "Override priority for the new qitem")
     .option("--tier <tier>", "Override tier for the new qitem")
     .option("--tags <tags>", "Comma-separated tags for the new qitem")
+    .option("--no-nudge", "Suppress the default nudge to the new destination")
     .option("--json", "JSON output for agents")
     .action(async (qitemId: string, opts: {
       from: string;
@@ -168,6 +172,7 @@ export function queueCommand(depsOverride?: QueueDeps): Command {
       priority?: string;
       tier?: string;
       tags?: string;
+      nudge?: boolean;
       json?: boolean;
     }) => {
       const deps = getDeps();
@@ -181,11 +186,72 @@ export function queueCommand(depsOverride?: QueueDeps): Command {
           priority: opts.priority,
           tier: opts.tier,
           tags,
+          nudge: opts.nudge,
         });
         printResult(opts.json ?? false, res.data, res.status);
       });
     });
 
+  cmd
+    .command("handoff-and-complete <qitemId>")
+    .description(
+      "Atomic close (state=done, closure_reason=handed_off_to) + create new qitem owned by --to. Variant of handoff that fully terminates the source qitem."
+    )
+    .requiredOption("--from <session>", "Source seat handing off")
+    .requiredOption("--to <session>", "Destination seat receiving the new qitem")
+    .option("--body <text>", "New qitem body (defaults to source body)")
+    .option("--note <text>", "Transition note")
+    .option("--priority <priority>", "Override priority for the new qitem")
+    .option("--tier <tier>", "Override tier for the new qitem")
+    .option("--tags <tags>", "Comma-separated tags for the new qitem")
+    .option("--no-nudge", "Suppress the default nudge to the new destination")
+    .option("--json", "JSON output for agents")
+    .action(async (qitemId: string, opts: {
+      from: string;
+      to: string;
+      body?: string;
+      note?: string;
+      priority?: string;
+      tier?: string;
+      tags?: string;
+      nudge?: boolean;
+      json?: boolean;
+    }) => {
+      const deps = getDeps();
+      const tags = opts.tags ? opts.tags.split(",").map((s) => s.trim()).filter(Boolean) : undefined;
+      await withClient(deps, async (client) => {
+        const res = await client.post<unknown>(`/api/queue/${encodeURIComponent(qitemId)}/handoff-and-complete`, {
+          fromSession: opts.from,
+          toSession: opts.to,
+          body: opts.body,
+          transitionNote: opts.note,
+          priority: opts.priority,
+          tier: opts.tier,
+          tags,
+          nudge: opts.nudge,
+        });
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
+  cmd
+    .command("whoami")
+    .description("Show the caller's queue position from the daemon's perspective")
+    .requiredOption("--session <session>", "Caller's session name")
+    .option("--recent-limit <n>", "How many recent active qitems to include", "25")
+    .option("--json", "JSON output for agents")
+    .action(async (opts: { session: string; recentLimit: string; json?: boolean }) => {
+      const deps = getDeps();
+      const params = new URLSearchParams({
+        session: opts.session,
+        recentLimit: opts.recentLimit,
+      });
+      await withClient(deps, async (client) => {
+        const res = await client.get<unknown>(`/api/queue/whoami?${params.toString()}`);
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
   cmd
     .command("fallback <qitemId>")
     .description("Reroute a qitem to a fallback destination (e.g. unreachable seat)")

packages/cli/test/queue.test.ts

@@ -0,0 +1,286 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import type { QueueDeps } from "../src/commands/queue.js";
+import { createProgram } from "../src/index.js";
+
+/**
+ * `rig queue` CLI tests — PL-004 Phase A revision (R1).
+ *
+ * Pattern mirrors compact-plan.test.ts: mock daemon-lifecycle to fake a
+ * running daemon, inject a clientFactory that returns a stubbed HTTP client.
+ * Tests assert: command parsing, HTTP request shape, non-2xx exit handling,
+ * hot-potato error rendering. No real daemon, no DB, no network.
+ */
+
+vi.mock("../src/daemon-lifecycle.js", async () => {
+  const actual = await vi.importActual<Record<string, unknown>>("../src/daemon-lifecycle.js");
+  return {
+    ...actual,
+    getDaemonStatus: vi.fn(async () => ({ state: "running", healthy: true, pid: 1234, port: 7433 })),
+    getDaemonUrl: vi.fn(() => "http://localhost:7433"),
+  };
+});
+
+interface StubResponse {
+  status: number;
+  data: unknown;
+}
+
+function makeDeps(opts?: {
+  routes?: Record<string, StubResponse>;
+}): { deps: QueueDeps; calls: Array<{ method: string; path: string; body?: unknown }> } {
+  const calls: Array<{ method: string; path: string; body?: unknown }> = [];
+  const routes = opts?.routes ?? {};
+  return {
+    calls,
+    deps: {
+      lifecycleDeps: {} as QueueDeps["lifecycleDeps"],
+      clientFactory: () => ({
+        get: vi.fn(async (path: string) => {
+          calls.push({ method: "GET", path });
+          return routes[`GET ${path}`] ?? { status: 200, data: {} };
+        }),
+        getText: vi.fn(async (path: string) => {
+          calls.push({ method: "GET", path });
+          return { status: 200, data: "" };
+        }),
+        post: vi.fn(async (path: string, body: unknown) => {
+          calls.push({ method: "POST", path, body });
+          return routes[`POST ${path}`] ?? { status: 201, data: { qitemId: "qitem-test-1" } };
+        }),
+        delete: vi.fn(async (path: string) => {
+          calls.push({ method: "DELETE", path });
+          return { status: 204, data: null };
+        }),
+        postText: vi.fn(async (path: string) => {
+          calls.push({ method: "POST", path });
+          return { status: 200, data: "" };
+        }),
+        postExpectText: vi.fn(async (path: string) => {
+          calls.push({ method: "POST", path });
+          return { status: 200, data: "" };
+        }),
+      }) as unknown as ReturnType<QueueDeps["clientFactory"]>,
+    },
+  };
+}
+
+describe("rig queue CLI", () => {
+  let logs: string[];
+  let errors: string[];
+
+  beforeEach(() => {
+    logs = [];
+    errors = [];
+    vi.spyOn(console, "log").mockImplementation((...args) => logs.push(args.join(" ")));
+    vi.spyOn(console, "error").mockImplementation((...args) => errors.push(args.join(" ")));
+    process.exitCode = undefined;
+  });
+
+  it("queue is registered on createProgram with all R1 subcommands", async () => {
+    const { deps } = makeDeps();
+    const program = createProgram({ queueDeps: deps });
+    const queueCmd = program.commands.find((c) => c.name() === "queue");
+    expect(queueCmd).toBeDefined();
+    const subs = queueCmd!.commands.map((c) => c.name()).sort();
+    // R1 ratified contract: handoff-and-complete + whoami present alongside the originals.
+    expect(subs).toContain("create");
+    expect(subs).toContain("handoff");
+    expect(subs).toContain("handoff-and-complete");
+    expect(subs).toContain("whoami");
+    expect(subs).toContain("update");
+    expect(subs).toContain("inbox-drop");
+    expect(subs).toContain("inbox-absorb");
+    expect(subs).toContain("inbox-deny");
+    expect(subs).toContain("list");
+    expect(subs).toContain("show");
+  });
+
+  it("create POSTs to /api/queue/create with sourceSession + destinationSession + body", async () => {
+    const { deps, calls } = makeDeps({
+      routes: { "POST /api/queue/create": { status: 201, data: { qitemId: "qitem-x", state: "pending" } } },
+    });
+    const program = createProgram({ queueDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "queue", "create",
+      "--source", "alice@rig",
+      "--destination", "bob@rig",
+      "--body", "do thing",
+      "--json",
+    ]);
+    const create = calls.find((c) => c.path === "/api/queue/create");
+    expect(create).toBeDefined();
+    const body = create!.body as Record<string, unknown>;
+    expect(body.sourceSession).toBe("alice@rig");
+    expect(body.destinationSession).toBe("bob@rig");
+    expect(body.body).toBe("do thing");
+    // R1: commander's --no-nudge sets opts.nudge to true by default.
+    // The CLI sends nudge: true, and the daemon treats nudge !== false as nudging.
+    expect(body.nudge).toBe(true);
+  });
+
+  it("create --no-nudge passes nudge: false to the daemon (cold-queue opt-out)", async () => {
+    const { deps, calls } = makeDeps();
+    const program = createProgram({ queueDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "queue", "create",
+      "--source", "alice@rig",
+      "--destination", "bob@rig",
+      "--body", "cold",
+      "--no-nudge",
+    ]);
+    const create = calls.find((c) => c.path === "/api/queue/create");
+    expect((create!.body as { nudge: boolean }).nudge).toBe(false);
+  });
+
+  it("update --state done WITHOUT --closure-reason renders structured hot-potato error and exits non-zero", async () => {
+    const { deps } = makeDeps({
+      routes: {
+        "POST /api/queue/qitem-x/update": {
+          status: 400,
+          data: {
+            error: "missing_closure_reason",
+            message: "state=done requires closure_reason; valid values: handed_off_to, blocked_on, denied, canceled, no-follow-on, escalation",
+            validReasons: ["handed_off_to", "blocked_on", "denied", "canceled", "no-follow-on", "escalation"],
+          },
+        },
+      },
+    });
+    const program = createProgram({ queueDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "queue", "update", "qitem-x",
+      "--actor", "bob@rig",
+      "--state", "done",
+      "--json",
+    ]);
+    expect(process.exitCode).toBe(1);
+    const out = logs.join("\n");
+    expect(out).toContain("missing_closure_reason");
+    expect(out).toContain("validReasons");
+  });
+
+  it("handoff-and-complete POSTs to /api/queue/:id/handoff-and-complete with from + to", async () => {
+    const { deps, calls } = makeDeps({
+      routes: {
+        "POST /api/queue/qitem-src/handoff-and-complete": {
+          status: 201,
+          data: {
+            closed: { state: "done", closureReason: "handed_off_to" },
+            created: { state: "pending", qitemId: "qitem-new" },
+          },
+        },
+      },
+    });
+    const program = createProgram({ queueDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "queue", "handoff-and-complete", "qitem-src",
+      "--from", "bob@rig",
+      "--to", "carol@rig",
+      "--body", "carol's piece",
+      "--json",
+    ]);
+    const call = calls.find((c) => c.path === "/api/queue/qitem-src/handoff-and-complete");
+    expect(call).toBeDefined();
+    const body = call!.body as Record<string, unknown>;
+    expect(body.fromSession).toBe("bob@rig");
+    expect(body.toSession).toBe("carol@rig");
+    expect(body.body).toBe("carol's piece");
+  });
+
+  it("whoami GETs /api/queue/whoami with session + recentLimit query params", async () => {
+    const { deps, calls } = makeDeps({
+      routes: {
+        "GET /api/queue/whoami?session=bob%40rig&recentLimit=10": {
+          status: 200,
+          data: {
+            session: "bob@rig",
+            asDestination: { pending: 2, inProgress: 1, blocked: 0, recent: [] },
+            asSource: { total: 5 },
+          },
+        },
+      },
+    });
+    const program = createProgram({ queueDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "queue", "whoami",
+      "--session", "bob@rig",
+      "--recent-limit", "10",
+      "--json",
+    ]);
+    const call = calls.find((c) => c.method === "GET" && c.path.startsWith("/api/queue/whoami"));
+    expect(call).toBeDefined();
+    expect(call!.path).toContain("session=bob%40rig");
+    expect(call!.path).toContain("recentLimit=10");
+  });
+
+  it("create against unknown destination rig surfaces 400 error and exits non-zero", async () => {
+    const { deps } = makeDeps({
+      routes: {
+        "POST /api/queue/create": {
+          status: 400,
+          data: {
+            error: "unknown_destination_rig",
+            message: "destination_session bob@phantom-rig references an unknown rig",
+          },
+        },
+      },
+    });
+    const program = createProgram({ queueDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "queue", "create",
+      "--source", "alice@known-rig",
+      "--destination", "bob@phantom-rig",
+      "--body", "x",
+      "--json",
+    ]);
+    expect(process.exitCode).toBe(1);
+    const out = logs.join("\n");
+    expect(out).toContain("unknown_destination_rig");
+  });
+
+  it("handoff with --no-nudge passes nudge: false through to daemon", async () => {
+    const { deps, calls } = makeDeps({
+      routes: {
+        "POST /api/queue/qitem-x/handoff": {
+          status: 201,
+          data: { closed: {}, created: {} },
+        },
+      },
+    });
+    const program = createProgram({ queueDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "queue", "handoff", "qitem-x",
+      "--from", "bob@rig",
+      "--to", "carol@rig",
+      "--no-nudge",
+    ]);
+    const call = calls.find((c) => c.path === "/api/queue/qitem-x/handoff");
+    expect((call!.body as { nudge: boolean }).nudge).toBe(false);
+  });
+
+  it("list constructs /api/queue/list with filter params", async () => {
+    const { deps, calls } = makeDeps({
+      routes: { "GET /api/queue/list?destinationSession=bob%40rig&state=pending&limit=50": { status: 200, data: [] } },
+    });
+    const program = createProgram({ queueDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "queue", "list",
+      "--destination", "bob@rig",
+      "--state", "pending",
+      "--limit", "50",
+      "--json",
+    ]);
+    const call = calls.find((c) => c.method === "GET" && c.path.startsWith("/api/queue/list"));
+    expect(call).toBeDefined();
+    expect(call!.path).toContain("destinationSession=bob%40rig");
+    expect(call!.path).toContain("state=pending");
+    expect(call!.path).toContain("limit=50");
+  });
+});

packages/cli/test/stream.test.ts

@@ -0,0 +1,209 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import type { StreamDeps } from "../src/commands/stream.js";
+import { createProgram } from "../src/index.js";
+
+/**
+ * `rig stream` CLI tests — PL-004 Phase A revision (R1).
+ *
+ * Covers parser behavior, HTTP request shape, and non-2xx exit handling for
+ * stream emit / list / show / archive. SSE consumer is daemon-side only at
+ * Phase A (cli-side `watch` consumer is not in scope for v1; the route is
+ * exposed for browser/dashboard tooling and in-process tests).
+ */
+
+vi.mock("../src/daemon-lifecycle.js", async () => {
+  const actual = await vi.importActual<Record<string, unknown>>("../src/daemon-lifecycle.js");
+  return {
+    ...actual,
+    getDaemonStatus: vi.fn(async () => ({ state: "running", healthy: true, pid: 1234, port: 7433 })),
+    getDaemonUrl: vi.fn(() => "http://localhost:7433"),
+  };
+});
+
+interface StubResponse {
+  status: number;
+  data: unknown;
+}
+
+function makeDeps(opts?: {
+  routes?: Record<string, StubResponse>;
+}): { deps: StreamDeps; calls: Array<{ method: string; path: string; body?: unknown }> } {
+  const calls: Array<{ method: string; path: string; body?: unknown }> = [];
+  const routes = opts?.routes ?? {};
+  return {
+    calls,
+    deps: {
+      lifecycleDeps: {} as StreamDeps["lifecycleDeps"],
+      clientFactory: () => ({
+        get: vi.fn(async (path: string) => {
+          calls.push({ method: "GET", path });
+          return routes[`GET ${path}`] ?? { status: 200, data: [] };
+        }),
+        getText: vi.fn(async (path: string) => {
+          calls.push({ method: "GET", path });
+          return { status: 200, data: "" };
+        }),
+        post: vi.fn(async (path: string, body: unknown) => {
+          calls.push({ method: "POST", path, body });
+          return routes[`POST ${path}`] ?? { status: 201, data: { streamItemId: "stream-test-1" } };
+        }),
+        delete: vi.fn(async (path: string) => {
+          calls.push({ method: "DELETE", path });
+          return { status: 204, data: null };
+        }),
+        postText: vi.fn(async (path: string) => {
+          calls.push({ method: "POST", path });
+          return { status: 200, data: "" };
+        }),
+        postExpectText: vi.fn(async (path: string) => {
+          calls.push({ method: "POST", path });
+          return { status: 200, data: "" };
+        }),
+      }) as unknown as ReturnType<StreamDeps["clientFactory"]>,
+    },
+  };
+}
+
+describe("rig stream CLI", () => {
+  let logs: string[];
+  let errors: string[];
+
+  beforeEach(() => {
+    logs = [];
+    errors = [];
+    vi.spyOn(console, "log").mockImplementation((...args) => logs.push(args.join(" ")));
+    vi.spyOn(console, "error").mockImplementation((...args) => errors.push(args.join(" ")));
+    process.exitCode = undefined;
+  });
+
+  it("stream is registered on createProgram with emit/list/show/archive subcommands", () => {
+    const { deps } = makeDeps();
+    const program = createProgram({ streamDeps: deps });
+    const streamCmd = program.commands.find((c) => c.name() === "stream");
+    expect(streamCmd).toBeDefined();
+    const subs = streamCmd!.commands.map((c) => c.name()).sort();
+    expect(subs).toContain("emit");
+    expect(subs).toContain("list");
+    expect(subs).toContain("show");
+    expect(subs).toContain("archive");
+  });
+
+  it("emit POSTs to /api/stream/emit with sourceSession + body", async () => {
+    const { deps, calls } = makeDeps({
+      routes: {
+        "POST /api/stream/emit": {
+          status: 201,
+          data: { streamItemId: "stream-emitted", streamSortKey: "01KQ...", tsEmitted: "2026-05-03T00:00:00Z" },
+        },
+      },
+    });
+    const program = createProgram({ streamDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "stream", "emit",
+      "--source", "alice@rig",
+      "--body", "stream content",
+      "--json",
+    ]);
+    const call = calls.find((c) => c.path === "/api/stream/emit");
+    expect(call).toBeDefined();
+    const body = call!.body as Record<string, unknown>;
+    expect(body.sourceSession).toBe("alice@rig");
+    expect(body.body).toBe("stream content");
+  });
+
+  it("emit with --id passes idempotent stream_item_id through", async () => {
+    const { deps, calls } = makeDeps();
+    const program = createProgram({ streamDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "stream", "emit",
+      "--source", "alice@rig",
+      "--body", "x",
+      "--id", "stream-fixed-id",
+    ]);
+    const call = calls.find((c) => c.path === "/api/stream/emit");
+    expect((call!.body as { streamItemId: string }).streamItemId).toBe("stream-fixed-id");
+  });
+
+  it("list GETs /api/stream/list with limit + sourceSession query", async () => {
+    const { deps, calls } = makeDeps({
+      routes: { "GET /api/stream/list?sourceSession=bob%40rig&limit=20": { status: 200, data: [] } },
+    });
+    const program = createProgram({ streamDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "stream", "list",
+      "--source", "bob@rig",
+      "--limit", "20",
+      "--json",
+    ]);
+    const call = calls.find((c) => c.method === "GET" && c.path.startsWith("/api/stream/list"));
+    expect(call).toBeDefined();
+    expect(call!.path).toContain("sourceSession=bob%40rig");
+    expect(call!.path).toContain("limit=20");
+  });
+
+  it("show GETs /api/stream/:id", async () => {
+    const { deps, calls } = makeDeps({
+      routes: { "GET /api/stream/stream-x": { status: 200, data: { streamItemId: "stream-x" } } },
+    });
+    const program = createProgram({ streamDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "stream", "show", "stream-x",
+      "--json",
+    ]);
+    const call = calls.find((c) => c.method === "GET" && c.path === "/api/stream/stream-x");
+    expect(call).toBeDefined();
+  });
+
+  it("archive POSTs to /api/stream/:id/archive", async () => {
+    const { deps, calls } = makeDeps({
+      routes: { "POST /api/stream/stream-x/archive": { status: 200, data: { ok: true } } },
+    });
+    const program = createProgram({ streamDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "stream", "archive", "stream-x",
+      "--json",
+    ]);
+    const call = calls.find((c) => c.method === "POST" && c.path === "/api/stream/stream-x/archive");
+    expect(call).toBeDefined();
+  });
+
+  it("non-2xx response on emit exits non-zero with structured error in JSON", async () => {
+    const { deps } = makeDeps({
+      routes: {
+        "POST /api/stream/emit": {
+          status: 400,
+          data: { error: "sourceSession is required" },
+        },
+      },
+    });
+    const program = createProgram({ streamDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "stream", "emit",
+      "--source", "alice@rig",
+      "--body", "x",
+      "--json",
+    ]);
+    expect(process.exitCode).toBe(1);
+    const out = logs.join("\n");
+    expect(out).toContain("sourceSession is required");
+  });
+
+  it("show 404 for nonexistent id exits non-zero", async () => {
+    const { deps } = makeDeps({
+      routes: { "GET /api/stream/missing": { status: 404, data: { error: "stream item not found" } } },
+    });
+    const program = createProgram({ streamDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "stream", "show", "missing",
+      "--json",
+    ]);
+    expect(process.exitCode).toBe(1);
+  });
+});

packages/daemon/src/domain/inbox-handler.ts

@@ -128,7 +128,7 @@ export class InboxHandler {
    * Idempotent on inbox_id: if already absorbed, returns the existing
    * `absorbed_qitem_id` rather than creating a duplicate.
    */
-  absorb(inboxId: string, receiverSession: string): { entry: InboxEntry; qitemId: string } {
+  async absorb(inboxId: string, receiverSession: string): Promise<{ entry: InboxEntry; qitemId: string }> {
     const entry = this.getById(inboxId);
     if (!entry) {
       throw new InboxHandlerError("inbox_not_found", `inbox ${inboxId} not found`);
@@ -149,12 +149,16 @@ export class InboxHandler {
       );
     }
 
-    const qitem = this.queueRepo.create({
+    // Inbox-absorb suppresses the default queue create-time nudge: the inbox
+    // surface already had its own delivery (the original drop), so a second
+    // nudge on absorb is duplicative. The destination knows it absorbed.
+    const qitem = await this.queueRepo.create({
       sourceSession: entry.senderSession,
       destinationSession: entry.destinationSession,
       body: entry.body,
       tags: entry.tags ?? undefined,
       priority: entry.urgency === "critical" ? "critical" : entry.urgency === "urgent" ? "urgent" : "routine",
+      nudge: false,
     });
 
     const ts = new Date().toISOString();

packages/daemon/src/domain/queue-repository.ts

@@ -74,6 +74,23 @@ interface QueueItemRow {
   resolution: string | null;
 }
 
+/**
+ * Async transport contract — exists in this domain module so QueueRepository
+ * can do durable+waking handoffs (Phase A contract: queue create / handoff /
+ * handoff-and-complete are nudging by default unless caller opts out).
+ *
+ * The wired-in implementation is `SessionTransport` (packages/daemon/src/
+ * domain/session-transport.ts), but the repository depends only on this
+ * minimal shape so test code can supply a stub.
+ */
+export interface QueueNudgeTransport {
+  send(
+    sessionName: string,
+    text: string,
+    opts?: { verify?: boolean }
+  ): Promise<{ ok: boolean; verified?: boolean; error?: string; reason?: string }>;
+}
+
 export interface QueueCreateInput {
   qitemId?: string;
   sourceSession: string;
@@ -84,6 +101,13 @@ export interface QueueCreateInput {
   tags?: string[];
   expiresAt?: string;
   chainOfRecord?: string[];
+  /**
+   * R1 fix (PL-004 Phase A revision): Phase A is durable + waking by default.
+   * When true (or omitted), the repository nudges the destination after the
+   * create transaction commits and persists last_nudge_attempt + last_nudge_result.
+   * Operators opt out with `nudge: false` for cold-queue cases.
+   */
+  nudge?: boolean;
 }
 
 export interface QueueUpdateInput {
@@ -104,8 +128,19 @@ export interface QueueHandoffInput {
   priority?: QueuePriority;
   tier?: string;
   tags?: string[];
+  /** Default true; nudge the destination after the close+create transaction. */
+  nudge?: boolean;
 }
 
+/**
+ * Like {@link QueueHandoffInput} but the source qitem is closed as `done`
+ * (terminal) instead of `handed-off` (intermediate). Use when the source seat
+ * is fully complete with the work and the new qitem is the canonical
+ * follow-on. Closure_reason is recorded as `handed_off_to` and the new qitem
+ * is created in the same atomic transaction.
+ */
+export interface QueueHandoffAndCompleteInput extends QueueHandoffInput {}
+
 export interface QueueClaimInput {
   qitemId: string;
   destinationSession: string;
@@ -150,19 +185,69 @@ export class QueueRepository {
   readonly transitionLog: QueueTransitionLog;
   private readonly eventBus: EventBus;
   private readonly validateRig: (sessionRef: string) => boolean;
+  private transport: QueueNudgeTransport | undefined;
 
   constructor(
     db: Database.Database,
     eventBus: EventBus,
-    opts?: { validateRig?: (sessionRef: string) => boolean }
+    opts?: {
+      validateRig?: (sessionRef: string) => boolean;
+      /**
+       * R1 fix (PL-004 Phase A revision): durable+waking-by-default transport
+       * for create / handoff / handoff-and-complete. When provided, the
+       * repository nudges the destination after the corresponding transaction
+       * commits and records last_nudge_attempt + last_nudge_result via
+       * recordNudgeAttempt(). When absent, no nudge is issued (caller is in
+       * a test or daemon-bootstrap path where transport is not yet wired).
+       */
+      transport?: QueueNudgeTransport;
+    }
   ) {
     this.db = db;
     this.eventBus = eventBus;
     this.transitionLog = new QueueTransitionLog(db);
     this.validateRig = opts?.validateRig ?? (() => true);
+    this.transport = opts?.transport;
   }
 
-  create(input: QueueCreateInput): QueueItem {
+  /**
+   * Attach the wake-path transport AFTER construction. Used by daemon
+   * startup, where SessionTransport is constructed later in the dep graph
+   * than QueueRepository (because SessionTransport needs agentActivityStore
+   * which itself needs eventBus). Calling this is safe at any time; create /
+   * handoff / handoff-and-complete will start nudging on the next call.
+   */
+  attachTransport(transport: QueueNudgeTransport): void {
+    this.transport = transport;
+  }
+
+  /**
+   * Issue a default nudge to the destination after a create / handoff /
+   * handoff-and-complete commit. Records the result via recordNudgeAttempt.
+   * Errors are caught and surfaced as nudge_result strings — they do not
+   * unwind the underlying queue mutation.
+   */
+  private async maybeNudge(
+    qitemId: string,
+    destinationSession: string,
+    nudgeOpt: boolean | undefined,
+  ): Promise<void> {
+    if (nudgeOpt === false) return;
+    if (!this.transport) return;
+    const text = `Queue handoff: ${qitemId} - check your queue.`;
+    try {
+      const res = await this.transport.send(destinationSession, text, { verify: true });
+      const result = res.ok
+        ? (res.verified ? "verified" : "sent-unverified")
+        : `failed:${res.error ?? res.reason ?? "unknown"}`;
+      this.recordNudgeAttempt(qitemId, result);
+    } catch (err) {
+      const msg = err instanceof Error ? err.message : String(err);
+      this.recordNudgeAttempt(qitemId, `failed:${msg}`);
+    }
+  }
+
+  async create(input: QueueCreateInput): Promise<QueueItem> {
     if (!this.validateRig(input.destinationSession)) {
       throw new QueueRepositoryError(
         "unknown_destination_rig",
@@ -219,6 +304,7 @@ export class QueueRepository {
 
     const persistedEvent = txn();
     this.eventBus.notifySubscribers(persistedEvent);
+    await this.maybeNudge(id, input.destinationSession, input.nudge);
     return this.getByIdOrThrow(id);
   }
 
@@ -227,7 +313,7 @@ export class QueueRepository {
    * closure_reason=handed_off_to) and create a new qitem owned by `toSession`,
    * with `handed_off_from` recording the chain. One atomic transaction.
    */
-  handoff(input: QueueHandoffInput): { closed: QueueItem; created: QueueItem } {
+  async handoff(input: QueueHandoffInput): Promise<{ closed: QueueItem; created: QueueItem }> {
     const source = this.getById(input.qitemId);
     if (!source) {
       throw new QueueRepositoryError(
@@ -333,12 +419,181 @@ export class QueueRepository {
       this.eventBus.notifySubscribers(e.payload as import("./types.js").PersistedEvent);
     }
 
+    await this.maybeNudge(newId, input.toSession, input.nudge);
+
     return {
       closed: this.getByIdOrThrow(source.qitemId),
       created: this.getByIdOrThrow(newId),
     };
   }
 
+  /**
+   * Variant of {@link handoff} that closes the source qitem as `done`
+   * (terminal closure) instead of `handed-off` (intermediate). Same atomic
+   * close+create, same chain_of_record semantics, same default-nudge behavior.
+   * Use when the source seat is fully complete with the work — no follow-up
+   * tracking needed against the source qitem.
+   */
+  async handoffAndComplete(input: QueueHandoffAndCompleteInput): Promise<{ closed: QueueItem; created: QueueItem }> {
+    const source = this.getById(input.qitemId);
+    if (!source) {
+      throw new QueueRepositoryError(
+        "qitem_not_found",
+        `qitem ${input.qitemId} not found`
+      );
+    }
+    if (source.state === "done" || source.state === "handed-off") {
+      throw new QueueRepositoryError(
+        "qitem_already_terminal",
+        `qitem ${input.qitemId} is already in terminal state ${source.state}`
+      );
+    }
+    if (!this.validateRig(input.toSession)) {
+      throw new QueueRepositoryError(
+        "unknown_destination_rig",
+        `to_session ${input.toSession} references an unknown rig`
+      );
+    }
+
+    const newId = newQitemId();
+    const ts = new Date().toISOString();
+    const body = input.body ?? source.body;
+    const priority = input.priority ?? source.priority;
+    const tier = input.tier ?? source.tier;
+    const tags = input.tags ? JSON.stringify(input.tags) : (source.tags ? JSON.stringify(source.tags) : null);
+    const chain = JSON.stringify([...(source.chainOfRecord ?? []), source.qitemId]);
+
+    const events: Array<{ name: string; payload: import("./types.js").RigEvent }> = [];
+
+    const txn = this.db.transaction(() => {
+      this.db
+        .prepare(
+          `UPDATE queue_items
+             SET state = 'done',
+                 ts_updated = ?,
+                 handed_off_to = ?,
+                 closure_reason = 'handed_off_to',
+                 closure_target = ?
+           WHERE qitem_id = ?`
+        )
+        .run(ts, input.toSession, input.toSession, source.qitemId);
+
+      this.transitionLog.append({
+        qitemId: source.qitemId,
+        state: "done",
+        actorSession: input.fromSession,
+        transitionNote: input.transitionNote ?? `handoff-and-complete to ${input.toSession}`,
+        closureReason: "handed_off_to",
+        closureTarget: input.toSession,
+      });
+
+      this.db
+        .prepare(
+          `INSERT INTO queue_items (
+            qitem_id, ts_created, ts_updated, source_session, destination_session,
+            state, priority, tier, tags, handed_off_from, chain_of_record, body
+          ) VALUES (?, ?, ?, ?, ?, 'pending', ?, ?, ?, ?, ?, ?)`
+        )
+        .run(
+          newId,
+          ts,
+          ts,
+          input.fromSession,
+          input.toSession,
+          priority,
+          tier,
+          tags,
+          source.qitemId,
+          chain,
+          body
+        );
+
+      this.transitionLog.append({
+        qitemId: newId,
+        state: "pending",
+        actorSession: input.fromSession,
+        transitionNote: `handoff-and-complete from ${source.qitemId}`,
+      });
+
+      const handoffEvent = this.eventBus.persistWithinTransaction({
+        type: "queue.handed_off",
+        qitemId: source.qitemId,
+        fromSession: input.fromSession,
+        toSession: input.toSession,
+        closureReason: "handed_off_to",
+      });
+      events.push({ name: "queue.handed_off", payload: handoffEvent });
+
+      const createdEvent = this.eventBus.persistWithinTransaction({
+        type: "queue.created",
+        qitemId: newId,
+        sourceSession: input.fromSession,
+        destinationSession: input.toSession,
+        priority,
+        tier,
+      });
+      events.push({ name: "queue.created", payload: createdEvent });
+    });
+
+    txn();
+    for (const e of events) {
+      this.eventBus.notifySubscribers(e.payload as import("./types.js").PersistedEvent);
+    }
+
+    await this.maybeNudge(newId, input.toSession, input.nudge);
+
+    return {
+      closed: this.getByIdOrThrow(source.qitemId),
+      created: this.getByIdOrThrow(newId),
+    };
+  }
+
+  /**
+   * `whoami` — return the seat's queue position from the daemon's perspective.
+   * Counts active qitems (pending + in-progress + blocked) destined for the
+   * caller, lists the most recent active qitems, and reports counts for the
+   * caller's outgoing source role too. Read-only; no mutations.
+   *
+   * Per PL-004 Phase A § Routes: GET /api/queue/whoami.
+   */
+  whoami(session: string, opts?: { recentLimit?: number }): {
+    session: string;
+    asDestination: { pending: number; inProgress: number; blocked: number; recent: QueueItem[] };
+    asSource: { total: number };
+  } {
+    const limit = Math.max(1, Math.min(opts?.recentLimit ?? 25, 200));
+    const countByState = (state: string): number => {
+      const row = this.db
+        .prepare(
+          `SELECT COUNT(*) AS n FROM queue_items WHERE destination_session = ? AND state = ?`
+        )
+        .get(session, state) as { n: number };
+      return row.n;
+    };
+    const recent = this.db
+      .prepare(
+        `SELECT * FROM queue_items
+          WHERE destination_session = ?
+            AND state IN ('pending','in-progress','blocked')
+          ORDER BY ts_updated DESC
+          LIMIT ?`
+      )
+      .all(session, limit) as QueueItemRow[];
+    const sourceTotalRow = this.db
+      .prepare(`SELECT COUNT(*) AS n FROM queue_items WHERE source_session = ?`)
+      .get(session) as { n: number };
+    return {
+      session,
+      asDestination: {
+        pending: countByState("pending"),
+        inProgress: countByState("in-progress"),
+        blocked: countByState("blocked"),
+        recent: recent.map((r) => this.rowToItem(r)),
+      },
+      asSource: { total: sourceTotalRow.n },
+    };
+  }
+
   /**
    * Mark a qitem `in-progress` (claim). Computes closure_required_at from tier.
    */

packages/daemon/src/routes/queue.ts

@@ -82,7 +82,7 @@ export function queueRoutes(): Hono {
     if (!body.body) return c.json({ error: "body is required" }, 400);
 
     try {
-      const item = getRepo(c).create({
+      const item = await getRepo(c).create({
         qitemId: body.qitemId,
         sourceSession: body.sourceSession,
         destinationSession: body.destinationSession,
@@ -92,6 +92,7 @@ export function queueRoutes(): Hono {
         tags: body.tags,
         expiresAt: body.expiresAt,
         chainOfRecord: body.chainOfRecord,
+        nudge: (body as { nudge?: boolean }).nudge,
       });
       return c.json(item, 201);
     } catch (err) {
@@ -169,7 +170,7 @@ export function queueRoutes(): Hono {
     if (!body.toSession) return c.json({ error: "toSession is required" }, 400);
 
     try {
-      const result = getRepo(c).handoff({
+      const result = await getRepo(c).handoff({
         qitemId,
         fromSession: body.fromSession,
         toSession: body.toSession,
@@ -178,6 +179,43 @@ export function queueRoutes(): Hono {
         priority: body.priority,
         tier: body.tier,
         tags: body.tags,
+        nudge: (body as { nudge?: boolean }).nudge,
+      });
+      return c.json(result, 201);
+    } catch (err) {
+      return errorResponse(c, err);
+    }
+  });
+
+  // POST /:qitemId/handoff-and-complete — variant of handoff that closes
+  // source as `done` (terminal) instead of `handed-off` (intermediate).
+  // Same atomic close+create + chain_of_record + default-nudge contract.
+  app.post("/:qitemId/handoff-and-complete", async (c) => {
+    const qitemId = c.req.param("qitemId");
+    const body = await c.req.json<{
+      fromSession?: string;
+      toSession?: string;
+      body?: string;
+      transitionNote?: string;
+      priority?: QueuePriority;
+      tier?: string;
+      tags?: string[];
+      nudge?: boolean;
+    }>().catch(() => ({} as never));
+    if (!body.fromSession) return c.json({ error: "fromSession is required" }, 400);
+    if (!body.toSession) return c.json({ error: "toSession is required" }, 400);
+
+    try {
+      const result = await getRepo(c).handoffAndComplete({
+        qitemId,
+        fromSession: body.fromSession,
+        toSession: body.toSession,
+        body: body.body,
+        transitionNote: body.transitionNote,
+        priority: body.priority,
+        tier: body.tier,
+        tags: body.tags,
+        nudge: body.nudge,
       });
       return c.json(result, 201);
     } catch (err) {
@@ -198,6 +236,17 @@ export function queueRoutes(): Hono {
     }
   });
 
+  // GET /whoami — caller's queue position from the daemon's perspective.
+  // MUST precede /:qitemId so the literal path wins.
+  app.get("/whoami", (c) => {
+    const session = c.req.query("session");
+    if (!session) return c.json({ error: "session is required" }, 400);
+    const recentLimit = c.req.query("recentLimit")
+      ? Number.parseInt(c.req.query("recentLimit")!, 10)
+      : undefined;
+    return c.json(getRepo(c).whoami(session, { recentLimit }));
+  });
+
   // GET /list — list with filters. MUST precede /:qitemId so the literal path wins.
   app.get("/list", (c) => {
     const destinationSession = c.req.query("destinationSession") || undefined;
@@ -274,7 +323,7 @@ export function queueRoutes(): Hono {
     const body = await c.req.json<{ receiverSession?: string }>().catch(() => ({} as never));
     if (!body.receiverSession) return c.json({ error: "receiverSession is required" }, 400);
     try {
-      const result = getInbox(c).absorb(inboxId, body.receiverSession);
+      const result = await getInbox(c).absorb(inboxId, body.receiverSession);
       return c.json(result);
     } catch (err) {
       return errorResponse(c, err);
@@ -343,8 +392,10 @@ export function queueRoutes(): Hono {
   });
 
   // ---- SSE watch over coordination events ----
+  // Mounted at both /watch (legacy alias) and /sse (Phase A contract per IMPL).
+  // Same handler; either path emits the identical event stream.
 
-  app.get("/watch", (c) => {
+  const sseHandler = (c: Parameters<typeof streamSSE>[0]) => {
     const eventBus = getEventBus(c);
 
     return streamSSE(c, async (stream) => {
@@ -371,7 +422,10 @@ export function queueRoutes(): Hono {
         unsubscribe();
       }
     });
-  });
+  };
+
+  app.get("/watch", sseHandler);
+  app.get("/sse", sseHandler);
 
   return app;
 }

packages/daemon/src/routes/stream.ts

@@ -83,8 +83,11 @@ export function streamRoutes(): Hono {
     return c.json({ ok: true });
   });
 
-  // GET /watch — SSE for new stream.emitted events
-  app.get("/watch", (c) => {
+  // GET /watch — SSE for new stream.emitted events.
+  // Mounted at both /watch (legacy alias) and /sse (Phase A contract per
+  // IMPL § Routes: GET /api/stream/sse). Same handler; either path emits
+  // the identical event stream.
+  const sseHandler = (c: Parameters<typeof streamSSE>[0]) => {
     const eventBus = getEventBus(c);
     const store = getStore(c);
 
@@ -124,7 +127,10 @@ export function streamRoutes(): Hono {
         unsubscribe();
       }
     });
-  });
+  };
+
+  app.get("/watch", sseHandler);
+  app.get("/sse", sseHandler);
 
   return app;
 }

packages/daemon/src/startup.ts

@@ -126,9 +126,23 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
   const rigRepo = new RigRepository(db);
   const sessionRegistry = new SessionRegistry(db);
   const eventBus = new EventBus(db);
+  // PL-004 Phase A revision (R1): topology-backed validateRig.
+  // Reject `<member>@<unknown-rig>` shapes by checking the rig portion
+  // against the rig registry. Bare ids without `@` are also rejected
+  // (no canonical rig binding).
+  const topologyValidateRig = (sessionRef: string): boolean => {
+    const m = /^[^@]+@(.+)$/.exec(sessionRef);
+    if (!m) return false;
+    const rigName = m[1]!;
+    return rigRepo.findRigsByName(rigName).length > 0;
+  };
   // PL-004 Phase A — shared coordination services. Constructed early so
   // both the queueRepo dep slot and inboxHandler can share one instance.
-  const queueRepoInstance = new QueueRepository(db, eventBus);
+  // Transport is wired after SessionTransport instantiation below via
+  // attachTransport().
+  const queueRepoInstance = new QueueRepository(db, eventBus, {
+    validateRig: topologyValidateRig,
+  });
 
   const tmuxAdapter = new TmuxAdapter(opts?.tmuxExec ?? execCommand);
   // cmuxFactory takes precedence (for tests), then cmuxExec-based CLI transport, then default
@@ -370,7 +384,13 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
     podBundleSourceResolver,
     runtimeAdapters: { "claude-code": claudeAdapter, "codex": codexAdapter, "terminal": new (await import("./adapters/terminal-adapter.js")).TerminalAdapter() },
     transcriptStore,
-    sessionTransport: new SessionTransport({ db, rigRepo, sessionRegistry, tmuxAdapter, agentActivityStore }),
+    sessionTransport: (() => {
+      const t = new SessionTransport({ db, rigRepo, sessionRegistry, tmuxAdapter, agentActivityStore });
+      // PL-004 Phase A revision (R1): wire QueueRepository's wake-path so
+      // create / handoff / handoff-and-complete nudge by default.
+      queueRepoInstance.attachTransport(t);
+      return t;
+    })(),
     chatRepo: new ChatRepository(db),
     streamStore: new StreamStore(db, eventBus),
     queueRepo: queueRepoInstance,

packages/daemon/test/inbox-handler.test.ts

@@ -82,14 +82,14 @@ describe("InboxHandler", () => {
     ).toThrow(InboxHandlerError);
   });
 
-  it("absorb promotes pending entry to a queue_item, emits inbox.absorbed", () => {
+  it("absorb promotes pending entry to a queue_item, emits inbox.absorbed", async () => {
     const entry = inbox.drop({
       destinationSession: "bob@rig",
       senderSession: "alice@rig",
       body: "review this",
       urgency: "urgent",
     });
-    const result = inbox.absorb(entry.inboxId, "bob@rig");
+    const result = await inbox.absorb(entry.inboxId, "bob@rig");
     expect(result.entry.state).toBe("absorbed");
     expect(result.entry.absorbedQitemId).toBe(result.qitemId);
 
@@ -102,27 +102,27 @@ describe("InboxHandler", () => {
     expect(captured.some((e) => e.type === "inbox.absorbed")).toBe(true);
   });
 
-  it("absorb is idempotent — second call returns same qitem_id", () => {
+  it("absorb is idempotent — second call returns same qitem_id", async () => {
     const entry = inbox.drop({
       destinationSession: "bob@rig",
       senderSession: "alice@rig",
       body: "x",
     });
-    const a = inbox.absorb(entry.inboxId, "bob@rig");
-    const b = inbox.absorb(entry.inboxId, "bob@rig");
+    const a = await inbox.absorb(entry.inboxId, "bob@rig");
+    const b = await inbox.absorb(entry.inboxId, "bob@rig");
     expect(a.qitemId).toBe(b.qitemId);
   });
 
-  it("absorb refuses if destination doesn't match", () => {
+  it("absorb refuses if destination doesn't match", async () => {
     const entry = inbox.drop({
       destinationSession: "bob@rig",
       senderSession: "alice@rig",
       body: "x",
     });
-    expect(() => inbox.absorb(entry.inboxId, "carol@rig")).toThrow(/destined for/);
+    await expect(inbox.absorb(entry.inboxId, "carol@rig")).rejects.toThrow(/destined for/);
   });
 
-  it("deny records reason + emits inbox.denied; cannot subsequently absorb", () => {
+  it("deny records reason + emits inbox.denied; cannot subsequently absorb", async () => {
     const entry = inbox.drop({
       destinationSession: "bob@rig",
       senderSession: "alice@rig",
@@ -132,7 +132,7 @@ describe("InboxHandler", () => {
     expect(denied.state).toBe("denied");
     expect(denied.deniedReason).toBe("off-topic-for-this-rig");
     expect(captured.some((e) => e.type === "inbox.denied")).toBe(true);
-    expect(() => inbox.absorb(entry.inboxId, "bob@rig")).toThrow(/denied/);
+    await expect(inbox.absorb(entry.inboxId, "bob@rig")).rejects.toThrow(/denied/);
   });
 
   it("listPending returns only pending entries for the given destination", () => {

packages/daemon/test/queue-repository.test.ts

@@ -31,8 +31,8 @@ describe("QueueRepository", () => {
 
   afterEach(() => db.close());
 
-  it("create stamps qitem_id + transition + queue.created event", () => {
-    const item = repo.create({
+  it("create stamps qitem_id + transition + queue.created event", async () => {
+    const item = await repo.create({
       sourceSession: "alice@rig-a",
       destinationSession: "bob@rig-b",
       body: "do the thing",
@@ -46,21 +46,21 @@ describe("QueueRepository", () => {
     expect(transitions[0]!.state).toBe("pending");
   });
 
-  it("create rejects unknown rig when validateRig denies", () => {
+  it("create rejects unknown rig when validateRig denies", async () => {
     const strictRepo = new QueueRepository(db, bus, {
       validateRig: (s) => s.endsWith("@known-rig"),
     });
-    expect(() =>
+    await expect(
       strictRepo.create({
         sourceSession: "alice@known-rig",
         destinationSession: "bob@phantom-rig",
         body: "x",
       })
-    ).toThrow(/unknown rig/);
+    ).rejects.toThrow(/unknown rig/);
   });
 
-  it("claim transitions pending → in-progress and computes closure_required_at from tier", () => {
-    const item = repo.create({
+  it("claim transitions pending → in-progress and computes closure_required_at from tier", async () => {
+    const item = await repo.create({
       sourceSession: "alice@rig",
       destinationSession: "bob@rig",
       body: "x",
@@ -73,8 +73,8 @@ describe("QueueRepository", () => {
     expect(captured.some((e) => e.type === "queue.claimed")).toBe(true);
   });
 
-  it("claim rejects mismatched destination", () => {
-    const item = repo.create({
+  it("claim rejects mismatched destination", async () => {
+    const item = await repo.create({
       sourceSession: "alice@rig",
       destinationSession: "bob@rig",
       body: "x",
@@ -84,8 +84,8 @@ describe("QueueRepository", () => {
     );
   });
 
-  it("update state=done WITHOUT closure_reason rejected with missing_closure_reason", () => {
-    const item = repo.create({
+  it("update state=done WITHOUT closure_reason rejected with missing_closure_reason", async () => {
+    const item = await repo.create({
       sourceSession: "alice@rig",
       destinationSession: "bob@rig",
       body: "x",
@@ -102,9 +102,9 @@ describe("QueueRepository", () => {
     }
   });
 
-  it("update accepts each of the 6 valid closure reasons", () => {
+  it("update accepts each of the 6 valid closure reasons", async () => {
     for (const reason of CLOSURE_REASONS) {
-      const item = repo.create({
+      const item = await repo.create({
         sourceSession: "alice@rig",
         destinationSession: "bob@rig",
         body: `for ${reason}`,
@@ -123,13 +123,13 @@ describe("QueueRepository", () => {
     }
   });
 
-  it("handoff is transactional: closes source as handed-off + creates new qitem", () => {
-    const item = repo.create({
+  it("handoff is transactional: closes source as handed-off + creates new qitem", async () => {
+    const item = await repo.create({
       sourceSession: "alice@rig",
       destinationSession: "bob@rig",
       body: "do it",
     });
-    const result = repo.handoff({
+    const result = await repo.handoff({
       qitemId: item.qitemId,
       fromSession: "bob@rig",
       toSession: "carol@rig",
@@ -147,8 +147,8 @@ describe("QueueRepository", () => {
     expect(captured.filter((e) => e.type === "queue.created")).toHaveLength(2); // create + handoff-create
   });
 
-  it("handoff refuses on already-terminal qitem", () => {
-    const item = repo.create({
+  it("handoff refuses on already-terminal qitem", async () => {
+    const item = await repo.create({
       sourceSession: "alice@rig",
       destinationSession: "bob@rig",
       body: "x",
@@ -160,17 +160,17 @@ describe("QueueRepository", () => {
       state: "done",
       closureReason: "no-follow-on",
     });
-    expect(() =>
+    await expect(
       repo.handoff({
         qitemId: item.qitemId,
         fromSession: "bob@rig",
         toSession: "carol@rig",
       })
-    ).toThrow(/terminal/);
+    ).rejects.toThrow(/terminal/);
   });
 
-  it("transitions are append-only — every state change adds a row", () => {
-    const item = repo.create({
+  it("transitions are append-only — every state change adds a row", async () => {
+    const item = await repo.create({
       sourceSession: "alice@rig",
       destinationSession: "bob@rig",
       body: "x",
@@ -194,8 +194,8 @@ describe("QueueRepository", () => {
     ]);
   });
 
-  it("findOverdue surfaces in-progress qitems past closure_required_at", () => {
-    const item = repo.create({
+  it("findOverdue surfaces in-progress qitems past closure_required_at", async () => {
+    const item = await repo.create({
       sourceSession: "alice@rig",
       destinationSession: "bob@rig",
       body: "x",
@@ -207,8 +207,8 @@ describe("QueueRepository", () => {
     expect(overdue.map((q) => q.qitemId)).toContain(item.qitemId);
   });
 
-  it("routeToFallback emits qitem.fallback_routed and rewrites destination", () => {
-    const item = repo.create({
+  it("routeToFallback emits qitem.fallback_routed and rewrites destination", async () => {
+    const item = await repo.create({
       sourceSession: "alice@rig",
       destinationSession: "bob@rig",
       body: "x",
@@ -219,14 +219,224 @@ describe("QueueRepository", () => {
     expect(captured.some((e) => e.type === "qitem.fallback_routed")).toBe(true);
   });
 
-  it("list filters by destination + state", () => {
-    const a = repo.create({ sourceSession: "x@r", destinationSession: "bob@r", body: "1" });
-    repo.create({ sourceSession: "x@r", destinationSession: "carol@r", body: "2" });
-    repo.create({ sourceSession: "x@r", destinationSession: "bob@r", body: "3" });
+  it("list filters by destination + state", async () => {
+    const a = await repo.create({ sourceSession: "x@r", destinationSession: "bob@r", body: "1" });
+    await repo.create({ sourceSession: "x@r", destinationSession: "carol@r", body: "2" });
+    await repo.create({ sourceSession: "x@r", destinationSession: "bob@r", body: "3" });
     repo.claim({ qitemId: a.qitemId, destinationSession: "bob@r" });
 
     expect(repo.list({ destinationSession: "bob@r" })).toHaveLength(2);
     expect(repo.list({ destinationSession: "bob@r", state: "in-progress" })).toHaveLength(1);
     expect(repo.list({ destinationSession: "bob@r", state: ["pending", "in-progress"] })).toHaveLength(2);
   });
+
+  // ---- PL-004 Phase A revision (R1) tests ----
+
+  describe("R1 default-nudge wiring", () => {
+    it("create nudges destination by default and persists last_nudge_attempt + last_nudge_result", async () => {
+      const sends: Array<{ session: string; text: string }> = [];
+      const stubTransport = {
+        send: async (sessionName: string, text: string) => {
+          sends.push({ session: sessionName, text });
+          return { ok: true, verified: true };
+        },
+      };
+      const nudgingRepo = new QueueRepository(db, bus, { transport: stubTransport });
+      const item = await nudgingRepo.create({
+        sourceSession: "alice@rig",
+        destinationSession: "bob@rig",
+        body: "ping me",
+      });
+      expect(sends).toHaveLength(1);
+      expect(sends[0]!.session).toBe("bob@rig");
+      expect(sends[0]!.text).toContain(item.qitemId);
+      const fresh = nudgingRepo.getById(item.qitemId)!;
+      expect(fresh.lastNudgeAttempt).not.toBeNull();
+      expect(fresh.lastNudgeResult).toBe("verified");
+    });
+
+    it("create with nudge:false does NOT call transport (cold-queue opt-out)", async () => {
+      const sends: Array<{ session: string; text: string }> = [];
+      const stubTransport = {
+        send: async (sessionName: string, text: string) => {
+          sends.push({ session: sessionName, text });
+          return { ok: true };
+        },
+      };
+      const nudgingRepo = new QueueRepository(db, bus, { transport: stubTransport });
+      const item = await nudgingRepo.create({
+        sourceSession: "alice@rig",
+        destinationSession: "bob@rig",
+        body: "cold",
+        nudge: false,
+      });
+      expect(sends).toHaveLength(0);
+      const fresh = nudgingRepo.getById(item.qitemId)!;
+      expect(fresh.lastNudgeAttempt).toBeNull();
+      expect(fresh.lastNudgeResult).toBeNull();
+    });
+
+    it("nudge failure is recorded as failed:<reason>; create still succeeds", async () => {
+      const stubTransport = {
+        send: async () => ({ ok: false, error: "tmux pane not found" }),
+      };
+      const nudgingRepo = new QueueRepository(db, bus, { transport: stubTransport });
+      const item = await nudgingRepo.create({
+        sourceSession: "alice@rig",
+        destinationSession: "bob@rig",
+        body: "x",
+      });
+      const fresh = nudgingRepo.getById(item.qitemId)!;
+      expect(fresh.lastNudgeResult).toMatch(/^failed:/);
+      // Item itself created normally — nudge failures don't unwind the create.
+      expect(fresh.state).toBe("pending");
+    });
+
+    it("handoff nudges new destination by default", async () => {
+      const sends: Array<{ session: string }> = [];
+      const stubTransport = {
+        send: async (sessionName: string) => {
+          sends.push({ session: sessionName });
+          return { ok: true, verified: true };
+        },
+      };
+      const nudgingRepo = new QueueRepository(db, bus, { transport: stubTransport });
+      const original = await nudgingRepo.create({
+        sourceSession: "alice@rig",
+        destinationSession: "bob@rig",
+        body: "x",
+        nudge: false, // suppress create-time nudge so we count only handoff
+      });
+      const result = await nudgingRepo.handoff({
+        qitemId: original.qitemId,
+        fromSession: "bob@rig",
+        toSession: "carol@rig",
+      });
+      expect(sends).toHaveLength(1);
+      expect(sends[0]!.session).toBe("carol@rig");
+      const fresh = nudgingRepo.getById(result.created.qitemId)!;
+      expect(fresh.lastNudgeResult).toBe("verified");
+    });
+
+    it("attachTransport() works after construction (post-hoc wiring path)", async () => {
+      const repoNoTransport = new QueueRepository(db, bus);
+      const sends: Array<{ session: string }> = [];
+      const stubTransport = {
+        send: async (s: string) => { sends.push({ session: s }); return { ok: true, verified: true }; },
+      };
+      // First create: no transport, no nudge
+      await repoNoTransport.create({
+        sourceSession: "alice@rig",
+        destinationSession: "bob@rig",
+        body: "before",
+      });
+      expect(sends).toHaveLength(0);
+      // Attach + create again
+      repoNoTransport.attachTransport(stubTransport);
+      await repoNoTransport.create({
+        sourceSession: "alice@rig",
+        destinationSession: "bob@rig",
+        body: "after",
+      });
+      expect(sends).toHaveLength(1);
+    });
+  });
+
+  describe("R1 handoff-and-complete", () => {
+    it("closes source as state=done with closure_reason=handed_off_to (terminal) and creates new qitem", async () => {
+      const original = await repo.create({
+        sourceSession: "alice@rig",
+        destinationSession: "bob@rig",
+        body: "review then route",
+      });
+      const result = await repo.handoffAndComplete({
+        qitemId: original.qitemId,
+        fromSession: "bob@rig",
+        toSession: "carol@rig",
+        body: "carol's follow-on",
+      });
+      expect(result.closed.state).toBe("done"); // not "handed-off"
+      expect(result.closed.closureReason).toBe("handed_off_to");
+      expect(result.closed.handedOffTo).toBe("carol@rig");
+      expect(result.created.state).toBe("pending");
+      expect(result.created.handedOffFrom).toBe(original.qitemId);
+      expect(result.created.destinationSession).toBe("carol@rig");
+      expect(result.created.body).toBe("carol's follow-on");
+      expect(result.created.chainOfRecord).toEqual([original.qitemId]);
+    });
+
+    it("refuses on already-terminal qitem", async () => {
+      const item = await repo.create({
+        sourceSession: "alice@rig",
+        destinationSession: "bob@rig",
+        body: "x",
+      });
+      repo.claim({ qitemId: item.qitemId, destinationSession: "bob@rig" });
+      repo.update({
+        qitemId: item.qitemId,
+        actorSession: "bob@rig",
+        state: "done",
+        closureReason: "no-follow-on",
+      });
+      await expect(
+        repo.handoffAndComplete({
+          qitemId: item.qitemId,
+          fromSession: "bob@rig",
+          toSession: "carol@rig",
+        })
+      ).rejects.toThrow(/terminal/);
+    });
+
+    it("respects validateRig (rejects unknown destination rig)", async () => {
+      const strictRepo = new QueueRepository(db, bus, {
+        validateRig: (s) => s.endsWith("@known-rig"),
+      });
+      const original = await strictRepo.create({
+        sourceSession: "alice@known-rig",
+        destinationSession: "bob@known-rig",
+        body: "x",
+      });
+      await expect(
+        strictRepo.handoffAndComplete({
+          qitemId: original.qitemId,
+          fromSession: "bob@known-rig",
+          toSession: "carol@phantom-rig",
+        })
+      ).rejects.toThrow(/unknown rig/);
+    });
+  });
+
+  describe("R1 whoami", () => {
+    it("returns counts + recent active qitems for the destination session", async () => {
+      const a = await repo.create({ sourceSession: "x@r", destinationSession: "bob@r", body: "1" });
+      await repo.create({ sourceSession: "x@r", destinationSession: "bob@r", body: "2" });
+      await repo.create({ sourceSession: "x@r", destinationSession: "carol@r", body: "3" });
+      repo.claim({ qitemId: a.qitemId, destinationSession: "bob@r" });
+      const whoami = repo.whoami("bob@r");
+      expect(whoami.session).toBe("bob@r");
+      expect(whoami.asDestination.pending).toBe(1);
+      expect(whoami.asDestination.inProgress).toBe(1);
+      expect(whoami.asDestination.recent).toHaveLength(2);
+    });
+
+    it("recent excludes terminal-state qitems", async () => {
+      const a = await repo.create({ sourceSession: "x@r", destinationSession: "bob@r", body: "1" });
+      repo.claim({ qitemId: a.qitemId, destinationSession: "bob@r" });
+      repo.update({ qitemId: a.qitemId, actorSession: "bob@r", state: "done", closureReason: "no-follow-on" });
+      const whoami = repo.whoami("bob@r");
+      expect(whoami.asDestination.pending).toBe(0);
+      expect(whoami.asDestination.inProgress).toBe(0);
+      expect(whoami.asDestination.recent).toHaveLength(0);
+    });
+
+    it("asSource.total counts all source-side qitems regardless of state", async () => {
+      await repo.create({ sourceSession: "alice@r", destinationSession: "bob@r", body: "1" });
+      await repo.create({ sourceSession: "alice@r", destinationSession: "carol@r", body: "2" });
+      const item = await repo.create({ sourceSession: "alice@r", destinationSession: "dan@r", body: "3" });
+      repo.claim({ qitemId: item.qitemId, destinationSession: "dan@r" });
+      repo.update({ qitemId: item.qitemId, actorSession: "dan@r", state: "done", closureReason: "no-follow-on" });
+      const whoami = repo.whoami("alice@r");
+      expect(whoami.asSource.total).toBe(3);
+    });
+  });
 });

packages/daemon/test/queue-routes.test.ts

@@ -235,4 +235,217 @@ describe("queue routes", () => {
     const data = (await res.json()) as unknown[];
     expect(data).toHaveLength(1);
   });
+
+  // ---- PL-004 Phase A revision (R1) route tests ----
+
+  describe("R1 cross-rig validation rejection", () => {
+    let strictDb: Database.Database;
+    let strictBus: EventBus;
+    let strictRepo: QueueRepository;
+    let strictApp: Hono;
+
+    beforeEach(() => {
+      strictDb = createDb();
+      migrate(strictDb, [
+        coreSchema,
+        eventsSchema,
+        queueItemsSchema,
+        queueTransitionsSchema,
+        inboxEntriesSchema,
+        outboxEntriesSchema,
+      ]);
+      strictBus = new EventBus(strictDb);
+      strictRepo = new QueueRepository(strictDb, strictBus, {
+        // Topology-backed validator stub: only `@known-rig` is recognized.
+        validateRig: (s) => /^[^@]+@known-rig$/.test(s),
+      });
+      const strictInbox = new InboxHandler(strictDb, strictBus, strictRepo);
+      const strictOutbox = new OutboxHandler(strictDb);
+      strictApp = buildApp({
+        eventBus: strictBus,
+        queueRepo: strictRepo,
+        inboxHandler: strictInbox,
+        outboxHandler: strictOutbox,
+      });
+    });
+
+    afterEach(() => strictDb.close());
+
+    it("POST /api/queue/create rejects unknown rig with 400 + structured error", async () => {
+      const res = await strictApp.request("/api/queue/create", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          sourceSession: "alice@known-rig",
+          destinationSession: "bob@phantom-rig",
+          body: "x",
+        }),
+      });
+      expect(res.status).toBe(400);
+      const body = (await res.json()) as { error: string; message: string };
+      expect(body.error).toBe("unknown_destination_rig");
+      expect(body.message).toMatch(/phantom-rig/);
+    });
+
+    it("POST /api/queue/create accepts known rig with 201", async () => {
+      const res = await strictApp.request("/api/queue/create", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          sourceSession: "alice@known-rig",
+          destinationSession: "bob@known-rig",
+          body: "ok",
+        }),
+      });
+      expect(res.status).toBe(201);
+    });
+
+    it("POST /api/queue/:id/handoff rejects unknown destination rig", async () => {
+      const created = await strictApp.request("/api/queue/create", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          sourceSession: "alice@known-rig",
+          destinationSession: "bob@known-rig",
+          body: "x",
+        }),
+      });
+      const item = (await created.json()) as { qitemId: string };
+      const res = await strictApp.request(`/api/queue/${item.qitemId}/handoff`, {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          fromSession: "bob@known-rig",
+          toSession: "carol@phantom-rig",
+        }),
+      });
+      expect(res.status).toBe(400);
+      const body = (await res.json()) as { error: string };
+      expect(body.error).toBe("unknown_destination_rig");
+    });
+
+    it("POST /api/queue/:id/handoff-and-complete rejects unknown destination rig", async () => {
+      const created = await strictApp.request("/api/queue/create", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          sourceSession: "alice@known-rig",
+          destinationSession: "bob@known-rig",
+          body: "x",
+        }),
+      });
+      const item = (await created.json()) as { qitemId: string };
+      const res = await strictApp.request(`/api/queue/${item.qitemId}/handoff-and-complete`, {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          fromSession: "bob@known-rig",
+          toSession: "carol@phantom-rig",
+        }),
+      });
+      expect(res.status).toBe(400);
+      const body = (await res.json()) as { error: string };
+      expect(body.error).toBe("unknown_destination_rig");
+    });
+  });
+
+  describe("R1 handoff-and-complete route", () => {
+    it("POST /api/queue/:id/handoff-and-complete closes source as done + creates new", async () => {
+      const created = await app.request("/api/queue/create", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ sourceSession: "alice@r", destinationSession: "bob@r", body: "x" }),
+      });
+      const item = (await created.json()) as { qitemId: string };
+      const res = await app.request(`/api/queue/${item.qitemId}/handoff-and-complete`, {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          fromSession: "bob@r",
+          toSession: "carol@r",
+          body: "carol's piece",
+        }),
+      });
+      expect(res.status).toBe(201);
+      const result = (await res.json()) as {
+        closed: { state: string; closureReason: string; handedOffTo: string };
+        created: { state: string; handedOffFrom: string; destinationSession: string; body: string };
+      };
+      expect(result.closed.state).toBe("done");
+      expect(result.closed.closureReason).toBe("handed_off_to");
+      expect(result.closed.handedOffTo).toBe("carol@r");
+      expect(result.created.state).toBe("pending");
+      expect(result.created.handedOffFrom).toBe(item.qitemId);
+      expect(result.created.body).toBe("carol's piece");
+    });
+
+    it("POST /api/queue/:id/handoff-and-complete returns 400 on missing fromSession or toSession", async () => {
+      const res = await app.request("/api/queue/some-id/handoff-and-complete", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ toSession: "carol@r" }),
+      });
+      expect(res.status).toBe(400);
+      const body = (await res.json()) as { error: string };
+      expect(body.error).toMatch(/fromSession/);
+    });
+  });
+
+  describe("R1 whoami route", () => {
+    it("GET /api/queue/whoami returns counts + recent for the session", async () => {
+      // Seed: 2 pending + 1 in-progress for bob; 1 unrelated for carol.
+      const a = await app.request("/api/queue/create", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ sourceSession: "alice@r", destinationSession: "bob@r", body: "1" }),
+      });
+      const itemA = (await a.json()) as { qitemId: string };
+      await app.request("/api/queue/create", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ sourceSession: "alice@r", destinationSession: "bob@r", body: "2" }),
+      });
+      await app.request("/api/queue/create", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ sourceSession: "alice@r", destinationSession: "carol@r", body: "3" }),
+      });
+      await app.request(`/api/queue/${itemA.qitemId}/claim`, {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ destinationSession: "bob@r" }),
+      });
+
+      const res = await app.request("/api/queue/whoami?session=bob@r");
+      expect(res.status).toBe(200);
+      const body = (await res.json()) as {
+        session: string;
+        asDestination: { pending: number; inProgress: number; recent: unknown[] };
+        asSource: { total: number };
+      };
+      expect(body.session).toBe("bob@r");
+      expect(body.asDestination.pending).toBe(1);
+      expect(body.asDestination.inProgress).toBe(1);
+      expect(body.asDestination.recent).toHaveLength(2);
+      expect(body.asSource.total).toBe(0);
+    });
+
+    it("GET /api/queue/whoami returns 400 without session query param", async () => {
+      const res = await app.request("/api/queue/whoami");
+      expect(res.status).toBe(400);
+    });
+  });
+
+  describe("R1 SSE alias", () => {
+    it("/api/queue/sse is mounted alongside /api/queue/watch (same handler)", async () => {
+      // Both routes resolve to handlers; HEAD-style probe by aborting after a short ping.
+      // Since SSE responses don't return a synchronous body, we just confirm the
+      // route is mounted (not 404). The /watch path is the legacy alias, /sse is canonical.
+      const watchRes = await app.request("/api/queue/watch", { method: "HEAD" });
+      const sseRes = await app.request("/api/queue/sse", { method: "HEAD" });
+      // Both should be the same status (Hono returns 200 for SSE before stream starts; HEAD is method-allowed by default for GET routes).
+      expect(watchRes.status).toBe(sseRes.status);
+      expect(watchRes.status).toBeLessThan(500);
+    });
+  });
 });

packages/daemon/test/stream-routes.test.ts

@@ -95,4 +95,12 @@ describe("stream routes", () => {
     const data = (await list.json()) as unknown[];
     expect(data).toHaveLength(0);
   });
+
+  // ---- PL-004 Phase A revision (R1): SSE alias ----
+  it("/api/stream/sse is mounted alongside /api/stream/watch (same handler)", async () => {
+    const watchRes = await app.request("/api/stream/watch", { method: "HEAD" });
+    const sseRes = await app.request("/api/stream/sse", { method: "HEAD" });
+    expect(watchRes.status).toBe(sseRes.status);
+    expect(watchRes.status).toBeLessThan(500);
+  });
 });