feat: expose workflow console api and RPC service with @effect/rpc
- Replace manual JSON-RPC protocol handling with @effect/rpc typed RPCs - Add evidence capture layer (per-call Ref, response body buffering/reconstruction) - Add service module with typed dispatch and layer composition via Layer.mergeAll - Add Hono API with /api/health, /api/connect, /api/rpc routes - Add browser DTO contracts, error mapping, body size limits - Add 13 Hono route tests and 12 target-policy tests passing - Delete old protocol.ts and protocol.test.ts (replaced by rpcs.ts/service.ts)
This commit is contained in:
@@ -14,6 +14,8 @@
|
||||
"typecheck": "tsc -p tsconfig.json --noEmit"
|
||||
},
|
||||
"dependencies": {
|
||||
"@effect/platform": "0.96.2",
|
||||
"@effect/rpc": "0.75.1",
|
||||
"effect": "3.21.4"
|
||||
},
|
||||
"devDependencies": {
|
||||
|
||||
@@ -0,0 +1,89 @@
|
||||
import { Context, Effect, Layer, Ref } from "effect";
|
||||
import { HttpClient, HttpClientRequest, HttpClientResponse } from "@effect/platform";
|
||||
|
||||
export type EvidenceRecord = {
|
||||
readonly request: {
|
||||
readonly url: string;
|
||||
readonly method: string;
|
||||
readonly body: unknown;
|
||||
};
|
||||
readonly response: { readonly status: number; readonly body: unknown };
|
||||
};
|
||||
|
||||
export const EvidenceRef = Context.GenericTag<Ref.Ref<EvidenceRecord | null>>(
|
||||
"EvidenceRef",
|
||||
);
|
||||
|
||||
export const makeEvidenceLayer = Layer.sync(
|
||||
EvidenceRef,
|
||||
() => Ref.unsafeMake<EvidenceRecord | null>(null),
|
||||
);
|
||||
|
||||
const readRequestBody = (
|
||||
request: HttpClientRequest.HttpClientRequest,
|
||||
): unknown => {
|
||||
const body = request.body;
|
||||
if (body._tag === "Uint8Array") {
|
||||
try {
|
||||
return JSON.parse(new TextDecoder().decode(body.body));
|
||||
} catch {
|
||||
return body.body;
|
||||
}
|
||||
}
|
||||
if (body._tag === "Raw") {
|
||||
return body.body;
|
||||
}
|
||||
return null;
|
||||
};
|
||||
|
||||
/**
|
||||
* Wrap an HttpClient to capture raw request/response evidence per-call.
|
||||
*
|
||||
* Buffers the response body once, records it for the evidence drawer,
|
||||
* then reconstructs the response so RpcClient can still read it.
|
||||
*/
|
||||
export const withEvidenceCapture = <E, R>(
|
||||
client: HttpClient.HttpClient.With<E, R>,
|
||||
ref: Ref.Ref<EvidenceRecord | null>,
|
||||
): HttpClient.HttpClient.With<E, R> =>
|
||||
client.pipe(
|
||||
HttpClient.transform((responseEffect, request) =>
|
||||
Effect.gen(function* () {
|
||||
const response = yield* responseEffect;
|
||||
|
||||
// Buffer the body text once so we can record it AND reconstruct the response
|
||||
const bodyText = yield* Effect.catchAll(
|
||||
response.text,
|
||||
() => Effect.succeed(""),
|
||||
);
|
||||
|
||||
let responseBody: unknown;
|
||||
try {
|
||||
responseBody = JSON.parse(bodyText);
|
||||
} catch {
|
||||
responseBody = bodyText;
|
||||
}
|
||||
|
||||
yield* Ref.set(ref, {
|
||||
request: {
|
||||
url: request.url,
|
||||
method: request.method,
|
||||
body: readRequestBody(request),
|
||||
},
|
||||
response: { status: response.status, body: responseBody },
|
||||
});
|
||||
|
||||
// Reconstruct a fresh response from the buffered text so downstream
|
||||
// consumers (RpcClient) can still read the body.
|
||||
return HttpClientResponse.fromWeb(
|
||||
request,
|
||||
new Response(bodyText, {
|
||||
status: response.status,
|
||||
headers: new Headers(
|
||||
Object.entries(response.headers as Record<string, string>),
|
||||
),
|
||||
}),
|
||||
);
|
||||
}),
|
||||
),
|
||||
);
|
||||
@@ -11,11 +11,20 @@ export {
|
||||
|
||||
export { normalizeLoopbackTarget } from "./target-policy.js";
|
||||
|
||||
export { decodeRpcResponse } from "./protocol.js";
|
||||
export { WorkflowHealth, WorkflowSourcesList, WorkflowRpcs } from "./rpcs.js";
|
||||
|
||||
export type {
|
||||
JsonRpcRequest,
|
||||
JsonRpcResponse,
|
||||
JsonRpcSuccess,
|
||||
JsonRpcFailure,
|
||||
} from "./protocol.js";
|
||||
export { WorkflowRpc, makeWorkflowRpcLayer } from "./service.js";
|
||||
export type { OperationExchange, WorkflowRpcError, OperationName } from "./service.js";
|
||||
|
||||
export {
|
||||
getOperationMeta,
|
||||
listOperations,
|
||||
} from "./method-registry.js";
|
||||
export type { OperationMeta } from "./method-registry.js";
|
||||
|
||||
export {
|
||||
EvidenceRef,
|
||||
makeEvidenceLayer,
|
||||
withEvidenceCapture,
|
||||
} from "./evidence.js";
|
||||
export type { EvidenceRecord } from "./evidence.js";
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { Effect, Ref } from "effect";
|
||||
import { RpcClient, RpcGroup, RpcSerialization } from "@effect/rpc";
|
||||
import { FetchHttpClient, HttpClient } from "@effect/platform";
|
||||
import { WorkflowHealth, WorkflowSourcesList } from "./rpcs.js";
|
||||
import { withEvidenceCapture, type EvidenceRecord } from "./evidence.js";
|
||||
|
||||
const LIVE = process.env.LIVE_PYTHON_SERVER === "1";
|
||||
const describeLive = LIVE ? describe : describe.skip;
|
||||
|
||||
const TARGET = "http://127.0.0.1:8765/rpc";
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
const run = <A>(eff: Effect.Effect<A, any, any>): Promise<A> =>
|
||||
Effect.runPromise(eff as Effect.Effect<A, never, never>);
|
||||
|
||||
describeLive("interop: live Python server", () => {
|
||||
it("workflow.health returns ok", async () => {
|
||||
const group = RpcGroup.make(WorkflowHealth);
|
||||
const result = await run(
|
||||
Effect.gen(function* () {
|
||||
const client = yield* RpcClient.make(group as any).pipe(
|
||||
Effect.provide(RpcSerialization.layerJsonRpc()),
|
||||
Effect.provide(FetchHttpClient.layer),
|
||||
Effect.provide(
|
||||
RpcClient.layerProtocolHttp({ url: TARGET } as any),
|
||||
),
|
||||
);
|
||||
return yield* (client as any)["workflow.health"]({});
|
||||
}).pipe(
|
||||
Effect.scoped,
|
||||
Effect.provide(FetchHttpClient.layer),
|
||||
Effect.provide(RpcSerialization.layerJsonRpc()),
|
||||
),
|
||||
);
|
||||
expect(result).toEqual({
|
||||
status: "ok",
|
||||
store_root: expect.any(String),
|
||||
});
|
||||
});
|
||||
|
||||
it("workflow.sources.list returns paginated results", async () => {
|
||||
const group = RpcGroup.make(WorkflowSourcesList);
|
||||
const result = await run(
|
||||
Effect.gen(function* () {
|
||||
const client = yield* RpcClient.make(group as any).pipe(
|
||||
Effect.provide(RpcSerialization.layerJsonRpc()),
|
||||
Effect.provide(FetchHttpClient.layer),
|
||||
Effect.provide(
|
||||
RpcClient.layerProtocolHttp({ url: TARGET } as any),
|
||||
),
|
||||
);
|
||||
return yield* (client as any)["workflow.sources.list"]({
|
||||
limit: 10,
|
||||
});
|
||||
}).pipe(
|
||||
Effect.scoped,
|
||||
Effect.provide(FetchHttpClient.layer),
|
||||
Effect.provide(RpcSerialization.layerJsonRpc()),
|
||||
),
|
||||
);
|
||||
const typed = result as {
|
||||
sources: unknown[];
|
||||
next_cursor: string | null;
|
||||
total: number;
|
||||
};
|
||||
expect(typed.sources).toBeInstanceOf(Array);
|
||||
expect(typeof typed.total).toBe("number");
|
||||
});
|
||||
|
||||
it("handles standard JSON-RPC errors gracefully", async () => {
|
||||
const group = RpcGroup.make(WorkflowSourcesList);
|
||||
const exit = await Effect.runPromiseExit(
|
||||
Effect.gen(function* () {
|
||||
const client = yield* RpcClient.make(group as any).pipe(
|
||||
Effect.provide(RpcSerialization.layerJsonRpc()),
|
||||
Effect.provide(FetchHttpClient.layer),
|
||||
Effect.provide(
|
||||
RpcClient.layerProtocolHttp({ url: TARGET } as any),
|
||||
),
|
||||
);
|
||||
return yield* (client as any)["workflow.sources.list"]({
|
||||
cursor: "nonexistent",
|
||||
});
|
||||
}).pipe(
|
||||
Effect.scoped,
|
||||
Effect.provide(FetchHttpClient.layer),
|
||||
Effect.provide(RpcSerialization.layerJsonRpc()),
|
||||
) as Effect.Effect<unknown, never, never>,
|
||||
);
|
||||
expect(exit._tag).toBe("Success");
|
||||
});
|
||||
|
||||
it("raw evidence capture works", async () => {
|
||||
const evidenceRef = Effect.runSync(Ref.make<EvidenceRecord | null>(null));
|
||||
|
||||
const group = RpcGroup.make(WorkflowHealth);
|
||||
const result = await run(
|
||||
Effect.gen(function* () {
|
||||
const httpClient = yield* HttpClient.HttpClient;
|
||||
const transformed = withEvidenceCapture(httpClient, evidenceRef);
|
||||
|
||||
const protocolLayer = RpcClient.layerProtocolHttp({
|
||||
url: TARGET,
|
||||
transformClient: () => transformed,
|
||||
} as any);
|
||||
|
||||
const client = yield* RpcClient.make(group as any).pipe(
|
||||
Effect.provide(RpcSerialization.layerJsonRpc()),
|
||||
Effect.provide(FetchHttpClient.layer),
|
||||
Effect.provide(protocolLayer),
|
||||
);
|
||||
|
||||
return yield* (client as any)["workflow.health"]({});
|
||||
}).pipe(
|
||||
Effect.scoped,
|
||||
Effect.provide(FetchHttpClient.layer),
|
||||
Effect.provide(RpcSerialization.layerJsonRpc()),
|
||||
),
|
||||
);
|
||||
|
||||
const evidence = Effect.runSync(Ref.get(evidenceRef));
|
||||
expect(evidence).not.toBeNull();
|
||||
expect(evidence!.request.url).toContain("/rpc");
|
||||
expect(evidence!.response.status).toBe(200);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,45 @@
|
||||
export type OperationMeta = {
|
||||
readonly method: string;
|
||||
readonly label: string;
|
||||
readonly explanation: string;
|
||||
readonly idempotency: "read";
|
||||
readonly equivalentCli: (params: unknown) => string;
|
||||
readonly interpret: (result: unknown) => unknown;
|
||||
};
|
||||
|
||||
const registry: ReadonlyMap<string, OperationMeta> = new Map([
|
||||
[
|
||||
"workflow.health",
|
||||
{
|
||||
method: "workflow.health",
|
||||
label: "Health check",
|
||||
explanation: "Check if the workflow server is running",
|
||||
idempotency: "read",
|
||||
equivalentCli: () => "uv run wf status",
|
||||
interpret: (result) => result,
|
||||
},
|
||||
],
|
||||
[
|
||||
"workflow.sources.list",
|
||||
{
|
||||
method: "workflow.sources.list",
|
||||
label: "List sources",
|
||||
explanation: "List registered data sources with pagination",
|
||||
idempotency: "read",
|
||||
equivalentCli: (params) => {
|
||||
const p = params as { cursor?: string; limit?: number };
|
||||
const parts = ["uv run wf source list"];
|
||||
if (p.limit != null) parts.push(`--limit ${p.limit}`);
|
||||
if (p.cursor != null) parts.push(`--cursor ${p.cursor}`);
|
||||
return parts.join(" ");
|
||||
},
|
||||
interpret: (result) => result,
|
||||
},
|
||||
],
|
||||
]);
|
||||
|
||||
export const getOperationMeta = (method: string): OperationMeta | undefined =>
|
||||
registry.get(method);
|
||||
|
||||
export const listOperations = (): ReadonlyArray<OperationMeta> =>
|
||||
Array.from(registry.values());
|
||||
@@ -1,84 +0,0 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { RpcDecodeError, RpcProtocolError, decodeRpcResponse } from "./protocol.js";
|
||||
|
||||
describe("decodeRpcResponse", () => {
|
||||
it("decodes success envelope with matching string id", () => {
|
||||
const payload = {
|
||||
jsonrpc: "2.0",
|
||||
id: "req-1",
|
||||
result: { status: "ok" },
|
||||
};
|
||||
expect(decodeRpcResponse(payload, "req-1")).toEqual({
|
||||
jsonrpc: "2.0",
|
||||
id: "req-1",
|
||||
result: { status: "ok" },
|
||||
});
|
||||
});
|
||||
|
||||
it("decodes error envelope with matching string id", () => {
|
||||
const payload = {
|
||||
jsonrpc: "2.0",
|
||||
id: "req-1",
|
||||
error: { code: -32000, message: "server error" },
|
||||
};
|
||||
const response = decodeRpcResponse(payload, "req-1");
|
||||
expect(response).toEqual({
|
||||
jsonrpc: "2.0",
|
||||
id: "req-1",
|
||||
error: { code: -32000, message: "server error" },
|
||||
});
|
||||
});
|
||||
|
||||
it("rejects mismatched id", () => {
|
||||
const payload = {
|
||||
jsonrpc: "2.0",
|
||||
id: "other",
|
||||
result: { status: "ok" },
|
||||
};
|
||||
expect(() => decodeRpcResponse(payload, "req-1")).toThrow(
|
||||
RpcProtocolError,
|
||||
);
|
||||
});
|
||||
|
||||
it("rejects envelope with both result and error", () => {
|
||||
const payload = {
|
||||
jsonrpc: "2.0",
|
||||
id: "req-1",
|
||||
result: { status: "ok" },
|
||||
error: { code: -32000, message: "err" },
|
||||
};
|
||||
expect(() => decodeRpcResponse(payload, "req-1")).toThrow(RpcDecodeError);
|
||||
});
|
||||
|
||||
it("rejects envelope with neither result nor error", () => {
|
||||
const payload = {
|
||||
jsonrpc: "2.0",
|
||||
id: "req-1",
|
||||
};
|
||||
expect(() => decodeRpcResponse(payload, "req-1")).toThrow(RpcDecodeError);
|
||||
});
|
||||
|
||||
it("rejects wrong jsonrpc version", () => {
|
||||
const payload = {
|
||||
jsonrpc: "1.0",
|
||||
id: "req-1",
|
||||
result: {},
|
||||
};
|
||||
expect(() => decodeRpcResponse(payload, "req-1")).toThrow(RpcDecodeError);
|
||||
});
|
||||
|
||||
it("rejects malformed error object", () => {
|
||||
const payload = {
|
||||
jsonrpc: "2.0",
|
||||
id: "req-1",
|
||||
error: { message: "missing code" },
|
||||
};
|
||||
expect(() => decodeRpcResponse(payload, "req-1")).toThrow(RpcDecodeError);
|
||||
});
|
||||
|
||||
it("rejects non-object payload", () => {
|
||||
expect(() => decodeRpcResponse("not an object", "req-1")).toThrow(
|
||||
RpcDecodeError,
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -1,113 +0,0 @@
|
||||
import { Schema } from "effect";
|
||||
import { RpcDecodeError, RpcProtocolError } from "./errors.js";
|
||||
|
||||
export { RpcDecodeError, RpcProtocolError };
|
||||
|
||||
const JsonRpcVersion = Schema.Literal("2.0");
|
||||
|
||||
const JsonRpcErrorObject = Schema.Struct({
|
||||
code: Schema.Number,
|
||||
message: Schema.String,
|
||||
data: Schema.optional(Schema.String),
|
||||
});
|
||||
|
||||
const decodeJsonRpcVersion = Schema.decodeUnknownSync(JsonRpcVersion);
|
||||
const decodeErrorObject = Schema.decodeUnknownSync(JsonRpcErrorObject);
|
||||
|
||||
export type JsonRpcRequest = {
|
||||
readonly jsonrpc: "2.0";
|
||||
readonly id: string;
|
||||
readonly method: string;
|
||||
readonly params: unknown;
|
||||
};
|
||||
|
||||
export type JsonRpcSuccess = {
|
||||
readonly jsonrpc: "2.0";
|
||||
readonly id: string;
|
||||
readonly result: unknown;
|
||||
};
|
||||
|
||||
export type JsonRpcFailure = {
|
||||
readonly jsonrpc: "2.0";
|
||||
readonly id: string;
|
||||
readonly error: {
|
||||
readonly code: number;
|
||||
readonly message: string;
|
||||
readonly data?: string;
|
||||
};
|
||||
};
|
||||
|
||||
export type JsonRpcResponse = JsonRpcSuccess | JsonRpcFailure;
|
||||
|
||||
function throwRpcDecode(message: string): never {
|
||||
throw new RpcDecodeError({ message });
|
||||
}
|
||||
|
||||
export function decodeRpcResponse(
|
||||
value: unknown,
|
||||
expectedId: string,
|
||||
): JsonRpcResponse {
|
||||
if (typeof value !== "object" || value === null) {
|
||||
throwRpcDecode("response must be a JSON object");
|
||||
}
|
||||
|
||||
const obj = value as Record<string, unknown>;
|
||||
|
||||
if (!("jsonrpc" in obj) || !("id" in obj)) {
|
||||
throwRpcDecode("response must contain 'jsonrpc' and 'id' fields");
|
||||
}
|
||||
|
||||
try {
|
||||
decodeJsonRpcVersion(obj.jsonrpc, { onExcessProperty: "error" });
|
||||
} catch {
|
||||
throwRpcDecode(`jsonrpc version must be "2.0", got ${JSON.stringify(obj.jsonrpc)}`);
|
||||
}
|
||||
|
||||
if (typeof obj.id !== "string") {
|
||||
throwRpcDecode(`response id must be a string, got ${typeof obj.id}`);
|
||||
}
|
||||
|
||||
if (obj.id !== expectedId) {
|
||||
throw new RpcProtocolError({
|
||||
message: `response id "${obj.id}" does not match expected id "${expectedId}"`,
|
||||
evidence: JSON.stringify(obj),
|
||||
});
|
||||
}
|
||||
|
||||
const hasResult = "result" in obj;
|
||||
const hasError = "error" in obj;
|
||||
|
||||
if (hasResult && hasError) {
|
||||
throwRpcDecode("response must not contain both 'result' and 'error'");
|
||||
}
|
||||
|
||||
if (!hasResult && !hasError) {
|
||||
throwRpcDecode("response must contain exactly one of 'result' or 'error'");
|
||||
}
|
||||
|
||||
if (hasResult) {
|
||||
return { jsonrpc: "2.0", id: obj.id, result: obj.result };
|
||||
}
|
||||
|
||||
if (typeof obj.error !== "object" || obj.error === null) {
|
||||
throwRpcDecode("'error' must be an object");
|
||||
}
|
||||
|
||||
let errorObj: { readonly code: number; readonly message: string; readonly data?: string | undefined };
|
||||
try {
|
||||
errorObj = decodeErrorObject(obj.error, { onExcessProperty: "error" });
|
||||
} catch (e) {
|
||||
const msg = e instanceof Error ? e.message : String(e);
|
||||
throwRpcDecode(`invalid error object: ${msg}`);
|
||||
}
|
||||
|
||||
const error: { code: number; message: string; data?: string } = {
|
||||
code: errorObj.code,
|
||||
message: errorObj.message,
|
||||
};
|
||||
if (errorObj.data !== undefined) {
|
||||
error.data = errorObj.data;
|
||||
}
|
||||
|
||||
return { jsonrpc: "2.0", id: obj.id, error };
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
import { Rpc, RpcGroup } from "@effect/rpc";
|
||||
import { Schema } from "effect";
|
||||
|
||||
const SourceSummarySchema = Schema.Struct({
|
||||
id: Schema.String,
|
||||
kind: Schema.String,
|
||||
enabled: Schema.Boolean,
|
||||
description: Schema.NullOr(Schema.String),
|
||||
tool_count: Schema.Number,
|
||||
node_spec_count: Schema.Number,
|
||||
reducer_count: Schema.Number,
|
||||
prompt_count: Schema.Number,
|
||||
resource_count: Schema.Number,
|
||||
});
|
||||
|
||||
export const WorkflowHealth = Rpc.make("workflow.health", {
|
||||
payload: Schema.Struct({}),
|
||||
success: Schema.Struct({
|
||||
status: Schema.Literal("ok"),
|
||||
store_root: Schema.String,
|
||||
}),
|
||||
error: Schema.Never,
|
||||
});
|
||||
|
||||
export const WorkflowSourcesList = Rpc.make("workflow.sources.list", {
|
||||
payload: Schema.Struct({
|
||||
cursor: Schema.optional(Schema.String),
|
||||
limit: Schema.optional(
|
||||
Schema.Number.pipe(Schema.greaterThan(0), Schema.lessThan(101)),
|
||||
),
|
||||
}),
|
||||
success: Schema.Struct({
|
||||
sources: Schema.Array(SourceSummarySchema),
|
||||
next_cursor: Schema.NullOr(Schema.String),
|
||||
total: Schema.Number,
|
||||
}),
|
||||
error: Schema.Never,
|
||||
});
|
||||
|
||||
export const WorkflowRpcs = RpcGroup.make(WorkflowHealth, WorkflowSourcesList);
|
||||
@@ -0,0 +1,262 @@
|
||||
import { Context, Effect, Layer, Ref } from "effect";
|
||||
import { RpcClient, RpcGroup, RpcSerialization } from "@effect/rpc";
|
||||
import { FetchHttpClient, HttpClient } from "@effect/platform";
|
||||
import { normalizeLoopbackTarget } from "./target-policy.js";
|
||||
import { getOperationMeta } from "./method-registry.js";
|
||||
import { WorkflowHealth, WorkflowSourcesList, WorkflowRpcs } from "./rpcs.js";
|
||||
import { withEvidenceCapture, type EvidenceRecord } from "./evidence.js";
|
||||
import {
|
||||
InvalidTargetError,
|
||||
UnknownOperationError,
|
||||
UpstreamConnectionError,
|
||||
UpstreamTimeoutError,
|
||||
RpcProtocolError,
|
||||
RpcRemoteError,
|
||||
} from "./errors.js";
|
||||
|
||||
export interface OperationExchange {
|
||||
readonly operation: string;
|
||||
readonly label: string;
|
||||
readonly interpreted: unknown;
|
||||
readonly exchange: { readonly request: unknown; readonly response: unknown };
|
||||
readonly equivalentCli: string;
|
||||
readonly durationMs: number;
|
||||
}
|
||||
|
||||
export type WorkflowRpcError =
|
||||
| InvalidTargetError
|
||||
| UnknownOperationError
|
||||
| UpstreamConnectionError
|
||||
| UpstreamTimeoutError
|
||||
| RpcProtocolError
|
||||
| RpcRemoteError;
|
||||
|
||||
export type OperationName = "workflow.health" | "workflow.sources.list";
|
||||
|
||||
const isOperationName = (s: string): s is OperationName =>
|
||||
s === "workflow.health" || s === "workflow.sources.list";
|
||||
|
||||
const rpcsByTag = new Map(
|
||||
[WorkflowHealth, WorkflowSourcesList].map((r) => [r._tag, r] as const),
|
||||
);
|
||||
|
||||
const interpretResult = (operation: string, result: unknown): unknown => {
|
||||
const meta = getOperationMeta(operation);
|
||||
return meta ? meta.interpret(result) : result;
|
||||
};
|
||||
|
||||
/**
|
||||
* Map an Effect failure cause to our domain errors.
|
||||
*
|
||||
* Handles both Effect-native errors (RequestError, ResponseError) and
|
||||
* foreign JSON-RPC errors from the Python server (which lack Effect's
|
||||
* Cause shape). For foreign errors, we read the captured raw response
|
||||
* evidence to extract the JSON-RPC error object.
|
||||
*/
|
||||
const mapCauseToError = (
|
||||
cause: unknown,
|
||||
evidence: EvidenceRecord | null,
|
||||
): WorkflowRpcError => {
|
||||
if (
|
||||
cause &&
|
||||
typeof cause === "object" &&
|
||||
"_tag" in cause &&
|
||||
typeof cause._tag === "string"
|
||||
) {
|
||||
const tag = cause._tag;
|
||||
if (tag === "RequestError" || tag === "HttpClientError") {
|
||||
const err = "error" in cause ? cause.error : undefined;
|
||||
const msg = err instanceof Error ? err.message : String(cause);
|
||||
if (msg.toLowerCase().includes("timeout")) {
|
||||
return new UpstreamTimeoutError({ message: msg });
|
||||
}
|
||||
return new UpstreamConnectionError({ message: msg });
|
||||
}
|
||||
if (tag === "ResponseError") {
|
||||
const err = "error" in cause ? cause.error : undefined;
|
||||
const msg = err instanceof Error ? err.message : String(cause);
|
||||
return new UpstreamConnectionError({ message: msg });
|
||||
}
|
||||
}
|
||||
|
||||
// Foreign JSON-RPC error: extract from captured raw response evidence
|
||||
if (evidence?.response?.body) {
|
||||
const body = evidence.response.body;
|
||||
if (typeof body === "object" && body !== null && "error" in body) {
|
||||
const rpcErr = (body as { error: unknown }).error;
|
||||
if (typeof rpcErr === "object" && rpcErr !== null) {
|
||||
const errObj = rpcErr as {
|
||||
message?: unknown;
|
||||
code?: unknown;
|
||||
data?: unknown;
|
||||
};
|
||||
return new RpcRemoteError({
|
||||
message: String(errObj.message ?? "remote error"),
|
||||
code: Number(errObj.code ?? -1),
|
||||
...(errObj.data != null ? { data: String(errObj.data) } : {}),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const msg = cause instanceof Error ? cause.message : String(cause);
|
||||
return new RpcProtocolError({ message: msg });
|
||||
};
|
||||
|
||||
export const WorkflowRpc = Context.GenericTag<{
|
||||
readonly execute: (
|
||||
operation: string,
|
||||
target: string,
|
||||
params: unknown,
|
||||
) => Effect.Effect<OperationExchange, WorkflowRpcError>;
|
||||
}>("WorkflowRpc");
|
||||
|
||||
/**
|
||||
* Dispatch an RPC call through a typed client.
|
||||
*
|
||||
* The `client` is the result of `RpcClient.make(group)` which has typed
|
||||
* methods like `client["workflow.health"]({})`. We use a dispatch map
|
||||
* to avoid unsafe casts.
|
||||
*/
|
||||
const dispatchRpc = (
|
||||
client: WorkflowRpcsClient,
|
||||
operation: OperationName,
|
||||
params: unknown,
|
||||
): Effect.Effect<unknown> => {
|
||||
switch (operation) {
|
||||
case "workflow.health":
|
||||
return client["workflow.health"](params as Record<string, never>);
|
||||
case "workflow.sources.list":
|
||||
return client["workflow.sources.list"](
|
||||
params as { cursor?: string; limit?: number },
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
// The typed client shape produced by RpcClient.make(WorkflowRpcs)
|
||||
type WorkflowRpcsClient = {
|
||||
readonly "workflow.health": (
|
||||
input: Record<string, never>,
|
||||
) => Effect.Effect<{ readonly status: "ok"; readonly store_root: string }>;
|
||||
readonly "workflow.sources.list": (input: {
|
||||
cursor?: string;
|
||||
limit?: number;
|
||||
}) => Effect.Effect<{
|
||||
readonly sources: ReadonlyArray<{
|
||||
readonly id: string;
|
||||
readonly kind: string;
|
||||
readonly enabled: boolean;
|
||||
readonly description: string | null;
|
||||
readonly tool_count: number;
|
||||
readonly node_spec_count: number;
|
||||
readonly reducer_count: number;
|
||||
readonly prompt_count: number;
|
||||
readonly resource_count: number;
|
||||
}>;
|
||||
readonly next_cursor: string | null;
|
||||
readonly total: number;
|
||||
}>;
|
||||
};
|
||||
|
||||
const executeImpl = (
|
||||
operation: string,
|
||||
target: string,
|
||||
params: unknown,
|
||||
): Effect.Effect<OperationExchange, WorkflowRpcError> => {
|
||||
let normalizedTarget: string;
|
||||
try {
|
||||
normalizedTarget = normalizeLoopbackTarget(target);
|
||||
} catch (e) {
|
||||
return Effect.fail(
|
||||
e instanceof InvalidTargetError
|
||||
? e
|
||||
: new InvalidTargetError({
|
||||
message: e instanceof Error ? e.message : String(e),
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
const meta = getOperationMeta(operation);
|
||||
if (!meta) {
|
||||
return Effect.fail(
|
||||
new UnknownOperationError({
|
||||
message: `unknown operation: ${operation}`,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
if (!isOperationName(operation)) {
|
||||
return Effect.fail(
|
||||
new UnknownOperationError({
|
||||
message: `unsupported operation: ${operation}`,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
const evidenceRef = Ref.unsafeMake<EvidenceRecord | null>(null);
|
||||
|
||||
return Effect.gen(function* () {
|
||||
const startTime = Date.now();
|
||||
|
||||
const rpcDef = rpcsByTag.get(operation)!;
|
||||
const group = RpcGroup.make(rpcDef);
|
||||
|
||||
const rpcClient = yield* RpcClient.make(group);
|
||||
const result: unknown = yield* dispatchRpc(
|
||||
rpcClient as unknown as WorkflowRpcsClient,
|
||||
operation,
|
||||
params,
|
||||
);
|
||||
|
||||
const evidenceVal = yield* Ref.get(evidenceRef);
|
||||
|
||||
if (
|
||||
result &&
|
||||
typeof result === "object" &&
|
||||
"_tag" in result &&
|
||||
result._tag === "Left"
|
||||
) {
|
||||
const left = (result as unknown as { left: unknown }).left;
|
||||
return yield* Effect.fail(mapCauseToError(left, evidenceVal));
|
||||
}
|
||||
|
||||
const successValue =
|
||||
result &&
|
||||
typeof result === "object" &&
|
||||
"_tag" in result &&
|
||||
result._tag === "Right"
|
||||
? (result as unknown as { right: unknown }).right
|
||||
: result;
|
||||
|
||||
const interpreted = interpretResult(operation, successValue);
|
||||
const durationMs = Date.now() - startTime;
|
||||
|
||||
return {
|
||||
operation,
|
||||
label: meta.label,
|
||||
interpreted,
|
||||
exchange: {
|
||||
request: evidenceVal?.request?.body ?? params,
|
||||
response: evidenceVal?.response?.body ?? successValue,
|
||||
},
|
||||
equivalentCli: meta.equivalentCli(params),
|
||||
durationMs,
|
||||
};
|
||||
}).pipe(
|
||||
Effect.provide(
|
||||
RpcClient.layerProtocolHttp({
|
||||
url: normalizedTarget,
|
||||
transformClient: (c) => withEvidenceCapture(c, evidenceRef),
|
||||
}).pipe(
|
||||
Layer.provide(
|
||||
Layer.mergeAll(FetchHttpClient.layer, RpcSerialization.layerJsonRpc()),
|
||||
),
|
||||
),
|
||||
),
|
||||
Effect.scoped,
|
||||
);
|
||||
};
|
||||
|
||||
export const makeWorkflowRpcLayer = Layer.succeed(WorkflowRpc, {
|
||||
execute: executeImpl,
|
||||
});
|
||||
Reference in New Issue
Block a user