diff --git a/src/trajectory/export.test.ts b/src/trajectory/export.test.ts index 60dba9df8462..3d6ef0efb2e5 100644 --- a/src/trajectory/export.test.ts +++ b/src/trajectory/export.test.ts @@ -1768,6 +1768,7 @@ describe("exportTrajectoryBundle", () => { terminalError: "non_deliverable_terminal_turn", assistantTexts: ["done"], finalPromptText: `final prompt from ${path.join(tmpDir, "prompt.txt")}`, + finalPromptTextOriginalLength: 12_345, itemLifecycle: { startedCount: 1, completedCount: 1, @@ -1849,6 +1850,7 @@ describe("exportTrajectoryBundle", () => { const tools = fs.readFileSync(path.join(outputDir, "tools.json"), "utf8"); expect(prompts).toContain("$WORKSPACE_DIR/AGENTS.md"); expect(artifacts).toContain("$WORKSPACE_DIR/prompt.txt"); + expect(JSON.parse(artifacts).finalPromptTextOriginalLength).toBe(12_345); expect(artifacts).toContain("non_deliverable_terminal_turn"); expect(systemPrompt).toContain("$WORKSPACE_DIR/instructions.md"); expect(tools).toContain("$WORKSPACE_DIR/docs"); diff --git a/src/trajectory/export.ts b/src/trajectory/export.ts index 8f888d63104e..57a24f4e8138 100644 --- a/src/trajectory/export.ts +++ b/src/trajectory/export.ts @@ -993,6 +993,9 @@ function buildArtifactsCapture(params: { compactionCount: runtimeArtifacts?.compactionCount ?? runtimeCompletion?.compactionCount, assistantTexts: runtimeArtifacts?.assistantTexts ?? runtimeCompletion?.assistantTexts, finalPromptText: runtimeArtifacts?.finalPromptText ?? runtimeCompletion?.finalPromptText, + finalPromptTextOriginalLength: + runtimeArtifacts?.finalPromptTextOriginalLength ?? + runtimeCompletion?.finalPromptTextOriginalLength, itemLifecycle: runtimeArtifacts?.itemLifecycle, toolMetas: runtimeArtifacts?.toolMetas, didSendViaMessagingTool: runtimeArtifacts?.didSendViaMessagingTool, diff --git a/src/trajectory/runtime-store.sqlite.test.ts b/src/trajectory/runtime-store.sqlite.test.ts index 39438c6c436e..39df27f2c1ea 100644 --- a/src/trajectory/runtime-store.sqlite.test.ts +++ b/src/trajectory/runtime-store.sqlite.test.ts @@ -2,7 +2,7 @@ import fs from "node:fs"; import os from "node:os"; import path from "node:path"; -import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { replaceSessionEntry } from "../config/sessions/session-accessor.js"; import { executeSqliteQuerySync, getNodeSqliteKysely } from "../infra/kysely-sync.js"; import type { DB as OpenClawAgentKyselyDatabase } from "../state/openclaw-agent-db.generated.js"; @@ -33,6 +33,7 @@ describe("SQLite trajectory runtime store", () => { }); afterEach(() => { + vi.useRealTimers(); closeOpenClawAgentDatabasesForTest(); closeOpenClawStateDatabaseForTest(); fs.rmSync(tempDir, { recursive: true, force: true }); @@ -86,6 +87,111 @@ describe("SQLite trajectory runtime store", () => { expect(events.map((event) => event.type)).toEqual(["event-3", "event-4"]); }); + it("drops old runs while retaining recent runs", async () => { + const now = Date.parse("2026-07-26T00:00:00.000Z"); + vi.useFakeTimers(); + vi.setSystemTime(now); + appendSqliteTrajectoryRuntimeEvents({ sessionId: "session-1", storePath }, [ + createTrajectoryEvent({ type: "current", ts: new Date(now).toISOString() }), + ]); + await addSession("history"); + appendSqliteTrajectoryRuntimeEvents({ sessionId: "history", storePath }, [ + createTrajectoryEvent({ + runId: "old-run", + sessionId: "history", + type: "old", + ts: new Date(now - 15 * 24 * 60 * 60 * 1_000).toISOString(), + }), + createTrajectoryEvent({ + runId: "recent-run", + sessionId: "history", + type: "recent", + ts: new Date(now - 13 * 24 * 60 * 60 * 1_000).toISOString(), + }), + ]); + + vi.advanceTimersByTime(60 * 60 * 1_000); + appendSqliteTrajectoryRuntimeEvents({ sessionId: "session-1", storePath }, [ + createTrajectoryEvent({ type: "sweep-trigger", ts: new Date(Date.now()).toISOString() }), + ]); + + await expect(runtimeEventTypes("history")).resolves.toEqual(["recent"]); + }); + + it("evicts oldest runs to the global byte budget without touching the current session", async () => { + const now = Date.parse("2026-07-26T00:00:00.000Z"); + vi.useFakeTimers(); + vi.setSystemTime(now); + appendSqliteTrajectoryRuntimeEvents({ sessionId: "session-1", storePath }, [ + createTrajectoryEvent({ payloadSize: 200, type: "current-initial" }), + ]); + for (const [index, sessionId] of ["oldest", "middle", "newest"].entries()) { + await addSession(sessionId); + appendSqliteTrajectoryRuntimeEvents({ sessionId, storePath }, [ + createTrajectoryEvent({ + payloadSize: 200, + runId: `${sessionId}-run`, + sessionId, + type: sessionId, + ts: new Date(now - (3 - index) * 24 * 60 * 60 * 1_000).toISOString(), + }), + ]); + } + const bytesBefore = runtimeBytesBySession(); + const trigger = createTrajectoryEvent({ + payloadSize: 200, + type: "current-newest", + ts: new Date(now + 60 * 60 * 1_000).toISOString(), + }); + const triggerBytes = Buffer.byteLength(JSON.stringify(trigger), "utf8") + 1; + const maxGlobalRuntimeBytes = + [...bytesBefore.values()].reduce((total, bytes) => total + bytes, 0) + + triggerBytes - + (bytesBefore.get("oldest") ?? 0) - + (bytesBefore.get("middle") ?? 0); + + vi.advanceTimersByTime(60 * 60 * 1_000); + appendSqliteTrajectoryRuntimeEvents( + { maxGlobalRuntimeBytes, sessionId: "session-1", storePath }, + [trigger], + ); + + await expect(runtimeEventTypes("oldest")).resolves.toEqual([]); + await expect(runtimeEventTypes("middle")).resolves.toEqual([]); + await expect(runtimeEventTypes("newest")).resolves.toEqual(["newest"]); + await expect(runtimeEventTypes("session-1")).resolves.toEqual([ + "current-initial", + "current-newest", + ]); + }); + + it("rate-limits the global sweep instead of running it on every insert", async () => { + const now = Date.parse("2026-07-26T00:00:00.000Z"); + vi.useFakeTimers(); + vi.setSystemTime(now); + appendSqliteTrajectoryRuntimeEvents({ sessionId: "session-1", storePath }, [ + createTrajectoryEvent({ type: "current" }), + ]); + await addSession("old-session"); + appendSqliteTrajectoryRuntimeEvents({ sessionId: "old-session", storePath }, [ + createTrajectoryEvent({ + sessionId: "old-session", + type: "old", + ts: new Date(now - 15 * 24 * 60 * 60 * 1_000).toISOString(), + }), + ]); + appendSqliteTrajectoryRuntimeEvents({ sessionId: "session-1", storePath }, [ + createTrajectoryEvent({ type: "same-window" }), + ]); + await expect(runtimeEventTypes("old-session")).resolves.toEqual(["old"]); + + vi.advanceTimersByTime(60 * 60 * 1_000); + appendSqliteTrajectoryRuntimeEvents({ sessionId: "session-1", storePath }, [ + createTrajectoryEvent({ type: "next-window", ts: new Date(Date.now()).toISOString() }), + ]); + await expect(runtimeEventTypes("old-session")).resolves.toEqual([]); + }); + it("cascades trajectory rows when the session row is deleted", async () => { appendSqliteTrajectoryRuntimeEvents({ sessionId: "session-1", storePath }, [ createTrajectoryEvent({ type: "model.started" }), @@ -102,21 +208,58 @@ describe("SQLite trajectory runtime store", () => { function sqlitePath(): string { return path.join(tempDir, "agents", "main", "agent", "openclaw-agent.sqlite"); } + + async function addSession(sessionId: string): Promise { + await replaceSessionEntry( + { sessionKey: `agent:main:${sessionId}`, storePath }, + { sessionId, updatedAt: Date.now() }, + ); + } + + async function runtimeEventTypes(sessionId: string): Promise { + const events = await loadSqliteTrajectoryRuntimeEvents({ sessionId, storePath }); + return events.map((event) => event.type); + } + + function runtimeBytesBySession(): Map { + const database = openOpenClawAgentDatabase({ agentId: "main", path: sqlitePath() }); + const db = getNodeSqliteKysely(database.db); + const rows = executeSqliteQuerySync( + database.db, + db.selectFrom("trajectory_runtime_events").select(["session_id", "event_json"]), + ).rows; + const bytesBySession = new Map(); + for (const row of rows) { + bytesBySession.set( + row.session_id, + (bytesBySession.get(row.session_id) ?? 0) + Buffer.byteLength(row.event_json, "utf8") + 1, + ); + } + return bytesBySession; + } }); -function createTrajectoryEvent(options: { seq?: number; type: string }): TrajectoryEvent { +function createTrajectoryEvent(options: { + payloadSize?: number; + runId?: string; + seq?: number; + sessionId?: string; + ts?: string; + type: string; +}): TrajectoryEvent { + const sessionId = options.sessionId ?? "session-1"; return { traceSchema: "openclaw-trajectory", schemaVersion: 1, - traceId: "session-1", + traceId: sessionId, source: "runtime", type: options.type, - ts: "2026-07-03T00:00:00.000Z", + ts: options.ts ?? "2026-07-03T00:00:00.000Z", seq: options.seq ?? 1, sourceSeq: options.seq ?? 1, - sessionId: "session-1", - sessionKey: "agent:main:main", - runId: "run-1", - data: { payload: "x".repeat(120) }, + sessionId, + sessionKey: `agent:main:${sessionId}`, + runId: options.runId ?? "run-1", + data: { payload: "x".repeat(options.payloadSize ?? 120) }, }; } diff --git a/src/trajectory/runtime-store.sqlite.ts b/src/trajectory/runtime-store.sqlite.ts index 83c181692159..80c66a88e9ae 100644 --- a/src/trajectory/runtime-store.sqlite.ts +++ b/src/trajectory/runtime-store.sqlite.ts @@ -22,14 +22,25 @@ type SqliteTrajectoryRuntimeDatabase = Pick< "trajectory_runtime_events" >; +const TRAJECTORY_RUNTIME_RETENTION_MAX_AGE_MS = 14 * 24 * 60 * 60 * 1_000; +const TRAJECTORY_RUNTIME_GLOBAL_MAX_BYTES = 512 * 1024 * 1024; +const TRAJECTORY_RUNTIME_GLOBAL_SWEEP_INTERVAL_MS = 60 * 60 * 1_000; +const TRAJECTORY_RUNTIME_DELETE_RUN_BATCH_SIZE = 100; + export type SqliteTrajectoryRuntimeScope = { agentId?: string; env?: NodeJS.ProcessEnv; + maxGlobalRuntimeBytes?: number; maxRuntimeBytes?: number; sessionId: string; storePath: string; }; +type SqliteTrajectoryRuntimeReadScope = Omit< + SqliteTrajectoryRuntimeScope, + "maxGlobalRuntimeBytes" | "maxRuntimeBytes" +>; + type SqliteTrajectoryRuntimeEventRow = { event: TrajectoryEvent; seq: number; @@ -40,6 +51,17 @@ type TrajectoryRuntimeRow = { seq: number; }; +type TrajectoryRuntimeRun = { + newestCreatedAt: number; + runId: string | null; + runtimeBytes: number; + sessionId: string; +}; + +// The runtime store owns this process-local cadence. Database handles are cached, +// so a WeakMap rate-limits work without retaining closed agent databases. +const lastGlobalSweepAtByDatabase = new WeakMap(); + /** Appends runtime trajectory events to the per-agent SQLite session store. */ export function appendSqliteTrajectoryRuntimeEvents( scope: SqliteTrajectoryRuntimeScope, @@ -53,6 +75,12 @@ export function appendSqliteTrajectoryRuntimeEvents( 1, Math.floor(scope.maxRuntimeBytes ?? TRAJECTORY_RUNTIME_CAPTURE_MAX_BYTES), ); + const maxGlobalRuntimeBytes = Math.max( + 1, + Math.floor(scope.maxGlobalRuntimeBytes ?? TRAJECTORY_RUNTIME_GLOBAL_MAX_BYTES), + ); + const sweepAt = Date.now(); + let sweptDatabase: OpenClawAgentDatabase | undefined; runOpenClawAgentWriteTransaction((database) => { const db = getTrajectoryKysely(database.db); let seq = readNextTrajectorySeq(database, scope.sessionId); @@ -71,26 +99,43 @@ export function appendSqliteTrajectoryRuntimeEvents( seq += 1; } trimSqliteTrajectoryRuntimeWindow(database, scope.sessionId, maxRuntimeBytes); + const lastSweptAt = lastGlobalSweepAtByDatabase.get(database); + if ( + lastSweptAt === undefined || + sweepAt < lastSweptAt || + sweepAt - lastSweptAt >= TRAJECTORY_RUNTIME_GLOBAL_SWEEP_INTERVAL_MS + ) { + sweepSqliteTrajectoryRuntimeRetention( + database, + scope.sessionId, + sweepAt, + maxGlobalRuntimeBytes, + ); + sweptDatabase = database; + } }, options); + if (sweptDatabase) { + lastGlobalSweepAtByDatabase.set(sweptDatabase, sweepAt); + } } /** Loads runtime trajectory events from per-agent SQLite rows in storage order. */ export async function loadSqliteTrajectoryRuntimeEvents( - scope: Omit, + scope: SqliteTrajectoryRuntimeReadScope, ): Promise { return loadSqliteTrajectoryRuntimeEventsSync(scope); } /** Loads runtime trajectory events synchronously for CLI and export paths. */ function loadSqliteTrajectoryRuntimeEventsSync( - scope: Omit, + scope: SqliteTrajectoryRuntimeReadScope, ): TrajectoryEvent[] { return loadSqliteTrajectoryRuntimeEventRowsSync(scope).map((row) => row.event); } /** Loads runtime trajectory event rows with storage seqs for follow/export cursors. */ export function loadSqliteTrajectoryRuntimeEventRowsSync( - scope: Omit & { + scope: SqliteTrajectoryRuntimeReadScope & { afterSeq?: number; maxEvents?: number; }, @@ -116,6 +161,96 @@ export function loadSqliteTrajectoryRuntimeEventRowsSync( })); } +function sweepSqliteTrajectoryRuntimeRetention( + database: OpenClawAgentDatabase, + currentSessionId: string, + now: number, + maxGlobalRuntimeBytes: number, +): void { + const runs = readSqliteTrajectoryRuntimeRuns(database); + let retainedBytes = runs.reduce((total, run) => total + run.runtimeBytes, 0); + const cutoff = now - TRAJECTORY_RUNTIME_RETENTION_MAX_AGE_MS; + const deletedRuns = new Set(); + for (const run of runs) { + if (run.sessionId !== currentSessionId && run.newestCreatedAt < cutoff) { + deletedRuns.add(run); + retainedBytes -= run.runtimeBytes; + } + } + for (const run of runs.toSorted(compareTrajectoryRuntimeRunsOldestFirst)) { + if (retainedBytes <= maxGlobalRuntimeBytes) { + break; + } + // The per-session trim already bounds the active writer. Preserve its just-written + // events while the global budget evicts complete older runs elsewhere. + if (run.sessionId === currentSessionId || deletedRuns.has(run)) { + continue; + } + deletedRuns.add(run); + retainedBytes -= run.runtimeBytes; + } + deleteSqliteTrajectoryRuntimeRuns(database, [...deletedRuns]); +} + +function readSqliteTrajectoryRuntimeRuns(database: OpenClawAgentDatabase): TrajectoryRuntimeRun[] { + const db = getTrajectoryKysely(database.db); + const rows = executeSqliteQuerySync( + database.db, + db + .selectFrom("trajectory_runtime_events") + .select(["session_id", "run_id"]) + .select((eb) => [ + eb.fn.max("created_at").as("newest_created_at"), + eb.fn + .sum(eb(eb.fn("octet_length", ["event_json"]), "+", 1)) + .as("runtime_bytes"), + ]) + .groupBy(["session_id", "run_id"]), + ).rows; + return rows.map((row) => ({ + newestCreatedAt: normalizeSqliteNumber(row.newest_created_at), + runId: row.run_id, + runtimeBytes: normalizeSqliteNumber(row.runtime_bytes), + sessionId: row.session_id, + })); +} + +function deleteSqliteTrajectoryRuntimeRuns( + database: OpenClawAgentDatabase, + runs: readonly TrajectoryRuntimeRun[], +): void { + const db = getTrajectoryKysely(database.db); + for (let index = 0; index < runs.length; index += TRAJECTORY_RUNTIME_DELETE_RUN_BATCH_SIZE) { + const batch = runs.slice(index, index + TRAJECTORY_RUNTIME_DELETE_RUN_BATCH_SIZE); + executeSqliteQuerySync( + database.db, + db + .deleteFrom("trajectory_runtime_events") + .where((eb) => + eb.or( + batch.map((run) => + eb.and([ + eb("session_id", "=", run.sessionId), + run.runId === null ? eb("run_id", "is", null) : eb("run_id", "=", run.runId), + ]), + ), + ), + ), + ); + } +} + +function compareTrajectoryRuntimeRunsOldestFirst( + left: TrajectoryRuntimeRun, + right: TrajectoryRuntimeRun, +): number { + return ( + left.newestCreatedAt - right.newestCreatedAt || + left.sessionId.localeCompare(right.sessionId) || + (left.runId ?? "").localeCompare(right.runId ?? "") + ); +} + function getTrajectoryKysely(database: import("node:sqlite").DatabaseSync) { return getNodeSqliteKysely(database); } diff --git a/src/trajectory/runtime.test.ts b/src/trajectory/runtime.test.ts index 7dcdab5b92b4..7442fe1de526 100644 --- a/src/trajectory/runtime.test.ts +++ b/src/trajectory/runtime.test.ts @@ -345,6 +345,52 @@ describe("trajectory runtime", () => { expect(second.data.truncated).toBeUndefined(); }); + it("caps large final prompts and records their original length", () => { + const writes: string[] = []; + const finalPromptText = `prompt-${"🙂".repeat(4_096)}`; + const recorder = createTrajectoryRuntimeRecorder({ + sessionId: "session-1", + sessionFile: "/tmp/session.jsonl", + writer: { + filePath: "/tmp/session.trajectory.jsonl", + write: (line) => { + writes.push(line); + }, + flush: async () => undefined, + }, + }); + + expectTrajectoryRuntimeRecorder(recorder).recordEvent("model.completed", { finalPromptText }); + + const parsed = JSON.parse(expectDefined(writes[0], "writes[0] test invariant")); + expect(Buffer.byteLength(parsed.data.finalPromptText, "utf8")).toBeLessThanOrEqual(4 * 1024); + expect(Buffer.byteLength(parsed.data.finalPromptText, "utf8")).toBeGreaterThan(4 * 1024 - 4); + expect(parsed.data.finalPromptTextOriginalLength).toBe(finalPromptText.length); + }); + + it("leaves short final prompts unchanged", () => { + const writes: string[] = []; + const recorder = createTrajectoryRuntimeRecorder({ + sessionId: "session-1", + sessionFile: "/tmp/session.jsonl", + writer: { + filePath: "/tmp/session.trajectory.jsonl", + write: (line) => { + writes.push(line); + }, + flush: async () => undefined, + }, + }); + + expectTrajectoryRuntimeRecorder(recorder).recordEvent("trace.artifacts", { + finalPromptText: "short prompt", + }); + + const parsed = JSON.parse(expectDefined(writes[0], "writes[0] test invariant")); + expect(parsed.data.finalPromptText).toBe("short prompt"); + expect(parsed.data.finalPromptTextOriginalLength).toBeUndefined(); + }); + it("redacts secrets before preserving usage in truncated runtime events", () => { const writes: string[] = []; const recorder = createTrajectoryRuntimeRecorder({ diff --git a/src/trajectory/runtime.ts b/src/trajectory/runtime.ts index 666641165fb1..abc6d9cf97b9 100644 --- a/src/trajectory/runtime.ts +++ b/src/trajectory/runtime.ts @@ -9,6 +9,7 @@ import type { OpenClawConfig } from "../config/types.openclaw.js"; import { redactSecrets } from "../logging/redact.js"; import { parseBooleanValue } from "../utils/boolean.js"; import { safeJsonStringify } from "../utils/safe-json.js"; +import { truncateUtf8Prefix } from "../utils/utf8-truncate.js"; import { TRAJECTORY_RUNTIME_CAPTURE_MAX_BYTES, TRAJECTORY_RUNTIME_EVENT_MAX_BYTES, @@ -42,6 +43,7 @@ const TRAJECTORY_RUNTIME_DATA_STRING_MAX_CHARS = 32_768; const TRAJECTORY_RUNTIME_DATA_ARRAY_MAX_ITEMS = 64; const TRAJECTORY_RUNTIME_DATA_OBJECT_MAX_KEYS = 64; const TRAJECTORY_RUNTIME_DATA_MAX_DEPTH = 6; +const TRAJECTORY_RUNTIME_FINAL_PROMPT_MAX_BYTES = 4 * 1024; const TRAJECTORY_RUNTIME_OVERSIZE_PRESERVED_DATA_KEYS = ["usage", "promptCache"] as const; type TrajectoryRuntimeWriterDiagnostics = QueuedFileWriterDiagnostics; @@ -182,10 +184,29 @@ function limitTrajectoryPayloadValue( } function sanitizeTrajectoryPayload(data: Record): Record { - return redactSecrets(sanitizeDiagnosticPayload(limitTrajectoryPayloadValue(data))) as Record< - string, - unknown - >; + const finalPromptText = data.finalPromptText; + const redactedFinalPromptText = + typeof finalPromptText === "string" ? (redactSecrets(finalPromptText) as string) : undefined; + const boundedData = + typeof finalPromptText === "string" && + typeof redactedFinalPromptText === "string" && + (Buffer.byteLength(finalPromptText, "utf8") > TRAJECTORY_RUNTIME_FINAL_PROMPT_MAX_BYTES || + Buffer.byteLength(redactedFinalPromptText, "utf8") > + TRAJECTORY_RUNTIME_FINAL_PROMPT_MAX_BYTES) + ? { + ...data, + finalPromptText: truncateUtf8Prefix( + redactedFinalPromptText, + TRAJECTORY_RUNTIME_FINAL_PROMPT_MAX_BYTES, + ), + finalPromptTextOriginalLength: finalPromptText.length, + } + : typeof redactedFinalPromptText === "string" + ? { ...data, finalPromptText: redactedFinalPromptText } + : data; + return redactSecrets( + sanitizeDiagnosticPayload(limitTrajectoryPayloadValue(boundedData)), + ) as Record; } function describeTrajectoryWriterFlushState(writer: TrajectoryRuntimeWriter): string | undefined {