Skip to content

Commit c397a02

Browse files
steipeteKevin ShenghuizjmySuko
committed
fix(queue): harden drain/abort/timeout race handling
- reject new lane enqueues once gateway drain begins - always reset lane draining state and isolate onWait callback failures - persist per-session abort cutoff and skip stale queued messages - avoid false 600s agentTurn timeout in isolated cron jobs Fixes #27407 Fixes #27332 Fixes #27427 Co-authored-by: Kevin Shenghui <[email protected]> Co-authored-by: zjmy <[email protected]> Co-authored-by: suko <[email protected]>
1 parent 1aef45b commit c397a02

13 files changed

Lines changed: 551 additions & 42 deletions

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ Docs: https://docs.openclaw.ai
1414
### Fixes
1515

1616
- Cron/Hooks isolated routing: preserve canonical `agent:*` session keys in isolated runs so already-qualified keys are not double-prefixed (for example `agent:main:main` no longer becomes `agent:main:agent:main:main`). Landed from contributor PR #27333 by @MaheshBhushan. (#27289, #27282)
17+
- Queue/Drain/Cron reliability: harden lane draining with guaranteed `draining` flag reset on synchronous pump failures, reject new queue enqueues during gateway restart drain windows (instead of silently killing accepted tasks), add `/stop` queued-backlog cutoff metadata with stale-message skipping (while avoiding cross-session native-stop cutoff bleed), and raise isolated cron `agentTurn` outer safety timeout to avoid false 10-minute timeout races against longer agent session timeouts. (#27407, #27332, #27427)
1718
- Security/Plugin channel HTTP auth: normalize protected `/api/channels` path checks against canonicalized request paths (case + percent-decoding + slash normalization), and fail closed on malformed `%`-encoded channel prefixes so alternate-path variants cannot bypass gateway auth.
1819
- Security/Exec approvals forwarding: prefer turn-source channel/account/thread metadata when resolving approval delivery targets so stale session routes do not misroute approval prompts.
1920
- Security/Sandbox path alias guard: reject broken symlink targets by resolving through existing ancestors and failing closed on out-of-root targets, preventing workspace-only `apply_patch` writes from escaping sandbox/workspace boundaries via dangling symlinks. This ships in the next npm release (`2026.2.26`). Thanks @tdjackey for reporting.

‎src/auto-reply/reply/abort.test.ts‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,8 +10,10 @@ import {
1010
isAbortRequestText,
1111
isAbortTrigger,
1212
resetAbortMemoryForTest,
13+
resolveAbortCutoffFromContext,
1314
resolveSessionEntryForKey,
1415
setAbortMemory,
16+
shouldSkipMessageByAbortCutoff,
1517
tryFastAbortFromMessage,
1618
} from "./abort.js";
1719
import { enqueueFollowupRun, getFollowupQueueDepth, type FollowupRun } from "./queue.js";
@@ -80,6 +82,9 @@ describe("abort detection", () => {
8082
sessionKey: string;
8183
from: string;
8284
to: string;
85+
targetSessionKey?: string;
86+
messageSid?: string;
87+
timestamp?: number;
8388
}) {
8489
return tryFastAbortFromMessage({
8590
ctx: buildTestCtx({
@@ -91,6 +96,9 @@ describe("abort detection", () => {
9196
Surface: "telegram",
9297
From: params.from,
9398
To: params.to,
99+
...(params.targetSessionKey ? { CommandTargetSessionKey: params.targetSessionKey } : {}),
100+
...(params.messageSid ? { MessageSid: params.messageSid } : {}),
101+
...(typeof params.timestamp === "number" ? { Timestamp: params.timestamp } : {}),
94102
}),
95103
cfg: params.cfg,
96104
});
@@ -221,6 +229,62 @@ describe("abort detection", () => {
221229
expect(getAbortMemory("session-2104")).toBe(true);
222230
});
223231

232+
it("extracts abort cutoff metadata from context", () => {
233+
expect(
234+
resolveAbortCutoffFromContext(
235+
buildTestCtx({
236+
MessageSid: "42",
237+
Timestamp: 123,
238+
}),
239+
),
240+
).toEqual({
241+
messageSid: "42",
242+
timestamp: 123,
243+
});
244+
});
245+
246+
it("treats numeric message IDs at or before cutoff as stale", () => {
247+
expect(
248+
shouldSkipMessageByAbortCutoff({
249+
cutoffMessageSid: "200",
250+
messageSid: "199",
251+
}),
252+
).toBe(true);
253+
expect(
254+
shouldSkipMessageByAbortCutoff({
255+
cutoffMessageSid: "200",
256+
messageSid: "200",
257+
}),
258+
).toBe(true);
259+
expect(
260+
shouldSkipMessageByAbortCutoff({
261+
cutoffMessageSid: "200",
262+
messageSid: "201",
263+
}),
264+
).toBe(false);
265+
});
266+
267+
it("falls back to timestamp cutoff when message IDs are unavailable", () => {
268+
expect(
269+
shouldSkipMessageByAbortCutoff({
270+
cutoffTimestamp: 2000,
271+
timestamp: 1999,
272+
}),
273+
).toBe(true);
274+
expect(
275+
shouldSkipMessageByAbortCutoff({
276+
cutoffTimestamp: 2000,
277+
timestamp: 2000,
278+
}),
279+
).toBe(true);
280+
expect(
281+
shouldSkipMessageByAbortCutoff({
282+
cutoffTimestamp: 2000,
283+
timestamp: 2001,
284+
}),
285+
).toBe(false);
286+
});
287+
224288
it("resolves session entry when key exists in store", () => {
225289
const store = {
226290
"session-1": { sessionId: "abc", updatedAt: 0 },
@@ -291,6 +355,64 @@ describe("abort detection", () => {
291355
expect(commandQueueMocks.clearCommandLane).toHaveBeenCalledWith(`session:${sessionKey}`);
292356
});
293357

358+
it("persists abort cutoff metadata on /stop when command and target session match", async () => {
359+
const sessionKey = "telegram:123";
360+
const sessionId = "session-123";
361+
const { storePath, cfg } = await createAbortConfig({
362+
sessionIdsByKey: { [sessionKey]: sessionId },
363+
});
364+
365+
const result = await runStopCommand({
366+
cfg,
367+
sessionKey,
368+
from: "telegram:123",
369+
to: "telegram:123",
370+
messageSid: "55",
371+
timestamp: 1234567890000,
372+
});
373+
374+
expect(result.handled).toBe(true);
375+
const store = JSON.parse(await fs.readFile(storePath, "utf8")) as Record<string, unknown>;
376+
const entry = store[sessionKey] as {
377+
abortedLastRun?: boolean;
378+
abortCutoffMessageSid?: string;
379+
abortCutoffTimestamp?: number;
380+
};
381+
expect(entry.abortedLastRun).toBe(true);
382+
expect(entry.abortCutoffMessageSid).toBe("55");
383+
expect(entry.abortCutoffTimestamp).toBe(1234567890000);
384+
});
385+
386+
it("does not persist cutoff metadata when native /stop targets a different session", async () => {
387+
const slashSessionKey = "telegram:slash:123";
388+
const targetSessionKey = "agent:main:telegram:group:123";
389+
const targetSessionId = "session-target";
390+
const { storePath, cfg } = await createAbortConfig({
391+
sessionIdsByKey: { [targetSessionKey]: targetSessionId },
392+
});
393+
394+
const result = await runStopCommand({
395+
cfg,
396+
sessionKey: slashSessionKey,
397+
from: "telegram:123",
398+
to: "telegram:123",
399+
targetSessionKey,
400+
messageSid: "999",
401+
timestamp: 1234567890000,
402+
});
403+
404+
expect(result.handled).toBe(true);
405+
const store = JSON.parse(await fs.readFile(storePath, "utf8")) as Record<string, unknown>;
406+
const entry = store[targetSessionKey] as {
407+
abortedLastRun?: boolean;
408+
abortCutoffMessageSid?: string;
409+
abortCutoffTimestamp?: number;
410+
};
411+
expect(entry.abortedLastRun).toBe(true);
412+
expect(entry.abortCutoffMessageSid).toBeUndefined();
413+
expect(entry.abortCutoffTimestamp).toBeUndefined();
414+
});
415+
294416
it("fast-abort stops active subagent runs for requester session", async () => {
295417
const sessionKey = "telegram:parent";
296418
const childKey = "agent:main:subagent:child-1";

‎src/auto-reply/reply/abort.ts‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -113,6 +113,80 @@ export function getAbortMemory(key: string): boolean | undefined {
113113
return ABORT_MEMORY.get(normalized);
114114
}
115115

116+
export type AbortCutoff = {
117+
messageSid?: string;
118+
timestamp?: number;
119+
};
120+
121+
export function resolveAbortCutoffFromContext(ctx: MsgContext): AbortCutoff | undefined {
122+
const messageSid =
123+
(typeof ctx.MessageSidFull === "string" && ctx.MessageSidFull.trim()) ||
124+
(typeof ctx.MessageSid === "string" && ctx.MessageSid.trim()) ||
125+
undefined;
126+
const timestamp =
127+
typeof ctx.Timestamp === "number" && Number.isFinite(ctx.Timestamp) ? ctx.Timestamp : undefined;
128+
if (!messageSid && timestamp === undefined) {
129+
return undefined;
130+
}
131+
return { messageSid, timestamp };
132+
}
133+
134+
function toNumericMessageSid(value: string | undefined): bigint | undefined {
135+
const trimmed = value?.trim();
136+
if (!trimmed || !/^\d+$/.test(trimmed)) {
137+
return undefined;
138+
}
139+
try {
140+
return BigInt(trimmed);
141+
} catch {
142+
return undefined;
143+
}
144+
}
145+
146+
export function shouldSkipMessageByAbortCutoff(params: {
147+
cutoffMessageSid?: string;
148+
cutoffTimestamp?: number;
149+
messageSid?: string;
150+
timestamp?: number;
151+
}): boolean {
152+
const cutoffSid = params.cutoffMessageSid?.trim();
153+
const currentSid = params.messageSid?.trim();
154+
if (cutoffSid && currentSid) {
155+
const cutoffNumeric = toNumericMessageSid(cutoffSid);
156+
const currentNumeric = toNumericMessageSid(currentSid);
157+
if (cutoffNumeric !== undefined && currentNumeric !== undefined) {
158+
return currentNumeric <= cutoffNumeric;
159+
}
160+
if (currentSid === cutoffSid) {
161+
return true;
162+
}
163+
}
164+
if (
165+
typeof params.cutoffTimestamp === "number" &&
166+
Number.isFinite(params.cutoffTimestamp) &&
167+
typeof params.timestamp === "number" &&
168+
Number.isFinite(params.timestamp)
169+
) {
170+
return params.timestamp <= params.cutoffTimestamp;
171+
}
172+
return false;
173+
}
174+
175+
function shouldPersistAbortCutoff(params: {
176+
commandSessionKey?: string;
177+
targetSessionKey?: string;
178+
}): boolean {
179+
const commandSessionKey = params.commandSessionKey?.trim();
180+
const targetSessionKey = params.targetSessionKey?.trim();
181+
if (!commandSessionKey || !targetSessionKey) {
182+
return true;
183+
}
184+
// Native targeted /stop can run from a slash/session-control key while the
185+
// actual target session uses different message id/timestamp spaces.
186+
// Persist cutoff only when command source and target are the same session.
187+
return commandSessionKey === targetSessionKey;
188+
}
189+
116190
function pruneAbortMemory(): void {
117191
if (ABORT_MEMORY.size <= ABORT_MEMORY_MAX) {
118192
return;
@@ -302,8 +376,16 @@ export async function tryFastAbortFromMessage(params: {
302376
`abort: cleared followups=${cleared.followupCleared} lane=${cleared.laneCleared} keys=${cleared.keys.join(",")}`,
303377
);
304378
}
379+
const abortCutoff = shouldPersistAbortCutoff({
380+
commandSessionKey: ctx.SessionKey,
381+
targetSessionKey: key ?? targetKey,
382+
})
383+
? resolveAbortCutoffFromContext(ctx)
384+
: undefined;
305385
if (entry && key) {
306386
entry.abortedLastRun = true;
387+
entry.abortCutoffMessageSid = abortCutoff?.messageSid;
388+
entry.abortCutoffTimestamp = abortCutoff?.timestamp;
307389
entry.updatedAt = Date.now();
308390
store[key] = entry;
309391
await updateSessionStore(storePath, (nextStore) => {
@@ -312,6 +394,8 @@ export async function tryFastAbortFromMessage(params: {
312394
return;
313395
}
314396
nextEntry.abortedLastRun = true;
397+
nextEntry.abortCutoffMessageSid = abortCutoff?.messageSid;
398+
nextEntry.abortCutoffTimestamp = abortCutoff?.timestamp;
315399
nextEntry.updatedAt = Date.now();
316400
nextStore[key] = nextEntry;
317401
});

‎src/auto-reply/reply/commands-session.ts‎

Lines changed: 23 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ import { normalizeUsageDisplay, resolveResponseUsageMode } from "../thinking.js"
1919
import {
2020
formatAbortReplyText,
2121
isAbortTrigger,
22+
resolveAbortCutoffFromContext,
2223
resolveSessionEntryForKey,
2324
setAbortMemory,
2425
stopSubagentsForRequester,
@@ -99,18 +100,27 @@ async function applyAbortTarget(params: {
99100
sessionStore?: Record<string, SessionEntry>;
100101
storePath?: string;
101102
abortKey?: string;
103+
abortCutoff?: { messageSid?: string; timestamp?: number };
102104
}) {
103105
const { abortTarget } = params;
104106
if (abortTarget.sessionId) {
105107
abortEmbeddedPiRun(abortTarget.sessionId);
106108
}
107109
if (abortTarget.entry && params.sessionStore && abortTarget.key) {
108110
abortTarget.entry.abortedLastRun = true;
111+
abortTarget.entry.abortCutoffMessageSid = params.abortCutoff?.messageSid;
112+
abortTarget.entry.abortCutoffTimestamp = params.abortCutoff?.timestamp;
109113
abortTarget.entry.updatedAt = Date.now();
110114
params.sessionStore[abortTarget.key] = abortTarget.entry;
111115
if (params.storePath) {
112116
await updateSessionStore(params.storePath, (store) => {
113-
store[abortTarget.key] = abortTarget.entry;
117+
store[abortTarget.key] = {
118+
...abortTarget.entry,
119+
abortedLastRun: true,
120+
abortCutoffMessageSid: params.abortCutoff?.messageSid,
121+
abortCutoffTimestamp: params.abortCutoff?.timestamp,
122+
updatedAt: Date.now(),
123+
};
114124
});
115125
}
116126
} else if (params.abortKey) {
@@ -503,6 +513,12 @@ export const handleStopCommand: CommandHandler = async (params, allowTextCommand
503513
sessionStore: params.sessionStore,
504514
storePath: params.storePath,
505515
abortKey: params.command.abortKey,
516+
abortCutoff:
517+
params.sessionKey?.trim() &&
518+
abortTarget.key?.trim() &&
519+
params.sessionKey.trim() === abortTarget.key.trim()
520+
? resolveAbortCutoffFromContext(params.ctx)
521+
: undefined,
506522
});
507523

508524
// Trigger internal hook for stop command
@@ -545,6 +561,12 @@ export const handleAbortTrigger: CommandHandler = async (params, allowTextComman
545561
sessionStore: params.sessionStore,
546562
storePath: params.storePath,
547563
abortKey: params.command.abortKey,
564+
abortCutoff:
565+
params.sessionKey?.trim() &&
566+
abortTarget.key?.trim() &&
567+
params.sessionKey.trim() === abortTarget.key.trim()
568+
? resolveAbortCutoffFromContext(params.ctx)
569+
: undefined,
548570
});
549571
return { shouldContinue: false, reply: { text: "⚙️ Agent was aborted." } };
550572
};

0 commit comments

Comments
 (0)