dsh-superagent/docs/008-手册.md

008 手册

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

添加一个 Source

这份食谱实现一个名为 webhook 的 Source kind。它通过一个虚构的 CLI hookq 读取一个假想的消息队列，并把消息转换为 Triggers。其他所有接口都是 dsh-superagent 已有的部分。

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

示例假定这样的 CLI 契约：

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

Kind 及其插件条目放在 src/sources/webhook.ts，仿照 src/sources/github.ts。若这个 Source 变大，就像 Feishu Source 那样把翻译、Channel 与轮询代码拆成独立模块。

1. 把事件映射到领域

写代码之前先定下稳定的映射。Trigger ids、Thread ids 和 Channel 地址都会被持久化，之后改动它们可能让既有记录失联。

概念	对 webhook 队列而言
Trigger	一条消息
Thread	一个 conversation 里的一条消息及其回复：<conversation>:<id of the first message>
Actor	发送者在队列上的 id，按允许名单写它的方式
Channel 地址	conversation，以及回复要挂在下面的那条消息
接收范围	一个 conversation，在准入表里有自己的一行

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

模块 import 后面各步骤要用到的东西：

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 and environment follow in the next steps.
}

• kind 是稳定的 Source id。它出现在 Trigger 的 source、Thread-key 前缀、source:webhook 设置 block 和 Source Policy 里。
• schema 校验已保存的配置并提供默认值。在操作者保存一个实例之前，registry 会用那些默认值创建一个启用状态的实例。默认值应当什么都不接收；本示例默认为没有 conversation。
• fields 无需自定义面板代码就生成 Source 设置表单。一个 rows 字段编辑以其第一列作键的条目；advanced 隐藏不常用的设置。若 Source 能把 ids 解析成显示名，就实现 refreshLabels。把这些名字存到 KV.sources.labels(kind) 下，带 choose 与 refresh 的列随即可提供它们。
• 可选的 facts(deployment) 展示由 setup 推导的值，例如某个 CLI 背后的账号。目前 setup 只检查 SETUP_FIELDS 里列出的 GitHub 与 Feishu 字段，所以本示例不提供 facts。
• health 声明这个 Source 的健康条目，初始为 starting。注册会认领每个名字，并拒绝与 runtime、reply、worker 或 effect 种类等条目的冲突。用 kind 的名字，并把 sync:<id> 留给 Policy 同步。一个 error 会让状态脚本以 2 退出、让每小时检查通知操作者，所以只在可能需要人来处理时上报它。

3. 把一条消息翻译成 Trigger

转换之前先校验 CLI 的应答：

/** One message as `hookq pull` and `hookq thread` print it. */
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>
/** The answer of `hookq pull`, whose items are read one by one, and of `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() })

/** The Trigger of one queue message; the agent's own messages produce none. */
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 },
  }
}

• 稳定的 ids。 通过 key 从队列 id 派生 Trigger id。Engine.receive 对已存储的 id 返回 duplicate，重复的 accept 也只创建一个 Ticket。因此重读消息是安全的。要把编辑当作新事件，就在 id 里包含编辑版本，如同 GitHub 对评论用 updated_at 那样。
• Threads。 第一个 Thread 是 primary，拥有那个 Eval。额外的 Threads 增加 Correlation 上下文。id 由 Source 选定；它的完整 key 是 webhook:<id>。
• 发送者。 在 actor 里使用 Source 原生的 id。bot 与 CI 设 human: false。名单里的 bot 只能成为 inferred 候选。空的 actor id——如 CI 的情形——使用针对未列出、未被 at 的发送者的准入单元格。只有消息确实 at 了智能体时才设 addressed 为 mention 或 direct；这让名单中者的消息获得 explicit 强度。绝不为智能体自己的消息产出 Trigger。
• 回复去向与元数据。 channel 指明回复去向；url 链接到源消息。channel.thread 为有序投递给消息分组：一个 Ticket 的较晚消息排在较早消息之后，直到某条较早消息失败三次。其他 Tickets 不排队。Source 专属字段留在 meta，比如接收它的 conversation 与 Thread root。meta 与 text 都是不可信数据，不是授权。

Agents 可以通过 orchestrator_read trigger/<id> 与 thread/<key> 读取新 kind。它们暴露消息文本、处置结果与相邻 Triggers。新 kind 不会自动获得内建技能、原始服务记录的资源或 pull-request 绑定。

文本搜索无需额外代码即可工作。当人们使用 Source 专属的指称——如消息 id、URL 或 #12——时才实现 find(query)。返回那些指称所标识的 Thread keys 与搜索词：

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

这里，查询里的队列消息 id（如 m42）会去搜索具有它的 Triggers。

4. Source 接收什么：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 对每个进来的 Trigger 经由 SourceRegistry 调用 scope。范围之外的消息作为 filtered 存储并带上原因，不创建 Thread。把消息交给 receive，而不是在轮询循环里悄悄丢弃，这样面板才能解释过滤。

observed: true 表示 Source 接收该范围内未被 at 的消息，允许对它们做准入表判断。row 标识对应的准入行。

重新校验会用 Mandate 存储的元数据调用 scope。让它只依赖 meta 与配置。恢复要求的是 listened，不是 observed。没有 scope 时，一个 kind 接收它产出的每个 Trigger，但不 observe 任何消息。

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。Source 插件被卸载期间，消息等待。发送可能在远端服务已收到之后才失败，所以重试不得重复发消息。用 message.id 作幂等键：hookq 支持 --key；GitHub 存一个隐藏的评论标记；Feishu 接受一个幂等键。
• 永久失败与可重试失败。 成功返回 sent，只有永久无法投递时才返回 gone，例如示例中的 410。一条 gone 消息作为 dropped 保留，不再排队；它出现在 GET sources 和 Source 页面上。其他失败要抛出，让 Outbox 在 5 秒后重试，倍增到 300 秒。CliError.code 是 CLI 输出报告的代码。
• 允许的目的地。 allowed(address, config) 只检查已保存的地址与当前配置。把每个必需的事实放进地址；例如 Feishu 记录 chat 是否 direct。一个不再被监听的目的地会被丢弃，而被停用的 Source 让消息保持等待。没有 allowed 时，Source 启用期间所有地址都允许。
• 链接。 link(address, links) 返回 SourceRegistry.link 供面板展示的 URL 与标签。

6. 供研判使用的 Thread 上下文

  threadContext: {
    messageId: trigger => String(trigger.meta.messageId ?? ''),
    // A queue thread is flat: its root, and the message a reply answers, are on its first page, as on 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 } }
    },
  },

主动研判需要 Trigger 周围的对话。orchestrator_eval 读一个初始窗口；orchestrator_eval_read 展开分页、root 或父消息。实现可选的 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 可描述特殊结果，如 Feishu 的 chat_neighbours。part 是 thread、root 或 parent。
• messageId(trigger) 在这些消息中找到 Trigger 自己的那条。
• images(text) 与 image(…) 供消息带图片的 kind 使用，如 Feishu；示例略去了它们。

没有 threadContext 的 kind 不给研判提供任何 Thread 上下文；Trigger 自己的文本仍可读。

若消息会回复更早的消息，就加 repliesTo(trigger)。它返回对应的 Trigger id，让跟进通知与 worker Inputs 能在已存储时展示被回复的消息。例如，若 hookq 提供 reply_to 且翻译把它存为 meta.replyTo：

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

7. 谁可以提需求：Source 的 Policy 部分

  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 校验 Source Policy，包括 allowed 发送者 ids。id 比较不区分大小写。准入表单独校验、跨 kinds 共享。这部分 Policy 存于 policy.sources.webhook，并参与 Policy 版本。
• fields 生成 Eval → On each Source 下的表单，从 allowed 开始。explicit 解释如何 at 智能体。可选的 unnamed: {person, bot} 为 Feishu 的 ou_… 这类不透明 ids 提供显示标签；示例省略它，因为队列 id 是可读的名字。
• rows(config) 列出准入范围及其可用单元格。每行的 id 必须匹配 scope 返回的 row。未被 at 的消息的单元格，只在 Source observe 那些消息的地方可用。没有 rows 的 kind 使用默认单元格。
• 可选的 bots 列出有 inferred 候选资格的 bots，使用与 allowed 相同的条目形状。没有它，任何 bot 都不被准入。
• 可选的 sync: {label, ready(part), run(host, part, signal)} 在启动与 Sync now 时刷新成员名单。Feishu 用 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 设置即使在 Source 停用时，也会在 prepare command 旁展示声明的 variables。启用期间，values 用 deployment、配置与发起的 Trigger 提供 Ticket 专属的值。

prepare command 读取环境变量；worker 在其指令中读取这些值。使用 DSH_SUPERAGENT_<KIND>_… 前缀，对来自其他 Source 的 Tickets 返回空值，且永不包含凭据。

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 ends its process on an unhandled rejection, so a failure that escapes the loop becomes the Source's health.
    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 注册 kind，并在销毁时注销。第二个用 SourceContext 运行轮询循环。销毁会 abort 循环并等待其完成，包括应用停止或重启期间。

async function watch(ctx: SourceContext, signal: AbortSignal): Promise<void> {
  await ctx.ready
  // Setup restarts dsh-superagent once it resolves the accounts; until then the Source reads nothing.
  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
        // One network failure is retried on the next round; three in a row need a person.
        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(() => {})
  }
}

/** Read the queue after the cursor, take in each message, then move the cursor past them. */
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)：Source 停用时它是 undefined。

• 游标。 用 KV（src/kv.ts）里有类型的条目把读取位置存进 ctx.state，例如 webhook: {cursor: record<number>('webhook-cursor')}。tests/store.spec.ts 要求 key 字面量只出现在那个 registry 里。外部包可以用导出的 record 与 family 定义 keys。把 record 设为 undefined 会移除它；ctx.atomic 把写入归组进一个事务。

  只有在每个被覆盖的 Trigger 都已存储之后才推进游标。被打断的一轮保持游标不变；稳定的 ids 让重放安全。对时间戳游标，src/sources/poll.ts 的 cursorAt(started) 会减去五秒，以容纳远端的迟到写入。

• Thread 元数据。 收到一个 Trigger 后，调用 ctx.threads.describe(key, {label, url, facts}) 填好它的展示信息。ctx.threads.patch(key, meta) 委托给 threads.patchMeta，设置 Source 拥有的键，或在值为 undefined 时移除它们。两者都会忽略不存在的 Threads，包括 filtered Triggers 的。

• 逐条错误。 计数畸形 item 并继续，而不是让整批失败。src/infra/commands.ts 里的 validItems(item, list, tally) 支持这种模式。对独立轮询的 Threads，用 src/sources/poll.ts 的 pollThreads；它把每次失败记为 pollError。它的 polled 谓词保留 live 订阅，以及较晚的 Trigger 与 Ticket 结束之后两周的窗口。

• 健康。 只通过 ctx.report 写声明过的条目。health(state, code, values) 提供固定消息，如 setupPending、sourceOff 与 pollFailures。上报 error 前先数连续失败次数。

• Abort。 把 signal 传给 ctx.commands.withSignal(signal) 与 delay，并在条目之间检查它，这样停止或重启能立刻结束循环。

本示例不发送 Source 发起的通知。需要它们的 Source 调用 ctx.notify；Outbox 像投递其他回复一样投递它们。

它也不把 Threads 绑定到 Ticket 完成。只在 Source 的 Thread 生命周期决定工作是否完成时才用 ctx.tickets.bindThread 与 threadClosed，如 GitHub pull requests 那样。Ticket 何时结束仍由 Engine 决定。

10. 注册插件

在 tsdown.config.ts 里把这个模块构建为它自己的 entry：

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 列表，处于禁用状态，就像 Feishu：

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

在 profile 的 cordis.patch.yml 里用 id: dsh-superagent-webhook 与 disabled: false 启用该插件。构建并重启。新 Source 会出现在 Sources 设置和 Eval → On each Source 下。

把它的模块与 spec 加进 007 架构 的 module 表和 test 表。tests/docs.spec.ts 检查两个索引是否完整。

11. 测试

仿照 tests/sources.spec.ts 的模式创建 tests/webhook.spec.ts：一个临时目录里的真实 Store、带 HealthBoard 的 SourceRegistry、用替身端口的 Engine，以及来自 sourceContextOf 的 SourceContext。

通过 spy Commands.run 来 mock CLI 调用。也要 spy 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' }] } } },
  }))
  // The registry claims the kind's health entries on the board, where the test reads them.
  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 })
  // The Source's own notices go to the outbox, which keeps their receipts and sends nothing here.
  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) }
}
/** The queue as `hookq` answers it. */
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)
  // An allowed sender who mentions the agent makes an explicit request.
  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) 构造一个队列 item。至少覆盖：

• 翻译：稳定的 id、Thread、actor、human 与 addressed，且智能体自己的消息不产生 Trigger；
• 一轮：Triggers 只存储一次、一个范围之外的被 filtered、游标在其之后移动、名单中者的 mention 被研判为 explicit；
• 被打断的一轮，其游标保持不变；
• send：sent、仅对 410 返回 gone、任何其他失败抛出，CLI 的错误用 commandError 构造；
• Thread 上下文：一页的消息与下一页的游标，如同 tests/registry.spec.ts 通过 SourceRegistry.threadContext 读取一个 kind 的 Threads；
• 插件：apply 注册 kind，并在 setup 等待时上报 setupPending；它的 disposers 注销 kind；逃出循环的失败变成该 Source 的健康状态，而不是 unhandled rejection；
• Source 的 Policy 部分及其准入行，如同 tests/registry.spec.ts 与 tests/policy.spec.ts 检查 Feishu Source 的那样。

检查清单

1. 映射 Triggers、Threads、actors、Channel 地址与接收范围。
2. 写出 kind：schema、fields、health，若 setup 推导了任何值再加 facts。
3. 把事件翻译成带稳定 id 的 Triggers，并丢掉智能体自己的。
4. 在 scope 里决定 Source 接收什么，只依据 meta。
5. 给它一个 Channel，其 send 每条消息只发一次，并且只在 conversation 永远无法接收时才回答 gone。
6. 若能读取 Thread，给研判提供它的 Thread 上下文。
7. 给它一个带 allowed 与准入行的 Policy 部分。
8. 注册它的执行环境变量。
9. 写好插件条目和一个循环：等待 setup、跟随开关、在 receive 之后移动游标、把错误保持在逐条层面、上报健康、abort 时停止；游标名字写进 KV。
10. 添加构建 entry、导出和 cordis.patch.yml insert，并把模块及其 spec 列进 007 的表格。
11. 像 tests/sources.spec.ts 测试其他 Sources 那样测试它。