From 13fdc945c993f6ee01d14df19c84c2a9dc887d79 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 19:52:39 +0200 Subject: [PATCH 1/8] test: an operator pages a gateway's events by worker (#911) --- e2e/gateway-event-relay.test.ts | 120 ++++++++++++++++++++++++++++++++ ui/src/state/workers.ts | 2 +- 2 files changed, 121 insertions(+), 1 deletion(-) diff --git a/e2e/gateway-event-relay.test.ts b/e2e/gateway-event-relay.test.ts index 124e23afc..5f5db3052 100644 --- a/e2e/gateway-event-relay.test.ts +++ b/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,78 @@ 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 { + await wire.hello({ 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 +317,52 @@ describe("gateway event relay", () => { }); }); +interface JoinedWorker { + readonly id: string; + readonly env: Awaited>; +} + +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 => { + 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 => { + 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 diff --git a/ui/src/state/workers.ts b/ui/src/state/workers.ts index afc855fc1..caaa61c59 100644 --- a/ui/src/state/workers.ts +++ b/ui/src/state/workers.ts @@ -178,7 +178,7 @@ export function useDeviceDrawer( } { const page = useWorkerPage(workerId); const search = useSearch(); - const workers = useLiveResource("/v1/workers").data?.workers; + const workers = useRouteQuery("/v1/workers").data?.workers; return { // why: each choice replaces the entry rather than adding one, as the worker page's tabs do. chooseTab: (tab) => setSearch(writeDrawerTab(search, tab), { replace: true }), From 791ce2070624b8f3b25fdc60da911cf48fed4f2a Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 19:55:03 +0200 Subject: [PATCH 2/8] feat(history): GET /v1/events takes a worker filter served from an index (#911) --- e2e/gateway-event-relay.test.ts | 5 +- src/contract/events-page.ts | 15 +++++- src/contract/operations.ts | 2 + src/contract/protocol.ts | 2 +- src/daemon-protocol/index.test.ts | 2 +- src/daemon/dispatcher.ts | 9 +++- src/gateway/dispatcher.ts | 8 +-- src/history/database.ts | 81 ++++++++++++++++++++++++++----- src/history/index.ts | 55 ++++++++++++++++----- src/http/admin-declarations.ts | 3 +- src/http/app.ts | 4 +- 11 files changed, 151 insertions(+), 35 deletions(-) diff --git a/e2e/gateway-event-relay.test.ts b/e2e/gateway-event-relay.test.ts index 5f5db3052..138f3c1be 100644 --- a/e2e/gateway-event-relay.test.ts +++ b/e2e/gateway-event-relay.test.ts @@ -252,7 +252,10 @@ describe("gateway event relay", () => { const wire = new SimlockWire(await new NodeIpcTransport().connect(first.env.socketPath)); try { - await wire.hello({ principal: "relay-by-worker" }); + 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( diff --git a/src/contract/events-page.ts b/src/contract/events-page.ts index 378b4462d..67a1cf80b 100644 --- a/src/contract/events-page.ts +++ b/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 }), }, }; } diff --git a/src/contract/operations.ts b/src/contract/operations.ts index afeb98edd..4fab44876 100644 --- a/src/contract/operations.ts +++ b/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), diff --git a/src/contract/protocol.ts b/src/contract/protocol.ts index 4d3b6f979..443745791 100644 --- a/src/contract/protocol.ts +++ b/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; diff --git a/src/daemon-protocol/index.test.ts b/src/daemon-protocol/index.test.ts index 555d77724..5cddc44ef 100644 --- a/src/daemon-protocol/index.test.ts +++ b/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'); }); diff --git a/src/daemon/dispatcher.ts b/src/daemon/dispatcher.ts index 5b8b429ed..1f6229926 100644 --- a/src/daemon/dispatcher.ts +++ b/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); diff --git a/src/gateway/dispatcher.ts b/src/gateway/dispatcher.ts index 1d237edb0..f451a720e 100644 --- a/src/gateway/dispatcher.ts +++ b/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 }), }), ); diff --git a/src/history/database.ts b/src/history/database.ts index 50f8091b0..89262de24 100644 --- a/src/history/database.ts +++ b/src/history/database.ts @@ -7,16 +7,41 @@ export const BUSY_TIMEOUT_MS = 5_000; export const INSERT_EVENT_SQL = "INSERT OR IGNORE INTO events (id, timestamp, seq, size, envelope) VALUES (?, ?, ?, ?, ?)"; -const SELECT_SINCE_SQL = - "SELECT envelope FROM events WHERE timestamp > ? ORDER BY timestamp, seq, rowid"; +// why: a row of bad JSON must index as no worker; a bare json_extract on it throws "malformed JSON" and fails the index build +const ENVELOPE_FIELD = (path: string) => + `(CASE WHEN json_valid(envelope) THEN json_extract(envelope, '${path}') END)`; +const WORKER_ID_SQL = ENVELOPE_FIELD("$.payload.workerId"); + +// why: a refused dial's workerId is only what the connection claimed, so worker.rejected never belongs to a worker +const WORKER_FILTER_SQL = `${WORKER_ID_SQL} = ? AND ${ENVELOPE_FIELD("$.event")} <> 'worker.rejected'`; + +const WORKER_INDEX_SQL = `CREATE INDEX IF NOT EXISTS events_worker ON events (${WORKER_ID_SQL}, timestamp, seq)`; + +function filtered(workerId: string | undefined): string { + return workerId === undefined ? "" : ` AND ${WORKER_FILTER_SQL}`; +} + +function selectSinceSql(workerId: string | undefined): string { + return `SELECT envelope FROM events WHERE timestamp > ?${filtered(workerId)} ORDER BY timestamp, seq, rowid`; +} const SELECT_CURSOR_SQL = "SELECT timestamp, seq, rowid AS position FROM events WHERE id = ?"; -const SELECT_BEFORE_SQL = - "SELECT envelope FROM events WHERE (timestamp, seq, rowid) < (?, ?, ?) ORDER BY timestamp DESC, seq DESC, rowid DESC LIMIT ?"; +function selectBeforeSql(workerId: string | undefined): string { + return `SELECT envelope FROM events WHERE (timestamp, seq, rowid) < (?, ?, ?)${filtered(workerId)} ORDER BY timestamp DESC, seq DESC, rowid DESC LIMIT ?`; +} -const SELECT_NEWEST_SQL = - "SELECT envelope FROM events ORDER BY timestamp DESC, seq DESC, rowid DESC LIMIT ?"; +function selectNewestSql(workerId: string | undefined): string { + return `SELECT envelope FROM events ${workerId === undefined ? "" : `WHERE ${WORKER_FILTER_SQL} `}ORDER BY timestamp DESC, seq DESC, rowid DESC LIMIT ?`; +} + +function withWorker( + workerId: string | undefined, + before: readonly (string | number)[], + after: readonly number[] = [], +): (string | number)[] { + return [...before, ...(workerId === undefined ? [] : [workerId]), ...after]; +} const SELECT_OLDEST_SQL = "SELECT timestamp FROM events ORDER BY timestamp, seq, rowid LIMIT 1"; @@ -44,6 +69,11 @@ export function configureDatabase(database: SqliteDatabase): void { ); } +// why: run after the file has opened, never inside `configureDatabase`, whose throws count as damage +export function createWorkerIndex(database: SqliteDatabase): void { + database.exec(WORKER_INDEX_SQL); +} + export function inTransaction(database: SqliteDatabase, work: () => T): T { database.exec("BEGIN"); try { @@ -64,8 +94,14 @@ export function envelopeSize(json: string): number { return Buffer.byteLength(json, "utf8"); } -export function selectEventsSince(database: SqliteDatabase, sinceTs: number): EventEnvelope[] { - return parseEnvelopes(database.prepare(SELECT_SINCE_SQL).all(sinceTs)); +export function selectEventsSince( + database: SqliteDatabase, + sinceTs: number, + workerId?: string, +): EventEnvelope[] { + return parseEnvelopes( + database.prepare(selectSinceSql(workerId)).all(...withWorker(workerId, [sinceTs])), + ); } function parseEnvelopes(rows: readonly Record[]): EventEnvelope[] { @@ -81,17 +117,38 @@ export function selectEventsBefore( database: SqliteDatabase, beforeId: string, limit: number, + workerId?: string, ): EventEnvelope[] | undefined { const cursor = database.prepare(SELECT_CURSOR_SQL).get(beforeId); if (cursor === undefined) return undefined; const rows = database - .prepare(SELECT_BEFORE_SQL) - .all(Number(cursor.timestamp), Number(cursor.seq), Number(cursor.position), limit); + .prepare(selectBeforeSql(workerId)) + .all( + ...withWorker( + workerId, + [Number(cursor.timestamp), Number(cursor.seq), Number(cursor.position)], + [limit], + ), + ); return parseEnvelopes(rows).reverse(); } -export function selectNewestEvents(database: SqliteDatabase, limit: number): EventEnvelope[] { - return parseEnvelopes(database.prepare(SELECT_NEWEST_SQL).all(limit)).reverse(); +export function selectNewestEvents( + database: SqliteDatabase, + limit: number, + workerId?: string, +): EventEnvelope[] { + return parseEnvelopes( + database.prepare(selectNewestSql(workerId)).all(...withWorker(workerId, [], [limit])), + ).reverse(); +} + +export function isWorkerEvent(envelope: EventEnvelope, workerId: string | undefined): boolean { + if (workerId === undefined) return true; + return ( + envelope.event !== "worker.rejected" && + (envelope.payload as { workerId?: unknown } | null)?.workerId === workerId + ); } export function selectOldestTimestamp(database: SqliteDatabase): number | undefined { diff --git a/src/history/index.ts b/src/history/index.ts index e6adeeecf..55b522f69 100644 --- a/src/history/index.ts +++ b/src/history/index.ts @@ -15,8 +15,10 @@ import { import { BUSY_TIMEOUT_MS, configureDatabase, + createWorkerIndex, envelopeSize, INSERT_EVENT_SQL, + isWorkerEvent, selectEventsBefore, selectEventsSince, selectNewestEvents, @@ -67,10 +69,14 @@ export type EventsPageRead = | { readonly kind: "unavailable" }; export interface EventHistoryHandle { - replay(input?: { readonly sinceTs?: number }): Promise; + replay(input?: { + readonly sinceTs?: number; + readonly workerId?: string; + }): Promise; replayPage(input: { readonly beforeId?: string; readonly limit: number; + readonly workerId?: string; }): Promise; status(): EventHistoryStatus; start(): Promise; @@ -136,31 +142,40 @@ class EventHistory implements EventHistoryHandle { }; } - async replay(input: { readonly sinceTs?: number } = {}): Promise { + async replay( + input: { readonly sinceTs?: number; readonly workerId?: string } = {}, + ): Promise { const { bus, logger } = this.#options; - if (input.sinceTs === undefined) return bus.replay(); + const ofWorker = (events: EventEnvelope[]) => + events.filter((envelope) => isWorkerEvent(envelope, input.workerId)); + if (input.sinceTs === undefined) return ofWorker(bus.replay()); const database = this.#database; - if (database === undefined || !this.#readable) return bus.replay({ sinceTs: input.sinceTs }); + if (database === undefined || !this.#readable) { + return ofWorker(bus.replay({ sinceTs: input.sinceTs })); + } try { - return selectEventsSince(database, input.sinceTs); + return selectEventsSince(database, input.sinceTs, input.workerId); } catch (error: unknown) { logger.warn("History read failed; replaying from memory", { error: describe(error), path: this.#path, }); - return bus.replay({ sinceTs: input.sinceTs }); + return ofWorker(bus.replay({ sinceTs: input.sinceTs })); } } async replayPage(input: { readonly beforeId?: string; readonly limit: number; + readonly workerId?: string; }): Promise { const database = this.#database; - if (input.beforeId === undefined) return { events: this.#newest(input.limit), kind: "events" }; + if (input.beforeId === undefined) { + return { events: this.#newest(input.limit, input.workerId), kind: "events" }; + } if (database === undefined || !this.#readable || !this.#writing) return { kind: "unavailable" }; try { - const events = selectEventsBefore(database, input.beforeId, input.limit); + const events = selectEventsBefore(database, input.beforeId, input.limit, input.workerId); if (events !== undefined) return { events, kind: "events" }; const oldestTs = selectOldestTimestamp(database); return oldestTs === undefined @@ -175,11 +190,11 @@ class EventHistory implements EventHistoryHandle { } } - #newest(limit: number): EventEnvelope[] { + #newest(limit: number, workerId: string | undefined): EventEnvelope[] { const database = this.#database; if (database !== undefined && this.#readable && this.#writing) { try { - return selectNewestEvents(database, limit); + return selectNewestEvents(database, limit, workerId); } catch (error: unknown) { this.#options.logger.warn("History read failed; paging from memory", { error: describe(error), @@ -187,7 +202,10 @@ class EventHistory implements EventHistoryHandle { }); } } - return this.#options.bus.replay().slice(-limit); + return this.#options.bus + .replay() + .filter((envelope) => isWorkerEvent(envelope, workerId)) + .slice(-limit); } async start(): Promise { @@ -210,6 +228,7 @@ class EventHistory implements EventHistoryHandle { } // why: nothing up to the rebuild awaits, so no event can land between the ring's copy and the live writer this.#database = database; + this.#indexWorkers(database); this.#migrate(database); const rebuilding = this.#prepare(database); if (rebuilding === undefined) return; @@ -228,6 +247,20 @@ class EventHistory implements EventHistoryHandle { this.#arm(); } + #indexWorkers(database: SqliteDatabase): void { + try { + createWorkerIndex(database); + } catch (error: unknown) { + this.#options.logger.warn( + "The history's worker index could not be built; reads by worker scan the table", + { + error: describe(error), + path: this.#path, + }, + ); + } + } + #migrate(database: SqliteDatabase): void { try { for (const projection of this.#options.projections) migrateProjection(database, projection); diff --git a/src/http/admin-declarations.ts b/src/http/admin-declarations.ts index 823aac969..72ee37bd2 100644 --- a/src/http/admin-declarations.ts +++ b/src/http/admin-declarations.ts @@ -157,6 +157,7 @@ const eventsQuerySchema = z.object({ before: firstValue(z.string()).optional(), limit: firstValue(z.string()).optional(), since: firstValue(z.string()).optional(), + worker: firstValue(z.string()).optional(), }); export const eventsRoute = declareRoute({ @@ -166,7 +167,7 @@ export const eventsRoute = declareRoute({ summary: "Replay stored events, newest page first", role: "admin", notes: - "`since` is a duration such as 5m; `before` and `limit` page back through the history and cannot be used with `since`.", + "`since` is a duration such as 5m; `before` and `limit` page back through the history and cannot be used with `since`. `worker` keeps only the events of that worker (1 to 128 characters); a gateway never returns a `worker.rejected` for it, and a worker answers only its own id.", request: { query: eventsQuerySchema }, reply: z.object({ events: OPERATIONS["events.replay"].output }), errors: { 422: ["HISTORY_NOT_KEPT"], 503: ["HISTORY_UNAVAILABLE"] }, diff --git a/src/http/app.ts b/src/http/app.ts index 1a0c528d4..b38a8e1c7 100644 --- a/src/http/app.ts +++ b/src/http/app.ts @@ -708,8 +708,8 @@ export function createHttpApp(deps: HttpGatewayDeps): OpenAPIHono & HttpApp ); app.openapi({ ...eventsRoute, middleware: [agentAuth] }, async (c) => { - const { before, limit, since } = c.req.valid("query"); - const answer = eventsPageQuery({ before, limit, since }); + const { before, limit, since, worker } = c.req.valid("query"); + const answer = eventsPageQuery({ before, limit, since, worker }); if ("refused" in answer) throw badRequest(answer.refused); const sinceTs = since === undefined ? undefined : deps.clock.now() - parseDuration(since); const events = await deps.dispatch( From 8fe1bd7e5b5492f5b8f25b3479f43e83c4b5ebe8 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 20:06:52 +0200 Subject: [PATCH 3/8] test: GET /v1/events refuses a bad worker and passes a good one to events.replay (#911) --- src/http/app.test.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/http/app.test.ts b/src/http/app.test.ts index 470070b87..5b335be6a 100644 --- a/src/http/app.test.ts +++ b/src/http/app.test.ts @@ -2389,7 +2389,7 @@ describe("operator-only listing routes", () => { oldestTs: 1_234, }); }; - const notKept = await app.request("/v1/events?before=evt_abc&limit=50", { + const notKept = await app.request("/v1/events?before=evt_abc&limit=50&worker=w1", { headers: operatorAuth, }); dispatcher.handlers["events.replay"] = () => { @@ -2397,7 +2397,7 @@ describe("operator-only listing routes", () => { }; const unavailable = await app.request("/v1/events?before=evt_abc", { headers: operatorAuth }); - expect(asked).toEqual([{ beforeId: "evt_abc", limit: 50 }]); + expect(asked).toEqual([{ beforeId: "evt_abc", limit: 50, workerId: "w1" }]); expect(notKept.status).toBe(422); expect(await notKept.json()).toMatchObject({ error: { code: "HISTORY_NOT_KEPT", oldestTs: 1_234 }, @@ -2414,6 +2414,8 @@ describe("operator-only listing routes", () => { ["since=abc", "Invalid duration: abc"], ["before=evt_abc&since=1h", "before and since cannot be used together"], ["limit=50&since=1h", "limit and since cannot be used together"], + ["worker=", "worker must be a worker id of 1 to 128 characters"], + [`worker=${"w".repeat(129)}`, "worker must be a worker id of 1 to 128 characters"], ])("GET /v1/events?%s answers 400 BAD_REQUEST before dispatching", async (query, message) => { const { app, dispatcher } = buildHarness(); dispatcher.handlers["events.replay"] = () => { From 52baa4497045d096bd2144770416dab68452d588 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 20:07:49 +0200 Subject: [PATCH 4/8] test: the history reads a worker's events without the database or the index (#911) --- src/history/history.test.ts | 70 +++++++++++++++++++++++++++++++++++++ 1 file changed, 70 insertions(+) diff --git a/src/history/history.test.ts b/src/history/history.test.ts index f5f16b9b8..4b1ca7114 100644 --- a/src/history/history.test.ts +++ b/src/history/history.test.ts @@ -501,6 +501,76 @@ describe("createEventHistory", () => { expect(fromStopped).toEqual({ events: stoppedHeld, kind: "events" }); }); + it("reads a worker's events without the database or the index: from memory with limit counted after the filter, past a failed index build that sets nothing aside, and past a stored row of bad JSON", async () => { + const memory = await rig({ capacity: 10 }); + const connected = (workerId: string) => + memory.bus.emit("worker.connected", { workerId }, "gateway"); + const first = connected("w1"); + const rejected = memory.bus.emit( + "worker.rejected", + { reason: "forbidden", workerId: "w1" }, + "gateway", + ); + connected("w2"); + connected("w2"); + const unstarted = memory.history(); + const newestOfW1 = await unstarted.replayPage({ limit: 2, workerId: "w1" }); + const recent = await unstarted.replay({ workerId: "w1" }); + + const unindexed = await rig(); + const noIndex = new ProbeSqliteOpener({ + fault: (call) => { + if (call.kind === "exec" && /CREATE INDEX.*events_worker/is.test(call.sql)) { + throw new Error("disk full"); + } + }, + }); + const scanning = unindexed.history({ sqlite: noIndex }); + await scanning.start(); + const written = unindexed.bus.emit("worker.connected", { workerId: "w1" }, "gateway"); + unindexed.bus.emit("worker.connected", { workerId: "w2" }, "gateway"); + const scanned = await scanning.replayPage({ limit: 5, workerId: "w1" }); + const scanningStatus = scanning.status(); + await scanning.stop(); + + const damaged = await rig(); + const seeded = damaged.history(); + await seeded.start(); + const good = damaged.bus.emit("worker.connected", { workerId: "w1" }, "gateway"); + await seeded.stop(); + const file = await new NodeSqliteOpener().open(damaged.dbPath, { readOnly: false }); + file.exec("DROP INDEX events_worker"); + file + .prepare("INSERT INTO events (id, timestamp, seq, size, envelope) VALUES (?, ?, ?, ?, ?)") + .run("evt_bad", NOW, 99, 5, "{nope"); + file.close(); + const reopened = damaged.history(); + await reopened.start(); + const afterBadRow = await reopened.replayPage({ limit: 5, workerId: "w1" }); + const reopenedStatus = reopened.status(); + await reopened.stop(); + const indexes = await new NodeSqliteOpener().open(damaged.dbPath, { readOnly: true }); + const indexed = indexes + .prepare("SELECT name FROM sqlite_master WHERE type = 'index' AND name = 'events_worker'") + .all(); + indexes.close(); + + expect(newestOfW1).toEqual({ events: [first], kind: "events" }); + expect(recent).toEqual([first]); + expect(rejected.payload).toMatchObject({ workerId: "w1" }); + expect(levelRecords(unindexed, "warn")).toBe(1); + expect(unindexed.logs.records.find((record) => record.level === "warn")?.fields).toMatchObject({ + error: "disk full", + }); + expect(await damagedNames(unindexed.directory)).toEqual([]); + expect(scanningStatus).toMatchObject({ opened: true, writing: true }); + expect(scanned).toEqual({ events: [written], kind: "events" }); + expect(levelRecords(damaged, "warn")).toBe(0); + expect(reopenedStatus).toMatchObject({ opened: true, writing: true }); + expect(indexed).toEqual([{ name: "events_worker" }]); + expect(afterBadRow).toEqual({ events: [good], kind: "events" }); + }); + it("opens a new file with incremental auto_vacuum, WAL, synchronous NORMAL and a 5 s busy timeout", async () => { const rigged = await rig(); const sqlite = new ProbeSqliteOpener(); From c9fdbf6e5beef3379a938485c3b65510bb163df7 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 20:13:42 +0200 Subject: [PATCH 5/8] feat(ui): the worker's Events tab scrolls back through that worker's history (#911) --- e2e/console/workers.spec.ts | 66 +++++++++++++++++++++++++-- ui/src/data/event-feed.ts | 28 ++++++++++-- ui/src/ds/feed.css | 14 +++++- ui/src/model/events.ts | 6 +-- ui/src/screens/fleet/fleet.css | 30 ++---------- ui/src/screens/fleet/worker-page.tsx | 68 ++++++++++++++++++++-------- ui/src/state/workers.ts | 44 ++++++++++++++---- 7 files changed, 191 insertions(+), 65 deletions(-) diff --git a/e2e/console/workers.spec.ts b/e2e/console/workers.spec.ts index 038aae0a8..918ecf88f 100644 --- a/e2e/console/workers.spec.ts +++ b/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,72 @@ 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(); + await lease(fleet.gateway, "console-e2e-new"); + await expect.poll(() => feed.getByRole("listitem").first().innerText()).not.toBe(newest); + 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 +953,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"]) { diff --git a/ui/src/data/event-feed.ts b/ui/src/data/event-feed.ts index 5b441268b..1881747be 100644 --- a/ui/src/data/event-feed.ts +++ b/ui/src/data/event-feed.ts @@ -22,6 +22,13 @@ import { const RETRY_BACKOFF_MS = [1_000, 2_000, 4_000, 8_000] as const; const RETRY_INTERVAL_MS = 10_000; +// why: a feed of one worker's events asks the daemon for that worker alone in every read, and keeps of +// the stream only the events that belong to it. +export interface FeedScope { + readonly workerId: string; + readonly keeps: (event: ConsoleEvent) => boolean; +} + export class EventFeed { readonly #get: (path: ApiPath) => Promise; readonly #clock: Clock; @@ -41,13 +48,26 @@ export class EventFeed { #active = false; readonly #showsHistory: boolean; + readonly #scope: FeedScope | undefined; // why: only a screen that scrolls back through the history asks for it; the others read the last // hour and nothing older, so a quiet hour does not fill their lists with older events. - constructor(get: (path: ApiPath) => Promise, clock: Clock, showsHistory = false) { + constructor( + get: (path: ApiPath) => Promise, + clock: Clock, + showsHistory = false, + scope?: FeedScope, + ) { this.#get = get; this.#clock = clock; this.#showsHistory = showsHistory; + this.#scope = scope; + } + + #read(query: string): Promise { + const worker = + this.#scope === undefined ? "" : `&worker=${encodeURIComponent(this.#scope.workerId)}`; + return this.#get(`/v1/events?${query}${worker}`); } start(): () => void { @@ -61,7 +81,7 @@ export class EventFeed { readonly streamEvent = (message: StreamEvent): void => { if (!this.#active) return; const event = readEvent(message.data); - if (event !== undefined) this.#add([event]); + if (event !== undefined && this.#scope?.keeps(event) !== false) this.#add([event]); }; readonly streamOpened = (opened: StreamOpened): void => { @@ -97,7 +117,7 @@ export class EventFeed { this.#retryTimer = undefined; this.#loading = true; const opens = this.#opens; - this.#get(`/v1/events?since=${since}`) + this.#read(`since=${since}`) .then(readReplay) .then( (events) => { @@ -143,7 +163,7 @@ export class EventFeed { ? `limit=${PAGE}` : `before=${encodeURIComponent(oldest.id)}&limit=${PAGE}`; this.#loadingOlder = true; - this.#get(`/v1/events?${query}`) + this.#read(query) .then(readOlder) .then( (page) => { diff --git a/ui/src/ds/feed.css b/ui/src/ds/feed.css index fdacad98b..26ee138eb 100644 --- a/ui/src/ds/feed.css +++ b/ui/src/ds/feed.css @@ -11,14 +11,24 @@ /* As a page's last part the feed fills the page's height; the box's height comes from the page, * never from its rows, so it still scrolls inside itself. */ .page > .feed:last-child, -.page > :last-child > .feed:last-child { +.page > :last-child > .feed:last-child, +.page > .tabs:last-child > .tab-panel > .feed:last-child { + display: flex; + flex: 1; + flex-direction: column; +} + +/* Tabs that end a page, with a feed as their panel's last part, pass the page's height down to it. */ +.page > .tabs:last-child, +.page > .tabs:last-child > .tab-panel { display: flex; flex: 1; flex-direction: column; } .page > .feed:last-child .feed-box, -.page > :last-child > .feed:last-child .feed-box { +.page > :last-child > .feed:last-child .feed-box, +.page > .tabs:last-child > .tab-panel > .feed:last-child .feed-box { flex: 1 1 0; height: auto; min-height: 20rem; diff --git a/ui/src/model/events.ts b/ui/src/model/events.ts index 17c26a5ca..27a2f0d42 100644 --- a/ui/src/model/events.ts +++ b/ui/src/model/events.ts @@ -72,6 +72,7 @@ function localMinute(timestamp: number): string { export function historyEnd( state: HistoryState, held: readonly ConsoleEvent[], + none = "No events yet.", ): HistoryEnd | undefined { if (held.length >= MAX_EVENTS) { return { @@ -83,10 +84,7 @@ export function historyEnd( if (state === "start") { return { retry: false, - text: - oldest === undefined - ? "No events yet." - : `Start of the history (${localMinute(oldest.timestamp)})`, + text: oldest === undefined ? none : `Start of the history (${localMinute(oldest.timestamp)})`, }; } if (state === "gone") return { retry: false, text: "Older events are no longer kept." }; diff --git a/ui/src/screens/fleet/fleet.css b/ui/src/screens/fleet/fleet.css index 2db02ceb1..e51ed9567 100644 --- a/ui/src/screens/fleet/fleet.css +++ b/ui/src/screens/fleet/fleet.css @@ -139,31 +139,11 @@ padding: var(--space-4) var(--space-6); } -.worker-events { - display: flex; - flex-direction: column; - margin: 0; - padding: 0; - list-style: none; - border: 1px solid var(--border); - background: var(--surface); -} - -.worker-events li { - display: grid; - grid-template-columns: auto minmax(0, 1fr); - gap: var(--space-1) var(--space-3); - padding: var(--space-3) var(--space-6); - border-bottom: 1px solid var(--border); - font-size: var(--text-sm); -} - -.worker-events li:last-child { - border-bottom: 0; -} - -.worker-events li > :last-child { - grid-column: 2; +.worker-event-time { + margin-right: var(--space-2); + color: var(--text-muted); + font-size: var(--text-xs); + font-weight: 400; } .panel .facts + .table-card { diff --git a/ui/src/screens/fleet/worker-page.tsx b/ui/src/screens/fleet/worker-page.tsx index d9db8ad2c..7481da5b2 100644 --- a/ui/src/screens/fleet/worker-page.tsx +++ b/ui/src/screens/fleet/worker-page.tsx @@ -5,9 +5,11 @@ import { type Column, DataTable, Facts, + Feed, Grid, Id, Link, + ListRow, Loaded, Meter, Page, @@ -21,11 +23,11 @@ import { import { type CapabilityRow, type DeviceRow, - type EventRow, type FigureCard, type InstallRow, type PlatformCard, useWorkerEvents, + type WorkerEventsPage, useWorkerPage, useWorkerPageMenu, type WarmRow, @@ -92,7 +94,7 @@ export function WorkerPageBody({ page }: { readonly page: WorkerPageView }) { {page.tab === "devices" ? : null} {page.tab === "host" ? : null} {page.tab === "warm-pool" ? : null} - {page.tab === "events" ? : null} + {page.tab === "events" ? : null} ); @@ -437,26 +439,56 @@ function Installs({ rows }: { readonly rows: readonly InstallRow[] }) { ); } +// why: one row of the feed is as tall as the feed says, so the rows can be placed by arithmetic. +const EVENT_ROW_PX = 60; + function WorkerEvents({ id }: { readonly id: string }) { const view = useWorkerEvents(id); - return {(rows) => }; + return {(page) => }; } -function WorkerEventList({ rows }: { readonly rows: readonly EventRow[] }) { - if (rows.length === 0) { - return

No events for this worker in the last hour.

; - } +function WorkerEventFeed({ page }: { readonly page: WorkerEventsPage }) { + const { end, loadOlder, rows } = page; + const endRow = + end === undefined ? null : ( +

+ {end.text} + {end.retry ? ( + + ) : null} +

+ ); + if (rows.length === 0) return endRow ??

Loading…

; return ( -
    - {rows.map((row) => ( -
  • - - {row.name} - {row.summary} -
    • - ))} -
+ rows[index]?.id ?? String(index)} + newText={(count) => (count === 1 ? "1 new event" : `${count} new events`)} + onNearEnd={loadOlder} + footer={endRow} + renderRow={(index) => { + const row = rows[index]; + return row === undefined ? null : ( + + + {row.name} + + } + sub={row.summary} + /> + ); + }} + /> ); } diff --git a/ui/src/state/workers.ts b/ui/src/state/workers.ts index caaa61c59..733463f95 100644 --- a/ui/src/state/workers.ts +++ b/ui/src/state/workers.ts @@ -1,3 +1,7 @@ +import { useEffect, useState, useSyncExternalStore } from "react"; + +import { EventFeed } from "../data/event-feed"; +import { browserClock } from "../data/live/browser"; import type { MenuEntry } from "../ds"; import type { LoadView } from "../ds/loaded"; import type { Paging } from "../ds/pager"; @@ -26,13 +30,13 @@ import { import { type StreamOffer, streamOffer } from "../model/screen"; import { type WorkerList, workerPath } from "../model/workers"; import { type EventRow, eventRows } from "../model/event-list"; -import { eventsOfWorker } from "../model/events"; +import { eventsOfWorker, type HistoryEnd, historyEnd } from "../model/events"; import { useActions } from "./actions"; -import { useRecentEvents } from "./events"; -import { useDaemonMode, useRouteQuery, useNow } from "./live"; +import { useDaemonMode, useLiveEvents, useRouteQuery, useNow, useStreamOpened } from "./live"; import { usePaging } from "./paging"; import { loadView, mapView } from "./resource"; import { navigate, setSearch, usePath, useSearch } from "./router"; +import { useApi } from "./services"; export type { EventRow } from "../model/event-list"; export type { @@ -199,12 +203,34 @@ export function useLeaseUdid(leaseId: string): string | undefined { ).data?.lease.udid; } -// why: the events the console holds, an hour of them, that belong to this worker, newest first. -export function useWorkerEvents(workerId: string): LoadView { - const feed = useRecentEvents(); +export interface WorkerEventsPage { + readonly rows: readonly EventRow[]; + readonly loadOlder: () => void; + // why: the last row says why the list stops; none while more may still load. + readonly end: HistoryEnd | undefined; +} + +// why: the tab holds its own feed of this worker's events, read from the daemon for that worker +// alone and scrolled back through its history; the stream adds the worker's new ones. +export function useWorkerEvents(workerId: string): LoadView { + const api = useApi(); const mode = useDaemonMode(); - const workers = useRouteQuery("/v1/workers").data?.workers ?? []; - return mapView(loadView(feed), (held) => - eventRows(eventsOfWorker(held, workerId, mode), workers, "all"), + const [feed] = useState( + () => + new EventFeed((path) => api.getJson(path), browserClock, true, { + keeps: (event) => eventsOfWorker([event], workerId, mode).length > 0, + workerId, + }), ); + useLiveEvents(feed.streamEvent); + useStreamOpened(feed.streamOpened); + useEffect(() => feed.start(), [feed]); + const state = useSyncExternalStore(feed.subscribe, feed.snapshot); + const history = useSyncExternalStore(feed.subscribe, feed.history); + const workers = useRouteQuery("/v1/workers").data?.workers ?? []; + return mapView(loadView(state), (held) => ({ + end: historyEnd(history, held, "No events for this worker yet."), + loadOlder: feed.loadOlder, + rows: eventRows(held, workers, "all"), + })); } From dd0527b646a34587e4f790392cab62158054d4e4 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 20:15:27 +0200 Subject: [PATCH 6/8] docs: the events route takes worker, the console lists the worker page's four tabs (#911) --- src/contract/AGENTS.md | 2 +- src/history/AGENTS.md | 2 +- src/http/AGENTS.md | 2 +- src/http/openapi.test.ts | 2 +- ui/AGENTS.md | 2 +- ui/src/data/event-feed.ts | 2 +- website/src/content/docs/docs/reference/events.mdx | 3 ++- website/src/content/docs/docs/worker/console.mdx | 5 +++-- 8 files changed, 11 insertions(+), 9 deletions(-) diff --git a/src/contract/AGENTS.md b/src/contract/AGENTS.md index a482aba9d..bad8c8107 100644 --- a/src/contract/AGENTS.md +++ b/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 diff --git a/src/history/AGENTS.md b/src/history/AGENTS.md index cd1a749eb..bf39247ca 100644 --- a/src/history/AGENTS.md +++ b/src/history/AGENTS.md @@ -25,7 +25,7 @@ Flows: [startup](../../docs/internal/flows/startup.md). Catalog: [EVENTS.md](../ 3. **Catch up** — the ring's events are written in `seq` order, not time order (`INSERT OR IGNORE` by `id`; `index.ts` `#start`). After a set-aside only the events seen since the factory ran are. A projection whose check failed takes no step meanwhile. 4. **Write** — each event is one transaction: `BEGIN`, insert the row, run every projection's `apply` when the row is new, `COMMIT`. 5. **Prune** — after start and every hour: delete rows older than `eventLog.retention`, then the oldest 1 000 at a time while the stored size is over `eventLog.maxBytes`, then each projection's `prune` (rows older than `metrics.retention`), then `incremental_vacuum` and `wal_checkpoint(TRUNCATE)`. -6. **Read** — `replay({ sinceTs })` selects `timestamp > ?` ordered by timestamp, seq, rowid; a row whose id fails `EVENT_ID_PATTERN` is skipped. `openHistoryReader` does the same read-only for the CLI when no daemon answers; a missing file is empty, an unreadable one throws. `replayPage({ beforeId?, limit })` never throws: with `beforeId` it selects the `limit` rows before that row's `(timestamp, seq, rowid)` and returns them oldest first (`not-kept` with the oldest row's timestamp when the id is not stored and the table has rows, `unavailable` when the file is not readable or not writing or the read fails); without it, the newest `limit` rows, or the ring's newest when the file is not readable or not writing or the read fails. +6. **Read** — `replay({ sinceTs?, workerId? })` selects `timestamp > ?` ordered by timestamp, seq, rowid; a row whose id fails `EVENT_ID_PATTERN` is skipped. `openHistoryReader` does the same read-only for the CLI when no daemon answers; a missing file is empty, an unreadable one throws. `replayPage({ beforeId?, limit, workerId? })` never throws: with `beforeId` it selects the `limit` rows before that row's `(timestamp, seq, rowid)` and returns them oldest first (`not-kept` with the oldest row's timestamp when the id is not stored and the table has rows, `unavailable` when the file is not readable or not writing or the read fails); without it, the newest `limit` rows, or the ring's newest when the file is not readable or not writing or the read fails. 7. **Stop** — `stop()` cancels the timer and closes the file. A `start()` still awaiting closes what it opened and returns. ## State and timers diff --git a/src/http/AGENTS.md b/src/http/AGENTS.md index f9d64d1f3..3fd45295b 100644 --- a/src/http/AGENTS.md +++ b/src/http/AGENTS.md @@ -88,7 +88,7 @@ agent-device front and the stream (`/v1/stream`). opened by an identity with `expiresAt` (a proxy JWT) ends then (`sse.ts`, `app.ts`). - `GET /v1/events/stream` — dispatches `events.subscribe` (admin, readiness), then subscribes to the `EventBus` itself. -- `GET /v1/events` — reads `since`, `before` and `limit` through `eventsPageQuery` (contract), refused as 400 before dispatching `events.replay`; `HISTORY_NOT_KEPT` maps to 422 with `oldestTs`, `HISTORY_UNAVAILABLE` to 503 here too. +- `GET /v1/events` — reads `since`, `before`, `limit` and `worker` through `eventsPageQuery` (contract), refused as 400 before dispatching `events.replay`; `HISTORY_NOT_KEPT` maps to 422 with `oldestTs`, `HISTORY_UNAVAILABLE` to 503 here too. - `GET /v1/stats` — parses `since`/`from`/`to` with `usageWindow` (contract) and `ledger=1`, all refused as 400 before dispatching `usage.get` (the dispatcher answers an agent token 403, `HISTORY_NOT_KEPT` maps to 422 with `oldestTs`, `HISTORY_UNAVAILABLE` to 503). - `GET /v1/inbox` → `inbox.get` (admin): the items that need the operator, decided by the daemon (`src/contract/inbox.ts`); an agent token gets 403 from the dispatcher. - `POST /v1/update` (optional body `{version, force}`, strict) → `daemon.update`: 202 when the call began the update (`started`), 200 when it followed one or found nothing to do (`upToDate`). The route declares both (`alsoStatus`). The role check is the dispatcher's; on a gateway it updates the gateway itself, with no lease wait. diff --git a/src/http/openapi.test.ts b/src/http/openapi.test.ts index 0903f7aac..028ab310d 100644 --- a/src/http/openapi.test.ts +++ b/src/http/openapi.test.ts @@ -309,7 +309,7 @@ describe("buildOpenApiDocument", () => { }).toEqual({ components: ["platform"], componentRemove: ["dryRun"], - events: ["before", "limit", "since"], + events: ["before", "limit", "since", "worker"], }); expect( Object.keys(replySchema(operation("/v1/events", "get"), "200").properties ?? {}), diff --git a/ui/AGENTS.md b/ui/AGENTS.md index 585a1e613..cc85c8c79 100644 --- a/ui/AGENTS.md +++ b/ui/AGENTS.md @@ -126,7 +126,7 @@ newer ones, and `loadOlder` adds the 100 before the oldest held (`GET /v1/events none held). It reports how the history stands (`HistoryState`: more, start, gone, failed); `ds/feed.tsx` calls `onNearEnd` within 20 rows of the end and draws the end row. -The Inbox drawer and the lease drawer share one list of the last hour per tab (`data/recent-feed.ts`, held by `LiveProvider`): it starts at the first user, reads `since=1h` once, takes the stream and the reconnect gap, and is dropped 60 s after the last user closes (ADR 0044 §1). The worker's Events tab reads the same list until #911. +The Inbox drawer and the lease drawer share one list of the last hour per tab (`data/recent-feed.ts`, held by `LiveProvider`): it starts at the first user, reads `since=1h` once, takes the stream and the reconnect gap, and is dropped 60 s after the last user closes (ADR 0044 §1). The worker's Events tab holds its own `EventFeed` (`state/workers.ts` `useWorkerEvents`), given a `FeedScope`: every read carries `&worker=` and the stream adds only that worker's events, and it scrolls back like Activity. Lists page in the URL (`state/paging.ts`, `model/paging.ts`). diff --git a/ui/src/data/event-feed.ts b/ui/src/data/event-feed.ts index 1881747be..45620d045 100644 --- a/ui/src/data/event-feed.ts +++ b/ui/src/data/event-feed.ts @@ -24,7 +24,7 @@ const RETRY_INTERVAL_MS = 10_000; // why: a feed of one worker's events asks the daemon for that worker alone in every read, and keeps of // the stream only the events that belong to it. -export interface FeedScope { +interface FeedScope { readonly workerId: string; readonly keeps: (event: ConsoleEvent) => boolean; } diff --git a/website/src/content/docs/docs/reference/events.mdx b/website/src/content/docs/docs/reference/events.mdx index 1deafa061..0312acd21 100644 --- a/website/src/content/docs/docs/reference/events.mdx +++ b/website/src/content/docs/docs/reference/events.mdx @@ -42,9 +42,10 @@ With `--follow`, each line that arrives live wraps the event in an `event` field | `GET /v1/events` | `{"events": [...]}`: the recent events kept in memory. | | `GET /v1/events?since=2h` | `{"events": [...]}`: events from the history, same duration format as the CLI. | | `GET /v1/events?before=evt_…&limit=100` | `{"events": [...]}`: the `limit` events just older than that event, oldest first. Fewer than `limit` means the history ends there. | +| `GET /v1/events?worker=w1&limit=100` | `{"events": [...]}`: only the events of that worker, with `before`, `limit` or `since` as above. | | `GET /v1/events/stream` | A server-sent event stream. Each message's `event` is the event name and its `data` the event as JSON. | -`limit` is 1 to 500, and 100 when only `before` is given; `limit` alone returns the newest events. `before` and `limit` cannot be combined with `since`. A bad `before` or `limit` is `400 BAD_REQUEST`; a `before` the history no longer holds is `422 HISTORY_NOT_KEPT` with `oldestTs`; an unreadable history is `503 HISTORY_UNAVAILABLE`. +`limit` is 1 to 500, and 100 when only `before` is given; `limit` alone returns the newest events. `before` and `limit` cannot be combined with `since`. `worker` is a worker id of 1 to 128 characters, as `GET /v1/workers` gives it. A gateway returns the events whose payload has that `workerId`, never a `worker.rejected` (its id is only what a refused connection claimed), and an unknown worker gives `{"events": []}`. A worker answers every event for its own id and none for another. A bad `before`, `limit` or `worker` is `400 BAD_REQUEST`; a `before` the history no longer holds is `422 HISTORY_NOT_KEPT` with `oldestTs`; an unreadable history is `503 HISTORY_UNAVAILABLE`. ```sh curl -N -H "Authorization: Bearer $SIMLOCK_TOKEN" https://simlock-gw.example.com/v1/events/stream diff --git a/website/src/content/docs/docs/worker/console.mdx b/website/src/content/docs/docs/worker/console.mdx index 2aaf9389c..98dd9d5ef 100644 --- a/website/src/content/docs/docs/worker/console.mdx +++ b/website/src/content/docs/docs/worker/console.mdx @@ -123,11 +123,12 @@ The square shows the worst thing wrong with the worker, and always has a word be **Issues** shows only workers whose square is not green. **Busy** shows workers using three quarters of their device slots or more. -Select the worker to see three tabs: +Select the worker to see its CPU, memory and disk, each with a bar, how many leases it holds, and a square per device slot for iOS and Android (leased, ready, booting, quarantined or free). Under them are four tabs: -- **Overview**: CPU, memory and disk, each with a bar; how many leases it holds; a square per device slot for iOS and Android (leased, ready, booting, quarantined or free); and every device, with its state, model, runtime, whether it is physical or virtual, and the lease holding it. +- **Devices**: every device, with its state, model, runtime, whether it is physical or virtual, and the lease holding it. - **Host**: the worker's id, health, operating system, CPU architecture, the version of each platform tool, each optional module (such as screen streaming) as `available`, `off` or `unavailable` with a reason, and the catalog of models and runtimes it can lease. - **Warm pool**: whether the worker keeps devices warm, the running slots it holds back, and per target and learned kind its origin (`configured` or `learned`), how many devices it wants, how many are ready and how many are booting. A target it cannot fill shows the reason, such as `runtime-missing`. Under it, **Installs in progress** lists runtimes and system images being downloaded. See [Capacity and the warm pool](/simlock/docs/worker/capacity). +- **Events**: this worker's events, newest first, back to the start of its kept history or to the console's cap of events. Older events load as you scroll to the end of the list, and new ones appear at the top. With none, the tab says `No events for this worker yet.` ### Open a device From 36dd2d7973e1ff0b34def08165401405fdca38d2 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 21:00:35 +0200 Subject: [PATCH 7/8] test(history): bound a worker id at the operation, and show a second worker's events never reach the tab (#911) --- e2e/console/workers.spec.ts | 4 ++++ src/contract/operations.test.ts | 8 +++++++- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/e2e/console/workers.spec.ts b/e2e/console/workers.spec.ts index 918ecf88f..fedd6beb7 100644 --- a/e2e/console/workers.spec.ts +++ b/e2e/console/workers.spec.ts @@ -871,8 +871,12 @@ test.describe("the workers page", () => { 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) => { diff --git a/src/contract/operations.test.ts b/src/contract/operations.test.ts index 5a941e0e1..f15e29408 100644 --- a/src/contract/operations.test.ts +++ b/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); From f1f874be280370a26e33e95fcfa2482ed82d1cc1 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 21:13:53 +0200 Subject: [PATCH 8/8] fix(ui): a worker's Events tab filters by the daemon mode as it is when each event arrives (#911) --- ui/src/state/worker-events.test.tsx | 120 ++++++++++++++++++++++++++++ ui/src/state/workers.ts | 10 ++- 2 files changed, 127 insertions(+), 3 deletions(-) create mode 100644 ui/src/state/worker-events.test.tsx diff --git a/ui/src/state/worker-events.test.tsx b/ui/src/state/worker-events.test.tsx new file mode 100644 index 000000000..7e3aadfdc --- /dev/null +++ b/ui/src/state/worker-events.test.tsx @@ -0,0 +1,120 @@ +import { renderToStaticMarkup } from "react-dom/server"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +import type { ApiClient, ApiResponse } from "../data/api"; +import type { DaemonMode } from "../data/mode"; +import { connect, fakeClock, settle } from "../data/live/test-support"; +import { createQueries } from "./queries"; +import { type Live, LiveContext } from "./live"; +import { ConsoleProvider } from "./services"; +import { useWorkerEvents } from "./workers"; + +// why: a server render never runs effects and keeps no state between renders, so the effects of each +// render are collected for the test to run, and useState and useRef keep their slot across renders. +const rendered = vi.hoisted(() => ({ + effects: [] as (() => void | (() => void))[], + slot: 0, + slots: [] as unknown[], +})); + +vi.mock("react", async (original) => ({ + ...(await original()), + useEffect: (effect: () => void | (() => void)) => { + rendered.effects.push(effect); + }, + useRef: (initial: T) => { + const at = rendered.slot++; + rendered.slots[at] ??= { current: initial }; + return rendered.slots[at]; + }, + useState: (initial: T | (() => T)) => { + const at = rendered.slot++; + rendered.slots[at] ??= typeof initial === "function" ? (initial as () => T)() : initial; + return [rendered.slots[at], () => undefined]; + }, + useSyncExternalStore: (_subscribe: unknown, snapshot: () => unknown) => snapshot(), +})); + +vi.mock("../data/live/browser", async () => ({ + browserClock: (await import("../data/live/test-support")).fakeClock, + documentVisibility: { hidden: () => false, subscribe: () => () => undefined }, +})); + +afterEach(() => { + vi.useRealTimers(); + vi.unstubAllGlobals(); +}); + +function streamed(id: string, seq: number): string { + const data = { + event: "lease.renewed", + id, + payload: { leaseId: "l1" }, + seq, + timestamp: Date.now(), + }; + return `event: lease.renewed\ndata: ${JSON.stringify(data)}\n\n`; +} + +describe("a worker's Events tab", () => { + it("follows the daemon mode as it arrives: a single host's streamed event, which names no worker, shows once the mode is known to be a worker's", async () => { + vi.stubGlobal("window", { location: { pathname: "/workers/w1", search: "?tab=events" } }); + const { connection, daemon, visibility } = connect(); + daemon.reads = (path): Promise> => + Promise.resolve({ body: path.startsWith("/v1/events") ? { events: [] } : {}, date: null }); + const queries = createQueries({ clock: fakeClock, connection, visibility }); + queries.client.setQueryData(["workers"], { workers: [] }); + let mode: DaemonMode | undefined; + const live = { + connection, + mode: { + daemon: () => ({}), + dispose: () => undefined, + mode: () => mode, + subscribe: () => () => undefined, + }, + queries, + recent: {}, + } as unknown as Live; + const services = { + api: { + getJson: (path: string) => daemon.get(path as never).then((answer) => answer.body), + } as unknown as ApiClient, + confirm: () => Promise.resolve(), + fetch: () => Promise.reject(new Error("unused")), + notify: () => undefined, + preferences: {} as never, + session: {} as never, + }; + let held: ReturnType | undefined; + const Probe = () => { + held = useWorkerEvents("w1"); + return null; + }; + const render = (): void => { + rendered.slot = 0; + renderToStaticMarkup( + + + + + , + ); + }; + const rows = (): number => (held?.kind === "ready" ? held.data.rows.length : -1); + + await settle(); + render(); + for (const effect of rendered.effects) effect(); + rendered.effects = []; + await settle(); + mode = "worker"; + render(); + for (const effect of rendered.effects) effect(); + rendered.effects = []; + daemon.openStream()?.send(streamed("e1", 1)); + await settle(); + render(); + expect(rows()).toBe(1); + }); +}); diff --git a/ui/src/state/workers.ts b/ui/src/state/workers.ts index 733463f95..3e718ec79 100644 --- a/ui/src/state/workers.ts +++ b/ui/src/state/workers.ts @@ -1,4 +1,4 @@ -import { useEffect, useState, useSyncExternalStore } from "react"; +import { useEffect, useRef, useState, useSyncExternalStore } from "react"; import { EventFeed } from "../data/event-feed"; import { browserClock } from "../data/live/browser"; @@ -214,11 +214,15 @@ export interface WorkerEventsPage { // alone and scrolled back through its history; the stream adds the worker's new ones. export function useWorkerEvents(workerId: string): LoadView { const api = useApi(); - const mode = useDaemonMode(); + // why: the feed outlives the render that built it, and the mode may arrive after the tab mounts, + // so the filter reads the mode as it is when each event comes. + const daemonMode = useDaemonMode(); + const mode = useRef(daemonMode); + mode.current = daemonMode; const [feed] = useState( () => new EventFeed((path) => api.getJson(path), browserClock, true, { - keeps: (event) => eventsOfWorker([event], workerId, mode).length > 0, + keeps: (event) => eventsOfWorker([event], workerId, mode.current).length > 0, workerId, }), );