feat: support inbound files across all channels

This commit is contained in:
xmanrui 2026-08-23 22:39:26 +08:00
parent 29bcb67a64
commit 0a26d7df92
45 changed files with 2651 additions and 343 deletions

View file

@ -0,0 +1,64 @@
async function cancelResponseBody(response) {
try {
await response?.body?.cancel?.();
} catch {
// Preserve the original download failure.
}
}
function hostedByMessagingPlatform(target, allowedHosts) {
return !Array.isArray(allowedHosts) || allowedHosts.some((rule) => (
typeof rule === 'string'
&& (target.hostname === rule
|| (rule.startsWith('.')
&& (target.hostname === rule.slice(1) || target.hostname.endsWith(rule))))
));
}
/**
* Open a channel-hosted ordinary file as a stream.
*
* This deliberately has no plugin-defined size, type, count, or download-time
* limit. The caller owns cancellation through its AbortSignal and the channel
* remains the authority for its own file limits.
*/
export async function fetchFileStream(url, {
fetchImpl = fetch,
headers,
signal,
allowedHosts,
} = {}) {
const target = new URL(url);
if (target.protocol !== 'https:') throw new Error('File download URL must use HTTPS');
if (!hostedByMessagingPlatform(target, allowedHosts)) {
throw new Error('File download URL is not hosted by the messaging platform');
}
const response = await fetchImpl(target, {
method: 'GET',
headers,
signal,
redirect: 'manual',
});
if (Number.isInteger(response?.status) && response.status >= 300 && response.status < 400) {
await cancelResponseBody(response);
const error = new Error(`File download redirect was blocked (HTTP ${response.status})`);
error.code = 'file-redirect-blocked';
throw error;
}
if (!response?.ok) {
await cancelResponseBody(response);
const error = new Error(`File download failed with HTTP ${response?.status ?? 'unknown'}`);
error.code = 'file-http-error';
error.status = response?.status;
throw error;
}
if (response.body?.[Symbol.asyncIterator]) return { stream: response.body };
if (typeof response.arrayBuffer === 'function') {
const data = Buffer.from(await response.arrayBuffer());
return {
stream: (async function* fileBody() { yield data; }()),
};
}
throw new Error('File download returned no readable body');
}

View file

@ -3,6 +3,10 @@ import { randomUUID } from 'node:crypto';
import { isAbsolute } from 'node:path';
import { adoptRegisteredWorkspaceSession } from './harness-session-binding.mjs';
import {
appendInboundFilesToPrompt,
InboundFileError,
} from './inbound-file.mjs';
import { outboundArtifactRegistry } from './semantic/artifact.mjs';
// Every channel plugin runs in the same Host process. Sharing ownership by
@ -466,6 +470,7 @@ export class HarnessClient {
#commandExecutor;
#controlExecutor;
#sessionMaintenanceExecutor;
#fileIngressExecutor;
#managedProcess = null;
#interactionRegistry;
#interactionOwnerships;
@ -486,6 +491,7 @@ export class HarnessClient {
commandExecutor,
controlExecutor,
sessionMaintenanceExecutor,
fileIngressExecutor,
}) {
if (typeof createWebSocket !== 'function') {
throw new TypeError('createWebSocket must be a function');
@ -509,6 +515,9 @@ export class HarnessClient {
&& typeof sessionMaintenanceExecutor !== 'function') {
throw new TypeError('sessionMaintenanceExecutor must be a function');
}
if (fileIngressExecutor !== undefined && typeof fileIngressExecutor !== 'function') {
throw new TypeError('fileIngressExecutor must be a function');
}
this.#baseUrl = new URL(baseUrl);
this.#workspace = workspace;
// Keep an omitted preset absent so session.create resolves the Host's current default.
@ -523,6 +532,7 @@ export class HarnessClient {
this.#commandExecutor = commandExecutor;
this.#controlExecutor = controlExecutor;
this.#sessionMaintenanceExecutor = sessionMaintenanceExecutor;
this.#fileIngressExecutor = fileIngressExecutor;
this.#interactionRegistry = interactionRegistry(this.#baseUrl.origin);
this.#interactionOwnerships = this.#interactionRegistry.ownerships;
this.#interactionClaims = this.#interactionRegistry.claims;
@ -1034,6 +1044,7 @@ export class HarnessClient {
? options.onInteractionResolved
: undefined;
const control = normalizeControl(options.control);
const inboundFiles = Array.isArray(options.files) ? options.files.filter(Boolean) : [];
await this.ensureRunning({ signal });
const before = await this.rpc(
'session.history',
@ -1075,6 +1086,9 @@ export class HarnessClient {
let interactionTask = null;
let artifactsDelivered = false;
let deliveredArtifactCount = 0;
let stagedInboundFiles = null;
let promptAccepted = false;
let turnFinished = false;
const deliverArtifacts = async () => {
if (!onArtifact || artifactsDelivered || tracker.turn === null) {
@ -1103,6 +1117,30 @@ export class HarnessClient {
const closeArtifactConsumer = outboundArtifactRegistry.openConsumer(sessionId, promptRpcId);
try {
if (inboundFiles.length > 0) {
if (!this.#fileIngressExecutor) {
throw new InboundFileError(
'inbound-file-ingress-unavailable',
'Harness file ingress is unavailable in this Host process.',
);
}
const sessionList = await this.rpc(
'session.list',
{},
30_000,
{ signal },
);
const sessionWorkspace = sessionList?.items?.find(
(item) => item?.sessionId === sessionId,
)?.cwd;
stagedInboundFiles = await this.#fileIngressExecutor({
sessionId,
workspace: sessionWorkspace,
files: inboundFiles,
signal,
});
prompt = appendInboundFilesToPrompt(prompt, stagedInboundFiles);
}
if (interactionSignal) {
let markOpen;
const opened = new Promise((resolve) => { markOpen = resolve; });
@ -1134,6 +1172,7 @@ export class HarnessClient {
content,
clientTimeZone: Intl.DateTimeFormat().resolvedOptions().timeZone,
}, 30_000, { rpcId: promptRpcId, signal });
promptAccepted = true;
try {
const deadline = Date.now() + timeoutMs;
@ -1159,6 +1198,7 @@ export class HarnessClient {
}
}
if (!tracker.finished) continue;
turnFinished = true;
// An accepted /stop revokes attachment delivery even when Harness
// preserved a useful partial text answer for the existing UX.
const artifactCount = ownership?.stopRequested
@ -1185,6 +1225,11 @@ export class HarnessClient {
throw turnStoppedError();
}
} finally {
if (stagedInboundFiles && (!promptAccepted || turnFinished)) {
await stagedInboundFiles.cleanup().catch((error) => {
console.warn(`[${this.#logPrefix}] unable to clean inbound files:`, error.message);
});
}
closeArtifactConsumer();
if (ownership) {
this.#unregisterControlOwnership(ownership);

View file

@ -0,0 +1,206 @@
import { createWriteStream } from 'node:fs';
import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises';
import { isAbsolute, join, relative, resolve } from 'node:path';
import { pipeline } from 'node:stream/promises';
const FILES_DIRECTORY = join('.dsh-im', 'inbound');
export class InboundFileError extends Error {
constructor(code, message, userMessage = '文件接收失败,请重新发送后再试。', options = {}) {
super(message, options);
this.name = 'InboundFileError';
this.code = code;
this.userMessage = userMessage;
}
}
function fileSources(message) {
return Array.isArray(message?.files) ? message.files.filter(Boolean) : [];
}
function displayName(value, fallback) {
if (typeof value !== 'string') return fallback;
const cleaned = value
.replaceAll('\\', '/')
.split('/')
.at(-1)
?.replace(/[\u0000-\u001f\u007f]/g, '')
.trim();
return cleaned || fallback;
}
function storageName(value, index) {
const cleaned = displayName(value, 'file')
.replace(/[^\p{L}\p{N}._ -]/gu, '_')
.replace(/^\.+/, '')
.slice(0, 160) || 'file';
return `${String(index + 1).padStart(2, '0')}-${cleaned}`;
}
function loadedFile(value) {
if (Buffer.isBuffer(value) || value instanceof Uint8Array) {
return { data: Buffer.from(value) };
}
const raw = value?.data ?? value?.buffer;
if (Buffer.isBuffer(raw) || raw instanceof Uint8Array) {
return {
data: Buffer.from(raw),
name: value?.name ?? value?.filename,
mediaType: value?.mediaType ?? value?.mimetype,
};
}
const stream = value?.stream ?? value;
if (stream && typeof stream[Symbol.asyncIterator] === 'function') {
return {
stream,
name: value?.name ?? value?.filename,
mediaType: value?.mediaType ?? value?.mimetype,
};
}
return null;
}
export function hasInboundFiles(message) {
return fileSources(message).length > 0;
}
/** Start provider downloads immediately while preserving the lazy file-source contract. */
export function prefetchInboundFiles(message, { signal } = {}) {
const sources = fileSources(message);
if (sources.length === 0) return message;
return {
...message,
files: sources.map((source) => {
if (source?.data !== undefined || typeof source?.load !== 'function') return source;
let download;
try {
download = Promise.resolve(source.load({ signal }));
} catch (error) {
download = Promise.reject(error);
}
download.catch(() => undefined);
return {
...source,
async load({ signal: loadSignal } = {}) {
loadSignal?.throwIfAborted();
const result = await download;
loadSignal?.throwIfAborted();
return result;
},
};
}),
};
}
export async function stageInboundFiles(message, {
workspace,
signal,
} = {}) {
const sources = fileSources(message);
if (sources.length === 0) return null;
if (typeof workspace !== 'string' || !isAbsolute(workspace)) {
throw new InboundFileError(
'inbound-file-workspace-unavailable',
'The Harness Session workspace is unavailable for inbound files.',
);
}
signal?.throwIfAborted();
const root = resolve(workspace, FILES_DIRECTORY);
await mkdir(root, { recursive: true, mode: 0o700 });
const directory = await mkdtemp(join(root, 'turn-'));
const files = [];
try {
for (const [index, source] of sources.entries()) {
signal?.throwIfAborted();
let value;
try {
value = source?.data === undefined
? await source?.load?.({ signal })
: source.data;
} catch (error) {
if (signal?.aborted) throw error;
throw new InboundFileError(
'inbound-file-download-failed',
`Unable to download inbound file ${index + 1}: ${error?.message ?? String(error)}`,
'文件下载失败,请重新发送后再试。',
{ cause: error },
);
}
const loaded = loadedFile(value);
if (!loaded) {
throw new InboundFileError(
'inbound-file-data-invalid',
`Inbound file ${index + 1} returned no readable data.`,
);
}
const name = displayName(loaded.name ?? source?.name, `file-${index + 1}`);
const path = join(directory, storageName(name, index));
if (loaded.data) {
await writeFile(path, loaded.data, { mode: 0o600, signal });
} else {
try {
await pipeline(
loaded.stream,
createWriteStream(path, { flags: 'wx', mode: 0o600 }),
{ signal },
);
} catch (error) {
if (signal?.aborted) throw error;
throw new InboundFileError(
'inbound-file-download-failed',
`Unable to stream inbound file ${index + 1}: ${error?.message ?? String(error)}`,
'文件下载失败,请重新发送后再试。',
{ cause: error },
);
}
}
const relativePath = relative(resolve(workspace), path);
if (!relativePath || relativePath.startsWith('..') || isAbsolute(relativePath)) {
throw new InboundFileError(
'inbound-file-path-invalid',
'The staged inbound file escaped the Harness Session workspace.',
);
}
files.push(Object.freeze({
name,
path: relativePath,
...(typeof (loaded.mediaType ?? source?.mediaType) === 'string'
&& (loaded.mediaType ?? source.mediaType).trim()
? { mediaType: (loaded.mediaType ?? source.mediaType).trim() }
: {}),
}));
}
return Object.freeze({
files: Object.freeze(files),
async cleanup() {
await rm(directory, { recursive: true, force: true });
},
});
} catch (error) {
await rm(directory, { recursive: true, force: true }).catch(() => undefined);
throw error;
}
}
export function appendInboundFilesToPrompt(prompt, staged) {
if (!staged?.files?.length) return prompt;
const manifest = [
'<dsh_im_files>',
JSON.stringify({
description: 'Files uploaded with this user message. Paths are relative to the current Harness workspace.',
files: staged.files,
}),
'</dsh_im_files>',
].join('\n');
if (Array.isArray(prompt)) return [...prompt, { type: 'text', text: manifest }];
const text = typeof prompt === 'string' ? prompt.trim() : '';
return text ? `${text}\n\n${manifest}` : manifest;
}
export function inboundFileUserMessage(error) {
return error instanceof InboundFileError ? error.userMessage : null;
}

View file

@ -23,6 +23,10 @@ import {
imagePromptUserMessage,
promptContentForMessage,
} from './image-prompt.mjs';
import {
hasInboundFiles,
inboundFileUserMessage,
} from './inbound-file.mjs';
import {
harnessAnswerForQuestion,
harnessQuestionText,
@ -50,6 +54,7 @@ function canClaimInteractionReply(message, pending, senderId) {
return pending.actor === senderId
&& (message.kind !== 'group' || message.addressed === true)
&& !hasInboundImages(message)
&& !hasInboundFiles(message)
&& Boolean(cleanText(message.content));
}
@ -171,7 +176,7 @@ export class TextHarnessBridge {
const key = `${kind}:${conversationId}`;
const pending = this.#pendingInteractions.get(key);
const text = cleanText(normalized.content);
const commandRunner = isControlCommand(text)
const commandRunner = hasInboundFiles(normalized) ? null : isControlCommand(text)
? runControlCommand
: (isModelCommand(text)
? runModelCommand
@ -194,7 +199,7 @@ export class TextHarnessBridge {
key,
actor: senderId,
messageId,
text: hasInboundImages(normalized) ? '' : normalized.content,
text: hasInboundImages(normalized) || hasInboundFiles(normalized) ? '' : normalized.content,
addressed: normalized.kind !== 'group' || normalized.addressed === true,
hasPendingQuestion: Boolean(pending),
questionCompletion: pending?.submitting || pending?.claimedReplyMessageId
@ -300,6 +305,7 @@ export class TextHarnessBridge {
{
signal: this.#signal,
hasImages: hasInboundImages(message),
hasFiles: hasInboundFiles(message),
pendingInteraction: this.#pendingInteractions.has(key)
|| this.#approvals.hasPending(key),
control: { owner: this, key },
@ -423,16 +429,17 @@ export class TextHarnessBridge {
return;
}
const hasImages = hasInboundImages(message);
if (!text && !hasImages) {
await this.#bot.sendText(target, '目前支持文字和图片消息。');
const hasFiles = hasInboundFiles(message);
if (!text && !hasImages && !hasFiles) {
await this.#bot.sendText(target, '目前支持文字、图片和文件消息。');
return;
}
const command = text.toLowerCase();
if (!hasImages && command === '/help') {
if (!hasImages && !hasFiles && command === '/help') {
await this.#bot.sendText(target, [
`${this.#descriptor.label}机器人已连接 DeepSeek Harness。`,
'',
'直接发送文字或图片即可继续当前会话。',
'直接发送文字、图片或文件即可继续当前会话。',
'/new 开启一个全新会话',
'/compact 压缩当前会话的较早上下文',
'/workspace 工作区绝对路径 切换工作区',
@ -453,12 +460,12 @@ export class TextHarnessBridge {
].join('\n'));
return;
}
if (!hasImages && command === '/status') {
if (!hasImages && !hasFiles && command === '/status') {
await this.#harness.ensureRunning({ signal: this.#signal });
await this.#bot.sendText(target, `${this.#descriptor.label}机器人与 DeepSeek Harness 连接正常。`);
return;
}
const workspaceCommand = !hasImages
const workspaceCommand = !hasImages && !hasFiles
? await runWorkspaceCommand(text, this.#harness, conversationKey)
: null;
if (workspaceCommand) {
@ -467,12 +474,12 @@ export class TextHarnessBridge {
}
return;
}
if (!hasImages && command === '/new') {
if (!hasImages && !hasFiles && command === '/new') {
await this.#state.clearSession(conversationKey);
await this.#bot.sendText(target, '已开启新会话。请发送你的问题。');
return;
}
const compactCommand = !hasImages
const compactCommand = !hasImages && !hasFiles
? await runCompactCommand(
text,
this.#harness,
@ -527,6 +534,7 @@ export class TextHarnessBridge {
requiresMention: message.kind === 'group',
}),
onInteractionResolved: (resolution) => this.#handleInteractionResolved(resolution),
files: message.files,
},
});
const visibleAnswer = !cleanText(answer) && artifacts.length > 0
@ -600,6 +608,18 @@ export class TextHarnessBridge {
}
return;
}
const fileErrorMessage = inboundFileUserMessage(error);
if (fileErrorMessage) {
try {
await this.#bot.sendText(target, fileErrorMessage);
} catch (sendError) {
this.#logger.error?.(
`[dsh-im:${this.#descriptor.key}] failed to send the file error reply:`,
sendError,
);
}
return;
}
this.#logger.error?.(`[dsh-im:${this.#descriptor.key}] failed to process a message:`, error);
try {
await this.#bot.sendText(target, '消息处理失败,请稍后重试。');
@ -642,7 +662,7 @@ export class TextHarnessBridge {
const target = message.replyTarget;
const text = cleanText(message.content);
if (!text || hasInboundImages(message)) {
if (!text || hasInboundImages(message) || hasInboundFiles(message)) {
try {
await this.#bot.sendText(target, '请用文字回答当前问题。');
} catch (error) {