diff --git a/.changeset/harden-task-id-overwrite-guards.md b/.changeset/harden-task-id-overwrite-guards.md new file mode 100644 index 000000000..4a268b2dd --- /dev/null +++ b/.changeset/harden-task-id-overwrite-guards.md @@ -0,0 +1,5 @@ +--- +"@runfusion/fusion": patch +--- + +Harden task creation so stale allocator state or colliding reservations fail safely instead of overwriting an existing task row or task directory. diff --git a/docs/architecture.md b/docs/architecture.md index 958c53e36..609b413b7 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -674,12 +674,15 @@ A lease is recoverable only when there is **no active local executor session for - Reserve/commit/abort execute under a process-local lock and a single SQLite transaction. Lazy reservation expiry cleanup runs inside those same transactions. - Default reservation TTL is `15 * 60 * 1000` ms (15 minutes). Expired/aborted reservations are **burned IDs** and are never reissued. - `committedClusterTaskCount` from allocator state is the only authoritative cluster-wide committed-task count. Local task-row counts and ID suffix math are not authoritative. + - Store open reconciles every known prefix in `distributed_task_id_state` to `max(current nextSequence, max(tasks suffix)+1, max(archivedTasks suffix)+1, max(reservation sequence)+1)`. This self-heals stale counters before ordinary task creation resumes. - Mesh allocator write routes (`/api/mesh/task-ids/reserve|commit|abort`) return `503` when the coordinator node is unreachable; they never fall back to local-only cluster ID issuance. - Cluster task creation now uses a strong-write reserve → create → replicate → commit/abort sequence. - Ordinary local task creation (`TaskStore.createTask()`, duplicate, and refine flows) now allocates IDs through the same distributed reserve/commit/abort lifecycle owned by `TaskStore`. - `POST /api/tasks` uses the store-owned allocator path for local creates rather than maintaining a separate route-local allocator implementation. - `POST /api/tasks` reserves a distributed ID, creates the authoritative local task with that reserved ID, then POSTs authenticated replication payloads to peer nodes. - - Creation self-heals stale ID overlap state: if a reserved `FN-*` collides with an existing task (`Task ID already exists...` or replicated-create collision), the route aborts that reservation, cleans up partial local state, reserves the next ID, and retries up to a bounded limit. + - All create-class writes now use conflict-raising inserts, not SQLite `ON CONFLICT ... DO UPDATE`. Existing task rows and `.fusion/tasks/{id}` contents always win over stale counters or colliding reservations. + - Local create paths perform a final active+archived existence check immediately before insert. If a reserved `FN-*` still collides, the reservation is aborted/burned and the create fails loudly instead of rewriting the existing task. + - Creation self-heals stale overlap state at the route layer: if a reserved `FN-*` collides with an existing task (`Task ID already exists...` or replicated-create collision), the route aborts that reservation, cleans up partial local state, reserves the next ID, and retries up to a bounded limit. - Replica apply uses `TaskStore.applyReplicatedTaskCreate(...)`, which is idempotent by task ID: replaying the same payload returns the existing task without creating duplicates. - If an incoming replicated payload conflicts with a different existing task record for the same ID, the apply path returns a deterministic collision error instead of overwriting data. - Any replication/coordinator failure aborts the reservation and returns write failure (`503`), so this path does not report success for local-only partial writes. diff --git a/docs/storage.md b/docs/storage.md index b4e3e425c..cc6059130 100644 --- a/docs/storage.md +++ b/docs/storage.md @@ -4,8 +4,9 @@ - `distributed_task_id_state` is the authoritative local task-ID allocator state. `nextSequence` is the active high-water mark used for local ID reservations. - `distributed_task_id_reservations` tracks reserve/commit/abort lifecycle entries. Aborted/expired reservations are burned and never reissued. -- `config.nextId` is retained only as a legacy compatibility field and optional seed source; runtime task creation no longer mutates it as allocator truth. -- Startup allocator reconciliation bumps each active prefix sequence to `max(current nextSequence, max(existing task suffix)+1)` across live + archived tasks to self-heal stale allocator drift. +- `config.nextId` is retained only as a deprecated legacy compatibility field and optional one-time seed source. Fusion still reads it during reconciliation, but runtime task creation and settings writes no longer mutate it. +- Startup/store-open allocator reconciliation bumps each active prefix sequence to `max(current nextSequence, max(tasks suffix)+1, max(archivedTasks suffix)+1, max(reservation sequence)+1)` so stale allocator rows self-heal before local task creation resumes. +- Create-class task persistence is intentionally non-destructive: new tasks use plain `INSERT` semantics, while `ON CONFLICT(id) DO UPDATE` remains update-only. If counters drift and a reserved ID still collides, the create fails and the existing SQLite row / task directory stays intact. ## SQLite write-path lock recovery (FN-4042 / FN-4083) diff --git a/docs/task-management.md b/docs/task-management.md index b9808581e..3369d9466 100644 --- a/docs/task-management.md +++ b/docs/task-management.md @@ -369,6 +369,26 @@ Archive entries preserve key metadata needed for restoration, including: - Moves task to `done` - Logs “Task restored from archive” when recovering from compact archive entry +### Task-ID collision safety and operator recovery + +- Ordinary task creation, duplicate, and refine flows now fail safely if the chosen task ID already exists in active storage or archive storage. Existing task rows/files always win; the new create attempt must retry with a fresh reservation instead of overwriting data. +- A failed create may burn a distributed reservation. Gaps in `FN-*` numbering are expected and are safer than reissuing a possibly-colliding ID. +- `config.nextId` is legacy/read-only. The live allocator state is `distributed_task_id_state.nextSequence`, reconciled on store open against live tasks, archived task snapshots, and reservation history. + +If you suspect **historical overwrites from pre-FN-4044 builds**, inspect surviving evidence in this order: + +1. `archive.db` / archived task snapshots for the missing ID +2. `.fusion/tasks//task.json.bak`, `PROMPT.md`, attachments, and any surviving worktree branch named for the task +3. agent run logs / task documents / activity log entries that still mention the original ID +4. git commits whose subject/body references the original task ID but no longer matches the current task metadata + +Recovery/backfill guidance: + +- If the original task row still exists in archive storage, unarchive or manually recreate the task from that snapshot. +- If only prompt/worktree/git evidence survives, create a replacement task with a new ID and copy over the recovered description, prompt, documents, and attachments manually. +- If both the active row and archive snapshot were overwritten, Fusion cannot reconstruct lost attachments/comments automatically; recreate them from git history, branch/worktree contents, screenshots, or external issue trackers. +- Record the incident in the replacement task so future audits understand why the task ID and commit history diverge. + ## GitHub Issue Import and PR Creation Import issues: diff --git a/packages/core/src/__tests__/distributed-task-id.test.ts b/packages/core/src/__tests__/distributed-task-id.test.ts index 73397f410..a5a494870 100644 --- a/packages/core/src/__tests__/distributed-task-id.test.ts +++ b/packages/core/src/__tests__/distributed-task-id.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from "vitest"; import { Database } from "../db.js"; -import { createDistributedTaskIdAllocator, DistributedTaskIdError } from "../distributed-task-id.js"; +import { createDistributedTaskIdAllocator, DistributedTaskIdError, reconcileTaskIdState } from "../distributed-task-id.js"; describe("distributed-task-id allocator", () => { const createAllocator = () => { @@ -84,6 +84,33 @@ describe("distributed-task-id allocator", () => { expect(state.nextSequence).toBe(3702); }); + it("reconciles stale state rows past live tasks, archived tasks, and reservations", () => { + const db = new Database("/tmp/fusion-test", { inMemory: true }); + db.init(); + const now = new Date().toISOString(); + + db.prepare( + "INSERT INTO tasks (id, description, \"column\", createdAt, updatedAt) VALUES (?, '', 'todo', ?, ?)", + ).run("FN-003", now, now); + db.prepare( + "INSERT INTO archivedTasks (id, data, archivedAt) VALUES (?, ?, ?)", + ).run("FN-005", JSON.stringify({ id: "FN-005" }), now); + db.prepare( + "INSERT INTO distributed_task_id_state (prefix, nextSequence, committedClusterTaskCount, lastCommittedTaskId, updatedAt) VALUES (?, ?, ?, ?, ?)", + ).run("FN", 2, 1, "FN-001", now); + db.prepare( + `INSERT INTO distributed_task_id_reservations ( + reservationId, prefix, nodeId, sequence, taskId, status, reason, expiresAt, createdAt, updatedAt + ) VALUES (?, ?, ?, ?, ?, 'reserved', NULL, ?, ?, ?)`, + ).run("res-7", "FN", "node-a", 7, "FN-007", new Date(Date.now() + 60_000).toISOString(), now, now); + + const reconciled = reconcileTaskIdState(db); + expect(reconciled).toContain("FN"); + + const state = db.prepare("SELECT nextSequence FROM distributed_task_id_state WHERE prefix = ?").get("FN") as { nextSequence: number }; + expect(state.nextSequence).toBe(8); + }); + it("skips stale overlapping nextSequence values and reserves the next free id", async () => { const db = new Database("/tmp/fusion-test", { inMemory: true }); db.init(); @@ -106,6 +133,26 @@ describe("distributed-task-id allocator", () => { expect(state.committedClusterTaskCount).toBe(1); }); + it("reconciles stale reservation sequences before allocating a new reservation", async () => { + const db = new Database("/tmp/fusion-test", { inMemory: true }); + db.init(); + const now = new Date().toISOString(); + db.prepare( + "INSERT INTO distributed_task_id_state (prefix, nextSequence, committedClusterTaskCount, lastCommittedTaskId, updatedAt) VALUES (?, ?, ?, ?, ?)", + ).run("FN", 2, 1, "FN-001", now); + db.prepare( + `INSERT INTO distributed_task_id_reservations ( + reservationId, prefix, nodeId, sequence, taskId, status, reason, expiresAt, createdAt, updatedAt + ) VALUES (?, ?, ?, ?, ?, 'reserved', NULL, ?, ?, ?)`, + ).run("res-2", "FN", "node-a", 2, "FN-002", new Date(Date.now() + 60_000).toISOString(), now, now); + + const allocator = createDistributedTaskIdAllocator(db); + const reservation = await allocator.reserveDistributedTaskId({ prefix: "FN", nodeId: "node-b" }); + + expect(reservation.taskId).toBe("FN-003"); + expect(reservation.sequence).toBe(3); + }); + it("state reports committed count independently from nextSequence", async () => { const { allocator } = createAllocator(); const first = await allocator.reserveDistributedTaskId({ prefix: "FN", nodeId: "node-a" }); diff --git a/packages/core/src/__tests__/store-create-collision.test.ts b/packages/core/src/__tests__/store-create-collision.test.ts new file mode 100644 index 000000000..d3d57065a --- /dev/null +++ b/packages/core/src/__tests__/store-create-collision.test.ts @@ -0,0 +1,117 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { readFile } from "node:fs/promises"; +import { join } from "node:path"; + +import { createTaskStoreTestHarness } from "./store-test-helpers.js"; + +describe("TaskStore collision guards", () => { + const harness = createTaskStoreTestHarness(); + let store = harness.store(); + let rootDir = harness.rootDir(); + + beforeEach(async () => { + await harness.beforeEach(); + store = harness.store(); + rootDir = harness.rootDir(); + }); + + afterEach(async () => { + await harness.afterEach(); + vi.restoreAllMocks(); + }); + + const forceAllocatorCollision = (taskId: string) => { + const allocator = store.getDistributedTaskIdAllocator(); + vi.spyOn(allocator, "reserveDistributedTaskId").mockResolvedValue({ + reservationId: `res-${taskId}`, + taskId, + sequence: Number.parseInt(taskId.split("-")[1] ?? "0", 10), + expiresAt: new Date(Date.now() + 60_000).toISOString(), + committedClusterTaskCount: 0, + }); + vi.spyOn(allocator, "commitDistributedTaskIdReservation").mockResolvedValue({ + reservationId: `res-${taskId}`, + taskId, + sequence: Number.parseInt(taskId.split("-")[1] ?? "0", 10), + committedAt: new Date().toISOString(), + committedClusterTaskCount: 1, + }); + vi.spyOn(allocator, "abortDistributedTaskIdReservation").mockResolvedValue({ + reservationId: `res-${taskId}`, + taskId, + sequence: Number.parseInt(taskId.split("-")[1] ?? "0", 10), + abortedAt: new Date().toISOString(), + committedClusterTaskCount: 0, + reason: "failed-create", + }); + }; + + it("createTask throws and preserves the existing task when the allocator returns a colliding id", async () => { + const original = await store.createTask({ title: "Original", description: "original task", column: "todo" }); + const originalPromptPath = join(rootDir, ".fusion", "tasks", original.id, "PROMPT.md"); + const originalPrompt = await readFile(originalPromptPath, "utf8"); + + forceAllocatorCollision(original.id); + await expect( + store.createTask({ title: "Replacement", description: "replacement task", column: "todo" }), + ).rejects.toThrow(`Task ID already exists: ${original.id}`); + + const persisted = await store.getTask(original.id); + const promptAfter = await readFile(originalPromptPath, "utf8"); + + expect(persisted.title).toBe("Original"); + expect(persisted.description).toBe("original task"); + expect(promptAfter).toBe(originalPrompt); + }); + + it("duplicateTask throws and preserves the unrelated task when its reserved id collides", async () => { + const source = await store.createTask({ title: "Source", description: "source task" }); + const victim = await store.createTask({ title: "Victim", description: "victim task", column: "todo" }); + + forceAllocatorCollision(victim.id); + await expect(store.duplicateTask(source.id)).rejects.toThrow(`Task ID already exists: ${victim.id}`); + + const persisted = await store.getTask(victim.id); + expect(persisted.title).toBe("Victim"); + expect(persisted.description).toBe("victim task"); + expect(persisted.sourceParentTaskId).toBeUndefined(); + }); + + it("refineTask throws and preserves the unrelated task when its reserved id collides", async () => { + const source = await store.createTask({ title: "Source", description: "source task", column: "todo" }); + await store.moveTask(source.id, "in-progress"); + await store.moveTask(source.id, "in-review"); + await store.moveTask(source.id, "done"); + const victim = await store.createTask({ title: "Victim", description: "victim task", column: "todo" }); + + forceAllocatorCollision(victim.id); + await expect(store.refineTask(source.id, "apply polish")).rejects.toThrow(`Task ID already exists: ${victim.id}`); + + const persisted = await store.getTask(victim.id); + expect(persisted.title).toBe("Victim"); + expect(persisted.description).toBe("victim task"); + expect(persisted.dependencies).toEqual([]); + }); + + it("createTask rejects archived-id collisions from stale distributed_task_id_state without overwriting archive data", async () => { + const archived = await store.createTask({ title: "Archived", description: "archived task", column: "todo" }); + await store.moveTask(archived.id, "in-progress"); + await store.moveTask(archived.id, "in-review"); + await store.moveTask(archived.id, "done"); + const archivedDetail = await store.getTask(archived.id); + await store.archiveTask(archived.id); + + const archivedPrefix = archived.id.split("-")[0]; + store.getDatabase().prepare("DELETE FROM distributed_task_id_reservations WHERE prefix = ?").run(archivedPrefix); + store.getDatabase().prepare("UPDATE distributed_task_id_state SET nextSequence = 1 WHERE prefix = ?").run(archivedPrefix); + + await expect(store.createTask({ title: "New", description: "new task" })).rejects.toThrow( + `Task ID already exists: ${archived.id}`, + ); + + const preservedArchive = await store.getTask(archived.id); + expect(preservedArchive.title).toBe(archivedDetail.title); + expect(preservedArchive.description).toBe(archivedDetail.description); + expect(preservedArchive.prompt).toBe(archivedDetail.prompt); + }); +}); diff --git a/packages/core/src/__tests__/store-migration.test.ts b/packages/core/src/__tests__/store-migration.test.ts index 5900b1752..892fb94c5 100644 --- a/packages/core/src/__tests__/store-migration.test.ts +++ b/packages/core/src/__tests__/store-migration.test.ts @@ -109,35 +109,35 @@ describe("TaskStore", () => { }); }); - describe("FTS5 corruption recovery during upsert", () => { - it("rebuilds FTS5 and retries once when upsert fails with an FTS corruption error", async () => { + describe("FTS5 corruption recovery during create inserts", () => { + it("rebuilds FTS5 and retries once when an insert fails with an FTS corruption error", async () => { const db = harness.store().getDatabase(); const rebuildSpy = vi.spyOn(db, "rebuildFts5Index").mockReturnValue(true); - const upsertSpy = vi.spyOn(harness.store() as any, "upsertTask"); - const originalUpsert = upsertSpy.getMockImplementation(); - upsertSpy + const insertSpy = vi.spyOn(harness.store() as any, "insertTask"); + const originalInsert = insertSpy.getMockImplementation(); + insertSpy .mockImplementationOnce(() => { throw new Error("SQLITE_CORRUPT: corruption found reading blob in fts5"); }) .mockImplementation((task: any) => { - if (originalUpsert) { - return originalUpsert(task); + if (originalInsert) { + return originalInsert(task); } - return (Object.getPrototypeOf(harness.store()) as any).upsertTask.call(harness.store(), task); + return (Object.getPrototypeOf(harness.store()) as any).insertTask.call(harness.store(), task); }); const created = await harness.store().createTask({ description: "Recover from FTS corruption" }); expect(created.id).toBeDefined(); expect(rebuildSpy).toHaveBeenCalledTimes(1); - expect(upsertSpy).toHaveBeenCalledTimes(2); + expect(insertSpy).toHaveBeenCalledTimes(2); }); it("propagates non-FTS errors without rebuild", async () => { const db = harness.store().getDatabase(); const rebuildSpy = vi.spyOn(db, "rebuildFts5Index").mockReturnValue(true); - vi.spyOn(harness.store() as any, "upsertTask").mockImplementationOnce(() => { + vi.spyOn(harness.store() as any, "insertTask").mockImplementationOnce(() => { throw new Error("constraint failed"); }); diff --git a/packages/core/src/__tests__/test-project.test.ts b/packages/core/src/__tests__/test-project.test.ts index e32a9c66b..b57788ae3 100644 --- a/packages/core/src/__tests__/test-project.test.ts +++ b/packages/core/src/__tests__/test-project.test.ts @@ -42,7 +42,7 @@ describe("test-project fixture", () => { const configRaw = await readFile(join(fixture.rootDir, ".fusion", "config.json"), "utf-8"); const config = JSON.parse(configRaw); - expect(config.nextId).toBe(1); + expect(config.nextId).toBeUndefined(); expect(config.settings.taskPrefix).toBe("FN"); const tasks = await fixture.store.listTasks(); diff --git a/packages/core/src/db.ts b/packages/core/src/db.ts index cad783ddd..53ff548ab 100644 --- a/packages/core/src/db.ts +++ b/packages/core/src/db.ts @@ -263,6 +263,9 @@ CREATE TABLE IF NOT EXISTS tasks ( ); -- Config table (single row with project settings) +-- nextId is a deprecated legacy allocator counter retained read-only for one +-- release so older databases/config consumers can still load it during the +-- distributed_task_id_state transition. CREATE TABLE IF NOT EXISTS config ( id INTEGER PRIMARY KEY CHECK (id = 1), nextId INTEGER DEFAULT 1, diff --git a/packages/core/src/distributed-task-id.ts b/packages/core/src/distributed-task-id.ts index 0a3f2727c..e0ce8357a 100644 --- a/packages/core/src/distributed-task-id.ts +++ b/packages/core/src/distributed-task-id.ts @@ -12,6 +12,7 @@ import type { } from "./types.js"; const DEFAULT_RESERVATION_TTL_MS = 15 * 60 * 1000; +const TASK_ID_PATTERN = /^([A-Z][A-Z0-9]*)-(\d+)$/; export interface DistributedTaskIdAllocator { formatDistributedTaskId(prefix: string, sequence: number): string; @@ -56,6 +57,164 @@ type ReservationRow = { abortedAt: string | null; }; +function parseTaskId(taskId: string): { prefix: string; sequence: number } | null { + const match = taskId.trim().toUpperCase().match(TASK_ID_PATTERN); + if (!match) { + return null; + } + + const sequence = Number.parseInt(match[2], 10); + if (!Number.isFinite(sequence)) { + return null; + } + + return { prefix: match[1], sequence }; +} + +function getConfiguredPrefixAndLegacyNextId(db: Database): { prefix: string; nextId: number | null } { + try { + const row = db + .prepare("SELECT nextId, settings FROM config WHERE id = 1") + .get() as { nextId: number | null; settings: string | null } | undefined; + if (!row) { + return { prefix: "KB", nextId: null }; + } + + const settings = row.settings ? (JSON.parse(row.settings) as { taskPrefix?: string }) : null; + return { + prefix: (settings?.taskPrefix ?? "KB").trim().toUpperCase(), + nextId: typeof row.nextId === "number" ? row.nextId : null, + }; + } catch { + return { prefix: "KB", nextId: null }; + } +} + +function getKnownPrefixes(db: Database): Set { + const prefixes = new Set(); + const configured = getConfiguredPrefixAndLegacyNextId(db).prefix; + if (configured) { + prefixes.add(configured); + } + + const addFromQuery = (sql: string, mapper: (row: Record) => string | undefined): void => { + try { + const rows = db.prepare(sql).all() as Array>; + for (const row of rows) { + const prefix = mapper(row)?.trim().toUpperCase(); + if (prefix) { + prefixes.add(prefix); + } + } + } catch { + // Best-effort for tests / partial schemas. + } + }; + + addFromQuery("SELECT prefix FROM distributed_task_id_state", (row) => row.prefix as string | undefined); + addFromQuery("SELECT prefix FROM distributed_task_id_reservations", (row) => row.prefix as string | undefined); + addFromQuery("SELECT id FROM tasks", (row) => parseTaskId(String(row.id ?? ""))?.prefix); + addFromQuery("SELECT id FROM archivedTasks", (row) => parseTaskId(String(row.id ?? ""))?.prefix); + + return prefixes; +} + +function getMaxTaskSequenceFromTable(db: Database, table: string, prefix: string): number { + try { + const rows = db.prepare(`SELECT id FROM ${table} WHERE id LIKE ?`).all(`${prefix}-%`) as Array<{ id: string }>; + let maxSequence = 0; + for (const row of rows) { + const parsed = parseTaskId(row.id); + if (parsed?.prefix === prefix && parsed.sequence > maxSequence) { + maxSequence = parsed.sequence; + } + } + return maxSequence; + } catch { + return 0; + } +} + +function getMaxReservationSequence(db: Database, prefix: string): number { + try { + const row = db + .prepare("SELECT MAX(sequence) AS maxSeq FROM distributed_task_id_reservations WHERE prefix = ?") + .get(prefix) as { maxSeq: number | null } | undefined; + return typeof row?.maxSeq === "number" ? row.maxSeq : 0; + } catch { + return 0; + } +} + +function getNextSequenceFloor(db: Database, prefix: string): number { + const configured = getConfiguredPrefixAndLegacyNextId(db); + let nextSequence = 1; + + if (configured.prefix === prefix && configured.nextId && configured.nextId > nextSequence) { + nextSequence = configured.nextId; + } + + const taskHighWaterMark = getMaxTaskSequenceFromTable(db, "tasks", prefix) + 1; + const archivedHighWaterMark = getMaxTaskSequenceFromTable(db, "archivedTasks", prefix) + 1; + const reservationHighWaterMark = getMaxReservationSequence(db, prefix) + 1; + + nextSequence = Math.max(nextSequence, taskHighWaterMark, archivedHighWaterMark, reservationHighWaterMark); + return nextSequence; +} + +function ensureStateRow(db: Database, prefix: string): void { + const nowIso = new Date().toISOString(); + const nextSequence = getNextSequenceFloor(db, prefix); + db.prepare( + `INSERT OR IGNORE INTO distributed_task_id_state ( + prefix, nextSequence, committedClusterTaskCount, lastCommittedTaskId, updatedAt + ) VALUES (?, ?, 0, NULL, ?)`, + ).run(prefix, nextSequence, nowIso); + db.prepare( + `UPDATE distributed_task_id_state + SET nextSequence = MAX(nextSequence, ?), + updatedAt = ? + WHERE prefix = ?`, + ).run(nextSequence, nowIso, prefix); +} + +export function reconcileTaskIdState(db: Database): string[] { + const nowIso = new Date().toISOString(); + return db.transaction(() => { + const reconciled: string[] = []; + for (const prefix of getKnownPrefixes(db)) { + const nextSequence = getNextSequenceFloor(db, prefix); + db.prepare( + `INSERT OR IGNORE INTO distributed_task_id_state ( + prefix, nextSequence, committedClusterTaskCount, lastCommittedTaskId, updatedAt + ) VALUES (?, ?, 0, NULL, ?)`, + ).run(prefix, nextSequence, nowIso); + + const before = db + .prepare("SELECT nextSequence FROM distributed_task_id_state WHERE prefix = ?") + .get(prefix) as { nextSequence: number } | undefined; + db.prepare( + `UPDATE distributed_task_id_state + SET nextSequence = MAX(nextSequence, ?), + updatedAt = ? + WHERE prefix = ?`, + ).run(nextSequence, nowIso, prefix); + const after = db + .prepare("SELECT nextSequence FROM distributed_task_id_state WHERE prefix = ?") + .get(prefix) as { nextSequence: number } | undefined; + + if (!before || !after || after.nextSequence !== before.nextSequence) { + reconciled.push(prefix); + } + } + + if (reconciled.length > 0) { + db.bumpLastModified(); + } + return reconciled; + }); +} + export function formatDistributedTaskId(prefix: string, sequence: number): string { const normalizedPrefix = prefix.trim().toUpperCase(); if (!normalizedPrefix) { @@ -105,70 +264,6 @@ export function createDistributedTaskIdAllocator(db: Database): DistributedTaskI return existsInTable("tasks") || existsInTable("archivedTasks"); }; - const ensureStateRow = (prefix: string): void => { - // Seed nextSequence past any pre-existing task ID for this prefix. Without - // this, projects whose tasks were originally allocated through - // TaskStore.allocateId() (config.nextId) would have mesh-routed task - // creates restart at 1 and collide with historical FN-001 / FN-002 / … - // IDs (regression introduced when the dashboard task-create route was - // wired to reserveDistributedTaskId in FN-3450). - // - // We take the max of: - // - 1 (historical default) - // - the legacy config.nextId counter, when the configured taskPrefix - // matches `prefix` - // - one past the highest numeric suffix on any existing task for this - // prefix (live tasks + archived), to handle DBs where config.nextId - // ever drifted below the real high-water mark - let seedSequence = 1; - try { - const configRow = db - .prepare("SELECT nextId, settings FROM config WHERE id = 1") - .get() as { nextId: number | null; settings: string | null } | undefined; - if (configRow) { - const settings = configRow.settings ? (JSON.parse(configRow.settings) as { taskPrefix?: string }) : null; - const configuredPrefix = (settings?.taskPrefix ?? "KB").trim().toUpperCase(); - if (configuredPrefix === prefix && typeof configRow.nextId === "number" && configRow.nextId > seedSequence) { - seedSequence = configRow.nextId; - } - } - } catch { - // Best-effort: if the config row/column is missing (fresh test DB) we - // fall back to the historical default of 1. - } - const idPattern = `${prefix}-%`; - const probeTable = (table: string): void => { - try { - const row = db - .prepare( - `SELECT MAX(CAST(substr(id, ${prefix.length + 2}) AS INTEGER)) AS maxSeq - FROM ${table} - WHERE id LIKE ? AND substr(id, ${prefix.length + 2}) GLOB '[0-9]*'`, - ) - .get(idPattern) as { maxSeq: number | null } | undefined; - if (row && typeof row.maxSeq === "number" && row.maxSeq + 1 > seedSequence) { - seedSequence = row.maxSeq + 1; - } - } catch { - // Table may not exist (tests, isolated DBs); ignore. - } - }; - probeTable("tasks"); - probeTable("archivedTasks"); - const nowIso = new Date().toISOString(); - db.prepare( - `INSERT OR IGNORE INTO distributed_task_id_state ( - prefix, nextSequence, committedClusterTaskCount, lastCommittedTaskId, updatedAt - ) VALUES (?, ?, 0, NULL, ?)` - ).run(prefix, seedSequence, nowIso); - db.prepare( - `UPDATE distributed_task_id_state - SET nextSequence = MAX(nextSequence, ?), - updatedAt = ? - WHERE prefix = ?` - ).run(seedSequence, nowIso, prefix); - }; - return { formatDistributedTaskId, reserveDistributedTaskId: async (input) => @@ -184,7 +279,7 @@ export function createDistributedTaskIdAllocator(db: Database): DistributedTaskI if (!prefix) { throw new DistributedTaskIdError("prefix is required", "invalid_prefix"); } - ensureStateRow(prefix); + ensureStateRow(db, prefix); const state = db .prepare( @@ -203,7 +298,7 @@ export function createDistributedTaskIdAllocator(db: Database): DistributedTaskI db.prepare( `INSERT INTO distributed_task_id_reservations ( reservationId, prefix, nodeId, sequence, taskId, status, reason, expiresAt, createdAt, updatedAt - ) VALUES (?, ?, ?, ?, ?, 'reserved', NULL, ?, ?, ?)` + ) VALUES (?, ?, ?, ?, ?, 'reserved', NULL, ?, ?, ?)`, ).run(reservationId, prefix, input.nodeId, sequence, taskId, expiresAt, nowIso, nowIso); db.prepare( @@ -252,7 +347,7 @@ export function createDistributedTaskIdAllocator(db: Database): DistributedTaskI WHERE reservationId = ?`, ).run(nowIso, nowIso, row.reservationId); - ensureStateRow(row.prefix); + ensureStateRow(db, row.prefix); db.prepare( `UPDATE distributed_task_id_state SET committedClusterTaskCount = committedClusterTaskCount + 1, @@ -308,7 +403,7 @@ export function createDistributedTaskIdAllocator(db: Database): DistributedTaskI ).run(input.reason, nowIso, nowIso, row.reservationId); } - ensureStateRow(row.prefix); + ensureStateRow(db, row.prefix); const state = db .prepare( "SELECT committedClusterTaskCount FROM distributed_task_id_state WHERE prefix = ?", @@ -334,7 +429,7 @@ export function createDistributedTaskIdAllocator(db: Database): DistributedTaskI if (!prefix) { throw new DistributedTaskIdError("prefix is required", "invalid_prefix"); } - ensureStateRow(prefix); + ensureStateRow(db, prefix); const row = db .prepare( `SELECT nextSequence, committedClusterTaskCount, lastCommittedTaskId diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index f5d43d68e..f507c3ad1 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -28,7 +28,7 @@ import { validateNodeOverrideChange } from "./node-override-guard.js"; import { sanitizeTitle } from "./ai-summarize.js"; import { assertProjectRootDir } from "./project-root-guard.js"; import { generateTaskLineageId, normalizeTaskCommitAssociation } from "./task-lineage.js"; -import { createDistributedTaskIdAllocator, resolveLocalNodeId, type DistributedTaskIdAllocator } from "./distributed-task-id.js"; +import { createDistributedTaskIdAllocator, reconcileTaskIdState, resolveLocalNodeId, type DistributedTaskIdAllocator } from "./distributed-task-id.js"; import { buildBootstrapPrompt, replicationCollisionError, @@ -585,6 +585,8 @@ export class TaskStore extends EventEmitter { private worktreeAllocationLock: Promise = Promise.resolve(); /** Promise chain for serializing config.json read-modify-write cycles */ private configLock: Promise = Promise.resolve(); + /** Startup/open guard for distributed_task_id_state reconciliation. */ + private taskIdStateReconciled = false; /** Cached workflow steps — invalidated on create/update/delete */ private workflowStepsCache: import("./types.js").WorkflowStep[] | null = null; /** Plugin-contributed workflow step templates injected by engine runtime. */ @@ -680,6 +682,7 @@ export class TaskStore extends EventEmitter { throw error; } this._db = db; + this.reconcileDistributedTaskIdStateOnOpen(); // Auto-migrate legacy data if needed if (detectLegacyData(this.fusionDir)) { // Note: migrateFromLegacy is async but we need sync access. @@ -705,6 +708,14 @@ export class TaskStore extends EventEmitter { return this._archiveDb; } + private reconcileDistributedTaskIdStateOnOpen(): void { + if (this.taskIdStateReconciled) { + return; + } + reconcileTaskIdState(this.db); + this.taskIdStateReconciled = true; + } + async init(): Promise { await mkdir(this.tasksDir, { recursive: true }); @@ -719,6 +730,8 @@ export class TaskStore extends EventEmitter { } this._db = db; } + + this.reconcileDistributedTaskIdStateOnOpen(); // Auto-migrate from legacy file-based storage if (detectLegacyData(this.fusionDir)) { @@ -726,12 +739,14 @@ export class TaskStore extends EventEmitter { } await this.migrateActiveArchivedTasksToArchiveDb(); await this.importLegacyAgentLogsOnce(); + this.taskIdStateReconciled = false; + this.reconcileDistributedTaskIdStateOnOpen(); // Write config.json for backward compatibility if it doesn't exist if (!existsSync(this.configPath)) { const config = await this.readConfig(); try { - await writeFile(this.configPath, JSON.stringify(config, null, 2)); + await writeFile(this.configPath, this.serializeConfigForDisk(config)); } catch (err) { storeLog.warn("Backward-compat config.json sync failed during init", { phase: "init:config-sync", @@ -1215,8 +1230,129 @@ export class TaskStore extends EventEmitter { return [...columns, limitedLog].join(", "); } + private getTaskPersistValues(task: Task): unknown[] { + return [ + task.id, + task.lineageId ?? generateTaskLineageId(), + task.title ?? null, + task.description, + normalizeTaskPriority(task.priority), + task.column, + task.status ?? null, + task.size ?? null, + task.reviewLevel ?? null, + task.currentStep || 0, + task.worktree ?? null, + task.blockedBy ?? null, + task.paused ? 1 : 0, + task.baseBranch ?? null, + task.branch ?? null, + task.executionStartBranch ?? null, + task.baseCommitSha ?? null, + task.modelPresetId ?? null, + task.modelProvider ?? null, + task.modelId ?? null, + task.validatorModelProvider ?? null, + task.validatorModelId ?? null, + task.planningModelProvider ?? null, + task.planningModelId ?? null, + task.mergeRetries ?? null, + task.workflowStepRetries ?? null, + task.stuckKillCount ?? 0, + task.postReviewFixCount ?? 0, + task.recoveryRetryCount ?? null, + task.taskDoneRetryCount ?? 0, + task.verificationFailureCount ?? 0, + task.mergeConflictBounceCount ?? 0, + task.nextRecoveryAt ?? null, + task.error ?? null, + task.summary ?? null, + task.thinkingLevel ?? null, + task.executionMode ?? null, + task.tokenUsage?.inputTokens ?? null, + task.tokenUsage?.outputTokens ?? null, + task.tokenUsage?.cachedTokens ?? null, + task.tokenUsage?.totalTokens ?? null, + task.tokenUsage?.firstUsedAt ?? null, + task.tokenUsage?.lastUsedAt ?? null, + task.createdAt, + task.updatedAt, + task.columnMovedAt ?? null, + task.executionStartedAt ?? null, + task.executionCompletedAt ?? null, + toJson(task.dependencies || []), + toJson(task.steps || []), + toJson(task.log || []), + toJson(task.attachments || []), + toJson(task.steeringComments || []), + toJson(task.comments || []), + toJsonNullable(task.review), + toJsonNullable(task.reviewState), + toJson(task.workflowStepResults || []), + toJsonNullable(task.prInfo), + toJsonNullable(task.issueInfo), + toJsonNullable(task.githubTracking), + task.sourceIssue?.provider ?? null, + task.sourceIssue?.repository ?? null, + task.sourceIssue?.externalIssueId ?? null, + task.sourceIssue?.issueNumber ?? null, + task.sourceIssue?.url ?? null, + toJsonNullable(task.mergeDetails), + task.breakIntoSubtasks ? 1 : 0, + toJson(task.enabledWorkflowSteps || []), + toJson(task.modifiedFiles || []), + task.missionId ?? null, + task.sliceId ?? null, + task.assignedAgentId ?? null, + task.pausedByAgentId ?? null, + task.assigneeUserId ?? null, + task.nodeId ?? null, + task.effectiveNodeId ?? null, + task.effectiveNodeSource ?? null, + task.sourceType ?? null, + task.sourceAgentId ?? null, + task.sourceRunId ?? null, + task.sourceSessionId ?? null, + task.sourceMessageId ?? null, + task.sourceParentTaskId ?? null, + toJsonNullable(task.sourceMetadata), + task.checkedOutBy ?? null, + task.checkedOutAt ?? null, + task.checkoutNodeId ?? null, + task.checkoutRunId ?? null, + task.checkoutLeaseRenewedAt ?? null, + task.checkoutLeaseEpoch ?? 0, + ]; + } + /** - * Upsert a task to the database. Used by create and update operations. + * Insert a brand-new task row. Create paths must use this so SQLite raises on + * duplicate IDs instead of silently rewriting the existing row. + */ + private insertTask(task: Task): void { + this.db.prepare(` + INSERT INTO tasks ( + id, lineageId, title, description, priority, "column", status, size, reviewLevel, currentStep, + worktree, blockedBy, paused, baseBranch, branch, executionStartBranch, baseCommitSha, modelPresetId, modelProvider, + modelId, validatorModelProvider, validatorModelId, planningModelProvider, planningModelId, mergeRetries, + workflowStepRetries, stuckKillCount, postReviewFixCount, recoveryRetryCount, taskDoneRetryCount, verificationFailureCount, mergeConflictBounceCount, nextRecoveryAt, error, + summary, thinkingLevel, executionMode, tokenUsageInputTokens, tokenUsageOutputTokens, tokenUsageCachedTokens, + tokenUsageTotalTokens, tokenUsageFirstUsedAt, tokenUsageLastUsedAt, createdAt, updatedAt, columnMovedAt, + executionStartedAt, executionCompletedAt, + dependencies, steps, log, attachments, steeringComments, + comments, review, reviewState, workflowStepResults, prInfo, issueInfo, githubTracking, + sourceIssueProvider, sourceIssueRepository, sourceIssueExternalIssueId, sourceIssueNumber, sourceIssueUrl, + mergeDetails, breakIntoSubtasks, enabledWorkflowSteps, modifiedFiles, missionId, sliceId, assignedAgentId, pausedByAgentId, assigneeUserId, nodeId, effectiveNodeId, effectiveNodeSource, sourceType, sourceAgentId, sourceRunId, sourceSessionId, sourceMessageId, sourceParentTaskId, sourceMetadata, checkedOutBy, checkedOutAt, checkoutNodeId, checkoutRunId, checkoutLeaseRenewedAt, checkoutLeaseEpoch + ) VALUES ( + ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? + ) + `).run(...this.getTaskPersistValues(task)); + this.db.bumpLastModified(); + } + + /** + * Upsert a task to the database. Update paths intentionally retain ON CONFLICT + * semantics; create paths must use insertTask() instead. */ private upsertTask(task: Task): void { this.db.prepare(` @@ -1325,101 +1461,64 @@ export class TaskStore extends EventEmitter { checkoutRunId = excluded.checkoutRunId, checkoutLeaseRenewedAt = excluded.checkoutLeaseRenewedAt, checkoutLeaseEpoch = excluded.checkoutLeaseEpoch - `).run( - task.id, - task.lineageId ?? generateTaskLineageId(), - task.title ?? null, - task.description, - normalizeTaskPriority(task.priority), - task.column, - task.status ?? null, - task.size ?? null, - task.reviewLevel ?? null, - task.currentStep || 0, - task.worktree ?? null, - task.blockedBy ?? null, - task.paused ? 1 : 0, - task.baseBranch ?? null, - task.branch ?? null, - task.executionStartBranch ?? null, - task.baseCommitSha ?? null, - task.modelPresetId ?? null, - task.modelProvider ?? null, - task.modelId ?? null, - task.validatorModelProvider ?? null, - task.validatorModelId ?? null, - task.planningModelProvider ?? null, - task.planningModelId ?? null, - task.mergeRetries ?? null, - task.workflowStepRetries ?? null, - task.stuckKillCount ?? 0, - task.postReviewFixCount ?? 0, - task.recoveryRetryCount ?? null, - task.taskDoneRetryCount ?? 0, - task.verificationFailureCount ?? 0, - task.mergeConflictBounceCount ?? 0, - task.nextRecoveryAt ?? null, - task.error ?? null, - task.summary ?? null, - task.thinkingLevel ?? null, - task.executionMode ?? null, - task.tokenUsage?.inputTokens ?? null, - task.tokenUsage?.outputTokens ?? null, - task.tokenUsage?.cachedTokens ?? null, - task.tokenUsage?.totalTokens ?? null, - task.tokenUsage?.firstUsedAt ?? null, - task.tokenUsage?.lastUsedAt ?? null, - task.createdAt, - task.updatedAt, - task.columnMovedAt ?? null, - task.executionStartedAt ?? null, - task.executionCompletedAt ?? null, - toJson(task.dependencies || []), - toJson(task.steps || []), - toJson(task.log || []), - toJson(task.attachments || []), - toJson(task.steeringComments || []), - toJson(task.comments || []), - toJsonNullable(task.review), - toJsonNullable(task.reviewState), - toJson(task.workflowStepResults || []), - toJsonNullable(task.prInfo), - toJsonNullable(task.issueInfo), - toJsonNullable(task.githubTracking), - task.sourceIssue?.provider ?? null, - task.sourceIssue?.repository ?? null, - task.sourceIssue?.externalIssueId ?? null, - task.sourceIssue?.issueNumber ?? null, - task.sourceIssue?.url ?? null, - toJsonNullable(task.mergeDetails), - task.breakIntoSubtasks ? 1 : 0, - toJson(task.enabledWorkflowSteps || []), - toJson(task.modifiedFiles || []), - task.missionId ?? null, - task.sliceId ?? null, - task.assignedAgentId ?? null, - task.pausedByAgentId ?? null, - task.assigneeUserId ?? null, - task.nodeId ?? null, - task.effectiveNodeId ?? null, - task.effectiveNodeSource ?? null, - task.sourceType ?? null, - task.sourceAgentId ?? null, - task.sourceRunId ?? null, - task.sourceSessionId ?? null, - task.sourceMessageId ?? null, - task.sourceParentTaskId ?? null, - toJsonNullable(task.sourceMetadata), - task.checkedOutBy ?? null, - task.checkedOutAt ?? null, - task.checkoutNodeId ?? null, - task.checkoutRunId ?? null, - task.checkoutLeaseRenewedAt ?? null, - task.checkoutLeaseEpoch ?? 0, - ); + `).run(...this.getTaskPersistValues(task)); this.db.bumpLastModified(); } + private isTaskIdConflictError(error: unknown): boolean { + const message = error instanceof Error ? error.message : String(error); + return /SQLITE_CONSTRAINT|UNIQUE constraint failed: tasks\.id|PRIMARY KEY constraint failed: tasks\.id/i.test(message); + } + + private logTaskCreateConflict(task: Task, operation: string, error: unknown): void { + storeLog.error("Refused colliding task create", { + phase: "task-create:id-conflict", + operation, + taskId: task.id, + column: task.column, + sourceType: task.sourceType, + error: error instanceof Error ? error.message : String(error), + }); + } + + private insertTaskWithFtsRecovery(task: Task, operation: string): void { + const normalizeConflict = (error: unknown): never => { + this.logTaskCreateConflict(task, operation, error); + throw new Error(`Task ID already exists: ${task.id}`); + }; + + try { + this.insertTask(task); + return; + } catch (error) { + if (this.isTaskIdConflictError(error)) { + normalizeConflict(error); + } + if (!this.db.isFts5CorruptionError(error)) { + throw error; + } + + console.warn(`[fusion:store] FTS5 corruption detected during insert for task ${task.id}; rebuilding index and retrying once`); + + try { + this.db.rebuildFts5Index(); + } catch (rebuildError) { + console.warn("[fusion:store] FTS5 rebuild failed; propagating original insert error", rebuildError); + throw error; + } + + try { + this.insertTask(task); + } catch (retryError) { + if (this.isTaskIdConflictError(retryError)) { + normalizeConflict(retryError); + } + console.warn("[fusion:store] Insert retry after FTS5 rebuild failed; propagating original insert error", retryError); + throw error; + } + } + } + private upsertTaskWithFtsRecovery(task: Task): void { try { this.upsertTask(task); @@ -1459,6 +1558,31 @@ export class TaskStore extends EventEmitter { return this.rowToTask(row); } + private isTaskIdPresentInArchivedTasksTable(id: string): boolean { + try { + const row = this.db.prepare("SELECT 1 as found FROM archivedTasks WHERE id = ? LIMIT 1").get(id) as { found?: number } | undefined; + return row?.found === 1; + } catch { + return false; + } + } + + private taskIdExistsAnywhere(id: string): boolean { + if (this.readTaskFromDb(id)) { + return true; + } + if (this.isTaskIdPresentInArchivedTasksTable(id)) { + return true; + } + return this.archiveDb.get(id) !== undefined; + } + + private assertTaskIdAvailable(id: string): void { + if (this.taskIdExistsAnywhere(id)) { + throw new Error(`Task ID already exists: ${id}`); + } + } + private isTaskArchived(id: string): boolean { const row = this.db.prepare('SELECT "column" FROM tasks WHERE id = ?').get(id) as { column: Column } | undefined; if (row) { @@ -1718,7 +1842,17 @@ export class TaskStore extends EventEmitter { } /** - * Write a task to SQLite (primary store) and also write task.json to disk + * Write a brand-new task to SQLite (primary store) and also write task.json to disk + * for backward compatibility and debugging. Create paths must call this variant + * so duplicate IDs fail safely instead of overwriting existing rows. + */ + private async atomicCreateTaskJson(dir: string, task: Task, operation: string): Promise { + this.insertTaskWithFtsRecovery(task, operation); + await this.writeTaskJsonFile(dir, task); + } + + /** + * Write an existing task to SQLite (primary store) and also write task.json to disk * for backward compatibility and debugging. */ private async atomicWriteTaskJson(dir: string, task: Task): Promise { @@ -2104,6 +2238,11 @@ export class TaskStore extends EventEmitter { }; } + private serializeConfigForDisk(config: BoardConfig): string { + const { nextId: _deprecatedNextId, ...configForDisk } = config as BoardConfig & { nextId?: number }; + return JSON.stringify(configForDisk, null, 2); + } + private async writeConfig( config: BoardConfig, options?: { nextWorkflowStepId?: number }, @@ -2119,12 +2258,18 @@ export class TaskStore extends EventEmitter { ? JSON.stringify(legacyWorkflowSteps) : "[]"; - // Use INSERT OR REPLACE to ensure the config row exists (handles edge case where row is missing) + // `config.nextId` is deprecated legacy state. Preserve the existing column + // value for one release, but stop writing new values so distributed_task_id_state + // remains the sole active allocator counter. this.db.prepare( - `INSERT OR REPLACE INTO config (id, nextId, nextWorkflowStepId, settings, workflowSteps, updatedAt) - VALUES (1, ?, ?, ?, ?, ?)`, + `INSERT INTO config (id, nextWorkflowStepId, settings, workflowSteps, updatedAt) + VALUES (1, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + nextWorkflowStepId = excluded.nextWorkflowStepId, + settings = excluded.settings, + workflowSteps = excluded.workflowSteps, + updatedAt = excluded.updatedAt`, ).run( - config.nextId || 1, nextWorkflowStepId, JSON.stringify(config.settings || {}), workflowStepsJson, @@ -2134,7 +2279,7 @@ export class TaskStore extends EventEmitter { // Also write config.json to disk for backward compatibility try { const tmpPath = this.configPath + ".tmp"; - await writeFile(tmpPath, JSON.stringify(config, null, 2)); + await writeFile(tmpPath, this.serializeConfigForDisk(config)); await rename(tmpPath, this.configPath); } catch (err) { // Best-effort: SQLite is the primary store @@ -2469,9 +2614,7 @@ export class TaskStore extends EventEmitter { throw new Error(`Task ${id} cannot depend on itself`); } - if (this.readTaskFromDb(id)) { - throw new Error(`Task ID already exists: ${id}`); - } + this.assertTaskIdAvailable(id); const title = input.title?.trim() || undefined; let resolvedWorkflowSteps: string[] | undefined = input.enabledWorkflowSteps?.length @@ -2589,9 +2732,10 @@ export class TaskStore extends EventEmitter { updatedAt: options?.updatedAt ?? now, }; + this.assertTaskIdAvailable(id); + const dir = this.taskDir(id); - await mkdir(dir, { recursive: true }); - await this.atomicWriteTaskJson(dir, task); + await this.atomicCreateTaskJson(dir, task, "createTask"); // Update cache if watcher is active if (this.isWatching) this.taskCache.set(id, { ...task }); @@ -2638,9 +2782,10 @@ export class TaskStore extends EventEmitter { baseBranch: sourceTask.baseBranch, }; + this.assertTaskIdAvailable(newId); + const newDir = this.taskDir(newId); - await mkdir(newDir, { recursive: true }); - await this.atomicWriteTaskJson(newDir, newTask); + await this.atomicCreateTaskJson(newDir, newTask, "duplicateTask"); await mkdir(newDir, { recursive: true }); await writeFile(join(newDir, "PROMPT.md"), sourceTask.prompt); @@ -2702,9 +2847,10 @@ export class TaskStore extends EventEmitter { attachments: sourceTask.attachments ? [...sourceTask.attachments] : undefined, }; + this.assertTaskIdAvailable(newId); + const newDir = this.taskDir(newId); - await mkdir(newDir, { recursive: true }); - await this.atomicWriteTaskJson(newDir, newTask); + await this.atomicCreateTaskJson(newDir, newTask, "refineTask"); const prompt = `# ${newTask.title}\n\n${newTask.description}\n`; await mkdir(newDir, { recursive: true }); await writeFile(join(newDir, "PROMPT.md"), prompt); @@ -6943,6 +7089,7 @@ ${stepsSection}`; if (this._db) { this._db.close(); this._db = null; + this.taskIdStateReconciled = false; } if (this._archiveDb) { this._archiveDb.close();