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

feat(FN-3464): wire shared state in peer exchange service (FN-3464 Step 1)

Browse Source 
Merges peer exchange shared state wiring (FN-3464) into the engine with updated tests, adds immediate wake controls for agent inbox and message API (FN-3087), and restores the narrow logs/system mouse auto-toggle policy with CLI documentation (FN-3708). Dependency graph plugin receives test and high

Fusion-Task-Id: FN-3464
This commit is contained in:
Fusion
2026-05-07 11:15:27 -07:00
committed by gsxdsm
parent 1b0f3de9f9
commit 29692dbbcc
4 changed files with 129 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
packages/engine/src/__tests__/peer-exchange-service.test.ts
View File

@@ -49,6 +49,9 @@ describe("PeerExchangeService", () => {
let mockReportMeshState: ReturnType<typeof vi.fn>;
let mockGetSettingsForSync: ReturnType<typeof vi.fn>;
let mockApplyRemoteSettings: ReturnType<typeof vi.fn>;
let mockGetProjectSettingsSnapshot: ReturnType<typeof vi.fn>;
let mockGetAuthMaterialSnapshot: ReturnType<typeof vi.fn>;
let mockApplyProjectSettingsSnapshot: ReturnType<typeof vi.fn>;
beforeEach(() => {
vi.useFakeTimers();
@@ -61,6 +64,9 @@ describe("PeerExchangeService", () => {
mockReportMeshState = vi.fn();
mockGetSettingsForSync = vi.fn();
mockApplyRemoteSettings = vi.fn();
mockGetProjectSettingsSnapshot = vi.fn();
mockGetAuthMaterialSnapshot = vi.fn();
mockApplyProjectSettingsSnapshot = vi.fn();
mockCentralCore = {
listNodes: mockListNodes,
@@ -69,6 +75,9 @@ describe("PeerExchangeService", () => {
reportMeshState: mockReportMeshState,
getSettingsForSync: mockGetSettingsForSync,
applyRemoteSettings: mockApplyRemoteSettings,
getProjectSettingsSnapshot: mockGetProjectSettingsSnapshot,
getAuthMaterialSnapshot: mockGetAuthMaterialSnapshot,
applyProjectSettingsSnapshot: mockApplyProjectSettingsSnapshot,
} as unknown as CentralCore;
mockFetch = vi.fn();
@@ -386,6 +395,8 @@ describe("PeerExchangeService", () => {
const service = new PeerExchangeService(mockCentralCore, { settingsSyncEnabled: true });
const payload = makeSettingsPayload({ checksum: "local-checksum-123" });
mockGetSettingsForSync.mockResolvedValue(payload);
mockGetProjectSettingsSnapshot.mockResolvedValue({ version: 1, exportedAt: payload.exportedAt, checksum: payload.checksum, payload: { global: {} } });
mockGetAuthMaterialSnapshot.mockReturnValue({ version: 1, exportedAt: payload.exportedAt, checksum: "auth-checksum", payload: { providerAuth: {} } });
setupSuccessfulSync();
await service.syncWithNode(makeNode());
@@ -395,6 +406,7 @@ describe("PeerExchangeService", () => {
const body = JSON.parse(call[1].body);
expect(body.settings).toBeDefined();
expect(body.settings.checksum).toBe("local-checksum-123");
expect(body.sharedState?.projectSettings?.checksum).toBe("local-checksum-123");
});
it("should include settings on first sync with a node (no throttle entry)", async () => {
@@ -537,7 +549,7 @@ describe("PeerExchangeService", () => {
const localPayload = makeSettingsPayload({ checksum: "local-checksum" });
const remotePayload = makeSettingsPayload({ checksum: "remote-checksum" });
mockGetSettingsForSync.mockResolvedValue(localPayload);
mockApplyRemoteSettings.mockResolvedValue({
mockApplyProjectSettingsSnapshot.mockResolvedValue({
success: true,
globalCount: 5,
projectCount: 2,
@@ -554,13 +566,16 @@ describe("PeerExchangeService", () => {
knownPeers: [],
newPeers: [],
timestamp: "2026-04-01T12:00:00.000Z",
settings: remotePayload,
sharedState: {
projectSettings: remotePayload,
authMaterial: { ...remotePayload, payload: { providerAuth: {} } },
},
}),
});
const result = await service.syncWithNode(makeNode());
expect(mockApplyRemoteSettings).toHaveBeenCalledWith(remotePayload);
expect(mockApplyProjectSettingsSnapshot).toHaveBeenCalledWith(remotePayload);
expect(result.settingsApplied).toBe(true);
expect(result.settingsVersion).toBe("remote-checksum");
});

97
packages/engine/src/peer-exchange-service.ts
View File

@@ -1,4 +1,9 @@
import type { CentralCore, GlobalSettings, SettingsSyncPayload } from "@fusion/core";
import type {
CentralCore,
GlobalSettings,
SettingsSyncPayload,
SharedMeshStatePayload,
} from "@fusion/core";
import type { NodeConfig, PeerSyncRequest, PeerSyncResponse } from "@fusion/core";
import { peerExchangeLog } from "./logger.js";
@@ -57,6 +62,8 @@ export class PeerExchangeService {
private lastSettingsSyncByNode = new Map<string, { version: string; timestamp: number }>();
/** Cached settings payload from the last successful getSettingsForSync call. */
private cachedSettingsPayload: SettingsSyncPayload | null = null;
/** Cached shared-state settings/auth payload built from canonical snapshots. */
private cachedSharedStatePayload: SharedMeshStatePayload | null = null;
/** Global settings provided via options. */
private globalSettings?: GlobalSettings;
/** Provider auth credentials provided via options. */
@@ -87,6 +94,7 @@ export class PeerExchangeService {
this.globalSettings = settings;
// Invalidate cache to ensure fresh payload on next sync
this.cachedSettingsPayload = null;
this.cachedSharedStatePayload = null;
}
/**
@@ -309,6 +317,7 @@ export class PeerExchangeService {
if (shouldIncludeSettings) {
request.settings = this.cachedSettingsPayload;
request.sharedState = await this.getSharedStateSettingsBundle();
}
} catch (err) {
// Log error but continue with peer sync
@@ -359,45 +368,46 @@ export class PeerExchangeService {
const mergeResult = await this.centralCore.mergePeers(peerResponse.knownPeers);
// ── Process remote settings if included in response ──
if (peerResponse.settings && this.settingsSyncEnabled) {
settingsVersion = peerResponse.settings.checksum;
if (this.settingsSyncEnabled && (peerResponse.sharedState || peerResponse.settings)) {
const remoteChecksum =
peerResponse.sharedState?.projectSettings?.checksum ??
peerResponse.settings?.checksum;
settingsVersion = remoteChecksum;
// Check if we should apply remote settings
// Apply if remote checksum is different from our cached checksum
const localChecksum = this.cachedSettingsPayload?.checksum ?? "";
if (peerResponse.settings.checksum !== localChecksum) {
if (remoteChecksum && remoteChecksum !== localChecksum) {
try {
const applyResult = await this.centralCore.applyRemoteSettings(peerResponse.settings);
const applyResult = await this.applyRemoteSharedState(peerResponse.sharedState, peerResponse.settings);
if (applyResult.success) {
settingsApplied = true;
peerExchangeLog.log(
`Applied remote settings from ${node.name} (version: ${peerResponse.settings.checksum}, ` +
`global: ${applyResult.globalCount}, projects: ${applyResult.projectCount}, auth: ${applyResult.authCount})`
`Applied remote settings from ${node.name} (version: ${remoteChecksum}, ` +
`global: ${applyResult.globalCount}, projects: ${applyResult.projectCount}, auth: ${applyResult.authCount})`,
);
// Invalidate cache to ensure fresh data on next sync
this.cachedSettingsPayload = null;
this.cachedSharedStatePayload = null;
} else {
peerExchangeLog.warn(
`Failed to apply remote settings from ${node.name}: ${applyResult.error}`
);
peerExchangeLog.warn(`Failed to apply remote settings from ${node.name}: ${applyResult.error}`);
}
} catch (err) {
const error = err instanceof Error ? err.message : String(err);
peerExchangeLog.warn(`Settings sync error with ${node.name}: ${error}`);
}
} else {
} else if (remoteChecksum) {
peerExchangeLog.log(
`Remote settings from ${node.name} are up-to-date (version: ${peerResponse.settings.checksum})`
`Remote settings from ${node.name} are up-to-date (version: ${remoteChecksum})`,
);
}
// Update throttle tracking
this.lastSettingsSyncByNode.set(node.id, {
version: peerResponse.settings.checksum,
timestamp: Date.now(),
});
if (remoteChecksum) {
this.lastSettingsSyncByNode.set(node.id, {
version: remoteChecksum,
timestamp: Date.now(),
});
}
}
peerExchangeLog.log(
@@ -431,4 +441,53 @@ export class PeerExchangeService {
return { nodeId: node.id, success: false, added: 0, updated: 0, error: message };
}
}
private async getSharedStateSettingsBundle(): Promise<SharedMeshStatePayload | undefined> {
if (this.cachedSharedStatePayload) {
return this.cachedSharedStatePayload;
}
const globalSettings = this.globalSettings ?? {};
const core = this.centralCore as CentralCore & {
getProjectSettingsSnapshot?: (settings: GlobalSettings) => Promise<SharedMeshStatePayload["projectSettings"]>;
getAuthMaterialSnapshot?: (
providerAuth?: Record<string, { type: "api_key" | "oauth"; key?: string; accessToken?: string; authenticated?: boolean }>,
) => SharedMeshStatePayload["authMaterial"];
};
if (!core.getProjectSettingsSnapshot || !core.getAuthMaterialSnapshot) {
return undefined;
}
const projectSettings = await core.getProjectSettingsSnapshot(globalSettings);
const authMaterial = core.getAuthMaterialSnapshot(this.providerAuth);
this.cachedSharedStatePayload = { projectSettings, authMaterial };
return this.cachedSharedStatePayload;
}
private async applyRemoteSharedState(
sharedState: SharedMeshStatePayload | undefined,
fallbackSettings: SettingsSyncPayload | undefined,
): Promise<{ success: boolean; globalCount: number; projectCount: number; authCount: number; error?: string }> {
const core = this.centralCore as CentralCore & {
applyProjectSettingsSnapshot?: (snapshot: NonNullable<SharedMeshStatePayload["projectSettings"]>) => Promise<{ success: boolean; globalCount: number; projectCount: number; authCount: number; error?: string }>;
applyAuthMaterialSnapshot?: (snapshot: NonNullable<SharedMeshStatePayload["authMaterial"]>) => { success?: boolean; authCount?: number; error?: string } | Record<string, unknown>;
};
if (sharedState?.projectSettings && core.applyProjectSettingsSnapshot) {
const result = await core.applyProjectSettingsSnapshot(sharedState.projectSettings);
if (sharedState.authMaterial && core.applyAuthMaterialSnapshot) {
const authResult = core.applyAuthMaterialSnapshot(sharedState.authMaterial);
const authCount =
typeof (authResult as { authCount?: number }).authCount === "number"
? (authResult as { authCount: number }).authCount
: Object.keys(sharedState.authMaterial.payload.providerAuth ?? {}).length;
return { ...result, authCount: Math.max(result.authCount, authCount) };
}
return result;
}
if (fallbackSettings) {
return this.centralCore.applyRemoteSettings(fallbackSettings);
}
return { success: true, globalCount: 0, projectCount: 0, authCount: 0 };
}
}