From 5e46859748ac616aad34bd4298e85f562a8963fa Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:19:38 +0200 Subject: [PATCH 01/20] test: an operator streams a device through a gateway (#909) --- src/daemon/stream-gateway.e2e.test.ts | 25 ++++++++++++++++++++++++- 1 file changed, 24 insertions(+), 1 deletion(-) diff --git a/src/daemon/stream-gateway.e2e.test.ts b/src/daemon/stream-gateway.e2e.test.ts index 940642611..52720adc2 100644 --- a/src/daemon/stream-gateway.e2e.test.ts +++ b/src/daemon/stream-gateway.e2e.test.ts @@ -140,12 +140,19 @@ describe("a screen stream through a gateway", () => { agent, ); const leaseId = (granted as { lease: { id: string } }).lease.id; + const deviceId = (granted as { lease: { deviceId: string } }).lease.deviceId; + const { workers } = await gateway.dispatch("worker.list", {}, admin); + const workerId = workers[0]?.id ?? ""; + const deviceStream = (path = `${workerId}/devices/${deviceId}`) => + fetch(`http://127.0.0.1:${String(port)}/v1/stream/workers/${path}`, { + headers: { authorization: `Bearer ${operator}` }, + }); const stream = (signal?: AbortSignal) => fetch(`http://127.0.0.1:${String(port)}/v1/stream/leases/${encodeURIComponent(leaseId)}`, { headers: { authorization: `Bearer ${operator}` }, ...(signal === undefined ? {} : { signal }), }); - return { gateway, leaseId, screen, stream }; + return { deviceStream, gateway, leaseId, screen, stream }; } it( @@ -174,6 +181,22 @@ describe("a screen stream through a gateway", () => { }, ); + it( + "relays an operator's device stream through the gateway: the worker's frames", + { timeout: 30_000 }, + async () => { + const { deviceStream } = await fleet(true); + const response = await deviceStream(); + expect(response.status).toBe(200); + expect(response.headers.get("simlock-stream-codec")).toBe("h264"); + const reader = response.body?.getReader(); + if (reader === undefined) throw new Error("no body"); + const first = await reader.read(); + expect(first.value?.[4]).toBe(SCREEN_FRAME_TAG.meta); + await reader.cancel(); + }, + ); + it( "a viewer leaving the gateway stops the capture on the worker", { timeout: 30_000 }, From 484383befd739435ae2f016c4de46e64b05b1060 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:22:28 +0200 Subject: [PATCH 02/20] feat(stream): a gateway relays a device's stream from a worker (#909) --- src/stream/gateway-link.test.ts | 2 +- src/stream/gateway-link.ts | 28 ++++++++++++++-- src/stream/mounts.ts | 30 ++++++++++++----- src/stream/service.ts | 57 +++++++++++++++++++++++++-------- src/stream/worker-link.ts | 30 ++++++++++++----- 5 files changed, 114 insertions(+), 33 deletions(-) diff --git a/src/stream/gateway-link.test.ts b/src/stream/gateway-link.test.ts index 9652572aa..a92d6f3e2 100644 --- a/src/stream/gateway-link.test.ts +++ b/src/stream/gateway-link.test.ts @@ -11,7 +11,7 @@ interface Relayed { readonly workerId: string; readonly url: string; readonly tenant: string; - readonly leaseId: string; + readonly leaseId: string | undefined; } class CodedError extends Error { diff --git a/src/stream/gateway-link.ts b/src/stream/gateway-link.ts index 827bdeab8..6d740b154 100644 --- a/src/stream/gateway-link.ts +++ b/src/stream/gateway-link.ts @@ -1,8 +1,9 @@ import type { ScreenMeta } from "../core/index.js"; +import type { StreamCaller } from "./caller.js"; import type { CapturedStream } from "./feed.js"; import { checkedMeta } from "./framing.js"; import { relayedRefusal, StreamRefusalError, streamUnavailable, unknownLease } from "./refusal.js"; -import type { StreamLink } from "./service.js"; +import type { StreamDeviceTarget, StreamLink } from "./service.js"; export interface StreamLeasePlacement { readonly ownerId: string; @@ -21,7 +22,7 @@ export interface FleetPort { call: { readonly request: Request; readonly tenant: string; - readonly leaseId: string; + readonly leaseId?: string; }, ): Promise; } @@ -34,7 +35,9 @@ const WORKER_STREAM_PATH = "/stream/leases/"; export function createGatewayStreamLink(options: GatewayStreamLinkOptions): StreamLink { return { - async serve(leaseId, caller, signal): Promise { + async serve(target, caller, signal): Promise { + if (typeof target !== "string") return serveDevice(options, target, caller, signal); + const leaseId = target; const held = options.fleet.holder(leaseId); if (held === undefined) throw unknownLease(leaseId); if (caller.role !== "operator" && held.ownerId !== caller.principal) { @@ -64,6 +67,25 @@ export function createGatewayStreamLink(options: GatewayStreamLinkOptions): Stre }; } +async function serveDevice( + options: GatewayStreamLinkOptions, + target: StreamDeviceTarget, + caller: StreamCaller, + signal: AbortSignal, +): Promise { + const workerId = target.workerId ?? ""; + const response = await relayed(workerId, target.deviceId, () => + options.fleet.relay(workerId, { + request: new Request( + `http://simlock-worker/stream/devices/${encodeURIComponent(target.deviceId)}`, + { method: "GET", signal }, + ), + tenant: `${options.fleet.requesterPrefix}${caller.principal}`, + }), + ); + return fromResponse(workerId, target.deviceId, response); +} + // why: the relay fails with the fleet's own codes -- a worker gone (`WORKER_UNREACHABLE`), a worker // that serves no stream (`RELAY_TARGET_MISSING`) -- and a viewer must get the picture's 503 with // that reason rather than a 500 that says nothing; anything without a code is a real fault. diff --git a/src/stream/mounts.ts b/src/stream/mounts.ts index 00e920cc6..3dcdc0860 100644 --- a/src/stream/mounts.ts +++ b/src/stream/mounts.ts @@ -1,4 +1,9 @@ -import { handleStreamRequest, leaseIdOf, type StreamServiceOptions } from "./service.js"; +import { + handleStreamRequest, + relayedDeviceTargetOf, + streamTargetOf, + type StreamServiceOptions, +} from "./service.js"; // why: the shape the HTTP layer mounts, declared here rather than imported from `src/http`, because a // stream is the same exchange whether a console on this host asks for it or a gateway relays one, and a @@ -24,7 +29,7 @@ export function createStreamMount(options: StreamServiceOptions): StreamMount { principal: identity.principal, role: identity.role === "operator" ? "operator" : "agent", }, - leaseIdOf(new URL(request.url).pathname), + streamTargetOf(new URL(request.url).pathname), ), }; } @@ -35,10 +40,19 @@ export function createStreamRelayTarget(options: StreamServiceOptions): StreamRe // worker checks its own registry against -- the same way `device.exec` names the requester a // gateway speaks for. Nothing on this door is a claim the worker takes on faith, and a relay is // admitted only from a gateway session, so it is as trusted as the uplink it came on and no more. - handleStreamRequest( - options, - request, - { principal: tenant, requesterId: tenant, role: "agent" }, - leaseId, - ); + leaseId === undefined + ? handleStreamRequest( + options, + request, + // why: a relay is admitted only from a gateway session, and the gateway has already refused a + // caller that is not an operator, so a device stream on this door is served whatever the tenant. + { principal: tenant, requesterId: tenant, role: "operator" }, + relayedDeviceTargetOf(new URL(request.url).pathname), + ) + : handleStreamRequest( + options, + request, + { principal: tenant, requesterId: tenant, role: "agent" }, + leaseId, + ); } diff --git a/src/stream/service.ts b/src/stream/service.ts index c35673f2d..cd4522b9f 100644 --- a/src/stream/service.ts +++ b/src/stream/service.ts @@ -2,8 +2,17 @@ import type { CapturedStream } from "./feed.js"; import type { StreamCaller } from "./caller.js"; import { noSuchStream, StreamRefusalError } from "./refusal.js"; +// why: a stream is asked for by the lease it films (a plain lease id, the route #910 removes) or by the +// device it films; `workerId` is absent on the relay door, where the gateway has already chosen the worker. +export interface StreamDeviceTarget { + readonly deviceId: string; + readonly workerId?: string; +} + +export type StreamTarget = string | StreamDeviceTarget; + export interface StreamLink { - serve(leaseId: string, caller: StreamCaller, signal: AbortSignal): Promise; + serve(target: StreamTarget, caller: StreamCaller, signal: AbortSignal): Promise; } export interface StreamServiceOptions { @@ -15,11 +24,27 @@ export interface StreamServiceOptions { } export function leaseIdOf(pathname: string): string | undefined { - const match = /^\/v1\/stream\/leases\/([^/]+)$/.exec(pathname); - const encoded = match?.[1]; - if (encoded === undefined) return undefined; + const encoded = /^\/v1\/stream\/leases\/([^/]+)$/.exec(pathname)?.[1]; + return encoded === undefined ? undefined : decoded(encoded); +} + +export function streamTargetOf(pathname: string): StreamTarget | undefined { + const device = /^\/v1\/stream\/workers\/([^/]+)\/devices\/([^/]+)$/.exec(pathname); + if (device?.[1] === undefined || device[2] === undefined) return leaseIdOf(pathname); + const workerId = decoded(device[1]); + const deviceId = decoded(device[2]); + return workerId === undefined || deviceId === undefined ? undefined : { deviceId, workerId }; +} + +export function relayedDeviceTargetOf(pathname: string): StreamTarget | undefined { + const encoded = /^\/stream\/devices\/([^/]+)$/.exec(pathname)?.[1]; + const deviceId = encoded === undefined ? undefined : decoded(encoded); + return deviceId === undefined ? undefined : { deviceId }; +} + +function decoded(segment: string): string | undefined { try { - return decodeURIComponent(encoded); + return decodeURIComponent(segment); } catch { return undefined; } @@ -29,9 +54,9 @@ export async function handleStreamRequest( options: StreamServiceOptions, request: Request, caller: StreamCaller, - leaseId: string | undefined, + target: StreamTarget | undefined, ): Promise { - if (leaseId === undefined) return refusalResponse(noSuchStream()); + if (target === undefined) return refusalResponse(noSuchStream()); if (request.method !== "GET") { // why: 405 rather than a 404 for a path that does exist -- a caller that posted to a stream is one // command away from the right ask, and "no such thing" would send it looking elsewhere. @@ -42,7 +67,7 @@ export async function handleStreamRequest( let stream: CapturedStream; try { - stream = await options.link.serve(leaseId, caller, request.signal); + stream = await options.link.serve(target, caller, request.signal); } catch (error) { const refusal = error instanceof StreamRefusalError @@ -53,8 +78,8 @@ export async function handleStreamRequest( error instanceof Error ? error.message : String(error), ); options.logger?.warn("Screen stream refused", { + ...targetFields(target), code: refusal.code, - leaseId, message: refusal.message, }); return refusalResponse(refusal); @@ -63,13 +88,13 @@ export async function handleStreamRequest( options.logger?.info("Screen stream opened", { codec: stream.meta.codec, height: stream.meta.height, - leaseId, + ...targetFields(target), width: stream.meta.width, }); - return streamResponse(leaseId, stream); + return streamResponse(typeof target === "string" ? target : undefined, stream); } -export function streamResponse(leaseId: string, stream: CapturedStream): Response { +export function streamResponse(leaseId: string | undefined, stream: CapturedStream): Response { const { meta } = stream; // why: the codec and size are headers as well as the body's first frame, because a browser must // decide how to read a body before it has read any of it, and because a gateway reads the headers @@ -82,12 +107,18 @@ export function streamResponse(leaseId: string, stream: CapturedStream): Respons "simlock-stream-fps": String(meta.fps), "simlock-stream-height": String(meta.height), "simlock-stream-width": String(meta.width), - "x-simlock-lease": leaseId, + ...(leaseId === undefined ? {} : { "x-simlock-lease": leaseId }), }, status: 200, }); } +function targetFields(target: StreamTarget): Record { + return typeof target === "string" + ? { leaseId: target } + : { deviceId: target.deviceId, workerId: target.workerId }; +} + // why: one body shape for every refusal this route can make -- `{ "error": { "code", "message" } }`, // the shape every other route answers with -- so a client that reads `error.code` needs no special // case for a stream that never started, and a gateway can forward a worker's code unchanged. diff --git a/src/stream/worker-link.ts b/src/stream/worker-link.ts index 83a4458d7..04cf02377 100644 --- a/src/stream/worker-link.ts +++ b/src/stream/worker-link.ts @@ -14,7 +14,7 @@ import type { CaptureRegistry } from "./captures.js"; import type { StreamCaller } from "./caller.js"; import { type CapturedStream, type FeedBounds, readScreenFeed } from "./feed.js"; import { notStreamable, StreamRefusalError, streamUnavailable, unknownLease } from "./refusal.js"; -import type { StreamLink } from "./service.js"; +import type { StreamLink, StreamTarget } from "./service.js"; export interface StreamRegistry { readonly snapshot: { @@ -42,12 +42,12 @@ export interface WorkerStreamLinkOptions { export function createWorkerStreamLink(options: WorkerStreamLinkOptions): StreamLink { return { - async serve(leaseId, caller, signal): Promise { + async serve(target, caller, signal): Promise { // why: ownership is proved before anything is asked of the machine. A link that looked for its // capture first would answer an agent one way for a lease this worker holds and another for a // lease that does not exist, which is a way of listing the fleet's leases. - const { device, lease } = requireLease(options.registry, leaseId, caller); - const slot = options.captures.begin(leaseId); + const { device, lease } = requireTarget(options.registry, target, caller); + const slot = options.captures.begin(lease?.id ?? device.id); // why: the listener, and not only the body's cancel, because a request aborted before its // first frame has no body being read yet -- which is what a page opened and closed quickly does. if (signal.aborted) void slot.release(); @@ -67,20 +67,22 @@ export function createWorkerStreamLink(options: WorkerStreamLinkOptions): Stream stream = await readScreenFeed(feed, options.bounds); } catch (error) { slot.abandon(); - if (slot.ended.aborted) throw unknownLease(leaseId); + if (slot.ended.aborted && lease !== undefined) throw unknownLease(lease.id); throw refusalFor(error); } // why: the lease is read again once there is a picture, because a capture can take seconds to // start and the lease it was asked for can end and the device be granted to someone else in // that time; a slot that was stopped says so even if a new lease reused the id. - const still = options.registry.snapshot.leases.find((candidate) => candidate.id === leaseId); + const still = options.registry.snapshot.leases.find( + (candidate) => candidate.id === lease?.id, + ); if (signal.aborted) { await slot.release(); throw streamUnavailable("The viewer went away before the picture started"); } - if (slot.ended.aborted || still?.deviceId !== lease.deviceId) { + if (lease !== undefined && (slot.ended.aborted || still?.deviceId !== lease.deviceId)) { await slot.release(); - throw unknownLease(leaseId); + throw unknownLease(lease.id); } return { ...stream, close: () => slot.release() }; }, @@ -105,6 +107,18 @@ function refusalFor(error: unknown): StreamRefusalError { return streamUnavailable(String(error)); } +function requireTarget( + registry: StreamRegistry, + target: StreamTarget, + caller: StreamCaller, +): { readonly lease?: LeaseRecord; readonly device: DeviceRecord } { + if (typeof target === "string") return requireLease(registry, target, caller); + const device = registry.snapshot.devices.find((candidate) => candidate.id === target.deviceId); + if (device === undefined) + throw new StreamRefusalError(404, "UNKNOWN_DEVICE", `Unknown device: ${target.deviceId}`); + return { device }; +} + // why: ownership is proved against this worker's own registry rather than taken from whoever relayed // the request. A caller that names a requester is checked against the lease's requester (the rule // `device.exec` follows); one that names none is checked against the principal that owns the lease. From 22ee23c02c69a95f0d8975e9bc1c9d84f832c99f Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:23:42 +0200 Subject: [PATCH 03/20] test: an operator asks a single worker for a device's stream (#909) --- src/daemon/main.ts | 1 + src/modules/testing.ts | 1 + src/modules/types.ts | 1 + src/stream/create.test.ts | 92 +++++++++++++++++++++++++++++++++++-- src/stream/create.ts | 1 + src/stream/test-fixtures.ts | 2 + src/stream/worker.ts | 1 + 7 files changed, 95 insertions(+), 4 deletions(-) diff --git a/src/daemon/main.ts b/src/daemon/main.ts index deac8aa0c..ec91fbf7e 100644 --- a/src/daemon/main.ts +++ b/src/daemon/main.ts @@ -505,6 +505,7 @@ async function composeDaemon( environment: process.env, events: { subscribe: (event, handler) => eventBus.subscribe(event, handler) }, filesystem, + instanceId, leaseEnvironment, nodePath: process.execPath, nodeVersion: process.versions.node, diff --git a/src/modules/testing.ts b/src/modules/testing.ts index 7b4d0a8ac..17974619b 100644 --- a/src/modules/testing.ts +++ b/src/modules/testing.ts @@ -148,6 +148,7 @@ export async function fakeWorkerContext(name: string, config: C): Promise bus.subscribe(event, handler) }, filesystem: new MemoryFilesystem(), + instanceId: "fake-worker", leaseEnvironment: {}, logger: new NoopLogger(), name, diff --git a/src/modules/types.ts b/src/modules/types.ts index ecc57fae6..77e950807 100644 --- a/src/modules/types.ts +++ b/src/modules/types.ts @@ -130,6 +130,7 @@ export interface ModuleScreens { export interface WorkerContext { readonly name: string; + readonly instanceId: string; readonly config: C; readonly daemonConfig: Config; readonly logger: Logger; diff --git a/src/stream/create.test.ts b/src/stream/create.test.ts index 775de6db9..b94144a66 100644 --- a/src/stream/create.test.ts +++ b/src/stream/create.test.ts @@ -1,24 +1,41 @@ import { describe, expect, it } from "vitest"; -import type { ScreenAvailability } from "../core/index.js"; +import { type ScreenAvailability, ScreenNotSupportedError } from "../core/index.js"; import { FakeClock, NoopLogger } from "../ports/index.js"; import { createWorkerStream } from "./create.js"; import { ScriptedFeed, TestRegistry } from "./test-fixtures.js"; -function worker(availability: readonly ScreenAvailability[] = [], sendsMeta = true) { +function worker( + availability: readonly ScreenAvailability[] = [], + sendsMeta = true, + unsupported = false, +) { const clock = new FakeClock(); const registry = new TestRegistry( [{ deviceId: "dev_1", id: "lse_1", ownerId: "tok_agent", requesterId: "gw:inst:agent-1" }], - [{ id: "dev_1", platform: "ios" }], + [ + { id: "dev_1", platform: "ios" }, + { id: "dev_ready", platform: "ios", state: "ready" }, + { id: "dev_leased", platform: "ios", state: "leased" }, + { id: "dev_shutdown", platform: "ios", state: "shutdown" }, + { id: "dev_provisioning", platform: "ios", state: "provisioning" }, + { id: "dev_reclaiming", platform: "ios", state: "reclaiming" }, + { id: "dev_quarantined", platform: "ios", state: "quarantined" }, + { id: "dev_absent", physical: true, platform: "ios", state: "absent" }, + ], ); const feeds: ScriptedFeed[] = []; + let opened = 0; const stream = createWorkerStream({ clock, + instanceId: "w1", logger: new NoopLogger(), name: "stream", registry, screens: { openScreen: async () => { + opened += 1; + if (unsupported) throw new ScreenNotSupportedError("no capture here"); const feed = new ScriptedFeed(); if (sendsMeta) feed.meta(); feeds.push(feed); @@ -28,9 +45,14 @@ function worker(availability: readonly ScreenAvailability[] = [], sendsMeta = tr }, settings: { bitrateKbps: 1500, fps: 30, maxDimension: 720 }, }); - return { clock, feeds, stream }; + return { clock, feeds, opened: () => opened, stream }; } +const OPERATOR = { principal: "tok_op", role: "operator" }; + +const deviceRequest = (worker: string, device: string) => + new Request(`http://daemon/v1/stream/workers/${worker}/devices/${device}`); + const relayed = () => new Request("http://simlock-worker/stream/leases/lse_1"); async function settledWithin(promise: Promise): Promise { @@ -121,4 +143,66 @@ describe("createWorkerStream", () => { ); expect(feeds[0]?.closes).toBe(1); }); + + it("serves an operator the stream of a ready or leased device, and refuses any other device with its reason before opening anything", async () => { + for (const device of ["dev_ready", "dev_leased"]) { + const { opened, stream } = worker(); + const served = await stream.mount.handle(deviceRequest("w1", device), OPERATOR); + expect(served.status, device).toBe(200); + expect(opened(), device).toBe(1); + await served.body?.cancel(); + } + const refusals: [string, string, number, string, string][] = [ + [ + "w1", + "dev_shutdown", + 409, + "DEVICE_NOT_RUNNING", + "The device is shutdown; only a ready or leased device streams", + ], + [ + "w1", + "dev_provisioning", + 409, + "DEVICE_NOT_RUNNING", + "The device is provisioning; only a ready or leased device streams", + ], + [ + "w1", + "dev_reclaiming", + 409, + "DEVICE_NOT_RUNNING", + "The device is reclaiming; only a ready or leased device streams", + ], + [ + "w1", + "dev_quarantined", + 409, + "DEVICE_NOT_RUNNING", + "The device is quarantined; only a ready or leased device streams", + ], + [ + "w1", + "dev_absent", + 409, + "DEVICE_NOT_RUNNING", + "The device is absent; only a ready or leased device streams", + ], + ["w1", "dev_nope", 404, "UNKNOWN_DEVICE", "Unknown device: dev_nope"], + ["w2", "dev_ready", 404, "UNKNOWN_WORKER", "Unknown worker: w2"], + ]; + for (const [workerId, device, status, code, message] of refusals) { + const { opened, stream } = worker(); + const refused = await stream.mount.handle(deviceRequest(workerId, device), OPERATOR); + expect(refused.status, device).toBe(status); + expect(await refused.json(), device).toEqual({ error: { code, message } }); + expect(opened(), device).toBe(0); + } + const { stream } = worker([], true, true); + const unsupported = await stream.mount.handle(deviceRequest("w1", "dev_ready"), OPERATOR); + expect(unsupported.status).toBe(415); + expect(((await unsupported.json()) as { error: { code: string } }).error.code).toBe( + "STREAM_NOT_SUPPORTED", + ); + }); }); diff --git a/src/stream/create.ts b/src/stream/create.ts index 7e47a8f81..94d25a53e 100644 --- a/src/stream/create.ts +++ b/src/stream/create.ts @@ -15,6 +15,7 @@ import { createWorkerStreamLink, type ScreenOpener, type StreamRegistry } from " export interface WorkerStreamOptions { readonly name: string; + readonly instanceId: string; readonly registry: StreamRegistry; readonly screens: ScreenOpener & { screenAvailability(): Promise; diff --git a/src/stream/test-fixtures.ts b/src/stream/test-fixtures.ts index 35d694e1c..bee0cc8cd 100644 --- a/src/stream/test-fixtures.ts +++ b/src/stream/test-fixtures.ts @@ -20,6 +20,7 @@ export interface TestDevice { readonly id: string; readonly platform: Platform; readonly physical?: boolean; + readonly state?: string; } export class TestRegistry implements StreamRegistry { @@ -42,6 +43,7 @@ export class TestRegistry implements StreamRegistry { enrolledApps: [], id: device.id, physical: device.physical === true, + state: device.state ?? "ready", spec: { platform: device.platform }, }) as unknown as DeviceRecord, ), diff --git a/src/stream/worker.ts b/src/stream/worker.ts index 38950216f..b0f8dbcc5 100644 --- a/src/stream/worker.ts +++ b/src/stream/worker.ts @@ -6,6 +6,7 @@ import type { StreamModuleConfig } from "./definition.js"; export function createWorker(ctx: WorkerContext): WorkerModule { const stream = createWorkerStream({ clock: ctx.clock, + instanceId: ctx.instanceId, logger: ctx.logger, name: ctx.name, registry: ctx.registry, From 9690e4dadc7d3f4dda71820f91f491b2ed4732ac Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:24:47 +0200 Subject: [PATCH 04/20] feat(stream): a worker streams a ready or leased device and refuses any other with its reason (#909) --- src/stream/create.ts | 1 + src/stream/refusal.ts | 20 ++++++++++++++++++-- src/stream/worker-link.test.ts | 4 ++++ src/stream/worker-link.ts | 30 +++++++++++++++++++++++------- 4 files changed, 46 insertions(+), 9 deletions(-) diff --git a/src/stream/create.ts b/src/stream/create.ts index 94d25a53e..b3c7992e9 100644 --- a/src/stream/create.ts +++ b/src/stream/create.ts @@ -46,6 +46,7 @@ export function createWorkerStream(options: WorkerStreamOptions): WorkerStream { link: createWorkerStreamLink({ bounds: { clock: options.clock, silenceMs: DEFAULT_SILENCE_MS, startMs: DEFAULT_START_MS }, captures, + instanceId: options.instanceId, registry: options.registry, screens: options.screens, settings: options.settings, diff --git a/src/stream/refusal.ts b/src/stream/refusal.ts index 6f0f44fc5..8775957a9 100644 --- a/src/stream/refusal.ts +++ b/src/stream/refusal.ts @@ -5,8 +5,8 @@ export class StreamRefusalError extends Error { // why: `code` is a free string rather than a closed union, because a gateway forwards the code its // worker answered with and that worker may be a build this one has never seen. The codes this - // service raises itself are `UNKNOWN_LEASE`, `BAD_REQUEST`, `STREAM_NOT_SUPPORTED`, - // `STREAM_UNAVAILABLE` and `INTERNAL`, and the HTTP API promises exactly these. + // service raises itself are `UNKNOWN_LEASE`, `UNKNOWN_DEVICE`, `UNKNOWN_WORKER`, `DEVICE_NOT_RUNNING`, + // `BAD_REQUEST`, `STREAM_NOT_SUPPORTED`, `STREAM_UNAVAILABLE` and `INTERNAL`, and the HTTP API promises exactly these. constructor( readonly status: number, readonly code: string, @@ -21,6 +21,22 @@ export function unknownLease(leaseId: string): StreamRefusalError { return new StreamRefusalError(404, "UNKNOWN_LEASE", `Unknown lease: ${leaseId}`); } +export function unknownDevice(deviceId: string): StreamRefusalError { + return new StreamRefusalError(404, "UNKNOWN_DEVICE", `Unknown device: ${deviceId}`); +} + +export function unknownWorker(workerId: string): StreamRefusalError { + return new StreamRefusalError(404, "UNKNOWN_WORKER", `Unknown worker: ${workerId}`); +} + +export function deviceNotRunning(state: string): StreamRefusalError { + return new StreamRefusalError( + 409, + "DEVICE_NOT_RUNNING", + `The device is ${state}; only a ready or leased device streams`, + ); +} + export function noSuchStream(): StreamRefusalError { // why: the same status and code as an unknown lease on purpose -- a daemon with `capabilities.stream.enabled` off // never mounts this route at all and answers any path it has not heard of as a 404, so a client that diff --git a/src/stream/worker-link.test.ts b/src/stream/worker-link.test.ts index 4e2bf4a72..82ae32fef 100644 --- a/src/stream/worker-link.test.ts +++ b/src/stream/worker-link.test.ts @@ -35,6 +35,7 @@ function setup(options: { refuseWith?: Error } = {}) { const captures = createCaptureRegistry(); const opened: Opened[] = []; const link = createWorkerStreamLink({ + instanceId: "w1", bounds: { clock, silenceMs: 1_000, startMs: 5_000 }, captures, registry, @@ -174,6 +175,7 @@ describe("the worker's stream link", () => { ); const captures = createCaptureRegistry(); const link = createWorkerStreamLink({ + instanceId: "w1", bounds: { clock: new FakeClock(), silenceMs: 1_000, startMs: 5_000 }, captures, registry, @@ -218,6 +220,7 @@ describe("the worker's stream link", () => { ); const captures = createCaptureRegistry(); const link = createWorkerStreamLink({ + instanceId: "w1", bounds: { clock: new FakeClock(), silenceMs: 1_000, startMs: 5_000 }, captures, registry, @@ -251,6 +254,7 @@ describe("the worker's stream link", () => { ); const captures = createCaptureRegistry(); const link = createWorkerStreamLink({ + instanceId: "w1", bounds: { clock: new FakeClock(), silenceMs: 1_000, startMs: 5_000 }, captures, registry, diff --git a/src/stream/worker-link.ts b/src/stream/worker-link.ts index 04cf02377..892ae6c89 100644 --- a/src/stream/worker-link.ts +++ b/src/stream/worker-link.ts @@ -13,7 +13,15 @@ import { import type { CaptureRegistry } from "./captures.js"; import type { StreamCaller } from "./caller.js"; import { type CapturedStream, type FeedBounds, readScreenFeed } from "./feed.js"; -import { notStreamable, StreamRefusalError, streamUnavailable, unknownLease } from "./refusal.js"; +import { + deviceNotRunning, + notStreamable, + StreamRefusalError, + streamUnavailable, + unknownDevice, + unknownLease, + unknownWorker, +} from "./refusal.js"; import type { StreamLink, StreamTarget } from "./service.js"; export interface StreamRegistry { @@ -33,6 +41,7 @@ export interface ScreenOpener { } export interface WorkerStreamLinkOptions { + readonly instanceId: string; readonly registry: StreamRegistry; readonly screens: ScreenOpener; readonly captures: CaptureRegistry; @@ -46,7 +55,7 @@ export function createWorkerStreamLink(options: WorkerStreamLinkOptions): Stream // why: ownership is proved before anything is asked of the machine. A link that looked for its // capture first would answer an agent one way for a lease this worker holds and another for a // lease that does not exist, which is a way of listing the fleet's leases. - const { device, lease } = requireTarget(options.registry, target, caller); + const { device, lease } = requireTarget(options, target, caller); const slot = options.captures.begin(lease?.id ?? device.id); // why: the listener, and not only the body's cancel, because a request aborted before its // first frame has no body being read yet -- which is what a page opened and closed quickly does. @@ -108,14 +117,21 @@ function refusalFor(error: unknown): StreamRefusalError { } function requireTarget( - registry: StreamRegistry, + options: WorkerStreamLinkOptions, target: StreamTarget, caller: StreamCaller, ): { readonly lease?: LeaseRecord; readonly device: DeviceRecord } { - if (typeof target === "string") return requireLease(registry, target, caller); - const device = registry.snapshot.devices.find((candidate) => candidate.id === target.deviceId); - if (device === undefined) - throw new StreamRefusalError(404, "UNKNOWN_DEVICE", `Unknown device: ${target.deviceId}`); + if (typeof target === "string") return requireLease(options.registry, target, caller); + // why: a worker answers only for its own id (ADR 0012 section 1); a gateway names the worker it relays to, + // and a request that reaches the wrong worker is refused rather than answered with the wrong screen. + if (target.workerId !== undefined && target.workerId !== options.instanceId) { + throw unknownWorker(target.workerId); + } + const device = options.registry.snapshot.devices.find( + (candidate) => candidate.id === target.deviceId, + ); + if (device === undefined) throw unknownDevice(target.deviceId); + if (device.state !== "ready" && device.state !== "leased") throw deviceNotRunning(device.state); return { device }; } From 88d284fc60d68ebc07cd00dfd15ef0dcde1ddd0f Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:25:14 +0200 Subject: [PATCH 05/20] test: an agent token asks for a device's stream (#909) --- src/stream/create.test.ts | 35 ++++++++++++++++++++++++++++++++++- 1 file changed, 34 insertions(+), 1 deletion(-) diff --git a/src/stream/create.test.ts b/src/stream/create.test.ts index b94144a66..1600666c4 100644 --- a/src/stream/create.test.ts +++ b/src/stream/create.test.ts @@ -2,7 +2,7 @@ import { describe, expect, it } from "vitest"; import { type ScreenAvailability, ScreenNotSupportedError } from "../core/index.js"; import { FakeClock, NoopLogger } from "../ports/index.js"; -import { createWorkerStream } from "./create.js"; +import { createGatewayStream, createWorkerStream } from "./create.js"; import { ScriptedFeed, TestRegistry } from "./test-fixtures.js"; function worker( @@ -205,4 +205,37 @@ describe("createWorkerStream", () => { "STREAM_NOT_SUPPORTED", ); }); + + it("answers an agent token asking for a device's stream with 404 UNKNOWN_DEVICE, on a worker and through a gateway, opening and relaying nothing", async () => { + const agent = { principal: "tok_agent", role: "agent" }; + const { opened, stream } = worker(); + expect((await stream.mount.handle(deviceRequest("w1", "dev_1"), OPERATOR)).status).toBe(200); + const onWorker = await stream.mount.handle(deviceRequest("w1", "dev_1"), agent); + expect(onWorker.status).toBe(404); + expect(await onWorker.json()).toEqual({ + error: { code: "UNKNOWN_DEVICE", message: "Unknown device: dev_1" }, + }); + expect(opened()).toBe(1); + + const relays: string[] = []; + const gateway = createGatewayStream({ + fleet: { + holder: () => undefined, + relay: async (workerId) => { + relays.push(workerId); + return new Response(null, { status: 500 }); + }, + requesterPrefix: "gw:inst:", + }, + logger: new NoopLogger(), + }); + const throughGateway = await gateway.mount.handle(deviceRequest("w1", "dev_1"), agent); + expect(throughGateway.status).toBe(404); + expect(await throughGateway.json()).toEqual({ + error: { code: "UNKNOWN_DEVICE", message: "Unknown device: dev_1" }, + }); + expect(relays).toEqual([]); + await gateway.mount.handle(deviceRequest("w1", "dev_1"), OPERATOR); + expect(relays).toEqual(["w1"]); + }); }); From 766264be9cbd89da1be55b61c4e546dacf566166 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:25:32 +0200 Subject: [PATCH 06/20] feat(stream): only an operator streams a device (#909) --- src/stream/gateway-link.ts | 11 ++++++++++- src/stream/worker-link.ts | 3 +++ 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/src/stream/gateway-link.ts b/src/stream/gateway-link.ts index 6d740b154..9b86a1600 100644 --- a/src/stream/gateway-link.ts +++ b/src/stream/gateway-link.ts @@ -2,7 +2,13 @@ import type { ScreenMeta } from "../core/index.js"; import type { StreamCaller } from "./caller.js"; import type { CapturedStream } from "./feed.js"; import { checkedMeta } from "./framing.js"; -import { relayedRefusal, StreamRefusalError, streamUnavailable, unknownLease } from "./refusal.js"; +import { + relayedRefusal, + StreamRefusalError, + streamUnavailable, + unknownDevice, + unknownLease, +} from "./refusal.js"; import type { StreamDeviceTarget, StreamLink } from "./service.js"; export interface StreamLeasePlacement { @@ -73,6 +79,9 @@ async function serveDevice( caller: StreamCaller, signal: AbortSignal, ): Promise { + // why: the same refusal an unknown device gets, before any relay, so an agent cannot reach a screen + // by naming a device. + if (caller.role !== "operator") throw unknownDevice(target.deviceId); const workerId = target.workerId ?? ""; const response = await relayed(workerId, target.deviceId, () => options.fleet.relay(workerId, { diff --git a/src/stream/worker-link.ts b/src/stream/worker-link.ts index 892ae6c89..20bd6430f 100644 --- a/src/stream/worker-link.ts +++ b/src/stream/worker-link.ts @@ -122,6 +122,9 @@ function requireTarget( caller: StreamCaller, ): { readonly lease?: LeaseRecord; readonly device: DeviceRecord } { if (typeof target === "string") return requireLease(options.registry, target, caller); + // why: only an operator watches a device, and an agent gets the answer an unknown device gets, so + // it cannot learn which devices exist or which one its lease holds. + if (caller.role !== "operator") throw unknownDevice(target.deviceId); // why: a worker answers only for its own id (ADR 0012 section 1); a gateway names the worker it relays to, // and a request that reaches the wrong worker is refused rather than answered with the wrong screen. if (target.workerId !== undefined && target.workerId !== options.instanceId) { From 4d92124d638f9a6e0831a7d8864131d20fa5a0b6 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:26:20 +0200 Subject: [PATCH 07/20] test: an operator asks a gateway for a device's stream (#909) --- src/stream/create.test.ts | 73 +++++++++++++++++++++++++++++++++ src/stream/gateway-link.test.ts | 1 + src/stream/gateway-link.ts | 1 + src/stream/gateway.ts | 1 + 4 files changed, 76 insertions(+) diff --git a/src/stream/create.test.ts b/src/stream/create.test.ts index 1600666c4..c3cf13624 100644 --- a/src/stream/create.test.ts +++ b/src/stream/create.test.ts @@ -48,6 +48,13 @@ function worker( return { clock, feeds, opened: () => opened, stream }; } +const STREAM_HEADERS = { + "simlock-stream-codec": "h264", + "simlock-stream-fps": "30", + "simlock-stream-height": "720", + "simlock-stream-width": "330", +}; + const OPERATOR = { principal: "tok_op", role: "operator" }; const deviceRequest = (worker: string, device: string) => @@ -221,6 +228,7 @@ describe("createWorkerStream", () => { const gateway = createGatewayStream({ fleet: { holder: () => undefined, + knowsWorker: () => true, relay: async (workerId) => { relays.push(workerId); return new Response(null, { status: 500 }); @@ -238,4 +246,69 @@ describe("createWorkerStream", () => { await gateway.mount.handle(deviceRequest("w1", "dev_1"), OPERATOR); expect(relays).toEqual(["w1"]); }); + + it("relays an operator's device stream through a gateway, keeping the worker's refusal code, and refuses an unknown or disconnected worker", async () => { + const refusal = (status: number, code: string, message: string) => + new Response(JSON.stringify({ error: { code, message } }), { status }); + const answers: Record Response | Promise> = { + w1: () => new Response(new ReadableStream(), { headers: STREAM_HEADERS }), + w_busy: () => + refusal( + 409, + "DEVICE_NOT_RUNNING", + "The device is shutdown; only a ready or leased device streams", + ), + w_missing: () => refusal(404, "UNKNOWN_DEVICE", "Unknown device: dev_9"), + w_gone: () => { + throw Object.assign(new Error("Worker w_gone is not connected"), { + code: "WORKER_UNREACHABLE", + }); + }, + }; + const relays: { workerId: string; url: string; tenant: string }[] = []; + const gateway = createGatewayStream({ + fleet: { + holder: () => undefined, + knowsWorker: (workerId) => workerId in answers, + relay: async (workerId, call) => { + relays.push({ tenant: call.tenant, url: call.request.url, workerId }); + return (answers[workerId] ?? (() => new Response(null, { status: 500 })))(); + }, + requesterPrefix: "gw:inst:", + }, + logger: new NoopLogger(), + }); + const ask = (workerId: string) => + gateway.mount.handle(deviceRequest(workerId, "dev_9"), OPERATOR); + + const served = await ask("w1"); + expect(served.status).toBe(200); + expect(served.headers.get("simlock-stream-codec")).toBe("h264"); + await served.body?.cancel(); + expect(relays).toEqual([ + { + tenant: "gw:inst:tok_op", + url: "http://simlock-worker/stream/devices/dev_9", + workerId: "w1", + }, + ]); + + const expected: [string, number, string, string][] = [ + [ + "w_busy", + 409, + "DEVICE_NOT_RUNNING", + "The device is shutdown; only a ready or leased device streams", + ], + ["w_missing", 404, "UNKNOWN_DEVICE", "Unknown device: dev_9"], + ["w_gone", 503, "WORKER_UNREACHABLE", "Worker w_gone is not connected"], + ["w_nobody", 404, "UNKNOWN_WORKER", "Unknown worker: w_nobody"], + ]; + for (const [workerId, status, code, message] of expected) { + const refused = await ask(workerId); + expect(refused.status, workerId).toBe(status); + expect(await refused.json(), workerId).toEqual({ error: { code, message } }); + } + expect(relays.map((relay) => relay.workerId)).toEqual(["w1", "w_busy", "w_missing", "w_gone"]); + }); }); diff --git a/src/stream/gateway-link.test.ts b/src/stream/gateway-link.test.ts index a92d6f3e2..fab8ed699 100644 --- a/src/stream/gateway-link.test.ts +++ b/src/stream/gateway-link.test.ts @@ -31,6 +31,7 @@ function fleet(answer: () => Response): { port: FleetPort; relayed: Relayed[] } leaseId === "w1.lse_9" ? { ownerId: "tok_agent", requesterId: "agent-1", workerId: "w1", workerLeaseId: "lse_9" } : undefined, + knowsWorker: () => true, relay: async (workerId, call) => { relayed.push({ leaseId: call.leaseId, diff --git a/src/stream/gateway-link.ts b/src/stream/gateway-link.ts index 9b86a1600..57e8ed36e 100644 --- a/src/stream/gateway-link.ts +++ b/src/stream/gateway-link.ts @@ -20,6 +20,7 @@ export interface StreamLeasePlacement { export interface FleetPort { holder(leaseId: string): StreamLeasePlacement | undefined; + knowsWorker(workerId: string): boolean; // why: `gw::`, the exact prefix this index recognizes a requester it // sent by, so a stream namespaces with the same value its leases do. readonly requesterPrefix: string; diff --git a/src/stream/gateway.ts b/src/stream/gateway.ts index 5b5d6c225..0953a15ee 100644 --- a/src/stream/gateway.ts +++ b/src/stream/gateway.ts @@ -17,6 +17,7 @@ export function createGateway(ctx: GatewayContext): GatewayM workerLeaseId: entry.workerLeaseId, }; }, + knowsWorker: (workerId) => ctx.worker(workerId) !== undefined, relay: (workerId, call) => ctx.relay(workerId, call.request, call.tenant, call.leaseId), requesterPrefix: ctx.requesterPrefix, }, From eacd6e23c56556a7f4e6cddda793e0bc985727ec Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:27:13 +0200 Subject: [PATCH 08/20] feat(stream): a gateway refuses an unknown worker and keeps a worker's refusal for a device stream (#909) --- src/stream/gateway-link.ts | 20 +++++++++++--------- src/stream/refusal.ts | 4 ++-- 2 files changed, 13 insertions(+), 11 deletions(-) diff --git a/src/stream/gateway-link.ts b/src/stream/gateway-link.ts index 57e8ed36e..69bac41c3 100644 --- a/src/stream/gateway-link.ts +++ b/src/stream/gateway-link.ts @@ -8,6 +8,7 @@ import { streamUnavailable, unknownDevice, unknownLease, + unknownWorker, } from "./refusal.js"; import type { StreamDeviceTarget, StreamLink } from "./service.js"; @@ -56,7 +57,7 @@ export function createGatewayStreamLink(options: GatewayStreamLinkOptions): Stre // a caller cannot reach a screen by naming someone else, and the worker proves ownership // against its own registry instead of trusting this relay. const namespacedRequesterId = `${options.fleet.requesterPrefix}${held.requesterId}`; - const response = await relayed(held.workerId, leaseId, () => + const response = await relayed(held.workerId, `lease ${leaseId}`, () => options.fleet.relay(held.workerId, { leaseId: held.workerLeaseId, request: new Request( @@ -69,7 +70,7 @@ export function createGatewayStreamLink(options: GatewayStreamLinkOptions): Stre tenant: namespacedRequesterId, }), ); - return fromResponse(held.workerId, leaseId, response); + return fromResponse(held.workerId, `lease ${leaseId}`, response); }, }; } @@ -84,7 +85,8 @@ async function serveDevice( // by naming a device. if (caller.role !== "operator") throw unknownDevice(target.deviceId); const workerId = target.workerId ?? ""; - const response = await relayed(workerId, target.deviceId, () => + if (!options.fleet.knowsWorker(workerId)) throw unknownWorker(workerId); + const response = await relayed(workerId, `device ${target.deviceId}`, () => options.fleet.relay(workerId, { request: new Request( `http://simlock-worker/stream/devices/${encodeURIComponent(target.deviceId)}`, @@ -93,7 +95,7 @@ async function serveDevice( tenant: `${options.fleet.requesterPrefix}${caller.principal}`, }), ); - return fromResponse(workerId, target.deviceId, response); + return fromResponse(workerId, `device ${target.deviceId}`, response); } // why: the relay fails with the fleet's own codes -- a worker gone (`WORKER_UNREACHABLE`), a worker @@ -101,7 +103,7 @@ async function serveDevice( // that reason rather than a 500 that says nothing; anything without a code is a real fault. async function relayed( workerId: string, - leaseId: string, + subject: string, relay: () => Promise, ): Promise { try { @@ -110,7 +112,7 @@ async function relayed( const code = (error as { readonly code?: unknown }).code; if (code === "RELAY_TARGET_MISSING") { throw streamUnavailable( - `Worker ${workerId} does not serve screen streams for lease ${leaseId}; turn on capabilities.stream.enabled there`, + `Worker ${workerId} does not serve screen streams for ${subject}; turn on capabilities.stream.enabled there`, ); } if (typeof code === "string" && /^[A-Z][A-Z0-9_]{0,63}$/.test(code)) { @@ -130,7 +132,7 @@ const MAX_REFUSAL_BYTES = 16 * 1024; async function fromResponse( workerId: string, - leaseId: string, + subject: string, response: Response, ): Promise { // why: the codec header is what tells a relayed stream from a relayed refusal -- a gateway reads a @@ -139,7 +141,7 @@ async function fromResponse( const codec = response.headers.get("simlock-stream-codec"); if (!response.ok || codec === null || response.body === null) { const reason = await readCapped(response, MAX_REFUSAL_BYTES); - throw relayedRefusal(workerId, leaseId, response.status, reason); + throw relayedRefusal(workerId, subject, response.status, reason); } let meta: ScreenMeta; try { @@ -152,7 +154,7 @@ async function fromResponse( } catch (error) { await response.body.cancel().catch(() => undefined); throw streamUnavailable( - `Worker ${workerId} answered the stream for lease ${leaseId} with a picture it cannot describe: ${ + `Worker ${workerId} answered the stream for ${subject} with a picture it cannot describe: ${ error instanceof Error ? error.message : String(error) }`, ); diff --git a/src/stream/refusal.ts b/src/stream/refusal.ts index 8775957a9..728b49131 100644 --- a/src/stream/refusal.ts +++ b/src/stream/refusal.ts @@ -59,7 +59,7 @@ const MAX_RELAYED_MESSAGE = 1_000; export function relayedRefusal( workerId: string, - leaseId: string, + subject: string, status: number, body: string, ): StreamRefusalError { @@ -80,7 +80,7 @@ export function relayedRefusal( return new StreamRefusalError( relayedStatus, "STREAM_UNAVAILABLE", - `Worker ${workerId} refused the stream for lease ${leaseId} with status ${String(status)}`, + `Worker ${workerId} refused the stream for ${subject} with status ${String(status)}`, ); } From 891fd0576bfa542d933f2c8d3dd517e45418d920 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:28:20 +0200 Subject: [PATCH 09/20] test: a watched device changes: leased, lease ended, hook while opening (#909) --- src/stream/create.test.ts | 54 ++++++++++++++++++++++++++++++++++++++- 1 file changed, 53 insertions(+), 1 deletion(-) diff --git a/src/stream/create.test.ts b/src/stream/create.test.ts index c3cf13624..f9f74a8c0 100644 --- a/src/stream/create.test.ts +++ b/src/stream/create.test.ts @@ -9,6 +9,7 @@ function worker( availability: readonly ScreenAvailability[] = [], sendsMeta = true, unsupported = false, + openGate: Promise = Promise.resolve(), ) { const clock = new FakeClock(); const registry = new TestRegistry( @@ -35,6 +36,7 @@ function worker( screens: { openScreen: async () => { opened += 1; + await openGate; if (unsupported) throw new ScreenNotSupportedError("no capture here"); const feed = new ScriptedFeed(); if (sendsMeta) feed.meta(); @@ -45,7 +47,7 @@ function worker( }, settings: { bitrateKbps: 1500, fps: 30, maxDimension: 720 }, }); - return { clock, feeds, opened: () => opened, stream }; + return { clock, feeds, opened: () => opened, registry, stream }; } const STREAM_HEADERS = { @@ -311,4 +313,54 @@ describe("createWorkerStream", () => { } expect(relays.map((relay) => relay.workerId)).toEqual(["w1", "w_busy", "w_missing", "w_gone"]); }); + + it("follows its device: a lease granted leaves the stream running, a lease ended stops the capture before the reclaim hook returns, and a hook that runs while the capture opens refuses with 409", async () => { + { + const { feeds, registry, stream } = worker(); + const ended = (leaseId: string) => + ({ deviceId: "dev_ready", leaseId }) as Parameters[0]; + const served = await stream.mount.handle(deviceRequest("w1", "dev_ready"), OPERATOR); + const reader = served.body!.getReader(); + await reader.read(); + + registry.devices = registry.devices.map((device) => + device.id === "dev_ready" ? { ...device, state: "leased" } : device, + ); + registry.leases.push({ + deviceId: "dev_ready", + id: "lse_2", + ownerId: "tok_agent", + requesterId: "agent-2", + }); + feeds[0]?.keyframe([9]); + expect((await reader.read()).done).toBe(false); + expect(feeds[0]?.closes).toBe(0); + + await stream.leaseEnd.leaseEnded(ended("lse_2"), new AbortController().signal); + expect(feeds[0]?.closes).toBe(1); + expect((await reader.read()).done).toBe(true); + } + + { + let openIt = () => {}; + const gate = new Promise((resolve) => (openIt = resolve)); + const { feeds, stream } = worker([], true, false, gate); + const asked = stream.mount.handle(deviceRequest("w1", "dev_ready"), OPERATOR); + await settledWithin(asked); + const ending = stream.leaseEnd.leaseEnded( + { deviceId: "dev_ready", leaseId: "lse_3" } as Parameters< + typeof stream.leaseEnd.leaseEnded + >[0], + new AbortController().signal, + ); + openIt(); + const refused = await asked; + await ending; + expect(refused.status).toBe(409); + expect(((await refused.json()) as { error: { code: string } }).error.code).toBe( + "DEVICE_NOT_RUNNING", + ); + expect(feeds[0]?.closes).toBe(1); + } + }); }); From f4c04a58e95fd71506881c1995b622ce2232eb34 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:29:00 +0200 Subject: [PATCH 10/20] feat(stream): a lease ending stops the captures on its device and a hook mid-open refuses 409 (#909) --- src/stream/captures.ts | 37 ++++++++++++++++++++++++++++++------- src/stream/refusal.ts | 8 ++++++++ src/stream/worker-link.ts | 17 +++++++++++++---- 3 files changed, 51 insertions(+), 11 deletions(-) diff --git a/src/stream/captures.ts b/src/stream/captures.ts index ddeeab3e0..5ab17cd8e 100644 --- a/src/stream/captures.ts +++ b/src/stream/captures.ts @@ -1,7 +1,8 @@ -// why: a capture belongs to a lease, not to a connection -- the lease is the thing that says this -// machine still owns that device. A viewer going away stops a capture through its own request; this -// covers what that cannot, a lease that ends somewhere else (released elsewhere, expired, the device -// lost) while somebody is still watching, or while a capture is still starting. +// why: a capture belongs to the lease or the device it films, not to a connection -- the lease is the thing +// that says this machine still owns that device, and the device is the thing that says it still runs. A +// viewer going away stops a capture through its own request; this covers what that cannot, a lease that +// ends somewhere else (released elsewhere, expired, the device lost) while somebody is still watching, or +// while a capture is still starting. import type { ModuleLeaseEndParticipant } from "../modules/index.js"; export interface CaptureSlot { @@ -21,6 +22,10 @@ export interface CaptureRegistry { begin(leaseId: string): CaptureSlot; stop(leaseId: string): Promise; leases(): string[]; + // why: a device's stream belongs to the device, not to a lease, because a device nobody leases streams + // too; its captures end when the device stops running or its lease ends, whoever the viewer is. + beginOnDevice(deviceId: string): CaptureSlot; + stopOnDevice(deviceId: string): Promise; } interface Slot { @@ -33,7 +38,13 @@ interface Slot { closing: Promise | undefined; } -export function createCaptureRegistry(): CaptureRegistry { +interface SlotTable { + begin(key: string): CaptureSlot; + stop(key: string): Promise; + keys(): string[]; +} + +function createSlotTable(): SlotTable { const running = new Map>(); const closeSlot = async (slot: Slot): Promise => { @@ -87,7 +98,7 @@ export function createCaptureRegistry(): CaptureRegistry { }, }; }, - leases: () => [...running.keys()], + keys: () => [...running.keys()], stop: async (leaseId) => { const forLease = running.get(leaseId); if (forLease === undefined) return 0; @@ -99,6 +110,18 @@ export function createCaptureRegistry(): CaptureRegistry { }; } +export function createCaptureRegistry(): CaptureRegistry { + const leases = createSlotTable(); + const devices = createSlotTable(); + return { + begin: leases.begin, + beginOnDevice: devices.begin, + leases: leases.keys, + stop: leases.stop, + stopOnDevice: devices.stop, + }; +} + export function createStreamLeaseEnd(captures: CaptureRegistry): ModuleLeaseEndParticipant { return { // why: waits for each capture's close, which a driver bounds, so a reclaim does not wipe or @@ -106,7 +129,7 @@ export function createStreamLeaseEnd(captures: CaptureRegistry): ModuleLeaseEndP // the wait if leasing gives up first. leaseEnded: async (ended, signal) => { await Promise.race([ - captures.stop(ended.leaseId), + Promise.all([captures.stop(ended.leaseId), captures.stopOnDevice(ended.deviceId)]), new Promise((resolve) => { if (signal.aborted) resolve(); else signal.addEventListener("abort", () => resolve(), { once: true }); diff --git a/src/stream/refusal.ts b/src/stream/refusal.ts index 728b49131..ac71b7cc4 100644 --- a/src/stream/refusal.ts +++ b/src/stream/refusal.ts @@ -37,6 +37,14 @@ export function deviceNotRunning(state: string): StreamRefusalError { ); } +export function deviceStopped(): StreamRefusalError { + return new StreamRefusalError( + 409, + "DEVICE_NOT_RUNNING", + "The device stopped before its picture started", + ); +} + export function noSuchStream(): StreamRefusalError { // why: the same status and code as an unknown lease on purpose -- a daemon with `capabilities.stream.enabled` off // never mounts this route at all and answers any path it has not heard of as a 404, so a client that diff --git a/src/stream/worker-link.ts b/src/stream/worker-link.ts index 20bd6430f..8fc08bb1f 100644 --- a/src/stream/worker-link.ts +++ b/src/stream/worker-link.ts @@ -15,6 +15,7 @@ import type { StreamCaller } from "./caller.js"; import { type CapturedStream, type FeedBounds, readScreenFeed } from "./feed.js"; import { deviceNotRunning, + deviceStopped, notStreamable, StreamRefusalError, streamUnavailable, @@ -56,7 +57,10 @@ export function createWorkerStreamLink(options: WorkerStreamLinkOptions): Stream // capture first would answer an agent one way for a lease this worker holds and another for a // lease that does not exist, which is a way of listing the fleet's leases. const { device, lease } = requireTarget(options, target, caller); - const slot = options.captures.begin(lease?.id ?? device.id); + const slot = + lease === undefined + ? options.captures.beginOnDevice(device.id) + : options.captures.begin(lease.id); // why: the listener, and not only the body's cancel, because a request aborted before its // first frame has no body being read yet -- which is what a page opened and closed quickly does. if (signal.aborted) void slot.release(); @@ -76,7 +80,8 @@ export function createWorkerStreamLink(options: WorkerStreamLinkOptions): Stream stream = await readScreenFeed(feed, options.bounds); } catch (error) { slot.abandon(); - if (slot.ended.aborted && lease !== undefined) throw unknownLease(lease.id); + if (slot.ended.aborted) + throw lease === undefined ? deviceStopped() : unknownLease(lease.id); throw refusalFor(error); } // why: the lease is read again once there is a picture, because a capture can take seconds to @@ -89,9 +94,13 @@ export function createWorkerStreamLink(options: WorkerStreamLinkOptions): Stream await slot.release(); throw streamUnavailable("The viewer went away before the picture started"); } - if (lease !== undefined && (slot.ended.aborted || still?.deviceId !== lease.deviceId)) { + if ( + lease === undefined + ? slot.ended.aborted + : slot.ended.aborted || still?.deviceId !== lease.deviceId + ) { await slot.release(); - throw unknownLease(lease.id); + throw lease === undefined ? deviceStopped() : unknownLease(lease.id); } return { ...stream, close: () => slot.release() }; }, From 16c8f13574c6a47bdc2eee65158bfb1597b6bc6f Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:29:39 +0200 Subject: [PATCH 11/20] test: a device that stops running ends its streams (#909) --- src/stream/create.test.ts | 52 +++++++++++++++++++++++++++++++++++++-- src/stream/create.ts | 3 ++- src/stream/worker.ts | 1 + 3 files changed, 53 insertions(+), 3 deletions(-) diff --git a/src/stream/create.test.ts b/src/stream/create.test.ts index f9f74a8c0..d1b4d1708 100644 --- a/src/stream/create.test.ts +++ b/src/stream/create.test.ts @@ -27,8 +27,17 @@ function worker( ); const feeds: ScriptedFeed[] = []; let opened = 0; + const handlers = new Map void)[]>(); const stream = createWorkerStream({ clock, + events: { + subscribe: (event, handler) => { + const forEvent = handlers.get(event) ?? []; + forEvent.push(handler as (typeof forEvent)[number]); + handlers.set(event, forEvent); + return () => undefined; + }, + }, instanceId: "w1", logger: new NoopLogger(), name: "stream", @@ -47,7 +56,10 @@ function worker( }, settings: { bitrateKbps: 1500, fps: 30, maxDimension: 720 }, }); - return { clock, feeds, opened: () => opened, registry, stream }; + const emit = (event: string, deviceId: string) => { + for (const handler of handlers.get(event) ?? []) handler({ payload: { deviceId } }); + }; + return { clock, emit, feeds, opened: () => opened, registry, stream }; } const STREAM_HEADERS = { @@ -314,7 +326,7 @@ describe("createWorkerStream", () => { expect(relays.map((relay) => relay.workerId)).toEqual(["w1", "w_busy", "w_missing", "w_gone"]); }); - it("follows its device: a lease granted leaves the stream running, a lease ended stops the capture before the reclaim hook returns, and a hook that runs while the capture opens refuses with 409", async () => { + it("follows its device: a lease granted leaves the stream running, a lease ended or a device that stops running ends the capture, and one that comes while the capture opens refuses with 409", async () => { { const { feeds, registry, stream } = worker(); const ended = (leaseId: string) => @@ -362,5 +374,41 @@ describe("createWorkerStream", () => { ); expect(feeds[0]?.closes).toBe(1); } + + for (const event of [ + "device.shutdown", + "device.absence-detected", + "device.quarantined", + "device.deleted", + ]) { + const { emit, feeds, stream } = worker(); + const watched = await stream.mount.handle(deviceRequest("w1", "dev_ready"), OPERATOR); + const other = await stream.mount.handle(deviceRequest("w1", "dev_leased"), OPERATOR); + const watchedReader = watched.body!.getReader(); + const otherReader = other.body!.getReader(); + await watchedReader.read(); + await otherReader.read(); + emit(event, "dev_ready"); + await settledWithin(Promise.resolve()); + expect(feeds[0]?.closes, event).toBe(1); + expect((await watchedReader.read()).done, event).toBe(true); + expect(feeds[1]?.closes, event).toBe(0); + feeds[1]?.keyframe([5]); + expect((await otherReader.read()).done, event).toBe(false); + await otherReader.cancel(); + } + + { + let openIt = () => {}; + const gate = new Promise((resolve) => (openIt = resolve)); + const { emit, feeds, stream } = worker([], true, false, gate); + const asked = stream.mount.handle(deviceRequest("w1", "dev_ready"), OPERATOR); + await settledWithin(asked); + emit("device.shutdown", "dev_ready"); + openIt(); + const refused = await asked; + expect(refused.status).toBe(409); + expect(feeds[0]?.closes).toBe(1); + } }); }); diff --git a/src/stream/create.ts b/src/stream/create.ts index b3c7992e9..26c8c876a 100644 --- a/src/stream/create.ts +++ b/src/stream/create.ts @@ -1,5 +1,5 @@ import type { ScreenAvailability, ScreenSettings } from "../core/index.js"; -import type { ModuleLeaseEndParticipant } from "../modules/index.js"; +import type { ModuleLeaseEndParticipant, WorkerContext } from "../modules/index.js"; import type { Clock, Logger } from "../ports/index.js"; import { createCaptureRegistry, createStreamLeaseEnd } from "./captures.js"; import { DEFAULT_SILENCE_MS, DEFAULT_START_MS } from "./feed.js"; @@ -16,6 +16,7 @@ import { createWorkerStreamLink, type ScreenOpener, type StreamRegistry } from " export interface WorkerStreamOptions { readonly name: string; readonly instanceId: string; + readonly events: WorkerContext["events"]; readonly registry: StreamRegistry; readonly screens: ScreenOpener & { screenAvailability(): Promise; diff --git a/src/stream/worker.ts b/src/stream/worker.ts index b0f8dbcc5..5aa026441 100644 --- a/src/stream/worker.ts +++ b/src/stream/worker.ts @@ -6,6 +6,7 @@ import type { StreamModuleConfig } from "./definition.js"; export function createWorker(ctx: WorkerContext): WorkerModule { const stream = createWorkerStream({ clock: ctx.clock, + events: ctx.events, instanceId: ctx.instanceId, logger: ctx.logger, name: ctx.name, From 8e43fdcba76adf1dbeeb6e5ec8b2f87535889105 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:29:50 +0200 Subject: [PATCH 12/20] feat(stream): a device that stops running ends the captures on it (#909) --- src/stream/create.ts | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/src/stream/create.ts b/src/stream/create.ts index 26c8c876a..a1c5f5bc6 100644 --- a/src/stream/create.ts +++ b/src/stream/create.ts @@ -13,6 +13,13 @@ import { import type { StreamServiceOptions } from "./service.js"; import { createWorkerStreamLink, type ScreenOpener, type StreamRegistry } from "./worker-link.js"; +const DEVICE_STOP_EVENTS = [ + "device.shutdown", + "device.absence-detected", + "device.quarantined", + "device.deleted", +] as const; + export interface WorkerStreamOptions { readonly name: string; readonly instanceId: string; @@ -54,6 +61,14 @@ export function createWorkerStream(options: WorkerStreamOptions): WorkerStream { }), logger: options.logger, }; + // why: a device that stops with no lease end (the warm pool's shutdown, the reaper, a phone unplugged, + // quarantine, a delete) runs no reclaim hook, and a capture does not end by itself when its device + // does -- a simulator's repeats its last frame -- so these four events are what end it. + for (const event of DEVICE_STOP_EVENTS) { + options.events.subscribe(event, (envelope) => { + void captures.stopOnDevice(envelope.payload.deviceId); + }); + } return { capabilitySource: { name: options.name, From 163cb471c766f3641607ee64e0d0cfbdc0185666 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:30:43 +0200 Subject: [PATCH 13/20] feat(stream): declare the device stream route in the OpenAPI document (#909) --- src/http/core-routes.test.ts | 5 ++++- src/http/openapi.test.ts | 28 ++++++++++++++++++++++- src/stream/definition.ts | 43 ++++++++++++++++++++++++++++++++++++ 3 files changed, 74 insertions(+), 2 deletions(-) diff --git a/src/http/core-routes.test.ts b/src/http/core-routes.test.ts index fe7619101..8f930f32e 100644 --- a/src/http/core-routes.test.ts +++ b/src/http/core-routes.test.ts @@ -85,7 +85,10 @@ describe("the routes createHttpApp registers", () => { expect(registered.some((route) => route.path.startsWith("/v1/stream"))).toBe(true); expect(registered.some((route) => route.path.startsWith("/v1/agent-device"))).toBe(true); expect(coverageProblems(core, declared)).toEqual([]); - expect(moduleRoutes).toEqual(["GET /v1/stream/leases/{id}"]); + expect(moduleRoutes).toEqual([ + "GET /v1/stream/leases/{id}", + "GET /v1/stream/workers/{workerId}/devices/{deviceId}", + ]); expect([...documented].sort()).toEqual([...declared, ...moduleRoutes].sort()); expect(coverageProblems([...core, "GET /v1/new"], declared)).toEqual([ "not declared: GET /v1/new", diff --git a/src/http/openapi.test.ts b/src/http/openapi.test.ts index f87249c45..5049d05ba 100644 --- a/src/http/openapi.test.ts +++ b/src/http/openapi.test.ts @@ -350,7 +350,7 @@ describe("buildOpenApiDocument with a module that carries http", () => { expect(result.valid).toBe(true); }); - it("an integrator reads GET /v1/stream/leases/{id} under stream, with its headers, binary 200, 405 and refusals, and no agent-device or uplink path", () => { + it("an integrator reads GET /v1/stream/leases/{id} and GET /v1/stream/workers/{workerId}/devices/{deviceId} under stream, with its headers, binary 200, 405 and refusals, and no agent-device or uplink path", () => { const document = buildOpenApiDocument(optionalModules); const operation = document.paths?.["/v1/stream/leases/{id}"]?.get; const ok = (operation?.responses ?? {})["200"] as OkResponse; @@ -387,6 +387,32 @@ describe("buildOpenApiDocument with a module that carries http", () => { "502": ["STREAM_UNAVAILABLE"], "503": ["STREAM_UNAVAILABLE", "WORKER_UNREACHABLE"], }); + const device = document.paths?.["/v1/stream/workers/{workerId}/devices/{deviceId}"]?.get; + expect(device?.tags).toEqual(["stream"]); + expect( + Object.fromEntries( + ["401", "403", "404", "405", "409", "415", "500", "502", "503"].map((status) => [ + status, + errorCodes(device, status), + ]), + ), + ).toEqual({ + "401": ["UNAUTHENTICATED"], + "403": ["FORBIDDEN"], + "404": ["UNKNOWN_DEVICE", "UNKNOWN_WORKER"], + "405": ["BAD_REQUEST"], + "409": ["DEVICE_NOT_RUNNING"], + "415": ["STREAM_NOT_SUPPORTED"], + "500": ["INTERNAL"], + "502": ["STREAM_UNAVAILABLE"], + "503": ["STREAM_UNAVAILABLE", "WORKER_UNREACHABLE"], + }); + expect(sortedKeys(((device?.responses ?? {})["200"] as OkResponse).headers)).toEqual([ + "simlock-stream-codec", + "simlock-stream-fps", + "simlock-stream-height", + "simlock-stream-width", + ]); expect( sortedKeys(document.paths).filter( (path) => path.startsWith("/v1/agent-device") || path === "/v1/uplink", diff --git a/src/stream/definition.ts b/src/stream/definition.ts index 16baa431e..6e8193b91 100644 --- a/src/stream/definition.ts +++ b/src/stream/definition.ts @@ -62,6 +62,49 @@ export const streamModule: OptionalModuleDefinition = { }, }, }, + { + method: "get", + path: "/v1/stream/workers/{workerId}/devices/{deviceId}", + description: + "Watch the screen of a running device, leased or not: operator tokens only. A single worker answers for its own id; a gateway relays to the named worker. Served only when capabilities.stream.enabled is true. A gateway keeps a worker's refusal with its code.", + statuses: { + 200: { + description: "The picture, as a binary body of frames; its codec and size are headers.", + contentTypes: [ + "application/vnd.simlock.stream+h264", + "application/vnd.simlock.stream+mjpeg", + ], + headers: { + "simlock-stream-codec": "The codec of the body: h264 or mjpeg.", + "simlock-stream-fps": "Frames per second.", + "simlock-stream-height": "Picture height in pixels.", + "simlock-stream-width": "Picture width in pixels.", + }, + }, + 404: { + description: "No such device or worker, or no such stream path.", + errorCodes: ["UNKNOWN_DEVICE", "UNKNOWN_WORKER"], + }, + 405: { description: "A stream is read, not written.", errorCodes: ["BAD_REQUEST"] }, + 409: { + description: "The device is not ready or leased.", + errorCodes: ["DEVICE_NOT_RUNNING"], + }, + 415: { + description: "The device cannot be streamed.", + errorCodes: ["STREAM_NOT_SUPPORTED"], + }, + 500: { description: "The stream failed unexpectedly.", errorCodes: ["INTERNAL"] }, + 502: { + description: "A gateway's worker answered success with no picture.", + errorCodes: ["STREAM_UNAVAILABLE"], + }, + 503: { + description: "The stream is not available, or a gateway cannot reach the worker.", + errorCodes: ["STREAM_UNAVAILABLE", "WORKER_UNREACHABLE"], + }, + }, + }, ], }, loadGateway: () => import("./gateway.js"), From 4ffc05579d5a85f635caa30f220913480e311f80 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:33:05 +0200 Subject: [PATCH 14/20] test(e2e): the slow-lane stream tests call the device route, and an unleased simulator streams (#909) --- docs/internal/ARCHITECTURE.md | 2 +- e2e/helpers/device-stream.ts | 16 ++++ e2e/helpers/index.ts | 1 + e2e/slow-android-physical.test.ts | 18 ++-- e2e/slow-android-stream.test.ts | 22 ++--- e2e/slow-ios-physical-manual.test.ts | 11 ++- e2e/slow-ios-physical.test.ts | 15 +-- e2e/slow-ios-stream.test.ts | 95 ++++++++++++++++++- src/stream/AGENTS.md | 59 +++++++----- .../using-devices/watching-the-screen.mdx | 2 +- 10 files changed, 185 insertions(+), 56 deletions(-) create mode 100644 e2e/helpers/device-stream.ts diff --git a/docs/internal/ARCHITECTURE.md b/docs/internal/ARCHITECTURE.md index 48a7515db..3e372a0ff 100644 --- a/docs/internal/ARCHITECTURE.md +++ b/docs/internal/ARCHITECTURE.md @@ -48,7 +48,7 @@ worker daemon ── uplink (WebSocket to /v1/uplink, worker dials) ──┘ | alerts | Gateway only: chosen fleet events POSTed to webhook destinations | [alerts](../../src/alerts/AGENTS.md) | | link | Byte channels over the uplink | [link](../../src/link/AGENTS.md) | | agent-device | Optional edge: the two agent-device daemons (simulators, phones) beside a worker, its HTTP front | [agent-device](../../src/agent-device/AGENTS.md) | -| stream | Optional edge: a leased device's live screen over HTTP, relayed by a gateway | [stream](../../src/stream/AGENTS.md) | +| stream | Optional edge: a running device's live screen over HTTP, relayed by a gateway | [stream](../../src/stream/AGENTS.md) | | auth | Credential → identity: tokens, admin secret, `hello`, uplink join token, HTTP strategies | [auth](../../src/auth/AGENTS.md) | | http | HTTP frontend: routes, SSE | [http](../../src/http/AGENTS.md) | | cli | Operator command surface | [cli](../../src/cli/AGENTS.md) | diff --git a/e2e/helpers/device-stream.ts b/e2e/helpers/device-stream.ts new file mode 100644 index 000000000..98ffd5cc4 --- /dev/null +++ b/e2e/helpers/device-stream.ts @@ -0,0 +1,16 @@ +import { readFile } from "node:fs/promises"; +import { join } from "node:path"; + +// why: the device stream is asked for by the worker's own id, which a single worker keeps in its +// `instance.json`; a lane that streams reads it there instead of asking the daemon, so the URL it +// builds is the one an operator's tool would. +export async function deviceStreamUrl( + base: string, + home: string, + deviceId: string, +): Promise { + const instance = JSON.parse(await readFile(join(home, "instance.json"), "utf8")) as { + readonly id: string; + }; + return `${base}/v1/stream/workers/${encodeURIComponent(instance.id)}/devices/${encodeURIComponent(deviceId)}`; +} diff --git a/e2e/helpers/index.ts b/e2e/helpers/index.ts index 9fb0cbb1a..601bc0da0 100644 --- a/e2e/helpers/index.ts +++ b/e2e/helpers/index.ts @@ -12,6 +12,7 @@ export { } from "./agent-device.js"; export type { Agent, AgentDeviceResult } from "./agent-device.js"; export { freeLoopbackPort } from "./port.js"; +export { deviceStreamUrl } from "./device-stream.js"; export { withHttpDaemon } from "./http-daemon.js"; // A flow that factors its own setup out of a test body needs to name what `withDaemon` hands // back; every other flow keeps it inferred. diff --git a/e2e/slow-android-physical.test.ts b/e2e/slow-android-physical.test.ts index 142d8e642..1e0ff34a7 100644 --- a/e2e/slow-android-physical.test.ts +++ b/e2e/slow-android-physical.test.ts @@ -2,7 +2,13 @@ import { execFile } from "node:child_process"; import { promisify } from "node:util"; import { describe, expect, it } from "vitest"; -import { type TestEnv, waitFor, waitForDeviceState, withHttpDaemon } from "./helpers/index.js"; +import { + deviceStreamUrl, + type TestEnv, + waitFor, + waitForDeviceState, + withHttpDaemon, +} from "./helpers/index.js"; const execFileAsync = promisify(execFile); @@ -30,7 +36,7 @@ interface Frame { } interface Grant { - readonly device: { readonly driverDeviceId: string }; + readonly device: { readonly driverDeviceId: string; readonly id: string }; readonly environment: Record; readonly lease: { readonly id: string }; } @@ -186,8 +192,8 @@ async function readFrames( return frames; } -async function streamOf(daemon: Rig, leaseId: string): Promise { - const response = await fetch(`${daemon.base}/v1/stream/leases/${encodeURIComponent(leaseId)}`, { +async function streamOf(daemon: Rig, deviceId: string): Promise { + const response = await fetch(await deviceStreamUrl(daemon.base, daemon.env.home, deviceId), { headers: daemon.auth, // why: covers the ~185 s worst case of opening the stream, so a stall fails here naming its step. signal: AbortSignal.timeout(STREAM_OPEN_BOUND), @@ -297,7 +303,7 @@ describe.skipIf(missing !== undefined)( ); it( - "a lease on the phone streams a meta frame and a key frame over GET /v1/stream/leases/{leaseId}, and after the lease ends the phone holds no simlock-scrcpy jar and no scrcpy server process and the host keeps no capture shell, with the phone's screen off when the stream asks and its stay-awake setting put back within 10 s of the lease's end", + "a lease on the phone streams a meta frame and a key frame over GET /v1/stream/workers/{workerId}/devices/{deviceId}, and after the lease ends the phone holds no simlock-scrcpy jar and no scrcpy server process and the host keeps no capture shell, with the phone's screen off when the stream asks and its stay-awake setting put back within 10 s of the lease's end", { timeout: 10 * MINUTE }, async () => { const stayOn = ( @@ -330,7 +336,7 @@ describe.skipIf(missing !== undefined)( await enroll(daemon.env, serial); const grant = await leasePhysical(daemon.env); - const response = await streamOf(daemon, grant.lease.id); + const response = await streamOf(daemon, grant.device.id); if (response.body === null) throw new Error("the stream answered without a body"); const frames = await readFrames( response.body, diff --git a/e2e/slow-android-stream.test.ts b/e2e/slow-android-stream.test.ts index 84849dd6a..8d82ce068 100644 --- a/e2e/slow-android-stream.test.ts +++ b/e2e/slow-android-stream.test.ts @@ -4,7 +4,7 @@ import { homedir } from "node:os"; import { join } from "node:path"; import { describe, expect, it } from "vitest"; -import { DEVICE_READY_MS, waitFor, withHttpDaemon } from "./helpers/index.js"; +import { DEVICE_READY_MS, deviceStreamUrl, waitFor, withHttpDaemon } from "./helpers/index.js"; const ANDROID_HOME = process.env.ANDROID_HOME ?? @@ -84,18 +84,17 @@ describe.skipIf(!hasAndroidSdk)( ); expect(lease.code, `lease failed: ${lease.stdout} ${lease.stderr}`).toBe(0); const leaseId = (lease.json as { lease: { id: string } }).lease.id; + const deviceId = (lease.json as { device: { id: string } }).device.id; + const streamUrl = await deviceStreamUrl(baseUrl, env.home, deviceId); const created = await env.cli(["token", "create", "--role", "operator"]); expect(created.code, created.stderr).toBe(0); const authorization = `Bearer ${(created.json as { secret: string }).secret}`; const viewer = new AbortController(); - const response = await fetch( - `${baseUrl}/v1/stream/leases/${encodeURIComponent(leaseId)}`, - { - headers: { authorization }, - signal: viewer.signal, - }, - ); + const response = await fetch(streamUrl, { + headers: { authorization }, + signal: viewer.signal, + }); expect( response.status, await response @@ -153,10 +152,7 @@ describe.skipIf(!hasAndroidSdk)( // why: an untouched emulator's encoder sends nothing once its picture is still, so a stream // held well past the 30 s silence bound proves the still screen is kept alive on a real // device, and the lease released under that viewer proves the stop on a device too. - const watching = await fetch( - `${baseUrl}/v1/stream/leases/${encodeURIComponent(leaseId)}`, - { headers: { authorization } }, - ); + const watching = await fetch(streamUrl, { headers: { authorization } }); expect(watching.status).toBe(200); if (watching.body === null) throw new Error("the stream answered without a body"); const watched = watching.body.getReader(); @@ -209,7 +205,7 @@ describe.skipIf(!hasAndroidSdk)( }, ); - const last = await fetch(`${baseUrl}/v1/stream/leases/${encodeURIComponent(leaseId)}`, { + const last = await fetch(streamUrl, { headers: { authorization }, }); expect(last.status).toBe(200); diff --git a/e2e/slow-ios-physical-manual.test.ts b/e2e/slow-ios-physical-manual.test.ts index e2aca2820..b8a230f44 100644 --- a/e2e/slow-ios-physical-manual.test.ts +++ b/e2e/slow-ios-physical-manual.test.ts @@ -2,7 +2,13 @@ import { execFile } from "node:child_process"; import { promisify } from "node:util"; import { describe, expect, it } from "vitest"; -import { freeLoopbackPort, waitFor, withDaemon, type TestEnv } from "./helpers/index.js"; +import { + deviceStreamUrl, + freeLoopbackPort, + waitFor, + withDaemon, + type TestEnv, +} from "./helpers/index.js"; const execFileAsync = promisify(execFile); @@ -166,12 +172,13 @@ describe.skipIf(process.platform !== "darwin" || missing !== undefined)( ); expect(leased.code, `lease failed: ${leased.stderr}`).toBe(0); const leaseId = (leased.json as { lease: { id: string } }).lease.id; + const deviceId = (leased.json as { device: { id: string } }).device.id; const created = await env.cli(["token", "create", "--role", "operator"]); expect(created.code, created.stderr).toBe(0); const viewer = new AbortController(); const response = await fetch( - `http://127.0.0.1:${String(env.port)}/v1/stream/leases/${encodeURIComponent(leaseId)}`, + await deviceStreamUrl(`http://127.0.0.1:${String(env.port)}`, env.home, deviceId), { headers: { authorization: `Bearer ${(created.json as { secret: string }).secret}` }, signal: viewer.signal, diff --git a/e2e/slow-ios-physical.test.ts b/e2e/slow-ios-physical.test.ts index d9b9cbc90..56b3ed946 100644 --- a/e2e/slow-ios-physical.test.ts +++ b/e2e/slow-ios-physical.test.ts @@ -8,6 +8,7 @@ import { describe, expect, it } from "vitest"; import { REPO_ROOT, + deviceStreamUrl, SCREEN_TAGS as TAG, type TestEnv, readScreenFrames as readFrames, @@ -44,7 +45,7 @@ const missing = : undefined; interface Grant { - readonly device: { readonly driverDeviceId: string }; + readonly device: { readonly driverDeviceId: string; readonly id: string }; readonly environment: Record; readonly lease: { readonly id: string }; } @@ -489,9 +490,9 @@ describe.skipIf(process.platform !== "darwin" || missing !== undefined)( const daemon = await rig(); await enroll(daemon.env, udid); const grant = await leasePhysical(daemon.env, "phone-lane-stream"); - const open = async (leaseId: string, signal: AbortSignal) => { + const open = async (deviceId: string, signal: AbortSignal) => { const response = await fetch( - `${daemon.base}/v1/stream/leases/${encodeURIComponent(leaseId)}`, + await deviceStreamUrl(daemon.base, daemon.env.home, deviceId), { headers: daemon.auth, signal }, ); if (response.body === null) throw new Error("the stream answered without a body"); @@ -514,7 +515,7 @@ describe.skipIf(process.platform !== "darwin" || missing !== undefined)( }; const viewer = new AbortController(); - const first = await open(grant.lease.id, viewer.signal); + const first = await open(grant.device.id, viewer.signal); const firstReader = await reading(first); expect(first.response.headers.get("simlock-stream-codec")).toBe("h264"); const frames = await readFrames(firstReader, (read) => @@ -544,7 +545,7 @@ describe.skipIf(process.platform !== "darwin" || missing !== undefined)( false, ); - const second = await open(grant.lease.id, new AbortController().signal); + const second = await open(grant.device.id, new AbortController().signal); if (second.response.status === 200) { await readFrames(second.reader!, (read) => read.some((frame) => frame.tag === TAG.keyframe), @@ -571,7 +572,7 @@ describe.skipIf(process.platform !== "darwin" || missing !== undefined)( // why: the stream's own first-frame bound is 30 s, so a body that produced no key frame by // then has already failed or will not fail visibly; the race only names the claim. const again = await leasePhysical(daemon.env, "phone-lane-stream-2"); - const reopened = await open(again.lease.id, new AbortController().signal); + const reopened = await open(again.device.id, new AbortController().signal); const reopenedReader = await reading(reopened); const keyFrameArrived = await Promise.race([ readFrames(reopenedReader, (read) => @@ -599,7 +600,7 @@ describe.skipIf(process.platform !== "darwin" || missing !== undefined)( const grant = await leasePhysical(daemon.env, "phone-lane-old-ios"); const asked = Date.now(); const response = await fetch( - `${daemon.base}/v1/stream/leases/${encodeURIComponent(grant.lease.id)}`, + await deviceStreamUrl(daemon.base, daemon.env.home, grant.device.id), { headers: daemon.auth, signal: AbortSignal.timeout(10_000) }, ); // why: the status is read before any reader is taken; a 200's endless body is never read. diff --git a/e2e/slow-ios-stream.test.ts b/e2e/slow-ios-stream.test.ts index 279b07680..7d9da7e90 100644 --- a/e2e/slow-ios-stream.test.ts +++ b/e2e/slow-ios-stream.test.ts @@ -6,6 +6,8 @@ import { describe, expect, it } from "vitest"; import { DEVICE_READY_MS, + deviceStreamUrl, + listDevices, REPO_ROOT, SCREEN_TAGS as TAG, readScreenFrames as readFrames, @@ -16,6 +18,8 @@ import { const execFileAsync = promisify(execFile); const STREAM_OPEN_BOUND = 240_000; +// why: an exact model name, as a warm target needs; the same model the other iOS lanes lease. +const WARM_MODEL = "iPhone 16"; const KEY_FRAME_BOUND = 120_000; async function within(work: Promise, bound: number, what: string): Promise { @@ -70,12 +74,13 @@ describe.skipIf(process.platform !== "darwin")( ); expect(lease.code, `lease failed: ${lease.stdout} ${lease.stderr}`).toBe(0); const leaseId = (lease.json as { lease: { id: string } }).lease.id; + const deviceId = (lease.json as { device: { id: string } }).device.id; const created = await env.cli(["token", "create", "--role", "operator"]); expect(created.code, created.stderr).toBe(0); const authorization = `Bearer ${(created.json as { secret: string }).secret}`; const open = async (signal: AbortSignal) => { const response = await within( - fetch(`${baseUrl}/v1/stream/leases/${encodeURIComponent(leaseId)}`, { + fetch(await deviceStreamUrl(baseUrl, env.home, deviceId), { headers: { authorization }, signal, }), @@ -167,6 +172,94 @@ describe.skipIf(process.platform !== "darwin")( }, ); + it( + "streams a simulator nobody has leased, and keeps streaming for 10 s when a lease is granted on that simulator", + { timeout: 480_000 }, + async () => { + expect(await capturesRunning(), "a capture was already running before the test").toBe(0); + const { env, port } = await withHttpDaemon((httpPort) => ({ + configOverrides: { + http: { enabled: true, host: "127.0.0.1", port: httpPort }, + capabilities: { + stream: { bitrateKbps: 1500, enabled: true, fps: 30, maxDimension: 720 }, + }, + warmPool: { targets: [{ count: 1, model: WARM_MODEL, platform: "ios" }] }, + }, + driver: "real", + })); + const baseUrl = `http://127.0.0.1:${String(port)}`; + try { + let warmId = ""; + await waitFor( + async () => { + warmId = + (await listDevices(env)).find((device) => device.state === "ready")?.id ?? ""; + return warmId !== ""; + }, + { + interval: 2_000, + label: "the warm pool keeps one simulator ready", + timeout: DEVICE_READY_MS, + }, + ); + const created = await env.cli(["token", "create", "--role", "operator"]); + expect(created.code, created.stderr).toBe(0); + const viewer = new AbortController(); + const response = await within( + fetch(await deviceStreamUrl(baseUrl, env.home, warmId), { + headers: { authorization: `Bearer ${(created.json as { secret: string }).secret}` }, + signal: viewer.signal, + }), + STREAM_OPEN_BOUND, + "the unleased simulator's stream to open", + ); + const detail = response.status === 200 ? "" : await response.text().catch(() => ""); + expect(response.status, detail).toBe(200); + if (response.body === null) throw new Error("the stream answered without a body"); + const reader = response.body.getReader(); + const first = await within( + readFrames(reader, (read) => read.some((frame) => frame.tag === TAG.keyframe)), + KEY_FRAME_BOUND, + "the unleased simulator's key frame", + ); + expect(first.some((frame) => frame.tag === TAG.keyframe)).toBe(true); + + const lease = await env.cli( + [ + "lease", + "--platform", + "ios", + "--device", + WARM_MODEL, + "--agent-id", + "ios-stream-watched", + "--detach", + ], + { timeout: DEVICE_READY_MS }, + ); + expect(lease.code, `lease failed: ${lease.stdout} ${lease.stderr}`).toBe(0); + expect( + (lease.json as { device: { id: string } }).device.id, + "the lease took another simulator", + ).toBe(warmId); + const keepUntil = Date.now() + 10_000; + const kept = await within( + readFrames(reader, () => Date.now() > keepUntil), + KEY_FRAME_BOUND, + "frames after the lease was granted", + ); + expect(kept.length, "the stream sent nothing for 10 s after the lease").toBeGreaterThan( + 0, + ); + viewer.abort(); + } finally { + await env + .cli(["nuke", "--delete-devices", "--yes"], { timeout: 120_000 }) + .catch(() => undefined); + } + }, + ); + it( "the simulator capture program exits 64 on missing usage and 1 when its device is not there", { timeout: 120_000 }, diff --git a/src/stream/AGENTS.md b/src/stream/AGENTS.md index d576b9713..a9781d91a 100644 --- a/src/stream/AGENTS.md +++ b/src/stream/AGENTS.md @@ -1,6 +1,6 @@ # Stream -Does: serves a leased device's live screen at `GET /v1/stream/leases/{leaseId}`, on a worker and through a gateway. +Does: serves a running device's live screen to operators at `GET /v1/stream/workers/{workerId}/devices/{deviceId}` (and, until #910, a lease's at `GET /v1/stream/leases/{leaseId}`), on a worker and through a gateway. Entry: `index.ts` → `streamModule`, an optional module (ADR 0030): `worker.ts` `createWorker`, `gateway.ts` `createGateway`, loaded by the module host only when turned on. Uses: core (`ScreenFeed`, frame tags, registry records), modules (the context and hook types), ports. @@ -9,43 +9,51 @@ Off unless `capabilities.stream.enabled` (false). Off: nothing loaded, and the ` ## Files -| File | Role | -| ------------------------- | --------------------------------------------------------------------------------- | -| `definition.ts` | `streamModule`: name `stream`, config section, `http` route data, the two loaders | -| `worker.ts`, `gateway.ts` | The module's two sides, built from the host's context | -| `create.ts` | The two factories; the `stream` capability read from each driver's availability | -| `worker-link.ts` | Worker: lease check, open through the driver, register, re-check after frame 1 | -| `gateway-link.ts` | Gateway: fleet ownership, relay to the holding worker, refusal and header check | -| `feed.ts` | `readScreenFeed`: meta frame first, start and silence bounds, close on any end | -| `framing.ts` | `MetaFrameScanner`, `checkedMeta`: codec and size a picture may claim | -| `captures.ts` | Captures per lease; the lease-end participant that stops them | -| `service.ts` | One request either door: method, refusal body, response headers, log line | -| `mounts.ts` | The HTTP mount and the relay target | -| `refusal.ts` | `StreamRefusalError`, each refusal's status and code, a relayed refusal checked | -| `caller.ts` | Who asks | -| `test-fixtures.ts` | `TestRegistry`, `ScriptedFeed` for tests | -| `testing.ts` | What other components' tests may import: `createStreamMount` | +| File | Role | +| ------------------------- | ------------------------------------------------------------------------------------------ | +| `definition.ts` | `streamModule`: name `stream`, config section, `http` route data, the two loaders | +| `worker.ts`, `gateway.ts` | The module's two sides, built from the host's context | +| `create.ts` | The two factories; the `stream` capability; the four device events that stop captures | +| `worker-link.ts` | Worker: lease or device check, open through the driver, register, re-check after frame 1 | +| `gateway-link.ts` | Gateway: fleet ownership (lease) or known worker (device), relay, refusal and header check | +| `feed.ts` | `readScreenFeed`: meta frame first, start and silence bounds, close on any end | +| `framing.ts` | `MetaFrameScanner`, `checkedMeta`: codec and size a picture may claim | +| `captures.ts` | Captures per lease and per device; the lease-end participant that stops them | +| `service.ts` | One request either door: method, refusal body, response headers, log line | +| `mounts.ts` | The HTTP mount and the relay target | +| `refusal.ts` | `StreamRefusalError`, each refusal's status and code, a relayed refusal checked | +| `caller.ts` | Who asks | +| `test-fixtures.ts` | `TestRegistry`, `ScriptedFeed` for tests | +| `testing.ts` | What other components' tests may import: `createStreamMount` | ## How it works Worker — `worker-link.ts` `createWorkerStreamLink`: -1. Lease from the registry snapshot. Agent: its requester (relayed) or principal must hold it. - Unknown and not yours both answer 404 `UNKNOWN_LEASE`; nothing is opened first. -2. `captures.begin(leaseId)` — registered before the capture starts. +1. Lease route: the lease from the registry snapshot. Agent: its requester (relayed) or principal + must hold it. Unknown and not yours both answer 404 `UNKNOWN_LEASE`; nothing is opened first. + Device route: operator only (an agent gets 404 `UNKNOWN_DEVICE`, as for any device), the path's + worker id must be this worker's own (404 `UNKNOWN_WORKER`), the device must exist (404 + `UNKNOWN_DEVICE`) and be `ready` or `leased` (409 `DEVICE_NOT_RUNNING`). A relayed device + request is served whatever the tenant: the gateway has already refused non-operators. +2. `captures.begin(leaseId)` or `captures.beginOnDevice(deviceId)` — registered before the capture starts. 3. `ctx.screens.openScreen(platform, device, settings)` (the daemon passes its `DriverCatalog`) → the driver's `ScreenFeed`. `ScreenNotSupportedError` → 415 `STREAM_NOT_SUPPORTED`; anything else → 503 `STREAM_UNAVAILABLE`. 4. `readScreenFeed`: the meta frame within the start bound, or 503 and close. -5. Lease re-read: gone, re-granted, or its slot stopped → close, 404. +5. Lease re-read: gone, re-granted, or its slot stopped → close, 404. A device slot stopped → close, 409 `DEVICE_NOT_RUNNING`. 6. 200; headers `simlock-stream-{codec,fps,width,height}`; body = the frames, meta first. Ends — each closes the capture once (`closedOnce`): the viewer cancels or aborts; the capture ends or errors; no byte within the silence bound; the lease ends (the `beforeDeviceReclaim` hook, `createStreamLeaseEnd` → -`captures.stop`, which waits for every close, bounded by the driver and by leasing's signal). +`captures.stop` and `captures.stopOnDevice`, which wait for every close, bounded by the driver and by leasing's signal); +its device stops with no lease end (`device.shutdown`, `device.absence-detected`, `device.quarantined` or +`device.deleted` → `captures.stopOnDevice`). A lease granted while streaming ends nothing. Gateway — `gateway-link.ts`: the fleet lease index names owner, worker and the worker's lease id. Not the owner (agent) → 404. Relays `GET /stream/leases/` as target `stream` -with tenant `` from the index, never from the caller. A refusal +with tenant `` from the index, never from the caller. A device +request: not an operator → 404 `UNKNOWN_DEVICE`; a worker not in the fleet → 404 `UNKNOWN_WORKER`; +else relays `GET /stream/devices/` with tenant `` and no lease id. A refusal body is read to 16 KiB; its code and message are forwarded only if the code matches `^[A-Z][A-Z0-9_]{0,63}$`, else the code becomes `STREAM_UNAVAILABLE`. The worker's 4xx or 5xx status is kept either way; a 2xx without a picture → 502. Headers go through `checkedMeta`. @@ -56,7 +64,7 @@ cannot says why in its 503); `unavailable` when none can, with each platform's r ## State and timers -Memory only: running captures per lease (`captures.ts`). +Memory only: running captures per lease and per device (`captures.ts`). | What | Default | Key | | ---------------------------- | -------------------- | ------------------------------------------------------------------------------------------------ | @@ -70,7 +78,8 @@ Memory only: running captures per lease (`captures.ts`). - No platform name here: a driver offers `screenCapture` or not (`core/screen.ts`). - Ownership before any work; unknown and not-yours answer the same. - A capture is registered before it starts and the lease is re-read after its first frame, - so a capture never outlives its lease (`worker-link.test.ts`). + so a capture never outlives its lease or its device's running state (`worker-link.test.ts`, + `create.test.ts`). - A gateway never parses a video byte and never takes a caller's word for the requester. - No business events: watching changes no lease or device. Each open and refusal is logged. diff --git a/website/src/content/docs/docs/using-devices/watching-the-screen.mdx b/website/src/content/docs/docs/using-devices/watching-the-screen.mdx index 26a3b294d..1ee014180 100644 --- a/website/src/content/docs/docs/using-devices/watching-the-screen.mdx +++ b/website/src/content/docs/docs/using-devices/watching-the-screen.mdx @@ -100,7 +100,7 @@ At the defaults, a screen that is busy redrawing costs about 1.4 Mbit/s, and a s ## Stream from your own tool -Anything with an agent or operator token can read a lease's screen over HTTP at `GET /v1/stream/leases/`, on a worker or a gateway with streaming on. An agent token can read only its own leases. The body is Simlock's own framing of H.264 frames, made for the console's player, and it runs until you close the request. Most users watch in the console instead. +Anything with an operator token can read the screen of a running device, leased or not, over HTTP at `GET /v1/stream/workers//devices/`, on a worker (which answers only for its own id) or on a gateway (which relays to the named worker), with streaming on. A device streams while it is `ready` or `leased`; any other state answers 409 `DEVICE_NOT_RUNNING`. The stream ends when the device stops running or its lease ends. An agent token cannot read a device's screen. The older `GET /v1/stream/leases/` still works for now. The body is Simlock's own framing of H.264 frames, made for the console's player, and it runs until you close the request. Most users watch in the console instead. ## When it does not work From d11c14967172e686c6799651b4ef0bb3b31c2ec1 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:34:36 +0200 Subject: [PATCH 15/20] refactor(stream): keep the worker link's serve under the complexity bound (#909) --- src/http/openapi.test.ts | 6 +++++- src/stream/worker-link.ts | 26 +++++++++++++++----------- 2 files changed, 20 insertions(+), 12 deletions(-) diff --git a/src/http/openapi.test.ts b/src/http/openapi.test.ts index 5049d05ba..599ed2488 100644 --- a/src/http/openapi.test.ts +++ b/src/http/openapi.test.ts @@ -407,7 +407,7 @@ describe("buildOpenApiDocument with a module that carries http", () => { "502": ["STREAM_UNAVAILABLE"], "503": ["STREAM_UNAVAILABLE", "WORKER_UNREACHABLE"], }); - expect(sortedKeys(((device?.responses ?? {})["200"] as OkResponse).headers)).toEqual([ + expect(sortedKeys(okHeaders(device))).toEqual([ "simlock-stream-codec", "simlock-stream-fps", "simlock-stream-height", @@ -422,6 +422,10 @@ describe("buildOpenApiDocument with a module that carries http", () => { }); }); +function okHeaders(operation: { responses?: object } | undefined): Record { + return ((operation?.responses ?? {}) as Record)["200"]?.headers ?? {}; +} + type OkResponse = { content?: Record; headers?: Record }; function sortedKeys(record: object | undefined): string[] { diff --git a/src/stream/worker-link.ts b/src/stream/worker-link.ts index 8fc08bb1f..af1202fc3 100644 --- a/src/stream/worker-link.ts +++ b/src/stream/worker-link.ts @@ -80,33 +80,37 @@ export function createWorkerStreamLink(options: WorkerStreamLinkOptions): Stream stream = await readScreenFeed(feed, options.bounds); } catch (error) { slot.abandon(); - if (slot.ended.aborted) - throw lease === undefined ? deviceStopped() : unknownLease(lease.id); + if (slot.ended.aborted) throw stopped(lease); throw refusalFor(error); } // why: the lease is read again once there is a picture, because a capture can take seconds to // start and the lease it was asked for can end and the device be granted to someone else in // that time; a slot that was stopped says so even if a new lease reused the id. - const still = options.registry.snapshot.leases.find( - (candidate) => candidate.id === lease?.id, - ); if (signal.aborted) { await slot.release(); throw streamUnavailable("The viewer went away before the picture started"); } - if ( - lease === undefined - ? slot.ended.aborted - : slot.ended.aborted || still?.deviceId !== lease.deviceId - ) { + if (slot.ended.aborted || !leaseHolds(options.registry, lease)) { await slot.release(); - throw lease === undefined ? deviceStopped() : unknownLease(lease.id); + throw stopped(lease); } return { ...stream, close: () => slot.release() }; }, }; } +// why: a stream of a device has no lease to lose, so it is only ever stopped; a lease's stream is also +// refused when the lease is gone or has been granted to another device since the capture began. +function leaseHolds(registry: StreamRegistry, lease: LeaseRecord | undefined): boolean { + if (lease === undefined) return true; + const still = registry.snapshot.leases.find((candidate) => candidate.id === lease.id); + return still?.deviceId === lease.deviceId; +} + +function stopped(lease: LeaseRecord | undefined): StreamRefusalError { + return lease === undefined ? deviceStopped() : unknownLease(lease.id); +} + // why: the feed's reader, the viewer and the lease's end can each close a capture, and a driver's // close stops processes on a device; it runs once, and every later caller waits on the same stop. function closedOnce(feed: ScreenFeed): ScreenFeed { From e5b63c7b8d1f07e27b192be75c39c304ac57a2fb Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:35:03 +0200 Subject: [PATCH 16/20] test(e2e): list the warm simulator through the CLI (#909) --- e2e/slow-ios-stream.test.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/e2e/slow-ios-stream.test.ts b/e2e/slow-ios-stream.test.ts index 7d9da7e90..1fd140794 100644 --- a/e2e/slow-ios-stream.test.ts +++ b/e2e/slow-ios-stream.test.ts @@ -7,7 +7,6 @@ import { describe, expect, it } from "vitest"; import { DEVICE_READY_MS, deviceStreamUrl, - listDevices, REPO_ROOT, SCREEN_TAGS as TAG, readScreenFrames as readFrames, From 4f69b66f5838998aa8d5090228ee61a3620e0c2a Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:35:30 +0200 Subject: [PATCH 17/20] test(e2e): list the warm simulator through the CLI (#909) --- e2e/slow-ios-stream.test.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/e2e/slow-ios-stream.test.ts b/e2e/slow-ios-stream.test.ts index 1fd140794..5aad600db 100644 --- a/e2e/slow-ios-stream.test.ts +++ b/e2e/slow-ios-stream.test.ts @@ -191,8 +191,9 @@ describe.skipIf(process.platform !== "darwin")( let warmId = ""; await waitFor( async () => { - warmId = - (await listDevices(env)).find((device) => device.state === "ready")?.id ?? ""; + const listed = await env.cli(["list", "--devices"]); + const devices = (listed.json ?? []) as { id: string; state: string }[]; + warmId = devices.find((device) => device.state === "ready")?.id ?? ""; return warmId !== ""; }, { From 8151aa8faf70cc3178ebaa2b936349f9691d07e6 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:42:46 +0200 Subject: [PATCH 18/20] test(stream): a device stopped while its first frame is awaited refuses 409; the configuration page names a running device (#909) --- src/stream/create.test.ts | 11 +++++++++++ .../src/content/docs/docs/reference/configuration.mdx | 2 +- 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/src/stream/create.test.ts b/src/stream/create.test.ts index d1b4d1708..cef17808e 100644 --- a/src/stream/create.test.ts +++ b/src/stream/create.test.ts @@ -410,5 +410,16 @@ describe("createWorkerStream", () => { expect(refused.status).toBe(409); expect(feeds[0]?.closes).toBe(1); } + + { + const { emit, feeds, stream } = worker([], false); + const asked = stream.mount.handle(deviceRequest("w1", "dev_ready"), OPERATOR); + await settledWithin(asked); + expect(feeds).toHaveLength(1); + emit("device.shutdown", "dev_ready"); + const refused = await asked; + expect(refused.status).toBe(409); + expect(feeds[0]?.closes).toBe(1); + } }); }); diff --git a/website/src/content/docs/docs/reference/configuration.mdx b/website/src/content/docs/docs/reference/configuration.mdx index 49bdfc5fe..76a25c6b2 100644 --- a/website/src/content/docs/docs/reference/configuration.mdx +++ b/website/src/content/docs/docs/reference/configuration.mdx @@ -311,7 +311,7 @@ A name Simlock does not know is warned about and ignored. | `capabilities.agent-device.ios.bundleId` | non-empty string | not set | Bundle ID of agent-device's iPhone runner. Passed to the phone daemon. | Worker | | `capabilities.agent-device.ios.provisioningProfile` | non-empty string | not set | Provisioning profile for manual signing of the iPhone runner. Setting it switches agent-device to manual signing. Passed to the phone daemon. | Worker | | `capabilities.agent-device.ios.signingIdentity` | non-empty string | not set | Code-signing identity for manual signing of the iPhone runner. Passed to the phone daemon. | Worker | -| `capabilities.stream.enabled` | boolean | `false` | Let people watch a leased device's screen. On a gateway: pass each stream through from the worker. | Both | +| `capabilities.stream.enabled` | boolean | `false` | Let people watch a running device's screen. On a gateway: pass each stream through from the worker. | Both | | `capabilities.stream.fps` | integer `1`–`60` | `30` | Highest frame rate. A still screen sends far fewer frames. | Worker | | `capabilities.stream.maxDimension` | integer `96`–`4096` | `720` | Longest side of the picture, in pixels. The setting that changes bandwidth most. | Worker | | `capabilities.stream.bitrateKbps` | integer `100`–`40000` | `1500` | Average bitrate in kilobits per second. | Worker | From b1aa159d9ffdbd63c60e3f1855961d00519650b7 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 12:56:36 +0200 Subject: [PATCH 19/20] test(e2e): a stream kept after the lease is still open at the end of the window (#909) --- e2e/slow-ios-stream.test.ts | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/e2e/slow-ios-stream.test.ts b/e2e/slow-ios-stream.test.ts index 5aad600db..94892d849 100644 --- a/e2e/slow-ios-stream.test.ts +++ b/e2e/slow-ios-stream.test.ts @@ -251,6 +251,14 @@ describe.skipIf(process.platform !== "darwin")( expect(kept.length, "the stream sent nothing for 10 s after the lease").toBeGreaterThan( 0, ); + // why: a stream that closed at the grant ends the read loop early with frames already + // buffered, so frames alone do not prove it kept going; the next read must still be open. + const stillOpen = await within( + reader.read(), + KEY_FRAME_BOUND, + "the stream's next read after the lease", + ); + expect(stillOpen.done, "the stream ended when the lease was granted").toBe(false); viewer.abort(); } finally { await env From 355ae8c90bcc6dec9bfc18366af9a02188bf7995 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 18:35:33 +0200 Subject: [PATCH 20/20] test(e2e): the stream lanes read the worker's instanceId, and the Android lane no longer reads an open stream up front (#909) --- e2e/helpers/device-stream.ts | 4 ++-- e2e/slow-android-stream.test.ts | 11 ++++------- 2 files changed, 6 insertions(+), 9 deletions(-) diff --git a/e2e/helpers/device-stream.ts b/e2e/helpers/device-stream.ts index 98ffd5cc4..c32ed1316 100644 --- a/e2e/helpers/device-stream.ts +++ b/e2e/helpers/device-stream.ts @@ -10,7 +10,7 @@ export async function deviceStreamUrl( deviceId: string, ): Promise { const instance = JSON.parse(await readFile(join(home, "instance.json"), "utf8")) as { - readonly id: string; + readonly instanceId: string; }; - return `${base}/v1/stream/workers/${encodeURIComponent(instance.id)}/devices/${encodeURIComponent(deviceId)}`; + return `${base}/v1/stream/workers/${encodeURIComponent(instance.instanceId)}/devices/${encodeURIComponent(deviceId)}`; } diff --git a/e2e/slow-android-stream.test.ts b/e2e/slow-android-stream.test.ts index 8d82ce068..ea11e7d13 100644 --- a/e2e/slow-android-stream.test.ts +++ b/e2e/slow-android-stream.test.ts @@ -95,13 +95,10 @@ describe.skipIf(!hasAndroidSdk)( headers: { authorization }, signal: viewer.signal, }); - expect( - response.status, - await response - .clone() - .text() - .catch(() => ""), - ).toBe(200); + // why: a 200 body is an endless stream, so its text is read only for a failure, never as + // an argument evaluated up front. + const detail = response.status === 200 ? "" : await response.text().catch(() => ""); + expect(response.status, detail).toBe(200); expect(response.headers.get("simlock-stream-codec")).toBe("h264"); if (response.body === null) throw new Error("the stream answered without a body");