Skip to content

Commit 23e768e

Browse files
wangmiao0668000666wangmiao0668000666
committed
fix(agents): keep lane-task timeout alive during long-running tools (fixes #94033)
1 parent 0cae5b3 commit 23e768e

4 files changed

Lines changed: 288 additions & 0 deletions

File tree

‎scripts/repro/issue-94033-cron-lane-timeout.mts‎

Lines changed: 113 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,113 @@
1+
#!/usr/bin/env node
2+
// Standalone real-environment proof for #94033.
3+
//
4+
// Reproduces the lane-task timeout sliding window from
5+
// `src/process/command-queue.ts`. A task that takes longer than the
6+
// configured `taskTimeoutMs` without reporting progress triggers
7+
// `CommandLaneTaskTimeoutError`.
8+
//
9+
// With the fix (heartbeat helper applied), `noteProgress` is invoked
10+
// periodically while the task is in flight, so the sliding window
11+
// stays fresh and the task completes successfully.
12+
//
13+
// Without the fix, the task exceeds `taskTimeoutMs` and the timeout
14+
// fires.
15+
//
16+
// Run: node --import tsx scripts/repro/issue-94033-cron-lane-timeout.mts
17+
import assert from "node:assert/strict";
18+
import {
19+
CommandLaneTaskTimeoutError,
20+
enqueueCommandInLane,
21+
resetCommandLane,
22+
} from "../../src/process/command-queue.ts";
23+
import { startLaneTaskProgressHeartbeat } from "../../src/agents/embedded-agent-runner/lane-heartbeat.ts";
24+
25+
const LANE = "repro-cron-94033";
26+
const TASK_TIMEOUT_MS = 600; // Short timeout for fast reproduction
27+
const HEARTBEAT_INTERVAL_MS = 100; // < grace window, similar pattern to real prod
28+
const TASK_DURATION_MS = 1_500; // > TASK_TIMEOUT_MS, would trigger timeout without heartbeat
29+
30+
async function runScenario({
31+
withHeartbeat,
32+
label,
33+
}: {
34+
withHeartbeat: boolean;
35+
label: string;
36+
}) {
37+
resetCommandLane(LANE);
38+
let progressNotes = 0;
39+
// The progress-at timestamp mirrors the production pattern:
40+
// `let laneTaskProgressAtMs = Date.now(); noteProgress = () => { laneTaskProgressAtMs = Date.now() }`
41+
// The lane-task timeout callback reads this same closure variable.
42+
let lastProgressAtMs = Date.now();
43+
const noteProgress = () => {
44+
lastProgressAtMs = Date.now();
45+
progressNotes += 1;
46+
};
47+
const taskPromise = enqueueCommandInLane(
48+
LANE,
49+
async () => {
50+
const heartbeat = withHeartbeat
51+
? startLaneTaskProgressHeartbeat(noteProgress, HEARTBEAT_INTERVAL_MS)
52+
: undefined;
53+
try {
54+
await new Promise<void>((resolve) => {
55+
setTimeout(resolve, TASK_DURATION_MS);
56+
});
57+
return "completed";
58+
} finally {
59+
heartbeat?.stop();
60+
}
61+
},
62+
{
63+
taskTimeoutMs: TASK_TIMEOUT_MS,
64+
taskTimeoutProgressAtMs: () => lastProgressAtMs,
65+
},
66+
);
67+
68+
const started = Date.now();
69+
let outcome: "completed" | "timed-out";
70+
try {
71+
await taskPromise;
72+
outcome = "completed";
73+
} catch (err) {
74+
if (err instanceof CommandLaneTaskTimeoutError) {
75+
outcome = "timed-out";
76+
} else {
77+
throw err;
78+
}
79+
}
80+
const elapsedMs = Math.round(Date.now() - started);
81+
82+
console.log(`[${label}] outcome=${outcome} elapsedMs=${elapsedMs} noteProgressCalls=${progressNotes}`);
83+
return { outcome, elapsedMs, progressNotes };
84+
}
85+
86+
console.log("=== Reproduction for issue #94033 ===");
87+
console.log(`Lane: ${LANE}`);
88+
console.log(`Configured taskTimeoutMs: ${TASK_TIMEOUT_MS}`);
89+
console.log(`Task runs for: ${TASK_DURATION_MS}ms (longer than timeout)`);
90+
console.log("");
91+
92+
const without = await runScenario({ withHeartbeat: false, label: "without-heartbeat" });
93+
console.log("");
94+
const withFix = await runScenario({ withHeartbeat: true, label: "with-heartbeat " });
95+
96+
console.log("");
97+
console.log("=== Results ===");
98+
console.log(
99+
`Without heartbeat (pre-fix): timed-out after ${without.elapsedMs}ms (timeout fired as expected)`,
100+
);
101+
console.log(
102+
`With heartbeat (post-fix): ${withFix.outcome} after ${withFix.elapsedMs}ms (noteProgress called ${withFix.progressNotes} times)`,
103+
);
104+
105+
assert.equal(without.outcome, "timed-out", "pre-fix scenario should time out");
106+
assert.equal(withFix.outcome, "completed", "post-fix scenario should complete");
107+
assert.ok(
108+
withFix.progressNotes >= 10,
109+
`heartbeat should fire many times during ${TASK_DURATION_MS}ms at ${HEARTBEAT_INTERVAL_MS}ms interval, got ${withFix.progressNotes}`,
110+
);
111+
112+
console.log("");
113+
console.log("PASS: heartbeat keeps the lane-task sliding window alive during long-running work.");

‎src/agents/embedded-agent-runner/lane-heartbeat.test.ts‎

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,103 @@
1+
// Focused tests for the lane-task progress heartbeat helper.
2+
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
3+
import { startLaneTaskProgressHeartbeat, withLaneTaskProgressHeartbeat } from "./lane-heartbeat.js";
4+
5+
describe("startLaneTaskProgressHeartbeat", () => {
6+
it("calls noteProgress at the requested interval", () => {
7+
vi.useFakeTimers();
8+
const noteProgress = vi.fn();
9+
const handle = startLaneTaskProgressHeartbeat(noteProgress, 1_000);
10+
expect(noteProgress).not.toHaveBeenCalled();
11+
vi.advanceTimersByTime(1_000);
12+
expect(noteProgress).toHaveBeenCalledTimes(1);
13+
vi.advanceTimersByTime(2_000);
14+
expect(noteProgress).toHaveBeenCalledTimes(3);
15+
handle.stop();
16+
vi.useRealTimers();
17+
});
18+
19+
it("stops calling noteProgress after stop()", () => {
20+
vi.useFakeTimers();
21+
const noteProgress = vi.fn();
22+
const handle = startLaneTaskProgressHeartbeat(noteProgress, 1_000);
23+
vi.advanceTimersByTime(1_000);
24+
expect(noteProgress).toHaveBeenCalledTimes(1);
25+
handle.stop();
26+
vi.advanceTimersByTime(5_000);
27+
expect(noteProgress).toHaveBeenCalledTimes(1);
28+
vi.useRealTimers();
29+
});
30+
31+
it("stop() is idempotent", () => {
32+
vi.useFakeTimers();
33+
const noteProgress = vi.fn();
34+
const handle = startLaneTaskProgressHeartbeat(noteProgress, 1_000);
35+
handle.stop();
36+
handle.stop();
37+
vi.advanceTimersByTime(5_000);
38+
expect(noteProgress).not.toHaveBeenCalled();
39+
vi.useRealTimers();
40+
});
41+
42+
it("uses a 20s default interval", () => {
43+
vi.useFakeTimers();
44+
const noteProgress = vi.fn();
45+
const handle = startLaneTaskProgressHeartbeat(noteProgress);
46+
vi.advanceTimersByTime(19_999);
47+
expect(noteProgress).not.toHaveBeenCalled();
48+
vi.advanceTimersByTime(1);
49+
expect(noteProgress).toHaveBeenCalledTimes(1);
50+
handle.stop();
51+
vi.useRealTimers();
52+
});
53+
});
54+
55+
describe("withLaneTaskProgressHeartbeat", () => {
56+
beforeEach(() => {
57+
vi.useFakeTimers();
58+
});
59+
afterEach(() => {
60+
vi.useRealTimers();
61+
});
62+
63+
it("stops the heartbeat when the task resolves", async () => {
64+
const noteProgress = vi.fn();
65+
const task = Promise.resolve("done");
66+
const wrapped = withLaneTaskProgressHeartbeat(noteProgress, task, 1_000);
67+
// Let the microtask queue drain so the finally block runs.
68+
await vi.advanceTimersByTimeAsync(0);
69+
await expect(wrapped).resolves.toBe("done");
70+
expect(noteProgress).not.toHaveBeenCalled();
71+
vi.advanceTimersByTime(5_000);
72+
expect(noteProgress).not.toHaveBeenCalled();
73+
});
74+
75+
it("stops the heartbeat when the task rejects", async () => {
76+
const noteProgress = vi.fn();
77+
const failure = new Error("boom");
78+
let rejectTask!: (err: Error) => void;
79+
const task = new Promise<string>((_resolve, reject) => {
80+
rejectTask = reject;
81+
});
82+
const wrapped = withLaneTaskProgressHeartbeat(noteProgress, task, 1_000);
83+
rejectTask(failure);
84+
await expect(wrapped).rejects.toBe(failure);
85+
vi.advanceTimersByTime(5_000);
86+
expect(noteProgress).not.toHaveBeenCalled();
87+
});
88+
89+
it("keeps ticking noteProgress until the task settles", async () => {
90+
const noteProgress = vi.fn();
91+
let resolveTask!: (value: string) => void;
92+
const task = new Promise<string>((resolve) => {
93+
resolveTask = resolve;
94+
});
95+
const wrapped = withLaneTaskProgressHeartbeat(noteProgress, task, 1_000);
96+
vi.advanceTimersByTime(2_500);
97+
expect(noteProgress).toHaveBeenCalledTimes(2);
98+
resolveTask("done");
99+
await expect(wrapped).resolves.toBe("done");
100+
vi.advanceTimersByTime(5_000);
101+
expect(noteProgress).toHaveBeenCalledTimes(2);
102+
});
103+
});

‎src/agents/embedded-agent-runner/lane-heartbeat.ts‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
// Lane-task progress heartbeat.
2+
//
3+
// The command-queue lane-task timeout uses a sliding window: every
4+
// `armTimeout()` cycle re-reads `lastProgressAtMs` (provided by
5+
// `taskTimeoutProgressAtMs` in the queue entry) and rejects with
6+
// `CommandLaneTaskTimeoutError` if `Date.now() - lastProgressAtMs > taskTimeoutMs`.
7+
//
8+
// The embedded runner only refreshes `lastProgressAtMs` via the
9+
// `noteLaneTaskProgress` callback, which fires from
10+
// `notifyExecutionPhase` / `notifyRunProgress` during the runner's own
11+
// lifecycle. Long-running tool execution (e.g. an `exec` that runs for
12+
// several minutes) happens *inside* the embedded attempt and emits no
13+
// runner callbacks, so the sliding window expires even though the task
14+
// is making progress.
15+
//
16+
// This helper bridges that gap by periodically calling a `noteProgress`
17+
// callback while a task is in flight. The default interval is 20s —
18+
// well under the 30s grace window in `command-queue.ts:265`
19+
// (`taskTimeoutMs + 30s`), so at least one `armTimeout` cycle always
20+
// sees fresh progress before deciding to fire.
21+
//
22+
// The interval is `.unref()`'d so the heartbeat never blocks process
23+
// exit. Callers must always invoke `stop()` once the task settles
24+
// (success, error, or abort) so the timer cannot outlive the task it
25+
// is observing.
26+
const DEFAULT_HEARTBEAT_INTERVAL_MS = 20_000;
27+
28+
export type LaneTaskProgressHeartbeat = { stop: () => void };
29+
30+
/**
31+
* Start a periodic heartbeat that calls `noteProgress` every
32+
* `intervalMs` (default 20s). Returns a handle whose `stop()` clears
33+
* the underlying timer; safe to call multiple times.
34+
*
35+
* The interval is `.unref()`'d so the timer never keeps the Node event
36+
* loop alive on its own.
37+
*/
38+
export function startLaneTaskProgressHeartbeat(
39+
noteProgress: () => void,
40+
intervalMs: number = DEFAULT_HEARTBEAT_INTERVAL_MS,
41+
): LaneTaskProgressHeartbeat {
42+
const handle = setInterval(noteProgress, intervalMs);
43+
handle.unref?.();
44+
return {
45+
stop: () => {
46+
clearInterval(handle);
47+
},
48+
};
49+
}
50+
51+
/**
52+
* Run `task` under a periodic progress heartbeat. The heartbeat is
53+
* stopped once the task settles (success or failure), so callers can
54+
* safely `await` the returned promise and never leak the interval.
55+
*/
56+
export function withLaneTaskProgressHeartbeat<T>(
57+
noteProgress: () => void,
58+
task: Promise<T>,
59+
intervalMs: number = DEFAULT_HEARTBEAT_INTERVAL_MS,
60+
): Promise<T> {
61+
const heartbeat = startLaneTaskProgressHeartbeat(noteProgress, intervalMs);
62+
return task.finally(() => {
63+
heartbeat.stop();
64+
});
65+
}

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

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -146,6 +146,7 @@ import {
146146
hasOutboundDeliveryEvidence,
147147
} from "./delivery-evidence.js";
148148
import { resolveEmbeddedRunFailureSignal } from "./failure-signal.js";
149+
import { startLaneTaskProgressHeartbeat } from "./lane-heartbeat.js";
149150
import { resolveGlobalLane, resolveSessionLane } from "./lanes.js";
150151
import { log } from "./logger.js";
151152
import { resolveModelAsync } from "./model.js";
@@ -1853,6 +1854,11 @@ async function runEmbeddedAgentInternal(
18531854
} else {
18541855
parentAbortSignal?.addEventListener("abort", relayParentAbort, { once: true });
18551856
}
1857+
// Keep the lane-task sliding window alive while the embedded attempt
1858+
// is in flight (notably during long-running tool execution such as
1859+
// `exec` commands). Without this heartbeat the lane-task timeout
1860+
// expires even though the attempt is still making progress. See #94033.
1861+
const laneHeartbeat = startLaneTaskProgressHeartbeat(noteLaneTaskProgress);
18561862
const rawAttempt = await runEmbeddedAttemptWithBackend({
18571863
sessionId: activeSessionId,
18581864
sessionKey: resolvedSessionKey,
@@ -2011,6 +2017,7 @@ async function runEmbeddedAgentInternal(
20112017
throw postCompactionAbortError ?? err;
20122018
})
20132019
.finally(() => {
2020+
laneHeartbeat.stop();
20142021
parentAbortSignal?.removeEventListener?.("abort", relayParentAbort);
20152022
if (postCompactionAbortController === attemptAbortController) {
20162023
postCompactionAbortController = undefined;

0 commit comments

Comments
 (0)