feat(bin): batch routine supervision wakes across the watcher and Pi primary

.pi/extensions/fm-primary-pi-watch.ts

@@ -54,6 +54,17 @@ type SessionGeneration = {
   retryFailures: number;
   restoring: boolean;
   seq: number;
+  wakeBatchTimer: ReturnType<typeof setTimeout> | null;
+  wakeBatch: WakeBatchItem[];
+  wakeBatchArm: ChildProcess | null;
+};
+
+type WakeRecovery = { generation: string; watcherPid: string };
+
+type WakeBatchItem = {
+  key: string;
+  messages: string[];
+  recoveries: WakeRecovery[];
 };
 
 function refreshWatchToolShell(
@@ -96,6 +107,8 @@ const armReadyTimeoutMs = positiveInteger(
   process.platform === "win32" ? 35000 : 12000,
 );
 const armRetireTimeoutMs = positiveInteger("FM_WATCH_ARM_RETIRE_TIMEOUT_MS", 1000);
+const wakeBatchLimit = positiveInteger("FM_WAKE_BATCH_LIMIT", 20);
+const wakeBatchSeconds = configuredWakeBatchSeconds();
 const repairOnlyHint = "call fm_watch_arm_pi again only after a later notification says the cycle is missing, failed, or unhealthy";
 const shuttingDownMessage = "watcher: not armed - Pi session is shutting down";
 
@@ -111,6 +124,18 @@ function positiveInteger(name: string, fallback: number): number {
   return Math.floor(value);
 }
 
+function configuredWakeBatchSeconds(): number {
+  const override = Number(process.env.FM_WAKE_BATCH_SECONDS);
+  if (Number.isFinite(override) && override > 0) return Math.floor(override);
+  try {
+    const value = Number(readFileSync(`${config}/wake-batch-seconds`, "utf8").trim());
+    if (Number.isFinite(value) && value > 0) return Math.floor(value);
+  } catch {
+    // An absent or unreadable local override keeps the safe shared default.
+  }
+  return 60;
+}
+
 function parentPid(pid: string): string {
   const result = spawnSync("ps", ["-o", "ppid=", "-p", pid], { encoding: "utf8" });
   if (result.status !== 0) return "";
@@ -194,6 +219,9 @@ function createGeneration(): SessionGeneration {
     retryFailures: 0,
     restoring: false,
     seq: 0,
+    wakeBatchTimer: null,
+    wakeBatch: [],
+    wakeBatchArm: null,
   };
 }
 
@@ -209,6 +237,10 @@ function stopGeneration(generation: SessionGeneration): void {
   generation.stopping = true;
   if (generation.retryTimer) clearTimeout(generation.retryTimer);
   generation.retryTimer = null;
+  if (generation.wakeBatchTimer) clearTimeout(generation.wakeBatchTimer);
+  generation.wakeBatchTimer = null;
+  generation.wakeBatch = [];
+  generation.wakeBatchArm = null;
   if (generation.child) generation.child.kill("SIGTERM");
   generation.child = null;
 }
@@ -250,6 +282,139 @@ export default function (pi: ExtensionAPI) {
     await pi.sendUserMessage(content, { deliverAs: "followUp" });
   }
 
+  function wakeIsUrgent(message: string): boolean {
+    const urgent = /(?:^|\n)(?:failed|blocked|needs-decision)(?:\s+\[[^\]]+\])?:|watcher: FAILED|(?:lost|no longer owns?) (?:the )?(?:session )?lock|lock ownership (?:was )?lost/i;
+    if (urgent.test(message)) return true;
+    const statusPaths = message.match(/[^\s()]+\.status\b/g) ?? [];
+    for (const statusPath of statusPaths) {
+      const path = resolve(statusPath);
+      if (!path.startsWith(`${resolve(state)}/`)) continue;
+      try {
+        const last = readFileSync(path, "utf8").trim().split(/\r?\n/).pop() ?? "";
+        if (/^(?:failed|blocked|needs-decision)(?:\s+\[[^\]]+\])?:/.test(last)) return true;
+      } catch {
+        // Missing or unreadable endpoints are surfaced by the watcher itself.
+      }
+    }
+    return false;
+  }
+
+  function wakeIdentity(message: string): string {
+    const leadingEndpoint = message.match(/^stale:\s*([^\s(]+)/)?.[1];
+    const identities = [
+      ...(message.match(/(?:^|[\s(])[^\s()]+\.(?:status|turn-ended)\b/g) ?? []),
+      ...(message.match(/\b(?:[A-Za-z0-9._-]+:)?w[A-Za-z0-9._-]+:p[A-Za-z0-9._-]+\b/g) ?? []),
+      ...(leadingEndpoint ? [leadingEndpoint] : []),
+    ].map((value) => value.trim()).sort();
+    const wakeClass = message.match(/^(signal|stale|check|heartbeat)/)?.[1] ?? "wake";
+    return identities.length > 0 ? `${wakeClass}:${[...new Set(identities)].join("|")}` : message.trim();
+  }
+
+  // <owning-arm-ended> is set only by the arm-end flush. bin/fm-watch-arm.sh prints
+  // its `watcher: started pid=...` line and then waits on that watcher, so an arm
+  // process only closes AFTER its own watcher pid has exited. Confirming a handling
+  // handshake there would always fail --handling-delivered's liveness gate and
+  // report a watcher failure that never happened, so an ended arm skips the
+  // confirmation: there is nothing left to confirm. A timer-driven flush still
+  // confirms, and a dangling arm that outlived its watcher is still retired below.
+  async function flushWakeBatch(owner: SessionGeneration, owningArmEnded = false): Promise<void> {
+    if (!generationIsLive(owner) || owner.wakeBatch.length === 0) return;
+    if (owner.wakeBatchTimer) clearTimeout(owner.wakeBatchTimer);
+    owner.wakeBatchTimer = null;
+    const items = owner.wakeBatch.splice(0, owner.wakeBatch.length);
+    owner.wakeBatchArm = null;
+    const urgentDetails: string[] = [];
+    const routineDetails: string[] = [];
+    let recovery: WakeRecovery | undefined;
+    for (const item of items) {
+      for (const message of item.messages) {
+        const detail = message.length > 800 ? `${message.slice(0, 785)} [truncated]` : message;
+        (wakeIsUrgent(message) ? urgentDetails : routineDetails).push(detail);
+      }
+      if (item.recoveries.length > 0) recovery = item.recoveries[item.recoveries.length - 1];
+    }
+    if (recovery && !owningArmEnded) {
+      const confirmed = confirmHandlingDeliveryWithRetry(owner, recovery);
+      if (!confirmed.ok) {
+        urgentDetails.push(confirmed.detail);
+        // A successor whose watcher died before confirming leaves owner.child
+        // pointing at an arm that will never supervise. Retiring it here keeps
+        // the dangling-arm recovery the pre-batching delivery path owned.
+        if (!pidAlive(recovery.watcherPid)) await retireArm(owner.child);
+      }
+    }
+    // An urgent wake is what triggered this flush, and it is appended last. Rendering
+    // in insertion order would push it past wakeBatchLimit behind the routine wakes it
+    // interrupted, so the follow-up would omit the very failure that broke the window.
+    // Urgent details are therefore rendered first and only routine details are omitted.
+    const details = [...urgentDetails, ...routineDetails];
+    const omitted = Math.max(0, details.length - wakeBatchLimit);
+    const shown = details.slice(0, wakeBatchLimit);
+    const message = shown.length === 1 && omitted === 0
+      ? shown[0]
+      : `batched ${details.length} watcher wakes:\n${shown.map((item) => `- ${item}`).join("\n")}${omitted > 0 ? `\n- ${omitted} more omitted` : ""}`;
+    await sendWake(owner, message);
+  }
+
+  // A batch must not outlive the arm that opened it while that arm is NOT being
+  // replaced: an actionable close immediately starts a successor and carries the
+  // supervision chain forward, so the batch keeps aggregating across it - that
+  // rotation is the ordinary case the wake-batch window exists for. A
+  // non-actionable close hands off to a bounded retry instead, so the batch is
+  // delivered here rather than waiting out a window under a cycle that may never
+  // come back. The durable wake queue and the recovery marker remain the only
+  // records of what was delivered and acknowledged, so a crash between this flush
+  // and its confirmation re-presents on the next drain rather than losing it.
+  function flushWakeBatchOnArmEnd(owner: SessionGeneration, armChild: ChildProcess): void {
+    if (owner.wakeBatchArm !== armChild) return;
+    if (owner.wakeBatch.length === 0) {
+      owner.wakeBatchArm = null;
+      return;
+    }
+    void flushWakeBatch(owner, true).catch(() => {
+      // Pi owns delivery errors; durable wakes remain available to the drain.
+    });
+  }
+
+  async function queueWake(owner: SessionGeneration, message: string, recovery?: WakeRecovery): Promise<void> {
+    if (!generationIsLive(owner)) return;
+    const key = wakeIdentity(message);
+    const existing = owner.wakeBatch.find((item) => item.key === key);
+    if (existing) {
+      // Identity collapse dedupes an unchanged repeat of the same wake, not a
+      // differently-typed one: two reasons for one endpoint (a paused recheck and
+      // then a vanished endpoint) must both reach the batch, or the second reason
+      // would only ever be seen by the drain.
+      if (!existing.messages.includes(message)) existing.messages.push(message);
+      if (recovery && !existing.recoveries.some((item) => item.generation === recovery.generation && item.watcherPid === recovery.watcherPid)) {
+        existing.recoveries.push(recovery);
+      }
+    } else {
+      owner.wakeBatch.push({ key, messages: [message], recoveries: recovery ? [recovery] : [] });
+    }
+    // A batch is owned by the CURRENT arm, not the one that opened it. Every wake
+    // in an aggregating batch arrives just after an actionable rotation started a
+    // successor, so pinning ownership to the opening arm would leave it pointing at
+    // a predecessor that has already closed and can never end again - and the
+    // arm-end flush would then be unreachable for exactly the multi-wake batches
+    // batching exists for. A wake queued with no live arm leaves the last live
+    // owner in place rather than orphaning the batch.
+    if (owner.child) owner.wakeBatchArm = owner.child;
+    if (wakeIsUrgent(message)) {
+      await flushWakeBatch(owner);
+      return;
+    }
+    if (!owner.wakeBatchTimer) {
+      owner.wakeBatchTimer = setTimeout(() => {
+        owner.wakeBatchTimer = null;
+        void flushWakeBatch(owner).catch(() => {
+          // Pi owns delivery errors; durable wakes remain available to the drain.
+        });
+      }, wakeBatchSeconds * 1000);
+      owner.wakeBatchTimer.unref();
+    }
+  }
+
   function confirmHandlingDelivery(recovery: { generation: string; watcherPid: string }): {
     ok: boolean;
     detail: string;
@@ -295,25 +460,14 @@ export default function (pi: ExtensionAPI) {
   async function deliverActionableWake(
     owner: SessionGeneration,
     message: string,
-    recovery?: { generation: string; watcherPid: string },
+    recovery?: WakeRecovery,
   ): Promise<void> {
     if (!generationIsLive(owner)) return;
-    if (recovery) {
-      const confirmed = confirmHandlingDeliveryWithRetry(owner, recovery);
-      if (!confirmed.ok) {
-        const watcherPid = recovery.watcherPid;
-        if (!pidAlive(watcherPid)) {
-          await retireArm(owner.child);
-        }
-        await sendWake(owner, `${message}\n\n${confirmed.detail}`);
-        return;
-      }
-    }
-    await sendWake(owner, message);
+    await queueWake(owner, message, recovery);
   }
 
   function surfaceFailure(owner: SessionGeneration, message: string): void {
-    void sendWake(owner, message).catch(() => {
+    void queueWake(owner, message).catch(() => {
       // Pi owns delivery errors; continuity restoration never waits on prompting.
     });
   }
@@ -517,13 +671,15 @@ export default function (pi: ExtensionAPI) {
         return;
       }
       if (owner.restoring) return;
+      flushWakeBatchOnArmEnd(owner, armChild);
       scheduleRetry(owner, classification.message, predecessor);
     });
     armChild.on("error", (error: Error) => {
       if (settled) return;
       settled = true;
       resolveClosed();
       settleReadiness(false);
+      flushWakeBatchOnArmEnd(owner, armChild);
       releaseChild();
       if (!generationIsLive(owner)) return;
       if (owner.restoring) return;

.pi/extensions/fm-primary-turnend-guard.ts

@@ -207,8 +207,8 @@ export default function (pi: ExtensionAPI) {
     await injectSessionstart(pi, source);
   });
 
-  // Pi's compaction equivalent. The digest is what a compacted session has just
-  // lost, so re-emitting it here is the point rather than a side effect.
+  // Pi's compaction equivalent. The source owner routes this to the compact
+  // recovery digest: ownership, actionable queue, task identities, next step.
   pi.on?.("session_compact", async () => {
     await injectSessionstart(pi, "compact");
   });

AGENTS.md

@@ -77,6 +77,7 @@ config/cmux-socket-password  optional cmux control-socket password; LOCAL, gitig
 config/wedge-alarm  optional away-mode wedge-alarm active-alert directives; LOCAL, gitignored; absent means auto (macOS Notification Center when available); see docs/wedge-alarm.md
 config/watched-tools.json  optional list of the tools this home depends on, read by the update check armed with bin/fm-tool-update-check.sh; LOCAL, gitignored, firstmate-maintained but human-editable, and NOT inherited by secondmate homes; see docs/configuration.md "Watched tool updates"
 config/attended-routine-status-absorb  optional "off" opt-out from default-on attended absorption of routine supervision wakes; LOCAL, gitignored; see docs/configuration.md "Attended routine status absorption"
+config/wake-batch-seconds  optional positive-integer Pi watcher follow-up aggregation window; LOCAL, gitignored, default 60 seconds; see docs/configuration.md "Pi watcher wake batching"
 config/x-mode.env    generated Relay watcher cadence; LOCAL, gitignored; source before arming watcher when present
 data/                personal fleet records; LOCAL, gitignored as a whole
   backlog.md         task queue, dependencies, history
@@ -123,11 +124,12 @@ state/               runtime records and signals; gitignored
   .<id>.open-decisions-cursor  per-task byte cursor and folded open-decision set bounding the OPEN DECISIONS scan's cost to new status-log appends; written only by fm-classify-lib.sh's status_open_decisions_incremental, removed by teardown, safe to delete (forces one full re-fold)
   .status-presentation-cursor .status-presentation-lock  fleet-wide per-task status identity/byte-offset manifest and serialization lock preventing already-presented status lines from being replayed as new; owned by fm-classify-lib.sh, with each task's row retired by teardown
   .status-absorbed-<id>  per-task receipt binding the status-file identity and byte endpoint the attended watcher absorbed instead of waking on; owned by fm-classify-lib.sh, retired once the drain presents those bytes, removed by teardown; never touch
+  .open-decisions-presentation  fleet-wide receipt of the last OPEN DECISIONS block a reader was actually shown; lets an unchanged repeat collapse to a count marker while session-recovery and mechanical drains leave it unspent; owned by fm-wake-drain.sh, safe to delete (forces one full re-print)
   .afk               durable away-mode flag; present = sub-supervisor may inject escalations (set by /afk, cleared on user return)
   .watch.lock .wake-queue.lock watcher singleton and queue serialization locks
   .claude-autoarm.lock .claude-autoarm-epoch .claude-autoarm-failure-notified .claude-autoarm-failure-alarmed .turnend-claude-blocks .turnend-claude-blocks.lock   Claude Stop auto-arm single-flight, epoch, failure-episode, attended-alarm, guard-budget, and budget-lock records; never touch
   .cursor-park-owner .cursor-park-owner.lock .turnend-cursor-blocks   Cursor stop-hook owner record, publication and commit lock, and bounded repair-nag budget; never touch
-  .hash-* .count-* .stale-* .stale-since-* .paused-* .wedge-escalations-* .writing-* .seen-* .hb-surfaced-* .last-* .heartbeat-streak   watcher internals; never touch
+  .hash-* .count-* .stale-* .stale-since-* .paused-* .endpoint-missing-* .wedge-escalations-* .writing-* .seen-* .hb-surfaced-* .last-* .heartbeat-streak   watcher internals; never touch
   .watch-triage.log  watcher's absorbed-wake debug log (size-capped); never relied on, safe to delete
   .last-watcher-beat watcher liveness beacon, touched every poll (including while absorbing benign wakes); guard scripts read it
   .subsuper-* .supervise-daemon.*   sub-supervisor internals; never touch
@@ -165,6 +167,7 @@ When that section reports its checks still in progress it names exactly what is
 3. **Wake queue** - when locked, presents the durable wake queue and prints the raw records prominently as this turn's first work queue; a clearly labeled status-event annotation may follow a valid `signal` record and includes every status line still unread at the presentation cursor, but never replaces the raw record or current-state reconciliation, and a lapsed watcher chain still surfaces here via the same guard alarm.
    Presented records remain durable until the handling turn runs the generation-bound acknowledgement printed by the drain.
    Every locked drain also prints a bounded fleet-wide `OPEN DECISIONS` section when durable decision records remain open, including when the queue itself is empty; reconcile those entries before continuing.
+   A later drain in the same session may print `OPEN DECISIONS: unchanged, N open` instead: that means the block you were already shown still stands unchanged, and it is just as actionable; a recovery drain (session start, `/clear`, compaction, away-mode return, or a drain that handles a watcher-down recovery episode) always prints the full block again.
    The same drain prints every still-unread `note:` line, pending-reply resolution, and routine status line the attended watcher absorbed instead of waking on, since the last presentation, in an unbounded `UNREAD STATUS` section, so an answer buried under a later routine line is not dropped; those lines are not re-printed after that presentation.
    It also prints a bounded `RECORD DIVERGENCE` section naming every captain call the status log reads as resolved while its backlog task is still held; nothing is closed for you, and `captain-hold-lifecycle` owns the reconciliation.
    When the lock could not be acquired and verified, the queue is left untouched because no session mutation is authorized, and the guard's tangle/watcher-liveness alarms still print in read-only advisory mode without drain, supervision repair, or checkout repair commands.