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-branch-supervision-omp.ts b/.omp/extensions/fm-branch-supervision-omp.ts
index c2e6b5eaab5..7ae9d522994 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,
@@ -849,21 +851,20 @@ ${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);
-    // 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");
+    try {
+      pi.events?.emit?.(FM_PRIMARY_WATCHER_WAKE_EVENT, wake);
+    } catch {}
+    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 ec01a0301b8..debc0bf7509 100644
--- a/.omp/extensions/fm-primary-omp.ts
+++ b/.omp/extensions/fm-primary-omp.ts
@@ -1,7 +1,18 @@
 // 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 { Buffer } from "node:buffer";
+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 {
@@ -15,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";
 
@@ -27,12 +40,25 @@ 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`;
+const notificationAcknowledgement = `${state}/.omp-primary-nextturn-ack`;
 
 type ProcessResult = {
   code: number;
   stderr: string;
 };
 
+type WakeNotificationClaim = {
+  pid: string;
+  instance: string;
+  session: string;
+  state: "pending" | "inflight";
+  id: string;
+  through: string;
+  key: string;
+  content: string;
+};
+
 function encodeOperationalInput(kind: "session-start" | "watcher" | "turn-end-guard", content: string): string {
   const result = spawnSync(operationalInputScript, ["encode", kind], {
     encoding: "utf8",
@@ -179,11 +205,255 @@ function runGuard(event: SessionStopEvent): Promise<ProcessResult> {
   });
 }
 
+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");
+  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;
+  const v4 = version === "fm-omp-primary-nextturn-notification-v4" && lines.length === 8;
+  const v5 = version === "fm-omp-primary-nextturn-notification-v5" && lines.length === 9;
+  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 = 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 || 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] !== ""
+  ) {
+    return undefined;
+  }
+  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 || v5 || v6 ? lines[2] : "",
+    session: lines[sessionIndex],
+    state: claimState,
+    id: v5 || v6 ? lines[5] : "",
+    through: v6 ? lines[6] : "0",
+    key: lines[keyIndex],
+    content: message,
+  };
+}
+
+function writeWakeNotificationClaim(claim: WakeNotificationClaim): 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-v6",
+        claim.pid,
+        claim.instance,
+        claim.session,
+        claim.state,
+        claim.id,
+        claim.through,
+        claim.key,
+        Buffer.from(claim.content, "utf8").toString("base64"),
+        "",
+      ].join("\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 = "";
+  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");
+
+  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 sessionIdentity = ctx.sessionManager.getSessionId();
+    notificationSession = createHash("sha256").update(sessionIdentity).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 discardWakeNotification = (expectedId = ""): void => {
+    const claim = readWakeNotificationClaim();
+    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 &&
+      (current.instance !== notificationInstance ||
+        current.session !== notificationSession ||
+        current.state === "pending")
+    ) {
+      return false;
+    }
+    writeWakeNotificationClaim({
+      pid: String(process.pid),
+      instance: notificationInstance,
+      session: notificationSession,
+      state: "pending",
+      id: randomUUID(),
+      through: queuedWakeSequence(),
+      key,
+      content,
+    });
+    return true;
+  };
+  const replayWakeNotification = (): void => {
+    reconcileWakeNotificationAcknowledgement();
+    const pending = readWakeNotificationClaim();
+    if (
+      !pending ||
+      (pending.instance === notificationInstance && pending.session === notificationSession)
+    ) {
+      return;
+    }
+    const replay = {
+      ...pending,
+      pid: String(process.pid),
+      instance: notificationInstance,
+      session: notificationSession,
+      state: "pending",
+      id: randomUUID(),
+    };
+    try {
+      writeWakeNotificationClaim(replay);
+      sendWakeNotification(pending.content);
+    } catch {
+      try {
+        writeWakeNotificationClaim(pending);
+      } catch {
+        // Keep the replacement claim if restoring the former process claim also fails.
+      }
+    }
+  };
+  const queueWakeNotification = (content: string, notificationKey: string, retainOnFailure = false): boolean => {
+    if (!claimWakeNotification(notificationKey, content)) return true;
+    try {
+      sendWakeNotification(content);
+      return true;
+    } catch (error) {
+      if (!retainOnFailure) 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", id: claim.id || randomUUID(), through: claim.through || queuedWakeSequence() });
+  };
+
+  omp.events?.on?.(FM_PRIMARY_WATCHER_WAKE_EVENT, (data) => {
+    if (runtime.firstmateOmpPrimaryNotificationBinding !== notificationBinding) return;
+    const wake = data as PrimaryWatcherWake;
+    if (
+      !wake ||
+      typeof wake.accept !== "function" ||
+      typeof wake.content !== "string" ||
+      typeof wake.notificationKey !== "string"
+    ) {
+      return;
+    }
+    queueWakeNotification(wake.content, wake.notificationKey, true);
+    wake.accept();
+  });
 
   // Supervision-branch dispatch handshake (docs/omp-supervision-branch.md).
   // Build one offer per ordinary actionable wake and emit it on the shared
@@ -216,18 +486,12 @@ export default function (omp: ExtensionAPI) {
     armReadyTimeoutEnv: "FM_OMP_ARM_READY_TIMEOUT_MS",
     repairToolName: "fm_watch_arm_omp",
     encodeOperationalInput,
-    sendFollowUp: async (content) => {
-      // Deliver a custom steer so OMP wakes idle sessions without touching the editable draft.
-      omp.sendMessage(
-        {
-          customType: "firstmate-watcher-wake",
-          content,
-          display: false,
-          attribution: "agent",
-          details: { kind: "watcher", runtime: "omp" },
-        },
-        { deliverAs: "steer", triggerTurn: true },
-      );
+    coalesceWakeNotification: true,
+    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.
+      queueWakeNotification(content, notificationKey);
     },
     offerWakeToBranch,
   });
@@ -238,18 +502,42 @@ export default function (omp: ExtensionAPI) {
   };
 
   omp.on("session_start", (_event, ctx) => {
+    setNotificationSession(ctx);
     taskInboxDoorbell.activate();
     watch.sessionStart();
     publishSecondmateSession(ctx);
     deliverSessionstartNudge();
+    replayWakeNotification();
   });
 
   omp.on("session_switch", (event, ctx) => {
+    setNotificationSession(ctx);
     watch.sessionShutdown();
     watch.sessionStart();
     publishSecondmateSession(ctx);
     deliverSessionstartNudge(event.reason === "new" || event.reason === "resume");
     watch.arm();
+    replayWakeNotification();
+  });
+
+  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 (
+      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 => {
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/bin/fm-primary-watch-core.ts b/bin/fm-primary-watch-core.ts
index 64b48694ace..6b8762d09c4 100644
--- a/bin/fm-primary-watch-core.ts
+++ b/bin/fm-primary-watch-core.ts
@@ -57,6 +57,7 @@ type SessionGeneration = {
   retryTimer: ReturnType<typeof setTimeout> | null;
   retryFailures: number;
   restoring: boolean;
+  notificationPending: boolean;
   seq: number;
 };
 
@@ -77,7 +78,12 @@ export type PrimaryWatchCoreOptions = {
   armReadyTimeoutEnv: string;
   repairToolName: string;
   encodeOperationalInput: (kind: "watcher", content: string) => string;
-  sendFollowUp: (content: string) => Promise<void>;
+  sendFollowUp: (content: string, notificationKey: string) => Promise<void>;
+  // 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<ArmResult>;
   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`;
@@ -337,13 +346,23 @@ export function createPrimaryWatchCore(options: PrimaryWatchCoreOptions): Primar
     };
   }
 
-  async function sendWake(owner: SessionGeneration, message: string): Promise<void> {
-    if (!generationIsLive(owner)) return;
+  async function sendWake(
+    owner: SessionGeneration,
+    message: string,
+    notificationKey = "wake-queue",
+  ): Promise<void> {
+    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, notificationKey);
+    } catch (error) {
+      if (coalesceWakeNotification) owner.notificationPending = false;
+      throw error;
+    }
   }
 
   function confirmHandlingDelivery(recovery: RecoveryHandoff): { ok: boolean; detail: string } {
@@ -404,7 +423,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;
       }
     }
@@ -414,7 +433,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 {
@@ -695,6 +714,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 +739,7 @@ export function createPrimaryWatchCore(options: PrimaryWatchCoreOptions): Primar
     arm: () => startArm(generation),
     armAndWait,
     markLoaded,
+    notificationTurnStarted,
     sessionShutdown,
     sessionStart,
   };
diff --git a/bin/fm-wake-drain.sh b/bin/fm-wake-drain.sh
index 89eaecbe4ff..7335ae519fe 100755
--- a/bin/fm-wake-drain.sh
+++ b/bin/fm-wake-drain.sh
@@ -32,6 +32,35 @@ 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 claim_through acknowledgement_tmp
+  [ -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
+  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 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
+  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 +357,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/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..f6e93892d6e 100644
--- a/docs/watcher-continuity.md
+++ b/docs/watcher-continuity.md
@@ -25,9 +25,10 @@ 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.
+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.
 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.
@@ -97,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.
diff --git a/tests/fm-omp-primary.test.sh b/tests/fm-omp-primary.test.sh
index 35b744c7650..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");
 }
@@ -593,10 +598,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 +741,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 +820,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 +841,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 +926,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 +935,327 @@ 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 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"
+  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" \
+    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";
+
+const handlers = new Map();
+const eventHandlers = 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) {
+    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); },
+  },
+  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 dispatch = await import(new URL("./lib/fm-branch-dispatch.ts", pathToFileURL(process.env.EXTENSION)).href);
+
+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 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: () => undefined,
+        getSessionId: () => sessionId,
+      },
+    },
+  );
+};
+const startNextTurn = async () => {
+  const notification = notifications.at(-1);
+  if (notification) {
+    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" }, {});
+};
+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}`);
+  }
+};
+const triggerCurrent = async (messageCount) => trigger(armCount(), messageCount);
+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(sessionId);
+  if (arm) await tool.execute();
+};
+
+try {
+  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" ||
+    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 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) {
+    throw new Error(`branch fallback did not share the pending next-turn notification: ${JSON.stringify(notifications)}`);
+  }
+
+  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 || claimSession() !== firstClaimSession) {
+    throw new Error(`same-session reload duplicated or changed a pending notification: ${JSON.stringify(notifications)}`);
+  }
+
+  await handlers.get("session_switch")(
+    { type: "session_switch", reason: "new" },
+    {
+      sessionManager: {
+        getSessionFile: () => undefined,
+        getSessionId: () => replacementSessionId,
+      },
+    },
+  );
+  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 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", 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");
+  }
+  if (notifications[2].message.content !== first.message.content) {
+    throw new Error("process restart did not re-notify the exact unacknowledged durable batch");
+  }
+  if (sendAttempts !== 3) {
+    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 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) {
+    throw new Error("failed next-turn submission retired durable wake rows");
+  }
+
+  await startNextTurn();
+  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 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)}`);
+  }
+  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(" | ")}`);
+  }
+  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",
+    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: () => undefined, getSessionId: () => "acknowledged-session" } },
+  );
+  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");
+  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 +1269,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
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