fix(daemon,cli): PL-004 Phase B R1 — L1→L2 FK + view.changed event bridge + notes 3+4

Closes guard BLOCK on Phase B R0 at openrig 4ac0d20 per qitem-20260503043604-21709333 + guard artifact pl004-phase-b-4ac0d20-guard-review-block-20260503.md.

==== BLOCKER 1: L1→L2 referential integrity + existence check ====

PRD § L2 (product-specs/coordination-primitive-system/README.md:113,118,136) defines stream_item_id as FK to L1. R0 implementation had UNIQUE-only without FK; tests classified arbitrary nonexistent IDs.

Fix:
- Migration 028: added FOREIGN KEY clause `stream_item_id TEXT NOT NULL UNIQUE REFERENCES stream_items(stream_item_id)`. PRAGMA foreign_keys=ON in connection.ts is the defense-in-depth safety net.
- project-classifier.ts: added explicit existence check (`SELECT 1 FROM stream_items WHERE stream_item_id = ?`) BEFORE INSERT, throwing structured `ProjectClassifierError("unknown_stream_item", ...)` with meta.streamItemId. Pre-check renders a clean error path instead of opaque SQLite FK violation message.
- routes/projects.ts: maps `unknown_stream_item` to HTTP 400.
- packages/cli/src/commands/project.ts: surfaces 400 error correctly via existing exit-1 pattern.

Test updates (tests now migrate streamItemsSchema + seed real stream_items rows):
- project-classifier.test.ts: imports streamItemsSchema + StreamStore. Helper `seedStreamItem()` emits a real stream item via Phase A's StreamStore. Updated all 7 existing tests to call seedStreamItem first. Added 2 NEW tests: (a) classify of nonexistent stream_item_id throws unknown_stream_item; (b) FK constraint blocks direct INSERT bypass (defense-in-depth).
- projects-routes.test.ts: imports streamItemsSchema + StreamStore. Helper `seedStreamItem()` for route tests. Updated all 3 existing project tests to seed first. Added 1 NEW test: POST /api/projects/project with nonexistent streamItemId returns 400 unknown_stream_item with structured message + streamItemId in meta.
- packages/cli/test/project.test.ts: added 1 NEW test for unknown_stream_item 400 surfacing + non-zero exit + error message rendering.

==== BLOCKER 2: view.changed event bridge ====

PRD § L5 acceptance criterion + slice IMPL § Guard Checkpoint Focus item 8: SSE consumers on /api/views/:name/sse must observe view.changed when underlying state mutates. R0 implementation only forwarded events that already had type view.changed, but no production path emitted view.changed from queue/project mutations.

Fix:
- NEW packages/daemon/src/domain/view-event-bridge.ts (96 lines): subscribes to event-bus and emits view.changed for affected built-in views via ViewProjector.notifyViewChanged. EVENT_TO_VIEWS mapping covers all 9 coordination event types (queue.created → recently-active, founder, pod-load, activity; queue.handed_off / queue.claimed / queue.unclaimed / qitem.fallback_routed / inbox.absorbed → recently-active, pod-load, activity; qitem.closure_overdue → recently-active, escalations, activity; inbox.denied + project.classified → activity). classifier.lease_* events deliberately NOT bridged (lease lifecycle is project SSE, not view SSE). view.changed itself NOT bridged (no feedback loop).
- Validation at wire time: bridge throws if EVENT_TO_VIEWS references unknown built-in view names (fail-fast guard against future built-in renames/removals).
- Returns ViewEventBridgeStop unsubscribe function for tests + graceful daemon shutdown.
- Best-effort isolation: bridge errors caught and dropped silently so they cannot unwind the underlying state mutation that triggered them.

Wired in startup.ts: viewProjectorInstance constructed early; wireViewEventBridge(eventBus, viewProjectorInstance) called immediately after.

NEW packages/daemon/test/view-event-bridge.test.ts (9 cases):
- queue.created → 4 view.changed (recently-active, founder, pod-load, activity)
- queue.handed_off → 3 view.changed
- queue.claimed → 3 view.changed
- inbox.absorbed → 3 view.changed
- inbox.denied → 1 view.changed (activity only)
- project.classified → 1 view.changed (activity only)
- classifier.lease_acquired → 0 view.changed (lease is project SSE not view SSE)
- view.changed → 0 view.changed (feedback-loop prevention)
- unsubscribe stops the bridge cleanly

NEW route-level test in views-routes.test.ts: "R1 BLOCKER 2: queue mutation triggers view.changed visible to SSE consumer". Wires the bridge in test, subscribes to /api/views/recently-active/sse, mutates queue state, reads SSE body via reader.read() with deadline, asserts view.changed event observed in stream payload before timeout. Cancels via reader.cancel() to release the stream lock cleanly.

==== NOTE 3: lease acquire stale/dead expiry opt-in ====

R0 lease acquire returned 409 lease_held even when holder was dead (only operator-verb reclaim or evaluateDeadness call could clear it). Per guard note, the route should cover stale/dead expiry behavior or scope explicitly.

Fix:
- routes/projects.ts POST /lease/acquire: added optional `evaluateDeadnessFirst: true` body parameter. When set, the route calls ClassifierLeaseManager.evaluateDeadness() before acquire, which clears stale TTL-expired or dead-holder leases. Default OFF so existing behavior is preserved; opt-in is explicit.
- packages/cli/src/commands/project.ts lease-acquire: added --evaluate-deadness-first flag passing through to the route body.

==== NOTE 4: pod-load v0 interpretation documented ====

PRD § L5 names pod-load as "per-pod queue-item counts" but Phase A's queue_items table lacks pod metadata. R0 implementation grouped by destination_session (per-seat). Per guard note, document v0 interpretation rather than calling PRD semantics unchanged.

Fix:
- view-projector.ts pod-load case: added inline doc explaining v0 interpretation is per-seat (destination_session) because Phase A lacks pod metadata; future Phase D workflow runtime can upgrade to GROUP BY pod_id with destination_session sub-grouping. Documented v0 behavior preserves the view name but reduces the grouping key.

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

Focused (under explicit temp-rooted OPENRIG_HOME + OPENRIG_DB):
- daemon: OPENRIG_HOME=/tmp/pl004-phase-b-r1-final-d OPENRIG_DB=/tmp/pl004-phase-b-r1-final-d/openrig.sqlite npx vitest run [6 Phase B daemon test files including new view-event-bridge.test.ts] → 70/70 PASS (R0 baseline 57 + 4 R1 BLOCKER 1 nonexistent-stream cases + 9 R1 BLOCKER 2 bridge cases = 70).
- cli: OPENRIG_HOME=/tmp/pl004-phase-b-r1-final-c OPENRIG_DB=/tmp/pl004-phase-b-r1-final-c/openrig.sqlite npx vitest run test/project.test.ts test/view.test.ts → 13/13 PASS (R0 baseline 12 + 1 R1 unknown_stream_item case).
- daemon tsc + build: exit 0; cli tsc + build: exit 0.

Compiled-output runtime checks:
- node import dist/domain/view-event-bridge.js → wireViewEventBridge: function ✓

Full per-package suites (NO env override; honest baseline):
- daemon: 2404/2404 PASS / 0 FAIL (Phase B R0 baseline 2391 + 13 new R1 = 2404).
- cli: 770/774 PASS / 4 FAIL named pre-existing flakes. Phase B R0 baseline 769/773; +1 new = 770/774. NO new regressions.

==== BOUNDARIES PRESERVED ====

- Worktree-only changes; main HEAD unchanged at 194a1ebb throughout (verified pre and post).
- 12 files changed (10 modified + 2 new), all in packages/daemon/{src,test}/* and packages/cli/{src,test}/*.
- Phase A surfaces NOT touched. Existing 22 + Phase A's 5 migrations NOT modified. Phase B R0's migrations 028-030 modified ONLY in 028 (added FK clause; safe additive change since the column already existed).
- Lifecycle v2 surfaces NOT touched. Phase C/D/E surfaces NOT touched.
- Velocity prior-art .mjs UNCHANGED.
- POC rigx project / view-proto NOT touched (preserved as-is for dogfood window).
- 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.

==== HANDOFF RECOMMENDATION: ready-for-guard-review ====

Both blockers fixed; both notes addressed. Driver-side focused verification passed (70 + 13 = 83 focused tests under explicit temp DB isolation + compiled-output runtime check + per-package baselines clean).

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

packages/cli/src/commands/project.ts

@@ -53,12 +53,17 @@ export function projectCommand(depsOverride?: ProjectDeps): Command {
     .command("lease-acquire")
     .description("Acquire the active classifier lease for the caller")
     .requiredOption("--session <session>", "Classifier session name")
+    .option(
+      "--evaluate-deadness-first",
+      "Before acquire, call evaluateDeadness to clear any stale TTL-expired or dead-holder lease (per PRD § L2 deadness-detection)",
+    )
     .option("--json", "JSON output for agents")
-    .action(async (opts: { session: string; json?: boolean }) => {
+    .action(async (opts: { session: string; evaluateDeadnessFirst?: boolean; json?: boolean }) => {
       const deps = getDeps();
       await withClient(deps, async (client) => {
         const res = await client.post<unknown>("/api/projects/lease/acquire", {
           classifierSession: opts.session,
+          evaluateDeadnessFirst: opts.evaluateDeadnessFirst,
         });
         printResult(opts.json ?? false, res.data, res.status);
       });

packages/cli/test/project.test.ts

@@ -99,6 +99,32 @@ describe("rig project CLI (PL-004 Phase B)", () => {
     expect(body.classificationDestination).toBe("planning@rig");
   });
 
+  it("R1 BLOCKER 1: classify with unknown_stream_item 400 surfaces error + non-zero exit", async () => {
+    const { deps } = makeDeps({
+      routes: {
+        "POST /api/projects/project": {
+          status: 400,
+          data: {
+            error: "unknown_stream_item",
+            message: "stream_item_id stream-nonexistent does not exist in stream_items",
+            streamItemId: "stream-nonexistent",
+          },
+        },
+      },
+    });
+    const program = createProgram({ projectDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "project", "classify", "stream-nonexistent",
+      "--session", "alice@rig",
+      "--json",
+    ]);
+    expect(process.exitCode).toBe(1);
+    const out = logs.join("\n");
+    expect(out).toContain("unknown_stream_item");
+    expect(out).toContain("does not exist");
+  });
+
   it("classify with idempotency_violation 409 surfaces error + non-zero exit", async () => {
     const { deps } = makeDeps({
       routes: {

packages/daemon/src/db/migrations/028_project_classifications.ts

@@ -3,21 +3,32 @@ import type { Migration } from "../migrate.js";
 /**
  * Project classifications (PL-004 Phase B; L2 Project / Classifier).
  *
- * Per PRD § L2 + slice IMPL § Write Set: agent-backed classification of
- * stream items. Daemon owns idempotency (UNIQUE on stream_item_id) and
- * lease validation (delegated to classifier-lease-manager). Classification
- * fields are agent judgment — daemon does NOT enforce taxonomies on them.
+ * Per PRD § L2 (`product-specs/coordination-primitive-system/README.md:113,118,136`)
+ * + slice IMPL § Write Set:118 + R1 guard (BLOCKER 1): agent-backed
+ * classification of stream items. Daemon owns:
+ *
+ * - Idempotency on stream_item_id (UNIQUE).
+ * - Lease validation (delegated to classifier-lease-manager).
+ * - L1→L2 referential integrity (FK to stream_items.stream_item_id).
+ *
+ * Classification fields are agent judgment — daemon does NOT enforce
+ * taxonomies on them.
  *
  * Idempotency contract: re-projection of the same stream_item_id MUST
  * fail with structured 409. The first row wins; the second attempt's
  * classification is rejected.
+ *
+ * Existence contract (R1): classification of a nonexistent stream_item_id
+ * MUST fail with structured 400-class `unknown_stream_item` error. Domain
+ * layer pre-checks for clean error rendering; FK constraint is a
+ * defense-in-depth safety net (PRAGMA foreign_keys = ON in connection.ts).
  */
 export const projectClassificationsSchema: Migration = {
   name: "028_project_classifications.sql",
   sql: `
     CREATE TABLE IF NOT EXISTS project_classifications (
       project_id TEXT PRIMARY KEY,
-      stream_item_id TEXT NOT NULL UNIQUE,
+      stream_item_id TEXT NOT NULL UNIQUE REFERENCES stream_items(stream_item_id),
       classification_type TEXT,
       classification_urgency TEXT,
       classification_maturity TEXT,

packages/daemon/src/domain/project-classifier.ts

@@ -104,6 +104,22 @@ export class ProjectClassifier {
     // lease_held by other session, or lease_expired.
     this.leaseManager.requireActiveHolder(input.classifierSession);
 
+    // R1 fix (BLOCKER 1): existence check on stream_item_id. The L1→L2
+    // FK in migration 028 is a defense-in-depth safety net, but the SQLite
+    // FK violation would surface as an opaque error string. Pre-checking
+    // here gives a clean structured error (`unknown_stream_item`) that
+    // routes can map to a 400-class status.
+    const streamRow = this.db
+      .prepare(`SELECT 1 FROM stream_items WHERE stream_item_id = ? LIMIT 1`)
+      .get(input.streamItemId) as { 1: number } | undefined;
+    if (!streamRow) {
+      throw new ProjectClassifierError(
+        "unknown_stream_item",
+        `stream_item_id ${input.streamItemId} does not exist in stream_items; emit it first via 'rig stream emit' or check the id`,
+        { streamItemId: input.streamItemId },
+      );
+    }
+
     // Pre-check idempotency for a clean error path (constraint violation
     // would also catch this but the SQLite error message is opaque).
     const existing = this.db

packages/daemon/src/domain/view-event-bridge.ts

@@ -0,0 +1,97 @@
+import type { EventBus } from "./event-bus.js";
+import type { ViewProjector } from "./view-projector.js";
+import type { PersistedEvent } from "./types.js";
+import { BUILT_IN_VIEW_NAMES } from "./view-projector.js";
+
+/**
+ * View event bridge (PL-004 Phase B R1; closes guard BLOCKER 2).
+ *
+ * Subscribes to coordination state-mutation events on the event-bus and
+ * emits `view.changed` for the affected built-in views via
+ * ViewProjector.notifyViewChanged. Without this bridge, SSE consumers
+ * connecting to /api/views/:name/sse never receive change notifications
+ * because nothing in production wires queue/project mutations to the
+ * view.changed event source.
+ *
+ * Per PRD § L5 acceptance criterion: "view-projector emits view.changed
+ * when underlying state changes" + slice IMPL § Guard Checkpoint Focus
+ * item 8 + R1 guard finding 2.
+ *
+ * Mapping (event type → affected built-in views):
+ * - queue.created            → recently-active, founder, pod-load, activity
+ * - queue.handed_off         → recently-active, pod-load, activity
+ * - queue.claimed            → recently-active, pod-load, activity
+ * - queue.unclaimed          → recently-active, pod-load, activity
+ * - qitem.fallback_routed    → recently-active, pod-load, activity
+ * - qitem.closure_overdue    → recently-active, escalations, activity
+ * - inbox.absorbed           → recently-active, pod-load, activity
+ * - inbox.denied             → activity
+ * - project.classified       → activity (downstream views may consume project_classifications)
+ *
+ * Conservative principle: when in doubt, emit. SSE consumers can filter
+ * by view name. Over-emission is correct (no missed events); the slight
+ * SSE traffic increase is acceptable for Phase B.
+ *
+ * The bridge does NOT emit view.changed for `view.changed` itself
+ * (avoid feedback loops) or for `classifier.lease_*` events (lease state
+ * is exposed via the project SSE, not the view SSE).
+ *
+ * Custom views: this bridge does NOT emit per-custom-view events. Custom
+ * view consumers can subscribe to /api/views/sse (the generic stream)
+ * and filter by viewName. A future enhancement could parse custom view
+ * SQL to determine which built-in event types it depends on; v0 keeps
+ * the bridge built-in-only.
+ */
+
+const EVENT_TO_VIEWS: Record<string, readonly string[]> = {
+  "queue.created":          ["recently-active", "founder", "pod-load", "activity"],
+  "queue.handed_off":       ["recently-active", "pod-load", "activity"],
+  "queue.claimed":          ["recently-active", "pod-load", "activity"],
+  "queue.unclaimed":        ["recently-active", "pod-load", "activity"],
+  "qitem.fallback_routed":  ["recently-active", "pod-load", "activity"],
+  "qitem.closure_overdue":  ["recently-active", "escalations", "activity"],
+  "inbox.absorbed":         ["recently-active", "pod-load", "activity"],
+  "inbox.denied":           ["activity"],
+  "project.classified":     ["activity"],
+};
+
+export interface ViewEventBridgeStop {
+  (): void;
+}
+
+/**
+ * Wire the bridge. Returns a function that unsubscribes (used in tests +
+ * for graceful daemon shutdown). The bridge subscribes to event-bus and
+ * only fires for known coordination event types.
+ */
+export function wireViewEventBridge(
+  eventBus: EventBus,
+  viewProjector: ViewProjector,
+): ViewEventBridgeStop {
+  // Validate the mapping references built-in view names. If a built-in is
+  // renamed/removed without updating EVENT_TO_VIEWS, fail fast at startup.
+  const builtInSet = new Set<string>(BUILT_IN_VIEW_NAMES as readonly string[]);
+  for (const [evt, views] of Object.entries(EVENT_TO_VIEWS)) {
+    for (const v of views) {
+      if (!builtInSet.has(v)) {
+        throw new Error(
+          `view-event-bridge: EVENT_TO_VIEWS maps event '${evt}' to unknown built-in view '${v}'; update mapping or built-in list`,
+        );
+      }
+    }
+  }
+
+  return eventBus.subscribe((event: PersistedEvent) => {
+    const affectedViews = EVENT_TO_VIEWS[event.type];
+    if (!affectedViews) return;
+    for (const viewName of affectedViews) {
+      try {
+        viewProjector.notifyViewChanged(viewName, event.type);
+      } catch {
+        // Best-effort: bridge errors must not unwind the underlying state
+        // mutation. Drop silently; SSE consumers' worst case is a missed
+        // wake-up, not state corruption.
+      }
+    }
+  });
+}

packages/daemon/src/domain/view-projector.ts

@@ -153,7 +153,15 @@ export class ViewProjector {
         params = [...rigParams, limit];
         break;
       case "pod-load":
-        // Per-destination qitem counts (active states only).
+        // R1 NOTE 4: PRD § L5 names this view "per-pod queue-item counts",
+        // but Phase A's queue_items table does not carry pod metadata —
+        // destination_session is `<member>@<rig>` shape and there is no
+        // pod_id column. v0 interpretation: group by destination_session
+        // (i.e., per-seat counts). When pod metadata becomes available
+        // (Phase D workflow runtime or later), this view should be
+        // upgraded to GROUP BY pod_id with destination_session aggregation
+        // as a sub-grouping. Documented v0 behavior preserves the view
+        // name but reduces the grouping key.
         sql = `
           SELECT destination_session AS pod, COUNT(*) AS active_count
           FROM queue_items

packages/daemon/src/routes/projects.ts

@@ -32,6 +32,7 @@ export function projectsRoutes(): Hono {
     if (err instanceof ProjectClassifierError) {
       const status = err.code === "idempotency_violation" ? 409
         : err.code === "project_not_found" ? 404
+        : err.code === "unknown_stream_item" ? 400
         : 500;
       return c.json({ error: err.code, message: err.message, ...(err.meta ?? {}) }, status as 200);
     }
@@ -51,10 +52,20 @@ export function projectsRoutes(): Hono {
   }
 
   // POST /lease/acquire — acquire active classifier lease for caller.
+  // R1 NOTE 3: optional `evaluateDeadnessFirst: true` causes the route to
+  // call evaluateDeadness BEFORE acquire, which clears stale TTL-expired or
+  // dead-holder leases. Without this opt-in, acquire returns 409 lease_held
+  // even when the holder is dead (the operator-verb reclaim path is the
+  // only other way to clear a dead lease without waiting for TTL+next
+  // evaluateDeadness call). Default OFF — operators / classifiers
+  // explicitly request the proactive cleanup.
   app.post("/lease/acquire", async (c) => {
-    const body = await c.req.json<{ classifierSession?: string }>().catch(() => ({} as never));
+    const body = await c.req.json<{ classifierSession?: string; evaluateDeadnessFirst?: boolean }>().catch(() => ({} as never));
     if (!body.classifierSession) return c.json({ error: "classifierSession is required" }, 400);
     try {
+      if (body.evaluateDeadnessFirst === true) {
+        getLease(c).evaluateDeadness();
+      }
       const lease = getLease(c).acquire(body.classifierSession);
       return c.json(lease, 201);
     } catch (err) {

packages/daemon/src/startup.ts

@@ -66,6 +66,7 @@ import { OutboxHandler } from "./domain/outbox-handler.js";
 import { ProjectClassifier } from "./domain/project-classifier.js";
 import { ClassifierLeaseManager } from "./domain/classifier-lease-manager.js";
 import { ViewProjector } from "./domain/view-projector.js";
+import { wireViewEventBridge } from "./domain/view-event-bridge.js";
 import { SpecReviewService } from "./domain/spec-review-service.js";
 import { SpecLibraryService } from "./domain/spec-library-service.js";
 import { WhoamiService } from "./domain/whoami-service.js";
@@ -153,6 +154,14 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
   // the leaseManager dep slot and project-classifier can share one instance.
   // isAlive is post-attached after whoami-service is constructed.
   const classifierLeaseManagerInstance = new ClassifierLeaseManager(db, eventBus);
+  // PL-004 Phase B — view-projector. Constructed early so both the
+  // viewProjector dep slot and the view-event-bridge can share one instance.
+  const viewProjectorInstance = new ViewProjector(db, eventBus);
+  // PL-004 Phase B R1 (closes guard BLOCKER 2): wire the view event bridge
+  // so queue/inbox/project mutations emit view.changed for affected built-in
+  // views. SSE consumers on /api/views/:name/sse now receive change events
+  // when underlying state mutates.
+  wireViewEventBridge(eventBus, viewProjectorInstance);
 
   const tmuxAdapter = new TmuxAdapter(opts?.tmuxExec ?? execCommand);
   // cmuxFactory takes precedence (for tests), then cmuxExec-based CLI transport, then default
@@ -426,7 +435,7 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
     outboxHandler: new OutboxHandler(db),
     classifierLeaseManager: classifierLeaseManagerInstance,
     projectClassifier: new ProjectClassifier(db, eventBus, classifierLeaseManagerInstance),
-    viewProjector: new ViewProjector(db, eventBus),
+    viewProjector: viewProjectorInstance,
     askService: (() => {
       const psProjectionService = new PsProjectionService({ db });
       const execDep = (cmd: string, args: string[]): Promise<{ stdout: string; exitCode: number }> =>

packages/daemon/test/project-classifier.test.ts

@@ -4,8 +4,10 @@ import { createDb } from "../src/db/connection.js";
 import { migrate } from "../src/db/migrate.js";
 import { coreSchema } from "../src/db/migrations/001_core_schema.js";
 import { eventsSchema } from "../src/db/migrations/003_events.js";
+import { streamItemsSchema } from "../src/db/migrations/023_stream_items.js";
 import { classifierLeasesSchema } from "../src/db/migrations/029_classifier_leases.js";
 import { projectClassificationsSchema } from "../src/db/migrations/028_project_classifications.js";
+import { StreamStore } from "../src/domain/stream-store.js";
 import { EventBus } from "../src/domain/event-bus.js";
 import { ClassifierLeaseManager } from "../src/domain/classifier-lease-manager.js";
 import {
@@ -20,14 +22,28 @@ describe("ProjectClassifier (PL-004 Phase B; L2 classifier write path)", () => {
   let bus: EventBus;
   let leaseMgr: ClassifierLeaseManager;
   let classifier: ProjectClassifier;
+  let streamStore: StreamStore;
   let captured: PersistedEvent[];
 
+  // R1 fix (BLOCKER 1): tests now migrate streamItemsSchema (Phase A
+  // migration 023) and seed real stream_items rows so the L1→L2 FK +
+  // existence check in project-classifier can be exercised end-to-end.
+  // Helper that emits a stream item via Phase A's StreamStore.
+  function seedStreamItem(streamItemId: string): void {
+    streamStore.emit({
+      streamItemId,
+      sourceSession: "discovery@rig",
+      body: `body for ${streamItemId}`,
+    });
+  }
+
   beforeEach(() => {
     db = createDb();
-    migrate(db, [coreSchema, eventsSchema, classifierLeasesSchema, projectClassificationsSchema]);
+    migrate(db, [coreSchema, eventsSchema, streamItemsSchema, classifierLeasesSchema, projectClassificationsSchema]);
     bus = new EventBus(db);
     leaseMgr = new ClassifierLeaseManager(db, bus);
     classifier = new ProjectClassifier(db, bus, leaseMgr);
+    streamStore = new StreamStore(db, bus);
     captured = [];
     bus.subscribe((e) => captured.push(e));
   });
@@ -36,6 +52,7 @@ describe("ProjectClassifier (PL-004 Phase B; L2 classifier write path)", () => {
 
   it("classify with valid lease creates project_classifications row + emits project.classified", () => {
     leaseMgr.acquire("alice@rig");
+    seedStreamItem("stream-1");
     const proj = classifier.classify({
       streamItemId: "stream-1",
       classifierSession: "alice@rig",
@@ -50,6 +67,7 @@ describe("ProjectClassifier (PL-004 Phase B; L2 classifier write path)", () => {
   });
 
   it("classify without active lease throws no_active_lease", () => {
+    seedStreamItem("stream-1");
     expect(() => classifier.classify({
       streamItemId: "stream-1",
       classifierSession: "alice@rig",
@@ -58,14 +76,49 @@ describe("ProjectClassifier (PL-004 Phase B; L2 classifier write path)", () => {
 
   it("classify by non-holder throws lease_held", () => {
     leaseMgr.acquire("alice@rig");
+    seedStreamItem("stream-1");
     expect(() => classifier.classify({
       streamItemId: "stream-1",
       classifierSession: "bob@rig",
     })).toThrow(/lease_held|alice@rig/);
   });
 
+  it("R1 BLOCKER 1: classify of nonexistent stream_item_id throws unknown_stream_item (no FK violation surfaced)", () => {
+    leaseMgr.acquire("alice@rig");
+    // No seed: stream_items has no row for "nonexistent-stream".
+    try {
+      classifier.classify({
+        streamItemId: "nonexistent-stream",
+        classifierSession: "alice@rig",
+        classificationType: "idea",
+      });
+      throw new Error("should have thrown");
+    } catch (err) {
+      expect(err).toBeInstanceOf(ProjectClassifierError);
+      expect((err as ProjectClassifierError).code).toBe("unknown_stream_item");
+      expect((err as ProjectClassifierError).meta?.streamItemId).toBe("nonexistent-stream");
+    }
+    // Defense-in-depth: confirm no row was inserted (existence check fired
+    // before INSERT, so FK constraint never had to defend).
+    const projectionAttempts = classifier.list();
+    expect(projectionAttempts).toHaveLength(0);
+  });
+
+  it("R1 BLOCKER 1: FK constraint is the safety net if existence check is bypassed", () => {
+    // Direct INSERT bypassing project-classifier should be blocked by the
+    // SQLite FK constraint (PRAGMA foreign_keys = ON in connection.ts).
+    expect(() => {
+      db.prepare(
+        `INSERT INTO project_classifications (
+          project_id, stream_item_id, classifier_session, ts_projected
+        ) VALUES (?, ?, ?, ?)`
+      ).run("proj-bypass", "nonexistent-via-bypass", "alice@rig", "2026-05-03T00:00:00Z");
+    }).toThrow(/FOREIGN KEY constraint failed/);
+  });
+
   it("classify is idempotent on stream_item_id (re-projection → idempotency_violation 409)", () => {
     leaseMgr.acquire("alice@rig");
+    seedStreamItem("stream-1");
     classifier.classify({
       streamItemId: "stream-1",
       classifierSession: "alice@rig",
@@ -89,6 +142,7 @@ describe("ProjectClassifier (PL-004 Phase B; L2 classifier write path)", () => {
 
   it("first classification's classifier_session is preserved on re-projection attempts", () => {
     leaseMgr.acquire("alice@rig");
+    seedStreamItem("stream-1");
     const first = classifier.classify({
       streamItemId: "stream-1",
       classifierSession: "alice@rig",
@@ -116,6 +170,8 @@ describe("ProjectClassifier (PL-004 Phase B; L2 classifier write path)", () => {
 
   it("classify accepts all 6 classification fields + action; null when unset", () => {
     leaseMgr.acquire("alice@rig");
+    seedStreamItem("stream-full");
+    seedStreamItem("stream-minimal");
     const full = classifier.classify({
       streamItemId: "stream-full",
       classifierSession: "alice@rig",
@@ -142,6 +198,9 @@ describe("ProjectClassifier (PL-004 Phase B; L2 classifier write path)", () => {
 
   it("list filters by classifierSession + classificationDestination", () => {
     leaseMgr.acquire("alice@rig");
+    seedStreamItem("s1");
+    seedStreamItem("s2");
+    seedStreamItem("s3");
     classifier.classify({
       streamItemId: "s1",
       classifierSession: "alice@rig",

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

@@ -5,11 +5,13 @@ import { createDb } from "../src/db/connection.js";
 import { migrate } from "../src/db/migrate.js";
 import { coreSchema } from "../src/db/migrations/001_core_schema.js";
 import { eventsSchema } from "../src/db/migrations/003_events.js";
+import { streamItemsSchema } from "../src/db/migrations/023_stream_items.js";
 import { classifierLeasesSchema } from "../src/db/migrations/029_classifier_leases.js";
 import { projectClassificationsSchema } from "../src/db/migrations/028_project_classifications.js";
 import { EventBus } from "../src/domain/event-bus.js";
 import { ClassifierLeaseManager } from "../src/domain/classifier-lease-manager.js";
 import { ProjectClassifier } from "../src/domain/project-classifier.js";
+import { StreamStore } from "../src/domain/stream-store.js";
 import { projectsRoutes } from "../src/routes/projects.js";
 
 function buildApp(opts: {
@@ -33,14 +35,27 @@ describe("projects routes (PL-004 Phase B)", () => {
   let bus: EventBus;
   let leaseMgr: ClassifierLeaseManager;
   let classifier: ProjectClassifier;
+  let streamStore: StreamStore;
   let app: Hono;
 
+  // R1 fix (BLOCKER 1): tests now migrate streamItemsSchema (Phase A
+  // migration 023) and seed real stream_items rows so the L1→L2 FK +
+  // existence check is exercised end-to-end through the route.
+  function seedStreamItem(streamItemId: string): void {
+    streamStore.emit({
+      streamItemId,
+      sourceSession: "discovery@rig",
+      body: `body for ${streamItemId}`,
+    });
+  }
+
   beforeEach(() => {
     db = createDb();
-    migrate(db, [coreSchema, eventsSchema, classifierLeasesSchema, projectClassificationsSchema]);
+    migrate(db, [coreSchema, eventsSchema, streamItemsSchema, classifierLeasesSchema, projectClassificationsSchema]);
     bus = new EventBus(db);
     leaseMgr = new ClassifierLeaseManager(db, bus);
     classifier = new ProjectClassifier(db, bus, leaseMgr);
+    streamStore = new StreamStore(db, bus);
     app = buildApp({ eventBus: bus, classifier, leaseMgr });
   });
 
@@ -59,6 +74,7 @@ describe("projects routes (PL-004 Phase B)", () => {
   });
 
   it("POST /api/projects/project requires lease + idempotent on stream_item_id", async () => {
+    seedStreamItem("stream-x");
     await app.request("/api/projects/lease/acquire", {
       method: "POST",
       headers: { "Content-Type": "application/json" },
@@ -92,6 +108,7 @@ describe("projects routes (PL-004 Phase B)", () => {
   });
 
   it("POST /api/projects/project without active lease returns 409 no_active_lease", async () => {
+    seedStreamItem("stream-x");
     const res = await app.request("/api/projects/project", {
       method: "POST",
       headers: { "Content-Type": "application/json" },
@@ -105,6 +122,29 @@ describe("projects routes (PL-004 Phase B)", () => {
     expect(err.error).toBe("no_active_lease");
   });
 
+  it("R1 BLOCKER 1: POST /api/projects/project with nonexistent stream_item_id returns 400 unknown_stream_item", async () => {
+    await app.request("/api/projects/lease/acquire", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ classifierSession: "alice@rig" }),
+    });
+    // Note: NO seedStreamItem call — stream-nonexistent has no row in stream_items.
+    const res = await app.request("/api/projects/project", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({
+        streamItemId: "stream-nonexistent",
+        classifierSession: "alice@rig",
+        classificationType: "idea",
+      }),
+    });
+    expect(res.status).toBe(400);
+    const err = (await res.json()) as { error: string; message: string; streamItemId: string };
+    expect(err.error).toBe("unknown_stream_item");
+    expect(err.message).toMatch(/does not exist/);
+    expect(err.streamItemId).toBe("stream-nonexistent");
+  });
+
   it("POST /api/projects/reclaim-classifier with --if-dead refuses on alive holder", async () => {
     await app.request("/api/projects/lease/acquire", {
       method: "POST",
@@ -161,6 +201,7 @@ describe("projects routes (PL-004 Phase B)", () => {
       body: JSON.stringify({ classifierSession: "alice@rig" }),
     });
     for (const [id, dest] of [["s1", "planning@rig"], ["s2", "delivery@rig"], ["s3", "planning@rig"]] as Array<[string, string]>) {
+      seedStreamItem(id);
       await app.request("/api/projects/project", {
         method: "POST",
         headers: { "Content-Type": "application/json" },

packages/daemon/test/view-event-bridge.test.ts

@@ -0,0 +1,199 @@
+import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import type Database from "better-sqlite3";
+import { createDb } from "../src/db/connection.js";
+import { migrate } from "../src/db/migrate.js";
+import { coreSchema } from "../src/db/migrations/001_core_schema.js";
+import { eventsSchema } from "../src/db/migrations/003_events.js";
+import { streamItemsSchema } from "../src/db/migrations/023_stream_items.js";
+import { queueItemsSchema } from "../src/db/migrations/024_queue_items.js";
+import { queueTransitionsSchema } from "../src/db/migrations/025_queue_transitions.js";
+import { inboxEntriesSchema } from "../src/db/migrations/026_inbox_entries.js";
+import { classifierLeasesSchema } from "../src/db/migrations/029_classifier_leases.js";
+import { projectClassificationsSchema } from "../src/db/migrations/028_project_classifications.js";
+import { viewsCustomSchema } from "../src/db/migrations/030_views_custom.js";
+import { EventBus } from "../src/domain/event-bus.js";
+import { QueueRepository } from "../src/domain/queue-repository.js";
+import { InboxHandler } from "../src/domain/inbox-handler.js";
+import { StreamStore } from "../src/domain/stream-store.js";
+import { ViewProjector } from "../src/domain/view-projector.js";
+import { ClassifierLeaseManager } from "../src/domain/classifier-lease-manager.js";
+import { ProjectClassifier } from "../src/domain/project-classifier.js";
+import { wireViewEventBridge } from "../src/domain/view-event-bridge.js";
+import type { PersistedEvent } from "../src/domain/types.js";
+
+/**
+ * View event bridge tests (PL-004 Phase B R1; closes guard BLOCKER 2).
+ *
+ * The bridge subscribes to coordination state-mutation events and emits
+ * view.changed for affected built-in views via ViewProjector.notifyViewChanged.
+ * Without this, /api/views/:name/sse never receives change notifications.
+ */
+
+describe("view-event-bridge (PL-004 Phase B R1; BLOCKER 2 fix)", () => {
+  let db: Database.Database;
+  let bus: EventBus;
+  let queueRepo: QueueRepository;
+  let inbox: InboxHandler;
+  let streamStore: StreamStore;
+  let projector: ViewProjector;
+  let leaseMgr: ClassifierLeaseManager;
+  let classifier: ProjectClassifier;
+  let captured: PersistedEvent[];
+
+  beforeEach(() => {
+    db = createDb();
+    migrate(db, [
+      coreSchema, eventsSchema,
+      streamItemsSchema, queueItemsSchema, queueTransitionsSchema, inboxEntriesSchema,
+      classifierLeasesSchema, projectClassificationsSchema, viewsCustomSchema,
+    ]);
+    bus = new EventBus(db);
+    queueRepo = new QueueRepository(db, bus);
+    inbox = new InboxHandler(db, bus, queueRepo);
+    streamStore = new StreamStore(db, bus);
+    projector = new ViewProjector(db, bus);
+    leaseMgr = new ClassifierLeaseManager(db, bus);
+    classifier = new ProjectClassifier(db, bus, leaseMgr);
+    captured = [];
+    bus.subscribe((e) => captured.push(e));
+    wireViewEventBridge(bus, projector);
+  });
+
+  afterEach(() => db.close());
+
+  function viewChangedEvents(): Array<{ viewName: string; cause: string }> {
+    return captured
+      .filter((e) => e.type === "view.changed")
+      .map((e) => ({ viewName: (e as { viewName: string }).viewName, cause: (e as { cause: string }).cause }));
+  }
+
+  it("queue.created triggers view.changed for recently-active + founder + pod-load + activity", async () => {
+    await queueRepo.create({
+      sourceSession: "alice@rig",
+      destinationSession: "bob@rig",
+      body: "test",
+      nudge: false,
+    });
+    const events = viewChangedEvents();
+    const viewNames = events.map((e) => e.viewName);
+    expect(viewNames).toContain("recently-active");
+    expect(viewNames).toContain("founder");
+    expect(viewNames).toContain("pod-load");
+    expect(viewNames).toContain("activity");
+    // Each event has cause = the source event type.
+    expect(events.every((e) => e.cause === "queue.created")).toBe(true);
+  });
+
+  it("queue.handed_off triggers view.changed for recently-active + pod-load + activity", async () => {
+    const item = await queueRepo.create({
+      sourceSession: "alice@rig",
+      destinationSession: "bob@rig",
+      body: "test",
+      nudge: false,
+    });
+    captured.length = 0; // reset to focus on the handoff event
+    await queueRepo.handoff({
+      qitemId: item.qitemId,
+      fromSession: "bob@rig",
+      toSession: "carol@rig",
+      nudge: false,
+    });
+    const events = viewChangedEvents();
+    expect(events.some((e) => e.viewName === "recently-active" && e.cause === "queue.handed_off")).toBe(true);
+    expect(events.some((e) => e.viewName === "pod-load" && e.cause === "queue.handed_off")).toBe(true);
+    expect(events.some((e) => e.viewName === "activity" && e.cause === "queue.handed_off")).toBe(true);
+    // queue.handed_off ALSO emits queue.created for the new qitem (which fires its own view.changed batch).
+  });
+
+  it("queue.claimed triggers view.changed for recently-active + pod-load + activity", async () => {
+    const item = await queueRepo.create({
+      sourceSession: "alice@rig",
+      destinationSession: "bob@rig",
+      body: "test",
+      nudge: false,
+    });
+    captured.length = 0;
+    queueRepo.claim({ qitemId: item.qitemId, destinationSession: "bob@rig" });
+    const events = viewChangedEvents();
+    const claimEvents = events.filter((e) => e.cause === "queue.claimed");
+    expect(claimEvents.map((e) => e.viewName).sort()).toEqual(["activity", "pod-load", "recently-active"]);
+  });
+
+  it("inbox.absorbed triggers view.changed for recently-active + pod-load + activity", async () => {
+    const drop = inbox.drop({
+      destinationSession: "bob@rig",
+      senderSession: "alice@rig",
+      body: "test",
+    });
+    captured.length = 0;
+    await inbox.absorb(drop.inboxId, "bob@rig");
+    const events = viewChangedEvents();
+    const absorbEvents = events.filter((e) => e.cause === "inbox.absorbed");
+    expect(absorbEvents.map((e) => e.viewName).sort()).toEqual(["activity", "pod-load", "recently-active"]);
+  });
+
+  it("inbox.denied triggers view.changed for activity only", () => {
+    const drop = inbox.drop({
+      destinationSession: "bob@rig",
+      senderSession: "alice@rig",
+      body: "test",
+    });
+    captured.length = 0;
+    inbox.deny(drop.inboxId, "bob@rig", "off-topic");
+    const events = viewChangedEvents();
+    const denyEvents = events.filter((e) => e.cause === "inbox.denied");
+    expect(denyEvents.map((e) => e.viewName)).toEqual(["activity"]);
+  });
+
+  it("project.classified triggers view.changed for activity only", () => {
+    streamStore.emit({ streamItemId: "stream-1", sourceSession: "discovery@rig", body: "x" });
+    leaseMgr.acquire("alice@rig");
+    captured.length = 0;
+    classifier.classify({
+      streamItemId: "stream-1",
+      classifierSession: "alice@rig",
+      classificationType: "idea",
+    });
+    const events = viewChangedEvents();
+    const classifyEvents = events.filter((e) => e.cause === "project.classified");
+    expect(classifyEvents.map((e) => e.viewName)).toEqual(["activity"]);
+  });
+
+  it("classifier.lease_acquired does NOT trigger view.changed (lease lifecycle is project SSE, not view SSE)", () => {
+    leaseMgr.acquire("alice@rig");
+    const events = viewChangedEvents();
+    expect(events.filter((e) => e.cause === "classifier.lease_acquired")).toHaveLength(0);
+  });
+
+  it("view.changed itself does NOT trigger view.changed (no feedback loop)", () => {
+    projector.notifyViewChanged("recently-active", "manual-test");
+    captured.length = 0;
+    projector.notifyViewChanged("recently-active", "manual-test-2");
+    // The bridge should not re-emit view.changed in response to view.changed.
+    const newViewChangedEvents = captured.filter((e) => e.type === "view.changed");
+    // Only the direct notifyViewChanged call's event; no echo.
+    expect(newViewChangedEvents).toHaveLength(1);
+    expect((newViewChangedEvents[0] as { cause: string }).cause).toBe("manual-test-2");
+  });
+
+  it("unsubscribe stops the bridge", async () => {
+    // Re-wire and capture the unsubscribe function.
+    const localCaptured: PersistedEvent[] = [];
+    bus.subscribe((e) => localCaptured.push(e));
+    const stop = wireViewEventBridge(bus, projector);
+    stop();
+    await queueRepo.create({
+      sourceSession: "alice@rig",
+      destinationSession: "bob@rig",
+      body: "test-after-unsubscribe",
+      nudge: false,
+    });
+    // The MAIN bridge from beforeEach is still wired; it emits view.changed.
+    // The local one we just stopped doesn't emit. Verify only ONE bridge fired
+    // (not two): activity should appear exactly once.
+    const activityEvents = captured.filter(
+      (e) => e.type === "view.changed" && (e as { viewName: string }).viewName === "activity" && (e as { cause: string }).cause === "queue.created",
+    );
+    expect(activityEvents).toHaveLength(1);
+  });
+});

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

@@ -11,6 +11,7 @@ import { viewsCustomSchema } from "../src/db/migrations/030_views_custom.js";
 import { EventBus } from "../src/domain/event-bus.js";
 import { QueueRepository } from "../src/domain/queue-repository.js";
 import { ViewProjector } from "../src/domain/view-projector.js";
+import { wireViewEventBridge } from "../src/domain/view-event-bridge.js";
 import { viewsRoutes } from "../src/routes/views.js";
 
 function buildApp(opts: {
@@ -156,4 +157,67 @@ describe("views routes (PL-004 Phase B)", () => {
       await res.body?.cancel();
     }
   });
+
+  // ---- R1 BLOCKER 2: SSE consumer observes view.changed when queue mutates ----
+  // Per guard finding 2: route-level test that mutates queue state and
+  // observes a view.changed event. Wires the view-event-bridge in the test
+  // so the production code path is exercised.
+  it("R1 BLOCKER 2: queue mutation triggers view.changed visible to SSE consumer", async () => {
+    // Wire the bridge so queue.created → view.changed flows through the
+    // event-bus the SSE handler is subscribed to.
+    wireViewEventBridge(bus, projector);
+
+    // Subscribe to the SSE stream and read until we see a view.changed
+    // line for the recently-active view (caused by queue.created), or
+    // bail out after a short timeout.
+    const sseResPromise = app.request("/api/views/recently-active/sse");
+
+    // Mutate queue state in parallel: create a new qitem.
+    await queueRepo.create({
+      sourceSession: "alice@rig",
+      destinationSession: "bob@rig",
+      body: "trigger view.changed",
+      nudge: false,
+    });
+
+    const res = await sseResPromise;
+    expect(res.status).toBe(200);
+    expect(res.headers.get("content-type") ?? "").toContain("text/event-stream");
+
+    // Read SSE body for up to ~1.5s and look for a view.changed event.
+    const reader = res.body!.getReader();
+    const decoder = new TextDecoder();
+    let buffer = "";
+    const deadline = Date.now() + 1500;
+    let observed = false;
+    try {
+      while (Date.now() < deadline && !observed) {
+        const readP = reader.read();
+        const tickP = new Promise<{ done: true; value: undefined }>((resolve) =>
+          setTimeout(() => resolve({ done: true, value: undefined }), 200),
+        );
+        const { done, value } = (await Promise.race([readP, tickP])) as { done: boolean; value?: Uint8Array };
+        if (!done && value) buffer += decoder.decode(value, { stream: true });
+        if (buffer.includes('"type":"view.changed"') && buffer.includes('"viewName":"recently-active"')) {
+          observed = true;
+          break;
+        }
+        // Push another mutation if we haven't seen the event yet — the
+        // first one may have raced past the SSE handler's subscription.
+        if (Date.now() < deadline && !observed) {
+          await queueRepo.create({
+            sourceSession: "alice@rig",
+            destinationSession: "bob@rig",
+            body: `nudge-${Date.now()}`,
+            nudge: false,
+          });
+        }
+      }
+    } finally {
+      // Cancel via the reader (releases the stream lock cleanly).
+      await reader.cancel().catch(() => {});
+    }
+
+    expect(observed).toBe(true);
+  });
 });