authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-06-20 12:25:16-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-06-20 12:25:16-07:00
log4c16c66368c934a3f23794fc4826a2c77fe9f323
tree7e08f5076a98fd2e0eab8ac840d9bcbdae2ec240
parent4407a5a6b716b1d37bed917345d79f627a42f59e
signature Signed by PGP key B5690EEEBB952194

[codex] Structure ACP transport errors (#3251)

Co-authored-by: codex <codex@users.noreply.github.com>

10 files changed, 593 insertions(+), 152 deletions(-)

packages/effect-acp/src/_internal/shared.ts+19-27
......@@ -1,30 +1,29 @@
11import * as Effect from "effect/Effect";
22import * as Schema from "effect/Schema";
3import * as SchemaIssue from "effect/SchemaIssue";
43import { RpcClientError } from "effect/unstable/rpc";
54
65import * as AcpSchema from "../_generated/schema.gen.ts";
76import * as AcpError from "../errors.ts";
87const isError = Schema.is(AcpSchema.Error);
9const isAcpRequestError = Schema.is(AcpError.AcpRequestError);
10
11const formatSchemaIssue = SchemaIssue.makeFormatterDefault();
128
139export const callRpc = <A>(
10 method: string,
1411 effect: Effect.Effect<A, RpcClientError.RpcClientError | AcpSchema.Error>,
1512): Effect.Effect<A, AcpError.AcpError> =>
1613 effect.pipe(
17 Effect.catchTag("RpcClientError", (error) =>
18 Effect.fail(
19 new AcpError.AcpTransportError({
20 detail: error.message,
21 cause: error,
22 }),
23 ),
24 ),
2514 Effect.catchIf(isError, (error) =>
2615 Effect.fail(AcpError.AcpRequestError.fromProtocolError(error)),
2716 ),
17 Effect.catchTags({
18 RpcClientError: (cause) =>
19 Effect.fail(
20 new AcpError.AcpTransportError({
21 operation: "call-rpc",
22 method,
23 cause,
24 }),
25 ),
26 }),
2827 );
2928
3029export const runHandler = Effect.fnUntraced(function* <A, B>(
......@@ -37,9 +36,7 @@ export const runHandler = Effect.fnUntraced(function* <A, B>(
3736 }
3837 return yield* handler(payload).pipe(
3938 Effect.mapError((error) =>
40 isAcpRequestError(error)
41 ? error.toProtocolError()
42 : AcpError.AcpRequestError.internalError(error.message).toProtocolError(),
39 AcpError.AcpRequestError.fromCoreHandlerError(error, method).toProtocolError(),
4340 ),
4441 );
4542});
......@@ -51,12 +48,7 @@ export function decodeExtRequestRegistration<A, I>(
5148) {
5249 return (params: unknown): Effect.Effect<unknown, AcpError.AcpError> =>
5350 Schema.decodeUnknownEffect(payload)(params).pipe(
54 Effect.mapError((error) =>
55 AcpError.AcpRequestError.invalidParams(
56 `Invalid ${method} payload: ${formatSchemaIssue(error.issue)}`,
57 { issue: error.issue },
58 ),
59 ),
51 Effect.mapError((error) => AcpError.AcpRequestError.invalidExtensionPayload(method, error)),
6052 Effect.flatMap((decoded) => handler(decoded)),
6153 );
6254}
......@@ -68,12 +60,12 @@ export function decodeExtNotificationRegistration<A, I>(
6860) {
6961 return (params: unknown): Effect.Effect<void, AcpError.AcpError> =>
7062 Schema.decodeUnknownEffect(payload)(params).pipe(
71 Effect.mapError(
72 (error) =>
73 new AcpError.AcpProtocolParseError({
74 detail: `Invalid ${method} notification payload: ${formatSchemaIssue(error.issue)}`,
75 cause: error,
76 }),
63 Effect.mapError((error) =>
64 AcpError.AcpProtocolParseError.fromSchemaError(
65 "decode-notification-payload",
66 method,
67 error,
68 ),
7769 ),
7870 Effect.flatMap((decoded) => handler(decoded)),
7971 );
packages/effect-acp/src/_internal/stdio.ts+1-1
......@@ -50,7 +50,7 @@ export const makeTerminationError = (
5050 Effect.match(handle.exitCode, {
5151 onFailure: (cause) =>
5252 new AcpError.AcpTransportError({
53 detail: "Failed to determine ACP process exit status",
53 operation: "read-process-exit-status",
5454 cause,
5555 }),
5656 onSuccess: (code) => new AcpError.AcpProcessExitedError({ code }),
packages/effect-acp/src/agent.ts+54-40
......@@ -288,12 +288,12 @@ export const make = Effect.fn("effect-acp/AcpAgent.make")(function* (
288288 notification.method === AGENT_METHODS.session_cancel
289289 ) {
290290 return decodeCancelNotification(notification.params).pipe(
291 Effect.mapError(
292 (error) =>
293 new AcpError.AcpProtocolParseError({
294 detail: `Invalid ${AGENT_METHODS.session_cancel} notification payload`,
295 cause: error,
296 }),
291 Effect.mapError((error) =>
292 AcpError.AcpProtocolParseError.fromSchemaError(
293 "decode-notification-payload",
294 AGENT_METHODS.session_cancel,
295 error,
296 ),
297297 ),
298298 Effect.flatMap((decoded) =>
299299 Effect.forEach(cancelHandlers, (handler) => handler(decoded), { discard: true }),
......@@ -376,41 +376,55 @@ export const make = Effect.fn("effect-acp/AcpAgent.make")(function* (
376376 },
377377 client: {
378378 requestPermission: (payload) =>
379 callRpc(rpc[CLIENT_METHODS.session_request_permission](payload)),
380 elicit: (payload) => callRpc(rpc[CLIENT_METHODS.session_elicitation](payload)),
381 readTextFile: (payload) => callRpc(rpc[CLIENT_METHODS.fs_read_text_file](payload)),
382 writeTextFile: (payload) => callRpc(rpc[CLIENT_METHODS.fs_write_text_file](payload)),
379 callRpc(
380 CLIENT_METHODS.session_request_permission,
381 rpc[CLIENT_METHODS.session_request_permission](payload),
382 ),
383 elicit: (payload) =>
384 callRpc(
385 CLIENT_METHODS.session_elicitation,
386 rpc[CLIENT_METHODS.session_elicitation](payload),
387 ),
388 readTextFile: (payload) =>
389 callRpc(CLIENT_METHODS.fs_read_text_file, rpc[CLIENT_METHODS.fs_read_text_file](payload)),
390 writeTextFile: (payload) =>
391 callRpc(CLIENT_METHODS.fs_write_text_file, rpc[CLIENT_METHODS.fs_write_text_file](payload)),
383392 createTerminal: (payload) =>
384 callRpc(rpc[CLIENT_METHODS.terminal_create](payload)).pipe(
385 Effect.map((response) =>
386 AcpTerminal.makeTerminal({
387 sessionId: payload.sessionId,
388 terminalId: response.terminalId,
389 output: callRpc(
390 rpc[CLIENT_METHODS.terminal_output]({
391 sessionId: payload.sessionId,
392 terminalId: response.terminalId,
393 }),
394 ),
395 waitForExit: callRpc(
396 rpc[CLIENT_METHODS.terminal_wait_for_exit]({
397 sessionId: payload.sessionId,
398 terminalId: response.terminalId,
399 }),
400 ),
401 kill: callRpc(
402 rpc[CLIENT_METHODS.terminal_kill]({
403 sessionId: payload.sessionId,
404 terminalId: response.terminalId,
405 }),
406 ),
407 release: callRpc(
408 rpc[CLIENT_METHODS.terminal_release]({
409 sessionId: payload.sessionId,
410 terminalId: response.terminalId,
411 }),
412 ),
413 }),
393 callRpc(CLIENT_METHODS.terminal_create, rpc[CLIENT_METHODS.terminal_create](payload)).pipe(
394 Effect.map(
395 (response) =>
396 ({
397 sessionId: payload.sessionId,
398 terminalId: response.terminalId,
399 output: callRpc(
400 CLIENT_METHODS.terminal_output,
401 rpc[CLIENT_METHODS.terminal_output]({
402 sessionId: payload.sessionId,
403 terminalId: response.terminalId,
404 }),
405 ),
406 waitForExit: callRpc(
407 CLIENT_METHODS.terminal_wait_for_exit,
408 rpc[CLIENT_METHODS.terminal_wait_for_exit]({
409 sessionId: payload.sessionId,
410 terminalId: response.terminalId,
411 }),
412 ),
413 kill: callRpc(
414 CLIENT_METHODS.terminal_kill,
415 rpc[CLIENT_METHODS.terminal_kill]({
416 sessionId: payload.sessionId,
417 terminalId: response.terminalId,
418 }),
419 ),
420 release: callRpc(
421 CLIENT_METHODS.terminal_release,
422 rpc[CLIENT_METHODS.terminal_release]({
423 sessionId: payload.sessionId,
424 terminalId: response.terminalId,
425 }),
426 ),
427 }) satisfies AcpTerminal.AcpTerminal,
414428 ),
415429 ),
416430 sessionUpdate: (payload) => transport.notify(CLIENT_METHODS.session_update, payload),
packages/effect-acp/src/client.test.ts+3-3
......@@ -147,7 +147,7 @@ it.layer(NodeServices.layer)("effect-acp client", (it) => {
147147 );
148148
149149 it.effect(
150 "returns formatted invalid params when a typed extension request payload is wrong",
150 "returns structured invalid params without exposing values from typed extension request payloads",
151151 () =>
152152 Effect.gen(function* () {
153153 const handle = yield* makeHandle({ ACP_MOCK_BAD_TYPED_REQUEST: "1" });
......@@ -213,8 +213,8 @@ it.layer(NodeServices.layer)("effect-acp client", (it) => {
213213 assert.fail("Expected prompt to fail for invalid typed extension payload");
214214 }
215215 const rendered = Cause.pretty(result.cause);
216 assert.include(rendered, "Invalid x/typed_request payload:");
217 assert.include(rendered, "Expected string, got 123");
216 assert.include(rendered, "Invalid payload for ACP extension method 'x/typed_request'.");
217 assert.notInclude(rendered, "Expected string, got 123");
218218 }),
219219 );
220220
packages/effect-acp/src/client.ts+25-12
......@@ -462,19 +462,32 @@ export const make = Effect.fn("effect-acp/AcpClient.make")(function* (
462462 notify: transport.notify,
463463 },
464464 agent: {
465 initialize: (payload) => callRpc(rpc[AGENT_METHODS.initialize](payload)),
466 authenticate: (payload) => callRpc(rpc[AGENT_METHODS.authenticate](payload)),
467 logout: (payload) => callRpc(rpc[AGENT_METHODS.logout](payload)),
468 createSession: (payload) => callRpc(rpc[AGENT_METHODS.session_new](payload)),
469 loadSession: (payload) => callRpc(rpc[AGENT_METHODS.session_load](payload)),
470 listSessions: (payload) => callRpc(rpc[AGENT_METHODS.session_list](payload)),
471 forkSession: (payload) => callRpc(rpc[AGENT_METHODS.session_fork](payload)),
472 resumeSession: (payload) => callRpc(rpc[AGENT_METHODS.session_resume](payload)),
473 closeSession: (payload) => callRpc(rpc[AGENT_METHODS.session_close](payload)),
474 setSessionModel: (payload) => callRpc(rpc[AGENT_METHODS.session_set_model](payload)),
465 initialize: (payload) =>
466 callRpc(AGENT_METHODS.initialize, rpc[AGENT_METHODS.initialize](payload)),
467 authenticate: (payload) =>
468 callRpc(AGENT_METHODS.authenticate, rpc[AGENT_METHODS.authenticate](payload)),
469 logout: (payload) => callRpc(AGENT_METHODS.logout, rpc[AGENT_METHODS.logout](payload)),
470 createSession: (payload) =>
471 callRpc(AGENT_METHODS.session_new, rpc[AGENT_METHODS.session_new](payload)),
472 loadSession: (payload) =>
473 callRpc(AGENT_METHODS.session_load, rpc[AGENT_METHODS.session_load](payload)),
474 listSessions: (payload) =>
475 callRpc(AGENT_METHODS.session_list, rpc[AGENT_METHODS.session_list](payload)),
476 forkSession: (payload) =>
477 callRpc(AGENT_METHODS.session_fork, rpc[AGENT_METHODS.session_fork](payload)),
478 resumeSession: (payload) =>
479 callRpc(AGENT_METHODS.session_resume, rpc[AGENT_METHODS.session_resume](payload)),
480 closeSession: (payload) =>
481 callRpc(AGENT_METHODS.session_close, rpc[AGENT_METHODS.session_close](payload)),
482 setSessionModel: (payload) =>
483 callRpc(AGENT_METHODS.session_set_model, rpc[AGENT_METHODS.session_set_model](payload)),
475484 setSessionConfigOption: (payload) =>
476 callRpc(rpc[AGENT_METHODS.session_set_config_option](payload)),
477 prompt: (payload) => callRpc(rpc[AGENT_METHODS.session_prompt](payload)),
485 callRpc(
486 AGENT_METHODS.session_set_config_option,
487 rpc[AGENT_METHODS.session_set_config_option](payload),
488 ),
489 prompt: (payload) =>
490 callRpc(AGENT_METHODS.session_prompt, rpc[AGENT_METHODS.session_prompt](payload)),
478491 cancel: (payload) => transport.notify(AGENT_METHODS.session_cancel, payload),
479492 },
480493 handleRequestPermission: (handler) =>
packages/effect-acp/src/errors.test.ts created+144
......@@ -0,0 +1,144 @@
1import { describe, expect, it } from "@effect/vitest";
2import * as Effect from "effect/Effect";
3import * as Schema from "effect/Schema";
4import * as RpcClientError from "effect/unstable/rpc/RpcClientError";
5
6import * as AcpSchema from "./_generated/schema.gen.ts";
7import { callRpc, runHandler } from "./_internal/shared.ts";
8import * as AcpError from "./errors.ts";
9
10const decodeNestedNumberPayload = Schema.decodeUnknownEffect(
11 Schema.Struct({ profile: Schema.Struct({ token: Schema.Number }) }),
12);
13const encodeUnknownJson = Schema.encodeSync(Schema.UnknownFromJsonString);
14
15describe("effect-acp errors", () => {
16 it.effect("retains RPC method and cause without deriving the message from the cause", () => {
17 const rootCause = new Error("connection details that must not become the public message");
18 const failure = new RpcClientError.RpcClientError({
19 reason: new RpcClientError.RpcClientDefect({
20 message: rootCause.message,
21 cause: rootCause,
22 }),
23 });
24
25 return Effect.gen(function* () {
26 const error = yield* callRpc("session/new", Effect.fail(failure)).pipe(Effect.flip);
27
28 expect(error).toMatchObject({
29 _tag: "AcpTransportError",
30 operation: "call-rpc",
31 method: "session/new",
32 cause: failure,
33 });
34 expect(error.message).toBe("ACP transport operation call-rpc failed for method session/new.");
35 expect(error.message).not.toContain(rootCause.message);
36 });
37 });
38
39 it.effect("preserves protocol request errors as request errors", () => {
40 const failure = AcpSchema.Error.make({
41 code: -32602,
42 message: "Invalid params",
43 data: { field: "sessionId" },
44 });
45
46 return Effect.gen(function* () {
47 const error = yield* callRpc("session/load", Effect.fail(failure)).pipe(Effect.flip);
48
49 expect(error).toMatchObject({
50 _tag: "AcpRequestError",
51 code: -32602,
52 errorMessage: "Invalid params",
53 data: { field: "sessionId" },
54 });
55 });
56 });
57
58 it("does not expose legacy diagnostic detail as the transport message", () => {
59 const cause = new Error("connection refused at a private endpoint");
60 const error = new AcpError.AcpTransportError({
61 detail: cause.message,
62 cause,
63 });
64
65 expect(error.message).toBe("ACP transport operation failed.");
66 expect(error.cause).toBe(cause);
67 });
68
69 it("preserves structured extension handler failures behind stable request errors", () => {
70 const cause = new AcpError.AcpTransportError({
71 operation: "read-input-stream",
72 cause: new Error("private transport diagnostics"),
73 });
74 const error = AcpError.AcpRequestError.fromExtensionHandlerError(cause, "x/test");
75
76 expect(error).toMatchObject({
77 code: -32603,
78 method: "x/test",
79 operation: "handle-extension-request",
80 cause,
81 });
82 expect(error.message).toBe("ACP extension request handler failed for method 'x/test'");
83 expect(error.message).not.toContain(cause.message);
84 });
85
86 it.effect("uses the structured mapper for core handler failures", () => {
87 const cause = new AcpError.AcpTransportError({
88 operation: "read-input-stream",
89 cause: new Error("private transport diagnostics"),
90 });
91
92 return Effect.gen(function* () {
93 const error = yield* runHandler(() => Effect.fail(cause), {}, "fs/read_text_file").pipe(
94 Effect.flip,
95 );
96
97 expect(error).toMatchObject({
98 code: -32603,
99 message: "ACP request handler failed for method 'fs/read_text_file'",
100 });
101 expect(error.message).not.toContain(cause.message);
102 });
103 });
104
105 it.effect("keeps invalid extension payload values only in the exact schema cause", () =>
106 Effect.gen(function* () {
107 const secret = "acp-schema-payload-secret";
108 const cause = yield* decodeNestedNumberPayload({ profile: { token: secret } }).pipe(
109 Effect.flip,
110 );
111 const error = AcpError.AcpRequestError.invalidExtensionPayload("x/private", cause);
112 const { cause: directCause, ...directDiagnostics } = error;
113
114 expect(directCause).toBe(cause);
115 expect(error).toMatchObject({
116 method: "x/private",
117 operation: "decode-extension-request-payload",
118 maximumPathDepth: 2,
119 });
120 expect(error.issueCount).toBeGreaterThan(0);
121 expect(error.issueKinds).toContain("Pointer");
122 expect(error.message).toBe("Invalid payload for ACP extension method 'x/private'.");
123 expect(error.message).not.toContain(secret);
124 expect(encodeUnknownJson(directDiagnostics)).not.toContain(secret);
125 expect(encodeUnknownJson(error.toProtocolError())).not.toContain(secret);
126
127 const protocolError = AcpError.AcpProtocolParseError.fromSchemaError(
128 "decode-notification-payload",
129 "x/private",
130 cause,
131 );
132 const { cause: protocolCause, ...protocolDiagnostics } = protocolError;
133 expect(protocolCause).toBe(cause);
134 expect(protocolError).toMatchObject({
135 method: "x/private",
136 operation: "decode-notification-payload",
137 maximumPathDepth: 2,
138 });
139 expect(protocolError.message).not.toContain(secret);
140 expect(encodeUnknownJson(protocolDiagnostics)).not.toContain(secret);
141 expect("detail" in protocolError).toBe(false);
142 }),
143 );
144});
packages/effect-acp/src/errors.ts+172-6
......@@ -1,7 +1,77 @@
11import * as Schema from "effect/Schema";
2import type * as SchemaIssue from "effect/SchemaIssue";
23
34import * as AcpSchema from "./_generated/schema.gen.ts";
45
6export const AcpRequestOperation = Schema.Literals([
7 "decode-extension-request-payload",
8 "handle-request",
9 "handle-extension-request",
10]);
11export type AcpRequestOperation = typeof AcpRequestOperation.Type;
12
13export const AcpSchemaIssueKind = Schema.Literals([
14 "Filter",
15 "Encoding",
16 "Pointer",
17 "Composite",
18 "AnyOf",
19 "InvalidType",
20 "InvalidValue",
21 "MissingKey",
22 "UnexpectedKey",
23 "Forbidden",
24 "OneOf",
25]);
26export type AcpSchemaIssueKind = typeof AcpSchemaIssueKind.Type;
27
28export interface AcpSchemaIssueDiagnostics {
29 readonly issueCount: number;
30 readonly issueKinds: ReadonlyArray<AcpSchemaIssueKind>;
31 readonly maximumPathDepth: number;
32}
33
34const schemaIssueDiagnostics = (root: SchemaIssue.Issue): AcpSchemaIssueDiagnostics => {
35 let issueCount = 0;
36 let maximumPathDepth = 0;
37 const issueKinds = new Set<AcpSchemaIssueKind>();
38
39 const visit = (issue: SchemaIssue.Issue, pathDepth: number): void => {
40 issueCount += 1;
41 issueKinds.add(issue._tag);
42 maximumPathDepth = Math.max(maximumPathDepth, pathDepth);
43 switch (issue._tag) {
44 case "Filter":
45 case "Encoding":
46 visit(issue.issue, pathDepth);
47 break;
48 case "Pointer":
49 visit(issue.issue, pathDepth + issue.path.length);
50 break;
51 case "Composite":
52 case "AnyOf":
53 for (const child of issue.issues) visit(child, pathDepth);
54 break;
55 }
56 };
57
58 visit(root, 0);
59 return {
60 issueCount,
61 issueKinds: [...issueKinds],
62 maximumPathDepth,
63 };
64};
65
66export interface AcpRequestDiagnostics {
67 readonly method?: string;
68 readonly operation?: AcpRequestOperation;
69 readonly cause?: unknown;
70 readonly issueCount?: number;
71 readonly issueKinds?: ReadonlyArray<AcpSchemaIssueKind>;
72 readonly maximumPathDepth?: number;
73}
74
575export class AcpSpawnError extends Schema.TaggedErrorClass<AcpSpawnError>()("AcpSpawnError", {
676 command: Schema.optional(Schema.String),
777 cause: Schema.Defect(),
......@@ -27,27 +97,68 @@ export class AcpProcessExitedError extends Schema.TaggedErrorClass<AcpProcessExi
2797 }
2898}
2999
100export const AcpProtocolParseOperation = Schema.Literals([
101 "encode-message",
102 "decode-wire-message",
103 "decode-notification-payload",
104]);
105export type AcpProtocolParseOperation = typeof AcpProtocolParseOperation.Type;
106
30107export class AcpProtocolParseError extends Schema.TaggedErrorClass<AcpProtocolParseError>()(
31108 "AcpProtocolParseError",
32109 {
33 detail: Schema.String,
34 cause: Schema.optional(Schema.Defect()),
110 operation: AcpProtocolParseOperation,
111 method: Schema.optionalKey(Schema.String),
112 issueCount: Schema.optionalKey(Schema.Number),
113 issueKinds: Schema.optionalKey(Schema.Array(AcpSchemaIssueKind)),
114 maximumPathDepth: Schema.optionalKey(Schema.Number),
115 cause: Schema.Defect(),
35116 },
36117) {
37118 override get message() {
38 return `Failed to parse ACP protocol message: ${this.detail}`;
119 const method = this.method === undefined ? "" : ` for method '${this.method}'`;
120 return `ACP protocol operation '${this.operation}' failed${method}.`;
121 }
122
123 static fromSchemaError(
124 operation: AcpProtocolParseOperation,
125 method: string,
126 cause: Schema.SchemaError,
127 ) {
128 return new AcpProtocolParseError({
129 operation,
130 method,
131 ...schemaIssueDiagnostics(cause.issue),
132 cause,
133 });
39134 }
40135}
41136
42137export class AcpTransportError extends Schema.TaggedErrorClass<AcpTransportError>()(
43138 "AcpTransportError",
44139 {
45 detail: Schema.String,
140 operation: Schema.optional(
141 Schema.Literals(["call-rpc", "read-input-stream", "read-process-exit-status"]),
142 ),
143 method: Schema.optional(Schema.String),
144 detail: Schema.optional(Schema.String),
46145 cause: Schema.Defect(),
47146 },
48147) {
49148 override get message() {
50 return this.detail;
149 const method = this.method ? ` for method ${this.method}` : "";
150 return this.operation
151 ? `ACP transport operation ${this.operation} failed${method}.`
152 : "ACP transport operation failed.";
153 }
154}
155
156export class AcpInputStreamEndedError extends Schema.TaggedErrorClass<AcpInputStreamEndedError>()(
157 "AcpInputStreamEndedError",
158 {},
159) {
160 override get message() {
161 return "ACP input stream ended.";
51162 }
52163}
53164
......@@ -55,6 +166,12 @@ export class AcpRequestError extends Schema.TaggedErrorClass<AcpRequestError>()(
55166 code: AcpSchema.ErrorCode,
56167 errorMessage: Schema.String,
57168 data: Schema.optional(Schema.Unknown),
169 method: Schema.optionalKey(Schema.String),
170 operation: Schema.optionalKey(AcpRequestOperation),
171 issueCount: Schema.optionalKey(Schema.Number),
172 issueKinds: Schema.optionalKey(Schema.Array(AcpSchemaIssueKind)),
173 maximumPathDepth: Schema.optionalKey(Schema.Number),
174 cause: Schema.optionalKey(Schema.Defect()),
58175}) {
59176 override get message() {
60177 return this.errorMessage;
......@@ -68,6 +185,36 @@ export class AcpRequestError extends Schema.TaggedErrorClass<AcpRequestError>()(
68185 });
69186 }
70187
188 static fromCoreHandlerError(error: AcpError, method: string) {
189 if (error._tag === "AcpRequestError") {
190 return error;
191 }
192 return AcpRequestError.internalError(
193 `ACP request handler failed for method '${method}'`,
194 undefined,
195 {
196 method,
197 operation: "handle-request",
198 cause: error,
199 },
200 );
201 }
202
203 static fromExtensionHandlerError(error: AcpError, method: string) {
204 if (error._tag === "AcpRequestError") {
205 return error;
206 }
207 return AcpRequestError.internalError(
208 `ACP extension request handler failed for method '${method}'`,
209 undefined,
210 {
211 method,
212 operation: "handle-extension-request",
213 cause: error,
214 },
215 );
216 }
217
71218 static parseError(message = "Parse error", data?: unknown) {
72219 return new AcpRequestError({
73220 code: -32700,
......@@ -99,11 +246,29 @@ export class AcpRequestError extends Schema.TaggedErrorClass<AcpRequestError>()(
99246 });
100247 }
101248
102 static internalError(message = "Internal error", data?: unknown) {
249 static invalidExtensionPayload(method: string, cause: Schema.SchemaError) {
250 const diagnostics = schemaIssueDiagnostics(cause.issue);
251 return new AcpRequestError({
252 code: -32602,
253 errorMessage: `Invalid payload for ACP extension method '${method}'.`,
254 data: diagnostics,
255 method,
256 operation: "decode-extension-request-payload",
257 ...diagnostics,
258 cause,
259 });
260 }
261
262 static internalError(
263 message = "Internal error",
264 data?: unknown,
265 diagnostics: AcpRequestDiagnostics = {},
266 ) {
103267 return new AcpRequestError({
104268 code: -32603,
105269 errorMessage: message,
106270 ...(data !== undefined ? { data } : {}),
271 ...diagnostics,
107272 });
108273 }
109274
......@@ -138,6 +303,7 @@ export const AcpError = Schema.Union([
138303 AcpProcessExitedError,
139304 AcpProtocolParseError,
140305 AcpTransportError,
306 AcpInputStreamEndedError,
141307]);
142308
143309export type AcpError = typeof AcpError.Type;
packages/effect-acp/src/protocol.test.ts+134-2
......@@ -48,6 +48,8 @@ const decodeExtRequest = Schema.decodeEffect(Schema.fromJsonString(ExtRequest));
4848const decodeRequestPermissionResponse = Schema.decodeEffect(
4949 Schema.fromJsonString(RequestPermissionResponse),
5050);
51const encodeUnknownJsonString = Schema.encodeUnknownSync(Schema.UnknownFromJsonString);
52const encoder = new TextEncoder();
5153const mockPeerPath = Effect.map(Effect.service(Path.Path), (path) =>
5254 path.join(import.meta.dirname, "../test/fixtures/acp-mock-peer.ts"),
5355);
......@@ -132,6 +134,49 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
132134 }),
133135 );
134136
137 it.effect("keeps invalid core notification values only in the schema cause", () =>
138 Effect.gen(function* () {
139 const secret = "acp-core-notification-secret-sentinel";
140 const { stdio, input } = yield* makeInMemoryStdio();
141 const termination = yield* Deferred.make<AcpError.AcpError>();
142 yield* AcpProtocol.makeAcpPatchedProtocol({
143 stdio,
144 serverRequestMethods: new Set(),
145 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
146 });
147
148 yield* Queue.offer(
149 input,
150 encoder.encode(
151 `${encodeUnknownJsonString({
152 jsonrpc: "2.0",
153 method: "session/update",
154 params: {
155 sessionId: { secret },
156 update: {
157 sessionUpdate: "plan",
158 entries: [],
159 },
160 },
161 })}\n`,
162 ),
163 );
164
165 const error = yield* Deferred.await(termination);
166 assert.instanceOf(error, AcpError.AcpProtocolParseError);
167 const parseError = error as AcpError.AcpProtocolParseError;
168 const { cause, ...directDiagnostics } = parseError;
169 assert.equal(parseError.operation, "decode-notification-payload");
170 assert.equal(parseError.method, "session/update");
171 assert.isAbove(parseError.issueCount ?? 0, 0);
172 assert.include(parseError.issueKinds ?? [], "Pointer");
173 assert.isAbove(parseError.maximumPathDepth ?? 0, 0);
174 assert.isTrue(Schema.isSchemaError(cause));
175 assert.notInclude(parseError.message, secret);
176 assert.notInclude(encodeUnknownJsonString(directDiagnostics), secret);
177 }),
178 );
179
135180 it.effect("logs outgoing notifications when logOutgoing is enabled", () =>
136181 Effect.gen(function* () {
137182 const { stdio } = yield* makeInMemoryStdio();
......@@ -172,6 +217,38 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
172217 }),
173218 );
174219
220 it.effect("logs decode failures without copying the cause or wire payload", () =>
221 Effect.gen(function* () {
222 const secret = "acp-wire-secret-sentinel";
223 const { stdio, input } = yield* makeInMemoryStdio();
224 const events: Array<AcpProtocol.AcpProtocolLogEvent> = [];
225 const termination = yield* Deferred.make<AcpError.AcpError>();
226 yield* AcpProtocol.makeAcpPatchedProtocol({
227 stdio,
228 serverRequestMethods: new Set(),
229 logIncoming: true,
230 logger: (event) =>
231 Effect.sync(() => {
232 events.push(event);
233 }),
234 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
235 });
236
237 yield* Queue.offer(input, encoder.encode(`{"secret":"${secret}"\n`));
238 yield* Deferred.await(termination);
239
240 const event = events.find(({ stage }) => stage === "decode_failed");
241 assert.deepEqual(event, {
242 direction: "incoming",
243 stage: "decode_failed",
244 payload: {
245 operation: "decode-wire-message",
246 },
247 });
248 assert.notInclude(encodeUnknownJsonString(event), secret);
249 }),
250 );
251
175252 it.effect("fails notification encoding through the declared ACP error channel", () =>
176253 Effect.gen(function* () {
177254 const { stdio } = yield* makeInMemoryStdio();
......@@ -182,13 +259,16 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
182259
183260 const bigintError = yield* transport.notify("x/test", 1n).pipe(Effect.flip);
184261 assert.instanceOf(bigintError, AcpError.AcpProtocolParseError);
185 assert.equal(bigintError.detail, "Failed to encode ACP message");
262 assert.equal(bigintError.operation, "encode-message");
263 assert.instanceOf(bigintError.cause, TypeError);
264 assert.equal(bigintError.message, "ACP protocol operation 'encode-message' failed.");
186265
187266 const circular: Record<string, unknown> = {};
188267 circular.self = circular;
189268 const circularError = yield* transport.notify("x/test", circular).pipe(Effect.flip);
190269 assert.instanceOf(circularError, AcpError.AcpProtocolParseError);
191 assert.equal(circularError.detail, "Failed to encode ACP message");
270 assert.equal(circularError.operation, "encode-message");
271 assert.instanceOf(circularError.cause, TypeError);
192272 }),
193273 );
194274
......@@ -381,14 +461,35 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
381461 assert.equal((message as { readonly _tag?: string })._tag, "ClientProtocolError");
382462 const defect = (message as { readonly error: { readonly reason: unknown } }).error.reason as {
383463 readonly _tag: string;
464 readonly message: string;
384465 readonly cause: unknown;
385466 };
386467 assert.equal(defect._tag, "RpcClientDefect");
468 assert.equal(defect.message, "ACP protocol terminated.");
387469 assert.instanceOf(defect.cause, AcpError.AcpProcessExitedError);
388470 assert.equal((defect.cause as AcpError.AcpProcessExitedError).code, 7);
389471 }),
390472 );
391473
474 it.effect("classifies an input stream ending without inventing a cause", () =>
475 Effect.gen(function* () {
476 const { stdio, input } = yield* makeInMemoryStdio();
477 const termination = yield* Deferred.make<AcpError.AcpError>();
478 yield* AcpProtocol.makeAcpPatchedProtocol({
479 stdio,
480 serverRequestMethods: new Set(),
481 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
482 });
483
484 yield* Queue.end(input);
485
486 const error = yield* Deferred.await(termination);
487 assert.instanceOf(error, AcpError.AcpInputStreamEndedError);
488 assert.equal(error.message, "ACP input stream ended.");
489 assert.equal("cause" in error, false);
490 }),
491 );
492
392493 it.effect("does not emit a second process-exit error after a decode failure", () =>
393494 Effect.gen(function* () {
394495 const handle = yield* makeHandle({
......@@ -413,9 +514,40 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
413514 assert.equal((message as { readonly _tag?: string })._tag, "ClientProtocolError");
414515 const defect = (message as { readonly error: { readonly reason: unknown } }).error.reason as {
415516 readonly _tag: string;
517 readonly message: string;
416518 readonly cause: unknown;
417519 };
418520 assert.equal(defect._tag, "RpcClientDefect");
521 assert.equal(defect.message, "ACP protocol terminated.");
522 assert.instanceOf(defect.cause, AcpError.AcpProtocolParseError);
523 }),
524 );
525
526 it.effect("keeps client send failure messages independent from the cause", () =>
527 Effect.gen(function* () {
528 const { stdio } = yield* makeInMemoryStdio();
529 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
530 stdio,
531 serverRequestMethods: new Set(),
532 });
533
534 const failure = yield* transport.clientProtocol
535 .send(0, {
536 _tag: "Request",
537 id: "request-1",
538 tag: "x/test",
539 payload: 1n,
540 headers: [],
541 })
542 .pipe(Effect.flip);
543 const defect = failure.reason as {
544 readonly _tag: string;
545 readonly message: string;
546 readonly cause: unknown;
547 };
548
549 assert.equal(defect._tag, "RpcClientDefect");
550 assert.equal(defect.message, "Failed to send ACP protocol message.");
419551 assert.instanceOf(defect.cause, AcpError.AcpProtocolParseError);
420552 }),
421553 );
packages/effect-acp/src/protocol.ts+41-41
......@@ -17,7 +17,6 @@ import * as AcpSchema from "./_generated/schema.gen.ts";
1717import { CLIENT_METHODS } from "./_generated/meta.gen.ts";
1818import * as AcpError from "./errors.ts";
1919const isAcpError = Schema.is(AcpError.AcpError);
20const isAcpRequestError = Schema.is(AcpError.AcpRequestError);
2120
2221export interface AcpProtocolLogEvent {
2322 readonly direction: "incoming" | "outgoing";
......@@ -114,7 +113,7 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
114113 try: () => parser.encode(message),
115114 catch: (cause) =>
116115 new AcpError.AcpProtocolParseError({
117 detail: "Failed to encode ACP message",
116 operation: "encode-message",
118117 cause,
119118 }),
120119 });
......@@ -184,7 +183,7 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
184183 _tag: "ClientProtocolError",
185184 error: new RpcClientError.RpcClientError({
186185 reason: new RpcClientError.RpcClientDefect({
187 message: error.message,
186 message: "ACP protocol terminated.",
188187 cause: error,
189188 }),
190189 }),
......@@ -243,7 +242,11 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
243242 }
244243 return options.onExtRequest(message.tag, message.payload).pipe(
245244 Effect.matchEffect({
246 onFailure: (error) => respondWithError(message.id, normalizeToRequestError(error)),
245 onFailure: (error) =>
246 respondWithError(
247 message.id,
248 AcpError.AcpRequestError.fromExtensionHandlerError(error, message.tag),
249 ),
247250 onSuccess: (value) => respondWithSuccess(message.id, value),
248251 }),
249252 );
......@@ -261,12 +264,12 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
261264 params,
262265 }) satisfies AcpIncomingNotification,
263266 ),
264 Effect.mapError(
265 (cause) =>
266 new AcpError.AcpProtocolParseError({
267 detail: `Invalid ${CLIENT_METHODS.session_update} notification payload`,
268 cause,
269 }),
267 Effect.mapError((cause) =>
268 AcpError.AcpProtocolParseError.fromSchemaError(
269 "decode-notification-payload",
270 CLIENT_METHODS.session_update,
271 cause,
272 ),
270273 ),
271274 Effect.flatMap(dispatchNotification),
272275 );
......@@ -281,12 +284,12 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
281284 params,
282285 }) satisfies AcpIncomingNotification,
283286 ),
284 Effect.mapError(
285 (cause) =>
286 new AcpError.AcpProtocolParseError({
287 detail: `Invalid ${CLIENT_METHODS.session_elicitation_complete} notification payload`,
288 cause,
289 }),
287 Effect.mapError((cause) =>
288 AcpError.AcpProtocolParseError.fromSchemaError(
289 "decode-notification-payload",
290 CLIENT_METHODS.session_elicitation_complete,
291 cause,
292 ),
290293 ),
291294 Effect.flatMap(dispatchNotification),
292295 );
......@@ -379,7 +382,7 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
379382 >,
380383 catch: (cause) =>
381384 new AcpError.AcpProtocolParseError({
382 detail: "Failed to decode ACP wire message",
385 operation: "decode-wire-message",
383386 cause,
384387 }),
385388 }),
......@@ -396,8 +399,13 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
396399 direction: "incoming",
397400 stage: "decode_failed",
398401 payload: {
399 detail: error.detail,
400 cause: error.cause,
402 operation: error.operation,
403 ...(error.method === undefined ? {} : { method: error.method }),
404 ...(error.issueCount === undefined ? {} : { issueCount: error.issueCount }),
405 ...(error.issueKinds === undefined ? {} : { issueKinds: error.issueKinds }),
406 ...(error.maximumPathDepth === undefined
407 ? {}
408 : { maximumPathDepth: error.maximumPathDepth }),
401409 },
402410 }),
403411 ),
......@@ -413,7 +421,7 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
413421 const normalized: AcpError.AcpError = isAcpError(error)
414422 ? error
415423 : new AcpError.AcpTransportError({
416 detail: error instanceof Error ? error.message : String(error),
424 operation: "read-input-stream",
417425 cause: error,
418426 });
419427 return handleTermination(() => Effect.succeed(normalized));
......@@ -421,13 +429,7 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
421429 onSuccess: () =>
422430 handleTermination(
423431 () =>
424 options.terminationError ??
425 Effect.succeed(
426 new AcpError.AcpTransportError({
427 detail: "ACP input stream ended",
428 cause: new Error("ACP input stream ended"),
429 }),
430 ),
432 options.terminationError ?? Effect.succeed(new AcpError.AcpInputStreamEndedError({})),
431433 ),
432434 }),
433435 Effect.forkScoped,
......@@ -441,7 +443,18 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
441443 Stream.runForEach((message) => f(message)),
442444 Effect.forever,
443445 ),
444 send: (_clientId, request) => offerOutgoing(request).pipe(Effect.mapError(toRpcClientError)),
446 send: (_clientId, request) =>
447 offerOutgoing(request).pipe(
448 Effect.mapError(
449 (error) =>
450 new RpcClientError.RpcClientError({
451 reason: new RpcClientError.RpcClientDefect({
452 message: "Failed to send ACP protocol message.",
453 cause: error,
454 }),
455 }),
456 ),
457 ),
445458 supportsAck: true,
446459 supportsTransferables: false,
447460 });
......@@ -521,16 +534,3 @@ function isProtocolError(
521534 typeof value.message === "string"
522535 );
523536}
524
525function normalizeToRequestError(error: AcpError.AcpError): AcpError.AcpRequestError {
526 return isAcpRequestError(error) ? error : AcpError.AcpRequestError.internalError(error.message);
527}
528
529function toRpcClientError(error: AcpError.AcpError): RpcClientError.RpcClientError {
530 return new RpcClientError.RpcClientError({
531 reason: new RpcClientError.RpcClientDefect({
532 message: error.message,
533 cause: error,
534 }),
535 });
536}
packages/effect-acp/src/terminal.ts-20
......@@ -23,23 +23,3 @@ export interface AcpTerminal {
2323 */
2424 readonly release: Effect.Effect<AcpSchema.ReleaseTerminalResponse, AcpError.AcpError>;
2525}
26
27export interface MakeTerminalOptions {
28 readonly sessionId: string;
29 readonly terminalId: string;
30 readonly output: Effect.Effect<AcpSchema.TerminalOutputResponse, AcpError.AcpError>;
31 readonly waitForExit: Effect.Effect<AcpSchema.WaitForTerminalExitResponse, AcpError.AcpError>;
32 readonly kill: Effect.Effect<AcpSchema.KillTerminalResponse, AcpError.AcpError>;
33 readonly release: Effect.Effect<AcpSchema.ReleaseTerminalResponse, AcpError.AcpError>;
34}
35
36export function makeTerminal(options: MakeTerminalOptions): AcpTerminal {
37 return {
38 sessionId: options.sessionId,
39 terminalId: options.terminalId,
40 output: options.output,
41 waitForExit: options.waitForExit,
42 kill: options.kill,
43 release: options.release,
44 };
45}