feat(daemon,cli): PL-004 Phase B coordination primitive — L2 Project (Classifier) + L5 View

Closes Phase B dispatch qitem-20260503040433-cfd76ab9 + delivery-orch handoff packet pl004-phase-b-driver-handoff-20260503.md. Foundation: openrig main 194a1ebb (Phase A R1 + SSE fix; locally promoted with caveats). Branch: phase-b-coordination-primitive-system (worktree-scoped; main NOT touched).

==== AUDIT (12 hard gates; ALL GREEN in clean Phase B worktree) ====

1. Phase A migrations 023-027 present (5 files at packages/daemon/src/db/migrations/02[3-7]_*.ts).
2. db/connection.ts + db/migrate.ts unchanged from Phase A; same Migration{name,sql} shape used.
3. Phase A stream-store.ts read interface evaluated — chose path (b) direct query against `stream_items` for the future "find unclassified" pattern (NOT touching stream-store; classifier write path doesn't actually need the listUnclassified pattern; project verb takes streamItemId from caller).
4. queue-repository.ts read APIs sufficient — view-projector queries queue_items + queue_transitions tables directly via shared db handle (read-only, no Phase A write contract changed).
5. event-bus.ts RigEvent union has Phase A's 9 events; Phase B appends 6 new events alongside (untouched: stream.emitted, queue.created, queue.handed_off, queue.claimed, queue.unclaimed, qitem.fallback_routed, qitem.closure_overdue, inbox.absorbed, inbox.denied).
6. whoami-service.ts + node-inventory.ts present; isAlive wired via direct sessions-table query (`SELECT 1 FROM sessions WHERE session_name = ? AND status = 'running'`) — simpler than threading whoami-service through the dep graph, same liveness signal.
7. POC rigx project / view-proto sources present at substrate (348 + 304 lines); contract reverse-engineered.
8. No in-flight Phase A revisions on main (verified via git log).
9. Cross-loop POC stable (per Phase A R1 promotion artifact).
10. Phase A locally promoted at 194a1ebb (delivery-ready/mode-2/coordination-primitive-system-phase-a.md status: promoted).
11. Test discipline: ALL test runs use explicit OPENRIG_HOME + OPENRIG_DB under /tmp/pl004-phase-b-* per Phase A R1 incident lesson.
12. SSE/static routes mounted BEFORE /:id catchalls in routes/projects.ts (sse + watch + lease + list before /:projectId) and routes/views.ts (list + sse + watch + /:viewName/sse before /:viewName) per Phase A R1 SSE fix lesson; live GET tests assert content-type: text/event-stream + regression guard against not_found shadowing.

==== R0 SCOPE DELIVERED (21 files; +2953 / -2) ====

Migrations (3 NEW; numbering 028-030 contiguous after Phase A's 027):
- 028_project_classifications: project_id PK + UNIQUE(stream_item_id) for idempotency + classifier_session + 6 classification fields + ts_projected.
- 029_classifier_leases: lease_id PK + classifier_session + acquired_at/expires_at/last_heartbeat + state enum + reclaim fields. Single-writer enforced via partial UNIQUE INDEX on state='active'.
- 030_views_custom: view_id PK + view_name UNIQUE + definition (operator-supplied SQL) + registered_by_session + last_evaluated_at.

Domain (3 NEW; per IMPL § Domain):
- classifier-lease-manager.ts (392 lines): acquire (idempotent same-session) / heartbeat (extends TTL) / evaluateDeadness (TTL OR isAlive=false → expired + classifier.dead event) / reclaim (operator-verb; --if-dead respects isAlive) / requireActiveHolder (validation gate for classifier write path). attachIsAlive() setter for post-construction wiring (whoami/sessions table later in dep graph).
- project-classifier.ts (222 lines): classify (lease check + idempotency check + INSERT + project.classified event) / getById / getByStreamItemId / list. Pure SQLite-canonical write; no Hono.
- view-projector.ts (350 lines): show (dispatches to 6 built-in OR custom) / 6 built-in views (recently-active, founder, pod-load, escalations, held, activity) as parameterized SQL with fixture-rig exclusion (OPENRIG_VIEW_INCLUDE_FIXTURES=1 opt-in) + --rig filter / registerCustomView (UNIQUE on view_name; reserved built-in names rejected; re-registration updates) / listCustomViews / notifyViewChanged (emits view.changed event for SSE consumers).

Routes (2 NEW):
- routes/projects.ts (183 lines): POST /lease/acquire, /lease/heartbeat, /reclaim-classifier, /project; GET /lease, /list, /sse + /watch (SSE handler dual-mounted), /:projectId. Routes mounted in correct order (literal before /:id catchall per R1 lesson).
- routes/views.ts (124 lines): GET /list, /sse + /watch (generic SSE), /:viewName/sse (view-filtered SSE), /:viewName (run); POST /custom/register. Routes mounted in correct order.

Event-bus extension (light-touch in domain/types.ts; +14 lines):
- 6 NEW Phase B events appended to RigEvent union: project.classified, classifier.lease_acquired, classifier.lease_expired, classifier.dead, classifier.reclaimed, view.changed. Phase A's 9 events untouched.

Startup wiring (light-touch in startup.ts; +33 lines):
- Imports for 3 new migrations + 3 new domain services. Migration list extended by 3 (028, 029, 030). classifierLeaseManagerInstance constructed early (shared between leaseManager dep slot and projectClassifier). attachIsAlive wired in the SessionTransport-construction IIFE alongside Phase A's queueRepoInstance.attachTransport (sessions-table query for liveness check).

Server wiring (light-touch in server.ts; +13 lines):
- AppDeps extended with projectClassifier?, classifierLeaseManager?, viewProjector?. Context setters for the 3 new services. app.route("/api/projects", projectsRoutes()) + app.route("/api/views", viewsRoutes()) mounted alongside Phase A routes.

CLI commands (2 NEW):
- commands/project.ts (194 lines): rig project lease-acquire / lease-heartbeat / lease-show / reclaim-classifier --if-dead / classify <streamItemId> --type --urgency --maturity --confidence --destination --action / list / show <projectId>. Mirrors Phase A's queue.ts pattern.
- commands/view.ts (101 lines): rig view list / show <viewName> --rig --limit --json / register --name --definition --session.

CLI registration (light-touch in cli/src/index.ts; +6 lines):
- Imports projectCommand + viewCommand. ProgramDeps extended with projectDeps? + viewDeps?. addCommand calls.

Tests (5 NEW daemon files + 2 NEW CLI files = 7 new test files; 69 new tests):
- classifier-lease-manager.test.ts (16 cases): acquire idempotency, lease_held conflict, heartbeat extension, deadness via isAlive AND TTL, reclaim semantics, --if-dead refusal-when-alive, requireActiveHolder validation, partial UNIQUE singleton enforcement at DB layer.
- project-classifier.test.ts (7 cases): valid lease + classify, no-lease rejection, lease_held by other session, idempotency_violation on re-projection (with all 6 closure-reason-equivalent fields preserved), null-default classification fields, list filtering.
- view-projector.test.ts (15 cases): all 6 built-in view names + queries, fixture exclusion default + opt-in, --rig filter, view_not_found, custom view registration + reserved-name rejection + re-registration update + list. notifyViewChanged event.
- projects-routes.test.ts (10 cases): all POSTs + GETs round-trip; idempotency 409; reclaim semantics including --if-dead; SSE route-order regression guard (3 cases per R1 SSE lesson: text/event-stream + handler-reached + not-shadowed).
- views-routes.test.ts (9 cases): /list + /show + /custom/register + reserved-name 409 + view_not_found 404 + SSE route-order regression guard (3 cases including /:viewName/sse).
- project.test.ts (6 CLI cases): registration check, lease-acquire POST shape, classify POST shape, idempotency_violation 409 surfacing + non-zero exit, reclaim-classifier --if-dead passthrough, list query params.
- view.test.ts (6 CLI cases): registration check, list GET, show with rig+limit query params, show without query params, view_not_found 404 surfacing + non-zero exit, register POST shape.

==== 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-d3 OPENRIG_DB=/tmp/pl004-phase-b-d3/openrig.sqlite npx vitest run [5 Phase B daemon test files] → 57/57 PASS.
- cli: OPENRIG_HOME=/tmp/pl004-phase-b-c OPENRIG_DB=/tmp/pl004-phase-b-c/openrig.sqlite npx vitest run test/project.test.ts test/view.test.ts → 12/12 PASS.
- daemon tsc + build: exit 0; cli tsc + build: exit 0.

Compiled-output runtime checks:
- node import dist/domain/classifier-lease-manager.js → ClassifierLeaseManager: function ✓
- node import dist/domain/view-projector.js → ViewProjector: function; BUILT_IN_VIEW_NAMES.length: 6 ✓
- node import dist/index.js → project subs: classify, lease-acquire, lease-heartbeat, lease-show, list, reclaim-classifier, show; view subs: list, register, show ✓

Full per-package suites (NO env override; honest baseline):
- daemon: 2391/2391 PASS / 0 FAIL (Phase A R1 baseline 2334 + 57 new Phase B = 2391).
- cli: 769/773 PASS / 4 FAIL named pre-existing flakes (chatroom wait timing, config-store env pollution, two preflight host-port). NO new regressions; matches Phase A R1 baseline.

==== STREAM-READ TRICHOTOMY CHOICE (gate 6) ====

Per advisory review Rec 2 + IMPL audit row 3: the classifier did not require a "find-unclassified" filter on stream-store because the project verb takes streamItemId from the caller. Each agent decides what to project; the daemon enforces the lease + idempotency. So none of the three escape-valves was triggered:
- (a) extending stream-store: not needed (no boundary touch on Phase A).
- (b) direct query: wired via shared db handle for view-projector (queue_items + queue_transitions read-only).
- (c) post-filter in code: not needed.

==== BOUNDARIES PRESERVED ====

- Worktree-only changes; main HEAD unchanged at 194a1ebb throughout (verified pre and post).
- 21 files changed (10 modified + 11 new), all in packages/daemon/{src,test}/* and packages/cli/{src/commands,src/index,test}/*.
- Phase A surfaces NOT touched: stream-store.ts, queue-repository.ts, queue-transition-log.ts, inbox-handler.ts, outbox-handler.ts, hot-potato-enforcer.ts, routes/{stream,queue}.ts, commands/{stream,queue}.ts. Existing 22 migrations + Phase A's 5 migrations (023-027) NOT modified.
- Lifecycle v2 surfaces NOT touched (restore-orchestrator.ts, seat-handover-service.ts, snapshot-repository.ts).
- 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 cross-host promotion added.
- 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 ====

Driver-side focused verification passed (57 + 12 = 69 focused tests under explicit temp DB isolation + compiled-output runtime checks + per-package baselines clean). All 12 audit hard gates GREEN. All 4 ratified Phase B contract surfaces (classifier lease + idempotency + reclaim + 6 built-in views + SSE) implemented per PRD § L2 + § L5.

Per IMPL § Driver Handoff Contract: hand to guard at the milestone-diff boundary. Guard adjudicates against the 18 § Guard Checkpoint Focus items. Then tester runs cross-loop dogfood proof per IMPL § Cross-Loop Dogfood Plan + Tester Proof Expectations.

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

packages/cli/src/commands/project.ts

@@ -0,0 +1,194 @@
+import { Command } from "commander";
+import { DaemonClient } from "../client.js";
+import { getDaemonStatus, getDaemonUrl } from "../daemon-lifecycle.js";
+import { realDeps } from "./daemon.js";
+import type { StatusDeps } from "./status.js";
+
+/**
+ * `rig project` — coordination primitive L2 (classifier) commands (PL-004 Phase B).
+ *
+ * Backed by `/api/projects`. Operates only via the daemon HTTP API.
+ *
+ * Per PRD § L2: classifier judgment stays with the agent; daemon enforces
+ * the lease + idempotency + reclaim contract.
+ */
+
+export interface ProjectDeps extends StatusDeps {}
+
+async function withClient<T>(
+  deps: ProjectDeps,
+  fn: (client: DaemonClient) => Promise<T>
+): Promise<T | undefined> {
+  const status = await getDaemonStatus(deps.lifecycleDeps);
+  if (status.state !== "running" || status.healthy === false) {
+    console.error("Daemon not running. Start it with: rig daemon start");
+    process.exitCode = 1;
+    return undefined;
+  }
+  const client = deps.clientFactory(getDaemonUrl(status));
+  return fn(client);
+}
+
+function printResult(json: boolean, body: unknown, status: number): void {
+  if (json) {
+    console.log(JSON.stringify(body));
+  } else {
+    console.log(JSON.stringify(body, null, 2));
+  }
+  if (status >= 400) process.exitCode = status >= 500 ? 2 : 1;
+}
+
+export function projectCommand(depsOverride?: ProjectDeps): Command {
+  const cmd = new Command("project").description(
+    "Coordination L2 — agent-backed classifier with daemon-enforced lease + idempotency + reclaim",
+  );
+  const getDeps = (): ProjectDeps => depsOverride ?? {
+    lifecycleDeps: realDeps(),
+    clientFactory: (url: string) => new DaemonClient(url),
+  };
+
+  // ---- Lease lifecycle ----
+
+  cmd
+    .command("lease-acquire")
+    .description("Acquire the active classifier lease for the caller")
+    .requiredOption("--session <session>", "Classifier session name")
+    .option("--json", "JSON output for agents")
+    .action(async (opts: { session: string; json?: boolean }) => {
+      const deps = getDeps();
+      await withClient(deps, async (client) => {
+        const res = await client.post<unknown>("/api/projects/lease/acquire", {
+          classifierSession: opts.session,
+        });
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
+  cmd
+    .command("lease-heartbeat")
+    .description("Send a heartbeat for an active classifier lease (extends TTL)")
+    .requiredOption("--lease-id <id>", "Lease ID")
+    .requiredOption("--session <session>", "Classifier session name")
+    .option("--json", "JSON output for agents")
+    .action(async (opts: { leaseId: string; session: string; json?: boolean }) => {
+      const deps = getDeps();
+      await withClient(deps, async (client) => {
+        const res = await client.post<unknown>("/api/projects/lease/heartbeat", {
+          leaseId: opts.leaseId,
+          classifierSession: opts.session,
+        });
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
+  cmd
+    .command("lease-show")
+    .description("Show the currently-active classifier lease")
+    .option("--json", "JSON output for agents")
+    .action(async (opts: { json?: boolean }) => {
+      const deps = getDeps();
+      await withClient(deps, async (client) => {
+        const res = await client.get<unknown>("/api/projects/lease");
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
+  // ---- Operator-verb reclaim ----
+
+  cmd
+    .command("reclaim-classifier")
+    .description(
+      "Operator-verb: reclaim the active classifier lease. Use --if-dead to refuse if holder is still alive.",
+    )
+    .requiredOption("--session <session>", "Session that will hold the new lease")
+    .option("--if-dead", "Only reclaim if the current holder is reported dead by the liveness check")
+    .option("--reason <text>", "Reclaim reason (free-form; recorded in classifier_leases.reclaim_reason)")
+    .option("--json", "JSON output for agents")
+    .action(async (opts: { session: string; ifDead?: boolean; reason?: string; json?: boolean }) => {
+      const deps = getDeps();
+      await withClient(deps, async (client) => {
+        const res = await client.post<unknown>("/api/projects/reclaim-classifier", {
+          byClassifierSession: opts.session,
+          ifDead: opts.ifDead,
+          reason: opts.reason,
+        });
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
+  // ---- Project a stream item ----
+
+  cmd
+    .command("classify <streamItemId>")
+    .description(
+      "Project a stream item with classification fields (idempotent on stream_item_id; requires active lease)",
+    )
+    .requiredOption("--session <session>", "Classifier session name (must hold active lease)")
+    .option("--type <type>", "Classification type (e.g., idea, bug, feature-request)")
+    .option("--urgency <urgency>", "Classification urgency (e.g., normal, high, critical)")
+    .option("--maturity <maturity>", "Classification maturity (e.g., concept, drafted, ratified)")
+    .option("--confidence <confidence>", "Classification confidence (e.g., low, medium, high)")
+    .option("--destination <destination>", "Classification destination (downstream slice/seat)")
+    .option("--action <action>", "Action type (e.g., create, advance)")
+    .option("--json", "JSON output for agents")
+    .action(async (streamItemId: string, opts: {
+      session: string;
+      type?: string;
+      urgency?: string;
+      maturity?: string;
+      confidence?: string;
+      destination?: string;
+      action?: string;
+      json?: boolean;
+    }) => {
+      const deps = getDeps();
+      await withClient(deps, async (client) => {
+        const res = await client.post<unknown>("/api/projects/project", {
+          streamItemId,
+          classifierSession: opts.session,
+          classificationType: opts.type,
+          classificationUrgency: opts.urgency,
+          classificationMaturity: opts.maturity,
+          classificationConfidence: opts.confidence,
+          classificationDestination: opts.destination,
+          action: opts.action,
+        });
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
+  // ---- List + show ----
+
+  cmd
+    .command("list")
+    .description("List project classifications with filters")
+    .option("--session <session>", "Filter by classifier session")
+    .option("--destination <destination>", "Filter by classification destination")
+    .option("--limit <n>", "Result limit", "100")
+    .option("--json", "JSON output for agents")
+    .action(async (opts: { session?: string; destination?: string; limit: string; json?: boolean }) => {
+      const deps = getDeps();
+      const params = new URLSearchParams();
+      if (opts.session) params.set("classifierSession", opts.session);
+      if (opts.destination) params.set("classificationDestination", opts.destination);
+      if (opts.limit) params.set("limit", opts.limit);
+      await withClient(deps, async (client) => {
+        const res = await client.get<unknown>(`/api/projects/list?${params.toString()}`);
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
+  cmd
+    .command("show <projectId>")
+    .description("Show one project classification")
+    .option("--json", "JSON output for agents")
+    .action(async (projectId: string, opts: { json?: boolean }) => {
+      const deps = getDeps();
+      await withClient(deps, async (client) => {
+        const res = await client.get<unknown>(`/api/projects/${encodeURIComponent(projectId)}`);
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
+  return cmd;
+}

packages/cli/src/commands/view.ts

@@ -0,0 +1,101 @@
+import { Command } from "commander";
+import { DaemonClient } from "../client.js";
+import { getDaemonStatus, getDaemonUrl } from "../daemon-lifecycle.js";
+import { realDeps } from "./daemon.js";
+import type { StatusDeps } from "./status.js";
+
+/**
+ * `rig view` — coordination primitive L5 (view) commands (PL-004 Phase B).
+ *
+ * Backed by `/api/views`. 6 built-in views (recently-active, founder,
+ * pod-load, escalations, held, activity) + custom view registration.
+ */
+
+export interface ViewDeps extends StatusDeps {}
+
+async function withClient<T>(
+  deps: ViewDeps,
+  fn: (client: DaemonClient) => Promise<T>
+): Promise<T | undefined> {
+  const status = await getDaemonStatus(deps.lifecycleDeps);
+  if (status.state !== "running" || status.healthy === false) {
+    console.error("Daemon not running. Start it with: rig daemon start");
+    process.exitCode = 1;
+    return undefined;
+  }
+  const client = deps.clientFactory(getDaemonUrl(status));
+  return fn(client);
+}
+
+function printResult(json: boolean, body: unknown, status: number): void {
+  if (json) {
+    console.log(JSON.stringify(body));
+  } else {
+    console.log(JSON.stringify(body, null, 2));
+  }
+  if (status >= 400) process.exitCode = status >= 500 ? 2 : 1;
+}
+
+export function viewCommand(depsOverride?: ViewDeps): Command {
+  const cmd = new Command("view").description(
+    "Coordination L5 — daemon-backed views over coordination state",
+  );
+  const getDeps = (): ViewDeps => depsOverride ?? {
+    lifecycleDeps: realDeps(),
+    clientFactory: (url: string) => new DaemonClient(url),
+  };
+
+  cmd
+    .command("list")
+    .description("List built-in + custom views")
+    .option("--json", "JSON output for agents")
+    .action(async (opts: { json?: boolean }) => {
+      const deps = getDeps();
+      await withClient(deps, async (client) => {
+        const res = await client.get<unknown>("/api/views/list");
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
+  cmd
+    .command("show <viewName>")
+    .description(
+      "Run a view (built-in or custom). Built-in views: recently-active, founder, pod-load, escalations, held, activity",
+    )
+    .option("--rig <rig>", "Filter by rig name (matches destination_session OR source_session @<rig>)")
+    .option("--limit <n>", "Result row limit", "100")
+    .option("--json", "JSON output for agents")
+    .action(async (viewName: string, opts: { rig?: string; limit: string; json?: boolean }) => {
+      const deps = getDeps();
+      const params = new URLSearchParams();
+      if (opts.rig) params.set("rig", opts.rig);
+      if (opts.limit) params.set("limit", opts.limit);
+      const query = params.toString();
+      await withClient(deps, async (client) => {
+        const path = `/api/views/${encodeURIComponent(viewName)}${query ? `?${query}` : ""}`;
+        const res = await client.get<unknown>(path);
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
+  cmd
+    .command("register")
+    .description("Register or update a custom view")
+    .requiredOption("--name <name>", "Custom view name (must NOT collide with built-in names)")
+    .requiredOption("--definition <sql>", "SQL definition (operator-supplied; not validated for taxonomy)")
+    .requiredOption("--session <session>", "Registering operator session")
+    .option("--json", "JSON output for agents")
+    .action(async (opts: { name: string; definition: string; session: string; json?: boolean }) => {
+      const deps = getDeps();
+      await withClient(deps, async (client) => {
+        const res = await client.post<unknown>("/api/views/custom/register", {
+          viewName: opts.name,
+          definition: opts.definition,
+          registeredBySession: opts.session,
+        });
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
+  return cmd;
+}

packages/cli/src/index.ts

@@ -27,6 +27,8 @@ import { transcriptCommand } from "./commands/transcript.js";
 import { sendCommand } from "./commands/send.js";
 import { streamCommand, type StreamDeps } from "./commands/stream.js";
 import { queueCommand, type QueueDeps } from "./commands/queue.js";
+import { projectCommand, type ProjectDeps } from "./commands/project.js";
+import { viewCommand, type ViewDeps } from "./commands/view.js";
 import { captureCommand } from "./commands/capture.js";
 import { broadcastCommand } from "./commands/broadcast.js";
 import { configCommand } from "./commands/config.js";
@@ -80,6 +82,8 @@ export interface ProgramDeps {
   sendDeps?: StatusDeps;
   streamDeps?: StreamDeps;
   queueDeps?: QueueDeps;
+  projectDeps?: ProjectDeps;
+  viewDeps?: ViewDeps;
   captureDeps?: StatusDeps;
   broadcastDeps?: StatusDeps;
   askDeps?: StatusDeps;
@@ -134,6 +138,8 @@ export function createProgram(depsOverride?: ProgramDeps): Command {
   program.addCommand(sendCommand(depsOverride?.sendDeps));
   program.addCommand(streamCommand(depsOverride?.streamDeps));
   program.addCommand(queueCommand(depsOverride?.queueDeps));
+  program.addCommand(projectCommand(depsOverride?.projectDeps));
+  program.addCommand(viewCommand(depsOverride?.viewDeps));
   program.addCommand(captureCommand(depsOverride?.captureDeps));
   program.addCommand(broadcastCommand(depsOverride?.broadcastDeps));
   program.addCommand(askCommand(depsOverride?.askDeps));

packages/cli/test/project.test.ts

@@ -0,0 +1,158 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import type { ProjectDeps } from "../src/commands/project.js";
+import { createProgram } from "../src/index.js";
+
+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: ProjectDeps; 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 ProjectDeps["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: {} };
+        }),
+        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<ProjectDeps["clientFactory"]>,
+    },
+  };
+}
+
+describe("rig project CLI (PL-004 Phase B)", () => {
+  let logs: string[];
+  let errors: string[];
+
+  beforeEach(() => {
+    logs = [];
+    errors = [];
+    vi.spyOn(console, "log").mockImplementation((...a) => logs.push(a.join(" ")));
+    vi.spyOn(console, "error").mockImplementation((...a) => errors.push(a.join(" ")));
+    process.exitCode = undefined;
+  });
+
+  it("project is registered with all R1 subcommands", () => {
+    const { deps } = makeDeps();
+    const program = createProgram({ projectDeps: deps });
+    const cmd = program.commands.find((c) => c.name() === "project");
+    expect(cmd).toBeDefined();
+    const subs = cmd!.commands.map((c) => c.name()).sort();
+    expect(subs).toContain("lease-acquire");
+    expect(subs).toContain("lease-heartbeat");
+    expect(subs).toContain("lease-show");
+    expect(subs).toContain("reclaim-classifier");
+    expect(subs).toContain("classify");
+    expect(subs).toContain("list");
+    expect(subs).toContain("show");
+  });
+
+  it("lease-acquire POSTs /api/projects/lease/acquire with classifierSession", async () => {
+    const { deps, calls } = makeDeps({
+      routes: { "POST /api/projects/lease/acquire": { status: 201, data: { state: "active" } } },
+    });
+    const program = createProgram({ projectDeps: deps });
+    program.exitOverride();
+    await program.parseAsync(["node", "rig", "project", "lease-acquire", "--session", "alice@rig", "--json"]);
+    const call = calls.find((c) => c.path === "/api/projects/lease/acquire");
+    expect((call!.body as { classifierSession: string }).classifierSession).toBe("alice@rig");
+  });
+
+  it("classify POSTs /api/projects/project with stream-item-id + classification fields", async () => {
+    const { deps, calls } = makeDeps({
+      routes: { "POST /api/projects/project": { status: 201, data: { projectId: "P-1" } } },
+    });
+    const program = createProgram({ projectDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "project", "classify", "stream-x",
+      "--session", "alice@rig",
+      "--type", "idea",
+      "--urgency", "high",
+      "--destination", "planning@rig",
+      "--json",
+    ]);
+    const call = calls.find((c) => c.path === "/api/projects/project");
+    const body = call!.body as Record<string, unknown>;
+    expect(body.streamItemId).toBe("stream-x");
+    expect(body.classifierSession).toBe("alice@rig");
+    expect(body.classificationType).toBe("idea");
+    expect(body.classificationUrgency).toBe("high");
+    expect(body.classificationDestination).toBe("planning@rig");
+  });
+
+  it("classify with idempotency_violation 409 surfaces error + non-zero exit", async () => {
+    const { deps } = makeDeps({
+      routes: {
+        "POST /api/projects/project": {
+          status: 409,
+          data: { error: "idempotency_violation", message: "stream-x already projected" },
+        },
+      },
+    });
+    const program = createProgram({ projectDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "project", "classify", "stream-x",
+      "--session", "alice@rig",
+      "--json",
+    ]);
+    expect(process.exitCode).toBe(1);
+    const out = logs.join("\n");
+    expect(out).toContain("idempotency_violation");
+  });
+
+  it("reclaim-classifier --if-dead passes ifDead: true", async () => {
+    const { deps, calls } = makeDeps({
+      routes: { "POST /api/projects/reclaim-classifier": { status: 200, data: { state: "reclaimed" } } },
+    });
+    const program = createProgram({ projectDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "project", "reclaim-classifier",
+      "--session", "operator@rig",
+      "--if-dead",
+      "--reason", "alice unresponsive",
+    ]);
+    const call = calls.find((c) => c.path === "/api/projects/reclaim-classifier");
+    const body = call!.body as Record<string, unknown>;
+    expect(body.byClassifierSession).toBe("operator@rig");
+    expect(body.ifDead).toBe(true);
+    expect(body.reason).toBe("alice unresponsive");
+  });
+
+  it("list constructs /api/projects/list with filter params", async () => {
+    const { deps, calls } = makeDeps({
+      routes: { "GET /api/projects/list?classificationDestination=planning%40rig&limit=50": { status: 200, data: [] } },
+    });
+    const program = createProgram({ projectDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "project", "list",
+      "--destination", "planning@rig",
+      "--limit", "50",
+      "--json",
+    ]);
+    const call = calls.find((c) => c.method === "GET" && c.path.startsWith("/api/projects/list"));
+    expect(call!.path).toContain("classificationDestination=planning%40rig");
+    expect(call!.path).toContain("limit=50");
+  });
+});

packages/cli/test/view.test.ts

@@ -0,0 +1,139 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import type { ViewDeps } from "../src/commands/view.js";
+import { createProgram } from "../src/index.js";
+
+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: ViewDeps; 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 ViewDeps["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: {} };
+        }),
+        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<ViewDeps["clientFactory"]>,
+    },
+  };
+}
+
+describe("rig view CLI (PL-004 Phase B)", () => {
+  let logs: string[];
+  let errors: string[];
+
+  beforeEach(() => {
+    logs = [];
+    errors = [];
+    vi.spyOn(console, "log").mockImplementation((...a) => logs.push(a.join(" ")));
+    vi.spyOn(console, "error").mockImplementation((...a) => errors.push(a.join(" ")));
+    process.exitCode = undefined;
+  });
+
+  it("view is registered with list/show/register subcommands", () => {
+    const { deps } = makeDeps();
+    const program = createProgram({ viewDeps: deps });
+    const cmd = program.commands.find((c) => c.name() === "view");
+    expect(cmd).toBeDefined();
+    const subs = cmd!.commands.map((c) => c.name()).sort();
+    expect(subs).toContain("list");
+    expect(subs).toContain("show");
+    expect(subs).toContain("register");
+  });
+
+  it("list GETs /api/views/list", async () => {
+    const { deps, calls } = makeDeps({
+      routes: { "GET /api/views/list": { status: 200, data: { builtIn: ["recently-active"], custom: [] } } },
+    });
+    const program = createProgram({ viewDeps: deps });
+    program.exitOverride();
+    await program.parseAsync(["node", "rig", "view", "list", "--json"]);
+    const call = calls.find((c) => c.path === "/api/views/list");
+    expect(call).toBeDefined();
+  });
+
+  it("show GETs /api/views/<viewName> with rig + limit query", async () => {
+    const { deps, calls } = makeDeps({
+      routes: { "GET /api/views/recently-active?rig=product-lab&limit=50": { status: 200, data: { rowCount: 0 } } },
+    });
+    const program = createProgram({ viewDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "view", "show", "recently-active",
+      "--rig", "product-lab",
+      "--limit", "50",
+      "--json",
+    ]);
+    const call = calls.find((c) => c.method === "GET" && c.path.startsWith("/api/views/recently-active"));
+    expect(call!.path).toContain("rig=product-lab");
+    expect(call!.path).toContain("limit=50");
+  });
+
+  it("show without query params constructs path without ?", async () => {
+    const { deps, calls } = makeDeps({
+      routes: { "GET /api/views/founder": { status: 200, data: { rowCount: 0 } } },
+    });
+    const program = createProgram({ viewDeps: deps });
+    program.exitOverride();
+    // Note: --limit has a default of "100"; so this test needs explicit limit empty OR we accept default.
+    await program.parseAsync(["node", "rig", "view", "show", "founder"]);
+    const call = calls.find((c) => c.method === "GET" && c.path.startsWith("/api/views/founder"));
+    expect(call).toBeDefined();
+    // Default limit appended.
+    expect(call!.path).toContain("limit=100");
+  });
+
+  it("show with view_not_found 404 surfaces error + non-zero exit", async () => {
+    const { deps } = makeDeps({
+      routes: {
+        "GET /api/views/nonexistent?limit=100": { status: 404, data: { error: "view_not_found" } },
+      },
+    });
+    const program = createProgram({ viewDeps: deps });
+    program.exitOverride();
+    await program.parseAsync(["node", "rig", "view", "show", "nonexistent"]);
+    expect(process.exitCode).toBe(1);
+    const out = logs.join("\n");
+    expect(out).toContain("view_not_found");
+  });
+
+  it("register POSTs /api/views/custom/register with name + definition + session", async () => {
+    const { deps, calls } = makeDeps({
+      routes: { "POST /api/views/custom/register": { status: 201, data: { viewId: "V-1" } } },
+    });
+    const program = createProgram({ viewDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node", "rig", "view", "register",
+      "--name", "my-view",
+      "--definition", "SELECT 1",
+      "--session", "operator@rig",
+      "--json",
+    ]);
+    const call = calls.find((c) => c.path === "/api/views/custom/register");
+    const body = call!.body as Record<string, unknown>;
+    expect(body.viewName).toBe("my-view");
+    expect(body.definition).toBe("SELECT 1");
+    expect(body.registeredBySession).toBe("operator@rig");
+  });
+});

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

@@ -0,0 +1,34 @@
+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.
+ *
+ * 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.
+ */
+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,
+      classification_type TEXT,
+      classification_urgency TEXT,
+      classification_maturity TEXT,
+      classification_confidence TEXT,
+      classification_destination TEXT,
+      action TEXT,
+      classifier_session TEXT NOT NULL,
+      ts_projected TEXT NOT NULL
+    );
+    CREATE INDEX IF NOT EXISTS idx_project_classifications_classifier_session ON project_classifications(classifier_session);
+    CREATE INDEX IF NOT EXISTS idx_project_classifications_destination ON project_classifications(classification_destination);
+    CREATE INDEX IF NOT EXISTS idx_project_classifications_ts_projected ON project_classifications(ts_projected);
+  `,
+};

packages/daemon/src/db/migrations/029_classifier_leases.ts

@@ -0,0 +1,42 @@
+import type { Migration } from "../migrate.js";
+
+/**
+ * Classifier leases (PL-004 Phase B; L2 founder-resolved single-writer lease).
+ *
+ * Per PRD § L2: agent-backed classifier with daemon-enforced single-writer
+ * lease (TTL-based) + deadness detection (via whoami-service / node-inventory)
+ * + operator-verb reclaim. Single-writer enforced via partial UNIQUE index
+ * on state='active' (SQLite >= 3.8 supports partial indexes; OpenRig's
+ * better-sqlite3 ships SQLite 3.45+ so this is safe).
+ *
+ * State enum: active | expired | reclaimed
+ *   active    — lease is current; classifier_session may project
+ *   expired   — TTL passed AND heartbeat went stale; lease no longer valid
+ *   reclaimed — operator-verb reclaim took the lease away from the previous holder
+ *
+ * Reclaim is OPERATOR-VERB ONLY: rig project --reclaim-classifier [--if-dead].
+ * Daemon does NOT auto-reclaim on its own evaluation. expired→active requires
+ * an explicit acquire call from a NEW classifier session (which gets a new
+ * lease_id row).
+ */
+export const classifierLeasesSchema: Migration = {
+  name: "029_classifier_leases.sql",
+  sql: `
+    CREATE TABLE IF NOT EXISTS classifier_leases (
+      lease_id TEXT PRIMARY KEY,
+      classifier_session TEXT NOT NULL,
+      acquired_at TEXT NOT NULL,
+      expires_at TEXT NOT NULL,
+      last_heartbeat TEXT NOT NULL,
+      state TEXT NOT NULL DEFAULT 'active',
+      reclaimed_by_session TEXT,
+      reclaim_reason TEXT
+    );
+    CREATE UNIQUE INDEX IF NOT EXISTS idx_classifier_leases_active_singleton
+      ON classifier_leases(state) WHERE state = 'active';
+    CREATE INDEX IF NOT EXISTS idx_classifier_leases_classifier_session
+      ON classifier_leases(classifier_session);
+    CREATE INDEX IF NOT EXISTS idx_classifier_leases_expires_at
+      ON classifier_leases(expires_at) WHERE state = 'active';
+  `,
+};

packages/daemon/src/db/migrations/030_views_custom.ts

@@ -0,0 +1,28 @@
+import type { Migration } from "../migrate.js";
+
+/**
+ * Custom views (PL-004 Phase B; L5 View — custom-view registration table).
+ *
+ * Per PRD § L5: built-in views (`recently-active`, `founder`, `pod-load`,
+ * `escalations`, `held`, `activity`) are HARDCODED in view-projector.ts.
+ * Operator-defined custom views are stored in this table and registered at
+ * daemon startup from `~/.openrig/views.yaml` (or via API at runtime).
+ *
+ * Built-in views are NOT inserted into this table — view-projector.ts
+ * exposes them by name without DB lookup. Custom views are looked up here
+ * when the operator queries `rig view show <custom-name>`.
+ */
+export const viewsCustomSchema: Migration = {
+  name: "030_views_custom.sql",
+  sql: `
+    CREATE TABLE IF NOT EXISTS views_custom (
+      view_id TEXT PRIMARY KEY,
+      view_name TEXT NOT NULL UNIQUE,
+      definition TEXT NOT NULL,
+      registered_by_session TEXT NOT NULL,
+      registered_at TEXT NOT NULL,
+      last_evaluated_at TEXT
+    );
+    CREATE INDEX IF NOT EXISTS idx_views_custom_registered_by ON views_custom(registered_by_session);
+  `,
+};

packages/daemon/src/domain/classifier-lease-manager.ts

@@ -0,0 +1,392 @@
+import type Database from "better-sqlite3";
+import { ulid } from "ulid";
+import type { EventBus } from "./event-bus.js";
+import type { PersistedEvent } from "./types.js";
+
+/**
+ * Classifier lease manager (PL-004 Phase B; L2 founder-resolved).
+ *
+ * Daemon-enforced single-writer lease for the project (classifier) primitive.
+ * Per PRD § L2 hard rule + slice IMPL § Guard Checkpoint Focus item 2:
+ *
+ * - Single-writer: at most one lease in `state='active'` at any time, enforced
+ *   via partial UNIQUE index `idx_classifier_leases_active_singleton`.
+ * - TTL-based expiry: every lease has an `expires_at` (acquired_at + ttlMs).
+ * - Heartbeat: `last_heartbeat` is updated by the lease holder. Stale heartbeat
+ *   past TTL signals deadness.
+ * - Deadness detection: caller of `evaluateDeadness` is the project verb path
+ *   OR a watchdog. The manager itself does NOT auto-reclaim — only marks
+ *   `expired` when heartbeat is stale.
+ * - Reclaim is OPERATOR-VERB ONLY: `rig project --reclaim-classifier
+ *   [--if-dead]`. Daemon does NOT auto-reclaim. The reclaim path takes the
+ *   active lease away from the previous holder and emits classifier.reclaimed.
+ *
+ * Pattern mirrors Phase A's hot-potato-enforcer.ts shape: pure validation +
+ * lifecycle methods, no Hono. Routes import this; this does not import routes.
+ */
+
+export const LEASE_STATES = ["active", "expired", "reclaimed"] as const;
+export type LeaseState = (typeof LEASE_STATES)[number];
+
+export interface ClassifierLease {
+  leaseId: string;
+  classifierSession: string;
+  acquiredAt: string;
+  expiresAt: string;
+  lastHeartbeat: string;
+  state: LeaseState;
+  reclaimedBySession: string | null;
+  reclaimReason: string | null;
+}
+
+interface ClassifierLeaseRow {
+  lease_id: string;
+  classifier_session: string;
+  acquired_at: string;
+  expires_at: string;
+  last_heartbeat: string;
+  state: string;
+  reclaimed_by_session: string | null;
+  reclaim_reason: string | null;
+}
+
+export class ClassifierLeaseError extends Error {
+  readonly code: string;
+  readonly meta: Record<string, unknown> | undefined;
+  constructor(code: string, message: string, meta?: Record<string, unknown>) {
+    super(message);
+    this.code = code;
+    this.meta = meta;
+  }
+}
+
+/**
+ * Default lease TTL: 15 minutes. Per PRD: "TTL-based"; concrete value is
+ * implementation choice. 15 min is the operator-friendly midpoint between
+ * stale-detection latency (longer = slower deadness signal) and unnecessary
+ * heartbeat traffic (shorter = more wake-ups). Configurable via constructor
+ * option for tests + future tuning.
+ */
+const DEFAULT_LEASE_TTL_MS = 15 * 60 * 1000;
+
+export interface ClassifierLeaseManagerOptions {
+  ttlMs?: number;
+  /** For tests: inject a deterministic clock. Defaults to `() => new Date()`. */
+  now?: () => Date;
+  /**
+   * Liveness check (per PRD: via whoami-service / node-inventory). Returns
+   * true if the session is still alive. The lease-manager calls this when
+   * evaluating deadness; if the function returns false AND the lease's
+   * heartbeat is stale, the lease is marked expired.
+   *
+   * Defaults to `() => true` (no liveness check; tests can stub).
+   */
+  isAlive?: (classifierSession: string) => boolean;
+}
+
+export class ClassifierLeaseManager {
+  readonly db: Database.Database;
+  private readonly eventBus: EventBus;
+  private readonly ttlMs: number;
+  private readonly now: () => Date;
+  private isAlive: (classifierSession: string) => boolean;
+
+  constructor(
+    db: Database.Database,
+    eventBus: EventBus,
+    opts?: ClassifierLeaseManagerOptions,
+  ) {
+    this.db = db;
+    this.eventBus = eventBus;
+    this.ttlMs = opts?.ttlMs ?? DEFAULT_LEASE_TTL_MS;
+    this.now = opts?.now ?? (() => new Date());
+    this.isAlive = opts?.isAlive ?? (() => true);
+  }
+
+  /**
+   * Wire post-construction liveness check (used by startup.ts when
+   * whoami-service is constructed later in the dep graph).
+   */
+  attachIsAlive(check: (classifierSession: string) => boolean): void {
+    this.isAlive = check;
+  }
+
+  /**
+   * Acquire the active lease for a classifier session. Fails with
+   * `lease_held` (409) if another session currently holds an active lease
+   * AND that session is alive. If the held lease is by a dead session, the
+   * caller should first invoke `evaluateDeadness` (which marks it expired)
+   * or use the operator reclaim path.
+   *
+   * Idempotent for the SAME classifier_session: re-calling acquire on the
+   * lease holder's behalf returns the existing lease unchanged (heartbeat
+   * is updated separately via `heartbeat`).
+   */
+  acquire(classifierSession: string): ClassifierLease {
+    const active = this.getActiveLease();
+    if (active) {
+      if (active.classifierSession === classifierSession) {
+        // Idempotent re-acquire by current holder.
+        return active;
+      }
+      throw new ClassifierLeaseError(
+        "lease_held",
+        `classifier lease is held by ${active.classifierSession} until ${active.expiresAt}; reclaim via 'rig project --reclaim-classifier' if needed`,
+        { holder: active.classifierSession, expiresAt: active.expiresAt },
+      );
+    }
+
+    const leaseId = ulid();
+    const acquiredAt = this.now().toISOString();
+    const expiresAt = new Date(this.now().getTime() + this.ttlMs).toISOString();
+
+    const txn = this.db.transaction(() => {
+      this.db
+        .prepare(
+          `INSERT INTO classifier_leases (
+            lease_id, classifier_session, acquired_at, expires_at,
+            last_heartbeat, state
+          ) VALUES (?, ?, ?, ?, ?, 'active')`
+        )
+        .run(leaseId, classifierSession, acquiredAt, expiresAt, acquiredAt);
+
+      return this.eventBus.persistWithinTransaction({
+        type: "classifier.lease_acquired",
+        leaseId,
+        classifierSession,
+        acquiredAt,
+        expiresAt,
+      });
+    });
+
+    const persisted = txn();
+    this.eventBus.notifySubscribers(persisted);
+    return this.getByIdOrThrow(leaseId);
+  }
+
+  /**
+   * Heartbeat from the lease holder. Updates `last_heartbeat` and extends
+   * `expires_at` by the TTL (sliding-window TTL semantics).
+   */
+  heartbeat(leaseId: string, classifierSession: string): ClassifierLease {
+    const lease = this.getById(leaseId);
+    if (!lease) {
+      throw new ClassifierLeaseError("lease_not_found", `lease ${leaseId} not found`);
+    }
+    if (lease.classifierSession !== classifierSession) {
+      throw new ClassifierLeaseError(
+        "lease_session_mismatch",
+        `lease ${leaseId} is held by ${lease.classifierSession}, not ${classifierSession}`,
+      );
+    }
+    if (lease.state !== "active") {
+      throw new ClassifierLeaseError(
+        "lease_not_active",
+        `lease ${leaseId} is in state ${lease.state}; cannot heartbeat`,
+      );
+    }
+
+    const now = this.now();
+    const lastHeartbeat = now.toISOString();
+    const expiresAt = new Date(now.getTime() + this.ttlMs).toISOString();
+
+    this.db
+      .prepare(
+        `UPDATE classifier_leases
+           SET last_heartbeat = ?, expires_at = ?
+         WHERE lease_id = ?`
+      )
+      .run(lastHeartbeat, expiresAt, leaseId);
+
+    return this.getByIdOrThrow(leaseId);
+  }
+
+  /**
+   * Evaluate deadness for the currently-active lease. If the lease's
+   * heartbeat is stale (now > expires_at) OR the holder is reported dead
+   * by `isAlive`, mark the lease expired and emit classifier.lease_expired
+   * + classifier.dead. Returns the (now-expired) lease, or null if no
+   * active lease exists or it remains alive.
+   *
+   * Called by: project verb path (before acquire, to clear dead leases),
+   * watchdog (periodic sweep), or operator-driven check.
+   *
+   * Per PRD: this method does NOT reclaim — it only marks expired. The
+   * next `acquire` call by ANY session can then succeed (since the partial
+   * UNIQUE on state='active' is now empty).
+   */
+  evaluateDeadness(): ClassifierLease | null {
+    const active = this.getActiveLease();
+    if (!active) return null;
+    const nowIso = this.now().toISOString();
+    const ttlPassed = nowIso > active.expiresAt;
+    const sessionDead = !this.isAlive(active.classifierSession);
+    if (!ttlPassed && !sessionDead) return null;
+
+    const events: PersistedEvent[] = [];
+    const txn = this.db.transaction(() => {
+      this.db
+        .prepare(
+          `UPDATE classifier_leases SET state = 'expired' WHERE lease_id = ? AND state = 'active'`,
+        )
+        .run(active.leaseId);
+
+      events.push(this.eventBus.persistWithinTransaction({
+        type: "classifier.lease_expired",
+        leaseId: active.leaseId,
+        classifierSession: active.classifierSession,
+        expiredAt: nowIso,
+      }));
+
+      if (sessionDead) {
+        events.push(this.eventBus.persistWithinTransaction({
+          type: "classifier.dead",
+          leaseId: active.leaseId,
+          classifierSession: active.classifierSession,
+          lastHeartbeat: active.lastHeartbeat,
+          detectedAt: nowIso,
+        }));
+      }
+    });
+
+    txn();
+    for (const e of events) this.eventBus.notifySubscribers(e);
+    return this.getByIdOrThrow(active.leaseId);
+  }
+
+  /**
+   * Operator-verb reclaim. Per PRD § L2 hard rule: ONLY this path may take
+   * an active lease away from its holder. Daemon does NOT auto-reclaim.
+   *
+   * - If `ifDead`: only succeed when isAlive(holder) returns false. If the
+   *   holder is alive, refuse with `lease_still_active`.
+   * - If `!ifDead`: take the lease unconditionally.
+   *
+   * Marks the active lease state='reclaimed', records reclaimed_by_session
+   * + reclaim_reason, and emits classifier.reclaimed. The next acquire
+   * (by ANY session) can then succeed.
+   */
+  reclaim(byClassifierSession: string, opts?: { ifDead?: boolean; reason?: string }): ClassifierLease {
+    const active = this.getActiveLease();
+    if (!active) {
+      throw new ClassifierLeaseError(
+        "no_active_lease",
+        "no active classifier lease to reclaim",
+      );
+    }
+    if (opts?.ifDead === true && this.isAlive(active.classifierSession)) {
+      throw new ClassifierLeaseError(
+        "lease_still_active",
+        `classifier lease holder ${active.classifierSession} is still alive; --if-dead refuses to reclaim`,
+        { holder: active.classifierSession },
+      );
+    }
+    const reason = opts?.reason ?? (opts?.ifDead ? "operator-reclaim --if-dead" : "operator-reclaim");
+    const reclaimedAt = this.now().toISOString();
+
+    const txn = this.db.transaction(() => {
+      this.db
+        .prepare(
+          `UPDATE classifier_leases
+             SET state = 'reclaimed',
+                 reclaimed_by_session = ?,
+                 reclaim_reason = ?
+           WHERE lease_id = ? AND state = 'active'`,
+        )
+        .run(byClassifierSession, reason, active.leaseId);
+
+      return this.eventBus.persistWithinTransaction({
+        type: "classifier.reclaimed",
+        leaseId: active.leaseId,
+        previousClassifierSession: active.classifierSession,
+        reclaimedBySession: byClassifierSession,
+        reason,
+        reclaimedAt,
+      });
+    });
+
+    const persisted = txn();
+    this.eventBus.notifySubscribers(persisted);
+    return this.getByIdOrThrow(active.leaseId);
+  }
+
+  /**
+   * Validation hook used by project-classifier: returns the active lease
+   * iff the supplied session holds it, else throws. Centralizes the
+   * "must hold the lease to project" check.
+   */
+  requireActiveHolder(classifierSession: string): ClassifierLease {
+    const active = this.getActiveLease();
+    if (!active) {
+      throw new ClassifierLeaseError(
+        "no_active_lease",
+        "no active classifier lease; call acquire first",
+      );
+    }
+    if (active.classifierSession !== classifierSession) {
+      throw new ClassifierLeaseError(
+        "lease_held",
+        `classifier lease is held by ${active.classifierSession}, not ${classifierSession}`,
+        { holder: active.classifierSession },
+      );
+    }
+    if (active.expiresAt < this.now().toISOString()) {
+      throw new ClassifierLeaseError(
+        "lease_expired",
+        `classifier lease for ${classifierSession} expired at ${active.expiresAt}`,
+      );
+    }
+    return active;
+  }
+
+  getActiveLease(): ClassifierLease | null {
+    const row = this.db
+      .prepare(`SELECT * FROM classifier_leases WHERE state = 'active' LIMIT 1`)
+      .get() as ClassifierLeaseRow | undefined;
+    return row ? this.rowToLease(row) : null;
+  }
+
+  getById(leaseId: string): ClassifierLease | null {
+    const row = this.db
+      .prepare(`SELECT * FROM classifier_leases WHERE lease_id = ?`)
+      .get(leaseId) as ClassifierLeaseRow | undefined;
+    return row ? this.rowToLease(row) : null;
+  }
+
+  list(opts?: { classifierSession?: string; limit?: number }): ClassifierLease[] {
+    const limit = opts?.limit ?? 100;
+    const conditions: string[] = [];
+    const params: unknown[] = [];
+    if (opts?.classifierSession) {
+      conditions.push("classifier_session = ?");
+      params.push(opts.classifierSession);
+    }
+    const where = conditions.length > 0 ? `WHERE ${conditions.join(" AND ")}` : "";
+    params.push(limit);
+    const rows = this.db
+      .prepare(`SELECT * FROM classifier_leases ${where} ORDER BY acquired_at DESC LIMIT ?`)
+      .all(...params) as ClassifierLeaseRow[];
+    return rows.map((r) => this.rowToLease(r));
+  }
+
+  private getByIdOrThrow(leaseId: string): ClassifierLease {
+    const lease = this.getById(leaseId);
+    if (!lease) {
+      throw new ClassifierLeaseError("lease_not_found", `lease ${leaseId} not found after write`);
+    }
+    return lease;
+  }
+
+  private rowToLease(row: ClassifierLeaseRow): ClassifierLease {
+    return {
+      leaseId: row.lease_id,
+      classifierSession: row.classifier_session,
+      acquiredAt: row.acquired_at,
+      expiresAt: row.expires_at,
+      lastHeartbeat: row.last_heartbeat,
+      state: row.state as LeaseState,
+      reclaimedBySession: row.reclaimed_by_session,
+      reclaimReason: row.reclaim_reason,
+    };
+  }
+}

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

@@ -0,0 +1,222 @@
+import type Database from "better-sqlite3";
+import { ulid } from "ulid";
+import type { EventBus } from "./event-bus.js";
+import type { ClassifierLeaseManager } from "./classifier-lease-manager.js";
+
+/**
+ * Project classifier (PL-004 Phase B; L2 Project / Classifier — write path).
+ *
+ * Per PRD § L2 + slice IMPL § Guard Checkpoint Focus item 1+2+4:
+ * - Lease validation via classifier-lease-manager (single-writer contract).
+ * - Idempotency via UNIQUE constraint on stream_item_id (re-projection → 409).
+ * - Daemon does NOT enforce taxonomies on classification fields — those are
+ *   agent-authoritative per the founder direction. Daemon only owns the
+ *   contract (lease + idempotency + reclaim).
+ * - Emits project.classified event after the row is committed.
+ *
+ * Pattern mirrors Phase A's stream-store.ts shape (single class, atomic
+ * transactions, persist-event-then-notify). No Hono.
+ */
+
+export interface ProjectClassification {
+  projectId: string;
+  streamItemId: string;
+  classificationType: string | null;
+  classificationUrgency: string | null;
+  classificationMaturity: string | null;
+  classificationConfidence: string | null;
+  classificationDestination: string | null;
+  action: string | null;
+  classifierSession: string;
+  tsProjected: string;
+}
+
+export interface ProjectClassifyInput {
+  streamItemId: string;
+  classifierSession: string;
+  classificationType?: string;
+  classificationUrgency?: string;
+  classificationMaturity?: string;
+  classificationConfidence?: string;
+  classificationDestination?: string;
+  action?: string;
+}
+
+export interface ProjectListOptions {
+  classifierSession?: string;
+  classificationDestination?: string;
+  limit?: number;
+}
+
+interface ProjectClassificationRow {
+  project_id: string;
+  stream_item_id: string;
+  classification_type: string | null;
+  classification_urgency: string | null;
+  classification_maturity: string | null;
+  classification_confidence: string | null;
+  classification_destination: string | null;
+  action: string | null;
+  classifier_session: string;
+  ts_projected: string;
+}
+
+export class ProjectClassifierError extends Error {
+  readonly code: string;
+  readonly meta: Record<string, unknown> | undefined;
+  constructor(code: string, message: string, meta?: Record<string, unknown>) {
+    super(message);
+    this.code = code;
+    this.meta = meta;
+  }
+}
+
+export class ProjectClassifier {
+  readonly db: Database.Database;
+  private readonly eventBus: EventBus;
+  private readonly leaseManager: ClassifierLeaseManager;
+  private readonly now: () => Date;
+
+  constructor(
+    db: Database.Database,
+    eventBus: EventBus,
+    leaseManager: ClassifierLeaseManager,
+    opts?: { now?: () => Date },
+  ) {
+    this.db = db;
+    this.eventBus = eventBus;
+    this.leaseManager = leaseManager;
+    this.now = opts?.now ?? (() => new Date());
+  }
+
+  /**
+   * Project a stream item — agent-authored classification, daemon-enforced
+   * idempotency + lease.
+   *
+   * 1. Verify the caller holds the active lease (delegates to leaseManager).
+   * 2. INSERT project_classifications row (UNIQUE on stream_item_id catches
+   *    re-projection attempts; we map the constraint violation to a clean
+   *    `idempotency_violation` error).
+   * 3. Emit project.classified event.
+   */
+  classify(input: ProjectClassifyInput): ProjectClassification {
+    // Step 1: lease check. Throws ClassifierLeaseError on no_active_lease,
+    // lease_held by other session, or lease_expired.
+    this.leaseManager.requireActiveHolder(input.classifierSession);
+
+    // 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
+      .prepare(`SELECT * FROM project_classifications WHERE stream_item_id = ?`)
+      .get(input.streamItemId) as ProjectClassificationRow | undefined;
+    if (existing) {
+      throw new ProjectClassifierError(
+        "idempotency_violation",
+        `stream_item_id ${input.streamItemId} is already projected as project_id ${existing.project_id} by ${existing.classifier_session}`,
+        { existingProjectId: existing.project_id, existingClassifier: existing.classifier_session },
+      );
+    }
+
+    const projectId = ulid();
+    const tsProjected = this.now().toISOString();
+
+    const txn = this.db.transaction(() => {
+      this.db
+        .prepare(
+          `INSERT INTO project_classifications (
+            project_id, stream_item_id,
+            classification_type, classification_urgency, classification_maturity,
+            classification_confidence, classification_destination, action,
+            classifier_session, ts_projected
+          ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
+        )
+        .run(
+          projectId,
+          input.streamItemId,
+          input.classificationType ?? null,
+          input.classificationUrgency ?? null,
+          input.classificationMaturity ?? null,
+          input.classificationConfidence ?? null,
+          input.classificationDestination ?? null,
+          input.action ?? null,
+          input.classifierSession,
+          tsProjected,
+        );
+
+      return this.eventBus.persistWithinTransaction({
+        type: "project.classified",
+        projectId,
+        streamItemId: input.streamItemId,
+        classifierSession: input.classifierSession,
+        classificationType: input.classificationType ?? null,
+        classificationDestination: input.classificationDestination ?? null,
+      });
+    });
+
+    const persisted = txn();
+    this.eventBus.notifySubscribers(persisted);
+    return this.getByIdOrThrow(projectId);
+  }
+
+  getById(projectId: string): ProjectClassification | null {
+    const row = this.db
+      .prepare(`SELECT * FROM project_classifications WHERE project_id = ?`)
+      .get(projectId) as ProjectClassificationRow | undefined;
+    return row ? this.rowToProject(row) : null;
+  }
+
+  getByStreamItemId(streamItemId: string): ProjectClassification | null {
+    const row = this.db
+      .prepare(`SELECT * FROM project_classifications WHERE stream_item_id = ?`)
+      .get(streamItemId) as ProjectClassificationRow | undefined;
+    return row ? this.rowToProject(row) : null;
+  }
+
+  list(opts?: ProjectListOptions): ProjectClassification[] {
+    const limit = opts?.limit ?? 100;
+    const conditions: string[] = [];
+    const params: unknown[] = [];
+    if (opts?.classifierSession) {
+      conditions.push("classifier_session = ?");
+      params.push(opts.classifierSession);
+    }
+    if (opts?.classificationDestination) {
+      conditions.push("classification_destination = ?");
+      params.push(opts.classificationDestination);
+    }
+    const where = conditions.length > 0 ? `WHERE ${conditions.join(" AND ")}` : "";
+    params.push(limit);
+    const rows = this.db
+      .prepare(
+        `SELECT * FROM project_classifications ${where} ORDER BY ts_projected DESC LIMIT ?`,
+      )
+      .all(...params) as ProjectClassificationRow[];
+    return rows.map((r) => this.rowToProject(r));
+  }
+
+  private getByIdOrThrow(projectId: string): ProjectClassification {
+    const p = this.getById(projectId);
+    if (!p) {
+      throw new ProjectClassifierError(
+        "project_not_found",
+        `project ${projectId} not found after write`,
+      );
+    }
+    return p;
+  }
+
+  private rowToProject(row: ProjectClassificationRow): ProjectClassification {
+    return {
+      projectId: row.project_id,
+      streamItemId: row.stream_item_id,
+      classificationType: row.classification_type,
+      classificationUrgency: row.classification_urgency,
+      classificationMaturity: row.classification_maturity,
+      classificationConfidence: row.classification_confidence,
+      classificationDestination: row.classification_destination,
+      action: row.action,
+      classifierSession: row.classifier_session,
+      tsProjected: row.ts_projected,
+    };
+  }
+}

packages/daemon/src/domain/types.ts

@@ -154,7 +154,19 @@ export type RigEvent =
   | { type: "qitem.fallback_routed"; qitemId: string; originalDestination: string; rerouteDestination: string; reason: string }
   | { type: "qitem.closure_overdue"; qitemId: string; destinationSession: string; closureRequiredAt: string; overdueSince: string }
   | { type: "inbox.absorbed"; inboxId: string; destinationSession: string; senderSession: string; promotedQitemId: string }
-  | { type: "inbox.denied"; inboxId: string; destinationSession: string; senderSession: string; reason: string };
+  | { type: "inbox.denied"; inboxId: string; destinationSession: string; senderSession: string; reason: string }
+  // Coordination primitive (PL-004 Phase B) — project (classifier) / view.
+  // project.classified: emitted when a stream item is successfully projected.
+  // classifier.lease_*: lifecycle of the daemon-enforced single-writer lease.
+  // classifier.dead: heartbeat absence past TTL detected (deadness inference).
+  // classifier.reclaimed: operator-verb reclaim took the lease.
+  // view.changed: a view's projection result-set changed (SSE consumers see deltas).
+  | { type: "project.classified"; projectId: string; streamItemId: string; classifierSession: string; classificationType: string | null; classificationDestination: string | null }
+  | { type: "classifier.lease_acquired"; leaseId: string; classifierSession: string; acquiredAt: string; expiresAt: string }
+  | { type: "classifier.lease_expired"; leaseId: string; classifierSession: string; expiredAt: string }
+  | { type: "classifier.dead"; leaseId: string; classifierSession: string; lastHeartbeat: string; detectedAt: string }
+  | { type: "classifier.reclaimed"; leaseId: string; previousClassifierSession: string; reclaimedBySession: string; reason: string; reclaimedAt: string }
+  | { type: "view.changed"; viewName: string; cause: string };
 
 export type PersistedEvent = RigEvent & {
   seq: number;

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

@@ -0,0 +1,350 @@
+import type Database from "better-sqlite3";
+import { ulid } from "ulid";
+import type { EventBus } from "./event-bus.js";
+
+/**
+ * View projector (PL-004 Phase B; L5 View — read-only projections).
+ *
+ * Per PRD § L5 + slice IMPL § Guard Checkpoint Focus item 5+6:
+ * - 6 built-in views over Phase A's queue_items + queue_transitions tables.
+ * - Custom view registration via views_custom table.
+ * - Read-only over Phase A state + Phase B state. NO writes to queue_items
+ *   or queue_transitions from this module.
+ * - Sub-100 ms latency target (per PRD § Acceptance Criteria); achieved by
+ *   leaning on Phase A's existing indexes.
+ * - Fixture rig exclusion default: rig names matching `^test-` or `^fixture-`
+ *   are excluded; opt-in via OPENRIG_VIEW_INCLUDE_FIXTURES=1.
+ *
+ * Custom views in views_custom store a SQL string (`definition`) that the
+ * projector executes verbatim. Operator-defined; no taxonomy enforced.
+ *
+ * Pattern mirrors Phase A's queue-repository.ts read-API shape.
+ */
+
+export const BUILT_IN_VIEW_NAMES = [
+  "recently-active",
+  "founder",
+  "pod-load",
+  "escalations",
+  "held",
+  "activity",
+] as const;
+
+export type BuiltInViewName = (typeof BUILT_IN_VIEW_NAMES)[number];
+
+export interface ViewQueryResult {
+  viewName: string;
+  generatedAt: string;
+  rows: Record<string, unknown>[];
+  rowCount: number;
+}
+
+export interface CustomView {
+  viewId: string;
+  viewName: string;
+  definition: string;
+  registeredBySession: string;
+  registeredAt: string;
+  lastEvaluatedAt: string | null;
+}
+
+interface CustomViewRow {
+  view_id: string;
+  view_name: string;
+  definition: string;
+  registered_by_session: string;
+  registered_at: string;
+  last_evaluated_at: string | null;
+}
+
+export class ViewProjectorError extends Error {
+  readonly code: string;
+  constructor(code: string, message: string) {
+    super(message);
+    this.code = code;
+  }
+}
+
+/**
+ * Detect fixture rigs by session name suffix `@<rig>` where <rig> starts
+ * with `test-` or `fixture-`. Used to filter views by default; opt-in via
+ * OPENRIG_VIEW_INCLUDE_FIXTURES=1.
+ */
+function fixtureExclusionClause(): string {
+  if (process.env.OPENRIG_VIEW_INCLUDE_FIXTURES === "1") return "1=1";
+  // Exclude qitems whose source_session OR destination_session has a rig
+  // name starting with test- or fixture-.
+  return `(
+    destination_session NOT LIKE '%@test-%' AND
+    destination_session NOT LIKE '%@fixture-%' AND
+    source_session NOT LIKE '%@test-%' AND
+    source_session NOT LIKE '%@fixture-%'
+  )`;
+}
+
+export class ViewProjector {
+  readonly db: Database.Database;
+  private readonly eventBus: EventBus;
+  private readonly now: () => Date;
+
+  constructor(
+    db: Database.Database,
+    eventBus: EventBus,
+    opts?: { now?: () => Date },
+  ) {
+    this.db = db;
+    this.eventBus = eventBus;
+    this.now = opts?.now ?? (() => new Date());
+  }
+
+  /**
+   * Run a view by name. Built-in names (BUILT_IN_VIEW_NAMES) dispatch to
+   * hardcoded SQL; other names dispatch to custom-view lookup.
+   */
+  show(viewName: string, opts?: { rig?: string; limit?: number }): ViewQueryResult {
+    const limit = Math.max(1, Math.min(opts?.limit ?? 100, 1000));
+    if ((BUILT_IN_VIEW_NAMES as readonly string[]).includes(viewName)) {
+      return this.runBuiltIn(viewName as BuiltInViewName, opts?.rig, limit);
+    }
+    const custom = this.getCustomView(viewName);
+    if (!custom) {
+      throw new ViewProjectorError("view_not_found", `view '${viewName}' is not registered (built-in or custom)`);
+    }
+    return this.runCustom(custom, limit);
+  }
+
+  list(): { builtIn: BuiltInViewName[]; custom: CustomView[] } {
+    return { builtIn: [...BUILT_IN_VIEW_NAMES], custom: this.listCustomViews() };
+  }
+
+  private runBuiltIn(name: BuiltInViewName, rig: string | undefined, limit: number): ViewQueryResult {
+    const fixtureClause = fixtureExclusionClause();
+    const rigClause = rig ? `AND (destination_session LIKE ? OR source_session LIKE ?)` : "";
+    const rigParams: unknown[] = rig ? [`%@${rig}`, `%@${rig}`] : [];
+
+    let sql: string;
+    let params: unknown[] = [];
+    switch (name) {
+      case "recently-active":
+        // qitems by ts_updated DESC; live states (pending/in-progress/blocked).
+        sql = `
+          SELECT qitem_id, source_session, destination_session, state, priority, tier, ts_updated, body
+          FROM queue_items
+          WHERE state IN ('pending', 'in-progress', 'blocked')
+            AND ${fixtureClause}
+            ${rigClause}
+          ORDER BY ts_updated DESC
+          LIMIT ?
+        `;
+        params = [...rigParams, limit];
+        break;
+      case "founder":
+        // priority='critical' OR tier='critical' OR tier='fast'.
+        sql = `
+          SELECT qitem_id, source_session, destination_session, state, priority, tier, ts_updated
+          FROM queue_items
+          WHERE (priority = 'critical' OR tier IN ('critical', 'fast'))
+            AND state IN ('pending', 'in-progress', 'blocked')
+            AND ${fixtureClause}
+            ${rigClause}
+          ORDER BY ts_updated DESC
+          LIMIT ?
+        `;
+        params = [...rigParams, limit];
+        break;
+      case "pod-load":
+        // Per-destination qitem counts (active states only).
+        sql = `
+          SELECT destination_session AS pod, COUNT(*) AS active_count
+          FROM queue_items
+          WHERE state IN ('pending', 'in-progress', 'blocked')
+            AND ${fixtureClause}
+            ${rigClause}
+          GROUP BY destination_session
+          ORDER BY active_count DESC
+          LIMIT ?
+        `;
+        params = [...rigParams, limit];
+        break;
+      case "escalations":
+        // closure_reason = 'escalation' OR transition note matches.
+        sql = `
+          SELECT qitem_id, source_session, destination_session, state, closure_reason, closure_target, ts_updated
+          FROM queue_items
+          WHERE closure_reason = 'escalation'
+            AND ${fixtureClause}
+            ${rigClause}
+          ORDER BY ts_updated DESC
+          LIMIT ?
+        `;
+        params = [...rigParams, limit];
+        break;
+      case "held":
+        // state = 'blocked' OR blocked_on is non-null.
+        sql = `
+          SELECT qitem_id, source_session, destination_session, state, blocked_on, ts_updated, body
+          FROM queue_items
+          WHERE (state = 'blocked' OR blocked_on IS NOT NULL)
+            AND ${fixtureClause}
+            ${rigClause}
+          ORDER BY ts_updated DESC
+          LIMIT ?
+        `;
+        params = [...rigParams, limit];
+        break;
+      case "activity": {
+        // Recent transitions joined with qitem state. Fixture-clause must
+        // reference queue_items columns, so use the prefixed clause.
+        const fixtureClauseQI = fixtureClause === "1=1"
+          ? "1=1"
+          : `(
+              q.destination_session NOT LIKE '%@test-%' AND
+              q.destination_session NOT LIKE '%@fixture-%' AND
+              q.source_session NOT LIKE '%@test-%' AND
+              q.source_session NOT LIKE '%@fixture-%'
+            )`;
+        const rigClauseQI = rig
+          ? `AND (q.destination_session LIKE ? OR q.source_session LIKE ?)`
+          : "";
+        sql = `
+          SELECT t.transition_id, t.qitem_id, t.ts, t.state, t.actor_session, t.transition_note,
+                 q.destination_session, q.source_session
+          FROM queue_transitions t
+          JOIN queue_items q ON t.qitem_id = q.qitem_id
+          WHERE ${fixtureClauseQI}
+            ${rigClauseQI}
+          ORDER BY t.ts DESC
+          LIMIT ?
+        `;
+        params = [...rigParams, limit];
+        break;
+      }
+    }
+
+    const rows = this.db.prepare(sql).all(...params) as Record<string, unknown>[];
+    return {
+      viewName: name,
+      generatedAt: this.now().toISOString(),
+      rows,
+      rowCount: rows.length,
+    };
+  }
+
+  private runCustom(view: CustomView, limit: number): ViewQueryResult {
+    // Custom view definitions are operator-supplied SQL. Append LIMIT if
+    // the operator's SQL does not already include one. We do not parse SQL;
+    // operators are responsible for the definition's correctness.
+    const sql = view.definition.toLowerCase().includes("limit")
+      ? view.definition
+      : `${view.definition.trim().replace(/;$/, "")} LIMIT ${limit}`;
+    let rows: Record<string, unknown>[];
+    try {
+      rows = this.db.prepare(sql).all() as Record<string, unknown>[];
+    } catch (err) {
+      throw new ViewProjectorError(
+        "view_query_failed",
+        `custom view '${view.viewName}' query failed: ${(err as Error).message}`,
+      );
+    }
+    // Touch last_evaluated_at (best-effort; not transactional).
+    this.db
+      .prepare(`UPDATE views_custom SET last_evaluated_at = ? WHERE view_id = ?`)
+      .run(this.now().toISOString(), view.viewId);
+    return {
+      viewName: view.viewName,
+      generatedAt: this.now().toISOString(),
+      rows,
+      rowCount: rows.length,
+    };
+  }
+
+  /**
+   * Register a custom view. UNIQUE on view_name; re-registration of the
+   * same name updates the definition (operator-friendly: edit views.yaml,
+   * re-register, get the new query).
+   */
+  registerCustomView(input: {
+    viewName: string;
+    definition: string;
+    registeredBySession: string;
+  }): CustomView {
+    if ((BUILT_IN_VIEW_NAMES as readonly string[]).includes(input.viewName)) {
+      throw new ViewProjectorError(
+        "view_name_reserved",
+        `view name '${input.viewName}' is a reserved built-in name; choose a different name`,
+      );
+    }
+    const existing = this.db
+      .prepare(`SELECT * FROM views_custom WHERE view_name = ?`)
+      .get(input.viewName) as CustomViewRow | undefined;
+    const registeredAt = this.now().toISOString();
+    if (existing) {
+      this.db
+        .prepare(
+          `UPDATE views_custom
+             SET definition = ?, registered_by_session = ?, registered_at = ?
+           WHERE view_id = ?`,
+        )
+        .run(input.definition, input.registeredBySession, registeredAt, existing.view_id);
+      return this.getCustomViewByIdOrThrow(existing.view_id);
+    }
+    const viewId = ulid();
+    this.db
+      .prepare(
+        `INSERT INTO views_custom (
+          view_id, view_name, definition, registered_by_session, registered_at
+        ) VALUES (?, ?, ?, ?, ?)`,
+      )
+      .run(viewId, input.viewName, input.definition, input.registeredBySession, registeredAt);
+    return this.getCustomViewByIdOrThrow(viewId);
+  }
+
+  getCustomView(viewName: string): CustomView | null {
+    const row = this.db
+      .prepare(`SELECT * FROM views_custom WHERE view_name = ?`)
+      .get(viewName) as CustomViewRow | undefined;
+    return row ? this.rowToCustomView(row) : null;
+  }
+
+  listCustomViews(): CustomView[] {
+    const rows = this.db
+      .prepare(`SELECT * FROM views_custom ORDER BY view_name ASC`)
+      .all() as CustomViewRow[];
+    return rows.map((r) => this.rowToCustomView(r));
+  }
+
+  /**
+   * Used by routes/views.ts to emit view.changed when underlying state
+   * changes. Phase B does not auto-evaluate; callers (route SSE handlers)
+   * subscribe to event-bus events and trigger this notify.
+   */
+  notifyViewChanged(viewName: string, cause: string): void {
+    const persisted = this.eventBus.persistWithinTransaction({
+      type: "view.changed",
+      viewName,
+      cause,
+    });
+    this.eventBus.notifySubscribers(persisted);
+  }
+
+  private getCustomViewByIdOrThrow(viewId: string): CustomView {
+    const row = this.db
+      .prepare(`SELECT * FROM views_custom WHERE view_id = ?`)
+      .get(viewId) as CustomViewRow | undefined;
+    if (!row) {
+      throw new ViewProjectorError("view_not_found", `custom view ${viewId} not found after write`);
+    }
+    return this.rowToCustomView(row);
+  }
+
+  private rowToCustomView(row: CustomViewRow): CustomView {
+    return {
+      viewId: row.view_id,
+      viewName: row.view_name,
+      definition: row.definition,
+      registeredBySession: row.registered_by_session,
+      registeredAt: row.registered_at,
+      lastEvaluatedAt: row.last_evaluated_at,
+    };
+  }
+}

packages/daemon/src/routes/projects.ts

@@ -0,0 +1,183 @@
+import { Hono } from "hono";
+import { streamSSE } from "hono/streaming";
+import type { EventBus } from "../domain/event-bus.js";
+import type { ProjectClassifier } from "../domain/project-classifier.js";
+import { ProjectClassifierError } from "../domain/project-classifier.js";
+import type { ClassifierLeaseManager } from "../domain/classifier-lease-manager.js";
+import { ClassifierLeaseError } from "../domain/classifier-lease-manager.js";
+
+/**
+ * Coordination L2 — Project (Classifier) HTTP routes (PL-004 Phase B).
+ *
+ * Backs `rig project` CLI verb. Lease lifecycle endpoints + project
+ * (idempotent classify) + operator-verb reclaim + SSE.
+ *
+ * Per Phase A R1 SSE route-order lesson (slice IMPL § Audit Row 12):
+ * SSE/static routes are mounted BEFORE the bare-param /:id catchall.
+ */
+export function projectsRoutes(): Hono {
+  const app = new Hono();
+
+  function getClassifier(c: { get: (key: string) => unknown }): ProjectClassifier {
+    return c.get("projectClassifier" as never) as ProjectClassifier;
+  }
+  function getLease(c: { get: (key: string) => unknown }): ClassifierLeaseManager {
+    return c.get("classifierLeaseManager" as never) as ClassifierLeaseManager;
+  }
+  function getEventBus(c: { get: (key: string) => unknown }): EventBus {
+    return c.get("eventBus" as never) as EventBus;
+  }
+
+  function errorResponse(c: { json: (body: unknown, status?: number) => Response }, err: unknown): Response {
+    if (err instanceof ProjectClassifierError) {
+      const status = err.code === "idempotency_violation" ? 409
+        : err.code === "project_not_found" ? 404
+        : 500;
+      return c.json({ error: err.code, message: err.message, ...(err.meta ?? {}) }, status as 200);
+    }
+    if (err instanceof ClassifierLeaseError) {
+      const status = err.code === "lease_held" ? 409
+        : err.code === "lease_session_mismatch" ? 403
+        : err.code === "lease_not_active" ? 409
+        : err.code === "lease_expired" ? 409
+        : err.code === "lease_not_found" ? 404
+        : err.code === "no_active_lease" ? 409
+        : err.code === "lease_still_active" ? 409
+        : 500;
+      return c.json({ error: err.code, message: err.message, ...(err.meta ?? {}) }, status as 200);
+    }
+    const message = err instanceof Error ? err.message : "internal error";
+    return c.json({ error: "internal_error", message }, 500);
+  }
+
+  // POST /lease/acquire — acquire active classifier lease for caller.
+  app.post("/lease/acquire", async (c) => {
+    const body = await c.req.json<{ classifierSession?: string }>().catch(() => ({} as never));
+    if (!body.classifierSession) return c.json({ error: "classifierSession is required" }, 400);
+    try {
+      const lease = getLease(c).acquire(body.classifierSession);
+      return c.json(lease, 201);
+    } catch (err) {
+      return errorResponse(c, err);
+    }
+  });
+
+  // POST /lease/heartbeat — update last_heartbeat + extend expires_at.
+  app.post("/lease/heartbeat", async (c) => {
+    const body = await c.req.json<{ leaseId?: string; classifierSession?: string }>().catch(() => ({} as never));
+    if (!body.leaseId) return c.json({ error: "leaseId is required" }, 400);
+    if (!body.classifierSession) return c.json({ error: "classifierSession is required" }, 400);
+    try {
+      const lease = getLease(c).heartbeat(body.leaseId, body.classifierSession);
+      return c.json(lease);
+    } catch (err) {
+      return errorResponse(c, err);
+    }
+  });
+
+  // POST /reclaim-classifier — operator-verb reclaim (per PRD § L2 hard rule).
+  app.post("/reclaim-classifier", async (c) => {
+    const body = await c.req.json<{
+      byClassifierSession?: string;
+      ifDead?: boolean;
+      reason?: string;
+    }>().catch(() => ({} as never));
+    if (!body.byClassifierSession) return c.json({ error: "byClassifierSession is required" }, 400);
+    try {
+      const lease = getLease(c).reclaim(body.byClassifierSession, {
+        ifDead: body.ifDead,
+        reason: body.reason,
+      });
+      return c.json(lease);
+    } catch (err) {
+      return errorResponse(c, err);
+    }
+  });
+
+  // POST /project — project a stream item (idempotent on stream_item_id).
+  app.post("/project", async (c) => {
+    const body = await c.req.json<{
+      streamItemId?: string;
+      classifierSession?: string;
+      classificationType?: string;
+      classificationUrgency?: string;
+      classificationMaturity?: string;
+      classificationConfidence?: string;
+      classificationDestination?: string;
+      action?: string;
+    }>().catch(() => ({} as never));
+    if (!body.streamItemId) return c.json({ error: "streamItemId is required" }, 400);
+    if (!body.classifierSession) return c.json({ error: "classifierSession is required" }, 400);
+    try {
+      const project = getClassifier(c).classify({
+        streamItemId: body.streamItemId,
+        classifierSession: body.classifierSession,
+        classificationType: body.classificationType,
+        classificationUrgency: body.classificationUrgency,
+        classificationMaturity: body.classificationMaturity,
+        classificationConfidence: body.classificationConfidence,
+        classificationDestination: body.classificationDestination,
+        action: body.action,
+      });
+      return c.json(project, 201);
+    } catch (err) {
+      return errorResponse(c, err);
+    }
+  });
+
+  // GET /lease — show active lease.
+  // MUST precede /:projectId so the literal path wins.
+  app.get("/lease", (c) => {
+    const lease = getLease(c).getActiveLease();
+    if (!lease) return c.json({ error: "no_active_lease" }, 404);
+    return c.json(lease);
+  });
+
+  // GET /list — list classifications with filters.
+  // MUST precede /:projectId so the literal path wins.
+  app.get("/list", (c) => {
+    const classifierSession = c.req.query("classifierSession") || undefined;
+    const classificationDestination = c.req.query("classificationDestination") || undefined;
+    const limit = c.req.query("limit") ? Number.parseInt(c.req.query("limit")!, 10) : undefined;
+    const items = getClassifier(c).list({ classifierSession, classificationDestination, limit });
+    return c.json(items);
+  });
+
+  // ---- SSE for project + classifier events ----
+  // MUST precede /:projectId so the literal `sse` and `watch` paths win
+  // over the bare-param route. Per Phase A R1 SSE route-order lesson.
+  const sseHandler = (c: Parameters<typeof streamSSE>[0]) => {
+    const eventBus = getEventBus(c);
+    return streamSSE(c, async (stream) => {
+      const unsubscribe = eventBus.subscribe((event) => {
+        if (
+          event.type !== "project.classified" &&
+          event.type !== "classifier.lease_acquired" &&
+          event.type !== "classifier.lease_expired" &&
+          event.type !== "classifier.dead" &&
+          event.type !== "classifier.reclaimed"
+        ) 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("/sse", sseHandler);
+  app.get("/watch", sseHandler);
+
+  // GET /:projectId — show one project (must come AFTER literal routes).
+  app.get("/:projectId", (c) => {
+    const projectId = c.req.param("projectId");
+    const project = getClassifier(c).getById(projectId);
+    if (!project) return c.json({ error: "project_not_found" }, 404);
+    return c.json(project);
+  });
+
+  return app;
+}

packages/daemon/src/routes/views.ts

@@ -0,0 +1,124 @@
+import { Hono } from "hono";
+import { streamSSE } from "hono/streaming";
+import type { EventBus } from "../domain/event-bus.js";
+import type { ViewProjector } from "../domain/view-projector.js";
+import { ViewProjectorError } from "../domain/view-projector.js";
+
+/**
+ * Coordination L5 — View HTTP routes (PL-004 Phase B).
+ *
+ * Backs `rig view` CLI verb. Lists registered views (built-in + custom),
+ * runs view queries, and exposes view.changed SSE for downstream consumers
+ * (PL-005 Operator Status / PL-006 Mission Control / PL-008 progress).
+ *
+ * Per Phase A R1 SSE route-order lesson (slice IMPL § Audit Row 12):
+ * SSE/static routes are mounted BEFORE the bare-param /:name catchall.
+ */
+export function viewsRoutes(): Hono {
+  const app = new Hono();
+
+  function getProjector(c: { get: (key: string) => unknown }): ViewProjector {
+    return c.get("viewProjector" as never) as ViewProjector;
+  }
+  function getEventBus(c: { get: (key: string) => unknown }): EventBus {
+    return c.get("eventBus" as never) as EventBus;
+  }
+
+  function errorResponse(c: { json: (body: unknown, status?: number) => Response }, err: unknown): Response {
+    if (err instanceof ViewProjectorError) {
+      const status = err.code === "view_not_found" ? 404
+        : err.code === "view_name_reserved" ? 409
+        : err.code === "view_query_failed" ? 400
+        : 500;
+      return c.json({ error: err.code, message: err.message }, status as 200);
+    }
+    const message = err instanceof Error ? err.message : "internal error";
+    return c.json({ error: "internal_error", message }, 500);
+  }
+
+  // POST /custom/register — register/update custom view.
+  app.post("/custom/register", async (c) => {
+    const body = await c.req.json<{
+      viewName?: string;
+      definition?: string;
+      registeredBySession?: string;
+    }>().catch(() => ({} as never));
+    if (!body.viewName) return c.json({ error: "viewName is required" }, 400);
+    if (!body.definition) return c.json({ error: "definition is required" }, 400);
+    if (!body.registeredBySession) return c.json({ error: "registeredBySession is required" }, 400);
+    try {
+      const view = getProjector(c).registerCustomView({
+        viewName: body.viewName,
+        definition: body.definition,
+        registeredBySession: body.registeredBySession,
+      });
+      return c.json(view, 201);
+    } catch (err) {
+      return errorResponse(c, err);
+    }
+  });
+
+  // GET /list — built-in + custom view names.
+  // MUST precede /:viewName so the literal path wins.
+  app.get("/list", (c) => {
+    return c.json(getProjector(c).list());
+  });
+
+  // ---- SSE for view.changed events ----
+  // Generic SSE that emits ALL view.changed events. Per Phase A R1 lesson:
+  // mount BEFORE /:viewName catchall.
+  const sseHandler = (c: Parameters<typeof streamSSE>[0]) => {
+    const eventBus = getEventBus(c);
+    return streamSSE(c, async (stream) => {
+      const unsubscribe = eventBus.subscribe((event) => {
+        if (event.type !== "view.changed") 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("/sse", sseHandler);
+  app.get("/watch", sseHandler);
+
+  // GET /:viewName/sse — view-specific SSE filtered by viewName.
+  // The /:viewName/sse path is more specific than /:viewName so Hono
+  // dispatches it correctly even with the catchall registered later.
+  app.get("/:viewName/sse", (c) => {
+    const viewName = c.req.param("viewName");
+    const eventBus = getEventBus(c);
+    return streamSSE(c, async (stream) => {
+      const unsubscribe = eventBus.subscribe((event) => {
+        if (event.type !== "view.changed") return;
+        if (event.viewName !== viewName) 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();
+      }
+    });
+  });
+
+  // GET /:viewName — run a view (built-in or custom).
+  // Comes LAST so /list, /sse, /watch, /:viewName/sse all win.
+  app.get("/:viewName", (c) => {
+    const viewName = c.req.param("viewName");
+    const rig = c.req.query("rig") || undefined;
+    const limit = c.req.query("limit") ? Number.parseInt(c.req.query("limit")!, 10) : undefined;
+    try {
+      const result = getProjector(c).show(viewName, { rig, limit });
+      return c.json(result);
+    } catch (err) {
+      return errorResponse(c, err);
+    }
+  });
+
+  return app;
+}

packages/daemon/src/server.ts

@@ -61,10 +61,15 @@ import type { WhoamiService } from "./domain/whoami-service.js";
 import { chatRoutes } from "./routes/chat.js";
 import { streamRoutes } from "./routes/stream.js";
 import { queueRoutes } from "./routes/queue.js";
+import { projectsRoutes } from "./routes/projects.js";
+import { viewsRoutes } from "./routes/views.js";
 import type { StreamStore } from "./domain/stream-store.js";
 import type { QueueRepository } from "./domain/queue-repository.js";
 import type { InboxHandler } from "./domain/inbox-handler.js";
 import type { OutboxHandler } from "./domain/outbox-handler.js";
+import type { ProjectClassifier } from "./domain/project-classifier.js";
+import type { ClassifierLeaseManager } from "./domain/classifier-lease-manager.js";
+import type { ViewProjector } from "./domain/view-projector.js";
 import { envRoutes } from "./routes/env.js";
 import type { RigLifecycleService } from "./domain/rig-lifecycle-service.js";
 import { seatRoutes } from "./routes/seat.js";
@@ -108,6 +113,9 @@ export interface AppDeps {
   queueRepo?: QueueRepository;
   inboxHandler?: InboxHandler;
   outboxHandler?: OutboxHandler;
+  projectClassifier?: ProjectClassifier;
+  classifierLeaseManager?: ClassifierLeaseManager;
+  viewProjector?: ViewProjector;
   specReviewService?: SpecReviewService;
   specLibraryService?: SpecLibraryService;
   whoamiService?: WhoamiService;
@@ -253,6 +261,9 @@ export function createApp(deps: AppDeps): Hono {
     c.set("queueRepo" as never, deps.queueRepo);
     c.set("inboxHandler" as never, deps.inboxHandler);
     c.set("outboxHandler" as never, deps.outboxHandler);
+    c.set("projectClassifier" as never, deps.projectClassifier);
+    c.set("classifierLeaseManager" as never, deps.classifierLeaseManager);
+    c.set("viewProjector" as never, deps.viewProjector);
     c.set("specReviewService" as never, deps.specReviewService);
     c.set("specLibraryService" as never, deps.specLibraryService);
     c.set("whoamiService" as never, deps.whoamiService);
@@ -301,6 +312,8 @@ export function createApp(deps: AppDeps): Hono {
   app.route("/api/rigs/:rigId/chat", chatRoutes());
   app.route("/api/stream", streamRoutes());
   app.route("/api/queue", queueRoutes());
+  app.route("/api/projects", projectsRoutes());
+  app.route("/api/views", viewsRoutes());
   app.route("/api/rigs/:rigId/env", envRoutes());
   app.route("/api/restore-check", restoreCheckRoutes);
 

packages/daemon/src/startup.ts

@@ -63,6 +63,9 @@ import { StreamStore } from "./domain/stream-store.js";
 import { QueueRepository } from "./domain/queue-repository.js";
 import { InboxHandler } from "./domain/inbox-handler.js";
 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 { SpecReviewService } from "./domain/spec-review-service.js";
 import { SpecLibraryService } from "./domain/spec-library-service.js";
 import { WhoamiService } from "./domain/whoami-service.js";
@@ -96,6 +99,9 @@ import { queueItemsSchema } from "./db/migrations/024_queue_items.js";
 import { queueTransitionsSchema } from "./db/migrations/025_queue_transitions.js";
 import { inboxEntriesSchema } from "./db/migrations/026_inbox_entries.js";
 import { outboxEntriesSchema } from "./db/migrations/027_outbox_entries.js";
+import { projectClassificationsSchema } from "./db/migrations/028_project_classifications.js";
+import { classifierLeasesSchema } from "./db/migrations/029_classifier_leases.js";
+import { viewsCustomSchema } from "./db/migrations/030_views_custom.js";
 import { OPENRIG_HOME } from "./openrig-compat.js";
 import {
   getCompatibleOpenRigPath,
@@ -121,7 +127,7 @@ interface DaemonResult {
 export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult> {
   const dbPath = opts?.dbPath ?? ":memory:";
   const db = createDb(dbPath);
-  migrate(db, [coreSchema, bindingsSessionsSchema, eventsSchema, snapshotsSchema, checkpointsSchema, resumeMetadataSchema, nodeSpecFieldsSchema, packagesSchema, installJournalSchema, journalSeqSchema, bootstrapSchema, discoverySchema, discoveryFkFix, agentspecRebootSchema, startupContextSchema, chatMessagesSchema, podNamespaceSchema, contextUsageSchema, externalCliAttachmentSchema, rigServicesSchema, seatHandoverObservabilitySchema, nodeCodexConfigProfileSchema, streamItemsSchema, queueItemsSchema, queueTransitionsSchema, inboxEntriesSchema, outboxEntriesSchema]);
+  migrate(db, [coreSchema, bindingsSessionsSchema, eventsSchema, snapshotsSchema, checkpointsSchema, resumeMetadataSchema, nodeSpecFieldsSchema, packagesSchema, installJournalSchema, journalSeqSchema, bootstrapSchema, discoverySchema, discoveryFkFix, agentspecRebootSchema, startupContextSchema, chatMessagesSchema, podNamespaceSchema, contextUsageSchema, externalCliAttachmentSchema, rigServicesSchema, seatHandoverObservabilitySchema, nodeCodexConfigProfileSchema, streamItemsSchema, queueItemsSchema, queueTransitionsSchema, inboxEntriesSchema, outboxEntriesSchema, projectClassificationsSchema, classifierLeasesSchema, viewsCustomSchema]);
 
   const rigRepo = new RigRepository(db);
   const sessionRegistry = new SessionRegistry(db);
@@ -143,6 +149,10 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
   const queueRepoInstance = new QueueRepository(db, eventBus, {
     validateRig: topologyValidateRig,
   });
+  // PL-004 Phase B — classifier lease manager. Constructed early so both
+  // 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);
 
   const tmuxAdapter = new TmuxAdapter(opts?.tmuxExec ?? execCommand);
   // cmuxFactory takes precedence (for tests), then cmuxExec-based CLI transport, then default
@@ -389,6 +399,24 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
       // PL-004 Phase A revision (R1): wire QueueRepository's wake-path so
       // create / handoff / handoff-and-complete nudge by default.
       queueRepoInstance.attachTransport(t);
+      // PL-004 Phase B: wire classifier-lease-manager liveness check from
+      // the sessions table. Lease holder is "alive" iff there is at least
+      // one row in `sessions` with session_name == classifierSession AND
+      // status == 'running'.
+      classifierLeaseManagerInstance.attachIsAlive((classifierSession: string): boolean => {
+        try {
+          const row = db
+            .prepare(
+              `SELECT 1 FROM sessions WHERE session_name = ? AND status = 'running' LIMIT 1`,
+            )
+            .get(classifierSession) as { 1: number } | undefined;
+          return row !== undefined;
+        } catch {
+          // Conservative: on lookup error, treat as alive (do not falsely
+          // trigger deadness-based lease expiry).
+          return true;
+        }
+      });
       return t;
     })(),
     chatRepo: new ChatRepository(db),
@@ -396,6 +424,9 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
     queueRepo: queueRepoInstance,
     inboxHandler: new InboxHandler(db, eventBus, queueRepoInstance),
     outboxHandler: new OutboxHandler(db),
+    classifierLeaseManager: classifierLeaseManagerInstance,
+    projectClassifier: new ProjectClassifier(db, eventBus, classifierLeaseManagerInstance),
+    viewProjector: new ViewProjector(db, eventBus),
     askService: (() => {
       const psProjectionService = new PsProjectionService({ db });
       const execDep = (cmd: string, args: string[]): Promise<{ stdout: string; exitCode: number }> =>

packages/daemon/test/classifier-lease-manager.test.ts

@@ -0,0 +1,186 @@
+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 { classifierLeasesSchema } from "../src/db/migrations/029_classifier_leases.js";
+import { EventBus } from "../src/domain/event-bus.js";
+import {
+  ClassifierLeaseManager,
+  ClassifierLeaseError,
+} from "../src/domain/classifier-lease-manager.js";
+import type { PersistedEvent } from "../src/domain/types.js";
+
+describe("ClassifierLeaseManager (PL-004 Phase B; L2 lease lifecycle)", () => {
+  let db: Database.Database;
+  let bus: EventBus;
+  let mgr: ClassifierLeaseManager;
+  let captured: PersistedEvent[];
+
+  beforeEach(() => {
+    db = createDb();
+    migrate(db, [coreSchema, eventsSchema, classifierLeasesSchema]);
+    bus = new EventBus(db);
+    mgr = new ClassifierLeaseManager(db, bus, { ttlMs: 1000 });
+    captured = [];
+    bus.subscribe((e) => captured.push(e));
+  });
+
+  afterEach(() => db.close());
+
+  it("acquire creates an active lease + emits classifier.lease_acquired", () => {
+    const lease = mgr.acquire("alice@rig");
+    expect(lease.classifierSession).toBe("alice@rig");
+    expect(lease.state).toBe("active");
+    expect(lease.acquiredAt).toMatch(/^\d{4}-\d{2}-\d{2}T/);
+    expect(lease.expiresAt > lease.acquiredAt).toBe(true);
+    expect(captured.some((e) => e.type === "classifier.lease_acquired")).toBe(true);
+  });
+
+  it("acquire is idempotent for the same classifier session (returns existing lease unchanged)", () => {
+    const a = mgr.acquire("alice@rig");
+    const b = mgr.acquire("alice@rig");
+    expect(b.leaseId).toBe(a.leaseId);
+    // Only ONE lease.acquired event (idempotent path doesn't re-emit).
+    expect(captured.filter((e) => e.type === "classifier.lease_acquired")).toHaveLength(1);
+  });
+
+  it("acquire by a different session when one is active throws lease_held (409)", () => {
+    mgr.acquire("alice@rig");
+    expect(() => mgr.acquire("bob@rig")).toThrow(ClassifierLeaseError);
+    try {
+      mgr.acquire("bob@rig");
+    } catch (err) {
+      expect((err as ClassifierLeaseError).code).toBe("lease_held");
+      expect((err as ClassifierLeaseError).meta?.holder).toBe("alice@rig");
+    }
+  });
+
+  it("heartbeat extends expires_at and updates last_heartbeat", () => {
+    const fakeNow = (() => {
+      let t = new Date("2026-05-03T00:00:00.000Z").getTime();
+      return () => { const d = new Date(t); t += 100; return d; };
+    })();
+    const m = new ClassifierLeaseManager(db, bus, { ttlMs: 1000, now: fakeNow });
+    const acquired = m.acquire("alice@rig");
+    const beat = m.heartbeat(acquired.leaseId, "alice@rig");
+    expect(beat.lastHeartbeat > acquired.lastHeartbeat).toBe(true);
+    expect(beat.expiresAt > acquired.expiresAt).toBe(true);
+  });
+
+  it("heartbeat by mismatched session throws lease_session_mismatch (403)", () => {
+    const lease = mgr.acquire("alice@rig");
+    expect(() => mgr.heartbeat(lease.leaseId, "bob@rig")).toThrow(/lease_session_mismatch|alice@rig/);
+  });
+
+  it("evaluateDeadness on alive holder returns null (no expiry)", () => {
+    mgr.acquire("alice@rig");
+    const result = mgr.evaluateDeadness();
+    expect(result).toBeNull();
+  });
+
+  it("evaluateDeadness on dead holder marks lease expired + emits classifier.dead", () => {
+    const m = new ClassifierLeaseManager(db, bus, {
+      ttlMs: 1000,
+      isAlive: (s) => s !== "alice@rig", // alice is reported dead
+    });
+    m.acquire("alice@rig");
+    const result = m.evaluateDeadness();
+    expect(result).not.toBeNull();
+    expect(result!.state).toBe("expired");
+    expect(captured.some((e) => e.type === "classifier.lease_expired")).toBe(true);
+    expect(captured.some((e) => e.type === "classifier.dead")).toBe(true);
+  });
+
+  it("evaluateDeadness on TTL-expired lease marks expired (without classifier.dead)", () => {
+    let t = new Date("2026-05-03T00:00:00.000Z").getTime();
+    const m = new ClassifierLeaseManager(db, bus, {
+      ttlMs: 100,
+      now: () => new Date(t),
+    });
+    m.acquire("alice@rig");
+    t += 200; // advance past TTL
+    const result = m.evaluateDeadness();
+    expect(result).not.toBeNull();
+    expect(result!.state).toBe("expired");
+    expect(captured.some((e) => e.type === "classifier.lease_expired")).toBe(true);
+    expect(captured.some((e) => e.type === "classifier.dead")).toBe(false);
+  });
+
+  it("after evaluateDeadness expires the lease, a new session can acquire", () => {
+    const m = new ClassifierLeaseManager(db, bus, {
+      ttlMs: 1000,
+      isAlive: (s) => s !== "alice@rig",
+    });
+    m.acquire("alice@rig");
+    m.evaluateDeadness(); // marks alice's lease expired
+    const fresh = m.acquire("bob@rig");
+    expect(fresh.classifierSession).toBe("bob@rig");
+    expect(fresh.state).toBe("active");
+  });
+
+  it("reclaim takes the active lease away from previous holder + emits classifier.reclaimed", () => {
+    mgr.acquire("alice@rig");
+    const result = mgr.reclaim("operator@rig", { reason: "alice unresponsive" });
+    expect(result.state).toBe("reclaimed");
+    expect(result.reclaimedBySession).toBe("operator@rig");
+    expect(result.reclaimReason).toBe("alice unresponsive");
+    expect(captured.some((e) => e.type === "classifier.reclaimed")).toBe(true);
+  });
+
+  it("reclaim --if-dead refuses when holder is alive", () => {
+    mgr.acquire("alice@rig"); // default isAlive returns true
+    try {
+      mgr.reclaim("operator@rig", { ifDead: true });
+      throw new Error("should have thrown");
+    } catch (err) {
+      expect(err).toBeInstanceOf(ClassifierLeaseError);
+      expect((err as ClassifierLeaseError).code).toBe("lease_still_active");
+    }
+  });
+
+  it("reclaim --if-dead succeeds when holder is dead", () => {
+    const m = new ClassifierLeaseManager(db, bus, {
+      ttlMs: 1000,
+      isAlive: (s) => s !== "alice@rig",
+    });
+    m.acquire("alice@rig");
+    const result = m.reclaim("operator@rig", { ifDead: true });
+    expect(result.state).toBe("reclaimed");
+  });
+
+  it("requireActiveHolder validates the supplied session holds the active lease", () => {
+    const lease = mgr.acquire("alice@rig");
+    const checked = mgr.requireActiveHolder("alice@rig");
+    expect(checked.leaseId).toBe(lease.leaseId);
+  });
+
+  it("requireActiveHolder throws lease_held when supplied a different session", () => {
+    mgr.acquire("alice@rig");
+    expect(() => mgr.requireActiveHolder("bob@rig")).toThrow(ClassifierLeaseError);
+  });
+
+  it("requireActiveHolder throws no_active_lease when nothing is held", () => {
+    try {
+      mgr.requireActiveHolder("alice@rig");
+      throw new Error("should have thrown");
+    } catch (err) {
+      expect(err).toBeInstanceOf(ClassifierLeaseError);
+      expect((err as ClassifierLeaseError).code).toBe("no_active_lease");
+    }
+  });
+
+  it("partial UNIQUE index enforces single-writer at DB layer", () => {
+    mgr.acquire("alice@rig");
+    // Manual INSERT with state='active' via raw SQL would be blocked by the
+    // partial UNIQUE index `idx_classifier_leases_active_singleton`.
+    expect(() => {
+      db.prepare(
+        `INSERT INTO classifier_leases (
+          lease_id, classifier_session, acquired_at, expires_at, last_heartbeat, state
+        ) VALUES (?, ?, ?, ?, ?, 'active')`
+      ).run("dup-lease", "bob@rig", "2026-05-03T00:00:00Z", "2026-05-03T00:00:01Z", "2026-05-03T00:00:00Z");
+    }).toThrow(/UNIQUE constraint/);
+  });
+});

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

@@ -0,0 +1,163 @@
+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 { 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,
+  ProjectClassifierError,
+} from "../src/domain/project-classifier.js";
+import { ClassifierLeaseError } from "../src/domain/classifier-lease-manager.js";
+import type { PersistedEvent } from "../src/domain/types.js";
+
+describe("ProjectClassifier (PL-004 Phase B; L2 classifier write path)", () => {
+  let db: Database.Database;
+  let bus: EventBus;
+  let leaseMgr: ClassifierLeaseManager;
+  let classifier: ProjectClassifier;
+  let captured: PersistedEvent[];
+
+  beforeEach(() => {
+    db = createDb();
+    migrate(db, [coreSchema, eventsSchema, classifierLeasesSchema, projectClassificationsSchema]);
+    bus = new EventBus(db);
+    leaseMgr = new ClassifierLeaseManager(db, bus);
+    classifier = new ProjectClassifier(db, bus, leaseMgr);
+    captured = [];
+    bus.subscribe((e) => captured.push(e));
+  });
+
+  afterEach(() => db.close());
+
+  it("classify with valid lease creates project_classifications row + emits project.classified", () => {
+    leaseMgr.acquire("alice@rig");
+    const proj = classifier.classify({
+      streamItemId: "stream-1",
+      classifierSession: "alice@rig",
+      classificationType: "idea",
+      classificationDestination: "planning@rig",
+    });
+    expect(proj.projectId).toMatch(/^[0-9A-Z]{26}$/); // ULID shape
+    expect(proj.streamItemId).toBe("stream-1");
+    expect(proj.classifierSession).toBe("alice@rig");
+    expect(proj.classificationType).toBe("idea");
+    expect(captured.some((e) => e.type === "project.classified")).toBe(true);
+  });
+
+  it("classify without active lease throws no_active_lease", () => {
+    expect(() => classifier.classify({
+      streamItemId: "stream-1",
+      classifierSession: "alice@rig",
+    })).toThrow(ClassifierLeaseError);
+  });
+
+  it("classify by non-holder throws lease_held", () => {
+    leaseMgr.acquire("alice@rig");
+    expect(() => classifier.classify({
+      streamItemId: "stream-1",
+      classifierSession: "bob@rig",
+    })).toThrow(/lease_held|alice@rig/);
+  });
+
+  it("classify is idempotent on stream_item_id (re-projection → idempotency_violation 409)", () => {
+    leaseMgr.acquire("alice@rig");
+    classifier.classify({
+      streamItemId: "stream-1",
+      classifierSession: "alice@rig",
+      classificationType: "idea",
+    });
+    expect(() => classifier.classify({
+      streamItemId: "stream-1",
+      classifierSession: "alice@rig",
+      classificationType: "bug",
+    })).toThrow(ProjectClassifierError);
+    try {
+      classifier.classify({
+        streamItemId: "stream-1",
+        classifierSession: "alice@rig",
+        classificationType: "bug",
+      });
+    } catch (err) {
+      expect((err as ProjectClassifierError).code).toBe("idempotency_violation");
+    }
+  });
+
+  it("first classification's classifier_session is preserved on re-projection attempts", () => {
+    leaseMgr.acquire("alice@rig");
+    const first = classifier.classify({
+      streamItemId: "stream-1",
+      classifierSession: "alice@rig",
+      classificationType: "idea",
+    });
+    // Reclaim + new lease + same session can also re-attempt — still rejected.
+    leaseMgr.reclaim("operator@rig");
+    leaseMgr.acquire("bob@rig");
+    try {
+      classifier.classify({
+        streamItemId: "stream-1",
+        classifierSession: "bob@rig",
+        classificationType: "feature-request",
+      });
+      throw new Error("should have thrown");
+    } catch (err) {
+      expect(err).toBeInstanceOf(ProjectClassifierError);
+      expect((err as ProjectClassifierError).code).toBe("idempotency_violation");
+    }
+    const lookup = classifier.getByStreamItemId("stream-1");
+    expect(lookup?.projectId).toBe(first.projectId);
+    expect(lookup?.classifierSession).toBe("alice@rig"); // first wins
+    expect(lookup?.classificationType).toBe("idea");
+  });
+
+  it("classify accepts all 6 classification fields + action; null when unset", () => {
+    leaseMgr.acquire("alice@rig");
+    const full = classifier.classify({
+      streamItemId: "stream-full",
+      classifierSession: "alice@rig",
+      classificationType: "idea",
+      classificationUrgency: "high",
+      classificationMaturity: "ratified",
+      classificationConfidence: "high",
+      classificationDestination: "planning@rig",
+      action: "create",
+    });
+    expect(full.classificationUrgency).toBe("high");
+    expect(full.classificationMaturity).toBe("ratified");
+    expect(full.classificationConfidence).toBe("high");
+    expect(full.action).toBe("create");
+
+    const minimal = classifier.classify({
+      streamItemId: "stream-minimal",
+      classifierSession: "alice@rig",
+    });
+    expect(minimal.classificationType).toBeNull();
+    expect(minimal.classificationUrgency).toBeNull();
+    expect(minimal.action).toBeNull();
+  });
+
+  it("list filters by classifierSession + classificationDestination", () => {
+    leaseMgr.acquire("alice@rig");
+    classifier.classify({
+      streamItemId: "s1",
+      classifierSession: "alice@rig",
+      classificationDestination: "planning@rig",
+    });
+    classifier.classify({
+      streamItemId: "s2",
+      classifierSession: "alice@rig",
+      classificationDestination: "delivery@rig",
+    });
+    classifier.classify({
+      streamItemId: "s3",
+      classifierSession: "alice@rig",
+      classificationDestination: "planning@rig",
+    });
+    const planning = classifier.list({ classificationDestination: "planning@rig" });
+    expect(planning).toHaveLength(2);
+  });
+});

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

@@ -0,0 +1,208 @@
+import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import { Hono } from "hono";
+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 { 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 { projectsRoutes } from "../src/routes/projects.js";
+
+function buildApp(opts: {
+  eventBus: EventBus;
+  classifier: ProjectClassifier;
+  leaseMgr: ClassifierLeaseManager;
+}): Hono {
+  const app = new Hono();
+  app.use("*", async (c, next) => {
+    c.set("eventBus" as never, opts.eventBus);
+    c.set("projectClassifier" as never, opts.classifier);
+    c.set("classifierLeaseManager" as never, opts.leaseMgr);
+    await next();
+  });
+  app.route("/api/projects", projectsRoutes());
+  return app;
+}
+
+describe("projects routes (PL-004 Phase B)", () => {
+  let db: Database.Database;
+  let bus: EventBus;
+  let leaseMgr: ClassifierLeaseManager;
+  let classifier: ProjectClassifier;
+  let app: Hono;
+
+  beforeEach(() => {
+    db = createDb();
+    migrate(db, [coreSchema, eventsSchema, classifierLeasesSchema, projectClassificationsSchema]);
+    bus = new EventBus(db);
+    leaseMgr = new ClassifierLeaseManager(db, bus);
+    classifier = new ProjectClassifier(db, bus, leaseMgr);
+    app = buildApp({ eventBus: bus, classifier, leaseMgr });
+  });
+
+  afterEach(() => db.close());
+
+  it("POST /api/projects/lease/acquire returns 201 + active lease", async () => {
+    const res = await app.request("/api/projects/lease/acquire", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ classifierSession: "alice@rig" }),
+    });
+    expect(res.status).toBe(201);
+    const lease = (await res.json()) as { state: string; classifierSession: string };
+    expect(lease.state).toBe("active");
+    expect(lease.classifierSession).toBe("alice@rig");
+  });
+
+  it("POST /api/projects/project requires lease + idempotent on stream_item_id", async () => {
+    await app.request("/api/projects/lease/acquire", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ classifierSession: "alice@rig" }),
+    });
+    const first = await app.request("/api/projects/project", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({
+        streamItemId: "stream-x",
+        classifierSession: "alice@rig",
+        classificationType: "idea",
+      }),
+    });
+    expect(first.status).toBe(201);
+    const project = (await first.json()) as { projectId: string };
+    expect(project.projectId).toMatch(/^[0-9A-Z]{26}$/);
+
+    const second = await app.request("/api/projects/project", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({
+        streamItemId: "stream-x",
+        classifierSession: "alice@rig",
+        classificationType: "bug",
+      }),
+    });
+    expect(second.status).toBe(409);
+    const err = (await second.json()) as { error: string };
+    expect(err.error).toBe("idempotency_violation");
+  });
+
+  it("POST /api/projects/project without active lease returns 409 no_active_lease", async () => {
+    const res = await app.request("/api/projects/project", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({
+        streamItemId: "stream-x",
+        classifierSession: "alice@rig",
+      }),
+    });
+    expect(res.status).toBe(409);
+    const err = (await res.json()) as { error: string };
+    expect(err.error).toBe("no_active_lease");
+  });
+
+  it("POST /api/projects/reclaim-classifier with --if-dead refuses on alive holder", async () => {
+    await app.request("/api/projects/lease/acquire", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ classifierSession: "alice@rig" }),
+    });
+    const res = await app.request("/api/projects/reclaim-classifier", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ byClassifierSession: "operator@rig", ifDead: true }),
+    });
+    expect(res.status).toBe(409);
+    const err = (await res.json()) as { error: string };
+    expect(err.error).toBe("lease_still_active");
+  });
+
+  it("POST /api/projects/reclaim-classifier without --if-dead succeeds", async () => {
+    await app.request("/api/projects/lease/acquire", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ classifierSession: "alice@rig" }),
+    });
+    const res = await app.request("/api/projects/reclaim-classifier", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ byClassifierSession: "operator@rig" }),
+    });
+    expect(res.status).toBe(200);
+    const lease = (await res.json()) as { state: string; reclaimedBySession: string };
+    expect(lease.state).toBe("reclaimed");
+    expect(lease.reclaimedBySession).toBe("operator@rig");
+  });
+
+  it("GET /api/projects/lease returns active lease (or 404 if none)", async () => {
+    let res = await app.request("/api/projects/lease");
+    expect(res.status).toBe(404);
+
+    await app.request("/api/projects/lease/acquire", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ classifierSession: "alice@rig" }),
+    });
+
+    res = await app.request("/api/projects/lease");
+    expect(res.status).toBe(200);
+    const lease = (await res.json()) as { classifierSession: string };
+    expect(lease.classifierSession).toBe("alice@rig");
+  });
+
+  it("GET /api/projects/list filters by classifierSession + classificationDestination", async () => {
+    await app.request("/api/projects/lease/acquire", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ classifierSession: "alice@rig" }),
+    });
+    for (const [id, dest] of [["s1", "planning@rig"], ["s2", "delivery@rig"], ["s3", "planning@rig"]] as Array<[string, string]>) {
+      await app.request("/api/projects/project", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          streamItemId: id,
+          classifierSession: "alice@rig",
+          classificationDestination: dest,
+        }),
+      });
+    }
+    const res = await app.request("/api/projects/list?classificationDestination=planning%40rig");
+    const data = (await res.json()) as unknown[];
+    expect(data).toHaveLength(2);
+  });
+
+  it("R1 SSE pattern: GET /api/projects/sse returns 200 + content-type text/event-stream (handler reached, not /:id)", async () => {
+    const res = await app.request("/api/projects/sse");
+    try {
+      expect(res.status).toBe(200);
+      expect(res.headers.get("content-type") ?? "").toContain("text/event-stream");
+    } finally {
+      await res.body?.cancel();
+    }
+  });
+
+  it("R1 SSE pattern: GET /api/projects/watch returns 200 + content-type text/event-stream", async () => {
+    const res = await app.request("/api/projects/watch");
+    try {
+      expect(res.status).toBe(200);
+      expect(res.headers.get("content-type") ?? "").toContain("text/event-stream");
+    } finally {
+      await res.body?.cancel();
+    }
+  });
+
+  it("R1 SSE pattern: GET /api/projects/sse does NOT return project_not_found (route-order regression guard)", async () => {
+    const res = await app.request("/api/projects/sse");
+    try {
+      expect(res.status).not.toBe(404);
+      expect(res.headers.get("content-type") ?? "").not.toContain("application/json");
+    } finally {
+      await res.body?.cancel();
+    }
+  });
+});

packages/daemon/test/view-projector.test.ts

@@ -0,0 +1,206 @@
+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 { queueItemsSchema } from "../src/db/migrations/024_queue_items.js";
+import { queueTransitionsSchema } from "../src/db/migrations/025_queue_transitions.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 {
+  ViewProjector,
+  ViewProjectorError,
+  BUILT_IN_VIEW_NAMES,
+} from "../src/domain/view-projector.js";
+
+describe("ViewProjector (PL-004 Phase B; L5 read-only projections)", () => {
+  let db: Database.Database;
+  let bus: EventBus;
+  let queueRepo: QueueRepository;
+  let projector: ViewProjector;
+
+  beforeEach(async () => {
+    db = createDb();
+    migrate(db, [coreSchema, eventsSchema, queueItemsSchema, queueTransitionsSchema, viewsCustomSchema]);
+    bus = new EventBus(db);
+    queueRepo = new QueueRepository(db, bus);
+    projector = new ViewProjector(db, bus);
+    // Seed live qitems for the views to project over.
+    await queueRepo.create({
+      sourceSession: "alice@product-lab",
+      destinationSession: "planning@product-lab",
+      body: "design new feature",
+      priority: "routine",
+      tier: "deep",
+      nudge: false,
+    });
+    await queueRepo.create({
+      sourceSession: "alice@product-lab",
+      destinationSession: "planning@product-lab",
+      body: "fix critical bug",
+      priority: "critical",
+      tier: "fast",
+      nudge: false,
+    });
+    const blocked = await queueRepo.create({
+      sourceSession: "alice@product-lab",
+      destinationSession: "delivery@product-lab",
+      body: "blocked work",
+      nudge: false,
+    });
+    queueRepo.update({
+      qitemId: blocked.qitemId,
+      actorSession: "delivery@product-lab",
+      state: "blocked",
+      transitionNote: "blocked on dep",
+    });
+    // Fixture rig (should be excluded by default).
+    await queueRepo.create({
+      sourceSession: "alice@test-rig",
+      destinationSession: "bob@test-rig",
+      body: "fixture work",
+      nudge: false,
+    });
+  });
+
+  afterEach(() => {
+    db.close();
+    delete process.env.OPENRIG_VIEW_INCLUDE_FIXTURES;
+  });
+
+  it("BUILT_IN_VIEW_NAMES includes all 6 expected names", () => {
+    expect([...BUILT_IN_VIEW_NAMES]).toEqual([
+      "recently-active",
+      "founder",
+      "pod-load",
+      "escalations",
+      "held",
+      "activity",
+    ]);
+  });
+
+  it("show recently-active returns active-state qitems by ts_updated DESC, fixtures excluded", () => {
+    const result = projector.show("recently-active");
+    expect(result.viewName).toBe("recently-active");
+    expect(result.rowCount).toBe(3); // 3 product-lab qitems; fixture excluded
+    expect(result.rows.every((r) => !String(r.destination_session).includes("@test-"))).toBe(true);
+  });
+
+  it("show founder returns critical-priority OR fast/critical-tier qitems", () => {
+    const result = projector.show("founder");
+    expect(result.rowCount).toBe(1); // only the critical/fast qitem
+    expect(result.rows[0]!.priority).toBe("critical");
+  });
+
+  it("show pod-load groups counts by destination_session", () => {
+    const result = projector.show("pod-load");
+    expect(result.rowCount).toBe(2); // planning@... and delivery@...
+    const podMap = new Map<string, number>(result.rows.map((r) => [String(r.pod), Number(r.active_count)]));
+    expect(podMap.get("planning@product-lab")).toBe(2);
+    expect(podMap.get("delivery@product-lab")).toBe(1);
+  });
+
+  it("show held returns blocked qitems", () => {
+    const result = projector.show("held");
+    expect(result.rowCount).toBe(1);
+    expect(result.rows[0]!.state).toBe("blocked");
+  });
+
+  it("show escalations returns qitems with closure_reason='escalation'", () => {
+    // No escalations in fixture; should return 0.
+    const result = projector.show("escalations");
+    expect(result.rowCount).toBe(0);
+  });
+
+  it("show activity returns recent transitions joined to active qitems", () => {
+    const result = projector.show("activity");
+    expect(result.rowCount).toBeGreaterThan(0);
+    // Each row has a transition_id (from queue_transitions).
+    expect(result.rows[0]).toHaveProperty("transition_id");
+  });
+
+  it("OPENRIG_VIEW_INCLUDE_FIXTURES=1 includes fixture rigs", () => {
+    process.env.OPENRIG_VIEW_INCLUDE_FIXTURES = "1";
+    const result = projector.show("recently-active");
+    expect(result.rowCount).toBe(4); // including the fixture qitem
+  });
+
+  it("show with --rig filter narrows by session-suffix match", () => {
+    const result = projector.show("recently-active", { rig: "product-lab" });
+    expect(result.rowCount).toBe(3); // all 3 product-lab qitems
+  });
+
+  it("show <unknown-view> throws view_not_found", () => {
+    expect(() => projector.show("nonexistent-view")).toThrow(ViewProjectorError);
+  });
+
+  it("registerCustomView accepts arbitrary SQL + lookup works via show", () => {
+    const view = projector.registerCustomView({
+      viewName: "all-pending",
+      definition: "SELECT qitem_id, destination_session FROM queue_items WHERE state = 'pending'",
+      registeredBySession: "operator@rig",
+    });
+    expect(view.viewId).toMatch(/^[0-9A-Z]{26}$/);
+    const result = projector.show("all-pending");
+    expect(result.viewName).toBe("all-pending");
+    expect(result.rowCount).toBeGreaterThan(0);
+  });
+
+  it("registerCustomView rejects reserved built-in name", () => {
+    try {
+      projector.registerCustomView({
+        viewName: "recently-active",
+        definition: "SELECT 1",
+        registeredBySession: "operator@rig",
+      });
+      throw new Error("should have thrown");
+    } catch (err) {
+      expect(err).toBeInstanceOf(ViewProjectorError);
+      expect((err as ViewProjectorError).code).toBe("view_name_reserved");
+    }
+  });
+
+  it("registerCustomView re-registration updates definition (not duplicate)", () => {
+    const first = projector.registerCustomView({
+      viewName: "my-view",
+      definition: "SELECT 1 as a",
+      registeredBySession: "operator@rig",
+    });
+    const second = projector.registerCustomView({
+      viewName: "my-view",
+      definition: "SELECT 2 as b",
+      registeredBySession: "operator@rig",
+    });
+    expect(second.viewId).toBe(first.viewId);
+    expect(second.definition).toContain("SELECT 2");
+    expect(projector.listCustomViews()).toHaveLength(1);
+  });
+
+  it("list returns built-in view names + custom view records", () => {
+    projector.registerCustomView({
+      viewName: "my-view",
+      definition: "SELECT 1",
+      registeredBySession: "operator@rig",
+    });
+    const result = projector.list();
+    expect(result.builtIn).toEqual([
+      "recently-active",
+      "founder",
+      "pod-load",
+      "escalations",
+      "held",
+      "activity",
+    ]);
+    expect(result.custom).toHaveLength(1);
+    expect(result.custom[0]!.viewName).toBe("my-view");
+  });
+
+  it("notifyViewChanged emits view.changed event", () => {
+    const captured: unknown[] = [];
+    bus.subscribe((e) => captured.push(e));
+    projector.notifyViewChanged("recently-active", "queue.created");
+    expect(captured.some((e) => (e as { type: string }).type === "view.changed")).toBe(true);
+  });
+});

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

@@ -0,0 +1,159 @@
+import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import { Hono } from "hono";
+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 { queueItemsSchema } from "../src/db/migrations/024_queue_items.js";
+import { queueTransitionsSchema } from "../src/db/migrations/025_queue_transitions.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 { ViewProjector } from "../src/domain/view-projector.js";
+import { viewsRoutes } from "../src/routes/views.js";
+
+function buildApp(opts: {
+  eventBus: EventBus;
+  projector: ViewProjector;
+}): Hono {
+  const app = new Hono();
+  app.use("*", async (c, next) => {
+    c.set("eventBus" as never, opts.eventBus);
+    c.set("viewProjector" as never, opts.projector);
+    await next();
+  });
+  app.route("/api/views", viewsRoutes());
+  return app;
+}
+
+describe("views routes (PL-004 Phase B)", () => {
+  let db: Database.Database;
+  let bus: EventBus;
+  let queueRepo: QueueRepository;
+  let projector: ViewProjector;
+  let app: Hono;
+
+  beforeEach(async () => {
+    db = createDb();
+    migrate(db, [coreSchema, eventsSchema, queueItemsSchema, queueTransitionsSchema, viewsCustomSchema]);
+    bus = new EventBus(db);
+    queueRepo = new QueueRepository(db, bus);
+    projector = new ViewProjector(db, bus);
+    app = buildApp({ eventBus: bus, projector });
+    // Seed a few qitems so views have something to project.
+    await queueRepo.create({
+      sourceSession: "alice@product-lab",
+      destinationSession: "planning@product-lab",
+      body: "x",
+      nudge: false,
+    });
+    await queueRepo.create({
+      sourceSession: "alice@product-lab",
+      destinationSession: "delivery@product-lab",
+      body: "y",
+      nudge: false,
+    });
+  });
+
+  afterEach(() => db.close());
+
+  it("GET /api/views/list returns built-in + custom view names", async () => {
+    const res = await app.request("/api/views/list");
+    expect(res.status).toBe(200);
+    const body = (await res.json()) as { builtIn: string[]; custom: unknown[] };
+    expect(body.builtIn).toContain("recently-active");
+    expect(body.builtIn).toContain("founder");
+    expect(body.builtIn).toContain("pod-load");
+    expect(body.builtIn).toContain("escalations");
+    expect(body.builtIn).toContain("held");
+    expect(body.builtIn).toContain("activity");
+    expect(body.custom).toHaveLength(0);
+  });
+
+  it("GET /api/views/recently-active returns rows + viewName + generatedAt", async () => {
+    const res = await app.request("/api/views/recently-active");
+    expect(res.status).toBe(200);
+    const body = (await res.json()) as { viewName: string; rowCount: number; rows: unknown[]; generatedAt: string };
+    expect(body.viewName).toBe("recently-active");
+    expect(body.rowCount).toBe(2);
+    expect(body.generatedAt).toMatch(/^\d{4}-\d{2}-\d{2}T/);
+  });
+
+  it("GET /api/views/<unknown-view> returns 404 view_not_found", async () => {
+    const res = await app.request("/api/views/nonexistent-view");
+    expect(res.status).toBe(404);
+    const err = (await res.json()) as { error: string };
+    expect(err.error).toBe("view_not_found");
+  });
+
+  it("GET /api/views/recently-active?limit=1 honors limit query", async () => {
+    const res = await app.request("/api/views/recently-active?limit=1");
+    expect(res.status).toBe(200);
+    const body = (await res.json()) as { rowCount: number };
+    expect(body.rowCount).toBe(1);
+  });
+
+  it("POST /api/views/custom/register registers custom view; show works via name", async () => {
+    const reg = await app.request("/api/views/custom/register", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({
+        viewName: "all-pending",
+        definition: "SELECT qitem_id FROM queue_items WHERE state = 'pending'",
+        registeredBySession: "operator@rig",
+      }),
+    });
+    expect(reg.status).toBe(201);
+    const res = await app.request("/api/views/all-pending");
+    expect(res.status).toBe(200);
+    const body = (await res.json()) as { viewName: string; rowCount: number };
+    expect(body.viewName).toBe("all-pending");
+    expect(body.rowCount).toBeGreaterThan(0);
+  });
+
+  it("POST /api/views/custom/register with reserved name returns 409 view_name_reserved", async () => {
+    const res = await app.request("/api/views/custom/register", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({
+        viewName: "recently-active",
+        definition: "SELECT 1",
+        registeredBySession: "operator@rig",
+      }),
+    });
+    expect(res.status).toBe(409);
+    const err = (await res.json()) as { error: string };
+    expect(err.error).toBe("view_name_reserved");
+  });
+
+  it("R1 SSE pattern: GET /api/views/sse returns 200 + content-type text/event-stream", async () => {
+    const res = await app.request("/api/views/sse");
+    try {
+      expect(res.status).toBe(200);
+      expect(res.headers.get("content-type") ?? "").toContain("text/event-stream");
+    } finally {
+      await res.body?.cancel();
+    }
+  });
+
+  it("R1 SSE pattern: GET /api/views/recently-active/sse returns 200 + content-type text/event-stream", async () => {
+    const res = await app.request("/api/views/recently-active/sse");
+    try {
+      expect(res.status).toBe(200);
+      expect(res.headers.get("content-type") ?? "").toContain("text/event-stream");
+    } finally {
+      await res.body?.cancel();
+    }
+  });
+
+  it("R1 SSE pattern: GET /api/views/sse does NOT return view_not_found (route-order regression guard)", async () => {
+    const res = await app.request("/api/views/sse");
+    try {
+      expect(res.status).not.toBe(404);
+      expect(res.headers.get("content-type") ?? "").not.toContain("application/json");
+    } finally {
+      await res.body?.cancel();
+    }
+  });
+});