fix: make Pi primary watcher away-mode aware and retire watcher markers on teardown

.opencode/plugins/fm-primary-turnend-guard.js

@@ -22,6 +22,10 @@ function runProcess(command, args, input = "") {
     });
     child.on("error", () => resolve({ code: 0, stdout: "", stderr: "" }));
     child.on("close", (code) => resolve({ code: code ?? 0, stdout, stderr }));
+    // A child that exits without reading its input breaks this pipe mid-write.
+    // Without a listener that EPIPE is an unhandled 'error' event and takes the
+    // whole host process down. The child's exit code and output still decide.
+    child.stdin.on("error", () => {});
     child.stdin.end(input);
   });
 }

.opencode/plugins/lib/fm-operational-input.js

@@ -25,6 +25,10 @@ export function encodeFirstmateOperationalInput(root, kind, content) {
       stderr += chunk.toString();
     });
     child.on("error", reject);
+    // An encoder that exits before reading its input breaks this pipe. Without
+    // a listener that EPIPE is an unhandled 'error' event and takes the whole
+    // host process down instead of rejecting this one call.
+    child.stdin.on("error", () => {});
     child.on("close", (code) => {
       if (code === 0 && stdout) {
         resolveResult(stdout);

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

@@ -10,7 +10,7 @@
 // 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 { existsSync, mkdirSync, readFileSync, 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";
@@ -126,6 +126,18 @@ function pidAlive(pid: string): boolean {
   }
 }
 
+// While state/.afk exists, the away daemon (bin/fm-supervise-daemon.sh) owns
+// fleet supervision, triage, and wake escalation. The Pi extension must not
+// spawn competing watcher children or deliver duplicate watcher wakes to the
+// primary session while away mode is active.
+function isAfkActive(): boolean {
+  try {
+    return existsSync(`${state}/.afk`);
+  } catch {
+    return false;
+  }
+}
+
 function lockOwnership(): LockOwnership {
   let lockPid = "";
   try {
@@ -243,7 +255,7 @@ export default function (pi: ExtensionAPI) {
     message: string,
     recovery?: { generation: string; watcherPid: string },
   ): Promise<void> {
-    if (!generationIsLive(owner)) return;
+    if (!generationIsLive(owner) || isAfkActive()) 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.`,
@@ -313,13 +325,15 @@ export default function (pi: ExtensionAPI) {
   }> {
     let failure = "";
     for (let attempt = 0; attempt <= retryLimit; attempt += 1) {
-      if (!generationIsLive(owner)) return { failure: "" };
+      if (!generationIsLive(owner) || isAfkActive()) return { failure: "" };
       const replacement = startArm(owner, predecessorArmPid);
+      if (isAfkActive()) return { failure: "" };
       const successorChild = owner.child;
       if (replacement.ok && successorChild && await waitForReadiness(successorChild)) {
         return { failure: "", recovery: armRecovery.get(successorChild) };
       }
       if (replacement.ok) {
+        if (isAfkActive()) return { failure: "" };
         failure = "watcher: FAILED - Pi extension could not verify a ready successor watcher";
         if (!(await retireArm(successorChild))) {
           return {
@@ -339,7 +353,7 @@ export default function (pi: ExtensionAPI) {
   }
 
   function scheduleRetry(owner: SessionGeneration, message: string, predecessorArmPid: string): void {
-    if (!generationIsLive(owner) || owner.child || owner.retryTimer) return;
+    if (!generationIsLive(owner) || owner.child || owner.retryTimer || isAfkActive()) 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}`);
@@ -352,9 +366,9 @@ export default function (pi: ExtensionAPI) {
     }
     const timer = setTimeout(() => {
       if (owner.retryTimer === timer) owner.retryTimer = null;
-      if (!generationIsLive(owner)) return;
+      if (!generationIsLive(owner) || isAfkActive()) return;
       const result = startArm(owner, predecessorArmPid);
-      if (!result.ok) {
+      if (!result.ok && !isAfkActive()) {
         surfaceFailure(owner, `watcher: FAILED - Pi extension could not launch a continuity retry\n${result.message}`);
       }
     }, retryDelay(owner.retryFailures));
@@ -364,6 +378,12 @@ export default function (pi: ExtensionAPI) {
 
   function startArm(owner: SessionGeneration, predecessorArmPid = ""): ArmResult {
     if (!generationIsLive(owner)) return { ok: false, message: shuttingDownMessage };
+    if (isAfkActive()) {
+      return {
+        ok: true,
+        message: "watcher: away mode active - supervise daemon owns watcher supervision",
+      };
+    }
     const ownership = lockOwnership();
     if (ownership === "other") return { ok: false, message: "watcher: read-only - session lock is held by another firstmate session" };
     if (ownership === "missing") {
@@ -445,6 +465,7 @@ export default function (pi: ExtensionAPI) {
       settleReadiness(false);
       releaseChild();
       if (!generationIsLive(owner)) return;
+      if (isAfkActive()) return;
       const classification = classifyClose(stdout, stderr, code, signal);
       const predecessor = String(armChild.pid ?? "");
       if (classification.kind === "actionable") {
@@ -453,7 +474,7 @@ export default function (pi: ExtensionAPI) {
         void (async () => {
           const restoration = await restoreAfterActionableClose(owner, predecessor);
           if (generationIsLive(owner)) owner.restoring = false;
-          if (!generationIsLive(owner)) return;
+          if (!generationIsLive(owner) || isAfkActive()) return;
           const message = restoration.failure ? `${classification.message}\n\n${restoration.failure}` : classification.message;
           await sendWake(owner, message, restoration.recovery);
         })().catch(() => {
@@ -469,7 +490,7 @@ export default function (pi: ExtensionAPI) {
       resolveClosed();
       settleReadiness(false);
       releaseChild();
-      if (!generationIsLive(owner)) return;
+      if (!generationIsLive(owner) || isAfkActive()) return;
       if (owner.restoring) return;
       scheduleRetry(owner, `watcher: FAILED - Pi extension arm child ${id} failed: ${error.message}`, String(armChild.pid ?? ""));
     });

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

@@ -163,6 +163,10 @@ function runGuard(): Promise<{ code: number; stderr: string }> {
     });
     child.on("error", () => resolveResult({ code: 0, stderr: "" }));
     child.on("close", (code) => resolveResult({ code: code ?? 0, stderr }));
+    // A guard that exits without reading its input breaks this pipe mid-write.
+    // Without a listener that EPIPE is an unhandled 'error' event and takes the
+    // whole host process down. The guard's exit code and stderr still decide.
+    child.stdin.on("error", () => {});
     child.stdin.end('{"stop_hook_active":false}');
   });
 }

bin/fm-brief.sh

@@ -278,14 +278,15 @@ Report only true captain-relevant outcomes or a declared external wait by append
 States: working, needs-decision, blocked, $PAUSED_VERB, done, failed.
 Use \`$PAUSED_VERB: {why}\` (distinct from \`blocked:\`) only when your domain is deliberately idling on a known external wait you expect to clear on its own; use \`blocked:\` when you are stuck and need firstmate to act.
 Use this only for material phase changes, a captain decision, a real blocker, a failure, or work ready for review.
+For a captain decision, append \`needs-decision [key=<slug>]: {summary of options}\`.
 This is also how you return the answer to a marked from-firstmate request above.
 A marked request requires one correlated answer after the work; it does not require a separate receipt or start acknowledgement.
 Never append \`working:\` merely to acknowledge receipt or announce that a marked request has started.
 When a routed-work phase has a supervisor-actionable material change worth reporting under the rule above, give that reported phase a stable key.
 If its first reportable event is \`working [key=<work-slug>]: {material phase}\`, use the same key on its later \`$PAUSED_VERB\`, \`done\`, \`failed\`, \`needs-decision\`, or \`blocked\` event so the earlier working phase is superseded.
 When a keyed phase ends without another reportable state, append \`resolved [key=<work-slug>]: {why it is no longer active}\`.
 \`resolved\` separately closes an escalated decision or blocker, and only a \`resolved\` line carrying that decision's exact key closes it: a later \`done\` or \`working\` event never does, even when the answer is what started that work.
-The main firstmate's answer normally writes that closing line at answer time; when a blocker or wait clears WITHOUT an answer from the main firstmate, append \`resolved: {how it cleared}\` yourself (keyed with \`[key=<slug>]\` if you opened it with one) as your domain resumes.
+The main firstmate's answer normally writes that closing line at answer time; when a blocker or wait clears WITHOUT an answer from the main firstmate, append \`resolved [key=<slug>]: {how it cleared}\` yourself (same \`[key=<slug>]\` if you opened it with one) as your domain resumes.
 Routine internal supervision, heartbeats, retries, and crewmate churn stay inside your own home and must not touch that status file.
 
 # Definition of done
@@ -368,9 +369,9 @@ The report is the only thing that survives, so anything worth keeping must be in
    treating it as a possible wedge. Use \`blocked:\` when you are stuck and need help.
 5. If you hit the same obstacle twice, append \`blocked: {why}\` and stop; firstmate will help.
 6. If a decision belongs to a human (product choices, destructive actions),
-   append \`needs-decision: {summary of options}\` and stop. Firstmate will reply with the decision.
+   append \`needs-decision [key=<slug>]: {summary of options}\` and stop. Firstmate will reply with the decision.
    A decision or blocker you opened stays open until a \`resolved\` line carrying its exact key lands; a later \`done:\` or \`working:\` line never closes it, even when the answer is what started that work.
-   Firstmate's reply normally writes that closing line at answer time; when a blocker or wait clears WITHOUT a firstmate reply, append \`resolved: {how it cleared}\` yourself (same \`[key=<slug>]\` if you opened it with one) as you resume.
+   Firstmate's reply normally writes that closing line at answer time; when a blocker or wait clears WITHOUT a firstmate reply, append \`resolved [key=<slug>]: {how it cleared}\` yourself (same \`[key=<slug>]\` if you opened it with one) as you resume.
 7. Never stop, restart, or update the shared \`no-mistakes\` daemon - it is one instance serving
    every lane/home, so restarting it kills other lanes' in-flight pipeline runs. On ANY no-mistakes
    daemon error, append \`blocked: {the daemon error}\` and stop; only firstmate manages the daemon.
@@ -456,11 +457,11 @@ land in the generation or the report.
    treating it as a possible wedge. Use \`blocked:\` when you are stuck and need help.
 5. If you hit the same obstacle twice, append \`blocked: {why}\` and stop; firstmate will help.
 6. If a decision belongs above you (product choices, destructive actions, ask-user findings),
-   append \`needs-decision: {summary of options}\` and stop. Firstmate will apply the configured authority and reply.
+   append \`needs-decision [key=<slug>]: {summary of options}\` and stop. Firstmate will apply the configured authority and reply.
    A decision or blocker you opened stays open until a \`resolved\` line carrying its exact key lands; a later \`done:\`
    or \`working:\` line never closes it, even when the answer is what started that work.
    Firstmate's reply normally writes that closing line at answer time; when a blocker or wait clears WITHOUT a firstmate reply,
-   append \`resolved: {how it cleared}\` yourself (same \`[key=<slug>]\` if you opened it with one) as you resume.
+   append \`resolved [key=<slug>]: {how it cleared}\` yourself (same \`[key=<slug>]\` if you opened it with one) as you resume.
 7. Never stop, restart, or update the shared \`no-mistakes\` daemon - it is one instance serving
    every lane/home, so restarting it kills other lanes' in-flight pipeline runs. On ANY no-mistakes
    daemon error, append \`blocked: {the daemon error}\` and stop; only firstmate manages the daemon.
@@ -581,9 +582,9 @@ $RULE1
    cadence instead of treating it as a possible wedge. Use \`blocked:\` when you are stuck and need help.
 5. If you hit the same obstacle twice, append \`blocked: {why}\` and stop; firstmate will help.
 6. If a decision belongs above the implementation worker (product choices, destructive actions, ask-user findings),
-   append \`needs-decision: {summary of options}\` and stop. Firstmate will apply the configured authority and reply with the decision.
+   append \`needs-decision [key=<slug>]: {summary of options}\` and stop. Firstmate will apply the configured authority and reply with the decision.
    A decision or blocker you opened stays open until a \`resolved\` line carrying its exact key lands; a later \`done:\` or \`working:\` line never closes it, even when the answer is what started that work.
-   Firstmate's reply normally writes that closing line at answer time; when a blocker or wait clears WITHOUT a firstmate reply, append \`resolved: {how it cleared}\` yourself (same \`[key=<slug>]\` if you opened it with one) as you resume.
+   Firstmate's reply normally writes that closing line at answer time; when a blocker or wait clears WITHOUT a firstmate reply, append \`resolved [key=<slug>]: {how it cleared}\` yourself (same \`[key=<slug>]\` if you opened it with one) as you resume.
 7. Never stop, restart, or update the shared \`no-mistakes\` daemon - it is one instance serving
    every lane/home, so restarting it kills other lanes' in-flight pipeline runs. On ANY no-mistakes
    daemon error, append \`blocked: {the daemon error}\` and stop; only firstmate manages the daemon.

bin/fm-remote-job-lib.sh

@@ -688,7 +688,11 @@ fm_remote_job_worker_ready_path() { printf '%s\n' "$FM_REMOTE_JOB_STATE/worker.r
 fm_remote_job_worker_identity_path() { printf '%s\n' "$FM_REMOTE_JOB_STATE/worker.identity"; }
 fm_remote_job_worker_lock_path() { printf '%s\n' "$FM_REMOTE_JOB_STATE/worker.lock"; }
 
-fm_remote_job_process_start() {
+# The wall-clock start stamp older records hold. procps derives it from a
+# sampled boot time, so a loaded machine reports it a second or more apart for
+# the same live process. Only read to recognize a record written before the
+# stable stamp below, and on platforms with no /proc to read.
+fm_remote_job_process_start_wall_clock() { # <pid>
   local pid=$1 ps_bin value
   if [ -x /bin/ps ]; then ps_bin=/bin/ps; elif [ -x /usr/bin/ps ]; then ps_bin=/usr/bin/ps; else return 1; fi
   value=$("$ps_bin" -p "$pid" -o lstart= 2>/dev/null) || return 1
@@ -697,6 +701,25 @@ fm_remote_job_process_start() {
   printf '%s\n' "$value"
 }
 
+# The pid-reuse guard for a recorded owner, so reading it twice for the same
+# live process has to give the same answer. The kernel's own start tick never
+# moves; the drifting wall-clock form made a healthy lock owner read as a
+# different process, which left ensure starting a second worker that could
+# never take the lock from the first.
+fm_remote_job_process_start() { # <pid>
+  local pid=$1 value
+  if [ -r "/proc/$pid/stat" ]; then
+    # Field 22 is starttime. comm (field 2) is parenthesized and can hold
+    # spaces, so drop everything through its closing paren before counting.
+    value=$(awk '{ sub(/^.*\) /, ""); print $20 }' "/proc/$pid/stat" 2>/dev/null || true)
+    case "$value" in
+      ''|*[!0-9]*) : ;;
+      *) printf 'boot-tick:%s\n' "$value"; return 0 ;;
+    esac
+  fi
+  fm_remote_job_process_start_wall_clock "$pid"
+}
+
 fm_remote_job_process_command() {
   local pid=$1 ps_bin value
   if [ -x /bin/ps ]; then ps_bin=/bin/ps; elif [ -x /usr/bin/ps ]; then ps_bin=/usr/bin/ps; else return 1; fi
@@ -798,7 +821,11 @@ fm_remote_job_lock_owner_matches_process() {
   [ "$pid" -gt 1 ] || return 1
   recorded_start=$(fm_remote_job_read_single_line "$lock/start" 256) || return 1
   actual_start=$(fm_remote_job_process_start "$pid") || return 1
-  [ "$recorded_start" = "$actual_start" ] || return 1
+  if [ "$recorded_start" != "$actual_start" ]; then
+    # A lock published before the stable stamp landed holds the wall-clock form.
+    actual_start=$(fm_remote_job_process_start_wall_clock "$pid") || return 1
+    [ "$recorded_start" = "$actual_start" ] || return 1
+  fi
   recorded_command=$(fm_remote_job_read_single_line "$lock/command" 8192) || return 1
   actual_command=$(fm_remote_job_process_command "$pid") || return 1
   [ "$recorded_command" = "$actual_command" ] || return 1
@@ -932,6 +959,33 @@ fm_remote_job_reload_launchagent() { # <account-home> <uid>
   fi
 }
 
+# Stop every Linux worker supervisor launched from <root>, by pid. owned_alive
+# reads worker.pid (the serving child). When that file is stale, a later start
+# would leave the old supervisor running in its own process group; under load
+# that duplicates into dozens of supervisors that never share the job queue.
+fm_remote_job_stop_stray_linux_workers() { # <root>
+  local root=$1 uid pid cmdline worker proc
+  uid=$(id -u 2>/dev/null || true)
+  case "$uid" in ''|*[!0-9]*) return 0 ;; esac
+  root=${root%/}
+  [ -n "$root" ] || return 0
+  worker="$root/bin/fm-remote-job-worker.sh"
+  for proc in /proc/[0-9]*; do
+    pid=${proc#/proc/}
+    case "$pid" in ''|*[!0-9]*) continue ;; esac
+    [ -r "$proc/cmdline" ] || continue
+    [ "$(awk '/^Uid:/{print $2; exit}' "$proc/status" 2>/dev/null)" = "$uid" ] || continue
+    cmdline=$(tr '\0' ' ' < "$proc/cmdline" 2>/dev/null) || continue
+    case "$cmdline" in
+      *"$worker --serve"*) continue ;;
+      *"$worker"*)
+        fm_remote_job_stop_worker_tree "$pid" || true
+        kill -KILL "$pid" 2>/dev/null || true
+        ;;
+    esac
+  done
+}
+
 fm_remote_job_start_linux_worker() { # <remote-root> <account-home>
   local root=$1 account_home=$2 worker pid
   worker="$root/bin/fm-remote-job-worker.sh"
@@ -953,6 +1007,7 @@ fm_remote_job_start_linux_worker() { # <remote-root> <account-home>
     wait "$pid" 2>/dev/null || true
     FM_REMOTE_JOB_REPAIRED=1
   fi
+  fm_remote_job_stop_stray_linux_workers "$root"
   # Job control puts the worker tree in its own process group, so a later stop
   # can signal every descendant at once without ever reaching the caller's own
   # group. Without this the group of a leaked worker is the launching command's.