fix(pi): recover captain input and replies lost to Pi's turn-start race

.gitignore

@@ -4,6 +4,7 @@ data/
 scratchpad*
 .no-mistakes/
 .lavish/
+.squish/
 .fm-secondmate-home
 .fm-secondmate-parent
 .DS_Store

docs/documentation-audiences.json

@@ -420,6 +420,10 @@
       "path": "docs/verification/muse.md",
       "audience": "maintainer-verification"
     },
+    {
+      "path": "docs/verification/pi-watch-extension-reply-recovery.md",
+      "audience": "maintainer-verification"
+    },
     {
       "path": "docs/verification/process-event-sources.md",
       "audience": "maintainer-verification"

docs/turnend-guard.md

@@ -52,7 +52,8 @@ If `jq` is missing or hook stdin is empty, the guard exits 0 because it cannot s
 - Claude registers two `Stop` hooks in `.claude/settings.json`, both anchored through `CLAUDE_PROJECT_DIR`: `bin/fm-turnend-guard.sh --claude`, and `bin/fm-claude-stop-autoarm.sh` with `asyncRewake: true` and `timeout: 28800`.
 - Codex registers a `Stop` hook in `.codex/hooks.json`, anchors the executable to the hook process working directory, verifies a Firstmate-shaped hook-bearing root, and passes the original payload to the shared guard.
 - OpenCode listens for `session.idle` in `.opencode/plugins/fm-primary-turnend-guard.js`, lets the watcher coordinator act first, and calls `client.session.promptAsync` once when the guard returns 2.
-- Pi listens for `agent_settled` in `.pi/extensions/fm-primary-turnend-guard.ts`, runs once per logical agent run, and calls `pi.sendUserMessage(..., { deliverAs: "followUp" })` once when the guard returns 2.
+- Pi listens for `agent_settled` in `.pi/extensions/fm-primary-turnend-guard.ts`, evaluates only the settle that drains its own count of logical runs in flight, and calls `pi.sendUserMessage(..., { deliverAs: "followUp" })` once when the guard returns 2.
+  The same handler owns one further follow-up path, taken only on a clean guard verdict and never alongside the guard's own, which recovers a captain message or a reply lost to Pi's turn-start race; [`watcher-continuity.md`](watcher-continuity.md#turn-settle-input-and-reply-recovery) owns that contract.
 - Cursor registers a `stop` hook in `.cursor/hooks.json` and delegates the whole turn boundary to `bin/fm-turnend-guard-cursor.sh`, the park described below.
   Cursor also loads `<project>/.claude/settings.json`, so every tracked Claude-shaped entrypoint whose event Cursor covers stands down on a Cursor-delivered payload through `bin/fm-hook-host-lib.sh`.
   That predicate reads the delivered payload's own `cursor_version`, never the environment: Cursor exports `CURSOR_INVOKED_AS`, `CURSOR_PROJECT_DIR`, and `CURSOR_VERSION` into every child process, so an environment guard would also disable the hooks of a Claude session started by hand from a Cursor pane, which is the hazard the `GROK_SESSION_ID` exclusion below records.
@@ -160,6 +161,7 @@ That warning uses `bin/fm-supervision-instructions.sh --repair-line`, so it alwa
 ## Regression coverage
 
 `tests/fm-turnend-guard.test.sh` covers the predicate, main and secondmate primary scope, child-worktree exclusion, `FM_HOME` and `FM_STATE_OVERRIDE` precedence, the live-lock and fresh-beacon guard predicate, the cooperative `--claude` open-generation claim wait, monotonic failed-epoch progression, bounded attended fail-open, post-alarm continuation suppression, positive recovery reset, generation and legacy claim cases that must block or clear instead of allowing a blind stop, Pi logical-run latching, missing-`jq` behavior, all five primary registrations, Grok native and legacy selection, typed field precedence, malformed input, and exactly-one-path safety.
+Its `test_pi_input_recovery_*` and `test_pi_reply_recovery_*` cases belong to the turn-settle recovery contract owned by [`watcher-continuity.md`](watcher-continuity.md#turn-settle-input-and-reply-recovery).
 `tests/fm-guard-stale-banner.test.sh` covers the pull-guard predicate, including the persistent-model fresh-leftover-beacon negative control, the auto-arm model's healthy fresh-beacon-without-a-watcher case and stale-beacon alarm, and the extension model's live-watcher path, ownership-qualified fresh hand-off, held-lock failures, independently broken ownership signals, stale-beacon alarm, queued-wake warning, and Pi and pi-signed harness routing.
 It also covers true-reason banner wording and reason-keyed episode dedup surviving a beacon mtime change.
 `tests/fm-cursor-primary.test.sh` covers the Cursor park end to end over real processes with no harness installed: each tracked Claude-shaped entrypoint standing down on a Cursor payload, both follow-up sources, the bounded repair nag and its reset, the nested loop bounds, supersession, away-mode and lock-ownership inertness, Pi-host stand-down without Cursor identity and continued parking when `PI_CODING_AGENT` leaks alongside `CURSOR_AGENT` or `CURSOR_INVOKED_AS`, child-worktree exclusion, and that the adapter never exits 2.

docs/verification/pi-agent-session-toctou-repro.mjs

@@ -0,0 +1,181 @@
+// Reproduction: AgentSession.prototype.prompt() has no atomic check-and-set
+// between reading `isStreaming` and committing to a new agent run inside
+// _runAgentPrompt(). Two concurrent prompt() calls that are both idle at the
+// moment they check isStreaming can both fall through to _runAgentPrompt(),
+// which unconditionally sets _isAgentRunActive = true and invokes
+// this.agent.prompt(messages) again - a genuine concurrent double-invocation.
+//
+// This drives the REAL, unmodified AgentSession.prototype.prompt from the
+// installed @earendil-works/pi-coding-agent package against a minimal stub
+// `this`, so the method under test is production code, not a reimplementation.
+import { pathToFileURL } from "node:url";
+
+const SDK_PATH = process.env.SDK_PATH;
+const { AgentSession } = await import(pathToFileURL(SDK_PATH).href);
+const promptFn = AgentSession.prototype.prompt;
+
+let concurrentRunAgentPromptCalls = 0;
+let maxConcurrentRunAgentPromptCalls = 0;
+const events = [];
+// Extension-visible event order, exactly as .pi/extensions/fm-primary-turnend-guard.ts
+// receives it. The losing prompt() call emits a settle of its own while the
+// winner is still running, so a settle is NOT proof that the session is idle.
+const extensionEvents = [];
+let settlesWhileAnotherRunWasLive = 0;
+// Everything the losing prompt() call manages to append. pi-agent-core's
+// Agent.prototype.prompt throws before normalizePromptInput/runPromptMessages,
+// so a losing captain message never becomes a transcript entry at all.
+const transcript = [];
+const capturedInputs = [];
+
+function log(label) {
+  events.push(`${(performance.now()).toFixed(2)}ms ${label}`);
+}
+
+// Faithful to agent-session.js lines 747-760 (_runAgentPrompt): sets
+// _isAgentRunActive = true as its very first (synchronous) statement, then
+// awaits the underlying agent run.
+async function fakeRunAgentPrompt(messages) {
+  this._isAgentRunActive = true;
+  concurrentRunAgentPromptCalls++;
+  maxConcurrentRunAgentPromptCalls = Math.max(maxConcurrentRunAgentPromptCalls, concurrentRunAgentPromptCalls);
+  const isLoser = concurrentRunAgentPromptCalls > 1;
+  log(`_runAgentPrompt ENTER (concurrent=${concurrentRunAgentPromptCalls}) messages=${JSON.stringify(messages).slice(0, 60)}`);
+  try {
+    if (isLoser) {
+      // pi-agent-core Agent.prototype.prompt (dist/agent.js) rejects at once
+      // when activeRun is set: "Agent is already processing a prompt." Nothing
+      // is appended, which is why the caller's message simply disappears.
+      throw new Error("Agent is already processing a prompt. Use steer() or followUp() to queue messages, or wait for completion.");
+    }
+    // Only the winner ever reaches the append path (message_end persistence).
+    for (const message of messages) transcript.push(message);
+    // Simulate real inference/tool-call latency.
+    await new Promise((r) => setTimeout(r, 40));
+    transcript.push({ role: "assistant", content: [{ type: "text", text: "Wake abgearbeitet." }], stopReason: "stop" });
+  } finally {
+    concurrentRunAgentPromptCalls--;
+    // agent-session.js _runAgentPrompt's finally block runs _emitAgentSettled()
+    // unconditionally, which clears _isAgentRunActive and emits agent_settled
+    // to every extension - even for a run that never produced anything.
+    this._isAgentRunActive = false;
+    log(`_runAgentPrompt EXIT -> _emitAgentSettled()`);
+    if (concurrentRunAgentPromptCalls > 0) settlesWhileAnotherRunWasLive++;
+    extensionEvents.push(`agent_settled(runsStillLive=${concurrentRunAgentPromptCalls})`);
+  }
+}
+
+function makeStubSession() {
+  return {
+    _isAgentRunActive: false,
+    get isStreaming() {
+      return this._isAgentRunActive;
+    },
+    _compactionAbortController: undefined,
+    _pendingNextTurnMessages: [],
+    _systemPromptOverride: undefined,
+    _baseSystemPrompt: "base",
+    promptTemplates: [],
+    model: { provider: "test" },
+    _modelRuntime: {
+      hasConfiguredAuth: () => true,
+      checkAuth: async () => "ok",
+      isUsingOAuth: () => false,
+    },
+    _extensionRunner: {
+      // The turn-end guard registers an `input` handler, so this reports true
+      // and prompt() really does emit the event - before its isStreaming
+      // check, and therefore for the losing call too.
+      hasHandlers: (event) => event === "input",
+      emitInput: async (text, images, source) => {
+        capturedInputs.push({ text, source });
+        return { action: "pass" };
+      },
+      // A REAL before_agent_start handler in this repo
+      // (.pi/extensions/fm-primary-turnend-guard.ts) awaits a spawned child
+      // process here. This delay stands in for that genuine async gap - it is
+      // not a contrived one - and is exactly the window a concurrently-fired
+      // watcher wake (fm-primary-pi-watch.ts sendWake) races against.
+      emitBeforeAgentStart: async () => {
+        log("emitBeforeAgentStart (simulating a real extension awaiting a child process)");
+        extensionEvents.push("before_agent_start");
+        await new Promise((r) => setTimeout(r, 20));
+        return undefined;
+      },
+    },
+    _findLastAssistantMessage: () => undefined,
+    _checkCompaction: async () => false,
+    _flushPendingBashMessages: () => {},
+    _expandSkillCommand: (t) => t,
+    _throwIfExtensionCommand: () => {},
+    _runAgentPrompt: fakeRunAgentPrompt,
+    agent: { state: { systemPrompt: "base" } },
+  };
+}
+
+const session = makeStubSession();
+
+// The watcher wake is started first here so the CAPTAIN's call is the one that
+// loses - the sub-case no tail inspection can detect, because the wake turn
+// answers normally and leaves a perfectly healthy transcript behind.
+log("watcher wake: prompt('FIRSTMATE WATCHER WAKE...') START");
+const wakeCall = promptFn.call(session, "FIRSTMATE WATCHER WAKE: stale", { streamingBehavior: "followUp", source: "extension" });
+
+// The wake above models fm-primary-pi-watch.ts's sendWake(): an unrelated
+// background watcher-close callback calling pi.sendUserMessage(...,
+// {deliverAs:"followUp"}) with zero coordination with the interactive call
+// below. sendUserMessage forwards to prompt() with streamingBehavior:
+// "followUp" and source: "extension" (agent-session.js sendUserMessage()).
+const CAPTAIN_TEXT = "bitte den Stand zusammenfassen";
+log(`captain call: prompt('${CAPTAIN_TEXT}') START`);
+const captainCall = promptFn.call(session, CAPTAIN_TEXT, { source: "interactive" });
+
+// allSettled, not all: the losing call rejects by design, exactly as the real
+// nested agent.prompt() does when it finds an active run.
+await Promise.allSettled([captainCall, wakeCall]);
+
+console.log(events.join("\n"));
+console.log(`\nmaxConcurrentRunAgentPromptCalls = ${maxConcurrentRunAgentPromptCalls}`);
+console.log(`extension event order: ${extensionEvents.join(" -> ")}`);
+console.log(`settlesWhileAnotherRunWasLive = ${settlesWhileAnotherRunWasLive}`);
+
+const capturedCaptainInput = capturedInputs.some((i) => i.source === "interactive" && i.text === CAPTAIN_TEXT);
+const captainTextInTranscript = transcript.some(
+  (m) => m.role === "user" && JSON.stringify(m.content ?? "").includes(CAPTAIN_TEXT),
+);
+const transcriptTail = transcript[transcript.length - 1];
+const tailLooksHealthy = transcriptTail?.role === "assistant" && transcriptTail?.stopReason === "stop";
+console.log(`\ncaptain input seen by the "input" event: ${capturedCaptainInput}`);
+console.log(`captain text present in the transcript: ${captainTextInTranscript}`);
+console.log(`transcript tail looks healthy: ${tailLooksHealthy}`);
+if (maxConcurrentRunAgentPromptCalls > 1) {
+  console.log("REPRODUCED: two concurrent prompt() calls both reached _runAgentPrompt() concurrently.");
+  if (settlesWhileAnotherRunWasLive > 0) {
+    // This is what the extension has to survive: it sees two
+    // before_agent_start events and then a settle that is NOT terminal. It
+    // counts logical runs in flight and leaves such a settle unevaluated, so
+    // only the trailing settle - the one that drains the counter - is judged.
+    // tests/fm-turnend-guard.test.sh's test_pi_reply_recovery_skips_a_spurious_mid_turn_settle
+    // and test_pi_reply_recovery_spurious_settle_never_doubles_a_healthy_answer
+    // replay exactly this event order against the real handler.
+    console.log("REPRODUCED: a spurious agent_settled fired while another logical run was still live.");
+  }
+  if (capturedCaptainInput && !captainTextInTranscript && tailLooksHealthy) {
+    // The BEFORE state for captain-input-loss recovery: the captain's message
+    // is gone from the transcript while the tail is a healthy assistant reply,
+    // so no tail inspection can detect it - but prompt()'s `input` event did
+    // see the message before the isStreaming check, which is what
+    // .pi/extensions/fm-primary-turnend-guard.ts records and resubmits.
+    // tests/fm-turnend-guard.test.sh's
+    // test_pi_input_recovery_resubmits_a_captain_message_lost_to_the_race
+    // drives the real handler through exactly this state (AFTER: exactly one
+    // resubmission carrying the lost text), and
+    // test_pi_input_recovery_stays_silent_for_a_delivered_captain_message is
+    // the negative control.
+    console.log("REPRODUCED: the captain's message was lost entirely while the transcript tail stayed healthy.");
+  }
+  process.exit(1);
+} else {
+  console.log("NOT REPRODUCED this run (race is timing-dependent).");
+  process.exit(0);
+}