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) <noreply@runfusion.ai>
This commit is contained in:
7
.changeset/fn-8969-plan-evidence-version-collision.md
Normal file
7
.changeset/fn-8969-plan-evidence-version-collision.md
Normal file
@@ -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.
|
||||||
@@ -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<PlanEvidenceAppendError>));
|
||||||
|
});
|
||||||
@@ -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);
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -79,6 +79,7 @@ import { GlobalSettingsStore } from "./config/global-settings.js";
|
|||||||
import { Database } from "./db/db.js";
|
import { Database } from "./db/db.js";
|
||||||
import { ArchiveDatabase } from "./db/archive-db.js";
|
import { ArchiveDatabase } from "./db/archive-db.js";
|
||||||
import { projectScopeFor, type AsyncDataLayer, type DbTransaction } from "./postgres/data-layer.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 { planningLifecycleLockTransportAvailability, withPlanningLifecycleAdvisoryLock, withPlanningLifecycleAdvisoryLocks } from "./postgres/advisory-locks.js";
|
||||||
import { MissionStore } from "./missions/mission-store.js";
|
import { MissionStore } from "./missions/mission-store.js";
|
||||||
import { AsyncMissionStore } from "./async-stores/async-mission-store.js";
|
import { AsyncMissionStore } from "./async-stores/async-mission-store.js";
|
||||||
@@ -1197,16 +1198,19 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
|
|||||||
return refineTaskImpl(this, id, feedback);
|
return refineTaskImpl(this, id, feedback);
|
||||||
}
|
}
|
||||||
/**
|
/**
|
||||||
* FNXC:SpecLock 2026-08-09-07:06:
|
* FNXC:SpecLock 2026-08-11-02:04:
|
||||||
* FN-8845 retains plan snapshots outside the mutable task row. Backend-only storage is deliberate:
|
* Evidence append uses the shared column-versioned, conflict-tolerant path. sourceHash remains
|
||||||
* SQLite is no longer a runtime path and silently falling back would lose audit history on restart.
|
* 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<import("./planner/spec-lock.js").CurrentPlanEvidence> {
|
async appendCurrentPlanEvidence(taskId: string, evidence: import("./planner/spec-lock.js").CurrentPlanEvidence): Promise<import("./planner/spec-lock.js").CurrentPlanEvidence> {
|
||||||
if (!this.asyncLayer) throw new Error("Spec-lock history requires PostgreSQL backend storage");
|
if (!this.asyncLayer) throw new Error("Spec-lock history requires PostgreSQL backend storage");
|
||||||
const projectId = this.asyncLayer.projectId ?? "";
|
const result = await appendPlanEvidenceInTransaction(this.asyncLayer.db, {
|
||||||
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();
|
projectId: this.asyncLayer.projectId,
|
||||||
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);
|
taskId,
|
||||||
return rows[0]!.snapshot as import("./planner/spec-lock.js").CurrentPlanEvidence;
|
buildEvidence: (version) => ({ ...evidence, version: Math.max(version, evidence.version) }),
|
||||||
|
});
|
||||||
|
return result.evidence;
|
||||||
}
|
}
|
||||||
async getLatestCurrentPlanEvidence(taskId: string): Promise<CurrentPlanEvidence | undefined> {
|
async getLatestCurrentPlanEvidence(taskId: string): Promise<CurrentPlanEvidence | undefined> {
|
||||||
if (!this.asyncLayer) throw new Error("Spec-lock history requires PostgreSQL backend storage");
|
if (!this.asyncLayer) throw new Error("Spec-lock history requires PostgreSQL backend storage");
|
||||||
@@ -1241,16 +1245,20 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
|
|||||||
sourceRevision: number,
|
sourceRevision: number,
|
||||||
liveTask?: Pick<Task, "dependencies" | "missionId" | "sliceId" | "sourceParentTaskId">,
|
liveTask?: Pick<Task, "dependencies" | "missionId" | "sliceId" | "sourceParentTaskId">,
|
||||||
): Promise<CurrentPlanEvidence> {
|
): Promise<CurrentPlanEvidence> {
|
||||||
const [prior, task] = await Promise.all([this.getLatestCurrentPlanEvidence(taskId), liveTask ?? this.getTask(taskId)]);
|
const task = liveTask ?? await this.getTask(taskId);
|
||||||
const candidate = createCurrentPlanEvidence({
|
if (!this.asyncLayer) throw new Error("Spec-lock history requires PostgreSQL backend storage");
|
||||||
version: (prior?.version ?? 0) + 1,
|
const result = await appendPlanEvidenceInTransaction(this.asyncLayer.db, {
|
||||||
sourceRevision,
|
projectId: this.asyncLayer.projectId,
|
||||||
capturedAt: new Date().toISOString(),
|
taskId,
|
||||||
prompt,
|
buildEvidence: (version) => createCurrentPlanEvidence({
|
||||||
bindings: specLockBindings(task),
|
version,
|
||||||
|
sourceRevision,
|
||||||
|
capturedAt: new Date().toISOString(),
|
||||||
|
prompt,
|
||||||
|
bindings: specLockBindings(task),
|
||||||
|
}),
|
||||||
});
|
});
|
||||||
if (prior?.sourceHash === candidate.sourceHash) return prior;
|
return result.evidence;
|
||||||
return this.appendCurrentPlanEvidence(taskId, candidate);
|
|
||||||
}
|
}
|
||||||
async appendSpecLock(taskId: string, lock: import("./planner/spec-lock.js").SpecLock): Promise<import("./planner/spec-lock.js").SpecLock> {
|
async appendSpecLock(taskId: string, lock: import("./planner/spec-lock.js").SpecLock): Promise<import("./planner/spec-lock.js").SpecLock> {
|
||||||
if (!this.asyncLayer) throw new Error("Spec-lock history requires PostgreSQL backend storage");
|
if (!this.asyncLayer) throw new Error("Spec-lock history requires PostgreSQL backend storage");
|
||||||
|
|||||||
@@ -27,11 +27,12 @@
|
|||||||
* integration tests consume. They program against the stable `AsyncDataLayer`
|
* integration tests consume. They program against the stable `AsyncDataLayer`
|
||||||
* interface (U4), not the underlying driver.
|
* 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 * 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 { 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";
|
import { storeLog } from "../../store.js";
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -186,11 +187,11 @@ export async function removeLineageReferences(
|
|||||||
const evidenceUnavailableChildIds: string[] = [];
|
const evidenceUnavailableChildIds: string[] = [];
|
||||||
let evidenceInsertAttempts = 0;
|
let evidenceInsertAttempts = 0;
|
||||||
/*
|
/*
|
||||||
FNXC:SpecLockLineageInvalidation 2026-08-10-14:47:
|
FNXC:SpecLockLineageInvalidation 2026-08-11-02:04:
|
||||||
Evidence inserts use the trigger-compatible blank value, while reads preserve an unbound layer's
|
FN-8969 converges lineage writes on the shared helper: it writes the trigger-compatible blank
|
||||||
project-agnostic scope instead of filtering for the blank value the trigger rewrites.
|
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) {
|
for (const child of returned) {
|
||||||
const prompt = promptByChildId.get(child.id);
|
const prompt = promptByChildId.get(child.id);
|
||||||
if (prompt === undefined) {
|
if (prompt === undefined) {
|
||||||
@@ -199,24 +200,24 @@ export async function removeLineageReferences(
|
|||||||
storeLog.warn(`[spec-lock] lineage evidence unavailable: missing PROMPT.md for ${child.id}`);
|
storeLog.warn(`[spec-lock] lineage evidence unavailable: missing PROMPT.md for ${child.id}`);
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
let resolved = false;
|
try {
|
||||||
for (let attempt = 0; attempt < 3 && !resolved; attempt += 1) {
|
const result = await appendPlanEvidenceInTransaction(tx, {
|
||||||
const latest = await tx.select({ version: schema.project.currentPlanEvidence.version }).from(schema.project.currentPlanEvidence)
|
projectId,
|
||||||
.where(and(projectScopeFor(schema.project.currentPlanEvidence.projectId, projectId), eq(schema.project.currentPlanEvidence.taskId, child.id))).orderBy(desc(schema.project.currentPlanEvidence.version)).limit(1);
|
taskId: child.id,
|
||||||
const computed = (latest[0]?.version ?? 0) + 1;
|
maxAttempts: 3,
|
||||||
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 } });
|
buildEvidence: (computed, attempt) => {
|
||||||
evidenceInsertAttempts += 1;
|
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 });
|
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 } });
|
||||||
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)
|
validateMatchedEvidence: (snapshot, _version, attempt) => {
|
||||||
.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 (snapshot.plan.sections.lineage.canonical.split("\n").includes(`parent-task:${parentId}`)) throw new LineageEvidenceAppendError(child.id, attempt + 1, "matched-row-not-truthful");
|
||||||
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, result.version);
|
||||||
evidenceVersionByChild.set(child.id, matching[0].version); resolved = true; break;
|
} 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 };
|
return { clearedChildIds: returned.map((row) => row.id), evidenceVersionByChild, evidenceUnavailableChildIds, evidenceInsertAttempts };
|
||||||
}
|
}
|
||||||
|
|||||||
77
packages/core/src/task-store/plan-evidence.ts
Normal file
77
packages/core/src/task-store/plan-evidence.ts
Normal file
@@ -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<AppendPlanEvidenceResult> {
|
||||||
|
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);
|
||||||
|
}
|
||||||
@@ -13,7 +13,7 @@ import { resolveCapacityPoolId } from "../workflows/workflow-capacity.js";
|
|||||||
import {resolveWorkflowIntakeFacts} from "./task-creation.js";
|
import {resolveWorkflowIntakeFacts} from "./task-creation.js";
|
||||||
import {TransitionRejectionError} from "./errors.js";
|
import {TransitionRejectionError} from "./errors.js";
|
||||||
import * as schema from "../postgres/schema/index.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 {mkdir, writeFile} from "node:fs/promises";
|
||||||
import {join} from "node:path";
|
import {join} from "node:path";
|
||||||
import type {Task, ColumnId, CheckoutClaimPrecondition, ActivityLogEntry, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, GoalCitation, GoalCitationFilter} from "../types.js";
|
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 {applyOriginalDescription} from "../tasks/original-description-policy.js";
|
||||||
import {isPlanReviewSatisfied} from "../planner/plan-approval.js";
|
import {isPlanReviewSatisfied} from "../planner/plan-approval.js";
|
||||||
import {createCurrentPlanEvidence} from "../planner/spec-lock.js";
|
import {createCurrentPlanEvidence} from "../planner/spec-lock.js";
|
||||||
|
import {appendPlanEvidenceInTransaction} from "./plan-evidence.js";
|
||||||
import {recordRunAuditEvent as recordRunAuditEventAsync} from "../postgres/data-layer.js";
|
import {recordRunAuditEvent as recordRunAuditEventAsync} from "../postgres/data-layer.js";
|
||||||
import {listGoalCitations as listGoalCitationsAsync} from "./async/async-events.js";
|
import {listGoalCitations as listGoalCitationsAsync} from "./async/async-events.js";
|
||||||
import type {RunAuditEventRow} from "../task-store/row-types.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 persist = async () => {
|
||||||
const row = await readTaskRowInTransaction(tx, id, { includeDeleted: true }, layer.projectId);
|
const row = await readTaskRowInTransaction(tx, id, { includeDeleted: true }, layer.projectId);
|
||||||
/*
|
/*
|
||||||
FNXC:SpecLock 2026-08-09-18:17:
|
FNXC:SpecLock 2026-08-11-02:04:
|
||||||
A full PROMPT.md rewrite publishes its evidence and clears approval in this one task-row
|
A full PROMPT.md rewrite publishes evidence and clears approval in this task-row transaction,
|
||||||
transaction. A database rollback therefore cannot leave either half visible by itself.
|
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) {
|
if (specPlanPrompt !== undefined) {
|
||||||
const projectId = layer.projectId ?? "";
|
await appendPlanEvidenceInTransaction(tx, {
|
||||||
const priorRows = await tx.select().from(schema.project.currentPlanEvidence)
|
projectId: layer.projectId,
|
||||||
.where(and(eq(schema.project.currentPlanEvidence.projectId, projectId), eq(schema.project.currentPlanEvidence.taskId, id)))
|
taskId: id,
|
||||||
.orderBy(desc(schema.project.currentPlanEvidence.version)).limit(1);
|
buildEvidence: (version) => createCurrentPlanEvidence({
|
||||||
const prior = priorRows[0]?.snapshot as import("../planner/spec-lock.js").CurrentPlanEvidence | undefined;
|
version,
|
||||||
const candidate = createCurrentPlanEvidence({
|
sourceRevision: Date.now(),
|
||||||
version: (prior?.version ?? 0) + 1,
|
capturedAt: new Date().toISOString(),
|
||||||
sourceRevision: Date.now(),
|
prompt: specPlanPrompt,
|
||||||
capturedAt: new Date().toISOString(),
|
bindings: {
|
||||||
prompt: specPlanPrompt,
|
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) {
|
if (row && row.deletedAt != null) {
|
||||||
return { deletedAt: row.deletedAt as string };
|
return { deletedAt: row.deletedAt as string };
|
||||||
|
|||||||
Reference in New Issue
Block a user