feat(stream): stream any running device by worker and device id

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) |

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<string> {
+  const instance = JSON.parse(await readFile(join(home, "instance.json"), "utf8")) as {
+    readonly instanceId: string;
+  };
+  return `${base}/v1/stream/workers/${encodeURIComponent(instance.instanceId)}/devices/${encodeURIComponent(deviceId)}`;
+}

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.

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<string, string>;
   readonly lease: { readonly id: string };
 }
@@ -186,8 +192,8 @@ async function readFrames(
   return frames;
 }
 
-async function streamOf(daemon: Rig, leaseId: string): Promise<Response> {
-  const response = await fetch(`${daemon.base}/v1/stream/leases/${encodeURIComponent(leaseId)}`, {
+async function streamOf(daemon: Rig, deviceId: string): Promise<Response> {
+  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,

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,25 +84,21 @@ 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,
-            },
-          );
-          expect(
-            response.status,
-            await response
-              .clone()
-              .text()
-              .catch(() => ""),
-          ).toBe(200);
+          const response = await fetch(streamUrl, {
+            headers: { authorization },
+            signal: viewer.signal,
+          });
+          // 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");
 
@@ -153,10 +149,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 +202,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);

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,

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<string, string>;
   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.

e2e/slow-ios-stream.test.ts

@@ -6,6 +6,7 @@ import { describe, expect, it } from "vitest";
 
 import {
   DEVICE_READY_MS,
+  deviceStreamUrl,
   REPO_ROOT,
   SCREEN_TAGS as TAG,
   readScreenFrames as readFrames,
@@ -16,6 +17,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<T>(work: Promise<T>, bound: number, what: string): Promise<T> {
@@ -70,12 +73,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 +171,103 @@ 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 () => {
+              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 !== "";
+            },
+            {
+              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,
+          );
+          // 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
+            .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 },

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,