transcript.js

// Best-effort capture of prompt/response content for a model.usage event.
//
// OpenClaw does not deliver prompt/completion text to third-party plugins (the
// `model.usage` diagnostic carries usage/cost only; message content is private
// data handed exclusively to bundled diagnostics services). To still populate
// the Langfuse generation's input/output, we read OpenClaw's per-session
// transcript, which records each turn's `model.completed` entry with
// `finalPromptText` (input) and `assistantTexts` (output).
//
// Storage backend note: OpenClaw >=2026.7 stores sessions in a canonical
// SQLite store (agents/<agentId>/agent/openclaw-agent.sqlite) instead of the
// old flat `agents/<agentId>/sessions/<sessionId>.trajectory.jsonl` file,
// which no longer exists there (that directory now holds only session
// locks). We read transcripts through the supported, storage-neutral plugin
// SDK surface (`openclaw/plugin-sdk/session-transcript-runtime`), which
// resolves the correct backend (SQLite or legacy file) by session identity.
// A direct on-disk `.trajectory.jsonl` read remains as a fallback for older
// hosts (<2026.7) where that SDK subpath does not exist.
//
// This is intentionally best-effort: any failure (session not found, not yet
// flushed, format change) returns null and never blocks usage forwarding.

import { openSync, readSync, readFileSync, statSync, closeSync } from "node:fs";
import { homedir } from "node:os";
import path from "node:path";
import { pathToFileURL } from "node:url";

// Cap how much of a (potentially long-lived) legacy transcript file we read;
// we only need the tail (most recent turns). Only used by the legacy
// file-based fallback.
const MAX_READ_BYTES = 2_000_000;

/** Resolve OpenClaw's state dir: ctx.stateDir, then env, then ~/.openclaw. */
export function resolveStateDir(stateDir) {
  if (typeof stateDir === "string" && stateDir.length > 0) return stateDir;
  if (process.env.OPENCLAW_STATE_DIR) return process.env.OPENCLAW_STATE_DIR;
  return path.join(homedir(), ".openclaw");
}

/** Path to a session's legacy trajectory transcript (pre-2026.7 hosts only). */
export function trajectoryPath(stateDir, agentId, sessionId) {
  return path.join(
    resolveStateDir(stateDir),
    "agents",
    agentId || "main",
    "sessions",
    `${sessionId}.trajectory.jsonl`,
  );
}

/**
 * Every on-disk layout a legacy transcript is known to use, in probe order.
 *
 * The nested `agents/<agentId>/sessions/` layout is not universal: OpenClaw
 * 2026.6.x writes the file flat in the state-dir root
 * (`~/.openclaw/<sessionId>.trajectory.jsonl`). Probing only the nested path
 * there finds nothing, and because this module is best-effort the miss is
 * silent — spans just ship with empty input/output. So try both and use
 * whichever exists.
 */
export function trajectoryPathCandidates(stateDir, agentId, sessionId) {
  const root = resolveStateDir(stateDir);
  return [
    trajectoryPath(root, agentId, sessionId),
    path.join(root, `${sessionId}.trajectory.jsonl`),
  ];
}

/** Read a byte window of a file as UTF-8 text. `from: "head" | "tail"`. */
function readWindow(file, maxBytes, from) {
  const { size } = statSync(file);
  if (size <= maxBytes) return readFileSync(file, "utf8");
  const fd = openSync(file, "r");
  try {
    const buf = Buffer.allocUnsafe(maxBytes);
    const start = from === "tail" ? size - maxBytes : 0;
    const bytes = readSync(fd, buf, 0, maxBytes, start);
    return buf.toString("utf8", 0, bytes);
  } finally {
    closeSync(fd);
  }
}

// The message tool actions that post a message and the argument keys its text
// is read from (OpenClaw MESSAGE_TOOL_SEND_ACTIONS and readMessagingText).
const MESSAGE_SEND_ACTIONS = new Set(["send", "thread-reply", "sendWithEffect", "sendAttachment", "upload-file"]);
const MESSAGE_TEXT_KEYS = ["content", "message", "text", "body"];
// OpenClaw's silent reply: the token alone (edge punctuation allowed) posts nothing.
const SILENT_REPLY = /^\W*NO_REPLY\W*$/;

/** The text an assistant `toolCall` block posts through the message tool, or undefined. */
function messageToolSendText(block) {
  if (block?.type !== "toolCall" || block.name !== "message" || typeof block.id !== "string") return undefined;
  let args = block.arguments;
  if (typeof args === "string") {
    try {
      args = JSON.parse(args);
    } catch {
      return undefined;
    }
  }
  if (!args || typeof args !== "object" || !MESSAGE_SEND_ACTIONS.has(String(args.action ?? "").trim())) {
    return undefined;
  }
  for (const key of MESSAGE_TEXT_KEYS) {
    if (typeof args[key] === "string" && args[key].trim()) return args[key];
  }
  return undefined;
}

/** Extract a prompt/response from a single parsed transcript event. */
function entryIO(obj) {
  // Transcript schema v3 (OpenClaw >=2026.7.2): message content lives on
  // `message` events (`{ type: "message", message: { role, content[] } }`)
  // instead of `model.completed` entries with finalPromptText/assistantTexts.
  if (obj?.type === "message" && obj.message) {
    const m = obj.message;
    const text = contentToText(m.content);
    if (text === undefined) return {};
    if (m.role === "user") return { input: text };
    if (m.role === "assistant") return { output: text };
    return {};
  }

  const data = obj?.data;
  if (!data) return {};
  if (obj.type === "model.completed") {
    const io = {};
    if (typeof data.finalPromptText === "string") io.input = data.finalPromptText;
    if (Array.isArray(data.assistantTexts) && data.assistantTexts.length > 0) {
      io.output = data.assistantTexts.join("\n");
    }
    return io;
  }
  if (obj.type === "prompt.submitted" && typeof data.prompt === "string") {
    return { input: data.prompt };
  }
  return {};
}

function parseLine(line) {
  const trimmed = line.trim();
  if (!trimmed) return null;
  try {
    return JSON.parse(trimmed);
  } catch {
    return null; // tolerate a truncated boundary line from a windowed read
  }
}

/** Flatten a tool-result `content` value (string | array of text blocks) to text. */
function contentToText(content) {
  if (typeof content === "string") return content;
  if (Array.isArray(content)) {
    const parts = [];
    for (const block of content) {
      if (typeof block === "string") parts.push(block);
      else if (block && typeof block.text === "string") parts.push(block.text);
    }
    if (parts.length > 0) return parts.join("\n");
  }
  return undefined;
}

/**
 * Pure: fold one conversation message's tool-call / tool-result content into
 * a `{ [toolCallId]: { name, input, output, isError } }` accumulator. Tool
 * inputs live on assistant `toolCall` content blocks ({ id, name, arguments });
 * tool outputs live on dedicated `toolResult` messages ({ toolCallId,
 * toolName, content, isError }). Shared by both transcript schemas below.
 */
function applyMessageToToolIO(msg, byId) {
  if (!msg || typeof msg !== "object") return;
  // Tool outputs: dedicated toolResult messages.
  if (msg.role === "toolResult" && typeof msg.toolCallId === "string") {
    const entry = (byId[msg.toolCallId] ??= {});
    const out = contentToText(msg.content);
    if (out !== undefined) entry.output = out;
    if (typeof msg.toolName === "string") entry.name ??= msg.toolName;
    if (typeof msg.isError === "boolean") entry.isError = msg.isError;
    return;
  }
  // Tool inputs: assistant toolCall content blocks.
  const content = msg.content;
  if (!Array.isArray(content)) return;
  for (const block of content) {
    if (block?.type !== "toolCall" || typeof block.id !== "string") continue;
    const entry = (byId[block.id] ??= {});
    if (typeof block.name === "string") entry.name = block.name;
    if (block.arguments !== undefined) {
      entry.input =
        typeof block.arguments === "string"
          ? block.arguments
          : JSON.stringify(block.arguments);
    }
  }
}

/**
 * Pure: extract per-tool input/output from an array of parsed transcript
 * events, keyed by the tool call id (which matches the `toolCallId` on
 * `tool.execution.*` diagnostic events). Two transcript shapes are supported:
 *
 * - Schema v3 (OpenClaw >=2026.7.2, current): each turn's messages are logged
 *   directly as individual `{ type: "message", message }` events, so we fold
 *   every one of those in encounter order.
 * - Legacy: a single `model.completed` entry carries a cumulative
 *   `messagesSnapshot` array (every tool call + result of the run so far); we
 *   keep only the LAST one (cumulative, so last wins).
 *
 * Returns a plain object { [toolCallId]: { name, input, output, isError } } or
 * null when none found.
 */
export function extractToolIO(events) {
  const byId = {};
  let sawAny = false;
  let snapshot;
  for (const obj of events) {
    if (obj?.type === "message" && obj.message) {
      applyMessageToToolIO(obj.message, byId);
      sawAny = true;
    } else if (obj?.type === "model.completed" && Array.isArray(obj?.data?.messagesSnapshot)) {
      // Walk forward; keep the latest snapshot (cumulative, so last wins).
      snapshot = obj.data.messagesSnapshot;
    }
  }
  if (snapshot) {
    for (const msg of snapshot) applyMessageToToolIO(msg, byId);
    sawAny = true;
  }
  if (!sawAny) return null;
  return Object.keys(byId).length > 0 ? byId : null;
}

/**
 * Pure: extract turn content from an array of parsed transcript events.
 * Returns { input, output, sessionInput } where `input`/`output` are the
 * latest turn (for the generation) and `sessionInput` is the first prompt in
 * the events (for trace-level aggregation). Returns null when nothing usable
 * is found.
 *
 * When `runId` is given AND at least one event actually carries a `runId`
 * field, `input`/`output` are taken ONLY from entries matching that run — a
 * session's trajectory file accumulates every turn, and without this a
 * still-in-flight write for turn N could silently return turn N-1's (stale)
 * content instead of null (which is what should trigger a retry).
 * `sessionInput` always looks at the whole session regardless of `runId`,
 * since it means "the first-ever prompt", not this turn's prompt.
 *
 * Schema v3 `{ type: "message", message }` events (OpenClaw >=2026.7.2) carry
 * no `runId` at all (only legacy `model.completed`/`prompt.submitted` entries
 * do) — every event would then mismatch any given `runId`, and this would
 * always return null even though the transcript has perfectly good turn
 * content. So the runId scope is only enforced when the events themselves
 * demonstrate they carry that field.
 */
export function extractContent(events, runId) {
  let input;
  let output;
  let sessionInput;
  // Schema v3 has no runId to scope by, so the turn is everything after its
  // user message: a reply from before it is the previous turn's, never this
  // one's. Sends through the message tool (how a group chat is usually
  // answered) are part of what the user got.
  let sent = [];
  const pendingSends = new Map();
  const scopeByRunId = runId !== undefined && events.some((e) => e?.runId !== undefined);
  for (const obj of events) {
    if (!obj) continue;
    const io = entryIO(obj);
    if (io.input !== undefined && sessionInput === undefined) sessionInput = io.input;
    if (scopeByRunId && obj.runId !== runId) continue;
    const message = obj.type === "message" ? obj.message : undefined;
    if (message?.role === "user" && io.input !== undefined) {
      output = undefined;
      sent = [];
      pendingSends.clear();
    }
    if (message?.role === "assistant" && Array.isArray(message.content)) {
      for (const block of message.content) {
        const text = messageToolSendText(block);
        if (text !== undefined) pendingSends.set(block.id, text);
      }
    }
    if (message?.role === "toolResult" && pendingSends.has(message.toolCallId)) {
      if (message.isError !== true) sent.push(pendingSends.get(message.toolCallId));
      pendingSends.delete(message.toolCallId);
    }
    if (io.input !== undefined) input = io.input;
    if (io.output !== undefined) output = io.output;
  }
  if (sent.length > 0) {
    // OpenClaw drops a final text that only repeats a sent one, or is the
    // silent token alone; anything else is delivered after the sends.
    const shown = [...sent];
    const final = output?.trim();
    if (final && !SILENT_REPLY.test(final) && !sent.some((s) => s.trim() === final)) shown.push(output);
    output = shown.join("\n\n");
  }
  if (input === undefined && output === undefined) return null;
  const out = {};
  if (input !== undefined) out.input = input;
  if (output !== undefined) out.output = output;
  if (sessionInput !== undefined) out.sessionInput = sessionInput;
  return out;
}

// ---------------------------------------------------------------------------
// Transcript event sourcing: prefer the supported plugin SDK surface (works
// against whatever backend OpenClaw actually uses — SQLite or file), and fall
// back to a direct legacy .trajectory.jsonl read for older hosts.
// ---------------------------------------------------------------------------

const SDK_TRANSCRIPT_SUBPATH = "plugin-sdk/session-transcript-runtime.js";

/**
 * Specifiers to try, in order, when loading the host's transcript SDK.
 *
 * The bare specifier only resolves when the plugin's directory has the host on
 * its `node_modules` walk-up chain. That holds for npm-project installs (the
 * host symlinks `<stateDir>/npm/node_modules/openclaw` -> its install root),
 * but NOT for extension installs under `<stateDir>/extensions/<id>/`: that
 * symlink is not on their chain, so the import fails, the legacy
 * `.trajectory.jsonl` fallback finds nothing on SQLite-backed hosts, and every
 * span silently ships with empty input/output.
 *
 * So we also derive absolute paths from the running host process itself — the
 * plugin is loaded in-process, so the host's own `dist/` is always on disk near
 * `process.argv[1]` (e.g. `/app/dist/index.js` -> `/app/dist/plugin-sdk/...`).
 */
function* sdkTranscriptSpecifiers() {
  yield "openclaw/plugin-sdk/session-transcript-runtime";

  const seen = new Set();
  for (const start of [process.argv[1], process.cwd()]) {
    if (!start) continue;
    let dir = path.resolve(start);
    // argv[1] is a file; cwd is a directory. Walking up from either covers both.
    for (let depth = 0; depth < 6; depth++) {
      const parent = path.dirname(dir);
      if (parent === dir) break;
      dir = parent;
      for (const base of [dir, path.join(dir, "dist")]) {
        const file = path.join(base, SDK_TRANSCRIPT_SUBPATH);
        if (seen.has(file)) continue;
        seen.add(file);
        yield pathToFileURL(file).href;
      }
    }
  }
}

// Lazily & safely import the modern session-transcript-runtime SDK surface.
// Hosts on OpenClaw <2026.7 may not expose this subpath at all; failing every
// candidate just means "use the legacy file fallback below".
let sdkModulePromise;
function loadSdkTranscriptModule(logger) {
  sdkModulePromise ??= (async () => {
    const tried = [];
    for (const spec of sdkTranscriptSpecifiers()) {
      try {
        const mod = await import(spec);
        if (mod && typeof mod.readSessionTranscriptEvents === "function") return mod;
        tried.push(`${spec}: no readSessionTranscriptEvents export`);
      } catch (err) {
        tried.push(`${spec}: ${err instanceof Error ? err.message : String(err)}`);
      }
    }
    // Listing every specifier tried is the point: a silent miss here is exactly
    // how empty input/output reaches the trace with nothing to explain it.
    logger?.debug?.(`benchgen: transcript SDK unavailable (${tried.join("; ")})`);
    return null;
  })();
  return sdkModulePromise;
}

/** Read + parse a legacy per-session .trajectory.jsonl file into events.
 * Returns an array of parsed events, or null. Never throws. */
function readLegacyTrajectoryEvents(stateDir, evt, logger) {
  const sessionId = evt?.sessionId ?? evt?.sessionKey;
  if (!sessionId) return null;

  const candidates = trajectoryPathCandidates(stateDir, evt?.agentId, sessionId);
  const misses = [];
  for (const file of candidates) {
    try {
      const { size } = statSync(file);
      const text =
        size <= MAX_READ_BYTES
          ? readFileSync(file, "utf8")
          : readWindow(file, MAX_READ_BYTES, "tail");
      return text
        .split("\n")
        .map(parseLine)
        .filter((obj) => obj !== null);
    } catch (err) {
      // This layout may not exist (expected on 2026.7+ SQLite-backed hosts, and
      // for whichever of the two layouts this host does not use) or the file may
      // be mid-write; this is best-effort, so try the next one.
      misses.push(`${file}: ${err instanceof Error ? err.message : String(err)}`);
    }
  }
  // Listing every path tried is the point: a silent miss here is exactly how
  // empty input/output reaches Langfuse with nothing to explain it.
  logger?.debug?.(`benchgen: legacy transcript read failed (${misses.join("; ")})`);
  return null;
}

/**
 * Read all transcript events for an event's session. Tries the storage-
 * neutral SDK surface first (`readSessionTranscriptEvents`), which resolves
 * whichever backend OpenClaw actually uses (SQLite-backed sessions on
 * >=2026.7, or a flat file on older hosts). Falls back to a direct legacy
 * `.trajectory.jsonl` read when the SDK subpath is unavailable or the read
 * comes back empty. Returns an array of parsed events, or null. Never throws.
 */
async function readTranscriptEvents(stateDir, evt, logger) {
  const sessionId = evt?.sessionId ?? evt?.sessionKey;
  if (!sessionId) return null;

  const sdk = await loadSdkTranscriptModule(logger);
  if (sdk && typeof sdk.readSessionTranscriptEvents === "function") {
    try {
      const events = await sdk.readSessionTranscriptEvents({
        agentId: evt.agentId || "main",
        sessionKey: evt.sessionKey,
        sessionId,
      });
      if (Array.isArray(events) && events.length > 0) return events;
    } catch (err) {
      logger?.debug?.(
        `benchgen: session-transcript-runtime read failed (${
          err instanceof Error ? err.message : String(err)
        })`,
      );
    }
  }

  return readLegacyTrajectoryEvents(stateDir, evt, logger);
}

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

// The trajectory file for the CURRENT turn is not guaranteed to be fully
// flushed to disk yet at the moment run.completed/model.usage fire — writes
// lag the diagnostic event by an unbounded amount (observed up to ~2-3s on
// some hosts, likely while the runtime finishes compaction/summary bookkeeping
// for the turn). Retry with a generous total budget rather than accepting a
// one-shot empty read; this only delays when the (background) Langfuse trace
// is finalized, it never blocks the user-facing chat response.
const RESOLVE_RETRY_ATTEMPTS = 12;
const RESOLVE_RETRY_DELAY_MS = 500;

/**
 * Build a content resolver bound to a state dir. Returns an async function
 * that, given a model.usage event, reads the session transcript and returns
 * { input, output, sessionInput } or null. Never throws.
 */
export function makeContentResolver(stateDir, logger) {
  return async (evt) => {
    for (let attempt = 0; attempt < RESOLVE_RETRY_ATTEMPTS; attempt++) {
      if (attempt > 0) await sleep(RESOLVE_RETRY_DELAY_MS);
      const events = await readTranscriptEvents(stateDir, evt, logger);
      if (events) {
        const content = extractContent(events, evt?.runId);
        if (content) return content;
      }
    }
    return null;
  };
}

/**
 * Build a tool-I/O resolver bound to a state dir. Returns an async function
 * that, given an event with a session id, reads the session transcript and
 * returns the per-tool I/O map { [toolCallId]: { name, input, output, isError } }
 * or null. Never throws.
 */
export function makeToolIOResolver(stateDir, logger) {
  return async (evt) => {
    for (let attempt = 0; attempt < RESOLVE_RETRY_ATTEMPTS; attempt++) {
      if (attempt > 0) await sleep(RESOLVE_RETRY_DELAY_MS);
      const events = await readTranscriptEvents(stateDir, evt, logger);
      if (events) {
        const io = extractToolIO(events);
        if (io) return io;
      }
    }
    return null;
  };
}