fix(daemon): PL-004 Phase A R1 — SSE route-order shadowing fix

Closes QA verdict revise-required from delivery-orch-lead per qitem-20260503024929-ea365d2b + handoff packet substrate/rigs/openrig-velocity/state/delivery-orch/pl004-phase-a-r1-sse-route-fix-driver-handoff-20260503.md.

Foundation: openrig fd4fb30 (PL-004 Phase A R1). Branch: phase-a-coordination-primitive-system (worktree-scoped; main NOT touched).

ROOT CAUSE:
The R1 commit fd4fb30 mounted GET /watch and GET /sse routes AFTER the bare-param catchalls (GET /:qitemId on queue.ts:278; GET /:streamItemId on stream.ts:69). Hono dispatches GET /api/queue/sse to the /:qitemId handler (with qitemId="sse"), which calls QueueRepository.getById("sse") → null → returns 404 with `{"error":"qitem_not_found"}`. Same problem on stream side: GET /api/stream/sse → /:streamItemId handler → 404 `{"error":"stream item not found"}`.

The QA proof caught this; my R1 test was inadequate (HEAD comparison; both /watch and /sse returned the same 404 status, so the "they match" assertion held while both were broken).

FIX:
1. Moved sseHandler definition + `app.get("/watch", sseHandler)` + `app.get("/sse", sseHandler)` BEFORE the bare /:qitemId GET in routes/queue.ts (now mounted alongside /whoami, /list, /overdue — matching the existing literal-first convention with explicit comment).
2. Same fix in routes/stream.ts: SSE routes now mounted BEFORE GET /:streamItemId.
3. Removed the duplicate handler definition + mounts at end of both files.

NEW LIVE-GET TESTS (replacing inadequate HEAD comparison):
queue-routes.test.ts (3 cases under "R1 SSE route — live GET reaches the SSE handler"):
- GET /api/queue/sse → 200 + content-type: text/event-stream (handler reached)
- GET /api/queue/watch → 200 + content-type: text/event-stream (handler reached)
- GET /api/queue/sse does NOT return qitem_not_found (route-order regression guard)

stream-routes.test.ts (3 cases):
- GET /api/stream/sse → 200 + content-type: text/event-stream
- GET /api/stream/watch → 200 + content-type: text/event-stream
- GET /api/stream/sse does NOT return stream-item-not-found (regression guard)

Each test cancels the response body (`res.body?.cancel()`) so the long-lived SSE stream releases cleanly. The content-type assertion proves the SSE handler is reached (vs the JSON 404 the catchall would emit).

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

Focused (under explicit temp-rooted OPENRIG_HOME + OPENRIG_DB):
- `OPENRIG_HOME=/tmp/pl004-r1-sse-final-d OPENRIG_DB=/tmp/pl004-r1-sse-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` → 93/93 PASS (89 R1 baseline + 4 new SSE route-order regression tests).
- daemon tsc --noEmit + npm run build: exit 0.

Compiled-output runtime live-GET check (the canonical anti-regression proof):
```
node -e '
  ... boot Hono app + queueRoutes() + streamRoutes() from dist/ ...
  for (const path of ["/api/queue/sse", "/api/queue/watch", "/api/stream/sse", "/api/stream/watch"]) {
    const res = await app.request(path);
    console.log(path, "→ status:", res.status, "ctype:", res.headers.get("content-type"));
    await res.body?.cancel();
  }
'
```
Output (all 4 paths):
- /api/queue/sse → status: 200 ctype: text/event-stream
- /api/queue/watch → status: 200 ctype: text/event-stream
- /api/stream/sse → status: 200 ctype: text/event-stream
- /api/stream/watch → status: 200 ctype: text/event-stream

Honest per-package baselines (NO env override):
- daemon: 2334/2334 PASS / 0 FAIL (R1 baseline 2330 + 4 new).
- 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 baseline.

DB ISOLATION:
All test runs used explicit temp-rooted OPENRIG_HOME + OPENRIG_DB under /tmp/pl004-r1-sse-*. No daemon launched. No production DB opened.

BOUNDARIES PRESERVED:
- Worktree-only changes; main checkout `/Users/wrandom/code/projects/openrig` HEAD unchanged at ab822eef throughout. Verified pre and post.
- 4 files changed (all in packages/daemon/{src,test}/routes-and-tests). No CLI changes; no domain changes.
- No host daemon mutation. No host credential mutation. No live rig topology mutation.
- No standing Codex auto-review or approval-heavy loop.
- Normal posture approvals_reviewer=user honored throughout.

LESSON LEARNED (filed inline; carry into future PR review discipline):
HEAD-method tests against SSE/streaming routes can pass while the underlying route-resolution is broken. Test discipline for streaming routes: GET the route, assert response.status + content-type header, then cancel the body to release the stream. Hono's app.request returns the Response synchronously even when the body is a stream; checking headers without consuming body avoids hangs.

HANDOFF RECOMMENDATION: ready-for-qa-proof.

QA's sustained proof loop now has a green-for-real SSE surface; can resume the cross-loop dogfood + Velocity-pair cross-rig validation + reissued VM proof per the original R1 dispatch.

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

packages/daemon/src/routes/queue.ts

@@ -265,6 +265,43 @@ export function queueRoutes(): Hono {
     return c.json(items);
   });
 
+  // ---- SSE watch over coordination events ----
+  // MUST precede /:qitemId so the literal `watch` and `sse` paths win
+  // over the bare-param route (otherwise GET /api/queue/sse resolves as
+  // /:qitemId with qitemId="sse" and returns 404 qitem_not_found).
+  // Mounted at both /watch (legacy alias) and /sse (Phase A contract per IMPL).
+  // Same handler; either path emits the identical event stream.
+  const sseHandler = (c: Parameters<typeof streamSSE>[0]) => {
+    const eventBus = getEventBus(c);
+    return streamSSE(c, async (stream) => {
+      const unsubscribe = eventBus.subscribe((event) => {
+        if (
+          event.type !== "queue.created" &&
+          event.type !== "queue.handed_off" &&
+          event.type !== "queue.claimed" &&
+          event.type !== "queue.unclaimed" &&
+          event.type !== "qitem.fallback_routed" &&
+          event.type !== "qitem.closure_overdue" &&
+          event.type !== "inbox.absorbed" &&
+          event.type !== "inbox.denied"
+        ) return;
+        const sse = { id: String(event.seq), data: JSON.stringify(event) };
+        stream.writeSSE(sse).catch(() => {});
+      });
+
+      try {
+        await new Promise<void>((resolve) => {
+          stream.onAbort(() => resolve());
+        });
+      } finally {
+        unsubscribe();
+      }
+    });
+  };
+
+  app.get("/watch", sseHandler);
+  app.get("/sse", sseHandler);
+
   // GET /:qitemId/transitions — registered before /:qitemId so the literal
   // suffix wins over the bare param route.
   app.get("/:qitemId/transitions", (c) => {
@@ -391,41 +428,5 @@ export function queueRoutes(): Hono {
     return c.json(getOutbox(c).listForSender(senderSession, limit));
   });
 
-  // ---- 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.
-
-  const sseHandler = (c: Parameters<typeof streamSSE>[0]) => {
-    const eventBus = getEventBus(c);
-
-    return streamSSE(c, async (stream) => {
-      const unsubscribe = eventBus.subscribe((event) => {
-        if (
-          event.type !== "queue.created" &&
-          event.type !== "queue.handed_off" &&
-          event.type !== "queue.claimed" &&
-          event.type !== "queue.unclaimed" &&
-          event.type !== "qitem.fallback_routed" &&
-          event.type !== "qitem.closure_overdue" &&
-          event.type !== "inbox.absorbed" &&
-          event.type !== "inbox.denied"
-        ) return;
-        const sse = { id: String(event.seq), data: JSON.stringify(event) };
-        stream.writeSSE(sse).catch(() => {});
-      });
-
-      try {
-        await new Promise<void>((resolve) => {
-          stream.onAbort(() => resolve());
-        });
-      } finally {
-        unsubscribe();
-      }
-    });
-  };
-
-  app.get("/watch", sseHandler);
-  app.get("/sse", sseHandler);
-
   return app;
 }

packages/daemon/src/routes/stream.ts

@@ -65,25 +65,10 @@ export function streamRoutes(): Hono {
     return c.json(items);
   });
 
-  // GET /:streamItemId — fetch one
-  app.get("/:streamItemId", (c) => {
-    const id = c.req.param("streamItemId");
-    const store = getStore(c);
-    const item = store.getById(id);
-    if (!item) return c.json({ error: "stream item not found" }, 404);
-    return c.json(item);
-  });
-
-  // POST /:streamItemId/archive
-  app.post("/:streamItemId/archive", (c) => {
-    const id = c.req.param("streamItemId");
-    const store = getStore(c);
-    const ok = store.archive(id);
-    if (!ok) return c.json({ error: "stream item not found or already archived" }, 404);
-    return c.json({ ok: true });
-  });
-
   // GET /watch — SSE for new stream.emitted events.
+  // MUST precede /:streamItemId so the literal `watch` and `sse` paths
+  // win over the bare-param route (otherwise GET /api/stream/sse resolves
+  // as /:streamItemId with id="sse" and returns 404 stream-item-not-found).
   // 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.
@@ -132,5 +117,23 @@ export function streamRoutes(): Hono {
   app.get("/watch", sseHandler);
   app.get("/sse", sseHandler);
 
+  // GET /:streamItemId — fetch one
+  app.get("/:streamItemId", (c) => {
+    const id = c.req.param("streamItemId");
+    const store = getStore(c);
+    const item = store.getById(id);
+    if (!item) return c.json({ error: "stream item not found" }, 404);
+    return c.json(item);
+  });
+
+  // POST /:streamItemId/archive
+  app.post("/:streamItemId/archive", (c) => {
+    const id = c.req.param("streamItemId");
+    const store = getStore(c);
+    const ok = store.archive(id);
+    if (!ok) return c.json({ error: "stream item not found or already archived" }, 404);
+    return c.json({ ok: true });
+  });
+
   return app;
 }

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

@@ -436,16 +436,45 @@ describe("queue routes", () => {
     });
   });
 
-  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);
+  describe("R1 SSE route — live GET reaches the SSE handler (not shadowed by /:qitemId)", () => {
+    // Live GET tests per QA finding: HEAD comparison was inadequate because
+    // dynamic route shadowing (/:qitemId catching `sse` and `watch` as ids)
+    // returns 404 with `qitem_not_found` instead of the SSE handler.
+    // Real GET that asserts content-type: text/event-stream proves the
+    // SSE handler is reached. We cancel the response body to release the
+    // long-lived stream.
+
+    it("GET /api/queue/sse returns 200 + content-type: text/event-stream (handler reached)", async () => {
+      const res = await app.request("/api/queue/sse");
+      try {
+        expect(res.status).toBe(200);
+        expect(res.headers.get("content-type") ?? "").toContain("text/event-stream");
+      } finally {
+        await res.body?.cancel();
+      }
+    });
+
+    it("GET /api/queue/watch returns 200 + content-type: text/event-stream (handler reached)", async () => {
+      const res = await app.request("/api/queue/watch");
+      try {
+        expect(res.status).toBe(200);
+        expect(res.headers.get("content-type") ?? "").toContain("text/event-stream");
+      } finally {
+        await res.body?.cancel();
+      }
+    });
+
+    it("GET /api/queue/sse does NOT return qitem_not_found (route-order regression guard)", async () => {
+      const res = await app.request("/api/queue/sse");
+      try {
+        // If /:qitemId catches `sse` as an id, it returns 404 JSON with
+        // {"error":"qitem_not_found"}. This must never happen.
+        expect(res.status).not.toBe(404);
+        const ct = res.headers.get("content-type") ?? "";
+        expect(ct).not.toContain("application/json");
+      } finally {
+        await res.body?.cancel();
+      }
     });
   });
 });

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

@@ -96,11 +96,40 @@ describe("stream routes", () => {
     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);
+  // ---- PL-004 Phase A revision (R1): SSE route — live GET reaches handler ----
+  // Per QA finding: HEAD comparison was inadequate. Dynamic route shadowing
+  // (/:streamItemId catching `sse` and `watch` as ids) returns 404 with
+  // `stream item not found` instead of the SSE handler. Live GET asserting
+  // content-type: text/event-stream proves the handler is reached.
+
+  it("GET /api/stream/sse returns 200 + content-type: text/event-stream (handler reached)", async () => {
+    const res = await app.request("/api/stream/sse");
+    try {
+      expect(res.status).toBe(200);
+      expect(res.headers.get("content-type") ?? "").toContain("text/event-stream");
+    } finally {
+      await res.body?.cancel();
+    }
+  });
+
+  it("GET /api/stream/watch returns 200 + content-type: text/event-stream (handler reached)", async () => {
+    const res = await app.request("/api/stream/watch");
+    try {
+      expect(res.status).toBe(200);
+      expect(res.headers.get("content-type") ?? "").toContain("text/event-stream");
+    } finally {
+      await res.body?.cancel();
+    }
+  });
+
+  it("GET /api/stream/sse does NOT return stream-item-not-found (route-order regression guard)", async () => {
+    const res = await app.request("/api/stream/sse");
+    try {
+      expect(res.status).not.toBe(404);
+      const ct = res.headers.get("content-type") ?? "";
+      expect(ct).not.toContain("application/json");
+    } finally {
+      await res.body?.cancel();
+    }
   });
 });