From 1899bf1f5a583826c7420ea80401e74d74b1b338 Mon Sep 17 00:00:00 2001 From: elpresidank Date: Tue, 2 Jun 2026 02:19:37 -0500 Subject: [PATCH] Use tagged socket errors in client API --- ts/EFFECT_NATIVE_REWRITE_AUDIT.md | 41 ++++++++++--- .../client/src/__tests__/flows-api.test.ts | 34 ++++++++++- .../client/src/socket/trustgraph-socket.ts | 57 ++++++++++++++----- 3 files changed, 108 insertions(+), 24 deletions(-) diff --git a/ts/EFFECT_NATIVE_REWRITE_AUDIT.md b/ts/EFFECT_NATIVE_REWRITE_AUDIT.md index 81eb1f4f..d13be74b 100644 --- a/ts/EFFECT_NATIVE_REWRITE_AUDIT.md +++ b/ts/EFFECT_NATIVE_REWRITE_AUDIT.md @@ -12,8 +12,8 @@ Verified source roots: - Effect v4 subtree: `/home/elpresidank/YeeBois/projects/beep-effect2/.repos/effect-v4` - Installed Effect beta used by this workspace: `ts/node_modules/effect` -Current signal counts from `ts/packages` after the 2026-06-02 Client -WebSocket adapter slice: +Current signal counts from `ts/packages` after the 2026-06-02 Client socket +tagged error slice: | Signal | Count | | --- | ---: | @@ -25,9 +25,9 @@ WebSocket adapter slice: | `makeAsyncProcessor` | 19 | | `receive(` | 18 | | `while (` | 9 | -| `new Error` | 12 | +| `new Error` | 8 | | `new Promise` | 10 | -| `JSON.parse` | 7 | +| `JSON.parse` | 4 | | `localStorage` | 8 | | `JSON.stringify` | 6 | | `setTimeout` | 4 | @@ -54,6 +54,10 @@ Notes: the previous constructor assertions. - The `new Error` count dropped because `websocket-adapter.ts` now throws `S.TaggedErrorClass` adapter errors. +- The latest client socket slice removed the remaining production + `trustgraph-socket.ts` normal `Error`, raw `JSON.parse`, and listener + `try`/`catch` matches. The remaining client socket modernization signal is + the shared `newableFactory` constructor assertion pattern. - `Record` and `throwLibrarianServiceError` are now clean in `ts/packages`. @@ -406,6 +410,30 @@ Notes: - `cd ts && bun run test` - `git diff --check` +### 2026-06-02: Client Socket Tagged Error And JSON Slice + +- Status: migrated and root-verified. +- Completed: + - `ts/packages/client/src/socket/trustgraph-socket.ts` now models socket API + failures with `TrustGraphSocketError` via `S.TaggedErrorClass`. + - Flow/blueprint JSON response parsing now uses Schema decoding through + `S.UnknownFromJsonString` instead of raw `JSON.parse`. + - Token-cost config JSON keeps the previous invalid-string fallback behavior + while decoding through Schema/Option. + - Connection-state listener isolation now uses `Result.try` and typed socket + errors instead of a local `try`/`catch`. + - Flow start, row embeddings, collection update, and response-error + failures now reject with tagged socket errors instead of normal `Error`. + - Flow API tests cover invalid JSON and response-error rejections. +- Verification: + - `bun run --cwd ts/packages/client build` + - `bun run --cwd ts/packages/client test -- src/__tests__/flows-api.test.ts` + - `cd ts && bun run check` + - `bun run --cwd ts/packages/client test` + - `cd ts && bun run build` + - `cd ts && bun run test` + - `git diff --check` + ## Subagent Findings To Preserve - MCP/workbench: @@ -431,8 +459,7 @@ Notes: - Gateway/client: - `EffectRpcClient` now owns its socket/RPC layer with `ManagedRuntime`. Remaining client cleanup should focus on `trustgraph-socket.ts` - higher-level normal `Error` throws/JSON parsing and the client newable - factory assertions. + and `effect-rpc-client.ts` newable factory assertions. - Knowledge streams still duplicate legacy end-of-stream handling. - WebSocket adapter host fallbacks now use `Result.try` and tagged adapter errors while preserving sync exports. @@ -461,8 +488,6 @@ Notes: - `EffectRpcClient` is now an internal managed runtime with Promise compatibility facades. - Expose Promise-returning methods through a thin adapter. - - Finish replacing remaining normal client `Error` constructors with tagged - errors before they cross into shared Effect code. - Replace remaining client newable factory assertions with a typed factory shape that preserves current constructor/function compatibility. - Tests: diff --git a/ts/packages/client/src/__tests__/flows-api.test.ts b/ts/packages/client/src/__tests__/flows-api.test.ts index e011af72..5321255f 100644 --- a/ts/packages/client/src/__tests__/flows-api.test.ts +++ b/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); + }); }); }); diff --git a/ts/packages/client/src/socket/trustgraph-socket.ts b/ts/packages/client/src/socket/trustgraph-socket.ts index c68adbcb..3d3183f2 100644 --- a/ts/packages/client/src/socket/trustgraph-socket.ts +++ b/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", + { + 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("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"); }); },