diff --git a/.github/pr-screenshots/1145/after-delete.png b/.github/pr-screenshots/1145/after-delete.png new file mode 100644 index 0000000000..602773d997 Binary files /dev/null and b/.github/pr-screenshots/1145/after-delete.png differ diff --git a/.github/pr-screenshots/1145/connect-dialog.png b/.github/pr-screenshots/1145/connect-dialog.png new file mode 100644 index 0000000000..34c630941f Binary files /dev/null and b/.github/pr-screenshots/1145/connect-dialog.png differ diff --git a/.github/pr-screenshots/1145/edit-dialog.png b/.github/pr-screenshots/1145/edit-dialog.png new file mode 100644 index 0000000000..fd520bfd0b Binary files /dev/null and b/.github/pr-screenshots/1145/edit-dialog.png differ diff --git a/.github/pr-screenshots/1145/local-at-answer.png b/.github/pr-screenshots/1145/local-at-answer.png new file mode 100644 index 0000000000..47bdc3493a Binary files /dev/null and b/.github/pr-screenshots/1145/local-at-answer.png differ diff --git a/.github/pr-screenshots/1145/row-receiving.png b/.github/pr-screenshots/1145/row-receiving.png new file mode 100644 index 0000000000..78130c32d1 Binary files /dev/null and b/.github/pr-screenshots/1145/row-receiving.png differ diff --git a/CHANGELOG.md b/CHANGELOG.md index cadc6ef58a..b89f28ee7d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ WeChat original voice can be prepared for playback, native quotes and retained l ### Added +- A group Channel can connect an external conversation: **Connect an external conversation** under External connectors in the Channel Profile picks a bound app and one of its known groups, and every message from that group then appears in the Channel. Member Bots get the messages as context only by default and wake only when someone @s them; the connection's wake setting can switch to a batch or every message, and a member's own wake policy in the Channel still wins. Each row shows its state and last message, with a switch to pause and **Delete** to stop new messages while keeping the ones already here ([#1145](https://github.com/BotHarness/BotHarness/issues/1145), [ADR-0143](docs/adr/0143-an-external-conversation-streams-into-a-channel-as-one-way-context.md), [External connectors guide](docs/channel-sidebar/external-connectors.md)). - Slack DMs to a bound app now reach the Bot's Inbox and get answered in the DM without opening a thread, and Discord apps appear in **Bind app** again. The development IM provider moves to a dsh-im build with Slack DM intake, Discord checked delivery and checked Slack/Discord reply mentions ([#1125](https://github.com/BotHarness/BotHarness/issues/1125), [External identities guide](docs/channel-sidebar/external-identities.md)). - A Bot can be shown as a transparent pixel Window Companion from its sidebar menu. It follows genuine Activity and new Human–Bot DM replies across Harness pages, supports eased drag poses, continuous inertial falling and a soft landing alongside a hover reading list, and restores its local selection after restart without replaying earlier messages. Reduced motion disables the drop and pose effects. This first slice supports one companion; pause stops walking, not Bot execution ([#1138](https://github.com/BotHarness/DeepSeekBot/issues/1138)). diff --git a/CHANGELOG.zh.md b/CHANGELOG.zh.md index c3d69948e5..04955bdd48 100644 --- a/CHANGELOG.zh.md +++ b/CHANGELOG.zh.md @@ -9,6 +9,7 @@ ### Added +- 群频道可以接入外部会话：在频道 Profile 的「外部连接器」中点 **接入外部会话**，选一个已绑定的应用和它已知的一个群，之后这个群里的每条消息都会出现在频道里。成员 Bot 默认只把这些消息当作上下文，只有被 @ 时才会被唤醒；接入的唤醒设置可以改成攒够条数后唤醒或每条立即唤醒，成员在频道里单独设置的唤醒策略仍然优先。每一行显示状态和最后一条消息的时间，可以用开关暂停，**删除** 后不再接收新消息，已接入的消息保留（[#1145](https://github.com/BotHarness/BotHarness/issues/1145)，[ADR-0143](docs/adr/0143-an-external-conversation-streams-into-a-channel-as-one-way-context.md)，[外部连接器教程](docs/channel-sidebar/external-connectors.md)）。 - 私聊已绑定的 Slack 应用，消息现在会进入 Bot 收件箱，Bot 直接在私聊里回复，不开 thread；Discord 应用也重新出现在 **绑定应用** 里。开发用的 IM Provider 换到新的 dsh-im 版本，包含 Slack 私聊接收、Discord checked 发送，以及 Slack/Discord 回复里的 checked @（[#1125](https://github.com/BotHarness/BotHarness/issues/1125)、[外部身份指南](docs/channel-sidebar/external-identities.md)）。 - 从侧栏 Bot 菜单可显示透明像素窗口伙伴，在 Harness 页面之间保留，跟随真实 Activity 和新的 Human–Bot 私聊回复，支持拖拽姿态缓动、连续惯性下落与柔和落地，以及悬浮阅读列表；减少动效时关闭下落和姿态效果。重启恢复本地选择，不重播历史消息；首个切片支持一个伙伴，暂停只停止走动，不暂停 Bot 执行（[#1138](https://github.com/BotHarness/DeepSeekBot/issues/1138)）。 diff --git a/CONTEXT.md b/CONTEXT.md index 8c4d42236e..215b081364 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -438,6 +438,10 @@ _Avoid_: default folder, inbox, fixed bottom bucket A configured connection from an external source, such as an IM conversation or later a webhook, to an explicit Channel or PersonaBot Inbox target; it carries inbound delivery and exposes outbound capabilities without becoming the Actor. For an IM conversation it is listed under External connectors (外部连接器); the PersonaBot's own external identity is a separate thing. _Avoid_: integration, adapter, connector (bare) +**Conversation ingest**: +A Channel-owned, one-way connection that places every message of one external conversation into that Channel as Source Events, with context-only member Admissions by default; its wake setting can switch to a batch or every message, and a member PersonaBot's own wake policy in that Channel wins. It is listed under External connectors as an external conversation (外部会话); it grants no reply or other authority to any PersonaBot. Until slice 9 converges them, a Bridge is the Bot-owned route and a Conversation ingest is the Channel-owned one. +_Avoid_: sync, mirror, Bridge (for this record) + **App**: The UI name (应用) for one authenticated Provider account, such as a Lark app, a Slack app, a Discord bot or a paired WeChat Bot. An App is bound to at most one PersonaBot; a PersonaBot may bind several Apps, including several of one platform. Settings → IM apps lists every App and the Bot that uses it. _Avoid_: IM account (in UI copy), connector, integration diff --git a/CONTEXT.zh.md b/CONTEXT.zh.md index 040cb75bc9..4c8c8edb8f 100644 --- a/CONTEXT.zh.md +++ b/CONTEXT.zh.md @@ -436,6 +436,10 @@ _避免使用_：default folder、inbox、fixed bottom bucket 从 external source（如 IM 会话，未来还有 webhook）到某个显式 Channel 或 PersonaBot Inbox target 的已配置连接；它承载 inbound delivery 并暴露 outbound capability，但不会成为 Actor。对于 IM 会话，它列在「外部连接器 / External connector」中；PersonaBot 自己的外部身份是另一回事。 _避免使用_：integration、adapter、裸用 connector +**Conversation ingest**： +外部会话接入。由 Channel 持有的单向连接：把某个外部会话的每条消息作为 Source Event 放进该 Channel，成员 PersonaBot 默认只获得「仅作上下文」的 Admission；接入的唤醒设置可改为攒够条数后唤醒或逐条唤醒，成员在该 Channel 中的唤醒策略优先。它列在「外部连接器」中，显示为外部会话；它不给任何 PersonaBot 回复或其他权限。在 slice 9 合并之前，Bridge 指 Bot 持有的路由，Conversation ingest 指 Channel 持有的接入。 +_避免使用_：sync、mirror、用 Bridge 指代这条记录 + **App**： UI 上叫「应用」：一个已认证 Provider account 的名称，例如 Lark 应用、Slack 应用、Discord bot 或已配对的微信 Bot。一个应用最多绑定到一个 PersonaBot；一个 PersonaBot 可以绑定多个应用，包括同一平台的多个应用。「设置 → IM 应用」列出每个应用及使用它的 Bot。 _避免使用_：UI 文案里的「IM账号」、connector、integration diff --git a/docs/channel-sidebar/external-connectors.md b/docs/channel-sidebar/external-connectors.md index 14a0dfcd27..06437822e7 100644 --- a/docs/channel-sidebar/external-connectors.md +++ b/docs/channel-sidebar/external-connectors.md @@ -31,4 +31,16 @@ Lark, Slack and Discord apps post to conversations directly, so they never need ![The send target dialog](/guides/channel-sidebar/22-conversation-authorization-zh.webp) +## Connect an external conversation to a group Channel + +A group Channel can take in everything said in an external group, the way a webhook feeds it. Open the group's detailed Profile, scroll to **External connectors** and click **Connect an external conversation**: + +1. Choose an **App**. Every app a member Bot has bound is listed, with the platform and the Bot that uses it. +2. Choose a **Conversation**. The list shows the groups that app already knows, for example a group where someone has @mentioned it. If yours is missing, @ the app once in that group. +3. Keep **Wake** on **Context only**, or pick **Wake after a batch** or **Wake on every message**, then click **Connect**. + +From then on every message from that group appears in the Channel. With **Context only**, member Bots read those messages only when someone @s them, locally or by @mentioning their own app in the group. A member's own wake policy set in this Channel takes precedence. Nothing is sent back to the external group. + +The row shows **Waiting for the first message** until one arrives, then **Receiving** and the time of the last message. The app must receive ordinary group messages: Lark needs the `im:message.group_msg` scope. Use the Switch to pause, and **Delete** in the edit dialog to stop new messages; messages already in the Channel stay. + A local Group still manages its connectors in the group’s detailed Profile. Platform setup: [Lark / Feishu](/docs/lark-connection), [Slack](/docs/slack-connection), [personal WeChat](/docs/wechat-connection). [All sidebar features](/docs/channel-sidebar). diff --git a/docs/channel-sidebar/external-connectors.zh.md b/docs/channel-sidebar/external-connectors.zh.md index 7a46e6b78e..710cdf4ee0 100644 --- a/docs/channel-sidebar/external-connectors.zh.md +++ b/docs/channel-sidebar/external-connectors.zh.md @@ -31,4 +31,16 @@ Lark、Slack 和 Discord 应用可以直接发往会话，永远不需要这一 ![发送目标弹窗](/guides/channel-sidebar/22-conversation-authorization-zh.webp) +## 把外部会话接入群频道 + +群频道可以像接 webhook 一样，接收一个外部群里的所有消息。打开群的详细 Profile，滚动到 **外部连接器**，点 **接入外部会话**： + +1. 选择 **应用**。这里列出成员 Bot 已绑定的所有应用，并注明平台和使用它的 Bot。 +2. 选择 **会话**。列表里是这个应用已知的群，比如有人 @ 过它的群。如果找不到你的群，先在那个群里 @ 一次应用。 +3. **唤醒** 保持 **仅作上下文**，或者选 **攒够条数后唤醒**、**每条立即唤醒**，然后点 **接入**。 + +之后这个群里的每条消息都会出现在频道里。选 **仅作上下文** 时，成员 Bot 只有在被 @ 时才会读到这些消息，无论是在频道里本地 @，还是在群里 @ 它自己的应用。成员在这个频道里单独设置的唤醒策略优先。不会有任何内容发回外部群。 + +在第一条消息到达之前，这一行显示 **等待第一条消息**，之后显示 **正在接收** 和最后一条消息的时间。应用需要能收到群里的普通消息：飞书需要开通 `im:message.group_msg` 权限。用开关可以暂停；在编辑窗口里点 **删除** 后不再接收新消息，已接入的消息保留。 + 本地群聊仍在群的详细 Profile 中管理外部连接器。平台配置：[Lark / 飞书](/zh/docs/lark-connection)、[Slack](/zh/docs/slack-connection)、[个人微信](/zh/docs/wechat-connection)。返回[侧栏总览](/zh/docs/channel-sidebar)。 diff --git a/packages/client/src/client/actions.ts b/packages/client/src/client/actions.ts index 03d093f1db..07315e7173 100644 --- a/packages/client/src/client/actions.ts +++ b/packages/client/src/client/actions.ts @@ -25,6 +25,10 @@ import type { ChannelBridgeInput, ChannelBridgeSnapshot, } from '../../../core/src/messaging/channel-bridge.js'; +import type { + ConversationIngestInput, + ConversationIngestSnapshot, +} from '../../../core/src/messaging/conversation-ingest.js'; import type { MessagingIdentity, MessagingIdentityInput, @@ -50,6 +54,8 @@ import type { MessagingTarget } from '../../../core/src/messaging/provider.js'; import { loadChannelBridges, manageChannelBridge, + loadChannelIngests, + manageChannelIngest, loadMessagingSnapshot, manageMessagingIdentity, manageMessagingConversation, @@ -257,6 +263,8 @@ export interface HostDirectoryListing { export interface BridgeActions { channelBridges(channelId: string): Promise; channelBridge(channelId: string, input: ChannelBridgeInput): Promise; + channelIngests(channelId: string): Promise; + channelIngest(channelId: string, input: ConversationIngestInput): Promise; messagingChannelTarget(slug: string, grantId: string, channelId: string | null): Promise; messagingThreadPolicy( slug: string, @@ -1804,6 +1812,8 @@ export function createActions( messagingSource: (slug, sourceEventId) => readMessagingSource(call, slug, sourceEventId), channelBridges: (channelId) => loadChannelBridges(call, channelId), channelBridge: (channelId, input) => manageChannelBridge(call, channelId, input), + channelIngests: (channelId) => loadChannelIngests(call, channelId), + channelIngest: (channelId, input) => manageChannelIngest(call, channelId, input), approvalRoute: (slug, pairingId, expectedRevision) => setApprovalRoute(call, slug, pairingId, expectedRevision), approvalTest: (slug) => testApprovalRoute(call, slug), diff --git a/packages/client/src/client/bridge.ts b/packages/client/src/client/bridge.ts index 58cf5017ba..c5d7168c24 100644 --- a/packages/client/src/client/bridge.ts +++ b/packages/client/src/client/bridge.ts @@ -52,6 +52,10 @@ import type { GroupReceptionInput } from '../../../core/src/messaging/group-poli import type { MessagingConversationInput } from '../../../core/src/messaging/conversations.js'; import type { ActivityOverview } from '../../../core/src/bridge/methods.js'; import type { ExternalSource } from '../../../core/src/messaging/inbound.js'; +import type { + ConversationIngestInput, + ConversationIngestSnapshot, +} from '../../../core/src/messaging/conversation-ingest.js'; import type { HumanAssignmentContext } from '../../../core/src/runtime/assignment-human-context.js'; export type { HumanAssignmentContext } from '../../../core/src/runtime/assignment-human-context.js'; import type { @@ -3486,6 +3490,56 @@ export async function loadChannelBridges( throw new BridgeCallError('invalid-response', 'Invalid Channel Bridge snapshot'); return value as unknown as ChannelBridgeSnapshot; } +export async function loadChannelIngests( + call: BridgeCall, + channelId: string, +): Promise { + const value = asRecord(await unwrap(call, 'channelIngests', { channelId })); + if ( + !value || + value['channelId'] !== channelId || + !Array.isArray(value['ingests']) || + !value['ingests'].every((item) => { + const row = asRecord(item); + const conversation = asRecord(row?.['conversation']); + const wake = asRecord(row?.['wake']); + return ( + row && + conversation && + wake && + ['id', 'channelId', 'platform', 'accountName', 'intakeAfter'].every( + (key) => typeof row[key] === 'string', + ) && + typeof conversation['id'] === 'string' && + typeof conversation['name'] === 'string' && + typeof row['enabled'] === 'boolean' && + Number.isInteger(row['revision']) && + ['mentions', 'digest', 'all'].includes(String(wake['mode'])) && + ['waiting', 'receiving', 'paused', 'unavailable'].includes(String(row['state'])) + ); + }) || + !Array.isArray(value['candidates']) || + !value['candidates'].every((item) => { + const row = asRecord(item); + return ( + row && + ['bindingId', 'botSlug', 'platform', 'accountName'].every( + (key) => typeof row[key] === 'string', + ) && + Array.isArray(row['conversations']) + ); + }) + ) + throw new BridgeCallError('invalid-response', 'Invalid external conversation snapshot'); + return value as unknown as ConversationIngestSnapshot; +} +export async function manageChannelIngest( + call: BridgeCall, + channelId: string, + input: ConversationIngestInput, +): Promise { + await unwrap(call, 'channelIngest', { channelId, input }); +} export async function manageChannelBridge( call: BridgeCall, channelId: string, diff --git a/packages/client/src/client/channel-bridge-list.tsx b/packages/client/src/client/channel-bridge-list.tsx index 814a8ef346..e89bba52d1 100644 --- a/packages/client/src/client/channel-bridge-list.tsx +++ b/packages/client/src/client/channel-bridge-list.tsx @@ -41,6 +41,7 @@ export function ChannelBridgeList({ botNames, actions, t, + showEmpty = true, children, }: { channelId: string; @@ -48,6 +49,7 @@ export function ChannelBridgeList({ botNames: ReadonlyMap; actions: Pick; t: BotHarnessTranslate; + showEmpty?: boolean; children?: ReactNode; }): ReactElement { const [snapshot, setSnapshot] = useState(); @@ -244,7 +246,7 @@ export function ChannelBridgeList({ } /> ))} - {snapshot && !snapshot.bridges.length ? ( + {showEmpty && snapshot && !snapshot.bridges.length ? (
  • {t('bridge.empty')}
    • ) : null} {children} diff --git a/packages/client/src/client/conversation-ingest-list.tsx b/packages/client/src/client/conversation-ingest-list.tsx new file mode 100644 index 0000000000..392fa296a0 --- /dev/null +++ b/packages/client/src/client/conversation-ingest-list.tsx @@ -0,0 +1,317 @@ +import { useRef, useState, type ReactElement } from 'react'; +import { Button, Switch, Tag } from '@deepseek-ai/dsh-client-ui-primitives'; + +import type { + ConversationIngestRow, + ConversationIngestSnapshot, + ConversationIngestWake, +} from '../../../core/src/messaging/conversation-ingest.js'; +import type { BridgeActions } from './actions.js'; +import { externalPlatformLabel } from './bridge-source-label.js'; +import { Combobox } from './combobox.js'; +import { formatRelativeTime } from './labels.js'; +import type { BotHarnessTranslate } from './locale.js'; +import { Modal } from './modal.js'; +import { useMountedResource } from './mounted-resource.js'; +import { SidebarCardRow } from './sidebar-card.js'; + +const WAKE: Record = { + mentions: { mode: 'mentions', count: 10, intervalSeconds: 300 }, + digest: { mode: 'digest', count: 10, intervalSeconds: 300 }, + all: { mode: 'all', count: 1, intervalSeconds: 10 }, +}; + +export function ConversationIngestRows({ + channelId, + botNames, + actions, + t, +}: { + channelId: string; + botNames: ReadonlyMap; + actions: Pick; + t: BotHarnessTranslate; +}): ReactElement { + const [snapshot, setSnapshot] = useState(); + const [mode, setMode] = useState<'add' | 'edit' | 'delete'>(); + const [selected, setSelected] = useState(); + const [bindingId, setBindingId] = useState(''); + const [conversationId, setConversationId] = useState(''); + const [wake, setWake] = useState('mentions'); + const [busy, setBusy] = useState(false); + const [error, setError] = useState(''); + const mounted = useRef(false); + const generation = useRef(0); + const refresh = async () => { + const version = generation.current; + const next = await actions.channelIngests(channelId); + if (mounted.current && version === generation.current) setSnapshot(next); + }; + const mount = useMountedResource(() => { + mounted.current = true; + ++generation.current; + void refresh().catch(() => { + if (mounted.current) setError(t('ingest.failed')); + }); + return () => { + mounted.current = false; + ++generation.current; + }; + }, [channelId, actions]); + const operate = async (operation: () => Promise, close = false) => { + if (busy) return; + setBusy(true); + setError(''); + try { + await operation(); + await refresh(); + if (close && mounted.current) setMode(undefined); + } catch (err) { + await refresh().catch(() => undefined); + const code = err instanceof Error && 'code' in err ? err.code : ''; + if (mounted.current) + setError(t(code === 'ingest-exists' ? 'ingest.exists' : 'ingest.failed')); + } finally { + if (mounted.current) setBusy(false); + } + }; + const open = (next: typeof mode, row?: ConversationIngestRow) => { + setMode(next); + setSelected(row); + setBindingId(''); + setConversationId(''); + setWake(row?.wake.mode ?? 'mentions'); + setError(''); + }; + const candidate = snapshot?.candidates.find((item) => item.bindingId === bindingId); + const connected = (id: string) => + snapshot?.ingests.some( + (row) => + row.conversation.id === id && + candidate !== undefined && + row.platform === candidate.platform && + row.accountName === candidate.accountName, + ) ?? false; + const update = ( + row: ConversationIngestRow, + patch: { enabled?: boolean; wake?: ConversationIngestWake }, + ) => + actions.channelIngest(channelId, { + kind: 'update', + ingestId: row.id, + expectedRevision: row.revision, + ...patch, + }); + const save = async () => { + if (mode === 'add') + await actions.channelIngest(channelId, { + kind: 'add', + bindingId, + conversation: { kind: 'group', id: conversationId }, + wake: WAKE[wake], + }); + else if (mode === 'delete' && selected) + await actions.channelIngest(channelId, { + kind: 'delete', + ingestId: selected.id, + expectedRevision: selected.revision, + }); + else if (selected) await update(selected, { wake: WAKE[wake] }); + }; + const close = () => { + if (!busy) setMode(undefined); + }; + const stateTone = (row: ConversationIngestRow) => + row.state === 'receiving' ? 'success' : row.state === 'unavailable' ? 'warning' : 'neutral'; + return ( + <> + {snapshot?.ingests.map((row) => ( + {t(`ingest.state.${row.state}`)}} + meta={[ + externalPlatformLabel(row.platform, t), + row.accountName, + t(`ingest.wake.${row.wake.mode}`), + ...(row.lastMessageAt + ? [ + t('ingest.lastMessage', { + time: formatRelativeTime(Date.parse(row.lastMessageAt), Date.now(), t), + }), + ] + : []), + ].join(' · ')} + muted={!row.enabled} + hint={t('ingest.edit', { name: row.conversation.name })} + dialog + disabled={busy} + onClick={() => open('edit', row)} + trailing={ + void operate(() => update(row, { enabled: checked }))} + /> + } + /> + ))} +
  • + + {error} + +
    • + ) : null} + + {mode === 'edit' && selected ? ( + <> + + + + ) : null} + + + + } + > +
    + {error ? ( +

    + {error} +

    + ) : null} + {mode === 'add' ? ( + <> + + {snapshot && !snapshot.candidates.length ? ( +

    {t('ingest.noApps')}

    + ) : null} + {candidate ? ( + + ) : null} + {candidate && !candidate.conversations.length ? ( +

    {t('ingest.noConversations')}

    + ) : null} + + ) : null} + {mode === 'delete' ? ( +

    {t('ingest.deleteImpact')}

    + ) : mode === undefined ? null : ( + <> +
    + {t('ingest.wake')} + + setWake(value === 'all' ? 'all' : value === 'digest' ? 'digest' : 'mentions') + } + options={(['mentions', 'digest', 'all'] as const).map((value) => ({ + value, + label: t(`ingest.wake.${value}`), + }))} + /> +
    +

    {t('ingest.wakeHint')}

    +

    {t('ingest.requirement')}

    + + )} +
    +     + + ); +} diff --git a/packages/client/src/client/group-profile.tsx b/packages/client/src/client/group-profile.tsx index dd5f84b672..0519b8ffeb 100644 --- a/packages/client/src/client/group-profile.tsx +++ b/packages/client/src/client/group-profile.tsx @@ -8,6 +8,7 @@ import { import { bridgeSourceLabel } from './bridge-source-label.js'; import { GroupWakePolicyTable } from './group-wake-policy-table.js'; import { ChannelBridgeList } from './channel-bridge-list.js'; +import { ConversationIngestRows } from './conversation-ingest-list.js'; import type { BridgeActions } from './actions.js'; import type { ReactElement } from 'react'; @@ -249,7 +250,12 @@ export function GroupProfileView({ }: GroupProfileProps & { actions: Pick< BridgeActions, - 'channelBridges' | 'channelBridge' | 'groupWakePolicies' | 'setGroupWakePolicy' + | 'channelBridges' + | 'channelBridge' + | 'channelIngests' + | 'channelIngest' + | 'groupWakePolicies' + | 'setGroupWakePolicy' >; onTogglePin(id: string): void; onClose(): void; @@ -301,14 +307,31 @@ export function GroupProfileView({ actions={actions} t={t} /> - +
    +
    +
    + {t('bridge.title')} +

    {t('bridge.summary')}

    +
    +
    + + + +
    ); } diff --git a/packages/client/src/client/locale.ts b/packages/client/src/client/locale.ts index a0c08ee5e6..ade70ddac8 100644 --- a/packages/client/src/client/locale.ts +++ b/packages/client/src/client/locale.ts @@ -774,6 +774,36 @@ export const zh = { 'bridge.confirmDelete': '删除此连接器', 'bridge.save': '保存外部连接器', 'bridge.empty': '尚未接入外部信息。', + 'ingest.add': '接入外部会话', + 'ingest.addTitle': '接入外部会话', + 'ingest.edit': '编辑外部会话：{name}', + 'ingest.delete': '删除外部会话', + 'ingest.confirmDelete': '删除此接入', + 'ingest.save': '接入', + 'ingest.update': '保存', + 'ingest.summary': '把外部群里的每条消息带进这个频道，作为成员 Bot 的上下文。只收不发。', + 'ingest.app': '应用', + 'ingest.appHint': '{platform} · 由 {bot} 使用', + 'ingest.conversation': '会话', + 'ingest.noApps': '还没有可用的应用。请先在某个成员 Bot 的「外部身份」中绑定一个应用。', + 'ingest.noConversations': '这个应用还没有已知的群。先在群里 @ 一次应用，它就会出现在这里。', + 'ingest.wake': '唤醒', + 'ingest.wake.mentions': '仅作上下文', + 'ingest.wake.digest': '攒够条数后唤醒', + 'ingest.wake.all': '每条立即唤醒', + 'ingest.wakeHint': + '仅作上下文时，Bot 只在被 @ 时读到这些消息。成员在频道中单独设置的唤醒策略优先。', + 'ingest.requirement': '需要应用能收到群里的普通消息。飞书需开通 im:message.group_msg 权限。', + 'ingest.enabled': '接收新消息', + 'ingest.enableFor': '接收外部会话：{name}', + 'ingest.deleteImpact': '停止把这个会话的新消息带进频道。已接入的消息保留。', + 'ingest.exists': '这个会话已经接入到本频道。', + 'ingest.failed': '未能完成操作。请刷新后重试。', + 'ingest.state.waiting': '等待第一条消息', + 'ingest.state.receiving': '正在接收', + 'ingest.state.paused': '已暂停', + 'ingest.state.unavailable': '连接不可用', + 'ingest.lastMessage': '最后消息 {time}', 'bridge.source': '来源', 'bridge.conversation': '外部会话', 'bridge.condition': '接收条件', @@ -2932,6 +2962,41 @@ My relationship with the Human, established lore and topics to avoid: [fill in]. 'bridge.confirmDelete': 'Delete this connector', 'bridge.save': 'Save connector', 'bridge.empty': 'No external sources connected.', + 'ingest.add': 'Connect an external conversation', + 'ingest.addTitle': 'Connect an external conversation', + 'ingest.edit': 'Edit external conversation: {name}', + 'ingest.delete': 'Delete external conversation', + 'ingest.confirmDelete': 'Delete this connection', + 'ingest.save': 'Connect', + 'ingest.update': 'Save', + 'ingest.summary': + 'Bring every message from an external group into this Channel as context for member Bots. Receive only.', + 'ingest.app': 'App', + 'ingest.appHint': '{platform} · used by {bot}', + 'ingest.conversation': 'Conversation', + 'ingest.noApps': + "No app is available yet. Bind an app in a member Bot's External identities first.", + 'ingest.noConversations': + 'This app has no known group yet. @ the app once in the group and it will show up here.', + 'ingest.wake': 'Wake', + 'ingest.wake.mentions': 'Context only', + 'ingest.wake.digest': 'Wake after a batch', + 'ingest.wake.all': 'Wake on every message', + 'ingest.wakeHint': + "With context only, Bots read these messages only when someone @s them. A member's own wake policy in this Channel takes precedence.", + 'ingest.requirement': + 'The app must receive ordinary group messages. Lark needs the im:message.group_msg scope.', + 'ingest.enabled': 'Receive new messages', + 'ingest.enableFor': 'Receive external conversation: {name}', + 'ingest.deleteImpact': + 'New messages from this conversation stop coming into the Channel. Messages already here stay.', + 'ingest.exists': 'This conversation is already connected to this Channel.', + 'ingest.failed': 'Could not finish. Refresh and try again.', + 'ingest.state.waiting': 'Waiting for the first message', + 'ingest.state.receiving': 'Receiving', + 'ingest.state.paused': 'Paused', + 'ingest.state.unavailable': 'Connection unavailable', + 'ingest.lastMessage': 'Last message {time}', 'bridge.source': 'Source', 'bridge.conversation': 'External conversation', 'bridge.condition': 'Conditions', diff --git a/packages/client/test/bot-sidebar.test.ts b/packages/client/test/bot-sidebar.test.ts index fc0654c554..a32e471200 100644 --- a/packages/client/test/bot-sidebar.test.ts +++ b/packages/client/test/bot-sidebar.test.ts @@ -131,6 +131,8 @@ function stubActions(): BridgeActions { }), groupWakePolicies: vi.fn(async () => []), channelBridges: vi.fn(async (channelId) => ({ channelId, bridges: [], sources: [] })), + channelIngests: vi.fn(async (channelId) => ({ channelId, ingests: [], candidates: [] })), + channelIngest: vi.fn(async () => undefined), channelBridge: vi.fn(async () => undefined), messagingChannelTarget: async () => undefined, messagingThreadPolicy: async () => undefined, diff --git a/packages/core/src/bridge/methods.ts b/packages/core/src/bridge/methods.ts index 02725e3d5c..203c2a9dab 100644 --- a/packages/core/src/bridge/methods.ts +++ b/packages/core/src/bridge/methods.ts @@ -13,6 +13,10 @@ import type { OverviewMemory } from '../memory/overview.js'; import type { UsageOverviewBuckets, UsageOverviewResult } from '../usage/overview.js'; import { markAllHumanMessagesRead } from '../channels/mark-all-read.js'; import { channelBridgeInput, type ChannelBridgeSnapshot } from '../messaging/channel-bridge.js'; +import { + conversationIngestInput, + type ConversationIngestSnapshot, +} from '../messaging/conversation-ingest.js'; import { externalMemberWake, messagingDefaultsInput, @@ -322,6 +326,8 @@ export interface BridgeMethods { >; pairingReview(payload: unknown): Promise>; channelBridges(payload: unknown): Promise>; + channelIngests(payload: unknown): Promise>; + channelIngest(payload: unknown): Promise>; channelBridge(payload: unknown): Promise>; messagingIdentity(payload: unknown): Promise>; messagingChannelTarget(payload: unknown): Promise>; @@ -947,6 +953,26 @@ export function createBridgeMethods(deps: BridgeMethodsDeps): BridgeMethods { if (!input.success) return Promise.resolve(invalidInput('Known Group Channel required')); return messagingCall((service) => service.channelBridges(input.data.channelId)); }, + channelIngests(payload) { + const input = z + .object({ channelId: z.string().min(1).max(128) }) + .strict() + .safeParse(payload); + if (!input.success) return Promise.resolve(invalidInput('Known Group Channel required')); + return messagingCall(async (service) => service.inbound.ingests(input.data.channelId)); + }, + channelIngest(payload) { + const input = z + .object({ channelId: z.string().min(1).max(128), input: conversationIngestInput }) + .strict() + .safeParse(payload); + if (!input.success) + return Promise.resolve(invalidInput('Invalid external conversation command')); + return messagingCall(async (service) => { + await service.inbound.ingest(input.data.channelId, input.data.input); + return { updated: true as const }; + }); + }, channelBridge(payload) { const input = z .object({ channelId: z.string().min(1).max(128), input: channelBridgeInput }) diff --git a/packages/core/src/bridge/rpc.ts b/packages/core/src/bridge/rpc.ts index 14db6e3169..a7513d01f2 100644 --- a/packages/core/src/bridge/rpc.ts +++ b/packages/core/src/bridge/rpc.ts @@ -10,6 +10,10 @@ import type { import type { MessagingIdentity, MessagingIdentityInput } from '../messaging/identity.js'; import type { MessagingConversationInput } from '../messaging/conversations.js'; import type { ChannelBridgeInput, ChannelBridgeSnapshot } from '../messaging/channel-bridge.js'; +import type { + ConversationIngestInput, + ConversationIngestSnapshot, +} from '../messaging/conversation-ingest.js'; import type { UsageOverviewPeriod } from '../usage/overview.js'; import type { MarketplaceDetail, @@ -193,6 +197,12 @@ export class BotharnessBridgeService extends TypertRemoteService { channelBridges(channelId: string): Promise { return unwrapAsync(this.methods.channelBridges({ channelId })); } + channelIngests(channelId: string): Promise { + return unwrapAsync(this.methods.channelIngests({ channelId })); + } + channelIngest(channelId: string, input: ConversationIngestInput): Promise<{ updated: true }> { + return unwrapAsync(this.methods.channelIngest({ channelId, input })); + } channelBridge(channelId: string, input: ChannelBridgeInput): Promise<{ updated: true }> { return unwrapAsync(this.methods.channelBridge({ channelId, input })); } @@ -1012,6 +1022,8 @@ markRemoteMethods(BotharnessBridgeService.prototype, [ 'messagingDefaultsSet', 'channelBridges', 'channelBridge', + 'channelIngests', + 'channelIngest', 'messagingChannelTarget', 'messagingReceive', 'messagingGroupPolicy', diff --git a/packages/core/src/database/schema-plan.ts b/packages/core/src/database/schema-plan.ts index b25435e96f..910b1a37ab 100644 --- a/packages/core/src/database/schema-plan.ts +++ b/packages/core/src/database/schema-plan.ts @@ -1678,6 +1678,31 @@ const SEVERAL_APPS_MIGRATION: SchemaMigration = { }, }; +const CONVERSATION_INGEST_MIGRATION: SchemaMigration = { + generation: 64, + module: 'messaging', + description: 'Let a Channel ingest an external conversation as one-way context', + migrate(database) { + database.exec(` + CREATE TABLE messaging_conversation_ingests ( + id TEXT PRIMARY KEY, + channel_id TEXT NOT NULL, + provider_id TEXT NOT NULL, + account_ref TEXT NOT NULL, + fingerprint TEXT NOT NULL, + conversation_kind TEXT NOT NULL CHECK (conversation_kind IN ('dm', 'group')), + conversation_id TEXT NOT NULL, + revision INTEGER NOT NULL CHECK (revision > 0), + body TEXT NOT NULL CHECK (json_valid(body)) + ); + CREATE UNIQUE INDEX messaging_conversation_ingest_target ON messaging_conversation_ingests + (channel_id, provider_id, fingerprint, conversation_kind, conversation_id); + CREATE INDEX messaging_conversation_ingest_source ON messaging_conversation_ingests + (provider_id, fingerprint, conversation_kind, conversation_id); + `); + }, +}; + export const BOT_HARNESS_SCHEMA_PLAN = defineSchemaPlan([ SESSION_OWNERSHIP_MIGRATION, MESSAGING_TRACER_MIGRATION, @@ -1741,4 +1766,5 @@ export const BOT_HARNESS_SCHEMA_PLAN = defineSchemaPlan([ BOUND_APP_ADMISSION_MIGRATION, CONVERSATION_LIST_MIGRATION, SEVERAL_APPS_MIGRATION, + CONVERSATION_INGEST_MIGRATION, ]); diff --git a/packages/core/src/messaging/channel-target.ts b/packages/core/src/messaging/channel-target.ts index 65703b7fe8..a386bf513c 100644 --- a/packages/core/src/messaging/channel-target.ts +++ b/packages/core/src/messaging/channel-target.ts @@ -98,10 +98,11 @@ export function projectBridgeMessage(source: ExternalSource, body: string): Chan export function placeBridgeSource( db: DatabaseSync, source: ExternalSource, - botSlug: string, + botSlug: string | undefined, ): { channelId: string; message: ChannelMessage; revision: number } | undefined { if (!source.localChannelId) return; - bridgeChannel(db, source.localChannelId, botSlug); + if (botSlug === undefined) humanBridgeChannel(db, source.localChannelId); + else bridgeChannel(db, source.localChannelId, botSlug); const prior = db .prepare( 'SELECT channel_id FROM channel_placements WHERE source_event_id = ? AND channel_id = ?', diff --git a/packages/core/src/messaging/conversation-ingest.ts b/packages/core/src/messaging/conversation-ingest.ts new file mode 100644 index 0000000000..f439889b11 --- /dev/null +++ b/packages/core/src/messaging/conversation-ingest.ts @@ -0,0 +1,187 @@ +import { z } from 'zod'; +import type { DatabaseSync } from 'node:sqlite'; +import type { ChannelRecord } from '../channels/channel.js'; +import type { BotSourcePolicyStore } from '../runtime/source-policy.js'; +import type { ConversationRef } from './conversations.js'; +import { MessagingError } from './provider.js'; + +export interface ConversationIngestWake { + mode: 'mentions' | 'digest' | 'all'; + count: number; + intervalSeconds: number; +} + +export interface ConversationIngest { + id: string; + channelId: string; + providerId: string; + accountRef: string; + fingerprint: string; + platform: string; + accountName: string; + conversation: ConversationRef & { name: string }; + revision: number; + enabled: boolean; + intakeAfter: string; + wake: ConversationIngestWake; + createdAt: string; + lastMessageAt?: string; +} + +export interface ConversationIngestRow extends ConversationIngest { + botSlug?: string; + state: 'waiting' | 'receiving' | 'paused' | 'unavailable'; +} + +export interface ConversationIngestCandidate { + bindingId: string; + botSlug: string; + platform: string; + accountName: string; + conversations: (ConversationRef & { name: string })[]; +} + +export interface ConversationIngestSnapshot { + channelId: string; + ingests: ConversationIngestRow[]; + candidates: ConversationIngestCandidate[]; +} + +export const DEFAULT_INGEST_WAKE: ConversationIngestWake = { + mode: 'mentions', + count: 10, + intervalSeconds: 300, +}; + +const wake = z + .object({ + mode: z.enum(['mentions', 'digest', 'all']), + count: z.number().int().min(1).max(100), + intervalSeconds: z + .number() + .int() + .min(10) + .max(24 * 60 * 60), + }) + .strict(); + +export const conversationIngestInput = z.discriminatedUnion('kind', [ + z + .object({ + kind: z.literal('add'), + bindingId: z.string().min(1).max(128), + conversation: z.object({ kind: z.literal('group'), id: z.string().min(1).max(256) }).strict(), + wake: wake.optional(), + }) + .strict(), + z + .object({ + kind: z.literal('update'), + ingestId: z.string().uuid(), + expectedRevision: z.number().int().positive(), + enabled: z.boolean().optional(), + wake: wake.optional(), + }) + .strict(), + z + .object({ + kind: z.literal('delete'), + ingestId: z.string().uuid(), + expectedRevision: z.number().int().positive(), + }) + .strict(), +]); +export type ConversationIngestInput = z.infer; + +export function channelIngests(db: DatabaseSync, channelId: string): ConversationIngest[] { + return ( + db + .prepare( + 'SELECT body FROM messaging_conversation_ingests WHERE channel_id = ? ORDER BY rowid', + ) + .all(channelId) as { body: string }[] + ).map((row) => JSON.parse(row.body) as ConversationIngest); +} + +export function readIngest(db: DatabaseSync, id: string): ConversationIngest { + const row = db.prepare('SELECT body FROM messaging_conversation_ingests WHERE id = ?').get(id) as + | { body: string } + | undefined; + if (!row) throw new MessagingError('ingest-unavailable'); + return JSON.parse(row.body) as ConversationIngest; +} + +export function matchingIngests( + db: DatabaseSync, + providerId: string, + fingerprint: string, + conversation: ConversationRef, +): ConversationIngest[] { + return ( + db + .prepare( + `SELECT body FROM messaging_conversation_ingests + WHERE provider_id = ? AND fingerprint = ? AND conversation_kind = ? AND conversation_id = ? + ORDER BY rowid`, + ) + .all(providerId, fingerprint, conversation.kind, conversation.id) as { body: string }[] + ).map((row) => JSON.parse(row.body) as ConversationIngest); +} + +export function writeIngest(db: DatabaseSync, value: ConversationIngest): void { + db.prepare( + `INSERT INTO messaging_conversation_ingests + (id, channel_id, provider_id, account_ref, fingerprint, conversation_kind, conversation_id, revision, body) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (id) DO UPDATE SET revision = excluded.revision, body = excluded.body`, + ).run( + value.id, + value.channelId, + value.providerId, + value.accountRef, + value.fingerprint, + value.conversation.kind, + value.conversation.id, + value.revision, + JSON.stringify(value), + ); +} + +export function deleteIngest(db: DatabaseSync, id: string): void { + db.prepare('DELETE FROM messaging_conversation_ingests WHERE id = ?').run(id); +} + +export function admitIngestMembers( + db: DatabaseSync, + sourceEventId: string, + channel: ChannelRecord, + ingest: ConversationIngest, + sourcePolicy: BotSourcePolicyStore, + isBotActive: (botSlug: string) => boolean, +): string[] { + const admitted: string[] = []; + for (const botSlug of channel.members) { + if (!isBotActive(botSlug)) continue; + const rule = sourcePolicy.resolveIn(db, botSlug, 'group-ordinary'); + const custom = channel.wakePolicies?.[botSlug]; + const policy = custom ?? { ...ingest.wake, revision: ingest.revision }; + const mode = policy.mode; + const result = db + .prepare(`INSERT OR IGNORE INTO inbox_admissions + (source_event_id, bot_slug, reason, source_policy_revision, source_policy_wake_mode, + wake_policy_revision, wake_mode, wake_count, wake_interval_ms) + VALUES (?, ?, 'group-ordinary', ?, ?, ?, ?, ?, ?)`) + .run( + sourceEventId, + botSlug, + rule.revision, + rule.wake, + policy.revision, + mode, + mode === 'all' ? 1 : mode === 'digest' ? policy.count : null, + mode === 'all' ? 0 : mode === 'digest' ? policy.intervalSeconds * 1000 : null, + ); + if (result.changes > 0) admitted.push(botSlug); + } + return admitted; +} diff --git a/packages/core/src/messaging/inbound.ts b/packages/core/src/messaging/inbound.ts index 55869a0af1..7fb09445df 100644 --- a/packages/core/src/messaging/inbound.ts +++ b/packages/core/src/messaging/inbound.ts @@ -1,4 +1,24 @@ -import { admissionBound, conversationName, hold, readBlock, removeHeld } from './conversations.js'; +import { + admissionBound, + conversationName, + hold, + listHeld, + readBlock, + removeHeld, +} from './conversations.js'; +import { + DEFAULT_INGEST_WAKE, + admitIngestMembers, + channelIngests, + deleteIngest, + matchingIngests, + readIngest, + writeIngest, + type ConversationIngest, + type ConversationIngestInput, + type ConversationIngestSnapshot, +} from './conversation-ingest.js'; +import type { MessagingIdentity } from './identity.js'; import type { BotPairing } from './pairing.js'; import { readMessagingIdentity } from './identity.js'; import { @@ -29,7 +49,7 @@ import { type GroupReceptionPolicy, } from './group-policy.js'; import type { BotSourcePolicyEditor } from '../runtime/source-policy.js'; -import { bridgeChannel, placeBridgeSource } from './channel-target.js'; +import { bridgeChannel, humanBridgeChannel, placeBridgeSource } from './channel-target.js'; import { admitBridgeMembers } from './member-admission.js'; import { messagingDefaults } from './defaults.js'; import { assertMessagingIdentity } from './identity.js'; @@ -39,6 +59,7 @@ import { type ChannelBridgeInput, } from './channel-bridge.js'; import type { ChannelMessageCommit } from '../channels/store.js'; +import type { ChannelRecord } from '../channels/channel.js'; import { createHash, randomUUID } from 'node:crypto'; import { relatedReport, recordReportEcho, type RelatedReport } from './report.js'; import type { DatabaseSync } from 'node:sqlite'; @@ -118,6 +139,8 @@ export interface ExternalSource { event: Omit; grantId: string; grantRevision: number; + ingestId?: string; + ingestRevision?: number; defaultRevision?: number; receptionRevision?: number; bridgeRevision?: number; @@ -141,6 +164,8 @@ export interface InboundMessaging { register(provider: MessagingProvider): () => void; setEnabled(botSlug: string, grantId: string, enabled: boolean): Promise; channelBridge(channelId: string, input: ChannelBridgeInput): Promise; + ingests(channelId: string): ConversationIngestSnapshot; + ingest(channelId: string, input: ConversationIngestInput): Promise; setChannelTarget(botSlug: string, grantId: string, channelId: string | null): Promise; policy(botSlug: string, grantId: string): GroupReceptionPolicy; setPolicy( @@ -198,6 +223,7 @@ export function createInboundMessaging(options: { onReceptionChanged?(): void; onPlaced?(commit: ChannelMessageCommit): void; onShared?(botSlugs: string[]): void; + onIngested?(channelId: string, messageId: string): void; warn?(message: string): void; }): InboundMessaging { const { database } = options; @@ -256,7 +282,10 @@ export function createInboundMessaging(options: { } >(); const retainedCursors = new Map(); - const sourceId = (value: MessagingGrant, event: MessagingInboundEvent) => + const sourceId = ( + value: Pick, + event: MessagingInboundEvent, + ) => 'im-' + createHash('sha256') .update( @@ -270,10 +299,10 @@ export function createInboundMessaging(options: { .digest('hex'); const persistSource = ( db: DatabaseSync, - value: MessagingGrant, + value: MessagingGrant | { ingest: ConversationIngest; botSlug: string }, event: MessagingInboundEvent, ): string => { - const id = sourceId(value, event); + const id = sourceId('ingest' in value ? value.ingest : value, event); const existing = db .prepare('SELECT body, payload_json FROM source_events WHERE source_event_id = ?') .get(id) as { body: string; payload_json: string } | undefined; @@ -315,24 +344,39 @@ export function createInboundMessaging(options: { } } else { const { text: _text, ...evidence } = event; - const external: Omit = { - id, - at: event.at, - platform: event.channel, - accountName: value.accountName, - conversationName: value.targetName, - event: evidence, - grantId: value.id, - grantRevision: value.revision, - defaultRevision: messagingDefaults(db, value.platform).revision, - ...(event.conversation.kind === 'group' - ? { receptionRevision: groupReceptionPolicy(db, value.id).revision } - : {}), - ...(value.channelBridge ? { bridgeRevision: value.channelBridge.revision } : {}), - ...(!value.bridgeRoutes && value.receiveTargetChannelId - ? { localChannelId: value.receiveTargetChannelId } - : {}), - }; + const external: Omit = + 'ingest' in value + ? { + id, + at: event.at, + platform: event.channel, + accountName: value.ingest.accountName, + conversationName: value.ingest.conversation.name, + event: evidence, + grantId: '', + grantRevision: 0, + ingestId: value.ingest.id, + ingestRevision: value.ingest.revision, + defaultRevision: messagingDefaults(db, value.ingest.platform).revision, + } + : { + id, + at: event.at, + platform: event.channel, + accountName: value.accountName, + conversationName: value.targetName, + event: evidence, + grantId: value.id, + grantRevision: value.revision, + defaultRevision: messagingDefaults(db, value.platform).revision, + ...(event.conversation.kind === 'group' + ? { receptionRevision: groupReceptionPolicy(db, value.id).revision } + : {}), + ...(value.channelBridge ? { bridgeRevision: value.channelBridge.revision } : {}), + ...(!value.bridgeRoutes && value.receiveTargetChannelId + ? { localChannelId: value.receiveTargetChannelId } + : {}), + }; db.prepare(`INSERT INTO source_events (source_event_id, source_kind, bot_slug, body, created_at, payload_json) VALUES (?, 'bridge-message', ?, ?, ?, ?)`).run( id, @@ -994,6 +1038,7 @@ export function createInboundMessaging(options: { ) as { payload_json: string; body: string } | undefined; if (!row) throw new MessagingError('source-unavailable'); const retained = (JSON.parse(row.payload_json) as { external: ExternalSource }).external; + if (retained.ingestId) throw new MessagingError('source-unavailable'); const routed = database.read((db) => db.prepare('SELECT 1 FROM messaging_source_paths WHERE source_event_id = ?').get(id), ); @@ -1087,7 +1132,7 @@ export function createInboundMessaging(options: { ) as { payload_json: string; body: string } | undefined; if (!row) throw new MessagingError('source-unavailable'); const retained = (JSON.parse(row.payload_json) as { external: ExternalSource }).external; - if (grant(retained.grantId).botSlug === botSlug) return read(botSlug, id); + if (!retained.ingestId && grant(retained.grantId).botSlug === botSlug) return read(botSlug, id); const channelId = database.read((db) => { const placements = db .prepare( @@ -1143,6 +1188,73 @@ export function createInboundMessaging(options: { }), ); }; + const placeIngests = (identity: MessagingIdentity, event: MessagingInboundEvent) => { + if (event.conversation.kind !== 'group') return; + const current = (db: DatabaseSync) => + matchingIngests(db, identity.providerId, identity.fingerprint, event.conversation).filter( + (item) => + item.enabled && + item.accountRef === identity.accountRef && + Date.parse(event.at) >= Date.parse(item.intakeAfter), + ); + if (!database.read(current).length) return; + const placed: ChannelMessageCommit[] = []; + const members = new Set(); + transaction( + (db) => { + for (const ingest of current(db)) { + let channel: ChannelRecord; + try { + channel = humanBridgeChannel(db, ingest.channelId); + } catch (error) { + if (!(error instanceof MessagingError)) throw error; + continue; + } + if (channel.type !== 'group') continue; + const id = persistSource(db, { ingest, botSlug: identity.botSlug }, event); + const row = db + .prepare('SELECT payload_json FROM source_events WHERE source_event_id = ?') + .get(id) as { payload_json: string }; + const source = (JSON.parse(row.payload_json) as { external: ExternalSource }).external; + const commit = placeBridgeSource( + db, + { ...source, body: event.text, localChannelId: ingest.channelId }, + undefined, + ); + if (!commit) continue; + placed.push(commit); + for (const slug of admitIngestMembers( + db, + id, + channel, + ingest, + options.sourcePolicy, + options.isBotActive, + )) + members.add(slug); + writeIngest(db, { ...ingest, lastMessageAt: event.at }); + } + }, + ['source-event', 'channel', 'bot-inbox'], + ); + for (const commit of placed) { + try { + options.onPlaced?.(commit); + } catch { + options.warn?.('ingest-channel-publication-failed'); + } + try { + options.onIngested?.(commit.channelId, commit.message.id); + } catch { + options.warn?.('ingest-admission-wake-failed'); + } + } + try { + if (members.size) options.onShared?.([...members]); + } catch { + options.warn?.('ingest-share-wake-failed'); + } + }; const defaultTraffic = async ( id: string, lease: { controller: AbortController; token: object }, @@ -1165,7 +1277,22 @@ export function createInboundMessaging(options: { throw new MessagingError('consumer-unavailable'); if (event.fingerprint !== identity.fingerprint || event.botId !== identity.accountRef) throw new MessagingError('untrusted-source'); - if (event.conversation.kind === 'group' && !event.mentionedAccount) return { accepted: true }; + if (event.conversation.kind === 'group' && !event.mentionedAccount) { + placeIngests(identity, event); + return { accepted: true }; + } + const result = await admitDefault(id, lease, identity, entry, event, signal); + placeIngests(identity, event); + return result; + }; + const admitDefault = async ( + id: string, + lease: { controller: AbortController; token: object }, + identity: MessagingIdentity, + entry: NonNullable>, + event: MessagingInboundEvent, + signal: AbortSignal, + ): Promise<{ accepted: true }> => { if ( identity.platform === 'weixin' && (event.conversation.kind !== 'dm' || @@ -1460,6 +1587,139 @@ export function createInboundMessaging(options: { for (const [id, retry] of retries) if (retry.token === token) stop(id); }; }, + ingests(channelId) { + return database.read((db) => { + humanBridgeChannel(db, channelId); + const bindings = ( + db + .prepare( + "SELECT id FROM messaging_bindings WHERE revoked_at IS NULL AND platform <> 'weixin' ORDER BY created_at", + ) + .all() as { id: string }[] + ).map((row) => readMessagingIdentity(db, row.id)); + const state = ( + ingest: ConversationIngest, + ): ConversationIngestSnapshot['ingests'][number]['state'] => { + if (!ingest.enabled) return 'paused'; + const binding = bindings.find( + (item) => + item.providerId === ingest.providerId && + item.fingerprint === ingest.fingerprint && + item.accountRef === ingest.accountRef, + ); + if (!binding || service.pairingReception(binding.id) !== 'receiving') + return 'unavailable'; + return ingest.lastMessageAt ? 'receiving' : 'waiting'; + }; + return { + channelId, + ingests: channelIngests(db, channelId).map((ingest) => { + const binding = bindings.find( + (item) => + item.providerId === ingest.providerId && item.fingerprint === ingest.fingerprint, + ); + return { + ...ingest, + ...(binding ? { botSlug: binding.botSlug } : {}), + state: state(ingest), + }; + }), + candidates: bindings.map((binding) => { + const known = new Map(); + for (const row of db + .prepare( + `SELECT body FROM messaging_grants WHERE binding_id = ? AND revoked_at IS NULL + AND json_extract(body, '$.receiveScope.kind') = 'group' ORDER BY created_at`, + ) + .all(binding.id) as { body: string }[]) { + const value = JSON.parse(row.body) as MessagingGrant; + if (value.receiveScope) + known.set(value.receiveScope.conversationId, value.targetName); + } + for (const held of listHeld(db, binding.id)) + if (held.conversation.kind === 'group' && !known.has(held.conversation.id)) + known.set(held.conversation.id, held.name); + return { + bindingId: binding.id, + botSlug: binding.botSlug, + platform: binding.platform, + accountName: binding.name, + conversations: [...known].map(([id, name]) => ({ kind: 'group' as const, id, name })), + }; + }), + }; + }); + }, + async ingest(channelId, input) { + transaction( + (db) => { + const channel = humanBridgeChannel(db, channelId); + if (channel.type !== 'group') throw new MessagingError('channel-unavailable'); + if (input.kind === 'add') { + const identity = readMessagingIdentity(db, input.bindingId); + if (identity.revokedAt || identity.platform === 'weixin') + throw new MessagingError('identity-unavailable'); + const entry = conversationEntries(db, identity.id, input.conversation).find( + (item) => item.receiveScope?.kind === 'group', + ); + const held = listHeld(db, identity.id).find( + (item) => + item.conversation.kind === 'group' && + item.conversation.id === input.conversation.id, + ); + if (!entry && !held) throw new MessagingError('conversation-unavailable'); + if ( + matchingIngests( + db, + identity.providerId, + identity.fingerprint, + input.conversation, + ).some((item) => item.channelId === channelId) + ) + throw new MessagingError('ingest-exists'); + const at = new Date().toISOString(); + writeIngest(db, { + id: randomUUID(), + channelId, + providerId: identity.providerId, + accountRef: identity.accountRef, + fingerprint: identity.fingerprint, + platform: identity.platform, + accountName: identity.name, + conversation: { + kind: 'group', + id: input.conversation.id, + name: entry?.targetName ?? held?.name ?? input.conversation.id, + }, + revision: 1, + enabled: true, + intakeAfter: at, + wake: input.wake ?? DEFAULT_INGEST_WAKE, + createdAt: at, + }); + return; + } + const value = readIngest(db, input.ingestId); + if (value.channelId !== channelId) throw new MessagingError('ingest-unavailable'); + if (value.revision !== input.expectedRevision) throw new MessagingError('ingest-changed'); + if (input.kind === 'delete') { + deleteIngest(db, value.id); + return; + } + writeIngest(db, { + ...value, + revision: value.revision + 1, + ...(input.enabled === undefined ? {} : { enabled: input.enabled }), + ...(input.enabled === true && !value.enabled + ? { intakeAfter: new Date().toISOString() } + : {}), + ...(input.wake ? { wake: input.wake } : {}), + }); + }, + ['grants'], + ); + receptionChanged(); + }, async channelBridge(channelId, rawInput) { const input = channelBridgeInput.parse(rawInput); const value = grant(input.grantId); diff --git a/packages/core/src/messaging/outbound.ts b/packages/core/src/messaging/outbound.ts index f8e044c28e..01b687672c 100644 --- a/packages/core/src/messaging/outbound.ts +++ b/packages/core/src/messaging/outbound.ts @@ -294,6 +294,7 @@ export function createOutboundMessaging(options: { onAdmitted?(botSlug: string, sourceEventId: string): void; onPlaced?(commit: ChannelMessageCommit): void; onShared?(botSlugs: string[]): void; + onIngested?(channelId: string, messageId: string): void; timeoutMs?: number; recover?: boolean; now?: () => Date; @@ -560,6 +561,7 @@ export function createOutboundMessaging(options: { ...(options.onReceptionChanged ? { onReceptionChanged: options.onReceptionChanged } : {}), ...(options.onPlaced ? { onPlaced: options.onPlaced } : {}), ...(options.onShared ? { onShared: options.onShared } : {}), + ...(options.onIngested ? { onIngested: options.onIngested } : {}), ...(options.warn === undefined ? {} : { warn: options.warn }), }); const sourceForReply = (botSlug: string, sourceEventId: string, value: MessagingGrant) => { diff --git a/packages/core/src/plugin.ts b/packages/core/src/plugin.ts index 5a16dd4422..c52f0f007d 100644 --- a/packages/core/src/plugin.ts +++ b/packages/core/src/plugin.ts @@ -388,6 +388,7 @@ export function createCore( onShared: (slugs) => { for (const slug of slugs) runtime?.resumePendingDigests?.(slug); }, + onIngested: (channelId, messageId) => runtime?.admitGroupMessage(channelId, messageId), recover: operationalDatabase.mode === 'ready', isBotActive: (slug) => { const bot = registry.get(slug); diff --git a/packages/core/test/bridge-rpc.test.ts b/packages/core/test/bridge-rpc.test.ts index daa7ea4446..79cedf2303 100644 --- a/packages/core/test/bridge-rpc.test.ts +++ b/packages/core/test/bridge-rpc.test.ts @@ -79,6 +79,8 @@ describe('bridge typert service', () => { 'messagingDefaultsSet', 'channelBridges', 'channelBridge', + 'channelIngests', + 'channelIngest', 'messagingChannelTarget', 'messagingReceive', 'messagingGroupPolicy', diff --git a/packages/core/test/messaging-conversation-ingest.test.ts b/packages/core/test/messaging-conversation-ingest.test.ts new file mode 100644 index 0000000000..8ebad627f2 --- /dev/null +++ b/packages/core/test/messaging-conversation-ingest.test.ts @@ -0,0 +1,326 @@ +import { setImmediate as tick } from 'node:timers/promises'; +import { afterEach, expect, it } from 'vitest'; +import { createCore, type BotHarnessCore } from '../src/plugin.js'; +import { attachOperationalModule } from '../src/database/owner.js'; +import { createDshImProvider, type DshImOutboundService } from '../src/messaging/dsh-im.js'; +import type { MessagingInboundEvent, MessagingReplyRoute } from '../src/messaging/provider.js'; +import type { OrchestratorAgentRun, BotAgentAdapter } from '../src/runtime/bot-runtime.js'; +import { createTempRoot } from './helpers.js'; + +const fingerprint = 'c'.repeat(64); +const cores: BotHarnessCore[] = []; +afterEach(async () => { + for (const core of cores.splice(0)) { + core.externalMessaging.close(); + await core.runtime.close(); + core.operationalDatabase.close(); + } +}); + +function dm(id: string, overrides: Partial = {}): MessagingInboundEvent { + return { + version: 1, + channel: 'feishu', + botId: 'lark-app', + fingerprint, + eventId: `ev-${id}`, + messageId: `om-${id}`, + actor: { kind: 'user', id: 'ou_owner', name: 'Owner' }, + conversation: { kind: 'dm', id: 'oc_owner' }, + mentions: [], + mentionedAccount: false, + at: '2026-10-07T00:00:00.000Z', + text: `hello ${id}`, + reply: { messageId: `om-${id}`, conversationId: 'oc_owner', actorId: 'ou_owner' }, + replay: { kind: 'provider-redelivery', resumeCursor: false, gapPossible: true }, + ...overrides, + }; +} + +function mention(id: string, mentioned = true): MessagingInboundEvent { + return dm(id, { + conversation: { kind: 'group', id: 'oc_team' }, + mentions: mentioned ? [{ id: 'ou-bot', key: '@_user_1' }] : [], + mentionedAccount: mentioned, + text: mentioned ? `@_user_1 question ${id}` : `chatter ${id}`, + reply: { + messageId: `om-${id}`, + conversationId: 'oc_team', + actorId: 'ou_owner', + threadId: 'omt-topic', + rootId: 'om-root', + }, + }); +} + +type Platform = 'feishu' | 'slack' | 'discord' | 'weixin'; + +async function fixture( + platform: Platform = 'feishu', + targets: Awaited> = [], +) { + const home = createTempRoot('botharness-default-traffic-'); + const runs: OrchestratorAgentRun[] = []; + const agents: BotAgentAdapter = { + async runOrchestrator(run) { + runs.push(run); + }, + async runAssignment() {}, + requestAssignment() { + return { delivery: 'steer' }; + }, + async stopAssignment() {}, + async close() {}, + }; + let core = createCore({ dshHome: home, agents }); + cores.push(core); + expect(core.registry.create({ slug: 'ada', displayName: 'Ada' }).ok).toBe(true); + type Consumer = Parameters>[1]; + let consumer: Consumer | undefined; + let subscriptions = 0; + const replies: { route: MessagingReplyRoute; text: string }[] = []; + const service: DshImOutboundService = { + contractVersion: 1, + replyContextVersion: 1, + replyReceiptVersion: 1, + replyFenceVersion: 1, + listBots: async () => [{ botId: 'lark-app', channel: platform }], + listTargets: async () => targets, + describeBot: async (botId) => ({ + version: 1, + channel: platform, + botId, + account: { fingerprint, name: 'Support Lark app' }, + connected: true, + capabilities: [ + 'proactive-text-checked', + 'exclusive-text-consumer', + 'ordinary-text-consumer', + 'reply-text-checked', + 'reply-context-checked', + 'reply-receipt-checked', + 'reply-fence-checked', + ], + }), + sendChecked: async () => ({ sent: true as const }), + consumeInbound: async (_id, input) => { + ++subscriptions; + consumer = input; + return () => { + --subscriptions; + }; + }, + qualifyReplyChecked: async (_botId, route) => route, + replyChecked: async (_botId, route, text, options) => { + if (options.beforeSend && !options.beforeSend()) + throw Object.assign(new Error('stale-route'), { code: 'stale-route' }); + replies.push({ route, text }); + return { + sent: true, + receipt: { + version: 1, + messageId: 'reply-' + route.messageId, + conversationId: route.conversationId, + }, + }; + }, + }; + const register = () => core.externalMessaging.register(createDshImProvider(service, platform)!); + let dispose = register(); + const settle = async () => { + for (let i = 0; i < 4; i++) await tick(); + await core.runtime.whenIdle(); + await tick(); + }; + const identity = await core.externalMessaging.identity('ada', { + kind: 'bind', + providerId: `dsh-im/${platform}`, + accountRef: 'lark-app', + fingerprint, + }); + await settle(); + const query = (sql: string) => + attachOperationalModule(core.operationalDatabase, 'test').read((db) => db.prepare(sql).all()); + return { + get core() { + return core; + }, + identity, + runs, + replies, + query, + settle, + get subscriptions() { + return subscriptions; + }, + async receive(event: MessagingInboundEvent) { + if (!consumer) throw new Error('Not subscribed'); + const result = await consumer.onEvent( + { ...event, channel: platform }, + { signal: consumer.signal }, + ); + await settle(); + return result; + }, + entries() { + return query( + "SELECT json_extract(body, '$.origin') AS origin, json_extract(body, '$.receiveScope.kind') AS kind, json_extract(body, '$.receiveScope.conversationId') AS conversation, revoked_at FROM messaging_grants ORDER BY created_at", + ); + }, + admissions() { + return query( + "SELECT a.reason, json_extract(e.payload_json, '$.external.event.messageId') AS messageId FROM inbox_admissions a JOIN source_events e USING(source_event_id) ORDER BY e.created_at, messageId", + ); + }, + sourceId(messageId: string) { + return ( + query( + `SELECT source_event_id FROM source_events WHERE json_extract(payload_json, '$.external.event.messageId') = '${messageId}'`, + )[0] as { source_event_id: string } + ).source_event_id; + }, + async update(input: { enabled?: boolean; newConversations?: 'auto' | 'ask' }) { + const current = (await core.externalMessaging.snapshot('ada')).identities![0]!; + await core.externalMessaging.identity('ada', { + kind: 'update', + id: current.id, + expectedRevision: current.revision, + name: current.name, + enabled: input.enabled ?? current.enabled, + ...(input.newConversations ? { newConversations: input.newConversations } : {}), + }); + await settle(); + }, + async restart() { + dispose(); + core.externalMessaging.close(); + await core.runtime.close(); + core.operationalDatabase.close(); + cores.splice(cores.indexOf(core), 1); + core = createCore({ dshHome: home, agents }); + cores.push(core); + dispose = register(); + await settle(); + }, + }; +} + +function chatter(id: string, mentioned = false): MessagingInboundEvent { + return { ...mention(id, mentioned), at: new Date(Date.now() + 1000).toISOString() }; +} + +async function connected() { + const fx = await fixture(); + expect(fx.core.registry.create({ slug: 'bea', displayName: 'Bea' }).ok).toBe(true); + await fx.receive(mention('first')); + const group = fx.core.channels.createGroup({ name: 'Team', members: ['ada', 'bea'] }); + const before = fx.core.externalMessaging.inbound.ingests(group.id); + expect(before.ingests).toEqual([]); + expect(before.candidates).toEqual([ + { + bindingId: fx.identity.id, + botSlug: 'ada', + platform: 'feishu', + accountName: expect.any(String), + conversations: [{ kind: 'group', id: 'oc_team', name: 'oc_team' }], + }, + ]); + await fx.core.externalMessaging.inbound.ingest(group.id, { + kind: 'add', + bindingId: fx.identity.id, + conversation: { kind: 'group', id: 'oc_team' }, + }); + await fx.settle(); + const placed = () => + fx.query( + `SELECT e.body, json_extract(e.payload_json, '$.external.ingestId') AS ingest + FROM channel_placements p JOIN source_events e USING(source_event_id) + WHERE p.channel_id = '${group.id}' ORDER BY p.revision`, + ) as { body: string; ingest: string | null }[]; + const ingest = () => fx.core.externalMessaging.inbound.ingests(group.id).ingests[0]!; + return { fx, group, placed, ingest }; +} + +it('streams every message of a connected group into the Channel as context without waking Bots', async () => { + const { fx, placed, ingest } = await connected(); + expect(ingest()).toMatchObject({ state: 'waiting', enabled: true, wake: { mode: 'mentions' } }); + for (const id of ['a', 'b', 'c']) await fx.receive(chatter(id)); + expect(placed().map((row) => row.body)).toEqual(['chatter a', 'chatter b', 'chatter c']); + expect(placed().every((row) => row.ingest === ingest().id)).toBe(true); + expect( + fx.query( + `SELECT bot_slug, reason, wake_mode, wake_count FROM inbox_admissions + WHERE source_event_id IN (SELECT source_event_id FROM channel_placements) ORDER BY bot_slug`, + ), + ).toEqual( + ['ada', 'ada', 'ada', 'bea', 'bea', 'bea'].map((bot_slug) => ({ + bot_slug, + reason: 'group-ordinary', + wake_mode: 'mentions', + wake_count: null, + })), + ); + expect(fx.runs).toHaveLength(1); + expect(ingest()).toMatchObject({ state: 'receiving', lastMessageAt: expect.any(String) }); +}); + +it('places a mention once and keeps it answerable through the bound app', async () => { + const { fx, placed } = await connected(); + await fx.receive(chatter('ask', true)); + expect(placed()).toEqual([{ body: '@_user_1 question ask', ingest: null }]); + expect(fx.runs).toHaveLength(2); + await fx.core.externalMessaging.reply('ada', fx.sourceId('om-ask'), 'Answer'); + await fx.settle(); + expect(fx.replies.at(-1)?.text).toBe('Answer'); +}); + +it('keeps streaming after a restart and stops on pause and delete while history stays', async () => { + const { fx, group, placed, ingest } = await connected(); + await fx.receive(chatter('a')); + await fx.restart(); + await fx.receive(chatter('b')); + expect(placed()).toHaveLength(2); + const inbound = () => fx.core.externalMessaging.inbound; + await inbound().ingest(group.id, { + kind: 'update', + ingestId: ingest().id, + expectedRevision: ingest().revision, + enabled: false, + }); + expect(ingest().state).toBe('paused'); + await fx.receive(chatter('c')); + expect(placed()).toHaveLength(2); + await expect( + inbound().ingest(group.id, { + kind: 'delete', + ingestId: ingest().id, + expectedRevision: 1, + }), + ).rejects.toThrow('ingest-changed'); + await inbound().ingest(group.id, { + kind: 'delete', + ingestId: ingest().id, + expectedRevision: ingest().revision, + }); + await fx.receive(chatter('d')); + expect(placed().map((row) => row.body)).toEqual(['chatter a', 'chatter b']); + expect(inbound().ingests(group.id).ingests).toEqual([]); +}); + +it('refuses unknown conversations and a second connection of the same group', async () => { + const { fx, group } = await connected(); + await expect( + fx.core.externalMessaging.inbound.ingest(group.id, { + kind: 'add', + bindingId: fx.identity.id, + conversation: { kind: 'group', id: 'oc_other' }, + }), + ).rejects.toThrow('conversation-unavailable'); + await expect( + fx.core.externalMessaging.inbound.ingest(group.id, { + kind: 'add', + bindingId: fx.identity.id, + conversation: { kind: 'group', id: 'oc_team' }, + }), + ).rejects.toThrow('ingest-exists'); +}); diff --git a/packages/core/test/plugin.test.ts b/packages/core/test/plugin.test.ts index 8c625b5f1f..3202eb8a5c 100644 --- a/packages/core/test/plugin.test.ts +++ b/packages/core/test/plugin.test.ts @@ -339,6 +339,8 @@ describe('plugin entry', () => { 'messagingDefaultsSet', 'channelBridges', 'channelBridge', + 'channelIngests', + 'channelIngest', 'messagingChannelTarget', 'messagingReceive', 'messagingGroupPolicy',