fix: harden workflow console rpc boundary
This commit is contained in:
@@ -1,5 +1,10 @@
|
||||
import { Data } from "effect";
|
||||
|
||||
export type RpcExchangeEvidence = {
|
||||
readonly request: unknown | null;
|
||||
readonly response: unknown | null;
|
||||
};
|
||||
|
||||
export class InvalidTargetError extends Data.TaggedError("InvalidTargetError")<{
|
||||
readonly message: string;
|
||||
}> {}
|
||||
@@ -14,31 +19,36 @@ export class UpstreamConnectionError extends Data.TaggedError(
|
||||
"UpstreamConnectionError",
|
||||
)<{
|
||||
readonly message: string;
|
||||
readonly exchange?: RpcExchangeEvidence;
|
||||
}> {}
|
||||
|
||||
export class UpstreamTimeoutError extends Data.TaggedError(
|
||||
"UpstreamTimeoutError",
|
||||
)<{
|
||||
readonly message: string;
|
||||
readonly exchange?: RpcExchangeEvidence;
|
||||
}> {}
|
||||
|
||||
export class UpstreamResponseTooLargeError extends Data.TaggedError(
|
||||
"UpstreamResponseTooLargeError",
|
||||
)<{
|
||||
readonly message: string;
|
||||
readonly exchange?: RpcExchangeEvidence;
|
||||
}> {}
|
||||
|
||||
export class RpcProtocolError extends Data.TaggedError("RpcProtocolError")<{
|
||||
readonly message: string;
|
||||
readonly evidence?: string;
|
||||
readonly exchange?: RpcExchangeEvidence;
|
||||
}> {}
|
||||
|
||||
export class RpcRemoteError extends Data.TaggedError("RpcRemoteError")<{
|
||||
readonly message: string;
|
||||
readonly code: number;
|
||||
readonly data?: string;
|
||||
readonly exchange?: RpcExchangeEvidence;
|
||||
}> {}
|
||||
|
||||
export class RpcDecodeError extends Data.TaggedError("RpcDecodeError")<{
|
||||
readonly message: string;
|
||||
readonly exchange?: RpcExchangeEvidence;
|
||||
}> {}
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
import { Context, Effect, Layer, Ref } from "effect";
|
||||
import { Context, Effect, Layer, Ref, Stream } from "effect";
|
||||
import { HttpClient, HttpClientRequest, HttpClientResponse } from "@effect/platform";
|
||||
import {
|
||||
RpcProtocolError,
|
||||
UpstreamResponseTooLargeError,
|
||||
} from "./errors.js";
|
||||
|
||||
export type EvidenceRecord = {
|
||||
readonly request: {
|
||||
@@ -7,7 +11,10 @@ export type EvidenceRecord = {
|
||||
readonly method: string;
|
||||
readonly body: unknown;
|
||||
};
|
||||
readonly response: { readonly status: number; readonly body: unknown };
|
||||
readonly response: {
|
||||
readonly status: number;
|
||||
readonly body: unknown;
|
||||
} | null;
|
||||
};
|
||||
|
||||
export const EvidenceRef = Context.GenericTag<Ref.Ref<EvidenceRecord | null>>(
|
||||
@@ -31,11 +38,91 @@ const readRequestBody = (
|
||||
}
|
||||
}
|
||||
if (body._tag === "Raw") {
|
||||
return body.body;
|
||||
if (typeof body.body !== "string") return body.body;
|
||||
try {
|
||||
return JSON.parse(body.body);
|
||||
} catch {
|
||||
return body.body;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
};
|
||||
|
||||
const readBoundedText = (
|
||||
response: HttpClientResponse.HttpClientResponse,
|
||||
maxResponseBytes: number,
|
||||
): Effect.Effect<string, never> => {
|
||||
const declaredLength = Number(response.headers["content-length"] ?? "0");
|
||||
if (Number.isFinite(declaredLength) && declaredLength > maxResponseBytes) {
|
||||
return Effect.die(
|
||||
new UpstreamResponseTooLargeError({
|
||||
message: `response exceeds ${maxResponseBytes} bytes`,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
return Stream.runFoldEffect(
|
||||
response.stream,
|
||||
{ chunks: [] as Uint8Array[], size: 0 },
|
||||
(accumulator, chunk) => {
|
||||
const size = accumulator.size + chunk.byteLength;
|
||||
if (size > maxResponseBytes) {
|
||||
return Effect.die(
|
||||
new UpstreamResponseTooLargeError({
|
||||
message: `response exceeds ${maxResponseBytes} bytes`,
|
||||
}),
|
||||
);
|
||||
}
|
||||
accumulator.chunks.push(chunk);
|
||||
return Effect.succeed({ chunks: accumulator.chunks, size });
|
||||
},
|
||||
).pipe(
|
||||
Effect.map(({ chunks, size }) => {
|
||||
const bytes = new Uint8Array(size);
|
||||
let offset = 0;
|
||||
for (const chunk of chunks) {
|
||||
bytes.set(chunk, offset);
|
||||
offset += chunk.byteLength;
|
||||
}
|
||||
return new TextDecoder().decode(bytes);
|
||||
}),
|
||||
Effect.orDie,
|
||||
);
|
||||
};
|
||||
|
||||
const isRecord = (value: unknown): value is Record<string, unknown> =>
|
||||
typeof value === "object" && value !== null;
|
||||
|
||||
const downstreamRpcBodyText = (responseBody: unknown, bodyText: string): string => {
|
||||
if (!isRecord(responseBody) || Array.isArray(responseBody) || !("jsonrpc" in responseBody)) {
|
||||
return bodyText;
|
||||
}
|
||||
|
||||
const requestId = "id" in responseBody ? String(responseBody.id) : "";
|
||||
if ("result" in responseBody) {
|
||||
return JSON.stringify([
|
||||
{
|
||||
_tag: "Exit",
|
||||
requestId,
|
||||
exit: { _tag: "Success", value: responseBody.result },
|
||||
},
|
||||
]);
|
||||
}
|
||||
if ("error" in responseBody) {
|
||||
return JSON.stringify([
|
||||
{
|
||||
_tag: "Exit",
|
||||
requestId,
|
||||
exit: {
|
||||
_tag: "Failure",
|
||||
cause: { _tag: "Fail", error: responseBody.error },
|
||||
},
|
||||
},
|
||||
]);
|
||||
}
|
||||
return bodyText;
|
||||
};
|
||||
|
||||
/**
|
||||
* Wrap an HttpClient to capture raw request/response evidence per-call.
|
||||
*
|
||||
@@ -45,17 +132,23 @@ const readRequestBody = (
|
||||
export const withEvidenceCapture = <E, R>(
|
||||
client: HttpClient.HttpClient.With<E, R>,
|
||||
ref: Ref.Ref<EvidenceRecord | null>,
|
||||
maxResponseBytes: number,
|
||||
): HttpClient.HttpClient.With<E, R> =>
|
||||
client.pipe(
|
||||
HttpClient.tapRequest((request) =>
|
||||
Ref.set(ref, {
|
||||
request: {
|
||||
url: request.url,
|
||||
method: request.method,
|
||||
body: readRequestBody(request),
|
||||
},
|
||||
response: null,
|
||||
}),
|
||||
),
|
||||
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(""),
|
||||
);
|
||||
const bodyText = yield* readBoundedText(response, maxResponseBytes);
|
||||
|
||||
let responseBody: unknown;
|
||||
try {
|
||||
@@ -73,15 +166,22 @@ export const withEvidenceCapture = <E, R>(
|
||||
response: { status: response.status, body: responseBody },
|
||||
});
|
||||
|
||||
// Reconstruct a fresh response from the buffered text so downstream
|
||||
// consumers (RpcClient) can still read the body.
|
||||
if (response.status >= 300 && response.status < 400) {
|
||||
return yield* Effect.die(
|
||||
new RpcProtocolError({
|
||||
message: "upstream redirects are not allowed",
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
// RpcClient's HTTP protocol expects Effect-RPC response messages, while
|
||||
// the Python wf server returns standard JSON-RPC objects. Preserve the
|
||||
// raw object for evidence and translate only the reconstructed body.
|
||||
return HttpClientResponse.fromWeb(
|
||||
request,
|
||||
new Response(bodyText, {
|
||||
new Response(downstreamRpcBodyText(responseBody, bodyText), {
|
||||
status: response.status,
|
||||
headers: new Headers(
|
||||
Object.entries(response.headers as Record<string, string>),
|
||||
),
|
||||
headers: new Headers(response.headers),
|
||||
}),
|
||||
);
|
||||
}),
|
||||
|
||||
@@ -1,127 +1,57 @@
|
||||
import { Effect, Either } from "effect";
|
||||
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";
|
||||
import { WorkflowRpc, makeWorkflowRpcLayer } from "./service.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>);
|
||||
const runOperation = (operation: string, params: unknown = {}) =>
|
||||
Effect.gen(function* () {
|
||||
const rpc = yield* WorkflowRpc;
|
||||
return yield* rpc.execute(operation, TARGET, params);
|
||||
}).pipe(Effect.provide(makeWorkflowRpcLayer()), Effect.runPromise);
|
||||
|
||||
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({
|
||||
it("workflow.health returns interpreted status and raw JSON-RPC evidence", async () => {
|
||||
const exchange = await runOperation("workflow.health");
|
||||
|
||||
expect(exchange.interpreted).toMatchObject({
|
||||
status: "ok",
|
||||
store_root: expect.any(String),
|
||||
storeRoot: expect.any(String),
|
||||
});
|
||||
expect(exchange.exchange.request).toMatchObject({
|
||||
jsonrpc: "2.0",
|
||||
method: "workflow.health",
|
||||
});
|
||||
expect(exchange.exchange.response).toMatchObject({
|
||||
jsonrpc: "2.0",
|
||||
result: { status: "ok" },
|
||||
});
|
||||
});
|
||||
|
||||
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("workflow.sources.list returns paginated interpreted results", async () => {
|
||||
const exchange = await runOperation("workflow.sources.list", { limit: 10 });
|
||||
|
||||
expect(exchange.interpreted).toMatchObject({
|
||||
sources: expect.any(Array),
|
||||
total: expect.any(Number),
|
||||
});
|
||||
expect(exchange.equivalentCli).toContain("uv run wf source list");
|
||||
});
|
||||
|
||||
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("unknown operations fail before reaching the server", async () => {
|
||||
const result = await Effect.gen(function* () {
|
||||
const rpc = yield* WorkflowRpc;
|
||||
return yield* rpc
|
||||
.execute("workflow.nope", TARGET, {})
|
||||
.pipe(Effect.either);
|
||||
}).pipe(Effect.provide(makeWorkflowRpcLayer()), Effect.runPromise);
|
||||
|
||||
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);
|
||||
expect(Either.isLeft(result)).toBe(true);
|
||||
if (Either.isLeft(result)) {
|
||||
expect(result.left._tag).toBe("UnknownOperationError");
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,3 +1,10 @@
|
||||
import { Schema } from "effect";
|
||||
import {
|
||||
WorkflowHealthResultSchema,
|
||||
WorkflowSourcesListPayloadSchema,
|
||||
WorkflowSourcesListResultSchema,
|
||||
} from "./rpcs.js";
|
||||
|
||||
export type OperationMeta = {
|
||||
readonly method: string;
|
||||
readonly label: string;
|
||||
@@ -16,7 +23,10 @@ const registry: ReadonlyMap<string, OperationMeta> = new Map([
|
||||
explanation: "Check if the workflow server is running",
|
||||
idempotency: "read",
|
||||
equivalentCli: () => "uv run wf status",
|
||||
interpret: (result) => result,
|
||||
interpret: (result) => {
|
||||
const decoded = Schema.decodeUnknownSync(WorkflowHealthResultSchema)(result);
|
||||
return { status: decoded.status, storeRoot: decoded.store_root };
|
||||
},
|
||||
},
|
||||
],
|
||||
[
|
||||
@@ -27,13 +37,37 @@ const registry: ReadonlyMap<string, OperationMeta> = new Map([
|
||||
explanation: "List registered data sources with pagination",
|
||||
idempotency: "read",
|
||||
equivalentCli: (params) => {
|
||||
const p = params as { cursor?: string; limit?: number };
|
||||
const p = Schema.decodeUnknownSync(WorkflowSourcesListPayloadSchema)(
|
||||
params,
|
||||
{ onExcessProperty: "error" },
|
||||
);
|
||||
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,
|
||||
interpret: (result) => {
|
||||
const decoded = Schema.decodeUnknownSync(
|
||||
WorkflowSourcesListResultSchema,
|
||||
)(result);
|
||||
return {
|
||||
sources: decoded.sources.map((source) => ({
|
||||
id: source.id,
|
||||
kind: source.kind,
|
||||
enabled: source.enabled,
|
||||
description: source.description,
|
||||
counts: {
|
||||
tools: source.tool_count,
|
||||
nodeSpecs: source.node_spec_count,
|
||||
reducers: source.reducer_count,
|
||||
prompts: source.prompt_count,
|
||||
resources: source.resource_count,
|
||||
},
|
||||
})),
|
||||
nextCursor: decoded.next_cursor,
|
||||
total: decoded.total,
|
||||
};
|
||||
},
|
||||
},
|
||||
],
|
||||
]);
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { Rpc, RpcGroup } from "@effect/rpc";
|
||||
import { Schema } from "effect";
|
||||
|
||||
const SourceSummarySchema = Schema.Struct({
|
||||
export const SourceSummarySchema = Schema.Struct({
|
||||
id: Schema.String,
|
||||
kind: Schema.String,
|
||||
enabled: Schema.Boolean,
|
||||
@@ -13,27 +13,33 @@ const SourceSummarySchema = Schema.Struct({
|
||||
resource_count: Schema.Number,
|
||||
});
|
||||
|
||||
export const WorkflowHealthPayloadSchema = Schema.Struct({});
|
||||
export const WorkflowHealthResultSchema = Schema.Struct({
|
||||
status: Schema.Literal("ok"),
|
||||
store_root: Schema.String,
|
||||
});
|
||||
|
||||
export const WorkflowHealth = Rpc.make("workflow.health", {
|
||||
payload: Schema.Struct({}),
|
||||
success: Schema.Struct({
|
||||
status: Schema.Literal("ok"),
|
||||
store_root: Schema.String,
|
||||
}),
|
||||
payload: WorkflowHealthPayloadSchema,
|
||||
success: WorkflowHealthResultSchema,
|
||||
error: Schema.Never,
|
||||
});
|
||||
|
||||
export const WorkflowSourcesListPayloadSchema = Schema.Struct({
|
||||
cursor: Schema.optional(Schema.String),
|
||||
limit: Schema.optional(
|
||||
Schema.Number.pipe(Schema.int(), Schema.between(1, 100)),
|
||||
),
|
||||
});
|
||||
export const WorkflowSourcesListResultSchema = Schema.Struct({
|
||||
sources: Schema.Array(SourceSummarySchema),
|
||||
next_cursor: Schema.NullOr(Schema.String),
|
||||
total: Schema.Number,
|
||||
});
|
||||
|
||||
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,
|
||||
}),
|
||||
payload: WorkflowSourcesListPayloadSchema,
|
||||
success: WorkflowSourcesListResultSchema,
|
||||
error: Schema.Never,
|
||||
});
|
||||
|
||||
|
||||
@@ -0,0 +1,190 @@
|
||||
import { Effect, Either } from "effect";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
RpcProtocolError,
|
||||
RpcRemoteError,
|
||||
UpstreamConnectionError,
|
||||
UpstreamResponseTooLargeError,
|
||||
UpstreamTimeoutError,
|
||||
} from "./errors.js";
|
||||
import {
|
||||
WorkflowRpc,
|
||||
makeWorkflowRpcLayer,
|
||||
type OperationExchange,
|
||||
type WorkflowRpcOptions,
|
||||
} from "./service.js";
|
||||
|
||||
type JsonRpcRequest = {
|
||||
readonly jsonrpc: "2.0";
|
||||
readonly id: number | string;
|
||||
readonly method: string;
|
||||
readonly params: unknown;
|
||||
};
|
||||
|
||||
const bodyText = async (
|
||||
body: RequestInit["body"] | null | undefined,
|
||||
): Promise<string> => {
|
||||
if (typeof body === "string") return body;
|
||||
if (body instanceof Uint8Array) return new TextDecoder().decode(body);
|
||||
if (body instanceof Blob) return body.text();
|
||||
if (body instanceof ReadableStream) return new Response(body).text();
|
||||
throw new Error("expected JSON-RPC request body");
|
||||
};
|
||||
|
||||
const requestBody = async (
|
||||
input: Parameters<typeof globalThis.fetch>[0],
|
||||
init?: RequestInit,
|
||||
): Promise<JsonRpcRequest> => {
|
||||
if (input instanceof Request) {
|
||||
return JSON.parse(await input.clone().text()) as JsonRpcRequest;
|
||||
}
|
||||
|
||||
return JSON.parse(await bodyText(init?.body ?? null)) as JsonRpcRequest;
|
||||
};
|
||||
|
||||
const jsonResponse = (body: unknown, status = 200): Response =>
|
||||
new Response(JSON.stringify(body), {
|
||||
status,
|
||||
headers: { "content-type": "application/json" },
|
||||
});
|
||||
|
||||
const runOperation = (
|
||||
options: WorkflowRpcOptions,
|
||||
operation: "workflow.health" | "workflow.sources.list" = "workflow.health",
|
||||
params: unknown = {},
|
||||
): Promise<OperationExchange> =>
|
||||
Effect.gen(function* () {
|
||||
const rpc = yield* WorkflowRpc;
|
||||
return yield* rpc.execute(
|
||||
operation,
|
||||
"http://127.0.0.1:8765/rpc",
|
||||
params,
|
||||
);
|
||||
}).pipe(Effect.provide(makeWorkflowRpcLayer(options)), Effect.runPromise);
|
||||
|
||||
const runEither = (
|
||||
options: WorkflowRpcOptions,
|
||||
): Promise<Either.Either<OperationExchange, unknown>> =>
|
||||
Effect.gen(function* () {
|
||||
const rpc = yield* WorkflowRpc;
|
||||
return yield* rpc
|
||||
.execute("workflow.health", "http://127.0.0.1:8765/rpc", {})
|
||||
.pipe(Effect.either);
|
||||
}).pipe(Effect.provide(makeWorkflowRpcLayer(options)), Effect.runPromise);
|
||||
|
||||
describe("WorkflowRpc", () => {
|
||||
it("uses @effect/rpc and returns exact raw request and response evidence", async () => {
|
||||
const fetch: typeof globalThis.fetch = async (input, init) => {
|
||||
const request = await requestBody(input, init);
|
||||
return jsonResponse({
|
||||
jsonrpc: "2.0",
|
||||
id: request.id,
|
||||
result: { status: "ok", store_root: "C:/store" },
|
||||
});
|
||||
};
|
||||
|
||||
const exchange = await runOperation({ fetch });
|
||||
|
||||
expect(exchange.target).toBe("http://127.0.0.1:8765/rpc");
|
||||
expect(exchange.interpreted).toEqual({
|
||||
status: "ok",
|
||||
storeRoot: "C:/store",
|
||||
});
|
||||
expect(exchange.exchange.request).toMatchObject({
|
||||
jsonrpc: "2.0",
|
||||
method: "workflow.health",
|
||||
params: {},
|
||||
});
|
||||
expect(exchange.exchange.response).toMatchObject({
|
||||
jsonrpc: "2.0",
|
||||
result: { status: "ok", store_root: "C:/store" },
|
||||
});
|
||||
});
|
||||
|
||||
it("requests manual redirect handling", async () => {
|
||||
let redirect: RequestInit["redirect"];
|
||||
const fetch: typeof globalThis.fetch = async (input, init) => {
|
||||
redirect = init?.redirect ?? (input instanceof Request ? input.redirect : undefined);
|
||||
const request = await requestBody(input, init);
|
||||
return jsonResponse({
|
||||
jsonrpc: "2.0",
|
||||
id: request.id,
|
||||
result: { status: "ok", store_root: "C:/store" },
|
||||
});
|
||||
};
|
||||
|
||||
await runOperation({ fetch });
|
||||
|
||||
expect(redirect).toBe("manual");
|
||||
});
|
||||
|
||||
it("maps a standard foreign JSON-RPC error and preserves evidence", async () => {
|
||||
const fetch: typeof globalThis.fetch = async (input, init) => {
|
||||
const request = await requestBody(input, init);
|
||||
return jsonResponse({
|
||||
jsonrpc: "2.0",
|
||||
id: request.id,
|
||||
error: { code: -32602, message: "Invalid params", data: { field: "x" } },
|
||||
});
|
||||
};
|
||||
|
||||
const result = await runEither({ fetch });
|
||||
|
||||
expect(Either.isLeft(result)).toBe(true);
|
||||
if (Either.isRight(result)) return;
|
||||
expect(result.left).toBeInstanceOf(RpcRemoteError);
|
||||
expect((result.left as RpcRemoteError).exchange?.response).toMatchObject({
|
||||
error: { code: -32602, message: "Invalid params" },
|
||||
});
|
||||
});
|
||||
|
||||
it("fails with a bounded timeout", async () => {
|
||||
const fetch: typeof globalThis.fetch = () => new Promise<Response>(() => {});
|
||||
|
||||
const result = await runEither({ fetch, timeoutMilliseconds: 5 });
|
||||
|
||||
expect(Either.isLeft(result)).toBe(true);
|
||||
if (Either.isRight(result)) return;
|
||||
expect(result.left).toBeInstanceOf(UpstreamTimeoutError);
|
||||
});
|
||||
|
||||
it("maps transport failures to upstream connection errors", async () => {
|
||||
const fetch: typeof globalThis.fetch = async () => {
|
||||
throw new Error("connection refused");
|
||||
};
|
||||
|
||||
const result = await runEither({ fetch });
|
||||
|
||||
expect(Either.isLeft(result)).toBe(true);
|
||||
if (Either.isRight(result)) return;
|
||||
expect(result.left).toBeInstanceOf(UpstreamConnectionError);
|
||||
});
|
||||
|
||||
it("rejects a response larger than the configured byte limit", async () => {
|
||||
const fetch: typeof globalThis.fetch = async (input, init) => {
|
||||
const request = await requestBody(input, init);
|
||||
return jsonResponse({
|
||||
jsonrpc: "2.0",
|
||||
id: request.id,
|
||||
result: { status: "ok", store_root: "x".repeat(512) },
|
||||
});
|
||||
};
|
||||
|
||||
const result = await runEither({ fetch, maxResponseBytes: 128 });
|
||||
|
||||
expect(Either.isLeft(result)).toBe(true);
|
||||
if (Either.isRight(result)) return;
|
||||
expect(result.left).toBeInstanceOf(UpstreamResponseTooLargeError);
|
||||
});
|
||||
|
||||
it("rejects a redirect response instead of decoding it", async () => {
|
||||
const fetch: typeof globalThis.fetch = async () =>
|
||||
new Response("", { status: 302, headers: { location: "/elsewhere" } });
|
||||
|
||||
const result = await runEither({ fetch });
|
||||
|
||||
expect(Either.isLeft(result)).toBe(true);
|
||||
if (Either.isRight(result)) return;
|
||||
expect(result.left).toBeInstanceOf(RpcProtocolError);
|
||||
});
|
||||
});
|
||||
+245
-221
@@ -1,24 +1,52 @@
|
||||
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 { FetchHttpClient } from "@effect/platform";
|
||||
import { RpcClient, RpcSerialization } from "@effect/rpc";
|
||||
import {
|
||||
Cause,
|
||||
Clock,
|
||||
Context,
|
||||
Effect,
|
||||
Layer,
|
||||
Option,
|
||||
Ref,
|
||||
Schema,
|
||||
} from "effect";
|
||||
import {
|
||||
InvalidTargetError,
|
||||
UnknownOperationError,
|
||||
UpstreamConnectionError,
|
||||
UpstreamTimeoutError,
|
||||
RpcDecodeError,
|
||||
type RpcExchangeEvidence,
|
||||
RpcProtocolError,
|
||||
RpcRemoteError,
|
||||
UnknownOperationError,
|
||||
UpstreamConnectionError,
|
||||
UpstreamResponseTooLargeError,
|
||||
UpstreamTimeoutError,
|
||||
} from "./errors.js";
|
||||
import { type EvidenceRecord, withEvidenceCapture } from "./evidence.js";
|
||||
import { getOperationMeta } from "./method-registry.js";
|
||||
import {
|
||||
WorkflowHealthPayloadSchema,
|
||||
WorkflowRpcs,
|
||||
WorkflowSourcesListPayloadSchema,
|
||||
} from "./rpcs.js";
|
||||
import { normalizeLoopbackTarget } from "./target-policy.js";
|
||||
|
||||
const DEFAULT_TIMEOUT_MILLISECONDS = 5_000;
|
||||
const DEFAULT_MAX_RESPONSE_BYTES = 4 * 1024 * 1024;
|
||||
|
||||
export type OperationName = "workflow.health" | "workflow.sources.list";
|
||||
|
||||
export interface WorkflowRpcOptions {
|
||||
readonly fetch?: typeof globalThis.fetch;
|
||||
readonly timeoutMilliseconds?: number;
|
||||
readonly maxResponseBytes?: number;
|
||||
}
|
||||
|
||||
export interface OperationExchange {
|
||||
readonly operation: string;
|
||||
readonly operation: OperationName;
|
||||
readonly target: string;
|
||||
readonly label: string;
|
||||
readonly interpreted: unknown;
|
||||
readonly exchange: { readonly request: unknown; readonly response: unknown };
|
||||
readonly exchange: RpcExchangeEvidence;
|
||||
readonly equivalentCli: string;
|
||||
readonly durationMs: number;
|
||||
}
|
||||
@@ -28,81 +56,224 @@ export type WorkflowRpcError =
|
||||
| UnknownOperationError
|
||||
| UpstreamConnectionError
|
||||
| UpstreamTimeoutError
|
||||
| UpstreamResponseTooLargeError
|
||||
| RpcProtocolError
|
||||
| RpcRemoteError;
|
||||
| RpcRemoteError
|
||||
| RpcDecodeError;
|
||||
|
||||
export type OperationName = "workflow.health" | "workflow.sources.list";
|
||||
const isOperationName = (value: string): value is OperationName =>
|
||||
value === "workflow.health" || value === "workflow.sources.list";
|
||||
|
||||
const isOperationName = (s: string): s is OperationName =>
|
||||
s === "workflow.health" || s === "workflow.sources.list";
|
||||
const toExchange = (evidence: EvidenceRecord | null): RpcExchangeEvidence => ({
|
||||
request: evidence?.request.body ?? null,
|
||||
response: evidence?.response?.body ?? null,
|
||||
});
|
||||
|
||||
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;
|
||||
const responseError = (
|
||||
evidence: EvidenceRecord | null,
|
||||
): { readonly code: number; readonly message: string; readonly data?: unknown } | null => {
|
||||
const body = evidence?.response?.body;
|
||||
if (typeof body !== "object" || body === null || !("error" in body)) return null;
|
||||
const error = body.error;
|
||||
if (typeof error !== "object" || error === null) return null;
|
||||
const code = "code" in error ? Number(error.code) : Number.NaN;
|
||||
const message = "message" in error ? String(error.message) : "remote error";
|
||||
if (!Number.isFinite(code)) return null;
|
||||
return {
|
||||
code,
|
||||
message,
|
||||
...("data" in error ? { data: error.data } : {}),
|
||||
};
|
||||
};
|
||||
|
||||
const containsTag = (
|
||||
value: unknown,
|
||||
tags: ReadonlySet<string>,
|
||||
depth = 0,
|
||||
): boolean => {
|
||||
if (depth > 6 || typeof value !== "object" || value === null) return false;
|
||||
if ("_tag" in value && typeof value._tag === "string" && tags.has(value._tag)) {
|
||||
return true;
|
||||
}
|
||||
return Object.values(value).some((child) => containsTag(child, tags, depth + 1));
|
||||
};
|
||||
|
||||
const domainErrorFromUnknown = (
|
||||
value: unknown,
|
||||
exchange: RpcExchangeEvidence,
|
||||
): WorkflowRpcError | null => {
|
||||
if (value instanceof UpstreamTimeoutError) {
|
||||
return new UpstreamTimeoutError({ message: value.message, exchange });
|
||||
}
|
||||
if (value instanceof UpstreamResponseTooLargeError) {
|
||||
return new UpstreamResponseTooLargeError({ message: value.message, exchange });
|
||||
}
|
||||
if (value instanceof RpcProtocolError) {
|
||||
return new RpcProtocolError({ message: value.message, exchange });
|
||||
}
|
||||
if (value instanceof RpcDecodeError) {
|
||||
return new RpcDecodeError({ message: value.message, exchange });
|
||||
}
|
||||
return null;
|
||||
};
|
||||
|
||||
/**
|
||||
* 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,
|
||||
cause: Cause.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 });
|
||||
}
|
||||
const exchange = toExchange(evidence);
|
||||
const remote = responseError(evidence);
|
||||
if (remote) {
|
||||
return new RpcRemoteError({
|
||||
message: remote.message,
|
||||
code: remote.code,
|
||||
...(remote.data === undefined ? {} : { data: JSON.stringify(remote.data) }),
|
||||
exchange,
|
||||
});
|
||||
}
|
||||
|
||||
// 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 failure = Option.getOrUndefined(Cause.failureOption(cause));
|
||||
const defect = Option.getOrUndefined(Cause.dieOption(cause));
|
||||
const known =
|
||||
domainErrorFromUnknown(failure, exchange) ??
|
||||
domainErrorFromUnknown(defect, exchange);
|
||||
if (known) return known;
|
||||
|
||||
if (evidence?.request && !evidence.response) {
|
||||
return new UpstreamConnectionError({
|
||||
message: "could not connect to the workflow RPC server",
|
||||
exchange,
|
||||
});
|
||||
}
|
||||
|
||||
const msg = cause instanceof Error ? cause.message : String(cause);
|
||||
return new RpcProtocolError({ message: msg });
|
||||
if (containsTag(cause, new Set(["RequestError", "ResponseError"]))) {
|
||||
return new UpstreamConnectionError({
|
||||
message: "could not connect to the workflow RPC server",
|
||||
exchange,
|
||||
});
|
||||
}
|
||||
|
||||
const description = String(Cause.squash(cause));
|
||||
if (description.toLowerCase().includes("parse") || description.includes("Schema")) {
|
||||
return new RpcDecodeError({
|
||||
message: "workflow RPC result did not match the expected schema",
|
||||
exchange,
|
||||
});
|
||||
}
|
||||
return new RpcProtocolError({
|
||||
message: "workflow RPC server returned an invalid response",
|
||||
exchange,
|
||||
});
|
||||
};
|
||||
|
||||
const decodeParams = <A, I>(
|
||||
schema: Schema.Schema<A, I>,
|
||||
params: unknown,
|
||||
): Effect.Effect<A, RpcDecodeError> =>
|
||||
Schema.decodeUnknown(schema, { onExcessProperty: "error" })(params).pipe(
|
||||
Effect.mapError(
|
||||
() => new RpcDecodeError({ message: "invalid workflow RPC parameters" }),
|
||||
),
|
||||
);
|
||||
|
||||
const executeImpl =
|
||||
(options: WorkflowRpcOptions) =>
|
||||
(
|
||||
operation: string,
|
||||
target: string,
|
||||
params: unknown,
|
||||
): Effect.Effect<OperationExchange, WorkflowRpcError> => {
|
||||
let normalizedTarget: string;
|
||||
try {
|
||||
normalizedTarget = normalizeLoopbackTarget(target);
|
||||
} catch (error) {
|
||||
return Effect.fail(
|
||||
error instanceof InvalidTargetError
|
||||
? error
|
||||
: new InvalidTargetError({ message: "invalid workflow RPC target" }),
|
||||
);
|
||||
}
|
||||
if (!isOperationName(operation)) {
|
||||
return Effect.fail(
|
||||
new UnknownOperationError({ message: `unknown operation: ${operation}` }),
|
||||
);
|
||||
}
|
||||
|
||||
return Effect.gen(function* () {
|
||||
const evidenceRef = yield* Ref.make<EvidenceRecord | null>(null);
|
||||
const startedAt = yield* Clock.currentTimeMillis;
|
||||
const timeoutMilliseconds =
|
||||
options.timeoutMilliseconds ?? DEFAULT_TIMEOUT_MILLISECONDS;
|
||||
const maxResponseBytes =
|
||||
options.maxResponseBytes ?? DEFAULT_MAX_RESPONSE_BYTES;
|
||||
|
||||
const fetchOptionsLayer = Layer.mergeAll(
|
||||
Layer.succeed(FetchHttpClient.Fetch, options.fetch ?? globalThis.fetch),
|
||||
Layer.succeed(FetchHttpClient.RequestInit, { redirect: "manual" }),
|
||||
);
|
||||
const fetchLayer = Layer.merge(
|
||||
FetchHttpClient.layer.pipe(Layer.provide(fetchOptionsLayer)),
|
||||
RpcSerialization.layerJsonRpc(),
|
||||
);
|
||||
const protocolLayer = RpcClient.layerProtocolHttp({
|
||||
url: normalizedTarget,
|
||||
transformClient: (client) =>
|
||||
withEvidenceCapture(client, evidenceRef, maxResponseBytes),
|
||||
}).pipe(Layer.provide(fetchLayer));
|
||||
|
||||
const call = Effect.gen(function* () {
|
||||
const client = yield* RpcClient.make(WorkflowRpcs);
|
||||
switch (operation) {
|
||||
case "workflow.health": {
|
||||
const payload = yield* decodeParams(WorkflowHealthPayloadSchema, params);
|
||||
return yield* client.workflow.health(payload);
|
||||
}
|
||||
case "workflow.sources.list": {
|
||||
const payload = yield* decodeParams(
|
||||
WorkflowSourcesListPayloadSchema,
|
||||
params,
|
||||
);
|
||||
return yield* client.workflow["sources.list"](payload);
|
||||
}
|
||||
}
|
||||
}).pipe(
|
||||
Effect.provide(protocolLayer),
|
||||
Effect.scoped,
|
||||
Effect.timeoutFail({
|
||||
duration: timeoutMilliseconds,
|
||||
onTimeout: () =>
|
||||
new UpstreamTimeoutError({
|
||||
message: "workflow RPC request timed out",
|
||||
}),
|
||||
}),
|
||||
Effect.catchAllCause((cause) =>
|
||||
Ref.get(evidenceRef).pipe(
|
||||
Effect.flatMap((evidence) => Effect.fail(mapCauseToError(cause, evidence))),
|
||||
),
|
||||
),
|
||||
);
|
||||
|
||||
const result = yield* call;
|
||||
const evidence = yield* Ref.get(evidenceRef);
|
||||
const metadata = getOperationMeta(operation);
|
||||
if (!metadata) {
|
||||
return yield* Effect.fail(
|
||||
new UnknownOperationError({ message: `unknown operation: ${operation}` }),
|
||||
);
|
||||
}
|
||||
const finishedAt = yield* Clock.currentTimeMillis;
|
||||
return {
|
||||
operation,
|
||||
target: normalizedTarget,
|
||||
label: metadata.label,
|
||||
interpreted: metadata.interpret(result),
|
||||
exchange: toExchange(evidence),
|
||||
equivalentCli: metadata.equivalentCli(params),
|
||||
durationMs: finishedAt - startedAt,
|
||||
};
|
||||
});
|
||||
};
|
||||
|
||||
export const WorkflowRpc = Context.GenericTag<{
|
||||
readonly execute: (
|
||||
operation: string,
|
||||
@@ -111,152 +282,5 @@ export const WorkflowRpc = Context.GenericTag<{
|
||||
) => 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,
|
||||
});
|
||||
export const makeWorkflowRpcLayer = (options: WorkflowRpcOptions = {}) =>
|
||||
Layer.succeed(WorkflowRpc, { execute: executeImpl(options) });
|
||||
|
||||
Reference in New Issue
Block a user