fix(server,test): W6-A stabilize CI — sync lost update, atomic recovery summary, serial PG tests

- downloads.ts observe: re-read download_task FOR UPDATE inside the tx and merge from the locked row; qB sync could overwrite imported back to importing when verifyAndImport committed between the sync read and write (E8 flake). - agent finalize: accept summaryPrefix so the recovery verdict is written with the terminal status in one update (E6 read a budget_exceeded run before the "未恢复:…" annotation landed); recover-summary step kept, now idempotent no-op. - organize.ts: drop local slotComplete and the once-subscription completion write; completion/history are decided only by applySlotEvent(verify_passed) (W5-C follow-up, removes double write). Manual slot organize uses the shared subscriptions.slotComplete. - vitest: server split into unit (parallel) and pg (file-serial, groupOrder 1) projects; packages/tasks file-serial; explicit worker concurrency in e2e. - docs: W6 stabilization record in 04-v0.1-acceptance.md. Test expectations unchanged. Full `deno task ci` x3 green: 369s / 341s / 337s. Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-09-28 00:32:57 +08:00
1 parent 36aa99b8aa
commit 887843f777
@@ -659,7 +659,29 @@ async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats)
 const sizeOf = (t: QbTorrent) => (t.totalSize ?? t.size) || null;
 const downloadedOf = (t: QbTorrent) => Math.max(0, Math.round((sizeOf(t) ?? 0) - t.amountLeft));

-async function observe(d: AoDeps, row: DownloadTaskRow, t: QbTorrent, now: Date, stats: SyncStats): Promise<boolean> {
+async function observe(d: AoDeps, seen: DownloadTaskRow, t: QbTorrent, now: Date, stats: SyncStats): Promise<boolean> {
+  // 行锁内按最新行合并:同步读到行之后入库事务可能已把 importing 推进到 imported,
+  // 用同步开始时的快照整行覆写会把状态写回 importing(丢失更新)
+  return d.ctx.db.transaction().execute(async (trx) => {
+    const row = await trx
+      .selectFrom("download_task")
+      .selectAll()
+      .where("id", "=", seen.id)
+      .forUpdate()
+      .executeTakeFirst();
+    if (!row || row.removed_at || row.status === "removed") return false;
+    return observeLocked(d, trx, row, t, now, stats);
+  });
+}
+
+async function observeLocked(
+  d: AoDeps,
+  trx: Exec,
+  row: DownloadTaskRow,
+  t: QbTorrent,
+  now: Date,
+  stats: SyncStats,
+): Promise<boolean> {
  const cur = { status: row.status, pauseReason: row.pause_reason };
  const obs = mapQbState(
    { state: t.rawState, progress: t.progress, amountLeft: t.amountLeft },
@@ -691,80 +713,71 @@ async function observe(d: AoDeps, row: DownloadTaskRow, t: QbTorrent, now: Date,
  }
  const becameComplete = merged.status === "completed" && !["completed", "importing", "imported"].includes(row.status);
  const changed = merged.status !== row.status || merged.pauseReason !== row.pause_reason;
-  await d.ctx.db.transaction().execute(async (trx) => {
-    await trx
-      .updateTable("download_task")
-      .set({
-        status: merged.status,
-        pause_reason: merged.status === "paused" ? merged.pauseReason : null,
-        raw_state: t.rawState,
-        progress: Math.min(1, Math.max(0, t.progress)),
-        size_bytes: sizeOf(t),
-        downloaded_bytes: downloaded,
-        remaining_bytes: Math.max(0, Math.round(t.amountLeft)),
-        ratio: t.ratio,
-        seeding_time_sec: t.seedingTimeSec,
-        save_path: t.savePath,
-        content_path: t.contentPath,
-        name: t.name,
-        tags: t.tags,
-        last_progress_at: lastProgressAt,
-        no_progress_since: noProgressSince,
-        last_synced_at: now,
-        ...(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,
-      );
-    if (becameComplete) {
-      stats.completed.push(row.id);
-      const plan = row.acquisition_plan_id
-        ? await trx
-            .selectFrom("acquisition_plan")
-            .select(["id", "slot_id"])
-            .where("id", "=", row.acquisition_plan_id)
-            .executeTakeFirst()
-        : null;
-      if (plan) await applySlotEvent(d, trx, plan.slot_id, { type: "download_completed" }, { planId: plan.id });
-    }
-    if (stalledNow) {
-      // 无进展:只记证据 + 待处理,保留原任务,不换源(REQ §8.4.2)
-      await upsertDecision(d, trx, {
-        kind: "download_stalled",
-        title: `下载无进展:${t.name}`,
-        summary: `已 ${Math.round((now.getTime() - (noProgressSince?.getTime() ?? now.getTime())) / 60000)} 分钟没有新增字节`,
-        subjectType: "download_task",
-        subjectId: row.id,
-        dedupeKey: `download_stalled:${row.id}`,
-        evidence: {
-          rawState: t.rawState,
-          progress: t.progress,
-          downloadedBytes: downloaded,
-          noProgressSince: noProgressSince?.toISOString(),
-          action: "record_evidence",
-        },
-      });
-    }
-    if ((merged.status === "failed" || merged.status === "missing") && changed) {
-      await upsertDecision(d, trx, {
-        kind: "download_failed",
-        title: `下载异常:${t.name}`,
-        summary: `qB 状态 ${t.rawState}`,
-        subjectType: "download_task",
-        subjectId: row.id,
-        dedupeKey: `download_failed:${row.id}`,
-        evidence: { rawState: t.rawState, progress: t.progress },
-      });
-    }
-  });
+  await trx
+    .updateTable("download_task")
+    .set({
+      status: merged.status,
+      pause_reason: merged.status === "paused" ? merged.pauseReason : null,
+      raw_state: t.rawState,
+      progress: Math.min(1, Math.max(0, t.progress)),
+      size_bytes: sizeOf(t),
+      downloaded_bytes: downloaded,
+      remaining_bytes: Math.max(0, Math.round(t.amountLeft)),
+      ratio: t.ratio,
+      seeding_time_sec: t.seedingTimeSec,
+      save_path: t.savePath,
+      content_path: t.contentPath,
+      name: t.name,
+      tags: t.tags,
+      last_progress_at: lastProgressAt,
+      no_progress_since: noProgressSince,
+      last_synced_at: now,
+      ...(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);
+  if (becameComplete) {
+    stats.completed.push(row.id);
+    const plan = row.acquisition_plan_id
+      ? await trx
+          .selectFrom("acquisition_plan")
+          .select(["id", "slot_id"])
+          .where("id", "=", row.acquisition_plan_id)
+          .executeTakeFirst()
+      : null;
+    if (plan) await applySlotEvent(d, trx, plan.slot_id, { type: "download_completed" }, { planId: plan.id });
+  }
+  if (stalledNow) {
+    // 无进展:只记证据 + 待处理,保留原任务,不换源(REQ §8.4.2)
+    await upsertDecision(d, trx, {
+      kind: "download_stalled",
+      title: `下载无进展:${t.name}`,
+      summary: `已 ${Math.round((now.getTime() - (noProgressSince?.getTime() ?? now.getTime())) / 60000)} 分钟没有新增字节`,
+      subjectType: "download_task",
+      subjectId: row.id,
+      dedupeKey: `download_stalled:${row.id}`,
+      evidence: {
+        rawState: t.rawState,
+        progress: t.progress,
+        downloadedBytes: downloaded,
+        noProgressSince: noProgressSince?.toISOString(),
+        action: "record_evidence",
+      },
+    });
+  }
+  if ((merged.status === "failed" || merged.status === "missing") && changed) {
+    await upsertDecision(d, trx, {
+      kind: "download_failed",
+      title: `下载异常:${t.name}`,
+      summary: `qB 状态 ${t.rawState}`,
+      subjectType: "download_task",
+      subjectId: row.id,
+      dedupeKey: `download_failed:${row.id}`,
+      evidence: { rawState: t.rawState, progress: t.progress },
+    });
+  }
  return changed;
 }

@@ -24,7 +24,6 @@ import {
  type ExistingPolicy,
  type ExistingTarget,
  findOtherVersions,
-  movieCompletion,
  type NewBinding,
  type OrganizePlan,
  overrideFor,
@@ -34,7 +33,6 @@ import {
  type QualityFacts,
  resolveTargetDirectory,
  type SourceFile,
-  unitCompletion,
 } from "@umt/domain";
 import { organizationStatusFromFiles, organizePlanToFileResults, toDomainPathBinding } from "@umt/mapping";
 import { withAttemptToken } from "@umt/tasks";
@@ -45,6 +43,7 @@ import { applySlotEvent } from "./downloads.ts";
 import { PathGuardError } from "./organize-guard.ts";
 import { type FrozenTransfer, finalizeTransfer, planTransfer, type TransferSpec } from "./organize-transfer.ts";
 import { hnrSatisfied } from "./retention.ts";
+import { slotComplete as sharedSlotComplete } from "./subscriptions.ts";

 // 入库:冻结整理计划(organization_record + 文件明细)→ 逐文件可恢复执行(操作账本 fencing)→
 // 资产/位置/路径绑定同事务登记 → Slot 推进 → Move 后移除 qB 任务记录(不删数据)。
@@ -1455,14 +1454,14 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise<Execu
            .where("id", "=", plan.slot_id)
            .executeTakeFirst();
          if (slot && okAny) {
-            const complete = await slotComplete(trx, slot, frozen, verifiedAssets);
            const old = slot.current_asset_id;
+            // 完成判定与订阅完成(含历史)统一由 applySlotEvent(verify_passed) 按总集数账本重算,这里不再本地判定
            await applySlotEvent(
              d,
              trx,
              slot.id,
-              { type: "verify_passed", complete },
-              { currentAssetId: verifiedAssets[0]!, reason: complete ? "verified" : "verified_partial" },
+              { type: "verify_passed", complete: false },
+              { currentAssetId: verifiedAssets[0]!, reason: "verified" },
            );
            if (old && old !== verifiedAssets[0] && plan.purpose === "upgrade") {
              // 升级:旧版本保留,处置只记待处理,不删
@@ -1482,14 +1481,6 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise<Execu
                },
              });
            }
-            if (complete && slot.subscription_id) {
-              await trx
-                .updateTable("subscription")
-                .set({ status: "completed", status_reason: "slot_satisfied" })
-                .where("id", "=", slot.subscription_id)
-                .where("mode", "=", "once")
-                .execute();
-            }
          } else if (slot && !okAny) {
            await applySlotEvent(d, trx, slot.id, { type: "verify_failed" }, { reason: `organize:${status}` });
          }
@@ -1505,7 +1496,7 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise<Execu
        .forUpdate()
        .executeTakeFirst();
      if (slot && !["reserved", "downloading", "verifying"].includes(slot.status)) {
-        const complete = await slotComplete(trx, slot, frozen, verifiedAssets);
+        const { complete } = await sharedSlotComplete(trx, slot, frozen.mediaType, verifiedAssets);
        await trx
          .updateTable("acquisition_slot")
          .set({
@@ -1545,56 +1536,6 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise<Execu
  return { recordId: result.recordId, status: result.status, succeeded: result.succeeded, failed: result.failed };
 }

-async function slotComplete(
-  exec: Exec,
-  slot: { id: string; media_entity_id: string; content_unit_id: string | null; completion: unknown },
-  frozen: FrozenPlan,
-  newAssets: string[],
-): Promise<boolean> {
-  const completion = (slot.completion ?? {}) as { kind?: string; cuts?: string[] };
-  if (frozen.mediaType === "movie") {
-    const assets = await exec
-      .selectFrom("media_asset")
-      .select(["id", "content_edition"])
-      .where("id", "in", newAssets)
-      .execute();
-    return movieCompletion(
-      completion.cuts as never,
-      assets.map((a) => ({
-        assetId: a.id,
-        verified: true,
-        ...(a.content_edition ? { cut: a.content_edition as never } : {}),
-      })),
-    ).complete;
-  }
-  // 剧集/综艺:按内容单元覆盖;单元未建档(总集数未知)时继续追踪,不声称完整
-  let q = exec
-    .selectFrom("content_unit")
-    .select(["id", "status"])
-    .where("media_entity_id", "=", slot.media_entity_id)
-    .where("kind", "=", "episode")
-    .where("content_type", "=", "main");
-  if (slot.content_unit_id) q = q.where("parent_unit_id", "=", slot.content_unit_id);
-  const units = await q.where("status", "!=", "retired").execute();
-  const required = units.map((u) => u.id);
-  if (required.length === 0) return false;
-  // 总集数只有在单元都已确认(元数据/人工)时才算已知;由发布标题或文件名临时建出的单元(provisional)
-  // 只说明"见过这些集",不说明"只有这些集":总集数未知时不判完整,订阅继续追踪(REQ §7.1.1)
-  const totalKnown = units.every((u) => u.status === "confirmed");
-  if (!totalKnown) return false;
-  const cov = await sql<{ asset_id: string; unit_id: string }>`
-    select c.media_asset_id as asset_id, c.content_unit_id as unit_id from asset_unit_coverage c
-      join media_asset a on a.id = c.media_asset_id
-     where a.media_entity_id = ${slot.media_entity_id} and a.status = 'active'`.execute(exec);
-  const byAsset = new Map<string, string[]>();
-  for (const r of cov.rows) byAsset.set(r.asset_id, [...(byAsset.get(r.asset_id) ?? []), r.unit_id]);
-  return unitCompletion(
-    required,
-    [...byAsset].map(([assetId, fileUnits]) => ({ assetId, verified: true, fileUnits })),
-    { totalKnown },
-  ).complete;
-}
-
 /** 按季号与集号区间找或建集单元(临时 provisional,挂在季单元下) */
 async function ensureEpisodeUnits(
  exec: Exec,
@@ -189,6 +189,7 @@ export async function finalize(
  runId: string,
  res: DriveResult,
  settled: AgentGoalReport | null,
+  opts: { summaryPrefix?: string } = {},
 ): Promise<{ status: AgentRunStatus; summary: string }> {
  const { ctx } = env;
  const run = await ctx.db.selectFrom("agent_run").selectAll().where("id", "=", runId).executeTakeFirstOrThrow();
@@ -213,6 +214,8 @@ export async function finalize(
  }
  // 模型文本只作参考:附在业务核查结论之后,明确标注未经核查
  if (res.finalText && res.status !== "failed") summary += `\n模型汇报(未经核查):${res.finalText}`;
+  // 系统核实结论(授权内恢复)与终态同一次写入:观察者看到终态时汇总已是最终文本
+  if (opts.summaryPrefix) summary = `${opts.summaryPrefix}\n${summary}`;
  await ctx.db
    .insertInto("artifact")
    .values({
@@ -90,8 +90,11 @@ export function registerAgentTasks(registry: TaskRegistry, env: () => AgentEnv)
            maxRounds: rounds,
          });
        }
-        const out = await t.step.run("finalize", () => finalize(e, runId, res, settled));
+        const out = await t.step.run("finalize", () =>
+          finalize(e, runId, res, settled, recovery ? { summaryPrefix: recovery.summary } : {}),
+        );
        if (recovery) {
+          // 已在 finalize 内带上核实结论;保留该步骤兼容旧 run 的步骤历史(已带前缀时幂等跳过)
          const rec = recovery;
          return t.step.run("recover-summary", () => annotateRunSummary(e.ctx, runId, rec));
        }