client: adopt official @modernrelay/omnigraph SDK + parallelize ego reads

packages/client/package.json

@@ -20,6 +20,7 @@
     "test": "vitest run"
   },
   "dependencies": {
+    "@modernrelay/omnigraph": "^0.6.0",
     "@omnigraph/runtime": "workspace:*",
     "@omnigraph/notebook-spec": "workspace:*"
   },

packages/client/src/http.test.ts

@@ -0,0 +1,125 @@
+import { describe, it, expect } from "vitest";
+import { Client, OmnigraphHttpError } from "./http.js";
+
+// Regexes copied from packages/web/src/error-classifier.ts — the message
+// contract this facade must preserve when re-wrapping SDK errors. (web is
+// not a dependency of @omnigraph/client, so we assert against copies.)
+const RE_UNAUTHORIZED = /returned 401|"code"\s*:\s*"unauthorized"/i;
+const RE_NETWORK = /Failed to fetch|NetworkError|TypeError: NetworkError|net::ERR/i;
+const RE_CONFLICT = /stale view of.*expected manifest table version|ExpectedVersionMismatch/i;
+
+function jsonResponse(body: unknown, status = 200): Response {
+  return new Response(JSON.stringify(body), {
+    status,
+    headers: { "content-type": "application/json" },
+  });
+}
+
+function fetchReturning(res: Response): typeof fetch {
+  return (async () => res) as unknown as typeof fetch;
+}
+
+function clientWith(fetchImpl: typeof fetch): Client {
+  return new Client({ baseUrl: "http://omnigraph.test", fetchImpl });
+}
+
+describe("Client (SDK-backed facade)", () => {
+  it("reshapes a successful /query response to colombo's snake_case ReadOutput", async () => {
+    const client = clientWith(
+      fetchReturning(
+        jsonResponse({
+          query_name: "q",
+          target: { branch: "main", snapshot: null },
+          row_count: 2,
+          columns: ["slug"],
+          rows: [{ slug: "a" }, { slug: "b" }],
+        }),
+      ),
+    );
+    const out = await client.query({ query: "query q() { ... }", name: "q" });
+    expect(out).toEqual({
+      query_name: "q",
+      target: "main",
+      row_count: 2,
+      columns: ["slug"],
+      rows: [{ slug: "a" }, { slug: "b" }],
+    });
+  });
+
+  it("defaults columns/rows and coerces a snapshot target to a string", async () => {
+    const client = clientWith(
+      fetchReturning(
+        jsonResponse({
+          query_name: "q",
+          target: { branch: null, snapshot: "snap-1" },
+          row_count: 0,
+        }),
+      ),
+    );
+    const out = await client.query({ query: "q" });
+    expect(out.target).toBe("snap-1");
+    expect(out.columns).toEqual([]);
+    expect(out.rows).toEqual([]);
+  });
+
+  it("reshapes /mutate and omits actor_id when null", async () => {
+    const client = clientWith(
+      fetchReturning(
+        jsonResponse({
+          branch: "review",
+          query_name: "m",
+          affected_nodes: 1,
+          affected_edges: 0,
+          actor_id: null,
+        }),
+      ),
+    );
+    const out = await client.mutate({ query: "update ...", name: "m" });
+    expect(out).toEqual({
+      branch: "review",
+      query_name: "m",
+      affected_nodes: 1,
+      affected_edges: 0,
+    });
+  });
+
+  it("wraps a 401 as OmnigraphHttpError matching the permission classifier", async () => {
+    const client = clientWith(
+      fetchReturning(jsonResponse({ error: "bad token", code: "unauthorized" }, 401)),
+    );
+    const err = await client.query({ query: "q" }).catch((e) => e);
+    expect(err).toBeInstanceOf(OmnigraphHttpError);
+    expect(err.message).toMatch(/returned 401/);
+    expect(err.message).toMatch(RE_UNAUTHORIZED);
+    expect(err.message).toMatch(/"code"\s*:\s*"unauthorized"/);
+  });
+
+  it("wraps a network failure so the network classifier fires", async () => {
+    const client = clientWith(
+      (async () => {
+        throw new TypeError("Failed to fetch");
+      }) as unknown as typeof fetch,
+    );
+    const err = await client.query({ query: "q" }).catch((e) => e);
+    expect(err).toBeInstanceOf(OmnigraphHttpError);
+    expect(err.message).toMatch(RE_NETWORK);
+  });
+
+  it("preserves the server error text so the conflict classifier fires", async () => {
+    const client = clientWith(
+      fetchReturning(
+        jsonResponse(
+          {
+            error:
+              "storage: stale view of table nodes/PolicyClause expected manifest table version 5",
+            code: "conflict",
+          },
+          409,
+        ),
+      ),
+    );
+    const err = await client.mutate({ query: "update ..." }).catch((e) => e);
+    expect(err).toBeInstanceOf(OmnigraphHttpError);
+    expect(err.message).toMatch(RE_CONFLICT);
+  });
+});

packages/client/src/http.ts

@@ -1,19 +1,33 @@
 /**
- * Thin HTTP client for omnigraph-server. Maps directly onto the OpenAPI
- * surface — no translation, no business logic. The `ServerSource` wrapper
- * on top translates fixture-DSL queries and MutationSpec into `.gq`.
+ * Thin facade over the official omnigraph SDK (`@modernrelay/omnigraph`).
+ *
+ * The SDK owns the HTTP transport, the OpenAPI-faithful types, and typed
+ * error classes. This `Client` keeps colombo's stable, snake_case surface
+ * (`query`/`mutate`/`branches`/`healthz` + `OmnigraphHttpError`) so the
+ * `ServerSource` adapter, its tests, and the web error-classifier are
+ * unaffected by the SDK swap. It reshapes the SDK's camelCase responses
+ * back to colombo's shapes and re-wraps thrown SDK errors as
+ * `OmnigraphHttpError` to preserve the message contract the UI matches on.
  */
 
+import {
+  Omnigraph,
+  NetworkError,
+  OmnigraphError,
+  type QueryInput as SdkQueryInput,
+  type MutationInput as SdkMutationInput,
+} from "@modernrelay/omnigraph";
+
 export interface ClientOptions {
   baseUrl: string;
   /** Bearer token. Falls back to `OMNIGRAPH_TOKEN` env var when unset. */
   token?: string;
   fetchImpl?: typeof fetch;
 }
 
-export interface ReadInput {
-  query_source: string;
-  query_name?: string;
+export interface QueryInput {
+  query: string;
+  name?: string;
   params?: Record<string, unknown>;
   branch?: string;
   snapshot?: string;
@@ -27,9 +41,9 @@ export interface ReadOutput {
   rows: Record<string, unknown>[];
 }
 
-export interface ChangeInput {
-  query_source: string;
-  query_name?: string;
+export interface MutateInput {
+  query: string;
+  name?: string;
   params?: Record<string, unknown>;
   branch?: string;
 }
@@ -58,68 +72,99 @@ export class OmnigraphHttpError extends Error {
 }
 
 export class Client {
-  private readonly baseUrl: string;
-  private readonly token: string | undefined;
-  private readonly fetchImpl: typeof fetch;
+  private readonly og: Omnigraph;
 
   constructor(opts: ClientOptions) {
-    this.baseUrl = opts.baseUrl.replace(/\/$/, "");
-    this.token = opts.token ?? process.env.OMNIGRAPH_TOKEN;
-    // Bind to globalThis: in browsers, calling `fetch` with `this`
-    // pointing at a non-Window receiver throws "Illegal invocation".
-    // Storing the global as a class member makes `this.fetchImpl(...)`
-    // call it as a method on the Client instance unless we bind it
-    // explicitly here. Node 18+'s fetch is permissive and worked
-    // without this, which is why the TUI never tripped.
-    this.fetchImpl = opts.fetchImpl ?? fetch.bind(globalThis);
+    const token =
+      opts.token ??
+      process.env.OMNIGRAPH_TOKEN ??
+      process.env.OMNIGRAPH_BEARER_TOKEN;
+    this.og = new Omnigraph({
+      baseUrl: opts.baseUrl,
+      ...(token !== undefined ? { token } : {}),
+      ...(opts.fetchImpl !== undefined ? { fetch: opts.fetchImpl } : {}),
+    });
   }
 
-  read(body: ReadInput, signal?: AbortSignal): Promise<ReadOutput> {
-    return this.json<ReadOutput>("POST", "/read", body, signal);
+  async query(body: QueryInput, signal?: AbortSignal): Promise<ReadOutput> {
+    try {
+      const r = await this.og.query(
+        body as SdkQueryInput,
+        signal ? { signal } : {},
+      );
+      return {
+        query_name: r.queryName,
+        target: r.target?.branch ?? r.target?.snapshot ?? "main",
+        row_count: r.rowCount,
+        columns: r.columns ?? [],
+        rows: (r.rows ?? []) as Record<string, unknown>[],
+      };
+    } catch (e) {
+      throw toHttpError(e, "/query");
+    }
   }
 
-  /**
-   * One-shot mutation. The server commits exactly one manifest version
-   * for the touched tables (atomic per call; cross-table OCC enforced via
-   * ManifestBatchPublisher CAS). Returns 409 with conflict details on
-   * concurrent-write loss.
-   */
-  change(body: ChangeInput, signal?: AbortSignal): Promise<ChangeOutput> {
-    return this.json<ChangeOutput>("POST", "/change", body, signal);
+  async mutate(body: MutateInput, signal?: AbortSignal): Promise<ChangeOutput> {
+    try {
+      const r = await this.og.mutate(
+        body as SdkMutationInput,
+        signal ? { signal } : {},
+      );
+      return {
+        branch: r.branch,
+        query_name: r.queryName,
+        affected_nodes: r.affectedNodes,
+        affected_edges: r.affectedEdges,
+        ...(r.actorId != null ? { actor_id: r.actorId } : {}),
+      };
+    } catch (e) {
+      throw toHttpError(e, "/mutate");
+    }
   }
 
-  branches(): Promise<BranchListOutput> {
-    return this.json<BranchListOutput>("GET", "/branches");
+  async branches(): Promise<BranchListOutput> {
+    try {
+      return { branches: await this.og.branches.list() };
+    } catch (e) {
+      throw toHttpError(e, "/branches");
+    }
   }
 
   async healthz(): Promise<void> {
-    const res = await this.fetchImpl(`${this.baseUrl}/healthz`, {
-      method: "GET",
-    });
-    if (!res.ok) {
-      throw new OmnigraphHttpError(res.status, "/healthz", await res.text());
+    try {
+      await this.og.health();
+    } catch (e) {
+      throw toHttpError(e, "/healthz");
     }
   }
+}
 
-  private async json<T>(
-    method: "GET" | "POST" | "DELETE",
-    path: string,
-    body?: unknown,
-    signal?: AbortSignal,
-  ): Promise<T> {
-    const headers: Record<string, string> = {};
-    if (this.token) headers["Authorization"] = `Bearer ${this.token}`;
-    if (body !== undefined) headers["Content-Type"] = "application/json";
-    const res = await this.fetchImpl(`${this.baseUrl}${path}`, {
-      method,
-      headers,
-      body: body === undefined ? undefined : JSON.stringify(body),
-      signal,
-    });
-    const text = await res.text();
-    if (!res.ok) {
-      throw new OmnigraphHttpError(res.status, path, text);
-    }
-    return text ? (JSON.parse(text) as T) : (undefined as unknown as T);
+/**
+ * Normalize a thrown SDK error into `OmnigraphHttpError`, preserving the
+ * message format (`omnigraph-server <path> returned <status>: <body>`) and
+ * embedding the server `code` so the web error-classifier's regexes still
+ * fire. AbortError is re-thrown unchanged so the runtime's cancellation
+ * logic still recognizes it.
+ */
+function toHttpError(e: unknown, path: string): unknown {
+  if (e instanceof Error && e.name === "AbortError") return e;
+  if (e instanceof NetworkError) {
+    return new OmnigraphHttpError(
+      0,
+      path,
+      JSON.stringify({ error: `Failed to fetch — ${e.message}` }),
+    );
+  }
+  if (e instanceof OmnigraphError) {
+    return new OmnigraphHttpError(
+      e.status,
+      path,
+      JSON.stringify({
+        error: e.message,
+        ...(e.code ? { code: e.code } : {}),
+      }),
+    );
   }
+  const message = e instanceof Error ? e.message : String(e);
+  return new OmnigraphHttpError(0, path, JSON.stringify({ error: message }));
 }

packages/client/src/index.ts

@@ -2,9 +2,9 @@ export {
   Client,
   OmnigraphHttpError,
   type ClientOptions,
-  type ReadInput,
+  type QueryInput,
   type ReadOutput,
-  type ChangeInput,
+  type MutateInput,
   type ChangeOutput,
   type BranchListOutput,
 } from "./http.js";