fix(pi): preserve direct reply continuity across compaction

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

@@ -4,16 +4,171 @@ 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 = {
   ok: boolean;
   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,15 +308,15 @@ 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.
       }
     });
     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);

.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;

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.