packages/coding-agent/src/core/session-manager.ts

import { createHash, randomUUID } from "node:crypto";
import {
  appendFileSync,
  copyFileSync,
  existsSync,
  mkdirSync,
  readFileSync,
  renameSync,
  rmSync,
  writeFileSync,
} from "node:fs";
import { homedir } from "node:os";
import { dirname, join, resolve } from "node:path";
import type { ThinkingLevel } from "@zpi-dev/agent";
import type { JsonValue, Message, SystemMessage, Usage } from "@zpi-dev/ai";
import { assertSupportedOptions, getCurrentTools, isJsonObject } from "@zpi-dev/ai";
import { applyMicrocompactBoundary, microcompactClearedIds } from "./microcompact.ts";
import {
  isSessionEntry,
  maxSessionEntryBytes,
  parseSessionTranscript,
  validateMessage,
} from "./record-validation.ts";
import { recoverStreamedToolJournal } from "./streaming-tool-journal.ts";
import type {
  SessionCurrentState,
  TranscriptSnapshot,
  TranscriptState,
  TranscriptStore,
} from "./transcript-store.ts";
import { getMessageProvenance, setMessageProvenance } from "./transcript-store.ts";
export interface SessionHeader {
  type: "session";
  version: 1 | 2;
  format: "zpi";
  id: string;
  timestamp: string;
  cwd: string;
}
interface EntryBase {
  id: string;
  parentId: string | null;
  timestamp: string;
}
export type SessionEntry = EntryBase &
  (
    | { type: "message"; message: Message }
    | { type: "model_change"; provider: string; modelId: string }
    | { type: "thinking_level_change"; thinkingLevel: ThinkingLevel }
    | { type: "session_info"; name: string }
    | { type: "custom"; customType: string; data?: JsonValue }
    | { type: "microcompact"; data: import("./microcompact.ts").MicrocompactBoundary }
    | {
        type: "compaction";
        summary: string;
        firstKeptEntryId: string;
        tokensBefore?: number;
        details?: { readFiles: string[]; modifiedFiles: string[] };
        usage?: Usage;
        systemMessage?: SystemMessage;
        contextMessage?: SystemMessage;
      }
  );
export interface SessionContext {
  messages: Message[];
  thinkingLevel: ThinkingLevel;
  model: { provider: string; modelId: string } | null;
  lastThinkingLevel: ThinkingLevel;
}
export type SessionEntryInput = SessionEntry extends infer Entry
  ? Entry extends SessionEntry
    ? Omit<Entry, keyof EntryBase>
    : never
  : never;
export class SessionManager {
  private header: SessionHeader;
  private store?: TranscriptStore;
  private currentState?: SessionCurrentState;
  private contextCache?: { generation: string; endSeq: number; context: SessionContext };
  private entries: SessionEntry[] = [];
  private file?: string;
  private storageError?: Error;
  private appendQueue: Promise<unknown> = Promise.resolve();
  private lastEntry?: SessionEntry;
  private appendListeners = new Set<(entry: SessionEntry) => void>();
  subscribeEntries(listener: (entry: SessionEntry) => void): () => void {
    this.appendListeners.add(listener);
    return () => {
      this.appendListeners.delete(listener);
    };
  }
  private constructor(cwd: string, id: string = randomUUID()) {
    this.header = {
      type: "session",
      version: 1,
      format: "zpi",
      id,
      timestamp: new Date().toISOString(),
      cwd: resolve(cwd),
    };
  }
  /** Open an already validated indexed store without reading historical message bodies. */
  static fromStore(store: TranscriptStore, currentState?: SessionCurrentState): SessionManager {
    const header = store.getHeader();
    const manager = new SessionManager(header.cwd, header.id);
    manager.header = header;
    manager.file = store.getFile();
    manager.store = store;
    manager.currentState = currentState;
    recoverStreamedToolJournal(manager);
    return manager;
  }
  static async openAsync(
    store: TranscriptStore,
    currentState?: SessionCurrentState,
  ): Promise<SessionManager> {
    return SessionManager.fromStore(store, currentState);
  }
  hasCurrentState(): boolean {
    return this.currentState !== undefined;
  }
  getCurrentState(): Pick<TranscriptState, "model" | "thinkingLevel" | "lastThinkingLevel" | "name"> {
    if (this.store && !this.currentState) throw new Error("Indexed session requires current-state reader");
    return this.currentState?.get() ?? this.getStateAt();
  }
  getStateAt(snapshot?: TranscriptSnapshot): TranscriptState {
    if (this.store) return this.store.getStateAt(snapshot ?? this.store.snapshot());
    const context = buildSessionContext(this.entries);
    return {
      model: context.model,
      thinkingLevel: context.thinkingLevel,
      lastThinkingLevel: context.lastThinkingLevel,
      hasThinkingEntry: this.entries.some((e) => e.type === "thinking_level_change"),
      lastMessageSeq: this.entries.findLastIndex((e) => e.type === "message") + 1,
      lastCompactionSeq: this.entries.findLastIndex((e) => e.type === "compaction") + 1,
      name: this.entries.findLast((e) => e.type === "session_info")?.name,
    };
  }
  *iterateEntries(): Iterable<SessionEntry> {
    if (!this.store) {
      yield* this.entries;
      return;
    }
    const snapshot = this.store.snapshot();
    for (const { entry } of this.store.readRange(1, snapshot.endSeq, snapshot)) yield entry;
  }
  readEntryById(id: string): SessionEntry | undefined {
    return this.store
      ? this.store.readEntryById(id, this.store.snapshot())?.entry
      : this.entries.find((e) => e.id === id);
  }
  *unresolvedToolJournals(): Iterable<SessionEntry[]> {
    if (this.store) yield* this.store.readUnresolvedToolJournals(this.store.snapshot());
    else yield this.entries;
  }
  /** Only effective retained entries, suitable for compaction and current usage. */
  getContextEntries(): SessionEntry[] {
    if (!this.store) return this.entries.slice();
    const snapshot = this.store.snapshot();
    const plan = this.store.getContextPlan(snapshot);
    const result: SessionEntry[] = [];
    for (const { entry } of this.store.readRange(plan.firstKeptSeq, snapshot.endSeq, snapshot, [
      "message",
      "compaction",
      "microcompact",
      "model_change",
    ]))
      result.push(entry);
    return result;
  }
  precedesCompaction(message: Message): boolean {
    const source = getMessageProvenance(message);
    return source !== undefined && source.seq < this.getStateAt().lastCompactionSeq;
  }
  async materializeContext(snapshot?: TranscriptSnapshot): Promise<SessionContext> {
    return this.buildSessionContext(snapshot);
  }
  static create(
    cwd: string,
    sessionDir = join(homedir(), ".zpi", "agent", "sessions"),
    options: { id?: string } = {},
  ): SessionManager {
    assertSupportedOptions(options, ["id"], "SessionManager.create");
    const manager = new SessionManager(cwd, options.id);
    if (!/^[a-zA-Z0-9_-]+$/.test(manager.header.id)) throw new Error("Invalid session id");
    manager.file = join(sessionDir, `${manager.header.id}.jsonl`);
    mkdirSync(sessionDir, { recursive: true });
    writeFileSync(manager.file, `${JSON.stringify(manager.header)}\n`, { flag: "wx", mode: 0o600 });
    return manager;
  }
  static inMemory(
    cwd = process.cwd(),
    options: { id?: string } = {},
    entries: SessionEntry[] = [],
  ): SessionManager {
    assertSupportedOptions(options, ["id"], "SessionManager.inMemory");
    const m = new SessionManager(cwd, options.id);
    if (entries.some((e) => !isSessionEntry(e))) throw new Error("Invalid in-memory session entries");
    m.entries = structuredClone(entries);
    return m;
  }
  static open(path: string, _sessionDir?: string, cwdOverride?: string): SessionManager {
    const raw = readFileSync(path, "utf8");
    const { header, entries, repairTo } = parseSessionTranscript(raw, { repairIncompleteTail: true });
    if (repairTo !== undefined) {
      copyFileSync(path, `${path}.corrupt-${Date.now()}-${randomUUID()}`);
      writeFileSync(path, repairTo, { mode: 0o600 });
    } else if (!raw.endsWith("\n")) appendFileSync(path, "\n");
    const m = new SessionManager(cwdOverride ?? header.cwd, header.id);
    m.header = { ...header, cwd: cwdOverride ?? header.cwd };
    m.file = path;
    m.entries = entries;
    recoverStreamedToolJournal(m);
    return m;
  }
  getSessionId(): string {
    return this.header.id;
  }
  getHeader(): SessionHeader {
    return { ...this.header };
  }
  getSessionFile(): string | undefined {
    return this.store ? this.store.getFile() : this.file;
  }
  getCwd(): string {
    return this.header.cwd;
  }
  /** Move a draft into its worktree before any model messages are committed. */
  setCwd(cwd: string): void {
    if (this.storageError) throw this.storageError;
    if (this.getStateAt().lastMessageSeq || this.getStateAt().lastCompactionSeq)
      throw new Error("Cannot move a session with messages");
    const header = { ...this.header, cwd: resolve(cwd) };
    if (this.store) this.store.replace(header, this.iterateEntries());
    else this.writeTranscript(header, this.entries);
    this.header = header;
  }
  getEntries(): SessionEntry[] {
    if (this.store) throw new Error("Indexed sessions require iterateEntries or getContextEntries");
    return this.entries.slice();
  }
  getLastEntry(): SessionEntry | undefined {
    return this.store
      ? (this.lastEntry ?? this.store.getTail(this.store.snapshot())?.entry)
      : this.entries.at(-1);
  }
  /** Serialize durable appends, including concurrently finishing tools, before notifying readers. */
  appendEntryAsync(fields: SessionEntryInput): Promise<string> {
    return this.enqueueWrite(() => this.persistEntry(fields));
  }
  private enqueueWrite<T>(write: () => T | Promise<T>): Promise<T> {
    const operation = this.appendQueue.then(() => {
      if (this.storageError) throw this.storageError;
      return write();
    });
    this.appendQueue = operation.catch(() => {});
    return operation;
  }
  private async persistEntry(fields: SessionEntryInput): Promise<string> {
    if (!this.store?.appendAsync) return this.append(fields);
    const entry = this.createEntry(fields);
    try {
      await this.store.appendAsync(entry);
    } catch (error) {
      this.storageError = new Error(`storage: ${error instanceof Error ? error.message : String(error)}`);
      throw this.storageError;
    }
    this.publishEntry(entry);
    return entry.id;
  }
  async flushAsync(): Promise<void> {
    await this.appendQueue;
    if (this.storageError) throw this.storageError;
  }
  appendMessageAsync(message: Message): Promise<string> {
    return this.appendEntryAsync({ type: "message", message });
  }
  appendCustomEntryAsync(customType: string, data?: JsonValue): Promise<string> {
    return this.appendEntryAsync({ type: "custom", customType, data });
  }
  setCwdAsync(cwd: string): Promise<void> {
    return this.enqueueWrite(async () => {
      if (!this.store?.replaceAsync) return this.setCwd(cwd);
      if (this.getStateAt().lastMessageSeq || this.getStateAt().lastCompactionSeq)
        throw new Error("Cannot move a session with messages");
      const header = { ...this.header, cwd: resolve(cwd) };
      await this.store.replaceAsync(header, this.iterateEntries());
      this.header = header;
      this.lastEntry = undefined;
      this.contextCache = undefined;
    });
  }
  replaceEntriesAsync(source: Iterable<SessionEntry>): Promise<void> {
    return this.enqueueWrite(async () => {
      if (!this.store?.replaceAsync) return this.replaceEntries(source);
      await this.store.replaceAsync(this.header, source, { remapIds: true });
      this.lastEntry = undefined;
      this.contextCache = undefined;
    });
  }
  /** Atomically replace a transcript after a stable fork or conversation rewind. */
  replaceEntries(source: Iterable<SessionEntry>): void {
    if (this.storageError) throw this.storageError;
    if (this.store) {
      this.store.replace(this.header, remapSessionEntries(source));
      this.contextCache = undefined;
      this.lastEntry = undefined;
      return;
    }
    const materialized: SessionEntry[] = Array.isArray(source) ? source : Array.from(source);
    const ids = new Map(materialized.map((entry) => [entry.id, randomUUID()]));
    if (ids.size !== materialized.length) throw new Error("Duplicate session entry IDs");
    let parentId: string | null = null;
    const entries = materialized.map((entry) => {
      const id = ids.get(entry.id);
      if (!id) throw new Error("Invalid session entry identity");
      const next = { ...structuredClone(entry), id, parentId };
      if (next.type === "compaction") {
        const boundary = ids.get(next.firstKeptEntryId);
        if (!boundary) throw new Error("Invalid compaction boundary");
        next.firstKeptEntryId = boundary;
      }
      if (
        next.type === "custom" &&
        next.customType === "zpi.streaming_tools" &&
        isJsonObject(next.data) &&
        typeof next.data.anchorId === "string"
      ) {
        const anchor = ids.get(next.data.anchorId);
        if (!anchor) throw new Error("Invalid tool journal anchor");
        next.data.anchorId = anchor;
      }
      if (!isSessionEntry(next) || Buffer.byteLength(JSON.stringify(next)) > maxSessionEntryBytes)
        throw new Error("Invalid session entry");
      parentId = next.id;
      return next;
    });
    this.writeTranscript(this.header, entries);
    this.entries = entries;
    this.contextCache = undefined;
  }
  private writeTranscript(header: SessionHeader, entries: SessionEntry[]): void {
    if (!this.file) return;
    const temporary = `${this.file}.${randomUUID()}.tmp`;
    try {
      writeFileSync(temporary, `${[header, ...entries].map((entry) => JSON.stringify(entry)).join("\n")}\n`, {
        mode: 0o600,
      });
      renameSync(temporary, this.file);
    } finally {
      rmSync(temporary, { force: true });
    }
  }
  getMicrocompactClearedIds(): string[] {
    if (!this.store) return this.entries.flatMap(microcompactClearedIds);
    const view = this.store.snapshot();
    return Array.from(this.store.readRange(1, view.endSeq, view, ["microcompact"])).flatMap(({ entry }) =>
      microcompactClearedIds(entry),
    );
  }
  buildSessionContext(snapshot?: TranscriptSnapshot): SessionContext {
    if (!this.store) return buildSessionContext(this.entries);
    const view = snapshot ?? this.store.snapshot();
    if (this.contextCache?.generation === view.generation && this.contextCache.endSeq === view.endSeq)
      return this.contextCache.context;
    const state = this.store.getStateAt(view);
    const plan = this.store.getContextPlan(view);
    const messages: Message[] = [];
    const compaction = plan.compaction;
    let start = 1;
    if (compaction) {
      const entry = compaction.entry;
      if (entry.type !== "compaction" || !entry.systemMessage)
        throw new Error("Invalid compaction: system snapshot is required");
      const boundary = this.store.readEntryById(entry.firstKeptEntryId, view);
      if (!boundary || boundary.seq !== plan.firstKeptSeq || boundary.seq >= compaction.seq)
        throw new Error("Invalid compaction boundary");
      if (boundary.entry.type === "message" && boundary.entry.message.role === "toolResult")
        throw new Error("Compaction boundary splits a tool turn");
      const summary: Message = {
        role: "user",
        synthetic: true,
        content: `Summary of earlier conversation:\n${entry.summary}`,
        timestamp: Date.parse(entry.timestamp),
      };
      setMessageProvenance(entry.systemMessage, {
        entryId: entry.id,
        seq: compaction.seq,
        kind: "compaction-system",
      });
      setMessageProvenance(summary, { entryId: entry.id, seq: compaction.seq, kind: "compaction-summary" });
      messages.push(entry.systemMessage, summary);
      start = boundary.seq;
    }
    const retainedCalls = new Set<string>();
    let contextRestored = false;
    for (const { seq, entry } of this.store.readRange(start, view.endSeq, view, [
      "message",
      "microcompact",
    ])) {
      if (compaction && seq >= compaction.seq && !contextRestored) {
        if (compaction.entry.type === "compaction" && compaction.entry.contextMessage)
          messages.push(compaction.entry.contextMessage);
        contextRestored = true;
      }
      if (entry.type !== "message") {
        const projected = applyMicrocompactBoundary(messages, microcompactClearedIds(entry));
        messages.splice(0, messages.length, ...projected);
        continue;
      }
      if (compaction && seq < compaction.seq && entry.message.role === "system") continue;
      if (compaction) {
        if (entry.message.role === "assistant") {
          retainedCalls.clear();
          for (const part of entry.message.content) if (part.type === "toolCall") retainedCalls.add(part.id);
        } else if (entry.message.role === "toolResult" && !retainedCalls.delete(entry.message.toolCallId)) {
          throw new Error("Compaction boundary splits a tool turn: missing retained declaration");
        }
      }
      setMessageProvenance(entry.message, { entryId: entry.id, seq, kind: "message" });
      messages.push(entry.message);
    }
    if (
      compaction &&
      !contextRestored &&
      compaction.entry.type === "compaction" &&
      compaction.entry.contextMessage
    )
      messages.push(compaction.entry.contextMessage);
    const context = {
      messages,
      model: state.model,
      thinkingLevel: state.thinkingLevel,
      lastThinkingLevel: state.lastThinkingLevel,
    };
    this.contextCache = { ...view, context };
    return context;
  }
  appendMessage(message: Message): string {
    validateMessage(message);
    return this.append({ type: "message", message });
  }
  appendModelChange(provider: string, modelId: string): string {
    if (this.store) {
      if (!this.currentState) throw new Error("Indexed session requires current-state writer");
      this.currentState.setModel(provider, modelId);
      return "";
    }
    return this.append({ type: "model_change", provider, modelId });
  }
  appendThinkingLevelChange(thinkingLevel: ThinkingLevel): string {
    if (this.store) {
      if (!this.currentState) throw new Error("Indexed session requires current-state writer");
      this.currentState.setThinkingLevel(thinkingLevel);
      return "";
    }
    return this.append({ type: "thinking_level_change", thinkingLevel });
  }
  appendSessionInfo(name: string): string {
    if (this.store) {
      if (!this.currentState) throw new Error("Indexed session requires current-state writer");
      this.currentState.setName(name);
      return "";
    }
    return this.append({ type: "session_info", name });
  }
  appendMicrocompact(data: import("./microcompact.ts").MicrocompactBoundary): string {
    return this.append({ type: "microcompact", data });
  }
  appendCustomEntry(customType: string, data?: JsonValue): string {
    return this.append({ type: "custom", customType, data });
  }
  appendCompaction(
    summary: string,
    firstKeptEntryId: string,
    tokensBefore?: number,
    details?: { readFiles: string[]; modifiedFiles: string[] },
    usage?: Usage,
  ): string {
    return this.append(this.compactionInput(summary, firstKeptEntryId, tokensBefore, details, usage));
  }
  appendCompactionAsync(...args: Parameters<SessionManager["appendCompaction"]>): Promise<string> {
    return this.enqueueWrite(() => this.persistEntry(this.compactionInput(...args)));
  }
  private compactionInput(
    summary: string,
    firstKeptEntryId: string,
    tokensBefore?: number,
    details?: { readFiles: string[]; modifiedFiles: string[] },
    usage?: Usage,
  ): Extract<SessionEntryInput, { type: "compaction" }> {
    if (!summary.trim() || !this.readEntryById(firstKeptEntryId))
      throw new Error("Invalid compaction boundary or summary");
    const current = this.buildSessionContext().messages;
    const snapshot = systemSnapshot(current, Date.now());
    return {
      type: "compaction",
      summary,
      firstKeptEntryId,
      tokensBefore,
      details,
      usage,
      ...snapshot,
    };
  }
  private createEntry(fields: SessionEntryInput): SessionEntry {
    return {
      ...fields,
      id: randomUUID(),
      parentId: this.getLastEntry()?.id ?? null,
      timestamp: new Date().toISOString(),
    } as SessionEntry;
  }
  private publishEntry(entry: SessionEntry): void {
    if (this.store) this.lastEntry = entry;
    else this.entries.push(entry);
    this.contextCache = undefined;
    for (const listener of this.appendListeners) listener(structuredClone(entry));
  }
  private append(fields: SessionEntryInput): string {
    if (this.storageError) throw this.storageError;
    const entry = this.createEntry(fields);
    if (!isSessionEntry(entry)) throw new Error("Invalid session entry");
    const line = JSON.stringify(entry);
    if (Buffer.byteLength(line) > maxSessionEntryBytes) throw new Error("Session entry exceeds 8 MiB");
    try {
      if (this.store) this.store.append(entry);
      else if (this.file) {
        if (!existsSync(dirname(this.file))) throw new Error("Session directory no longer exists");
        appendFileSync(this.file, `${line}\n`, { mode: 0o600 });
      }
    } catch (e) {
      this.storageError = new Error(`storage: ${e instanceof Error ? e.message : String(e)}`);
      throw this.storageError;
    }
    this.publishEntry(entry);
    return entry.id;
  }
}

export type CompactionEntry = Extract<SessionEntry, { type: "compaction" }>;
/** Entries are append-only. Like Pi, context projection shares immutable message records. */
export function sessionEntryToContextMessages(entry: SessionEntry): Message[] {
  return entry.type === "message" ? [entry.message] : [];
}
function systemSnapshot(
  messages: Message[],
  timestamp: number,
): {
  systemMessage: SystemMessage;
  contextMessage?: SystemMessage;
} {
  const initial = messages[0]?.role === "system" ? messages[0] : undefined;
  const systemMessage: SystemMessage = {
    ...initial,
    role: "system",
    content: initial?.content ?? "",
    toolsAdded: getCurrentTools(messages),
    toolsRemoved: undefined,
    timestamp,
  };
  const sections: Record<string, string | null> = Object.create(null);
  const content: string[] = [];
  for (const message of messages.slice(initial ? 1 : 0)) {
    if (message.role !== "system") continue;
    const text =
      typeof message.content === "string" ? message.content : message.content.map((c) => c.text).join("\n");
    if (text) content.push(text);
    // Keep tombstones: summaries and retained text must not revive withdrawn blocks.
    Object.assign(sections, message.sections);
  }
  return {
    systemMessage,
    ...(content.length || Object.keys(sections).length
      ? {
          contextMessage: { role: "system", content: content.join("\n\n"), sections, timestamp },
        }
      : {}),
  };
}
export function buildSessionContext(entries: SessionEntry[]): SessionContext {
  const context: SessionContext = {
    messages: [],
    thinkingLevel: "off",
    model: null,
    lastThinkingLevel: "medium",
  };
  let compactionIndex = -1;
  for (let i = 0; i < entries.length; i++) {
    const entry = entries[i];
    if (entry.type === "message")
      setMessageProvenance(entry.message, { entryId: entry.id, seq: i + 1, kind: "message" });
    if (entry.type === "compaction") compactionIndex = i;
    if (entry.type === "model_change") context.model = { provider: entry.provider, modelId: entry.modelId };
    else if (entry.type === "message" && entry.message.role === "assistant")
      context.model = { provider: entry.message.provider, modelId: entry.message.model };
    else if (entry.type === "thinking_level_change") {
      context.thinkingLevel = entry.thinkingLevel;
      if (entry.thinkingLevel !== "off") context.lastThinkingLevel = entry.thinkingLevel;
    }
  }
  const compaction = entries[compactionIndex];
  if (compaction?.type !== "compaction") {
    context.messages = replayMicrocompact(entries.flatMap(sessionEntryToContextMessages), entries);
    return context;
  }
  const keepIndex = entries.findIndex((e) => e.id === compaction.firstKeptEntryId);
  if (keepIndex < 0 || keepIndex >= compactionIndex) throw new Error("Invalid compaction boundary");
  const timestamp = Date.parse(compaction.timestamp);
  const snapshot = compaction.systemMessage
    ? { systemMessage: compaction.systemMessage, contextMessage: compaction.contextMessage }
    : systemSnapshot(entries.slice(0, compactionIndex).flatMap(sessionEntryToContextMessages), timestamp);
  context.messages = [
    snapshot.systemMessage,
    {
      role: "user",
      synthetic: true,
      content: `Summary of earlier conversation:\n${compaction.summary}`,
      timestamp,
    },
    ...entries
      .slice(keepIndex, compactionIndex)
      .flatMap(sessionEntryToContextMessages)
      .filter((m) => m.role !== "system"),
    ...(snapshot.contextMessage ? [snapshot.contextMessage] : []),
    ...entries.slice(compactionIndex + 1).flatMap(sessionEntryToContextMessages),
  ];
  context.messages = replayMicrocompact(context.messages, entries);
  return context;
}

function replayMicrocompact(messages: Message[], entries: SessionEntry[]): Message[] {
  for (const [index, entry] of entries.entries()) {
    if (entry.type !== "microcompact") continue;
    // A persisted boundary can only clear results that existed when it was written.
    const preceding = messages.filter((message) => (getMessageProvenance(message)?.seq ?? Infinity) <= index);
    const projected = applyMicrocompactBoundary(preceding, microcompactClearedIds(entry));
    const replacements = new Map(preceding.map((message, i) => [message, projected[i]]));
    messages = messages.map((message) => replacements.get(message) ?? message);
  }
  return messages;
}

/** Stream ID remapping without retaining the whole transcript. */
export function* remapSessionEntries(source: Iterable<SessionEntry>): Iterable<SessionEntry> {
  const nonce = randomUUID();
  const remap = (id: string) =>
    createHash("sha256").update(nonce).update("\0").update(id).digest("hex").slice(0, 32);
  let parentId: string | null = null;
  for (const entry of source) {
    const next: SessionEntry = { ...structuredClone(entry), id: remap(entry.id), parentId };
    if (next.type === "compaction") next.firstKeptEntryId = remap(next.firstKeptEntryId);
    if (
      next.type === "custom" &&
      next.customType === "zpi.streaming_tools" &&
      isJsonObject(next.data) &&
      typeof next.data.anchorId === "string"
    )
      next.data.anchorId = remap(next.data.anchorId);
    if (!isSessionEntry(next) || Buffer.byteLength(JSON.stringify(next)) > maxSessionEntryBytes)
      throw new Error("Invalid session entry");
    parentId = next.id;
    yield next;
  }
}