authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-06-20 15:12:31-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-06-20 15:12:31-07:00
log6d8d995621065e0513def2fa28bf0f7d8b54dd90
tree1789927186d2595e39444380775fc8110831c4c4
parentb6c590af389d09400fa4f99a56ffc33e9ed4ebf1
signature Signed by PGP key B5690EEEBB952194

[codex] Correlate protocol request failures (#3433)

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

8 files changed, 314 insertions(+), 59 deletions(-)

packages/effect-acp/src/_internal/shared.ts+1-1
...@@ -12,7 +12,7 @@ export const callRpc = <A>(...@@ -12,7 +12,7 @@ export const callRpc = <A>(
12): Effect.Effect<A, AcpError.AcpError> =>12): Effect.Effect<A, AcpError.AcpError> =>
13 effect.pipe(13 effect.pipe(
14 Effect.catchIf(isError, (error) =>14 Effect.catchIf(isError, (error) =>
15 Effect.fail(AcpError.AcpRequestError.fromProtocolError(error)),15 Effect.fail(AcpError.AcpRequestError.fromProtocolError(error, { method })),
16 ),16 ),
17 Effect.catchTags({17 Effect.catchTags({
18 RpcClientError: (cause) =>18 RpcClientError: (cause) =>
packages/effect-acp/src/errors.test.ts+2
...@@ -51,6 +51,8 @@ describe("effect-acp errors", () => {...@@ -51,6 +51,8 @@ describe("effect-acp errors", () => {
51 code: -32602,51 code: -32602,
52 errorMessage: "Invalid params",52 errorMessage: "Invalid params",
53 data: { field: "sessionId" },53 data: { field: "sessionId" },
54 method: "session/load",
55 operation: "receive-response",
54 });56 });
55 });57 });
56 });58 });
packages/effect-acp/src/errors.ts+65-1
...@@ -5,8 +5,11 @@ import * as AcpSchema from "./_generated/schema.gen.ts";...@@ -5,8 +5,11 @@ import * as AcpSchema from "./_generated/schema.gen.ts";
55
6export const AcpRequestOperation = Schema.Literals([6export const AcpRequestOperation = Schema.Literals([
7 "decode-extension-request-payload",7 "decode-extension-request-payload",
8 "encode-extension-response",
8 "handle-request",9 "handle-request",
9 "handle-extension-request",10 "handle-extension-request",
11 "receive-response",
12 "receive-streaming-response",
10]);13]);
11export type AcpRequestOperation = typeof AcpRequestOperation.Type;14export type AcpRequestOperation = typeof AcpRequestOperation.Type;
1215
...@@ -65,6 +68,7 @@ const schemaIssueDiagnostics = (root: SchemaIssue.Issue): AcpSchemaIssueDiagnost...@@ -65,6 +68,7 @@ const schemaIssueDiagnostics = (root: SchemaIssue.Issue): AcpSchemaIssueDiagnost
6568
66export interface AcpRequestDiagnostics {69export interface AcpRequestDiagnostics {
67 readonly method?: string;70 readonly method?: string;
71 readonly requestId?: string;
68 readonly operation?: AcpRequestOperation;72 readonly operation?: AcpRequestOperation;
69 readonly cause?: unknown;73 readonly cause?: unknown;
70 readonly issueCount?: number;74 readonly issueCount?: number;
...@@ -109,6 +113,7 @@ export class AcpProtocolParseError extends Schema.TaggedErrorClass<AcpProtocolPa...@@ -109,6 +113,7 @@ export class AcpProtocolParseError extends Schema.TaggedErrorClass<AcpProtocolPa
109 {113 {
110 operation: AcpProtocolParseOperation,114 operation: AcpProtocolParseOperation,
111 method: Schema.optionalKey(Schema.String),115 method: Schema.optionalKey(Schema.String),
116 requestId: Schema.optionalKey(Schema.String),
112 issueCount: Schema.optionalKey(Schema.Number),117 issueCount: Schema.optionalKey(Schema.Number),
113 issueKinds: Schema.optionalKey(Schema.Array(AcpSchemaIssueKind)),118 issueKinds: Schema.optionalKey(Schema.Array(AcpSchemaIssueKind)),
114 maximumPathDepth: Schema.optionalKey(Schema.Number),119 maximumPathDepth: Schema.optionalKey(Schema.Number),
...@@ -132,6 +137,19 @@ export class AcpProtocolParseError extends Schema.TaggedErrorClass<AcpProtocolPa...@@ -132,6 +137,19 @@ export class AcpProtocolParseError extends Schema.TaggedErrorClass<AcpProtocolPa
132 cause,137 cause,
133 });138 });
134 }139 }
140
141 static fromEncodingError(
142 method: string | undefined,
143 requestId: string | undefined,
144 cause: unknown,
145 ) {
146 return new AcpProtocolParseError({
147 operation: "encode-message",
148 ...(method === undefined ? {} : { method }),
149 ...(requestId === undefined ? {} : { requestId }),
150 cause,
151 });
152 }
135}153}
136154
137export class AcpTransportError extends Schema.TaggedErrorClass<AcpTransportError>()(155export class AcpTransportError extends Schema.TaggedErrorClass<AcpTransportError>()(
...@@ -167,6 +185,7 @@ export class AcpRequestError extends Schema.TaggedErrorClass<AcpRequestError>()(...@@ -167,6 +185,7 @@ export class AcpRequestError extends Schema.TaggedErrorClass<AcpRequestError>()(
167 errorMessage: Schema.String,185 errorMessage: Schema.String,
168 data: Schema.optional(Schema.Unknown),186 data: Schema.optional(Schema.Unknown),
169 method: Schema.optionalKey(Schema.String),187 method: Schema.optionalKey(Schema.String),
188 requestId: Schema.optionalKey(Schema.String),
170 operation: Schema.optionalKey(AcpRequestOperation),189 operation: Schema.optionalKey(AcpRequestOperation),
171 issueCount: Schema.optionalKey(Schema.Number),190 issueCount: Schema.optionalKey(Schema.Number),
172 issueKinds: Schema.optionalKey(Schema.Array(AcpSchemaIssueKind)),191 issueKinds: Schema.optionalKey(Schema.Array(AcpSchemaIssueKind)),
...@@ -177,14 +196,59 @@ export class AcpRequestError extends Schema.TaggedErrorClass<AcpRequestError>()(...@@ -177,14 +196,59 @@ export class AcpRequestError extends Schema.TaggedErrorClass<AcpRequestError>()(
177 return this.errorMessage;196 return this.errorMessage;
178 }197 }
179198
180 static fromProtocolError(error: AcpSchema.Error) {199 static fromProtocolError(
200 error: AcpSchema.Error,
201 context: {
202 readonly method: string;
203 readonly requestId?: string;
204 readonly cause?: unknown;
205 },
206 ) {
181 return new AcpRequestError({207 return new AcpRequestError({
182 code: error.code,208 code: error.code,
183 errorMessage: error.message,209 errorMessage: error.message,
184 ...(error.data !== undefined ? { data: error.data } : {}),210 ...(error.data !== undefined ? { data: error.data } : {}),
211 method: context.method,
212 ...(context.requestId === undefined ? {} : { requestId: context.requestId }),
213 operation: "receive-response",
214 cause: context.cause ?? error,
185 });215 });
186 }216 }
187217
218 static fromExtensionResponseFailure(method: string, requestId: string, cause: unknown) {
219 return AcpRequestError.internalError("Extension request failed", undefined, {
220 method,
221 requestId,
222 operation: "receive-response",
223 cause,
224 });
225 }
226
227 static fromExtensionResponseEncodingError(
228 method: string,
229 requestId: string,
230 cause: AcpProtocolParseError,
231 ) {
232 return AcpRequestError.internalError("Internal error", undefined, {
233 method,
234 requestId,
235 operation: "encode-extension-response",
236 cause,
237 });
238 }
239
240 static unsupportedStreamingResponse(method: string, requestId: string) {
241 return AcpRequestError.internalError(
242 "Streaming extension responses are not supported",
243 undefined,
244 {
245 method,
246 requestId,
247 operation: "receive-streaming-response",
248 },
249 );
250 }
251
188 static fromCoreHandlerError(error: AcpError, method: string) {252 static fromCoreHandlerError(error: AcpError, method: string) {
189 if (error._tag === "AcpRequestError") {253 if (error._tag === "AcpRequestError") {
190 return error;254 return error;
packages/effect-acp/src/protocol.test.ts+73-1
...@@ -260,15 +260,33 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {...@@ -260,15 +260,33 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
260 const bigintError = yield* transport.notify("x/test", 1n).pipe(Effect.flip);260 const bigintError = yield* transport.notify("x/test", 1n).pipe(Effect.flip);
261 assert.instanceOf(bigintError, AcpError.AcpProtocolParseError);261 assert.instanceOf(bigintError, AcpError.AcpProtocolParseError);
262 assert.equal(bigintError.operation, "encode-message");262 assert.equal(bigintError.operation, "encode-message");
263 assert.equal(bigintError.method, "x/test");
263 assert.instanceOf(bigintError.cause, TypeError);264 assert.instanceOf(bigintError.cause, TypeError);
264 assert.equal(bigintError.message, "ACP protocol operation 'encode-message' failed.");265 assert.equal(
266 bigintError.message,
267 "ACP protocol operation 'encode-message' failed for method 'x/test'.",
268 );
265269
266 const circular: Record<string, unknown> = {};270 const circular: Record<string, unknown> = {};
267 circular.self = circular;271 circular.self = circular;
268 const circularError = yield* transport.notify("x/test", circular).pipe(Effect.flip);272 const circularError = yield* transport.notify("x/test", circular).pipe(Effect.flip);
269 assert.instanceOf(circularError, AcpError.AcpProtocolParseError);273 assert.instanceOf(circularError, AcpError.AcpProtocolParseError);
270 assert.equal(circularError.operation, "encode-message");274 assert.equal(circularError.operation, "encode-message");
275 assert.equal(circularError.method, "x/test");
271 assert.instanceOf(circularError.cause, TypeError);276 assert.instanceOf(circularError.cause, TypeError);
277
278 const requestError = yield* transport.request("x/request", 1n).pipe(
279 Effect.match({
280 onFailure: (error) => error,
281 onSuccess: () => assert.fail("Expected request encoding to fail"),
282 }),
283 );
284 assert.instanceOf(requestError, AcpError.AcpProtocolParseError);
285 assert.deepInclude(requestError, {
286 operation: "encode-message",
287 method: "x/request",
288 requestId: "1",
289 });
272 }),290 }),
273 );291 );
274292
...@@ -310,6 +328,60 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {...@@ -310,6 +328,60 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
310 }),328 }),
311 );329 );
312330
331 it.effect("correlates extension response errors with the originating request", () =>
332 Effect.gen(function* () {
333 const { stdio, input, output } = yield* makeInMemoryStdio();
334 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
335 stdio,
336 serverRequestMethods: new Set(),
337 });
338
339 const response = yield* transport
340 .request("x/private", { hello: "world" })
341 .pipe(Effect.forkScoped);
342 yield* Queue.take(output);
343 yield* Queue.offer(
344 input,
345 encoder.encode(
346 `${encodeUnknownJsonString({
347 jsonrpc: "2.0",
348 id: 1,
349 error: {
350 _tag: "Cause",
351 code: -32602,
352 message: "Invalid params",
353 data: [
354 {
355 _tag: "Fail",
356 error: {
357 code: -32602,
358 message: "Invalid params",
359 data: { field: "hello" },
360 },
361 },
362 ],
363 },
364 })}\n`,
365 ),
366 );
367
368 const error = yield* Fiber.join(response).pipe(
369 Effect.match({
370 onFailure: (error) => error,
371 onSuccess: () => assert.fail("Expected extension request to fail"),
372 }),
373 );
374 assert.instanceOf(error, AcpError.AcpRequestError);
375 assert.deepInclude(error, {
376 code: -32602,
377 errorMessage: "Invalid params",
378 method: "x/private",
379 requestId: "1",
380 operation: "receive-response",
381 });
382 }),
383 );
384
313 it.effect("preserves zero-valued ids for inbound core client requests", () =>385 it.effect("preserves zero-valued ids for inbound core client requests", () =>
314 Effect.gen(function* () {386 Effect.gen(function* () {
315 const { stdio, input, output } = yield* makeInMemoryStdio();387 const { stdio, input, output } = yield* makeInMemoryStdio();
packages/effect-acp/src/protocol.ts+67-31
...@@ -66,6 +66,11 @@ export interface AcpPatchedProtocol {...@@ -66,6 +66,11 @@ export interface AcpPatchedProtocol {
66 readonly notify: (method: string, payload: unknown) => Effect.Effect<void, AcpError.AcpError>;66 readonly notify: (method: string, payload: unknown) => Effect.Effect<void, AcpError.AcpError>;
67}67}
6868
69interface AcpPendingRequest {
70 readonly deferred: Deferred.Deferred<unknown, AcpError.AcpError>;
71 readonly method: string;
72}
73
69const decodeSessionUpdate = Schema.decodeUnknownEffect(AcpSchema.SessionNotification);74const decodeSessionUpdate = Schema.decodeUnknownEffect(AcpSchema.SessionNotification);
70const decodeElicitationComplete = Schema.decodeUnknownEffect(75const decodeElicitationComplete = Schema.decodeUnknownEffect(
71 AcpSchema.ElicitationCompleteNotification,76 AcpSchema.ElicitationCompleteNotification,
...@@ -83,9 +88,7 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi...@@ -83,9 +88,7 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
83 const outgoing = yield* Queue.unbounded<string | Uint8Array, Cause.Done<void>>();88 const outgoing = yield* Queue.unbounded<string | Uint8Array, Cause.Done<void>>();
84 const nextRequestId = yield* Ref.make(1n);89 const nextRequestId = yield* Ref.make(1n);
85 const terminationHandled = yield* Ref.make(false);90 const terminationHandled = yield* Ref.make(false);
86 const extPending = yield* Ref.make(91 const extPending = yield* Ref.make(new Map<string, AcpPendingRequest>());
87 new Map<string, Deferred.Deferred<unknown, AcpError.AcpError>>(),
88 );
8992
90 const logProtocol = (event: AcpProtocolLogEvent) => {93 const logProtocol = (event: AcpProtocolLogEvent) => {
91 if (event.direction === "incoming" && !options.logIncoming) {94 if (event.direction === "incoming" && !options.logIncoming) {
...@@ -109,13 +112,17 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi...@@ -109,13 +112,17 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
109 payload: message,112 payload: message,
110 });113 });
111114
115 const method = message._tag === "Request" ? message.tag : undefined;
116 const encodedRequestId =
117 message._tag === "Request"
118 ? message.id
119 : "requestId" in message
120 ? message.requestId
121 : undefined;
122 const requestId = encodedRequestId === "" ? undefined : encodedRequestId;
112 const encoded = yield* Effect.try({123 const encoded = yield* Effect.try({
113 try: () => parser.encode(message),124 try: () => parser.encode(message),
114 catch: (cause) =>125 catch: (cause) => AcpError.AcpProtocolParseError.fromEncodingError(method, requestId, cause),
115 new AcpError.AcpProtocolParseError({
116 operation: "encode-message",
117 cause,
118 }),
119 });126 });
120127
121 if (encoded) {128 if (encoded) {
...@@ -131,16 +138,16 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi...@@ -131,16 +138,16 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
131138
132 const resolveExtPending = (139 const resolveExtPending = (
133 requestId: string,140 requestId: string,
134 onFound: (deferred: Deferred.Deferred<unknown, AcpError.AcpError>) => Effect.Effect<void>,141 onFound: (pendingRequest: AcpPendingRequest) => Effect.Effect<void>,
135 ) =>142 ) =>
136 Ref.modify(extPending, (pending) => {143 Ref.modify(extPending, (pending) => {
137 const deferred = pending.get(requestId);144 const pendingRequest = pending.get(requestId);
138 if (!deferred) {145 if (!pendingRequest) {
139 return [Effect.void, pending] as const;146 return [Effect.void, pending] as const;
140 }147 }
141 const next = new Map(pending);148 const next = new Map(pending);
142 next.delete(requestId);149 next.delete(requestId);
143 return [onFound(deferred), next] as const;150 return [onFound(pendingRequest), next] as const;
144 }).pipe(Effect.flatten);151 }).pipe(Effect.flatten);
145152
146 const removeExtPending = (requestId: string) =>153 const removeExtPending = (requestId: string) =>
...@@ -154,15 +161,15 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi...@@ -154,15 +161,15 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
154 });161 });
155162
156 const completeExtPendingFailure = (requestId: string, error: AcpError.AcpError) =>163 const completeExtPendingFailure = (requestId: string, error: AcpError.AcpError) =>
157 resolveExtPending(requestId, (deferred) => Deferred.fail(deferred, error));164 resolveExtPending(requestId, ({ deferred }) => Deferred.fail(deferred, error));
158165
159 const completeExtPendingSuccess = (requestId: string, value: unknown) =>166 const completeExtPendingSuccess = (requestId: string, value: unknown) =>
160 resolveExtPending(requestId, (deferred) => Deferred.succeed(deferred, value));167 resolveExtPending(requestId, ({ deferred }) => Deferred.succeed(deferred, value));
161168
162 const failAllExtPending = (error: AcpError.AcpError) =>169 const failAllExtPending = (error: AcpError.AcpError) =>
163 Ref.getAndSet(extPending, new Map()).pipe(170 Ref.getAndSet(extPending, new Map()).pipe(
164 Effect.flatMap((pending) =>171 Effect.flatMap((pending) =>
165 Effect.forEach([...pending.values()], (deferred) => Deferred.fail(deferred, error), {172 Effect.forEach([...pending.values()], ({ deferred }) => Deferred.fail(deferred, error), {
166 discard: true,173 discard: true,
167 }),174 }),
168 ),175 ),
...@@ -303,7 +310,26 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi...@@ -303,7 +310,26 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
303310
304 if (!options.serverRequestMethods.has(message.tag)) {311 if (!options.serverRequestMethods.has(message.tag)) {
305 return handleExtRequest(message).pipe(312 return handleExtRequest(message).pipe(
306 Effect.catch(() => respondWithError(message.id, AcpError.AcpRequestError.internalError())),313 Effect.catchTags({
314 AcpProtocolParseError: (error) =>
315 Effect.logWarning(error).pipe(
316 Effect.annotateLogs({
317 method: message.tag,
318 requestId: message.id,
319 operation: error.operation,
320 }),
321 Effect.andThen(
322 respondWithError(
323 message.id,
324 AcpError.AcpRequestError.fromExtensionResponseEncodingError(
325 message.tag,
326 message.id,
327 error,
328 ),
329 ),
330 ),
331 ),
332 }),
307 Effect.asVoid,333 Effect.asVoid,
308 );334 );
309 }335 }
...@@ -314,7 +340,8 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi...@@ -314,7 +340,8 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
314 const handleExitEncoded = (message: RpcMessage.ResponseExitEncoded) =>340 const handleExitEncoded = (message: RpcMessage.ResponseExitEncoded) =>
315 Ref.get(extPending).pipe(341 Ref.get(extPending).pipe(
316 Effect.flatMap((pending) => {342 Effect.flatMap((pending) => {
317 if (!pending.has(message.requestId)) {343 const pendingRequest = pending.get(message.requestId);
344 if (!pendingRequest) {
318 return Queue.offer(clientQueue, message).pipe(Effect.asVoid);345 return Queue.offer(clientQueue, message).pipe(Effect.asVoid);
319 }346 }
320 if (message.exit._tag === "Success") {347 if (message.exit._tag === "Success") {
...@@ -324,12 +351,20 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi...@@ -324,12 +351,20 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
324 if (failure && isProtocolError(failure.error)) {351 if (failure && isProtocolError(failure.error)) {
325 return completeExtPendingFailure(352 return completeExtPendingFailure(
326 message.requestId,353 message.requestId,
327 AcpError.AcpRequestError.fromProtocolError(failure.error),354 AcpError.AcpRequestError.fromProtocolError(failure.error, {
355 method: pendingRequest.method,
356 requestId: message.requestId,
357 cause: message.exit.cause,
358 }),
328 );359 );
329 }360 }
330 return completeExtPendingFailure(361 return completeExtPendingFailure(
331 message.requestId,362 message.requestId,
332 AcpError.AcpRequestError.internalError("Extension request failed"),363 AcpError.AcpRequestError.fromExtensionResponseFailure(
364 pendingRequest.method,
365 message.requestId,
366 message.exit.cause,
367 ),
333 );368 );
334 }),369 }),
335 );370 );
...@@ -344,16 +379,18 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi...@@ -344,16 +379,18 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
344 return handleExitEncoded(message);379 return handleExitEncoded(message);
345 case "Chunk":380 case "Chunk":
346 return Ref.get(extPending).pipe(381 return Ref.get(extPending).pipe(
347 Effect.flatMap((pending) =>382 Effect.flatMap((pending) => {
348 pending.has(message.requestId)383 const pendingRequest = pending.get(message.requestId);
384 return pendingRequest
349 ? completeExtPendingFailure(385 ? completeExtPendingFailure(
350 message.requestId,386 message.requestId,
351 AcpError.AcpRequestError.internalError(387 AcpError.AcpRequestError.unsupportedStreamingResponse(
352 "Streaming extension responses are not supported",388 pendingRequest.method,
389 message.requestId,
353 ),390 ),
354 )391 )
355 : Queue.offer(clientQueue, message).pipe(Effect.asVoid),392 : Queue.offer(clientQueue, message).pipe(Effect.asVoid);
356 ),393 }),
357 );394 );
358 case "Defect":395 case "Defect":
359 case "ClientProtocolError":396 case "ClientProtocolError":
...@@ -401,6 +438,7 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi...@@ -401,6 +438,7 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
401 payload: {438 payload: {
402 operation: error.operation,439 operation: error.operation,
403 ...(error.method === undefined ? {} : { method: error.method }),440 ...(error.method === undefined ? {} : { method: error.method }),
441 ...(error.requestId === undefined ? {} : { requestId: error.requestId }),
404 ...(error.issueCount === undefined ? {} : { issueCount: error.issueCount }),442 ...(error.issueCount === undefined ? {} : { issueCount: error.issueCount }),
405 ...(error.issueKinds === undefined ? {} : { issueKinds: error.issueKinds }),443 ...(error.issueKinds === undefined ? {} : { issueKinds: error.issueKinds }),
406 ...(error.maximumPathDepth === undefined444 ...(error.maximumPathDepth === undefined
...@@ -494,18 +532,16 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi...@@ -494,18 +532,16 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
494 (current) => [current, current + 1n] as const,532 (current) => [current, current + 1n] as const,
495 );533 );
496 const deferred = yield* Deferred.make<unknown, AcpError.AcpError>();534 const deferred = yield* Deferred.make<unknown, AcpError.AcpError>();
497 yield* Ref.update(extPending, (pending) => new Map(pending).set(String(requestId), deferred));535 yield* Ref.update(extPending, (pending) =>
536 new Map(pending).set(String(requestId), { deferred, method }),
537 );
498 yield* offerOutgoing({538 yield* offerOutgoing({
499 _tag: "Request",539 _tag: "Request",
500 id: String(requestId),540 id: String(requestId),
501 tag: method,541 tag: method,
502 payload,542 payload,
503 headers: [],543 headers: [],
504 }).pipe(544 }).pipe(Effect.tapError(() => removeExtPending(String(requestId))));
505 Effect.catch((error) =>
506 removeExtPending(String(requestId)).pipe(Effect.andThen(Effect.fail(error))),
507 ),
508 );
509 return yield* Deferred.await(deferred).pipe(545 return yield* Deferred.await(deferred).pipe(
510 Effect.onInterrupt(() => removeExtPending(String(requestId))),546 Effect.onInterrupt(() => removeExtPending(String(requestId))),
511 );547 );
packages/effect-codex-app-server/src/errors.ts+14-2
...@@ -5,6 +5,7 @@ export const CodexAppServerRequestOperation = Schema.Literals([...@@ -5,6 +5,7 @@ export const CodexAppServerRequestOperation = Schema.Literals([
5 "decode-payload",5 "decode-payload",
6 "encode-payload",6 "encode-payload",
7 "handle-request",7 "handle-request",
8 "receive-response",
8]);9]);
9export type CodexAppServerRequestOperation = typeof CodexAppServerRequestOperation.Type;10export type CodexAppServerRequestOperation = typeof CodexAppServerRequestOperation.Type;
1011
...@@ -83,6 +84,7 @@ const payloadKind = (payload: unknown): CodexAppServerPayloadKind => {...@@ -83,6 +84,7 @@ const payloadKind = (payload: unknown): CodexAppServerPayloadKind => {
8384
84export interface CodexAppServerRequestDiagnostics {85export interface CodexAppServerRequestDiagnostics {
85 readonly method?: string;86 readonly method?: string;
87 readonly requestId?: string;
86 readonly operation?: CodexAppServerRequestOperation;88 readonly operation?: CodexAppServerRequestOperation;
87 readonly cause?: unknown;89 readonly cause?: unknown;
88 readonly issueCount?: number;90 readonly issueCount?: number;
...@@ -154,6 +156,7 @@ export class CodexAppServerProtocolParseError extends Schema.TaggedErrorClass<Co...@@ -154,6 +156,7 @@ export class CodexAppServerProtocolParseError extends Schema.TaggedErrorClass<Co
154 {156 {
155 operation: CodexAppServerProtocolParseOperation,157 operation: CodexAppServerProtocolParseOperation,
156 method: Schema.optionalKey(Schema.String),158 method: Schema.optionalKey(Schema.String),
159 requestId: Schema.optionalKey(Schema.String),
157 detail: Schema.optionalKey(Schema.String),160 detail: Schema.optionalKey(Schema.String),
158 issueCount: Schema.optionalKey(Schema.Number),161 issueCount: Schema.optionalKey(Schema.Number),
159 issueKinds: Schema.optionalKey(Schema.Array(CodexAppServerSchemaIssueKind)),162 issueKinds: Schema.optionalKey(Schema.Array(CodexAppServerSchemaIssueKind)),
...@@ -169,7 +172,7 @@ export class CodexAppServerProtocolParseError extends Schema.TaggedErrorClass<Co...@@ -169,7 +172,7 @@ export class CodexAppServerProtocolParseError extends Schema.TaggedErrorClass<Co
169 static fromSchemaError(172 static fromSchemaError(
170 operation: CodexAppServerProtocolParseOperation,173 operation: CodexAppServerProtocolParseOperation,
171 cause: Schema.SchemaError,174 cause: Schema.SchemaError,
172 context: { readonly method?: string } = {},175 context: { readonly method?: string; readonly requestId?: string } = {},
173 ) {176 ) {
174 return new CodexAppServerProtocolParseError({177 return new CodexAppServerProtocolParseError({
175 operation,178 operation,
...@@ -235,6 +238,7 @@ export class CodexAppServerRequestError extends Schema.TaggedErrorClass<CodexApp...@@ -235,6 +238,7 @@ export class CodexAppServerRequestError extends Schema.TaggedErrorClass<CodexApp
235 errorMessage: Schema.String,238 errorMessage: Schema.String,
236 data: Schema.optional(Schema.Unknown),239 data: Schema.optional(Schema.Unknown),
237 method: Schema.optionalKey(Schema.String),240 method: Schema.optionalKey(Schema.String),
241 requestId: Schema.optionalKey(Schema.String),
238 operation: Schema.optionalKey(CodexAppServerRequestOperation),242 operation: Schema.optionalKey(CodexAppServerRequestOperation),
239 issueCount: Schema.optionalKey(Schema.Number),243 issueCount: Schema.optionalKey(Schema.Number),
240 issueKinds: Schema.optionalKey(Schema.Array(CodexAppServerSchemaIssueKind)),244 issueKinds: Schema.optionalKey(Schema.Array(CodexAppServerSchemaIssueKind)),
...@@ -247,11 +251,19 @@ export class CodexAppServerRequestError extends Schema.TaggedErrorClass<CodexApp...@@ -247,11 +251,19 @@ export class CodexAppServerRequestError extends Schema.TaggedErrorClass<CodexApp
247 return this.errorMessage;251 return this.errorMessage;
248 }252 }
249253
250 static fromProtocolError(error: CodexAppServerProtocolErrorShape) {254 static fromProtocolError(
255 error: CodexAppServerProtocolErrorShape,
256 method: string,
257 requestId: string,
258 ) {
251 return new CodexAppServerRequestError({259 return new CodexAppServerRequestError({
252 code: error.code,260 code: error.code,
253 errorMessage: error.message,261 errorMessage: error.message,
254 ...(error.data !== undefined ? { data: error.data } : {}),262 ...(error.data !== undefined ? { data: error.data } : {}),
263 method,
264 requestId,
265 operation: "receive-response",
266 cause: error,
255 });267 });
256 }268 }
257269
packages/effect-codex-app-server/src/protocol.test.ts+52-1
...@@ -212,10 +212,11 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => {...@@ -212,10 +212,11 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => {
212 const bigintError = yield* transport.notify("x/test", 1n).pipe(Effect.flip);212 const bigintError = yield* transport.notify("x/test", 1n).pipe(Effect.flip);
213 assert.instanceOf(bigintError, CodexError.CodexAppServerProtocolParseError);213 assert.instanceOf(bigintError, CodexError.CodexAppServerProtocolParseError);
214 assert.equal(bigintError.operation, "encode-wire-message");214 assert.equal(bigintError.operation, "encode-wire-message");
215 assert.equal(bigintError.method, "x/test");
215 assert.exists(bigintError.cause);216 assert.exists(bigintError.cause);
216 assert.equal(217 assert.equal(
217 bigintError.message,218 bigintError.message,
218 "Codex App Server protocol operation 'encode-wire-message' failed.",219 "Codex App Server protocol operation 'encode-wire-message' failed for method 'x/test'.",
219 );220 );
220221
221 const circular: Record<string, unknown> = {};222 const circular: Record<string, unknown> = {};
...@@ -223,7 +224,57 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => {...@@ -223,7 +224,57 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => {
223 const circularError = yield* transport.notify("x/test", circular).pipe(Effect.flip);224 const circularError = yield* transport.notify("x/test", circular).pipe(Effect.flip);
224 assert.instanceOf(circularError, CodexError.CodexAppServerProtocolParseError);225 assert.instanceOf(circularError, CodexError.CodexAppServerProtocolParseError);
225 assert.equal(circularError.operation, "encode-wire-message");226 assert.equal(circularError.operation, "encode-wire-message");
227 assert.equal(circularError.method, "x/test");
226 assert.exists(circularError.cause);228 assert.exists(circularError.cause);
229
230 const requestError = yield* transport.request("x/request", 1n).pipe(
231 Effect.match({
232 onFailure: (error) => error,
233 onSuccess: () => assert.fail("Expected request encoding to fail"),
234 }),
235 );
236 assert.instanceOf(requestError, CodexError.CodexAppServerProtocolParseError);
237 assert.deepInclude(requestError, {
238 operation: "encode-wire-message",
239 method: "x/request",
240 requestId: "1",
241 });
242 }),
243 );
244
245 it.effect("correlates response errors with the originating request", () =>
246 Effect.gen(function* () {
247 const { stdio, input, output } = yield* makeInMemoryStdio();
248 const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ stdio });
249
250 const response = yield* transport.request("thread/start", {}).pipe(Effect.forkScoped);
251 yield* Queue.take(output);
252 yield* Queue.offer(
253 input,
254 encodeJsonl({
255 id: 1,
256 error: {
257 code: -32602,
258 message: "Invalid params",
259 data: { field: "cwd" },
260 },
261 }),
262 );
263
264 const error = yield* Fiber.join(response).pipe(
265 Effect.match({
266 onFailure: (error) => error,
267 onSuccess: () => assert.fail("Expected Codex App Server request to fail"),
268 }),
269 );
270 assert.instanceOf(error, CodexError.CodexAppServerRequestError);
271 assert.deepInclude(error, {
272 code: -32602,
273 errorMessage: "Invalid params",
274 method: "thread/start",
275 requestId: "1",
276 operation: "receive-response",
277 });
227 }),278 }),
228 );279 );
229280
packages/effect-codex-app-server/src/protocol.ts+40-22
...@@ -67,6 +67,11 @@ export interface CodexAppServerPatchedProtocol {...@@ -67,6 +67,11 @@ export interface CodexAppServerPatchedProtocol {
67 ) => Effect.Effect<void, CodexError.CodexAppServerError>;67 ) => Effect.Effect<void, CodexError.CodexAppServerError>;
68}68}
6969
70interface CodexAppServerPendingRequest {
71 readonly deferred: Deferred.Deferred<unknown, CodexError.CodexAppServerError>;
72 readonly method: string;
73}
74
70function isObject(value: unknown): value is Record<string, unknown> {75function isObject(value: unknown): value is Record<string, unknown> {
71 return typeof value === "object" && value !== null;76 return typeof value === "object" && value !== null;
72}77}
...@@ -94,9 +99,21 @@ const encodeWireMessage = (...@@ -94,9 +99,21 @@ const encodeWireMessage = (
94): Effect.Effect<string, CodexError.CodexAppServerProtocolParseError> =>99): Effect.Effect<string, CodexError.CodexAppServerProtocolParseError> =>
95 encodeJsonString(message).pipe(100 encodeJsonString(message).pipe(
96 Effect.map((encoded) => `${encoded}\n`),101 Effect.map((encoded) => `${encoded}\n`),
97 Effect.mapError((cause) =>102 Effect.mapError((cause) => {
98 CodexError.CodexAppServerProtocolParseError.fromSchemaError("encode-wire-message", cause),103 const method = typeof message.method === "string" ? message.method : undefined;
99 ),104 const requestId =
105 typeof message.id === "string" || typeof message.id === "number"
106 ? String(message.id)
107 : undefined;
108 return CodexError.CodexAppServerProtocolParseError.fromSchemaError(
109 "encode-wire-message",
110 cause,
111 {
112 ...(method === undefined ? {} : { method }),
113 ...(requestId === undefined ? {} : { requestId }),
114 },
115 );
116 }),
100 );117 );
101118
102const decodeWireMessage = (119const decodeWireMessage = (
...@@ -138,9 +155,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa...@@ -138,9 +155,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
138 const outgoing = yield* Queue.unbounded<string, Cause.Done<void>>();155 const outgoing = yield* Queue.unbounded<string, Cause.Done<void>>();
139 const incomingNotifications = yield* Queue.unbounded<CodexAppServerIncomingNotification>();156 const incomingNotifications = yield* Queue.unbounded<CodexAppServerIncomingNotification>();
140 const incomingRequests = yield* Queue.unbounded<CodexAppServerIncomingRequest>();157 const incomingRequests = yield* Queue.unbounded<CodexAppServerIncomingRequest>();
141 const pending = yield* Ref.make(158 const pending = yield* Ref.make(new Map<string, CodexAppServerPendingRequest>());
142 new Map<string, Deferred.Deferred<unknown, CodexError.CodexAppServerError>>(),
143 );
144 const nextRequestId = yield* Ref.make(1);159 const nextRequestId = yield* Ref.make(1);
145 const remainder = yield* Ref.make("");160 const remainder = yield* Ref.make("");
146 const terminationHandled = yield* Ref.make(false);161 const terminationHandled = yield* Ref.make(false);
...@@ -161,7 +176,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa...@@ -161,7 +176,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
161 const failAllPending = (error: CodexError.CodexAppServerError) =>176 const failAllPending = (error: CodexError.CodexAppServerError) =>
162 Ref.get(pending).pipe(177 Ref.get(pending).pipe(
163 Effect.flatMap((current) =>178 Effect.flatMap((current) =>
164 Effect.forEach([...current.values()], (deferred) => Deferred.fail(deferred, error), {179 Effect.forEach([...current.values()], ({ deferred }) => Deferred.fail(deferred, error), {
165 discard: true,180 discard: true,
166 }),181 }),
167 ),182 ),
...@@ -214,18 +229,16 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa...@@ -214,18 +229,16 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
214229
215 const resolvePending = (230 const resolvePending = (
216 requestId: string,231 requestId: string,
217 handler: (232 handler: (pendingRequest: CodexAppServerPendingRequest) => Effect.Effect<void>,
218 deferred: Deferred.Deferred<unknown, CodexError.CodexAppServerError>,
219 ) => Effect.Effect<void>,
220 ) =>233 ) =>
221 Ref.modify(pending, (current) => {234 Ref.modify(pending, (current) => {
222 const deferred = current.get(requestId);235 const pendingRequest = current.get(requestId);
223 if (!deferred) {236 if (!pendingRequest) {
224 return [Effect.void, current] as const;237 return [Effect.void, current] as const;
225 }238 }
226 const next = new Map(current);239 const next = new Map(current);
227 next.delete(requestId);240 next.delete(requestId);
228 return [handler(deferred), next] as const;241 return [handler(pendingRequest), next] as const;
229 }).pipe(Effect.flatten);242 }).pipe(Effect.flatten);
230243
231 const respond = (requestId: string | number, result: unknown) =>244 const respond = (requestId: string | number, result: unknown) =>
...@@ -240,14 +253,20 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa...@@ -240,14 +253,20 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
240 const requestId = String(response.id);253 const requestId = String(response.id);
241 const protocolError = response.error;254 const protocolError = response.error;
242 if (protocolError !== undefined) {255 if (protocolError !== undefined) {
243 return resolvePending(requestId, (deferred) =>256 return resolvePending(requestId, ({ deferred, method }) =>
244 Deferred.fail(257 Deferred.fail(
245 deferred,258 deferred,
246 CodexError.CodexAppServerRequestError.fromProtocolError(protocolError),259 CodexError.CodexAppServerRequestError.fromProtocolError(
260 protocolError,
261 method,
262 requestId,
263 ),
247 ),264 ),
248 );265 );
249 }266 }
250 return resolvePending(requestId, (deferred) => Deferred.succeed(deferred, response.result));267 return resolvePending(requestId, ({ deferred }) =>
268 Deferred.succeed(deferred, response.result),
269 );
251 };270 };
252271
253 const handleRequest = (request: CodexAppServerIncomingRequest) =>272 const handleRequest = (request: CodexAppServerIncomingRequest) =>
...@@ -322,6 +341,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa...@@ -322,6 +341,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
322 payload: {341 payload: {
323 operation: error.operation,342 operation: error.operation,
324 ...(error.method === undefined ? {} : { method: error.method }),343 ...(error.method === undefined ? {} : { method: error.method }),
344 ...(error.requestId === undefined ? {} : { requestId: error.requestId }),
325 ...(error.issueCount === undefined ? {} : { issueCount: error.issueCount }),345 ...(error.issueCount === undefined ? {} : { issueCount: error.issueCount }),
326 ...(error.issueKinds === undefined ? {} : { issueKinds: error.issueKinds }),346 ...(error.issueKinds === undefined ? {} : { issueKinds: error.issueKinds }),
327 ...(error.maximumPathDepth === undefined347 ...(error.maximumPathDepth === undefined
...@@ -375,16 +395,14 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa...@@ -375,16 +395,14 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
375 (current) => [current, current + 1] as const,395 (current) => [current, current + 1] as const,
376 );396 );
377 const deferred = yield* Deferred.make<unknown, CodexError.CodexAppServerError>();397 const deferred = yield* Deferred.make<unknown, CodexError.CodexAppServerError>();
378 yield* Ref.update(pending, (current) => new Map(current).set(String(requestId), deferred));398 yield* Ref.update(pending, (current) =>
399 new Map(current).set(String(requestId), { deferred, method }),
400 );
379 yield* offerOutgoing({401 yield* offerOutgoing({
380 id: requestId,402 id: requestId,
381 method,403 method,
382 ...(payload !== undefined ? { params: payload } : {}),404 ...(payload !== undefined ? { params: payload } : {}),
383 }).pipe(405 }).pipe(Effect.tapError(() => removePending(String(requestId))));
384 Effect.catch((error) =>
385 removePending(String(requestId)).pipe(Effect.andThen(Effect.fail(error))),
386 ),
387 );
388 return yield* Deferred.await(deferred).pipe(406 return yield* Deferred.await(deferred).pipe(
389 Effect.onInterrupt(() => removePending(String(requestId))),407 Effect.onInterrupt(() => removePending(String(requestId))),
390 );408 );