Skip to content

Commit 470098b

Browse files
steipetesteipete
committed
fix: keep embedded run lanes from wedging
1 parent b83b639 commit 470098b

10 files changed

Lines changed: 303 additions & 37 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ Docs: https://docs.openclaw.ai
2626

2727
### Fixes
2828

29+
- Agents/Codex: bound embedded-run cleanup, trajectory flushing, and command-lane task timeouts after runtime failures, so Discord and other chat sessions return to idle instead of staying stuck in processing. Thanks @vincentkoc.
2930
- Web fetch: add a documented `tools.web.fetch.ssrfPolicy.allowIpv6UniqueLocalRange` opt-in and thread it through cache keys and DNS/IP checks so trusted fake-IP proxy stacks using `fc00::/7` can work without broad private-network access. Fixes #74351. Thanks @jeffrey701.
3031
- OpenAI Codex: restore `/verbose full` persistence and app-server tool-output forwarding, and retry Gateway E2E temp-home cleanup so debug runs do not regress on stale validation or cleanup flakes. Thanks @vincentkoc.
3132
- Anthropic/Meridian: preserve text and thinking content seeded on `content_block_start` in anthropic-messages streams, so `[thinking, text]` replies no longer persist as empty turns or trigger empty-response fallbacks. Fixes #74410. Thanks @vyctorbrzezowski.

‎extensions/codex/src/app-server/run-attempt.ts‎

Lines changed: 19 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import {
2828
registerNativeHookRelay,
2929
setActiveEmbeddedRun,
3030
supportsModelTools,
31+
runAgentCleanupStep,
3132
type AgentMessage,
3233
type EmbeddedRunAttemptParams,
3334
type EmbeddedRunAttemptResult,
@@ -682,7 +683,15 @@ export async function runCodexAppServerAttempt(
682683
notificationCleanup();
683684
requestCleanup();
684685
nativeHookRelay?.unregister();
685-
await trajectoryRecorder?.flush();
686+
await runAgentCleanupStep({
687+
runId: params.runId,
688+
sessionId: params.sessionId,
689+
step: "codex-trajectory-flush-startup-failure",
690+
log: embeddedAgentLog,
691+
cleanup: async () => {
692+
await trajectoryRecorder?.flush();
693+
},
694+
});
686695
params.abortSignal?.removeEventListener("abort", abortFromUpstream);
687696
throw error;
688697
}
@@ -886,7 +895,15 @@ export async function runCodexAppServerAttempt(
886895
aborted: runAbortController.signal.aborted,
887896
});
888897
}
889-
await trajectoryRecorder?.flush();
898+
await runAgentCleanupStep({
899+
runId: params.runId,
900+
sessionId: params.sessionId,
901+
step: "codex-trajectory-flush",
902+
log: embeddedAgentLog,
903+
cleanup: async () => {
904+
await trajectoryRecorder?.flush();
905+
},
906+
});
890907
userInputBridge?.cancelPending();
891908
clearTimeout(timeout);
892909
clearTurnCompletionIdleTimer();

‎src/agents/pi-embedded-runner/run.ts‎

Lines changed: 62 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import { formatErrorMessage } from "../../infra/errors.js";
1212
import { getGlobalHookRunner } from "../../plugins/hook-runner-global.js";
1313
import { resolveProviderAuthProfileId } from "../../plugins/provider-runtime.js";
1414
import { enqueueCommandInLane } from "../../process/command-queue.js";
15+
import type { CommandQueueEnqueueOptions } from "../../process/command-queue.types.js";
1516
import { normalizeOptionalString } from "../../shared/string-coerce.js";
1617
import { sanitizeForLog } from "../../terminal/ansi.js";
1718
import { resolveUserPath } from "../../utils.js";
@@ -76,6 +77,7 @@ import {
7677
pickFallbackThinkingLevel,
7778
} from "../pi-embedded-helpers.js";
7879
import { resolveProviderIdForAuth } from "../provider-auth-aliases.js";
80+
import { runAgentCleanupStep } from "../run-cleanup-timeout.js";
7981
import { buildAgentRuntimeAuthPlan } from "../runtime-plan/auth.js";
8082
import { buildAgentRuntimePlan } from "../runtime-plan/build.js";
8183
import { ensureRuntimePluginsLoaded } from "../runtime-plugins.js";
@@ -159,8 +161,26 @@ import { createUsageAccumulator, mergeUsageIntoAccumulator } from "./usage-accum
159161
type ApiKeyInfo = ResolvedProviderAuth;
160162

161163
const MAX_SAME_MODEL_IDLE_TIMEOUT_RETRIES = 1;
164+
const EMBEDDED_RUN_LANE_TIMEOUT_GRACE_MS = 30_000;
162165
type EmbeddedRunAttemptForRunner = Awaited<ReturnType<typeof runEmbeddedAttemptWithBackend>>;
163166

167+
function resolveEmbeddedRunLaneTimeoutMs(timeoutMs: number): number | undefined {
168+
if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) {
169+
return undefined;
170+
}
171+
return Math.floor(timeoutMs) + EMBEDDED_RUN_LANE_TIMEOUT_GRACE_MS;
172+
}
173+
174+
function withEmbeddedRunLaneTimeout(
175+
opts: CommandQueueEnqueueOptions | undefined,
176+
laneTaskTimeoutMs: number | undefined,
177+
): CommandQueueEnqueueOptions | undefined {
178+
if (laneTaskTimeoutMs === undefined || opts?.taskTimeoutMs !== undefined) {
179+
return opts;
180+
}
181+
return { ...opts, taskTimeoutMs: laneTaskTimeoutMs };
182+
}
183+
164184
function normalizeEmbeddedRunAttemptResult(
165185
attempt: EmbeddedRunAttemptForRunner,
166186
): EmbeddedRunAttemptForRunner {
@@ -292,10 +312,15 @@ export async function runEmbeddedPiAgent(
292312
}
293313
const sessionLane = resolveSessionLane(params.sessionKey?.trim() || params.sessionId);
294314
const globalLane = resolveGlobalLane(params.lane);
295-
const enqueueGlobal =
296-
params.enqueue ?? ((task, opts) => enqueueCommandInLane(globalLane, task, opts));
297-
const enqueueSession =
298-
params.enqueue ?? ((task, opts) => enqueueCommandInLane(sessionLane, task, opts));
315+
const laneTaskTimeoutMs = resolveEmbeddedRunLaneTimeoutMs(params.timeoutMs);
316+
const withLaneTimeout = (opts?: CommandQueueEnqueueOptions) =>
317+
withEmbeddedRunLaneTimeout(opts, laneTaskTimeoutMs);
318+
const enqueueGlobal = <T>(task: () => Promise<T>, opts?: CommandQueueEnqueueOptions) =>
319+
params.enqueue
320+
? params.enqueue(task, withLaneTimeout(opts))
321+
: enqueueCommandInLane(globalLane, task, withLaneTimeout(opts));
322+
const enqueueSession = <T>(task: () => Promise<T>, opts?: CommandQueueEnqueueOptions) =>
323+
params.enqueue ? params.enqueue(task, opts) : enqueueCommandInLane(sessionLane, task, opts);
299324
const channelHint = params.messageChannel ?? params.messageProvider;
300325
const resolvedToolResultFormat =
301326
params.toolResultFormat ??
@@ -2489,26 +2514,42 @@ export async function runEmbeddedPiAgent(
24892514
}
24902515
} finally {
24912516
forgetPromptBuildDrainCacheForRun(params.runId);
2492-
await contextEngine.dispose?.();
24932517
stopRuntimeAuthRefreshTimer();
2518+
await runAgentCleanupStep({
2519+
runId: params.runId,
2520+
sessionId: params.sessionId,
2521+
step: "context-engine-dispose",
2522+
log,
2523+
cleanup: async () => {
2524+
await contextEngine.dispose?.();
2525+
},
2526+
});
24942527
if (params.cleanupBundleMcpOnRunEnd === true) {
2495-
const onError = (error: unknown, sessionId: string) => {
2496-
log.warn(
2497-
`bundle-mcp cleanup failed after run for ${sessionId}: ${formatErrorMessage(error)}`,
2498-
);
2499-
};
2500-
const retiredBySessionKey = await retireSessionMcpRuntimeForSessionKey({
2501-
sessionKey: params.sessionKey,
2502-
reason: "embedded-run-end",
2503-
onError,
2528+
await runAgentCleanupStep({
2529+
runId: params.runId,
2530+
sessionId: params.sessionId,
2531+
step: "bundle-mcp-retire",
2532+
log,
2533+
cleanup: async () => {
2534+
const onError = (error: unknown, sessionId: string) => {
2535+
log.warn(
2536+
`bundle-mcp cleanup failed after run for ${sessionId}: ${formatErrorMessage(error)}`,
2537+
);
2538+
};
2539+
const retiredBySessionKey = await retireSessionMcpRuntimeForSessionKey({
2540+
sessionKey: params.sessionKey,
2541+
reason: "embedded-run-end",
2542+
onError,
2543+
});
2544+
if (!retiredBySessionKey) {
2545+
await retireSessionMcpRuntime({
2546+
sessionId: params.sessionId,
2547+
reason: "embedded-run-end",
2548+
onError,
2549+
});
2550+
}
2551+
},
25042552
});
2505-
if (!retiredBySessionKey) {
2506-
await retireSessionMcpRuntime({
2507-
sessionId: params.sessionId,
2508-
reason: "embedded-run-end",
2509-
onError,
2510-
});
2511-
}
25122553
}
25132554
}
25142555
});

‎src/agents/pi-embedded-runner/run/attempt.ts‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,7 @@ import {
119119
import { wrapStreamFnTextTransforms } from "../../plugin-text-transforms.js";
120120
import { describeProviderRequestRoutingSummary } from "../../provider-attribution.js";
121121
import { registerProviderStreamForModel } from "../../provider-stream.js";
122+
import { runAgentCleanupStep } from "../../run-cleanup-timeout.js";
122123
import { collectRuntimeChannelCapabilities } from "../../runtime-capabilities.js";
123124
import {
124125
logAgentRuntimeToolDiagnostics,
@@ -3299,7 +3300,15 @@ export async function runEmbeddedAttempt(
32993300
promptError: promptError ? formatErrorMessage(promptError) : undefined,
33003301
});
33013302
}
3302-
await trajectoryRecorder?.flush();
3303+
await runAgentCleanupStep({
3304+
runId: params.runId,
3305+
sessionId: params.sessionId,
3306+
step: "pi-trajectory-flush",
3307+
log,
3308+
cleanup: async () => {
3309+
await trajectoryRecorder?.flush();
3310+
},
3311+
});
33033312
// Always tear down the session (and release the lock) before we leave this attempt.
33043313
//
33053314
// BUGFIX: Wait for the agent to be truly idle before flushing pending tool results.

‎src/agents/run-cleanup-timeout.test.ts‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
2+
import { AGENT_CLEANUP_STEP_TIMEOUT_MS, runAgentCleanupStep } from "./run-cleanup-timeout.js";
3+
4+
describe("agent cleanup timeout", () => {
5+
const log = {
6+
warn: vi.fn(),
7+
};
8+
9+
beforeEach(() => {
10+
vi.useFakeTimers();
11+
log.warn.mockClear();
12+
});
13+
14+
afterEach(() => {
15+
vi.useRealTimers();
16+
});
17+
18+
it("returns after the cleanup timeout when a cleanup step stalls", async () => {
19+
const cleanup = vi.fn(async () => new Promise<never>(() => {}));
20+
21+
const result = runAgentCleanupStep({
22+
runId: "run-1",
23+
sessionId: "session-1",
24+
step: "bundle-mcp-retire",
25+
cleanup,
26+
log,
27+
});
28+
29+
await vi.advanceTimersByTimeAsync(AGENT_CLEANUP_STEP_TIMEOUT_MS);
30+
await expect(result).resolves.toBeUndefined();
31+
32+
expect(cleanup).toHaveBeenCalledTimes(1);
33+
expect(log.warn).toHaveBeenCalledWith(expect.stringContaining("agent cleanup timed out:"));
34+
expect(log.warn).toHaveBeenCalledWith(expect.stringContaining("step=bundle-mcp-retire"));
35+
});
36+
37+
it("logs cleanup rejection without throwing", async () => {
38+
await expect(
39+
runAgentCleanupStep({
40+
runId: "run-2",
41+
sessionId: "session-2",
42+
step: "context-engine-dispose",
43+
cleanup: async () => {
44+
throw new Error("dispose failed");
45+
},
46+
log,
47+
}),
48+
).resolves.toBeUndefined();
49+
50+
expect(log.warn).toHaveBeenCalledWith(expect.stringContaining("agent cleanup failed:"));
51+
expect(log.warn).toHaveBeenCalledWith(expect.stringContaining("dispose failed"));
52+
});
53+
});

‎src/agents/run-cleanup-timeout.ts‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
import { formatErrorMessage } from "../infra/errors.js";
2+
3+
export const AGENT_CLEANUP_STEP_TIMEOUT_MS = 10_000;
4+
5+
export type AgentCleanupLogger = {
6+
warn: (message: string) => void;
7+
};
8+
9+
export async function runAgentCleanupStep(params: {
10+
runId: string;
11+
sessionId: string;
12+
step: string;
13+
cleanup: () => Promise<void>;
14+
log: AgentCleanupLogger;
15+
timeoutMs?: number;
16+
}): Promise<void> {
17+
const timeoutMs = Math.max(1, Math.floor(params.timeoutMs ?? AGENT_CLEANUP_STEP_TIMEOUT_MS));
18+
let timeoutHandle: ReturnType<typeof setTimeout> | undefined;
19+
let timedOut = false;
20+
const cleanupPromise = Promise.resolve().then(params.cleanup);
21+
const observedCleanupPromise = cleanupPromise.catch((error) => {
22+
if (!timedOut) {
23+
params.log.warn(
24+
`agent cleanup failed: runId=${params.runId} sessionId=${params.sessionId} step=${params.step} error=${formatErrorMessage(error)}`,
25+
);
26+
}
27+
});
28+
const timeoutPromise = new Promise<"timeout">((resolve) => {
29+
timeoutHandle = setTimeout(() => {
30+
timedOut = true;
31+
resolve("timeout");
32+
}, timeoutMs);
33+
timeoutHandle.unref?.();
34+
});
35+
const result = await Promise.race([
36+
observedCleanupPromise.then(() => "done" as const),
37+
timeoutPromise,
38+
]);
39+
if (timeoutHandle) {
40+
clearTimeout(timeoutHandle);
41+
}
42+
if (result === "timeout") {
43+
params.log.warn(
44+
`agent cleanup timed out: runId=${params.runId} sessionId=${params.sessionId} step=${params.step} timeoutMs=${timeoutMs}`,
45+
);
46+
void cleanupPromise.catch((error) => {
47+
params.log.warn(
48+
`agent cleanup rejected after timeout: runId=${params.runId} sessionId=${params.sessionId} step=${params.step} error=${formatErrorMessage(error)}`,
49+
);
50+
});
51+
}
52+
}

‎src/plugin-sdk/agent-harness-runtime.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@ export { VERSION as OPENCLAW_VERSION } from "../version.js";
6060
export { formatErrorMessage } from "../infra/errors.js";
6161
export { formatApprovalDisplayPath } from "../infra/approval-display-paths.js";
6262
export { emitAgentEvent, onAgentEvent, resetAgentEventsForTest } from "../infra/agent-events.js";
63+
export { runAgentCleanupStep } from "../agents/run-cleanup-timeout.js";
6364
export { log as embeddedAgentLog } from "../agents/pi-embedded-runner/logger.js";
6465
export { buildAgentRuntimePlan } from "../agents/runtime-plan/build.js";
6566
export { classifyEmbeddedPiRunResultForModelFallback } from "../agents/pi-embedded-runner/result-fallback-classifier.js";

‎src/process/command-queue.test.ts‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ type CommandQueueModule = typeof import("./command-queue.js");
2222

2323
let clearCommandLane: CommandQueueModule["clearCommandLane"];
2424
let CommandLaneClearedError: CommandQueueModule["CommandLaneClearedError"];
25+
let CommandLaneTaskTimeoutError: CommandQueueModule["CommandLaneTaskTimeoutError"];
2526
let enqueueCommand: CommandQueueModule["enqueueCommand"];
2627
let enqueueCommandInLane: CommandQueueModule["enqueueCommandInLane"];
2728
let GatewayDrainingError: CommandQueueModule["GatewayDrainingError"];
@@ -63,6 +64,7 @@ describe("command queue", () => {
6364
({
6465
clearCommandLane,
6566
CommandLaneClearedError,
67+
CommandLaneTaskTimeoutError,
6668
enqueueCommand,
6769
enqueueCommandInLane,
6870
GatewayDrainingError,
@@ -326,6 +328,42 @@ describe("command queue", () => {
326328
await expect(other).resolves.toBe("other");
327329
});
328330

331+
it("task timeout releases a stuck lane and drains queued work", async () => {
332+
const lane = `timeout-lane-${Date.now()}-${Math.random().toString(16).slice(2)}`;
333+
setCommandLaneConcurrency(lane, 1);
334+
335+
vi.useFakeTimers();
336+
try {
337+
const first = enqueueCommandInLane(lane, async () => new Promise<never>(() => {}), {
338+
taskTimeoutMs: 25,
339+
});
340+
const firstRejected = expect(first).rejects.toBeInstanceOf(CommandLaneTaskTimeoutError);
341+
let secondRan = false;
342+
const second = enqueueCommandInLane(lane, async () => {
343+
secondRan = true;
344+
return "second";
345+
});
346+
347+
expect(secondRan).toBe(false);
348+
expect(getCommandLaneSnapshot(lane)).toMatchObject({
349+
activeCount: 1,
350+
queuedCount: 1,
351+
});
352+
353+
await vi.advanceTimersByTimeAsync(25);
354+
355+
await firstRejected;
356+
await expect(second).resolves.toBe("second");
357+
expect(secondRan).toBe(true);
358+
expect(getCommandLaneSnapshot(lane)).toMatchObject({
359+
activeCount: 0,
360+
queuedCount: 0,
361+
});
362+
} finally {
363+
vi.useRealTimers();
364+
}
365+
});
366+
329367
it("getCommandLaneSnapshot reports active and queued work for one lane", async () => {
330368
const lane = `snapshot-lane-${Date.now()}-${Math.random().toString(16).slice(2)}`;
331369
setCommandLaneConcurrency(lane, 1);

0 commit comments

Comments
 (0)