merge: fix/c-acquisition

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

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

@@ -34,6 +34,7 @@ import { hashJson, jsonb, startOfDay, toApi } from "../util.ts";
 import {
   type AoDeps,
   automationBlock,
+  CANDIDATE_ARTIFACT_TITLE,
   downloaderPathOf,
   type Exec,
   enqueueTracked,
@@ -47,9 +48,13 @@ import { autoFillCheck, type ForceFill } from "./evaluate-autofill.ts";
 import { entityOriginalLanguage, loadRuleCatalog, mergeHard, type RuleCatalog, resolveRuleChain } from "./rules.ts";
 import {
   clearDeferral,
+  clearReleaseDeferrals,
+  clearSubscriptionDeferrals,
+  gateRetryDelayMs,
   rangeOf,
   recordDeferral,
   scheduleSlotReevaluate,
+  scheduleSlotRetry,
   seasonLedgers,
   slotCoveredUnits,
   slotInflightPlans,
@@ -245,6 +250,11 @@ export async function createSlotsFromTemplates(
   }
 }
 
+/**
+ * 预算用量。并发 = 在跑的下载任务(不含 stalled:无进展任务不占名额,否则卡住几个就把追更全挡住)
+ * + 已预留但还没建下载任务的计划(刚预留、等补投;防止一轮评估连续预留超过并发上限)。
+ * 已建下载任务的预留计划按任务计,不重复算。
+ */
 async function budgetUsage(exec: Exec, now: Date) {
   const day = startOfDay(now);
   const r = await sql<{ new_titles: number; bytes: number; reserved: number; active: number; reserved_count: number }>`
@@ -253,8 +263,11 @@ async function budgetUsage(exec: Exec, now: Date) {
          and evaluation->>'mode' = 'new_title')::int as new_titles,
       (select coalesce(sum(reserved_bytes),0) from acquisition_plan where created_at >= ${day} and status in ('submitted','completed'))::bigint as bytes,
       (select coalesce(sum(reserved_bytes),0) from acquisition_plan where created_at >= ${day} and status = 'reserved')::bigint as reserved,
-      (select count(*) from download_task where managed and removed_at is null and status in ('queued','submitting','downloading','stalled'))::int as active,
+      (select count(*) from download_task where managed and removed_at is null and status in ('queued','submitting','downloading'))::int as active,
-      (select count(*) from acquisition_plan where status = 'reserved')::int as reserved_count`.execute(exec);
+      (select count(*) from acquisition_plan p where p.status = 'reserved'
+         and not exists (select 1 from download_task dt where dt.acquisition_plan_id = p.id and dt.removed_at is null))::int as reserved_count`.execute(
+    exec,
+  );
   const row = r.rows[0]!;
   return {
     newTitlesToday: row.new_titles,
@@ -360,11 +373,22 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon
   const { d, now } = c;
   const db = d.ctx.db;
   const skip = (reason: string) => c.out.skipped.push({ sourceReleaseId: rel.id, reason });
-  if (!rel.media_entity_id || rel.resolution_status !== "resolved") return skip("release_unresolved");
+  // 不再参与评估的资源:延后行一并清掉,免得到期空转(L4)
-  if (rel.availability !== "active") return skip(`release_${rel.availability}`);
+  const drop = async (reason: string) => {
+    await clearReleaseDeferrals(db, rel.id);
+    return skip(reason);
+  };
+  if (!rel.media_entity_id || rel.resolution_status !== "resolved") return drop("release_unresolved");
+  if (rel.availability !== "active") return drop(`release_${rel.availability}`);
   const entity = await resolveEntity(db, rel.media_entity_id);
-  if (!entity || entity.status === "archived") return skip("entity_unavailable");
+  if (!entity || entity.status === "archived") return drop("entity_unavailable");
-  if (entity.kind !== "movie" && episodeUnresolved(rel)) return skip("episode_unresolved");
+  if (entity.kind !== "movie" && episodeUnresolved(rel)) {
+    // 订阅侧可见:相关收录版本各记一条候选评估;确认后重发 release.resolved 正常评估时按 (收录版本, 资源) 覆盖
+    await noteEpisodeUnresolved(c, rel, entity.id, onlySlotId);
+    return drop("episode_unresolved");
+  }
+  // 已确认的资源:先清掉之前的"集数待确认"记录(换算到别的季时下面可能不再进入该收录版本,不会被覆盖)
+  if (entity.kind !== "movie" && episodeResolutionOf(rel)) await clearEpisodeUnresolvedNotes(c, rel.id, entity.id);
   const parsed = parsedOf(rel);
   const units = await releaseUnits(
     db,
@@ -463,6 +487,7 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon
         .set({ status: "completed", status_reason: "stop_after_reached" })
         .where("id", "=", s.id)
         .execute();
+      await clearSubscriptionDeferrals(db, s.id);
       s.status = "completed";
     }
   }
@@ -477,7 +502,11 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon
   for (const slot of slots) {
     if (onlySlotId && slot.id !== onlySlotId) continue;
     const sub = slot.subscription_id ? activeSubs.get(slot.subscription_id) : undefined;
-    if (slot.subscription_id && !sub) continue; // 订阅已暂停/完成/取消
+    if (slot.subscription_id && !sub) {
+      // 订阅已暂停/完成/取消:不评估,也不留延后行
+      await clearDeferral(db, slot.id, rel.id);
+      continue;
+    }
     if (slot.content_unit_id && !units.includes(slot.content_unit_id)) {
       // Slot 限定某季:季单元是否是本发布单元的父级
       const seasonOk = await db
@@ -523,6 +552,37 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon
   }
 }
 
+/** 集数待确认:给仍在追踪的收录版本各写一条 skipped 候选评估(原因 episode_unresolved) */
+async function noteEpisodeUnresolved(c: Ctx, rel: SourceReleaseRow, entityId: string, onlySlotId?: string) {
+  let q = c.d.ctx.db
+    .selectFrom("acquisition_slot as s")
+    .leftJoin("subscription as sub", "sub.id", "s.subscription_id")
+    .select("s.id")
+    .where("s.media_entity_id", "=", entityId)
+    .where((eb) => eb.or([eb("s.subscription_id", "is", null), eb("sub.status", "in", [...TRACKING_SUB_STATUSES])]));
+  if (onlySlotId) q = q.where("s.id", "=", onlySlotId);
+  for (const s of await q.execute())
+    await note(c, {
+      slotId: s.id,
+      sourceReleaseId: rel.id,
+      title: rel.title,
+      planId: null,
+      decision: "skipped",
+      mode: null,
+      rejected: [{ code: "episode_unresolved", message: "集数待确认，确认后自动重新评估" }],
+      reasons: ["episode_unresolved"],
+    });
+}
+
+async function clearEpisodeUnresolvedNotes(c: Ctx, sourceReleaseId: string, entityId: string) {
+  await sql`
+    delete from artifact
+     where title = ${CANDIDATE_ARTIFACT_TITLE} and subject_type = 'acquisition_slot'
+       and subject_id in (select id from acquisition_slot where media_entity_id = ${entityId})
+       and content->>'sourceReleaseId' = ${sourceReleaseId}
+       and content->'reasons' ? 'episode_unresolved'`.execute(c.d.ctx.db);
+}
+
 /** 站点字段 → 评估上下文(做种数、促销下载系数、副标题/标签、站点 id、发布后分钟数) */
 export function contextOf(rel: SourceReleaseRow, now: Date = new Date()): EvaluationContext {
   const promo = (rel.promotion ?? {}) as { kind?: string; raw?: { downloadFactor?: number | null } };
@@ -607,6 +667,10 @@ async function evaluateForSlot(c: Ctx, p: SlotInput): Promise<{ planId: string |
   }
   const inflight = await slotInflightPlans(db, slot.id);
   const inflightKeys = inflight.flatMap((x) => x.unitKeys);
+  const deferWithRetry = async (slotId: string, releaseId: string, reason: string, unitKeys: string[]) => {
+    const attempts = await recordDeferral(db, { slotId, sourceReleaseId: releaseId, reason, unitKeys });
+    await scheduleSlotRetry(db, slotId, new Date(now.getTime() + gateRetryDelayMs(attempts)));
+  };
   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, {
@@ -785,9 +849,21 @@ async function evaluateForSlot(c: Ctx, p: SlotInput): Promise<{ planId: string |
   planUnits = fill.planUnits;
   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 });
+    // 被挡住不是"不要":写延后并按次数退避排重评(电影 / 已满足的剧集收录版本没有周期重评,全靠这里)
+    const reason = `gate:${gate.kind}`;
+    if (!c.draft) await deferWithRetry(slot.id, rel.id, reason, planUnits);
+    await note(c, {
+      ...base,
+      decision: "deferred",
+      mode,
+      rejected: gate.rejected,
+      reasons: [`deferred:${reason}`, ...gate.reasons],
+      ...evalExtra,
+      unitKeys: planUnits,
+    });
     c.out.skipped.push({ sourceReleaseId: rel.id, reason: gate.reasons[0] ?? "gated" });
-    return done;
+    if (!c.draft) c.out.deferred.push({ sourceReleaseId: rel.id, slotId: slot.id, reason });
+    return { planId: null, deferred: !c.draft };
   }
   const r = await reserve(c, {
     slot,
@@ -803,6 +879,27 @@ async function evaluateForSlot(c: Ctx, p: SlotInput): Promise<{ planId: string |
     ...(fill.action === "fill" ? { priority: fill.priority, evaluationExtra: fill.evaluation } : {}),
   });
   if (r.conflict) return defer("units_inflight", r.conflict, mode);
+  if (r.retryAt) {
+    // 同一资源上次计划失败,冷却中:延后到冷却结束再试
+    await recordDeferral(db, {
+      slotId: slot.id,
+      sourceReleaseId: rel.id,
+      reason: "retry_cooldown",
+      unitKeys: planUnits,
+    });
+    await scheduleSlotRetry(db, slot.id, r.retryAt);
+    await note(c, {
+      ...base,
+      decision: "deferred",
+      mode,
+      rejected: [],
+      reasons: ["deferred:retry_cooldown", `retry_at:${r.retryAt.toISOString()}`],
+      ...evalExtra,
+      unitKeys: planUnits,
+    });
+    c.out.deferred.push({ sourceReleaseId: rel.id, slotId: slot.id, reason: "retry_cooldown" });
+    return { planId: null, deferred: true };
+  }
   const planId = r.planId;
   if (!planId) {
     await note(c, { ...base, decision: "skipped", mode, rejected: [], reasons: ["slot_changed_or_reserved"] });
@@ -905,8 +1002,11 @@ interface GateOk {
   volumeKey: string;
   savePath: string;
 }
+/** 闸门类型:写进延后原因(gate:automation / gate:downloader / gate:budget / gate:cache) */
+type GateKind = "automation" | "downloader" | "budget" | "cache";
 interface GateNo {
   ok: false;
+  kind: GateKind;
   reasons: string[];
   rejected: { code: string; message: string }[];
 }
@@ -920,13 +1020,13 @@ async function gates(
 ): Promise<GateOk | GateNo> {
   const { d } = c;
   const blocked = await automationBlock(d);
-  if (blocked) return { ok: false, reasons: [blocked], rejected: [] };
+  if (blocked) return { ok: false, kind: "automation", reasons: [blocked], rejected: [] };
   const sp = (sub?.params ?? {}) as { downloaderInstanceId?: string; downloadSavePath?: string };
   const inst = await pickInstance(d.ctx.db, await loadInstances(d.ctx.db), {
     explicitId: sp.downloaderInstanceId,
     siteId: rel.site_id,
   });
-  if (!inst?.savePath) return { ok: false, reasons: ["no_downloader_instance"], rejected: [] };
+  if (!inst?.savePath) return { ok: false, kind: "downloader", reasons: ["no_downloader_instance"], rejected: [] };
   // 订阅指定的保存路径:本机路径经下载器路径映射换成下载器侧路径;本来就是下载器侧路径的原样用
   let savePath = inst.savePath;
   if (sp.downloadSavePath) {
@@ -934,6 +1034,7 @@ async function gates(
     if (!mapped)
       return {
         ok: false,
+        kind: "downloader",
         reasons: ["download_save_path_unmapped"],
         rejected: [
           {
@@ -958,6 +1059,7 @@ async function gates(
   if (b.decision !== "allow") {
     return {
       ok: false,
+      kind: "budget",
       reasons: [b.decision === "unknown" ? "budget_unknown_size" : "budget_exceeded"],
       rejected: b.violations.map((v) => ({
         code: `budget.${v.kind}`,
@@ -969,11 +1071,12 @@ async function gates(
   const vol = (await loadVolumes(d, d.ctx.db, await loadInstances(d.ctx.db))).find(
     (v) => v.volumeKey === inst.volumeKey,
   );
-  if (!vol) return { ok: false, reasons: ["volume_state_unknown"], rejected: [] };
+  if (!vol) return { ok: false, kind: "cache", reasons: ["volume_state_unknown"], rejected: [] };
   const dv = decideVolume(vol, cache, rel.size_bytes !== null ? { bytes: rel.size_bytes } : {});
   if (dv.decision !== "allow") {
     return {
       ok: false,
+      kind: "cache",
       reasons: dv.reasons.length ? dv.reasons.map((r) => `cache:${r}`) : ["cache:blocked"],
       rejected: [
         {
@@ -998,6 +1101,16 @@ export function subscriptionSavePath(t: Pick<InstanceTopology, "inst" | "mapping
   return hostPathOf(t, p) ? p : null;
 }
 
+/**
+ * 旧计划是否让位给新计划(T1-3c):失败 / 取消 / 过期可以重建(站点 Cookie 修好、下载器恢复、集数映射改号后重评),
+ * 但用户明确不要的(在下载器里移除、手动取消)和整理阶段失败的(文件已下好,重下同一种子无意义,交给整理重试)仍挡住。
+ */
+function planRetryable(p: { status: string; status_reason: string | null }): boolean {
+  if (!["failed", "cancelled", "expired"].includes(p.status)) return false;
+  const r = p.status_reason ?? "";
+  return r !== "download_removed" && r !== "cancelled_by_user" && !r.startsWith("organize:");
+}
+
 async function reserve(
   c: Ctx,
   p: {
@@ -1017,7 +1130,7 @@ async function reserve(
     priority?: AcquisitionPriority;
     evaluationExtra?: Record<string, unknown>;
   },
-): Promise<{ planId: string | null; conflict?: string[] }> {
+): Promise<{ planId: string | null; conflict?: string[]; retryAt?: Date }> {
   const { d, now } = c;
   const immediate = p.sub?.priority === "immediate";
   const priority = immediate ? PRIORITY.immediate : (p.priority ?? PRIORITY[priorityOf(p.mode, immediate)]);
@@ -1109,6 +1222,35 @@ async function reserve(
     } else if (slot.revision !== p.slot.revision || busy) {
       return { planId: null };
     }
+    // 幂等键只挡在途 / 已完成的计划;失败 / 取消 / 过期的旧计划改名让位(Slot 行锁内,同 Slot 的预留串行;
+    // 改名与插入在同一事务,唯一约束仍兜底)
+    const key = `acq:${slot.id}:${p.rel.id}`;
+    const prev = await trx
+      .selectFrom("acquisition_plan")
+      .select(["id", "status", "status_reason", "updated_at"])
+      .where("idempotency_key", "=", key)
+      .forUpdate()
+      .executeTakeFirst();
+    if (prev) {
+      if (!planRetryable(prev)) return { planId: null };
+      if (prev.status === "failed") {
+        // 失败的冷却一段再重试(按该资源已失败次数翻倍),免得持续失败时每轮都建计划、开待处理
+        const retired = await trx
+          .selectFrom("acquisition_plan")
+          .select((eb) => eb.fn.countAll<number>().as("n"))
+          .where("slot_id", "=", slot.id)
+          .where("source_release_id", "=", p.rel.id)
+          .where("idempotency_key", "like", `${key}:retired:%`)
+          .executeTakeFirstOrThrow();
+        const retryAt = new Date(prev.updated_at.getTime() + gateRetryDelayMs(Number(retired.n) + 1));
+        if (retryAt > now) return { planId: null, retryAt };
+      }
+      await trx
+        .updateTable("acquisition_plan")
+        .set({ idempotency_key: `${key}:retired:${prev.id}` })
+        .where("id", "=", prev.id)
+        .execute();
+    }
     let st: SlotState = { status: slot.status, hasAsset: slot.current_asset_id !== null };
     if (!busy) {
       if (st.status === "satisfied") st = transition(st, { type: "upgrade_found" });
@@ -1131,7 +1273,7 @@ async function reserve(
         }),
         payload_hash: hashJson(payload),
         expected_revision: slot.revision + 1,
-        idempotency_key: `acq:${slot.id}:${p.rel.id}`,
+        idempotency_key: key,
         unit_keys: p.unitKeys,
       })
       .onConflict((oc) => oc.column("idempotency_key").doNothing())

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

@@ -577,6 +577,7 @@ export async function completeSubscription(
     .returningAll()
     .executeTakeFirst();
   if (!done) return false;
+  await clearSubscriptionDeferrals(trx, done.id);
   await trx
     .insertInto("subscription_history")
     .values({
@@ -597,16 +598,56 @@ export async function completeSubscription(
 
 // ── 延后与重评调度 ──
 
+/** 写 / 更新延后行,返回累计次数(首次 1) */
 export async function recordDeferral(
   exec: Exec,
   input: { slotId: string; sourceReleaseId: string; reason: string; unitKeys: readonly string[] },
-): Promise<void> {
+): Promise<number> {
-  await sql`
+  const r = await sql<{ attempts: number }>`
     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);
+          updated_at = now()
+    returning attempts`.execute(exec);
+  return r.rows[0]?.attempts ?? 1;
+}
+
+/** 某资源的全部延后行(资源过期 / 不可评估时不再留着空转) */
+export async function clearReleaseDeferrals(exec: Exec, sourceReleaseId: string): Promise<void> {
+  await exec.deleteFrom("slot_deferral").where("source_release_id", "=", sourceReleaseId).execute();
+}
+
+/** 某订阅下收录版本的全部延后行(订阅完成 / 暂停 / 取消后不再重评) */
+export async function clearSubscriptionDeferrals(exec: Exec, subscriptionId: string): Promise<void> {
+  await exec
+    .deleteFrom("slot_deferral")
+    .where("slot_id", "in", (eb) =>
+      eb.selectFrom("acquisition_slot").select("id").where("subscription_id", "=", subscriptionId),
+    )
+    .execute();
+}
+
+/**
+ * 被闸门挡住(自动化开关 / 下载器 / 预算 / 缓存卷)或失败计划冷却中的延后:按次数退避,
+ * 30 分钟起翻倍,封顶 6 小时。不等事件触发(预算释放、缓存卷恢复没有统一事件),到点由 sweepDueSlots 重评。
+ */
+export const GATE_RETRY_BASE_MS = 30 * 60_000;
+export const GATE_RETRY_MAX_MS = 6 * 3600_000;
+export const gateRetryDelayMs = (attempts: number): number =>
+  Math.min(GATE_RETRY_BASE_MS * 2 ** Math.max(0, Math.min(attempts, 16) - 1), GATE_RETRY_MAX_MS);
+
+/**
+ * 最晚在 at 重评该收录版本(已排得更早的不推迟);电影收录版本与已满足的剧集收录版本也适用
+ * (scheduleSlotReevaluate 只给未满足的剧集排周期)。
+ */
+export async function scheduleSlotRetry(exec: Exec, slotId: string, at: Date): Promise<void> {
+  await exec
+    .updateTable("acquisition_slot")
+    .set({ next_evaluate_at: at })
+    .where("id", "=", slotId)
+    .where((eb) => eb.or([eb("next_evaluate_at", "is", null), eb("next_evaluate_at", ">", at)]))
+    .execute();
 }
 
 export async function clearDeferral(exec: Exec, slotId: string, sourceReleaseId: string): Promise<void> {

apps/server/test/acquisition-organize/blocked-reevaluate.test.ts

@@ -0,0 +1,354 @@
+import { createMockQb, type MockQb } from "@umt/testkit";
+import { sql } from "kysely";
+import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
+import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts";
+import {
+  completeSubscription,
+  GATE_RETRY_BASE_MS,
+  sweepDueSlots,
+} from "../../src/modules/acquisition-organize/services/subscriptions.ts";
+import { createHarness, fakeDisk, type Harness, pgAvailable, seedBase, seedEntity, seedRelease } from "./harness.ts";
+
+// 获取环节被挡住 / 失败后的重评(T1-3):闸门不过写延后并排重评(电影、已满足剧集收录版本);
+// 并发预算不计 stalled;失败 / 取消的旧计划不挡同一资源重建计划;订阅完成 / 资源过期清延后行。
+
+const hasPg = await pgAvailable();
+const SIZE = 300_000_000;
+
+describe.skipIf(!hasPg)("被挡住与失败后的重评", () => {
+  let h: Harness;
+  let qb: MockQb;
+  let base: Awaited<ReturnType<typeof seedBase>>;
+  const disk = fakeDisk(10_000_000_000);
+
+  beforeAll(async () => {
+    qb = await createMockQb({ addVisibleAfterPolls: 0 });
+    disk.setPct(10);
+    h = await createHarness("blocked_reevaluate", { diskUsage: disk.fn });
+    base = await seedBase(h.db, { libraryPath: "/tmp/umt-none-lib", cachePath: "/tmp/umt-none-cache", qbUrl: qb.url });
+  }, 60_000);
+
+  afterAll(async () => {
+    await h?.close();
+    await qb?.close();
+  });
+
+  const db = () => h.db;
+  // 用例之间不共享在途:上一个用例留下的预留计划会占并发预算
+  beforeEach(async () => {
+    await db().updateTable("acquisition_plan").set({ status: "cancelled" }).where("status", "=", "reserved").execute();
+  });
+  const hash = () => Math.random().toString(16).slice(2).padEnd(40, "0").slice(0, 40);
+
+  async function movieSlot(title: string) {
+    const entityId = await seedEntity(db(), { kind: "movie", title, year: 2024 });
+    const sub = await db()
+      .insertInto("subscription")
+      .values({ media_entity_id: entityId, media_kind: "movie", mode: "once", 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({ resolutions: ["1080p"] }),
+        upgrade_threshold: JSON.stringify({}),
+      })
+      .returning("id")
+      .executeTakeFirstOrThrow();
+    return { entityId, subId: sub.id, slotId: slot.id };
+  }
+
+  async function seriesSlot(title: string, status: "empty" | "satisfied" = "empty") {
+    const entityId = await seedEntity(db(), { kind: "series", title, year: 2024 });
+    const sub = await db()
+      .insertInto("subscription")
+      .values({ media_entity_id: entityId, media_kind: "series", mode: "continuous", status: "active" })
+      .returning("id")
+      .executeTakeFirstOrThrow();
+    const slot = await db()
+      .insertInto("acquisition_slot")
+      .values({
+        media_entity_id: entityId,
+        subscription_id: sub.id,
+        name: "1080p",
+        status,
+        hard_conditions: JSON.stringify({ resolutions: ["1080p"] }),
+        upgrade_threshold: JSON.stringify({}),
+      })
+      .returning("id")
+      .executeTakeFirstOrThrow();
+    return { entityId, subId: sub.id, slotId: slot.id };
+  }
+
+  const release = (entityId: string, title: string) =>
+    seedRelease(db(), { siteId: base.siteId, entityId, title, size: SIZE, hash: hash() });
+  const deferrals = (slotId: string) =>
+    db().selectFrom("slot_deferral").selectAll().where("slot_id", "=", slotId).execute();
+  const slotRow = (slotId: string) =>
+    db().selectFrom("acquisition_slot").selectAll().where("id", "=", slotId).executeTakeFirstOrThrow();
+  const candidate = async (slotId: string, releaseId: string) =>
+    (
+      await sql<{ content: { decision: string; reasons: string[] } }>`
+        select content from artifact where title = 'candidate_evaluation' and subject_id = ${slotId}
+           and content->>'sourceReleaseId' = ${releaseId}`.execute(db())
+    ).rows[0]?.content;
+  /** 模拟计划结束后收录版本释放(applySlotEvent release 的效果) */
+  const releaseSlot = (slotId: string) =>
+    db()
+      .updateTable("acquisition_slot")
+      .set({ status: "empty", reserved_plan_id: null })
+      .where("id", "=", slotId)
+      .execute();
+
+  /** 缓存卷超阈值挡住 → 延后 + 排重评;恢复后到期重评能建计划 */
+  async function gateThenRecover(slotId: string, releaseId: string) {
+    disk.setPct(95);
+    const t0 = Date.now();
+    const ev = await evaluateAcquisition(h.d, { sourceReleaseId: releaseId });
+    expect(ev.plans).toHaveLength(0);
+    expect(ev.skipped[0]?.reason).toMatch(/^cache:/);
+    expect(ev.deferred).toContainEqual({ sourceReleaseId: releaseId, slotId, reason: "gate:cache" });
+    const rows = await deferrals(slotId);
+    expect(rows.map((r) => [r.source_release_id, r.reason])).toEqual([[releaseId, "gate:cache"]]);
+    expect((await candidate(slotId, releaseId))?.decision).toBe("deferred");
+    const due = (await slotRow(slotId)).next_evaluate_at;
+    expect(due).not.toBeNull();
+    const delay = due!.getTime() - t0;
+    expect(delay).toBeGreaterThan(GATE_RETRY_BASE_MS - 60_000);
+    expect(delay).toBeLessThanOrEqual(GATE_RETRY_BASE_MS + 60_000);
+
+    // 恢复后到期:巡检入队按收录版本重评,先评延后的资源
+    disk.setPct(10);
+    await db()
+      .updateTable("acquisition_slot")
+      .set({ next_evaluate_at: new Date(Date.now() - 1000) })
+      .where("id", "=", slotId)
+      .execute();
+    h.engine.enqueued.length = 0;
+    expect(await sweepDueSlots(h.d)).toBeGreaterThanOrEqual(1);
+    const job = h.engine.enqueued.find(
+      (e) =>
+        e.task === "evaluateAcquisition" && (e.input as { acquisitionSlotId?: string }).acquisitionSlotId === slotId,
+    );
+    expect(job).toBeDefined();
+    const ev2 = await evaluateAcquisition(h.d, job!.input as { acquisitionSlotId: string });
+    expect(ev2.plans, JSON.stringify(ev2)).toHaveLength(1);
+    expect(await deferrals(slotId)).toHaveLength(0);
+    expect((await candidate(slotId, releaseId))?.decision).toBe("accept");
+  }
+
+  it("电影收录版本:缓存卷超阈值写延后 gate:cache 并排重评,恢复后重评建计划", async () => {
+    const { entityId, slotId } = await movieSlot("Gate Movie");
+    const rel = await release(entityId, "Gate.Movie.2024.1080p.WEB-DL.x264-GRP");
+    await gateThenRecover(slotId, rel);
+  });
+
+  it("已满足的剧集收录版本:新集被挡住同样写延后并重评", async () => {
+    const { entityId, slotId } = await seriesSlot("Gate Series", "satisfied");
+    const rel = await release(entityId, "Gate.Series.S01E05.1080p.WEB-DL.x264-GRP");
+    await gateThenRecover(slotId, rel);
+  });
+
+  it("再次被挡住按次数退避(第 2 次 60 分钟)", async () => {
+    const { entityId, slotId } = await movieSlot("Backoff Movie");
+    const rel = await release(entityId, "Backoff.Movie.2024.1080p.WEB-DL.x264-GRP");
+    await h.setSettings("automation", { enabled: false });
+    try {
+      await evaluateAcquisition(h.d, { sourceReleaseId: rel });
+      await db().updateTable("acquisition_slot").set({ next_evaluate_at: null }).where("id", "=", slotId).execute();
+      const t0 = Date.now();
+      await evaluateAcquisition(h.d, { sourceReleaseId: rel });
+      const [row] = await deferrals(slotId);
+      expect(row).toMatchObject({ reason: "gate:automation", attempts: 2 });
+      const delay = (await slotRow(slotId)).next_evaluate_at!.getTime() - t0;
+      expect(delay).toBeGreaterThan(2 * GATE_RETRY_BASE_MS - 60_000);
+      expect(delay).toBeLessThanOrEqual(2 * GATE_RETRY_BASE_MS + 60_000);
+    } finally {
+      await h.setSettings("automation", { enabled: true });
+    }
+  });
+
+  it("预算并发:stalled 任务不占并发名额,在跑的任务占", async () => {
+    await h.setSettings("budget", { maxConcurrentDownloads: 1 });
+    try {
+      const busyTask = await db()
+        .insertInto("download_task")
+        .values({
+          instance_id: base.instanceId,
+          torrent_hash: hash(),
+          name: "stalled-one",
+          size_bytes: 1_000_000,
+          status: "stalled",
+          managed: true,
+        })
+        .returning("id")
+        .executeTakeFirstOrThrow();
+      const a = await movieSlot("Budget Movie A");
+      const relA = await release(a.entityId, "Budget.Movie.A.2024.1080p.WEB-DL.x264-GRP");
+      const evA = await evaluateAcquisition(h.d, { sourceReleaseId: relA });
+      expect(evA.plans, JSON.stringify(evA)).toHaveLength(1);
+      // 刚预留、还没建下载任务的计划算一个名额
+      const b = await movieSlot("Budget Movie B");
+      const relB = await release(b.entityId, "Budget.Movie.B.2024.1080p.WEB-DL.x264-GRP");
+      const evB = await evaluateAcquisition(h.d, { sourceReleaseId: relB });
+      expect(evB.plans).toHaveLength(0);
+      expect(evB.skipped[0]?.reason).toBe("budget_exceeded");
+      expect((await deferrals(b.slotId))[0]?.reason).toBe("gate:budget");
+      // A 结束、stalled 任务变成下载中:仍占满
+      await db().updateTable("acquisition_plan").set({ status: "cancelled" }).where("id", "=", evA.plans[0]!).execute();
+      await db().updateTable("download_task").set({ status: "downloading" }).where("id", "=", busyTask.id).execute();
+      expect((await evaluateAcquisition(h.d, { sourceReleaseId: relB })).plans).toHaveLength(0);
+      await db().updateTable("download_task").set({ status: "stalled" }).where("id", "=", busyTask.id).execute();
+      const evB2 = await evaluateAcquisition(h.d, { sourceReleaseId: relB });
+      expect(evB2.plans).toHaveLength(1);
+      expect(await deferrals(b.slotId)).toHaveLength(0);
+      await db().deleteFrom("download_task").where("id", "=", busyTask.id).execute();
+      await db()
+        .updateTable("acquisition_plan")
+        .set({ status: "cancelled" })
+        .where("id", "=", evB2.plans[0]!)
+        .execute();
+    } finally {
+      await h.setSettings("budget", {});
+    }
+  });
+
+  it("计划失败后同一资源能重建计划(冷却后);在途 / 已完成 / 用户移除的仍挡住", async () => {
+    const { entityId, slotId } = await movieSlot("Retry Movie");
+    const rel = await release(entityId, "Retry.Movie.2024.1080p.WEB-DL.x264-GRP");
+    const ev1 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
+    expect(ev1.plans, JSON.stringify(ev1)).toHaveLength(1);
+    const p1 = ev1.plans[0]!;
+    // 失败时间 = updated_at;触发器会把它改成 now(),回拨时在事务里临时停用本 schema 的触发器
+    const fail = (id: string, reason: string, agoMs: number) =>
+      db()
+        .transaction()
+        .execute(async (trx) => {
+          await sql`alter table acquisition_plan disable trigger user`.execute(trx);
+          await trx
+            .updateTable("acquisition_plan")
+            .set({ status: "failed", status_reason: reason, updated_at: new Date(Date.now() - agoMs) })
+            .where("id", "=", id)
+            .execute();
+          await sql`alter table acquisition_plan enable trigger user`.execute(trx);
+        });
+
+    // 刚失败:冷却中 → 延后 retry_cooldown,排到冷却结束
+    await fail(p1, "source_failed: cookie expired", 0);
+    await releaseSlot(slotId);
+    const ev2 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
+    expect(ev2.plans).toHaveLength(0);
+    expect(ev2.deferred).toContainEqual({ sourceReleaseId: rel, slotId, reason: "retry_cooldown" });
+    expect((await deferrals(slotId))[0]?.reason).toBe("retry_cooldown");
+    expect((await slotRow(slotId)).next_evaluate_at).not.toBeNull();
+
+    // 冷却过后:建新计划,旧计划的幂等键让位
+    await fail(p1, "source_failed: cookie expired", GATE_RETRY_BASE_MS + 60_000);
+    const ev3 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
+    expect(ev3.plans, JSON.stringify(ev3)).toHaveLength(1);
+    const p2 = ev3.plans[0]!;
+    expect(p2).not.toBe(p1);
+    const keys = await db()
+      .selectFrom("acquisition_plan")
+      .select(["id", "idempotency_key"])
+      .where("id", "in", [p1, p2])
+      .execute();
+    expect(keys.find((k) => k.id === p2)?.idempotency_key).toBe(`acq:${slotId}:${rel}`);
+    expect(keys.find((k) => k.id === p1)?.idempotency_key).toBe(`acq:${slotId}:${rel}:retired:${p1}`);
+    expect(await deferrals(slotId)).toHaveLength(0);
+
+    // 第二次失败:冷却翻倍(30 分钟前失败仍在冷却)
+    await fail(p2, "rejected", GATE_RETRY_BASE_MS + 60_000);
+    await releaseSlot(slotId);
+    expect((await evaluateAcquisition(h.d, { sourceReleaseId: rel })).plans).toHaveLength(0);
+    await fail(p2, "rejected", 2 * GATE_RETRY_BASE_MS + 60_000);
+    const ev4 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
+    expect(ev4.plans).toHaveLength(1);
+    const p3 = ev4.plans[0]!;
+
+    // 已完成:挡住
+    await db().updateTable("acquisition_plan").set({ status: "completed" }).where("id", "=", p3).execute();
+    await releaseSlot(slotId);
+    const ev5 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
+    expect(ev5.plans).toHaveLength(0);
+    expect((await candidate(slotId, rel))?.reasons).toContain("slot_changed_or_reserved");
+
+    // 用户在下载器里移除:不自动重下
+    await fail(p3, "download_removed", 24 * 3600_000);
+    const ev6 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
+    expect(ev6.plans).toHaveLength(0);
+  });
+
+  it("取消的旧计划:重置订阅取消的立即让位,用户手动取消的仍挡住", async () => {
+    const { entityId, slotId } = await movieSlot("Cancel Movie");
+    const rel = await release(entityId, "Cancel.Movie.2024.1080p.WEB-DL.x264-GRP");
+    const ev1 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
+    expect(ev1.plans).toHaveLength(1);
+    const cancel = (id: string, reason: string) =>
+      db()
+        .updateTable("acquisition_plan")
+        .set({ status: "cancelled", status_reason: reason })
+        .where("id", "=", id)
+        .execute();
+    await cancel(ev1.plans[0]!, "subscription_reset");
+    await releaseSlot(slotId);
+    const ev2 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
+    expect(ev2.plans, JSON.stringify(ev2)).toHaveLength(1);
+    await cancel(ev2.plans[0]!, "cancelled_by_user");
+    await releaseSlot(slotId);
+    expect((await evaluateAcquisition(h.d, { sourceReleaseId: rel })).plans).toHaveLength(0);
+  });
+
+  it("在途计划(已提交)仍挡住同一资源", async () => {
+    const { entityId } = await seriesSlot("Inflight Series");
+    const rel = await release(entityId, "Inflight.Series.S01E01.1080p.WEB-DL.x264-GRP");
+    const ev1 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
+    expect(ev1.plans, JSON.stringify(ev1)).toHaveLength(1);
+    await db().updateTable("acquisition_plan").set({ status: "submitted" }).where("id", "=", ev1.plans[0]!).execute();
+    const ev2 = await evaluateAcquisition(h.d, { sourceReleaseId: rel });
+    expect(ev2.plans).toHaveLength(0);
+    expect(ev2.deferred[0]?.reason).toBe("units_inflight");
+  });
+
+  it("订阅完成、订阅暂停、资源过期时清掉延后行", async () => {
+    disk.setPct(95);
+    try {
+      // 完成
+      const a = await movieSlot("Cleanup Movie A");
+      const relA = await release(a.entityId, "Cleanup.Movie.A.2024.1080p.WEB-DL.x264-GRP");
+      await evaluateAcquisition(h.d, { sourceReleaseId: relA });
+      expect(await deferrals(a.slotId)).toHaveLength(1);
+      const subA = await db()
+        .selectFrom("subscription")
+        .selectAll()
+        .where("id", "=", a.subId)
+        .executeTakeFirstOrThrow();
+      expect(await completeSubscription(h.d, db(), subA, "test")).toBe(true);
+      expect(await deferrals(a.slotId)).toHaveLength(0);
+
+      // 暂停:到期重评时跳过并清掉
+      const b = await movieSlot("Cleanup Movie B");
+      const relB = await release(b.entityId, "Cleanup.Movie.B.2024.1080p.WEB-DL.x264-GRP");
+      await evaluateAcquisition(h.d, { sourceReleaseId: relB });
+      expect(await deferrals(b.slotId)).toHaveLength(1);
+      await db().updateTable("subscription").set({ status: "paused" }).where("id", "=", b.subId).execute();
+      await evaluateAcquisition(h.d, { acquisitionSlotId: b.slotId });
+      expect(await deferrals(b.slotId)).toHaveLength(0);
+
+      // 资源过期
+      const c = await movieSlot("Cleanup Movie C");
+      const relC = await release(c.entityId, "Cleanup.Movie.C.2024.1080p.WEB-DL.x264-GRP");
+      await evaluateAcquisition(h.d, { sourceReleaseId: relC });
+      expect(await deferrals(c.slotId)).toHaveLength(1);
+      await db().updateTable("source_release").set({ availability: "expired" }).where("id", "=", relC).execute();
+      const ev = await evaluateAcquisition(h.d, { sourceReleaseId: relC });
+      expect(ev.skipped[0]?.reason).toBe("release_expired");
+      expect(await deferrals(c.slotId)).toHaveLength(0);
+    } finally {
+      disk.setPct(10);
+    }
+  });
+});

apps/server/test/acquisition-organize/episode-resolution-evaluate.test.ts

@@ -156,4 +156,47 @@ describe.skipIf(!hasPg)("订阅评估跳过集数待确认的资源", () => {
     expect(ev.plans).toHaveLength(0);
     expect(ev.skipped).toEqual([{ sourceReleaseId: id, reason: "episode_unresolved" }]);
   });
+
+  it("集数待确认在订阅侧可见:写候选记录 episode_unresolved;确认后正常评估并覆盖", async () => {
+    const slot = await h.db
+      .selectFrom("acquisition_slot")
+      .select("id")
+      .where("media_entity_id", "=", entityId)
+      .executeTakeFirstOrThrow();
+    const candidate = async (releaseId: string) =>
+      (
+        await h.db
+          .selectFrom("artifact")
+          .select("content")
+          .where("title", "=", "candidate_evaluation")
+          .where("subject_id", "=", slot.id)
+          .execute()
+      )
+        .map(
+          (r) => r.content as { sourceReleaseId: string; decision: string; reasons: string[]; planId: string | null },
+        )
+        .find((c) => c.sourceReleaseId === releaseId);
+    const id = await seed(LILITH_40);
+    await evaluateAcquisition(h.d, { sourceReleaseId: id });
+    expect(await candidate(id)).toMatchObject({ decision: "skipped", reasons: ["episode_unresolved"] });
+
+    // 人工确认为 S02E12(与 `- 24` 的换算结果相同),重发评估:正常建计划,记录被覆盖
+    const fixed = resolved("[Lilith-Raws] Kusuriya no Hitorigoto - 24 [Baha][WEB-DL][1080p][AVC AAC][CHT][MP4]");
+    const cols = toParsedColumns(fixed.parsed);
+    await h.db
+      .updateTable("source_release")
+      .set({
+        parsed: JSON.stringify(cols.parsed),
+        parser_version: cols.parserVersion,
+        episode_resolution: JSON.stringify(fixed.resolution),
+      })
+      .where("id", "=", id)
+      .execute();
+    const ev = await evaluateAcquisition(h.d, { sourceReleaseId: id });
+    expect(ev.plans, JSON.stringify(ev)).toHaveLength(1);
+    const c = await candidate(id);
+    expect(c?.decision).toBe("accept");
+    expect(c?.planId).toBe(ev.plans[0]);
+    expect(c?.reasons).not.toContain("episode_unresolved");
+  });
 });

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

@@ -12,7 +12,7 @@ import { completeDownload, createE2E, type E2E, pgAvailable, seedBase, waitFor }
 // 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),计划结束后重评时已覆盖,不重复下载;
+//   · E02 另一版本与在途 E02 单元冲突 → 延后(候选评估 deferred;订阅接收即完成后延后行清掉),不重复下载;
 //   · refreshMetadata(编排循环按周期入队)把 TMDB 季集数写进季单元 → 三集全部入库后 once 订阅完成并写历史。
 
 const hasPg = await pgAvailable();
@@ -216,14 +216,24 @@ describe.skipIf(!hasPg)("E9 连续更新剧集:Slot 忙时到达的集全部入
       60_000,
       "另一版本识别",
     );
+    // 三集都被下载器接收后 once 订阅即完成,完成时延后行随之清掉;以候选评估记录为准
     await waitFor(
       async () =>
-        e.db
+        (
-          .selectFrom("slot_deferral")
+          await e.db
-          .selectAll()
+            .selectFrom("artifact")
-          .where("slot_id", "=", slotId)
+            .select("content")
-          .where("source_release_id", "=", loserRel.id)
+            .where("title", "=", "candidate_evaluation")
-          .executeTakeFirst(),
+            .where("subject_id", "=", slotId)
+            .execute()
+        ).find((r) => {
+          const c = r.content as { sourceReleaseId: string; decision: string; reasons: string[] };
+          return (
+            c.sourceReleaseId === loserRel.id &&
+            c.decision === "deferred" &&
+            c.reasons.includes("deferred:units_inflight")
+          );
+        }),
       60_000,
       "另一版本延后",
     );