openclaw/src/auto-reply/reply/session-transcript-replay.ts

import fs from "node:fs";
import fsp from "node:fs/promises";
import path from "node:path";
import { CURRENT_SESSION_VERSION } from "../../config/sessions/version.js";

/** Tail kept so DM continuity survives silent session rotations. */
export const DEFAULT_REPLAY_MAX_MESSAGES = 6;

type SessionRecord = {
  type?: unknown;
  id?: unknown;
  parentId?: unknown;
  timestamp?: unknown;
  message?: { role?: unknown };
};
type KeptRecord = { role: "user" | "assistant"; line: string };

function isValidReplayTimestamp(value: unknown): boolean {
  if (typeof value === "number") {
    return Number.isFinite(value);
  }
  return typeof value === "string" && value.trim().length > 0;
}

function replayableRole(record: SessionRecord | null): "user" | "assistant" | undefined {
  if (
    !record ||
    record.type !== "message" ||
    typeof record.id !== "string" ||
    record.id.trim().length === 0 ||
    !isValidReplayTimestamp(record.timestamp) ||
    !(
      record.parentId === null ||
      record.parentId === undefined ||
      typeof record.parentId === "string"
    )
  ) {
    return undefined;
  }
  const role = record.message?.role;
  return role === "user" || role === "assistant" ? role : undefined;
}

/**
 * Copy the tail of user/assistant JSONL records from a prior transcript into a
 * freshly-rotated one. Tool, system, and compaction records are skipped so
 * replay cannot reshape tool/role ordering, and the tail is aligned and
 * coalesced into alternating user/assistant turns so role-ordering resets
 * cannot immediately recur. Uses async I/O so long transcripts do not block
 * the event loop. Returns 0 on any error.
 */
export async function replayRecentUserAssistantMessages(params: {
  sourceTranscript?: string;
  targetTranscript: string;
  newSessionId: string;
  maxMessages?: number;
}): Promise<number> {
  const max = Math.max(0, params.maxMessages ?? DEFAULT_REPLAY_MAX_MESSAGES);
  const src = params.sourceTranscript;
  if (max === 0 || !src || !fs.existsSync(src)) {
    return 0;
  }
  try {
    const kept: KeptRecord[] = [];
    for (const line of (await fsp.readFile(src, "utf-8")).split(/\r?\n/)) {
      if (!line.trim()) {
        continue;
      }
      try {
        const role = replayableRole(JSON.parse(line) as SessionRecord | null);
        if (role) {
          kept.push({ role, line });
        }
      } catch {
        // Skip malformed lines.
      }
    }
    if (kept.length === 0) {
      return 0;
    }
    let startIdx = Math.max(0, kept.length - max);
    while (startIdx < kept.length && kept[startIdx].role === "assistant") {
      startIdx += 1;
    }
    if (startIdx === kept.length) {
      // Retained window is assistant-only; replaying would re-create the same
      // role-ordering hazard this reset path is recovering from.
      return 0;
    }
    const tail = coalesceAlternatingReplayTail(kept.slice(startIdx)).map((entry) => entry.line);
    if (!fs.existsSync(params.targetTranscript)) {
      await fsp.mkdir(path.dirname(params.targetTranscript), { recursive: true });
      const header = JSON.stringify({
        type: "session",
        version: CURRENT_SESSION_VERSION,
        id: params.newSessionId,
        timestamp: new Date().toISOString(),
        cwd: process.cwd(),
      });
      await fsp.writeFile(params.targetTranscript, `${header}\n`, {
        encoding: "utf-8",
        mode: 0o600,
      });
    }
    await fsp.appendFile(params.targetTranscript, `${tail.join("\n")}\n`, "utf-8");
    return tail.length;
  } catch {
    return 0;
  }
}

// Keep the newest record from each same-role run, preserving original JSONL bytes
// for replay while ensuring strict provider alternation.
function coalesceAlternatingReplayTail(entries: KeptRecord[]): KeptRecord[] {
  const tail: KeptRecord[] = [];
  for (const entry of entries) {
    const lastIdx = tail.length - 1;
    if (lastIdx >= 0 && tail[lastIdx]?.role === entry.role) {
      tail[lastIdx] = entry;
      continue;
    }
    tail.push(entry);
  }
  return tail;
}