Gateway: T3 RPC client, host connection, awareness port, storage, API contract

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<string, Project>();
+  readonly threads = new Map<string, ThreadShell>();
+  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<typeof setTimeout> | 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<BoundedThread> {
+    return this.http<BoundedThread>(`/api/orchestration/threads/${encodeURIComponent(threadId)}/bounded`);
+  }
+
+  /** The session's expiry, from `/api/auth/session`. */
+  async sessionExpiresAt(): Promise<string | null> {
+    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<T>(path: string, init: RequestInit = {}): Promise<T> {
+    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<T>;
+  }
+
+  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<EnvironmentDescriptor>("/.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<ServerConfig>("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 });
+  }
+}