Skip to content

Commit dcfc1f1

Browse files
steipetesteipete
committed
test: split ACP attachment resolution from dispatch flow
1 parent b43d73b commit dcfc1f1

3 files changed

Lines changed: 117 additions & 83 deletions

File tree

‎src/auto-reply/reply/dispatch-acp-attachments.ts‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
import type { AcpTurnAttachment } from "../../acp/control-plane/manager.types.js";
2+
import type { OpenClawConfig } from "../../config/config.js";
3+
import { logVerbose } from "../../globals.js";
4+
import type { FinalizedMsgContext } from "../templating.js";
5+
6+
let dispatchAcpMediaRuntimePromise: Promise<
7+
typeof import("./dispatch-acp-media.runtime.js")
8+
> | null = null;
9+
10+
export function loadDispatchAcpMediaRuntime() {
11+
dispatchAcpMediaRuntimePromise ??= import("./dispatch-acp-media.runtime.js");
12+
return dispatchAcpMediaRuntimePromise;
13+
}
14+
15+
export type DispatchAcpAttachmentRuntime = Pick<
16+
Awaited<ReturnType<typeof loadDispatchAcpMediaRuntime>>,
17+
| "MediaAttachmentCache"
18+
| "isMediaUnderstandingSkipError"
19+
| "normalizeAttachments"
20+
| "resolveMediaAttachmentLocalRoots"
21+
>;
22+
23+
const ACP_ATTACHMENT_MAX_BYTES = 10 * 1024 * 1024;
24+
const ACP_ATTACHMENT_TIMEOUT_MS = 1_000;
25+
26+
export async function resolveAcpAttachments(params: {
27+
ctx: FinalizedMsgContext;
28+
cfg: OpenClawConfig;
29+
runtime?: DispatchAcpAttachmentRuntime;
30+
}): Promise<AcpTurnAttachment[]> {
31+
const runtime = params.runtime ?? (await loadDispatchAcpMediaRuntime());
32+
const mediaAttachments = runtime
33+
.normalizeAttachments(params.ctx)
34+
.map((attachment) =>
35+
attachment.path?.trim() ? { ...attachment, url: undefined } : attachment,
36+
);
37+
const cache = new runtime.MediaAttachmentCache(mediaAttachments, {
38+
localPathRoots: runtime.resolveMediaAttachmentLocalRoots({
39+
cfg: params.cfg,
40+
ctx: params.ctx,
41+
}),
42+
});
43+
const results: AcpTurnAttachment[] = [];
44+
for (const attachment of mediaAttachments) {
45+
const mediaType = attachment.mime ?? "application/octet-stream";
46+
if (!mediaType.startsWith("image/")) {
47+
continue;
48+
}
49+
if (!attachment.path?.trim()) {
50+
continue;
51+
}
52+
try {
53+
const { buffer } = await cache.getBuffer({
54+
attachmentIndex: attachment.index,
55+
maxBytes: ACP_ATTACHMENT_MAX_BYTES,
56+
timeoutMs: ACP_ATTACHMENT_TIMEOUT_MS,
57+
});
58+
results.push({
59+
mediaType,
60+
data: buffer.toString("base64"),
61+
});
62+
} catch (error) {
63+
if (runtime.isMediaUnderstandingSkipError(error)) {
64+
logVerbose(`dispatch-acp: skipping attachment #${attachment.index + 1} (${error.reason})`);
65+
} else {
66+
const errorName = error instanceof Error ? error.name : typeof error;
67+
logVerbose(
68+
`dispatch-acp: failed to read attachment #${attachment.index + 1} (${errorName})`,
69+
);
70+
}
71+
}
72+
}
73+
return results;
74+
}

‎src/auto-reply/reply/dispatch-acp.test.ts‎

Lines changed: 39 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,9 @@ import { AcpRuntimeError } from "../../acp/runtime/errors.js";
66
import type { AcpSessionStoreEntry } from "../../acp/runtime/session-meta.js";
77
import type { OpenClawConfig } from "../../config/config.js";
88
import type { SessionBindingRecord } from "../../infra/outbound/session-binding-service.js";
9+
import type { MediaUnderstandingSkipError } from "../../media-understanding/errors.js";
910
import { withFetchPreconnect } from "../../test-utils/fetch-mock.js";
11+
import { resolveAcpAttachments } from "./dispatch-acp-attachments.js";
1012
import type { ReplyDispatcher } from "./reply-dispatcher.js";
1113
import { buildTestCtx } from "./test-ctx.js";
1214
import { createAcpSessionMeta, createAcpTestConfig } from "./test-fixtures/acp-runtime.js";
@@ -277,6 +279,11 @@ describe("tryDispatchAcpReply", () => {
277279
({ tryDispatchAcpReply } = await import("./dispatch-acp.js"));
278280
managerMocks.resolveSession.mockReset();
279281
managerMocks.runTurn.mockReset();
282+
managerMocks.runTurn.mockImplementation(
283+
async ({ onEvent }: { onEvent?: (event: unknown) => Promise<void> }) => {
284+
await onEvent?.({ type: "done" });
285+
},
286+
);
280287
managerMocks.getObservabilitySnapshot.mockReset();
281288
managerMocks.getObservabilitySnapshot.mockReturnValue({
282289
turns: { queueDepth: 0 },
@@ -435,39 +442,54 @@ describe("tryDispatchAcpReply", () => {
435442
});
436443

437444
it("forwards normalized image attachments into ACP turns", async () => {
438-
setReadyAcpResolution();
439445
const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "dispatch-acp-"));
440446
const imagePath = path.join(tempDir, "inbound.png");
441447
try {
442448
await fs.writeFile(imagePath, "image-bytes");
443-
managerMocks.runTurn.mockResolvedValue(undefined);
444-
445-
await runDispatch({
446-
bodyForAgent: " ",
449+
const attachments = await resolveAcpAttachments({
447450
cfg: createAcpTestConfig({
448451
channels: {
449452
imessage: {
450453
attachmentRoots: [tempDir],
451454
},
452455
},
453456
}),
454-
ctxOverrides: {
457+
ctx: buildTestCtx({
458+
Provider: "imessage",
459+
Surface: "imessage",
455460
MediaPath: imagePath,
456461
MediaType: "image/png",
457-
},
458-
});
459-
460-
expect(managerMocks.runTurn).toHaveBeenCalledWith(
461-
expect.objectContaining({
462-
text: "",
463-
attachments: [
462+
}),
463+
runtime: {
464+
MediaAttachmentCache: class {
465+
async getBuffer() {
466+
return {
467+
buffer: Buffer.from("image-bytes"),
468+
mime: "image/png",
469+
fileName: "inbound.png",
470+
size: "image-bytes".length,
471+
};
472+
}
473+
} as unknown as typeof import("./dispatch-acp-media.runtime.js").MediaAttachmentCache,
474+
isMediaUnderstandingSkipError: (_error: unknown): _error is MediaUnderstandingSkipError =>
475+
false,
476+
normalizeAttachments: (ctx) => [
464477
{
465-
mediaType: "image/png",
466-
data: Buffer.from("image-bytes").toString("base64"),
478+
path: ctx.MediaPath,
479+
mime: ctx.MediaType,
480+
index: 0,
467481
},
468482
],
469-
}),
470-
);
483+
resolveMediaAttachmentLocalRoots: () => [tempDir],
484+
},
485+
});
486+
487+
expect(attachments).toEqual([
488+
{
489+
mediaType: "image/png",
490+
data: Buffer.from("image-bytes").toString("base64"),
491+
},
492+
]);
471493
} finally {
472494
await fs.rm(tempDir, { recursive: true, force: true });
473495
}

‎src/auto-reply/reply/dispatch-acp.ts‎

Lines changed: 4 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,3 @@
1-
import type { AcpTurnAttachment } from "../../acp/control-plane/manager.types.js";
21
import { resolveAcpAgentPolicyError, resolveAcpDispatchPolicyError } from "../../acp/policy.js";
32
import { formatAcpRuntimeErrorText } from "../../acp/runtime/error-text.js";
43
import { toAcpRuntimeError } from "../../acp/runtime/errors.js";
@@ -18,15 +17,13 @@ import { resolveStatusTtsSnapshot } from "../../tts/status-config.js";
1817
import { resolveConfiguredTtsMode } from "../../tts/tts-config.js";
1918
import type { FinalizedMsgContext } from "../templating.js";
2019
import { createAcpReplyProjector } from "./acp-projector.js";
20+
import { loadDispatchAcpMediaRuntime, resolveAcpAttachments } from "./dispatch-acp-attachments.js";
2121
import {
2222
createAcpDispatchDeliveryCoordinator,
2323
type AcpDispatchDeliveryCoordinator,
2424
} from "./dispatch-acp-delivery.js";
2525
import type { ReplyDispatcher, ReplyDispatchKind } from "./reply-dispatcher.js";
2626

27-
let dispatchAcpMediaRuntimePromise: Promise<
28-
typeof import("./dispatch-acp-media.runtime.js")
29-
> | null = null;
3027
let dispatchAcpManagerRuntimePromise: Promise<
3128
typeof import("./dispatch-acp-manager.runtime.js")
3229
> | null = null;
@@ -36,11 +33,6 @@ let dispatchAcpSessionRuntimePromise: Promise<
3633
let dispatchAcpTtsRuntimePromise: Promise<typeof import("./dispatch-acp-tts.runtime.js")> | null =
3734
null;
3835

39-
function loadDispatchAcpMediaRuntime() {
40-
dispatchAcpMediaRuntimePromise ??= import("./dispatch-acp-media.runtime.js");
41-
return dispatchAcpMediaRuntimePromise;
42-
}
43-
4436
function loadDispatchAcpManagerRuntime() {
4537
dispatchAcpManagerRuntimePromise ??= import("./dispatch-acp-manager.runtime.js");
4638
return dispatchAcpManagerRuntimePromise;
@@ -99,62 +91,6 @@ function hasInboundMediaForAcp(ctx: FinalizedMsgContext): boolean {
9991
);
10092
}
10193

102-
const ACP_ATTACHMENT_MAX_BYTES = 10 * 1024 * 1024;
103-
const ACP_ATTACHMENT_TIMEOUT_MS = 1_000;
104-
105-
async function resolveAcpAttachments(
106-
ctx: FinalizedMsgContext,
107-
cfg: OpenClawConfig,
108-
): Promise<AcpTurnAttachment[]> {
109-
if (!hasInboundMediaForAcp(ctx)) {
110-
return [];
111-
}
112-
const {
113-
MediaAttachmentCache,
114-
isMediaUnderstandingSkipError,
115-
normalizeAttachments,
116-
resolveMediaAttachmentLocalRoots,
117-
} = await loadDispatchAcpMediaRuntime();
118-
const mediaAttachments = normalizeAttachments(ctx).map((attachment) =>
119-
attachment.path?.trim() ? { ...attachment, url: undefined } : attachment,
120-
);
121-
const cache = new MediaAttachmentCache(mediaAttachments, {
122-
localPathRoots: resolveMediaAttachmentLocalRoots({ cfg, ctx }),
123-
});
124-
const results: AcpTurnAttachment[] = [];
125-
for (const attachment of mediaAttachments) {
126-
const mediaType = attachment.mime ?? "application/octet-stream";
127-
if (!mediaType.startsWith("image/")) {
128-
continue;
129-
}
130-
if (!attachment.path?.trim()) {
131-
continue;
132-
}
133-
try {
134-
const { buffer } = await cache.getBuffer({
135-
attachmentIndex: attachment.index,
136-
maxBytes: ACP_ATTACHMENT_MAX_BYTES,
137-
timeoutMs: ACP_ATTACHMENT_TIMEOUT_MS,
138-
});
139-
results.push({
140-
mediaType,
141-
data: buffer.toString("base64"),
142-
});
143-
} catch (error) {
144-
if (isMediaUnderstandingSkipError(error)) {
145-
logVerbose(`dispatch-acp: skipping attachment #${attachment.index + 1} (${error.reason})`);
146-
} else {
147-
const errorName = error instanceof Error ? error.name : typeof error;
148-
logVerbose(
149-
`dispatch-acp: failed to read attachment #${attachment.index + 1} (${errorName})`,
150-
);
151-
}
152-
// Skip unreadable files. Text content should still be delivered.
153-
}
154-
}
155-
return results;
156-
}
157-
15894
function resolveAcpRequestId(ctx: FinalizedMsgContext): string {
15995
const id = ctx.MessageSidFull ?? ctx.MessageSid ?? ctx.MessageSidFirst ?? ctx.MessageSidLast;
16096
if (typeof id === "string" && id.trim()) {
@@ -492,7 +428,9 @@ export async function tryDispatchAcpReply(params: {
492428
}
493429

494430
const promptText = resolveAcpPromptText(params.ctx);
495-
const attachments = await resolveAcpAttachments(params.ctx, params.cfg);
431+
const attachments = hasInboundMediaForAcp(params.ctx)
432+
? await resolveAcpAttachments({ ctx: params.ctx, cfg: params.cfg })
433+
: [];
496434
if (!promptText && attachments.length === 0) {
497435
const counts = params.dispatcher.getQueuedCounts();
498436
delivery.applyRoutedCounts(counts);

0 commit comments

Comments
 (0)