Skip to content

Commit 3b04ea6

Browse files
vincentkocvincentkoc
committed
fix(cron): release cancelled embedded lanes
1 parent ebb60a0 commit 3b04ea6

7 files changed

Lines changed: 401 additions & 13 deletions

File tree

‎src/agents/embedded-agent-runner/run.compaction-loop-guard.test.ts‎

Lines changed: 200 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -187,6 +187,206 @@ describe("post-compaction loop guard wired into runEmbeddedAgent", () => {
187187
expect(attemptSignalReason).toBeInstanceOf(PostCompactionLoopPersistedError);
188188
});
189189

190+
it("releases the lane after a post-compaction abort when the backend ignores cancellation", async () => {
191+
vi.useFakeTimers();
192+
let settleIgnoredAttempt: ((value: ReturnType<typeof makeAttemptResult>) => void) | undefined;
193+
let resolveAttemptAborted: (() => void) | undefined;
194+
const attemptAbortedPromise = new Promise<void>((resolve) => {
195+
resolveAttemptAborted = resolve;
196+
});
197+
try {
198+
const overflowError = makeOverflowError();
199+
let attemptAborted = false;
200+
mockedRunEmbeddedAttempt.mockResolvedValueOnce(
201+
makeAttemptResult({ promptError: overflowError }),
202+
);
203+
mockedRunEmbeddedAttempt.mockImplementationOnce(async (attemptParams: unknown) => {
204+
const { abortSignal, onToolOutcome } = attemptParams as {
205+
abortSignal?: AbortSignal;
206+
onToolOutcome?: ToolOutcomeObserver;
207+
};
208+
for (let i = 0; i < 3; i += 1) {
209+
await executeWrappedToolOutcome(
210+
"gateway",
211+
{ action: "lookup", path: "x" },
212+
"identical-result",
213+
onToolOutcome,
214+
);
215+
}
216+
attemptAborted = abortSignal?.aborted ?? false;
217+
resolveAttemptAborted?.();
218+
return await new Promise((resolve) => {
219+
settleIgnoredAttempt = resolve;
220+
});
221+
});
222+
mockedCompactDirect.mockResolvedValueOnce(
223+
makeCompactionSuccess({
224+
summary: "Compacted session",
225+
firstKeptEntryId: "entry-5",
226+
tokensBefore: 150000,
227+
}),
228+
);
229+
230+
const run = runEmbeddedAgent({
231+
...baseParams,
232+
runId: "run-post-compaction-abort-lane-release",
233+
timeoutMs: 48 * 60 * 60 * 1000,
234+
});
235+
let settled = false;
236+
void run
237+
.finally(() => {
238+
settled = true;
239+
})
240+
.catch(() => {});
241+
242+
await attemptAbortedPromise;
243+
expect(attemptAborted).toBe(true);
244+
await vi.advanceTimersByTimeAsync(30_001);
245+
246+
expect(settled).toBe(true);
247+
await expect(run).rejects.toMatchObject({ name: "CommandLaneTaskTimeoutError" });
248+
} finally {
249+
settleIgnoredAttempt?.(makeAttemptResult());
250+
vi.useRealTimers();
251+
}
252+
});
253+
254+
it("keeps a native lane alive while a tool is still running", async () => {
255+
vi.useFakeTimers();
256+
let settleAttempt: ((value: ReturnType<typeof makeAttemptResult>) => void) | undefined;
257+
let resolveAttemptStarted: (() => void) | undefined;
258+
const attemptStarted = new Promise<void>((resolve) => {
259+
resolveAttemptStarted = resolve;
260+
});
261+
try {
262+
mockedRunEmbeddedAttempt.mockImplementationOnce(async (attemptParams: unknown) => {
263+
const { onAttemptTimeoutArmed } = attemptParams as {
264+
onAttemptTimeoutArmed?: () => void;
265+
};
266+
resolveAttemptStarted?.();
267+
onAttemptTimeoutArmed?.();
268+
return await new Promise((resolve) => {
269+
settleAttempt = resolve;
270+
});
271+
});
272+
273+
const run = runEmbeddedAgent({
274+
...baseParams,
275+
runId: "run-native-tool-heartbeat",
276+
timeoutMs: 1,
277+
agentHarnessRuntimeOverride: "openclaw",
278+
});
279+
let settled = false;
280+
void run
281+
.finally(() => {
282+
settled = true;
283+
})
284+
.catch(() => {});
285+
286+
await vi.advanceTimersByTimeAsync(0);
287+
await attemptStarted;
288+
await vi.advanceTimersByTimeAsync(30_001);
289+
290+
expect(settled).toBe(false);
291+
settleAttempt?.(makeAttemptResult());
292+
await expect(run).resolves.toMatchObject({ meta: { aborted: false } });
293+
} finally {
294+
vi.useRealTimers();
295+
}
296+
});
297+
298+
it("releases a native lane when a timed-out attempt ignores cancellation", async () => {
299+
vi.useFakeTimers();
300+
let settleAttempt: ((value: ReturnType<typeof makeAttemptResult>) => void) | undefined;
301+
let resolveAttemptStarted: (() => void) | undefined;
302+
const attemptStarted = new Promise<void>((resolve) => {
303+
resolveAttemptStarted = resolve;
304+
});
305+
try {
306+
mockedRunEmbeddedAttempt.mockImplementationOnce(async (attemptParams: unknown) => {
307+
const { onAttemptTimeoutArmed, onAttemptTimeout } = attemptParams as {
308+
onAttemptTimeoutArmed?: () => void;
309+
onAttemptTimeout?: (reason: Error) => void;
310+
};
311+
resolveAttemptStarted?.();
312+
onAttemptTimeoutArmed?.();
313+
onAttemptTimeout?.(new Error("attempt timed out"));
314+
return await new Promise((resolve) => {
315+
settleAttempt = resolve;
316+
});
317+
});
318+
319+
const run = runEmbeddedAgent({
320+
...baseParams,
321+
runId: "run-native-timeout-lane-release",
322+
timeoutMs: 48 * 60 * 60 * 1000,
323+
agentHarnessRuntimeOverride: "openclaw",
324+
});
325+
let settled = false;
326+
void run
327+
.finally(() => {
328+
settled = true;
329+
})
330+
.catch(() => {});
331+
332+
await vi.advanceTimersByTimeAsync(0);
333+
await attemptStarted;
334+
await vi.advanceTimersByTimeAsync(30_001);
335+
336+
expect(settled).toBe(true);
337+
await expect(run).rejects.toMatchObject({ name: "CommandLaneTaskTimeoutError" });
338+
} finally {
339+
settleAttempt?.(makeAttemptResult());
340+
vi.useRealTimers();
341+
}
342+
});
343+
344+
it("releases a native lane when an explicit abort ignores cancellation", async () => {
345+
vi.useFakeTimers();
346+
let settleAttempt: ((value: ReturnType<typeof makeAttemptResult>) => void) | undefined;
347+
let resolveAttemptStarted: (() => void) | undefined;
348+
const attemptStarted = new Promise<void>((resolve) => {
349+
resolveAttemptStarted = resolve;
350+
});
351+
try {
352+
mockedRunEmbeddedAttempt.mockImplementationOnce(async (attemptParams: unknown) => {
353+
const { onAttemptTimeoutArmed, onAttemptAbort } = attemptParams as {
354+
onAttemptTimeoutArmed?: () => void;
355+
onAttemptAbort?: () => void;
356+
};
357+
resolveAttemptStarted?.();
358+
onAttemptTimeoutArmed?.();
359+
onAttemptAbort?.();
360+
return await new Promise((resolve) => {
361+
settleAttempt = resolve;
362+
});
363+
});
364+
365+
const run = runEmbeddedAgent({
366+
...baseParams,
367+
runId: "run-native-abort-lane-release",
368+
timeoutMs: 48 * 60 * 60 * 1000,
369+
agentHarnessRuntimeOverride: "openclaw",
370+
});
371+
let settled = false;
372+
void run
373+
.finally(() => {
374+
settled = true;
375+
})
376+
.catch(() => {});
377+
378+
await vi.advanceTimersByTimeAsync(0);
379+
await attemptStarted;
380+
await vi.advanceTimersByTimeAsync(30_001);
381+
382+
expect(settled).toBe(true);
383+
await expect(run).rejects.toMatchObject({ name: "CommandLaneTaskTimeoutError" });
384+
} finally {
385+
settleAttempt?.(makeAttemptResult());
386+
vi.useRealTimers();
387+
}
388+
});
389+
190390
it("does not abort when the result hash changes across post-compaction attempts (progress was made)", async () => {
191391
const overflowError = makeOverflowError();
192392
// Attempt 1: overflow → triggers compaction.

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

Lines changed: 62 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -237,6 +237,7 @@ type ApiKeyInfo = ResolvedProviderAuth;
237237

238238
const MAX_SAME_MODEL_IDLE_TIMEOUT_RETRIES = 1;
239239
const EMBEDDED_RUN_LANE_TIMEOUT_GRACE_MS = 30_000;
240+
const EMBEDDED_RUN_LANE_HEARTBEAT_MS = EMBEDDED_RUN_LANE_TIMEOUT_GRACE_MS / 2;
240241
const MID_TURN_PRECHECK_CONTINUATION_PROMPT =
241242
"Continue from the current transcript after the latest tool result. Do not repeat the original user request, and do not rerun completed tools unless the transcript shows they are still needed.";
242243
const COMPACTION_CONTINUATION_RETRY_INSTRUCTION =
@@ -637,6 +638,8 @@ async function runEmbeddedAgentInternal(
637638
};
638639
const sessionQueuePriority = resolveEmbeddedRunSessionQueuePriority(params.trigger);
639640
const laneTaskTimeoutMs = resolveEmbeddedRunLaneTimeoutMs(params.timeoutMs);
641+
const laneTaskAbortController = new AbortController();
642+
const laneTaskReleaseController = new AbortController();
640643
let laneTaskProgressAtMs = Date.now();
641644
const noteLaneTaskProgress = () => {
642645
laneTaskProgressAtMs = Date.now();
@@ -661,6 +664,9 @@ async function runEmbeddedAgentInternal(
661664
{
662665
...opts,
663666
taskTimeoutProgressAtMs: () => laneTaskProgressAtMs,
667+
taskTimeoutAbortSignal: laneTaskAbortController.signal,
668+
taskTimeoutAbortGraceMs: EMBEDDED_RUN_LANE_TIMEOUT_GRACE_MS,
669+
taskTimeoutReleaseSignal: laneTaskReleaseController.signal,
664670
},
665671
laneTaskTimeoutMs,
666672
);
@@ -1480,6 +1486,7 @@ async function runEmbeddedAgentInternal(
14801486
const verdict = postCompactionGuard.observe(observation);
14811487
if (verdict.shouldAbort) {
14821488
postCompactionAbortError ??= PostCompactionLoopPersistedError.fromVerdict(verdict);
1489+
laneTaskAbortController.abort(postCompactionAbortError);
14831490
postCompactionAbortController?.abort(postCompactionAbortError);
14841491
}
14851492
};
@@ -1846,17 +1853,56 @@ async function runEmbeddedAgentInternal(
18461853
postCompactionAbortController = attemptAbortController;
18471854
const parentAbortSignal = params.abortSignal;
18481855
const relayParentAbort = (): void => {
1856+
laneTaskAbortController.abort(parentAbortSignal?.reason);
18491857
attemptAbortController.abort(parentAbortSignal?.reason);
18501858
};
18511859
if (parentAbortSignal?.aborted) {
18521860
relayParentAbort();
18531861
} else {
18541862
parentAbortSignal?.addEventListener("abort", relayParentAbort, { once: true });
18551863
}
1856-
// Periodically report progress during long-running tool execution
1857-
// so the lane timeout does not expire while a tool is still running
1858-
// (e.g., exec commands taking >5 minutes) (#94033).
1859-
const progressInterval = setInterval(() => noteLaneTaskProgress(), 30_000);
1864+
// Native attempts start the heartbeat only after their own timeout
1865+
// watchdog is armed, keeping preflight inside the requested deadline.
1866+
let progressInterval: ReturnType<typeof setInterval> | undefined;
1867+
const stopLaneProgressHeartbeat = () => {
1868+
if (progressInterval) {
1869+
clearInterval(progressInterval);
1870+
progressInterval = undefined;
1871+
}
1872+
attemptAbortController.signal.removeEventListener("abort", stopLaneProgressHeartbeat);
1873+
};
1874+
const startLaneProgressHeartbeat = () => {
1875+
if (progressInterval || attemptAbortController.signal.aborted) {
1876+
return;
1877+
}
1878+
progressInterval = setInterval(
1879+
() => noteLaneTaskProgress(),
1880+
EMBEDDED_RUN_LANE_HEARTBEAT_MS,
1881+
);
1882+
progressInterval.unref?.();
1883+
attemptAbortController.signal.addEventListener("abort", stopLaneProgressHeartbeat, {
1884+
once: true,
1885+
});
1886+
};
1887+
// Timeout recovery can continue after an attempt returns, but a native
1888+
// transport that ignores its timeout releases the lane after one grace.
1889+
let timeoutReleaseTimer: ReturnType<typeof setTimeout> | undefined;
1890+
const clearAttemptTimeoutRelease = () => {
1891+
if (timeoutReleaseTimer) {
1892+
clearTimeout(timeoutReleaseTimer);
1893+
timeoutReleaseTimer = undefined;
1894+
}
1895+
};
1896+
const armAttemptTimeoutRelease = (reason: Error) => {
1897+
if (timeoutReleaseTimer) {
1898+
return;
1899+
}
1900+
timeoutReleaseTimer = setTimeout(
1901+
() => laneTaskReleaseController.abort(reason),
1902+
EMBEDDED_RUN_LANE_TIMEOUT_GRACE_MS,
1903+
);
1904+
timeoutReleaseTimer.unref?.();
1905+
};
18601906
const rawAttempt = await runEmbeddedAttemptWithBackend({
18611907
sessionId: activeSessionId,
18621908
sessionKey: resolvedSessionKey,
@@ -1960,6 +2006,16 @@ async function runEmbeddedAgentInternal(
19602006
runId: params.runId,
19612007
lifecycleGeneration,
19622008
abortSignal: attemptAbortController.signal,
2009+
onAttemptTimeoutArmed: pluginHarnessOwnsTransport
2010+
? undefined
2011+
: startLaneProgressHeartbeat,
2012+
onAttemptTimeout: pluginHarnessOwnsTransport ? undefined : armAttemptTimeoutRelease,
2013+
onAttemptAbort: pluginHarnessOwnsTransport
2014+
? undefined
2015+
: () => {
2016+
stopLaneProgressHeartbeat();
2017+
laneTaskAbortController.abort();
2018+
},
19632019
replyOperation: params.replyOperation,
19642020
shouldEmitToolResult: params.shouldEmitToolResult,
19652021
shouldEmitToolOutput: params.shouldEmitToolOutput,
@@ -2015,7 +2071,8 @@ async function runEmbeddedAgentInternal(
20152071
throw postCompactionAbortError ?? err;
20162072
})
20172073
.finally(() => {
2018-
clearInterval(progressInterval);
2074+
clearAttemptTimeoutRelease();
2075+
stopLaneProgressHeartbeat();
20192076
parentAbortSignal?.removeEventListener?.("abort", relayParentAbort);
20202077
if (postCompactionAbortController === attemptAbortController) {
20212078
postCompactionAbortController = undefined;

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

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3388,7 +3388,9 @@ export async function runEmbeddedAttempt(
33883388
}
33893389
}
33903390
if (isTimeout) {
3391-
runAbortController.abort(reason ?? makeTimeoutAbortReason());
3391+
const timeoutReason = reason instanceof Error ? reason : makeTimeoutAbortReason();
3392+
params.onAttemptTimeout?.(timeoutReason);
3393+
runAbortController.abort(timeoutReason);
33923394
} else {
33933395
runAbortController.abort(reason);
33943396
}
@@ -3698,6 +3700,7 @@ export async function runEmbeddedAttempt(
36983700

36993701
const abortActiveRunExternally = (reason?: "user_abort" | "restart" | "superseded") => {
37003702
externalAbort = true;
3703+
params.onAttemptAbort?.();
37013704
abortRun(false, reason === "restart" ? createAgentRunRestartAbortError() : undefined);
37023705
};
37033706
const queueHandle: EmbeddedAgentQueueHandle & {
@@ -3793,6 +3796,7 @@ export async function runEmbeddedAttempt(
37933796
);
37943797
};
37953798
scheduleAbortTimer(params.timeoutMs, "initial");
3799+
params.onAttemptTimeoutArmed?.();
37963800

37973801
let messagesSnapshot: AgentMessage[] = [];
37983802
let sessionIdUsed = activeSession.sessionId;

‎src/agents/embedded-agent-runner/run/types.ts‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,12 @@ export type EmbeddedRunAttemptParams = EmbeddedRunAttemptBase & {
7373
agentHarnessTaskRuntimeScope?: AgentHarnessTaskRuntimeScope;
7474
/** Live observer called after wrapped tool outcomes are recorded. */
7575
onToolOutcome?: ToolOutcomeObserver;
76+
/** Signals that the attempt's own run-timeout watchdog is active. */
77+
onAttemptTimeoutArmed?: () => void;
78+
/** Signals that this attempt's timeout has fired and must unwind promptly. */
79+
onAttemptTimeout?: (reason: Error) => void;
80+
/** Signals an explicit cancellation through the active native run handle. */
81+
onAttemptAbort?: () => void;
7682
/** Supplies run-global model-call ordering for parallel tool outcomes. */
7783
allocateToolOutcomeOrdinal?: (toolCallId?: string) => number;
7884
model: Model;

0 commit comments

Comments
 (0)