From fa552b0faaa800c76ec1bdd2633f172fa58dd45a Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 11 Apr 2026 16:39:09 +0700 Subject: [PATCH] Add mutable daemon config RPC support --- packages/server/src/client/daemon-client.ts | 44 ++++- .../server/src/server/daemon-config-store.ts | 165 ++++++++++++++++++ packages/server/src/server/session.ts | 25 +++ .../websocket-server.notifications.test.ts | 4 + .../websocket-server.relay-reconnect.test.ts | 4 + .../server/src/server/websocket-server.ts | 25 +++ packages/server/src/shared/messages.ts | 66 +++++++ 7 files changed, 331 insertions(+), 2 deletions(-) create mode 100644 packages/server/src/server/daemon-config-store.ts diff --git a/packages/server/src/client/daemon-client.ts b/packages/server/src/client/daemon-client.ts index fdee4a5b8..894af94c3 100644 --- a/packages/server/src/client/daemon-client.ts +++ b/packages/server/src/client/daemon-client.ts @@ -70,6 +70,10 @@ import type { AgentProvider, AgentSessionConfig, } from "../server/agent/agent-sdk-types.js"; +import type { + MutableDaemonConfig, + MutableDaemonConfigPatch, +} from "../shared/messages.js"; import { getAgentProviderDefinition } from "../server/agent/provider-manifest.js"; import { isRelayClientWebSocketUrl } from "../shared/daemon-endpoints.js"; import { @@ -516,10 +520,18 @@ type WaitHandle = { }; type RpcWaitResult = { kind: "ok"; value: T } | { kind: "error"; error: DaemonRpcError }; -type CorrelatedResponseMessage = Extract< +type GetDaemonConfigResponse = Extract< SessionOutboundMessage, - { payload: { requestId: string } } + { type: "get_daemon_config_response" } >; +type SetDaemonConfigResponse = Extract< + SessionOutboundMessage, + { type: "set_daemon_config_response" } +>; +type CorrelatedResponseMessage = + | Extract + | GetDaemonConfigResponse + | SetDaemonConfigResponse; type CorrelatedResponseType = CorrelatedResponseMessage["type"]; type CorrelatedResponsePayload = Extract< CorrelatedResponseMessage, @@ -2706,6 +2718,34 @@ export class DaemonClient { }); } + async getDaemonConfig( + requestId?: string, + ): Promise<{ requestId: string; config: MutableDaemonConfig }> { + return this.sendCorrelatedSessionRequest({ + requestId, + message: { + type: "get_daemon_config_request", + }, + responseType: "get_daemon_config_response", + timeout: 10000, + }); + } + + async patchDaemonConfig( + config: MutableDaemonConfigPatch, + requestId?: string, + ): Promise<{ requestId: string; config: MutableDaemonConfig }> { + return this.sendCorrelatedSessionRequest({ + requestId, + message: { + type: "set_daemon_config_request", + config, + }, + responseType: "set_daemon_config_response", + timeout: 10000, + }); + } + async refreshProvidersSnapshot(options?: { cwd?: string; requestId?: string; diff --git a/packages/server/src/server/daemon-config-store.ts b/packages/server/src/server/daemon-config-store.ts new file mode 100644 index 000000000..7719ff49a --- /dev/null +++ b/packages/server/src/server/daemon-config-store.ts @@ -0,0 +1,165 @@ +import { + loadPersistedConfig, + savePersistedConfig, + type PersistedConfig, +} from "./persisted-config.js"; +import { + MutableDaemonConfigSchema, + MutableDaemonConfigPatchSchema, +} from "../shared/messages.js"; + +export type { MutableDaemonConfig, MutableDaemonConfigPatch } from "../shared/messages.js"; + +type MutableDaemonConfig = import("../shared/messages.js").MutableDaemonConfig; +type MutableDaemonConfigPatch = import("../shared/messages.js").MutableDaemonConfigPatch; + +type LoggerLike = { + child(bindings: Record): LoggerLike; + info(...args: any[]): void; +}; + +type ConfigListener = (config: MutableDaemonConfig) => void; +type FieldChangeHandler = (value: unknown) => void; + +function getLogger(logger: LoggerLike | undefined): LoggerLike | undefined { + return logger?.child({ module: "daemon-config-store" }); +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function deepMerge>(current: T, patch: Record): T { + const next: Record = { ...current }; + + for (const [key, patchValue] of Object.entries(patch)) { + if (patchValue === undefined) { + continue; + } + const currentValue = next[key]; + if (isRecord(currentValue) && isRecord(patchValue)) { + next[key] = deepMerge(currentValue, patchValue); + continue; + } + next[key] = patchValue; + } + + return next as T; +} + +function getValueAtPath(config: MutableDaemonConfig, path: string): unknown { + return path + .split(".") + .reduce((value, segment) => (isRecord(value) ? value[segment] : undefined), config); +} + +function isEqualValue(a: unknown, b: unknown): boolean { + return JSON.stringify(a) === JSON.stringify(b); +} + +export class DaemonConfigStore { + private current: MutableDaemonConfig; + private readonly paseoHome: string; + private readonly logger: LoggerLike | undefined; + private readonly changeListeners = new Set(); + private readonly fieldChangeHandlers = new Map>(); + + constructor( + paseoHome: string, + initial: MutableDaemonConfig, + logger?: LoggerLike, + ) { + this.paseoHome = paseoHome; + this.logger = getLogger(logger); + this.current = MutableDaemonConfigSchema.parse(initial); + } + + public get(): MutableDaemonConfig { + return this.current; + } + + public patch(partial: MutableDaemonConfigPatch): MutableDaemonConfig { + const parsedPatch = MutableDaemonConfigPatchSchema.parse(partial); + const next = MutableDaemonConfigSchema.parse(deepMerge(this.current, parsedPatch)); + + const changedFieldPaths = Array.from(this.fieldChangeHandlers.keys()).filter((path) => { + return !isEqualValue(getValueAtPath(this.current, path), getValueAtPath(next, path)); + }); + + if (changedFieldPaths.length === 0 && isEqualValue(this.current, next)) { + return this.current; + } + + // Persist before updating in-memory state so that if persistence fails, + // runtime and disk stay consistent. + this.persistConfig(next); + this.current = next; + + for (const path of changedFieldPaths) { + const handlers = this.fieldChangeHandlers.get(path); + if (!handlers) { + continue; + } + const value = getValueAtPath(next, path); + for (const handler of handlers) { + handler(value); + } + } + + for (const listener of this.changeListeners) { + listener(next); + } + + return next; + } + + public onFieldChange(path: string, handler: FieldChangeHandler): () => void { + const handlers = this.fieldChangeHandlers.get(path) ?? new Set(); + handlers.add(handler); + this.fieldChangeHandlers.set(path, handlers); + + return () => { + const currentHandlers = this.fieldChangeHandlers.get(path); + if (!currentHandlers) { + return; + } + currentHandlers.delete(handler); + if (currentHandlers.size === 0) { + this.fieldChangeHandlers.delete(path); + } + }; + } + + public onChange(listener: ConfigListener): () => void { + this.changeListeners.add(listener); + return () => { + this.changeListeners.delete(listener); + }; + } + + private persistConfig(config: MutableDaemonConfig): void { + const persisted = loadPersistedConfig(this.paseoHome, this.logger); + const nextPersisted = mergeMutableConfigIntoPersistedConfig({ + persisted, + mutable: config, + }); + savePersistedConfig(this.paseoHome, nextPersisted, this.logger); + } +} + +function mergeMutableConfigIntoPersistedConfig(params: { + persisted: PersistedConfig; + mutable: MutableDaemonConfig; +}): PersistedConfig { + const { persisted, mutable } = params; + return { + ...persisted, + daemon: { + ...persisted.daemon, + mcp: { + ...persisted.daemon?.mcp, + injectIntoAgents: mutable.mcp.injectIntoAgents, + }, + }, + }; +} diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 07ca2a656..57f81427a 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -68,6 +68,7 @@ import { experimental_createMCPClient } from "ai"; import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; import type { VoiceCallerContext, VoiceSpeakHandler } from "./voice-types.js"; import { BackgroundGitFetchManager } from "./background-git-fetch-manager.js"; +import type { DaemonConfigStore } from "./daemon-config-store.js"; import { buildProviderRegistry } from "./agent/provider-registry.js"; import type { AgentProviderRuntimeSettingsMap } from "./agent/provider-launch-config.js"; @@ -415,6 +416,7 @@ export type SessionOptions = { loopService: LoopService; checkoutDiffManager: CheckoutDiffManager; backgroundGitFetchManager: BackgroundGitFetchManager; + daemonConfigStore: DaemonConfigStore; mcpBaseUrl?: string | null; stt: Resolvable; tts: Resolvable; @@ -597,6 +599,7 @@ export class Session { private readonly loopService: LoopService; private readonly checkoutDiffManager: CheckoutDiffManager; private readonly backgroundGitFetchManager: BackgroundGitFetchManager; + private readonly daemonConfigStore: DaemonConfigStore; private readonly mcpBaseUrl: string | null; private readonly downloadTokenStore: DownloadTokenStore; private readonly pushTokenStore: PushTokenStore; @@ -661,6 +664,7 @@ export class Session { loopService, checkoutDiffManager, backgroundGitFetchManager, + daemonConfigStore, mcpBaseUrl, stt, tts, @@ -689,6 +693,7 @@ export class Session { this.loopService = loopService; this.checkoutDiffManager = checkoutDiffManager; this.backgroundGitFetchManager = backgroundGitFetchManager; + this.daemonConfigStore = daemonConfigStore; this.mcpBaseUrl = mcpBaseUrl ?? null; this.terminalManager = terminalManager; this.providerSnapshotManager = providerSnapshotManager ?? null; @@ -1611,6 +1616,26 @@ export class Session { await this.handleWaitForFinish(msg.agentId, msg.requestId, msg.timeoutMs); break; + case "get_daemon_config_request": + this.emit({ + type: "get_daemon_config_response", + payload: { + requestId: msg.requestId, + config: this.daemonConfigStore.get(), + }, + }); + break; + + case "set_daemon_config_request": + this.emit({ + type: "set_daemon_config_response", + payload: { + requestId: msg.requestId, + config: this.daemonConfigStore.patch(msg.config), + }, + }); + break; + case "dictation_stream_start": { const unavailable = this.resolveVoiceFeatureUnavailableContext("dictation"); diff --git a/packages/server/src/server/websocket-server.notifications.test.ts b/packages/server/src/server/websocket-server.notifications.test.ts index 25e9f00d6..bf7f84a50 100644 --- a/packages/server/src/server/websocket-server.notifications.test.ts +++ b/packages/server/src/server/websocket-server.notifications.test.ts @@ -65,6 +65,9 @@ function createServer(agentManagerOverrides?: Record) { getAgent: vi.fn(() => null), ...agentManagerOverrides, }; + const daemonConfigStore = { + onChange: vi.fn(() => () => {}), + }; const server = new VoiceAssistantWebSocketServer( {} as any, @@ -74,6 +77,7 @@ function createServer(agentManagerOverrides?: Record) { {} as any, {} as any, "/tmp/paseo-test", + daemonConfigStore as any, null, { allowedOrigins: new Set() }, undefined, diff --git a/packages/server/src/server/websocket-server.relay-reconnect.test.ts b/packages/server/src/server/websocket-server.relay-reconnect.test.ts index c258af0fd..a6c70e40a 100644 --- a/packages/server/src/server/websocket-server.relay-reconnect.test.ts +++ b/packages/server/src/server/websocket-server.relay-reconnect.test.ts @@ -147,6 +147,9 @@ function createLogger() { function createServer(options?: { speechReadiness?: SpeechReadinessSnapshot | null }) { const speechReadiness = options?.speechReadiness ?? null; + const daemonConfigStore = { + onChange: vi.fn(() => () => {}), + }; return new VoiceAssistantWebSocketServer( {} as any, createLogger() as any, @@ -165,6 +168,7 @@ function createServer(options?: { speechReadiness?: SpeechReadinessSnapshot | nu {} as any, {} as any, "/tmp/paseo-test", + daemonConfigStore as any, null, { allowedOrigins: new Set() }, speechReadiness diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 360a0dd3a..37a4310b9 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -13,6 +13,7 @@ import type { LoopService } from "./loop-service.js"; import type { ScheduleService } from "./schedule/service.js"; import type { CheckoutDiffManager, CheckoutDiffMetrics } from "./checkout-diff-manager.js"; import { BackgroundGitFetchManager } from "./background-git-fetch-manager.js"; +import type { DaemonConfigStore, MutableDaemonConfig } from "./daemon-config-store.js"; import { type ServerInfoStatusPayload, type WSHelloMessage, @@ -239,6 +240,7 @@ export class VoiceAssistantWebSocketServer { private readonly backgroundGitFetchManager: BackgroundGitFetchManager; private readonly downloadTokenStore: DownloadTokenStore; private readonly paseoHome: string; + private readonly daemonConfigStore: DaemonConfigStore; private readonly pushTokenStore: PushTokenStore; private readonly pushService: PushService; private readonly mcpBaseUrl: string | null; @@ -276,6 +278,7 @@ export class VoiceAssistantWebSocketServer { private readonly requestLatencies = new Map(); private runtimeMetricsInterval: ReturnType | null = null; private unsubscribeSpeechReadiness: (() => void) | null = null; + private unsubscribeDaemonConfigChange: (() => void) | null = null; constructor( server: HTTPServer, @@ -285,6 +288,7 @@ export class VoiceAssistantWebSocketServer { agentStorage: AgentStorage, downloadTokenStore: DownloadTokenStore, paseoHome: string, + daemonConfigStore: DaemonConfigStore, mcpBaseUrl: string | null, wsConfig: WebSocketServerConfig, speech?: SpeechService | null, @@ -333,6 +337,7 @@ export class VoiceAssistantWebSocketServer { }); this.downloadTokenStore = downloadTokenStore; this.paseoHome = paseoHome; + this.daemonConfigStore = daemonConfigStore; this.mcpBaseUrl = mcpBaseUrl; this.speech = speech ?? null; this.terminalManager = terminalManager ?? null; @@ -352,6 +357,9 @@ export class VoiceAssistantWebSocketServer { this.unsubscribeSpeechReadiness = this.speech?.onReadinessChange((snapshot) => { this.publishSpeechReadiness(snapshot); }) ?? null; + this.unsubscribeDaemonConfigChange = this.daemonConfigStore.onChange((config) => { + this.broadcastDaemonConfigChanged(config); + }); const pushLogger = this.logger.child({ module: "push" }); this.pushTokenStore = new PushTokenStore(pushLogger, join(paseoHome, "push-tokens.json")); @@ -442,6 +450,8 @@ export class VoiceAssistantWebSocketServer { public async close(): Promise { this.unsubscribeSpeechReadiness?.(); this.unsubscribeSpeechReadiness = null; + this.unsubscribeDaemonConfigChange?.(); + this.unsubscribeDaemonConfigChange = null; if (this.runtimeMetricsInterval) { clearInterval(this.runtimeMetricsInterval); this.runtimeMetricsInterval = null; @@ -637,6 +647,7 @@ export class VoiceAssistantWebSocketServer { scheduleService: this.scheduleService, checkoutDiffManager: this.checkoutDiffManager, backgroundGitFetchManager: this.backgroundGitFetchManager, + daemonConfigStore: this.daemonConfigStore, mcpBaseUrl: this.mcpBaseUrl, stt: () => this.speech?.resolveStt() ?? null, tts: () => this.speech?.resolveTts() ?? null, @@ -802,10 +813,24 @@ export class VoiceAssistantWebSocketServer { }; } + private createDaemonConfigChangedMessage(config: MutableDaemonConfig): WSOutboundMessage { + return wrapSessionMessage({ + type: "status", + payload: { + status: "daemon_config_changed", + config, + }, + }); + } + private broadcastCapabilitiesUpdate(): void { this.broadcast(this.createServerInfoMessage()); } + private broadcastDaemonConfigChanged(config: MutableDaemonConfig): void { + this.broadcast(this.createDaemonConfigChangedMessage(config)); + } + private bindSocketHandlers(ws: WebSocketLike): void { ws.on("message", (data) => { void this.handleRawMessage(ws, data); diff --git a/packages/server/src/shared/messages.ts b/packages/server/src/shared/messages.ts index 886d13941..59c971040 100644 --- a/packages/server/src/shared/messages.ts +++ b/packages/server/src/shared/messages.ts @@ -47,6 +47,29 @@ import { LoopLogsResponseSchema, LoopStopResponseSchema, } from "../server/loop/rpc-schemas.js"; +// --------------------------------------------------------------------------- +// Mutable daemon config schemas (shared between server store and client) +// --------------------------------------------------------------------------- + +export const MutableDaemonConfigSchema = z + .object({ + mcp: z + .object({ + injectIntoAgents: z.boolean(), + }) + .passthrough(), + }) + .passthrough(); + +export const MutableDaemonConfigPatchSchema = z + .object({ + mcp: MutableDaemonConfigSchema.shape.mcp.partial().optional(), + }) + .partial() + .passthrough(); + +export type MutableDaemonConfig = z.infer; +export type MutableDaemonConfigPatch = z.infer; import type { LiteralUnion } from "./literal-union.js"; import type { AgentCapabilityFlags, @@ -725,6 +748,17 @@ export const WaitForFinishRequestSchema = z.object({ timeoutMs: z.number().int().positive().optional(), }); +export const GetDaemonConfigRequestMessageSchema = z.object({ + type: z.literal("get_daemon_config_request"), + requestId: z.string(), +}); + +export const SetDaemonConfigRequestMessageSchema = z.object({ + type: z.literal("set_daemon_config_request"), + requestId: z.string(), + config: MutableDaemonConfigPatchSchema, +}); + // ============================================================================ // Dictation Streaming (lossless, resumable) // ============================================================================ @@ -1397,6 +1431,8 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ SetVoiceModeMessageSchema, SendAgentMessageRequestSchema, WaitForFinishRequestSchema, + GetDaemonConfigRequestMessageSchema, + SetDaemonConfigRequestMessageSchema, DictationStreamStartMessageSchema, DictationStreamChunkMessageSchema, DictationStreamFinishMessageSchema, @@ -1732,6 +1768,13 @@ export const ShutdownRequestedStatusPayloadSchema = z.object({ requestId: z.string(), }); +export const DaemonConfigChangedStatusPayloadSchema = z + .object({ + status: z.literal("daemon_config_changed"), + config: MutableDaemonConfigSchema, + }) + .passthrough(); + export const KnownStatusPayloadSchema = z.discriminatedUnion("status", [ AgentCreatedStatusPayloadSchema, AgentCreateFailedStatusPayloadSchema, @@ -1739,6 +1782,7 @@ export const KnownStatusPayloadSchema = z.discriminatedUnion("status", [ AgentRefreshedStatusPayloadSchema, ShutdownRequestedStatusPayloadSchema, RestartRequestedStatusPayloadSchema, + DaemonConfigChangedStatusPayloadSchema, ]); export type KnownStatusPayload = z.infer; @@ -2028,6 +2072,26 @@ export const WaitForFinishResponseMessageSchema = z.object({ }), }); +export const GetDaemonConfigResponseMessageSchema = z.object({ + type: z.literal("get_daemon_config_response"), + payload: z + .object({ + requestId: z.string(), + config: MutableDaemonConfigSchema, + }) + .passthrough(), +}); + +export const SetDaemonConfigResponseMessageSchema = z.object({ + type: z.literal("set_daemon_config_response"), + payload: z + .object({ + requestId: z.string(), + config: MutableDaemonConfigSchema, + }) + .passthrough(), +}); + export const AgentPermissionRequestMessageSchema = z.object({ type: z.literal("agent_permission_request"), payload: z.object({ @@ -2658,6 +2722,8 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ FetchAgentTimelineResponseMessageSchema, SendAgentMessageResponseMessageSchema, SetVoiceModeResponseMessageSchema, + GetDaemonConfigResponseMessageSchema, + SetDaemonConfigResponseMessageSchema, SetAgentModeResponseMessageSchema, SetAgentModelResponseMessageSchema, SetAgentThinkingResponseMessageSchema,