feat: send workspace images to QQ chats

Worker results may report image attachments stored inside the workspace;
the runtime validates them (containment, png/jpg magic, size) and
delivers them with the pending event. Assistants gain a send_image
action so users can ask for an image later. QQ uploads via /files with
base64 file_data and sends msg_type 7 rich media, sharing the same
msg_id/msg_seq counter as text replies; non-image adapters flatten
images to text.
This commit is contained in:
zenord
2026-08-19 14:39:18 +08:00
parent 7aa610294c
commit 6b506b8c55
17 changed files with 659 additions and 62 deletions
+99 -29
View File
@@ -4,7 +4,8 @@ import type { PlatformAdapter } from "./adapter.js";
import { CommandRouter, type ParsedCommand } from "./command-router.js";
import { chatKeyFor } from "./durable-session-store.js";
import type { Proposal } from "./proposal-store.js";
import type { IncomingMessage, MessageTarget } from "./types.js";
import type { IncomingMessage, MessageTarget, OutgoingImage, OutgoingImageRef } from "./types.js";
import { readOutboundImage } from "./workspace-images.js";
export interface GatewayResult { ok: boolean; reply?: string; ignored?: boolean; error?: string }
@@ -19,6 +20,11 @@ interface ChatTarget {
lastEventAt?: number;
}
interface PendingEvent {
text: string;
images?: OutgoingImageRef[];
}
export interface GatewayOptions {
now?: () => number;
}
@@ -27,6 +33,28 @@ export interface GatewayOptions {
const GROUP_PASSIVE_WINDOW_MS = 4.5 * 60_000;
const C2C_PASSIVE_WINDOW_MS = 55 * 60_000;
// Reads image files at send time (the worker may have updated them); unreadable files degrade
// to a text note, and adapters without image support get a plain [图片] text line instead.
async function resolveOutboundImages(text: string, refs: OutgoingImageRef[] | undefined, imageCapable: boolean): Promise<{ text: string; images: OutgoingImage[] }> {
if (!refs || refs.length === 0) return { text, images: [] };
const images: OutgoingImage[] = [];
const failures: string[] = [];
for (const ref of refs) {
const image = await readOutboundImage(ref);
if (image) images.push(image);
else failures.push(ref.filename || "image");
}
let finalText = text;
if (failures.length > 0) {
finalText = [finalText, ...failures.map((name) => `(图片 ${name} 发送失败:文件不存在或已失效)`)].filter((part) => part.length > 0).join("\n");
}
if (!imageCapable && images.length > 0) {
finalText = [finalText, ...images.map((image) => `[图片] ${image.filename || "image"}`)].filter((part) => part.length > 0).join("\n");
return { text: finalText, images: [] };
}
return { text: finalText, images };
}
class ReplyStream {
private tail = Promise.resolve();
@@ -36,29 +64,47 @@ class ReplyStream {
private readonly nextSequence: () => number | undefined
) {}
enqueue(text: string, onError?: (error: unknown) => void): Promise<void> {
const replySequence = this.nextSequence();
this.tail = this.tail.then(() => this.adapter.sendMessage({
enqueue(entry: string | PendingEvent, onError?: (error: unknown) => void): Promise<void> {
const request = typeof entry === "string" ? { text: entry } : entry;
this.tail = this.tail.then(() => this.deliver(request)).catch((error: unknown) => {
console.error(`Gateway delivery send failed (platform=${this.message.platform})`);
onError?.(error);
});
return this.tail;
}
private async deliver(request: PendingEvent): Promise<void> {
const { text, images } = await resolveOutboundImages(request.text, request.images, this.adapter.supportsImages === true);
if (text || images.length === 0) await this.send({ text });
for (const image of images) {
try {
await this.send({ text: "", images: [image] });
} catch (error) {
console.error(`Gateway image delivery failed (platform=${this.message.platform})`);
await this.send({ text: `(图片 ${image.filename || "image"} 发送失败)` }).catch(() => undefined);
}
}
}
private send(piece: { text: string; images?: OutgoingImage[] }): Promise<void> {
return this.adapter.sendMessage({
target: {
platform: this.message.platform,
chatId: this.message.chatId,
userId: this.message.userId,
raw: this.message.raw
},
text,
text: piece.text,
replyTo: this.message.messageId,
replySequence
})).catch((error: unknown) => {
console.error(`Gateway delivery send failed (platform=${this.message.platform})`);
onError?.(error);
replySequence: this.nextSequence(),
images: piece.images
});
return this.tail;
}
}
export class Gateway {
private readonly chatTargets = new Map<string, ChatTarget>();
private readonly pendingEvents = new Map<string, string[]>();
private readonly pendingEvents = new Map<string, PendingEvent[]>();
// QQ deduplicates passive replies by msg_id + msg_seq: all sends referencing the same inbound
// message (normal replies, events, redelivered pending events) share one counter per chat+messageId.
private readonly messageSequences = new Map<string, Map<string, number>>();
@@ -110,16 +156,16 @@ export class Gateway {
if (options.synchronous) return this.promptResult(message);
try {
const reply = (await this.runtime.prompt({
const response = await this.runtime.prompt({
platform: message.platform,
chatId: message.chatId,
userId: message.userId,
text: message.text,
messageId: message.messageId,
attachments: message.attachments
})).text;
await this.replyStream(chatKey, message, adapter).enqueue(reply);
return { ok: true, reply };
});
await this.replyStream(chatKey, message, adapter).enqueue({ text: response.text, images: response.images });
return { ok: true, reply: response.text };
} catch (error) {
const errorText = error instanceof Error ? error.message : String(error);
const reply = `Agent error: ${errorText}`;
@@ -128,7 +174,7 @@ export class Gateway {
}
}
async sendEvent(chatKey: string, text: string): Promise<void> {
async sendEvent(chatKey: string, text: string, images?: OutgoingImageRef[]): Promise<void> {
if (this.closed) return;
const entry = this.chatTargets.get(chatKey);
if (!entry) return;
@@ -139,21 +185,44 @@ export class Gateway {
if (!freshInbound) {
// No fresh passive window and proactive messages may be unauthorized: skip and remind on the next inbound.
this.recordEventDelivery(chatKey, "skipped", "no fresh passive window");
this.queuePendingEvent(chatKey, text);
this.queuePendingEvent(chatKey, { text, images });
console.log(`Gateway event skipped: no fresh passive window (platform=${entry.target.platform})`);
return;
}
try {
await entry.adapter.sendMessage({
target: entry.target,
text,
replyTo: entry.lastInboundMessageId,
replySequence: this.nextReplySequence(chatKey, entry.lastInboundMessageId)
});
const resolved = await resolveOutboundImages(text, images, entry.adapter.supportsImages === true);
if (resolved.text || resolved.images.length === 0) {
await entry.adapter.sendMessage({
target: entry.target,
text: resolved.text,
replyTo: entry.lastInboundMessageId,
replySequence: this.nextReplySequence(chatKey, entry.lastInboundMessageId)
});
}
for (const image of resolved.images) {
try {
await entry.adapter.sendMessage({
target: entry.target,
text: "",
replyTo: entry.lastInboundMessageId,
replySequence: this.nextReplySequence(chatKey, entry.lastInboundMessageId),
images: [image]
});
} catch (imageError) {
// Image upload/send failures never block the event text; degrade to a text note.
console.error(`Gateway event image send failed (platform=${entry.target.platform})`);
await entry.adapter.sendMessage({
target: entry.target,
text: `(图片 ${image.filename || "image"} 发送失败)`,
replyTo: entry.lastInboundMessageId,
replySequence: this.nextReplySequence(chatKey, entry.lastInboundMessageId)
}).catch(() => undefined);
}
}
this.recordEventDelivery(chatKey, "success");
} catch (error) {
this.recordEventDelivery(chatKey, "failed", safeEventError(error));
this.queuePendingEvent(chatKey, text);
this.queuePendingEvent(chatKey, { text, images });
console.error(`Gateway event send failed (platform=${entry.target.platform})`);
}
}
@@ -166,9 +235,9 @@ export class Gateway {
entry.lastEventAt = this.now();
}
private queuePendingEvent(chatKey: string, text: string): void {
private queuePendingEvent(chatKey: string, event: PendingEvent): void {
const pending = this.pendingEvents.get(chatKey) || [];
pending.push(text);
pending.push(event);
this.pendingEvents.set(chatKey, pending);
}
@@ -178,11 +247,12 @@ export class Gateway {
if (!pending || pending.length === 0) return;
this.pendingEvents.delete(chatKey);
const stream = this.replyStream(chatKey, message, adapter);
for (const text of pending) {
await stream.enqueue(text, (error) => {
for (const event of pending) {
// Images are re-read from disk at redelivery time (the worker may have updated them).
await stream.enqueue(event, (error) => {
// Redelivery failed: keep the event queued for the next inbound and record the failure.
this.recordEventDelivery(chatKey, "failed", safeEventError(error));
this.queuePendingEvent(chatKey, text);
this.queuePendingEvent(chatKey, event);
});
}
}