fix(agents): add trajectory flush timeout diagnostics

galiniliev · web-flow · commit ddeaebfc6807 · 2026-05-18T23:29:37.000-07:00
Adds bounded queued-writer diagnostics to pi-trajectory-flush cleanup timeout warnings so operators can see pending write count, queued bytes, active operation, and append size without exposing paths or payloads. Closes #82961
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -6,6 +6,7 @@ Docs: https://docs.openclaw.ai
 
 ### Fixes
 
+- Agents: include bounded trajectory queued-writer diagnostics in `pi-trajectory-flush` timeout warnings so flush stalls show pending writes, queued bytes, and append state. Fixes #82961. (#82962) Thanks @galiniliev.
 - Agents/subagents: recover stale completion announces by retrying unsupported transcript-wait wakes without transcript waiting and forcing a message-tool handoff when the requester run is already stale. Fixes #83699. (#83700) Thanks @galiniliev.
 - Agents/subagents: skip stale embedded-run wake probes for dormant completion requesters, so late subagent completions go straight to requester-agent/direct handoff instead of producing `reason=no_active_run` queue noise. (#82964) Thanks @galiniliev.
 - CLI: retry config snapshot reads after a transient failure so one rejected read no longer poisons later commands in the same process. (#83931) Thanks @honor2030.
diff --git a/src/agents/pi-embedded-runner/run/attempt.ts b/src/agents/pi-embedded-runner/run/attempt.ts
@@ -4788,6 +4788,7 @@ export async function runEmbeddedAttempt(
         sessionId: params.sessionId,
         step: "pi-trajectory-flush",
         log,
+        getTimeoutDetails: () => trajectoryRecorder?.describeFlushState(),
         cleanup: async () => {
           await trajectoryRecorder?.flush();
         },
diff --git a/src/agents/queued-file-writer.test.ts b/src/agents/queued-file-writer.test.ts
@@ -92,4 +92,34 @@ describe("getQueuedFileWriter", () => {
 
     expect(fs.readFileSync(filePath, "utf8")).toBe("12345\n");
   });
+
+  it("reports pending queue diagnostics before flush drains writes", async () => {
+    const tmpDir = makeTempDir();
+    const filePath = path.join(tmpDir, "trace.jsonl");
+    const writer = getQueuedFileWriter(new Map(), filePath, {
+      maxFileBytes: 1024,
+      maxQueuedBytes: 1024,
+      yieldBeforeWrite: true,
+    });
+
+    writer.write("line\n");
+
+    expect(writer.describeQueue?.()).toEqual({
+      pendingWrites: 1,
+      queuedBytes: 5,
+      activeOperation: "idle",
+      activeWriteBytes: undefined,
+      maxFileBytes: 1024,
+      maxQueuedBytes: 1024,
+      yieldBeforeWrite: true,
+    });
+
+    await writer.flush();
+
+    expect(writer.describeQueue?.()).toMatchObject({
+      pendingWrites: 0,
+      queuedBytes: 0,
+      activeOperation: "idle",
+    });
+  });
 });
diff --git a/src/agents/queued-file-writer.ts b/src/agents/queued-file-writer.ts
@@ -4,10 +4,21 @@ import { appendRegularFile, resolveRegularFileAppendFlags } from "../infra/fs-sa
 
 export type QueuedFileWriteResult = "queued" | "dropped";
 
+export type QueuedFileWriterDiagnostics = {
+  pendingWrites: number;
+  queuedBytes: number;
+  activeOperation: "idle" | "mkdir" | "yield" | "file-append";
+  activeWriteBytes?: number;
+  maxFileBytes?: number;
+  maxQueuedBytes?: number;
+  yieldBeforeWrite: boolean;
+};
+
 export type QueuedFileWriter = {
   filePath: string;
   write: (line: string) => unknown;
   flush: () => Promise<void>;
+  describeQueue?: () => QueuedFileWriterDiagnostics;
 };
 
 type QueuedFileWriterOptions = {
@@ -50,7 +61,10 @@ export function getQueuedFileWriter(
   const dir = path.dirname(filePath);
   const ready = fs.mkdir(dir, { recursive: true, mode: 0o700 }).catch(() => undefined);
   let queue: Promise<unknown> = Promise.resolve();
+  let pendingWrites = 0;
   let queuedBytes = 0;
+  let activeOperation: QueuedFileWriterDiagnostics["activeOperation"] = "idle";
+  let activeWriteBytes: number | undefined;
 
   const writer: QueuedFileWriter = {
     filePath,
@@ -62,20 +76,45 @@ export function getQueuedFileWriter(
       ) {
         return "dropped";
       }
+      pendingWrites += 1;
       queuedBytes += lineBytes;
       queue = queue
-        .then(() => ready)
-        .then(() => (options.yieldBeforeWrite ? waitForImmediate() : undefined))
-        .then(() => safeAppendFile(filePath, line, options))
+        .then(async () => {
+          activeOperation = "mkdir";
+          await ready;
+        })
+        .then(async () => {
+          if (options.yieldBeforeWrite) {
+            activeOperation = "yield";
+            await waitForImmediate();
+          }
+        })
+        .then(async () => {
+          activeOperation = "file-append";
+          activeWriteBytes = lineBytes;
+          await safeAppendFile(filePath, line, options);
+        })
         .catch(() => undefined)
         .finally(() => {
+          pendingWrites = Math.max(0, pendingWrites - 1);
           queuedBytes = Math.max(0, queuedBytes - lineBytes);
+          activeWriteBytes = undefined;
+          activeOperation = pendingWrites > 0 ? activeOperation : "idle";
         });
       return "queued";
     },
     flush: async () => {
       await queue;
     },
+    describeQueue: () => ({
+      pendingWrites,
+      queuedBytes,
+      activeOperation,
+      activeWriteBytes,
+      maxFileBytes: options.maxFileBytes,
+      maxQueuedBytes: options.maxQueuedBytes,
+      yieldBeforeWrite: options.yieldBeforeWrite === true,
+    }),
   };
 
   writers.set(filePath, writer);
diff --git a/src/agents/run-cleanup-timeout.test.ts b/src/agents/run-cleanup-timeout.test.ts
@@ -65,6 +65,50 @@ describe("agent cleanup timeout", () => {
     );
   });
 
+  it("includes cleanup timeout details when the cleanup step exposes them", async () => {
+    const cleanup = vi.fn(async () => new Promise<never>(() => {}));
+
+    const result = runAgentCleanupStep({
+      runId: "run-trajectory",
+      sessionId: "session-trajectory",
+      step: "pi-trajectory-flush",
+      cleanup,
+      log,
+      timeoutMs: 5,
+      getTimeoutDetails: () => "pendingWrites=2 queuedBytes=128 activeOperation=file-append",
+    });
+
+    await vi.advanceTimersByTimeAsync(5);
+    await expect(result).resolves.toBeUndefined();
+
+    expect(log.warn).toHaveBeenCalledWith(
+      "agent cleanup timed out: runId=run-trajectory sessionId=session-trajectory step=pi-trajectory-flush timeoutMs=5 details=pendingWrites=2 queuedBytes=128 activeOperation=file-append",
+    );
+  });
+
+  it("does not fail cleanup when timeout details throw", async () => {
+    const cleanup = vi.fn(async () => new Promise<never>(() => {}));
+
+    const result = runAgentCleanupStep({
+      runId: "run-trajectory",
+      sessionId: "session-trajectory",
+      step: "pi-trajectory-flush",
+      cleanup,
+      log,
+      timeoutMs: 5,
+      getTimeoutDetails: () => {
+        throw new Error("details unavailable");
+      },
+    });
+
+    await vi.advanceTimersByTimeAsync(5);
+    await expect(result).resolves.toBeUndefined();
+
+    expect(log.warn).toHaveBeenCalledWith(
+      "agent cleanup timed out: runId=run-trajectory sessionId=session-trajectory step=pi-trajectory-flush timeoutMs=5 detailsError=details unavailable",
+    );
+  });
+
   it("uses the general cleanup timeout environment override for other cleanup steps", async () => {
     const cleanup = vi.fn(async () => new Promise<never>(() => {}));
 
diff --git a/src/agents/run-cleanup-timeout.ts b/src/agents/run-cleanup-timeout.ts
@@ -28,6 +28,17 @@ function parseTimeoutEnvValue(value: string | undefined): number | undefined {
   return normalized > 0 ? normalized : undefined;
 }
 
+function resolveCleanupTimeoutDetails(
+  getTimeoutDetails: (() => string | undefined) | undefined,
+): string {
+  try {
+    const timeoutDetails = getTimeoutDetails?.()?.trim();
+    return timeoutDetails ? ` details=${timeoutDetails}` : "";
+  } catch (error) {
+    return ` detailsError=${formatErrorMessage(error)}`;
+  }
+}
+
 export function resolveAgentCleanupStepTimeoutMs(params: {
   step: string;
   timeoutMs?: number;
@@ -54,6 +65,7 @@ export async function runAgentCleanupStep(params: {
   sessionId: string;
   step: string;
   cleanup: () => Promise<void>;
+  getTimeoutDetails?: () => string | undefined;
   log: AgentCleanupLogger;
   env?: NodeJS.ProcessEnv;
   timeoutMs?: number;
@@ -88,8 +100,9 @@ export async function runAgentCleanupStep(params: {
     clearTimeout(timeoutHandle);
   }
   if (result === "timeout") {
+    const details = resolveCleanupTimeoutDetails(params.getTimeoutDetails);
     params.log.warn(
-      `agent cleanup timed out: runId=${params.runId} sessionId=${params.sessionId} step=${params.step} timeoutMs=${timeoutMs}`,
+      `agent cleanup timed out: runId=${params.runId} sessionId=${params.sessionId} step=${params.step} timeoutMs=${timeoutMs}${details}`,
     );
     void cleanupPromise.catch((error) => {
       params.log.warn(
diff --git a/src/trajectory/runtime.test.ts b/src/trajectory/runtime.test.ts
@@ -169,6 +169,33 @@ describe("trajectory runtime", () => {
     expect(truncated?.data.droppedEvents).toBeGreaterThan(0);
   });
 
+  it("describes queued writer state for cleanup timeout logs", () => {
+    const recorder = createTrajectoryRuntimeRecorder({
+      sessionId: "session-1",
+      sessionFile: "/tmp/session.jsonl",
+      writer: {
+        filePath: "/tmp/session.trajectory.jsonl",
+        write: () => "queued",
+        flush: async () => undefined,
+        describeQueue: () => ({
+          pendingWrites: 2,
+          queuedBytes: 256,
+          activeOperation: "file-append",
+          activeWriteBytes: 128,
+          maxFileBytes: 1024,
+          maxQueuedBytes: 1024,
+          yieldBeforeWrite: true,
+        }),
+      },
+    });
+
+    const runtimeRecorder = expectTrajectoryRuntimeRecorder(recorder);
+
+    expect(runtimeRecorder.describeFlushState()).toBe(
+      "pendingWrites=2 queuedBytes=256 activeOperation=file-append yieldBeforeWrite=true activeWriteBytes=128 maxQueuedBytes=1024 maxFileBytes=1024",
+    );
+  });
+
   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
@@ -45,6 +45,7 @@ type TrajectoryRuntimeRecorder = {
   filePath: string;
   recordEvent: (type: string, data?: Record<string, unknown>) => void;
   flush: () => Promise<void>;
+  describeFlushState: () => string | undefined;
 };
 
 const writers = new Map<string, QueuedFileWriter>();
@@ -208,6 +209,29 @@ function sanitizeTrajectoryPayload(data: Record<string, unknown>): Record<string
   >;
 }
 
+function describeTrajectoryWriterFlushState(writer: QueuedFileWriter): string | undefined {
+  const diagnostics = writer.describeQueue?.();
+  if (!diagnostics) {
+    return undefined;
+  }
+  const parts = [
+    `pendingWrites=${diagnostics.pendingWrites}`,
+    `queuedBytes=${diagnostics.queuedBytes}`,
+    `activeOperation=${diagnostics.activeOperation}`,
+    `yieldBeforeWrite=${diagnostics.yieldBeforeWrite}`,
+  ];
+  if (diagnostics.activeWriteBytes !== undefined) {
+    parts.push(`activeWriteBytes=${diagnostics.activeWriteBytes}`);
+  }
+  if (diagnostics.maxQueuedBytes !== undefined) {
+    parts.push(`maxQueuedBytes=${diagnostics.maxQueuedBytes}`);
+  }
+  if (diagnostics.maxFileBytes !== undefined) {
+    parts.push(`maxFileBytes=${diagnostics.maxFileBytes}`);
+  }
+  return parts.join(" ");
+}
+
 export function toTrajectoryToolDefinitions(
   tools: ReadonlyArray<{ name?: string; description?: string; parameters?: unknown }>,
 ): TrajectoryToolDefinition[] {
@@ -361,5 +385,6 @@ export function createTrajectoryRuntimeRecorder(
         writers.delete(filePath);
       }
     },
+    describeFlushState: () => describeTrajectoryWriterFlushState(writer),
   };
 }
