diff --git a/src/cli.test.ts b/src/cli.test.ts index 1d982415..216b2a90 100644 --- a/src/cli.test.ts +++ b/src/cli.test.ts @@ -125,14 +125,17 @@ try { ] : request.method === "hello" ? { - state: "ready", - protocolVersion: LOCAL_AGENT_DAEMON_PROTOCOL_VERSION, - pid: process.pid, - endpoint: daemonSocket, - startedAt: "now", - activeTurns: 0, - runtimeCount: 0, - clientConnections: 1, + status: { + state: "ready", + protocolVersion: LOCAL_AGENT_DAEMON_PROTOCOL_VERSION, + pid: process.pid, + endpoint: daemonSocket, + startedAt: "now", + activeTurns: 0, + runtimeCount: 0, + clientConnections: 1, + }, + configMatches: true, } : null; socket.end(encodeLocalAgentDaemonResponse({ diff --git a/src/local-agent-client.ts b/src/local-agent-client.ts index 5d31464b..4e6367d9 100644 --- a/src/local-agent-client.ts +++ b/src/local-agent-client.ts @@ -6,6 +6,7 @@ import { fileURLToPath } from "node:url"; import { matchError, Result, type Result as BetterResult } from "better-result"; import type { ServerConfig } from "./config.js"; import { + AgentDaemonConfigChangedError, AgentDaemonInvalidRequestError, AgentDaemonInvalidResponseError, AgentDaemonProtocolMismatchError, @@ -23,6 +24,7 @@ import { decodeAgentRecord, decodeAgentRecordList, decodeAgentWaitResults, + decodeDaemonHello, decodeDaemonLogs, decodeDaemonStatus, decodeLocalAgentDaemonResponse, @@ -33,6 +35,7 @@ import { type LocalAgentDaemonResponse, type LocalAgentDaemonStatus, } from "./local-agent-daemon-protocol.js"; +import { localAgentProviderConfigRevision } from "./local-agent-config.js"; import { LOCAL_AGENT_DAEMON_PROTOCOL_VERSION, ensureLocalAgentDaemonSecret, @@ -68,6 +71,7 @@ type RequestError = export interface LocalAgentClientOptions { stateDir: string; + configRevision: string; configDir?: string; startupTimeoutMs?: number; requestTimeoutMs?: number; @@ -78,6 +82,7 @@ export interface LocalAgentClientOptions { export class LocalAgentClient { private readonly stateDir: string; private readonly paths: LocalAgentDaemonPaths; + private readonly configRevision: string; private readonly endpoint: string; private readonly startupTimeoutMs: number; private readonly requestTimeoutMs: number; @@ -86,6 +91,7 @@ export class LocalAgentClient { constructor(options: LocalAgentClientOptions) { this.stateDir = options.stateDir; + this.configRevision = options.configRevision; this.paths = localAgentDaemonPaths(options.stateDir); this.endpoint = options.endpoint ?? this.paths.endpoint; this.startupTimeoutMs = options.startupTimeoutMs ?? DEFAULT_STARTUP_TIMEOUT_MS; @@ -226,6 +232,7 @@ export class LocalAgentClient { authToken: authToken.value, method: "hello", params: {}, + configRevision: this.configRevision, }, this.requestTimeoutMs); if (response.isErr()) { if ( @@ -253,8 +260,28 @@ export class LocalAgentClient { } return error.code === "DAEMON_UNAVAILABLE" ? Result.ok(undefined) : Result.err(error); } - const decoded = decodeValue(response.value.result, "hello", decodeDaemonStatus); - return decoded.map((status) => status.state === "ready" ? status : undefined); + const decoded = decodeValue(response.value.result, "hello", decodeDaemonHello); + if (decoded.isErr()) return decoded; + if (!decoded.value.configMatches) { + return this.replaceIdleChangedDaemon(authToken.value, decoded.value.status); + } + return Result.ok(decoded.value.status.state === "ready" ? decoded.value.status : undefined); + } + + private async replaceIdleChangedDaemon( + authToken: string, + status: LocalAgentDaemonStatus, + ): Promise> { + const changed = new AgentDaemonConfigChangedError({ + code: "DAEMON_CONFIG_CHANGED", + operation: "startup", + retryable: true, + message: status.activeTurns > 0 + ? "The local agent daemon is running active turns with an older provider configuration. Retry after they finish." + : "The local agent daemon is using an older provider configuration.", + }); + if (status.activeTurns > 0) return Result.err(changed); + return this.stopIdleDaemon(authToken, LOCAL_AGENT_DAEMON_PROTOCOL_VERSION, status, changed); } private async replaceIdleOlderDaemon( @@ -268,6 +295,7 @@ export class LocalAgentClient { authToken, method: "hello", params: {}, + configRevision: this.configRevision, }, this.requestTimeoutMs); if (statusResponse.isErr() || !statusResponse.value.ok) return Result.err(mismatch); const status = decodeValue(statusResponse.value.result, "hello", decodeDaemonStatus); @@ -282,14 +310,27 @@ export class LocalAgentClient { })); } + return this.stopIdleDaemon(authToken, protocolVersion, status.value, mismatch); + } + + private async stopIdleDaemon( + authToken: string, + protocolVersion: number, + status: LocalAgentDaemonStatus, + cause: AgentDaemonProtocolMismatchError | AgentDaemonConfigChangedError, + ): Promise> { const stopResponse = await sendRequest(this.endpoint, { requestId: randomUUID(), protocolVersion, authToken, method: "daemon.stop", - params: {}, + // Older daemons do not support atomic idle replacement. Their existing + // best-effort upgrade path remains available through the legacy shape. + params: protocolVersion === LOCAL_AGENT_DAEMON_PROTOCOL_VERSION + ? { ifIdle: true } + : {}, }, this.requestTimeoutMs); - if (stopResponse.isErr() || !stopResponse.value.ok) return Result.err(mismatch); + if (stopResponse.isErr() || !stopResponse.value.ok) return Result.err(cause); const deadline = Date.now() + this.startupTimeoutMs; while (Date.now() < deadline) { @@ -300,28 +341,33 @@ export class LocalAgentClient { authToken, method: "hello", params: {}, + configRevision: this.configRevision, }, Math.min(this.requestTimeoutMs, 250)); if (probe.isErr() && probe.error.code === "DAEMON_UNAVAILABLE") { - if (!existsSync(this.paths.lockPath) || !isProcessAlive(status.value.pid)) { + if (!existsSync(this.paths.lockPath) || !isProcessAlive(status.pid)) { return Result.ok(undefined); } continue; } - if ( - probe.isOk() - && probe.value.protocolVersion >= LOCAL_AGENT_DAEMON_PROTOCOL_VERSION - ) { + if (probe.isOk() && probe.value.protocolVersion > protocolVersion) { // Another client completed the replacement while this client was // waiting for the old endpoint to disappear. return this.tryHello(); } + if (probe.isOk() && probe.value.ok && protocolVersion === LOCAL_AGENT_DAEMON_PROTOCOL_VERSION) { + const hello = decodeValue(probe.value.result, "hello", decodeDaemonHello); + if (hello.isErr()) return hello; + if (hello.value.configMatches && hello.value.status.state === "ready") { + return Result.ok(hello.value.status); + } + } } return Result.err(new AgentDaemonStartupError({ code: "DAEMON_STARTUP_FAILURE", operation: "startup", retryable: true, - cause: mismatch, - message: "The older local agent daemon did not stop in time for the upgrade.", + cause, + message: "The local agent daemon did not stop in time for replacement.", })); } @@ -430,9 +476,13 @@ export class LocalAgentClient { } export function createLocalAgentClient( - config: Pick, + config: Pick, ): LocalAgentClient { - return new LocalAgentClient({ configDir: config.configDir, stateDir: config.stateDir }); + return new LocalAgentClient({ + configDir: config.configDir, + stateDir: config.stateDir, + configRevision: localAgentProviderConfigRevision(config.subagents), + }); } export function spawnLocalAgentDaemon( @@ -607,6 +657,7 @@ function isRequestError( AgentDaemonStartupError: () => "daemon" as const, AgentDaemonTimeoutError: () => "daemon" as const, AgentDaemonProtocolMismatchError: () => "daemon" as const, + AgentDaemonConfigChangedError: () => "daemon" as const, AgentDaemonUnauthorizedError: () => "daemon" as const, AgentDaemonInvalidRequestError: () => "daemon" as const, AgentDaemonInvalidResponseError: () => "daemon" as const, diff --git a/src/local-agent-config.test.ts b/src/local-agent-config.test.ts index a27812a1..237f92b7 100644 --- a/src/local-agent-config.test.ts +++ b/src/local-agent-config.test.ts @@ -1,6 +1,7 @@ import assert from "node:assert/strict"; import { isSubagentProviderEnabled, + localAgentProviderConfigRevision, localAgentProviderEnvironment, subagentProviderConfig, subagentsConfigSchema, @@ -55,6 +56,33 @@ assert.deepEqual(inherited, { OPENAI_API_KEY: "inherited", UNCHANGED: "yes", }); +assert.equal( + localAgentProviderConfigRevision(config), + localAgentProviderConfigRevision(subagentsConfigSchema.parse({ + enabled: true, + providers: [ + { id: "claude", enabled: false, model: "sonnet" }, + { + id: "codex", + enabled: true, + effort: "high", + model: "gpt-5.4", + command: "/opt/bin/codex-wrapper", + env: { EMPTY_VALUE: "", OPENAI_API_KEY: "configured" }, + }, + ], + })), + "provider and environment key order must not restart the daemon", +); +assert.notEqual( + localAgentProviderConfigRevision(config), + localAgentProviderConfigRevision(subagentsConfigSchema.parse({ + ...config, + providers: config.providers.map((provider) => provider.id === "codex" + ? { ...provider, command: "/opt/bin/another-wrapper" } + : provider), + })), +); assert.throws( () => subagentsConfigSchema.parse({ enabled: true, diff --git a/src/local-agent-config.ts b/src/local-agent-config.ts index ae3f4201..572b2dfd 100644 --- a/src/local-agent-config.ts +++ b/src/local-agent-config.ts @@ -1,3 +1,4 @@ +import { createHash } from "node:crypto"; import * as z from "zod/v4"; import { LOCAL_AGENT_PROVIDERS, @@ -93,3 +94,25 @@ export function providerCommandVariable(provider: LocalAgentProvider): string | return undefined; } } + +export function localAgentProviderConfigRevision(config: SubagentsConfig): string { + const providers = [...config.providers] + .sort((left, right) => left.id.localeCompare(right.id)) + .map((provider) => ({ + id: provider.id, + enabled: provider.enabled, + ...(provider.model ? { model: provider.model } : {}), + ...(provider.effort ? { effort: provider.effort } : {}), + ...(provider.command ? { command: provider.command } : {}), + ...(provider.env && Object.keys(provider.env).length > 0 + ? { + env: Object.fromEntries( + Object.entries(provider.env).sort(([left], [right]) => left.localeCompare(right)), + ), + } + : {}), + })); + return createHash("sha256") + .update(JSON.stringify({ enabled: config.enabled, providers })) + .digest("hex"); +} diff --git a/src/local-agent-daemon-lifecycle.ts b/src/local-agent-daemon-lifecycle.ts index 250bb231..5ac32b93 100644 --- a/src/local-agent-daemon-lifecycle.ts +++ b/src/local-agent-daemon-lifecycle.ts @@ -12,7 +12,7 @@ import { } from "node:fs"; import { join, resolve } from "node:path"; -export const LOCAL_AGENT_DAEMON_PROTOCOL_VERSION = 4; +export const LOCAL_AGENT_DAEMON_PROTOCOL_VERSION = 5; export const LOCAL_AGENT_DAEMON_SOCKET_NAME = "agentd.sock"; export const LOCAL_AGENT_DAEMON_PID_NAME = "agentd.pid"; export const LOCAL_AGENT_DAEMON_LOCK_NAME = "agentd.lock"; diff --git a/src/local-agent-daemon-main.ts b/src/local-agent-daemon-main.ts index 0121441d..51a9fc47 100644 --- a/src/local-agent-daemon-main.ts +++ b/src/local-agent-daemon-main.ts @@ -10,6 +10,7 @@ import { import { LocalAgentManager } from "./local-agent-manager.js"; import { LocalAgentRuntimePool } from "./local-agent-runtime-pool.js"; import { LocalAgentStore } from "./local-agent-store.js"; +import { localAgentProviderConfigRevision } from "./local-agent-config.js"; const config = loadConfig(); const DEFAULT_DAEMON_SHUTDOWN_TIMEOUT_MS = 10_000; @@ -33,6 +34,7 @@ const manager = new LocalAgentManager({ const daemon = new LocalAgentDaemon({ stateDir: paths.stateDir, manager, + configRevision: localAgentProviderConfigRevision(config.subagents), onLockAcquired: () => { const reconciled = manager.reconcileActiveRuns(); if (reconciled.isErr()) throw reconciled.error; diff --git a/src/local-agent-daemon-protocol.test.ts b/src/local-agent-daemon-protocol.test.ts index 24acaf31..95d75045 100644 --- a/src/local-agent-daemon-protocol.test.ts +++ b/src/local-agent-daemon-protocol.test.ts @@ -2,6 +2,7 @@ import assert from "node:assert/strict"; import { decodeAgentRecord, decodeAgentWaitResults, + decodeDaemonHello, decodeLocalAgentDaemonRequest, decodeLocalAgentDaemonResponse, encodeLocalAgentDaemonResponse, @@ -55,6 +56,53 @@ const directRequest = decodeLocalAgentDaemonRequest({ if (directRequest.method !== "agent.start") throw new Error("expected agent.start request"); assert.equal(directRequest.params.workspaceId, undefined); +const helloRequest = decodeLocalAgentDaemonRequest({ + requestId: "req_hello", + protocolVersion: LOCAL_AGENT_DAEMON_PROTOCOL_VERSION, + authToken: "test-secret", + method: "hello", + params: {}, + configRevision: "provider-config-revision", +}); +assert.equal(helloRequest.method, "hello"); +if (helloRequest.method !== "hello") throw new Error("expected hello request"); +assert.equal(helloRequest.configRevision, "provider-config-revision"); +const conditionalStop = decodeLocalAgentDaemonRequest({ + requestId: "req_stop", + protocolVersion: LOCAL_AGENT_DAEMON_PROTOCOL_VERSION, + authToken: "test-secret", + method: "daemon.stop", + params: { ifIdle: true }, +}); +assert.equal(conditionalStop.method, "daemon.stop"); +if (conditionalStop.method !== "daemon.stop") throw new Error("expected daemon.stop request"); +assert.equal(conditionalStop.params.ifIdle, true); +assert.deepEqual(decodeDaemonHello({ + status: { + state: "ready", + protocolVersion: LOCAL_AGENT_DAEMON_PROTOCOL_VERSION, + pid: 123, + endpoint: "/tmp/agentd.sock", + startedAt: "now", + activeTurns: 0, + runtimeCount: 0, + clientConnections: 1, + }, + configMatches: false, +}), { + status: { + state: "ready", + protocolVersion: LOCAL_AGENT_DAEMON_PROTOCOL_VERSION, + pid: 123, + endpoint: "/tmp/agentd.sock", + startedAt: "now", + activeTurns: 0, + runtimeCount: 0, + clientConnections: 1, + }, + configMatches: false, +}); + assert.throws( () => decodeLocalAgentDaemonRequest({ requestId: "req_2", diff --git a/src/local-agent-daemon-protocol.ts b/src/local-agent-daemon-protocol.ts index bf87a471..a19571e3 100644 --- a/src/local-agent-daemon-protocol.ts +++ b/src/local-agent-daemon-protocol.ts @@ -23,7 +23,7 @@ export type LocalAgentDaemonMethod = | "daemon.logs"; export type LocalAgentDaemonRequest = - | AgentDaemonRequestBase<"hello", Record> + | (AgentDaemonRequestBase<"hello", Record> & { configRevision?: string }) | AgentDaemonRequestBase<"agent.start", StartLocalAgentInput> | AgentDaemonRequestBase<"agent.continue", { id: string; prompt: string; scope: LocalAgentWorkspaceScope; overrides?: RunOverrides }> | AgentDaemonRequestBase<"agent.get", { id: string; scope: LocalAgentWorkspaceScope }> @@ -34,7 +34,7 @@ export type LocalAgentDaemonRequest = timeoutMs?: number; }> | AgentDaemonRequestBase<"daemon.status", Record> - | AgentDaemonRequestBase<"daemon.stop", Record> + | AgentDaemonRequestBase<"daemon.stop", { ifIdle?: boolean }> | AgentDaemonRequestBase<"daemon.logs", { lines?: number }>; interface AgentDaemonRequestBase< @@ -59,6 +59,11 @@ export interface LocalAgentDaemonStatus { clientConnections: number; } +export interface LocalAgentDaemonHello { + status: LocalAgentDaemonStatus; + configMatches: boolean; +} + export interface LocalAgentDaemonErrorPayload { code: string; message: string; @@ -102,9 +107,24 @@ export function decodeLocalAgentDaemonRequest(value: unknown): LocalAgentDaemonR switch (method) { case "hello": + return { + requestId, + protocolVersion, + authToken, + method, + params: decodeEmptyParams(params), + configRevision: optionalString(record?.configRevision), + }; case "daemon.status": - case "daemon.stop": return { requestId, protocolVersion, authToken, method, params: decodeEmptyParams(params) } as LocalAgentDaemonRequest; + case "daemon.stop": + return { + requestId, + protocolVersion, + authToken, + method, + params: decodeStopParams(params), + }; case "agent.start": return { requestId, @@ -269,6 +289,14 @@ export function decodeDaemonStatus(value: unknown): LocalAgentDaemonStatus { }; } +export function decodeDaemonHello(value: unknown): LocalAgentDaemonHello { + const record = asRecord(value); + return { + status: decodeDaemonStatus(record?.status), + configMatches: requiredBoolean(record?.configMatches, "configMatches"), + }; +} + export function decodeDaemonLogs(value: unknown): string { if (typeof value !== "string") throw new LocalAgentDaemonProtocolError("INVALID_RESULT", "Daemon returned invalid logs."); return value; @@ -331,6 +359,13 @@ function decodeListScope(value: unknown): LocalAgentWorkspaceScope { return decodeWorkspaceScope(value); } +function decodeStopParams(value: unknown): { ifIdle?: boolean } { + const record = asRecord(value); + if (!record) throw new LocalAgentDaemonProtocolError("INVALID_PARAMS", "Daemon stop options must be an object."); + const ifIdle = optionalBoolean(record.ifIdle); + return ifIdle === undefined ? {} : { ifIdle }; +} + function decodeWaitParams(value: unknown): { ids: string[]; scope: LocalAgentWorkspaceScope; @@ -414,6 +449,13 @@ function requiredInteger(value: unknown, field: string): number { return value; } +function requiredBoolean(value: unknown, field: string): boolean { + if (typeof value !== "boolean") { + throw new LocalAgentDaemonProtocolError("INVALID_PROTOCOL", `Invalid ${field}.`); + } + return value; +} + function optionalString(value: unknown): string | undefined { if (typeof value !== "string") return undefined; const trimmed = value.trim(); diff --git a/src/local-agent-daemon.test.ts b/src/local-agent-daemon.test.ts index 4d196ef3..466cf066 100644 --- a/src/local-agent-daemon.test.ts +++ b/src/local-agent-daemon.test.ts @@ -24,6 +24,7 @@ import type { RunOverrides, StartLocalAgentInput } from "./local-agent-manager.j import type { LocalAgentRecord } from "./local-agent-store.js"; const root = await mkdtemp(join(tmpdir(), "devspace-agentd-test-")); +const CONFIG_REVISION = "test-provider-config"; const record: LocalAgentRecord = { id: "agt_test", workspaceId: "ws_test", @@ -41,14 +42,27 @@ class FakeManager implements LocalAgentDaemonManager { closed = false; lastInput?: StartLocalAgentInput; blockWaitUntilAbort = false; + blockStartUntilRelease = false; + startStarted = false; waitStarted = false; waitAborted = false; + private releaseStart?: () => void; async start(input: StartLocalAgentInput) { this.lastInput = input; + this.startStarted = true; + if (this.blockStartUntilRelease) { + await new Promise((resolveStart) => { this.releaseStart = resolveStart; }); + this.activeTurnCount = 1; + } return Result.ok(record); } + releaseBlockedStart(): void { + this.releaseStart?.(); + this.releaseStart = undefined; + } + async continue( _agentId: string, _prompt: string, @@ -92,11 +106,13 @@ class FakeManager implements LocalAgentDaemonManager { const manager = new FakeManager(); const daemon = new LocalAgentDaemon({ stateDir: join(root, "state"), + configRevision: CONFIG_REVISION, manager, idleShutdownMs: 60_000, }); const client = new LocalAgentClient({ stateDir: join(root, "state"), + configRevision: CONFIG_REVISION, startupTimeoutMs: 2_000, requestTimeoutMs: 2_000, spawnDaemon: () => { void daemon.start(); }, @@ -106,6 +122,7 @@ const missingDaemonStateDir = join(root, "missing-daemon-state"); let diagnosticSpawnCount = 0; const missingDaemonClient = new LocalAgentClient({ stateDir: missingDaemonStateDir, + configRevision: CONFIG_REVISION, startupTimeoutMs: 50, requestTimeoutMs: 50, spawnDaemon: () => { diagnosticSpawnCount += 1; }, @@ -169,12 +186,14 @@ const idleManager = new FakeManager(); idleManager.activeTurnCount = 0; const idleDaemon = new LocalAgentDaemon({ stateDir: idleStateDir, + configRevision: CONFIG_REVISION, manager: idleManager, idleShutdownMs: 200, idleCheckIntervalMs: 10, }); const idleClient = new LocalAgentClient({ stateDir: idleStateDir, + configRevision: CONFIG_REVISION, startupTimeoutMs: 2_000, requestTimeoutMs: 2_000, spawnDaemon: () => { void idleDaemon.start(); }, @@ -193,11 +212,13 @@ const ownerManager = new FakeManager(); const competingManager = new FakeManager(); const ownerDaemon = new LocalAgentDaemon({ stateDir: ownershipStateDir, + configRevision: CONFIG_REVISION, manager: ownerManager, idleShutdownMs: 60_000, }); const competingDaemon = new LocalAgentDaemon({ stateDir: ownershipStateDir, + configRevision: CONFIG_REVISION, manager: competingManager, idleShutdownMs: 60_000, }); @@ -223,6 +244,7 @@ try { assert.equal(readFileSync(ownerDaemon.paths.pidPath, "utf8"), pidBefore); const ownerClient = new LocalAgentClient({ stateDir: ownershipStateDir, + configRevision: CONFIG_REVISION, spawnDaemon: () => { throw new Error("the winning daemon should already be reachable"); }, }); assert.equal(unwrap(await ownerClient.status()).pid, process.pid); @@ -233,6 +255,7 @@ try { const startupFailureClient = new LocalAgentClient({ stateDir: join(root, "startup-failure-state"), + configRevision: CONFIG_REVISION, startupTimeoutMs: 20, requestTimeoutMs: 10, spawnDaemon: () => { throw new Error("spawn failed"); }, @@ -241,6 +264,125 @@ const startupFailure = await startupFailureClient.ensureReady(); assert.equal(startupFailure.isErr(), true); if (startupFailure.isErr()) assert.equal(startupFailure.error.code, "DAEMON_STARTUP_FAILURE"); +// Keep Unix socket paths below macOS's short sockaddr_un path limit. +const staleIdleStateDir = join(root, "si"); +const staleIdleManager = new FakeManager(); +staleIdleManager.activeTurnCount = 0; +const staleIdleDaemon = new LocalAgentDaemon({ + stateDir: staleIdleStateDir, + configRevision: "old-provider-config", + manager: staleIdleManager, + idleShutdownMs: 60_000, +}); +const currentManager = new FakeManager(); +currentManager.activeTurnCount = 0; +const currentDaemon = new LocalAgentDaemon({ + stateDir: staleIdleStateDir, + configRevision: CONFIG_REVISION, + manager: currentManager, + idleShutdownMs: 60_000, +}); +let currentDaemonSpawns = 0; +const staleIdleClient = new LocalAgentClient({ + stateDir: staleIdleStateDir, + configRevision: CONFIG_REVISION, + startupTimeoutMs: 2_000, + requestTimeoutMs: 500, + spawnDaemon: () => { + currentDaemonSpawns += 1; + void currentDaemon.start(); + }, +}); +try { + await staleIdleDaemon.start(); + assert.equal(unwrap(await staleIdleClient.ensureReady()).state, "ready"); + assert.equal(staleIdleManager.closed, true); + assert.equal(currentDaemonSpawns, 1); +} finally { + await staleIdleDaemon.close(); + await currentDaemon.close(); +} + +const staleActiveStateDir = join(root, "sa"); +const staleActiveManager = new FakeManager(); +const staleActiveDaemon = new LocalAgentDaemon({ + stateDir: staleActiveStateDir, + configRevision: "old-provider-config", + manager: staleActiveManager, + idleShutdownMs: 60_000, +}); +let staleActiveSpawns = 0; +const staleActiveClient = new LocalAgentClient({ + stateDir: staleActiveStateDir, + configRevision: CONFIG_REVISION, + startupTimeoutMs: 500, + requestTimeoutMs: 500, + spawnDaemon: () => { staleActiveSpawns += 1; }, +}); +try { + await staleActiveDaemon.start(); + const changed = await staleActiveClient.ensureReady(); + assert.equal(changed.isErr(), true); + if (changed.isErr()) { + assert.equal(changed.error.code, "DAEMON_CONFIG_CHANGED"); + assert.equal(changed.error.retryable, true); + } + assert.equal(staleActiveSpawns, 0); + assert.equal(staleActiveManager.closed, false); + assert.deepEqual(Object.keys(unwrap(await staleActiveClient.status())).sort(), [ + "activeTurns", + "clientConnections", + "endpoint", + "pid", + "protocolVersion", + "runtimeCount", + "startedAt", + "state", + ]); +} finally { + await staleActiveDaemon.close(); +} + +const configRaceStateDir = join(root, "sr"); +const configRaceManager = new FakeManager(); +configRaceManager.activeTurnCount = 0; +configRaceManager.blockStartUntilRelease = true; +const configRaceDaemon = new LocalAgentDaemon({ + stateDir: configRaceStateDir, + configRevision: "old-provider-config", + manager: configRaceManager, + idleShutdownMs: 60_000, +}); +const matchingRaceClient = new LocalAgentClient({ + stateDir: configRaceStateDir, + configRevision: "old-provider-config", + spawnDaemon: () => { throw new Error("the existing daemon should be used"); }, +}); +const changedRaceClient = new LocalAgentClient({ + stateDir: configRaceStateDir, + configRevision: CONFIG_REVISION, + spawnDaemon: () => { throw new Error("a busy daemon must not be replaced"); }, +}); +try { + await configRaceDaemon.start(); + const starting = matchingRaceClient.run({ + target: "reviewer", + prompt: "race with replacement", + workspaceId: record.workspaceId, + workspaceRoot: record.workspaceRoot, + }); + await waitFor(() => configRaceManager.startStarted); + const changed = await changedRaceClient.ensureReady(); + assert.equal(changed.isErr(), true); + if (changed.isErr()) assert.equal(changed.error.code, "DAEMON_CONFIG_CHANGED"); + assert.equal(configRaceManager.closed, false); + configRaceManager.releaseBlockedStart(); + unwrap(await starting); +} finally { + configRaceManager.releaseBlockedStart(); + await configRaceDaemon.close(); +} + const upgradeStateDir = join(root, "upgrade-state"); await mkdir(upgradeStateDir, { recursive: true }); const upgradePaths = localAgentDaemonPaths(upgradeStateDir); @@ -306,6 +448,7 @@ const replacementManager = new FakeManager(); replacementManager.activeTurnCount = 0; const replacementDaemon = new LocalAgentDaemon({ stateDir: upgradeStateDir, + configRevision: CONFIG_REVISION, manager: replacementManager, idleShutdownMs: 60_000, }); @@ -313,6 +456,7 @@ let replacementSpawns = 0; let spawnedBeforeLegacyLockReleased = false; const upgradeClient = new LocalAgentClient({ stateDir: upgradeStateDir, + configRevision: CONFIG_REVISION, startupTimeoutMs: 2_000, requestTimeoutMs: 500, spawnDaemon: () => { @@ -368,20 +512,24 @@ const replacementRaceServer = createNetServer((socket) => { })); return; } + const status = { + state: request.method === "daemon.stop" ? "stopping" as const : "ready" as const, + protocolVersion: replacementRaceProtocol, + pid: process.pid, + endpoint: replacementRacePaths.endpoint, + startedAt: "now", + activeTurns: 0, + runtimeCount: 0, + clientConnections: 1, + }; socket.end(encodeLocalAgentDaemonResponse({ requestId: request.requestId, protocolVersion: replacementRaceProtocol, ok: true, - result: { - state: request.method === "daemon.stop" ? "stopping" : "ready", - protocolVersion: replacementRaceProtocol, - pid: process.pid, - endpoint: replacementRacePaths.endpoint, - startedAt: "now", - activeTurns: 0, - runtimeCount: 0, - clientConnections: 1, - }, + result: request.method === "hello" + && replacementRaceProtocol === LOCAL_AGENT_DAEMON_PROTOCOL_VERSION + ? { status, configMatches: true } + : status, }), () => { if (request.method === "daemon.stop") { replacementRaceProtocol = LOCAL_AGENT_DAEMON_PROTOCOL_VERSION; @@ -395,6 +543,7 @@ await new Promise((resolveListen, rejectListen) => { }); const replacementRaceClient = new LocalAgentClient({ stateDir: replacementRaceStateDir, + configRevision: CONFIG_REVISION, startupTimeoutMs: 500, requestTimeoutMs: 100, spawnDaemon: () => { @@ -447,6 +596,7 @@ await new Promise((resolveListen, rejectListen) => { try { const timeoutClient = new LocalAgentClient({ stateDir: timeoutStateDir, + configRevision: CONFIG_REVISION, endpoint: timeoutPaths.endpoint, requestTimeoutMs: 20, spawnDaemon: () => { throw new Error("existing daemon should be used"); }, @@ -484,6 +634,7 @@ await new Promise((resolveListen, rejectListen) => { try { const invalidClient = new LocalAgentClient({ stateDir: invalidStateDir, + configRevision: CONFIG_REVISION, endpoint: invalidPaths.endpoint, requestTimeoutMs: 50, spawnDaemon: () => { throw new Error("existing daemon should be used"); }, @@ -507,6 +658,7 @@ const socketManager = new FakeManager(); socketManager.activeTurnCount = 0; const socketDaemon = new LocalAgentDaemon({ stateDir: socketStateDir, + configRevision: CONFIG_REVISION, manager: socketManager, requestReadTimeoutMs: 30, shutdownTimeoutMs: 100, @@ -550,6 +702,7 @@ try { authToken: "wrong-secret", method: "hello", params: {}, + configRevision: CONFIG_REVISION, }) + "\n"); assert.equal(unauthorized.ok, false); if (!unauthorized.ok) assert.equal(unauthorized.error.code, "DAEMON_UNAUTHORIZED"); diff --git a/src/local-agent-daemon.ts b/src/local-agent-daemon.ts index 75a3b6dd..aac96e9d 100644 --- a/src/local-agent-daemon.ts +++ b/src/local-agent-daemon.ts @@ -70,6 +70,7 @@ export interface LocalAgentDaemonManager { export interface LocalAgentDaemonOptions { stateDir: string; manager: LocalAgentDaemonManager; + configRevision: string; idleShutdownMs?: number; idleCheckIntervalMs?: number; requestReadTimeoutMs?: number; @@ -83,6 +84,7 @@ export interface LocalAgentDaemonOptions { export class LocalAgentDaemon { readonly paths: LocalAgentDaemonPaths; private readonly manager: LocalAgentDaemonManager; + private readonly configRevision: string; private readonly lock: LocalAgentDaemonLock; private readonly idleShutdownMs: number; private readonly idleCheckIntervalMs: number; @@ -99,12 +101,14 @@ export class LocalAgentDaemon { private startedAt?: string; private accepting = false; private stopping = false; + private activeTurnRequests = 0; private authToken?: string; private ownsLock = false; constructor(options: LocalAgentDaemonOptions) { this.paths = options.paths ?? localAgentDaemonPaths(options.stateDir); this.manager = options.manager; + this.configRevision = options.configRevision; this.lock = new LocalAgentDaemonLock(this.paths); this.idleShutdownMs = options.idleShutdownMs ?? DEFAULT_DAEMON_IDLE_SHUTDOWN_MS; this.idleCheckIntervalMs = options.idleCheckIntervalMs ?? DEFAULT_IDLE_CHECK_INTERVAL_MS; @@ -294,6 +298,12 @@ export class LocalAgentDaemon { ); } this.assertAuthenticated(request.authToken); + if (request.method === "hello" && !request.configRevision) { + throw new LocalAgentDaemonProtocolError( + "INVALID_REQUEST", + "Daemon hello requires a provider configuration revision.", + ); + } if (!this.accepting && request.method !== "hello" && request.method !== "daemon.status") { throw new AgentDaemonUnavailableError({ code: "DAEMON_UNAVAILABLE", @@ -305,11 +315,14 @@ export class LocalAgentDaemon { switch (request.method) { case "hello": - return this.status(); + return { + status: this.status(), + configMatches: request.configRevision === this.configRevision, + }; case "agent.start": - return unwrapManagerResult(await this.manager.start(request.params)); + return this.runTurnRequest(() => this.manager.start(request.params)); case "agent.continue": - return unwrapManagerResult(await this.manager.continue( + return this.runTurnRequest(() => this.manager.continue( request.params.id, request.params.prompt, request.params.overrides, @@ -329,6 +342,18 @@ export class LocalAgentDaemon { case "daemon.status": return this.status(); case "daemon.stop": + if (request.params.ifIdle) { + this.accepting = false; + if (this.activeTurnRequests > 0 || this.manager.activeTurnCount > 0) { + this.accepting = true; + throw new AgentDaemonUnavailableError({ + code: "DAEMON_UNAVAILABLE", + operation: "daemon.stop", + retryable: true, + message: "Local agent daemon became busy before it could be replaced.", + }); + } + } this.stopping = true; this.accepting = false; return this.status(); @@ -337,6 +362,17 @@ export class LocalAgentDaemon { } } + private async runTurnRequest( + operation: () => Promise>, + ): Promise { + this.activeTurnRequests += 1; + try { + return unwrapManagerResult(await operation()); + } finally { + this.activeTurnRequests -= 1; + } + } + private writeError(socket: Socket, requestId: string, error: LocalAgentDaemonErrorPayload): void { socket.end(encodeLocalAgentDaemonResponse({ requestId, diff --git a/src/local-agent-errors.ts b/src/local-agent-errors.ts index 0df50b86..32c33cd7 100644 --- a/src/local-agent-errors.ts +++ b/src/local-agent-errors.ts @@ -103,6 +103,10 @@ export class AgentDaemonProtocolMismatchError extends TaggedError( "AgentDaemonProtocolMismatchError", )() {} +export class AgentDaemonConfigChangedError extends TaggedError( + "AgentDaemonConfigChangedError", +)() {} + export class AgentDaemonUnauthorizedError extends TaggedError( "AgentDaemonUnauthorizedError", )() {} @@ -124,6 +128,7 @@ export type AgentDaemonError = | AgentDaemonStartupError | AgentDaemonTimeoutError | AgentDaemonProtocolMismatchError + | AgentDaemonConfigChangedError | AgentDaemonUnauthorizedError | AgentDaemonInvalidRequestError | AgentDaemonInvalidResponseError @@ -179,6 +184,7 @@ export function isAgentDaemonError(error: unknown): error is AgentDaemonError { || AgentDaemonStartupError.is(error) || AgentDaemonTimeoutError.is(error) || AgentDaemonProtocolMismatchError.is(error) + || AgentDaemonConfigChangedError.is(error) || AgentDaemonUnauthorizedError.is(error) || AgentDaemonInvalidRequestError.is(error) || AgentDaemonInvalidResponseError.is(error) @@ -207,6 +213,7 @@ export function toAgentErrorPayload(error: LocalAgentError): AgentErrorPayload { AgentDaemonStartupError: daemonErrorPayload, AgentDaemonTimeoutError: daemonErrorPayload, AgentDaemonProtocolMismatchError: daemonErrorPayload, + AgentDaemonConfigChangedError: daemonErrorPayload, AgentDaemonUnauthorizedError: daemonErrorPayload, AgentDaemonInvalidRequestError: daemonErrorPayload, AgentDaemonInvalidResponseError: daemonErrorPayload, @@ -315,6 +322,13 @@ export function agentErrorFromPayload(payload: { retryable, message: payload.message, }); + case "DAEMON_CONFIG_CHANGED": + return new AgentDaemonConfigChangedError({ + code: payload.code, + operation: payload.operation ?? "hello", + retryable, + message: payload.message, + }); case "DAEMON_UNAUTHORIZED": return new AgentDaemonUnauthorizedError({ code: payload.code,