diff --git a/openclaw.plugin.json b/openclaw.plugin.json index e3b8c04..86de7fd 100644 --- a/openclaw.plugin.json +++ b/openclaw.plugin.json @@ -169,6 +169,18 @@ }, "additionalProperties": false }, + "files": { + "type": "object", + "properties": { + "inboundDir": { + "type": "string" + }, + "outboundDir": { + "type": "string" + } + }, + "additionalProperties": false + }, "experimentalChannels": { "type": "boolean" }, @@ -406,6 +418,18 @@ }, "additionalProperties": false }, + "files": { + "type": "object", + "properties": { + "inboundDir": { + "type": "string" + }, + "outboundDir": { + "type": "string" + } + }, + "additionalProperties": false + }, "experimentalChannels": { "type": "boolean" }, diff --git a/src/actions/message-actions.ts b/src/actions/message-actions.ts index 118b51f..90fd753 100644 --- a/src/actions/message-actions.ts +++ b/src/actions/message-actions.ts @@ -2,6 +2,7 @@ import { readStringArrayParam } from "openclaw/plugin-sdk/channel-actions"; import type { ChannelMessageActionName } from "openclaw/plugin-sdk/channel-contract"; import type { OpenClawConfig } from "openclaw/plugin-sdk/channel-core"; import { normalizePollInput, resolvePollMaxSelections } from "openclaw/plugin-sdk/poll-runtime"; +import { cleanupStagedOutboundFiles } from "../channel/media/outbound-files.js"; import { buildComposedMessages } from "../channel/media/simplex-media.js"; import { renderSimplexPollText } from "../channel/messaging/simplex-outbound.js"; import { resolveSimplexChatItemId } from "../simplex/runtime/api.js"; @@ -61,6 +62,9 @@ async function sendActionComposedMessages(params: { ttl: params.ttl ?? params.account.config.messageTtlSeconds, }), }); + // Files staged into a shared outbound dir are no longer needed once the + // send command has been issued (simplex-chat has consumed the path). + await cleanupStagedOutboundFiles(params.composedMessages); return { messageId: resolveSimplexChatItemId(chatItems[0]) }; } diff --git a/src/channel/events/simplex-inbound-files.test.ts b/src/channel/events/simplex-inbound-files.test.ts index a843eea..4d97e5a 100644 --- a/src/channel/events/simplex-inbound-files.test.ts +++ b/src/channel/events/simplex-inbound-files.test.ts @@ -4,8 +4,10 @@ import { setSimplexRuntime } from "../runtime.js"; import { dispatchInbound, finalizePendingFile, + markFileAccepted, type PendingInboundFile, queuePendingFile, + shouldRetryFileAccept, } from "./simplex-inbound-files.js"; const simplexClientMock = vi.hoisted(() => ({ @@ -135,6 +137,29 @@ describe("simplex inbound live replies", () => { expect(cancelFile).not.toHaveBeenCalled(); }); + it("tracks accept retries for queued pending files", async () => { + vi.useFakeTimers(); + installRuntime(async () => undefined); + const current = pending(); + current.fileId = 42; + current.client = fileClient(vi.fn(async () => undefined)); + + // Unknown file: nothing queued, nothing to retry. + expect(shouldRetryFileAccept(current.account.accountId, 42)).toBe(false); + + // Queued but not yet accepted (initial /freceive failed): retry wanted. + queuePendingFile({ pending: current, accountId: current.account.accountId, fileId: 42 }); + expect(shouldRetryFileAccept(current.account.accountId, 42)).toBe(true); + + // Accepted: no further retries. + markFileAccepted(current.account.accountId, 42); + expect(shouldRetryFileAccept(current.account.accountId, 42)).toBe(false); + + // Finalized: entry removed entirely. + await finalizePendingFile({ accountId: current.account.accountId, fileId: 42 }); + expect(shouldRetryFileAccept(current.account.accountId, 42)).toBe(false); + }); + it("replaces an older pending file timeout for the same key", async () => { vi.useFakeTimers(); installRuntime(async () => undefined); diff --git a/src/channel/events/simplex-inbound-files.ts b/src/channel/events/simplex-inbound-files.ts index 339fcee..13a0653 100644 --- a/src/channel/events/simplex-inbound-files.ts +++ b/src/channel/events/simplex-inbound-files.ts @@ -1,8 +1,12 @@ +import { readFile } from "node:fs/promises"; +import path from "node:path"; import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contract"; import type { OpenClawConfig } from "openclaw/plugin-sdk/channel-core"; import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; +import { SIMPLEX_CHANNEL_ID } from "../../constants.js"; import type { SimplexClient } from "../../simplex/runtime/client.js"; import type { ResolvedSimplexAccount } from "../../types/config.js"; +import { resolveSimplexMediaMaxBytes } from "../media/simplex-media.js"; import { createSimplexLiveReplyController, type SimplexLiveReplyPayload, @@ -11,6 +15,27 @@ import { getSimplexRuntime } from "../runtime.js"; const PENDING_FILE_TIMEOUT_MS = 90_000; +/** + * Where simplex-chat saves received files when started without --files-folder. + */ +const DEFAULT_INBOUND_DIR = "/tmp"; + +/** + * Directory where the simplex-chat runtime saves received files (its + * --files-folder). The WS API reports received files with a path relative to + * this directory (`fileSource.filePath` is typically just the file name), so + * the plugin needs it to locate the file on disk. Defaults to /tmp, where + * simplex-chat saves files when no --files-folder is configured. + */ +function resolveSimplexInboundDir(params: { + cfg: OpenClawConfig; + accountId?: string | null; +}): string { + const channel = params.cfg.channels?.[SIMPLEX_CHANNEL_ID]; + const account = params.accountId ? channel?.accounts?.[params.accountId] : undefined; + return account?.files?.inboundDir ?? channel?.files?.inboundDir ?? DEFAULT_INBOUND_DIR; +} + type SimplexReplyPayload = { text?: string; mediaUrl?: string; @@ -37,6 +62,7 @@ export type PendingInboundFile = { type QueuedPendingInboundFile = { pending: PendingInboundFile; timeout: ReturnType; + accepted: boolean; }; const pendingFiles = new Map(); @@ -45,6 +71,12 @@ function pendingKey(accountId: string, fileId: number): string { return `${accountId}:${fileId}`; } +export function isFileAutoAcceptEnabled(account: ResolvedSimplexAccount): boolean { + return ( + account.config.filePolicy?.autoAccept ?? account.config.connection?.autoAcceptFiles !== false + ); +} + export async function requestFileDownload(params: { fileId: number; account: ResolvedSimplexAccount; @@ -52,9 +84,7 @@ export async function requestFileDownload(params: { runtime: RuntimeEnv; }): Promise { const { fileId, account, client, runtime } = params; - const autoAccept = - account.config.filePolicy?.autoAccept ?? account.config.connection?.autoAcceptFiles !== false; - if (!autoAccept) { + if (!isFileAutoAcceptEnabled(account)) { return false; } try { @@ -66,6 +96,23 @@ export async function requestFileDownload(params: { return true; } +/** + * Whether a pending file still needs its download accepted. The initial + * /freceive issued on newChatItems can fail because the XFTP file + * description is not ready yet; the accept is retried on rcvFileDescrReady. + */ +export function shouldRetryFileAccept(accountId: string, fileId: number): boolean { + const queued = pendingFiles.get(pendingKey(accountId, fileId)); + return Boolean(queued && !queued.accepted); +} + +export function markFileAccepted(accountId: string, fileId: number): void { + const queued = pendingFiles.get(pendingKey(accountId, fileId)); + if (queued) { + queued.accepted = true; + } +} + export function queuePendingFile(params: { pending: PendingInboundFile; accountId: string; @@ -99,13 +146,14 @@ export function queuePendingFile(params: { }); }, PENDING_FILE_TIMEOUT_MS); timeout.unref?.(); - pendingFiles.set(key, { pending, timeout }); + pendingFiles.set(key, { pending, timeout, accepted: false }); } export async function finalizePendingFile(params: { accountId: string; fileId: number; filePath?: string; + fileName?: string; }): Promise { const queued = pendingFiles.get(pendingKey(params.accountId, params.fileId)); if (!queued) { @@ -114,10 +162,50 @@ export async function finalizePendingFile(params: { pendingFiles.delete(pendingKey(params.accountId, params.fileId)); clearTimeout(queued.timeout); const { pending } = queued; - const mediaPath = params.filePath?.trim() || undefined; + let rawPath = params.filePath?.trim() || undefined; + // The WS API reports received files relative to the runtime's + // --files-folder (fileSource.filePath is typically just the file name). + // Resolve against the configured inbound dir (default: /tmp, where + // simplex-chat saves files when no --files-folder is configured). + if (rawPath && !path.isAbsolute(rawPath)) { + rawPath = path.join( + resolveSimplexInboundDir({ + cfg: pending.cfg, + accountId: pending.account.accountId, + }), + rawPath + ); + } + let mediaPath: string | undefined; let mediaType: string | undefined; - if (mediaPath) { - mediaType = await getSimplexRuntime().media.detectMime({ filePath: mediaPath }); + if (rawPath) { + const core = getSimplexRuntime(); + // Stage the file into OpenClaw's shared media store (media/inbound/*), + // like other bundled channels. The raw path from the SimpleX runtime + // (e.g. /tmp/... or ~/.simplex/files/...) lives outside the store, + // so media tools and sandboxed workspaces would reject it. + try { + mediaType = await core.media.detectMime({ filePath: rawPath }); + const buffer = await readFile(rawPath); + const saved = await core.channel.media.saveMediaBuffer( + buffer, + mediaType, + "inbound", + resolveSimplexMediaMaxBytes({ + cfg: pending.cfg, + accountId: pending.account.accountId, + }), + params.fileName ?? path.basename(rawPath), + rawPath + ); + mediaPath = saved.path; + mediaType = saved.contentType ?? mediaType; + } catch (err) { + pending.runtime.error?.( + `[${params.accountId}] SimpleX inbound media staging failed, using raw path: ${String(err)}` + ); + mediaPath = rawPath; + } } await dispatchInbound({ pending, mediaPath, mediaType }); } @@ -143,6 +231,7 @@ export async function dispatchInbound(params: { const ctxPayload = { ...pending.ctxPayload, MediaPath: mediaPath, + MediaPaths: mediaPath ? [mediaPath] : undefined, MediaType: mediaType, MediaUrl: mediaPath, }; diff --git a/src/channel/events/simplex-monitor.ts b/src/channel/events/simplex-monitor.ts index fabc57e..b5a0df8 100644 --- a/src/channel/events/simplex-monitor.ts +++ b/src/channel/events/simplex-monitor.ts @@ -25,9 +25,12 @@ import { dispatchInbound, finalizePendingFile, hasPendingFile, + isFileAutoAcceptEnabled, + markFileAccepted, type PendingInboundFile, queuePendingFile, requestFileDownload, + shouldRetryFileAccept, } from "./simplex-inbound-files.js"; export type SimplexMonitorOpts = { @@ -163,8 +166,19 @@ async function handleSimplexEvent(params: { if (event.type === "rcvFileDescrReady") { const fileId = (event as { rcvFileTransfer?: { fileId?: unknown } })?.rcvFileTransfer?.fileId; - if (typeof fileId === "number" && Number.isInteger(fileId) && fileId > 0) { - await requestFileDownload({ fileId, account, client, runtime }); + if ( + typeof fileId === "number" && + Number.isInteger(fileId) && + fileId > 0 && + // Retry the accept for a queued pending file whose initial /freceive + // failed (issued on newChatItems, before the file description was + // ready). Skip files we never queued (e.g. oversize) or that were + // already accepted. + shouldRetryFileAccept(account.accountId, fileId) + ) { + if (await requestFileDownload({ fileId, account, client, runtime })) { + markFileAccepted(account.accountId, fileId); + } } return; } @@ -179,6 +193,7 @@ async function handleSimplexEvent(params: { accountId: account.accountId, fileId, filePath, + fileName: typeof file?.fileName === "string" ? file.fileName : undefined, }); } return; @@ -328,9 +343,17 @@ async function handleSimplexEvent(params: { ); continue; } - const accepted = await requestFileDownload({ fileId, account, client, runtime }); - if (accepted) { + if (isFileAutoAcceptEnabled(account)) { + // Queue BEFORE attempting the accept: at this point the file may + // still be an invitation whose XFTP file description is not ready, + // so /freceive can fail. The accept is retried on rcvFileDescrReady + // and the dispatch happens on rcvFileComplete with the downloaded + // file's absolute path. If the transfer never completes, the + // pending-file timeout dispatches without media. queuePendingFile({ pending, accountId: account.accountId, fileId }); + if (await requestFileDownload({ fileId, account, client, runtime })) { + markFileAccepted(account.accountId, fileId); + } continue; } } diff --git a/src/channel/media/outbound-files.ts b/src/channel/media/outbound-files.ts new file mode 100644 index 0000000..60b110b --- /dev/null +++ b/src/channel/media/outbound-files.ts @@ -0,0 +1,91 @@ +import { randomUUID } from "node:crypto"; +import { copyFile, unlink, writeFile } from "node:fs/promises"; +import path from "node:path"; +import type { OpenClawConfig } from "openclaw/plugin-sdk/channel-core"; +import { SIMPLEX_CHANNEL_ID } from "../../constants.js"; +import type { SimplexComposedMessage } from "../../types/simplex.js"; + +/** + * Shared outbound directory support. + * + * When OpenClaw and simplex-chat run in separate containers, they exchange + * files through a shared volume rather than a shared filesystem. + * For example: + * + * /simplex/inbound files received by the bot (readable by OpenClaw) + * /simplex/outbound files OpenClaw writes for the bot to send + * + * If `files.outboundDir` is configured for this channel, outbound media is + * written there before the send command is issued, so the path in + * `fileSource.filePath` is readable by simplex-chat. Files staged by this + * module are tracked and deleted after the send completes. + * + * If `files.outboundDir` is not configured, the legacy single-filesystem + * behavior applies and this module is inert. + */ + +const STAGED_FILES = new Set(); + +export function resolveSimplexOutboundDir(params: { + cfg: OpenClawConfig; + accountId?: string | null; +}): string | undefined { + const channel = params.cfg.channels?.[SIMPLEX_CHANNEL_ID]; + const account = params.accountId ? channel?.accounts?.[params.accountId] : undefined; + return account?.files?.outboundDir ?? channel?.files?.outboundDir; +} + +/** + * Whether a local path is already readable by simplex-chat without staging + * (it lives in the shared outbound dir). + */ +export function isSimplexReadablePath(filePath: string, outboundDir: string): boolean { + return filePath.startsWith(outboundDir.endsWith(path.sep) ? outboundDir : outboundDir + path.sep); +} + +function stagedFileName(fileName?: string): string { + const base = fileName ? path.basename(fileName) : ""; + return base ? `${randomUUID()}-${base}` : randomUUID(); +} + +/** Write a media buffer into the shared outbound dir; returns the staged path. */ +export async function stageOutboundBuffer(params: { + outboundDir: string; + buffer: Uint8Array; + fileName?: string; +}): Promise { + const filePath = path.join(params.outboundDir, stagedFileName(params.fileName)); + await writeFile(filePath, params.buffer); + STAGED_FILES.add(filePath); + return filePath; +} + +/** Copy a local file into the shared outbound dir; returns the staged path. */ +export async function stageOutboundLocalFile(params: { + outboundDir: string; + sourcePath: string; +}): Promise { + const filePath = path.join(params.outboundDir, stagedFileName(params.sourcePath)); + await copyFile(params.sourcePath, filePath); + STAGED_FILES.add(filePath); + return filePath; +} + +/** + * Delete any staged outbound files referenced by the given composed messages. + * Call after the send completes. Only files staged by this module are + * deleted; pre-existing files in the outbound dir are left alone. + */ +export async function cleanupStagedOutboundFiles( + composedMessages: SimplexComposedMessage[] +): Promise { + for (const message of composedMessages) { + const filePath = message.fileSource?.filePath; + if (filePath && STAGED_FILES.has(filePath)) { + STAGED_FILES.delete(filePath); + await unlink(filePath).catch(() => { + // Best-effort: the file may already be gone. + }); + } + } +} diff --git a/src/channel/media/simplex-media.ts b/src/channel/media/simplex-media.ts index 459af51..e9fc268 100644 --- a/src/channel/media/simplex-media.ts +++ b/src/channel/media/simplex-media.ts @@ -1,9 +1,16 @@ import path from "node:path"; import type { OpenClawConfig } from "openclaw/plugin-sdk/channel-core"; import { resolveChannelMediaMaxBytes } from "openclaw/plugin-sdk/media-runtime"; +import { resolveMediaBufferPath } from "openclaw/plugin-sdk/media-store"; import { SIMPLEX_CHANNEL_ID } from "../../constants.js"; import type { SimplexComposedMessage, SimplexMsgContent } from "../../types/simplex.js"; import { getSimplexRuntime } from "../runtime.js"; +import { + isSimplexReadablePath, + resolveSimplexOutboundDir, + stageOutboundBuffer, + stageOutboundLocalFile, +} from "./outbound-files.js"; const DEFAULT_MAX_BYTES = 5 * 1024 * 1024; @@ -25,6 +32,7 @@ export function resolveSimplexMediaMaxBytes(params: { async function resolveMediaPath(params: { mediaUrl: string; maxBytes: number; + outboundDir?: string; }): Promise<{ path: string; contentType?: string; fileName?: string }> { const core = getSimplexRuntime(); const mediaUrlLower = params.mediaUrl.toLowerCase(); @@ -34,6 +42,16 @@ async function resolveMediaPath(params: { maxBytes: params.maxBytes, filePathHint: params.mediaUrl, }); + // Shared outbound dir configured: write the buffer where simplex-chat + // can read it (cross-container deployments). + if (params.outboundDir) { + const staged = await stageOutboundBuffer({ + outboundDir: params.outboundDir, + buffer: fetched.buffer, + fileName: fetched.fileName, + }); + return { path: staged, contentType: fetched.contentType, fileName: fetched.fileName }; + } const saved = await core.channel.media.saveMediaBuffer( fetched.buffer, fetched.contentType, @@ -43,9 +61,29 @@ async function resolveMediaPath(params: { ); return { path: saved.path, contentType: saved.contentType, fileName: fetched.fileName }; } - const contentType = await core.media.detectMime({ filePath: params.mediaUrl }); - const fileName = path.basename(params.mediaUrl); - return { path: params.mediaUrl, contentType, fileName }; + // media:/// references (e.g. media://inbound/ for files + // staged into OpenClaw's media store): resolve to the physical path via + // the store's read-side helper, then treat as a local path below. + let localPath = params.mediaUrl; + if (mediaUrlLower.startsWith("media://")) { + const match = /^media:\/\/([^/]+)\/(.+)$/.exec(params.mediaUrl); + if (!match?.[1] || !match[2]) { + throw new Error(`Invalid media reference: ${params.mediaUrl}`); + } + localPath = await resolveMediaBufferPath(match[2], match[1]); + } + const contentType = await core.media.detectMime({ filePath: localPath }); + const fileName = path.basename(localPath); + // Local path that simplex-chat cannot read: copy it into the shared + // outbound dir first. + if (params.outboundDir && !isSimplexReadablePath(localPath, params.outboundDir)) { + const staged = await stageOutboundLocalFile({ + outboundDir: params.outboundDir, + sourcePath: localPath, + }); + return { path: staged, contentType, fileName }; + } + return { path: localPath, contentType, fileName }; } function buildMediaMsgContent(params: { @@ -124,13 +162,17 @@ export async function buildComposedMessages(params: { cfg: params.cfg, accountId: params.accountId, }); + const outboundDir = resolveSimplexOutboundDir({ + cfg: params.cfg, + accountId: params.accountId, + }); for (let i = 0; i < mediaList.length; i += 1) { const mediaUrl = mediaList[i]; if (!mediaUrl) { continue; } - const resolved = await resolveMediaPath({ mediaUrl, maxBytes }); + const resolved = await resolveMediaPath({ mediaUrl, maxBytes, outboundDir }); const caption = i === 0 ? text : ""; const msgContent = buildMediaMsgContent({ text: caption, diff --git a/src/channel/messaging/simplex-send.ts b/src/channel/messaging/simplex-send.ts index d72cbb0..f435ecc 100644 --- a/src/channel/messaging/simplex-send.ts +++ b/src/channel/messaging/simplex-send.ts @@ -4,6 +4,7 @@ import { parseSimplexNumericId, resolveSimplexChatItemId } from "../../simplex/r import { withSimplexClient } from "../../simplex/runtime/transport.js"; import type { ResolvedSimplexAccount } from "../../types/config.js"; import type { SimplexComposedMessage } from "../../types/simplex.js"; +import { cleanupStagedOutboundFiles } from "../media/outbound-files.js"; import { buildComposedMessages } from "../media/simplex-media.js"; export async function sendSimplexComposedMessages(params: { @@ -27,6 +28,9 @@ export async function sendSimplexComposedMessages(params: { ttl: params.ttl, liveMessage: params.liveMessage, }); + // Files staged into a shared outbound dir are no longer needed once the + // send command has been issued (simplex-chat has consumed the path). + await cleanupStagedOutboundFiles(params.composedMessages); const messageId = resolveSimplexChatItemId(chatItems[0]); return { messageId, diff --git a/src/config/config-schema.ts b/src/config/config-schema.ts index 66e1e59..e1c72e2 100644 --- a/src/config/config-schema.ts +++ b/src/config/config-schema.ts @@ -59,6 +59,21 @@ const SimplexFilePolicySchema = z }) .strict(); +// Shared-volume file exchange (cross-container deployments). +// inboundDir is the simplex-chat runtime's --files-folder (e.g. +// /simplex/inbound); the WS API reports received files relative to it, so +// the plugin needs it to locate received files on disk. Defaults to /tmp, +// where simplex-chat saves files when no --files-folder is configured. +// outboundDir is a directory writable by OpenClaw and readable by +// simplex-chat (e.g. /simplex/outbound); outbound media is staged there +// before sending. Unset = default single-filesystem behavior. +const SimplexFilesSchema = z + .object({ + inboundDir: z.string().optional(), + outboundDir: z.string().optional(), + }) + .strict(); + export const SimplexAccountConfigSchema = z .object({ name: z.string().optional(), @@ -77,6 +92,7 @@ export const SimplexAccountConfigSchema = z streaming: SimplexStreamingSchema.optional(), messageTtlSeconds: z.number().int().positive().optional(), filePolicy: SimplexFilePolicySchema.optional(), + files: SimplexFilesSchema.optional(), experimentalChannels: z.boolean().optional(), groupPolicy: GroupPolicySchema.optional(), groupAllowFrom: SimplexAllowFromListSchema,