diff --git a/extensions/tlon/src/monitor/index.test.ts b/extensions/tlon/src/monitor/index.test.ts index ef1114a69e8a..f596c053c2e3 100644 --- a/extensions/tlon/src/monitor/index.test.ts +++ b/extensions/tlon/src/monitor/index.test.ts @@ -1,14 +1,17 @@ -// Tlon monitor tests cover authentication retry scheduling and shutdown lifecycle. +// Tlon monitor tests cover authentication, inbound context, and shutdown lifecycle. import { createServer, type Server } from "node:http"; import type { AddressInfo } from "node:net"; +import { buildChannelInboundEventContext } from "openclaw/plugin-sdk/channel-inbound"; +import { saveRemoteMedia } from "openclaw/plugin-sdk/media-runtime"; import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime"; -import { afterEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; const { authenticateMock, sleepWithAbortMock, sseClientMock, ingressMock, + inboundRuntimeMock, settingsManagerMock, realUrbitFixture, } = vi.hoisted(() => ({ @@ -27,6 +30,18 @@ const { start: vi.fn(), stop: vi.fn().mockResolvedValue(undefined), }, + inboundRuntimeMock: { + buildContext: vi.fn(), + dispatch: vi.fn().mockResolvedValue(undefined), + resolveAgentRoute: vi.fn(() => ({ + accountId: "default", + agentId: "main", + dmScope: "main", + sessionKey: "agent:main:main", + })), + resolveEffectiveMessagesConfig: vi.fn(() => ({ responsePrefix: undefined })), + shouldComputeCommandAuthorized: vi.fn(() => false), + }, settingsManagerMock: { load: vi.fn().mockResolvedValue({}), onChange: vi.fn().mockReturnValue(() => {}), @@ -52,6 +67,11 @@ vi.mock("openclaw/plugin-sdk/runtime-env", async (importOriginal) => { }; }); +vi.mock("openclaw/plugin-sdk/media-runtime", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, saveRemoteMedia: vi.fn() }; +}); + vi.mock("../runtime.js", () => ({ getTlonRuntime: () => ({ config: { @@ -62,6 +82,7 @@ vi.mock("../runtime.js", () => ({ ship: "~zod", url: realUrbitFixture.url, network: { dangerouslyAllowPrivateNetwork: true }, + ownerShip: "~nec", }, }, }), @@ -69,6 +90,21 @@ vi.mock("../runtime.js", () => ({ logging: { getChildLogger: () => ({}), }, + channel: { + commands: { + shouldComputeCommandAuthorized: inboundRuntimeMock.shouldComputeCommandAuthorized, + }, + inbound: { + buildContext: inboundRuntimeMock.buildContext, + dispatch: inboundRuntimeMock.dispatch, + }, + reply: { + resolveEffectiveMessagesConfig: inboundRuntimeMock.resolveEffectiveMessagesConfig, + }, + routing: { + resolveAgentRoute: inboundRuntimeMock.resolveAgentRoute, + }, + }, }), })); @@ -100,6 +136,13 @@ vi.mock("./ingress.js", () => ({ })); import { monitorTlonProvider } from "./index.js"; +import { extractMessageText } from "./utils.js"; + +const saveRemoteMediaMock = vi.mocked(saveRemoteMedia); + +beforeEach(() => { + inboundRuntimeMock.buildContext.mockImplementation(buildChannelInboundEventContext); +}); afterEach(async () => { vi.clearAllMocks(); @@ -143,6 +186,100 @@ describe("monitorTlonProvider authentication retry", () => { }); }); +describe("monitorTlonProvider inbound media truth", () => { + it.each([ + { + name: "a failed download beside successful images", + imageCount: 3, + failedIndexes: [1], + expectedAttachments: 2, + expectedNotice: "[tlon attachment unavailable]", + }, + { + name: "images beyond the eight-image cap", + imageCount: 10, + failedIndexes: [], + expectedAttachments: 8, + expectedNotice: "[tlon 2 attachments unavailable]", + }, + ])( + "reports $name to the model without changing command text", + async ({ imageCount, failedIndexes, expectedAttachments, expectedNotice }) => { + const controller = new AbortController(); + const runtime = { error: vi.fn(), exit: vi.fn(), log: vi.fn() } satisfies RuntimeEnv; + authenticateMock.mockResolvedValueOnce("urbauth-~zod=proof"); + ingressMock.receive.mockResolvedValueOnce({ kind: "ignored" }); + saveRemoteMediaMock.mockImplementation(async ({ url }) => { + const index = Number(new URL(url).pathname.slice(1, -4)); + if (failedIndexes.includes(index)) { + throw new Error("download failed"); + } + return { + id: `photo-${index}.png`, + path: `/tmp/openclaw/media/inbound/photo-${index}.png`, + size: 10, + contentType: "image/png", + }; + }); + const content = [ + { inline: ["/status"] }, + ...Array.from({ length: imageCount }, (_, index) => ({ + block: { image: { src: `https://example.com/${index}.png` } }, + })), + ]; + const originalText = extractMessageText(content); + + const monitor = monitorTlonProvider({ abortSignal: controller.signal, runtime }); + try { + await vi.waitFor(() => expect(sseClientMock.connect).toHaveBeenCalledOnce()); + const chatSubscription = sseClientMock.subscribe.mock.calls + .map(([subscription]) => subscription) + .find((subscription) => subscription.app === "chat"); + if (!chatSubscription) { + throw new Error("expected chat subscription"); + } + await chatSubscription.event({ + whom: "~nec", + id: `dm-media-${imageCount}`, + response: { + add: { + essay: { + author: "~nec", + content, + sent: 1_700_000_000_000, + }, + }, + }, + }); + + expect(inboundRuntimeMock.dispatch).toHaveBeenCalledOnce(); + const dispatchCall = inboundRuntimeMock.dispatch.mock.calls[0]; + if (!dispatchCall) { + throw new Error("expected inbound dispatch call"); + } + const [{ ctxPayload, replyOptions }] = dispatchCall; + expect(ctxPayload.BodyForAgent).toBe(`${originalText}\n\n${expectedNotice}`); + expect(ctxPayload.RawBody).toBe(originalText); + expect(ctxPayload.CommandBody).toBe(originalText); + expect(ctxPayload.BodyForCommands).toBe(originalText); + expect(ctxPayload.Attachments).toHaveLength(expectedAttachments); + expect(replyOptions.media).toHaveLength(expectedAttachments); + expect(replyOptions.media).toEqual( + Array.from({ length: Math.min(imageCount, 8) }, (_, index) => index) + .filter((index) => !failedIndexes.includes(index)) + .map((index) => ({ + path: `/tmp/openclaw/media/inbound/photo-${index}.png`, + contentType: "image/png", + })), + ); + } finally { + controller.abort(); + await monitor; + } + }, + ); +}); + describe("monitorTlonProvider shutdown", () => { it("does not authenticate when the shutdown signal is already aborted", async () => { const controller = new AbortController(); diff --git a/extensions/tlon/src/monitor/index.ts b/extensions/tlon/src/monitor/index.ts index ad41c9fc8022..2f33163c9222 100644 --- a/extensions/tlon/src/monitor/index.ts +++ b/extensions/tlon/src/monitor/index.ts @@ -1,5 +1,8 @@ import { resolveHumanDelayConfig } from "openclaw/plugin-sdk/agent-runtime"; -import { createChannelInboundEnvelopeBuilder } from "openclaw/plugin-sdk/channel-inbound"; +import { + createChannelInboundEnvelopeBuilder, + formatInboundMediaUnavailableText, +} from "openclaw/plugin-sdk/channel-inbound"; import { bindIngressLifecycleToReplyOptions, waitUntilAbort, @@ -330,9 +333,11 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise = []; + let unavailableMediaCount = 0; if (messageContent) { try { - attachments = await downloadMessageImages(messageContent); + ({ attachments, unavailableCount: unavailableMediaCount } = + await downloadMessageImages(messageContent)); if (attachments.length > 0) { runtime.log?.(`[tlon] Downloaded ${attachments.length} image(s) from message`); } @@ -505,6 +510,13 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise 0 + ? formatInboundMediaUnavailableText({ + body: commandBody, + notice: `[tlon ${unavailableMediaCount > 1 ? `${unavailableMediaCount} attachments` : "attachment"} unavailable]`, + }) + : commandBody; const tlonConversationId = isGroup ? (groupChannel ?? channelNest ?? senderShip) : senderShip; const ctxPayload = core.channel.inbound.buildContext({ channel: "tlon", @@ -535,7 +547,7 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise { contentType: "image/png", })); - const images = await downloadMessageImages(content); + const result = await downloadMessageImages(content); - expect(images).toHaveLength(8); + expect(result).toMatchObject({ unavailableCount: 2 }); + expect(result.attachments).toHaveLength(8); expect(saveRemoteMediaMock.mock.calls.map(([options]) => options.url)).toEqual( Array.from({ length: 8 }, (_, index) => `https://example.com/${index}.png`), ); @@ -87,12 +88,15 @@ describe("tlon monitor media", () => { ssrfPolicy: undefined, requestInit: { method: "GET" }, }); - expect(result).toEqual([ - { path: "/tmp/openclaw/media/inbound/photo---uuid.png", contentType: "image/png" }, - ]); + expect(result).toEqual({ + attachments: [ + { path: "/tmp/openclaw/media/inbound/photo---uuid.png", contentType: "image/png" }, + ], + unavailableCount: 0, + }); }); - it("returns null when the fetch exceeds the image cap", async () => { + it("reports an unavailable image when the fetch exceeds the image cap", async () => { saveRemoteMediaMock.mockRejectedValue( new Error( `Failed to fetch media from https://example.com/photo.png: payload exceeds maxBytes ${MAX_IMAGE_BYTES}`, @@ -103,7 +107,7 @@ describe("tlon monitor media", () => { { block: { image: { src: "https://example.com/photo.png" } } }, ]); - expect(result).toEqual([]); + expect(result).toEqual({ attachments: [], unavailableCount: 1 }); expect(readRemoteMediaBufferMock).not.toHaveBeenCalled(); }); }); diff --git a/extensions/tlon/src/monitor/media.ts b/extensions/tlon/src/monitor/media.ts index eb6c66083018..7a4e5cc95bb9 100644 --- a/extensions/tlon/src/monitor/media.ts +++ b/extensions/tlon/src/monitor/media.ts @@ -14,18 +14,10 @@ import { TLON_MEDIA_FETCH_TIMEOUTS } from "../media-fetch-timeouts.js"; const MAX_IMAGES_PER_MESSAGE = 8; -interface ExtractedImage { - url: string; - alt?: string; -} - -interface DownloadedMedia { - localPath: string; - contentType: string; - originalUrl: string; -} - +type ExtractedImages = { images: Array<{ url: string }>; unavailableCount: number }; +type DownloadedMedia = { localPath: string; contentType: string }; type TlonInboundMedia = { path: string; contentType: string }; +type TlonInboundMediaDownload = { attachments: TlonInboundMedia[]; unavailableCount: number }; /** Keeps Tlon's shipped path-duplicating prompt bytes paired with ordered facts. */ export function buildTlonInboundMediaPrompt( @@ -47,28 +39,27 @@ export function buildTlonInboundMediaPrompt( /** * Extract image blocks from Tlon message content. - * Returns array of image URLs found in the message. + * Returns up to the download cap plus the number omitted by that cap. */ -function extractImageBlocks(content: unknown): ExtractedImage[] { +function extractImageBlocks(content: unknown): ExtractedImages { if (!content || !Array.isArray(content)) { - return []; + return { images: [], unavailableCount: 0 }; } - const images: ExtractedImage[] = []; + const images: Array<{ url: string }> = []; + let unavailableCount = 0; for (const verse of content) { if (verse?.block?.image?.src) { - images.push({ - url: verse.block.image.src, - alt: verse.block.image.alt, - }); if (images.length >= MAX_IMAGES_PER_MESSAGE) { - break; + unavailableCount++; + continue; } + images.push({ url: verse.block.image.src }); } } - return images; + return { images, unavailableCount }; } /** @@ -97,7 +88,6 @@ async function downloadMedia(url: string, mediaDir?: string): Promise> { - const images = extractImageBlocks(content); - if (images.length === 0) { - return []; - } - - const attachments: Array<{ path: string; contentType: string }> = []; +): Promise { + const { images, unavailableCount: overCapCount } = extractImageBlocks(content); + const attachments: TlonInboundMedia[] = []; + let unavailableCount = overCapCount; for (const image of images) { const downloaded = await downloadMedia(image.url, mediaDir); @@ -166,8 +152,10 @@ export async function downloadMessageImages( path: downloaded.localPath, contentType: downloaded.contentType, }); + } else { + unavailableCount++; } } - return attachments; + return { attachments, unavailableCount }; }