merge: fix/h-visibility

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

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

@@ -15,7 +15,7 @@ import { registerNamingRoutes } from "./routes/naming.ts";
 import { registerOrganizationRoutes } from "./routes/organization.ts";
 import { registerRuleRoutes } from "./routes/rules.ts";
 import { type AoDeps, automationBlock, enqueueSweep, enqueueTracked, timeSlot } from "./services/common.ts";
-import { SWEEP_SLOT_MS, sweepReservedPlans, syncDownloads } from "./services/downloads.ts";
+import { reportVolumeStates, SWEEP_SLOT_MS, sweepReservedPlans, syncDownloads } from "./services/downloads.ts";
 import { recoverStalledRecords } from "./services/organize-records.ts";
 import { sweepDueSlots, sweepMetadataRefresh, sweepSubscriptionSearch } from "./services/subscriptions.ts";
 import { organizeDecisionHook } from "./services/takeover.ts";
@@ -139,11 +139,17 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se
         },
         run: guarded(d, async (d) => {
           const cache = await d.ctx.settings.get("cache");
-          const over = (await loadVolumes(d)).some(
-            (v) => (usedPercent(v) ?? 0) >= volumeThresholds(v, cache).hardThresholdPercent,
-          );
-          if (!over) return "未超阈值";
-          return (await enqueueThresholdCleanup(d)) ? "已入队清理" : "清理进行中";
+          const vols = await loadVolumes(d);
+          // 读不到容量的卷不当 0%:开待处理(同步每轮也会判定),摘要里写明
+          await reportVolumeStates(d, vols);
+          const unknown = vols.filter((v) => usedPercent(v) === null).length;
+          const note = unknown ? `，${unknown} 个下载目录读不到容量` : "";
+          const over = vols.some((v) => {
+            const pct = usedPercent(v);
+            return pct !== null && pct >= volumeThresholds(v, cache).hardThresholdPercent;
+          });
+          if (!over) return `未超阈值${note}`;
+          return ((await enqueueThresholdCleanup(d)) ? "已入队清理" : "清理进行中") + note;
         }),
       });
       registerScheduledJob(ctx, {

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

@@ -117,12 +117,42 @@ export const DEFAULT_OPTIONS: AoOptions = {
 
 const str = (v: unknown) => (typeof v === "string" && v.length > 0 ? v : undefined);
 
+/** qB 客户端按 ServerContext 共享缓存(实例配置变了重建):下载同步、面板、Agent 的 download.recheck 共用同一登录会话 */
+const sharedClients = new WeakMap<ServerContext, Map<string, { stamp: string; client: DownloaderClient }>>();
+
+/** 按实例取 qB 客户端(凭证现场经 secrets 解出,解析失败按无凭证处理并记警告) */
+export async function downloaderClientFor(ctx: ServerContext, inst: DownloaderInstanceRow): Promise<DownloaderClient> {
+  let clients = sharedClients.get(ctx);
+  if (!clients) {
+    clients = new Map();
+    sharedClients.set(ctx, clients);
+  }
+  const stamp = `${inst.updated_at.toISOString()}|${inst.base_url}`;
+  const hit = clients.get(inst.id);
+  if (hit && hit.stamp === stamp) return hit.client;
+  const settings = (inst.settings ?? {}) as Record<string, unknown>;
+  let secret: string | undefined;
+  if (inst.secret_ref) {
+    try {
+      secret = await ctx.secrets.resolve(inst.secret_ref);
+    } catch (err) {
+      ctx.logger.warn({ err: { message: (err as Error).message } }, "凭证解析失败");
+    }
+  }
+  const client = createQbClient({
+    instanceId: inst.id,
+    baseUrl: inst.base_url,
+    ...qbAuthConfig({ authKind: settings.authKind, username: inst.username, secret: secret ?? null }),
+  });
+  clients.set(inst.id, { stamp, client });
+  return client;
+}
+
 /** 用 ctx 组装默认端口;overrides 逐项覆盖(测试) */
 export function createPorts(
   ctx: ServerContext,
   overrides: Partial<Omit<AoPorts, "options">> & { options?: Partial<AoOptions> } = {},
 ): AoPorts {
-  const clients = new Map<string, { stamp: string; client: DownloaderClient }>();
   const resolveSecret = async (ref: string | null): Promise<string | undefined> => {
     if (!ref) return undefined;
     try {
@@ -159,20 +189,7 @@ export function createPorts(
       if (r.created) await enqueueNewEntityRefresh(ctx, r.mediaEntityId);
       return r.mediaEntityId;
     },
-    async downloader(inst) {
-      const stamp = `${inst.updated_at.toISOString()}|${inst.base_url}`;
-      const hit = clients.get(inst.id);
-      if (hit && hit.stamp === stamp) return hit.client;
-      const settings = (inst.settings ?? {}) as Record<string, unknown>;
-      const secret = await resolveSecret(inst.secret_ref);
-      const client = createQbClient({
-        instanceId: inst.id,
-        baseUrl: inst.base_url,
-        ...qbAuthConfig({ authKind: settings.authKind, username: inst.username, secret: secret ?? null }),
-      });
-      clients.set(inst.id, { stamp, client });
-      return client;
-    },
+    downloader: (inst) => downloaderClientFor(ctx, inst),
     async torrentSource({ release, site }) {
       // 永久性条件用 SiteError:submitDownload 据此把计划判失败并释放 Slot,而不是反复重试、Slot 一直 reserved
       if (!site) throw new SiteError("", "not_found", "站点不存在，无法获取种子");

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

@@ -124,7 +124,7 @@ export function instanceLive(d: AoDeps, inst: DownloaderInstanceRow): Promise<In
 
 /**
  * 还没提交到下载器、在等条件的计划(DL-3):草案且有待批准的审批(等批准);已预留但还没有下载任务
- * (等下载盘空间 / 自动化开启 / 提交任务执行)。没有审批的草案(智能体只生成没提交)不算在等。
+ * (等下载盘空间 / 下载目录读不到容量 / 自动化开启 / 提交任务执行)。没有审批的草案(智能体只生成没提交)不算在等。
  */
 export function waitingPlansQuery(db: Exec, instanceId?: string) {
   return db
@@ -161,7 +161,9 @@ export function waitingPlansQuery(db: Exec, instanceId?: string) {
 export function planWaitOf(status: string, deferred: string | null): PlanWait {
   if (status === "draft") return "approval";
   if (!deferred) return "queued";
-  if (deferred.startsWith("cache:") || deferred === "volume_state_unknown") return "space";
+  // 下载目录读不到容量是配置问题(路径映射 / 挂载),不是空间不够
+  if (deferred === "volume_state_unknown" || deferred === "cache:volume_state_unknown") return "volume_unknown";
+  if (deferred.startsWith("cache:")) return "space";
   if (deferred === "automation_disabled" || deferred === "read_only_reconcile") return "automation_off";
   return "queued";
 }

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

@@ -2,6 +2,7 @@ import {
   assertSafeToDeleteData,
   type DownloaderClient,
   DownloaderError,
+  type QbFile,
   type QbTorrent,
   SiteError,
   type TorrentSource,
@@ -268,6 +269,8 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
   const site = await db.selectFrom("site").selectAll().where("id", "=", rel.site_id).executeTakeFirst();
   const client = await d.ports.downloader(topo.inst);
   const tag = topo.inst.managed_tag;
+  // 季包 / 多集包要按缺集选文件:先不让它开始下载(对照 MP QK1),选完文件再恢复
+  const pack = needsPackSelection(plan, rel);
   // 评估时定下的保存路径(订阅 downloadSavePath 经路径映射后的下载器侧路径),没有用实例默认
   const savePath = ev.savePath || topo.savePath;
 
@@ -280,6 +283,10 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
   });
   let dtId: string | undefined;
   let inUseBy: string | undefined;
+  /** commit 时计划已不是 reserved(提交期间被取消等):不落任务行,事后撤销 qB 里的任务 */
+  let lostPlan: string | undefined;
+  /** 已存在的任务原本没有管理标签、是本次补上的 */
+  let tagAdded = false;
   const result = await withAttemptToken<SubmitResult>(
     db,
     `submit:${plan.id}`,
@@ -304,6 +311,8 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
         savePath,
         tags: [tag],
         ...(topo.category ? { category: topo.category } : {}),
+        // 磁力暂停添加时 qB 不取元数据:改用「拿到元数据后停止」;.torrent / URL 直接暂停添加
+        ...(pack ? (source.kind === "magnet" ? { stopCondition: "MetadataReceived" as const } : { paused: true }) : {}),
       });
       switch (out.kind) {
         case "added": {
@@ -314,7 +323,10 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
         }
         case "already_exists": {
           // 原本已存在:接管,只补管理标签,不改保存路径/其它标签/文件选择
-          if (!out.existing.tags.includes(tag)) await client.addTags([out.hash], [tag]);
+          if (!out.existing.tags.includes(tag)) {
+            await client.addTags([out.hash], [tag]);
+            tagAdded = true;
+          }
           return { kind: "already_exists", ...view(out.existing), tags: [...new Set([...out.existing.tags, tag])] };
         }
         case "accepted_unconfirmed":
@@ -336,6 +348,17 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
     {
       ...(opts.runId ? { runId: opts.runId } : {}),
       commit: async (trx, res) => {
+        // 复核计划仍是 reserved(取消接口同样锁计划行):qB 回查期间被取消的,不把它写回 submitted
+        const cur = await trx
+          .selectFrom("acquisition_plan")
+          .select("status")
+          .where("id", "=", plan.id)
+          .forUpdate()
+          .executeTakeFirst();
+        if (cur?.status !== "reserved") {
+          lostPlan = cur?.status ?? "missing";
+          return;
+        }
         if (res.kind === "rejected" || res.kind === "source_failed") {
           const r = res;
           await failPlanTx(d, trx, plan.id, plan.slot_id, `${r.kind}: ${r.reason}`);
@@ -402,8 +425,9 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
               tags: r.tags,
               managed: true,
               origin: r.kind === "already_exists" ? "adopted" : "submitted",
-              status: r.kind === "accepted_unconfirmed" ? "submitting" : "queued",
-              pause_reason: null,
+              status:
+                r.kind === "accepted_unconfirmed" ? "submitting" : pack && r.kind === "added" ? "paused" : "queued",
+              pause_reason: pack && r.kind === "added" ? "policy" : null,
               removed_at: null,
               completed_at: null,
               no_progress_since: null,
@@ -420,12 +444,13 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
           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)
+                                     save_path, content_path, category, tags, managed, origin, status, pause_reason, size_bytes, next_check_at)
           values (${topo.inst.id}, ${r.hash}, ${r.hash.length === 40 ? r.hash : null}, ${r.hash.length === 64 ? r.hash : null},
                   ${plan.id}, ${ev.manual ? plan.id : null}, ${rel.id}, coalesce((select media_entity_id from acquisition_slot where id = ${plan.slot_id}), ${ev.mediaEntityId ?? rel.media_entity_id}::uuid),
                   ${r.name}, ${r.savePath}, ${r.contentPath}, ${topo.category},
                   ${r.tags}, true, ${r.kind === "already_exists" ? "adopted" : "submitted"},
-                  ${r.kind === "accepted_unconfirmed" ? "submitting" : "queued"}, ${rel.size_bytes}, now())
+                  ${r.kind === "accepted_unconfirmed" ? "submitting" : pack && r.kind === "added" ? "paused" : "queued"},
+                  ${pack && r.kind === "added" ? "policy" : null}, ${rel.size_bytes}, now())
           on conflict (instance_id, torrent_hash) do update
             set acquisition_plan_id = coalesce(download_task.acquisition_plan_id, excluded.acquisition_plan_id),
                 import_plan_id = coalesce(excluded.import_plan_id, download_task.import_plan_id),
@@ -435,9 +460,17 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
           returning id, (xmax = 0) as inserted`.execute(trx);
           dtId = ins.rows[0]!.id;
         }
+        // 暂停添加的季包:选文件挂起,选完(或元数据超时整包)由提交后续步骤 / 同步恢复下载
+        const pendingSel = pack && r.kind !== "already_exists";
         await trx
           .updateTable("acquisition_plan")
-          .set({ status: "submitted", status_reason: r.kind })
+          .set({
+            status: "submitted",
+            status_reason: r.kind,
+            ...(pendingSel
+              ? { file_selection: jsonb({ mode: "pending", since: d.ports.now().toISOString(), attempts: 0 }) }
+              : {}),
+          })
           .where("id", "=", plan.id)
           .execute();
         await applySlotEvent(
@@ -456,14 +489,18 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
       },
     },
   );
+  if (lostPlan) {
+    await undoLostSubmit(d, client, result, tag, tagAdded, plan.id, lostPlan);
+    return { status: "skipped", reason: `plan_${lostPlan}` };
+  }
   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) {
-    await selectPackFiles(d, client, hash, plan, rel).catch((err) =>
-      d.ctx.logger.warn({ planId: plan.id, err: { message: errMessage(err) } }, "季包选文件失败，按整包下载"),
-    );
+  // 季包/多集包:按缺集选文件再恢复下载(接管的已有任务不改文件选择);文件清单还拿不到(磁力元数据未就绪)
+  // 时留给同步重试,失败不影响提交结果
+  if (pack && result.kind !== "already_exists") {
+    const fresh = await db.selectFrom("acquisition_plan").selectAll().where("id", "=", plan.id).executeTakeFirst();
+    if (fresh && isPendingSelection(fresh)) await settlePackFiles(d, client, hash, fresh, rel);
   }
   if (result.kind !== "accepted_unconfirmed") await afterDownloadAccepted(d, plan.id);
   if (!dtId) {
@@ -495,11 +532,150 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
 }
 
 const PACK_FILE_LIST = { retries: 3, intervalMs: 300 } as const;
+/** 季包文件清单(磁力元数据)最多等多久;超时仍拿不到就整包下载并记录原因 */
+export const PACK_METADATA_TIMEOUT_MS = 10 * 60_000;
+
+/** 资源是季包(没有集号也没有日期)或多集包 */
+function isPackRelease(rel: SourceReleaseRow): boolean {
+  const parsed = parsedOf(rel);
+  const isPack = !parsed.episodes && !parsed.airDate;
+  const multi = !!parsed.episodes && parsed.episodes.end > parsed.episodes.start;
+  return isPack || multi;
+}
+
+/** 订阅计划下季包 / 多集包、还没选过文件:提交时暂停添加,选完文件再恢复 */
+function needsPackSelection(plan: AcquisitionPlanRow, rel: SourceReleaseRow): boolean {
+  return !!plan.slot_id && plan.unit_keys.length > 0 && plan.file_selection === null && isPackRelease(rel);
+}
+
+interface PendingSelection {
+  mode: "pending";
+  since: string;
+  attempts: number;
+}
+
+const isPendingSelection = (plan: Pick<AcquisitionPlanRow, "file_selection">) =>
+  (plan.file_selection as { mode?: string } | null)?.mode === "pending";
 
 /**
- * 季包按缺集选文件(对照 MP 从整季包按集选文件,仅 qB):提交后读文件清单,把不需要单元的文件优先级置 0;
+ * 暂停添加的季包:读文件清单 → 选文件 → 恢复下载 → 记录选择。
+ * 文件清单还拿不到(磁力元数据未就绪)且没到 PACK_METADATA_TIMEOUT_MS:只累加重试次数,留给下一轮同步;
+ * 超时仍拿不到:整包下载并记 file_list_timeout。恢复下载失败时选择保持 pending,下一轮同步重做(选文件幂等)。
+ * 返回 true 表示已收尾(不再 pending)。
+ */
+async function settlePackFiles(
+  d: AoDeps,
+  client: DownloaderClient,
+  hash: string,
+  plan: AcquisitionPlanRow,
+  rel: SourceReleaseRow,
+): Promise<boolean> {
+  const pending = plan.file_selection as unknown as PendingSelection;
+  const now = d.ports.now();
+  const waitedMs = now.getTime() - new Date(pending.since).getTime();
+  const files = await client.files(hash, PACK_FILE_LIST).catch(() => [] as QbFile[]);
+  const record = (sel: Record<string, unknown>, unitKeys?: string[]) =>
+    d.ctx.db
+      .updateTable("acquisition_plan")
+      .set({ file_selection: jsonb(sel), ...(unitKeys ? { unit_keys: unitKeys } : {}) })
+      .where("id", "=", plan.id)
+      .execute();
+  if (files.length === 0 && waitedMs < PACK_METADATA_TIMEOUT_MS) {
+    await record({ ...pending, attempts: pending.attempts + 1 });
+    return false;
+  }
+  let sel: { selection: Record<string, unknown>; unitKeys?: string[] };
+  if (files.length === 0) {
+    sel = {
+      selection: { mode: "whole", reason: "file_list_timeout", waitedSec: Math.round(waitedMs / 1000) },
+    };
+  } else {
+    try {
+      sel = await selectPackFiles(d, client, hash, plan, rel, files);
+    } catch (err) {
+      d.ctx.logger.warn({ planId: plan.id, err: { message: errMessage(err) } }, "季包选文件失败，按整包下载");
+      sel = { selection: { mode: "whole", reason: "select_failed", error: errMessage(err).slice(0, 300) } };
+    }
+  }
+  try {
+    await client.resume([hash]);
+  } catch (err) {
+    d.ctx.logger.warn(
+      { planId: plan.id, err: { message: errMessage(err) } },
+      "季包选完文件后恢复下载失败，下一轮同步重试",
+    );
+    return false;
+  }
+  await record(sel.selection, sel.unitKeys);
+  return true;
+}
+
+/** 同步时收尾挂起的季包选文件(本轮在 qB 里看得到的任务) */
+async function settlePendingPacks(d: AoDeps, client: DownloaderClient, topo: InstanceTopology, seen: Set<string>) {
+  const rows = await d.ctx.db
+    .selectFrom("acquisition_plan as p")
+    .innerJoin("download_task as dt", "dt.acquisition_plan_id", "p.id")
+    .selectAll("p")
+    .select("dt.torrent_hash")
+    .where("dt.instance_id", "=", topo.inst.id)
+    .where("dt.removed_at", "is", null)
+    .where("dt.status", "!=", "removed")
+    .where("p.status", "=", "submitted")
+    .where(sql<string>`p.file_selection->>'mode'`, "=", "pending")
+    .execute();
+  for (const { torrent_hash, ...plan } of rows) {
+    if (!seen.has(torrent_hash)) continue;
+    try {
+      const rel = await d.ctx.db
+        .selectFrom("source_release")
+        .selectAll()
+        .where("id", "=", plan.source_release_id)
+        .executeTakeFirstOrThrow();
+      await settlePackFiles(d, client, torrent_hash, plan as AcquisitionPlanRow, rel);
+    } catch (err) {
+      d.ctx.logger.warn({ planId: plan.id, err: { message: errMessage(err) } }, "季包选文件重试失败");
+    }
+  }
+}
+
+/**
+ * 提交期间计划被取消(commit 复核不是 reserved):任务行没落,qB 里刚加的任务暂停、去掉管理标签、
+ * 打上 <管理标签>-cancelled 留给人删(不自动删数据);原本就在 qB 里的任务只撤掉本次补上的管理标签。
+ */
+async function undoLostSubmit(
+  d: AoDeps,
+  client: DownloaderClient,
+  result: SubmitResult,
+  tag: string,
+  tagAdded: boolean,
+  planId: string,
+  planStatus: string,
+) {
+  if (!("hash" in result)) return;
+  try {
+    if (result.kind === "already_exists") {
+      if (tagAdded) await client.removeTags([result.hash], [tag]);
+    } else {
+      await client.pause([result.hash]);
+      await client.removeTags([result.hash], [tag]);
+      await client.addTags([result.hash], [`${tag}-cancelled`]);
+    }
+    d.ctx.logger.warn(
+      { planId, planStatus, hash: result.hash, kind: result.kind },
+      "提交期间计划已结束，已撤下 qB 里的任务（暂停并去掉管理标签）",
+    );
+  } catch (err) {
+    d.ctx.logger.error(
+      { planId, planStatus, hash: result.hash, err: { message: errMessage(err) } },
+      "提交期间计划已结束，撤下 qB 任务失败，需要到 qB 里手动处理",
+    );
+  }
+}
+
+/**
+ * 季包按缺集选文件(对照 MP 从整季包按集选文件,仅 qB):按文件清单把不需要单元的文件优先级置 0;
  * 需要单元 = 计划预留的集(季通配 = 该季未覆盖、且不在其它在途计划里的集)。
- * 文件清单拿不到 / 有视频文件认不出集号 → 整包下载并记录理由;计划的 unit_keys 收敛为实际选中的集。
+ * 有视频文件认不出集号 → 整包下载并记录理由;计划的 unit_keys 收敛为实际选中的集。返回要记录的选择(由调用方写)。
  */
 async function selectPackFiles(
   d: AoDeps,
@@ -507,20 +683,12 @@ async function selectPackFiles(
   hash: string,
   plan: AcquisitionPlanRow,
   rel: SourceReleaseRow,
-): Promise<void> {
-  if (!plan.slot_id) return;
+  files: QbFile[],
+): Promise<{ selection: Record<string, unknown>; unitKeys?: string[] }> {
   const parsed = parsedOf(rel);
-  const isPack = !parsed.episodes && !parsed.airDate;
-  const multi = !!parsed.episodes && parsed.episodes.end > parsed.episodes.start;
-  if (!isPack && !multi) return;
-  const record = (sel: Record<string, unknown>, unitKeys?: string[]) =>
-    d.ctx.db
-      .updateTable("acquisition_plan")
-      .set({ file_selection: jsonb(sel), ...(unitKeys ? { unit_keys: unitKeys } : {}) })
-      .where("id", "=", plan.id)
-      .execute();
-  const files = await client.files(hash, PACK_FILE_LIST).catch(() => []);
-  if (files.length === 0) return void (await record({ mode: "whole", reason: "file_list_unavailable" }));
+  const whole = (selection: Record<string, unknown>) => ({ selection: { mode: "whole", ...selection } });
+  const slotId = plan.slot_id;
+  if (!slotId) return whole({ reason: "no_slot" });
 
   // 季通配:展开为"该季未覆盖且不在其它在途计划里"
   const wildSeasons = new Set(
@@ -535,7 +703,7 @@ async function selectPackFiles(
     const slot = await d.ctx.db
       .selectFrom("acquisition_slot")
       .select(["id", "current_asset_id", "media_entity_id"])
-      .where("id", "=", plan.slot_id)
+      .where("id", "=", slotId)
       .executeTakeFirstOrThrow();
     const slots = await d.ctx.db
       .selectFrom("acquisition_slot")
@@ -543,7 +711,7 @@ async function selectPackFiles(
       .where("media_entity_id", "=", slot.media_entity_id)
       .execute();
     const covered = await slotCoveredUnits(d.ctx.db, slot, { entityWide: slots.length <= 1 });
-    const others = await slotInflightPlans(d.ctx.db, plan.slot_id, plan.id);
+    const others = await slotInflightPlans(d.ctx.db, slotId, plan.id);
     blocked = new Set([...covered.keys, ...others.flatMap((o) => o.unitKeys)]);
   }
   const wantedKey = (k: string) => {
@@ -588,13 +756,12 @@ async function selectPackFiles(
   const videos = rows.filter((r) => r.kind === "video");
   const unmapped = videos.filter((r) => r.keys.length === 0 && !r.special);
   if (videos.length === 0 || unmapped.length > 0)
-    return void (await record({
-      mode: "whole",
+    return whole({
       reason: videos.length === 0 ? "no_video_files" : "unmapped_video_file",
       files: unmapped.slice(0, 20).map((r) => r.name),
-    }));
+    });
   const selected = [...new Set(videos.filter((r) => r.wanted).flatMap((r) => r.keys.filter(wantedKey)))];
-  if (selected.length === 0) return void (await record({ mode: "whole", reason: "no_missing_units_in_pack" }));
+  if (selected.length === 0) return whole({ reason: "no_missing_units_in_pack" });
   // 视频不需要 → 不下;附属文件(字幕等)有集号且不需要 → 不下;无集号的附属文件保留
   const skip = rows.filter((r) => (r.kind === "video" ? !r.wanted : r.keys.length > 0 && !r.wanted));
   if (skip.length > 0)
@@ -603,15 +770,15 @@ async function selectPackFiles(
       skip.map((r) => r.index),
       0,
     );
-  await record(
-    {
+  return {
+    selection: {
       mode: skip.length > 0 ? "selected" : "all_needed",
       selectedUnits: selected,
       skippedFiles: skip.slice(0, 200).map((r) => r.name),
       totalFiles: files.length,
     },
-    selected,
-  );
+    unitKeys: selected,
+  };
 }
 
 async function failPlanTx(d: AoDeps, trx: Exec, planId: string, slotId: string | null, reason: string) {
@@ -716,8 +883,16 @@ export interface SyncStats {
   removed: number;
   thresholdPaused: number;
   thresholdResumed: number;
+  /** 本轮首次在 qB 里确认的「已收单未确认」提交的计划(同步后补调 afterDownloadAccepted) */
+  confirmedPlans: string[];
 }
 
+/** 下载器连续不可达多久开 downloader_offline 待处理 */
+export const DOWNLOADER_OFFLINE_DECISION_MS = 10 * 60_000;
+
+/** 在途(还没下完)的任务状态:停用的实例上有这些任务时仍同步 */
+const INFLIGHT_STATUSES: DownloadStatus[] = ["queued", "submitting", "downloading", "stalled", "paused"];
+
 /** 连续多少轮同步(列表 + 按 hash 查)都查不到才判 removed;与资产对账的 missingConfirmAfter 同口径 */
 export const MISSING_CONFIRM_ROUNDS = 3;
 
@@ -731,23 +906,44 @@ export async function syncDownloads(d: AoDeps): Promise<SyncStats> {
     removed: 0,
     thresholdPaused: 0,
     thresholdResumed: 0,
+    confirmedPlans: [],
   };
   const insts = await loadInstances(d.ctx.db);
   for (const topo of insts) {
     stats.instances++;
     await syncInstance(d, topo, stats);
   }
+  // 停用的实例上还有在途任务:只同步这些任务的状态(不接管新任务、不开离线事项、不计入失联统计)
+  const enabled = new Set(insts.map((t) => t.inst.id));
+  for (const topo of await loadInstances(d.ctx.db, { enabledOnly: false })) {
+    if (enabled.has(topo.inst.id)) continue;
+    const inflight = await d.ctx.db
+      .selectFrom("download_task")
+      .select("id")
+      .where("instance_id", "=", topo.inst.id)
+      .where("removed_at", "is", null)
+      .where("managed", "=", true)
+      .where("status", "in", INFLIGHT_STATUSES)
+      .executeTakeFirst();
+    if (inflight) await syncInstance(d, topo, stats, { observeOnly: true });
+  }
+  for (const planId of stats.confirmedPlans) {
+    await afterDownloadAccepted(d, planId).catch((err) =>
+      d.ctx.logger.warn({ planId, err: { message: errMessage(err) } }, "补记订阅进度失败"),
+    );
+  }
   await enforceThresholds(d, insts, stats);
   return stats;
 }
 
-async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats) {
+async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats, opts: { observeOnly?: boolean } = {}) {
   const db = d.ctx.db;
   const now = d.ports.now();
-  const client = await d.ports.downloader(topo.inst);
-  const sourceTags = await qbSourceTags(d, topo);
   let list: QbTorrent[];
+  let client: DownloaderClient;
   try {
+    client = await d.ports.downloader(topo.inst);
+    const sourceTags = await qbSourceTags(d, topo);
     const tags = [...new Set([topo.inst.managed_tag, ...sourceTags.map((s) => s.config.tag!)])];
     const batches = await Promise.all(tags.map((tag) => client.list({ tag })));
     list = [...new Map(batches.flat().map((t) => [t.hash, t])).values()].filter(
@@ -755,13 +951,15 @@ async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats)
     );
   } catch (err) {
     // 下载器失联:不改任何任务状态(失败 ≠ 不存在)
-    stats.unreachable.push(topo.inst.id);
     await db
       .updateTable("downloader_instance")
       .set({ health_status: "offline" })
       .where("id", "=", topo.inst.id)
       .execute();
     d.ctx.logger.warn({ instanceId: topo.inst.id, err: { message: errMessage(err) } }, "qB 同步失败");
+    if (opts.observeOnly) return;
+    stats.unreachable.push(topo.inst.id);
+    await reportDownloaderOffline(d, topo, now, err);
     return;
   }
   await db
@@ -769,6 +967,7 @@ async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats)
     .set({ health_status: "healthy", last_seen_at: now })
     .where("id", "=", topo.inst.id)
     .execute();
+  await closeRecoveredDecision(d, db, `downloader_offline:${topo.inst.id}`, "下载器已恢复连接");
   const rows = await db
     .selectFrom("download_task")
     .selectAll()
@@ -783,10 +982,11 @@ async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats)
     const row = byHash.get(t.hash);
     if (row) {
       if (await observe(d, row, t, now, stats)) stats.updated++;
-    } else {
+    } else if (!opts.observeOnly) {
       await adopt(d, topo, t, now, stats);
     }
   }
+  await settlePendingPacks(d, client, topo, seen);
   for (const row of rows) {
     if (seen.has(row.torrent_hash) || !row.managed) continue;
     let still: QbTorrent | null;
@@ -920,9 +1120,21 @@ async function observeLocked(
     );
   // 卡住 / 异常后恢复正常(进度增长或状态恢复):关闭同一任务未关闭的无进展 / 异常事项
   if (changed && RECOVERED.has(merged.status)) {
-    if (row.status === "stalled") await closeRecoveredDecision(d, trx, `download_stalled:${row.id}`, merged.status);
+    const note = `下载已恢复（${merged.status}）`;
+    if (row.status === "stalled") await closeRecoveredDecision(d, trx, `download_stalled:${row.id}`, note);
     if (row.status === "failed" || row.status === "missing")
-      await closeRecoveredDecision(d, trx, `download_failed:${row.id}`, merged.status);
+      await closeRecoveredDecision(d, trx, `download_failed:${row.id}`, note);
+  }
+  // 「已收单未确认」的提交第一次在 qB 里看到:计划改记已确认,同步后补调 afterDownloadAccepted(订阅进度 / 提前完成)
+  if (row.status === "submitting" && row.acquisition_plan_id) {
+    const confirmed = await trx
+      .updateTable("acquisition_plan")
+      .set({ status_reason: "accepted_confirmed" })
+      .where("id", "=", row.acquisition_plan_id)
+      .where("status_reason", "=", "accepted_unconfirmed")
+      .returning("id")
+      .executeTakeFirst();
+    if (confirmed) stats.confirmedPlans.push(confirmed.id);
   }
   if (becameComplete) {
     stats.completed.push(row.id);
@@ -974,7 +1186,7 @@ const RECOVERED = new Set<DownloadStatus>(["queued", "downloading", "completed",
  * 卡住 / 异常事项自动关闭:只关 open / awaiting_user;Agent 正在调查(investigating)的由它自己核实后关闭。
  * resolution 不带处理选项,executeResolution 不会再执行任何动作。
  */
-async function closeRecoveredDecision(d: AoDeps, trx: Exec, dedupeKey: string, status: DownloadStatus) {
+async function closeRecoveredDecision(d: AoDeps, trx: Exec, dedupeKey: string, note: string) {
   const by = "system:download-sync";
   const rows = await trx
     .updateTable("decision")
@@ -982,7 +1194,7 @@ async function closeRecoveredDecision(d: AoDeps, trx: Exec, dedupeKey: string, s
       status: "resolved",
       resolved_by: by,
       resolved_at: d.ports.now(),
-      resolution: jsonb({ key: "download_recovered", optionKey: null, by, note: `下载已恢复（${status}）` }),
+      resolution: jsonb({ key: "download_recovered", optionKey: null, by, note }),
     })
     .where("dedupe_key", "=", dedupeKey)
     .where("status", "in", ["open", "awaiting_user"])
@@ -998,6 +1210,82 @@ async function closeRecoveredDecision(d: AoDeps, trx: Exec, dedupeKey: string, s
   }
 }
 
+/** 同一 dedupeKey 已有未关闭的事项(巡检类事项每轮都会判定,已开着就不再刷新、不累加次数) */
+async function hasOpenDecision(exec: Exec, dedupeKey: string): Promise<boolean> {
+  const r = await exec
+    .selectFrom("decision")
+    .select("id")
+    .where("dedupe_key", "=", dedupeKey)
+    .where("status", "in", ["open", "investigating", "awaiting_user"])
+    .executeTakeFirst();
+  return !!r;
+}
+
+/** 下载器连续不可达超过 DOWNLOADER_OFFLINE_DECISION_MS:按实例开一条待处理,恢复连接后由同步关闭 */
+async function reportDownloaderOffline(d: AoDeps, topo: InstanceTopology, now: Date, err: unknown) {
+  const since = topo.inst.last_seen_at ?? topo.inst.created_at;
+  const offlineMs = now.getTime() - since.getTime();
+  if (offlineMs < DOWNLOADER_OFFLINE_DECISION_MS) return;
+  const dedupeKey = `downloader_offline:${topo.inst.id}`;
+  if (await hasOpenDecision(d.ctx.db, dedupeKey)) return;
+  await upsertDecision(d, d.ctx.db, {
+    kind: "downloader_offline",
+    severity: "critical",
+    title: `下载器连不上：${topo.inst.name}`,
+    summary:
+      `${topo.inst.last_seen_at ? `已 ${Math.round(offlineMs / 60_000)} 分钟` : "添加后一直"}连不上 ${topo.inst.base_url}，` +
+      "新下载提交不了，在下的任务状态也不更新。检查下载器是否在运行、地址和登录信息是否正确",
+    subjectType: "downloader_instance",
+    subjectId: topo.inst.id,
+    dedupeKey,
+    evidence: {
+      instanceId: topo.inst.id,
+      baseUrl: topo.inst.base_url,
+      lastSeenAt: topo.inst.last_seen_at?.toISOString() ?? null,
+      error: errMessage(err).slice(0, 500),
+    },
+  });
+}
+
+/**
+ * 下载目录读不到容量(statfs 失败、保存路径没有路径映射):按卷开 critical 待处理,读得到后关闭。
+ * 读不到时新下载一律延后(不能把"未知"当"有空间"),所以这是配置问题(路径映射 / 容器挂载),不是空间问题。
+ */
+export async function reportVolumeStates(d: AoDeps, vols: VolumeState[]): Promise<void> {
+  for (const v of vols) {
+    const dedupeKey = `cache_volume_unknown:${v.volumeKey}`;
+    if (v.snapshot) {
+      await closeRecoveredDecision(d, d.ctx.db, dedupeKey, "下载目录已能读到容量");
+      continue;
+    }
+    if (await hasOpenDecision(d.ctx.db, dedupeKey)) continue;
+    const usage = v.usage && v.usage.status !== "ok" ? (v.usage as { code?: string; message?: string }) : null;
+    const savePaths = v.instances.map((t) => ({
+      instance: t.inst.name,
+      savePath: t.savePath,
+      localPath: t.localSavePath,
+    }));
+    const why = v.path
+      ? `读不到本机路径 ${v.path} 的容量${usage?.code ? `（${usage.code}）` : ""}，检查容器是否挂载了下载目录`
+      : "下载器的保存路径没法换算成本机路径，检查下载器的保存路径和路径映射";
+    await upsertDecision(d, d.ctx.db, {
+      kind: "cache_volume_unknown",
+      severity: "critical",
+      title: `下载目录读不到容量：${v.instances.map((t) => t.inst.name).join("、")}`,
+      summary: `${why}。读到之前新下载一律延后`,
+      subjectType: "cache_volume",
+      dedupeKey,
+      evidence: {
+        volumeKey: v.volumeKey,
+        path: v.path,
+        reason: v.path ? "statfs_failed" : "save_path_unmapped",
+        usage: usage ? { code: usage.code ?? null, message: usage.message ?? null } : null,
+        instances: savePaths,
+      },
+    });
+  }
+}
+
 /** 用户给已有任务贴了管理标签:登记接管,不改保存路径与其它标签 */
 async function adopt(d: AoDeps, topo: InstanceTopology, t: QbTorrent, now: Date, stats: SyncStats) {
   const db = d.ctx.db;
@@ -1092,20 +1380,31 @@ async function adopt(d: AoDeps, topo: InstanceTopology, t: QbTorrent, now: Date,
 
 async function enforceThresholds(d: AoDeps, insts: InstanceTopology[], stats: SyncStats) {
   if (insts.length === 0) return;
+  const vols = await loadVolumes(d, d.ctx.db, insts);
+  await reportVolumeStates(d, vols);
   // 只读对账模式禁止自动写入(含对 qB 的暂停/恢复)
   if ((await d.ctx.settings.get("automation")).readOnlyReconcile) return;
   const cache = await d.ctx.settings.get("cache");
-  const vols = await loadVolumes(d, d.ctx.db, insts);
   for (const v of vols) {
     const dv = decideVolume(v, cache);
-    if (dv.decision === "pause_managed" && dv.pauseTaskIds.length > 0) {
-      const n = await pauseForThreshold(d, v, dv.pauseTaskIds);
+    // 排队中(queuedDL)的纳管任务一并暂停:否则其它任务被暂停后 qB 的队列会把它拉起来
+    const pauseIds =
+      dv.decision === "pause_managed"
+        ? [
+            ...new Set([
+              ...dv.pauseTaskIds,
+              ...v.inflight.filter((t) => t.managed && t.status === "queued").map((t) => t.taskId),
+            ]),
+          ]
+        : [];
+    if (pauseIds.length > 0) {
+      const n = await pauseForThreshold(d, v, pauseIds);
       stats.thresholdPaused += n;
       await d.ctx.outbox.emit(d.ctx.db, "download.threshold", {
         volumeKey: v.volumeKey,
         state: "tripped",
         usedPercent: dv.usedPct === undefined ? null : Math.min(100, dv.usedPct),
-        pausedTaskIds: dv.pauseTaskIds,
+        pausedTaskIds: pauseIds,
       });
     } else if (dv.resumeThresholdPaused) {
       const ids = await resumeThresholdPaused(d, v);
@@ -1272,8 +1571,9 @@ export async function cleanupCandidates(d: AoDeps, v: VolumeState): Promise<Clea
       const site = rel
         ? await db.selectFrom("site").select("config").where("id", "=", rel.site_id).executeTakeFirst()
         : null;
-      const days = instancePolicy(topo.inst).seedRetentionDays ?? retention.seedRetentionDays;
-      const lock = await lockReason(d, row, rel ?? null, site ?? null, days, now);
+      const policy = instancePolicy(topo.inst);
+      const days = policy.seedRetentionDays ?? retention.seedRetentionDays;
+      const lock = await lockReason(d, row, rel ?? null, site ?? null, days, now, policy.hnrUnknownCleanable === true);
       out.push({ row, topo, bytes: row.size_bytes ?? 0, cleanable: lock === null, lockReason: lock });
     }
   }
@@ -1287,10 +1587,13 @@ async function lockReason(
   site: { config: unknown } | null,
   days: number,
   now: Date,
+  /** 下载器设置「H&R 未知也可清理」:保种天数到期后,H&R 要求未知按已满足处理 */
+  hnrUnknownCleanable = false,
 ): Promise<string | null> {
   if (!retentionElapsed(row, days, now)) return "retention_days";
   const hnr = hnrSatisfied(row, rel as never, site as never);
-  if (hnr !== true) return hnr === false ? "hnr_unsatisfied" : "hnr_unknown";
+  if (hnr === false) return "hnr_unsatisfied";
+  if (hnr === "unknown" && !hnrUnknownCleanable) return "hnr_unknown";
   // 已验证入库:最近一次整理记录成功,且没有失败/排队的文件(部分入库不能整种清理)
   const rec = await d.ctx.db
     .selectFrom("organization_record")
@@ -1385,11 +1688,16 @@ export async function cleanupDownloadCache(
     if (pct > th.cleanupTargetPercent && pct >= th.hardThresholdPercent) {
       entry.action = entry.removed.length > 0 ? "cleaned" : "insufficient";
       // 没有可清理对象:待处理 + 阈值继续阻止新下载(不突破保留要求)
+      const hnrUnknown = entry.skipped.filter((x) => x.reason === "hnr_unknown").length;
       await upsertDecision(d, db, {
         kind: "space_insufficient",
         severity: "critical",
         title: `下载缓存卷 ${v.volumeKey} 空间不足`,
-        summary: `使用率 ${pct.toFixed(1)}%，可清理任务已用尽；新下载已暂停`,
+        summary:
+          `使用率 ${pct.toFixed(1)}%，可清理任务已用尽；新下载已暂停` +
+          (hnrUnknown
+            ? `。另有 ${hnrUnknown} 个任务因为 H&R 要求未知没有清理，确认没有 H&R 要求的话，可在下载器设置里打开「H&R 未知也可清理」`
+            : ""),
         subjectType: "cache_volume",
         dedupeKey: `space_insufficient:${v.volumeKey}`,
         evidence: { usedPercent: pct, locked: entry.skipped },

apps/server/src/modules/agent/capabilities-download.ts

@@ -1,7 +1,6 @@
-import { createQbClient, type DownloaderClient, qbAuthConfig } from "@umt/adapters";
 import { CapabilityError, type CapabilityRegistry } from "@umt/agent";
 import { capabilitySpec, type DownloadStatus } from "@umt/contracts";
-import type { DownloaderInstanceRow } from "@umt/db";
+import { downloaderClientFor } from "../acquisition-organize/ports.ts";
 import type { ServerContext } from "../types.ts";
 import { planBinding } from "./authorize.ts";
 import type { AgentModuleOptions } from "./runner.ts";
@@ -15,16 +14,6 @@ const RECHECK_FROM: DownloadStatus[] = ["missing", "failed", "stalled", "paused"
 /** resume 允许的起点 */
 const RESUME_FROM: DownloadStatus[] = ["paused", "stalled", "failed"];
 
-async function defaultClient(ctx: ServerContext, inst: DownloaderInstanceRow): Promise<DownloaderClient> {
-  const settings = (inst.settings ?? {}) as Record<string, unknown>;
-  const secret = inst.secret_ref ? await ctx.secrets.resolve(inst.secret_ref) : undefined;
-  return createQbClient({
-    instanceId: inst.id,
-    baseUrl: inst.base_url,
-    ...qbAuthConfig({ authKind: settings.authKind, username: inst.username, secret: secret ?? null }),
-  });
-}
-
 /** 持续授权的对象范围:按下载任务 → 计划 → Slot → 作品核对(authorize 对非计划类能力拿不到目标作品) */
 async function checkGrantScope(ctx: ServerContext, grantId: string, planId: string | null) {
   const g = await ctx.db.selectFrom("automation_grant").select("scope").where("id", "=", grantId).executeTakeFirst();
@@ -75,7 +64,8 @@ export function registerDownloadRecoveryCapabilities(
       .where("id", "=", dt.instance_id)
       .executeTakeFirst();
     if (!inst) throw new CapabilityError("not_found", "下载器实例不存在");
-    const client = opts.downloader ? await opts.downloader(inst) : await defaultClient(ctx, inst);
+    // 与下载同步共用按实例缓存的客户端(同一登录会话与凭证处理),不另建
+    const client = opts.downloader ? await opts.downloader(inst) : await downloaderClientFor(ctx, inst);
     try {
       if (input.action === "recheck") await client.recheck([dt.torrent_hash]);
       else await client.resume([dt.torrent_hash]);