fix(substrate): review-feedback correctness pass — catch widening, init flag, host requirement, write-back guard, idempotency key

Addresses 7 inline review findings (Copilot + secret-mars + arc0btc) on
PR #5 — all substantive items resolved in-tree, no follow-ups left.

**Catch widening (Copilot #4, secret-mars [blocking-risk], arc0btc #1)**

`runSubstrateIntakeTick` previously only caught `claimNextJob`. Both
`jobRowToTaskInput` and `enqueueTask` ran outside the try, so a malformed
JobRow or a local SQLite lock contention escaped as a thrown error from
`runOnce`. Each call site now has its own catch with a distinct skip
reason — `job-parse-fail` and `local-enqueue-fail` — so the contract's
"NEVER throws" guarantee is real. The substrate job stays under lease in
both failure modes; `releaseExpiredLeases` reconciles on the next cycle.

**`substrateDbInitialized` retry on credential fail (Copilot #1, all 3
reviewers seconded)**

The flag was set to `true` BEFORE `createSubstrateConnection` succeeded.
A transient credential read miss at first tick (e.g. encrypted blob not
yet available) permanently disabled substrate intake for the process
lifetime — the contract's documented `[substrate] skip
reason=credential-fail` log line would never re-fire. Flag now sets only
inside the success branch; next tick retries.

**`runSubstrateWriteBack` swallows transient PG blips (arc0btc #3)**

Was the only substrate fn whose Postgres calls were unguarded — a
mid-write-back connection reset would propagate out of `runOnce` even
though the local task already finished cleanly. Wrapped both
`completeJob` and `failJob` in a single try; new
`[substrate] write-back error=<msg> jobs.id=<id>` log line on transient
fail. Lease recovery still reconciles eventually.

**Default host removed (Copilot #3, arc0btc seconded)**

`host` no longer defaults to a hard-coded private IP. When
`substrate.enabled: true`, an explicit `substrate.host` is required —
`createSubstrateConnection` throws a clear error if unset. Closes the
"dev/test slot misconfigured at enabled:true accidentally connects to
prod" footgun.

**Self-contained log fallbacks (secret-mars #5)**

The empty-else branches on `completeJob`/`failJob` `{ ok: false }`
results relied on the substrate-db package emitting its own
`[substrate] complete-epoch-mismatch ...` log. If that package's log
format ever changes, those branches went silent. Added
`[substrate] complete-failed jobs.id=<id> epoch=<n>` and
`[substrate] fail-failed ...` fallbacks so this code is self-contained.

**Idempotency-key threading (arc0btc #2 design, secret-mars idempotency
follow-up)**

Closes the "side-effecting jobs can execute twice when a lease expires
mid-flight" hazard structurally — the fence on `jobs.claim_epoch` only
protects the status write, not the action. `jobRowToTaskInput` now
threads `payload.idempotency_key = "substrate-<job_id>-e<claim_epoch>"`
so downstream side-effecting handlers (email send, PR open, tx
broadcast) can dedup against their own per-handler key store. Bounds the
*consequence* from "action runs twice" to "action runs once, second
attempt no-ops."

Substrate tasks also enqueue with `priority: 1` so they execute on the
same or next tick — the lease window now tracks real execution latency
instead of waiting behind lower-priority work, which shrinks the "lease
expires while task waits in local queue" window further.

**Silent null claim (Copilot #5)**

Contract said null-claim is silent; impl logged `[substrate] claim ...
result=none` every tick. Dropped the log — quiet-tick visibility lives
in successful-claim and idle-dispatch event lines, not substrate
tick-rate noise.

**Other nits (Copilot #2, arc0btc #5)**

- Removed unused `getTaskById` import in `substrate.ts`.
- Documented `substrate` block's shallow-merge behavior in `types.ts`
  (unlike `profiles`/`adapters` which deep-merge). Slots that extend a
  base and want to flip only `isLeaseRecoveryOwner: true` must repeat
  the whole substrate block.

Deferred (already declared out of scope by reviewers):
- `pg_advisory_lock` guard on `isLeaseRecoveryOwner` race (secret-mars
  [suggestion]) — multi-slot mis-config protection is a follow-up.
- `_substrate_*` payload-injection vs strict validators (secret-mars
  [suggestion]) — naming is defensive prefixing already.

Contract block in src/substrate.ts updated with all new log lines and a
new "Side-effect duplicate-execution guard" section.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: whoabuddy

src/runtime.ts

@@ -210,15 +210,18 @@ export async function runOnce(db: Database, config: RuntimeConfig): Promise<Reco
     // ---------------------------------------------------------------------------
     const subConfig = resolveSubstrateConfig(config);
     if (subConfig) {
-      // Lazy-init the substrate DB connection (credential resolved once per process).
+      // Lazy-init the substrate DB connection (credential resolved on success;
+      // re-attempted next tick on failure so a transient credential / host
+      // read miss does not permanently disable substrate intake).
       if (!substrateDbInitialized) {
-        substrateDbInitialized = true;
         try {
           substrateDb = await createSubstrateConnection(subConfig);
+          substrateDbInitialized = true; // only mark initialized on success
         } catch (error) {
           const msg = error instanceof Error ? error.message : String(error);
           console.error(`[substrate] skip reason=credential-fail error=${msg}`);
           substrateDb = null;
+          // substrateDbInitialized stays false — next tick retries.
         }
       }
 

src/substrate.ts

@@ -5,18 +5,36 @@
 // Zero behavior change on any slot where substrate is not configured.
 //
 // Contract (stable, parseable log lines):
-//   [substrate] claim slot=<id> kinds=<list> result=<jobs.id|none>
+//   [substrate] claim slot=<id> kinds=<list> result=<jobs.id>     (success — null claim is silent)
 //   [substrate] complete jobs.id=<id> epoch=<n>
 //   [substrate] fail jobs.id=<id> epoch=<n> reason=<...>
 //   [substrate] lease-recovery released=<n>
 //   [substrate] skip reason=credential-fail error=<...>
 //   [substrate] skip reason=db-unreachable error=<...>
+//   [substrate] skip reason=job-parse-fail error=<...> jobs.id=<id>
+//   [substrate] skip reason=local-enqueue-fail error=<...> jobs.id=<id>
 //   [substrate] complete-epoch-mismatch jobs.id=<id> expected=<n> actual=<m> — ...
+//   [substrate] complete-failed jobs.id=<id> epoch=<n>            (self-contained fallback)
+//   [substrate] fail-failed jobs.id=<id> epoch=<n>                (self-contained fallback)
+//   [substrate] write-back error=<...> jobs.id=<id>               (transient PG blip; swallowed)
 //
 // Failure conditions:
 //   - Substrate unreachable → tick logs [substrate] skip and continues (NOT crash)
-//   - Credential resolution fail → tick logs [substrate] skip with distinct line
-//   - claimNextJob returns null → no-op tick (silent)
+//   - Credential resolution fail → tick logs [substrate] skip with distinct line; flag NOT
+//     marked initialized so the next tick retries.
+//   - claimNextJob returns null → no-op tick (silent — quiet-tick visibility goes through
+//     successful-claim and idle-dispatch event lines, not substrate-specific log spam)
+//   - jobRowToTaskInput / enqueueTask throw → caught with distinct reason; substrate job
+//     stays held under lease and gets returned by releaseExpiredLeases.
+//   - completeJob / failJob throw mid-write-back → caught; lease recovery reconciles.
+//
+// Side-effect duplicate-execution guard:
+//   Substrate `jobs.id` + `claim_epoch` are threaded into the local TaskInput as both
+//   `payload._substrate_*` fields (for write-back fencing) AND a top-level
+//   `payload.idempotency_key = "substrate-<job_id>-e<claim_epoch>"` (for downstream
+//   side-effecting handlers to dedup against). Substrate tasks also enqueue with
+//   `priority: 1` to maximize same/next-tick execution and minimize the window
+//   where a lease can expire on a queued-but-unrun task.
 // ---------------------------------------------------------------------------
 
 import type { Database } from "bun:sqlite";
@@ -30,7 +48,7 @@ import {
 } from "@genesis-works/substrate-db";
 import type { RuntimeConfig, SubstrateConfig, TaskInput, CanonicalOutcome, TaskRecord } from "./types";
 import { resolveCredentialRefs } from "./credentials";
-import { enqueueTask, getTaskById } from "./db";
+import { enqueueTask } from "./db";
 
 // ---------------------------------------------------------------------------
 // Types
@@ -73,6 +91,15 @@ export function resolveSubstrateConfig(config: RuntimeConfig): SubstrateConfig |
 export async function createSubstrateConnection(
   sub: SubstrateConfig
 ): Promise<SubstrateDb> {
+  // Host is REQUIRED when substrate is enabled — no implicit default to any private IP.
+  // A misconfigured slot with substrate.enabled=true but no host would otherwise quietly
+  // connect to whatever sits at the prior default; explicit host config is the safer floor.
+  if (!sub.host || sub.host.trim().length === 0) {
+    throw new Error(
+      'Substrate enabled but substrate.host is not set — explicit host required (no implicit default).'
+    );
+  }
+
   // Resolve the credential — the env key SUBSTRATE_DB_CREDENTIAL holds the id.
   // We use resolveCredentialRefs on a fake env object to reuse the existing pattern.
   const fakeEnv: Record<string, string> = {
@@ -85,7 +112,7 @@ export async function createSubstrateConnection(
   }
 
   return createSubstrateClient({
-    host: sub.host ?? "192.168.1.31",
+    host: sub.host,
     port: sub.port ?? 5432,
     database: sub.database ?? "substrate",
     user: sub.user ?? "substrate_app",
@@ -101,6 +128,12 @@ export async function createSubstrateConnection(
 
 export function jobRowToTaskInput(job: JobRow): TaskInput {
   const p = (job.payload ?? {}) as Record<string, unknown>;
+  // Side-effect duplicate guard: every substrate-sourced task carries a stable
+  // idempotency_key derived from (job_id, claim_epoch). Downstream handlers that
+  // perform external side-effects (email send, PR open, tx broadcast) check this
+  // key in their own dedup store before acting — bounds duplicate execution from
+  // lease-expiry-mid-flight scenarios to a no-op on the second handler call.
+  const idempotencyKey = `substrate-${job.id}-e${job.claim_epoch}`;
   return {
     kind: job.kind,
     source: `substrate:${job.id}`,        // ties runtime task back to the substrate job
@@ -110,9 +143,13 @@ export function jobRowToTaskInput(job: JobRow): TaskInput {
       ...p,
       _substrate_job_id: job.id,           // load-bearing: write-back uses this
       _substrate_claim_epoch: job.claim_epoch, // load-bearing: epoch fencing
+      idempotency_key: idempotencyKey,     // dedup key for side-effecting handlers
     },
     requested_profile: typeof p.requested_profile === "string" ? p.requested_profile : undefined,
     requested_adapter: typeof p.requested_adapter === "string" ? p.requested_adapter : undefined,
+    // Priority 1 maximizes same/next-tick execution so the lease window roughly
+    // tracks real execution latency instead of waiting behind lower-priority work.
+    priority: 1,
     max_attempts: 1, // substrate handles retry via releaseExpiredLeases
   };
 }
@@ -128,7 +165,13 @@ export function jobRowToTaskInput(job: JobRow): TaskInput {
 // Returns { claimed: false } if substrate is disabled, unreachable, or no jobs.
 // Returns { claimed: true, jobId, taskId, epochUsed } on successful enqueue.
 //
-// NEVER throws — all errors are logged and result in { claimed: false }.
+// NEVER throws — every code path inside is wrapped. Distinct skip reasons:
+//   db-unreachable      — claimNextJob threw (Postgres connectivity)
+//   job-parse-fail      — jobRowToTaskInput threw on a malformed JobRow
+//   local-enqueue-fail  — enqueueTask threw (sqlite lock, validator, etc.)
+// In the latter two, the substrate `jobs` row stays held under lease and is
+// returned by releaseExpiredLeases — another slot (or this slot on a later
+// tick) re-claims with a fresh epoch.
 // ---------------------------------------------------------------------------
 
 export async function runSubstrateIntakeTick(
@@ -148,17 +191,37 @@ export async function runSubstrateIntakeTick(
     return { claimed: false, reason: "db-unreachable" };
   }
 
-  const kindList = sub.kinds.join(",");
+  // Silent no-op on null claim (per contract). Successful claims and idle dispatch
+  // are already logged elsewhere; substrate-specific result=none lines would just
+  // be tick-rate noise on a queue that's frequently empty.
   if (!job) {
-    console.info(`[substrate] claim slot=${sub.slotId} kinds=${kindList} result=none`);
     return { claimed: false };
   }
 
+  const kindList = sub.kinds.join(",");
   console.info(`[substrate] claim slot=${sub.slotId} kinds=${kindList} result=${job.id}`);
 
-  // Convert to TaskInput and enqueue locally
-  const taskInput = jobRowToTaskInput(job);
-  const task = enqueueTask(localDb, config, taskInput);
+  // Convert to TaskInput and enqueue locally — both can throw on malformed payload
+  // or local DB pressure. A throw here leaves the substrate job under lease so it
+  // gets re-claimed by another slot via releaseExpiredLeases. We DO NOT release the
+  // lease eagerly here: doing so would race with the lease-recovery owner.
+  let taskInput: TaskInput;
+  try {
+    taskInput = jobRowToTaskInput(job);
+  } catch (error) {
+    const msg = error instanceof Error ? error.message : String(error);
+    console.error(`[substrate] skip reason=job-parse-fail error=${msg} jobs.id=${job.id}`);
+    return { claimed: false, reason: "job-parse-fail" };
+  }
+
+  let task;
+  try {
+    task = enqueueTask(localDb, config, taskInput);
+  } catch (error) {
+    const msg = error instanceof Error ? error.message : String(error);
+    console.error(`[substrate] skip reason=local-enqueue-fail error=${msg} jobs.id=${job.id}`);
+    return { claimed: false, reason: "local-enqueue-fail" };
+  }
 
   return {
     claimed: true,
@@ -200,33 +263,45 @@ export async function runSubstrateWriteBack(
 
   const isSuccess = outcome.status === "completed";
 
-  if (isSuccess) {
-    const result = await completeJob(
-      substrateDb,
-      jobId,
-      {
-        task_id: task.task_id,
-        status: "completed",
-        operator_summary: outcome.operator_summary,
-        artifact_paths: outcome.artifact_paths ?? [],
-        machine_status: outcome.machine_status,
-      },
-      undefined,
-      claimEpoch
-    );
-    if (result.ok) {
-      console.info(`[substrate] complete jobs.id=${jobId} epoch=${claimEpoch ?? "none"}`);
-    } else {
-      // conflict is already logged by completeJob with the [substrate] prefix
-    }
-  } else {
-    const reason = outcome.operator_summary.slice(0, 500);
-    const result = await failJob(substrateDb, jobId, reason, claimEpoch);
-    if (result.ok) {
-      console.info(`[substrate] fail jobs.id=${jobId} epoch=${claimEpoch ?? "none"} reason=${reason.slice(0, 100)}`);
+  // The whole substrate write-back is wrapped: a transient PG blip mid-write
+  // must NOT propagate out of runOnce and turn a clean local-task completion
+  // into an errored tick. The substrate `jobs` row stays held under lease and
+  // releaseExpiredLeases reconciles on the next cycle.
+  try {
+    if (isSuccess) {
+      const result = await completeJob(
+        substrateDb,
+        jobId,
+        {
+          task_id: task.task_id,
+          status: "completed",
+          operator_summary: outcome.operator_summary,
+          artifact_paths: outcome.artifact_paths ?? [],
+          machine_status: outcome.machine_status,
+        },
+        undefined,
+        claimEpoch
+      );
+      if (result.ok) {
+        console.info(`[substrate] complete jobs.id=${jobId} epoch=${claimEpoch ?? "none"}`);
+      } else {
+        // Self-contained fallback log: don't assume the substrate-db package's
+        // log format stays stable across versions. Mismatch / already-done
+        // both land here; the package may also log its own line.
+        console.warn(`[substrate] complete-failed jobs.id=${jobId} epoch=${claimEpoch ?? "none"}`);
+      }
     } else {
-      // conflict is already logged by failJob
+      const reason = outcome.operator_summary.slice(0, 500);
+      const result = await failJob(substrateDb, jobId, reason, claimEpoch);
+      if (result.ok) {
+        console.info(`[substrate] fail jobs.id=${jobId} epoch=${claimEpoch ?? "none"} reason=${reason.slice(0, 100)}`);
+      } else {
+        console.warn(`[substrate] fail-failed jobs.id=${jobId} epoch=${claimEpoch ?? "none"}`);
+      }
     }
+  } catch (error) {
+    const msg = error instanceof Error ? error.message : String(error);
+    console.error(`[substrate] write-back error=${msg} jobs.id=${jobId}`);
   }
 }
 

src/types.ts

@@ -53,6 +53,10 @@ export type RuntimeConfig = {
   profiles: Record<string, string>;
   adapters: Record<string, AdapterConfig>;
   // Opt-in substrate dispatch intake (Phase 5 — disabled by default).
+  // Note: this block is shallow-overridden by mergeRuntimeConfig (unlike profiles /
+  // adapters which deep-merge). Slots that extend a base and want to set only
+  // `isLeaseRecoveryOwner: true` must repeat the whole substrate block in their
+  // host config; a deep-merge here is a follow-up if that pattern becomes common.
   substrate?: SubstrateConfig;
 };