diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index b024c5e85..697d33741 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -35,7 +35,7 @@ edit yourself keeps the mode you give it until the next `config set`; run | `gateway.url` | **Worker side.** Base URL of the gateway this worker joins, e.g. `wss://gw.example:4700`; the worker dials `/v1/uplink` and upgrades it to a WebSocket. Joining grants that gateway the `admin` role on this daemon, and `gateway.token` rides on the upgrade request as a bearer credential, so use `wss://` — or plain `ws://` only over loopback or inside your own tunnel. `http://` and `https://` are rejected at load. Unset means "do not join a fleet": the default, and the only thing that changes about a joined worker. | unset | | `gateway.token` | **Worker side.** The join token (`simlock token create --role worker`, minted on the gateway) this worker presents when it opens its uplink. Required whenever `gateway.url` is set. `simlock config` shows it; the daemon log and the `daemon.started` event show `[redacted]`. It is stored in plain text in `config.json`, which `simlock config set` writes owner-only. | unset | | `gateway.label` | **Worker side.** Display name for this worker in `simlock worker list`, `status`, the console, and on the lease's `worker` block. Display-only: nothing routes on it and it need not be unique. | the worker's own id | -| `agentDevice.enabled` | **Worker side.** Run one agent-device daemon on this worker, started and watched by Simlock. It sees Simlock's devices and no others, refuses `boot` and `shutdown`, and keeps its files in `~/.simlock/agent-device`. If agent-device is missing, older than 0.21.25, speaks another RPC protocol or needs a newer Node than Simlock runs on, it is not started and Simlock runs on without it. A daemon a killed Simlock left behind is stopped at the next start. | `false` | +| `agentDevice.enabled` | **Worker side.** Run one agent-device daemon on this worker, started and watched by Simlock. It sees Simlock's devices and no others, refuses `boot` and `shutdown`, and keeps its files in `~/.simlock/agent-device`. If agent-device is missing, older than 0.21.25, speaks another RPC protocol or needs a newer Node than Simlock runs on, it is not started and Simlock runs on without it. A daemon a killed Simlock left behind is stopped at the next start. With `http.enabled` also `true`, the agent-device CLI reaches it at `/v1/agent-device` (see [HTTP-API.md](HTTP-API.md#v1agent-device)). | `false` | | `agentDevice.command` | **Worker side.** The agent-device executable: a name looked up on `PATH`, or a path. Used only when `agentDevice.enabled` is `true`. | `agent-device` | | `exec.timeoutMs` | **Worker side.** How long one `device.exec` command (`simlock simctl` / `simlock adb` against a gateway or over HTTP) may run before the worker kills it and the operation fails with `EXEC_TIMEOUT`. Authoritative: it bounds the process that actually runs. | `10 minutes` | | `gateway.routing` | **Gateway side.** Which routing policy places a queued request on a worker. `warm-then-free` is the only policy in v1. Among the workers that can serve the request, in order: a worker whose health is not `running`, or that has requests of its own waiting, is left out; a worker holding a ready device that fits goes first (the device's mode must fit the request, a request with no mode fits a device that serves the worker's default mode, and its runtime must be the one requested or, with none requested, the one the worker would pick: its default runtime if the model pairs with it, otherwise the only runtime the model pairs with); otherwise a worker with no free running slot for the platform or overall is left out (a running device that is not leased counts as free: the worker shuts it down to make room), a worker under its RAM budget for the platform is preferred over one at it (a worker at its budget is still asked when it is the only one left), and the one with the most free running capacity for the platform wins. A worker that is left out is busy, not unable: the request waits for it. Ties go to the lowest worker id. | `warm-then-free` | diff --git a/docs/HTTP-API.md b/docs/HTTP-API.md index 825b50fbf..e8d26a5a3 100644 --- a/docs/HTTP-API.md +++ b/docs/HTTP-API.md @@ -1453,6 +1453,72 @@ from that worker's local agents, each with the **`workerId`** whose queue it is in. A request the gateway has sent to a worker is listed once, as the gateway's own. A worker's `createdAt` comes from that worker's clock. +### `/v1/agent-device/*` + +A **worker** with `agentDevice.enabled` and `http.enabled` both `true` serves +agent-device's proxy URL here. Point the agent-device CLI at it and give it +your Simlock token: + +```bash +export AGENT_DEVICE_DAEMON_AUTH_TOKEN= +agent-device connect proxy --daemon-base-url http://:/v1/agent-device +agent-device open MyApp --platform ios --device "iPhone 17 Pro" +agent-device snapshot +agent-device close +``` + +With `agentDevice.enabled` `false`, and on a gateway, every `/v1/agent-device` +URL answers `404`. + +- **Every route needs the Simlock token.** No token is `401`, a worker join + token is `403`, and so is an `operator` token, with + `refused: agent-device through simlock needs an agent token`. +- **Each agent needs its own Simlock token.** Simlock tells agents apart by + token: two agents sharing one token share one agent-device session. +- **`open` leases a device from Simlock.** The device is picked by Simlock, + and waits in Simlock's queue, up to 330 seconds, when every match is taken. + The lease is an ordinary Simlock lease: `simlock list` shows it, owned by the + token's principal, and the lease ID is the 32-hex ID the CLI chose. `close` + ends the agent-device lease and then frees the Simlock lease, so Simlock + wipes the device for the next agent. +- **Heartbeats** reach agent-device unchanged. They do not renew the Simlock + lease yet, so a Simlock lease ends at its own TTL; pass `ttlMs` on `open` + for a longer one. + +What `open` asks Simlock for: + +| The agent passes | Simlock is asked for | +| --- | --- | +| `--platform ios` or `android` | that platform, any device Simlock grants | +| `--target mobile`, `tv`, `desktop` | the class `phone`, `tv`, `desktop` | +| `--device "iPhone 17 Pro"` | that model, and no class: the model wins over `--target` | +| `ttlMs` | that TTL, at least 5 seconds; none sent means `lease.defaultTtlMs` | + +Messages the route prints. The agent-device CLI reads the message in the body +(a JSON-RPC error on `/rpc`, `{ "ok": false, "error", "code" }` elsewhere), not +the status. + +- `agent-device is not available on this simlock: `: agent-device is + turned on but missing, too old, starting or crashed. The rest of Simlock + works as before. +- `timed out waiting for a device; nothing was leased`. +- `refused by simlock: `: Simlock refused the lease request, for + example a model it does not know, or a `ttlMs` above `lease.maxTtlMs`. +- `refused: only iOS and Android`, `refused: name --platform ios or android`. +- `refused: simlock picks the device`: `--udid` or `--serial` on `open`. +- `refused: this agent-device is too old for simlock; upgrade it`: the CLI + sent no lease ID. +- `refused: send lease calls as lease RPCs`: a lease call inside + `agent_device.command`. +- `refused: ttlMs must be a positive number of milliseconds`. +- `refused: simlock leases only plain devices`: `backend`, + `retainOnClose: true`, a `provider*` or `aws*` key, or a `leaseProvider` + other than `proxy`. +- `refused: agent-device through simlock needs an agent token`. +- `refused: request body over 1 MiB` and `refused: send one JSON-RPC request` + (a body that is not JSON, a JSON array, or an object with no string + `method`); both answer with `id: null`. + ## Errors Every failure is the same shape the daemon protocol uses: diff --git a/docs/internal/COMPONENTS.md b/docs/internal/COMPONENTS.md index 03980e904..a8d506f29 100644 --- a/docs/internal/COMPONENTS.md +++ b/docs/internal/COMPONENTS.md @@ -114,6 +114,10 @@ modules with. | `AgentDeviceSupervisor` | `src/agent-device/host/agent-device-supervisor.ts` | Starting, watching and stopping the one agent-device daemon: reaping one a killed simlock left (by a pid record and an identity check), the private state dir and policy file, the daemon's environment, readiness through `daemon.json`, the port and `/health`, restart on death, and the availability state (`off`, `starting`, `running`, `unavailable` with a reason and remedy). | Serve agent-device traffic; report to `simlock doctor`; decide which devices exist (the drivers' environments do). | | Install lookup, compatibility, daemon environment | `src/agent-device/host/install.ts`, `src/agent-device/shared/` | Finding the agent-device install and its daemon entry; the minimum versions and RPC protocol; the environment and policy data. | Run a process. | | `AgentDeviceHealthProbe` | `src/agent-device/host/health.ts` | The `GET /health` read, with a time limit. | Judge the answer. | +| Front | `src/agent-device/front/` | The `/v1/agent-device/*` mount: the role check, the `/rpc` body check, turning an agent-device `allocate` into a simlock `lease.request` (and its cancel when the CLI leaves) and a `release` into a `lease.release`, and the front map of the leases it made. | Check a token (the app does); run the proxy; renew leases (#473). | +| `AgentDeviceHost`, `AgentDeviceUpstream` | `src/agent-device/host/host.ts`, `host/upstream.ts` | `@agent-device/proxy` in front of the daemon, one proxy per daemon start; the tenant each request acts as; the port the daemon is reached through and its supervisor adapter. | Know simlock leases. | +| Wire rules | `src/agent-device/shared/` | The path prefixes, the `/rpc` body reader, the call classifier, the lease-call field reader, the reply reader and the front's replies. | Decide what a call means. | +| `FakeAgentDevice` | `src/agent-device/testing.ts` | The in-memory agent-device daemon the proxy sits in front of in tests. | Appear in production code. | ## Drivers diff --git a/package.json b/package.json index 8c7bc5295..84f9c8000 100644 --- a/package.json +++ b/package.json @@ -64,6 +64,7 @@ "mutate": "node scripts/mutate.mjs" }, "dependencies": { + "@agent-device/proxy": "file:vendor/agent-device-proxy-0.21.27-dev-54e4e4d.tgz", "@hono/node-server": "2.0.10", "@hono/zod-validator": "^0.9.1", "@modelcontextprotocol/sdk": "1.29.0", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 3de6a5b20..27ca0cc80 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -211,6 +211,9 @@ importers: .: dependencies: + '@agent-device/proxy': + specifier: file:vendor/agent-device-proxy-0.21.27-dev-54e4e4d.tgz + version: file:vendor/agent-device-proxy-0.21.27-dev-54e4e4d.tgz '@hono/node-server': specifier: 2.0.10 version: 2.0.10(hono@4.13.5) @@ -311,6 +314,11 @@ importers: packages: + '@agent-device/proxy@file:vendor/agent-device-proxy-0.21.27-dev-54e4e4d.tgz': + resolution: {integrity: sha512-/Q0a2HgHF82DmisG8y5sYsguc5sBc8nEuWuylaemVUWaPmqpJTvwV6aVVsGcluJInU29361hyN27zSbl1v7FiQ==, tarball: file:vendor/agent-device-proxy-0.21.27-dev-54e4e4d.tgz} + version: 0.21.27-dev + engines: {node: '>=22.12'} + '@axe-core/playwright@4.13.0': resolution: {integrity: sha512-6YLx+kxXu5GJceG4ozFg+33a2EMTdjYwWGloJ3sb9Kta5pp+ZNS53uxGVog5JetIY8s++P5UrtX+cri+u0VAVg==} peerDependencies: @@ -3093,6 +3101,8 @@ packages: snapshots: + '@agent-device/proxy@file:vendor/agent-device-proxy-0.21.27-dev-54e4e4d.tgz': {} + '@axe-core/playwright@4.13.0(playwright-core@1.63.0)': dependencies: axe-core: 4.13.0 diff --git a/src/agent-device/edge.test.ts b/src/agent-device/edge.test.ts new file mode 100644 index 000000000..f24f26917 --- /dev/null +++ b/src/agent-device/edge.test.ts @@ -0,0 +1,1199 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { createServer } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; + +import { FakeDriver } from "../core/testing.js"; +import { startDaemon, type StartDaemonOptions } from "../daemon/main.js"; +import type { DaemonServer } from "../daemon/server.js"; +import { + CryptoTokenSecrets, + FakeClock, + JsonLinesLogger, + MemoryFilesystem, + MemoryLogSink, +} from "../ports/index.js"; +import { FakeAgentDevice } from "./testing.js"; + +const runningDaemons: DaemonServer[] = []; +const temporaryDirectories: string[] = []; + +afterEach(async () => { + await Promise.all(runningDaemons.splice(0).map((daemon) => daemon.stop("test"))); + await Promise.all( + temporaryDirectories.splice(0).map((path) => rm(path, { force: true, recursive: true })), + ); +}); + +const secrets = { + one: "slk_test_agent_one", + operator: "slk_test_operator", + two: "slk_test_agent_two", + worker: "slk_test_worker_join", +} as const; +const principals = { one: "tok_one", two: "tok_two" } as const; + +const notAvailable = "agent-device is not available on this simlock: "; +const agentTokenOnly = "refused: agent-device through simlock needs an agent token"; +const queueTimeoutMessage = "timed out waiting for a device; nothing was leased"; +const defaultTtlMs = 600_000; +const maxTtlMs = 1_800_000; +const scope = { clientId: "client-1", leaseProvider: "proxy", runId: "run-1" } as const; + +const adminSession = { + manageEventSubscription: () => undefined, + principal: "admin", + role: "admin", +} as const; + +function sessionFor(principal: string) { + return { manageEventSubscription: () => undefined, principal, role: "agent" } as const; +} + +interface Envelope { + readonly id?: unknown; + readonly jsonrpc?: string; + readonly result?: Record; + readonly error?: { + readonly code: number; + readonly message: string; + readonly data?: { readonly code: string; readonly message: string }; + }; +} + +interface SetUpOptions { + readonly drivers?: (clock: FakeClock) => FakeDriver[]; + readonly enabled?: boolean; + readonly gatewayMode?: boolean; + readonly maxRunning?: number; + readonly slowStart?: boolean; + readonly upstream?: FakeAgentDevice; +} + +interface Rig { + readonly clock: FakeClock; + readonly daemon: DaemonServer; + readonly driver: FakeDriver; + readonly fake: FakeAgentDevice; + readonly filesystem: MemoryFilesystem; + readonly origin: string; + readonly statePath: string; + readonly started: Promise; +} + +const phoneModels = { + defaultModels: { phone: ["iPhone 16"] }, + modelClasses: { "iPhone 16": "phone" }, +} as const; + +function iosDriver(clock: FakeClock): FakeDriver { + return new FakeDriver({ + availableOsVersions: ["26.5"], + clock, + defaultModels: { desktop: ["Mac"], phone: ["iPhone 16"], tv: ["Apple TV 4K"] }, + knownModels: ["iPhone 16", "iPhone 17 Pro", "Apple TV 4K", "Mac"], + modelClasses: { + "Apple TV 4K": "tv", + "iPhone 16": "phone", + "iPhone 17 Pro": "phone", + Mac: "desktop", + }, + platform: "ios", + }); +} + +function androidDriver(clock: FakeClock): FakeDriver { + return new FakeDriver({ + availableOsVersions: ["35"], + clock, + defaultModels: { phone: ["Pixel 8"] }, + knownModels: ["Pixel 8"], + modelClasses: { "Pixel 8": "phone" }, + platform: "android", + }); +} + +async function freeLoopbackPort(): Promise { + const server = createServer(); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + await new Promise((resolve) => server.close(() => resolve())); + if (typeof address !== "object" || address === null) throw new Error("no port bound"); + return address.port; +} + +async function sleep(ms: number): Promise { + await new Promise((resolve) => setTimeout(resolve, ms)); +} + +async function setUp(options: SetUpOptions = {}): Promise { + const directory = await mkdtemp(join(tmpdir(), "simlock-agent-device-edge-")); + temporaryDirectories.push(directory); + const clock = new FakeClock(1_000); + const filesystem = new MemoryFilesystem(); + await filesystem.mkdirp(directory); + const hasher = new CryptoTokenSecrets(); + const record = (id: string, secret: string, role: string) => ({ + createdAt: 0, + hash: hasher.hash(secret), + id, + role, + }); + await filesystem.writeFileAtomic( + join(directory, "tokens.json"), + JSON.stringify([ + record(principals.one, secrets.one, "agent"), + record(principals.two, secrets.two, "agent"), + record("tok_operator", secrets.operator, "operator"), + record("tok_worker", secrets.worker, "worker"), + ]), + ); + const port = await freeLoopbackPort(); + const fake = options.upstream ?? new FakeAgentDevice(); + const drivers = options.drivers?.(clock) ?? [iosDriver(clock)]; + const statePath = join(directory, "state.json"); + const started = startDaemon({ + agentDeviceUpstream: fake, + clock, + configOverrides: { + agentDevice: { enabled: options.enabled ?? true }, + capacity: { config: { maxRunning: options.maxRunning ?? 4 }, strategy: "fixed" }, + http: { enabled: true, host: "127.0.0.1", port }, + lease: { defaultTtlMs, maxTtlMs }, + ...(options.gatewayMode === true ? { mode: "gateway" as const } : {}), + }, + dataDirectory: directory, + drivers, + filesystem, + logger: new JsonLinesLogger({ clock, level: "debug", sink: new MemoryLogSink() }), + statePath, + version: "1.2.3", + } as StartDaemonOptions).then((daemon) => { + runningDaemons.push(daemon); + return daemon; + }); + const origin = `http://127.0.0.1:${port}`; + if (options.slowStart === true) { + await waitListening(origin); + return { + clock, + daemon: undefined as never, + driver: drivers[0] as FakeDriver, + fake, + filesystem, + origin, + started, + statePath, + }; + } + const daemon = await started; + return { + clock, + daemon, + driver: drivers[0] as FakeDriver, + fake, + filesystem, + origin, + started, + statePath, + }; +} + +async function waitListening(origin: string): Promise { + for (let attempt = 0; attempt < 400; attempt += 1) { + try { + await fetch(`${origin}/v1/healthz`); + return; + } catch { + await sleep(5); + } + } + throw new Error("the daemon never listened"); +} + +async function advance(rig: Rig, ms: number, step = 1_000): Promise { + for (let elapsed = 0; elapsed < ms; elapsed += step) { + rig.clock.advance(Math.min(step, ms - elapsed)); + await sleep(3); + } +} + +interface SendOptions { + readonly body?: string; + readonly headers?: Record; + readonly method?: string; + readonly signal?: AbortSignal; + readonly token?: string | null; +} + +function send(rig: Rig, path: string, options: SendOptions = {}): Promise { + const token = options.token === undefined ? secrets.one : options.token; + return fetch(`${rig.origin}/v1/agent-device${path}`, { + ...(options.body === undefined ? {} : { body: options.body }), + headers: { + ...(token === null ? {} : { authorization: `Bearer ${token}` }), + ...(options.body === undefined ? {} : { "content-type": "application/json" }), + ...options.headers, + }, + method: options.method ?? (options.body === undefined ? "GET" : "POST"), + ...(options.signal === undefined ? {} : { signal: options.signal }), + }); +} + +function rpcBody(method: string, params: Record, id: unknown = 1): string { + return JSON.stringify({ id, jsonrpc: "2.0", method, params }); +} + +function rpc( + rig: Rig, + method: string, + params: Record, + options: Omit & { readonly id?: unknown } = {}, +): Promise { + return send(rig, "/rpc", { ...options, body: rpcBody(method, params, options.id ?? 1) }); +} + +async function bodyOf(response: Response): Promise> { + const text = await response.text(); + expect( + text.trimStart().startsWith("{"), + `expected a JSON object body, got ${response.status} ${text.slice(0, 80)}`, + ).toBe(true); + return JSON.parse(text) as Record; +} + +async function envelopeOf(response: Response): Promise { + return (await bodyOf(response)) as Envelope; +} + +const hex = (n: number): string => n.toString(16).padStart(32, "0"); + +function allocateParams(n: number, extra: Record = {}): Record { + return { ...scope, leaseId: hex(n), platform: "ios", ttlMs: 120_000, ...extra }; +} + +function allocate( + rig: Rig, + n: number, + extra: Record = {}, + options: Omit = {}, +): Promise { + return rpc(rig, "agent_device.lease.allocate", allocateParams(n, extra), options); +} + +async function granted( + rig: Rig, + n: number, + extra: Record = {}, + options: Omit = {}, +): Promise { + const answer = await envelopeOf(await allocate(rig, n, extra, options)); + expect(answer.result).toMatchObject({ ok: true }); +} + +function release(rig: Rig, n: number, options: Omit = {}): Promise { + return rpc(rig, "agent_device.lease.release", { ...scope, leaseId: hex(n) }, options); +} + +function command( + rig: Rig, + name: string, + n: number | undefined, + options: Omit = {}, + extra: Record = {}, +): Promise { + return rpc( + rig, + "agent_device.command", + { + command: name, + ...(n === undefined ? {} : { meta: { leaseId: hex(n) } }), + ...extra, + }, + options, + ); +} + +interface LeaseView { + readonly id: string; + readonly deviceId: string; + readonly requesterId: string; + readonly ownerId: string; + readonly ttlMs: number; + readonly ttlDeadline: number; + readonly lastRenewedAt: number; +} + +async function leases(rig: Rig): Promise { + return (await rig.daemon.dispatch( + "list.get", + { kind: "leases" }, + adminSession, + )) as unknown as readonly LeaseView[]; +} + +async function waiting( + rig: Rig, +): Promise { + return (await rig.daemon.dispatch("list.get", { kind: "requests" }, adminSession)) as never; +} + +async function devices(rig: Rig): Promise< + readonly { + readonly id: string; + readonly state: string; + readonly address?: string; + readonly driverDeviceId?: string; + }[] +> { + return (await rig.daemon.dispatch("list.get", { kind: "devices" }, adminSession)) as never; +} + +async function requestedSpecs(rig: Rig, requester: string): Promise { + const events = await rig.daemon.dispatch("events.replay", {}, adminSession); + return events + .filter((entry) => entry.event === "lease.requested") + .map((entry) => entry.payload as { requester: string; requestSpec: unknown }) + .filter((payload) => payload.requester === requester) + .map((payload) => payload.requestSpec); +} + +async function grantedDevice(rig: Rig) { + const [device] = await devices(rig); + expect(device?.address).toBeDefined(); + return device as { readonly address: string; readonly driverDeviceId?: string }; +} + +function receivedAllocate(rig: Rig, n: number): Record { + const params = allocatesReceived(rig, n)[0]?.params; + expect(params).toBeDefined(); + return params as Record; +} + +function allocatesReceived(rig: Rig, n?: number) { + return rig.fake + .rpcRequests("agent_device.lease.allocate") + .filter((request) => n === undefined || request.params?.["leaseId"] === hex(n)); +} + +async function simlockMessage(rig: Rig, input: Record): Promise { + try { + await rig.daemon.dispatch("lease.request", input, sessionFor(principals.one)); + } catch (error) { + return (error as Error).message; + } + throw new Error("simlock accepted a request the test expected it to refuse"); +} + +function track(promise: Promise): { + readonly done: () => boolean; + readonly promise: Promise; +} { + let settled = false; + const tracked = promise.finally(() => { + settled = true; + }); + tracked.catch(() => undefined); + return { done: () => settled, promise: tracked }; +} + +function refusal(message: string, code = "INVALID_ARGS") { + return { code: -32000, data: { code, message }, message }; +} + +describe("the agent-device front, through the whole worker daemon", () => { + it("with agentDevice.enabled false, and on a gateway-mode daemon, /v1/agent-device/health answers 404, as today", async () => { + const enabled = await setUp(); + expect((await send(enabled, "/health")).status).toBe(200); + + const off = await setUp({ enabled: false }); + expect((await send(off, "/health")).status).toBe(404); + + const gateway = await setUp({ gatewayMode: true }); + expect((await send(gateway, "/health")).status).toBe(404); + }); + + it("/v1/agent-device/rpc with no simlock token answers 401, with a worker join token 403, and with an operator token 403 `refused: agent-device through simlock needs an agent token`; the agent-device daemon receives nothing", async () => { + const rig = await setUp(); + expect((await command(rig, "snapshot", undefined)).status).toBe(200); + const received = rig.fake.requests.length; + + expect((await command(rig, "snapshot", undefined, { token: null })).status).toBe(401); + expect((await command(rig, "snapshot", undefined, { token: secrets.worker })).status).toBe(403); + const operator = await command(rig, "snapshot", undefined, { token: secrets.operator }); + expect(operator.status).toBe(403); + expect(await operator.text()).toContain(agentTokenOnly); + + expect(rig.fake.requests).toHaveLength(received); + }); + + it("/v1/agent-device/health with a simlock token answers the proxy's health, with the agent-device daemon's health under `upstream`", async () => { + const rig = await setUp(); + + const response = await send(rig, "/health"); + + expect(response.status).toBe(200); + expect(await response.json()).toMatchObject({ + service: "agent-device-proxy", + upstream: { service: "agent-device-daemon" }, + }); + }); + + it("with agentDevice.enabled true and agent-device unavailable, /health, /rpc and an upload each answer `agent-device is not available on this simlock: `, nothing is leased, and POST /v1/lease-requests still grants a device", async () => { + const rig = await setUp(); + const reason = "found 0.20.3, need 0.21.25 or later"; + rig.fake.setUnavailable(reason); + + const health = await send(rig, "/health"); + const rpcAnswer = await allocate(rig, 1); + const upload = await send(rig, "/upload/preflight", { body: "{}" }); + + expect(await health.text()).toContain(`${notAvailable}${reason}`); + expect((await envelopeOf(rpcAnswer)).error?.message).toBe(`${notAvailable}${reason}`); + expect(await upload.text()).toContain(`${notAvailable}${reason}`); + expect(await leases(rig)).toEqual([]); + expect(rig.fake.requests).toEqual([]); + + const granted = await fetch(`${rig.origin}/v1/lease-requests`, { + body: JSON.stringify({ device: "iPhone 16", platform: "ios" }), + headers: { authorization: `Bearer ${secrets.one}`, "content-type": "application/json" }, + method: "POST", + }); + expect(granted.status).toBe(201); + }); + + it("after the agent-device daemon restarts with a new base URL, token and generation, the next request reaches the new daemon", async () => { + const rig = await setUp(); + expect(await bodyOf(await send(rig, "/health"))).toMatchObject({ + upstream: { service: "agent-device-daemon" }, + }); + const before = rig.fake.requests.length; + + rig.fake.restart(); + const response = await send(rig, "/health"); + + expect(await bodyOf(response)).toMatchObject({ upstream: { service: "agent-device-daemon" } }); + expect(rig.fake.requests.length).toBeGreaterThan(before); + }); + + it("commands sent with two simlock tokens reach the agent-device daemon as two tenants, each its token's principal, though the client named another tenant", async () => { + const rig = await setUp(); + const named = { meta: { tenantId: "someone-else" }, tenantId: "someone-else" }; + + await command(rig, "snapshot", undefined, { token: secrets.one }, named); + await command(rig, "snapshot", undefined, { token: secrets.two }, named); + + expect(rig.fake.rpcRequests("agent_device.command").map((request) => request.tenant)).toEqual([ + principals.one, + principals.two, + ]); + }); + + it("allocate `--platform ios` makes a simlock lease whose ID and requester are the CLI's lease ID, owned by the token's principal, and the lease request names no osVersion, mode, imageTag or physical device", async () => { + const rig = await setUp(); + + const answer = await allocate(rig, 1); + + expect((await envelopeOf(answer)).result).toMatchObject({ ok: true }); + expect(await leases(rig)).toMatchObject([ + { id: hex(1), ownerId: principals.one, requesterId: hex(1) }, + ]); + expect(await requestedSpecs(rig, hex(1))).toEqual([{ platform: "ios" }]); + + const bare = await setUp({ + drivers: (clock) => [ + new FakeDriver({ + availableOsVersions: [], + clock, + knownModels: ["iPhone 16"], + platform: "ios", + }), + ], + }); + const refused = await allocate(bare, 2); + expect((await envelopeOf(refused)).error?.message).toContain("refused by simlock: "); + expect(bare.driver.calls.map((call) => call.operation)).not.toContain("installComponent"); + }); + + it("the simlock lease's TTL is the CLI's `ttlMs`, 5 s when the CLI sent less, and `lease.defaultTtlMs` when it sent none; the allocate the agent-device daemon receives carries `ttlMs` equal to the ceiling in each case", async () => { + const rig = await setUp(); + + await granted(rig, 1, { ttlMs: 120_000 }); + await granted(rig, 2, { ttlMs: 1_000 }); + await granted(rig, 3, { ttlMs: undefined }); + + const byId = new Map((await leases(rig)).map((lease) => [lease.id, lease.ttlMs])); + expect(byId.get(hex(1))).toBe(120_000); + expect(byId.get(hex(2))).toBe(5_000); + expect(byId.get(hex(3))).toBe(defaultTtlMs); + expect(allocatesReceived(rig).map((request) => request.params?.["ttlMs"])).toEqual([ + rig.fake.leaseMaxTtlMs, + rig.fake.leaseMaxTtlMs, + rig.fake.leaseMaxTtlMs, + ]); + }); + + it('allocate `--target mobile`, `tv` and `desktop` ask simlock for class `phone`, `tv` and `desktop`; allocate `--device "iPhone 17 Pro"`, alone and with `--target tv`, asks for that model and no class', async () => { + const rig = await setUp({ maxRunning: 8 }); + + await granted(rig, 1, { target: "mobile" }); + await granted(rig, 2, { target: "tv" }); + await granted(rig, 3, { target: "desktop" }); + await granted(rig, 4, { device: "iPhone 17 Pro" }); + await granted(rig, 5, { device: "iPhone 17 Pro", target: "tv" }); + + expect(await requestedSpecs(rig, hex(1))).toEqual([{ class: "phone", platform: "ios" }]); + expect(await requestedSpecs(rig, hex(2))).toEqual([{ class: "tv", platform: "ios" }]); + expect(await requestedSpecs(rig, hex(3))).toEqual([{ class: "desktop", platform: "ios" }]); + expect(await requestedSpecs(rig, hex(4))).toEqual([ + { model: "iPhone 17 Pro", platform: "ios" }, + ]); + expect(await requestedSpecs(rig, hex(5))).toEqual([ + { model: "iPhone 17 Pro", platform: "ios" }, + ]); + }); + + it("an allocate whose `ttlMs` is above `lease.maxTtlMs` answers `refused by simlock:` with simlock's `BAD_REQUEST` message and leaves no simlock lease; one with `ttlMs: null` gets `lease.defaultTtlMs`", async () => { + const rig = await setUp(); + expect((await envelopeOf(await allocate(rig, 1, { ttlMs: null }))).result).toMatchObject({ + ok: true, + }); + expect((await leases(rig)).map((lease) => lease.ttlMs)).toEqual([defaultTtlMs]); + + const over = maxTtlMs + 1; + const expected = await simlockMessage(rig, { leaseId: hex(9), platform: "ios", ttlMs: over }); + const answer = await envelopeOf(await allocate(rig, 2, { ttlMs: over })); + + expect(answer.error?.message).toBe(`refused by simlock: ${expected}`); + expect((await leases(rig)).map((lease) => lease.id)).toEqual([hex(1)]); + }); + + it('allocate `--device "My test phone"` answers `refused by simlock:` with simlock\'s message, and leaves no simlock lease', async () => { + const rig = await setUp(); + expect( + (await envelopeOf(await allocate(rig, 1, { device: "iPhone 17 Pro" }))).result, + ).toMatchObject({ ok: true }); + const expected = await simlockMessage(rig, { + leaseId: hex(9), + model: "My test phone", + platform: "ios", + }); + + const answer = await envelopeOf(await allocate(rig, 2, { device: "My test phone" })); + + expect(answer.error).toMatchObject(refusal(`refused by simlock: ${expected}`)); + expect((await leases(rig)).map((lease) => lease.id)).toEqual([hex(1)]); + }); + + describe("each input refused before simlock", () => { + const tooOld = "refused: this agent-device is too old for simlock; upgrade it"; + const ttlRefusal = "refused: ttlMs must be a positive number of milliseconds"; + const plainOnly = "refused: simlock leases only plain devices"; + const lifecycle = "refused: send lease calls as lease RPCs"; + const cases: readonly { + readonly title: string; + readonly message: string; + readonly method?: string; + readonly params: (n: number) => Record; + }[] = [ + { + message: "refused: only iOS and Android", + params: (n) => allocateParams(n, { platform: "macos" }), + title: "--platform macos", + }, + { + message: "refused: simlock picks the device", + params: (n) => allocateParams(n, { udid: "ABC" }), + title: "--udid", + }, + { + message: "refused: simlock picks the device", + params: (n) => allocateParams(n, { serial: "emulator-5554" }), + title: "--serial", + }, + { + message: "refused: name --platform ios or android", + params: (n) => allocateParams(n, { platform: undefined }), + title: "no platform", + }, + { + message: tooOld, + params: (n) => allocateParams(n, { leaseId: undefined }), + title: "no lease ID", + }, + { + message: tooOld, + params: (n) => allocateParams(n, { leaseId: hex(n).toUpperCase().replace(/[0-9]/g, "A") }), + title: "a lease ID that is not 32 lowercase hex", + }, + { + message: ttlRefusal, + params: (n) => allocateParams(n, { ttlMs: "abc" }), + title: 'a ttlMs of "abc"', + }, + { + message: ttlRefusal, + params: (n) => allocateParams(n, { ttlMs: -1 }), + title: "a ttlMs of -1", + }, + { + message: ttlRefusal, + params: (n) => allocateParams(n, { ttlMs: 0 }), + title: "a ttlMs of 0", + }, + { + message: plainOnly, + params: (n) => allocateParams(n, { backend: "cloud" }), + title: "backend", + }, + { + message: plainOnly, + params: (n) => allocateParams(n, { retainOnClose: true }), + title: "retainOnClose: true", + }, + { + message: plainOnly, + params: (n) => allocateParams(n, { providerName: "x" }), + title: "a provider* key", + }, + { + message: plainOnly, + params: (n) => allocateParams(n, { leaseProvider: "aws" }), + title: 'leaseProvider: "aws"', + }, + ...["agent_device.command", "agent-device.command"].flatMap((method) => + ["lease_allocate", "lease_heartbeat", "lease_release"].map((name) => ({ + message: lifecycle, + method, + params: (n: number) => ({ command: name, meta: { leaseId: hex(n) } }), + title: `${method} carrying ${name}`, + })), + ), + ]; + + it.each(cases)( + "$title answers its message, simlock stores no lease request, and the agent-device daemon receives nothing", + async ({ message, method, params }) => { + const rig = await setUp(); + const accepted = await allocate(rig, 1); + expect((await envelopeOf(accepted)).result).toMatchObject({ ok: true }); + const heldBefore = (await leases(rig)).map((lease) => lease.id); + const receivedBefore = rig.fake.requests.length; + + const answer = await rpc(rig, method ?? "agent_device.lease.allocate", params(2)); + + expect((await envelopeOf(answer)).error).toMatchObject(refusal(message)); + expect(await requestedSpecs(rig, hex(2))).toEqual([]); + expect(await waiting(rig)).toEqual([]); + expect((await leases(rig)).map((lease) => lease.id)).toEqual(heldBefore); + expect(rig.fake.requests).toHaveLength(receivedBefore); + }, + ); + }); + + it("allocate and release sent as `agent-device.lease.allocate` and `agent-device.lease.release` act exactly as the `agent_device.lease.*` spelling", async () => { + const rig = await setUp(); + + const allocated = await rpc(rig, "agent-device.lease.allocate", allocateParams(1)); + + expect((await envelopeOf(allocated)).result).toMatchObject({ ok: true }); + expect((await leases(rig)).map((lease) => lease.id)).toEqual([hex(1)]); + expect(rig.fake.leases.has(hex(1))).toBe(true); + + const released = await rpc(rig, "agent-device.lease.release", { ...scope, leaseId: hex(1) }); + + expect((await envelopeOf(released)).result).toMatchObject({ ok: true }); + expect(await leases(rig)).toEqual([]); + expect(rig.fake.leases.has(hex(1))).toBe(false); + }); + + it("after the grant, the allocate the agent-device daemon receives names the granted device's address by `udid` on iOS and by `serial` on Android (the emulator serial, not the AVD name), and carries no `device`, `deviceKey` or `target`; the CLI gets the daemon's reply with the same lease ID", async () => { + const ios = await setUp(); + const iosAnswer = await envelopeOf( + await allocate(ios, 1, { device: "iPhone 17 Pro", deviceKey: "saved", target: "mobile" }), + ); + const iosDevice = await grantedDevice(ios); + const iosParams = receivedAllocate(ios, 1); + + expect(iosAnswer.result).toMatchObject({ lease: { leaseId: hex(1) }, ok: true }); + expect(iosParams["udid"]).toBe(iosDevice.address); + expect(iosParams).not.toHaveProperty("serial"); + for (const dropped of ["device", "deviceKey", "target"]) { + expect(iosParams).not.toHaveProperty(dropped); + } + + const android = await setUp({ drivers: (clock) => [androidDriver(clock)] }); + const androidAnswer = await envelopeOf( + await allocate(android, 2, { device: "Pixel 8", platform: "android" }), + ); + const androidDevice = await grantedDevice(android); + const androidParams = receivedAllocate(android, 2); + + expect(androidAnswer.result).toMatchObject({ lease: { leaseId: hex(2) }, ok: true }); + expect(androidParams["serial"]).toBe(androidDevice.address); + expect(androidParams["serial"]).not.toBe(androidDevice.driverDeviceId); + expect(androidParams).not.toHaveProperty("udid"); + }); + + it("with every matching device leased, allocate waits, and is granted and answered once that device is released and reclaimed", async () => { + const rig = await setUp({ maxRunning: 1 }); + expect((await envelopeOf(await allocate(rig, 1))).result).toMatchObject({ ok: true }); + + const second = track(allocate(rig, 2, {}, { token: secrets.two })); + await expect + .poll(async () => (await waiting(rig)).map((request) => request.requesterId)) + .toEqual([hex(2)]); + expect(second.done()).toBe(false); + + await release(rig, 1); + const answer = await envelopeOf(await second.promise); + + expect(answer.result).toMatchObject({ lease: { leaseId: hex(2) }, ok: true }); + expect((await leases(rig)).map((lease) => lease.id)).toEqual([hex(2)]); + }); + + it("a CLI that hangs up while its allocate waits, and one that hangs up before simlock admitted the request, leave simlock's queue, and neither request is granted when a device frees", async () => { + const rig = await setUp({ maxRunning: 1 }); + await granted(rig, 1); + const hangUp = new AbortController(); + const second = track(allocate(rig, 2, {}, { signal: hangUp.signal, token: secrets.two })); + await sleep(50); + expect(second.done()).toBe(false); + await expect.poll(async () => (await waiting(rig)).length).toBe(1); + + hangUp.abort(); + await second.promise.catch(() => undefined); + await expect.poll(async () => (await waiting(rig)).length).toBe(0); + await release(rig, 1); + await sleep(50); + + expect(await leases(rig)).toEqual([]); + expect(allocatesReceived(rig, 2)).toEqual([]); + + const slow = await setUp({ + drivers: (clock) => [ + new FakeDriver({ + ...phoneModels, + availableOsVersions: ["26.5"], + clock, + knownModels: ["iPhone 16"], + latencyMs: { listManaged: 30_000 }, + platform: "ios", + }), + ], + slowStart: true, + }); + const early = new AbortController(); + const parked = track( + fetch(`${slow.origin}/v1/agent-device/rpc`, { + body: rpcBody("agent_device.lease.allocate", allocateParams(3)), + headers: { authorization: `Bearer ${secrets.one}`, "content-type": "application/json" }, + method: "POST", + signal: early.signal, + }), + ); + await sleep(100); + expect(parked.done()).toBe(false); + early.abort(); + await parked.promise.catch(() => undefined); + await sleep(50); + slow.clock.advance(30_000); + const ready = { ...slow, daemon: await slow.started }; + await sleep(100); + + expect(await leases(ready)).toEqual([]); + expect(await requestedSpecs(ready, hex(3))).toEqual([]); + expect(allocatesReceived(ready, 3)).toEqual([]); + }); + + it("a CLI that hangs up after device work started for it gets no agent-device lease: the grant that lands later is released at once and never reaches the agent-device daemon; when that device work fails and the request goes back to the queue, it is never granted", async () => { + const slowDevice = (clock: FakeClock) => [ + new FakeDriver({ + ...phoneModels, + availableOsVersions: ["26.5"], + clock, + knownModels: ["iPhone 16"], + latencyMs: { makeReady: 10_000 }, + platform: "ios", + }), + ]; + const rig = await setUp({ drivers: slowDevice }); + const hangUp = new AbortController(); + const first = track(allocate(rig, 1, {}, { signal: hangUp.signal })); + await sleep(50); + expect(first.done()).toBe(false); + await expect + .poll(() => rig.driver.calls.some((call) => call.operation === "makeReady")) + .toBe(true); + + hangUp.abort(); + await first.promise.catch(() => undefined); + await advance(rig, 12_000); + await expect.poll(async () => (await leases(rig)).length).toBe(0); + + expect(allocatesReceived(rig, 1)).toEqual([]); + + const failing = await setUp({ drivers: slowDevice }); + failing.driver.failOn("makeReady", 1, new Error("boot failed")); + const abort = new AbortController(); + const second = track(allocate(failing, 2, {}, { signal: abort.signal })); + await sleep(50); + expect(second.done()).toBe(false); + await expect + .poll(() => failing.driver.calls.some((call) => call.operation === "makeReady")) + .toBe(true); + abort.abort(); + await second.promise.catch(() => undefined); + + await advance(failing, 15_000); + + await expect.poll(async () => (await waiting(failing)).length).toBe(0); + expect(await leases(failing)).toEqual([]); + expect(allocatesReceived(failing, 2)).toEqual([]); + }); + + it("`DELETE /v1/lease-requests/:id` for the front's waiting request answers the allocate `refused by simlock:` with simlock's message, and nothing is leased", async () => { + const rig = await setUp({ maxRunning: 1 }); + await granted(rig, 1, {}, { token: secrets.two }); + const second = track(allocate(rig, 2)); + await expect.poll(async () => (await waiting(rig)).length).toBe(1); + const [request] = await waiting(rig); + + const cancelled = await fetch(`${rig.origin}/v1/lease-requests/${request?.id}`, { + headers: { authorization: `Bearer ${secrets.one}` }, + method: "DELETE", + }); + const answer = await envelopeOf(await second.promise); + + expect(cancelled.status).toBe(204); + expect(answer.error?.message).toMatch(/^refused by simlock: .+/); + expect((await leases(rig)).map((lease) => lease.id)).toEqual([hex(1)]); + expect(await waiting(rig)).toEqual([]); + }); + + it("stopping the daemon while an allocate waits ends with no simlock lease, and the lease request it waited on is cancelled, not open, in state.json", async () => { + const rig = await setUp({ maxRunning: 1 }); + await granted(rig, 1, {}, { token: secrets.two }); + const second = track(allocate(rig, 2)); + await expect.poll(async () => (await waiting(rig)).length).toBe(1); + + await rig.daemon.stop("test"); + await second.promise.catch(() => undefined); + + const state = await rig.filesystem.readFile(rig.statePath); + expect(second.done()).toBe(true); + expect(state).toContain(hex(1)); + const records = JSON.parse(state ?? "{}") as { + leaseRequests?: readonly { leaseId?: string; state: string }[]; + }; + expect(records.leaseRequests?.filter((record) => record.leaseId === hex(2))).toEqual([ + expect.objectContaining({ state: "cancelled" }), + ]); + expect(records.leaseRequests?.some((record) => record.state === "open")).toBe(false); + }); + + it("an allocate still waiting 330 s after the front admitted it answers `timed out waiting for a device; nothing was leased`, and simlock's queue is empty", async () => { + const rig = await setUp({ maxRunning: 1 }); + await granted(rig, 1, { ttlMs: maxTtlMs }, { token: secrets.two }); + const second = track(allocate(rig, 2)); + await expect.poll(async () => (await waiting(rig)).length).toBe(1); + + await advance(rig, 329_000); + expect(second.done()).toBe(false); + await advance(rig, 1_000); + const answer = await envelopeOf(await second.promise); + + expect(answer.error).toMatchObject(refusal(queueTimeoutMessage, "COMMAND_FAILED")); + expect(await waiting(rig)).toEqual([]); + expect((await leases(rig)).map((lease) => lease.id)).toEqual([hex(1)]); + }); + + it("an agent-device allocate that answers an error, and a grant that names no device address, release the simlock lease, and the CLI gets the error", async () => { + const rig = await setUp(); + rig.fake.failNextAllocate({ code: "NO_DEVICE", message: "no such device" }); + + const answer = await envelopeOf(await allocate(rig, 1)); + + expect(answer.error).toMatchObject({ + data: { code: "NO_DEVICE", message: "no such device" }, + message: "no such device", + }); + await expect.poll(async () => (await leases(rig)).length).toBe(0); + + class NoAddressDriver extends FakeDriver { + override async makeReady(...arguments_: Parameters) { + const ready = await super.makeReady(...arguments_); + return { ...ready, address: undefined as unknown as string }; + } + } + const bare = await setUp({ + drivers: (clock) => [ + new NoAddressDriver({ + ...phoneModels, + availableOsVersions: ["26.5"], + clock, + knownModels: ["iPhone 16"], + platform: "ios", + }), + ], + }); + + const noAddress = await envelopeOf(await allocate(bare, 2)); + + expect(noAddress.error?.data?.code).toBe("COMMAND_FAILED"); + await expect.poll(async () => (await leases(bare)).length).toBe(0); + expect(allocatesReceived(bare, 2)).toEqual([]); + }); + + it("an agent-device allocate with no usable reply (an unreachable daemon, a non-JSON body) releases the simlock lease and answers `COMMAND_FAILED`; when agent-device answers the follow-up release with a scope error, the simlock lease stays until its TTL, and when it answers `LEASE_NOT_FOUND`, the simlock lease is released", async () => { + const rig = await setUp(); + expect((await envelopeOf(await allocate(rig, 1))).result).toMatchObject({ ok: true }); + rig.fake.setUnreachable(true); + const unreachable = await envelopeOf(await allocate(rig, 2)); + rig.fake.setUnreachable(false); + + expect(unreachable.error?.data?.code).toBe("COMMAND_FAILED"); + await expect.poll(async () => (await leases(rig)).map((lease) => lease.id)).toEqual([hex(1)]); + + rig.fake.answerNextAllocateWith("bad gateway"); + const notJson = await envelopeOf(await allocate(rig, 3)); + + expect(notJson.error?.data?.code).toBe("COMMAND_FAILED"); + await expect.poll(async () => (await leases(rig)).map((lease) => lease.id)).toEqual([hex(1)]); + + rig.fake.answerNextAllocateWith("bad gateway"); + rig.fake.failNextRelease({ code: "UNAUTHORIZED", message: "lease scope mismatch" }); + const kept = await envelopeOf(await allocate(rig, 4, { ttlMs: 60_000 })); + + expect(kept.error?.data?.code).toBe("COMMAND_FAILED"); + expect((await leases(rig)).map((lease) => lease.id)).toContain(hex(4)); + await advance(rig, 61_000); + await expect + .poll(async () => (await leases(rig)).map((lease) => lease.id)) + .not.toContain(hex(4)); + }); + + it("the follow-up release after an allocate with no usable reply carries the allocate's runId, leaseProvider and clientId and not its deviceKey", async () => { + const rig = await setUp(); + rig.fake.answerNextAllocateWith("bad gateway"); + + const answer = await envelopeOf(await allocate(rig, 1, { deviceKey: "saved-device" })); + + expect(answer.error?.data?.code).toBe("COMMAND_FAILED"); + const params = rig.fake.rpcRequests("agent_device.lease.release")[0]?.params; + expect(params).toMatchObject({ ...scope, leaseId: hex(1) }); + expect(params).not.toHaveProperty("deviceKey"); + }); + + it("a CLI that hangs up while agent-device makes its lease leaves neither lease: the agent-device lease is released with the reply's scope, then the simlock lease; when that release fails, the simlock lease stays until its TTL", async () => { + const rig = await setUp(); + rig.fake.holdAllocate(); + rig.fake.nameDeviceKeyInNextReply("device-key-1"); + const hangUp = new AbortController(); + const first = track(allocate(rig, 1, {}, { signal: hangUp.signal })); + await sleep(50); + expect(first.done()).toBe(false); + await expect.poll(() => allocatesReceived(rig, 1).length).toBe(1); + + hangUp.abort(); + await first.promise.catch(() => undefined); + await sleep(50); + rig.fake.letAllocateGo(); + + await expect.poll(async () => (await leases(rig)).length).toBe(0); + expect(rig.fake.leases.has(hex(1))).toBe(false); + expect(rig.fake.rpcRequests("agent_device.lease.release")[0]?.params).toMatchObject({ + ...scope, + deviceKey: "device-key-1", + }); + + rig.fake.holdAllocate(); + rig.fake.failNextRelease({ code: "COMMAND_FAILED", message: "release failed" }); + const abort = new AbortController(); + const second = track(allocate(rig, 2, { ttlMs: 60_000 }, { signal: abort.signal })); + await sleep(50); + expect(second.done()).toBe(false); + await expect.poll(() => allocatesReceived(rig, 2).length).toBe(1); + abort.abort(); + await second.promise.catch(() => undefined); + await sleep(50); + rig.fake.letAllocateGo(); + await expect.poll(() => rig.fake.rpcRequests("agent_device.lease.release").length).toBe(2); + + expect((await leases(rig)).map((lease) => lease.id)).toEqual([hex(2)]); + await advance(rig, 61_000); + await expect.poll(async () => (await leases(rig)).length).toBe(0); + }); + + it("a heartbeat reaches the agent-device daemon unchanged, its reply reaches the CLI, and the simlock lease's deadline does not move", async () => { + const rig = await setUp(); + await granted(rig, 1); + const [before] = await leases(rig); + rig.clock.advance(10_000); + const params = { ...scope, leaseId: hex(1), ttlMs: 90_000 }; + + const answer = await envelopeOf( + await rpc(rig, "agent_device.lease.heartbeat", params, { id: "hb-1" }), + ); + + expect(answer).toMatchObject({ id: "hb-1", result: { ok: true } }); + expect(rig.fake.rpcRequests("agent_device.lease.heartbeat")[0]?.params).toMatchObject(params); + expect((await leases(rig))[0]).toEqual(before); + }); + + it("release, and release after `close` (which the daemon answers `released: false`), go to the agent-device daemon first, then release the simlock lease, and the device is reclaimed and free", async () => { + const rig = await setUp(); + let heldAtDaemonRelease: readonly string[] = []; + rig.fake.onRelease = async () => { + heldAtDaemonRelease = (await leases(rig)).map((lease) => lease.id); + }; + await granted(rig, 1); + + const released = await envelopeOf(await release(rig, 1)); + + expect(released.result).toMatchObject({ ok: true, released: true }); + expect(heldAtDaemonRelease).toEqual([hex(1)]); + expect(await leases(rig)).toEqual([]); + await expect + .poll(async () => (await devices(rig)).map((device) => device.state)) + .toEqual(["ready"]); + + await granted(rig, 2); + await command(rig, "close", 2); + const afterClose = await envelopeOf(await release(rig, 2)); + + expect(afterClose.result).toMatchObject({ ok: true, released: false }); + expect(heldAtDaemonRelease).toEqual([hex(2)]); + expect(await leases(rig)).toEqual([]); + await expect + .poll(async () => (await devices(rig)).map((device) => device.state)) + .toEqual(["ready"]); + }); + + it("a release the agent-device daemon fails keeps the simlock lease and answers the daemon's error", async () => { + const rig = await setUp(); + await granted(rig, 1); + rig.fake.failNextRelease({ code: "COMMAND_FAILED", message: "daemon could not release" }); + + const answer = await envelopeOf(await release(rig, 1)); + + expect(answer.error).toMatchObject({ + data: { code: "COMMAND_FAILED" }, + message: "daemon could not release", + }); + expect((await leases(rig)).map((lease) => lease.id)).toEqual([hex(1)]); + }); + + it("a release after `simlock release --all` ended the simlock lease answers the daemon's reply", async () => { + const rig = await setUp(); + await granted(rig, 1); + await rig.daemon.dispatch("lease.release-all", {}, adminSession); + expect(await leases(rig)).toEqual([]); + + const answer = await envelopeOf(await release(rig, 1)); + + expect(answer.result).toMatchObject({ ok: true, released: true }); + expect(rig.fake.leases.has(hex(1))).toBe(false); + }); + + it("a release from a second token naming the first token's lease ID leaves the first token's simlock lease in place", async () => { + const rig = await setUp(); + await granted(rig, 1); + + const answer = await envelopeOf(await release(rig, 1, { token: secrets.two })); + + expect(answer.error).toBeDefined(); + expect(rig.fake.rpcRequests("agent_device.lease.release")).toHaveLength(1); + expect((await leases(rig)).map((lease) => lease.id)).toEqual([hex(1)]); + expect(rig.fake.leases.has(hex(1))).toBe(true); + + rig.fake.letAnyTenantRelease(); + const crossed = await envelopeOf(await release(rig, 1, { token: secrets.two })); + + expect(crossed.result).toMatchObject({ ok: true }); + expect(rig.fake.leases.has(hex(1))).toBe(false); + expect((await leases(rig)).map((lease) => lease.id)).toEqual([hex(1)]); + }); + + it("a command's reply, a streamed progress body included, reaches the CLI unchanged, and a CLI that hangs up during it aborts the request the agent-device daemon was answering", async () => { + const rig = await setUp(); + const chunks = ['{"progress":1}\n', '{"progress":2}\n', '{"done":true}\n']; + rig.fake.streamNextCommand(chunks); + + const streamed = await command(rig, "install", undefined); + + expect(await streamed.text()).toBe(chunks.join("")); + + const plain = await envelopeOf(await command(rig, "snapshot", undefined)); + expect(plain.result).toMatchObject({ command: "snapshot", ok: true }); + + rig.fake.holdCommand(); + const hangUp = new AbortController(); + const held = track(command(rig, "snapshot", undefined, { signal: hangUp.signal })); + await sleep(50); + expect(held.done()).toBe(false); + await expect.poll(() => rig.fake.rpcRequests("agent_device.command").length).toBe(3); + hangUp.abort(); + await held.promise.catch(() => undefined); + + await expect.poll(() => rig.fake.abortedCommands).toEqual(["snapshot"]); + }); + + it("an upload preflight answers a ticket URL under `/v1/agent-device/upload/direct/` on simlock's own origin, carrying the agent's own simlock token, and a `PUT` to it reaches the agent-device daemon", async () => { + const rig = await setUp(); + + const preflight = await send(rig, "/upload/preflight", { body: "{}" }); + expect(preflight.status).toBe(200); + const ticket = ( + (await preflight.json()) as { upload: { url: string; headers: Record } } + ).upload; + + expect(ticket.url.startsWith(`${rig.origin}/v1/agent-device/upload/direct/`)).toBe(true); + const carried = Object.entries(ticket.headers).find( + ([name]) => name.toLowerCase() === "authorization", + ); + expect(carried?.[1]).toBe(`Bearer ${secrets.one}`); + + const put = await fetch(ticket.url, { + body: "artifact bytes", + headers: ticket.headers, + method: "PUT", + }); + + expect(put.status).toBe(200); + expect(rig.fake.uploads).toEqual(["artifact bytes"]); + }); + + it("an `/rpc` body over 1 MiB, one that is not JSON, a JSON array, and an object with no `method` are each refused with their message and `id: null`, also while agent-device is unavailable, and neither simlock nor the agent-device daemon sees them", async () => { + const rig = await setUp(); + expect((await command(rig, "snapshot", undefined)).status).toBe(200); + const oneMiB = 1024 * 1024; + const bodies: readonly [string, string][] = [ + [ + JSON.stringify({ + id: 1, + jsonrpc: "2.0", + method: "agent_device.command", + params: { pad: "x".repeat(oneMiB) }, + }), + "refused: request body over 1 MiB", + ], + ["{ not json", "refused: send one JSON-RPC request"], + ["[]", "refused: send one JSON-RPC request"], + [JSON.stringify({ id: 3, jsonrpc: "2.0" }), "refused: send one JSON-RPC request"], + ]; + + for (const unavailable of [false, true]) { + rig.fake.setUnavailable(unavailable ? "starting" : undefined); + const receivedBefore = rig.fake.requests.length; + for (const [body, message] of bodies) { + const answer = await envelopeOf(await send(rig, "/rpc", { body })); + expect(answer).toMatchObject({ error: refusal(message), id: null, jsonrpc: "2.0" }); + } + expect(rig.fake.requests).toHaveLength(receivedBefore); + } + expect(await waiting(rig)).toEqual([]); + expect(await leases(rig)).toEqual([]); + }); +}); diff --git a/src/agent-device/front/allocate.test.ts b/src/agent-device/front/allocate.test.ts new file mode 100644 index 000000000..1a02cc2a9 --- /dev/null +++ b/src/agent-device/front/allocate.test.ts @@ -0,0 +1,156 @@ +import { describe, expect, it, vi } from "vitest"; + +import type { HttpDispatch } from "../../http/dispatcher-session.js"; +import type { TokenIdentity } from "../../http/token-store.js"; +import { FakeClock, NoopLogger } from "../../ports/index.js"; +import { allocate } from "./allocate.js"; +import type { FrontContext } from "./context.js"; +import { FrontMap } from "./front-map.js"; + +const agent: TokenIdentity = { requesterId: "tok_a", role: "agent" }; +const leaseId = "00000000000000000000000000000001"; + +interface AdmitSession { + readonly onRequestAdmitted?: (id: string, replayed: boolean) => void; +} +type Handler = (input: Record, session: AdmitSession) => Promise; + +function setUp() { + const clock = new FakeClock(1_000); + const map = new FrontMap(clock); + const hangUp = new AbortController(); + const calls: { readonly operation: string; readonly input: Record }[] = []; + const handlers: Record = { + "lease.list": async () => ({ leases: [] }), + "lease.request": async () => { + throw new Error("no device"); + }, + }; + const dispatch = (async (operation: string, input: Record, session: never) => { + calls.push({ input, operation }); + const handler = handlers[operation]; + if (handler === undefined) throw new Error(`unexpected ${operation}`); + return handler(input, session); + }) as unknown as HttpDispatch; + const context = { + clock, + dispatch, + link: { forward: vi.fn(), status: vi.fn() }, + logger: new NoopLogger(), + map, + } as unknown as FrontContext; + const run = () => + allocate(context, { + call: { + body: {}, + id: 1, + kind: "request", + method: "agent_device.lease.allocate", + params: { leaseId, platform: "ios" }, + text: "", + }, + identity: agent, + leaseMaxTtlMs: 600_000, + request: new Request("http://front.test/rpc", { method: "POST", signal: hangUp.signal }), + }); + const requests = () => calls.filter((call) => call.operation === "lease.request"); + const cancels = () => calls.filter((call) => call.operation === "lease.cancel"); + return { cancels, clock, handlers, hangUp, map, requests, run }; +} + +// why: the request must stay open until the test admits and settles it, to hang up in between. +function parkedRequest(rig: ReturnType) { + let admit!: () => void; + let settle!: () => void; + rig.handlers["lease.request"] = (_input, session) => + new Promise((_resolve, reject) => { + admit = () => session.onRequestAdmitted?.("req_1", false); + settle = () => reject(new Error("cancelled")); + }); + return { admit: () => admit(), settle: () => settle() }; +} + +describe("the allocate front", () => { + it("offers simlock no download: the lease request names no allowDownload", async () => { + const rig = setUp(); + + await rig.run(); + + expect(rig.requests()).toHaveLength(1); + expect(rig.requests()[0]?.input).not.toHaveProperty("allowDownload"); + }); + + it("cancels the lease request when the CLI hangs up before the daemon admits it, as soon as the daemon does", async () => { + const rig = setUp(); + const parked = parkedRequest(rig); + const done = rig.run(); + await vi.waitFor(() => expect(rig.requests()).toHaveLength(1)); + + rig.hangUp.abort(); + expect(rig.cancels()).toEqual([]); + parked.admit(); + + await vi.waitFor(() => expect(rig.cancels()).toHaveLength(1)); + expect(rig.cancels()[0]?.input).toEqual({ requesterId: leaseId }); + parked.settle(); + await done; + }); + + it("repeats the cancel every second while the request is not cancellable, and stops once it is", async () => { + const rig = setUp(); + const parked = parkedRequest(rig); + const answers = ["not-cancellable", "not-cancellable", "cancelled"]; + rig.handlers["lease.cancel"] = async () => ({ result: answers.shift() }); + const done = rig.run(); + await vi.waitFor(() => expect(rig.requests()).toHaveLength(1)); + parked.admit(); + + rig.hangUp.abort(); + await vi.waitFor(() => expect(rig.cancels()).toHaveLength(1)); + await vi.waitFor(() => expect(rig.clock.pendingTimerCount).toBe(1)); + rig.clock.advance(999); + expect(rig.cancels()).toHaveLength(1); + rig.clock.advance(1); + await vi.waitFor(() => expect(rig.cancels()).toHaveLength(2)); + await vi.waitFor(() => expect(rig.clock.pendingTimerCount).toBe(1)); + rig.clock.advance(1_000); + await vi.waitFor(() => expect(rig.cancels()).toHaveLength(3)); + + expect(rig.clock.pendingTimerCount).toBe(0); + parked.settle(); + await done; + expect(rig.cancels()).toHaveLength(3); + }); + + it("refreshes the front map at each allocate", async () => { + const rig = setUp(); + const refresh = vi.spyOn(rig.map, "refresh"); + + await rig.run(); + await rig.run(); + + expect(refresh).toHaveBeenCalledTimes(2); + }); + + it("gives up on a lease.list that does not answer within 5 s and goes on to the lease request", async () => { + const rig = setUp(); + rig.map.add("old", { address: "a", deadline: 500, identity: agent, ttlMs: 60_000 }); + let lists = 0; + rig.handlers["lease.list"] = async () => { + lists += 1; + return lists === 1 ? new Promise(() => undefined) : { leases: [] }; + }; + const done = rig.run(); + await vi.waitFor(() => expect(lists).toBe(1)); + + await new Promise((resolve) => setImmediate(resolve)); + rig.clock.advance(4_999); + await new Promise((resolve) => setImmediate(resolve)); + expect(rig.requests()).toEqual([]); + rig.clock.advance(1); + await done; + + expect(rig.requests()).toHaveLength(1); + expect(rig.map.get("old")?.endedAt).toBeUndefined(); + }); +}); diff --git a/src/agent-device/front/allocate.ts b/src/agent-device/front/allocate.ts new file mode 100644 index 000000000..3b44469df --- /dev/null +++ b/src/agent-device/front/allocate.ts @@ -0,0 +1,357 @@ +import type { z } from "zod"; + +import type { OPERATIONS } from "../../contract/index.js"; +import { buildHttpSession } from "../../http/dispatcher-session.js"; +import { mapError } from "../../http/errors.js"; +import type { TokenIdentity } from "../../http/token-store.js"; +import type { Clock } from "../../ports/index.js"; +import { frontReply, failure, refusal, type FrontReply } from "../shared/replies.js"; +import { leaseCallFields, leaseIdOf, type RpcRead } from "../shared/rpc.js"; +import { exchange, leaseCall, releaseSimlock, relay, type FrontContext } from "./context.js"; + +type Grant = z.infer<(typeof OPERATIONS)["lease.request"]["output"]>; + +const QUEUE_WAIT_MS = 330_000; +const MIN_TTL_MS = 5_000; +const CANCEL_RETRY_MS = 1_000; +const LEASE_GONE = new Set(["LEASE_NOT_FOUND", "LEASE_EXPIRED", "LEASE_REVOKED"]); +const CLASS_OF_TARGET: Record = { + desktop: "desktop", + mobile: "phone", + tv: "tv", +}; +const CALL_SCOPE_KEYS = ["runId", "leaseProvider", "clientId"] as const; +// why: only agent-device's reply names the device key it gave the lease, since the front stripped the CLI's key from the allocate it forwarded. +const REPLY_SCOPE_KEYS = [...CALL_SCOPE_KEYS, "deviceKey"] as const; + +const present = (value: unknown): boolean => value !== undefined && value !== null; + +interface Plan { + readonly leaseId: string; + readonly platform: "ios" | "android"; + readonly model?: string; + readonly class?: "phone" | "tv" | "desktop"; + readonly ttlMs?: number; +} + +function notPlain(params: Record): boolean { + const provider = params["leaseProvider"]; + return ( + present(params["backend"]) || + params["retainOnClose"] === true || + (present(provider) && provider !== "proxy") || + Object.entries(params).some( + ([key, value]) => present(value) && (key.startsWith("provider") || key.startsWith("aws")), + ) + ); +} + +function deviceRefusal(fields: ReturnType): FrontReply | undefined { + if (!present(fields.platform)) return refusal("refused: name --platform ios or android"); + if (fields.platform !== "ios" && fields.platform !== "android") { + return refusal("refused: only iOS and Android"); + } + if (present(fields.udid) || present(fields.serial)) { + return refusal("refused: simlock picks the device"); + } + return undefined; +} + +function refusalFor(params: Record): FrontReply | undefined { + const fields = leaseCallFields(params); + const ttl = fields.ttlMs; + if (leaseIdOf(fields.leaseId) === undefined) { + return refusal("refused: this agent-device is too old for simlock; upgrade it"); + } + if (notPlain(params)) return refusal("refused: simlock leases only plain devices"); + const device = deviceRefusal(fields); + if (device !== undefined) return device; + if (present(ttl) && !(typeof ttl === "number" && Number.isFinite(ttl) && ttl > 0)) { + return refusal("refused: ttlMs must be a positive number of milliseconds"); + } + return undefined; +} + +function planOf(params: Record): Plan { + const fields = leaseCallFields(params); + const device = + typeof fields.device === "string" && fields.device !== "" ? fields.device : undefined; + const klass = typeof fields.target === "string" ? CLASS_OF_TARGET[fields.target] : undefined; + return { + leaseId: fields.leaseId as string, + platform: fields.platform as "ios" | "android", + ...(device === undefined ? (klass === undefined ? {} : { class: klass }) : { model: device }), + ...(typeof fields.ttlMs === "number" ? { ttlMs: Math.max(fields.ttlMs, MIN_TTL_MS) } : {}), + }; +} + +function pause(clock: Clock, ms: number, wakeOn: (wake: () => void) => void): Promise { + return new Promise((resolve) => { + const timer = clock.setTimer(ms, resolve); + wakeOn(() => { + clock.cancel(timer); + resolve(); + }); + }); +} + +type LeaseOutcome = + | { readonly kind: "granted"; readonly grant: Grant } + | { readonly kind: "failed"; readonly reply: FrontReply } + | { readonly kind: "gone" }; + +// why: a request parked while the daemon starts cannot be called back, so wait on a read that parks at the same gate and give up if the CLI leaves, or a CLI that left would still be admitted. +async function stillWaiting( + context: FrontContext, + identity: TokenIdentity, + signal: AbortSignal, +): Promise { + if (signal.aborted) return false; + const gone = new Promise((resolve) => { + signal.addEventListener("abort", () => resolve(false), { once: true }); + }); + const ready = context.dispatch("lease.list", {}, buildHttpSession(identity)).then( + () => true, + () => true, + ); + return Promise.race([ready, gone]); +} + +async function requestLease( + context: FrontContext, + identity: TokenIdentity, + plan: Plan, + signal: AbortSignal, +): Promise { + if (!(await stillWaiting(context, identity, signal))) return { kind: "gone" }; + let admitted = false; + let aborted = signal.aborted; + let settled = false; + let cancelling = false; + let wakeCancel: (() => void) | undefined; + const session = buildHttpSession(identity); + const cancelUntilGone = async (): Promise => { + const until = context.clock.now() + QUEUE_WAIT_MS; + while (!settled && context.clock.now() < until) { + try { + const { result } = await context.dispatch( + "lease.cancel", + { requesterId: plan.leaseId }, + session, + ); + if (result !== "not-cancellable") return; + } catch { + return; + } + await pause(context.clock, CANCEL_RETRY_MS, (wake) => { + wakeCancel = wake; + }); + } + }; + const startCancel = (): void => { + if (cancelling) return; + cancelling = true; + void cancelUntilGone(); + }; + const onAbort = (): void => { + aborted = true; + if (admitted) startCancel(); + }; + signal.addEventListener("abort", onAbort, { once: true }); + context.map.startWaiting(plan.leaseId); + try { + const grant = await context.dispatch( + "lease.request", + { + leaseId: plan.leaseId, + platform: plan.platform, + requesterId: plan.leaseId, + timeoutMs: QUEUE_WAIT_MS, + ...(plan.model === undefined ? {} : { model: plan.model }), + ...(plan.class === undefined ? {} : { class: plan.class }), + ...(plan.ttlMs === undefined ? {} : { ttlMs: plan.ttlMs }), + }, + buildHttpSession(identity, { + onRequestAdmitted: () => { + admitted = true; + if (aborted) startCancel(); + }, + }), + ); + return { grant, kind: "granted" }; + } catch (error: unknown) { + const mapped = mapError(error); + return { + kind: "failed", + reply: + mapped.code === "QUEUE_TIMEOUT" + ? failure("timed out waiting for a device; nothing was leased", 504) + : refusal(`refused by simlock: ${mapped.message}`), + }; + } finally { + settled = true; + context.map.stopWaiting(plan.leaseId); + wakeCancel?.(); + signal.removeEventListener("abort", onAbort); + } +} + +function scopeOf( + source: Record | undefined, + keys: readonly string[], +): Record { + const scope: Record = {}; + for (const key of keys) { + if (source?.[key] !== undefined) scope[key] = source[key]; + } + return scope; +} + +async function releaseAgentDevice( + context: FrontContext, + request: Request, + identity: TokenIdentity, + leaseId: string, + scope: Record, +) { + return exchange( + context, + leaseCall(request, "agent_device.lease.release", { ...scope, leaseId }, "simlock-release"), + identity, + ); +} + +function forwardedBody( + call: RpcRead & { kind: "request" }, + plan: Plan, + address: string, + ceiling: number, +) { + const { device: _device, deviceKey: _deviceKey, target: _target, ...rest } = call.params; + return { + ...call.body, + params: { + ...rest, + ttlMs: ceiling, + ...(plan.platform === "ios" ? { udid: address } : { serial: address }), + }, + }; +} + +export interface AllocateArguments { + readonly request: Request; + readonly identity: TokenIdentity; + readonly call: RpcRead & { readonly kind: "request" }; + readonly leaseMaxTtlMs: number; +} + +interface Granted { + readonly request: Request; + readonly identity: TokenIdentity; + readonly call: RpcRead & { readonly kind: "request" }; + readonly plan: Plan; + readonly grant: Grant; + readonly address: string; +} + +export async function allocate( + context: FrontContext, + { request, identity, call, leaseMaxTtlMs }: AllocateArguments, +): Promise { + const refused = refusalFor(call.params); + if (refused !== undefined) return frontReply(refused, call); + const plan = planOf(call.params); + await context.map.refresh((who) => listLeases(context, who)); + if (request.signal.aborted) return undefined; + const outcome = await requestLease(context, identity, plan, request.signal); + if (outcome.kind === "gone") return undefined; + if (outcome.kind === "failed") return frontReply(outcome.reply, call); + const { grant } = outcome; + const address = grant.device.address; + if (request.signal.aborted || address === undefined || address === "") { + await releaseSimlock(context, identity, plan.leaseId); + return request.signal.aborted + ? undefined + : frontReply(failure("simlock granted a device with no address; nothing was leased"), call); + } + const forwarded = new Request(request.url, { + body: JSON.stringify(forwardedBody(call, plan, address, leaseMaxTtlMs)), + headers: leaseCall(request, "agent_device.lease.allocate", {}, null).headers, + method: "POST", + }); + const reply = await exchange(context, forwarded, identity); + const made = { address, call, grant, identity, plan, request }; + if (reply.reading.kind === "success") return keepOrRelease(context, made, reply, reply.reading); + if (reply.reading.kind === "error") { + await releaseSimlock(context, identity, plan.leaseId); + return relay(reply); + } + await settleNoReply(context, made); + return frontReply(failure("agent-device did not answer the allocate"), call); +} + +async function keepOrRelease( + context: FrontContext, + { request, identity, plan, grant, address }: Granted, + reply: Parameters[0], + reading: { readonly result: Record }, +): Promise { + if (!request.signal.aborted) { + context.map.add(plan.leaseId, { + address, + deadline: grant.lease.ttlDeadline, + identity, + ttlMs: grant.lease.ttlMs, + }); + return relay(reply); + } + const lease = reading.result["lease"]; + const named = + typeof lease === "object" && lease !== null ? (lease as Record) : {}; + const ended = await releaseAgentDevice( + context, + request, + identity, + plan.leaseId, + scopeOf(named, REPLY_SCOPE_KEYS), + ); + if (ended.reading.kind === "success") await releaseSimlock(context, identity, plan.leaseId); + return undefined; +} + +async function settleNoReply( + context: FrontContext, + { request, identity, call, plan }: Granted, +): Promise { + const ended = await releaseAgentDevice( + context, + request, + identity, + plan.leaseId, + scopeOf(call.params, CALL_SCOPE_KEYS), + ); + const holds = + ended.reading.kind === "error" && + (ended.reading.code === undefined || !LEASE_GONE.has(ended.reading.code)); + if (!holds) await releaseSimlock(context, identity, plan.leaseId); +} + +const LIST_TIMEOUT_MS = 5_000; + +async function listLeases(context: FrontContext, identity: TokenIdentity) { + let timer: ReturnType | undefined; + const timeout = new Promise((_resolve, reject) => { + timer = context.clock.setTimer(LIST_TIMEOUT_MS, () => + reject(new Error("lease.list timed out")), + ); + }); + try { + const { leases } = await Promise.race([ + context.dispatch("lease.list", {}, buildHttpSession(identity)), + timeout, + ]); + return leases.map((lease) => ({ id: lease.id, ttlDeadline: lease.ttlDeadline })); + } finally { + if (timer !== undefined) context.clock.cancel(timer); + } +} diff --git a/src/agent-device/front/context.ts b/src/agent-device/front/context.ts new file mode 100644 index 000000000..b70cf3a32 --- /dev/null +++ b/src/agent-device/front/context.ts @@ -0,0 +1,83 @@ +import type { HttpDispatch } from "../../http/dispatcher-session.js"; +import type { TokenIdentity } from "../../http/token-store.js"; +import type { Clock, Logger } from "../../ports/index.js"; +import { buildHttpSession } from "../../http/dispatcher-session.js"; +import { mapError } from "../../http/errors.js"; +import { readReply, type ReplyReading } from "../shared/rpc.js"; +import type { FrontMap } from "./front-map.js"; +import type { HostLink } from "./host-link.js"; + +export interface FrontContext { + readonly link: HostLink; + readonly map: FrontMap; + readonly dispatch: HttpDispatch; + readonly clock: Clock; + readonly logger: Logger; +} + +export interface Exchange { + readonly response: Response | undefined; + readonly text: string | undefined; + readonly reading: ReplyReading; +} + +export async function exchange( + context: FrontContext, + request: Request, + identity: TokenIdentity, +): Promise { + try { + const response = await context.link.forward(request, identity); + const text = await response.text(); + return { reading: readReply(text), response, text }; + } catch (error) { + context.logger.warn("agent-device lease call got no usable reply", { + message: error instanceof Error ? error.message : String(error), + }); + return { reading: { kind: "unusable" }, response: undefined, text: undefined }; + } +} + +export function relay(reply: { + response: Response | undefined; + text: string | undefined; +}): Response { + const headers = new Headers(reply.response?.headers); + headers.delete("content-length"); + headers.delete("content-encoding"); + return new Response(reply.text ?? "", { headers, status: reply.response?.status ?? 502 }); +} + +export function leaseCall( + origin: Request, + method: "agent_device.lease.allocate" | "agent_device.lease.release", + params: Record, + id: unknown, +): Request { + const headers = new Headers(origin.headers); + headers.delete("content-length"); + headers.set("content-type", "application/json"); + return new Request(origin.url, { + body: JSON.stringify({ id, jsonrpc: "2.0", method, params }), + headers, + method: "POST", + }); +} + +export async function releaseSimlock( + context: FrontContext, + identity: TokenIdentity, + leaseId: string, +): Promise { + try { + await context.dispatch("lease.release", { leaseId }, buildHttpSession(identity)); + return true; + } catch (error) { + if (mapError(error).code === "UNKNOWN_LEASE") return true; + context.logger.warn("agent-device front could not release a simlock lease", { + leaseId, + message: error instanceof Error ? error.message : String(error), + }); + return false; + } +} diff --git a/src/agent-device/front/front-map.test.ts b/src/agent-device/front/front-map.test.ts new file mode 100644 index 000000000..e06388851 --- /dev/null +++ b/src/agent-device/front/front-map.test.ts @@ -0,0 +1,100 @@ +import { describe, expect, it, vi } from "vitest"; + +import type { TokenIdentity } from "../../http/token-store.js"; +import { FakeClock } from "../../ports/index.js"; +import { FrontMap } from "./front-map.js"; + +const agent: TokenIdentity = { requesterId: "tok_a", role: "agent" }; +const ttlMs = 60_000; + +function entry(deadline: number) { + return { address: "device-addr", deadline, identity: agent, ttlMs }; +} + +describe("the front map", () => { + it("drops an ended entry one TTL after it ended, at the next read or write", () => { + const clock = new FakeClock(1_000); + const map = new FrontMap(clock); + map.add("lease-a", entry(70_000)); + map.markEnded("lease-a"); + + clock.advance(ttlMs - 1); + expect(map.get("lease-a")).toMatchObject({ endedAt: 1_000 }); + + clock.advance(1); + map.add("lease-b", entry(500_000)); + expect(map.get("lease-a")).toBeUndefined(); + expect(map.get("lease-b")).toBeDefined(); + expect(clock.pendingTimerCount).toBe(0); + }); + + it("at an allocate marks an entry past its latest deadline ended when lease.list no longer shows its lease, and lets one it still shows take the listed deadline", async () => { + const clock = new FakeClock(1_000); + const map = new FrontMap(clock); + map.add("gone", entry(2_000)); + map.add("kept", entry(2_000)); + map.add("fresh", entry(9_000_000)); + clock.advance(10_000); + const list = vi.fn(async () => [{ id: "kept", ttlDeadline: 500_000 }]); + + await map.refresh(list); + + expect(list).toHaveBeenCalledTimes(1); + expect(list).toHaveBeenCalledWith(agent); + expect(map.get("gone")?.endedAt).toBe(11_000); + expect(map.get("kept")).toMatchObject({ deadline: 500_000 }); + expect(map.get("kept")?.endedAt).toBeUndefined(); + expect(map.get("fresh")).toMatchObject({ deadline: 9_000_000 }); + expect(map.get("fresh")?.endedAt).toBeUndefined(); + }); + + it("keeps a lease ID in the waiting list while any allocate with it still waits", () => { + const map = new FrontMap(new FakeClock(1_000)); + map.startWaiting("lease-a"); + map.startWaiting("lease-a"); + map.startWaiting("lease-b"); + + map.stopWaiting("lease-a"); + expect(map.waitingIds()).toEqual(["lease-a", "lease-b"]); + + map.stopWaiting("lease-a"); + map.stopWaiting("lease-b"); + expect(map.waitingIds()).toEqual([]); + }); + + it("leaves every entry unchanged when lease.list fails for its principal", async () => { + const clock = new FakeClock(1_000); + const map = new FrontMap(clock); + map.add("lease-a", entry(2_000)); + clock.advance(10_000); + + await map.refresh(async () => { + throw new Error("lease.list refused"); + }); + + expect(map.get("lease-a")).toMatchObject({ deadline: 2_000 }); + expect(map.get("lease-a")?.endedAt).toBeUndefined(); + }); + + it("asks lease.list for at most 100 principals, the ones holding the oldest deadlines", async () => { + const clock = new FakeClock(1_000); + const map = new FrontMap(clock); + for (let n = 0; n < 101; n += 1) { + const identity: TokenIdentity = { requesterId: `tok_${n}`, role: "agent" }; + map.add(`lease-${n}`, { address: "a", deadline: 2_100 - n, identity, ttlMs }); + } + clock.advance(10_000); + const asked: string[] = []; + + await map.refresh(async (who) => { + asked.push(who.requesterId); + return []; + }); + + expect(asked).toHaveLength(100); + expect(asked).toContain("tok_100"); + expect(asked).not.toContain("tok_0"); + expect(map.get("lease-100")?.endedAt).toBeDefined(); + expect(map.get("lease-0")?.endedAt).toBeUndefined(); + }); +}); diff --git a/src/agent-device/front/front-map.ts b/src/agent-device/front/front-map.ts new file mode 100644 index 000000000..3cfdb4d05 --- /dev/null +++ b/src/agent-device/front/front-map.ts @@ -0,0 +1,110 @@ +import type { TokenIdentity } from "../../http/token-store.js"; +import type { Clock } from "../../ports/index.js"; + +const MAX_PRINCIPALS_PER_REFRESH = 100; + +export interface FrontMapEntry { + readonly identity: TokenIdentity; + readonly address: string; + readonly ttlMs: number; + readonly deadline: number; +} + +export interface ListedLease { + readonly id: string; + readonly ttlDeadline: number; +} + +interface Held { + readonly identity: TokenIdentity; + readonly address: string; + readonly ttlMs: number; + deadline: number; + endedAt?: number; +} + +export class FrontMap { + readonly #held = new Map(); + readonly #waiting = new Map(); + + startWaiting(leaseId: string): void { + this.#waiting.set(leaseId, (this.#waiting.get(leaseId) ?? 0) + 1); + } + + stopWaiting(leaseId: string): void { + const left = (this.#waiting.get(leaseId) ?? 0) - 1; + if (left > 0) this.#waiting.set(leaseId, left); + else this.#waiting.delete(leaseId); + } + + waitingIds(): readonly string[] { + return [...this.#waiting.keys()]; + } + + constructor(readonly clock: Clock) {} + + add(leaseId: string, entry: FrontMapEntry): void { + this.#prune(); + this.#held.set(leaseId, { ...entry }); + } + + get(leaseId: string): (FrontMapEntry & { readonly endedAt?: number }) | undefined { + this.#prune(); + return this.#held.get(leaseId); + } + + markEnded(leaseId: string): void { + const held = this.#held.get(leaseId); + if (held !== undefined) held.endedAt ??= this.clock.now(); + } + + async refresh(list: (identity: TokenIdentity) => Promise): Promise { + this.#prune(); + const now = this.clock.now(); + const byPrincipal = new Map< + string, + { identity: TokenIdentity; oldest: number; ids: string[] } + >(); + for (const [id, held] of this.#held) { + if (held.endedAt !== undefined || held.deadline > now) continue; + const group = byPrincipal.get(held.identity.requesterId); + if (group === undefined) { + byPrincipal.set(held.identity.requesterId, { + identity: held.identity, + ids: [id], + oldest: held.deadline, + }); + } else { + group.ids.push(id); + group.oldest = Math.min(group.oldest, held.deadline); + } + } + const groups = [...byPrincipal.values()] + .sort((a, b) => a.oldest - b.oldest) + .slice(0, MAX_PRINCIPALS_PER_REFRESH); + await Promise.all( + groups.map(async (group) => { + let listed: readonly ListedLease[]; + try { + listed = await list(group.identity); + } catch { + return; + } + for (const id of group.ids) { + const held = this.#held.get(id); + if (held === undefined) continue; + const found = listed.find((lease) => lease.id === id); + if (found === undefined) held.endedAt ??= this.clock.now(); + else held.deadline = found.ttlDeadline; + } + }), + ); + } + + #prune(): void { + const now = this.clock.now(); + for (const [id, held] of this.#held) { + if (held.endedAt !== undefined && now - held.endedAt >= held.ttlMs) this.#held.delete(id); + } + } +} diff --git a/src/agent-device/front/front.ts b/src/agent-device/front/front.ts new file mode 100644 index 000000000..290305673 --- /dev/null +++ b/src/agent-device/front/front.ts @@ -0,0 +1,101 @@ +import type { HttpMount } from "../../http/app.js"; +import type { TokenIdentity } from "../../http/token-store.js"; +import { FRONT_MOUNT, FRONT_PREFIX } from "../shared/paths.js"; +import { failure, frontReply, notAvailable, refusal } from "../shared/replies.js"; +import { classifyCall, readRpcBody, type RpcRead } from "../shared/rpc.js"; +import { allocate } from "./allocate.js"; +import type { FrontContext } from "./context.js"; +import { release } from "./release.js"; + +type RpcRequest = RpcRead & { readonly kind: "request" }; + +const AGENT_TOKEN_ONLY = "refused: agent-device through simlock needs an agent token"; +const CLIENT_GONE = 499; + +function carry(request: Request, body: string): Request { + const headers = new Headers(request.headers); + headers.delete("content-length"); + return new Request(request.url, { + body, + headers, + method: request.method, + signal: request.signal, + }); +} + +// why: the allocate keeps cleaning up (cancel, release) after the CLI hung up, while the handler answers at once. +function untilHangUp(signal: AbortSignal, work: Promise): Promise { + const gone = new Promise((resolve) => { + const answer = () => resolve(new Response(null, { status: CLIENT_GONE })); + if (signal.aborted) answer(); + else signal.addEventListener("abort", answer, { once: true }); + }); + return Promise.race([work.then((response) => response ?? gone), gone]); +} + +async function handleRpc( + context: FrontContext, + request: Request, + identity: TokenIdentity, +): Promise { + const read = await readRpcBody(request); + if (read.kind === "too-large") { + return frontReply(refusal("refused: request body over 1 MiB"), { id: null }); + } + if (read.kind === "malformed") { + return frontReply(refusal("refused: send one JSON-RPC request"), { id: null }); + } + const status = await context.link.status(); + if (status.state === "unavailable") return frontReply(notAvailable(status.reason), read); + const kind = classifyCall(read.method, read.params); + if (kind === "wrapped-lease") { + return frontReply(refusal("refused: send lease calls as lease RPCs"), read); + } + if (kind === "allocate") { + const work = allocate(context, { + call: read, + identity, + leaseMaxTtlMs: status.leaseMaxTtlMs, + request, + }).catch((error: unknown) => { + context.logger.error("agent-device allocate failed", { + message: error instanceof Error ? error.message : String(error), + }); + return frontReply(failure("agent-device allocate failed"), read); + }); + return untilHangUp(request.signal, work); + } + return kind === "release" + ? release(context, identity, read as RpcRequest, carry(request, read.text)) + : passThrough(context, carry(request, read.text), identity, read); +} + +async function passThrough( + context: FrontContext, + request: Request, + identity: TokenIdentity, + rpc: { readonly id: unknown } | undefined, +): Promise { + try { + return await context.link.forward(request, identity); + } catch (error) { + return frontReply(notAvailable(error instanceof Error ? error.message : String(error)), rpc); + } +} + +export function createFront(context: FrontContext): HttpMount { + return { + handle: async (request, identity) => { + const isRpc = + request.method === "POST" && new URL(request.url).pathname === `${FRONT_PREFIX}/rpc`; + if (identity.role !== "agent") { + return frontReply(refusal(AGENT_TOKEN_ONLY, 403), isRpc ? { id: null } : undefined); + } + if (isRpc) return handleRpc(context, request, identity); + const status = await context.link.status(); + if (status.state === "unavailable") return frontReply(notAvailable(status.reason), undefined); + return passThrough(context, request, identity, undefined); + }, + prefix: FRONT_MOUNT, + }; +} diff --git a/src/agent-device/front/host-link.ts b/src/agent-device/front/host-link.ts new file mode 100644 index 000000000..9dac488f1 --- /dev/null +++ b/src/agent-device/front/host-link.ts @@ -0,0 +1,22 @@ +import type { TokenIdentity } from "../../http/token-store.js"; +import type { AgentDeviceHost } from "../host/host.js"; + +export interface HostLink { + status(): Promise< + | { readonly state: "running"; readonly leaseMaxTtlMs: number } + | { readonly state: "unavailable"; readonly reason: string } + >; + forward(request: Request, identity: TokenIdentity): Promise; +} + +export class InProcessHostLink implements HostLink { + constructor(private readonly host: AgentDeviceHost) {} + + status(): ReturnType { + return this.host.status(); + } + + forward(request: Request, identity: TokenIdentity): Promise { + return this.host.forward(request, identity); + } +} diff --git a/src/agent-device/front/release.ts b/src/agent-device/front/release.ts new file mode 100644 index 000000000..793e75277 --- /dev/null +++ b/src/agent-device/front/release.ts @@ -0,0 +1,28 @@ +import type { TokenIdentity } from "../../http/token-store.js"; +import { failure, frontReply } from "../shared/replies.js"; +import { leaseCallFields, leaseIdOf, type RpcRead } from "../shared/rpc.js"; +import { exchange, relay, releaseSimlock, type FrontContext } from "./context.js"; + +export async function release( + context: FrontContext, + identity: TokenIdentity, + call: RpcRead & { readonly kind: "request" }, + forwarded: Request, +): Promise { + const reply = await exchange(context, forwarded, identity); + if (reply.response === undefined) { + return frontReply(failure("agent-device did not answer the release"), call); + } + const leaseId = leaseIdOf(leaseCallFields(call.params).leaseId); + const entry = leaseId === undefined ? undefined : context.map.get(leaseId); + if ( + leaseId !== undefined && + entry?.identity.requesterId === identity.requesterId && + entry.endedAt === undefined && + reply.reading.kind === "success" && + (await releaseSimlock(context, entry.identity, leaseId)) + ) { + context.map.markEnded(leaseId); + } + return relay(reply); +} diff --git a/src/agent-device/host/host.test.ts b/src/agent-device/host/host.test.ts new file mode 100644 index 000000000..a3497f9c5 --- /dev/null +++ b/src/agent-device/host/host.test.ts @@ -0,0 +1,43 @@ +import { describe, expect, it, vi } from "vitest"; + +import { AgentDeviceHost } from "./host.js"; +import type { AgentDeviceUpstream } from "./upstream.js"; + +vi.mock("@agent-device/proxy", () => { + throw new Error("module not found"); +}); + +const running = { + baseUrl: "http://127.0.0.1:47001", + generation: 1, + leaseMaxTtlMs: 600_000, + state: "running", + token: "daemon-token", +} as const; + +describe("the agent-device host", () => { + it("answers unavailable with `could not load @agent-device/proxy` when the proxy package fails to load, before any allocate is served", async () => { + const upstream = { + current: () => running, + fetch: async () => new Response(), + } as unknown as AgentDeviceUpstream; + const host = new AgentDeviceHost(upstream); + + expect(await host.status()).toEqual({ + reason: expect.stringMatching(/^could not load @agent-device\/proxy: ./), + state: "unavailable", + }); + }); + + it("passes the upstream's own reason through while the daemon is unavailable", async () => { + const upstream = { + current: () => ({ reason: "agent-device is stopped", state: "unavailable" }), + fetch: async () => new Response(), + } as unknown as AgentDeviceUpstream; + + expect(await new AgentDeviceHost(upstream).status()).toEqual({ + reason: "agent-device is stopped", + state: "unavailable", + }); + }); +}); diff --git a/src/agent-device/host/host.ts b/src/agent-device/host/host.ts new file mode 100644 index 000000000..77841d7f6 --- /dev/null +++ b/src/agent-device/host/host.ts @@ -0,0 +1,130 @@ +import { randomUUID } from "node:crypto"; + +import type { TokenIdentity } from "../../http/token-store.js"; +import { FRONT_PREFIX, PROXY_BASE_PATH } from "../shared/paths.js"; +import type { AgentDeviceUpstream } from "./upstream.js"; + +type ProxyModule = typeof import("@agent-device/proxy"); +type DaemonProxy = ReturnType; + +// why: the proxy hands `authenticate` a copy of the request, so the tag travels as a header the front always overwrites. +const TAG_HEADER = "x-simlock-front-tag"; +const PREFLIGHT_PATH = `${PROXY_BASE_PATH}/upload/preflight`; + +export type HostStatus = + | { readonly state: "running"; readonly leaseMaxTtlMs: number } + | { readonly state: "unavailable"; readonly reason: string }; + +function stripBodyHeaders(headers: Headers): Headers { + const copy = new Headers(headers); + copy.delete("content-length"); + copy.delete("content-encoding"); + return copy; +} + +export class AgentDeviceHost { + readonly #identities = new Map(); + #loading: Promise | undefined; + #loadError: string | undefined; + #proxy: { readonly generation: number; readonly proxy: DaemonProxy } | undefined; + + constructor(readonly upstream: AgentDeviceUpstream) {} + + async status(): Promise { + const current = this.upstream.current(); + if (current.state === "unavailable") return current; + await this.#load(); + if (this.#loadError !== undefined) return { reason: this.#loadError, state: "unavailable" }; + return { leaseMaxTtlMs: current.leaseMaxTtlMs, state: "running" }; + } + + tag(request: Request, identity: TokenIdentity): string { + const tag = randomUUID(); + this.#identities.set(tag, identity); + request.headers.set(TAG_HEADER, tag); + return tag; + } + + authenticate(request: Request): { readonly tenantId: string } | null { + const tag = request.headers.get(TAG_HEADER); + const identity = tag === null ? undefined : this.#identities.get(tag); + return identity === undefined ? null : { tenantId: identity.requesterId }; + } + + async forward(request: Request, identity: TokenIdentity): Promise { + const status = await this.status(); + if (status.state === "unavailable") throw new Error(status.reason); + const module = await this.#load(); + const current = this.upstream.current(); + if (module === undefined || current.state === "unavailable") { + throw new Error(this.#loadError ?? "unavailable"); + } + if (this.#proxy?.generation !== current.generation) { + this.#proxy = { + generation: current.generation, + proxy: module.createDaemonProxy({ + authenticate: (tagged) => this.authenticate(tagged), + upstreamBaseUrl: current.baseUrl, + upstreamFetch: (upstream) => this.upstream.fetch(upstream), + upstreamToken: current.token, + }), + }; + } + const proxied = this.#rewrite(request); + const tag = this.tag(proxied, identity); + try { + const response = await this.#proxy.proxy.handle(proxied); + return new URL(proxied.url).pathname === PREFLIGHT_PATH + ? await this.#withTicketPrefix(response) + : response; + } finally { + this.#identities.delete(tag); + } + } + + #load(): Promise { + this.#loading ??= import("@agent-device/proxy").then( + (module) => module, + (error: unknown) => { + const reason = error instanceof Error ? error.message : String(error); + this.#loadError = `could not load @agent-device/proxy: ${reason}`; + return undefined; + }, + ); + return this.#loading; + } + + #rewrite(request: Request): Request { + const url = new URL(request.url); + if (url.pathname === FRONT_PREFIX || url.pathname.startsWith(`${FRONT_PREFIX}/`)) { + url.pathname = PROXY_BASE_PATH + url.pathname.slice(FRONT_PREFIX.length); + } + return new Request(url, { + headers: request.headers, + method: request.method, + signal: request.signal, + ...(request.body === null ? {} : { body: request.body, duplex: "half" }), + } as RequestInit); + } + + async #withTicketPrefix(response: Response): Promise { + // why: the proxy builds the ticket URL from the path it saw, and the client reached it under `/v1`. + const text = await response.text(); + let body = text; + try { + const parsed = JSON.parse(text) as { upload?: { url?: unknown } }; + if (typeof parsed.upload?.url === "string") { + const ticket = new URL(parsed.upload.url); + ticket.pathname = `/v1${ticket.pathname}`; + parsed.upload.url = ticket.toString(); + body = JSON.stringify(parsed); + } + } catch { + body = text; + } + return new Response(body, { + headers: stripBodyHeaders(response.headers), + status: response.status, + }); + } +} diff --git a/src/agent-device/host/upstream.test.ts b/src/agent-device/host/upstream.test.ts new file mode 100644 index 000000000..0363f8e83 --- /dev/null +++ b/src/agent-device/host/upstream.test.ts @@ -0,0 +1,154 @@ +import { describe, expect, it, vi } from "vitest"; + +import { + FakeClock, + FakeProcessSupervisor, + FakeTcpProbe, + JsonLinesLogger, + MemoryFilesystem, + MemoryLogSink, + ScriptedProcessRunner, + type ScriptedProcessExpectation, +} from "../../ports/index.js"; +import { FakeAgentDeviceHealth } from "../testing.js"; +import { agentDeviceLeaseMaxTtlMs } from "../shared/daemon-environment.js"; +import { AgentDeviceSupervisor } from "./agent-device-supervisor.js"; +import { AgentDeviceHost } from "./host.js"; +import { SupervisorAgentDeviceUpstream, type AgentDeviceUpstream } from "./upstream.js"; + +const home = "/home/simlock/.simlock"; +const nodePath = "/opt/node/bin/node"; +const packageRoot = "/opt/ad/node_modules/agent-device"; +const realCli = `${packageRoot}/bin/agent-device.mjs`; +const internalEntry = `${packageRoot}/dist/src/internal/daemon.js`; +const httpPort = 4310; +const daemonToken = "daemon-token"; +const leaseMaxTtlMs = 4 * 60 * 60_000; + +async function arrange(version: string) { + const clock = new FakeClock(1_000_000); + const filesystem = new MemoryFilesystem(); + const probe = new FakeTcpProbe(); + const health = new FakeAgentDeviceHealth(); + health.respondWith({ rpcProtocolVersion: 2, service: "agent-device-daemon" }); + await filesystem.mkdirp(home); + await filesystem.mkdirp("/usr/local/bin"); + await filesystem.mkdirp(`${packageRoot}/bin`); + await filesystem.mkdirp(`${packageRoot}/dist/src/internal`); + await filesystem.writeFileAtomic(realCli, "#!/usr/bin/env node\n"); + filesystem.defineAttributes(realCli, { mode: 0o755 }); + await filesystem.writeFileAtomic(`${packageRoot}/package.json`, '{"name":"agent-device"}'); + await filesystem.writeFileAtomic(internalEntry, ""); + filesystem.defineSymlink("/usr/local/bin/agent-device", realCli); + const processes = new FakeProcessSupervisor([]); + const expectations: ScriptedProcessExpectation[] = [ + { + match: { args: [realCli, "--version"], command: nodePath }, + result: { code: 0, stderr: "", stdout: `${version}\n` }, + }, + { + hangs: true, + match: { args: [internalEntry, `--simlock-home=${home}`], command: nodePath }, + }, + ]; + const runner = new ScriptedProcessRunner(expectations); + const spawn = runner.spawn.bind(runner); + vi.spyOn(runner, "spawn").mockImplementation((command, args, options) => { + const handle = spawn(command, args, options); + if (command !== nodePath || args[0] !== internalEntry) return handle; + processes.markAlive(handle.pid); + void filesystem.writeFileAtomic( + `${home}/agent-device/daemon.json`, + JSON.stringify({ httpPort, pid: handle.pid, token: daemonToken, version }), + ); + probe.startListening(httpPort); + return handle; + }); + const supervisor = new AgentDeviceSupervisor({ + baseEnvironment: { HOME: "/home/operator", PATH: "/usr/local/bin:/usr/bin" }, + clock, + config: { command: "agent-device", enabled: true }, + driverEnvironment: {}, + filesystem, + healthProbe: health, + home, + leaseMaxTtlMs, + logger: new JsonLinesLogger({ clock, level: "debug", sink: new MemoryLogSink() }), + nodePath, + nodeVersion: "22.14.0", + platforms: ["ios", "android"], + processRunner: runner, + processSupervisor: processes, + tcpProbe: probe, + }); + return { clock, supervisor, upstream: new SupervisorAgentDeviceUpstream(supervisor) }; +} + +async function flush(): Promise { + await new Promise((resolve) => setImmediate(resolve)); +} + +async function started(harness: Awaited>): Promise { + let settled = false; + const starting = harness.supervisor.start().finally(() => { + settled = true; + }); + starting.catch(() => undefined); + for (let tick = 0; tick < 4_000 && !settled; tick += 1) { + await flush(); + harness.clock.advance(100); + } + await starting; +} + +describe("the agent-device upstream adapter over the supervisor", () => { + it("reports running with the supervisor's base URL, token, generation and the ceiling", async () => { + const harness = await arrange("0.21.25-dev"); + await started(harness); + const availability = harness.supervisor.availability(); + if (availability.state !== "running") throw new Error(`got ${availability.state}`); + + expect(harness.upstream.current()).toEqual({ + baseUrl: availability.baseUrl, + generation: availability.generation, + leaseMaxTtlMs: agentDeviceLeaseMaxTtlMs(leaseMaxTtlMs), + state: "running", + token: daemonToken, + }); + expect(availability.baseUrl).toContain(String(httpPort)); + }); + + it("reports unavailable with the supervisor's message", async () => { + const harness = await arrange("0.20.3"); + await started(harness); + + expect(harness.upstream.current()).toEqual({ + reason: "found 0.20.3, need 0.21.25 or later", + state: "unavailable", + }); + }); + + it("reports unavailable with starting while the supervisor starts", async () => { + const harness = await arrange("0.21.25-dev"); + const starting = harness.supervisor.start(); + starting.catch(() => undefined); + + expect(harness.upstream.current()).toEqual({ reason: "starting", state: "unavailable" }); + }); +}); + +describe("the agent-device host's authenticate", () => { + const upstream: AgentDeviceUpstream = { + current: () => ({ reason: "starting", state: "unavailable" }), + fetch: async () => new Response(null), + }; + + it("returns the tagged identity's principal as the tenant, and null for a request the front did not tag", () => { + const host = new AgentDeviceHost(upstream); + const tagged = new Request("http://worker/v1/agent-device/rpc"); + host.tag(tagged, { requesterId: "tok_agent", role: "agent" }); + + expect(host.authenticate(tagged)).toEqual({ tenantId: "tok_agent" }); + expect(host.authenticate(new Request("http://worker/v1/agent-device/rpc"))).toBeNull(); + }); +}); diff --git a/src/agent-device/host/upstream.ts b/src/agent-device/host/upstream.ts new file mode 100644 index 000000000..f63cacd6d --- /dev/null +++ b/src/agent-device/host/upstream.ts @@ -0,0 +1,44 @@ +import type { AgentDeviceAvailability } from "./agent-device-supervisor.js"; + +export type AgentDeviceUpstreamState = + | { + readonly state: "running"; + readonly baseUrl: string; + readonly token: string; + readonly generation: number; + readonly leaseMaxTtlMs: number; + } + | { readonly state: "unavailable"; readonly reason: string }; + +export interface AgentDeviceUpstream { + current(): AgentDeviceUpstreamState; + fetch(request: Request): Promise; +} + +export class SupervisorAgentDeviceUpstream implements AgentDeviceUpstream { + constructor(readonly supervisor: { availability(): AgentDeviceAvailability }) {} + + // fallow-ignore-next-line unused-class-member -- reached through the AgentDeviceUpstream port by the host. + current(): AgentDeviceUpstreamState { + const availability = this.supervisor.availability(); + switch (availability.state) { + case "running": + return { + baseUrl: availability.baseUrl, + generation: availability.generation, + leaseMaxTtlMs: availability.leaseMaxTtlMs, + state: "running", + token: availability.token, + }; + case "unavailable": + return { reason: availability.message, state: "unavailable" }; + default: + return { reason: availability.state === "off" ? "off" : "starting", state: "unavailable" }; + } + } + + // fallow-ignore-next-line unused-class-member -- reached through the AgentDeviceUpstream port by the host. + async fetch(request: Request): Promise { + return globalThis.fetch(request); + } +} diff --git a/src/agent-device/index.ts b/src/agent-device/index.ts index 7d711624c..f4be9891b 100644 --- a/src/agent-device/index.ts +++ b/src/agent-device/index.ts @@ -1,5 +1,43 @@ +import type { HttpDispatch } from "../http/dispatcher-session.js"; +import type { HttpMount } from "../http/app.js"; +import type { Clock, Logger } from "../ports/index.js"; +import { FrontMap } from "./front/front-map.js"; +import { createFront } from "./front/front.js"; +import { InProcessHostLink, type HostLink } from "./front/host-link.js"; +import { AgentDeviceHost } from "./host/host.js"; +import type { AgentDeviceUpstream } from "./host/upstream.js"; + export { agentDeviceFindings } from "./host/availability-findings.js"; export { AgentDeviceSupervisor } from "./host/agent-device-supervisor.js"; -// fallow-ignore-next-line unused-type -- the availability contract the front (#472) and doctor (#471) read. +// fallow-ignore-next-line unused-type -- the shape `AgentDeviceSupervisor.availability()` returns, which `agentDeviceFindings` reads. export type { AgentDeviceAvailability } from "./host/agent-device-supervisor.js"; +export type { AgentDeviceUpstream } from "./host/upstream.js"; +export { SupervisorAgentDeviceUpstream } from "./host/upstream.js"; export { type AgentDeviceHealthProbe, NodeAgentDeviceHealthProbe } from "./host/health.js"; + +export interface AgentDeviceEdgeParts { + readonly link: HostLink; + readonly map: FrontMap; +} + +export function createAgentDeviceEdgeParts( + upstream: AgentDeviceUpstream, + clock: Clock, +): AgentDeviceEdgeParts { + return { link: new InProcessHostLink(new AgentDeviceHost(upstream)), map: new FrontMap(clock) }; +} + +// why: a stop refuses every dispatch, so waiting allocates' lease requests are cancelled through leasing directly, before the gateway stops; otherwise they stay open across a restart. +export async function cancelWaitingAllocates( + parts: AgentDeviceEdgeParts, + cancel: (requesterId: string) => Promise, +): Promise { + await Promise.allSettled(parts.map.waitingIds().map((id) => cancel(id))); +} + +export function createAgentDeviceFront( + parts: AgentDeviceEdgeParts, + deps: { readonly dispatch: HttpDispatch; readonly clock: Clock; readonly logger: Logger }, +): HttpMount { + return createFront({ ...deps, link: parts.link, map: parts.map }); +} diff --git a/src/agent-device/shared/paths.ts b/src/agent-device/shared/paths.ts new file mode 100644 index 000000000..18a61fc15 --- /dev/null +++ b/src/agent-device/shared/paths.ts @@ -0,0 +1,5 @@ +export const FRONT_MOUNT = "agent-device"; + +export const FRONT_PREFIX = `/v1/${FRONT_MOUNT}`; + +export const PROXY_BASE_PATH = "/agent-device"; diff --git a/src/agent-device/shared/replies.ts b/src/agent-device/shared/replies.ts new file mode 100644 index 000000000..241b7d9fc --- /dev/null +++ b/src/agent-device/shared/replies.ts @@ -0,0 +1,44 @@ +export type ReplyCode = "INVALID_ARGS" | "COMMAND_FAILED"; + +export interface FrontReply { + readonly message: string; + readonly code: ReplyCode; + readonly status: number; +} + +export function notAvailable(reason: string): FrontReply { + return { + code: "COMMAND_FAILED", + message: `agent-device is not available on this simlock: ${reason}`, + status: 503, + }; +} + +export function refusal(message: string, status = 400): FrontReply { + return { code: "INVALID_ARGS", message, status }; +} + +export function failure(message: string, status = 502): FrontReply { + return { code: "COMMAND_FAILED", message, status }; +} + +export function frontReply(reply: FrontReply, rpc: { readonly id: unknown } | undefined): Response { + if (rpc === undefined) { + return Response.json( + { code: reply.code, error: reply.message, ok: false }, + { status: reply.status }, + ); + } + return Response.json( + { + error: { + code: -32000, + data: { code: reply.code, message: reply.message }, + message: reply.message, + }, + id: rpc.id, + jsonrpc: "2.0", + }, + { status: reply.status }, + ); +} diff --git a/src/agent-device/shared/rpc.test.ts b/src/agent-device/shared/rpc.test.ts new file mode 100644 index 000000000..ff964b359 --- /dev/null +++ b/src/agent-device/shared/rpc.test.ts @@ -0,0 +1,21 @@ +import { describe, expect, it } from "vitest"; + +import { commandLeaseId } from "./rpc.js"; + +const first = "a".repeat(32); +const second = "b".repeat(32); + +describe("a command's lease ID", () => { + it("is read from params.meta.leaseId, else params.flags.leaseId, and a value that is not 32 lowercase hex is no lease ID", () => { + expect(commandLeaseId({ meta: { leaseId: first } })).toBe(first); + expect(commandLeaseId({ flags: { leaseId: second } })).toBe(second); + expect(commandLeaseId({ flags: { leaseId: second }, meta: { leaseId: first } })).toBe(first); + + expect(commandLeaseId({ meta: { leaseId: first.toUpperCase() } })).toBeUndefined(); + expect(commandLeaseId({ meta: { leaseId: first.slice(1) } })).toBeUndefined(); + expect(commandLeaseId({ meta: { leaseId: `${first}0` } })).toBeUndefined(); + expect(commandLeaseId({ meta: { leaseId: `${"g".repeat(32)}` } })).toBeUndefined(); + expect(commandLeaseId({ meta: { leaseId: 7 } })).toBeUndefined(); + expect(commandLeaseId({})).toBeUndefined(); + }); +}); diff --git a/src/agent-device/shared/rpc.ts b/src/agent-device/shared/rpc.ts new file mode 100644 index 000000000..603584251 --- /dev/null +++ b/src/agent-device/shared/rpc.ts @@ -0,0 +1,136 @@ +const MAX_RPC_BODY_BYTES = 1024 * 1024; +const LEASE_ID = /^[0-9a-f]{32}$/; + +export function leaseIdOf(value: unknown): string | undefined { + return typeof value === "string" && LEASE_ID.test(value) ? value : undefined; +} + +function record(value: unknown): Record | undefined { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? (value as Record) + : undefined; +} + +export type RpcRead = + | { + readonly kind: "request"; + readonly text: string; + readonly body: Record; + readonly id: unknown; + readonly method: string; + readonly params: Record; + } + | { readonly kind: "too-large" } + | { readonly kind: "malformed" }; + +async function readBounded(body: ReadableStream | null): Promise { + if (body === null) return ""; + const reader = body.getReader(); + const chunks: Uint8Array[] = []; + let size = 0; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + size += value.byteLength; + if (size > MAX_RPC_BODY_BYTES) { + await reader.cancel(); + return undefined; + } + chunks.push(value); + } + return Buffer.concat(chunks).toString("utf8"); +} + +export async function readRpcBody(request: Request): Promise { + const text = await readBounded(request.body); + if (text === undefined) return { kind: "too-large" }; + let parsed: unknown; + try { + parsed = JSON.parse(text); + } catch { + return { kind: "malformed" }; + } + const body = record(parsed); + if (body === undefined || typeof body["method"] !== "string") return { kind: "malformed" }; + return { + body, + id: body["id"] ?? null, + kind: "request", + method: body["method"], + params: record(body["params"]) ?? {}, + text, + }; +} + +export type CallClass = "allocate" | "heartbeat" | "release" | "wrapped-lease" | "command"; + +const LEASE_METHOD = /^agent[_-]device\.lease\.(allocate|heartbeat|release)$/; +const COMMAND_METHOD = /^agent[_-]device\.command$/; +const WRAPPED_LEASE = /^lease_(allocate|heartbeat|release)$/; + +export function classifyCall(method: string, params: Record): CallClass { + const lease = LEASE_METHOD.exec(method)?.[1]; + if (lease !== undefined) return lease as "allocate" | "heartbeat" | "release"; + if (COMMAND_METHOD.test(method)) { + const command = params["command"]; + if (typeof command === "string" && WRAPPED_LEASE.test(command)) return "wrapped-lease"; + } + return "command"; +} + +export interface LeaseCallFields { + readonly leaseId: unknown; + readonly ttlMs: unknown; + readonly platform: unknown; + readonly target: unknown; + readonly device: unknown; + readonly udid: unknown; + readonly serial: unknown; + readonly deviceKey: unknown; +} + +export function leaseCallFields(params: Record): LeaseCallFields { + return { + device: params["device"], + deviceKey: params["deviceKey"], + leaseId: params["leaseId"], + platform: params["platform"], + serial: params["serial"], + target: params["target"], + ttlMs: params["ttlMs"], + udid: params["udid"], + }; +} + +export function commandLeaseId(params: unknown): string | undefined { + const fields = record(params); + const meta = record(fields?.["meta"]); + const flags = record(fields?.["flags"]); + return leaseIdOf(meta?.["leaseId"] ?? flags?.["leaseId"]); +} + +export type ReplyReading = + | { readonly kind: "success"; readonly result: Record } + | { readonly kind: "error"; readonly code: string | undefined } + | { readonly kind: "unusable" }; + +export function readReply(text: string): ReplyReading { + let parsed: unknown; + try { + parsed = JSON.parse(text); + } catch { + return { kind: "unusable" }; + } + const body = record(parsed); + if (body === undefined || body["jsonrpc"] !== "2.0") return { kind: "unusable" }; + const error = record(body["error"]); + if (error !== undefined) { + const code = record(error["data"])?.["code"]; + return { code: typeof code === "string" ? code : undefined, kind: "error" }; + } + const result = record(body["result"]); + if (result !== undefined && result["ok"] === true && body["error"] === undefined) { + return { kind: "success", result }; + } + return { kind: "unusable" }; +} diff --git a/src/agent-device/testing.ts b/src/agent-device/testing.ts index a80d577d8..d5f31a593 100644 --- a/src/agent-device/testing.ts +++ b/src/agent-device/testing.ts @@ -1,4 +1,5 @@ import type { AgentDeviceHealthProbe } from "./host/health.js"; +import type { AgentDeviceUpstream, AgentDeviceUpstreamState } from "./host/upstream.js"; type Answer = { readonly body: unknown } | { readonly error: Error } | "never"; @@ -22,3 +23,300 @@ export class FakeAgentDeviceHealth implements AgentDeviceHealthProbe { return this.#answer.body; } } + +export interface FakeAgentDeviceRequest { + readonly method: string; + readonly path: string; + readonly tenant: string | undefined; + readonly rpcMethod?: string; + readonly params?: Readonly>; +} + +interface FakeAgentDeviceLease { + readonly leaseId: string; + readonly tenant: string | undefined; + readonly scope: { + readonly runId: unknown; + readonly leaseProvider: unknown; + readonly clientId: unknown; + readonly deviceKey: unknown; + }; + readonly ttlMs: unknown; + ended: boolean; +} + +interface FakeFailure { + readonly code: string; + readonly message: string; +} + +interface Gate { + readonly promise: Promise; + open(): void; +} + +function gate(): Gate { + let open!: () => void; + const promise = new Promise((resolve) => { + open = resolve; + }); + return { open, promise }; +} + +async function readRpc(request: Request, tenantHeader: string | undefined) { + const body = (await request.json()) as { + readonly id?: unknown; + readonly method?: string; + readonly params?: Record; + }; + const params = body.params ?? {}; + const meta = (params["meta"] ?? {}) as Record; + const stamped = params["tenantId"] ?? meta["tenantId"]; + const tenant = tenantHeader ?? (typeof stamped === "string" ? stamped : undefined); + return { id: body.id, meta, method: body.method ?? "", params, tenant }; +} + +function envelope(id: unknown, result: unknown): Response { + return Response.json({ id, jsonrpc: "2.0", result }); +} + +function errorEnvelope(id: unknown, failure: FakeFailure): Response { + return Response.json({ + error: { code: -32000, data: failure, message: failure.message }, + id, + jsonrpc: "2.0", + }); +} + +export class FakeAgentDevice implements AgentDeviceUpstream { + readonly requests: FakeAgentDeviceRequest[] = []; + readonly leases = new Map(); + readonly abortedCommands: string[] = []; + readonly uploads: string[] = []; + baseUrl = "http://127.0.0.1:47001"; + token = "daemon-token-1"; + generation = 1; + leaseMaxTtlMs: number; + onRelease: (() => Promise) | undefined; + #unavailable: string | undefined; + #unreachable = false; + #failAllocate: FakeFailure | undefined; + #failRelease: FakeFailure | undefined; + #allocateBody: string | undefined; + #replyDeviceKey: string | undefined; + #anyTenantReleases = false; + #allocateGate: Gate | undefined; + #commandGate: Gate | undefined; + #streamChunks: readonly string[] | undefined; + #restarts = 0; + + constructor(options: { readonly leaseMaxTtlMs?: number } = {}) { + this.leaseMaxTtlMs = options.leaseMaxTtlMs ?? 4 * 60 * 60_000 + 10_000; + } + + // fallow-ignore-next-line unused-class-member -- reached through the AgentDeviceUpstream port by the host. + current(): AgentDeviceUpstreamState { + if (this.#unavailable !== undefined) { + return { reason: this.#unavailable, state: "unavailable" }; + } + return { + baseUrl: this.baseUrl, + generation: this.generation, + leaseMaxTtlMs: this.leaseMaxTtlMs, + state: "running", + token: this.token, + }; + } + + setUnavailable(reason: string | undefined): void { + this.#unavailable = reason; + } + + setUnreachable(unreachable: boolean): void { + this.#unreachable = unreachable; + } + + failNextAllocate(failure: FakeFailure): void { + this.#failAllocate = failure; + } + + failNextRelease(failure: FakeFailure): void { + this.#failRelease = failure; + } + + answerNextAllocateWith(body: string): void { + this.#allocateBody = body; + } + + nameDeviceKeyInNextReply(deviceKey: string): void { + this.#replyDeviceKey = deviceKey; + } + + // why: a daemon that finds a lease whichever tenant names it lets a cross-tenant release through to agent-device; only the simlock release is guarded by the front's owner check. + letAnyTenantRelease(): void { + this.#anyTenantReleases = true; + } + + holdAllocate(): void { + this.#allocateGate = gate(); + } + + letAllocateGo(): void { + this.#allocateGate?.open(); + } + + holdCommand(): void { + this.#commandGate = gate(); + } + + streamNextCommand(chunks: readonly string[]): void { + this.#streamChunks = chunks; + } + + restart(): void { + this.#restarts += 1; + this.generation += 1; + this.baseUrl = `http://127.0.0.1:${47_001 + this.#restarts}`; + this.token = `daemon-token-${this.generation}`; + this.leases.clear(); + } + + rpcRequests(rpcMethod: string): readonly FakeAgentDeviceRequest[] { + return this.requests.filter((request) => request.rpcMethod === rpcMethod); + } + + // fallow-ignore-next-line unused-class-member -- reached through the AgentDeviceUpstream port by the host. + async fetch(request: Request): Promise { + if (this.#unreachable) throw new TypeError("fetch failed"); + const url = new URL(request.url); + const base = new URL(this.baseUrl); + if (url.origin !== base.origin) throw new TypeError("fetch failed: wrong origin"); + if (request.headers.get("authorization") !== `Bearer ${this.token}`) { + return Response.json( + { code: "UNAUTHORIZED", error: "bad token", ok: false }, + { status: 401 }, + ); + } + const tenantHeader = request.headers.get("x-agent-device-tenant") ?? undefined; + if (url.pathname === "/health") { + this.requests.push({ method: request.method, path: url.pathname, tenant: tenantHeader }); + return Response.json({ ok: true, rpcProtocolVersion: 2, service: "agent-device-daemon" }); + } + if (url.pathname === "/upload/preflight") { + this.requests.push({ method: request.method, path: url.pathname, tenant: tenantHeader }); + return Response.json({ + ok: true, + upload: { headers: {}, url: `${this.baseUrl}/upload/direct/ticket-1` }, + }); + } + if (url.pathname.startsWith("/upload/direct/")) { + this.requests.push({ method: request.method, path: url.pathname, tenant: tenantHeader }); + this.uploads.push(await request.text()); + return Response.json({ ok: true }); + } + if (url.pathname !== "/rpc") return new Response("Not found", { status: 404 }); + return this.#rpc(request, tenantHeader); + } + + async #rpc(request: Request, tenantHeader: string | undefined): Promise { + const call = await readRpc(request, tenantHeader); + const { id, meta, method, params, tenant } = call; + this.requests.push({ method: request.method, params, path: "/rpc", rpcMethod: method, tenant }); + switch (method.replace(/^agent[_-]device\./, "")) { + case "lease.allocate": + return this.#allocate(id, params, tenant); + case "lease.heartbeat": + return envelope(id, { lease: params, ok: true }); + case "lease.release": + await this.onRelease?.(); + return this.#release(id, params, tenant); + default: + return this.#command(request, id, params, meta); + } + } + + async #allocate( + id: unknown, + params: Record, + tenant: string | undefined, + ): Promise { + if (this.#allocateGate !== undefined) await this.#allocateGate.promise; + const raw = this.#allocateBody; + this.#allocateBody = undefined; + if (raw !== undefined) return new Response(raw, { status: 200 }); + const failure = this.#failAllocate; + this.#failAllocate = undefined; + if (failure !== undefined) return errorEnvelope(id, failure); + const leaseId = String(params["leaseId"]); + const scope = { + clientId: params["clientId"], + deviceKey: this.#replyDeviceKey ?? params["deviceKey"], + leaseProvider: params["leaseProvider"], + runId: params["runId"], + }; + this.#replyDeviceKey = undefined; + this.leases.set(leaseId, { ended: false, leaseId, scope, tenant, ttlMs: params["ttlMs"] }); + return envelope(id, { + lease: { ...scope, leaseId, tenantId: tenant, ttlMs: params["ttlMs"] }, + ok: true, + }); + } + + #release(id: unknown, params: Record, tenant: string | undefined): Response { + const failure = this.#failRelease; + this.#failRelease = undefined; + if (failure !== undefined) return errorEnvelope(id, failure); + const lease = this.leases.get(String(params["leaseId"])); + if (lease === undefined || (!this.#anyTenantReleases && lease.tenant !== tenant)) { + return errorEnvelope(id, { code: "LEASE_NOT_FOUND", message: "lease not found" }); + } + if ( + lease.scope.runId !== params["runId"] || + lease.scope.leaseProvider !== params["leaseProvider"] || + lease.scope.clientId !== params["clientId"] || + lease.scope.deviceKey !== params["deviceKey"] + ) { + return errorEnvelope(id, { code: "UNAUTHORIZED", message: "lease scope mismatch" }); + } + this.leases.delete(lease.leaseId); + return envelope(id, { ok: true, released: !lease.ended }); + } + + async #command( + request: Request, + id: unknown, + params: Record, + meta: Record, + ): Promise { + const command = String(params["command"] ?? ""); + if (this.#commandGate !== undefined) { + const aborted = new Promise((_resolve, reject) => { + request.signal.addEventListener("abort", () => { + this.abortedCommands.push(command); + reject(new Error("aborted")); + }); + }); + await Promise.race([this.#commandGate.promise, aborted]); + } + const chunks = this.#streamChunks; + this.#streamChunks = undefined; + if (chunks !== undefined) { + const encoder = new TextEncoder(); + return new Response( + new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(encoder.encode(chunk)); + controller.close(); + }, + }), + { headers: { "content-type": "application/x-ndjson" } }, + ); + } + const leaseId = meta["leaseId"]; + if (command === "close" && typeof leaseId === "string") { + const lease = this.leases.get(leaseId); + if (lease !== undefined) lease.ended = true; + } + return envelope(id, { command, ok: true }); + } +} diff --git a/src/daemon/main.ts b/src/daemon/main.ts index ad5096a29..e59280425 100644 --- a/src/daemon/main.ts +++ b/src/daemon/main.ts @@ -5,7 +5,12 @@ import { pathToFileURL } from "node:url"; import { agentDeviceFindings, AgentDeviceSupervisor, + cancelWaitingAllocates, + createAgentDeviceEdgeParts, + createAgentDeviceFront, + SupervisorAgentDeviceUpstream, type AgentDeviceHealthProbe, + type AgentDeviceUpstream, NodeAgentDeviceHealthProbe, } from "../agent-device/index.js"; import { EVENT_FILE_NAME, EventBus, EventHistory, type EventBusLogger } from "../bus/index.js"; @@ -51,6 +56,7 @@ import { iosPrerequisites, } from "../drivers/ios/index.js"; import { createHttpApp } from "../http/app.js"; +import type { HttpDispatch } from "../http/dispatcher-session.js"; import { HttpGateway } from "../http/server.js"; import { TokenStore } from "../http/token-store.js"; import { @@ -108,6 +114,7 @@ import { createCredentialRoleResolver } from "./session.js"; export interface StartDaemonOptions { readonly agentDeviceHealth?: AgentDeviceHealthProbe; + readonly agentDeviceUpstream?: AgentDeviceUpstream; readonly clock?: Clock; readonly configOverrides?: ConfigOverrides; readonly configPath?: string; @@ -426,6 +433,13 @@ export async function startDaemon(options: StartDaemonOptions = {}): Promise core.closeWarmPool(), @@ -505,6 +519,9 @@ export async function startDaemon(options: StartDaemonOptions = {}): Promise { + if (agentDeviceParts !== undefined) { + await cancelWaitingAllocates(agentDeviceParts, (id) => leasing.cancelPending(id)); + } await gatewayStarted.catch(() => undefined); await stopHttpGateway?.(); // Stopped with the other auxiliary frontends, and for the same reason: the uplink is a @@ -525,7 +542,7 @@ export async function startDaemon(options: StartDaemonOptions = {}): Promise { const httpLogger = logger.child("http"); + const dispatch: HttpDispatch = (operation, input, session) => + daemon.dispatch(operation, input, session); const app = createHttpApp({ clock, config, - dispatch: (operation, input, session) => daemon.dispatch(operation, input, session), + dispatch, eventBus, idGenerator, leaseRequests: leasing.requests, logger: httpLogger, + mounts: + agentDeviceParts === undefined + ? [] + : [ + createAgentDeviceFront(agentDeviceParts, { + clock, + dispatch, + logger: logger.child("agent-device"), + }), + ], ownerRoutedFacts: daemon.ownerRoutedFacts, registry, tokens, diff --git a/src/http/app.ts b/src/http/app.ts index b9c45700a..1edd9787c 100644 --- a/src/http/app.ts +++ b/src/http/app.ts @@ -1,5 +1,5 @@ import { zValidator } from "@hono/zod-validator"; -import { Hono } from "hono"; +import { type Context, Hono } from "hono"; import { z } from "zod"; import type { EventBus } from "../bus/index.js"; @@ -80,6 +80,12 @@ export interface HttpGatewayDeps { /** `false` on a gateway, which never downloads: `allowDownload` then does not answer the lease * request `POST` early. Default `true`. */ readonly answerDownloadsEarly?: boolean | undefined; + readonly mounts?: readonly HttpMount[]; +} + +export interface HttpMount { + readonly prefix: string; + handle(request: Request, identity: TokenIdentity): Promise; } type Env = AuthEnv; @@ -713,6 +719,12 @@ export function createHttpApp(deps: HttpGatewayDeps): Hono & HttpAppDisposa }); }); + for (const mount of deps.mounts ?? []) { + const handler = (c: Context) => mount.handle(c.req.raw, c.get("identity")); + app.all(`/v1/${mount.prefix}`, agentAuth, handler); + app.all(`/v1/${mount.prefix}/*`, agentAuth, handler); + } + // `notices` holds a subscription for the app's lifetime -- exposed here rather than left to // leak, so a caller composing this app into a longer-lived process (the daemon) can // unsubscribe it on shutdown. diff --git a/src/source-hygiene.test.ts b/src/source-hygiene.test.ts index c5310fc52..2a2f52b3b 100644 --- a/src/source-hygiene.test.ts +++ b/src/source-hygiene.test.ts @@ -9,6 +9,10 @@ const REPO_ROOT = join(dirname(fileURLToPath(import.meta.url)), ".."); const NUL = 0x00; +/** why: built binaries committed for tests or vendored packages are never searched as text. */ +const isCommittedBinary = (name: string): boolean => + name.startsWith("e2e/fixtures/") || (name.startsWith("vendor/") && name.endsWith(".tgz")); + /** * why: a raw NUL byte in a text file makes `grep` and `ripgrep` classify the whole file as * binary and print `binary file matches` instead of the matching lines -- so the file @@ -17,8 +21,8 @@ const NUL = 0x00; * * Enumerated from `git ls-files` rather than a directory walk so the check sees every * tracked file in the repository (testing rule 4), not just the directories this test - * happened to name. The one exception is `e2e/fixtures/`: built binaries committed for tests, - * never searched as text, so they are skipped. Untracked and ignored files are out of scope: they are not part of + * happened to name. The exceptions are `e2e/fixtures/` and the vendored `vendor/*.tgz` tarballs: built binaries + * committed for tests, never searched as text, so they are skipped. Untracked and ignored files are out of scope: they are not part of * what anyone clones or searches. */ describe("tracked files", () => { @@ -30,7 +34,7 @@ describe("tracked files", () => { .toString("utf8") .split("\0") .filter((name) => name.length > 0) - .filter((name) => !name.startsWith("e2e/fixtures/")) + .filter((name) => !isCommittedBinary(name)) // A submodule or a path deleted from the working tree is still listed by // `git ls-files`; only regular files can be read. .filter((name) => { @@ -47,12 +51,15 @@ describe("tracked files", () => { expect(trackedFiles.length).toBeGreaterThan(100); }); - it("hold no path under e2e/fixtures/ and still hold e2e/appduct-fixtures.test.ts", () => { - expect(trackedFiles.filter((name) => name.startsWith("e2e/fixtures/"))).toEqual([]); + it("hold no committed binary and still hold e2e/appduct-fixtures.test.ts", () => { + expect(trackedFiles.filter(isCommittedBinary)).toEqual([]); + expect(trackedFiles.some((name) => name.startsWith("vendor/") && name.endsWith(".tgz"))).toBe( + false, + ); expect(trackedFiles).toContain("e2e/appduct-fixtures.test.ts"); }); - it("outside e2e/fixtures/ contain no raw NUL byte, so grep and ripgrep can search them", () => { + it("outside the committed binaries contain no raw NUL byte, so grep and ripgrep can search them", () => { const offenders = trackedFiles .map((name) => ({ name, contents: readFileSync(join(REPO_ROOT, name)) })) .filter(({ contents }) => contents.includes(NUL)) diff --git a/vendor/README.md b/vendor/README.md new file mode 100644 index 000000000..5ca546bd6 --- /dev/null +++ b/vendor/README.md @@ -0,0 +1,13 @@ +# vendor + +`agent-device-proxy-0.21.27-dev-54e4e4d.tgz` is `@agent-device/proxy`, packed +from the `simlock-integration` branch of agent-device at commit +`54e4e4db971d75ba1ebeb75c0ab7bce45c7d9b28`: + +```bash +pnpm --filter @agent-device/proxy pack --pack-destination +``` + +Its `prepack` runs the package's own build, which bundles agent-device's +internal packages into `dist`, so the tarball has no `workspace:` dependency. +It is swapped for the npm release later. diff --git a/vendor/agent-device-proxy-0.21.27-dev-54e4e4d.tgz b/vendor/agent-device-proxy-0.21.27-dev-54e4e4d.tgz new file mode 100644 index 000000000..a7b1c2217 Binary files /dev/null and b/vendor/agent-device-proxy-0.21.27-dev-54e4e4d.tgz differ