diff --git a/gateway/src/api.ts b/gateway/src/api.ts new file mode 100644 index 0000000..8ba7467 --- /dev/null +++ b/gateway/src/api.ts @@ -0,0 +1,257 @@ +// The watch's REST API, as described in docs/api.md and typed in contract.ts. + +import type { BunRequest } from "bun"; + +import type { + ErrorResponse, + NewThreadRequest, + NewThreadResponse, + PairRequest, + PairResponse, + ProjectsResponse, + PushRegistration, + RespondRequest, + SendMessageRequest, +} from "./contract.ts"; +import type { Db, DeviceRow } from "./db.ts"; +import type { Hub } from "./hub.ts"; +import { HostError, type T3Host } from "./t3/host.ts"; +import { RpcError } from "./t3/rpc.ts"; +import type { ApprovalDecision, ModelSelection, ThreadShell } from "./t3/types.ts"; +import { buildFeed, buildThreadDetail, hostSummary, type SummaryCache } from "./views.ts"; + +export interface ApiDeps { + db: Db; + hub: Hub; + cache: SummaryCache; + log?: (message: string) => void; +} + +/** Wrong pairing codes allowed before every open code is voided. */ +const PAIR_ATTEMPTS = 10; +const PAIR_WINDOW_MS = 15 * 60_000; +const DECISIONS: ReadonlySet = new Set([ + "accept", + "acceptForSession", + "acceptAlways", + "decline", + "cancel", +]); + +class HttpError extends Error { + constructor( + readonly status: number, + message: string, + ) { + super(message); + } +} + +const json = (body: unknown, status = 200) => Response.json(body, { status }); +const accepted = () => new Response(null, { status: 202 }); + +export function createRoutes({ db, hub, cache, log = () => {} }: ApiDeps) { + let pairFailures: number[] = []; + + /** Wraps a handler with device auth and error mapping. */ + const authed = +

(handler: (req: BunRequest

, device: DeviceRow) => Promise | Response) => + (req: BunRequest

) => + handle(async () => { + const token = req.headers.get("authorization")?.match(/^Bearer (.+)$/)?.[1]; + const device = token ? db.devices.authenticate(token) : null; + if (!device) throw new HttpError(401, "This watch is not paired."); + return handler(req, device); + }); + + async function handle(run: () => Promise | Response): Promise { + try { + return await run(); + } catch (error) { + if (error instanceof HttpError) return json({ error: error.message } satisfies ErrorResponse, error.status); + if (error instanceof HostError) { + return json({ error: error.message } satisfies ErrorResponse, error.status === 503 ? 503 : 502); + } + if (error instanceof RpcError) return json({ error: error.message } satisfies ErrorResponse, 502); + log(`unhandled: ${error instanceof Error ? error.stack : String(error)}`); + return json({ error: "Something went wrong in the gateway." } satisfies ErrorResponse, 500); + } + } + + const findHost = (hostId: string): T3Host => { + const host = hub.get(hostId); + if (!host) throw new HttpError(404, "Unknown host."); + return host; + }; + + const findThread = (hostId: string, threadId: string): { host: T3Host; thread: ThreadShell } => { + const host = findHost(hostId); + const thread = host.threads.get(threadId); + if (!thread) throw new HttpError(404, "Unknown thread."); + return { host, thread }; + }; + + const resolveModel = (host: T3Host, projectId: string): ModelSelection | null => + host.projects.get(projectId)?.defaultModelSelection ?? host.serverConfig?.settings?.defaultModelSelection ?? null; + + return { + "/v1/health": { + GET: () => json({ ok: true, hosts: hub.list().map(hostSummary) }), + }, + + "/v1/pair": { + POST: (req: BunRequest<"/v1/pair">) => + handle(async () => { + const body = await readBody(req); + const code = String(body.code ?? "").trim(); + const name = String(body.deviceName ?? "").trim() || "Apple Watch"; + pairFailures = pairFailures.filter((at) => Date.now() - at < PAIR_WINDOW_MS); + if (pairFailures.length >= PAIR_ATTEMPTS) throw new HttpError(429, "Too many attempts. Make a new code."); + if (!/^\d{6}$/.test(code) || !db.pairing.redeem(code)) { + pairFailures.push(Date.now()); + if (pairFailures.length >= PAIR_ATTEMPTS) db.pairing.voidAll(); + throw new HttpError(400, "That code is wrong or has expired."); + } + const deviceToken = `wc_${Buffer.from(crypto.getRandomValues(new Uint8Array(32))).toString("base64url")}`; + const deviceId = db.devices.create(name, deviceToken); + log(`paired device ${name} (${deviceId})`); + return json({ deviceId, deviceToken } satisfies PairResponse); + }), + }, + + "/v1/device/push": { + PUT: authed<"/v1/device/push">(async (req, device) => { + const body = await readBody(req); + if (!/^[0-9a-fA-F]{32,200}$/.test(body.apnsToken ?? "")) throw new HttpError(400, "apnsToken must be hex."); + if (body.apnsEnvironment !== "sandbox" && body.apnsEnvironment !== "production") { + throw new HttpError(400, "apnsEnvironment must be sandbox or production."); + } + db.devices.setPush(device.id, body.apnsToken.toLowerCase(), body.apnsEnvironment); + return new Response(null, { status: 204 }); + }), + }, + + "/v1/feed": { + GET: authed(async () => json(await buildFeed(hub.list(), cache))), + }, + + "/v1/projects": { + GET: authed(() => + json({ + hosts: hub.list().map((host) => ({ + ...hostSummary(host), + projects: [...host.projects.values()] + .sort((a, b) => a.title.localeCompare(b.title)) + .map((project) => ({ id: project.id, title: project.title, model: resolveModel(host, project.id)?.model ?? null })), + })), + } satisfies ProjectsResponse), + ), + }, + + "/v1/threads": { + POST: authed<"/v1/threads">(async (req) => { + const body = await readBody(req); + const text = String(body.text ?? "").trim(); + if (!text) throw new HttpError(400, "Say what the thread should do."); + const host = findHost(String(body.hostId)); + if (!host.projects.has(String(body.projectId))) throw new HttpError(404, "Unknown project."); + const modelSelection = resolveModel(host, body.projectId); + if (!modelSelection) throw new HttpError(400, "That project has no default model. Set one in T3 Code."); + const result = await host.launchThread({ + commandId: crypto.randomUUID(), + creationSource: "mobile", + projectId: body.projectId, + title: titleFrom(text), + generateTitle: true, + modelSelection, + runtimeMode: host.serverConfig?.settings?.defaultRuntimeMode ?? "full-access", + interactionMode: "default", + workspaceStrategy: { type: "root" }, + initialMessage: { messageId: crypto.randomUUID(), text, attachments: [] }, + }); + return json({ hostId: host.id, threadId: result.threadId } satisfies NewThreadResponse, 201); + }), + }, + + "/v1/threads/:hostId/:threadId": { + GET: authed<"/v1/threads/:hostId/:threadId">(async (req) => { + const { host, thread } = findThread(req.params.hostId, req.params.threadId); + return json(await buildThreadDetail(host, thread, cache)); + }), + }, + + "/v1/threads/:hostId/:threadId/messages": { + POST: authed<"/v1/threads/:hostId/:threadId/messages">(async (req) => { + const { host, thread } = findThread(req.params.hostId, req.params.threadId); + const text = String((await readBody(req)).text ?? "").trim(); + if (!text) throw new HttpError(400, "The message is empty."); + await host.dispatch({ + type: "message.dispatch", + commandId: crypto.randomUUID(), + threadId: thread.id, + messageId: crypto.randomUUID(), + text, + attachments: [], + createdBy: "user", + creationSource: "mobile", + dispatchMode: { type: "queue_after_active" }, + }); + return accepted(); + }), + }, + + "/v1/threads/:hostId/:threadId/respond": { + POST: authed<"/v1/threads/:hostId/:threadId/respond">(async (req) => { + const { host, thread } = findThread(req.params.hostId, req.params.threadId); + const body = await readBody(req); + if (thread.pendingRuntimeRequest?.id !== body.requestId) { + throw new HttpError(409, "That request was already answered."); + } + const base = { type: "runtime-request.respond", commandId: crypto.randomUUID(), threadId: thread.id, requestId: body.requestId } as const; + if ("decision" in body) { + if (!DECISIONS.has(body.decision)) throw new HttpError(400, "Unknown decision."); + await host.dispatch({ ...base, decision: body.decision }); + } else if ("answers" in body && body.answers && typeof body.answers === "object") { + await host.dispatch({ ...base, answers: body.answers }); + } else { + throw new HttpError(400, "Send a decision or answers."); + } + cache.forget(host, thread.id); + return accepted(); + }), + }, + + "/v1/threads/:hostId/:threadId/stop": { + POST: authed<"/v1/threads/:hostId/:threadId/stop">(async (req) => { + const { host, thread } = findThread(req.params.hostId, req.params.threadId); + if (host.supports("threadStop")) { + await host.dispatch({ type: "thread.stop", commandId: crypto.randomUUID(), threadId: thread.id }); + } else if (thread.activeRunId) { + await host.dispatch({ + type: "run.interrupt", + commandId: crypto.randomUUID(), + threadId: thread.id, + runId: thread.activeRunId, + }); + } else { + throw new HttpError(409, "Nothing is running."); + } + return accepted(); + }), + }, + }; +} + +async function readBody(req: Request): Promise & T> { + try { + return (await req.json()) as Partial & T; + } catch { + throw new HttpError(400, "The body must be JSON."); + } +} + +/** A thread title from the first line of the prompt. T3 replaces it with a generated one. */ +function titleFrom(text: string): string { + const line = text.split("\n")[0]?.trim() ?? text; + return line.length > 80 ? `${line.slice(0, 79)}…` : line; +} diff --git a/gateway/src/apns.ts b/gateway/src/apns.ts new file mode 100644 index 0000000..0871ff4 --- /dev/null +++ b/gateway/src/apns.ts @@ -0,0 +1,114 @@ +// Sends push notifications through APNs with token-based auth: an ES256 JWT signed by the .p8 key +// from the Apple Developer portal, over HTTP/2. No dependencies. + +import { createPrivateKey, sign, type KeyObject } from "node:crypto"; +import { connect, type ClientHttp2Session } from "node:http2"; + +export interface ApnsConfig { + keyId: string; + teamId: string; + /** The .p8 key's PEM text. */ + key: string; + /** The app's bundle ID. */ + topic: string; +} + +export interface ApnsNotification { + deviceToken: string; + environment: "sandbox" | "production"; + payload: object; + /** Replaces an earlier notification with the same id on the device. */ + collapseId?: string; +} + +/** Device tokens APNs reported as no longer valid. */ +export type ApnsResult = { ok: true } | { ok: false; status: number; reason: string; tokenInvalid: boolean }; + +const HOSTS = { + sandbox: "https://api.sandbox.push.apple.com", + production: "https://api.push.apple.com", +} as const; + +// Apple rejects tokens older than an hour and throttles ones refreshed more often than every 20 minutes. +const JWT_LIFETIME_MS = 40 * 60_000; + +export class Apns { + private readonly key: KeyObject; + private jwt: { value: string; issuedAt: number } | null = null; + private readonly sessions = new Map(); + + constructor(private readonly config: ApnsConfig) { + this.key = createPrivateKey(config.key); + } + + async send(notification: ApnsNotification): Promise { + const body = JSON.stringify(notification.payload); + const session = this.session(HOSTS[notification.environment]); + return new Promise((resolve) => { + const request = session.request({ + ":method": "POST", + ":path": `/3/device/${notification.deviceToken}`, + authorization: `bearer ${this.token()}`, + "apns-topic": this.config.topic, + "apns-push-type": "alert", + "apns-priority": "10", + ...(notification.collapseId ? { "apns-collapse-id": notification.collapseId } : {}), + "content-type": "application/json", + }); + let status = 0; + let response = ""; + request.on("response", (headers) => { + status = Number(headers[":status"]); + }); + request.setEncoding("utf8"); + request.on("data", (chunk: string) => (response += chunk)); + request.on("end", () => { + if (status === 200) return resolve({ ok: true }); + const reason = safeReason(response); + resolve({ + ok: false, + status, + reason, + tokenInvalid: status === 410 || reason === "BadDeviceToken" || reason === "DeviceTokenNotForTopic", + }); + }); + request.on("error", (error) => resolve({ ok: false, status: 0, reason: error.message, tokenInvalid: false })); + request.end(body); + }); + } + + close() { + for (const session of this.sessions.values()) session.close(); + this.sessions.clear(); + } + + private session(origin: string): ClientHttp2Session { + const existing = this.sessions.get(origin); + if (existing && !existing.closed && !existing.destroyed) return existing; + const session = connect(origin); + session.on("error", () => this.sessions.delete(origin)); + session.on("close", () => this.sessions.delete(origin)); + this.sessions.set(origin, session); + return session; + } + + private token(): string { + if (this.jwt && Date.now() - this.jwt.issuedAt < JWT_LIFETIME_MS) return this.jwt.value; + const issuedAt = Date.now(); + const header = base64url(JSON.stringify({ alg: "ES256", kid: this.config.keyId })); + const claims = base64url(JSON.stringify({ iss: this.config.teamId, iat: Math.floor(issuedAt / 1000) })); + const signature = sign("sha256", Buffer.from(`${header}.${claims}`), { key: this.key, dsaEncoding: "ieee-p1363" }); + this.jwt = { value: `${header}.${claims}.${signature.toString("base64url")}`, issuedAt }; + return this.jwt.value; + } +} + +const base64url = (text: string) => Buffer.from(text).toString("base64url"); + +function safeReason(body: string): string { + try { + return (JSON.parse(body) as { reason?: string }).reason ?? body; + } catch { + return body; + } +} diff --git a/gateway/src/cli.ts b/gateway/src/cli.ts new file mode 100755 index 0000000..f20f031 --- /dev/null +++ b/gateway/src/cli.ts @@ -0,0 +1,118 @@ +#!/usr/bin/env bun +// Operator commands. They edit the same SQLite file the running gateway reads, so changes apply +// within a few seconds. In the container: `docker exec -it wristcode-gateway wristcode `. + +import { loadConfig } from "./config.ts"; +import { openDb } from "./db.ts"; +import { ORCHESTRATION_PROTOCOL_VERSION, type EnvironmentDescriptor } from "./t3/types.ts"; + +const USAGE = `wristcode + + host add [--label ] + Connect a T3 Code server. Make the credential on that host with: + t3 auth pairing create --scope orchestration:read --scope orchestration:operate --label "WristCode gateway" + Run it again with a fresh credential to renew the gateway's token. + host list + host remove + pair + Print a 6 digit code to enter on the watch. Good for 5 minutes. + device list + device revoke `; + +const PAIRING_CODE_TTL_MS = 5 * 60_000; + +const db = openDb(loadConfig().dbPath); +const [group, command, ...rest] = process.argv.slice(2); + +try { + await run(); +} catch (error) { + console.error(error instanceof Error ? error.message : String(error)); + process.exitCode = 1; +} finally { + db.close(); +} + +async function run() { + switch (`${group} ${command ?? ""}`.trim()) { + case "host add": + return hostAdd(rest); + case "host list": + for (const host of db.hosts.list()) { + console.log(`${host.label}\t${host.url}\t${host.id}\ttoken expires ${host.token_expires_at ?? "unknown"}`); + } + return; + case "host remove": + if (!rest[0]) throw new Error(USAGE); + console.log(db.hosts.remove(rest[0]) ? "Removed." : "No such host."); + return; + case "pair": { + const code = String(crypto.getRandomValues(new Uint32Array(1))[0]! % 1_000_000).padStart(6, "0"); + db.pairing.create(code, PAIRING_CODE_TTL_MS); + console.log(`Enter this code on the watch within 5 minutes: ${code}`); + return; + } + case "device list": + for (const device of db.devices.list()) { + const state = device.revoked_at ? "revoked" : device.apns_token ? `push ${device.apns_env}` : "no push"; + console.log(`${device.name}\t${device.id}\t${state}\tlast seen ${device.last_seen_at ?? "never"}`); + } + return; + case "device revoke": + if (!rest[0]) throw new Error(USAGE); + console.log(`Revoked ${db.devices.revoke(rest[0])} device(s).`); + return; + default: + console.log(USAGE); + } +} + +/** Exchanges a T3 pairing credential for a bearer session and saves the host. */ +async function hostAdd(args: string[]) { + const labelFlag = args.indexOf("--label"); + const label = labelFlag === -1 ? undefined : args[labelFlag + 1]; + const [rawUrl, credential] = args.filter((_, index) => labelFlag === -1 || (index !== labelFlag && index !== labelFlag + 1)); + if (!rawUrl || !credential) throw new Error(USAGE); + const url = new URL(rawUrl).origin; + + const descriptor = await getJson(`${url}/.well-known/t3/environment`); + if ((descriptor.orchestrationProtocolVersion ?? 1) !== ORCHESTRATION_PROTOCOL_VERSION) { + throw new Error(`${url} speaks orchestration protocol ${descriptor.orchestrationProtocolVersion ?? 1}, this gateway speaks ${ORCHESTRATION_PROTOCOL_VERSION}.`); + } + + const response = await fetch(`${url}/oauth/token`, { + method: "POST", + headers: { "content-type": "application/x-www-form-urlencoded" }, + body: new URLSearchParams({ + grant_type: "urn:ietf:params:oauth:grant-type:token-exchange", + subject_token: credential, + subject_token_type: "urn:t3:params:oauth:token-type:environment-bootstrap", + requested_token_type: "urn:ietf:params:oauth:token-type:access_token", + scope: "orchestration:read orchestration:operate", + client_label: "WristCode gateway", + client_device_type: "bot", + }), + }); + if (!response.ok) throw new Error(`${url} refused the pairing credential (${response.status}): ${await response.text()}`); + const token = (await response.json()) as { access_token: string; expires_in?: number; scope?: string }; + const scopes = token.scope?.split(" ") ?? []; + if (!scopes.includes("orchestration:read") || !scopes.includes("orchestration:operate")) { + throw new Error(`The credential only granted "${token.scope}". Make one with orchestration:read and orchestration:operate.`); + } + + db.hosts.upsert({ + id: descriptor.environmentId, + label: label ?? descriptor.label, + url, + environment_id: descriptor.environmentId, + token: token.access_token, + token_expires_at: token.expires_in ? new Date(Date.now() + token.expires_in * 1000).toISOString() : null, + }); + console.log(`Added ${label ?? descriptor.label} (T3 ${descriptor.serverVersion}). The gateway connects within 15 seconds.`); +} + +async function getJson(url: string): Promise { + const response = await fetch(url); + if (!response.ok) throw new Error(`${url} returned ${response.status}`); + return response.json() as Promise; +} diff --git a/gateway/src/config.ts b/gateway/src/config.ts new file mode 100644 index 0000000..dbc6ace --- /dev/null +++ b/gateway/src/config.ts @@ -0,0 +1,26 @@ +// Settings from the environment. Only the database path is needed to run. Without the APNs +// settings the gateway still serves the watch, it just logs pushes instead of sending them. + +import { readFileSync } from "node:fs"; + +import type { ApnsConfig } from "./apns.ts"; + +export interface Config { + port: number; + dbPath: string; + apns: ApnsConfig | null; +} + +export function loadConfig(env = process.env): Config { + const keyId = env.APNS_KEY_ID; + const teamId = env.APNS_TEAM_ID; + const key = env.APNS_KEY ?? (env.APNS_KEY_PATH ? readFileSync(env.APNS_KEY_PATH, "utf8") : undefined); + return { + port: Number(env.PORT ?? 8080), + dbPath: env.WRISTCODE_DB ?? "data/wristcode.sqlite", + apns: + keyId && teamId && key + ? { keyId, teamId, key: key.replace(/\\n/g, "\n"), topic: env.APNS_TOPIC ?? "org.gbrown.wristcode" } + : null, + }; +} diff --git a/gateway/src/hub.ts b/gateway/src/hub.ts new file mode 100644 index 0000000..a0eb8d9 --- /dev/null +++ b/gateway/src/hub.ts @@ -0,0 +1,57 @@ +// The set of T3 hosts, kept in step with the `hosts` table. The CLI edits the table and the running +// gateway picks the change up on its next sync, so adding or renewing a host needs no restart. + +import type { Db, HostRow } from "./db.ts"; +import { T3Host, type HostOptions } from "./t3/host.ts"; + +export class Hub { + private readonly hosts = new Map(); + + constructor( + private readonly db: Db, + private readonly options: HostOptions, + ) {} + + /** Starts new hosts, restarts ones whose URL or token changed, and stops removed ones. */ + sync() { + const rows = this.db.hosts.list(); + const seen = new Set(); + for (const row of rows) { + seen.add(row.id); + const current = this.hosts.get(row.id); + if (current && current.row.url === row.url && current.row.token === row.token && current.row.label === row.label) { + current.row = row; + continue; + } + current?.host.stop(); + const host = new T3Host( + { id: row.id, label: row.label, url: row.url, token: row.token, environmentId: row.environment_id }, + this.options, + ); + this.hosts.set(row.id, { host, row }); + host.start(); + } + for (const [id, { host }] of this.hosts) { + if (seen.has(id)) continue; + host.stop(); + this.hosts.delete(id); + } + } + + list(): T3Host[] { + return [...this.hosts.values()].map(({ host }) => host); + } + + get(id: string): T3Host | undefined { + return this.hosts.get(id)?.host; + } + + rows(): HostRow[] { + return [...this.hosts.values()].map(({ row }) => row); + } + + stop() { + for (const { host } of this.hosts.values()) host.stop(); + this.hosts.clear(); + } +} diff --git a/gateway/src/main.ts b/gateway/src/main.ts new file mode 100644 index 0000000..76000b5 --- /dev/null +++ b/gateway/src/main.ts @@ -0,0 +1,80 @@ +// Starts the gateway: connects to every T3 host in the database, serves the watch API, and pushes +// notifications. + +import { mkdirSync } from "node:fs"; +import { dirname } from "node:path"; + +import { Apns } from "./apns.ts"; +import { createRoutes } from "./api.ts"; +import { loadConfig } from "./config.ts"; +import { openDb } from "./db.ts"; +import { Hub } from "./hub.ts"; +import { Notifier, type SendPush } from "./notify.ts"; +import { SummaryCache } from "./views.ts"; + +const HOST_SYNC_MS = 15_000; +const TOKEN_CHECK_MS = 6 * 60 * 60_000; + +const log = (message: string) => console.log(`${new Date().toISOString()} ${message}`); + +const config = loadConfig(); +mkdirSync(dirname(config.dbPath), { recursive: true }); +const db = openDb(config.dbPath); +const cache = new SummaryCache(log); +const apns = config.apns ? new Apns(config.apns) : null; +if (!apns) log("APNs is not configured. Pushes will only be logged."); + +const sendPush: SendPush = async (push) => { + const devices = db.devices.active().filter((device) => device.apns_token && device.apns_env); + if (!apns || devices.length === 0) { + log(`push (not sent): [${push.category}] ${push.title}: ${push.body}`); + return; + } + const payload = { + aps: { + alert: { title: push.title, ...(push.subtitle ? { subtitle: push.subtitle } : {}), body: push.body }, + sound: "default", + category: push.category, + "thread-id": push.data.threadId || push.data.hostId, + }, + ...push.data, + }; + await Promise.all( + devices.map(async (device) => { + const result = await apns.send({ + deviceToken: device.apns_token!, + environment: device.apns_env!, + payload, + collapseId: push.collapseId, + }); + if (result.ok) return; + log(`APNs refused a push to ${device.name}: ${result.status} ${result.reason}`); + if (result.tokenInvalid) db.devices.clearPush(device.apns_token!); + }), + ); +}; + +const notifier = new Notifier(sendPush, cache, log); +const hub = new Hub(db, { onThreadChange: notifier.onThreadChange, onStatus: notifier.onHostStatus, log }); +hub.sync(); +setInterval(() => hub.sync(), HOST_SYNC_MS); +notifier.checkTokenExpiry(hub.rows()); +setInterval(() => notifier.checkTokenExpiry(hub.rows()), TOKEN_CHECK_MS); + +const server = Bun.serve({ + port: config.port, + routes: createRoutes({ db, hub, cache, log }), + fetch: () => Response.json({ error: "Not found." }, { status: 404 }), +}); +log(`WristCode gateway listening on :${server.port} with ${hub.list().length} host(s)`); + +for (const signal of ["SIGINT", "SIGTERM"] as const) { + process.on(signal, () => { + log(`${signal}: shutting down`); + void server.stop(); + hub.stop(); + apns?.close(); + db.close(); + process.exit(0); + }); +} diff --git a/gateway/src/notify.ts b/gateway/src/notify.ts new file mode 100644 index 0000000..e6060c1 --- /dev/null +++ b/gateway/src/notify.ts @@ -0,0 +1,133 @@ +// Decides when to push. A thread notifies when its phase becomes one the user cares about +// (waiting for them, finished, failed), and again only after it moves on. A new approval on a +// thread already waiting counts as a change, so each request notifies once. Nothing notifies while +// a host loads its first snapshot, so a restart never replays old news. + +import { threadPhase, type Phase } from "./awareness.ts"; +import type { PushCategory, PushPayload } from "./contract.ts"; +import type { HostRow } from "./db.ts"; +import type { T3Host, ThreadChange } from "./t3/host.ts"; +import { trimText, type SummaryCache, type ThreadSummary } from "./views.ts"; + +export interface Push { + title: string; + subtitle?: string; + body: string; + category: PushCategory; + data: PushPayload; + /** Replaces an earlier push about the same thing on the watch. */ + collapseId: string; +} + +export type SendPush = (push: Push) => Promise; + +const NOTIFYING: ReadonlySet = new Set(["waiting_for_approval", "waiting_for_input", "completed", "failed"]); +const BODY_CHARS = 180; +const SUMMARY_WAIT_MS = 3_000; +const TOKEN_WARNING_MS = 7 * 24 * 60 * 60_000; + +export class Notifier { + /** The last phase key seen per thread, `phase:requestId`. */ + private readonly keys = new Map(); + /** Hosts already warned about, so each warning goes out once a day at most. */ + private readonly warned = new Map(); + + constructor( + private readonly send: SendPush, + private readonly cache: SummaryCache, + private readonly log: (message: string) => void = () => {}, + ) {} + + onThreadChange = ({ host, previous, next, initial }: ThreadChange) => { + const id = `${host.id}:${(next ?? previous)?.id}`; + if (!next) { + this.keys.delete(id); + return; + } + const phase = threadPhase(next); + const key = `${phase}:${next.pendingRuntimeRequest?.id ?? ""}`; + const before = this.keys.get(id); + this.keys.set(id, key); + if (initial || phase === null || !NOTIFYING.has(phase) || key === before) return; + void this.notify(host, next, phase).catch((error: unknown) => + this.log(`push for ${id} failed: ${error instanceof Error ? error.message : String(error)}`), + ); + }; + + /** Warns once when a host starts rejecting the gateway's token. */ + onHostStatus = (host: T3Host) => { + if (host.status !== "auth_invalid") return; + this.warnHost(host.id, host.label, `T3 rejected the gateway's token. Pair ${host.label} again.`); + }; + + /** Warns about tokens that expire within a week. Run it daily. */ + checkTokenExpiry(rows: HostRow[], now = Date.now()) { + for (const row of rows) { + if (!row.token_expires_at) continue; + const left = Date.parse(row.token_expires_at) - now; + if (left > TOKEN_WARNING_MS) continue; + const days = Math.max(0, Math.ceil(left / (24 * 60 * 60_000))); + this.warnHost(row.id, row.label, `The gateway's T3 token expires in ${days} day${days === 1 ? "" : "s"}. Pair ${row.label} again.`); + } + } + + private warnHost(hostId: string, label: string, body: string) { + const last = this.warned.get(hostId) ?? 0; + if (Date.now() - last < 24 * 60 * 60_000) return; + this.warned.set(hostId, Date.now()); + void this.send({ + title: label, + body, + category: "host", + data: { hostId, threadId: "" }, + collapseId: `host:${hostId}`, + }).catch(() => {}); + } + + private async notify(host: T3Host, thread: NonNullable, phase: Phase) { + const summary = await Promise.race([ + this.cache.get(host, thread).catch(() => undefined), + Bun.sleep(SUMMARY_WAIT_MS).then(() => undefined), + ]); + const project = host.projects.get(thread.projectId)?.title; + const requestId = thread.pendingRuntimeRequest?.id; + await this.send({ + title: thread.title, + subtitle: project ? `${host.label} · ${project}` : host.label, + body: body(phase, thread.lastError ?? null, summary), + category: category(phase), + data: { hostId: host.id, threadId: thread.id, ...(requestId ? { requestId } : {}) }, + collapseId: `${host.id}:${thread.id}`.slice(0, 64), + }); + } +} + +function category(phase: Phase): PushCategory { + switch (phase) { + case "waiting_for_approval": + return "approval"; + case "waiting_for_input": + return "input"; + case "failed": + return "failed"; + default: + return "finished"; + } +} + +function body(phase: Phase, lastError: string | null, summary: ThreadSummary | undefined): string { + switch (phase) { + case "waiting_for_approval": { + const prompt = summary?.request?.prompt; + return prompt ? `Wants to run: ${trimText(prompt, BODY_CHARS)}` : "Needs your approval."; + } + case "waiting_for_input": { + const question = summary?.request?.questions[0]?.question; + return question ? trimText(question, BODY_CHARS) : "Has a question for you."; + } + case "failed": + return lastError ? trimText(lastError, BODY_CHARS) : "The run failed."; + default: + return summary?.lastAssistantMessage ? trimText(summary.lastAssistantMessage, BODY_CHARS) : "Finished."; + } +} diff --git a/gateway/src/t3/types.ts b/gateway/src/t3/types.ts index 2afc15e..d59f995 100644 --- a/gateway/src/t3/types.ts +++ b/gateway/src/t3/types.ts @@ -155,7 +155,7 @@ export interface EnvironmentDescriptor { export interface ServerConfig { environment?: { capabilities?: Record }; - settings?: { defaultModelSelection?: ModelSelection | null }; + settings?: { defaultModelSelection?: ModelSelection | null; defaultRuntimeMode?: string }; } /** Commands sent through `orchestration.dispatchCommand`. */ diff --git a/gateway/src/views.ts b/gateway/src/views.ts new file mode 100644 index 0000000..37887c1 --- /dev/null +++ b/gateway/src/views.ts @@ -0,0 +1,208 @@ +// Turns T3 state into the shapes in contract.ts. Thread shells carry status but not text, so the +// last message and the pending request's details come from each thread's bounded snapshot. Those +// are large (hundreds of KB), so they are reduced to a small summary and cached per thread until +// the thread's `updatedAt` moves. + +import { threadPhase, type Phase } from "./awareness.ts"; +import type { Feed, FeedItem, HostSummary, Message, PendingRequest, ThreadDetail } from "./contract.ts"; +import type { T3Host } from "./t3/host.ts"; +import { + isApprovalRequest, + isUserInputRequest, + type BoundedThread, + type ThreadShell, +} from "./t3/types.ts"; + +const LAST_MESSAGE_CHARS = 280; +const MESSAGE_CHARS = 2000; +const MESSAGES_PER_THREAD = 20; +const RECENT_WINDOW_MS = 24 * 60 * 60_000; +const RECENT_LIMIT = 10; +/** How long a feed request waits for summaries it does not have yet. */ +const SUMMARY_WAIT_MS = 2_000; + +export interface ThreadSummary { + messages: Message[]; + lastAssistantMessage: string | null; + request: PendingRequest | null; +} + +/** Trims text for a small screen: collapses whitespace and cuts at a word near `max`. */ +export function trimText(text: string, max: number): string { + const flat = text.replace(/\s+/g, " ").trim(); + if (flat.length <= max) return flat; + const cut = flat.slice(0, max); + const space = cut.lastIndexOf(" "); + return `${(space > max * 0.6 ? cut.slice(0, space) : cut).trimEnd()}…`; +} + +/** Reduces a bounded thread snapshot to what the watch shows. */ +export function summarize(bounded: BoundedThread, thread: ThreadShell): ThreadSummary { + const { messages, turnItems } = bounded.projection; + const visible = messages.filter( + (message) => message.text.trim().length > 0 && (message.role === "user" || message.role === "assistant"), + ); + const lastAssistant = visible.findLast((message) => message.role === "assistant"); + return { + messages: visible.slice(-MESSAGES_PER_THREAD).map((message) => ({ + id: message.id, + role: message.role as Message["role"], + text: message.text.length > MESSAGE_CHARS ? `${message.text.slice(0, MESSAGE_CHARS)}…` : message.text, + createdAt: message.createdAt, + })), + lastAssistantMessage: lastAssistant ? trimText(lastAssistant.text, LAST_MESSAGE_CHARS) : null, + request: pendingRequest(thread, turnItems), + }; +} + +function pendingRequest(thread: ThreadShell, turnItems: BoundedThread["projection"]["turnItems"]): PendingRequest | null { + const pending = thread.pendingRuntimeRequest; + if (!pending || pending.kind === "auth_refresh") return null; + const base = { id: pending.id, kind: pending.kind, prompt: null, reason: null, options: [], questions: [] }; + const item = turnItems.findLast( + (candidate) => (isApprovalRequest(candidate) || isUserInputRequest(candidate)) && candidate.requestId === pending.id, + ); + if (item && isApprovalRequest(item)) { + const options = item.options?.length + ? item.options.map(({ decision, label }) => ({ decision, label })) + : [ + { decision: "accept" as const, label: "Allow" }, + { decision: "decline" as const, label: "Deny" }, + ]; + return { ...base, prompt: item.prompt ?? item.title ?? null, reason: item.reason ?? null, options }; + } + if (item && isUserInputRequest(item)) { + return { + ...base, + questions: item.questions.map((question) => ({ + id: question.id, + header: question.header, + question: question.question, + options: question.options.map((option) => ({ label: option.label, value: option.value ?? option.label })), + allowCustomAnswer: question.allowCustomAnswer ?? false, + })), + }; + } + // The turn item may not be in the bounded window yet. Offer the plain choices for approvals. + return pending.kind === "user_input" + ? base + : { + ...base, + options: [ + { decision: "accept", label: "Allow" }, + { decision: "decline", label: "Deny" }, + ], + }; +} + +/** Per-thread summaries, refetched when a thread's `updatedAt` changes. */ +export class SummaryCache { + private readonly entries = new Map(); + private readonly inflight = new Map>(); + + constructor(private readonly log: (message: string) => void = () => {}) {} + + /** The cached summary, if it matches the thread's current version. */ + peek(host: T3Host, thread: ThreadShell): ThreadSummary | undefined { + const entry = this.entries.get(key(host, thread.id)); + return entry?.updatedAt === thread.updatedAt ? entry.summary : undefined; + } + + get(host: T3Host, thread: ThreadShell): Promise { + const cached = this.peek(host, thread); + if (cached) return Promise.resolve(cached); + const k = `${key(host, thread.id)}@${thread.updatedAt}`; + const running = this.inflight.get(k); + if (running) return running; + const promise = host + .bounded(thread.id) + .then((bounded) => { + const summary = summarize(bounded, thread); + this.entries.set(key(host, thread.id), { updatedAt: thread.updatedAt, summary }); + return summary; + }) + .finally(() => this.inflight.delete(k)); + this.inflight.set(k, promise); + return promise; + } + + forget(host: T3Host, threadId: string) { + this.entries.delete(key(host, threadId)); + } +} + +const key = (host: T3Host, threadId: string) => `${host.id}:${threadId}`; + +export const hostSummary = (host: T3Host): HostSummary => ({ + id: host.id, + label: host.label, + status: host.status, + error: host.lastError, +}); + +function feedItem(host: T3Host, thread: ThreadShell, phase: Phase, summary: ThreadSummary | undefined): FeedItem { + return { + hostId: host.id, + hostLabel: host.label, + threadId: thread.id, + projectId: thread.projectId, + projectTitle: host.projects.get(thread.projectId)?.title ?? "", + title: thread.title, + phase, + updatedAt: thread.updatedAt, + lastMessage: summary?.lastAssistantMessage ?? null, + request: summary?.request ?? null, + }; +} + +/** Builds the feed across every host, waiting briefly for summaries it has not read yet. */ +export async function buildFeed(hosts: T3Host[], cache: SummaryCache, now = Date.now()): Promise { + const candidates: Array<{ host: T3Host; thread: ThreadShell; phase: Phase }> = []; + for (const host of hosts) { + for (const thread of host.threads.values()) { + const phase = threadPhase(thread); + if (phase === null || thread.archivedAt || thread.deletedAt) continue; + if ((phase === "completed" || phase === "failed") && now - Date.parse(thread.updatedAt) > RECENT_WINDOW_MS) continue; + candidates.push({ host, thread, phase }); + } + } + + const byNewest = (a: { thread: ThreadShell }, b: { thread: ThreadShell }) => + b.thread.updatedAt.localeCompare(a.thread.updatedAt); + const needsYou = candidates + .filter(({ phase }) => phase === "waiting_for_approval" || phase === "waiting_for_input") + .sort((a, b) => + (a.thread.pendingRuntimeRequest?.createdAt ?? "").localeCompare(b.thread.pendingRuntimeRequest?.createdAt ?? ""), + ); + const working = candidates.filter(({ phase }) => phase === "starting" || phase === "running").sort(byNewest); + const recent = candidates + .filter(({ phase }) => phase === "completed" || phase === "failed") + .sort(byNewest) + .slice(0, RECENT_LIMIT); + + const shown = [...needsYou, ...working, ...recent]; + const missing = shown + .filter(({ host, thread }) => !cache.peek(host, thread)) + .map(({ host, thread }) => cache.get(host, thread).catch(() => undefined)); + if (missing.length > 0) await Promise.race([Promise.all(missing), Bun.sleep(SUMMARY_WAIT_MS)]); + + const toItem = ({ host, thread, phase }: (typeof candidates)[number]) => + feedItem(host, thread, phase, cache.peek(host, thread)); + return { + needsYou: needsYou.map(toItem), + working: working.map(toItem), + recent: recent.map(toItem), + hosts: hosts.map(hostSummary), + }; +} + +/** One thread with its recent messages. Fetches the summary if it is not cached. */ +export async function buildThreadDetail(host: T3Host, thread: ThreadShell, cache: SummaryCache): Promise { + const summary = await cache.get(host, thread); + const phase = threadPhase(thread) ?? (thread.status === "failed" ? "failed" : "completed"); + return { + item: feedItem(host, thread, phase, summary), + messages: summary.messages, + canStop: thread.activeRunId !== null, + }; +} diff --git a/gateway/test/notify.test.ts b/gateway/test/notify.test.ts new file mode 100644 index 0000000..43235f4 --- /dev/null +++ b/gateway/test/notify.test.ts @@ -0,0 +1,109 @@ +import { describe, expect, it } from "bun:test"; + +import { Notifier, type Push } from "../src/notify.ts"; +import type { T3Host } from "../src/t3/host.ts"; +import type { ThreadShell } from "../src/t3/types.ts"; +import { SummaryCache } from "../src/views.ts"; + +const host = { + id: "host-1", + label: "Mac mini", + projects: new Map([["project-1", { id: "project-1", title: "wristcode" }]]), + bounded: () => Promise.reject(new Error("offline")), +} as unknown as T3Host; + +const shell = (overrides: Partial = {}): ThreadShell => ({ + id: "thread-1", + projectId: "project-1", + title: "Watch app scaffold", + createdAt: "2026-11-01T09:00:00.000Z", + updatedAt: "2026-11-01T09:00:00.000Z", + status: "running", + activeRunId: "run-1", + pendingRuntimeRequest: null, + modelSelection: { instanceId: "claudeAgent", model: "claude-opus-5-5" }, + lineage: { relationshipToParent: null }, + ...overrides, +}); + +const request = (id: string) => ({ id, kind: "command", createdAt: "2026-11-01T09:00:00.000Z" }); + +function setup() { + const pushes: Push[] = []; + const notifier = new Notifier(async (push) => void pushes.push(push), new SummaryCache()); + let previous: ThreadShell | undefined; + const change = (next: ThreadShell | undefined, initial = false) => { + notifier.onThreadChange({ host, previous, next, initial }); + previous = next; + }; + // Pushes are sent after a summary lookup, which fails fast here. + const settle = () => Bun.sleep(5); + return { pushes, change, settle, notifier }; +} + +describe("Notifier", () => { + it("stays quiet while loading the first snapshot", async () => { + const { pushes, change, settle } = setup(); + change(shell({ status: "completed" }), true); + change(shell({ status: "completed" })); + await settle(); + expect(pushes).toEqual([]); + }); + + it("notifies once when a running thread finishes", async () => { + const { pushes, change, settle } = setup(); + change(shell(), true); + change(shell({ status: "completed" })); + change(shell({ status: "completed", updatedAt: "2026-11-01T09:01:00.000Z" })); + await settle(); + expect(pushes.map((push) => [push.category, push.subtitle, push.data.threadId])).toEqual([ + ["finished", "Mac mini · wristcode", "thread-1"], + ]); + }); + + it("notifies for each new approval request, with its request id", async () => { + const { pushes, change, settle } = setup(); + change(shell(), true); + change(shell({ pendingRuntimeRequest: request("r1") })); + change(shell({ pendingRuntimeRequest: request("r2") })); + await settle(); + expect(pushes.map((push) => [push.category, push.data.requestId])).toEqual([ + ["approval", "r1"], + ["approval", "r2"], + ]); + }); + + it("does not announce a turn the user stopped", async () => { + const { pushes, change, settle } = setup(); + change(shell(), true); + change(shell({ status: "completed", latestRunStopRequested: true })); + await settle(); + expect(pushes).toEqual([]); + }); + + it("reports failures with the server's error", async () => { + const { pushes, change, settle } = setup(); + change(shell(), true); + change(shell({ status: "failed", lastError: "Provider quota exceeded" })); + await settle(); + expect(pushes.map((push) => [push.category, push.body])).toEqual([["failed", "Provider quota exceeded"]]); + }); + + it("warns about a token that expires within a week, once a day", async () => { + const { pushes, notifier } = setup(); + const row = { + id: "host-1", + label: "Mac mini", + url: "https://t3mac.gbrown.org", + environment_id: "host-1", + token: "t", + token_expires_at: new Date(Date.now() + 3 * 24 * 60 * 60_000).toISOString(), + created_at: "", + }; + notifier.checkTokenExpiry([row]); + notifier.checkTokenExpiry([row]); + expect(pushes.map((push) => [push.category, push.body])).toEqual([ + ["host", "The gateway's T3 token expires in 3 days. Pair Mac mini again."], + ]); + }); +}); diff --git a/gateway/test/rpc.test.ts b/gateway/test/rpc.test.ts new file mode 100644 index 0000000..1d73003 --- /dev/null +++ b/gateway/test/rpc.test.ts @@ -0,0 +1,90 @@ +// The RPC client against an in-process server that frames messages the way Effect RPC does. +import { afterEach, describe, expect, it } from "bun:test"; +import type { Server, ServerWebSocket } from "bun"; + +import { RpcError, RpcSocket } from "../src/t3/rpc.ts"; + +type Frame = { _tag: string; id?: string; requestId?: string; tag?: string; payload?: unknown }; + +let server: Server | undefined; +afterEach(() => server?.stop(true)); + +/** Starts a server whose `onRequest` decides how to answer each Request frame. */ +function serve(onRequest: (ws: ServerWebSocket, frame: Frame, received: Frame[]) => void) { + const received: Frame[] = []; + server = Bun.serve({ + port: 0, + fetch: (req, srv) => (srv.upgrade(req) ? undefined : new Response("no", { status: 400 })), + websocket: { + message(ws, data) { + const frame = JSON.parse(String(data)) as Frame; + received.push(frame); + if (frame._tag === "Request") onRequest(ws, frame, received); + }, + }, + }); + return { url: `ws://localhost:${server.port}`, received }; +} + +const send = (ws: ServerWebSocket, frame: object) => ws.send(JSON.stringify(frame)); + +describe("RpcSocket", () => { + it("resolves a unary call with its success value", async () => { + const { url } = serve((ws, frame) => + send(ws, { _tag: "Exit", requestId: frame.id, exit: { _tag: "Success", value: { echoed: frame.payload } } }), + ); + const socket = await RpcSocket.connect(url); + expect(await socket.call("server.probe", { a: 1 })).toEqual({ echoed: { a: 1 } }); + socket.close(); + }); + + it("rejects with the server's typed error", async () => { + const error = { _tag: "EnvironmentAuthorizationError", message: "missing scope", requiredScope: "orchestration:operate" }; + const { url } = serve((ws, frame) => + send(ws, { _tag: "Exit", requestId: frame.id, exit: { _tag: "Failure", cause: [{ _tag: "Fail", error }] } }), + ); + const socket = await RpcSocket.connect(url); + const failure = await socket.call("orchestration.dispatchCommand", {}).catch((e: unknown) => e); + expect(failure).toBeInstanceOf(RpcError); + expect((failure as RpcError).error).toEqual(error); + socket.close(); + }); + + it("acks every chunk so the server sends the next one", async () => { + const { url, received } = serve((ws, frame) => { + // Like Effect RPC, send one chunk, then wait for its Ack before the next. + const values = [[1, 2], [3]]; + const next = () => { + const value = values.shift(); + if (value) send(ws, { _tag: "Chunk", requestId: frame.id, values: value }); + else send(ws, { _tag: "Exit", requestId: frame.id, exit: { _tag: "Success", value: null } }); + }; + const waitForAck = setInterval(() => { + const acks = received.filter((f) => f._tag === "Ack").length; + if (acks === 2 - values.length) next(); + if (values.length === 0 && acks === 2) clearInterval(waitForAck); + }, 2); + next(); + }); + const socket = await RpcSocket.connect(url); + const seen: unknown[] = []; + await socket.stream("orchestration.subscribeShell", {}, (value) => seen.push(value)).done; + expect(seen).toEqual([1, 2, 3]); + expect(received.filter((f) => f._tag === "Ack")).toHaveLength(2); + socket.close(); + }); + + it("fails pending calls when the socket closes", async () => { + const { url } = serve((ws) => ws.close()); + const socket = await RpcSocket.connect(url); + expect(socket.call("server.probe", {})).rejects.toThrow("socket closed"); + await socket.closed; + }); + + it("closes a silent socket", async () => { + // This server never answers Pings. + const { url } = serve(() => {}); + const socket = await RpcSocket.connect(url, { pingIntervalMs: 10, silenceTimeoutMs: 40 }); + await socket.closed; + }); +}); diff --git a/gateway/test/views.test.ts b/gateway/test/views.test.ts new file mode 100644 index 0000000..b19f84d --- /dev/null +++ b/gateway/test/views.test.ts @@ -0,0 +1,104 @@ +import { describe, expect, it } from "bun:test"; + +import type { BoundedThread, ThreadShell } from "../src/t3/types.ts"; +import { summarize, trimText } from "../src/views.ts"; + +const thread = (overrides: Partial = {}): ThreadShell => ({ + id: "thread-1", + projectId: "project-1", + title: "Receipt OCR", + createdAt: "2026-11-01T09:00:00.000Z", + updatedAt: "2026-11-01T09:00:00.000Z", + status: "running", + activeRunId: "run-1", + pendingRuntimeRequest: null, + modelSelection: { instanceId: "claudeAgent", model: "claude-opus-5-5" }, + lineage: { relationshipToParent: null }, + ...overrides, +}); + +const message = (role: string, text: string, id = text) => ({ + id, + role, + text, + createdAt: "2026-11-01T09:00:00.000Z", + updatedAt: "2026-11-01T09:00:00.000Z", +}); + +const bounded = (turnItems: BoundedThread["projection"]["turnItems"], messages = [message("assistant", "Tests written.")]) => + ({ snapshotSequence: 1, projection: { thread: thread(), messages, turnItems } }) as BoundedThread; + +describe("trimText", () => { + it("collapses whitespace and cuts at a word", () => { + expect(trimText("one two\n\nthree", 50)).toBe("one two three"); + expect(trimText("alpha beta gamma delta", 15)).toBe("alpha beta…"); + }); +}); + +describe("summarize", () => { + it("takes the last assistant message and skips empty or system ones", () => { + const summary = summarize( + bounded([], [message("user", "Add OCR"), message("assistant", "Done, all green."), message("system", "noise"), message("assistant", " ")]), + thread(), + ); + expect(summary.lastAssistantMessage).toBe("Done, all green."); + expect(summary.messages.map((m) => m.role)).toEqual(["user", "assistant"]); + }); + + it("reads the pending approval's prompt and options", () => { + const summary = summarize( + bounded([ + { + type: "approval_request", + requestId: "r1", + requestKind: "command", + prompt: "bun test --filter ocr", + options: [ + { decision: "accept", label: "Allow once" }, + { decision: "acceptAlways", label: "Always allow", warning: "Applies to every thread" }, + { decision: "decline", label: "Deny" }, + ], + } as BoundedThread["projection"]["turnItems"][number], + ]), + thread({ pendingRuntimeRequest: { id: "r1", kind: "command", createdAt: "2026-11-01T09:00:00.000Z" } }), + ); + expect(summary.request).toEqual({ + id: "r1", + kind: "command", + prompt: "bun test --filter ocr", + reason: null, + options: [ + { decision: "accept", label: "Allow once" }, + { decision: "acceptAlways", label: "Always allow" }, + { decision: "decline", label: "Deny" }, + ], + questions: [], + }); + }); + + it("offers Allow and Deny when the approval's turn item is not loaded yet", () => { + const summary = summarize( + bounded([]), + thread({ pendingRuntimeRequest: { id: "r9", kind: "file-change", createdAt: "2026-11-01T09:00:00.000Z" } }), + ); + expect(summary.request?.options.map((o) => o.decision)).toEqual(["accept", "decline"]); + }); + + it("reads questions, using the label when an option has no value", () => { + const summary = summarize( + bounded([ + { + type: "user_input_request", + requestId: "q1", + questions: [ + { id: "db", header: "Database", question: "Which database?", options: [{ label: "SQLite", description: "" }] }, + ], + } as BoundedThread["projection"]["turnItems"][number], + ]), + thread({ pendingRuntimeRequest: { id: "q1", kind: "user_input", createdAt: "2026-11-01T09:00:00.000Z" } }), + ); + expect(summary.request?.questions).toEqual([ + { id: "db", header: "Database", question: "Which database?", options: [{ label: "SQLite", value: "SQLite" }], allowCustomAnswer: false }, + ]); + }); +});