fix(session): make compaction snapshots crash-safe

src/agent/turn/TurnRunner.ts

@@ -263,10 +263,11 @@ export class TurnRunner {
             await this.transcript.recordAgentStatusMessage?.(options.sessionId, options.turnId, status);
           },
           onCompactPersisted: async ({ boundary, messages: compactMessages }) => {
-            await this.transcript.recordControlBoundary?.(options.sessionId, options.turnId, boundary);
-            for (const message of compactMessages) {
-              await this.transcript.recordDurableMessage(options.sessionId, options.turnId, message);
-            }
+            if (boundary.kind !== "compact" || boundary.subtype !== "compact_boundary") return;
+            await this.transcript.recordControlBoundary?.(options.sessionId, options.turnId, {
+              ...boundary,
+              snapshot: { version: 1, messages: compactMessages },
+            });
           },
         });
         let runResult: TurnRunnerResult | undefined;

src/session/transcript/CompactSnapshot.ts

@@ -0,0 +1,58 @@
+import type { CanonicalMessage } from "../../model/index.js";
+import type { AgentTranscriptEntry } from "./TranscriptEntry.js";
+
+/** JSON on disk is untrusted even when it parses as a complete record. */
+export function readCompactSnapshot(entry: AgentTranscriptEntry): CanonicalMessage[] | undefined {
+  if (entry.type !== "control_boundary") return undefined;
+  const boundary = entry.boundary;
+  if (!isRecord(boundary) || boundary.kind !== "compact" || boundary.subtype !== "compact_boundary") {
+    return undefined;
+  }
+  const snapshot: unknown = boundary.snapshot;
+  if (!isRecord(snapshot) || snapshot.version !== 1 || !Array.isArray(snapshot.messages) ||
+      snapshot.messages.length === 0 || !snapshot.messages.every(isMessage)) {
+    return undefined;
+  }
+  return snapshot.messages;
+}
+
+function isMessage(value: unknown): value is CanonicalMessage {
+  return isRecord(value) && (value.role === "user" || value.role === "assistant") &&
+    (value.metadata === undefined || isRecord(value.metadata)) &&
+    Array.isArray(value.content) && value.content.every(isContentBlock);
+}
+
+function isContentBlock(value: unknown): boolean {
+  if (!isRecord(value)) return false;
+  switch (value.type) {
+    case "text":
+    case "thinking":
+      return typeof value.text === "string";
+    case "image":
+    case "audio":
+      return (value.source === "base64" || value.source === "url") &&
+        typeof value.data === "string" && typeof value.mimeType === "string";
+    case "pdf":
+      return value.source === "base64" && typeof value.data === "string" &&
+        value.mimeType === "application/pdf" && typeof value.bytes === "number";
+    case "tool_call":
+      return typeof value.id === "string" && typeof value.name === "string";
+    case "tool_result":
+      return typeof value.toolCallId === "string" && Array.isArray(value.content) &&
+        value.content.every((block: unknown) => isRecord(block) &&
+          ["text", "image", "pdf"].includes(String(block.type)) && isContentBlock(block));
+    case "tool_result_reference":
+    case "media_reference":
+      return typeof value.path === "string" && typeof value.originalBytes === "number" &&
+        typeof value.preview === "string" && typeof value.hasMore === "boolean" &&
+        (value.type === "tool_result_reference"
+          ? typeof value.toolCallId === "string"
+          : typeof value.mimeType === "string" && ["image", "pdf", "audio"].includes(String(value.mediaType)));
+    default:
+      return false;
+  }
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null && !Array.isArray(value);
+}

src/session/transcript/JsonlTranscriptWriter.ts

@@ -1,5 +1,5 @@
 import { randomUUID } from "node:crypto";
-import { mkdir, appendFile } from "node:fs/promises";
+import { mkdir, appendFile, open } from "node:fs/promises";
 import { basename, dirname, join, relative } from "node:path";
 import type { CanonicalMessage } from "../../model/index.js";
 import type { AgentTurnResult } from "../../agent/protocol/result.js";
@@ -44,6 +44,7 @@ export class JsonlTranscriptWriter implements AgentTranscriptWriter {
   private closed = false;
   private writeChain: Promise<void> = Promise.resolve();
   private lastEntryId: string | null = null;
+  private tailPrepared = false;
   private readonly now: () => Date;
 
   constructor(private readonly options: JsonlTranscriptWriterOptions) {
@@ -150,16 +151,42 @@ export class JsonlTranscriptWriter implements AgentTranscriptWriter {
 
   recordEntry(entry: AgentTranscriptEntry): Promise<void> {
     if (this.closed) return Promise.resolve();
+    // Capture the whole record before queued IO, so callers cannot mutate a
+    // replacement snapshot while an earlier append is still pending.
+    const line = `${JSON.stringify(entry)}\n`;
+    const flush = entry.type === "control_boundary" &&
+      entry.boundary.kind === "compact" && entry.boundary.subtype === "compact_boundary";
     this.sequence = Math.max(this.sequence, entry.sequence);
     this.lastEntryId = entry.entryId ?? this.lastEntryId;
     this.writeChain = this.writeChain.then(async () => {
       if (this.closed) return;
       await mkdir(dirname(this.options.path), { recursive: true, mode: 0o700 });
-      await appendFile(this.options.path, `${JSON.stringify(entry)}\n`, { encoding: "utf8", mode: 0o600 });
+      await this.prepareTail();
+      await appendFile(this.options.path, line, { encoding: "utf8", mode: 0o600, flush });
     });
     return this.writeChain;
   }
 
+  private async prepareTail(): Promise<void> {
+    if (this.tailPrepared) return;
+    const file = await open(this.options.path, "a+", 0o600);
+    try {
+      const { size } = await file.stat();
+      if (size > 0) {
+        const lastByte = Buffer.alloc(1);
+        await file.read(lastByte, 0, 1, size - 1);
+        if (lastByte[0] !== 0x0a) {
+          // Keep crash debris for diagnostics, but never concatenate the next
+          // entry onto it. Also preserves valid JSON lacking a final newline.
+          await file.appendFile("\n");
+        }
+      }
+      this.tailPrepared = true;
+    } finally {
+      await file.close();
+    }
+  }
+
   /**
    * C3.S1 — record the parent-side `subagent_started` reference. The full
    * directive lives in the sidechain transcript; we keep only a truncated

src/session/transcript/TranscriptEntry.ts

@@ -101,6 +101,8 @@ export type AgentControlBoundaryTranscriptEntry = AgentTranscriptEntryBase & {
         kind: "compact";
         subtype: "compact_boundary";
         compactMetadata: CompactBoundaryMetadata;
+        /** Complete replacement context committed in the same JSONL record. */
+        snapshot?: { version: 1; messages: CanonicalMessage[] };
       }
     | {
         kind: "compact";

src/session/transcript/TranscriptReplay.ts

@@ -2,6 +2,7 @@ import { cloneMessage, cloneMessages, type CanonicalMessage, type CanonicalUsage
 import type { AgentEvent } from "../../agent/protocol/events.js";
 import type { AgentPermissionDenial, AgentTurnResult } from "../../agent/protocol/result.js";
 import type { AgentTranscriptDiagnostic, AgentTranscriptEntry, SessionMetadataValue } from "./TranscriptEntry.js";
+import { readCompactSnapshot } from "./CompactSnapshot.js";
 
 export type AgentTranscriptReplayResult = {
   messages: CanonicalMessage[];
@@ -12,26 +13,21 @@ export type AgentTranscriptReplayResult = {
   diagnostics: AgentTranscriptDiagnostic[];
   /**
    * Index of the last compact_boundary entry consumed during replay. When
-   * present, only messages after this entry are kept in `messages`.
+   * present, its snapshot and messages after it are kept in `messages`.
    */
   lastCompactBoundaryIndex?: number;
-  /** Last compact boundary entry encountered (for resume relink). */
+  /** Last valid compact snapshot boundary (for resume relink). */
   lastCompactBoundary?: AgentTranscriptEntry & { type: "control_boundary" };
 };
 
 /**
- * Find the index of the last compact boundary entry. Used by resume / replay
- * to slice messages after the boundary.
+ * Only a self-contained, valid snapshot authorizes dropping prior context.
+ * Legacy boundaries cannot prove that all replacement messages were written.
  */
 export function findLastCompactBoundaryIndex(entries: AgentTranscriptEntry[]): number {
   for (let index = entries.length - 1; index >= 0; index -= 1) {
     const entry = entries[index];
-    if (
-      entry.type === "control_boundary" &&
-      entry.boundary.kind === "compact" &&
-      "subtype" in entry.boundary &&
-      entry.boundary.subtype === "compact_boundary"
-    ) {
+    if (readCompactSnapshot(entry) !== undefined) {
       return index;
     }
   }
@@ -79,6 +75,9 @@ export function replayTranscriptEntries(entries: AgentTranscriptEntry[]): AgentT
       case "assistant_message":
       case "tool_result_message":
       case "durable_message":
+        // Legacy replacement records have no completeness guarantee. Their
+        // original history is retained instead; do not duplicate a partial copy.
+        if (entry.message.metadata?.compactReplacement === true) break;
         if (!completedTurnIds.has(entry.turnId)) {
           diagnostics.push({
             code: "transcript_entry_invalid",
@@ -106,12 +105,18 @@ export function replayTranscriptEntries(entries: AgentTranscriptEntry[]): AgentT
         }
         break;
       case "control_boundary":
-        if (
-          entry.boundary.kind === "compact" &&
-          "subtype" in entry.boundary &&
-          entry.boundary.subtype === "compact_boundary"
-        ) {
+        if (index === lastBoundaryIndex) {
           lastCompactBoundary = entry;
+          const snapshot = readCompactSnapshot(entry)!;
+          messages.push(...cloneMessages(snapshot));
+          events.push(...snapshot.map((message) => projectMessageEvent(entry.sessionId, entry.turnId, message)));
+        } else if (entry.boundary?.kind === "compact" && entry.boundary.subtype === "compact_boundary" &&
+                   readCompactSnapshot(entry) === undefined) {
+          diagnostics.push({
+            code: "transcript_entry_invalid",
+            severity: "warning",
+            message: "Ignoring compact boundary without a valid complete snapshot; retaining prior context.",
+          });
         }
         break;
       case "session_metadata":

src/web/server/forkSession.ts

@@ -13,6 +13,7 @@ import { dirname, isAbsolute, join, relative, resolve } from "node:path";
 import { platform } from "node:process";
 import type { CanonicalContentBlock, CanonicalMessage } from "../../model/index.js";
 import { getPilotProjectChatDir } from "../../pilot/index.js";
+import { readCompactSnapshot } from "../../session/transcript/CompactSnapshot.js";
 import { readTranscript } from "../../session/transcript/TranscriptReader.js";
 import {
   sanitizeSessionIdForPath,
@@ -225,63 +226,59 @@ function markMessageAsForkCarryover(
   };
 }
 
-function retargetTranscriptEntryAuxiliaryPaths(
+function mapTranscriptEntryMessages(
   entry: AgentTranscriptEntry,
-  sourceSessionDir: string,
-  targetSessionDir: string,
+  transform: (message: CanonicalMessage) => CanonicalMessage,
 ): AgentTranscriptEntry {
   if (entry.type === "accepted_input") {
-    return {
-      ...entry,
-      messages: entry.messages.map((message) => ({
-        ...message,
-        content: message.content.map((block) =>
-          retargetContentBlock(block, sourceSessionDir, targetSessionDir),
-        ),
-      })),
-    };
+    return { ...entry, messages: entry.messages.map(transform) };
   }
   if (
     entry.type === "assistant_message" ||
     entry.type === "tool_result_message" ||
     entry.type === "durable_message"
   ) {
-    return {
-      ...entry,
-      message: {
-        ...entry.message,
-        content: entry.message.content.map((block) =>
-          retargetContentBlock(block, sourceSessionDir, targetSessionDir),
-        ),
-      },
-    };
+    return { ...entry, message: transform(entry.message) };
+  }
+  if (
+    entry.type === "control_boundary" &&
+    entry.boundary.kind === "compact" &&
+    entry.boundary.subtype === "compact_boundary"
+  ) {
+    const messages = readCompactSnapshot(entry);
+    if (messages) {
+      return {
+        ...entry,
+        boundary: {
+          ...entry.boundary,
+          snapshot: { version: 1, messages: messages.map(transform) },
+        },
+      };
+    }
   }
   return entry;
 }
 
+function retargetTranscriptEntryAuxiliaryPaths(
+  entry: AgentTranscriptEntry,
+  sourceSessionDir: string,
+  targetSessionDir: string,
+): AgentTranscriptEntry {
+  return mapTranscriptEntryMessages(entry, (message) => ({
+    ...message,
+    content: message.content.map((block) =>
+      retargetContentBlock(block, sourceSessionDir, targetSessionDir),
+    ),
+  }));
+}
+
 function markTranscriptEntryAsForkCarryover(
   entry: AgentTranscriptEntry,
   sourceSessionId: string,
 ): AgentTranscriptEntry {
-  if (entry.type === "accepted_input") {
-    return {
-      ...entry,
-      messages: entry.messages.map((message) =>
-        markMessageAsForkCarryover(message, sourceSessionId, entry.turnId),
-      ),
-    };
-  }
-  if (
-    entry.type === "assistant_message" ||
-    entry.type === "tool_result_message" ||
-    entry.type === "durable_message"
-  ) {
-    return {
-      ...entry,
-      message: markMessageAsForkCarryover(entry.message, sourceSessionId, entry.turnId),
-    };
-  }
-  return entry;
+  return mapTranscriptEntryMessages(entry, (message) =>
+    markMessageAsForkCarryover(message, sourceSessionId, entry.turnId),
+  );
 }
 
 function retargetAcceptedInputEntry(
@@ -327,7 +324,8 @@ function retargetEntriesToSession(
     if (
       entry.type === "assistant_message" ||
       entry.type === "tool_result_message" ||
-      entry.type === "durable_message"
+      entry.type === "durable_message" ||
+      entry.type === "control_boundary"
     ) {
       const retargeted = {
         ...retargetTranscriptEntryAuxiliaryPaths(