From 7152fe90e1c4ff5c5738f39dc5175a99fb3e67e8 Mon Sep 17 00:00:00 2001 From: userName Date: Mon, 25 May 2026 18:29:27 +0800 Subject: [PATCH] POK-53: persist inbound Lark file attachments --- src/channels/lark/inbound.ts | 383 +++++++++++++++++++++- tests/channels/lark/inbound/media.test.ts | 166 ++++++++++ 2 files changed, 548 insertions(+), 1 deletion(-) diff --git a/src/channels/lark/inbound.ts b/src/channels/lark/inbound.ts index ada8f15..a818ee0 100644 --- a/src/channels/lark/inbound.ts +++ b/src/channels/lark/inbound.ts @@ -8,6 +8,8 @@ import { Buffer } from "node:buffer"; import { randomUUID } from "node:crypto"; +import { mkdir, writeFile } from "node:fs/promises"; +import path from "node:path"; import * as Lark from "@larksuiteoapi/node-sdk"; import { type AgentUserImagePayload, @@ -39,6 +41,7 @@ import { type RuntimeStatusService, } from "@/src/runtime/status.js"; import { createSubsystemLogger } from "@/src/shared/logger.js"; +import { POKOCLAW_WORKSPACE_DIR } from "@/src/shared/paths.js"; import type { StorageDb } from "@/src/storage/db/client.js"; import { AgentsRepo } from "@/src/storage/repos/agents.repo.js"; import { ChannelInstancesRepo } from "@/src/storage/repos/channel-instances.repo.js"; @@ -62,6 +65,7 @@ const LARK_INBOUND_LOG_PREVIEW_MAX_LENGTH = 144; const LARK_CARD_ACTION_LOG_PREVIEW_MAX_LENGTH = 320; const LARK_INTERACTIVE_TEXT_NODE_LIMIT = 48; const LARK_INTERACTIVE_TEXT_CHAR_LIMIT = 4_000; +const LARK_INBOUND_ASSET_DIR = path.join(POKOCLAW_WORKSPACE_DIR, "lark-inbound"); const LARK_INTERACTIVE_TEXT_TRUNCATED_NOTICE = "[卡片内容过长，引用文本已截断]"; const RUN_CARD_OBJECT_KIND = "run_card"; const STEER_PENDING_REACTION_EMOJI = "Typing"; @@ -143,6 +147,7 @@ export interface CreateLarkInboundRuntimeInput { }): void; }; steerReactionState?: LarkSteerReactionState; + assetStoreDir?: string; } export interface LarkInboundRuntimeStatus { @@ -198,6 +203,7 @@ export interface NormalizedLarkTextMessage { chatType: string | null; text: string; imageKeys?: string[]; + fileAssets?: LarkInboundFileDescriptor[]; createdAt?: Date; } @@ -208,6 +214,21 @@ interface LarkInboundImageAsset { mimeType: string; } +type LarkInboundFileAssetKind = "audio" | "file"; + +interface LarkInboundFileDescriptor { + kind: LarkInboundFileAssetKind; + fileKey: string; + fileName?: string; +} + +interface LarkInboundSavedFileAsset extends LarkInboundFileDescriptor { + messageId: string; + localPath: string; + mimeType: string; + byteLength: number; +} + interface LarkQuotedMessage { messageType: string; text: string; @@ -302,6 +323,7 @@ export function normalizeLarkTextMessage( return { skipReason: "text message content is empty" }; } const imageKeys = extractLarkImageKeys(messageType, contentRaw); + const fileAssets = extractLarkFileDescriptors(messageType, contentRaw); const parentMessageId = readString(message.parent_id); const threadId = readString(message.thread_id); @@ -326,6 +348,12 @@ export function normalizeLarkTextMessage( chatType, imageKeyCount: imageKeys.length, imageKeys, + fileAssetCount: fileAssets.length, + fileAssets: fileAssets.map((asset) => ({ + kind: asset.kind, + fileKey: asset.fileKey, + fileName: asset.fileName ?? null, + })), contentPreview: truncateLogText(text, LARK_INBOUND_LOG_PREVIEW_MAX_LENGTH), }); @@ -340,6 +368,7 @@ export function normalizeLarkTextMessage( chatType, text, ...(imageKeys.length === 0 ? {} : { imageKeys }), + ...(fileAssets.length === 0 ? {} : { fileAssets }), ...(createdAt == null ? {} : { createdAt }), }; } @@ -361,6 +390,7 @@ export function createLarkMessageReceiveHandler(input: { }) => Promise; taskThreads?: CreateLarkInboundRuntimeInput["taskThreads"]; steerReactionState?: LarkSteerReactionState; + assetStoreDir?: string; }): (data: unknown) => Promise { return async (data: unknown) => { const normalized = normalizeLarkTextMessage(data); @@ -571,10 +601,22 @@ export function createLarkMessageReceiveHandler(input: { contentPreview: truncateLogText(hydrated.text, LARK_INBOUND_LOG_PREVIEW_MAX_LENGTH), }); - const content = + const baseContent = route.kind === "ordinary_thread" ? hydrated.text : await buildInboundMessageContent(hydrated, input); + const savedFileAssets = + input.clients == null + ? [] + : await fetchAndStoreLarkInboundFileAssets({ + installationId: input.installationId, + messageId: hydrated.messageId, + descriptors: hydrated.fileAssets ?? [], + assetStoreDir: input.assetStoreDir ?? LARK_INBOUND_ASSET_DIR, + clients: input.clients, + ...(hydrated.createdAt == null ? {} : { createdAt: hydrated.createdAt }), + }); + const content = appendSavedFileAssetPaths(baseContent, savedFileAssets); const imageAssets = input.clients == null ? [] @@ -590,6 +632,24 @@ export function createLarkMessageReceiveHandler(input: { }); const runtimeImages = buildLarkInboundRuntimeImages(imageAssets); + if ((hydrated.fileAssets?.length ?? 0) > 0 || savedFileAssets.length > 0) { + logger.info("prepared lark inbound file assets", { + installationId: input.installationId, + chatId: hydrated.chatId, + messageId: hydrated.messageId, + requestedFileAssetCount: hydrated.fileAssets?.length ?? 0, + savedFileAssetCount: savedFileAssets.length, + savedFileAssets: savedFileAssets.map((asset) => ({ + kind: asset.kind, + fileKey: asset.fileKey, + fileName: asset.fileName ?? null, + localPath: asset.localPath, + mimeType: asset.mimeType, + byteLength: asset.byteLength, + })), + }); + } + if ((hydrated.imageKeys?.length ?? 0) > 0 || runtimeImages.length > 0) { logger.info("prepared lark inbound image payloads", { installationId: input.installationId, @@ -1185,6 +1245,7 @@ export function createLarkInboundRuntime(input: CreateLarkInboundRuntimeInput): ...(input.steerReactionState == null ? {} : { steerReactionState: input.steerReactionState }), + ...(input.assetStoreDir == null ? {} : { assetStoreDir: input.assetStoreDir }), }); const onCardAction = createLarkCardActionHandler({ installationId: installation.installationId, @@ -2336,6 +2397,98 @@ function extractLarkImageKeys(messageType: string, content: string): string[] { } } +function extractLarkFileDescriptors( + messageType: string, + content: string, +): LarkInboundFileDescriptor[] { + if (content.length === 0) { + return []; + } + + try { + const parsed = JSON.parse(content); + if (!isRecord(parsed)) { + return []; + } + + switch (messageType) { + case "audio": { + const fileKey = readString(parsed.file_key); + return fileKey == null ? [] : [{ kind: "audio", fileKey }]; + } + case "file": { + const fileKey = readString(parsed.file_key); + if (fileKey == null) { + return []; + } + const fileName = readString(parsed.file_name) ?? readString(parsed.name); + return [{ kind: "file", fileKey, ...(fileName == null ? {} : { fileName }) }]; + } + case "media": { + const fileKey = readString(parsed.file_key); + if (fileKey == null) { + return []; + } + const fileName = readString(parsed.file_name) ?? readString(parsed.name); + return [{ kind: "file", fileKey, ...(fileName == null ? {} : { fileName }) }]; + } + case "post": + return extractLarkPostFileDescriptors(parsed); + default: + return []; + } + } catch { + return []; + } +} + +function extractLarkPostFileDescriptors( + parsed: Record, +): LarkInboundFileDescriptor[] { + const body = unwrapLarkPostContent(parsed); + if (body == null) { + return []; + } + + const descriptors: LarkInboundFileDescriptor[] = []; + const content = Array.isArray(body.content) ? body.content : []; + for (const paragraph of content) { + if (!Array.isArray(paragraph)) { + continue; + } + for (const element of paragraph) { + if (!isRecord(element) || readString(element.tag) !== "media") { + continue; + } + const fileKey = readString(element.file_key); + if (fileKey == null) { + continue; + } + const fileName = readString(element.file_name) ?? readString(element.name); + descriptors.push({ + kind: "file", + fileKey, + ...(fileName == null ? {} : { fileName }), + }); + } + } + return dedupeLarkFileDescriptors(descriptors); +} + +function dedupeLarkFileDescriptors( + descriptors: LarkInboundFileDescriptor[], +): LarkInboundFileDescriptor[] { + const seen = new Set(); + return descriptors.filter((descriptor) => { + const key = `${descriptor.kind}:${descriptor.fileKey}`; + if (seen.has(key)) { + return false; + } + seen.add(key); + return true; + }); +} + function parseLarkMessageContent(messageType: string, content: string): string { if (content.length === 0) { return ""; @@ -3000,6 +3153,158 @@ async function fetchLarkInboundImageAssets(input: { return assets; } +async function fetchAndStoreLarkInboundFileAssets(input: { + installationId: string; + messageId: string; + createdAt?: Date; + descriptors: LarkInboundFileDescriptor[]; + assetStoreDir: string; + clients: { + getOrCreate(installationId: string): LarkSdkClient; + }; +}): Promise { + if (input.descriptors.length === 0) { + logger.debug("no lark inbound file assets to fetch", { + installationId: input.installationId, + messageId: input.messageId, + }); + return []; + } + + logger.debug("fetching lark inbound file assets", { + installationId: input.installationId, + messageId: input.messageId, + fileAssetCount: input.descriptors.length, + fileAssets: input.descriptors.map((asset) => ({ + kind: asset.kind, + fileKey: asset.fileKey, + fileName: asset.fileName ?? null, + })), + }); + + const client = input.clients.getOrCreate(input.installationId); + const targetDir = buildLarkInboundAssetMessageDir({ + assetStoreDir: input.assetStoreDir, + messageId: input.messageId, + ...(input.createdAt == null ? {} : { createdAt: input.createdAt }), + }); + const savedAssets: LarkInboundSavedFileAsset[] = []; + for (const descriptor of input.descriptors) { + try { + logger.debug("requesting lark inbound file resource", { + installationId: input.installationId, + messageId: input.messageId, + kind: descriptor.kind, + fileKey: descriptor.fileKey, + fileName: descriptor.fileName ?? null, + }); + const response = await client.sdk.im.messageResource.get({ + path: { + message_id: input.messageId, + file_key: descriptor.fileKey, + }, + params: { + type: descriptor.kind === "audio" ? "audio" : "file", + }, + }); + const buffer = await readLarkResourceBuffer(response.getReadableStream()); + if (buffer.length === 0) { + throw new Error("empty lark inbound file resource"); + } + const mimeType = normalizeLarkFileMimeType(response.headers?.["content-type"]); + await mkdir(targetDir, { recursive: true }); + const localPath = path.join( + targetDir, + buildLarkInboundAssetFileName({ + descriptor, + mimeType, + index: savedAssets.length, + }), + ); + await writeFile(localPath, buffer); + savedAssets.push({ + ...descriptor, + messageId: input.messageId, + localPath, + mimeType, + byteLength: buffer.length, + }); + logger.info("saved lark inbound file resource", { + installationId: input.installationId, + messageId: input.messageId, + kind: descriptor.kind, + fileKey: descriptor.fileKey, + fileName: descriptor.fileName ?? null, + localPath, + mimeType, + byteLength: buffer.length, + }); + } catch (error) { + logger.warn("failed to fetch lark inbound file resource", { + installationId: input.installationId, + messageId: input.messageId, + kind: descriptor.kind, + fileKey: descriptor.fileKey, + fileName: descriptor.fileName ?? null, + error: error instanceof Error ? error.message : String(error), + }); + } + } + + logger.debug("finished fetching lark inbound file assets", { + installationId: input.installationId, + messageId: input.messageId, + requestedFileAssetCount: input.descriptors.length, + savedFileAssetCount: savedAssets.length, + }); + + return savedAssets; +} + +function appendSavedFileAssetPaths( + content: string, + savedAssets: LarkInboundSavedFileAsset[], +): string { + if (savedAssets.length === 0) { + return content; + } + + const lines = savedAssets.map((asset) => { + const label = asset.fileName == null ? asset.fileKey : asset.fileName; + return `- ${asset.kind}: ${label} -> ${asset.localPath}`; + }); + return [content, "", "Saved local files:", ...lines].join("\n"); +} + +function buildLarkInboundAssetMessageDir(input: { + assetStoreDir: string; + createdAt?: Date; + messageId: string; +}): string { + const dateSegment = + input.createdAt == null || Number.isNaN(input.createdAt.getTime()) + ? "undated" + : input.createdAt.toISOString().slice(0, 10); + return path.join(input.assetStoreDir, dateSegment, sanitizePathSegment(input.messageId)); +} + +function buildLarkInboundAssetFileName(input: { + descriptor: LarkInboundFileDescriptor; + mimeType: string; + index: number; +}): string { + const fileName = sanitizeFileName(input.descriptor.fileName); + const key = sanitizePathSegment(input.descriptor.fileKey).slice(0, 32); + const fallbackExt = extensionFromMimeType(input.mimeType) ?? ".bin"; + if (fileName != null) { + const parsed = path.parse(fileName); + const ext = parsed.ext.length > 0 ? parsed.ext : fallbackExt; + const name = parsed.name.length > 0 ? parsed.name : input.descriptor.kind; + return `${name}-${key}${ext}`; + } + return `${input.descriptor.kind}-${key || input.index}${fallbackExt}`; +} + async function readLarkResourceBuffer(stream: NodeJS.ReadableStream): Promise { const chunks: Buffer[] = []; for await (const chunk of stream) { @@ -3032,6 +3337,82 @@ function normalizeLarkImageMimeType(value: unknown): string { return "image/png"; } +function normalizeLarkFileMimeType(value: unknown): string { + const normalized = normalizeContentTypeHeader(value); + return normalized ?? "application/octet-stream"; +} + +function normalizeContentTypeHeader(value: unknown): string | null { + if (typeof value === "string") { + const normalized = value.split(";")[0]?.trim().toLowerCase(); + return normalized == null || normalized.length === 0 ? null : normalized; + } + + if (Array.isArray(value)) { + for (const item of value) { + if (typeof item !== "string") { + continue; + } + const normalized = item.split(";")[0]?.trim().toLowerCase(); + if (normalized != null && normalized.length > 0) { + return normalized; + } + } + } + + return null; +} + +function extensionFromMimeType(mimeType: string): string | null { + switch (mimeType.toLowerCase()) { + case "audio/mpeg": + case "audio/mp3": + return ".mp3"; + case "audio/mp4": + case "audio/x-m4a": + return ".m4a"; + case "audio/ogg": + return ".ogg"; + case "audio/wav": + case "audio/x-wav": + return ".wav"; + case "audio/webm": + return ".webm"; + case "video/mp4": + return ".mp4"; + case "application/pdf": + return ".pdf"; + case "text/plain": + return ".txt"; + default: + return null; + } +} + +function sanitizeFileName(value: string | undefined): string | null { + if (value == null) { + return null; + } + const baseName = path.basename(value.replaceAll("\\", "/")).trim(); + if (baseName.length === 0 || baseName === "." || baseName === "..") { + return null; + } + const sanitized = [...baseName] + .map((char) => (isUnsafeFileNameChar(char) ? "_" : char)) + .join("") + .replace(/\s+/g, " "); + return sanitized.length === 0 ? null : sanitized; +} + +function isUnsafeFileNameChar(char: string): boolean { + return char.charCodeAt(0) < 32 || '<>:"/\\|?*'.includes(char); +} + +function sanitizePathSegment(value: string): string { + const sanitized = value.trim().replace(/[^a-zA-Z0-9._-]/g, "_"); + return sanitized.length === 0 ? "unknown" : sanitized; +} + async function fetchQuotedLarkMessage(input: { client: LarkSdkClient; messageId: string; diff --git a/tests/channels/lark/inbound/media.test.ts b/tests/channels/lark/inbound/media.test.ts index 6e7aee2..0076ea0 100644 --- a/tests/channels/lark/inbound/media.test.ts +++ b/tests/channels/lark/inbound/media.test.ts @@ -1,3 +1,6 @@ +import { mkdtemp, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; import { Readable } from "node:stream"; import { describe, expect, test, vi } from "vitest"; import type { LarkSdkClient } from "@/src/channels/lark/client.js"; @@ -11,6 +14,169 @@ import { ChannelSurfacesRepo } from "@/src/storage/repos/channel-surfaces.repo.j import { makeMessageEvent, makeRawMessage, seedFixture, withHandle } from "./fixtures.js"; describe("lark inbound message media handling", () => { + test("downloads inbound audio messages to the local workspace and forwards the saved path", async () => { + const assetStoreDir = await mkdtemp(path.join(tmpdir(), "pokoclaw-lark-audio-")); + try { + await withHandle(async (handle) => { + seedFixture(handle); + const surfacesRepo = new ChannelSurfacesRepo(handle.storage.db); + surfacesRepo.upsert({ + id: "surface_1", + channelType: "lark", + channelInstallationId: "default", + conversationId: "conv_main", + branchId: "branch_main", + surfaceKey: buildLarkChatSurfaceKey("oc_chat_1"), + surfaceObjectJson: JSON.stringify({ chat_id: "oc_chat_1" }), + }); + + const submitMessage = vi.fn(async () => ({ status: "started" as const })); + const messageResourceGet = vi.fn(async () => ({ + headers: { "content-type": "audio/mpeg" }, + getReadableStream: () => Readable.from(Buffer.from("fake-audio")), + })); + const clients = { + getOrCreate: vi.fn(() => ({ + sdk: { + im: { + messageResource: { + get: messageResourceGet, + }, + }, + }, + })), + } as unknown as { getOrCreate(installationId: string): LarkSdkClient }; + + const handler = createLarkMessageReceiveHandler({ + installationId: "default", + storage: handle.storage.db, + ingress: { submitMessage, submitApprovalDecision: vi.fn(() => false) }, + control: new RuntimeControlService(new SessionRunAbortRegistry()), + clients, + assetStoreDir, + }); + + await handler(makeMessageEvent("audio", { file_key: "audio_v3_123" })); + + const localPath = path.join( + assetStoreDir, + "2026-03-27", + "om_msg_1", + "audio-audio_v3_123.mp3", + ); + + expect(messageResourceGet).toHaveBeenCalledExactlyOnceWith({ + path: { + message_id: "om_msg_1", + file_key: "audio_v3_123", + }, + params: { + type: "audio", + }, + }); + await expect(readFile(localPath, "utf8")).resolves.toBe("fake-audio"); + expect(submitMessage).toHaveBeenCalledExactlyOnceWith({ + sessionId: "sess_chat_1", + scenario: "chat", + content: [ + "[\u8bed\u97f3 audio_v3_123]", + "", + "Saved local files:", + `- audio: audio_v3_123 -> ${localPath}`, + ].join("\n"), + channelMessageId: "om_msg_1", + createdAt: new Date("2026-03-27T00:00:00.000Z"), + }); + }); + } finally { + await rm(assetStoreDir, { recursive: true, force: true }); + } + }); + + test("downloads uploaded recording files to the local workspace", async () => { + const assetStoreDir = await mkdtemp(path.join(tmpdir(), "pokoclaw-lark-file-")); + try { + await withHandle(async (handle) => { + seedFixture(handle); + const surfacesRepo = new ChannelSurfacesRepo(handle.storage.db); + surfacesRepo.upsert({ + id: "surface_1", + channelType: "lark", + channelInstallationId: "default", + conversationId: "conv_main", + branchId: "branch_main", + surfaceKey: buildLarkChatSurfaceKey("oc_chat_1"), + surfaceObjectJson: JSON.stringify({ chat_id: "oc_chat_1" }), + }); + + const submitMessage = vi.fn(async () => ({ status: "started" as const })); + const messageResourceGet = vi.fn(async () => ({ + headers: { "content-type": "audio/x-m4a" }, + getReadableStream: () => Readable.from(Buffer.from("fake-m4a")), + })); + const clients = { + getOrCreate: vi.fn(() => ({ + sdk: { + im: { + messageResource: { + get: messageResourceGet, + }, + }, + }, + })), + } as unknown as { getOrCreate(installationId: string): LarkSdkClient }; + + const handler = createLarkMessageReceiveHandler({ + installationId: "default", + storage: handle.storage.db, + ingress: { submitMessage, submitApprovalDecision: vi.fn(() => false) }, + control: new RuntimeControlService(new SessionRunAbortRegistry()), + clients, + assetStoreDir, + }); + + await handler( + makeMessageEvent("file", { + file_key: "file_v3_recording_1", + file_name: "meeting recording.m4a", + }), + ); + + const localPath = path.join( + assetStoreDir, + "2026-03-27", + "om_msg_1", + "meeting recording-file_v3_recording_1.m4a", + ); + + expect(messageResourceGet).toHaveBeenCalledExactlyOnceWith({ + path: { + message_id: "om_msg_1", + file_key: "file_v3_recording_1", + }, + params: { + type: "file", + }, + }); + await expect(readFile(localPath, "utf8")).resolves.toBe("fake-m4a"); + expect(submitMessage).toHaveBeenCalledExactlyOnceWith({ + sessionId: "sess_chat_1", + scenario: "chat", + content: [ + "[\u6587\u4ef6 file_v3_recording_1]", + "", + "Saved local files:", + `- file: meeting recording.m4a -> ${localPath}`, + ].join("\n"), + channelMessageId: "om_msg_1", + createdAt: new Date("2026-03-27T00:00:00.000Z"), + }); + }); + } finally { + await rm(assetStoreDir, { recursive: true, force: true }); + } + }); + test("downloads inbound image messages and forwards them in userPayload", async () => { await withHandle(async (handle) => { seedFixture(handle);