diff --git a/.pi/extensions/fm-primary-pi-watch.ts b/.pi/extensions/fm-primary-pi-watch.ts
index 2b4cde1fa36..04b04f7b019 100644
--- a/.pi/extensions/fm-primary-pi-watch.ts
+++ b/.pi/extensions/fm-primary-pi-watch.ts
@@ -4,7 +4,7 @@ import { createHash } from "node:crypto";
 import { mkdirSync, readFileSync, writeFileSync } from "node:fs";
 import { dirname, resolve } from "node:path";
 import { fileURLToPath } from "node:url";
-import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
+import type { ExtensionAPI, SessionEntry } from "@earendil-works/pi-coding-agent";
 import { Type } from "typebox";
 
 type ArmResult = {
@@ -12,8 +12,163 @@ type ArmResult = {
   message: string;
 };
 
+type DirectExchangeEvent = {
+  version: 1;
+  event: "submitted" | "admitted" | "delivered" | "answered";
+  exchangeId: string;
+  at: number;
+  inputText?: string;
+  imageCount?: number;
+  inputContent?: unknown;
+  delivery?: "immediate" | "steer" | "followUp";
+  content?: unknown;
+  runKey?: string;
+};
+
+type DirectExchangeState = {
+  exchangeId: string;
+  submittedAt: number;
+  inputText: string;
+  imageCount: number;
+  inputContent?: unknown;
+  delivery: "immediate" | "steer" | "followUp";
+  admittedAt?: number;
+  deliveredAt?: number;
+  deliveredContent?: unknown;
+  deliveredRecordIndex?: number;
+  deliveredRunKey?: string;
+  answeredAt?: number;
+  answerContent?: unknown;
+};
+
 type LockOwnership = "owned" | "missing" | "other";
 
+const directExchangeEntryType = "firstmate-direct-exchange";
+const directInputObservationEntryType = "firstmate-direct-input-observation";
+const continuityMessageType = "firstmate-direct-exchange-continuity";
+
+function cloneJson(value: unknown): unknown {
+  try {
+    return JSON.parse(JSON.stringify(value));
+  } catch {
+    return String(value);
+  }
+}
+
+function textContent(content: unknown): string {
+  if (typeof content === "string") return content;
+  if (!Array.isArray(content)) return "";
+  return content
+    .filter((part): part is { type: "text"; text: string } => {
+      if (!part || typeof part !== "object") return false;
+      const candidate = part as { type?: unknown; text?: unknown };
+      return candidate.type === "text" && typeof candidate.text === "string";
+    })
+    .map((part) => part.text)
+    .join("\n");
+}
+
+function parseDirectExchangeEvent(entry: SessionEntry): DirectExchangeEvent | undefined {
+  if (entry.type !== "custom" || entry.customType !== directExchangeEntryType) return undefined;
+  const data = entry.data;
+  if (!data || typeof data !== "object") return undefined;
+  const candidate = data as Partial<DirectExchangeEvent>;
+  if (
+    candidate.version !== 1 ||
+    !["submitted", "admitted", "delivered", "answered"].includes(candidate.event ?? "") ||
+    typeof candidate.exchangeId !== "string" ||
+    !candidate.exchangeId ||
+    typeof candidate.at !== "number"
+  ) {
+    return undefined;
+  }
+  return candidate as DirectExchangeEvent;
+}
+
+function foldDirectExchanges(entries: SessionEntry[]): DirectExchangeState[] {
+  const ordered: DirectExchangeState[] = [];
+  const byId = new Map<string, DirectExchangeState>();
+  entries.forEach((entry, index) => {
+    const event = parseDirectExchangeEvent(entry);
+    if (!event) return;
+    if (event.event === "submitted") {
+      if (byId.has(event.exchangeId) || typeof event.inputText !== "string") return;
+      const state: DirectExchangeState = {
+        exchangeId: event.exchangeId,
+        submittedAt: event.at,
+        inputText: event.inputText,
+        imageCount: event.imageCount ?? 0,
+        delivery: event.delivery ?? "immediate",
+      };
+      if (event.inputContent !== undefined) state.inputContent = event.inputContent;
+      byId.set(event.exchangeId, state);
+      ordered.push(state);
+      return;
+    }
+    const state = byId.get(event.exchangeId);
+    if (!state) return;
+    if (event.event === "admitted") {
+      state.admittedAt = event.at;
+    } else if (event.event === "delivered") {
+      state.admittedAt ??= event.at;
+      state.deliveredAt = event.at;
+      state.deliveredContent = event.content;
+      state.deliveredRecordIndex = index;
+      state.deliveredRunKey = event.runKey;
+    } else if (event.event === "answered") {
+      state.answeredAt = event.at;
+      state.answerContent = event.content;
+    }
+  });
+  return ordered;
+}
+
+function contentPresent(messages: Array<{ role: string; content?: unknown }>, role: string, content: unknown): boolean {
+  const expected = JSON.stringify(content);
+  return messages.some((message) => message.role === role && JSON.stringify(message.content) === expected);
+}
+
+function renderDirectExchangeContinuity(
+  entries: SessionEntry[],
+  messages: Array<{ role: string; content?: unknown }>,
+): string | undefined {
+  const exchanges = foldDirectExchanges(entries);
+  const sections: string[] = [];
+  const latestAnswered = [...exchanges].reverse().find((exchange) => exchange.answerContent !== undefined);
+  if (
+    latestAnswered &&
+    (!contentPresent(messages, "user", latestAnswered.deliveredContent) ||
+      !contentPresent(messages, "assistant", latestAnswered.answerContent))
+  ) {
+    sections.push(
+      [
+        `Exchange ${latestAnswered.exchangeId}: ANSWERED`,
+        `Human input, exact JSON: ${JSON.stringify(latestAnswered.deliveredContent)}`,
+        `Assistant answer, exact JSON: ${JSON.stringify(latestAnswered.answerContent)}`,
+      ].join("\n"),
+    );
+  }
+  for (const exchange of exchanges) {
+    if (exchange.answerContent !== undefined) continue;
+    if (exchange.deliveredContent !== undefined) {
+      sections.push(
+        [
+          `Exchange ${exchange.exchangeId}: OPEN_REPLY_OBLIGATION`,
+          `Human input, exact JSON: ${JSON.stringify(exchange.deliveredContent)}`,
+          "No completed assistant answer was observed before compaction.",
+        ].join("\n"),
+      );
+    }
+  }
+  if (sections.length === 0) return undefined;
+  return [
+    "FIRSTMATE DIRECT EXCHANGE CONTINUITY",
+    "This is extension-generated context metadata, not human-authored input.",
+    "Watcher and turn-end supervision prompts are custom messages, not captain-authored requests.",
+    ...sections,
+  ].join("\n\n");
+}
+
 const extensionFile = fileURLToPath(import.meta.url);
 const extensionDir = dirname(extensionFile);
 const root = resolve(extensionDir, "../..");
@@ -90,6 +245,16 @@ function failureLine(stdout: string, stderr: string, code: number | null): strin
 }
 
 export default function (pi: ExtensionAPI) {
+  let exchangeSequence = 0;
+  let agentRunSequence = 0;
+  let activeAgentRunKey: string | undefined;
+  let replyCohort = new Set<string>();
+  let agentStartAwaitingHumanDelivery = false;
+
+  function appendDirectExchange(event: DirectExchangeEvent): void {
+    pi.appendEntry(directExchangeEntryType, event);
+  }
+
   function stopArm(): void {
     if (child) child.kill("SIGTERM");
     child = null;
@@ -100,10 +265,15 @@ export default function (pi: ExtensionAPI) {
   };
   process.once("exit", cleanupOnProcessExit);
 
-  async function sendWake(message: string) {
-    await pi.sendUserMessage(
-      `FIRSTMATE WATCHER WAKE: ${message}\n\nRun bin/fm-wake-drain.sh first, handle the queued wake, then resume Pi supervision.`,
-      { deliverAs: "followUp" },
+  function sendWake(message: string): void {
+    pi.sendMessage(
+      {
+        customType: "firstmate-watcher-wake",
+        content: `FIRSTMATE WATCHER WAKE: ${message}\n\nRun bin/fm-wake-drain.sh first, handle the queued wake, then resume Pi supervision.`,
+        display: true,
+        details: { version: 1, source: "firstmate-extension", kind: "watcher-wake" },
+      },
+      { deliverAs: "followUp", triggerTurn: true },
     );
   }
 
@@ -138,7 +308,7 @@ export default function (pi: ExtensionAPI) {
       const failure = reason ? "" : failureLine(stdout, stderr, code);
       if (!reason && !failure) return;
       try {
-        await sendWake(reason || failure);
+        sendWake(reason || failure);
       } catch {
         // Pi owns delivery errors; fail open so the extension never wedges the session.
       }
@@ -146,7 +316,7 @@ export default function (pi: ExtensionAPI) {
     child.on("error", async (error: Error) => {
       child = null;
       try {
-        await sendWake(`watcher: FAILED - Pi extension arm child ${id} failed: ${error.message}`);
+        sendWake(`watcher: FAILED - Pi extension arm child ${id} failed: ${error.message}`);
       } catch {
         // Fail open.
       }
@@ -157,6 +327,126 @@ export default function (pi: ExtensionAPI) {
   pi.on?.("session_start", () => {
     markLoaded();
   });
+
+  pi.on("agent_start", () => {
+    activeAgentRunKey = `${process.pid}:${++agentRunSequence}`;
+    agentStartAwaitingHumanDelivery = true;
+  });
+
+  function newExchangeId(at: number, content: unknown): string {
+    return createHash("sha256")
+      .update(`${at}\0${++exchangeSequence}\0${JSON.stringify(content)}`)
+      .digest("hex")
+      .slice(0, 16);
+  }
+
+  function admitExchange(exchangeId: string, at: number): void {
+    appendDirectExchange({ version: 1, event: "admitted", exchangeId, at });
+  }
+
+  function createAdmittedExchange(
+    content: unknown,
+    delivery: "immediate" | "steer" | "followUp",
+    at: number,
+  ): string {
+    const exchangeId = newExchangeId(at, content);
+    appendDirectExchange({
+      version: 1,
+      event: "submitted",
+      exchangeId,
+      at,
+      inputText: textContent(content),
+      imageCount: Array.isArray(content)
+        ? content.filter((part) => part && typeof part === "object" && (part as { type?: unknown }).type === "image").length
+        : 0,
+      inputContent: content,
+      delivery,
+    });
+    admitExchange(exchangeId, at);
+    return exchangeId;
+  }
+
+  pi.on("input", (event) => {
+    if (event.source === "extension") return;
+    const at = Date.now();
+    const inputContent = cloneJson([{ type: "text", text: event.text }, ...(event.images ?? [])]);
+    const delivery = event.streamingBehavior ?? "immediate";
+    pi.appendEntry(directInputObservationEntryType, {
+      version: 1,
+      observationId: newExchangeId(at, inputContent),
+      at,
+      inputText: event.text,
+      imageCount: event.images?.length ?? 0,
+      inputContent,
+      delivery,
+    });
+  });
+
+  pi.on("message_end", (event, ctx) => {
+    if (event.message.role === "user") {
+      const content = cloneJson(event.message.content);
+      const exchangeId = createAdmittedExchange(content, "steer", event.message.timestamp);
+      appendDirectExchange({
+        version: 1,
+        event: "delivered",
+        exchangeId,
+        at: event.message.timestamp,
+        content,
+        runKey: activeAgentRunKey,
+      });
+      if (agentStartAwaitingHumanDelivery) {
+        replyCohort = new Set<string>();
+        agentStartAwaitingHumanDelivery = false;
+      }
+      replyCohort.add(exchangeId);
+      return;
+    }
+    if (event.message.role !== "assistant" || event.message.stopReason !== "stop") return;
+    const answerText = textContent(event.message.content);
+    if (!answerText) return;
+    const branch = ctx.sessionManager.getBranch();
+    const open = foldDirectExchanges(branch).filter((exchange) => {
+      if (exchange.deliveredContent === undefined || exchange.answerContent !== undefined) return false;
+      if (!replyCohort.has(exchange.exchangeId)) return false;
+      const deliveredIndex = exchange.deliveredRecordIndex;
+      if (deliveredIndex === undefined) return false;
+      return !branch.slice(deliveredIndex + 1).some((entry) => entry.type === "custom_message");
+    });
+    for (const exchange of open) {
+      appendDirectExchange({
+        version: 1,
+        event: "answered",
+        exchangeId: exchange.exchangeId,
+        at: event.message.timestamp,
+        content: cloneJson(event.message.content),
+      });
+    }
+    if (open.length > 0) replyCohort.clear();
+  });
+
+  pi.on("context", (event, ctx) => {
+    const continuity = renderDirectExchangeContinuity(
+      ctx.sessionManager.getBranch(),
+      event.messages,
+    );
+    if (!continuity) return;
+    const message = {
+      role: "custom" as const,
+      customType: continuityMessageType,
+      content: continuity,
+      display: false,
+      details: { version: 1, source: "firstmate-extension", kind: "direct-exchange-continuity" },
+      timestamp: Date.now(),
+    };
+    let insertAt = 0;
+    for (let i = event.messages.length - 1; i >= 0; i -= 1) {
+      if (event.messages[i]?.role === "user") {
+        insertAt = i;
+        break;
+      }
+    }
+    return { messages: [...event.messages.slice(0, insertAt), message, ...event.messages.slice(insertAt)] };
+  });
   pi.on?.("session_shutdown", () => {
     stopArm();
     process.off("exit", cleanupOnProcessExit);
diff --git a/.pi/extensions/fm-primary-turnend-guard.ts b/.pi/extensions/fm-primary-turnend-guard.ts
index f6b68497f35..a8631c99876 100644
--- a/.pi/extensions/fm-primary-turnend-guard.ts
+++ b/.pi/extensions/fm-primary-turnend-guard.ts
@@ -132,11 +132,17 @@ export default function (pi: ExtensionAPI) {
 
     guardFollowupActive = true;
     try {
-      await pi.sendUserMessage(
-        "TURN WOULD END BLIND - supervision is off. " +
-          "Resume supervision according to the session-start operating block before ending the turn.\n\n" +
-          result.stderr,
-        { deliverAs: "followUp" },
+      pi.sendMessage(
+        {
+          customType: "firstmate-turnend-guard",
+          content:
+            "TURN WOULD END BLIND - supervision is off. " +
+            "Resume supervision according to the session-start operating block before ending the turn.\n\n" +
+            result.stderr,
+          display: true,
+          details: { version: 1, source: "firstmate-extension", kind: "turnend-guard" },
+        },
+        { deliverAs: "followUp", triggerTurn: true },
       );
     } catch {
       guardFollowupActive = false;
diff --git a/docs/architecture.md b/docs/architecture.md
index 1048e75f3b3..992a83e374b 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -58,6 +58,7 @@ This Firstmate tooling change requires no web-app, API, or realtime deployment.
 
 At session start, `bin/fm-session-start.sh` emits exactly one primary-harness supervision block rendered by `bin/fm-supervision-instructions.sh` from `docs/supervision-protocols/`.
 That block owns the live wait shape for the running primary harness: Claude and Grok use background-notify cycles, Codex uses bounded foreground checkpoints, Pi uses its two tracked primary extensions, and OpenCode uses its TUI plugin.
+Pi's supervision prompts remain custom-message context while exact direct exchanges and open reply obligations survive compaction through the contract in [`supervision-protocols/pi.md`](supervision-protocols/pi.md#input-provenance-and-compaction-continuity).
 `bin/fm-watch-arm.sh` remains the verified arm wrapper for protocols that call it; it forks the watcher as a tracked child, verifies it is genuinely alive with a fresh liveness beacon, and prints an honest initial status (`started` / `attached` / restart-only `healthy` / `FAILED`, the last exiting non-zero).
 On `attached` it stays live until that existing cycle ends so background-notify harnesses do not get an empty false wake from a healthy no-op exit.
 Because the initial status stays in the task's buffer long after the instant it describes, an attach is always closed by a terminal `FAILED` line - either `attached cycle ended` when the holder stops passing the liveness proof, or `attach interrupted` when the arm itself is signalled away - with the beacon age re-measured at exit and a non-zero status.
diff --git a/docs/supervision-protocols/pi.md b/docs/supervision-protocols/pi.md
index 19a188b6074..7d95c973494 100644
--- a/docs/supervision-protocols/pi.md
+++ b/docs/supervision-protocols/pi.md
@@ -1,4 +1,4 @@
-Mode: Pi extension background wake.
+Mode: Pi extension background wake with direct-exchange compaction continuity.
 
 When this session owns supervision and away mode is not active:
 1. Drain first with `bin/fm-wake-drain.sh`.
@@ -6,7 +6,7 @@ When this session owns supervision and away mode is not active:
 3. Arm supervision with the `fm_watch_arm_pi` tool.
    Use `/fm-watch-arm-pi` only as a human-entered fallback.
    Never run `bin/fm-watch-arm.sh` through Pi's bash tool because that foreground arm can wedge the agent and bypasses extension-owned cleanup.
-4. The extension starts `bin/fm-watch-arm.sh --restart`, keeps the child attached to the live Pi process, and sends a follow-up user message when the child exits with an actionable watcher reason.
+4. The extension starts `bin/fm-watch-arm.sh --restart`, keeps the child attached to the live Pi process, and sends a context-participating `firstmate-watcher-wake` custom follow-up when the child exits with an actionable watcher reason.
 5. If the extension says the watcher is already healthy, do not start another cycle.
 6. If the extension reports a watcher failure, drain queued wakes, inspect the failure text, and restart Pi with both extensions loaded if needed.
 7. Never use shell `&` for watcher supervision.
@@ -16,6 +16,60 @@ 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.
 
+## Input provenance and compaction continuity
+
+This document owns the Pi-specific boundary between direct captain input, automated supervision prompts, pending input, and rebuilt post-compaction model context.
+The watcher and turn-end extensions use Pi's context-participating `sendMessage()` custom-message path with distinct `firstmate-watcher-wake` and `firstmate-turnend-guard` types, `deliverAs: "followUp"`, and `triggerTurn: true`.
+They never use `sendUserMessage()` for automation, so supervision remains visible to the model without becoming a human-authored `role: "user"` turn.
+
+The watcher extension observes Pi's `input` event and records `interactive` or `rpc` submissions as provisional non-context `firstmate-direct-input-observation` entries before Pi queues or delivers them.
+Each provisional record carries the exact text-and-image content from that boundary, including image data and MIME type, so compaction before delivery cannot reduce an attachment to a count.
+Only exact user content delivered by `message_end` becomes a fresh admitted and delivered exchange.
+The delivery boundary never associates a user message with an observation or an older exchange by content, order, or fallback, so a consumed observation followed by an identical accepted retry remains unambiguous.
+Global queue state and later lifecycle events never promote or match a provisional observation because they cannot identify which input Pi accepted.
+Unadmitted provisional records never become reply obligations.
+Each exact user delivery joins an in-process logical reply cohort awaiting the next completed answer.
+A retry `agent_start` retains that cohort until Pi delivers another human message, so a successful retry can close the original exchange without another user `message_end`.
+The first human delivery after an `agent_start` replaces a retained cohort, and additional human deliveries before the answer join it, so an answer to a newer request cannot close an older failed request.
+It records an exact completed assistant answer only for the active cohort, on `stopReason: "stop"`, when no custom message intervened after delivery, then clears the answered cohort.
+These append-only records survive compaction and session resume without changing Pi's queue ordering.
+A queued submission remains owned by Pi until exact delivery, so continuity metadata never bypasses the steering or follow-up queue.
+Queue disappearance without an exact user delivery remains durable observation evidence but never creates an exchange or reply obligation.
+
+Before each model request, the extension compares Pi's compaction-aware message list with the full active branch.
+When compaction omitted the latest completed direct exchange, the hook injects one hidden `firstmate-direct-exchange-continuity` custom message containing the exact JSON user content, the exact JSON assistant answer, and `ANSWERED`.
+Every delivered input without a completed answer is included distinctly as `OPEN_REPLY_OBLIGATION`.
+The continuity message identifies itself as extension-generated metadata and states that watcher and guard prompts are custom messages rather than captain-authored requests.
+It is inserted before the current human user message, preserving that user message as the final prompt and avoiding tool-call or tool-result adjacency changes.
+The mechanism does not cancel compaction, enlarge `keepRecentTokens`, alter the cut point, or refuse a turn; malformed or absent continuity state simply leaves Pi's ordinary context unchanged.
+
+### 2026-08-25 incident evidence
+
+The source was the live Pi 0.84.2 JSONL session `2026-08-24T12-45-47-533Z_01a033ce-368d-7c69-9c34-acd78d33130d.jsonl` under the primary's Pi session directory.
+The captain's input carried message timestamp `2026-08-25T03:28:00.086Z` and was persisted as entry `e5eafcbb` at `03:28:06.116Z`, a 6.030-second steering delay while the current tool turn finished.
+The exact assistant answer was persisted as `f2a84da7` at `03:28:48.884Z`.
+An extension watcher input had been generated earlier at `03:27:33.185Z` but was persisted as `85a87db6` at `03:28:48.886Z`, a 75.701-second follow-up delay and two milliseconds after the direct answer.
+This ordering proves that the submitted human steering input was delayed but neither omitted nor starved: both its exact user entry and exact assistant answer exist before the older extension follow-up was delivered.
+The JSONL contains no additional submitted human entry between that answer and the watcher follow-up; editor text that was never submitted is outside session evidence and is not claimed either way.
+Compaction `d4b4f008` followed at `03:29:11.300Z` with `firstKeptEntryId=85a87db6`, so the exact direct exchange remained only inside a lossy summary while the automated follow-up and later supervision turns occupied the live tail.
+
+### 2026-08-25 regression evidence
+
+Deterministic command: `tests/fm-pi-watch-extension.test.sh`.
+Observed output included `ok - Pi compaction continuity preserves exact human exchange across automated custom prompts`.
+The regression uses Pi 0.84.2's installed `Agent` with a controlled assistant event stream to prove a later human image steer receives its own provider turn before an older automation follow-up, then uses installed `SessionManager` to build the actual compaction-aware context after an exact human question and answer, a custom watcher prompt, an assistant supervision response, and a compaction whose first-kept entry is the watcher custom message.
+It then delivers a referring human follow-up and proves the chained context hook adds the exact prior question and answer as `ANSWERED`, adds the follow-up as `OPEN_REPLY_OBLIGATION`, keeps both records custom rather than user-authored, and leaves still-pending steering submissions under Pi's ownership.
+A compaction-before-delivery fixture records text plus an image provisionally, cuts at a later custom watcher entry before any user message exists, proves the durable session contract retains exact image data and MIME type without injecting it into context, then delivers the exact input and proves a later compaction restores it as `OPEN_REPLY_OBLIGATION`.
+It also proves a consumed steer does not become continuity metadata while unrelated automation is pending or after that automation drains.
+A separate compaction fixture cuts an unanswered direct question behind a custom watcher turn and proves the exact question returns as `OPEN_REPLY_OBLIGATION` rather than being inferred from summary prose.
+`tests/fm-pi-retry-continuity.test.sh` exercises the installed-Pi proof that an error followed by a retry without another human delivery closes the retained exchange, while a new human delivery on the retry replaces the retained cohort so its answer cannot close the older failed exchange.
+
+Live command: `FM_PI_COMPACTION_LIVE_E2E=1 FM_PI_LIVE_AUTH_DIR='/Users/dongkeun/.pi/firstmate-local' tests/fm-pi-primary-compaction-live-e2e.test.sh`.
+Observed output: `ok - Pi 0.84.2 live compaction rebuilt the exact answered captain exchange across a custom watcher turn (firstKeptType=message)`.
+The smoke used a cloned project, private tmux socket, isolated `PI_CODING_AGENT_DIR`, isolated `FM_HOME`, copied read-only credential input, synthetic isolated watcher arm, real Pi `/compact`, and a later model turn.
+The isolated session proved the compaction cut after the direct exchange, captured the provider-bound rebuilt context with the exact question, exact answer, `ANSWERED`, and custom-message provenance, and observed the resumed model identify the prior direct question and report that it had been answered.
+It did not touch the live primary session, home, session file, lock, wake queue, or watcher.
+
 Verification on 2026-07-09 used Pi 0.80.5, an isolated `PI_CODING_AGENT_DIR`, an isolated `FM_HOME`, and the dedicated tmux socket `fm-pi-q6-lab`.
 The command `Use the fm_watch_arm_pi custom tool now. Do not use bash.` rendered `watcher: started Pi extension arm child 1`, then the model returned `DONE` without the prior `result.content.filter(...)` crash.
 The extension tool returned Pi's required text `content` plus structured `details` and used `Type.Object({})` for its parameter schema.
diff --git a/docs/turnend-guard.md b/docs/turnend-guard.md
index 3444171615d..28098c9af82 100644
--- a/docs/turnend-guard.md
+++ b/docs/turnend-guard.md
@@ -42,7 +42,7 @@ All verified primary harnesses have a tracked integration:
 - `claude`: `.claude/settings.json` registers a `Stop` hook command anchored through `"$CLAUDE_PROJECT_DIR"/bin/fm-turnend-guard.sh`.
 - `codex`: `.codex/hooks.json` registers a `Stop` hook that reads the hook payload once, anchors the executable to the hook command process working directory, verifies that root is firstmate-shaped and hook-bearing, and pipes the original payload to that checkout's `bin/fm-turnend-guard.sh`.
 - `opencode`: `.opencode/plugins/fm-primary-turnend-guard.js` listens for `session.idle`, lets the watcher-arm coordinator handle normal idle supervision first, runs the shared guard only when that coordinator does not act, and uses `client.session.promptAsync` to force one follow-up prompt when the guard returns 2.
-- `pi`: `.pi/extensions/fm-primary-turnend-guard.ts` listens for `agent_settled`, marks the extension version loaded for session-start checks, runs the shared guard once per logical agent run, and uses `pi.sendUserMessage(..., { deliverAs: "followUp" })` to force one follow-up prompt when the guard returns 2.
+- `pi`: `.pi/extensions/fm-primary-turnend-guard.ts` listens for `agent_settled`, marks the extension version loaded for session-start checks, runs the shared guard once per logical agent run, and uses a triggering `firstmate-turnend-guard` custom follow-up when the guard returns 2; [`supervision-protocols/pi.md`](supervision-protocols/pi.md#input-provenance-and-compaction-continuity) owns its input-provenance and compaction behavior.
 - `grok`: `.grok/hooks/fm-primary-turnend-guard.json` registers a `Stop` hook that invokes `bin/fm-turnend-guard-grok.sh`.
   The adapter runs the shared guard and, when it returns 2, invokes `grok --resume <sessionId> -p <guard-reason>` with `GROK_TURNEND_GUARD_ACTIVE=1`.
   It does not pass `--permission-mode`, so the passive Stop hook cannot grant stronger tool permissions than Grok's resumed-session default.
@@ -149,3 +149,4 @@ No Herdr command was issued and no fleet state was touched; the experiment wrote
 `tests/fm-turnend-guard.test.sh` covers the shared predicate, primary scoping (including a secondmate's own home being guarded like the main primary while its child worktrees stay exempt), `FM_HOME` and `FM_STATE_OVERRIDE` precedence, Pi logical-run latch behavior for no-tool and multi-tool runs, fail-open behavior without `jq`, tracked hook registration for all five harnesses, and the Grok adapter's forced-resume loop guard and permission-mode regression.
 The default behavior suite does not invoke live language-model harnesses.
 `FM_PI_LIVE_E2E=1 tests/fm-pi-primary-live-e2e.test.sh` opts into the isolated interactive Pi regression recorded above.
+`FM_PI_COMPACTION_LIVE_E2E=1 FM_PI_LIVE_AUTH_DIR=<isolated-smoke-auth-source> tests/fm-pi-primary-compaction-live-e2e.test.sh` opts into the isolated real-compaction regression owned and evidenced by [`supervision-protocols/pi.md`](supervision-protocols/pi.md#2026-08-25-regression-evidence).
diff --git a/tests/behavior-test-durations.tsv b/tests/behavior-test-durations.tsv
index b83432a3e21..fabadb0c6b7 100644
--- a/tests/behavior-test-durations.tsv
+++ b/tests/behavior-test-durations.tsv
@@ -79,9 +79,14 @@
 2770	tests/fm-no-mistakes-runtime.test.sh
 7260	tests/fm-no-mistakes-worker.test.sh
 2000	tests/fm-pi-account-home.test.sh
+4500	tests/fm-pi-direct-continuity-types.test.sh
+4250	tests/fm-pi-direct-continuity.test.sh
+5300	tests/fm-pi-exact-delivery-association.test.sh
+1500	tests/fm-pi-primary-compaction-live-e2e.test.sh
 10	tests/fm-pi-primary-live-e2e.test.sh
 80	tests/fm-pi-primary-types.test.sh
 4000	tests/fm-pi-refresh.test.sh
+3750	tests/fm-pi-retry-continuity.test.sh
 3170	tests/fm-pi-watch-extension.test.sh
 1574	tests/fm-pr-merge.test.sh
 802	tests/fm-prompt-exec.test.sh
diff --git a/tests/fm-pi-direct-continuity-types.test.sh b/tests/fm-pi-direct-continuity-types.test.sh
new file mode 100755
index 00000000000..2a9d7db1163
--- /dev/null
+++ b/tests/fm-pi-direct-continuity-types.test.sh
@@ -0,0 +1,55 @@
+#!/usr/bin/env bash
+# shellcheck source=tests/test-entry.sh
+. "$(dirname "$0")/test-entry.sh"
+set -u
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+pi_package=${FM_PI_PACKAGE_DIR:-}
+tsc_bin=
+selected_node_dir=
+
+while IFS= read -r node_bin; do
+  node_dir=$(dirname "$node_bin")
+  [ -n "$selected_node_dir" ] || selected_node_dir=$node_dir
+  if [ -z "$pi_package" ] && [ -x "$node_dir/npm" ]; then
+    candidate=$(PATH="$node_dir:$PATH" "$node_dir/npm" root -g 2>/dev/null)/@earendil-works/pi-coding-agent
+    if [ -f "$candidate/package.json" ]; then
+      pi_package=$candidate
+    fi
+  fi
+done < <(
+  {
+    type -a -p node 2>/dev/null
+    for candidate_node in /opt/homebrew/opt/node/bin/node /usr/local/opt/node/bin/node; do
+      [ -x "$candidate_node" ] && printf '%s\n' "$candidate_node"
+    done
+  } | awk '!seen[$0]++'
+)
+
+for candidate in \
+  "$(command -v tsc 2>/dev/null || true)" \
+  "$HOME"/.nvm/versions/node/*/lib/node_modules/typescript/bin/tsc \
+  "$HOME"/.nvm/versions/node/*/lib/node_modules/*/node_modules/.bin/tsc \
+  /opt/homebrew/lib/node_modules/typescript/bin/tsc \
+  /opt/homebrew/lib/node_modules/*/node_modules/.bin/tsc \
+  /usr/local/lib/node_modules/typescript/bin/tsc \
+  /usr/local/lib/node_modules/*/node_modules/.bin/tsc; do
+  [ -x "$candidate" ] || continue
+  tsc_bin=$candidate
+  break
+done
+
+[ -n "$pi_package" ] && [ -f "$pi_package/package.json" ] || { echo "not ok - installed Pi package not found" >&2; exit 1; }
+[ -n "$tsc_bin" ] || { echo "not ok - installed TypeScript compiler not found" >&2; exit 1; }
+[ -n "$selected_node_dir" ] || { echo "not ok - Node runtime not found" >&2; exit 1; }
+
+out=$(env -u FM_TEST_RUNNER_ACTIVE PATH="$(dirname "$tsc_bin"):$selected_node_dir:$PATH" FM_PI_PACKAGE_DIR="$pi_package" "$ROOT/tests/fm-pi-primary-types.test.sh") || {
+  printf '%s\n' "$out" >&2
+  exit 1
+}
+printf '%s\n' "$out" | grep -Eq '^ok - Pi primary extensions pass strict no-emit typecheck against Pi ' || {
+  printf '%s\n' "$out" >&2
+  echo "not ok - strict Pi consumer typecheck did not run" >&2
+  exit 1
+}
+printf '%s\n' 'ok - exact Pi continuity consumer typecheck passed'
diff --git a/tests/fm-pi-direct-continuity.test.sh b/tests/fm-pi-direct-continuity.test.sh
new file mode 100755
index 00000000000..25fd724c636
--- /dev/null
+++ b/tests/fm-pi-direct-continuity.test.sh
@@ -0,0 +1,43 @@
+#!/usr/bin/env bash
+# shellcheck source=tests/test-entry.sh
+. "$(dirname "$0")/test-entry.sh"
+set -u
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+selected_node=
+pi_package=${FM_PI_PACKAGE_DIR:-}
+
+while IFS= read -r node_bin; do
+  [ -n "$node_bin" ] || continue
+  node_dir=$(dirname "$node_bin")
+  if [ -z "$selected_node" ] && [ "$("$node_bin" -p 'process.features?.typescript ? "yes" : "no"' 2>/dev/null)" = yes ]; then
+    selected_node=$node_dir
+  fi
+  if [ -z "$pi_package" ] && [ -x "$node_dir/npm" ]; then
+    candidate=$(PATH="$node_dir:$PATH" "$node_dir/npm" root -g 2>/dev/null)/@earendil-works/pi-coding-agent
+    if [ -f "$candidate/package.json" ]; then
+      pi_package=$candidate
+    fi
+  fi
+done < <(
+  {
+    type -a -p node 2>/dev/null
+    for candidate_node in /opt/homebrew/opt/node/bin/node /usr/local/opt/node/bin/node; do
+      [ -x "$candidate_node" ] && printf '%s\n' "$candidate_node"
+    done
+  } | awk '!seen[$0]++'
+)
+
+[ -n "$selected_node" ] || { echo "not ok - no available Node runtime supports TypeScript import" >&2; exit 1; }
+[ -n "$pi_package" ] && [ -f "$pi_package/package.json" ] || { echo "not ok - installed Pi package not found" >&2; exit 1; }
+
+out=$(env -u FM_TEST_RUNNER_ACTIVE PATH="$selected_node:$PATH" FM_PI_PACKAGE_DIR="$pi_package" FM_PI_EXACT_DELIVERY_PROOF=1 "$ROOT/tests/fm-pi-watch-extension.test.sh") || {
+  printf '%s\n' "$out" >&2
+  exit 1
+}
+printf '%s\n' "$out" | grep -Fqx 'PI_EXACT_DELIVERY_ASSOCIATION_PROOF_OK' || {
+  printf '%s\n' "$out" >&2
+  echo "not ok - exact Pi delivery association proof token missing" >&2
+  exit 1
+}
+printf '%s\n' 'PI_EXACT_DELIVERY_ASSOCIATION_PROOF_OK'
diff --git a/tests/fm-pi-exact-delivery-association.test.sh b/tests/fm-pi-exact-delivery-association.test.sh
new file mode 100755
index 00000000000..5106f9f7e17
--- /dev/null
+++ b/tests/fm-pi-exact-delivery-association.test.sh
@@ -0,0 +1,17 @@
+#!/usr/bin/env bash
+# shellcheck source=tests/test-entry.sh
+. "$(dirname "$0")/test-entry.sh"
+set -u
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+
+out=$(env -u FM_TEST_RUNNER_ACTIVE "$ROOT/tests/fm-pi-direct-continuity.test.sh") || {
+  printf '%s\n' "$out" >&2
+  exit 1
+}
+printf '%s\n' "$out" | grep -Fqx 'PI_EXACT_DELIVERY_ASSOCIATION_PROOF_OK' || {
+  printf '%s\n' "$out" >&2
+  echo "not ok - installed Pi exact-delivery assertions did not complete" >&2
+  exit 1
+}
+printf '%s\n' 'ok - installed Pi exact delivery association preserved'
diff --git a/tests/fm-pi-primary-compaction-live-e2e.test.sh b/tests/fm-pi-primary-compaction-live-e2e.test.sh
new file mode 100755
index 00000000000..1d7e876e77b
--- /dev/null
+++ b/tests/fm-pi-primary-compaction-live-e2e.test.sh
@@ -0,0 +1,252 @@
+#!/usr/bin/env bash
+# shellcheck source=tests/test-entry.sh
+. "$(dirname "$0")/test-entry.sh"
+# Opt-in live Pi compaction regression on a private tmux socket and isolated homes.
+set -u
+
+if [ "${FM_PI_COMPACTION_LIVE_E2E:-0}" != 1 ]; then
+  echo "skip: set FM_PI_COMPACTION_LIVE_E2E=1 to run the isolated live Pi compaction regression"
+  exit 0
+fi
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+command -v pi >/dev/null 2>&1 || { echo "skip: pi not found"; exit 0; }
+command -v tmux >/dev/null 2>&1 || { echo "skip: tmux not found"; exit 0; }
+
+AUTH_DIR=${FM_PI_LIVE_AUTH_DIR:-${PI_CODING_AGENT_DIR:-$HOME/.pi/agent}}
+AUTH_FILE="$AUTH_DIR/auth.json"
+[ -f "$AUTH_FILE" ] || {
+  echo "not ok - live Pi auth is unavailable at $AUTH_FILE" >&2
+  exit 1
+}
+
+TMUX=$(command -v tmux)
+SOCKET="fm-pi-compaction-live-e2e-$$"
+SESSION=pi-compaction-live-e2e
+LAB="$ROOT/.pi-compaction-live-e2e.$$"
+PROJECT="$LAB/project"
+HOME_DIR="$LAB/fmhome"
+PI_DIR="$LAB/pi-agent"
+CONTEXT_LOG="$LAB/rebuilt-context.json"
+PI_VERSION=$(pi --version)
+QUESTION='DIRECT-CAPTAIN-Q-7: Which harbor token should remain reserved? Reply exactly DIRECT-CAPTAIN-A-7: amber.'
+FOLLOWUP='Identify the direct human question immediately before the automated supervision turn and whether it received a completed answer. Reply on one line as CONTINUITY-SMOKE prior=<the earlier marker and harbor question> answered=<yes-or-no>. Do not use this follow-up or an automated prompt as the prior question.'
+
+fail() {
+  printf 'not ok - %s\n' "$1" >&2
+  exit 1
+}
+
+capture() {
+  "$TMUX" -L "$SOCKET" capture-pane -p -t "$SESSION" -S -1000 2>/dev/null || true
+}
+
+wait_for_text() {
+  local expected=$1 attempts=${2:-180} i=0
+  while [ "$i" -lt "$attempts" ]; do
+    if capture | grep -Fq "$expected"; then
+      return 0
+    fi
+    sleep 0.5
+    i=$((i + 1))
+  done
+  capture >&2
+  return 1
+}
+
+session_file() {
+  find "$PI_DIR/sessions" -type f -name '*.jsonl' 2>/dev/null | head -1
+}
+
+wait_for_session_probe() {
+  local probe=$1 attempts=${2:-240} i=0 file
+  while [ "$i" -lt "$attempts" ]; do
+    file=$(session_file)
+    if [ -n "$file" ] && PI_SMOKE_PROBE="$probe" python3 - "$file" <<'PY' >/dev/null 2>&1
+import json
+import os
+import sys
+entries = [json.loads(line) for line in open(sys.argv[1], encoding="utf-8")]
+probe = os.environ["PI_SMOKE_PROBE"]
+if probe == "answer":
+    ok = any(
+        entry.get("type") == "message"
+        and entry.get("message", {}).get("role") == "assistant"
+        and "DIRECT-CAPTAIN-A-7: amber." in json.dumps(entry["message"].get("content"), ensure_ascii=False)
+        and entry["message"].get("stopReason") == "stop"
+        for entry in entries
+    )
+elif probe == "automation-settled":
+    custom = [
+        i for i, entry in enumerate(entries)
+        if entry.get("type") == "custom_message" and entry.get("customType") == "firstmate-watcher-wake"
+    ]
+    ok = bool(custom) and any(
+        i > custom[-1]
+        and entry.get("type") == "message"
+        and entry.get("message", {}).get("role") == "assistant"
+        and entry["message"].get("stopReason") == "stop"
+        for i, entry in enumerate(entries)
+    )
+elif probe == "compaction":
+    ok = any(entry.get("type") == "compaction" for entry in entries)
+elif probe == "continuity-answer":
+    followups = [
+        i for i, entry in enumerate(entries)
+        if entry.get("type") == "message"
+        and entry.get("message", {}).get("role") == "user"
+        and "Identify the direct human question immediately before" in json.dumps(entry["message"].get("content"), ensure_ascii=False)
+    ]
+    ok = bool(followups) and any(
+        i > followups[-1]
+        and entry.get("type") == "message"
+        and entry.get("message", {}).get("role") == "assistant"
+        and entry["message"].get("stopReason") == "stop"
+        and "CONTINUITY-SMOKE" in json.dumps(entry["message"].get("content"), ensure_ascii=False)
+        and "DIRECT-CAPTAIN-Q-7" in json.dumps(entry["message"].get("content"), ensure_ascii=False)
+        and "answered=yes" in json.dumps(entry["message"].get("content"), ensure_ascii=False)
+        for i, entry in enumerate(entries)
+    )
+else:
+    ok = False
+raise SystemExit(0 if ok else 1)
+PY
+    then
+      return 0
+    fi
+    sleep 0.5
+    i=$((i + 1))
+  done
+  capture >&2
+  return 1
+}
+
+send_prompt() {
+  local prompt=$1
+  "$TMUX" -L "$SOCKET" send-keys -t "$SESSION" -l "$prompt"
+  "$TMUX" -L "$SOCKET" send-keys -t "$SESSION" Enter
+}
+
+lab_pid_is_safe() {
+  local pid=$1 command
+  command=$(ps -p "$pid" -o command= 2>/dev/null || true)
+  case "$command" in
+    *"$LAB"*) return 0 ;;
+    *) return 1 ;;
+  esac
+}
+
+cleanup() {
+  local pid
+  "$TMUX" -L "$SOCKET" kill-server 2>/dev/null || true
+  while IFS= read -r pid; do
+    if lab_pid_is_safe "$pid"; then
+      kill -TERM "$pid" 2>/dev/null || true
+    fi
+  done < <(find "$HOME_DIR/state" -maxdepth 3 -type f -name pid -exec sed -n '1p' {} \; 2>/dev/null)
+  rm -rf "$LAB"
+}
+trap cleanup EXIT
+
+mkdir -p "$LAB"
+git clone -q "$ROOT" "$PROJECT"
+cp "$ROOT/.pi/extensions/fm-primary-pi-watch.ts" "$PROJECT/.pi/extensions/fm-primary-pi-watch.ts"
+cp "$ROOT/.pi/extensions/fm-primary-turnend-guard.ts" "$PROJECT/.pi/extensions/fm-primary-turnend-guard.ts"
+mkdir -p "$HOME_DIR/state" "$HOME_DIR/config" "$PI_DIR" "$PROJECT/.pi/extensions"
+cp "$AUTH_FILE" "$PI_DIR/auth.json"
+chmod 600 "$PI_DIR/auth.json"
+cat > "$PI_DIR/settings.json" <<'JSON'
+{
+  "compaction": {
+    "enabled": true,
+    "reserveTokens": 16384,
+    "keepRecentTokens": 1
+  }
+}
+JSON
+cat > "$PROJECT/.pi/extensions/zz-compaction-context-capture.ts" <<'TS'
+import { writeFileSync } from "node:fs";
+import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
+
+export default function (pi: ExtensionAPI) {
+  pi.on("context", (event) => {
+    if (!event.messages.some((message) => message.role === "custom" && message.customType === "firstmate-direct-exchange-continuity")) return;
+    writeFileSync(process.env.FM_PI_CONTEXT_LOG!, JSON.stringify(event.messages, null, 2));
+  });
+}
+TS
+cat > "$PROJECT/bin/fm-watch-arm.sh" <<'SH'
+#!/usr/bin/env bash
+set -u
+count_file="$FM_HOME/state/.compaction-smoke-arm-count"
+count=0
+[ ! -f "$count_file" ] || count=$(cat "$count_file")
+count=$((count + 1))
+printf '%s\n' "$count" > "$count_file"
+if [ "$count" -eq 1 ]; then
+  printf 'signal: isolated compaction smoke supervision turn\n'
+  exit 0
+fi
+trap 'exit 0' TERM INT
+while :; do sleep 1; done
+SH
+chmod +x "$PROJECT/bin/fm-watch-arm.sh"
+
+"$TMUX" -L "$SOCKET" new-session -d -s "$SESSION" -c "$PROJECT" \
+  "env PI_CODING_AGENT_DIR='$PI_DIR' FM_HOME='$HOME_DIR' FM_ROOT_OVERRIDE='$PROJECT' FM_STATE_OVERRIDE='$HOME_DIR/state' FM_CONFIG_OVERRIDE='$HOME_DIR/config' FM_PI_CONTEXT_LOG='$CONTEXT_LOG' bash -lc 'printf \"%s\\n\" \"\$\$\" > \"\$FM_HOME/state/.lock\"; pi --approve --provider openai-codex --model gpt-5.6-sol --thinking low; rc=\$?; printf \"PI_EXIT=%s\\n\" \"\$rc\"; sleep 300'"
+"$TMUX" -L "$SOCKET" resize-window -t "$SESSION" -x 220 -y 60
+
+wait_for_text "fm-primary-pi-watch.ts" 120 || fail "Pi primary extensions did not load"
+send_prompt "$QUESTION"
+wait_for_session_probe answer 240 || fail "direct captain answer did not complete"
+
+send_prompt "/fm-watch-arm-pi"
+wait_for_text "FIRSTMATE WATCHER WAKE" 180 || fail "custom watcher prompt did not render"
+wait_for_session_probe automation-settled 300 || fail "automated supervision turn did not settle"
+
+send_prompt "/compact"
+wait_for_session_probe compaction 480 || fail "real Pi compaction did not complete"
+FILE=$(session_file)
+[ -n "$FILE" ] || fail "isolated Pi session file is missing"
+BOUNDARY=$(python3 - "$FILE" <<'PY'
+import json
+import sys
+entries = [json.loads(line) for line in open(sys.argv[1], encoding="utf-8")]
+question = next(i for i, e in enumerate(entries) if e.get("type") == "message" and e.get("message", {}).get("role") == "user" and "DIRECT-CAPTAIN-Q-7" in json.dumps(e["message"].get("content"), ensure_ascii=False))
+answer = next(i for i, e in enumerate(entries) if e.get("type") == "message" and e.get("message", {}).get("role") == "assistant" and "DIRECT-CAPTAIN-A-7" in json.dumps(e["message"].get("content"), ensure_ascii=False) and e["message"].get("stopReason") == "stop")
+automation = next(i for i, e in enumerate(entries) if e.get("type") == "custom_message" and e.get("customType") == "firstmate-watcher-wake")
+compaction_index = max(i for i, e in enumerate(entries) if e.get("type") == "compaction")
+compaction = entries[compaction_index]
+if not (question < answer < automation < compaction_index):
+    raise SystemExit("unexpected interleaving")
+kept = compaction.get("firstKeptEntryId", "")
+kept_index = next((i for i, e in enumerate(entries) if e.get("id") == kept), len(entries))
+if kept_index <= answer:
+    raise SystemExit("compaction retained the original direct exchange instead of exercising continuity recovery")
+kept_type = "none" if not kept else entries[kept_index].get("type", "unknown")
+print(f"firstKeptType={kept_type}")
+PY
+) || fail "real compaction did not cut after the direct exchange"
+
+send_prompt "$FOLLOWUP"
+wait_for_session_probe continuity-answer 300 || fail "resumed model did not identify the prior question and completed answer"
+
+for _ in $(seq 1 100); do
+  [ -f "$CONTEXT_LOG" ] && break
+  sleep 0.1
+done
+[ -f "$CONTEXT_LOG" ] || fail "post-compaction provider context was not captured"
+python3 - "$CONTEXT_LOG" <<'PY' || fail "post-compaction model context lacks exact direct-exchange continuity"
+import json
+import sys
+messages = json.load(open(sys.argv[1], encoding="utf-8"))
+records = [m for m in messages if m.get("role") == "custom" and m.get("customType") == "firstmate-direct-exchange-continuity"]
+assert records, "continuity custom message absent"
+content = records[-1]["content"]
+assert "DIRECT-CAPTAIN-Q-7: Which harbor token should remain reserved?" in content
+assert "DIRECT-CAPTAIN-A-7: amber." in content
+assert "ANSWERED" in content
+assert "not human-authored input" in content
+PY
+
+printf 'ok - Pi %s live compaction rebuilt the exact answered captain exchange across a custom watcher turn (%s)\n' "$PI_VERSION" "$BOUNDARY"
diff --git a/tests/fm-pi-retry-continuity.test.sh b/tests/fm-pi-retry-continuity.test.sh
new file mode 100755
index 00000000000..b36e911c63b
--- /dev/null
+++ b/tests/fm-pi-retry-continuity.test.sh
@@ -0,0 +1,42 @@
+#!/usr/bin/env bash
+# shellcheck source=tests/test-entry.sh
+. "$(dirname "$0")/test-entry.sh"
+set -u
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+selected_node=
+pi_package=${FM_PI_PACKAGE_DIR:-}
+
+while IFS= read -r node_bin; do
+  node_dir=$(dirname "$node_bin")
+  if [ -z "$selected_node" ] && [ "$("$node_bin" -p 'process.features?.typescript ? "yes" : "no"' 2>/dev/null)" = yes ]; then
+    selected_node=$node_dir
+  fi
+  if [ -z "$pi_package" ] && [ -x "$node_dir/npm" ]; then
+    candidate=$(PATH="$node_dir:$PATH" "$node_dir/npm" root -g 2>/dev/null)/@earendil-works/pi-coding-agent
+    if [ -f "$candidate/package.json" ]; then
+      pi_package=$candidate
+    fi
+  fi
+done < <(
+  {
+    type -a -p node 2>/dev/null
+    for candidate_node in /opt/homebrew/opt/node/bin/node /usr/local/opt/node/bin/node; do
+      [ -x "$candidate_node" ] && printf '%s\n' "$candidate_node"
+    done
+  } | awk '!seen[$0]++'
+)
+
+[ -n "$selected_node" ] || { echo "not ok - no available Node runtime supports TypeScript import" >&2; exit 1; }
+[ -n "$pi_package" ] && [ -f "$pi_package/package.json" ] || { echo "not ok - installed Pi package not found" >&2; exit 1; }
+
+out=$(env -u FM_TEST_RUNNER_ACTIVE PATH="$selected_node:$PATH" FM_PI_PACKAGE_DIR="$pi_package" FM_PI_RETRY_CONTINUITY_PROOF=1 "$ROOT/tests/fm-pi-watch-extension.test.sh") || {
+  printf '%s\n' "$out" >&2
+  exit 1
+}
+printf '%s\n' "$out" | grep -Fqx 'PI_RETRY_CONTINUITY_PROOF_OK' || {
+  printf '%s\n' "$out" >&2
+  echo "not ok - Pi retry continuity proof token missing" >&2
+  exit 1
+}
+printf '%s\n' 'PI_RETRY_CONTINUITY_PROOF_OK'
diff --git a/tests/fm-pi-watch-extension.test.sh b/tests/fm-pi-watch-extension.test.sh
index 7695055e3ce..3593a853b1b 100755
--- a/tests/fm-pi-watch-extension.test.sh
+++ b/tests/fm-pi-watch-extension.test.sh
@@ -39,7 +39,6 @@ test_tracked_extension_present_and_self_hashing() {
   assert_contains "$text" "fm_watch_arm_pi" "tracked extension missing tool name"
   assert_contains "$text" "fm-watch-arm-pi" "tracked extension missing command name"
   assert_contains "$text" "fm-watch-arm.sh" "tracked extension missing watcher arm"
-  assert_contains "$text" "sendUserMessage" "tracked extension missing Pi wake API"
   assert_contains "$text" "deliverAs: \"followUp\"" "tracked extension missing followUp delivery"
   assert_contains "$text" ".pi-watch-extension-loaded" "tracked extension missing loaded marker"
   assert_contains "$text" 'createHash("sha256").update(readFileSync(extensionFile)).digest("hex")' "tracked extension does not self-hash its own content for extensionVersion"
@@ -97,14 +96,19 @@ import { pathToFileURL } from "node:url";
 let handler = null;
 let notification = "";
 let prompt = "";
+let promptType = "";
+let promptOptions = null;
 const pi = {
   on() {},
+  appendEntry() {},
   registerCommand(name, options) {
     if (name === "fm-watch-arm-pi") handler = options.handler;
   },
   registerTool() {},
-  sendUserMessage: async (message) => {
-    prompt = message;
+  sendMessage(message, options) {
+    prompt = message.content;
+    promptType = message.customType;
+    promptOptions = options;
   },
 };
 writeFileSync(`${process.env.FM_HOME}/state/.lock`, `${process.pid}\n`);
@@ -136,6 +140,14 @@ if (!prompt.includes("FIRSTMATE WATCHER WAKE")) {
   console.error(`missing follow-up prompt: ${prompt}`);
   process.exit(1);
 }
+if (promptType !== "firstmate-watcher-wake") {
+  console.error(`wake was not a custom watcher message: ${promptType}`);
+  process.exit(1);
+}
+if (promptOptions?.deliverAs !== "followUp" || promptOptions?.triggerTurn !== true) {
+  console.error(`unexpected custom wake delivery: ${JSON.stringify(promptOptions)}`);
+  process.exit(1);
+}
 if (!prompt.includes("external healthy watcher")) {
   console.error(prompt);
   process.exit(1);
@@ -172,11 +184,12 @@ import { pathToFileURL } from "node:url";
 let tool = null;
 const pi = {
   on() {},
+  appendEntry() {},
   registerCommand() {},
   registerTool(candidate) {
     if (candidate.name === "fm_watch_arm_pi") tool = candidate;
   },
-  sendUserMessage: async () => {},
+  sendMessage() {},
 };
 writeFileSync(`${process.env.FM_HOME}/state/.lock`, `${process.pid}\n`);
 const mod = await import(pathToFileURL(process.env.PLUGIN).href);
@@ -220,9 +233,10 @@ const pi = {
   on(event, handler) {
     handlers.set(event, handler);
   },
+  appendEntry() {},
   registerCommand() {},
   registerTool() {},
-  sendUserMessage: async () => {},
+  sendMessage() {},
 };
 const before = process.listenerCount("exit");
 const mod = await import(pathToFileURL(process.env.PLUGIN).href);
@@ -266,11 +280,12 @@ import { pathToFileURL } from "node:url";
 let tool = null;
 const pi = {
   on() {},
+  appendEntry() {},
   registerCommand() {},
   registerTool(candidate) {
     if (candidate.name === "fm_watch_arm_pi") tool = candidate;
   },
-  sendUserMessage: async () => {},
+  sendMessage() {},
 };
 writeFileSync(`${process.env.FM_HOME}/state/.lock`, `${process.pid}\n`);
 const mod = await import(pathToFileURL(process.env.PLUGIN).href);
@@ -300,6 +315,543 @@ EOF
   pass "Pi process-exit cleanup stops the attached arm child"
 }
 
+test_pi_compaction_preserves_direct_exchange_and_pending_input() {
+  local repo home plugin pi_bin pi_package_dir out status
+  fm_node_supports_ts_import || { pass "node lacks .ts import support, skipping Pi compaction-continuity check"; return; }
+  pi_bin=$(command -v pi 2>/dev/null || true)
+  [ -n "$pi_bin" ] || { pass "Pi is unavailable, skipping installed SessionManager compaction-continuity check"; return; }
+  pi_package_dir=$(cd "$(dirname "$pi_bin")/../lib/node_modules/@earendil-works/pi-coding-agent" 2>/dev/null && pwd -P) || {
+    pass "installed Pi package is unavailable, skipping SessionManager compaction-continuity check"
+    return
+  }
+  repo="$TMP_ROOT/pi-compaction-continuity-root"
+  home="$TMP_ROOT/pi-compaction-continuity-home"
+  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 'signal: deterministic supervision prompt\n'
+SH
+  chmod +x "$repo/bin/fm-watch-arm.sh"
+  out=$(NODE_OPTIONS=--disable-warning=ExperimentalWarning PLUGIN="$plugin" PI_PACKAGE_DIR="$pi_package_dir" REPO="$repo" FM_HOME="$home" FM_ROOT_OVERRIDE="$repo" node --input-type=module 2>&1 <<'EOF'
+import { writeFileSync } from "node:fs";
+import { pathToFileURL } from "node:url";
+
+const { SessionManager } = await import(pathToFileURL(`${process.env.PI_PACKAGE_DIR}/dist/index.js`).href);
+const { Agent } = await import(pathToFileURL(`${process.env.PI_PACKAGE_DIR}/node_modules/@earendil-works/pi-agent-core/dist/index.js`).href);
+const { AssistantMessageEventStream } = await import(pathToFileURL(`${process.env.PI_PACKAGE_DIR}/node_modules/@earendil-works/pi-ai/dist/index.js`).href);
+let sessionManager = SessionManager.inMemory(process.env.REPO);
+const handlers = new Map();
+let tool = null;
+let automatedMessage = null;
+let automatedOptions = null;
+let pendingMessages = false;
+
+const pi = {
+  on(event, handler) {
+    handlers.set(event, handler);
+  },
+  appendEntry(customType, data) {
+    sessionManager.appendCustomEntry(customType, data);
+  },
+  registerCommand() {},
+  registerTool(candidate) {
+    if (candidate.name === "fm_watch_arm_pi") tool = candidate;
+  },
+  sendMessage(message, options) {
+    automatedMessage = message;
+    automatedOptions = options;
+  },
+};
+const ctx = {
+  get sessionManager() {
+    return sessionManager;
+  },
+  hasPendingMessages: () => pendingMessages,
+};
+async function input(text, streamingBehavior, images) {
+  await handlers.get("input")(
+    { type: "input", text, images, source: "interactive", streamingBehavior },
+    ctx,
+  );
+}
+async function finishMessage(message) {
+  await handlers.get("message_end")({ type: "message_end", message }, ctx);
+  sessionManager.appendMessage(message);
+}
+async function startAgent() {
+  await handlers.get("agent_start")({ type: "agent_start" }, ctx);
+}
+
+writeFileSync(`${process.env.FM_HOME}/state/.lock`, `${process.pid}\n`);
+const mod = await import(pathToFileURL(process.env.PLUGIN).href);
+mod.default(pi);
+if (!tool) throw new Error("Pi watch tool was not registered");
+
+const providerContexts = [];
+let releaseInitial;
+let providerCall = 0;
+const usage = {
+  input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0,
+  cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
+};
+const agent = new Agent({
+  initialState: {
+    systemPrompt: "queue regression",
+    model: { id: "fixture", name: "fixture", api: "openai-completions", provider: "fixture", baseUrl: "http://fixture", reasoning: false, input: ["text", "image"], cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, contextWindow: 100000, maxTokens: 1000 },
+    tools: [],
+  },
+  convertToLlm: (messages) => messages.map((message) => message.role === "custom"
+    ? { role: "user", content: [{ type: "text", text: message.content }], timestamp: message.timestamp }
+    : message),
+  streamFn: (_model, context) => {
+    providerContexts.push(structuredClone(context.messages));
+    const call = ++providerCall;
+    const stream = new AssistantMessageEventStream();
+    const finish = () => {
+      const message = { role: "assistant", content: [{ type: "text", text: `provider-${call}` }], api: "openai-completions", provider: "fixture", model: "fixture", usage, stopReason: "stop", timestamp: Date.now() };
+      stream.push({ type: "start", partial: { ...message, content: [], stopReason: "pending" } });
+      stream.push({ type: "done", reason: "stop", message });
+    };
+    if (call === 1) releaseInitial = finish;
+    else queueMicrotask(finish);
+    return stream;
+  },
+});
+const exactImage = { type: "image", data: "QUEUE-ORDER-IMAGE-DATA", mimeType: "image/png" };
+const initialRun = agent.prompt("INITIAL-HELD-RESPONSE");
+for (let i = 0; i < 100 && !releaseInitial; i += 1) await new Promise((resolve) => setTimeout(resolve, 1));
+if (!releaseInitial) throw new Error("controlled provider did not hold the initial response");
+agent.followUp({ role: "custom", customType: "firstmate-watcher-wake", content: "OLDER-AUTOMATION-FOLLOWUP", display: true, timestamp: Date.now() });
+agent.steer({ role: "user", content: [{ type: "text", text: "LATER-HUMAN-IMAGE-STEER" }, exactImage], timestamp: Date.now() });
+releaseInitial();
+await initialRun;
+await agent.waitForIdle();
+if (providerContexts.length !== 3) throw new Error(`expected three provider turns, got ${providerContexts.length}`);
+const secondTail = providerContexts[1].at(-1);
+const thirdTail = providerContexts[2].at(-1);
+if (secondTail?.role !== "user" || JSON.stringify(secondTail.content) !== JSON.stringify([{ type: "text", text: "LATER-HUMAN-IMAGE-STEER" }, exactImage])) {
+  throw new Error(`human image steer was reordered or altered: ${JSON.stringify(secondTail)}`);
+}
+if (thirdTail?.role !== "user" || JSON.stringify(thirdTail.content) !== JSON.stringify([{ type: "text", text: "OLDER-AUTOMATION-FOLLOWUP" }])) {
+  throw new Error(`older automation was starved or delivered early: ${JSON.stringify(thirdTail)}`);
+}
+
+const question = "DIRECT-CAPTAIN-Q-7: Which harbor token should remain reserved?";
+const answer = "DIRECT-CAPTAIN-A-7: amber.";
+const followup = "Which question did I ask before the automated supervision turn, and was it answered?";
+const queued = "QUEUED-CAPTAIN-Q-8: preserve this pending input too";
+const questionContent = [{ type: "text", text: question }];
+const answerContent = [{ type: "text", text: answer }];
+
+await input(question);
+await startAgent();
+await finishMessage({ role: "user", content: questionContent, timestamp: 1700000000100 });
+await finishMessage({
+  role: "assistant",
+  content: answerContent,
+  stopReason: "stop",
+  timestamp: 1700000000200,
+});
+
+await tool.execute("tool-compaction", {}, undefined, undefined, {});
+for (let i = 0; i < 100 && !automatedMessage; i += 1) {
+  await new Promise((resolve) => setTimeout(resolve, 10));
+}
+if (!automatedMessage) throw new Error("watcher did not produce an automated prompt");
+if (automatedMessage.customType !== "firstmate-watcher-wake") {
+  throw new Error(`watcher prompt masqueraded as user input: ${automatedMessage.customType}`);
+}
+if (automatedOptions?.deliverAs !== "followUp" || automatedOptions?.triggerTurn !== true) {
+  throw new Error(`wrong watcher queue behavior: ${JSON.stringify(automatedOptions)}`);
+}
+const automatedEntryId = sessionManager.appendCustomMessageEntry(
+  automatedMessage.customType,
+  automatedMessage.content,
+  true,
+  automatedMessage.details,
+);
+sessionManager.appendMessage({
+  role: "assistant",
+  content: [{ type: "text", text: "Automated supervision handled." }],
+  stopReason: "stop",
+  timestamp: 1700000000300,
+});
+const compactionId = sessionManager.appendCompaction(
+  "Lossy summary without the direct captain exchange.",
+  automatedEntryId,
+  131874,
+  { fixture: "automated-boundary" },
+  true,
+);
+const compaction = sessionManager.getEntry(compactionId);
+if (compaction?.type !== "compaction" || compaction.firstKeptEntryId !== automatedEntryId) {
+  throw new Error("fixture did not compact at the automated-message boundary");
+}
+
+await input(followup);
+const followupContent = [{ type: "text", text: followup }];
+await finishMessage({ role: "user", content: followupContent, timestamp: 1700000000400 });
+const rebuilt = sessionManager.buildSessionContext().messages;
+if (rebuilt.some((message) => message.role === "user" && JSON.stringify(message.content) === JSON.stringify(questionContent))) {
+  throw new Error("SessionManager fixture unexpectedly retained the pre-boundary captain question");
+}
+if (!rebuilt.some((message) => message.role === "custom" && message.customType === "firstmate-watcher-wake")) {
+  throw new Error("SessionManager fixture lost the retained custom supervision prompt");
+}
+const result = await handlers.get("context")({ type: "context", messages: rebuilt }, ctx);
+const continuity = result?.messages?.find(
+  (message) => message.role === "custom" && message.customType === "firstmate-direct-exchange-continuity",
+);
+if (!continuity) throw new Error("rebuilt context lacks direct-exchange continuity");
+if (continuity.content.includes("Lossy summary without") && !continuity.content.includes(question)) {
+  throw new Error("continuity trusted the lossy summary instead of the exact exchange");
+}
+if (!continuity.content.includes("ANSWERED") || !continuity.content.includes(JSON.stringify(questionContent))) {
+  throw new Error(`exact answered question missing: ${continuity.content}`);
+}
+if (!continuity.content.includes(JSON.stringify(answerContent))) {
+  throw new Error(`exact assistant answer missing: ${continuity.content}`);
+}
+if (!continuity.content.includes("OPEN_REPLY_OBLIGATION") || !continuity.content.includes(JSON.stringify(followupContent))) {
+  throw new Error(`follow-up reply obligation missing: ${continuity.content}`);
+}
+if (!continuity.content.includes("not human-authored input") || !continuity.content.includes("custom messages")) {
+  throw new Error("continuity does not distinguish extension prompts from human input");
+}
+const continuityIndex = result.messages.indexOf(continuity);
+const followupIndex = result.messages.findIndex(
+  (message) => message.role === "user" && JSON.stringify(message.content) === JSON.stringify(followupContent),
+);
+if (continuityIndex < 0 || followupIndex < 0 || continuityIndex >= followupIndex) {
+  throw new Error("continuity metadata displaced the current human follow-up as the final prompt");
+}
+
+pendingMessages = true;
+await input(queued, "steer");
+const whilePending = await handlers.get("context")({ type: "context", messages: rebuilt }, ctx);
+const pendingContinuity = whilePending?.messages?.find(
+  (message) => message.role === "custom" && message.customType === "firstmate-direct-exchange-continuity",
+);
+if (pendingContinuity?.content.includes(queued)) {
+  throw new Error("pending human input bypassed Pi's queue and was injected early");
+}
+pendingMessages = false;
+sessionManager = SessionManager.inMemory(process.env.REPO);
+const provisional = "PROVISIONAL-CONSUMED-Q-8B";
+pendingMessages = true;
+await input(provisional, "steer");
+const provisionalResult = await handlers.get("context")({ type: "context", messages: rebuilt }, ctx);
+if (provisionalResult?.messages?.some((message) => message.customType === "firstmate-direct-exchange-continuity" && message.content.includes(provisional))) {
+  throw new Error("consumed provisional input was admitted by unrelated pending automation");
+}
+pendingMessages = false;
+const afterAutomationDrain = await handlers.get("context")({ type: "context", messages: rebuilt }, ctx);
+if (afterAutomationDrain?.messages?.some((message) => message.customType === "firstmate-direct-exchange-continuity" && message.content.includes(provisional))) {
+  throw new Error("consumed provisional input became continuity after unrelated automation drained");
+}
+pendingMessages = true;
+await input(provisional, "steer");
+await finishMessage({ role: "user", content: [{ type: "text", text: provisional }], timestamp: 1700000000410 });
+pendingMessages = false;
+const consumedRetryBoundaryId = sessionManager.appendCustomMessageEntry(
+  "firstmate-watcher-wake",
+  "FIRSTMATE WATCHER WAKE: unrelated automation after identical accepted retry",
+  true,
+  { version: 1, source: "firstmate-extension", kind: "watcher-wake" },
+);
+sessionManager.appendCompaction(
+  "Lossy summary that omits the identical accepted retry.",
+  consumedRetryBoundaryId,
+  131874,
+  { fixture: "consumed-identical-retry-boundary" },
+  true,
+);
+const consumedRetryResult = await handlers.get("context")(
+  { type: "context", messages: sessionManager.buildSessionContext().messages },
+  ctx,
+);
+const consumedRetryContinuity = consumedRetryResult?.messages?.find(
+  (message) => message.role === "custom" && message.customType === "firstmate-direct-exchange-continuity",
+);
+const consumedRetryObligations = consumedRetryContinuity?.content.match(/OPEN_REPLY_OBLIGATION/g) ?? [];
+if (consumedRetryObligations.length !== 1 || !consumedRetryContinuity.content.includes(JSON.stringify([{ type: "text", text: provisional }]))) {
+  throw new Error(`consumed input was confused with identical accepted retry: ${consumedRetryContinuity?.content}`);
+}
+const consumedRetryEntries = sessionManager.getBranch().filter((entry) => entry.type === "custom");
+const consumedRetryObservations = consumedRetryEntries.filter(
+  (entry) => entry.customType === "firstmate-direct-input-observation" && entry.data?.inputText === provisional,
+);
+const consumedRetryExchanges = consumedRetryEntries.filter(
+  (entry) => entry.customType === "firstmate-direct-exchange" && entry.data?.event === "submitted" && entry.data?.inputText === provisional,
+);
+if (consumedRetryObservations.length !== 2 || consumedRetryExchanges.length !== 1) {
+  throw new Error("input observations were promoted instead of admitting only the delivered identical retry");
+}
+
+sessionManager = SessionManager.inMemory(process.env.REPO);
+const duplicate = "DUPLICATE-QUEUED-CAPTAIN-Q: preserve both obligations";
+const duplicateContent = [{ type: "text", text: duplicate }];
+await input(duplicate, "steer");
+await input(duplicate, "steer");
+await finishMessage({ role: "user", content: duplicateContent, timestamp: 1700000000420 });
+await finishMessage({ role: "user", content: duplicateContent, timestamp: 1700000000430 });
+const duplicateEvents = sessionManager.getBranch().filter(
+  (entry) => entry.type === "custom" && entry.customType === "firstmate-direct-exchange",
+);
+const duplicateSubmissions = duplicateEvents.filter(
+  (entry) => entry.data?.event === "submitted" && entry.data?.inputText === duplicate,
+);
+const duplicateDeliveries = duplicateEvents.filter(
+  (entry) => entry.data?.event === "delivered" && JSON.stringify(entry.data?.content) === JSON.stringify(duplicateContent),
+);
+if (duplicateSubmissions.length !== 2 || new Set(duplicateSubmissions.map((entry) => entry.data.exchangeId)).size !== 2) {
+  throw new Error("duplicate submissions did not retain distinct exchange identities");
+}
+if (
+  duplicateDeliveries.length !== 2 ||
+  duplicateDeliveries[0]?.data?.exchangeId !== duplicateSubmissions[0]?.data?.exchangeId ||
+  duplicateDeliveries[1]?.data?.exchangeId !== duplicateSubmissions[1]?.data?.exchangeId
+) {
+  throw new Error("duplicate exact deliveries were not associated in submission order");
+}
+const duplicateBoundaryId = sessionManager.appendCustomMessageEntry(
+  "firstmate-watcher-wake",
+  "FIRSTMATE WATCHER WAKE: compact duplicate direct inputs",
+  true,
+  { version: 1, source: "firstmate-extension", kind: "watcher-wake" },
+);
+sessionManager.appendCompaction(
+  "Lossy summary that omits both duplicate direct inputs.",
+  duplicateBoundaryId,
+  131874,
+  { fixture: "duplicate-input-boundary" },
+  true,
+);
+const duplicateResult = await handlers.get("context")(
+  { type: "context", messages: sessionManager.buildSessionContext().messages },
+  ctx,
+);
+const duplicateContinuity = duplicateResult?.messages?.find(
+  (message) => message.role === "custom" && message.customType === "firstmate-direct-exchange-continuity",
+);
+const duplicateObligations = duplicateContinuity?.content.match(/OPEN_REPLY_OBLIGATION/g) ?? [];
+if (duplicateObligations.length !== 2) {
+  throw new Error(`compaction did not preserve two duplicate reply obligations: ${duplicateContinuity?.content}`);
+}
+
+sessionManager = SessionManager.inMemory(process.env.REPO);
+pendingMessages = false;
+const imageQueued = "QUEUED-CAPTAIN-IMAGE-Q-9: identify the attached harbor signal";
+const image = { type: "image", data: "iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB", mimeType: "image/png" };
+const imageSubmittedContent = [{ type: "text", text: imageQueued }, image];
+pendingMessages = true;
+await input(imageQueued, "followUp", [image]);
+pendingMessages = false;
+const imageBoundaryId = sessionManager.appendCustomMessageEntry(
+  "firstmate-watcher-wake",
+  "FIRSTMATE WATCHER WAKE: compaction before queued image delivery",
+  true,
+  { version: 1, source: "firstmate-extension", kind: "watcher-wake" },
+);
+sessionManager.appendCompaction(
+  "Lossy summary that omits the queued image submission.",
+  imageBoundaryId,
+  131874,
+  { fixture: "image-before-delivery-boundary" },
+  true,
+);
+const imageRebuilt = sessionManager.buildSessionContext().messages;
+if (JSON.stringify(imageRebuilt).includes(image.data)) {
+  throw new Error("image-before-delivery fixture unexpectedly retained the queued image in Pi context");
+}
+const imageBeforeDelivery = await handlers.get("context")({ type: "context", messages: imageRebuilt }, ctx);
+if (imageBeforeDelivery?.messages?.some((message) => message.customType === "firstmate-direct-exchange-continuity" && message.content.includes(imageQueued))) {
+  throw new Error("unadmitted queued image bypassed Pi ownership before delivery");
+}
+const imageSubmissionEntry = sessionManager.getBranch().find(
+  (entry) => entry.type === "custom" && entry.customType === "firstmate-direct-input-observation",
+);
+if (JSON.stringify(imageSubmissionEntry?.data?.inputContent) !== JSON.stringify(imageSubmittedContent)) {
+  throw new Error("provisional session contract did not preserve exact queued image content");
+}
+await finishMessage({ role: "user", content: imageSubmittedContent, timestamp: 1700000000450 });
+const deliveredBoundaryId = sessionManager.appendCustomMessageEntry(
+  "firstmate-watcher-wake",
+  "FIRSTMATE WATCHER WAKE: compaction after exact image delivery",
+  true,
+  { version: 1, source: "firstmate-extension", kind: "watcher-wake" },
+);
+sessionManager.appendCompaction(
+  "Lossy summary that omits the delivered image question.",
+  deliveredBoundaryId,
+  131874,
+  { fixture: "image-after-delivery-boundary" },
+  true,
+);
+const imageAfterDelivery = await handlers.get("context")(
+  { type: "context", messages: sessionManager.buildSessionContext().messages },
+  ctx,
+);
+const imageContinuity = imageAfterDelivery?.messages?.find(
+  (message) => message.role === "custom" && message.customType === "firstmate-direct-exchange-continuity",
+);
+if (!imageContinuity?.content.includes("OPEN_REPLY_OBLIGATION") || !imageContinuity.content.includes(JSON.stringify(imageSubmittedContent))) {
+  throw new Error(`compaction lost exact admitted image obligation: ${imageContinuity?.content}`);
+}
+
+sessionManager = SessionManager.inMemory(process.env.REPO);
+pendingMessages = false;
+const unanswered = "UNANSWERED-CAPTAIN-Q-10: Which reply remains due?";
+const unansweredContent = [{ type: "text", text: unanswered }];
+await input(unanswered);
+await finishMessage({ role: "user", content: unansweredContent, timestamp: 1700000000500 });
+const openAutomationId = sessionManager.appendCustomMessageEntry(
+  "firstmate-watcher-wake",
+  "FIRSTMATE WATCHER WAKE: deterministic open-obligation boundary",
+  true,
+  { version: 1, source: "firstmate-extension", kind: "watcher-wake" },
+);
+sessionManager.appendMessage({
+  role: "assistant",
+  content: [{ type: "text", text: "Automation handled without answering the captain." }],
+  stopReason: "stop",
+  timestamp: 1700000000600,
+});
+sessionManager.appendCompaction(
+  "Lossy summary that omits the unanswered direct question.",
+  openAutomationId,
+  131874,
+  { fixture: "open-obligation-boundary" },
+  true,
+);
+const openRebuilt = sessionManager.buildSessionContext().messages;
+if (openRebuilt.some((message) => message.role === "user")) {
+  throw new Error("open-obligation fixture unexpectedly retained the captain user message");
+}
+const openResult = await handlers.get("context")({ type: "context", messages: openRebuilt }, ctx);
+const openContinuity = openResult?.messages?.find(
+  (message) => message.role === "custom" && message.customType === "firstmate-direct-exchange-continuity",
+);
+if (!openContinuity?.content.includes("OPEN_REPLY_OBLIGATION") || !openContinuity.content.includes(JSON.stringify(unansweredContent))) {
+  throw new Error("compaction lost the exact unanswered captain obligation");
+}
+
+sessionManager = SessionManager.inMemory(process.env.REPO);
+const abortedQuestion = [{ type: "text", text: "ABORTED-RUN-CAPTAIN-Q: which answer remains due?" }];
+const laterQuestion = [{ type: "text", text: "LATER-RUN-CAPTAIN-Q: answer only this question" }];
+const joinedQuestion = [{ type: "text", text: "JOINED-RUN-CAPTAIN-Q: preserve joined delivery" }];
+const laterAnswer = [{ type: "text", text: "LATER-RUN-CAPTAIN-A: this question only" }];
+await startAgent();
+await finishMessage({ role: "user", content: abortedQuestion, timestamp: 1700000000700 });
+await startAgent();
+await finishMessage({ role: "user", content: laterQuestion, timestamp: 1700000000800 });
+await finishMessage({ role: "user", content: joinedQuestion, timestamp: 1700000000850 });
+await finishMessage({
+  role: "assistant",
+  content: laterAnswer,
+  stopReason: "stop",
+  timestamp: 1700000000900,
+});
+const runExchangeEvents = sessionManager.getBranch().filter(
+  (entry) => entry.type === "custom" && entry.customType === "firstmate-direct-exchange",
+);
+const laterRunAnswers = runExchangeEvents.filter(
+  (entry) => entry.data?.event === "answered" && JSON.stringify(entry.data?.content) === JSON.stringify(laterAnswer),
+);
+if (laterRunAnswers.length !== 2) {
+  throw new Error(`same-run deliveries were not both answered: ${JSON.stringify(laterRunAnswers)}`);
+}
+const runBoundaryId = sessionManager.appendCustomMessageEntry(
+  "firstmate-watcher-wake",
+  "FIRSTMATE WATCHER WAKE: compact run-attribution fixture",
+  true,
+  { version: 1, source: "firstmate-extension", kind: "watcher-wake" },
+);
+sessionManager.appendCompaction(
+  "Lossy summary that omits both run-attribution questions.",
+  runBoundaryId,
+  131874,
+  { fixture: "run-attribution-boundary" },
+  true,
+);
+const runResult = await handlers.get("context")(
+  { type: "context", messages: sessionManager.buildSessionContext().messages },
+  ctx,
+);
+const runContinuity = runResult?.messages?.find(
+  (message) => message.role === "custom" && message.customType === "firstmate-direct-exchange-continuity",
+);
+if (!runContinuity?.content.includes(`OPEN_REPLY_OBLIGATION\nHuman input, exact JSON: ${JSON.stringify(abortedQuestion)}`)) {
+  throw new Error(`aborted run obligation was incorrectly closed: ${runContinuity?.content}`);
+}
+if (!runContinuity.content.includes(`ANSWERED\nHuman input, exact JSON: ${JSON.stringify(joinedQuestion)}`)) {
+  throw new Error(`joined delivery in the later run was not answered: ${runContinuity.content}`);
+}
+if (!runContinuity.content.includes(JSON.stringify(laterAnswer))) {
+  throw new Error(`later run answer was not preserved: ${runContinuity.content}`);
+}
+
+sessionManager = SessionManager.inMemory(process.env.REPO);
+const retryQuestion = [{ type: "text", text: "RETRY-CAPTAIN-Q: close this after retry" }];
+const retryAnswer = [{ type: "text", text: "RETRY-CAPTAIN-A: closed after retry" }];
+await startAgent();
+await finishMessage({ role: "user", content: retryQuestion, timestamp: 1700000001000 });
+await startAgent();
+await finishMessage({ role: "assistant", content: retryAnswer, stopReason: "stop", timestamp: 1700000001100 });
+const retryEvents = sessionManager.getBranch().filter(
+  (entry) => entry.type === "custom" && entry.customType === "firstmate-direct-exchange",
+);
+const retryAnswered = retryEvents.find(
+  (entry) => entry.data?.event === "answered" && JSON.stringify(entry.data?.content) === JSON.stringify(retryAnswer),
+);
+if (!retryAnswered) throw new Error("successful retry did not answer the retained captain exchange");
+
+sessionManager = SessionManager.inMemory(process.env.REPO);
+const failedQuestion = [{ type: "text", text: "FAILED-CAPTAIN-Q: leave this open" }];
+const replacementQuestion = [{ type: "text", text: "REPLACEMENT-CAPTAIN-Q: answer this instead" }];
+const replacementAnswer = [{ type: "text", text: "REPLACEMENT-CAPTAIN-A: answered replacement" }];
+await startAgent();
+await finishMessage({ role: "user", content: failedQuestion, timestamp: 1700000001200 });
+await startAgent();
+await finishMessage({ role: "user", content: replacementQuestion, timestamp: 1700000001300 });
+await finishMessage({ role: "assistant", content: replacementAnswer, stopReason: "stop", timestamp: 1700000001400 });
+const replacementBoundaryId = sessionManager.appendCustomMessageEntry(
+  "firstmate-watcher-wake",
+  "FIRSTMATE WATCHER WAKE: retry replacement boundary",
+  true,
+  { version: 1, source: "firstmate-extension", kind: "watcher-wake" },
+);
+sessionManager.appendCompaction(
+  "Lossy summary that omits retry lifecycle exchanges.",
+  replacementBoundaryId,
+  131874,
+  { fixture: "retry-replacement-boundary" },
+  true,
+);
+const replacementResult = await handlers.get("context")(
+  { type: "context", messages: sessionManager.buildSessionContext().messages },
+  ctx,
+);
+const replacementContinuity = replacementResult?.messages?.find(
+  (message) => message.role === "custom" && message.customType === "firstmate-direct-exchange-continuity",
+);
+if (!replacementContinuity?.content.includes(`OPEN_REPLY_OBLIGATION\nHuman input, exact JSON: ${JSON.stringify(failedQuestion)}`)) {
+  throw new Error(`failed original question was incorrectly closed: ${replacementContinuity?.content}`);
+}
+if (!replacementContinuity.content.includes(`ANSWERED\nHuman input, exact JSON: ${JSON.stringify(replacementQuestion)}`)) {
+  throw new Error(`replacement question was not answered: ${replacementContinuity.content}`);
+}
+EOF
+)
+  status=$?
+  [ "$status" -eq 0 ] || printf '%s\n' "$out" >&2
+  expect_code 0 "$status" "Pi compaction continuity must preserve exact direct exchange and pending-input state"
+  [ -z "$out" ] || fail "Pi compaction-continuity test printed output: $out"
+  pass "Pi compaction continuity preserves exact human exchange across automated custom prompts"
+}
+
 test_opencode_primary_watch_plugin_uses_effective_state_home() {
   local plugin repo home log out status
   plugin="$TMP_ROOT/opencode-effective-state-primary.mjs"
@@ -863,12 +1415,25 @@ EOF
   pass "OpenCode healthy arm output does not suppress the turn-end guard"
 }
 
+if [ "${FM_PI_EXACT_DELIVERY_PROOF:-0}" = 1 ]; then
+  test_pi_compaction_preserves_direct_exchange_and_pending_input
+  printf '%s\n' 'PI_EXACT_DELIVERY_ASSOCIATION_PROOF_OK'
+  exit 0
+fi
+
+if [ "${FM_PI_RETRY_CONTINUITY_PROOF:-0}" = 1 ]; then
+  test_pi_compaction_preserves_direct_exchange_and_pending_input
+  printf '%s\n' 'PI_RETRY_CONTINUITY_PROOF_OK'
+  exit 0
+fi
+
 test_tracked_extension_present_and_self_hashing
 test_spawn_template_mentions_pi_watch_placeholder
 test_pi_extension_reports_external_healthy_watcher
 test_pi_tool_returns_agent_tool_result
 test_pi_process_exit_cleanup_listener_lifecycle
 test_pi_process_exit_cleanup_stops_arm_child
+test_pi_compaction_preserves_direct_exchange_and_pending_input
 test_opencode_primary_watch_plugin_uses_effective_state_home
 test_opencode_primary_watch_plugin_sources_effective_config
 test_opencode_primary_watch_plugin_requires_session_lock
@@ -878,3 +1443,4 @@ test_opencode_watch_arm_rejects_spoofed_secondmate_markers
 test_opencode_primary_watch_plugin_rearms_after_wake
 test_opencode_watch_arm_coordinates_with_turnend_guard
 test_opencode_healthy_arm_output_does_not_suppress_guard
+printf '%s\n' 'PI_DIRECT_CONTINUITY_PROOF_OK'
diff --git a/tests/fm-session-start.test.sh b/tests/fm-session-start.test.sh
index 38ff43ee952..b7f038e71c3 100755
--- a/tests/fm-session-start.test.sh
+++ b/tests/fm-session-start.test.sh
@@ -679,7 +679,7 @@ EOF
   block_count=$(printf '%s\n' "$out" | grep -c '^SUPERVISION OPERATING INSTRUCTIONS - primary harness:')
   [ "$block_count" -eq 1 ] || fail "expected exactly one supervision block, got $block_count"
   assert_contains "$out" "SUPERVISION OPERATING INSTRUCTIONS - primary harness: pi" "pi supervision block missing"
-  assert_contains "$out" "Mode: Pi extension background wake." "pi snippet missing from session start"
+  assert_contains "$out" "Mode: Pi extension background wake with direct-exchange compaction continuity." "pi snippet missing from session start"
   assert_contains "$out" "PI_WATCH_EXTENSION: not loaded" "pi extension load diagnostic missing"
   assert_contains "$out" "restart plain pi so $root/.pi/extensions/fm-primary-turnend-guard.ts and $root/.pi/extensions/fm-primary-pi-watch.ts auto-load" "pi extension load diagnostic omits the turn-end guard extension"
 
diff --git a/tests/fm-turnend-guard.test.sh b/tests/fm-turnend-guard.test.sh
index b7867b106a1..497d25eca21 100755
--- a/tests/fm-turnend-guard.test.sh
+++ b/tests/fm-turnend-guard.test.sh
@@ -771,7 +771,6 @@ test_pi_extension_forces_followup() {
   content=$(cat "$ext")
   assert_contains "$content" 'agent_settled' "pi extension must run after one logical agent run settles"
   assert_contains "$content" 'fm-turnend-guard.sh' "pi extension must invoke the shared guard"
-  assert_contains "$content" 'sendUserMessage' "pi extension must force a follow-up turn"
   assert_contains "$content" 'deliverAs: "followUp"' "pi extension must queue the follow-up safely"
   assert_contains "$content" 'guardFollowupActive' "pi extension must carry a logical-run loop guard"
   assert_not_contains "$content" 'skipNextTurnEnd' "pi extension kept the internal-turn loop guard"
@@ -816,11 +815,12 @@ const pi = {
   on(event, handler) {
     handlers.set(event, handler);
   },
-  async sendUserMessage(message, options) {
+  sendMessage(message, options) {
     prompts += 1;
-    if (!message.includes("TURN WOULD END BLIND")) throw new Error(`unexpected prompt: ${message}`);
-    if (options?.deliverAs !== "followUp") throw new Error("guard prompt was not a follow-up");
-    await handlers.get("agent_settled")?.({ type: "agent_settled" }, {});
+    if (message.customType !== "firstmate-turnend-guard") throw new Error(`unexpected prompt type: ${message.customType}`);
+    if (!message.content.includes("TURN WOULD END BLIND")) throw new Error(`unexpected prompt: ${message.content}`);
+    if (options?.deliverAs !== "followUp" || options?.triggerTurn !== true) throw new Error("guard prompt was not a triggering follow-up");
+    void handlers.get("agent_settled")?.({ type: "agent_settled" }, {});
   },
 };
 const mod = await import(pathToFileURL(process.env.PLUGIN).href);
@@ -876,10 +876,10 @@ const pi = {
   on(event, handler) {
     handlers.set(event, handler);
   },
-  async sendUserMessage() {
+  sendMessage() {
     attempts += 1;
     if (attempts === 1) throw new Error("synthetic delivery failure");
-    await handlers.get("agent_settled")?.({ type: "agent_settled" }, {});
+    void handlers.get("agent_settled")?.({ type: "agent_settled" }, {});
   },
 };
 const mod = await import(pathToFileURL(process.env.PLUGIN).href);
diff --git a/tests/test-capabilities.tsv b/tests/test-capabilities.tsv
index b60c5a2f412..5b188f07e5f 100644
--- a/tests/test-capabilities.tsv
+++ b/tests/test-capabilities.tsv
@@ -72,6 +72,11 @@ fm-no-mistakes-reattach.test.sh	hermetic
 fm-no-mistakes-runtime.test.sh	hermetic
 fm-no-mistakes-worker.test.sh	hermetic
 fm-pi-account-home.test.sh	hermetic
+fm-pi-direct-continuity.test.sh	hermetic
+fm-pi-direct-continuity-types.test.sh	hermetic
+fm-pi-exact-delivery-association.test.sh	hermetic
+fm-pi-primary-compaction-live-e2e.test.sh	hermetic
+fm-pi-retry-continuity.test.sh	hermetic
 fm-pi-primary-live-e2e.test.sh	hermetic
 fm-pi-primary-types.test.sh	hermetic
 fm-pi-refresh.test.sh	hermetic