From 980f3a0c8fcc67c7bbc59ad01c7de011c4187fdb Mon Sep 17 00:00:00 2001 From: lofyer Date: Thu, 13 Aug 2026 02:09:09 +0800 Subject: [PATCH] fix: preserve channel message delivery --- src/main/assistant/assistant-database.test.ts | 73 ++++++- src/main/assistant/assistant-database.ts | 66 +++++- src/main/assistant/heartbeat-database.test.ts | 2 +- src/main/channels/channel-driver.ts | 24 ++- src/main/channels/channel-service.test.ts | 189 ++++++++++++++++- src/main/channels/channel-service.ts | 200 ++++++++++++------ .../channels/dingtalk-channel-driver.test.ts | 42 ++++ src/main/channels/dingtalk-channel-driver.ts | 9 +- src/main/channels/sqlite-channel-state.ts | 8 +- .../channels/wechat-channel-driver.test.ts | 1 + src/main/channels/wechat-channel-driver.ts | 1 + .../channels/wecom-channel-driver.test.ts | 39 ++++ src/main/channels/wecom-channel-driver.ts | 9 +- src/main/ipc.ts | 52 ++--- src/shared/channel-contracts.ts | 1 + 15 files changed, 586 insertions(+), 130 deletions(-) diff --git a/src/main/assistant/assistant-database.test.ts b/src/main/assistant/assistant-database.test.ts index 718c5a9..de765d8 100644 --- a/src/main/assistant/assistant-database.test.ts +++ b/src/main/assistant/assistant-database.test.ts @@ -98,7 +98,7 @@ describe('AssistantDatabase', () => { database.close() }) - it('migrates existing databases to schema version 18', async () => { + it('migrates existing databases to schema version 19', async () => { const directory = await mkdtemp( join(tmpdir(), 'goodbuddy-assistant-migration-') ) @@ -127,7 +127,7 @@ describe('AssistantDatabase', () => { user_version: number } ).user_version - ).toBe(18) + ).toBe(19) expect( current .prepare( @@ -231,7 +231,7 @@ describe('AssistantDatabase', () => { user_version: number } ).user_version - ).toBe(18) + ).toBe(19) expect( current .prepare( @@ -630,9 +630,18 @@ describe('AssistantDatabase', () => { const databasePath = join(directory, 'assistant.sqlite') const database = new AssistantDatabase(databasePath) database.initialize('C:\\Workspace') - expect(database.claimChannelEvent('weixin', 'event-1')).toBe(true) - expect(database.claimChannelEvent('weixin', 'event-1')).toBe(false) - expect(database.claimChannelEvent('dingtalk', 'event-1')).toBe(true) + expect( + database.claimChannelEvent('weixin', 'account-1', 'event-1') + ).toBe(true) + expect( + database.claimChannelEvent('weixin', 'account-1', 'event-1') + ).toBe(false) + expect( + database.claimChannelEvent('weixin', 'account-2', 'event-1') + ).toBe(true) + expect( + database.claimChannelEvent('dingtalk', 'account-1', 'event-1') + ).toBe(true) const entry = database.enqueueChannelResult({ channel: 'weixin', @@ -656,12 +665,58 @@ describe('AssistantDatabase', () => { const reopened = new AssistantDatabase(databasePath) reopened.initialize('C:\\Workspace') - expect(reopened.claimChannelEvent('weixin', 'event-1')).toBe( - false - ) + expect( + reopened.claimChannelEvent('weixin', 'account-1', 'event-1') + ).toBe(false) reopened.close() }) + it('preserves legacy channel event claims while adding account identity', async () => { + const directory = await mkdtemp( + join(tmpdir(), 'goodbuddy-channel-event-migration-') + ) + temporaryDirectories.push(directory) + const databasePath = join(directory, 'assistant.sqlite') + const initial = new AssistantDatabase(databasePath) + initial.initialize('C:\\Workspace') + initial.close() + + const legacy = new DatabaseSync(databasePath) + legacy.exec(` + DROP TABLE channel_events; + CREATE TABLE channel_events ( + channel TEXT NOT NULL, + event_id TEXT NOT NULL, + claimed_at INTEGER NOT NULL, + PRIMARY KEY(channel, event_id) + ); + CREATE INDEX channel_events_claimed_at + ON channel_events(claimed_at); + INSERT INTO channel_events(channel, event_id, claimed_at) + VALUES ('weixin', 'legacy-event', 1); + PRAGMA user_version = 18; + `) + legacy.close() + + const migrated = new AssistantDatabase(databasePath) + migrated.initialize('C:\\Workspace') + expect( + migrated.claimChannelEvent( + 'weixin', + 'default', + 'legacy-event' + ) + ).toBe(false) + expect( + migrated.claimChannelEvent( + 'weixin', + 'new-account', + 'legacy-event' + ) + ).toBe(true) + migrated.close() + }) + it('safely deletes a confirmed project and its scoped data', async () => { const database = await createDatabase() const project = database.createProject({ diff --git a/src/main/assistant/assistant-database.ts b/src/main/assistant/assistant-database.ts index fcb8389..e3e55e1 100644 --- a/src/main/assistant/assistant-database.ts +++ b/src/main/assistant/assistant-database.ts @@ -1636,15 +1636,19 @@ export class AssistantDatabase { } } - claimChannelEvent(channel: string, eventId: string): boolean { + claimChannelEvent( + channel: string, + accountId: string, + eventId: string + ): boolean { const database = this.requireDatabase() const result = database .prepare( `INSERT OR IGNORE INTO channel_events - (channel, event_id, claimed_at) - VALUES (?, ?, ?)` + (channel, account_id, event_id, claimed_at) + VALUES (?, ?, ?, ?)` ) - .run(channel, eventId, Date.now()) + .run(channel, accountId, eventId, Date.now()) if (result.changes === 1) { this.channelEventWrites += 1 if (this.channelEventWrites % 128 === 0) { @@ -1664,12 +1668,17 @@ export class AssistantDatabase { return result.changes === 1 } - releaseChannelEvent(channel: string, eventId: string): void { + releaseChannelEvent( + channel: string, + accountId: string, + eventId: string + ): void { this.requireDatabase() .prepare( - 'DELETE FROM channel_events WHERE channel = ? AND event_id = ?' + `DELETE FROM channel_events + WHERE channel = ? AND account_id = ? AND event_id = ?` ) - .run(channel, eventId) + .run(channel, accountId, eventId) } enqueueChannelResult(message: ChannelResultMessage): { @@ -4475,12 +4484,12 @@ export class AssistantDatabase { const version = database .prepare('PRAGMA user_version') .get() as { user_version: number } - if (version.user_version > 18) { + if (version.user_version > 19) { throw new Error( `当前 GoodBuddy 不支持助理数据库版本 ${version.user_version}，请升级应用后重试` ) } - if (version.user_version === 18) { + if (version.user_version === 19) { return } if (version.user_version < 1) { @@ -5162,9 +5171,10 @@ export class AssistantDatabase { database.exec(` CREATE TABLE IF NOT EXISTS channel_events ( channel TEXT NOT NULL, + account_id TEXT NOT NULL DEFAULT 'default', event_id TEXT NOT NULL, claimed_at INTEGER NOT NULL, - PRIMARY KEY(channel, event_id) + PRIMARY KEY(channel, account_id, event_id) ); CREATE INDEX IF NOT EXISTS channel_events_claimed_at ON channel_events(claimed_at); @@ -5353,6 +5363,42 @@ export class AssistantDatabase { throw error } } + if (version.user_version < 19) { + database.exec('BEGIN IMMEDIATE') + try { + const eventColumns = new Set( + ( + database + .prepare('PRAGMA table_info(channel_events)') + .all() as Array<{ name: string }> + ).map((column) => column.name) + ) + if (!eventColumns.has('account_id')) { + database.exec(` + ALTER TABLE channel_events RENAME TO channel_events_legacy; + DROP INDEX IF EXISTS channel_events_claimed_at; + CREATE TABLE channel_events ( + channel TEXT NOT NULL, + account_id TEXT NOT NULL DEFAULT 'default', + event_id TEXT NOT NULL, + claimed_at INTEGER NOT NULL, + PRIMARY KEY(channel, account_id, event_id) + ); + INSERT INTO channel_events + (channel, account_id, event_id, claimed_at) + SELECT channel, 'default', event_id, claimed_at + FROM channel_events_legacy; + DROP TABLE channel_events_legacy; + CREATE INDEX channel_events_claimed_at + ON channel_events(claimed_at); + `) + } + database.exec('PRAGMA user_version = 19; COMMIT;') + } catch (error) { + database.exec('ROLLBACK') + throw error + } + } } private requireDatabase(): DatabaseSync { diff --git a/src/main/assistant/heartbeat-database.test.ts b/src/main/assistant/heartbeat-database.test.ts index 921eb99..ffa61de 100644 --- a/src/main/assistant/heartbeat-database.test.ts +++ b/src/main/assistant/heartbeat-database.test.ts @@ -100,7 +100,7 @@ describe('AssistantDatabase heartbeat persistence', () => { ).count check.close() migrated.close() - expect(version).toBe(18) + expect(version).toBe(19) expect(heartbeatTableCount).toBe(3) }) diff --git a/src/main/channels/channel-driver.ts b/src/main/channels/channel-driver.ts index d36fb63..c357ce5 100644 --- a/src/main/channels/channel-driver.ts +++ b/src/main/channels/channel-driver.ts @@ -20,8 +20,16 @@ export interface ChannelDriver { } export interface DedupStore { - claim(channel: string, eventId: string): boolean | Promise - release(channel: string, eventId: string): void | Promise + claim( + channel: string, + accountId: string, + eventId: string + ): boolean | Promise + release( + channel: string, + accountId: string, + eventId: string + ): void | Promise } export class MemoryDedupStore implements DedupStore { @@ -33,8 +41,8 @@ export class MemoryDedupStore implements DedupStore { } } - claim(channel: string, eventId: string): boolean { - const key = this.key(channel, eventId) + claim(channel: string, accountId: string, eventId: string): boolean { + const key = this.key(channel, accountId, eventId) if (this.claimed.has(key)) { return false } @@ -50,16 +58,16 @@ export class MemoryDedupStore implements DedupStore { return true } - release(channel: string, eventId: string): void { - this.claimed.delete(this.key(channel, eventId)) + release(channel: string, accountId: string, eventId: string): void { + this.claimed.delete(this.key(channel, accountId, eventId)) } clear(): void { this.claimed.clear() } - private key(channel: string, eventId: string): string { - return `${channel}\u0000${eventId}` + private key(channel: string, accountId: string, eventId: string): string { + return `${channel}\u0000${accountId}\u0000${eventId}` } } diff --git a/src/main/channels/channel-service.test.ts b/src/main/channels/channel-service.test.ts index 718eccd..1ca6cc7 100644 --- a/src/main/channels/channel-service.test.ts +++ b/src/main/channels/channel-service.test.ts @@ -50,6 +50,7 @@ function inbound( ): ChannelInboundText { return { channel: 'fake', + accountId: 'default', eventId: 'event-1', senderId: 'allowed-user', conversationId: 'conversation-1', @@ -83,6 +84,7 @@ describe('channel contracts', () => { }) ).toEqual({ channel: 'fake', + accountId: 'default', eventId: 'event-1', senderId: 'user-1', conversationId: 'direct-1', @@ -135,7 +137,7 @@ describe('channel contracts', () => { }) describe('ChannelService', () => { - it('acknowledges first and denies all senders when no allowlist is configured', async () => { + it('acknowledges after accepting input and denies all senders when no allowlist is configured', async () => { const driver = new FakeChannelDriver() const executor = vi.fn() const service = new ChannelService(driver, executor) @@ -149,6 +151,17 @@ describe('ChannelService', () => { await service.stop() }) + it('does not acknowledge malformed input', async () => { + const driver = new FakeChannelDriver() + const service = new ChannelService(driver, vi.fn()) + await service.start() + + await driver.emit({ channel: 'fake' }) + + expect(driver.acknowledgements).toBe(0) + await service.stop() + }) + it('executes an allowed request asynchronously with the normalized ask mode', async () => { const driver = new FakeChannelDriver() let finish: ((value: { status: string; output: string }) => void) | undefined @@ -173,6 +186,9 @@ describe('ChannelService', () => { }) expect(driver.acknowledgements).toBe(1) + await vi.waitFor(() => { + expect(executor).toHaveBeenCalledOnce() + }) expect(executor).toHaveBeenCalledWith( expect.objectContaining({ text: '帮我分析', @@ -280,9 +296,10 @@ describe('ChannelService', () => { it('deduplicates by channel and event id', async () => { const store = new MemoryDedupStore() - expect(store.claim('first', 'same-id')).toBe(true) - expect(store.claim('first', 'same-id')).toBe(false) - expect(store.claim('second', 'same-id')).toBe(true) + expect(store.claim('first', 'account-1', 'same-id')).toBe(true) + expect(store.claim('first', 'account-1', 'same-id')).toBe(false) + expect(store.claim('first', 'account-2', 'same-id')).toBe(true) + expect(store.claim('second', 'account-1', 'same-id')).toBe(true) const driver = new FakeChannelDriver() const executor = vi.fn(async () => ({ @@ -303,6 +320,154 @@ describe('ChannelService', () => { await service.stop() }) + it('does not deduplicate matching event ids from different accounts', async () => { + const driver = new FakeChannelDriver() + const executor = vi.fn(async () => ({ + status: 'completed', + output: 'done' + })) + const service = new ChannelService(driver, executor, { + allowedSenderIds: ['allowed-user'] + }) + await service.start() + + await driver.emit( + inbound({ + accountId: 'account-1', + eventId: 'shared-event', + conversationId: 'shared-conversation' + }) + ) + await driver.emit( + inbound({ + accountId: 'account-2', + eventId: 'shared-event', + conversationId: 'shared-conversation' + }) + ) + + await waitForSent(driver, 2) + expect(executor).toHaveBeenCalledTimes(2) + await service.stop() + }) + + it('serializes requests from the same conversation', async () => { + const driver = new FakeChannelDriver() + const finishes: Array<() => void> = [] + const executor = vi.fn( + (message: ChannelInboundText) => + new Promise<{ status: string; output: string }>((resolve) => { + finishes.push(() => + resolve({ + status: 'completed', + output: message.eventId + }) + ) + }) + ) + const service = new ChannelService(driver, executor, { + allowedSenderIds: ['allowed-user'], + maximumConcurrency: 2 + }) + await service.start() + + await driver.emit(inbound({ eventId: 'first' })) + await driver.emit(inbound({ eventId: 'second' })) + + expect(executor).toHaveBeenCalledOnce() + finishes[0]?.() + await vi.waitFor(() => { + expect(executor).toHaveBeenCalledTimes(2) + }) + finishes[1]?.() + await waitForSent(driver, 2) + expect(driver.sent.map((message) => message.output)).toEqual([ + 'first', + 'second' + ]) + await service.stop() + }) + + it('keeps failed deliveries in the outbox without sending a second result', async () => { + class FailingDriver extends FakeChannelDriver { + attempts = 0 + + override async send( + message: ChannelResultMessage, + signal: AbortSignal + ): Promise { + void message + void signal + this.attempts += 1 + throw new Error('offline') + } + } + + const driver = new FailingDriver() + const outbox = new MemoryOutbox() + const service = new ChannelService( + driver, + async () => ({ status: 'completed', output: '完成' }), + { + allowedSenderIds: ['allowed-user'], + outbox + } + ) + await service.start() + await driver.emit(inbound({ eventId: 'delivery-failure' })) + + await vi.waitFor(() => { + expect(driver.attempts).toBe(1) + }) + expect(await outbox.listUndelivered()).toEqual([ + expect.objectContaining({ + state: 'failed', + attempts: 1, + message: expect.objectContaining({ + eventId: 'delivery-failure', + status: 'completed' + }) + }) + ]) + await service.stop() + }) + + it('releases the event claim when no durable result can be queued', async () => { + const driver = new FakeChannelDriver() + const store = new MemoryDedupStore() + const outbox = { + enqueue: vi.fn(() => { + throw new Error('database unavailable') + }), + markDelivered: vi.fn(), + markFailed: vi.fn(), + listUndelivered: vi.fn(() => []) + } + const deliveryFailure = vi.fn() + const executor = vi.fn(async () => ({ + status: 'completed', + output: '完成' + })) + const service = new ChannelService(driver, executor, { + allowedSenderIds: ['allowed-user'], + dedupStore: store, + outbox, + onDeliveryFailure: deliveryFailure + }) + await service.start() + await driver.emit(inbound({ eventId: 'retryable' })) + await vi.waitFor(() => { + expect(outbox.enqueue).toHaveBeenCalledOnce() + }) + await driver.emit(inbound({ eventId: 'retryable' })) + await vi.waitFor(() => { + expect(outbox.enqueue).toHaveBeenCalledTimes(2) + }) + expect(executor).toHaveBeenCalledTimes(2) + expect(deliveryFailure).toHaveBeenCalled() + await service.stop() + }) + it('enforces concurrency and input length limits', async () => { const driver = new FakeChannelDriver() let finish: (() => void) | undefined @@ -320,8 +485,20 @@ describe('ChannelService', () => { await service.start() await driver.emit(inbound({ eventId: 'active', text: '12345' })) - await driver.emit(inbound({ eventId: 'busy', text: '12345' })) - await driver.emit(inbound({ eventId: 'too-long', text: '123456' })) + await driver.emit( + inbound({ + eventId: 'busy', + conversationId: 'conversation-2', + text: '12345' + }) + ) + await driver.emit( + inbound({ + eventId: 'too-long', + conversationId: 'conversation-3', + text: '123456' + }) + ) await waitForSent(driver, 2) expect(driver.sent).toEqual( diff --git a/src/main/channels/channel-service.ts b/src/main/channels/channel-service.ts index 3ea1ef4..ca23103 100644 --- a/src/main/channels/channel-service.ts +++ b/src/main/channels/channel-service.ts @@ -89,8 +89,8 @@ export class ChannelService { private readonly outbox: Outbox private readonly onDeliveryFailure?: (error: unknown) => void private readonly onDeliverySuccess?: () => void - private readonly tasks = new Set>() private readonly active = new Map() + private readonly conversationTails = new Map>() private state: ServiceState = 'idle' private stopPromise?: Promise @@ -149,18 +149,17 @@ export class ChannelService { this.state = 'running' try { await this.driver.start(async (rawMessage, acknowledge) => { - await acknowledge() if (this.state !== 'running') { + await acknowledge() return } - const task = this.process(rawMessage).catch(() => { - // Processing failures are converted to bounded channel results. - }) - this.tasks.add(task) - void task.finally(() => { - this.tasks.delete(task) - }) + try { + this.enqueue(rawMessage) + await acknowledge() + } catch (error) { + this.onDeliveryFailure?.(error) + } }) await this.retryUndelivered() } catch (error) { @@ -170,9 +169,12 @@ export class ChannelService { } cancel(eventId: string): boolean { - const controller = this.active.get( - this.activeKey(this.driver.channel, eventId) - ) + const suffix = `\u0000${eventId}` + const controller = [...this.active.entries()].find( + ([key]) => + key.startsWith(`${this.driver.channel}\u0000`) && + key.endsWith(suffix) + )?.[1] if (!controller) { return false } @@ -201,7 +203,7 @@ export class ChannelService { const driverStop = Promise.resolve().then(() => this.driver.stop()) const results = await Promise.allSettled([ driverStop, - ...this.tasks + ...this.conversationTails.values() ]) const driverResult = results[0] if (driverResult?.status === 'rejected') { @@ -259,73 +261,96 @@ export class ChannelService { const claimed = await this.dedupStore.claim( message.channel, + message.accountId, message.eventId ) if (!claimed) { return } - if (message.text.length > this.maximumInputLength) { - await this.deliver( - this.result(message, { - status: 'rejected', - error: `消息过长，最多允许 ${this.maximumInputLength} 个字符` - }), - new AbortController().signal - ) - return - } - - if (this.active.size >= this.maximumConcurrency) { - await this.deliver( - this.result(message, { - status: 'busy', - error: '当前请求较多，请稍后重试' - }), - new AbortController().signal - ) - return - } - - const key = this.activeKey(message.channel, message.eventId) - const controller = new AbortController() - this.active.set(key, controller) + let durableResult = false try { - const rawResult = await this.execute(message, controller.signal) - if (controller.signal.aborted) { - await this.deliver( + if (message.text.length > this.maximumInputLength) { + durableResult = await this.tryDeliver( this.result(message, { - status: 'cancelled', - error: '请求已取消' + status: 'rejected', + error: `消息过长，最多允许 ${this.maximumInputLength} 个字符` }), new AbortController().signal ) return } - const result = channelExecutorResultSchema.safeParse(rawResult) - if (!result.success) { - await this.deliver( + if (this.active.size >= this.maximumConcurrency) { + durableResult = await this.tryDeliver( this.result(message, { - status: 'failed', - error: '请求返回了无效结果' + status: 'busy', + error: '当前请求较多，请稍后重试' }), - controller.signal + new AbortController().signal ) return } - await this.deliver(this.result(message, result.data), controller.signal) - } catch { - const cancelled = controller.signal.aborted - await this.deliver( - this.result(message, { - status: cancelled ? 'cancelled' : 'failed', - error: cancelled ? '请求已取消' : '请求处理失败' - }), - new AbortController().signal + + const key = this.activeKey( + message.channel, + message.accountId, + message.eventId ) + const controller = new AbortController() + this.active.set(key, controller) + try { + let rawResult: Awaited> + try { + rawResult = await this.execute(message, controller.signal) + } catch { + const cancelled = controller.signal.aborted + durableResult = await this.tryDeliver( + this.result(message, { + status: cancelled ? 'cancelled' : 'failed', + error: cancelled ? '请求已取消' : '请求处理失败' + }), + new AbortController().signal + ) + return + } + if (controller.signal.aborted) { + durableResult = await this.tryDeliver( + this.result(message, { + status: 'cancelled', + error: '请求已取消' + }), + new AbortController().signal + ) + return + } + + const result = channelExecutorResultSchema.safeParse(rawResult) + if (!result.success) { + durableResult = await this.tryDeliver( + this.result(message, { + status: 'failed', + error: '请求返回了无效结果' + }), + controller.signal + ) + return + } + durableResult = await this.tryDeliver( + this.result(message, result.data), + controller.signal + ) + } finally { + this.active.delete(key) + } } finally { - this.active.delete(key) + if (!durableResult) { + await this.dedupStore.release( + message.channel, + message.accountId, + message.eventId + ) + } } } @@ -420,7 +445,7 @@ export class ChannelService { private async deliver( message: ChannelResultMessage, signal: AbortSignal - ): Promise { + ): Promise { const entry = await this.outbox.enqueue(message) try { await this.driver.send(message, signal) @@ -429,11 +454,60 @@ export class ChannelService { } catch (error) { await this.outbox.markFailed(entry.id) this.onDeliveryFailure?.(error) - throw error + } + return true + } + + private async tryDeliver( + message: ChannelResultMessage, + signal: AbortSignal + ): Promise { + try { + return await this.deliver(message, signal) + } catch (error) { + this.onDeliveryFailure?.(error) + return false } } - private activeKey(channel: string, eventId: string): string { - return `${channel}\u0000${eventId}` + private activeKey( + channel: string, + accountId: string, + eventId: string + ): string { + return `${channel}\u0000${accountId}\u0000${eventId}` + } + + private enqueue(rawMessage: unknown): void { + const parsed = channelInboundTextSchema.safeParse(rawMessage) + if (!parsed.success) { + throw new Error('通道消息格式无效') + } + if (parsed.data.channel !== this.driver.channel) { + throw new Error('通道消息来源不匹配') + } + const key = + `${parsed.data.channel}\u0000${parsed.data.accountId}` + + `\u0000${parsed.data.conversationId}` + const previous = this.conversationTails.get(key) ?? Promise.resolve() + const task = + this.conversationTails.has(key) + ? previous + .catch(() => undefined) + .then(() => this.process(parsed.data)) + : this.process(parsed.data) + const tail = task.then( + () => undefined, + () => undefined + ) + this.conversationTails.set(key, tail) + void tail.finally(() => { + if (this.conversationTails.get(key) === tail) { + this.conversationTails.delete(key) + } + }) + void task.catch(() => { + // The event claim is released when no durable result could be recorded. + }) } } diff --git a/src/main/channels/dingtalk-channel-driver.test.ts b/src/main/channels/dingtalk-channel-driver.test.ts index 6a7a168..c0ab321 100644 --- a/src/main/channels/dingtalk-channel-driver.test.ts +++ b/src/main/channels/dingtalk-channel-driver.test.ts @@ -66,6 +66,7 @@ describe('DingTalkChannelDriver', () => { expect(messages).toEqual([ { channel: 'dingtalk', + accountId: 'client-id', eventId: 'event-1', senderId: 'user-1', conversationId: 'conversation-1', @@ -177,4 +178,45 @@ describe('DingTalkChannelDriver', () => { { status: 'SUCCESS' } ) }) + + it('rejects unsupported attachments without consuming the reply context', async () => { + const transport = new FakeTransport() + const driver = new DingTalkChannelDriver({ + clientId: 'client-id', + clientSecret: 'client-secret', + allowedSenderIds: ['user-1'], + transportFactory: { + create: async () => transport + } + }) + await driver.start(() => undefined) + await transport.listener?.(envelope('media-event')) + + const message = { + channel: 'dingtalk' as const, + eventId: 'media-event', + conversationId: 'conversation-1', + recipientId: 'user-1', + status: 'completed', + output: '文件已生成', + attachments: [ + { + name: 'result.txt', + mimeType: 'text/plain', + size: 2, + kind: 'file' as const, + dataBase64: 'b2s=' + } + ] + } + await expect( + driver.send(message, new AbortController().signal) + ).rejects.toThrow('暂不支持发送附件') + await driver.send( + { ...message, attachments: undefined }, + new AbortController().signal + ) + expect(transport.replyText).toHaveBeenCalledOnce() + await driver.stop() + }) }) diff --git a/src/main/channels/dingtalk-channel-driver.ts b/src/main/channels/dingtalk-channel-driver.ts index 8f0c010..942c546 100644 --- a/src/main/channels/dingtalk-channel-driver.ts +++ b/src/main/channels/dingtalk-channel-driver.ts @@ -194,6 +194,7 @@ function resultText(message: ChannelResultMessage): string { export class DingTalkChannelDriver implements ChannelDriver { readonly channel = 'dingtalk' + private readonly accountId: string private readonly driver: DingTalkDriver private readonly maximumContexts: number @@ -201,6 +202,7 @@ export class DingTalkChannelDriver implements ChannelDriver { private handler?: ChannelInboundHandler constructor(options: DingTalkChannelDriverOptions) { + this.accountId = options.clientId this.maximumContexts = maximumReplyContexts( options.maximumReplyContexts ) @@ -230,6 +232,9 @@ export class DingTalkChannelDriver implements ChannelDriver { message: ChannelResultMessage, signal: AbortSignal ): Promise { + if (message.attachments?.length) { + throw new Error('钉钉通道暂不支持发送附件') + } const record = this.replyContexts.get(message.eventId) if ( !record || @@ -245,7 +250,8 @@ export class DingTalkChannelDriver implements ChannelDriver { await this.driver.reply(record.context, resultText(message)) } catch { throw new Error('钉钉消息回复失败') - } finally { + } + if (!message.attachments?.length) { this.replyContexts.delete(message.eventId) } } @@ -276,6 +282,7 @@ export class DingTalkChannelDriver implements ChannelDriver { this.enforceContextLimit() const inbound: ChannelInboundText = { channel: this.channel, + accountId: this.accountId, eventId: message.dedupeKey, senderId: message.senderId, conversationId: message.conversationId, diff --git a/src/main/channels/sqlite-channel-state.ts b/src/main/channels/sqlite-channel-state.ts index 4981d85..53f4450 100644 --- a/src/main/channels/sqlite-channel-state.ts +++ b/src/main/channels/sqlite-channel-state.ts @@ -9,12 +9,12 @@ import type { ChannelResultMessage } from '../../shared/channel-contracts' export class SqliteChannelDedupStore implements DedupStore { constructor(private readonly database: AssistantDatabase) {} - claim(channel: string, eventId: string): boolean { - return this.database.claimChannelEvent(channel, eventId) + claim(channel: string, accountId: string, eventId: string): boolean { + return this.database.claimChannelEvent(channel, accountId, eventId) } - release(channel: string, eventId: string): void { - this.database.releaseChannelEvent(channel, eventId) + release(channel: string, accountId: string, eventId: string): void { + this.database.releaseChannelEvent(channel, accountId, eventId) } } diff --git a/src/main/channels/wechat-channel-driver.test.ts b/src/main/channels/wechat-channel-driver.test.ts index 6858cfe..8da6ac5 100644 --- a/src/main/channels/wechat-channel-driver.test.ts +++ b/src/main/channels/wechat-channel-driver.test.ts @@ -77,6 +77,7 @@ describe('WechatChannelDriver', () => { expect(handler).toHaveBeenCalledWith( expect.objectContaining({ channel: 'weixin', + accountId: 'bot-account', eventId: 'event-1', senderId: 'sender-1', workMode: 'ask', diff --git a/src/main/channels/wechat-channel-driver.ts b/src/main/channels/wechat-channel-driver.ts index 23e8276..23c82f0 100644 --- a/src/main/channels/wechat-channel-driver.ts +++ b/src/main/channels/wechat-channel-driver.ts @@ -186,6 +186,7 @@ export class WechatChannelDriver implements ChannelDriver { this.handler?.( { channel: this.channel, + accountId: this.settings.accountId, eventId: message.eventId, senderId: message.senderId, conversationId: message.conversationId, diff --git a/src/main/channels/wecom-channel-driver.test.ts b/src/main/channels/wecom-channel-driver.test.ts index 89d2b2e..c9095e0 100644 --- a/src/main/channels/wecom-channel-driver.test.ts +++ b/src/main/channels/wecom-channel-driver.test.ts @@ -87,6 +87,7 @@ describe('WeComChannelDriver', () => { transport.emit(groupFrame('event-2', 'request-2')) expect(messages[0]).toEqual({ channel: 'wecom', + accountId: 'bot-1', eventId: 'event-1', senderId: 'user-1', conversationId: 'group-1', @@ -130,4 +131,42 @@ describe('WeComChannelDriver', () => { await driver.stop() expect(transport.disconnect).toHaveBeenCalledOnce() }) + + it('rejects unsupported attachments without consuming the reply context', async () => { + const transport = new FakeTransport() + const driver = new WeComChannelDriver({ + botId: 'bot-1', + secret: 'secret', + transportFactory: () => transport + }) + await driver.start(() => undefined) + transport.emit(groupFrame('media-event', 'media-request')) + + const message = { + channel: 'wecom' as const, + eventId: 'media-event', + conversationId: 'group-1', + recipientId: 'user-1', + status: 'completed', + output: '文件已生成', + attachments: [ + { + name: 'result.txt', + mimeType: 'text/plain', + size: 2, + kind: 'file' as const, + dataBase64: 'b2s=' + } + ] + } + await expect( + driver.send(message, new AbortController().signal) + ).rejects.toThrow('回复失败') + await driver.send( + { ...message, attachments: undefined }, + new AbortController().signal + ) + expect(transport.replyStream).toHaveBeenCalledOnce() + await driver.stop() + }) }) diff --git a/src/main/channels/wecom-channel-driver.ts b/src/main/channels/wecom-channel-driver.ts index 766897b..d99f8af 100644 --- a/src/main/channels/wecom-channel-driver.ts +++ b/src/main/channels/wecom-channel-driver.ts @@ -40,12 +40,14 @@ function resultText(message: ChannelResultMessage): string { export class WeComChannelDriver implements ChannelDriver { readonly channel = 'wecom' + private readonly accountId: string private readonly driver: WeComDriver private readonly maximumContexts: number private readonly replyContexts = new Map() private handler?: ChannelInboundHandler constructor(options: WeComChannelDriverOptions) { + this.accountId = options.botId this.maximumContexts = maximumReplyContexts( options.maximumReplyContexts ) @@ -86,11 +88,13 @@ export class WeComChannelDriver implements ChannelDriver { try { signal.throwIfAborted() await this.driver.reply(record.context, { - text: resultText(message) + text: resultText(message), + attachments: message.attachments }) } catch { throw new Error('企业微信消息回复失败') - } finally { + } + if (!message.attachments?.length) { this.replyContexts.delete(message.eventId) } } @@ -119,6 +123,7 @@ export class WeComChannelDriver implements ChannelDriver { this.enforceContextLimit() const inbound: ChannelInboundText = { channel: this.channel, + accountId: this.accountId, eventId: message.eventId, senderId: message.userId, conversationId: message.conversationId, diff --git a/src/main/ipc.ts b/src/main/ipc.ts index 2344dce..138fea7 100644 --- a/src/main/ipc.ts +++ b/src/main/ipc.ts @@ -1571,30 +1571,30 @@ export function registerIpcHandlers( contextManager.remove(contextId) } } - const remoteConversation = - assistantDatabase.getOrCreateRemoteConversation({ - projectId: project.id, - channel, - accountId: 'default', - externalConversationId: message.conversationId, - conversationType: message.conversationType, - title: `${channelLabel} · ****${identitySuffix}`, - accountDisplay: senderDisplay, - runtimeSelection + try { + const remoteConversation = + assistantDatabase.getOrCreateRemoteConversation({ + projectId: project.id, + channel, + accountId: message.accountId, + externalConversationId: message.conversationId, + conversationType: message.conversationType, + title: `${channelLabel} · ****${identitySuffix}`, + accountDisplay: senderDisplay, + runtimeSelection + }) + assistantDatabase.appendRemoteConversationMessage({ + conversationId: remoteConversation.id, + role: 'user', + content: parsed.prompt, + attachments: publicAttachments, + status: `${channelLabel} · ${ + parsed.workMode === 'execute' + ? '执行' + : '对话' + }` }) - assistantDatabase.appendRemoteConversationMessage({ - conversationId: remoteConversation.id, - role: 'user', - content: parsed.prompt, - attachments: publicAttachments, - status: `${channelLabel} · ${ - parsed.workMode === 'execute' - ? '执行' - : '对话' - }` - }) - publishRemoteConversationChange() - + publishRemoteConversationChange() const remoteTaskId = randomUUID() assistantDatabase.createTask({ id: remoteTaskId, @@ -1658,7 +1658,6 @@ export function registerIpcHandlers( detail: unavailable, status: 'failed' }) - releaseRemoteContexts() return { status: 'failed', error: unavailable } } if ( @@ -1692,7 +1691,6 @@ export function registerIpcHandlers( detail: unavailable, status: 'failed' }) - releaseRemoteContexts() return { status: 'failed', error: unavailable } } } @@ -1781,8 +1779,10 @@ export function registerIpcHandlers( status: result.status === 'completed' ? 'completed' : 'failed' }) - releaseRemoteContexts() return result + } finally { + releaseRemoteContexts() + } } const channelManager = channelSettingsStore ? new ChannelManager(channelSettingsStore, channelExecutor, { diff --git a/src/shared/channel-contracts.ts b/src/shared/channel-contracts.ts index 8f27485..d26dada 100644 --- a/src/shared/channel-contracts.ts +++ b/src/shared/channel-contracts.ts @@ -105,6 +105,7 @@ export const channelInboundTextSchema = z .trim() .min(1) .max(CHANNEL_LIMITS.maximumChannelLength), + accountId: channelIdentifierSchema.default('default'), eventId: z .string() .trim()