feat(ao,sources): W5-C subscription completion, Slot × unit reservation, deferral, pack file selection, active search

- completion by season episode_total ledger (applySlotEvent verify_passed hook overrides organize's
  complete): unknown total keeps tracking and records suspected-missing; pending never completes;
  once / continuous current-season completion writes subscription_history
- evaluation: missing = release units ∩ episodeRange − covered − in-flight; series slots may hold
  several unit-disjoint in-flight plans; overlapping/busy releases are deferred (slot_deferral) and
  re-evaluated when a plan ends (next_evaluate_at due) or on the periodic slot re-evaluation
- per-unit upgrade via slot_unit ledger; site scope (params.siteIds); site facts in evaluation context
- season packs: after submit read qB files and setFilePriority(0) for covered/in-flight episodes;
  whole pack with recorded reason when undeterminable; removed downloads release their reservation
- searchSubscription task (title/aliases × enabled sites via liveSearch, snapshot + rate limiter):
  on create, every searchIntervalHours (default 24h) and POST /subscriptions/{id}/search
- refreshMetadata writes season totals (only grows, floored by covered, lock respected) and turns
  metadata_pending subscriptions active; plan loop refreshes tracked series every 6h
- endpoints: subscription progress, episode-total, subscription-history; e9 continuous-series e2e

Co-Authored-By: Claude Opus 5.5 <[email protected]>

apps/server/src/modules/acquisition-organize/__tests__/rules-api.test.ts

@@ -50,7 +50,9 @@ describe.skipIf(!hasPg)("规则 API(真实 PG)", () => {
       mode: "continuous",
       seasonScope: "following_seasons",
       origin: "user",
-      status: "active",
+      // 剧集没有已接受的 TMDB 身份 → 总集数无来源:pending(照常搜索/评估,不能完成)
+      status: "pending",
+      statusReason: "metadata_pending",
     });
     const once = await h.api("PATCH", `/subscriptions/${sub.body.data.id}`, {
       seasonScope: "current_season",

apps/server/src/modules/acquisition-organize/__tests__/subscription.test.ts

@@ -0,0 +1,317 @@
+import { createQbClient } from "@umt/adapters";
+import { createMockQb, createTempTree, type MockQb, makeTorrent, type TempTree } from "@umt/testkit";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { evaluateAcquisition } from "../services/acquisition.ts";
+import { submitDownload, syncDownloads } from "../services/downloads.ts";
+import { executeRecord, planImport } from "../services/organize.ts";
+import { applyEpisodeTotal, subscriptionProgress, sweepDueSlots } from "../services/subscriptions.ts";
+import { createHarness, fakeDisk, type Harness, pgAvailable, seedBase, seedEntity, seedRelease } from "./harness.ts";
+
+// W5-C 订阅增强集成:Slot×单元多在途 + 忙时延后重评 + 移除释放;季包按缺集选文件(mock qB files/filePrio);
+// 总集数未知不结束(疑似缺集);once 完成写历史;pending 不完成;分集洗版只替换能提升的单元。
+
+const hasPg = await pgAvailable();
+const SIZE = 300_000;
+
+describe.skipIf(!hasPg)("订阅增强(真实 PG + mock qB + 临时目录)", () => {
+  let h: Harness;
+  let qb: MockQb;
+  let lib: TempTree;
+  let cache: TempTree;
+  const disk = fakeDisk(10_000_000_000);
+  const torrents = new Map<string, Uint8Array>();
+  let base: Awaited<ReturnType<typeof seedBase>>;
+
+  beforeAll(async () => {
+    qb = await createMockQb({ addVisibleAfterPolls: 0 });
+    lib = await createTempTree({}, { sentinel: true, prefix: "umt-lib-" });
+    cache = await createTempTree({}, { prefix: "umt-cache-" });
+    disk.state.watch = [cache.root];
+    disk.setPct(10);
+    h = await createHarness("subscription", {
+      diskUsage: disk.fn,
+      torrentSource: async ({ release }) => ({ kind: "file", bytes: torrents.get(release.id)! }),
+    });
+    base = await seedBase(h.db, { libraryPath: lib.root, cachePath: cache.root, qbUrl: qb.url });
+  }, 60_000);
+
+  afterAll(async () => {
+    await h?.close();
+    await qb?.close();
+    await lib?.cleanup();
+    await cache?.cleanup();
+  });
+
+  const db = () => h.db;
+
+  async function seriesSlot(
+    title: string,
+    opts: { mode?: "once" | "continuous"; status?: "active" | "pending"; hard?: object; upgrade?: object } = {},
+  ) {
+    const entityId = await seedEntity(h.db, { kind: "series", title, year: 2024 });
+    const sub = await db()
+      .insertInto("subscription")
+      .values({
+        media_entity_id: entityId,
+        media_kind: "series",
+        mode: opts.mode ?? "continuous",
+        status: opts.status ?? "active",
+      })
+      .returning("id")
+      .executeTakeFirstOrThrow();
+    const slot = await db()
+      .insertInto("acquisition_slot")
+      .values({
+        media_entity_id: entityId,
+        subscription_id: sub.id,
+        name: "1080p",
+        hard_conditions: JSON.stringify(opts.hard ?? { resolutions: ["1080p"] }),
+        upgrade_threshold: JSON.stringify(opts.upgrade ?? {}),
+      })
+      .returning("id")
+      .executeTakeFirstOrThrow();
+    return { entityId, subId: sub.id, slotId: slot.id };
+  }
+
+  async function release(entityId: string, title: string, files?: { path: string; length: number }[]) {
+    const t = files
+      ? makeTorrent({ name: title, files, pieceSeed: Math.floor(Math.random() * 1e6) })
+      : makeTorrent({ name: `${title}.mkv`, length: SIZE, pieceSeed: Math.floor(Math.random() * 1e6) });
+    const id = await seedRelease(h.db, { siteId: base.siteId, entityId, title, size: SIZE, hash: t.hash });
+    torrents.set(id, t.bytes);
+    return { id, hash: t.hash, name: t.name };
+  }
+
+  const slotRow = (id: string) =>
+    db().selectFrom("acquisition_slot").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
+  const planRow = (id: string) =>
+    db().selectFrom("acquisition_plan").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
+  const subRow = (id: string) =>
+    db().selectFrom("subscription").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
+
+  async function planFor(relId: string) {
+    const ev = await evaluateAcquisition(h.d, { sourceReleaseId: relId });
+    expect(ev.plans, JSON.stringify(ev)).toHaveLength(1);
+    return ev.plans[0]!;
+  }
+  async function submit(planId: string) {
+    const s = await submitDownload(h.d, planId);
+    expect(s.status).toBe("submitted");
+    return s.downloadTaskId!;
+  }
+  async function completeAndImport(hash: string, dtId: string) {
+    await qb.complete(hash, { writeFiles: true });
+    const sync = await syncDownloads(h.d);
+    expect(sync.completed).toContain(dtId);
+    const p = await planImport(h.d, { downloadTaskId: dtId, trigger: "auto", actorType: "system" });
+    expect(p.status, JSON.stringify(p)).toBe("running");
+    const ex = await executeRecord(h.d, p.recordId!);
+    expect(ex.status).toBe("succeeded");
+  }
+
+  it("Slot×单元多在途:两集同时在途;重叠单元的发布延后;下载被移除即释放预留,重评后补上", async () => {
+    const { entityId, slotId } = await seriesSlot("熊家餐馆");
+    const e1 = await release(entityId, "The.Bear.S01E01.1080p.WEB-DL.x264-GRP");
+    const e2 = await release(entityId, "The.Bear.S01E02.1080p.WEB-DL.x264-GRP");
+    const p1 = await planFor(e1.id);
+    const p2 = await planFor(e2.id); // Slot 已 reserved,单元不相交 → 第二个在途计划
+    expect((await planRow(p1)).unit_keys).toEqual(["s1e1"]);
+    expect((await planRow(p2)).unit_keys).toEqual(["s1e2"]);
+    expect((await slotRow(slotId)).status).toBe("reserved");
+
+    // 同一集的另一个版本:单元在途 → 延后(deferred),不丢
+    const e2b = await release(entityId, "The.Bear.S01E02.1080p.WEB-DL.x265-OTHER");
+    const ev = await evaluateAcquisition(h.d, { sourceReleaseId: e2b.id });
+    expect(ev.plans).toHaveLength(0);
+    expect(ev.deferred).toEqual([{ sourceReleaseId: e2b.id, slotId, reason: "units_inflight" }]);
+    expect(
+      await db().selectFrom("slot_deferral").select(["unit_keys"]).where("slot_id", "=", slotId).execute(),
+    ).toEqual([{ unit_keys: ["s1e2"] }]);
+
+    const dt1 = await submit(p1);
+    const dt2 = await submit(p2);
+    expect((await slotRow(slotId)).status).toBe("downloading");
+    await completeAndImport(e1.hash, dt1);
+    // E01 入库,但 E02 仍在途:Slot 显示在途阶段而不是静止态;分集账本登记 E01
+    const s1 = await slotRow(slotId);
+    expect(s1.status).toBe("downloading");
+    expect(s1.current_asset_id).not.toBeNull();
+    expect(s1.next_evaluate_at).not.toBeNull();
+    const ledger = await db()
+      .selectFrom("slot_unit as su")
+      .innerJoin("content_unit as u", "u.id", "su.content_unit_id")
+      .select(["u.episode_number", "su.plan_id"])
+      .where("su.slot_id", "=", slotId)
+      .execute();
+    expect(ledger).toEqual([{ episode_number: 1, plan_id: p1 }]);
+
+    // 用户在 qB 里删掉 E02:确认移除 → 计划失败释放单元 → Slot 到期重评 → 延后的 E02 另一版本得到计划
+    const qc = createQbClient({ instanceId: "t", baseUrl: qb.url, username: "admin", password: "adminadmin" });
+    await qc.delete([e2.hash], { deleteFiles: false });
+    await syncDownloads(h.d);
+    expect(await planRow(p2)).toMatchObject({ status: "failed", status_reason: "download_removed" });
+    expect((await slotRow(slotId)).status).toBe("attention");
+    h.engine.enqueued.length = 0;
+    expect(await sweepDueSlots(h.d)).toBeGreaterThanOrEqual(1);
+    expect(h.engine.enqueued).toContainEqual(
+      expect.objectContaining({ task: "evaluateAcquisition", input: { acquisitionSlotId: slotId } }),
+    );
+    const re = await evaluateAcquisition(h.d, { acquisitionSlotId: slotId });
+    expect(re.plans).toHaveLength(1);
+    expect(await planRow(re.plans[0]!)).toMatchObject({ source_release_id: e2b.id, unit_keys: ["s1e2"] });
+    // e2b 的延后已消化;被移除的原 E02 此时单元在途(e2b)→ 转为延后,等 e2b 结束后再判
+    const left = await db()
+      .selectFrom("slot_deferral")
+      .select("source_release_id")
+      .where("slot_id", "=", slotId)
+      .execute();
+    expect(left.map((x) => x.source_release_id)).toEqual([e2.id]);
+    // 候选评估记录带 trace / unitKeys
+    const art = await db()
+      .selectFrom("artifact")
+      .select("content")
+      .where("subject_id", "=", slotId)
+      .where("title", "=", "candidate_evaluation")
+      .execute();
+    const byRel = new Map(art.map((a) => [(a.content as { sourceReleaseId: string }).sourceReleaseId, a.content]));
+    expect(byRel.get(e2b.id)).toMatchObject({ decision: "accept", unitKeys: ["s1e2"], pri: 100 });
+  }, 60_000);
+
+  it("季包:总集数已知 → 只预留缺集;提交后读文件清单,已覆盖/在途的集文件优先级置 0", async () => {
+    const { entityId, slotId } = await seriesSlot("选集剧");
+    await db()
+      .transaction()
+      .execute((trx) => applyEpisodeTotal(trx, { entityId, season: 1, fresh: 4, now: new Date() }));
+    const e1 = await release(entityId, "Pick.Show.S01E01.1080p.WEB-DL.x264-GRP");
+    await completeAndImport(e1.hash, await submit(await planFor(e1.id)));
+    const e2 = await release(entityId, "Pick.Show.S01E02.1080p.WEB-DL.x264-GRP");
+    const p2 = await planFor(e2.id); // E02 在途(未提交)
+
+    const packTitle = "Pick.Show.S01.1080p.WEB-DL.x264-GRP";
+    const files = [1, 2, 3, 4].map((e) => ({ path: `Pick.Show.S01E0${e}.1080p.WEB-DL.x264-GRP.mkv`, length: SIZE }));
+    const pack = await release(entityId, packTitle, [
+      ...files,
+      { path: "Pick.Show.S01E01.chs.srt", length: 100 },
+      { path: "Pick.Show.S01E03.chs.srt", length: 100 },
+    ]);
+    const pp = await planFor(pack.id);
+    expect((await planRow(pp)).unit_keys).toEqual(["s1e3", "s1e4"]);
+    const before = qb.callsTo("torrents/filePrio").length;
+    await submit(pp);
+    const prio = qb.callsTo("torrents/filePrio").slice(before);
+    expect(prio).toHaveLength(1);
+    const t = qb.get(pack.hash)!;
+    const off = t.files.filter((f) => f.priority === 0).map((f) => f.name.split("/").pop());
+    expect(off.sort()).toEqual(
+      [
+        "Pick.Show.S01E01.1080p.WEB-DL.x264-GRP.mkv",
+        "Pick.Show.S01E01.chs.srt",
+        "Pick.Show.S01E02.1080p.WEB-DL.x264-GRP.mkv",
+      ].sort(),
+    );
+    expect((await planRow(pp)).file_selection).toMatchObject({ mode: "selected", selectedUnits: ["s1e3", "s1e4"] });
+    // 进度:E01 已覆盖,E02–E04 在途
+    const prog = await subscriptionProgress(db(), (await slotRow(slotId)).subscription_id!);
+    expect(prog!.seasons[0]).toMatchObject({ seasonNumber: 1, episodeTotal: 4, totalKnown: true, missing: [] });
+    expect(prog!.seasons[0]!.units.map((u) => u.state)).toEqual(["covered", "inflight", "inflight", "inflight"]);
+    expect(prog!.slots[0]!.inflightPlans.map((x) => x.planId).sort()).toEqual([p2, pp].sort());
+
+    // 认不出集号的整包:整包下载并记理由
+    const { entityId: e2Id } = await seriesSlot("整包剧");
+    const odd = await release(e2Id, "Odd.Show.S01.1080p.WEB-DL.x264-GRP", [
+      { path: "disc1.mkv", length: SIZE },
+      { path: "disc2.mkv", length: SIZE },
+    ]);
+    const po = await planFor(odd.id);
+    expect((await planRow(po)).unit_keys).toEqual(["s1*"]);
+    const n0 = qb.callsTo("torrents/filePrio").length;
+    await submit(po);
+    expect(qb.callsTo("torrents/filePrio").length).toBe(n0);
+    expect((await planRow(po)).file_selection).toMatchObject({ mode: "whole", reason: "unmapped_video_file" });
+  }, 60_000);
+
+  it("总集数未知:入库后不结束、记疑似缺集;总集数就绪后补齐 → once 订阅完成并写历史", async () => {
+    const { entityId, slotId, subId } = await seriesSlot("未知集数剧", { mode: "once" });
+    for (const ep of [1, 3]) {
+      const r = await release(entityId, `Unknown.Total.S01E0${ep}.1080p.WEB-DL.x264-GRP`);
+      await completeAndImport(r.hash, await submit(await planFor(r.id)));
+    }
+    expect((await slotRow(slotId)).status).toBe("empty");
+    expect(await subRow(subId)).toMatchObject({ status: "active" });
+    const dec = await db()
+      .selectFrom("decision")
+      .selectAll()
+      .where("dedupe_key", "=", `suspected_missing:${slotId}`)
+      .executeTakeFirstOrThrow();
+    expect(dec.evidence).toMatchObject({ suspectedMissing: [{ season: 1, episodes: [2] }] });
+
+    await db()
+      .transaction()
+      .execute((trx) => applyEpisodeTotal(trx, { entityId, season: 1, fresh: 3, now: new Date() }));
+    const r2 = await release(entityId, "Unknown.Total.S01E02.1080p.WEB-DL.x264-GRP");
+    await completeAndImport(r2.hash, await submit(await planFor(r2.id)));
+    expect((await slotRow(slotId)).status).toBe("satisfied");
+    expect(await subRow(subId)).toMatchObject({ status: "completed", status_reason: "slot_satisfied" });
+    const hist = await db()
+      .selectFrom("subscription_history")
+      .selectAll()
+      .where("subscription_id", "=", subId)
+      .execute();
+    expect(hist).toHaveLength(1);
+    expect(hist[0]).toMatchObject({ reason: "slot_satisfied", media_entity_id: entityId, mode: "once" });
+    expect((hist[0]!.progress as { seasons: { complete: boolean }[] }).seasons[0]!.complete).toBe(true);
+    // 历史 API + 重新订阅(改回 active)
+    const api = await h.api("GET", `/subscription-history?subscriptionId=${subId}`);
+    expect(api.body.data.items).toHaveLength(1);
+    const re = await h.api("PATCH", `/subscriptions/${subId}`, { status: "active" });
+    expect(re.body.data.status).toBe("active");
+  }, 90_000);
+
+  it("pending(TMDB 未就绪)照常下载但不完成;暂停的订阅不评估", async () => {
+    const { entityId, slotId, subId } = await seriesSlot("待定剧", { mode: "once", status: "pending" });
+    await db()
+      .transaction()
+      .execute((trx) => applyEpisodeTotal(trx, { entityId, season: 1, fresh: 1, now: new Date() }));
+    const r = await release(entityId, "Pending.Show.S01E01.1080p.WEB-DL.x264-GRP");
+    await completeAndImport(r.hash, await submit(await planFor(r.id)));
+    expect((await slotRow(slotId)).status).toBe("empty");
+    expect((await subRow(subId)).status).toBe("pending");
+
+    const paused = await seriesSlot("暂停剧", { status: "active" });
+    await db().updateTable("subscription").set({ status: "paused" }).where("id", "=", paused.subId).execute();
+    const pr = await release(paused.entityId, "Paused.Show.S01E01.1080p.WEB-DL.x264-GRP");
+    const ev = await evaluateAcquisition(h.d, { sourceReleaseId: pr.id });
+    expect(ev).toMatchObject({ evaluated: 0, plans: [] });
+  }, 60_000);
+
+  it("分集洗版:只替换能提升的单元;同档不换(trace 可读)", async () => {
+    const { entityId, slotId } = await seriesSlot("洗版剧", {
+      hard: { resolutions: ["720p", "1080p"] },
+      upgrade: { enabled: true, allowResolutionUpgrade: true },
+    });
+    const r = await release(entityId, "Wash.Show.S01E01.720p.WEB-DL.x264-GRP");
+    await completeAndImport(r.hash, await submit(await planFor(r.id)));
+    const same = await release(entityId, "Wash.Show.S01E01.720p.WEB-DL.x264-OTHER");
+    const ev = await evaluateAcquisition(h.d, { sourceReleaseId: same.id });
+    expect(ev.plans).toHaveLength(0);
+    const up = await release(entityId, "Wash.Show.S01E01-E02.1080p.WEB-DL.x264-GRP");
+    // E02 还缺 → 追更(fill);只有 E01 覆盖时的纯升级场景见下一条
+    const fill = await planFor(up.id);
+    expect(await planRow(fill)).toMatchObject({ purpose: "fill", unit_keys: ["s1e2"] });
+    await db().updateTable("acquisition_plan").set({ status: "cancelled" }).where("id", "=", fill).execute();
+    await db().updateTable("acquisition_slot").set({ status: "empty" }).where("id", "=", slotId).execute();
+    const up1 = await release(entityId, "Wash.Show.S01E01.1080p.WEB-DL.x264-GRP");
+    const pu = await planFor(up1.id);
+    expect(await planRow(pu)).toMatchObject({ purpose: "upgrade", unit_keys: ["s1e1"] });
+    const art = await db()
+      .selectFrom("artifact")
+      .select("content")
+      .where("subject_id", "=", slotId)
+      .where("title", "=", "candidate_evaluation")
+      .execute();
+    const sameEval = art
+      .map((a) => a.content as { sourceReleaseId: string; trace?: string[] })
+      .find((a) => a.sourceReleaseId === same.id);
+    expect(sameEval?.trace?.[0]).toBe("与当前版本同档,不反复更换");
+  }, 60_000);
+});

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

@@ -12,6 +12,7 @@ import { registerOrganizationRoutes } from "./routes/organization.ts";
 import { registerRuleRoutes } from "./routes/rules.ts";
 import { type AoDeps, automationBlock, enqueueSweep, enqueueTracked, timeSlot } from "./services/common.ts";
 import { SWEEP_SLOT_MS, sweepReservedPlans, syncDownloads } from "./services/downloads.ts";
+import { sweepDueSlots, sweepMetadataRefresh, sweepSubscriptionSearch } from "./services/subscriptions.ts";
 import { loadVolumes, usedPercent } from "./services/volumes.ts";
 import { errMessage } from "./util.ts";
 
@@ -55,6 +56,10 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se
       if (over) await enqueueThresholdCleanup(d);
       await sweepReservedPlans(d);
       await sweepPendingImports(d);
+      // W5-C:Slot 到期重评(忙时延后的发布 / 定时)、订阅周期主动搜索、追踪剧集的总集数刷新
+      await sweepDueSlots(d);
+      await sweepSubscriptionSearch(d);
+      await sweepMetadataRefresh(d);
     }
     const every = d.ports.options.reconcileIntervalMs;
     if (every > 0) {

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

@@ -41,6 +41,10 @@ export interface AoOptions {
   missingConfirmAfter: number;
   /** 后台资产对账周期(0 = 不自动跑) */
   reconcileIntervalMs: number;
+  /** 未满足 Slot 的定时重评周期(next_evaluate_at),缺省 6h;0 = 只在计划结束时重评 */
+  slotReevaluateIntervalMs?: number;
+  /** 追踪中剧集的元数据(总集数)刷新周期,缺省 6h;0 = 关 */
+  metadataRefreshIntervalMs?: number;
 }
 
 export interface AoPorts {

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

@@ -6,6 +6,7 @@ import {
   type PreferenceSuppression,
   type PreferenceTag,
   type Subscription,
+  type SubscriptionHistory,
 } from "@umt/contracts";
 import { type DB, sql, type Transaction } from "@umt/db";
 import { evaluateCandidate, parseRelease } from "@umt/domain";
@@ -16,6 +17,7 @@ import { ApiException } from "../../../http/errors.ts";
 import type { ModuleApp } from "../../types.ts";
 import { createSlotsFromTemplates, publishedAcquisitionPolicies } from "../services/acquisition.ts";
 import type { AoDeps } from "../services/common.ts";
+import { applyEpisodeTotal, enqueueSubscriptionSearch, subscriptionProgress } from "../services/subscriptions.ts";
 import { actorOf, conflict, errMessage, hashJson, jsonb, notFound, toApi, toApiList } from "../util.ts";
 import { mount, paginate } from "./helpers.ts";
 
@@ -137,12 +139,25 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) {
           .executeTakeFirst();
         if (!e) throw new ApiException("bad_request", "mediaEntityId 不存在");
       }
+      // pending:TMDB 未就绪(未建档,或剧集没有已接受的 TMDB 身份 → 总集数无来源):照常搜索/评估,不能完成
+      const tmdbReady =
+        !!body.mediaEntityId &&
+        (body.mediaKind === "movie" ||
+          !!(await db()
+            .selectFrom("external_identity")
+            .select("id")
+            .where("media_entity_id", "=", body.mediaEntityId)
+            .where("provider", "=", "tmdb")
+            .where("status", "=", "accepted")
+            .executeTakeFirst()));
       const row = await db()
         .transaction()
         .execute(async (trx) => {
           const r = await trx
             .insertInto("subscription")
             .values({
+              status: tmdbReady ? "active" : "pending",
+              status_reason: tmdbReady ? null : "metadata_pending",
               media_entity_id: body.mediaEntityId ?? null,
               season_unit_id: body.seasonUnitId ?? null,
               title_query: body.titleQuery ?? null,
@@ -181,7 +196,21 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) {
           }
           return r;
         });
-      if (row.media_entity_id) await kickEvaluate(d, { subscriptionId: row.id });
+      if (row.media_entity_id) {
+        await kickEvaluate(d, { subscriptionId: row.id });
+        // 新订阅:立即主动搜索一次(之后按周期);剧集先刷新元数据拿季总集数
+        await enqueueSubscriptionSearch(d, row.id, "created").catch((err) =>
+          d.ctx.logger.debug({ err: { message: errMessage(err) } }, "订阅搜索入队失败"),
+        );
+        if (row.media_kind !== "movie" && tmdbReady)
+          await d.ctx.engine
+            .enqueue(
+              "refreshMetadata",
+              { mediaEntityId: row.media_entity_id },
+              { idempotencyKey: `refreshMetadata:subscription:${row.id}`, traceId: row.id },
+            )
+            .catch((err) => d.ctx.logger.debug({ err: { message: errMessage(err) } }, "元数据刷新入队失败"));
+      }
       return toApi<Subscription>(row);
     },
     { tag: TAG },
@@ -210,6 +239,94 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) {
     { tag: TAG },
   );
 
+  mount(
+    app,
+    Api.subscriptions.search,
+    async (_c, { params }) => {
+      const sub = await db()
+        .selectFrom("subscription")
+        .select(["id", "media_entity_id", "status"])
+        .where("id", "=", params.id)
+        .executeTakeFirst();
+      if (!sub) notFound("订阅");
+      if (!sub!.media_entity_id) throw new ApiException("bad_request", "订阅尚未关联作品,无法按标题搜索");
+      try {
+        return await enqueueSubscriptionSearch(d, sub!.id, "manual");
+      } catch (err) {
+        throw new ApiException("upstream_unavailable", `搜索任务入队失败:${errMessage(err)}`);
+      }
+    },
+    { tag: TAG },
+  );
+  mount(
+    app,
+    Api.subscriptions.progress,
+    async (_c, { params }) => (await subscriptionProgress(db(), params.id)) ?? notFound("订阅"),
+    { tag: TAG },
+  );
+  mount(
+    app,
+    Api.subscriptions.setEpisodeTotal,
+    async (_c, { params, body }) => {
+      const sub = await db()
+        .selectFrom("subscription")
+        .select(["id", "media_entity_id", "media_kind"])
+        .where("id", "=", params.id)
+        .executeTakeFirst();
+      if (!sub) notFound("订阅");
+      if (!sub!.media_entity_id || sub!.media_kind === "movie")
+        throw new ApiException("bad_request", "只有已关联作品的剧集/综艺订阅有季总集数");
+      await db()
+        .transaction()
+        .execute(async (trx) => {
+          const now = d.ports.now();
+          if (body.locked) {
+            await applyEpisodeTotal(trx, {
+              entityId: sub!.media_entity_id!,
+              season: body.seasonNumber,
+              manual: { total: body.total, locked: true },
+              now,
+            });
+          } else {
+            // 未锁定:按注入口径只升
+            await applyEpisodeTotal(trx, {
+              entityId: sub!.media_entity_id!,
+              season: body.seasonNumber,
+              injected: body.total,
+              now,
+            });
+          }
+        });
+      return (await subscriptionProgress(db(), params.id))!;
+    },
+    { tag: TAG },
+  );
+  mount(
+    app,
+    Api.subscriptionHistory.list,
+    async (_c, { query }) => {
+      let q = db().selectFrom("subscription_history").selectAll();
+      if (query.mediaEntityId) q = q.where("media_entity_id", "=", query.mediaEntityId);
+      if (query.subscriptionId) q = q.where("subscription_id", "=", query.subscriptionId);
+      const p = await paginate(q, query);
+      return { items: toApiList<SubscriptionHistory>(p.rows), nextCursor: p.nextCursor };
+    },
+    { tag: TAG },
+  );
+  mount(
+    app,
+    Api.subscriptionHistory.get,
+    async (_c, { params }) => {
+      const r = await db()
+        .selectFrom("subscription_history")
+        .selectAll()
+        .where("id", "=", params.id)
+        .executeTakeFirst();
+      return toApi<SubscriptionHistory>(r ?? notFound("订阅历史"));
+    },
+    { tag: TAG },
+  );
+
   // ── 收录 Slot ──
   mount(
     app,

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

@@ -10,19 +10,36 @@ import { sql } from "@umt/db";
 import {
   type AcquisitionMode,
   checkBudget,
+  computeUnitGaps,
   determineMode,
+  type EvaluationContext,
+  episodeKey,
   evaluateCandidate,
+  improvableUnits,
   type ParsedRelease,
   parseRelease,
   priorityOf,
   type QualityFacts,
   type SlotEvaluation,
   type SlotState,
+  seasonWildcard,
   transition,
+  unitsOverlap,
 } from "@umt/domain";
 import { fromParsedColumns, slotSpecFromRow, toContractSlotEvaluation } from "@umt/mapping";
 import { hashJson, jsonb, startOfDay, toApi } from "../util.ts";
 import { type AoDeps, automationBlock, type Exec, enqueueTracked, loadInstances, recordCandidate } from "./common.ts";
+import {
+  clearDeferral,
+  rangeOf,
+  recordDeferral,
+  scheduleSlotReevaluate,
+  seasonLedgers,
+  slotCoveredUnits,
+  slotInflightPlans,
+  subParams,
+  TRACKING_SUB_STATUSES,
+} from "./subscriptions.ts";
 import { decideVolume, loadVolumes } from "./volumes.ts";
 
 // evaluateAcquisition:release.resolved → 匹配订阅/Slot → 模式 → 确定性评估 → 预算/缓存卷 → 冻结计划并预留。
@@ -45,6 +62,8 @@ export interface EvaluateOutcome {
   evaluated: number;
   plans: string[];
   skipped: { sourceReleaseId: string; reason: string }[];
+  /** Slot 忙(单元在途冲突 / 电影 Slot 在途)而延后的发布;计划结束后重评 */
+  deferred: { sourceReleaseId: string; slotId: string; reason: string }[];
 }
 
 const PRIORITY: Record<ReturnType<typeof priorityOf>, AcquisitionPriority> = {
@@ -110,25 +129,6 @@ async function releaseUnits(
   return [...ids, ...keys];
 }
 
-/** Slot 已满足的单元:该 Slot 的计划带来的资产 + 当前资产的覆盖(content_unit id 与合成键) */
-async function satisfiedUnits(exec: Exec, slot: AcquisitionSlotRow): Promise<string[]> {
-  const assets = await sql<{ id: string; media_info: unknown }>`
-    select distinct a.id, a.media_info from media_asset a
-      left join download_task dt on dt.id = a.download_task_id
-      left join acquisition_plan p on p.id = dt.acquisition_plan_id
-     where a.status = 'active' and a.parent_asset_id is null
-       and (p.slot_id = ${slot.id} or a.id = ${slot.current_asset_id})`.execute(exec);
-  if (assets.rows.length === 0) return [];
-  const ids = assets.rows.map((a) => a.id);
-  const cov = await exec
-    .selectFrom("asset_unit_coverage")
-    .select("content_unit_id")
-    .where("media_asset_id", "in", ids)
-    .execute();
-  const keys = assets.rows.flatMap((a) => (a.media_info as { _umt?: { unitKeys?: string[] } })?._umt?.unitKeys ?? []);
-  return [...cov.map((c) => c.content_unit_id), ...keys];
-}
-
 async function resolveEntity(exec: Exec, id: string): Promise<MediaEntityRow | null> {
   let cur = await exec.selectFrom("media_entity").selectAll().where("id", "=", id).executeTakeFirst();
   for (let i = 0; cur?.merged_into_id && i < 10; i++) {
@@ -229,7 +229,12 @@ async function note(c: Ctx, base: Omit<CandidateEvaluation, "evaluatedAt">) {
 }
 
 export async function evaluateAcquisition(d: AoDeps, input: EvaluateInput): Promise<EvaluateOutcome> {
-  const c: Ctx = { d, now: d.ports.now(), out: { evaluated: 0, plans: [], skipped: [] }, draft: input.draft === true };
+  const c: Ctx = {
+    d,
+    now: d.ports.now(),
+    out: { evaluated: 0, plans: [], skipped: [], deferred: [] },
+    draft: input.draft === true,
+  };
   const db = d.ctx.db;
   if (input.sourceReleaseId) {
     const rel = await db
@@ -266,6 +271,22 @@ export async function evaluateAcquisition(d: AoDeps, input: EvaluateInput): Prom
       )?.media_entity_id ?? null;
   }
   if (!entityId) return c.out;
+  // 先评该 Slot 延后的发布(按集号升序,连续更新的集依次补上),再评最近已识别的发布
+  const deferred = input.acquisitionSlotId
+    ? await db
+        .selectFrom("slot_deferral as sd")
+        .innerJoin("source_release as r", "r.id", "sd.source_release_id")
+        .selectAll("r")
+        .where("sd.slot_id", "=", input.acquisitionSlotId)
+        .orderBy("sd.created_at")
+        .execute()
+    : [];
+  const epStart = (r: SourceReleaseRow) => {
+    const p = parsedOf(r);
+    return (p.season ?? 1) * 100_000 + (p.episodes?.start ?? 0);
+  };
+  deferred.sort((a, b) => epStart(a) - epStart(b));
+  const seen = new Set(deferred.map((r) => r.id));
   const rels = await db
     .selectFrom("source_release")
     .selectAll()
@@ -275,7 +296,8 @@ export async function evaluateAcquisition(d: AoDeps, input: EvaluateInput): Prom
     .orderBy("first_seen_at", "desc")
     .limit(50)
     .execute();
-  for (const r of rels) await evaluateRelease(c, r, [], input.acquisitionSlotId);
+  for (const r of [...deferred, ...rels.filter((x) => !seen.has(x.id))])
+    await evaluateRelease(c, r, [], input.acquisitionSlotId);
   return c.out;
 }
 
@@ -372,7 +394,11 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon
   // 订阅停止条件
   for (const s of subs) {
     const stopAfter = (s.params as { stopAfter?: string } | null)?.stopAfter;
-    if (s.status === "active" && stopAfter && Date.parse(stopAfter) < now.getTime()) {
+    if (
+      (TRACKING_SUB_STATUSES as readonly string[]).includes(s.status) &&
+      stopAfter &&
+      Date.parse(stopAfter) < now.getTime()
+    ) {
       await db
         .updateTable("subscription")
         .set({ status: "completed", status_reason: "stop_after_reached" })
@@ -381,7 +407,10 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon
       s.status = "completed";
     }
   }
-  const activeSubs = new Map<string, SubscriptionRow>(subs.filter((s) => s.status === "active").map((s) => [s.id, s]));
+  // pending(TMDB 未就绪)照常评估与下载,只是不能完成
+  const activeSubs = new Map<string, SubscriptionRow>(
+    subs.filter((s) => (TRACKING_SUB_STATUSES as readonly string[]).includes(s.status)).map((s) => [s.id, s]),
+  );
   const tracked = activeSubs.size > 0 || slots.some((s) => s.subscription_id === null);
   const releaseSeason = parsed.season ?? (parsed.episodes ? 1 : undefined);
 
@@ -416,97 +445,293 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon
       if (su?.season_number !== null && su?.season_number !== undefined && su.season_number !== releaseSeason) continue;
     }
     c.out.evaluated++;
-    const base = { slotId: slot.id, sourceReleaseId: rel.id, title: rel.title, planId: null as string | null };
+    const outcome = await evaluateForSlot(c, {
-    if (BUSY.includes(slot.status)) {
-      await note(c, { ...base, decision: "skipped", mode: null, rejected: [], reasons: [`slot_busy:${slot.status}`] });
-      continue;
-    }
-    const spec = slotSpecFromRow(toApi<AcquisitionSlot>(slot));
-    const satisfied = await satisfiedUnits(db, slot);
-    const slotSatisfied = slot.status === "satisfied" || slot.status === "upgrade_pending";
-    const gaps = units.filter((u) => !satisfied.includes(u));
-    const current = await currentFacts(db, slot.current_asset_id);
-    const fill = isMovie ? !slotSatisfied : gaps.length > 0 || !current;
-    const ev: SlotEvaluation = evaluateCandidate(spec, parsed, fill ? null : current);
-    const flat = toContractSlotEvaluation(ev);
-    if (!flat.acceptable) {
-      await note(c, {
-        ...base,
-        decision: ev.decision as "reject" | "unknown_field",
-        mode: null,
-        rejected: flat.rejected,
-        reasons: flat.reasons,
-      });
-      continue;
-    }
-    const mode: AcquisitionMode | null =
-      forcedMode ??
-      determineMode({
-        entity: { tracked, kind: isMovie ? "movie" : "tv" },
-        releaseUnits: units,
-        satisfiedUnits: satisfied,
-        slotSatisfied,
-        now: now.toISOString(),
-        newTitleWindowDays: d.ports.options.newTitleWindowDays,
-        upgradeCandidate: ev.decision === "upgrade",
-      }).mode;
-    if (!mode) {
-      await note(c, {
-        ...base,
-        decision: "skipped",
-        mode: null,
-        rejected: [],
-        reasons: ["mode:none", ...flat.reasons],
-      });
-      continue;
-    }
-    if (planned) {
-      await note(c, { ...base, decision: "skipped", mode, rejected: [], reasons: [`covered_by_plan:${planned}`] });
-      continue;
-    }
-    const gate = await gates(c, rel, mode, sub);
-    if (!gate.ok) {
-      await note(c, { ...base, decision: "skipped", mode, rejected: gate.rejected, reasons: gate.reasons });
-      skip(gate.reasons[0] ?? "gated");
-      continue;
-    }
-    const planId = await reserve(c, {
-      slot,
       rel,
-      mode,
+      entity,
-      ev,
+      parsed,
+      units,
+      slot,
       sub,
-      instanceId: gate.instanceId,
+      tracked,
-      volumeKey: gate.volumeKey,
+      forcedMode,
-      savePath: gate.savePath,
+      releaseSeason: releaseSeason ?? 1,
+      slotCount: slots.length,
+      alreadyPlanned: planned,
     });
-    if (!planId) {
+    if (outcome.planId) planned = outcome.planId;
-      await note(c, { ...base, decision: "skipped", mode, rejected: [], reasons: ["slot_changed_or_reserved"] });
+    if (!outcome.deferred) await clearDeferral(db, slot.id, rel.id);
-      continue;
+    if (!isMovie && sub) await scheduleSlotReevaluate(d, db, slot.id);
-    }
+  }
-    planned = planId;
+}
-    c.out.plans.push(planId);
+
+/** 站点字段 → 评估上下文(做种数、促销下载系数、副标题/标签、站点 id) */
+function contextOf(rel: SourceReleaseRow): EvaluationContext {
+  const promo = (rel.promotion ?? {}) as { kind?: string; raw?: { downloadFactor?: number | null } };
+  const factor =
+    typeof promo.raw?.downloadFactor === "number"
+      ? promo.raw.downloadFactor
+      : promo.kind === "free" || promo.kind === "free2up"
+        ? 0
+        : promo.kind === "50pctdown" || promo.kind === "50pctdown2up"
+          ? 0.5
+          : promo.kind === "30pctdown"
+            ? 0.3
+            : promo.kind === "none" || promo.kind === "2up"
+              ? 1
+              : undefined;
+  return {
+    siteId: rel.site_id,
+    ...(rel.subtitle ? { subtitle: rel.subtitle } : {}),
+    labels: rel.tags_raw ?? [],
+    ...(rel.seeders !== null && rel.seeders !== undefined ? { seeders: rel.seeders } : {}),
+    ...(factor !== undefined ? { downloadVolumeFactor: factor } : {}),
+  };
+}
+
+interface SlotInput {
+  rel: SourceReleaseRow;
+  entity: MediaEntityRow;
+  parsed: ParsedRelease;
+  units: string[];
+  slot: AcquisitionSlotRow;
+  sub: SubscriptionRow | undefined;
+  tracked: boolean;
+  forcedMode: AcquisitionMode | null;
+  releaseSeason: number;
+  slotCount: number;
+  alreadyPlanned: string | null;
+}
+
+/**
+ * 单个 Slot 的评估(对照 MP 缺集 = 媒体库已有 ∩ 订阅范围 − note):
+ * - 电影 Slot 有在途计划 → 延后(deferred);
+ * - 剧集:需要的单元 = 候选单元 − 已覆盖 − 在途;全部在途 → 延后;都已覆盖 → 分集洗版(只取能提升的单元);
+ * - 季包:总集数已知按缺集展开,未知用整季通配预留(提交后开文件清单再按缺集选文件)。
+ */
+async function evaluateForSlot(c: Ctx, p: SlotInput): Promise<{ planId: string | null; deferred: boolean }> {
+  const { d, now } = c;
+  const db = d.ctx.db;
+  const { rel, parsed, slot, sub } = p;
+  const isMovie = p.entity.kind === "movie";
+  const base = { slotId: slot.id, sourceReleaseId: rel.id, title: rel.title, planId: null as string | null };
+  const done = { planId: null, deferred: false };
+  const params = subParams(sub);
+  if (params.siteIds?.length && !params.siteIds.includes(rel.site_id)) {
+    await note(c, { ...base, decision: "skipped", mode: null, rejected: [], reasons: ["site_out_of_scope"] });
+    return done;
+  }
+  const spec = slotSpecFromRow(toApi<AcquisitionSlot>(slot));
+  const ctx = contextOf(rel);
+  const inflight = await slotInflightPlans(db, slot.id);
+  const inflightKeys = inflight.flatMap((x) => x.unitKeys);
+  const defer = async (reason: string, unitKeys: string[], mode: AcquisitionMode | null = null) => {
+    await recordDeferral(db, { slotId: slot.id, sourceReleaseId: rel.id, reason, unitKeys });
     await note(c, {
       ...base,
-      planId,
+      decision: "deferred",
-      decision: ev.decision as "accept" | "upgrade",
       mode,
       rejected: [],
-      reasons: flat.reasons,
+      reasons: [`deferred:${reason}`, ...inflight.map((x) => `inflight:${x.id}`)],
+      unitKeys,
     });
-    if (c.draft) continue;
+    c.out.deferred.push({ sourceReleaseId: rel.id, slotId: slot.id, reason });
-    await enqueueTracked(d, {
+    return { planId: null, deferred: true };
-      kind: "submitDownload",
+  };
-      idempotencyKey: `plan:${planId}`,
+
-      input: { acquisitionPlanId: planId },
+  let ev: SlotEvaluation;
-      subjectType: "acquisition_plan",
+  let planUnits: string[] = [];
-      subjectId: planId,
+  let releaseKeys: string[] = [];
-    }).then(
+  let covered = new Set<string>();
-      ({ taskId }) => db.updateTable("acquisition_plan").set({ task_id: taskId }).where("id", "=", planId).execute(),
+  const slotSatisfied = slot.status === "satisfied" || slot.status === "upgrade_pending";
-      (err) =>
+  if (isMovie) {
-        d.ctx.logger.warn({ err: { message: (err as Error).message }, planId }, "submitDownload 入队失败,留给巡检补投"),
+    if (inflight.length > 0 || BUSY.includes(slot.status)) return defer("slot_busy", []);
-    );
+    const current = await currentFacts(db, slot.current_asset_id);
+    ev = evaluateCandidate(spec, parsed, slotSatisfied ? current : null, ctx);
+  } else {
+    const cov = await slotCoveredUnits(db, slot, { entityWide: p.slotCount <= 1 });
+    covered = cov.keys;
+    const range = rangeOf(sub);
+    const isPack = !parsed.episodes && !parsed.airDate;
+    let wanted: string[];
+    let reservedHit: string[];
+    if (isPack) {
+      const ledger = (await seasonLedgers(db, p.entity.id)).get(p.releaseSeason);
+      if (ledger?.known && ledger.total) {
+        releaseKeys = Array.from({ length: ledger.total }, (_, i) => episodeKey(p.releaseSeason, i + 1));
+        const g = computeUnitGaps({
+          units: releaseKeys,
+          covered: [...covered],
+          inflight: inflightKeys,
+          ...(range ? { range } : {}),
+        });
+        wanted = g.missing;
+        reservedHit = g.reserved;
+        ctx.unitCount = ledger.total;
+      } else {
+        // 总集数未知:整季通配预留;与同季在途单元冲突 → 延后
+        releaseKeys = [seasonWildcard(p.releaseSeason)];
+        reservedHit = unitsOverlap(releaseKeys, inflightKeys);
+        wanted = reservedHit.length ? [] : releaseKeys;
+      }
+    } else {
+      releaseKeys = unitKeysOf(parsed);
+      const g = computeUnitGaps({
+        units: releaseKeys,
+        covered: [...covered],
+        inflight: inflightKeys,
+        ...(range ? { range } : {}),
+      });
+      if (releaseKeys.length > 0 && g.outOfRange.length === releaseKeys.length) {
+        await note(c, { ...base, decision: "skipped", mode: null, rejected: [], reasons: ["episode_out_of_range"] });
+        return done;
+      }
+      wanted = g.missing;
+      reservedHit = g.reserved;
+    }
+    if (wanted.length > 0) {
+      ev = evaluateCandidate(spec, parsed, null, ctx);
+      planUnits = wanted;
+    } else if (reservedHit.length > 0) {
+      return defer("units_inflight", reservedHit);
+    } else {
+      // 都已覆盖:分集洗版,只取能提升的单元(对照 MP get_best_version_interested_episodes)
+      const unitCurrent = await unitQuality(db, slot, releaseKeys);
+      const withCurrent = releaseKeys.filter((k) => unitCurrent[k]);
+      const firstCurrent = withCurrent.length
+        ? unitCurrent[withCurrent[0]!]!
+        : await currentFacts(db, slot.current_asset_id);
+      if (!spec.upgrade?.enabled || withCurrent.length === 0) {
+        ev = evaluateCandidate(spec, parsed, firstCurrent ?? {}, ctx);
+      } else {
+        const imp = improvableUnits(spec, parsed, withCurrent, unitCurrent, ctx);
+        if (imp.units.length === 0) {
+          ev = evaluateCandidate(spec, parsed, firstCurrent, ctx);
+        } else {
+          const busyHit = unitsOverlap(imp.units, inflightKeys);
+          if (busyHit.length) return defer("units_inflight", busyHit, "upgrade");
+          ev = evaluateCandidate(spec, parsed, unitCurrent[imp.units[0]!] ?? null, ctx);
+          planUnits = imp.units;
+        }
+      }
+    }
   }
+  const flat = toContractSlotEvaluation(ev);
+  const evalExtra = { pri: flat.pri, trace: flat.trace };
+  if (!flat.acceptable) {
+    await note(c, {
+      ...base,
+      decision: ev.decision as "reject" | "unknown_field",
+      mode: null,
+      rejected: flat.rejected,
+      reasons: flat.reasons,
+      ...evalExtra,
+    });
+    return done;
+  }
+  const mode: AcquisitionMode | null =
+    p.forcedMode ??
+    determineMode({
+      entity: { tracked: p.tracked, kind: isMovie ? "movie" : "tv" },
+      releaseUnits: isMovie ? p.units : releaseKeys,
+      satisfiedUnits: isMovie ? [] : releaseKeys.filter((k) => !planUnits.includes(k) || covered.has(k)),
+      slotSatisfied,
+      now: now.toISOString(),
+      newTitleWindowDays: d.ports.options.newTitleWindowDays,
+      upgradeCandidate: ev.decision === "upgrade",
+    }).mode;
+  if (!mode) {
+    await note(c, {
+      ...base,
+      decision: "skipped",
+      mode: null,
+      rejected: [],
+      reasons: ["mode:none", ...flat.reasons],
+      ...evalExtra,
+    });
+    return done;
+  }
+  if (p.alreadyPlanned) {
+    await note(c, {
+      ...base,
+      decision: "skipped",
+      mode,
+      rejected: [],
+      reasons: [`covered_by_plan:${p.alreadyPlanned}`],
+    });
+    return done;
+  }
+  const gate = await gates(c, rel, mode, sub);
+  if (!gate.ok) {
+    await note(c, { ...base, decision: "skipped", mode, rejected: gate.rejected, reasons: gate.reasons, ...evalExtra });
+    c.out.skipped.push({ sourceReleaseId: rel.id, reason: gate.reasons[0] ?? "gated" });
+    return done;
+  }
+  const r = await reserve(c, {
+    slot,
+    rel,
+    mode,
+    ev,
+    sub,
+    instanceId: gate.instanceId,
+    volumeKey: gate.volumeKey,
+    savePath: gate.savePath,
+    unitKeys: planUnits,
+    multiInflight: !isMovie,
+  });
+  if (r.conflict) return defer("units_inflight", r.conflict, mode);
+  const planId = r.planId;
+  if (!planId) {
+    await note(c, { ...base, decision: "skipped", mode, rejected: [], reasons: ["slot_changed_or_reserved"] });
+    return done;
+  }
+  c.out.plans.push(planId);
+  await note(c, {
+    ...base,
+    planId,
+    decision: ev.decision as "accept" | "upgrade",
+    mode,
+    rejected: [],
+    reasons: flat.reasons,
+    ...evalExtra,
+    unitKeys: planUnits,
+  });
+  if (c.draft) return { planId, deferred: false };
+  await enqueueTracked(d, {
+    kind: "submitDownload",
+    idempotencyKey: `plan:${planId}`,
+    input: { acquisitionPlanId: planId },
+    subjectType: "acquisition_plan",
+    subjectId: planId,
+  }).then(
+    ({ taskId }) => db.updateTable("acquisition_plan").set({ task_id: taskId }).where("id", "=", planId).execute(),
+    (err) =>
+      d.ctx.logger.warn({ err: { message: (err as Error).message }, planId }, "submitDownload 入队失败,留给巡检补投"),
+  );
+  return { planId, deferred: false };
+}
+
+/** 分集质量账本:合成单元键 → 该单元当前版本的交付特征(含入库时的阶梯 pri) */
+async function unitQuality(
+  exec: Exec,
+  slot: AcquisitionSlotRow,
+  keys: readonly string[],
+): Promise<Record<string, QualityFacts | undefined>> {
+  const rows = await exec
+    .selectFrom("slot_unit as su")
+    .innerJoin("content_unit as u", "u.id", "su.content_unit_id")
+    .select(["u.season_number", "u.episode_number", "su.quality", "su.pri"])
+    .where("su.slot_id", "=", slot.id)
+    .where("su.media_asset_id", "is not", null)
+    .execute();
+  const out: Record<string, QualityFacts | undefined> = {};
+  const want = new Set(keys);
+  for (const r of rows) {
+    if (r.episode_number === null) continue;
+    const k = episodeKey(r.season_number ?? 1, r.episode_number);
+    if (!want.has(k)) continue;
+    out[k] = { ...((r.quality ?? {}) as QualityFacts), ...(r.pri !== null ? { pri: r.pri } : {}) };
+  }
+  return out;
 }
 
 interface GateOk {
@@ -588,8 +813,12 @@ async function reserve(
     instanceId: string;
     volumeKey: string;
     savePath: string;
+    /** 计划预留的单元(剧集);电影为空 */
+    unitKeys: string[];
+    /** 剧集 Slot:允许与其它在途计划并存(单元不相交) */
+    multiInflight: boolean;
   },
-): Promise<string | null> {
+): Promise<{ planId: string | null; conflict?: string[] }> {
   const { d, now } = c;
   const immediate = p.sub?.priority === "immediate";
   const priority = PRIORITY[priorityOf(p.mode, immediate)];
@@ -599,6 +828,7 @@ async function reserve(
     instanceId: p.instanceId,
     savePath: p.savePath,
     mode: p.mode,
+    unitKeys: p.unitKeys,
   };
   const evaluation = jsonb({
     ...toContractSlotEvaluation(p.ev),
@@ -607,6 +837,7 @@ async function reserve(
     volumeKey: p.volumeKey,
     savePath: p.savePath,
   });
+  const purpose = p.mode === "upgrade" ? ("upgrade" as const) : ("fill" as const);
   if (c.draft) {
     // 草案:同一 Slot+发布复用未提交的草案;Slot 状态与预算不动,expected_revision 记下当前 Slot 版本 +1
     const key = `draft:${p.slot.id}:${p.rel.id}`;
@@ -616,20 +847,26 @@ async function reserve(
       .where("idempotency_key", "=", key)
       .executeTakeFirst();
     if (hit) {
-      if (hit.status !== "draft") return null;
+      if (hit.status !== "draft") return { planId: null };
       await d.ctx.db
         .updateTable("acquisition_plan")
-        .set({ evaluation, payload_hash: hashJson(payload), expected_revision: p.slot.revision + 1, priority })
+        .set({
+          evaluation,
+          payload_hash: hashJson(payload),
+          expected_revision: p.slot.revision + 1,
+          priority,
+          unit_keys: p.unitKeys,
+        })
         .where("id", "=", hit.id)
         .execute();
-      return hit.id;
+      return { planId: hit.id };
     }
     const row = await d.ctx.db
       .insertInto("acquisition_plan")
       .values({
         slot_id: p.slot.id,
         source_release_id: p.rel.id,
-        purpose: p.mode === "upgrade" ? "upgrade" : "fill",
+        purpose,
         status: "draft",
         priority,
         evaluation,
@@ -642,63 +879,73 @@ async function reserve(
         payload_hash: hashJson(payload),
         expected_revision: p.slot.revision + 1,
         idempotency_key: key,
+        unit_keys: p.unitKeys,
       })
       .onConflict((oc) => oc.column("idempotency_key").doNothing())
       .returning("id")
       .executeTakeFirst();
-    return row?.id ?? null;
+    return { planId: row?.id ?? null };
   }
-  try {
+  return d.ctx.db.transaction().execute(async (trx) => {
-    return await d.ctx.db.transaction().execute(async (trx) => {
+    const slot = await trx
-      const slot = await trx
+      .selectFrom("acquisition_slot")
-        .selectFrom("acquisition_slot")
+      .selectAll()
-        .selectAll()
+      .where("id", "=", p.slot.id)
-        .where("id", "=", p.slot.id)
+      .forUpdate()
-        .forUpdate()
+      .executeTakeFirst();
-        .executeTakeFirst();
+    if (!slot) return { planId: null };
-      if (!slot || slot.revision !== p.slot.revision || BUSY.includes(slot.status)) return null;
+    const busy = BUSY.includes(slot.status);
-      let st: SlotState = { status: slot.status, hasAsset: slot.current_asset_id !== null };
+    if (p.multiInflight) {
+      // Slot×单元预留:在 Slot 行锁内复核与其它在途计划的单元不相交
+      const others = await slotInflightPlans(trx, slot.id);
+      const hit = unitsOverlap(
+        p.unitKeys,
+        others.flatMap((o) => o.unitKeys),
+      );
+      if (hit.length) return { planId: null, conflict: hit };
+    } else if (slot.revision !== p.slot.revision || busy) {
+      return { planId: null };
+    }
+    let st: SlotState = { status: slot.status, hasAsset: slot.current_asset_id !== null };
+    if (!busy) {
       if (st.status === "satisfied") st = transition(st, { type: "upgrade_found" });
       st = transition(st, { type: "reserve" });
-      const plan = await trx
+    }
-        .insertInto("acquisition_plan")
+    const plan = await trx
-        .values({
+      .insertInto("acquisition_plan")
-          slot_id: slot.id,
+      .values({
-          source_release_id: p.rel.id,
+        slot_id: slot.id,
-          purpose: p.mode === "upgrade" ? "upgrade" : "fill",
+        source_release_id: p.rel.id,
-          status: "reserved",
+        purpose,
-          priority,
+        status: "reserved",
-          evaluation,
+        priority,
-          reserved_bytes: p.rel.size_bytes ?? 0,
+        evaluation,
-          budget_reservation: jsonb({
+        reserved_bytes: p.rel.size_bytes ?? 0,
-            bytes: p.rel.size_bytes,
+        budget_reservation: jsonb({
-            countsAsNewTitle: p.mode === "new_title",
+          bytes: p.rel.size_bytes,
-            day: startOfDay(now).toISOString().slice(0, 10),
+          countsAsNewTitle: p.mode === "new_title",
-          }),
+          day: startOfDay(now).toISOString().slice(0, 10),
-          payload_hash: hashJson(payload),
+        }),
-          expected_revision: slot.revision + 1,
+        payload_hash: hashJson(payload),
-          idempotency_key: `acq:${slot.id}:${p.rel.id}`,
+        expected_revision: slot.revision + 1,
-        })
+        idempotency_key: `acq:${slot.id}:${p.rel.id}`,
-        .onConflict((oc) => oc.column("idempotency_key").doNothing())
+        unit_keys: p.unitKeys,
-        .returning("id")
+      })
-        .executeTakeFirst();
+      .onConflict((oc) => oc.column("idempotency_key").doNothing())
-      if (!plan) return null;
+      .returning("id")
-      await trx
+      .executeTakeFirst();
-        .updateTable("acquisition_slot")
+    if (!plan) return { planId: null };
-        .set({
+    await trx
-          status: st.status,
+      .updateTable("acquisition_slot")
-          reserved_plan_id: plan.id,
+      .set({
-          revision: slot.revision + 1,
+        status: st.status,
-          last_evaluated_at: now,
+        ...(busy ? {} : { reserved_plan_id: plan.id }),
-          status_reason: `reserved:${p.mode}`,
+        revision: slot.revision + 1,
-        })
+        last_evaluated_at: now,
-        .where("id", "=", slot.id)
+        status_reason: busy ? `reserved_more:${p.mode}` : `reserved:${p.mode}`,
-        .execute();
+      })
-      return plan.id;
+      .where("id", "=", slot.id)
-    });
+      .execute();
-  } catch (err) {
+    return { planId: plan.id };
-    // 部分唯一索引:同一 Slot 同时只能有一个有效预留
+  });
-    if ((err as { code?: string }).code === "23505") return null;
-    throw err;
-  }
 }

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

@@ -1,10 +1,38 @@
-import { assertSafeToDeleteData, DownloaderError, type QbTorrent, SiteError, type TorrentSource } from "@umt/adapters";
+import {
+  assertSafeToDeleteData,
+  type DownloaderClient,
+  DownloaderError,
+  type QbTorrent,
+  SiteError,
+  type TorrentSource,
+} from "@umt/adapters";
 import type { DownloadStatus, PauseReason } from "@umt/contracts";
-import { type DB, type DownloadTaskRow, sql, type Transaction } from "@umt/db";
+import {
-import { mapQbState, type SlotEvent, transition } from "@umt/domain";
+  type AcquisitionPlanRow,
+  type DB,
+  type DownloadTaskRow,
+  type SourceReleaseRow,
+  sql,
+  type Transaction,
+} from "@umt/db";
+import {
+  episodeKey,
+  fileKindOf,
+  mapQbState,
+  type PlanStage,
+  parseReleasePath,
+  parseUnitKey,
+  type SlotEvent,
+  type SlotState,
+  stageAfter,
+  transition,
+  transitionWithInflight,
+  unitsOverlap,
+} from "@umt/domain";
 import { mergeObservedDownloadState, toDomainDownloadStatus } from "@umt/mapping";
 import { withAttemptToken } from "@umt/tasks";
 import { errMessage, jsonb } from "../util.ts";
+import { parsedOf } from "./acquisition.ts";
 import {
   type AoDeps,
   automationBlock,
@@ -19,17 +47,69 @@ import {
   upsertDecision,
 } from "./common.ts";
 import { hnrSatisfied, retentionElapsed } from "./retention.ts";
+import {
+  afterSlotVerified,
+  type SlotCompletion,
+  slotComplete,
+  slotCoveredUnits,
+  slotInflightPlans,
+} from "./subscriptions.ts";
 import { decideVolume, loadVolumes, usedPercent, type VolumeState } from "./volumes.ts";
 
 // 下载:提交(幂等 + fencing)、qB 同步与标签接管、无进展证据、缓存卷硬阈值、缓存清理。
 
-/** Slot 状态推进(非法迁移只记日志不抛:并发下 Slot 可能已被人工处理) */
+/** 参与 Slot×单元多在途聚合的事件(其余事件按严格状态机) */
+const PLAN_EVENTS = new Set<SlotEvent["type"]>([
+  "reserve",
+  "release",
+  "download_started",
+  "download_completed",
+  "download_failed",
+  "verify_passed",
+  "verify_failed",
+]);
+
+/** 核验事件的来源计划与本次核验涉及的资产(organize 未传 planId 时按资产→下载任务→计划反查) */
+async function verifiedContext(
+  trx: Exec,
+  slotId: string,
+  assetId: string | null | undefined,
+): Promise<{ planId: string | null; assets: string[] }> {
+  if (!assetId) return { planId: null, assets: [] };
+  const a = await trx
+    .selectFrom("media_asset as a")
+    .leftJoin("download_task as dt", "dt.id", "a.download_task_id")
+    .leftJoin("acquisition_plan as p", "p.id", "dt.acquisition_plan_id")
+    .select(["dt.id as dt_id", "p.id as plan_id", "p.slot_id"])
+    .where("a.id", "=", assetId)
+    .executeTakeFirst();
+  const assets = new Set<string>([assetId]);
+  if (a?.dt_id) {
+    const rows = await sql<{ media_asset_id: string }>`
+      select f.media_asset_id from organization_file_result f
+        join organization_record r on r.id = f.organization_record_id
+       where r.download_task_id = ${a.dt_id} and f.role = 'main' and f.media_asset_id is not null
+         and r.id = (select id from organization_record where download_task_id = ${a.dt_id} order by created_at desc limit 1)`.execute(
+      trx,
+    );
+    for (const r of rows.rows) assets.add(r.media_asset_id);
+  }
+  return { planId: a?.slot_id === slotId ? (a?.plan_id ?? null) : null, assets: [...assets] };
+}
+
+/**
+ * Slot 状态推进(非法迁移只记日志不抛:并发下 Slot 可能已被人工处理)。
+ * 剧集 Slot 可有多个在途计划(Slot×单元预留):按"其它在途计划的阶段 + 本计划事件后阶段"聚合,
+ * Slot 显示最靠后阶段;最后一个计划结束才落到静止态。
+ * 核验通过时由本模块按季总集数账本重算完成(覆盖调用方传入的 complete),并写分集质量账本、
+ * 完成订阅(写历史);计划结束(释放/失败/核验)时标记 Slot 到期重评,处理忙时延后的发布。
+ */
 export async function applySlotEvent(
   d: AoDeps,
   trx: Exec,
   slotId: string,
   event: SlotEvent,
-  extra: { currentAssetId?: string | null; reason?: string } = {},
+  extra: { currentAssetId?: string | null; reason?: string; planId?: string | null } = {},
 ): Promise<boolean> {
   const slot = await trx
     .selectFrom("acquisition_slot")
@@ -38,26 +118,63 @@ export async function applySlotEvent(
     .forUpdate()
     .executeTakeFirst();
   if (!slot) return false;
+  let planId = extra.planId ?? null;
+  let completion: SlotCompletion | null = null;
+  let verified: string[] = [];
+  if (event.type === "verify_passed") {
+    const vc = await verifiedContext(trx, slotId, extra.currentAssetId);
+    planId ??= vc.planId;
+    verified = vc.assets;
+    const entity = await trx
+      .selectFrom("media_entity")
+      .select("kind")
+      .where("id", "=", slot.media_entity_id)
+      .executeTakeFirst();
+    completion = await slotComplete(trx, slot, entity?.kind === "movie" ? "movie" : "tv", verified);
+    event = { type: "verify_passed", complete: completion.complete };
+  }
+  let others: PlanStage[] = [];
+  if (PLAN_EVENTS.has(event.type)) {
+    const inflight = await slotInflightPlans(trx, slotId);
+    if (!planId && event.type === "verify_failed") {
+      // organize 阻塞/失败未传 planId:在途里处于核验阶段的那个视为本计划
+      planId = inflight.find((p) => p.stage === "verifying")?.id ?? null;
+    }
+    others = inflight.filter((p) => p.id !== planId).map((p) => p.stage);
+  }
+  const state = { status: slot.status, hasAsset: slot.current_asset_id !== null };
+  let next: SlotState | null;
   try {
-    const next = transition({ status: slot.status, hasAsset: slot.current_asset_id !== null }, event);
+    next = PLAN_EVENTS.has(event.type) ? transitionWithInflight(state, event, others) : transition(state, event);
-    await trx
-      .updateTable("acquisition_slot")
-      .set({
-        status: next.status,
-        revision: slot.revision + 1,
-        ...(extra.currentAssetId !== undefined ? { current_asset_id: extra.currentAssetId } : {}),
-        ...(next.status !== "reserved" && next.status !== "downloading" && next.status !== "verifying"
-          ? { reserved_plan_id: null }
-          : {}),
-        status_reason: extra.reason ?? event.type,
-      })
-      .where("id", "=", slotId)
-      .execute();
-    return true;
   } catch (err) {
     d.ctx.logger.warn({ slotId, event: event.type, err: { message: errMessage(err) } }, "Slot 状态迁移被拒绝");
     return false;
   }
+  if (!next) {
+    d.ctx.logger.warn({ slotId, event: event.type, from: slot.status }, "Slot 状态迁移被拒绝(已被人工处理?)");
+    return false;
+  }
+  const ended = PLAN_EVENTS.has(event.type) && stageAfter(event.type) === null;
+  const busy = next.status === "reserved" || next.status === "downloading" || next.status === "verifying";
+  await trx
+    .updateTable("acquisition_slot")
+    .set({
+      status: next.status,
+      revision: slot.revision + 1,
+      ...(extra.currentAssetId !== undefined ? { current_asset_id: extra.currentAssetId } : {}),
+      ...(!busy ? { reserved_plan_id: null } : {}),
+      status_reason:
+        extra.reason === "verified" || extra.reason === "verified_partial"
+          ? completion?.complete
+            ? "verified"
+            : "verified_partial"
+          : (extra.reason ?? event.type),
+      ...(ended ? { next_evaluate_at: d.ports.now() } : {}),
+    })
+    .where("id", "=", slotId)
+    .execute();
+  if (completion) await afterSlotVerified(d, trx, { slotId, planId, verifiedAssets: verified, completion });
+  return true;
 }
 
 async function emitStatus(
@@ -212,7 +329,13 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
           .set({ status: "submitted", status_reason: r.kind })
           .where("id", "=", plan.id)
           .execute();
-        await applySlotEvent(d, trx, plan.slot_id, { type: "download_started" }, { reason: `download:${r.kind}` });
+        await applySlotEvent(
+          d,
+          trx,
+          plan.slot_id,
+          { type: "download_started" },
+          { reason: `download:${r.kind}`, planId: plan.id },
+        );
         await d.ctx.outbox.emit(
           trx,
           "download.submitted",
@@ -224,6 +347,12 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
   );
   if (result.kind === "rejected" || result.kind === "source_failed") return { status: "failed", reason: result.reason };
   const hash = (result as Extract<SubmitResult, { hash: string }>).hash;
+  // 季包/多集包:按缺集选文件(接管的已有任务不改文件选择);失败不影响提交结果
+  if (result.kind === "added" && plan.unit_keys.length > 0 && plan.file_selection === null) {
+    await selectPackFiles(d, client, hash, plan, rel).catch((err) =>
+      d.ctx.logger.warn({ planId: plan.id, err: { message: errMessage(err) } }, "季包选文件失败,按整包下载"),
+    );
+  }
   if (!dtId) {
     // 账本已有结果(重放):按 hash 找回任务行
     dtId = (
@@ -238,13 +367,107 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
   return { status: "submitted", reason: result.kind, ...(dtId ? { downloadTaskId: dtId } : {}) };
 }
 
+const PACK_FILE_LIST = { retries: 3, intervalMs: 300 } as const;
+
+/**
+ * 季包按缺集选文件(对照 MP 从整季包按集选文件,仅 qB):提交后读文件清单,把不需要单元的文件优先级置 0;
+ * 需要单元 = 计划预留的集(季通配 = 该季未覆盖、且不在其它在途计划里的集)。
+ * 文件清单拿不到 / 有视频文件认不出集号 → 整包下载并记录理由;计划的 unit_keys 收敛为实际选中的集。
+ */
+async function selectPackFiles(
+  d: AoDeps,
+  client: DownloaderClient,
+  hash: string,
+  plan: AcquisitionPlanRow,
+  rel: SourceReleaseRow,
+): Promise<void> {
+  const parsed = parsedOf(rel);
+  const isPack = !parsed.episodes && !parsed.airDate;
+  const multi = !!parsed.episodes && parsed.episodes.end > parsed.episodes.start;
+  if (!isPack && !multi) return;
+  const record = (sel: Record<string, unknown>, unitKeys?: string[]) =>
+    d.ctx.db
+      .updateTable("acquisition_plan")
+      .set({ file_selection: jsonb(sel), ...(unitKeys ? { unit_keys: unitKeys } : {}) })
+      .where("id", "=", plan.id)
+      .execute();
+  const files = await client.files(hash, PACK_FILE_LIST).catch(() => []);
+  if (files.length === 0) return void (await record({ mode: "whole", reason: "file_list_unavailable" }));
+
+  // 季通配:展开为"该季未覆盖且不在其它在途计划里"
+  const wildSeasons = new Set(
+    plan.unit_keys
+      .map(parseUnitKey)
+      .filter((u) => u?.wildcard)
+      .map((u) => u!.season),
+  );
+  const explicit = new Set(plan.unit_keys.filter((k) => !parseUnitKey(k)?.wildcard));
+  let blocked = new Set<string>();
+  if (wildSeasons.size) {
+    const slot = await d.ctx.db
+      .selectFrom("acquisition_slot")
+      .select(["id", "current_asset_id", "media_entity_id"])
+      .where("id", "=", plan.slot_id)
+      .executeTakeFirstOrThrow();
+    const slots = await d.ctx.db
+      .selectFrom("acquisition_slot")
+      .select("id")
+      .where("media_entity_id", "=", slot.media_entity_id)
+      .execute();
+    const covered = await slotCoveredUnits(d.ctx.db, slot, { entityWide: slots.length <= 1 });
+    const others = await slotInflightPlans(d.ctx.db, plan.slot_id, plan.id);
+    blocked = new Set([...covered.keys, ...others.flatMap((o) => o.unitKeys)]);
+  }
+  const wantedKey = (k: string) => {
+    if (explicit.has(k)) return true;
+    const u = parseUnitKey(k);
+    return !!u && wildSeasons.has(u.season) && !blocked.has(k) && unitsOverlap([k], [...blocked]).length === 0;
+  };
+  const rows = files.map((f) => {
+    const kind = fileKindOf(f.name);
+    const p = parseReleasePath(f.name);
+    const keys: string[] = [];
+    if (p.episodes)
+      for (let e = p.episodes.start; e <= p.episodes.end && keys.length < 200; e++)
+        keys.push(episodeKey(p.season ?? parsed.season ?? 1, e));
+    return { index: f.index, name: f.name, kind, keys, wanted: keys.some(wantedKey) };
+  });
+  const videos = rows.filter((r) => r.kind === "video");
+  const unmapped = videos.filter((r) => r.keys.length === 0);
+  if (videos.length === 0 || unmapped.length > 0)
+    return void (await record({
+      mode: "whole",
+      reason: videos.length === 0 ? "no_video_files" : "unmapped_video_file",
+      files: unmapped.slice(0, 20).map((r) => r.name),
+    }));
+  const selected = [...new Set(videos.filter((r) => r.wanted).flatMap((r) => r.keys.filter(wantedKey)))];
+  if (selected.length === 0) return void (await record({ mode: "whole", reason: "no_missing_units_in_pack" }));
+  // 视频不需要 → 不下;附属文件(字幕等)有集号且不需要 → 不下;无集号的附属文件保留
+  const skip = rows.filter((r) => (r.kind === "video" ? !r.wanted : r.keys.length > 0 && !r.wanted));
+  if (skip.length > 0)
+    await client.setFilePriority(
+      hash,
+      skip.map((r) => r.index),
+      0,
+    );
+  await record(
+    {
+      mode: skip.length > 0 ? "selected" : "all_needed",
+      selectedUnits: selected,
+      skippedFiles: skip.slice(0, 200).map((r) => r.name),
+      totalFiles: files.length,
+    },
+    selected,
+  );
+}
+
 async function failPlanTx(d: AoDeps, trx: Exec, planId: string, slotId: string, reason: string) {
   await trx
     .updateTable("acquisition_plan")
     .set({ status: "failed", status_reason: reason.slice(0, 500) })
     .where("id", "=", planId)
     .execute();
-  await applySlotEvent(d, trx, slotId, { type: "release" }, { reason: `submit_failed` });
+  await applySlotEvent(d, trx, slotId, { type: "release" }, { reason: `submit_failed`, planId });
   await upsertDecision(d, trx, {
     kind: "download_failed",
     title: "下载提交失败",
@@ -256,6 +479,25 @@ async function failPlanTx(d: AoDeps, trx: Exec, planId: string, slotId: string,
   });
 }
 
+/** qB 里已确认没有该任务(用户移除):在途计划释放(失败),其单元不再算"已预留",Slot 到期重评 */
+async function releaseRemovedPlan(d: AoDeps, trx: Exec, planId: string) {
+  const plan = await trx
+    .updateTable("acquisition_plan")
+    .set({ status: "failed", status_reason: "download_removed" })
+    .where("id", "=", planId)
+    .where("status", "=", "submitted")
+    .returning(["id", "slot_id"])
+    .executeTakeFirst();
+  if (plan)
+    await applySlotEvent(
+      d,
+      trx,
+      plan.slot_id,
+      { type: "download_failed", reason: "download_removed" },
+      { reason: "download_removed", planId: plan.id },
+    );
+}
+
 async function failPlan(d: AoDeps, planId: string, slotId: string, reason: string): Promise<SubmitOutcome> {
   await d.ctx.db.transaction().execute((trx) => failPlanTx(d, trx, planId, slotId, reason));
   return { status: "failed", reason };
@@ -407,6 +649,7 @@ async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats)
           .where("id", "=", row.id)
           .execute();
         await emitStatus(d, trx, row.id, row.status, "removed", null);
+        if (row.acquisition_plan_id) await releaseRemovedPlan(d, trx, row.acquisition_plan_id);
         stats.removed++;
       }
     });
@@ -486,11 +729,11 @@ async function observe(d: AoDeps, row: DownloadTaskRow, t: QbTorrent, now: Date,
       const plan = row.acquisition_plan_id
         ? await trx
             .selectFrom("acquisition_plan")
-            .select(["slot_id"])
+            .select(["id", "slot_id"])
             .where("id", "=", row.acquisition_plan_id)
             .executeTakeFirst()
         : null;
-      if (plan) await applySlotEvent(d, trx, plan.slot_id, { type: "download_completed" });
+      if (plan) await applySlotEvent(d, trx, plan.slot_id, { type: "download_completed" }, { planId: plan.id });
     }
     if (stalledNow) {
       // 无进展:只记证据 + 待处理,保留原任务,不换源(REQ §8.4.2)

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

@@ -0,0 +1,800 @@
+import type { AcquisitionSlot, SubscriptionParams, SubscriptionProgress, UnitProgressState } from "@umt/contracts";
+import type { AcquisitionSlotRow, SubscriptionRow } from "@umt/db";
+import { sql } from "@umt/db";
+import {
+  episodeKey,
+  movieCompletion,
+  type ParsedRelease,
+  parseUnitKey,
+  priorityLadder,
+  type QualityFacts,
+  resolveEpisodeTotal,
+  type SeasonProgress,
+  seasonProgress,
+  type UnitRange,
+} from "@umt/domain";
+import { slotSpecFromRow } from "@umt/mapping";
+import { errMessage, jsonb, toApi } from "../util.ts";
+import { type AoDeps, type Exec, upsertDecision } from "./common.ts";
+
+// W5-C 订阅业务:季总集数账本、缺集/在途、完成判定与历史、Slot 重评调度、分集质量账本、主动搜索调度。
+// 对照 docs/porting/moviepilot-subscribe.md §3/§7/§8/§9;差距清单 docs/plan/05-subscription-gap.md。
+
+/** 在途计划状态(等价 MP note:提交成功即计入,失败/移除释放) */
+export const INFLIGHT_PLAN_STATUSES = ["reserved", "submitted"] as const;
+/** 仍参与评估/搜索的订阅状态(pending = TMDB 未就绪:可搜索/评估,不可完成) */
+export const TRACKING_SUB_STATUSES = ["active", "pending"] as const;
+
+export const DEFAULT_SLOT_REEVALUATE_MS = 6 * 3600_000;
+export const DEFAULT_METADATA_REFRESH_MS = 6 * 3600_000;
+
+export function subParams(sub: Pick<SubscriptionRow, "params"> | null | undefined): SubscriptionParams {
+  return (sub?.params ?? {}) as SubscriptionParams;
+}
+
+export function rangeOf(sub: Pick<SubscriptionRow, "params"> | null | undefined): UnitRange | undefined {
+  const r = subParams(sub).episodeRange;
+  if (!r || (r.start === undefined && r.end === undefined)) return undefined;
+  return { ...(r.start !== undefined ? { start: r.start } : {}), ...(r.end !== undefined ? { end: r.end } : {}) };
+}
+
+// ── 季总集数账本 ──
+
+export interface SeasonLedger {
+  seasonUnitId: string | null;
+  seasonNumber: number;
+  total: number | null;
+  known: boolean;
+  source: string | null;
+  locked: boolean;
+}
+
+/** 作品各季的总集数账本(季单元 episode_total);只列季号 ≥ 1 的正片季 */
+export async function seasonLedgers(exec: Exec, entityId: string): Promise<Map<number, SeasonLedger>> {
+  const rows = await exec
+    .selectFrom("content_unit")
+    .select(["id", "season_number", "episode_total", "episode_total_source", "episode_total_locked"])
+    .where("media_entity_id", "=", entityId)
+    .where("kind", "=", "season")
+    .where("status", "!=", "retired")
+    .where("season_number", "is not", null)
+    .orderBy("created_at")
+    .execute();
+  const out = new Map<number, SeasonLedger>();
+  for (const r of rows) {
+    const n = r.season_number!;
+    if (n < 1 || out.has(n)) continue;
+    out.set(n, {
+      seasonUnitId: r.id,
+      seasonNumber: n,
+      total: r.episode_total,
+      known: r.episode_total !== null && r.episode_total > 0,
+      source: r.episode_total_source,
+      locked: r.episode_total_locked,
+    });
+  }
+  return out;
+}
+
+/** 找或建季单元(TMDB 季信息/人工设置总集数时用;编号来自元数据,状态 confirmed) */
+export async function ensureSeasonUnit(exec: Exec, entityId: string, season: number): Promise<string> {
+  const hit = await exec
+    .selectFrom("content_unit")
+    .select("id")
+    .where("media_entity_id", "=", entityId)
+    .where("kind", "=", "season")
+    .where("season_number", "=", season)
+    .where("status", "!=", "retired")
+    .orderBy("created_at")
+    .executeTakeFirst();
+  if (hit) return hit.id;
+  return (
+    await exec
+      .insertInto("content_unit")
+      .values({ media_entity_id: entityId, kind: "season", season_number: season, status: "confirmed" })
+      .returning("id")
+      .executeTakeFirstOrThrow()
+  ).id;
+}
+
+/** 某季已入库覆盖的最高集号(回退兜底用) */
+async function coveredMaxEpisode(exec: Exec, entityId: string, season: number): Promise<number | null> {
+  const r = await sql<{ max: number | null }>`
+    select max(u.episode_number)::int as max from asset_unit_coverage c
+      join content_unit u on u.id = c.content_unit_id
+      join media_asset a on a.id = c.media_asset_id
+     where u.media_entity_id = ${entityId} and u.kind = 'episode' and u.season_number = ${season}
+       and a.status = 'active'`.execute(exec);
+  return r.rows[0]?.max ?? null;
+}
+
+/**
+ * 写入季总集数(元数据刷新 / 注入 / 人工):按 resolveEpisodeTotal 只升、回退不低于已覆盖最高集、锁定不改。
+ * manual=true 时为人工设置:可调低并可锁定。
+ */
+export async function applyEpisodeTotal(
+  exec: Exec,
+  input: {
+    entityId: string;
+    season: number;
+    fresh?: number | null;
+    injected?: number | null;
+    manual?: { total: number; locked: boolean };
+    now: Date;
+  },
+): Promise<{ seasonUnitId: string; total: number | null; known: boolean; changed: boolean }> {
+  const unitId = await ensureSeasonUnit(exec, input.entityId, input.season);
+  const cur = await exec
+    .selectFrom("content_unit")
+    .select(["episode_total", "episode_total_locked", "episode_total_source"])
+    .where("id", "=", unitId)
+    .forUpdate()
+    .executeTakeFirstOrThrow();
+  if (input.manual) {
+    await exec
+      .updateTable("content_unit")
+      .set({
+        episode_total: input.manual.total,
+        episode_total_source: "manual",
+        episode_total_locked: input.manual.locked,
+        episode_total_at: input.now,
+      })
+      .where("id", "=", unitId)
+      .execute();
+    return {
+      seasonUnitId: unitId,
+      total: input.manual.total,
+      known: true,
+      changed: input.manual.total !== cur.episode_total,
+    };
+  }
+  const r = resolveEpisodeTotal({
+    previous: cur.episode_total,
+    fresh: input.fresh ?? null,
+    injected: input.injected ?? null,
+    coveredMax: await coveredMaxEpisode(exec, input.entityId, input.season),
+    locked: cur.episode_total_locked,
+  });
+  if (r.changed) {
+    await exec
+      .updateTable("content_unit")
+      .set({ episode_total: r.total, episode_total_source: r.source, episode_total_at: input.now })
+      .where("id", "=", unitId)
+      .execute();
+  }
+  return { seasonUnitId: unitId, total: r.total, known: r.known, changed: r.changed };
+}
+
+// ── 覆盖 / 在途 ──
+
+/** Slot 已覆盖的内容单元:该 Slot 计划产出资产 + 当前资产 + 分集账本(content_unit id 与合成键两种口径) */
+export async function slotCoveredUnits(
+  exec: Exec,
+  slot: Pick<AcquisitionSlotRow, "id" | "current_asset_id" | "media_entity_id">,
+  opts: { entityWide?: boolean } = {},
+): Promise<{ keys: Set<string>; episodes: Map<number, Set<number>> }> {
+  const r = await sql<{
+    unit_id: string;
+    season_number: number | null;
+    episode_number: number | null;
+    kind: string;
+    air_date: string | null;
+  }>`
+    with assets as (
+      select a.id from media_asset a
+        left join download_task dt on dt.id = a.download_task_id
+        left join acquisition_plan p on p.id = dt.acquisition_plan_id
+       where a.status = 'active' and a.parent_asset_id is null
+         and (p.slot_id = ${slot.id} or a.id = ${slot.current_asset_id}
+              or a.id in (select media_asset_id from slot_unit where slot_id = ${slot.id})
+              ${opts.entityWide ? sql`or a.media_entity_id = ${slot.media_entity_id}` : sql``})
+    )
+    select u.id as unit_id, u.season_number, u.episode_number, u.kind, to_char(u.air_date, 'YYYYMMDD') as air_date
+      from asset_unit_coverage c
+      join assets on assets.id = c.media_asset_id
+      join content_unit u on u.id = c.content_unit_id
+    union
+    select u.id, u.season_number, u.episode_number, u.kind, to_char(u.air_date, 'YYYYMMDD') from slot_unit su
+      join content_unit u on u.id = su.content_unit_id
+     where su.slot_id = ${slot.id} and su.media_asset_id is not null`.execute(exec);
+  const keys = new Set<string>();
+  const episodes = new Map<number, Set<number>>();
+  for (const row of r.rows) {
+    keys.add(row.unit_id);
+    // 综艺按播出日期成单元:与发布侧合成键 d{YYYYMMDD} 同口径
+    if (row.air_date) keys.add(`d${row.air_date}`);
+    if (row.kind === "episode" && row.episode_number !== null) {
+      const s = row.season_number ?? 1;
+      keys.add(episodeKey(s, row.episode_number));
+      const set = episodes.get(s) ?? new Set<number>();
+      set.add(row.episode_number);
+      episodes.set(s, set);
+    }
+  }
+  // 入库时记在资产上的合成键(无 content_unit 的旧数据)
+  const legacy = await sql<{ keys: string[] | null }>`
+    select array(select jsonb_array_elements_text(coalesce(a.media_info->'_umt'->'unitKeys', '[]'::jsonb))) as keys
+      from media_asset a
+      left join download_task dt on dt.id = a.download_task_id
+      left join acquisition_plan p on p.id = dt.acquisition_plan_id
+     where a.status = 'active' and a.parent_asset_id is null
+       and (p.slot_id = ${slot.id} or a.id = ${slot.current_asset_id})`.execute(exec);
+  for (const row of legacy.rows)
+    for (const k of row.keys ?? []) {
+      keys.add(k);
+      const p = parseUnitKey(k);
+      if (p?.episode !== undefined) {
+        const set = episodes.get(p.season) ?? new Set<number>();
+        set.add(p.episode);
+        episodes.set(p.season, set);
+      }
+    }
+  return { keys, episodes };
+}
+
+export interface InflightPlan {
+  id: string;
+  status: string;
+  unitKeys: string[];
+  stage: "reserved" | "downloading" | "verifying";
+}
+
+/** Slot 的在途计划(reserved/submitted)及其阶段 */
+export async function slotInflightPlans(exec: Exec, slotId: string, excludePlanId?: string): Promise<InflightPlan[]> {
+  let q = exec
+    .selectFrom("acquisition_plan as p")
+    .leftJoin("download_task as dt", (j) =>
+      j.onRef("dt.acquisition_plan_id", "=", "p.id").on("dt.removed_at", "is", null),
+    )
+    .select(["p.id", "p.status", "p.unit_keys", "dt.status as dt_status"])
+    .where("p.slot_id", "=", slotId)
+    .where("p.status", "in", [...INFLIGHT_PLAN_STATUSES]);
+  if (excludePlanId) q = q.where("p.id", "!=", excludePlanId);
+  const rows = await q.execute();
+  const byId = new Map<string, InflightPlan>();
+  for (const r of rows) {
+    const stage: InflightPlan["stage"] =
+      r.status === "reserved"
+        ? "reserved"
+        : r.dt_status === "completed" || r.dt_status === "importing"
+          ? "verifying"
+          : "downloading";
+    const prev = byId.get(r.id);
+    if (!prev || stage === "verifying") byId.set(r.id, { id: r.id, status: r.status, unitKeys: r.unit_keys, stage });
+  }
+  return [...byId.values()];
+}
+
+// ── 完成判定 ──
+
+export interface SlotCompletion {
+  complete: boolean;
+  seasons: SeasonProgress[];
+  /** 继续追踪时的疑似缺集(总集数未知,已见最高集以下的空洞) */
+  suspectedMissing: { season: number; episodes: number[] }[];
+}
+
+/** 剧集 Slot 覆盖的季:Slot 限定季单元 > 订阅季单元 > 作品已知的所有正片季 */
+async function scopeSeasons(
+  exec: Exec,
+  slot: Pick<AcquisitionSlotRow, "media_entity_id" | "content_unit_id">,
+  sub: Pick<SubscriptionRow, "season_unit_id"> | null,
+  covered: Map<number, Set<number>>,
+  ledgers: Map<number, SeasonLedger>,
+): Promise<number[]> {
+  const unitId = slot.content_unit_id ?? sub?.season_unit_id ?? null;
+  if (unitId) {
+    const u = await exec
+      .selectFrom("content_unit")
+      .select(["season_number", "kind"])
+      .where("id", "=", unitId)
+      .executeTakeFirst();
+    if (u?.season_number !== null && u?.season_number !== undefined) return [u.season_number];
+  }
+  const eps = await exec
+    .selectFrom("content_unit")
+    .select("season_number")
+    .distinct()
+    .where("media_entity_id", "=", slot.media_entity_id)
+    .where("kind", "=", "episode")
+    .where("content_type", "=", "main")
+    .where("status", "!=", "retired")
+    .where("season_number", ">=", 1)
+    .execute();
+  const set = new Set<number>([...ledgers.keys(), ...covered.keys(), ...eps.map((e) => e.season_number!)]);
+  return [...set].filter((s) => s >= 1).sort((a, b) => a - b);
+}
+
+/**
+ * Slot 完成判定。电影按内容版本;剧集/综艺按季单元总集数(TMDB 账本)与实际文件覆盖:
+ * 总集数未知 → 不完成、继续追踪并报告疑似缺集;订阅 pending(TMDB 未就绪)→ 不完成。
+ */
+export async function slotComplete(
+  exec: Exec,
+  slot: Pick<
+    AcquisitionSlotRow,
+    "id" | "media_entity_id" | "content_unit_id" | "completion" | "subscription_id" | "current_asset_id"
+  >,
+  mediaType: "movie" | "tv",
+  newAssets: readonly string[],
+): Promise<SlotCompletion> {
+  const completion = (slot.completion ?? {}) as { kind?: string; cuts?: string[] };
+  if (mediaType === "movie") {
+    const assets = newAssets.length
+      ? await exec
+          .selectFrom("media_asset")
+          .select(["id", "content_edition"])
+          .where("id", "in", [...newAssets])
+          .execute()
+      : [];
+    const complete = movieCompletion(
+      completion.cuts as never,
+      assets.map((a) => ({
+        assetId: a.id,
+        verified: true,
+        ...(a.content_edition ? { cut: a.content_edition as never } : {}),
+      })),
+    ).complete;
+    return { complete, seasons: [], suspectedMissing: [] };
+  }
+  const sub = slot.subscription_id
+    ? ((await exec.selectFrom("subscription").selectAll().where("id", "=", slot.subscription_id).executeTakeFirst()) ??
+      null)
+    : null;
+  const slotCount = await exec
+    .selectFrom("acquisition_slot")
+    .select((eb) => eb.fn.countAll<number>().as("n"))
+    .where("media_entity_id", "=", slot.media_entity_id)
+    .executeTakeFirstOrThrow();
+  // 单 Slot 作品:作品下所有资产都算该 Slot 的覆盖(手动整理/历史导入);多 Slot 只算本 Slot 的
+  const covered = await slotCoveredUnits(exec, slot, { entityWide: Number(slotCount.n) <= 1 });
+  const inflight = await slotInflightPlans(exec, slot.id);
+  const inflightEps = new Map<number, number[]>();
+  for (const p of inflight)
+    for (const k of p.unitKeys) {
+      const u = parseUnitKey(k);
+      if (u?.episode !== undefined) inflightEps.set(u.season, [...(inflightEps.get(u.season) ?? []), u.episode]);
+    }
+  const ledgers = await seasonLedgers(exec, slot.media_entity_id);
+  const seasons = await scopeSeasons(exec, slot, sub, covered.episodes, ledgers);
+  const range = rangeOf(sub);
+  const progress = seasons.map((s) => {
+    const l = ledgers.get(s);
+    return seasonProgress({
+      season: s,
+      total: l?.total ?? null,
+      totalKnown: !!l?.known,
+      covered: [...(covered.episodes.get(s) ?? [])],
+      inflight: inflightEps.get(s) ?? [],
+      ...(range ? { range } : {}),
+    });
+  });
+  const complete = sub?.status !== "pending" && progress.length > 0 && progress.every((p) => p.complete);
+  return {
+    complete,
+    seasons: progress,
+    suspectedMissing: progress
+      .filter((p) => p.suspectedMissing.length > 0)
+      .map((p) => ({ season: p.season, episodes: p.suspectedMissing })),
+  };
+}
+
+// ── 核验通过后:分集账本、订阅完成、重评调度 ──
+
+const qualityOf = (mediaInfo: unknown): QualityFacts =>
+  ((mediaInfo as { _umt?: { quality?: QualityFacts } })?._umt?.quality ?? {}) as QualityFacts;
+
+/**
+ * 核验入库后(organize.verifyAndImport 在同一事务里调用):
+ * 1) 分集质量账本:该计划覆盖的单元指向新资产(升级计划只替换计划内单元;其余单元缺账本时补登);
+ * 2) 订阅完成:once 订阅完成;continuous + 仅本季 的剧集订阅季完整也完成;pending 不完成;写历史;
+ * 3) 疑似缺集记待处理(info);4) Slot 标记到期重评(处理忙时延后的发布)。
+ */
+export async function afterSlotVerified(
+  d: AoDeps,
+  trx: Exec,
+  input: { slotId: string; planId: string | null; verifiedAssets: readonly string[]; completion: SlotCompletion },
+): Promise<void> {
+  const slot = await trx.selectFrom("acquisition_slot").selectAll().where("id", "=", input.slotId).executeTakeFirst();
+  if (!slot) return;
+  const now = d.ports.now();
+  const plan = input.planId
+    ? await trx
+        .selectFrom("acquisition_plan")
+        .select(["id", "purpose", "unit_keys"])
+        .where("id", "=", input.planId)
+        .executeTakeFirst()
+    : undefined;
+  if (input.verifiedAssets.length) {
+    const spec = slotSpecFromRow(toApi<AcquisitionSlot>(slot));
+    const cov = await sql<{
+      asset_id: string;
+      unit_id: string;
+      media_info: unknown;
+      season: number | null;
+      ep: number | null;
+    }>`
+      select c.media_asset_id as asset_id, c.content_unit_id as unit_id, a.media_info,
+             u.season_number as season, u.episode_number as ep
+        from asset_unit_coverage c
+        join media_asset a on a.id = c.media_asset_id
+        join content_unit u on u.id = c.content_unit_id
+       where c.media_asset_id in (${sql.join([...input.verifiedAssets])}) and u.kind = 'episode'`.execute(trx);
+    const planUnits = new Set(plan?.unit_keys ?? []);
+    for (const r of cov.rows) {
+      const q = qualityOf(r.media_info);
+      const pri = priorityLadder(spec, { raw: "", hdr: [], isEncode: false, ...q } as unknown as ParsedRelease).pri;
+      const inPlan = planUnits.has(r.unit_id) || (r.ep !== null && planUnits.has(episodeKey(r.season ?? 1, r.ep)));
+      const replace = plan?.purpose === "upgrade" ? inPlan : true;
+      const values = {
+        slot_id: slot.id,
+        content_unit_id: r.unit_id,
+        media_asset_id: r.asset_id,
+        plan_id: plan?.id ?? null,
+        quality: jsonb(q),
+        pri,
+      };
+      if (replace) {
+        await trx
+          .insertInto("slot_unit")
+          .values(values)
+          .onConflict((oc) =>
+            oc.columns(["slot_id", "content_unit_id"]).doUpdateSet({
+              media_asset_id: values.media_asset_id,
+              plan_id: values.plan_id,
+              quality: values.quality,
+              pri: values.pri,
+            }),
+          )
+          .execute();
+      } else {
+        await trx
+          .insertInto("slot_unit")
+          .values(values)
+          .onConflict((oc) => oc.columns(["slot_id", "content_unit_id"]).doNothing())
+          .execute();
+      }
+    }
+  }
+
+  if (input.completion.suspectedMissing.length) {
+    const desc = input.completion.suspectedMissing
+      .map((s) => `S${String(s.season).padStart(2, "0")}:${s.episodes.map((e) => `E${e}`).join(",")}`)
+      .join(";");
+    await upsertDecision(d, trx, {
+      kind: "other",
+      severity: "info",
+      title: "疑似缺集(总集数未知,继续追踪)",
+      summary: `已入库的最高集以下仍缺:${desc}`.slice(0, 500),
+      subjectType: "acquisition_slot",
+      subjectId: slot.id,
+      dedupeKey: `suspected_missing:${slot.id}`,
+      evidence: { slotId: slot.id, suspectedMissing: input.completion.suspectedMissing },
+    });
+  }
+
+  if (input.completion.complete && slot.subscription_id) {
+    const sub = await trx
+      .selectFrom("subscription")
+      .selectAll()
+      .where("id", "=", slot.subscription_id)
+      .executeTakeFirst();
+    const seasonDone = sub?.media_kind !== "movie" && sub?.season_scope === "current_season";
+    if (sub && sub.status === "active" && (sub.mode === "once" || seasonDone)) {
+      await completeSubscription(d, trx, sub, "slot_satisfied", { seasons: input.completion.seasons });
+    }
+  }
+  // 计划结束:Slot 到期重评(忙时延后的发布、连续更新的下一集)
+  await markSlotDue(trx, slot.id, now);
+}
+
+/** Slot 标记立即到期重评(事务内写,由编排循环 sweepDueSlots 消费) */
+export async function markSlotDue(exec: Exec, slotId: string, now: Date): Promise<void> {
+  await exec.updateTable("acquisition_slot").set({ next_evaluate_at: now }).where("id", "=", slotId).execute();
+}
+
+/** 订阅完成:状态 completed + 写 subscription_history(行保留,可改回 active 重新订阅) */
+export async function completeSubscription(
+  d: AoDeps,
+  trx: Exec,
+  sub: SubscriptionRow,
+  reason: string,
+  progress: Record<string, unknown> = {},
+): Promise<boolean> {
+  const done = await trx
+    .updateTable("subscription")
+    .set({ status: "completed", status_reason: reason, next_search_at: null })
+    .where("id", "=", sub.id)
+    .where("status", "in", [...TRACKING_SUB_STATUSES])
+    .returningAll()
+    .executeTakeFirst();
+  if (!done) return false;
+  await trx
+    .insertInto("subscription_history")
+    .values({
+      subscription_id: done.id,
+      media_entity_id: done.media_entity_id,
+      season_unit_id: done.season_unit_id,
+      media_kind: done.media_kind,
+      mode: done.mode,
+      reason,
+      snapshot: jsonb(toApi(done)),
+      progress: jsonb(progress),
+      completed_at: d.ports.now(),
+    })
+    .execute();
+  d.ctx.logger.info({ subscriptionId: sub.id, reason }, "订阅完成,已写历史");
+  return true;
+}
+
+// ── 延后与重评调度 ──
+
+export async function recordDeferral(
+  exec: Exec,
+  input: { slotId: string; sourceReleaseId: string; reason: string; unitKeys: readonly string[] },
+): Promise<void> {
+  await sql`
+    insert into slot_deferral (slot_id, source_release_id, reason, unit_keys)
+    values (${input.slotId}, ${input.sourceReleaseId}, ${input.reason}, ${[...input.unitKeys]})
+    on conflict (slot_id, source_release_id) do update
+      set reason = excluded.reason, unit_keys = excluded.unit_keys, attempts = slot_deferral.attempts + 1,
+          updated_at = now()`.execute(exec);
+}
+
+export async function clearDeferral(exec: Exec, slotId: string, sourceReleaseId: string): Promise<void> {
+  await exec
+    .deleteFrom("slot_deferral")
+    .where("slot_id", "=", slotId)
+    .where("source_release_id", "=", sourceReleaseId)
+    .execute();
+}
+
+/**
+ * 到期 Slot:next_evaluate_at <= now → 入队 evaluateAcquisition{acquisitionSlotId}(先评 deferred,再评最近发布);
+ * 仍需追踪的 Slot 顺延一个周期,否则清空。返回入队数。
+ */
+export async function sweepDueSlots(d: AoDeps): Promise<number> {
+  const now = d.ports.now();
+  const every = d.ports.options.slotReevaluateIntervalMs ?? DEFAULT_SLOT_REEVALUATE_MS;
+  const due = await d.ctx.db.transaction().execute(async (trx) => {
+    const rows = await sql<{ id: string; status: string; sub_status: string | null; has_sub: boolean; due: Date }>`
+      select s.id, s.status, sub.status as sub_status, s.subscription_id is not null as has_sub, s.next_evaluate_at as due
+        from acquisition_slot s left join subscription sub on sub.id = s.subscription_id
+       where s.next_evaluate_at is not null and s.next_evaluate_at <= ${now}
+       order by s.next_evaluate_at limit 20
+       for update of s skip locked`.execute(trx);
+    for (const r of rows.rows) {
+      const tracking =
+        !r.has_sub || (r.sub_status !== null && (TRACKING_SUB_STATUSES as readonly string[]).includes(r.sub_status));
+      const keep = tracking && every > 0 && r.status !== "satisfied";
+      await trx
+        .updateTable("acquisition_slot")
+        .set({ next_evaluate_at: keep ? new Date(now.getTime() + every) : null })
+        .where("id", "=", r.id)
+        .execute();
+    }
+    return rows.rows;
+  });
+  let n = 0;
+  for (const r of due) {
+    try {
+      await d.ctx.engine.enqueue(
+        "evaluateAcquisition",
+        { acquisitionSlotId: r.id },
+        { idempotencyKey: `slot-due:${r.id}:${new Date(r.due).getTime()}`, traceId: r.id },
+      );
+      n++;
+    } catch (err) {
+      d.ctx.logger.warn({ slotId: r.id, err: { message: errMessage(err) } }, "Slot 重评入队失败");
+    }
+  }
+  return n;
+}
+
+/** 首次进入追踪的 Slot 排上周期重评(只在未排期时设置,不覆盖"立即到期"标记) */
+export async function scheduleSlotReevaluate(d: AoDeps, exec: Exec, slotId: string): Promise<void> {
+  const every = d.ports.options.slotReevaluateIntervalMs ?? DEFAULT_SLOT_REEVALUATE_MS;
+  if (every <= 0) return;
+  await exec
+    .updateTable("acquisition_slot")
+    .set({ next_evaluate_at: new Date(d.ports.now().getTime() + every) })
+    .where("id", "=", slotId)
+    .where("next_evaluate_at", "is", null)
+    .where("status", "!=", "satisfied")
+    .execute();
+}
+
+/** 手动/新建订阅的主动搜索入队(searchSubscription 由 sources-identity 注册) */
+export async function enqueueSubscriptionSearch(
+  d: AoDeps,
+  subscriptionId: string,
+  reason: "created" | "manual" | "schedule",
+): Promise<{ runId: string; deduplicated: boolean }> {
+  const now = d.ports.now();
+  const key =
+    reason === "schedule"
+      ? `searchSubscription:${subscriptionId}:${Math.floor(now.getTime() / 3600_000)}`
+      : `searchSubscription:${subscriptionId}:${reason}:${now.getTime()}`;
+  return d.ctx.engine.enqueue(
+    "searchSubscription",
+    { subscriptionId, manual: reason === "manual" },
+    { idempotencyKey: key, traceId: subscriptionId },
+  );
+}
+
+/** 周期主动搜索:next_search_at 到期的追踪中订阅(新建时由搜索任务排期;0 = 关闭周期搜索) */
+export async function sweepSubscriptionSearch(d: AoDeps): Promise<number> {
+  const now = d.ports.now();
+  const rows = await d.ctx.db.transaction().execute(async (trx) => {
+    const due = await sql<{ id: string }>`
+      select id from subscription
+       where status in ('active', 'pending') and next_search_at is not null and next_search_at <= ${now}
+       order by next_search_at limit 10 for update skip locked`.execute(trx);
+    for (const r of due.rows) {
+      // 先顺延一小时租约:任务成功后会按周期重写
+      await trx
+        .updateTable("subscription")
+        .set({ next_search_at: new Date(now.getTime() + 3600_000) })
+        .where("id", "=", r.id)
+        .execute();
+    }
+    return due.rows;
+  });
+  let n = 0;
+  for (const r of rows) {
+    try {
+      await enqueueSubscriptionSearch(d, r.id, "schedule");
+      n++;
+    } catch (err) {
+      d.ctx.logger.warn({ subscriptionId: r.id, err: { message: errMessage(err) } }, "订阅搜索入队失败");
+    }
+  }
+  return n;
+}
+
+/** 追踪中剧集的元数据刷新(总集数):上次刷新超过周期(或从未刷新)且有 TMDB 身份 → 入队 refreshMetadata */
+export async function sweepMetadataRefresh(d: AoDeps): Promise<number> {
+  const every = d.ports.options.metadataRefreshIntervalMs ?? DEFAULT_METADATA_REFRESH_MS;
+  if (every <= 0) return 0;
+  const now = d.ports.now();
+  const before = new Date(now.getTime() - every);
+  const rows = await sql<{ id: string }>`
+    select distinct e.id from subscription s
+      join media_entity e on e.id = s.media_entity_id
+     where s.status in ('active', 'pending') and s.media_kind <> 'movie' and e.status <> 'archived'
+       and (e.last_metadata_refresh_at is null or e.last_metadata_refresh_at < ${before})
+       and exists (select 1 from external_identity x where x.media_entity_id = e.id and x.provider = 'tmdb' and x.status = 'accepted')
+     limit 20`.execute(d.ctx.db);
+  const slot = Math.floor(now.getTime() / every);
+  let n = 0;
+  for (const r of rows.rows) {
+    try {
+      await d.ctx.engine.enqueue(
+        "refreshMetadata",
+        { mediaEntityId: r.id },
+        { idempotencyKey: `refreshMetadata:tracking:${r.id}:${slot}`, traceId: r.id },
+      );
+      n++;
+    } catch (err) {
+      d.ctx.logger.warn({ mediaEntityId: r.id, err: { message: errMessage(err) } }, "元数据刷新入队失败");
+    }
+  }
+  return n;
+}
+
+// ── 订阅进度(API / UI) ──
+
+export async function subscriptionProgress(exec: Exec, subId: string): Promise<SubscriptionProgress | null> {
+  const sub = await exec.selectFrom("subscription").selectAll().where("id", "=", subId).executeTakeFirst();
+  if (!sub) return null;
+  const slots = sub.media_entity_id
+    ? await exec
+        .selectFrom("acquisition_slot")
+        .selectAll()
+        .where("media_entity_id", "=", sub.media_entity_id)
+        .where((eb) => eb.or([eb("subscription_id", "=", sub.id), eb("subscription_id", "is", null)]))
+        .orderBy("created_at")
+        .execute()
+    : [];
+  const slotOut: SubscriptionProgress["slots"] = [];
+  for (const s of slots) {
+    const inflight = await slotInflightPlans(exec, s.id);
+    const deferred = await exec
+      .selectFrom("slot_deferral")
+      .select((eb) => eb.fn.countAll<number>().as("n"))
+      .where("slot_id", "=", s.id)
+      .executeTakeFirstOrThrow();
+    slotOut.push({
+      slotId: s.id,
+      name: s.name,
+      status: s.status,
+      inflightPlans: inflight.map((p) => ({ planId: p.id, status: p.status, unitKeys: p.unitKeys })),
+      deferred: Number(deferred.n),
+      nextEvaluateAt: s.next_evaluate_at ? s.next_evaluate_at.toISOString() : null,
+    });
+  }
+  const seasons: SubscriptionProgress["seasons"] = [];
+  if (sub.media_entity_id && sub.media_kind !== "movie") {
+    const main = slots[0];
+    const ledgers = await seasonLedgers(exec, sub.media_entity_id);
+    const covered = main
+      ? await slotCoveredUnits(exec, main, { entityWide: slots.length <= 1 })
+      : { keys: new Set<string>(), episodes: new Map<number, Set<number>>() };
+    const inflightEps = new Map<number, Set<number>>();
+    for (const s of slotOut)
+      for (const p of s.inflightPlans)
+        for (const k of p.unitKeys) {
+          const u = parseUnitKey(k);
+          if (u?.episode !== undefined)
+            inflightEps.set(u.season, (inflightEps.get(u.season) ?? new Set()).add(u.episode));
+        }
+    const nums = main
+      ? await scopeSeasons(exec, main, sub, covered.episodes, ledgers)
+      : [...ledgers.keys()].sort((a, b) => a - b);
+    const units = await exec
+      .selectFrom("content_unit")
+      .select(["id", "season_number", "episode_number"])
+      .where("media_entity_id", "=", sub.media_entity_id)
+      .where("kind", "=", "episode")
+      .where("status", "!=", "retired")
+      .execute();
+    const ledgerRows = main
+      ? await exec
+          .selectFrom("slot_unit")
+          .select(["content_unit_id", "quality", "pri"])
+          .where("slot_id", "=", main.id)
+          .execute()
+      : [];
+    const range = rangeOf(sub);
+    for (const n of nums) {
+      const l = ledgers.get(n);
+      const p = seasonProgress({
+        season: n,
+        total: l?.total ?? null,
+        totalKnown: !!l?.known,
+        covered: [...(covered.episodes.get(n) ?? [])],
+        inflight: [...(inflightEps.get(n) ?? [])],
+        ...(range ? { range } : {}),
+      });
+      const maxEp = Math.max(p.total ?? 0, ...p.covered, ...p.inflight, ...p.suspectedMissing, 0);
+      const list: SubscriptionProgress["seasons"][number]["units"] = [];
+      for (let e = 1; e <= maxEp && e <= 2000; e++) {
+        const cu = units.find((u) => (u.season_number ?? 1) === n && u.episode_number === e);
+        const lr = cu ? ledgerRows.find((r) => r.content_unit_id === cu.id) : undefined;
+        const q = (lr?.quality ?? null) as { resolution?: string; source?: string } | null;
+        const state: UnitProgressState =
+          (range?.start !== undefined && e < range.start) || (range?.end !== undefined && e > range.end)
+            ? "out_of_range"
+            : p.covered.includes(e)
+              ? "covered"
+              : p.inflight.includes(e)
+                ? "inflight"
+                : p.suspectedMissing.includes(e)
+                  ? "suspected_missing"
+                  : "missing";
+        list.push({
+          episode: e,
+          state,
+          contentUnitId: cu?.id ?? null,
+          quality: lr ? { resolution: q?.resolution ?? null, source: q?.source ?? null, pri: lr.pri } : null,
+        });
+      }
+      seasons.push({
+        seasonUnitId: l?.seasonUnitId ?? null,
+        seasonNumber: n,
+        episodeTotal: l?.total ?? null,
+        totalKnown: !!l?.known,
+        totalSource: l?.source ?? null,
+        totalLocked: !!l?.locked,
+        units: list,
+        missing: p.missing,
+        suspectedMissing: p.suspectedMissing,
+        complete: p.complete,
+      });
+    }
+  }
+  const completable =
+    sub.status !== "pending" &&
+    (sub.media_kind === "movie" || (seasons.length > 0 && seasons.every((s) => s.totalKnown)));
+  return { subscriptionId: sub.id, status: sub.status, completable, seasons, slots: slotOut };
+}

apps/server/src/modules/sources-identity/service/metadata-totals.ts

@@ -0,0 +1,82 @@
+import type { CandidateSeason } from "@umt/adapters";
+import { sql } from "@umt/db";
+import { resolveEpisodeTotal } from "@umt/domain";
+import type { Exec } from "./common.ts";
+
+// W5-C 季总集数账本(content_unit(kind=season).episode_total):refreshMetadata 用 TMDB seasons[].episodeCount 刷新。
+// 规则在 domain resolveEpisodeTotal:只升;TMDB 回退不低于已覆盖最高集;人工锁定不改;未公布(0)保持未知。
+// 季单元不存在时按元数据建(confirmed);不预建集单元(集单元仍由发布/文件懒建,provisional 不代表总数)。
+
+export interface SeasonTotalsResult {
+  seasons: { season: number; total: number | null; known: boolean; changed: boolean }[];
+  /** 本次后该作品的追踪订阅从 pending 转 active 的数量 */
+  activated: number;
+}
+
+async function ensureSeason(exec: Exec, entityId: string, season: number, name: string | null): Promise<string> {
+  const hit = await exec
+    .selectFrom("content_unit")
+    .select("id")
+    .where("media_entity_id", "=", entityId)
+    .where("kind", "=", "season")
+    .where("season_number", "=", season)
+    .where("status", "!=", "retired")
+    .orderBy("created_at")
+    .executeTakeFirst();
+  if (hit) return hit.id;
+  return (
+    await exec
+      .insertInto("content_unit")
+      .values({ media_entity_id: entityId, kind: "season", season_number: season, status: "confirmed", title: name })
+      .returning("id")
+      .executeTakeFirstOrThrow()
+  ).id;
+}
+
+export async function applySeasonTotals(
+  exec: Exec,
+  entityId: string,
+  seasons: readonly CandidateSeason[],
+  now: Date,
+): Promise<SeasonTotalsResult> {
+  const out: SeasonTotalsResult = { seasons: [], activated: 0 };
+  for (const s of seasons) {
+    if (s.seasonNumber < 1) continue; // S00 特别篇不计入正片季
+    const unitId = await ensureSeason(exec, entityId, s.seasonNumber, s.name);
+    const cur = await exec
+      .selectFrom("content_unit")
+      .select(["episode_total", "episode_total_locked"])
+      .where("id", "=", unitId)
+      .forUpdate()
+      .executeTakeFirstOrThrow();
+    const covered = await sql<{ max: number | null }>`
+      select max(u.episode_number)::int as max from asset_unit_coverage c
+        join content_unit u on u.id = c.content_unit_id
+        join media_asset a on a.id = c.media_asset_id
+       where u.media_entity_id = ${entityId} and u.kind = 'episode' and u.season_number = ${s.seasonNumber}
+         and a.status = 'active'`.execute(exec);
+    const r = resolveEpisodeTotal({
+      previous: cur.episode_total,
+      fresh: s.episodeCount,
+      coveredMax: covered.rows[0]?.max ?? null,
+      locked: cur.episode_total_locked,
+    });
+    if (r.changed)
+      await exec
+        .updateTable("content_unit")
+        .set({ episode_total: r.total, episode_total_source: r.source, episode_total_at: now })
+        .where("id", "=", unitId)
+        .execute();
+    out.seasons.push({ season: s.seasonNumber, total: r.total, known: r.known, changed: r.changed });
+  }
+  // TMDB 已就绪:因元数据未就绪而 pending 的订阅转 active(仍按总集数是否已知决定能否完成)
+  const act = await exec
+    .updateTable("subscription")
+    .set({ status: "active", status_reason: "metadata_ready" })
+    .where("media_entity_id", "=", entityId)
+    .where("status", "=", "pending")
+    .where("status_reason", "=", "metadata_pending")
+    .executeTakeFirst();
+  out.activated = Number(act.numUpdatedRows ?? 0);
+  return out;
+}

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

@@ -6,6 +6,7 @@ import { jsonb } from "../rows.ts";
 import type { SourcesRuntime } from "../runtime.ts";
 import { type Actor, audit, type Exec } from "./common.ts";
 import { resolveEntityId, seedAliases } from "./entities.ts";
+import { applySeasonTotals, type SeasonTotalsResult } from "./metadata-totals.ts";
 
 // refreshMetadata:字段级刷新(REQ §7.3)。来源值写 metadata_field;最终值 = 覆盖(锁)> 已锁字段保持 > 来源值。
 // 只更新 media_entity 的展示字段与别名,不触碰 path_binding(路径绑定不随元数据变化)。
@@ -34,6 +35,8 @@ export interface RefreshResult {
   reason?: string;
   changedFields: string[];
   lockedFields: string[];
+  /** 剧集:季总集数账本刷新结果(W5-C) */
+  seasons?: SeasonTotalsResult["seasons"];
 }
 
 export async function refreshMetadata(rt: SourcesRuntime, mediaEntityId: string): Promise<RefreshResult> {
@@ -95,8 +98,13 @@ export async function refreshMetadata(rt: SourcesRuntime, mediaEntityId: string)
       .execute();
     await seedAliases(trx, id, detail);
     const r = await project(trx, id, now);
+    // 季总集数账本:只升;回退不低于已覆盖最高集;锁定不改(订阅完成判定据此)
+    const totals =
+      ident.entity_type === "tv" && detail.seasons?.length
+        ? await applySeasonTotals(trx, id, detail.seasons, now)
+        : null;
     await trx.updateTable("media_entity").set({ last_metadata_refresh_at: now }).where("id", "=", id).execute();
-    return { status: "refreshed" as const, ...r };
+    return { status: "refreshed" as const, ...r, ...(totals ? { seasons: totals.seasons } : {}) };
   });
 }
 

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

@@ -5,6 +5,7 @@ import { ingestFeed } from "./service/feeds.ts";
 import { refreshMetadata } from "./service/metadata.ts";
 import { resolveRelease } from "./service/resolve.ts";
 import { snapshotSite } from "./service/snapshot.ts";
+import { liveSearch } from "./service/workbench.ts";
 
 // 持久任务:输入只放业务 ID,处理器读业务表。外部请求失败的退避/熔断由适配器状态与业务表承担,
 // 凭据失效返回正常结果(不重试);只有意外错误(DB、TMDB 网络)抛出交给引擎重试。
@@ -53,6 +54,14 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu
       retryPolicy: NETWORK_RETRY,
       handler: (t) => t.step.run("refresh", () => refreshMetadata(rt(), t.input.mediaEntityId)),
     }),
+    defineTask({
+      name: "searchSubscription",
+      version: 1,
+      inputSchema: z.object({ subscriptionId: id, manual: z.boolean().optional() }),
+      retryPolicy: NETWORK_RETRY,
+      handler: (t) =>
+        t.step.run("search", () => searchSubscription(rt(), t.input.subscriptionId, { manual: !!t.input.manual })),
+    }),
     defineTask({
       name: "snapshotSite",
       version: 1,
@@ -70,6 +79,106 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu
   );
 }
 
+// ── 订阅主动搜索(W5-C,对照 MP SubscribeSearchOwner):按作品标题/原名/别名在已开启站点搜索 ──
+
+/** 主动搜索周期缺省 24h;subscription.params.searchIntervalHours 覆盖(0 = 关闭周期搜索) */
+export const DEFAULT_SEARCH_INTERVAL_HOURS = 24;
+/** 每次最多用几个检索词(标题、原名、别名),控制站点请求量 */
+export const SEARCH_MAX_KEYWORDS = 3;
+
+export interface SubscriptionSearchResult {
+  status: "ok" | "skipped";
+  reason?: string;
+  keywords: string[];
+  sites: string[];
+  searches: number;
+  errors: { siteId: string; keyword: string; error: string }[];
+  nextSearchAt: string | null;
+}
+
+/**
+ * 主动搜索:对订阅站点范围内已开启且健康的站点,按标题/原名/别名经站点适配器 search(共享限流器、实时补查合并与失败退避,
+ * 记 site_snapshot kind=live_search);新结果 upsert 为 source_release 并发 release.observed → 识别 → release.resolved →
+ * evaluateAcquisition(与 RSS 同一路径);已识别的旧结果由本任务末尾入队的订阅重评覆盖。
+ */
+export async function searchSubscription(
+  rt: SourcesRuntime,
+  subscriptionId: string,
+  opts: { manual?: boolean } = {},
+): Promise<SubscriptionSearchResult> {
+  const { db } = rt.ctx;
+  const none = { keywords: [], sites: [], searches: 0, errors: [], nextSearchAt: null };
+  const sub = await db.selectFrom("subscription").selectAll().where("id", "=", subscriptionId).executeTakeFirst();
+  if (!sub) return { status: "skipped", reason: "not_found", ...none };
+  if (!opts.manual && sub.status !== "active" && sub.status !== "pending")
+    return { status: "skipped", reason: `subscription_${sub.status}`, ...none };
+  if (!sub.media_entity_id) return { status: "skipped", reason: "no_entity", ...none };
+  const automation = await rt.ctx.settings.get("automation");
+  if (!opts.manual && (!automation.enabled || automation.readOnlyReconcile))
+    return { status: "skipped", reason: "automation_blocked", ...none };
+  const entity = await db
+    .selectFrom("media_entity")
+    .select(["id", "title", "original_title"])
+    .where("id", "=", sub.media_entity_id)
+    .executeTakeFirst();
+  if (!entity) return { status: "skipped", reason: "entity_not_found", ...none };
+  const aliases = await db
+    .selectFrom("title_alias")
+    .select(["alias", "kind"])
+    .where("media_entity_id", "=", entity.id)
+    .where("content_unit_id", "is", null)
+    .orderBy("created_at")
+    .limit(20)
+    .execute();
+  const seen = new Set<string>();
+  const keywords: string[] = [];
+  for (const k of [entity.title, entity.original_title, ...aliases.map((a) => a.alias)]) {
+    const v = (k ?? "").trim();
+    const norm = v.toLowerCase();
+    if (!v || seen.has(norm)) continue;
+    seen.add(norm);
+    keywords.push(v);
+    if (keywords.length >= SEARCH_MAX_KEYWORDS) break;
+  }
+  const params = (sub.params ?? {}) as { siteIds?: string[]; searchIntervalHours?: number };
+  let sq = db.selectFrom("site").selectAll().where("enabled", "=", true).where("health_status", "!=", "failing");
+  if (params.siteIds?.length) sq = sq.where("id", "in", params.siteIds);
+  const sites = await sq.orderBy("name").execute();
+  const errors: SubscriptionSearchResult["errors"] = [];
+  let searches = 0;
+  for (const site of sites) {
+    for (const kw of keywords) {
+      searches++;
+      const r = await liveSearch(rt, site, kw).catch((e) => ({ error: (e as Error).message, coalesced: false }));
+      if (r.error) errors.push({ siteId: site.id, keyword: kw, error: r.error });
+    }
+  }
+  const hours = params.searchIntervalHours ?? DEFAULT_SEARCH_INTERVAL_HOURS;
+  const now = rt.now();
+  const next = hours > 0 ? new Date(now.getTime() + hours * 3600_000) : null;
+  await db
+    .updateTable("subscription")
+    .set({ last_searched_at: now, next_search_at: next })
+    .where("id", "=", sub.id)
+    .execute();
+  // 已识别的旧结果:按订阅重评(新结果走 release.observed → resolved → evaluate)
+  await rt.ctx.engine
+    .enqueue(
+      "evaluateAcquisition",
+      { subscriptionId: sub.id },
+      { idempotencyKey: `search-eval:${sub.id}:${now.getTime()}`, traceId: sub.id },
+    )
+    .catch(() => {});
+  return {
+    status: "ok",
+    keywords,
+    sites: sites.map((s) => s.id),
+    searches,
+    errors,
+    nextSearchAt: next ? next.toISOString() : null,
+  };
+}
+
 type P = Record<string, unknown>;
 
 export const sourcesOutboxRoutes: Record<string, OutboxRouteFn> = {

apps/server/test/e2e/e9-continuous-series.test.ts

@@ -0,0 +1,274 @@
+import {
+  createMockRssServer,
+  type MockFeedItem,
+  type MockRssServer,
+  TMDB_FIXTURES,
+  type TmdbFixture,
+} from "@umt/testkit";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { TORRENT_SIZE, torrentFor } from "./fixtures/modules.ts";
+import { completeDownload, createE2E, type E2E, pgAvailable, seedBase, waitFor } from "./harness.ts";
+
+// E9 连续更新剧集(W5-C,对照 MP 缺集 − note 与批量下载):全部经真实任务引擎 + outbox 中继。
+// 季共 3 集(mock TMDB),新作品策略按 once 订阅。E01 还在下载(Slot 忙)时 E02、E03 与 E02 的另一个版本先后到达:
+//   · E02/E03 与在途的 E01 单元不相交 → 各自得到计划并同时下载(Slot×单元预留),不再因 slot_busy 丢集;
+//   · E02 另一版本与在途 E02 单元冲突 → 延后(slot_deferral),计划结束后重评时已覆盖,不重复下载;
+//   · refreshMetadata(编排循环按周期入队)把 TMDB 季集数写进季单元 → 三集全部入库后 once 订阅完成并写历史。
+
+const hasPg = await pgAvailable();
+const BASE = "https://audiences.example";
+const EP = (n: number, grp = "x264-GRP") => `Three.Eps.S01E0${n}.1080p.WEB-DL.${grp}`;
+const ALT = EP(2, "x265-ALT");
+
+const SHOW: TmdbFixture = {
+  type: "tv",
+  id: 990001,
+  original_title: "Three Eps",
+  original_language: "en",
+  origin_country: ["US"],
+  date: "2026-09-01",
+  titles: { "zh-CN": "三集剧", "en-US": "Three Eps" },
+  popularity: 50,
+  alternative_titles: [],
+  external_ids: {},
+  seasons: [
+    {
+      season_number: 1,
+      air_date: "2026-09-01",
+      names: { "zh-CN": "第 1 季", "en-US": "Season 1" },
+      episodes: [1, 2, 3].map((i) => ({
+        episode_number: i,
+        air_date: `2026-09-0${i}`,
+        names: { "zh-CN": `第 ${i} 集`, "en-US": `Episode ${i}` },
+      })),
+    },
+  ],
+};
+
+const item = (id: number, title: string): MockFeedItem => ({
+  guid: `aud-${id}`,
+  title,
+  link: `${BASE}/details.php?id=${id}&hit=1`,
+  enclosureUrl: `${BASE}/download.php?id=${id}&passkey=E9PASSKEY`,
+  size: TORRENT_SIZE,
+  pubDate: new Date(),
+  categories: ["剧集"],
+});
+
+describe.skipIf(!hasPg)("E9 连续更新剧集:Slot 忙时到达的集全部入库", () => {
+  let e: E2E;
+  let rss: MockRssServer;
+  let feedId: string;
+  let entityId: string;
+  let slotId: string;
+  let subId: string;
+
+  let lastSlot = -1;
+  const fetchFeed = async () => {
+    const slotOf = () => Math.floor(Date.now() / 10_000);
+    if (slotOf() === lastSlot) await new Promise((r) => setTimeout(r, (lastSlot + 1) * 10_000 - Date.now() + 50));
+    lastSlot = slotOf();
+    const r = await e.api("POST", `/feed-sources/${feedId}/fetch`);
+    expect(r.status).toBe(200);
+  };
+  const dtOf = (title: string) =>
+    e.db.selectFrom("download_task").selectAll().where("torrent_hash", "=", torrentFor(title).hash).executeTakeFirst();
+  const importedFor = async (title: string) => {
+    const dt = await waitFor(async () => dtOf(title), 60_000, `${title} download_task`);
+    return waitFor(
+      async () =>
+        e.db
+          .selectFrom("organization_record")
+          .selectAll()
+          .where("download_task_id", "=", dt.id)
+          .where("status", "in", ["succeeded", "failed", "partial"])
+          .executeTakeFirst(),
+      90_000,
+      `${title} 整理记录结束`,
+    );
+  };
+
+  beforeAll(async () => {
+    e = await createE2E("e9", { tmdbFixtures: [...TMDB_FIXTURES, SHOW] });
+    const base = await seedBase(e, { baseUrl: `${BASE}/` });
+    rss = await createMockRssServer({ passkey: "E9PASSKEY", items: [item(930001, EP(1))] });
+    const urlRef = await e.booted.ctx.secrets.create({ purpose: "feed_url", plaintext: rss.feedUrl });
+    feedId = (
+      await e.db
+        .insertInto("feed_source")
+        .values({
+          site_id: base.siteId,
+          name: "观众-剧集",
+          purpose: "series",
+          url_secret_ref: urlRef,
+          url_display: rss.feedUrl.replace(/passkey=[^&]+/, "passkey=[REDACTED]"),
+          poll_interval_sec: 3600,
+        })
+        .returning("id")
+        .executeTakeFirstOrThrow()
+    ).id;
+    const created = await e.api("POST", "/policy-revisions", {
+      policyKey: "series",
+      kind: "acquisition",
+      title: "剧集 1080p",
+      scope: { mediaKinds: ["series"] },
+      rules: {
+        newTitle: {
+          enabled: true,
+          windowDays: 3650,
+          mediaKinds: ["series"],
+          slots: [{ name: "1080p", hardConditions: { resolutions: ["1080p"] } }],
+        },
+      },
+    });
+    expect(created.status).toBe(201);
+    const pub = await e.api("POST", `/policy-revisions/${created.body.data.id}/publish`, {
+      expectedRevision: created.body.data.revision,
+    });
+    expect(pub.status).toBe(200);
+  }, 120_000);
+
+  afterAll(async () => {
+    await rss?.close();
+    await e?.close();
+  }, 60_000);
+
+  it("E01 下载中;TMDB 季集数写入季单元(总集数 3)", async () => {
+    await fetchFeed();
+    const rel = await waitFor(
+      async () =>
+        e.db
+          .selectFrom("source_release")
+          .selectAll()
+          .where("title", "=", EP(1))
+          .where("resolution_status", "=", "resolved")
+          .executeTakeFirst(),
+      60_000,
+      "E01 识别完成",
+    );
+    entityId = rel.media_entity_id!;
+    const sub = await waitFor(
+      async () =>
+        e.db.selectFrom("subscription").selectAll().where("media_entity_id", "=", entityId).executeTakeFirst(),
+      30_000,
+      "新作品订阅",
+    );
+    subId = sub.id;
+    expect(sub).toMatchObject({ mode: "once", status: "active" });
+    slotId = (
+      await e.db
+        .selectFrom("acquisition_slot")
+        .select("id")
+        .where("media_entity_id", "=", entityId)
+        .executeTakeFirstOrThrow()
+    ).id;
+    await waitFor(async () => e.qb.get(torrentFor(EP(1)).hash), 60_000, "qB 收到 E01");
+    const season = await waitFor(
+      async () =>
+        e.db
+          .selectFrom("content_unit")
+          .select(["episode_total", "episode_total_source"])
+          .where("media_entity_id", "=", entityId)
+          .where("kind", "=", "season")
+          .where("episode_total", "is not", null)
+          .executeTakeFirst(),
+      60_000,
+      "季总集数刷新",
+    );
+    expect(season).toEqual({ episode_total: 3, episode_total_source: "tmdb" });
+  }, 180_000);
+
+  it("Slot 忙时 E02、E03 与 E02 另一版本到达:E02/E03 同时在途,另一版本延后", async () => {
+    rss.addItem(item(930002, EP(2)));
+    rss.addItem(item(930003, EP(3)));
+    rss.addItem(item(930004, ALT));
+    await fetchFeed();
+    // E02(任一版本)与 E03 都进了 qB;E01 仍未完成 → 三个在途计划并存
+    await waitFor(async () => e.qb.get(torrentFor(EP(3)).hash), 90_000, "qB 收到 E03");
+    const e02 = await waitFor(
+      async () => e.qb.get(torrentFor(EP(2)).hash) ?? e.qb.get(torrentFor(ALT).hash),
+      60_000,
+      "qB 收到 E02",
+    );
+    const inflight = await e.db
+      .selectFrom("acquisition_plan")
+      .select(["unit_keys", "status"])
+      .where("slot_id", "=", slotId)
+      .where("status", "in", ["reserved", "submitted"])
+      .execute();
+    expect(inflight.map((p) => p.unit_keys[0]).sort()).toEqual(["s1e1", "s1e2", "s1e3"]);
+    const slot = await e.db
+      .selectFrom("acquisition_slot")
+      .selectAll()
+      .where("id", "=", slotId)
+      .executeTakeFirstOrThrow();
+    expect(slot.status).toBe("downloading");
+    // 另一版本 E02:延后而不是丢弃
+    const loser = e02.name.includes("x265-ALT") ? EP(2) : ALT;
+    const loserRel = await waitFor(
+      async () =>
+        e.db
+          .selectFrom("source_release")
+          .select("id")
+          .where("title", "=", loser)
+          .where("resolution_status", "=", "resolved")
+          .executeTakeFirst(),
+      60_000,
+      "另一版本识别",
+    );
+    await waitFor(
+      async () =>
+        e.db
+          .selectFrom("slot_deferral")
+          .selectAll()
+          .where("slot_id", "=", slotId)
+          .where("source_release_id", "=", loserRel.id)
+          .executeTakeFirst(),
+      60_000,
+      "另一版本延后",
+    );
+  }, 180_000);
+
+  it("三集依次完成全部入库;once 订阅完成并写历史;延后的版本重评后不重复下载", async () => {
+    const e02Title = e.qb.get(torrentFor(EP(2)).hash) ? EP(2) : ALT;
+    for (const t of [EP(3), e02Title, EP(1)]) {
+      await completeDownload(e, t);
+      const rec = await importedFor(t);
+      expect(rec.status).toBe("succeeded");
+    }
+    const cov = await e.db
+      .selectFrom("asset_unit_coverage as c")
+      .innerJoin("content_unit as u", "u.id", "c.content_unit_id")
+      .innerJoin("media_asset as a", "a.id", "c.media_asset_id")
+      .select(["u.episode_number"])
+      .where("a.media_entity_id", "=", entityId)
+      .where("a.status", "=", "active")
+      .orderBy("u.episode_number")
+      .execute();
+    expect(cov.map((c) => c.episode_number)).toEqual([1, 2, 3]);
+    const sub = await waitFor(
+      async () => {
+        const s = await e.db.selectFrom("subscription").selectAll().where("id", "=", subId).executeTakeFirstOrThrow();
+        return s.status === "completed" ? s : null;
+      },
+      30_000,
+      "订阅完成",
+    );
+    expect(sub.status_reason).toBe("slot_satisfied");
+    const slot = await e.db
+      .selectFrom("acquisition_slot")
+      .selectAll()
+      .where("id", "=", slotId)
+      .executeTakeFirstOrThrow();
+    expect(slot.status).toBe("satisfied");
+    const hist = await e.api("GET", `/subscription-history?subscriptionId=${subId}`);
+    expect(hist.body.data.items).toHaveLength(1);
+    expect(hist.body.data.items[0]).toMatchObject({ reason: "slot_satisfied", mediaEntityId: entityId });
+    // 进度 API:三集已覆盖,总集数已知
+    const prog = await e.api("GET", `/subscriptions/${subId}/progress`);
+    expect(prog.body.data.seasons[0]).toMatchObject({ episodeTotal: 3, totalKnown: true, complete: true });
+    // 延后的另一版本:计划结束后重评,E02 已覆盖 → 不下载;三个种子各提交一次
+    await new Promise((r) => setTimeout(r, 1500));
+    expect(e.qb.callsTo("torrents/add")).toHaveLength(3);
+  }, 300_000);
+});

apps/server/test/sources-identity/mteam-tasks.test.ts

@@ -72,7 +72,13 @@ describe.skipIf(!up)("M-Team 快照与任务注册", () => {
   it("模块注册 4 个任务与 3 条 outbox 路由;任务处理器经 step.run 调服务", async () => {
     const reg = new TaskRegistry();
     h.module.registerTasks(reg, h.ctx);
-    expect(reg.names()).toEqual(["ingestFeed", "refreshMetadata", "resolveObservation", "snapshotSite"]);
+    expect(reg.names()).toEqual([
+      "ingestFeed",
+      "refreshMetadata",
+      "resolveObservation",
+      "searchSubscription",
+      "snapshotSite",
+    ]);
     expect(Object.keys(h.module.outboxRoutes!(h.ctx)).sort()).toEqual([
       "feed.due",
       "feed.gap_suspected",

apps/server/test/sources-identity/subscription-search.test.ts

@@ -0,0 +1,149 @@
+import { createMockNexusPhpSite, type MockNexusPhpSite } from "@umt/testkit";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { applySeasonTotals } from "../../src/modules/sources-identity/service/metadata-totals.ts";
+import { searchSubscription } from "../../src/modules/sources-identity/tasks.ts";
+import { createHarness, type Harness, json, pgAvailable } from "./harness.ts";
+
+// W5-C 订阅主动搜索(mock NexusPHP 站点:适配器 search + 限流 + 快照)与季总集数账本(refreshMetadata 的写入部分)。
+
+const up = await pgAvailable();
+
+describe.skipIf(!up)("订阅主动搜索与季总集数", () => {
+  let h: Harness;
+  let mock: MockNexusPhpSite;
+  let siteId: string;
+
+  beforeAll(async () => {
+    mock = await createMockNexusPhpSite();
+    h = await createHarness("subsearch");
+    const ok = await json(
+      await h.req("/sites", {
+        body: {
+          key: "audiences",
+          name: "观众",
+          adapter: "nexusphp",
+          baseUrl: `${mock.baseUrl}/`,
+          rateLimit: { minIntervalMs: 0, concurrency: 4 },
+        },
+      }),
+    );
+    expect(ok.status).toBe(200);
+    siteId = ok.body.data.id;
+    await h.req("/site-accounts", { body: { siteId, label: "主号", authKind: "cookie", credential: mock.cookie } });
+  });
+  afterAll(async () => {
+    await h?.close();
+    await mock?.close();
+  });
+
+  const entity = async (title: string, kind: "movie" | "series" = "movie") =>
+    (
+      await h.db
+        .insertInto("media_entity")
+        .values({ kind, title, original_title: title, year: 2021, status: "active" })
+        .returning("id")
+        .executeTakeFirstOrThrow()
+    ).id;
+  const subscribe = async (entityId: string, extra: Record<string, unknown> = {}) =>
+    (
+      await h.db
+        .insertInto("subscription")
+        .values({ media_entity_id: entityId, media_kind: "movie", mode: "once", ...extra })
+        .returning("id")
+        .executeTakeFirstOrThrow()
+    ).id;
+
+  it("按标题在已开启站点搜索:结果 upsert 为 source_release(新结果发 release.observed),排下次搜索并入队订阅重评", async () => {
+    const e = await entity("Dune");
+    const subId = await subscribe(e);
+    const searches = () => mock.requests.filter((r) => r.path.includes("search=")).length;
+    const before = searches();
+    const r = await searchSubscription(h.rt, subId);
+    expect(r).toMatchObject({ status: "ok", keywords: ["Dune"], sites: [siteId], searches: 1, errors: [] });
+    expect(searches() - before).toBe(1);
+    const rels = await h.db
+      .selectFrom("source_release")
+      .select(["id", "title"])
+      .where("site_id", "=", siteId)
+      .execute();
+    expect(rels.some((x) => /Dune/i.test(x.title))).toBe(true);
+    expect((await h.outbox("release.observed")).length).toBeGreaterThanOrEqual(1);
+    const snap = await h.db
+      .selectFrom("site_snapshot")
+      .selectAll()
+      .where("query", "=", "Dune")
+      .executeTakeFirstOrThrow();
+    expect(snap.kind).toBe("live_search");
+    const sub = await h.db.selectFrom("subscription").selectAll().where("id", "=", subId).executeTakeFirstOrThrow();
+    expect(sub.last_searched_at).not.toBeNull();
+    expect(sub.next_search_at!.getTime() - sub.last_searched_at!.getTime()).toBe(24 * 3600_000);
+    expect(h.engine.enqueued.at(-1)).toMatchObject({ task: "evaluateAcquisition", input: { subscriptionId: subId } });
+  });
+
+  it("站点范围外不搜;暂停订阅只允许手动;searchIntervalHours=0 关闭周期", async () => {
+    const e = await entity("Bear");
+    const scoped = await subscribe(e, {
+      params: JSON.stringify({ siteIds: ["00000000-0000-4000-8000-000000000000"] }),
+    });
+    expect(await searchSubscription(h.rt, scoped)).toMatchObject({ status: "ok", sites: [], searches: 0 });
+    const paused = await subscribe(e, { status: "paused", params: JSON.stringify({ searchIntervalHours: 0 }) });
+    expect(await searchSubscription(h.rt, paused)).toMatchObject({ status: "skipped", reason: "subscription_paused" });
+    const manual = await searchSubscription(h.rt, paused, { manual: true });
+    expect(manual).toMatchObject({ status: "ok", searches: 1, nextSearchAt: null });
+  });
+
+  it("季总集数:TMDB 集数写季单元(S00 不计、未公布保持未知);回退接受但被已覆盖最高集兜底;锁定不改;pending 订阅转 active", async () => {
+    const e = await entity("Series T", "series");
+    const sub = await subscribe(e, { media_kind: "series", status: "pending", status_reason: "metadata_pending" });
+    const now = new Date();
+    const r1 = await h.db.transaction().execute((trx) =>
+      applySeasonTotals(
+        trx,
+        e,
+        [
+          { seasonNumber: 0, episodeCount: 3, name: "特别篇", airDate: null },
+          { seasonNumber: 1, episodeCount: 8, name: "第 1 季", airDate: null },
+          { seasonNumber: 2, episodeCount: 0, name: "第 2 季", airDate: null },
+        ],
+        now,
+      ),
+    );
+    expect(r1.seasons).toEqual([
+      { season: 1, total: 8, known: true, changed: true },
+      { season: 2, total: null, known: false, changed: false },
+    ]);
+    expect(r1.activated).toBe(1);
+    expect(
+      (await h.db.selectFrom("subscription").select("status").where("id", "=", sub).executeTakeFirstOrThrow()).status,
+    ).toBe("active");
+    const r2 = await h.db
+      .transaction()
+      .execute((trx) =>
+        applySeasonTotals(trx, e, [{ seasonNumber: 1, episodeCount: 12, name: null, airDate: null }], now),
+      );
+    expect(r2.seasons[0]).toMatchObject({ total: 12, changed: true });
+    await h.db
+      .updateTable("content_unit")
+      .set({ episode_total_locked: true })
+      .where("media_entity_id", "=", e)
+      .where("season_number", "=", 1)
+      .execute();
+    const r3 = await h.db
+      .transaction()
+      .execute((trx) =>
+        applySeasonTotals(trx, e, [{ seasonNumber: 1, episodeCount: 20, name: null, airDate: null }], now),
+      );
+    expect(r3.seasons[0]).toMatchObject({ total: 12, changed: false });
+    const units = await h.db
+      .selectFrom("content_unit")
+      .select(["kind", "season_number", "episode_total", "episode_total_source"])
+      .where("media_entity_id", "=", e)
+      .orderBy("season_number")
+      .execute();
+    // 只建季单元,不预建集单元
+    expect(units).toEqual([
+      { kind: "season", season_number: 1, episode_total: 12, episode_total_source: "tmdb" },
+      { kind: "season", season_number: 2, episode_total: null, episode_total_source: null },
+    ]);
+  });
+});