fix: bound trajectory runtime flush (#77154)

steipete · web-flow · commit 474bea162b4d · 2026-05-04T09:48:03.000+01:00
* fix: bound trajectory runtime flush

* fix: keep trajectory export cap compatible

* test: keep followup delivery test pure
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -85,6 +85,7 @@ Docs: https://docs.openclaw.ai
 - Google Meet: fork the caller's current agent transcript into agent-mode meeting consultant sessions, so Meet replies inherit the context from the tool call that joined the meeting.
 - iOS/mobile pairing: reject non-loopback `ws://` setup URLs before QR/setup-code issuance and let the iOS Gateway settings screen scan QR codes or paste full setup-code messages. Thanks @BunsDev.
 - Control UI: keep Gateway Access inputs and locale picker contained inside the card at narrow and tablet widths.
+- Agents/trajectory: bound runtime trajectory capture and yield queued sidecar writes so oversized traces stop recording instead of monopolizing Gateway cleanup. Fixes #77124. Thanks @loyur.
 - Telegram/streaming: sanitize tool-progress draft preview backticks before shared compaction, so long backtick-heavy progress text still renders inside the safe code-formatted preview instead of collapsing to an ellipsis.
 - UI/chat: remove the unsupported `line-clamp` declaration from the chat queue text rule to eliminate Firefox console noise without changing visible truncation behavior. Thanks @ZanderH-code.
 - Agents/Pi: suppress persistence for synthetic mid-turn overflow continuation prompts, so transcript-retry recovery does not write the "continue from transcript" prompt as a new user turn. Thanks @vincentkoc.
diff --git a/docs/tools/trajectory.md b/docs/tools/trajectory.md
@@ -181,7 +181,7 @@ OpenClaw redacts sensitive values before writing export files:
 
 The exporter also bounds input size:
 
-- runtime sidecar files: 50 MiB
+- runtime sidecar files: live capture stops at 10 MiB and records a truncation event when space remains; export accepts existing runtime sidecars up to 50 MiB
 - session files: 50 MiB
 - runtime events: 200,000
 - total exported events: 250,000
diff --git a/src/agents/queued-file-writer.test.ts b/src/agents/queued-file-writer.test.ts
@@ -80,4 +80,16 @@ describe("getQueuedFileWriter", () => {
 
     expect(fs.readFileSync(filePath, "utf8")).toBe("12345\n");
   });
+
+  it("drops writes that would exceed the pending queue cap", async () => {
+    const tmpDir = makeTempDir();
+    const filePath = path.join(tmpDir, "trace.jsonl");
+    const writer = getQueuedFileWriter(new Map(), filePath, { maxQueuedBytes: 6 });
+
+    expect(writer.write("12345\n")).toBe("queued");
+    expect(writer.write("after\n")).toBe("dropped");
+    await writer.flush();
+
+    expect(fs.readFileSync(filePath, "utf8")).toBe("12345\n");
+  });
 });
diff --git a/src/agents/queued-file-writer.ts b/src/agents/queued-file-writer.ts
@@ -2,14 +2,18 @@ import nodeFs from "node:fs";
 import fs from "node:fs/promises";
 import path from "node:path";
 
+export type QueuedFileWriteResult = "queued" | "dropped";
+
 export type QueuedFileWriter = {
   filePath: string;
-  write: (line: string) => void;
+  write: (line: string) => unknown;
   flush: () => Promise<void>;
 };
 
 type QueuedFileWriterOptions = {
   maxFileBytes?: number;
+  maxQueuedBytes?: number;
+  yieldBeforeWrite?: boolean;
 };
 
 type QueuedFileAppendFlagConstants = Pick<
@@ -111,6 +115,12 @@ async function safeAppendFile(
   }
 }
 
+function waitForImmediate(): Promise<void> {
+  return new Promise((resolve) => {
+    setImmediate(resolve);
+  });
+}
+
 export function getQueuedFileWriter(
   writers: Map<string, QueuedFileWriter>,
   filePath: string,
@@ -123,15 +133,29 @@ export function getQueuedFileWriter(
 
   const dir = path.dirname(filePath);
   const ready = fs.mkdir(dir, { recursive: true, mode: 0o700 }).catch(() => undefined);
-  let queue = Promise.resolve();
+  let queue: Promise<unknown> = Promise.resolve();
+  let queuedBytes = 0;
 
   const writer: QueuedFileWriter = {
     filePath,
     write: (line: string) => {
+      const lineBytes = Buffer.byteLength(line, "utf8");
+      if (
+        options.maxQueuedBytes !== undefined &&
+        queuedBytes + lineBytes > options.maxQueuedBytes
+      ) {
+        return "dropped";
+      }
+      queuedBytes += lineBytes;
       queue = queue
         .then(() => ready)
+        .then(() => (options.yieldBeforeWrite ? waitForImmediate() : undefined))
         .then(() => safeAppendFile(filePath, line, options))
-        .catch(() => undefined);
+        .catch(() => undefined)
+        .finally(() => {
+          queuedBytes = Math.max(0, queuedBytes - lineBytes);
+        });
+      return "queued";
     },
     flush: async () => {
       await queue;
diff --git a/src/auto-reply/reply/followup-delivery.test.ts b/src/auto-reply/reply/followup-delivery.test.ts
@@ -1,7 +1,11 @@
-import { describe, expect, it } from "vitest";
+import { describe, expect, it, vi } from "vitest";
 import type { OpenClawConfig } from "../../config/config.js";
 import { resolveFollowupDeliveryPayloads } from "./followup-delivery.js";
 
+vi.mock("../../channels/plugins/index.js", () => ({
+  getChannelPlugin: () => undefined,
+}));
+
 const baseConfig = {} as OpenClawConfig;
 
 describe("resolveFollowupDeliveryPayloads", () => {
diff --git a/src/trajectory/export.test.ts b/src/trajectory/export.test.ts
@@ -4,7 +4,7 @@ import path from "node:path";
 import type { Message, Usage } from "@mariozechner/pi-ai";
 import { afterAll, describe, expect, it } from "vitest";
 import { exportTrajectoryBundle, resolveDefaultTrajectoryExportDir } from "./export.js";
-import { resolveTrajectoryPointerFilePath } from "./paths.js";
+import { TRAJECTORY_RUNTIME_FILE_MAX_BYTES, resolveTrajectoryPointerFilePath } from "./paths.js";
 import type { TrajectoryEvent } from "./types.js";
 
 const tempRoot = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-trajectory-"));
@@ -272,7 +272,7 @@ describe("exportTrajectoryBundle", () => {
     const outputDir = path.join(tmpDir, "bundle");
     writeSimpleSessionFile(sessionFile);
     fs.closeSync(fs.openSync(runtimeFile, "w"));
-    fs.truncateSync(runtimeFile, 50 * 1024 * 1024 + 1);
+    fs.truncateSync(runtimeFile, TRAJECTORY_RUNTIME_FILE_MAX_BYTES + 1);
 
     await expect(
       exportTrajectoryBundle({
diff --git a/src/trajectory/paths.ts b/src/trajectory/paths.ts
@@ -2,6 +2,7 @@ import fs from "node:fs";
 import path from "node:path";
 import { resolveHomeRelativePath } from "../infra/home-dir.js";
 
+export const TRAJECTORY_RUNTIME_CAPTURE_MAX_BYTES = 10 * 1024 * 1024;
 export const TRAJECTORY_RUNTIME_FILE_MAX_BYTES = 50 * 1024 * 1024;
 export const TRAJECTORY_RUNTIME_EVENT_MAX_BYTES = 256 * 1024;
 
diff --git a/src/trajectory/runtime.test.ts b/src/trajectory/runtime.test.ts
@@ -88,7 +88,7 @@ describe("trajectory runtime", () => {
     expect(JSON.stringify(parsed.data)).not.toContain("sk-other-secret-token");
   });
 
-  it("truncates events that exceed the runtime event byte limit", () => {
+  it("bounds large runtime event fields before serialization", () => {
     const writes: string[] = [];
     const recorder = createTrajectoryRuntimeRecorder({
       sessionId: "session-1",
@@ -108,15 +108,53 @@ describe("trajectory runtime", () => {
 
     expect(writes).toHaveLength(1);
     const parsed = JSON.parse(writes[0]);
-    expect(parsed.data).toMatchObject({
+    expect(parsed.data.prompt).toMatchObject({
       truncated: true,
-      reason: "trajectory-event-size-limit",
+      reason: "trajectory-field-size-limit",
     });
     expect(Buffer.byteLength(writes[0], "utf8")).toBeLessThanOrEqual(
       TRAJECTORY_RUNTIME_EVENT_MAX_BYTES + 1,
     );
   });
 
+  it("stops runtime capture at the file budget and records a truncation event", async () => {
+    const writes: string[] = [];
+    const recorder = createTrajectoryRuntimeRecorder({
+      sessionId: "session-1",
+      sessionFile: "/tmp/session.jsonl",
+      maxRuntimeFileBytes: 900,
+      writer: {
+        filePath: "/tmp/session.trajectory.jsonl",
+        write: (line) => {
+          writes.push(line);
+        },
+        flush: async () => undefined,
+      },
+    });
+
+    recorder?.recordEvent("context.compiled", {
+      prompt: "x".repeat(180),
+    });
+    recorder?.recordEvent("prompt.submitted", {
+      prompt: "y".repeat(180),
+    });
+    recorder?.recordEvent("model.completed", {
+      get prompt() {
+        throw new Error("stopped recorder should not read dropped payloads");
+      },
+    });
+    await recorder?.flush();
+
+    const parsed = writes.map((line) => JSON.parse(line));
+    expect(parsed.map((event) => event.type)).toContain("trace.truncated");
+    const truncated = parsed.find((event) => event.type === "trace.truncated");
+    expect(truncated?.data).toMatchObject({
+      reason: "trajectory-runtime-file-size-limit",
+      limitBytes: 900,
+    });
+    expect(truncated?.data.droppedEvents).toBeGreaterThan(0);
+  });
+
   it("writes a session-adjacent pointer when using an override directory", () => {
     const tmpDir = makeTempDir();
     const sessionFile = path.join(tmpDir, "session.jsonl");
diff --git a/src/trajectory/runtime.ts b/src/trajectory/runtime.ts
