From 972f6e4be5d1d5c4ff82f8b0fe82df720ed591fe Mon Sep 17 00:00:00 2001 From: Gabriel Brown Date: Sun, 1 Nov 2026 02:04:16 -0500 Subject: [PATCH] Gateway: T3 RPC client, host connection, awareness port, storage, API contract --- .gitignore | 1 + docs/api.md | 164 ++++++++++++++++++++ docs/plan.md | 18 +-- gateway/bun.lock | 26 ++++ gateway/package.json | 18 +++ gateway/src/awareness.ts | 76 ++++++++++ gateway/src/contract.ts | 134 +++++++++++++++++ gateway/src/db.ts | 140 +++++++++++++++++ gateway/src/t3/host.ts | 267 +++++++++++++++++++++++++++++++++ gateway/src/t3/rpc.ts | 183 ++++++++++++++++++++++ gateway/src/t3/types.ts | 199 ++++++++++++++++++++++++ gateway/test/awareness.test.ts | 98 ++++++++++++ gateway/tsconfig.json | 17 +++ 13 files changed, 1332 insertions(+), 9 deletions(-) create mode 100644 docs/api.md create mode 100644 gateway/bun.lock create mode 100644 gateway/package.json create mode 100644 gateway/src/awareness.ts create mode 100644 gateway/src/contract.ts create mode 100644 gateway/src/db.ts create mode 100644 gateway/src/t3/host.ts create mode 100644 gateway/src/t3/rpc.ts create mode 100644 gateway/src/t3/types.ts create mode 100644 gateway/test/awareness.test.ts create mode 100644 gateway/tsconfig.json diff --git a/.gitignore b/.gitignore index bfbe491..ad1ca07 100644 --- a/.gitignore +++ b/.gitignore @@ -11,3 +11,4 @@ xcuserdata/ DerivedData/ build/ *.p8 +gateway/data/ diff --git a/docs/api.md b/docs/api.md new file mode 100644 index 0000000..94dc095 --- /dev/null +++ b/docs/api.md @@ -0,0 +1,164 @@ +# Gateway API + +The watch app talks to the gateway over HTTPS with JSON bodies. The TypeScript types in +[`gateway/src/contract.ts`](../gateway/src/contract.ts) are the source of truth, and the watch +mirrors them in `watch/WristCode/API/Models.swift`. Change both together. + +The base URL is `https://wristcode.gbrown.org`. Every route except `GET /v1/health` and +`POST /v1/pair` needs `Authorization: Bearer `. Timestamps are ISO 8601 strings in UTC. + +Errors return a non-2xx status with `{ "error": "" }`. + +| Status | Meaning | +| --- | --- | +| 400 | The request body is wrong | +| 401 | The device token is missing or revoked. Show the pairing screen | +| 404 | Unknown host, thread, or project | +| 409 | The request is stale, for example an approval that was already answered | +| 429 | Too many pairing attempts | +| 502 | The T3 server rejected the call. The message says why | +| 503 | That host is offline | + +## Pairing + +`wristcode pair` on the gateway prints a 6 digit code that is good for 5 minutes. The watch sends it +once and stores the token in its Keychain. + +``` +POST /v1/pair +{ "code": "482913", "deviceName": "Gabriel's Apple Watch" } + +200 { "deviceId": "9b0c...", "deviceToken": "wc_..." } +``` + +After pairing, and again whenever APNs hands the watch a new token: + +``` +PUT /v1/device/push +{ "apnsToken": "", "apnsEnvironment": "sandbox" } + +204 +``` + +Debug builds use `sandbox`, TestFlight and App Store builds use `production`. + +## Feed + +``` +GET /v1/feed + +200 { + "needsYou": [FeedItem], + "working": [FeedItem], + "recent": [FeedItem], + "hosts": [{ "id": "...", "label": "Mac mini", "status": "online", "error": null }] +} +``` + +`needsYou` holds threads waiting for an approval or an answer, oldest first. `working` holds running +threads, most recently updated first. `recent` holds threads that completed or failed in the last 24 +hours, at most 10. Host `status` is one of `connecting`, `online`, `offline`, `auth_invalid`, or +`incompatible`. + +``` +FeedItem { + "hostId": "...", "hostLabel": "Mac mini", + "threadId": "...", "projectId": "...", "projectTitle": "bill-tracker", + "title": "Receipt OCR", + "phase": "waiting_for_approval", + "updatedAt": "2026-11-01T09:41:00.000Z", + "lastMessage": "Parser and tests are written. I want to run the suite.", + "request": PendingRequest | null +} +``` + +`phase` is one of `starting`, `running`, `waiting_for_approval`, `waiting_for_input`, `completed`, or +`failed`. `lastMessage` is the agent's latest message trimmed to about 280 characters. It can be null +for a moment after a thread changes. + +``` +PendingRequest { + "id": "...", + "kind": "command", + "prompt": "bun test --filter ocr", + "reason": null, + "options": [{ "decision": "accept", "label": "Allow" }, { "decision": "decline", "label": "Deny" }], + "questions": [] +} +``` + +For questions (`kind: "user_input"`), `options` is empty and `questions` lists +`{ id, header, question, options: [{ label, value }], allowCustomAnswer }`. + +## Thread + +``` +GET /v1/threads/:hostId/:threadId + +200 { "item": FeedItem, "messages": [Message], "canStop": true } +Message { "id": "...", "role": "assistant", "text": "...", "createdAt": "..." } +``` + +`messages` holds the last 20, oldest first, each trimmed to 2000 characters. + +``` +POST /v1/threads/:hostId/:threadId/messages +{ "text": "Yes, commit it" } +202 +``` + +A message to a busy thread waits until the current turn ends. Nothing is interrupted. + +``` +POST /v1/threads/:hostId/:threadId/respond +{ "requestId": "...", "decision": "accept" } +or +{ "requestId": "...", "answers": { "": "" } } +202 +``` + +``` +POST /v1/threads/:hostId/:threadId/stop +202 +``` + +## New thread + +``` +GET /v1/projects + +200 { "hosts": [{ "id": "...", "label": "Mac mini", "status": "online", "error": null, + "projects": [{ "id": "...", "title": "wristcode", "model": "claude-opus-5-5" }] }] } +``` + +A project with `"model": null` has no default model, so the watch should not offer it. + +``` +POST /v1/threads +{ "hostId": "...", "projectId": "...", "text": "Add a complication for the waiting count" } + +201 { "hostId": "...", "threadId": "..." } +``` + +The new thread uses the project's default model, or the host's, and the host's default runtime mode. + +## Notifications + +The gateway pushes when a thread starts waiting for you, finishes, or fails. Each push has an +`aps.category` and the custom keys `hostId`, `threadId`, and, for approvals and questions, +`requestId`. + +| Category | When | Actions the watch registers | +| --- | --- | --- | +| `approval` | A thread needs an approval | Allow (`accept`), Deny (`decline`) | +| `input` | A thread asked a question | Open | +| `finished` | A turn completed | Reply (text input), Open | +| `failed` | A turn failed | Open | +| `host` | A host's T3 token expires within 7 days, or was rejected | none | + +## Health + +``` +GET /v1/health +200 { "ok": true, "hosts": [HostSummary] } +``` diff --git a/docs/plan.md b/docs/plan.md index 8f96a53..a9f26fe 100644 --- a/docs/plan.md +++ b/docs/plan.md @@ -130,8 +130,9 @@ to end check is a manual run against server-gib's own T3 before deploying. - Nginx Proxy Manager gets a proxy host for `wristcode.gbrown.org` that points at `wristcode-gateway:8080`, using the `*.gbrown.org` certificate. DNS and the Cloudflare tunnel already cover the subdomain. -- Deploys: Watchtower pulls `:latest` nightly. A `wristcode` entry in `~/Server/ci-deploy.sh` would - make deploys immediate after CI. +- Deploys: after pushing the image, CI calls `~/Server/ci-deploy.sh wristcode` on server-gib over + SSH with a deploy key, which pulls and restarts the stack. Watchtower's nightly pull is the + fallback. - Local development: `bun --watch src/main.ts` against real T3 hosts, with no local image builds. ## Watch app @@ -193,11 +194,10 @@ Phases 1 and 2 happen on the Fedora desktop. Phase 4 runs on the Mac mini in par phase 2 API is fixed. The gateway owns `gateway/`, the Mac mini owns `watch/`, and both pull before they push to `docs/`. -## Open decisions +## Decisions -- **Runtime mode for threads started from the watch.** Either inherit the host's default - (`full-access` on these hosts) or force `approval-required`, so that anything risky waits for a - tap on the wrist. -- **Instant deploys.** Add a `wristcode` entry to `~/Server/ci-deploy.sh` with a CI deploy key, or - rely on Watchtower's nightly pull. -- **Bundle ID.** For example `org.gbrown.wristcode`. +- Threads started from the watch inherit the host's default runtime mode (`full-access` on these + hosts). +- CI deploys right away through a `wristcode` entry in `~/Server/ci-deploy.sh`, with Watchtower's + nightly pull as the fallback. +- The bundle ID is `org.gbrown.wristcode`. diff --git a/gateway/bun.lock b/gateway/bun.lock new file mode 100644 index 0000000..22fa54b --- /dev/null +++ b/gateway/bun.lock @@ -0,0 +1,26 @@ +{ + "lockfileVersion": 2, + "configVersion": 1, + "workspaces": { + "": { + "name": "wristcode-gateway", + "devDependencies": { + "@types/bun": "latest", + "typescript": "^5.9.0", + }, + }, + }, + "packages": { + "@types/bun": ["@types/bun@1.4.4", "", { "dependencies": { "bun-types": "1.4.4" } }, "sha512-9wkRDXUXK124pEnAutE6n/l18j+i9GC5IAHK8TmX/2nlySETmky0OU6yL5U8pW1AoKh++AN9auDCr647BfurjQ=="], + + "@types/node": ["@types/node@26.6.6", "", { "dependencies": { "undici-types": "~8.9.0" } }, "sha512-MzQj+8Kx3XI40OmUAE6Qoh58ZMNstNNMwn2nfd3qWsfe7nDNSqJNzes4njhz6AHwrSHOffSLwXU6yb474Q4aog=="], + + "bun-types": ["bun-types@1.4.4", "", { "dependencies": { "@types/node": "*", "undici-types": "*" } }, "sha512-yjrJh5oGX7mMPKGd19YXyIjmEuj4G4zNC30bs7PM9Az7NQ3ndF3j6hbeRdvJTRQEhJ/FYQwW0a2moweEi2hMxQ=="], + + "typescript": ["typescript@5.9.3", "", { "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" } }, "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw=="], + + "undici-types": ["undici-types@8.12.3", "", {}, "sha512-EXE4ydL3S76dFb27UFjNVZHTyv/TLdATj/KUe29ptdcg6qXL48J2tZ3126E5kICj5jn781OR+XlVH1N6Mz8jbA=="], + + "@types/node/undici-types": ["undici-types@8.9.0", "", {}, "sha512-KTDyRTYX8sWmKXAikPHHSyc63CRPETMctyjKFupcC6OBLXT3xsN0e9aF7m+mIXutFWpUXuedtowG7iLOzp0kQg=="], + } +} diff --git a/gateway/package.json b/gateway/package.json new file mode 100644 index 0000000..b78b3f4 --- /dev/null +++ b/gateway/package.json @@ -0,0 +1,18 @@ +{ + "name": "wristcode-gateway", + "version": "0.1.0", + "private": true, + "type": "module", + "license": "MIT", + "bin": { "wristcode": "src/cli.ts" }, + "scripts": { + "dev": "bun --watch src/main.ts", + "start": "bun src/main.ts", + "test": "bun test", + "typecheck": "tsc --noEmit" + }, + "devDependencies": { + "@types/bun": "latest", + "typescript": "^5.9.0" + } +} diff --git a/gateway/src/awareness.ts b/gateway/src/awareness.ts new file mode 100644 index 0000000..c1ae698 --- /dev/null +++ b/gateway/src/awareness.ts @@ -0,0 +1,76 @@ +// What a thread is doing, from the watch's point of view. Ported from T3 Code's +// `packages/shared/src/agentAwareness.ts` (MIT), which drives T3's own mobile notifications. + +import type { ThreadShell } from "./t3/types.ts"; + +export type Phase = + | "starting" + | "running" + | "waiting_for_approval" + | "waiting_for_input" + | "completed" + | "failed"; + +type AwarenessFields = Pick< + ThreadShell, + | "activityRunStatus" + | "lineage" + | "latestRunStopRequested" + | "pendingBackgroundTasks" + | "pendingRuntimeRequest" + | "settledOverride" + | "status" +>; + +/** + * The phase to show for a thread, or null when the watch should not show it: subagent children, + * threads that never ran or were stopped, and finished work the user already settled. + */ +export function threadPhase(thread: AwarenessFields): Phase | null { + if (thread.lineage.relationshipToParent === "subagent") return null; + const phase = resolvePhase(thread); + if (phase === null) return null; + if (phase === "completed" && thread.latestRunStopRequested === true) return null; + if ( + thread.settledOverride === "settled" && + (phase === "completed" || phase === "failed") && + !backgroundWorkHoldsCompletion(thread.pendingBackgroundTasks ?? []) + ) { + return null; + } + return phase; +} + +function resolvePhase(thread: AwarenessFields): Phase | null { + const runStatus = thread.activityRunStatus ?? thread.status; + const request = thread.pendingRuntimeRequest; + if ( + request?.kind === "user_input" && + (request.blocking !== false || !(runStatus === "preparing" || runStatus === "starting" || runStatus === "running")) + ) { + return "waiting_for_input"; + } + if (request && request.kind !== "user_input" && request.kind !== "auth_refresh") { + return "waiting_for_approval"; + } + switch (runStatus) { + case "preparing": + case "starting": + return "starting"; + case "running": + case "waiting": + return "running"; + case "completed": + return backgroundWorkHoldsCompletion(thread.pendingBackgroundTasks ?? []) ? "running" : "completed"; + case "failed": + return "failed"; + default: + return null; + } +} + +// Commands and monitors are as often a dev server or its watchdog as a wait, so they hold +// completion only when the agent marks them awaited. +function backgroundWorkHoldsCompletion(tasks: NonNullable): boolean { + return tasks.some((task) => task.awaited === true || task.kind === "subagent" || task.kind === "background_task"); +} diff --git a/gateway/src/contract.ts b/gateway/src/contract.ts new file mode 100644 index 0000000..b3a1c3d --- /dev/null +++ b/gateway/src/contract.ts @@ -0,0 +1,134 @@ +// The JSON the watch app reads and sends. This is the contract between the gateway and the +// watch's Codable models (watch/WristCode/API/Models.swift). docs/api.md describes it for humans. +// Change both sides together. + +import type { Phase } from "./awareness.ts"; +import type { ApprovalDecision } from "./t3/types.ts"; +import type { HostStatus } from "./t3/host.ts"; + +export interface HostSummary { + id: string; + label: string; + status: HostStatus; + error: string | null; +} + +/** Something a thread is waiting on: an approval, or questions for the user. */ +export interface PendingRequest { + id: string; + /** `command`, `file-change`, `permission`, `user_input`, ... */ + kind: string; + /** What the agent wants to do, such as the command line. Null when the server gave none. */ + prompt: string | null; + reason: string | null; + /** Approval choices, in the server's order. Empty for questions. */ + options: Array<{ decision: ApprovalDecision; label: string }>; + /** Questions for the user. Empty for approvals. */ + questions: Array<{ + id: string; + header: string; + question: string; + options: Array<{ label: string; value: string }>; + allowCustomAnswer: boolean; + }>; +} + +export interface FeedItem { + hostId: string; + hostLabel: string; + threadId: string; + projectId: string; + projectTitle: string; + title: string; + phase: Phase; + updatedAt: string; + /** The agent's latest message, trimmed for the watch. Null until the gateway has read it. */ + lastMessage: string | null; + request: PendingRequest | null; +} + +export interface Feed { + /** Waiting for approval or input. Oldest request first. */ + needsYou: FeedItem[]; + /** Starting or running. Most recently updated first. */ + working: FeedItem[]; + /** Completed or failed in the last 24 hours. Most recent first, at most 10. */ + recent: FeedItem[]; + hosts: HostSummary[]; +} + +export interface Message { + id: string; + role: "user" | "assistant" | "system"; + text: string; + createdAt: string; +} + +export interface ThreadDetail { + item: FeedItem; + /** Oldest first, at most 20. */ + messages: Message[]; + /** True while a run is active, so Stop makes sense. */ + canStop: boolean; +} + +export interface ProjectsResponse { + hosts: Array< + HostSummary & { + projects: Array<{ + id: string; + title: string; + /** The model a new thread will use, such as `claude-opus-5-5`. Null when none is configured. */ + model: string | null; + }>; + } + >; +} + +export interface PairRequest { + code: string; + deviceName: string; +} + +export interface PairResponse { + deviceId: string; + deviceToken: string; +} + +export interface PushRegistration { + apnsToken: string; + apnsEnvironment: "sandbox" | "production"; +} + +export interface SendMessageRequest { + text: string; +} + +export type RespondRequest = + | { requestId: string; decision: ApprovalDecision } + | { requestId: string; answers: Record }; + +export interface NewThreadRequest { + hostId: string; + projectId: string; + text: string; +} + +export interface NewThreadResponse { + hostId: string; + threadId: string; +} + +export interface ErrorResponse { + error: string; +} + +/** Custom keys on every push notification, beside `aps`. */ +export interface PushPayload { + hostId: string; + threadId: string; + requestId?: string; +} + +/** Notification categories the watch registers actions for. */ +export type PushCategory = "approval" | "finished" | "failed" | "input" | "host"; diff --git a/gateway/src/db.ts b/gateway/src/db.ts new file mode 100644 index 0000000..f889828 --- /dev/null +++ b/gateway/src/db.ts @@ -0,0 +1,140 @@ +// SQLite state: the T3 hosts the gateway connects to, the watches paired with it, and pending +// pairing codes. The CLI and the server open the same file, and the server rereads `hosts` so a +// `wristcode host add` takes effect without a restart. + +import { Database } from "bun:sqlite"; + +export interface HostRow { + id: string; + label: string; + url: string; + environment_id: string; + token: string; + token_expires_at: string | null; + created_at: string; +} + +export interface DeviceRow { + id: string; + name: string; + token_hash: string; + apns_token: string | null; + apns_env: "sandbox" | "production" | null; + created_at: string; + last_seen_at: string | null; + revoked_at: string | null; +} + +const SCHEMA = ` +create table if not exists hosts ( + id text primary key, + label text not null, + url text not null unique, + environment_id text not null unique, + token text not null, + token_expires_at text, + created_at text not null +); +create table if not exists devices ( + id text primary key, + name text not null, + token_hash text not null unique, + apns_token text, + apns_env text, + created_at text not null, + last_seen_at text, + revoked_at text +); +create table if not exists pairing_codes ( + code_hash text primary key, + expires_at text not null, + used_at text +); +`; + +export const sha256 = (value: string) => new Bun.CryptoHasher("sha256").update(value).digest("hex"); + +const now = () => new Date().toISOString(); + +export function openDb(path: string) { + const db = new Database(path, { create: true, strict: true }); + db.run("pragma journal_mode = wal"); + db.run("pragma busy_timeout = 5000"); + db.run(SCHEMA); + + return { + hosts: { + list: () => db.query("select * from hosts order by created_at").all(), + upsert(host: Omit) { + db.query( + `insert into hosts (id, label, url, environment_id, token, token_expires_at, created_at) + values ($id, $label, $url, $environment_id, $token, $token_expires_at, $created_at) + on conflict(environment_id) do update set + label = excluded.label, url = excluded.url, token = excluded.token, + token_expires_at = excluded.token_expires_at`, + ).run({ ...host, created_at: now() }); + }, + remove: (idOrLabel: string) => + db.query("delete from hosts where id = $v or label = $v").run({ v: idOrLabel }).changes > 0, + }, + + devices: { + list: () => db.query("select * from devices order by created_at").all(), + active: () => db.query("select * from devices where revoked_at is null").all(), + create(name: string, token: string): string { + const id = crypto.randomUUID(); + db.query( + "insert into devices (id, name, token_hash, created_at) values ($id, $name, $hash, $created_at)", + ).run({ id, name, hash: sha256(token), created_at: now() }); + return id; + }, + /** The active device that owns this bearer token, marking it seen. */ + authenticate(token: string): DeviceRow | null { + const device = db + .query("select * from devices where token_hash = $hash and revoked_at is null") + .get({ hash: sha256(token) }); + if (device) db.query("update devices set last_seen_at = $at where id = $id").run({ at: now(), id: device.id }); + return device; + }, + setPush(id: string, apnsToken: string, apnsEnv: "sandbox" | "production") { + db.query("update devices set apns_token = $apnsToken, apns_env = $apnsEnv where id = $id").run({ + id, + apnsToken, + apnsEnv, + }); + }, + clearPush: (apnsToken: string) => + db.query("update devices set apns_token = null where apns_token = $apnsToken").run({ apnsToken }), + revoke: (idOrName: string) => + db.query("update devices set revoked_at = $at where (id = $v or name = $v) and revoked_at is null").run({ + at: now(), + v: idOrName, + }).changes, + }, + + pairing: { + create(code: string, ttlMs: number) { + db.query("insert into pairing_codes (code_hash, expires_at) values ($hash, $expires_at)").run({ + hash: sha256(code), + expires_at: new Date(Date.now() + ttlMs).toISOString(), + }); + }, + /** Marks the code used and returns true if it was valid. */ + redeem(code: string): boolean { + return ( + db + .query( + "update pairing_codes set used_at = $at where code_hash = $hash and used_at is null and expires_at > $at", + ) + .run({ hash: sha256(code), at: now() }).changes > 0 + ); + }, + /** Voids every open code, after too many wrong guesses. */ + voidAll: () => db.query("update pairing_codes set used_at = $at where used_at is null").run({ at: now() }), + }, + + close: () => db.close(), + }; +} + +export type Db = ReturnType; diff --git a/gateway/src/t3/host.ts b/gateway/src/t3/host.ts new file mode 100644 index 0000000..52088ad --- /dev/null +++ b/gateway/src/t3/host.ts @@ -0,0 +1,267 @@ +// One T3 Code server. Keeps a shell stream open, holds every project and thread shell in memory, +// and reconnects with backoff. Writes go over the same socket, and thread pages go over HTTP. + +import { RpcError, RpcSocket } from "./rpc.ts"; +import { + ORCHESTRATION_PROTOCOL_VERSION, + type BoundedThread, + type Command, + type EnvironmentDescriptor, + type LaunchThreadInput, + type Project, + type ServerConfig, + type ShellStreamItem, + type ThreadShell, +} from "./types.ts"; + +export type HostStatus = + | "connecting" + | "online" + | "offline" + /** The token expired or was revoked. A new one has to be paired in. */ + | "auth_invalid" + /** The server speaks another protocol version or is a different environment. */ + | "incompatible"; + +export interface HostConfig { + id: string; + label: string; + /** The server's HTTP origin, such as `https://t3mac.gbrown.org`. */ + url: string; + token: string; + environmentId: string; +} + +export interface ThreadChange { + host: T3Host; + previous: ThreadShell | undefined; + next: ThreadShell | undefined; + /** True while loading the first snapshot after start, when nothing should notify. */ + initial: boolean; +} + +export interface HostOptions { + onThreadChange?: (change: ThreadChange) => void; + onStatus?: (host: T3Host) => void; + log?: (message: string) => void; +} + +const BACKOFF_MIN_MS = 1_000; +const BACKOFF_MAX_MS = 5 * 60_000; +const AUTH_RETRY_MS = 10 * 60_000; + +export class HostError extends Error { + constructor( + message: string, + readonly status: number, + ) { + super(message); + this.name = "HostError"; + } +} + +export class T3Host { + status: HostStatus = "connecting"; + lastError: string | null = null; + readonly projects = new Map(); + readonly threads = new Map(); + serverConfig: ServerConfig | null = null; + + private socket: RpcSocket | null = null; + private sequence: number | null = null; + private hasSnapshot = false; + private stopped = false; + private attempt = 0; + private retryTimer: ReturnType | null = null; + + constructor( + readonly config: HostConfig, + private readonly options: HostOptions = {}, + ) {} + + get id() { + return this.config.id; + } + + get label() { + return this.config.label; + } + + start() { + this.stopped = false; + void this.connect(); + } + + stop() { + this.stopped = true; + if (this.retryTimer) clearTimeout(this.retryTimer); + this.socket?.close(); + } + + /** Sends one orchestration command and resolves with its sequence. */ + async dispatch(command: Command): Promise<{ sequence: number }> { + return this.requireSocket().call("orchestration.dispatchCommand", command); + } + + async launchThread(input: LaunchThreadInput): Promise<{ threadId: string }> { + return this.requireSocket().call("orchestration.launchThread", input); + } + + /** The recent window of one thread: its messages and turn items. */ + bounded(threadId: string): Promise { + return this.http(`/api/orchestration/threads/${encodeURIComponent(threadId)}/bounded`); + } + + /** The session's expiry, from `/api/auth/session`. */ + async sessionExpiresAt(): Promise { + const session = await this.http<{ expiresAt?: string }>("/api/auth/session"); + return session.expiresAt ?? null; + } + + supports(capability: string): boolean { + return this.serverConfig?.environment?.capabilities?.[capability] === true; + } + + private requireSocket(): RpcSocket { + if (this.socket === null || this.status !== "online") { + throw new HostError(`${this.label} is ${this.status}`, 503); + } + return this.socket; + } + + private async http(path: string, init: RequestInit = {}): Promise { + const response = await fetch(new URL(path, this.config.url), { + ...init, + headers: { + authorization: `Bearer ${this.config.token}`, + "x-t3-orchestration-protocol": String(ORCHESTRATION_PROTOCOL_VERSION), + ...init.headers, + }, + }); + if (!response.ok) { + const body = await response.text(); + throw new HostError(`${this.label} ${path} returned ${response.status}: ${body.slice(0, 200)}`, response.status); + } + return response.json() as Promise; + } + + private setStatus(status: HostStatus, error: string | null = null) { + this.status = status; + this.lastError = error; + this.options.onStatus?.(this); + } + + private async connect() { + if (this.stopped) return; + this.setStatus("connecting", this.lastError); + try { + const descriptor = await this.http("/.well-known/t3/environment"); + const protocol = descriptor.orchestrationProtocolVersion ?? 1; + if (protocol !== ORCHESTRATION_PROTOCOL_VERSION) { + throw new HostError(`speaks orchestration protocol ${protocol}, the gateway speaks ${ORCHESTRATION_PROTOCOL_VERSION}`, 426); + } + if (descriptor.environmentId !== this.config.environmentId) { + throw new HostError(`is environment ${descriptor.environmentId}, expected ${this.config.environmentId}`, 409); + } + const { ticket } = await this.http<{ ticket: string }>("/api/auth/websocket-ticket/rpc", { method: "POST" }); + const wsUrl = new URL("/ws", this.config.url); + wsUrl.protocol = wsUrl.protocol === "https:" ? "wss:" : "ws:"; + wsUrl.searchParams.set("wsTicket", ticket); + wsUrl.searchParams.set("orchestrationProtocol", String(ORCHESTRATION_PROTOCOL_VERSION)); + const socket = await RpcSocket.connect(wsUrl.href); + this.socket = socket; + this.serverConfig = await socket.call("server.getConfig", {}); + this.subscribe(socket); + this.attempt = 0; + this.setStatus("online"); + await socket.closed; + this.socket = null; + this.scheduleRetry("offline", "connection closed"); + } catch (error) { + this.socket?.close(); + this.socket = null; + const message = error instanceof Error ? error.message : String(error); + if (error instanceof HostError && error.status === 401) this.scheduleRetry("auth_invalid", message); + else if (error instanceof HostError && (error.status === 426 || error.status === 409)) this.scheduleRetry("incompatible", message); + else this.scheduleRetry("offline", message); + } + } + + private subscribe(socket: RpcSocket) { + const input = this.hasSnapshot && this.sequence !== null ? { afterSequence: this.sequence } : {}; + const { done } = socket.stream("orchestration.subscribeShell", input, (value) => + this.apply(value as ShellStreamItem), + ); + // The stream only ends when the socket does, or on a server error. Either way, start over. + done.then( + () => socket.close(), + (error: RpcError) => { + this.options.log?.(`${this.label}: shell stream ended: ${error.message}`); + socket.close(); + }, + ); + } + + private scheduleRetry(status: HostStatus, error: string) { + if (this.stopped) return; + this.setStatus(status, error); + this.options.log?.(`${this.label}: ${status}: ${error}`); + const delay = + status === "auth_invalid" || status === "incompatible" + ? AUTH_RETRY_MS + : Math.min(BACKOFF_MAX_MS, BACKOFF_MIN_MS * 2 ** this.attempt++) * (0.5 + Math.random() / 2); + this.retryTimer = setTimeout(() => void this.connect(), delay); + } + + /** Applies one shell stream item to the in-memory state, reporting each thread that changed. */ + apply(item: ShellStreamItem) { + switch (item.kind) { + case "snapshot": { + const { snapshot } = item; + for (const project of snapshot.projects) this.projects.set(project.id, project); + const known = new Set(snapshot.projects.map((project) => project.id)); + for (const id of this.projects.keys()) if (!known.has(id)) this.projects.delete(id); + this.sequence = Math.max(this.sequence ?? 0, snapshot.snapshotSequence); + // The server also sends project-only snapshots with no threads: one first on every resume, + // and more as project metadata fills in. T3's own client keeps its threads for those, and + // so does the gateway. + if (snapshot.threads.length === 0 && this.hasSnapshot) return; + const initial = !this.hasSnapshot; + this.hasSnapshot = true; + const next = new Map(snapshot.threads.map((thread) => [thread.id, thread])); + for (const [id, previous] of this.threads) { + if (!next.has(id)) this.changeThread(previous, undefined, initial); + } + for (const thread of next.values()) this.changeThread(this.threads.get(thread.id), thread, initial); + return; + } + case "project.updated": + this.projects.set(item.project.id, item.project); + this.sequence = item.sequence; + return; + case "project.removed": + this.projects.delete(item.projectId); + this.sequence = item.sequence; + return; + case "thread.updated": + this.sequence = item.sequence; + if (item.location === "active") this.changeThread(this.threads.get(item.thread.id), item.thread, false); + else this.changeThread(this.threads.get(item.thread.id), undefined, false); + return; + case "thread.removed": { + this.sequence = item.sequence; + const previous = this.threads.get(item.threadId); + if (previous) this.changeThread(previous, undefined, false); + return; + } + case "synchronized": + return; + } + } + + private changeThread(previous: ThreadShell | undefined, next: ThreadShell | undefined, initial: boolean) { + if (next) this.threads.set(next.id, next); + else if (previous) this.threads.delete(previous.id); + this.options.onThreadChange?.({ host: this, previous, next, initial }); + } +} diff --git a/gateway/src/t3/rpc.ts b/gateway/src/t3/rpc.ts new file mode 100644 index 0000000..69938eb --- /dev/null +++ b/gateway/src/t3/rpc.ts @@ -0,0 +1,183 @@ +// A minimal client for T3 Code's WebSocket RPC, which speaks Effect RPC's JSON framing. +// One socket multiplexes unary calls and streams by request id. Streams are flow controlled: the +// server waits for an Ack after every Chunk, so `stream` acks each one after its handler runs. +// See docs/t3-protocol.md, section 2. + +type FromServer = + | { _tag: "Chunk"; requestId: string; values: unknown[] } + | { _tag: "Exit"; requestId: string; exit: Exit } + | { _tag: "Defect"; defect: unknown } + | { _tag: "Pong" } + | { _tag: "ClientProtocolError"; error: unknown }; + +type Exit = + | { _tag: "Success"; value: unknown } + | { _tag: "Failure"; cause: Array<{ _tag: string; error?: unknown; defect?: unknown }> }; + +/** A call that ended in a Failure exit, or a socket that closed under it. */ +export class RpcError extends Error { + constructor( + message: string, + /** The typed error from the server, such as `{ _tag: "EnvironmentAuthorizationError", ... }`. */ + readonly error?: unknown, + ) { + super(message); + this.name = "RpcError"; + } +} + +interface Pending { + onChunk?: (value: unknown) => void; + resolve: (value: unknown) => void; + reject: (error: RpcError) => void; +} + +export interface RpcSocketOptions { + pingIntervalMs?: number; + /** Close the socket when the server has sent nothing for this long. */ + silenceTimeoutMs?: number; +} + +/** One open RPC WebSocket. It never reconnects; the caller makes a new one. */ +export class RpcSocket { + private nextId = 1; + private readonly pending = new Map(); + private lastFrameAt = Date.now(); + private readonly timer: ReturnType; + /** Resolves when the socket closes, for any reason. */ + readonly closed: Promise; + + private constructor( + private readonly ws: WebSocket, + options: RpcSocketOptions, + ) { + const pingIntervalMs = options.pingIntervalMs ?? 5_000; + const silenceTimeoutMs = options.silenceTimeoutMs ?? 15_000; + this.closed = new Promise((resolve) => ws.addEventListener("close", () => resolve())); + ws.addEventListener("message", (event) => this.onMessage(event.data)); + ws.addEventListener("close", () => this.failAll("socket closed")); + this.timer = setInterval(() => { + if (Date.now() - this.lastFrameAt > silenceTimeoutMs) { + this.close(); + return; + } + this.send({ _tag: "Ping" }); + }, pingIntervalMs); + void this.closed.then(() => clearInterval(this.timer)); + } + + /** Opens a socket and resolves once it is connected. */ + static connect(url: string, options: RpcSocketOptions = {}): Promise { + return new Promise((resolve, reject) => { + const ws = new WebSocket(url); + const onError = () => reject(new RpcError(`could not connect to ${new URL(url).host}`)); + ws.addEventListener("error", onError, { once: true }); + ws.addEventListener( + "open", + () => { + ws.removeEventListener("error", onError); + resolve(new RpcSocket(ws, options)); + }, + { once: true }, + ); + }); + } + + /** Calls a unary method and resolves with its success value. */ + call(tag: string, payload: unknown): Promise { + return new Promise((resolve, reject) => { + const id = this.request(tag, payload, { resolve: resolve as (v: unknown) => void, reject }); + if (id === null) reject(new RpcError("socket is not open")); + }); + } + + /** + * Opens a stream. `onChunk` runs once per value, in order. The promise settles when the stream + * ends or fails. Call `cancel` to interrupt it without closing the socket. + */ + stream(tag: string, payload: unknown, onChunk: (value: unknown) => void) { + let id: string | null = null; + const done = new Promise((resolve, reject) => { + id = this.request(tag, payload, { onChunk, resolve: () => resolve(), reject }); + if (id === null) reject(new RpcError("socket is not open")); + }); + return { + done, + cancel: () => { + if (id !== null) this.send({ _tag: "Interrupt", requestId: id }); + }, + }; + } + + close() { + clearInterval(this.timer); + this.ws.close(); + } + + private request(tag: string, payload: unknown, pending: Pending): string | null { + if (this.ws.readyState !== WebSocket.OPEN) return null; + const id = String(this.nextId++); + this.pending.set(id, pending); + this.send({ _tag: "Request", id, tag, payload, headers: [] }); + return id; + } + + private send(frame: object) { + if (this.ws.readyState === WebSocket.OPEN) this.ws.send(JSON.stringify(frame)); + } + + private onMessage(data: unknown) { + this.lastFrameAt = Date.now(); + if (typeof data !== "string") return; + const parsed: unknown = JSON.parse(data); + const frames = (Array.isArray(parsed) ? parsed : [parsed]) as FromServer[]; + for (const frame of frames) this.onFrame(frame); + } + + private onFrame(frame: FromServer) { + switch (frame._tag) { + case "Chunk": { + const requestId = String(frame.requestId); + const pending = this.pending.get(requestId); + if (!pending) return; + for (const value of frame.values) pending.onChunk?.(value); + this.send({ _tag: "Ack", requestId: frame.requestId }); + return; + } + case "Exit": { + const requestId = String(frame.requestId); + const pending = this.pending.get(requestId); + if (!pending) return; + this.pending.delete(requestId); + if (frame.exit._tag === "Success") pending.resolve(frame.exit.value); + else pending.reject(failureToError(frame.exit.cause)); + return; + } + case "Defect": + this.failAll(`server defect: ${JSON.stringify(frame.defect)}`); + this.close(); + return; + case "ClientProtocolError": + this.failAll(`protocol error: ${JSON.stringify(frame.error)}`); + this.close(); + return; + case "Pong": + return; + } + } + + private failAll(message: string) { + for (const pending of this.pending.values()) pending.reject(new RpcError(message)); + this.pending.clear(); + } +} + +function failureToError(cause: Extract["cause"]): RpcError { + const first = cause[0]; + if (first?.error !== undefined) { + const error = first.error as { _tag?: string; message?: string }; + return new RpcError(error.message ?? error._tag ?? "call failed", first.error); + } + if (first?._tag === "Interrupt") return new RpcError("interrupted"); + return new RpcError(`call failed: ${JSON.stringify(first?.defect ?? cause)}`); +} diff --git a/gateway/src/t3/types.ts b/gateway/src/t3/types.ts new file mode 100644 index 0000000..2afc15e --- /dev/null +++ b/gateway/src/t3/types.ts @@ -0,0 +1,199 @@ +// Narrow types for the parts of T3 Code's orchestration protocol (v2) the gateway reads and writes. +// They cover only the fields we use. Unknown fields pass through untouched, and every enum leaves +// room for values a newer server may add. Source of truth: docs/t3-protocol.md. + +/** A string union that still accepts values a newer server adds. */ +type Open = T | (string & {}); + +export type RunStatus = Open< + | "idle" + | "preparing" + | "queued" + | "starting" + | "running" + | "waiting" + | "completed" + | "interrupted" + | "failed" + | "cancelled" + | "rolled_back" +>; + +export type RuntimeRequestKind = Open< + | "command" + | "file-read" + | "file-change" + | "mcp-elicitation" + | "permission" + | "user_input" + | "dynamic_tool_call" + | "auth_refresh" +>; + +export type ApprovalDecision = "accept" | "acceptForSession" | "acceptAlways" | "decline" | "cancel"; + +export interface ModelSelection { + instanceId: string; + model: string; + options?: unknown; +} + +export interface Project { + id: string; + title: string; + workspaceRoot: string; + defaultModelSelection: ModelSelection | null; + updatedAt: string; +} + +export interface ThreadShell { + id: string; + projectId: string; + title: string; + createdAt: string; + updatedAt: string; + archivedAt?: string | null; + deletedAt?: string | null; + status: RunStatus; + activityRunStatus?: RunStatus | null; + activeRunId: string | null; + latestRunStopRequested?: boolean; + pendingRuntimeRequest: { + id: string; + kind: RuntimeRequestKind; + createdAt: string; + blocking?: boolean; + } | null; + pendingBackgroundTasks?: Array<{ kind: Open<"command" | "monitor" | "subagent" | "background_task">; awaited?: boolean }>; + lastError?: string | null; + modelSelection: ModelSelection; + runtimeMode?: string; + lineage: { relationshipToParent: Open<"subagent"> | null }; + settledOverride?: Open<"settled"> | null; +} + +export interface ShellSnapshot { + snapshotSequence: number; + projects: Project[]; + threads: ThreadShell[]; +} + +export type ShellStreamItem = + | { kind: "snapshot"; snapshot: ShellSnapshot } + | { kind: "project.updated"; sequence: number; project: Project } + | { kind: "project.removed"; sequence: number; projectId: string } + | { kind: "thread.updated"; sequence: number; location: Open<"active" | "archive">; thread: ThreadShell } + | { kind: "thread.removed"; sequence: number; location: Open<"active" | "archive">; threadId: string } + | { kind: "synchronized" }; + +export interface ConversationMessage { + id: string; + role: Open<"user" | "assistant" | "system">; + text: string; + streaming?: boolean; + createdAt: string; + updatedAt: string; +} + +export interface ApprovalOption { + decision: ApprovalDecision; + label: string; + warning?: string; +} + +export interface UserInputQuestion { + id: string; + header: string; + question: string; + options: Array<{ label: string; description: string; value?: string }>; + allowCustomAnswer?: boolean; + multiSelect?: boolean; +} + +/** Any turn item. Narrow with the guards below to read the two types the gateway uses. */ +export interface TurnItem { + type: string; + title?: string | null; +} + +export interface ApprovalRequestItem extends TurnItem { + type: "approval_request"; + requestId: string; + requestKind: RuntimeRequestKind; + prompt?: string; + reason?: string; + options?: ApprovalOption[]; +} + +export interface UserInputRequestItem extends TurnItem { + type: "user_input_request"; + requestId: string; + questions: UserInputQuestion[]; +} + +export const isApprovalRequest = (item: TurnItem): item is ApprovalRequestItem => + item.type === "approval_request"; + +export const isUserInputRequest = (item: TurnItem): item is UserInputRequestItem => + item.type === "user_input_request"; + +export interface BoundedThread { + snapshotSequence: number; + projection: { + thread: ThreadShell; + messages: ConversationMessage[]; + turnItems: TurnItem[]; + }; +} + +export interface EnvironmentDescriptor { + environmentId: string; + label: string; + serverVersion: string; + orchestrationProtocolVersion?: number; +} + +export interface ServerConfig { + environment?: { capabilities?: Record }; + settings?: { defaultModelSelection?: ModelSelection | null }; +} + +/** Commands sent through `orchestration.dispatchCommand`. */ +export type Command = + | { + type: "message.dispatch"; + commandId: string; + threadId: string; + messageId: string; + text: string; + attachments: []; + createdBy: "user"; + creationSource: "mobile"; + dispatchMode: { type: "queue_after_active" }; + } + | { + type: "runtime-request.respond"; + commandId: string; + threadId: string; + requestId: string; + decision?: ApprovalDecision; + answers?: Record; + } + | { type: "thread.stop"; commandId: string; threadId: string } + | { type: "run.interrupt"; commandId: string; threadId: string; runId: string }; + +export interface LaunchThreadInput { + commandId: string; + creationSource: "mobile"; + projectId: string; + title: string; + generateTitle: boolean; + modelSelection: ModelSelection; + runtimeMode: string; + interactionMode: "default"; + workspaceStrategy: { type: "root" }; + initialMessage: { messageId: string; text: string; attachments: [] }; +} + +/** The protocol version this gateway speaks. A server on another version is refused. */ +export const ORCHESTRATION_PROTOCOL_VERSION = 2; diff --git a/gateway/test/awareness.test.ts b/gateway/test/awareness.test.ts new file mode 100644 index 0000000..a9eed50 --- /dev/null +++ b/gateway/test/awareness.test.ts @@ -0,0 +1,98 @@ +// Cases ported from T3 Code's agentAwareness.test.ts, so the gateway notifies when T3's own app would. +import { describe, expect, it } from "bun:test"; + +import { threadPhase } from "../src/awareness.ts"; +import type { ThreadShell } from "../src/t3/types.ts"; + +const thread = (overrides: Partial = {}): ThreadShell => ({ + id: "thread-2", + projectId: "project-1", + title: "Integrate orchestration", + createdAt: "2026-05-22T12:00:00.000Z", + updatedAt: "2026-05-22T12:00:00.000Z", + status: "running", + activeRunId: "run-1", + pendingRuntimeRequest: null, + modelSelection: { instanceId: "codex", model: "gpt-5.4" }, + lineage: { relationshipToParent: null }, + settledOverride: null, + ...overrides, +}); + +const request = (kind: string, blocking?: boolean): ThreadShell["pendingRuntimeRequest"] => ({ + id: `request-${kind}`, + kind, + createdAt: "2026-05-22T12:00:00.000Z", + ...(blocking === undefined ? {} : { blocking }), +}); + +describe("threadPhase", () => { + it("does not announce completion after a manual stop", () => { + expect(threadPhase(thread({ status: "completed" }))).toBe("completed"); + expect(threadPhase(thread({ status: "completed", latestRunStopRequested: true }))).toBeNull(); + }); + + it.each(["running", "completed", "failed"])("hides %s subagent activity", (status) => { + expect(threadPhase(thread({ status, lineage: { relationshipToParent: "subagent" } }))).toBeNull(); + }); + + it("keeps an older activity run visible over a newer cancelled run", () => { + expect(threadPhase(thread({ status: "cancelled", activityRunStatus: "running" }))).toBe("running"); + }); + + it.each(["completed", "failed"])("hides settled %s work", (status) => { + expect(threadPhase(thread({ status, settledOverride: "settled" }))).toBeNull(); + }); + + it.each([ + ["starting", "starting"], + ["running", "running"], + ["waiting", "running"], + ] as const)("keeps settled %s work visible", (status, phase) => { + expect(threadPhase(thread({ status, settledOverride: "settled" }))).toBe(phase); + }); + + it.each([ + ["user_input", "waiting_for_input"], + ["command", "waiting_for_approval"], + ] as const)("keeps a settled terminal thread with a pending %s request", (kind, phase) => { + expect( + threadPhase(thread({ status: "completed", settledOverride: "settled", pendingRuntimeRequest: request(kind) })), + ).toBe(phase); + }); + + it("keeps settled work held open by an awaited monitor", () => { + expect( + threadPhase( + thread({ + status: "completed", + settledOverride: "settled", + pendingBackgroundTasks: [{ kind: "monitor", awaited: true }], + }), + ), + ).toBe("running"); + }); + + it.each([ + ["only a dev server", "completed", [{ kind: "command" }]], + ["a dev server's watchdog monitor", "completed", [{ kind: "monitor" }]], + ["a command the agent waits on", "running", [{ kind: "command", awaited: true }]], + ["a dev server and a subagent", "running", [{ kind: "command" }, { kind: "subagent" }]], + ] as const)("reports a completed run waiting on %s as %s", (_case, phase, tasks) => { + expect(threadPhase(thread({ status: "completed", pendingBackgroundTasks: [...tasks] }))).toBe(phase); + }); + + it("reports an async question as input only after the working run completes", () => { + const asking = thread({ pendingRuntimeRequest: request("user_input", false) }); + expect(threadPhase(asking)).toBe("running"); + expect(threadPhase({ ...asking, status: "completed" })).toBe("waiting_for_input"); + }); + + it("does not present authentication refreshes as approvals", () => { + expect(threadPhase(thread({ pendingRuntimeRequest: request("auth_refresh") }))).toBe("running"); + }); + + it("hides threads that never ran", () => { + expect(threadPhase(thread({ status: "idle" }))).toBeNull(); + }); +}); diff --git a/gateway/tsconfig.json b/gateway/tsconfig.json new file mode 100644 index 0000000..2deeb74 --- /dev/null +++ b/gateway/tsconfig.json @@ -0,0 +1,17 @@ +{ + "compilerOptions": { + "target": "ESNext", + "module": "Preserve", + "moduleResolution": "bundler", + "lib": ["ESNext"], + "types": ["bun"], + "strict": true, + "noUncheckedIndexedAccess": true, + "exactOptionalPropertyTypes": true, + "verbatimModuleSyntax": true, + "allowImportingTsExtensions": true, + "noEmit": true, + "skipLibCheck": true + }, + "include": ["src", "test"] +}