Skip to content

Commit df74a84

Browse files
NianJiuZstNianJiuZst
committed
fix(gateway): close chat.abort send race
1 parent 28beea9 commit df74a84

2 files changed

Lines changed: 274 additions & 118 deletions

File tree

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

Lines changed: 130 additions & 114 deletions
Original file line numberDiff line numberDiff line change
@@ -2190,127 +2190,139 @@ export const chatHandlers: GatewayRequestHandlers = {
21902190
return;
21912191
}
21922192
}
2193+
// Register before awaited attachment/model preparation so an immediate
2194+
// chat.abort can still find and stop the active run.
2195+
const activeRunAbort = registerChatAbortController({
2196+
chatAbortControllers: context.chatAbortControllers,
2197+
runId: clientRunId,
2198+
sessionId: backingSessionId ?? clientRunId,
2199+
sessionKey: rawSessionKey,
2200+
timeoutMs,
2201+
now,
2202+
ownerConnId: normalizeOptionalText(client?.connId),
2203+
ownerDeviceId: normalizeOptionalText(client?.connect?.device?.id),
2204+
providerId: resolvedSessionModel.provider,
2205+
authProviderId: resolvedSessionAuthProvider,
2206+
kind: "chat-send",
2207+
});
2208+
if (!activeRunAbort.registered) {
2209+
respond(true, { runId: clientRunId, status: "in_flight" as const }, undefined, {
2210+
cached: true,
2211+
runId: clientRunId,
2212+
});
2213+
return;
2214+
}
2215+
21932216
const explicitOriginTargetsPlugin = explicitOriginTargetsPluginBinding(
21942217
explicitOriginResult.value,
21952218
);
21962219
let modelOverride: string | undefined;
21972220
let modelOverrideFallbacks: string[] | undefined;
2198-
if (normalizedAttachments.length > 0) {
2199-
try {
2200-
await measureDiagnosticsTimelineSpan(
2201-
"gateway.chat_send.prepare_attachments",
2202-
async () => {
2203-
const hasImageAttachments = await hasImageChatAttachments(normalizedAttachments);
2204-
const supportsSessionModelImages = await resolveGatewayModelSupportsImages({
2205-
loadGatewayModelCatalog: context.loadGatewayModelCatalog,
2206-
provider: resolvedSessionModel.provider,
2207-
model: resolvedSessionModel.model,
2208-
});
2209-
const explicitOriginSupportsInlineImages =
2210-
explicitOriginTargetsAcpSession(explicitOriginResult.value) ||
2211-
explicitOriginTargetsPlugin;
2212-
const imageModelPlan = await resolveImageModelOverridePlan({
2213-
cfg,
2214-
agentId,
2215-
defaultProvider: resolvedConfiguredDefaultModel.provider,
2216-
defaultModel: resolvedConfiguredDefaultModel.model,
2217-
hasImageAttachments,
2218-
sessionModelSupportsImages:
2219-
supportsSessionModelImages || explicitOriginSupportsInlineImages,
2220-
modelSupportsImages: (ref) =>
2221-
resolveGatewayModelSupportsImages({
2222-
loadGatewayModelCatalog: context.loadGatewayModelCatalog,
2223-
provider: ref.provider,
2224-
model: ref.model,
2225-
}),
2226-
});
2227-
if (imageModelPlan.kind === "inline-image-model") {
2228-
modelOverride = imageModelPlan.modelOverride;
2229-
modelOverrideFallbacks = imageModelPlan.modelOverrideFallbacks;
2230-
}
2231-
// Bound plugin sessions own the real recipient model, so keep image
2232-
// attachments even when the parent OpenClaw session model is text-only.
2233-
const supportsImages =
2234-
imageModelPlan.kind === "inline-session" ||
2235-
imageModelPlan.kind === "inline-image-model" ||
2236-
explicitOriginSupportsInlineImages;
2237-
const routeImageOffloadsAsMediaPaths = !supportsImages;
2238-
const parsed = await parseMessageWithAttachments(
2239-
inboundMessage,
2240-
normalizedAttachments,
2241-
{
2242-
maxBytes: resolveChatAttachmentMaxBytes(cfg),
2243-
log: context.logGateway,
2244-
supportsImages,
2245-
// chat.send routes selected offloadedRefs into ctx.MediaPaths below
2246-
// so the auto-reply stage pipeline can surface them to the agent.
2247-
acceptNonImage: true,
2221+
let dispatchScheduled = false;
2222+
try {
2223+
if (normalizedAttachments.length > 0) {
2224+
try {
2225+
await measureDiagnosticsTimelineSpan(
2226+
"gateway.chat_send.prepare_attachments",
2227+
async () => {
2228+
const hasImageAttachments = await hasImageChatAttachments(normalizedAttachments);
2229+
const supportsSessionModelImages = await resolveGatewayModelSupportsImages({
2230+
loadGatewayModelCatalog: context.loadGatewayModelCatalog,
2231+
provider: resolvedSessionModel.provider,
2232+
model: resolvedSessionModel.model,
2233+
});
2234+
const explicitOriginSupportsInlineImages =
2235+
explicitOriginTargetsAcpSession(explicitOriginResult.value) ||
2236+
explicitOriginTargetsPlugin;
2237+
const imageModelPlan = await resolveImageModelOverridePlan({
2238+
cfg,
2239+
agentId,
2240+
defaultProvider: resolvedConfiguredDefaultModel.provider,
2241+
defaultModel: resolvedConfiguredDefaultModel.model,
2242+
hasImageAttachments,
2243+
sessionModelSupportsImages:
2244+
supportsSessionModelImages || explicitOriginSupportsInlineImages,
2245+
modelSupportsImages: (ref) =>
2246+
resolveGatewayModelSupportsImages({
2247+
loadGatewayModelCatalog: context.loadGatewayModelCatalog,
2248+
provider: ref.provider,
2249+
model: ref.model,
2250+
}),
2251+
});
2252+
if (imageModelPlan.kind === "inline-image-model") {
2253+
modelOverride = imageModelPlan.modelOverride;
2254+
modelOverrideFallbacks = imageModelPlan.modelOverrideFallbacks;
2255+
}
2256+
// Bound plugin sessions own the real recipient model, so keep image
2257+
// attachments even when the parent OpenClaw session model is text-only.
2258+
const supportsImages =
2259+
imageModelPlan.kind === "inline-session" ||
2260+
imageModelPlan.kind === "inline-image-model" ||
2261+
explicitOriginSupportsInlineImages;
2262+
const routeImageOffloadsAsMediaPaths = !supportsImages;
2263+
const parsed = await parseMessageWithAttachments(
2264+
inboundMessage,
2265+
normalizedAttachments,
2266+
{
2267+
maxBytes: resolveChatAttachmentMaxBytes(cfg),
2268+
log: context.logGateway,
2269+
supportsImages,
2270+
// chat.send routes selected offloadedRefs into ctx.MediaPaths below
2271+
// so the auto-reply stage pipeline can surface them to the agent.
2272+
acceptNonImage: true,
2273+
},
2274+
);
2275+
parsedMessage = stripTrailingOffloadedMediaMarkers(
2276+
parsed.message,
2277+
routeImageOffloadsAsMediaPaths
2278+
? parsed.offloadedRefs.filter((ref) => ref.mimeType.startsWith("image/"))
2279+
: [],
2280+
);
2281+
parsedImages = parsed.images;
2282+
imageOrder = routeImageOffloadsAsMediaPaths ? [] : parsed.imageOrder;
2283+
offloadedRefs = parsed.offloadedRefs;
2284+
({
2285+
paths: mediaPathOffloadPaths,
2286+
types: mediaPathOffloadTypes,
2287+
workspaceDir: mediaPathOffloadWorkspaceDir,
2288+
} = await prestageMediaPathOffloads({
2289+
offloadedRefs,
2290+
// Text-only image offloads need ctx.MediaPaths so media-understanding
2291+
// can describe them via agents.defaults.imageModel. Vision-capable
2292+
// image offloads stay as prompt refs for native image loading.
2293+
includeImageRefs: routeImageOffloadsAsMediaPaths,
2294+
cfg,
2295+
sessionKey,
2296+
agentId,
2297+
}));
2298+
},
2299+
{
2300+
phase: "agent-turn",
2301+
config: cfg,
2302+
attributes: {
2303+
attachmentCount: normalizedAttachments.length,
2304+
hasExplicitOrigin: explicitOriginResult.value !== undefined,
22482305
},
2249-
);
2250-
parsedMessage = stripTrailingOffloadedMediaMarkers(
2251-
parsed.message,
2252-
routeImageOffloadsAsMediaPaths
2253-
? parsed.offloadedRefs.filter((ref) => ref.mimeType.startsWith("image/"))
2254-
: [],
2255-
);
2256-
parsedImages = parsed.images;
2257-
imageOrder = routeImageOffloadsAsMediaPaths ? [] : parsed.imageOrder;
2258-
offloadedRefs = parsed.offloadedRefs;
2259-
({
2260-
paths: mediaPathOffloadPaths,
2261-
types: mediaPathOffloadTypes,
2262-
workspaceDir: mediaPathOffloadWorkspaceDir,
2263-
} = await prestageMediaPathOffloads({
2264-
offloadedRefs,
2265-
// Text-only image offloads need ctx.MediaPaths so media-understanding
2266-
// can describe them via agents.defaults.imageModel. Vision-capable
2267-
// image offloads stay as prompt refs for native image loading.
2268-
includeImageRefs: routeImageOffloadsAsMediaPaths,
2269-
cfg,
2270-
sessionKey,
2271-
agentId,
2272-
}));
2273-
},
2274-
{
2275-
phase: "agent-turn",
2276-
config: cfg,
2277-
attributes: {
2278-
attachmentCount: normalizedAttachments.length,
2279-
hasExplicitOrigin: explicitOriginResult.value !== undefined,
22802306
},
2281-
},
2282-
);
2283-
} catch (err) {
2284-
logAttachmentFailure(context.logGateway, "chat.send attachment parse/stage failed", err);
2285-
respond(
2286-
false,
2287-
undefined,
2288-
errorShape(
2289-
err instanceof MediaOffloadError ? ErrorCodes.UNAVAILABLE : ErrorCodes.INVALID_REQUEST,
2290-
String(err),
2291-
),
2292-
);
2293-
return;
2307+
);
2308+
} catch (err) {
2309+
logAttachmentFailure(context.logGateway, "chat.send attachment parse/stage failed", err);
2310+
respond(
2311+
false,
2312+
undefined,
2313+
errorShape(
2314+
err instanceof MediaOffloadError
2315+
? ErrorCodes.UNAVAILABLE
2316+
: ErrorCodes.INVALID_REQUEST,
2317+
String(err),
2318+
),
2319+
);
2320+
return;
2321+
}
22942322
}
2295-
}
22962323

2297-
try {
2298-
const activeRunAbort = registerChatAbortController({
2299-
chatAbortControllers: context.chatAbortControllers,
2300-
runId: clientRunId,
2301-
sessionId: backingSessionId ?? clientRunId,
2302-
sessionKey: rawSessionKey,
2303-
timeoutMs,
2304-
now,
2305-
ownerConnId: normalizeOptionalText(client?.connId),
2306-
ownerDeviceId: normalizeOptionalText(client?.connect?.device?.id),
2307-
providerId: resolvedSessionModel.provider,
2308-
authProviderId: resolvedSessionAuthProvider,
2309-
kind: "chat-send",
2310-
});
2311-
if (!activeRunAbort.registered) {
2312-
respond(true, { runId: clientRunId, status: "in_flight" as const }, undefined, {
2313-
cached: true,
2324+
if (activeRunAbort.controller.signal.aborted) {
2325+
respond(true, { runId: clientRunId, status: "started" as const }, undefined, {
23142326
runId: clientRunId,
23152327
});
23162328
return;
@@ -2945,9 +2957,8 @@ export const chatHandlers: GatewayRequestHandlers = {
29452957
activeRunAbort.cleanup();
29462958
context.removeChatRun(clientRunId, clientRunId, sessionKey);
29472959
});
2960+
dispatchScheduled = true;
29482961
} catch (err) {
2949-
context.chatAbortControllers.delete(clientRunId);
2950-
context.removeChatRun(clientRunId, clientRunId, sessionKey);
29512962
const error = errorShape(ErrorCodes.UNAVAILABLE, String(err));
29522963
const payload = {
29532964
runId: clientRunId,
@@ -2968,6 +2979,11 @@ export const chatHandlers: GatewayRequestHandlers = {
29682979
runId: clientRunId,
29692980
error: formatForLog(err),
29702981
});
2982+
} finally {
2983+
if (!dispatchScheduled) {
2984+
activeRunAbort.cleanup();
2985+
context.removeChatRun(clientRunId, clientRunId, sessionKey);
2986+
}
29712987
}
29722988
},
29732989
"chat.inject": async ({ params, respond, context }) => {

0 commit comments

Comments
 (0)