feat(daemon,cli): PL-004 Phase C — daemon-native Watchdog supervision tree

Adds the Watchdog primitive from PL-004 Phase C:

Migrations (031, 032):
  - 031_watchdog_jobs: canonical scheduler state (policy enum + intervals
    + lifecycle). Three v1 policies: periodic-reminder, artifact-pool-ready,
    edge-artifact-required. workflow-keepalive is intentionally rejected
    at registration with policy_deferred_to_phase_d (ships in Phase D).
  - 032_watchdog_history: append-only audit. Records sent / skipped /
    terminal outcomes only; pure not_due polls are not recorded.

Domain (TypeScript ports of POC .mjs policies preserving observable semantics):
  - WatchdogJobsRepository (CRUD + register-time enum gate + lifecycle)
  - WatchdogHistoryLog (append-only writer; no update/delete API surface)
  - WatchdogPolicyEngine (resolve policy by name, parse spec_yaml context,
    dispatch evaluate, record meaningful outcomes, emit events, call
    injected delivery for action=send)
  - WatchdogScheduler (tick-driven daemon-native scheduler joining the
    supervision tree; isDue gate; sequential per-tick evaluation;
    graceful in-flight await on stop; SQLite-canonical restart recovery)
  - 3 policy modules + shared artifact-pool helpers

Routes (per Phase A R1 SSE route-order discipline; SSE before /:job_id):
  - POST /api/watchdog/register
  - GET /api/watchdog/list
  - GET /api/watchdog/sse + /watch
  - GET /api/watchdog/:job_id and :/:job_id/status
  - POST /api/watchdog/:job_id/stop

5 new RigEvents:
  - watchdog.evaluation_fired / _skipped / _terminal
  - watchdog.job_registered / _stopped

CLI:
  - rig watchdog register/list/show/status/stop

Wiring:
  - startup.ts constructs repo + log early, engine + scheduler after
    SessionTransport is available (delivery is wired via SessionTransport.send)
  - index.ts starts the scheduler post-listen and stops it on SIGINT/SIGTERM

Tests: 62 daemon tests + 7 CLI tests covering jobs-repo, history-log
append-only contract, three-policy semantics, engine dispatch+history+events,
scheduler dueness/restart-recovery/error-resilience, routes including SSE
route-order regression guard, CLI subcommand registration + happy paths +
workflow-keepalive Phase D rejection surfacing. Full daemon vitest baseline:
2471/2471 pass; CLI watchdog tests 7/7 pass; pre-existing CLI failures
(host-registry/config/specs/preflight/client/chatroom) confirmed present
on Phase B HEAD pre-Phase-C and unrelated to this change.

workflow-keepalive deferred to Phase D paired with workflow_instances.

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

docs/as-built/architecture.md

@@ -325,6 +325,8 @@ Migration boundary:
 - `021_seat_handover_observability.ts` adds seat-handover observability fields.
 - `022_node_codex_config_profile.ts` adds Codex config-profile field on nodes.
 - `023_stream_items.ts` through `027_outbox_entries.ts` add the PL-004 Phase A coordination tables (see § Coordination Layer below).
+- `028_project_classifications.ts` through `030_views_custom.ts` add the PL-004 Phase B classifier + view tables.
+- `031_watchdog_jobs.ts` and `032_watchdog_history.ts` add the PL-004 Phase C watchdog supervision-tree tables. `watchdog_jobs` is the canonical scheduler state (policy, target_session, intervals, lifecycle); `watchdog_history` is the append-only audit of meaningful evaluations (`sent`/`skipped`/`terminal`). Pure `not_due` polls are not recorded. Three v1 policies: `periodic-reminder`, `artifact-pool-ready`, `edge-artifact-required`. The fourth POC policy `workflow-keepalive` is intentionally rejected at registration with `policy_deferred_to_phase_d` and ships in Phase D paired with workflow_instances.
 
 ### Coordination tables (PL-004 Phase A)
 

docs/as-built/cli-reference.md

@@ -700,6 +700,20 @@ Closure-reason semantics:
 - `no-follow-on` — terminal completion, nothing else needed.
 - `escalation` — kicked up to a higher tier (target = escalation target).
 
+## Coordination Watchdog (PL-004 Phase C)
+
+`rig watchdog` registers, lists, inspects, and stops daemon-native scheduler jobs. The scheduler runs inside the daemon supervision tree and persists state in SQLite (`watchdog_jobs`/`watchdog_history`); jobs survive daemon restarts. Three policies are available at v1: `periodic-reminder`, `artifact-pool-ready`, and `edge-artifact-required`. The fourth POC policy `workflow-keepalive` is rejected with `policy_deferred_to_phase_d` and ships in Phase D.
+
+### `rig watchdog`
+
+- `register --spec <path> --policy <name> --target-session <s> --interval-seconds <n> --registered-by <s>` — register a job from a YAML spec; `--active-wake-interval-seconds` and `--scan-interval-seconds` are pool-ready-specific opt-ins.
+- `list` — list all jobs (active + stopped + terminal).
+- `show <job_id>` — show one job.
+- `status <job_id>` — show job + recent evaluation history (last 20 entries).
+- `stop <job_id> [--reason <text>]` — operator stop; scheduler skips the job thereafter.
+
+History records only meaningful evaluations: `sent` (delivery executed), `skipped` (policy returned skip with a reason such as `no_actionable_artifacts`), or `terminal` (policy declared the job done). Pure `not_due` polls are not recorded.
+
 ## Commands Not Present
 
 These are not current top-level `rig` commands:

docs/as-built/codemap.md

@@ -89,6 +89,23 @@ Fast lookup:
 - `packages/daemon/src/db/migrations/025_queue_transitions.ts`: PL-004 Phase A — append-only queue state transition log. Exports: queueTransitionsSchema. Related: `packages/daemon/src/db/migrate.js`.
 - `packages/daemon/src/db/migrations/026_inbox_entries.ts`: PL-004 Phase A — mailbox-style inbox for asynchronous deposit. Exports: inboxEntriesSchema. Related: `packages/daemon/src/db/migrate.js`.
 - `packages/daemon/src/db/migrations/027_outbox_entries.ts`: PL-004 Phase A — sender-side outbox audit. Exports: outboxEntriesSchema. Related: `packages/daemon/src/db/migrate.js`.
+- `packages/daemon/src/db/migrations/028_project_classifications.ts`: PL-004 Phase B — L2 project classifications table (idempotent on stream_item_id; FK to stream_items). Exports: projectClassificationsSchema. Related: `packages/daemon/src/db/migrate.js`.
+- `packages/daemon/src/db/migrations/029_classifier_leases.ts`: PL-004 Phase B — single-writer classifier lease table (partial UNIQUE INDEX on state='active'). Exports: classifierLeasesSchema. Related: `packages/daemon/src/db/migrate.js`.
+- `packages/daemon/src/db/migrations/030_views_custom.ts`: PL-004 Phase B — custom view registry. Exports: viewsCustomSchema. Related: `packages/daemon/src/db/migrate.js`.
+- `packages/daemon/src/db/migrations/031_watchdog_jobs.ts`: PL-004 Phase C — watchdog jobs (canonical scheduler state; policy enum + intervals + lifecycle). Exports: watchdogJobsSchema. Related: `packages/daemon/src/db/migrate.js`.
+- `packages/daemon/src/db/migrations/032_watchdog_history.ts`: PL-004 Phase C — append-only watchdog evaluation history (sent/skipped/terminal; not_due polls not recorded). Exports: watchdogHistorySchema. Related: `packages/daemon/src/db/migrate.js`.
+
+### Domain: Coordination Watchdog (PL-004 Phase C)
+- `packages/daemon/src/domain/watchdog-jobs-repository.ts`: Persistence for watchdog_jobs. Enforces three-policy enum (periodic-reminder, artifact-pool-ready, edge-artifact-required) at register-time; rejects workflow-keepalive with policy_deferred_to_phase_d. Exports: WatchdogJobsRepository, WatchdogJobsError, WatchdogJob, PHASE_C_POLICIES, WORKFLOW_KEEPALIVE_DEFERRED_POLICY. Related: `packages/daemon/src/db/migrations/031_watchdog_jobs.js`.
+- `packages/daemon/src/domain/watchdog-history-log.ts`: Append-only writer for watchdog_history. Records meaningful evaluations only (sent/skipped/terminal). API surface intentionally lacks update/delete. Exports: WatchdogHistoryLog, WatchdogHistoryEntry, WatchdogOutcome. Related: `packages/daemon/src/db/migrations/032_watchdog_history.js`.
+- `packages/daemon/src/domain/watchdog-policy-engine.ts`: Resolves policy by name, parses spec_yaml context, dispatches policy.evaluate(), records meaningful outcomes via history-log, emits watchdog.* events, calls injected delivery callback for action=send. Exports: WatchdogPolicyEngine, DeliveryFn, DeliveryRequest, DeliveryOutcome. Related: `packages/daemon/src/domain/watchdog-jobs-repository.js`, `packages/daemon/src/domain/watchdog-history-log.js`, `packages/daemon/src/domain/event-bus.js`.
+- `packages/daemon/src/domain/watchdog-scheduler.ts`: Tick-driven daemon-native scheduler that joins the supervision tree. Filters listActive() through isDue(), evaluates due jobs sequentially, supports start/stop with graceful in-flight await. Exports: WatchdogScheduler, isDue. Related: `packages/daemon/src/domain/watchdog-policy-engine.js`, `packages/daemon/src/domain/watchdog-jobs-repository.js`.
+- `packages/daemon/src/domain/policies/types.ts`: Policy contract (Policy, PolicyJob, PolicyEvaluation). Exports: Policy, PolicyJob, PolicyEvaluation. Related: none.
+- `packages/daemon/src/domain/policies/periodic-reminder.ts`: Policy that returns send {target, message} unconditionally; throws policy_spec_invalid if context.target.session or context.message missing. Translated from POC `lib/policies/periodic-reminder.mjs`. Exports: periodicReminderPolicy. Related: `packages/daemon/src/domain/policies/types.js`.
+- `packages/daemon/src/domain/policies/artifact-pool-ready.ts`: Policy that scans context.pools and skips with no_actionable_artifacts when empty; otherwise sends a formatted message capped at max_items. Translated from POC `lib/policies/artifact-pool-ready.mjs`. Exports: artifactPoolReadyPolicy. Related: `packages/daemon/src/domain/policies/artifact-pool-helpers.js`, `packages/daemon/src/domain/policies/types.js`.
+- `packages/daemon/src/domain/policies/edge-artifact-required.ts`: Policy that scans source_pools and target_pools, computes source-key per artifact, and sends an edge-repair message for sources lacking matching downstream artifacts. Skips with no_missing_edge_artifacts when all sources matched. Translated from POC `lib/policies/edge-artifact-required.mjs`. Exports: edgeArtifactRequiredPolicy. Related: `packages/daemon/src/domain/policies/artifact-pool-helpers.js`, `packages/daemon/src/domain/policies/types.js`.
+- `packages/daemon/src/domain/policies/artifact-pool-helpers.ts`: Shared filesystem scanner + formatter for artifact-pool-ready and edge-artifact-required. Translated from POC `lib/policies/artifact-pool.js`. Exports: scanArtifactPools, formatArtifactList, sourceKeyFor, ArtifactPoolSpec, ScannedArtifact. Related: none.
+- `packages/daemon/src/routes/watchdog.ts`: Hono route module for `/api/watchdog` (`POST /register`, `GET /list`, `GET /sse`, `GET /watch`, `GET /:job_id`, `GET /:job_id/status`, `POST /:job_id/stop`). SSE/literal paths mounted before /:job_id catchall per Phase A R1 route-order discipline. Exports: watchdogRoutes. Related: `packages/daemon/src/domain/watchdog-jobs-repository.js`, `packages/daemon/src/domain/watchdog-history-log.js`, `packages/daemon/src/domain/event-bus.js`.
 
 ### Domain: Core Runtime
 - `packages/daemon/src/domain/errors.ts`: Domain-specific error types, currently including `RigNotFoundError`. Exports: RigNotFoundError. Related: none.

packages/cli/src/commands/watchdog.ts

@@ -0,0 +1,153 @@
+import { readFileSync } from "node:fs";
+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 watchdog` — coordination primitive Watchdog (PL-004 Phase C).
+ *
+ * Backed by `/api/watchdog`. Operates only via the daemon HTTP API.
+ *
+ * Per PRD § Watchdog: scheduler is daemon-native and joins the
+ * supervision tree. Three policies in scope at v1; workflow-keepalive
+ * deferred to Phase D.
+ */
+
+export interface WatchdogDeps extends StatusDeps {}
+
+async function withClient<T>(
+  deps: WatchdogDeps,
+  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 watchdogCommand(depsOverride?: WatchdogDeps): Command {
+  const cmd = new Command("watchdog").description(
+    "Coordination Watchdog — daemon-native scheduler for periodic-reminder, artifact-pool-ready, edge-artifact-required (PL-004 Phase C; workflow-keepalive ships in Phase D)",
+  );
+  const getDeps = (): WatchdogDeps =>
+    depsOverride ?? {
+      lifecycleDeps: realDeps(),
+      clientFactory: (url: string) => new DaemonClient(url),
+    };
+
+  cmd
+    .command("register")
+    .description("Register a new watchdog job from a YAML spec file")
+    .requiredOption("--spec <path>", "Path to YAML spec file (policy + target + interval + context)")
+    .requiredOption("--policy <policy>", "Policy name (one of: periodic-reminder, artifact-pool-ready, edge-artifact-required)")
+    .requiredOption("--target-session <session>", "Canonical <member>@<rig> target")
+    .requiredOption("--interval-seconds <n>", "Evaluation interval (positive integer)")
+    .requiredOption("--registered-by <session>", "Registering session (for audit)")
+    .option("--active-wake-interval-seconds <n>", "(pool-ready-specific) wake-up cadence when actionable artifacts exist")
+    .option("--scan-interval-seconds <n>", "(pool-ready-specific) artifact pool scan cadence")
+    .option("--json", "JSON output for agents")
+    .action(async (opts: {
+      spec: string;
+      policy: string;
+      targetSession: string;
+      intervalSeconds: string;
+      registeredBy: string;
+      activeWakeIntervalSeconds?: string;
+      scanIntervalSeconds?: string;
+      json?: boolean;
+    }) => {
+      const deps = getDeps();
+      let specYaml: string;
+      try {
+        specYaml = readFileSync(opts.spec, "utf-8");
+      } catch (err) {
+        console.error(`Failed to read spec file ${opts.spec}: ${err instanceof Error ? err.message : err}`);
+        process.exitCode = 1;
+        return;
+      }
+      await withClient(deps, async (client) => {
+        const res = await client.post<unknown>("/api/watchdog/register", {
+          policy: opts.policy,
+          specYaml,
+          targetSession: opts.targetSession,
+          intervalSeconds: Number.parseInt(opts.intervalSeconds, 10),
+          activeWakeIntervalSeconds: opts.activeWakeIntervalSeconds
+            ? Number.parseInt(opts.activeWakeIntervalSeconds, 10)
+            : undefined,
+          scanIntervalSeconds: opts.scanIntervalSeconds
+            ? Number.parseInt(opts.scanIntervalSeconds, 10)
+            : undefined,
+          registeredBySession: opts.registeredBy,
+        });
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
+  cmd
+    .command("list")
+    .description("List all watchdog jobs")
+    .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/watchdog/list");
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
+  cmd
+    .command("show <jobId>")
+    .description("Show one watchdog job")
+    .option("--json", "JSON output for agents")
+    .action(async (jobId: string, opts: { json?: boolean }) => {
+      const deps = getDeps();
+      await withClient(deps, async (client) => {
+        const res = await client.get<unknown>(`/api/watchdog/${encodeURIComponent(jobId)}`);
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
+  cmd
+    .command("status <jobId>")
+    .description("Show one watchdog job + recent evaluation history")
+    .option("--json", "JSON output for agents")
+    .action(async (jobId: string, opts: { json?: boolean }) => {
+      const deps = getDeps();
+      await withClient(deps, async (client) => {
+        const res = await client.get<unknown>(`/api/watchdog/${encodeURIComponent(jobId)}/status`);
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
+  cmd
+    .command("stop <jobId>")
+    .description("Stop a watchdog job (operator-stopped; scheduler skips it)")
+    .option("--reason <text>", "Stop reason (free-form; recorded in terminal_reason)")
+    .option("--json", "JSON output for agents")
+    .action(async (jobId: string, opts: { reason?: string; json?: boolean }) => {
+      const deps = getDeps();
+      await withClient(deps, async (client) => {
+        const res = await client.post<unknown>(`/api/watchdog/${encodeURIComponent(jobId)}/stop`, {
+          reason: opts.reason,
+        });
+        printResult(opts.json ?? false, res.data, res.status);
+      });
+    });
+
+  return cmd;
+}

packages/cli/src/index.ts

@@ -29,6 +29,7 @@ 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 { watchdogCommand, type WatchdogDeps } from "./commands/watchdog.js";
 import { captureCommand } from "./commands/capture.js";
 import { broadcastCommand } from "./commands/broadcast.js";
 import { configCommand } from "./commands/config.js";
@@ -84,6 +85,7 @@ export interface ProgramDeps {
   queueDeps?: QueueDeps;
   projectDeps?: ProjectDeps;
   viewDeps?: ViewDeps;
+  watchdogDeps?: WatchdogDeps;
   captureDeps?: StatusDeps;
   broadcastDeps?: StatusDeps;
   askDeps?: StatusDeps;
@@ -140,6 +142,7 @@ export function createProgram(depsOverride?: ProgramDeps): Command {
   program.addCommand(queueCommand(depsOverride?.queueDeps));
   program.addCommand(projectCommand(depsOverride?.projectDeps));
   program.addCommand(viewCommand(depsOverride?.viewDeps));
+  program.addCommand(watchdogCommand(depsOverride?.watchdogDeps));
   program.addCommand(captureCommand(depsOverride?.captureDeps));
   program.addCommand(broadcastCommand(depsOverride?.broadcastDeps));
   program.addCommand(askCommand(depsOverride?.askDeps));

packages/cli/test/watchdog.test.ts

@@ -0,0 +1,233 @@
+import { mkdtempSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import type { WatchdogDeps } from "../src/commands/watchdog.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: WatchdogDeps;
+  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 WatchdogDeps["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<WatchdogDeps["clientFactory"]>,
+    },
+  };
+}
+
+describe("rig watchdog CLI (PL-004 Phase C)", () => {
+  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("watchdog is registered with all v1 subcommands", () => {
+    const { deps } = makeDeps();
+    const program = createProgram({ watchdogDeps: deps });
+    const cmd = program.commands.find((c) => c.name() === "watchdog");
+    expect(cmd).toBeDefined();
+    const subs = cmd!.commands.map((c) => c.name()).sort();
+    expect(subs).toContain("register");
+    expect(subs).toContain("list");
+    expect(subs).toContain("show");
+    expect(subs).toContain("status");
+    expect(subs).toContain("stop");
+  });
+
+  it("register POSTs /api/watchdog/register with spec file contents + structured fields", async () => {
+    const { deps, calls } = makeDeps({
+      routes: {
+        "POST /api/watchdog/register": {
+          status: 201,
+          data: { jobId: "01J..." },
+        },
+      },
+    });
+    const tmp = mkdtempSync(join(tmpdir(), "wd-cli-"));
+    const spec = join(tmp, "spec.yaml");
+    const yaml =
+      "policy: periodic-reminder\ntarget: alice@rig\ninterval_seconds: 60\ncontext:\n  target:\n    session: alice@rig\n  message: ping\n";
+    writeFileSync(spec, yaml);
+    const program = createProgram({ watchdogDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node",
+      "rig",
+      "watchdog",
+      "register",
+      "--spec",
+      spec,
+      "--policy",
+      "periodic-reminder",
+      "--target-session",
+      "alice@rig",
+      "--interval-seconds",
+      "60",
+      "--registered-by",
+      "ops@kernel",
+      "--json",
+    ]);
+    const call = calls.find((c) => c.path === "/api/watchdog/register");
+    expect(call).toBeDefined();
+    const body = call!.body as Record<string, unknown>;
+    expect(body.policy).toBe("periodic-reminder");
+    expect(body.specYaml).toBe(yaml);
+    expect(body.targetSession).toBe("alice@rig");
+    expect(body.intervalSeconds).toBe(60);
+    expect(body.registeredBySession).toBe("ops@kernel");
+  });
+
+  it("register propagates daemon-side workflow-keepalive rejection (Phase D deferral surfaced via 400)", async () => {
+    const { deps } = makeDeps({
+      routes: {
+        "POST /api/watchdog/register": {
+          status: 400,
+          data: {
+            error: "policy_deferred_to_phase_d",
+            message: "policy 'workflow-keepalive' ships in PL-004 Phase D",
+          },
+        },
+      },
+    });
+    const tmp = mkdtempSync(join(tmpdir(), "wd-cli-"));
+    const spec = join(tmp, "spec.yaml");
+    writeFileSync(spec, "policy: workflow-keepalive\n");
+    const program = createProgram({ watchdogDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node",
+      "rig",
+      "watchdog",
+      "register",
+      "--spec",
+      spec,
+      "--policy",
+      "workflow-keepalive",
+      "--target-session",
+      "alice@rig",
+      "--interval-seconds",
+      "60",
+      "--registered-by",
+      "ops@kernel",
+      "--json",
+    ]);
+    expect(process.exitCode).toBe(1);
+    expect(logs.some((l) => l.includes("policy_deferred_to_phase_d"))).toBe(true);
+  });
+
+  it("list GETs /api/watchdog/list", async () => {
+    const { deps, calls } = makeDeps({
+      routes: { "GET /api/watchdog/list": { status: 200, data: [] } },
+    });
+    const program = createProgram({ watchdogDeps: deps });
+    program.exitOverride();
+    await program.parseAsync(["node", "rig", "watchdog", "list", "--json"]);
+    expect(calls.find((c) => c.path === "/api/watchdog/list")).toBeDefined();
+  });
+
+  it("show GETs /api/watchdog/:jobId", async () => {
+    const { deps, calls } = makeDeps({
+      routes: { "GET /api/watchdog/job-1": { status: 200, data: { jobId: "job-1" } } },
+    });
+    const program = createProgram({ watchdogDeps: deps });
+    program.exitOverride();
+    await program.parseAsync(["node", "rig", "watchdog", "show", "job-1", "--json"]);
+    expect(calls.find((c) => c.path === "/api/watchdog/job-1")).toBeDefined();
+  });
+
+  it("status GETs /api/watchdog/:jobId/status", async () => {
+    const { deps, calls } = makeDeps({
+      routes: {
+        "GET /api/watchdog/job-1/status": {
+          status: 200,
+          data: { job: { jobId: "job-1" }, recentHistory: [] },
+        },
+      },
+    });
+    const program = createProgram({ watchdogDeps: deps });
+    program.exitOverride();
+    await program.parseAsync(["node", "rig", "watchdog", "status", "job-1", "--json"]);
+    expect(calls.find((c) => c.path === "/api/watchdog/job-1/status")).toBeDefined();
+  });
+
+  it("stop POSTs /api/watchdog/:jobId/stop with reason", async () => {
+    const { deps, calls } = makeDeps({
+      routes: {
+        "POST /api/watchdog/job-1/stop": { status: 200, data: { state: "stopped" } },
+      },
+    });
+    const program = createProgram({ watchdogDeps: deps });
+    program.exitOverride();
+    await program.parseAsync([
+      "node",
+      "rig",
+      "watchdog",
+      "stop",
+      "job-1",
+      "--reason",
+      "tester stop",
+      "--json",
+    ]);
+    const call = calls.find((c) => c.path === "/api/watchdog/job-1/stop");
+    expect(call).toBeDefined();
+    expect((call!.body as { reason: string }).reason).toBe("tester stop");
+  });
+});

packages/daemon/src/db/migrations/031_watchdog_jobs.ts

@@ -0,0 +1,46 @@
+import type { Migration } from "../migrate.js";
+
+/**
+ * Watchdog jobs (PL-004 Phase C; daemon-native Watchdog supervision tree).
+ *
+ * Per PRD § Watchdog + slice IMPL § Write Set: SQLite-canonical watchdog
+ * jobs that join the daemon supervision tree. Three policy enum values
+ * SHIPPED at Phase C v1: periodic-reminder, artifact-pool-ready,
+ * edge-artifact-required. The fourth POC policy (workflow-keepalive)
+ * ships in Phase D paired with workflow_instances; daemon registration
+ * MUST reject it with a structured deferred-to-phase-d error.
+ *
+ * State enum:
+ *   active   - scheduler is evaluating this job on its interval
+ *   stopped  - operator stopped the job; scheduler skips it
+ *   terminal - policy declared the job complete (e.g., workflow finished)
+ *
+ * Schema details mirror the POC declarative spec (target_session,
+ * interval_seconds, optional pool-ready-specific scan/active intervals).
+ * spec_yaml preserves the original operator-supplied YAML for audit.
+ */
+export const watchdogJobsSchema: Migration = {
+  name: "031_watchdog_jobs.sql",
+  sql: `
+    CREATE TABLE IF NOT EXISTS watchdog_jobs (
+      job_id TEXT PRIMARY KEY,
+      policy TEXT NOT NULL,
+      spec_yaml TEXT NOT NULL,
+      target_session TEXT NOT NULL,
+      interval_seconds INTEGER NOT NULL,
+      active_wake_interval_seconds INTEGER,
+      scan_interval_seconds INTEGER,
+      last_evaluation_at TEXT,
+      last_fire_at TEXT,
+      state TEXT NOT NULL DEFAULT 'active',
+      registered_by_session TEXT NOT NULL,
+      registered_at TEXT NOT NULL,
+      terminal_reason TEXT
+    );
+    CREATE INDEX IF NOT EXISTS idx_watchdog_jobs_state ON watchdog_jobs(state);
+    CREATE INDEX IF NOT EXISTS idx_watchdog_jobs_policy ON watchdog_jobs(policy);
+    CREATE INDEX IF NOT EXISTS idx_watchdog_jobs_target_session ON watchdog_jobs(target_session);
+    CREATE INDEX IF NOT EXISTS idx_watchdog_jobs_active_next_eval
+      ON watchdog_jobs(last_evaluation_at) WHERE state = 'active';
+  `,
+};

packages/daemon/src/db/migrations/032_watchdog_history.ts

@@ -0,0 +1,47 @@
+import type { Migration } from "../migrate.js";
+
+/**
+ * Watchdog history (PL-004 Phase C; append-only audit log).
+ *
+ * Per PRD § Watchdog + slice IMPL § Guard Checkpoint Focus item 2:
+ * append-only history of meaningful watchdog events. Pure `not_due`
+ * polls are NOT recorded (matches POC; minimizes table size). Only
+ * `sent` (delivery executed), `skipped` (policy ran but skipped per
+ * its own logic — e.g., suppress_if_recent_success or
+ * no_actionable_artifacts), and `terminal` (policy declared job done)
+ * are recorded.
+ *
+ * Append-only contract: writers only INSERT. UPDATE/DELETE are not
+ * exposed by WatchdogHistoryLog API; direct SQL UPDATE/DELETE would
+ * succeed at the DB level (SQLite has no view/role layer) but is a
+ * contract violation enforced at the domain-layer API boundary.
+ *
+ * Outcome enum:
+ *   sent     - policy.evaluate() returned action=send and delivery executed
+ *   skipped  - policy.evaluate() returned action=skip (with reason)
+ *   terminal - policy.evaluate() returned action=terminal (job declared done)
+ *
+ * FK to watchdog_jobs(job_id) for referential integrity. Indexes on
+ * (job_id, evaluated_at DESC) for "recent history for one job" queries
+ * and (outcome, evaluated_at DESC) for cross-job outcome queries.
+ */
+export const watchdogHistorySchema: Migration = {
+  name: "032_watchdog_history.sql",
+  sql: `
+    CREATE TABLE IF NOT EXISTS watchdog_history (
+      history_id TEXT PRIMARY KEY,
+      job_id TEXT NOT NULL REFERENCES watchdog_jobs(job_id),
+      evaluated_at TEXT NOT NULL,
+      outcome TEXT NOT NULL,
+      skip_reason TEXT,
+      delivery_target_session TEXT,
+      delivery_status TEXT,
+      delivery_message TEXT,
+      evaluation_notes TEXT
+    );
+    CREATE INDEX IF NOT EXISTS idx_watchdog_history_job_recent
+      ON watchdog_history(job_id, evaluated_at DESC);
+    CREATE INDEX IF NOT EXISTS idx_watchdog_history_outcome_recent
+      ON watchdog_history(outcome, evaluated_at DESC);
+  `,
+};

packages/daemon/src/domain/policies/artifact-pool-helpers.ts

@@ -0,0 +1,136 @@
+// PL-004 Phase C: shared artifact-pool helpers (TypeScript port of POC
+// `lib/policies/artifact-pool.js`).
+//
+// Used by artifact-pool-ready and edge-artifact-required policies. Pure
+// filesystem scanner; no event-bus, no DB, no Hono. Reverse-engineers
+// the POC contract for include_statuses + frontmatter extraction.
+
+import { existsSync, readdirSync, readFileSync, statSync } from "node:fs";
+import { join, basename } from "node:path";
+
+export interface ArtifactPoolSpec {
+  /** Absolute path to the pool directory (or `paths` array variant). */
+  path?: string;
+  paths?: string[];
+  /** File extensions to include (default: ['.md']). */
+  extensions?: string[];
+  /** Filter on frontmatter `status:` field (default: include all). */
+  include_statuses?: string[];
+  /** Frontmatter field used for keying artifacts (default: 'entry'). */
+  key_field?: string;
+}
+
+export interface ScannedArtifact {
+  /** Absolute path to the artifact file. */
+  path: string;
+  /** Parsed YAML frontmatter (top-level keys only). */
+  frontmatter: Record<string, string>;
+}
+
+const FRONTMATTER_RE = /^---\n([\s\S]*?)\n---/;
+
+/**
+ * Parse top-level YAML frontmatter from a markdown-style document.
+ * Only extracts `key: value` pairs (no nested structures, no arrays).
+ * Mirrors POC's lightweight extraction in lib/policies/artifact-pool.js.
+ */
+function parseFrontmatter(content: string): Record<string, string> {
+  const match = FRONTMATTER_RE.exec(content);
+  if (!match) return {};
+  const block = match[1] ?? "";
+  const fm: Record<string, string> = {};
+  for (const line of block.split("\n")) {
+    const idx = line.indexOf(":");
+    if (idx === -1) continue;
+    const key = line.slice(0, idx).trim();
+    let value = line.slice(idx + 1).trim();
+    if (value.startsWith('"') && value.endsWith('"')) value = value.slice(1, -1);
+    if (value.startsWith("'") && value.endsWith("'")) value = value.slice(1, -1);
+    if (key) fm[key] = value;
+  }
+  return fm;
+}
+
+/**
+ * Scan a single pool directory for artifacts matching extensions +
+ * include_statuses. Returns ScannedArtifact[] with parsed frontmatter.
+ */
+function scanPool(spec: ArtifactPoolSpec & { path: string }): ScannedArtifact[] {
+  if (!existsSync(spec.path)) return [];
+  let entries: string[];
+  try {
+    entries = readdirSync(spec.path);
+  } catch {
+    return [];
+  }
+  const exts = spec.extensions ?? [".md"];
+  const includeStatuses = spec.include_statuses;
+  const out: ScannedArtifact[] = [];
+  for (const entry of entries) {
+    if (!exts.some((e) => entry.endsWith(e))) continue;
+    const fullPath = join(spec.path, entry);
+    let st;
+    try {
+      st = statSync(fullPath);
+    } catch {
+      continue;
+    }
+    if (!st.isFile()) continue;
+    let content: string;
+    try {
+      content = readFileSync(fullPath, "utf-8");
+    } catch {
+      continue;
+    }
+    const fm = parseFrontmatter(content);
+    if (includeStatuses && includeStatuses.length > 0) {
+      if (!includeStatuses.includes(fm.status ?? "")) continue;
+    }
+    out.push({ path: fullPath, frontmatter: fm });
+  }
+  return out;
+}
+
+/**
+ * Scan one or more artifact pools per spec(s). Mirrors POC
+ * `scanArtifactPools(pools)`. Returns the flat union of matches across
+ * all pools. Missing/unreadable pools yield empty contributions.
+ */
+export async function scanArtifactPools(
+  pools: ArtifactPoolSpec | ArtifactPoolSpec[] | undefined,
+): Promise<ScannedArtifact[]> {
+  if (!pools) return [];
+  const list = Array.isArray(pools) ? pools : [pools];
+  const out: ScannedArtifact[] = [];
+  for (const spec of list) {
+    if (spec.path) {
+      out.push(...scanPool({ ...spec, path: spec.path }));
+    } else if (Array.isArray(spec.paths)) {
+      for (const p of spec.paths) out.push(...scanPool({ ...spec, path: p }));
+    }
+  }
+  return out;
+}
+
+/**
+ * Format an artifact list as bullet lines. Mirrors POC formatArtifactList.
+ */
+export function formatArtifactList(artifacts: ScannedArtifact[], maxItems: number): string {
+  return artifacts
+    .slice(0, maxItems)
+    .map((a) => `- ${basename(a.path)} (${a.path})`)
+    .join("\n");
+}
+
+/**
+ * Compute the source-key for an artifact (used by edge-artifact-required).
+ * Mirrors POC sourceKeyFor: prefer frontmatter[keyField], else basename
+ * sans .md extension.
+ */
+export function sourceKeyFor(artifact: ScannedArtifact, keyField: string): string {
+  const value = artifact.frontmatter[keyField];
+  if (value === undefined || value === null || value === "") {
+    return basename(artifact.path).replace(/\.md$/, "");
+  }
+  return value;
+}

packages/daemon/src/domain/policies/artifact-pool-ready.ts

@@ -0,0 +1,53 @@
+// PL-004 Phase C: artifact-pool-ready policy (TypeScript port of
+// POC `lib/policies/artifact-pool-ready.mjs`, 29 lines).
+//
+// Scans context.pools, returns:
+//   - skip { reason: "no_actionable_artifacts" } if pool empty
+//   - send { message: "Artifact pool ready: <label> has N actionable
+//     artifact(s)..." + bullet list capped at max_items (default 5) }
+
+import type { Policy, PolicyEvaluation, PolicyJob } from "./types.js";
+import {
+  type ArtifactPoolSpec,
+  formatArtifactList,
+  scanArtifactPools,
+} from "./artifact-pool-helpers.js";
+
+interface ArtifactPoolReadyContext {
+  target?: { session?: string };
+  pools?: ArtifactPoolSpec | ArtifactPoolSpec[];
+  label?: string;
+  max_items?: number;
+  message_template?: string;
+}
+
+export const artifactPoolReadyPolicy: Policy = {
+  name: "artifact-pool-ready",
+  async evaluate(job: PolicyJob): Promise<PolicyEvaluation> {
+    const context = job.context as ArtifactPoolReadyContext;
+    const targetSession = context.target?.session;
+    if (!targetSession) {
+      throw Object.assign(new Error("artifact-pool-ready requires context.target.session"), {
+        code: "policy_spec_invalid",
+        policy: "artifact-pool-ready",
+        field: "context.target.session",
+      });
+    }
+    const artifacts = await scanArtifactPools(context.pools);
+    if (artifacts.length === 0) {
+      return { action: "skip", reason: "no_actionable_artifacts" };
+    }
+    const label = context.label ?? "artifact pool";
+    const maxItems = context.max_items ?? 5;
+    const list = formatArtifactList(artifacts, maxItems);
+    const message =
+      context.message_template ??
+      `Artifact pool ready: ${label} has ${artifacts.length} actionable artifact(s).\n\n${list}`;
+    return {
+      action: "send",
+      target: targetSession,
+      message,
+      notes: { artifact_count: artifacts.length, label },
+    };
+  },
+};

packages/daemon/src/domain/policies/edge-artifact-required.ts

@@ -0,0 +1,89 @@
+// PL-004 Phase C: edge-artifact-required policy (TypeScript port of
+// POC `lib/policies/edge-artifact-required.mjs`, 70 lines).
+//
+// Scans both source pools and target pools. For each source artifact,
+// computes its source-key (frontmatter[key_field] OR basename-sans-md).
+// If targets array contains no artifact with matching source-key,
+// the source is "missing a downstream edge artifact." Emits a send
+// only when one or more sources are missing edges; otherwise skip with
+// reason "no_missing_edge_artifacts".
+
+import type { Policy, PolicyEvaluation, PolicyJob } from "./types.js";
+import {
+  type ArtifactPoolSpec,
+  type ScannedArtifact,
+  formatArtifactList,
+  scanArtifactPools,
+  sourceKeyFor,
+} from "./artifact-pool-helpers.js";
+
+interface EdgeArtifactRequiredContext {
+  target?: { session?: string };
+  source_pools?: ArtifactPoolSpec | ArtifactPoolSpec[];
+  target_pools?: ArtifactPoolSpec | ArtifactPoolSpec[];
+  source_key_field?: string;
+  target_key_field?: string;
+  label?: string;
+  max_items?: number;
+  message_template?: string;
+}
+
+function targetContainsKey(
+  targets: ScannedArtifact[],
+  key: string,
+  targetKeyField: string,
+): boolean {
+  for (const t of targets) {
+    if (sourceKeyFor(t, targetKeyField) === key) return true;
+  }
+  return false;
+}
+
+export const edgeArtifactRequiredPolicy: Policy = {
+  name: "edge-artifact-required",
+  async evaluate(job: PolicyJob): Promise<PolicyEvaluation> {
+    const context = job.context as EdgeArtifactRequiredContext;
+    const targetSession = context.target?.session;
+    if (!targetSession) {
+      throw Object.assign(new Error("edge-artifact-required requires context.target.session"), {
+        code: "policy_spec_invalid",
+        policy: "edge-artifact-required",
+        field: "context.target.session",
+      });
+    }
+    const sourceKeyField = context.source_key_field ?? "entry";
+    const targetKeyField = context.target_key_field ?? "entry";
+    const sources = await scanArtifactPools(context.source_pools);
+    // POC override: target scan ignores include_statuses (so we can detect
+    // any matching downstream artifact regardless of its lifecycle state).
+    const targetPoolsRaw = context.target_pools;
+    const targetPoolsList = targetPoolsRaw
+      ? Array.isArray(targetPoolsRaw)
+        ? targetPoolsRaw
+        : [targetPoolsRaw]
+      : [];
+    const targets = await scanArtifactPools(
+      targetPoolsList.map((p) => ({ ...p, include_statuses: [] })),
+    );
+    const missing: ScannedArtifact[] = [];
+    for (const s of sources) {
+      const key = sourceKeyFor(s, sourceKeyField);
+      if (!targetContainsKey(targets, key, targetKeyField)) missing.push(s);
+    }
+    if (missing.length === 0) {
+      return { action: "skip", reason: "no_missing_edge_artifacts" };
+    }
+    const label = context.label ?? "edge";
+    const maxItems = context.max_items ?? 5;
+    const list = formatArtifactList(missing, maxItems);
+    const message =
+      context.message_template ??
+      `Artifact edge repair needed: ${label} has ${missing.length} upstream artifact(s) with no matching downstream artifact.\n\n${list}`;
+    return {
+      action: "send",
+      target: targetSession,
+      message,
+      notes: { missing_count: missing.length, label },
+    };
+  },
+};

packages/daemon/src/domain/policies/periodic-reminder.ts

@@ -0,0 +1,38 @@
+// PL-004 Phase C: periodic-reminder policy (TypeScript port of
+// POC `lib/policies/periodic-reminder.mjs`, 15 lines).
+//
+// Contract: explicit context.target.session and context.message both
+// required. Throws for missing either (hard contract violation).
+// Returns action=send unconditionally when invoked (the scheduler
+// gates on interval; this policy itself is stateless).
+
+import type { Policy, PolicyEvaluation, PolicyJob } from "./types.js";
+
+interface PeriodicReminderContext {
+  target?: { session?: string };
+  message?: string;
+}
+
+export const periodicReminderPolicy: Policy = {
+  name: "periodic-reminder",
+  async evaluate(job: PolicyJob): Promise<PolicyEvaluation> {
+    const context = job.context as PeriodicReminderContext;
+    const targetSession = context.target?.session;
+    const message = context.message;
+    if (!targetSession) {
+      throw Object.assign(new Error("periodic-reminder requires context.target.session"), {
+        code: "policy_spec_invalid",
+        policy: "periodic-reminder",
+        field: "context.target.session",
+      });
+    }
+    if (!message) {
+      throw Object.assign(new Error("periodic-reminder requires context.message"), {
+        code: "policy_spec_invalid",
+        policy: "periodic-reminder",
+        field: "context.message",
+      });
+    }
+    return { action: "send", target: targetSession, message };
+  },
+};

packages/daemon/src/domain/policies/types.ts

@@ -0,0 +1,39 @@
+// PL-004 Phase C: shared policy contract types.
+//
+// Each watchdog policy implements `evaluate(job)` returning a
+// PolicyEvaluation. Pure: no side-effects, no event-bus, no DB.
+// The watchdog-policy-engine maps `action: send` to a delivery call,
+// records meaningful outcomes via watchdog-history-log, and emits
+// the corresponding RigEvent.
+
+export interface PolicyJob {
+  jobId: string;
+  policy: string;
+  targetSession: string;
+  intervalSeconds: number;
+  activeWakeIntervalSeconds: number | null;
+  scanIntervalSeconds: number | null;
+  /** Parsed `context:` block from the operator-supplied spec_yaml. */
+  context: Record<string, unknown>;
+  lastEvaluationAt: string | null;
+  lastFireAt: string | null;
+  registeredBySession: string;
+  registeredAt: string;
+}
+
+export type PolicyEvaluation =
+  | { action: "send"; target: string; message: string; notes?: Record<string, unknown> }
+  | { action: "skip"; reason: string; notes?: Record<string, unknown> }
+  | { action: "terminal"; reason: string; notes?: Record<string, unknown> };
+
+export interface Policy {
+  /** Stable identifier matching watchdog_jobs.policy enum. */
+  readonly name: string;
+  /**
+   * Pure evaluation. No I/O beyond filesystem reads (artifact-pool
+   * scans). Throws only for hard contract violations (missing required
+   * spec fields); recoverable conditions (no actionable artifacts,
+   * recent successful run) MUST return action=skip.
+   */
+  evaluate(job: PolicyJob): Promise<PolicyEvaluation>;
+}

packages/daemon/src/domain/types.ts

@@ -172,7 +172,17 @@ export type RigEvent =
   | { 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 };
+  | { type: "view.changed"; viewName: string; cause: string }
+  // PL-004 Phase C: daemon-native Watchdog supervision tree events.
+  // Three policies in scope (periodic-reminder, artifact-pool-ready,
+  // edge-artifact-required); workflow-keepalive deferred to Phase D.
+  // Pure `not_due` polls are NOT recorded in history and NOT emitted
+  // as events; only meaningful evaluations + lifecycle transitions are.
+  | { type: "watchdog.evaluation_fired"; jobId: string; policy: string; targetSession: string; deliveryStatus: string }
+  | { type: "watchdog.evaluation_skipped"; jobId: string; policy: string; skipReason: string }
+  | { type: "watchdog.evaluation_terminal"; jobId: string; policy: string; terminalReason: string }
+  | { type: "watchdog.job_registered"; jobId: string; policy: string; targetSession: string; registeredBy: string }
+  | { type: "watchdog.job_stopped"; jobId: string; reason: string };
 
 export type PersistedEvent = RigEvent & {
   seq: number;

packages/daemon/src/domain/watchdog-history-log.ts

@@ -0,0 +1,136 @@
+import type Database from "better-sqlite3";
+import { ulid } from "ulid";
+
+/**
+ * Watchdog history log (PL-004 Phase C; append-only audit writer).
+ *
+ * Per slice IMPL § Guard Checkpoint Focus item 2:
+ * - Pure `not_due` polls are NOT recorded (matches POC; minimizes
+ *   table size). Only `sent` / `skipped` / `terminal` outcomes pass
+ *   through this log.
+ * - Append-only API surface: only `record()` exposed. UPDATE/DELETE
+ *   are not exposed. SQLite has no view/role layer so direct SQL
+ *   could mutate, but the contract is enforced at this domain
+ *   boundary.
+ */
+
+export type WatchdogOutcome = "sent" | "skipped" | "terminal";
+
+export interface WatchdogHistoryRecordInput {
+  jobId: string;
+  evaluatedAt: string;
+  outcome: WatchdogOutcome;
+  skipReason?: string | null;
+  deliveryTargetSession?: string | null;
+  deliveryStatus?: string | null;
+  deliveryMessage?: string | null;
+  evaluationNotes?: Record<string, unknown> | null;
+}
+
+export interface WatchdogHistoryEntry {
+  historyId: string;
+  jobId: string;
+  evaluatedAt: string;
+  outcome: WatchdogOutcome;
+  skipReason: string | null;
+  deliveryTargetSession: string | null;
+  deliveryStatus: string | null;
+  deliveryMessage: string | null;
+  evaluationNotes: Record<string, unknown> | null;
+}
+
+interface HistoryRow {
+  history_id: string;
+  job_id: string;
+  evaluated_at: string;
+  outcome: string;
+  skip_reason: string | null;
+  delivery_target_session: string | null;
+  delivery_status: string | null;
+  delivery_message: string | null;
+  evaluation_notes: string | null;
+}
+
+export class WatchdogHistoryLog {
+  constructor(private readonly db: Database.Database) {}
+
+  /**
+   * Append a meaningful evaluation outcome. Returns the persisted entry.
+   * Caller MUST NOT invoke this for `not_due` pure-skip outcomes — those
+   * are not recorded (per POC + IMPL guidance).
+   */
+  record(input: WatchdogHistoryRecordInput): WatchdogHistoryEntry {
+    const historyId = ulid();
+    this.db
+      .prepare(
+        `INSERT INTO watchdog_history (
+          history_id, job_id, evaluated_at, outcome,
+          skip_reason, delivery_target_session, delivery_status,
+          delivery_message, evaluation_notes
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+      )
+      .run(
+        historyId,
+        input.jobId,
+        input.evaluatedAt,
+        input.outcome,
+        input.skipReason ?? null,
+        input.deliveryTargetSession ?? null,
+        input.deliveryStatus ?? null,
+        input.deliveryMessage ?? null,
+        input.evaluationNotes ? JSON.stringify(input.evaluationNotes) : null,
+      );
+    return {
+      historyId,
+      jobId: input.jobId,
+      evaluatedAt: input.evaluatedAt,
+      outcome: input.outcome,
+      skipReason: input.skipReason ?? null,
+      deliveryTargetSession: input.deliveryTargetSession ?? null,
+      deliveryStatus: input.deliveryStatus ?? null,
+      deliveryMessage: input.deliveryMessage ?? null,
+      evaluationNotes: input.evaluationNotes ?? null,
+    };
+  }
+
+  listForJob(jobId: string, limit = 50): WatchdogHistoryEntry[] {
+    const rows = this.db
+      .prepare(
+        `SELECT * FROM watchdog_history WHERE job_id = ?
+         ORDER BY evaluated_at DESC, history_id DESC LIMIT ?`,
+      )
+      .all(jobId, limit) as HistoryRow[];
+    return rows.map(rowToEntry);
+  }
+
+  listAll(limit = 100): WatchdogHistoryEntry[] {
+    const rows = this.db
+      .prepare(
+        `SELECT * FROM watchdog_history
+         ORDER BY evaluated_at DESC, history_id DESC LIMIT ?`,
+      )
+      .all(limit) as HistoryRow[];
+    return rows.map(rowToEntry);
+  }
+
+  countForJob(jobId: string): number {
+    const row = this.db
+      .prepare(`SELECT COUNT(*) AS n FROM watchdog_history WHERE job_id = ?`)
+      .get(jobId) as { n: number };
+    return row.n;
+  }
+}
+
+function rowToEntry(row: HistoryRow): WatchdogHistoryEntry {
+  return {
+    historyId: row.history_id,
+    jobId: row.job_id,
+    evaluatedAt: row.evaluated_at,
+    outcome: row.outcome as WatchdogOutcome,
+    skipReason: row.skip_reason,
+    deliveryTargetSession: row.delivery_target_session,
+    deliveryStatus: row.delivery_status,
+    deliveryMessage: row.delivery_message,
+    evaluationNotes: row.evaluation_notes ? JSON.parse(row.evaluation_notes) : null,
+  };
+}

packages/daemon/src/domain/watchdog-jobs-repository.ts

@@ -0,0 +1,230 @@
+import type Database from "better-sqlite3";
+import { ulid } from "ulid";
+
+/**
+ * Watchdog jobs repository (PL-004 Phase C).
+ *
+ * Owns all reads/writes against `watchdog_jobs`. Pure persistence; no
+ * event-bus, no scheduler, no policy dispatch. Composed by the
+ * scheduler and policy engine.
+ *
+ * Phase C v1 enforces the policy enum against three values:
+ * periodic-reminder, artifact-pool-ready, edge-artifact-required.
+ * `workflow-keepalive` is intentionally rejected with a structured
+ * `policy_deferred_to_phase_d` error.
+ */
+
+export const PHASE_C_POLICIES = [
+  "periodic-reminder",
+  "artifact-pool-ready",
+  "edge-artifact-required",
+] as const;
+
+export type WatchdogPolicyName = (typeof PHASE_C_POLICIES)[number];
+
+export const WORKFLOW_KEEPALIVE_DEFERRED_POLICY = "workflow-keepalive";
+
+export type WatchdogJobState = "active" | "stopped" | "terminal";
+
+export interface WatchdogJob {
+  jobId: string;
+  policy: WatchdogPolicyName;
+  specYaml: string;
+  targetSession: string;
+  intervalSeconds: number;
+  activeWakeIntervalSeconds: number | null;
+  scanIntervalSeconds: number | null;
+  lastEvaluationAt: string | null;
+  lastFireAt: string | null;
+  state: WatchdogJobState;
+  registeredBySession: string;
+  registeredAt: string;
+  terminalReason: string | null;
+}
+
+export interface RegisterWatchdogJobInput {
+  policy: string;
+  specYaml: string;
+  targetSession: string;
+  intervalSeconds: number;
+  activeWakeIntervalSeconds?: number | null;
+  scanIntervalSeconds?: number | null;
+  registeredBySession: string;
+}
+
+interface JobRow {
+  job_id: string;
+  policy: string;
+  spec_yaml: string;
+  target_session: string;
+  interval_seconds: number;
+  active_wake_interval_seconds: number | null;
+  scan_interval_seconds: number | null;
+  last_evaluation_at: string | null;
+  last_fire_at: string | null;
+  state: string;
+  registered_by_session: string;
+  registered_at: string;
+  terminal_reason: string | null;
+}
+
+export class WatchdogJobsError extends Error {
+  constructor(
+    public readonly code: string,
+    message: string,
+    public readonly details?: Record<string, unknown>,
+  ) {
+    super(message);
+    this.name = "WatchdogJobsError";
+  }
+}
+
+export class WatchdogJobsRepository {
+  constructor(
+    private readonly db: Database.Database,
+    private readonly now: () => Date = () => new Date(),
+  ) {}
+
+  register(input: RegisterWatchdogJobInput): WatchdogJob {
+    if (input.policy === WORKFLOW_KEEPALIVE_DEFERRED_POLICY) {
+      throw new WatchdogJobsError(
+        "policy_deferred_to_phase_d",
+        `policy 'workflow-keepalive' ships in PL-004 Phase D paired with workflow_instances; not available in Phase C v1`,
+        { policy: input.policy, phase: "C", deferredTo: "D" },
+      );
+    }
+    if (!PHASE_C_POLICIES.includes(input.policy as WatchdogPolicyName)) {
+      throw new WatchdogJobsError(
+        "policy_unknown",
+        `unknown watchdog policy '${input.policy}'; Phase C v1 supports: ${PHASE_C_POLICIES.join(", ")}`,
+        { policy: input.policy, supported: [...PHASE_C_POLICIES] },
+      );
+    }
+    if (!Number.isInteger(input.intervalSeconds) || input.intervalSeconds <= 0) {
+      throw new WatchdogJobsError(
+        "interval_invalid",
+        `interval_seconds must be a positive integer (got ${input.intervalSeconds})`,
+        { intervalSeconds: input.intervalSeconds },
+      );
+    }
+    if (!input.targetSession || !input.targetSession.includes("@")) {
+      throw new WatchdogJobsError(
+        "target_session_invalid",
+        `target_session must be canonical '<member>@<rig>' (got '${input.targetSession}')`,
+        { targetSession: input.targetSession },
+      );
+    }
+    const jobId = ulid();
+    const registeredAt = this.now().toISOString();
+    this.db
+      .prepare(
+        `INSERT INTO watchdog_jobs (
+          job_id, policy, spec_yaml, target_session,
+          interval_seconds, active_wake_interval_seconds, scan_interval_seconds,
+          last_evaluation_at, last_fire_at, state,
+          registered_by_session, registered_at, terminal_reason
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, NULL, NULL, 'active', ?, ?, NULL)`,
+      )
+      .run(
+        jobId,
+        input.policy,
+        input.specYaml,
+        input.targetSession,
+        input.intervalSeconds,
+        input.activeWakeIntervalSeconds ?? null,
+        input.scanIntervalSeconds ?? null,
+        input.registeredBySession,
+        registeredAt,
+      );
+    return this.getByIdOrThrow(jobId);
+  }
+
+  getById(jobId: string): WatchdogJob | null {
+    const row = this.db
+      .prepare(`SELECT * FROM watchdog_jobs WHERE job_id = ?`)
+      .get(jobId) as JobRow | undefined;
+    return row ? rowToJob(row) : null;
+  }
+
+  getByIdOrThrow(jobId: string): WatchdogJob {
+    const job = this.getById(jobId);
+    if (!job) {
+      throw new WatchdogJobsError(
+        "job_not_found",
+        `watchdog job ${jobId} not found`,
+        { jobId },
+      );
+    }
+    return job;
+  }
+
+  listAll(): WatchdogJob[] {
+    const rows = this.db
+      .prepare(`SELECT * FROM watchdog_jobs ORDER BY registered_at ASC`)
+      .all() as JobRow[];
+    return rows.map(rowToJob);
+  }
+
+  listActive(): WatchdogJob[] {
+    const rows = this.db
+      .prepare(`SELECT * FROM watchdog_jobs WHERE state = 'active' ORDER BY registered_at ASC`)
+      .all() as JobRow[];
+    return rows.map(rowToJob);
+  }
+
+  recordEvaluation(jobId: string, evaluatedAt: string, fired: boolean): void {
+    if (fired) {
+      this.db
+        .prepare(
+          `UPDATE watchdog_jobs SET last_evaluation_at = ?, last_fire_at = ? WHERE job_id = ?`,
+        )
+        .run(evaluatedAt, evaluatedAt, jobId);
+    } else {
+      this.db
+        .prepare(`UPDATE watchdog_jobs SET last_evaluation_at = ? WHERE job_id = ?`)
+        .run(evaluatedAt, jobId);
+    }
+  }
+
+  markTerminal(jobId: string, reason: string): void {
+    this.db
+      .prepare(
+        `UPDATE watchdog_jobs SET state = 'terminal', terminal_reason = ? WHERE job_id = ?`,
+      )
+      .run(reason, jobId);
+  }
+
+  stop(jobId: string, reason = "operator_stopped"): WatchdogJob {
+    const existing = this.getByIdOrThrow(jobId);
+    if (existing.state === "terminal") {
+      throw new WatchdogJobsError(
+        "job_terminal",
+        `cannot stop watchdog job ${jobId}: state is terminal`,
+        { jobId, state: existing.state },
+      );
+    }
+    if (existing.state === "stopped") return existing;
+    this.db
+      .prepare(`UPDATE watchdog_jobs SET state = 'stopped', terminal_reason = ? WHERE job_id = ?`)
+      .run(reason, jobId);
+    return this.getByIdOrThrow(jobId);
+  }
+}
+
+function rowToJob(row: JobRow): WatchdogJob {
+  return {
+    jobId: row.job_id,
+    policy: row.policy as WatchdogPolicyName,
+    specYaml: row.spec_yaml,
+    targetSession: row.target_session,
+    intervalSeconds: row.interval_seconds,
+    activeWakeIntervalSeconds: row.active_wake_interval_seconds,
+    scanIntervalSeconds: row.scan_interval_seconds,
+    lastEvaluationAt: row.last_evaluation_at,
+    lastFireAt: row.last_fire_at,
+    state: row.state as WatchdogJobState,
+    registeredBySession: row.registered_by_session,
+    registeredAt: row.registered_at,
+    terminalReason: row.terminal_reason,
+  };
+}

packages/daemon/src/domain/watchdog-policy-engine.ts

@@ -0,0 +1,411 @@
+import type { EventBus } from "./event-bus.js";
+import type { Policy, PolicyEvaluation, PolicyJob } from "./policies/types.js";
+import { artifactPoolReadyPolicy } from "./policies/artifact-pool-ready.js";
+import { edgeArtifactRequiredPolicy } from "./policies/edge-artifact-required.js";
+import { periodicReminderPolicy } from "./policies/periodic-reminder.js";
+import {
+  type WatchdogJob,
+  type WatchdogJobsRepository,
+} from "./watchdog-jobs-repository.js";
+import type { WatchdogHistoryEntry, WatchdogHistoryLog } from "./watchdog-history-log.js";
+
+/**
+ * Watchdog policy engine (PL-004 Phase C).
+ *
+ * Owns:
+ *   1. Policy resolution by name (registry).
+ *   2. spec_yaml → context parse for the policy.
+ *   3. Dispatch evaluate(job) → PolicyEvaluation.
+ *   4. For action=send: invoke delivery callback, record history,
+ *      emit watchdog.evaluation_fired.
+ *   5. For action=skip: record history, emit watchdog.evaluation_skipped.
+ *   6. For action=terminal: mark job terminal, record history, emit
+ *      watchdog.evaluation_terminal.
+ *
+ * The actual delivery is injected as a callback so tests can stub it
+ * and so the daemon supervision tree can wire it to SessionTransport
+ * without this module needing a concrete dependency.
+ */
+
+export interface DeliveryRequest {
+  targetSession: string;
+  message: string;
+}
+
+export interface DeliveryOutcome {
+  status: "ok" | "failed";
+  error?: string;
+}
+
+export type DeliveryFn = (req: DeliveryRequest) => Promise<DeliveryOutcome>;
+
+export interface PolicyContextParser {
+  /** Parse the operator-supplied spec_yaml into the `context:` block. */
+  (specYaml: string): Record<string, unknown>;
+}
+
+interface WatchdogPolicyEngineDeps {
+  jobsRepo: WatchdogJobsRepository;
+  historyLog: WatchdogHistoryLog;
+  eventBus: EventBus;
+  deliver: DeliveryFn;
+  /** Optional override: defaults to a YAML-aware parser. */
+  parseContext?: PolicyContextParser;
+  now?: () => Date;
+}
+
+const POLICY_REGISTRY: Map<string, Policy> = new Map([
+  [periodicReminderPolicy.name, periodicReminderPolicy],
+  [artifactPoolReadyPolicy.name, artifactPoolReadyPolicy],
+  [edgeArtifactRequiredPolicy.name, edgeArtifactRequiredPolicy],
+]);
+
+export interface EvaluationResult {
+  job: WatchdogJob;
+  outcome: PolicyEvaluation;
+  history: WatchdogHistoryEntry | null;
+  delivery: DeliveryOutcome | null;
+}
+
+export class WatchdogPolicyEngine {
+  private readonly jobsRepo: WatchdogJobsRepository;
+  private readonly historyLog: WatchdogHistoryLog;
+  private readonly eventBus: EventBus;
+  private readonly deliver: DeliveryFn;
+  private readonly parseContext: PolicyContextParser;
+  private readonly now: () => Date;
+
+  constructor(deps: WatchdogPolicyEngineDeps) {
+    this.jobsRepo = deps.jobsRepo;
+    this.historyLog = deps.historyLog;
+    this.eventBus = deps.eventBus;
+    this.deliver = deps.deliver;
+    this.parseContext = deps.parseContext ?? defaultParseContext;
+    this.now = deps.now ?? (() => new Date());
+  }
+
+  resolvePolicy(name: string): Policy | undefined {
+    return POLICY_REGISTRY.get(name);
+  }
+
+  /**
+   * Evaluate one watchdog job. Side-effects: history record (for
+   * meaningful outcomes), event emission, and (on send) delivery.
+   *
+   * Pure `not_due` decisions are NOT this engine's concern; the
+   * scheduler decides dueness and only calls evaluate for due jobs.
+   * If the policy itself returns skip with an internal reason
+   * (no_actionable_artifacts, etc.) we DO record that.
+   */
+  async evaluate(job: WatchdogJob): Promise<EvaluationResult> {
+    const policy = this.resolvePolicy(job.policy);
+    if (!policy) {
+      const reason = `unknown_policy:${job.policy}`;
+      this.jobsRepo.markTerminal(job.jobId, reason);
+      const evaluatedAt = this.now().toISOString();
+      const history = this.historyLog.record({
+        jobId: job.jobId,
+        evaluatedAt,
+        outcome: "terminal",
+        skipReason: reason,
+      });
+      this.eventBus.emit({
+        type: "watchdog.evaluation_terminal",
+        jobId: job.jobId,
+        policy: job.policy,
+        terminalReason: reason,
+      });
+      return {
+        job: this.jobsRepo.getByIdOrThrow(job.jobId),
+        outcome: { action: "terminal", reason },
+        history,
+        delivery: null,
+      };
+    }
+
+    const context = this.parseContext(job.specYaml);
+    const policyJob: PolicyJob = {
+      jobId: job.jobId,
+      policy: job.policy,
+      targetSession: job.targetSession,
+      intervalSeconds: job.intervalSeconds,
+      activeWakeIntervalSeconds: job.activeWakeIntervalSeconds,
+      scanIntervalSeconds: job.scanIntervalSeconds,
+      context,
+      lastEvaluationAt: job.lastEvaluationAt,
+      lastFireAt: job.lastFireAt,
+      registeredBySession: job.registeredBySession,
+      registeredAt: job.registeredAt,
+    };
+
+    const outcome = await policy.evaluate(policyJob);
+    const evaluatedAt = this.now().toISOString();
+
+    if (outcome.action === "send") {
+      const delivery = await this.deliver({
+        targetSession: outcome.target,
+        message: outcome.message,
+      });
+      const history = this.historyLog.record({
+        jobId: job.jobId,
+        evaluatedAt,
+        outcome: "sent",
+        deliveryTargetSession: outcome.target,
+        deliveryStatus: delivery.status,
+        deliveryMessage: outcome.message,
+        evaluationNotes: outcome.notes ?? null,
+      });
+      this.jobsRepo.recordEvaluation(job.jobId, evaluatedAt, true);
+      this.eventBus.emit({
+        type: "watchdog.evaluation_fired",
+        jobId: job.jobId,
+        policy: job.policy,
+        targetSession: outcome.target,
+        deliveryStatus: delivery.status,
+      });
+      return {
+        job: this.jobsRepo.getByIdOrThrow(job.jobId),
+        outcome,
+        history,
+        delivery,
+      };
+    }
+
+    if (outcome.action === "skip") {
+      const history = this.historyLog.record({
+        jobId: job.jobId,
+        evaluatedAt,
+        outcome: "skipped",
+        skipReason: outcome.reason,
+        evaluationNotes: outcome.notes ?? null,
+      });
+      this.jobsRepo.recordEvaluation(job.jobId, evaluatedAt, false);
+      this.eventBus.emit({
+        type: "watchdog.evaluation_skipped",
+        jobId: job.jobId,
+        policy: job.policy,
+        skipReason: outcome.reason,
+      });
+      return {
+        job: this.jobsRepo.getByIdOrThrow(job.jobId),
+        outcome,
+        history,
+        delivery: null,
+      };
+    }
+
+    // terminal
+    this.jobsRepo.markTerminal(job.jobId, outcome.reason);
+    const history = this.historyLog.record({
+      jobId: job.jobId,
+      evaluatedAt,
+      outcome: "terminal",
+      skipReason: outcome.reason,
+      evaluationNotes: outcome.notes ?? null,
+    });
+    this.eventBus.emit({
+      type: "watchdog.evaluation_terminal",
+      jobId: job.jobId,
+      policy: job.policy,
+      terminalReason: outcome.reason,
+    });
+    return {
+      job: this.jobsRepo.getByIdOrThrow(job.jobId),
+      outcome,
+      history,
+      delivery: null,
+    };
+  }
+}
+
+/**
+ * Default spec_yaml parser. Extracts the `context:` block as a
+ * generic Record<string, unknown>. Uses a minimal recursive
+ * indentation-based parser sufficient for the POC spec shape:
+ *   policy: ...
+ *   target: <member>@<rig>
+ *   interval_seconds: 1800
+ *   context:
+ *     target:
+ *       session: ...
+ *     message: "..."
+ *     pools:
+ *       - path: /abs/...
+ *         include_statuses: [ready]
+ *
+ * For robust YAML, a future maintenance pass can swap in `yaml` or
+ * `js-yaml`; the POC schemas are simple enough that this parser
+ * handles them in-tree without an extra dependency.
+ */
+function defaultParseContext(specYaml: string): Record<string, unknown> {
+  const lines = specYaml.split("\n");
+  let i = 0;
+  while (i < lines.length) {
+    const line = lines[i] ?? "";
+    if (/^context\s*:\s*$/.test(line)) {
+      i++;
+      const block: string[] = [];
+      while (i < lines.length) {
+        const sub = lines[i] ?? "";
+        if (sub.length === 0) {
+          block.push(sub);
+          i++;
+          continue;
+        }
+        if (sub.startsWith(" ") || sub.startsWith("\t")) {
+          block.push(sub);
+          i++;
+          continue;
+        }
+        break;
+      }
+      // Strip leading two-space indent to make it a top-level YAML doc.
+      const dedented = block
+        .map((l) => (l.startsWith("  ") ? l.slice(2) : l))
+        .join("\n");
+      return parseSimpleYaml(dedented) as Record<string, unknown>;
+    }
+    i++;
+  }
+  return {};
+}
+
+/**
+ * Minimal indentation-aware YAML parser. Supports nested mappings,
+ * inline `[a, b]` arrays, dash-prefixed sequences of mappings, and
+ * scalar string values (quoted or unquoted). Sufficient for the POC
+ * spec set; not a general YAML implementation.
+ */
+function parseSimpleYaml(src: string): unknown {
+  const lines = src.split("\n").filter((l) => l.trim().length > 0 && !l.trim().startsWith("#"));
+  const idx = { i: 0 };
+  return parseYamlBlock(lines, idx, 0);
+}
+
+function indentOf(line: string): number {
+  let n = 0;
+  while (n < line.length && line[n] === " ") n++;
+  return n;
+}
+
+function parseScalar(raw: string): unknown {
+  const trimmed = raw.trim();
+  if (trimmed === "null" || trimmed === "~") return null;
+  if (trimmed === "true") return true;
+  if (trimmed === "false") return false;
+  if (/^-?\d+$/.test(trimmed)) return Number.parseInt(trimmed, 10);
+  if (/^-?\d+\.\d+$/.test(trimmed)) return Number.parseFloat(trimmed);
+  if (trimmed.startsWith("[") && trimmed.endsWith("]")) {
+    const inner = trimmed.slice(1, -1).trim();
+    if (inner === "") return [];
+    return inner.split(",").map((p) => parseScalar(p));
+  }
+  if (trimmed.startsWith('"') && trimmed.endsWith('"')) return trimmed.slice(1, -1);
+  if (trimmed.startsWith("'") && trimmed.endsWith("'")) return trimmed.slice(1, -1);
+  return trimmed;
+}
+
+function parseYamlBlock(lines: string[], idx: { i: number }, expectedIndent: number): unknown {
+  if (idx.i >= lines.length) return null;
+  const first = lines[idx.i] ?? "";
+  const firstIndent = indentOf(first);
+  if (firstIndent < expectedIndent) return null;
+  const trimmed = first.trim();
+  if (trimmed.startsWith("- ")) {
+    return parseYamlSeq(lines, idx, firstIndent);
+  }
+  return parseYamlMap(lines, idx, firstIndent);
+}
+
+function parseYamlMap(
+  lines: string[],
+  idx: { i: number },
+  blockIndent: number,
+): Record<string, unknown> {
+  const result: Record<string, unknown> = {};
+  while (idx.i < lines.length) {
+    const line = lines[idx.i] ?? "";
+    const ind = indentOf(line);
+    if (ind < blockIndent) break;
+    if (ind > blockIndent) break;
+    const trimmed = line.trim();
+    const colonIdx = trimmed.indexOf(":");
+    if (colonIdx === -1) break;
+    const key = trimmed.slice(0, colonIdx).trim();
+    const valuePart = trimmed.slice(colonIdx + 1).trim();
+    idx.i++;
+    if (valuePart.length > 0) {
+      result[key] = parseScalar(valuePart);
+    } else {
+      // Nested block: probe next line indent.
+      if (idx.i < lines.length) {
+        const next = lines[idx.i] ?? "";
+        const nextIndent = indentOf(next);
+        if (nextIndent > blockIndent) {
+          result[key] = parseYamlBlock(lines, idx, nextIndent);
+          continue;
+        }
+      }
+      result[key] = null;
+    }
+  }
+  return result;
+}
+
+function parseYamlSeq(
+  lines: string[],
+  idx: { i: number },
+  blockIndent: number,
+): unknown[] {
+  const result: unknown[] = [];
+  while (idx.i < lines.length) {
+    const line = lines[idx.i] ?? "";
+    const ind = indentOf(line);
+    if (ind < blockIndent) break;
+    if (ind > blockIndent) break;
+    const trimmed = line.trim();
+    if (!trimmed.startsWith("- ")) break;
+    const rest = trimmed.slice(2).trim();
+    idx.i++;
+    if (rest.includes(":")) {
+      // First key of an inline mapping element. Synthesize a virtual
+      // sub-block by re-injecting the rest as a key + scanning further
+      // lines at deeper indent.
+      const colonIdx = rest.indexOf(":");
+      const k = rest.slice(0, colonIdx).trim();
+      const v = rest.slice(colonIdx + 1).trim();
+      const elem: Record<string, unknown> = {};
+      if (v.length > 0) {
+        elem[k] = parseScalar(v);
+      } else if (idx.i < lines.length && indentOf(lines[idx.i] ?? "") > blockIndent + 2) {
+        elem[k] = parseYamlBlock(lines, idx, indentOf(lines[idx.i] ?? ""));
+      } else {
+        elem[k] = null;
+      }
+      // Continue gathering same-element keys at indent blockIndent + 2.
+      while (idx.i < lines.length) {
+        const sub = lines[idx.i] ?? "";
+        const subIndent = indentOf(sub);
+        if (subIndent !== blockIndent + 2) break;
+        const subTrim = sub.trim();
+        if (subTrim.startsWith("- ")) break;
+        const sCol = subTrim.indexOf(":");
+        if (sCol === -1) break;
+        const sk = subTrim.slice(0, sCol).trim();
+        const sv = subTrim.slice(sCol + 1).trim();
+        idx.i++;
+        if (sv.length > 0) {
+          elem[sk] = parseScalar(sv);
+        } else if (idx.i < lines.length && indentOf(lines[idx.i] ?? "") > blockIndent + 2) {
+          elem[sk] = parseYamlBlock(lines, idx, indentOf(lines[idx.i] ?? ""));
+        } else {
+          elem[sk] = null;
+        }
+      }
+      result.push(elem);
+    } else {
+      result.push(parseScalar(rest));
+    }
+  }
+  return result;
+}

packages/daemon/src/domain/watchdog-scheduler.ts

@@ -0,0 +1,146 @@
+import type { WatchdogJob, WatchdogJobsRepository } from "./watchdog-jobs-repository.js";
+import type { WatchdogPolicyEngine } from "./watchdog-policy-engine.js";
+
+/**
+ * Watchdog scheduler (PL-004 Phase C; daemon-native supervision tree
+ * member). Joins createDaemon's lifecycle: started by startup.ts after
+ * the policy engine is ready, stopped during graceful shutdown.
+ *
+ * Loop semantics:
+ *   - Wakes every `tickIntervalMs` (default 1000) and queries the
+ *     repository for active jobs whose `interval_seconds` has elapsed
+ *     since `last_evaluation_at` (or who have never been evaluated).
+ *   - For each due job, calls policyEngine.evaluate(job). The engine
+ *     records meaningful outcomes; pure not-due polls are filtered
+ *     here and never reach history.
+ *   - SQLite is the canonical schedule state — restart recovery is
+ *     automatic: on startup, `listActive()` returns the same set
+ *     and `last_evaluation_at` determines next-due.
+ *
+ * Concurrency:
+ *   - One in-flight tick at a time. If a tick takes longer than
+ *     tickIntervalMs (long policy.evaluate or slow delivery), the
+ *     next tick is delayed; we don't queue overlapping ticks.
+ *   - Within a tick, jobs are evaluated sequentially. This bounds
+ *     resource use and matches the POC's single-loop behavior.
+ *
+ * Shutdown:
+ *   - stop() sets shuttingDown=true, clears the timer, and awaits the
+ *     in-flight tick (if any). Idempotent.
+ */
+
+export interface WatchdogSchedulerDeps {
+  jobsRepo: WatchdogJobsRepository;
+  policyEngine: WatchdogPolicyEngine;
+  /** Tick wake-up cadence. Default 1000 ms. */
+  tickIntervalMs?: number;
+  /** Override clock for tests. */
+  now?: () => Date;
+  /** Override timer scheduler for tests. */
+  setTimer?: (cb: () => void, ms: number) => NodeJS.Timeout;
+  clearTimer?: (handle: NodeJS.Timeout) => void;
+  /** Notification on tick errors (for telemetry; defaults to console.error). */
+  onTickError?: (err: unknown) => void;
+}
+
+export class WatchdogScheduler {
+  private readonly jobsRepo: WatchdogJobsRepository;
+  private readonly policyEngine: WatchdogPolicyEngine;
+  private readonly tickIntervalMs: number;
+  private readonly now: () => Date;
+  private readonly setTimer: (cb: () => void, ms: number) => NodeJS.Timeout;
+  private readonly clearTimer: (handle: NodeJS.Timeout) => void;
+  private readonly onTickError: (err: unknown) => void;
+
+  private timer: NodeJS.Timeout | null = null;
+  private inflight: Promise<void> | null = null;
+  private shuttingDown = false;
+  private started = false;
+
+  constructor(deps: WatchdogSchedulerDeps) {
+    this.jobsRepo = deps.jobsRepo;
+    this.policyEngine = deps.policyEngine;
+    this.tickIntervalMs = deps.tickIntervalMs ?? 1000;
+    this.now = deps.now ?? (() => new Date());
+    this.setTimer = deps.setTimer ?? ((cb, ms) => setTimeout(cb, ms));
+    this.clearTimer = deps.clearTimer ?? ((handle) => clearTimeout(handle));
+    this.onTickError = deps.onTickError ?? ((err) => console.error("[watchdog] tick error", err));
+  }
+
+  start(): void {
+    if (this.started) return;
+    this.started = true;
+    this.shuttingDown = false;
+    this.scheduleNextTick();
+  }
+
+  async stop(): Promise<void> {
+    if (!this.started) return;
+    this.shuttingDown = true;
+    if (this.timer) {
+      this.clearTimer(this.timer);
+      this.timer = null;
+    }
+    if (this.inflight) {
+      try {
+        await this.inflight;
+      } catch {
+        // Errors already logged via onTickError; suppress during shutdown.
+      }
+    }
+    this.started = false;
+  }
+
+  isRunning(): boolean {
+    return this.started && !this.shuttingDown;
+  }
+
+  /**
+   * Run one tick synchronously (await result). Exposed for tests so
+   * they can drive ticks without timers. Production path calls this
+   * via the timer loop.
+   */
+  async runTickNow(): Promise<void> {
+    if (this.inflight) {
+      await this.inflight;
+      return;
+    }
+    this.inflight = this.tick();
+    try {
+      await this.inflight;
+    } finally {
+      this.inflight = null;
+    }
+  }
+
+  private scheduleNextTick(): void {
+    if (this.shuttingDown) return;
+    this.timer = this.setTimer(() => {
+      void this.runTickNow()
+        .catch((err) => this.onTickError(err))
+        .finally(() => this.scheduleNextTick());
+    }, this.tickIntervalMs);
+  }
+
+  private async tick(): Promise<void> {
+    if (this.shuttingDown) return;
+    const nowMs = this.now().getTime();
+    const active = this.jobsRepo.listActive();
+    for (const job of active) {
+      if (this.shuttingDown) return;
+      if (!isDue(job, nowMs)) continue;
+      try {
+        await this.policyEngine.evaluate(job);
+      } catch (err) {
+        this.onTickError(err);
+      }
+    }
+  }
+}
+
+export function isDue(job: WatchdogJob, nowMs: number): boolean {
+  if (!job.lastEvaluationAt) return true;
+  const last = Date.parse(job.lastEvaluationAt);
+  if (Number.isNaN(last)) return true;
+  return nowMs - last >= job.intervalSeconds * 1000;
+}

packages/daemon/src/index.ts

@@ -6,15 +6,35 @@ export async function startServer(port?: number) {
   const p = port ?? parseInt(readOpenRigEnv("OPENRIG_PORT", "RIGGED_PORT") ?? "7433", 10);
   const dbPath = readOpenRigEnv("OPENRIG_DB", "RIGGED_DB") ?? "openrig.sqlite";
 
-  const { app, contextMonitor } = await createDaemon({ dbPath });
+  const { app, contextMonitor, deps } = await createDaemon({ dbPath });
 
   const h = readOpenRigEnv("OPENRIG_HOST", "RIGGED_HOST") ?? "127.0.0.1";
 
-  return serve({ fetch: app.fetch, port: p, hostname: h }, (info) => {
+  const server = serve({ fetch: app.fetch, port: p, hostname: h }, (info) => {
     console.log(`OpenRig daemon listening on http://localhost:${info.port}`);
     // Start context monitor polling only after successful server bind
     contextMonitor.start();
+    // PL-004 Phase C: start watchdog scheduler. Joins the supervision
+    // tree post-bind so the HTTP surface is ready before the first
+    // tick (matches contextMonitor pattern).
+    deps.watchdogScheduler?.start();
   });
+
+  // PL-004 Phase C: graceful shutdown — stop scheduler before process
+  // exit so any in-flight policy evaluation completes (or is awaited).
+  const shutdown = async (sig: string) => {
+    console.log(`OpenRig daemon received ${sig}; shutting down`);
+    try {
+      await deps.watchdogScheduler?.stop();
+    } catch (err) {
+      console.error("[watchdog] shutdown error", err);
+    }
+    process.exit(0);
+  };
+  process.once("SIGINT", () => void shutdown("SIGINT"));
+  process.once("SIGTERM", () => void shutdown("SIGTERM"));
+
+  return server;
 }
 
 // Only start the server when this file is executed directly (not imported).

packages/daemon/src/routes/watchdog.ts

@@ -0,0 +1,170 @@
+import { Hono } from "hono";
+import { streamSSE } from "hono/streaming";
+import type { EventBus } from "../domain/event-bus.js";
+import type { WatchdogHistoryLog } from "../domain/watchdog-history-log.js";
+import {
+  type WatchdogJobsRepository,
+  WatchdogJobsError,
+} from "../domain/watchdog-jobs-repository.js";
+
+/**
+ * Watchdog HTTP routes (PL-004 Phase C). Backs `rig watchdog` CLI verb.
+ *
+ * Per Phase A R1 SSE route-order lesson: SSE/static routes mounted
+ * BEFORE the bare-param /:job_id catchall so the literal `/sse` and
+ * literal action paths win over the param route.
+ *
+ * Endpoints:
+ *   POST /register          register a new watchdog job
+ *   GET  /list              list all watchdog jobs (active + stopped + terminal)
+ *   GET  /sse               SSE stream of watchdog.* events
+ *   GET  /:job_id           show one job
+ *   GET  /:job_id/status    job + recent history (compact summary)
+ *   POST /:job_id/stop      operator stop
+ */
+export function watchdogRoutes(): Hono {
+  const app = new Hono();
+
+  function getJobsRepo(c: { get: (key: string) => unknown }): WatchdogJobsRepository {
+    return c.get("watchdogJobsRepo" as never) as WatchdogJobsRepository;
+  }
+  function getHistoryLog(c: { get: (key: string) => unknown }): WatchdogHistoryLog {
+    return c.get("watchdogHistoryLog" as never) as WatchdogHistoryLog;
+  }
+  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 WatchdogJobsError) {
+      const status =
+        err.code === "job_not_found" ? 404
+        : err.code === "policy_unknown" ? 400
+        : err.code === "policy_deferred_to_phase_d" ? 400
+        : err.code === "interval_invalid" ? 400
+        : err.code === "target_session_invalid" ? 400
+        : err.code === "job_terminal" ? 409
+        : 500;
+      return c.json(
+        { error: err.code, message: err.message, ...(err.details ?? {}) },
+        status as 200,
+      );
+    }
+    const message = err instanceof Error ? err.message : "internal error";
+    return c.json({ error: "internal_error", message }, 500);
+  }
+
+  app.post("/register", async (c) => {
+    const body = await c.req
+      .json<{
+        policy?: string;
+        specYaml?: string;
+        targetSession?: string;
+        intervalSeconds?: number;
+        activeWakeIntervalSeconds?: number;
+        scanIntervalSeconds?: number;
+        registeredBySession?: string;
+      }>()
+      .catch(() => ({} as never));
+    if (!body.policy) return c.json({ error: "policy is required" }, 400);
+    if (!body.specYaml) return c.json({ error: "specYaml is required" }, 400);
+    if (!body.targetSession) return c.json({ error: "targetSession is required" }, 400);
+    if (typeof body.intervalSeconds !== "number") {
+      return c.json({ error: "intervalSeconds is required" }, 400);
+    }
+    if (!body.registeredBySession) return c.json({ error: "registeredBySession is required" }, 400);
+    try {
+      const job = getJobsRepo(c).register({
+        policy: body.policy,
+        specYaml: body.specYaml,
+        targetSession: body.targetSession,
+        intervalSeconds: body.intervalSeconds,
+        activeWakeIntervalSeconds: body.activeWakeIntervalSeconds,
+        scanIntervalSeconds: body.scanIntervalSeconds,
+        registeredBySession: body.registeredBySession,
+      });
+      getEventBus(c).emit({
+        type: "watchdog.job_registered",
+        jobId: job.jobId,
+        policy: job.policy,
+        targetSession: job.targetSession,
+        registeredBy: job.registeredBySession,
+      });
+      return c.json(job, 201);
+    } catch (err) {
+      return errorResponse(c, err);
+    }
+  });
+
+  // GET /list — list all watchdog jobs.
+  // Literal path before /:job_id (per Phase A R1 SSE route-order lesson).
+  app.get("/list", (c) => {
+    const jobs = getJobsRepo(c).listAll();
+    return c.json(jobs);
+  });
+
+  // SSE for watchdog.* events. MUST precede /:job_id so the literal
+  // path wins. 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 !== "watchdog.evaluation_fired" &&
+          event.type !== "watchdog.evaluation_skipped" &&
+          event.type !== "watchdog.evaluation_terminal" &&
+          event.type !== "watchdog.job_registered" &&
+          event.type !== "watchdog.job_stopped"
+        ) 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);
+
+  app.get("/:job_id/status", (c) => {
+    const jobId = c.req.param("job_id");
+    const job = getJobsRepo(c).getById(jobId);
+    if (!job) return c.json({ error: "job_not_found", jobId }, 404);
+    const recentHistory = getHistoryLog(c).listForJob(jobId, 20);
+    return c.json({ job, recentHistory });
+  });
+
+  app.post("/:job_id/stop", async (c) => {
+    const jobId = c.req.param("job_id");
+    const body = await c.req
+      .json<{ reason?: string }>()
+      .catch(() => ({} as { reason?: string }));
+    try {
+      const job = getJobsRepo(c).stop(jobId, body.reason ?? "operator_stopped");
+      getEventBus(c).emit({
+        type: "watchdog.job_stopped",
+        jobId: job.jobId,
+        reason: body.reason ?? "operator_stopped",
+      });
+      return c.json(job);
+    } catch (err) {
+      return errorResponse(c, err);
+    }
+  });
+
+  app.get("/:job_id", (c) => {
+    const jobId = c.req.param("job_id");
+    const job = getJobsRepo(c).getById(jobId);
+    if (!job) return c.json({ error: "job_not_found", jobId }, 404);
+    return c.json(job);
+  });
+
+  return app;
+}

packages/daemon/src/server.ts

@@ -63,6 +63,7 @@ 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 { watchdogRoutes } from "./routes/watchdog.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";
@@ -70,6 +71,10 @@ 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 type { WatchdogJobsRepository } from "./domain/watchdog-jobs-repository.js";
+import type { WatchdogHistoryLog } from "./domain/watchdog-history-log.js";
+import type { WatchdogPolicyEngine } from "./domain/watchdog-policy-engine.js";
+import type { WatchdogScheduler } from "./domain/watchdog-scheduler.js";
 import { envRoutes } from "./routes/env.js";
 import type { RigLifecycleService } from "./domain/rig-lifecycle-service.js";
 import { seatRoutes } from "./routes/seat.js";
@@ -116,6 +121,10 @@ export interface AppDeps {
   projectClassifier?: ProjectClassifier;
   classifierLeaseManager?: ClassifierLeaseManager;
   viewProjector?: ViewProjector;
+  watchdogJobsRepo?: WatchdogJobsRepository;
+  watchdogHistoryLog?: WatchdogHistoryLog;
+  watchdogPolicyEngine?: WatchdogPolicyEngine;
+  watchdogScheduler?: WatchdogScheduler;
   specReviewService?: SpecReviewService;
   specLibraryService?: SpecLibraryService;
   whoamiService?: WhoamiService;
@@ -264,6 +273,10 @@ export function createApp(deps: AppDeps): Hono {
     c.set("projectClassifier" as never, deps.projectClassifier);
     c.set("classifierLeaseManager" as never, deps.classifierLeaseManager);
     c.set("viewProjector" as never, deps.viewProjector);
+    c.set("watchdogJobsRepo" as never, deps.watchdogJobsRepo);
+    c.set("watchdogHistoryLog" as never, deps.watchdogHistoryLog);
+    c.set("watchdogPolicyEngine" as never, deps.watchdogPolicyEngine);
+    c.set("watchdogScheduler" as never, deps.watchdogScheduler);
     c.set("specReviewService" as never, deps.specReviewService);
     c.set("specLibraryService" as never, deps.specLibraryService);
     c.set("whoamiService" as never, deps.whoamiService);
@@ -314,6 +327,7 @@ export function createApp(deps: AppDeps): Hono {
   app.route("/api/queue", queueRoutes());
   app.route("/api/projects", projectsRoutes());
   app.route("/api/views", viewsRoutes());
+  app.route("/api/watchdog", watchdogRoutes());
   app.route("/api/rigs/:rigId/env", envRoutes());
   app.route("/api/restore-check", restoreCheckRoutes);
 

packages/daemon/src/startup.ts

@@ -67,6 +67,10 @@ import { ProjectClassifier } from "./domain/project-classifier.js";
 import { ClassifierLeaseManager } from "./domain/classifier-lease-manager.js";
 import { ViewProjector } from "./domain/view-projector.js";
 import { wireViewEventBridge } from "./domain/view-event-bridge.js";
+import { WatchdogJobsRepository } from "./domain/watchdog-jobs-repository.js";
+import { WatchdogHistoryLog } from "./domain/watchdog-history-log.js";
+import { WatchdogPolicyEngine } from "./domain/watchdog-policy-engine.js";
+import { WatchdogScheduler } from "./domain/watchdog-scheduler.js";
 import { SpecReviewService } from "./domain/spec-review-service.js";
 import { SpecLibraryService } from "./domain/spec-library-service.js";
 import { WhoamiService } from "./domain/whoami-service.js";
@@ -103,6 +107,8 @@ 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 { watchdogJobsSchema } from "./db/migrations/031_watchdog_jobs.js";
+import { watchdogHistorySchema } from "./db/migrations/032_watchdog_history.js";
 import { OPENRIG_HOME } from "./openrig-compat.js";
 import {
   getCompatibleOpenRigPath,
@@ -128,7 +134,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, projectClassificationsSchema, classifierLeasesSchema, viewsCustomSchema]);
+  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, watchdogJobsSchema, watchdogHistorySchema]);
 
   const rigRepo = new RigRepository(db);
   const sessionRegistry = new SessionRegistry(db);
@@ -163,6 +169,12 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
   // when underlying state mutates.
   wireViewEventBridge(eventBus, viewProjectorInstance);
 
+  // PL-004 Phase C — watchdog supervision tree. Repository + history-log
+  // are constructed early; the policy engine + scheduler are constructed
+  // after SessionTransport is available so the engine can wire delivery.
+  const watchdogJobsRepoInstance = new WatchdogJobsRepository(db);
+  const watchdogHistoryLogInstance = new WatchdogHistoryLog(db);
+
   const tmuxAdapter = new TmuxAdapter(opts?.tmuxExec ?? execCommand);
   // cmuxFactory takes precedence (for tests), then cmuxExec-based CLI transport, then default
   const cmuxFactory = opts?.cmuxFactory
@@ -436,6 +448,8 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
     classifierLeaseManager: classifierLeaseManagerInstance,
     projectClassifier: new ProjectClassifier(db, eventBus, classifierLeaseManagerInstance),
     viewProjector: viewProjectorInstance,
+    watchdogJobsRepo: watchdogJobsRepoInstance,
+    watchdogHistoryLog: watchdogHistoryLogInstance,
     askService: (() => {
       const psProjectionService = new PsProjectionService({ db });
       const execDep = (cmd: string, args: string[]): Promise<{ stdout: string; exitCode: number }> =>
@@ -514,6 +528,34 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
     }
   } catch { /* best-effort — reference docs are not critical to daemon operation */ }
 
+  // PL-004 Phase C — watchdog policy engine + scheduler. Wired here
+  // (after deps construction) so the engine can dispatch deliveries
+  // through the live SessionTransport. Scheduler is started by
+  // index.ts after listen() so the daemon's HTTP surface is ready
+  // before the scheduler's first tick.
+  const sessionTransport = deps.sessionTransport;
+  if (sessionTransport) {
+    const watchdogPolicyEngine = new WatchdogPolicyEngine({
+      jobsRepo: watchdogJobsRepoInstance,
+      historyLog: watchdogHistoryLogInstance,
+      eventBus,
+      deliver: async ({ targetSession, message }) => {
+        try {
+          const result = await sessionTransport.send(targetSession, message);
+          return result.ok ? { status: "ok" } : { status: "failed", error: result.error };
+        } catch (err) {
+          return { status: "failed", error: err instanceof Error ? err.message : String(err) };
+        }
+      },
+    });
+    const watchdogScheduler = new WatchdogScheduler({
+      jobsRepo: watchdogJobsRepoInstance,
+      policyEngine: watchdogPolicyEngine,
+    });
+    deps.watchdogPolicyEngine = watchdogPolicyEngine;
+    deps.watchdogScheduler = watchdogScheduler;
+  }
+
   // Context monitor — constructed before createApp so routes can access pollOnce for refresh.
   // Caller (index.ts) starts polling after listen.
   const { ContextMonitor } = await import("./domain/context-monitor.js");

packages/daemon/test/watchdog-history-log.test.ts

@@ -0,0 +1,121 @@
+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 { watchdogJobsSchema } from "../src/db/migrations/031_watchdog_jobs.js";
+import { watchdogHistorySchema } from "../src/db/migrations/032_watchdog_history.js";
+import { WatchdogJobsRepository } from "../src/domain/watchdog-jobs-repository.js";
+import { WatchdogHistoryLog } from "../src/domain/watchdog-history-log.js";
+
+describe("WatchdogHistoryLog (PL-004 Phase C; append-only audit)", () => {
+  let db: Database.Database;
+  let jobsRepo: WatchdogJobsRepository;
+  let log: WatchdogHistoryLog;
+  let jobId: string;
+
+  beforeEach(() => {
+    db = createDb();
+    migrate(db, [coreSchema, eventsSchema, watchdogJobsSchema, watchdogHistorySchema]);
+    jobsRepo = new WatchdogJobsRepository(db);
+    log = new WatchdogHistoryLog(db);
+    jobId = jobsRepo.register({
+      policy: "periodic-reminder",
+      specYaml: "policy: periodic-reminder",
+      targetSession: "a@rig",
+      intervalSeconds: 60,
+      registeredBySession: "ops@kernel",
+    }).jobId;
+  });
+
+  afterEach(() => db.close());
+
+  it("record(sent) persists delivery target/status/message + ULID id", () => {
+    const e = log.record({
+      jobId,
+      evaluatedAt: "2026-05-03T07:00:00.000Z",
+      outcome: "sent",
+      deliveryTargetSession: "a@rig",
+      deliveryStatus: "ok",
+      deliveryMessage: "hello",
+    });
+    expect(e.historyId).toMatch(/^[0-9A-Z]{26}$/);
+    expect(e.outcome).toBe("sent");
+    expect(e.deliveryTargetSession).toBe("a@rig");
+    expect(e.deliveryStatus).toBe("ok");
+    expect(e.deliveryMessage).toBe("hello");
+  });
+
+  it("record(skipped) persists reason + null delivery fields", () => {
+    const e = log.record({
+      jobId,
+      evaluatedAt: "2026-05-03T07:00:00.000Z",
+      outcome: "skipped",
+      skipReason: "no_actionable_artifacts",
+    });
+    expect(e.outcome).toBe("skipped");
+    expect(e.skipReason).toBe("no_actionable_artifacts");
+    expect(e.deliveryTargetSession).toBeNull();
+  });
+
+  it("record(terminal) persists reason", () => {
+    const e = log.record({
+      jobId,
+      evaluatedAt: "2026-05-03T07:00:00.000Z",
+      outcome: "terminal",
+      skipReason: "policy_done",
+    });
+    expect(e.outcome).toBe("terminal");
+    expect(e.skipReason).toBe("policy_done");
+  });
+
+  it("record JSON-encodes evaluation_notes and decodes on read", () => {
+    log.record({
+      jobId,
+      evaluatedAt: "2026-05-03T07:00:00.000Z",
+      outcome: "sent",
+      deliveryTargetSession: "a@rig",
+      deliveryStatus: "ok",
+      deliveryMessage: "x",
+      evaluationNotes: { artifact_count: 7, label: "foo" },
+    });
+    const list = log.listForJob(jobId);
+    expect(list[0]?.evaluationNotes).toEqual({ artifact_count: 7, label: "foo" });
+  });
+
+  it("listForJob returns DESC by evaluated_at", () => {
+    log.record({ jobId, evaluatedAt: "2026-05-03T07:00:00.000Z", outcome: "sent", deliveryTargetSession: "a@rig", deliveryStatus: "ok", deliveryMessage: "first" });
+    log.record({ jobId, evaluatedAt: "2026-05-03T07:01:00.000Z", outcome: "sent", deliveryTargetSession: "a@rig", deliveryStatus: "ok", deliveryMessage: "second" });
+    const list = log.listForJob(jobId);
+    expect(list[0]?.deliveryMessage).toBe("second");
+    expect(list[1]?.deliveryMessage).toBe("first");
+  });
+
+  it("countForJob returns the number of recorded entries", () => {
+    expect(log.countForJob(jobId)).toBe(0);
+    log.record({ jobId, evaluatedAt: "2026-05-03T07:00:00.000Z", outcome: "skipped", skipReason: "x" });
+    log.record({ jobId, evaluatedAt: "2026-05-03T07:00:00.000Z", outcome: "skipped", skipReason: "x" });
+    expect(log.countForJob(jobId)).toBe(2);
+  });
+
+  it("FK violation: record() against unknown job_id throws SQLite FK error", () => {
+    expect(() =>
+      log.record({
+        jobId: "unknown-job-id",
+        evaluatedAt: "2026-05-03T07:00:00.000Z",
+        outcome: "skipped",
+        skipReason: "x",
+      }),
+    ).toThrow();
+  });
+
+  it("API surface does NOT expose update or delete (append-only contract)", () => {
+    const proto = Object.getPrototypeOf(log) as Record<string, unknown>;
+    const names = Object.getOwnPropertyNames(proto);
+    expect(names).not.toContain("update");
+    expect(names).not.toContain("delete");
+    expect(names).not.toContain("remove");
+    expect(names).not.toContain("modify");
+  });
+});

packages/daemon/test/watchdog-jobs-repository.test.ts

@@ -0,0 +1,150 @@
+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 { watchdogJobsSchema } from "../src/db/migrations/031_watchdog_jobs.js";
+import { watchdogHistorySchema } from "../src/db/migrations/032_watchdog_history.js";
+import {
+  PHASE_C_POLICIES,
+  WatchdogJobsError,
+  WatchdogJobsRepository,
+} from "../src/domain/watchdog-jobs-repository.js";
+
+describe("WatchdogJobsRepository (PL-004 Phase C)", () => {
+  let db: Database.Database;
+  let repo: WatchdogJobsRepository;
+
+  beforeEach(() => {
+    db = createDb();
+    migrate(db, [coreSchema, eventsSchema, watchdogJobsSchema, watchdogHistorySchema]);
+    repo = new WatchdogJobsRepository(db);
+  });
+
+  afterEach(() => db.close());
+
+  function validInput(overrides: Record<string, unknown> = {}) {
+    return {
+      policy: "periodic-reminder",
+      specYaml: "policy: periodic-reminder\ntarget: a@rig\ninterval_seconds: 60\ncontext:\n  target:\n    session: a@rig\n  message: hello\n",
+      targetSession: "a@rig",
+      intervalSeconds: 60,
+      registeredBySession: "ops@kernel",
+      ...overrides,
+    };
+  }
+
+  it("register stores all three v1 policies", () => {
+    for (const p of PHASE_C_POLICIES) {
+      const job = repo.register(validInput({ policy: p }));
+      expect(job.policy).toBe(p);
+      expect(job.state).toBe("active");
+      expect(job.jobId).toMatch(/^[0-9A-Z]{26}$/);
+    }
+  });
+
+  it("register rejects workflow-keepalive with policy_deferred_to_phase_d", () => {
+    try {
+      repo.register(validInput({ policy: "workflow-keepalive" }));
+      throw new Error("should have thrown");
+    } catch (err) {
+      expect(err).toBeInstanceOf(WatchdogJobsError);
+      expect((err as WatchdogJobsError).code).toBe("policy_deferred_to_phase_d");
+    }
+  });
+
+  it("register rejects unknown policy with policy_unknown", () => {
+    try {
+      repo.register(validInput({ policy: "totally-bogus" }));
+      throw new Error("should have thrown");
+    } catch (err) {
+      expect(err).toBeInstanceOf(WatchdogJobsError);
+      expect((err as WatchdogJobsError).code).toBe("policy_unknown");
+    }
+  });
+
+  it("register rejects non-positive interval_seconds with interval_invalid", () => {
+    for (const bad of [0, -1, 1.5]) {
+      try {
+        repo.register(validInput({ intervalSeconds: bad }));
+        throw new Error(`should have thrown for ${bad}`);
+      } catch (err) {
+        expect(err).toBeInstanceOf(WatchdogJobsError);
+        expect((err as WatchdogJobsError).code).toBe("interval_invalid");
+      }
+    }
+  });
+
+  it("register rejects target_session without @ as target_session_invalid", () => {
+    try {
+      repo.register(validInput({ targetSession: "no-at-here" }));
+      throw new Error("should have thrown");
+    } catch (err) {
+      expect(err).toBeInstanceOf(WatchdogJobsError);
+      expect((err as WatchdogJobsError).code).toBe("target_session_invalid");
+    }
+  });
+
+  it("listActive returns only state=active jobs in registration order", () => {
+    const a = repo.register(validInput({ targetSession: "a@rig" }));
+    const b = repo.register(validInput({ targetSession: "b@rig" }));
+    const c = repo.register(validInput({ targetSession: "c@rig" }));
+    repo.stop(b.jobId);
+    const active = repo.listActive();
+    expect(active.map((j) => j.jobId)).toEqual([a.jobId, c.jobId]);
+  });
+
+  it("recordEvaluation(fired=true) updates last_evaluation_at + last_fire_at", () => {
+    const job = repo.register(validInput());
+    repo.recordEvaluation(job.jobId, "2026-05-03T07:00:00.000Z", true);
+    const after = repo.getByIdOrThrow(job.jobId);
+    expect(after.lastEvaluationAt).toBe("2026-05-03T07:00:00.000Z");
+    expect(after.lastFireAt).toBe("2026-05-03T07:00:00.000Z");
+  });
+
+  it("recordEvaluation(fired=false) updates only last_evaluation_at", () => {
+    const job = repo.register(validInput());
+    repo.recordEvaluation(job.jobId, "2026-05-03T07:00:00.000Z", false);
+    const after = repo.getByIdOrThrow(job.jobId);
+    expect(after.lastEvaluationAt).toBe("2026-05-03T07:00:00.000Z");
+    expect(after.lastFireAt).toBeNull();
+  });
+
+  it("markTerminal sets state=terminal + terminal_reason", () => {
+    const job = repo.register(validInput());
+    repo.markTerminal(job.jobId, "policy_returned_terminal");
+    const after = repo.getByIdOrThrow(job.jobId);
+    expect(after.state).toBe("terminal");
+    expect(after.terminalReason).toBe("policy_returned_terminal");
+  });
+
+  it("stop sets state=stopped + records reason", () => {
+    const job = repo.register(validInput());
+    const stopped = repo.stop(job.jobId, "operator stop reason");
+    expect(stopped.state).toBe("stopped");
+    expect(stopped.terminalReason).toBe("operator stop reason");
+  });
+
+  it("stop on already-terminal job throws job_terminal", () => {
+    const job = repo.register(validInput());
+    repo.markTerminal(job.jobId, "done");
+    try {
+      repo.stop(job.jobId);
+      throw new Error("should have thrown");
+    } catch (err) {
+      expect(err).toBeInstanceOf(WatchdogJobsError);
+      expect((err as WatchdogJobsError).code).toBe("job_terminal");
+    }
+  });
+
+  it("getByIdOrThrow throws job_not_found for unknown id", () => {
+    try {
+      repo.getByIdOrThrow("does-not-exist");
+      throw new Error("should have thrown");
+    } catch (err) {
+      expect(err).toBeInstanceOf(WatchdogJobsError);
+      expect((err as WatchdogJobsError).code).toBe("job_not_found");
+    }
+  });
+});

packages/daemon/test/watchdog-policies.test.ts

@@ -0,0 +1,240 @@
+import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import { mkdirSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { artifactPoolReadyPolicy } from "../src/domain/policies/artifact-pool-ready.js";
+import { edgeArtifactRequiredPolicy } from "../src/domain/policies/edge-artifact-required.js";
+import { periodicReminderPolicy } from "../src/domain/policies/periodic-reminder.js";
+import type { PolicyJob } from "../src/domain/policies/types.js";
+
+function makeJob(overrides: Partial<PolicyJob> & { context: Record<string, unknown> }): PolicyJob {
+  return {
+    jobId: "job-1",
+    policy: "periodic-reminder",
+    targetSession: "a@rig",
+    intervalSeconds: 60,
+    activeWakeIntervalSeconds: null,
+    scanIntervalSeconds: null,
+    lastEvaluationAt: null,
+    lastFireAt: null,
+    registeredBySession: "ops@kernel",
+    registeredAt: "2026-05-03T07:00:00.000Z",
+    ...overrides,
+  };
+}
+
+describe("periodicReminderPolicy", () => {
+  it("returns send with explicit target+message from context", async () => {
+    const out = await periodicReminderPolicy.evaluate(
+      makeJob({
+        context: { target: { session: "alice@rig" }, message: "ping" },
+      }),
+    );
+    expect(out).toEqual({ action: "send", target: "alice@rig", message: "ping" });
+  });
+
+  it("throws policy_spec_invalid when context.target.session is missing", async () => {
+    try {
+      await periodicReminderPolicy.evaluate(makeJob({ context: { message: "x" } }));
+      throw new Error("should have thrown");
+    } catch (err) {
+      expect((err as Error & { code: string }).code).toBe("policy_spec_invalid");
+    }
+  });
+
+  it("throws policy_spec_invalid when context.message is missing", async () => {
+    try {
+      await periodicReminderPolicy.evaluate(makeJob({ context: { target: { session: "a@rig" } } }));
+      throw new Error("should have thrown");
+    } catch (err) {
+      expect((err as Error & { code: string }).code).toBe("policy_spec_invalid");
+    }
+  });
+});
+
+describe("artifactPoolReadyPolicy", () => {
+  let tmp: string;
+  beforeEach(() => {
+    tmp = join(tmpdir(), `watchdog-pool-${Date.now()}-${Math.random()}`);
+    mkdirSync(tmp, { recursive: true });
+  });
+  afterEach(() => rmSync(tmp, { recursive: true, force: true }));
+
+  it("skip with reason no_actionable_artifacts when pool empty", async () => {
+    const out = await artifactPoolReadyPolicy.evaluate(
+      makeJob({
+        context: { target: { session: "a@rig" }, pools: { path: tmp } },
+      }),
+    );
+    expect(out).toEqual({ action: "skip", reason: "no_actionable_artifacts" });
+  });
+
+  it("send with formatted message when pool has actionable artifacts", async () => {
+    writeFileSync(join(tmp, "a.md"), "---\nstatus: ready\n---\nbody-a\n");
+    writeFileSync(join(tmp, "b.md"), "---\nstatus: ready\n---\nbody-b\n");
+    const out = await artifactPoolReadyPolicy.evaluate(
+      makeJob({
+        context: {
+          target: { session: "a@rig" },
+          pools: { path: tmp, include_statuses: ["ready"] },
+          label: "things",
+        },
+      }),
+    );
+    expect(out.action).toBe("send");
+    if (out.action !== "send") return;
+    expect(out.target).toBe("a@rig");
+    expect(out.message).toMatch(/things has 2 actionable artifact/);
+    expect(out.message).toContain("a.md");
+    expect(out.message).toContain("b.md");
+  });
+
+  it("respects include_statuses filter (artifact with non-matching status is excluded)", async () => {
+    writeFileSync(join(tmp, "a.md"), "---\nstatus: ready\n---\nbody-a\n");
+    writeFileSync(join(tmp, "b.md"), "---\nstatus: draft\n---\nbody-b\n");
+    const out = await artifactPoolReadyPolicy.evaluate(
+      makeJob({
+        context: {
+          target: { session: "a@rig" },
+          pools: { path: tmp, include_statuses: ["ready"] },
+        },
+      }),
+    );
+    expect(out.action).toBe("send");
+    if (out.action !== "send") return;
+    expect(out.message).toMatch(/has 1 actionable/);
+    expect(out.message).toContain("a.md");
+    expect(out.message).not.toContain("b.md");
+  });
+
+  it("respects max_items cap in formatted bullet list", async () => {
+    for (const c of ["a", "b", "c", "d", "e", "f", "g"]) {
+      writeFileSync(join(tmp, `${c}.md`), "---\nstatus: ready\n---\n");
+    }
+    const out = await artifactPoolReadyPolicy.evaluate(
+      makeJob({
+        context: {
+          target: { session: "a@rig" },
+          pools: { path: tmp, include_statuses: ["ready"] },
+          max_items: 3,
+        },
+      }),
+    );
+    expect(out.action).toBe("send");
+    if (out.action !== "send") return;
+    expect(out.message).toMatch(/has 7 actionable/);
+    const bulletCount = out.message.split("\n").filter((l) => l.startsWith("- ")).length;
+    expect(bulletCount).toBe(3);
+  });
+
+  it("missing pool directory yields skip (no actionable)", async () => {
+    rmSync(tmp, { recursive: true, force: true });
+    const out = await artifactPoolReadyPolicy.evaluate(
+      makeJob({
+        context: { target: { session: "a@rig" }, pools: { path: tmp } },
+      }),
+    );
+    expect(out).toEqual({ action: "skip", reason: "no_actionable_artifacts" });
+  });
+});
+
+describe("edgeArtifactRequiredPolicy", () => {
+  let src: string;
+  let tgt: string;
+  beforeEach(() => {
+    src = join(tmpdir(), `wd-edge-src-${Date.now()}-${Math.random()}`);
+    tgt = join(tmpdir(), `wd-edge-tgt-${Date.now()}-${Math.random()}`);
+    mkdirSync(src, { recursive: true });
+    mkdirSync(tgt, { recursive: true });
+  });
+  afterEach(() => {
+    rmSync(src, { recursive: true, force: true });
+    rmSync(tgt, { recursive: true, force: true });
+  });
+
+  it("skip when every source key has matching target", async () => {
+    writeFileSync(join(src, "x.md"), "---\nentry: x\nstatus: ready\n---\n");
+    writeFileSync(join(tgt, "x-downstream.md"), "---\nentry: x\nstatus: anything\n---\n");
+    const out = await edgeArtifactRequiredPolicy.evaluate(
+      makeJob({
+        context: {
+          target: { session: "a@rig" },
+          source_pools: { path: src, include_statuses: ["ready"] },
+          target_pools: { path: tgt },
+        },
+      }),
+    );
+    expect(out).toEqual({ action: "skip", reason: "no_missing_edge_artifacts" });
+  });
+
+  it("send with missing-edge list when sources lack matching targets", async () => {
+    writeFileSync(join(src, "x.md"), "---\nentry: x\nstatus: ready\n---\n");
+    writeFileSync(join(src, "y.md"), "---\nentry: y\nstatus: ready\n---\n");
+    writeFileSync(join(tgt, "x-downstream.md"), "---\nentry: x\n---\n");
+    const out = await edgeArtifactRequiredPolicy.evaluate(
+      makeJob({
+        context: {
+          target: { session: "a@rig" },
+          source_pools: { path: src, include_statuses: ["ready"] },
+          target_pools: { path: tgt },
+          label: "myedge",
+        },
+      }),
+    );
+    expect(out.action).toBe("send");
+    if (out.action !== "send") return;
+    expect(out.target).toBe("a@rig");
+    expect(out.message).toMatch(/myedge has 1 upstream artifact/);
+    expect(out.message).toContain("y.md");
+    expect(out.message).not.toContain("x.md");
+  });
+
+  it("source key falls back to basename-sans-md when frontmatter[key_field] absent", async () => {
+    writeFileSync(join(src, "no-frontmatter-key.md"), "---\nstatus: ready\n---\n");
+    const out = await edgeArtifactRequiredPolicy.evaluate(
+      makeJob({
+        context: {
+          target: { session: "a@rig" },
+          source_pools: { path: src, include_statuses: ["ready"] },
+          target_pools: { path: tgt },
+        },
+      }),
+    );
+    expect(out.action).toBe("send");
+    if (out.action !== "send") return;
+    expect(out.message).toContain("no-frontmatter-key.md");
+  });
+
+  it("target match honors target_key_field override", async () => {
+    writeFileSync(join(src, "x.md"), "---\nentry: foo\nstatus: ready\n---\n");
+    writeFileSync(join(tgt, "x.md"), "---\nlinks_to: foo\n---\n");
+    const out = await edgeArtifactRequiredPolicy.evaluate(
+      makeJob({
+        context: {
+          target: { session: "a@rig" },
+          source_pools: { path: src, include_statuses: ["ready"] },
+          target_pools: { path: tgt },
+          source_key_field: "entry",
+          target_key_field: "links_to",
+        },
+      }),
+    );
+    expect(out).toEqual({ action: "skip", reason: "no_missing_edge_artifacts" });
+  });
+
+  it("target scan ignores include_statuses (matches POC override)", async () => {
+    writeFileSync(join(src, "x.md"), "---\nentry: x\nstatus: ready\n---\n");
+    // Target has matching key but a non-listed status; match should still count.
+    writeFileSync(join(tgt, "x-down.md"), "---\nentry: x\nstatus: rejected\n---\n");
+    const out = await edgeArtifactRequiredPolicy.evaluate(
+      makeJob({
+        context: {
+          target: { session: "a@rig" },
+          source_pools: { path: src, include_statuses: ["ready"] },
+          target_pools: { path: tgt, include_statuses: ["accepted"] },
+        },
+      }),
+    );
+    expect(out).toEqual({ action: "skip", reason: "no_missing_edge_artifacts" });
+  });
+});

packages/daemon/test/watchdog-policy-engine.test.ts

@@ -0,0 +1,160 @@
+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 { watchdogJobsSchema } from "../src/db/migrations/031_watchdog_jobs.js";
+import { watchdogHistorySchema } from "../src/db/migrations/032_watchdog_history.js";
+import { EventBus } from "../src/domain/event-bus.js";
+import { WatchdogJobsRepository } from "../src/domain/watchdog-jobs-repository.js";
+import { WatchdogHistoryLog } from "../src/domain/watchdog-history-log.js";
+import {
+  type DeliveryFn,
+  WatchdogPolicyEngine,
+} from "../src/domain/watchdog-policy-engine.js";
+import type { PersistedEvent } from "../src/domain/types.js";
+
+describe("WatchdogPolicyEngine (PL-004 Phase C)", () => {
+  let db: Database.Database;
+  let bus: EventBus;
+  let jobsRepo: WatchdogJobsRepository;
+  let log: WatchdogHistoryLog;
+  let captured: PersistedEvent[];
+  let deliveryCalls: Array<{ targetSession: string; message: string }>;
+  let deliver: DeliveryFn;
+
+  function makeEngine(deliverOverride?: DeliveryFn): WatchdogPolicyEngine {
+    return new WatchdogPolicyEngine({
+      jobsRepo,
+      historyLog: log,
+      eventBus: bus,
+      deliver: deliverOverride ?? deliver,
+    });
+  }
+
+  beforeEach(() => {
+    db = createDb();
+    migrate(db, [coreSchema, eventsSchema, watchdogJobsSchema, watchdogHistorySchema]);
+    bus = new EventBus(db);
+    jobsRepo = new WatchdogJobsRepository(db);
+    log = new WatchdogHistoryLog(db);
+    captured = [];
+    bus.subscribe((e) => captured.push(e));
+    deliveryCalls = [];
+    deliver = async (req) => {
+      deliveryCalls.push(req);
+      return { status: "ok" };
+    };
+  });
+
+  afterEach(() => db.close());
+
+  it("evaluate(periodic-reminder) routes through delivery + records sent + emits evaluation_fired", async () => {
+    const engine = makeEngine();
+    const job = jobsRepo.register({
+      policy: "periodic-reminder",
+      specYaml: "policy: periodic-reminder\ntarget: alice@rig\ninterval_seconds: 60\ncontext:\n  target:\n    session: alice@rig\n  message: ping\n",
+      targetSession: "alice@rig",
+      intervalSeconds: 60,
+      registeredBySession: "ops@kernel",
+    });
+    const result = await engine.evaluate(job);
+    expect(result.outcome.action).toBe("send");
+    expect(deliveryCalls).toEqual([{ targetSession: "alice@rig", message: "ping" }]);
+    expect(log.countForJob(job.jobId)).toBe(1);
+    const list = log.listForJob(job.jobId);
+    expect(list[0]?.outcome).toBe("sent");
+    expect(list[0]?.deliveryStatus).toBe("ok");
+    expect(captured.some((e) => e.type === "watchdog.evaluation_fired")).toBe(true);
+    const after = jobsRepo.getByIdOrThrow(job.jobId);
+    expect(after.lastEvaluationAt).not.toBeNull();
+    expect(after.lastFireAt).not.toBeNull();
+  });
+
+  it("evaluate records skipped + emits evaluation_skipped + does NOT touch last_fire_at", async () => {
+    const engine = makeEngine(async () => {
+      throw new Error("delivery should NOT be called for skip");
+    });
+    // artifact-pool-ready against an empty pool returns skip(no_actionable_artifacts).
+    const job = jobsRepo.register({
+      policy: "artifact-pool-ready",
+      specYaml:
+        "policy: artifact-pool-ready\ntarget: alice@rig\ninterval_seconds: 60\ncontext:\n  target:\n    session: alice@rig\n  pools:\n    path: /nonexistent-pool-dir-for-test\n",
+      targetSession: "alice@rig",
+      intervalSeconds: 60,
+      registeredBySession: "ops@kernel",
+    });
+    const result = await engine.evaluate(job);
+    expect(result.outcome.action).toBe("skip");
+    if (result.outcome.action !== "skip") return;
+    expect(result.outcome.reason).toBe("no_actionable_artifacts");
+    expect(deliveryCalls).toEqual([]);
+    expect(log.countForJob(job.jobId)).toBe(1);
+    expect(log.listForJob(job.jobId)[0]?.outcome).toBe("skipped");
+    expect(captured.some((e) => e.type === "watchdog.evaluation_skipped")).toBe(true);
+    const after = jobsRepo.getByIdOrThrow(job.jobId);
+    expect(after.lastEvaluationAt).not.toBeNull();
+    expect(after.lastFireAt).toBeNull();
+  });
+
+  it("delivery failure is recorded as sent outcome with delivery_status=failed", async () => {
+    const failingDeliver: DeliveryFn = async () => ({ status: "failed", error: "transport denied" });
+    const engine = makeEngine(failingDeliver);
+    const job = jobsRepo.register({
+      policy: "periodic-reminder",
+      specYaml:
+        "policy: periodic-reminder\ntarget: alice@rig\ninterval_seconds: 60\ncontext:\n  target:\n    session: alice@rig\n  message: ping\n",
+      targetSession: "alice@rig",
+      intervalSeconds: 60,
+      registeredBySession: "ops@kernel",
+    });
+    await engine.evaluate(job);
+    const list = log.listForJob(job.jobId);
+    expect(list[0]?.outcome).toBe("sent");
+    expect(list[0]?.deliveryStatus).toBe("failed");
+  });
+
+  it("unknown policy at evaluate-time marks job terminal + emits evaluation_terminal", async () => {
+    const engine = makeEngine();
+    const job = jobsRepo.register({
+      policy: "periodic-reminder",
+      specYaml: "policy: periodic-reminder\ncontext:\n  target:\n    session: a@rig\n  message: x\n",
+      targetSession: "a@rig",
+      intervalSeconds: 60,
+      registeredBySession: "ops@kernel",
+    });
+    // Hand-craft a job carrying an unknown policy name. The engine should
+    // treat this defensively: terminal + history record + event.
+    const synthetic = { ...job, policy: "totally-bogus" as never };
+    const result = await engine.evaluate(synthetic);
+    expect(result.outcome.action).toBe("terminal");
+    const after = jobsRepo.getByIdOrThrow(job.jobId);
+    expect(after.state).toBe("terminal");
+    expect(captured.some((e) => e.type === "watchdog.evaluation_terminal")).toBe(true);
+  });
+
+  it("policy registry resolves the three v1 policies", () => {
+    const engine = makeEngine();
+    expect(engine.resolvePolicy("periodic-reminder")?.name).toBe("periodic-reminder");
+    expect(engine.resolvePolicy("artifact-pool-ready")?.name).toBe("artifact-pool-ready");
+    expect(engine.resolvePolicy("edge-artifact-required")?.name).toBe("edge-artifact-required");
+    expect(engine.resolvePolicy("workflow-keepalive")).toBeUndefined();
+  });
+
+  it("default YAML parser extracts nested context block", async () => {
+    const engine = makeEngine();
+    const job = jobsRepo.register({
+      policy: "periodic-reminder",
+      specYaml:
+        "policy: periodic-reminder\ntarget: a@rig\ninterval_seconds: 60\ncontext:\n  target:\n    session: parsed-session@rig\n  message: parsed-message\n",
+      targetSession: "a@rig",
+      intervalSeconds: 60,
+      registeredBySession: "ops@kernel",
+    });
+    await engine.evaluate(job);
+    expect(deliveryCalls).toEqual([
+      { targetSession: "parsed-session@rig", message: "parsed-message" },
+    ]);
+  });
+});

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

@@ -0,0 +1,204 @@
+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 { watchdogJobsSchema } from "../src/db/migrations/031_watchdog_jobs.js";
+import { watchdogHistorySchema } from "../src/db/migrations/032_watchdog_history.js";
+import { EventBus } from "../src/domain/event-bus.js";
+import { WatchdogJobsRepository } from "../src/domain/watchdog-jobs-repository.js";
+import { WatchdogHistoryLog } from "../src/domain/watchdog-history-log.js";
+import { watchdogRoutes } from "../src/routes/watchdog.js";
+
+function buildApp(opts: {
+  eventBus: EventBus;
+  jobsRepo: WatchdogJobsRepository;
+  historyLog: WatchdogHistoryLog;
+}): Hono {
+  const app = new Hono();
+  app.use("*", async (c, next) => {
+    c.set("eventBus" as never, opts.eventBus);
+    c.set("watchdogJobsRepo" as never, opts.jobsRepo);
+    c.set("watchdogHistoryLog" as never, opts.historyLog);
+    await next();
+  });
+  app.route("/api/watchdog", watchdogRoutes());
+  return app;
+}
+
+describe("watchdog routes (PL-004 Phase C)", () => {
+  let db: Database.Database;
+  let bus: EventBus;
+  let jobsRepo: WatchdogJobsRepository;
+  let log: WatchdogHistoryLog;
+  let app: Hono;
+
+  const validRegisterBody = {
+    policy: "periodic-reminder",
+    specYaml:
+      "policy: periodic-reminder\ntarget: alice@rig\ninterval_seconds: 60\ncontext:\n  target:\n    session: alice@rig\n  message: ping\n",
+    targetSession: "alice@rig",
+    intervalSeconds: 60,
+    registeredBySession: "ops@kernel",
+  };
+
+  beforeEach(() => {
+    db = createDb();
+    migrate(db, [coreSchema, eventsSchema, watchdogJobsSchema, watchdogHistorySchema]);
+    bus = new EventBus(db);
+    jobsRepo = new WatchdogJobsRepository(db);
+    log = new WatchdogHistoryLog(db);
+    app = buildApp({ eventBus: bus, jobsRepo, historyLog: log });
+  });
+
+  afterEach(() => db.close());
+
+  it("POST /register returns 201 + persists job + emits watchdog.job_registered", async () => {
+    const captured: Array<{ type: string }> = [];
+    bus.subscribe((e) => captured.push(e));
+    const res = await app.request("/api/watchdog/register", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify(validRegisterBody),
+    });
+    expect(res.status).toBe(201);
+    const body = (await res.json()) as { jobId: string; policy: string; state: string };
+    expect(body.jobId).toMatch(/^[0-9A-Z]{26}$/);
+    expect(body.policy).toBe("periodic-reminder");
+    expect(body.state).toBe("active");
+    expect(captured.some((e) => e.type === "watchdog.job_registered")).toBe(true);
+  });
+
+  it("POST /register rejects workflow-keepalive with 400 + policy_deferred_to_phase_d", async () => {
+    const res = await app.request("/api/watchdog/register", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ ...validRegisterBody, policy: "workflow-keepalive" }),
+    });
+    expect(res.status).toBe(400);
+    const body = (await res.json()) as { error: string };
+    expect(body.error).toBe("policy_deferred_to_phase_d");
+  });
+
+  it("POST /register rejects unknown policy with 400 + policy_unknown", async () => {
+    const res = await app.request("/api/watchdog/register", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ ...validRegisterBody, policy: "totally-bogus" }),
+    });
+    expect(res.status).toBe(400);
+    const body = (await res.json()) as { error: string };
+    expect(body.error).toBe("policy_unknown");
+  });
+
+  it("POST /register rejects missing required field with 400", async () => {
+    const res = await app.request("/api/watchdog/register", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ policy: "periodic-reminder" }),
+    });
+    expect(res.status).toBe(400);
+  });
+
+  it("GET /list lists all jobs", async () => {
+    jobsRepo.register({ ...validRegisterBody, targetSession: "a@rig" });
+    jobsRepo.register({ ...validRegisterBody, targetSession: "b@rig" });
+    const res = await app.request("/api/watchdog/list");
+    expect(res.status).toBe(200);
+    const list = (await res.json()) as Array<{ targetSession: string }>;
+    expect(list).toHaveLength(2);
+  });
+
+  it("GET /:job_id returns job by id", async () => {
+    const job = jobsRepo.register(validRegisterBody);
+    const res = await app.request(`/api/watchdog/${job.jobId}`);
+    expect(res.status).toBe(200);
+    const body = (await res.json()) as { jobId: string };
+    expect(body.jobId).toBe(job.jobId);
+  });
+
+  it("GET /:job_id returns 404 for unknown id", async () => {
+    const res = await app.request("/api/watchdog/unknown-id");
+    expect(res.status).toBe(404);
+    const body = (await res.json()) as { error: string };
+    expect(body.error).toBe("job_not_found");
+  });
+
+  it("GET /:job_id/status returns job + recent history", async () => {
+    const job = jobsRepo.register(validRegisterBody);
+    log.record({
+      jobId: job.jobId,
+      evaluatedAt: "2026-05-03T07:00:00.000Z",
+      outcome: "sent",
+      deliveryTargetSession: "alice@rig",
+      deliveryStatus: "ok",
+      deliveryMessage: "ping",
+    });
+    const res = await app.request(`/api/watchdog/${job.jobId}/status`);
+    expect(res.status).toBe(200);
+    const body = (await res.json()) as { job: { jobId: string }; recentHistory: Array<{ outcome: string }> };
+    expect(body.job.jobId).toBe(job.jobId);
+    expect(body.recentHistory).toHaveLength(1);
+    expect(body.recentHistory[0]?.outcome).toBe("sent");
+  });
+
+  it("POST /:job_id/stop stops job + emits watchdog.job_stopped", async () => {
+    const captured: Array<{ type: string }> = [];
+    bus.subscribe((e) => captured.push(e));
+    const job = jobsRepo.register(validRegisterBody);
+    const res = await app.request(`/api/watchdog/${job.jobId}/stop`, {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ reason: "tester stopped" }),
+    });
+    expect(res.status).toBe(200);
+    const body = (await res.json()) as { state: string };
+    expect(body.state).toBe("stopped");
+    expect(captured.some((e) => e.type === "watchdog.job_stopped")).toBe(true);
+  });
+
+  it("POST /:job_id/stop on terminal job returns 409 + job_terminal", async () => {
+    const job = jobsRepo.register(validRegisterBody);
+    jobsRepo.markTerminal(job.jobId, "done");
+    const res = await app.request(`/api/watchdog/${job.jobId}/stop`, {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({}),
+    });
+    expect(res.status).toBe(409);
+    const body = (await res.json()) as { error: string };
+    expect(body.error).toBe("job_terminal");
+  });
+
+  it("R1 SSE pattern: GET /api/watchdog/sse returns 200 + content-type text/event-stream", async () => {
+    const res = await app.request("/api/watchdog/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/watchdog/watch returns 200 + content-type text/event-stream", async () => {
+    const res = await app.request("/api/watchdog/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/watchdog/sse does NOT return job_not_found (route-order regression guard)", async () => {
+    const res = await app.request("/api/watchdog/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/watchdog-scheduler.test.ts

@@ -0,0 +1,221 @@
+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 { watchdogJobsSchema } from "../src/db/migrations/031_watchdog_jobs.js";
+import { watchdogHistorySchema } from "../src/db/migrations/032_watchdog_history.js";
+import { EventBus } from "../src/domain/event-bus.js";
+import { WatchdogJobsRepository } from "../src/domain/watchdog-jobs-repository.js";
+import { WatchdogHistoryLog } from "../src/domain/watchdog-history-log.js";
+import { WatchdogPolicyEngine } from "../src/domain/watchdog-policy-engine.js";
+import { WatchdogScheduler, isDue } from "../src/domain/watchdog-scheduler.js";
+
+describe("WatchdogScheduler (PL-004 Phase C)", () => {
+  let db: Database.Database;
+  let bus: EventBus;
+  let jobsRepo: WatchdogJobsRepository;
+  let log: WatchdogHistoryLog;
+  let deliveries: Array<{ targetSession: string; message: string }>;
+
+  beforeEach(() => {
+    db = createDb();
+    migrate(db, [coreSchema, eventsSchema, watchdogJobsSchema, watchdogHistorySchema]);
+    bus = new EventBus(db);
+    jobsRepo = new WatchdogJobsRepository(db);
+    log = new WatchdogHistoryLog(db);
+    deliveries = [];
+  });
+
+  afterEach(() => db.close());
+
+  function makeEngine() {
+    return new WatchdogPolicyEngine({
+      jobsRepo,
+      historyLog: log,
+      eventBus: bus,
+      deliver: async (req) => {
+        deliveries.push(req);
+        return { status: "ok" };
+      },
+    });
+  }
+
+  it("isDue returns true for never-evaluated jobs", () => {
+    const job = jobsRepo.register({
+      policy: "periodic-reminder",
+      specYaml: "context:\n  target:\n    session: a@rig\n  message: x\n",
+      targetSession: "a@rig",
+      intervalSeconds: 30,
+      registeredBySession: "ops@kernel",
+    });
+    expect(isDue(job, Date.now())).toBe(true);
+  });
+
+  it("isDue returns false when interval not yet elapsed", () => {
+    const job = jobsRepo.register({
+      policy: "periodic-reminder",
+      specYaml: "context:\n  target:\n    session: a@rig\n  message: x\n",
+      targetSession: "a@rig",
+      intervalSeconds: 30,
+      registeredBySession: "ops@kernel",
+    });
+    const tenSecondsAgo = new Date(Date.now() - 10_000).toISOString();
+    jobsRepo.recordEvaluation(job.jobId, tenSecondsAgo, true);
+    const updated = jobsRepo.getByIdOrThrow(job.jobId);
+    expect(isDue(updated, Date.now())).toBe(false);
+  });
+
+  it("isDue returns true when interval elapsed", () => {
+    const job = jobsRepo.register({
+      policy: "periodic-reminder",
+      specYaml: "context:\n  target:\n    session: a@rig\n  message: x\n",
+      targetSession: "a@rig",
+      intervalSeconds: 30,
+      registeredBySession: "ops@kernel",
+    });
+    const longAgo = new Date(Date.now() - 60_000).toISOString();
+    jobsRepo.recordEvaluation(job.jobId, longAgo, true);
+    const updated = jobsRepo.getByIdOrThrow(job.jobId);
+    expect(isDue(updated, Date.now())).toBe(true);
+  });
+
+  it("runTickNow only evaluates due active jobs", async () => {
+    const j1 = jobsRepo.register({
+      policy: "periodic-reminder",
+      specYaml: "context:\n  target:\n    session: a@rig\n  message: m1\n",
+      targetSession: "a@rig",
+      intervalSeconds: 30,
+      registeredBySession: "ops@kernel",
+    });
+    const j2 = jobsRepo.register({
+      policy: "periodic-reminder",
+      specYaml: "context:\n  target:\n    session: b@rig\n  message: m2\n",
+      targetSession: "b@rig",
+      intervalSeconds: 30,
+      registeredBySession: "ops@kernel",
+    });
+    jobsRepo.recordEvaluation(j1.jobId, new Date(Date.now() - 5_000).toISOString(), true);
+
+    const engine = makeEngine();
+    const sched = new WatchdogScheduler({ jobsRepo, policyEngine: engine });
+    await sched.runTickNow();
+    expect(deliveries).toEqual([{ targetSession: "b@rig", message: "m2" }]);
+  });
+
+  it("runTickNow evaluates a due job at every subsequent due tick", async () => {
+    const job = jobsRepo.register({
+      policy: "periodic-reminder",
+      specYaml: "context:\n  target:\n    session: a@rig\n  message: m\n",
+      targetSession: "a@rig",
+      intervalSeconds: 1,
+      registeredBySession: "ops@kernel",
+    });
+    const engine = makeEngine();
+    const sched = new WatchdogScheduler({ jobsRepo, policyEngine: engine });
+    await sched.runTickNow();
+    expect(deliveries.length).toBe(1);
+    // Re-mark as if evaluation was 5s ago to make the next tick due again.
+    jobsRepo.recordEvaluation(job.jobId, new Date(Date.now() - 5_000).toISOString(), true);
+    await sched.runTickNow();
+    expect(deliveries.length).toBe(2);
+  });
+
+  it("stopped jobs are excluded from evaluation", async () => {
+    const job = jobsRepo.register({
+      policy: "periodic-reminder",
+      specYaml: "context:\n  target:\n    session: a@rig\n  message: m\n",
+      targetSession: "a@rig",
+      intervalSeconds: 30,
+      registeredBySession: "ops@kernel",
+    });
+    jobsRepo.stop(job.jobId);
+    const engine = makeEngine();
+    const sched = new WatchdogScheduler({ jobsRepo, policyEngine: engine });
+    await sched.runTickNow();
+    expect(deliveries).toEqual([]);
+  });
+
+  it("terminal jobs are excluded from evaluation", async () => {
+    const job = jobsRepo.register({
+      policy: "periodic-reminder",
+      specYaml: "context:\n  target:\n    session: a@rig\n  message: m\n",
+      targetSession: "a@rig",
+      intervalSeconds: 30,
+      registeredBySession: "ops@kernel",
+    });
+    jobsRepo.markTerminal(job.jobId, "done");
+    const engine = makeEngine();
+    const sched = new WatchdogScheduler({ jobsRepo, policyEngine: engine });
+    await sched.runTickNow();
+    expect(deliveries).toEqual([]);
+  });
+
+  it("policy evaluation errors are caught and tick continues for siblings", async () => {
+    jobsRepo.register({
+      policy: "periodic-reminder",
+      specYaml: "context:\n  message: only-message-no-target\n",
+      targetSession: "a@rig",
+      intervalSeconds: 30,
+      registeredBySession: "ops@kernel",
+    });
+    const ok = jobsRepo.register({
+      policy: "periodic-reminder",
+      specYaml: "context:\n  target:\n    session: ok@rig\n  message: ok\n",
+      targetSession: "ok@rig",
+      intervalSeconds: 30,
+      registeredBySession: "ops@kernel",
+    });
+    const engine = makeEngine();
+    const errors: unknown[] = [];
+    const sched = new WatchdogScheduler({
+      jobsRepo,
+      policyEngine: engine,
+      onTickError: (err) => errors.push(err),
+    });
+    await sched.runTickNow();
+    expect(errors.length).toBe(1);
+    expect(deliveries).toEqual([{ targetSession: "ok@rig", message: "ok" }]);
+    // After successful eval the second job's last_evaluation_at should be set.
+    const after = jobsRepo.getByIdOrThrow(ok.jobId);
+    expect(after.lastEvaluationAt).not.toBeNull();
+  });
+
+  it("start + stop are idempotent and do not throw", async () => {
+    const engine = makeEngine();
+    const sched = new WatchdogScheduler({ jobsRepo, policyEngine: engine, tickIntervalMs: 60_000 });
+    sched.start();
+    sched.start();
+    expect(sched.isRunning()).toBe(true);
+    await sched.stop();
+    await sched.stop();
+    expect(sched.isRunning()).toBe(false);
+  });
+
+  it("recovers schedule across restart via SQLite (new repo + new scheduler picks up active jobs)", async () => {
+    jobsRepo.register({
+      policy: "periodic-reminder",
+      specYaml: "context:\n  target:\n    session: persist@rig\n  message: persisted\n",
+      targetSession: "persist@rig",
+      intervalSeconds: 1,
+      registeredBySession: "ops@kernel",
+    });
+    // Simulate restart: new repo + new history-log + new engine + new scheduler
+    // sharing the same DB handle. Phase A/B pattern; SQLite is canonical.
+    const repo2 = new WatchdogJobsRepository(db);
+    const log2 = new WatchdogHistoryLog(db);
+    const engine2 = new WatchdogPolicyEngine({
+      jobsRepo: repo2,
+      historyLog: log2,
+      eventBus: bus,
+      deliver: async (req) => {
+        deliveries.push(req);
+        return { status: "ok" };
+      },
+    });
+    const sched2 = new WatchdogScheduler({ jobsRepo: repo2, policyEngine: engine2 });
+    await sched2.runTickNow();
+    expect(deliveries).toEqual([{ targetSession: "persist@rig", message: "persisted" }]);
+  });
+});