Skip to content

Commit 59ddd5f

Browse files
xydt-tanshanshanxydt-tanshanshan
committed
[AI] fix(compaction): pass cancellation-only signal to preflight compaction
Preflight compaction received replyOperation.abortSignal (~60s lifecycle timeout) via compactWithSafetyTimeout's Promise.race, causing slow compactions to be killed before the configured compaction.timeoutSeconds. Create createPreflightCompactionCancelSignal with three-layer protection: 1. abortSignal.reason inspection: propagate explicit cancel, filter TimeoutError (lifecycle) through a one-shot event listener 2. When TimeoutError is filtered, start a polling interval on replyOperation.result (set only on explicit user/restart abort) 3. Check replyOperation.result immediately after construction for the already-aborted case compactWithSafetyTimeout remains the sole timeout owner. Related to #95553
1 parent 33eb6ab commit 59ddd5f

2 files changed

Lines changed: 178 additions & 1 deletion

File tree

‎src/auto-reply/reply/agent-runner-memory.test.ts‎

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ import {
1414
import type { TemplateContext } from "../templating.js";
1515
import type { ReplyPayload } from "../types.js";
1616
import {
17+
createPreflightCompactionCancelSignal,
1718
runMemoryFlushIfNeeded,
1819
runPreflightCompactionIfNeeded,
1920
setAgentRunnerMemoryTestDeps,
@@ -117,6 +118,7 @@ type CompactEmbeddedAgentSessionParams = {
117118
sessionFile?: string;
118119
sessionId?: string;
119120
trigger?: string;
121+
abortSignal?: AbortSignal;
120122
};
121123

122124
function requireRefreshQueuedFollowupSessionCall(index = 0) {
@@ -2394,4 +2396,127 @@ describe("runMemoryFlushIfNeeded", () => {
23942396
expect(flushCall.bootstrapPromptWarningSignaturesSeen).toEqual(["sig-a", "sig-b"]);
23952397
expect(flushCall.bootstrapPromptWarningSignature).toBe("sig-b");
23962398
});
2399+
2400+
it("preflight compaction passes a cancellation-only signal that excludes lifecycle timeout", async () => {
2401+
const sessionFile = path.join(rootDir, "session.jsonl");
2402+
await fs.writeFile(
2403+
sessionFile,
2404+
`${JSON.stringify({ message: { role: "user", content: "x".repeat(5_000) } })}\n`,
2405+
"utf8",
2406+
);
2407+
registerMemoryFlushPlanResolverForTest(() => ({
2408+
softThresholdTokens: 1,
2409+
forceFlushTranscriptBytes: 1_000_000_000,
2410+
reserveTokensFloor: 0,
2411+
prompt: "Pre-compaction memory flush.\nNO_REPLY",
2412+
systemPrompt: "Write memory to memory/YYYY-MM-DD.md.",
2413+
relativePath: "memory/2023-11-14.md",
2414+
}));
2415+
compactEmbeddedAgentSessionMock.mockResolvedValueOnce({
2416+
ok: true,
2417+
compacted: true,
2418+
result: { tokensAfter: 42 },
2419+
});
2420+
const sessionEntry: SessionEntry = {
2421+
sessionId: "session",
2422+
sessionFile,
2423+
updatedAt: Date.now(),
2424+
totalTokens: 120,
2425+
totalTokensFresh: true,
2426+
};
2427+
const replyOperation = createReplyOperation();
2428+
const onCompactionNotice = vi.fn();
2429+
2430+
await runPreflightCompactionIfNeeded({
2431+
cfg: {
2432+
agents: {
2433+
defaults: {
2434+
compaction: {
2435+
memoryFlush: {},
2436+
},
2437+
},
2438+
},
2439+
},
2440+
followupRun: createTestFollowupRun({
2441+
sessionId: "session",
2442+
sessionFile,
2443+
sessionKey: "agent:main:main",
2444+
}),
2445+
defaultModel: "anthropic/claude-opus-4-6",
2446+
agentCfgContextTokens: 100,
2447+
sessionEntry,
2448+
sessionStore: { "agent:main:main": sessionEntry },
2449+
sessionKey: "agent:main:main",
2450+
storePath: path.join(rootDir, "sessions.json"),
2451+
isHeartbeat: false,
2452+
replyOperation,
2453+
onCompactionNotice,
2454+
});
2455+
2456+
expect(compactEmbeddedAgentSessionMock).toHaveBeenCalledTimes(1);
2457+
const compactCall = requireCompactEmbeddedAgentSessionCall();
2458+
expect(compactCall.abortSignal).toBeDefined();
2459+
expect(compactCall.abortSignal).not.toBe(replyOperation.abortSignal);
2460+
});
2461+
});
2462+
2463+
describe("createPreflightCompactionCancelSignal", () => {
2464+
function createTestReplyOperation(
2465+
ctrl: AbortController,
2466+
result?: ReplyOperation["result"],
2467+
): ReplyOperation {
2468+
const op = createReplyOperation();
2469+
Object.defineProperty(op, "abortSignal", {
2470+
value: ctrl.signal,
2471+
writable: false,
2472+
});
2473+
if (result !== undefined) {
2474+
Object.defineProperty(op, "result", {
2475+
value: result,
2476+
writable: true,
2477+
});
2478+
}
2479+
return op;
2480+
}
2481+
2482+
it("aborts on user abort", () => {
2483+
const replyCtrl = new AbortController();
2484+
const replyOperation = createTestReplyOperation(replyCtrl);
2485+
const { signal: cancelSignal } = createPreflightCompactionCancelSignal(replyOperation);
2486+
2487+
replyCtrl.abort(new Error("Reply operation aborted by user"));
2488+
expect(cancelSignal.aborted).toBe(true);
2489+
});
2490+
2491+
it("does not abort on lifecycle timeout (TimeoutError)", () => {
2492+
const replyCtrl = new AbortController();
2493+
const replyOperation = createTestReplyOperation(replyCtrl);
2494+
const { signal: cancelSignal2 } = createPreflightCompactionCancelSignal(replyOperation);
2495+
2496+
const timeout = new Error("chat run timed out");
2497+
timeout.name = "TimeoutError";
2498+
replyCtrl.abort(timeout);
2499+
expect(cancelSignal2.aborted).toBe(false);
2500+
});
2501+
2502+
it("aborts on explicit cancel after lifecycle timeout via replyOperation.result", async () => {
2503+
vi.useFakeTimers();
2504+
const replyCtrl = new AbortController();
2505+
const replyOperation = createTestReplyOperation(replyCtrl);
2506+
2507+
const timeout = new Error("chat run timed out");
2508+
timeout.name = "TimeoutError";
2509+
replyCtrl.abort(timeout);
2510+
2511+
const { signal: cancelSignal3 } = createPreflightCompactionCancelSignal(replyOperation);
2512+
expect(cancelSignal3.aborted).toBe(false);
2513+
2514+
(replyOperation as { result: ReplyOperation["result"] }).result = {
2515+
kind: "aborted",
2516+
code: "aborted_by_user",
2517+
};
2518+
await vi.advanceTimersByTimeAsync(600);
2519+
expect(cancelSignal3.aborted).toBe(true);
2520+
vi.useRealTimers();
2521+
});
23972522
});

‎src/auto-reply/reply/agent-runner-memory.ts‎

Lines changed: 53 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -718,6 +718,56 @@ async function estimatePromptTokensFromSessionTranscript(params: {
718718
}
719719
}
720720

721+
export function createPreflightCompactionCancelSignal(replyOperation: ReplyOperation): { signal: AbortSignal; cleanup: () => void } {
722+
const ctrl = new AbortController();
723+
let cancelInterval: ReturnType<typeof setInterval> | undefined;
724+
725+
const clearPolling = () => {
726+
if (cancelInterval) {
727+
clearInterval(cancelInterval);
728+
cancelInterval = undefined;
729+
}
730+
};
731+
732+
const startCancelPolling = () => {
733+
cancelInterval = setInterval(() => {
734+
if (ctrl.signal.aborted) {
735+
clearPolling();
736+
return;
737+
}
738+
if (replyOperation.result?.kind === "aborted") {
739+
ctrl.abort(new Error("Preflight compaction cancelled"));
740+
clearPolling();
741+
}
742+
}, 500);
743+
ctrl.signal.addEventListener("abort", clearPolling);
744+
};
745+
746+
if (replyOperation.abortSignal.aborted) {
747+
const reason = replyOperation.abortSignal.reason;
748+
if (!(reason instanceof Error && reason.name === "TimeoutError")) {
749+
ctrl.abort(reason);
750+
} else {
751+
startCancelPolling();
752+
}
753+
} else {
754+
replyOperation.abortSignal.addEventListener("abort", () => {
755+
const reason = replyOperation.abortSignal.reason;
756+
if (!(reason instanceof Error && reason.name === "TimeoutError")) {
757+
ctrl.abort(reason);
758+
return;
759+
}
760+
startCancelPolling();
761+
}, { once: true });
762+
}
763+
764+
if (replyOperation.result?.kind === "aborted" && !ctrl.signal.aborted) {
765+
ctrl.abort(new Error("Preflight compaction cancelled"));
766+
}
767+
768+
return { signal: ctrl.signal, cleanup: clearPolling };
769+
}
770+
721771
/** Runs preflight compaction when session state exceeds configured thresholds. */
722772
export async function runPreflightCompactionIfNeeded(params: {
723773
cfg: OpenClawConfig;
@@ -943,6 +993,7 @@ export async function runPreflightCompactionIfNeeded(params: {
943993
params.sessionKey ?? params.followupRun.run.sessionKey,
944994
{ storePath: params.storePath },
945995
);
996+
const cancelSignal = createPreflightCompactionCancelSignal(params.replyOperation);
946997
const result = await deps.compactEmbeddedAgentSession({
947998
sessionId: entry.sessionId,
948999
sessionKey: params.sessionKey,
@@ -978,8 +1029,9 @@ export async function runPreflightCompactionIfNeeded(params: {
9781029
contextTokenBudget: contextWindowTokens,
9791030
currentTokenCount: tokenCountForCompaction ?? freshPersistedTokens,
9801031
ownerNumbers: params.followupRun.run.ownerNumbers,
981-
abortSignal: params.replyOperation.abortSignal,
1032+
abortSignal: cancelSignal.signal,
9821033
});
1034+
cancelSignal.cleanup();
9831035

9841036
if (!result?.ok) {
9851037
const reason = result?.reason ?? "not_compacted";

0 commit comments

Comments
 (0)