feat(FN-4044): harden task ID collision guards and allocation safety
Hardens task ID collision guards across distributed instances by adding `isAllocated` checks in `store.ts` and rewriting `distributed-task-id.ts` with deterministic collision-safe allocation logic, backed by a new `store-create-collision.test.ts` suite reproducing the race path. Updates architecture Fusion-Task-Id: FN-4044
This commit is contained in:
Fusion
5
.changeset/harden-task-id-overwrite-guards.md
Normal file
5
.changeset/harden-task-id-overwrite-guards.md
Normal file
@@ -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.
|
||||
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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/<id>/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:
|
||||
|
||||
@@ -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" });
|
||||
|
||||
117
packages/core/src/__tests__/store-create-collision.test.ts
Normal file
117
packages/core/src/__tests__/store-create-collision.test.ts
Normal file
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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");
|
||||
});
|
||||
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<string> {
|
||||
const prefixes = new Set<string>();
|
||||
const configured = getConfiguredPrefixAndLegacyNextId(db).prefix;
|
||||
if (configured) {
|
||||
prefixes.add(configured);
|
||||
}
|
||||
|
||||
const addFromQuery = (sql: string, mapper: (row: Record<string, unknown>) => string | undefined): void => {
|
||||
try {
|
||||
const rows = db.prepare(sql).all() as Array<Record<string, unknown>>;
|
||||
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
|
||||
|
||||
@@ -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<TaskStoreEvents> {
|
||||
private worktreeAllocationLock: Promise<void> = Promise.resolve();
|
||||
/** Promise chain for serializing config.json read-modify-write cycles */
|
||||
private configLock: Promise<void> = 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<TaskStoreEvents> {
|
||||
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<TaskStoreEvents> {
|
||||
return this._archiveDb;
|
||||
}
|
||||
|
||||
private reconcileDistributedTaskIdStateOnOpen(): void {
|
||||
if (this.taskIdStateReconciled) {
|
||||
return;
|
||||
}
|
||||
reconcileTaskIdState(this.db);
|
||||
this.taskIdStateReconciled = true;
|
||||
}
|
||||
|
||||
async init(): Promise<void> {
|
||||
await mkdir(this.tasksDir, { recursive: true });
|
||||
|
||||
@@ -719,6 +730,8 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
|
||||
}
|
||||
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<TaskStoreEvents> {
|
||||
}
|
||||
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<TaskStoreEvents> {
|
||||
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<TaskStoreEvents> {
|
||||
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<TaskStoreEvents> {
|
||||
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<TaskStoreEvents> {
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<void> {
|
||||
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<void> {
|
||||
@@ -2104,6 +2238,11 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
|
||||
};
|
||||
}
|
||||
|
||||
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<TaskStoreEvents> {
|
||||
? 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<TaskStoreEvents> {
|
||||
// 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<TaskStoreEvents> {
|
||||
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<TaskStoreEvents> {
|
||||
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<TaskStoreEvents> {
|
||||
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<TaskStoreEvents> {
|
||||
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();
|
||||
|
||||
Reference in New Issue
Block a user