ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-12 21:53:00 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(agents): honor run abort signal in image and pdf tools (#112644)

Browse Source 
* fix(agents): honor run abort signal in image and pdf tools

The image and pdf agent tools declared `execute: async (_toolCallId, args)`
and dropped the run abort signal that `wrapToolWithAbortSignal` supplies as
the third execute argument. The wrapper only races the execute promise, so an
aborted run kept sequentially downloading images/PDFs (up to the per-tool cap,
each up to the byte cap) and still issued a paid vision/PDF-model call for a
dead run.

Thread the signal into the existing `requestInit: { signal }` seam (which the
media fetch layer already merges into the download fetch) and add
`signal.throwIfAborted()` between sequential loop items and before the paid
model call. No new media-options signal field; non-abort behavior is unchanged.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(agents): keep the pdf test suite under the lint ceiling; guard model dispatch

Follow-up on the abort-signal change, resolving the check-lint failure and a
review finding on the same seam.

max-lines
---------
Adding the two pdf abort tests pushed src/agents/tools/pdf-tool.test.ts to 1018
effective lines against a ceiling of 1000, failing check-lint. Resolved without
touching config/max-lines-baseline.txt — a suppression there would also have
tripped the max-lines ratchet.

- The new tests declared their own `describe` with `beforeEach`/`afterEach`
  hooks identical to the existing `describe("createPdfTool")`. They exercise
  that same tool, so they now live in it and the duplicated scaffolding is gone.
- `stubPdfToolInfra`, `createPdfModelRegistry` and `FAKE_PDF_MEDIA` moved to
  pdf-tool.test-support.ts, which already exists for exactly this. They go
  through `createPdfToolInfraStub(completeMock)` rather than being exported
  directly, because the stub wires the suite's own `complete` mock into the
  model registry and vi.mock handles are file-scoped. All 21 existing call
  sites are unchanged.

Abort propagation into model dispatch
-------------------------------------
The previous revision stopped cancellation at download boundaries and before
the first model call, but `runImagePrompt`/`runPdfPrompt` never saw the signal.
A run cancelled while the first provider request was in flight could still
issue the remaining ones — the image path dispatches `describeImage` once per
image in a sequential loop, so a dead run kept paying for every later image.

Both now take an optional `signal` and check it immediately before each
provider dispatch (3 sites in image-tool, 4 in pdf-tool).

Forwarding the signal further, into the provider transports themselves, is a
different seam and is deliberately left out of this PR.

Also declares `requestInit` on `ImageToolLoadWebMediaOptions`. The local facade
omitted it while the underlying loader accepts it (web-media.ts declares it and
forwards it to readRemoteMediaBuffer), so the option worked at runtime and only
compiled because spread properties skip excess-property checking. A non-spread
call site would not have.

tsgo core + core-test, oxlint, oxfmt clean; pdf-tool and image-tool suites 266
tests pass.

* test(agents): prove in-flight media aborts

* test(agents): narrow PDF loader options

* test(agents): reject abort mocks with errors

* fix(agents): propagate media cancellation

* test(agents): type PDF abort fixture

* test(agents): type prepared runtime snapshot

* fix(agents): normalize abort rejection errors

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Peter Steinberger <steipete@gmail.com>
This commit is contained in:
Masato Hoshino
2026-07-25 21:17:09 +09:00
committed by GitHub
parent b26776fbca
commit bdcc5d54a5
29 changed files with 987 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/codex/media-understanding-provider.test.ts
+98 -1
View File
@@ -4,6 +4,7 @@ import { afterEach, describe, expect, it, vi } from "vitest";
import { buildCodexMediaUnderstandingProvider } from "./media-understanding-provider.js";
import type { CodexAppServerClient } from "./src/app-server/client.js";
import type { CodexServerNotification, JsonValue } from "./src/app-server/protocol.js";
import type { CodexAppServerClientFactory } from "./src/app-server/shared-client.js";
const sharedClientMocks = vi.hoisted(() => ({
createIsolatedCodexAppServerClient: vi.fn(),
@@ -84,9 +85,11 @@ function turnStartResult(status = "inProgress", items: JsonValue[] = []) {
function createFakeClient(options?: {
inputModalities?: string[];
completeWithItems?: boolean;
deferTurnCompletion?: boolean;
notifyError?: string;
approvalRequestMethod?: string;
responseText?: string;
onTurnStart?: () => void;
}) {
const notifications = new Set<(notification: CodexServerNotification) => void>();
const requestHandlers = new Set<(request: { method: string }) => JsonValue | undefined>();
@@ -104,6 +107,7 @@ function createFakeClient(options?: {
return threadStartResult();
}
if (method === "turn/start") {
options?.onTurnStart?.();
if (options?.approvalRequestMethod) {
for (const handler of requestHandlers) {
const response = handler({ method: options.approvalRequestMethod });
@@ -128,7 +132,7 @@ function createFakeClient(options?: {
},
});
}
} else if (!options?.completeWithItems) {
} else if (!options?.completeWithItems && !options?.deferTurnCompletion) {
for (const notify of notifications) {
notify({
method: "item/agentMessage/delta",
@@ -190,6 +194,63 @@ describe("codex media understanding provider", () => {
sharedClientMocks.createIsolatedCodexAppServerClient.mockReset();
});
it("does not start a bounded turn for an already-aborted media request", async () => {
const clientFactory = vi.fn();
const provider = buildCodexMediaUnderstandingProvider({ clientFactory });
const controller = new AbortController();
controller.abort(new Error("caller cancelled Codex media request"));
await expect(
provider.describeImage?.({
buffer: Buffer.from("image-bytes"),
fileName: "image.png",
mime: "image/png",
provider: "codex",
model: "gpt-5.4",
timeoutMs: 30_000,
signal: controller.signal,
cfg: {},
agentDir: "/tmp/openclaw-agent",
}),
).rejects.toThrow("caller cancelled Codex media request");
expect(clientFactory).not.toHaveBeenCalled();
});
it("abandons app-server startup when the media request aborts", async () => {
const clientFactory = vi.fn<CodexAppServerClientFactory>(
async (options) =>
await new Promise<never>((_, reject) => {
options?.abandonSignal?.addEventListener(
"abort",
() => {
const reason = options.abandonSignal?.reason;
reject(reason instanceof Error ? reason : new Error("Codex startup aborted"));
},
{ once: true },
);
}),
);
const provider = buildCodexMediaUnderstandingProvider({ clientFactory });
const controller = new AbortController();
const result = provider.describeImage?.({
buffer: Buffer.from("image-bytes"),
fileName: "image.png",
mime: "image/png",
provider: "codex",
model: "gpt-5.4",
timeoutMs: 30_000,
signal: controller.signal,
cfg: {},
agentDir: "/tmp/openclaw-agent",
});
await vi.waitFor(() => expect(clientFactory).toHaveBeenCalledOnce());
controller.abort(new Error("caller cancelled Codex startup"));
await expect(result).rejects.toThrow("caller cancelled Codex startup");
expect(clientFactory.mock.calls[0]?.[0]?.abandonSignal).toBe(controller.signal);
});
it("runs image understanding through a bounded Codex app-server turn", async () => {
const { client, requests } = createFakeClient();
const clientFactory = vi.fn(async () => client);
@@ -333,6 +394,42 @@ describe("codex media understanding provider", () => {
expect(requests[2]?.params).toEqual(expect.objectContaining({ cwd: "/tmp/openclaw-agent" }));
});
it("interrupts a configured app-server turn when the media request aborts", async () => {
const controller = new AbortController();
const { client, requests } = createFakeClient({
deferTurnCompletion: true,
onTurnStart: () => setTimeout(() => controller.abort(new Error("media cancelled")), 0),
});
const provider = buildCodexMediaUnderstandingProvider({
pluginConfig: {
appServer: {
transport: "websocket",
url: "ws://127.0.0.1:4501",
},
},
clientFactory: async () => client,
});
await expect(
provider.describeImage?.({
buffer: Buffer.from("image-bytes"),
fileName: "image.png",
mime: "image/png",
provider: "codex",
model: "gpt-5.4",
timeoutMs: 30_000,
signal: controller.signal,
cfg: {},
agentDir: "/tmp/openclaw-agent",
}),
).rejects.toThrow();
expect(requests).toContainEqual({
method: "turn/interrupt",
params: { threadId: "thread-1", turnId: "turn-1" },
});
});
it("passes the scoped auth store into isolated app-server startup", async () => {
const { client } = createFakeClient();
sharedClientMocks.createIsolatedCodexAppServerClient.mockResolvedValue(client);
extensions/codex/media-understanding-provider.ts
+5
View File
@@ -51,6 +51,7 @@ export function buildCodexMediaUnderstandingProvider(
prompt: req.prompt,
maxTokens: req.maxTokens,
timeoutMs: req.timeoutMs,
...(req.signal ? { signal: req.signal } : {}),
profile: req.profile,
preferredProfile: req.preferredProfile,
authStore: req.authStore,
@@ -72,12 +73,14 @@ async function describeCodexImages(
if (!model) {
throw new Error("Codex image understanding requires model id.");
}
req.signal?.throwIfAborted();
const { text } = await runBoundedCodexAppServerTurn({
config: req.cfg,
model: { mode: "required", id: model },
profile: req.profile,
timeoutMs: req.timeoutMs,
signal: req.signal,
agentDir: req.agentDir,
authProfileStore: req.authStore,
options,
@@ -115,12 +118,14 @@ async function extractCodexStructured(
if (!req.input.some((entry) => entry.type === "image")) {
throw new Error("Codex structured extraction requires at least one image input.");
}
req.signal?.throwIfAborted();
const { text } = await runBoundedCodexAppServerTurn({
config: req.cfg,
model: { mode: "required", id: model },
profile: req.profile,
timeoutMs: req.timeoutMs,
signal: req.signal,
agentDir: req.agentDir,
authProfileStore: req.authStore,
options,
extensions/codex/src/app-server/attempt-client-cleanup.ts
+16 -5
View File
@@ -93,18 +93,29 @@ export function interruptCodexTurnBestEffort(
timeoutMs?: number;
},
): void {
void interruptCodexTurnAndWaitBestEffort(client, params);
}
/** Sends a bounded turn interrupt and waits for Codex to confirm terminal abort handling. */
export async function interruptCodexTurnAndWaitBestEffort(
client: CodexAppServerClient,
params: {
threadId: string;
turnId: string;
timeoutMs?: number;
},
): Promise<void> {
const requestOptions =
params.timeoutMs && Number.isFinite(params.timeoutMs) && params.timeoutMs > 0
? { timeoutMs: params.timeoutMs }
: undefined;
const requestParams = { threadId: params.threadId, turnId: params.turnId };
try {
const interrupt = requestOptions
// Non-empty interrupts resolve after Codex emits TurnAborted; the empty
// startup form resolves after Op::Interrupt is submitted because no turn exists yet.
await (requestOptions
? client.request("turn/interrupt", requestParams, requestOptions)
: client.request("turn/interrupt", requestParams);
void Promise.resolve(interrupt).catch((error: unknown) => {
embeddedAgentLog.debug("codex app-server turn interrupt failed during abort", { error });
});
: client.request("turn/interrupt", requestParams));
} catch (error) {
embeddedAgentLog.debug("codex app-server turn interrupt failed during abort", { error });
}
extensions/codex/src/app-server/bounded-turn.ts
+37 -3
View File
@@ -4,6 +4,10 @@ import type { AuthProfileStore } from "openclaw/plugin-sdk/agent-runtime";
import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
import { resolveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime";
import { resolvePreferredOpenClawTmpDir, withTempWorkspace } from "openclaw/plugin-sdk/temp-path";
import {
CODEX_APP_SERVER_INTERRUPT_TIMEOUT_MS,
interruptCodexTurnAndWaitBestEffort,
} from "./attempt-client-cleanup.js";
import {
isRetryableErrorNotification,
readCodexNotificationItem,
@@ -154,6 +158,7 @@ async function runBoundedCodexAppServerTurnInWorkspace(
agentDir,
config: params.config,
timeoutMs,
...(params.signal ? { abandonSignal: params.signal } : {}),
})
: await import("./shared-client.js").then(({ createIsolatedCodexAppServerClient }) =>
createIsolatedCodexAppServerClient({
@@ -163,10 +168,30 @@ async function runBoundedCodexAppServerTurnInWorkspace(
agentDir,
authProfileStore: params.authProfileStore,
config: params.config,
...(params.signal ? { abandonSignal: params.signal } : {}),
}),
);
const abortController = new AbortController();
const abortFromCaller = () => abortController.abort(params.signal?.reason ?? "aborted");
let activeThreadId: string | undefined;
let activeTurnId = "";
let interruptPromise: Promise<void> | undefined;
const requestInterrupt = () => {
if (!activeThreadId || interruptPromise) {
return;
}
// Codex serializes start/interrupt per thread; an empty turn id is its
// explicit startup-interrupt contract while turn/start is still resolving.
interruptPromise = interruptCodexTurnAndWaitBestEffort(client, {
threadId: activeThreadId,
turnId: activeTurnId,
timeoutMs: CODEX_APP_SERVER_INTERRUPT_TIMEOUT_MS,
});
};
const abortRun = (reason: unknown) => {
abortController.abort(reason);
requestInterrupt();
};
const abortFromCaller = () => abortRun(params.signal?.reason ?? "aborted");
if (params.signal?.aborted) {
abortFromCaller();
} else {
@@ -174,9 +199,9 @@ async function runBoundedCodexAppServerTurnInWorkspace(
}
const remainingRunMs = deadline - Date.now();
if (remainingRunMs <= 0) {
abortController.abort("timeout");
abortRun("timeout");
}
const timeout = setTimeout(() => abortController.abort("timeout"), Math.max(1, remainingRunMs));
const timeout = setTimeout(() => abortRun("timeout"), Math.max(1, remainingRunMs));
timeout.unref?.();
let retrySelection = false;
@@ -218,6 +243,10 @@ async function runBoundedCodexAppServerTurnInWorkspace(
{ timeoutMs, signal: abortController.signal },
),
);
activeThreadId = thread.thread.id;
if (abortController.signal.aborted) {
requestInterrupt();
}
if (params.requireNoExternalCapabilities) {
// Attest the started thread before injecting historical tool evidence.
// Otherwise inherited MCP state could act on a finalization-only turn.
@@ -254,6 +283,10 @@ async function runBoundedCodexAppServerTurnInWorkspace(
{ timeoutMs, signal: abortController.signal },
),
);
activeTurnId = turn.turn.id;
if (abortController.signal.aborted) {
requestInterrupt();
}
return {
...(await collector.collect(turn.turn, {
timeoutMs,
@@ -274,6 +307,7 @@ async function runBoundedCodexAppServerTurnInWorkspace(
} finally {
clearTimeout(timeout);
params.signal?.removeEventListener("abort", abortFromCaller);
await interruptPromise;
if (ownsClient) {
client.close();
}
extensions/deepgram/audio.ts
+1
View File
@@ -85,6 +85,7 @@ export async function transcribeDeepgramAudio(
headers,
body,
timeoutMs: params.timeoutMs,
...(params.signal ? { signal: params.signal } : {}),
fetchFn,
allowPrivateNetwork,
dispatcherPolicy,
extensions/elevenlabs/media-understanding-provider.ts
+1
View File
@@ -55,6 +55,7 @@ export async function transcribeElevenLabsAudio(
headers,
body: form,
timeoutMs: req.timeoutMs,
...(req.signal ? { signal: req.signal } : {}),
fetchFn,
allowPrivateNetwork,
dispatcherPolicy,
extensions/google/media-understanding-provider.ts
+2
View File
@@ -35,6 +35,7 @@ async function generateGeminiInlineDataText(params: {
model?: string;
prompt?: string;
timeoutMs: number;
signal?: AbortSignal;
fetchFn?: typeof fetch;
defaultBaseUrl: string;
defaultModel: string;
@@ -90,6 +91,7 @@ async function generateGeminiInlineDataText(params: {
headers,
body,
timeoutMs: params.timeoutMs,
...(params.signal ? { signal: params.signal } : {}),
fetchFn,
allowPrivateNetwork,
dispatcherPolicy,
extensions/moonshot/media-understanding-provider.ts
+1
View File
@@ -58,6 +58,7 @@ async function describeMoonshotVideo(
headers,
body,
timeoutMs: params.timeoutMs,
...(params.signal ? { signal: params.signal } : {}),
fetchFn,
allowPrivateNetwork,
dispatcherPolicy,
extensions/openrouter/media-understanding-provider.ts
+1
View File
@@ -141,6 +141,7 @@ async function transcribeOpenRouterAudio(
...(temperature !== undefined ? { temperature } : {}),
},
timeoutMs: params.timeoutMs,
...(params.signal ? { signal: params.signal } : {}),
fetchFn,
allowPrivateNetwork,
dispatcherPolicy,
extensions/qwen/media-understanding-provider.ts
+1
View File
@@ -52,6 +52,7 @@ async function describeQwenVideo(params: VideoDescriptionRequest): Promise<Video
buffer: params.buffer,
}),
timeoutMs: params.timeoutMs,
...(params.signal ? { signal: params.signal } : {}),
fetchFn,
allowPrivateNetwork,
dispatcherPolicy,
extensions/xai/stt.ts
+1
View File
@@ -57,6 +57,7 @@ async function transcribeXaiAudio(
headers,
body: form,
timeoutMs: params.timeoutMs,
...(params.signal ? { signal: params.signal } : {}),
fetchFn,
allowPrivateNetwork,
dispatcherPolicy,
src/agents/minimax-vlm.ts
+2
View File
@@ -92,6 +92,7 @@ export async function minimaxUnderstandImage(params: {
modelBaseUrl?: string;
provider?: string;
timeoutMs?: number;
signal?: AbortSignal;
/** Operator-configured private-network policy from the provider request config. */
allowPrivateNetwork?: boolean;
/** Resolved model request transport metadata, including proxy and TLS policy. */
@@ -152,6 +153,7 @@ export async function minimaxUnderstandImage(params: {
image_url: imageDataUrl,
},
timeoutMs,
...(params.signal ? { signal: params.signal } : {}),
fetchFn: fetch,
allowPrivateNetwork,
ssrfPolicy,
src/agents/model-fallback.test.ts
+28
View File
@@ -4489,6 +4489,34 @@ describe("runWithImageModelFallback", () => {
["google", "gemini-2.5-flash-image-preview"],
]);
});
it("preserves caller cancellation without starting an image fallback", async () => {
const controller = new AbortController();
const reason = new Error("caller cancelled image fallback");
const run = vi.fn(async () => {
controller.abort(reason);
throw Object.assign(new Error("socket hang up"), { code: "ECONNRESET" });
});
await expect(
runWithImageModelFallback({
cfg: makeCfg({
agents: {
defaults: {
imageModel: {
primary: "openai/gpt-5.4-mini",
fallbacks: ["google/gemini-2.5-flash"],
},
},
},
}),
abortSignal: controller.signal,
run,
}),
).rejects.toBe(reason);
expect(run).toHaveBeenCalledOnce();
});
});
describe("runWithModelFallback preserved prompt errors", () => {
src/agents/model-fallback.ts
+5
View File
@@ -2080,6 +2080,7 @@ export async function runWithImageModelFallback<T>(params: {
modelOverride?: string;
run: (provider: string, model: string) => Promise<T>;
onError?: ModelFallbackErrorHandler;
abortSignal?: AbortSignal;
}): Promise<ModelFallbackRunResult<T>> {
const candidates = resolveImageFallbackCandidates({
cfg: params.cfg,
@@ -2102,6 +2103,10 @@ export async function runWithImageModelFallback<T>(params: {
attempts,
attempt: i + 1,
total: candidates.length,
abortSignal: params.abortSignal,
}).catch((error: unknown) => {
params.abortSignal?.throwIfAborted();
throw error;
});
if ("success" in attemptRun) {
return attemptRun.success;
src/agents/tools/image-tool.test.ts
+158
View File
@@ -3299,4 +3299,162 @@ describe("image compression policy", () => {
});
});
});
type MockImageLoadWebMedia = Awaited<
ReturnType<
NonNullable<
NonNullable<Parameters<typeof testing.setProviderDepsForTest>[0]>["loadImageWebMediaRuntime"]
>
>
>["loadWebMedia"];
describe("image tool run abort", () => {
afterEach(() => {
imageProviderHarness.reset();
testing.setProviderDepsForTest();
});
function makeDescribeSpies() {
const describeImage = vi.fn(async (params: ImageDescriptionRequest) => ({
text: "ok",
model: params.model,
}));
const describeImages = vi.fn(async (params: ImagesDescriptionRequest) => ({
text: "ok",
model: params.model,
}));
return { describeImage, describeImages };
}
function installAbortImageDeps(
loadWebMedia: MockImageLoadWebMedia,
spies: ReturnType<typeof makeDescribeSpies>,
providers: MediaUnderstandingProvider[] = [minimaxProvider, moonshotProvider],
) {
installImageUnderstandingProviderDeps(providers, {
loadImageWebMediaRuntime: async () => ({
loadWebMedia,
optimizeImageBufferForWebMedia: async ({ buffer, contentType, fileName }) => ({
buffer,
contentType: contentType ?? "image/png",
kind: "image",
fileName,
}),
}),
describeImageWithModel: spies.describeImage,
describeImagesWithModel: spies.describeImages,
});
}
it("forwards the run signal through the provider request contract", async () => {
vi.stubEnv("MINIMAX_API_KEY", "minimax-test");
const loadWebMedia: MockImageLoadWebMedia = vi.fn(async () => ({
buffer: Buffer.from(ONE_PIXEL_PNG_B64, "base64"),
contentType: "image/png",
kind: "image" as const,
}));
const spies = makeDescribeSpies();
installAbortImageDeps(loadWebMedia, spies, [{ id: "minimax", capabilities: ["image"] }]);
const controller = new AbortController();
await withTempAgentDir(async (agentDir) => {
const tool = createRequiredImageTool({ config: createMinimaxImageConfig(), agentDir });
await tool.execute(
"t1",
{
prompt: "Describe the images.",
images: ["https://example.test/a.png", "https://example.test/b.png"],
},
controller.signal,
);
});
expect(spies.describeImages).toHaveBeenCalledWith(
expect.objectContaining({ signal: controller.signal }),
);
});
it("throws before downloading or calling the provider when the run signal is already aborted", async () => {
vi.stubEnv("MINIMAX_API_KEY", "minimax-test");
const loadWebMedia: MockImageLoadWebMedia = vi.fn(async () => ({
buffer: Buffer.from(ONE_PIXEL_PNG_B64, "base64"),
contentType: "image/png",
kind: "image" as const,
}));
const spies = makeDescribeSpies();
installAbortImageDeps(loadWebMedia, spies);
await withTempAgentDir(async (agentDir) => {
const tool = createRequiredImageTool({ config: createMinimaxImageConfig(), agentDir });
const controller = new AbortController();
controller.abort();
await expect(
tool.execute(
"t1",
{
prompt: "Describe the images.",
images: ["https://example.test/a.png", "https://example.test/b.png"],
},
controller.signal,
),
).rejects.toThrow();
// Aborted run must not spend bandwidth on downloads or a paid vision call.
expect(loadWebMedia).not.toHaveBeenCalled();
expect(spies.describeImage).not.toHaveBeenCalled();
expect(spies.describeImages).not.toHaveBeenCalled();
});
});
it("stops remaining downloads and skips the provider call when aborted mid-run", async () => {
vi.stubEnv("MINIMAX_API_KEY", "minimax-test");
const controller = new AbortController();
let markDownloadStarted: (() => void) | undefined;
const downloadStarted = new Promise<void>((resolve) => {
markDownloadStarted = resolve;
});
const loadWebMedia: MockImageLoadWebMedia = vi.fn(async (_url, options) => {
const downloadSignal = options?.requestInit?.signal;
expect(downloadSignal).toBe(controller.signal);
markDownloadStarted?.();
return await new Promise<never>((_, reject) => {
downloadSignal?.addEventListener(
"abort",
() => reject(new Error("aborted", { cause: downloadSignal.reason })),
{ once: true },
);
});
});
const spies = makeDescribeSpies();
installAbortImageDeps(loadWebMedia, spies);
await withTempAgentDir(async (agentDir) => {
const tool = createRequiredImageTool({ config: createMinimaxImageConfig(), agentDir });
const execution = tool.execute(
"t1",
{
prompt: "Describe the images.",
images: [
"https://example.test/a.png",
"https://example.test/b.png",
"https://example.test/c.png",
],
},
controller.signal,
);
await downloadStarted;
controller.abort();
await expect(execution).rejects.toThrow();
// Only the first image is fetched; the loop exits before the rest and the
// paid vision provider is never called for the dead run.
expect(loadWebMedia).toHaveBeenCalledTimes(1);
expect(spies.describeImage).not.toHaveBeenCalled();
expect(spies.describeImages).not.toHaveBeenCalled();
});
});
});
/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */
src/agents/tools/image-tool.ts
+22 -1
View File
@@ -100,6 +100,7 @@ type ImageToolLoadWebMediaOptions = {
inboundRoots?: readonly string[];
ssrfPolicy?: ReturnType<typeof resolveRemoteMediaSsrfPolicy>;
readIdleTimeoutMs?: number;
requestInit?: RequestInit;
};
type ImageWebMediaRuntime = {
@@ -704,6 +705,7 @@ async function runImagePrompt(params: {
images: Array<{ buffer: Buffer; mimeType: string }>;
workspaceDir?: string;
preparedModelRuntime?: PreparedModelRuntimeSnapshot;
signal?: AbortSignal;
}): Promise<{
text: string;
provider: string;
@@ -723,6 +725,7 @@ async function runImagePrompt(params: {
const result = await runWithImageModelFallback({
cfg: effectiveCfg,
modelOverride: params.modelOverride,
abortSignal: params.signal,
run: async (provider, modelId) => {
const timeoutMs = resolveImageToolTimeoutMs({
cfg: providerCfg,
@@ -740,6 +743,8 @@ async function runImagePrompt(params: {
) {
const describeImages =
imageProvider?.describeImages ?? imageToolProviderDeps.describeImagesWithModel;
// A run cancelled mid-dispatch must not buy another provider call.
params.signal?.throwIfAborted();
const described = await describeImages({
images: params.images.map((image, index) => ({
buffer: image.buffer,
@@ -751,6 +756,7 @@ async function runImagePrompt(params: {
prompt: params.prompt,
maxTokens: resolveImageToolMaxTokens(undefined),
timeoutMs,
...(params.signal ? { signal: params.signal } : {}),
cfg: providerCfg,
...(params.agentId ? { agentId: params.agentId } : {}),
agentDir: params.agentDir,
@@ -769,6 +775,8 @@ async function runImagePrompt(params: {
if (!image) {
throw new Error("Image input disappeared during model execution");
}
// A run cancelled mid-dispatch must not buy another provider call.
params.signal?.throwIfAborted();
const described = await describeImage({
buffer: image.buffer,
fileName: "image-1",
@@ -778,6 +786,7 @@ async function runImagePrompt(params: {
prompt: params.prompt,
maxTokens: resolveImageToolMaxTokens(undefined),
timeoutMs,
...(params.signal ? { signal: params.signal } : {}),
cfg: providerCfg,
...(params.agentId ? { agentId: params.agentId } : {}),
agentDir: params.agentDir,
@@ -792,6 +801,8 @@ async function runImagePrompt(params: {
const parts: string[] = [];
for (const [index, image] of params.images.entries()) {
// A run cancelled mid-dispatch must not buy another provider call.
params.signal?.throwIfAborted();
const described = await describeImage({
buffer: image.buffer,
fileName: `image-${index + 1}`,
@@ -801,6 +812,7 @@ async function runImagePrompt(params: {
prompt: `${params.prompt}\n\nDescribe image ${index + 1} of ${params.images.length}.`,
maxTokens: resolveImageToolMaxTokens(undefined),
timeoutMs,
...(params.signal ? { signal: params.signal } : {}),
cfg: providerCfg,
...(params.agentId ? { agentId: params.agentId } : {}),
agentDir: params.agentDir,
@@ -907,7 +919,7 @@ export function createImageTool(options?: {
maxBytesMb: optionalFiniteNumberSchema({ exclusiveMinimum: 0 }),
maxImages: optionalPositiveIntegerSchema(),
}),
execute: async (_toolCallId, args) => {
execute: async (_toolCallId, args, signal) => {
const record = args && typeof args === "object" ? (args as Record<string, unknown>) : {};
// MARK: - Normalize image + images input and dedupe while preserving order
@@ -1014,6 +1026,9 @@ export function createImageTool(options?: {
const loadedImages: LoadedImageForTool[] = [];
for (const imageRawInput of imageInputs) {
// Stop before starting the next sequential download/decode when the run
// was aborted, so a dead run cannot keep pulling up to maxImages remote images.
signal?.throwIfAborted();
const trimmed = imageRawInput.trim();
const imageRaw = trimmed.startsWith("@") ? trimmed.slice(1).trim() : trimmed;
if (!imageRaw) {
@@ -1126,6 +1141,9 @@ export function createImageTool(options?: {
inboundRoots: mediaInboundRoots,
ssrfPolicy: remoteMediaSsrfPolicy,
...(isHttpUrl ? { readIdleTimeoutMs: REMOTE_MEDIA_READ_IDLE_TIMEOUT_MS } : {}),
// Forward the run abort signal into the fetch layer so an abort
// mid-download disconnects the in-flight socket.
...(signal ? { requestInit: { signal } } : {}),
imageCompression,
});
if (media.kind !== "image") {
@@ -1153,8 +1171,11 @@ export function createImageTool(options?: {
return await buildNativeImageToolResult(loadedImages, options?.config);
}
// Do not issue a paid vision-provider call for an already-aborted run.
signal?.throwIfAborted();
// Text-only runs delegate image understanding to the configured fallback model.
const result = await runImagePrompt({
signal,
cfg: options?.config,
agentId: options?.agentId,
agentDir,
src/agents/tools/pdf-native-providers.ts
+6
View File
@@ -41,6 +41,7 @@ type NativePdfJsonRequest = {
failureLabel: string;
responseLabel: string;
nonJsonMessage: string;
signal?: AbortSignal;
};
async function postNativePdfJson(params: NativePdfJsonRequest): Promise<Record<string, unknown>> {
@@ -56,6 +57,7 @@ async function postNativePdfJson(params: NativePdfJsonRequest): Promise<Record<s
headers,
body: params.body,
timeoutMs: NATIVE_PDF_PROVIDER_FETCH_TIMEOUT_MS,
...(params.signal ? { signal: params.signal } : {}),
fetchFn: fetch,
allowPrivateNetwork: params.allowPrivateNetwork,
ssrfPolicy: params.ssrfPolicy,
@@ -113,6 +115,7 @@ export async function anthropicAnalyzePdf(params: {
maxTokens?: number;
baseUrl?: string;
requestConfig?: NativePdfProviderRequestConfig;
signal?: AbortSignal;
}): Promise<string> {
const apiKey = normalizeSecretInput(params.apiKey);
if (!apiKey) {
@@ -170,6 +173,7 @@ export async function anthropicAnalyzePdf(params: {
failureLabel: "Anthropic PDF request failed",
responseLabel: "Anthropic PDF response",
nonJsonMessage: "Anthropic PDF response was not JSON.",
signal: params.signal,
});
const responseContent = json.content as AnthropicResponseContent | undefined;
@@ -206,6 +210,7 @@ export async function geminiAnalyzePdf(params: {
pdfs: PdfInput[];
baseUrl?: string;
requestConfig?: NativePdfProviderRequestConfig;
signal?: AbortSignal;
}): Promise<string> {
const apiKey = normalizeSecretInput(params.apiKey);
if (!apiKey) {
@@ -266,6 +271,7 @@ export async function geminiAnalyzePdf(params: {
failureLabel: "Gemini PDF request failed",
responseLabel: "Gemini PDF response",
nonJsonMessage: "Gemini PDF response was not JSON.",
signal: params.signal,
});
const candidates = json.candidates as GeminiCandidate[] | undefined;
src/agents/tools/pdf-tool.runtime-abort.test.ts
+115
View File
@@ -0,0 +1,115 @@
// PDF runtime-abort coverage keeps prepared-runtime acquisition cancellable and leak-free.
import { afterEach, describe, expect, it, vi } from "vitest";
import type { OpenClawConfig } from "../../config/config.js";
import * as pdfExtractModule from "../../media/pdf-extract.js";
import * as preparedModelRuntime from "../prepared-model-runtime.js";
import { createPdfToolInfraStub, withTempPdfAgentDir } from "./pdf-tool.test-support.js";
const completeMock = vi.hoisted(() => vi.fn());
const registerProviderStreamForModelMock = vi.hoisted(() => vi.fn());
vi.mock("../../llm/stream.js", async () => {
const actual = await vi.importActual<typeof import("../../llm/stream.js")>("../../llm/stream.js");
return { ...actual, complete: completeMock };
});
vi.mock("../provider-stream.js", () => ({
registerProviderStreamForModel: registerProviderStreamForModelMock,
}));
const { createPdfModelRegistry, stubPdfToolInfra } = createPdfToolInfraStub(completeMock);
describe("PDF tool prepared-runtime cancellation", () => {
afterEach(() => {
completeMock.mockReset();
vi.restoreAllMocks();
});
it("rejects before deferred acquisition resolves, then releases the late lease", async () => {
await withTempPdfAgentDir(async (agentDir) => {
await stubPdfToolInfra(agentDir, { provider: "anthropic" });
const cfg = {
agents: { defaults: { pdfModel: { primary: "anthropic/claude-opus-4-6" } } },
} as OpenClawConfig;
const modelRegistry = createPdfModelRegistry(() => ({
provider: "anthropic",
api: "anthropic-messages",
maxTokens: 8192,
input: ["text", "document"],
}));
const release = vi.fn();
let finishAcquisition!: (
value: Awaited<ReturnType<typeof preparedModelRuntime.acquireAgentRunPreparedModelRuntime>>,
) => void;
vi.mocked(preparedModelRuntime.acquireAgentRunPreparedModelRuntime).mockImplementationOnce(
() =>
new Promise((resolve) => {
finishAcquisition = resolve;
}),
);
const tool = (await import("./pdf-tool.js")).createPdfTool({ config: cfg, agentDir });
if (!tool) {
throw new Error("expected PDF tool");
}
const controller = new AbortController();
const execution = tool.execute(
"t1",
{ prompt: "summarize", pdf: "/tmp/a.pdf" },
controller.signal,
);
await vi.waitFor(() =>
expect(preparedModelRuntime.acquireAgentRunPreparedModelRuntime).toHaveBeenCalledOnce(),
);
const assertion = expect(execution).rejects.toThrow("PDF runtime cancelled");
controller.abort(new Error("PDF runtime cancelled"));
await assertion;
expect(release).not.toHaveBeenCalled();
finishAcquisition({
snapshot: {
agentDir,
config: cfg,
// Cancellation releases this late lease before its stores can be used.
createStores: () => ({ authStorage: {}, modelRegistry }),
} as never,
release,
});
await vi.waitFor(() => expect(release).toHaveBeenCalledOnce());
expect(completeMock).not.toHaveBeenCalled();
});
});
it("releases the runtime when a generic provider ignores cancellation", async () => {
await withTempPdfAgentDir(async (agentDir) => {
const { release } = await stubPdfToolInfra(agentDir, { provider: "openai" });
vi.spyOn(pdfExtractModule, "extractPdfContent").mockResolvedValue({
text: "extractable text",
images: [],
});
completeMock.mockImplementationOnce(() => new Promise(() => {}));
const cfg = {
agents: { defaults: { pdfModel: { primary: "openai/gpt-5.4-mini" } } },
} as OpenClawConfig;
const tool = (await import("./pdf-tool.js")).createPdfTool({ config: cfg, agentDir });
if (!tool) {
throw new Error("expected PDF tool");
}
const controller = new AbortController();
const execution = tool.execute(
"t1",
{ prompt: "summarize", pdf: "/tmp/a.pdf" },
controller.signal,
);
await vi.waitFor(() => expect(completeMock).toHaveBeenCalledOnce());
const options = completeMock.mock.calls[0]?.[2];
expect(options?.signal).toBe(controller.signal);
const assertion = expect(execution).rejects.toThrow("PDF provider cancelled");
controller.abort(new Error("PDF provider cancelled"));
await assertion;
expect(release).toHaveBeenCalledOnce();
});
});
});
src/agents/tools/pdf-tool.test-support.ts
+92 -1
View File
@@ -3,7 +3,15 @@
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { vi } from "vitest";
import { type Mock, vi } from "vitest";
import * as webMedia from "../../media/web-media.js";
import * as modelAuth from "../model-auth.js";
import * as modelsConfig from "../models-config.js";
import * as preparedModelRuntime from "../prepared-model-runtime.js";
import {
getModelRegistryRuntime,
initializeModelRegistryRuntime,
} from "../sessions/model-registry-runtime.js";
export async function withTempPdfAgentDir<T>(run: (agentDir: string) => Promise<T>): Promise<T> {
const agentDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-pdf-"));
@@ -27,3 +35,86 @@ export function resetPdfToolAuthEnv(): void {
vi.stubEnv("GH_TOKEN", "");
vi.stubEnv("GITHUB_TOKEN", "");
}
export const FAKE_PDF_MEDIA = {
kind: "document",
buffer: Buffer.from("%PDF-1.4 fake"),
contentType: "application/pdf",
fileName: "doc.pdf",
} as const;
// The PDF tool suites share this module-boundary stub. It is built through a
// factory rather than exported directly because it wires the suite's own
// `complete` mock into the model registry, and vi.mock handles are file-scoped
// — a plain export would capture the wrong (or no) mock.
export function createPdfToolInfraStub(completeMock: Mock) {
function createPdfModelRegistry(find: () => unknown) {
const modelRegistry = { find };
initializeModelRegistryRuntime(modelRegistry);
getModelRegistryRuntime(modelRegistry).llmRuntime.complete = completeMock;
return modelRegistry;
}
async function stubPdfToolInfra(
agentDir: string,
params?: {
mockLoad?: boolean;
provider?: string;
input?: string[];
api?: string;
modelFound?: boolean;
},
) {
// Keep PDF tool tests focused on orchestration; provider discovery, auth, and
// remote media loading are replaced with narrow spies at the module boundary.
const loadSpy = vi.spyOn(webMedia, "loadWebMediaRaw");
if (params?.mockLoad !== false) {
loadSpy.mockResolvedValue(FAKE_PDF_MEDIA as never);
}
const setRuntimeApiKey = vi.fn();
const authStorage = { setRuntimeApiKey };
const find =
params?.modelFound === false
? () => null
: () =>
({
provider: params?.provider ?? "anthropic",
api:
params?.api ??
(params?.provider === "openai"
? "openai-chatgpt-responses"
: params?.provider === "openai"
? "openai-responses"
: "anthropic-messages"),
maxTokens: 8192,
input: params?.input ?? ["text", "document"],
}) as never;
const modelRegistry = createPdfModelRegistry(find);
const release = vi.fn();
vi.spyOn(preparedModelRuntime, "acquireAgentRunPreparedModelRuntime").mockImplementation(
async (input) =>
({
snapshot: {
agentDir: input.agentDir,
config: input.config,
workspaceDir: input.workspaceDir,
createStores: () => ({ authStorage, modelRegistry }),
},
release,
}) as never,
);
vi.spyOn(modelsConfig, "ensureOpenClawModelsJson").mockResolvedValue({
agentDir,
wrote: false,
});
vi.spyOn(modelAuth, "getApiKeyForModel").mockResolvedValue({ apiKey: "test-key" } as never);
vi.spyOn(modelAuth, "requireApiKey").mockReturnValue("test-key");
return { loadSpy, release, setRuntimeApiKey };
}
return { createPdfModelRegistry, stubPdfToolInfra };
}
src/agents/tools/pdf-tool.test.ts
+82 -79
View File
@@ -10,15 +10,15 @@ import * as webMedia from "../../media/web-media.js";
import { withEnvAsync } from "../../test-utils/env.js";
import type { AuthProfileStore } from "../auth-profiles/types.js";
import * as modelAuth from "../model-auth.js";
import * as modelsConfig from "../models-config.js";
import * as preparedModelRuntime from "../prepared-model-runtime.js";
import {
getModelRegistryRuntime,
initializeModelRegistryRuntime,
} from "../sessions/model-registry-runtime.js";
import * as pdfNativeProviders from "./pdf-native-providers.js";
import * as pdfModelConfigModule from "./pdf-tool.model-config.js";
import { resetPdfToolAuthEnv, withTempPdfAgentDir } from "./pdf-tool.test-support.js";
import {
createPdfToolInfraStub,
FAKE_PDF_MEDIA,
resetPdfToolAuthEnv,
withTempPdfAgentDir,
} from "./pdf-tool.test-support.js";
const completeMock = vi.hoisted(() => vi.fn());
const registerProviderStreamForModelMock = vi.hoisted(() => vi.fn());
@@ -35,6 +35,8 @@ vi.mock("../provider-stream.js", () => ({
registerProviderStreamForModel: registerProviderStreamForModelMock,
}));
const { createPdfModelRegistry, stubPdfToolInfra } = createPdfToolInfraStub(completeMock);
type PdfToolModule = typeof import("./pdf-tool.js");
let createPdfTool: PdfToolModule["createPdfTool"];
@@ -49,12 +51,6 @@ const ANTHROPIC_PDF_MODEL = "anthropic/claude-opus-4-6";
const GOOGLE_PDF_MODEL = "google/gemini-2.5-pro";
const OPENAI_PDF_MODEL = "openai/gpt-5.4-mini";
const CODEX_PDF_MODEL = "openai/gpt-5.4";
const FAKE_PDF_MEDIA = {
kind: "document",
buffer: Buffer.from("%PDF-1.4 fake"),
contentType: "application/pdf",
fileName: "doc.pdf",
} as const;
function requirePdfTool(
tool: Awaited<ReturnType<typeof loadCreatePdfTool>> extends (...args: any[]) => infer R
@@ -118,73 +114,6 @@ function firstCompletionContext(): { systemPrompt?: string } | undefined {
return context;
}
function createPdfModelRegistry(find: () => unknown) {
const modelRegistry = { find };
initializeModelRegistryRuntime(modelRegistry);
getModelRegistryRuntime(modelRegistry).llmRuntime.complete = completeMock;
return modelRegistry;
}
async function stubPdfToolInfra(
agentDir: string,
params?: {
mockLoad?: boolean;
provider?: string;
input?: string[];
api?: string;
modelFound?: boolean;
},
) {
// Keep PDF tool tests focused on orchestration; provider discovery, auth, and
// remote media loading are replaced with narrow spies at the module boundary.
const loadSpy = vi.spyOn(webMedia, "loadWebMediaRaw");
if (params?.mockLoad !== false) {
loadSpy.mockResolvedValue(FAKE_PDF_MEDIA as never);
}
const setRuntimeApiKey = vi.fn();
const authStorage = { setRuntimeApiKey };
const find =
params?.modelFound === false
? () => null
: () =>
({
provider: params?.provider ?? "anthropic",
api:
params?.api ??
(params?.provider === "openai"
? "openai-chatgpt-responses"
: params?.provider === "openai"
? "openai-responses"
: "anthropic-messages"),
maxTokens: 8192,
input: params?.input ?? ["text", "document"],
}) as never;
const modelRegistry = createPdfModelRegistry(find);
vi.spyOn(preparedModelRuntime, "acquireAgentRunPreparedModelRuntime").mockImplementation(
async (input) =>
({
snapshot: {
agentDir: input.agentDir,
config: input.config,
workspaceDir: input.workspaceDir,
createStores: () => ({ authStorage, modelRegistry }),
},
release: vi.fn(),
}) as never,
);
vi.spyOn(modelsConfig, "ensureOpenClawModelsJson").mockResolvedValue({
agentDir,
wrote: false,
});
vi.spyOn(modelAuth, "getApiKeyForModel").mockResolvedValue({ apiKey: "test-key" } as never);
vi.spyOn(modelAuth, "requireApiKey").mockReturnValue("test-key");
return { loadSpy, setRuntimeApiKey };
}
async function withManagedInboundPdf(
run: (params: { stateDir: string; mediaId: string; mediaPath: string }) => Promise<void>,
) {
@@ -1062,4 +991,78 @@ describe("createPdfTool", () => {
});
});
});
it("throws before loading or calling the model when the run signal is already aborted", async () => {
await withTempPdfAgentDir(async (agentDir) => {
const { loadSpy } = await stubPdfToolInfra(agentDir, { provider: "anthropic" });
const nativeSpy = vi.spyOn(pdfNativeProviders, "anthropicAnalyzePdf");
nativeSpy.mockResolvedValue("native summary");
const tool = requirePdfTool(
(await loadCreatePdfTool())({ config: withPdfModel(ANTHROPIC_PDF_MODEL), agentDir }),
);
const controller = new AbortController();
controller.abort();
await expect(
tool.execute(
"t1",
{ prompt: "summarize", pdfs: ["/tmp/a.pdf", "/tmp/b.pdf"] },
controller.signal,
),
).rejects.toThrow();
// Aborted run must not spend bandwidth on downloads or a paid model call.
expect(loadSpy).not.toHaveBeenCalled();
expect(nativeSpy).not.toHaveBeenCalled();
expect(completeMock).not.toHaveBeenCalled();
});
});
it("stops remaining downloads and skips the model call when aborted mid-run", async () => {
await withTempPdfAgentDir(async (agentDir) => {
const { loadSpy } = await stubPdfToolInfra(agentDir, {
mockLoad: false,
provider: "anthropic",
});
const controller = new AbortController();
let markDownloadStarted: (() => void) | undefined;
const downloadStarted = new Promise<void>((resolve) => {
markDownloadStarted = resolve;
});
loadSpy.mockImplementation(async (_url, options) => {
const downloadSignal =
typeof options === "object" ? options.requestInit?.signal : undefined;
expect(downloadSignal).toBe(controller.signal);
markDownloadStarted?.();
return await new Promise<never>((_, reject) => {
downloadSignal?.addEventListener(
"abort",
() => reject(new Error("aborted", { cause: downloadSignal.reason })),
{ once: true },
);
});
});
const nativeSpy = vi.spyOn(pdfNativeProviders, "anthropicAnalyzePdf");
nativeSpy.mockResolvedValue("native summary");
const tool = requirePdfTool(
(await loadCreatePdfTool())({ config: withPdfModel(ANTHROPIC_PDF_MODEL), agentDir }),
);
const execution = tool.execute(
"t1",
{ prompt: "summarize", pdfs: ["/tmp/a.pdf", "/tmp/b.pdf", "/tmp/c.pdf"] },
controller.signal,
);
await downloadStarted;
controller.abort();
await expect(execution).rejects.toThrow();
// Only the first PDF is fetched; the loop exits before the rest and the
// paid model call never fires for the dead run.
expect(loadSpy).toHaveBeenCalledTimes(1);
expect(nativeSpy).not.toHaveBeenCalled();
expect(completeMock).not.toHaveBeenCalled();
});
});
});
src/agents/tools/pdf-tool.ts
+62 -12
View File
@@ -21,6 +21,7 @@ import { loadWebMediaRaw } from "../../media/web-media.js";
import { resolveUserPath } from "../../utils.js";
import { resolveDefaultAgentDir } from "../agent-scope.js";
import type { AuthProfileStore } from "../auth-profiles/types.js";
import { abortable } from "../embedded-agent-runner/run/abortable.js";
import { applySecretRefHeaderSentinels } from "../model-auth.js";
import {
acquireAgentRunPreparedModelRuntime,
@@ -157,6 +158,7 @@ async function runPdfPrompt(params: {
password?: string;
pageNumbers?: number[];
getExtractions: () => Promise<PdfExtractedContent[]>;
signal?: AbortSignal;
}): Promise<{
text: string;
provider: string;
@@ -166,17 +168,34 @@ async function runPdfPrompt(params: {
}> {
const requestedCfg = applyImageModelConfigDefaults(params.cfg, params.pdfModelConfig);
const preparedRuntimeLease = params.preparedModelRuntime
? { snapshot: params.preparedModelRuntime, release: () => {} }
: await acquireAgentRunPreparedModelRuntime({
agentDir: params.agentDir,
...(params.agentId ? { agentId: params.agentId } : {}),
config: requestedCfg ?? {},
inheritedAuthDir: resolveDefaultAgentDir(requestedCfg ?? {}),
...(params.workspaceDir ? { workspaceDir: params.workspaceDir } : {}),
});
let preparedRuntimeLease: Awaited<ReturnType<typeof acquireAgentRunPreparedModelRuntime>>;
if (params.preparedModelRuntime) {
preparedRuntimeLease = { snapshot: params.preparedModelRuntime, release: () => {} };
} else {
const acquireRuntime = acquireAgentRunPreparedModelRuntime({
agentDir: params.agentDir,
...(params.agentId ? { agentId: params.agentId } : {}),
config: requestedCfg ?? {},
inheritedAuthDir: resolveDefaultAgentDir(requestedCfg ?? {}),
...(params.workspaceDir ? { workspaceDir: params.workspaceDir } : {}),
});
try {
preparedRuntimeLease = params.signal
? await abortable(params.signal, acquireRuntime)
: await acquireRuntime;
} catch (error) {
if (params.signal?.aborted) {
void acquireRuntime.then(
(late) => late.release(),
() => undefined,
);
}
throw error;
}
}
try {
params.signal?.throwIfAborted();
const preparedRuntime = preparedRuntimeLease.snapshot;
const runtimeAgentDir = preparedRuntime.agentDir;
const runtimeWorkspaceDir = preparedRuntime.workspaceDir ?? params.workspaceDir;
@@ -205,6 +224,7 @@ async function runPdfPrompt(params: {
const result = await runWithImageModelFallback({
cfg: effectiveCfg,
modelOverride: params.modelOverride,
abortSignal: params.signal,
run: async (provider, modelId) => {
const model = bindModelLlmRuntime(
applySecretRefHeaderSentinels(
@@ -238,6 +258,8 @@ async function runPdfPrompt(params: {
}));
if (provider === "anthropic") {
// A run cancelled mid-dispatch must not buy another provider call.
params.signal?.throwIfAborted();
const text = await anthropicAnalyzePdf({
apiKey,
modelId,
@@ -249,11 +271,14 @@ async function runPdfPrompt(params: {
headers: model.headers,
request: getModelProviderRequestTransport(model),
},
signal: params.signal,
});
return { text, provider, model: modelId, native: true };
}
if (provider === "google") {
// A run cancelled mid-dispatch must not buy another provider call.
params.signal?.throwIfAborted();
const text = await geminiAnalyzePdf({
apiKey,
modelId,
@@ -264,6 +289,7 @@ async function runPdfPrompt(params: {
headers: model.headers,
request: getModelProviderRequestTransport(model),
},
signal: params.signal,
});
return { text, provider, model: modelId, native: true };
}
@@ -293,19 +319,31 @@ async function runPdfPrompt(params: {
images: [],
}));
const context = buildPdfExtractionContext(params.prompt, textOnlyExtractions, model);
const message = await complete(model, context, {
// A run cancelled mid-dispatch must not buy another provider call.
params.signal?.throwIfAborted();
const completion = complete(model, context, {
apiKey,
maxTokens: resolvePdfToolMaxTokens(model.maxTokens),
signal: params.signal,
});
const message = params.signal
? await abortable(params.signal, completion)
: await completion;
const text = coercePdfAssistantText({ message, provider, model: modelId });
return { text, provider, model: modelId, native: false };
}
const context = buildPdfExtractionContext(params.prompt, extractions, model);
const message = await complete(model, context, {
// A run cancelled mid-dispatch must not buy another provider call.
params.signal?.throwIfAborted();
const completion = complete(model, context, {
apiKey,
maxTokens: resolvePdfToolMaxTokens(model.maxTokens),
signal: params.signal,
});
const message = params.signal
? await abortable(params.signal, completion)
: await completion;
const text = coercePdfAssistantText({ message, provider, model: modelId });
return { text, provider, model: modelId, native: false };
},
@@ -392,7 +430,7 @@ export function createPdfTool(options?: {
name: "pdf",
description,
parameters: PdfToolSchema,
execute: async (_toolCallId, args) => {
execute: async (_toolCallId, args, signal) => {
const record = args && typeof args === "object" ? (args as Record<string, unknown>) : {};
// MARK: - Normalize pdf + pdfs input
@@ -459,6 +497,9 @@ export function createPdfTool(options?: {
}> = [];
for (const pdfRaw of pdfInputs) {
// Stop before starting the next sequential download when the run was
// aborted, so a dead run cannot keep pulling remote PDFs.
signal?.throwIfAborted();
const trimmed = normalizeMediaReferenceSource(pdfRaw);
const refInfo = classifyMediaReferenceSource(trimmed);
const { isHttpUrl } = refInfo;
@@ -519,6 +560,9 @@ export function createPdfTool(options?: {
localRoots,
...(isHttpUrl ? { readIdleTimeoutMs: REMOTE_MEDIA_READ_IDLE_TIMEOUT_MS } : {}),
ssrfPolicy: remoteMediaSsrfPolicy,
// Forward the run abort signal into the fetch layer so an abort
// mid-download disconnects the in-flight socket.
...(signal ? { requestInit: { signal } } : {}),
});
if (media.kind !== "document") {
@@ -550,6 +594,9 @@ export function createPdfTool(options?: {
const getExtractions = async (): Promise<PdfExtractedContent[]> => {
const extractedAll: PdfExtractedContent[] = [];
for (const pdf of loadedPdfs) {
// Extraction is sequential and can be CPU-heavy. Do not start the next
// document after the owning agent run has been cancelled.
signal?.throwIfAborted();
const extracted = await extractPdfContent({
buffer: pdf.buffer,
maxPages: configuredMaxPages,
@@ -564,7 +611,10 @@ export function createPdfTool(options?: {
return extractedAll;
};
// Do not issue a paid PDF-model call for an already-aborted run.
signal?.throwIfAborted();
const result = await runPdfPrompt({
signal,
cfg: options?.config,
agentId: options?.agentId,
agentDir,
src/media-understanding/image.runtime-timeout.test.ts
+124
View File
@@ -620,6 +620,50 @@ describe("describeImageWithModel", () => {
},
);
it("does not start the reasoning-only retry after caller cancellation", async () => {
const controller = new AbortController();
discoverModelsMock.mockReturnValue({
find: vi.fn(() => ({
api: "openai-responses",
provider: "openai",
id: "gpt-5.4-mini",
input: ["text", "image"],
baseUrl: "https://api.openai.com/v1",
})),
});
completeMock.mockImplementationOnce(async () => {
controller.abort(new Error("caller cancelled image description"));
return {
role: "assistant",
api: "openai-responses",
provider: "openai",
model: "gpt-5.4-mini",
stopReason: "stop",
timestamp: Date.now(),
content: [{ type: "thinking", thinking: "internal", thinkingSignature: "reasoning" }],
};
});
await expect(
describeImageWithModel({
cfg: {},
agentDir: "/tmp/openclaw-agent",
provider: "openai",
model: "gpt-5.4-mini",
buffer: Buffer.from("png-bytes"),
fileName: "image.png",
mime: "image/png",
prompt: "Describe the image.",
timeoutMs: 1000,
signal: controller.signal,
}),
).rejects.toThrow("caller cancelled image description");
expect(completeMock).toHaveBeenCalledOnce();
const options = requireFirstMockCall(completeMock, "cancelled image completion")[2];
expect(options?.signal?.aborted).toBe(true);
});
it("rejects when a generic image completion ignores the abort signal", async () => {
vi.useFakeTimers();
discoverModelsMock.mockReturnValue({
@@ -659,6 +703,39 @@ describe("describeImageWithModel", () => {
expect(options.timeoutMs).toBe(25);
});
it("releases the prepared runtime when a provider ignores caller cancellation", async () => {
discoverModelsMock.mockReturnValue({
find: vi.fn(() => ({
api: "openai-responses",
provider: "openai",
id: "gpt-5.4-mini",
input: ["text", "image"],
baseUrl: "https://api.openai.com/v1",
})),
});
completeMock.mockImplementation(() => new Promise(() => {}));
const controller = new AbortController();
const result = describeImageWithModel({
cfg: {},
agentDir: "/tmp/openclaw-agent",
provider: "openai",
model: "gpt-5.4-mini",
buffer: Buffer.from("png-bytes"),
fileName: "image.png",
mime: "image/png",
prompt: "Describe the image.",
timeoutMs: 60_000,
signal: controller.signal,
});
await vi.waitFor(() => expect(completeMock).toHaveBeenCalledOnce());
const assertion = expect(result).rejects.toThrow("caller cancelled provider request");
controller.abort(new Error("caller cancelled provider request"));
await assertion;
expect(releasePreparedModelRuntimeMock).toHaveBeenCalledOnce();
});
it("keeps the full configured timeout for provider requests after slow setup", async () => {
vi.useFakeTimers();
const slowSetupMs = 400;
@@ -791,4 +868,51 @@ describe("describeImageWithModel", () => {
await vi.waitFor(() => expect(releasePreparedModelRuntimeMock).toHaveBeenCalledOnce());
expect(completeMock).not.toHaveBeenCalled();
});
it("releases a prepared generation when cancellation wins during setup", async () => {
let finishResolution!: (value: {
authStorage: typeof preparedAuthStorage;
model: { provider: string; id: string; api: string; input: string[] };
modelRegistry: object;
}) => void;
resolveModelAsyncMock.mockImplementationOnce(
() =>
new Promise((resolve) => {
finishResolution = resolve;
}),
);
const controller = new AbortController();
const result = describeImageWithModel({
cfg: {},
agentDir: "/tmp/openclaw-agent",
provider: "openai",
model: "gpt-5.4-mini",
buffer: Buffer.from("png-bytes"),
fileName: "image.png",
mime: "image/png",
prompt: "Describe the image.",
timeoutMs: 1000,
signal: controller.signal,
});
await vi.waitFor(() => expect(resolveModelAsyncMock).toHaveBeenCalledOnce());
const assertion = expect(result).rejects.toThrow("caller cancelled during setup");
controller.abort(new Error("caller cancelled during setup"));
await assertion;
expect(releasePreparedModelRuntimeMock).not.toHaveBeenCalled();
finishResolution({
authStorage: preparedAuthStorage,
model: {
provider: "openai",
id: "gpt-5.4-mini",
api: "openai-responses",
input: ["text", "image"],
},
modelRegistry: {},
});
await vi.waitFor(() => expect(releasePreparedModelRuntimeMock).toHaveBeenCalledOnce());
expect(completeMock).not.toHaveBeenCalled();
});
});
src/media-understanding/image.test.ts
+30 -1
View File
@@ -187,7 +187,7 @@ vi.mock("../infra/net/fetch-guard.js", async () => {
};
});
const { describeImageWithModel } = await import("./image.js");
const { describeImageWithModel, describeImagesWithModel } = await import("./image.js");
describe("describeImageWithModel", () => {
afterEach(() => {
@@ -328,6 +328,35 @@ describe("describeImageWithModel", () => {
expect(completeMock).not.toHaveBeenCalled();
});
it("does not start another MiniMax request after caller cancellation", async () => {
const controller = new AbortController();
fetchMock.mockImplementationOnce(async () => {
controller.abort(new Error("caller cancelled MiniMax image batch"));
return Response.json({
base_resp: { status_code: 0 },
content: "first image",
});
});
await expect(
describeImagesWithModel({
cfg: {},
agentDir: "/tmp/openclaw-agent",
provider: "minimax-portal",
model: "MiniMax-VL-01",
images: [
{ buffer: Buffer.from("first"), fileName: "first.png", mime: "image/png" },
{ buffer: Buffer.from("second"), fileName: "second.png", mime: "image/png" },
],
timeoutMs: 1000,
signal: controller.signal,
}),
).rejects.toThrow("caller cancelled MiniMax image batch");
expect(fetchMock).toHaveBeenCalledOnce();
expect(completeMock).not.toHaveBeenCalled();
});
it("carries resolved MiniMax model transport policy into the VLM request", async () => {
discoverModelsMock.mockReturnValue({
find: vi.fn(() =>
src/media-understanding/image.ts
+54 -6
View File
@@ -194,6 +194,7 @@ async function describeImagesWithMinimax(params: {
images: Array<{ buffer: Buffer; mime?: string }>;
allowPrivateNetwork?: boolean;
request?: ModelProviderRequestTransportOverrides;
signal?: AbortSignal;
}): Promise<ImagesDescriptionResult> {
const responses: string[] = [];
// MiniMax VLM handles its own outbound fetch, so unwrap only at this final handoff.
@@ -203,6 +204,9 @@ async function describeImagesWithMinimax(params: {
);
const apiKey = runtimeValue;
for (const [index, image] of params.images.entries()) {
// One MiniMax request is issued per image, so cancellation must gate every
// iteration or a dead run can continue buying calls after the first image.
params.signal?.throwIfAborted();
const prompt =
params.images.length > 1
? `${params.prompt}\n\nDescribe image ${index + 1} of ${params.images.length} independently.`
@@ -216,6 +220,7 @@ async function describeImagesWithMinimax(params: {
timeoutMs: params.timeoutMs,
allowPrivateNetwork: params.allowPrivateNetwork,
request: params.request,
signal: params.signal,
});
responses.push(params.images.length > 1 ? `Image ${index + 1}:\n${text.trim()}` : text.trim());
}
@@ -353,23 +358,52 @@ async function withImageDescriptionTimeout<T>(params: {
task: Promise<T>;
timeoutMs: number | undefined;
controller: AbortController;
signal?: AbortSignal;
createTimeoutError: (timeoutMs: number) => Error;
}): Promise<T> {
if (params.timeoutMs === undefined) {
params.signal?.throwIfAborted();
if (params.timeoutMs === undefined && !params.signal) {
return await params.task;
}
let timeout: NodeJS.Timeout | undefined;
try {
return await Promise.race([
params.task,
let removeAbortListener: (() => void) | undefined;
const races: Promise<T>[] = [params.task];
if (params.timeoutMs !== undefined) {
races.push(
new Promise<never>((_, reject) => {
timeout = setTimeout(() => {
params.controller.abort();
reject(params.createTimeoutError(params.timeoutMs!));
}, params.timeoutMs);
}),
]);
);
}
if (params.signal) {
races.push(
new Promise<never>((_, reject) => {
const onAbort = () => {
try {
params.signal?.throwIfAborted();
} catch (error) {
reject(
error instanceof Error
? error
: new Error("image description aborted", { cause: error }),
);
}
};
params.signal?.addEventListener("abort", onAbort, { once: true });
removeAbortListener = () => params.signal?.removeEventListener("abort", onAbort);
if (params.signal?.aborted) {
onAbort();
}
}),
);
}
try {
return await Promise.race(races);
} finally {
removeAbortListener?.();
if (timeout) {
clearTimeout(timeout);
}
@@ -381,8 +415,12 @@ async function describeImagesWithModelInternal(
options: { onPayload?: ProviderStreamOptions["onPayload"] } = {},
): Promise<ImagesDescriptionResult> {
const prompt = params.prompt ?? "Describe the image.";
params.signal?.throwIfAborted();
const startedAtMs = Date.now();
const controller = new AbortController();
const requestSignal = params.signal
? AbortSignal.any([params.signal, controller.signal])
: controller.signal;
const configuredTimeoutMs = resolveImageDescriptionTimeoutMs(params.timeoutMs);
const allowPrivateNetwork = resolveConfiguredProviderAllowPrivateNetwork(
params.cfg,
@@ -396,6 +434,7 @@ async function describeImagesWithModelInternal(
try {
const resolved = await withImageDescriptionTimeout({
controller,
signal: params.signal,
timeoutMs: configuredTimeoutMs,
createTimeoutError: (timeoutMs) =>
buildImageDescriptionTimeoutError({ phase: "setup", timeoutMs }),
@@ -411,11 +450,13 @@ async function describeImagesWithModelInternal(
(late) => late.release(),
() => undefined,
);
params.signal?.throwIfAborted();
if (!isMinimaxVlmModel(params.provider, params.model) || !isUnknownModelError(err)) {
throw err;
}
const fallback = await withImageDescriptionTimeout({
controller,
signal: params.signal,
timeoutMs: configuredTimeoutMs,
createTimeoutError: (timeoutMs) =>
buildImageDescriptionTimeoutError({ phase: "setup", timeoutMs }),
@@ -430,11 +471,13 @@ async function describeImagesWithModelInternal(
timeoutMs: params.timeoutMs,
images: params.images,
allowPrivateNetwork,
signal: params.signal,
});
}
const apiKey = runtimeValue;
try {
params.signal?.throwIfAborted();
const setupDurationMs = Date.now() - startedAtMs;
if (isMinimaxVlmModel(model.provider, model.id)) {
@@ -447,6 +490,7 @@ async function describeImagesWithModelInternal(
timeoutMs: params.timeoutMs,
images: params.images,
request: getModelProviderRequestTransport(model),
signal: params.signal,
});
}
@@ -468,13 +512,14 @@ async function describeImagesWithModelInternal(
const maxTokens = resolveImageToolMaxTokens(model.maxTokens, params.maxTokens);
const completeImage = async (onPayload?: ProviderStreamOptions["onPayload"]) => {
params.signal?.throwIfAborted();
const payloadHandler = composeImageDescriptionPayloadHandlers(onPayload, options.onPayload);
const timeoutMs = configuredTimeoutMs;
const headers = buildImageRequestHeaders(model);
const streamOptions = {
apiKey,
maxTokens,
signal: controller.signal,
signal: requestSignal,
...(timeoutMs !== undefined ? { timeoutMs } : {}),
...(headers ? { headers } : {}),
...(payloadHandler ? { onPayload: payloadHandler } : {}),
@@ -484,6 +529,7 @@ async function describeImagesWithModelInternal(
: complete(model, context, streamOptions);
return await withImageDescriptionTimeout({
controller,
signal: params.signal,
timeoutMs,
createTimeoutError: (requestTimeoutMs) =>
buildImageDescriptionTimeoutError({
@@ -509,6 +555,7 @@ async function describeImagesWithModelInternal(
}
}
params.signal?.throwIfAborted();
const retryMessage = await completeImage(disableReasoningForImageRetryPayload);
const text = coerceImageAssistantText({
message: retryMessage,
@@ -535,6 +582,7 @@ function toImagesDescriptionRequest(params: ImageDescriptionRequest): ImagesDesc
prompt: params.prompt,
maxTokens: params.maxTokens,
timeoutMs: params.timeoutMs,
...(params.signal ? { signal: params.signal } : {}),
profile: params.profile,
preferredProfile: params.preferredProfile,
authStore: params.authStore,
src/media-understanding/openai-compatible-audio.ts
+1
View File
@@ -68,6 +68,7 @@ export async function transcribeOpenAiCompatibleAudio(
headers,
body: form,
timeoutMs: params.timeoutMs,
...(params.signal ? { signal: params.signal } : {}),
fetchFn,
pinDns: false,
allowPrivateNetwork,
src/media-understanding/shared.ts
+6
View File
@@ -653,6 +653,7 @@ type GuardedPostRequestParams<TBody> = GuardedProviderRequestParams &
headers: Headers;
body: TBody;
timeoutMs?: number;
signal?: AbortSignal;
fetchFn: typeof fetch;
};
@@ -663,6 +664,7 @@ export async function postTranscriptionRequest(params: GuardedPostRequestParams<
method: "POST",
headers: params.headers,
body: params.body,
...(params.signal ? { signal: params.signal } : {}),
},
timeoutMs: params.timeoutMs,
fetchFn: params.fetchFn,
@@ -682,6 +684,7 @@ async function postGuardedRequest(params: {
retry?: TransientProviderRetryConfig;
}) {
const operation = async () => {
params.init.signal?.throwIfAborted();
const result = await fetchWithTimeoutGuarded(
params.url,
params.init,
@@ -707,6 +710,7 @@ async function postGuardedRequest(params: {
provider: "provider-http",
stage: params.retryStage,
retry: params.retry,
signal: params.init.signal ?? undefined,
operation,
});
}
@@ -718,6 +722,7 @@ export async function postJsonRequest(params: GuardedPostRequestParams<unknown>)
method: "POST",
headers: params.headers,
body: JSON.stringify(params.body),
...(params.signal ? { signal: params.signal } : {}),
},
timeoutMs: params.timeoutMs,
fetchFn: params.fetchFn,
@@ -734,6 +739,7 @@ export async function postMultipartRequest(params: GuardedPostRequestParams<Body
method: "POST",
headers: params.headers,
body: params.body,
...(params.signal ? { signal: params.signal } : {}),
},
timeoutMs: params.timeoutMs,
fetchFn: params.fetchFn,
src/media-understanding/types.ts
+5
View File
@@ -118,6 +118,7 @@ export type AudioTranscriptionRequest = {
prompt?: string;
query?: Record<string, string | number | boolean>;
timeoutMs: number;
signal?: AbortSignal;
fetchFn?: typeof fetch;
};
@@ -139,6 +140,7 @@ export type VideoDescriptionRequest = {
model?: string;
prompt?: string;
timeoutMs: number;
signal?: AbortSignal;
fetchFn?: typeof fetch;
};
@@ -154,6 +156,7 @@ export type ImageDescriptionRequest = {
prompt?: string;
maxTokens?: number;
timeoutMs: number;
signal?: AbortSignal;
profile?: string;
preferredProfile?: string;
authStore?: AuthProfileStore;
@@ -179,6 +182,7 @@ export type ImagesDescriptionRequest = {
prompt?: string;
maxTokens?: number;
timeoutMs: number;
signal?: AbortSignal;
profile?: string;
preferredProfile?: string;
authStore?: AuthProfileStore;
@@ -223,6 +227,7 @@ export type StructuredExtractionRequest = {
jsonSchema?: unknown;
jsonMode?: boolean;
timeoutMs: number;
signal?: AbortSignal;
profile?: string;
preferredProfile?: string;
authStore?: AuthProfileStore;
src/provider-runtime/operation-retry.test.ts
+20
View File
@@ -113,6 +113,26 @@ describe("executeProviderOperationWithRetry", () => {
expect(operation).toHaveBeenCalledTimes(1);
});
it("does not start another attempt after caller cancellation", async () => {
const controller = new AbortController();
const operation = vi.fn(async () => {
controller.abort(new Error("caller cancelled provider read"));
throw Object.assign(new Error("socket hang up"), { code: "ECONNRESET" });
});
await expect(
executeProviderOperationWithRetry({
provider: "test",
stage: "read",
operation,
signal: controller.signal,
retry: { attempts: 2, baseDelayMs: 0, maxDelayMs: 0 },
}),
).rejects.toThrow("caller cancelled provider read");
expect(operation).toHaveBeenCalledOnce();
});
it("does not retry create operations by default", async () => {
const operation = vi.fn(async () => {
throw Object.assign(new Error("write EPIPE"), { code: "EPIPE" });
src/provider-runtime/operation-retry.ts
+11 -1
View File
@@ -249,16 +249,26 @@ export async function executeProviderOperationWithRetry<T>(params: {
stage: ProviderOperationRetryStage;
operation: () => Promise<T>;
retry?: TransientProviderRetryConfig;
signal?: AbortSignal;
}): Promise<T> {
const retryConfig = providerOperationRetryConfig(params.stage, params.retry);
const retryOptions = resolveTransientProviderRetryOptions(retryConfig);
const resolvedRetryOptions = resolveTransientProviderRetryOptions(retryConfig);
const retrySignal =
params.signal && resolvedRetryOptions?.signal
? AbortSignal.any([params.signal, resolvedRetryOptions.signal])
: (params.signal ?? resolvedRetryOptions?.signal);
const retryOptions = resolvedRetryOptions
? { ...resolvedRetryOptions, ...(retrySignal ? { signal: retrySignal } : {}) }
: undefined;
const maxAttempts = resolveTransientProviderAttempts(retryOptions);
let lastError: unknown;
for (let attemptNumber = 1; attemptNumber <= maxAttempts; attemptNumber += 1) {
params.signal?.throwIfAborted();
try {
return await params.operation();
} catch (error) {
params.signal?.throwIfAborted();
lastError = error;
const message = formatErrorMessage(error);
if (