diff --git a/changes/1031-lark-native-questions.md b/changes/1031-lark-native-questions.md new file mode 100644 index 0000000000..28308f99c2 --- /dev/null +++ b/changes/1031-lark-native-questions.md @@ -0,0 +1,11 @@ +--- +section: Added +--- + +## en + +- A paired Lark management DM can answer native Bot questions through a private form, a quoted reply or `/answer`, with the original session staying the single authority and uncertain submissions never replayed ([#1031](https://github.com/BotHarness/DeepSeekBot/issues/1031)). + +## zh + +- 已配对的 Lark 管理私聊可通过私密表单、引用回复或 `/answer` 回答 Bot 的原生提问;答案始终以原会话为准,结果未确认的提交不会重放([#1031](https://github.com/BotHarness/DeepSeekBot/issues/1031))。 diff --git a/docs/bot-cli.md b/docs/bot-cli.md index f6c5a9b423..c3b1254029 100644 --- a/docs/bot-cli.md +++ b/docs/bot-cli.md @@ -400,6 +400,8 @@ deepseekbot approval-routes deepseekbot approval-route --pairing deepseekbot approval-test deepseekbot approval-retry --delivery +deepseekbot question-deliveries +deepseekbot question-retry --delivery deepseekbot failure-alerts deepseekbot failure-alerts-set --enabled deepseekbot failure-alerts-test diff --git a/docs/bot-cli.zh.md b/docs/bot-cli.zh.md index 14572691b2..7144355c09 100644 --- a/docs/bot-cli.zh.md +++ b/docs/bot-cli.zh.md @@ -386,6 +386,8 @@ deepseekbot approval-routes deepseekbot approval-route --pairing deepseekbot approval-test deepseekbot approval-retry --delivery +deepseekbot question-deliveries +deepseekbot question-retry --delivery deepseekbot failure-alerts deepseekbot failure-alerts-set --enabled deepseekbot failure-alerts-test diff --git a/docs/dev/verification/lark-1031-native-questions.md b/docs/dev/verification/lark-1031-native-questions.md new file mode 100644 index 0000000000..5d7b3a31c5 --- /dev/null +++ b/docs/dev/verification/lark-1031-native-questions.md @@ -0,0 +1,24 @@ +# Lark native question qualification (#1031) + +## Candidate and scope + +A paired Lark management DM answers a native Bot question through a private form or an explicit text reply. The original session stays the one authority, and an unconfirmed submission is never replayed. + +Provider: `34b53227d4dc6a1e32059a7489e4e0fe280aeeb4`, 425 runtime files. Core: this branch on DSH `0.2.1-rc.3`, with `deepseek-v4.1-flash` driving real model turns. + +## Native qualification — 2026-11-03 JST + +An isolated Host connected the dedicated `BotHarness Lark QA1` app, which is visible only to the test account. The app gained `im:chat:read`, `im:resource` and the long-connection `card.action.trigger` callback in published version 1.0.1. A `/pair` request from the real DM was approved with only the `answer` capability, then selected as the management route. + +| Native test | Actual action in Lark | Result | +| -------------------------- | ------------------------------------------------------- | ----------------------------------------------------------------------------------- | +| Two-question form | Chose `Red` and typed free text for the second question | One resolution with the actor recorded, the Bot continued, card updated to Answered | +| Single question, `/answer` | Sent `/answer Tuesday` in the DM | One resolution, the Bot replied `LARK_TEXT_OK Tuesday`, card updated to Answered | + +Delivery records show `sent`, `accepted` and `updated` for both. Web shows the same answers, and the Management notifications panel lists both in light and dark. + +## Not covered natively + +Quoted-reply answers, cancelled or expired cards, a Web and Lark answer race, restart, and unconfirmed-submission repair have automated coverage only. The form's duplicated text in the first test came from the test typing it twice. + +The QA binding was removed and the Host stopped. Screenshots are local evidence in the task's verify folder. diff --git a/docs/lark-connection.md b/docs/lark-connection.md index 9c39b2d615..6523839de8 100644 --- a/docs/lark-connection.md +++ b/docs/lark-connection.md @@ -348,7 +348,7 @@ private-origin check is retained; a group pairing does not create a management d Removing a capability or revoking/reassigning the person refuses delayed decisions. If reassignment invalidates the route, explicitly select/save its new current destination again. Legacy `/pair` management grants remain independent and do not enable ordinary chat by themselves. -Native Questions, group execution-approval destinations and pairing-code issuance are separate work. +Group execution-approval destinations and pairing-code issuance are separate work. A management DM with the **Answer formal questions** capability receives native questions as a private form and can also answer with a quoted reply or `/answer `; the original session accepts one answer, and an unconfirmed submission is never replayed ([#1031](https://github.com/BotHarness/DeepSeekBot/issues/1031)). Qualification is recorded in [the verification note](dev/verification/lark-1031-native-questions.md). This slice supports Lark private **Allow once** and **Reject** cards through a qualified Provider. The #1033 source candidate adds the separately confirmed exact-rule path below. Group execution approvals, native question forms and non-blocking native waits are separate slices. The released dsh-im package number alone does not imply card capability; an unavailable Provider remains unavailable. diff --git a/packages/client/src/client/actions.ts b/packages/client/src/client/actions.ts index 13f4aaf064..9b0dd31e76 100644 --- a/packages/client/src/client/actions.ts +++ b/packages/client/src/client/actions.ts @@ -36,6 +36,7 @@ import { setApprovalRoute, testApprovalRoute, retryApprovalNotification, + retryQuestionNotification, setFailureAlerts, testFailureAlerts, retryFailureAlerts, @@ -475,6 +476,7 @@ export interface BridgeActions { approvalRoute(slug: string, pairingId: string | null, expectedRevision: number): Promise; approvalTest(slug: string): Promise; approvalRetry(slug: string, id: string): Promise; + questionRetry(slug: string, id: string): Promise; failureAlertsSet(slug: string, enabled: boolean, expectedRevision: number): Promise; failureAlertsTest(slug: string): Promise; failureAlertsRetry(slug: string, id: string): Promise; @@ -2894,6 +2896,7 @@ export function createActions( setApprovalRoute(call, slug, pairingId, expectedRevision), approvalTest: (slug) => testApprovalRoute(call, slug), approvalRetry: (slug, id) => retryApprovalNotification(call, slug, id), + questionRetry: (slug, id) => retryQuestionNotification(call, slug, id), failureAlertsSet: (slug, enabled, expectedRevision) => setFailureAlerts(call, slug, enabled, expectedRevision), failureAlertsTest: (slug) => testFailureAlerts(call, slug), diff --git a/packages/client/src/client/bridge.ts b/packages/client/src/client/bridge.ts index d68a741c72..f58c61a4ee 100644 --- a/packages/client/src/client/bridge.ts +++ b/packages/client/src/client/bridge.ts @@ -5947,6 +5947,14 @@ export async function retryApprovalNotification( await unwrap(call, 'approvalRetry', { slug, id }); } +export async function retryQuestionNotification( + call: BridgeCall, + slug: string, + id: string, +): Promise { + await unwrap(call, 'questionRetry', { slug, id }); +} + export async function setFailureAlerts( call: BridgeCall, slug: string, diff --git a/packages/client/src/client/external-entries.tsx b/packages/client/src/client/external-entries.tsx index d6571c0efe..05d0b91a02 100644 --- a/packages/client/src/client/external-entries.tsx +++ b/packages/client/src/client/external-entries.tsx @@ -1,3 +1,4 @@ +import { QuestionNotifications } from './question-notifications.js'; import { useState, type ReactElement } from 'react'; import { Button, Tag } from '@deepseek-ai/dsh-client-ui-primitives'; import { ReachablePostDialog } from './reachable-post.js'; @@ -232,6 +233,13 @@ function IdentitiesForBot({ retry: (id) => operate(() => actions.failureAlertsRetry(slug, id), 'approval'), }} /> + operate(() => actions.questionRetry(slug, id), 'approval')} + openSession={actions.openSession} + /> ) : null} diff --git a/packages/client/src/client/locale.ts b/packages/client/src/client/locale.ts index 8c90f9c49b..7db502775d 100644 --- a/packages/client/src/client/locale.ts +++ b/packages/client/src/client/locale.ts @@ -806,6 +806,21 @@ export const zh = { 'failureAlerts.state.failed': '确定未发送', 'failureAlerts.state.unknown-outcome': '结果未知', 'failureAlerts.state.cancelled': '已取消', + 'questionIm.title': 'Lark 提问通知', + 'questionIm.hint': + '同一管理私聊可回答原生提问,需要单独授予“回答”能力。原会话接收答案后才会显示已回答。', + 'questionIm.empty': '暂无提问通知。', + 'questionIm.reconcile': '核对原提问', + 'questionIm.repair': '修复通知', + 'questionIm.status.pending': '原提问等待回答', + 'questionIm.status.answered': '已回答', + 'questionIm.status.cancelled': '已取消', + 'questionIm.status.expired': '已失效', + 'questionIm.submission.none': '尚未提交答案', + 'questionIm.submission.submitting': '正在提交答案', + 'questionIm.submission.accepted': '原会话已接受答案', + 'questionIm.submission.refused': '答案未被接受,原提问继续等待', + 'questionIm.submission.unknown-outcome': '答案结果未确认;核对后再继续,系统不会重放答案。', 'approvalIm.title': '管理通知', 'approvalIm.configuration': '通知设置', 'approvalIm.platform': 'Lark 私聊', @@ -827,7 +842,7 @@ export const zh = { 'approvalIm.refresh': '刷新通知', 'approvalIm.destination': '管理私聊', 'approvalIm.off': '关闭自动通知', - 'approvalIm.empty': '先在私聊发起 /pair,再在 Web 授予批准或拒绝能力。', + 'approvalIm.empty': '先在私聊发起 /pair,再在 Web 授予批准、拒绝或回答能力。', 'approvalIm.save': '保存目的地', 'approvalIm.test': '发送测试卡片', 'approvalIm.once': @@ -5301,6 +5316,22 @@ export const en = { 'failureAlerts.state.failed': 'Known unsent', 'failureAlerts.state.unknown-outcome': 'Unknown outcome', 'failureAlerts.state.cancelled': 'Cancelled', + 'questionIm.title': 'Lark question notifications', + 'questionIm.hint': + 'The same management DM can answer native questions with a separate answer capability. Answered appears only after the original session accepts the answer.', + 'questionIm.empty': 'No question notifications yet.', + 'questionIm.reconcile': 'Check original question', + 'questionIm.repair': 'Repair notification', + 'questionIm.status.pending': 'Original question awaits an answer', + 'questionIm.status.answered': 'Answered', + 'questionIm.status.cancelled': 'Cancelled', + 'questionIm.status.expired': 'Expired', + 'questionIm.submission.none': 'No answer submitted', + 'questionIm.submission.submitting': 'Submitting answer', + 'questionIm.submission.accepted': 'Original session accepted the answer', + 'questionIm.submission.refused': 'Answer not accepted; original question still waits', + 'questionIm.submission.unknown-outcome': + 'Answer unconfirmed; check before continuing. The answer will not be replayed.', 'approvalIm.title': 'Management notifications', 'approvalIm.configuration': 'Notification settings', 'approvalIm.platform': 'Lark DM', @@ -5322,7 +5353,8 @@ export const en = { 'approvalIm.refresh': 'Refresh notifications', 'approvalIm.destination': 'Management DM', 'approvalIm.off': 'Automatic notifications off', - 'approvalIm.empty': 'Send /pair in the Bot DM, then grant approve or reject capability in Web.', + 'approvalIm.empty': + 'Send /pair in the Bot DM, then grant approve, reject or answer capability in Web.', 'approvalIm.save': 'Save destination', 'approvalIm.test': 'Send test card', 'approvalIm.once': diff --git a/packages/client/src/client/question-notifications.tsx b/packages/client/src/client/question-notifications.tsx new file mode 100644 index 0000000000..268437369e --- /dev/null +++ b/packages/client/src/client/question-notifications.tsx @@ -0,0 +1,70 @@ +import type { ReactElement } from 'react'; +import { Button, Tag } from '@deepseek-ai/dsh-client-ui-primitives'; +import type { QuestionDelivery } from '../../../core/src/messaging/question-messaging.js'; +import type { BotHarnessTranslate } from './locale.js'; + +export function QuestionNotifications({ + deliveries, + busy, + repair, + openSession, + t, +}: { + deliveries: QuestionDelivery[]; + busy: boolean; + repair(id: string): Promise; + openSession(id: string): void; + t: BotHarnessTranslate; +}): ReactElement { + return ( +
+ {t('questionIm.title')} +

{t('questionIm.hint')}

+ {deliveries.length === 0 ?

{t('questionIm.empty')}

: null} + {deliveries.map((delivery) => { + const canRepair = + (delivery.submission === 'unknown-outcome' && + ['pending', 'answered'].includes(delivery.status)) || + (delivery.delivery === 'failed' && + delivery.attempts < 3 && + delivery.status === 'pending') || + (delivery.receipt !== undefined && delivery.update === 'failed'); + return ( +
+
+ {delivery.reference} + {t(`questionIm.status.${delivery.status}`)} + {t(`approvalIm.delivery.${delivery.delivery}`)} +
+

{t(`questionIm.submission.${delivery.submission}`)}

+ {delivery.update === 'unknown-outcome' || delivery.update === 'failed' ? ( +

{t('approvalIm.updateUnconfirmed')}

+ ) : null} + {delivery.delivery === 'unknown-outcome' ? ( +

{t('approvalIm.unknownHint')}

+ ) : null} +
+ + {canRepair ? ( + + ) : null} +
+
+ ); + })} +
+ ); +} diff --git a/packages/client/test/bot-sidebar.test.ts b/packages/client/test/bot-sidebar.test.ts index eb7387c34b..58cbda6931 100644 --- a/packages/client/test/bot-sidebar.test.ts +++ b/packages/client/test/bot-sidebar.test.ts @@ -259,6 +259,7 @@ function stubActions(): BridgeActions { approvalRoute: vi.fn(async () => {}), approvalTest: vi.fn(async () => {}), approvalRetry: vi.fn(async () => {}), + questionRetry: vi.fn(), failureAlertsSet: vi.fn(async () => {}), failureAlertsTest: vi.fn(async () => {}), failureAlertsRetry: vi.fn(async () => {}), diff --git a/packages/core/src/bots/cli-messaging.ts b/packages/core/src/bots/cli-messaging.ts index 1170140b80..4b7f39863a 100644 --- a/packages/core/src/bots/cli-messaging.ts +++ b/packages/core/src/bots/cli-messaging.ts @@ -110,6 +110,14 @@ export const MESSAGING_COMMANDS = [ command: 'approval-retry', description: 'retry one failed IM approval delivery', }, + { + command: 'question-retry', + description: 'repair one IM question notification or unconfirmed answer', + }, + { + command: 'question-deliveries', + description: "read a Bot's IM question notifications", + }, { command: 'failure-alerts', description: "read a Bot's IM failure-alert setting and deliveries", @@ -270,6 +278,7 @@ export async function prepareMessaging( case 'channel-bridge': case 'approval-routes': case 'approval-test': + case 'question-deliveries': case 'failure-alerts': case 'failure-alerts-test': break; @@ -324,6 +333,7 @@ export async function prepareMessaging( if (values.pairing !== 'none') uuid(values.pairing, '--pairing'); break; case 'approval-retry': + case 'question-retry': case 'failure-alerts-retry': uuid(values.delivery, '--delivery'); break; @@ -528,6 +538,13 @@ export async function runMessaging(operation: MessagingOperation, rpc: Rpc): Pro slug: target, id: values.delivery!.trim(), }); + case 'question-retry': + return rpc('questionRetry', { + slug: target, + id: values.delivery!.trim(), + }); + case 'question-deliveries': + return { deliveries: (await snapshot())['questions'] ?? [] }; case 'failure-alerts': return record((await snapshot())['failureAlerts']); case 'failure-alerts-set': { diff --git a/packages/core/src/bridge/methods.ts b/packages/core/src/bridge/methods.ts index 4588b4b8d4..d50b63dbd3 100644 --- a/packages/core/src/bridge/methods.ts +++ b/packages/core/src/bridge/methods.ts @@ -654,6 +654,7 @@ export interface BridgeMethods { ): Promise< BridgeResult<{ delivery: import('../messaging/approval-messaging.js').ApprovalDelivery }> >; + questionRetry(payload: unknown): Promise>; approvalRetry( payload: unknown, ): Promise< @@ -2255,6 +2256,17 @@ export function createBridgeMethods(deps: BridgeMethodsDeps): BridgeMethods { delivery: await service.approvals.test(input.data.slug), })); }, + questionRetry(payload) { + const input = z + .object({ slug: z.string().min(1), id: z.string().uuid() }) + .strict() + .safeParse(payload); + if (!input.success) return Promise.resolve(invalidInput('Invalid question notification')); + return messagingCall(async (service) => { + await service.questions.retry(input.data.slug, input.data.id); + return { updated: true as const }; + }); + }, approvalRetry(payload) { const input = z .object({ slug: z.string().min(1), id: z.string().uuid() }) diff --git a/packages/core/src/bridge/rpc.ts b/packages/core/src/bridge/rpc.ts index 615a55539d..202b40c8db 100644 --- a/packages/core/src/bridge/rpc.ts +++ b/packages/core/src/bridge/rpc.ts @@ -511,6 +511,9 @@ export class BotharnessBridgeService extends TypertRemoteService { ): Promise<{ delivery: import('../messaging/approval-messaging.js').ApprovalDelivery }> { return unwrapAsync(this.methods.approvalTest({ slug })); } + questionRetry(slug: string, id: string): Promise<{ updated: true }> { + return unwrapAsync(this.methods.questionRetry({ slug, id })); + } approvalRetry( slug: string, id: string, @@ -2390,6 +2393,7 @@ markRemoteMethods(BotharnessBridgeService.prototype, [ 'approvalRoute', 'approvalTest', 'approvalRetry', + 'questionRetry', 'failureAlertsSet', 'failureAlertsTest', 'failureAlertsRetry', diff --git a/packages/core/src/channels/channel.ts b/packages/core/src/channels/channel.ts index d85ff19a2a..2742338636 100644 --- a/packages/core/src/channels/channel.ts +++ b/packages/core/src/channels/channel.ts @@ -929,6 +929,11 @@ export function isChannelMessage(value: unknown): value is ChannelMessage { if (questionResolution !== undefined) { if (typeof questionResolution !== 'object' || questionResolution === null) return false; const resolution = questionResolution as Record; + if ( + resolution['actor'] !== undefined && + (resolution['state'] !== 'answered' || !parseToolApprovalActor(resolution['actor'])) + ) + return false; if ( typeof resolution['requestMessageId'] !== 'string' || message['replyTo'] !== resolution['requestMessageId'] || diff --git a/packages/core/src/channels/store.ts b/packages/core/src/channels/store.ts index b997da4e04..df30e36350 100644 --- a/packages/core/src/channels/store.ts +++ b/packages/core/src/channels/store.ts @@ -453,6 +453,7 @@ export interface ChannelStore { id: string, message: ChannelMessage, origin?: ChannelMessageOrigin, + beforeCommit?: () => boolean, ): Promise; readMessages(id: string, options?: ChannelReadOptions): ChannelMessage[]; queryMessages(id: string, options?: ChannelMessageQueryOptions): ChannelMessageQueryPage; @@ -922,7 +923,7 @@ export function createChannelStore(options: ChannelStoreOptions): ChannelStore { write(renamed); return renamed; }, - appendMessage(id, message, origin) { + appendMessage(id, message, origin, beforeCommit) { return enqueue(id, () => { const record = read(id); if (record === undefined) return undefined; @@ -935,6 +936,7 @@ export function createChannelStore(options: ChannelStoreOptions): ChannelStore { } assertAttachmentRefs(message.attachments ?? []); assertGrantRequestReply(record, message, priorMessages); + if (beforeCommit && !beforeCommit()) return undefined; const durableMessage = { ...message }; delete durableMessage.grantRequestResolved; delete durableMessage.replyToPreview; diff --git a/packages/core/src/channels/user-questions.ts b/packages/core/src/channels/user-questions.ts index 5d27316e59..ec5bfd990e 100644 --- a/packages/core/src/channels/user-questions.ts +++ b/packages/core/src/channels/user-questions.ts @@ -12,6 +12,7 @@ import type { import { dmChannelId, type ChannelMessage } from './channel.js'; import type { ChannelStore } from './store.js'; import type { SessionOwnership } from '../sessions/ownership.js'; +import type { ToolApprovalActor } from '../workspaces/tool-approval.js'; export interface ChannelQuestionRequest { sessionId: string; @@ -51,6 +52,7 @@ export interface ChannelQuestionResolution { requestMessageId: string; state: 'answered' | 'cancelled'; answers?: AskUserQuestionAnswerItem[]; + actor?: ToolApprovalActor; } export interface ChannelQuestionNotice extends ChannelQuestionRequest { @@ -113,6 +115,9 @@ export class ChannelUserQuestions { readonly #changed: (slug: string, count: number) => void; readonly #pending = new Map(); readonly #listeners = new Set<() => void>(); + readonly #observers = new Set< + (notice: ChannelQuestionNotice, status: 'pending' | 'answered' | 'cancelled') => void + >(); readonly #timed: TimedQuestionPort | undefined; readonly #askChannel: ((sessionId: string) => string | undefined) | undefined; @@ -223,6 +228,8 @@ export class ChannelUserQuestions { pending.committed = true; if (this.status(pending.botSlug, message.id) === 'pending') this.#publishAttention(pending.botSlug); + if (this.#pending.get(message.id) === pending) + this.#observe(message.id, pending, 'pending'); } return await answer; } finally { @@ -276,6 +283,19 @@ export class ChannelUserQuestions { }; } + observe( + listener: (notice: ChannelQuestionNotice, status: 'pending' | 'answered' | 'cancelled') => void, + ): () => void { + this.#observers.add(listener); + return () => { + this.#observers.delete(listener); + }; + } + + pending(botSlug: string, messageId: string): ChannelQuestionNotice | undefined { + return this.requests(botSlug).find((notice) => notice.messageId === messageId); + } + activeSessionIds(): string[] { return this.activeMessageIds().flatMap((id) => { const pending = this.#pending.get(id); @@ -318,17 +338,26 @@ export class ChannelUserQuestions { botSlug: string, messageId: string, answer: AskUserQuestionAnswer, + external?: { actor: ToolApprovalActor; authorized(): boolean }, ): Promise { const pending = this.#pending.get(messageId); if ( pending === undefined || pending.botSlug !== botSlug || pending.deciding || - pending.submitted + pending.submitted || + (external !== undefined && !pending.committed) ) return false; + if (external && !external.authorized()) return false; if (!this.#holds(messageId, pending)) return false; if (!validAnswer(pending.questions, answer)) return false; + if ( + this.#channels + .readMessages(pending.channelId) + .some((message) => message.userQuestionResolution?.requestMessageId === messageId) + ) + return false; pending.deciding = true; try { if (pending.deferred && pending.callId !== undefined) { @@ -383,15 +412,21 @@ export class ChannelUserQuestions { requestMessageId: messageId, state: 'answered', answers: answer.answers, + ...(external ? { actor: external.actor } : {}), }, }; - if ((await this.#channels.appendMessage(pending.channelId, decision)) === undefined) + if ( + (await this.#channels.appendMessage(pending.channelId, decision, undefined, () => + external ? this.#pending.get(messageId) === pending && external.authorized() : true, + )) === undefined + ) return false; if (this.#pending.get(messageId) !== pending || pending.signal?.aborted) return false; this.#pending.delete(messageId); this.#publishAttention(pending.botSlug); pending.signal?.removeEventListener('abort', pending.abort); pending.resolve(answer); + this.#observe(messageId, pending, 'answered'); return true; } finally { pending.deciding = false; @@ -402,6 +437,35 @@ export class ChannelUserQuestions { } } + reconcile(botSlug: string, messageId: string): boolean { + const pending = this.#pending.get(messageId); + if ( + !pending?.committed || + pending.deciding || + pending.deferred || + pending.botSlug !== botSlug || + this.status(botSlug, messageId, pending.channelId) !== 'pending' + ) + return false; + const resolution = this.#channels + .readMessages(pending.channelId) + .find( + (message) => message.userQuestionResolution?.requestMessageId === messageId, + )?.userQuestionResolution; + if ( + resolution?.state !== 'answered' || + !resolution.answers || + !validAnswer(pending.questions, { answers: resolution.answers }) + ) + return false; + this.#pending.delete(messageId); + this.#publishAttention(botSlug); + pending.signal?.removeEventListener('abort', pending.abort); + pending.resolve({ answers: resolution.answers }); + this.#observe(messageId, pending, 'answered'); + return true; + } + cancelSession(sessionId: string): void { for (const [id, pending] of this.#pending) { if (pending.sessionId !== sessionId) continue; @@ -568,6 +632,31 @@ export class ChannelUserQuestions { return true; } + #observe( + messageId: string, + pending: Pending, + status: 'pending' | 'answered' | 'cancelled', + ): void { + for (const listener of this.#observers) { + try { + listener( + { + botSlug: pending.botSlug, + channelId: pending.channelId, + ...(pending.groupName === undefined ? {} : { groupName: pending.groupName }), + messageId, + sessionId: pending.sessionId, + ...(pending.callId === undefined ? {} : { callId: pending.callId }), + questions: pending.questions, + }, + status, + ); + } catch { + this.#warn('user-question-notification-failed'); + } + } + } + cancelChannel(channelId: string): void { for (const [messageId, pending] of this.#pending) if (pending.channelId === channelId) this.#cancel(messageId, false); @@ -596,6 +685,7 @@ export class ChannelUserQuestions { (saved) => { if (saved === undefined) this.#warn(`botharness.channel_question.cancel_append_failed request=${messageId}`); + else this.#observe(messageId, pending, 'cancelled'); }, () => this.#warn(`botharness.channel_question.cancel_append_failed request=${messageId}`), ); diff --git a/packages/core/src/database/schema-plan.ts b/packages/core/src/database/schema-plan.ts index 5ba72fd0ad..75a1d7760f 100644 --- a/packages/core/src/database/schema-plan.ts +++ b/packages/core/src/database/schema-plan.ts @@ -2618,6 +2618,20 @@ const BOT_LAN_ADMISSION_MIGRATION: SchemaMigration = { }, }; +const QUESTION_MESSAGING_MIGRATION: SchemaMigration = { + generation: 116, + module: 'messaging', + description: 'Retain checked native question notification references and receipts', + migrate(database) { + database.exec(` + CREATE TABLE messaging_question_deliveries ( + id TEXT PRIMARY KEY, bot_slug TEXT NOT NULL, body TEXT NOT NULL CHECK(json_valid(body)) + ); + CREATE INDEX messaging_question_deliveries_bot ON messaging_question_deliveries(bot_slug); + `); + }, +}; + export const BOT_HARNESS_SCHEMA_PLAN = defineSchemaPlan([ SESSION_OWNERSHIP_MIGRATION, MESSAGING_TRACER_MIGRATION, @@ -2733,4 +2747,5 @@ export const BOT_HARNESS_SCHEMA_PLAN = defineSchemaPlan([ GRANT_RECEPTION_EVIDENCE_MIGRATION, HUMAN_ATTENTION_CANDIDATE_INDEX_MIGRATION, BOT_LAN_ADMISSION_MIGRATION, + QUESTION_MESSAGING_MIGRATION, ]); diff --git a/packages/core/src/messaging/approval-messaging.ts b/packages/core/src/messaging/approval-messaging.ts index 1d79f29163..0a0e62082e 100644 --- a/packages/core/src/messaging/approval-messaging.ts +++ b/packages/core/src/messaging/approval-messaging.ts @@ -174,7 +174,7 @@ export function createApprovalMessaging(options: { paired.revision !== value.pairingRevision || paired.status !== 'approved' || (paired.purpose === 'conversation' && paired.conversationKind !== 'dm') || - !paired.capabilities.some((c) => c === 'approve' || c === 'reject') + !paired.capabilities.some((c) => c === 'approve' || c === 'reject' || c === 'answer') ) throw new MessagingError('approval-route-unavailable'); const provider = options.provider(identity.providerId); @@ -551,6 +551,14 @@ export function createApprovalMessaging(options: { if (status === 'pending') { if (read(notice.messageId)) return; const selected = route(notice.botSlug); + if ( + selected && + !pairing + .list(notice.botSlug) + .find((value) => value.id === selected.pairingId) + ?.capabilities.some((c) => c === 'approve' || c === 'reject') + ) + return; if (!selected) return; write({ id: notice.messageId, @@ -626,7 +634,7 @@ export function createApprovalMessaging(options: { .filter( (p) => p.status === 'approved' && - p.capabilities.some((c) => c === 'approve' || c === 'reject'), + p.capabilities.some((c) => c === 'approve' || c === 'reject' || c === 'answer'), ) .map((p) => { let ready = false; diff --git a/packages/core/src/messaging/dsh-im.ts b/packages/core/src/messaging/dsh-im.ts index 03635ac663..d454be33e8 100644 --- a/packages/core/src/messaging/dsh-im.ts +++ b/packages/core/src/messaging/dsh-im.ts @@ -9,6 +9,9 @@ import type { MessagingHistoryQuery, MessagingHistoryPage, MessagingApprovalCard, + MessagingCardAction, + MessagingQuestionAction, + MessagingQuestionCard, MessagingPairingCard, MessagingApprovalAck, MessagingTypingState, @@ -36,6 +39,18 @@ export interface DshImOutboundService { options: { expectedFingerprint: string; signal: AbortSignal; beforeSend(): boolean }, ): Promise<{ accepted: true }>; approvalCardVersion?: 1; + questionCardVersion?: 1; + questionCardChecked?( + botId: string, + route: MessagingReplyRoute | MessagingReceipt, + card: MessagingQuestionCard, + options: { + expectedFingerprint: string; + signal: AbortSignal; + beforeSend(): boolean; + update?: boolean; + }, + ): Promise<{ sent?: true; updated?: true; receipt?: MessagingReceipt }>; approvalRuleCardVersion?: 1 | 2; pairingCardVersion?: 1; approvalCardChecked?( @@ -685,6 +700,55 @@ export function createDshImProvider( host.approvalRuleCardVersion === 2 && allOpaqueAccounts.get(ref) === fingerprint, } : {}), + ...(platform === 'feishu' && + host.questionCardVersion === 1 && + typeof host.questionCardChecked === 'function' + ? { + async questionCard(input: Parameters>[0]) { + input.signal.throwIfAborted(); + const info = await host.describeBot(input.accountRef); + if (info.account.fingerprint !== input.fingerprint) + throw new MessagingError('rebind-required'); + if ( + !info.connected || + ![ + 'question-card-checked', + 'question-card-update-checked', + 'question-action-consumer', + ].every((cap) => info.capabilities.includes(cap)) + ) + throw new MessagingProviderError('capability-unavailable', 'not-started'); + try { + const result = await host.questionCardChecked!( + input.accountRef, + input.route, + input.card, + { + expectedFingerprint: input.fingerprint, + signal: input.signal, + beforeSend: input.beforeSend, + ...(input.update ? { update: true } : {}), + }, + ); + if (input.update) { + if (result.updated !== true) + throw new MessagingProviderError('provider-result-unknown', 'unknown'); + return { updated: true as const }; + } + if ( + result.sent !== true || + result.receipt?.version !== 1 || + !result.receipt.messageId || + result.receipt.conversationId !== input.route.conversationId + ) + throw new MessagingProviderError('provider-result-unknown', 'unknown'); + return { sent: true as const, receipt: result.receipt }; + } catch (error) { + throw providerFailure(error); + } + }, + } + : {}), ...(host.replyContextVersion === 1 && host.replyReceiptVersion === 1 && host.replyFenceVersion === 1 && @@ -882,7 +946,21 @@ export function createDshImProvider( conversationId: identifier, messageId: identifier, requestId: z.string().uuid(), - action: z.enum(['allowed-once', 'rejected']), + action: z.enum(['allowed-once', 'rejected', 'answer']), + values: z + .array( + z + .object({ + selected: z + .array(z.number().int().min(0).max(99)) + .max(100), + custom: z.string().min(1).max(2000).optional(), + }) + .strict(), + ) + .min(1) + .max(3) + .optional(), }) .strict() ).parse(raw); @@ -904,7 +982,27 @@ export function createDshImProvider( event.fingerprint !== input.fingerprint ) throw new MessagingError('untrusted-source'); - return input.onAction!(event, context.signal); + if ('values' in event || event.action === 'answer') { + const { values } = event as { values?: MessagingQuestionAction['values'] }; + if ( + event.action !== 'answer' || + !values || + host.questionCardVersion !== 1 || + !info.capabilities.includes('question-action-consumer') + ) + throw new MessagingError('untrusted-source'); + return input.onAction!( + { + ...(event as Omit), + values: values.map(({ selected, custom }) => ({ + selected, + ...(custom === undefined ? {} : { custom }), + })), + }, + context.signal, + ); + } + return input.onAction!(event as MessagingCardAction, context.signal); }, } : {}), diff --git a/packages/core/src/messaging/inbound.ts b/packages/core/src/messaging/inbound.ts index b73a6c56fd..e5ef7c1e3a 100644 --- a/packages/core/src/messaging/inbound.ts +++ b/packages/core/src/messaging/inbound.ts @@ -239,9 +239,22 @@ export interface InboundMessaging { } export function createInboundMessaging(options: { + onQuestionAction?( + providerId: string, + event: import('./provider.js').MessagingQuestionAction, + signal: AbortSignal, + ): Promise; + onQuestionText?( + providerId: string, + event: MessagingInboundEvent, + signal: AbortSignal, + ): Promise; onApprovalAction?( providerId: string, - event: import('./provider.js').MessagingCardAction, + event: Exclude< + import('./provider.js').MessagingCardAction, + import('./provider.js').MessagingQuestionAction + >, signal: AbortSignal, ): Promise; database: OperationalDatabaseModulePort; @@ -1833,6 +1846,7 @@ export function createInboundMessaging(options: { event: MessagingInboundEvent, signal: AbortSignal, ) => { + if (await options.onQuestionText?.(provider.id, event, signal)) return true; if (!options.pairing || !chatPairingApplies(event) || provider.id !== `dsh-im/${event.channel}`) return false; signal.throwIfAborted(); @@ -2215,8 +2229,13 @@ export function createInboundMessaging(options: { async (event, signal) => (await controlIntake(provider, event, signal)) || (await commandIntake(provider, event, signal)), - options.onApprovalAction - ? (event, signal) => options.onApprovalAction!(provider.id, event, signal) + options.onApprovalAction || options.onQuestionAction + ? (event, signal) => + event.action === 'answer' + ? (options.onQuestionAction?.(provider.id, event, signal) ?? + Promise.resolve({ accepted: true as const, status: 'refused' as const })) + : (options.onApprovalAction?.(provider.id, event, signal) ?? + Promise.resolve({ accepted: true as const, status: 'refused' as const })) : undefined, ), } diff --git a/packages/core/src/messaging/outbound.ts b/packages/core/src/messaging/outbound.ts index 5d86bf9ca4..25b8988d23 100644 --- a/packages/core/src/messaging/outbound.ts +++ b/packages/core/src/messaging/outbound.ts @@ -17,6 +17,11 @@ import { import { externalPermissionGroups, type ExternalPermissionGroup } from './permission-groups.js'; import { behaviorPermissions, type BehaviorPermission } from './behavior-permissions.js'; import { createPairingCards } from './pairing-cards.js'; +import { + createQuestionMessaging, + type QuestionDelivery, + type QuestionMessaging, +} from './question-messaging.js'; import { createMessagingDirectory, type MessagingDirectoryQuery, @@ -236,6 +241,7 @@ export type MessagingApp = MessagingAccount & { }; export interface MessagingSnapshot { + questions?: QuestionDelivery[]; permissionGroups?: ExternalPermissionGroup[]; behaviorPermissions?: BehaviorPermission[]; senderPolicy?: SenderPolicy; @@ -281,6 +287,7 @@ export interface MessagingSnapshot { } export interface OutboundMessaging { + questions: QuestionMessaging; privateInputReplyAuthority( botSlug: string, sourceEventId: string, @@ -1103,7 +1110,18 @@ export function createOutboundMessaging(options: { ...(options.now === undefined ? {} : { now: options.now }), ...(options.warn === undefined ? {} : { warn: options.warn }), }); + const questions = createQuestionMessaging({ + database, + pairing, + approvals, + provider: (id) => provider(id).provider, + isBotActive: options.isBotActive, + ...(options.recover === undefined ? {} : { recover: options.recover }), + ...(options.warn ? { warn: options.warn } : {}), + }); const inbound = createInboundMessaging({ + onQuestionAction: (providerId, event, signal) => questions.action(providerId, event, signal), + onQuestionText: (providerId, event, signal) => questions.text(providerId, event, signal), onApprovalAction: (providerId, event, signal) => event.kind === 'pairing' ? pairingCards.action(providerId, event, signal) @@ -1384,6 +1402,7 @@ export function createOutboundMessaging(options: { } const service: OutboundMessaging = { + questions, privateInputReplyAuthority(botSlug, sourceEventId, authority) { const key = privateReplyKey(botSlug, sourceEventId); let authorities = privateReplies.get(key); @@ -2582,6 +2601,7 @@ export function createOutboundMessaging(options: { senderPolicy: database.read((db) => readSenderPolicy(db, botSlug)), pairings: pairing.list(botSlug), approvals: approvals.snapshot(botSlug), + questions: questions.snapshot(botSlug), failureAlerts: failureAlerts.snapshot(botSlug), pairingReceivers: identities .filter((i) => ['feishu', 'slack'].includes(i.platform) && !i.revokedAt) @@ -3323,6 +3343,7 @@ export function createOutboundMessaging(options: { } finally { closed = true; approvals.close(); + questions.close(); failureAlerts.close(); pairingCards.close(); feedback.close(); diff --git a/packages/core/src/messaging/provider.ts b/packages/core/src/messaging/provider.ts index b162fea60b..6147b3eb69 100644 --- a/packages/core/src/messaging/provider.ts +++ b/packages/core/src/messaging/provider.ts @@ -219,7 +219,8 @@ export interface MessagingPairingAction { export type MessagingCardAction = | MessagingApprovalAction | MessagingApprovalRuleAction - | MessagingPairingAction; + | MessagingPairingAction + | MessagingQuestionAction; export interface MessagingApprovalAck { accepted: true; @@ -233,6 +234,27 @@ export interface MessagingSetup { kind: 'credentials' | 'qr'; } +export interface MessagingQuestionCard { + requestId: string; + reference: string; + questions: import('@deepseek-ai/dsh-user-questions/types').AskUserQuestionItem[]; + status: 'pending' | 'answered' | 'cancelled' | 'expired' | 'web-required'; + detail: string; +} + +export interface MessagingQuestionAction { + version: 1; + channel: 'feishu'; + botId: string; + fingerprint: string; + actorId: string; + conversationId: string; + messageId: string; + requestId: string; + action: 'answer'; + values: { selected: number[]; custom?: string }[]; +} + export interface MessagingProvider { approvalRuleCardSupported?: boolean; approvalRuleAllOpaqueCardSupported?(accountRef: string, fingerprint: string): boolean; @@ -273,6 +295,15 @@ export interface MessagingProvider { beforeSend(): boolean; update?: boolean; }): Promise<{ sent?: true; updated?: true; receipt?: MessagingReceipt }>; + questionCard?(input: { + accountRef: string; + fingerprint: string; + route: MessagingReplyRoute | MessagingReceipt; + card: MessagingQuestionCard; + signal: AbortSignal; + beforeSend(): boolean; + update?: boolean; + }): Promise<{ sent?: true; updated?: true; receipt?: MessagingReceipt }>; qualifyReply?(input: { accountRef: string; fingerprint: string; diff --git a/packages/core/src/messaging/question-messaging.ts b/packages/core/src/messaging/question-messaging.ts new file mode 100644 index 0000000000..36ba23a73b --- /dev/null +++ b/packages/core/src/messaging/question-messaging.ts @@ -0,0 +1,598 @@ +import { createHash } from 'node:crypto'; +import type { AskUserQuestionAnswer } from '@deepseek-ai/dsh-user-questions/types'; +import type { ChannelStore } from '../channels/store.js'; +import type { ChannelQuestionNotice, ChannelUserQuestions } from '../channels/user-questions.js'; +import type { OperationalDatabaseModulePort } from '../database/owner.js'; +import type { ToolApprovalActor } from '../workspaces/tool-approval.js'; +import type { ApprovalMessaging, ApprovalRoute } from './approval-messaging.js'; +import { assertMessagingIdentity } from './identity.js'; +import type { BotPairing } from './pairing.js'; +import { + MessagingError, + MessagingProviderError, + type MessagingInboundEvent, + type MessagingProvider, + type MessagingQuestionAction, + type MessagingQuestionCard, + type MessagingReceipt, +} from './provider.js'; + +export interface QuestionDelivery { + id: string; + botSlug: string; + channelId: string; + sessionId: string; + reference: string; + questionHash: string; + route: ApprovalRoute; + createdAt: string; + status: 'pending' | 'answered' | 'cancelled' | 'expired'; + delivery: 'pending' | 'sending' | 'sent' | 'failed' | 'unknown-outcome'; + attempts: number; + receipt?: MessagingReceipt; + submission: 'none' | 'submitting' | 'accepted' | 'refused' | 'unknown-outcome'; + update: 'none' | 'sending' | 'updated' | 'failed' | 'unknown-outcome'; + reason?: string; +} + +export interface QuestionMessaging { + attach(owner: ChannelUserQuestions, channels: ChannelStore): () => void; + refresh(): void; + snapshot(botSlug: string): QuestionDelivery[]; + retry(botSlug: string, id: string): Promise; + action( + providerId: string, + event: MessagingQuestionAction, + signal: AbortSignal, + ): Promise<{ accepted: true; status: 'queued' | 'refused' }>; + text(providerId: string, event: MessagingInboundEvent, signal: AbortSignal): Promise; + close(): void; +} + +export function createQuestionMessaging(options: { + database: OperationalDatabaseModulePort; + pairing: BotPairing; + approvals: ApprovalMessaging; + provider(id: string): MessagingProvider; + isBotActive(slug: string): boolean; + recover?: boolean; + warn?(message: string): void; +}): QuestionMessaging { + const { database, pairing } = options; + let owner: ChannelUserQuestions | undefined; + let channels: ChannelStore | undefined; + let detach: (() => void) | undefined; + let closed = false; + const controller = new AbortController(); + const busy = new Set(); + const rows = (slug: string): QuestionDelivery[] => + database.read((db) => + ( + db + .prepare( + 'SELECT body FROM messaging_question_deliveries WHERE bot_slug = ? ORDER BY rowid DESC LIMIT 50', + ) + .all(slug) as { body: string }[] + ).map((row) => JSON.parse(row.body) as QuestionDelivery), + ); + const read = (id: string): QuestionDelivery | undefined => + closed + ? undefined + : database.read((db) => { + const row = db + .prepare('SELECT body FROM messaging_question_deliveries WHERE id = ?') + .get(id) as { body: string } | undefined; + return row ? (JSON.parse(row.body) as QuestionDelivery) : undefined; + }); + const write = (value: QuestionDelivery) => { + if (closed) return; + database.transaction( + (db) => + db + .prepare( + 'INSERT INTO messaging_question_deliveries(id, bot_slug, body) VALUES(?, ?, ?) ON CONFLICT(id) DO UPDATE SET body = excluded.body', + ) + .run(value.id, value.botSlug, JSON.stringify(value)), + ['question-messaging'], + ); + }; + const request = (value: QuestionDelivery) => { + const question = channels?.message(value.channelId, value.id)?.userQuestionRequest; + return question?.sessionId === value.sessionId && + createHash('sha256').update(JSON.stringify(question.questions)).digest('hex') === + value.questionHash + ? question + : undefined; + }; + const resolution = (value: QuestionDelivery) => + channels + ?.readMessages(value.channelId) + .find((message) => message.userQuestionResolution?.requestMessageId === value.id) + ?.userQuestionResolution; + const context = (value: QuestionDelivery) => { + if (closed || !options.isBotActive(value.botSlug)) throw new MessagingError('bot-unavailable'); + const selected = options.approvals.snapshot(value.botSlug).route; + if ( + !selected || + selected.revision !== value.route.revision || + selected.pairingId !== value.route.pairingId + ) + throw new MessagingError('approval-route-stale'); + const identity = database.read((db) => assertMessagingIdentity(db, value.route.bindingId)); + if (identity.botSlug !== value.botSlug || identity.platform !== 'feishu') + throw new MessagingError('untrusted-source'); + const paired = pairing.list(value.botSlug).find((item) => item.id === value.route.pairingId); + if ( + !paired || + paired.status !== 'approved' || + paired.bindingId !== identity.id || + paired.revision !== value.route.pairingRevision || + !paired.capabilities.includes('answer') + ) + throw new MessagingError('question-answer-unavailable'); + const provider = options.provider(identity.providerId); + if (!provider.questionCard) throw new MessagingError('capability-unavailable'); + return { identity, paired, provider }; + }; + const safe = (value: QuestionDelivery) => { + try { + context(value); + return !!request(value); + } catch { + return false; + } + }; + const live = (value: QuestionDelivery) => + !!owner?.pending(value.botSlug, value.id) && !!request(value); + const state = (value: QuestionDelivery): QuestionDelivery['status'] => + resolution(value)?.state ?? + (value.submission === 'accepted' + ? 'answered' + : value.status === 'pending' && !live(value) + ? 'expired' + : value.status); + const render = (value: QuestionDelivery): MessagingQuestionCard => { + const status = state(value); + const answers = resolution(value)?.answers; + const detail = answers + ? answers.map((answer) => answer.custom ?? answer.selected.join(', ')).join('\n') + : value.submission === 'unknown-outcome' + ? 'Answer not confirmed; inspect and reconcile in Web / 回答结果未确认,请在 Web 核对' + : value.submission === 'refused' + ? 'Answer not accepted; the original question still waits / 回答未被接受,原提问继续等待' + : value.submission === 'submitting' + ? 'Checking the native question owner / 正在核验原生提问' + : `Choose answers or enter your own, then Submit / 选择或填写答案后提交。\nFor a single question you may reply to this card, or send /answer ${value.reference} your answer. / 单个问题也可回复此卡片,或使用 /answer ${value.reference} 答案。`; + return { + requestId: value.id, + reference: value.reference, + questions: request(value)?.questions ?? [], + status: value.submission === 'unknown-outcome' && live(value) ? 'web-required' : status, + detail, + }; + }; + const trace = (phase: string, value: QuestionDelivery, startedAt: number) => + options.warn?.( + `question-notification ${JSON.stringify({ + initiator: 'native-question', + phase, + outcome: value.delivery, + submission: value.submission, + update: value.update, + durationMs: Math.round(performance.now() - startedAt), + reason: value.reason ?? 'none', + })}`, + ); + const bounded = (run: (signal: AbortSignal) => Promise): Promise => { + const signal = AbortSignal.any([controller.signal, AbortSignal.timeout(15000)]); + return Promise.race([ + run(signal), + new Promise((_, reject) => + signal.addEventListener( + 'abort', + () => reject(new MessagingProviderError('question-result-unknown', 'unknown')), + { once: true }, + ), + ), + ]); + }; + const update = async (id: string): Promise => { + const value = read(id); + if (!value?.receipt || busy.has(id) || !safe(value)) return; + const startedAt = performance.now(); + const card = render(value); + busy.add(id); + write({ ...value, status: state(value), update: 'sending' }); + try { + const { identity, provider } = context(value); + const result = await bounded((signal) => + provider.questionCard!({ + accountRef: identity.accountRef, + fingerprint: identity.fingerprint, + route: value.receipt!, + card, + signal, + update: true, + beforeSend: () => safe(value) && render(read(id)!).status === card.status, + }), + ); + if (result.updated !== true) + throw new MessagingProviderError('provider-result-unknown', 'unknown'); + const latest = read(id); + if (latest) write({ ...latest, update: 'updated' }); + } catch (error) { + const latest = read(id); + if (latest) + write({ + ...latest, + update: + error instanceof MessagingProviderError && error.disposition === 'not-started' + ? 'failed' + : 'unknown-outcome', + }); + } finally { + busy.delete(id); + const latest = read(id); + if (latest) { + trace('update', latest, startedAt); + if (render(latest).status !== card.status) void update(id); + } + } + }; + const send = async (id: string): Promise => { + const value = read(id); + if ( + !value || + busy.has(id) || + value.attempts >= 3 || + !['pending', 'failed'].includes(value.delivery) + ) + return; + const startedAt = performance.now(); + busy.add(id); + let started = false; + write({ ...value, attempts: value.attempts + 1 }); + try { + const { identity, paired, provider } = context(value); + if (!live(value) || !paired.messageIds?.[0]) + throw new MessagingProviderError('question-owner-unavailable', 'not-started'); + write({ ...value, attempts: value.attempts + 1, delivery: 'sending' }); + started = true; + const result = await bounded((signal) => + provider.questionCard!({ + accountRef: identity.accountRef, + fingerprint: identity.fingerprint, + route: { + messageId: paired.messageIds![0]!, + actorId: paired.actorId, + conversationId: paired.conversationId, + }, + card: render(value), + signal, + beforeSend: () => safe(value) && live(value), + }), + ); + if ( + result.sent !== true || + result.receipt?.version !== 1 || + result.receipt.conversationId !== paired.conversationId + ) + throw new MessagingProviderError('provider-result-unknown', 'unknown'); + const latest = read(id); + if (latest) write({ ...latest, delivery: 'sent', receipt: result.receipt }); + } catch (error) { + const latest = read(id); + if (latest) + write({ + ...latest, + delivery: + !started || + (error instanceof MessagingProviderError && error.disposition === 'not-started') + ? 'failed' + : 'unknown-outcome', + reason: + error instanceof MessagingError || error instanceof MessagingProviderError + ? error.code + : 'provider-result-unknown', + }); + } finally { + busy.delete(id); + const latest = read(id); + if (latest) { + trace('send', latest, startedAt); + if (latest.receipt && state(latest) !== 'pending') void update(id); + } + } + }; + const notify = (notice: ChannelQuestionNotice, status: 'pending' | 'answered' | 'cancelled') => { + if (closed) return; + if (status === 'pending') { + if (read(notice.messageId)) return; + const route = options.approvals.snapshot(notice.botSlug).route; + const question = channels?.message(notice.channelId, notice.messageId)?.userQuestionRequest; + if (!route || !question) return; + write({ + id: notice.messageId, + botSlug: notice.botSlug, + channelId: notice.channelId, + sessionId: notice.sessionId, + reference: notice.messageId.replaceAll('-', '').slice(0, 12).toUpperCase(), + questionHash: createHash('sha256').update(JSON.stringify(question.questions)).digest('hex'), + route, + createdAt: new Date().toISOString(), + status: 'pending', + delivery: 'pending', + attempts: 0, + submission: 'none', + update: 'none', + }); + void send(notice.messageId).catch(() => options.warn?.('question-notification-send-failed')); + } else { + const value = read(notice.messageId); + if (!value) return; + write({ + ...value, + status, + submission: status === 'answered' ? 'accepted' : value.submission, + }); + void update(value.id); + } + }; + const authorized = ( + value: QuestionDelivery, + providerId: string, + event: { botId: string; fingerprint: string; actorId: string; conversationId: string }, + ) => { + try { + const { identity, paired } = context(value); + const current = pairing.assert(value.botSlug, identity.id, event.actorId, 'answer'); + return ( + safe(value) && + live(value) && + identity.providerId === providerId && + identity.accountRef === event.botId && + identity.fingerprint === event.fingerprint && + paired.id === current.id && + current.revision === value.route.pairingRevision && + paired.conversationId === event.conversationId + ); + } catch { + return false; + } + }; + const submit = ( + value: QuestionDelivery, + answer: AskUserQuestionAnswer, + providerId: string, + event: { + botId: string; + fingerprint: string; + actorId: string; + conversationId: string; + messageId: string; + }, + signal: AbortSignal, + ) => { + if ( + signal.aborted || + value.status !== 'pending' || + ['submitting', 'unknown-outcome'].includes(value.submission) || + value.delivery !== 'sent' || + !authorized(value, providerId, event) + ) + return false; + const actor: ToolApprovalActor = { + platform: 'feishu', + bindingId: value.route.bindingId, + fingerprint: event.fingerprint, + actorId: event.actorId, + pairingId: value.route.pairingId, + pairingRevision: value.route.pairingRevision, + conversationId: event.conversationId, + messageId: event.messageId, + }; + write({ ...value, submission: 'submitting' }); + queueMicrotask(() => { + void owner + ?.answer(value.botSlug, value.id, answer, { + actor, + authorized: () => !signal.aborted && authorized(value, providerId, event), + }) + .then((accepted) => { + const latest = read(value.id); + if (latest) + write({ + ...latest, + status: state(latest), + submission: accepted ? 'accepted' : 'refused', + }); + void update(value.id); + }) + .catch(() => { + const latest = read(value.id); + if (latest) + write({ + ...latest, + submission: 'unknown-outcome', + reason: 'question-answer-unconfirmed', + }); + void update(value.id); + }); + }); + return true; + }; + if (options.recover !== false) + database.transaction((db) => + db + .prepare( + "UPDATE messaging_question_deliveries SET body = json_set(body, '$.status', CASE WHEN json_extract(body, '$.status') = 'pending' THEN 'expired' ELSE json_extract(body, '$.status') END, '$.delivery', CASE WHEN json_extract(body, '$.delivery') = 'sending' THEN 'unknown-outcome' ELSE json_extract(body, '$.delivery') END, '$.submission', CASE WHEN json_extract(body, '$.submission') = 'submitting' THEN 'unknown-outcome' ELSE json_extract(body, '$.submission') END, '$.update', CASE WHEN json_extract(body, '$.status') = 'pending' AND json_extract(body, '$.receipt') IS NOT NULL THEN 'none' WHEN json_extract(body, '$.update') = 'sending' THEN 'unknown-outcome' ELSE json_extract(body, '$.update') END)", + ) + .run(), + ); + const service: QuestionMessaging = { + attach(nextOwner, nextChannels) { + detach?.(); + owner = nextOwner; + channels = nextChannels; + const unsubscribe = owner.observe(notify); + detach = unsubscribe; + service.refresh(); + return () => { + unsubscribe(); + if (detach === unsubscribe) { + owner = undefined; + channels = undefined; + detach = undefined; + } + }; + }, + refresh() { + if (closed || !channels) return; + const pendingUpdates = database.read((db) => + ( + db + .prepare( + "SELECT body FROM messaging_question_deliveries WHERE json_extract(body, '$.receipt') IS NOT NULL AND json_extract(body, '$.status') != 'pending' AND json_extract(body, '$.update') IN ('none', 'failed') ORDER BY rowid DESC LIMIT 50", + ) + .all() as { body: string }[] + ).map((row) => JSON.parse(row.body) as QuestionDelivery), + ); + for (const value of pendingUpdates) void update(value.id); + }, + snapshot(botSlug) { + return closed ? [] : rows(botSlug).map((value) => ({ ...value, status: state(value) })); + }, + async retry(botSlug, id) { + const value = read(id); + if (!value || value.botSlug !== botSlug || !safe(value)) + throw new MessagingError('question-repair-unavailable'); + if (value.submission === 'unknown-outcome') { + if (owner?.reconcile(botSlug, id)) return; + if (!live(value) || resolution(value)) + throw new MessagingError('question-owner-unavailable'); + write({ ...value, submission: 'refused', reason: 'original-question-still-pending' }); + await update(id); + } else if (value.delivery === 'failed' && live(value) && value.attempts < 3) await send(id); + else if (value.receipt && value.update === 'failed') await update(id); + else throw new MessagingError('question-repair-unavailable'); + }, + async action(providerId, event, signal) { + const value = read(event.requestId); + const question = value && request(value); + if ( + !value || + !question || + value.receipt?.messageId !== event.messageId || + value.receipt.conversationId !== event.conversationId || + event.values.length !== question.questions.length || + event.values.some((answer, index) => + answer.selected.some( + (pick) => + !Number.isSafeInteger(pick) || + pick < 0 || + pick >= (question.questions[index]!.options?.length ?? 0), + ), + ) + ) + return { accepted: true, status: 'refused' }; + const answers = event.values.map((answer, index) => ({ + id: question.questions[index]!.id, + selected: answer.selected.map((pick) => question.questions[index]!.options![pick]!.label), + ...(answer.custom === undefined ? {} : { custom: answer.custom }), + })); + return { + accepted: true, + status: submit(value, { answers }, providerId, event, signal) ? 'queued' : 'refused', + }; + }, + async text(providerId, event, signal) { + if (closed) return false; + const command = /^\/answer(?:\s|$)/.test(event.text.trim()); + const candidates = database.read((db) => + ( + db + .prepare( + "SELECT body FROM messaging_question_deliveries WHERE json_extract(body, '$.status') = 'pending' AND json_extract(body, '$.receipt.conversationId') = ? ORDER BY rowid DESC LIMIT 100", + ) + .all(event.conversation.id) as { body: string }[] + ).map((row) => JSON.parse(row.body) as QuestionDelivery), + ); + const quoted = candidates.find((value) => value.receipt?.messageId === event.reply.parentId); + if (!command && !quoted) return false; + if ( + event.channel !== 'feishu' || + event.conversation.kind !== 'dm' || + event.attachments?.length + ) + return true; + const match = /^\/answer\s+([A-Fa-f0-9]{12})\s+([\s\S]+)$/.exec(event.text.trim()); + const value = + quoted ?? + (match ? candidates.find((item) => item.reference === match[1]!.toUpperCase()) : undefined); + const ambiguous = + command && + (!match || (quoted !== undefined && quoted.reference !== match[1]!.toUpperCase())); + const trusted = candidates.find((item) => + authorized(item, providerId, { + ...event, + actorId: event.actor.id, + conversationId: event.conversation.id, + }), + ); + if (!trusted || signal.aborted) return true; + const question = value && request(value); + const text = match ? match[2]! : event.text.trim(); + if ( + !ambiguous && + value && + question?.questions.length === 1 && + text.length > 0 && + text.length <= 2000 + ) { + submit( + value, + { answers: [{ id: question.questions[0]!.id, selected: [], custom: text }] }, + providerId, + { ...event, actorId: event.actor.id, conversationId: event.conversation.id }, + signal, + ); + } else { + const { identity, provider } = context(trusted); + await provider.reply?.({ + accountRef: identity.accountRef, + fingerprint: identity.fingerprint, + route: event.reply, + text: + 'Select the original question form, or use /answer for one question. / 请使用原提问表单;单个问题可用 /answer <编号> <答案>。\n' + + candidates + .filter((item) => + authorized(item, providerId, { + ...event, + actorId: event.actor.id, + conversationId: event.conversation.id, + }), + ) + .map((item) => item.reference) + .join('\n'), + signal, + beforeSend: () => + authorized(trusted, providerId, { + ...event, + actorId: event.actor.id, + conversationId: event.conversation.id, + }), + }); + } + return true; + }, + close() { + closed = true; + controller.abort(); + detach?.(); + busy.clear(); + owner = undefined; + channels = undefined; + }, + }; + return service; +} diff --git a/packages/core/src/plugin.ts b/packages/core/src/plugin.ts index e3d6c1813e..bbbc3dd05c 100644 --- a/packages/core/src/plugin.ts +++ b/packages/core/src/plugin.ts @@ -2057,6 +2057,11 @@ export function apply(ctx: Context, config: BotHarnessConfig): void { 'botharness: companion question observation', ); registerChannelUserQuestions(ctx, userQuestions); + if (core.operationalDatabase.mode === 'ready') + ctx.effect( + () => core.externalMessaging.questions.attach(userQuestions, core.channels), + 'botharness: checked IM question answers', + ); const approvedCalls = new Map boolean>(); const deviceCallReleases = new Map void>(); ctx.effect(() => () => approvedCalls.clear(), 'botharness: one-call approval execution fences'); diff --git a/packages/core/test/bridge-rpc.test.ts b/packages/core/test/bridge-rpc.test.ts index 48163b1027..e634f53e05 100644 --- a/packages/core/test/bridge-rpc.test.ts +++ b/packages/core/test/bridge-rpc.test.ts @@ -625,6 +625,7 @@ describe('bridge typert service', () => { 'approvalRoute', 'approvalTest', 'approvalRetry', + 'questionRetry', 'failureAlertsSet', 'failureAlertsTest', 'failureAlertsRetry', diff --git a/packages/core/test/human-attention-index.test.ts b/packages/core/test/human-attention-index.test.ts index e54a92e37f..168f61594e 100644 --- a/packages/core/test/human-attention-index.test.ts +++ b/packages/core/test/human-attention-index.test.ts @@ -62,7 +62,7 @@ describe('Human attention candidate indexes', () => { mountOperationalDatabase({ dshHome, schemaPlan: BOT_HARNESS_SCHEMA_PLAN }), ); expect(upgraded.mode).toBe('ready'); - expect(upgraded.generation).toBe(115); + expect(upgraded.generation).toBe(116); const current = attachOperationalModule(upgraded, 'attention-index-upgrade'); const after = current.read((connection) => createHash('sha256') diff --git a/packages/core/test/lark-questions.test.ts b/packages/core/test/lark-questions.test.ts new file mode 100644 index 0000000000..2a57e9f107 --- /dev/null +++ b/packages/core/test/lark-questions.test.ts @@ -0,0 +1,402 @@ +import { afterEach, expect, it, vi } from 'vitest'; +import type { Agent } from '@deepseek-ai/dsh-agent'; +import type { AskUserQuestionItem } from '@deepseek-ai/dsh-user-questions/types'; +import { createCore, type BotHarnessCore } from '../src/plugin.js'; +import { ChannelUserQuestions } from '../src/channels/user-questions.js'; +import { createDshImProvider, type DshImOutboundService } from '../src/messaging/dsh-im.js'; +import { + MessagingProviderError, + type MessagingQuestionAction, + type MessagingInboundEvent, +} from '../src/messaging/provider.js'; +import { createTempRoot } from './helpers.js'; +import { createQuestionMessaging } from '../src/messaging/question-messaging.js'; +import { attachOperationalModule } from '../src/database/owner.js'; +import { createBridgeMethods } from '../src/bridge/methods.js'; + +const fingerprint = 'a'.repeat(64); +const cores: BotHarnessCore[] = []; +const owners: ChannelUserQuestions[] = []; +afterEach(async () => { + for (const owner of owners.splice(0)) owner.close(); + for (const core of cores.splice(0)) { + core.externalMessaging.close(); + await core.runtime.close(); + core.operationalDatabase.close(); + } +}); +async function fixture() { + const core = createCore({ dshHome: createTempRoot('bh-lark-question-') }); + cores.push(core); + core.registry.create({ slug: 'ada', displayName: 'Ada' }); + core.channels.getOrCreateDm('ada', 'Ada'); + let receiver: Parameters>[1] | undefined; + let fail: 'none' | 'known' | 'unknown' = 'none'; + const cards = vi.fn>( + async (_bot, route, card, input) => { + expect(input.beforeSend?.()).toBe(true); + if (fail !== 'none') + throw new MessagingProviderError('simulated', fail === 'known' ? 'not-started' : 'unknown'); + return input.update + ? { updated: true } + : { + sent: true, + receipt: { + version: 1, + messageId: 'om_' + card.requestId, + conversationId: route.conversationId, + }, + }; + }, + ); + const replies = vi.fn>( + async (_bot, _route, _text, input) => { + expect(input.beforeSend?.()).toBe(true); + return { sent: true }; + }, + ); + const host: DshImOutboundService = { + contractVersion: 1, + approvalCardVersion: 1, + questionCardVersion: 1, + replyFenceVersion: 1, + listBots: async () => [{ botId: 'lark-qa', channel: 'feishu' }], + listTargets: async () => [], + describeBot: async (botId) => ({ + version: 1, + botId, + channel: 'feishu', + connected: true, + account: { fingerprint, name: 'QA Lark' }, + capabilities: [ + 'proactive-text-checked', + 'exclusive-text-consumer', + 'reply-text-checked', + 'approval-card-checked', + 'approval-card-update-checked', + 'approval-action-consumer', + 'question-card-checked', + 'question-card-update-checked', + 'question-action-consumer', + ], + }), + sendChecked: async () => ({ sent: true }), + replyChecked: replies, + approvalCardChecked: async () => ({ sent: true }), + questionCardChecked: cards, + consumeInbound: async (_bot, input) => { + receiver = input; + return () => {}; + }, + }; + core.externalMessaging.register(createDshImProvider(host)!); + await core.externalMessaging.identity('ada', { + kind: 'bind', + providerId: 'dsh-im/feishu', + accountRef: 'lark-qa', + fingerprint, + }); + await vi.waitFor(() => expect(receiver).toBeDefined()); + const event: MessagingInboundEvent = { + version: 1, + channel: 'feishu', + botId: 'lark-qa', + fingerprint, + eventId: 'pair', + messageId: 'om_pair', + actor: { kind: 'user', id: 'ou_alice', name: 'Alice QA' }, + conversation: { kind: 'dm', id: 'oc_private' }, + mentions: [], + mentionedAccount: false, + at: new Date().toISOString(), + text: '/pair', + reply: { messageId: 'om_pair', conversationId: 'oc_private', actorId: 'ou_alice' }, + replay: { kind: 'provider-redelivery', resumeCursor: false, gapPossible: true }, + }; + await receiver!.onEvent(event, { signal: receiver!.signal }); + const pairing = core.externalMessaging.pairing.list('ada')[0]!; + const approved = core.externalMessaging.pairing.review('ada', { + id: pairing.id, + expectedRevision: 1, + kind: 'approve', + capabilities: ['answer'], + }); + await core.externalMessaging.approvals.setRoute('ada', pairing.id, 0); + const sessionId = 'bh-question-qa'; + core.ownership.claim({ + sessionId, + botSlug: 'ada', + rootRole: 'orchestrator', + at: new Date().toISOString(), + }); + const agent = { session: { id: sessionId, header: { cwd: process.cwd() } } } as Agent; + const owner = new ChannelUserQuestions(core.channels, core.ownership); + owners.push(owner); + core.externalMessaging.questions.attach(owner, core.channels); + const management = createBridgeMethods({ + registry: core.registry, + states: core.states, + channels: core.channels, + ownership: core.ownership, + roster: core.roster, + externalMessaging: core.externalMessaging, + }); + const start = async ( + questions: AskUserQuestionItem[] = [ + { + id: 'choice', + question: 'Choose a branch', + options: [ + { label: 'main', description: 'Current' }, + { label: 'qa', description: 'Test' }, + ], + }, + ], + ) => { + const previous = new Set( + core.externalMessaging.questions.snapshot('ada').map((value) => value.id), + ); + const answer = owner.ask({ agent, questions }); + void answer.catch(() => undefined); + await vi.waitFor(() => + expect( + core.externalMessaging.questions + .snapshot('ada') + .some((d) => !previous.has(d.id) && d.delivery !== 'pending' && d.delivery !== 'sending'), + ).toBe(true), + ); + const delivery = core.externalMessaging.questions.snapshot('ada')[0]!; + await vi.waitFor(() => + expect( + core.externalMessaging.questions.snapshot('ada').find((d) => d.id === delivery.id) + ?.delivery, + ).not.toMatch(/pending|sending/), + ); + return { answer, delivery }; + }; + const action = (id: string, changes: Partial = {}) => + receiver!.onAction!( + { + version: 1, + channel: 'feishu', + botId: 'lark-qa', + fingerprint, + actorId: 'ou_alice', + conversationId: 'oc_private', + messageId: 'om_' + id, + requestId: id, + action: 'answer', + values: [{ selected: [0] }], + ...changes, + }, + { signal: receiver!.signal }, + ); + const text = (text: string, parentId?: string) => { + const messageId = crypto.randomUUID(); + return receiver!.onEvent( + { + ...event, + eventId: crypto.randomUUID(), + messageId, + text, + reply: { ...event.reply, messageId, ...(parentId ? { parentId } : {}) }, + }, + { signal: receiver!.signal }, + ); + }; + return { + core, + owner, + cards, + replies, + approved, + start, + action, + text, + host, + agent, + management, + failure(value: typeof fail) { + fail = value; + }, + }; +} + +it('answer-only pairing settles the original native question, with actual actor and one canonical answer', async () => { + const f = await fixture(), + pending = await f.start([ + { + id: 'choice', + question: 'Choose a branch', + options: [{ label: 'main', description: 'Current' }], + }, + { id: 'notes', question: 'What should I remember?' }, + ]); + expect( + await f.action(pending.delivery.id, { + values: [{ selected: [0] }, { selected: [], custom: 'Remember this' }], + }), + ).toMatchObject({ status: 'queued' }); + expect(await pending.answer).toEqual({ + answers: [ + { id: 'choice', selected: ['main'] }, + { id: 'notes', selected: [], custom: 'Remember this' }, + ], + }); + const resolutions = f.core.channels + .readMessages('dm-ada') + .filter((m) => m.userQuestionResolution); + expect(resolutions).toHaveLength(1); + expect(resolutions[0]?.userQuestionResolution?.actor).toMatchObject({ + actorId: 'ou_alice', + pairingId: f.approved.id, + }); + expect(await f.action(pending.delivery.id)).toMatchObject({ status: 'refused' }); + expect(f.core.channels.readMessages('dm-ada').some((m) => m.toolApprovalDecision)).toBe(false); + await vi.waitFor(() => expect(f.cards.mock.calls.at(-1)?.[2].status).toBe('answered')); +}); + +it('forged actor, receipt, shape and revoked answer authority preserve the original ask', async () => { + const f = await fixture(), + pending = await f.start(); + for (const changed of [ + { actorId: 'ou_other' }, + { messageId: 'om_other' }, + { conversationId: 'oc_other' }, + ]) + expect(await f.action(pending.delivery.id, changed)).toMatchObject({ status: 'refused' }); + await expect(f.action(pending.delivery.id, { values: [] })).rejects.toThrow(); + expect(await f.action(pending.delivery.id, { values: [{ selected: [99] }] })).toMatchObject({ + status: 'refused', + }); + expect(f.owner.pending('ada', pending.delivery.id)).toBeDefined(); + f.core.externalMessaging.pairing.review('ada', { + kind: 'revoke', + id: f.approved.id, + expectedRevision: 2, + }); + expect(await f.action(pending.delivery.id)).toMatchObject({ status: 'refused' }); + expect(f.owner.pending('ada', pending.delivery.id)).toBeDefined(); +}); + +it('Web and Lark answer races have exactly one winner', async () => { + const f = await fixture(), + pending = await f.start(); + const web = f.owner.answer('ada', pending.delivery.id, { + answers: [{ id: 'choice', selected: ['qa'] }], + }); + await f.action(pending.delivery.id); + expect(await web).toBe(true); + expect(await pending.answer).toEqual({ answers: [{ id: 'choice', selected: ['qa'] }] }); + expect( + f.core.channels.readMessages('dm-ada').filter((m) => m.userQuestionResolution), + ).toHaveLength(1); +}); + +it('known-unsent delivery can be repaired, while uncertain delivery cannot be resent', async () => { + const f = await fixture(); + f.failure('known'); + const first = await f.start(); + expect(first.delivery.delivery).toBe('failed'); + f.failure('none'); + await f.core.externalMessaging.questions.retry('ada', first.delivery.id); + expect(f.core.externalMessaging.questions.snapshot('ada')[0]?.delivery).toBe('sent'); + f.failure('unknown'); + const second = await f.start(); + expect(second.delivery.delivery).toBe('unknown-outcome'); + const count = f.cards.mock.calls.length; + await expect(f.core.externalMessaging.questions.retry('ada', second.delivery.id)).rejects.toThrow( + 'question-repair-unavailable', + ); + expect(f.cards).toHaveBeenCalledTimes(count); +}); + +it('explicit references disambiguate questions; ordinary conversation and mismatched replies never answer', async () => { + const f = await fixture(), + first = await f.start(), + second = await f.start(); + await f.text('ordinary clarification'); + expect(f.owner.pending('ada', first.delivery.id)).toBeDefined(); + await f.text('/answer'); + expect(f.replies.mock.calls.at(-1)?.[2]).toContain(first.delivery.reference); + await f.text(`/answer ${second.delivery.reference} wrong question`, 'om_' + first.delivery.id); + expect(f.owner.pending('ada', first.delivery.id)).toBeDefined(); + await f.text(`/answer ${first.delivery.reference} custom branch`); + expect(await first.answer).toEqual({ + answers: [{ id: 'choice', selected: [], custom: 'custom branch' }], + }); + expect(f.owner.pending('ada', second.delivery.id)).toBeDefined(); +}); + +it('uncertain submission reconciles the original owner without replaying the answer', async () => { + const f = await fixture(), + pending = await f.start(); + const append = f.core.channels.appendMessage; + vi.spyOn(f.core.channels, 'appendMessage').mockImplementationOnce(async (...args) => { + await append(...args); + throw new Error('Result lost after commit'); + }); + await f.action(pending.delivery.id); + await vi.waitFor(() => + expect(f.core.externalMessaging.questions.snapshot('ada')[0]?.submission).toBe( + 'unknown-outcome', + ), + ); + expect(f.owner.pending('ada', pending.delivery.id)).toBeDefined(); + expect(await f.action(pending.delivery.id)).toMatchObject({ status: 'refused' }); + expect( + await f.management.questionRetry({ slug: 'other-bot', id: pending.delivery.id }), + ).toMatchObject({ ok: false, error: { code: 'question-repair-unavailable' } }); + expect(await f.management.questionRetry({ slug: 'ada', id: 'not-a-question' })).toMatchObject({ + ok: false, + }); + expect(await f.management.questionRetry({ slug: 'ada', id: pending.delivery.id })).toEqual({ + ok: true, + value: { updated: true }, + }); + expect(await pending.answer).toEqual({ answers: [{ id: 'choice', selected: ['main'] }] }); + expect( + f.core.channels.readMessages('dm-ada').filter((m) => m.userQuestionResolution), + ).toHaveLength(1); +}); + +it('a restarted notification service expires old controls and updates the original receipt without restoring the native owner', async () => { + const f = await fixture(), + pending = await f.start(); + f.core.externalMessaging.questions.close(); + const nextOwner = new ChannelUserQuestions(f.core.channels, f.core.ownership); + owners.push(nextOwner); + const restarted = createQuestionMessaging({ + database: attachOperationalModule(f.core.operationalDatabase, 'question-recovery-test'), + pairing: f.core.externalMessaging.pairing, + approvals: f.core.externalMessaging.approvals, + provider: () => createDshImProvider(f.host)!, + isBotActive: () => true, + }); + try { + restarted.attach(nextOwner, f.core.channels); + expect(restarted.snapshot('ada')[0]?.status).toBe('expired'); + expect(nextOwner.pending('ada', pending.delivery.id)).toBeUndefined(); + expect( + await restarted.action( + 'dsh-im/feishu', + { + version: 1, + channel: 'feishu', + botId: 'lark-qa', + fingerprint, + actorId: 'ou_alice', + conversationId: 'oc_private', + messageId: 'om_' + pending.delivery.id, + requestId: pending.delivery.id, + action: 'answer', + values: [{ selected: [0] }], + }, + new AbortController().signal, + ), + ).toMatchObject({ status: 'refused' }); + await vi.waitFor(() => expect(f.cards.mock.calls.at(-1)?.[2].status).toBe('expired')); + } finally { + restarted.close(); + } +}); diff --git a/packages/core/test/plugin.test.ts b/packages/core/test/plugin.test.ts index 93b2511e49..7d5ab4ce60 100644 --- a/packages/core/test/plugin.test.ts +++ b/packages/core/test/plugin.test.ts @@ -576,6 +576,7 @@ describe('plugin entry', () => { 'approvalRoute', 'approvalTest', 'approvalRetry', + 'questionRetry', 'failureAlertsSet', 'failureAlertsTest', 'failureAlertsRetry', diff --git a/packages/core/test/user-questions.test.ts b/packages/core/test/user-questions.test.ts index 5b63effaed..29de1822f4 100644 --- a/packages/core/test/user-questions.test.ts +++ b/packages/core/test/user-questions.test.ts @@ -60,6 +60,105 @@ function fixture(role: 'orchestrator' | 'assignment' = 'orchestrator', timed?: T } describe('native DSH questions in a PersonaBot DM', () => { + it('publishes only committed requests and accepted answers, while a broken observer cannot settle the owner', async () => { + const state = fixture(); + const notices: string[] = []; + state.answerer.observe((notice, status) => { + expect(state.channels.message(channelId, notice.messageId)).toBeDefined(); + notices.push(status); + }); + state.answerer.observe(() => { + throw new Error('observer failed'); + }); + const wait = state.answerer.ask({ agent: state.agent, questions }); + await vi.waitFor(() => expect(notices).toEqual(['pending'])); + const id = state.answerer.activeMessageIds()[0]!; + expect(state.answerer.pending('ada', id)?.sessionId).toBe(sessionId); + expect( + await state.answerer.answer('ada', id, { + answers: [{ id: questions[0]!.id, selected: ['main'] }], + }), + ).toBe(true); + await wait; + expect(notices).toEqual(['pending', 'answered']); + expect(state.warn).toHaveBeenCalledWith('user-question-notification-failed'); + }); + + it('rechecks external authority inside the canonical Channel commit queue and preserves the ask when it is revoked', async () => { + const state = fixture(); + const wait = state.answerer.ask({ agent: state.agent, questions }); + await vi.waitFor(() => expect(state.answerer.activeMessageIds()).toHaveLength(1)); + const id = state.answerer.activeMessageIds()[0]!; + let authorized = true; + await vi.waitFor(() => expect(state.answerer.pending('ada', id)).toBeDefined()); + const answer = { answers: [{ id: questions[0]!.id, selected: ['main'] }] }; + const result = state.answerer.answer('ada', id, answer, { + actor: { + platform: 'feishu', + bindingId: 'binding', + fingerprint: 'a'.repeat(64), + actorId: 'ou_human', + pairingId: 'pairing', + pairingRevision: 1, + conversationId: 'oc_dm', + messageId: 'om_card', + }, + authorized: () => authorized, + }); + authorized = false; + expect(await result).toBe(false); + expect(state.channels.readMessages(channelId)).toHaveLength(1); + expect(state.answerer.status('ada', id)).toBe('pending'); + expect(await state.answerer.answer('ada', id, answer)).toBe(true); + expect(await wait).toEqual(answer); + }); + + it('keeps the original native ask after a failed answer write and accepts only one winner', async () => { + const state = fixture(); + const wait = state.answerer.ask({ agent: state.agent, questions }); + await vi.waitFor(() => expect(state.answerer.activeMessageIds()).toHaveLength(1)); + const id = state.answerer.activeMessageIds()[0]!; + const answer = { answers: [{ id: questions[0]!.id, selected: ['main'] }] }; + await vi.waitFor(() => expect(state.answerer.pending('ada', id)).toBeDefined()); + vi.spyOn(state.channels, 'appendMessage').mockRejectedValueOnce( + new Error('answer write failed'), + ); + await expect(state.answerer.answer('ada', id, answer)).rejects.toThrow('answer write failed'); + expect(state.answerer.status('ada', id)).toBe('pending'); + expect( + await Promise.all([ + state.answerer.answer('ada', id, answer), + state.answerer.answer('ada', id, answer), + ]), + ).toEqual([true, false]); + expect(await wait).toEqual(answer); + expect( + state.channels.readMessages(channelId).filter((m) => m.userQuestionResolution), + ).toHaveLength(1); + }); + + it('requires explicit owner reconciliation after a write commits but reports an uncertain result', async () => { + const state = fixture(); + const wait = state.answerer.ask({ agent: state.agent, questions }); + await vi.waitFor(() => expect(state.answerer.activeMessageIds()).toHaveLength(1)); + const id = state.answerer.activeMessageIds()[0]!; + await vi.waitFor(() => expect(state.answerer.pending('ada', id)).toBeDefined()); + const answer = { answers: [{ id: questions[0]!.id, selected: ['main'] }] }; + const append = state.channels.appendMessage.bind(state.channels); + vi.spyOn(state.channels, 'appendMessage').mockImplementationOnce(async (...args) => { + await append(...args); + throw new Error('post-commit write outcome unknown'); + }); + await expect(state.answerer.answer('ada', id, answer)).rejects.toThrow('unknown'); + expect(await state.answerer.answer('ada', id, answer)).toBe(false); + expect(state.answerer.pending('ada', id)).toBeDefined(); + expect(state.answerer.reconcile('ada', id)).toBe(true); + expect(await wait).toEqual(answer); + expect( + state.channels.readMessages(channelId).filter((m) => m.userQuestionResolution), + ).toHaveLength(1); + }); + it('commits a card, validates the Human answer, and resumes the same waiting request', async () => { const state = fixture(); const wait = state.answerer.ask({ agent: state.agent, questions }); diff --git a/scripts/dev-im-provider.mjs b/scripts/dev-im-provider.mjs index c951ce6b33..09b8d21ae8 100644 --- a/scripts/dev-im-provider.mjs +++ b/scripts/dev-im-provider.mjs @@ -5,11 +5,11 @@ import { join } from 'node:path'; export const qualifiedImProvider = Object.freeze({ package: '@xmanrui/dsh-im', packageVersion: '4.32.0', - source: '518314f6cf495aea093c0a93b0107de25b66e60d', - spec: 'github:DoodleBears/dsh-im#518314f6cf495aea093c0a93b0107de25b66e60d', + source: 'e655636c061fa0aed8c77c23f7e87cb5abf79b65', + spec: 'github:DoodleBears/dsh-im#e655636c061fa0aed8c77c23f7e87cb5abf79b65', dsh: '0.2.1-rc.2', - runtimeFiles: 424, - runtimeSha256: 'bc54309720ce66906b562d7afaae99669c2177d67717623f6999e906abe473d1', + runtimeFiles: 425, + runtimeSha256: '7b9cbc1706ed42a114d10d106ddf41d1fcbbc3f93dcfbe358950802e695f1647', }); export function withQualifiedImProvider(manifest) {