dsh-superagent 文档|008 实践手册

本手册演示如何扩展 dsh-superagent。它使用 001 领域模型 与 002 授权 中的术语，以及 007 架构 中的各个组件。

添加一个来源（Source）

这份食谱实现一个名为 webhook 的来源种类（Source kind）。它通过一个虚构的命令行工具 hookq 读取一个假想的消息队列，并把消息转换成触发（Trigger）。其余所有接口都属于 dsh-superagent 本身。

实现有四项职责：翻译事件、定义接收与授权规则、提供回复与上下文、维护一个轮询循环。下面的编号步骤按顺序把这些部分搭起来。

这个示例假定如下 CLI 约定：

• hookq pull --after <n> --json
   打印位置 n 之后的消息，最旧的在前：{"me": "<the agent's id on the queue>", "items": […], "next": <position>}。一个条目有 id、conversation、thread（开启其所属线程的那条消息的 id，在该条消息本身上则没有此字段）、带 id 和 bot 的 sender、mentions_me、text、at 和 url。
• hookq reply --conversation <id> --to <message id> --key <key> --text <text>
   在某条消息下面回复，每个 key 只回复一次。对于已被删除的对话，它会失败，并在标准错误上给出 {"error": {"code": 410, "message": "…"}}。
• hookq thread --conversation <id> --root <message id> --json
  ，用 --after <cursor> 取后一页，打印一个线程的一页，最旧的在前：{"items": […], "next": "<cursor>"}，最后一页没有 next。

把这个种类及其插件入口放在 src/sources/webhook.ts，照 src/sources/github.ts 的样子写。如果这个来源不断膨胀，就照飞书（Feishu）来源的做法，把翻译、Channel 和轮询代码拆成独立的模块。

1. 把事件映射到领域上

在写代码之前先选定稳定的映射。触发 id、线程 id 和渠道（Channel）地址都会被持久化，所以日后更改它们可能会断掉已有记录之间的联系。

2. 配置：schema、字段、事实（facts）与健康

模块导入下面各步骤要用到的东西：

import type { Context } from '@deepseek-ai/cordis' import { setTimeout as delay } from 'node:timers/promises' import { z } from 'zod' import { key, threadKey, type OutboundMessage, type Trigger } from '../core/index.ts' import { CliError } from '../infra/commands.ts' import { KV } from '../kv.ts' import { health, sourceContext, type SourceContext } from './host.ts' import type { SourceKind } from './kind.ts' 

const ConfigSchema = z.object({   conversations: z.array(z.object({     id: z.string().regex(/^[a-z0-9-]+$/),     listen: z.enum(['mentions', 'all']),   }).strict()).default([]),   pollSeconds: z.number().int().min(10).max(3600).default(30), }).strict() export type WebhookConfig = z.infer<typeof ConfigSchema>  export const webhookKind: SourceKind<WebhookConfig> = {   kind: 'webhook',   title: { zh: 'Webhook 队列', en: 'Webhook queue' },   description: { zh: '接收 webhook 队列里下列对话的消息。', en: 'Receives the messages of these conversations from the webhook queue.' },   schema: ConfigSchema,   fields: [     { key: 'conversations', type: 'rows', label: { zh: '接收的对话', en: 'Conversations' }, columns: [       { key: 'id', type: 'text', label: { zh: '对话', en: 'Conversation' }, pattern: '[a-z0-9-]+' },       { key: 'listen', type: 'select', label: { zh: '接收', en: 'Receive' }, options: [         { value: 'mentions', label: { zh: '仅 @', en: '@ only' } }, { value: 'all', label: { zh: '全部消息', en: 'All' } }] },     ] },     { key: 'pollSeconds', type: 'number', advanced: true, min: 10, max: 3600,       label: { zh: '轮询间隔 / 秒', en: 'Polling interval / seconds' } },   ],   health: ['webhook'],   // find、scope、channel、threadContext、policy 和 environment 在后续步骤中给出。 } 

• kind
   是来源的稳定 id。它会出现在触发（Trigger）的 source 字段、线程键前缀、source:webhook 设置块以及来源策略（Source Policy）中。
• schema
   校验被存储的配置并提供默认值。在运维人员保存某个实例之前，注册表会依据这些默认值创建一个已启用的实例。默认值应当做成“什么都不接收”；这个示例默认没有任何对话。
• fields
   生成来源（Sources）的设置表单，不需要自定义面板代码。一个 rows 字段按其第一列作为键来编辑条目；advanced 隐藏不常用的设置。如果该来源能把 id 解析成显示名，就实现 refreshLabels，并把这些名字存到 KV.sources.labels(kind) 之下。带 choose 和 refresh 的列随即可供它们选择。
• 可选的 facts(deployment) 展示由 setup 推导出的值，例如某个 CLI 背后的账号。setup 目前只检查 SETUP_FIELDS 中列出的 GitHub 和飞书字段，所以这个示例不提供任何事实。
• health
   声明该来源的健康条目，初始为 starting。注册时会逐一认领这些名称，并拒绝与别处的条目冲突：例如 runtime、worker 这些健康条目，以及 reply 等各个效果类型（effect kinds）。用该种类自己的名字，并把 sync:<id> 保留给策略同步。error 会让状态脚本以退出码 2 退出，并让每小时检查通知运维人员，因此只有在可能需要人工介入时才报告它。

3. 把一条消息翻译成触发

先校验 CLI 的响应，再对其进行转换：

/** 一条队列消息，即 `hookq pull` 和 `hookq thread` 打印出来的样子。 */ const MessageSchema = z.looseObject({   id: z.string().min(1), conversation: z.string().min(1), thread: z.string().optional(),   sender: z.looseObject({ id: z.string(), bot: z.boolean().default(false) }), mentions_me: z.boolean().default(false),   text: z.string(), at: z.string(), url: z.string().optional(), }) type QueueMessage = z.infer<typeof MessageSchema> /** `hookq pull` 的应答（其条目会被逐条读取），以及 `hookq thread` 的应答。 */ const PageSchema = z.looseObject({ me: z.string(), items: z.array(z.unknown()), next: z.number().int() }) const ThreadPageSchema = z.looseObject({ items: z.array(MessageSchema), next: z.string().optional() }) 

/** 一条队列消息对应的触发；智能体自己的消息不产生触发。 */ export function webhookTrigger(message: QueueMessage, me: string): Trigger | undefined {   if (message.sender.id === me) return undefined   const thread = `${message.conversation}:${message.thread ?? message.id}`   return {     id: key('webhook', message.id),     source: 'webhook',     threads: [{ source: 'webhook', id: thread }],     actor: { source: 'webhook', id: message.sender.id },     human: !message.sender.bot,     ...(message.mentions_me ? { addressed: 'mention' as const } : {}),     at: message.at,     text: message.text,     channel: { source: 'webhook', thread: `webhook:${thread}`,       address: { conversation: message.conversation, message: message.id, ...(message.url ? { url: message.url } : {}) } },     ...(message.url ? { url: message.url } : {}),     meta: { conversation: message.conversation, messageId: message.id, root: message.thread ?? message.id },   } } 

• 稳定的 id。
   通过 key 从队列 id 派生触发 id。Engine.receive 对已经存过的 id 返回 duplicate，而重复的 accept 只会创建一个工单（Ticket）。因此重复读取消息是安全的。若要把编辑当作新事件，就在 id 里带上一个编辑版本，就像 GitHub 对评论使用 updated_at 那样。
• 线程。
   第一个线程是主线程，并拥有那次评估（Eval）。额外的线程添加关联（Correlation）上下文。来源自己挑选 id；它的完整键是 webhook:<id>。
• 发送者。
   在 actor 中使用来源原生的 id。对机器人和 CI 设 human: false。被列在名单上的机器人只能成为推断候选（inferred candidates）。空的行动者 id（例如 CI 那种）会使用“不在名单上、未被指名”的发送者所对应的那个准入格。只有当消息确实指名了智能体时，才把 addressed 设为 mention 或 direct；这会让一个已允许之人的消息获得明确（explicit）强度。绝不要为智能体自己的消息产出触发。
• 回复目的地与元数据。channel
   指明回复去往何处；url 链接到源消息。channel.thread 把消息归组以保证有序投递：一个工单的后续消息要排在它先前的消息之后，直到某条先前的消息失败三次为止。其他工单的消息不必等待。把来源特有的字段留在 meta 里，例如接收它的对话和线程根消息。meta 和 text 都是不可信数据，不是授权依据。

智能体可以通过 orchestrator_read trigger/<id> 和 thread/<key> 读取这个新种类。这些读取会暴露消息文本、去向（disposition）和相邻的触发。一个新种类不会自动获得内建技能、访问原始服务记录所需的资源，或拉取请求绑定。

文本搜索无需额外代码即可工作。当人们使用来源特有的引用（如消息 id、URL 或 #12）时，实现 find(query)。返回这些引用所标识的线程键和搜索词：

  find: query => ({ threads: [], terms: [...new Set(query.match(/\bm[0-9]+\b/g) ?? [])] }), 

这里，查询中出现的队列消息 id（如 m42）会被拿去搜索对应的那条触发。

4. 来源接收什么：接收范围（scope）

  scope({ meta }, config) {     const conversation = config.conversations.find(item => item.id === meta.conversation)     if (!conversation) return { listened: false, reason: 'conversation_not_listed' }     return { listened: true, observed: conversation.listen === 'all', row: conversation.id }   }, 

引擎（Engine）针对每个传入的触发，通过 SourceRegistry 调用 scope。范围之外的消息会被存为 filtered（已过滤）并带上其原因，且不创建线程。要把消息传给 receive，而不是在轮询循环里悄悄丢弃，这样面板才能解释过滤情况。

observed: true 表示该来源在这个范围内接收未被指名（unaddressed）的消息，从而允许对它们做准入表决策。row 用来标识对应的准入行。

重新校验（Revalidation）会带着授权（Mandate）所存储的元数据调用 scope。要让它只依赖 meta 和配置。恢复（resume）需要 listened，并不需要 observed。若没有 scope，一个种类会接收它产生的每一个触发，但什么都不观察。

5. 回复：渠道（Channel）

  channel: {     allowed: (address, config) => config.conversations.some(item => item.id === address.conversation),     send: sendWebhook,     link: address => ({ url: address.url ?? '',       label: { zh: `Webhook 对话 ${address.conversation}`, en: `Webhook conversation ${address.conversation}` } }),   }, 

export async function sendWebhook(host: Pick<SourceContext, 'commands'>, message: OutboundMessage): Promise<'sent' | 'gone'> {   const { conversation, message: to } = message.channel.address   if (!conversation || !to) throw new Error('Missing webhook reply target')   try {     await host.commands.run('hookq', ['reply', '--conversation', conversation, '--to', to, '--key', message.id,       '--text', message.text])     return 'sent'   } catch (error) {     if (error instanceof CliError && error.code === 410) return 'gone'     throw error   } } 

• 幂等发送。
   Outbox（发件箱）调用 SourceRegistry.send，后者调用 channel.send。在来源插件被卸载期间，消息会处于等待状态（waiting）。发送可能在远程服务已经收到之后才失败，因此重试不得重复消息。用 message.id 作为幂等键：hookq 支持 --key；GitHub 存一个隐藏的评论标记；飞书接受一个幂等键。
• 永久失败与可重试失败。
   成功时返回 sent，只有当投递永久不可能时才返回 gone，比如示例中的 410。一条 gone 消息会作为 dropped（无需处理）保留，永不再次入队；它会出现在 GET sources 和来源页面上。其他失败要抛出异常，让 Outbox 在 5 秒后重试，间隔翻倍直到 300 秒。CliError.code 是 CLI 输出所报告的错误码。
• 允许的目的地。allowed(address, config)
   只检查已保存的地址和当前配置。要把每一个必需的事实放进地址里；例如飞书会记录该聊天是否为单聊。一个不再被监听的目的地会被丢弃，而已停用的来源会让消息处于等待状态（waiting）。若没有 allowed，在来源启用期间所有地址都被允许。
• 链接。link(address, links)
   返回面板通过 SourceRegistry.link 显示的 URL 和标签。

6. 供评估使用的线程上下文

  threadContext: {     messageId: trigger => String(trigger.meta.messageId ?? ''),     // 队列线程是平铺的：它的根消息、以及某条回复所回答的那条消息，都在它的第一页上，正如 GitHub 那样。     read: async (commands, trigger, _part, cursor) => {       const page = ThreadPageSchema.parse(JSON.parse(await commands.run('hookq', ['thread', '--conversation',         String(trigger.meta.conversation), '--root', String(trigger.meta.root), '--json', ...(cursor ? ['--after', cursor] : [])])))       const messages = page.items.map(item => ({ id: item.id, from: item.sender.id, at: item.at, text: item.text,         ...(item.sender.bot ? { bot: true as const } : {}), ...(item.url ? { url: item.url } : {}) }))       return { messages, ...(page.next ? { next: page.next } : {}), body: { messages } }     },   }, 

主动评估需要某个触发前后的对话内容。orchestrator_eval 读取一个初始窗口；orchestrator_eval_read 展开分页、根消息或父消息。实现可选的 threadContext 接口来提供这些读取：

• read(commands, trigger, part, cursor, config)
   返回一个 ContextPart。它的 messages 使用 ContextMessage：{id, from, at, text}，外加适用的 bot、deleted、reply_to、url、path 和 line 字段。最后一页要省略 next。body 是展示给正在做评估的会话（Session）的内容，这里就是同样的那些消息。可选的 status 可以描述特殊结果，例如飞书的 chat_neighbours。part 是 thread、root 或 parent。
• messageId(trigger)
   在这些消息之中找到该触发自己的那条消息。
• images(text)
   和 image(…) 留给消息携带图片的种类，飞书的消息就带图片；这个示例不实现它们。

没有 threadContext 的种类不会给评估提供任何线程上下文；触发自己的文本仍然可读。

如果消息会回复更早的消息，就加上 repliesTo(trigger)。它返回相应的触发 id，让跟进通知和工作器（worker）的输入（Input）能在已存储的情况下显示所回复的那条消息。例如，若 hookq 提供 reply_to，并且翻译时把它存为 meta.replyTo：

  repliesTo: trigger => typeof trigger.meta.replyTo === 'string' ? key('webhook', trigger.meta.replyTo) : undefined, 

7. 谁可以提出请求：策略中属于来源的那一部分

  policy: {     schema: z.object({ allowed: z.array(z.object({ id: z.string().min(1), name: z.string().trim().max(100).optional() }).strict())       .default([]) }).strict(),     fields: [{ key: 'allowed', type: 'rows', label: { zh: '能发起工作的发送者', en: 'Senders who may originate work' }, columns: [       { key: 'id', type: 'text', label: { zh: '发送者', en: 'Sender' } },       { key: 'name', type: 'text', label: { zh: '名字', en: 'Name' }, optional: true }] }],     explicit: { zh: '名单里的人在消息里 @ 智能体。', en: 'Someone on the list mentions the agent in a message.' },     rows: stored => ConfigSchema.parse(stored).conversations.map(conversation => {       const all = conversation.listen === 'all'       return { id: conversation.id, label: conversation.id, note: all ? { zh: '全部消息', en: 'All' } : { zh: '仅 @', en: '@ only' },         cells: { listedUnaddressed: all, unlistedAddressed: true, unlistedUnaddressed: all } }     }),   }, 

• schema
   校验来源策略，其中包括 allowed 发送者 id。id 的比较不区分大小写。准入表另行校验，且各来源种类共享同一套。这一部分策略保存在 policy.sources.webhook，并参与策略版本（Policy version）的计算。
• fields
   生成 评估（Eval）→ 各来源设置（On each Source） 之下的表单，以 allowed 开头。explicit 说明如何指名智能体。可选的 unnamed: {person, bot} 为不透明的 id（如飞书的 ou_…）提供显示标签；这个示例省略了它，因为队列 id 本身就是可读的名字。
• rows(config)
   列出准入范围（receiving scope）及其可用的单元格。每一行的 id 必须与 scope 返回的 row 一致。未被指名消息的单元格，只在来源确实观察那些消息的范围里才可用。没有 rows 的种类使用默认单元格。
• 可选的 bots 列出有资格成为推断候选的机器人，使用与 allowed 相同的条目形状。没有它，就不会准入任何机器人。
• 可选的 sync: {label, ready(part), run(host, part, signal)} 在启动时以及点击立即同步（Sync now）时刷新成员名单。飞书使用的是聊天（chat）成员名单。失败会显示在 sync:<id> 之下，并在一分钟后重试，间隔翻倍直到一小时，直至成功。

8. 供执行器（Executor）使用的变量

  environment: {     variables: [{ key: 'DSH_SUPERAGENT_WEBHOOK_CONVERSATION', description: {       zh: '接下工单的 webhook 对话；工单不是从这个来源来的时为空',       en: 'The webhook conversation that started the ticket; empty when the ticket did not start on this Source' } }],     values: ({ trigger }) => ({       DSH_SUPERAGENT_WEBHOOK_CONVERSATION: trigger?.source === 'webhook' ? String(trigger.meta.conversation ?? '') : '' }),   }, 

执行设置（Execution）会在准备命令（prepare command）旁边展示已声明的 variables，即使该来源处于停用状态。在启用期间，values 用部署、配置和发起该工单的那个触发，提供工单特定的取值。

准备命令读取环境变量；工作器在其指令中读取这些取值。使用 DSH_SUPERAGENT_<KIND>_… 前缀，对来自另一个来源的工单返回空值，并且绝不包含任何凭据。

9. 插件入口及其循环

export const name = 'dsh-superagent-webhook' export const inject = ['orchestrator'] export function apply(ctx: Context): void {   ctx.effect(() => ctx.orchestrator.sources.register(webhookKind), 'webhook: source kind')   ctx.effect(() => {     const source = sourceContext(ctx.orchestrator, webhookKind)     const controller = new AbortController()     // DSH 在遇到未处理的 promise 拒绝（unhandled rejection）时会结束它的进程，因此逃逸出循环的失败会变成该来源的健康状态。     const running = watch(source, controller.signal)       .catch((error: unknown) => { source.report('webhook', { state: 'error', detail: String(error) }) })     return async () => { controller.abort(); await running }   }) } 

插件注入 orchestrator。它的第一个 effect 注册这个种类，并在其卸载（dispose）时把该种类注销。第二个 effect 带着一个 SourceContext 跑轮询循环。卸载会中止这个循环并等待它跑完，包括在应用停止或重启期间。

async function watch(ctx: SourceContext, signal: AbortSignal): Promise<void> {   await ctx.ready   // setup 解析好账号之后会重启一次 dsh-superagent；在此之前该来源什么都不读。   if (!ctx.configured()) { ctx.report('webhook', health('paused', 'setupPending')); return }   let failures = 0   while (!signal.aborted) {     const config = ctx.config<WebhookConfig>('webhook')     if (!config) ctx.report('webhook', health('paused', 'sourceOff'))     else {       const at = new Date().toISOString()       try {         const { read, unreadable } = await pollQueue(ctx, signal)         failures = 0         ctx.report('webhook', { state: 'ready', detail: `${read} messages, ${unreadable} unreadable`, at })       } catch (error) {         failures += 1         // 一次网络失败会在下一轮重试；连续三次就需要人来处理。         if (!signal.aborted && failures >= 3) {           ctx.report('webhook', health('error', 'pollFailures', { count: String(failures), error: String(error) }))         }       }     }     await delay((config?.pollSeconds ?? 30) * 1000, undefined, { signal }).catch(() => {})   } }  /** 读取游标之后的队列，逐条收下每条消息，然后把游标推进到它们之后。 */ export async function pollQueue(ctx: SourceContext, signal: AbortSignal): Promise<{ read: number; unreadable: number }> {   const after = ctx.state.get(KV.webhook.cursor) ?? 0   const output = await ctx.commands.withSignal(signal).run('hookq', ['pull', '--after', String(after), '--json'])   const page = PageSchema.parse(JSON.parse(output))   let unreadable = 0   for (const item of page.items) {     if (signal.aborted) throw new Error('The webhook Source stopped during the read')     const message = MessageSchema.safeParse(item)     if (!message.success) { unreadable++; continue }     const trigger = webhookTrigger(message.data, page.me)     if (!trigger) continue     await ctx.receive(trigger)     const conversation = { label: { zh: '对话', en: 'Conversation' }, value: message.data.conversation }     for (const ref of trigger.threads) ctx.threads.describe(threadKey(ref), { facts: { conversation } })   }   ctx.state.set(KV.webhook.cursor, page.next)   return { read: page.items.length, unreadable } } 

• setup 与开关。
   等待 ctx.ready，并且在 ctx.configured() 为 false 期间什么都不读。每一轮都读取 ctx.config(kind)：来源关闭时它是 undefined。
• 游标。
   在 ctx.state 中以 KV（src/kv.ts）里的类型化条目存储已读位置，例如 webhook: {cursor: record<number>('webhook-cursor')}。tests/store.spec.ts 要求键字面量只出现在那个注册表里。外部包可以用导出的 record 和 family 定义键。把一条记录设为 undefined 会删除它；ctx.atomic 把多个写入归组进一个事务。

只能在所有被覆盖的触发都已存储之后才推进游标。被打断的一轮不会改变游标；稳定的 id 使重放是安全的。对于时间戳游标，src/sources/poll.ts 中的 cursorAt(started) 会减去五秒，以容忍远端迟到的写入。 - 线程元数据。 收到一个触发之后，调用 ctx.threads.describe(key, {label, url, facts}) 来填充它的显示信息。ctx.threads.patch(key, meta) 委托给 threads.patchMeta，设置由来源拥有的键，或在取值为 undefined 时移除它们。两者都会忽略不存在的线程，包括那些属于 filtered（已过滤）触发的线程。 - 逐条错误。 统计格式错误的条目并继续，而不是让整个批次失败。src/infra/commands.ts 中的 validItems(item, list, tally) 支持这种写法。对于各自独立轮询的线程，使用 src/sources/poll.ts 中的 pollThreads；它把每一次失败记为 pollError。它的 polled 谓词会保持活动订阅，并在最后一个触发与工单结束两者之中较晚者之后保留两周的窗口。 - 健康。 通过 ctx.report 只写入已声明的健康条目。health(state, code, values) 提供像 setupPending、sourceOff 和 pollFailures 这样的固定消息。在报告 error 之前先累计连续失败的次数。 - 中止。 把信号传给 ctx.commands.withSignal(signal) 和 delay，并在条目之间检查它，这样无论是停止还是重启，循环都会立刻结束。

这个示例不发送由来源发起的通知。需要发送这些通知的来源调用 ctx.notify；Outbox 会像投递其他回复一样投递它们。

它也不把线程绑定到工单完成。只有当某个来源的线程生命周期决定工作是否完成时——就像 GitHub 的拉取请求那样——才使用 ctx.tickets.bindThread 和 threadClosed。工单何时结束，仍然由引擎决定。

10. 注册这个插件

把该模块在 tsdown.config.ts 中构建成一个独立入口：

entry: { index: 'src/index.ts', 'github-watch': 'src/sources/github.ts', 'feishu-watch': 'src/sources/feishu.ts',   'webhook-watch': 'src/sources/webhook.ts' }, 

然后在 package.json 里导出构建好的文件，往 exports 中加入 "./webhook-watch": "./lib/webhook-watch.js"，并把该插件加入 bundle 的 cordis.patch.yml 的 insert 列表、保持禁用，就像飞书那样：

    - id: dsh-superagent-webhook       name: dsh-superagent/webhook-watch       disabled: true 

在某个 profile 的 cordis.patch.yml 中用 id: dsh-superagent-webhook 和 disabled: false 启用该插件。构建并重启。新的来源会出现在设置 → 来源（Sources）这一分组里，以及 评估（Eval）→ 各来源设置（On each Source） 之下。

把它的模块与规格加入 007 架构 中的模块表和测试表。tests/docs.spec.ts 会检查这两份索引是否完整。

11. 测试

创建 tests/webhook.spec.ts，沿用 tests/sources.spec.ts 中的写法：在临时目录里放一个真实的 Store，一个带 HealthBoard 的 SourceRegistry，一个用替身端口的 Engine，以及一个由 sourceContextOf 创建的 SourceContext。

通过对 Commands.run 打桩来模拟 CLI 调用。还要对 withSignal 打桩，因为它通常会返回一个新的 Commands 实例。

function host(overrides: Partial<SourceContextDeps> = {}) {   const store = new Store(join(mkdtempSync(join(tmpdir(), 'webhook-')), 'db.sqlite'))   store.saveSettings(SettingsSchema.parse({     sources: { webhook: { enabled: true, config: { conversations: [{ id: 'ops', listen: 'mentions' }] } } },     policy: { sources: { webhook: { allowed: [{ id: 'alice' }] } } },   }))   // 注册表会在看板上认领该种类声明的健康条目，测试就从这里读取它们。   const board = new HealthBoard()   const sources = new SourceRegistry(() => store.settings(), board)   sources.register(webhookKind)   const engine = new Engine({ repo: store, scope: sources, policy: () => settingsPolicy(store.settings(), sources),     assessor: { request: vi.fn() }, executor: { start: vi.fn(), deliver: vi.fn(), stop: vi.fn() },     outbox: { send: vi.fn(), status: () => undefined, withdraw: vi.fn() } })   const commands = new Commands(() => testDeployment(), {}, new AbortController().signal)   const receive = vi.fn((trigger: Trigger) => { const result = engine.receive(trigger); engine.evaluateDue(); return result })   // 来源自己发出的通知进入 outbox，这里只保留它们的回执，并不发送任何东西。   const deps: SourceContextDeps = { store, engine, outbox: { send: message => { store.receipts.add(message) } }, commands, sources,     deployment: testDeployment(), ready: Promise.resolve(), configured: () => true, receive, ...overrides }   return { ...deps, store, engine, commands, receive, health: board.all, ctx: sourceContextOf(deps, webhookKind) } } /** 队列，即 `hookq` 应答的样子。 */ function queue(h: ReturnType<typeof host>, answer: () => unknown) {   vi.spyOn(h.commands, 'withSignal').mockReturnValue(h.commands)   return vi.spyOn(h.commands, 'run').mockImplementation(async () => JSON.stringify(answer())) }  it('takes in what the queue holds after the cursor, once, and moves the cursor after it', async () => {   const h = host()   queue(h, () => ({ me: 'agent', items: [message('m1'), { id: 'broken' }, message('m2', { conversation: 'elsewhere' })], next: 3 }))   expect(await pollQueue(h.ctx, new AbortController().signal)).toEqual({ read: 3, unreadable: 1 })   expect(await h.receive.mock.results[0]!.value).toBe('stored')   expect(await h.receive.mock.results[1]!.value).toBe('filtered')   expect(h.store.get(KV.webhook.cursor)).toBe(3)   // 一位在允许名单上、且 @ 了智能体的发送者会构成一次明确请求。   expect(h.engine.candidate(key('webhook', 'm1'))?.strength).toBe('explicit')   await pollQueue(h.ctx, new AbortController().signal)   expect(await h.receive.mock.results[2]!.value).toBe('duplicate')   h.store.close() }) 

message(id, fields) 构造一个队列条目。至少要覆盖：

• 翻译：稳定的 id、线程、行动者、human 和 addressed，并且智能体自己的消息不产生触发；
• 一轮：触发只被存储一次，一个落在范围之外的为 filtered，游标被移到它们之后，一个允许名单上的发送者的 @ 被评估为 explicit；
• 被打断的一轮，它会保留自己的游标；
• send
  ：sent，只有 410 才返回 gone，任何其他失败都抛出异常，其中 CLI 的错误由 commandError 构造；
• 线程上下文：一页的消息和下一页的游标，就像 tests/registry.spec.ts 通过 SourceRegistry.threadContext 读取一个种类的线程那样；
• 插件：apply 注册该种类，并在 setup 尚未完成时报告 setupPending，它的卸载（dispose）函数会注销该种类，而一个逃逸出循环的失败会变成该来源的健康状态，而不是一个未处理的 promise 拒绝；
• 策略中属于该来源的那一部分及其准入行，就像 tests/registry.spec.ts 和 tests/policy.spec.ts 检查飞书来源的那一部分那样。

检查清单

2. 映射好触发、线程、行动者、渠道（Channel）地址和接收范围。
4. 写出这个种类：schema、fields、health，以及 setup 若有推导则加上 facts。
6. 把事件翻译成带稳定 id 的触发，并丢掉智能体自己的那些。
8. 在 scope 中决定该来源接收什么，只依据 meta 判断。
10. 给它一个渠道（Channel），其 send 每条消息只投递一次，并且只有当对话永远无法收到时才回答 gone。
12. 如果它能读取某个线程，就给评估提供它的线程上下文。
14. 给它策略中的一部分，含 allowed 和准入行。
16. 注册它的执行环境变量。
18. 写出插件入口和一个循环：等待 setup、跟随开关、在 receive 之后移动游标、把错误控制在单条条目、报告健康、并在中止时停止；在 KV 中为游标命名。
20. 添加构建入口、导出项和 cordis.patch.yml 中的 insert，并在 007 的表格里列出该模块及其规格。
22. 照 tests/sources.spec.ts 测试其他来源的方式测试它。