feat(history): the worker's Events tab scrolls back through its history

e2e/console/workers.spec.ts

@@ -781,7 +781,7 @@ test.describe("the workers page", () => {
     }
   });
 
-  test("on a gateway the Events tab lists only that worker's events, on a single host all of them, and with none it says No events for this worker in the last hour.", async ({
+  test("on a gateway the Events tab lists only that worker's events, on a single host all of them, and scrolls back through that worker's history, shows its new events at the top, and with none says No events for this worker yet.", async ({
     page,
   }) => {
     const fleet = await startFleet([
@@ -814,12 +814,76 @@ test.describe("the workers page", () => {
       await expect(forB.getByText("worker.drain-started").first()).toBeVisible();
       await expect(forB.getByText("lease.granted")).toHaveCount(0);
 
+      const idA = idOf("worker-a");
+      const feed = page.getByRole("region", { name: "Event feed for this worker" });
+      const asked: string[] = [];
+      const answered: { readonly payload?: { readonly workerId?: string } }[] = [];
+      page.on("request", (request) => {
+        const url = new URL(request.url());
+        if (url.pathname === "/v1/events") asked.push(url.search);
+      });
+      page.on("response", (response) => {
+        if (new URL(response.url()).pathname !== "/v1/events") return;
+        void response.json().then((body: { events?: (typeof answered)[number][] }) => {
+          answered.push(...(body.events ?? []));
+        });
+      });
+      // why: a page of older events holds 100, so more than 100 of worker A's own are needed.
+      for (let round = 0; round < 25; round += 1) {
+        const granted = await fleet.gateway.cli([
+          "lease",
+          "--platform",
+          "ios",
+          "--device",
+          "iPhone 16",
+          "--os",
+          "18.4",
+          "--detach",
+          "--agent-id",
+          `history-${round}`,
+        ]);
+        expect(granted.code, granted.stderr).toBe(0);
+        const released = await fleet.gateway.cli([
+          "release",
+          (JSON.parse(granted.stdout) as { lease: { id: string } }).lease.id,
+        ]);
+        expect(released.code, released.stderr).toBe(0);
+      }
+      // why: taller than the feed's own height cap, so only a feed that fills the page reaches the bottom.
+      await page.setViewportSize({ height: 1_800, width: 1_280 });
+      await eventsOf(fleet.gateway, idA);
+      await expect(feed).toBeVisible();
+      const box = await feed.boundingBox();
+      expect((box?.y ?? 0) + (box?.height ?? 0)).toBeGreaterThan(1_800 - 120);
+      await expect(page.getByText("Start of the history (", { exact: false })).toHaveCount(0);
+      await expect
+        .poll(async () => {
+          await feed.evaluate((element) => element.scrollTo({ top: element.scrollHeight }));
+          return page.getByText("Start of the history (", { exact: false }).count();
+        })
+        .toBe(1);
+      expect(asked.some((query) => query.includes("before="))).toBe(true);
+      expect(asked.filter((query) => !query.includes(`worker=${encodeURIComponent(idA)}`))).toEqual(
+        [],
+      );
+      expect(answered.length).toBeGreaterThan(100);
+      expect(answered.filter((event) => event.payload?.workerId !== idA)).toEqual([]);
+
+      await feed.evaluate((element) => element.scrollTo({ top: 0 }));
+      const newest = await feed.getByRole("listitem").first().innerText();
+      expect((await fleet.gateway.cli(["worker", "undrain", idOf("worker-b")])).code).toBe(0);
+      await lease(fleet.gateway, "console-e2e-other", "iPhone 17");
+      await lease(fleet.gateway, "console-e2e-new");
+      await expect.poll(() => feed.getByRole("listitem").first().innerText()).not.toBe(newest);
+      await expect(feed.getByText("console-e2e-new").first()).toBeVisible();
+      await expect(feed.getByText("console-e2e-other")).toHaveCount(0);
+
       const replay = (url: URL) => url.pathname === "/v1/events";
       await page.route(replay, async (route) => {
         await route.fulfill({ json: { events: [] } });
       });
       await eventsOf(fleet.gateway, idOf("worker-a"));
-      await expect(page.getByText("No events for this worker in the last hour.")).toBeVisible();
+      await expect(page.getByText("No events for this worker yet.")).toBeVisible();
       await page.unroute(replay);
 
       const [only] = await workerList(host);
@@ -893,7 +957,7 @@ test.describe("the workers page", () => {
       await expect(
         page
           .getByRole("list", { name: "Events for this worker" })
-          .or(page.getByText("No events for this worker in the last hour.")),
+          .or(page.getByText("No events for this worker yet.")),
       ).toBeVisible();
 
       for (const tab of ["Events", "Host", "Warm pool", "Devices"]) {

e2e/gateway-event-relay.test.ts

@@ -5,6 +5,8 @@ import { join } from "node:path";
 
 import { describe, expect, it } from "vitest";
 
+import { NodeIpcTransport } from "../src/ports/index.js";
+import { SimlockWire } from "../src/simlock-client/wire.js";
 import { waitFor, withDaemon, withHttpDaemon } from "./helpers/index.js";
 import type { RecordedEvent } from "./helpers/events.js";
 
@@ -195,6 +197,81 @@ describe("gateway event relay", () => {
     await held.waitForExit(15_000);
   });
 
+  it("a gateway pages its events by worker through GET /v1/events, and a worker answers its own id only", async () => {
+    const { first, gateway, port, read, second, secret } = await gatewayWithTwoWorkers();
+    const holders = [
+      first.env.cliBackground(leaseArgs("by-worker-one-a")),
+      first.env.cliBackground(leaseArgs("by-worker-one-b")),
+      second.env.cliBackground(leaseArgs("by-worker-two")),
+    ];
+    for (const holder of holders) await holder.firstStdoutLine();
+    const refused = await withDaemon({
+      configOverrides: {
+        gateway: {
+          token: (
+            (await gateway.cli(["token", "create", "--role", "operator"])).json as {
+              secret: string;
+            }
+          ).secret,
+          url: `ws://127.0.0.1:${port}`,
+        },
+      },
+    });
+    expect(refused.home).toBeDefined();
+    let all: RecordedEvent[] = [];
+    await waitFor(
+      async () => {
+        all = await read("?limit=500");
+        return all.some((event) => event.event === "worker.rejected");
+      },
+      { label: "the gateway kept the refused dial's worker.rejected", timeout: 20_000 },
+    );
+    const claimed = all
+      .filter((event) => event.event === "worker.rejected")
+      .map((event) => (event.payload as { workerId: string }).workerId)[0];
+    const ofWorker = (id: string) =>
+      all.filter(
+        (event) =>
+          event.event !== "worker.rejected" &&
+          (event.payload as { workerId?: string }).workerId === id,
+      );
+    const expected = ofWorker(first.id);
+    expect(expected.length).toBeGreaterThanOrEqual(5);
+    expect(idsOf(ofWorker(second.id)).some((id) => idsOf(expected).includes(id))).toBe(false);
+
+    const newest = await read(`?worker=${first.id}&limit=100`);
+    expect(idsOf(newest)).toEqual(idsOf(expected));
+    const lastTwo = await read(`?worker=${first.id}&limit=2`);
+    expect(idsOf(lastTwo)).toEqual(idsOf(expected.slice(-2)));
+    const olderTwo = await read(`?worker=${first.id}&before=${lastTwo[0]?.id}&limit=2`);
+    expect(idsOf(olderTwo)).toEqual(idsOf(expected.slice(-4, -2)));
+    expect(idsOf(await read(`?worker=${first.id}&since=1h`))).toEqual(idsOf(expected));
+    expect(idsOf(await read(`?worker=${first.id}`))).toEqual(idsOf(expected));
+    expect(await read(`?worker=${claimed}&limit=100`)).toEqual([]);
+    expect(await read("?worker=nobody&limit=100")).toEqual([]);
+
+    const wire = new SimlockWire(await new NodeIpcTransport().connect(first.env.socketPath));
+    try {
+      const adminToken = (
+        await readFile(join(first.env.dataDirectory, "admin.token"), "utf8")
+      ).trim();
+      await wire.hello({ credential: adminToken, principal: "relay-by-worker" });
+      const own = (await wire.call("events.replay", {})) as RecordedEvent[];
+      expect(own.length).toBeGreaterThan(0);
+      expect(
+        idsOf((await wire.call("events.replay", { workerId: first.id })) as RecordedEvent[]),
+      ).toEqual(idsOf(own));
+      expect(await wire.call("events.replay", { workerId: second.id })).toEqual([]);
+    } finally {
+      await wire.close();
+    }
+    expect(secret).toBeDefined();
+    for (const holder of holders) {
+      holder.kill("SIGTERM");
+      await holder.waitForExit(15_000);
+    }
+  });
+
   it("simlock events --follow on a gateway prints an event that arrives during the replay once", async () => {
     const { gateway, worker } = await gatewayWithWorker();
     const gate = await gateReplay(gateway.socketPath);
@@ -243,6 +320,52 @@ describe("gateway event relay", () => {
   });
 });
 
+interface JoinedWorker {
+  readonly id: string;
+  readonly env: Awaited<ReturnType<typeof withDaemon>>;
+}
+
+async function gatewayWithTwoWorkers() {
+  const { env: gateway, port } = await withHttpDaemon((port) => ({
+    configOverrides: { http: { host: "127.0.0.1", port } },
+    role: "gateway",
+    driver: "none",
+  }));
+  const minted = await gateway.cli(["token", "create", "--role", "worker"]);
+  const { secret } = minted.json as { secret: string };
+  const join = async (label: string): Promise<JoinedWorker> => {
+    const env = await withDaemon({
+      configOverrides: { gateway: { label, token: secret, url: `ws://127.0.0.1:${port}` } },
+      driverScript: { ios: { knownModels: ["iPhone 17"], availableOsVersions: ["18.4"] } },
+    });
+    let id: string | undefined;
+    await waitFor(
+      async () => {
+        const listed = await gateway.cli(["worker", "list", "--json"]);
+        const workers = (listed.json as { workers: { id: string; label?: string }[] }).workers;
+        id = workers.find((worker) => worker.label === label)?.id;
+        return id !== undefined;
+      },
+      { label: `${label} joined the gateway`, timeout: 30_000 },
+    );
+    return { env, id: id as string };
+  };
+  const first = await join("worker-one");
+  const second = await join("worker-two");
+  const operator = await gateway.cli(["token", "create", "--role", "operator"]);
+  const authorization = `Bearer ${(operator.json as { secret: string }).secret}`;
+  const read = async (query: string): Promise<RecordedEvent[]> => {
+    const response = await fetch(`http://127.0.0.1:${String(port)}/v1/events${query}`, {
+      headers: { authorization },
+    });
+    expect(response.status, query).toBe(200);
+    return ((await response.json()) as { events: RecordedEvent[] }).events;
+  };
+  return { first, gateway, operator, port, read, second, secret };
+}
+
+const idsOf = (events: readonly RecordedEvent[]) => events.map((event) => event.id);
+
 /**
  * A unix-socket proxy in front of the gateway's daemon socket that holds the follower's
  * `events.replay` request until `release()`. Everything else, pushes included, passes through, so

src/contract/AGENTS.md

@@ -70,7 +70,7 @@ Three identities:
 
 - `SimlockErrorCode` is a closed union (`ErrorDetailsMap`). `ERROR_TABLE` gives each code
   `kind` (`transport` | `protocol` | `domain`), `cliExitCode` and `httpStatus`.
-- `HISTORY_NOT_KEPT` (exit 12, 422, `details.oldestTs`) and `HISTORY_UNAVAILABLE` (exit 1, 503) are `usage.get`'s and `events.replay`'s with `beforeId`; `includeLedger` adds `requests[]` (at most 100,000, newest; `ledgerTruncated` when cut); the window it takes is bounded by `usageWindowSchema` (integers to 8.64e15, `from < to`, 90 days at most) and `usageWindow` (`usage-window.ts`) parses the CLI's flags once for every surface. `events.replay` takes `beforeId` and `limit` (1 to 500), refused with `sinceTs`; `eventsPageQuery` (`events-page.ts`) parses the HTTP query once, with the bounds.
+- `HISTORY_NOT_KEPT` (exit 12, 422, `details.oldestTs`) and `HISTORY_UNAVAILABLE` (exit 1, 503) are `usage.get`'s and `events.replay`'s with `beforeId`; `includeLedger` adds `requests[]` (at most 100,000, newest; `ledgerTruncated` when cut); the window it takes is bounded by `usageWindowSchema` (integers to 8.64e15, `from < to`, 90 days at most) and `usageWindow` (`usage-window.ts`) parses the CLI's flags once for every surface. `events.replay` takes `beforeId` and `limit` (1 to 500), refused with `sinceTs`, and `workerId` (1 to 128 characters) with any of them; `eventsPageQuery` (`events-page.ts`) parses the HTTP query once, with the bounds.
 - `fromWireError` wraps an unknown code as `UNKNOWN_DAEMON_ERROR`. That code and `CANCELLED`
   are built client side only.
 - `CODES_WITH_DECLARED_DETAILS`: codes whose `details` are contract. The compiler keeps it

src/contract/events-page.ts

@@ -7,15 +7,27 @@ export interface EventsPageAsk {
   readonly before?: string | undefined;
   readonly limit?: string | undefined;
   readonly since?: string | undefined;
+  readonly worker?: string | undefined;
 }
 
 export type EventsPageAnswer =
-  | { readonly page: { readonly beforeId?: string; readonly limit?: number } }
+  | {
+      readonly page: {
+        readonly beforeId?: string;
+        readonly limit?: number;
+        readonly workerId?: string;
+      };
+    }
   | { readonly refused: string };
 
 const LIMIT_DIGITS = /^[1-9]\d{0,2}$/;
 
+export const EVENTS_WORKER_MAX = 128;
+
 export function eventsPageQuery(ask: EventsPageAsk): EventsPageAnswer {
+  if (ask.worker !== undefined && (ask.worker === "" || ask.worker.length > EVENTS_WORKER_MAX)) {
+    return { refused: "worker must be a worker id of 1 to 128 characters" };
+  }
   if (ask.before !== undefined && !EVENT_ID_PATTERN.test(ask.before)) {
     return { refused: "before must be an event id" };
   }
@@ -33,6 +45,7 @@ export function eventsPageQuery(ask: EventsPageAsk): EventsPageAnswer {
     page: {
       ...(ask.before === undefined ? {} : { beforeId: ask.before }),
       ...(limit === undefined ? {} : { limit }),
+      ...(ask.worker === undefined ? {} : { workerId: ask.worker }),
     },
   };
 }

src/contract/operations.test.ts

@@ -1728,10 +1728,16 @@ describe("usage.get input", () => {
 describe("events.replay input", () => {
   const parse = (input: unknown) => OPERATIONS["events.replay"].input.safeParse(input);
 
-  it("accepts beforeId and limit in bounds, and refuses limit 0, 501 and 1.5, a beforeId of evt_ or abc, and beforeId or limit with sinceTs", () => {
+  it("accepts beforeId and limit in bounds, and refuses limit 0, 501 and 1.5, a beforeId of evt_ or abc, and beforeId or limit with sinceTs; accepts a workerId of 1 to 128 characters with any of them, and refuses an empty one or 129", () => {
     expect(parse({ beforeId: "evt_abc", limit: 1 }).success).toBe(true);
     expect(parse({ limit: 500 }).success).toBe(true);
     expect(parse({ sinceTs: 5 }).success).toBe(true);
+    for (const workerId of ["w", "w".repeat(128)]) {
+      expect(parse({ workerId }).success).toBe(true);
+      expect(parse({ limit: 5, workerId }).success).toBe(true);
+      expect(parse({ sinceTs: 5, workerId }).success).toBe(true);
+    }
+    for (const workerId of ["", "w".repeat(129)]) expect(parse({ workerId }).success).toBe(false);
 
     for (const limit of [0, 501, 1.5]) expect(parse({ limit }).success).toBe(false);
     for (const beforeId of ["evt_", "abc"]) expect(parse({ beforeId }).success).toBe(false);

src/contract/operations.ts

@@ -12,6 +12,7 @@ import { z } from "zod";
 import { CONSOLE_URL_MAX_LENGTH } from "./console-url.js";
 import {
   EVENTS_PAGE_MAX,
+  EVENTS_WORKER_MAX,
   EVENTS_PAGE_WITH_SINCE_MESSAGE,
   eventsPageWithoutSince,
 } from "./events-page.js";
@@ -700,6 +701,7 @@ export const eventsReplay = defineOperation({
       beforeId: z.string().regex(EVENT_ID_PATTERN).optional(),
       limit: z.number().int().min(1).max(EVENTS_PAGE_MAX).optional(),
       sinceTs: z.number().optional(),
+      workerId: z.string().min(1).max(EVENTS_WORKER_MAX).optional(),
     })
     .refine(eventsPageWithoutSince, { message: EVENTS_PAGE_WITH_SINCE_MESSAGE }),
   output: z.array(eventEnvelopeSchema),

src/contract/protocol.ts

@@ -10,7 +10,7 @@ export interface ProtocolRange {
 const protocolRangeSchema = z.object({ min: z.number().int(), max: z.number().int() });
 
 /** Widen only once a second version is kept alive side by side with the first; a bump that leaves no shim behind moves both ends. */
-export const PROTOCOL_VERSION_RANGE: ProtocolRange = { min: 59, max: 59 };
+export const PROTOCOL_VERSION_RANGE: ProtocolRange = { min: 60, max: 60 };
 
 const LEGACY_DAEMON_PROTOCOL_VERSION = 2;
 

src/daemon-protocol/index.test.ts

@@ -9,7 +9,7 @@ import {
 
 describe("daemon protocol", () => {
   it("keeps the current protocol version and newline framing", () => {
-    expect(DAEMON_PROTOCOL_VERSION).toBe(59);
+    expect(DAEMON_PROTOCOL_VERSION).toBe(60);
     expect(serializeFrame({ id: 1, type: "hello" })).toBe('{"id":1,"type":"hello"}\n');
   });
 

src/daemon/dispatcher.ts

@@ -851,8 +851,12 @@ export class Dispatcher {
 
   #usageGet: Handler<"usage.get"> = (input) => usageAnswer(this.#usage.get(input));
 
-  #eventsReplay: Handler<"events.replay"> = (input) =>
-    input.beforeId === undefined && input.limit === undefined
+  #eventsReplay: Handler<"events.replay"> = (input) => {
+    // why: a worker's own events carry no workerId, so `worker` can only name this daemon or no one (ADR 0012 §1)
+    if (input.workerId !== undefined && input.workerId !== this.options.instanceId) {
+      return Promise.resolve([]);
+    }
+    return input.beforeId === undefined && input.limit === undefined
       ? this.options.eventHistory.replay(
           input.sinceTs === undefined ? {} : { sinceTs: input.sinceTs },
         )
@@ -862,6 +866,7 @@ export class Dispatcher {
             limit: input.limit ?? EVENTS_PAGE_DEFAULT,
           }),
         );
+  };
 
   #eventsSubscribe: Handler<"events.subscribe"> = (_input, session) => {
     const subscriptionId = session.manageEventSubscription(true);

src/gateway/dispatcher.ts

@@ -374,13 +374,15 @@ export class GatewayDispatcher {
 
   #eventsReplay: Handler<"events.replay"> = (input) =>
     input.beforeId === undefined && input.limit === undefined
-      ? this.options.eventHistory.replay(
-          input.sinceTs === undefined ? {} : { sinceTs: input.sinceTs },
-        )
+      ? this.options.eventHistory.replay({
+          ...(input.sinceTs === undefined ? {} : { sinceTs: input.sinceTs }),
+          ...(input.workerId === undefined ? {} : { workerId: input.workerId }),
+        })
       : eventsPageAnswer(
           this.options.eventHistory.replayPage({
             ...(input.beforeId === undefined ? {} : { beforeId: input.beforeId }),
             limit: input.limit ?? EVENTS_PAGE_DEFAULT,
+            ...(input.workerId === undefined ? {} : { workerId: input.workerId }),
           }),
         );