diff --git a/docs/adr/0006-daemon-rpc-protocol-version.md b/docs/adr/0006-daemon-rpc-protocol-version.md index 7c2a7b1c52..a95ced5e55 100644 --- a/docs/adr/0006-daemon-rpc-protocol-version.md +++ b/docs/adr/0006-daemon-rpc-protocol-version.md @@ -84,3 +84,29 @@ the gap and its reason instead of implying coverage. See `test/wire-compat/READM Legacy remote daemons without `rpcProtocolVersion` remain reachable. This keeps the first release of the proxy compatible with older HTTP daemons, but it means absence of the marker is not proof of compatibility. + +## Daemon capabilities and quiesce + +A daemon on the local state directory publishes `capabilities` in `daemon.json`. This is the +command-level metadata the decision above prefers to a protocol bump. A daemon that predates a +capability omits it, so a client treats absence as "cannot do this" and never infers it from the +package version. + +`quiesce` is how a client replaces an incompatible daemon without destroying other clients' work. +The client sends `daemon_quiesce` with the daemon token and the pid and start time recorded in +`daemon.json`. The daemon checks the token and that it is that instance, refuses a request that +reached it as a remote-origin one, then closes admission and counts what it holds in a single +synchronous step: requests in flight, sessions, repair sessions and live leases, and whether the +daemon is still starting up (`startingUp`), because before that it has no drain path. Any of them +refuses with `daemon-busy` and reopens admission in that same step. Otherwise it answers +`draining`, keeps admission closed, and shuts down. The request is answered before dispatch, so +it joins neither the in-flight count nor the command registry. + +The daemon accepts a quiesce only when it can prove its own start time and the request names both +the pid and that start time; a daemon that cannot is never addressed. A client never signals a +daemon that did not accept one, except to retire it after it answered `draining`. + +The census counts what the contract names and nothing else. A resumable upload still streaming and +a host-administration human-control hold that names no lease are outside it, the same boundary +idle reap draws, so a quiesce can end them. A daemon started with an HTTP auth hook marks every +HTTP request remote-origin and so refuses a quiesce over HTTP; the socket transport is unaffected. diff --git a/packages/daemon-contracts/src/daemon-registration-owner.ts b/packages/daemon-contracts/src/daemon-registration-owner.ts index c2276c9f2d..55d1833059 100644 --- a/packages/daemon-contracts/src/daemon-registration-owner.ts +++ b/packages/daemon-contracts/src/daemon-registration-owner.ts @@ -26,6 +26,7 @@ import { withDaemonDiagnosticsScope } from './daemon-diagnostics-scope.ts'; import type { DaemonCodeOrigin } from '@agent-device/host-kit/code-signature'; import { resolveDaemonPaths, + type DaemonCapability, type DaemonPaths, type DaemonServerMode, } from './daemon-resolution.ts'; @@ -51,6 +52,8 @@ export type DaemonRegistrationFields = Readonly<{ codeOrigin: DaemonCodeOrigin; codeSignature: string; policyDigest?: string; + /** What this daemon can do beyond the base RPC contract; absent from daemons that predate it. */ + capabilities?: readonly DaemonCapability[]; }>; type DaemonRegistrationRemoval = | Readonly<{ state: 'removed' }> diff --git a/packages/daemon-contracts/src/daemon-resolution.ts b/packages/daemon-contracts/src/daemon-resolution.ts index 2081fac0cb..8a5b1d1ff3 100644 --- a/packages/daemon-contracts/src/daemon-resolution.ts +++ b/packages/daemon-contracts/src/daemon-resolution.ts @@ -13,7 +13,9 @@ export type { DaemonServerMode, DaemonTransportPreference }; * daemon both read this subset — the client to find or launch a daemon, the daemon to bind one — * so it sits at the process root rather than under `packages/daemon/src/`, where a client would have to * reach daemon internals to resolve an address. `packages/daemon/src/core/config.ts` adds the request-scoping - * rules that only the daemon applies on top of this. + * rules that only the daemon applies on top of this. What a daemon publishes about itself + * (`capabilities`) and the one request that asks it to leave (`quiesce`) belong here for the same + * reason: the client addresses a daemon with them and the daemon answers by them. */ export type DaemonPaths = { @@ -94,3 +96,53 @@ export function resolveDaemonTransportPreference( if (normalized === 'dual') return 'auto'; return 'auto'; } + +// The quiesce request is how a client asks an idle daemon to leave so a compatible one can take +// its state directory. The client and the daemon share this vocabulary, so neither owns it. + +/** What a daemon can do beyond the base RPC contract, published in its `daemon.json`. */ +export const DAEMON_CAPABILITIES = ['quiesce'] as const; +export type DaemonCapability = (typeof DAEMON_CAPABILITIES)[number]; + +export const DAEMON_QUIESCE_COMMAND = 'daemon_quiesce'; + +/** The reason a daemon gives every request it refuses once it has begun to leave. */ +export const DAEMON_SHUTTING_DOWN_REASON = 'daemon_shutting_down'; + +export const DAEMON_QUIESCE_REASONS = { + busy: 'daemon-busy', + instanceMismatch: 'daemon-instance-mismatch', +} as const; + +/** The daemon instance a quiesce request is addressed to; any other instance refuses it. */ +export type DaemonQuiesceFence = Readonly<{ + pid: number; + /** Null when the daemon could not prove its start time, as its `daemon.json` records. */ + processStartTime: string | null; +}>; + +/** + * What a daemon holds. It leaves a quiesce request only when all four counts are zero and it is + * not still starting up, because before that it has no drain path to run. + */ +export type DaemonQuiesceCensus = Readonly<{ + /** True from process start until the daemon can drain. */ + startingUp: boolean; + inFlightRequests: number; + sessions: number; + repairSessions: number; + leases: number; +}>; + +export function readDaemonQuiesceFence(input: unknown): DaemonQuiesceFence | undefined { + if (typeof input !== 'object' || input === null) return undefined; + const { pid, processStartTime } = input as Record; + if (!Number.isInteger(pid)) return undefined; + if (processStartTime !== null && typeof processStartTime !== 'string') return undefined; + return { pid: pid as number, processStartTime }; +} + +export function readDaemonCapabilities(value: unknown): readonly DaemonCapability[] { + if (!Array.isArray(value)) return []; + return DAEMON_CAPABILITIES.filter((capability) => value.includes(capability)); +} diff --git a/packages/daemon/src/core/__tests__/session-store-lifetime.test.ts b/packages/daemon/src/core/__tests__/session-store-lifetime.test.ts index a528264685..61d9bf852b 100644 --- a/packages/daemon/src/core/__tests__/session-store-lifetime.test.ts +++ b/packages/daemon/src/core/__tests__/session-store-lifetime.test.ts @@ -2,6 +2,7 @@ import assert from 'node:assert/strict'; import fs from 'node:fs'; import { test } from 'vitest'; import { AppError } from '@agent-device/kernel/errors'; +import { DAEMON_SHUTTING_DOWN_REASON } from '@agent-device/daemon-contracts/daemon-resolution'; import { makeSession, makeRepairCompleteSession, @@ -110,7 +111,7 @@ test('shutdown closes draft admission while allowing the current lifetime to set store.closeAdmission(); assert.throws( () => store.publish('late-draft', makeSession('late-draft')), - (error) => error instanceof AppError && error.details?.reason === 'daemon_shutting_down', + (error) => error instanceof AppError && error.details?.reason === DAEMON_SHUTTING_DOWN_REASON, ); store.update(ref, { appName: 'Settled' }); assert.equal(store.requireCurrent(ref).appName, 'Settled'); diff --git a/packages/daemon/src/server/http/daemon-runtime-quiesce.test.ts b/packages/daemon/src/server/http/daemon-runtime-quiesce.test.ts new file mode 100644 index 0000000000..f14afa29b1 --- /dev/null +++ b/packages/daemon/src/server/http/daemon-runtime-quiesce.test.ts @@ -0,0 +1,400 @@ +import fs from 'node:fs'; +import net from 'node:net'; +import { afterEach, expect, test, vi } from 'vitest'; +import { mkdtempForTestSync } from '@agent-device/test-kit'; +import { + DAEMON_QUIESCE_COMMAND, + resolveDaemonPaths, + type DaemonQuiesceFence, +} from '@agent-device/daemon-contracts/daemon-resolution'; +import { readCurrentOwnerIdentity } from '@agent-device/host-kit/process'; +import type { DaemonResponse } from '@agent-device/kernel/contracts'; + +function buildDaemonQuiesceRequest(params: { + token: string; + fence: DaemonQuiesceFence; + requestId: string; +}) { + return { + token: params.token, + session: 'default', + command: DAEMON_QUIESCE_COMMAND, + positionals: [], + input: { pid: params.fence.pid, processStartTime: params.fence.processStartTime }, + meta: { requestId: params.requestId }, + }; +} + +// A quiesce is the daemon's own answer to "may I leave?", so these tests boot the real runtime and +// ask over its real socket: the census, the closing of admission and the drain are one flow. +const probe = vi.hoisted(() => ({ + /** Holds daemon startup after its servers listen, until released. */ + startupGate: undefined as Promise | undefined, + store: undefined as import('../../core/session-store.ts').SessionStore | undefined, + registry: undefined as import('../../resources/lease-registry.ts').LeaseRegistry | undefined, + held: undefined as Promise | undefined, + entered: undefined as (() => void) | undefined, +})); + +vi.mock('@agent-device/daemon-contracts/device-claims', async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + reconcileOrphanedDeviceClaims: async ( + ...args: Parameters + ) => { + await probe.startupGate; + return await actual.reconcileOrphanedDeviceClaims(...args); + }, + }; +}); +vi.mock('../../core/session-store.ts', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + SessionStore: class extends actual.SessionStore { + constructor(...args: ConstructorParameters) { + super(...args); + probe.store = this; + } + }, + }; +}); +vi.mock('../../resources/lease-registry.ts', async (importOriginal) => { + const actual = await importOriginal(); + class RecordedLeaseRegistry extends actual.LeaseRegistry { + constructor(...args: ConstructorParameters) { + super(...args); + probe.registry = this; + } + } + return { ...actual, LeaseRegistry: RecordedLeaseRegistry }; +}); +vi.mock('../request-router.ts', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + createRequestHandler: (...args: Parameters) => { + const dispatch = actual.createRequestHandler(...args); + return async (request: Parameters[0]) => { + if (probe.held) { + probe.entered?.(); + await probe.held; + } + return await dispatch(request); + }; + }, + }; +}); + +import { daemonRuntimeDepsFixture } from '../../__tests__/support/daemon-runtime-deps-fixture.ts'; +import { makeRepairArmedSession, makeSession } from '../../__tests__/support/session-factories.ts'; +import { startDaemonRuntime } from './daemon-runtime.ts'; + +afterEach(() => { + probe.startupGate = undefined; + probe.held = undefined; + probe.entered = undefined; + probe.store = undefined; + probe.registry = undefined; +}); + +type Booted = { + runtime: NonNullable>>; + stateDir: string; + exit: Promise; + quiesce(overrides?: { + token?: string; + pid?: number; + processStartTime?: string | null; + }): Promise; + sessionList(): Promise; +}; + +async function bootDaemon(options: { serverMode?: 'socket' | 'dual' } = {}): Promise { + const stateDir = mkdtempForTestSync('agent-device-daemon-quiesce-'); + const paths = resolveDaemonPaths(stateDir); + let resolveExit: (code: number) => void = () => {}; + const exit = new Promise((resolve) => { + resolveExit = resolve; + }); + const runtime = await startDaemonRuntime({ + composeDeps: daemonRuntimeDepsFixture(), + env: { + ...process.env, + AGENT_DEVICE_STATE_DIR: stateDir, + AGENT_DEVICE_DAEMON_SERVER_MODE: options.serverMode ?? 'socket', + AGENT_DEVICE_DAEMON_IDLE_TIMEOUT_MS: '0', + }, + exit: resolveExit, + registerProcessHandlers: false, + stderr: { write: () => {} }, + stdout: { write: () => {} }, + }); + if (!runtime?.socketPort) throw new Error('daemon did not start'); + const info = JSON.parse(fs.readFileSync(paths.infoPath, 'utf8')) as { + pid: number; + processStartTime?: string; + }; + const socketPort = runtime.socketPort; + const send = (request: object) => sendSocketRequest(socketPort, request); + return { + runtime, + stateDir, + exit, + quiesce: (overrides = {}) => + send( + buildDaemonQuiesceRequest({ + token: overrides.token ?? runtime.token, + fence: { + pid: overrides.pid ?? info.pid, + processStartTime: + overrides.processStartTime !== undefined + ? overrides.processStartTime + : (info.processStartTime ?? null), + }, + requestId: 'quiesce-test', + }), + ), + sessionList: () => + send({ + token: runtime.token, + session: 'default', + command: 'session_list', + positionals: [], + meta: { requestId: 'session-list-test' }, + }), + }; +} + +async function sendSocketRequest(port: number, request: object): Promise { + return await new Promise((resolve, reject) => { + const socket = net.createConnection({ host: '127.0.0.1', port }); + let buffer = ''; + socket.setEncoding('utf8'); + socket.on('connect', () => socket.write(`${JSON.stringify(request)}\n`)); + socket.on('data', (chunk) => { + buffer += chunk; + if (!buffer.includes('\n')) return; + socket.end(); + resolve(JSON.parse(buffer) as DaemonResponse); + }); + socket.on('error', reject); + }); +} + +function reasonOf(response: DaemonResponse): unknown { + return response.ok ? undefined : response.error.details?.reason; +} + +test('an idle daemon accepts a quiesce, drains and releases its state directory', async () => { + const daemon = await bootDaemon(); + const paths = resolveDaemonPaths(daemon.stateDir); + const published = JSON.parse(fs.readFileSync(paths.infoPath, 'utf8')) as { + capabilities?: unknown; + }; + expect(published.capabilities).toEqual(['quiesce']); + + expect(await daemon.quiesce()).toEqual({ ok: true, data: { outcome: 'draining' } }); + + expect(await daemon.exit).toBe(0); + expect(fs.existsSync(paths.infoPath)).toBe(false); + expect(fs.existsSync(paths.lockPath)).toBe(false); +}); + +test('a daemon with a request in flight refuses to quiesce and keeps serving', async () => { + const daemon = await bootDaemon(); + let release: () => void = () => {}; + const entered = new Promise((resolve) => { + probe.entered = resolve; + }); + probe.held = new Promise((resolve) => { + release = resolve; + }); + const running = daemon.sessionList(); + await entered; + + try { + const refusal = await daemon.quiesce(); + expect(refusal.ok).toBe(false); + expect(reasonOf(refusal)).toBe('daemon-busy'); + expect(!refusal.ok && refusal.error.details).toMatchObject({ inFlightRequests: 1 }); + + release(); + expect((await running).ok).toBe(true); + // Admission reopened with the refusal: the daemon still serves and can still be quiesced once idle. + probe.held = undefined; + expect((await daemon.sessionList()).ok).toBe(true); + expect((await daemon.quiesce()).ok).toBe(true); + await daemon.exit; + } finally { + release(); + await running.catch(() => {}); + await daemon.runtime.shutdown(); + } +}); + +test('a daemon holding a session, a repair session or a live lease refuses to quiesce', async () => { + const daemon = await bootDaemon(); + try { + const openRef = probe.store!.publish('open-session', makeSession('open-session')); + const withSession = await daemon.quiesce(); + expect(reasonOf(withSession)).toBe('daemon-busy'); + expect(!withSession.ok && withSession.error.details).toMatchObject({ + sessions: 1, + repairSessions: 0, + }); + probe.store!.retire(openRef); + + const repairRef = probe.store!.publish('repairing', makeRepairArmedSession('repairing')); + const withRepair = await daemon.quiesce(); + expect(!withRepair.ok && withRepair.error.details).toMatchObject({ + sessions: 0, + repairSessions: 1, + }); + probe.store!.retire(repairRef); + + const lease = probe.registry!.allocateLease({ + tenantId: 'tenant-a', + runId: 'run-a', + ttlMs: 60_000, + }); + const withLease = await daemon.quiesce(); + expect(!withLease.ok && withLease.error.details).toMatchObject({ leases: 1 }); + probe.registry!.releaseLease({ leaseId: lease.leaseId }); + + expect((await daemon.quiesce()).ok).toBe(true); + expect(await daemon.exit).toBe(0); + } finally { + await daemon.runtime.shutdown(); + } +}); + +test('a quiesce with the wrong token or for another daemon instance leaves the daemon running', async () => { + const daemon = await bootDaemon(); + try { + const wrongToken = await daemon.quiesce({ token: 'not-the-token' }); + expect(!wrongToken.ok && wrongToken.error.code).toBe('UNAUTHORIZED'); + + const otherInstance = await daemon.quiesce({ pid: readCurrentOwnerIdentity().pid + 1 }); + expect(reasonOf(otherInstance)).toBe('daemon-instance-mismatch'); + // A recycled pid is a different instance: the start time is part of the fence. + const recycledPid = await daemon.quiesce({ processStartTime: 'a different start time' }); + expect(reasonOf(recycledPid)).toBe('daemon-instance-mismatch'); + + expect((await daemon.sessionList()).ok).toBe(true); + } finally { + await daemon.runtime.shutdown(); + } +}); + +async function quiesceOverHttp( + daemon: Booted, + headers: Record, +): Promise<{ result?: DaemonResponse; error?: { data?: { code?: string } } }> { + const { httpPort, token } = daemon.runtime; + const info = JSON.parse( + fs.readFileSync(resolveDaemonPaths(daemon.stateDir).infoPath, 'utf8'), + ) as { pid: number; processStartTime?: string }; + const response = await fetch(`http://127.0.0.1:${httpPort}/rpc`, { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${token}`, + ...headers, + }, + body: JSON.stringify({ + jsonrpc: '2.0', + id: 'http-quiesce', + method: 'agent_device.command', + params: buildDaemonQuiesceRequest({ + token, + fence: { pid: info.pid, processStartTime: info.processStartTime ?? null }, + requestId: 'http-quiesce', + }), + }), + }); + return (await response.json()) as never; +} + +test('a quiesce that arrives as a remote-origin HTTP request is refused', async () => { + const daemon = await bootDaemon({ serverMode: 'dual' }); + try { + const body = await quiesceOverHttp(daemon, { 'x-agent-device-network-access': 'public-only' }); + expect(JSON.stringify(body)).toContain('"UNAUTHORIZED"'); + expect((await daemon.sessionList()).ok).toBe(true); + } finally { + await daemon.runtime.shutdown(); + } +}); + +test('a local HTTP quiesce drains an idle daemon like a socket one', async () => { + const daemon = await bootDaemon({ serverMode: 'dual' }); + const body = await quiesceOverHttp(daemon, {}); + expect(body.result).toEqual({ ok: true, data: { outcome: 'draining' } }); + expect(await daemon.exit).toBe(0); +}); + +test('a quiesce asked while the daemon is still starting up is refused as busy, then accepted', async () => { + const stateDir = mkdtempForTestSync('agent-device-daemon-quiesce-startup-'); + const paths = resolveDaemonPaths(stateDir); + let openGate: () => void = () => {}; + probe.startupGate = new Promise((resolve) => { + openGate = resolve; + }); + let resolveExit: (code: number) => void = () => {}; + const exit = new Promise((resolve) => { + resolveExit = resolve; + }); + const starting = startDaemonRuntime({ + composeDeps: daemonRuntimeDepsFixture(), + env: { + ...process.env, + AGENT_DEVICE_STATE_DIR: stateDir, + AGENT_DEVICE_DAEMON_SERVER_MODE: 'socket', + AGENT_DEVICE_DAEMON_IDLE_TIMEOUT_MS: '0', + }, + exit: resolveExit, + registerProcessHandlers: false, + stderr: { write: () => {} }, + stdout: { write: () => {} }, + }); + try { + // Published, listening, and still held inside startup. + for (let polls = 0; !fs.existsSync(paths.infoPath); polls += 1) { + if (polls >= 500) throw new Error('daemon did not publish daemon.json during startup'); + await new Promise((resolve) => setTimeout(resolve, 10)); + } + const info = JSON.parse(fs.readFileSync(paths.infoPath, 'utf8')) as { + port: number; + token: string; + pid: number; + processStartTime?: string; + }; + + const quiesceRequest = buildDaemonQuiesceRequest({ + token: info.token, + fence: { pid: info.pid, processStartTime: info.processStartTime ?? null }, + requestId: 'startup-quiesce', + }); + const response = await sendSocketRequest(info.port, quiesceRequest); + // The daemon is still starting up, so it refuses as busy instead of promising a drain it cannot run yet. + expect(response.ok).toBe(false); + expect(reasonOf(response)).toBe('daemon-busy'); + expect(!response.ok && response.error.details).toMatchObject({ + startingUp: true, + inFlightRequests: 0, + }); + + openGate(); + await starting; + const again = await sendSocketRequest(info.port, quiesceRequest); + expect(again).toEqual({ ok: true, data: { outcome: 'draining' } }); + expect(await exit).toBe(0); + expect(fs.existsSync(paths.infoPath)).toBe(false); + } finally { + openGate(); + await starting.then((runtime) => runtime?.shutdown()); + } +}); diff --git a/packages/daemon/src/server/http/daemon-runtime.ts b/packages/daemon/src/server/http/daemon-runtime.ts index 3de749dc9a..9e46c73fbb 100644 --- a/packages/daemon/src/server/http/daemon-runtime.ts +++ b/packages/daemon/src/server/http/daemon-runtime.ts @@ -9,6 +9,7 @@ import crypto from 'node:crypto'; import { asAppError, AppError, normalizeError } from '@agent-device/kernel/errors'; import { SessionStore } from '../../core/session-store.ts'; import { + DAEMON_CAPABILITIES, resolveDaemonPaths, resolveDaemonServerMode, } from '@agent-device/daemon-contracts/daemon-resolution'; @@ -43,7 +44,11 @@ import { closeDaemonServers } from './server-shutdown.ts'; import type { DaemonInvokeFn } from '../../core/daemon-request.ts'; import type { SessionRef, SessionState } from '../../core/session-state.ts'; import type { RuntimeHintValues } from '@agent-device/contracts/application-lifecycle-runtime'; -import { createDaemonIdleReap } from '../../sessions/server/daemon-idle-reap.ts'; +import { + createDaemonIdleReap, + createDaemonQuiesce, + daemonShuttingDownResponse, +} from '../../sessions/server/daemon-idle-reap.ts'; import { withDaemonDiagnosticsScope } from '@agent-device/daemon-contracts/daemon-diagnostics-scope'; import { createSessionIdleExpiry } from './daemon-session-idle-expiry.ts'; import { resolveSessionIdleExpiryMs } from '../session-idle-expiry.ts'; @@ -622,16 +627,24 @@ export async function startDaemonRuntime( env, }); - const handleRequest: DaemonInvokeFn = async (req) => { - if (shuttingDown) - return { - ok: false, - error: { - code: 'COMMAND_FAILED', - message: 'Daemon is shutting down.', - details: { reason: 'daemon_shutting_down' }, - }, - }; + // The quiesce request is answered before dispatch and never joins `inFlightRequests`, so the + // count it reads needs no allowance for it. The servers listen and `daemon.json` is published + // before `shutdown` exists, and a daemon answers `draining` only once it has a drain path to run. + let startingUp = true; + const quiesce = createDaemonQuiesce({ + token, + identity: { pid: daemonIdentity.pid, processStartTime: daemonIdentity.startTime }, + sessionStore, + leaseRegistry, + inFlightRequestCount: () => inFlightRequests.size, + startingUp: () => startingUp, + drain: () => { + void shutdown(); + }, + }); + + const handleRequest: DaemonInvokeFn = quiesce.admit(async (req) => { + if (shuttingDown) return daemonShuttingDownResponse(); idleReap.cancel(); const dispatch = dispatchRequest(req); inFlightRequests.add(dispatch); @@ -646,7 +659,7 @@ export async function startDaemonRuntime( // the reaper's view of a session's deadline identical to the lock-guarded one. sessionIdleExpiry.noteSessionsChanged(); } - }; + }); const openDaemonServers = async (): Promise<{ servers: DaemonServer[]; @@ -713,6 +726,7 @@ export async function startDaemonRuntime( codeOrigin: daemonCodeOrigin, codeSignature: daemonCodeSignature, policyDigest: daemonPolicy?.digest, + capabilities: DAEMON_CAPABILITIES, }); if (socketPort) stdout.write(`AGENT_DEVICE_DAEMON_PORT=${socketPort}\n`); if (httpPort) stdout.write(`AGENT_DEVICE_DAEMON_HTTP_PORT=${httpPort}\n`); @@ -918,6 +932,8 @@ export async function startDaemonRuntime( exit(shutdownOptions.exitCode ?? 0); }; + startingUp = false; + if (options.registerProcessHandlers !== false) { process.on('SIGINT', () => { void shutdown(); diff --git a/packages/daemon/src/sessions/server/daemon-idle-reap.test.ts b/packages/daemon/src/sessions/server/daemon-idle-reap.test.ts index bc08e79861..e663a012d8 100644 --- a/packages/daemon/src/sessions/server/daemon-idle-reap.test.ts +++ b/packages/daemon/src/sessions/server/daemon-idle-reap.test.ts @@ -7,6 +7,7 @@ import { SessionStore } from '../../core/session-store.ts'; import type { SessionState } from '../../core/session-state.ts'; import { createDaemonIdleReap, + createDaemonQuiesce, hasActiveRecording, hasReapBlockingOpenSessions, isDaemonIdle, @@ -292,3 +293,36 @@ test('cancel prevents a scheduled reap from firing', async () => { assert.equal(reaped, 0); }); + +test('a daemon that cannot prove its start time refuses every quiesce', async () => { + const drain = vi.fn(); + const quiesce = createDaemonQuiesce({ + token: 'daemon-token', + identity: { pid: process.pid, processStartTime: null }, + sessionStore, + leaseRegistry: { listActiveLeases: () => [] }, + inFlightRequestCount: () => 0, + startingUp: () => false, + drain, + }); + const admitted = vi.fn(async () => ({ ok: true as const })); + + const response = await quiesce.admit(admitted)({ + token: 'daemon-token', + session: 'default', + command: 'daemon_quiesce', + positionals: [], + input: { pid: process.pid, processStartTime: null }, + }); + + assert.equal(response.ok, false); + assert.equal(!response.ok && response.error.details?.reason, 'daemon-instance-mismatch'); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(drain.mock.calls.length, 0); + // Admission stays open: a refusal never leaves the daemon closed. + assert.deepEqual(await quiesce.admit(admitted)(makeRequest('devices')), { ok: true }); +}); + +function makeRequest(command: string) { + return { token: 'daemon-token', session: 'default', command, positionals: [] }; +} diff --git a/packages/daemon/src/sessions/server/daemon-idle-reap.ts b/packages/daemon/src/sessions/server/daemon-idle-reap.ts index 3f749e7f68..c65cdd2fee 100644 --- a/packages/daemon/src/sessions/server/daemon-idle-reap.ts +++ b/packages/daemon/src/sessions/server/daemon-idle-reap.ts @@ -1,3 +1,16 @@ +import { + DAEMON_QUIESCE_COMMAND, + DAEMON_QUIESCE_REASONS, + DAEMON_SHUTTING_DOWN_REASON, + readDaemonQuiesceFence, + type DaemonQuiesceCensus, + type DaemonQuiesceFence, +} from '@agent-device/daemon-contracts/daemon-resolution'; +import { timingSafeStringEqual } from '@agent-device/host-kit/transport'; +import type { DaemonResponse } from '@agent-device/kernel/contracts'; +import { AppError, normalizeError } from '@agent-device/kernel/errors'; +import type { DaemonInvokeFn, DaemonRequest } from '../../core/daemon-request.ts'; +import type { LeaseRegistry } from '../../resources/lease-registry.ts'; import { emitDiagnostic } from '@agent-device/host-kit/diagnostics'; import type { SessionStore } from '../../core/session-store.ts'; import type { SessionState } from '../../core/session-state.ts'; @@ -129,3 +142,116 @@ export function createDaemonIdleReap(params: { return { noteActivity, cancel, idleMs }; } + +/** + * ADR 0006: a daemon leaves for a newer one only when it holds nothing, and it decides that + * itself. A client that read "idle" and then signaled would race every other client of the state + * directory; the daemon is the one place the answer and the closing of admission can be one step. + */ +export type DaemonQuiesce = Readonly<{ + /** + * Puts the quiesce request and the closed admission it leaves behind in front of `handler`: a + * quiesce request gets its verdict, and any other request gets the shutting-down refusal once a + * quiesce has closed admission. + */ + admit(handler: DaemonInvokeFn): DaemonInvokeFn; +}>; + +export function daemonShuttingDownResponse(): DaemonResponse { + return { + ok: false, + error: { + code: 'COMMAND_FAILED', + message: 'Daemon is shutting down.', + details: { reason: DAEMON_SHUTTING_DOWN_REASON }, + }, + }; +} + +export function createDaemonQuiesce(params: { + token: string; + identity: DaemonQuiesceFence; + sessionStore: Pick; + leaseRegistry: Pick; + /** Requests being served, not counting the quiesce request itself. */ + inFlightRequestCount: () => number; + /** Whether the daemon has yet to build the drain path `drain` runs. */ + startingUp: () => boolean; + /** Starts the drain; called at most once, after the response is on its way. */ + drain: () => void; +}): DaemonQuiesce { + let admitting = true; + const refuse = (error: AppError): DaemonResponse => ({ ok: false, error: normalizeError(error) }); + + const quiesce = (req: DaemonRequest): DaemonResponse => { + if (typeof req.token !== 'string' || !timingSafeStringEqual(req.token, params.token)) { + return refuse(new AppError('UNAUTHORIZED', 'Invalid daemon token.')); + } + // A remote client reaches this daemon through the proxy with the shared token, and only the + // host that owns the state directory may retire the daemon on it. + if (req.internal?.publicNetworkOnly) { + return refuse( + new AppError('UNAUTHORIZED', 'Daemon quiesce is not available to remote clients.'), + ); + } + const fence = readDaemonQuiesceFence(req.input); + // A pid alone can name a recycled process, so a daemon that cannot prove its start time + // cannot be addressed at all. + if (!fence || fence.pid !== params.identity.pid) return refuse(instanceMismatch()); + if (params.identity.processStartTime === null) return refuse(instanceMismatch()); + if (fence.processStartTime !== params.identity.processStartTime) { + return refuse(instanceMismatch()); + } + // Admission closes and the census is read in the same synchronous step, so no request can be + // admitted between the two; a refusal reopens admission in that same step. + if (!admitting) return { ok: true, data: { outcome: 'draining' } }; + admitting = false; + const census = readCensus(); + if (isHeld(census)) { + admitting = true; + return refuse( + new AppError('COMMAND_FAILED', 'Daemon is busy and was not asked to leave.', { + reason: DAEMON_QUIESCE_REASONS.busy, + ...census, + }), + ); + } + setImmediate(params.drain); + return { ok: true, data: { outcome: 'draining' } }; + }; + + const readCensus = (): DaemonQuiesceCensus => { + const sessions = params.sessionStore.toArray(); + const repairSessions = sessions.filter(isUncommittedRepairSession).length; + return { + startingUp: params.startingUp(), + inFlightRequests: params.inFlightRequestCount(), + sessions: sessions.length - repairSessions, + repairSessions, + leases: params.leaseRegistry.listActiveLeases().length, + }; + }; + + return { + admit: (handler) => async (req) => { + if (req.command === DAEMON_QUIESCE_COMMAND) return quiesce(req); + return admitting ? await handler(req) : daemonShuttingDownResponse(); + }, + }; +} + +function isHeld(census: DaemonQuiesceCensus): boolean { + return ( + census.startingUp || + census.inFlightRequests > 0 || + census.sessions > 0 || + census.repairSessions > 0 || + census.leases > 0 + ); +} + +function instanceMismatch(): AppError { + return new AppError('COMMAND_FAILED', 'Quiesce request names a different daemon instance.', { + reason: DAEMON_QUIESCE_REASONS.instanceMismatch, + }); +} diff --git a/src/daemon-client/__tests__/daemon-client-metadata.test.ts b/src/daemon-client/__tests__/daemon-client-metadata.test.ts index 63554c782e..7c664f5969 100644 --- a/src/daemon-client/__tests__/daemon-client-metadata.test.ts +++ b/src/daemon-client/__tests__/daemon-client-metadata.test.ts @@ -52,3 +52,14 @@ test('a registration this version did not write reads back unreported', () => { assert.equal(readDaemonInfo(infoPath)?.codeOrigin, undefined); } }); + +test('a daemon record without capabilities reads as a daemon with none', () => { + const [, infoPath] = scratchStateDir(); + fs.writeFileSync(infoPath, JSON.stringify({ httpPort: 41_234, token: 'legacy', pid: 1 })); + assert.deepEqual(readDaemonInfo(infoPath)?.capabilities, []); + fs.writeFileSync( + infoPath, + JSON.stringify({ httpPort: 41_234, token: 'new', pid: 1, capabilities: ['quiesce', 'future'] }), + ); + assert.deepEqual(readDaemonInfo(infoPath)?.capabilities, ['quiesce']); +}); diff --git a/src/daemon-client/daemon-client-metadata.ts b/src/daemon-client/daemon-client-metadata.ts index 5f35583f15..785666401f 100644 --- a/src/daemon-client/daemon-client-metadata.ts +++ b/src/daemon-client/daemon-client-metadata.ts @@ -2,9 +2,10 @@ import fs from 'node:fs'; import { isProcessPid } from '@agent-device/host-kit/process'; import type { DaemonCodeOrigin } from '@agent-device/host-kit/code-signature'; - import { + readDaemonCapabilities, resolveDaemonPaths, + type DaemonCapability, type DaemonPaths, type DaemonServerMode, } from '@agent-device/daemon-contracts/daemon-resolution'; @@ -21,6 +22,8 @@ export type DaemonInfo = { processStartTime?: string; /** ADR 0029: digest of the daemon policy the daemon enforces; absent when it has none. */ policyDigest?: string; + /** Absent from daemons that predate capabilities, which therefore have none. */ + capabilities?: readonly DaemonCapability[]; baseUrl?: string; remoteInstanceId?: string; remoteUpstreamInstanceId?: string; @@ -49,6 +52,7 @@ export function readDaemonInfo(infoPath: string): DaemonInfo | null { codeSignature: readOptionalString(parsed.codeSignature), processStartTime: readOptionalString(parsed.processStartTime), policyDigest: readOptionalString(parsed.policyDigest), + capabilities: readDaemonCapabilities(parsed.capabilities), }; }