feat: 生命周期广播事件与链式钩子设施（ADR-006）

- contracts:补齐生命周期广播事件(订阅、下载、整理结束、刮削、字幕、作品关注 / 元数据 / 季集结构、站点可达性) 的类型与 payload schema,LIFECYCLE_EVENTS 清单与 lifecycleKey 幂等键;HOOK_POINTS 链式钩子挂点; FEATURES 功能清单与 SLOT_POSITIONS 前端插槽位置 - outbox 中继支持一条事件路由给多个订阅方(其余订阅方用 <键>#<任务名>);生命周期事件允许多模块订阅, 没有订阅方时也被中继认领并镜像到 SSE - 服务端 HookRegistry:按注册顺序执行、rewrite / intercept 两类、单钩子超时与异常隔离、无注册零开销; ServerModule 增加 manifest 与 registerHooks,bootstrap 装配 ctx.hooks - ADR-006 与 server README「如何写一个功能模块」 Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
2026-10-06 19:27:02 +08:00
1 parent 0f656f0f58
commit 1db689d9a7
@@ -4,8 +4,11 @@ import { Id, NonNegInt, Timestamp } from "./common.ts";
 import { DownloadStatus, PauseReason, TaskKind, TaskStatus } from "./execution.ts";
 import { IdentifyCompletedPayload } from "./files.ts";
 import { EvidenceLevel } from "./identity.ts";
-import { OrganizationStatus } from "./organization.ts";
+import { OrganizationAction, OrganizationStatus, OrganizationTrigger } from "./organization.ts";
+import { SubscriptionMediaKind, SubscriptionOrigin, SubscriptionStatus } from "./policy.ts";
+import { ScrapeStatus, ScrapeTrigger } from "./scrape.ts";
 import { FeedDuePayload, FeedGapSuspectedPayload, FeedHealth, ReleaseObservedPayload } from "./sources.ts";
+import { SubtitleLanguage, SubtitleSource } from "./subtitles.ts";

 /** SSE 事件信封:seq 在进程内单调递增,客户端断线后用 ?after=<seq> 补读 */
 export const EventEnvelope = z.object({
@@ -24,10 +27,11 @@ export type EventEnvelope<P = unknown> = {
 /**
 * 已知事件类型:`域.动作`。新增事件先在这里加类型与 payload schema。
 *
- * 没有单独的 download.completed / organization.completed 事件,约定用现有事件表达(W2 收口):
- * - "下载完成" = `download.status_changed` 且 `to === "completed"`(outbox 幂等键 `download.completed:<id>`,只发一次;
- *   acquisition-organize 据此入队 verifyAndImport);
- * - "入库完成" = `organization.changed` 且 `status === "succeeded"`(或 partial);`changeSeq` 与公开增量 API 游标一致。
+ * 状态流事件(`download.status_changed`、`organization.changed`)每次状态变化都发,给 SSE 与增量游标用;
+ * 生命周期广播事件(LIFECYCLE_EVENTS,ADR-006)只在业务节点发一次,给功能模块订阅:
+ * - 下载完成 = `download.completed`(与 `download.status_changed → completed` 同事务发出;acquisition-organize
+ *   仍按后者入队 verifyAndImport,幂等键 `download.completed:<id>` 属于后者);
+ * - 整理结束 = `organization.finished`,`outcome` 区分成功 / 部分成功 / 失败。
 */
 export const EventType = {
  heartbeat: "system.heartbeat",
@@ -59,6 +63,24 @@ export const EventType = {
  decisionUpdated: "decision.updated",

  identifyCompleted: "identify.completed",
+
+  // ── 生命周期广播事件(ADR-006):业务节点事后通知,经 outbox,至少一次 ──
+  subscriptionCreated: "subscription.created",
+  subscriptionUpdated: "subscription.updated",
+  subscriptionCompleted: "subscription.completed",
+  subscriptionCancelled: "subscription.cancelled",
+  downloadCompleted: "download.completed",
+  downloadFailed: "download.failed",
+  downloadRemoved: "download.removed",
+  organizationFinished: "organization.finished",
+  scrapeCompleted: "scrape.completed",
+  scrapeFailed: "scrape.failed",
+  subtitleWritten: "subtitle.written",
+  mediaWatched: "media.watched",
+  mediaMetadataRefreshed: "media.metadata_refreshed",
+  mediaEpisodeStructureChanged: "media.episode_structure_changed",
+  siteUnreachable: "site.unreachable",
+  siteRecovered: "site.recovered",
 } as const;
 export type EventType = (typeof EventType)[keyof typeof EventType];

@@ -172,6 +194,99 @@ export const DecisionUpdatedPayload = z.object({
  status: DecisionStatus,
 });

+// ── 生命周期广播事件 payload(ADR-006)──
+// payload 只放业务 ID 与订阅方做判断要用的少量事实;详情由订阅方按 ID 查业务表(事件可能晚到、重复到)。
+
+const SubscriptionRef = z.object({
+  subscriptionId: Id,
+  mediaEntityId: Id.nullable(),
+  mediaKind: SubscriptionMediaKind,
+});
+/** 订阅已建(用户 / 智能体 / 策略) */
+export const SubscriptionCreatedPayload = SubscriptionRef.extend({
+  origin: SubscriptionOrigin,
+  status: SubscriptionStatus,
+});
+/** 订阅已改:参数或状态(暂停 / 恢复 / 重新订阅);changes 是改动的字段名 */
+export const SubscriptionUpdatedPayload = SubscriptionRef.extend({
+  status: SubscriptionStatus,
+  changes: z.array(z.string()),
+});
+/** 订阅已完成:reason 如 slot_satisfied / download_accepted / stop_after_reached */
+export const SubscriptionCompletedPayload = SubscriptionRef.extend({ reason: z.string() });
+/** 订阅已取消(删除订阅即取消,行保留) */
+export const SubscriptionCancelledPayload = SubscriptionRef.extend({ reason: z.string() });
+
+const DownloadRef = z.object({
+  downloadTaskId: Id,
+  planId: Id.nullable(),
+  mediaEntityId: Id.nullable(),
+  name: z.string(),
+  /** 任务行代数:removed 行复活后 +1 */
+  generation: NonNegInt,
+});
+/** 下载已完成(种子下完,尚未整理) */
+export const DownloadCompletedPayload = DownloadRef;
+/** 下载失败 */
+export const DownloadFailedPayload = DownloadRef.extend({ from: DownloadStatus.nullable() });
+/** 下载已移除(用户删除 / 清理 / 下载器里找不到) */
+export const DownloadRemovedPayload = DownloadRef.extend({ from: DownloadStatus.nullable() });
+
+/** 整理结束:一条整理记录进入 succeeded / partial / failed(统一 organization.changed 的终态) */
+export const OrganizationFinishedPayload = z.object({
+  recordId: Id,
+  outcome: z.enum(["succeeded", "partial", "failed"]),
+  action: OrganizationAction,
+  trigger: OrganizationTrigger,
+  mediaEntityId: Id.nullable(),
+  downloadTaskId: Id.nullable(),
+  reason: z.string().nullable(),
+  changeSeq: z.string().regex(/^\d+$/),
+});
+
+const ScrapeRef = z.object({
+  scrapeRecordId: Id,
+  mediaEntityId: Id.nullable(),
+  organizationRecordId: Id.nullable(),
+  trigger: ScrapeTrigger,
+  status: ScrapeStatus,
+  written: NonNegInt,
+  failed: NonNegInt,
+});
+/** 刮削完成(succeeded / partial / skipped) */
+export const ScrapeCompletedPayload = ScrapeRef;
+/** 刮削失败 */
+export const ScrapeFailedPayload = ScrapeRef.extend({ reason: z.string().nullable() });
+
+/** 字幕已写入媒体库(一条写入历史一条事件) */
+export const SubtitleWrittenPayload = z.object({
+  writeId: Id,
+  mediaEntityId: Id,
+  videoLocationId: Id,
+  language: SubtitleLanguage,
+  source: SubtitleSource,
+});
+
+/** 作品进入关注(建订阅、开始下载、用户搜索 / 手动认定);reason 为触发来源 */
+export const MediaWatchedPayload = z.object({ mediaEntityId: Id, reason: z.string() });
+/** 作品元数据已刷新;changedFields 为本次变化的字段 */
+export const MediaMetadataRefreshedPayload = z.object({
+  mediaEntityId: Id,
+  full: z.boolean(),
+  changedFields: z.array(z.string()),
+});
+/** 季集结构已切换(TMDB 默认季 ⇄ 剧集组);null 表示默认季 */
+export const MediaEpisodeStructureChangedPayload = z.object({
+  mediaEntityId: Id,
+  fromGroupId: z.string().nullable(),
+  toGroupId: z.string().nullable(),
+});
+
+/** 站点不可达:登录失效、列表抓取连续失败或检测失败;reason 为原因码 */
+export const SiteUnreachablePayload = z.object({ siteId: Id, reason: z.string(), message: z.string().nullable() });
+/** 站点恢复 */
+export const SiteRecoveredPayload = z.object({ siteId: Id, reason: z.string() });
+
 /** 事件类型 → payload schema */
 export const EVENT_PAYLOADS = {
  "system.heartbeat": HeartbeatPayload,
@@ -195,6 +310,22 @@ export const EVENT_PAYLOADS = {
  "decision.created": DecisionCreatedPayload,
  "decision.updated": DecisionUpdatedPayload,
  "identify.completed": IdentifyCompletedPayload,
+  "subscription.created": SubscriptionCreatedPayload,
+  "subscription.updated": SubscriptionUpdatedPayload,
+  "subscription.completed": SubscriptionCompletedPayload,
+  "subscription.cancelled": SubscriptionCancelledPayload,
+  "download.completed": DownloadCompletedPayload,
+  "download.failed": DownloadFailedPayload,
+  "download.removed": DownloadRemovedPayload,
+  "organization.finished": OrganizationFinishedPayload,
+  "scrape.completed": ScrapeCompletedPayload,
+  "scrape.failed": ScrapeFailedPayload,
+  "subtitle.written": SubtitleWrittenPayload,
+  "media.watched": MediaWatchedPayload,
+  "media.metadata_refreshed": MediaMetadataRefreshedPayload,
+  "media.episode_structure_changed": MediaEpisodeStructureChangedPayload,
+  "site.unreachable": SiteUnreachablePayload,
+  "site.recovered": SiteRecoveredPayload,
 } as const satisfies Record<EventType, z.ZodType>;

 export type EventPayload<T extends EventType> = z.infer<(typeof EVENT_PAYLOADS)[T]>;
@@ -210,6 +341,89 @@ export const TypedEventEnvelope = z.discriminatedUnion(
  ) as unknown as [z.ZodObject, ...z.ZodObject[]],
 );

+/**
+ * 生命周期广播事件清单(ADR-006):给功能清单、智能体提示与文档用。
+ * 已有的 release.resolved 也算一条(资源识别完成)。emittedBy 是发出的模块,说明写在什么时候发。
+ */
+export const LIFECYCLE_EVENTS: readonly { type: EventType; label: string; emittedBy: string; when: string }[] = [
+  { type: "release.resolved", label: "资源识别完成", emittedBy: "sources-identity", when: "来源资源认到本地作品" },
+  {
+    type: "subscription.created",
+    label: "订阅已建",
+    emittedBy: "acquisition-organize",
+    when: "新建订阅（含策略自动建）",
+  },
+  {
+    type: "subscription.updated",
+    label: "订阅已改",
+    emittedBy: "acquisition-organize",
+    when: "改参数、暂停、恢复、重新订阅",
+  },
+  {
+    type: "subscription.completed",
+    label: "订阅已完成",
+    emittedBy: "acquisition-organize",
+    when: "收录版本全部满足或到达停止日期",
+  },
+  { type: "subscription.cancelled", label: "订阅已取消", emittedBy: "acquisition-organize", when: "删除订阅" },
+  { type: "download.submitted", label: "下载已提交", emittedBy: "acquisition-organize", when: "种子交给下载器" },
+  { type: "download.completed", label: "下载已完成", emittedBy: "acquisition-organize", when: "下载器报告下完" },
+  { type: "download.failed", label: "下载失败", emittedBy: "acquisition-organize", when: "下载任务进入失败" },
+  {
+    type: "download.removed",
+    label: "下载已移除",
+    emittedBy: "acquisition-organize",
+    when: "任务被删除或在下载器里消失",
+  },
+  {
+    type: "organization.finished",
+    label: "整理结束",
+    emittedBy: "acquisition-organize",
+    when: "整理记录成功、部分成功或失败",
+  },
+  {
+    type: "scrape.completed",
+    label: "刮削完成",
+    emittedBy: "acquisition-organize",
+    when: "刮削记录成功 / 部分成功 / 跳过",
+  },
+  { type: "scrape.failed", label: "刮削失败", emittedBy: "acquisition-organize", when: "刮削记录失败" },
+  { type: "subtitle.written", label: "字幕已写入", emittedBy: "subtitles", when: "补字幕写入一个字幕文件" },
+  {
+    type: "media.watched",
+    label: "作品进入关注",
+    emittedBy: "sources-identity / acquisition-organize",
+    when: "建订阅、开始下载、搜索认定作品",
+  },
+  {
+    type: "media.metadata_refreshed",
+    label: "作品元数据已刷新",
+    emittedBy: "sources-identity",
+    when: "刷新元数据任务完成",
+  },
+  {
+    type: "media.episode_structure_changed",
+    label: "季集结构已切换",
+    emittedBy: "sources-identity",
+    when: "作品切换剧集组",
+  },
+  {
+    type: "site.unreachable",
+    label: "站点不可达",
+    emittedBy: "sources-identity",
+    when: "登录失效、列表连续抓取失败或检测失败",
+  },
+  { type: "site.recovered", label: "站点恢复", emittedBy: "sources-identity", when: "不可达的站点再次正常" },
+];
+
+/**
+ * 生命周期事件的 outbox 幂等键:`<type>/<部分…>`。用斜杠与已有的 `<type>:<id>` 键分开
+ * (`download.completed:<id>` 已被 download.status_changed 占用,outbox 幂等键全表唯一)。
+ */
+export function lifecycleKey(type: EventType, ...parts: (string | number)[]): string {
+  return [type, ...parts].join("/");
+}
+
 export function isEventType(type: string): type is EventType {
  return Object.hasOwn(EVENT_PAYLOADS, type);
 }
@@ -0,0 +1,119 @@
+import type { KnownCapabilityId } from "./capability.ts";
+import type { EventType } from "./events.ts";
+import type { HookPoint } from "./hooks.ts";
+
+// 功能模块约定(ADR-006):外围功能各自声明一份 manifest,汇总成功能清单。
+// 功能清单只给智能体看(系统提示的「功能清单」一节),不做开关;前端插槽默认直接显示。
+
+/** 前端插槽位置(apps/web/app/lib/slots.ts 按这些位置注册组件) */
+export const SLOT_POSITIONS = {
+  "sidebar.entry": "侧栏入口",
+  "entity.actions": "作品页操作区",
+  "entity.panels": "作品页面板区",
+  "organize.record.actions": "整理记录行操作",
+  "settings.sections": "设置分区",
+} as const;
+export type SlotPosition = keyof typeof SLOT_POSITIONS;
+
+export interface FeatureManifest {
+  /** 与服务端模块名一致(modules/<id>/) */
+  id: string;
+  name: string;
+  /** 一句话说明这个功能做什么 */
+  summary: string;
+  /** 怎么用:入口在哪、智能体能做什么、不能做什么 */
+  usage: string;
+  /** 订阅的生命周期事件 */
+  subscribes: readonly EventType[];
+  /** 提供的能力(智能体可调用) */
+  capabilities: readonly KnownCapabilityId[];
+  /** 页面入口(界面上的位置) */
+  pages: readonly string[];
+  /** 挂在哪些前端插槽 */
+  slots: readonly SlotPosition[];
+  /** 注册的链式钩子 */
+  hooks: readonly HookPoint[];
+}
+
+/** 功能清单。新增外围功能:在这里加一份,服务端模块 manifest 指向它(featureManifest(id)) */
+export const FEATURES = [
+  {
+    id: "subtitles",
+    name: "补字幕",
+    summary: "给媒体库里的视频补字幕：上传或从 SubHD / 字幕库在线搜索下载，预览配对后写到视频同目录，可看历史、撤销。",
+    usage:
+      "入口在作品页「补字幕」和整理记录行菜单「补字幕」；字幕站在站点页单列一组。智能体没有补字幕能力，用户问到时指引去作品页操作；整理结束事件已订阅，整理后自动补字幕还没做。",
+    subscribes: ["organization.finished"],
+    capabilities: [],
+    pages: ["作品页「补字幕」", "整理记录行菜单「补字幕」", "站点页「字幕站」"],
+    slots: ["entity.actions", "organize.record.actions"],
+    hooks: [],
+  },
+  {
+    id: "browser",
+    name: "原站视图",
+    summary: "在服务端浏览器里打开站点原页面：人工接管、过验证、从页面下载种子或字幕。",
+    usage: "入口在站点页「原站视图」；智能体可以读原站页面、点击、下载种子，挑战页走视觉子流程。",
+    subscribes: [],
+    capabilities: [
+      "browser.snapshot",
+      "browser.navigate",
+      "browser.interact",
+      "browser.download",
+      "browser.solve_challenge",
+    ],
+    pages: ["站点页「原站视图」"],
+    slots: [],
+    hooks: [],
+  },
+  {
+    id: "desktop",
+    name: "桌面模式",
+    summary: "原站视图的远程桌面：直接操作浏览器容器的桌面，处理复杂验证。",
+    usage: "入口在原站视图的桌面模式；只给人用，智能体不操作。",
+    subscribes: [],
+    capabilities: [],
+    pages: ["原站视图「桌面模式」"],
+    slots: [],
+    hooks: [],
+  },
+  {
+    id: "websearch",
+    name: "网页搜索",
+    summary: "用 Brave 搜索站外公开信息，并读取搜索结果的正文（结果是不可信数据）。",
+    usage: "在设置 → 智能体 → 网页搜索里填 Key 后可用；智能体先搜索，再读其中一条结果的正文。",
+    subscribes: [],
+    capabilities: ["web.search", "web.fetch"],
+    pages: ["设置 → 智能体 → 网页搜索"],
+    slots: [],
+    hooks: [],
+  },
+  {
+    id: "mcp",
+    name: "MCP 服务",
+    summary: "把能力以 MCP 工具对外提供，外部客户端用 API Key 接入，授权与智能体相同。",
+    usage: "在设置 → 智能体 → 工具与 MCP 启用并选允许的 API Key；智能体自己不需要经 MCP 调用。",
+    subscribes: [],
+    capabilities: [],
+    pages: ["设置 → 智能体 → 工具与 MCP"],
+    slots: [],
+    hooks: [],
+  },
+  {
+    id: "schedules",
+    name: "定时任务与巡检",
+    summary: "各模块的周期任务统一调度；内置智能体巡检模板（默认关闭）。",
+    usage: "在日志 → 定时任务里查看、启停；智能体只能起草未启用的巡检，由用户检查后启用。",
+    subscribes: [],
+    capabilities: ["schedule.draft_check"],
+    pages: ["日志 → 定时任务"],
+    slots: [],
+    hooks: [],
+  },
+] as const satisfies readonly FeatureManifest[];
+
+export type FeatureId = (typeof FEATURES)[number]["id"];
+
+export function featureManifest(id: FeatureId): FeatureManifest {
+  return FEATURES.find((f) => f.id === id)!;
+}
@@ -0,0 +1,44 @@
+// 链式钩子挂点(ADR-006,参考 MoviePilot ChainEventType):同步、按注册顺序执行、可改写或拦截。
+// 这里只登记挂点名称与语义(给功能清单、智能体提示与文档用);入参 / 返回值类型在服务端
+// apps/server/src/hooks/points.ts,因为它们用到 domain 的解析结果等类型。
+
+/**
+ * - rewrite:每个钩子拿到上一个钩子的结果,返回新值或 undefined(不改);
+ * - intercept:任一钩子返回拒绝原因即停止,后面的钩子不再执行。
+ */
+export type HookKind = "rewrite" | "intercept";
+
+export const HOOK_POINTS = {
+  "name.recognize": {
+    kind: "rewrite",
+    label: "名称识别",
+    when: "资源标题解析之后、按名称召回作品之前",
+    effect: "改写解析结果（标题、年份、季集、外部 ID）",
+  },
+  "organize.rename": {
+    kind: "rewrite",
+    label: "整理改名",
+    when: "按命名模板生成目标路径之后",
+    effect: "改写主视频的目标文件名（同主干的字幕等附属文件跟着改）",
+  },
+  "organize.intercept": {
+    kind: "intercept",
+    label: "整理拦截",
+    when: "整理计划生成后、写整理记录之前",
+    effect: "返回原因即拒绝本次整理（记录为失败，原因 intercepted:<原因>）",
+  },
+  "resource.selection": {
+    kind: "rewrite",
+    label: "资源选择",
+    when: "订阅 / 收录版本重评时候选排好序之后、逐条评估之前",
+    effect: "调整候选顺序或过滤掉一部分（不能加入新候选）",
+  },
+  "subscription.completion.check": {
+    kind: "intercept",
+    label: "订阅完成判定",
+    when: "订阅满足完成条件、标记完成之前",
+    effect: "追加完成条件：返回原因即本次不完成，订阅继续追踪",
+  },
+} as const satisfies Record<string, { kind: HookKind; label: string; when: string; effect: string }>;
+
+export type HookPoint = keyof typeof HOOK_POINTS;
@@ -14,6 +14,8 @@ export * from "./download-board.ts";
 export * from "./episode-mapping.ts";
 export * from "./events.ts";
 export * from "./execution.ts";
+export * from "./features.ts";
+export * from "./hooks.ts";
 export * from "./identity.ts";
 export * from "./instances.ts";
 export * from "./library-versions.ts";
@@ -21,8 +21,11 @@ export interface OutboxRoute {
  priority?: number;
 }

-/** 返回 null 表示该事件无需建任务,直接标记已投递 */
-export type OutboxRouteFn = (event: OutboxEvent) => OutboxRoute | null;
+/**
+ * 返回 null 表示该事件无需建任务,直接标记已投递。
+ * 返回数组 = 广播给多个订阅方(ADR-006):第一个任务用事件的幂等键,其余用 `<键>#<任务名>`,重投时由引擎幂等去重。
+ */
+export type OutboxRouteFn = (event: OutboxEvent) => OutboxRoute | readonly OutboxRoute[] | null;

 export interface OutboxRelayOptions {
  db: AnyDb;
@@ -82,11 +85,13 @@ export class OutboxRelay {
      for (const ev of rows.rows) {
        try {
          const routeFn = this.opts.routes[ev.type];
-          const route = routeFn ? routeFn(ev) : null;
-          if (route) {
-            const payloadTrace = (ev.payload as { traceId?: unknown } | null)?.traceId;
+          const routed = routeFn ? routeFn(ev) : null;
+          const routes = (routed === null ? [] : Array.isArray(routed) ? routed : [routed]) as readonly OutboxRoute[];
+          const baseKey = ev.idempotency_key ?? OutboxRelay.idempotencyKey(ev.id);
+          const payloadTrace = (ev.payload as { traceId?: unknown } | null)?.traceId;
+          for (const [i, route] of routes.entries()) {
            await this.opts.engine.enqueue(route.task, route.input, {
-              idempotencyKey: ev.idempotency_key ?? OutboxRelay.idempotencyKey(ev.id),
+              idempotencyKey: i === 0 ? baseKey : `${baseKey}#${route.task}`,
              traceId: route.traceId ?? (typeof payloadTrace === "string" ? payloadTrace : randomUUID()),
              ...(route.priority === undefined ? {} : { priority: route.priority }),
            });
@@ -0,0 +1,88 @@
+import { sql } from "kysely";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { type EnqueueOptions, OutboxRelay, type TaskEngine } from "../src/index.ts";
+import {
+  connectDb,
+  dropSchemas,
+  migrateBizSchema,
+  pgAvailable,
+  resetSchemas,
+  type TestDb,
+  testSchemas,
+} from "./helpers.ts";
+
+// 生命周期广播(ADR-006):一条 outbox 事件路由给多个订阅方任务。第一个任务用事件的幂等键,
+// 其余用 <键>#<任务名>;失败重投时已入队的由引擎幂等键去重,没入队的补上。
+
+const schemas = testSchemas("outbox_bc");
+const hasPg = await pgAvailable();
+
+describe.skipIf(!hasPg)("OutboxRelay 广播给多个订阅方", () => {
+  let db: TestDb;
+
+  beforeAll(async () => {
+    await resetSchemas(schemas);
+    db = connectDb(schemas.biz);
+    await migrateBizSchema(db, schemas.biz);
+  }, 30_000);
+  afterAll(async () => {
+    await db?.destroy();
+    await dropSchemas(schemas);
+  }, 30_000);
+
+  it("每个订阅方各入队一次,幂等键按订阅方区分;中途失败重投时同键去重", async () => {
+    await sql`INSERT INTO outbox_event (type, payload, idempotency_key)
+      VALUES ('thing.done', '{"id":"t1"}'::jsonb, 'thing.done/t1')`.execute(db);
+    const keys = new Map<string, number>();
+    let failB = true;
+    const engine: Pick<TaskEngine, "enqueue"> = {
+      enqueue: async (task: string, _input: unknown, opts: EnqueueOptions) => {
+        if (task === "b" && failB) {
+          failB = false;
+          throw new Error("引擎暂时不可用");
+        }
+        keys.set(opts.idempotencyKey, (keys.get(opts.idempotencyKey) ?? 0) + 1);
+        return { runId: opts.idempotencyKey, deduplicated: false };
+      },
+    };
+    const relay = new OutboxRelay({
+      db,
+      engine,
+      routes: {
+        "thing.done": (ev) => [
+          { task: "a", input: ev.payload },
+          { task: "b", input: ev.payload },
+        ],
+      },
+    });
+    const first = await relay.pollOnce();
+    expect(first).toEqual({ published: 0, failed: 1 });
+    // 退避期过了再投:a 用同一把键再入一次(真实引擎按幂等键返回已有 run),b 补上
+    await sql`UPDATE outbox_event SET next_attempt_at = now()`.execute(db);
+    expect(await relay.pollOnce()).toEqual({ published: 1, failed: 0 });
+    expect([...keys.keys()].sort()).toEqual(["thing.done/t1", "thing.done/t1#b"]);
+    expect(keys.get("thing.done/t1")).toBe(2);
+    expect(keys.get("thing.done/t1#b")).toBe(1);
+    const row = await sql<{ status: string }>`SELECT status FROM outbox_event`.execute(db);
+    expect(row.rows).toEqual([{ status: "dispatched" }]);
+  });
+
+  it("单个订阅方保持原来的键;返回空数组 / null 直接标记已投递", async () => {
+    await sql`DELETE FROM outbox_event`.execute(db);
+    await sql`INSERT INTO outbox_event (type, payload, idempotency_key) VALUES
+      ('one.sub', '{}'::jsonb, 'k-one'), ('no.sub', '{}'::jsonb, 'k-none')`.execute(db);
+    const keys: string[] = [];
+    const relay = new OutboxRelay({
+      db,
+      engine: {
+        enqueue: async (_t: string, _i: unknown, o: EnqueueOptions) => {
+          keys.push(o.idempotencyKey);
+          return { runId: "x", deduplicated: false };
+        },
+      },
+      routes: { "one.sub": () => [{ task: "a", input: {} }], "no.sub": () => [] },
+    });
+    expect(await relay.pollOnce()).toEqual({ published: 2, failed: 0 });
+    expect(keys).toEqual(["k-one"]);
+  });
+});