root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(FN-5242): add merge queue storage and lease API to task store

Browse Source 
- docs(FN-5242): complete Step 7 — document merge queue storage
- docs(FN-5242): complete Step 5 — add changeset and exports
- fix(FN-5242): order merge queue audit assertions chronologically
- test(FN-5242): complete Step 4 — cover merge queue leasing
- feat(FN-5242): complete Step 2 — add merge queue lease API
- feat(FN-5242): complete Step 1 — add mergeQueue schema v89

Fusion-Task-Id: FN-5242
This commit is contained in:
Fusion (runfusion.ai)
2026-05-19 21:14:49 -07:00
committed by gsxdsm
parent 17eb85ed81
commit 6b24b56b81
15 changed files with 636 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
packages/core/src/__tests__/db-migrate.test.ts
View File

@@ -715,7 +715,7 @@ describe("schema migration", () => {
const row = db.prepare("SELECT deletedAt FROM tasks WHERE id = 'FN-legacy'").get() as { deletedAt: string | null };
expect(row.deletedAt).toBeNull();
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
db.close();
});
@@ -748,7 +748,7 @@ describe("schema migration", () => {
{ id: "WS-001", mode: "prompt", gateMode: "advisory" },
{ id: "WS-002", mode: "script", gateMode: "advisory" },
]);
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
db.close();
});
@@ -798,7 +798,7 @@ describe("schema migration", () => {
reviewerContextRetryCount: 0,
reviewerFallbackRetryCount: 0,
});
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
db.close();
});
@@ -827,7 +827,7 @@ describe("schema migration", () => {
const columns = db.prepare("PRAGMA table_info(milestones)").all() as Array<{ name: string }>;
expect(columns.map((column) => column.name)).toContain("acceptanceCriteria");
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
db.close();
});
@@ -862,7 +862,7 @@ describe("schema migration", () => {
{ id: "WS-002", mode: "script", enabled: 1, gateMode: "advisory" },
{ id: "WS-003", mode: "prompt", enabled: 0, gateMode: "advisory" },
]);
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
db.close();
});

34
packages/core/src/__tests__/db.test.ts
View File

@@ -296,7 +296,7 @@ describe("Database", () => {
});
it("seeds schema version", () => {
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
});
it("includes tokenUsageCacheWriteTokens on freshly initialized tasks table", () => {
@@ -339,7 +339,7 @@ describe("Database", () => {
it("is idempotent - calling init() twice does not fail", () => {
expect(() => db.init()).not.toThrow();
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
});
it("does not overwrite existing config on re-init", () => {
// Update the config
@@ -1404,7 +1404,7 @@ describe("schema migrations", () => {
db.init();
// Verify version bumped to 29 (includes v1→v2 through v26→v29)
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
// Verify new columns exist and existing data is intact
const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
@@ -1429,11 +1429,11 @@ describe("schema migrations", () => {
const db = new Database(fusionDir);
db.init();
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
// Re-init should not fail
db.init();
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
db.close();
});
@@ -1468,7 +1468,7 @@ describe("schema migrations", () => {
db.init();
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
expect(cols.map((col) => col.name)).toContain("priority");
@@ -1509,7 +1509,7 @@ describe("schema migrations", () => {
db.init();
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
const colNames = cols.map((col) => col.name);
@@ -1581,7 +1581,7 @@ describe("schema migrations", () => {
db.init();
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
const colNames = cols.map((col) => col.name);
@@ -1821,7 +1821,7 @@ describe("schema migrations", () => {
db.init();
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
const cols = db.prepare("PRAGMA table_info(chat_messages)").all() as Array<{ name: string }>;
expect(cols.map((col) => col.name)).toContain("attachments");
@@ -1895,7 +1895,7 @@ describe("schema migrations", () => {
db.init();
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'agentRatings'").all() as Array<{ name: string }>;
expect(tables).toEqual([{ name: "agentRatings" }]);
@@ -1919,7 +1919,7 @@ describe("schema migrations", () => {
db.init();
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'mission_events'").all() as Array<{ name: string }>;
expect(tables).toEqual([{ name: "mission_events" }]);
@@ -2023,7 +2023,7 @@ describe("schema migrations", () => {
db.init();
// Verify version bumped to 29
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
// Verify new columns exist and existing data is intact
const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
@@ -2242,7 +2242,7 @@ describe("schema migrations", () => {
localDb.init();
expect(localDb.getSchemaVersion()).toBe(88);
expect(localDb.getSchemaVersion()).toBe(89);
const columns = localDb.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
expect(columns.map((column) => column.name)).toContain("tokenUsageCacheWriteTokens");
@@ -2553,7 +2553,7 @@ describe("createDatabase factory", () => {
const db = createDatabase(fusionDir);
db.init();
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
expect(db.getLastModified()).toBeGreaterThan(0);
db.close();
@@ -2707,7 +2707,7 @@ describe("migration v77 task token budget columns", () => {
migrated = new Database(fusion);
migrated.init();
expect(migrated.getSchemaVersion()).toBe(88);
expect(migrated.getSchemaVersion()).toBe(89);
const rows = migrated.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
const names = new Set(rows.map((row) => row.name));
expect(names.has("tokenBudgetSoftAlertedAt")).toBe(true);
@@ -2753,7 +2753,7 @@ describe("migration v67 drops orphan project auth tables", () => {
migrated = new Database(fusion);
migrated.init();
expect(migrated.getSchemaVersion()).toBe(88);
expect(migrated.getSchemaVersion()).toBe(89);
const tables = migrated
.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'project_auth_%'")
.all() as Array<{ name: string }>;
@@ -2780,7 +2780,7 @@ describe("migration v67 drops orphan project auth tables", () => {
try {
fresh.init();
expect(fresh.getSchemaVersion()).toBe(88);
expect(fresh.getSchemaVersion()).toBe(89);
const tables = fresh
.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'project_auth_%'")
.all() as Array<{ name: string }>;

10
packages/core/src/__tests__/insight-store.test.ts
View File

@@ -1000,7 +1000,7 @@ describe("Migration: pre-33 DB upgrade", () => {
// Step 1: Create a fresh database at v33 (runs all migrations up to 33)
const db1 = createDatabase(legacyDir);
db1.init();
expect(db1.getSchemaVersion()).toBe(88);
expect(db1.getSchemaVersion()).toBe(89);
db1.close();
// Step 2: Manually downgrade to version 32 and drop insight tables
@@ -1035,7 +1035,7 @@ describe("Migration: pre-33 DB upgrade", () => {
expect(tableNamesBefore).not.toContain("project_insight_runs");
// Now run init — this triggers the v32→v33 migration
db3.init();
expect(db3.getSchemaVersion()).toBe(88);
expect(db3.getSchemaVersion()).toBe(89);
// Step 4: Verify insight tables exist after migration
const tablesAfter = db3.prepare(
@@ -1066,12 +1066,12 @@ describe("Migration: pre-33 DB upgrade", () => {
try {
const db1 = createDatabase(testDir);
db1.init();
expect(db1.getSchemaVersion()).toBe(88);
expect(db1.getSchemaVersion()).toBe(89);
db1.close();
const db2 = createDatabase(testDir);
expect(() => db2.init()).not.toThrow();
expect(db2.getSchemaVersion()).toBe(88);
expect(db2.getSchemaVersion()).toBe(89);
db2.close();
} finally {
rmSync(testDir, { recursive: true, force: true });
@@ -1085,7 +1085,7 @@ describe("Migration: pre-33 DB upgrade", () => {
// Step 1: Create a fresh DB and run migrations
const db1 = createDatabase(compatDir);
db1.init();
expect(db1.getSchemaVersion()).toBe(88);
expect(db1.getSchemaVersion()).toBe(89);
// Step 2: Strip lifecycle and cancelledAt columns by recreating the
// table without them. This simulates a DB that was created before the

2
packages/core/src/__tests__/mission-store.test.ts
View File

@@ -2821,7 +2821,7 @@ describe("MissionStore", () => {
describe("Loop State & Validator Run Schema (v31)", () => {
it("schema version is 40 after migration", () => {
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
});
it("mission_features table has loop state columns", () => {

2
packages/core/src/__tests__/run-audit.test.ts
View File

@@ -584,7 +584,7 @@ describe("Run Audit", () => {
});
it("schema version is bumped to 40", () => {
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
});
});
});

6
packages/core/src/__tests__/secrets-schema.test.ts
View File

@@ -42,7 +42,7 @@ describe("secrets schema migrations", () => {
const version = db
.prepare("SELECT value FROM __meta WHERE key = 'schemaVersion'")
.get() as { value: string };
expect(version.value).toBe("88");
expect(version.value).toBe("89");
} finally {
db.close();
rmSync(dir, { recursive: true, force: true });
@@ -105,7 +105,7 @@ describe("secrets schema migrations", () => {
const version = db
.prepare("SELECT value FROM __meta WHERE key = 'schemaVersion'")
.get() as { value: string };
expect(version.value).toBe("88");
expect(version.value).toBe("89");
} finally {
db.close();
rmSync(dir, { recursive: true, force: true });
@@ -155,7 +155,7 @@ describe("secrets schema migrations", () => {
.prepare("SELECT value FROM __meta WHERE key = 'schemaVersion'")
.get() as { value: string };
expect(projectVersion.value).toBe("88");
expect(projectVersion.value).toBe("89");
expect(centralVersion.value).toBe("13");
} finally {
projectDb.close();

258
packages/core/src/__tests__/store-merge-queue.test.ts Normal file
View File

@@ -0,0 +1,258 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { mkdtempSync } from "node:fs";
import { rm } from "node:fs/promises";
import { join } from "node:path";
import { tmpdir } from "node:os";
import { TaskStore, MergeQueueLeaseOwnershipError, MergeQueueTaskNotFoundError } from "../store.js";
function makeTmpDir(): string {
return mkdtempSync(join(tmpdir(), "kb-merge-queue-test-"));
}
describe("TaskStore merge queue", () => {
let rootDir: string;
let globalDir: string;
let store: TaskStore;
const extraStores: TaskStore[] = [];
beforeEach(async () => {
rootDir = makeTmpDir();
globalDir = join(rootDir, ".fusion-global");
store = new TaskStore(rootDir, globalDir);
await store.init();
});
afterEach(async () => {
vi.useRealTimers();
for (const extraStore of extraStores.splice(0)) {
extraStore.close();
}
store.close();
await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 });
});
async function createTask(priority: "low" | "normal" | "high" | "urgent" = "normal"): Promise<string> {
const task = await store.createTask({ description: `merge queue ${priority}`, priority });
return task.id;
}
function getTableNames(): string[] {
return (store.getDatabase().prepare("SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name").all() as Array<{ name: string }>).map((row) => row.name);
}
it("creates the mergeQueue table and indexes on fresh init", () => {
expect(getTableNames()).toContain("mergeQueue");
const indexes = store.getDatabase().prepare("PRAGMA index_list('mergeQueue')").all() as Array<{ name: string }>;
expect(indexes.map((row) => row.name)).toEqual(
expect.arrayContaining(["idx_mergeQueue_lease_ready", "idx_mergeQueue_leaseExpiresAt"]),
);
expect(store.getDatabase().getSchemaVersion()).toBe(89);
});
it("migrates a legacy v88 database and preserves task rows", async () => {
const existingTask = await store.createTask({ description: "legacy row survives", priority: "high" });
const db = store.getDatabase();
db.exec("DROP INDEX IF EXISTS idx_mergeQueue_lease_ready");
db.exec("DROP INDEX IF EXISTS idx_mergeQueue_leaseExpiresAt");
db.exec("DROP TABLE IF EXISTS mergeQueue");
db.prepare("UPDATE __meta SET value = '88' WHERE key = 'schemaVersion'").run();
store.close();
const reopened = new TaskStore(rootDir, globalDir);
extraStores.push(reopened);
await reopened.init();
const tables = reopened.getDatabase().prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'mergeQueue'").all() as Array<{ name: string }>;
expect(tables).toEqual([{ name: "mergeQueue" }]);
expect((await reopened.getTask(existingTask.id))?.description).toBe("legacy row survives");
});
it("enqueueMergeQueue is idempotent and preserves existing attempt state", async () => {
const taskId = await createTask();
const first = store.enqueueMergeQueue(taskId, { now: "2026-05-19T00:00:00.000Z" });
const second = store.enqueueMergeQueue(taskId, { now: "2026-05-19T00:00:05.000Z" });
expect(first).toEqual(second);
expect(store.peekMergeQueue()).toHaveLength(1);
expect(store.peekMergeQueue()[0].attemptCount).toBe(0);
const events = store.getRunAuditEvents({ taskId, mutationType: "mergeQueue:enqueue" });
expect(events).toHaveLength(2);
expect(events[0].metadata).toMatchObject({ alreadyEnqueued: true, taskId, enqueuedAt: first.enqueuedAt, priority: "normal" });
expect(events[1].metadata).toMatchObject({ alreadyEnqueued: false, taskId, enqueuedAt: first.enqueuedAt, priority: "normal" });
});
it("throws MergeQueueTaskNotFoundError for unknown tasks", () => {
expect(() => store.enqueueMergeQueue("FN-999999")).toThrow(MergeQueueTaskNotFoundError);
});
it("leases in priority order regardless of enqueue order", async () => {
const lowTaskId = await createTask("low");
const urgentTaskId = await createTask("urgent");
const normalTaskId = await createTask("normal");
store.enqueueMergeQueue(lowTaskId, { now: "2026-05-19T00:00:00.000Z" });
store.enqueueMergeQueue(urgentTaskId, { now: "2026-05-19T00:00:01.000Z" });
store.enqueueMergeQueue(normalTaskId, { now: "2026-05-19T00:00:02.000Z" });
expect(store.acquireMergeQueueLease("worker-1", { leaseDurationMs: 60_000, now: "2026-05-19T00:01:00.000Z" })?.taskId).toBe(urgentTaskId);
expect(store.acquireMergeQueueLease("worker-2", { leaseDurationMs: 60_000, now: "2026-05-19T00:01:01.000Z" })?.taskId).toBe(normalTaskId);
expect(store.acquireMergeQueueLease("worker-3", { leaseDurationMs: 60_000, now: "2026-05-19T00:01:02.000Z" })?.taskId).toBe(lowTaskId);
});
it("uses FIFO ordering within the same priority", async () => {
const firstTaskId = await createTask();
const secondTaskId = await createTask();
store.enqueueMergeQueue(firstTaskId, { now: "2026-05-19T00:00:00.000Z" });
store.enqueueMergeQueue(secondTaskId, { now: "2026-05-19T00:00:00.005Z" });
expect(store.acquireMergeQueueLease("worker-1", { leaseDurationMs: 60_000, now: "2026-05-19T00:01:00.000Z" })?.taskId).toBe(firstTaskId);
expect(store.acquireMergeQueueLease("worker-2", { leaseDurationMs: 60_000, now: "2026-05-19T00:01:01.000Z" })?.taskId).toBe(secondTaskId);
});
it("allows exactly one worker to lease a single queued task across competing stores", async () => {
const storeA = new TaskStore(rootDir, globalDir);
const storeB = new TaskStore(rootDir, globalDir);
extraStores.push(storeA, storeB);
await storeA.init();
await storeB.init();
const taskId = await createTask();
for (let index = 0; index < 20; index += 1) {
store.enqueueMergeQueue(taskId, { now: `2026-05-19T00:00:${String(index).padStart(2, "0")}.000Z` });
const [leaseA, leaseB] = await Promise.all([
Promise.resolve().then(() => storeA.acquireMergeQueueLease("worker-a", { leaseDurationMs: 60_000, now: `2026-05-19T00:10:${String(index).padStart(2, "0")}.000Z` })),
Promise.resolve().then(() => storeB.acquireMergeQueueLease("worker-b", { leaseDurationMs: 60_000, now: `2026-05-19T00:10:${String(index).padStart(2, "0")}.000Z` })),
]);
expect([Boolean(leaseA), Boolean(leaseB)].filter(Boolean)).toHaveLength(1);
const leased = (leaseA ?? leaseB)!;
expect(leased.taskId).toBe(taskId);
store.releaseMergeQueueLease(taskId, leased.leasedBy!, { kind: "success" });
expect(store.peekMergeQueue()).toHaveLength(0);
}
});
it("recovers expired leases and makes the task leasable again", async () => {
vi.useFakeTimers();
vi.setSystemTime(new Date("2026-05-19T00:00:00.000Z"));
const taskId = await createTask();
store.enqueueMergeQueue(taskId);
const firstLease = store.acquireMergeQueueLease("worker-a", { leaseDurationMs: 50 });
expect(firstLease?.leasedBy).toBe("worker-a");
vi.setSystemTime(new Date("2026-05-19T00:00:01.000Z"));
const recovered = store.recoverExpiredMergeQueueLeases();
expect(recovered).toHaveLength(1);
expect(recovered[0]).toMatchObject({ taskId, leasedBy: null, leasedAt: null, leaseExpiresAt: null });
const expiredEvents = store.getRunAuditEvents({ taskId, mutationType: "mergeQueue:lease-expired" });
expect(expiredEvents).toHaveLength(1);
expect(expiredEvents[0].metadata).toMatchObject({
taskId,
previousLeasedBy: "worker-a",
previousLeaseExpiresAt: firstLease?.leaseExpiresAt,
recoveredAt: "2026-05-19T00:00:01.000Z",
});
const [workerBLease, workerASecondAttempt] = await Promise.all([
Promise.resolve().then(() => store.acquireMergeQueueLease("worker-b", { leaseDurationMs: 60_000 })),
Promise.resolve().then(() => store.acquireMergeQueueLease("worker-a", { leaseDurationMs: 60_000 })),
]);
expect(workerBLease?.taskId).toBe(taskId);
expect(workerASecondAttempt).toBeNull();
});
it("guards lease release by current owner", async () => {
const taskId = await createTask();
store.enqueueMergeQueue(taskId, { now: "2026-05-19T00:00:00.000Z" });
const lease = store.acquireMergeQueueLease("worker-a", { leaseDurationMs: 60_000, now: "2026-05-19T00:01:00.000Z" });
expect(lease?.taskId).toBe(taskId);
expect(() => store.releaseMergeQueueLease(taskId, "worker-b", { kind: "success" })).toThrow(MergeQueueLeaseOwnershipError);
expect(store.peekMergeQueue()[0]).toMatchObject({ taskId, leasedBy: "worker-a" });
});
it("releases failed work back to the queue and increments attemptCount", async () => {
const taskId = await createTask();
store.enqueueMergeQueue(taskId, { now: "2026-05-19T00:00:00.000Z" });
const lease = store.acquireMergeQueueLease("worker-a", { leaseDurationMs: 60_000, now: "2026-05-19T00:01:00.000Z" });
expect(lease?.taskId).toBe(taskId);
store.releaseMergeQueueLease(taskId, "worker-a", { kind: "failure", error: "boom" });
const queued = store.peekMergeQueue()[0];
expect(queued).toMatchObject({
taskId,
leasedBy: null,
leasedAt: null,
leaseExpiresAt: null,
attemptCount: 1,
lastError: "boom",
});
expect(store.acquireMergeQueueLease("worker-b", { leaseDurationMs: 60_000, now: "2026-05-19T00:02:00.000Z" })?.taskId).toBe(taskId);
});
it("emits one audit event for each merge queue mutation path", async () => {
vi.useFakeTimers();
vi.setSystemTime(new Date("2026-05-19T00:00:00.000Z"));
const failureTaskId = await createTask();
const expiryTaskId = await createTask("urgent");
store.enqueueMergeQueue(failureTaskId);
store.acquireMergeQueueLease("worker-a", { leaseDurationMs: 60_000 });
store.releaseMergeQueueLease(failureTaskId, "worker-a", { kind: "failure", error: "boom" });
store.enqueueMergeQueue(expiryTaskId);
store.acquireMergeQueueLease("worker-b", { leaseDurationMs: 10 });
vi.setSystemTime(new Date("2026-05-19T00:00:01.000Z"));
store.recoverExpiredMergeQueueLeases();
const auditRows = store.getDatabase().prepare(`
SELECT taskId, mutationType, target, metadata
FROM runAuditEvents
WHERE mutationType LIKE 'mergeQueue:%'
ORDER BY timestamp ASC, rowid ASC
`).all() as Array<{
taskId: string | null;
mutationType: string;
target: string;
metadata: string | null;
}>;
const auditEvents = auditRows.map((row) => ({
taskId: row.taskId,
mutationType: row.mutationType,
target: row.target,
metadata: row.metadata ? JSON.parse(row.metadata) as Record<string, unknown> : undefined,
}));
const enqueueEvents = auditEvents.filter((event) => event.mutationType === "mergeQueue:enqueue" && event.target === failureTaskId);
expect(enqueueEvents).toHaveLength(1);
expect(Object.keys(enqueueEvents[0].metadata ?? {}).sort()).toEqual(["alreadyEnqueued", "enqueuedAt", "priority", "taskId"]);
const acquiredEvents = auditEvents.filter(
(event) => event.mutationType === "mergeQueue:lease-acquired" && event.target === failureTaskId && event.metadata?.workerId === "worker-a",
);
expect(acquiredEvents).toHaveLength(1);
expect(Object.keys(acquiredEvents[0].metadata ?? {}).sort()).toEqual(["leaseExpiresAt", "priority", "taskId", "workerId"]);
const releasedEvents = auditEvents.filter(
(event) => event.mutationType === "mergeQueue:lease-released" && event.target === failureTaskId && event.metadata?.workerId === "worker-a",
);
expect(releasedEvents).toHaveLength(1);
expect(Object.keys(releasedEvents[0].metadata ?? {}).sort()).toEqual(["attemptCount", "error", "outcome", "taskId", "workerId"]);
const expiredEvents = auditEvents.filter(
(event) => event.mutationType === "mergeQueue:lease-expired" && (event.target === expiryTaskId || event.metadata?.taskId === expiryTaskId),
);
expect(expiredEvents).toHaveLength(1);
expect(Object.keys(expiredEvents[0].metadata ?? {}).sort()).toEqual(["previousLeaseExpiresAt", "previousLeasedBy", "recoveredAt", "taskId"]);
});
});

2
packages/core/src/__tests__/task-documents.test.ts
View File

@@ -51,7 +51,7 @@ describe("TaskStore task documents", () => {
expect(tableNames.has("task_documents")).toBe(true);
expect(tableNames.has("task_document_revisions")).toBe(true);
expect(db.getSchemaVersion()).toBe(88);
expect(db.getSchemaVersion()).toBe(89);
const index = db
.prepare(

40
packages/core/src/db.ts
View File

@@ -120,7 +120,7 @@ export function probeFts5(db: DatabaseSync): boolean {
// ── Schema Definition ────────────────────────────────────────────────
const SCHEMA_VERSION = 88;
const SCHEMA_VERSION = 89;
function normalizeTaskComments(
steeringComments: SteeringComment[] | undefined,
@@ -496,6 +496,19 @@ CREATE TABLE IF NOT EXISTS agentBlockedStates (
FOREIGN KEY (agentId) REFERENCES agents(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS mergeQueue (
taskId TEXT PRIMARY KEY REFERENCES tasks(id) ON DELETE CASCADE,
enqueuedAt TEXT NOT NULL,
priority TEXT NOT NULL DEFAULT 'normal',
leasedBy TEXT,
leasedAt TEXT,
leaseExpiresAt TEXT,
attemptCount INTEGER NOT NULL DEFAULT 0,
lastError TEXT
);
CREATE INDEX IF NOT EXISTS idx_mergeQueue_lease_ready ON mergeQueue(leasedBy, priority, enqueuedAt);
CREATE INDEX IF NOT EXISTS idx_mergeQueue_leaseExpiresAt ON mergeQueue(leaseExpiresAt);
-- Task documents (key-value store per task with revision tracking)
CREATE TABLE IF NOT EXISTS task_documents (
id TEXT PRIMARY KEY,
@@ -3465,6 +3478,31 @@ export class Database {
});
}
if (version < 89) {
this.applyMigration(89, () => {
this.db.exec(`
CREATE TABLE IF NOT EXISTS mergeQueue (
taskId TEXT PRIMARY KEY REFERENCES tasks(id) ON DELETE CASCADE,
enqueuedAt TEXT NOT NULL,
priority TEXT NOT NULL DEFAULT 'normal',
leasedBy TEXT,
leasedAt TEXT,
leaseExpiresAt TEXT,
attemptCount INTEGER NOT NULL DEFAULT 0,
lastError TEXT
)
`);
this.db.exec(`
CREATE INDEX IF NOT EXISTS idx_mergeQueue_lease_ready
ON mergeQueue(leasedBy, priority, enqueuedAt)
`);
this.db.exec(`
CREATE INDEX IF NOT EXISTS idx_mergeQueue_leaseExpiresAt
ON mergeQueue(leaseExpiresAt)
`);
});
}
}
/**

5
packages/core/src/index.ts
View File

@@ -1,5 +1,5 @@
export { COLUMNS, DEFAULT_COLUMN, isColumn, normalizeColumn, COLUMN_LABELS, COLUMN_DESCRIPTIONS, VALID_TRANSITIONS, DEFAULT_SETTINGS, DEFAULT_GLOBAL_SETTINGS, DEFAULT_PROJECT_SETTINGS, GLOBAL_SETTINGS_KEYS, PROJECT_SETTINGS_KEYS, isGlobalSettingsKey, isProjectSettingsKey, resolvePersistAgentThinkingLog, THINKING_LEVELS, THEME_MODES, COLOR_THEMES, WORKFLOW_STEP_TEMPLATES, AGENT_PERMISSIONS, PERMANENT_AGENT_ACTION_CATEGORIES, AGENT_PERMISSION_POLICY_ACTION_CATEGORIES, AGENT_PROVISIONING_APPROVAL_MODES, SANDBOX_PROVISIONING_APPROVAL_MODES, AGENT_PERMISSION_POLICY_PRESET_IDS, LEGACY_AGENT_PERMISSION_POLICY_ACTION_CATEGORY_ALIASES, APPROVAL_REQUEST_STATUSES, APPROVAL_REQUEST_AUDIT_EVENT_TYPES, normalizeApprovalRequestActionCategory, isValidApprovalRequestTransition, agentToConfigSnapshot, diffConfigSnapshots, isEphemeralAgent, hasAgentIdentity, CheckoutConflictError, DEFAULT_HEARTBEAT_PROCEDURE_PATH, getDefaultHeartbeatProcedurePath, EXECUTION_MODES, DEFAULT_EXECUTION_MODE, TASK_PRIORITIES, DEFAULT_TASK_PRIORITY, HIGH_FANOUT_BLOCKER_TODO_THRESHOLD, STALE_HIGH_FANOUT_BLOCKER_AGE_THRESHOLD_MS, DASHBOARD_USER_ID, normalizeMessageParticipant, validateMessageMetadata, validateDockerNodeConfig, sanitizeDockerNodeConfigForResponse, normalizeMergeConflictStrategy, normalizeMergeStrategyOverlapBehavior, normalizePostMergeAuditMode, POST_MERGE_AUDIT_MODES, normalizeMergeAuditAutoRecovery, MERGE_AUDIT_AUTO_RECOVERY_MODES, normalizeAutoRecovery, AUTO_RECOVERY_MODES, buildResearchDocumentKey, REPO_OVERRIDE_RE, SHARED_STATE_SNAPSHOT_VERSION } from "./types.js";
export type { Column, IssueInfo, IssueState, TaskSourceIssue, PrInfo, PrConflictState, PrConflictDiagnostics, PrCheckState, PrCheckStatus, PrStatus, Task, TaskTokenUsage, TaskAttachment, TaskComment, TaskCommentInput, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, TaskCreateInput, MeshReplicatedTaskCreatePayload, MeshReplicatedTaskApplyResult, TaskSource, SourceType, TaskDetail, RetrySummary, InboxTask, TodoList, TodoItem, TodoListCreateInput, TodoListUpdateInput, TodoItemCreateInput, TodoItemUpdateInput, TodoListWithItems, AgentLogEntry, AgentLogType, AgentRole, BoardConfig, DistributedTaskIdReserveInput, DistributedTaskIdReserveResult, DistributedTaskIdCommitInput, DistributedTaskIdCommitResult, DistributedTaskIdAbortInput, DistributedTaskIdAbortResult, DistributedTaskIdStateInput, DistributedTaskIdStateResult, AutostashOrphanRecord, AutostashOutcome, MergeDetails, MergeResult, MergeConflictStrategy, CanonicalMergeConflictStrategy, MergeStrategyOverlapBehavior, PostMergeAuditMode, MergeAuditAutoRecoveryMode, AutoRecoveryMode, AutoRecoveryFailureClass, AutoRecoverySettings, DirectMergeCommitStrategy, Settings, GlobalSettings, ProjectSettings, SecretsEnvConfig, WebSearchBackend, ResearchEnabledSources, ResearchGlobalDefaults, ResearchProjectLimits, ResearchProjectSettings, SandboxBackendName, SandboxFailureMode, SandboxPolicy, SandboxProjectSettings, EvalFollowUpPolicy, EvalProjectSettings, ResolvedEvalSettings, SettingsScope, DaemonTokenSettings, TaskStep, StepStatus, TaskLogEntry, RunMutationContext, ActivityLogEntry, ActivityEventType, ThinkingLevel, ThemeMode, ColorTheme, ExecutionMode, TaskPriority, UnavailableNodePolicy, OwningNodeHandoffPolicy, PlanningQuestion, PlanningSummary, PlanningResponse, PlanningQuestionType, ArchivedTaskEntry, BatchStatusRequest, BatchStatusResponse, BatchStatusEntry, BatchStatusResult, GithubIssueAction, ModelPreset, WorkflowStep, WorkflowStepMode, WorkflowStepGateMode, WorkflowStepPhase, WorkflowStepInput, WorkflowStepResult, WorkflowStepTemplate, Agent, OrgTreeNode, AgentState, AgentDetail, AgentCreateInput, AgentUpdateInput, AgentApiKey, AgentApiKeyCreateResult, AgentCapability, AgentPromptTemplate, AgentPromptsConfig, AgentPermission, PermanentAgentActionCategory, PermanentAgentSensitiveActionCategory, PermanentAgentGatingContext, AgentPermissionPolicy, AgentPermissionPolicyRules, AgentPermissionPolicyActionCategory, AgentProvisioningApprovalMode, SandboxProvisioningApprovalMode, LegacyAgentPermissionPolicyActionCategory, ApprovalRequestActionCategoryInput, ApprovalRequestActionCategory, AgentPermissionPolicyDisposition, AgentPermissionPolicyPresetId, ApprovalRequestStatus, ApprovalRequestAuditEventType, ApprovalRequestActorSnapshot, ApprovalRequestTargetAction, ApprovalRequestAuditEvent, ApprovalRequest, ApprovalRequestCreateInput, ApprovalRequestDecisionInput, ApprovalRequestCompletionInput, ApprovalRequestListInput, TaskAssignSource, AgentAccessState, AgentHeartbeatConfig, AgentBudgetConfig, AgentBudgetStatus, InstructionsBundleConfig, MessageResponseMode, AgentHeartbeatEvent, AgentHeartbeatRun, BlockedStateSnapshot, HeartbeatInvocationSource, AgentTaskSession, AgentRating, AgentRatingSummary, AgentRatingInput, AgentConfigSnapshot, RevisionFieldDiff, AgentConfigRevision, AgentStats, ReflectionTrigger, ReflectionMetrics, AgentReflection, AgentPerformanceSummary, NtfyNotificationEvent, NotificationEvent, NotificationPayload, NotificationProviderConfig, CustomProvider, SteeringComment, ParticipantType, MessageType, Message, MessageCreateInput, MessageFilter, MessageMetadata, MessageReplyReference, Mailbox, CheckoutLease, CheckoutClaimPrecondition, TaskClaimRow, CentralClaimStore, RunAuditDomain, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, AgentMemoryInclusionMode, HeartbeatPromptTemplate, HeartbeatScopeDisciplineMode, WorktrunkSettings, WorktrunkOnFailure } from "./types.js";
export type { Column, IssueInfo, IssueState, TaskSourceIssue, PrInfo, PrConflictState, PrConflictDiagnostics, PrCheckState, PrCheckStatus, PrStatus, Task, TaskTokenUsage, TaskAttachment, TaskComment, TaskCommentInput, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, TaskCreateInput, MeshReplicatedTaskCreatePayload, MeshReplicatedTaskApplyResult, TaskSource, SourceType, TaskDetail, RetrySummary, InboxTask, TodoList, TodoItem, TodoListCreateInput, TodoListUpdateInput, TodoItemCreateInput, TodoItemUpdateInput, TodoListWithItems, AgentLogEntry, AgentLogType, AgentRole, BoardConfig, DistributedTaskIdReserveInput, DistributedTaskIdReserveResult, DistributedTaskIdCommitInput, DistributedTaskIdCommitResult, DistributedTaskIdAbortInput, DistributedTaskIdAbortResult, DistributedTaskIdStateInput, DistributedTaskIdStateResult, AutostashOrphanRecord, AutostashOutcome, MergeDetails, MergeResult, MergeConflictStrategy, CanonicalMergeConflictStrategy, MergeStrategyOverlapBehavior, PostMergeAuditMode, MergeAuditAutoRecoveryMode, AutoRecoveryMode, AutoRecoveryFailureClass, AutoRecoverySettings, DirectMergeCommitStrategy, Settings, GlobalSettings, ProjectSettings, SecretsEnvConfig, WebSearchBackend, ResearchEnabledSources, ResearchGlobalDefaults, ResearchProjectLimits, ResearchProjectSettings, SandboxBackendName, SandboxFailureMode, SandboxPolicy, SandboxProjectSettings, EvalFollowUpPolicy, EvalProjectSettings, ResolvedEvalSettings, SettingsScope, DaemonTokenSettings, TaskStep, StepStatus, TaskLogEntry, RunMutationContext, ActivityLogEntry, ActivityEventType, ThinkingLevel, ThemeMode, ColorTheme, ExecutionMode, TaskPriority, MergeQueueEntry, MergeQueueEnqueueOptions, MergeQueueAcquireOptions, MergeQueueReleaseOutcome, UnavailableNodePolicy, OwningNodeHandoffPolicy, PlanningQuestion, PlanningSummary, PlanningResponse, PlanningQuestionType, ArchivedTaskEntry, BatchStatusRequest, BatchStatusResponse, BatchStatusEntry, BatchStatusResult, GithubIssueAction, ModelPreset, WorkflowStep, WorkflowStepMode, WorkflowStepGateMode, WorkflowStepPhase, WorkflowStepInput, WorkflowStepResult, WorkflowStepTemplate, Agent, OrgTreeNode, AgentState, AgentDetail, AgentCreateInput, AgentUpdateInput, AgentApiKey, AgentApiKeyCreateResult, AgentCapability, AgentPromptTemplate, AgentPromptsConfig, AgentPermission, PermanentAgentActionCategory, PermanentAgentSensitiveActionCategory, PermanentAgentGatingContext, AgentPermissionPolicy, AgentPermissionPolicyRules, AgentPermissionPolicyActionCategory, AgentProvisioningApprovalMode, SandboxProvisioningApprovalMode, LegacyAgentPermissionPolicyActionCategory, ApprovalRequestActionCategoryInput, ApprovalRequestActionCategory, AgentPermissionPolicyDisposition, AgentPermissionPolicyPresetId, ApprovalRequestStatus, ApprovalRequestAuditEventType, ApprovalRequestActorSnapshot, ApprovalRequestTargetAction, ApprovalRequestAuditEvent, ApprovalRequest, ApprovalRequestCreateInput, ApprovalRequestDecisionInput, ApprovalRequestCompletionInput, ApprovalRequestListInput, TaskAssignSource, AgentAccessState, AgentHeartbeatConfig, AgentBudgetConfig, AgentBudgetStatus, InstructionsBundleConfig, MessageResponseMode, AgentHeartbeatEvent, AgentHeartbeatRun, BlockedStateSnapshot, HeartbeatInvocationSource, AgentTaskSession, AgentRating, AgentRatingSummary, AgentRatingInput, AgentConfigSnapshot, RevisionFieldDiff, AgentConfigRevision, AgentStats, ReflectionTrigger, ReflectionMetrics, AgentReflection, AgentPerformanceSummary, NtfyNotificationEvent, NotificationEvent, NotificationPayload, NotificationProviderConfig, CustomProvider, SteeringComment, ParticipantType, MessageType, Message, MessageCreateInput, MessageFilter, MessageMetadata, MessageReplyReference, Mailbox, CheckoutLease, CheckoutClaimPrecondition, TaskClaimRow, CentralClaimStore, RunAuditDomain, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, AgentMemoryInclusionMode, HeartbeatPromptTemplate, HeartbeatScopeDisciplineMode, WorktrunkSettings, WorktrunkOnFailure } from "./types.js";
export { AGENT_VALID_TRANSITIONS, DUPLICATE_OF_METADATA_KEY } from "./types.js";
export {
resolveWorktrunkSettings,
@@ -130,6 +130,9 @@ export {
detectSelfDefeatingDependency,
SelfDefeatingDependencyError,
TaskDeletedError,
MergeQueueTaskNotFoundError,
MergeQueueLeaseOwnershipError,
InvalidMergeQueueLeaseDurationError,
} from "./store.js";
export {
STOPWORDS,

259
packages/core/src/store.ts
View File

@@ -3,7 +3,7 @@ import { randomUUID } from "node:crypto";
import { mkdir, readdir, readFile, writeFile, rename, unlink } from "node:fs/promises";
import { join } from "node:path";
import { existsSync, watch, type FSWatcher } from "node:fs";
import type { Task, TaskDetail, TaskCreateInput, TaskAttachment, AgentLogEntry, BoardConfig, Column, CheckoutClaimPrecondition, MergeResult, Settings, GlobalSettings, ProjectSettings, ActivityLogEntry, ActivityEventType, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, InboxTask, TaskLogEntry, RunMutationContext, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, ArchivedTaskEntry, ArchiveAgentLogMode, TaskPriority, SourceType, WorkflowStepTemplate, Agent, AutostashOrphanRecord, TaskCommitAssociation, TaskCommitAssociationMatchSource, TaskCommitAssociationConfidence, GithubIssueAction } from "./types.js";
import type { Task, TaskDetail, TaskCreateInput, TaskAttachment, AgentLogEntry, BoardConfig, Column, CheckoutClaimPrecondition, MergeResult, Settings, GlobalSettings, ProjectSettings, ActivityLogEntry, ActivityEventType, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, InboxTask, TaskLogEntry, RunMutationContext, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, ArchivedTaskEntry, ArchiveAgentLogMode, TaskPriority, SourceType, WorkflowStepTemplate, Agent, AutostashOrphanRecord, TaskCommitAssociation, TaskCommitAssociationMatchSource, TaskCommitAssociationConfidence, GithubIssueAction, MergeQueueEntry, MergeQueueEnqueueOptions, MergeQueueAcquireOptions, MergeQueueReleaseOutcome } from "./types.js";
import { createActivityLogSnapshot, createRunAuditSnapshot, createTaskMetadataSnapshot, toTaskMetadataRecord, validateSnapshotEnvelope, type ActivityLogSnapshot, type RunAuditSnapshot, type TaskMetadataSnapshot } from "./shared-mesh-state.js";
import { VALID_TRANSITIONS, DEFAULT_SETTINGS, isGlobalOnlySettingsKey, WORKFLOW_STEP_TEMPLATES, validateDocumentKey } from "./types.js";
import { DEFAULT_PROJECT_SETTINGS } from "./settings-schema.js";
@@ -262,6 +262,17 @@ interface RunAuditEventRow {
metadata: string | null;
}
interface MergeQueueRow {
taskId: string;
enqueuedAt: string;
priority: string;
leasedBy: string | null;
leasedAt: string | null;
leaseExpiresAt: string | null;
attemptCount: number;
lastError: string | null;
}
/** Database row shape for the config table. */
interface ConfigRow {
nextId: number;
@@ -794,6 +805,35 @@ export function detectSelfDefeatingDependency(
};
}
export class MergeQueueTaskNotFoundError extends Error {
constructor(public readonly taskId: string) {
super(`Cannot enqueue merge queue entry for missing task ${taskId}`);
this.name = "MergeQueueTaskNotFoundError";
}
}
export class MergeQueueLeaseOwnershipError extends Error {
constructor(
public readonly taskId: string,
public readonly workerId: string,
public readonly currentOwner: string | null,
) {
super(
currentOwner
? `Worker ${workerId} does not own merge queue lease for ${taskId}; current owner is ${currentOwner}`
: `Worker ${workerId} cannot release merge queue lease for ${taskId}; the entry is not currently leased`,
);
this.name = "MergeQueueLeaseOwnershipError";
}
}
export class InvalidMergeQueueLeaseDurationError extends Error {
constructor(public readonly leaseDurationMs: number) {
super(`merge queue leaseDurationMs must be > 0 (received ${leaseDurationMs})`);
this.name = "InvalidMergeQueueLeaseDurationError";
}
}
export class TaskStore extends EventEmitter<TaskStoreEvents> {
private static readonly ACTIVE_TASKS_WHERE = '"deletedAt" IS NULL';
@@ -5615,6 +5655,19 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
// ── Run Audit APIs ───────────────────────────────────────────────────
private rowToMergeQueueEntry(row: MergeQueueRow): MergeQueueEntry {
return {
taskId: row.taskId,
enqueuedAt: row.enqueuedAt,
priority: normalizeTaskPriority(row.priority),
leasedBy: row.leasedBy,
leasedAt: row.leasedAt,
leaseExpiresAt: row.leaseExpiresAt,
attemptCount: row.attemptCount,
lastError: row.lastError,
};
}
/**
* Convert a database row to a RunAuditEvent object.
*/
@@ -5750,6 +5803,210 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
return rows.map((row) => this.rowToRunAuditEvent(row));
}
enqueueMergeQueue(taskId: string, opts: MergeQueueEnqueueOptions = {}): MergeQueueEntry {
return this.db.transactionImmediate(() => {
const existing = this.db.prepare("SELECT * FROM mergeQueue WHERE taskId = ?").get(taskId) as MergeQueueRow | undefined;
const taskRow = this.db.prepare("SELECT priority FROM tasks WHERE id = ?").get(taskId) as { priority: string | null } | undefined;
if (!taskRow) {
throw new MergeQueueTaskNotFoundError(taskId);
}
const now = opts.now ?? new Date().toISOString();
const priority = opts.priority ?? normalizeTaskPriority(taskRow.priority);
let entry: MergeQueueEntry;
let alreadyEnqueued = true;
if (existing) {
entry = this.rowToMergeQueueEntry(existing);
} else {
this.db.prepare(`
INSERT INTO mergeQueue (taskId, enqueuedAt, priority, attemptCount)
VALUES (?, ?, ?, 0)
ON CONFLICT(taskId) DO NOTHING
`).run(taskId, now, priority);
const inserted = this.db.prepare("SELECT * FROM mergeQueue WHERE taskId = ?").get(taskId) as MergeQueueRow | undefined;
if (!inserted) {
throw new Error(`Failed to read merge queue entry for ${taskId} after enqueue`);
}
entry = this.rowToMergeQueueEntry(inserted);
alreadyEnqueued = false;
}
this.insertRunAuditEventRow({
taskId,
domain: "database",
mutationType: "mergeQueue:enqueue",
target: taskId,
metadata: {
taskId,
priority: entry.priority,
enqueuedAt: entry.enqueuedAt,
alreadyEnqueued,
},
});
return entry;
});
}
acquireMergeQueueLease(workerId: string, opts: MergeQueueAcquireOptions): MergeQueueEntry | null {
if (opts.leaseDurationMs <= 0) {
throw new InvalidMergeQueueLeaseDurationError(opts.leaseDurationMs);
}
return this.db.transactionImmediate(() => {
const now = opts.now ?? new Date().toISOString();
const leaseExpiresAt = new Date(Date.parse(now) + opts.leaseDurationMs).toISOString();
const leased = this.db.prepare(`
UPDATE mergeQueue
SET leasedBy = ?, leasedAt = ?, leaseExpiresAt = ?
WHERE taskId = (
SELECT taskId FROM mergeQueue
WHERE leasedBy IS NULL OR leaseExpiresAt <= ?
ORDER BY CASE priority
WHEN 'urgent' THEN 0
WHEN 'high' THEN 1
WHEN 'normal' THEN 2
WHEN 'low' THEN 3
ELSE 4
END ASC,
enqueuedAt ASC
LIMIT 1
)
RETURNING *
`).get(workerId, now, leaseExpiresAt, now) as MergeQueueRow | undefined;
if (!leased) {
return null;
}
const entry = this.rowToMergeQueueEntry(leased);
this.insertRunAuditEventRow({
taskId: entry.taskId,
domain: "database",
mutationType: "mergeQueue:lease-acquired",
target: entry.taskId,
metadata: {
taskId: entry.taskId,
workerId,
leaseExpiresAt: entry.leaseExpiresAt,
priority: entry.priority,
},
});
return entry;
});
}
releaseMergeQueueLease(taskId: string, workerId: string, outcome: MergeQueueReleaseOutcome): void {
this.db.transactionImmediate(() => {
const current = this.db.prepare("SELECT leasedBy FROM mergeQueue WHERE taskId = ?").get(taskId) as { leasedBy: string | null } | undefined;
if (!current || current.leasedBy !== workerId) {
throw new MergeQueueLeaseOwnershipError(taskId, workerId, current?.leasedBy ?? null);
}
if (outcome.kind === "success") {
this.db.prepare("DELETE FROM mergeQueue WHERE taskId = ? AND leasedBy = ?").run(taskId, workerId);
this.insertRunAuditEventRow({
taskId,
domain: "database",
mutationType: "mergeQueue:lease-released",
target: taskId,
metadata: {
taskId,
workerId,
outcome: "success",
},
});
return;
}
const released = this.db.prepare(`
UPDATE mergeQueue
SET leasedBy = NULL,
leasedAt = NULL,
leaseExpiresAt = NULL,
attemptCount = attemptCount + 1,
lastError = ?
WHERE taskId = ? AND leasedBy = ?
RETURNING *
`).get(outcome.error, taskId, workerId) as MergeQueueRow | undefined;
if (!released) {
throw new MergeQueueLeaseOwnershipError(taskId, workerId, null);
}
const entry = this.rowToMergeQueueEntry(released);
this.insertRunAuditEventRow({
taskId,
domain: "database",
mutationType: "mergeQueue:lease-released",
target: taskId,
metadata: {
taskId,
workerId,
outcome: "failure",
attemptCount: entry.attemptCount,
error: outcome.error,
},
});
});
}
recoverExpiredMergeQueueLeases(now: string = new Date().toISOString()): MergeQueueEntry[] {
return this.db.transactionImmediate(() => {
const expired = this.db.prepare(`
SELECT * FROM mergeQueue
WHERE leasedBy IS NOT NULL AND leaseExpiresAt <= ?
ORDER BY leaseExpiresAt ASC, enqueuedAt ASC
`).all(now) as MergeQueueRow[];
if (expired.length === 0) {
return [];
}
const recoveredRows = this.db.prepare(`
UPDATE mergeQueue
SET leasedBy = NULL,
leasedAt = NULL,
leaseExpiresAt = NULL
WHERE leasedBy IS NOT NULL AND leaseExpiresAt <= ?
RETURNING *
`).all(now) as MergeQueueRow[];
const previousByTaskId = new Map(expired.map((row) => [row.taskId, row]));
for (const row of recoveredRows) {
const previous = previousByTaskId.get(row.taskId);
this.insertRunAuditEventRow({
taskId: row.taskId,
domain: "database",
mutationType: "mergeQueue:lease-expired",
target: row.taskId,
metadata: {
taskId: row.taskId,
previousLeasedBy: previous?.leasedBy ?? null,
previousLeaseExpiresAt: previous?.leaseExpiresAt ?? null,
recoveredAt: now,
},
});
}
return recoveredRows.map((row) => this.rowToMergeQueueEntry(row));
});
}
peekMergeQueue(): MergeQueueEntry[] {
const rows = this.db.prepare(`
SELECT * FROM mergeQueue
ORDER BY CASE priority
WHEN 'urgent' THEN 0
WHEN 'high' THEN 1
WHEN 'normal' THEN 2
WHEN 'low' THEN 3
ELSE 4
END ASC,
enqueuedAt ASC
`).all() as MergeQueueRow[];
return rows.map((row) => this.rowToMergeQueueEntry(row));
}
// ── End Run Audit APIs ───────────────────────────────────────────────
/**

38
packages/core/src/types.ts
View File

@@ -38,6 +38,31 @@ export type TaskPriority = (typeof TASK_PRIORITIES)[number];
*/
export const DEFAULT_TASK_PRIORITY: TaskPriority = "normal";
export interface MergeQueueEntry {
taskId: string;
enqueuedAt: string;
priority: TaskPriority;
leasedBy: string | null;
leasedAt: string | null;
leaseExpiresAt: string | null;
attemptCount: number;
lastError: string | null;
}
export interface MergeQueueEnqueueOptions {
priority?: TaskPriority;
now?: string;
}
export interface MergeQueueAcquireOptions {
leaseDurationMs: number;
now?: string;
}
export type MergeQueueReleaseOutcome =
| { kind: "success" }
| { kind: "failure"; error: string };
/**
* Dashboard high-fan-out blocker threshold. A blocker is considered high impact
* when at least this many active todo tasks are waiting on it.
@@ -4944,6 +4969,13 @@ export interface AgentPromptsConfig {
* - "sandbox": Sandbox backend lifecycle events for user-configured command execution */
export type RunAuditDomain = "database" | "git" | "filesystem" | "sandbox";
export type RunAuditMutationType =
| "mergeQueue:enqueue"
| "mergeQueue:lease-acquired"
| "mergeQueue:lease-released"
| "mergeQueue:lease-expired"
| (string & {});
/** Input for recording a run-audit event. */
export interface RunAuditEventInput {
/** ISO-8601 timestamp when the event occurred. Defaults to current time if not provided. */
@@ -4957,7 +4989,7 @@ export interface RunAuditEventInput {
/** The domain/category of the mutation. */
domain: RunAuditDomain;
/** Type of mutation (e.g., "task:update", "git:commit", "file:write"). */
mutationType: string;
mutationType: RunAuditMutationType;
/** Target of the mutation (e.g., task ID, file path, branch name). */
target: string;
/** Optional structured metadata about the mutation (compact, actionable data). */
@@ -4979,7 +5011,7 @@ export interface RunAuditEvent {
/** The domain/category of the mutation */
domain: RunAuditDomain;
/** Type of mutation (e.g., "task:update", "git:commit", "file:write") */
mutationType: string;
mutationType: RunAuditMutationType;
/** Target of the mutation (e.g., task ID, file path, branch name) */
target: string;
/** Optional structured metadata about the mutation */
@@ -4997,7 +5029,7 @@ export interface RunAuditEventFilter {
/** Filter by domain. */
domain?: RunAuditDomain;
/** Filter by mutation type. */
mutationType?: string;
mutationType?: RunAuditMutationType;
/** Start of time range (inclusive). */
startTime?: string;
/** End of time range (inclusive). */