Skip to content

Commit eb0ef79

Browse files
nex-dhnex-dh
committed
Polish chat abort preprocessing cleanup
1 parent 20ca5fe commit eb0ef79

2 files changed

Lines changed: 219 additions & 4 deletions

File tree

‎src/gateway/server-methods/chat.directive-tags.test.ts‎

Lines changed: 170 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3084,6 +3084,176 @@ describe("chat directive tag stripping for non-streaming final payloads", () =>
30843084
expect(mockState.deleteMediaBufferCalls).toEqual([{ id: "saved-media", subdir: "inbound" }]);
30853085
});
30863086

3087+
it("lets abort win when attachment preprocessing fails after cancellation", async () => {
3088+
createTranscriptFixture("openclaw-chat-send-abort-stage-error-");
3089+
mockState.finalText = "ok";
3090+
mockState.sessionEntry = {
3091+
modelProvider: "test-provider",
3092+
model: "vision-model",
3093+
};
3094+
mockState.modelCatalog = [
3095+
{
3096+
provider: "test-provider",
3097+
id: "vision-model",
3098+
name: "Vision model",
3099+
input: ["text", "image"],
3100+
},
3101+
];
3102+
mockState.savedMediaResults = [
3103+
{ path: "/home/user/.openclaw/media/inbound/report.pdf", contentType: "application/pdf" },
3104+
];
3105+
mockState.sandboxWorkspace = { workspaceDir: "/sandbox/workspace" };
3106+
const stageError = Object.assign(new Error("ENOSPC: no space left on device"), {
3107+
code: "ENOSPC",
3108+
});
3109+
mockState.stageSandboxMediaError = stageError;
3110+
let releaseSave!: () => void;
3111+
mockState.saveMediaWait = new Promise<void>((resolve) => {
3112+
releaseSave = resolve;
3113+
});
3114+
const respond = vi.fn();
3115+
const abortRespond = vi.fn();
3116+
const context = createChatContext();
3117+
const pdf = Buffer.from("%PDF-1.4\n%µ¶\n1 0 obj\n<<>>\nendobj\n").toString("base64");
3118+
const runId = "idem-abort-stage-error";
3119+
3120+
const send = chatHandlers["chat.send"]({
3121+
params: {
3122+
sessionKey: "main",
3123+
message: "read this",
3124+
idempotencyKey: runId,
3125+
attachments: [
3126+
{
3127+
type: "file",
3128+
mimeType: "application/pdf",
3129+
fileName: "report.pdf",
3130+
content: pdf,
3131+
},
3132+
],
3133+
},
3134+
respond: respond as unknown as Parameters<(typeof chatHandlers)["chat.send"]>[0]["respond"],
3135+
req: {} as never,
3136+
client: null,
3137+
isWebchatConnect: () => false,
3138+
context: context as GatewayRequestContext,
3139+
});
3140+
3141+
await waitForAssertion(() => {
3142+
expect(mockState.activeSaveMediaCalls).toBe(1);
3143+
});
3144+
3145+
await chatHandlers["chat.abort"]({
3146+
params: { sessionKey: "main", runId },
3147+
respond: abortRespond as unknown as Parameters<
3148+
(typeof chatHandlers)["chat.abort"]
3149+
>[0]["respond"],
3150+
req: {} as never,
3151+
client: null,
3152+
isWebchatConnect: () => false,
3153+
context: context as GatewayRequestContext,
3154+
});
3155+
3156+
expect(lastRespondCall(abortRespond)?.[0]).toBe(true);
3157+
expect(lastRespondCall(abortRespond)?.[1]).toEqual({
3158+
ok: true,
3159+
aborted: true,
3160+
runIds: [runId],
3161+
});
3162+
3163+
releaseSave();
3164+
await send;
3165+
3166+
expect(lastRespondCall(respond)?.[0]).toBe(true);
3167+
expect(lastRespondCall(respond)?.[1]).toEqual({ runId, status: "started" });
3168+
expect(lastRespondCall(respond)?.[2]).toBeUndefined();
3169+
expect(mockState.lastDispatchCtx).toBeUndefined();
3170+
expect(context.logGateway.error).not.toHaveBeenCalled();
3171+
expect(mockState.deleteMediaBufferCalls).toEqual([{ id: "saved-media", subdir: "inbound" }]);
3172+
});
3173+
3174+
it("cleans offloaded media when abort wins after attachment preprocessing", async () => {
3175+
createTranscriptFixture("openclaw-chat-send-abort-cleans-offload-");
3176+
mockState.finalText = "ok";
3177+
mockState.sessionEntry = {
3178+
modelProvider: "test-provider",
3179+
model: "vision-model",
3180+
};
3181+
mockState.modelCatalog = [
3182+
{
3183+
provider: "test-provider",
3184+
id: "vision-model",
3185+
name: "Vision model",
3186+
input: ["text", "image"],
3187+
},
3188+
];
3189+
mockState.savedMediaResults = [
3190+
{ path: "/home/user/.openclaw/media/inbound/report.pdf", contentType: "application/pdf" },
3191+
];
3192+
mockState.sandboxWorkspace = { workspaceDir: "/sandbox/workspace" };
3193+
mockState.stagedRelativePaths = ["media/inbound/report.pdf"];
3194+
let releaseSave!: () => void;
3195+
mockState.saveMediaWait = new Promise<void>((resolve) => {
3196+
releaseSave = resolve;
3197+
});
3198+
const respond = vi.fn();
3199+
const abortRespond = vi.fn();
3200+
const context = createChatContext();
3201+
const pdf = Buffer.from("%PDF-1.4\n%µ¶\n1 0 obj\n<<>>\nendobj\n").toString("base64");
3202+
const runId = "idem-abort-clean-offload";
3203+
3204+
const send = chatHandlers["chat.send"]({
3205+
params: {
3206+
sessionKey: "main",
3207+
message: "read this",
3208+
idempotencyKey: runId,
3209+
attachments: [
3210+
{
3211+
type: "file",
3212+
mimeType: "application/pdf",
3213+
fileName: "report.pdf",
3214+
content: pdf,
3215+
},
3216+
],
3217+
},
3218+
respond: respond as unknown as Parameters<(typeof chatHandlers)["chat.send"]>[0]["respond"],
3219+
req: {} as never,
3220+
client: null,
3221+
isWebchatConnect: () => false,
3222+
context: context as GatewayRequestContext,
3223+
});
3224+
3225+
await waitForAssertion(() => {
3226+
expect(mockState.activeSaveMediaCalls).toBe(1);
3227+
});
3228+
3229+
await chatHandlers["chat.abort"]({
3230+
params: { sessionKey: "main", runId },
3231+
respond: abortRespond as unknown as Parameters<
3232+
(typeof chatHandlers)["chat.abort"]
3233+
>[0]["respond"],
3234+
req: {} as never,
3235+
client: null,
3236+
isWebchatConnect: () => false,
3237+
context: context as GatewayRequestContext,
3238+
});
3239+
3240+
expect(lastRespondCall(abortRespond)?.[0]).toBe(true);
3241+
expect(lastRespondCall(abortRespond)?.[1]).toEqual({
3242+
ok: true,
3243+
aborted: true,
3244+
runIds: [runId],
3245+
});
3246+
3247+
releaseSave();
3248+
await send;
3249+
3250+
expect(lastRespondCall(respond)?.[0]).toBe(true);
3251+
expect(lastRespondCall(respond)?.[1]).toEqual({ runId, status: "started" });
3252+
expect(lastRespondCall(respond)?.[2]).toBeUndefined();
3253+
expect(mockState.lastDispatchCtx).toBeUndefined();
3254+
expect(mockState.deleteMediaBufferCalls).toEqual([{ id: "saved-media", subdir: "inbound" }]);
3255+
});
3256+
30873257
it("logs chat.send attachment parse failures with stack details", async () => {
30883258
createTranscriptFixture("openclaw-chat-send-attachment-parse-stack-");
30893259
const respond = vi.fn();

‎src/gateway/server-methods/chat.ts‎

Lines changed: 49 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -951,6 +951,38 @@ function stripTrailingOffloadedMediaMarkers(message: string, refs: OffloadedRef[
951951
// Returned paths are absolute media-store paths when no sandbox is active, or
952952
// sandbox-relative paths plus `workspaceDir` when sandboxing is active. Host-side
953953
// media-understanding uses MediaWorkspaceDir to resolve those relative paths.
954+
async function cleanupChatSendPreDispatchMedia(params: {
955+
offloadedRefs: OffloadedRef[];
956+
mediaPathOffloadPaths: string[];
957+
mediaPathOffloadWorkspaceDir?: string;
958+
logGateway: GatewayRequestContext["logGateway"];
959+
}) {
960+
const cleanupTasks: Promise<unknown>[] = params.offloadedRefs.map((ref) =>
961+
deleteMediaBuffer(ref.id, "inbound"),
962+
);
963+
if (params.mediaPathOffloadWorkspaceDir) {
964+
const workspaceRoot = path.resolve(params.mediaPathOffloadWorkspaceDir);
965+
for (const stagedPath of new Set(params.mediaPathOffloadPaths)) {
966+
if (!stagedPath || path.isAbsolute(stagedPath)) {
967+
continue;
968+
}
969+
const target = path.resolve(workspaceRoot, stagedPath);
970+
if (target === workspaceRoot || !target.startsWith(`${workspaceRoot}${path.sep}`)) {
971+
continue;
972+
}
973+
cleanupTasks.push(fs.promises.rm(target, { force: true }));
974+
}
975+
}
976+
const results = await Promise.allSettled(cleanupTasks);
977+
for (const result of results) {
978+
if (result.status === "rejected") {
979+
params.logGateway.warn(
980+
`chat.send aborted attachment cleanup failed: ${formatForLog(result.reason)}`,
981+
);
982+
}
983+
}
984+
}
985+
954986
async function prestageMediaPathOffloads(params: {
955987
offloadedRefs: OffloadedRef[];
956988
includeImageRefs?: boolean;
@@ -2193,6 +2225,12 @@ export const chatHandlers: GatewayRequestHandlers = {
21932225
const explicitOriginTargetsPlugin = explicitOriginTargetsPluginBinding(
21942226
explicitOriginResult.value,
21952227
);
2228+
const ackPayload = {
2229+
runId: clientRunId,
2230+
status: "started" as const,
2231+
};
2232+
// Register before awaited attachment/model preprocessing so an immediate
2233+
// chat.abort can find and cancel the run (issue #84176).
21962234
const activeRunAbort = registerChatAbortController({
21972235
chatAbortControllers: context.chatAbortControllers,
21982236
runId: clientRunId,
@@ -2301,6 +2339,11 @@ export const chatHandlers: GatewayRequestHandlers = {
23012339
},
23022340
);
23032341
} catch (err) {
2342+
if (activeRunAbort.controller.signal.aborted) {
2343+
activeRunAbort.cleanup();
2344+
respond(true, ackPayload, undefined, { runId: clientRunId });
2345+
return;
2346+
}
23042347
activeRunAbort.cleanup();
23052348
logAttachmentFailure(context.logGateway, "chat.send attachment parse/stage failed", err);
23062349
respond(
@@ -2315,11 +2358,13 @@ export const chatHandlers: GatewayRequestHandlers = {
23152358
}
23162359
}
23172360

2318-
const ackPayload = {
2319-
runId: clientRunId,
2320-
status: "started" as const,
2321-
};
23222361
if (activeRunAbort.controller.signal.aborted) {
2362+
await cleanupChatSendPreDispatchMedia({
2363+
offloadedRefs,
2364+
mediaPathOffloadPaths,
2365+
mediaPathOffloadWorkspaceDir,
2366+
logGateway: context.logGateway,
2367+
});
23232368
activeRunAbort.cleanup();
23242369
respond(true, ackPayload, undefined, { runId: clientRunId });
23252370
return;

0 commit comments

Comments
 (0)