feat(FN-3960): add mesh replay outage contracts and runtime hooks

- Add core outage schema and mesh replay contract types, with CentralCore/CentralDB support for queue and snapshot persistence - Expose runtime replay hooks in engine health monitoring and peer exchange paths to drive outage recovery flows - Expand protocol and architecture docs for shared mesh replay behavior and finalized contract expectations - Add regression tests across core and engine for queue sequencing, snapshot contracts, and replay integration Fusion-Task-Id: FN-3960
2026-05-10 19:11:31 -07:00
parent 7d67dc3597
commit a5a794d12e
14 changed files with 717 additions and 15 deletions
--- a/packages/core/src/tests/central-core.test.ts
+++ b/packages/core/src/tests/central-core.test.ts
@@ -1392,6 +1392,116 @@ describe("CentralCore", () => {
      expect(remote?.nodeType).toBe("remote");
    });

+    describe("mesh outage persistence", () => {
+      it("persists and reloads mesh snapshot records", async () => {
+        await central.recordMeshSnapshot({
+          nodeId: "node-a",
+          projectId: "proj-1",
+          scope: "mesh.state",
+          payload: { value: 1 },
+          snapshotVersion: "a".repeat(64),
+          capturedAt: "2026-05-10T00:00:00.000Z",
+          sourceNodeId: "node-b",
+        });
+
+        const loaded = await central.getLatestMeshSnapshot({ nodeId: "node-a", projectId: "proj-1", scope: "mesh.state" });
+        expect(loaded?.payload).toEqual({ value: 1 });
+        expect(loaded?.snapshotVersion).toBe("a".repeat(64));
+        expect(loaded?.sourceNodeId).toBe("node-b");
+      });
+
+      it("supports queue lifecycle transitions and filters", async () => {
+        const entry = await central.enqueueMeshWrite({
+          originNodeId: "origin-1",
+          targetNodeId: "target-1",
+          projectId: "proj-1",
+          scope: "mesh.settings",
+          entityType: "project-settings",
+          entityId: "settings",
+          operation: "upsert",
+          payload: { ok: true },
+          intentVersion: "v1",
+        });
+
+        expect(entry.status).toBe("pending");
+
+        const replaying = await central.markMeshWriteReplayStarted(entry.id);
+        expect(replaying.status).toBe("replaying");
+        expect(replaying.attemptCount).toBe(1);
+
+        const failed = await central.markMeshWriteFailed(entry.id, { lastError: "timeout" });
+        expect(failed.status).toBe("failed");
+        expect(failed.lastError).toBe("timeout");
+
+        const failedRows = await central.listPendingMeshWrites({ targetNodeId: "target-1", status: "failed" });
+        expect(failedRows.map((row) => row.id)).toContain(entry.id);
+
+        const applied = await central.markMeshWriteApplied(entry.id, {});
+        expect(applied.status).toBe("applied");
+        expect(applied.appliedAt).toBeTruthy();
+      });
+
+      it("computes degraded read state from durable snapshot and queue", async () => {
+        await central.recordMeshSnapshot({
+          nodeId: "node-degraded",
+          scope: "mesh.tasks",
+          payload: { tasks: [] },
+          snapshotVersion: "b".repeat(64),
+          capturedAt: new Date(Date.now() - 5_000).toISOString(),
+          sourceNodeId: "node-source",
+        });
+
+        await central.enqueueMeshWrite({
+          originNodeId: "origin-2",
+          targetNodeId: "target-2",
+          scope: "mesh.tasks",
+          entityType: "task",
+          entityId: "T-1",
+          operation: "create",
+          payload: { id: "T-1" },
+          intentVersion: "v1",
+        });
+
+        const state = await central.getMeshDegradedReadState({ nodeId: "node-degraded", scope: "mesh.tasks" });
+        expect(state.mode).toBe("degraded");
+        expect(state.sourceNodeId).toBe("node-source");
+        expect(state.snapshotVersion).toBe("b".repeat(64));
+        expect(state.stalenessMs).toBeGreaterThanOrEqual(0);
+        expect(state.queueDepth).toBeGreaterThanOrEqual(1);
+      });
+
+      it("keeps queue and snapshots across close and re-init", async () => {
+        const initial = new CentralCore(tempDir);
+        await initial.init();
+        await initial.recordMeshSnapshot({
+          nodeId: "node-restart",
+          scope: "mesh.restart",
+          payload: { restart: true },
+          snapshotVersion: "c".repeat(64),
+          capturedAt: "2026-05-10T00:00:00.000Z",
+        });
+        const queued = await initial.enqueueMeshWrite({
+          originNodeId: "origin-restart",
+          targetNodeId: "target-restart",
+          scope: "mesh.restart",
+          entityType: "task",
+          entityId: "R-1",
+          operation: "update",
+          payload: { id: "R-1" },
+          intentVersion: "v1",
+        });
+        await initial.close();
+
+        const restarted = new CentralCore(tempDir);
+        await restarted.init();
+        const snapshot = await restarted.getLatestMeshSnapshot({ nodeId: "node-restart", scope: "mesh.restart" });
+        const queue = await restarted.listPendingMeshWrites({ targetNodeId: "target-restart" });
+        expect(snapshot?.payload).toEqual({ restart: true });
+        expect(queue.map((row) => row.id)).toContain(queued.id);
+        await restarted.close();
+      });
+    });
+
    describe("peer exchange methods", () => {
      it("should register a gossip peer and preserve its nodeId", async () => {
        const peerInfo = {
--- a/packages/core/src/tests/central-db.test.ts
+++ b/packages/core/src/tests/central-db.test.ts
@@ -39,7 +39,7 @@ describe("CentralDatabase", () => {

    it("should initialize schema version", () => {
      db.init();
-      expect(db.getSchemaVersion()).toBe(9);
+      expect(db.getSchemaVersion()).toBe(10);
    });

    it("should seed lastModified on init", () => {
@@ -95,6 +95,8 @@ describe("CentralDatabase", () => {
      expect(tableNames).toContain("nodes");
      expect(tableNames).toContain("peerNodes");
      expect(tableNames).toContain("projectNodePathMappings");
+      expect(tableNames).toContain("meshSharedSnapshots");
+      expect(tableNames).toContain("meshWriteQueue");
      expect(tableNames).toContain("__meta");
    });

@@ -217,7 +219,7 @@ describe("CentralDatabase", () => {

      db.init();

-      expect(db.getSchemaVersion()).toBe(9);
+      expect(db.getSchemaVersion()).toBe(10);

      const nodeColumns = db.prepare("PRAGMA table_info(nodes)").all() as Array<{ name: string }>;
      const nodeColumnNames = nodeColumns.map((column) => column.name);
@@ -282,7 +284,7 @@ describe("CentralDatabase", () => {

      db.init();

-      expect(db.getSchemaVersion()).toBe(9);
+      expect(db.getSchemaVersion()).toBe(10);

      const nodeColumns = db.prepare("PRAGMA table_info(nodes)").all() as Array<{ name: string }>;
      const nodeColumnNames = nodeColumns.map((column) => column.name);
@@ -370,7 +372,7 @@ describe("CentralDatabase", () => {

      db.init();

-      expect(db.getSchemaVersion()).toBe(9);
+      expect(db.getSchemaVersion()).toBe(10);

      const nodeColumns = db.prepare("PRAGMA table_info(nodes)").all() as Array<{ name: string }>;
      expect(nodeColumns.map((column) => column.name)).toContain("dockerConfig");
@@ -520,7 +522,7 @@ describe("CentralDatabase", () => {

      db.init();

-      expect(db.getSchemaVersion()).toBe(9);
+      expect(db.getSchemaVersion()).toBe(10);

      const mappings = db
        .prepare("SELECT projectId, nodeId, path FROM projectNodePathMappings ORDER BY projectId")
@@ -531,6 +533,30 @@ describe("CentralDatabase", () => {
        { projectId: "proj_2", nodeId: "node_local", path: "/tmp/proj-2" },
      ]);
    });
+
+    it("should migrate from v9 to v10 with mesh outage tables", () => {
+      db.exec(`
+        CREATE TABLE IF NOT EXISTS __meta (key TEXT PRIMARY KEY, value TEXT);
+        CREATE TABLE IF NOT EXISTS plugin_installs (id TEXT PRIMARY KEY, name TEXT NOT NULL, version TEXT NOT NULL, path TEXT NOT NULL, createdAt TEXT NOT NULL, updatedAt TEXT NOT NULL);
+        CREATE TABLE IF NOT EXISTS project_plugin_states (projectPath TEXT NOT NULL, pluginId TEXT NOT NULL, enabled INTEGER NOT NULL DEFAULT 0, state TEXT NOT NULL DEFAULT 'installed', createdAt TEXT NOT NULL, updatedAt TEXT NOT NULL, PRIMARY KEY (projectPath, pluginId));
+      `);
+      db.prepare("INSERT INTO __meta (key, value) VALUES ('schemaVersion', '9')").run();
+      db.prepare("INSERT INTO __meta (key, value) VALUES ('lastModified', ?)").run(String(Date.now()));
+
+      db.init();
+
+      expect(db.getSchemaVersion()).toBe(10);
+
+      const snapshotCols = db.prepare("PRAGMA table_info(meshSharedSnapshots)").all() as Array<{ name: string }>;
+      expect(snapshotCols.map((c) => c.name)).toEqual(
+        expect.arrayContaining(["nodeId", "projectId", "scope", "payload", "snapshotVersion", "capturedAt", "sourceNodeId", "sourceRunId", "staleAfter", "updatedAt"]),
+      );
+
+      const queueCols = db.prepare("PRAGMA table_info(meshWriteQueue)").all() as Array<{ name: string }>;
+      expect(queueCols.map((c) => c.name)).toEqual(
+        expect.arrayContaining(["id", "originNodeId", "targetNodeId", "projectId", "scope", "entityType", "entityId", "operation", "payload", "intentVersion", "status", "attemptCount", "lastAttemptAt", "lastError", "createdAt", "updatedAt", "appliedAt"]),
+      );
+    });
  });

  describe("transactions", () => {
--- a/packages/core/src/tests/mesh-replication-protocol.test.ts
+++ b/packages/core/src/tests/mesh-replication-protocol.test.ts
@@ -3,6 +3,7 @@ import {
  SHARED_MESH_PROTOCOL_ID,
  SHARED_MESH_PROTOCOL_VERSION,
  classifyReadStaleness,
+  compareMeshWriteReplayOrder,
  createFenceToken,
  getCoordinationModeForEntity,
  getDefaultWriteClassForEntity,
@@ -11,6 +12,7 @@ import {
  isMeshWriteClass,
  isProtocolRef,
  isQuorumSatisfied,
+  isRetryableMeshWriteFailure,
 } from "../mesh-replication-protocol.js";

 describe("mesh-replication-protocol", () => {
@@ -57,6 +59,23 @@ describe("mesh-replication-protocol", () => {
    expect(isProtocolRef({ protocol: "fusion.shared-mesh", version: "2.0" })).toBe(false);
  });

+  it("orders replay rows by createdAt then id", () => {
+    const a = { createdAt: "2026-05-10T00:00:00.000Z", id: "b" };
+    const b = { createdAt: "2026-05-10T00:00:00.000Z", id: "a" };
+    const c = { createdAt: "2026-05-10T00:00:01.000Z", id: "a" };
+
+    expect(compareMeshWriteReplayOrder(a, b)).toBeGreaterThan(0);
+    expect(compareMeshWriteReplayOrder(b, a)).toBeLessThan(0);
+    expect(compareMeshWriteReplayOrder(a, c)).toBeLessThan(0);
+  });
+
+  it("classifies retryable mesh failures", () => {
+    expect(isRetryableMeshWriteFailure(503)).toBe(true);
+    expect(isRetryableMeshWriteFailure(409)).toBe(false);
+    expect(isRetryableMeshWriteFailure(undefined, "TypeError: fetch failed ECONNREFUSED")).toBe(true);
+    expect(isRetryableMeshWriteFailure(undefined, "validation error 422")).toBe(false);
+  });
+
  it("classifies read staleness from queue depth and lag", () => {
    const fresh = classifyReadStaleness({ queueDepth: 0, observedAt: "2026-05-05T00:00:10.000Z", lastGlobalCommitAt: "2026-05-05T00:00:10.000Z" });
    expect(fresh.isStale).toBe(false);
--- a/packages/core/src/central-core.ts
+++ b/packages/core/src/central-core.ts
@@ -68,6 +68,16 @@ import type {
  ProjectNodePathMapping,
  ProjectNodePathMappingUpsertInput,
  ProjectNodePathMappingDeleteInput,
+  MeshSnapshotQuery,
+  MeshSnapshotRecord,
+  MeshSnapshotRecordInput,
+  MeshWriteQueueEntry,
+  MeshWriteQueueFilter,
+  MeshWriteQueueInput,
+  MeshWriteApplyResult,
+  MeshWriteFailureResult,
+  MeshDegradedReadState,
+  MeshWriteReplaySummary,
 } from "./types.js";
 import { getAppVersion, parseSemver } from "./app-version.js";
 import { validateDockerNodeConfig } from "./types.js";
@@ -1400,6 +1410,168 @@ export class CentralCore extends EventEmitter<CentralCoreEvents> {
    return snapshots;
  }

+  async recordMeshSnapshot(input: MeshSnapshotRecordInput): Promise<MeshSnapshotRecord> {
+    this.ensureInitialized();
+    const now = new Date().toISOString();
+    this.db!.prepare(
+      `INSERT INTO meshSharedSnapshots (nodeId, projectId, scope, payload, snapshotVersion, capturedAt, sourceNodeId, sourceRunId, staleAfter, updatedAt)
+       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+       ON CONFLICT(nodeId, projectId, scope) DO UPDATE SET
+         payload = excluded.payload,
+         snapshotVersion = excluded.snapshotVersion,
+         capturedAt = excluded.capturedAt,
+         sourceNodeId = excluded.sourceNodeId,
+         sourceRunId = excluded.sourceRunId,
+         staleAfter = excluded.staleAfter,
+         updatedAt = excluded.updatedAt`
+    ).run(
+      input.nodeId,
+      input.projectId ?? null,
+      input.scope,
+      JSON.stringify(input.payload),
+      input.snapshotVersion,
+      input.capturedAt,
+      input.sourceNodeId ?? null,
+      input.sourceRunId ?? null,
+      input.staleAfter ?? null,
+      now,
+    );
+    this.db!.bumpLastModified();
+    return { ...input, projectId: input.projectId ?? null, sourceNodeId: input.sourceNodeId ?? null, sourceRunId: input.sourceRunId ?? null, staleAfter: input.staleAfter ?? null, updatedAt: now };
+  }
+
+  async getLatestMeshSnapshot(query: MeshSnapshotQuery): Promise<MeshSnapshotRecord | null> {
+    this.ensureInitialized();
+    const row = this.db!.prepare(
+      `SELECT * FROM meshSharedSnapshots WHERE nodeId = ? AND projectId IS ? AND scope = ?`
+    ).get(query.nodeId, query.projectId ?? null, query.scope) as {
+      nodeId: string; projectId: string | null; scope: string; payload: string; snapshotVersion: string; capturedAt: string; sourceNodeId: string | null; sourceRunId: string | null; staleAfter: string | null; updatedAt: string;
+    } | undefined;
+    if (!row) return null;
+    return {
+      nodeId: row.nodeId,
+      projectId: row.projectId,
+      scope: row.scope,
+      payload: fromJson<Record<string, unknown>>(row.payload) ?? {},
+      snapshotVersion: row.snapshotVersion,
+      capturedAt: row.capturedAt,
+      sourceNodeId: row.sourceNodeId,
+      sourceRunId: row.sourceRunId,
+      staleAfter: row.staleAfter,
+      updatedAt: row.updatedAt,
+    };
+  }
+
+  async enqueueMeshWrite(input: MeshWriteQueueInput): Promise<MeshWriteQueueEntry> {
+    this.ensureInitialized();
+    const now = new Date().toISOString();
+    const id = `mq_${randomUUID().replace(/-/g, "").slice(0, 24)}`;
+    this.db!.prepare(
+      `INSERT INTO meshWriteQueue (id, originNodeId, targetNodeId, projectId, scope, entityType, entityId, operation, payload, intentVersion, status, attemptCount, createdAt, updatedAt)
+       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', 0, ?, ?)`
+    ).run(
+      id,
+      input.originNodeId,
+      input.targetNodeId,
+      input.projectId ?? null,
+      input.scope,
+      input.entityType,
+      input.entityId,
+      input.operation,
+      JSON.stringify(input.payload),
+      input.intentVersion,
+      now,
+      now,
+    );
+    this.db!.bumpLastModified();
+    return (await this.listPendingMeshWrites({ targetNodeId: input.targetNodeId })).find((entry) => entry.id === id)!;
+  }
+
+  async listPendingMeshWrites(filter: MeshWriteQueueFilter = {}): Promise<MeshWriteQueueEntry[]> {
+    this.ensureInitialized();
+    const conditions: string[] = [];
+    const values: Array<string> = [];
+    if (filter.originNodeId) { conditions.push("originNodeId = ?"); values.push(filter.originNodeId); }
+    if (filter.targetNodeId) { conditions.push("targetNodeId = ?"); values.push(filter.targetNodeId); }
+    if (filter.status) { conditions.push("status = ?"); values.push(filter.status); }
+    const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(" AND ")}` : "";
+    const rows = this.db!.prepare(
+      `SELECT * FROM meshWriteQueue ${whereClause} ORDER BY createdAt ASC, id ASC`
+    ).all(...values) as Array<{ id: string; originNodeId: string; targetNodeId: string; projectId: string | null; scope: string; entityType: string; entityId: string; operation: string; payload: string; intentVersion: string; status: MeshWriteQueueEntry["status"]; attemptCount: number; lastAttemptAt: string | null; lastError: string | null; createdAt: string; updatedAt: string; appliedAt: string | null }>;
+    return rows.map((row) => ({ ...row, payload: fromJson<Record<string, unknown>>(row.payload) ?? {} }));
+  }
+
+  async markMeshWriteReplayStarted(id: string): Promise<MeshWriteQueueEntry> {
+    this.ensureInitialized();
+    const now = new Date().toISOString();
+    this.db!.prepare(
+      `UPDATE meshWriteQueue SET status = 'replaying', attemptCount = attemptCount + 1, lastAttemptAt = ?, updatedAt = ? WHERE id = ?`
+    ).run(now, now, id);
+    this.db!.bumpLastModified();
+    return this.getMeshWriteQueueEntryById(id);
+  }
+
+  async markMeshWriteApplied(id: string, result: MeshWriteApplyResult): Promise<MeshWriteQueueEntry> {
+    this.ensureInitialized();
+    const now = new Date().toISOString();
+    this.db!.prepare(
+      `UPDATE meshWriteQueue SET status = 'applied', appliedAt = ?, updatedAt = ? WHERE id = ?`
+    ).run(result.appliedAt ?? now, now, id);
+    this.db!.bumpLastModified();
+    return this.getMeshWriteQueueEntryById(id);
+  }
+
+  async markMeshWriteFailed(id: string, result: MeshWriteFailureResult): Promise<MeshWriteQueueEntry> {
+    this.ensureInitialized();
+    const now = new Date().toISOString();
+    this.db!.prepare(
+      `UPDATE meshWriteQueue SET status = 'failed', lastError = ?, updatedAt = ? WHERE id = ?`
+    ).run(result.lastError, now, id);
+    this.db!.bumpLastModified();
+    return this.getMeshWriteQueueEntryById(id);
+  }
+
+  async getMeshDegradedReadState(query: MeshSnapshotQuery): Promise<MeshDegradedReadState> {
+    this.ensureInitialized();
+    const snapshot = await this.getLatestMeshSnapshot(query);
+    const now = Date.now();
+    const counts = this.db!.prepare(
+      `SELECT
+        SUM(CASE WHEN status IN ('pending','replaying','failed') THEN 1 ELSE 0 END) AS queueDepth,
+        SUM(CASE WHEN status = 'pending' THEN 1 ELSE 0 END) AS pendingWriteCount,
+        SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) AS failedWriteCount
+       FROM meshWriteQueue`
+    ).get() as { queueDepth: number | null; pendingWriteCount: number | null; failedWriteCount: number | null };
+
+    const asOf = snapshot?.capturedAt ?? new Date(now).toISOString();
+    return {
+      mode: snapshot ? "degraded" : "fresh",
+      asOf,
+      sourceNodeId: snapshot?.sourceNodeId ?? null,
+      snapshotVersion: snapshot?.snapshotVersion ?? null,
+      stalenessMs: Math.max(0, now - Date.parse(asOf)),
+      queueDepth: counts.queueDepth ?? 0,
+      pendingWriteCount: counts.pendingWriteCount ?? 0,
+      failedWriteCount: counts.failedWriteCount ?? 0,
+    };
+  }
+
+  async replayPendingMeshWritesForNode(targetNodeId: string): Promise<MeshWriteReplaySummary> {
+    this.ensureInitialized();
+    const pending = await this.listPendingMeshWrites({ targetNodeId, status: "pending" });
+    return { replayed: pending.length, applied: 0, failed: 0, queuedWriteIds: pending.map((entry) => entry.id) };
+  }
+
+  private getMeshWriteQueueEntryById(id: string): MeshWriteQueueEntry {
+    const row = this.db!.prepare(`SELECT * FROM meshWriteQueue WHERE id = ?`).get(id) as
+      | { id: string; originNodeId: string; targetNodeId: string; projectId: string | null; scope: string; entityType: string; entityId: string; operation: string; payload: string; intentVersion: string; status: MeshWriteQueueEntry["status"]; attemptCount: number; lastAttemptAt: string | null; lastError: string | null; createdAt: string; updatedAt: string; appliedAt: string | null }
+      | undefined;
+    if (!row) {
+      throw new Error(`Mesh write queue entry not found: ${id}`);
+    }
+    return { ...row, payload: fromJson<Record<string, unknown>>(row.payload) ?? {} };
+  }
+
  /**
   * Collect a fresh local mesh state snapshot.
   */
--- a/packages/core/src/central-db.ts
+++ b/packages/core/src/central-db.ts
@@ -23,7 +23,7 @@ export { toJson, toJsonNullable, fromJson };

 // ── Schema Definition ───────────────────────────────────────────────────

-const CENTRAL_SCHEMA_VERSION = 9;
+const CENTRAL_SCHEMA_VERSION = 10;

 const CENTRAL_SCHEMA_SQL = `
 -- Projects table (project registry)
@@ -210,6 +210,44 @@ CREATE TABLE IF NOT EXISTS project_plugin_states (
 CREATE INDEX IF NOT EXISTS idxProjectPluginStatesProjectPath ON project_plugin_states(projectPath);
 CREATE INDEX IF NOT EXISTS idxProjectPluginStatesPluginId ON project_plugin_states(pluginId);

+-- Durable mesh shared-state snapshots
+CREATE TABLE IF NOT EXISTS meshSharedSnapshots (
+  nodeId TEXT NOT NULL,
+  projectId TEXT,
+  scope TEXT NOT NULL,
+  payload TEXT NOT NULL,
+  snapshotVersion TEXT NOT NULL,
+  capturedAt TEXT NOT NULL,
+  sourceNodeId TEXT,
+  sourceRunId TEXT,
+  staleAfter TEXT,
+  updatedAt TEXT NOT NULL,
+  PRIMARY KEY (nodeId, projectId, scope)
+);
+CREATE INDEX IF NOT EXISTS idxMeshSharedSnapshotsLookup ON meshSharedSnapshots(nodeId, projectId, scope);
+
+-- Durable offline write queue + history
+CREATE TABLE IF NOT EXISTS meshWriteQueue (
+  id TEXT PRIMARY KEY,
+  originNodeId TEXT NOT NULL,
+  targetNodeId TEXT NOT NULL,
+  projectId TEXT,
+  scope TEXT NOT NULL,
+  entityType TEXT NOT NULL,
+  entityId TEXT NOT NULL,
+  operation TEXT NOT NULL,
+  payload TEXT NOT NULL,
+  intentVersion TEXT NOT NULL,
+  status TEXT NOT NULL CHECK (status IN ('pending', 'replaying', 'applied', 'failed')),
+  attemptCount INTEGER NOT NULL DEFAULT 0,
+  lastAttemptAt TEXT,
+  lastError TEXT,
+  createdAt TEXT NOT NULL,
+  updatedAt TEXT NOT NULL,
+  appliedAt TEXT
+);
+CREATE INDEX IF NOT EXISTS idxMeshWriteQueueReplay ON meshWriteQueue(targetNodeId, status, createdAt, id);
+
 -- Schema version tracking
 CREATE TABLE IF NOT EXISTS __meta (
  key TEXT PRIMARY KEY,
@@ -351,6 +389,44 @@ CREATE INDEX IF NOT EXISTS idxProjectPluginStatesProjectPath ON project_plugin_s
 CREATE INDEX IF NOT EXISTS idxProjectPluginStatesPluginId ON project_plugin_states(pluginId);
 `;

+const CENTRAL_SCHEMA_V10_MIGRATION_SQL = `
+CREATE TABLE IF NOT EXISTS meshSharedSnapshots (
+  nodeId TEXT NOT NULL,
+  projectId TEXT,
+  scope TEXT NOT NULL,
+  payload TEXT NOT NULL,
+  snapshotVersion TEXT NOT NULL,
+  capturedAt TEXT NOT NULL,
+  sourceNodeId TEXT,
+  sourceRunId TEXT,
+  staleAfter TEXT,
+  updatedAt TEXT NOT NULL,
+  PRIMARY KEY (nodeId, projectId, scope)
+);
+CREATE INDEX IF NOT EXISTS idxMeshSharedSnapshotsLookup ON meshSharedSnapshots(nodeId, projectId, scope);
+
+CREATE TABLE IF NOT EXISTS meshWriteQueue (
+  id TEXT PRIMARY KEY,
+  originNodeId TEXT NOT NULL,
+  targetNodeId TEXT NOT NULL,
+  projectId TEXT,
+  scope TEXT NOT NULL,
+  entityType TEXT NOT NULL,
+  entityId TEXT NOT NULL,
+  operation TEXT NOT NULL,
+  payload TEXT NOT NULL,
+  intentVersion TEXT NOT NULL,
+  status TEXT NOT NULL CHECK (status IN ('pending', 'replaying', 'applied', 'failed')),
+  attemptCount INTEGER NOT NULL DEFAULT 0,
+  lastAttemptAt TEXT,
+  lastError TEXT,
+  createdAt TEXT NOT NULL,
+  updatedAt TEXT NOT NULL,
+  appliedAt TEXT
+);
+CREATE INDEX IF NOT EXISTS idxMeshWriteQueueReplay ON meshWriteQueue(targetNodeId, status, createdAt, id);
+`;
+
 // ── Central Database Class ────────────────────────────────────────────────

 export class CentralDatabase {
@@ -478,6 +554,11 @@ export class CentralDatabase {
      migrated = true;
    }

+    if (currentVersion < 10) {
+      this.db.exec(CENTRAL_SCHEMA_V10_MIGRATION_SQL);
+      migrated = true;
+    }
+
    if (migrated) {
      this.db
        .prepare("INSERT INTO __meta (key, value) VALUES ('schemaVersion', ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value")
--- a/packages/core/src/index.ts
+++ b/packages/core/src/index.ts
@@ -455,6 +455,17 @@ export type {
  IsolationMode,
  MeshDiscovery,
  MeshClusterSnapshot,
+  MeshDegradedReadState,
+  MeshSnapshotQuery,
+  MeshSnapshotRecord,
+  MeshSnapshotRecordInput,
+  MeshWriteApplyResult,
+  MeshWriteFailureResult,
+  MeshWriteQueueEntry,
+  MeshWriteQueueFilter,
+  MeshWriteQueueInput,
+  MeshWriteQueueStatus,
+  MeshWriteReplaySummary,
  MigrationOptions,
  NodeConfig,
  NodeMeshState,
--- a/packages/core/src/mesh-replication-protocol.ts
+++ b/packages/core/src/mesh-replication-protocol.ts
@@ -1,4 +1,4 @@
-import type { NodeMeshState, PeerSyncRequest, PeerSyncResponse, SettingsSyncPayload } from "./types.js";
+import type { MeshWriteQueueEntry, NodeMeshState, PeerSyncRequest, PeerSyncResponse, SettingsSyncPayload } from "./types.js";

 export const SHARED_MESH_PROTOCOL_ID = "fusion.shared-mesh" as const;
 export const SHARED_MESH_PROTOCOL_VERSION = "1.0" as const;
@@ -159,6 +159,25 @@ export function isProtocolRef(value: { protocol?: string; version?: string } | n
  return value?.protocol === SHARED_MESH_PROTOCOL_ID && value.version === SHARED_MESH_PROTOCOL_VERSION;
 }

+const RETRYABLE_NODE_ERROR_CODES = ["ECONNREFUSED", "ENOTFOUND", "ETIMEDOUT", "ECONNRESET"] as const;
+
+export function isRetryableMeshWriteFailure(statusCode?: number, errorMessage?: string): boolean {
+  if (statusCode !== undefined) {
+    return statusCode === 502 || statusCode === 503 || statusCode === 504;
+  }
+  const message = (errorMessage ?? "").toLowerCase();
+  if (message.includes("timeout") || message.includes("abort")) return true;
+  if (message.includes("typeerror") || message.includes("network")) return true;
+  return RETRYABLE_NODE_ERROR_CODES.some((code) => message.includes(code.toLowerCase()));
+}
+
+export function compareMeshWriteReplayOrder(a: Pick<MeshWriteQueueEntry, "createdAt" | "id">, b: Pick<MeshWriteQueueEntry, "createdAt" | "id">): number {
+  if (a.createdAt === b.createdAt) {
+    return a.id.localeCompare(b.id);
+  }
+  return a.createdAt.localeCompare(b.createdAt);
+}
+
 export function classifyReadStaleness(params: {
  queueDepth: number;
  lastGlobalCommitAt?: string;
--- a/packages/core/src/types.ts
+++ b/packages/core/src/types.ts
@@ -2860,6 +2860,85 @@ export interface SnapshotBase {
  checksum: string;
 }

+export type MeshWriteQueueStatus = "pending" | "replaying" | "applied" | "failed";
+
+export interface MeshSnapshotQuery {
+  nodeId: string;
+  projectId?: string | null;
+  scope: string;
+}
+
+export interface MeshSnapshotRecordInput {
+  nodeId: string;
+  projectId?: string | null;
+  scope: string;
+  payload: Record<string, unknown>;
+  snapshotVersion: string;
+  capturedAt: string;
+  sourceNodeId?: string | null;
+  sourceRunId?: string | null;
+  staleAfter?: string | null;
+}
+
+export interface MeshSnapshotRecord extends MeshSnapshotRecordInput {
+  updatedAt: string;
+}
+
+export interface MeshWriteQueueInput {
+  originNodeId: string;
+  targetNodeId: string;
+  projectId?: string | null;
+  scope: string;
+  entityType: string;
+  entityId: string;
+  operation: string;
+  payload: Record<string, unknown>;
+  intentVersion: string;
+}
+
+export interface MeshWriteQueueFilter {
+  originNodeId?: string;
+  targetNodeId?: string;
+  status?: MeshWriteQueueStatus;
+}
+
+export interface MeshWriteQueueEntry extends MeshWriteQueueInput {
+  id: string;
+  status: MeshWriteQueueStatus;
+  attemptCount: number;
+  lastAttemptAt?: string | null;
+  lastError?: string | null;
+  createdAt: string;
+  updatedAt: string;
+  appliedAt?: string | null;
+}
+
+export interface MeshWriteApplyResult {
+  appliedAt?: string;
+}
+
+export interface MeshWriteFailureResult {
+  lastError: string;
+}
+
+export interface MeshWriteReplaySummary {
+  replayed: number;
+  applied: number;
+  failed: number;
+  queuedWriteIds: string[];
+}
+
+export interface MeshDegradedReadState {
+  mode: "fresh" | "degraded";
+  asOf: string;
+  sourceNodeId: string | null;
+  snapshotVersion: string | null;
+  stalenessMs: number;
+  queueDepth: number;
+  pendingWriteCount: number;
+  failedWriteCount: number;
+}
+
 export interface SharedMeshStatePayload {
  taskMetadata?: SnapshotBase & { payload: { tasks: Task[] } };
  missionHierarchy?: SnapshotBase & {