fix: preserve OpenCode watcher continuity failure diagnostics

.opencode/plugins/fm-primary-watch-arm.js

@@ -1,5 +1,6 @@
 import { spawn, spawnSync } from "node:child_process";
-import { existsSync, readFileSync, readdirSync, realpathSync } from "node:fs";
+import { existsSync, readFileSync, readdirSync, realpathSync, statSync, unlinkSync, writeFileSync } from "node:fs";
+import { createHash } from "node:crypto";
 import { resolve } from "node:path";
 import { encodeFirstmateOperationalInput } from "./lib/fm-operational-input.js";
 import { eventMatchesRoot } from "./lib/fm-event-location.js";
@@ -28,6 +29,7 @@ const ARM_RETIRE_TIMEOUT_MS = positiveInteger("FM_WATCH_ARM_RETIRE_TIMEOUT_MS",
 const REARM_RETRY_BASE_MS = positiveInteger("FM_WATCH_REARM_RETRY_BASE_MS", 250);
 const REARM_RETRY_MAX_MS = positiveInteger("FM_WATCH_REARM_RETRY_MAX_MS", 4000);
 const REARM_RETRY_LIMIT = positiveInteger("FM_WATCH_REARM_RETRY_LIMIT", 5);
+const DIAGNOSTIC_DEDUPE_MS = positiveInteger("FM_WATCH_DIAGNOSTIC_DEDUPE_MS", 300000);
 
 let child = null;
 let armStatus = "idle";
@@ -83,6 +85,76 @@ function runProcess(command, args, options = {}) {
   });
 }
 
+function markerIsStale(marker, now) {
+  try {
+    return now - statSync(marker).mtimeMs >= DIAGNOSTIC_DEDUPE_MS;
+  } catch {
+    return false;
+  }
+}
+
+function expireStaleMarker(marker) {
+  if (!markerIsStale(marker, Date.now())) return false;
+  try {
+    unlinkSync(marker);
+  } catch {
+    return false;
+  }
+  return true;
+}
+
+function sweepStaleMarkers(state) {
+  let names;
+  try {
+    names = readdirSync(state);
+  } catch {
+    return;
+  }
+  const now = Date.now();
+  for (const name of names) {
+    if (!name.startsWith(".opencode-watch-diagnostic-") || !markerIsStale(`${state}/${name}`, now)) continue;
+    try {
+      unlinkSync(`${state}/${name}`);
+    } catch {}
+  }
+}
+
+function publishDiagnostic(paths, detail) {
+  const message = `check: OpenCode watcher continuity failure: ${String(detail).trim() || "unknown failure"}`;
+  const digest = createHash("sha256").update(message).digest("hex").slice(0, 24);
+  const marker = `${paths.state}/.opencode-watch-diagnostic-${digest}`;
+  sweepStaleMarkers(paths.state);
+  try {
+    writeFileSync(marker, `${message}\n`, { flag: "wx" });
+  } catch (error) {
+    if (error?.code !== "EEXIST" || !expireStaleMarker(marker)) return;
+    try {
+      writeFileSync(marker, `${message}\n`, { flag: "wx" });
+    } catch {
+      return;
+    }
+  }
+  const script = [
+    'set -u',
+    'root=$1 home=$2 state=$3 key=$4 message=$5',
+    'FM_HOME="$home" FM_STATE_OVERRIDE="$state" FM_ROOT_OVERRIDE="$root"',
+    '. "$root/bin/fm-wake-lib.sh"',
+    'fm_wake_append check "$key" "$message"',
+  ].join("; ");
+  const result = spawnSync("bash", ["-c", script, "fm-opencode-watch-diagnostic", paths.root, paths.home, paths.state, `opencode-watch:${digest}`, message], {
+    cwd: paths.root,
+    encoding: "utf8",
+    env: { ...process.env, FM_HOME: paths.home, FM_STATE_OVERRIDE: paths.state, FM_ROOT_OVERRIDE: paths.root },
+  });
+  if (result.status !== 0) {
+    try {
+      unlinkSync(marker);
+    } catch {
+      // The diagnostic remains durably marked if its queue publication raced.
+    }
+  }
+}
+
 async function resolveRoot(anchor) {
   if (!anchor) return "";
   const result = await runProcess("git", ["-C", anchor, "rev-parse", "--show-toplevel"]);
@@ -307,8 +379,9 @@ function wakePrompt(reason) {
 }
 
 function surfaceFailure(paths, client, sessionID, reason) {
-  void sendPrompt(paths, client, sessionID, wakePrompt(reason)).catch(() => {
-    // OpenCode owns delivery errors; continuity restoration never waits on prompting.
+  publishDiagnostic(paths, reason);
+  void sendPrompt(paths, client, sessionID, wakePrompt(reason)).catch((error) => {
+    publishDiagnostic(paths, `prompt delivery failed: ${String(error?.message ?? error)}`);
   });
 }
 
@@ -463,6 +536,7 @@ function spawnArm(paths, sessionID, client, predecessorArmPid = "") {
       void restoration.then(async (result) => {
         try {
           const message = result.failure ? `${classification.message}\n\n${result.failure}` : classification.message;
+          if (result.failure) publishDiagnostic(paths, result.failure);
           await deliverActionableWake(paths, client, sessionID, message, result.recovery);
         } finally {
           if (restorationInFlight === restoration) restorationInFlight = null;
@@ -543,9 +617,10 @@ export default {
   id: "fm-primary-watch-arm",
   setup(ctx) {
     const controller = new AbortController();
+    let paths = effectivePaths(resolvePath(ctx.location?.directory || process.cwd()));
     void (async () => {
       const root = await resolveRoot(ctx.location.directory);
-      const paths = effectivePaths(root);
+      paths = effectivePaths(root);
       globalThis[COORDINATOR_KEY] = {
         ensureArmed: (sessionID, activeCtx) => ensureArm(paths, sessionID, activeCtx ?? ctx),
       };
@@ -555,9 +630,13 @@ export default {
         if (!eventMatchesRoot(event, root)) continue;
         const sessionID = event.data?.sessionID;
         if (!sessionID) continue;
-        void ensureArm(paths, sessionID, ctx);
+        void ensureArm(paths, sessionID, ctx).catch((error) => {
+          surfaceFailure(paths, ctx, sessionID, `watcher: FAILED - OpenCode watcher arm failed: ${String(error?.message ?? error)}`);
+        });
       }
-    })().catch(() => {});
+    })().catch((error) => {
+      publishDiagnostic(paths, `OpenCode watcher plugin initialization failed: ${String(error?.message ?? error)}`);
+    });
     return () => controller.abort();
   },
 };

docs/configuration.md

@@ -2378,6 +2378,7 @@ FM_WATCH_ARM_RETIRE_TIMEOUT_MS=1000   # milliseconds Pi/OpenCode wait for an unr
 FM_WATCH_REARM_RETRY_BASE_MS=250   # Pi/OpenCode adapter base delay for continuity restoration retries
 FM_WATCH_REARM_RETRY_MAX_MS=4000   # Pi/OpenCode adapter cap for exponential continuity retry delay
 FM_WATCH_REARM_RETRY_LIMIT=5   # Pi/OpenCode adapter launch-failure retries before surfacing restoration failure
+FM_WATCH_DIAGNOSTIC_DEDUPE_MS=300000   # milliseconds the OpenCode watcher plugin suppresses republication of an identical continuity-failure diagnostic; the marker is swept once this window expires, so a recurring failure surfaces again
 FM_WATCH_CYCLE_LOG_MAX_BYTES=262144   # size cap for the arm-owned watcher lifecycle ledger
 FM_WATCH_CYCLE_LOG_KEEP_LINES=1000   # newest complete lifecycle rows considered when the ledger is capped
 FM_WATCH_EXTENSION_LOG_KEEP_LINES=0   # opt-in Pi extension diagnostic log (state/.watch-extension.log); unset, empty, non-numeric, zero, or negative disables logging, a positive value keeps that many newest rows; logging never changes supervision behavior

docs/watcher-continuity.md

@@ -149,6 +149,9 @@ The Pi extension diagnostic log is opt-in and off by default: only a positive FM
 docs/configuration.md owns the knob's default and accepted values.
 A failed confirmation is never swallowed.
 
+OpenCode additionally preserves the diagnostic itself, not only the typed wake: an arm error, an undeliverable prompt, an exhausted restoration, or a plugin initialization failure each appends one durable `check: OpenCode watcher continuity failure` queue row carrying the exact detail, so the outcome reaches the drain even when the prompt path drops it.
+Identical diagnostics publish exactly one row per window - [`configuration.md`](configuration.md) owns the window's knob and default - after which the stale marker is swept and a recurring failure surfaces again.
+
 ### Readiness timeout and retry
 
 The adapter waits at most one readiness timeout per attempt.
@@ -435,6 +438,8 @@ The same suite covers ordinary same-process session replacement for `/new`, `/re
 - A failed confirmation for a stale successor that spares a newer arm started by a repair.
 - A repair, a scheduled retry, and a deferred close over a dead-but-unclosed arm child that each start a fresh arm instead of stalling.
 
+`tests/fm-opencode-primary-watch-arm.test.sh` pins the OpenCode durable-diagnostic contract against the real plugin: a prompt that always rejects still leaves one durable row for the failed wake and one for the delivery failure, a refused event stream leaves one initialization-failure row, an unrestorable successor surfaces its retry detail durably, and an identical diagnostic is republished only after its dedupe marker expires while the next publication sweeps other stale markers.
+
 The guard and session-start suites prove that active generation evidence tolerates a fresh-beacon handoff.
 They also prove that a legacy or handoff-phase watcher marker from an absent replacement extension still raises the outage diagnostic.
 

tests/fm-opencode-primary-watch-arm.test.sh

@@ -0,0 +1,201 @@
+#!/usr/bin/env bash
+set -u
+
+# shellcheck source=tests/lib.sh
+. "$(dirname "${BASH_SOURCE[0]}")/lib.sh"
+
+PLUGIN="$ROOT/.opencode/plugins/fm-primary-watch-arm.js"
+TMP_ROOT=$(fm_test_tmproot fm-opencode-primary-watch-arm)
+
+make_fixture() {
+  local repo="$TMP_ROOT/$1" home="$TMP_ROOT/$1-home"
+  mkdir -p "$repo/bin" "$home/state" "$home/config"
+  git init -q "$repo"
+  : > "$repo/AGENTS.md"
+  cp "$ROOT/bin/fm-wake-lib.sh" "$ROOT/bin/fm-path-lib.sh" "$repo/bin/"
+  : > "$home/state/task.meta"
+  printf '%s\t%s\n' "$repo" "$home"
+}
+
+test_prompt_failure_publishes_durable_diagnostics() {
+  local fixture repo home log stop out status
+  fixture=$(make_fixture prompt-failure)
+  repo=${fixture%%$'\t'*}
+  home=${fixture#*$'\t'}
+  log="$TMP_ROOT/prompt-failure.log"
+  stop="$TMP_ROOT/prompt-failure.stop"
+  cat > "$repo/bin/fm-watch-arm.sh" <<'SH'
+#!/usr/bin/env bash
+if [ "${1:-}" = --handling-delivered ]; then exit 0; fi
+count=$(test -f "$FM_ARM_LOG" && wc -l < "$FM_ARM_LOG" || printf 0)
+printf 'arm=%s\n' "$$" >> "$FM_ARM_LOG"
+printf 'watcher: started pid=%s (beacon fresh)\n' "$$"
+if [ "$count" -eq 0 ]; then
+  printf 'signal: terminal outcome\n'
+  exit 0
+fi
+trap 'exit 0' TERM INT
+while [ ! -e "$FM_STOP_FILE" ]; do sleep 0.02; done
+SH
+  chmod +x "$repo/bin/fm-watch-arm.sh"
+  out=$(PLUGIN="$PLUGIN" WORKTREE="$repo" FM_HOME="$home" FM_ARM_LOG="$log" FM_STOP_FILE="$stop" FM_WATCH_REARM_RETRY_BASE_MS=5 FM_WATCH_REARM_RETRY_MAX_MS=10 node 2>&1 <<'EOF'
+import { existsSync, readFileSync, writeFileSync } from "node:fs";
+import { pathToFileURL } from "node:url";
+
+const mod = await import(pathToFileURL(process.env.PLUGIN).href);
+const client = { session: { promptAsync: async () => { throw new Error("prompt unavailable"); } } };
+writeFileSync(`${process.env.FM_HOME}/state/.lock`, `${process.pid}\n`);
+const hooks = mod.default.setup({
+  location: { directory: process.env.WORKTREE },
+  event: { subscribe: async function* () { yield { type: "session.idle", data: { sessionID: "session-test" } }; await new Promise(() => {}); } },
+});
+const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
+const queuePath = `${process.env.FM_HOME}/state/.wake-queue`;
+const countOccurrences = (text, needle) => text.split(needle).length - 1;
+for (let i = 0; i < 500 && !existsSync(`${process.env.FM_ARM_LOG}`); i += 1) await sleep(10);
+let queue = "";
+for (let i = 0; i < 500; i += 1) {
+  if (existsSync(queuePath)) {
+    queue = readFileSync(queuePath, "utf8");
+    if (queue.includes("prompt delivery failed")) break;
+  }
+  await sleep(10);
+}
+if (!queue.includes("OpenCode watcher continuity failure")) throw new Error(`missing durable prompt failure: ${queue}`);
+if (!queue.includes("could not deliver an actionable wake")) throw new Error(`missing wake failure detail: ${queue}`);
+if (!queue.includes("prompt delivery failed")) throw new Error(`missing durable delivery failure: ${queue}`);
+const rows = queue.trim().split("\n");
+if (rows.length !== 2) throw new Error(`expected each failure exactly once: ${queue}`);
+if (countOccurrences(queue, "could not deliver an actionable wake") !== 1) throw new Error(`prompt failure was repeated: ${queue}`);
+if (countOccurrences(queue, "prompt delivery failed") !== 1) throw new Error(`delivery failure was repeated: ${queue}`);
+writeFileSync(process.env.FM_STOP_FILE, "stop\n");
+hooks();
+EOF
+  )
+  status=$?
+  expect_code 0 "$status" "OpenCode prompt failures must publish durable diagnostics: $out"
+  [ -z "$out" ] || fail "OpenCode prompt failure test printed output: $out"
+  pass "OpenCode prompt failure publishes durable diagnostics"
+}
+
+test_plugin_initialization_failure_publishes_durable_diagnostic() {
+  local fixture repo home out status
+  fixture=$(make_fixture initialization-failure)
+  repo=${fixture%%$'\t'*}
+  home=${fixture#*$'\t'}
+  out=$(PLUGIN="$PLUGIN" WORKTREE="$repo" FM_HOME="$home" node 2>&1 <<'EOF'
+import { existsSync, readFileSync } from "node:fs";
+import { pathToFileURL } from "node:url";
+
+const mod = await import(pathToFileURL(process.env.PLUGIN).href);
+mod.default.setup({
+  location: { directory: process.env.WORKTREE },
+  event: { subscribe: async function* () { throw new Error("event stream unavailable"); } },
+});
+for (let i = 0; i < 500 && !existsSync(`${process.env.FM_HOME}/state/.wake-queue`); i += 1) await new Promise((resolve) => setTimeout(resolve, 10));
+const queue = readFileSync(`${process.env.FM_HOME}/state/.wake-queue`, "utf8");
+if (!queue.includes("plugin initialization failed")) throw new Error(`missing durable initialization failure: ${queue}`);
+EOF
+  )
+  status=$?
+  expect_code 0 "$status" "OpenCode plugin initialization failures must be durable: $out"
+  [ -z "$out" ] || fail "OpenCode initialization failure test printed output: $out"
+  pass "OpenCode plugin initialization failure publishes a durable diagnostic"
+}
+
+test_missing_successor_surfaces_durable_terminal_failure() {
+  local fixture repo home log out status
+  fixture=$(make_fixture missing-successor)
+  repo=${fixture%%$'\t'*}
+  home=${fixture#*$'\t'}
+  log="$TMP_ROOT/missing-successor.log"
+  cat > "$repo/bin/fm-watch-arm.sh" <<'SH'
+#!/usr/bin/env bash
+if [ "${1:-}" = --handling-delivered ]; then exit 0; fi
+count=$(test -f "$FM_ARM_LOG" && wc -l < "$FM_ARM_LOG" || printf 0)
+printf 'arm=%s\n' "$$" >> "$FM_ARM_LOG"
+if [ "$count" -eq 0 ]; then
+  printf 'watcher: started pid=%s (beacon fresh)\n' "$$"
+  printf 'signal: terminal outcome\n'
+else
+  printf 'watcher: FAILED - successor unavailable\n' >&2
+  exit 1
+fi
+SH
+  chmod +x "$repo/bin/fm-watch-arm.sh"
+  out=$(PLUGIN="$PLUGIN" WORKTREE="$repo" FM_HOME="$home" FM_ARM_LOG="$log" FM_WATCH_REARM_RETRY_BASE_MS=5 FM_WATCH_REARM_RETRY_MAX_MS=10 FM_WATCH_REARM_RETRY_LIMIT=1 node 2>&1 <<'EOF'
+import { existsSync, readFileSync, writeFileSync } from "node:fs";
+import { pathToFileURL } from "node:url";
+
+const mod = await import(pathToFileURL(process.env.PLUGIN).href);
+let prompt = "";
+const client = { session: { promptAsync: async (request) => { prompt = request.body.parts[0].text; } } };
+writeFileSync(`${process.env.FM_HOME}/state/.lock`, `${process.pid}\n`);
+mod.default.setup({
+  location: { directory: process.env.WORKTREE },
+  event: { subscribe: async function* () { yield { type: "session.idle", data: { sessionID: "session-test" } }; await new Promise(() => {}); } },
+});
+for (let i = 0; i < 500 && !existsSync(`${process.env.FM_HOME}/state/.wake-queue`); i += 1) await new Promise((resolve) => setTimeout(resolve, 10));
+const queue = readFileSync(`${process.env.FM_HOME}/state/.wake-queue`, "utf8");
+if (!queue.includes("could not restore watcher continuity after 1 retries")) throw new Error(`missing durable successor failure: ${queue}`);
+EOF
+  )
+  status=$?
+  expect_code 0 "$status" "OpenCode missing successors must surface a durable terminal failure: $out"
+  [ -z "$out" ] || fail "OpenCode missing successor test printed output: $out"
+  pass "OpenCode missing successor surfaces a durable terminal failure"
+}
+
+test_dedupe_marker_expiry_republishes_identical_diagnostic() {
+  local fixture repo home out status
+  fixture=$(make_fixture dedupe-expiry)
+  repo=${fixture%%$'\t'*}
+  home=${fixture#*$'\t'}
+  out=$(PLUGIN="$PLUGIN" WORKTREE="$repo" FM_HOME="$home" node 2>&1 <<'EOF'
+import { existsSync, readFileSync, readdirSync, utimesSync, writeFileSync } from "node:fs";
+import { pathToFileURL } from "node:url";
+
+const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
+const mod = await import(pathToFileURL(process.env.PLUGIN).href);
+const setupPlugin = () => mod.default.setup({
+  location: { directory: process.env.WORKTREE },
+  event: { subscribe: async function* () { throw new Error("event stream unavailable"); } },
+});
+const stateDir = `${process.env.FM_HOME}/state`;
+const queuePath = `${stateDir}/.wake-queue`;
+const readRows = () => (existsSync(queuePath) ? readFileSync(queuePath, "utf8").trim().split("\n").filter(Boolean) : []);
+const readMarkers = () => readdirSync(stateDir).filter((name) => name.startsWith(".opencode-watch-diagnostic-"));
+setupPlugin();
+for (let i = 0; i < 500 && readRows().length === 0; i += 1) await sleep(10);
+if (!readRows()[0]?.includes("plugin initialization failed")) throw new Error("missing first durable initialization failure");
+setupPlugin();
+await sleep(1500);
+if (readRows().length !== 1) throw new Error(`identical failure inside the dedupe window was repeated: ${readRows().join("\n")}`);
+const markers = readMarkers();
+if (markers.length !== 1) throw new Error(`expected one dedupe marker: ${markers.join(", ")}`);
+const stale = new Date(Date.now() - 31 * 24 * 60 * 60 * 1000);
+utimesSync(`${stateDir}/${markers[0]}`, stale, stale);
+writeFileSync(`${stateDir}/.opencode-watch-diagnostic-staleleftover`, "stale\n");
+utimesSync(`${stateDir}/.opencode-watch-diagnostic-staleleftover`, stale, stale);
+setupPlugin();
+for (let i = 0; i < 500 && readRows().length < 2; i += 1) await sleep(10);
+const rows = readRows();
+if (rows.length !== 2 || !rows[1].includes("plugin initialization failed")) {
+  throw new Error(`expired marker did not republish the recurring failure: ${rows.join("\n")}`);
+}
+const remaining = readMarkers();
+if (remaining.length !== 1 || remaining.includes(".opencode-watch-diagnostic-staleleftover")) {
+  throw new Error(`stale dedupe markers were not swept: ${remaining.join(", ")}`);
+}
+EOF
+  )
+  status=$?
+  expect_code 0 "$status" "OpenCode diagnostic dedupe markers must expire: $out"
+  [ -z "$out" ] || fail "OpenCode dedupe expiry test printed output: $out"
+  pass "OpenCode dedupe marker expiry republishes an identical diagnostic"
+}
+
+test_prompt_failure_publishes_durable_diagnostics
+test_plugin_initialization_failure_publishes_durable_diagnostic
+test_missing_successor_surfaces_durable_terminal_failure
+test_dedupe_marker_expiry_republishes_identical_diagnostic