diff --git a/src/__tests__/daemon-proxy.test.ts b/src/__tests__/daemon-proxy.test.ts index 6471b9526..5f1409d58 100644 --- a/src/__tests__/daemon-proxy.test.ts +++ b/src/__tests__/daemon-proxy.test.ts @@ -4,6 +4,7 @@ import crypto from 'node:crypto'; import http from 'node:http'; import { createDaemonProxyServer } from '../remote/daemon-proxy.ts'; import { createDaemonHttpServer } from '../daemon/server/http-server.ts'; +import { getRequestSignal } from '@agent-device/host-kit/request'; import { executeRunScriptHttpRequest } from '../daemon/adapters/maestro/run-script-http.ts'; import { DAEMON_HTTP_NETWORK_ACCESS_HEADER, @@ -173,6 +174,139 @@ test('proxy enforces public-only Maestro HTTP policy on a local daemon', async ( } }); +test('daemon proxy cancels the upstream daemon request when its client disconnects', async (t) => { + if (await skipWhenLoopbackUnavailable(t)) return; + + let upstreamStarted!: (requestId: string | undefined) => void; + const started = new Promise((resolve) => { + upstreamStarted = resolve; + }); + let upstreamCanceled!: (reason: string) => void; + const canceled = new Promise((resolve) => { + upstreamCanceled = resolve; + }); + const env = { ...process.env }; + delete env.AGENT_DEVICE_HTTP_AUTH_HOOK; + delete env.AGENT_DEVICE_HTTP_AUTH_EXPORT; + const daemon = await createDaemonHttpServer({ + token: 'daemon-secret', + env, + handleRequest: async (request) => { + const requestId = request.meta?.requestId; + const signal = getRequestSignal(requestId); + upstreamStarted(requestId); + if (!signal) { + upstreamCanceled('no request signal was registered'); + } else if (signal.aborted) { + upstreamCanceled('aborted'); + } else { + signal.addEventListener('abort', () => upstreamCanceled('aborted'), { once: true }); + } + await canceled; + return { ok: false, error: { code: 'COMMAND_FAILED', message: 'request canceled' } }; + }, + }); + const proxy = createDaemonProxyServer({ + upstreamBaseUrl: `http://127.0.0.1:${await listenOnLoopback(daemon)}`, + upstreamToken: 'daemon-secret', + clientToken: 'proxy-secret', + }); + + try { + const proxyPort = await listenOnLoopback(proxy); + const client = http.request({ + host: '127.0.0.1', + port: proxyPort, + method: 'POST', + path: '/agent-device/rpc', + headers: { 'content-type': 'application/json', authorization: 'Bearer proxy-secret' }, + }); + client.on('error', () => {}); + client.end( + JSON.stringify({ + jsonrpc: '2.0', + id: 'req-disconnect', + method: 'agent_device.command', + params: { + token: 'proxy-secret', + session: 'default', + command: 'snapshot', + positionals: [], + flags: {}, + }, + }), + ); + const requestId = await started; + assert.match(String(requestId), /req-disconnect/); + + client.destroy(); + + const timeout = new Promise((resolve) => { + setTimeout(() => resolve('upstream request was never canceled'), 5000).unref(); + }); + assert.equal(await Promise.race([canceled, timeout]), 'aborted'); + } finally { + await closeLoopbackServer(proxy); + await closeLoopbackServer(daemon); + } +}); + +test('daemon proxy forwards a request diagnostics record fetch and nothing else on that path', async (t) => { + if (await skipWhenLoopbackUnavailable(t)) return; + + const upstreamRequests: Array<{ method: string; url: string; auth: string }> = []; + const upstream = http.createServer((req, res) => { + upstreamRequests.push({ + method: req.method ?? '', + url: req.url ?? '', + auth: String(req.headers.authorization ?? ''), + }); + res.setHeader('content-type', 'application/x-ndjson'); + res.end('{"phase":"request_failed"}\n'); + }); + const proxy = createDaemonProxyServer({ + upstreamBaseUrl: `http://127.0.0.1:${await listenOnLoopback(upstream)}`, + upstreamToken: 'daemon-secret', + clientToken: 'proxy-secret', + }); + + try { + const proxyPort = await listenOnLoopback(proxy); + const record = `/agent-device/sessions/default/requests/req%3A1/diagnostics`; + const headers = { authorization: 'Bearer proxy-secret' }; + + const fetched = await fetch(`http://127.0.0.1:${proxyPort}${record}`, { headers }); + assert.equal(fetched.status, 200); + assert.equal(await fetched.text(), '{"phase":"request_failed"}\n'); + assert.deepEqual(upstreamRequests, [ + { + method: 'GET', + url: '/sessions/default/requests/req%3A1/diagnostics', + auth: 'Bearer daemon-secret', + }, + ]); + + const unauthenticated = await fetch(`http://127.0.0.1:${proxyPort}${record}`); + assert.equal(unauthenticated.status, 401); + const posted = await fetch(`http://127.0.0.1:${proxyPort}${record}`, { + method: 'POST', + headers, + }); + assert.equal(posted.status, 404); + const enumerated = await fetch( + `http://127.0.0.1:${proxyPort}/agent-device/sessions/default/requests`, + { + headers, + }, + ); + assert.equal(enumerated.status, 404); + assert.equal(upstreamRequests.length, 1, 'only the record fetch reaches the daemon'); + } finally { + await closeLoopbackServer(proxy); + await closeLoopbackServer(upstream); + } +}); + test('daemon proxy rejects unauthenticated rpc requests', async (t) => { if (await skipWhenLoopbackUnavailable(t)) return; diff --git a/src/daemon-client/__tests__/daemon-client-health-compat.test.ts b/src/daemon-client/__tests__/daemon-client-health-compat.test.ts new file mode 100644 index 000000000..df007d55e --- /dev/null +++ b/src/daemon-client/__tests__/daemon-client-health-compat.test.ts @@ -0,0 +1,94 @@ +import assert from 'node:assert/strict'; +import http from 'node:http'; +import { test } from 'vitest'; +import { DAEMON_RPC_PROTOCOL_VERSION } from '@agent-device/contracts/daemon-http'; +import { readRemoteDaemonHealth } from '../daemon-client-transport.ts'; +import { + closeLoopbackServer, + listenOnLoopback, + skipWhenLoopbackUnavailable, +} from '../../__tests__/test-utils/loopback.ts'; + +/** + * ADR 0006 health compatibility across every link a command RPC crosses. `sendToDaemon` + * runs this check before the RPC (daemon-client.test.ts pins that order); these cases pin + * what the check itself accepts and refuses when a proxy sits in front of the daemon. + */ + +const DAEMON_LINK = { ok: true, service: 'agent-device-daemon', version: '98.0.0' } as const; + +async function withHealthServer( + payload: Record, + run: (baseUrl: string) => Promise, +): Promise { + const server = http.createServer((req, res) => { + assert.equal(req.url, '/agent-device/health'); + res.setHeader('content-type', 'application/json'); + res.end(JSON.stringify(payload)); + }); + try { + const port = await listenOnLoopback(server); + return await run(`http://127.0.0.1:${port}/agent-device`); + } finally { + await closeLoopbackServer(server); + } +} + +function proxyHealth(upstreamRpcProtocolVersion: number): Record { + return { + ok: true, + service: 'agent-device-proxy', + version: '99.0.0', + rpcProtocolVersion: DAEMON_RPC_PROTOCOL_VERSION, + upstream: { ...DAEMON_LINK, rpcProtocolVersion: upstreamRpcProtocolVersion }, + }; +} + +test('a proxy whose daemon speaks the same protocol passes with the upstream link readable', async (t) => { + if (await skipWhenLoopbackUnavailable(t)) return; + + await withHealthServer(proxyHealth(DAEMON_RPC_PROTOCOL_VERSION), async (baseUrl) => { + const health = await readRemoteDaemonHealth({ baseUrl, token: 'proxy-token', pid: 0 }); + assert.equal(health.reachable, true); + assert.equal(health.service, 'agent-device-proxy'); + assert.deepEqual(health.upstream, { + service: 'agent-device-daemon', + version: '98.0.0', + rpcProtocolVersion: DAEMON_RPC_PROTOCOL_VERSION, + }); + }); +}); + +test('a proxy whose daemon speaks another protocol is refused, naming the daemon link', async (t) => { + if (await skipWhenLoopbackUnavailable(t)) return; + + await withHealthServer(proxyHealth(DAEMON_RPC_PROTOCOL_VERSION + 1), async (baseUrl) => { + await assert.rejects( + readRemoteDaemonHealth({ baseUrl, token: 'proxy-token', pid: 0 }), + (error: unknown) => { + const details = (error as { code?: string; details?: Record }).details; + assert.equal((error as { code?: string }).code, 'COMMAND_FAILED'); + assert.match(String((error as Error).message), /RPC protocol is incompatible/); + assert.equal(details?.remoteService, 'agent-device-daemon'); + assert.equal(details?.remoteVersion, '98.0.0'); + assert.equal(details?.remoteRpcProtocolVersion, DAEMON_RPC_PROTOCOL_VERSION + 1); + assert.equal(details?.supportedRpcProtocolVersion, DAEMON_RPC_PROTOCOL_VERSION); + return true; + }, + ); + }); +}); + +test('a daemon health payload without an upstream link parses as before', async (t) => { + if (await skipWhenLoopbackUnavailable(t)) return; + + await withHealthServer( + { ...DAEMON_LINK, rpcProtocolVersion: DAEMON_RPC_PROTOCOL_VERSION }, + async (baseUrl) => { + const health = await readRemoteDaemonHealth({ baseUrl, token: 'daemon-token', pid: 0 }); + assert.equal(health.reachable, true); + assert.equal(health.upstream, undefined); + assert.equal(health.rpcProtocolVersion, DAEMON_RPC_PROTOCOL_VERSION); + }, + ); +}); diff --git a/src/daemon-client/daemon-client-transport.ts b/src/daemon-client/daemon-client-transport.ts index 5cc3137bb..8b81e0619 100644 --- a/src/daemon-client/daemon-client-transport.ts +++ b/src/daemon-client/daemon-client-transport.ts @@ -36,8 +36,15 @@ export type RemoteDaemonHealth = { service?: string; version?: string; rpcProtocolVersion?: number; + /** The daemon behind a proxy, as the proxy's health reported it. */ + upstream?: RemoteDaemonHealthLink; }; +type RemoteDaemonHealthLink = Pick< + RemoteDaemonHealth, + 'service' | 'version' | 'rpcProtocolVersion' +>; + export async function canConnect( info: DaemonInfo, preference: DaemonTransportPreference, @@ -86,17 +93,21 @@ function canConnectHttp(info: DaemonInfo): Promise { export async function readRemoteDaemonHealth(info: DaemonInfo): Promise { const health = await readDaemonHttpHealth(info); if (!info.baseUrl || !health.reachable) return health; - if ( - typeof health.rpcProtocolVersion === 'number' && - health.rpcProtocolVersion !== DAEMON_RPC_PROTOCOL_VERSION - ) { + // Every link a command RPC crosses has to speak the client's protocol: a proxy that reports a + // skewed daemon behind it fails here, before the RPC, exactly like a skewed proxy does. + const incompatible = [health, health.upstream].find( + (link) => + typeof link?.rpcProtocolVersion === 'number' && + link.rpcProtocolVersion !== DAEMON_RPC_PROTOCOL_VERSION, + ); + if (incompatible) { throw new AppError('COMMAND_FAILED', 'Remote daemon RPC protocol is incompatible', { daemonBaseUrl: info.baseUrl, clientVersion: readVersion(), - remoteVersion: health.version, - remoteService: health.service, + remoteVersion: incompatible.version, + remoteService: incompatible.service, supportedRpcProtocolVersion: DAEMON_RPC_PROTOCOL_VERSION, - remoteRpcProtocolVersion: health.rpcProtocolVersion, + remoteRpcProtocolVersion: incompatible.rpcProtocolVersion, hint: 'Upgrade agent-device on the client or remote host so both support the same daemon RPC protocol.', }); } @@ -156,22 +167,29 @@ async function readDaemonHttpHealth(info: DaemonInfo): Promise { try { - const parsed = JSON.parse(body) as { - service?: unknown; - version?: unknown; - rpcProtocolVersion?: unknown; - }; + const parsed = JSON.parse(body) as { upstream?: unknown }; + const upstream = + parsed.upstream && typeof parsed.upstream === 'object' + ? readHealthLink(parsed.upstream as Record) + : undefined; return { - service: typeof parsed.service === 'string' ? parsed.service : undefined, - version: typeof parsed.version === 'string' ? parsed.version : undefined, - rpcProtocolVersion: - typeof parsed.rpcProtocolVersion === 'number' ? parsed.rpcProtocolVersion : undefined, + ...readHealthLink(parsed as Record), + ...(upstream ? { upstream } : {}), }; } catch { return {}; } } +function readHealthLink(parsed: Record): RemoteDaemonHealthLink { + return { + service: typeof parsed.service === 'string' ? parsed.service : undefined, + version: typeof parsed.version === 'string' ? parsed.version : undefined, + rpcProtocolVersion: + typeof parsed.rpcProtocolVersion === 'number' ? parsed.rpcProtocolVersion : undefined, + }; +} + export async function sendRequest( info: DaemonInfo, req: DaemonRequest, diff --git a/src/remote/daemon-proxy.ts b/src/remote/daemon-proxy.ts index a460e6eb9..44e9f3878 100644 --- a/src/remote/daemon-proxy.ts +++ b/src/remote/daemon-proxy.ts @@ -133,13 +133,38 @@ async function forwardProxyRequest(params: { const response = await options.fetchImpl(upstreamUrl, { method, headers, - signal: AbortSignal.timeout(options.upstreamTimeoutMs), + signal: upstreamRequestSignal(req, res, options.upstreamTimeoutMs), ...(body ? { body, duplex: 'half' as const } : {}), }); await sendProxyResponse({ req, res, route, response, clientToken: options.clientToken }); } +/** + * The upstream request lives exactly as long as the client keeps waiting for it. The daemon's + * HTTP boundary turns a vanished client into request cancellation, so the proxy has to drop its + * own upstream socket for a remote client's disconnect to reach in-flight runner work. + */ +function upstreamRequestSignal( + req: IncomingMessage, + res: ServerResponse, + timeoutMs: number, +): AbortSignal { + const clientGone = new AbortController(); + const abortIfResponseIncomplete = () => { + if (res.writableFinished || clientGone.signal.aborted) return; + clientGone.abort( + new AppError( + 'COMMAND_FAILED', + 'Proxy client disconnected before the upstream response ended', + ), + ); + }; + req.on('aborted', abortIfResponseIncomplete); + res.on('close', abortIfResponseIncomplete); + return AbortSignal.any([clientGone.signal, AbortSignal.timeout(timeoutMs)]); +} + async function sendProxyResponse(params: { req: IncomingMessage; res: ServerResponse; @@ -304,9 +329,27 @@ function isSupportedDaemonRoute(route: string, method: string | undefined): bool if (isSupportedUploadRoute(route, method)) return true; if (route === '/artifacts' || route === '/artifacts/') return method === 'GET'; if (route.startsWith('/artifacts/')) return method === 'GET'; + if (isRequestDiagnosticsRoute(route)) return method === 'GET'; return false; } +/** + * `GET /sessions//requests//diagnostics` (#1801): the record a failed + * command names. A remote client localizes its `logPath` from it, so a client behind the + * proxy keeps exactly the failure envelope a client on the daemon host gets. + */ +function isRequestDiagnosticsRoute(route: string): boolean { + const segments = route.split('/'); + return ( + segments.length === 6 && + segments[1] === 'sessions' && + segments[3] === 'requests' && + segments[5] === 'diagnostics' && + segments[2] !== '' && + segments[4] !== '' + ); +} + function isSupportedUploadRoute(route: string, method: string | undefined): boolean { if (route === '/upload') return method === 'POST'; if (isUploadPreflightRoute(route)) return method === 'POST'; @@ -439,6 +482,7 @@ function sendUnauthorized(res: ServerResponse, route: string, rpcId: unknown): v } function sendProxyError(res: ServerResponse, error: unknown): void { + if (res.destroyed) return; if (res.headersSent) { res.destroy(error instanceof Error ? error : undefined); return; diff --git a/test/integration/provider-scenarios/harness.ts b/test/integration/provider-scenarios/harness.ts index de234bdf1..72ac5f1f5 100644 --- a/test/integration/provider-scenarios/harness.ts +++ b/test/integration/provider-scenarios/harness.ts @@ -21,7 +21,11 @@ import type { AppleSimulatorScreenRecordingProcess } from '../../../src/platform import { trackDownloadableArtifact } from '../../../src/daemon/artifact-tracking.ts'; import { LeaseRegistry } from '../../../src/daemon/lease-registry.ts'; import { SessionStore } from '../../../src/daemon/session-store.ts'; -import type { DaemonRequest, DaemonResponse } from '../../../src/daemon/daemon-request.ts'; +import type { + DaemonInvokeFn, + DaemonRequest, + DaemonResponse, +} from '../../../src/daemon/daemon-request.ts'; import type { SessionState } from '../../../src/daemon/session-state.ts'; import { runCmdBackground } from '@agent-device/host-kit/command'; import { createOwnedProcessRecordStore } from '@agent-device/host-kit/process'; @@ -65,6 +69,9 @@ export type ProviderScenarioHarness = { }, ) => Promise; client: () => AgentDeviceClient; + /** The scenario daemon's request boundary, for mounting it behind a real HTTP server. */ + handleRequest: DaemonInvokeFn; + token: string; session: (name?: string) => SessionState | undefined; sessionDir: (name?: string) => string; setSession: (name: string, session: SessionState) => void; @@ -209,6 +216,8 @@ export async function createProviderScenarioHarness( `direct-${command}-${Date.now()}`, ), client: () => createAgentDeviceClient({}, { transport }), + handleRequest, + token: PROVIDER_SCENARIO_TOKEN, session: (name = 'default') => sessionStore.get(name), sessionDir: (name = 'default') => sessionStore.resolveSessionDir(name), setSession: (name, session) => sessionStore.set(name, session), diff --git a/test/integration/provider-scenarios/remote-proxy-parity.test.ts b/test/integration/provider-scenarios/remote-proxy-parity.test.ts new file mode 100644 index 000000000..20214e473 --- /dev/null +++ b/test/integration/provider-scenarios/remote-proxy-parity.test.ts @@ -0,0 +1,550 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { test } from 'vitest'; +import { DEFAULT_PROXY_LEASE_TTL_MS } from '../../../src/core/lease-scope.ts'; +import { sendToDaemon } from '../../../src/daemon-client/daemon-client.ts'; +import { LeaseRegistry } from '../../../src/daemon/lease-registry.ts'; +import { createDaemonHttpServer } from '../../../src/daemon/server/http-server.ts'; +import { resolveSessionRequestLogPath } from '../../../src/daemon/session-artifact-paths.ts'; +import type { DaemonRequest, DaemonResponse } from '../../../src/daemon/daemon-request.ts'; +import { createDaemonProxyServer } from '../../../src/remote/daemon-proxy.ts'; +import { AppError, type DaemonError } from '@agent-device/kernel/errors'; +import { + closeLoopbackServer, + listenOnLoopback, + skipWhenLoopbackUnavailable, +} from '../../../src/__tests__/test-utils/loopback.ts'; +import { PROVIDER_SCENARIO_IOS_SIMULATOR } from './fixtures.ts'; +import { createProviderScenarioHarness, type ProviderScenarioHarness } from './harness.ts'; +import { + createAppleRunnerProviderFromTranscript, + createRecordingAppleToolProvider, + simctlDeviceLifecycleHandler, + type FlatToolCall, +} from './providers.ts'; +import { PARALLEL_PROVIDER_SCENARIO_TIMEOUT_MS } from './test-timeouts.ts'; +import { createProviderTranscript, type ProviderScenarioTranscript } from './transcript.ts'; + +/** + * #2198 slice B: the proxy publishes exactly what the daemon publishes. Both legs run the same + * script over the same deterministic Simulator acquisition fixture; the only things allowed to + * differ are transport timing and request identity. + */ + +type ScenarioRequest = Omit; +type ScenarioLeg = (request: ScenarioRequest) => Promise; + +const SIM = PROVIDER_SCENARIO_IOS_SIMULATOR; +const APP = 'com.apple.Preferences'; +const PROXY_TOKEN = 'proxy-parity-token'; +const VOLATILE_KEYS = new Set([ + 'requestId', + 'diagnosticId', + 'logPath', + 'timing', + 'durationMs', + 'elapsedMs', + 'timestamp', + 'capturedAt', + 'startedAt', + 'completedAt', + 'measuredAt', + 'maxMs', + 'p50Ms', + 'p95Ms', + // Session identity minted per open; refs are compared, the generation stamp is not. + 'refsGeneration', +]); +/** Fields the wire may re-home (paths) or re-issue (ids) but must never lose. */ +const PRESERVED_ERROR_KEYS = ['hint', 'details', 'diagnosticId', 'logPath'] as const; + +type ParityWorld = { + daemon: ProviderScenarioHarness; + appleTool: { calls: FlatToolCall[] }; + runnerTranscript: ProviderScenarioTranscript; + close: () => Promise; +}; + +function scriptedTree() { + return { + nodes: [ + { + index: 0, + type: 'XCUIElementTypeCell', + label: 'General', + identifier: 'General', + rect: { x: 16, y: 100, width: 360, height: 44 }, + enabled: true, + hittable: true, + }, + { + index: 1, + type: 'XCUIElementTypeApplication', + label: 'Settings', + identifier: APP, + rect: { x: 0, y: 0, width: 393, height: 852 }, + enabled: true, + hittable: true, + }, + ], + truncated: false, + }; +} + +type ParityWorldOptions = { leaseRegistry?: LeaseRegistry }; + +async function createParityWorld(options: ParityWorldOptions = {}): Promise { + const runnerTranscript = createProviderTranscript([ + { + command: 'ios.runner.snapshot', + deviceId: SIM.id, + platform: 'apple', + repeat: true, + result: scriptedTree, + }, + ]); + const appleTool = createRecordingAppleToolProvider({ + simctl: simctlDeviceLifecycleHandler('com.apple.CoreSimulator.SimRuntime.iOS-18-0', [ + { name: SIM.name, udid: SIM.id }, + ]), + }); + const daemon = await createProviderScenarioHarness({ + ...(options.leaseRegistry ? { leaseRegistry: options.leaseRegistry } : {}), + platformRuntime: true, + appleRunnerProvider: () => + createAppleRunnerProviderFromTranscript(runnerTranscript, 'ios.runner'), + appleToolProvider: () => appleTool.provider, + deviceInventoryProvider: async () => [SIM], + }); + return { daemon, appleTool, runnerTranscript, close: () => daemon.close() }; +} + +function parityScript(): readonly ScenarioRequest[] { + const flags = { platform: 'ios', udid: SIM.id } as const; + return [ + { session: 'default', command: 'open', positionals: [APP], flags }, + { + session: 'default', + command: 'snapshot', + positionals: [], + flags: { snapshotInteractiveOnly: true }, + }, + { session: 'default', command: 'snapshot', positionals: [], flags: { snapshotRaw: true } }, + { + session: 'default', + command: 'diff', + positionals: ['snapshot'], + flags: { snapshotInteractiveOnly: true }, + }, + { session: 'default', command: 'click', positionals: ['@e404'], flags: {} }, + { session: 'default', command: 'appstate', positionals: [], flags }, + { session: 'default', command: 'close', positionals: [], flags: {} }, + // A fresh session after cleanup starts its comparison state from nothing. + { session: 'default', command: 'open', positionals: [APP], flags }, + { + session: 'default', + command: 'diff', + positionals: ['snapshot'], + flags: { snapshotInteractiveOnly: true }, + }, + { session: 'default', command: 'close', positionals: [], flags: {} }, + ]; +} + +/** + * The published form of a response with transport identity removed: JSON drops `undefined` + * exactly like the wire does, every path under the world's session root becomes the same + * token, and per-request log files lose the request id in their name. + */ +function stripVolatile(value: unknown, sessionRoot: string): unknown { + return stripVolatileKeys(JSON.parse(JSON.stringify(value)), sessionRoot); +} + +function stripVolatileKeys(value: unknown, sessionRoot: string): unknown { + if (Array.isArray(value)) return value.map((entry) => stripVolatileKeys(entry, sessionRoot)); + if (typeof value === 'string') { + return value + .split(sessionRoot) + .join('') + .replace(/\/requests\/[^/]+\.ndjson$/, '/requests/.ndjson'); + } + if (!value || typeof value !== 'object') return value; + return Object.fromEntries( + Object.entries(value as Record) + .filter(([key]) => !VOLATILE_KEYS.has(key)) + .map(([key, entry]) => [key, stripVolatileKeys(entry, sessionRoot)]), + ); +} + +function sessionRootOf(daemon: ProviderScenarioHarness): string { + return path.dirname(daemon.sessionDir()); +} + +const ERROR_ENVELOPE_KEYS = [ + 'hint', + 'diagnosticId', + 'logPath', + 'logPathUnavailable', + 'diagnosticsRecord', + 'retriable', + 'supportedOn', + 'requestId', +] as const; + +/** + * The remote client raises daemon failures as `AppError`s whose details carry the error + * envelope; fold that back into the daemon's own response shape so both legs compare alike. + */ +function responseFromClientError(error: unknown): DaemonResponse { + if (!(error instanceof AppError)) throw error; + const details: Record = { ...(error.details ?? {}) }; + const envelope: Partial> = {}; + for (const key of ERROR_ENVELOPE_KEYS) { + if (key in details) { + envelope[key] = details[key]; + delete details[key]; + } + } + delete envelope.requestId; + return { + ok: false, + error: { + code: error.code, + message: error.message, + ...(Object.keys(details).length > 0 ? { details } : {}), + ...envelope, + } as DaemonError, + }; +} + +async function runLeg(leg: ScenarioLeg, legName: string): Promise { + const responses: DaemonResponse[] = []; + for (const [index, request] of parityScript().entries()) { + responses.push(await leg({ ...request, meta: { requestId: `${legName}-${index + 1}` } })); + } + return responses; +} + +async function withProxiedWorld( + run: (context: { world: ParityWorld; proxied: ScenarioLeg }) => Promise, + options: ParityWorldOptions = {}, +): Promise { + const world = await createParityWorld(options); + const upstream = await createDaemonHttpServer({ + token: world.daemon.token, + handleRequest: world.daemon.handleRequest, + // Same composition as the daemon runtime: remote clients localize failure records. + resolveRequestDiagnosticsPath: (ref) => + resolveSessionRequestLogPath(world.daemon.sessionDir(ref.session), ref.requestId), + }); + const proxy = createDaemonProxyServer({ + upstreamBaseUrl: `http://127.0.0.1:${await listenOnLoopback(upstream)}`, + upstreamToken: world.daemon.token, + clientToken: PROXY_TOKEN, + }); + // The remote client's own state dir: localized failure records land here, not in the + // developer's real state dir. + const clientStateDir = fs.mkdtempSync(path.join(os.tmpdir(), 'agent-device-proxy-parity-')); + try { + const daemonBaseUrl = `http://127.0.0.1:${await listenOnLoopback(proxy)}/agent-device`; + const proxied: ScenarioLeg = async (request) => { + try { + return await sendToDaemon( + { ...request, flags: { ...request.flags, daemonBaseUrl, stateDir: clientStateDir } }, + { authToken: PROXY_TOKEN }, + ); + } catch (error) { + return responseFromClientError(error); + } + }; + return await run({ world, proxied }); + } finally { + await closeLoopbackServer(proxy); + await closeLoopbackServer(upstream); + await world.close(); + fs.rmSync(clientStateDir, { recursive: true, force: true }); + } +} + +type LegRun = { responses: DaemonResponse[]; sessionRoot: string }; + +function assertPublishedParity(direct: LegRun, proxied: LegRun): void { + const script = parityScript(); + assert.equal(proxied.responses.length, direct.responses.length); + direct.responses.forEach((directResponse, index) => { + const proxiedResponse = proxied.responses[index]; + const step = `${index + 1}:${script[index]?.command}`; + assert.deepEqual( + stripVolatile(proxiedResponse, proxied.sessionRoot), + stripVolatile(directResponse, direct.sessionRoot), + `proxy diverged from direct execution at step ${step}`, + ); + if (directResponse.ok || !proxiedResponse || proxiedResponse.ok) return; + assertErrorEnvelopeKept(directResponse.error, proxiedResponse.error, step); + }); +} + +/** + * The remote client may add to the envelope (it materializes the failure record locally, + * which yields a logPath the daemon never published); it may not lose anything. + */ +function assertErrorEnvelopeKept(direct: DaemonError, proxied: DaemonError, step: string): void { + for (const key of PRESERVED_ERROR_KEYS) { + if (direct[key] === undefined) continue; + assert.equal( + typeof proxied[key], + typeof direct[key], + `proxy lost error.${key} at step ${step}`, + ); + } +} + +function baselineInitialized(response: DaemonResponse | undefined): boolean | undefined { + if (!response?.ok) return undefined; + return (response.data as { baselineInitialized?: boolean } | undefined)?.baselineInitialized; +} + +/** The script exercised what it claims to: successes, one typed failure, a fresh baseline. */ +function assertScriptOutcomes(responses: DaemonResponse[]): void { + const [open, snapshot, raw, diff, missingRef, , , , freshDiff] = responses; + assert.equal(open?.ok, true); + assert.equal(snapshot?.ok, true); + assert.equal(raw?.ok, true); + assert.equal(baselineInitialized(diff), false); + assert.equal(missingRef?.ok, false); + assert.equal( + baselineInitialized(freshDiff), + true, + 'a session reopened after cleanup must not compare against the previous session tree', + ); +} + +test( + 'Provider-backed integration proxy execution publishes what direct daemon execution publishes', + async (t) => { + if (await skipWhenLoopbackUnavailable(t, 'daemon proxy parity coverage')) return; + + const directWorld = await createParityWorld(); + let direct: LegRun; + try { + direct = { + responses: await runLeg( + async (request) => + await directWorld.daemon.handleRequest({ ...request, token: directWorld.daemon.token }), + 'direct', + ), + sessionRoot: sessionRootOf(directWorld.daemon), + }; + } finally { + await directWorld.close(); + } + + const proxied = await withProxiedWorld(async ({ world, proxied: leg }) => ({ + responses: await runLeg(leg, 'proxied'), + sessionRoot: sessionRootOf(world.daemon), + })); + + assertPublishedParity(direct, proxied); + assertScriptOutcomes(direct.responses); + }, + PARALLEL_PROVIDER_SCENARIO_TIMEOUT_MS, +); + +test( + 'Provider-backed integration proxy clients contending for one device fail at lease admission', + async (t) => { + if (await skipWhenLoopbackUnavailable(t, 'daemon proxy parity coverage')) return; + + await withProxiedWorld(async ({ world, proxied }) => { + const flags = { platform: 'ios', udid: SIM.id } as const; + const first = await proxied({ + session: 'first', + command: 'open', + positionals: [APP], + flags, + meta: { cwd: '/workspace/first' }, + }); + assert.equal(first.ok, true, JSON.stringify(first)); + + const callsBefore = world.appleTool.calls.length; + const remainingBefore = world.runnerTranscript.remaining.length; + const second = await proxied({ + session: 'second', + command: 'open', + positionals: [APP], + flags, + meta: { cwd: '/workspace/second' }, + }); + assert.equal(second.ok, false); + if (second.ok) return; + assert.equal(second.error.code, 'DEVICE_IN_USE'); + assert.equal(typeof second.error.hint, 'string'); + + const callsAfter = world.appleTool.calls.slice(callsBefore); + assert.deepEqual( + callsAfter.filter(([, subcommand]) => subcommand !== 'list'), + [], + `the refused open must not reach platform work: ${JSON.stringify(callsAfter)}`, + ); + assert.equal(world.runnerTranscript.remaining.length, remainingBefore); + + const close = await proxied({ + session: 'first', + command: 'close', + positionals: [], + flags: {}, + meta: { cwd: '/workspace/first' }, + }); + assert.equal(close.ok, true, JSON.stringify(close)); + }); + }, + PARALLEL_PROVIDER_SCENARIO_TIMEOUT_MS, +); + +const LEASE_SCOPE = { + tenantId: 'team-a', + runId: 'run-a', + clientId: 'client-a', + deviceKey: SIM.id, + leaseBackend: 'ios-simulator', +} as const; + +test( + 'Provider-backed integration proxy lease expiry tears the session down and a reacquired lease starts with no comparison state', + async (t) => { + if (await skipWhenLoopbackUnavailable(t, 'daemon proxy parity coverage')) return; + + let now = 1_000_000; + const leaseRegistry = new LeaseRegistry({ now: () => now }); + await withProxiedWorld( + async ({ world, proxied }) => { + const session = 'leased'; + const flags = { platform: 'ios', udid: SIM.id } as const; + const allocate = async (): Promise => { + const response = await proxied({ + session, + command: 'lease_allocate', + positionals: [], + flags: {}, + meta: LEASE_SCOPE, + }); + assert.equal(response.ok, true, JSON.stringify(response)); + const leaseId = (response.ok ? response.data : {})?.lease as { leaseId?: string }; + assert.equal(typeof leaseId?.leaseId, 'string'); + return leaseId.leaseId!; + }; + const run = async ( + command: string, + positionals: string[], + leaseId: string, + stepFlags: DaemonRequest['flags'] = flags, + ) => + await proxied({ + session, + command, + positionals, + flags: stepFlags, + meta: { ...LEASE_SCOPE, leaseId }, + }); + + const firstLease = await allocate(); + assert.equal((await run('open', [APP], firstLease)).ok, true); + assert.equal( + (await run('snapshot', [], firstLease, { snapshotInteractiveOnly: true })).ok, + true, + ); + assert.equal( + baselineInitialized( + await run('diff', ['snapshot'], firstLease, { snapshotInteractiveOnly: true }), + ), + false, + 'the leased session holds comparison state before it expires', + ); + + // The lease lapses without a heartbeat; the next request through the proxy finds it expired. + now += DEFAULT_PROXY_LEASE_TTL_MS + 1; + const expired = await run('diff', ['snapshot'], firstLease, { + snapshotInteractiveOnly: true, + }); + assert.equal(expired.ok, false); + if (expired.ok) return; + assert.equal(expired.error.code, 'UNAUTHORIZED'); + assert.equal(expired.error.details?.reason, 'LEASE_NOT_FOUND'); + assert.equal(typeof expired.error.hint, 'string'); + assert.equal(world.daemon.session(session), undefined, 'expiry tears the session down'); + + const secondLease = await allocate(); + assert.notEqual(secondLease, firstLease); + assert.equal((await run('open', [APP], secondLease)).ok, true); + assert.equal( + baselineInitialized( + await run('diff', ['snapshot'], secondLease, { snapshotInteractiveOnly: true }), + ), + true, + 'a reacquired lease must not compare against the expired session tree', + ); + assert.equal((await run('close', [], secondLease, {})).ok, true); + }, + { leaseRegistry }, + ); + }, + PARALLEL_PROVIDER_SCENARIO_TIMEOUT_MS, +); + +test( + 'Provider-backed integration proxy lease heartbeat renews the lease and keeps the session and its comparison state', + async (t) => { + if (await skipWhenLoopbackUnavailable(t, 'daemon proxy parity coverage')) return; + + let now = 2_000_000; + const leaseRegistry = new LeaseRegistry({ now: () => now }); + await withProxiedWorld( + async ({ world, proxied }) => { + const session = 'renewed'; + const flags = { platform: 'ios', udid: SIM.id } as const; + const allocated = await proxied({ + session, + command: 'lease_allocate', + positionals: [], + flags: {}, + meta: LEASE_SCOPE, + }); + assert.equal(allocated.ok, true, JSON.stringify(allocated)); + const lease = (allocated.ok ? allocated.data : {})?.lease as { leaseId: string }; + const meta = { ...LEASE_SCOPE, leaseId: lease.leaseId }; + const run = async ( + command: string, + positionals: string[], + stepFlags: DaemonRequest['flags'], + ) => await proxied({ session, command, positionals, flags: stepFlags, meta }); + + assert.equal((await run('open', [APP], flags)).ok, true); + assert.equal((await run('snapshot', [], { snapshotInteractiveOnly: true })).ok, true); + const heartbeatExpiry = async (): Promise => { + const heartbeat = await run('lease_heartbeat', [], {}); + assert.equal(heartbeat.ok, true, JSON.stringify(heartbeat)); + const renewed = (heartbeat.ok ? heartbeat.data : {})?.lease as { expiresAt: number }; + return renewed.expiresAt; + }; + + // One explicit heartbeat through the proxy just before the lease would lapse moves the + // expiry forward; a request past the old expiry but inside the new window still finds + // the session and its diff baseline. + const firstExpiry = await heartbeatExpiry(); + now = firstExpiry - 1_000; + const renewedExpiry = await heartbeatExpiry(); + assert.ok(renewedExpiry > firstExpiry, 'the heartbeat moved the expiry forward'); + now = firstExpiry + 1_000; + const diff = await run('diff', ['snapshot'], { snapshotInteractiveOnly: true }); + assert.equal(diff.ok, true, JSON.stringify(diff)); + assert.equal(baselineInitialized(diff), false, 'the renewed session kept its baseline'); + assert.notEqual(world.daemon.session(session), undefined); + assert.equal((await run('close', [], {})).ok, true); + }, + { leaseRegistry }, + ); + }, + PARALLEL_PROVIDER_SCENARIO_TIMEOUT_MS, +); diff --git a/test/wire-compat/ledger.json b/test/wire-compat/ledger.json index d13ec1847..57ea07121 100644 --- a/test/wire-compat/ledger.json +++ b/test/wire-compat/ledger.json @@ -61,10 +61,12 @@ "src/daemon-client/daemon-client-rpc.ts#rejectDaemonHttpRpcError": "sha256:83b0312fe88bc3b3497de799e23d8d14455f665b7cf880f4617cd62b181351cd", "src/daemon-client/daemon-client-rpc.ts#resolveDaemonHttpResult": "sha256:296f9d376ce67c8cb20209bdf1c59c2423ffcfa35b5565a99e70271263149048", "src/daemon-client/daemon-client-rpc.ts#toDaemonHttpRpcError": "sha256:888246763c48670e7da893054d025744654f8715c3b4906312617a2b5028316b", - "src/daemon-client/daemon-client-transport.ts#RemoteDaemonHealth": "sha256:38fe712390d247de59a74b9dc209fdf407ff5f9b1e32b698ae71ebc57569b66d", + "src/daemon-client/daemon-client-transport.ts#RemoteDaemonHealth": "sha256:e1d2251a5bc1607539ec4fbe73c4a9b2dd5fa2859293c442565eb457db47f42f", + "src/daemon-client/daemon-client-transport.ts#RemoteDaemonHealthLink": "sha256:e145d58c1b8c209cdc95ed9e3540b5ffbb9c2ac2eedce3bb1b34ecd7b7bd9dd0", "src/daemon-client/daemon-client-transport.ts#readDaemonHttpHealth": "sha256:5e75af39e96045ce7faba986c900744cee76cdaa1e0df360ea38f3c020520711", - "src/daemon-client/daemon-client-transport.ts#readHealthPayload": "sha256:5b15e14319b16aebf32d07336986a7f1f2a6b13cec1823ca9aa7c9f01bf1b0b6", - "src/daemon-client/daemon-client-transport.ts#readRemoteDaemonHealth": "sha256:b22111f693ecb65705195a66bce617e58441483ebfd67cf264d6900ba1dcb401", + "src/daemon-client/daemon-client-transport.ts#readHealthLink": "sha256:7f020e3f8bf11f6286d4c466d83606a9819a3a6d4f7db70b427fa8835ce0c5be", + "src/daemon-client/daemon-client-transport.ts#readHealthPayload": "sha256:4e85ffc3e35e02379c393e9312344757e003cf1f0ad9eb8d1f77d90c81c861f1", + "src/daemon-client/daemon-client-transport.ts#readRemoteDaemonHealth": "sha256:d833b61b242d0f594d69282b2de2362066c1bcfee2f0c687bd2394d585550c22", "src/daemon/downloadable-artifact-http.ts#DownloadableArtifactHttpAuthorizer": "sha256:1b2702a929ca9170db2ca97c08e3ab67e17edb3ee75325a576c4c1b9cdbebb44", "src/daemon/downloadable-artifact-http.ts#DownloadableArtifactHttpRoute": "sha256:e63c4581ccde668913914149c9092d16ecf8a5cbd7ab33c6eb8617e77fc1015e", "src/daemon/downloadable-artifact-http.ts#handleArtifactDownload": "sha256:7f96d17b7c605230fb3cc21ceaa5b2e515d7445653ff95b2b0f6f15fa6214d4e", @@ -178,7 +180,7 @@ { "declaration": "packages/contracts/src/daemon-http.ts#buildDaemonHealthPayload", "digest": "sha256:709899c02b2a5f6ec2a995350ac7b5791d67db12a631b109af5074427291edda", - "rationale": "#2318 moves the daemon HTTP wire contract into @agent-device/contracts and hands buildDaemonHealthPayload the version its caller advertises instead of reading the package version itself. The added parameter is caller-local \u2014 each side passes its own readVersion() \u2014 and the /health payload a released peer sends or parses is byte-identical." + "rationale": "#2318 moves the daemon HTTP wire contract into @agent-device/contracts and hands buildDaemonHealthPayload the version its caller advertises instead of reading the package version itself. The added parameter is caller-local — each side passes its own readVersion() — and the /health payload a released peer sends or parses is byte-identical." }, { "declaration": "packages/kernel/src/contracts.ts#DaemonArtifactKnownType", @@ -188,7 +190,7 @@ { "declaration": "src/remote/daemon-artifacts.ts#DownloadRemoteArtifactParams", "digest": "sha256:3ec63be2dbdb542e30b19d1500bc16874607823afb7ae392e92b0b535865859e", - "rationale": "#2246 adds the optional isDirectory field. It is pure client-local state \u2014 never serialized, never sent to the daemon \u2014 that tells the CLIENT'S OWN download logic to extract a tar body instead of writing it verbatim; the GET /artifacts/:id request and response framing are unchanged. Every existing call site (screenshot, recording) omits it and keeps writing a single file exactly as before." + "rationale": "#2246 adds the optional isDirectory field. It is pure client-local state — never serialized, never sent to the daemon — that tells the CLIENT'S OWN download logic to extract a tar body instead of writing it verbatim; the GET /artifacts/:id request and response framing are unchanged. Every existing call site (screenshot, recording) omits it and keeps writing a single file exactly as before." }, { "declaration": "src/remote/daemon-artifacts.ts#downloadRemoteArtifact", @@ -254,6 +256,21 @@ "declaration": "src/daemon-client/daemon-client-rpc.ts#appErrorFromDaemonError", "digest": "sha256:ac4761006c71d93ccba9f8e37cbd4cf48283dbb22737c163dc9c1a878e154eea", "rationale": "#1862 client-side only: this rehydrates the new optional structured `cause` when present. Released daemons that omit it follow the unchanged path, and no request field or existing response field is narrowed." + }, + { + "declaration": "src/daemon-client/daemon-client-transport.ts#RemoteDaemonHealth", + "digest": "sha256:e1d2251a5bc1607539ec4fbe73c4a9b2dd5fa2859293c442565eb457db47f42f", + "rationale": "#2198 slice B teaches the client to read the `upstream` link a proxy's /health already nests. The field is optional and additive: a daemon or proxy that does not send it parses exactly as before, and no request shape changes. RemoteDaemonHealth gains the optional `upstream` link; every released field stays." + }, + { + "declaration": "src/daemon-client/daemon-client-transport.ts#readHealthPayload", + "digest": "sha256:4e85ffc3e35e02379c393e9312344757e003cf1f0ad9eb8d1f77d90c81c861f1", + "rationale": "#2198 slice B teaches the client to read the `upstream` link a proxy's /health already nests. The field is optional and additive: a daemon or proxy that does not send it parses exactly as before, and no request shape changes. readHealthPayload reads the same top-level fields through readHealthLink and additionally the nested `upstream` object when present; a payload without it yields the previous shape." + }, + { + "declaration": "src/daemon-client/daemon-client-transport.ts#readRemoteDaemonHealth", + "digest": "sha256:d833b61b242d0f594d69282b2de2362066c1bcfee2f0c687bd2394d585550c22", + "rationale": "#2198 slice B: the ADR 0006 refusal now covers every link a command RPC crosses, so a proxy whose daemon speaks another protocol version fails at health, before the RPC, exactly like a skewed proxy. The comparison against DAEMON_RPC_PROTOCOL_VERSION is unchanged and strictly wider; a compatible chain passes as before." } ] } diff --git a/test/wire-compat/surface.ts b/test/wire-compat/surface.ts index 4cbfb8512..0c333f6ed 100644 --- a/test/wire-compat/surface.ts +++ b/test/wire-compat/surface.ts @@ -106,7 +106,9 @@ export const WIRE_SURFACE: readonly WireSurfaceGroup[] = [ ...from( CLIENT_TRANSPORT, 'RemoteDaemonHealth', + 'RemoteDaemonHealthLink', 'readHealthPayload', + 'readHealthLink', 'readDaemonHttpHealth', 'readRemoteDaemonHealth', ), diff --git a/test/wire-compat/wire-mutations.test.ts b/test/wire-compat/wire-mutations.test.ts index 0c868f334..6021400a8 100644 --- a/test/wire-compat/wire-mutations.test.ts +++ b/test/wire-compat/wire-mutations.test.ts @@ -146,7 +146,7 @@ const MUTATIONS: readonly WireMutation[] = [ { breakClass: 'health consumer: the client stops reading the advertised protocol version', file: 'src/daemon-client/daemon-client-transport.ts', - name: 'readHealthPayload', + name: 'readHealthLink', from: "typeof parsed.rpcProtocolVersion === 'number' ? parsed.rpcProtocolVersion : undefined", to: 'undefined', }, @@ -154,7 +154,7 @@ const MUTATIONS: readonly WireMutation[] = [ breakClass: 'health consumer: the mismatch refusal ADR 0006 built is weakened', file: 'src/daemon-client/daemon-client-transport.ts', name: 'readRemoteDaemonHealth', - from: 'health.rpcProtocolVersion !== DAEMON_RPC_PROTOCOL_VERSION', + from: 'link.rpcProtocolVersion !== DAEMON_RPC_PROTOCOL_VERSION', to: 'false', }, {