008 食谱：添加新的 Source

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

添加一个 Source

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

实现有四项职责：转换事件、定义接收与授权规则、提供回复和上下文，以及维护一个轮询循环。带编号的步骤按顺序构建这些部分。

本示例假设如下 CLI 契约：

• hookq pull --after <n> --json 打印位置 n 之后的消息，最旧的在前：{"me": "<智能体在队列上的 id>", "items": […], "next": <position>}。一个条目包含 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 只回复一次。对一个已被删除的对话，它会在标准错误输出上以 {"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 逐渐变大，就像飞书 Source 那样把转换、Channel 和轮询代码拆分到各自的模块中。

1. 把事件映射到领域

在编写代码之前先选定稳定的映射。Trigger id、Thread id 和 Channel 地址都会被持久化，之后再更改它们可能使既有记录失去关联。

2. 配置：schema、fields、facts 与 health

该模块导入下面步骤会用到的内容：

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 设置块以及 Source Policy 中。
• schema 校验存储的配置并提供默认值。在操作者保存一个实例之前，registry 会根据这些默认值创建一个已启用的实例。默认值应当不接收任何内容；本示例默认没有任何对话。
• fields 无需自定义面板代码即可生成 Source 设置表单。rows 字段编辑以第一列为键的条目；advanced 隐藏较少使用的设置。如果这个 Source 能把 id 解析为显示名称，实现 refreshLabels。把这些名称保存在 KV.sources.labels(kind) 之下。带 choose 和 refresh 的列随后即可提供这些名称。
• 可选的 facts(deployment) 展示由 setup 推导出的值，例如某个 CLI 背后的账号。Setup 目前只检查 SETUP_FIELDS 中列出的 GitHub 和飞书字段，因此本示例不提供 facts。
• health 声明该 Source 的健康条目，初始为 starting。注册时会认领每个名称，并拒绝与 runtime、reply、worker 或 effect 种类等条目发生冲突。使用该 kind 自己的名称，并为 Policy 同步保留 sync:<id>。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 },
  }
}

• 稳定的 id。通过 key 由队列 id 推导 Trigger id。Engine.receive 对已存储的 id 返回 duplicate，而重复的 accept 也只会创建一个 Ticket。因此重新读取消息是安全的。若要把编辑当作新事件处理，就把编辑版本纳入 id，就像 GitHub 使用评论的 updated_at 那样。
• Thread。第一个 Thread 是主 Thread，拥有该 Eval。额外的 Thread 增加 Correlation 上下文。Source 自行选择 id；其完整 key 是 webhook:<id>。
• 发送者。在 actor 中使用 Source 原生的 id。对 bot 和 CI 设置 human: false。名单中的 bot 只能成为 inferred（推断）候选。空的 actor id（例如 CI 的情况）使用为未列出且未指名的发送者准备的准入单元格。仅当消息确实在向智能体说话时才把 addressed 设为 mention 或 direct；这会让被允许的人的消息获得 explicit 强度。绝不为智能体自己的消息生成 Trigger。
• 回复目的地与元数据。channel 标识回复发往何处；url 链接到源消息。channel.thread 将消息分组以按序投递：一个 Ticket 的后续消息排在它较早的消息之后等待，直到某条较早的消息已失败三次。其他 Ticket 不必等待。把 Source 特有的字段保存在 meta 中，例如接收对话和 Thread 根。meta 和 text 都属于不可信数据，不构成授权。

agent 可以通过 orchestrator_read trigger/<id> 和 thread/<key> 读取新的 kind。它们暴露消息文本、处置（disposition）和相邻的 Trigger。新的 kind 不会自动获得内置 skill、原始服务记录的 resources 或 PR 绑定。

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

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

在这里，查询中的队列消息 id（例如 m42）会在 Trigger 中搜索它。

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 通过 SourceRegistry 为每个进入的 Trigger 调用 scope。超出范围的消息会连同其原因被存储为 filtered，并且不创建 Thread。把消息交给 receive，而不是在轮询循环中悄悄丢弃，这样面板才能解释过滤情况。

observed: true 表示 Source 在该范围内也接收未指名的消息，从而允许准入表对它们作出决定。row 标识对应的准入行。

重新验证（revalidation）时会使用 Mandate 存储的元数据调用 scope。让它只依赖 meta 和配置。恢复（resuming）要求的是 listened，而非 observed。没有 scope 时，一个 kind 会接收它产生的每个 Trigger，但不观察任何消息。

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 存储一个隐藏的评论标记；飞书接受幂等键。
• 永久失败与可重试失败。成功时返回 sent，仅当投递永久不可能时才返回 gone，例如本示例中的 410。gone 的消息会被保留为 dropped，不再入队；它会出现在 GET sources 和 Source 页面中。其他失败则抛出异常，让 Outbox 在 5 秒后重试，间隔按倍增长直至 300 秒。CliError.code 是 CLI 输出报告的错误码。
• 允许的目的地。allowed(address, config) 只检查已保存的地址和当前配置。把所有必需的事实放进地址；例如，飞书记录该会话是否为单聊。不再被接收的目的地会被丢弃，而被禁用的 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 展开后续页、根消息或父消息。实现可选的 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) 在其中找到 Trigger 自身的那条消息。
• images(text) 和 image(…) 面向消息携带图片的 kind，如飞书那样；本示例省略它们。

没有 threadContext 的 kind 不会为评估提供 Thread 上下文；Trigger 自身的文本仍然可读。

如果消息会回复更早的消息，添加 repliesTo(trigger)。它返回对应的 Trigger id，使得后续通知和 worker 的 Input 在已存储的情况下能够显示被回复的消息。例如，如果 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 发送者 id。id 比较不区分大小写。准入表单独校验，并在各 kind 之间共享。这部分 Policy 保存在 policy.sources.webhook，并计入 Policy 版本。
• fields 生成 Eval → On each Source 之下的表单，以 allowed 开头。explicit 解释如何向智能体发起对话。可选的 unnamed: {person, bot} 为不透明 id（例如飞书的 ou_…）提供显示标签；本示例省略它，因为队列 id 是可读的名称。
• rows(config) 列出准入范围及其可用单元格。每个行 id 必须与 scope 返回的 row 匹配。未指名消息的单元格仅在这些消息被 Source 观察的范围内可用。没有 rows 的 kind 使用默认单元格。
• 可选的 bots 列出有资格成为推断候选的 bot，条目形状与 allowed 相同。没有它，任何 bot 都不会被接纳。
• 可选的 sync: {label, ready(part), run(host, part, signal)} 在启动时以及点击 Sync now 时刷新成员关系。飞书使用会话成员关系。失败会出现在 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 ?? '') : '' }),
  },

执行设置会在 prepare 命令旁边展示声明的 variables，即使该 Source 已被禁用。启用期间，values 使用部署信息、配置和来源 Trigger 提供 Ticket 特定的值。

prepare 命令读取环境变量；worker 在其指示中读取它们的值。使用 DSH_SUPERAGENT_<KIND>_… 前缀，对来自其他 Source 的 Ticket 返回空值，并且绝不包含凭据。

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，并在处置时注销它。第二个 effect 使用一个 SourceContext 运行轮询循环。处置会中止循环并等待其完成，包括在应用停止或重启期间。

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 字面量只出现在该注册表中。外部包可以用导出的 record 和 family 定义 key。设置 undefined 会删除一条记录；ctx.atomic 把多次写入组成一个事务。

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

• Thread 元数据。接收一个 Trigger 之后，调用 ctx.threads.describe(key, {label, url, facts}) 填充其显示信息。ctx.threads.patch(key, meta) 委托给 threads.patchMeta，设置 Source 拥有的 key，或在值为 undefined 时移除它们。两者都会忽略不存在的 Thread，包括被过滤的 Trigger 的 Thread。
• 逐条目的错误。统计畸形条目并继续，而不是让整批失败。src/infra/commands.ts 中的 validItems(item, list, tally) 支持这种模式。对于独立轮询的 Thread，使用 src/sources/poll.ts 的 pollThreads；它把每次失败记录为 pollError。其 polled 谓词保持活跃的订阅，并在最后一条 Trigger 与 Ticket 结束时间中较晚者之后维持两周的窗口。
• 健康。只通过 ctx.report 写入已声明的条目。health(state, code, values) 提供固定消息，例如 setupPending、sourceOff 和 pollFailures。在报告 error 之前先统计连续失败次数。
• 中止。把 signal 传给 ctx.commands.withSignal(signal) 和 delay，并在条目之间检查它，使停止或重启立即结束循环。

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

它也不把 Thread 绑定到 Ticket 完成。仅当 Source 的 Thread 生命周期决定工作是否完成时才使用 ctx.tickets.bindThread 和 threadClosed，就像 GitHub PR 那样。Ticket 何时结束仍由 Engine 决定。

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 启用该插件。构建并重启。新的 Source 会出现在 Sources 设置和 Eval → On each Source 之下。

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

11. 测试

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

通过侦听（spy）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' }] } } },
  }))
  // 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) 构建一个队列条目。至少覆盖以下方面：

• 转换：稳定的 id、Thread、actor、human 和 addressed，以及不为智能体自己的消息生成 Trigger；
• 一轮轮询：Trigger 只存储一次，一个超出范围的消息为 filtered，游标随之推进，被允许发送者的提及被评估为 explicit；
• 被中断的一轮，其游标保持不变；
• send：返回 sent，仅对 410 返回 gone，其他任何失败都抛出异常，CLI 的错误由 commandError 构建；
• Thread 上下文：一页的消息和下一页的游标，如 tests/registry.spec.ts 通过 SourceRegistry.threadContext 读取一个 kind 的 Thread 那样；
• 插件：apply 注册该 kind 并在 setup 等待期间报告 setupPending，其处置器注销该 kind，逃逸出循环的失败成为 Source 的健康状态而不是未处理的 rejection；
• Source 的 Policy 部分及其准入行，如 tests/registry.spec.ts 和 tests/policy.spec.ts 检查飞书 Source 的那样。

清单

1. 映射 Trigger、Thread、actor、Channel 地址和接收范围。
2. 编写 kind：schema、fields、health，以及 setup 有所推导时的 facts。
3. 把事件转换为带稳定 id 的 Trigger，并丢弃智能体自己的消息。
4. 在 scope 中只依据 meta 决定 Source 接收什么。
5. 为它提供一个 Channel，其 send 把每条消息只发送一次，并且仅当对话永远无法接收时才回答 gone。
6. 如果它能读取 Thread，为评估提供其 Thread 上下文。
7. 为它提供 Policy 的一部分，包含 allowed 和准入行。
8. 注册它的执行环境变量。
9. 编写插件入口和一个循环：等待 setup、跟随开关、在 receive 之后推进游标、逐条目地保留错误、报告健康状态并在中止时停止；在 KV 中为游标命名。
10. 添加构建入口、导出和 cordis.patch.yml 的 insert，并在 007 的表中列出该模块及其 spec。
11. 像 tests/sources.spec.ts 测试其他 Source 那样测试它。