diff --git a/.opencode/plugins/fm-primary-watch-arm.js b/.opencode/plugins/fm-primary-watch-arm.js index b50436b951d..d9a11fd6328 100644 --- a/.opencode/plugins/fm-primary-watch-arm.js +++ b/.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(); }, }; diff --git a/docs/configuration.md b/docs/configuration.md index 63695826c71..718cb5f1053 100644 --- a/docs/configuration.md +++ b/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 diff --git a/docs/watcher-continuity.md b/docs/watcher-continuity.md index f14717596b4..a927bddddb3 100644 --- a/docs/watcher-continuity.md +++ b/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. diff --git a/tests/fm-opencode-primary-watch-arm.test.sh b/tests/fm-opencode-primary-watch-arm.test.sh new file mode 100755 index 00000000000..d0faaff9729 --- /dev/null +++ b/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