[codex] Correlate protocol request failures

packages/effect-acp/src/_internal/shared.ts

@@ -12,7 +12,7 @@ export const callRpc = <A>(
 ): Effect.Effect<A, AcpError.AcpError> =>
   effect.pipe(
     Effect.catchIf(isError, (error) =>
-      Effect.fail(AcpError.AcpRequestError.fromProtocolError(error)),
+      Effect.fail(AcpError.AcpRequestError.fromProtocolError(error, { method })),
     ),
     Effect.catchTags({
       RpcClientError: (cause) =>

packages/effect-acp/src/errors.test.ts

@@ -51,6 +51,8 @@ describe("effect-acp errors", () => {
         code: -32602,
         errorMessage: "Invalid params",
         data: { field: "sessionId" },
+        method: "session/load",
+        operation: "receive-response",
       });
     });
   });

packages/effect-acp/src/errors.ts

@@ -5,8 +5,11 @@ import * as AcpSchema from "./_generated/schema.gen.ts";
 
 export const AcpRequestOperation = Schema.Literals([
   "decode-extension-request-payload",
+  "encode-extension-response",
   "handle-request",
   "handle-extension-request",
+  "receive-response",
+  "receive-streaming-response",
 ]);
 export type AcpRequestOperation = typeof AcpRequestOperation.Type;
 
@@ -65,6 +68,7 @@ const schemaIssueDiagnostics = (root: SchemaIssue.Issue): AcpSchemaIssueDiagnost
 
 export interface AcpRequestDiagnostics {
   readonly method?: string;
+  readonly requestId?: string;
   readonly operation?: AcpRequestOperation;
   readonly cause?: unknown;
   readonly issueCount?: number;
@@ -109,6 +113,7 @@ export class AcpProtocolParseError extends Schema.TaggedErrorClass<AcpProtocolPa
   {
     operation: AcpProtocolParseOperation,
     method: Schema.optionalKey(Schema.String),
+    requestId: Schema.optionalKey(Schema.String),
     issueCount: Schema.optionalKey(Schema.Number),
     issueKinds: Schema.optionalKey(Schema.Array(AcpSchemaIssueKind)),
     maximumPathDepth: Schema.optionalKey(Schema.Number),
@@ -132,6 +137,19 @@ export class AcpProtocolParseError extends Schema.TaggedErrorClass<AcpProtocolPa
       cause,
     });
   }
+
+  static fromEncodingError(
+    method: string | undefined,
+    requestId: string | undefined,
+    cause: unknown,
+  ) {
+    return new AcpProtocolParseError({
+      operation: "encode-message",
+      ...(method === undefined ? {} : { method }),
+      ...(requestId === undefined ? {} : { requestId }),
+      cause,
+    });
+  }
 }
 
 export class AcpTransportError extends Schema.TaggedErrorClass<AcpTransportError>()(
@@ -167,6 +185,7 @@ export class AcpRequestError extends Schema.TaggedErrorClass<AcpRequestError>()(
   errorMessage: Schema.String,
   data: Schema.optional(Schema.Unknown),
   method: Schema.optionalKey(Schema.String),
+  requestId: Schema.optionalKey(Schema.String),
   operation: Schema.optionalKey(AcpRequestOperation),
   issueCount: Schema.optionalKey(Schema.Number),
   issueKinds: Schema.optionalKey(Schema.Array(AcpSchemaIssueKind)),
@@ -177,14 +196,59 @@ export class AcpRequestError extends Schema.TaggedErrorClass<AcpRequestError>()(
     return this.errorMessage;
   }
 
-  static fromProtocolError(error: AcpSchema.Error) {
+  static fromProtocolError(
+    error: AcpSchema.Error,
+    context: {
+      readonly method: string;
+      readonly requestId?: string;
+      readonly cause?: unknown;
+    },
+  ) {
     return new AcpRequestError({
       code: error.code,
       errorMessage: error.message,
       ...(error.data !== undefined ? { data: error.data } : {}),
+      method: context.method,
+      ...(context.requestId === undefined ? {} : { requestId: context.requestId }),
+      operation: "receive-response",
+      cause: context.cause ?? error,
     });
   }
 
+  static fromExtensionResponseFailure(method: string, requestId: string, cause: unknown) {
+    return AcpRequestError.internalError("Extension request failed", undefined, {
+      method,
+      requestId,
+      operation: "receive-response",
+      cause,
+    });
+  }
+
+  static fromExtensionResponseEncodingError(
+    method: string,
+    requestId: string,
+    cause: AcpProtocolParseError,
+  ) {
+    return AcpRequestError.internalError("Internal error", undefined, {
+      method,
+      requestId,
+      operation: "encode-extension-response",
+      cause,
+    });
+  }
+
+  static unsupportedStreamingResponse(method: string, requestId: string) {
+    return AcpRequestError.internalError(
+      "Streaming extension responses are not supported",
+      undefined,
+      {
+        method,
+        requestId,
+        operation: "receive-streaming-response",
+      },
+    );
+  }
+
   static fromCoreHandlerError(error: AcpError, method: string) {
     if (error._tag === "AcpRequestError") {
       return error;

packages/effect-acp/src/protocol.test.ts

@@ -260,15 +260,33 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
       const bigintError = yield* transport.notify("x/test", 1n).pipe(Effect.flip);
       assert.instanceOf(bigintError, AcpError.AcpProtocolParseError);
       assert.equal(bigintError.operation, "encode-message");
+      assert.equal(bigintError.method, "x/test");
       assert.instanceOf(bigintError.cause, TypeError);
-      assert.equal(bigintError.message, "ACP protocol operation 'encode-message' failed.");
+      assert.equal(
+        bigintError.message,
+        "ACP protocol operation 'encode-message' failed for method 'x/test'.",
+      );
 
       const circular: Record<string, unknown> = {};
       circular.self = circular;
       const circularError = yield* transport.notify("x/test", circular).pipe(Effect.flip);
       assert.instanceOf(circularError, AcpError.AcpProtocolParseError);
       assert.equal(circularError.operation, "encode-message");
+      assert.equal(circularError.method, "x/test");
       assert.instanceOf(circularError.cause, TypeError);
+
+      const requestError = yield* transport.request("x/request", 1n).pipe(
+        Effect.match({
+          onFailure: (error) => error,
+          onSuccess: () => assert.fail("Expected request encoding to fail"),
+        }),
+      );
+      assert.instanceOf(requestError, AcpError.AcpProtocolParseError);
+      assert.deepInclude(requestError, {
+        operation: "encode-message",
+        method: "x/request",
+        requestId: "1",
+      });
     }),
   );
 
@@ -310,6 +328,60 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
     }),
   );
 
+  it.effect("correlates extension response errors with the originating request", () =>
+    Effect.gen(function* () {
+      const { stdio, input, output } = yield* makeInMemoryStdio();
+      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
+        stdio,
+        serverRequestMethods: new Set(),
+      });
+
+      const response = yield* transport
+        .request("x/private", { hello: "world" })
+        .pipe(Effect.forkScoped);
+      yield* Queue.take(output);
+      yield* Queue.offer(
+        input,
+        encoder.encode(
+          `${encodeUnknownJsonString({
+            jsonrpc: "2.0",
+            id: 1,
+            error: {
+              _tag: "Cause",
+              code: -32602,
+              message: "Invalid params",
+              data: [
+                {
+                  _tag: "Fail",
+                  error: {
+                    code: -32602,
+                    message: "Invalid params",
+                    data: { field: "hello" },
+                  },
+                },
+              ],
+            },
+          })}\n`,
+        ),
+      );
+
+      const error = yield* Fiber.join(response).pipe(
+        Effect.match({
+          onFailure: (error) => error,
+          onSuccess: () => assert.fail("Expected extension request to fail"),
+        }),
+      );
+      assert.instanceOf(error, AcpError.AcpRequestError);
+      assert.deepInclude(error, {
+        code: -32602,
+        errorMessage: "Invalid params",
+        method: "x/private",
+        requestId: "1",
+        operation: "receive-response",
+      });
+    }),
+  );
+
   it.effect("preserves zero-valued ids for inbound core client requests", () =>
     Effect.gen(function* () {
       const { stdio, input, output } = yield* makeInMemoryStdio();