lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden workflow console runtime

Browse Source
This commit is contained in:
lda
2026-07-02 17:35:16 +07:00 Verified
parent bcf31583af
commit ea1276badd
22 changed files with 684 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/packages/rpc/src/evidence.ts
+52 -2
View File
@@ -1,5 +1,10 @@
import { Context, Effect, Layer, Ref, Stream } from "effect";
import { HttpClient, HttpClientRequest, HttpClientResponse } from "@effect/platform";
import {
HttpBody,
HttpClient,
HttpClientRequest,
HttpClientResponse,
} from "@effect/platform";
import {
RpcProtocolError,
UpstreamResponseTooLargeError,
@@ -48,6 +53,44 @@ const readRequestBody = (
return null;
};
const bodyText = (body: HttpBody.HttpBody): string | null => {
if (body._tag === "Uint8Array") {
return new TextDecoder().decode(body.body);
}
if (body._tag === "Raw" && typeof body.body === "string") {
return body.body;
}
return null;
};
const sanitizeJsonRpcRequest = (
request: HttpClientRequest.HttpClientRequest,
): HttpClientRequest.HttpClientRequest => {
const text = bodyText(request.body);
if (text === null) return request;
let body: unknown;
try {
body = JSON.parse(text);
} catch {
return request;
}
if (!isRecord(body) || Array.isArray(body) || body.jsonrpc !== "2.0") {
return request;
}
const sanitized = {
jsonrpc: "2.0",
method: body.method,
params: "params" in body ? body.params : undefined,
id: "id" in body ? body.id : undefined,
};
return HttpClientRequest.modify(request, {
body: HttpBody.text(JSON.stringify(sanitized), "application/json"),
});
};
const readBoundedText = (
response: HttpClientResponse.HttpClientResponse,
maxResponseBytes: number,
@@ -135,6 +178,7 @@ export const withEvidenceCapture = <E, R>(
maxResponseBytes: number,
): HttpClient.HttpClient.With<E, R> =>
client.pipe(
HttpClient.mapRequest(sanitizeJsonRpcRequest),
HttpClient.tapRequest((request) =>
Ref.set(ref, {
request: {
@@ -177,11 +221,17 @@ export const withEvidenceCapture = <E, R>(
// 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.
const headers = new Headers(response.headers);
// The body is rewritten for Effect-RPC, so upstream body metadata no
// longer describes the reconstructed Response.
headers.delete("content-length");
headers.delete("content-encoding");
headers.delete("transfer-encoding");
return HttpClientResponse.fromWeb(
request,
new Response(downstreamRpcBodyText(responseBody, bodyText), {
status: response.status,
headers: new Headers(response.headers),
headers,
}),
);
}),
web/packages/rpc/src/index.ts
+5 -1
View File
@@ -20,7 +20,11 @@ export {
getOperationMeta,
listOperations,
} from "./method-registry.js";
export type { OperationMeta } from "./method-registry.js";
export type {
OperationMeta,
WorkflowHealthInterpreted,
WorkflowSourcesListInterpreted,
} from "./method-registry.js";
export {
EvidenceRef,
web/packages/rpc/src/method-registry.ts
+77 -56
View File
@@ -14,66 +14,87 @@ export type OperationMeta = {
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) => {
const decoded = Schema.decodeUnknownSync(WorkflowHealthResultSchema)(result);
return { status: decoded.status, storeRoot: decoded.store_root };
},
export type WorkflowHealthInterpreted = {
readonly status: "ok";
readonly storeRoot: string;
};
export type WorkflowSourcesListInterpreted = {
readonly sources: ReadonlyArray<{
readonly id: string;
readonly kind: string;
readonly enabled: boolean;
readonly description: string | null;
readonly counts: {
readonly tools: number;
readonly nodeSpecs: number;
readonly reducers: number;
readonly prompts: number;
readonly resources: number;
};
}>;
readonly nextCursor: string | null;
readonly total: number;
};
const operationEntries: ReadonlyArray<OperationMeta> = [
{
method: "workflow.health",
label: "Health check",
explanation: "Check if the workflow server is running",
idempotency: "read",
equivalentCli: () => "uv run wf status",
interpret: (result): WorkflowHealthInterpreted => {
const decoded = Schema.decodeUnknownSync(WorkflowHealthResultSchema)(result);
return { status: decoded.status, storeRoot: decoded.store_root };
},
],
[
"workflow.sources.list",
{
method: "workflow.sources.list",
label: "List sources",
explanation: "List registered data sources with pagination",
idempotency: "read",
equivalentCli: (params) => {
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) => {
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,
};
},
},
{
method: "workflow.sources.list",
label: "List sources",
explanation: "List registered data sources with pagination",
idempotency: "read",
equivalentCli: (params) => {
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): WorkflowSourcesListInterpreted => {
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,
};
},
},
];
const registry: ReadonlyMap<string, OperationMeta> = new Map(
operationEntries.map((entry) => [entry.method, entry]),
);
export const getOperationMeta = (method: string): OperationMeta | undefined =>
registry.get(method);
export const listOperations = (): ReadonlyArray<OperationMeta> =>
Array.from(registry.values());
operationEntries;
web/packages/rpc/src/rpcs.ts
+11 -6
View File
@@ -1,16 +1,21 @@
import { Rpc, RpcGroup } from "@effect/rpc";
import { Schema } from "effect";
const NonNegativeIntegerSchema = Schema.Number.pipe(
Schema.int(),
Schema.between(0, Number.MAX_SAFE_INTEGER),
);
export 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,
tool_count: NonNegativeIntegerSchema,
node_spec_count: NonNegativeIntegerSchema,
reducer_count: NonNegativeIntegerSchema,
prompt_count: NonNegativeIntegerSchema,
resource_count: NonNegativeIntegerSchema,
});
export const WorkflowHealthPayloadSchema = Schema.Struct({});
@@ -34,7 +39,7 @@ export const WorkflowSourcesListPayloadSchema = Schema.Struct({
export const WorkflowSourcesListResultSchema = Schema.Struct({
sources: Schema.Array(SourceSummarySchema),
next_cursor: Schema.NullOr(Schema.String),
total: Schema.Number,
total: NonNegativeIntegerSchema,
});
export const WorkflowSourcesList = Rpc.make("workflow.sources.list", {
web/packages/rpc/src/service.test.ts
+63
View File
@@ -1,6 +1,7 @@
import { Effect, Either } from "effect";
import { describe, expect, it } from "vitest";
import {
RpcDecodeError,
RpcProtocolError,
RpcRemoteError,
UpstreamConnectionError,
@@ -138,6 +139,68 @@ describe("WorkflowRpc", () => {
});
});
it("maps malformed successful results to decode errors with 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: "wrong", store_root: "C:/store" },
});
};
const result = await runEither({ fetch });
expect(Either.isLeft(result)).toBe(true);
if (Either.isRight(result)) return;
expect(result.left).toBeInstanceOf(RpcDecodeError);
expect((result.left as RpcDecodeError).exchange?.response).toMatchObject({
result: { status: "wrong", store_root: "C:/store" },
});
});
it("rejects invalid source count shapes as decode errors", async () => {
const fetch: typeof globalThis.fetch = async (input, init) => {
const request = await requestBody(input, init);
return jsonResponse({
jsonrpc: "2.0",
id: request.id,
result: {
sources: [
{
id: "local.demo",
kind: "python",
enabled: true,
description: null,
tool_count: -1,
node_spec_count: 0,
reducer_count: 0,
prompt_count: 0,
resource_count: 0,
},
],
next_cursor: null,
total: 1,
},
});
};
const result = await Effect.gen(function* () {
const rpc = yield* WorkflowRpc;
return yield* rpc
.execute(
"workflow.sources.list",
"http://127.0.0.1:8765/rpc",
{},
)
.pipe(Effect.either);
}).pipe(Effect.provide(makeWorkflowRpcLayer({ fetch })), Effect.runPromise);
expect(Either.isLeft(result)).toBe(true);
if (Either.isRight(result)) return;
expect(result.left).toBeInstanceOf(RpcDecodeError);
});
it("fails with a bounded timeout", async () => {
const fetch: typeof globalThis.fetch = () => new Promise<Response>(() => {});
web/packages/rpc/src/service.ts
+36 -3
View File
@@ -154,7 +154,13 @@ const mapCauseToError = (
}
const description = String(Cause.squash(cause));
if (description.toLowerCase().includes("parse") || description.includes("Schema")) {
const lowerDescription = description.toLowerCase();
if (
lowerDescription.includes("parse") ||
lowerDescription.includes("schema") ||
lowerDescription.includes("decode") ||
lowerDescription.includes("expected")
) {
return new RpcDecodeError({
message: "workflow RPC result did not match the expected schema",
exchange,
@@ -176,6 +182,27 @@ const decodeParams = <A, I>(
),
);
const decodeOperationMetadata = (
metadata: NonNullable<ReturnType<typeof getOperationMeta>>,
result: unknown,
params: unknown,
evidence: EvidenceRecord | null,
): Effect.Effect<
{ readonly interpreted: unknown; readonly equivalentCli: string },
RpcDecodeError
> =>
Effect.try({
try: () => ({
interpreted: metadata.interpret(result),
equivalentCli: metadata.equivalentCli(params),
}),
catch: () =>
new RpcDecodeError({
message: "workflow RPC result did not match the expected schema",
exchange: toExchange(evidence),
}),
});
const executeImpl =
(options: WorkflowRpcOptions) =>
(
@@ -261,14 +288,20 @@ const executeImpl =
new UnknownOperationError({ message: `unknown operation: ${operation}` }),
);
}
const decodedMetadata = yield* decodeOperationMetadata(
metadata,
result,
params,
evidence,
);
const finishedAt = yield* Clock.currentTimeMillis;
return {
operation,
target: normalizedTarget,
label: metadata.label,
interpreted: metadata.interpret(result),
interpreted: decodedMetadata.interpreted,
exchange: toExchange(evidence),
equivalentCli: metadata.equivalentCli(params),
equivalentCli: decodedMetadata.equivalentCli,
durationMs: finishedAt - startedAt,
};
});
web/packages/rpc/tsconfig.json
+2 -1
View File
@@ -10,5 +10,6 @@
"rootDir": "src",
"outDir": "dist"
},
"include": ["src"]
"include": ["src"],
"exclude": ["src/**/*.test.ts"]
}