Use tagged socket errors in client API

ts/packages/client/src/__tests__/flows-api.test.ts

@@ -1,5 +1,5 @@
 import { describe, it, expect, vi, beforeEach } from "vitest";
-import { FlowsApi } from "../socket/trustgraph-socket";
+import { FlowsApi, TrustGraphSocketError } from "../socket/trustgraph-socket";
 import { FlowResponse } from "../models/messages";
 
 describe("FlowsApi", () => {
@@ -194,6 +194,15 @@ describe("FlowsApi", () => {
       );
       expect(result).toEqual(flowDefinition); // Result should be parsed JSON
     });
+
+    it("rejects invalid flow JSON with a tagged socket error", async () => {
+      const mockResponse: FlowResponse = {
+        flow: "not json",
+      };
+      mockApi.makeRequest.mockResolvedValue(mockResponse);
+
+      await expect(flowsApi.getFlow("broken-flow")).rejects.toBeInstanceOf(TrustGraphSocketError);
+    });
   });
 
   describe("getFlowBlueprint", () => {
@@ -217,5 +226,28 @@ describe("FlowsApi", () => {
       );
       expect(result).toEqual(blueprintDefinition); // Result should be parsed JSON
     });
+
+    it("rejects invalid blueprint JSON with a tagged socket error", async () => {
+      const mockResponse: FlowResponse = {
+        "blueprint-definition": "not json",
+      };
+      mockApi.makeRequest.mockResolvedValue(mockResponse);
+
+      await expect(flowsApi.getFlowBlueprint("broken-blueprint")).rejects.toBeInstanceOf(TrustGraphSocketError);
+    });
+  });
+
+  describe("flow errors", () => {
+    it("rejects start-flow response errors with a tagged socket error", async () => {
+      const mockResponse: FlowResponse = {
+        error: {
+          type: "flow-error",
+          message: "nope",
+        },
+      };
+      mockApi.makeRequest.mockResolvedValue(mockResponse);
+
+      await expect(flowsApi.startFlow("id", "default", "desc")).rejects.toBeInstanceOf(TrustGraphSocketError);
+    });
   });
 });

ts/packages/client/src/socket/trustgraph-socket.ts

@@ -8,7 +8,7 @@ import {
   makeEffectRpcClient,
 } from "./effect-rpc-client.js";
 import { getDefaultSocketUrl, getRandomValues } from "./websocket-adapter.js";
-import { Clock, Effect } from "effect";
+import { Clock, Effect, Option, Result, Schema as S } from "effect";
 
 // Import all message types for different services
 import type {
@@ -127,6 +127,21 @@ function toErrorMessage(value: unknown, fallback: string): string {
   return fallback;
 }
 
+export class TrustGraphSocketError extends S.TaggedErrorClass<TrustGraphSocketError>()(
+  "TrustGraphSocketError",
+  {
+    message: S.String,
+    operation: S.String,
+  },
+) {}
+
+const socketError = (operation: string, message: string): TrustGraphSocketError =>
+  TrustGraphSocketError.make({ operation, message });
+
+function throwSocketError(operation: string, message: string): never {
+  throw socketError(operation, message);
+}
+
 function dispatchOptions(
   timeoutMs: number | undefined,
   retries: number | undefined,
@@ -165,10 +180,12 @@ function streamingMetadataFrom(source: {
 
 function throwIfResponseError(error: ResponseError | undefined): void {
   if (error !== undefined) {
-    throw new Error(error.message);
+    throwSocketError("response-error", error.message);
   }
 }
 
+const decodeJsonUnknown = S.decodeUnknownOption(S.UnknownFromJsonString);
+
 export interface ConfigValueEntry {
   workspace?: string;
   type?: string;
@@ -194,11 +211,15 @@ function asConfigValues(response: unknown): ConfigValueEntry[] {
 
 function parseConfigJson(value: unknown): unknown {
   if (typeof value !== "string") return value;
-  try {
-    return JSON.parse(value);
-  } catch {
-    return value;
+  return Option.getOrElse(decodeJsonUnknown(value), () => value);
+}
+
+function parseResponseJson(value: string | undefined, operation: string): unknown {
+  const parsed = Option.getOrUndefined(decodeJsonUnknown(value ?? "{}"));
+  if (parsed === undefined) {
+    return throwSocketError(operation, "Response JSON could not be decoded");
   }
+  return parsed;
 }
 
 const currentEpochSeconds = (): number =>
@@ -526,10 +547,16 @@ export function makeBaseApi(
   const notifyStateChange = () => {
     const state = getConnectionState();
     connectionStateListeners.forEach((listener) => {
-      try {
-        listener(state);
-      } catch (error) {
-        logClientError("Error in connection state listener", error);
+      const result = Result.try({
+        try: () => listener(state),
+        catch: (error) =>
+          socketError(
+            "connection-state-listener",
+            toErrorMessage(error, "Error in connection state listener"),
+          ),
+      });
+      if (Result.isFailure(result)) {
+        logClientError("Error in connection state listener", result.failure);
       }
     });
   };
@@ -1038,7 +1065,7 @@ export function makeFlowsApi(api: BaseApi) {
             },
             60000,
           )
-          .then((r) => JSON.parse(r.flow ?? "{}")); // Parse JSON flow definition
+          .then((r) => parseResponseJson(r.flow, "get-flow"));
       },
 
 
@@ -1188,7 +1215,7 @@ export function makeFlowsApi(api: BaseApi) {
             },
             60000,
           )
-          .then((r) => JSON.parse(r["blueprint-definition"] ?? "{}"));
+          .then((r) => parseResponseJson(r["blueprint-definition"], "get-blueprint"));
       },
 
 
@@ -1236,7 +1263,7 @@ export function makeFlowsApi(api: BaseApi) {
           .makeRequest<FlowRequest, FlowResponse>("flow", request, 30000)
           .then((response) => {
             if (response.error !== undefined) {
-              throw new Error(toErrorMessage(response.error, "Flow start failed"));
+              throwSocketError("start-flow", toErrorMessage(response.error, "Flow start failed"));
             }
             return response;
           });
@@ -2023,7 +2050,7 @@ export function makeFlowApi(api: BaseApi, flowId: string) {
           )
           .then((r) => {
             if (r.error !== undefined) {
-              throw new Error(r.error.message);
+              throwSocketError("row-embeddings-query", r.error.message);
             }
             return r.matches ?? [];
           });
@@ -2475,7 +2502,7 @@ export function makeCollectionManagementApi(api: BaseApi) {
             ) {
               return r.collections[0];
             }
-            throw new Error("Failed to update collection");
+            return throwSocketError("update-collection", "Failed to update collection");
           });
       },