fix(organize): deduplicate concurrent identity recovery plans

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

@@ -161,6 +161,10 @@ export class OrganizeRecoveryError extends Error {
   }
 }
 
+class OrganizeRetryConflictError extends Error {
+  readonly code = "conflict";
+}
+
 interface RetryOptions {
   mediaEntityId?: string;
   idempotencyKey?: string;
@@ -204,12 +208,16 @@ async function retryLocked(
   if (overrides.idempotencyKey) {
     const existing = await d.ctx.db
       .selectFrom("organization_record")
-      .select(["id", "task_id", "status", "reason"])
+      .select(["id", "task_id", "status", "reason", "media_entity_id"])
       .where(sql<string>`plan->>'idempotencyKey'`, "=", overrides.idempotencyKey)
+      .where(sql<string>`plan->>'retryOfRecordId'`, "=", recordId)
       .where("status", "in", ["queued", "running", "succeeded", "partial"])
       .executeTakeFirst();
-    if (existing?.task_id)
+    if (existing?.task_id) {
+      if (overrides.mediaEntityId && existing.media_entity_id !== overrides.mediaEntityId)
+        throw new OrganizeRetryConflictError("这条整理事项已按其他作品处理，刷新后查看结果");
       return { recordId: existing.id, taskId: existing.task_id, status: existing.status, reason: existing.reason };
+    }
   }
   const resolved = await resolveOrganizeRecovery(d.ctx.db, record);
   if (!resolved.recovery.ready) throw new OrganizeRecoveryError(resolved.recovery);

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

@@ -341,7 +341,14 @@ async function reorganizeAs(
   const recordId = row.subject_type === "organization_record" ? row.subject_id : null;
   if (!recordId) throw new HookError("not_found", "这条事项没有关联的整理记录");
   const dtId = ev.downloadTaskId;
-  const r = await retryOrganizeRecord(d, recordId, actor, via, { mediaEntityId: entityId });
+  const r = await retryOrganizeRecord(d, recordId, actor, via, {
+    mediaEntityId: entityId,
+    idempotencyKey: `organize-identity:${row.id}`,
+  }).catch((error: unknown) => {
+    if (error instanceof Error && (error as { code?: string }).code === "conflict")
+      throw new HookError("conflict", error.message);
+    throw error;
+  });
   if (!["running", "queued", "succeeded", "partial"].includes(r.status))
     throw new HookError("conflict", `作品已确认，但重新整理没能执行：${r.reason ?? r.status}`);
   if (dtId)

apps/server/test/acquisition-organize/organize-identity-recovery.test.ts

@@ -1,6 +1,7 @@
 import { randomUUID } from "node:crypto";
 import { symlink, unlink, writeFile } from "node:fs/promises";
 import { join } from "node:path";
+import { sql } from "@umt/db";
 import { createTempTree, type TempTree } from "@umt/testkit";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
 import { executeRecord, planImport } from "../../src/modules/acquisition-organize/services/organize.ts";
@@ -10,6 +11,7 @@ import {
 } from "../../src/modules/acquisition-organize/services/organize-retry.ts";
 import {
   assignOrganizeIdentity,
+  OrganizeDecisionError,
   organizeDecisionHook,
 } from "../../src/modules/acquisition-organize/services/takeover.ts";
 import { autoInvestigate, checkRecovery } from "../../src/modules/agent/investigate.ts";
@@ -123,6 +125,50 @@ describe.skipIf(!hasPg)("按路径身份补齐后恢复原整理", () => {
       ).plan,
     ).toMatchObject({ retryOfRecordId: plan.recordId });
   });
+  it("人工与智能体并发指定同一作品，只冻结并入队一次整理计划", async () => {
+    const { plan, dec } = await unknown("Concurrent.Path", { action: "copy" });
+    const entityId = await seedEntity(h.db, { kind: "movie", title: "Concurrent Path", year: 2022 });
+    const before = h.engine.enqueued.length;
+    const [agent, human] = await Promise.all([
+      assignOrganizeIdentity(h.d, dec.id, { mediaEntityId: entityId }, actor, "标题一致"),
+      organizeDecisionHook(h.d).execute(
+        dec,
+        { optionKey: "select_entity", identity: { mediaEntityId: entityId } },
+        actor,
+      ),
+    ]);
+    expect(agent.recordId).toBe(human.recordId);
+    expect(h.engine.enqueued.slice(before)).toHaveLength(1);
+    const records = await h.db
+      .selectFrom("organization_record")
+      .select("id")
+      .where(sql<string>`plan->>'retryOfRecordId'`, "=", plan.recordId!)
+      .execute();
+    expect(records).toHaveLength(1);
+  });
+  it("同一事项并发指定不同作品时拒绝后到的选择，不创建第二个计划", async () => {
+    const { plan, dec } = await unknown("Conflicting.Path", { action: "copy" });
+    const entities = await Promise.all(
+      ["First Choice", "Second Choice"].map((title) => seedEntity(h.db, { kind: "movie", title, year: 2022 })),
+    );
+    const before = h.engine.enqueued.length;
+    const results = await Promise.allSettled(
+      entities.map((mediaEntityId) =>
+        organizeDecisionHook(h.d).execute(dec, { optionKey: "select_entity", identity: { mediaEntityId } }, actor),
+      ),
+    );
+    expect(results.filter((result) => result.status === "fulfilled")).toHaveLength(1);
+    const rejected = results.find((result) => result.status === "rejected");
+    expect(rejected?.status === "rejected" && rejected.reason).toMatchObject({ code: "conflict" });
+    expect(rejected?.status === "rejected" && rejected.reason).toBeInstanceOf(OrganizeDecisionError);
+    expect(h.engine.enqueued.slice(before)).toHaveLength(1);
+    const records = await h.db
+      .selectFrom("organization_record")
+      .select("id")
+      .where(sql<string>`plan->>'retryOfRecordId'`, "=", plan.recordId!)
+      .execute();
+    expect(records).toHaveLength(1);
+  });
   it("身份已找到但重新规划失败，保持原事项待处理", async () => {
     const { dec } = await unknown("Blocked.Path", { targetLibraryRootId: randomUUID() });
     const entityId = await seedEntity(h.db, { kind: "movie", title: "Blocked Path", year: 2022 });