diff --git a/docs/architecture.md b/docs/architecture.md index 32749129f..48b6ae8f5 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -471,6 +471,12 @@ Implemented in `agent-heartbeat.ts`: - Canonical replication/write-coordination contract: [`docs/shared-mesh-protocol.md`](./shared-mesh-protocol.md) - Defines protocol versioning, write classes, quorum/ack semantics, lease epochs/fencing, offline queue/replay, reconciliation outcomes, restart recovery hooks, and degraded-read staleness metadata. - Existing `/api/mesh/sync` and settings-sync payloads remain the active exchange primitives while follow-on runtime tasks implement full v1 coordinator/quorum behavior. +- Distributed task-ID allocation (`packages/core/src/distributed-task-id.ts`) is the first mesh-aware coordinated write primitive. + - Durable state lives in SQLite tables `distributed_task_id_state` (prefix sequence + authoritative committed count) and `distributed_task_id_reservations` (reservation lifecycle rows). + - 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. + - 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. - Process lifecycle ownership: - `fn serve` / `fn dashboard` start a single process-level `PeerExchangeService` and stop it during shutdown. - `CentralCore.startDiscovery()` is invoked from CLI startup only after HTTP bind completes so discovery advertises the actual listening port. diff --git a/docs/multi-project.md b/docs/multi-project.md index 43361c162..cd96e0c6b 100644 --- a/docs/multi-project.md +++ b/docs/multi-project.md @@ -35,6 +35,7 @@ Peer/mesh coordination spans core + engine, with startup ownership in CLI proces - `NodeDiscovery` and `NodeConnection` in `@fusion/core` handle discovery and remote node connectivity/auth primitives. - `PeerExchangeService` in `@fusion/engine` coordinates node-to-node sync/exchange workflows. - Canonical replication semantics live in [`docs/shared-mesh-protocol.md`](./shared-mesh-protocol.md). That protocol separates strongly coordinated shared state from append-only streams, queued replay classes, and node-local runtime state. +- Distributed task-ID allocation is one strongly coordinated shared-state path: reserve/commit/abort are coordinator-mediated writes, and cluster-wide committed task totals come from allocator `committedClusterTaskCount` state (not per-node local task counts). - `runServe()` and `runDashboard()` (CLI) own process-level mesh service lifecycle: - start one process-wide `PeerExchangeService` instance - call `CentralCore.startDiscovery()` only after the HTTP server is listening and the real bound port is known diff --git a/packages/core/src/__tests__/db.test.ts b/packages/core/src/__tests__/db.test.ts index 070ede951..b0764a21e 100644 --- a/packages/core/src/__tests__/db.test.ts +++ b/packages/core/src/__tests__/db.test.ts @@ -116,6 +116,8 @@ describe("Database", () => { expect(tableNames).toContain("project_auth_memberships"); expect(tableNames).toContain("project_auth_providers"); expect(tableNames).toContain("project_auth_sessions"); + expect(tableNames).toContain("distributed_task_id_state"); + expect(tableNames).toContain("distributed_task_id_reservations"); }); it("creates all expected indexes", () => { @@ -127,6 +129,8 @@ describe("Database", () => { expect(indexNames).toContain("idxActivityLogTimestamp"); expect(indexNames).toContain("idxActivityLogType"); expect(indexNames).toContain("idxActivityLogTaskId"); + expect(indexNames).toContain("idxDistributedTaskIdReservationsPrefixStatus"); + expect(indexNames).toContain("idxDistributedTaskIdReservationsExpiry"); expect(indexNames).toContain("idxActivityLogTaskIdTimestamp"); expect(indexNames).toContain("idxActivityLogTypeTimestamp"); expect(indexNames).toContain("idxArchivedTasksId"); @@ -171,7 +175,7 @@ describe("Database", () => { }); it("seeds schema version", () => { - expect(db.getSchemaVersion()).toBe(64); + expect(db.getSchemaVersion()).toBe(65); }); it("seeds lastModified", () => { const ts = db.getLastModified(); @@ -193,7 +197,7 @@ describe("Database", () => { it("is idempotent - calling init() twice does not fail", () => { expect(() => db.init()).not.toThrow(); - expect(db.getSchemaVersion()).toBe(64); + expect(db.getSchemaVersion()).toBe(65); }); it("does not overwrite existing config on re-init", () => { // Update the config @@ -966,7 +970,7 @@ describe("schema migrations", () => { db.init(); // Verify version bumped to 29 (includes v1→v2 through v26→v29) - expect(db.getSchemaVersion()).toBe(64); + expect(db.getSchemaVersion()).toBe(65); // Verify new columns exist and existing data is intact const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -991,11 +995,11 @@ describe("schema migrations", () => { const db = new Database(fusionDir); db.init(); - expect(db.getSchemaVersion()).toBe(64); + expect(db.getSchemaVersion()).toBe(65); // Re-init should not fail db.init(); - expect(db.getSchemaVersion()).toBe(64); + expect(db.getSchemaVersion()).toBe(65); db.close(); }); @@ -1030,7 +1034,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(64); + expect(db.getSchemaVersion()).toBe(65); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; expect(cols.map((col) => col.name)).toContain("priority"); @@ -1071,7 +1075,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(64); + expect(db.getSchemaVersion()).toBe(65); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; const colNames = cols.map((col) => col.name); @@ -1140,7 +1144,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(64); + expect(db.getSchemaVersion()).toBe(65); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; const colNames = cols.map((col) => col.name); @@ -1243,7 +1247,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(64); + expect(db.getSchemaVersion()).toBe(65); const cols = db.prepare("PRAGMA table_info(chat_messages)").all() as Array<{ name: string }>; expect(cols.map((col) => col.name)).toContain("attachments"); @@ -1317,7 +1321,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(64); + expect(db.getSchemaVersion()).toBe(65); 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" }]); @@ -1341,7 +1345,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(64); + expect(db.getSchemaVersion()).toBe(65); 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" }]); @@ -1445,7 +1449,7 @@ describe("schema migrations", () => { db.init(); // Verify version bumped to 29 - expect(db.getSchemaVersion()).toBe(64); + expect(db.getSchemaVersion()).toBe(65); // Verify new columns exist and existing data is intact const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -1914,7 +1918,7 @@ describe("createDatabase factory", () => { const db = createDatabase(fusionDir); db.init(); - expect(db.getSchemaVersion()).toBe(64); + expect(db.getSchemaVersion()).toBe(65); expect(db.getLastModified()).toBeGreaterThan(0); db.close(); @@ -2043,7 +2047,7 @@ describe("migration v63 project auth tables", () => { const migrated = new Database(fusion); migrated.init(); - expect(migrated.getSchemaVersion()).toBe(64); + expect(migrated.getSchemaVersion()).toBe(65); const tables = migrated .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'project_auth_%' ORDER BY name") .all() as Array<{ name: string }>; diff --git a/packages/core/src/__tests__/distributed-task-id.test.ts b/packages/core/src/__tests__/distributed-task-id.test.ts new file mode 100644 index 000000000..88754d94e --- /dev/null +++ b/packages/core/src/__tests__/distributed-task-id.test.ts @@ -0,0 +1,75 @@ +import { describe, expect, it } from "vitest"; +import { Database } from "../db.js"; +import { createDistributedTaskIdAllocator, DistributedTaskIdError } from "../distributed-task-id.js"; + +describe("distributed-task-id allocator", () => { + const createAllocator = () => { + const db = new Database("/tmp/fusion-test", { inMemory: true }); + db.init(); + return { db, allocator: createDistributedTaskIdAllocator(db) }; + }; + + it("returns unique sequential IDs across concurrent reservations", async () => { + const { allocator } = createAllocator(); + const reservations = await Promise.all( + Array.from({ length: 10 }, () => allocator.reserveDistributedTaskId({ prefix: "fn", nodeId: "node-a" })), + ); + const ids = reservations.map((r) => r.taskId); + expect(new Set(ids).size).toBe(10); + expect(ids[0]).toBe("FN-001"); + expect(ids[9]).toBe("FN-010"); + }); + + it("commit increments committedClusterTaskCount by one", async () => { + const { allocator } = createAllocator(); + const reservation = await allocator.reserveDistributedTaskId({ prefix: "FN", nodeId: "node-a" }); + const committed = await allocator.commitDistributedTaskIdReservation({ + reservationId: reservation.reservationId, + nodeId: "node-a", + }); + expect(committed.committedClusterTaskCount).toBe(reservation.committedClusterTaskCount + 1); + }); + + it("abort burns the sequence and does not increment committed count", async () => { + const { allocator } = createAllocator(); + const reservation = await allocator.reserveDistributedTaskId({ prefix: "FN", nodeId: "node-a" }); + const aborted = await allocator.abortDistributedTaskIdReservation({ + reservationId: reservation.reservationId, + nodeId: "node-a", + reason: "failed-create", + }); + expect(aborted.committedClusterTaskCount).toBe(reservation.committedClusterTaskCount); + const state = await allocator.getDistributedTaskIdState({ prefix: "FN" }); + expect(state.burnedReservationCount).toBe(1); + }); + + it("expired reservations cannot be committed and count as burned", async () => { + const { allocator } = createAllocator(); + const reservation = await allocator.reserveDistributedTaskId({ + prefix: "FN", + nodeId: "node-a", + ttlMs: 1, + }); + await new Promise((resolve) => setTimeout(resolve, 5)); + await expect( + allocator.commitDistributedTaskIdReservation({ reservationId: reservation.reservationId, nodeId: "node-a" }), + ).rejects.toBeInstanceOf(DistributedTaskIdError); + + const state = await allocator.getDistributedTaskIdState({ prefix: "FN" }); + expect(state.burnedReservationCount).toBe(1); + expect(state.committedClusterTaskCount).toBe(0); + }); + + it("state reports committed count independently from nextSequence", async () => { + const { allocator } = createAllocator(); + const first = await allocator.reserveDistributedTaskId({ prefix: "FN", nodeId: "node-a" }); + await allocator.abortDistributedTaskIdReservation({ reservationId: first.reservationId, nodeId: "node-a", reason: "abort" }); + + const second = await allocator.reserveDistributedTaskId({ prefix: "FN", nodeId: "node-a" }); + await allocator.commitDistributedTaskIdReservation({ reservationId: second.reservationId, nodeId: "node-a" }); + + const state = await allocator.getDistributedTaskIdState({ prefix: "FN" }); + expect(state.nextSequence).toBe(3); + expect(state.committedClusterTaskCount).toBe(1); + }); +}); diff --git a/packages/core/src/__tests__/insight-store.test.ts b/packages/core/src/__tests__/insight-store.test.ts index 5981a29db..4e7dbdb96 100644 --- a/packages/core/src/__tests__/insight-store.test.ts +++ b/packages/core/src/__tests__/insight-store.test.ts @@ -886,7 +886,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(64); + expect(db1.getSchemaVersion()).toBe(65); db1.close(); // Step 2: Manually downgrade to version 32 and drop insight tables @@ -921,7 +921,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(64); + expect(db3.getSchemaVersion()).toBe(65); // Step 4: Verify insight tables exist after migration const tablesAfter = db3.prepare( @@ -952,12 +952,12 @@ describe("Migration: pre-33 DB upgrade", () => { try { const db1 = createDatabase(testDir); db1.init(); - expect(db1.getSchemaVersion()).toBe(64); + expect(db1.getSchemaVersion()).toBe(65); db1.close(); const db2 = createDatabase(testDir); expect(() => db2.init()).not.toThrow(); - expect(db2.getSchemaVersion()).toBe(64); + expect(db2.getSchemaVersion()).toBe(65); db2.close(); } finally { rmSync(testDir, { recursive: true, force: true }); @@ -971,7 +971,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(64); + expect(db1.getSchemaVersion()).toBe(65); // Step 2: Strip lifecycle and cancelledAt columns by recreating the // table without them. This simulates a DB that was created before the diff --git a/packages/core/src/__tests__/mission-store.test.ts b/packages/core/src/__tests__/mission-store.test.ts index db00eb79c..f544ea2ef 100644 --- a/packages/core/src/__tests__/mission-store.test.ts +++ b/packages/core/src/__tests__/mission-store.test.ts @@ -2629,7 +2629,7 @@ describe("MissionStore", () => { describe("Loop State & Validator Run Schema (v31)", () => { it("schema version is 40 after migration", () => { - expect(db.getSchemaVersion()).toBe(64); + expect(db.getSchemaVersion()).toBe(65); }); it("mission_features table has loop state columns", () => { diff --git a/packages/core/src/__tests__/roadmap-store.test.ts b/packages/core/src/__tests__/roadmap-store.test.ts index a3609f3cc..9cab98e5d 100644 --- a/packages/core/src/__tests__/roadmap-store.test.ts +++ b/packages/core/src/__tests__/roadmap-store.test.ts @@ -742,7 +742,7 @@ describe("RoadmapStore", () => { describe("schema version", () => { it("schema version is 40 after init", () => { - expect(db.getSchemaVersion()).toBe(64); + expect(db.getSchemaVersion()).toBe(65); }); }); diff --git a/packages/core/src/__tests__/run-audit.test.ts b/packages/core/src/__tests__/run-audit.test.ts index 5f1e0c147..fdcbc5499 100644 --- a/packages/core/src/__tests__/run-audit.test.ts +++ b/packages/core/src/__tests__/run-audit.test.ts @@ -465,7 +465,7 @@ describe("Run Audit", () => { }); it("schema version is bumped to 40", () => { - expect(db.getSchemaVersion()).toBe(64); + expect(db.getSchemaVersion()).toBe(65); }); }); }); diff --git a/packages/core/src/__tests__/store.test.ts b/packages/core/src/__tests__/store.test.ts index d6af18a90..5872a9422 100644 --- a/packages/core/src/__tests__/store.test.ts +++ b/packages/core/src/__tests__/store.test.ts @@ -11350,6 +11350,14 @@ describe("RunMutationContext", () => { }); }); + describe("distributed task-id allocator seam", () => { + it("returns a stable allocator instance", () => { + const first = store.getDistributedTaskIdAllocator(); + const second = store.getDistributedTaskIdAllocator(); + expect(first).toBe(second); + }); + }); + describe("FTS5 corruption recovery during upsert", () => { it("rebuilds FTS5 and retries once when upsert fails with an FTS corruption error", async () => { const db = store.getDatabase(); diff --git a/packages/core/src/__tests__/task-documents.test.ts b/packages/core/src/__tests__/task-documents.test.ts index 0b95bb760..984532910 100644 --- a/packages/core/src/__tests__/task-documents.test.ts +++ b/packages/core/src/__tests__/task-documents.test.ts @@ -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(64); + expect(db.getSchemaVersion()).toBe(65); const index = db .prepare( diff --git a/packages/core/src/db.ts b/packages/core/src/db.ts index 2b9092b4d..426eebbc7 100644 --- a/packages/core/src/db.ts +++ b/packages/core/src/db.ts @@ -88,7 +88,7 @@ export function probeFts5(db: DatabaseSync): boolean { // ── Schema Definition ──────────────────────────────────────────────── -const SCHEMA_VERSION = 64; +const SCHEMA_VERSION = 65; function normalizeTaskComments( steeringComments: SteeringComment[] | undefined, @@ -237,6 +237,35 @@ CREATE TABLE IF NOT EXISTS config ( updatedAt TEXT ); +CREATE TABLE IF NOT EXISTS distributed_task_id_state ( + prefix TEXT PRIMARY KEY, + nextSequence INTEGER NOT NULL, + committedClusterTaskCount INTEGER NOT NULL, + lastCommittedTaskId TEXT, + updatedAt TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS distributed_task_id_reservations ( + reservationId TEXT PRIMARY KEY, + prefix TEXT NOT NULL, + nodeId TEXT NOT NULL, + sequence INTEGER NOT NULL, + taskId TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ('reserved', 'committed', 'aborted', 'expired')), + reason TEXT CHECK (reason IS NULL OR reason IN ('abort', 'expired', 'failed-create')), + expiresAt TEXT NOT NULL, + committedAt TEXT, + abortedAt TEXT, + createdAt TEXT NOT NULL, + updatedAt TEXT NOT NULL, + FOREIGN KEY (prefix) REFERENCES distributed_task_id_state(prefix) ON DELETE CASCADE, + UNIQUE(prefix, sequence), + UNIQUE(prefix, taskId) +); + +CREATE INDEX IF NOT EXISTS idxDistributedTaskIdReservationsPrefixStatus ON distributed_task_id_reservations(prefix, status); +CREATE INDEX IF NOT EXISTS idxDistributedTaskIdReservationsExpiry ON distributed_task_id_reservations(status, expiresAt); + -- Workflow step definitions CREATE TABLE IF NOT EXISTS workflow_steps ( id TEXT PRIMARY KEY, @@ -2706,6 +2735,41 @@ export class Database { }); } + if (version < 65) { + this.applyMigration(65, () => { + this.db.exec(` + CREATE TABLE IF NOT EXISTS distributed_task_id_state ( + prefix TEXT PRIMARY KEY, + nextSequence INTEGER NOT NULL, + committedClusterTaskCount INTEGER NOT NULL, + lastCommittedTaskId TEXT, + updatedAt TEXT NOT NULL + ) + `); + this.db.exec(` + CREATE TABLE IF NOT EXISTS distributed_task_id_reservations ( + reservationId TEXT PRIMARY KEY, + prefix TEXT NOT NULL, + nodeId TEXT NOT NULL, + sequence INTEGER NOT NULL, + taskId TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ('reserved', 'committed', 'aborted', 'expired')), + reason TEXT CHECK (reason IS NULL OR reason IN ('abort', 'expired', 'failed-create')), + expiresAt TEXT NOT NULL, + committedAt TEXT, + abortedAt TEXT, + createdAt TEXT NOT NULL, + updatedAt TEXT NOT NULL, + FOREIGN KEY (prefix) REFERENCES distributed_task_id_state(prefix) ON DELETE CASCADE, + UNIQUE(prefix, sequence), + UNIQUE(prefix, taskId) + ) + `); + this.db.exec(`CREATE INDEX IF NOT EXISTS idxDistributedTaskIdReservationsPrefixStatus ON distributed_task_id_reservations(prefix, status)`); + this.db.exec(`CREATE INDEX IF NOT EXISTS idxDistributedTaskIdReservationsExpiry ON distributed_task_id_reservations(status, expiresAt)`); + }); + } + } /** diff --git a/packages/core/src/distributed-task-id.ts b/packages/core/src/distributed-task-id.ts new file mode 100644 index 000000000..657899399 --- /dev/null +++ b/packages/core/src/distributed-task-id.ts @@ -0,0 +1,289 @@ +import { randomUUID } from "node:crypto"; +import type { Database } from "./db.js"; +import type { + DistributedTaskIdAbortInput, + DistributedTaskIdAbortResult, + DistributedTaskIdCommitInput, + DistributedTaskIdCommitResult, + DistributedTaskIdReserveInput, + DistributedTaskIdReserveResult, + DistributedTaskIdStateInput, + DistributedTaskIdStateResult, +} from "./types.js"; + +const DEFAULT_RESERVATION_TTL_MS = 15 * 60 * 1000; + +export interface DistributedTaskIdAllocator { + formatDistributedTaskId(prefix: string, sequence: number): string; + reserveDistributedTaskId(input: DistributedTaskIdReserveInput): Promise; + commitDistributedTaskIdReservation(input: DistributedTaskIdCommitInput): Promise; + abortDistributedTaskIdReservation(input: DistributedTaskIdAbortInput): Promise; + getDistributedTaskIdState(input: DistributedTaskIdStateInput): Promise; +} + +export class DistributedTaskIdError extends Error { + constructor( + message: string, + public readonly code: + | "reservation_not_found" + | "reservation_not_owned" + | "reservation_expired" + | "reservation_finalized" + | "invalid_prefix", + ) { + super(message); + } +} + +type ReservationRow = { + reservationId: string; + prefix: string; + nodeId: string; + sequence: number; + taskId: string; + status: "reserved" | "committed" | "aborted" | "expired"; + reason: "abort" | "expired" | "failed-create" | null; + expiresAt: string; + committedAt: string | null; + abortedAt: string | null; +}; + +export function formatDistributedTaskId(prefix: string, sequence: number): string { + const normalizedPrefix = prefix.trim().toUpperCase(); + if (!normalizedPrefix) { + throw new DistributedTaskIdError("prefix is required", "invalid_prefix"); + } + return `${normalizedPrefix}-${String(sequence).padStart(3, "0")}`; +} + +export function createDistributedTaskIdAllocator(db: Database): DistributedTaskIdAllocator { + let opLock: Promise = Promise.resolve(); + const withLock = async (fn: () => Promise): Promise => { + const prev = opLock; + let resolve!: () => void; + opLock = new Promise((r) => { + resolve = r; + }); + await prev; + try { + return await fn(); + } finally { + resolve(); + } + }; + + const expireReservations = (nowIso: string): number => { + const result = db.prepare( + `UPDATE distributed_task_id_reservations + SET status = 'expired', reason = 'expired', abortedAt = ? + WHERE status = 'reserved' AND expiresAt <= ?`, + ).run(nowIso, nowIso) as { changes?: number }; + return result.changes ?? 0; + }; + + const ensureStateRow = (prefix: string): void => { + db.prepare( + `INSERT OR IGNORE INTO distributed_task_id_state ( + prefix, nextSequence, committedClusterTaskCount, lastCommittedTaskId, updatedAt + ) VALUES (?, 1, 0, NULL, ?)` + ).run(prefix, new Date().toISOString()); + }; + + return { + formatDistributedTaskId, + reserveDistributedTaskId: async (input) => + withLock(async () => { + const ttlMs = input.ttlMs ?? DEFAULT_RESERVATION_TTL_MS; + const now = new Date(); + const nowIso = now.toISOString(); + const expiresAt = new Date(now.getTime() + ttlMs).toISOString(); + + return db.transaction(() => { + expireReservations(nowIso); + const prefix = input.prefix.trim().toUpperCase(); + if (!prefix) { + throw new DistributedTaskIdError("prefix is required", "invalid_prefix"); + } + ensureStateRow(prefix); + + const state = db + .prepare( + "SELECT nextSequence, committedClusterTaskCount FROM distributed_task_id_state WHERE prefix = ?", + ) + .get(prefix) as { nextSequence: number; committedClusterTaskCount: number }; + + const sequence = state.nextSequence; + const taskId = formatDistributedTaskId(prefix, sequence); + const reservationId = randomUUID(); + + db.prepare( + `INSERT INTO distributed_task_id_reservations ( + reservationId, prefix, nodeId, sequence, taskId, status, reason, expiresAt, createdAt, updatedAt + ) VALUES (?, ?, ?, ?, ?, 'reserved', NULL, ?, ?, ?)` + ).run(reservationId, prefix, input.nodeId, sequence, taskId, expiresAt, nowIso, nowIso); + + db.prepare( + "UPDATE distributed_task_id_state SET nextSequence = ?, updatedAt = ? WHERE prefix = ?", + ).run(sequence + 1, nowIso, prefix); + db.bumpLastModified(); + + return { + reservationId, + taskId, + sequence, + expiresAt, + committedClusterTaskCount: state.committedClusterTaskCount, + }; + }); + }), + commitDistributedTaskIdReservation: async (input) => + withLock(async () => { + const nowIso = new Date().toISOString(); + return db.transaction(() => { + expireReservations(nowIso); + const row = db + .prepare( + `SELECT reservationId, prefix, nodeId, sequence, taskId, status, reason, expiresAt, committedAt, abortedAt + FROM distributed_task_id_reservations + WHERE reservationId = ?`, + ) + .get(input.reservationId) as ReservationRow | undefined; + + if (!row) { + throw new DistributedTaskIdError("reservation not found", "reservation_not_found"); + } + if (row.nodeId !== input.nodeId) { + throw new DistributedTaskIdError("reservation belongs to a different node", "reservation_not_owned"); + } + if (row.status === "expired") { + throw new DistributedTaskIdError("reservation has expired", "reservation_expired"); + } + if (row.status !== "reserved") { + throw new DistributedTaskIdError("reservation already finalized", "reservation_finalized"); + } + + db.prepare( + `UPDATE distributed_task_id_reservations + SET status = 'committed', committedAt = ?, updatedAt = ? + WHERE reservationId = ?`, + ).run(nowIso, nowIso, row.reservationId); + + ensureStateRow(row.prefix); + db.prepare( + `UPDATE distributed_task_id_state + SET committedClusterTaskCount = committedClusterTaskCount + 1, + lastCommittedTaskId = ?, + updatedAt = ? + WHERE prefix = ?`, + ).run(row.taskId, nowIso, row.prefix); + + const state = db + .prepare( + "SELECT committedClusterTaskCount FROM distributed_task_id_state WHERE prefix = ?", + ) + .get(row.prefix) as { committedClusterTaskCount: number }; + db.bumpLastModified(); + + return { + reservationId: row.reservationId, + taskId: row.taskId, + sequence: row.sequence, + committedClusterTaskCount: state.committedClusterTaskCount, + committedAt: nowIso, + }; + }); + }), + abortDistributedTaskIdReservation: async (input) => + withLock(async () => { + const nowIso = new Date().toISOString(); + return db.transaction(() => { + expireReservations(nowIso); + const row = db + .prepare( + `SELECT reservationId, prefix, nodeId, sequence, taskId, status, reason, expiresAt, committedAt, abortedAt + FROM distributed_task_id_reservations + WHERE reservationId = ?`, + ) + .get(input.reservationId) as ReservationRow | undefined; + + if (!row) { + throw new DistributedTaskIdError("reservation not found", "reservation_not_found"); + } + if (row.nodeId !== input.nodeId) { + throw new DistributedTaskIdError("reservation belongs to a different node", "reservation_not_owned"); + } + if (row.status === "committed") { + throw new DistributedTaskIdError("reservation already finalized", "reservation_finalized"); + } + + if (row.status === "reserved") { + db.prepare( + `UPDATE distributed_task_id_reservations + SET status = 'aborted', reason = ?, abortedAt = ?, updatedAt = ? + WHERE reservationId = ?`, + ).run(input.reason, nowIso, nowIso, row.reservationId); + } + + ensureStateRow(row.prefix); + const state = db + .prepare( + "SELECT committedClusterTaskCount FROM distributed_task_id_state WHERE prefix = ?", + ) + .get(row.prefix) as { committedClusterTaskCount: number }; + db.bumpLastModified(); + + return { + reservationId: row.reservationId, + taskId: row.taskId, + sequence: row.sequence, + committedClusterTaskCount: state.committedClusterTaskCount, + abortedAt: nowIso, + }; + }); + }), + getDistributedTaskIdState: async (input) => + withLock(async () => { + const nowIso = new Date().toISOString(); + return db.transaction(() => { + expireReservations(nowIso); + const prefix = input.prefix.trim().toUpperCase(); + if (!prefix) { + throw new DistributedTaskIdError("prefix is required", "invalid_prefix"); + } + ensureStateRow(prefix); + const row = db + .prepare( + `SELECT nextSequence, committedClusterTaskCount, lastCommittedTaskId + FROM distributed_task_id_state + WHERE prefix = ?`, + ) + .get(prefix) as { + nextSequence: number; + committedClusterTaskCount: number; + lastCommittedTaskId: string | null; + }; + + const active = db + .prepare( + `SELECT COUNT(*) AS count FROM distributed_task_id_reservations + WHERE prefix = ? AND status = 'reserved'`, + ) + .get(prefix) as { count: number }; + const burned = db + .prepare( + `SELECT COUNT(*) AS count FROM distributed_task_id_reservations + WHERE prefix = ? AND status IN ('aborted', 'expired')`, + ) + .get(prefix) as { count: number }; + + return { + nextSequence: row.nextSequence, + committedClusterTaskCount: row.committedClusterTaskCount, + activeReservationCount: active.count, + burnedReservationCount: burned.count, + lastCommittedTaskId: row.lastCommittedTaskId ?? undefined, + }; + }); + }), + }; +} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 62c221da0..462699b77 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -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, THINKING_LEVELS, THEME_MODES, COLOR_THEMES, WORKFLOW_STEP_TEMPLATES, AGENT_PERMISSIONS, agentToConfigSnapshot, diffConfigSnapshots, isEphemeralAgent, hasAgentIdentity, CheckoutConflictError, DEFAULT_HEARTBEAT_PROCEDURE_PATH, getDefaultHeartbeatProcedurePath, EXECUTION_MODES, DEFAULT_EXECUTION_MODE, TASK_PRIORITIES, DEFAULT_TASK_PRIORITY, DASHBOARD_USER_ID, normalizeMessageParticipant, validateMessageMetadata, validateDockerNodeConfig, sanitizeDockerNodeConfigForResponse, normalizeMergeConflictStrategy, buildResearchDocumentKey, PROJECT_AUTH_ROLES } from "./types.js"; -export type { Column, IssueInfo, IssueState, TaskSourceIssue, PrInfo, PrStatus, Task, TaskTokenUsage, TaskAttachment, TaskComment, TaskCommentInput, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, TaskCreateInput, TaskSource, SourceType, TaskDetail, InboxTask, TodoList, TodoItem, TodoListCreateInput, TodoListUpdateInput, TodoItemCreateInput, TodoItemUpdateInput, TodoListWithItems, ProjectAuthRole, ProjectAuthUser, ProjectAuthMembership, ProjectAuthProvider, ProjectAuthSession, ProjectAuthUserCreateInput, ProjectAuthMembershipCreateInput, ProjectAuthProviderCreateInput, ProjectAuthSessionCreateInput, AgentLogEntry, AgentLogType, AgentRole, BoardConfig, AutostashOutcome, MergeDetails, MergeResult, MergeConflictStrategy, CanonicalMergeConflictStrategy, Settings, GlobalSettings, ProjectSettings, WebSearchBackend, ResearchEnabledSources, ResearchGlobalDefaults, ResearchProjectLimits, ResearchProjectSettings, SettingsScope, DaemonTokenSettings, TaskStep, StepStatus, TaskLogEntry, RunMutationContext, ActivityLogEntry, ActivityEventType, ThinkingLevel, ThemeMode, ColorTheme, ExecutionMode, TaskPriority, UnavailableNodePolicy, PlanningQuestion, PlanningSummary, PlanningResponse, PlanningQuestionType, ArchivedTaskEntry, BatchStatusRequest, BatchStatusResponse, BatchStatusEntry, BatchStatusResult, ModelPreset, WorkflowStep, WorkflowStepMode, WorkflowStepPhase, WorkflowStepInput, WorkflowStepResult, WorkflowStepTemplate, Agent, OrgTreeNode, AgentState, AgentDetail, AgentCreateInput, AgentUpdateInput, AgentApiKey, AgentApiKeyCreateResult, AgentCapability, AgentPromptTemplate, AgentPromptsConfig, AgentPermission, 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, RunAuditDomain, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter } from "./types.js"; +export type { Column, IssueInfo, IssueState, TaskSourceIssue, PrInfo, PrStatus, Task, TaskTokenUsage, TaskAttachment, TaskComment, TaskCommentInput, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, TaskCreateInput, TaskSource, SourceType, TaskDetail, InboxTask, TodoList, TodoItem, TodoListCreateInput, TodoListUpdateInput, TodoItemCreateInput, TodoItemUpdateInput, TodoListWithItems, ProjectAuthRole, ProjectAuthUser, ProjectAuthMembership, ProjectAuthProvider, ProjectAuthSession, ProjectAuthUserCreateInput, ProjectAuthMembershipCreateInput, ProjectAuthProviderCreateInput, ProjectAuthSessionCreateInput, AgentLogEntry, AgentLogType, AgentRole, BoardConfig, DistributedTaskIdReserveInput, DistributedTaskIdReserveResult, DistributedTaskIdCommitInput, DistributedTaskIdCommitResult, DistributedTaskIdAbortInput, DistributedTaskIdAbortResult, DistributedTaskIdStateInput, DistributedTaskIdStateResult, AutostashOutcome, MergeDetails, MergeResult, MergeConflictStrategy, CanonicalMergeConflictStrategy, Settings, GlobalSettings, ProjectSettings, WebSearchBackend, ResearchEnabledSources, ResearchGlobalDefaults, ResearchProjectLimits, ResearchProjectSettings, SettingsScope, DaemonTokenSettings, TaskStep, StepStatus, TaskLogEntry, RunMutationContext, ActivityLogEntry, ActivityEventType, ThinkingLevel, ThemeMode, ColorTheme, ExecutionMode, TaskPriority, UnavailableNodePolicy, PlanningQuestion, PlanningSummary, PlanningResponse, PlanningQuestionType, ArchivedTaskEntry, BatchStatusRequest, BatchStatusResponse, BatchStatusEntry, BatchStatusResult, ModelPreset, WorkflowStep, WorkflowStepMode, WorkflowStepPhase, WorkflowStepInput, WorkflowStepResult, WorkflowStepTemplate, Agent, OrgTreeNode, AgentState, AgentDetail, AgentCreateInput, AgentUpdateInput, AgentApiKey, AgentApiKeyCreateResult, AgentCapability, AgentPromptTemplate, AgentPromptsConfig, AgentPermission, 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, RunAuditDomain, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter } from "./types.js"; export { AGENT_VALID_TRANSITIONS } from "./types.js"; export * from "./mesh-replication-protocol.js"; export { @@ -46,6 +46,12 @@ export type { ReflectionStoreEvents } from "./reflection-store.js"; export { MessageStore } from "./message-store.js"; export type { MessageStoreEvents } from "./message-store.js"; export { TaskStore } from "./store.js"; +export { + createDistributedTaskIdAllocator, + formatDistributedTaskId, + DistributedTaskIdError, +} from "./distributed-task-id.js"; +export type { DistributedTaskIdAllocator } from "./distributed-task-id.js"; export { ProjectAuthStore } from "./project-auth-store.js"; export { Database, createDatabase, toJson, toJsonNullable, fromJson } from "./db.js"; export type { Statement } from "./db.js"; diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index b9f5b00f4..4db5b5970 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -27,6 +27,7 @@ import { createLogger } from "./logger.js"; import { validateNodeOverrideChange } from "./node-override-guard.js"; import { sanitizeTitle } from "./ai-summarize.js"; import { assertProjectRootDir } from "./project-root-guard.js"; +import { createDistributedTaskIdAllocator, type DistributedTaskIdAllocator } from "./distributed-task-id.js"; /** Database row shape for the tasks table (all columns). */ interface TaskRow { @@ -526,6 +527,8 @@ export class TaskStore extends EventEmitter { private evalStore: EvalStore | null = null; /** Cached ProjectAuthStore instance */ private projectAuthStore: ProjectAuthStore | null = null; + /** Cached distributed task-id allocator instance. */ + private distributedTaskIdAllocator: DistributedTaskIdAllocator | null = null; /** Buffer for batching agent log writes to reduce WAL pressure. */ private agentLogBuffer: Array<{ @@ -6497,6 +6500,13 @@ ${stepsSection}`; return this.db; } + getDistributedTaskIdAllocator(): DistributedTaskIdAllocator { + if (!this.distributedTaskIdAllocator) { + this.distributedTaskIdAllocator = createDistributedTaskIdAllocator(this.db); + } + return this.distributedTaskIdAllocator; + } + /** * Perform a simple database health check. * Returns true if the database responds correctly, false otherwise. diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index 050cec3bf..40ba46149 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -2208,6 +2208,59 @@ export interface BoardConfig { settings?: Settings; } +export interface DistributedTaskIdReserveInput { + prefix: string; + nodeId: string; + ttlMs?: number; +} + +export interface DistributedTaskIdReserveResult { + reservationId: string; + taskId: string; + sequence: number; + expiresAt: string; + committedClusterTaskCount: number; +} + +export interface DistributedTaskIdCommitInput { + reservationId: string; + nodeId: string; +} + +export interface DistributedTaskIdCommitResult { + reservationId: string; + taskId: string; + sequence: number; + committedClusterTaskCount: number; + committedAt: string; +} + +export interface DistributedTaskIdAbortInput { + reservationId: string; + nodeId: string; + reason: "abort" | "expired" | "failed-create"; +} + +export interface DistributedTaskIdAbortResult { + reservationId: string; + taskId: string; + sequence: number; + committedClusterTaskCount: number; + abortedAt: string; +} + +export interface DistributedTaskIdStateInput { + prefix: string; +} + +export interface DistributedTaskIdStateResult { + nextSequence: number; + committedClusterTaskCount: number; + activeReservationCount: number; + burnedReservationCount: number; + lastCommittedTaskId?: string; +} + /** * Outcome of restoring the developer's pre-merge autostash after the merge * completes. Surfaced on MergeResult so the UI / dashboard can show whether diff --git a/packages/dashboard/app/components/ChatView.css b/packages/dashboard/app/components/ChatView.css index 340159395..7f1888ebc 100644 --- a/packages/dashboard/app/components/ChatView.css +++ b/packages/dashboard/app/components/ChatView.css @@ -1105,14 +1105,11 @@ .chat-thread[style*="--keyboard-overlap"] { height: calc(var(--vv-height, calc(100dvh - var(--keyboard-overlap, 0px))) - var(--header-height)); max-height: calc(var(--vv-height, calc(100dvh - var(--keyboard-overlap, 0px))) - var(--header-height)); - /* The previous translateY(--vv-offset-top) compensated for iOS - shifting the visual viewport on focus, but it also amplified - jitter during a swipe — visualViewport scroll events update - --vv-offset-top rapidly, the transform follows, and the thread - jutters / shifts ~300px / exposes the body background. Dropping - the transform lets the browser place the thread where iOS - expects; height-shrink to vv-height still keeps it inside the - visible area. */ + /* useMobileKeyboard pins --vv-offset-top across pan-time scroll + events (only resize/focus update it), so this transform anchors + the thread on initial focus without juddering during swipes. */ + transform: translateY(var(--vv-offset-top, 0px)); + will-change: transform; } /* On mobile, the "New Chat" affordance uses a full-width pinned footer. */ diff --git a/packages/dashboard/app/components/ChatView.tsx b/packages/dashboard/app/components/ChatView.tsx index cb731bb63..869b19d49 100644 --- a/packages/dashboard/app/components/ChatView.tsx +++ b/packages/dashboard/app/components/ChatView.tsx @@ -773,7 +773,6 @@ export function ChatView({ projectId, addToast }: ChatViewProps) { const messagesContainerRef = useRef(null); const inputRef = useRef(null); const preserveComposerFocusRef = useRef(false); - const handledMobileSendRef = useRef(false); const fileInputRef = useRef(null); const pendingAttachmentsRef = useRef([]); const mentionCursorPosRef = useRef(0); @@ -1088,12 +1087,6 @@ export function ChatView({ projectId, addToast }: ChatViewProps) { }); }, []); - const markPreserveComposerFocus = useCallback(() => { - if (typeof window === "undefined") return; - if (window.innerWidth > 768) return; - preserveComposerFocusRef.current = true; - }, []); - const handleSkillSelect = useCallback( (skill: DiscoveredSkill) => { setMessageInput((currentInput) => { diff --git a/packages/dashboard/app/hooks/useMobileKeyboard.ts b/packages/dashboard/app/hooks/useMobileKeyboard.ts index eed8b85fb..87dd40bbc 100644 --- a/packages/dashboard/app/hooks/useMobileKeyboard.ts +++ b/packages/dashboard/app/hooks/useMobileKeyboard.ts @@ -133,6 +133,8 @@ export function useMobileKeyboard( return; } + // Full update for resize + focus transitions (real keyboard + // open/close events). const update = () => { const metrics = getKeyboardMetrics(); setKeyboardOverlap(metrics.overlap); @@ -141,15 +143,29 @@ export function useMobileKeyboard( setKeyboardOpen(metrics.open); }; + // Scroll-only update: visualViewport.scroll fires at 60fps during + // an iOS pan with the keyboard up. Routing offsetTop through React + // state on every event amplifies the pan into a visible judder via + // the .chat-thread translateY(--vv-offset-top) transform. We skip + // offsetTop here; it stays pinned to whatever resize/focus last + // captured. Other metrics still update so a true viewport shrink + // is reflected. + const updateScrollOnly = () => { + const metrics = getKeyboardMetrics(); + setKeyboardOverlap(metrics.overlap); + setViewportHeight(metrics.vvHeight); + setKeyboardOpen(metrics.open); + }; + update(); vv.addEventListener("resize", update); - vv.addEventListener("scroll", update); + vv.addEventListener("scroll", updateScrollOnly); document.addEventListener("focusin", update); document.addEventListener("focusout", update); return () => { vv.removeEventListener("resize", update); - vv.removeEventListener("scroll", update); + vv.removeEventListener("scroll", updateScrollOnly); document.removeEventListener("focusin", update); document.removeEventListener("focusout", update); setKeyboardOverlap(0); diff --git a/packages/dashboard/src/__tests__/mesh-routes.test.ts b/packages/dashboard/src/__tests__/mesh-routes.test.ts index 635cf6777..75ef35cb5 100644 --- a/packages/dashboard/src/__tests__/mesh-routes.test.ts +++ b/packages/dashboard/src/__tests__/mesh-routes.test.ts @@ -24,6 +24,10 @@ const mockUpdateNode = vi.fn(); const mockGetLocalNode = vi.fn(); const mockGetSettingsForSync = vi.fn(); const mockApplyRemoteSettings = vi.fn(); +const mockReserveDistributedTaskId = vi.fn(); +const mockCommitDistributedTaskIdReservation = vi.fn(); +const mockAbortDistributedTaskIdReservation = vi.fn(); +const mockGetDistributedTaskIdState = vi.fn(); // Mock GlobalSettingsStore const mockGetSettings = vi.fn().mockResolvedValue({}); @@ -86,6 +90,15 @@ class MockStore extends EventEmitter { return mockGlobalSettingsStore; } + getDistributedTaskIdAllocator() { + return { + reserveDistributedTaskId: mockReserveDistributedTaskId, + commitDistributedTaskIdReservation: mockCommitDistributedTaskIdReservation, + abortDistributedTaskIdReservation: mockAbortDistributedTaskIdReservation, + getDistributedTaskIdState: mockGetDistributedTaskIdState, + }; + } + async listTasks(): Promise { return []; } @@ -173,6 +186,10 @@ describe("POST /api/mesh/sync", () => { }); mockGetNode.mockResolvedValue(undefined); mockUpdateNode.mockResolvedValue({ id: "node_remote", status: "online" }); + mockReserveDistributedTaskId.mockResolvedValue({ reservationId: "res-1", taskId: "FN-001", sequence: 1, expiresAt: "2030-01-01T00:00:00.000Z", committedClusterTaskCount: 0 }); + mockCommitDistributedTaskIdReservation.mockResolvedValue({ reservationId: "res-1", taskId: "FN-001", sequence: 1, committedClusterTaskCount: 1, committedAt: "2030-01-01T00:00:00.000Z" }); + mockAbortDistributedTaskIdReservation.mockResolvedValue({ reservationId: "res-1", taskId: "FN-001", sequence: 1, committedClusterTaskCount: 0, abortedAt: "2030-01-01T00:00:00.000Z" }); + mockGetDistributedTaskIdState.mockResolvedValue({ nextSequence: 2, committedClusterTaskCount: 1, activeReservationCount: 0, burnedReservationCount: 0, lastCommittedTaskId: "FN-001" }); mockGetLocalNode.mockResolvedValue({ id: "node_local", name: "local", @@ -654,3 +671,58 @@ describe("POST /api/mesh/sync", () => { }); }); }); + +describe("/api/mesh/task-ids routes", () => { + let app: ReturnType; + + beforeEach(async () => { + vi.clearAllMocks(); + mockInit.mockResolvedValue(undefined); + mockClose.mockResolvedValue(undefined); + mockGetLocalNode.mockResolvedValue({ id: "node_local", type: "local", name: "local", status: "online", maxConcurrent: 4, createdAt: "2026-04-01T10:00:00.000Z", updatedAt: "2026-04-01T12:00:00.000Z" }); + mockGetNode.mockResolvedValue(undefined); + mockReserveDistributedTaskId.mockResolvedValue({ reservationId: "res-1", taskId: "FN-001", sequence: 1, expiresAt: "2030-01-01T00:00:00.000Z", committedClusterTaskCount: 0 }); + mockCommitDistributedTaskIdReservation.mockResolvedValue({ reservationId: "res-1", taskId: "FN-001", sequence: 1, committedClusterTaskCount: 1, committedAt: "2030-01-01T00:00:00.000Z" }); + mockAbortDistributedTaskIdReservation.mockResolvedValue({ reservationId: "res-1", taskId: "FN-001", sequence: 1, committedClusterTaskCount: 0, abortedAt: "2030-01-01T00:00:00.000Z" }); + mockGetDistributedTaskIdState.mockResolvedValue({ nextSequence: 2, committedClusterTaskCount: 1, activeReservationCount: 0, burnedReservationCount: 0, lastCommittedTaskId: "FN-001" }); + app = createServer(new MockStore() as unknown as TaskStore); + }); + + it("reserves distributed task ids locally", async () => { + const response = await request(app, "POST", "/api/mesh/task-ids/reserve", JSON.stringify({ prefix: "FN", nodeId: "node-a" }), { "Content-Type": "application/json" }); + expect(response.status).toBe(200); + expect(mockReserveDistributedTaskId).toHaveBeenCalledWith({ prefix: "FN", nodeId: "node-a", ttlMs: undefined }); + expect((response.body as any).committedClusterTaskCount).toBe(0); + }); + + it("returns allocator state with authoritative committedClusterTaskCount", async () => { + const response = await request(app, "GET", "/api/mesh/task-ids/state?prefix=FN"); + expect(response.status).toBe(200); + expect((response.body as any).committedClusterTaskCount).toBe(1); + }); + + it("rejects bad requests", async () => { + const response = await request(app, "POST", "/api/mesh/task-ids/abort", JSON.stringify({ reservationId: "r", nodeId: "n", reason: "bad" }), { "Content-Type": "application/json" }); + expect(response.status).toBe(400); + }); + + it("rejects unauthorized mesh caller", async () => { + mockGetNode.mockResolvedValue(makeNodeConfig({ id: "node_remote_1", apiKey: "secret" })); + const response = await request( + app, + "POST", + "/api/mesh/task-ids/reserve", + JSON.stringify({ prefix: "FN", nodeId: "node-a", senderNodeId: "node_remote_1" }), + { "Content-Type": "application/json", Authorization: "Bearer wrong" }, + ); + expect(response.status).toBe(401); + }); + + it("returns 503 when coordinator is unreachable for writes", async () => { + mockGetNode.mockResolvedValue(makeNodeConfig({ id: "node_remote_1", url: "https://remote.example.com", apiKey: "secret" })); + vi.stubGlobal("fetch", vi.fn().mockRejectedValue(new Error("network down"))); + const response = await request(app, "POST", "/api/mesh/task-ids/commit", JSON.stringify({ reservationId: "res-1", nodeId: "node-a", coordinatorNodeId: "node_remote_1" }), { "Content-Type": "application/json" }); + expect(response.status).toBe(503); + vi.unstubAllGlobals(); + }); +}); diff --git a/packages/dashboard/src/routes/register-mesh-routes.ts b/packages/dashboard/src/routes/register-mesh-routes.ts index 0096ece0e..8b82f1653 100644 --- a/packages/dashboard/src/routes/register-mesh-routes.ts +++ b/packages/dashboard/src/routes/register-mesh-routes.ts @@ -1,9 +1,58 @@ import { ApiError, badRequest } from "../api-error.js"; import type { ApiRouteRegistrar } from "./types.js"; +import { fetchFromRemoteNode } from "./register-settings-sync-helpers.js"; export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => { const { router, store, emitRemoteRouteDiagnostic, rethrowAsApiError } = ctx; + const resolveAllocator = async (coordinatorNodeId?: string) => { + const { CentralCore } = await import("@fusion/core"); + const central = new CentralCore(); + await central.init(); + if (!coordinatorNodeId) { + await central.close(); + return { mode: "local" as const }; + } + const coordinator = await central.getNode(coordinatorNodeId); + if (coordinator?.type === "local") { + await central.close(); + return { mode: "local" as const }; + } + await central.close(); + if (!coordinator) { + throw new ApiError(503, "Allocator coordinator is unavailable"); + } + return { mode: "remote" as const, coordinator }; + }; + + const mapCoordinatorWriteError = (err: unknown): never => { + if (err instanceof ApiError && [502, 504].includes(err.statusCode)) { + throw new ApiError(503, "Allocator coordinator is unavailable"); + } + throw err; + }; + + const requireMeshAuth = async ( + req: { headers: { authorization?: string } }, + res: { status: (code: number) => { json: (payload: unknown) => void } }, + senderNodeId?: string, + ): Promise => { + if (!senderNodeId) return true; + const { CentralCore } = await import("@fusion/core"); + const central = new CentralCore(); + await central.init(); + const senderNode = await central.getNode(senderNodeId); + await central.close(); + if (!senderNode?.apiKey) return true; + const authHeader = req.headers.authorization; + const token = authHeader?.startsWith("Bearer ") ? authHeader.slice(7) : undefined; + if (!token || token !== senderNode.apiKey) { + res.status(401).json({ error: "Unauthorized" }); + return false; + } + return true; + }; + // ── Mesh Topology Routes ──────────────────────────────────────────────── /** @@ -67,6 +116,124 @@ export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => { * Request body: PeerSyncRequest (may include optional settings field) * Response body: PeerSyncResponse (may include optional settings field) */ + router.post("/mesh/task-ids/reserve", async (req, res) => { + try { + const prefix = String(req.body?.prefix ?? "").trim(); + const nodeId = String(req.body?.nodeId ?? "").trim(); + const ttlMs = req.body?.ttlMs; + const coordinatorNodeId = typeof req.body?.coordinatorNodeId === "string" ? req.body.coordinatorNodeId : undefined; + const senderNodeId = typeof req.body?.senderNodeId === "string" ? req.body.senderNodeId : undefined; + if (!prefix) throw badRequest("prefix is required"); + if (!nodeId) throw badRequest("nodeId is required"); + if (!(await requireMeshAuth(req, res, senderNodeId))) return; + + const target = await resolveAllocator(coordinatorNodeId); + if (target.mode === "remote") { + try { + const remote = await fetchFromRemoteNode(target.coordinator, "/api/mesh/task-ids/reserve", { + method: "POST", + body: { prefix, nodeId, ttlMs }, + }); + res.json(remote); + return; + } catch (err) { + mapCoordinatorWriteError(err); + } + } + + const result = await store.getDistributedTaskIdAllocator().reserveDistributedTaskId({ prefix, nodeId, ttlMs }); + res.json(result); + } catch (err: unknown) { + if (err instanceof ApiError) throw err; + rethrowAsApiError(err); + } + }); + + router.post("/mesh/task-ids/commit", async (req, res) => { + try { + const reservationId = String(req.body?.reservationId ?? "").trim(); + const nodeId = String(req.body?.nodeId ?? "").trim(); + const coordinatorNodeId = typeof req.body?.coordinatorNodeId === "string" ? req.body.coordinatorNodeId : undefined; + const senderNodeId = typeof req.body?.senderNodeId === "string" ? req.body.senderNodeId : undefined; + if (!reservationId) throw badRequest("reservationId is required"); + if (!nodeId) throw badRequest("nodeId is required"); + if (!(await requireMeshAuth(req, res, senderNodeId))) return; + + const target = await resolveAllocator(coordinatorNodeId); + if (target.mode === "remote") { + try { + const remote = await fetchFromRemoteNode(target.coordinator, "/api/mesh/task-ids/commit", { + method: "POST", + body: { reservationId, nodeId }, + }); + res.json(remote); + return; + } catch (err) { + mapCoordinatorWriteError(err); + } + } + + const result = await store.getDistributedTaskIdAllocator().commitDistributedTaskIdReservation({ reservationId, nodeId }); + res.json(result); + } catch (err: unknown) { + if (err instanceof ApiError) throw err; + if (err instanceof Error && err.message.toLowerCase().includes("expired")) { + throw new ApiError(409, err.message); + } + rethrowAsApiError(err); + } + }); + + router.post("/mesh/task-ids/abort", async (req, res) => { + try { + const reservationId = String(req.body?.reservationId ?? "").trim(); + const nodeId = String(req.body?.nodeId ?? "").trim(); + const reason = req.body?.reason; + const coordinatorNodeId = typeof req.body?.coordinatorNodeId === "string" ? req.body.coordinatorNodeId : undefined; + const senderNodeId = typeof req.body?.senderNodeId === "string" ? req.body.senderNodeId : undefined; + if (!reservationId) throw badRequest("reservationId is required"); + if (!nodeId) throw badRequest("nodeId is required"); + if (reason !== "abort" && reason !== "expired" && reason !== "failed-create") { + throw badRequest("reason must be one of: abort, expired, failed-create"); + } + if (!(await requireMeshAuth(req, res, senderNodeId))) return; + + const target = await resolveAllocator(coordinatorNodeId); + if (target.mode === "remote") { + try { + const remote = await fetchFromRemoteNode(target.coordinator, "/api/mesh/task-ids/abort", { + method: "POST", + body: { reservationId, nodeId, reason }, + }); + res.json(remote); + return; + } catch (err) { + mapCoordinatorWriteError(err); + } + } + + const result = await store.getDistributedTaskIdAllocator().abortDistributedTaskIdReservation({ reservationId, nodeId, reason }); + res.json(result); + } catch (err: unknown) { + if (err instanceof ApiError) throw err; + rethrowAsApiError(err); + } + }); + + router.get("/mesh/task-ids/state", async (req, res) => { + try { + const prefix = String(req.query?.prefix ?? "").trim(); + const senderNodeId = typeof req.query?.senderNodeId === "string" ? req.query.senderNodeId : undefined; + if (!prefix) throw badRequest("prefix is required"); + if (!(await requireMeshAuth(req, res, senderNodeId))) return; + const result = await store.getDistributedTaskIdAllocator().getDistributedTaskIdState({ prefix }); + res.json(result); + } catch (err: unknown) { + if (err instanceof ApiError) throw err; + rethrowAsApiError(err); + } + }); + router.post("/mesh/sync", async (req, res) => { try { const { CentralCore } = await import("@fusion/core");