diff --git a/apps/server/src/modules/acquisition-organize/services/organize-retry.ts b/apps/server/src/modules/acquisition-organize/services/organize-retry.ts index 4ad84165..26d5f32c 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize-retry.ts +++ b/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`plan->>'idempotencyKey'`, "=", overrides.idempotencyKey) + .where(sql`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); diff --git a/apps/server/src/modules/acquisition-organize/services/takeover.ts b/apps/server/src/modules/acquisition-organize/services/takeover.ts index feb4d107..5ccea6e4 100644 --- a/apps/server/src/modules/acquisition-organize/services/takeover.ts +++ b/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) diff --git a/apps/server/test/acquisition-organize/organize-identity-recovery.test.ts b/apps/server/test/acquisition-organize/organize-identity-recovery.test.ts index 827ff566..2c9a649c 100644 --- a/apps/server/test/acquisition-organize/organize-identity-recovery.test.ts +++ b/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`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`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 });