fix(daemon): PL-004 Phase C R1 — POC contract parity for watchdog policies + active-wake throttle

Repairs guard BLOCK on commit 48c0267:

1. scan_interval_seconds vs active_wake_interval_seconds (blocker 1):
   - Scheduler isDue uses scan_interval_seconds when set, falling back
     to interval_seconds. Mirrors POC engine.mjs:38-46.
   - Policy engine enforces active_wake_interval_seconds as a re-delivery
     throttle: when state.actionable was already true AND wake-window
     not elapsed since last_fire_at, returns quiet skip with reason
     active_wake_not_due (does NOT call delivery, preserves last_fire_at).
     Mirrors POC engine.mjs:49-64 and :243-263.
   - New columns on watchdog_jobs: actionable INTEGER NOT NULL DEFAULT 0,
     last_actionable_at TEXT. Engine sets these on every meaningful
     evaluation; clears on skip; preserves last_actionable_at across
     continued-actionable windows. Migration extended in-place since
     031 has not yet been promoted past Phase B HEAD.
   - Ported POC active-wake regression (tests/watchdog-active-wake-interval.test.sh)
     into daemon test as a 6-tick state-machine regression.

2. edge-artifact-required POC contract (blocker 2):
   - Spec uses context.source / context.target (singular pool specs)
     not source_pools / target_pools. Each may have path or paths.
   - Edge-satisfaction is body-match: a downstream artifact satisfies
     a source iff target.raw INCLUDES the source key. NOT a frontmatter
     field match. Mirrors POC edge-artifact-required.mjs:11-12.
   - Target scan overrides include_statuses to [] (POC override) so any
     downstream state counts as 'exists'.
   - Delivery target is job.target.session (top-level), not from context.
   - Label key is context.edge_label (POC parity).
   - Daemon test ports the POC pilot-a body-reference scenario.

3. Artifact-pool scanner POC parity (blocker 3):
   - Default ignores README.md and .DS_Store always applied.
   - Configured pool.ignore_names extends default ignores.
   - pool.recursive=true descends into subdirectories.
   - Malformed-frontmatter excluded unless pool.include_malformed_frontmatter.
   - Raw content preserved on every artifact (used by edge-artifact-required
     body-match).
   - ENOENT-tolerant pool path.
   - artifact-pool-ready and artifact-pool-helpers tests cover all 5 POC
     scanner behaviors.

Engine state-machine alignment to POC:
   - PolicyJob shape uses {target: {session}, message?, context}. Engine
     parses spec_yaml top-level target + message blocks and falls back to
     {session: registered targetSession} when spec lacks target.
   - Policies all read job.target.session for delivery target and
     job.message OR context.message for periodic-reminder.
   - PolicyEvaluation send action returns target as object {session: ...}
     to match POC contract (engine extracts session string for delivery).
   - QUIET_SKIP_REASONS set (not_due, no_actionable_artifacts,
     no_missing_edge_artifacts, active_wake_not_due) suppresses both
     history record AND event emit. Loud skips still record + emit.
     Mirrors POC shouldAppendHistory (engine.mjs:99-112).

Verification:
   - tsc clean for daemon and cli.
   - Per-package vitest baseline under explicit OPENRIG_HOME +
     OPENRIG_DB=:memory: temp env:
     - packages/daemon: 2485 / 2485 pass (was 2471; +14 new watchdog
       tests covering POC contract + active-wake state machine).
     - packages/cli: 7 / 7 watchdog tests pass.

Boundaries preserved:
   - Worktree-only on phase-c-coordination-primitive-system branch;
     main untouched.
   - No-touch surfaces (Phase A stream/queue/hot-potato, Phase B
     project/classifier/view, Lifecycle v2, migrations 001-030, workflow
     runtime, progress-rail code, POC services/watchdog state) unchanged.
   - workflow-keepalive still rejected at register with
     policy_deferred_to_phase_d; not added as a product policy module.
   - No fleet sweep, no protected-host mutation, no external spend, no
     destructive lifecycle, no VM-boundary actions.

Docs updated in architecture.md and cli-reference.md to reflect the
quiet-skip parity + scan/wake cadence semantics. codemap entries from
48c0267 still describe the same files.

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

docs/as-built/architecture.md

@@ -326,7 +326,7 @@ Migration boundary:
 - `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.
+- `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, scan + active-wake intervals, actionable + last_actionable_at active-wake state, lifecycle); `watchdog_history` is the append-only audit of meaningful evaluations only (loud `sent`/`skipped`/`terminal`). Quiet skip reasons (`not_due`, `no_actionable_artifacts`, `no_missing_edge_artifacts`, `active_wake_not_due`) are NOT recorded and do NOT emit `watchdog.*` events — POC parity so agents are not woken about scheduler polls. Scan cadence (`scan_interval_seconds`) is enforced by the scheduler; re-delivery cadence (`active_wake_interval_seconds`) is enforced by the policy engine and only suppresses repeated wakes during a continuously-actionable window. 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

@@ -712,7 +712,7 @@ Closure-reason semantics:
 - `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.
+History records only loud evaluations: `sent` (delivery executed) or `terminal` (policy declared the job done). Quiet skip reasons (`not_due`, `no_actionable_artifacts`, `no_missing_edge_artifacts`, `active_wake_not_due`) are NOT recorded — POC parity so agents are not woken about scheduler polls. Loud `skipped` rows are recorded only if a policy returns a non-quiet reason.
 
 ## Commands Not Present
 

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

@@ -15,8 +15,20 @@ import type { Migration } from "../migrate.js";
  *   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).
+ * Cadence columns mirror POC engine semantics:
+ *   - interval_seconds is the legacy / fallback cadence.
+ *   - scan_interval_seconds overrides the scheduler isDue cadence
+ *     (how often to run the policy at all).
+ *   - active_wake_interval_seconds throttles re-delivery: when the
+ *     pool stays continuously actionable, fire only every Nth scan.
+ *
+ * Active-wake state (R1 fix; POC parity for active-wake throttle):
+ *   - actionable: 1 iff the most recent scan returned action=send.
+ *     Used to detect newly-actionable transitions (no throttle on
+ *     transition; throttle on continued actionable state).
+ *   - last_actionable_at: first attempt time during the current
+ *     actionable window. Cleared when the policy returns skip.
+ *
  * spec_yaml preserves the original operator-supplied YAML for audit.
  */
 export const watchdogJobsSchema: Migration = {
@@ -32,6 +44,8 @@ export const watchdogJobsSchema: Migration = {
       scan_interval_seconds INTEGER,
       last_evaluation_at TEXT,
       last_fire_at TEXT,
+      actionable INTEGER NOT NULL DEFAULT 0,
+      last_actionable_at TEXT,
       state TEXT NOT NULL DEFAULT 'active',
       registered_by_session TEXT NOT NULL,
       registered_at TEXT NOT NULL,

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

@@ -1,136 +1,259 @@
-// PL-004 Phase C: shared artifact-pool helpers (TypeScript port of POC
-// `lib/policies/artifact-pool.js`).
+// PL-004 Phase C R1: 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.
+// R1 fix (guard blocker 3): full POC parity for the scanner. The
+// artifact-pool-ready and edge-artifact-required policies depend on:
+//   - Default ignores: README.md and .DS_Store always excluded.
+//   - Configured ignore_names: per-pool extra exclusions.
+//   - Recursive scan when pool.recursive=true.
+//   - Malformed-frontmatter exclusion unless pool.include_malformed_frontmatter.
+//   - Raw content preserved on every artifact (used by edge-artifact-required
+//     for body-reference target satisfaction).
+//   - Missing pool path returns empty (ENOENT-tolerant).
+//
+// Pure filesystem scanner; no event-bus, no DB, no Hono.
 
-import { existsSync, readdirSync, readFileSync, statSync } from "node:fs";
-import { join, basename } from "node:path";
+import { readdirSync, readFileSync, statSync } from "node:fs";
+import { basename, join } from "node:path";
+
+const DEFAULT_EXTENSIONS = [".md"];
+const DEFAULT_IGNORE_NAMES = ["README.md", ".DS_Store"];
 
 export interface ArtifactPoolSpec {
-  /** Absolute path to the pool directory (or `paths` array variant). */
+  /** Absolute path to the pool directory. POC contract: one path per pool. */
   path?: string;
+  /** Convenience array form; expanded to one pool per path. */
   paths?: string[];
   /** File extensions to include (default: ['.md']). */
   extensions?: string[];
-  /** Filter on frontmatter `status:` field (default: include all). */
+  /** Filter on frontmatter `status:` field. Empty/absent = include all. */
   include_statuses?: string[];
   /** Frontmatter field used for keying artifacts (default: 'entry'). */
   key_field?: string;
+  /** Per-pool extra ignore names (added to defaults). */
+  ignore_names?: string[];
+  /** When true, descend into subdirectories. Default false. */
+  recursive?: boolean;
+  /**
+   * When true, artifacts whose frontmatter cannot be parsed are still
+   * included with empty frontmatter. Default false: malformed frontmatter
+   * causes the artifact to be excluded (matches POC behavior so that
+   * agents do not get woken about half-written drafts).
+   */
+  include_malformed_frontmatter?: boolean;
 }
 
 export interface ScannedArtifact {
   /** Absolute path to the artifact file. */
   path: string;
+  /** The pool path that produced this artifact. */
+  pool_path: string;
+  /** Full file content (used by edge-artifact-required body-match). */
+  raw: string;
   /** Parsed YAML frontmatter (top-level keys only). */
   frontmatter: Record<string, string>;
+  /** Parse error message when include_malformed_frontmatter=true; else null. */
+  frontmatter_parse_error: string | null;
+  /** Convenience accessor for frontmatter.status; null when absent. */
+  status: string | null;
 }
 
-const FRONTMATTER_RE = /^---\n([\s\S]*?)\n---/;
+const FRONTMATTER_OPEN = "---\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.
+ * Returns { raw, frontmatter, parseError } so callers can decide whether
+ * to include malformed artifacts.
+ *
+ * Detection of malformed frontmatter mirrors POC behavior: lines that
+ * fail the `key: value` shape after the opener are treated as parse
+ * errors (rather than silently dropped) so include_malformed_frontmatter
+ * gating works as POC tests expect.
  */
-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;
+function readFrontmatter(filePath: string): {
+  raw: string;
+  frontmatter: Record<string, string>;
+  frontmatter_parse_error: string | null;
+} {
+  const raw = readFileSync(filePath, "utf-8");
+  if (!raw.startsWith(FRONTMATTER_OPEN)) {
+    return { raw, frontmatter: {}, frontmatter_parse_error: null };
   }
-  return fm;
+  const endIdx = raw.indexOf("\n---\n", FRONTMATTER_OPEN.length);
+  if (endIdx === -1) {
+    return { raw, frontmatter: {}, frontmatter_parse_error: null };
+  }
+  const block = raw.slice(FRONTMATTER_OPEN.length, endIdx);
+  const fm: Record<string, string> = {};
+  let parseError: string | null = null;
+  for (const line of block.split("\n")) {
+    if (line.trim().length === 0) continue;
+    const colonIdx = line.indexOf(":");
+    if (colonIdx === -1) {
+      parseError = `frontmatter line missing ':' (${line.slice(0, 40)})`;
+      break;
+    }
+    const key = line.slice(0, colonIdx).trim();
+    const valueRaw = line.slice(colonIdx + 1);
+    let value = valueRaw.trim();
+    // Detect malformed values like 'broken: value: still broken' where the
+    // value side itself contains another `:` outside quotes. POC treats
+    // these as parse errors so half-written drafts do not slip through.
+    const stripped = value.replace(/^"(.*)"$/, "$1").replace(/^'(.*)'$/, "$1");
+    if (
+      stripped !== value &&
+      stripped.length === value.length - 2
+    ) {
+      // Was quoted; treat as a clean string regardless of inner colons.
+      value = stripped;
+    } else if (value.includes(":")) {
+      parseError = `frontmatter value contains unescaped ':' for key '${key}'`;
+      break;
+    }
+    if (!key) {
+      parseError = "frontmatter line missing key";
+      break;
+    }
+    fm[key] = value;
+  }
+  return {
+    raw,
+    frontmatter: parseError ? {} : fm,
+    frontmatter_parse_error: parseError,
+  };
 }
 
-/**
- * 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[];
+function listFiles(rootDir: string, recursive: boolean): string[] {
+  let entries: import("node:fs").Dirent[];
   try {
-    entries = readdirSync(spec.path);
-  } catch {
-    return [];
+    entries = readdirSync(rootDir, { withFileTypes: true });
+  } catch (err) {
+    if ((err as NodeJS.ErrnoException).code === "ENOENT") return [];
+    throw err;
   }
-  const exts = spec.extensions ?? [".md"];
-  const includeStatuses = spec.include_statuses;
-  const out: ScannedArtifact[] = [];
+  const out: string[] = [];
   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 {
+    const full = join(rootDir, entry.name);
+    if (entry.isDirectory()) {
+      if (recursive) out.push(...listFiles(full, recursive));
       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 });
+    if (entry.isFile()) out.push(full);
   }
-  return out;
+  return out.sort((a, b) => a.localeCompare(b));
+}
+
+function shouldIncludeFile(filePath: string, pool: ArtifactPoolSpec): boolean {
+  const name = basename(filePath);
+  const ignoreNames = new Set([
+    ...DEFAULT_IGNORE_NAMES,
+    ...(Array.isArray(pool.ignore_names) ? pool.ignore_names : []),
+  ]);
+  if (ignoreNames.has(name)) return false;
+  const exts = Array.isArray(pool.extensions) ? pool.extensions : DEFAULT_EXTENSIONS;
+  return exts.some((e) => name.endsWith(e));
+}
+
+function statusAllowed(status: string | null, pool: ArtifactPoolSpec): boolean {
+  const include = pool.include_statuses;
+  if (!Array.isArray(include) || include.length === 0) return true;
+  return status !== null && include.includes(status);
 }
 
 /**
  * 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.
+ * all pools, sorted by absolute path. Missing pools yield empty.
+ *
+ * Throws if `pools` is undefined / empty / shapeless — POC contract:
+ * the policy spec MUST declare at least one pool.
  */
 export async function scanArtifactPools(
   pools: ArtifactPoolSpec | ArtifactPoolSpec[] | undefined,
 ): Promise<ScannedArtifact[]> {
+  const expanded = expandPools(pools);
+  if (expanded.length === 0) {
+    throw new Error("artifact pool policy: context pool list is required");
+  }
+  const out: ScannedArtifact[] = [];
+  for (const pool of expanded) {
+    if (!pool.path) {
+      throw new Error("artifact pool policy: every pool requires a path");
+    }
+    const files = listFiles(pool.path, Boolean(pool.recursive));
+    for (const filePath of files) {
+      if (!shouldIncludeFile(filePath, pool)) continue;
+      let st;
+      try {
+        st = statSync(filePath);
+      } catch {
+        continue;
+      }
+      if (!st.isFile()) continue;
+      let parsed;
+      try {
+        parsed = readFrontmatter(filePath);
+      } catch {
+        continue;
+      }
+      if (parsed.frontmatter_parse_error && !pool.include_malformed_frontmatter) {
+        continue;
+      }
+      const status = parsed.frontmatter.status ?? null;
+      if (!statusAllowed(status, pool)) continue;
+      out.push({
+        path: filePath,
+        pool_path: pool.path,
+        raw: parsed.raw,
+        frontmatter: parsed.frontmatter,
+        frontmatter_parse_error: parsed.frontmatter_parse_error,
+        status,
+      });
+    }
+  }
+  return out.sort((a, b) => a.path.localeCompare(b.path));
+}
+
+/**
+ * Expand the convenience input shape (single object, array of objects,
+ * or `paths: [...]` shorthand) into a flat array of single-path pools.
+ */
+function expandPools(
+  pools: ArtifactPoolSpec | ArtifactPoolSpec[] | undefined,
+): ArtifactPoolSpec[] {
   if (!pools) return [];
   const list = Array.isArray(pools) ? pools : [pools];
-  const out: ScannedArtifact[] = [];
+  const out: ArtifactPoolSpec[] = [];
   for (const spec of list) {
     if (spec.path) {
-      out.push(...scanPool({ ...spec, path: spec.path }));
+      out.push(spec);
     } else if (Array.isArray(spec.paths)) {
-      for (const p of spec.paths) out.push(...scanPool({ ...spec, path: p }));
+      for (const p of spec.paths) out.push({ ...spec, path: p, paths: undefined });
     }
   }
   return out;
 }
 
 /**
- * Format an artifact list as bullet lines. Mirrors POC formatArtifactList.
+ * Format an artifact list as bullet lines. Mirrors POC formatArtifactList:
+ * `- /full/absolute/path`. The POC favors the full path so receivers can
+ * cd / open / cat the artifact directly without further lookup.
  */
 export function formatArtifactList(artifacts: ScannedArtifact[], maxItems: number): string {
   return artifacts
     .slice(0, maxItems)
-    .map((a) => `- ${basename(a.path)} (${a.path})`)
+    .map((a) => `- ${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.
+ * Compute the source-key for an artifact. 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;
+  return String(value);
 }

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

@@ -1,10 +1,10 @@
-// PL-004 Phase C: artifact-pool-ready policy (TypeScript port of
-// POC `lib/policies/artifact-pool-ready.mjs`, 29 lines).
+// PL-004 Phase C R1: artifact-pool-ready policy (TypeScript port of
+// POC `lib/policies/artifact-pool-ready.mjs`).
 //
-// 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) }
+// POC contract: target.session is required (top-level `target:`).
+// Scans context.pools (array). Empty → skip(no_actionable_artifacts).
+// Otherwise sends a formatted message ending with the standard POC
+// pool-ready instruction line.
 
 import type { Policy, PolicyEvaluation, PolicyJob } from "./types.js";
 import {
@@ -14,38 +14,41 @@ import {
 } from "./artifact-pool-helpers.js";
 
 interface ArtifactPoolReadyContext {
-  target?: { session?: string };
   pools?: ArtifactPoolSpec | ArtifactPoolSpec[];
   label?: string;
   max_items?: number;
-  message_template?: string;
 }
 
+const POC_POOL_READY_TRAILER =
+  "Claim and process the next artifact, or terminally classify it with evidence. " +
+  "This is a pool-ready wake, not an approval gate, and it should not require orch to bridge the edge.";
+
 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"), {
+    if (!job.target?.session) {
+      throw Object.assign(new Error("artifact-pool-ready: target.session is required"), {
         code: "policy_spec_invalid",
         policy: "artifact-pool-ready",
-        field: "context.target.session",
+        field: "target.session",
       });
     }
+    const context = job.context as ArtifactPoolReadyContext;
     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 maxItems = Number(context.max_items ?? 5);
     const list = formatArtifactList(artifacts, maxItems);
+    const hiddenCount = Math.max(0, artifacts.length - maxItems);
+    const suffix = hiddenCount > 0 ? `\n- ... ${hiddenCount} more` : "";
     const message =
-      context.message_template ??
-      `Artifact pool ready: ${label} has ${artifacts.length} actionable artifact(s).\n\n${list}`;
+      job.message ??
+      `Artifact pool ready: ${label} has ${artifacts.length} actionable artifact(s).\n${list}${suffix}\n\n${POC_POOL_READY_TRAILER}`;
     return {
       action: "send",
-      target: targetSession,
+      target: job.target,
       message,
       notes: { artifact_count: artifacts.length, label },
     };

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

@@ -1,87 +1,112 @@
-// PL-004 Phase C: edge-artifact-required policy (TypeScript port of
-// POC `lib/policies/edge-artifact-required.mjs`, 70 lines).
+// PL-004 Phase C R1: edge-artifact-required policy (TypeScript port of
+// POC `lib/policies/edge-artifact-required.mjs`).
 //
-// 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".
+// R1 fix (guard blocker 2): preserves the POC contract.
+//   - Spec uses `context.source` (singular) and `context.target`
+//     (singular) pool specs. Each may have `path` or `paths`.
+//   - Edge-satisfaction: a downstream target artifact satisfies a
+//     source iff target.raw CONTAINS the source key. NOT a frontmatter
+//     match — body reference suffices.
+//   - Target scan overrides include_statuses to [] so any downstream
+//     state counts as "exists".
+//   - Delivery target is `job.target.session` (top-level).
+//   - Label key is `context.edge_label` (matches POC).
 
 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;
+  source?: ArtifactPoolSpec & { paths?: string[] };
+  target?: ArtifactPoolSpec & { paths?: string[] };
+  edge_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;
+function targetContainsKey(targets: ScannedArtifact[], key: string): boolean {
+  return targets.some((t) => t.raw.includes(key));
+}
+
+function expandPoolSides(
+  spec: (ArtifactPoolSpec & { paths?: string[] }) | undefined,
+): ArtifactPoolSpec[] {
+  if (!spec) return [];
+  if (Array.isArray(spec.paths)) {
+    return spec.paths.map((p) => ({ ...spec, path: p, paths: undefined }));
   }
-  return false;
+  return [spec];
 }
 
 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"), {
+    if (!job.target?.session) {
+      throw Object.assign(new Error("edge-artifact-required: target.session is required"), {
         code: "policy_spec_invalid",
         policy: "edge-artifact-required",
-        field: "context.target.session",
+        field: "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 context = job.context as EdgeArtifactRequiredContext;
+    const source = context.source;
+    const target = context.target;
+    if (!source?.path && !Array.isArray(source?.paths)) {
+      throw Object.assign(
+        new Error("edge-artifact-required: context.source.path is required"),
+        {
+          code: "policy_spec_invalid",
+          policy: "edge-artifact-required",
+          field: "context.source.path",
+        },
+      );
+    }
+    if (!target?.path && !Array.isArray(target?.paths)) {
+      throw Object.assign(
+        new Error("edge-artifact-required: context.target.path is required"),
+        {
+          code: "policy_spec_invalid",
+          policy: "edge-artifact-required",
+          field: "context.target.path",
+        },
+      );
+    }
+    const sourcePools = expandPoolSides(source);
+    const targetPools = expandPoolSides(target).map((p) => ({
+      ...p,
+      include_statuses: [] as string[],
+    }));
+    const keyField = source?.key_field ?? "entry";
+
+    const sources = await scanArtifactPools(sourcePools);
+    const targets = await scanArtifactPools(targetPools);
     const missing: ScannedArtifact[] = [];
     for (const s of sources) {
-      const key = sourceKeyFor(s, sourceKeyField);
-      if (!targetContainsKey(targets, key, targetKeyField)) missing.push(s);
+      const key = sourceKeyFor(s, keyField);
+      if (!targetContainsKey(targets, key)) 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 label = context.edge_label ?? "artifact edge";
+    const maxItems = Number(context.max_items ?? 5);
+    const list = missing
+      .slice(0, maxItems)
+      .map((a) => `- ${sourceKeyFor(a, keyField)} (${a.path})`)
+      .join("\n");
+    const hiddenCount = Math.max(0, missing.length - maxItems);
+    const suffix = hiddenCount > 0 ? `\n- ... ${hiddenCount} more` : "";
     const message =
-      context.message_template ??
-      `Artifact edge repair needed: ${label} has ${missing.length} upstream artifact(s) with no matching downstream artifact.\n\n${list}`;
+      job.message ??
+      `Artifact edge repair needed: ${label} has ${missing.length} upstream artifact(s) with no matching downstream artifact.\n${list}${suffix}\n\n` +
+        "Producer loop owns creating the missing downstream artifact. " +
+        "Do not wait for orch to bridge this manually; repair the edge or record the blocker with evidence.";
     return {
       action: "send",
-      target: targetSession,
+      target: job.target,
       message,
       notes: { missing_count: missing.length, label },
     };

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

@@ -1,38 +1,36 @@
 // PL-004 Phase C: periodic-reminder policy (TypeScript port of
-// POC `lib/policies/periodic-reminder.mjs`, 15 lines).
+// POC `lib/policies/periodic-reminder.mjs`).
 //
-// 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).
+// POC contract: target.session is required (top-level `target:` in
+// spec_yaml). Message comes from `job.message` (top-level) OR
+// `context.message`. 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> {
+    if (!job.target?.session) {
+      throw Object.assign(new Error("periodic-reminder: target.session is required"), {
+        code: "policy_spec_invalid",
+        policy: "periodic-reminder",
+        field: "target.session",
+      });
+    }
     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",
-      });
-    }
+    const message = job.message ?? context.message;
     if (!message) {
-      throw Object.assign(new Error("periodic-reminder requires context.message"), {
+      throw Object.assign(new Error("periodic-reminder: message is required (top-level message or context.message)"), {
         code: "policy_spec_invalid",
         policy: "periodic-reminder",
-        field: "context.message",
+        field: "message",
       });
     }
-    return { action: "send", target: targetSession, message };
+    return { action: "send", target: job.target, message };
   },
 };

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

@@ -9,7 +9,14 @@
 export interface PolicyJob {
   jobId: string;
   policy: string;
-  targetSession: string;
+  /**
+   * POC-shape target object. Built from spec_yaml top-level `target:`
+   * block when present; otherwise falls back to `{session: registered
+   * targetSession}`. Policies access `job.target.session` per POC.
+   */
+  target: { session: string };
+  /** Optional top-level message override (POC pattern: `job.message`). */
+  message?: string;
   intervalSeconds: number;
   activeWakeIntervalSeconds: number | null;
   scanIntervalSeconds: number | null;
@@ -22,7 +29,7 @@ export interface PolicyJob {
 }
 
 export type PolicyEvaluation =
-  | { action: "send"; target: string; message: string; notes?: Record<string, unknown> }
+  | { action: "send"; target: { session: string }; message: string; notes?: Record<string, unknown> }
   | { action: "skip"; reason: string; notes?: Record<string, unknown> }
   | { action: "terminal"; reason: string; notes?: Record<string, unknown> };
 

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

@@ -36,6 +36,8 @@ export interface WatchdogJob {
   scanIntervalSeconds: number | null;
   lastEvaluationAt: string | null;
   lastFireAt: string | null;
+  actionable: boolean;
+  lastActionableAt: string | null;
   state: WatchdogJobState;
   registeredBySession: string;
   registeredAt: string;
@@ -62,6 +64,8 @@ interface JobRow {
   scan_interval_seconds: number | null;
   last_evaluation_at: string | null;
   last_fire_at: string | null;
+  actionable: number;
+  last_actionable_at: string | null;
   state: string;
   registered_by_session: string;
   registered_at: string;
@@ -186,6 +190,38 @@ export class WatchdogJobsRepository {
     }
   }
 
+  /**
+   * R1 fix: write the actionable-state machine columns. Mirrors POC
+   * engine's `state.actionable` + `state.last_actionable_at`. Called
+   * by the policy engine after every meaningful evaluation:
+   *   - newActionable=false (skip): clears actionable + last_actionable_at.
+   *   - newActionable=true with no preserveLastActionableAt: stamps
+   *     last_actionable_at to evaluatedAt (newly actionable).
+   *   - newActionable=true with preserveLastActionableAt set: keeps the
+   *     existing first-actionable timestamp (continued actionable window).
+   */
+  setActionable(
+    jobId: string,
+    newActionable: boolean,
+    evaluatedAt: string,
+    preserveLastActionableAt: string | null = null,
+  ): void {
+    if (!newActionable) {
+      this.db
+        .prepare(
+          `UPDATE watchdog_jobs SET actionable = 0, last_actionable_at = NULL WHERE job_id = ?`,
+        )
+        .run(jobId);
+      return;
+    }
+    const stamp = preserveLastActionableAt ?? evaluatedAt;
+    this.db
+      .prepare(
+        `UPDATE watchdog_jobs SET actionable = 1, last_actionable_at = ? WHERE job_id = ?`,
+      )
+      .run(stamp, jobId);
+  }
+
   markTerminal(jobId: string, reason: string): void {
     this.db
       .prepare(
@@ -222,6 +258,8 @@ function rowToJob(row: JobRow): WatchdogJob {
     scanIntervalSeconds: row.scan_interval_seconds,
     lastEvaluationAt: row.last_evaluation_at,
     lastFireAt: row.last_fire_at,
+    actionable: row.actionable !== 0,
+    lastActionableAt: row.last_actionable_at,
     state: row.state as WatchdogJobState,
     registeredBySession: row.registered_by_session,
     registeredAt: row.registered_at,

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

@@ -10,21 +10,28 @@ import {
 import type { WatchdogHistoryEntry, WatchdogHistoryLog } from "./watchdog-history-log.js";
 
 /**
- * Watchdog policy engine (PL-004 Phase C).
+ * Watchdog policy engine (PL-004 Phase C R1).
  *
- * 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.
+ * Owns the per-evaluation state machine that mirrors POC engine
+ * `lib/engine.mjs`:
+ *   1. Resolve policy by name from registry. Unknown policy → terminal.
+ *   2. Parse spec_yaml into top-level `target:` + top-level `message?:` +
+ *      nested `context:` block, then build a PolicyJob with the
+ *      resolved target object (falls back to registered targetSession
+ *      when spec lacks an explicit `target:`).
+ *   3. Dispatch policy.evaluate(policyJob).
+ *   4. Action handling:
+ *      - skip: clear actionable=false; record history + emit event
+ *        ONLY if reason is "loud" (not in QUIET_SKIP_REASONS).
+ *      - send: enforce active-wake throttle. If state.actionable was
+ *        already true AND active_wake_interval_seconds is set AND not
+ *        elapsed since last_fire_at → emit/record `active_wake_not_due`
+ *        (quiet skip per POC). Otherwise call delivery, record history
+ *        with sent outcome, emit evaluation_fired, set actionable=true.
+ *      - terminal: mark job terminal, record history, emit 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.
+ * `not_due` polls are filtered upstream by the scheduler and never
+ * reach this engine.
  */
 
 export interface DeliveryRequest {
@@ -40,8 +47,16 @@ export interface DeliveryOutcome {
 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>;
+  /**
+   * Parse the operator-supplied spec_yaml into the structured fields the
+   * engine needs. Returns the top-level `target` (object), top-level
+   * `message` (string), and the `context:` block (Record).
+   */
+  (specYaml: string): {
+    target: { session: string } | null;
+    message: string | null;
+    context: Record<string, unknown>;
+  };
 }
 
 interface WatchdogPolicyEngineDeps {
@@ -49,8 +64,7 @@ interface WatchdogPolicyEngineDeps {
   historyLog: WatchdogHistoryLog;
   eventBus: EventBus;
   deliver: DeliveryFn;
-  /** Optional override: defaults to a YAML-aware parser. */
-  parseContext?: PolicyContextParser;
+  parseSpec?: PolicyContextParser;
   now?: () => Date;
 }
 
@@ -60,11 +74,26 @@ const POLICY_REGISTRY: Map<string, Policy> = new Map([
   [edgeArtifactRequiredPolicy.name, edgeArtifactRequiredPolicy],
 ]);
 
+/**
+ * Quiet skip reasons — POC `shouldAppendHistory` (engine.mjs:99-112)
+ * suppresses these from history. Same set must NOT emit watchdog.*
+ * events for parity with POC SSE behavior; agents never see scheduler
+ * polls just because the pool was empty or the wake throttle was active.
+ */
+const QUIET_SKIP_REASONS = new Set<string>([
+  "not_due",
+  "no_actionable_artifacts",
+  "no_missing_edge_artifacts",
+  "active_wake_not_due",
+]);
+
 export interface EvaluationResult {
   job: WatchdogJob;
-  outcome: PolicyEvaluation;
+  outcome: PolicyEvaluation | { action: "skip"; reason: "active_wake_not_due" };
   history: WatchdogHistoryEntry | null;
   delivery: DeliveryOutcome | null;
+  /** True if this evaluation produced a history record + event. */
+  meaningful: boolean;
 }
 
 export class WatchdogPolicyEngine {
@@ -72,7 +101,7 @@ export class WatchdogPolicyEngine {
   private readonly historyLog: WatchdogHistoryLog;
   private readonly eventBus: EventBus;
   private readonly deliver: DeliveryFn;
-  private readonly parseContext: PolicyContextParser;
+  private readonly parseSpec: PolicyContextParser;
   private readonly now: () => Date;
 
   constructor(deps: WatchdogPolicyEngineDeps) {
@@ -80,7 +109,7 @@ export class WatchdogPolicyEngine {
     this.historyLog = deps.historyLog;
     this.eventBus = deps.eventBus;
     this.deliver = deps.deliver;
-    this.parseContext = deps.parseContext ?? defaultParseContext;
+    this.parseSpec = deps.parseSpec ?? defaultParseSpec;
     this.now = deps.now ?? (() => new Date());
   }
 
@@ -88,21 +117,13 @@ export class WatchdogPolicyEngine {
     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);
+    const evaluatedAt = this.now().toISOString();
+
     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,
@@ -120,18 +141,21 @@ export class WatchdogPolicyEngine {
         outcome: { action: "terminal", reason },
         history,
         delivery: null,
+        meaningful: true,
       };
     }
 
-    const context = this.parseContext(job.specYaml);
+    const parsed = this.parseSpec(job.specYaml);
+    const target = parsed.target ?? { session: job.targetSession };
     const policyJob: PolicyJob = {
       jobId: job.jobId,
       policy: job.policy,
-      targetSession: job.targetSession,
+      target,
+      message: parsed.message ?? undefined,
       intervalSeconds: job.intervalSeconds,
       activeWakeIntervalSeconds: job.activeWakeIntervalSeconds,
       scanIntervalSeconds: job.scanIntervalSeconds,
-      context,
+      context: parsed.context,
       lastEvaluationAt: job.lastEvaluationAt,
       lastFireAt: job.lastFireAt,
       registeredBySession: job.registeredBySession,
@@ -139,39 +163,22 @@ export class WatchdogPolicyEngine {
     };
 
     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") {
+      // POC parity: skip clears actionable. Loud-vs-quiet decides
+      // whether to record + emit.
+      this.jobsRepo.recordEvaluation(job.jobId, evaluatedAt, false);
+      this.jobsRepo.setActionable(job.jobId, false, evaluatedAt);
+      const isQuiet = QUIET_SKIP_REASONS.has(outcome.reason);
+      if (isQuiet) {
+        return {
+          job: this.jobsRepo.getByIdOrThrow(job.jobId),
+          outcome,
+          history: null,
+          delivery: null,
+          meaningful: false,
+        };
+      }
       const history = this.historyLog.record({
         jobId: job.jobId,
         evaluatedAt,
@@ -179,7 +186,6 @@ export class WatchdogPolicyEngine {
         skipReason: outcome.reason,
         evaluationNotes: outcome.notes ?? null,
       });
-      this.jobsRepo.recordEvaluation(job.jobId, evaluatedAt, false);
       this.eventBus.emit({
         type: "watchdog.evaluation_skipped",
         jobId: job.jobId,
@@ -191,91 +197,196 @@ export class WatchdogPolicyEngine {
         outcome,
         history,
         delivery: null,
+        meaningful: true,
       };
     }
 
-    // terminal
-    this.jobsRepo.markTerminal(job.jobId, outcome.reason);
+    if (outcome.action === "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,
+        meaningful: true,
+      };
+    }
+
+    // outcome.action === "send".
+    // POC active-wake throttle (engine.mjs:49-64, :243-263):
+    //   - If state.actionable was already true AND active_wake_interval
+    //     is set AND wake-window has not elapsed → quiet skip. Preserves
+    //     existing last_fire_at and last_actionable_at.
+    //   - Otherwise: deliver, set actionable=true, stamp last_fire_at +
+    //     last_actionable_at (preserve existing first-actionable timestamp).
+    if (
+      job.actionable &&
+      job.activeWakeIntervalSeconds !== null &&
+      job.lastFireAt !== null
+    ) {
+      const lastFireMs = Date.parse(job.lastFireAt);
+      const nowMs = Date.parse(evaluatedAt);
+      const intervalMs = job.activeWakeIntervalSeconds * 1000;
+      if (Number.isFinite(lastFireMs) && nowMs - lastFireMs < intervalMs) {
+        // Quiet skip: scan happened, pool still actionable, but the
+        // wake window is closed. Update last_evaluation_at, do NOT
+        // touch last_fire_at, keep actionable=true and preserve
+        // last_actionable_at.
+        this.jobsRepo.recordEvaluation(job.jobId, evaluatedAt, false);
+        this.jobsRepo.setActionable(job.jobId, true, evaluatedAt, job.lastActionableAt);
+        return {
+          job: this.jobsRepo.getByIdOrThrow(job.jobId),
+          outcome: { action: "skip", reason: "active_wake_not_due" },
+          history: null,
+          delivery: null,
+          meaningful: false,
+        };
+      }
+    }
+
+    const delivery = await this.deliver({
+      targetSession: outcome.target.session,
+      message: outcome.message,
+    });
     const history = this.historyLog.record({
       jobId: job.jobId,
       evaluatedAt,
-      outcome: "terminal",
-      skipReason: outcome.reason,
+      outcome: "sent",
+      deliveryTargetSession: outcome.target.session,
+      deliveryStatus: delivery.status,
+      deliveryMessage: outcome.message,
       evaluationNotes: outcome.notes ?? null,
     });
+    this.jobsRepo.recordEvaluation(job.jobId, evaluatedAt, true);
+    this.jobsRepo.setActionable(job.jobId, true, evaluatedAt, job.lastActionableAt);
     this.eventBus.emit({
-      type: "watchdog.evaluation_terminal",
+      type: "watchdog.evaluation_fired",
       jobId: job.jobId,
       policy: job.policy,
-      terminalReason: outcome.reason,
+      targetSession: outcome.target.session,
+      deliveryStatus: delivery.status,
     });
     return {
       job: this.jobsRepo.getByIdOrThrow(job.jobId),
       outcome,
       history,
-      delivery: null,
+      delivery,
+      meaningful: true,
     };
   }
 }
 
 /**
- * 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:
+ * Default spec_yaml parser. Extracts:
+ *   - top-level `target:` block (POC contract: `{session: "..."}`)
+ *   - top-level `message:` scalar (POC pattern for periodic-reminder)
+ *   - nested `context:` block as Record<string, unknown>
+ *
+ * Uses a minimal recursive indentation-based parser sufficient for the
+ * POC spec set:
  *   policy: ...
- *   target: <member>@<rig>
+ *   target:
+ *     session: ...
+ *   message: "..."
  *   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
+ * For richer YAML, a future maintenance pass can swap in `yaml` /
  * `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> {
+function defaultParseSpec(specYaml: string): {
+  target: { session: string } | null;
+  message: string | null;
+  context: Record<string, unknown>;
+} {
   const lines = specYaml.split("\n");
+  const result: { target: { session: string } | null; message: string | null; context: Record<string, unknown> } = {
+    target: null,
+    message: null,
+    context: {},
+  };
   let i = 0;
   while (i < lines.length) {
     const line = lines[i] ?? "";
-    if (/^context\s*:\s*$/.test(line)) {
+    if (line.trimStart() !== line) {
+      // Indented line at top level — skip until we re-hit indent 0.
       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>;
+      continue;
     }
+    const trimmed = line.trim();
+    if (trimmed.length === 0 || trimmed.startsWith("#")) {
+      i++;
+      continue;
+    }
+    const colonIdx = trimmed.indexOf(":");
+    if (colonIdx === -1) {
+      i++;
+      continue;
+    }
+    const key = trimmed.slice(0, colonIdx).trim();
+    const value = trimmed.slice(colonIdx + 1).trim();
+    if (value.length > 0) {
+      // Inline scalar.
+      if (key === "message") {
+        result.message = stripQuotes(value);
+      }
+      i++;
+      continue;
+    }
+    // Block. Collect indented lines below.
     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;
+    }
+    const dedented = block.map((l) => (l.startsWith("  ") ? l.slice(2) : l)).join("\n");
+    if (key === "context") {
+      result.context = parseSimpleYaml(dedented) as Record<string, unknown>;
+    } else if (key === "target") {
+      const parsed = parseSimpleYaml(dedented);
+      if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) {
+        const sess = (parsed as Record<string, unknown>).session;
+        if (typeof sess === "string") result.target = { session: sess };
+      }
+    }
   }
-  return {};
+  return result;
+}
+
+function stripQuotes(value: string): string {
+  if (value.startsWith('"') && value.endsWith('"')) return value.slice(1, -1);
+  if (value.startsWith("'") && value.endsWith("'")) return value.slice(1, -1);
+  return value;
 }
 
-/**
- * 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 };
@@ -300,9 +411,7 @@ function parseScalar(raw: string): unknown {
     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;
+  return stripQuotes(trimmed);
 }
 
 function parseYamlBlock(lines: string[], idx: { i: number }, expectedIndent: number): unknown {
@@ -337,7 +446,6 @@ function parseYamlMap(
     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);
@@ -368,9 +476,6 @@ function parseYamlSeq(
     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();
@@ -382,7 +487,6 @@ function parseYamlSeq(
       } 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);

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

@@ -138,9 +138,17 @@ export class WatchdogScheduler {
   }
 }
 
+/**
+ * R1 fix: scan cadence is governed by `scan_interval_seconds` when
+ * supplied, falling back to `interval_seconds`. Mirrors POC engine
+ * (lib/engine.mjs:38-46) where `last_scan_at` and `scan_interval_seconds`
+ * gate the policy invocation. Wake-cadence (`active_wake_interval_seconds`)
+ * is enforced one layer up by the policy engine, NOT here.
+ */
 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;
+  const cadenceSeconds = job.scanIntervalSeconds ?? job.intervalSeconds;
+  return nowMs - last >= cadenceSeconds * 1000;
 }

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

@@ -35,11 +35,13 @@ describe("WatchdogJobsRepository (PL-004 Phase C)", () => {
     };
   }
 
-  it("register stores all three v1 policies", () => {
+  it("register stores all three v1 policies + actionable defaults to false", () => {
     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.actionable).toBe(false);
+      expect(job.lastActionableAt).toBeNull();
       expect(job.jobId).toMatch(/^[0-9A-Z]{26}$/);
     }
   });
@@ -138,6 +140,31 @@ describe("WatchdogJobsRepository (PL-004 Phase C)", () => {
     }
   });
 
+  it("setActionable(true) sets actionable=1 + last_actionable_at to evaluatedAt by default", () => {
+    const job = repo.register(validInput());
+    repo.setActionable(job.jobId, true, "2026-05-03T07:00:00.000Z");
+    const after = repo.getByIdOrThrow(job.jobId);
+    expect(after.actionable).toBe(true);
+    expect(after.lastActionableAt).toBe("2026-05-03T07:00:00.000Z");
+  });
+
+  it("setActionable(true) preserves last_actionable_at when preserve arg passed (continued window)", () => {
+    const job = repo.register(validInput());
+    repo.setActionable(job.jobId, true, "2026-05-03T07:00:00.000Z");
+    repo.setActionable(job.jobId, true, "2026-05-03T07:01:00.000Z", "2026-05-03T07:00:00.000Z");
+    const after = repo.getByIdOrThrow(job.jobId);
+    expect(after.lastActionableAt).toBe("2026-05-03T07:00:00.000Z");
+  });
+
+  it("setActionable(false) clears actionable + last_actionable_at", () => {
+    const job = repo.register(validInput());
+    repo.setActionable(job.jobId, true, "2026-05-03T07:00:00.000Z");
+    repo.setActionable(job.jobId, false, "2026-05-03T07:01:00.000Z");
+    const after = repo.getByIdOrThrow(job.jobId);
+    expect(after.actionable).toBe(false);
+    expect(after.lastActionableAt).toBeNull();
+  });
+
   it("getByIdOrThrow throws job_not_found for unknown id", () => {
     try {
       repo.getByIdOrThrow("does-not-exist");

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

@@ -11,7 +11,7 @@ function makeJob(overrides: Partial<PolicyJob> & { context: Record<string, unkno
   return {
     jobId: "job-1",
     policy: "periodic-reminder",
-    targetSession: "a@rig",
+    target: { session: "a@rig" },
     intervalSeconds: 60,
     activeWakeIntervalSeconds: null,
     scanIntervalSeconds: null,
@@ -23,28 +23,42 @@ function makeJob(overrides: Partial<PolicyJob> & { context: Record<string, unkno
   };
 }
 
-describe("periodicReminderPolicy", () => {
-  it("returns send with explicit target+message from context", async () => {
+describe("periodicReminderPolicy (POC contract)", () => {
+  it("returns send with target object + message from job.message", async () => {
     const out = await periodicReminderPolicy.evaluate(
       makeJob({
-        context: { target: { session: "alice@rig" }, message: "ping" },
+        target: { session: "alice@rig" },
+        message: "ping",
+        context: {},
       }),
     );
-    expect(out).toEqual({ action: "send", target: "alice@rig", message: "ping" });
+    expect(out).toEqual({ action: "send", target: { session: "alice@rig" }, message: "ping" });
   });
 
-  it("throws policy_spec_invalid when context.target.session is missing", async () => {
+  it("returns send with message from context.message when job.message absent", async () => {
+    const out = await periodicReminderPolicy.evaluate(
+      makeJob({
+        target: { session: "alice@rig" },
+        context: { message: "ctx-ping" },
+      }),
+    );
+    expect(out).toEqual({ action: "send", target: { session: "alice@rig" }, message: "ctx-ping" });
+  });
+
+  it("throws policy_spec_invalid when target.session is missing", async () => {
     try {
-      await periodicReminderPolicy.evaluate(makeJob({ context: { message: "x" } }));
+      await periodicReminderPolicy.evaluate(
+        makeJob({ target: { session: "" }, 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 () => {
+  it("throws policy_spec_invalid when no message anywhere", async () => {
     try {
-      await periodicReminderPolicy.evaluate(makeJob({ context: { target: { session: "a@rig" } } }));
+      await periodicReminderPolicy.evaluate(makeJob({ context: {} }));
       throw new Error("should have thrown");
     } catch (err) {
       expect((err as Error & { code: string }).code).toBe("policy_spec_invalid");
@@ -52,7 +66,7 @@ describe("periodicReminderPolicy", () => {
   });
 });
 
-describe("artifactPoolReadyPolicy", () => {
+describe("artifactPoolReadyPolicy (POC contract)", () => {
   let tmp: string;
   beforeEach(() => {
     tmp = join(tmpdir(), `watchdog-pool-${Date.now()}-${Math.random()}`);
@@ -63,7 +77,7 @@ describe("artifactPoolReadyPolicy", () => {
   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 } },
+        context: { pools: [{ path: tmp }] },
       }),
     );
     expect(out).toEqual({ action: "skip", reason: "no_actionable_artifacts" });
@@ -74,19 +88,21 @@ describe("artifactPoolReadyPolicy", () => {
     writeFileSync(join(tmp, "b.md"), "---\nstatus: ready\n---\nbody-b\n");
     const out = await artifactPoolReadyPolicy.evaluate(
       makeJob({
+        target: { session: "a@rig" },
         context: {
-          target: { session: "a@rig" },
-          pools: { path: tmp, include_statuses: ["ready"] },
+          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.target).toEqual({ session: "a@rig" });
     expect(out.message).toMatch(/things has 2 actionable artifact/);
     expect(out.message).toContain("a.md");
     expect(out.message).toContain("b.md");
+    // POC trailer message
+    expect(out.message).toContain("Claim and process the next artifact");
   });
 
   it("respects include_statuses filter (artifact with non-matching status is excluded)", async () => {
@@ -94,10 +110,7 @@ describe("artifactPoolReadyPolicy", () => {
     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"] },
-        },
+        context: { pools: [{ path: tmp, include_statuses: ["ready"] }] },
       }),
     );
     expect(out.action).toBe("send");
@@ -107,15 +120,14 @@ describe("artifactPoolReadyPolicy", () => {
     expect(out.message).not.toContain("b.md");
   });
 
-  it("respects max_items cap in formatted bullet list", async () => {
+  it("respects max_items cap (formatted bullet list capped)", 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"] },
+          pools: [{ path: tmp, include_statuses: ["ready"] }],
           max_items: 3,
         },
       }),
@@ -123,22 +135,115 @@ describe("artifactPoolReadyPolicy", () => {
     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;
+    const bulletCount = out.message
+      .split("\n")
+      .filter((l) => l.startsWith("- ") && !l.startsWith("- ..."))
+      .length;
     expect(bulletCount).toBe(3);
+    expect(out.message).toContain("- ... 4 more");
   });
 
-  it("missing pool directory yields skip (no actionable)", async () => {
+  it("missing pool directory yields skip (ENOENT-tolerant)", async () => {
     rmSync(tmp, { recursive: true, force: true });
     const out = await artifactPoolReadyPolicy.evaluate(
       makeJob({
-        context: { target: { session: "a@rig" }, pools: { path: tmp } },
+        context: { pools: [{ path: tmp }] },
+      }),
+    );
+    expect(out).toEqual({ action: "skip", reason: "no_actionable_artifacts" });
+  });
+
+  // R1 fix: POC scanner parity (guard blocker 3).
+  it("default-ignores README.md (POC parity)", async () => {
+    writeFileSync(join(tmp, "README.md"), "# Pool docs\n");
+    writeFileSync(join(tmp, "ready.md"), "---\nstatus: ready\n---\n");
+    const out = await artifactPoolReadyPolicy.evaluate(
+      makeJob({
+        context: { pools: [{ path: tmp, include_statuses: ["ready"] }] },
+      }),
+    );
+    expect(out.action).toBe("send");
+    if (out.action !== "send") return;
+    expect(out.message).toContain("ready.md");
+    expect(out.message).not.toContain("README.md");
+  });
+
+  it("default-ignores .DS_Store (POC parity)", async () => {
+    writeFileSync(join(tmp, ".DS_Store"), "binary-junk");
+    writeFileSync(join(tmp, "ready.md"), "---\nstatus: ready\n---\n");
+    const out = await artifactPoolReadyPolicy.evaluate(
+      makeJob({
+        context: { pools: [{ path: tmp, include_statuses: ["ready"] }] },
+      }),
+    );
+    expect(out.action).toBe("send");
+    if (out.action !== "send") return;
+    expect(out.message).not.toContain(".DS_Store");
+  });
+
+  it("excludes malformed-frontmatter artifacts unless include_malformed_frontmatter=true", async () => {
+    writeFileSync(join(tmp, "ready.md"), "---\nstatus: ready\n---\n");
+    writeFileSync(
+      join(tmp, "malformed.md"),
+      "---\nstatus: ready\nbroken: value: still broken\n---\n",
+    );
+    const out = await artifactPoolReadyPolicy.evaluate(
+      makeJob({
+        context: { pools: [{ path: tmp, include_statuses: ["ready"] }] },
+      }),
+    );
+    expect(out.action).toBe("send");
+    if (out.action !== "send") return;
+    expect(out.message).toContain("ready.md");
+    expect(out.message).not.toContain("malformed.md");
+  });
+
+  it("supports configured ignore_names (POC parity)", async () => {
+    writeFileSync(join(tmp, "skip-me.md"), "---\nstatus: ready\n---\n");
+    writeFileSync(join(tmp, "include-me.md"), "---\nstatus: ready\n---\n");
+    const out = await artifactPoolReadyPolicy.evaluate(
+      makeJob({
+        context: {
+          pools: [{ path: tmp, include_statuses: ["ready"], ignore_names: ["skip-me.md"] }],
+        },
+      }),
+    );
+    expect(out.action).toBe("send");
+    if (out.action !== "send") return;
+    expect(out.message).toContain("include-me.md");
+    expect(out.message).not.toContain("skip-me.md");
+  });
+
+  it("recursive=true descends into subdirectories", async () => {
+    const sub = join(tmp, "sub");
+    mkdirSync(sub, { recursive: true });
+    writeFileSync(join(sub, "nested-ready.md"), "---\nstatus: ready\n---\n");
+    const out = await artifactPoolReadyPolicy.evaluate(
+      makeJob({
+        context: {
+          pools: [{ path: tmp, include_statuses: ["ready"], recursive: true }],
+        },
+      }),
+    );
+    expect(out.action).toBe("send");
+    if (out.action !== "send") return;
+    expect(out.message).toContain("nested-ready.md");
+  });
+
+  it("recursive default=false excludes subdirectories", async () => {
+    const sub = join(tmp, "sub");
+    mkdirSync(sub, { recursive: true });
+    writeFileSync(join(sub, "nested-ready.md"), "---\nstatus: ready\n---\n");
+    const out = await artifactPoolReadyPolicy.evaluate(
+      makeJob({
+        context: { pools: [{ path: tmp, include_statuses: ["ready"] }] },
       }),
     );
     expect(out).toEqual({ action: "skip", reason: "no_actionable_artifacts" });
   });
 });
 
-describe("edgeArtifactRequiredPolicy", () => {
+describe("edgeArtifactRequiredPolicy (POC contract)", () => {
   let src: string;
   let tgt: string;
   beforeEach(() => {
@@ -152,89 +257,117 @@ describe("edgeArtifactRequiredPolicy", () => {
     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");
+  // R1 fix (guard blocker 2): POC contract uses context.source / context.target.
+  it("skip when downstream raw content references source key (POC body-match)", async () => {
+    writeFileSync(
+      join(src, "x.md"),
+      "---\nentry: coordination-stream-queue-view-intake-pilot-a-vertical\nstatus: shipped\n---\n# Pilot A\n",
+    );
+    writeFileSync(
+      join(tgt, "old.md"),
+      "---\nstatus: closed\n---\n# Old item\n\nThis item mentions agent-starter-v1-vertical only.\n",
+    );
+    writeFileSync(
+      join(tgt, "new.md"),
+      "---\nstatus: ready\n---\nLifecycle edge for coordination-stream-queue-view-intake-pilot-a-vertical.\n",
+    );
     const out = await edgeArtifactRequiredPolicy.evaluate(
       makeJob({
+        target: { session: "delivery-orch-lead@openrig-velocity" },
         context: {
-          target: { session: "a@rig" },
-          source_pools: { path: src, include_statuses: ["ready"] },
-          target_pools: { path: tgt },
+          edge_label: "delivery-to-lifecycle",
+          source: { path: src, include_statuses: ["shipped"], key_field: "entry" },
+          target: { 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");
+  it("send when downstream raw does NOT reference source key (POC body-match)", async () => {
+    writeFileSync(
+      join(src, "x.md"),
+      "---\nentry: coordination-stream-queue-view-intake-pilot-a-vertical\nstatus: shipped\n---\n# Pilot A\n",
+    );
+    writeFileSync(
+      join(tgt, "old.md"),
+      "---\nstatus: closed\n---\n# Old item\n\nThis item mentions agent-starter-v1-vertical only.\n",
+    );
     const out = await edgeArtifactRequiredPolicy.evaluate(
       makeJob({
+        target: { session: "delivery-orch-lead@openrig-velocity" },
         context: {
-          target: { session: "a@rig" },
-          source_pools: { path: src, include_statuses: ["ready"] },
-          target_pools: { path: tgt },
-          label: "myedge",
+          edge_label: "delivery-to-lifecycle",
+          source: { path: src, include_statuses: ["shipped"], key_field: "entry" },
+          target: { path: tgt },
         },
       }),
     );
     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");
+    expect(out.target).toEqual({ session: "delivery-orch-lead@openrig-velocity" });
+    expect(out.message).toContain("delivery-to-lifecycle");
+    expect(out.message).toContain("coordination-stream-queue-view-intake-pilot-a-vertical");
+    expect(out.message).toContain("Producer loop owns creating the missing downstream artifact");
   });
 
   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({
+        target: { session: "a@rig" },
         context: {
-          target: { session: "a@rig" },
-          source_pools: { path: src, include_statuses: ["ready"] },
-          target_pools: { path: tgt },
+          source: { path: src, include_statuses: ["ready"] },
+          target: { path: tgt },
         },
       }),
     );
     expect(out.action).toBe("send");
     if (out.action !== "send") return;
-    expect(out.message).toContain("no-frontmatter-key.md");
+    expect(out.message).toContain("no-frontmatter-key");
   });
 
-  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 () => {
+  it("target scan ignores source's include_statuses (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");
+    // Target has matching key in body but a non-listed status; match still counts.
+    writeFileSync(join(tgt, "x-down.md"), "---\nstatus: rejected\n---\nReferences x here.\n");
     const out = await edgeArtifactRequiredPolicy.evaluate(
       makeJob({
+        target: { session: "a@rig" },
         context: {
-          target: { session: "a@rig" },
-          source_pools: { path: src, include_statuses: ["ready"] },
-          target_pools: { path: tgt, include_statuses: ["accepted"] },
+          source: { path: src, include_statuses: ["ready"] },
+          target: { path: tgt, include_statuses: ["accepted"] },
         },
       }),
     );
     expect(out).toEqual({ action: "skip", reason: "no_missing_edge_artifacts" });
   });
+
+  it("throws policy_spec_invalid when context.source is missing", async () => {
+    try {
+      await edgeArtifactRequiredPolicy.evaluate(
+        makeJob({
+          target: { session: "a@rig" },
+          context: { target: { path: tgt } },
+        }),
+      );
+      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.target is missing", async () => {
+    try {
+      await edgeArtifactRequiredPolicy.evaluate(
+        makeJob({
+          target: { session: "a@rig" },
+          context: { source: { path: src } },
+        }),
+      );
+      throw new Error("should have thrown");
+    } catch (err) {
+      expect((err as Error & { code: string }).code).toBe("policy_spec_invalid");
+    }
+  });
 });

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

@@ -1,4 +1,7 @@
 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 type Database from "better-sqlite3";
 import { createDb } from "../src/db/connection.js";
 import { migrate } from "../src/db/migrate.js";
@@ -15,7 +18,7 @@ import {
 } from "../src/domain/watchdog-policy-engine.js";
 import type { PersistedEvent } from "../src/domain/types.js";
 
-describe("WatchdogPolicyEngine (PL-004 Phase C)", () => {
+describe("WatchdogPolicyEngine (PL-004 Phase C R1)", () => {
   let db: Database.Database;
   let bus: EventBus;
   let jobsRepo: WatchdogJobsRepository;
@@ -24,12 +27,13 @@ describe("WatchdogPolicyEngine (PL-004 Phase C)", () => {
   let deliveryCalls: Array<{ targetSession: string; message: string }>;
   let deliver: DeliveryFn;
 
-  function makeEngine(deliverOverride?: DeliveryFn): WatchdogPolicyEngine {
+  function makeEngine(opts?: { deliver?: DeliveryFn; now?: () => Date }): WatchdogPolicyEngine {
     return new WatchdogPolicyEngine({
       jobsRepo,
       historyLog: log,
       eventBus: bus,
-      deliver: deliverOverride ?? deliver,
+      deliver: opts?.deliver ?? deliver,
+      now: opts?.now,
     });
   }
 
@@ -50,61 +54,70 @@ describe("WatchdogPolicyEngine (PL-004 Phase C)", () => {
 
   afterEach(() => db.close());
 
-  it("evaluate(periodic-reminder) routes through delivery + records sent + emits evaluation_fired", async () => {
+  it("evaluate(periodic-reminder) routes through delivery + records sent + emits evaluation_fired + sets actionable=true", 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",
+      specYaml:
+        "policy: periodic-reminder\ntarget:\n  session: alice@rig\ninterval_seconds: 60\nmessage: ping\n",
       targetSession: "alice@rig",
       intervalSeconds: 60,
       registeredBySession: "ops@kernel",
     });
-    const result = await engine.evaluate(job);
-    expect(result.outcome.action).toBe("send");
+    await engine.evaluate(job);
     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();
+    expect(after.actionable).toBe(true);
+    expect(after.lastActionableAt).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");
+  it("evaluate(quiet skip — no_actionable_artifacts) does NOT record history nor emit event (POC parity)", async () => {
+    const engine = makeEngine({
+      deliver: 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();
+    const tmp = join(tmpdir(), `wd-eng-empty-${Date.now()}-${Math.random()}`);
+    mkdirSync(tmp, { recursive: true });
+    try {
+      const job = jobsRepo.register({
+        policy: "artifact-pool-ready",
+        specYaml:
+          `policy: artifact-pool-ready\ntarget:\n  session: alice@rig\ninterval_seconds: 60\ncontext:\n  pools:\n    - path: ${tmp}\n      include_statuses: [ready]\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(result.meaningful).toBe(false);
+      expect(deliveryCalls).toEqual([]);
+      expect(log.countForJob(job.jobId)).toBe(0);
+      expect(captured.some((e) => e.type === "watchdog.evaluation_skipped")).toBe(false);
+      const after = jobsRepo.getByIdOrThrow(job.jobId);
+      expect(after.lastEvaluationAt).not.toBeNull();
+      expect(after.lastFireAt).toBeNull();
+      expect(after.actionable).toBe(false);
+    } finally {
+      rmSync(tmp, { recursive: true, force: true });
+    }
   });
 
-  it("delivery failure is recorded as sent outcome with delivery_status=failed", async () => {
+  it("delivery failure recorded as sent with delivery_status=failed (still meaningful)", async () => {
     const failingDeliver: DeliveryFn = async () => ({ status: "failed", error: "transport denied" });
-    const engine = makeEngine(failingDeliver);
+    const engine = makeEngine({ deliver: 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",
+        "policy: periodic-reminder\ntarget:\n  session: alice@rig\ninterval_seconds: 60\nmessage: ping\n",
       targetSession: "alice@rig",
       intervalSeconds: 60,
       registeredBySession: "ops@kernel",
@@ -115,17 +128,16 @@ describe("WatchdogPolicyEngine (PL-004 Phase C)", () => {
     expect(list[0]?.deliveryStatus).toBe("failed");
   });
 
-  it("unknown policy at evaluate-time marks job terminal + emits evaluation_terminal", async () => {
+  it("unknown policy at evaluate-time marks job terminal + records + 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",
+      specYaml:
+        "policy: periodic-reminder\ntarget:\n  session: a@rig\nmessage: 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");
@@ -134,7 +146,7 @@ describe("WatchdogPolicyEngine (PL-004 Phase C)", () => {
     expect(captured.some((e) => e.type === "watchdog.evaluation_terminal")).toBe(true);
   });
 
-  it("policy registry resolves the three v1 policies", () => {
+  it("policy registry resolves the three v1 policies (workflow-keepalive absent)", () => {
     const engine = makeEngine();
     expect(engine.resolvePolicy("periodic-reminder")?.name).toBe("periodic-reminder");
     expect(engine.resolvePolicy("artifact-pool-ready")?.name).toBe("artifact-pool-ready");
@@ -142,13 +154,13 @@ describe("WatchdogPolicyEngine (PL-004 Phase C)", () => {
     expect(engine.resolvePolicy("workflow-keepalive")).toBeUndefined();
   });
 
-  it("default YAML parser extracts nested context block", async () => {
+  it("default spec parser extracts top-level target + context + message", 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",
+        "policy: periodic-reminder\ntarget:\n  session: parsed-session@rig\nmessage: parsed-message\ninterval_seconds: 60\n",
+      targetSession: "registered-fallback@rig",
       intervalSeconds: 60,
       registeredBySession: "ops@kernel",
     });
@@ -157,4 +169,130 @@ describe("WatchdogPolicyEngine (PL-004 Phase C)", () => {
       { targetSession: "parsed-session@rig", message: "parsed-message" },
     ]);
   });
+
+  it("falls back to registered targetSession when spec lacks top-level target", async () => {
+    const engine = makeEngine();
+    const job = jobsRepo.register({
+      policy: "periodic-reminder",
+      // No top-level target. Engine should synthesize {session: targetSession}.
+      specYaml: "policy: periodic-reminder\nmessage: ping\ninterval_seconds: 60\n",
+      targetSession: "fallback@rig",
+      intervalSeconds: 60,
+      registeredBySession: "ops@kernel",
+    });
+    await engine.evaluate(job);
+    expect(deliveryCalls).toEqual([{ targetSession: "fallback@rig", message: "ping" }]);
+  });
+
+  // R1 fix (guard blocker 1): port the POC active-wake regression.
+  // POC source: tests/watchdog-active-wake-interval.test.sh
+  it("active-wake throttle: re-delivery suppressed during wake window; fires when window elapses", async () => {
+    const tmp = join(tmpdir(), `wd-actwake-${Date.now()}-${Math.random()}`);
+    mkdirSync(tmp, { recursive: true });
+    try {
+      const job = jobsRepo.register({
+        policy: "artifact-pool-ready",
+        specYaml:
+          `policy: artifact-pool-ready\ntarget:\n  session: loop-head@example\ninterval_seconds: 30\nscan_interval_seconds: 30\nactive_wake_interval_seconds: 600\ncontext:\n  pools:\n    - path: ${tmp}\n      include_statuses: [ready]\n`,
+        targetSession: "loop-head@example",
+        intervalSeconds: 30,
+        scanIntervalSeconds: 30,
+        activeWakeIntervalSeconds: 600,
+        registeredBySession: "ops@kernel",
+      });
+
+      // Tick 1: empty pool → quiet skip(no_actionable_artifacts), actionable=false.
+      let nowMs = 0;
+      let engine = makeEngine({ now: () => new Date(nowMs) });
+      let r = await engine.evaluate(jobsRepo.getByIdOrThrow(job.jobId));
+      expect(r.outcome).toEqual({ action: "skip", reason: "no_actionable_artifacts" });
+      expect(deliveryCalls.length).toBe(0);
+
+      // Make pool actionable.
+      writeFileSync(join(tmp, "ready.md"), "---\nstatus: ready\n---\n");
+
+      // Tick 2: newly actionable → fires (no throttle on transition).
+      nowMs = 30_000;
+      engine = makeEngine({ now: () => new Date(nowMs) });
+      r = await engine.evaluate(jobsRepo.getByIdOrThrow(job.jobId));
+      expect(r.outcome.action).toBe("send");
+      expect(deliveryCalls.length).toBe(1);
+      const afterFire = jobsRepo.getByIdOrThrow(job.jobId);
+      expect(afterFire.actionable).toBe(true);
+      expect(afterFire.lastFireAt).toBe("1970-01-01T00:00:30.000Z");
+
+      // Tick 3 at +30s after fire (60s mark): pool still actionable but
+      // wake window (600s) not elapsed → quiet skip(active_wake_not_due).
+      nowMs = 60_000;
+      engine = makeEngine({ now: () => new Date(nowMs) });
+      r = await engine.evaluate(jobsRepo.getByIdOrThrow(job.jobId));
+      expect(r.outcome).toEqual({ action: "skip", reason: "active_wake_not_due" });
+      expect(deliveryCalls.length).toBe(1);
+      // last_fire_at preserved.
+      expect(jobsRepo.getByIdOrThrow(job.jobId).lastFireAt).toBe("1970-01-01T00:00:30.000Z");
+      // actionable still true.
+      expect(jobsRepo.getByIdOrThrow(job.jobId).actionable).toBe(true);
+
+      // Tick 4 at 630_000ms (>= last_fire + 600_000ms): wake window
+      // elapsed → fires again.
+      nowMs = 630_000;
+      engine = makeEngine({ now: () => new Date(nowMs) });
+      r = await engine.evaluate(jobsRepo.getByIdOrThrow(job.jobId));
+      expect(r.outcome.action).toBe("send");
+      expect(deliveryCalls.length).toBe(2);
+      const after2ndFire = jobsRepo.getByIdOrThrow(job.jobId);
+      expect(after2ndFire.lastFireAt).toBe("1970-01-01T00:10:30.000Z");
+
+      // Tick 5: pool empties → quiet skip resets actionable.
+      rmSync(join(tmp, "ready.md"));
+      nowMs = 660_000;
+      engine = makeEngine({ now: () => new Date(nowMs) });
+      r = await engine.evaluate(jobsRepo.getByIdOrThrow(job.jobId));
+      expect(r.outcome).toEqual({ action: "skip", reason: "no_actionable_artifacts" });
+      expect(jobsRepo.getByIdOrThrow(job.jobId).actionable).toBe(false);
+
+      // Tick 6: pool actionable again → fires immediately (newly actionable).
+      writeFileSync(join(tmp, "ready.md"), "---\nstatus: ready\n---\n");
+      nowMs = 690_000;
+      engine = makeEngine({ now: () => new Date(nowMs) });
+      r = await engine.evaluate(jobsRepo.getByIdOrThrow(job.jobId));
+      expect(r.outcome.action).toBe("send");
+      expect(deliveryCalls.length).toBe(3);
+
+      // History reflects only the 3 meaningful events.
+      expect(log.countForJob(job.jobId)).toBe(3);
+      const entries = log.listForJob(job.jobId);
+      // All sent (no quiet skip rows).
+      for (const e of entries) expect(e.outcome).toBe("sent");
+    } finally {
+      rmSync(tmp, { recursive: true, force: true });
+    }
+  });
+
+  it("active-wake throttle does NOT apply when active_wake_interval_seconds is null (every fire goes through)", async () => {
+    const tmp = join(tmpdir(), `wd-actwake-null-${Date.now()}-${Math.random()}`);
+    mkdirSync(tmp, { recursive: true });
+    writeFileSync(join(tmp, "ready.md"), "---\nstatus: ready\n---\n");
+    try {
+      const job = jobsRepo.register({
+        policy: "artifact-pool-ready",
+        specYaml:
+          `policy: artifact-pool-ready\ntarget:\n  session: x@rig\ninterval_seconds: 30\ncontext:\n  pools:\n    - path: ${tmp}\n      include_statuses: [ready]\n`,
+        targetSession: "x@rig",
+        intervalSeconds: 30,
+        registeredBySession: "ops@kernel",
+      });
+      let nowMs = 0;
+      let engine = makeEngine({ now: () => new Date(nowMs) });
+      await engine.evaluate(jobsRepo.getByIdOrThrow(job.jobId));
+      expect(deliveryCalls.length).toBe(1);
+      nowMs = 1000;
+      engine = makeEngine({ now: () => new Date(nowMs) });
+      await engine.evaluate(jobsRepo.getByIdOrThrow(job.jobId));
+      // Without wake throttle, every send fires.
+      expect(deliveryCalls.length).toBe(2);
+    } finally {
+      rmSync(tmp, { recursive: true, force: true });
+    }
+  });
 });

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

@@ -153,16 +153,17 @@ describe("WatchdogScheduler (PL-004 Phase C)", () => {
   });
 
   it("policy evaluation errors are caught and tick continues for siblings", async () => {
+    // Job 1 has no message anywhere — periodic-reminder throws policy_spec_invalid.
     jobsRepo.register({
       policy: "periodic-reminder",
-      specYaml: "context:\n  message: only-message-no-target\n",
+      specYaml: "policy: periodic-reminder\ntarget:\n  session: a@rig\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",
+      specYaml: "policy: periodic-reminder\ntarget:\n  session: ok@rig\nmessage: ok\n",
       targetSession: "ok@rig",
       intervalSeconds: 30,
       registeredBySession: "ops@kernel",