diff --git a/.agents/skills/afk/SKILL.md b/.agents/skills/afk/SKILL.md
index 2b68f29ed9..57422fe535 100644
--- a/.agents/skills/afk/SKILL.md
+++ b/.agents/skills/afk/SKILL.md
@@ -51,6 +51,9 @@ batched digest rather than per-wake injections.
 
 3. **Do not separately arm `fm-watch.sh`.** The daemon manages the watcher as
    its child; the singleton lock no-ops a stray arm harmlessly.
+   On Pi and pi-signed, the loaded watcher extension observes the durable flag,
+   yields its exact attached arm cycle, and resumes one ordinary cycle after
+   return; no watcher command or Pi restart belongs in this lifecycle.
 
 4. **Acknowledge** in `AGENTS.md` section 9 language: "Captain, away mode is active; I will batch routine updates and surface only decisions, failures, credentials, or review-ready work until you return."
 
@@ -151,6 +154,8 @@ It self-handles the routine majority without consuming a firstmate turn.
 Captain-relevant events, plus a bounded recheck of a declared external wait that remains idle, escalate to firstmate's context as one pre-read, single-line, batched digest.
 The classification predicates (the captain-relevant verb set, declared-pause vocabulary, signal/stale tests, and fleet-scan) live in the shared `bin/fm-classify-lib.sh`, the same library the always-on watcher uses for its own triage when afk is off, so the two modes apply one identical policy.
 While `state/.afk` exists the daemon owns the watcher, so the watcher reverts to one-shot and lets the daemon do the triage - the two never run their triage at the same time.
+The daemon classifies unseen logical wakes from durable queue state without consuming them, closing the Pi handoff interval while preserving `bin/fm-wake-drain.sh` as the sole queue consumer.
+`docs/watcher-continuity.md` owns the exact transfer and deduplication contract.
 
 Classify each wake this way:
 
@@ -229,7 +234,8 @@ the operational prefix lets firstmate distinguish it from a real captain message
 
 ## Stale-artifact lifecycle
 
-Treat `state/.subsuper-escalations`, its `.since` sidecar, and `state/.subsuper-inject-wedged` as session-scoped delivery artifacts, not as the durable work record.
+Treat the daemon's buffered escalation, queue-classification cursor, crash-pending record, check-wake deduplication ledger, and wedge marker as session-scoped delivery artifacts, not as the durable work record.
+Their exact paths and cleanup mechanics remain owned by the producing scripts.
 Always enter through `bin/fm-afk-launch.sh`, which clears prior-session artifacts only for a fresh entry and preserves the current session's buffer on refresh.
 Always exit through `bin/fm-afk-launch.sh stop`, which keeps `state/.afk` present through the daemon's shutdown flush and clears it last.
 `docs/herdr-backend.md` "Away-mode supervisor support" owns the current mechanism, and `docs/verification/runtime-backends.md` "Away-mode transport" owns active evidence.
diff --git a/.pi/extensions/fm-primary-pi-watch.ts b/.pi/extensions/fm-primary-pi-watch.ts
index 923ec6c310..d93a38abe2 100644
--- a/.pi/extensions/fm-primary-pi-watch.ts
+++ b/.pi/extensions/fm-primary-pi-watch.ts
@@ -9,8 +9,8 @@
 // quit leaves the final generation stopped so late callbacks cannot rearm. Stale
 // callbacks from a prior generation are no-ops against the active replacement.
 import { spawn, spawnSync, type ChildProcess } from "node:child_process";
-import { createHash } from "node:crypto";
-import { mkdirSync, readFileSync, writeFileSync } from "node:fs";
+import { createHash, randomBytes } from "node:crypto";
+import { existsSync, mkdirSync, readFileSync, unlinkSync, writeFileSync } from "node:fs";
 import { dirname, resolve } from "node:path";
 import { fileURLToPath } from "node:url";
 import type { ExtensionAPI, Theme } from "@earendil-works/pi-coding-agent";
@@ -21,7 +21,10 @@ import {
   calmTranscriptClassIsVisible,
   FIRSTMATE_CALM_PRESENTATION_EVENT,
 } from "./lib/fm-calm-visibility.ts";
-import { encodeFirstmateOperationalInput } from "./lib/fm-operational-input.ts";
+import {
+  classifyFirstmateCurrentOperationalText,
+  encodeFirstmateOperationalInput,
+} from "./lib/fm-operational-input.ts";
 
 type ArmResult = {
   ok: boolean;
@@ -51,6 +54,9 @@ type SessionGeneration = {
   stopping: boolean;
   child: ChildProcess | null;
   retryTimer: ReturnType<typeof setTimeout> | null;
+  awayMonitor: ReturnType<typeof setInterval> | null;
+  awayResumePending: boolean;
+  awayResumePredecessor: string;
   retryFailures: number;
   restoring: boolean;
   seq: number;
@@ -83,11 +89,14 @@ const fmRoot = process.env.FM_ROOT_OVERRIDE || root;
 const state = process.env.FM_STATE_OVERRIDE || `${fmHome}/state`;
 const config = process.env.FM_CONFIG_OVERRIDE || `${fmHome}/config`;
 const armScript = `${fmRoot}/bin/fm-watch-arm.sh`;
+const watchScript = `${fmRoot}/bin/fm-watch.sh`;
+const armTreeRetireScript = `${fmRoot}/bin/fm-pi-arm-tree-retire.sh`;
 const marker = `${state}/.pi-watch-extension-loaded`;
 const extensionVersion = `sha256:${createHash("sha256").update(readFileSync(extensionFile)).digest("hex")}`;
 const retryBaseMs = positiveInteger("FM_WATCH_REARM_RETRY_BASE_MS", 250);
 const retryMaxMs = positiveInteger("FM_WATCH_REARM_RETRY_MAX_MS", 4000);
 const retryLimit = positiveInteger("FM_WATCH_REARM_RETRY_LIMIT", 5);
+const awayHandoffPollMs = positiveInteger("FM_PI_AFK_HANDOFF_POLL_MS", 50);
 // 35s on Windows so the budget stays above arm's MSYS confirm default (30s in
 // bin/fm-watch-arm.sh): a slow but successful Git Bash cold start must not be
 // SIGTERMed mid-confirmation. Conditioned on win32 so other platforms keep 12s.
@@ -96,14 +105,22 @@ const armReadyTimeoutMs = positiveInteger(
   process.platform === "win32" ? 35000 : 12000,
 );
 const armRetireTimeoutMs = positiveInteger("FM_WATCH_ARM_RETIRE_TIMEOUT_MS", 1000);
+const armRetireRetryMs = positiveInteger("FM_PI_ARM_RETIRE_RETRY_MS", 50);
 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";
+const awayModeMessage = "watcher: stood down - away mode owns supervision; Pi will resume automatically when away mode ends";
 
 let nextGenerationId = 0;
 let activeGeneration: SessionGeneration | null = null;
 const armReadiness = new WeakMap<ChildProcess, Promise<boolean>>();
 const armClose = new WeakMap<ChildProcess, Promise<void>>();
 const armRecovery = new WeakMap<ChildProcess, { generation: string; watcherPid: string }>();
+const armTreeTokens = new WeakMap<ChildProcess, string>();
+const awayYieldedArms = new WeakSet<ChildProcess>();
+const retainedArmTrees = new Set<ChildProcess>();
+const armRetirementTimers = new Map<ChildProcess, ReturnType<typeof setTimeout>>();
+const armRetirementConfirmedHandlers = new WeakMap<ChildProcess, () => void>();
+const armRetirementWaiters = new Map<SessionGeneration, () => void>();
 
 function positiveInteger(name: string, fallback: number): number {
   const value = Number(process.env[name]);
@@ -143,6 +160,83 @@ function lockOwnership(): LockOwnership {
   return pidAlive(lockPid) ? "other" : "missing";
 }
 
+function awayModeActive(): boolean {
+  return existsSync(`${state}/.afk`);
+}
+
+function clearArmTreePidFile(armChild: ChildProcess): void {
+  if (process.platform !== "win32") return;
+  const token = armTreeTokens.get(armChild);
+  if (!token) return;
+  try {
+    unlinkSync(`${state}/.pi-arm-wrapper-${token}.pid`);
+  } catch {
+  }
+}
+
+function armTreeIdentityPublished(armChild: ChildProcess): boolean {
+  if (process.platform !== "win32") return false;
+  const token = armTreeTokens.get(armChild);
+  if (!token) return false;
+  return existsSync(`${state}/.pi-arm-wrapper-${token}.pid`) || existsSync(`${state}/.pi-arm-retirement-${token}`);
+}
+
+function terminateArmTree(armChild: ChildProcess): boolean {
+  const pid = armChild.pid;
+  if (process.platform === "win32" && pid) {
+    const token = armTreeTokens.get(armChild);
+    if (!token) return false;
+    const result = spawnSync("bash", [armTreeRetireScript, String(pid), token, state, watchScript, fmHome], {
+      stdio: "ignore",
+      windowsHide: true,
+    });
+    return result.status === 0;
+  }
+  if (process.platform !== "win32" && pid) {
+    try {
+      // The arm is spawned as a dedicated process-group leader below. Signal
+      // only that exact group so bash's trap and its watcher child retire
+      // together, without scanning processes or reaching a sibling home.
+      process.kill(-pid, "SIGTERM");
+      return true;
+    } catch {
+      // If the group already disappeared, ChildProcess.kill provides the
+      // ordinary exact-child result and preserves the existing race behavior.
+    }
+  }
+  return armChild.kill("SIGTERM");
+}
+
+function completeArmTreeRetirement(armChild: ChildProcess): void {
+  const timer = armRetirementTimers.get(armChild);
+  if (timer) clearTimeout(timer);
+  armRetirementTimers.delete(armChild);
+  retainedArmTrees.delete(armChild);
+  clearArmTreePidFile(armChild);
+  armRetirementConfirmedHandlers.get(armChild)?.();
+  if (retainedArmTrees.size !== 0) return;
+  const waiters = [...armRetirementWaiters.values()];
+  armRetirementWaiters.clear();
+  for (const resume of waiters) resume();
+}
+
+function requestArmTreeRetirement(armChild: ChildProcess): boolean {
+  if (terminateArmTree(armChild)) {
+    completeArmTreeRetirement(armChild);
+    return true;
+  }
+  if (process.platform !== "win32") return false;
+  retainedArmTrees.add(armChild);
+  if (!armRetirementTimers.has(armChild)) {
+    const timer = setTimeout(() => {
+      armRetirementTimers.delete(armChild);
+      requestArmTreeRetirement(armChild);
+    }, armRetireRetryMs);
+    armRetirementTimers.set(armChild, timer);
+  }
+  return false;
+}
+
 function markLoaded(): void {
   if (lockOwnership() === "other") return;
   mkdirSync(state, { recursive: true });
@@ -191,6 +285,9 @@ function createGeneration(): SessionGeneration {
     stopping: false,
     child: null,
     retryTimer: null,
+    awayMonitor: null,
+    awayResumePending: false,
+    awayResumePredecessor: "",
     retryFailures: 0,
     restoring: false,
     seq: 0,
@@ -209,8 +306,16 @@ function stopGeneration(generation: SessionGeneration): void {
   generation.stopping = true;
   if (generation.retryTimer) clearTimeout(generation.retryTimer);
   generation.retryTimer = null;
-  if (generation.child) generation.child.kill("SIGTERM");
-  generation.child = null;
+  if (generation.awayMonitor) clearInterval(generation.awayMonitor);
+  generation.awayMonitor = null;
+  generation.awayResumePending = false;
+  generation.awayResumePredecessor = "";
+  armRetirementWaiters.delete(generation);
+  if (generation.child) {
+    const armChild = generation.child;
+    const retired = requestArmTreeRetirement(armChild);
+    if ((process.platform !== "win32" || retired) && generation.child === armChild) generation.child = null;
+  }
 }
 
 const cleanupOnProcessExit = () => {
@@ -238,16 +343,70 @@ export default function (pi: ExtensionAPI) {
     !calmPresentation.stockExportRendering &&
     !calmTranscriptClassIsVisible(itemClass);
 
+  function yieldToAway(owner: SessionGeneration, predecessorArmPid = ""): void {
+    if (!generationIsLive(owner)) return;
+    owner.awayResumePending = true;
+    if (predecessorArmPid) owner.awayResumePredecessor = predecessorArmPid;
+    if (owner.retryTimer) clearTimeout(owner.retryTimer);
+    owner.retryTimer = null;
+    const armChild = owner.child;
+    if (!armChild || awayYieldedArms.has(armChild)) return;
+    owner.awayResumePredecessor = String(armChild.pid ?? owner.awayResumePredecessor);
+    // This exact ChildProcess is the extension's home-scoped arm. The arm owns
+    // and retires its watcher child on TERM, so no process scan or broad kill is
+    // needed and another Firstmate home cannot be affected.
+    const retired = requestArmTreeRetirement(armChild);
+    if (process.platform === "win32" || retired) awayYieldedArms.add(armChild);
+  }
+
+  function resumeAfterAway(owner: SessionGeneration): void {
+    if (!generationIsLive(owner) || awayModeActive() || !owner.awayResumePending) return;
+    if (owner.child || owner.retryTimer || owner.restoring) return;
+    const predecessor = owner.awayResumePredecessor;
+    owner.awayResumePending = false;
+    owner.awayResumePredecessor = "";
+    const result = startArm(owner, predecessor);
+    if (result.message === awayModeMessage) return;
+    if (!result.ok) {
+      scheduleRetry(owner, `watcher: FAILED - Pi extension could not resume continuity after away mode\n${result.message}`, predecessor);
+    }
+  }
+
+  function reconcileAwayOwnership(owner: SessionGeneration): void {
+    if (!generationIsLive(owner)) return;
+    if (awayModeActive()) {
+      yieldToAway(owner);
+      return;
+    }
+    resumeAfterAway(owner);
+  }
+
+  function startAwayMonitor(owner: SessionGeneration): void {
+    if (!generationIsLive(owner) || owner.awayMonitor) return;
+    reconcileAwayOwnership(owner);
+    const monitor = setInterval(() => reconcileAwayOwnership(owner), awayHandoffPollMs);
+    monitor.unref();
+    owner.awayMonitor = monitor;
+  }
+
   async function sendWake(
     owner: SessionGeneration,
     message: string,
     recovery?: { generation: string; watcherPid: string },
   ): Promise<void> {
     if (!generationIsLive(owner)) return;
+    if (awayModeActive()) {
+      yieldToAway(owner);
+      return;
+    }
     const content = encodeFirstmateOperationalInput(
       "watcher",
       `FIRSTMATE WATCHER WAKE: ${message}\n\nRun bin/fm-wake-drain.sh first and handle the queued wake. Watcher continuity is extension-owned.`,
     );
+    if (awayModeActive()) {
+      yieldToAway(owner);
+      return;
+    }
     await pi.sendUserMessage(content, { deliverAs: "followUp" });
     if (recovery) {
       const result = spawnSync(
@@ -294,7 +453,7 @@ export default function (pi: ExtensionAPI) {
 
   async function retireArm(armChild: ChildProcess | null): Promise<boolean> {
     if (!armChild) return true;
-    armChild.kill("SIGTERM");
+    requestArmTreeRetirement(armChild);
     const closed = armClose.get(armChild);
     if (!closed) return false;
     return new Promise((resolveRetired) => {
@@ -314,6 +473,10 @@ export default function (pi: ExtensionAPI) {
     let failure = "";
     for (let attempt = 0; attempt <= retryLimit; attempt += 1) {
       if (!generationIsLive(owner)) return { failure: "" };
+      if (awayModeActive()) {
+        yieldToAway(owner, predecessorArmPid);
+        return { failure: "" };
+      }
       const replacement = startArm(owner, predecessorArmPid);
       const successorChild = owner.child;
       if (replacement.ok && successorChild && await waitForReadiness(successorChild)) {
@@ -340,6 +503,10 @@ export default function (pi: ExtensionAPI) {
 
   function scheduleRetry(owner: SessionGeneration, message: string, predecessorArmPid: string): void {
     if (!generationIsLive(owner) || owner.child || owner.retryTimer) return;
+    if (awayModeActive()) {
+      yieldToAway(owner, predecessorArmPid);
+      return;
+    }
     const ownership = lockOwnership();
     if (ownership !== "owned") {
       surfaceFailure(owner, `watcher: FAILED - Pi extension cannot restore continuity because this session no longer owns the lock\n${message}`);
@@ -353,6 +520,10 @@ export default function (pi: ExtensionAPI) {
     const timer = setTimeout(() => {
       if (owner.retryTimer === timer) owner.retryTimer = null;
       if (!generationIsLive(owner)) return;
+      if (awayModeActive()) {
+        yieldToAway(owner, predecessorArmPid);
+        return;
+      }
       const result = startArm(owner, predecessorArmPid);
       if (!result.ok) {
         surfaceFailure(owner, `watcher: FAILED - Pi extension could not launch a continuity retry\n${result.message}`);
@@ -364,6 +535,23 @@ export default function (pi: ExtensionAPI) {
 
   function startArm(owner: SessionGeneration, predecessorArmPid = ""): ArmResult {
     if (!generationIsLive(owner)) return { ok: false, message: shuttingDownMessage };
+    if (awayModeActive()) {
+      yieldToAway(owner, predecessorArmPid);
+      return { ok: true, message: awayModeMessage };
+    }
+    if (process.platform === "win32" && retainedArmTrees.size !== 0) {
+      armRetirementWaiters.set(owner, () => {
+        if (!generationIsLive(owner)) return;
+        const result = startArm(owner, predecessorArmPid);
+        if (!result.ok) {
+          scheduleRetry(owner, `watcher: FAILED - Pi extension could not resume after exact predecessor retirement\n${result.message}`, predecessorArmPid);
+        }
+      });
+      return {
+        ok: true,
+        message: "watcher: waiting - exact prior Windows arm retirement is still converging; automatic re-arm pending",
+      };
+    }
     const ownership = lockOwnership();
     if (ownership === "other") return { ok: false, message: "watcher: read-only - session lock is held by another firstmate session" };
     if (ownership === "missing") {
@@ -386,6 +574,10 @@ export default function (pi: ExtensionAPI) {
       };
     }
     const id = ++owner.seq;
+    const armTreeToken = randomBytes(24).toString("hex");
+    const armLaunch = process.platform === "win32"
+      ? "wrapper_pid=${BASHPID:-$$}; wrapper_pid_file=${FM_PI_ARM_WRAPPER_PID_FILE:?}; wrapper_pid_tmp=$wrapper_pid_file.$wrapper_pid; printf '%s\\n' \"$wrapper_pid\" > \"$wrapper_pid_tmp\" && mv -f \"$wrapper_pid_tmp\" \"$wrapper_pid_file\" || { rm -f \"$wrapper_pid_tmp\"; exit 125; }; config_dir=\"${FM_CONFIG_OVERRIDE:-$FM_HOME/config}\"; [ -f \"$config_dir/x-mode.env\" ] && . \"$config_dir/x-mode.env\"; exec \"$FM_WATCH_ARM_SCRIPT\" --restart"
+      : "config_dir=\"${FM_CONFIG_OVERRIDE:-$FM_HOME/config}\"; [ -f \"$config_dir/x-mode.env\" ] && . \"$config_dir/x-mode.env\"; exec \"$FM_WATCH_ARM_SCRIPT\" --restart";
     const env = {
       ...process.env,
       FM_HOME: fmHome,
@@ -393,16 +585,21 @@ export default function (pi: ExtensionAPI) {
       FM_CONFIG_OVERRIDE: config,
       FM_WATCH_ARM_SCRIPT: armScript,
       FM_WATCH_PREDECESSOR_ARM_PID: predecessorArmPid,
+      FM_PI_ARM_TREE_TOKEN: armTreeToken,
+      ...(process.platform === "win32" ? { FM_PI_ARM_WRAPPER_PID_FILE: `${state}/.pi-arm-wrapper-${armTreeToken}.pid` } : {}),
     };
-    const armChild = spawn("bash", ["-lc", "config_dir=\"${FM_CONFIG_OVERRIDE:-$FM_HOME/config}\"; [ -f \"$config_dir/x-mode.env\" ] && . \"$config_dir/x-mode.env\"; exec \"$FM_WATCH_ARM_SCRIPT\" --restart"], {
+    const armChild = spawn("bash", ["-lc", armLaunch], {
       cwd: fmRoot,
       env,
+      detached: process.platform !== "win32",
       stdio: ["ignore", "pipe", "pipe"],
     });
+    armTreeTokens.set(armChild, armTreeToken);
     owner.child = armChild;
     let stdout = "";
     let stderr = "";
     let settled = false;
+    let finishSettledArm: (() => void) | null = null;
     let readinessSettled = false;
     let resolveReadiness: (ready: boolean) => void = () => {};
     let resolveClosed: () => void = () => {};
@@ -430,6 +627,13 @@ export default function (pi: ExtensionAPI) {
     const releaseChild = (): void => {
       if (owner.child === armChild) owner.child = null;
     };
+    const finishArmEvent = (): void => {
+      if (!finishSettledArm || retainedArmTrees.has(armChild)) return;
+      const finish = finishSettledArm;
+      finishSettledArm = null;
+      finish();
+    };
+    armRetirementConfirmedHandlers.set(armChild, finishArmEvent);
     armChild.stdout.on("data", (chunk: Buffer) => {
       stdout += chunk.toString();
       observeEstablishedArm();
@@ -441,37 +645,76 @@ export default function (pi: ExtensionAPI) {
     armChild.on("close", (code: number | null, signal: NodeJS.Signals | null) => {
       if (settled) return;
       settled = true;
-      resolveClosed();
-      settleReadiness(false);
-      releaseChild();
-      if (!generationIsLive(owner)) return;
-      const classification = classifyClose(stdout, stderr, code, signal);
-      const predecessor = String(armChild.pid ?? "");
-      if (classification.kind === "actionable") {
-        owner.retryFailures = 0;
-        owner.restoring = true;
-        void (async () => {
-          const restoration = await restoreAfterActionableClose(owner, predecessor);
-          if (generationIsLive(owner)) owner.restoring = false;
-          if (!generationIsLive(owner)) return;
-          const message = restoration.failure ? `${classification.message}\n\n${restoration.failure}` : classification.message;
-          await sendWake(owner, message, restoration.recovery);
-        })().catch(() => {
-        });
-        return;
+      finishSettledArm = () => {
+        clearArmTreePidFile(armChild);
+        resolveClosed();
+        settleReadiness(false);
+        const yieldedToAway = awayYieldedArms.delete(armChild);
+        releaseChild();
+        if (!generationIsLive(owner)) return;
+        const classification = classifyClose(stdout, stderr, code, signal);
+        const predecessor = String(armChild.pid ?? "");
+        if (yieldedToAway || awayModeActive()) {
+          owner.retryFailures = 0;
+          owner.restoring = false;
+          owner.awayResumePending = true;
+          owner.awayResumePredecessor = predecessor;
+          if (awayModeActive()) yieldToAway(owner, predecessor);
+          else resumeAfterAway(owner);
+          return;
+        }
+        if (classification.kind === "actionable") {
+          owner.retryFailures = 0;
+          owner.restoring = true;
+          void (async () => {
+            const restoration = await restoreAfterActionableClose(owner, predecessor);
+            if (generationIsLive(owner)) owner.restoring = false;
+            if (!generationIsLive(owner)) return;
+            if (awayModeActive() || owner.awayResumePending) {
+              yieldToAway(owner, predecessor);
+              return;
+            }
+            const message = restoration.failure ? `${classification.message}\n\n${restoration.failure}` : classification.message;
+            await sendWake(owner, message, restoration.recovery);
+          })().catch(() => {
+          });
+          return;
+        }
+        if (owner.restoring) return;
+        scheduleRetry(owner, classification.message, predecessor);
+      };
+      if (retainedArmTrees.has(armChild) && !armTreeIdentityPublished(armChild)) {
+        completeArmTreeRetirement(armChild);
+      } else {
+        finishArmEvent();
       }
-      if (owner.restoring) return;
-      scheduleRetry(owner, classification.message, predecessor);
     });
     armChild.on("error", (error: Error) => {
       if (settled) return;
       settled = true;
-      resolveClosed();
-      settleReadiness(false);
-      releaseChild();
-      if (!generationIsLive(owner)) return;
-      if (owner.restoring) return;
-      scheduleRetry(owner, `watcher: FAILED - Pi extension arm child ${id} failed: ${error.message}`, String(armChild.pid ?? ""));
+      finishSettledArm = () => {
+        clearArmTreePidFile(armChild);
+        resolveClosed();
+        settleReadiness(false);
+        releaseChild();
+        if (!generationIsLive(owner)) return;
+        if (awayYieldedArms.delete(armChild) || awayModeActive()) {
+          owner.retryFailures = 0;
+          owner.restoring = false;
+          owner.awayResumePending = true;
+          owner.awayResumePredecessor = String(armChild.pid ?? "");
+          if (awayModeActive()) yieldToAway(owner, owner.awayResumePredecessor);
+          else resumeAfterAway(owner);
+          return;
+        }
+        if (owner.restoring) return;
+        scheduleRetry(owner, `watcher: FAILED - Pi extension arm child ${id} failed: ${error.message}`, String(armChild.pid ?? ""));
+      };
+      if (retainedArmTrees.has(armChild) && !armTreeIdentityPublished(armChild)) {
+        completeArmTreeRetirement(armChild);
+      } else {
+        finishArmEvent();
+      }
     });
     return {
       ok: true,
@@ -483,11 +726,24 @@ export default function (pi: ExtensionAPI) {
     if (generation.stopping) generation = createGeneration();
     activateGeneration(generation);
     markLoaded();
+    startAwayMonitor(generation);
+    if (process.platform === "win32" && retainedArmTrees.size !== 0) startArm(generation);
   });
   pi.on?.("session_shutdown", () => {
     stopGeneration(generation);
   });
 
+  // A watcher follow-up can already be queued in Pi when AFK is entered. Pi's
+  // input event is the last public pre-agent boundary, so handle that typed
+  // extension message without starting a model turn. Marked away-supervisor
+  // escalations are a different operational kind and continue normally.
+  pi.on?.("input", (event) => {
+    if (event.source !== "extension" || !awayModeActive()) return { action: "continue" };
+    if (classifyFirstmateCurrentOperationalText(event.text)?.trim() !== "watcher") return { action: "continue" };
+    yieldToAway(generation);
+    return { action: "handled" };
+  });
+
   pi.registerCommand?.("fm-watch-arm-pi", {
     description: "Arm firstmate watcher supervision through the Pi extension instead of foreground bash.",
     handler: async (_args, ctx) => {
diff --git a/.pi/extensions/fm-primary-turnend-guard.ts b/.pi/extensions/fm-primary-turnend-guard.ts
index 1b2a3ec39a..33e35373cc 100644
--- a/.pi/extensions/fm-primary-turnend-guard.ts
+++ b/.pi/extensions/fm-primary-turnend-guard.ts
@@ -226,7 +226,21 @@ export default function (pi: ExtensionAPI) {
     return { block: true, reason: result.stderr.trim() || "denied by the watcher-arm PreToolUse seatbelt" };
   });
 
+  // A guard follow-up queued just before AFK entry must not become an away-mode
+  // model turn. Actionable away-supervisor messages carry a different kind and
+  // remain deliverable.
+  pi.on("input", (event) => {
+    if (event.source !== "extension" || !existsSync(`${state}/.afk`)) return { action: "continue" };
+    return classifyFirstmateCurrentOperationalText(event.text)?.trim() === "turn-end-guard"
+      ? { action: "handled" }
+      : { action: "continue" };
+  });
+
   pi.on("agent_settled", async () => {
+    if (existsSync(`${state}/.afk`)) {
+      guardFollowupActive = false;
+      return;
+    }
     if (guardFollowupActive) {
       guardFollowupActive = false;
       return;
diff --git a/bin/fm-afk-launch.sh b/bin/fm-afk-launch.sh
index 5df2a9d991..477675d7a7 100755
--- a/bin/fm-afk-launch.sh
+++ b/bin/fm-afk-launch.sh
@@ -360,11 +360,12 @@ fm_afk_launch_restore_backup() {  # <backup> <had-afk>
   rm -f "$FM_AFK_LAUNCH_STATE/.afk" \
     "$FM_AFK_LAUNCH_STATE/.subsuper-escalations" \
     "$FM_AFK_LAUNCH_STATE/.subsuper-escalations.since" \
-    "$FM_AFK_LAUNCH_STATE/.subsuper-inject-wedged" || result=1
+    "$FM_AFK_LAUNCH_STATE/.subsuper-inject-wedged" \
+    "$FM_AFK_LAUNCH_STATE/.subsuper-seen-wake-seq" || result=1
   if [ "$had_afk" -eq 1 ]; then
     cp "$backup/.afk" "$FM_AFK_LAUNCH_STATE/.afk" || result=1
   fi
-  for artifact in .subsuper-escalations .subsuper-escalations.since .subsuper-inject-wedged; do
+  for artifact in .subsuper-escalations .subsuper-escalations.since .subsuper-inject-wedged .subsuper-seen-wake-seq; do
     if [ -e "$backup/$artifact" ]; then
       cp -p "$backup/$artifact" "$FM_AFK_LAUNCH_STATE/$artifact" || result=1
     fi
@@ -487,7 +488,7 @@ fm_afk_launch_start() {
     had_afk=1
     cp "$FM_AFK_LAUNCH_STATE/.afk" "$backup/.afk" || { rm -rf "$backup"; return 1; }
   fi
-  for artifact in .subsuper-escalations .subsuper-escalations.since .subsuper-inject-wedged; do
+  for artifact in .subsuper-escalations .subsuper-escalations.since .subsuper-inject-wedged .subsuper-seen-wake-seq; do
     if [ -e "$FM_AFK_LAUNCH_STATE/$artifact" ]; then
       cp -p "$FM_AFK_LAUNCH_STATE/$artifact" "$backup/$artifact" || { rm -rf "$backup"; return 1; }
     fi
@@ -545,7 +546,7 @@ fm_afk_launch_start_native() {
     had_afk=1
     cp "$FM_AFK_LAUNCH_STATE/.afk" "$backup/.afk" || { rm -rf "$backup"; return 1; }
   fi
-  for artifact in .subsuper-escalations .subsuper-escalations.since .subsuper-inject-wedged; do
+  for artifact in .subsuper-escalations .subsuper-escalations.since .subsuper-inject-wedged .subsuper-seen-wake-seq; do
     if [ -e "$FM_AFK_LAUNCH_STATE/$artifact" ]; then
       cp -p "$FM_AFK_LAUNCH_STATE/$artifact" "$backup/$artifact" || { rm -rf "$backup"; return 1; }
     fi
diff --git a/bin/fm-afk-return.sh b/bin/fm-afk-return.sh
index b38c1e07c4..fb84f64eb9 100755
--- a/bin/fm-afk-return.sh
+++ b/bin/fm-afk-return.sh
@@ -124,7 +124,10 @@ clear_delivery_artifacts() {
   rm -f \
     "$STATE/.subsuper-escalations" \
     "$STATE/.subsuper-escalations.since" \
-    "$STATE/.subsuper-inject-wedged"
+    "$STATE/.subsuper-inject-wedged" \
+    "$STATE/.subsuper-seen-wake-seq" \
+    "$STATE/.subsuper-pending-wake" \
+    "$STATE/.subsuper-seen-wake-checks"
 }
 
 return_guard() {
@@ -204,8 +207,8 @@ return_reconcile() {
     return 3
   fi
 
-  rm -f "$GATE"
-  clear_delivery_artifacts
+  clear_delivery_artifacts || { rm -f "$evidence" "$blockers" "$drain_err"; return 1; }
+  rm -f "$GATE" || { rm -f "$evidence" "$blockers" "$drain_err"; return 1; }
   rm -f "$evidence" "$blockers" "$drain_err"
   printf 'fm-afk-return: catch-up clear; ordinary captain work may proceed\n'
   return 0
diff --git a/bin/fm-afk-start.sh b/bin/fm-afk-start.sh
index e86c54f170..042c360f07 100755
--- a/bin/fm-afk-start.sh
+++ b/bin/fm-afk-start.sh
@@ -63,7 +63,8 @@ fm_afk_clear_stale_artifacts() {  # <state-dir>
   local state=$1
   rm -f "$state/.subsuper-escalations" \
         "$state/.subsuper-escalations.since" \
-        "$state/.subsuper-inject-wedged" 2>/dev/null
+        "$state/.subsuper-inject-wedged" \
+        "$state/.subsuper-seen-wake-seq" 2>/dev/null
 }
 
 daemon_lock_owner() {
diff --git a/bin/fm-pi-arm-tree-retire.sh b/bin/fm-pi-arm-tree-retire.sh
new file mode 100755
index 0000000000..117643f18e
--- /dev/null
+++ b/bin/fm-pi-arm-tree-retire.sh
@@ -0,0 +1,201 @@
+#!/usr/bin/env bash
+set -u
+
+native_pid=${1:-}
+token=${2:-}
+state=${3:-}
+watch_path=${4:-}
+home=${5:-}
+case "$native_pid" in ''|*[!0-9]*|0|1) exit 2 ;; esac
+case "$token" in ''|*[!0-9a-f]*) exit 2 ;; esac
+[ -n "$state" ] && [ -n "$watch_path" ] && [ -n "$home" ] || exit 2
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+FM_HOME=$home
+FM_STATE_OVERRIDE=$state
+export FM_HOME FM_STATE_OVERRIDE
+# shellcheck source=bin/fm-wake-lib.sh
+. "$SCRIPT_DIR/fm-wake-lib.sh"
+
+proc_root=${FM_PROC_ROOT_OVERRIDE:-/proc}
+wrapper_pid_file="$state/.pi-arm-wrapper-$token.pid"
+retirement_dir="$state/.pi-arm-retirement-$token"
+FM_PI_ARM_WRAPPER_PID_FILE=$wrapper_pid_file
+export FM_PI_ARM_WRAPPER_PID_FILE
+
+read_single_pid() {
+  awk '
+    NR == 1 && /^[0-9]+$/ && $0 != "0" && $0 != "1" { pid = $0; next }
+    { invalid = 1 }
+    END {
+      if (!invalid && pid != "") print pid
+      else exit 1
+    }
+  ' "$1" 2>/dev/null
+}
+
+cleanup_snapshot() {
+  rm -f \
+    "$1/wrapper-msys-pid" \
+    "$1/wrapper-native-pid" \
+    "$1/watcher-recorded" \
+    "$1/watcher-msys-pid" \
+    "$1/watcher-native-pid" \
+    "$1/watcher-native-pid.tmp" \
+    "$1/watcher-identity" \
+    "$1/watcher-recorded.tmp" 2>/dev/null || true
+  rmdir "$1" 2>/dev/null || true
+}
+
+read_watcher_lock_snapshot() {
+  local lock_home lock_path current_identity
+  watcher_pid=$(read_single_pid "$state/.watch.lock/pid") || return 1
+  lock_home=$(cat "$state/.watch.lock/fm-home" 2>/dev/null || true)
+  lock_path=$(cat "$state/.watch.lock/watcher-path" 2>/dev/null || true)
+  watcher_identity=$(cat "$state/.watch.lock/pid-identity" 2>/dev/null || true)
+  [ "$lock_home" = "$home" ] || return 1
+  [ "$lock_path" = "$watch_path" ] || return 1
+  [ -n "$watcher_identity" ] || return 1
+  watcher_native_pid=
+  current_identity=$(fm_pid_identity "$watcher_pid" 2>/dev/null || true)
+  if [ "$current_identity" = "$watcher_identity" ]; then
+    watcher_native_pid=$(read_single_pid "$proc_root/$watcher_pid/winpid") || return 1
+  fi
+  watcher_recorded=1
+}
+
+record_snapshot() {
+  local i=0 tmp mapped_native_pid
+  while [ "$i" -lt 20 ]; do
+    msys_pid=$(read_single_pid "$wrapper_pid_file" 2>/dev/null || true)
+    [ -n "$msys_pid" ] && break
+    sleep 0.05
+    i=$((i + 1))
+  done
+  [ -n "${msys_pid:-}" ] || return 1
+  mapped_native_pid=$(read_single_pid "$proc_root/$msys_pid/winpid") || return 1
+  [ "$mapped_native_pid" = "$native_pid" ] || return 1
+  tree_matches || return 1
+
+  watcher_recorded=0
+  watcher_pid=
+  watcher_native_pid=
+  watcher_identity=
+  if [ -e "$state/.watch.lock" ]; then
+    read_watcher_lock_snapshot || return 1
+  fi
+
+  tmp="$retirement_dir.${BASHPID:-$$}"
+  mkdir "$tmp" 2>/dev/null || return 1
+  if ! printf '%s\n' "$msys_pid" > "$tmp/wrapper-msys-pid" \
+    || ! printf '%s\n' "$native_pid" > "$tmp/wrapper-native-pid" \
+    || ! printf '%s\n' "$watcher_recorded" > "$tmp/watcher-recorded" \
+    || ! printf '%s\n' "$watcher_pid" > "$tmp/watcher-msys-pid" \
+    || ! printf '%s\n' "$watcher_native_pid" > "$tmp/watcher-native-pid" \
+    || ! printf '%s\n' "$watcher_identity" > "$tmp/watcher-identity" \
+    || ! mv "$tmp" "$retirement_dir"; then
+    cleanup_snapshot "$tmp"
+    return 1
+  fi
+}
+
+tree_matches() {
+  local current_native_pid
+  current_native_pid=$(read_single_pid "$proc_root/$msys_pid/winpid") || return 1
+  [ "$current_native_pid" = "$native_pid" ] \
+    && [ -r "$proc_root/$msys_pid/environ" ] \
+    && tr '\0' '\n' < "$proc_root/$msys_pid/environ" 2>/dev/null \
+      | grep -F -x -- "FM_PI_ARM_TREE_TOKEN=$token" >/dev/null 2>&1
+}
+
+if [ ! -e "$retirement_dir" ]; then
+  record_snapshot || exit 3
+fi
+
+msys_pid=$(read_single_pid "$retirement_dir/wrapper-msys-pid") || exit 3
+recorded_native_pid=$(read_single_pid "$retirement_dir/wrapper-native-pid") || exit 3
+[ "$recorded_native_pid" = "$native_pid" ] || exit 3
+watcher_recorded=$(cat "$retirement_dir/watcher-recorded" 2>/dev/null || true)
+case "$watcher_recorded" in 0|1) ;; *) exit 3 ;; esac
+watcher_pid=
+watcher_native_pid=
+watcher_identity=
+if [ "$watcher_recorded" -eq 1 ]; then
+  watcher_pid=$(read_single_pid "$retirement_dir/watcher-msys-pid") || exit 3
+  watcher_native_pid=$(cat "$retirement_dir/watcher-native-pid" 2>/dev/null || true)
+  case "$watcher_native_pid" in
+    '') ;;
+    *[!0-9]*|0|1) exit 3 ;;
+  esac
+  watcher_identity=$(cat "$retirement_dir/watcher-identity" 2>/dev/null || true)
+  [ -n "$watcher_identity" ] || exit 3
+fi
+watcher_matches() {
+  local current
+  [ "$watcher_recorded" -eq 1 ] || return 1
+  current=$(fm_pid_identity "$watcher_pid") || return 1
+  [ "$current" = "$watcher_identity" ]
+}
+
+record_late_watcher() {
+  [ "$watcher_recorded" -eq 0 ] || return 0
+  [ -e "$state/.watch.lock" ] || return 0
+  read_watcher_lock_snapshot || return 1
+  if ! printf '%s\n' "$watcher_pid" > "$retirement_dir/watcher-msys-pid" \
+    || ! printf '%s\n' "$watcher_native_pid" > "$retirement_dir/watcher-native-pid" \
+    || ! printf '%s\n' "$watcher_identity" > "$retirement_dir/watcher-identity" \
+    || ! printf '1\n' > "$retirement_dir/watcher-recorded.tmp" \
+    || ! mv -f "$retirement_dir/watcher-recorded.tmp" "$retirement_dir/watcher-recorded"; then
+    return 1
+  fi
+  watcher_recorded=1
+}
+
+bind_watcher_native_pid() {
+  local current_native_pid
+  current_native_pid=$(read_single_pid "$proc_root/$watcher_pid/winpid") || return 1
+  if [ -n "$watcher_native_pid" ]; then
+    [ "$current_native_pid" = "$watcher_native_pid" ]
+    return
+  fi
+  if ! printf '%s\n' "$current_native_pid" > "$retirement_dir/watcher-native-pid.tmp" \
+    || ! mv -f "$retirement_dir/watcher-native-pid.tmp" "$retirement_dir/watcher-native-pid"; then
+    return 1
+  fi
+  watcher_native_pid=$current_native_pid
+}
+
+taskkill_status=0
+if tree_matches; then
+  MSYS2_ARG_CONV_EXCL='*' taskkill.exe /PID "$native_pid" /T /F >/dev/null 2>&1 || taskkill_status=$?
+fi
+i=0
+while [ "$i" -lt 5 ] && tree_matches; do
+  sleep 0.05
+  i=$((i + 1))
+done
+i=0
+while [ "$watcher_recorded" -eq 0 ] && [ "$i" -lt 5 ]; do
+  record_late_watcher || exit 1
+  [ "$watcher_recorded" -eq 1 ] && break
+  sleep 0.05
+  i=$((i + 1))
+done
+if watcher_matches; then
+  bind_watcher_native_pid || exit 1
+  MSYS2_ARG_CONV_EXCL='*' taskkill.exe /PID "$watcher_native_pid" /T /F >/dev/null 2>&1 || taskkill_status=$?
+fi
+i=0
+while [ "$i" -lt 20 ]; do
+  if ! tree_matches && ! watcher_matches; then
+    if [ "$taskkill_status" -eq 0 ]; then
+      cleanup_snapshot "$retirement_dir"
+      rm -f "$wrapper_pid_file" 2>/dev/null || true
+      exit 0
+    fi
+    exit 1
+  fi
+  sleep 0.05
+  i=$((i + 1))
+done
+exit 1
diff --git a/bin/fm-supervise-daemon.sh b/bin/fm-supervise-daemon.sh
index ff123a7faf..81d67bae5d 100755
--- a/bin/fm-supervise-daemon.sh
+++ b/bin/fm-supervise-daemon.sh
@@ -37,8 +37,9 @@
 #     to daemon-owned one-shot behavior and enqueues every wake to
 #     state/.wake-queue BEFORE advancing its suppression markers, so a
 #     crash/restart/missed injection is recovered on the next fm-wake-drain.sh.
-#     After a watcher cycle, the daemon handles every durable row through that
-#     drain and acknowledges it only after routing completes.
+#     The daemon lock-snapshots unseen queue rows, classifies the newest row per
+#     logical wake, and makes interrupted classification idempotent without
+#     consuming the queue; fm-wake-drain.sh remains the only consumer.
 #   - Fail-safe-to-escalate: any wake the classifier cannot confidently mark
 #     routine is escalated.
 #   - Bounded wedge latency: a stale pane without a declared external wait is
@@ -427,7 +428,7 @@ classify_unknown() {  # <reason>
 
 # --- stale marker + escalation buffer (stateful, but via explicit state dir) -
 # Marker:   state/.subsuper-stale-<key>   contains the epoch first seen idle.
-# Buffer:   state/.subsuper-escalations    one distilled line per escalation.
+# Buffer:   state/.subsuper-escalations    one internal record per escalation.
 # Seen:     state/.subsuper-seen-status-<task>  last status line the scan
 #           escalated, so the catch-all does not re-fire the same terminal.
 
@@ -632,13 +633,35 @@ stale_window_is_busy() {  # <window> <state>
   [ "${verdict%% *}" = busy ]
 }
 
-escalate_add() {  # <state> <distilled-item>
-  local state=$1 item=$2 buf
+escalate_add() {  # <state> <distilled-item> [wake-id]
+  local state=$1 item=$2 wake_id=${3:-} buf tmp was_empty=0
   buf="$state/.subsuper-escalations"
+  if [ -n "$wake_id" ]; then
+    case "$wake_id" in *[!0-9-]*|*-*-*|-*|*-) return 2 ;; esac
+    if awk -F '\t' -v id="@wake:$wake_id" '$1 == id { found = 1 } END { exit !found }' "$buf" 2>/dev/null; then
+      [ -r "${buf}.since" ] || _now > "${buf}.since"
+      return $?
+    fi
+    [ -s "$buf" ] || was_empty=1
+    tmp="$buf.queue.${BASHPID:-$$}"
+    if ! {
+      [ ! -e "$buf" ] || cat "$buf"
+      printf '@wake:%s\t%s\n' "$wake_id" "$item"
+    } > "$tmp" || ! mv "$tmp" "$buf"; then
+      rm -f "$tmp" 2>/dev/null || true
+      return 1
+    fi
+    [ "$was_empty" -eq 0 ] || _now > "${buf}.since"
+    return $?
+  fi
   [ -s "$buf" ] || _now > "${buf}.since"
   printf '%s\n' "$item" >> "$buf"
 }
 
+escalate_buffer_items() {  # <buffer>
+  awk '{ sub(/^@wake:[0-9]+-[0-9]+\t/, ""); print }' "$1" 2>/dev/null
+}
+
 # Flush the escalation buffer as ONE batched, single-line digest to the
 # supervisor pane. Returns 0 on successful inject (or empty buffer), non-zero on
 # inject failure (buffer preserved for retry / catch-up).
@@ -648,7 +671,7 @@ escalate_flush() {  # <state>
   [ -s "$buf" ] || return 0
   n=$(wc -l < "$buf" 2>/dev/null || echo 0)
   # Join buffered items with the literal " | " separator into one digest line.
-  msg=$(awk 'NR>1{printf " | "} {printf "%s",$0} END{print ""}' "$buf" 2>/dev/null)
+  msg=$(escalate_buffer_items "$buf" | awk 'NR>1{printf " | "} {printf "%s",$0} END{print ""}')
   # Single-line wrapper: no embedded newlines (inject_msg also collapses as a
   # safety net, but keeping the source single-line makes the intent explicit).
   msg=$(printf 'Supervisor escalate (%s event(s)): %s (pre-read; re-arm not needed — watcher daemon-managed)' "$n" "$msg")
@@ -912,7 +935,7 @@ inject_wedge_alarm() {  # <state> <age-seconds>
   {
     printf 'fm away-mode inject WEDGED: %ss undelivered as of %s\n' "$age" "$(date '+%Y-%m-%dT%H:%M:%S%z')"
     printf 'The supervisor pane could not accept an escalation. Buffered items:\n'
-    cat "$state/.subsuper-escalations" 2>/dev/null
+    escalate_buffer_items "$state/.subsuper-escalations"
   } 2>/dev/null > "$marker" || true
   target="${FM_SUPERVISOR_TARGET:-$FM_SUPERVISOR_TARGET_DEFAULT}"
   backend="${FM_SUPERVISOR_BACKEND:-$FM_SUPERVISOR_BACKEND_DEFAULT}"
@@ -945,7 +968,8 @@ _oldest_line_age() {  # <buf> -> seconds since the oldest buffered item first ar
 }
 
 # --- housekeeping (runs every tick while the watcher is mid-cycle) ----------
-# Four cheap jobs, each guarded so an empty/quiet fleet costs near zero:
+# Five cheap jobs, each guarded so an empty/quiet fleet costs near zero:
+#  0) classify durable queue rows not yet observed by this away session.
 #  1) batch flush: if the escalation buffer's oldest content is older than
 #     ESCALATE_BATCH_SECS (or batching is disabled), inject one digest.
 #  1b) max-defer escape: if the buffer is STILL undelivered past MAX_DEFER_SECS,
@@ -962,6 +986,7 @@ housekeeping() {  # <state>
   local state=$1 now due f key task win marker age last max_defer oldest pause_secs
   now=$(_now)
   migrate_watcher_pause_markers "$state"
+  classify_queued_wakes "$state" || true
 
   # (1) batch flush
   if [ "${FM_ESCALATE_BATCH_SECS:-$ESCALATE_BATCH_SECS_DEFAULT}" -le 0 ]; then
@@ -1201,12 +1226,12 @@ is_wake_reason() {  # <reason>
 
 # --- dispatch one wake reason to self-handle or escalate --------------------
 # Side effects: logging, marker records, escalation buffer appends.
-handle_wake() {  # <reason> <state>
-  local reason=$1 state=$2 decision action distilled task last stale_detail
+handle_wake() {  # <reason> <state> [wake-id]
+  local reason=$1 state=$2 wake_id=${3:-} decision action distilled task last stale_detail
   local kind="" arg=""
   if should_force_self "$reason"; then
     log "wake force-self (FM_INJECT_SKIP): $reason"
-    return
+    return 0
   fi
   case "$reason" in
     signal:*) kind=signal; arg="${reason#signal: }"
@@ -1228,12 +1253,14 @@ handle_wake() {  # <reason> <state>
   case "$action" in
     escalate)
       log "escalate: $reason -> $distilled"
-      escalate_add "$state" "$distilled"
+      escalate_add "$state" "$distilled" "$wake_id" || return 1
       # A terminal-stale escalate must not leave a persistence marker behind, or
       # housekeeping re-escalates the same pane as a false wedge later.
       [ "$kind" = "stale" ] && stale_marker_remove "$arg" "$state"
       mark_escalated_seen "$kind" "$arg" "$state"
-      [ "${FM_ESCALATE_BATCH_SECS:-$ESCALATE_BATCH_SECS_DEFAULT}" -le 0 ] && { escalate_flush "$state" || true; }
+      if [ -z "$wake_id" ] && [ "${FM_ESCALATE_BATCH_SECS:-$ESCALATE_BATCH_SECS_DEFAULT}" -le 0 ]; then
+        escalate_flush "$state" || true
+      fi
       ;;
     pause)
       # Declared external-wait pause: record a pause marker (long re-surface
@@ -1278,39 +1305,106 @@ handle_wake() {  # <reason> <state>
       log "self-handle: $reason -> $distilled"
       ;;
   esac
+  return 0
 }
 
-handle_durable_wakes() {  # <watcher-reason> <state>
-  local fallback_reason=$1 state=$2 out err tab epoch sequence kind key payload rest
-  local handled=0 ack_through ack_generation
-  out=$(mktemp "$state/.subsuper-wake-drain.XXXXXX") || return 1
-  err=$(mktemp "$state/.subsuper-wake-drain.XXXXXX") || { rm -f "$out"; return 1; }
-  if ! "$FM_DAEMON_DIR/fm-wake-drain.sh" > "$out" 2> "$err"; then
-    cat "$err" >&2
-    rm -f "$out" "$err"
+QUEUED_WAKE_CLASSIFICATION_ACTIVE=0
+
+queued_wake_cursor_advance() {  # <cursor-file> <seq>
+  local cursor_file=$1 seq=$2 current tmp
+  current=$(cat "$cursor_file" 2>/dev/null || echo 0)
+  case "$current" in ''|*[!0-9]*) current=0 ;; esac
+  [ "$current" -lt "$seq" ] || return 0
+  tmp="$cursor_file.${BASHPID:-$$}"
+  if ! printf '%s\n' "$seq" > "$tmp" || ! mv "$tmp" "$cursor_file"; then
+    rm -f "$tmp" 2>/dev/null || true
     return 1
   fi
+}
 
-  tab=$(printf '\t')
-  while IFS="$tab" read -r epoch sequence kind key payload rest; do
-    case "$epoch" in ''|*[!0-9]*) continue ;; esac
-    case "$sequence" in ''|*[!0-9]*) continue ;; esac
-    case "$kind" in signal|stale|check|heartbeat) ;; *) continue ;; esac
-    handle_wake "$payload" "$state"
-    handled=$((handled + 1))
-  done < "$out"
-  [ "$handled" -gt 0 ] || handle_wake "$fallback_reason" "$state"
-
-  ack_through=$(sed -n 's/^WAKE_ACK_REQUIRED:.*--ack-through \([0-9][0-9]*\) --recovery-generation [A-Za-z0-9._-][A-Za-z0-9._-]*$/\1/p' "$err" | tail -1)
-  ack_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' "$err" | tail -1)
-  grep -v '^WAKE_ACK_REQUIRED:' "$err" >&2 || true
-  rm -f "$out" "$err"
-  if [ -z "$ack_through" ] || [ -z "$ack_generation" ]; then
-    log "wake drain omitted its generation-bound acknowledgement; retaining durable wakes"
-    return 1
+queued_wake_check_is_duplicate() {  # <state> <key> <payload>
+  local state=$1 key=$2 payload=$3 ledger _seen_epoch seen_seq seen_key seen_payload
+  ledger="$state/.subsuper-seen-wake-checks"
+  [ -r "$ledger" ] || return 1
+  while IFS="$(printf '\t')" read -r _seen_epoch seen_seq seen_key seen_payload; do
+    [ "$seen_key" = "$key" ] || continue
+    fm_wake_record_seq_present "$seen_seq" && return 0
+    [ "$seen_payload" = "$payload" ] || return 1
+    escalate_buffer_items "$state/.subsuper-escalations" | grep -F -x -- "$payload" >/dev/null 2>&1
+    return $?
+  done < "$ledger"
+  return 1
+}
+
+queued_wake_check_store() {  # <state> <epoch> <seq> <key> <payload>
+  local state=$1 epoch=$2 seq=$3 key=$4 payload=$5 ledger tmp seen_epoch seen_seq seen_key seen_payload
+  ledger="$state/.subsuper-seen-wake-checks"
+  tmp="$ledger.${BASHPID:-$$}"
+  : > "$tmp" || return 1
+  if [ -r "$ledger" ]; then
+    while IFS="$(printf '\t')" read -r seen_epoch seen_seq seen_key seen_payload; do
+      [ "$seen_key" = "$key" ] && continue
+      printf '%s\t%s\t%s\t%s\n' "$seen_epoch" "$seen_seq" "$seen_key" "$seen_payload" >> "$tmp" || {
+        rm -f "$tmp" 2>/dev/null || true
+        return 1
+      }
+    done < "$ledger"
+  fi
+  if ! printf '%s\t%s\t%s\t%s\n' "$epoch" "$seq" "$key" "$payload" >> "$tmp" \
+    || ! mv "$tmp" "$ledger"; then
+      rm -f "$tmp" 2>/dev/null || true
+      return 1
+  fi
+}
+
+classify_pending_queued_wake() {  # <state> <pending-file> <cursor-file>
+  local state=$1 pending_file=$2 cursor_file=$3 epoch seq kind key payload extra
+  IFS="$(printf '\t')" read -r epoch seq kind key payload extra < "$pending_file" || return 1
+  case "$epoch" in ''|*[!0-9]*) return 1 ;; esac
+  case "$seq" in ''|*[!0-9]*) return 1 ;; esac
+  case "$kind" in signal|stale|check|heartbeat) ;; *) return 1 ;; esac
+  [ -n "$key" ] && [ -n "$payload" ] && [ -z "$extra" ] || return 1
+  queued_wake_cursor_advance "$cursor_file" "$seq" || return 1
+  if [ "$kind" = check ] && queued_wake_check_is_duplicate "$state" "$key" "$payload"; then
+    queued_wake_check_store "$state" "$epoch" "$seq" "$key" "$payload" || return 1
+    rm -f "$pending_file"
+    return $?
+  fi
+  handle_wake "$payload" "$state" "$epoch-$seq" || return 1
+  if [ "$kind" = check ]; then
+    queued_wake_check_store "$state" "$epoch" "$seq" "$key" "$payload" || return 1
   fi
-  "$FM_DAEMON_DIR/fm-wake-drain.sh" --ack-through "$ack_through" \
-    --recovery-generation "$ack_generation"
+  rm -f "$pending_file"
+}
+
+classify_queued_wakes() {  # <state>
+  local state=$1 cursor_file pending_file seen rows epoch seq kind key payload tmp
+  afk_active "$state" || return 0
+  type fm_wake_records_after_seq >/dev/null 2>&1 || return 0
+  cursor_file="$state/.subsuper-seen-wake-seq"
+  pending_file="$state/.subsuper-pending-wake"
+  QUEUED_WAKE_CLASSIFICATION_ACTIVE=1
+  if [ -e "$pending_file" ]; then
+    classify_pending_queued_wake "$state" "$pending_file" "$cursor_file" || return 1
+  fi
+  seen=$(cat "$cursor_file" 2>/dev/null || echo 0)
+  case "$seen" in ''|*[!0-9]*) seen=0 ;; esac
+  rows=$(fm_wake_records_after_seq "$seen") || return 1
+  if [ -z "$rows" ]; then
+    QUEUED_WAKE_CLASSIFICATION_ACTIVE=0
+    return 0
+  fi
+  while IFS="$(printf '\t')" read -r epoch seq kind key payload; do
+    [ -n "$seq" ] || continue
+    tmp="$pending_file.${BASHPID:-$$}"
+    if ! printf '%s\t%s\t%s\t%s\t%s\n' "$epoch" "$seq" "$kind" "$key" "$payload" > "$tmp" \
+      || ! mv "$tmp" "$pending_file"; then
+      rm -f "$tmp" 2>/dev/null || true
+      return 1
+    fi
+    classify_pending_queued_wake "$state" "$pending_file" "$cursor_file" || return 1
+  done <<< "$rows"
+  QUEUED_WAKE_CLASSIFICATION_ACTIVE=0
 }
 
 # --- log --------------------------------------------------------------------
@@ -1451,7 +1545,7 @@ fm_super_main() {
   cleanup() {
     trap - TERM INT
     wedge_alarm_stop_active_notifier
-    escalate_flush "$STATE" 2>/dev/null || true
+    [ "$QUEUED_WAKE_CLASSIFICATION_ACTIVE" -eq 1 ] || escalate_flush "$STATE" 2>/dev/null || true
     if [ -n "${WATCHER_PID:-}" ]; then
       kill "$WATCHER_PID" 2>/dev/null || true
       wait "$WATCHER_PID" 2>/dev/null || true
@@ -1539,9 +1633,7 @@ fm_super_main() {
           continue
         fi
         log "wake: $reason"
-        if ! handle_durable_wakes "$reason" "$STATE"; then
-          log "durable wake handling was not acknowledged; restarting for recovery"
-        fi
+        classify_queued_wakes "$STATE" || true
         trim_log
       fi
       start_watcher || continue
diff --git a/bin/fm-test-run.sh b/bin/fm-test-run.sh
index af55d99777..bb47ab2778 100755
--- a/bin/fm-test-run.sh
+++ b/bin/fm-test-run.sh
@@ -180,7 +180,7 @@ family_for_basename() {
     fm-tangle-guard.test.sh|fm-update.test.sh)
       printf '%s\n' session-bootstrap
       ;;
-    fm-afk-pi-herdr-return-e2e.test.sh|\
+    fm-afk-pi-herdr-return-e2e.test.sh|fm-pi-afk-handoff-live-e2e.test.sh|\
     fm-cmux-claude-composer-live-e2e.test.sh|\
     fm-composer-matrix-live-e2e.test.sh|\
     fm-codex-continuity-live-e2e.test.sh|fm-grok-continuity-live-e2e.test.sh|\
@@ -377,6 +377,7 @@ portable_serial_weight_hints() {
   cat <<'EOF'
 tests/fm-afk-inject-e2e.test.sh 34019
 tests/fm-afk-pi-herdr-return-e2e.test.sh 42
+tests/fm-pi-afk-handoff-live-e2e.test.sh 30
 tests/fm-afk-return.test.sh 1105
 tests/fm-ask-user-authority.test.sh 68
 tests/fm-backend-cmux-smoke.test.sh 29
diff --git a/bin/fm-wake-lib.sh b/bin/fm-wake-lib.sh
index 1a739c9f96..b2767fd1e3 100755
--- a/bin/fm-wake-lib.sh
+++ b/bin/fm-wake-lib.sh
@@ -982,6 +982,53 @@ fm_wake_queued_keys_locked() {
     "$FM_WAKE_QUEUE" 2>/dev/null || true
 }
 
+# fm_wake_records_after_seq <seq>
+# Print the newest complete, valid record per logical wake whose sequence is
+# newer than <seq>. The append lock makes the snapshot indivisible with appends
+# and drains; callers may then classify the captured rows without consuming the
+# durable queue.
+fm_wake_records_after_seq() {
+  local seen=$1 status=0
+  case "$seen" in
+    ''|*[!0-9]*) printf 'fm_wake_records_after_seq: invalid sequence: %s\n' "$seen" >&2; return 2 ;;
+  esac
+  fm_lock_acquire_wait "$FM_WAKE_QUEUE_LOCK" || return 1
+  awk -F '\t' -v seen="$seen" '
+    NF == 5 && $1 ~ /^[0-9]+$/ && $2 ~ /^[0-9]+$/ &&
+      ($3 == "signal" || $3 == "stale" || $3 == "check" || $3 == "heartbeat") &&
+      $4 != "" && $5 != "" && ($2 + 0) > seen {
+        dedupe = $3 SUBSEP $4
+        if ($3 == "heartbeat") dedupe = "heartbeat"
+        count++
+        row[count] = $0
+        row_dedupe[count] = dedupe
+        row_seq[count] = $2 + 0
+        latest[dedupe] = $2 + 0
+      }
+    END {
+      for (i = 1; i <= count; i++) {
+        if (row_seq[i] == latest[row_dedupe[i]]) print row[i]
+      }
+    }
+  ' "$FM_WAKE_QUEUE" 2>/dev/null || status=$?
+  fm_lock_release "$FM_WAKE_QUEUE_LOCK"
+  return "$status"
+}
+
+fm_wake_record_seq_present() {
+  local seq=$1 status=1
+  case "$seq" in
+    ''|*[!0-9]*) return 2 ;;
+  esac
+  fm_lock_acquire_wait "$FM_WAKE_QUEUE_LOCK" || return 1
+  if awk -F '\t' -v seq="$seq" 'NF == 5 && ($2 + 0) == seq { found = 1 } END { exit !found }' \
+    "$FM_WAKE_QUEUE" 2>/dev/null; then
+    status=0
+  fi
+  fm_lock_release "$FM_WAKE_QUEUE_LOCK"
+  return "$status"
+}
+
 fm_wake_restore_queue() {
   local drained=$1 restore
   restore="$STATE/.wake-queue.restore.$(fm_current_pid)"
diff --git a/docs/supervision-protocols/pi.md b/docs/supervision-protocols/pi.md
index 8dcaa13238..ba2cad0904 100644
--- a/docs/supervision-protocols/pi.md
+++ b/docs/supervision-protocols/pi.md
@@ -19,6 +19,11 @@ When this session owns supervision and away mode is not active:
 11. Never use shell `&` for watcher supervision.
    The arm mechanism above is extension-owned, not a model tool call, but a manual recovery probe that backgrounds, pipes, or bundles the arm is denied automatically by the PreToolUse seatbelt (`bin/fm-arm-pretool-check.sh`, wired into the turn-end guard extension at `__FM_PI_TURNEND_EXT__`).
 
+When `state/.afk` is present, the already-loaded extensions stand down automatically: the watcher extension retires its exact attached arm cycle, both extensions absorb routine extension-origin follow-ups, and the away daemon becomes the only watcher owner.
+Marked `away-supervisor` decisions, failures, blockers, credential needs, checks, and review-ready results continue into Pi.
+On return, clearing `.afk` makes the same watcher generation restore one ordinary cycle automatically; do not call `fm_watch_arm_pi` for the handoff or return path.
+The exact ownership and queue-transfer contract lives in [`watcher-continuity.md`](../watcher-continuity.md).
+
 The turn-end guard extension lives at `__FM_PI_TURNEND_EXT__`.
 The watcher extension lives at `__FM_PI_EXT__`.
 Both are tracked, project-local `.pi/extensions/*.ts` files that Pi auto-discovers once the project is trusted; `bin/fm-session-start.sh` reports when the running Pi session has not loaded both required extensions.
diff --git a/docs/verification/supervision.md b/docs/verification/supervision.md
index d0837023d3..7d164922e3 100644
--- a/docs/verification/supervision.md
+++ b/docs/verification/supervision.md
@@ -446,11 +446,32 @@ Observed guarantee: after ordinary `session_shutdown` for `/new`, `/resume`, and
 Stale prior-generation tool callbacks could not mutate the active child, repeated transitions kept exactly one live arm cycle, and terminal `quit` still refused late rearm.
 Plain Pi and pi-signed share the same tracked `.pi/extensions/fm-primary-pi-watch.ts` path, so both inherit the generation owner; other primary harnesses are not applicable because they do not use this Pi extension lifecycle.
 
+The Pi-to-AFK ownership handoff was verified on 2026-08-10 with installed Pi 0.84.1 and Herdr 0.7.3 in a helper-provisioned named non-default session, a throwaway Firstmate home, the real tracked Pi extension, and no model-provider request.
+The reproduced failure PID remained the same live bash wrapper at failure: PID, PPID, sleeping foreground state, start time, and command matched launch, while Node had observed neither `exit` nor `close`.
+The signal boundary was the wrapper's bash wait: signaling only its PID left that wrapper alive after its watcher exited.
+The live correction exercised the POSIX path: each Pi arm owns a dedicated process group, and handoff retires that group before the daemon acquires the watcher singleton.
+The cross-platform retirement contract is owned by [`watcher-continuity.md`](../watcher-continuity.md#ownership); `tests/fm-pi-watch-extension.test.sh` covers the Windows native-to-MSYS identity binding, fail-closed identity refusals, partial-retirement retry, and single replacement convergence.
+
+```sh
+HERDR_LAB_HELPER='/data/1/projects/bin/fm-herdr-lab.sh'
+HERDR_LAB_SESSION=$("$HERDR_LAB_HELPER" name pi-afk-watcher-handoff-h1)
+export HERDR_LAB_HELPER HERDR_LAB_SESSION
+FM_PI_AFK_HANDOFF_LIVE_E2E=1 tests/fm-pi-afk-handoff-live-e2e.test.sh
+```
+
+Observed guarantee: ordinary Pi supervision first acquired an extension-owned wrapper and watcher; AFK entry retired both exact identities and let the daemon acquire the singleton; routine progress produced no Pi turn; one actionable result arrived once through the typed `away-supervisor` path; the raw at-least-once queue remained lossless and drained to one logical signal; return stopped the daemon and restored one extension-owned cycle without a command; the resumed cycle launched its successor; and a second entry/return converged identically.
+
+The affected harness axis is plain Pi plus pi-signed because both load the same extension.
+Claude's AFK-aware Stop auto-arm, OpenCode's separate TUI plugin, Codex's foreground checkpoints, Grok's tracked background arm, and the unknown-harness fallback were inspected and are not applicable to this Pi extension ownership defect.
+The affected away-daemon backend axis is tmux and Herdr: deterministic lifecycle tests cover the shared daemon contract, and the live regression covers Herdr's real non-visible supervisor path.
+Zellij, Orca, and cmux integration surfaces were inspected and are not applicable because `bin/fm-afk-launch.sh` and `bin/fm-supervise-daemon.sh` explicitly refuse them as away-supervisor backends.
+
 Deterministic entry points:
 
 ```sh
 tests/fm-pi-watch-extension.test.sh
 tests/fm-pi-primary-types.test.sh
+tests/fm-pi-afk-handoff-live-e2e.test.sh
 tests/fm-watcher-lock.test.sh
 tests/fm-watch-arm.test.sh
 tests/fm-wake-queue.test.sh
diff --git a/docs/watcher-continuity.md b/docs/watcher-continuity.md
index 71f09f142b..777eeee3a5 100644
--- a/docs/watcher-continuity.md
+++ b/docs/watcher-continuity.md
@@ -9,6 +9,12 @@ Pi's `.pi/extensions/fm-primary-pi-watch.ts` and OpenCode's `.opencode/plugins/f
 Each adapter starts the next arm before delivering the wake prompt, checks current session-lock ownership at launch, preserves one child or scheduled retry at a time, and applies bounded exponential retry after an unexpected or failed close.
 A failed follow-up never cancels continuity restoration.
 Pi same-process session replacement follows the generation-owner contract in `.pi/extensions/fm-primary-pi-watch.ts`.
+Pi away-mode transfer follows that same generation owner.
+When `state/.afk` is present at session start or appears during a live arm, the extension cancels pending retries and retires its exact attached arm tree without a process scan, broad signal, or sibling-home impact.
+On POSIX, every arm is a dedicated process-group leader and the extension retires that exact group.
+On Windows, the extension binds the native child to its published MSYS wrapper identity and the home-scoped watcher-lock identity, retains ownership through partial retirement, and starts no replacement until both recorded identities are gone.
+The loaded extension then suppresses its typed watcher and turn-end-guard inputs while AFK remains active, but leaves typed `away-supervisor` escalations deliverable.
+When the flag clears, the same generation automatically restores one extension-owned cycle; repeated entry and return converge without another model arm command.
 Cursor's `.cursor/hooks.json` `stop` hook (`bin/fm-turnend-guard-cursor.sh`) owns routine tokenless re-arm for a Cursor primary by parking that awaited hook on `bin/fm-watch-arm.sh` and returning an actionable close as one follow-up; [`turnend-guard.md`](turnend-guard.md#harness-integrations) owns its loop bounds and supersession baton.
 Claude's `.claude/settings.json` Stop `asyncRewake` hook (`bin/fm-claude-stop-autoarm.sh`) owns routine tokenless re-arm.
 The hook fires on every Stop, and an eligible primary with supervision need admits one home-scoped owner that foregrounds `bin/fm-watch-arm.sh` inside the hook-owned process tree.
@@ -29,6 +35,10 @@ When that retained arm later closes, its actual close is classified as a new sup
 After the configured retry bound is exhausted, it delivers the original wake with a typed continuity-restoration failure even if every successor arm hung without reporting readiness.
 This is deliberate Option B ordering: the fleet is protected before the model handles the wake whenever restoration succeeds, but the model is never left blind when it does not.
 
+During Pi-to-AFK transfer, the daemon snapshots unseen durable queue rows and classifies the newest record for each logical wake without consuming it.
+Its away-session sequence cursor, crash-pending record, and check-wake ledger make repeated or interrupted classification idempotent, while `bin/fm-wake-drain.sh` remains the sole consumer and retains its logical wake deduplication.
+This closes the interval in which the retiring extension-owned watcher can enqueue after `.afk` appears but before the daemon's own watcher acquires the singleton lock.
+
 Claude's Stop hook starts the successor arm at the next Stop after the handling turn, rather than before notification as Pi and OpenCode do.
 The durable wake queue preserves actionable events during the residual active-turn window, and the bounded turn-end guard enforces recovery at Stop when no watcher or auto-arm claim is present.
 For every supported arm path, a successor that observes an accepted down stretch emits `check: rearm-resurface` through the ordinary durable handling path before settling into its live wait.
@@ -76,6 +86,8 @@ Only the watcher process touches `state/.last-watcher-beat`; no helper process c
 ## Regression coverage
 
 `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.
+It also covers startup while already away, live exact-child yield, routine input absorption, actionable away input pass-through, automatic return, repeated convergence, and sibling-home isolation.
+Its Windows fixtures cover native-to-MSYS identity binding, missing, ambiguous, or reused wrapper publications, a surviving recorded watcher, and retry convergence after partial tree retirement without a duplicate replacement.
 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.
 `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-watcher-lock.test.sh` covers verified-successor attach, recovery publication before stale-lock removal, the typed self-eviction failure, bounded and successor-linked lifecycle rows, and a SIGSTOP counterfactual that distinguishes a live PID from a stale beacon before classifying termination.
@@ -83,6 +95,8 @@ The same suite covers ordinary same-process session replacement for `/new`, `/re
 `tests/fm-claude-stop-autoarm.test.sh` covers the auto-arm's scope, stale and live session owners, unchanged AFK and need boundaries, single-flight, bounded failure retries, benign live-watcher cycle ends, one-notice failure episodes, and exit-2 translation.
 `FM_CLAUDE_LIVE_E2E=1 tests/fm-claude-stop-autoarm-live-e2e.test.sh` starts with the reproduced stale-lock state, runs session start first, completes two tokenless cycles, and checks the competing-live-owner negative control.
 `tests/fm-turnend-guard.test.sh` covers the cooperative `--claude` guard, including monotonic failed-epoch progression, the integrated bounded fail-open, post-alarm continuation suppression, and positive recovery reset.
+`tests/fm-daemon.test.sh` covers non-consuming queue classification, routine absorption, actionable escalation, same-snapshot compaction, later at-least-once check deduplication, and both sides of the cursor/effect crash boundary.
+`FM_PI_AFK_HANDOFF_LIVE_E2E=1 tests/fm-pi-afk-handoff-live-e2e.test.sh` drives a real installed Pi extension through a named isolated Herdr session and throwaway Firstmate home.
 
 ## Active limits and verification
 
diff --git a/tests/fm-afk-launch.test.sh b/tests/fm-afk-launch.test.sh
index 6d0c7bd9d1..2ad5aeaacd 100755
--- a/tests/fm-afk-launch.test.sh
+++ b/tests/fm-afk-launch.test.sh
@@ -41,7 +41,7 @@ GLOBAL_CLEANUP() {
 trap GLOBAL_CLEANUP EXIT
 
 # ---------------------------------------------------------------------------
-# UNIT 1: fm_afk_clear_stale_artifacts removes exactly the three stale artifacts.
+# UNIT 1: fm_afk_clear_stale_artifacts removes the away-session delivery artifacts.
 # ---------------------------------------------------------------------------
 unit_clear_stale() {
   local st
@@ -50,6 +50,7 @@ unit_clear_stale() {
   : > "$st/state/.subsuper-escalations"
   : > "$st/state/.subsuper-escalations.since"
   : > "$st/state/.subsuper-inject-wedged"
+  : > "$st/state/.subsuper-seen-wake-seq"
   : > "$st/state/.wake-queue"          # durable queue must be untouched
   # Source fm-afk-start.sh inside a child bash (it sets `set -eu` and would
   # otherwise leak that into this test shell) and call the clear helper.
@@ -57,8 +58,9 @@ unit_clear_stale() {
     bash -c '. "$1"; fm_afk_clear_stale_artifacts "$2"' _ "$START" "$st/state"
   if [ ! -e "$st/state/.subsuper-escalations" ] \
      && [ ! -e "$st/state/.subsuper-escalations.since" ] \
-     && [ ! -e "$st/state/.subsuper-inject-wedged" ]; then
-    pass "clear-stale: removes escalations buffer, sidecar, and wedge marker"
+     && [ ! -e "$st/state/.subsuper-inject-wedged" ] \
+     && [ ! -e "$st/state/.subsuper-seen-wake-seq" ]; then
+    pass "clear-stale: removes escalation delivery state and the queue-observation cursor"
   else
     fail "clear-stale: stale artifacts survived"
   fi
@@ -128,6 +130,7 @@ unit_fresh_vs_refresh() {
   mkdir -p "$st/state"
   : > "$st/state/.subsuper-escalations"
   : > "$st/state/.subsuper-inject-wedged"
+  printf '12\n' > "$st/state/.subsuper-seen-wake-seq"
   # A live "daemon": a real process whose identity the lock records, so
   # daemon_lock_held_by_live_daemon returns true (a refresh).
   sleep 600 &
@@ -137,7 +140,9 @@ unit_fresh_vs_refresh() {
   printf '%s' "$sleep_pid" > "$lock/pid"
   ( . "$ROOT/bin/fm-wake-lib.sh"; fm_pid_identity "$sleep_pid" > "$lock/pid-identity" 2>/dev/null ) || true
   FM_HOME="$st" FM_STATE_OVERRIDE="$st/state" "$START" >/dev/null 2>&1
-  if [ -e "$st/state/.subsuper-escalations" ] && [ -e "$st/state/.subsuper-inject-wedged" ]; then
+  if [ -e "$st/state/.subsuper-escalations" ] \
+    && [ -e "$st/state/.subsuper-inject-wedged" ] \
+    && [ "$(cat "$st/state/.subsuper-seen-wake-seq")" = 12 ]; then
     pass "refresh: daemon already alive - stale artifacts preserved (current session's buffer kept)"
   else
     fail "refresh: incorrectly cleared the current session's buffered escalations"
@@ -218,12 +223,14 @@ unit_failed_start_rolls_back_state() {
   mkdir -p "$st/state"
   printf 'pending\n' > "$st/state/.subsuper-escalations"
   printf 'wedged\n' > "$st/state/.subsuper-inject-wedged"
+  printf '19\n' > "$st/state/.subsuper-seen-wake-seq"
   if FM_HOME="$st" FM_STATE_OVERRIDE="$st/state" FM_SUPERVISOR_TARGET=unused \
     FM_SUPERVISOR_BACKEND=unsupported "$LAUNCH" start >/dev/null 2>&1; then
     fail "failed start: unsupported backend unexpectedly succeeded"
   elif [ ! -e "$st/state/.afk" ] \
     && [ "$(cat "$st/state/.subsuper-escalations")" = pending ] \
-    && [ "$(cat "$st/state/.subsuper-inject-wedged")" = wedged ]; then
+    && [ "$(cat "$st/state/.subsuper-inject-wedged")" = wedged ] \
+    && [ "$(cat "$st/state/.subsuper-seen-wake-seq")" = 19 ]; then
     pass "failed start: away flag and delivery artifacts roll back"
   else
     fail "failed start: left false away state or discarded delivery artifacts"
diff --git a/tests/fm-afk-return.test.sh b/tests/fm-afk-return.test.sh
index 537b1bff97..4aebbe1b60 100755
--- a/tests/fm-afk-return.test.sh
+++ b/tests/fm-afk-return.test.sh
@@ -84,6 +84,9 @@ test_return_gate_orders_catchup_before_bearings() {
   date +%s > "$dir/home/state/.afk"
   printf 'repair-task.status: blocked synthetic dependency\n' > "$dir/home/state/.subsuper-escalations"
   printf 'fm away-mode inject WEDGED: 4555s undelivered\n' > "$dir/home/state/.subsuper-inject-wedged"
+  printf '2\n' > "$dir/home/state/.subsuper-seen-wake-seq"
+  printf '1784074271\t2\tcheck\trepair-check\tcheck: repair required\n' > "$dir/home/state/.subsuper-pending-wake"
+  printf '1784074271\t2\trepair-check\tcheck: repair required\n' > "$dir/home/state/.subsuper-seen-wake-checks"
   {
     printf '1784074271\t2\tsignal\trepair-task.status\tsignal: synthetic status\n'
     printf 'wake annotation: latest wake-EVENT observed at drain, not current state: repair-task.status: blocked synthetic dependency\n'
@@ -105,6 +108,8 @@ test_return_gate_orders_catchup_before_bearings() {
   [ "$(wc -l < "$dir/home/stop.log" | tr -d ' ')" -eq 1 ] || fail "return begin did not stop away mode exactly once"
   [ -s "$dir/home/state/.fake-drain" ] || fail "blocked return acknowledged its emitted wake before handling completed"
   [ ! -e "$dir/home/state/.fake-drain-acks" ] || fail "blocked return crossed the post-handling acknowledgement boundary"
+  [ -e "$dir/home/state/.subsuper-pending-wake" ] || fail "blocked return cleared the pending-wake journal before catch-up succeeded"
+  [ -e "$dir/home/state/.subsuper-seen-wake-checks" ] || fail "blocked return cleared the check dedupe ledger before catch-up succeeded"
 
   # The exact incident regression: Bearings is an ordinary request and must
   # refuse before reading/rendering while this shared gate remains open.
@@ -141,6 +146,9 @@ test_return_gate_orders_catchup_before_bearings() {
   [ ! -s "$dir/home/state/.fake-drain" ] || fail "explicit post-handling acknowledgement left the handled wake durable"
   [ "$(cat "$dir/home/state/.fake-drain-acks" 2>/dev/null || true)" = 2 ] \
     || fail "explicit post-handling acknowledgement used the wrong wake sequence"
+  [ ! -e "$dir/home/state/.subsuper-seen-wake-seq" ] || fail "successful check left the away queue cursor behind"
+  [ ! -e "$dir/home/state/.subsuper-pending-wake" ] || fail "successful check left the pending-wake journal behind"
+  [ ! -e "$dir/home/state/.subsuper-seen-wake-checks" ] || fail "successful check left the check dedupe ledger behind"
 
   out=$(run_return "$dir" check) || fail "an already-clear repeated check should be idempotent: $out"
   [ ! -e "$gate" ] || fail "idempotent clear check recreated a gate"
diff --git a/tests/fm-daemon.test.sh b/tests/fm-daemon.test.sh
index 2fe02fb431..5f50c8ceea 100755
--- a/tests/fm-daemon.test.sh
+++ b/tests/fm-daemon.test.sh
@@ -658,6 +658,109 @@ test_handle_wake_routes_self_and_escalate() {
   pass "handle_wake routes routine->self and captain->escalate"
 }
 
+test_away_queue_handoff_classifies_once_without_consuming() {
+  local dir state drain_out drain_err latest sequence generation
+  dir=$(make_supercase away-queue-handoff)
+  state="$dir/state"
+  drain_out="$dir/drain.out"
+  drain_err="$dir/drain.err"
+  : > "$state/.afk"
+  printf 'done: PR https://example.test/pr/afk-handoff\n' > "$state/handoff-done.status"
+  append_wake "$state" heartbeat heartbeat heartbeat
+  append_wake "$state" signal handoff-done "signal: $state/handoff-done.status"
+  append_wake "$state" signal handoff-done "signal: $state/handoff-done.status"
+  append_wake "$state" check handoff-review "check: review ready"
+  append_wake "$state" check handoff-review "check: review ready"
+
+  FM_STATE_OVERRIDE="$state" bash -c '
+    . "$1"
+    . "$2"
+    classify_queued_wakes "$3"
+  ' _ "$DAEMON" "$ROOT/bin/fm-wake-lib.sh" "$state" \
+    || fail "away daemon could not classify the handoff queue snapshot"
+
+  [ "$(wc -l < "$state/.wake-queue" | tr -d ' ')" -eq 5 ] \
+    || fail "away queue classifier consumed or lost durable records"
+  [ "$(cat "$state/.subsuper-seen-wake-seq")" = 5 ] \
+    || fail "away queue classifier did not advance its session cursor"
+  [ "$(wc -l < "$state/.subsuper-escalations" | tr -d ' ')" -eq 2 ] \
+    || fail "handoff duplicates were not compacted before escalation side effects"
+
+  FM_STATE_OVERRIDE="$state" bash -c '
+    . "$1"
+    . "$2"
+    classify_queued_wakes "$3"
+  ' _ "$DAEMON" "$ROOT/bin/fm-wake-lib.sh" "$state" \
+    || fail "away daemon could not repeat its queue classification idempotently"
+  [ "$(wc -l < "$state/.subsuper-escalations" | tr -d ' ')" -eq 2 ] \
+    || fail "away queue cursor delivered an actionable event twice"
+
+  append_wake "$state" check handoff-review "check: review ready with newer diagnostics"
+  FM_STATE_OVERRIDE="$state" bash -c '
+    . "$1"
+    . "$2"
+    classify_queued_wakes "$3"
+  ' _ "$DAEMON" "$ROOT/bin/fm-wake-lib.sh" "$state" \
+    || fail "away daemon could not classify a later handoff duplicate"
+  [ "$(cat "$state/.subsuper-seen-wake-seq")" = 6 ] \
+    || fail "away queue classifier did not advance past a later logical duplicate"
+  [ "$(wc -l < "$state/.subsuper-escalations" | tr -d ' ')" -eq 2 ] \
+    || fail "a same-key handoff observation with changed diagnostics was delivered twice"
+
+  latest=$(awk -F '\t' '$3 == "check" && $4 == "handoff-review" { row = $0 } END { print row }' "$state/.wake-queue")
+  printf '%s\n' "$latest" > "$state/.subsuper-pending-wake"
+  FM_STATE_OVERRIDE="$state" bash -c '
+    . "$1"
+    . "$2"
+    classify_queued_wakes "$3"
+  ' _ "$DAEMON" "$ROOT/bin/fm-wake-lib.sh" "$state" \
+    || fail "away daemon could not recover a crash after buffering before pending cleanup"
+  [ "$(wc -l < "$state/.subsuper-escalations" | tr -d ' ')" -eq 2 ] \
+    || fail "crash replay duplicated an already-buffered escalation"
+
+  append_wake "$state" check handoff-blocker "check: credentials required"
+  latest=$(awk -F '\t' '$3 == "check" && $4 == "handoff-blocker" { row = $0 } END { print row }' "$state/.wake-queue")
+  printf '%s\n' "$latest" > "$state/.subsuper-pending-wake"
+  printf '7\n' > "$state/.subsuper-seen-wake-seq"
+  FM_STATE_OVERRIDE="$state" bash -c '
+    . "$1"
+    . "$2"
+    classify_queued_wakes "$3"
+  ' _ "$DAEMON" "$ROOT/bin/fm-wake-lib.sh" "$state" \
+    || fail "away daemon could not recover a crash after cursor commit before escalation"
+  [ "$(wc -l < "$state/.subsuper-escalations" | tr -d ' ')" -eq 3 ] \
+    || fail "cursor-first crash recovery lost or duplicated its pending escalation"
+  [ ! -e "$state/.subsuper-pending-wake" ] \
+    || fail "crash recovery left its durable pending wake behind"
+
+  awk -F '\t' '$3 != "handoff-blocker" { print }' "$state/.subsuper-seen-wake-checks" \
+    > "$state/.subsuper-seen-wake-checks.tmp"
+  mv "$state/.subsuper-seen-wake-checks.tmp" "$state/.subsuper-seen-wake-checks"
+  printf '%s\n' "$latest" > "$state/.subsuper-pending-wake"
+  FM_STATE_OVERRIDE="$state" bash -c '
+    . "$1"
+    . "$2"
+    classify_queued_wakes "$3"
+  ' _ "$DAEMON" "$ROOT/bin/fm-wake-lib.sh" "$state" \
+    || fail "away daemon could not recover a crash after escalation before logical commit"
+  [ "$(wc -l < "$state/.subsuper-escalations" | tr -d ' ')" -eq 3 ] \
+    || fail "effect-first crash replay duplicated its buffered escalation"
+
+  FM_STATE_OVERRIDE="$state" "$ROOT/bin/fm-wake-drain.sh" > "$drain_out" 2> "$drain_err" \
+    || fail "handoff queue could not be presented by its sole drain owner"
+  [ "$(grep -c $'\theartbeat\t\|\tsignal\t\|\tcheck\t' "$drain_out")" -eq 4 ] \
+    || fail "handoff queue drain did not preserve each logical routine and actionable record"
+  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' "$drain_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' "$drain_err")
+  [ "$sequence" = 7 ] && [ -n "$generation" ] \
+    || fail "handoff queue drain did not require acknowledgement of its exact presented generation"
+  FM_STATE_OVERRIDE="$state" "$ROOT/bin/fm-wake-drain.sh" --ack-through "$sequence" \
+    --recovery-generation "$generation" \
+    || fail "handled handoff queue could not be acknowledged"
+  [ ! -s "$state/.wake-queue" ] || fail "handoff drain left consumed records in the queue"
+  pass "away handoff classifies routine/actionable queue rows once without consuming or duplicating them"
+}
+
 test_inject_skip_forces_self() {
   local dir state
   dir=$(make_supercase skip)
@@ -1864,6 +1967,7 @@ test_escalate_batches_into_one_digest
 test_escalate_batch_age_uses_first_append
 test_heartbeat_scan_dedup
 test_handle_wake_routes_self_and_escalate
+test_away_queue_handoff_classifies_once_without_consuming
 test_inject_skip_forces_self
 test_is_wake_reason_distinguishes_status_stdout
 test_terminal_stale_escalate_leaves_no_marker
diff --git a/tests/fm-pi-afk-handoff-live-e2e.test.sh b/tests/fm-pi-afk-handoff-live-e2e.test.sh
new file mode 100755
index 0000000000..50bb4918e6
--- /dev/null
+++ b/tests/fm-pi-afk-handoff-live-e2e.test.sh
@@ -0,0 +1,347 @@
+#!/usr/bin/env bash
+# Real installed-Pi/Herdr regression for the Pi-to-AFK watcher ownership handoff.
+# All Herdr lifecycle and task calls are isolated through fm-herdr-lab.sh.
+set -u
+
+# shellcheck source=tests/lib.sh
+. "$(dirname "${BASH_SOURCE[0]}")/lib.sh"
+
+if [ "${FM_PI_AFK_HANDOFF_LIVE_E2E:-0}" != 1 ]; then
+  echo "skip: set FM_PI_AFK_HANDOFF_LIVE_E2E=1 to run the real Pi AFK handoff regression"
+  exit 0
+fi
+
+for tool in herdr jq pi; do
+  command -v "$tool" >/dev/null 2>&1 || { echo "skip: $tool not found"; exit 0; }
+done
+
+LAB_HELPER=${HERDR_LAB_HELPER:-$ROOT/bin/fm-herdr-lab.sh}
+SESSION=${HERDR_LAB_SESSION:-$("$LAB_HELPER" name fm-pi-afk-handoff-live-e2e)}
+TMP_ROOT=$(fm_test_tmproot fm-pi-afk-handoff-live-e2e)
+HOME_DIR="$TMP_ROOT/home"
+STATE="$HOME_DIR/state"
+PROJECT="$TMP_ROOT/project"
+PI_DIR="$TMP_ROOT/pi-agent"
+FAKEBIN="$TMP_ROOT/fakebin"
+CAPTURE="$TMP_ROOT/pi-prompts.jsonl"
+ORIGINAL_PATH=$PATH
+PRIMARY_PANE=
+PRIMARY_TARGET=
+DAEMON_STARTED=0
+
+cleanup() {
+  local rc=$?
+  trap - EXIT
+  if [ "$DAEMON_STARTED" -eq 1 ]; then
+    PATH="$FAKEBIN:$ORIGINAL_PATH" HERDR_SESSION="$SESSION" FM_HOME="$HOME_DIR" FM_STATE_OVERRIDE="$STATE" \
+      FM_SUPERVISOR_BACKEND=herdr FM_SUPERVISOR_TARGET="$PRIMARY_TARGET" \
+      "$ROOT/bin/fm-afk-launch.sh" stop >/dev/null 2>&1 || true
+  fi
+  "$LAB_HELPER" teardown "$SESSION" || rc=1
+  fm_test_cleanup
+  exit "$rc"
+}
+trap cleanup EXIT
+"$LAB_HELPER" provision "$SESSION"
+
+mkdir -p "$HOME_DIR"/{state,data,config,projects} "$PROJECT" "$PI_DIR" "$FAKEBIN"
+printf '# Synthetic isolated Firstmate primary\n' > "$PROJECT/AGENTS.md"
+
+CAPTURE_EXT="$TMP_ROOT/capture-extension.ts"
+cat > "$CAPTURE_EXT" <<'TS'
+import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
+import { appendFileSync } from "node:fs";
+export default function (pi: ExtensionAPI) {
+  pi.registerProvider("fm-local", {
+    name: "Firstmate local regression",
+    baseUrl: "http://127.0.0.1:9/v1",
+    apiKey: "local-regression-only",
+    api: "openai-completions",
+    models: [{
+      id: "fm-local",
+      name: "Firstmate local regression",
+      reasoning: false,
+      input: ["text"],
+      cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
+      contextWindow: 8192,
+      maxTokens: 256,
+    }],
+  });
+  pi.on("project_trust", () => ({ trusted: "yes", remember: false }));
+  pi.on("before_agent_start", (event, ctx) => {
+    appendFileSync(process.env.FM_PI_CAPTURE_PATH!, `${JSON.stringify({ prompt: event.prompt })}\n`);
+    ctx.abort();
+  });
+}
+TS
+
+# Production backend calls inherit this shim; it accepts only this lab session
+# and routes every operation through the helper so no ambient/default call can
+# escape the test.
+cat > "$FAKEBIN/herdr" <<EOF
+#!/usr/bin/env bash
+set -euo pipefail
+helper='$LAB_HELPER'
+session='$SESSION'
+real_path='$ORIGINAL_PATH'
+args=("\$@")
+n=\${#args[@]}
+if [ "\$n" -ge 2 ] && [ "\${args[\$((n-2))]}" = --session ]; then
+  [ "\${args[\$((n-1))]}" = "\$session" ] || { echo 'wrapper refused foreign session' >&2; exit 97; }
+  args=("\${args[@]:0:\$((n-2))}")
+else
+  [ "\${HERDR_SESSION:-}" = "\$session" ] || { echo 'wrapper requires isolated session' >&2; exit 98; }
+fi
+PATH="\$real_path" exec "\$helper" run "\$session" "\${args[@]}"
+EOF
+chmod +x "$FAKEBIN/herdr"
+
+cat > "$TMP_ROOT/daemon-entry" <<EOF
+#!/usr/bin/env bash
+export PATH='$FAKEBIN:$ORIGINAL_PATH'
+export HERDR_SESSION='$SESSION'
+export FM_HOME='$HOME_DIR'
+export FM_STATE_OVERRIDE='$STATE'
+export FM_ROOT_OVERRIDE='$ROOT'
+export FM_ESCALATE_BATCH_SECS=0
+export FM_HOUSEKEEPING_TICK=1
+export FM_POLL=1
+export FM_SIGNAL_GRACE=0
+export FM_HEARTBEAT=999999
+export FM_CHECK_INTERVAL=999999
+export FM_STALE_ESCALATE_SECS=999999
+exec '$ROOT/bin/fm-afk-start.sh'
+EOF
+chmod +x "$TMP_ROOT/daemon-entry"
+
+PRIMARY_OUT=$("$LAB_HELPER" run "$SESSION" workspace create --cwd "$PROJECT" --label pi-afk-handoff --no-focus)
+PRIMARY_PANE=$(printf '%s' "$PRIMARY_OUT" | jq -r '.result.root_pane.pane_id')
+PRIMARY_TARGET="$SESSION:$PRIMARY_PANE"
+PI_CMD=$(printf 'printf "%%s\\n" "$$" > %q; exec env PI_CODING_AGENT_DIR=%q FM_HOME=%q FM_STATE_OVERRIDE=%q FM_ROOT_OVERRIDE=%q FM_PI_CAPTURE_PATH=%q FM_PI_AFK_HANDOFF_POLL_MS=20 FM_POLL=1 FM_SIGNAL_GRACE=0 pi --approve --model fm-local/fm-local --no-session --no-context-files --no-extensions -e %q -e %q' \
+  "$STATE/.lock" "$PI_DIR" "$HOME_DIR" "$STATE" "$ROOT" "$CAPTURE" "$CAPTURE_EXT" "$ROOT/.pi/extensions/fm-primary-pi-watch.ts")
+"$LAB_HELPER" run "$SESSION" pane run "$PRIMARY_PANE" "$PI_CMD" >/dev/null
+
+wait_for_idle() {
+  local stable=0 status _
+  for _ in $(seq 1 240); do
+    status=$("$LAB_HELPER" run "$SESSION" agent get "$PRIMARY_PANE" 2>/dev/null \
+      | jq -r '.result.agent.agent_status // empty' 2>/dev/null || true)
+    case "$status" in
+      idle|done|blocked) stable=$((stable + 1)); [ "$stable" -ge 3 ] && return 0 ;;
+      *) stable=0 ;;
+    esac
+    sleep 0.25
+  done
+  return 1
+}
+
+wait_for_file() {  # <path>
+  local path=$1 _
+  for _ in $(seq 1 240); do [ -s "$path" ] && return 0; sleep 0.05; done
+  return 1
+}
+
+watcher_pid() {
+  local pid
+  pid=$(cat "$STATE/.watch.lock/pid" 2>/dev/null || true)
+  case "$pid" in ''|*[!0-9]*) return 1 ;; esac
+  kill -0 "$pid" 2>/dev/null || return 1
+  printf '%s\n' "$pid"
+}
+
+wait_for_watcher_change() {  # <old-pid>
+  local old=$1 current _
+  for _ in $(seq 1 240); do
+    current=$(watcher_pid 2>/dev/null || true)
+    if [ -n "$current" ] && [ "$current" != "$old" ]; then printf '%s\n' "$current"; return 0; fi
+    sleep 0.05
+  done
+  return 1
+}
+
+wait_for_extension_watcher() {  # <old-pid> <pi-pid>
+  local old=$1 pi_pid=$2 current parent grand stable=0 prior='' _
+  for _ in $(seq 1 240); do
+    current=$(watcher_pid 2>/dev/null || true)
+    parent=$(ps -p "$current" -o ppid= 2>/dev/null | tr -d '[:space:]')
+    grand=$(ps -p "$parent" -o ppid= 2>/dev/null | tr -d '[:space:]')
+    if [ -n "$current" ] && [ "$current" != "$old" ] && [ "$grand" = "$pi_pid" ]; then
+      if [ "$current" = "$prior" ]; then stable=$((stable + 1)); else stable=1; prior=$current; fi
+      [ "$stable" -ge 3 ] && { printf '%s\n' "$current"; return 0; }
+    else
+      stable=0
+      prior=
+    fi
+    sleep 0.05
+  done
+  return 1
+}
+
+wait_for_pid_dead() {  # <pid>
+  local pid=$1 _
+  for _ in $(seq 1 160); do
+    kill -0 "$pid" 2>/dev/null || return 0
+    sleep 0.05
+  done
+  return 1
+}
+
+process_identity() {  # <pid>
+  ps -p "$1" -o ppid=,lstart=,args= 2>/dev/null | sed 's/^[[:space:]]*//'
+}
+
+queued_signal_count() {
+  [ -s "$STATE/.wake-queue" ] || { echo 0; return; }
+  awk -F '\t' '$3 == "signal" { count++ } END { print count + 0 }' "$STATE/.wake-queue"
+}
+
+queue_cursor() {
+  local cursor
+  cursor=$(cat "$STATE/.subsuper-seen-wake-seq" 2>/dev/null || true)
+  case "$cursor" in ''|*[!0-9]*) cursor=0 ;; esac
+  printf '%s\n' "$cursor"
+}
+
+wait_for_capture_kind() {  # <kind> <count>
+  local kind=$1 count=$2 seen _
+  for _ in $(seq 1 240); do
+    if [ -s "$CAPTURE" ]; then
+      seen=$(jq -s --arg needle "FIRSTMATE_OP: v1 $kind:" '[.[] | select(.prompt | contains($needle))] | length' "$CAPTURE" 2>/dev/null) || seen=0
+    else
+      seen=0
+    fi
+    [ "$seen" -ge "$count" ] && return 0
+    sleep 0.1
+  done
+  return 1
+}
+
+wait_for_empty_composer() {
+  local composer _
+  for _ in $(seq 1 160); do
+    composer=$(PATH="$FAKEBIN:$ORIGINAL_PATH" HERDR_SESSION="$SESSION" FM_HOME="$HOME_DIR" FM_STATE_OVERRIDE="$STATE" \
+      FM_SUPERVISOR_BACKEND=herdr FM_SUPERVISOR_TARGET="$PRIMARY_TARGET" \
+      bash -c '. "$1"; fm_backend_composer_state herdr "$2"' _ "$ROOT/bin/fm-backend.sh" "$PRIMARY_TARGET" 2>/dev/null || true)
+    [ "$composer" = empty ] && return 0
+    sleep 0.1
+  done
+  return 1
+}
+
+wait_for_idle || fail "real Pi primary did not become stably idle"
+wait_for_file "$STATE/.pi-watch-extension-loaded" || fail "real Pi did not load the production watcher extension"
+"$LAB_HELPER" run "$SESSION" pane send-text "$PRIMARY_PANE" '/fm-watch-arm-pi' >/dev/null
+"$LAB_HELPER" run "$SESSION" pane send-keys "$PRIMARY_PANE" enter >/dev/null
+EXT_WATCHER=$(wait_for_watcher_change "") || fail "Pi extension did not acquire the initial watcher cycle"
+wait_for_idle || fail "real Pi did not settle after the watcher command"
+wait_for_empty_composer || fail "real Pi composer was not empty before AFK entry"
+PI_PID=$(cat "$STATE/.lock")
+EXT_ARM=$(ps -eo pid=,ppid=,args= | awk -v parent="$PI_PID" '$2 == parent && /fm-watch-arm\.sh/ { print $1; exit }')
+[ -n "$EXT_ARM" ] || fail "could not identify the extension-owned arm child"
+EXT_ARM_IDENTITY=$(process_identity "$EXT_ARM")
+EXT_WATCHER_IDENTITY=$(cat "$STATE/.watch.lock/pid-identity")
+pass "real Pi extension acquired an ordinary home-scoped watcher cycle"
+
+PATH="$FAKEBIN:$ORIGINAL_PATH" HERDR_SESSION="$SESSION" FM_HOME="$HOME_DIR" FM_STATE_OVERRIDE="$STATE" \
+  FM_SUPERVISOR_BACKEND=herdr FM_SUPERVISOR_TARGET="$PRIMARY_TARGET" FM_AFK_LAUNCH_ENTRY="$TMP_ROOT/daemon-entry" \
+  "$ROOT/bin/fm-afk-launch.sh" start >/dev/null
+DAEMON_STARTED=1
+wait_for_file "$STATE/.supervise-daemon.pid" || fail "away daemon did not acquire its lifecycle"
+AWAY_WATCHER=$(wait_for_watcher_change "$EXT_WATCHER") || fail "away daemon did not acquire the watcher after Pi yielded"
+wait_for_pid_dead "$EXT_ARM" || [ "$(process_identity "$EXT_ARM")" != "$EXT_ARM_IDENTITY" ] \
+  || fail "the exact Pi extension arm identity survived the AFK ownership handoff"
+CURRENT_WATCHER_IDENTITY=$(FM_STATE_OVERRIDE="$STATE" bash -c '. "$1"; fm_pid_identity "$2"' _ "$ROOT/bin/fm-wake-lib.sh" "$EXT_WATCHER" 2>/dev/null || true)
+[ "$CURRENT_WATCHER_IDENTITY" != "$EXT_WATCHER_IDENTITY" ] \
+  || fail "the exact Pi extension watcher identity survived the AFK ownership handoff"
+grep -F "arm_pid=$EXT_ARM" "$STATE/.watch-cycle-exits.log" | grep -F $'reason=arm-interrupted' >/dev/null \
+  || fail "the exact arm did not record its signal-driven retirement"
+pass "real Pi yielded its exact cycle and the away daemon acquired monitoring"
+
+if [ -s "$CAPTURE" ]; then CAPTURE_BEFORE=$(wc -l < "$CAPTURE"); else CAPTURE_BEFORE=0; fi
+printf 'working: routine heartbeat-equivalent progress\n' > "$STATE/pi-afk-live.status"
+for _ in $(seq 1 240); do
+  [ "$(queued_signal_count)" -ge 1 ] && break
+  sleep 0.05
+done
+[ "$(queued_signal_count)" -ge 1 ] \
+  || fail "away watcher did not durably queue the routine signal"
+for _ in $(seq 1 240); do
+  [ "$(queue_cursor)" -ge "$(awk -F '\t' '$3 == "signal" { seq=$2 } END { print seq + 0 }' "$STATE/.wake-queue")" ] && break
+  sleep 0.05
+done
+ROUTINE_SEQ=$(awk -F '\t' '$3 == "signal" { seq=$2 } END { print seq + 0 }' "$STATE/.wake-queue")
+[ "$(queue_cursor)" -ge "$ROUTINE_SEQ" ] \
+  || fail "away daemon did not classify the routine queued wake"
+sleep 1.1
+if [ -s "$CAPTURE" ]; then CAPTURE_AFTER=$(wc -l < "$CAPTURE"); else CAPTURE_AFTER=0; fi
+[ "$CAPTURE_AFTER" -eq "$CAPTURE_BEFORE" ] || fail "routine away progress created a Pi agent turn"
+[ ! -s "$STATE/.subsuper-escalations" ] || fail "routine away progress was escalated"
+pass "real away daemon absorbed routine progress without Pi model injection"
+
+printf 'done: actionable AFK handoff result\n' >> "$STATE/pi-afk-live.status"
+for _ in $(seq 1 240); do
+  [ "$(queued_signal_count)" -ge 2 ] && break
+  sleep 0.05
+done
+[ "$(queued_signal_count)" -ge 2 ] \
+  || fail "away watcher did not durably queue the actionable signal"
+wait_for_empty_composer || fail "real Pi composer was not empty for actionable away delivery"
+wait_for_capture_kind away-supervisor 1 || {
+  pane=$("$LAB_HELPER" run "$SESSION" pane read "$PRIMARY_PANE" --source recent --lines 120 2>/dev/null || true)
+  fail "actionable away result did not reach Pi through the marked supervisor path; buffer=$(cat "$STATE/.subsuper-escalations" 2>/dev/null || true); daemon=$(tail -8 "$STATE/.supervise-daemon.log" 2>/dev/null || true); pane=$pane"
+}
+sleep 2
+AWAY_COUNT=$(jq -s '[.[] | select(.prompt | contains("FIRSTMATE_OP: v1 away-supervisor:"))] | length' "$CAPTURE")
+[ "$AWAY_COUNT" -eq 1 ] || fail "actionable away result was delivered $AWAY_COUNT times"
+SIGNAL_COUNT=$(queued_signal_count)
+# fm-watch intentionally retains both pre-grace and post-grace observations in
+# the raw at-least-once queue. The daemon cursor must classify all four rows,
+# while status-seen and drain dedupe ensure each logical update delivers once.
+[ "$SIGNAL_COUNT" -eq 4 ] \
+  || fail "the raw at-least-once queue did not retain both observations of each logical signal (count=$SIGNAL_COUNT)"
+FM_ROOT_OVERRIDE="$PROJECT" FM_HOME="$HOME_DIR" FM_STATE_OVERRIDE="$STATE" "$ROOT/bin/fm-wake-drain.sh" > "$TMP_ROOT/handoff-drain.out" \
+  || fail "handoff queue drain failed"
+[ "$(grep -c $'\tsignal\t' "$TMP_ROOT/handoff-drain.out")" -eq 1 ] \
+  || fail "the sole consumer did not apply its one-record-per-kind/key compaction"
+[ ! -s "$STATE/.wake-queue" ] || fail "the sole consumer left handoff records queued"
+pass "real actionable delivery stayed marked while the lossless queue drained to one logical signal"
+
+FM_ROOT_OVERRIDE="$PROJECT" PATH="$FAKEBIN:$ORIGINAL_PATH" HERDR_SESSION="$SESSION" FM_HOME="$HOME_DIR" FM_STATE_OVERRIDE="$STATE" \
+  FM_SUPERVISOR_BACKEND=herdr FM_SUPERVISOR_TARGET="$PRIMARY_TARGET" "$ROOT/bin/fm-afk-return.sh" begin >/dev/null \
+  || fail "away return did not complete"
+DAEMON_STARTED=0
+[ ! -e "$STATE/.afk" ] || fail "away return left the AFK marker active"
+RESUMED_WATCHER=$(wait_for_extension_watcher "$AWAY_WATCHER" "$PI_PID") \
+  || fail "loaded Pi extension did not resume monitoring automatically"
+
+printf 'done: post-return supervision wake\n' > "$STATE/pi-afk-resumed.status"
+for _ in $(seq 1 240); do [ "$(queued_signal_count)" -ge 1 ] && break; sleep 0.05; done
+[ "$(queued_signal_count)" -ge 1 ] || fail "resumed Pi watcher did not durably queue the ordinary wake"
+for _ in $(seq 1 240); do
+  grep -F "arm_pid=$(ps -p "$RESUMED_WATCHER" -o ppid= | tr -d '[:space:]')" "$STATE/.watch-cycle-exits.log" 2>/dev/null \
+    | grep -F $'reason=actionable-signal' | grep -E $'successor=started:[0-9]+' >/dev/null && break
+  sleep 0.05
+done
+grep -F "arm_pid=$(ps -p "$RESUMED_WATCHER" -o ppid= | tr -d '[:space:]')" "$STATE/.watch-cycle-exits.log" 2>/dev/null \
+  | grep -F $'reason=actionable-signal' | grep -E $'successor=started:[0-9]+' >/dev/null \
+  || fail "resumed Pi cycle did not record its actionable close and automatic successor"
+POST_RETURN_WATCHER=$(wait_for_extension_watcher "$RESUMED_WATCHER" "$PI_PID") \
+  || fail "resumed Pi cycle did not establish its automatic successor"
+pass "real return stopped away ownership and resumed extension monitoring without manual re-arm"
+
+# One more complete transition proves idempotent convergence and no duplicate
+# cycle on repeated entry/return.
+PATH="$FAKEBIN:$ORIGINAL_PATH" HERDR_SESSION="$SESSION" FM_HOME="$HOME_DIR" FM_STATE_OVERRIDE="$STATE" \
+  FM_SUPERVISOR_BACKEND=herdr FM_SUPERVISOR_TARGET="$PRIMARY_TARGET" FM_AFK_LAUNCH_ENTRY="$TMP_ROOT/daemon-entry" \
+  "$ROOT/bin/fm-afk-launch.sh" start >/dev/null
+DAEMON_STARTED=1
+SECOND_AWAY=$(wait_for_watcher_change "$POST_RETURN_WATCHER") || fail "repeated AFK entry did not converge to daemon ownership"
+FM_ROOT_OVERRIDE="$PROJECT" PATH="$FAKEBIN:$ORIGINAL_PATH" HERDR_SESSION="$SESSION" FM_HOME="$HOME_DIR" FM_STATE_OVERRIDE="$STATE" \
+  FM_SUPERVISOR_BACKEND=herdr FM_SUPERVISOR_TARGET="$PRIMARY_TARGET" "$ROOT/bin/fm-afk-return.sh" begin >/dev/null \
+  || fail "repeated away return did not complete"
+DAEMON_STARTED=0
+SECOND_RESUMED=$(wait_for_extension_watcher "$SECOND_AWAY" "$PI_PID") \
+  || fail "repeated return did not restore Pi ownership"
+sleep 1
+[ "$(watcher_pid)" = "$SECOND_RESUMED" ] || fail "repeated return launched a second concurrent watcher cycle"
+pass "real repeated Pi AFK entry and return converged to one extension-owned cycle"
diff --git a/tests/fm-pi-watch-extension.test.sh b/tests/fm-pi-watch-extension.test.sh
index 9e29adc793..8677e8f56d 100755
--- a/tests/fm-pi-watch-extension.test.sh
+++ b/tests/fm-pi-watch-extension.test.sh
@@ -24,7 +24,9 @@ install_pi_watch_extension_fixture() {
   cp "$ROOT/.pi/extensions/lib/fm-operational-input.ts" "$repo/.pi/extensions/lib/fm-operational-input.ts"
   mkdir -p "$repo/bin"
   cp "$ROOT/bin/fm-operational-input.sh" "$repo/bin/fm-operational-input.sh"
-  chmod +x "$repo/bin/fm-operational-input.sh"
+  cp "$ROOT/bin/fm-pi-arm-tree-retire.sh" "$repo/bin/fm-pi-arm-tree-retire.sh"
+  cp "$ROOT/bin/fm-wake-lib.sh" "$repo/bin/fm-wake-lib.sh"
+  chmod +x "$repo/bin/fm-operational-input.sh" "$repo/bin/fm-pi-arm-tree-retire.sh" "$repo/bin/fm-wake-lib.sh"
   cat > "$repo/node_modules/@earendil-works/pi-coding-agent/package.json" <<'JSON'
 {"name":"@earendil-works/pi-coding-agent","type":"module","exports":"./index.js"}
 JSON
@@ -95,6 +97,7 @@ if (!handler) {
   console.error("Pi watch command was not registered");
   process.exit(1);
 }
+
 const result = await handler("", {
   ui: {
     notify(message) {
@@ -137,6 +140,674 @@ EOF
   pass "Pi extension reports external healthy watcher output"
 }
 
+test_pi_afk_handoff_yields_and_resumes_exact_cycle() {
+  local repo home plugin log out status
+  repo="$TMP_ROOT/pi-afk-handoff-root"
+  home="$TMP_ROOT/pi-afk-handoff-home"
+  log="$TMP_ROOT/pi-afk-handoff.log"
+  mkdir -p "$repo/bin" "$home/state" "$home/config"
+  install_pi_watch_extension_fixture "$repo"
+  plugin="$repo/.pi/extensions/fm-primary-pi-watch.ts"
+  cat > "$repo/bin/fm-watch-arm.sh" <<'SH'
+#!/usr/bin/env bash
+printf 'start=%s predecessor=%s\n' "$$" "${FM_WATCH_PREDECESSOR_ARM_PID:-none}" >> "${FM_ARM_LOG:?}"
+trap 'printf "term=%s\n" "$$" >> "$FM_ARM_LOG"; exit 0' TERM INT
+while :; do sleep 0.02; done
+SH
+  chmod +x "$repo/bin/fm-watch-arm.sh"
+  out=$(PLUGIN="$plugin" FM_HOME="$home" FM_ROOT_OVERRIDE="$repo" FM_ARM_LOG="$log" FM_PI_AFK_HANDOFF_POLL_MS=5 FM_WATCH_REARM_RETRY_BASE_MS=5 FM_WATCH_REARM_RETRY_MAX_MS=10 node --input-type=module 2>&1 <<'EOF'
+import { existsSync, readFileSync, rmSync, writeFileSync } from "node:fs";
+import { pathToFileURL } from "node:url";
+
+const handlers = new Map();
+let tool = null;
+const prompts = [];
+const pi = {
+  on(event, handler) {
+    handlers.set(event, handler);
+  },
+  registerCommand() {},
+  registerTool(candidate) {
+    if (candidate.name === "fm_watch_arm_pi") tool = candidate;
+  },
+  sendUserMessage: async (message) => {
+    prompts.push(message);
+  },
+  events: { on() {} },
+};
+const state = `${process.env.FM_HOME}/state`;
+const afk = `${state}/.afk`;
+const rows = () => existsSync(process.env.FM_ARM_LOG)
+  ? readFileSync(process.env.FM_ARM_LOG, "utf8").trim().split(/\n/).filter(Boolean)
+  : [];
+const starts = () => rows().filter((line) => line.startsWith("start="));
+const terms = () => rows().filter((line) => line.startsWith("term="));
+async function waitFor(predicate, label) {
+  for (let i = 0; i < 500; i += 1) {
+    if (predicate()) return;
+    await new Promise((resolve) => setTimeout(resolve, 10));
+  }
+  throw new Error(`timeout waiting for ${label}: ${rows().join(" | ")}`);
+}
+
+writeFileSync(`${state}/.lock`, `${process.pid}\n`);
+writeFileSync(afk, "away\n");
+const mod = await import(pathToFileURL(process.env.PLUGIN).href);
+mod.default(pi);
+await handlers.get("session_start")?.({ type: "session_start", reason: "startup" }, {});
+const stoodDown = await tool.execute("startup-away", {}, undefined, undefined, {});
+if (!stoodDown.details?.ok || !stoodDown.content[0]?.text.includes("away mode owns supervision")) {
+  throw new Error(`startup-away did not stand down: ${JSON.stringify(stoodDown.details)}`);
+}
+await new Promise((resolve) => setTimeout(resolve, 80));
+if (starts().length !== 0) throw new Error(`startup-away armed ${starts().length} children`);
+
+rmSync(afk);
+await waitFor(() => starts().length === 1, "automatic first post-away arm");
+const firstPid = starts()[0].match(/start=(\d+)/)?.[1];
+if (!firstPid) throw new Error(`missing first child identity: ${starts()[0]}`);
+
+writeFileSync(afk, "away-again\n");
+await waitFor(() => terms().length === 1, "exact first arm retirement");
+if (terms()[0] !== `term=${firstPid}`) {
+  throw new Error(`handoff retired the wrong child: start=${firstPid} term=${terms()[0]}`);
+}
+await new Promise((resolve) => setTimeout(resolve, 60));
+if (starts().length !== 1) throw new Error(`extension re-armed while away: ${rows().join(" | ")}`);
+if (prompts.length !== 0) throw new Error(`exact-child yield injected ${prompts.length} Pi prompts`);
+
+const input = handlers.get("input");
+if (!input) throw new Error("Pi input handoff boundary was not registered");
+const watcher = input({
+  type: "input",
+  source: "extension",
+  text: "\u2063FIRSTMATE_OP: v1 watcher: routine heartbeat",
+});
+if (watcher?.action !== "handled") throw new Error(`away watcher input was not absorbed: ${JSON.stringify(watcher)}`);
+const actionable = input({
+  type: "input",
+  source: "extension",
+  text: "\u2063FIRSTMATE_OP: v1 away-supervisor: actionable blocker",
+});
+if (actionable?.action !== "continue") throw new Error(`away-supervisor input was suppressed: ${JSON.stringify(actionable)}`);
+
+rmSync(afk);
+await waitFor(() => starts().length === 2, "automatic first return arm");
+writeFileSync(afk, "away-third\n");
+await waitFor(() => terms().length === 2, "second exact arm retirement");
+rmSync(afk);
+await waitFor(() => starts().length === 3, "automatic repeated return arm");
+await new Promise((resolve) => setTimeout(resolve, 80));
+if (starts().length !== 3) throw new Error(`repeated convergence duplicated cycles: ${rows().join(" | ")}`);
+if (prompts.length !== 0) throw new Error(`repeated AFK lifecycle injected ${prompts.length} Pi prompts`);
+await handlers.get("session_shutdown")?.({ type: "session_shutdown", reason: "quit" }, {});
+await waitFor(() => terms().length === 3, "shutdown retirement");
+EOF
+)
+  status=$?
+  expect_code 0 "$status" "Pi AFK lifecycle must stand down, yield its exact child, absorb watcher input, and resume once"
+  [ -z "$out" ] || fail "Pi AFK handoff test printed output: $out"
+  pass "Pi AFK handoff yields the exact arm and resumes one cycle across repeated entry and return"
+}
+
+test_pi_afk_windows_handoff_retires_exact_tree() {
+  local repo home plugin fakebin proc_root log watcher_file taskkill_log out status
+  repo="$TMP_ROOT/pi-afk-windows-tree-root"
+  home="$TMP_ROOT/pi-afk-windows-tree-home"
+  fakebin="$TMP_ROOT/pi-afk-windows-tree-bin"
+  proc_root="$TMP_ROOT/pi-afk-windows-tree-proc"
+  log="$TMP_ROOT/pi-afk-windows-tree.log"
+  watcher_file="$TMP_ROOT/pi-afk-windows-tree-watcher"
+  taskkill_log="$TMP_ROOT/pi-afk-windows-taskkill.log"
+  mkdir -p "$repo/bin" "$home/state" "$home/config" "$fakebin" "$proc_root"
+  install_pi_watch_extension_fixture "$repo"
+  plugin="$repo/.pi/extensions/fm-primary-pi-watch.ts"
+  cat > "$repo/bin/fm-watch-arm.sh" <<'SH'
+#!/usr/bin/env bash
+. "$(dirname "$0")/fm-wake-lib.sh"
+mkdir -p "$FM_PROC_ROOT_OVERRIDE/$FM_FAKE_MSYS_PID"
+printf '%s\n' "$FM_FAKE_MSYS_PID" > "$FM_PI_ARM_WRAPPER_PID_FILE"
+printf '%s\n' "$$" > "$FM_PROC_ROOT_OVERRIDE/$FM_FAKE_MSYS_PID/winpid"
+printf 'FM_PI_ARM_TREE_TOKEN=%s\000' "$FM_PI_ARM_TREE_TOKEN" > "$FM_PROC_ROOT_OVERRIDE/$FM_FAKE_MSYS_PID/environ"
+sleep 300 &
+watcher=$!
+mkdir -p "$STATE/.watch.lock"
+printf '%s\n' "$watcher" > "$STATE/.watch.lock/pid"
+printf '%s\n' "$FM_HOME" > "$STATE/.watch.lock/fm-home"
+printf '%s\n' "$(dirname "$0")/fm-watch.sh" > "$STATE/.watch.lock/watcher-path"
+fm_pid_identity "$watcher" > "$STATE/.watch.lock/pid-identity"
+mkdir -p "$FM_PROC_ROOT_OVERRIDE/$watcher"
+printf '%s\n' "$watcher" > "$FM_PROC_ROOT_OVERRIDE/$watcher/winpid"
+printf '%s\n' "$watcher" > "${FM_FAKE_WATCHER_FILE:?}"
+printf 'start=%s watcher=%s\n' "$$" "$watcher" >> "${FM_ARM_LOG:?}"
+trap 'kill -TERM "$watcher" 2>/dev/null || true; wait "$watcher" 2>/dev/null || true; exit 0' TERM INT
+wait "$watcher"
+SH
+  cat > "$fakebin/taskkill.exe" <<'SH'
+#!/usr/bin/env bash
+printf '%s\n' "$*" >> "${FM_FAKE_TASKKILL_LOG:?}"
+pid=
+while [ "$#" -gt 0 ]; do
+  if [ "$1" = /PID ] && [ "$#" -ge 2 ]; then pid=$2; shift 2; continue; fi
+  shift
+done
+case "$pid" in ''|*[!0-9]*) exit 2 ;; esac
+watcher=$(cat "${FM_FAKE_WATCHER_FILE:?}")
+kill -TERM "$watcher" 2>/dev/null || true
+kill -TERM "$pid" 2>/dev/null || true
+msys_pid=$(cat "${FM_PI_ARM_WRAPPER_PID_FILE:?}" 2>/dev/null || true)
+rm -f "${FM_PROC_ROOT_OVERRIDE:?}/$msys_pid/winpid" "${FM_PROC_ROOT_OVERRIDE:?}/$msys_pid/environ"
+rm -f "${FM_PROC_ROOT_OVERRIDE:?}/$watcher/winpid"
+exit 0
+SH
+  chmod +x "$repo/bin/fm-watch-arm.sh" "$fakebin/taskkill.exe"
+  out=$(PATH="$fakebin:$PATH" PLUGIN="$plugin" FM_HOME="$home" FM_ROOT_OVERRIDE="$repo" FM_ARM_LOG="$log" \
+    FM_FAKE_WATCHER_FILE="$watcher_file" FM_FAKE_TASKKILL_LOG="$taskkill_log" FM_FAKE_MSYS_PID=424242 \
+    FM_PROC_ROOT_OVERRIDE="$proc_root" FM_PI_AFK_HANDOFF_POLL_MS=5 \
+    node --input-type=module 2>&1 <<'EOF'
+import { spawn, spawnSync } from "node:child_process";
+import { existsSync, readFileSync, readdirSync, writeFileSync } from "node:fs";
+import { pathToFileURL } from "node:url";
+
+Object.defineProperty(process, "platform", { value: "win32" });
+const handlers = new Map();
+let tool = null;
+const pi = {
+  on(event, handler) { handlers.set(event, handler); },
+  registerCommand() {},
+  registerTool(candidate) { if (candidate.name === "fm_watch_arm_pi") tool = candidate; },
+  sendUserMessage: async () => { throw new Error("Windows AFK handoff injected a Pi prompt"); },
+  events: { on() {} },
+};
+const state = `${process.env.FM_HOME}/state`;
+const identity = (pid) => {
+  const stat = spawnSync("ps", ["-p", String(pid), "-o", "stat="], { encoding: "utf8" }).stdout.trim();
+  if (!stat || /^Z/.test(stat)) return "";
+  return spawnSync("ps", ["-p", String(pid), "-o", "lstart=", "-o", "command="], {
+    encoding: "utf8",
+    env: { ...process.env, LC_ALL: "C" },
+  }).stdout.trim();
+};
+const waitFor = async (predicate, label) => {
+  for (let i = 0; i < 500; i += 1) {
+    if (predicate()) return;
+    await new Promise((resolve) => setTimeout(resolve, 10));
+  }
+  throw new Error(`timeout waiting for ${label}`);
+};
+
+writeFileSync(`${state}/.lock`, `${process.pid}\n`);
+const mod = await import(pathToFileURL(process.env.PLUGIN).href);
+mod.default(pi);
+await handlers.get("session_start")?.({ type: "session_start", reason: "startup" }, {});
+await tool.execute("arm", {}, undefined, undefined, {});
+await waitFor(() => existsSync(process.env.FM_ARM_LOG) && existsSync(process.env.FM_FAKE_WATCHER_FILE), "arm tree");
+const row = readFileSync(process.env.FM_ARM_LOG, "utf8").trim();
+const match = row.match(/^start=(\d+) watcher=(\d+)$/);
+if (!match) throw new Error(`invalid arm identities: ${row}`);
+const wrapperPid = Number(match[1]);
+const watcherPid = Number(match[2]);
+const wrapperIdentity = identity(wrapperPid);
+const watcherIdentity = identity(watcherPid);
+if (!wrapperIdentity || !watcherIdentity) throw new Error("arm tree identities were not live before handoff");
+const wrapperPidFiles = readdirSync(state).filter((name) => /^\.pi-arm-wrapper-[0-9a-f]+\.pid$/.test(name));
+if (wrapperPidFiles.length !== 1) throw new Error(`ambiguous wrapper PID publication: ${wrapperPidFiles.join(" | ")}`);
+const wrapperPidFile = `${state}/${wrapperPidFiles[0]}`;
+const msysPid = Number(readFileSync(wrapperPidFile, "utf8").trim());
+if (msysPid !== Number(process.env.FM_FAKE_MSYS_PID) || msysPid === wrapperPid) {
+  throw new Error(`wrapper did not publish its distinct MSYS PID: native=${wrapperPid} msys=${msysPid}`);
+}
+const sibling = spawn("sleep", ["300"], { stdio: "ignore" });
+let siblingIdentity = "";
+await waitFor(() => {
+  siblingIdentity = identity(sibling.pid);
+  return Boolean(siblingIdentity);
+}, "unrelated sibling identity");
+try {
+  writeFileSync(`${state}/.afk`, "away\n");
+  await waitFor(() => existsSync(process.env.FM_FAKE_TASKKILL_LOG), "Windows tree retirement");
+  await waitFor(() => identity(wrapperPid) !== wrapperIdentity && identity(watcherPid) !== watcherIdentity, "retired wrapper and watcher identities");
+  await waitFor(() => !existsSync(wrapperPidFile), "wrapper PID publication cleanup");
+  const taskkill = readFileSync(process.env.FM_FAKE_TASKKILL_LOG, "utf8").trim().split("\n");
+  if (taskkill[0] !== `/PID ${wrapperPid} /T /F` || taskkill.some((line) => line !== `/PID ${wrapperPid} /T /F` && line !== `/PID ${watcherPid} /T /F`)) {
+    throw new Error(`wrong rooted taskkill: ${taskkill.join(" | ")}`);
+  }
+  if (identity(sibling.pid) !== siblingIdentity) throw new Error("Windows tree retirement changed an unrelated sibling identity");
+} finally {
+  sibling.kill("SIGTERM");
+  await handlers.get("session_shutdown")?.({ type: "session_shutdown", reason: "quit" }, {});
+}
+EOF
+)
+  status=$?
+  expect_code 0 "$status" "Pi Windows AFK handoff must retire the exact wrapper and watcher identities"
+  [ -z "$out" ] || fail "Pi Windows AFK tree-retirement test printed output: $out"
+  pass "Pi Windows AFK handoff retires one identity-bounded arm tree"
+}
+
+test_pi_afk_windows_namespace_identity_refusals() {
+  local dir home state fakebin proc_root ready taskkill_log token watch_path wrapper watcher wrapper_expected watcher_expected
+  local msys_pid wrapper_pid_file out status wrapper_current watcher_current
+  dir="$TMP_ROOT/pi-afk-windows-namespace-refusals"
+  home="$dir/home"
+  state="$home/state"
+  fakebin="$dir/bin"
+  proc_root="$dir/proc"
+  ready="$dir/ready"
+  taskkill_log="$dir/taskkill.log"
+  token=0123456789abcdef0123456789abcdef
+  watch_path="$dir/fm-watch.sh"
+  msys_pid=434343
+  wrapper_pid_file="$state/.pi-arm-wrapper-$token.pid"
+  mkdir -p "$state" "$fakebin" "$proc_root/$msys_pid"
+  cat > "$fakebin/taskkill.exe" <<'SH'
+#!/usr/bin/env bash
+printf '%s\n' "$*" >> "${FM_FAKE_TASKKILL_LOG:?}"
+exit 0
+SH
+  chmod +x "$fakebin/taskkill.exe"
+  FM_HOME="$home" FM_STATE_OVERRIDE="$state" FM_PROC_ROOT_OVERRIDE="$proc_root" FM_PI_ARM_TREE_TOKEN="$token" \
+    FM_READY_FILE="$ready" FM_WATCH_PATH="$watch_path" bash -c '
+      . "$1"
+      sleep 300 &
+      watcher=$!
+      mkdir -p "$STATE/.watch.lock"
+      printf "%s\n" "$watcher" > "$STATE/.watch.lock/pid"
+      printf "%s\n" "$FM_HOME" > "$STATE/.watch.lock/fm-home"
+      printf "%s\n" "$FM_WATCH_PATH" > "$STATE/.watch.lock/watcher-path"
+      fm_pid_identity "$watcher" > "$STATE/.watch.lock/pid-identity"
+      printf "%s\n" "$watcher" > "$FM_READY_FILE"
+      wait "$watcher"
+    ' _ "$ROOT/bin/fm-wake-lib.sh" &
+  wrapper=$!
+  for _ in $(seq 1 300); do [ -s "$ready" ] && break; sleep 0.01; done
+  [ -s "$ready" ] || fail "Windows namespace refusal fixture did not publish its watcher"
+  watcher=$(cat "$ready")
+  wrapper_expected=$(FM_HOME="$home" FM_STATE_OVERRIDE="$state" FM_PROC_ROOT_OVERRIDE="$proc_root" \
+    bash -c '. "$1"; fm_pid_identity "$2"' _ "$ROOT/bin/fm-wake-lib.sh" "$wrapper") \
+    || fail "Windows namespace refusal fixture did not capture the wrapper identity"
+  watcher_expected=$(cat "$state/.watch.lock/pid-identity")
+
+  set +e
+  out=$(PATH="$fakebin:$PATH" FM_ROOT_OVERRIDE="$ROOT" FM_PROC_ROOT_OVERRIDE="$proc_root" FM_FAKE_TASKKILL_LOG="$taskkill_log" \
+    "$ROOT/bin/fm-pi-arm-tree-retire.sh" "$wrapper" "$token" "$state" "$watch_path" "$home" 2>&1)
+  status=$?
+  set -e
+  [ "$status" -ne 0 ] || fail "Windows retirement accepted an absent wrapper PID publication"
+
+  printf '%s\n%s\n' "$msys_pid" "$((msys_pid + 1))" > "$wrapper_pid_file"
+  set +e
+  out=$(PATH="$fakebin:$PATH" FM_ROOT_OVERRIDE="$ROOT" FM_PROC_ROOT_OVERRIDE="$proc_root" FM_FAKE_TASKKILL_LOG="$taskkill_log" \
+    "$ROOT/bin/fm-pi-arm-tree-retire.sh" "$wrapper" "$token" "$state" "$watch_path" "$home" 2>&1)
+  status=$?
+  set -e
+  [ "$status" -ne 0 ] || fail "Windows retirement accepted an ambiguous wrapper PID publication"
+
+  printf '%s\n' "$msys_pid" > "$wrapper_pid_file"
+  set +e
+  out=$(PATH="$fakebin:$PATH" FM_ROOT_OVERRIDE="$ROOT" FM_PROC_ROOT_OVERRIDE="$proc_root" FM_FAKE_TASKKILL_LOG="$taskkill_log" \
+    "$ROOT/bin/fm-pi-arm-tree-retire.sh" "$wrapper" "$token" "$state" "$watch_path" "$home" 2>&1)
+  status=$?
+  set -e
+  [ "$status" -ne 0 ] || fail "Windows retirement accepted an absent MSYS-to-native PID mapping"
+
+  printf '%s\n' "$((wrapper + 1))" > "$proc_root/$msys_pid/winpid"
+  printf 'FM_PI_ARM_TREE_TOKEN=%s\000' "$token" > "$proc_root/$msys_pid/environ"
+  set +e
+  out=$(PATH="$fakebin:$PATH" FM_ROOT_OVERRIDE="$ROOT" FM_PROC_ROOT_OVERRIDE="$proc_root" FM_FAKE_TASKKILL_LOG="$taskkill_log" \
+    "$ROOT/bin/fm-pi-arm-tree-retire.sh" "$wrapper" "$token" "$state" "$watch_path" "$home" 2>&1)
+  status=$?
+  set -e
+  [ "$status" -ne 0 ] || fail "Windows retirement accepted a mismatched native PID namespace mapping"
+
+  printf '%s\n' "$wrapper" > "$proc_root/$msys_pid/winpid"
+  printf 'FM_PI_ARM_TREE_TOKEN=reused\000' > "$proc_root/$msys_pid/environ"
+  set +e
+  out=$(PATH="$fakebin:$PATH" FM_ROOT_OVERRIDE="$ROOT" FM_PROC_ROOT_OVERRIDE="$proc_root" FM_FAKE_TASKKILL_LOG="$taskkill_log" \
+    "$ROOT/bin/fm-pi-arm-tree-retire.sh" "$wrapper" "$token" "$state" "$watch_path" "$home" 2>&1)
+  status=$?
+  set -e
+  [ "$status" -ne 0 ] || fail "Windows retirement accepted a reused MSYS wrapper identity"
+  [ ! -e "$taskkill_log" ] || fail "Windows namespace refusal reached taskkill: $(cat "$taskkill_log")"
+  [ -z "$out" ] || fail "Windows namespace refusal printed output: $out"
+  wrapper_current=$(FM_HOME="$home" FM_STATE_OVERRIDE="$state" FM_PROC_ROOT_OVERRIDE="$proc_root" \
+    bash -c '. "$1"; fm_pid_identity "$2"' _ "$ROOT/bin/fm-wake-lib.sh" "$wrapper" 2>/dev/null || true)
+  watcher_current=$(FM_HOME="$home" FM_STATE_OVERRIDE="$state" FM_PROC_ROOT_OVERRIDE="$proc_root" \
+    bash -c '. "$1"; fm_pid_identity "$2"' _ "$ROOT/bin/fm-wake-lib.sh" "$watcher" 2>/dev/null || true)
+  [ "$wrapper_current" = "$wrapper_expected" ] || fail "Windows namespace refusal changed the wrapper identity"
+  [ "$watcher_current" = "$watcher_expected" ] || fail "Windows namespace refusal changed the watcher identity"
+  kill -TERM "$watcher" 2>/dev/null || true
+  wait "$wrapper" 2>/dev/null || true
+  pass "Pi Windows retirement refuses absent, ambiguous, mismatched, and reused PID identities"
+}
+
+test_pi_windows_prepublication_partial_retirement_retries_before_rearm() {
+  local repo home plugin fakebin proc_root log taskkill_log publish_started partial allow identities out status
+  repo="$TMP_ROOT/pi-windows-retirement-retry-root"
+  home="$TMP_ROOT/pi-windows-retirement-retry-home"
+  fakebin="$TMP_ROOT/pi-windows-retirement-retry-bin"
+  proc_root="$TMP_ROOT/pi-windows-retirement-retry-proc"
+  log="$TMP_ROOT/pi-windows-retirement-retry.log"
+  taskkill_log="$TMP_ROOT/pi-windows-retirement-retry-taskkill.log"
+  publish_started="$TMP_ROOT/pi-windows-retirement-retry-publish-started"
+  partial="$TMP_ROOT/pi-windows-retirement-retry-partial"
+  allow="$TMP_ROOT/pi-windows-retirement-retry-allow"
+  identities="$TMP_ROOT/pi-windows-retirement-retry-identities"
+  mkdir -p "$repo/bin" "$home/state" "$home/config" "$fakebin" "$proc_root" "$identities"
+  install_pi_watch_extension_fixture "$repo"
+  plugin="$repo/.pi/extensions/fm-primary-pi-watch.ts"
+  cat > "$fakebin/mv" <<'SH'
+#!/usr/bin/env bash
+case "${*: -1}" in
+  *.pi-arm-wrapper-*.pid)
+    if [ ! -e "${FM_PUBLISH_STARTED:?}" ]; then
+      : > "$FM_PUBLISH_STARTED"
+      sleep 0.15
+    fi
+    ;;
+esac
+exec /bin/mv "$@"
+SH
+  cat > "$repo/bin/fm-watch-arm.sh" <<'SH'
+#!/usr/bin/env bash
+. "$(dirname "$0")/fm-wake-lib.sh"
+count=0
+[ ! -f "${FM_ARM_LOG:?}" ] || count=$(wc -l < "$FM_ARM_LOG")
+count=$((count + 1))
+msys_pid=$((${FM_FAKE_MSYS_BASE:?} + count))
+sleep 0.15
+mkdir -p "$FM_PROC_ROOT_OVERRIDE/$msys_pid"
+printf '%s\n' "$msys_pid" > "$FM_PI_ARM_WRAPPER_PID_FILE"
+printf '%s\n' "$$" > "$FM_PROC_ROOT_OVERRIDE/$msys_pid/winpid"
+printf 'FM_PI_ARM_TREE_TOKEN=%s\000' "$FM_PI_ARM_TREE_TOKEN" > "$FM_PROC_ROOT_OVERRIDE/$msys_pid/environ"
+sleep 300 &
+watcher=$!
+mkdir -p "$STATE/.watch.lock" "$FM_PROC_ROOT_OVERRIDE/$watcher"
+printf '%s\n' "$watcher" > "$STATE/.watch.lock/pid"
+printf '%s\n' "$FM_HOME" > "$STATE/.watch.lock/fm-home"
+printf '%s\n' "$(dirname "$0")/fm-watch.sh" > "$STATE/.watch.lock/watcher-path"
+fm_pid_identity "$watcher" > "$STATE/.watch.lock/pid-identity"
+printf '%s\n' "$watcher" > "$FM_PROC_ROOT_OVERRIDE/$watcher/winpid"
+fm_pid_identity "$$" > "${FM_IDENTITY_DIR:?}/$count.wrapper"
+fm_pid_identity "$watcher" > "$FM_IDENTITY_DIR/$count.watcher"
+printf '%s\t%s\t%s\n' "$$" "$watcher" "$msys_pid" >> "$FM_ARM_LOG"
+wait "$watcher"
+SH
+  cat > "$fakebin/taskkill.exe" <<'SH'
+#!/usr/bin/env bash
+pid=
+while [ "$#" -gt 0 ]; do
+  if [ "$1" = /PID ] && [ "$#" -ge 2 ]; then pid=$2; shift 2; continue; fi
+  shift
+done
+case "$pid" in ''|*[!0-9]*) exit 2 ;; esac
+printf '%s\n' "$pid" >> "${FM_FAKE_TASKKILL_LOG:?}"
+row=$(awk -F '\t' -v pid="$pid" '$1 == pid { value = $0 } END { print value }' "${FM_ARM_LOG:?}")
+if [ -n "$row" ]; then
+  watcher=$(printf '%s\n' "$row" | awk -F '\t' '{ print $2 }')
+  msys_pid=$(printf '%s\n' "$row" | awk -F '\t' '{ print $3 }')
+  kill -KILL "$pid" 2>/dev/null || true
+  rm -f "${FM_PROC_ROOT_OVERRIDE:?}/$msys_pid/winpid" "$FM_PROC_ROOT_OVERRIDE/$msys_pid/environ"
+  if [ ! -e "${FM_ALLOW_RETIREMENT:?}" ]; then
+    : > "${FM_PARTIAL_MARKER:?}"
+    exit 0
+  fi
+  kill -TERM "$watcher" 2>/dev/null || true
+  rm -f "$FM_PROC_ROOT_OVERRIDE/$watcher/winpid"
+  exit 0
+fi
+if [ ! -e "${FM_ALLOW_RETIREMENT:?}" ]; then
+  : > "${FM_PARTIAL_MARKER:?}"
+  exit 1
+fi
+kill -TERM "$pid" 2>/dev/null || true
+rm -f "${FM_PROC_ROOT_OVERRIDE:?}/$pid/winpid"
+exit 0
+SH
+  chmod +x "$fakebin/mv" "$fakebin/taskkill.exe" "$repo/bin/fm-watch-arm.sh"
+  out=$(PATH="$fakebin:$PATH" PLUGIN="$plugin" FM_HOME="$home" FM_ROOT_OVERRIDE="$repo" FM_PROC_ROOT_OVERRIDE="$proc_root" \
+    FM_ARM_LOG="$log" FM_FAKE_TASKKILL_LOG="$taskkill_log" FM_PUBLISH_STARTED="$publish_started" \
+    FM_PARTIAL_MARKER="$partial" FM_ALLOW_RETIREMENT="$allow" FM_IDENTITY_DIR="$identities" \
+    FM_FAKE_MSYS_BASE=454500 FM_PI_ARM_RETIRE_RETRY_MS=10 FM_PI_AFK_HANDOFF_POLL_MS=5 \
+    node --input-type=module 2>&1 <<'EOF'
+import { spawn, spawnSync } from "node:child_process";
+import { existsSync, readFileSync, readdirSync, writeFileSync } from "node:fs";
+import { pathToFileURL } from "node:url";
+
+Object.defineProperty(process, "platform", { value: "win32" });
+const handlers = new Map();
+let tool = null;
+const pi = {
+  on(event, handler) { handlers.set(event, handler); },
+  registerCommand() {},
+  registerTool(candidate) { if (candidate.name === "fm_watch_arm_pi") tool = candidate; },
+  sendUserMessage: async () => { throw new Error("Windows retirement retry injected a Pi prompt"); },
+  events: { on() {} },
+};
+const state = `${process.env.FM_HOME}/state`;
+const identity = (pid) => {
+  const stat = spawnSync("ps", ["-p", String(pid), "-o", "stat="], { encoding: "utf8" }).stdout.trim();
+  if (!stat || /^Z/.test(stat)) return "";
+  return spawnSync("ps", ["-p", String(pid), "-o", "lstart=", "-o", "command="], {
+    encoding: "utf8",
+    env: { ...process.env, LC_ALL: "C" },
+  }).stdout.trim();
+};
+const rows = () => existsSync(process.env.FM_ARM_LOG)
+  ? readFileSync(process.env.FM_ARM_LOG, "utf8").trim().split("\n").filter(Boolean)
+  : [];
+const waitFor = async (predicate, label) => {
+  for (let i = 0; i < 500; i += 1) {
+    if (predicate()) return;
+    await new Promise((resolve) => setTimeout(resolve, 10));
+  }
+  throw new Error(`timeout waiting for ${label}`);
+};
+const parse = (row) => row.split("\t").map(Number);
+
+writeFileSync(`${state}/.lock`, `${process.pid}\n`);
+const mod = await import(pathToFileURL(process.env.PLUGIN).href);
+mod.default(pi);
+await handlers.get("session_start")?.({ type: "session_start", reason: "startup" }, {});
+await tool.execute("arm", {}, undefined, undefined, {});
+await handlers.get("session_shutdown")?.({ type: "session_shutdown", reason: "new" }, {});
+if (!existsSync(process.env.FM_PUBLISH_STARTED)) throw new Error("shutdown did not overlap wrapper PID publication");
+await handlers.get("session_start")?.({ type: "session_start", reason: "new" }, {});
+await waitFor(
+  () => rows().length === 1 && existsSync(process.env.FM_PARTIAL_MARKER),
+  "retained partial retirement before automatic replacement",
+);
+const [wrapperPid, watcherPid] = parse(rows()[0]);
+const wrapperExpected = readFileSync(`${process.env.FM_IDENTITY_DIR}/1.wrapper`, "utf8").trim();
+const watcherExpected = readFileSync(`${process.env.FM_IDENTITY_DIR}/1.watcher`, "utf8").trim();
+if (identity(wrapperPid) === wrapperExpected) throw new Error("partial retirement left the exact wrapper identity running");
+if (identity(watcherPid) !== watcherExpected) throw new Error("partial-retirement fixture lost its exact surviving watcher identity");
+if (readdirSync(state).filter((name) => /^\.pi-arm-wrapper-[0-9a-f]+\.pid$/.test(name)).length !== 1) {
+  throw new Error("partial retirement did not retain one exact wrapper publication");
+}
+if (readdirSync(state).filter((name) => /^\.pi-arm-retirement-[0-9a-f]+$/.test(name)).length !== 1) {
+  throw new Error("partial retirement did not retain one exact identity snapshot");
+}
+const sibling = spawn("sleep", ["300"], { stdio: "ignore" });
+const siblingExpected = identity(sibling.pid);
+try {
+  writeFileSync(process.env.FM_ALLOW_RETIREMENT, "allow\n");
+  await waitFor(() => identity(watcherPid) !== watcherExpected, "exact watcher retirement retry");
+  await waitFor(() => rows().length === 2, "one automatic replacement arm");
+  await new Promise((resolve) => setTimeout(resolve, 100));
+  if (rows().length !== 2) throw new Error(`retirement convergence launched duplicate arms: ${rows().join(" | ")}`);
+  const [replacementWrapper, replacementWatcher] = parse(rows()[1]);
+  if (!identity(replacementWrapper) || !identity(replacementWatcher)) throw new Error("replacement cycle was not live after exact cleanup");
+  if (identity(sibling.pid) !== siblingExpected) throw new Error("retirement retry changed an unrelated sibling identity");
+  const activeWrapperPublications = readdirSync(state).filter((name) => /^\.pi-arm-wrapper-[0-9a-f]+\.pid$/.test(name));
+  const retainedSnapshots = readdirSync(state).filter((name) => /^\.pi-arm-retirement-[0-9a-f]+$/.test(name));
+  if (activeWrapperPublications.length !== 1 || retainedSnapshots.length !== 0) {
+    throw new Error(`confirmed predecessor retirement retained stale artifacts: ${[...activeWrapperPublications, ...retainedSnapshots].join(" | ")}`);
+  }
+  await handlers.get("session_shutdown")?.({ type: "session_shutdown", reason: "quit" }, {});
+  await waitFor(() => !identity(replacementWrapper) && !identity(replacementWatcher), "terminal replacement retirement");
+  await waitFor(
+    () => !readdirSync(state).some((name) => /^\.pi-arm-(?:wrapper|retirement)-/.test(name)),
+    "terminal retirement artifact cleanup",
+  );
+} finally {
+  sibling.kill("SIGTERM");
+}
+EOF
+)
+  status=$?
+  expect_code 0 "$status" "Pi Windows prepublication and partial retirement must converge before one replacement arm"
+  [ -z "$out" ] || fail "Pi Windows retirement-retry test printed output: $out"
+  pass "Pi Windows retains exact retirement through prepublication and partial-kill retries"
+}
+
+test_pi_afk_windows_partial_tree_retirement_fails_closed() {
+  local dir home state fakebin proc_root ready partial token watch_path msys_pid wrapper_pid_file wrapper watcher expected current out status
+  dir="$TMP_ROOT/pi-afk-windows-partial-tree"
+  home="$dir/home"
+  state="$home/state"
+  fakebin="$dir/bin"
+  proc_root="$dir/proc"
+  ready="$dir/ready"
+  partial="$dir/partial"
+  token=0123456789abcdef0123456789abcdef
+  watch_path="$dir/fm-watch.sh"
+  msys_pid=444444
+  wrapper_pid_file="$state/.pi-arm-wrapper-$token.pid"
+  mkdir -p "$state" "$fakebin" "$proc_root/$msys_pid"
+  cat > "$fakebin/taskkill.exe" <<'SH'
+#!/usr/bin/env bash
+pid=
+while [ "$#" -gt 0 ]; do
+  if [ "$1" = /PID ] && [ "$#" -ge 2 ]; then pid=$2; shift 2; continue; fi
+  shift
+done
+case "$pid" in ''|*[!0-9]*) exit 2 ;; esac
+if [ ! -e "${FM_PARTIAL_MARKER:?}" ]; then
+  kill -KILL "$pid" 2>/dev/null || true
+  msys_pid=$(cat "${FM_PI_ARM_WRAPPER_PID_FILE:?}" 2>/dev/null || true)
+  rm -f "${FM_PROC_ROOT_OVERRIDE:?}/$msys_pid/winpid" "${FM_PROC_ROOT_OVERRIDE:?}/$msys_pid/environ"
+  : > "$FM_PARTIAL_MARKER"
+  exit 0
+fi
+exit 1
+SH
+  chmod +x "$fakebin/taskkill.exe"
+  FM_HOME="$home" FM_STATE_OVERRIDE="$state" FM_PROC_ROOT_OVERRIDE="$proc_root" FM_PI_ARM_TREE_TOKEN="$token" \
+    FM_READY_FILE="$ready" FM_WATCH_PATH="$watch_path" \
+    bash -c '
+      . "$1"
+      sleep 300 &
+      watcher=$!
+      mkdir -p "$STATE/.watch.lock"
+      printf "%s\n" "$watcher" > "$STATE/.watch.lock/pid"
+      printf "%s\n" "$FM_HOME" > "$STATE/.watch.lock/fm-home"
+      printf "%s\n" "$FM_WATCH_PATH" > "$STATE/.watch.lock/watcher-path"
+      fm_pid_identity "$watcher" > "$STATE/.watch.lock/pid-identity"
+      {
+        printf "%s\n" "$watcher"
+        cat "$STATE/.watch.lock/pid-identity"
+      } > "$FM_READY_FILE"
+      wait "$watcher"
+    ' _ "$ROOT/bin/fm-wake-lib.sh" &
+  wrapper=$!
+  for _ in $(seq 1 300); do [ -s "$ready" ] && break; sleep 0.01; done
+  [ -s "$ready" ] || fail "partial Windows retirement fixture did not publish watcher identity"
+  watcher=$(head -1 "$ready")
+  expected=$(tail -n +2 "$ready")
+  printf '%s\n' "$msys_pid" > "$wrapper_pid_file"
+  printf '%s\n' "$wrapper" > "$proc_root/$msys_pid/winpid"
+  printf 'FM_PI_ARM_TREE_TOKEN=%s\000' "$token" > "$proc_root/$msys_pid/environ"
+  mkdir -p "$proc_root/$watcher"
+  printf '%s\n' "$watcher" > "$proc_root/$watcher/winpid"
+  set +e
+  out=$(PATH="$fakebin:$PATH" FM_ROOT_OVERRIDE="$ROOT" FM_PROC_ROOT_OVERRIDE="$proc_root" FM_PI_ARM_WRAPPER_PID_FILE="$wrapper_pid_file" FM_PARTIAL_MARKER="$partial" \
+    "$ROOT/bin/fm-pi-arm-tree-retire.sh" "$wrapper" "$token" "$state" "$watch_path" "$home" 2>&1)
+  status=$?
+  set -e
+  [ "$status" -ne 0 ] || fail "partial Windows tree kill was accepted despite a surviving watcher identity"
+  [ -e "$wrapper_pid_file" ] || fail "partial Windows tree kill cleared its retained wrapper identity"
+  [ -e "$state/.pi-arm-retirement-$token" ] || fail "partial Windows tree kill cleared its retained retirement snapshot"
+  current=$(FM_HOME="$home" FM_STATE_OVERRIDE="$state" FM_PROC_ROOT_OVERRIDE="$proc_root" bash -c '. "$1"; fm_pid_identity "$2"' \
+    _ "$ROOT/bin/fm-wake-lib.sh" "$watcher" 2>/dev/null || true)
+  [ "$current" = "$expected" ] || fail "partial Windows fixture did not retain the recorded watcher identity"
+  kill -TERM "$watcher" 2>/dev/null || true
+  wait "$wrapper" 2>/dev/null || true
+  [ -z "$out" ] || fail "partial Windows tree retirement printed output: $out"
+  pass "Pi Windows AFK retirement rejects a surviving recorded watcher"
+}
+
+test_pi_afk_handoff_is_home_scoped() {
+  local repo runner home_a home_b log_a log_b ready_a ready_b stop_a stop_b pid_a pid_b out status
+  repo="$TMP_ROOT/pi-afk-isolation-root"
+  runner="$repo/runner.mjs"
+  home_a="$TMP_ROOT/pi-afk-isolation-home-a"
+  home_b="$TMP_ROOT/pi-afk-isolation-home-b"
+  log_a="$TMP_ROOT/pi-afk-isolation-a.log"
+  log_b="$TMP_ROOT/pi-afk-isolation-b.log"
+  ready_a="$TMP_ROOT/pi-afk-isolation-a.ready"
+  ready_b="$TMP_ROOT/pi-afk-isolation-b.ready"
+  stop_a="$TMP_ROOT/pi-afk-isolation-a.stop"
+  stop_b="$TMP_ROOT/pi-afk-isolation-b.stop"
+  mkdir -p "$repo/bin" "$home_a/state" "$home_a/config" "$home_b/state" "$home_b/config"
+  install_pi_watch_extension_fixture "$repo"
+  cat > "$repo/bin/fm-watch-arm.sh" <<'SH'
+#!/usr/bin/env bash
+printf 'start=%s\n' "$$" >> "${FM_ARM_LOG:?}"
+trap 'printf "term=%s\n" "$$" >> "$FM_ARM_LOG"; exit 0' TERM INT
+while :; do sleep 0.02; done
+SH
+  chmod +x "$repo/bin/fm-watch-arm.sh"
+  cat > "$runner" <<'JS'
+import { existsSync, writeFileSync } from "node:fs";
+import { pathToFileURL } from "node:url";
+const handlers = new Map();
+let tool = null;
+const pi = {
+  on(event, handler) { handlers.set(event, handler); },
+  registerCommand() {},
+  registerTool(candidate) { if (candidate.name === "fm_watch_arm_pi") tool = candidate; },
+  sendUserMessage: async () => { throw new Error("isolated AFK handoff injected a Pi prompt"); },
+  events: { on() {} },
+};
+writeFileSync(`${process.env.FM_HOME}/state/.lock`, `${process.pid}\n`);
+const mod = await import(pathToFileURL(process.env.PLUGIN).href);
+mod.default(pi);
+await handlers.get("session_start")?.({ type: "session_start", reason: "startup" }, {});
+await tool.execute("arm", {}, undefined, undefined, {});
+writeFileSync(process.env.FM_READY, "ready\n");
+while (!existsSync(process.env.FM_STOP)) await new Promise((resolve) => setTimeout(resolve, 10));
+await handlers.get("session_shutdown")?.({ type: "session_shutdown", reason: "quit" }, {});
+await new Promise((resolve) => setTimeout(resolve, 80));
+JS
+  PLUGIN="$repo/.pi/extensions/fm-primary-pi-watch.ts" FM_HOME="$home_a" FM_ROOT_OVERRIDE="$repo" FM_ARM_LOG="$log_a" FM_READY="$ready_a" FM_STOP="$stop_a" FM_PI_AFK_HANDOFF_POLL_MS=5 node "$runner" >"$TMP_ROOT/isolation-a.out" 2>&1 &
+  pid_a=$!
+  PLUGIN="$repo/.pi/extensions/fm-primary-pi-watch.ts" FM_HOME="$home_b" FM_ROOT_OVERRIDE="$repo" FM_ARM_LOG="$log_b" FM_READY="$ready_b" FM_STOP="$stop_b" FM_PI_AFK_HANDOFF_POLL_MS=5 node "$runner" >"$TMP_ROOT/isolation-b.out" 2>&1 &
+  pid_b=$!
+  status=0
+  for _ in $(seq 1 300); do
+    [ -s "$ready_a" ] && [ -s "$ready_b" ] && [ -s "$log_a" ] && [ -s "$log_b" ] && break
+    sleep 0.01
+  done
+  if [ ! -s "$ready_a" ] || [ ! -s "$ready_b" ] || [ ! -s "$log_a" ] || [ ! -s "$log_b" ]; then
+    status=1
+  else
+    : > "$home_a/state/.afk"
+    for _ in $(seq 1 300); do grep -q '^term=' "$log_a" 2>/dev/null && break; sleep 0.01; done
+    grep -q '^term=' "$log_a" 2>/dev/null || status=1
+    grep -q '^term=' "$log_b" 2>/dev/null && status=1
+    kill -0 "$pid_b" 2>/dev/null || status=1
+  fi
+  : > "$stop_a"
+  : > "$stop_b"
+  wait "$pid_a" || status=1
+  wait "$pid_b" || status=1
+  out=$(cat "$TMP_ROOT/isolation-a.out" "$TMP_ROOT/isolation-b.out")
+  expect_code 0 "$status" "Pi AFK handoff must retire only the exact home-scoped extension child"
+  [ -z "$out" ] || fail "Pi sibling-home isolation test printed output: $out"
+  pass "Pi AFK handoff leaves a sibling home's live cycle untouched"
+}
+
 test_pi_tool_returns_agent_tool_result() {
   local repo home plugin out status
   repo="$TMP_ROOT/pi-tool-result-root"
@@ -2151,6 +2822,12 @@ EOF
 }
 
 test_pi_extension_reports_external_healthy_watcher
+test_pi_afk_handoff_yields_and_resumes_exact_cycle
+test_pi_afk_windows_handoff_retires_exact_tree
+test_pi_afk_windows_namespace_identity_refusals
+test_pi_windows_prepublication_partial_retirement_retries_before_rearm
+test_pi_afk_windows_partial_tree_retirement_fails_closed
+test_pi_afk_handoff_is_home_scoped
 test_pi_tool_returns_agent_tool_result
 test_pi_redundant_tool_call_is_owned_noop
 test_pi_scheduled_retry_call_is_owned_noop
diff --git a/tests/fm-turnend-guard.test.sh b/tests/fm-turnend-guard.test.sh
index ac02c7c37c..d35f8c2c49 100755
--- a/tests/fm-turnend-guard.test.sh
+++ b/tests/fm-turnend-guard.test.sh
@@ -1084,6 +1084,72 @@ EOF
   pass ".pi primary extension: delivery failure resets the logical-run latch"
 }
 
+test_pi_extension_suppresses_guard_turns_while_afk() {
+  local repo home ext log out status
+  repo="$TMP_ROOT/pi-afk-guard-root"
+  home="$TMP_ROOT/pi-afk-guard-home"
+  ext="$repo/.pi/extensions/fm-primary-turnend-guard.ts"
+  log="$TMP_ROOT/pi-afk-guard.log"
+  mkdir -p "$repo/.pi/extensions/lib" "$repo/bin" "$home/state"
+  cp "$ROOT/.pi/extensions/fm-primary-turnend-guard.ts" "$ext"
+  cp "$ROOT/.pi/extensions/lib/fm-operational-input.ts" "$repo/.pi/extensions/lib/fm-operational-input.ts"
+  cp "$ROOT/bin/fm-operational-input.sh" "$repo/bin/fm-operational-input.sh"
+  cat > "$repo/bin/fm-turnend-guard.sh" <<'SH'
+#!/usr/bin/env bash
+cat >/dev/null
+printf 'guard\n' >> "${FM_GUARD_LOG:?}"
+printf 'AFK guard probe\n' >&2
+exit 2
+SH
+  cat > "$repo/bin/fm-arm-pretool-check.sh" <<'SH'
+#!/usr/bin/env bash
+exit 0
+SH
+  chmod +x "$repo/bin/fm-turnend-guard.sh" "$repo/bin/fm-arm-pretool-check.sh" "$repo/bin/fm-operational-input.sh"
+  out=$(PLUGIN="$ext" FM_HOME="$home" FM_GUARD_LOG="$log" node --input-type=module 2>&1 <<'EOF'
+import { existsSync, rmSync, writeFileSync } from "node:fs";
+import { pathToFileURL } from "node:url";
+const handlers = new Map();
+let prompts = 0;
+const pi = {
+  on(event, handler) { handlers.set(event, handler); },
+  sendUserMessage: async () => { prompts += 1; },
+};
+const afk = `${process.env.FM_HOME}/state/.afk`;
+writeFileSync(afk, "away\n");
+const mod = await import(pathToFileURL(process.env.PLUGIN).href);
+mod.default(pi);
+const input = handlers.get("input");
+if (!input) throw new Error("turn-end input boundary was not registered");
+const guardInput = input({
+  type: "input",
+  source: "extension",
+  text: "\u2063FIRSTMATE_OP: v1 turn-end-guard: routine repair",
+});
+if (guardInput?.action !== "handled") throw new Error(`AFK guard input was not absorbed: ${JSON.stringify(guardInput)}`);
+const awayInput = input({
+  type: "input",
+  source: "extension",
+  text: "\u2063FIRSTMATE_OP: v1 away-supervisor: actionable result",
+});
+if (awayInput?.action !== "continue") throw new Error(`actionable away input was blocked: ${JSON.stringify(awayInput)}`);
+await handlers.get("agent_settled")?.({ type: "agent_settled" }, {});
+if (existsSync(process.env.FM_GUARD_LOG) || prompts !== 0) {
+  throw new Error(`AFK agent settlement ran guard=${existsSync(process.env.FM_GUARD_LOG)} prompts=${prompts}`);
+}
+rmSync(afk);
+await handlers.get("agent_settled")?.({ type: "agent_settled" }, {});
+if (!existsSync(process.env.FM_GUARD_LOG) || prompts !== 1) {
+  throw new Error(`attended guard did not resume: guard=${existsSync(process.env.FM_GUARD_LOG)} prompts=${prompts}`);
+}
+EOF
+)
+  status=$?
+  expect_code 0 "$status" "Pi turn-end guard must remain tokenless during AFK and resume after return"
+  [ -z "$out" ] || fail "Pi AFK turn-end guard test printed output: $out"
+  pass ".pi primary extension: AFK absorbs guard turns while actionable away delivery stays live"
+}
+
 # --- --claude cooperative mode -----------------------------------------------
 # In --claude mode the guard ignores stop_hook_active (Claude marks every stop
 # after ANY stop-hook continuation true, including asyncRewake rewake turns) and
@@ -1647,6 +1713,7 @@ test_codex_hook_ignores_nested_git_root_guard
 test_opencode_plugin_anchors_guard_to_worktree
 test_pi_extension_injects_once_per_logical_agent_run
 test_pi_extension_retries_after_followup_delivery_failure
+test_pi_extension_suppresses_guard_turns_while_afk
 test_hook_claude_mode_reblocks_stop_hook_active_when_unhealthy
 test_hook_claude_mode_reblocks_x_mode_without_tasks
 test_hook_claude_mode_allows_when_autoarm_owner_alive