refactor(ao): background loops only orchestrate; cleanup/reconcile endpoints enqueue

- separate qB sync timer from the planning timer
- threshold cleanup, reserved-plan and pending-import sweeps, 6h reconcile enqueue
  engine tasks with time-slot idempotency keys instead of running inline
- pending-import sweep selects by completed_at (sync refreshes updated_at every round)
- POST /download-cache/cleanup and /library/reconcile return {taskId, runId} (202);
  web jumps to the engine run page
- test: a 50MB verifyAndImport in progress does not delay qB sync

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

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

@@ -0,0 +1,122 @@
+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 "../services/acquisition.ts";
+import { submitDownload, syncDownloads } from "../services/downloads.ts";
+import { executeRecord, planImport } from "../services/organize.ts";
+import {
+  createHarness,
+  fakeDisk,
+  type Harness,
+  pgAvailable,
+  seedBase,
+  seedEntity,
+  seedRelease,
+  seedSlot,
+} from "./harness.ts";
+
+// W4-D P0:后台循环只编排。补入库在巡检里只入队 verifyAndImport;一个 50MB 的入库(复制 + sha256 + 慢 ffprobe)
+// 进行中时,qB 同步定时器仍按周期打到 mock qB(用 mock qB 的请求时间戳断言)。
+
+const hasPg = await pgAvailable();
+const SIZE = 50 * 1024 * 1024;
+const SYNC_MS = 200;
+const PROBE_DELAY_MS = 2500;
+
+describe.skipIf(!hasPg)("acquisition-organize 后台循环不被入库阻塞", () => {
+  let h: Harness;
+  let qb: MockQb;
+  let lib: TempTree;
+  let cache: TempTree;
+  const disk = fakeDisk(10 * 1024 ** 3);
+  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("bgloop", {
+      diskUsage: disk.fn,
+      torrentSource: async ({ release }) => ({ kind: "file", bytes: torrents.get(release.id)! }),
+      // 慢 ffprobe:让一次入库稳定地持续数秒
+      probe: async () => {
+        await new Promise((r) => setTimeout(r, PROBE_DELAY_MS));
+        return { status: "unknown", reason: "ffprobe_unavailable" };
+      },
+      options: { backgroundLoops: true, syncIntervalMs: SYNC_MS, planIntervalMs: SYNC_MS, reconcileIntervalMs: 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?.module.stop?.();
+    await h?.close();
+    await qb?.close();
+    await lib?.cleanup();
+    await cache?.cleanup();
+  });
+
+  it("巡检只入队 verifyAndImport;50MB 入库进行中 qB 同步仍按时执行", async () => {
+    const entityId = await seedEntity(h.db, { kind: "movie", title: "长片", year: 2023 });
+    await seedSlot(h.db, { entityId });
+    const t = makeTorrent({ name: "Long.Movie.2023.1080p.BluRay.x264-GRP.mkv", length: SIZE, pieceSeed: 7 });
+    const relId = await seedRelease(h.db, {
+      siteId: base.siteId,
+      entityId,
+      title: "Long.Movie.2023.1080p.BluRay.x264-GRP",
+      size: SIZE,
+      hash: t.hash,
+    });
+    torrents.set(relId, t.bytes);
+    const ev = await evaluateAcquisition(h.d, { sourceReleaseId: relId });
+    const sub = await submitDownload(h.d, ev.plans[0]!);
+    expect(sub.status).toBe("submitted");
+    await qb.complete(t.hash, { writeFiles: true });
+    expect((await syncDownloads(h.d)).completed).toContain(sub.downloadTaskId);
+    const dtId = sub.downloadTaskId!;
+    // 完成超过 5 分钟且没有整理记录 → 巡检对象
+    await sql`update download_task set completed_at = now() - interval '10 minutes' where id = ${dtId}`.execute(h.db);
+
+    await h.module.start?.(h.ctx);
+    const deadline = Date.now() + 10_000;
+    while (!h.engine.enqueued.some((e) => e.task === "verifyAndImport") && Date.now() < deadline) {
+      await new Promise((r) => setTimeout(r, 50));
+    }
+    const enq = h.engine.enqueued.find((e) => e.task === "verifyAndImport");
+    expect(enq?.input).toMatchObject({ downloadTaskId: dtId });
+    expect(enq?.key).toMatch(new RegExp(`^verifyAndImport:import:${dtId}:\\d+$`));
+    // 循环没有自己执行入库
+    const recs = await h.db
+      .selectFrom("organization_record")
+      .select("id")
+      .where("download_task_id", "=", dtId)
+      .execute();
+    expect(recs).toHaveLength(0);
+    // 同一时间槽重复巡检用同一个引擎幂等键(真实引擎去重;假引擎只记录)
+    await new Promise((r) => setTimeout(r, SYNC_MS * 3));
+    const keys = new Set(h.engine.enqueued.filter((e) => e.task === "verifyAndImport").map((e) => e.key));
+    expect(keys.size).toBe(1);
+
+    // 模拟 worker 执行这次 verifyAndImport(同进程,与同步循环并发)
+    const t0 = Date.now();
+    const plan = await planImport(h.d, { downloadTaskId: dtId, trigger: "auto", actorType: "system" });
+    expect(plan.status).toBe("running");
+    const ex = await executeRecord(h.d, plan.recordId!);
+    const t1 = Date.now();
+    expect(ex).toMatchObject({ status: "succeeded", succeeded: 1 });
+    expect(t1 - t0).toBeGreaterThanOrEqual(PROBE_DELAY_MS);
+
+    const stamps = qb
+      .callsTo("torrents/info")
+      .map((c) => c.at)
+      .filter((at) => at >= t0 && at <= t1);
+    // 每 200ms 一轮(同步本身几十 ms);入库期间至少有 1/3 的轮次、最大间隔不超过 1s
+    expect(stamps.length).toBeGreaterThanOrEqual(Math.floor((t1 - t0) / SYNC_MS / 3));
+    const gaps = [t0, ...stamps, t1].slice(1).map((v, i) => v - [t0, ...stamps][i]!);
+    expect(Math.max(...gaps)).toBeLessThan(1000);
+  }, 90_000);
+});

apps/server/src/modules/acquisition-organize/__tests__/harness.ts

@@ -117,6 +117,7 @@ export function fakeDisk(total: number) {
 type Loose = any;
 
 export interface Harness {
+  module: ReturnType<typeof createAcquisitionOrganizeModule>;
   db: Kysely<DB>;
   ctx: ServerContext;
   d: AoDeps;
@@ -213,6 +214,7 @@ export async function createHarness(name: string, overrides: AoOverrides = {}):
   });
   const d = module.deps();
   return {
+    module,
     db,
     ctx,
     d,

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

@@ -180,7 +180,7 @@ export function registerAoCapabilities(registry: CapabilityRegistry, deps: () =>
       throw new CapabilityError("precondition", `计划状态为 ${plan.status},不能提交`);
     }
     // 与自动路径同一幂等键:同一计划只会有一个 submitDownload 任务
-    const taskId = await enqueueTracked(d, {
+    const { taskId } = await enqueueTracked(d, {
       kind: "submitDownload",
       idempotencyKey: `plan:${plan.id}`,
       input: { acquisitionPlanId: plan.id },

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

@@ -1,4 +1,5 @@
 import { type EventPayload, EventType } from "@umt/contracts";
+import { sql } from "@umt/db";
 import type { ServerContext, ServerModule } from "../types.ts";
 import { registerAoCapabilities } from "./capabilities.ts";
 import { registerAoTasks } from "./handlers.ts";
@@ -9,10 +10,8 @@ import { registerDownloadRoutes } from "./routes/downloads.ts";
 import { registerLibraryRoutes } from "./routes/library.ts";
 import { registerOrganizationRoutes } from "./routes/organization.ts";
 import { registerRuleRoutes } from "./routes/rules.ts";
-import { type AoDeps, automationBlock } from "./services/common.ts";
-import { cleanupDownloadCache, sweepReservedPlans, syncDownloads } from "./services/downloads.ts";
-import { executeRecord, planImport } from "./services/organize.ts";
-import { reconcileLibrary } from "./services/reconcile.ts";
+import { type AoDeps, automationBlock, enqueueSweep, enqueueTracked, timeSlot } from "./services/common.ts";
+import { SWEEP_SLOT_MS, sweepReservedPlans, syncDownloads } from "./services/downloads.ts";
 import { loadVolumes, usedPercent } from "./services/volumes.ts";
 import { errMessage } from "./util.ts";
 
@@ -25,33 +24,49 @@ import { errMessage } from "./util.ts";
 
 export type AoOverrides = Partial<Omit<AoPorts, "options">> & { options?: Partial<AoOptions> };
 
-/** 模块工厂:测试注入 mock qB / 固定磁盘用量 / 假种子来源;正式装配用 acquisitionOrganizeModule */
+/** 超阈值自动清理的时间槽(同槽只入队一次) */
+const CLEANUP_SLOT_MS = 5 * 60_000;
+
+/**
+ * 模块工厂:测试注入 mock qB / 固定磁盘用量 / 假种子来源;正式装配用 acquisitionOrganizeModule。
+ *
+ * 后台只做编排,两个独立定时器互不阻塞:
+ *   sync  —— qB 同步 + 缓存阈值暂停/恢复(syncDownloads),只有轻量 HTTP 与 DB 写;
+ *   plan  —— 阈值判断后入队 cleanupDownloadCache、巡检补投 submitDownload / verifyAndImport、按周期入队 reconcileLibrary。
+ * 复制/删种/全量 stat 这些重活都进任务引擎(租约 + attemptToken),不在循环里 await。
+ */
 export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): ServerModule & { deps(): AoDeps } {
   let deps: AoDeps | null = null;
-  let loop: ReturnType<typeof setTimeout> | null = null;
+  const timers = new Set<ReturnType<typeof setTimeout>>();
   let stopped = false;
-  let lastReconcile = 0;
+  let lastReconcileSlot: string | null = null;
   // register* 阶段只保存 ctx;端口在首次使用时组装(ctx 其它字段此时才完整)
   const lazy = (ctx: ServerContext): AoDeps => {
     deps ??= { ctx, ports: createPorts(ctx, overrides) };
     return deps;
   };
 
-  /** worker 后台一轮:同步 qB → 阈值 → 必要时清理 → 补提交 → 补入库 → 低频资产对账 */
-  const tick = async (d: AoDeps) => {
-    await syncDownloads(d);
+  /** 编排一轮:阈值 → 入队清理;补投提交/入库;周期入队资产对账 */
+  const planTick = async (d: AoDeps) => {
     const blocked = await automationBlock(d);
     if (!blocked) {
       const cache = await d.ctx.settings.get("cache");
       const over = (await loadVolumes(d)).some((v) => (usedPercent(v) ?? 0) >= cache.hardThresholdPercent);
-      if (over) await cleanupDownloadCache(d);
+      if (over) await enqueueThresholdCleanup(d);
       await sweepReservedPlans(d);
       await sweepPendingImports(d);
     }
     const every = d.ports.options.reconcileIntervalMs;
-    if (every > 0 && Date.now() - lastReconcile >= every) {
-      lastReconcile = Date.now();
-      await reconcileLibrary(d);
+    if (every > 0) {
+      const slot = timeSlot(d.ports.now(), every);
+      if (slot !== lastReconcileSlot) {
+        await enqueueTracked(d, {
+          kind: "reconcileLibrary",
+          idempotencyKey: `auto:${slot}`,
+          input: {},
+        });
+        lastReconcileSlot = slot;
+      }
     }
   };
 
@@ -94,30 +109,64 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se
       const d = lazy(ctx);
       if (!d.ports.options.backgroundLoops) return;
       stopped = false;
-      const schedule = async () => {
-        if (stopped) return;
+      const intervalMs = async () => {
         const cache = await ctx.settings.get("cache").catch(() => null);
-        const ms = d.ports.options.syncIntervalMs ?? (cache?.checkIntervalSec ?? 60) * 1000;
-        loop = setTimeout(async () => {
-          try {
-            await tick(d);
-          } catch (err) {
-            ctx.logger.error({ err: { message: errMessage(err) } }, "acquisition-organize 后台轮询失败");
-          }
-          void schedule();
-        }, ms);
+        return d.ports.options.syncIntervalMs ?? (cache?.checkIntervalSec ?? 60) * 1000;
       };
-      void schedule();
+      /** 自调度循环:上一轮结束后再排下一轮,单个循环内不重入;两个循环互不等待 */
+      const loop = (name: string, fn: (d: AoDeps) => Promise<unknown>, period: () => Promise<number>) => {
+        const schedule = async () => {
+          if (stopped) return;
+          const t = setTimeout(
+            async () => {
+              timers.delete(t);
+              try {
+                await fn(d);
+              } catch (err) {
+                ctx.logger.error(
+                  { err: { message: errMessage(err) }, loop: name },
+                  "acquisition-organize 后台轮询失败",
+                );
+              }
+              void schedule();
+            },
+            await period(),
+          );
+          timers.add(t);
+          if (stopped) clearTimeout(t);
+        };
+        void schedule();
+      };
+      loop("sync", syncDownloads, intervalMs);
+      loop("plan", planTick, async () => d.ports.options.planIntervalMs ?? (await intervalMs()));
     },
     stop() {
       stopped = true;
-      if (loop) clearTimeout(loop);
+      for (const t of timers) clearTimeout(t);
+      timers.clear();
     },
   };
   return module;
 }
 
-/** 下载已完成、没有整理记录(入库事件被总开关延后等)的纳管任务:直接补做入库 */
+/** 超阈值:按时间槽入队一次 cleanupDownloadCache;上一次清理仍在进行则跳过 */
+async function enqueueThresholdCleanup(d: AoDeps): Promise<void> {
+  const busy = await d.ctx.db
+    .selectFrom("task")
+    .select("id")
+    .where("kind", "=", "cleanupDownloadCache")
+    .where("status", "in", ["pending", "running"])
+    .where("updated_at", ">", new Date(d.ports.now().getTime() - 30 * 60_000))
+    .executeTakeFirst();
+  if (busy) return;
+  await enqueueTracked(d, {
+    kind: "cleanupDownloadCache",
+    idempotencyKey: `threshold:${timeSlot(d.ports.now(), CLEANUP_SLOT_MS)}`,
+    input: {},
+  });
+}
+
+/** 下载已完成、没有整理记录(入库事件被总开关延后等)的纳管任务:入队 verifyAndImport(业务行 import:<id>) */
 async function sweepPendingImports(d: AoDeps): Promise<number> {
   const rows = await d.ctx.db
     .selectFrom("download_task as dt")
@@ -125,7 +174,8 @@ async function sweepPendingImports(d: AoDeps): Promise<number> {
     .where("dt.status", "=", "completed")
     .where("dt.managed", "=", true)
     .where("dt.removed_at", "is", null)
-    .where("dt.updated_at", "<", new Date(d.ports.now().getTime() - 5 * 60_000))
+    // 用 completed_at:qB 同步每轮都会写 download_task(updated_at 触发器随之刷新),按 updated_at 永远选不中
+    .where(sql<Date>`coalesce(dt.completed_at, dt.updated_at)`, "<", new Date(d.ports.now().getTime() - 5 * 60_000))
     .where((eb) =>
       eb.not(
         eb.exists(
@@ -135,14 +185,24 @@ async function sweepPendingImports(d: AoDeps): Promise<number> {
     )
     .limit(10)
     .execute();
+  const slot = timeSlot(d.ports.now(), SWEEP_SLOT_MS);
   let n = 0;
   for (const r of rows) {
     try {
-      const p = await planImport(d, { downloadTaskId: r.id, trigger: "auto", actorType: "system", actorId: null });
-      if (p.recordId && (p.status === "running" || p.status === "queued")) await executeRecord(d, p.recordId);
-      n++;
+      const q = await enqueueSweep(
+        d,
+        {
+          kind: "verifyAndImport",
+          idempotencyKey: `import:${r.id}`,
+          input: { downloadTaskId: r.id },
+          subjectType: "download_task",
+          subjectId: r.id,
+        },
+        slot,
+      );
+      if (q) n++;
     } catch (err) {
-      d.ctx.logger.warn({ downloadTaskId: r.id, err: { message: errMessage(err) } }, "补入库失败");
+      d.ctx.logger.warn({ downloadTaskId: r.id, err: { message: errMessage(err) } }, "补入库入队失败");
     }
   }
   return n;

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

@@ -23,6 +23,10 @@ export type { DownloaderInstanceRow, MediaServerInstanceRow, SiteRow, SourceRele
 export interface AoOptions {
   /** 下载同步/阈值巡检周期;缺省按 settings.cache.checkIntervalSec */
   syncIntervalMs?: number;
+  /** 编排循环(阈值清理入队/巡检补投/对账入队)周期;缺省同 syncIntervalMs */
+  planIntervalMs?: number;
+  /** 手动整理按路径递归遍历的文件数上限(超出截断;dryRun 返回 truncated,正式执行拒绝) */
+  walkMaxFiles: number;
   /** worker 启动时是否跑后台循环(测试里手动调用) */
   backgroundLoops: boolean;
   /** 下载中且字节不增长超过该时长 → 记证据 + 待处理(不换源) */
@@ -59,6 +63,7 @@ export const DEFAULT_OPTIONS: AoOptions = {
   newTitleWindowDays: 30,
   missingConfirmAfter: 3,
   reconcileIntervalMs: 6 * 3600_000,
+  walkMaxFiles: 5000,
 };
 
 const str = (v: unknown) => (typeof v === "string" && v.length > 0 ? v : undefined);
@@ -137,7 +142,7 @@ export function createPorts(
         ...(str(settings.username) ? { username: str(settings.username)! } : {}),
       });
     },
-    options: { ...DEFAULT_OPTIONS },
+    options: { ...DEFAULT_OPTIONS, walkMaxFiles: ctx.config?.ORGANIZE_WALK_MAX_FILES ?? DEFAULT_OPTIONS.walkMaxFiles },
   };
   const { options, ...rest } = overrides;
   return { ...ports, ...rest, options: { ...ports.options, ...options } };

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

@@ -2,10 +2,9 @@ import { type AcquisitionPlan, Api, type DownloadTask } from "@umt/contracts";
 import { z } from "zod";
 import { ApiException } from "../../../http/errors.ts";
 import type { ModuleApp } from "../../types.ts";
-import type { AoDeps } from "../services/common.ts";
-import { cleanupDownloadCache, userPause, userResume } from "../services/downloads.ts";
-import { reconcileLibrary } from "../services/reconcile.ts";
-import { notFound, toApi, toApiList } from "../util.ts";
+import { type AoDeps, enqueueTracked, timeSlot } from "../services/common.ts";
+import { userPause, userResume } from "../services/downloads.ts";
+import { actorOf, notFound, toApi, toApiList } from "../util.ts";
 import { mount, paginate } from "./helpers.ts";
 
 // 下载计划 / 下载任务(含用户暂停与恢复)/ 手动触发缓存清理与资产对账。
@@ -83,18 +82,28 @@ export function registerDownloadRoutes(app: ModuleApp, d: AoDeps) {
     { tag: TAG },
   );
 
+  // 清理/对账是重活(删种删数据、全量 stat):入队返回 {taskId, runId},由任务页 / SSE 追踪。
+  // 幂等键按分钟槽:连点只入队一次。
+  const accepted = z.object({ taskId: z.uuid(), runId: z.string() });
+  const minute = () => timeSlot(d.ports.now(), 60_000);
   mount(
     app,
     {
       method: "post",
       path: "/download-cache/cleanup",
       scope: "admin",
-      summary: "立即按阈值清理下载缓存(只清已验证入库且满足保留/H&R 的纳管任务)",
+      summary: "入队一轮缓存清理(只清已验证入库且满足保留/H&R 的纳管任务);返回任务 id,进度见任务页",
       body: z.object({ downloaderInstanceId: z.uuid().optional() }),
-      response: z.object({ report: z.record(z.string(), z.unknown()) }),
+      response: accepted,
     },
-    async (_c, { body }) => ({ report: await cleanupDownloadCache(d, body) }),
-    { tag: TAG, status: 200 },
+    async (c, { body }) =>
+      enqueueTracked(d, {
+        kind: "cleanupDownloadCache",
+        idempotencyKey: `manual:${body.downloaderInstanceId ?? "all"}:${minute()}`,
+        input: body.downloaderInstanceId ? { downloaderInstanceId: body.downloaderInstanceId } : {},
+        triggeredBy: actorOf(c),
+      }),
+    { tag: TAG, status: 202 },
   );
   mount(
     app,
@@ -102,14 +111,17 @@ export function registerDownloadRoutes(app: ModuleApp, d: AoDeps) {
       method: "post",
       path: "/library/reconcile",
       scope: "admin",
-      summary: "立即做一轮资产对账(库根哨兵 + 增量 stat;不删记录、不补下载)",
+      summary: "入队一轮资产对账(库根哨兵 + 增量 stat;不删记录、不补下载);返回任务 id,进度见任务页",
       body: z.object({ libraryRootId: z.uuid().optional() }),
-      response: z.object({ runId: z.uuid(), stats: z.record(z.string(), z.unknown()) }),
+      response: accepted,
     },
-    async (_c, { body }) => {
-      const r = await reconcileLibrary(d, body);
-      return { runId: r.runId, stats: r.stats };
-    },
-    { tag: TAG, status: 200 },
+    async (c, { body }) =>
+      enqueueTracked(d, {
+        kind: "reconcileLibrary",
+        idempotencyKey: `manual:${body.libraryRootId ?? "all"}:${minute()}`,
+        input: body.libraryRootId ? { libraryRootId: body.libraryRootId } : {},
+        triggeredBy: actorOf(c),
+      }),
+    { tag: TAG, status: 202 },
   );
 }

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

@@ -36,7 +36,7 @@ export function mount(
   app: ModuleApp,
   maybe: RouteDef | undefined,
   handler: Handler,
-  opts: { tag?: string; status?: 200 | 201 } = {},
+  opts: { tag?: string; status?: 200 | 201 | 202 } = {},
 ) {
   if (!maybe) throw new Error("contracts 未定义该端点");
   const def = maybe;

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

@@ -502,7 +502,7 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon
       subjectType: "acquisition_plan",
       subjectId: planId,
     }).then(
-      (taskId) => db.updateTable("acquisition_plan").set({ task_id: taskId }).where("id", "=", planId).execute(),
+      ({ 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 入队失败,留给巡检补投"),
     );

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

@@ -123,21 +123,56 @@ export async function markTask(
     .execute();
 }
 
-/** 建业务任务行 + 引擎入队(幂等键一致);返回 task.id */
-export async function enqueueTracked(d: AoDeps, t: Omit<TaskRowInput, "runId">): Promise<string> {
+/**
+ * 建业务任务行 + 引擎入队。引擎幂等键缺省 `<kind>:<idempotencyKey>`(与业务行一致);
+ * 巡检补投等需要"同一业务行、按时间槽重投"的场景传 engineKey。
+ */
+export async function enqueueTracked(
+  d: AoDeps,
+  t: Omit<TaskRowInput, "runId">,
+  opts: { engineKey?: string } = {},
+): Promise<{ taskId: string; runId: string }> {
   const taskId = await ensureTaskRow(d.ctx.db, t);
   const { runId } = await d.ctx.engine.enqueue(
     t.kind,
     { ...t.input, taskId },
     {
-      idempotencyKey: `${t.kind}:${t.idempotencyKey}`,
+      idempotencyKey: opts.engineKey ?? `${t.kind}:${t.idempotencyKey}`,
       traceId: taskId,
     },
   );
   await d.ctx.db.updateTable("task").set({ engine_run_id: runId }).where("id", "=", taskId).execute();
-  return taskId;
+  return { taskId, runId };
 }
 
+const ACTIVE_RUN = new Set(["pending", "running", "waiting"]);
+
+/**
+ * 巡检补投:同一业务行(kind + idempotencyKey)上次的引擎 run 仍在进行就跳过;
+ * 否则按时间槽(slot)生成引擎幂等键重投(引擎幂等键 24h 内有效,同槽重复巡检只入队一次)。
+ * 返回 null 表示跳过。
+ */
+export async function enqueueSweep(
+  d: AoDeps,
+  t: Omit<TaskRowInput, "runId">,
+  slot: string,
+): Promise<{ taskId: string; runId: string } | null> {
+  const prev = await d.ctx.db
+    .selectFrom("task")
+    .select(["engine_run_id"])
+    .where("kind", "=", t.kind)
+    .where("idempotency_key", "=", t.idempotencyKey)
+    .executeTakeFirst();
+  if (prev?.engine_run_id) {
+    const run = await d.ctx.engine.get(prev.engine_run_id).catch(() => null);
+    if (run && ACTIVE_RUN.has(run.status)) return null;
+  }
+  return enqueueTracked(d, t, { engineKey: `${t.kind}:${t.idempotencyKey}:${slot}` });
+}
+
+/** 时间槽标签:floor(now / periodMs) */
+export const timeSlot = (now: Date, periodMs: number) => String(Math.floor(now.getTime() / Math.max(1, periodMs)));
+
 // ── 实例拓扑:路径映射、保存路径、缓存卷 ──
 
 export interface InstanceTopology {

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

@@ -10,10 +10,12 @@ import {
   automationBlock,
   downloaderPathOf,
   type Exec,
+  enqueueSweep,
   hostPathOf,
   type InstanceTopology,
   loadInstances,
   loadLibraryRoots,
+  timeSlot,
   upsertDecision,
 } from "./common.ts";
 import { hnrSatisfied, retentionElapsed } from "./retention.ts";
@@ -260,6 +262,13 @@ async function failPlan(d: AoDeps, planId: string, slotId: string, reason: strin
 }
 
 /** 巡检:reserved 但没有下载任务的计划(入队失败、被阈值/总开关延后)→ 条件满足时直接补提交 */
+/** 巡检补投的时间槽:同一计划/下载在一个槽内最多重投一次 */
+export const SWEEP_SLOT_MS = 5 * 60_000;
+
+/**
+ * 已预留但一直没有下载任务的计划:入队 submitDownload(与自动路径同一业务行 plan:<id>),不在巡检里直接提交。
+ * 上次的 run 仍在进行则跳过;返回本轮入队数。
+ */
 export async function sweepReservedPlans(d: AoDeps): Promise<number> {
   if (await automationBlock(d)) return 0;
   const stale = await d.ctx.db
@@ -271,13 +280,32 @@ export async function sweepReservedPlans(d: AoDeps): Promise<number> {
     .where("p.updated_at", "<", new Date(d.ports.now().getTime() - 60_000))
     .limit(20)
     .execute();
+  const slot = timeSlot(d.ports.now(), SWEEP_SLOT_MS);
   let n = 0;
   for (const p of stale) {
-    const r = await submitDownload(d, p.id).catch((err) => {
-      d.ctx.logger.warn({ planId: p.id, err: { message: errMessage(err) } }, "补提交失败");
-      return null;
-    });
-    if (r?.status === "submitted") n++;
+    try {
+      const r = await enqueueSweep(
+        d,
+        {
+          kind: "submitDownload",
+          idempotencyKey: `plan:${p.id}`,
+          input: { acquisitionPlanId: p.id },
+          subjectType: "acquisition_plan",
+          subjectId: p.id,
+        },
+        slot,
+      );
+      if (!r) continue;
+      n++;
+      await d.ctx.db
+        .updateTable("acquisition_plan")
+        .set({ task_id: r.taskId })
+        .where("id", "=", p.id)
+        .where("task_id", "is", null)
+        .execute();
+    } catch (err) {
+      d.ctx.logger.warn({ planId: p.id, err: { message: errMessage(err) } }, "补提交入队失败");
+    }
   }
   return n;
 }

apps/web/app/api/downloads.ts

@@ -102,14 +102,14 @@ export function useCacheVolumes() {
   });
 }
 
-/** 手动清理:只清已验证入库且满足保留/H&R 的纳管任务 */
+/** 手动清理:服务端入队 cleanupDownloadCache(只清已验证入库且满足保留/H&R 的纳管任务),返回任务 id */
 export function useCacheCleanup() {
   const qc = useQueryClient();
   return useMutation({
     mutationFn: (downloaderInstanceId?: string) =>
       apiMutate("/api/v1/download-cache/cleanup", {
         body: downloaderInstanceId ? { downloaderInstanceId } : {},
-      }).then((d) => z.object({ report: z.record(z.string(), z.unknown()) }).parse(d)),
+      }).then((d) => z.object({ taskId: z.string(), runId: z.string() }).parse(d)),
     onSuccess: () => qc.invalidateQueries({ queryKey: downloadKeys.all }),
   });
 }

apps/web/app/components/downloads/cache-volumes.tsx

@@ -1,5 +1,6 @@
 import type { Dashboard } from "@umt/contracts";
 import { Brush, HardDrive } from "lucide-react";
+import { useNavigate } from "react-router";
 import { toast } from "sonner";
 import { useCacheCleanup, useCacheVolumes } from "~/api/downloads.ts";
 import { ConfirmDialog } from "~/components/console/confirm-dialog.tsx";
@@ -79,6 +80,7 @@ function VolumeBar({ v }: { v: Volume }) {
 export function CacheVolumes() {
   const vols = useCacheVolumes();
   const cleanup = useCacheCleanup();
+  const navigate = useNavigate();
   return (
     <section className="pg-slab mb-6 rounded-xl p-4">
       <div className="mb-3 flex flex-wrap items-center justify-between gap-2">
@@ -96,10 +98,8 @@ export function CacheVolumes() {
           onConfirm={() =>
             cleanup.mutateAsync(undefined).then(
               (r) => {
-                const vols = (r.report as { volumes?: { removed?: unknown[]; skipped?: unknown[] }[] }).volumes ?? [];
-                const removed = vols.reduce((n, v) => n + (v.removed?.length ?? 0), 0);
-                const skipped = vols.reduce((n, v) => n + (v.skipped?.length ?? 0), 0);
-                toast.success(`清理完成:移除 ${removed} 个任务的缓存${skipped ? `,${skipped} 个因保留/H&R 跳过` : ""}`);
+                toast.success(`已排队清理任务 ${r.taskId.slice(0, 8)}`, { description: "后台执行,结果见任务详情" });
+                navigate(`/console/engine/${r.runId}`);
               },
               (e) => toast.error(errorMessage(e)),
             )