merge: fix/b-download

Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
2026-10-03 21:23:04 +08:00
commit 15d7f2ec2f
@@ -198,13 +198,17 @@ async function emitStatus(
  from: DownloadStatus | null,
  to: DownloadStatus,
  pauseReason: PauseReason | null,
+  /** 任务行的代数:removed 行复活后 +1,完成事件换一把幂等键,复活后的完成才能再次触发整理 */
+  generation = 0,
 ) {
  await d.ctx.outbox.emit(
    exec,
    "download.status_changed",
    { downloadTaskId: id, from, to, pauseReason },
-    // 完成事件幂等:只触发一次入库;其它状态变化不去重
-    to === "completed" ? { idempotencyKey: `download.completed:${id}` } : {},
+    // 完成事件幂等:同一代只触发一次入库;其它状态变化不去重
+    to === "completed"
+      ? { idempotencyKey: generation > 0 ? `download.completed:${id}:g${generation}` : `download.completed:${id}` }
+      : {},
  );
 }

@@ -275,6 +279,7 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
    tags: t.tags,
  });
  let dtId: string | undefined;
+  let inUseBy: string | undefined;
  const result = await withAttemptToken<SubmitResult>(
    db,
    `submit:${plan.id}`,
@@ -337,7 +342,82 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
          return;
        }
        const r = res as Extract<SubmitResult, { hash: string }>;
-        const ins = await sql<{ id: string; inserted: boolean }>`
+        const existing = await trx
+          .selectFrom("download_task")
+          .select(["id", "acquisition_plan_id", "status", "removed_at"])
+          .where("instance_id", "=", topo.inst.id)
+          .where("torrent_hash", "=", r.hash)
+          .forUpdate()
+          .executeTakeFirst();
+        const removed = !!existing && (existing.removed_at !== null || existing.status === "removed");
+        if (
+          !ev.manual &&
+          existing &&
+          !removed &&
+          existing.acquisition_plan_id &&
+          existing.acquisition_plan_id !== plan.id
+        ) {
+          // 同一种子的任务行已挂着另一个计划(手动先下 / 另一 Slot 跨轮次选中同一资源):任务行只推进它挂着的计划,
+          // 本计划挂不上去,取消并释放 Slot 预留,不留没有任务的 submitted 孤儿(该资源对本 Slot 不再计划)。
+          // 手动计划不受影响:它经 import_plan_id 挂到已有任务,由整理收尾结束(organize-finalize 的 importPlanId)
+          const owner = existing.acquisition_plan_id;
+          inUseBy = owner;
+          await trx
+            .updateTable("acquisition_plan")
+            .set({ status: "cancelled", status_reason: `torrent_in_use_by_plan:${owner}` })
+            .where("id", "=", plan.id)
+            .execute();
+          await applySlotEvent(
+            d,
+            trx,
+            plan.slot_id,
+            { type: "release" },
+            { reason: "torrent_in_use_by_plan", planId: plan.id },
+          );
+          return;
+        }
+        if (existing && removed) {
+          // 同一种子的旧任务行已 removed(用户删过 / 清理删过 / Move 后删过):复活并改挂本计划,
+          // 换代让完成事件能再次触发整理
+          const slotEntity = plan.slot_id
+            ? (
+                await trx
+                  .selectFrom("acquisition_slot")
+                  .select("media_entity_id")
+                  .where("id", "=", plan.slot_id)
+                  .executeTakeFirst()
+              )?.media_entity_id
+            : null;
+          await trx
+            .updateTable("download_task")
+            .set((eb) => ({
+              acquisition_plan_id: plan.id,
+              import_plan_id: ev.manual ? plan.id : null,
+              source_release_id: rel.id,
+              media_entity_id: slotEntity ?? ev.mediaEntityId ?? rel.media_entity_id,
+              name: r.name,
+              save_path: r.savePath,
+              content_path: r.contentPath,
+              category: topo.category,
+              tags: r.tags,
+              managed: true,
+              origin: r.kind === "already_exists" ? "adopted" : "submitted",
+              status: r.kind === "accepted_unconfirmed" ? "submitting" : "queued",
+              pause_reason: null,
+              removed_at: null,
+              completed_at: null,
+              no_progress_since: null,
+              last_progress_at: null,
+              missing_since: null,
+              missing_count: 0,
+              generation: eb("generation", "+", 1),
+              next_check_at: d.ports.now(),
+            }))
+            .where("id", "=", existing.id)
+            .execute();
+          dtId = existing.id;
+        } else {
+          const ins = await sql<{ id: string; inserted: boolean }>`
          insert into download_task (instance_id, torrent_hash, infohash_v1, infohash_v2, acquisition_plan_id, import_plan_id, source_release_id,
                                     media_entity_id, name,
                                     save_path, content_path, category, tags, managed, origin, status, size_bytes, next_check_at)
@@ -353,7 +433,8 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
                media_entity_id = coalesce(download_task.media_entity_id, excluded.media_entity_id),
                managed = true, tags = excluded.tags, updated_at = now()
          returning id, (xmax = 0) as inserted`.execute(trx);
-        dtId = ins.rows[0]!.id;
+          dtId = ins.rows[0]!.id;
+        }
        await trx
          .updateTable("acquisition_plan")
          .set({ status: "submitted", status_reason: r.kind })
@@ -369,13 +450,14 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
        await d.ctx.outbox.emit(
          trx,
          "download.submitted",
-          { downloadTaskId: dtId, instanceId: topo.inst.id, torrentHash: r.hash, planId: plan.id, name: r.name },
+          { downloadTaskId: dtId!, instanceId: topo.inst.id, torrentHash: r.hash, planId: plan.id, name: r.name },
          { idempotencyKey: `download.submitted:${plan.id}` },
        );
      },
    },
  );
  if (result.kind === "rejected" || result.kind === "source_failed") return { status: "failed", reason: result.reason };
+  if (inUseBy) return { status: "skipped", reason: "torrent_in_use_by_plan" };
  const hash = (result as Extract<SubmitResult, { hash: string }>).hash;
  // 季包/多集包:按缺集选文件(接管的已有任务不改文件选择);失败不影响提交结果
  if (result.kind === "added" && plan.unit_keys.length > 0 && plan.file_selection === null) {
@@ -636,6 +718,9 @@ export interface SyncStats {
  thresholdResumed: number;
 }

+/** 连续多少轮同步(列表 + 按 hash 查)都查不到才判 removed;与资产对账的 missingConfirmAfter 同口径 */
+export const MISSING_CONFIRM_ROUNDS = 3;
+
 export async function syncDownloads(d: AoDeps): Promise<SyncStats> {
  const stats: SyncStats = {
    instances: 0,
@@ -715,13 +800,20 @@ async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats)
        // 管理标签被移除:停止后续自动动作,保留历史与资产
        await trx
          .updateTable("download_task")
-          .set({ managed: false, tags: still.tags, last_synced_at: now })
+          .set({ managed: false, tags: still.tags, last_synced_at: now, missing_since: null, missing_count: 0 })
+          .where("id", "=", row.id)
+          .execute();
+      } else if (row.missing_count + 1 < MISSING_CONFIRM_ROUNDS) {
+        // 列表与按 hash 都查不到:先记缺席(qB 刚重启、任务还没加载完时会短暂返回空 / 部分列表)
+        await trx
+          .updateTable("download_task")
+          .set({ missing_since: row.missing_since ?? now, missing_count: row.missing_count + 1, last_synced_at: now })
          .where("id", "=", row.id)
          .execute();
      } else {
        await trx
          .updateTable("download_task")
-          .set({ status: "removed", removed_at: now, last_synced_at: now })
+          .set({ status: "removed", removed_at: now, last_synced_at: now, missing_count: row.missing_count + 1 })
          .where("id", "=", row.id)
          .execute();
        await emitStatus(d, trx, row.id, row.status, "removed", null);
@@ -771,7 +863,8 @@ async function observeLocked(
  let noProgressSince = row.no_progress_since;
  let lastProgressAt = row.last_progress_at;
  let stalledNow = false;
-  if (obs.status === "downloading") {
+  // importing / imported 期间(qB 重新校验 / 移动等中间态)不做无进展判定
+  if (obs.status === "downloading" && (merged.status === "downloading" || merged.status === "stalled")) {
    if (downloaded > row.downloaded_bytes || !lastProgressAt) {
      lastProgressAt = now;
      noProgressSince = null;
@@ -809,12 +902,28 @@ async function observeLocked(
      last_progress_at: lastProgressAt,
      no_progress_since: noProgressSince,
      last_synced_at: now,
+      missing_since: null,
+      missing_count: 0,
      ...(becameComplete ? { completed_at: now } : {}),
    })
    .where("id", "=", row.id)
    .execute();
  if (changed)
-    await emitStatus(d, trx, row.id, row.status, merged.status, merged.status === "paused" ? merged.pauseReason : null);
+    await emitStatus(
+      d,
+      trx,
+      row.id,
+      row.status,
+      merged.status,
+      merged.status === "paused" ? merged.pauseReason : null,
+      row.generation,
+    );
+  // 卡住 / 异常后恢复正常(进度增长或状态恢复):关闭同一任务未关闭的无进展 / 异常事项
+  if (changed && RECOVERED.has(merged.status)) {
+    if (row.status === "stalled") await closeRecoveredDecision(d, trx, `download_stalled:${row.id}`, merged.status);
+    if (row.status === "failed" || row.status === "missing")
+      await closeRecoveredDecision(d, trx, `download_failed:${row.id}`, merged.status);
+  }
  if (becameComplete) {
    stats.completed.push(row.id);
    const plan = row.acquisition_plan_id
@@ -858,6 +967,37 @@ async function observeLocked(
  return changed;
 }

+/** 视为"恢复正常"的状态(用户暂停不算恢复,事项留给人看) */
+const RECOVERED = new Set<DownloadStatus>(["queued", "downloading", "completed", "importing", "imported"]);
+
+/**
+ * 卡住 / 异常事项自动关闭:只关 open / awaiting_user;Agent 正在调查(investigating)的由它自己核实后关闭。
+ * resolution 不带处理选项,executeResolution 不会再执行任何动作。
+ */
+async function closeRecoveredDecision(d: AoDeps, trx: Exec, dedupeKey: string, status: DownloadStatus) {
+  const by = "system:download-sync";
+  const rows = await trx
+    .updateTable("decision")
+    .set({
+      status: "resolved",
+      resolved_by: by,
+      resolved_at: d.ports.now(),
+      resolution: jsonb({ key: "download_recovered", optionKey: null, by, note: `下载已恢复（${status}）` }),
+    })
+    .where("dedupe_key", "=", dedupeKey)
+    .where("status", "in", ["open", "awaiting_user"])
+    .returning("id")
+    .execute();
+  for (const r of rows) {
+    await d.ctx.outbox.emit(
+      trx,
+      "decision.updated",
+      { decisionId: r.id, status: "resolved" },
+      { idempotencyKey: `decision.updated:${r.id}:resolved` },
+    );
+  }
+}
+
 /** 用户给已有任务贴了管理标签:登记接管,不改保存路径与其它标签 */
 async function adopt(d: AoDeps, topo: InstanceTopology, t: QbTorrent, now: Date, stats: SyncStats) {
  const db = d.ctx.db;
@@ -901,12 +1041,49 @@ async function adopt(d: AoDeps, topo: InstanceTopology, t: QbTorrent, now: Date,
        last_progress_at: now,
        ...(st.status === "completed" ? { completed_at: now } : {}),
      })
-      .onConflict((oc) => oc.columns(["instance_id", "torrent_hash"]).doNothing())
-      .returning("id")
+      // 同一种子的旧任务行已 removed:复活为接管任务(原计划已在移除时结束,不再挂回);未移除的冲突不动
+      .onConflict((oc) =>
+        oc
+          .columns(["instance_id", "torrent_hash"])
+          .doUpdateSet((eb) => ({
+            acquisition_plan_id: null,
+            import_plan_id: null,
+            source_release_id: eb.fn.coalesce("download_task.source_release_id", "excluded.source_release_id"),
+            media_entity_id: eb.fn.coalesce("download_task.media_entity_id", "excluded.media_entity_id"),
+            name: eb.ref("excluded.name"),
+            save_path: eb.ref("excluded.save_path"),
+            content_path: eb.ref("excluded.content_path"),
+            category: eb.ref("excluded.category"),
+            tags: eb.ref("excluded.tags"),
+            managed: true,
+            origin: "adopted",
+            status: eb.ref("excluded.status"),
+            pause_reason: eb.ref("excluded.pause_reason"),
+            raw_state: eb.ref("excluded.raw_state"),
+            progress: eb.ref("excluded.progress"),
+            size_bytes: eb.ref("excluded.size_bytes"),
+            downloaded_bytes: eb.ref("excluded.downloaded_bytes"),
+            remaining_bytes: eb.ref("excluded.remaining_bytes"),
+            ratio: eb.ref("excluded.ratio"),
+            seeding_time_sec: eb.ref("excluded.seeding_time_sec"),
+            last_synced_at: eb.ref("excluded.last_synced_at"),
+            last_progress_at: eb.ref("excluded.last_progress_at"),
+            completed_at: eb.ref("excluded.completed_at"),
+            no_progress_since: null,
+            removed_at: null,
+            missing_since: null,
+            missing_count: 0,
+            generation: eb("download_task.generation", "+", 1),
+          }))
+          .where((eb) =>
+            eb.or([eb("download_task.removed_at", "is not", null), eb("download_task.status", "=", "removed")]),
+          ),
+      )
+      .returning(["id", "generation"])
      .executeTakeFirst();
    if (!ins) return;
    stats.adopted++;
-    await emitStatus(d, trx, ins.id, null, st.status, st.status === "paused" ? st.pauseReason : null);
+    await emitStatus(d, trx, ins.id, null, st.status, st.status === "paused" ? st.pauseReason : null, ins.generation);
    if (st.status === "completed") stats.completed.push(ins.id);
  });
 }
@@ -0,0 +1,324 @@
+import { createQbClient } from "@umt/adapters";
+import { sql } from "@umt/db";
+import { createMockQb, createTempTree, type MockQb, makeTorrent, type TempTree } from "@umt/testkit";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts";
+import {
+  cleanupCandidates,
+  MISSING_CONFIRM_ROUNDS,
+  submitDownload,
+  syncDownloads,
+} from "../../src/modules/acquisition-organize/services/downloads.ts";
+import { executeRecord, planImport } from "../../src/modules/acquisition-organize/services/organize.ts";
+import { loadVolumes } from "../../src/modules/acquisition-organize/services/volumes.ts";
+import {
+  createHarness,
+  fakeDisk,
+  type Harness,
+  pgAvailable,
+  seedBase,
+  seedEntity,
+  seedRelease,
+  seedSlot,
+} from "./harness.ts";
+
+// 下载调度卡死修复:removed 行再次提交复活、同 hash 第二个计划不成孤儿、已入库任务不被 qB 中间态改回、
+// 多轮缺席才判移除、卡住 / 异常恢复后事项自动关闭。真实 PG + mock qB;无进展阈值设为 0 方便构造 stalled。
+
+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);
+  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(40);
+    h = await createHarness("dl_recovery", {
+      diskUsage: disk.fn,
+      torrentSource: async ({ release }) => ({ kind: "file", bytes: torrents.get(release.id)! }),
+      options: { noProgressMs: 0 },
+    });
+    base = await seedBase(h.db, { libraryPath: lib.root, cachePath: cache.root, qbUrl: qb.url });
+    await h.setSettings("retention", { seedRetentionDays: 0 });
+  }, 60_000);
+
+  afterAll(async () => {
+    await h?.close();
+    await qb?.close();
+    await lib?.cleanup();
+    await cache?.cleanup();
+  });
+
+  const ext = () => createQbClient({ instanceId: "t", baseUrl: qb.url, username: "admin", password: "adminadmin" });
+  const dtRow = (id: string) =>
+    h.db.selectFrom("download_task").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
+  const planRow = (id: string) =>
+    h.db.selectFrom("acquisition_plan").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
+  const slotRow = (id: string) =>
+    h.db.selectFrom("acquisition_slot").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
+  const decision = (key: string) =>
+    h.db
+      .selectFrom("decision")
+      .selectAll()
+      .where("dedupe_key", "=", key)
+      .orderBy("created_at", "desc")
+      .executeTakeFirst();
+
+  async function movie(title: string, year: number) {
+    const entityId = await seedEntity(h.db, { kind: "movie", title, year });
+    const slotId = await seedSlot(h.db, { entityId });
+    return { entityId, slotId };
+  }
+
+  async function release(entityId: string, title: string) {
+    const t = 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 };
+  }
+
+  async function planFor(releaseId: string) {
+    const ev = await evaluateAcquisition(h.d, { sourceReleaseId: releaseId });
+    expect(ev.plans).toHaveLength(1);
+    return ev.plans[0]!;
+  }
+
+  /** 手动获取:同一资源可再次明确选择(slot_id 为空、evaluation.manual) */
+  async function manualPlanLike(planId: string, entityId: string) {
+    const r = await sql<{ id: string }>`
+      insert into acquisition_plan (slot_id, source_release_id, purpose, status, priority, evaluation, reserved_bytes,
+                                    budget_reservation, payload_hash, idempotency_key, status_reason)
+      select null, source_release_id, 'fill', 'reserved', 'immediate',
+             evaluation || ${JSON.stringify({ manual: true, mediaEntityId: entityId, importMode: "add" })}::jsonb,
+             reserved_bytes, '{}'::jsonb, payload_hash, ${`manual:${crypto.randomUUID()}`}, 'manual_selection'
+        from acquisition_plan where id = ${planId}
+      returning id`.execute(h.db);
+    return r.rows[0]!.id;
+  }
+
+  async function syncTimes(n: number) {
+    for (let i = 0; i < n; i++) await syncDownloads(h.d);
+  }
+
+  it("removed 行再次提交:行复活挂到新计划,完成后换代发完成事件并能入库", async () => {
+    const { entityId } = await movie("复活片", 2021);
+    const rel = await release(entityId, "Revive.Movie.2021.1080p.BluRay.x264-GRP");
+    const p1 = await planFor(rel.id);
+    const s1 = await submitDownload(h.d, p1);
+    expect(s1).toMatchObject({ status: "submitted", reason: "added" });
+    const dtId = s1.downloadTaskId!;
+
+    await ext().delete([rel.hash], { deleteFiles: false });
+    await syncTimes(MISSING_CONFIRM_ROUNDS);
+    expect(await dtRow(dtId)).toMatchObject({ status: "removed" });
+    expect((await dtRow(dtId)).removed_at).not.toBeNull();
+    expect(await planRow(p1)).toMatchObject({ status: "failed", status_reason: "download_removed" });
+
+    // 用户再次明确选择同一资源
+    const p2 = await manualPlanLike(p1, entityId);
+    const s2 = await submitDownload(h.d, p2);
+    expect(s2).toMatchObject({ status: "submitted", reason: "added", downloadTaskId: dtId });
+    const revived = await dtRow(dtId);
+    expect(revived).toMatchObject({
+      status: "queued",
+      removed_at: null,
+      completed_at: null,
+      acquisition_plan_id: p2,
+      import_plan_id: p2,
+      origin: "submitted",
+      missing_count: 0,
+      generation: 1,
+    });
+    expect(await planRow(p2)).toMatchObject({ status: "submitted", status_reason: "added" });
+
+    await qb.complete(rel.hash, { writeFiles: true });
+    const sync = await syncDownloads(h.d);
+    expect(sync.completed).toContain(dtId);
+    const done = (await h.outbox("download.status_changed")).filter(
+      (e) => e.payload.downloadTaskId === dtId && e.payload.to === "completed",
+    );
+    expect(done.map((e) => e.idempotency_key)).toEqual([`download.completed:${dtId}:g1`]);
+
+    const plan = await planImport(h.d, { downloadTaskId: dtId, trigger: "auto", actorType: "system" });
+    expect(plan.status).toBe("running");
+    expect(await executeRecord(h.d, plan.recordId!)).toMatchObject({ status: "succeeded", succeeded: 1 });
+    expect((await dtRow(dtId)).status).toBe("imported");
+  }, 60_000);
+
+  it("removed 行被重新贴管理标签:接管时复活为 adopted 行", async () => {
+    const { entityId } = await movie("接管复活", 2020);
+    const rel = await release(entityId, "Adopt.Revive.2020.1080p.WEB-DL.x264-GRP");
+    const s = await submitDownload(h.d, await planFor(rel.id));
+    const dtId = s.downloadTaskId!;
+    await ext().delete([rel.hash], { deleteFiles: false });
+    await syncTimes(MISSING_CONFIRM_ROUNDS);
+    expect((await dtRow(dtId)).status).toBe("removed");
+
+    qb.seed({ hash: rel.hash, name: "Adopt.Revive.2020.1080p.WEB-DL.x264-GRP.mkv", tags: ["umt"], size: SIZE });
+    qb.setProgress(rel.hash, 0.2);
+    const st = await syncDownloads(h.d);
+    expect(st.adopted).toBe(1);
+    expect(await dtRow(dtId)).toMatchObject({
+      origin: "adopted",
+      acquisition_plan_id: null,
+      removed_at: null,
+      status: "downloading",
+      generation: 1,
+    });
+  }, 60_000);
+
+  it("同 hash 已挂在途计划:第二个计划取消并释放 Slot 预留,不留 submitted 孤儿", async () => {
+    const { entityId } = await movie("同种双计划", 2022);
+    const rel = await release(entityId, "Shared.Hash.2022.1080p.BluRay.x264-GRP");
+    const p1 = await planFor(rel.id);
+    const s1 = await submitDownload(h.d, p1);
+    expect(s1.status).toBe("submitted");
+
+    // 另一个 Slot(另一订阅 / 版本)跨轮次选中同一资源
+    const slot2 = await seedSlot(h.db, { entityId, name: "second" });
+    const ev = await evaluateAcquisition(h.d, { acquisitionSlotId: slot2 });
+    expect(ev.plans).toHaveLength(1);
+    const p2 = ev.plans[0]!;
+    expect((await slotRow(slot2)).status).toBe("reserved");
+    const out = await submitDownload(h.d, p2);
+    expect(out).toEqual({ status: "skipped", reason: "torrent_in_use_by_plan" });
+    expect(await planRow(p2)).toMatchObject({ status: "cancelled", status_reason: `torrent_in_use_by_plan:${p1}` });
+    const s2 = await slotRow(slot2);
+    expect(["reserved", "downloading", "verifying"]).not.toContain(s2.status);
+    expect(s2.reserved_plan_id).toBeNull();
+    // 任务行仍只推进第一个计划;同一资源只剩一个 submitted 计划
+    expect((await dtRow(s1.downloadTaskId!)).acquisition_plan_id).toBe(p1);
+    expect(
+      await h.db
+        .selectFrom("acquisition_plan")
+        .select("id")
+        .where("source_release_id", "=", rel.id)
+        .where("status", "=", "submitted")
+        .execute(),
+    ).toEqual([{ id: p1 }]);
+  }, 60_000);
+
+  it("已入库任务经 checkingResumeData / checkingUP / moving 再回到完成:仍是 imported,completed_at 不变,清理仍能选中", async () => {
+    const { entityId } = await movie("回退片", 2019);
+    const rel = await release(entityId, "Regress.Movie.2019.1080p.BluRay.x264-GRP");
+    const s = await submitDownload(h.d, await planFor(rel.id));
+    const dtId = s.downloadTaskId!;
+    await qb.complete(rel.hash, { writeFiles: true });
+    await syncDownloads(h.d);
+    const plan = await planImport(h.d, { downloadTaskId: dtId, trigger: "auto", actorType: "system" });
+    expect(await executeRecord(h.d, plan.recordId!)).toMatchObject({ status: "succeeded" });
+    const before = await dtRow(dtId);
+    expect(before.status).toBe("imported");
+    expect(before.completed_at).not.toBeNull();
+
+    const t = qb.get(rel.hash)!;
+    const done = t.state;
+    for (const [state, progress] of [
+      ["checkingResumeData", 1],
+      ["queuedDL", 1],
+      ["checkingUP", 0.4],
+      ["moving", 1],
+    ] as const) {
+      t.state = state;
+      t.progress = progress;
+      await syncDownloads(h.d);
+      const row = await dtRow(dtId);
+      expect(row.status, state).toBe("imported");
+      expect(row.raw_state, state).toBe(state);
+    }
+    qb.setProgress(rel.hash, 1);
+    t.state = done;
+    await syncDownloads(h.d);
+    const after = await dtRow(dtId);
+    expect(after.status).toBe("imported");
+    expect(after.completed_at?.getTime()).toBe(before.completed_at!.getTime());
+    expect(after.no_progress_since).toBeNull();
+    const completedEvents = (await h.outbox("download.status_changed")).filter(
+      (e) => e.payload.downloadTaskId === dtId && e.payload.from === "imported",
+    );
+    expect(completedEvents).toHaveLength(0);
+
+    const vol = (await loadVolumes(h.d))[0]!;
+    const cand = (await cleanupCandidates(h.d, vol)).find((c) => c.row.id === dtId);
+    expect(cand).toMatchObject({ cleanable: true, lockReason: null });
+  }, 60_000);
+
+  it("qB 一轮查不到不判移除;连续缺席达到阈值才判移除;中途重新出现清掉缺席", async () => {
+    const { entityId } = await movie("缺席片", 2018);
+    const rel = await release(entityId, "Missing.Movie.2018.1080p.BluRay.x264-GRP");
+    const p = await planFor(rel.id);
+    const s = await submitDownload(h.d, p);
+    const dtId = s.downloadTaskId!;
+    const name = qb.get(rel.hash)!.name;
+
+    // qB 刚重启:列表与按 hash 都查不到
+    await ext().delete([rel.hash], { deleteFiles: false });
+    const st1 = await syncDownloads(h.d);
+    expect(st1.removed).toBe(0);
+    const r1 = await dtRow(dtId);
+    expect(r1).toMatchObject({ removed_at: null, missing_count: 1 });
+    expect(r1.status).not.toBe("removed");
+    expect(r1.missing_since).not.toBeNull();
+    expect((await planRow(p)).status).toBe("submitted");
+
+    // 任务加载回来:缺席清零
+    qb.seed({ hash: rel.hash, name, tags: ["umt"], size: SIZE });
+    qb.setProgress(rel.hash, 0.3);
+    await syncDownloads(h.d);
+    expect(await dtRow(dtId)).toMatchObject({ missing_count: 0, missing_since: null, removed_at: null });
+
+    // 真被删:连续缺席到阈值才判移除
+    await ext().delete([rel.hash], { deleteFiles: false });
+    await syncTimes(MISSING_CONFIRM_ROUNDS - 1);
+    expect((await dtRow(dtId)).removed_at).toBeNull();
+    expect((await dtRow(dtId)).missing_count).toBe(MISSING_CONFIRM_ROUNDS - 1);
+    const st = await syncDownloads(h.d);
+    expect(st.removed).toBe(1);
+    expect(await dtRow(dtId)).toMatchObject({ status: "removed" });
+    expect(await planRow(p)).toMatchObject({ status: "failed", status_reason: "download_removed" });
+  }, 60_000);
+
+  it("stalled / 异常后恢复:同一任务未关闭的事项自动关闭为 resolved", async () => {
+    const { entityId } = await movie("卡住片", 2017);
+    const rel = await release(entityId, "Stalled.Movie.2017.1080p.BluRay.x264-GRP");
+    const s = await submitDownload(h.d, await planFor(rel.id));
+    const dtId = s.downloadTaskId!;
+    qb.setState(rel.hash, "downloading");
+    qb.setProgress(rel.hash, 0.2);
+    await syncDownloads(h.d); // 有进展
+    await syncDownloads(h.d); // 字节不增长,阈值 0 → stalled
+    expect((await dtRow(dtId)).status).toBe("stalled");
+    expect(await decision(`download_stalled:${dtId}`)).toMatchObject({ kind: "download_stalled", status: "open" });
+
+    qb.setProgress(rel.hash, 0.5);
+    await syncDownloads(h.d);
+    expect((await dtRow(dtId)).status).toBe("downloading");
+    const closed = await decision(`download_stalled:${dtId}`);
+    expect(closed).toMatchObject({ status: "resolved", resolved_by: "system:download-sync" });
+    expect((await h.outbox("decision.updated")).map((e) => e.payload)).toContainEqual({
+      decisionId: closed!.id,
+      status: "resolved",
+    });
+
+    // qB 报错 → download_failed;恢复下载 → 关闭
+    qb.setError(rel.hash);
+    await syncDownloads(h.d);
+    expect((await dtRow(dtId)).status).toBe("failed");
+    expect(await decision(`download_failed:${dtId}`)).toMatchObject({ status: "open" });
+    qb.setState(rel.hash, "downloading");
+    qb.setProgress(rel.hash, 0.7);
+    await syncDownloads(h.d);
+    expect((await dtRow(dtId)).status).toBe("downloading");
+    expect(await decision(`download_failed:${dtId}`)).toMatchObject({ status: "resolved" });
+  }, 60_000);
+});
@@ -2,7 +2,11 @@ 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 "../../src/modules/acquisition-organize/services/acquisition.ts";
-import { submitDownload, syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts";
+import {
+  MISSING_CONFIRM_ROUNDS,
+  submitDownload,
+  syncDownloads,
+} from "../../src/modules/acquisition-organize/services/downloads.ts";
 import { executeRecord, planImport } from "../../src/modules/acquisition-organize/services/organize.ts";
 import {
  applyEpisodeTotal,
@@ -159,7 +163,7 @@ describe.skipIf(!hasPg)("订阅增强(真实 PG + mock qB + 临时目录)", () =
    // 用户在 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);
+    for (let i = 0; i < MISSING_CONFIRM_ROUNDS; i++) 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;
@@ -41,6 +41,8 @@ describe.skipIf(!hasPg)("E15 Agent 质量(fake 模型 + 真实能力)", () => {
          name: "Dune.Part.Two.2024.1080p.BluRay.x264-GRP",
          status: "failed",
          last_error: JSON.stringify({ message: "qB 状态 missingFiles" }),
+          // 已连续缺席两轮:下一轮同步即确认移除(多轮缺席才判移除)
+          missing_count: 2,
        })
        .returning("id")
        .executeTakeFirstOrThrow()
@@ -2,7 +2,7 @@ import { createQbClient } from "@umt/adapters";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
 import type { createAcquisitionOrganizeModule } from "../../src/modules/acquisition-organize/index.ts";
 import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts";
-import { syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts";
+import { MISSING_CONFIRM_ROUNDS, syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts";
 import { torrentFor } from "./fixtures/modules.ts";
 import { createE2E, type E2E, pgAvailable, seedBase, waitFor } from "./harness.ts";

@@ -206,7 +206,8 @@ describe.skipIf(!hasPg)("E8 AT24 标签移除 / 任务消失", () => {
    });
    await ext.delete([ids.D.hash], { deleteFiles: false });
    const externalDeletes = e.qb.callsTo("torrents/delete").length;
-    await syncDownloads(deps());
+    // 连续多轮查不到才确认移除
+    for (let i = 0; i < MISSING_CONFIRM_ROUNDS; i++) await syncDownloads(deps());
    const row = await dt("D");
    expect(row.status).toBe("removed");
    expect(row.removed_at).not.toBeNull();
@@ -223,6 +223,11 @@ export const DownloadTask = EntityBase.extend({
  removedAt: Timestamp.nullable(),
  fileList: z.array(z.object({ path: z.string(), size: NonNegInt, progress: z.number().optional() })).nullable(),
  lastError: JsonObject.nullable(),
+  /** 下载器里连续查不到的起点与轮数:达到阈值才判 removed,中途出现即清零 */
+  missingSince: Timestamp.nullable().optional(),
+  missingCount: NonNegInt.optional(),
+  /** removed 行被再次提交 / 接管复活的次数(完成事件幂等键随之换代) */
+  generation: NonNegInt.optional(),
 });
 export type DownloadTask = z.infer<typeof DownloadTask>;
 export const DownloadTaskCreate = DownloadTask.pick({
@@ -0,0 +1,25 @@
+import type { Kysely } from "kysely";
+import { execDdl } from "../src/migration-helpers.ts";
+
+// 下载调度卡死修复(docs/plan/system-review.md ③):
+// - missing_since / missing_count:qB 列表与按 hash 查都查不到时先记缺席,连续多轮才判 removed,中途出现即清零
+// - generation:同一 (实例, hash) 的 removed 行被再次提交 / 接管而复活时 +1,完成事件幂等键随之换代,
+//   复活后的完成仍能触发整理
+// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型
+export async function up(db: Kysely<any>): Promise<void> {
+  await execDdl(
+    db,
+    `alter table download_task add column if not exists missing_since timestamptz;
+     alter table download_task add column if not exists missing_count integer not null default 0;
+     alter table download_task add column if not exists generation integer not null default 0;`,
+  );
+}
+// biome-ignore lint/suspicious/noExplicitAny: 同上
+export async function down(db: Kysely<any>): Promise<void> {
+  await execDdl(
+    db,
+    `alter table download_task drop column if exists missing_since;
+     alter table download_task drop column if exists missing_count;
+     alter table download_task drop column if exists generation;`,
+  );
+}
@@ -719,6 +719,9 @@ export interface DownloadTaskTable {
  updated_at: GeneratedTimestamptz;
  media_entity_id: string | null;
  import_plan_id: string | null;
+  missing_since: Timestamptz | null;
+  missing_count: Generated<number>;
+  generation: Generated<number>;
 }

 export interface DownloaderInstanceTable {
@@ -62,7 +62,9 @@ export function toDomainDownloadStatus(status: DownloadStatus, pauseReason: Paus

 /**
 * 同步轮询时合并"下载器观察"与"当前业务状态":
- * - 业务叠加态在其底层下载状态不变时保留(importing/imported 遇 completed、stalled 遇 downloading、submitting 遇 queued);
+ * - importing / imported 只由整理侧推进:下载器的任何中间态(checkingResumeData / queuedDL / checkingUP / moving,
+ *   乃至出错 / 缺文件)都不把它改回下载阶段,同步只更新原始字段;只有 removed(调用方确认移除)例外;
+ * - 其余业务叠加态在其底层下载状态不变时保留(stalled 遇 downloading、submitting 遇 queued);
 * - 暂停原因为 policy/agent 且下载器仍是暂停时保留原因,不被改写成 user;
 * - 其余以观察为准。
 */
@@ -72,7 +74,7 @@ export function mergeObservedDownloadState(
 ): ContractDownloadState {
  const next = TO_CONTRACT[observed];
  const keep =
-    ((current.status === "importing" || current.status === "imported") && observed === "completed") ||
+    ((current.status === "importing" || current.status === "imported") && observed !== "removed") ||
    (current.status === "stalled" && observed === "downloading") ||
    (current.status === "submitting" && observed === "queued");
  if (keep) return { ...current };
@@ -57,6 +57,11 @@ describe("下载状态", () => {
      st("paused", "threshold"),
    );
    expect(M.mergeObservedDownloadState(st("imported"), "removed")).toEqual(st("removed"));
+    // 已进入整理侧:qB 重启 / 重新校验 / 移动的中间态不把状态改回下载阶段
+    for (const o of ["queued", "downloading", "paused_by_user", "error", "missing_files"] as const) {
+      expect(M.mergeObservedDownloadState(st("imported"), o), o).toEqual(st("imported"));
+      expect(M.mergeObservedDownloadState(st("importing"), o), o).toEqual(st("importing"));
+    }
  });
 });