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

feat(FN-3451): add shared state snapshots for mesh sync

Browse Source 
This merge completes the mesh sync shared-state pipeline (FN-3451 Steps 1–3) by adding agent run snapshot limits, applying inbound snapshots, and returning shared state snapshots, while hardening agent store cache initialization during sync. It also unifies agent action gating classifications across

Fusion-Task-Id: FN-3451
This commit is contained in:
Fusion
2026-05-08 14:03:17 -07:00
committed by gsxdsm
parent 07cd3d065c
commit ae6fdf9329
5 changed files with 172 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
packages/core/src/__tests__/agent-store.test.ts
View File

@@ -13,6 +13,7 @@
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
import { AgentStore } from "../agent-store.js";
import { TaskStore } from "../store.js";
import { validateSnapshotEnvelope } from "../shared-mesh-state.js";
import { rm } from "node:fs/promises";
import { join } from "node:path";
import { mkdtempSync, existsSync, mkdirSync, writeFileSync } from "node:fs";
@@ -2885,8 +2886,17 @@ describe("AgentStore", () => {
const agent = await store.createAgent({ name: "Snapshot Agent", role: "executor" });
await store.setLastBlockedState(agent.id, { taskId: "FN-1", blockedBy: "dep", recordedAt: new Date().toISOString(), contextHash: "h" });
const run1 = await store.startHeartbeatRun(agent.id);
await store.endHeartbeatRun(run1.id, "completed");
const run2 = await store.startHeartbeatRun(agent.id);
await store.endHeartbeatRun(run2.id, "completed");
const agentSnapshot = await store.getAgentSnapshot();
const runSnapshot = store.getAgentRunSnapshot();
const limitedRunSnapshot = store.getAgentRunSnapshot(1);
validateSnapshotEnvelope(agentSnapshot);
validateSnapshotEnvelope(runSnapshot);
const applyAgent = await store.applyAgentSnapshot(agentSnapshot);
const applyRun = await store.applyAgentRunSnapshot(runSnapshot);
@@ -2894,8 +2904,12 @@ describe("AgentStore", () => {
const runSnapshot2 = store.getAgentRunSnapshot();
expect(applyAgent.appliedAgents).toBeGreaterThan(0);
expect(agentSnapshot.payload.agents.length).toBeGreaterThan(0);
expect(agentSnapshot.payload.blockedStates.length).toBe(1);
expect(agentSnapshot2.payload).toEqual(agentSnapshot.payload);
expect(runSnapshot2.payload).toEqual(runSnapshot.payload);
expect(limitedRunSnapshot.payload.runs).toHaveLength(1);
expect(limitedRunSnapshot.payload.runs[0]?.id).toBe(run2.id);
expect(applyRun.applied + applyRun.skipped).toBeGreaterThanOrEqual(0);
});
});

15
packages/core/src/agent-store.ts
View File

@@ -2189,12 +2189,19 @@ export class AgentStore extends EventEmitter {
return { appliedAgents, appliedBlockedStates };
}
getAgentRunSnapshot(): AgentRunSnapshot {
const runs = this.db.prepare("SELECT data FROM agentRuns ORDER BY startedAt ASC").all() as Array<{ data: string }>;
const parsed = runs
getAgentRunSnapshot(limit?: number): AgentRunSnapshot {
const normalizedLimit = typeof limit === "number" && Number.isFinite(limit) && limit > 0 ? Math.floor(limit) : undefined;
const query = normalizedLimit
? "SELECT data FROM agentRuns ORDER BY startedAt DESC LIMIT ?"
: "SELECT data FROM agentRuns ORDER BY startedAt ASC";
const rows = normalizedLimit
? (this.db.prepare(query).all(normalizedLimit) as Array<{ data: string }>)
: (this.db.prepare(query).all() as Array<{ data: string }>);
const parsed = rows
.map((row) => this.parseJson<AgentHeartbeatRun | null>(row.data, null))
.filter((run): run is AgentHeartbeatRun => run !== null);
return createAgentRunSnapshot(parsed);
const orderedRuns = normalizedLimit ? parsed.reverse() : parsed;
return createAgentRunSnapshot(orderedRuns);
}
async applyAgentRunSnapshot(snapshot: AgentRunSnapshot): Promise<{ applied: number; skipped: number }> {

4
packages/core/src/central-core.ts
View File

@@ -3120,11 +3120,11 @@ export class CentralCore extends EventEmitter<CentralCoreEvents> {
async applyProjectSettingsSnapshot(snapshot: ProjectSettingsSnapshot): Promise<SettingsSyncResult> {
validateSnapshotEnvelope(snapshot);
const payloadWithoutChecksum: Omit<SettingsSyncPayload, "checksum"> = {
version: 1,
exportedAt: snapshot.exportedAt,
global: snapshot.payload.global,
projects: snapshot.payload.projects,
providerAuth: undefined,
exportedAt: snapshot.exportedAt,
version: 1,
};
const checksum = createHash("sha256")
.update(JSON.stringify(payloadWithoutChecksum))

2
packages/core/src/shared-mesh-state.ts
View File

@@ -22,6 +22,8 @@ import type {
Slice,
} from "./mission-types.js";
export const SHARED_STATE_DEFAULT_LIMIT = 10_000;
export interface SharedSnapshotEnvelope<TPayload> {
version: number;
exportedAt: string;

143
packages/dashboard/src/routes/register-mesh-routes.ts
View File

@@ -379,6 +379,146 @@ export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => {
}
}
// ── Shared state sync: apply inbound domain snapshots independently ──
const { AgentStore, SHARED_STATE_DEFAULT_LIMIT, validateSnapshotEnvelope } = await import("@fusion/core");
const sharedState = req.body?.sharedState;
if (sharedState && typeof sharedState === "object") {
const missionStore = store.getMissionStore();
const fusionDir = store.getFusionDir();
let agentStore: InstanceType<typeof AgentStore> | null = null;
const ensureAgentStore = async (): Promise<InstanceType<typeof AgentStore>> => {
if (agentStore) return agentStore;
const newStore = new AgentStore({ rootDir: fusionDir, taskStore: store });
await newStore.init();
agentStore = newStore;
return agentStore;
};
const applyDomain = async (domain: string, fn: () => Promise<void> | void): Promise<void> => {
try {
await fn();
} catch (err) {
emitRemoteRouteDiagnostic({
route: "mesh-sync",
message: `Failed to apply shared state domain: ${domain}`,
nodeId: senderNodeId,
upstreamPath: "/api/mesh/sync",
operationStage: `apply-shared-state-${domain}`,
level: "warn",
error: err,
});
}
};
await applyDomain("task-metadata", async () => {
if (!sharedState.taskMetadata) return;
validateSnapshotEnvelope(sharedState.taskMetadata);
await store.applyTaskMetadataSnapshot(sharedState.taskMetadata as Parameters<typeof store.applyTaskMetadataSnapshot>[0]);
});
await applyDomain("mission-hierarchy", async () => {
if (!sharedState.missionHierarchy) return;
validateSnapshotEnvelope(sharedState.missionHierarchy);
missionStore.applyMissionHierarchySnapshot(sharedState.missionHierarchy as Parameters<typeof missionStore.applyMissionHierarchySnapshot>[0]);
});
await applyDomain("agents", async () => {
if (!sharedState.agents) return;
validateSnapshotEnvelope(sharedState.agents);
const activeAgentStore = await ensureAgentStore();
await activeAgentStore.applyAgentSnapshot(sharedState.agents as Parameters<typeof activeAgentStore.applyAgentSnapshot>[0]);
});
await applyDomain("agent-runs", async () => {
if (!sharedState.agentRuns) return;
validateSnapshotEnvelope(sharedState.agentRuns);
const activeAgentStore = await ensureAgentStore();
await activeAgentStore.applyAgentRunSnapshot(sharedState.agentRuns as Parameters<typeof activeAgentStore.applyAgentRunSnapshot>[0]);
});
await applyDomain("activity-log", async () => {
if (!sharedState.activityLog) return;
validateSnapshotEnvelope(sharedState.activityLog);
store.applyActivityLogSnapshot(sharedState.activityLog as Parameters<typeof store.applyActivityLogSnapshot>[0]);
});
await applyDomain("run-audit", async () => {
if (!sharedState.runAudit) return;
validateSnapshotEnvelope(sharedState.runAudit);
store.applyRunAuditSnapshot(sharedState.runAudit as Parameters<typeof store.applyRunAuditSnapshot>[0]);
});
await applyDomain("project-settings", async () => {
if (!sharedState.projectSettings) return;
validateSnapshotEnvelope(sharedState.projectSettings);
const result = await central.applyProjectSettingsSnapshot(sharedState.projectSettings as Parameters<typeof central.applyProjectSettingsSnapshot>[0]);
if (!result.success) {
throw new Error(result.error ?? "applyProjectSettingsSnapshot failed");
}
});
if (sharedState.authMaterial) {
emitRemoteRouteDiagnostic({
route: "mesh-sync",
message: "Ignoring sharedState.authMaterial; use dedicated auth sync routes (/api/nodes/:id/auth/*)",
nodeId: senderNodeId,
upstreamPath: "/api/mesh/sync",
operationStage: "apply-shared-state-auth-material",
level: "info",
});
}
// Intentionally do not close this per-request AgentStore wrapper.
// AgentStore uses a process-wide DB cache by rootDir; closing here would
// invalidate shared connections used by long-lived runtime stores.
}
// Build shared-state response from fresh local snapshots per request.
const responseSharedState: Record<string, unknown> = {};
const collectSnapshot = async (domain: string, fn: () => Promise<unknown>): Promise<void> => {
try {
const snapshot = await fn();
if (!snapshot) {
emitRemoteRouteDiagnostic({
route: "mesh-sync",
message: `No shared state snapshot available for domain: ${domain}`,
nodeId: senderNodeId,
upstreamPath: "/api/mesh/sync",
operationStage: `build-shared-state-${domain}`,
level: "info",
});
return;
}
responseSharedState[domain] = snapshot;
} catch (err) {
emitRemoteRouteDiagnostic({
route: "mesh-sync",
message: `Failed to build shared state snapshot for domain: ${domain}`,
nodeId: senderNodeId,
upstreamPath: "/api/mesh/sync",
operationStage: `build-shared-state-${domain}`,
level: "warn",
error: err,
});
}
};
await collectSnapshot("taskMetadata", async () => store.getTaskMetadataSnapshot());
await collectSnapshot("missionHierarchy", async () => store.getMissionStore().getMissionHierarchySnapshot());
await collectSnapshot("activityLog", async () => store.getActivityLogSnapshot(SHARED_STATE_DEFAULT_LIMIT));
await collectSnapshot("runAudit", async () => store.getRunAuditSnapshot({ limit: SHARED_STATE_DEFAULT_LIMIT }));
const responseAgentStore = new AgentStore({ rootDir: store.getFusionDir(), taskStore: store });
await responseAgentStore.init();
await collectSnapshot("agents", async () => responseAgentStore.getAgentSnapshot());
await collectSnapshot("agentRuns", async () => responseAgentStore.getAgentRunSnapshot(SHARED_STATE_DEFAULT_LIMIT));
await collectSnapshot("projectSettings", async () => {
const localGlobal = await store.getGlobalSettingsStore().getSettings();
return central.getProjectSettingsSnapshot(localGlobal);
});
await central.close();
// Return sync response
@@ -394,6 +534,9 @@ export const registerMeshRoutes: ApiRouteRegistrar = (ctx) => {
if (responseSettings) {
response.settings = responseSettings;
}
if (Object.keys(responseSharedState).length > 0) {
response.sharedState = responseSharedState;
}
res.json(response);
} catch (err: unknown) {