authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-10-06 16:01:44-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-06 16:01:44-07:00
log8ddf200e8f637eaf9a4f0d3036ece8f1d3869d10
treebbebf4ca5e62214311157db13ce2616451d58bd6
parentf4f148eb670622a049ae6561d7795011e383fc43
signature Signed by PGP key B5690EEEBB952194

fix(server): one failing RPC handler no longer ends the client's other requests (#15515)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

10 files changed, 454 insertions(+), 16 deletions(-)

apps/server/src/observability/DefectReporter.test.ts created+102
......@@ -0,0 +1,102 @@
1import { assert, describe, it } from "@effect/vitest";
2import * as Cause from "effect/Cause";
3import * as Effect from "effect/Effect";
4import * as ErrorReporter from "effect/ErrorReporter";
5import * as Exit from "effect/Exit";
6import * as Fiber from "effect/Fiber";
7import * as Layer from "effect/Layer";
8import * as Logger from "effect/Logger";
9import * as Queue from "effect/Queue";
10import * as Schema from "effect/Schema";
11import * as Stream from "effect/Stream";
12import { Rpc, RpcClient, RpcGroup, RpcServer } from "effect/rpc";
13
14import { WS_RPC_SERVER_OPTIONS } from "../ws.ts";
15import * as DefectReporter from "./DefectReporter.ts";
16
17class TestRpcs extends RpcGroup.make(
18 Rpc.make("subscribe", { success: Schema.Number, stream: true }),
19 Rpc.make("boom", { success: Schema.Void }),
20) {}
21type TestRpc = RpcGroup.Rpcs<typeof TestRpcs>;
22
23/** Runs `body` with every error log captured. */
24const withErrorLogs = <A, E, R>(
25 body: (logs: Queue.Queue<Cause.Cause<unknown>>) => Effect.Effect<A, E, R>,
26) =>
27 Effect.gen(function* () {
28 const logs = yield* Queue.unbounded<Cause.Cause<unknown>>();
29 const logger = Logger.make(({ cause, logLevel }) => {
30 if (logLevel === "Error") Queue.offerUnsafe(logs, cause);
31 });
32 return yield* body(logs).pipe(
33 Effect.provide(Logger.layer([logger], { mergeWithExisting: false })),
34 );
35 });
36
37const errorMessage = (cause: Cause.Cause<unknown>) => (Cause.squash(cause) as Error).message;
38
39describe("DefectReporter", () => {
40 it.effect("a dying RPC handler fails alone, and its defect is logged", () =>
41 withErrorLogs((logs) =>
42 Effect.gen(function* () {
43 const subscription = yield* Queue.unbounded<number>();
44 // The same client/server pairing as RpcTest.makeClient, with ws.ts's options.
45 let client!: Effect.Success<
46 ReturnType<typeof RpcClient.makeNoSerialization<TestRpc, never>>
47 >;
48 const server = yield* RpcServer.makeNoSerialization(TestRpcs, {
49 ...WS_RPC_SERVER_OPTIONS,
50 onFromServer: (response) => client.write(response),
51 }).pipe(
52 // Provided to the handlers, as ws.ts does.
53 Effect.provide(
54 TestRpcs.toLayer({
55 subscribe: () => Stream.fromQueue(subscription),
56 boom: () => Effect.die(new Error("handler bug")),
57 }).pipe(Layer.provide(DefectReporter.layer)),
58 ),
59 );
60 client = yield* RpcClient.makeNoSerialization(TestRpcs, {
61 supportsAck: true,
62 onFromClient: ({ message }) => server.write(0, message),
63 });
64
65 const received = yield* Queue.unbounded<number>();
66 const sibling = yield* client.client.subscribe().pipe(
67 Stream.runForEach((value) => Queue.offer(received, value)),
68 Effect.forkScoped,
69 );
70 yield* Queue.offer(subscription, 1);
71 assert.equal(yield* Queue.take(received), 1);
72
73 const boom = yield* Effect.exit(client.client.boom());
74 assert.isTrue(Exit.hasDies(boom));
75 // The server reports before it answers, so the log is already written.
76 const logged = yield* Queue.clear(logs);
77 assert.deepEqual(logged.map(errorMessage), ["handler bug"]);
78
79 // The sibling subscription on the same client keeps delivering.
80 yield* Queue.offer(subscription, 2);
81 const next = yield* Queue.take(received).pipe(
82 Effect.raceFirst(Fiber.await(sibling).pipe(Effect.as("subscription ended"))),
83 );
84 assert.equal(next, 2);
85 assert.equal(yield* Queue.size(logs), 0);
86 }),
87 ).pipe(Effect.scoped),
88 );
89
90 it.effect("logs defects, not typed failures or interrupts", () =>
91 withErrorLogs((logs) =>
92 Effect.gen(function* () {
93 yield* ErrorReporter.report(Cause.fail(new Error("expected")));
94 yield* ErrorReporter.report(Cause.interrupt());
95 yield* ErrorReporter.report(Cause.die(new Error("bug")));
96
97 // Reporters log synchronously, so the logs are already written.
98 assert.deepEqual((yield* Queue.clear(logs)).map(errorMessage), ["bug"]);
99 }).pipe(Effect.provide(DefectReporter.layer)),
100 ),
101 );
102});
apps/server/src/observability/DefectReporter.ts created+24
......@@ -0,0 +1,24 @@
1import * as Cause from "effect/Cause";
2import * as Effect from "effect/Effect";
3import * as ErrorReporter from "effect/ErrorReporter";
4
5/**
6 * Logs the defects of WebSocket RPC handlers. RpcServer reports every failed
7 * handler exit; typed failures are expected responses, so only dies are logged.
8 */
9const reporter: ErrorReporter.ErrorReporter = {
10 [ErrorReporter.TypeId]: ErrorReporter.TypeId,
11 report: ({ cause, fiber }) => {
12 for (const reason of cause.reasons) {
13 if (reason._tag !== "Die" || ErrorReporter.isIgnored(reason.defect)) continue;
14 // Reporters are called synchronously from the failing fiber. Logging with
15 // its context keeps its loggers and annotations, and a fork cannot throw
16 // back into the server that reported.
17 Effect.runForkWith(fiber.context)(
18 Effect.logError("Unhandled defect", Cause.fromReasons([reason])),
19 );
20 }
21 },
22};
23
24export const layer = ErrorReporter.layer([reporter]);
apps/server/src/ws.ts+13-1
......@@ -210,6 +210,7 @@ import * as WorktreeSetupTracker from "./project/WorktreeSetupTracker.ts";
210210import * as ServerEnvironment from "./environment/ServerEnvironment.ts";
211211import * as DirectEndpoints from "./environment/DirectEndpoints.ts";
212212import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts";
213import * as DefectReporter from "./observability/DefectReporter.ts";
213214import * as BackgroundPolicy from "./background/BackgroundPolicy.ts";
214215import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts";
215216import { requiredScopeForDeviceList, rpcAuthorizationError } from "./auth/RpcAuthorization.ts";
......@@ -3787,6 +3788,14 @@ const layerWsRpc = (
37873788 }),
37883789 );
37893790
3791// A defect in a handler's effect fails only its own request. RpcServer's default
3792// sends a socket-level Defect frame instead, and the client ends every pending
3793// request on the socket with it. DefectReporter logs these defects.
3794export const WS_RPC_SERVER_OPTIONS = {
3795 disableTracing: true,
3796 disableFatalDefects: true,
3797} as const;
3798
37903799export const layer = Layer.unwrap(
37913800 Effect.gen(function* () {
37923801 const previewAutomationBroker = yield* PreviewAutomationBroker.PreviewAutomationBroker;
......@@ -3830,7 +3839,7 @@ export const layer = Layer.unwrap(
38303839 yield* analytics.record("client.connected", clientAnalyticsProps);
38313840 const rpcWebSocketHttpEffect = yield* Effect.gen(function* () {
38323841 const { protocol, httpEffect } = yield* RpcServer.makeProtocolWithHttpEffectWebsocket;
3833 yield* RpcServer.make(ServerWsRpcGroup, { disableTracing: true }).pipe(
3842 yield* RpcServer.make(ServerWsRpcGroup, WS_RPC_SERVER_OPTIONS).pipe(
38343843 Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)),
38353844 Effect.provide(RpcAuthorization.layer(session.scopes)),
38363845 Effect.forkScoped,
......@@ -3847,6 +3856,9 @@ export const layer = Layer.unwrap(
38473856 serverBrowser,
38483857 ).pipe(
38493858 Layer.provideMerge(RpcSerialization.layerJson),
3859 // Request fibers run in the handlers' context, so this reporter sees
3860 // their defects, not the rest of the server's.
3861 Layer.provide(DefectReporter.layer),
38503862 Layer.provide(Layer.succeed(SqlClient.SqlClient, sql)),
38513863 Layer.provide(AgentSessionScanner.layer),
38523864 Layer.provide(ProviderMaintenanceRunner.layer),
packages/effect-acp/src/_internal/shared.ts+17
......@@ -1,3 +1,4 @@
1import * as Cause from "effect/Cause";
12import * as Effect from "effect/Effect";
23import * as Schema from "effect/Schema";
34import { RpcClientError } from "effect/rpc";
......@@ -29,6 +30,19 @@ export const callRpc = <A>(
2930 }),
3031 );
3132
33/**
34 * Runs a notification handler so it cannot stop the caller: a typed failure is
35 * dropped, as notifications have no reply, and a defect is logged.
36 */
37export const isolateNotificationHandler = <E, R>(effect: Effect.Effect<void, E, R>) =>
38 effect.pipe(
39 Effect.catchCause((cause) =>
40 Cause.hasDies(cause)
41 ? Effect.logError("ACP notification handler failed", cause)
42 : Effect.void,
43 ),
44 );
45
3246export const runHandler = Effect.fnUntraced(function* <A, B, Args extends ReadonlyArray<unknown>>(
3347 handler: ((payload: A, ...args: Args) => Effect.Effect<B, AcpError.AcpError>) | undefined,
3448 payload: A,
......@@ -39,6 +53,9 @@ export const runHandler = Effect.fnUntraced(function* <A, B, Args extends Readon
3953 return yield* Effect.fail(AcpError.AcpRequestError.methodNotFound(method).toProtocolError());
4054 }
4155 return yield* handler(payload, ...args).pipe(
56 Effect.tapDefect((defect) =>
57 Effect.logError(`ACP request handler failed for '${method}'`, defect),
58 ),
4259 Effect.mapError((error) =>
4360 AcpError.AcpRequestError.fromCoreHandlerError(error, method).toProtocolError(),
4461 ),
packages/effect-acp/src/agent.test.ts+38
......@@ -35,6 +35,14 @@ const SessionCancelNotification = jsonRpcNotification(
3535const ExtPingNotification = jsonRpcNotification("x/ping", Schema.Struct({ count: Schema.Number }));
3636const ExtRequest = jsonRpcRequest("x/test", Schema.Struct({ hello: Schema.String }));
3737const ExtResponse = jsonRpcResponse(Schema.Struct({ ok: Schema.Boolean }));
38/** A response whose cause is a handler's defect, as RpcServer encodes it. */
39const DieResponse = Schema.Struct({
40 id: Schema.Number,
41 error: Schema.Struct({
42 _tag: Schema.Literal("Cause"),
43 data: Schema.Tuple([Schema.Struct({ _tag: Schema.Literal("Die") })]),
44 }),
45});
3846const decodeRequestPermissionRequest = Schema.decodeEffect(
3947 Schema.fromJsonString(RequestPermissionRequest),
4048);
......@@ -296,3 +304,33 @@ it.effect("effect-acp agent uses distinct ids for RPC calls and extension reques
296304 }).pipe(Effect.provide(context), Effect.ensuring(Scope.close(scope, Exit.void)));
297305 }),
298306);
307
308it.effect("effect-acp agent answers a request whose handler dies with an error for it", () =>
309 Effect.gen(function* () {
310 const { stdio, input, output } = yield* makeInMemoryStdio();
311 const scope = yield* Scope.make();
312 const context = yield* Layer.buildWithScope(AcpAgent.layer(stdio), scope);
313 const agent = yield* Effect.service(AcpAgent.AcpAgent).pipe(Effect.provide(context));
314 yield* agent.handleInitialize(() => Effect.die(new Error("handler bug")));
315
316 yield* Queue.offer(
317 input,
318 yield* encodeJsonl(InitializeRequest, {
319 jsonrpc: "2.0",
320 id: 7,
321 method: "initialize",
322 params: {
323 protocolVersion: 2,
324 capabilities: {},
325 info: { name: "effect-acp-test", version: "0.0.0" },
326 },
327 headers: [],
328 }),
329 );
330 const response = yield* Queue.take(output).pipe(
331 Effect.flatMap(Schema.decodeEffect(Schema.fromJsonString(DieResponse))),
332 );
333 assert.equal(response.id, 7);
334 yield* Scope.close(scope, Exit.void);
335 }),
336);
packages/effect-acp/src/agent.ts+11-2
......@@ -1,5 +1,6 @@
11import * as Context from "effect/Context";
22import * as Effect from "effect/Effect";
3import * as ErrorReporter from "effect/ErrorReporter";
34import * as Layer from "effect/Layer";
45import * as Schema from "effect/Schema";
56import * as Scope from "effect/Scope";
......@@ -18,6 +19,7 @@ import {
1819 callRpc,
1920 decodeExtNotificationRegistration,
2021 decodeExtRequestRegistration,
22 isolateNotificationHandler,
2123 runHandler,
2224} from "./_internal/shared.ts";
2325
......@@ -280,7 +282,11 @@ export const make = Effect.fn("effect-acp/AcpAgent.make")(function* (
280282 ),
281283 ),
282284 Effect.flatMap((decoded) =>
283 Effect.forEach(cancelHandlers, (handler) => handler(decoded), { discard: true }),
285 Effect.forEach(
286 cancelHandlers,
287 (handler) => isolateNotificationHandler(handler(decoded)),
288 { discard: true },
289 ),
284290 ),
285291 );
286292 }
......@@ -421,7 +427,10 @@ export const make = Effect.fn("effect-acp/AcpAgent.make")(function* (
421427 }),
422428 );
423429
424 yield* RpcServer.make(AcpRpcs.AgentRpcs).pipe(
430 yield* RpcServer.make(AcpRpcs.AgentRpcs, { disableFatalDefects: true }).pipe(
431 // runHandler logs handler defects with their method. A reporter inherited
432 // from the caller (a WebSocket request, say) would log them again.
433 Effect.provideService(ErrorReporter.CurrentErrorReporters, new Set()),
425434 Effect.provideService(RpcServer.Protocol, transport.serverProtocol),
426435 Effect.provide(layerAgentHandler),
427436 Effect.forkScoped,
packages/effect-acp/src/client.test.ts+123
......@@ -5,11 +5,13 @@ import * as Effect from "effect/Effect";
55import * as Exit from "effect/Exit";
66import * as Fiber from "effect/Fiber";
77import * as Layer from "effect/Layer";
8import * as Logger from "effect/Logger";
89import * as Queue from "effect/Queue";
910import * as Ref from "effect/Ref";
1011import * as Schema from "effect/Schema";
1112import * as Scope from "effect/Scope";
1213import * as Stream from "effect/Stream";
14import * as TestClock from "effect/testing/TestClock";
1315import { ChildProcess, ChildProcessSpawner } from "effect/process";
1416
1517import * as NodeServices from "@effect/platform-node/NodeServices";
......@@ -50,6 +52,16 @@ const PermissionRequest = jsonRpcRequest(
5052const PermissionResponse = jsonRpcResponse(AcpSchema.RequestPermissionResponse);
5153const ElicitationRequest = jsonRpcRequest("elicitation/create", AcpSchema.CreateElicitationRequest);
5254const ElicitationResponse = jsonRpcResponse(AcpSchema.CreateElicitationResponse);
55/** A JSON-RPC error response; only its id and the presence of an error matter here. */
56const ErrorResponse = Schema.Struct({ id: Schema.String, error: Schema.Unknown });
57/** A response whose cause is a handler's defect, as RpcServer encodes it. */
58const DieResponse = Schema.Struct({
59 id: Schema.String,
60 error: Schema.Struct({
61 _tag: Schema.Literal("Cause"),
62 data: Schema.Tuple([Schema.Struct({ _tag: Schema.Literal("Die") })]),
63 }),
64});
5365const decodePromptRequestLine = Schema.decodeEffect(Schema.fromJsonString(PromptRequest));
5466const XAiPromptCompleteNotification = jsonRpcNotification(
5567 "_x.ai/session/prompt_complete",
......@@ -1130,6 +1142,117 @@ it.layer(NodeServices.layer)("effect-acp client", (it) => {
11301142 }),
11311143 );
11321144
1145 it.effect("answers each request whose handler dies, and keeps reading", () =>
1146 Effect.gen(function* () {
1147 const errorLogs = yield* Queue.unbounded<string>();
1148 const logger = Logger.make(({ logLevel, message }) => {
1149 if (logLevel === "Error") Queue.offerUnsafe(errorLogs, String([message].flat()[0]));
1150 });
1151 const { stdio, input, output } = yield* makeInMemoryStdio();
1152 const scope = yield* Scope.make();
1153 const acp = yield* AcpClient.make(stdio).pipe(
1154 Effect.provideService(Scope.Scope, scope),
1155 Effect.provide(Logger.layer([logger])),
1156 );
1157 const bug = () => Effect.die(new Error("handler bug"));
1158 yield* acp.handleRequestPermission(bug);
1159 yield* acp.handleExtRequest("x/dies", Schema.Unknown, bug);
1160 yield* acp.handleExtNotification("x/notification-dies", Schema.Unknown, bug);
1161 yield* acp.handleSessionUpdate(bug);
1162 const updates = yield* Queue.unbounded<string>();
1163 yield* acp.handleSessionUpdate((notification) =>
1164 Queue.offer(updates, notification.sessionId).pipe(Effect.asVoid),
1165 );
1166 yield* acp.handleExtRequest("x/test", Schema.Struct({ hello: Schema.String }), () =>
1167 Effect.succeed({ ok: true }),
1168 );
1169 const send = <E>(line: Effect.Effect<Uint8Array, E>) =>
1170 Effect.flatMap(line, (bytes) => Queue.offer(input, bytes));
1171 const decodeDie = Schema.decodeEffect(Schema.fromJsonString(DieResponse));
1172 // A stopped reader leaves these waiting; fail instead of hanging.
1173 const next = <A>(queue: Queue.Dequeue<A>) =>
1174 TestClock.withLive(Queue.take(queue).pipe(Effect.timeout("2 seconds")));
1175
1176 yield* send(
1177 encodeJsonl(PermissionRequest, {
1178 jsonrpc: "2.0",
1179 id: "permission-a",
1180 method: "session/request_permission",
1181 params: {
1182 sessionId: "session-1",
1183 title: "Tool",
1184 subject: {
1185 type: "tool_call" as const,
1186 toolCall: { toolCallId: "tool-1", title: "Tool" },
1187 },
1188 options: [{ optionId: "allow", name: "Allow", kind: "allow_once" as const }],
1189 },
1190 headers: [],
1191 }),
1192 );
1193 assert.equal((yield* next(output).pipe(Effect.flatMap(decodeDie))).id, "permission-a");
1194
1195 yield* send(
1196 encodeJsonl(jsonRpcRequest("x/dies", Schema.Unknown), {
1197 jsonrpc: "2.0",
1198 id: "ext-a",
1199 method: "x/dies",
1200 params: {},
1201 headers: [],
1202 }),
1203 );
1204 const extDied = yield* next(output).pipe(
1205 Effect.flatMap(Schema.decodeEffect(Schema.fromJsonString(ErrorResponse))),
1206 );
1207 assert.equal(extDied.id, "ext-a");
1208
1209 yield* send(
1210 encodeJsonl(jsonRpcNotification("x/notification-dies", Schema.Unknown), {
1211 jsonrpc: "2.0",
1212 method: "x/notification-dies",
1213 params: {},
1214 }),
1215 );
1216 yield* send(
1217 encodeJsonl(SessionUpdateNotification, {
1218 jsonrpc: "2.0",
1219 method: "session/update",
1220 params: {
1221 sessionId: "session-1",
1222 update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: "hi" } },
1223 },
1224 }),
1225 );
1226
1227 // The next session handler still ran.
1228 assert.equal(yield* next(updates), "session-1");
1229
1230 // The reader survived all three: a later request is still answered.
1231 yield* send(
1232 encodeJsonl(ExtRequest, {
1233 jsonrpc: "2.0",
1234 id: "ext-b",
1235 method: "x/test",
1236 params: { hello: "world" },
1237 headers: [],
1238 }),
1239 );
1240 const answered = yield* next(output).pipe(
1241 Effect.flatMap(Schema.decodeEffect(Schema.fromJsonString(ExtResponse))),
1242 );
1243 assert.deepEqual([answered.id, answered.result], ["ext-b", { ok: true }]);
1244
1245 // Every defect was logged at Error, once.
1246 assert.deepEqual((yield* Queue.clear(errorLogs)).toSorted(), [
1247 "ACP extension request handler failed for 'x/dies'",
1248 "ACP notification handler failed",
1249 "ACP notification handler failed",
1250 "ACP request handler failed for 'session/request_permission'",
1251 ]);
1252 yield* Scope.close(scope, Exit.void);
1253 }),
1254 );
1255
11331256 it.effect("answers elicitation/create with the flat action shape", () =>
11341257 Effect.gen(function* () {
11351258 const { stdio, input, output } = yield* makeInMemoryStdio();
packages/effect-acp/src/client.ts+12-4
......@@ -1,6 +1,7 @@
11import * as Context from "effect/Context";
22import * as Deferred from "effect/Deferred";
33import * as Effect from "effect/Effect";
4import * as ErrorReporter from "effect/ErrorReporter";
45import * as Layer from "effect/Layer";
56import * as Schema from "effect/Schema";
67import * as Predicate from "effect/Predicate";
......@@ -22,6 +23,7 @@ import {
2223 callRpc,
2324 decodeExtNotificationRegistration,
2425 decodeExtRequestRegistration,
26 isolateNotificationHandler,
2527 runHandler,
2628} from "./_internal/shared.ts";
2729import { makeChildStdio, makeTerminationError } from "./_internal/stdio.ts";
......@@ -838,9 +840,12 @@ export const make = Effect.fn("effect-acp/AcpClient.make")(function* (
838840 registration: BufferedNotificationHandler<A>,
839841 notification: A,
840842 ) =>
841 Effect.forEach(registration.handlers, (handler) => handler(notification).pipe(Effect.ignore), {
842 discard: true,
843 });
843 // One handler failing or dying does not stop the others, or the reader.
844 Effect.forEach(
845 registration.handlers,
846 (handler) => isolateNotificationHandler(handler(notification)),
847 { discard: true },
848 );
844849
845850 const flushBufferedNotifications = <A>(registration: BufferedNotificationHandler<A>) =>
846851 Effect.suspend(() => {
......@@ -1099,7 +1104,10 @@ export const make = Effect.fn("effect-acp/AcpClient.make")(function* (
10991104 }),
11001105 );
11011106
1102 yield* RpcServer.make(AcpRpcs.CompatClientRpcs).pipe(
1107 yield* RpcServer.make(AcpRpcs.CompatClientRpcs, { disableFatalDefects: true }).pipe(
1108 // runHandler logs handler defects with their method. A reporter inherited
1109 // from the caller (a WebSocket request, say) would log them again.
1110 Effect.provideService(ErrorReporter.CurrentErrorReporters, new Set()),
11031111 Effect.provideService(RpcServer.Protocol, transport.serverProtocol),
11041112 Effect.provide(layerClientHandler),
11051113 Effect.forkScoped,
packages/effect-acp/src/protocol.test.ts+69
......@@ -11,6 +11,7 @@ import * as Sink from "effect/Sink";
1111import * as Stdio from "effect/Stdio";
1212import * as Stream from "effect/Stream";
1313import * as Ref from "effect/Ref";
14import * as TestClock from "effect/testing/TestClock";
1415import { ChildProcess, ChildProcessSpawner } from "effect/process";
1516
1617import { it, assert } from "@effect/vitest";
......@@ -257,6 +258,42 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
257258 }),
258259 );
259260
261 it.effect("terminates when a callback on the reader dies", () =>
262 Effect.gen(function* () {
263 const { stdio, input } = yield* makeInMemoryStdio();
264 const termination = yield* Deferred.make<AcpError.AcpError>();
265 yield* AcpProtocol.makeAcpPatchedProtocol({
266 stdio,
267 serverRequestMethods: new Set(),
268 transformSessionUpdate: () => {
269 throw new Error("normalizer bug");
270 },
271 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
272 });
273
274 yield* Queue.offer(
275 input,
276 encoder.encode(
277 `${encodeUnknownJsonString({
278 jsonrpc: "2.0",
279 method: "session/update",
280 params: {
281 sessionId: "session-1",
282 update: { sessionUpdate: "plan", entries: [] },
283 },
284 })}\n`,
285 ),
286 );
287
288 // Pending requests are failed through termination instead of hanging.
289 const error = yield* TestClock.withLive(
290 Deferred.await(termination).pipe(Effect.timeout("2 seconds")),
291 );
292 assert.instanceOf(error, AcpError.AcpTransportError);
293 assert.equal((error as AcpError.AcpTransportError).operation, "read-input-stream");
294 }),
295 );
296
260297 it.effect("logs outgoing notifications when logOutgoing is enabled", () =>
261298 Effect.gen(function* () {
262299 const { stdio } = yield* makeInMemoryStdio();
......@@ -500,6 +537,38 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
500537 }),
501538 );
502539
540 it.effect("answers an extension request whose handler also dies as an internal error", () =>
541 Effect.gen(function* () {
542 const { stdio, input, output } = yield* makeInMemoryStdio();
543 yield* AcpProtocol.makeAcpPatchedProtocol({
544 stdio,
545 serverRequestMethods: new Set(),
546 // The typed failure must not hide the defect from its cleanup.
547 onExtRequest: () =>
548 Effect.fail(AcpError.AcpRequestError.invalidParams("bad params")).pipe(
549 Effect.ensuring(Effect.die(new Error("cleanup bug"))),
550 ),
551 });
552
553 yield* Queue.offer(
554 input,
555 yield* encodeJsonl(ExtRequest, {
556 jsonrpc: "2.0",
557 id: 9,
558 method: "x/test",
559 params: { hello: "world" },
560 headers: [],
561 }),
562 );
563
564 const response = yield* Schema.decodeUnknownEffect(
565 Schema.fromJsonString(JsonRpcErrorResponse),
566 )(yield* Queue.take(output));
567 assert.equal(response.id, 9);
568 assert.equal(response.error.code, -32603);
569 }),
570 );
571
503572 it.effect("preserves numeric ids for inbound extension requests", () =>
504573 Effect.gen(function* () {
505574 const { stdio, input, output } = yield* makeInMemoryStdio();
packages/effect-acp/src/protocol.ts+45-9
......@@ -3,6 +3,7 @@ import * as Cause from "effect/Cause";
33import * as Effect from "effect/Effect";
44import * as Deferred from "effect/Deferred";
55import * as Exit from "effect/Exit";
6import * as Option from "effect/Option";
67import * as Queue from "effect/Queue";
78import * as Ref from "effect/Ref";
89import type * as PlatformError from "effect/PlatformError";
......@@ -20,6 +21,7 @@ import * as AcpSchema from "./schema.ts";
2021import * as AcpSchemaV1 from "./_generated/schema-v1.gen.ts";
2122import { CLIENT_METHODS } from "./_generated/meta.gen.ts";
2223import * as AcpError from "./errors.ts";
24import { isolateNotificationHandler } from "./_internal/shared.ts";
2325const isAcpError = Schema.is(AcpError.AcpError);
2426
2527export interface AcpProtocolLogEvent {
......@@ -385,8 +387,10 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
385387 const dispatchNotification = (notification: AcpIncomingNotification) =>
386388 Queue.offer(notificationQueue, notification).pipe(
387389 Effect.andThen(
390 // A failing or dying handler must not stop the reader, or every later
391 // message on the connection goes unanswered.
388392 options.onNotification
389 ? options.onNotification(notification).pipe(Effect.ignore)
393 ? isolateNotificationHandler(options.onNotification(notification))
390394 : Effect.void,
391395 ),
392396 Effect.asVoid,
......@@ -458,12 +462,38 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
458462 method: message.tag,
459463 })
460464 .pipe(
461 Effect.matchEffect({
462 onFailure: (error) =>
463 respondWithError(
464 message.id,
465 AcpError.AcpRequestError.fromExtensionHandlerError(error, message.tag),
466 ),
465 Effect.matchCauseEffect({
466 // A dying handler answers its own request, like a core handler, and
467 // leaves the reader running. A defect wins over a typed failure in
468 // the same cause, so it is never hidden behind an expected error.
469 onFailure: (cause) => {
470 const failure = Cause.hasDies(cause) ? Option.none() : Cause.findErrorOption(cause);
471 if (Option.isSome(failure)) {
472 return respondWithError(
473 message.id,
474 AcpError.AcpRequestError.fromExtensionHandlerError(failure.value, message.tag),
475 );
476 }
477 return Effect.logError(
478 `ACP extension request handler failed for '${message.tag}'`,
479 cause,
480 ).pipe(
481 Effect.andThen(
482 respondWithError(
483 message.id,
484 AcpError.AcpRequestError.internalError(
485 `ACP extension request handler failed for method '${message.tag}'`,
486 undefined,
487 {
488 method: message.tag,
489 operation: "handle-extension-request",
490 cause: Cause.squash(cause),
491 },
492 ),
493 ),
494 ),
495 );
496 },
467497 onSuccess: (value) => respondWithSuccess(message.id, value),
468498 }),
469499 );
......@@ -685,8 +715,14 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
685715 ),
686716 ),
687717 ),
688 Effect.matchEffect({
689 onFailure: (error) => {
718 // Anything that ends the reader, including a defect in a callback it runs,
719 // terminates the connection so pending requests fail instead of hanging.
720 Effect.matchCauseEffect({
721 onFailure: (cause) => {
722 // The reader's own interruption never gets here: an interrupted fiber
723 // skips failure handlers. An interrupt raised inside it ends it too.
724 const failure = Cause.findErrorOption(cause);
725 const error = Option.isSome(failure) ? failure.value : Cause.squash(cause);
690726 const normalized: AcpError.AcpError = isAcpError(error)
691727 ? error
692728 : new AcpError.AcpTransportError({