From 44479851700cfc2b79eacf17126e0fe8da45734b Mon Sep 17 00:00:00 2001 From: Pranay Pratyush Date: Mon, 31 Aug 2026 12:26:28 +0530 Subject: [PATCH 01/11] feat(omp): batch primary wake continuations --- .agents/skills/harness-adapters/SKILL.md | 2 +- .omp/extensions/fm-primary-omp.ts | 8 +- bin/fm-primary-watch-core.ts | 24 ++- docs/supervision-protocols/omp.md | 2 +- docs/verification/runtime-backends.md | 2 +- docs/watcher-continuity.md | 6 +- tests/fm-omp-primary.test.sh | 229 +++++++++++++++++++++-- 7 files changed, 251 insertions(+), 22 deletions(-) diff --git a/.agents/skills/harness-adapters/SKILL.md b/.agents/skills/harness-adapters/SKILL.md index 62ae4f482f2..6d6e959da11 100644 --- a/.agents/skills/harness-adapters/SKILL.md +++ b/.agents/skills/harness-adapters/SKILL.md @@ -336,7 +336,7 @@ Before killing and respawning a live OMP agent for a wedged PROVIDER STREAM, fol **Primary-session integration fact (verified 2026-07-31, OMP 17.1.8).** Plain OMP started from the Firstmate root discovers `.omp/extensions/fm-primary-omp.ts` natively, including in a fresh checkout before canonical `state/` exists; `omp -e .omp/extensions/fm-primary-omp.ts` remains the explicit recovery fallback. The adapter publishes the OMP marker shape owned by [configuration](../../../docs/configuration.md#harness-support), delivers the session-start instruction on native `session_start` and `session_switch` events, and owns watcher generations through `fm_watch_arm_omp` plus `/new` and `/resume` continuity. -It routes watcher follow-ups through OMP's `sendUserMessage(content)` with no explicit `deliverAs`, so OMP starts a turn when idle and steers while streaming instead of queueing the notification; it runs the shared turn-end predicate through native `session_stop`, and applies the shared watcher-arm, persistent-directory, and delegation-shaped tool safety checks before tool execution. +Primary watcher notification delivery and durable wake acknowledgement boundaries are owned by [`docs/watcher-continuity.md`](../../../docs/watcher-continuity.md); the adapter also runs the shared turn-end predicate through native `session_stop` and applies the shared watcher-arm, persistent-directory, and delegation-shaped tool safety checks before tool execution. `bin/fm-session-start.sh` rejects a missing, stale, foreign-PID, or version-mismatched loaded marker and prints both native-discovery and explicit `-e` recovery commands. The authoritative operating procedure is `docs/supervision-protocols/omp.md`. diff --git a/.omp/extensions/fm-primary-omp.ts b/.omp/extensions/fm-primary-omp.ts index ec01a0301b8..f27da05a6b5 100644 --- a/.omp/extensions/fm-primary-omp.ts +++ b/.omp/extensions/fm-primary-omp.ts @@ -216,8 +216,11 @@ export default function (omp: ExtensionAPI) { armReadyTimeoutEnv: "FM_OMP_ARM_READY_TIMEOUT_MS", repairToolName: "fm_watch_arm_omp", encodeOperationalInput, + coalesceWakeNotification: true, sendFollowUp: async (content) => { - // Deliver a custom steer so OMP wakes idle sessions without touching the editable draft. + // Queue one hidden continuation after prompt unwinding without touching + // the editable draft. The shared core coalesces concurrent closes until + // before_agent_start observes this next turn. omp.sendMessage( { customType: "firstmate-watcher-wake", @@ -226,7 +229,7 @@ export default function (omp: ExtensionAPI) { attribution: "agent", details: { kind: "watcher", runtime: "omp" }, }, - { deliverAs: "steer", triggerTurn: true }, + { deliverAs: "nextTurn", triggerTurn: true }, ); }, offerWakeToBranch, @@ -253,6 +256,7 @@ export default function (omp: ExtensionAPI) { }); omp.on("before_agent_start", (): BeforeAgentStartEventResult | undefined => { + watch.notificationTurnStarted(); if (!pendingStartupNudge) return undefined; const content = pendingStartupNudge; pendingStartupNudge = ""; diff --git a/bin/fm-primary-watch-core.ts b/bin/fm-primary-watch-core.ts index 64b48694ace..2294ba8f183 100644 --- a/bin/fm-primary-watch-core.ts +++ b/bin/fm-primary-watch-core.ts @@ -57,6 +57,7 @@ type SessionGeneration = { retryTimer: ReturnType | null; retryFailures: number; restoring: boolean; + notificationPending: boolean; seq: number; }; @@ -78,6 +79,11 @@ export type PrimaryWatchCoreOptions = { repairToolName: string; encodeOperationalInput: (kind: "watcher", content: string) => string; sendFollowUp: (content: string) => Promise; + // OMP's hidden next-turn transport can retain one pending continuation while + // the current prompt unwinds. Its adapter resets this latch when the next + // agent turn begins, so concurrent actionable closes coalesce without + // changing the durable wake queue or its acknowledgement ownership. + coalesceWakeNotification?: boolean; // Optional supervision-branch dispatch handshake. When supplied (the OMP // adapter with its branch extension loaded), the core offers each ordinary // actionable wake to the branch before delivering it to main; a synchronous @@ -93,6 +99,7 @@ export type PrimaryWatchCore = { arm: () => ArmResult; armAndWait: () => Promise; markLoaded: () => void; + notificationTurnStarted: () => void; sessionShutdown: () => void; sessionStart: () => void; }; @@ -165,6 +172,7 @@ function createGeneration(): SessionGeneration { retryTimer: null, retryFailures: 0, restoring: false, + notificationPending: false, seq: 0, }; } @@ -213,6 +221,7 @@ export function createPrimaryWatchCore(options: PrimaryWatchCoreOptions): Primar repairToolName, encodeOperationalInput, sendFollowUp, + coalesceWakeNotification = false, offerWakeToBranch, } = options; const armScript = `${fmRoot}/bin/fm-watch-arm.sh`; @@ -338,12 +347,18 @@ export function createPrimaryWatchCore(options: PrimaryWatchCoreOptions): Primar } async function sendWake(owner: SessionGeneration, message: string): Promise { - if (!generationIsLive(owner)) return; + if (!generationIsLive(owner) || (coalesceWakeNotification && owner.notificationPending)) return; const content = encodeOperationalInput( "watcher", `FIRSTMATE WATCHER WAKE: ${message}\n\nRun bin/fm-wake-drain.sh first and handle the queued wake. Watcher continuity is extension-owned.`, ); - await sendFollowUp(content); + if (coalesceWakeNotification) owner.notificationPending = true; + try { + await sendFollowUp(content); + } catch (error) { + if (coalesceWakeNotification) owner.notificationPending = false; + throw error; + } } function confirmHandlingDelivery(recovery: RecoveryHandoff): { ok: boolean; detail: string } { @@ -695,6 +710,10 @@ export function createPrimaryWatchCore(options: PrimaryWatchCoreOptions): Primar }; } + function notificationTurnStarted(): void { + if (coalesceWakeNotification) generation.notificationPending = false; + } + function sessionStart(): void { if (activeBinding !== binding) return; if (generation.stopping) generation = createGeneration(); @@ -716,6 +735,7 @@ export function createPrimaryWatchCore(options: PrimaryWatchCoreOptions): Primar arm: () => startArm(generation), armAndWait, markLoaded, + notificationTurnStarted, sessionShutdown, sessionStart, }; diff --git a/docs/supervision-protocols/omp.md b/docs/supervision-protocols/omp.md index 05bb9929dba..e372f6a14af 100644 --- a/docs/supervision-protocols/omp.md +++ b/docs/supervision-protocols/omp.md @@ -12,7 +12,7 @@ When this session owns supervision and away mode is not active: 6. The extension starts `bin/fm-watch-arm.sh --restart`, keeps the child attached to the live OMP process, and owns every later successor launch. The tool and the fallback command return only after that child reports readiness, so a `watcher: FAILED` readiness timeout is a real failure to handle under step 11 rather than a slow success. 7. OMP `/new` and `/resume` events inject the session-start instruction exactly once for the new conversation, replace the prior extension generation, and restore the watcher without a foreground watcher command. -8. After an actionable child close, the shared watcher core rechecks session-lock ownership and verifies one successor before it delivers the follow-up notification. +8. After an actionable child close, the shared watcher core rechecks session-lock ownership, verifies one successor, and confirms the recovery handoff before it schedules one hidden `nextTurn` notification with `triggerTurn`. 9. Ordinary work, turn completion, and ordinary notification handling must not call `fm_watch_arm_omp` again because continuity is extension-owned. 10. An unexpected child close enters bounded exponential retry, and an exhausted retry or lost session lock is surfaced as a watcher failure. 11. Missing, failed, or unhealthy cycle only: drain queued notifications, inspect the failure, call `fm_watch_arm_omp`, and restart with the explicit `-e` fallback if the integration is missing or stale. diff --git a/docs/verification/runtime-backends.md b/docs/verification/runtime-backends.md index b283c6ad97f..001881db281 100644 --- a/docs/verification/runtime-backends.md +++ b/docs/verification/runtime-backends.md @@ -616,7 +616,7 @@ ok - a captain-worthy wake opens exactly one follow-up turn on MAIN (real SDK) ok - OMP supervision branch live guard passed against @oh-my-pi/pi-coding-agent 17.3.4 ``` -The guard proves a broken branch (an unresolvable model pin) falls the wake back to main through the primary adapter's watcher-wake steer with triggerTurn, not sendUserMessage, leaving the wake queue durable. +The guard proves a broken branch (an unresolvable model pin) falls the wake back to main while leaving the wake queue durable. It proves a resident second AgentSession is created and remains re-promptable on a later wake without its turn output reaching main or replacing main's terminal resume breadcrumb, and that a routine verdict opens no new main turn while a captain verdict opens exactly one follow-up turn. The captain sub-check is skipped, not passed, on a run where the model judges the captain-worthy fixture routine. diff --git a/docs/watcher-continuity.md b/docs/watcher-continuity.md index 290a93e255f..e0b1b7dc019 100644 --- a/docs/watcher-continuity.md +++ b/docs/watcher-continuity.md @@ -25,9 +25,9 @@ While supervision is still needed and away mode remains inactive, an actionable ## Actionable wake ordering After an actionable Pi, OMP, or OpenCode child close, the adapter starts and verifies one singleton successor before it delivers the original wake. -OMP delivers that follow-up through its host API as a hidden custom `steer` with `triggerTurn`, which starts an idle handling turn without touching an editable TUI draft. -It confirms the handling handoff against that successor before scheduling the follow-up, retries once against the current generation and successor, and treats a failed confirmation as a restoration failure: it classifies the error, retires a successor that is no longer alive, and surfaces exactly one typed message. -A failed confirmation is never swallowed. +For OMP, the shared core confirms the recovery handoff before the adapter schedules at most one hidden custom `nextTurn` notification with `triggerTurn`; OMP consumes it on the next agent turn, including after prompt unwinding, without touching an editable TUI draft. +Notification delivery does not acknowledge durable wake rows; only the exact generation-bound `WAKE_ACK_REQUIRED` command printed by `bin/fm-wake-drain.sh` may retire them. +A failed confirmation is never swallowed: the core retries once against the current generation and successor, then classifies the failure, retires a successor that is no longer alive, and surfaces exactly one typed message. It waits at most one readiness timeout per attempt, then sends TERM and waits a bounded retirement confirmation before the next lock-verified exponential retry. If the unready arm does not retire within that bound, the adapter keeps ownership, starts no overlapping retry, and delivers the typed fallback immediately. When that retained arm later closes, its actual close is classified as a new supervised event without replaying the earlier fallback. diff --git a/tests/fm-omp-primary.test.sh b/tests/fm-omp-primary.test.sh index 35b744c7650..5f641c91cbd 100755 --- a/tests/fm-omp-primary.test.sh +++ b/tests/fm-omp-primary.test.sh @@ -593,10 +593,11 @@ if (watcherMessages.length !== 1 || !watcherMessages[0].message.content.includes } if ( watcherMessages[0].message.customType !== "firstmate-watcher-wake" || - watcherMessages[0].options?.deliverAs !== "steer" || + watcherMessages[0].message.display !== false || + watcherMessages[0].options?.deliverAs !== "nextTurn" || watcherMessages[0].options?.triggerTurn !== true ) { - throw new Error(`OMP watcher notification did not preserve the editable draft delivery mode: ${JSON.stringify(watcherMessages[0])}`); + throw new Error(`OMP watcher notification was not a hidden next-turn continuation: ${JSON.stringify(watcherMessages[0])}`); } if (!existsSync(`${process.env.FM_STATE_OVERRIDE}/watch-successor-ready`)) { throw new Error("OMP actionable notification arrived before successor readiness"); @@ -735,9 +736,9 @@ JS # The shared core delivers the recovery handshake for every runtime bound to it, # so OMP must confirm a handling delivery exactly like Pi and OpenCode do: start # and verify the successor, run fm-watch-arm.sh --handling-delivered for the -# generation the successor reported, and only then deliver the wake steer. -# Upstream covers Pi and OpenCode; this pins the fork's OMP binding of the same -# contract so a future adapter change cannot silently drop it. +# generation the successor reported, and only then schedule the hidden next-turn +# notification. Upstream covers Pi and OpenCode; this pins the fork's OMP binding +# of the same contract so a future adapter change cannot silently drop it. test_native_omp_confirms_recovery_handling_delivery() { local fixture out status=0 fixture="$TMP_ROOT/native-handling-delivery" @@ -814,9 +815,9 @@ for (let i = 0; i < 400 && !armRows().some((row) => row.startsWith("confirmed ") const rows = armRows(); const arms = rows.filter((row) => row.startsWith("arm=")); if (arms.length !== 2) throw new Error(`expected one successor arm, got ${arms.length}: ${rows.join(" | ")}`); -if (steers !== 1) throw new Error(`expected exactly one wake steer, got ${steers}`); -if (deliveryOptions?.deliverAs !== "steer" || deliveryOptions?.triggerTurn !== true) { - throw new Error(`wake was not delivered as a turn-triggering steer: ${JSON.stringify(deliveryOptions)}`); +if (steers !== 1) throw new Error(`expected exactly one wake notification, got ${steers}`); +if (deliveryOptions?.deliverAs !== "nextTurn" || deliveryOptions?.triggerTurn !== true) { + throw new Error(`wake was not delivered as a turn-triggering next-turn continuation: ${JSON.stringify(deliveryOptions)}`); } if (rowsAtDelivery !== 2) throw new Error(`wake delivery began before successor establishment (${rowsAtDelivery} arm rows)`); const confirmations = rows.filter((row) => row.startsWith("confirmed ")); @@ -835,8 +836,8 @@ JS ) || status=$? printf 'stop\n' > "$TMP_ROOT/native-handling-delivery.stop" 2>/dev/null || true expect_code 0 "$status" "OMP recovery handling delivery" - assert_contains "$out" omp-handling-delivery-ok "OMP did not confirm its recovery handling delivery after the wake steer" - pass "OMP confirms the recovery handling handshake after delivering its wake steer" + assert_contains "$out" omp-handling-delivery-ok "OMP did not confirm recovery handling before its next-turn notification" + pass "OMP confirms the recovery handling handshake before its hidden next-turn notification" } # A refused handling handshake must be classified and surfaced exactly once @@ -920,7 +921,7 @@ if (!steer.includes("FIRSTMATE WATCHER WAKE")) throw new Error(`missing follow-u if (!steer.includes("handling delivery confirmation was rejected")) { throw new Error(`refused handshake was swallowed: ${steer}`); } -if (steers !== 1) throw new Error(`refused handshake was not a single typed steer, got ${steers}`); +if (steers !== 1) throw new Error(`refused handshake was not a single typed notification, got ${steers}`); const refusals = armRows().filter((row) => row.startsWith("refused ")); if (refusals.length < 1) throw new Error(`handling-delivered was never attempted: ${armRows().join(" | ")}`); console.log("omp-refused-handshake-ok"); @@ -929,7 +930,210 @@ JS printf 'stop\n' > "$TMP_ROOT/native-handling-refused.stop" 2>/dev/null || true expect_code 0 "$status" "OMP refused handling delivery" assert_contains "$out" omp-refused-handshake-ok "OMP swallowed a refused handling handshake" - pass "OMP surfaces a refused handling handshake as one typed wake" + pass "OMP surfaces a refused handling handshake as one hidden notification" +} + +# OMP's hidden next-turn queue retains custom messages while a prompt unwinds. +# This fixture drives successive durable wake batches through the native primary +# adapter and verifies the core contributes one continuation until OMP begins +# that next turn. It also keeps the durable rows and a rejected recovery +# generation intact, so neither notification delivery nor a failed handoff can +# retire or strand replayable work. +test_native_omp_coalesces_pending_next_turn_notifications() { + local fixture out status=0 + fixture="$TMP_ROOT/native-next-turn-batching" + mkdir -p "$fixture/.omp/extensions" "$fixture/bin" "$fixture/config" "$fixture/state" + : > "$fixture/AGENTS.md" + git init -q -b main "$fixture" + cp "$ROOT/.omp/extensions/fm-primary-omp.ts" "$fixture/.omp/extensions/fm-primary-omp.ts" + mkdir -p "$fixture/.omp/extensions/lib" + cp "$ROOT/.omp/extensions/lib/fm-branch-dispatch.ts" "$fixture/.omp/extensions/lib/fm-branch-dispatch.ts" + cp "$ROOT/.omp/extensions/lib/fm-task-inbox-doorbell.ts" "$fixture/.omp/extensions/lib/fm-task-inbox-doorbell.ts" + cp "$ROOT/bin/fm-primary-watch-core.ts" "$fixture/bin/fm-primary-watch-core.ts" + cp "$ROOT/bin/fm-primary-scope-lib.sh" "$fixture/bin/fm-primary-scope-lib.sh" + cp "$ROOT/bin/fm-gate-refuse-lib.sh" "$fixture/bin/fm-gate-refuse-lib.sh" + cp "$ROOT/bin/fm-operational-input.sh" "$fixture/bin/fm-operational-input.sh" + cp "$ROOT/bin/fm-sessionstart-nudge.sh" "$fixture/bin/fm-sessionstart-nudge.sh" + cp "$ROOT/bin/fm-pi-compatible-runtimes" "$fixture/bin/fm-pi-compatible-runtimes" + cat > "$fixture/bin/fm-watch-arm.sh" <<'SH' +#!/usr/bin/env bash +set -u +state=${FM_STATE_OVERRIDE:?} +if [ "${1:-}" = --handling-delivered ]; then + printf 'handling generation=%s watcher=%s\n' "$2" "$4" >> "${FM_ARM_LOG:?}" + if [ -e "$state/reject-confirmation" ]; then + echo "watcher: recovery generation mismatch" >&2 + exit 1 + fi + exit 0 +fi +count=$(cat "$state/arm-count" 2>/dev/null || printf 0) +count=$((count + 1)) +printf '%s\n' "$count" > "$state/arm-count" +printf 'arm=%s count=%s\n' "$$" "$count" >> "${FM_ARM_LOG:?}" +printf 'watcher: started pid=%s (beacon fresh) recovery-generation=fixture-generation\n' "$$" +trap 'exit 0' TERM INT +while [ ! -e "$state/watch-trigger-$count" ]; do sleep 0.01; done +rm -f "$state/watch-trigger-$count" +printf 'signal: synthetic durable batch %s\n' "$count" +SH + cat > "$fixture/bin/fm-sessionstart-nudge.sh" <<'SH' +#!/usr/bin/env bash +exit 0 +SH + chmod +x "$fixture/bin/"*.sh + + out=$(EXTENSION="$fixture/.omp/extensions/fm-primary-omp.ts" \ + FM_HOME="$fixture" FM_ROOT_OVERRIDE="$fixture" FM_STATE_OVERRIDE="$fixture/state" \ + FM_ARM_LOG="$TMP_ROOT/native-next-turn-batching.log" \ + FM_WAKE_LIB="$ROOT/bin/fm-wake-lib.sh" \ + node --input-type=module 2>&1 <<'JS' +import { spawnSync } from "node:child_process"; +import { existsSync, readFileSync, writeFileSync } from "node:fs"; +import { pathToFileURL } from "node:url"; + +const handlers = new Map(); +const notifications = []; +let tool = null; +let sendAttempts = 0; +let failNextNotification = false; +const state = process.env.FM_STATE_OVERRIDE; +const armRows = () => (existsSync(process.env.FM_ARM_LOG) + ? readFileSync(process.env.FM_ARM_LOG, "utf8").trim().split("\n") + : []); +const armCount = () => armRows().filter((row) => row.startsWith("arm=")).length; +const waitFor = async (predicate, description) => { + for (let attempt = 0; attempt < 400; attempt += 1) { + if (predicate()) return; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + throw new Error(`timed out waiting for ${description}`); +}; +const api = { + zod: { object: () => ({}) }, + on(name, handler) { handlers.set(name, handler); }, + registerCommand() {}, + registerTool(candidate) { + if (candidate.name === "fm_watch_arm_omp") tool = candidate; + }, + sendMessage(message, options) { + sendAttempts += 1; + if (failNextNotification) { + failNextNotification = false; + throw new Error("synthetic next-turn submission failure"); + } + notifications.push({ message, options }); + }, +}; +writeFileSync(`${state}/.lock`, `${process.pid}\n`); +process.argv[1] = process.env.EXTENSION; +const extension = await import(`${pathToFileURL(process.env.EXTENSION).href}?batching=${Date.now()}`); +extension.default(api); +if (!tool) throw new Error("OMP did not register its watcher arm tool"); + +const queue = `${state}/.wake-queue`; +const queueRows = [ + "1\t1\tsignal\tworker.status\tsignal: worker first", + "2\t2\tsignal\tworker.status\tsignal: worker latest", + "3\t3\tstale\tworker\tstale: worker", +].join("\n") + "\n"; +writeFileSync(queue, queueRows); +const deduped = spawnSync( + "bash", + ["-c", '. "$1"; fm_wake_print_deduped "$2"', "bash", process.env.FM_WAKE_LIB, queue], + { encoding: "utf8", env: { ...process.env, FM_STATE_OVERRIDE: state } }, +); +if (deduped.status !== 0 || deduped.stdout.trim().split("\n").length !== 2) { + throw new Error(`durable batch was not deduplicated: ${deduped.stderr || deduped.stdout}`); +} + +const startNextTurn = async () => { + await handlers.get("before_agent_start")({ type: "before_agent_start" }, {}); +}; +const trigger = async (count, messageCount) => { + writeFileSync(`${state}/watch-trigger-${count}`, "go\n"); + await waitFor(() => armCount() >= count + 1, `successor ${count + 1}`); + if (messageCount !== undefined) { + await waitFor(() => notifications.length >= messageCount, `notification ${messageCount}`); + } +}; + +try { + await tool.execute(); + await waitFor(() => armCount() === 1, "initial watcher"); + await trigger(1, 1); + if (readFileSync(queue, "utf8") !== queueRows) { + throw new Error("hidden notification acknowledged durable wake rows"); + } + const first = notifications[0]; + if ( + first?.message?.customType !== "firstmate-watcher-wake" || + first.message?.display !== false || + first.options?.deliverAs !== "nextTurn" || + first.options?.triggerTurn !== true + ) { + throw new Error(`first notification was not hidden next-turn delivery: ${JSON.stringify(first)}`); + } + + await trigger(2); + await new Promise((resolve) => setTimeout(resolve, 40)); + if (notifications.length !== 1) { + throw new Error(`prompt-unwind activity created ${notifications.length} pending notifications`); + } + + await startNextTurn(); + await trigger(3, 2); + if (notifications.length !== 2) { + throw new Error("next OMP turn did not release the pending notification latch"); + } + + await startNextTurn(); + failNextNotification = true; + await trigger(4, 3); + if (sendAttempts !== 4 || !notifications[2].message.content.includes("could not deliver an actionable wake")) { + throw new Error(`failed next-turn submission did not surface one replayable failure: ${JSON.stringify(notifications)}`); + } + if (readFileSync(queue, "utf8") !== queueRows) { + throw new Error("failed next-turn submission retired durable wake rows"); + } + + await startNextTurn(); + await trigger(5, 4); + if (!notifications[3].message.content.includes("signal: synthetic durable batch 5")) { + throw new Error("durable wake did not replay after next-turn submission failure"); + } + + await startNextTurn(); + writeFileSync(`${state}/reject-confirmation`, "reject\n"); + await trigger(6, 5); + const mismatch = notifications[4]; + if (!mismatch.message.content.includes("recovery generation mismatch")) { + throw new Error(`generation-mismatched handoff was not surfaced: ${JSON.stringify(mismatch)}`); + } + if (readFileSync(queue, "utf8") !== queueRows) { + throw new Error("generation-mismatched handoff retired durable wake rows"); + } + if (!notifications.every((notification) => + notification.options?.deliverAs === "nextTurn" && + notification.options?.triggerTurn === true && + notification.message?.display === false, + )) { + throw new Error(`notification mode drifted across recovery: ${JSON.stringify(notifications)}`); + } + const handling = armRows().filter((row) => row.startsWith("handling ")); + if (!handling.every((row) => row.includes("generation=fixture-generation"))) { + throw new Error(`handling confirmation lost its recovery generation: ${handling.join(" | ")}`); + } + console.log("omp-next-turn-batching-ok"); +} finally { + writeFileSync(`${state}/watch-trigger-${armCount()}`, "stop\n"); + await handlers.get("session_shutdown")({ type: "session_shutdown" }, {}); +} +JS + ) || status=$? + expect_code 0 "$status" "OMP next-turn durable wake batching" + assert_contains "$out" omp-next-turn-batching-ok "OMP did not coalesce hidden next-turn wake notifications" + pass "OMP batches durable wake rows into replayable hidden next-turn notifications" } test_resolve_path_uses_node_when_readlink_f_is_unavailable @@ -943,3 +1147,4 @@ test_primary_marker_refuses_whitespace_identity test_native_primary_extension_contract test_native_omp_confirms_recovery_handling_delivery test_native_omp_refused_handling_delivery_is_typed_once +test_native_omp_coalesces_pending_next_turn_notifications From e7c6cc67b15520752b50961ccadf6b77fe69b10c Mon Sep 17 00:00:00 2001 From: Pranay Pratyush Date: Mon, 31 Aug 2026 12:39:35 +0530 Subject: [PATCH 02/11] fix(omp): persist primary wake claims --- .omp/extensions/fm-primary-omp.ts | 139 ++++++++++++++++++++++++++---- bin/fm-primary-watch-core.ts | 17 ++-- docs/watcher-continuity.md | 1 + tests/fm-omp-primary.test.sh | 37 +++++--- 4 files changed, 164 insertions(+), 30 deletions(-) diff --git a/.omp/extensions/fm-primary-omp.ts b/.omp/extensions/fm-primary-omp.ts index f27da05a6b5..2558ea6e5b7 100644 --- a/.omp/extensions/fm-primary-omp.ts +++ b/.omp/extensions/fm-primary-omp.ts @@ -1,7 +1,17 @@ // Firstmate primary integration for OMP. // OMP-native session, stop, tool-call, and shutdown events stay in this adapter. import { spawn, spawnSync } from "node:child_process"; -import { renameSync, unlinkSync, writeFileSync } from "node:fs"; +import { createHash, randomUUID } from "node:crypto"; +import { + closeSync, + lstatSync, + mkdirSync, + openSync, + readFileSync, + renameSync, + unlinkSync, + writeFileSync, +} from "node:fs"; import { dirname, isAbsolute, resolve } from "node:path"; import { fileURLToPath } from "node:url"; import type { @@ -27,12 +37,19 @@ const state = process.env.FM_STATE_OVERRIDE || `${fmHome}/state`; const config = process.env.FM_CONFIG_OVERRIDE || `${fmHome}/config`; const marker = `${state}/.omp-primary-extension-loaded`; const operationalInputScript = `${fmRoot}/bin/fm-operational-input.sh`; +const notificationClaim = `${state}/.omp-primary-nextturn-notification`; type ProcessResult = { code: number; stderr: string; }; +type WakeNotificationClaim = { + pid: string; + session: string; + key: string; +}; + function encodeOperationalInput(kind: "session-start" | "watcher" | "turn-end-guard", content: string): string { const result = spawnSync(operationalInputScript, ["encode", kind], { encoding: "utf8", @@ -179,12 +196,94 @@ function runGuard(event: SessionStopEvent): Promise { }); } +function readWakeNotificationClaim(): WakeNotificationClaim | undefined { + let content = ""; + try { + const stats = lstatSync(notificationClaim); + if (!stats.isFile() || stats.isSymbolicLink()) return undefined; + content = readFileSync(notificationClaim, "utf8"); + } catch { + return undefined; + } + const lines = content.split("\n"); + if ( + lines.length !== 5 || + lines[0] !== "fm-omp-primary-nextturn-notification-v1" || + !/^[0-9]+$/u.test(lines[1]) || + !/^[a-f0-9]{64}$/u.test(lines[2]) || + !/^[A-Za-z0-9._-]+$/u.test(lines[3]) || + lines[4] !== "" + ) { + return undefined; + } + return { pid: lines[1], session: lines[2], key: lines[3] }; +} + +function writeWakeNotificationClaim(session: string, key: string): void { + mkdirSync(state, { recursive: true }); + const temporary = `${notificationClaim}.tmp.${process.pid}.${randomUUID()}`; + let descriptor = -1; + try { + descriptor = openSync(temporary, "wx", 0o600); + writeFileSync( + descriptor, + `fm-omp-primary-nextturn-notification-v1\n${process.pid}\n${session}\n${key}\n`, + "utf8", + ); + closeSync(descriptor); + descriptor = -1; + renameSync(temporary, notificationClaim); + } catch (error) { + if (descriptor >= 0) { + try { + closeSync(descriptor); + } catch { + // Preserve the publication error; the descriptor may already be closed. + } + } + try { + unlinkSync(temporary); + } catch { + // The temporary file may not have been created. + } + throw error; + } +} + export default function (omp: ExtensionAPI) { if (!primaryIntegrationApplies()) return; publishNativeProcessIdentity(); const taskInboxDoorbell = installTaskInboxDoorbell(omp); let pendingStartupNudge = ""; + let notificationSession = createHash("sha256").update("unknown").digest("hex"); + + const setNotificationSession = (ctx: ExtensionContext): void => { + const sessionFile = ctx.sessionManager.getSessionFile() || "unknown"; + notificationSession = createHash("sha256").update(sessionFile).digest("hex"); + }; + const releaseWakeNotification = (): void => { + const claim = readWakeNotificationClaim(); + if (!claim || claim.pid !== String(process.pid)) return; + try { + unlinkSync(notificationClaim); + } catch { + // A later durable wake reclaims any claim that this turn could not retire. + } + }; + const claimWakeNotification = (key: string): boolean => { + const current = readWakeNotificationClaim(); + if ( + current?.pid === String(process.pid) && + current.session === notificationSession && + current.key === key + ) { + return false; + } + writeWakeNotificationClaim(notificationSession, key); + return true; + }; + // Supervision-branch dispatch handshake (docs/omp-supervision-branch.md). // Build one offer per ordinary actionable wake and emit it on the shared // event bus; a live, enabled branch extension calls accept() synchronously @@ -217,20 +316,27 @@ export default function (omp: ExtensionAPI) { repairToolName: "fm_watch_arm_omp", encodeOperationalInput, coalesceWakeNotification: true, - sendFollowUp: async (content) => { - // Queue one hidden continuation after prompt unwinding without touching - // the editable draft. The shared core coalesces concurrent closes until - // before_agent_start observes this next turn. - omp.sendMessage( - { - customType: "firstmate-watcher-wake", - content, - display: false, - attribution: "agent", - details: { kind: "watcher", runtime: "omp" }, - }, - { deliverAs: "nextTurn", triggerTurn: true }, - ); + releaseWakeNotification, + sendFollowUp: async (content, notificationKey) => { + // Reuse a claim from a same-session extension reload, but let a new + // session or process replay the durable batch. The claim never retires + // rows; only the drain acknowledgement owns that transition. + if (!claimWakeNotification(notificationKey)) return; + try { + omp.sendMessage( + { + customType: "firstmate-watcher-wake", + content, + display: false, + attribution: "agent", + details: { kind: "watcher", runtime: "omp" }, + }, + { deliverAs: "nextTurn", triggerTurn: true }, + ); + } catch (error) { + releaseWakeNotification(); + throw error; + } }, offerWakeToBranch, }); @@ -241,6 +347,7 @@ export default function (omp: ExtensionAPI) { }; omp.on("session_start", (_event, ctx) => { + setNotificationSession(ctx); taskInboxDoorbell.activate(); watch.sessionStart(); publishSecondmateSession(ctx); @@ -248,6 +355,8 @@ export default function (omp: ExtensionAPI) { }); omp.on("session_switch", (event, ctx) => { + releaseWakeNotification(); + setNotificationSession(ctx); watch.sessionShutdown(); watch.sessionStart(); publishSecondmateSession(ctx); diff --git a/bin/fm-primary-watch-core.ts b/bin/fm-primary-watch-core.ts index 2294ba8f183..35f14211956 100644 --- a/bin/fm-primary-watch-core.ts +++ b/bin/fm-primary-watch-core.ts @@ -78,12 +78,13 @@ export type PrimaryWatchCoreOptions = { armReadyTimeoutEnv: string; repairToolName: string; encodeOperationalInput: (kind: "watcher", content: string) => string; - sendFollowUp: (content: string) => Promise; + sendFollowUp: (content: string, notificationKey: string) => Promise; // OMP's hidden next-turn transport can retain one pending continuation while // the current prompt unwinds. Its adapter resets this latch when the next // agent turn begins, so concurrent actionable closes coalesce without // changing the durable wake queue or its acknowledgement ownership. coalesceWakeNotification?: boolean; + releaseWakeNotification?: () => void; // Optional supervision-branch dispatch handshake. When supplied (the OMP // adapter with its branch extension loaded), the core offers each ordinary // actionable wake to the branch before delivering it to main; a synchronous @@ -222,6 +223,7 @@ export function createPrimaryWatchCore(options: PrimaryWatchCoreOptions): Primar encodeOperationalInput, sendFollowUp, coalesceWakeNotification = false, + releaseWakeNotification, offerWakeToBranch, } = options; const armScript = `${fmRoot}/bin/fm-watch-arm.sh`; @@ -346,7 +348,11 @@ export function createPrimaryWatchCore(options: PrimaryWatchCoreOptions): Primar }; } - async function sendWake(owner: SessionGeneration, message: string): Promise { + async function sendWake( + owner: SessionGeneration, + message: string, + notificationKey = "wake-queue", + ): Promise { if (!generationIsLive(owner) || (coalesceWakeNotification && owner.notificationPending)) return; const content = encodeOperationalInput( "watcher", @@ -354,7 +360,7 @@ export function createPrimaryWatchCore(options: PrimaryWatchCoreOptions): Primar ); if (coalesceWakeNotification) owner.notificationPending = true; try { - await sendFollowUp(content); + await sendFollowUp(content, notificationKey); } catch (error) { if (coalesceWakeNotification) owner.notificationPending = false; throw error; @@ -419,7 +425,7 @@ export function createPrimaryWatchCore(options: PrimaryWatchCoreOptions): Primar const confirmed = confirmHandlingDeliveryWithRetry(owner, recovery); if (!confirmed.ok) { if (!pidAlive(recovery.watcherPid)) await retireArm(owner.child); - await sendWake(owner, `${message}\n\n${confirmed.detail}`); + await sendWake(owner, `${message}\n\n${confirmed.detail}`, recovery.generation); return; } } @@ -429,7 +435,7 @@ export function createPrimaryWatchCore(options: PrimaryWatchCoreOptions): Primar // repair the watcher cycle. Absent a branch, this is a no-op and every wake // goes to main exactly as before. if (!repairFailed && offerWakeToBranch?.(message)) return; - await sendWake(owner, message); + await sendWake(owner, message, recovery?.generation ?? "wake-queue"); } function surfaceFailure(owner: SessionGeneration, message: string): void { @@ -712,6 +718,7 @@ export function createPrimaryWatchCore(options: PrimaryWatchCoreOptions): Primar function notificationTurnStarted(): void { if (coalesceWakeNotification) generation.notificationPending = false; + releaseWakeNotification?.(); } function sessionStart(): void { diff --git a/docs/watcher-continuity.md b/docs/watcher-continuity.md index e0b1b7dc019..c19a94c34c6 100644 --- a/docs/watcher-continuity.md +++ b/docs/watcher-continuity.md @@ -26,6 +26,7 @@ While supervision is still needed and away mode remains inactive, an actionable After an actionable Pi, OMP, or OpenCode child close, the adapter starts and verifies one singleton successor before it delivers the original wake. For OMP, the shared core confirms the recovery handoff before the adapter schedules at most one hidden custom `nextTurn` notification with `triggerTurn`; OMP consumes it on the next agent turn, including after prompt unwinding, without touching an editable TUI draft. +The adapter preserves that pending session-and-recovery claim across an extension reload in the same live conversation, while a replacement session or process re-presents the unacknowledged durable batch instead of trusting a lost continuation. Notification delivery does not acknowledge durable wake rows; only the exact generation-bound `WAKE_ACK_REQUIRED` command printed by `bin/fm-wake-drain.sh` may retire them. A failed confirmation is never swallowed: the core retries once against the current generation and successor, then classifies the failure, retires a successor that is no longer alive, and surfaces exactly one typed message. It waits at most one readiness timeout per attempt, then sends TERM and waits a bounded retirement confirmation before the next lock-verified exponential retry. diff --git a/tests/fm-omp-primary.test.sh b/tests/fm-omp-primary.test.sh index 5f641c91cbd..12d02752dd6 100755 --- a/tests/fm-omp-primary.test.sh +++ b/tests/fm-omp-primary.test.sh @@ -1057,6 +1057,13 @@ const trigger = async (count, messageCount) => { await waitFor(() => notifications.length >= messageCount, `notification ${messageCount}`); } }; +const reloadExtension = async (label) => { + const reloaded = await import(`${pathToFileURL(process.env.EXTENSION).href}?${label}=${Date.now()}`); + reloaded.default(api); + if (!tool) throw new Error(`OMP did not register its watcher arm tool after ${label}`); + await tool.execute(); +}; + try { await tool.execute(); @@ -1075,21 +1082,31 @@ try { throw new Error(`first notification was not hidden next-turn delivery: ${JSON.stringify(first)}`); } - await trigger(2); + await reloadExtension("same-session-reload"); + await waitFor(() => armCount() >= 3, "same-session reload watcher"); + await trigger(3); await new Promise((resolve) => setTimeout(resolve, 40)); if (notifications.length !== 1) { - throw new Error(`prompt-unwind activity created ${notifications.length} pending notifications`); + throw new Error(`same-session reload duplicated a pending notification: ${JSON.stringify(notifications)}`); } - await startNextTurn(); - await trigger(3, 2); - if (notifications.length !== 2) { - throw new Error("next OMP turn did not release the pending notification latch"); + const claimPath = `${state}/.omp-primary-nextturn-notification`; + const staleClaim = readFileSync(claimPath, "utf8").split("\n"); + staleClaim[1] = "1"; + writeFileSync(claimPath, staleClaim.join("\n")); + await reloadExtension("process-restart"); + await waitFor(() => armCount() >= 5, "restart watcher"); + await trigger(5, 2); + if (readFileSync(queue, "utf8") !== queueRows) { + throw new Error("process-restart replay retired durable wake rows"); + } + if (!notifications[1].message.content.includes("signal: synthetic durable batch 5")) { + throw new Error("process restart did not replay the unacknowledged durable batch"); } await startNextTurn(); failNextNotification = true; - await trigger(4, 3); + await trigger(6, 3); if (sendAttempts !== 4 || !notifications[2].message.content.includes("could not deliver an actionable wake")) { throw new Error(`failed next-turn submission did not surface one replayable failure: ${JSON.stringify(notifications)}`); } @@ -1098,14 +1115,14 @@ try { } await startNextTurn(); - await trigger(5, 4); - if (!notifications[3].message.content.includes("signal: synthetic durable batch 5")) { + await trigger(7, 4); + if (!notifications[3].message.content.includes("signal: synthetic durable batch 7")) { throw new Error("durable wake did not replay after next-turn submission failure"); } await startNextTurn(); writeFileSync(`${state}/reject-confirmation`, "reject\n"); - await trigger(6, 5); + await trigger(8, 5); const mismatch = notifications[4]; if (!mismatch.message.content.includes("recovery generation mismatch")) { throw new Error(`generation-mismatched handoff was not surfaced: ${JSON.stringify(mismatch)}`); From 5774ece3e67ce08a6bc6b378a6e2d1d28ff91992 Mon Sep 17 00:00:00 2001 From: Pranay Pratyush Date: Mon, 31 Aug 2026 12:57:15 +0530 Subject: [PATCH 03/11] fix(omp): replay pending wake batches --- .omp/extensions/fm-primary-omp.ts | 78 +++++++++++++++++++++++-------- docs/watcher-continuity.md | 2 +- tests/fm-omp-primary.test.sh | 45 ++++++++++++------ 3 files changed, 90 insertions(+), 35 deletions(-) diff --git a/.omp/extensions/fm-primary-omp.ts b/.omp/extensions/fm-primary-omp.ts index 2558ea6e5b7..90493b95d80 100644 --- a/.omp/extensions/fm-primary-omp.ts +++ b/.omp/extensions/fm-primary-omp.ts @@ -1,6 +1,7 @@ // Firstmate primary integration for OMP. // OMP-native session, stop, tool-call, and shutdown events stay in this adapter. import { spawn, spawnSync } from "node:child_process"; +import { Buffer } from "node:buffer"; import { createHash, randomUUID } from "node:crypto"; import { closeSync, @@ -48,6 +49,7 @@ type WakeNotificationClaim = { pid: string; session: string; key: string; + content: string; }; function encodeOperationalInput(kind: "session-start" | "watcher" | "turn-end-guard", content: string): string { @@ -207,19 +209,21 @@ function readWakeNotificationClaim(): WakeNotificationClaim | undefined { } const lines = content.split("\n"); if ( - lines.length !== 5 || - lines[0] !== "fm-omp-primary-nextturn-notification-v1" || + lines.length !== 6 || + lines[0] !== "fm-omp-primary-nextturn-notification-v2" || !/^[0-9]+$/u.test(lines[1]) || !/^[a-f0-9]{64}$/u.test(lines[2]) || !/^[A-Za-z0-9._-]+$/u.test(lines[3]) || - lines[4] !== "" + lines[5] !== "" ) { return undefined; } - return { pid: lines[1], session: lines[2], key: lines[3] }; + const message = Buffer.from(lines[4], "base64").toString("utf8"); + if (Buffer.from(message, "utf8").toString("base64") !== lines[4]) return undefined; + return { pid: lines[1], session: lines[2], key: lines[3], content: message }; } -function writeWakeNotificationClaim(session: string, key: string): void { +function writeWakeNotificationClaim(claim: WakeNotificationClaim): void { mkdirSync(state, { recursive: true }); const temporary = `${notificationClaim}.tmp.${process.pid}.${randomUUID()}`; let descriptor = -1; @@ -227,7 +231,14 @@ function writeWakeNotificationClaim(session: string, key: string): void { descriptor = openSync(temporary, "wx", 0o600); writeFileSync( descriptor, - `fm-omp-primary-nextturn-notification-v1\n${process.pid}\n${session}\n${key}\n`, + [ + "fm-omp-primary-nextturn-notification-v2", + claim.pid, + claim.session, + claim.key, + Buffer.from(claim.content, "utf8").toString("base64"), + "", + ].join("\n"), "utf8", ); closeSync(descriptor); @@ -262,6 +273,18 @@ export default function (omp: ExtensionAPI) { const sessionFile = ctx.sessionManager.getSessionFile() || "unknown"; notificationSession = createHash("sha256").update(sessionFile).digest("hex"); }; + const sendWakeNotification = (content: string): void => { + omp.sendMessage( + { + customType: "firstmate-watcher-wake", + content, + display: false, + attribution: "agent", + details: { kind: "watcher", runtime: "omp" }, + }, + { deliverAs: "nextTurn", triggerTurn: true }, + ); + }; const releaseWakeNotification = (): void => { const claim = readWakeNotificationClaim(); if (!claim || claim.pid !== String(process.pid)) return; @@ -271,7 +294,7 @@ export default function (omp: ExtensionAPI) { // A later durable wake reclaims any claim that this turn could not retire. } }; - const claimWakeNotification = (key: string): boolean => { + const claimWakeNotification = (key: string, content: string): boolean => { const current = readWakeNotificationClaim(); if ( current?.pid === String(process.pid) && @@ -280,9 +303,34 @@ export default function (omp: ExtensionAPI) { ) { return false; } - writeWakeNotificationClaim(notificationSession, key); + writeWakeNotificationClaim({ + pid: String(process.pid), + session: notificationSession, + key, + content, + }); return true; }; + const replayWakeNotification = (): void => { + const pending = readWakeNotificationClaim(); + if ( + !pending || + (pending.pid === String(process.pid) && pending.session === notificationSession) + ) { + return; + } + const replay = { ...pending, pid: String(process.pid), session: notificationSession }; + try { + writeWakeNotificationClaim(replay); + sendWakeNotification(pending.content); + } catch { + try { + writeWakeNotificationClaim(pending); + } catch { + // Keep the replacement claim if restoring the former process claim also fails. + } + } + }; // Supervision-branch dispatch handshake (docs/omp-supervision-branch.md). // Build one offer per ordinary actionable wake and emit it on the shared @@ -321,18 +369,9 @@ export default function (omp: ExtensionAPI) { // Reuse a claim from a same-session extension reload, but let a new // session or process replay the durable batch. The claim never retires // rows; only the drain acknowledgement owns that transition. - if (!claimWakeNotification(notificationKey)) return; + if (!claimWakeNotification(notificationKey, content)) return; try { - omp.sendMessage( - { - customType: "firstmate-watcher-wake", - content, - display: false, - attribution: "agent", - details: { kind: "watcher", runtime: "omp" }, - }, - { deliverAs: "nextTurn", triggerTurn: true }, - ); + sendWakeNotification(content); } catch (error) { releaseWakeNotification(); throw error; @@ -352,6 +391,7 @@ export default function (omp: ExtensionAPI) { watch.sessionStart(); publishSecondmateSession(ctx); deliverSessionstartNudge(); + replayWakeNotification(); }); omp.on("session_switch", (event, ctx) => { diff --git a/docs/watcher-continuity.md b/docs/watcher-continuity.md index c19a94c34c6..d7cc79ca000 100644 --- a/docs/watcher-continuity.md +++ b/docs/watcher-continuity.md @@ -26,7 +26,7 @@ While supervision is still needed and away mode remains inactive, an actionable After an actionable Pi, OMP, or OpenCode child close, the adapter starts and verifies one singleton successor before it delivers the original wake. For OMP, the shared core confirms the recovery handoff before the adapter schedules at most one hidden custom `nextTurn` notification with `triggerTurn`; OMP consumes it on the next agent turn, including after prompt unwinding, without touching an editable TUI draft. -The adapter preserves that pending session-and-recovery claim across an extension reload in the same live conversation, while a replacement session or process re-presents the unacknowledged durable batch instead of trusting a lost continuation. +The adapter retains the exact batch payload in that pending session-and-recovery claim across an extension reload in the same live conversation, while a replacement session or process re-presents the unacknowledged durable batch instead of trusting a lost continuation. Notification delivery does not acknowledge durable wake rows; only the exact generation-bound `WAKE_ACK_REQUIRED` command printed by `bin/fm-wake-drain.sh` may retire them. A failed confirmation is never swallowed: the core retries once against the current generation and successor, then classifies the failure, retires a successor that is no longer alive, and surfaces exactly one typed message. It waits at most one readiness timeout per attempt, then sends TERM and waits a bounded retirement confirmation before the next lock-verified exponential retry. diff --git a/tests/fm-omp-primary.test.sh b/tests/fm-omp-primary.test.sh index 12d02752dd6..ea7455e9092 100755 --- a/tests/fm-omp-primary.test.sh +++ b/tests/fm-omp-primary.test.sh @@ -936,9 +936,9 @@ JS # OMP's hidden next-turn queue retains custom messages while a prompt unwinds. # This fixture drives successive durable wake batches through the native primary # adapter and verifies the core contributes one continuation until OMP begins -# that next turn. It also keeps the durable rows and a rejected recovery -# generation intact, so neither notification delivery nor a failed handoff can -# retire or strand replayable work. +# that next turn. +# It proves a same-session reload retains that continuation and a replacement +# primary re-notifies the exact unacknowledged batch without retiring its rows. test_native_omp_coalesces_pending_next_turn_notifications() { local fixture out status=0 fixture="$TMP_ROOT/native-next-turn-batching" @@ -987,6 +987,7 @@ SH FM_HOME="$fixture" FM_ROOT_OVERRIDE="$fixture" FM_STATE_OVERRIDE="$fixture/state" \ FM_ARM_LOG="$TMP_ROOT/native-next-turn-batching.log" \ FM_WAKE_LIB="$ROOT/bin/fm-wake-lib.sh" \ + FM_OMP_TASK_INBOX_DIR="" FM_OMP_TASK_DOORBELL_READY="" \ node --input-type=module 2>&1 <<'JS' import { spawnSync } from "node:child_process"; import { existsSync, readFileSync, writeFileSync } from "node:fs"; @@ -1047,6 +1048,14 @@ if (deduped.status !== 0 || deduped.stdout.trim().split("\n").length !== 2) { throw new Error(`durable batch was not deduplicated: ${deduped.stderr || deduped.stdout}`); } +const firstSession = `${state}/first-session.jsonl`; +const replacementSession = `${state}/replacement-session.jsonl`; +const startSession = async (sessionFile) => { + await handlers.get("session_start")( + { type: "session_start" }, + { sessionManager: { getSessionFile: () => sessionFile } }, + ); +}; const startNextTurn = async () => { await handlers.get("before_agent_start")({ type: "before_agent_start" }, {}); }; @@ -1057,15 +1066,17 @@ const trigger = async (count, messageCount) => { await waitFor(() => notifications.length >= messageCount, `notification ${messageCount}`); } }; -const reloadExtension = async (label) => { +const reloadExtension = async (label, sessionFile, arm = true) => { const reloaded = await import(`${pathToFileURL(process.env.EXTENSION).href}?${label}=${Date.now()}`); reloaded.default(api); if (!tool) throw new Error(`OMP did not register its watcher arm tool after ${label}`); - await tool.execute(); + await startSession(sessionFile); + if (arm) await tool.execute(); }; try { + await startSession(firstSession); await tool.execute(); await waitFor(() => armCount() === 1, "initial watcher"); await trigger(1, 1); @@ -1082,7 +1093,7 @@ try { throw new Error(`first notification was not hidden next-turn delivery: ${JSON.stringify(first)}`); } - await reloadExtension("same-session-reload"); + await reloadExtension("same-session-reload", firstSession); await waitFor(() => armCount() >= 3, "same-session reload watcher"); await trigger(3); await new Promise((resolve) => setTimeout(resolve, 40)); @@ -1094,19 +1105,23 @@ try { const staleClaim = readFileSync(claimPath, "utf8").split("\n"); staleClaim[1] = "1"; writeFileSync(claimPath, staleClaim.join("\n")); - await reloadExtension("process-restart"); - await waitFor(() => armCount() >= 5, "restart watcher"); - await trigger(5, 2); + await reloadExtension("process-restart", replacementSession, false); + await waitFor(() => notifications.length >= 2, "original batch replay after restart"); if (readFileSync(queue, "utf8") !== queueRows) { throw new Error("process-restart replay retired durable wake rows"); } - if (!notifications[1].message.content.includes("signal: synthetic durable batch 5")) { - throw new Error("process restart did not replay the unacknowledged durable batch"); + if (notifications[1].message.content !== first.message.content) { + throw new Error("process restart did not re-notify the exact unacknowledged durable batch"); + } + if (sendAttempts !== 2) { + throw new Error(`process restart did not make exactly one original-batch replay attempt: ${sendAttempts}`); } + await tool.execute(); + await waitFor(() => armCount() >= 5, "restart watcher"); await startNextTurn(); failNextNotification = true; - await trigger(6, 3); + await trigger(5, 3); if (sendAttempts !== 4 || !notifications[2].message.content.includes("could not deliver an actionable wake")) { throw new Error(`failed next-turn submission did not surface one replayable failure: ${JSON.stringify(notifications)}`); } @@ -1115,14 +1130,14 @@ try { } await startNextTurn(); - await trigger(7, 4); - if (!notifications[3].message.content.includes("signal: synthetic durable batch 7")) { + await trigger(6, 4); + if (!notifications[3].message.content.includes("signal: synthetic durable batch 6")) { throw new Error("durable wake did not replay after next-turn submission failure"); } await startNextTurn(); writeFileSync(`${state}/reject-confirmation`, "reject\n"); - await trigger(8, 5); + await trigger(7, 5); const mismatch = notifications[4]; if (!mismatch.message.content.includes("recovery generation mismatch")) { throw new Error(`generation-mismatched handoff was not surfaced: ${JSON.stringify(mismatch)}`); From 51a2eca3ddf4238dbb9c3e9b6b707b3180c63e76 Mon Sep 17 00:00:00 2001 From: Pranay Pratyush Date: Mon, 31 Aug 2026 17:01:13 +0530 Subject: [PATCH 04/11] no-mistakes(review): Preserve OMP wake claims across switches and fallback --- .omp/extensions/fm-branch-supervision-omp.ts | 14 ++++---- .omp/extensions/fm-primary-omp.ts | 35 ++++++++++++++----- .omp/extensions/lib/fm-branch-dispatch.ts | 20 +++++++++++ tests/fm-omp-primary.test.sh | 36 ++++++++++++++------ 4 files changed, 79 insertions(+), 26 deletions(-) diff --git a/.omp/extensions/fm-branch-supervision-omp.ts b/.omp/extensions/fm-branch-supervision-omp.ts index c2e6b5eaab5..d1afafb3b57 100644 --- a/.omp/extensions/fm-branch-supervision-omp.ts +++ b/.omp/extensions/fm-branch-supervision-omp.ts @@ -83,7 +83,9 @@ import type { Model } from "@oh-my-pi/pi-ai"; import type { Effort } from "@oh-my-pi/pi-catalog/effort"; import { activateEligibleRowsOwner, + createPrimaryWatcherWake, FM_BRANCH_DISPATCH_EVENT, + FM_PRIMARY_WATCHER_WAKE_EVENT, releaseEligibleRowsSnapshot, rollbackEligibleRowsOwnerActivation, scopeForUnreadWake, @@ -850,20 +852,18 @@ ${context.command} // Marked operational like every watcher injection, so the wake is never // mistaken for captain input (away-mode return semantics, mirror filter). const content = encodeOperationalInput(body); - // Deliver through the exact main-wake mechanism the primary OMP adapter uses - // (fm-primary-omp.ts sendFollowUp): a custom watcher-wake message delivered - // as a steer with triggerTurn. Unlike sendUserMessage/followUp, this - // reliably wakes an idle or interrupted main under OMP steer/continuation - // semantics, so a broken branch never strands the wake. + const wake = createPrimaryWatcherWake(encodeOperationalInput(body), "branch-fallback"); + pi.events?.emit?.(FM_PRIMARY_WATCHER_WAKE_EVENT, wake); + if (wake.accepted) return; pi.sendMessage( { customType: "firstmate-watcher-wake", - content, + content: wake.content, display: false, attribution: "agent", details: { kind: "watcher", runtime: "omp" }, }, - { deliverAs: "steer", triggerTurn: true }, + { deliverAs: "nextTurn", triggerTurn: true }, ); } diff --git a/.omp/extensions/fm-primary-omp.ts b/.omp/extensions/fm-primary-omp.ts index 90493b95d80..278eb38e550 100644 --- a/.omp/extensions/fm-primary-omp.ts +++ b/.omp/extensions/fm-primary-omp.ts @@ -26,7 +26,9 @@ import { createPrimaryWatchCore, ompNativeProcessIdentity } from "../../bin/fm-p import { createBranchDispatchOffer, FM_BRANCH_DISPATCH_EVENT, + FM_PRIMARY_WATCHER_WAKE_EVENT, scopeForUnreadWake, + type PrimaryWatcherWake, } from "./lib/fm-branch-dispatch.ts"; import { installTaskInboxDoorbell } from "./lib/fm-task-inbox-doorbell.ts"; @@ -331,6 +333,29 @@ export default function (omp: ExtensionAPI) { } } }; + const queueWakeNotification = (content: string, notificationKey: string): void => { + if (!claimWakeNotification(notificationKey, content)) return; + try { + sendWakeNotification(content); + } catch (error) { + releaseWakeNotification(); + throw error; + } + }; + + omp.events?.on?.(FM_PRIMARY_WATCHER_WAKE_EVENT, (data) => { + const wake = data as PrimaryWatcherWake; + if ( + !wake || + typeof wake.accept !== "function" || + typeof wake.content !== "string" || + typeof wake.notificationKey !== "string" + ) { + return; + } + wake.accept(); + queueWakeNotification(wake.content, wake.notificationKey); + }); // Supervision-branch dispatch handshake (docs/omp-supervision-branch.md). // Build one offer per ordinary actionable wake and emit it on the shared @@ -369,13 +394,7 @@ export default function (omp: ExtensionAPI) { // Reuse a claim from a same-session extension reload, but let a new // session or process replay the durable batch. The claim never retires // rows; only the drain acknowledgement owns that transition. - if (!claimWakeNotification(notificationKey, content)) return; - try { - sendWakeNotification(content); - } catch (error) { - releaseWakeNotification(); - throw error; - } + queueWakeNotification(content, notificationKey); }, offerWakeToBranch, }); @@ -395,13 +414,13 @@ export default function (omp: ExtensionAPI) { }); omp.on("session_switch", (event, ctx) => { - releaseWakeNotification(); setNotificationSession(ctx); watch.sessionShutdown(); watch.sessionStart(); publishSecondmateSession(ctx); deliverSessionstartNudge(event.reason === "new" || event.reason === "resume"); watch.arm(); + replayWakeNotification(); }); omp.on("before_agent_start", (): BeforeAgentStartEventResult | undefined => { diff --git a/.omp/extensions/lib/fm-branch-dispatch.ts b/.omp/extensions/lib/fm-branch-dispatch.ts index f2d88f2fe35..a1680983cd2 100644 --- a/.omp/extensions/lib/fm-branch-dispatch.ts +++ b/.omp/extensions/lib/fm-branch-dispatch.ts @@ -16,6 +16,26 @@ import { readdirSync, readFileSync } from "node:fs"; // main can repair the watcher cycle (fm_watch_arm_pi lives on main). export const FM_BRANCH_DISPATCH_EVENT = "fm-branch-supervision:dispatch"; +export const FM_PRIMARY_WATCHER_WAKE_EVENT = "fm-branch-supervision:main-wake"; + +export interface PrimaryWatcherWake { + content: string; + notificationKey: string; + accepted: boolean; + accept(): void; +} + +export function createPrimaryWatcherWake(content: string, notificationKey: string): PrimaryWatcherWake { + const wake: PrimaryWatcherWake = { + content, + notificationKey, + accepted: false, + accept() { + wake.accepted = true; + }, + }; + return wake; +} export type UnreadWakeScopeStatus = "safe" | "empty" | "unsafe"; diff --git a/tests/fm-omp-primary.test.sh b/tests/fm-omp-primary.test.sh index ea7455e9092..a979125121a 100755 --- a/tests/fm-omp-primary.test.sh +++ b/tests/fm-omp-primary.test.sh @@ -1066,6 +1066,7 @@ const trigger = async (count, messageCount) => { await waitFor(() => notifications.length >= messageCount, `notification ${messageCount}`); } }; +const triggerCurrent = async (messageCount) => trigger(armCount(), messageCount); const reloadExtension = async (label, sessionFile, arm = true) => { const reloaded = await import(`${pathToFileURL(process.env.EXTENSION).href}?${label}=${Date.now()}`); reloaded.default(api); @@ -1095,25 +1096,38 @@ try { await reloadExtension("same-session-reload", firstSession); await waitFor(() => armCount() >= 3, "same-session reload watcher"); - await trigger(3); + await triggerCurrent(); await new Promise((resolve) => setTimeout(resolve, 40)); if (notifications.length !== 1) { throw new Error(`same-session reload duplicated a pending notification: ${JSON.stringify(notifications)}`); } + await handlers.get("session_switch")( + { type: "session_switch", reason: "new" }, + { sessionManager: { getSessionFile: () => replacementSession } }, + ); + await waitFor(() => notifications.length >= 2, "original batch replay after session switch"); + if (readFileSync(queue, "utf8") !== queueRows) { + throw new Error("session-switch replay retired durable wake rows"); + } + if (notifications[1].message.content !== first.message.content) { + throw new Error("session switch did not re-notify the exact unacknowledged durable batch"); + } + const claimPath = `${state}/.omp-primary-nextturn-notification`; const staleClaim = readFileSync(claimPath, "utf8").split("\n"); staleClaim[1] = "1"; writeFileSync(claimPath, staleClaim.join("\n")); - await reloadExtension("process-restart", replacementSession, false); - await waitFor(() => notifications.length >= 2, "original batch replay after restart"); + const restartedSession = `${state}/restarted-session.jsonl`; + await reloadExtension("process-restart", restartedSession, false); + await waitFor(() => notifications.length >= 3, "original batch replay after restart"); if (readFileSync(queue, "utf8") !== queueRows) { throw new Error("process-restart replay retired durable wake rows"); } - if (notifications[1].message.content !== first.message.content) { + if (notifications[2].message.content !== first.message.content) { throw new Error("process restart did not re-notify the exact unacknowledged durable batch"); } - if (sendAttempts !== 2) { + if (sendAttempts !== 3) { throw new Error(`process restart did not make exactly one original-batch replay attempt: ${sendAttempts}`); } @@ -1121,8 +1135,8 @@ try { await waitFor(() => armCount() >= 5, "restart watcher"); await startNextTurn(); failNextNotification = true; - await trigger(5, 3); - if (sendAttempts !== 4 || !notifications[2].message.content.includes("could not deliver an actionable wake")) { + await triggerCurrent(4); + if (sendAttempts !== 5 || !notifications[3].message.content.includes("could not deliver an actionable wake")) { throw new Error(`failed next-turn submission did not surface one replayable failure: ${JSON.stringify(notifications)}`); } if (readFileSync(queue, "utf8") !== queueRows) { @@ -1130,15 +1144,15 @@ try { } await startNextTurn(); - await trigger(6, 4); - if (!notifications[3].message.content.includes("signal: synthetic durable batch 6")) { + await triggerCurrent(5); + if (!notifications[4].message.content.includes("signal: synthetic durable batch")) { throw new Error("durable wake did not replay after next-turn submission failure"); } await startNextTurn(); writeFileSync(`${state}/reject-confirmation`, "reject\n"); - await trigger(7, 5); - const mismatch = notifications[4]; + await triggerCurrent(6); + const mismatch = notifications[5]; if (!mismatch.message.content.includes("recovery generation mismatch")) { throw new Error(`generation-mismatched handoff was not surfaced: ${JSON.stringify(mismatch)}`); } From 3439534965096e3989662c1e17702a46359c3e65 Mon Sep 17 00:00:00 2001 From: Pranay Pratyush Date: Mon, 31 Aug 2026 17:17:19 +0530 Subject: [PATCH 05/11] no-mistakes(review): Harden OMP wake claim ownership and batching --- .omp/extensions/fm-branch-supervision-omp.ts | 1 - .omp/extensions/fm-primary-omp.ts | 53 ++++++++++++-------- tests/fm-omp-primary.test.sh | 19 ++++--- 3 files changed, 44 insertions(+), 29 deletions(-) diff --git a/.omp/extensions/fm-branch-supervision-omp.ts b/.omp/extensions/fm-branch-supervision-omp.ts index d1afafb3b57..c09d73f8087 100644 --- a/.omp/extensions/fm-branch-supervision-omp.ts +++ b/.omp/extensions/fm-branch-supervision-omp.ts @@ -851,7 +851,6 @@ ${context.command} const body = `FIRSTMATE WATCHER WAKE: ${message}\n\nRun bin/fm-wake-drain.sh first and handle the queued wake. (Supervision branch unavailable, falling back to main: ${detail})`; // Marked operational like every watcher injection, so the wake is never // mistaken for captain input (away-mode return semantics, mirror filter). - const content = encodeOperationalInput(body); const wake = createPrimaryWatcherWake(encodeOperationalInput(body), "branch-fallback"); pi.events?.emit?.(FM_PRIMARY_WATCHER_WAKE_EVENT, wake); if (wake.accepted) return; diff --git a/.omp/extensions/fm-primary-omp.ts b/.omp/extensions/fm-primary-omp.ts index 278eb38e550..b481b24ae8a 100644 --- a/.omp/extensions/fm-primary-omp.ts +++ b/.omp/extensions/fm-primary-omp.ts @@ -49,6 +49,7 @@ type ProcessResult = { type WakeNotificationClaim = { pid: string; + instance: string; session: string; key: string; content: string; @@ -210,19 +211,23 @@ function readWakeNotificationClaim(): WakeNotificationClaim | undefined { return undefined; } const lines = content.split("\n"); + const version = lines[0]; + const v2 = version === "fm-omp-primary-nextturn-notification-v2" && lines.length === 6; + const v3 = version === "fm-omp-primary-nextturn-notification-v3" && lines.length === 7; + if (!v2 && !v3) return undefined; + const offset = v3 ? 1 : 0; if ( - lines.length !== 6 || - lines[0] !== "fm-omp-primary-nextturn-notification-v2" || !/^[0-9]+$/u.test(lines[1]) || - !/^[a-f0-9]{64}$/u.test(lines[2]) || - !/^[A-Za-z0-9._-]+$/u.test(lines[3]) || - lines[5] !== "" + (v3 && !/^[a-f0-9-]{36}$/u.test(lines[2])) || + !/^[a-f0-9]{64}$/u.test(lines[2 + offset]) || + !/^[A-Za-z0-9._-]+$/u.test(lines[3 + offset]) || + lines[5 + offset] !== "" ) { return undefined; } - const message = Buffer.from(lines[4], "base64").toString("utf8"); - if (Buffer.from(message, "utf8").toString("base64") !== lines[4]) return undefined; - return { pid: lines[1], session: lines[2], key: lines[3], content: message }; + const message = Buffer.from(lines[4 + offset], "base64").toString("utf8"); + if (Buffer.from(message, "utf8").toString("base64") !== lines[4 + offset]) return undefined; + return { pid: lines[1], instance: v3 ? lines[2] : "", session: lines[2 + offset], key: lines[3 + offset], content: message }; } function writeWakeNotificationClaim(claim: WakeNotificationClaim): void { @@ -234,8 +239,9 @@ function writeWakeNotificationClaim(claim: WakeNotificationClaim): void { writeFileSync( descriptor, [ - "fm-omp-primary-nextturn-notification-v2", + "fm-omp-primary-nextturn-notification-v3", claim.pid, + claim.instance, claim.session, claim.key, Buffer.from(claim.content, "utf8").toString("base64"), @@ -268,6 +274,8 @@ export default function (omp: ExtensionAPI) { publishNativeProcessIdentity(); const taskInboxDoorbell = installTaskInboxDoorbell(omp); let pendingStartupNudge = ""; + const runtime = globalThis as typeof globalThis & { firstmateOmpPrimaryNotificationInstance?: string }; + const notificationInstance = runtime.firstmateOmpPrimaryNotificationInstance ??= randomUUID(); let notificationSession = createHash("sha256").update("unknown").digest("hex"); @@ -289,7 +297,7 @@ export default function (omp: ExtensionAPI) { }; const releaseWakeNotification = (): void => { const claim = readWakeNotificationClaim(); - if (!claim || claim.pid !== String(process.pid)) return; + if (!claim || claim.instance !== notificationInstance) return; try { unlinkSync(notificationClaim); } catch { @@ -298,15 +306,10 @@ export default function (omp: ExtensionAPI) { }; const claimWakeNotification = (key: string, content: string): boolean => { const current = readWakeNotificationClaim(); - if ( - current?.pid === String(process.pid) && - current.session === notificationSession && - current.key === key - ) { - return false; - } + if (current) return false; writeWakeNotificationClaim({ pid: String(process.pid), + instance: notificationInstance, session: notificationSession, key, content, @@ -317,11 +320,16 @@ export default function (omp: ExtensionAPI) { const pending = readWakeNotificationClaim(); if ( !pending || - (pending.pid === String(process.pid) && pending.session === notificationSession) + (pending.instance === notificationInstance && pending.session === notificationSession) ) { return; } - const replay = { ...pending, pid: String(process.pid), session: notificationSession }; + const replay = { + ...pending, + pid: String(process.pid), + instance: notificationInstance, + session: notificationSession, + }; try { writeWakeNotificationClaim(replay); sendWakeNotification(pending.content); @@ -333,10 +341,11 @@ export default function (omp: ExtensionAPI) { } } }; - const queueWakeNotification = (content: string, notificationKey: string): void => { - if (!claimWakeNotification(notificationKey, content)) return; + const queueWakeNotification = (content: string, notificationKey: string): boolean => { + if (!claimWakeNotification(notificationKey, content)) return true; try { sendWakeNotification(content); + return true; } catch (error) { releaseWakeNotification(); throw error; @@ -353,8 +362,8 @@ export default function (omp: ExtensionAPI) { ) { return; } - wake.accept(); queueWakeNotification(wake.content, wake.notificationKey); + wake.accept(); }); // Supervision-branch dispatch handshake (docs/omp-supervision-branch.md). diff --git a/tests/fm-omp-primary.test.sh b/tests/fm-omp-primary.test.sh index a979125121a..83d657a041c 100755 --- a/tests/fm-omp-primary.test.sh +++ b/tests/fm-omp-primary.test.sh @@ -994,6 +994,7 @@ import { existsSync, readFileSync, writeFileSync } from "node:fs"; import { pathToFileURL } from "node:url"; const handlers = new Map(); +const eventHandlers = new Map(); const notifications = []; let tool = null; let sendAttempts = 0; @@ -1013,6 +1014,10 @@ const waitFor = async (predicate, description) => { const api = { zod: { object: () => ({}) }, on(name, handler) { handlers.set(name, handler); }, + events: { + on(name, handler) { eventHandlers.set(name, handler); }, + emit(name, data) { eventHandlers.get(name)?.(data); }, + }, registerCommand() {}, registerTool(candidate) { if (candidate.name === "fm_watch_arm_omp") tool = candidate; @@ -1031,6 +1036,7 @@ process.argv[1] = process.env.EXTENSION; const extension = await import(`${pathToFileURL(process.env.EXTENSION).href}?batching=${Date.now()}`); extension.default(api); if (!tool) throw new Error("OMP did not register its watcher arm tool"); +const dispatch = await import(new URL("./lib/fm-branch-dispatch.ts", pathToFileURL(process.env.EXTENSION)).href); const queue = `${state}/.wake-queue`; const queueRows = [ @@ -1093,6 +1099,11 @@ try { ) { throw new Error(`first notification was not hidden next-turn delivery: ${JSON.stringify(first)}`); } + const fallback = dispatch.createPrimaryWatcherWake("fallback batch", "branch-fallback"); + api.events.emit(dispatch.FM_PRIMARY_WATCHER_WAKE_EVENT, fallback); + if (!fallback.accepted || notifications.length !== 1) { + throw new Error(`branch fallback did not share the pending next-turn notification: ${JSON.stringify(notifications)}`); + } await reloadExtension("same-session-reload", firstSession); await waitFor(() => armCount() >= 3, "same-session reload watcher"); @@ -1114,12 +1125,8 @@ try { throw new Error("session switch did not re-notify the exact unacknowledged durable batch"); } - const claimPath = `${state}/.omp-primary-nextturn-notification`; - const staleClaim = readFileSync(claimPath, "utf8").split("\n"); - staleClaim[1] = "1"; - writeFileSync(claimPath, staleClaim.join("\n")); - const restartedSession = `${state}/restarted-session.jsonl`; - await reloadExtension("process-restart", restartedSession, false); + delete globalThis.firstmateOmpPrimaryNotificationInstance; + await reloadExtension("process-restart", replacementSession, false); await waitFor(() => notifications.length >= 3, "original batch replay after restart"); if (readFileSync(queue, "utf8") !== queueRows) { throw new Error("process-restart replay retired durable wake rows"); From 110f90510b32c0976b87f66427549537c068a622 Mon Sep 17 00:00:00 2001 From: Pranay Pratyush Date: Mon, 31 Aug 2026 17:25:37 +0530 Subject: [PATCH 06/11] no-mistakes(review): Retain OMP wake claims through handling --- .omp/extensions/fm-primary-omp.ts | 61 ++++++++++++++++++++++++------- bin/fm-primary-watch-core.ts | 3 -- tests/fm-omp-primary.test.sh | 1 + 3 files changed, 49 insertions(+), 16 deletions(-) diff --git a/.omp/extensions/fm-primary-omp.ts b/.omp/extensions/fm-primary-omp.ts index b481b24ae8a..6544c33972b 100644 --- a/.omp/extensions/fm-primary-omp.ts +++ b/.omp/extensions/fm-primary-omp.ts @@ -51,6 +51,7 @@ type WakeNotificationClaim = { pid: string; instance: string; session: string; + state: "pending" | "inflight"; key: string; content: string; }; @@ -214,20 +215,33 @@ function readWakeNotificationClaim(): WakeNotificationClaim | undefined { const version = lines[0]; const v2 = version === "fm-omp-primary-nextturn-notification-v2" && lines.length === 6; const v3 = version === "fm-omp-primary-nextturn-notification-v3" && lines.length === 7; - if (!v2 && !v3) return undefined; - const offset = v3 ? 1 : 0; + const v4 = version === "fm-omp-primary-nextturn-notification-v4" && lines.length === 8; + if (!v2 && !v3 && !v4) return undefined; + const claimState = v4 ? lines[4] : "pending"; + const sessionIndex = v2 ? 2 : 3; + const keyIndex = v4 ? 5 : sessionIndex + 1; + const contentIndex = keyIndex + 1; if ( !/^[0-9]+$/u.test(lines[1]) || (v3 && !/^[a-f0-9-]{36}$/u.test(lines[2])) || - !/^[a-f0-9]{64}$/u.test(lines[2 + offset]) || - !/^[A-Za-z0-9._-]+$/u.test(lines[3 + offset]) || - lines[5 + offset] !== "" + (v4 && !/^[a-f0-9-]{36}$/u.test(lines[2])) || + (claimState !== "pending" && claimState !== "inflight") || + !/^[a-f0-9]{64}$/u.test(lines[sessionIndex]) || + !/^[A-Za-z0-9._-]+$/u.test(lines[keyIndex]) || + lines[contentIndex + 1] !== "" ) { return undefined; } - const message = Buffer.from(lines[4 + offset], "base64").toString("utf8"); - if (Buffer.from(message, "utf8").toString("base64") !== lines[4 + offset]) return undefined; - return { pid: lines[1], instance: v3 ? lines[2] : "", session: lines[2 + offset], key: lines[3 + offset], content: message }; + const message = Buffer.from(lines[contentIndex], "base64").toString("utf8"); + if (Buffer.from(message, "utf8").toString("base64") !== lines[contentIndex]) return undefined; + return { + pid: lines[1], + instance: v3 || v4 ? lines[2] : "", + session: lines[sessionIndex], + state: claimState, + key: lines[keyIndex], + content: message, + }; } function writeWakeNotificationClaim(claim: WakeNotificationClaim): void { @@ -239,10 +253,11 @@ function writeWakeNotificationClaim(claim: WakeNotificationClaim): void { writeFileSync( descriptor, [ - "fm-omp-primary-nextturn-notification-v3", + "fm-omp-primary-nextturn-notification-v4", claim.pid, claim.instance, claim.session, + claim.state, claim.key, Buffer.from(claim.content, "utf8").toString("base64"), "", @@ -295,7 +310,7 @@ export default function (omp: ExtensionAPI) { { deliverAs: "nextTurn", triggerTurn: true }, ); }; - const releaseWakeNotification = (): void => { + const discardWakeNotification = (): void => { const claim = readWakeNotificationClaim(); if (!claim || claim.instance !== notificationInstance) return; try { @@ -306,11 +321,19 @@ export default function (omp: ExtensionAPI) { }; const claimWakeNotification = (key: string, content: string): boolean => { const current = readWakeNotificationClaim(); - if (current) return false; + if ( + current && + (current.instance !== notificationInstance || + current.session !== notificationSession || + current.state === "pending") + ) { + return false; + } writeWakeNotificationClaim({ pid: String(process.pid), instance: notificationInstance, session: notificationSession, + state: "pending", key, content, }); @@ -347,10 +370,22 @@ export default function (omp: ExtensionAPI) { sendWakeNotification(content); return true; } catch (error) { - releaseWakeNotification(); + discardWakeNotification(); throw error; } }; + const markWakeNotificationInflight = (): void => { + const claim = readWakeNotificationClaim(); + if ( + !claim || + claim.instance !== notificationInstance || + claim.session !== notificationSession || + claim.state === "inflight" + ) { + return; + } + writeWakeNotificationClaim({ ...claim, state: "inflight" }); + }; omp.events?.on?.(FM_PRIMARY_WATCHER_WAKE_EVENT, (data) => { const wake = data as PrimaryWatcherWake; @@ -398,7 +433,6 @@ export default function (omp: ExtensionAPI) { repairToolName: "fm_watch_arm_omp", encodeOperationalInput, coalesceWakeNotification: true, - releaseWakeNotification, sendFollowUp: async (content, notificationKey) => { // Reuse a claim from a same-session extension reload, but let a new // session or process replay the durable batch. The claim never retires @@ -434,6 +468,7 @@ export default function (omp: ExtensionAPI) { omp.on("before_agent_start", (): BeforeAgentStartEventResult | undefined => { watch.notificationTurnStarted(); + markWakeNotificationInflight(); if (!pendingStartupNudge) return undefined; const content = pendingStartupNudge; pendingStartupNudge = ""; diff --git a/bin/fm-primary-watch-core.ts b/bin/fm-primary-watch-core.ts index 35f14211956..6b8762d09c4 100644 --- a/bin/fm-primary-watch-core.ts +++ b/bin/fm-primary-watch-core.ts @@ -84,7 +84,6 @@ export type PrimaryWatchCoreOptions = { // agent turn begins, so concurrent actionable closes coalesce without // changing the durable wake queue or its acknowledgement ownership. coalesceWakeNotification?: boolean; - releaseWakeNotification?: () => void; // Optional supervision-branch dispatch handshake. When supplied (the OMP // adapter with its branch extension loaded), the core offers each ordinary // actionable wake to the branch before delivering it to main; a synchronous @@ -223,7 +222,6 @@ export function createPrimaryWatchCore(options: PrimaryWatchCoreOptions): Primar encodeOperationalInput, sendFollowUp, coalesceWakeNotification = false, - releaseWakeNotification, offerWakeToBranch, } = options; const armScript = `${fmRoot}/bin/fm-watch-arm.sh`; @@ -718,7 +716,6 @@ export function createPrimaryWatchCore(options: PrimaryWatchCoreOptions): Primar function notificationTurnStarted(): void { if (coalesceWakeNotification) generation.notificationPending = false; - releaseWakeNotification?.(); } function sessionStart(): void { diff --git a/tests/fm-omp-primary.test.sh b/tests/fm-omp-primary.test.sh index 83d657a041c..9d86ac36047 100755 --- a/tests/fm-omp-primary.test.sh +++ b/tests/fm-omp-primary.test.sh @@ -1113,6 +1113,7 @@ try { throw new Error(`same-session reload duplicated a pending notification: ${JSON.stringify(notifications)}`); } + await startNextTurn(); await handlers.get("session_switch")( { type: "session_switch", reason: "new" }, { sessionManager: { getSessionFile: () => replacementSession } }, From 6bc7f7139d443bce3a80daf2dc9f06ae85982119 Mon Sep 17 00:00:00 2001 From: Pranay Pratyush Date: Mon, 31 Aug 2026 17:39:33 +0530 Subject: [PATCH 07/11] no-mistakes(review): Reconcile OMP claims with acknowledged watcher delivery --- .omp/extensions/fm-primary-omp.ts | 69 ++++++++++++++++++++++++++----- bin/fm-wake-drain.sh | 20 +++++++++ tests/fm-omp-primary.test.sh | 27 ++++++++++++ 3 files changed, 105 insertions(+), 11 deletions(-) diff --git a/.omp/extensions/fm-primary-omp.ts b/.omp/extensions/fm-primary-omp.ts index 6544c33972b..ba195c6a8ff 100644 --- a/.omp/extensions/fm-primary-omp.ts +++ b/.omp/extensions/fm-primary-omp.ts @@ -41,6 +41,7 @@ const config = process.env.FM_CONFIG_OVERRIDE || `${fmHome}/config`; const marker = `${state}/.omp-primary-extension-loaded`; const operationalInputScript = `${fmRoot}/bin/fm-operational-input.sh`; const notificationClaim = `${state}/.omp-primary-nextturn-notification`; +const notificationAcknowledgement = `${state}/.omp-primary-nextturn-ack`; type ProcessResult = { code: number; @@ -52,6 +53,7 @@ type WakeNotificationClaim = { instance: string; session: string; state: "pending" | "inflight"; + id: string; key: string; content: string; }; @@ -216,16 +218,19 @@ function readWakeNotificationClaim(): WakeNotificationClaim | undefined { const v2 = version === "fm-omp-primary-nextturn-notification-v2" && lines.length === 6; const v3 = version === "fm-omp-primary-nextturn-notification-v3" && lines.length === 7; const v4 = version === "fm-omp-primary-nextturn-notification-v4" && lines.length === 8; - if (!v2 && !v3 && !v4) return undefined; + const v5 = version === "fm-omp-primary-nextturn-notification-v5" && lines.length === 9; + if (!v2 && !v3 && !v4 && !v5) return undefined; const claimState = v4 ? lines[4] : "pending"; + const v5ClaimState = v5 ? lines[4] : claimState; const sessionIndex = v2 ? 2 : 3; - const keyIndex = v4 ? 5 : sessionIndex + 1; + const keyIndex = v5 ? 6 : v4 ? 5 : sessionIndex + 1; const contentIndex = keyIndex + 1; if ( !/^[0-9]+$/u.test(lines[1]) || (v3 && !/^[a-f0-9-]{36}$/u.test(lines[2])) || - (v4 && !/^[a-f0-9-]{36}$/u.test(lines[2])) || - (claimState !== "pending" && claimState !== "inflight") || + ((v4 || v5) && !/^[a-f0-9-]{36}$/u.test(lines[2])) || + (v5 && !/^[a-f0-9-]{36}$/u.test(lines[5])) || + (v5ClaimState !== "pending" && v5ClaimState !== "inflight") || !/^[a-f0-9]{64}$/u.test(lines[sessionIndex]) || !/^[A-Za-z0-9._-]+$/u.test(lines[keyIndex]) || lines[contentIndex + 1] !== "" @@ -236,9 +241,10 @@ function readWakeNotificationClaim(): WakeNotificationClaim | undefined { if (Buffer.from(message, "utf8").toString("base64") !== lines[contentIndex]) return undefined; return { pid: lines[1], - instance: v3 || v4 ? lines[2] : "", + instance: v3 || v4 || v5 ? lines[2] : "", session: lines[sessionIndex], - state: claimState, + state: v5ClaimState, + id: v5 ? lines[5] : "", key: lines[keyIndex], content: message, }; @@ -253,11 +259,12 @@ function writeWakeNotificationClaim(claim: WakeNotificationClaim): void { writeFileSync( descriptor, [ - "fm-omp-primary-nextturn-notification-v4", + "fm-omp-primary-nextturn-notification-v5", claim.pid, claim.instance, claim.session, claim.state, + claim.id, claim.key, Buffer.from(claim.content, "utf8").toString("base64"), "", @@ -310,16 +317,35 @@ export default function (omp: ExtensionAPI) { { deliverAs: "nextTurn", triggerTurn: true }, ); }; - const discardWakeNotification = (): void => { + const discardWakeNotification = (expectedId = ""): void => { const claim = readWakeNotificationClaim(); - if (!claim || claim.instance !== notificationInstance) return; + if (!claim || (expectedId ? claim.id !== expectedId : claim.instance !== notificationInstance)) return; try { unlinkSync(notificationClaim); } catch { // A later durable wake reclaims any claim that this turn could not retire. } }; + const reconcileWakeNotificationAcknowledgement = (): void => { + let acknowledged = ""; + try { + const stats = lstatSync(notificationAcknowledgement); + if (!stats.isFile() || stats.isSymbolicLink()) return; + acknowledged = readFileSync(notificationAcknowledgement, "utf8"); + } catch { + return; + } + const claim = readWakeNotificationClaim(); + if (claim?.id === acknowledged.trim()) { + discardWakeNotification(claim.id); + if (readWakeNotificationClaim()?.id === claim.id) return; + } + try { + unlinkSync(notificationAcknowledgement); + } catch {} + }; const claimWakeNotification = (key: string, content: string): boolean => { + reconcileWakeNotificationAcknowledgement(); const current = readWakeNotificationClaim(); if ( current && @@ -334,12 +360,14 @@ export default function (omp: ExtensionAPI) { instance: notificationInstance, session: notificationSession, state: "pending", + id: randomUUID(), key, content, }); return true; }; const replayWakeNotification = (): void => { + reconcileWakeNotificationAcknowledgement(); const pending = readWakeNotificationClaim(); if ( !pending || @@ -352,6 +380,8 @@ export default function (omp: ExtensionAPI) { pid: String(process.pid), instance: notificationInstance, session: notificationSession, + state: "pending", + id: randomUUID(), }; try { writeWakeNotificationClaim(replay); @@ -384,7 +414,7 @@ export default function (omp: ExtensionAPI) { ) { return; } - writeWakeNotificationClaim({ ...claim, state: "inflight" }); + writeWakeNotificationClaim({ ...claim, state: "inflight", id: claim.id || randomUUID() }); }; omp.events?.on?.(FM_PRIMARY_WATCHER_WAKE_EVENT, (data) => { @@ -466,9 +496,26 @@ export default function (omp: ExtensionAPI) { replayWakeNotification(); }); - omp.on("before_agent_start", (): BeforeAgentStartEventResult | undefined => { + omp.on("message_start", (event) => { + const message = event.message as { role?: unknown; customType?: unknown; content?: unknown }; + const claim = readWakeNotificationClaim(); + if ( + message.role !== "custom" || + message.customType !== "firstmate-watcher-wake" || + typeof message.content !== "string" || + !claim || + claim.instance !== notificationInstance || + claim.session !== notificationSession || + claim.state !== "pending" || + claim.content !== message.content + ) { + return; + } watch.notificationTurnStarted(); markWakeNotificationInflight(); + }); + + omp.on("before_agent_start", (): BeforeAgentStartEventResult | undefined => { if (!pendingStartupNudge) return undefined; const content = pendingStartupNudge; pendingStartupNudge = ""; diff --git a/bin/fm-wake-drain.sh b/bin/fm-wake-drain.sh index 89eaecbe4ff..93927b5ef29 100755 --- a/bin/fm-wake-drain.sh +++ b/bin/fm-wake-drain.sh @@ -32,6 +32,25 @@ RECOVERY_ACK_REQUIRED=false RECOVERY_ACK_MOVED=false ACK_THROUGH= ACK_GENERATION= +OMP_NOTIFICATION_CLAIM="$STATE/.omp-primary-nextturn-notification" +OMP_NOTIFICATION_ACKNOWLEDGEMENT="$STATE/.omp-primary-nextturn-ack" + +publish_omp_notification_acknowledgement() { + local claim_id acknowledgement_tmp + [ "$ACTOR" = main ] || return 0 + [ -f "$OMP_NOTIFICATION_CLAIM" ] && [ ! -L "$OMP_NOTIFICATION_CLAIM" ] || return 0 + [ "$(sed -n '1p' "$OMP_NOTIFICATION_CLAIM" 2>/dev/null)" = fm-omp-primary-nextturn-notification-v5 ] || return 0 + [ "$(sed -n '5p' "$OMP_NOTIFICATION_CLAIM" 2>/dev/null)" = inflight ] || return 0 + claim_id=$(sed -n '6p' "$OMP_NOTIFICATION_CLAIM" 2>/dev/null) || return 1 + case "$claim_id" in ''|*[!a-f0-9-]*) return 0 ;; esac + acknowledgement_tmp=$(mktemp "$STATE/.omp-primary-nextturn-ack.XXXXXX") || return 1 + if ! chmod 0600 "$acknowledgement_tmp" \ + || ! printf '%s\n' "$claim_id" > "$acknowledgement_tmp" \ + || ! _fm_atomic_replace "$acknowledgement_tmp" "$OMP_NOTIFICATION_ACKNOWLEDGEMENT"; then + rm -f -- "$acknowledgement_tmp" + return 1 + fi +} # --- per-actor consume (docs/omp-supervision-branch.md "Per-actor acknowledgement") -- # main (FM_SUPERVISION_ACTOR unset or "main", via fm-lease-lib.sh's fm_lease_actor @@ -328,6 +347,7 @@ if [ -n "$ACK_THROUGH" ]; then consume_actor_rows_locked "$MAIN_ROWS_FILE" "$ACK_THROUGH" || exit 1 fi fi + publish_omp_notification_acknowledgement || exit 1 fm_lock_release "$FM_WAKE_QUEUE_LOCK" DRAIN_LOCK_HELD=false if [ "$RECOVERY_ACK_MOVED" = true ]; then diff --git a/tests/fm-omp-primary.test.sh b/tests/fm-omp-primary.test.sh index 9d86ac36047..8dd51c63096 100755 --- a/tests/fm-omp-primary.test.sh +++ b/tests/fm-omp-primary.test.sh @@ -987,6 +987,7 @@ SH FM_HOME="$fixture" FM_ROOT_OVERRIDE="$fixture" FM_STATE_OVERRIDE="$fixture/state" \ FM_ARM_LOG="$TMP_ROOT/native-next-turn-batching.log" \ FM_WAKE_LIB="$ROOT/bin/fm-wake-lib.sh" \ + FM_WAKE_DRAIN="$ROOT/bin/fm-wake-drain.sh" \ FM_OMP_TASK_INBOX_DIR="" FM_OMP_TASK_DOORBELL_READY="" \ node --input-type=module 2>&1 <<'JS' import { spawnSync } from "node:child_process"; @@ -1063,6 +1064,13 @@ const startSession = async (sessionFile) => { ); }; const startNextTurn = async () => { + const notification = notifications.at(-1); + if (notification) { + await handlers.get("message_start")( + { type: "message_start", message: { ...notification.message, role: "custom" } }, + {}, + ); + } await handlers.get("before_agent_start")({ type: "before_agent_start" }, {}); }; const trigger = async (count, messageCount) => { @@ -1099,6 +1107,7 @@ try { ) { throw new Error(`first notification was not hidden next-turn delivery: ${JSON.stringify(first)}`); } + await handlers.get("before_agent_start")({ type: "before_agent_start", prompt: "unrelated turn" }, {}); const fallback = dispatch.createPrimaryWatcherWake("fallback batch", "branch-fallback"); api.events.emit(dispatch.FM_PRIMARY_WATCHER_WAKE_EVENT, fallback); if (!fallback.accepted || notifications.length !== 1) { @@ -1178,6 +1187,24 @@ try { if (!handling.every((row) => row.includes("generation=fixture-generation"))) { throw new Error(`handling confirmation lost its recovery generation: ${handling.join(" | ")}`); } + await startNextTurn(); + writeFileSync(`${state}/.watcher-down`, "pending:handling:fixture-generation\n"); + const acknowledged = spawnSync(process.env.FM_WAKE_DRAIN, ["--ack-through", "3", "--recovery-generation", "fixture-generation"], { + encoding: "utf8", + env: { ...process.env, FM_HOME: process.env.FM_HOME, FM_STATE_OVERRIDE: state }, + }); + if (acknowledged.status !== 0 || !existsSync(`${state}/.omp-primary-nextturn-ack`)) { + throw new Error(`durable wake acknowledgement did not retire its notification claim: ${acknowledged.stderr}`); + } + const notificationsBeforeAcknowledgedSwitch = notifications.length; + await handlers.get("session_switch")( + { type: "session_switch", reason: "resume" }, + { sessionManager: { getSessionFile: () => `${state}/acknowledged-session.jsonl` } }, + ); + await new Promise((resolve) => setTimeout(resolve, 40)); + if (notifications.length !== notificationsBeforeAcknowledgedSwitch) { + throw new Error(`acknowledged durable wake replayed after replacement: ${JSON.stringify(notifications)}`); + } console.log("omp-next-turn-batching-ok"); } finally { writeFileSync(`${state}/watch-trigger-${armCount()}`, "stop\n"); From db8eea7053df7245ba0d20d7c5e6643cb87b7e90 Mon Sep 17 00:00:00 2001 From: Pranay Pratyush Date: Mon, 31 Aug 2026 18:01:58 +0530 Subject: [PATCH 08/11] no-mistakes(review): Bind OMP claim retirement to acknowledged queue cutoffs --- .omp/extensions/fm-branch-supervision-omp.ts | 4 +- .omp/extensions/fm-primary-omp.ts | 42 ++++++++++++++------ bin/fm-wake-drain.sh | 12 ++++-- tests/fm-omp-primary.test.sh | 5 +++ 4 files changed, 47 insertions(+), 16 deletions(-) diff --git a/.omp/extensions/fm-branch-supervision-omp.ts b/.omp/extensions/fm-branch-supervision-omp.ts index c09d73f8087..7ae9d522994 100644 --- a/.omp/extensions/fm-branch-supervision-omp.ts +++ b/.omp/extensions/fm-branch-supervision-omp.ts @@ -852,7 +852,9 @@ ${context.command} // Marked operational like every watcher injection, so the wake is never // mistaken for captain input (away-mode return semantics, mirror filter). const wake = createPrimaryWatcherWake(encodeOperationalInput(body), "branch-fallback"); - pi.events?.emit?.(FM_PRIMARY_WATCHER_WAKE_EVENT, wake); + try { + pi.events?.emit?.(FM_PRIMARY_WATCHER_WAKE_EVENT, wake); + } catch {} if (wake.accepted) return; pi.sendMessage( { diff --git a/.omp/extensions/fm-primary-omp.ts b/.omp/extensions/fm-primary-omp.ts index ba195c6a8ff..5aaf62cce0e 100644 --- a/.omp/extensions/fm-primary-omp.ts +++ b/.omp/extensions/fm-primary-omp.ts @@ -54,6 +54,7 @@ type WakeNotificationClaim = { session: string; state: "pending" | "inflight"; id: string; + through: string; key: string; content: string; }; @@ -219,18 +220,19 @@ function readWakeNotificationClaim(): WakeNotificationClaim | undefined { const v3 = version === "fm-omp-primary-nextturn-notification-v3" && lines.length === 7; const v4 = version === "fm-omp-primary-nextturn-notification-v4" && lines.length === 8; const v5 = version === "fm-omp-primary-nextturn-notification-v5" && lines.length === 9; - if (!v2 && !v3 && !v4 && !v5) return undefined; - const claimState = v4 ? lines[4] : "pending"; - const v5ClaimState = v5 ? lines[4] : claimState; + const v6 = version === "fm-omp-primary-nextturn-notification-v6" && lines.length === 10; + if (!v2 && !v3 && !v4 && !v5 && !v6) return undefined; + const claimState = v4 || v5 || v6 ? lines[4] : "pending"; const sessionIndex = v2 ? 2 : 3; - const keyIndex = v5 ? 6 : v4 ? 5 : sessionIndex + 1; + const keyIndex = v6 ? 7 : v5 ? 6 : v4 ? 5 : sessionIndex + 1; const contentIndex = keyIndex + 1; if ( !/^[0-9]+$/u.test(lines[1]) || (v3 && !/^[a-f0-9-]{36}$/u.test(lines[2])) || - ((v4 || v5) && !/^[a-f0-9-]{36}$/u.test(lines[2])) || - (v5 && !/^[a-f0-9-]{36}$/u.test(lines[5])) || - (v5ClaimState !== "pending" && v5ClaimState !== "inflight") || + ((v4 || v5 || v6) && !/^[a-f0-9-]{36}$/u.test(lines[2])) || + ((v5 || v6) && !/^[a-f0-9-]{36}$/u.test(lines[5])) || + (v6 && !/^[0-9]+$/u.test(lines[6])) || + (claimState !== "pending" && claimState !== "inflight") || !/^[a-f0-9]{64}$/u.test(lines[sessionIndex]) || !/^[A-Za-z0-9._-]+$/u.test(lines[keyIndex]) || lines[contentIndex + 1] !== "" @@ -241,10 +243,11 @@ function readWakeNotificationClaim(): WakeNotificationClaim | undefined { if (Buffer.from(message, "utf8").toString("base64") !== lines[contentIndex]) return undefined; return { pid: lines[1], - instance: v3 || v4 || v5 ? lines[2] : "", + instance: v3 || v4 || v5 || v6 ? lines[2] : "", session: lines[sessionIndex], - state: v5ClaimState, - id: v5 ? lines[5] : "", + state: claimState, + id: v5 || v6 ? lines[5] : "", + through: v6 ? lines[6] : "0", key: lines[keyIndex], content: message, }; @@ -259,12 +262,13 @@ function writeWakeNotificationClaim(claim: WakeNotificationClaim): void { writeFileSync( descriptor, [ - "fm-omp-primary-nextturn-notification-v5", + "fm-omp-primary-nextturn-notification-v6", claim.pid, claim.instance, claim.session, claim.state, claim.id, + claim.through, claim.key, Buffer.from(claim.content, "utf8").toString("base64"), "", @@ -301,6 +305,19 @@ export default function (omp: ExtensionAPI) { let notificationSession = createHash("sha256").update("unknown").digest("hex"); + const queuedWakeSequence = (): string => { + let maximum = "0"; + try { + for (const line of readFileSync(`${state}/.wake-queue`, "utf8").split("\n")) { + const sequence = line.split("\t")[1] || ""; + if (/^[0-9]+$/u.test(sequence) && (sequence.length > maximum.length || sequence.length === maximum.length && sequence > maximum)) { + maximum = sequence; + } + } + } catch {} + return maximum; + }; + const setNotificationSession = (ctx: ExtensionContext): void => { const sessionFile = ctx.sessionManager.getSessionFile() || "unknown"; notificationSession = createHash("sha256").update(sessionFile).digest("hex"); @@ -361,6 +378,7 @@ export default function (omp: ExtensionAPI) { session: notificationSession, state: "pending", id: randomUUID(), + through: queuedWakeSequence(), key, content, }); @@ -414,7 +432,7 @@ export default function (omp: ExtensionAPI) { ) { return; } - writeWakeNotificationClaim({ ...claim, state: "inflight", id: claim.id || randomUUID() }); + writeWakeNotificationClaim({ ...claim, state: "inflight", id: claim.id || randomUUID(), through: claim.through || queuedWakeSequence() }); }; omp.events?.on?.(FM_PRIMARY_WATCHER_WAKE_EVENT, (data) => { diff --git a/bin/fm-wake-drain.sh b/bin/fm-wake-drain.sh index 93927b5ef29..5a05eee986e 100755 --- a/bin/fm-wake-drain.sh +++ b/bin/fm-wake-drain.sh @@ -36,13 +36,19 @@ OMP_NOTIFICATION_CLAIM="$STATE/.omp-primary-nextturn-notification" OMP_NOTIFICATION_ACKNOWLEDGEMENT="$STATE/.omp-primary-nextturn-ack" publish_omp_notification_acknowledgement() { - local claim_id acknowledgement_tmp + local claim_id claim_through acknowledgement_tmp [ "$ACTOR" = main ] || return 0 [ -f "$OMP_NOTIFICATION_CLAIM" ] && [ ! -L "$OMP_NOTIFICATION_CLAIM" ] || return 0 - [ "$(sed -n '1p' "$OMP_NOTIFICATION_CLAIM" 2>/dev/null)" = fm-omp-primary-nextturn-notification-v5 ] || return 0 - [ "$(sed -n '5p' "$OMP_NOTIFICATION_CLAIM" 2>/dev/null)" = inflight ] || return 0 + [ "$(sed -n '1p' "$OMP_NOTIFICATION_CLAIM" 2>/dev/null)" = fm-omp-primary-nextturn-notification-v6 ] || return 0 + case "$(sed -n '5p' "$OMP_NOTIFICATION_CLAIM" 2>/dev/null)" in pending|inflight) ;; *) return 0 ;; esac claim_id=$(sed -n '6p' "$OMP_NOTIFICATION_CLAIM" 2>/dev/null) || return 1 + claim_through=$(sed -n '7p' "$OMP_NOTIFICATION_CLAIM" 2>/dev/null) || return 1 case "$claim_id" in ''|*[!a-f0-9-]*) return 0 ;; esac + case "$claim_through" in ''|*[!0-9]*) return 0 ;; esac + if [ "${#claim_through}" -gt "${#ACK_THROUGH}" ] \ + || { [ "${#claim_through}" -eq "${#ACK_THROUGH}" ] && [[ "$claim_through" > "$ACK_THROUGH" ]]; }; then + return 0 + fi acknowledgement_tmp=$(mktemp "$STATE/.omp-primary-nextturn-ack.XXXXXX") || return 1 if ! chmod 0600 "$acknowledgement_tmp" \ || ! printf '%s\n' "$claim_id" > "$acknowledgement_tmp" \ diff --git a/tests/fm-omp-primary.test.sh b/tests/fm-omp-primary.test.sh index 8dd51c63096..b8d293c1f33 100755 --- a/tests/fm-omp-primary.test.sh +++ b/tests/fm-omp-primary.test.sh @@ -1188,6 +1188,11 @@ try { throw new Error(`handling confirmation lost its recovery generation: ${handling.join(" | ")}`); } await startNextTurn(); + const acknowledgedFollowUp = dispatch.createPrimaryWatcherWake("acknowledged follow-up", "branch-fallback"); + api.events.emit(dispatch.FM_PRIMARY_WATCHER_WAKE_EVENT, acknowledgedFollowUp); + if (!acknowledgedFollowUp.accepted || notifications.length !== 7) { + throw new Error(`in-flight wake did not accept one pending follow-up: ${JSON.stringify(notifications)}`); + } writeFileSync(`${state}/.watcher-down`, "pending:handling:fixture-generation\n"); const acknowledged = spawnSync(process.env.FM_WAKE_DRAIN, ["--ack-through", "3", "--recovery-generation", "fixture-generation"], { encoding: "utf8", From bf823ad9c7a1608cdad2d9d106c3c50ff66ce097 Mon Sep 17 00:00:00 2001 From: Pranay Pratyush Date: Mon, 31 Aug 2026 18:30:12 +0530 Subject: [PATCH 09/11] no-mistakes(review): Harden OMP wake claim continuity --- .omp/extensions/fm-primary-omp.ts | 15 +++++++++---- bin/fm-wake-drain.sh | 10 ++++++--- tests/fm-omp-primary.test.sh | 17 +++++++++----- tests/fm-wake-queue.test.sh | 37 +++++++++++++++++++++++++++++++ 4 files changed, 67 insertions(+), 12 deletions(-) diff --git a/.omp/extensions/fm-primary-omp.ts b/.omp/extensions/fm-primary-omp.ts index 5aaf62cce0e..e7390d050b4 100644 --- a/.omp/extensions/fm-primary-omp.ts +++ b/.omp/extensions/fm-primary-omp.ts @@ -300,7 +300,12 @@ export default function (omp: ExtensionAPI) { publishNativeProcessIdentity(); const taskInboxDoorbell = installTaskInboxDoorbell(omp); let pendingStartupNudge = ""; - const runtime = globalThis as typeof globalThis & { firstmateOmpPrimaryNotificationInstance?: string }; + const runtime = globalThis as typeof globalThis & { + firstmateOmpPrimaryNotificationBinding?: string; + firstmateOmpPrimaryNotificationInstance?: string; + }; + const notificationBinding = randomUUID(); + runtime.firstmateOmpPrimaryNotificationBinding = notificationBinding; const notificationInstance = runtime.firstmateOmpPrimaryNotificationInstance ??= randomUUID(); let notificationSession = createHash("sha256").update("unknown").digest("hex"); @@ -412,13 +417,13 @@ export default function (omp: ExtensionAPI) { } } }; - const queueWakeNotification = (content: string, notificationKey: string): boolean => { + const queueWakeNotification = (content: string, notificationKey: string, retainOnFailure = false): boolean => { if (!claimWakeNotification(notificationKey, content)) return true; try { sendWakeNotification(content); return true; } catch (error) { - discardWakeNotification(); + if (!retainOnFailure) discardWakeNotification(); throw error; } }; @@ -436,6 +441,7 @@ export default function (omp: ExtensionAPI) { }; omp.events?.on?.(FM_PRIMARY_WATCHER_WAKE_EVENT, (data) => { + if (runtime.firstmateOmpPrimaryNotificationBinding !== notificationBinding) return; const wake = data as PrimaryWatcherWake; if ( !wake || @@ -445,7 +451,7 @@ export default function (omp: ExtensionAPI) { ) { return; } - queueWakeNotification(wake.content, wake.notificationKey); + queueWakeNotification(wake.content, wake.notificationKey, true); wake.accept(); }); @@ -515,6 +521,7 @@ export default function (omp: ExtensionAPI) { }); omp.on("message_start", (event) => { + if (runtime.firstmateOmpPrimaryNotificationBinding !== notificationBinding) return; const message = event.message as { role?: unknown; customType?: unknown; content?: unknown }; const claim = readWakeNotificationClaim(); if ( diff --git a/bin/fm-wake-drain.sh b/bin/fm-wake-drain.sh index 5a05eee986e..7335ae519fe 100755 --- a/bin/fm-wake-drain.sh +++ b/bin/fm-wake-drain.sh @@ -37,7 +37,6 @@ OMP_NOTIFICATION_ACKNOWLEDGEMENT="$STATE/.omp-primary-nextturn-ack" publish_omp_notification_acknowledgement() { local claim_id claim_through acknowledgement_tmp - [ "$ACTOR" = main ] || return 0 [ -f "$OMP_NOTIFICATION_CLAIM" ] && [ ! -L "$OMP_NOTIFICATION_CLAIM" ] || return 0 [ "$(sed -n '1p' "$OMP_NOTIFICATION_CLAIM" 2>/dev/null)" = fm-omp-primary-nextturn-notification-v6 ] || return 0 case "$(sed -n '5p' "$OMP_NOTIFICATION_CLAIM" 2>/dev/null)" in pending|inflight) ;; *) return 0 ;; esac @@ -45,8 +44,13 @@ publish_omp_notification_acknowledgement() { claim_through=$(sed -n '7p' "$OMP_NOTIFICATION_CLAIM" 2>/dev/null) || return 1 case "$claim_id" in ''|*[!a-f0-9-]*) return 0 ;; esac case "$claim_through" in ''|*[!0-9]*) return 0 ;; esac - if [ "${#claim_through}" -gt "${#ACK_THROUGH}" ] \ - || { [ "${#claim_through}" -eq "${#ACK_THROUGH}" ] && [[ "$claim_through" > "$ACK_THROUGH" ]]; }; then + if awk -F '\t' -v cutoff="$claim_through" ' + $2 ~ /^[0-9]+$/ && (length($2) < length(cutoff) || (length($2) == length(cutoff) && ("x" $2) <= ("x" cutoff))) { + outstanding=1 + exit + } + END { exit outstanding ? 0 : 1 } + ' "$FM_WAKE_QUEUE"; then return 0 fi acknowledgement_tmp=$(mktemp "$STATE/.omp-primary-nextturn-ack.XXXXXX") || return 1 diff --git a/tests/fm-omp-primary.test.sh b/tests/fm-omp-primary.test.sh index b8d293c1f33..c9477972378 100755 --- a/tests/fm-omp-primary.test.sh +++ b/tests/fm-omp-primary.test.sh @@ -1014,7 +1014,15 @@ const waitFor = async (predicate, description) => { }; const api = { zod: { object: () => ({}) }, - on(name, handler) { handlers.set(name, handler); }, + on(name, handler) { + if (name === "message_start") { + const registered = handlers.get(name) || []; + registered.push(handler); + handlers.set(name, registered); + return; + } + handlers.set(name, handler); + }, events: { on(name, handler) { eventHandlers.set(name, handler); }, emit(name, data) { eventHandlers.get(name)?.(data); }, @@ -1066,10 +1074,9 @@ const startSession = async (sessionFile) => { const startNextTurn = async () => { const notification = notifications.at(-1); if (notification) { - await handlers.get("message_start")( - { type: "message_start", message: { ...notification.message, role: "custom" } }, - {}, - ); + for (const handler of handlers.get("message_start")) { + await handler({ type: "message_start", message: { ...notification.message, role: "custom" } }, {}); + } } await handlers.get("before_agent_start")({ type: "before_agent_start" }, {}); }; diff --git a/tests/fm-wake-queue.test.sh b/tests/fm-wake-queue.test.sh index 6d64c09deef..f0d5d37c955 100755 --- a/tests/fm-wake-queue.test.sh +++ b/tests/fm-wake-queue.test.sh @@ -887,6 +887,42 @@ test_main_drain_excludes_rows_already_granted_to_branch() { pass "a main drain excludes a branch-granted row and acknowledges only its own presented rows" } +test_omp_claim_waits_for_all_scoped_rows() { + local dir state grant sequence generation + grant="$ROOT/bin/fm-wake-grant.sh" + dir=$(make_case omp-claim-scoped-rows) + state="$dir/state" + append_wake "$state" check "some-poll.check.sh" "check: some-poll" || fail "check append failed" + append_wake "$state" signal "task-a.status" "signal: task-a" || fail "signal append failed" + FM_STATE_OVERRIDE="$state" "$grant" activate "$$" omp-claim || fail "branch owner activation failed" + FM_STATE_OVERRIDE="$state" "$grant" publish omp-claim 2 || fail "branch grant publication failed" + printf '%s\n' \ + fm-omp-primary-nextturn-notification-v6 \ + 1 \ + 11111111-1111-4111-8111-111111111111 \ + aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa \ + pending \ + 22222222-2222-4222-8222-222222222222 \ + 2 \ + wake-queue \ + d2FrZQ== \ + > "$state/.omp-primary-nextturn-notification" + FM_STATE_OVERRIDE="$state" FM_SUPERVISION_ACTOR=branch "$DRAIN" > "$dir/branch.out" 2> "$dir/branch.err" \ + || fail "branch drain failed: $(cat "$dir/branch.err")" + sequence=$(sed -n 's/^WAKE_ACK_REQUIRED:.*--ack-through \([0-9][0-9]*\) --recovery-generation [A-Za-z0-9._-][A-Za-z0-9._-]*$/\1/p' "$dir/branch.err") + generation=$(sed -n 's/^WAKE_ACK_REQUIRED:.*--ack-through [0-9][0-9]* --recovery-generation \([A-Za-z0-9._-][A-Za-z0-9._-]*\)$/\1/p' "$dir/branch.err") + FM_STATE_OVERRIDE="$state" FM_SUPERVISION_ACTOR=branch "$DRAIN" --ack-through "$sequence" --recovery-generation "$generation" \ + || fail "branch acknowledgement failed" + [ ! -e "$state/.omp-primary-nextturn-ack" ] || fail "branch acknowledgement retired a claim with main-owned rows remaining" + FM_STATE_OVERRIDE="$state" "$DRAIN" > "$dir/main.out" 2> "$dir/main.err" || fail "main drain failed: $(cat "$dir/main.err")" + sequence=$(sed -n 's/^WAKE_ACK_REQUIRED:.*--ack-through \([0-9][0-9]*\) --recovery-generation [A-Za-z0-9._-][A-Za-z0-9._-]*$/\1/p' "$dir/main.err") + generation=$(sed -n 's/^WAKE_ACK_REQUIRED:.*--ack-through [0-9][0-9]* --recovery-generation \([A-Za-z0-9._-][A-Za-z0-9._-]*\)$/\1/p' "$dir/main.err") + FM_STATE_OVERRIDE="$state" "$DRAIN" --ack-through "$sequence" --recovery-generation "$generation" \ + || fail "main acknowledgement failed" + [ -e "$state/.omp-primary-nextturn-ack" ] || fail "acknowledging every claimed row did not retire the OMP notification claim" + pass "OMP notification acknowledgement waits for all scoped rows through its cutoff" +} + test_branch_owner_activation_rollback_stops_after_publication() { local dir state grant status=0 grant="$ROOT/bin/fm-wake-grant.sh" @@ -931,4 +967,5 @@ test_marker_transitions_survive_reentry_from_an_exiting_frame test_handling_confirmation_is_bounded_by_foreign_marker_lock test_branch_actor_scoped_ack_never_swallows_a_main_owned_row test_main_drain_excludes_rows_already_granted_to_branch +test_omp_claim_waits_for_all_scoped_rows test_branch_owner_activation_rollback_stops_after_publication From f9df832beb4f6159e24a0f958de03e774a8d0e19 Mon Sep 17 00:00:00 2001 From: Pranay Pratyush Date: Mon, 31 Aug 2026 18:35:56 +0530 Subject: [PATCH 10/11] no-mistakes(document): Document OMP next-turn coverage --- docs/watcher-continuity.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/watcher-continuity.md b/docs/watcher-continuity.md index d7cc79ca000..f6e93892d6e 100644 --- a/docs/watcher-continuity.md +++ b/docs/watcher-continuity.md @@ -98,7 +98,7 @@ Only the watcher process touches `state/.last-watcher-beat`; no helper process c `tests/fm-pi-watch-extension.test.sh` checks Pi's first-cycle-or-explicit-repair tool metadata and ownership-based redundant-call no-ops, then simulates actionable and empty child closes against the actual Pi and OpenCode close handlers, blocks prompt delivery to prove the successor launches first, verifies single-flight behavior, changes the session lock before close to prove ownership is rechecked, and hangs each successor arm to prove bounded fallback delivery includes the typed restoration failure. The same suite covers ordinary same-process session replacement for `/new`, `/resume`, and `/fork`, same-instance shutdown-plus-start, stale prior-generation callbacks, repeated transitions with exactly one live cycle, disappearance of the shutting-down refusal after a valid replacement activates, and terminal quit still refusing late rearm. It also covers a fresh factory bind without a prior shutdown: the superseded binding retires its arm child, cannot arm or reclaim ownership through retained session callbacks, the live binding keeps exactly one arm child, and repeated binds never accumulate more than one process-exit fallback. -`tests/fm-omp-primary.test.sh` covers OMP's binding of the same core to its native session, watcher, and shutdown surfaces, pins the input-preserving custom-steer delivery contract, pins the recovery handling handshake OMP performs before delivering that steer, and pins the single typed wake a refused handshake produces. +`tests/fm-omp-primary.test.sh` covers OMP's binding of the same core to its native session, watcher, and shutdown surfaces, pins hidden custom `nextTurn` delivery with `triggerTurn`, the pre-delivery recovery handshake, batching of one pending notification across a same-session reload, exact unacknowledged-batch replay after session or process replacement, and acknowledgement-owned claim retirement. The opt-in `tests/fm-omp-primary-live-e2e.test.sh` proves a real OMP watcher wake reaches the session while its exact pending draft remains intact. `tests/fm-watch-arm.test.sh` covers durable queue replay, real remote parent-replies ingestion into the authoritative status log, decision-only OPEN DECISIONS recovery, interrupted handling replay, generation-bound acknowledgement, a persistent live successor after recovery, a watcher close inside the handling window that must leave the printed acknowledgement valid, and the self-healing moved-generation acknowledgement that consumes its handled rows and names its remedy. `tests/fm-watch-recovery-loop.test.sh` covers the once-per-generation announcement bound with the real Pi extension against a refused handling handshake, and a handling successor that must surface a real crew event instead of going blind. From 9871ff1b1ca9bac1b987b81ee22228c6dc208c8e Mon Sep 17 00:00:00 2001 From: Pranay Pratyush Date: Tue, 1 Sep 2026 02:42:51 +0530 Subject: [PATCH 11/11] fix(omp): identify fileless notification sessions --- .omp/extensions/fm-primary-omp.ts | 4 +- tests/fm-omp-primary.test.sh | 63 ++++++++++++++++++++++--------- 2 files changed, 48 insertions(+), 19 deletions(-) diff --git a/.omp/extensions/fm-primary-omp.ts b/.omp/extensions/fm-primary-omp.ts index e7390d050b4..debc0bf7509 100644 --- a/.omp/extensions/fm-primary-omp.ts +++ b/.omp/extensions/fm-primary-omp.ts @@ -324,8 +324,8 @@ export default function (omp: ExtensionAPI) { }; const setNotificationSession = (ctx: ExtensionContext): void => { - const sessionFile = ctx.sessionManager.getSessionFile() || "unknown"; - notificationSession = createHash("sha256").update(sessionFile).digest("hex"); + const sessionIdentity = ctx.sessionManager.getSessionId(); + notificationSession = createHash("sha256").update(sessionIdentity).digest("hex"); }; const sendWakeNotification = (content: string): void => { omp.sendMessage( diff --git a/tests/fm-omp-primary.test.sh b/tests/fm-omp-primary.test.sh index c9477972378..02bd72809e9 100755 --- a/tests/fm-omp-primary.test.sh +++ b/tests/fm-omp-primary.test.sh @@ -282,7 +282,7 @@ const api = { process.argv[1] = process.env.EXTENSION; const extension = await import(`${pathToFileURL(process.env.EXTENSION).href}?fresh-native=${Date.now()}`); extension.default(api); -const context = { sessionManager: { getSessionFile: () => "" } }; +const context = { sessionManager: { getSessionFile: () => "", getSessionId: () => "fresh-session" } }; await handlers.get("session_start")({ type: "session_start" }, context); const first = await handlers.get("before_agent_start")({ type: "before_agent_start" }, {}); const second = await handlers.get("before_agent_start")({ type: "before_agent_start" }, {}); @@ -488,7 +488,12 @@ let markerLines = readFileSync(marker, "utf8").trim().split("\n"); if (markerLines.length !== 4 || markerLines[1] !== String(process.pid)) { throw new Error(`invalid OMP primary marker ${markerLines.join("|")}`); } -const extensionContext = { sessionManager: { getSessionFile: () => `${process.env.FIXTURE}/omp-session.jsonl` } }; +const extensionContext = { + sessionManager: { + getSessionFile: () => `${process.env.FIXTURE}/omp-session.jsonl`, + getSessionId: () => "native-session", + }, +}; if (existsSync(process.env.FM_OMP_TASK_DOORBELL_READY)) { throw new Error("OMP primary doorbell published readiness before session initialization"); } @@ -990,6 +995,7 @@ SH FM_WAKE_DRAIN="$ROOT/bin/fm-wake-drain.sh" \ FM_OMP_TASK_INBOX_DIR="" FM_OMP_TASK_DOORBELL_READY="" \ node --input-type=module 2>&1 <<'JS' +import { createHash } from "node:crypto"; import { spawnSync } from "node:child_process"; import { existsSync, readFileSync, writeFileSync } from "node:fs"; import { pathToFileURL } from "node:url"; @@ -1063,12 +1069,19 @@ if (deduped.status !== 0 || deduped.stdout.trim().split("\n").length !== 2) { throw new Error(`durable batch was not deduplicated: ${deduped.stderr || deduped.stdout}`); } -const firstSession = `${state}/first-session.jsonl`; -const replacementSession = `${state}/replacement-session.jsonl`; -const startSession = async (sessionFile) => { +const firstSessionId = "first-session"; +const replacementSessionId = "replacement-session"; +const notificationClaim = `${state}/.omp-primary-nextturn-notification`; +const claimSession = () => readFileSync(notificationClaim, "utf8").split("\n")[3]; +const startSession = async (sessionId) => { await handlers.get("session_start")( { type: "session_start" }, - { sessionManager: { getSessionFile: () => sessionFile } }, + { + sessionManager: { + getSessionFile: () => undefined, + getSessionId: () => sessionId, + }, + }, ); }; const startNextTurn = async () => { @@ -1088,23 +1101,27 @@ const trigger = async (count, messageCount) => { } }; const triggerCurrent = async (messageCount) => trigger(armCount(), messageCount); -const reloadExtension = async (label, sessionFile, arm = true) => { +const reloadExtension = async (label, sessionId, arm = true) => { const reloaded = await import(`${pathToFileURL(process.env.EXTENSION).href}?${label}=${Date.now()}`); reloaded.default(api); if (!tool) throw new Error(`OMP did not register its watcher arm tool after ${label}`); - await startSession(sessionFile); + await startSession(sessionId); if (arm) await tool.execute(); }; - try { - await startSession(firstSession); + await startSession(firstSessionId); await tool.execute(); await waitFor(() => armCount() === 1, "initial watcher"); await trigger(1, 1); + const firstClaimSession = claimSession(); + const expectedFirstClaimSession = createHash("sha256").update(firstSessionId).digest("hex"); if (readFileSync(queue, "utf8") !== queueRows) { throw new Error("hidden notification acknowledged durable wake rows"); } + if (firstClaimSession !== expectedFirstClaimSession) { + throw new Error(`initial claim used the wrong fileless session identity: ${firstClaimSession}`); + } const first = notifications[0]; if ( first?.message?.customType !== "firstmate-watcher-wake" || @@ -1121,18 +1138,22 @@ try { throw new Error(`branch fallback did not share the pending next-turn notification: ${JSON.stringify(notifications)}`); } - await reloadExtension("same-session-reload", firstSession); + await reloadExtension("same-session-reload", firstSessionId); await waitFor(() => armCount() >= 3, "same-session reload watcher"); await triggerCurrent(); await new Promise((resolve) => setTimeout(resolve, 40)); - if (notifications.length !== 1) { - throw new Error(`same-session reload duplicated a pending notification: ${JSON.stringify(notifications)}`); + if (notifications.length !== 1 || claimSession() !== firstClaimSession) { + throw new Error(`same-session reload duplicated or changed a pending notification: ${JSON.stringify(notifications)}`); } - await startNextTurn(); await handlers.get("session_switch")( { type: "session_switch", reason: "new" }, - { sessionManager: { getSessionFile: () => replacementSession } }, + { + sessionManager: { + getSessionFile: () => undefined, + getSessionId: () => replacementSessionId, + }, + }, ); await waitFor(() => notifications.length >= 2, "original batch replay after session switch"); if (readFileSync(queue, "utf8") !== queueRows) { @@ -1141,9 +1162,17 @@ try { if (notifications[1].message.content !== first.message.content) { throw new Error("session switch did not re-notify the exact unacknowledged durable batch"); } + const replacementClaimSession = claimSession(); + const expectedReplacementClaimSession = createHash("sha256").update(replacementSessionId).digest("hex"); + if ( + replacementClaimSession !== expectedReplacementClaimSession || + replacementClaimSession === firstClaimSession + ) { + throw new Error(`session switch did not claim the replacement fileless session: ${replacementClaimSession}`); + } delete globalThis.firstmateOmpPrimaryNotificationInstance; - await reloadExtension("process-restart", replacementSession, false); + await reloadExtension("process-restart", replacementSessionId, false); await waitFor(() => notifications.length >= 3, "original batch replay after restart"); if (readFileSync(queue, "utf8") !== queueRows) { throw new Error("process-restart replay retired durable wake rows"); @@ -1211,7 +1240,7 @@ try { const notificationsBeforeAcknowledgedSwitch = notifications.length; await handlers.get("session_switch")( { type: "session_switch", reason: "resume" }, - { sessionManager: { getSessionFile: () => `${state}/acknowledged-session.jsonl` } }, + { sessionManager: { getSessionFile: () => undefined, getSessionId: () => "acknowledged-session" } }, ); await new Promise((resolve) => setTimeout(resolve, 40)); if (notifications.length !== notificationsBeforeAcknowledgedSwitch) {