From 33d4fa46c5a4e941d1ec7add04cab6d785d9d0da Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Mon, 10 Aug 2026 19:43:13 -0700 Subject: [PATCH] FN-8969: fix plan evidence version collisions Prevent plan writes from permanently failing when durable and snapshot evidence versions diverge. - Centralize conflict-tolerant plan-evidence appends with durable version allocation - Reuse the helper for prompt, lineage, and project evidence writes - Cover version drift, retry, deduplication, and concurrent PostgreSQL writes Files changed: .../fn-8969-plan-evidence-version-collision.md | 7 ++ .../__tests__/plan-evidence-next-version.test.ts | 49 ++++++++++++++ .../plan-evidence-version-collision.pg.test.ts | 72 ++++++++++++++++++++ packages/core/src/store.ts | 40 ++++++----- .../core/src/task-store/async/async-lifecycle.ts | 49 +++++++------- packages/core/src/task-store/plan-evidence.ts | 77 ++++++++++++++++++++++ packages/core/src/task-store/project-store-ops.ts | 47 +++++++------ 7 files changed, 276 insertions(+), 65 deletions(-) Fusion-Task-Id: FN-8969 Fusion-Task-Lineage: 5fba0bc8-bcf6-42ea-b617-3484919ed2fa Co-authored-by: Fusion (runfusion.ai) --- ...fn-8969-plan-evidence-version-collision.md | 7 ++ .../plan-evidence-next-version.test.ts | 49 ++++++++++++ ...plan-evidence-version-collision.pg.test.ts | 72 +++++++++++++++++ packages/core/src/store.ts | 40 ++++++---- .../src/task-store/async/async-lifecycle.ts | 49 ++++++------ packages/core/src/task-store/plan-evidence.ts | 77 +++++++++++++++++++ .../core/src/task-store/project-store-ops.ts | 47 ++++++----- 7 files changed, 276 insertions(+), 65 deletions(-) create mode 100644 .changeset/fn-8969-plan-evidence-version-collision.md create mode 100644 packages/core/src/__tests__/plan-evidence-next-version.test.ts create mode 100644 packages/core/src/__tests__/postgres/plan-evidence-version-collision.pg.test.ts create mode 100644 packages/core/src/task-store/plan-evidence.ts diff --git a/.changeset/fn-8969-plan-evidence-version-collision.md b/.changeset/fn-8969-plan-evidence-version-collision.md new file mode 100644 index 0000000000..66165a9fdc --- /dev/null +++ b/.changeset/fn-8969-plan-evidence-version-collision.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Fix plan writes failing permanently after a plan-evidence version collision. +category: fix +dev: Converges evidence writers on the conflict-tolerant appendPlanEvidenceInTransaction helper. diff --git a/packages/core/src/__tests__/plan-evidence-next-version.test.ts b/packages/core/src/__tests__/plan-evidence-next-version.test.ts new file mode 100644 index 0000000000..19fca1e9fc --- /dev/null +++ b/packages/core/src/__tests__/plan-evidence-next-version.test.ts @@ -0,0 +1,49 @@ +import { expect, it } from "vitest"; +import { appendPlanEvidenceInTransaction, PlanEvidenceAppendError } from "../task-store/plan-evidence.js"; +import { createCurrentPlanEvidence, type CurrentPlanEvidence } from "../planner/spec-lock.js"; + +const prompt = "# Task\n\n## Mission\n\nEvidence\n\n## Steps\n\n1. Keep it\n"; +const evidence = (version: number) => createCurrentPlanEvidence({ version, sourceRevision: 1, capturedAt: "2026-08-11T02:04:00.000Z", prompt }); + +function fakeTx(latestVersions: number[], conflicts: Array<"insert" | "dedupe"> = [], matching?: CurrentPlanEvidence) { + const inserted: CurrentPlanEvidence[] = []; + let selectCount = 0; + const tx = { + select: () => ({ from: () => ({ where: () => ({ orderBy: () => ({ limit: async () => [{ version: latestVersions[Math.min(selectCount++, latestVersions.length - 1)] }] }), limit: async () => matching ? [{ version: matching.version, snapshot: matching }] : [] }) }) }), + insert: () => ({ values: (row: { snapshot: CurrentPlanEvidence }) => ({ onConflictDoNothing: () => ({ returning: async () => { + const outcome = conflicts.shift(); + if (outcome === "insert" || outcome === "dedupe") return []; + inserted.push(row.snapshot); + return [{ version: row.snapshot.version }]; + } }) }) }), + }; + return { tx, inserted }; +} + +it("derives the next evidence version from the durable column", async () => { + const { tx, inserted } = fakeTx([5]); + const result = await appendPlanEvidenceInTransaction(tx as never, { taskId: "KB-1", buildEvidence: evidence }); + expect(result.version).toBe(6); + expect(inserted[0]?.version).toBe(6); +}); + +it("retries a primary-key conflict with a freshly-read version", async () => { + const { tx, inserted } = fakeTx([5, 6], ["insert"]); + const result = await appendPlanEvidenceInTransaction(tx as never, { taskId: "KB-1", buildEvidence: evidence }); + expect(result).toMatchObject({ version: 7, attempts: 2, deduped: false }); + expect(inserted[0]?.version).toBe(7); +}); + +it("returns the stored snapshot for a source-hash dedupe conflict", async () => { + const stored = evidence(5); + const { tx, inserted } = fakeTx([5], ["dedupe"], stored); + const result = await appendPlanEvidenceInTransaction(tx as never, { taskId: "KB-1", buildEvidence: () => stored }); + expect(result).toMatchObject({ evidence: stored, version: 5, deduped: true }); + expect(inserted).toEqual([]); +}); + +it("throws a named error after conflicts cannot produce a durable row", async () => { + const { tx } = fakeTx([5, 6, 7], ["insert", "insert", "insert"]); + await expect(appendPlanEvidenceInTransaction(tx as never, { taskId: "KB-1", buildEvidence: evidence })) + .rejects.toEqual(expect.objectContaining({ name: "PlanEvidenceAppendError", taskId: "KB-1", attempts: 3 } satisfies Partial)); +}); diff --git a/packages/core/src/__tests__/postgres/plan-evidence-version-collision.pg.test.ts b/packages/core/src/__tests__/postgres/plan-evidence-version-collision.pg.test.ts new file mode 100644 index 0000000000..8ab47f1f80 --- /dev/null +++ b/packages/core/src/__tests__/postgres/plan-evidence-version-collision.pg.test.ts @@ -0,0 +1,72 @@ +/* +FNXC:SpecLock 2026-08-11-02:04: +FN-8969 reproduces the operator-visible FN-8964 wedge against PostgreSQL: durable row versions, +not stale snapshot payload versions, must advance every authoritative PROMPT.md write. +*/ +import { afterAll, afterEach, beforeAll, beforeEach, expect, it } from "vitest"; +import { and, eq } from "drizzle-orm"; +import { + createSharedPgTaskStoreTestHarness, + pgDescribe, + type SharedPgTaskStoreHarness, +} from "../../__test-utils__/pg-test-harness.js"; +import { type TaskStore } from "../../store.js"; +import * as schema from "../../postgres/schema/index.js"; +import { createCurrentPlanEvidence } from "../../planner/spec-lock.js"; + +const prompt = (body: string) => `# Task\n\n## Mission\n\n${body}\n\n## File Scope\n\n- packages/core/src/store.ts\n\n## Steps\n\n1. Preserve plan evidence\n\n## Completion Criteria\n\n- [ ] Plan write succeeds\n\n## Do NOT\n\n- Drop evidence\n\n## Dependencies\n\n- None\n`; + +pgDescribe("plan evidence version collisions", () => { + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ prefix: "fusion_plan_evidence_collision" }); + let store: TaskStore; + + beforeAll(h.beforeAll); + afterAll(h.afterAll); + beforeEach(async () => { await h.beforeEach(); store = h.store(); }); + afterEach(h.afterEach); + + async function seedDriftedEvidence(taskId: string) { + await h.layer().db.delete(schema.project.currentPlanEvidence).where(eq(schema.project.currentPlanEvidence.taskId, taskId)); + const evidence = createCurrentPlanEvidence({ version: 4, sourceRevision: Date.now(), capturedAt: new Date().toISOString(), prompt: prompt("stale snapshot") }); + await h.layer().db.insert(schema.project.currentPlanEvidence).values({ + projectId: "", taskId, version: 5, sourceRevision: evidence.sourceRevision, + sourceHash: evidence.sourceHash, capturedAt: evidence.capturedAt, snapshot: evidence, + }); + } + + it.sequential("self-heals a durable-version/snapshot-version wedge and dedupes a repeated prompt", async () => { + const task = await store.createTask({ description: "drifted evidence prompt write" }); + await seedDriftedEvidence(task.id); + const revised = prompt("new authoritative content"); + + await expect(store.updateTask(task.id, { prompt: revised })).resolves.toBeDefined(); + expect((await store.getTask(task.id)).prompt).toBe(revised); + const afterFirst = await h.layer().db.select({ version: schema.project.currentPlanEvidence.version }) + .from(schema.project.currentPlanEvidence).where(and(eq(schema.project.currentPlanEvidence.taskId, task.id))).orderBy(schema.project.currentPlanEvidence.version); + expect(afterFirst.map((row) => row.version)).toContain(6); + + await expect(store.updateTask(task.id, { prompt: revised })).resolves.toBeDefined(); + const afterRepeat = await h.layer().db.select({ version: schema.project.currentPlanEvidence.version }) + .from(schema.project.currentPlanEvidence).where(and(eq(schema.project.currentPlanEvidence.taskId, task.id))); + expect(afterRepeat).toHaveLength(afterFirst.length); + }); + + it.sequential("uses unbound-layer reads and accepts concurrent distinct and identical prompt writes", async () => { + const task = await store.createTask({ description: "unbound concurrent plan writes" }); + await seedDriftedEvidence(task.id); + await expect(Promise.all([ + store.updateTask(task.id, { prompt: prompt("concurrent one") }), + store.updateTask(task.id, { prompt: prompt("concurrent two") }), + ])).resolves.toHaveLength(2); + const distinct = await h.layer().db.select({ version: schema.project.currentPlanEvidence.version, sourceHash: schema.project.currentPlanEvidence.sourceHash }) + .from(schema.project.currentPlanEvidence).where(eq(schema.project.currentPlanEvidence.taskId, task.id)).orderBy(schema.project.currentPlanEvidence.version); + expect(distinct.map((row) => row.version)).toEqual(expect.arrayContaining([5, 6, 7])); + + const same = prompt("same concurrent content"); + await expect(Promise.all([store.updateTask(task.id, { prompt: same }), store.updateTask(task.id, { prompt: same })])).resolves.toHaveLength(2); + const rows = await h.layer().db.select({ sourceHash: schema.project.currentPlanEvidence.sourceHash }) + .from(schema.project.currentPlanEvidence).where(eq(schema.project.currentPlanEvidence.taskId, task.id)); + const sameEvidence = createCurrentPlanEvidence({ version: 1, sourceRevision: 0, capturedAt: "", prompt: same }); + expect(rows.filter((row) => row.sourceHash === sameEvidence.sourceHash)).toHaveLength(1); + }); +}); diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 2695889e9c..f919968b58 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -79,6 +79,7 @@ import { GlobalSettingsStore } from "./config/global-settings.js"; import { Database } from "./db/db.js"; import { ArchiveDatabase } from "./db/archive-db.js"; import { projectScopeFor, type AsyncDataLayer, type DbTransaction } from "./postgres/data-layer.js"; +import { appendPlanEvidenceInTransaction } from "./task-store/plan-evidence.js"; import { planningLifecycleLockTransportAvailability, withPlanningLifecycleAdvisoryLock, withPlanningLifecycleAdvisoryLocks } from "./postgres/advisory-locks.js"; import { MissionStore } from "./missions/mission-store.js"; import { AsyncMissionStore } from "./async-stores/async-mission-store.js"; @@ -1197,16 +1198,19 @@ export class TaskStore extends EventEmitter { return refineTaskImpl(this, id, feedback); } /** - * FNXC:SpecLock 2026-08-09-07:06: - * FN-8845 retains plan snapshots outside the mutable task row. Backend-only storage is deliberate: - * SQLite is no longer a runtime path and silently falling back would lose audit history on restart. + * FNXC:SpecLock 2026-08-11-02:04: + * Evidence append uses the shared column-versioned, conflict-tolerant path. sourceHash remains + * the idempotence key, so callers receive the durable matching snapshot rather than a fabricated + * candidate after a concurrent insert. */ async appendCurrentPlanEvidence(taskId: string, evidence: import("./planner/spec-lock.js").CurrentPlanEvidence): Promise { if (!this.asyncLayer) throw new Error("Spec-lock history requires PostgreSQL backend storage"); - const projectId = this.asyncLayer.projectId ?? ""; - await this.asyncLayer.db.insert(schema.project.currentPlanEvidence).values({ projectId, taskId, version: evidence.version, sourceRevision: evidence.sourceRevision, sourceHash: evidence.sourceHash, capturedAt: evidence.capturedAt, snapshot: evidence }).onConflictDoNothing(); - const rows = await this.asyncLayer.db.select().from(schema.project.currentPlanEvidence).where(and(projectScopeFor(schema.project.currentPlanEvidence.projectId, this.asyncLayer.projectId), eq(schema.project.currentPlanEvidence.taskId, taskId), eq(schema.project.currentPlanEvidence.sourceHash, evidence.sourceHash))).limit(1); - return rows[0]!.snapshot as import("./planner/spec-lock.js").CurrentPlanEvidence; + const result = await appendPlanEvidenceInTransaction(this.asyncLayer.db, { + projectId: this.asyncLayer.projectId, + taskId, + buildEvidence: (version) => ({ ...evidence, version: Math.max(version, evidence.version) }), + }); + return result.evidence; } async getLatestCurrentPlanEvidence(taskId: string): Promise { if (!this.asyncLayer) throw new Error("Spec-lock history requires PostgreSQL backend storage"); @@ -1241,16 +1245,20 @@ export class TaskStore extends EventEmitter { sourceRevision: number, liveTask?: Pick, ): Promise { - const [prior, task] = await Promise.all([this.getLatestCurrentPlanEvidence(taskId), liveTask ?? this.getTask(taskId)]); - const candidate = createCurrentPlanEvidence({ - version: (prior?.version ?? 0) + 1, - sourceRevision, - capturedAt: new Date().toISOString(), - prompt, - bindings: specLockBindings(task), + const task = liveTask ?? await this.getTask(taskId); + if (!this.asyncLayer) throw new Error("Spec-lock history requires PostgreSQL backend storage"); + const result = await appendPlanEvidenceInTransaction(this.asyncLayer.db, { + projectId: this.asyncLayer.projectId, + taskId, + buildEvidence: (version) => createCurrentPlanEvidence({ + version, + sourceRevision, + capturedAt: new Date().toISOString(), + prompt, + bindings: specLockBindings(task), + }), }); - if (prior?.sourceHash === candidate.sourceHash) return prior; - return this.appendCurrentPlanEvidence(taskId, candidate); + return result.evidence; } async appendSpecLock(taskId: string, lock: import("./planner/spec-lock.js").SpecLock): Promise { if (!this.asyncLayer) throw new Error("Spec-lock history requires PostgreSQL backend storage"); diff --git a/packages/core/src/task-store/async/async-lifecycle.ts b/packages/core/src/task-store/async/async-lifecycle.ts index 1fb63cade4..08a50790dd 100644 --- a/packages/core/src/task-store/async/async-lifecycle.ts +++ b/packages/core/src/task-store/async/async-lifecycle.ts @@ -27,11 +27,12 @@ * integration tests consume. They program against the stable `AsyncDataLayer` * interface (U4), not the underlying driver. */ -import { and, desc, eq, ne, sql } from "drizzle-orm"; +import { and, eq, ne, sql } from "drizzle-orm"; import * as schema from "../../postgres/schema/index.js"; -import { projectScopeFor, type AsyncDataLayer, type DbTransaction } from "../../postgres/data-layer.js"; +import { type AsyncDataLayer, type DbTransaction } from "../../postgres/data-layer.js"; import { ACTIVE_TASK_FILTER } from "./async-persistence.js"; -import { createCurrentPlanEvidence, type CurrentPlanEvidence } from "../../planner/spec-lock.js"; +import { createCurrentPlanEvidence } from "../../planner/spec-lock.js"; +import { appendPlanEvidenceInTransaction, PlanEvidenceAppendError } from "../plan-evidence.js"; import { storeLog } from "../../store.js"; /** @@ -186,11 +187,11 @@ export async function removeLineageReferences( const evidenceUnavailableChildIds: string[] = []; let evidenceInsertAttempts = 0; /* - FNXC:SpecLockLineageInvalidation 2026-08-10-14:47: - Evidence inserts use the trigger-compatible blank value, while reads preserve an unbound layer's - project-agnostic scope instead of filtering for the blank value the trigger rewrites. + FNXC:SpecLockLineageInvalidation 2026-08-11-02:04: + FN-8969 converges lineage writes on the shared helper: it writes the trigger-compatible blank + value but reads with unbound-safe project scope, derives versions from the durable column, and + dedupes only by sourceHash while preserving this path's lineage truthfulness validator. */ - const evidenceProjectId = projectId ?? ""; for (const child of returned) { const prompt = promptByChildId.get(child.id); if (prompt === undefined) { @@ -199,24 +200,24 @@ export async function removeLineageReferences( storeLog.warn(`[spec-lock] lineage evidence unavailable: missing PROMPT.md for ${child.id}`); continue; } - let resolved = false; - for (let attempt = 0; attempt < 3 && !resolved; attempt += 1) { - const latest = await tx.select({ version: schema.project.currentPlanEvidence.version }).from(schema.project.currentPlanEvidence) - .where(and(projectScopeFor(schema.project.currentPlanEvidence.projectId, projectId), eq(schema.project.currentPlanEvidence.taskId, child.id))).orderBy(desc(schema.project.currentPlanEvidence.version)).limit(1); - const computed = (latest[0]?.version ?? 0) + 1; - const evidence = createCurrentPlanEvidence({ version: evidenceTargetVersionForTest?.(child.id, computed, attempt) ?? computed, sourceRevision: Date.now(), capturedAt: nowIso, prompt, bindings: { dependencies: (child.dependencies as string[] | undefined) ?? [], missionId: child.missionId ?? undefined, sliceId: child.sliceId ?? undefined } }); - evidenceInsertAttempts += 1; - const inserted = await tx.insert(schema.project.currentPlanEvidence).values({ projectId: evidenceProjectId, taskId: child.id, version: evidence.version, sourceRevision: evidence.sourceRevision, sourceHash: evidence.sourceHash, capturedAt: evidence.capturedAt, snapshot: evidence }).onConflictDoNothing().returning({ version: schema.project.currentPlanEvidence.version }); - if (inserted[0]) { evidenceVersionByChild.set(child.id, inserted[0].version); resolved = true; break; } - const matching = await tx.select({ version: schema.project.currentPlanEvidence.version, snapshot: schema.project.currentPlanEvidence.snapshot }).from(schema.project.currentPlanEvidence) - .where(and(projectScopeFor(schema.project.currentPlanEvidence.projectId, projectId), eq(schema.project.currentPlanEvidence.taskId, child.id), eq(schema.project.currentPlanEvidence.sourceHash, evidence.sourceHash))).limit(1); - if (matching[0]) { - const snapshot = matching[0].snapshot as CurrentPlanEvidence; - if (snapshot.plan.sections.lineage.canonical.split("\n").includes(`parent-task:${parentId}`)) throw new LineageEvidenceAppendError(child.id, attempt + 1, "matched-row-not-truthful"); - evidenceVersionByChild.set(child.id, matching[0].version); resolved = true; break; - } + try { + const result = await appendPlanEvidenceInTransaction(tx, { + projectId, + taskId: child.id, + maxAttempts: 3, + buildEvidence: (computed, attempt) => { + evidenceInsertAttempts += 1; + return createCurrentPlanEvidence({ version: evidenceTargetVersionForTest?.(child.id, computed, attempt) ?? computed, sourceRevision: Date.now(), capturedAt: nowIso, prompt, bindings: { dependencies: (child.dependencies as string[] | undefined) ?? [], missionId: child.missionId ?? undefined, sliceId: child.sliceId ?? undefined } }); + }, + validateMatchedEvidence: (snapshot, _version, attempt) => { + if (snapshot.plan.sections.lineage.canonical.split("\n").includes(`parent-task:${parentId}`)) throw new LineageEvidenceAppendError(child.id, attempt + 1, "matched-row-not-truthful"); + }, + }); + evidenceVersionByChild.set(child.id, result.version); + } catch (error) { + if (error instanceof PlanEvidenceAppendError) throw new LineageEvidenceAppendError(child.id, 3, "no-durable-version"); + throw error; } - if (!resolved) throw new LineageEvidenceAppendError(child.id, 3, "no-durable-version"); } return { clearedChildIds: returned.map((row) => row.id), evidenceVersionByChild, evidenceUnavailableChildIds, evidenceInsertAttempts }; } diff --git a/packages/core/src/task-store/plan-evidence.ts b/packages/core/src/task-store/plan-evidence.ts new file mode 100644 index 0000000000..80248f32b5 --- /dev/null +++ b/packages/core/src/task-store/plan-evidence.ts @@ -0,0 +1,77 @@ +import { and, desc, eq } from "drizzle-orm"; +import { projectScopeFor, type AsyncDataLayer, type DbTransaction } from "../postgres/data-layer.js"; +import * as schema from "../postgres/schema/index.js"; +import type { CurrentPlanEvidence } from "../planner/spec-lock.js"; + +export class PlanEvidenceAppendError extends Error { + constructor(readonly taskId: string, readonly attempts: number) { + super(`Plan evidence append exhausted for ${taskId} after ${attempts} attempts`); + this.name = "PlanEvidenceAppendError"; + } +} + +export type AppendPlanEvidenceOptions = { + projectId?: string; + taskId: string; + buildEvidence: (version: number, attempt: number) => CurrentPlanEvidence; + maxAttempts?: number; + /** Validates a source-hash dedupe row before it is treated as durable success. */ + validateMatchedEvidence?: (evidence: CurrentPlanEvidence, version: number, attempt: number) => void; +}; + +export type AppendPlanEvidenceResult = { + evidence: CurrentPlanEvidence; + version: number; + deduped: boolean; + attempts: number; +}; + +/** + * FNXC:SpecLock 2026-08-11-02:04: + * FN-8969/FN-8964 require every plan-evidence writer to derive versions from the durable column, + * not snapshot JSON: a stale embedded version otherwise wedges future operator PROMPT.md writes. + * PK races retry while sourceHash conflicts return their existing immutable evidence, preserving + * append-only history without allowing a conflict to hard-fail a visible plan write. + */ +export async function appendPlanEvidenceInTransaction( + tx: DbTransaction | AsyncDataLayer["db"], + options: AppendPlanEvidenceOptions, +): Promise { + const maxAttempts = options.maxAttempts ?? 3; + for (let attempt = 0; attempt < maxAttempts; attempt += 1) { + const latest = await tx.select({ version: schema.project.currentPlanEvidence.version }) + .from(schema.project.currentPlanEvidence) + .where(and( + projectScopeFor(schema.project.currentPlanEvidence.projectId, options.projectId), + eq(schema.project.currentPlanEvidence.taskId, options.taskId), + )) + .orderBy(desc(schema.project.currentPlanEvidence.version)) + .limit(1); + const evidence = options.buildEvidence((latest[0]?.version ?? 0) + 1, attempt); + const inserted = await tx.insert(schema.project.currentPlanEvidence).values({ + projectId: options.projectId ?? "", + taskId: options.taskId, + version: evidence.version, + sourceRevision: evidence.sourceRevision, + sourceHash: evidence.sourceHash, + capturedAt: evidence.capturedAt, + snapshot: evidence, + }).onConflictDoNothing().returning({ version: schema.project.currentPlanEvidence.version }); + if (inserted[0]) return { evidence, version: inserted[0].version, deduped: false, attempts: attempt + 1 }; + + const matching = await tx.select({ version: schema.project.currentPlanEvidence.version, snapshot: schema.project.currentPlanEvidence.snapshot }) + .from(schema.project.currentPlanEvidence) + .where(and( + projectScopeFor(schema.project.currentPlanEvidence.projectId, options.projectId), + eq(schema.project.currentPlanEvidence.taskId, options.taskId), + eq(schema.project.currentPlanEvidence.sourceHash, evidence.sourceHash), + )) + .limit(1); + if (matching[0]) { + const stored = matching[0].snapshot as CurrentPlanEvidence; + options.validateMatchedEvidence?.(stored, matching[0].version, attempt); + return { evidence: stored, version: matching[0].version, deduped: true, attempts: attempt + 1 }; + } + } + throw new PlanEvidenceAppendError(options.taskId, maxAttempts); +} diff --git a/packages/core/src/task-store/project-store-ops.ts b/packages/core/src/task-store/project-store-ops.ts index 77551be7d2..40d77b83da 100644 --- a/packages/core/src/task-store/project-store-ops.ts +++ b/packages/core/src/task-store/project-store-ops.ts @@ -13,7 +13,7 @@ import { resolveCapacityPoolId } from "../workflows/workflow-capacity.js"; import {resolveWorkflowIntakeFacts} from "./task-creation.js"; import {TransitionRejectionError} from "./errors.js"; import * as schema from "../postgres/schema/index.js"; -import {and, desc, eq, inArray, isNull, ne, or, sql} from "drizzle-orm"; +import {and, eq, inArray, isNull, ne, or, sql} from "drizzle-orm"; import {mkdir, writeFile} from "node:fs/promises"; import {join} from "node:path"; import type {Task, ColumnId, CheckoutClaimPrecondition, ActivityLogEntry, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, GoalCitation, GoalCitationFilter} from "../types.js"; @@ -43,6 +43,7 @@ import {recordActivityLogEntry as recordActivityLogEntryAsync} from "./async/asy import {applyOriginalDescription} from "../tasks/original-description-policy.js"; import {isPlanReviewSatisfied} from "../planner/plan-approval.js"; import {createCurrentPlanEvidence} from "../planner/spec-lock.js"; +import {appendPlanEvidenceInTransaction} from "./plan-evidence.js"; import {recordRunAuditEvent as recordRunAuditEventAsync} from "../postgres/data-layer.js"; import {listGoalCitations as listGoalCitationsAsync} from "./async/async-events.js"; import type {RunAuditEventRow} from "../task-store/row-types.js"; @@ -137,33 +138,29 @@ export async function atomicWriteTaskJsonWithAuditImpl(store: TaskStore, dir: st const persist = async () => { const row = await readTaskRowInTransaction(tx, id, { includeDeleted: true }, layer.projectId); /* - FNXC:SpecLock 2026-08-09-18:17: - A full PROMPT.md rewrite publishes its evidence and clears approval in this one task-row - transaction. A database rollback therefore cannot leave either half visible by itself. + FNXC:SpecLock 2026-08-11-02:04: + A full PROMPT.md rewrite publishes evidence and clears approval in this task-row transaction, + so rollback cannot expose either half alone. FN-8969 requires the shared append path here: + it reads the durable version column with unbound-safe scope, retries PK races, and dedupes by + sourceHash rather than trusting a snapshot version that can be stale. */ if (specPlanPrompt !== undefined) { - const projectId = layer.projectId ?? ""; - const priorRows = await tx.select().from(schema.project.currentPlanEvidence) - .where(and(eq(schema.project.currentPlanEvidence.projectId, projectId), eq(schema.project.currentPlanEvidence.taskId, id))) - .orderBy(desc(schema.project.currentPlanEvidence.version)).limit(1); - const prior = priorRows[0]?.snapshot as import("../planner/spec-lock.js").CurrentPlanEvidence | undefined; - const candidate = createCurrentPlanEvidence({ - version: (prior?.version ?? 0) + 1, - sourceRevision: Date.now(), - capturedAt: new Date().toISOString(), - prompt: specPlanPrompt, + await appendPlanEvidenceInTransaction(tx, { + projectId: layer.projectId, + taskId: id, + buildEvidence: (version) => createCurrentPlanEvidence({ + version, + sourceRevision: Date.now(), + capturedAt: new Date().toISOString(), + prompt: specPlanPrompt, + bindings: { + dependencies: task.dependencies ?? [], + missionId: task.missionId, + sliceId: task.sliceId, + sourceParentTaskId: task.sourceParentTaskId, + }, + }), }); - if (prior?.sourceHash !== candidate.sourceHash) { - await tx.insert(schema.project.currentPlanEvidence).values({ - projectId, - taskId: id, - version: candidate.version, - sourceRevision: candidate.sourceRevision, - sourceHash: candidate.sourceHash, - capturedAt: candidate.capturedAt, - snapshot: candidate, - }); - } } if (row && row.deletedAt != null) { return { deletedAt: row.deletedAt as string };