feat: 业务节点发生命周期事件并调用链式钩子

- 订阅已建 / 已改 / 已完成 / 已取消,下载完成 / 失败 / 移除,整理结束(统一 organization.changed 终态),
  刮削完成 / 失败,作品进入关注,元数据已刷新(任务调用方),季集结构切换,站点不可达 / 恢复:
  都在调用方事务内写 outbox,带幂等键
- 钩子挂点:name.recognize(resolveRelease / identifyParsed)、organize.rename 与 organize.intercept(planImport)、
  resource.selection(订阅重评)、subscription.completion.check(completeSubscription)
- 测试:各节点事件 payload 与去重、整理改名 / 拦截 / 出错不影响整理、名称识别改写、资源选择、订阅完成拦截

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei

apps/server/src/modules/acquisition-organize/routes/subscriptions.ts

@@ -14,6 +14,7 @@ import type { SubscriptionRow } from "@umt/db";
 import type { Context } from "hono";
 import type { AuthEnv } from "../../../http/auth.ts";
 import { ApiException } from "../../../http/errors.ts";
+import { emitMediaWatched } from "../../sources-identity/service/lifecycle-events.ts";
 import { enqueueWatchedRefresh } from "../../sources-identity/service/watched.ts";
 import type { ModuleApp } from "../../types.ts";
 import { ensureSubscriptionSlot } from "../services/acquisition.ts";
@@ -29,6 +30,7 @@ import {
   slotsOfSub,
   subscriptionCoverage,
 } from "../services/coverage.ts";
+import { emitSubscriptionEvent } from "../services/lifecycle-events.ts";
 import { loadRuleCatalog } from "../services/rules.ts";
 import { CALENDAR_MAX_DAYS, subscriptionCalendar } from "../services/subscription-calendar.ts";
 import { resetSubscription } from "../services/subscription-reset.ts";
@@ -376,6 +378,9 @@ export function registerSubscriptionRoutes(app: ModuleApp, d: AoDeps) {
           // 订阅总有自己的收录版本:策略模板,没有就建一个默认版本(质量预设随后由 applyDerived 写入)
           await ensureSubscriptionSlot(trx, r);
           await applyDerived(trx, r, norm, now);
+          // 生命周期广播(ADR-006):订阅已建;关联作品 = 作品进入关注
+          await emitSubscriptionEvent(d, trx, "subscription.created", r);
+          if (r.media_entity_id) await emitMediaWatched(d.ctx, trx, r.media_entity_id, "subscription", r.id);
           return r;
         });
       if (row.media_entity_id) {
@@ -436,6 +441,7 @@ export function registerSubscriptionRoutes(app: ModuleApp, d: AoDeps) {
           if (r.status !== "cancelled") await ensureSubscriptionSlot(trx, r);
           if (norm) await applyDerived(trx, r, norm, d.ports.now(), subParams(cur));
           else if (body.status !== undefined) await syncPrivateRuleGroup(trx, r, d.ports.now());
+          await emitSubscriptionChange(trx, cur, r, Object.keys(body));
           return r;
         });
       invalidateCoverage(r.id);
@@ -446,6 +452,21 @@ export function registerSubscriptionRoutes(app: ModuleApp, d: AoDeps) {
     { tag: TAG },
   );
 
+  /** 生命周期广播(ADR-006):改成 cancelled 发「已取消」,其余改动发「已改」 */
+  const emitSubscriptionChange = async (
+    trx: Exec,
+    before: SubscriptionRow,
+    after: SubscriptionRow,
+    changes: string[],
+    reason?: string,
+  ) => {
+    if (after.status === "cancelled" && before.status !== "cancelled")
+      await emitSubscriptionEvent(d, trx, "subscription.cancelled", after, {
+        reason: reason ?? after.status_reason ?? "",
+      });
+    else if (changes.length) await emitSubscriptionEvent(d, trx, "subscription.updated", after, { changes });
+  };
+
   // ── 状态动作 ──
   const setStatus = async (
     sub: SubscriptionRow,
@@ -456,16 +477,22 @@ export function registerSubscriptionRoutes(app: ModuleApp, d: AoDeps) {
     const now = d.ports.now();
     const searchOn = (subParams(sub).searchIntervalHours ?? DEFAULT_SEARCH_INTERVAL_HOURS) > 0;
     const r = await db()
-      .updateTable("subscription")
-      .set({
-        status: to,
-        status_reason: reason,
-        ...(to === "paused" || to === "cancelled" ? { next_search_at: null } : {}),
-        ...(extra.nextSearchNow && searchOn ? { next_search_at: now } : {}),
-      })
-      .where("id", "=", sub.id)
-      .returningAll()
-      .executeTakeFirstOrThrow();
+      .transaction()
+      .execute(async (trx) => {
+        const r = await trx
+          .updateTable("subscription")
+          .set({
+            status: to,
+            status_reason: reason,
+            ...(to === "paused" || to === "cancelled" ? { next_search_at: null } : {}),
+            ...(extra.nextSearchNow && searchOn ? { next_search_at: now } : {}),
+          })
+          .where("id", "=", sub.id)
+          .returningAll()
+          .executeTakeFirstOrThrow();
+        await emitSubscriptionChange(trx, sub, r, ["status"], reason);
+        return r;
+      });
     // 私有规则组随订阅删除(取消),重新订阅时按 params 重建
     if (to === "cancelled" || sub.status === "cancelled") await syncPrivateRuleGroup(db(), r, now);
     invalidateCoverage(sub.id);

apps/server/src/modules/acquisition-organize/services/acquisition.ts

@@ -1,4 +1,10 @@
-import type { AcquisitionPolicyRules, AcquisitionPriority, CandidateEvaluation, SlotTemplate } from "@umt/contracts";
+import {
+  type AcquisitionPolicyRules,
+  type AcquisitionPriority,
+  type CandidateEvaluation,
+  lifecycleKey,
+  type SlotTemplate,
+} from "@umt/contracts";
 import type { AcquisitionSlotRow, MediaEntityRow, SourceReleaseRow, SubscriptionRow } from "@umt/db";
 import { sql } from "@umt/db";
 import {
@@ -23,6 +29,7 @@ import {
   unitsOverlap,
 } from "@umt/domain";
 import { toContractSlotEvaluation } from "@umt/mapping";
+import { hasHook, rewriteHook } from "../../../hooks/index.ts";
 import { hashJson, jsonb, startOfDay } from "../util.ts";
 import {
   type AoDeps,
@@ -38,6 +45,7 @@ import {
   recordCandidate,
 } from "./common.ts";
 import { autoFillCheck, type ForceFill } from "./evaluate-autofill.ts";
+import { emitSubscriptionEvent } from "./lifecycle-events.ts";
 import { contextOf, episodeResolutionOf, parsedOf, unitKeysOf } from "./release-facts.ts";
 import { entityOriginalLanguage, loadRuleCatalog, type RuleCatalog } from "./rules.ts";
 import {
@@ -430,7 +438,12 @@ async function reevaluate(c: Ctx, input: EvaluateInput): Promise<EvaluateOutcome
     focus,
     rels.filter((x) => !seen.has(x.id)),
   );
-  const ordered = await rankReleases(c, entity, sub, focus[0], [...deferred, ...fresh]);
+  const ranked = await rankReleases(c, entity, sub, focus[0], [...deferred, ...fresh]);
+  const ordered = await selectResources(c, ranked, {
+    mediaEntityId: entity.id,
+    subscriptionId: sub?.id ?? null,
+    slotId: input.acquisitionSlotId ?? null,
+  });
   const scope = input.acquisitionSlotId ? { slotId: input.acquisitionSlotId } : sub ? { subscriptionId: sub.id } : {};
   for (const r of ordered) await evaluateRelease(c, r, [], scope);
   return c.out;
@@ -500,6 +513,33 @@ async function rankReleases(
   })).map((x) => byParsed.get(x.candidate)!);
 }
 
+/**
+ * 链式钩子 resource.selection(ADR-006):在择优排序之后调整顺序或过滤。钩子只能重排 / 删减,
+ * 返回里不认识的 ID 忽略、重复的只取第一次;没有钩子时原样返回。
+ */
+export async function selectResources(
+  c: Ctx,
+  rels: SourceReleaseRow[],
+  context: { mediaEntityId: string; subscriptionId: string | null; slotId: string | null },
+): Promise<SourceReleaseRow[]> {
+  if (!hasHook(c.d.ctx, "resource.selection") || rels.length === 0) return rels;
+  const byId = new Map(rels.map((r) => [r.id, r]));
+  const picked = await rewriteHook(
+    c.d.ctx,
+    "resource.selection",
+    rels.map((r) => ({ sourceReleaseId: r.id, title: r.title, parsed: parsedOf(r) })),
+    context,
+  );
+  const out: SourceReleaseRow[] = [];
+  for (const p of picked) {
+    const r = byId.get(p.sourceReleaseId);
+    if (!r) continue;
+    byId.delete(p.sourceReleaseId);
+    out.push(r);
+  }
+  return out;
+}
+
 /** 评估范围:只评某个收录版本,或只评某个订阅的收录版本;缺省评作品的全部收录版本 */
 interface EvalScope {
   slotId?: string | undefined;
@@ -600,8 +640,9 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon
           policy_revision_id: pol.row.id,
           created_by: `policy:${pol.row.policy_key}`,
         })
-        .returning("id")
+        .returningAll()
         .executeTakeFirstOrThrow();
+      await emitSubscriptionEvent(c.d, trx, "subscription.created", sub);
       await createSlotsFromTemplates(trx, {
         entityId: entity.id,
         subscriptionId: sub.id,
@@ -634,6 +675,10 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon
         .where("id", "=", s.id)
         .execute();
       await clearSubscriptionDeferrals(db, s.id);
+      await emitSubscriptionEvent(c.d, db, "subscription.completed", s, {
+        reason: "stop_after_reached",
+        key: lifecycleKey("subscription.completed", s.id, `stop_after:${stopAfter}`),
+      });
       s.status = "completed";
     }
   }

apps/server/src/modules/acquisition-organize/services/downloads.ts

@@ -35,6 +35,7 @@ import {
 import { mergeObservedDownloadState, toDomainDownloadStatus } from "@umt/mapping";
 import { withAttemptToken } from "@umt/tasks";
 import { createEntityEpisodeMapper, publishedAtOf } from "../../sources-identity/service/episode-resolution.ts";
+import { emitMediaWatched } from "../../sources-identity/service/lifecycle-events.ts";
 import { enqueueWatchedRefresh } from "../../sources-identity/service/watched.ts";
 import { errMessage, jsonb } from "../util.ts";
 import { parsedOf } from "./acquisition.ts";
@@ -52,6 +53,7 @@ import {
   timeSlot,
   upsertDecision,
 } from "./common.ts";
+import { emitDownloadLifecycle } from "./lifecycle-events.ts";
 import { matchesQbSource, qbSourceTags } from "./organize-config.ts";
 import { hnrSatisfied, retentionElapsed } from "./retention.ts";
 import {
@@ -212,6 +214,8 @@ async function emitStatus(
       ? { idempotencyKey: generation > 0 ? `download.completed:${id}:g${generation}` : `download.completed:${id}` }
       : {},
   );
+  // 生命周期广播(ADR-006):下载已完成 / 失败 / 已移除
+  await emitDownloadLifecycle(d, exec, id, from, to);
 }
 
 // ── 提交 ──
@@ -269,7 +273,10 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
     .executeTakeFirstOrThrow();
   // 开始下载 = 作品进入关注范围:从没刷新过元数据的(自动建档时不在关注范围)补排一次 full 刷新(集数两层,见 watched.ts)
   const watchedEntity = ev.mediaEntityId ?? rel.media_entity_id;
-  if (watchedEntity) await enqueueWatchedRefresh(d.ctx, watchedEntity, "download");
+  if (watchedEntity) {
+    await enqueueWatchedRefresh(d.ctx, watchedEntity, "download");
+    await emitMediaWatched(d.ctx, db, watchedEntity, "download", plan.id);
+  }
   const site = await db.selectFrom("site").selectAll().where("id", "=", rel.site_id).executeTakeFirst();
   const client = await d.ports.downloader(topo.inst);
   const tag = topo.inst.managed_tag;

apps/server/src/modules/acquisition-organize/services/lifecycle-events.ts

@@ -0,0 +1,140 @@
+import { type DownloadStatus, lifecycleKey, type OrganizationStatus } from "@umt/contracts";
+import type { ScrapeRecordRow, SubscriptionRow } from "@umt/db";
+import type { AoDeps, Exec } from "./common.ts";
+
+// 生命周期广播事件(ADR-006)在本模块的发出点:都在调用方的业务事务里写 outbox,订阅方按 ID 回查业务表。
+
+type SubEvent = "subscription.created" | "subscription.updated" | "subscription.completed" | "subscription.cancelled";
+
+type SubRef = Pick<SubscriptionRow, "id" | "media_entity_id" | "media_kind" | "status" | "origin">;
+
+/** 订阅已建 / 已改 / 已完成 / 已取消 */
+export async function emitSubscriptionEvent(
+  d: AoDeps,
+  exec: Exec,
+  type: SubEvent,
+  sub: SubRef,
+  extra: { reason?: string; changes?: string[]; key?: string } = {},
+): Promise<void> {
+  const ref = { subscriptionId: sub.id, mediaEntityId: sub.media_entity_id, mediaKind: sub.media_kind };
+  const opts = extra.key ? { idempotencyKey: extra.key } : {};
+  switch (type) {
+    case "subscription.created":
+      return d.ctx.outbox.emit(
+        exec,
+        type,
+        { ...ref, origin: sub.origin, status: sub.status },
+        { idempotencyKey: lifecycleKey(type, sub.id) },
+      );
+    case "subscription.updated":
+      return d.ctx.outbox.emit(exec, type, { ...ref, status: sub.status, changes: extra.changes ?? [] }, opts);
+    case "subscription.completed":
+    case "subscription.cancelled":
+      return d.ctx.outbox.emit(exec, type, { ...ref, reason: extra.reason ?? "" }, opts);
+  }
+}
+
+const DOWNLOAD_END: Partial<Record<DownloadStatus, "download.completed" | "download.failed" | "download.removed">> = {
+  completed: "download.completed",
+  failed: "download.failed",
+  removed: "download.removed",
+};
+
+/** 下载状态进入 completed / failed / removed 时发对应的生命周期事件(每代每种只发一次) */
+export async function emitDownloadLifecycle(
+  d: AoDeps,
+  exec: Exec,
+  id: string,
+  from: DownloadStatus | null,
+  to: DownloadStatus,
+): Promise<void> {
+  const type = DOWNLOAD_END[to];
+  if (!type || from === to) return;
+  const row = await exec
+    .selectFrom("download_task as t")
+    .leftJoin("acquisition_plan as p", "p.id", "t.acquisition_plan_id")
+    .leftJoin("acquisition_slot as s", "s.id", "p.slot_id")
+    .select([
+      "t.name",
+      "t.generation",
+      "t.acquisition_plan_id",
+      "t.media_entity_id",
+      "s.media_entity_id as slot_entity",
+    ])
+    .where("t.id", "=", id)
+    .executeTakeFirst();
+  if (!row) return;
+  const ref = {
+    downloadTaskId: id,
+    planId: row.acquisition_plan_id,
+    mediaEntityId: row.media_entity_id ?? row.slot_entity ?? null,
+    name: row.name,
+    generation: row.generation,
+  };
+  const key = { idempotencyKey: lifecycleKey(type, id, `g${row.generation}`) };
+  if (type === "download.completed") await d.ctx.outbox.emit(exec, type, ref, key);
+  else await d.ctx.outbox.emit(exec, type, { ...ref, from }, key);
+}
+
+const FINISHED: readonly OrganizationStatus[] = ["succeeded", "partial", "failed"];
+
+/** 整理记录进入终态时发 organization.finished(按 change_seq 去重:同一次状态变化只发一次) */
+export async function emitOrganizationFinished(
+  d: AoDeps,
+  exec: Exec,
+  r: {
+    id: string;
+    status: OrganizationStatus;
+    action: string;
+    trigger: string;
+    media_entity_id: string | null;
+    download_task_id: string | null;
+    reason: string | null;
+    change_seq: number | string;
+  },
+): Promise<void> {
+  if (!FINISHED.includes(r.status)) return;
+  await d.ctx.outbox.emit(
+    exec,
+    "organization.finished",
+    {
+      recordId: r.id,
+      outcome: r.status as "succeeded" | "partial" | "failed",
+      action: r.action as never,
+      trigger: r.trigger as never,
+      mediaEntityId: r.media_entity_id,
+      downloadTaskId: r.download_task_id,
+      reason: r.reason,
+      changeSeq: String(r.change_seq),
+    },
+    { idempotencyKey: lifecycleKey("organization.finished", r.id, String(r.change_seq)) },
+  );
+}
+
+/** 刮削记录结束:failed → scrape.failed;succeeded / partial / skipped → scrape.completed */
+export async function emitScrapeFinished(d: AoDeps, row: ScrapeRecordRow): Promise<void> {
+  if (row.status === "queued" || row.status === "running") return;
+  const ref = {
+    scrapeRecordId: row.id,
+    mediaEntityId: row.media_entity_id,
+    organizationRecordId: row.organization_record_id,
+    trigger: row.trigger,
+    status: row.status,
+    written: row.written,
+    failed: row.failed,
+  };
+  const at = row.finished_at?.getTime() ?? 0;
+  if (row.status === "failed")
+    await d.ctx.outbox.emit(
+      d.ctx.db,
+      "scrape.failed",
+      { ...ref, reason: row.reason },
+      {
+        idempotencyKey: lifecycleKey("scrape.failed", row.id, at),
+      },
+    );
+  else
+    await d.ctx.outbox.emit(d.ctx.db, "scrape.completed", ref, {
+      idempotencyKey: lifecycleKey("scrape.completed", row.id, at),
+    });
+}

apps/server/src/modules/acquisition-organize/services/organize-hooks.ts

@@ -0,0 +1,89 @@
+import { basename, dirname, extname, join } from "node:path";
+import type { OrganizePlan } from "@umt/domain";
+import { hasHook, interceptHook, rewriteHook } from "../../../hooks/index.ts";
+import type { ServerContext } from "../../types.ts";
+
+// 整理计划上的链式钩子(ADR-006):organize.rename 改写主视频目标文件名、organize.intercept 拒绝整理。
+// 都在计划生成之后、写整理记录之前调用;没有注册钩子时什么都不做。
+
+const BAD_NAME = /[\\/\0]/;
+const usableName = (n: unknown): boolean =>
+  typeof n === "string" && n.trim() !== "" && !BAD_NAME.test(n) && n !== "." && n !== "..";
+
+/**
+ * organize.rename:逐个主视频把模板生成的目标文件名交给钩子改写,同主干的附属文件(字幕、NFO 等)跟着换主干。
+ * 只改文件名不改目录;返回带路径分隔符或空名的钩子结果跳过。改名后撞名的项转 hold + duplicate_target(与保留原名同一处理)。
+ * 改名后的目标是否已存在由执行阶段按冲突处理(计划期的存在性探测在改名之前)。
+ */
+export async function applyRenameHook(
+  ctx: Pick<ServerContext, "hooks">,
+  plan: OrganizePlan,
+  base: { mediaEntityId: string; mediaType: "movie" | "tv" },
+): Promise<number> {
+  if (!hasHook(ctx, "organize.rename")) return 0;
+  const renamed = new Map<string, string>();
+  for (const main of plan.files.filter((f) => f.role === "main")) {
+    const name = basename(main.dst);
+    // 返回带路径分隔符或空名的钩子结果跳过(记日志),其余钩子照常
+    const next = await rewriteHook(
+      ctx,
+      "organize.rename",
+      name,
+      {
+        targetPath: main.dst,
+        sourcePath: main.src,
+        mediaEntityId: base.mediaEntityId,
+        mediaType: base.mediaType,
+        season: main.season ?? null,
+        episodes: main.episodes ?? null,
+      },
+      usableName,
+    );
+    if (next === name) continue;
+    const oldStem = name.slice(0, name.length - extname(name).length);
+    const newStem = next.slice(0, next.length - extname(next).length);
+    const to = join(dirname(main.dst), next);
+    renamed.set(main.dst, to);
+    main.dst = to;
+    for (const f of plan.files) {
+      if (f.mainSrc !== main.src) continue;
+      const side = basename(f.dst);
+      if (!side.startsWith(oldStem)) continue;
+      const nd = join(dirname(f.dst), newStem + side.slice(oldStem.length));
+      renamed.set(f.dst, nd);
+      f.dst = nd;
+    }
+  }
+  if (renamed.size === 0) return 0;
+  for (const c of plan.conflicts) if (c.dst && renamed.has(c.dst)) c.dst = renamed.get(c.dst)!;
+  for (const pc of plan.preconditions)
+    if (pc.kind === "target_absent" || pc.kind === "verify_after_copy")
+      pc.paths = pc.paths.map((p) => renamed.get(p) ?? p);
+  const byDst = new Map<string, OrganizePlan["files"]>();
+  for (const f of plan.files) byDst.set(f.dst, [...(byDst.get(f.dst) ?? []), f]);
+  for (const [dst, fs] of byDst) {
+    if (fs.length < 2 || plan.conflicts.some((c) => c.kind === "duplicate_target" && c.dst === dst)) continue;
+    for (const f of fs) f.action = "hold";
+    plan.conflicts.push({ kind: "duplicate_target", blocking: true, dst, detail: fs.map((f) => f.src).join(" | ") });
+  }
+  plan.requiresApproval = plan.conflicts.some((c) => c.blocking);
+  return renamed.size;
+}
+
+/** organize.intercept:返回拒绝原因(含拒绝的钩子名),放行返回 null */
+export async function organizeIntercept(
+  ctx: Pick<ServerContext, "hooks">,
+  plan: OrganizePlan,
+  base: {
+    mediaEntityId: string;
+    mediaType: "movie" | "tv";
+    action: string;
+    trigger: string;
+    downloadTaskId: string | null;
+  },
+) {
+  return interceptHook(ctx, "organize.intercept", () => ({
+    ...base,
+    files: plan.files.map((f) => ({ src: f.src, dst: f.dst, kind: f.kind, role: f.role })),
+  }));
+}

apps/server/src/modules/acquisition-organize/services/organize-plan.ts

@@ -42,10 +42,12 @@ import { publishedAcquisitionPolicies, unitKeysOf } from "./acquisition.ts";
 import { classifyEntity } from "./classification.ts";
 import { type AoDeps, type Exec, hostPathOf, loadInstances, upsertDecision } from "./common.ts";
 import { applySlotEvent } from "./downloads.ts";
+import { emitOrganizationFinished } from "./lifecycle-events.ts";
 import { effectiveNaming, ensureNamingVersion, loadNamingEntity, namingByVersion } from "./naming.ts";
 import { resolveOrganizeSource } from "./organize-config.ts";
 import { PathGuardError } from "./organize-guard.ts";
 import { successfulOrganizeCounts } from "./organize-history.ts";
+import { applyRenameHook, organizeIntercept } from "./organize-hooks.ts";
 import { planTransfer } from "./organize-transfer.ts";
 // 入库:冻结整理计划(organization_record + 文件明细)→ 逐文件可恢复执行(操作账本 fencing)→
 // 资产/位置/路径绑定同事务登记 → Slot 推进 → Move 后移除 qB 任务记录(不删数据)。
@@ -925,8 +927,19 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
         plan.requiresApproval = true;
       }
   }
+  // 链式钩子 organize.rename(ADR-006):模板命名时可改写主视频目标文件名
+  if (!inPlace && !keepNames) await applyRenameHook(d.ctx, plan, { mediaEntityId: entity.id, mediaType });
   if (keepNames && !inPlace) keepOriginalNames(plan);
   if (req.keepBoth && !inPlace) await keepBothTargets(plan);
+  // 链式钩子 organize.intercept:返回原因即拒绝本次整理(记为失败)
+  const veto = await organizeIntercept(d.ctx, plan, {
+    mediaEntityId: entity.id,
+    mediaType,
+    action,
+    trigger: req.trigger,
+    downloadTaskId: dt?.id ?? null,
+  });
+  if (veto) return fail(`intercepted:${veto.reason}`, { hook: veto.by, reason: veto.reason }, entity.id);
   const frozen: FrozenPlan = {
     version: 1,
     ...(req.scrapeAfterImport !== undefined ? { scrapeAfterImport: req.scrapeAfterImport } : {}),
@@ -1288,7 +1301,7 @@ export async function writeFailedRecord(
 export async function emitRecord(d: AoDeps, exec: Exec, recordId: string): Promise<void> {
   const r = await exec
     .selectFrom("organization_record")
-    .select(["status", "change_seq", "media_entity_id"])
+    .select(["id", "status", "change_seq", "media_entity_id", "action", "trigger", "download_task_id", "reason"])
     .where("id", "=", recordId)
     .executeTakeFirstOrThrow();
   await d.ctx.outbox.emit(exec, "organization.changed", {
@@ -1297,6 +1310,8 @@ export async function emitRecord(d: AoDeps, exec: Exec, recordId: string): Promi
     changeSeq: String(r.change_seq),
     mediaEntityId: r.media_entity_id,
   });
+  // 终态另发生命周期广播(ADR-006):整理成功 / 部分成功 / 失败
+  await emitOrganizationFinished(d, exec, r);
 }
 
 /**

apps/server/src/modules/acquisition-organize/services/scrape.ts

@@ -30,6 +30,7 @@ import {
 } from "../../sources-identity/service/episode-resolution.ts";
 import { errMessage, jsonb } from "../util.ts";
 import type { AoDeps } from "./common.ts";
+import { emitScrapeFinished } from "./lifecycle-events.ts";
 import {
   type EntityForNfo,
   episodeMetadataMissing,
@@ -930,6 +931,7 @@ export async function runScrapeRecord(
       .execute()
       .catch(() => {});
     if (dirs.size) void d.ctx.events.publish("files.changed", { rootId: rec.root_id, dirs: [...dirs] });
+    await emitScrapeFinished(d, row);
     return scrapeRecordView(row);
   };
   const fail = (reason: string) => finish({ status: "failed", reason, files: [] });

apps/server/src/modules/acquisition-organize/services/subscriptions.ts

@@ -1,4 +1,10 @@
-import type { AcquisitionSlot, SubscriptionParams, SubscriptionProgress, UnitProgressState } from "@umt/contracts";
+import {
+  type AcquisitionSlot,
+  lifecycleKey,
+  type SubscriptionParams,
+  type SubscriptionProgress,
+  type UnitProgressState,
+} from "@umt/contracts";
 import type { AcquisitionSlotRow, SourceReleaseRow, SubscriptionRow } from "@umt/db";
 import { sql } from "@umt/db";
 import {
@@ -20,8 +26,10 @@ import {
   type UnitRange,
 } from "@umt/domain";
 import { slotSpecFromRow } from "@umt/mapping";
+import { interceptHook } from "../../../hooks/index.ts";
 import { errMessage, jsonb, toApi } from "../util.ts";
 import { type AoDeps, type Exec, upsertDecision } from "./common.ts";
+import { emitSubscriptionEvent } from "./lifecycle-events.ts";
 import { contextOf, parsedOf, unitKeysOf } from "./release-facts.ts";
 import { entityOriginalLanguage, loadRuleCatalog, mergeHard, type RuleCatalog, resolveRuleChain } from "./rules.ts";
 
@@ -868,6 +876,18 @@ export async function completeSubscription(
   reason: string,
   progress: Record<string, unknown> = {},
 ): Promise<boolean> {
+  // 链式钩子 subscription.completion.check(ADR-006):追加完成条件,拒绝时订阅继续追踪
+  const veto = await interceptHook(d.ctx, "subscription.completion.check", () => ({
+    subscriptionId: sub.id,
+    mediaEntityId: sub.media_entity_id,
+    mediaKind: sub.media_kind,
+    reason,
+    progress,
+  }));
+  if (veto) {
+    d.ctx.logger.info({ subscriptionId: sub.id, reason, veto }, "订阅完成被钩子拦下，继续追踪");
+    return false;
+  }
   const done = await trx
     .updateTable("subscription")
     .set({ status: "completed", status_reason: reason, next_search_at: null })
@@ -877,7 +897,7 @@ export async function completeSubscription(
     .executeTakeFirst();
   if (!done) return false;
   await clearSubscriptionDeferrals(trx, done.id);
-  await trx
+  const history = await trx
     .insertInto("subscription_history")
     .values({
       subscription_id: done.id,
@@ -890,7 +910,12 @@ export async function completeSubscription(
       progress: jsonb(progress),
       completed_at: d.ports.now(),
     })
-    .execute();
+    .returning("id")
+    .executeTakeFirstOrThrow();
+  await emitSubscriptionEvent(d, trx, "subscription.completed", done, {
+    reason,
+    key: lifecycleKey("subscription.completed", done.id, history.id),
+  });
   d.ctx.logger.info({ subscriptionId: sub.id, reason }, "订阅完成，已写历史");
   return true;
 }

apps/server/src/modules/sources-identity/service/account.ts

@@ -30,6 +30,7 @@ import type { SourcesRuntime } from "../runtime.ts";
 import { capabilitiesOf, LOGIN_KINDS } from "../site-kinds.ts";
 import { type Actor, audit, closeDecisions, type Exec, openDecision, SYSTEM } from "./common.ts";
 import { cookieFromBrowser, writeBrowserCookie } from "./credentials.ts";
+import { emitSiteReachability } from "./lifecycle-events.ts";
 import { browserOnlyCookiesOf, resolveUserAgent, type UserAgentDefaults } from "./user-agent.ts";
 
 // W9-A 站点中心(docs/design/site-center.md):凭证写入、刷新登录(账密 takelogin / Cookie / API Key 校验)、
@@ -865,12 +866,20 @@ export async function checkSiteHealth(
       ? (HEALTH_OF[status] ?? "degraded")
       : site.health_status;
   const last = { ok, status, latencyMs, message, detail };
+  // 生命周期广播(ADR-006):登录失效 / 访问不了 ⇄ 正常,只在跨越时发
+  const bad = (h: string, st: unknown) => h === "failing" || st === "unreachable";
+  const wasBad = bad(site.health_status, (site.last_health as { status?: unknown } | null)?.status);
+  const isBad = site.enabled && bad(healthStatus, status);
   await rt.ctx.db.transaction().execute(async (trx) => {
     await trx
       .updateTable("site")
       .set({ last_health: jsonb(last), last_health_at: now, health_status: healthStatus })
       .where("id", "=", site.id)
       .execute();
+    if (isBad && !wasBad)
+      await emitSiteReachability(rt.ctx, trx, site.id, { reachable: false, reason: status, message });
+    else if (wasBad && healthStatus === "healthy")
+      await emitSiteReachability(rt.ctx, trx, site.id, { reachable: true, reason: "health_check" });
     // 首页显示已登录:顺手记上次登录
     if (ok) {
       await trx

apps/server/src/modules/sources-identity/service/episode-groups.ts

@@ -8,12 +8,13 @@ import {
   TmdbError,
   type TmdbRichDetail,
 } from "@umt/adapters";
-import type {
-  EpisodeGroupList,
-  EpisodeGroupSetResult,
-  EpisodeStructureCurrent,
-  EpisodeStructureSeason,
-  EpisodeStructureSeasons,
+import {
+  type EpisodeGroupList,
+  type EpisodeGroupSetResult,
+  type EpisodeStructureCurrent,
+  type EpisodeStructureSeason,
+  type EpisodeStructureSeasons,
+  lifecycleKey,
 } from "@umt/contracts";
 import { sql } from "@umt/db";
 import { ApiException } from "../../../http/errors.ts";
@@ -357,6 +358,13 @@ export async function setEpisodeGroup(
       },
     );
     const reresolve = await reresolveEntityEpisodes(rt.ctx, trx, tv.id);
+    // 生命周期广播(ADR-006):季集结构已切换
+    await rt.ctx.outbox.emit(
+      trx,
+      "media.episode_structure_changed",
+      { mediaEntityId: tv.id, fromGroupId: fromId, toGroupId: groupId },
+      { idempotencyKey: lifecycleKey("media.episode_structure_changed", tv.id, now.toISOString()) },
+    );
     return { ...r, reresolve };
   });
   let refresh = "refreshed";

apps/server/src/modules/sources-identity/service/lifecycle-events.ts

@@ -0,0 +1,55 @@
+import { lifecycleKey } from "@umt/contracts";
+import type { DB, Kysely, Transaction } from "@umt/db";
+import type { ServerContext } from "../../types.ts";
+
+// 生命周期广播事件(ADR-006)在来源与识别侧的发出点。元数据刷新、进入关注的判定本身在 metadata.ts / watched.ts,
+// 这里只在调用方发事件,不改那两处。
+
+type Exec = Kysely<DB> | Transaction<DB>;
+
+/**
+ * 作品进入关注:建订阅、开始下载、用户搜索 / 手动认定时由调用方发。
+ * 同一作品同一触发对象(订阅、下载计划、资源)只发一次;订阅方自己判断作品是否已经处理过。
+ */
+export async function emitMediaWatched(
+  ctx: Pick<ServerContext, "outbox">,
+  exec: Exec,
+  mediaEntityId: string,
+  reason: string,
+  subjectId: string,
+): Promise<void> {
+  await ctx.outbox.emit(
+    exec,
+    "media.watched",
+    { mediaEntityId, reason },
+    { idempotencyKey: lifecycleKey("media.watched", mediaEntityId, reason, subjectId) },
+  );
+}
+
+/** 作品元数据刷新完成(刷新任务的调用方发;按任务 run 去重) */
+export async function emitMetadataRefreshed(
+  ctx: Pick<ServerContext, "outbox">,
+  exec: Exec,
+  input: { mediaEntityId: string; full: boolean; changedFields: string[]; runId: string },
+): Promise<void> {
+  await ctx.outbox.emit(
+    exec,
+    "media.metadata_refreshed",
+    { mediaEntityId: input.mediaEntityId, full: input.full, changedFields: input.changedFields },
+    { idempotencyKey: lifecycleKey("media.metadata_refreshed", input.mediaEntityId, input.runId) },
+  );
+}
+
+/** 站点不可达 / 恢复。只在状态跨越时由调用方发(连续失败到阈值、登录失效、检测结果变化) */
+export async function emitSiteReachability(
+  ctx: Pick<ServerContext, "outbox">,
+  exec: Exec,
+  siteId: string,
+  change: { reachable: false; reason: string; message: string | null } | { reachable: true; reason: string },
+  key?: string,
+): Promise<void> {
+  const opts = key ? { idempotencyKey: key } : {};
+  if (change.reachable) await ctx.outbox.emit(exec, "site.recovered", { siteId, reason: change.reason }, opts);
+  else
+    await ctx.outbox.emit(exec, "site.unreachable", { siteId, reason: change.reason, message: change.message }, opts);
+}

apps/server/src/modules/sources-identity/service/resolve.ts

@@ -21,6 +21,7 @@ import {
   toParsedColumns,
   toReleaseResolutionStatus,
 } from "@umt/mapping";
+import { rewriteHook } from "../../../hooks/index.ts";
 import { jsonb } from "../rows.ts";
 import type { SourcesRuntime } from "../runtime.ts";
 import { addSiteAliases, type SiteAliasOutcome } from "./alias-guard.ts";
@@ -52,6 +53,7 @@ import {
   openGroupDecision,
   refreshGroupDecision,
 } from "./identity-decisions.ts";
+import { emitMediaWatched } from "./lifecycle-events.ts";
 import { loadSubject, searchBangumi } from "./meta-subjects.ts";
 import { tmdbRecallInput, tmdbToCandidate } from "./tmdb-candidate.ts";
 import { loadRecognizeCache, type RecognizeMeta, recognizeMetaOf, saveRecognizeCache } from "./tmdb-recognize-cache.ts";
@@ -480,7 +482,18 @@ export async function resolveRelease(rt: SourcesRuntime, input: ResolveInput): P
       mediaEntityId: release.media_entity_id,
     };
   }
-  const parsed = parseRelease(release.title, release.subtitle ? { subtitle: release.subtitle } : {});
+  // 链式钩子 name.recognize(ADR-006):可改写解析结果;没有钩子时原样
+  const parsed = await rewriteHook(
+    ctx,
+    "name.recognize",
+    parseRelease(release.title, release.subtitle ? { subtitle: release.subtitle } : {}),
+    () => ({
+      title: release.title,
+      subtitle: release.subtitle ?? null,
+      sourceReleaseId: release.id,
+      trigger: input.trigger ?? (input.force ? "review" : "observed"),
+    }),
+  );
   const cols = toParsedColumns(parsed);
   // 站点给出的外部 ID(列表 / 详情页的 IMDb、豆瓣、TMDB 链接)作为显式 ID 证据参与打分
   const siteIds = siteExplicitIds(release.raw);
@@ -909,6 +922,9 @@ async function bindAuto(
       } else if (userAsked && result.mediaEntityId) {
         await enqueueWatchedRefresh(ctx, result.mediaEntityId, `lookup:${lookup}`);
       }
+      // 用户搜索 / 手动认定 = 作品进入关注(ADR-006 生命周期广播)
+      if (userAsked && result.mediaEntityId)
+        await emitMediaWatched(ctx, ctx.db, result.mediaEntityId, `lookup:${lookup}`, release.id);
       return result;
     } catch (e) {
       if (e instanceof IdentityConflictError && attempt < 2) continue;
@@ -1160,13 +1176,20 @@ export interface NameIdentifyResult {
  */
 export async function identifyParsed(
   rt: SourcesRuntime,
-  parsed: ParsedRelease,
+  input: ParsedRelease,
   opts: { trigger?: ResolveTrigger; evidence?: Record<string, unknown> } = {},
 ): Promise<NameIdentifyResult> {
   const { ctx } = rt;
   const none = { mediaEntityId: null, contentUnitIds: [] as string[], candidates: [], created: false };
-  if (!parsed.title || parsed.auxiliaryOnly) return { outcome: "unparseable", reason: "unparseable_title", ...none };
   const trigger = opts.trigger ?? "takeover";
+  // 链式钩子 name.recognize(ADR-006):整理 / 接管按名称识别时同样可改写解析结果
+  const parsed = await rewriteHook(ctx, "name.recognize", input, () => ({
+    title: input.raw,
+    subtitle: null,
+    sourceReleaseId: null,
+    trigger,
+  }));
+  if (!parsed.title || parsed.auxiliaryOnly) return { outcome: "unparseable", reason: "unparseable_title", ...none };
   const local = await localCandidates(ctx.db, parsed, null);
   const lookup = await lookupReason(rt, parsed, trigger, local.length > 0, null);
   let remote: MetadataCandidate[] = [];

apps/server/src/modules/sources-identity/service/snapshot.ts

@@ -1,6 +1,6 @@
 import { createHash } from "node:crypto";
 import { type SiteDetailPage, SiteError, type SitePage, type SiteRelease } from "@umt/adapters";
-import type { SiteSnapshotKind } from "@umt/contracts";
+import { lifecycleKey, type SiteSnapshotKind } from "@umt/contracts";
 import { type SiteRow, sql } from "@umt/db";
 import { parseRelease } from "@umt/domain";
 import { siteReleaseToSourceRelease } from "@umt/mapping";
@@ -10,6 +10,7 @@ import { siteCan } from "../site-kinds.ts";
 import { enqueueRelogin } from "./account.ts";
 import { closeDecisions, type Exec, openDecision } from "./common.ts";
 import { hasDetailPage, saveDetailPage } from "./detail-page.ts";
+import { emitSiteReachability } from "./lifecycle-events.ts";
 
 // 站点快照(周期列表)、补漏(Feed 疑似缺口,有页数预算)、实时补查(工作台搜索)。
 // 所有请求经站点共享限流器;结果按 site+站点资源 ID upsert 到 source_release,并记录 site_snapshot 覆盖范围。
@@ -114,6 +115,12 @@ async function onAuthError(rt: SourcesRuntime, trx: Exec, site: SiteRow, e: Site
     .where("status", "in", ["unknown", "active"])
     .execute();
   await trx.updateTable("site").set({ health_status: "failing" }).where("id", "=", site.id).execute();
+  if (site.health_status !== "failing")
+    await emitSiteReachability(rt.ctx, trx, site.id, {
+      reachable: false,
+      reason: e.kind,
+      message: e.message.slice(0, 300),
+    });
   await openDecision(rt.ctx, trx, {
     kind: "source_unreachable",
     severity: "warning",
@@ -247,6 +254,23 @@ export async function snapshotSite(rt: SourcesRuntime, input: SnapshotInput): Pr
       const prevStreak = Number((site.snapshot_cursor as { failStreak?: unknown } | null)?.failStreak) || 0;
       const failStreak = status === "failed" ? prevStreak + 1 : 0;
       await syncSnapshotFailureDecision(rt, trx, site, failStreak, errJson);
+      // 生命周期广播(ADR-006):连续失败到开待处理的阈值 = 不可达;从阈值以上恢复 = 恢复
+      if (failStreak === SNAPSHOT_FAIL_DECISION_AT)
+        await emitSiteReachability(
+          rt.ctx,
+          trx,
+          site.id,
+          { reachable: false, reason: "snapshot_failed", message: errJson?.message.slice(0, 300) ?? null },
+          lifecycleKey("site.unreachable", site.id, snap.id),
+        );
+      else if (failStreak === 0 && prevStreak >= SNAPSHOT_FAIL_DECISION_AT)
+        await emitSiteReachability(
+          rt.ctx,
+          trx,
+          site.id,
+          { reachable: true, reason: "snapshot_ok" },
+          lifecycleKey("site.recovered", site.id, snap.id),
+        );
       await trx
         .updateTable("site")
         .set({

apps/server/src/modules/sources-identity/tasks.ts

@@ -5,6 +5,7 @@ import type { SourcesRuntime } from "./runtime.ts";
 import { refreshSiteAccount } from "./service/account.ts";
 import { AUTO_ENRICH_TASK, runAutoEnrich } from "./service/enrich.ts";
 import { ingestFeed } from "./service/feeds.ts";
+import { emitMetadataRefreshed } from "./service/lifecycle-events.ts";
 import { refreshMetadata } from "./service/metadata.ts";
 import { resolveRelease } from "./service/resolve.ts";
 import { snapshotSite } from "./service/snapshot.ts";
@@ -75,7 +76,18 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu
       inputSchema: z.object({ mediaEntityId: id, full: z.boolean().optional() }),
       retryPolicy: NETWORK_RETRY,
       handler: (t) =>
-        t.step.run("refresh", () => refreshMetadata(rt(), t.input.mediaEntityId, { full: !!t.input.full })),
+        t.step.run("refresh", async () => {
+          const r = await refreshMetadata(rt(), t.input.mediaEntityId, { full: !!t.input.full });
+          // 生命周期广播(ADR-006):元数据已刷新(在调用方发,刷新本身在 metadata.ts)
+          if (r.status === "refreshed")
+            await emitMetadataRefreshed(rt().ctx, rt().ctx.db, {
+              mediaEntityId: t.input.mediaEntityId,
+              full: !!t.input.full,
+              changedFields: r.changedFields,
+              runId: t.runId,
+            });
+          return r;
+        }),
     }),
     defineTask({
       name: "searchSubscription",