import * as Path from "effect/Path";import * as AcpError from "./errors.ts";import * as Effect from "effect/Effect";import * as Deferred from "effect/Deferred";import * as Fiber from "effect/Fiber";import * as Queue from "effect/Queue";import * as Schema from "effect/Schema";import * as Stream from "effect/Stream";import * as Ref from "effect/Ref";import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import { it, assert } from "@effect/vitest";import * as NodeServices from "@effect/platform-node/NodeServices"; import * as AcpSchema from "./_generated/schema.gen.ts";import * as AcpProtocol from "./protocol.ts";import {  encodeJsonl,  jsonRpcNotification,  jsonRpcRequest,  jsonRpcResponse,} from "./_internal/shared.ts";import { makeInMemoryStdio, makeTerminationError, makeChildStdio } from "./_internal/stdio.ts"; const SessionCancelNotification = jsonRpcNotification(  "session/cancel",  AcpSchema.CancelNotification,);const SessionUpdateNotification = jsonRpcNotification(  "session/update",  AcpSchema.SessionNotification,);const ElicitationCompleteNotification = jsonRpcNotification(  "session/elicitation/complete",  AcpSchema.ElicitationCompleteNotification,);const RequestPermissionRequest = jsonRpcRequest(  "session/request_permission",  AcpSchema.RequestPermissionRequest,);const RequestPermissionResponse = jsonRpcResponse(AcpSchema.RequestPermissionResponse);const ExtRequest = jsonRpcRequest("x/test", Schema.Struct({ hello: Schema.String }));const ExtResponse = jsonRpcResponse(Schema.Struct({ ok: Schema.Boolean }));const decodeSessionCancelNotification = Schema.decodeEffect(  Schema.fromJsonString(SessionCancelNotification),);const decodeExtRequest = Schema.decodeEffect(Schema.fromJsonString(ExtRequest));const decodeExtResponse = Schema.decodeEffect(Schema.fromJsonString(ExtResponse));const decodeRequestPermissionResponse = Schema.decodeEffect(  Schema.fromJsonString(RequestPermissionResponse),);const encodeUnknownJsonString = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown));const encoder = new TextEncoder();const mockPeerPath = Effect.map(Effect.service(Path.Path), (path) =>  path.join(import.meta.dirname, "../test/fixtures/acp-mock-peer.ts"),);const mockPeerArgs = (path: string) => [path]; const makeHandle = (env?: Record<string, string>) =>  Effect.gen(function* () {    const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;    const path = yield* Path.Path;    const command = ChildProcess.make(process.execPath, mockPeerArgs(yield* mockPeerPath), {      cwd: path.join(import.meta.dirname, ".."),      ...(env ? { env: { ...process.env, ...env } } : {}),    });    return yield* spawner.spawn(command);  }); it.layer(NodeServices.layer)("effect-acp protocol", (it) => {  it.effect(    "emits exact JSON-RPC notifications and decodes inbound session/update and elicitation completion",    () =>      Effect.gen(function* () {        const { stdio, input, output } = yield* makeInMemoryStdio();        const transport = yield* AcpProtocol.makeAcpPatchedProtocol({          stdio,          serverRequestMethods: new Set(),        });         const notifications =          yield* Deferred.make<ReadonlyArray<AcpProtocol.AcpIncomingNotification>>();        yield* transport.incoming.pipe(          Stream.take(2),          Stream.runCollect,          Effect.flatMap((notificationChunk) => Deferred.succeed(notifications, notificationChunk)),          Effect.forkScoped,        );         yield* transport.notify("session/cancel", { sessionId: "session-1" });        const outbound = yield* Queue.take(output);        assert.deepEqual(yield* decodeSessionCancelNotification(outbound), {          jsonrpc: "2.0",          method: "session/cancel",          params: {            sessionId: "session-1",          },        });         yield* Queue.offer(          input,          yield* encodeJsonl(SessionUpdateNotification, {            jsonrpc: "2.0",            method: "session/update",            params: {              sessionId: "session-1",              update: {                sessionUpdate: "plan",                entries: [                  {                    content: "Inspect repository",                    priority: "high",                    status: "in_progress",                  },                ],              },            },          }),        );         yield* Queue.offer(          input,          yield* encodeJsonl(ElicitationCompleteNotification, {            jsonrpc: "2.0",            method: "session/elicitation/complete",            params: {              elicitationId: "elicitation-1",            },          }),        );         const [update, completion] = yield* Deferred.await(notifications);        assert.equal(update?._tag, "SessionUpdate");        assert.equal(completion?._tag, "ElicitationComplete");      }),  );   it.effect("keeps only recent raw notifications after their callbacks run", () =>    Effect.gen(function* () {      const { stdio, input } = yield* makeInMemoryStdio();      const handled = yield* Deferred.make<void>();      let handledCount = 0;      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({        stdio,        serverRequestMethods: new Set(),        onNotification: () =>          Effect.sync(() => ++handledCount).pipe(            Effect.flatMap((count) =>              count === 64 ? Deferred.succeed(handled, undefined).pipe(Effect.asVoid) : Effect.void,            ),          ),      });       const messages = Array.from({ length: 64 }, (_, index) =>        encodeUnknownJsonString({          jsonrpc: "2.0",          method: "x/performance",          params: { index },        }),      );      yield* Queue.offer(input, encoder.encode(`${messages.join("\n")}\n`));      yield* Deferred.await(handled);       const retained = yield* transport.incoming.pipe(Stream.take(32), Stream.runCollect);       assert.equal(handledCount, 64);      assert.equal(retained.length, 32);      assert.deepEqual(retained[0]?.params, { index: 32 });      assert.deepEqual(retained[31]?.params, { index: 63 });    }),  );   it.effect("keeps invalid core notification values only in the schema cause", () =>    Effect.gen(function* () {      const secret = "acp-core-notification-secret-sentinel";      const { stdio, input } = yield* makeInMemoryStdio();      const termination = yield* Deferred.make<AcpError.AcpError>();      let transformCalls = 0;      yield* AcpProtocol.makeAcpPatchedProtocol({        stdio,        serverRequestMethods: new Set(),        transformSessionUpdate: (notification) => {          transformCalls += 1;          return notification;        },        onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),      });       yield* Queue.offer(        input,        encoder.encode(          `${encodeUnknownJsonString({            jsonrpc: "2.0",            method: "session/update",            params: {              sessionId: { secret },              update: {                sessionUpdate: "plan",                entries: [],              },            },          })}\n`,        ),      );       const error = yield* Deferred.await(termination);      assert.equal(transformCalls, 0);      assert.instanceOf(error, AcpError.AcpProtocolParseError);      const parseError = error as AcpError.AcpProtocolParseError;      const { cause, ...directDiagnostics } = parseError;      assert.equal(parseError.operation, "decode-notification-payload");      assert.equal(parseError.method, "session/update");      assert.isAbove(parseError.issueCount ?? 0, 0);      assert.include(parseError.issueKinds ?? [], "Pointer");      assert.isAbove(parseError.maximumPathDepth ?? 0, 0);      assert.isTrue(Schema.isSchemaError(cause));      assert.notInclude(parseError.message, secret);      assert.notInclude(encodeUnknownJsonString(directDiagnostics), secret);    }),  );   it.effect("logs outgoing notifications when logOutgoing is enabled", () =>    Effect.gen(function* () {      const { stdio } = yield* makeInMemoryStdio();      const events: Array<AcpProtocol.AcpProtocolLogEvent> = [];      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({        stdio,        serverRequestMethods: new Set(),        logOutgoing: true,        logger: (event) =>          Effect.sync(() => {            events.push(event);          }),      });       yield* transport.notify("session/cancel", { sessionId: "session-1" });       // A notification must not carry `id` or `headers`. Grok CLI drops frames that do.      assert.deepEqual(events, [        {          direction: "outgoing",          stage: "decoded",          payload: {            _tag: "Notification",            tag: "session/cancel",            payload: {              sessionId: "session-1",            },          },        },        {          direction: "outgoing",          stage: "raw",          payload:            '{"jsonrpc":"2.0","method":"session/cancel","params":{"sessionId":"session-1"}}\n',        },      ]);    }),  );   it.effect("logs decode failures without copying the cause or wire payload", () =>    Effect.gen(function* () {      const secret = "acp-wire-secret-sentinel";      const { stdio, input } = yield* makeInMemoryStdio();      const events: Array<AcpProtocol.AcpProtocolLogEvent> = [];      const termination = yield* Deferred.make<AcpError.AcpError>();      yield* AcpProtocol.makeAcpPatchedProtocol({        stdio,        serverRequestMethods: new Set(),        logIncoming: true,        logger: (event) =>          Effect.sync(() => {            events.push(event);          }),        onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),      });       yield* Queue.offer(input, encoder.encode(`{"secret":"${secret}"\n`));      yield* Deferred.await(termination);       const event = events.find(({ stage }) => stage === "decode_failed");      assert.deepEqual(event, {        direction: "incoming",        stage: "decode_failed",        payload: {          operation: "decode-wire-message",        },      });      assert.notInclude(encodeUnknownJsonString(event), secret);    }),  );   it.effect("fails notification encoding through the declared ACP error channel", () =>    Effect.gen(function* () {      const { stdio } = yield* makeInMemoryStdio();      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({        stdio,        serverRequestMethods: new Set(),      });       // Notifications encode through Schema, so the cause is the schema failure rather      // than the raw TypeError JSON.stringify throws. The ACP error shape is what callers see.      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.isDefined(bigintError.cause);      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.isDefined(circularError.cause);       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,      });    }),  );   it.effect("supports generic extension requests over the patched transport", () =>    Effect.gen(function* () {      const { stdio, input, output } = yield* makeInMemoryStdio();      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({        stdio,        serverRequestMethods: new Set(),      });       const response = yield* transport        .request("x/test", { hello: "world" })        .pipe(Effect.forkScoped);      const outbound = yield* Queue.take(output);      assert.deepEqual(yield* decodeExtRequest(outbound), {        jsonrpc: "2.0",        id: 1,        method: "x/test",        params: {          hello: "world",        },        headers: [],      });       yield* Queue.offer(        input,        yield* encodeJsonl(ExtResponse, {          jsonrpc: "2.0",          id: 1,          result: {            ok: true,          },        }),      );       const resolved = yield* Fiber.join(response);      assert.deepEqual(resolved, { ok: true });    }),  );   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 numeric ids for inbound extension requests", () =>    Effect.gen(function* () {      const { stdio, input, output } = yield* makeInMemoryStdio();      yield* AcpProtocol.makeAcpPatchedProtocol({        stdio,        serverRequestMethods: new Set(),        onExtRequest: () => Effect.succeed({ ok: true }),      });       yield* Queue.offer(        input,        yield* encodeJsonl(ExtRequest, {          jsonrpc: "2.0",          id: 7,          method: "x/test",          params: {            hello: "world",          },          headers: [],        }),      );       const outbound = yield* Queue.take(output);      assert.deepEqual(yield* decodeExtResponse(outbound), {        jsonrpc: "2.0",        id: 7,        result: {          ok: true,        },      });    }),  );   it.effect("preserves zero-valued ids for inbound core client requests", () =>    Effect.gen(function* () {      const { stdio, input, output } = yield* makeInMemoryStdio();      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({        stdio,        serverRequestMethods: new Set(["session/request_permission"]),      });      const inboundRequest = yield* Deferred.make<unknown>();       yield* transport.serverProtocol        .run((_clientId, message) => Deferred.succeed(inboundRequest, message).pipe(Effect.asVoid))        .pipe(Effect.forkScoped);       yield* Queue.offer(        input,        yield* encodeJsonl(RequestPermissionRequest, {          jsonrpc: "2.0",          id: 0,          method: "session/request_permission",          params: {            sessionId: "session-1",            toolCall: {              toolCallId: "tool-1",              title: "Allow mock action",            },            options: [{ optionId: "allow", name: "Allow", kind: "allow_once" }],          },          headers: [],        }),      );       const message = yield* Deferred.await(inboundRequest);      assert.deepEqual(message, {        _tag: "Request",        id: 0,        tag: "session/request_permission",        payload: {          sessionId: "session-1",          toolCall: {            toolCallId: "tool-1",

Lines 1–500 of 770. Load more