Add mutable daemon config RPC support

This commit is contained in:
Mohamed Boudra
2026-04-11 16:39:09 +07:00
parent 51e96fa303
commit fa552b0faa
7 changed files with 331 additions and 2 deletions

View File

@@ -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<T> = {
};
type RpcWaitResult<T> = { 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<SessionOutboundMessage, { payload: { requestId: string } }>
| GetDaemonConfigResponse
| SetDaemonConfigResponse;
type CorrelatedResponseType = CorrelatedResponseMessage["type"];
type CorrelatedResponsePayload<TType extends CorrelatedResponseType> = 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;

View File

@@ -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<string, unknown>): 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<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value);
}
function deepMerge<T extends Record<string, unknown>>(current: T, patch: Record<string, unknown>): T {
const next: Record<string, unknown> = { ...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<unknown>((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<ConfigListener>();
private readonly fieldChangeHandlers = new Map<string, Set<FieldChangeHandler>>();
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<FieldChangeHandler>();
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,
},
},
};
}

View File

@@ -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<SpeechToTextProvider | null>;
tts: Resolvable<TextToSpeechProvider | null>;
@@ -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");

View File

@@ -65,6 +65,9 @@ function createServer(agentManagerOverrides?: Record<string, unknown>) {
getAgent: vi.fn(() => null),
...agentManagerOverrides,
};
const daemonConfigStore = {
onChange: vi.fn(() => () => {}),
};
const server = new VoiceAssistantWebSocketServer(
{} as any,
@@ -74,6 +77,7 @@ function createServer(agentManagerOverrides?: Record<string, unknown>) {
{} as any,
{} as any,
"/tmp/paseo-test",
daemonConfigStore as any,
null,
{ allowedOrigins: new Set() },
undefined,

View File

@@ -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

View File

@@ -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<string, number[]>();
private runtimeMetricsInterval: ReturnType<typeof setInterval> | 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<void> {
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);

View File

@@ -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<typeof MutableDaemonConfigSchema>;
export type MutableDaemonConfigPatch = z.infer<typeof MutableDaemonConfigPatchSchema>;
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<typeof KnownStatusPayloadSchema>;
@@ -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,