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 mockPeerPath = Effect.map(Effect.service(Path.Path), (path) =>  path.join(import.meta.dirname, "../test/fixtures/acp-mock-peer.ts"),); const makeHandle = (env?: Record<string, string>) =>  Effect.gen(function* () {    const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;    const path = yield* Path.Path;    const command = ChildProcess.make("bun", ["run", yield* mockPeerPath], {      cwd: path.join(import.meta.dirname, ".."),      shell: process.platform === "win32",      ...(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* Schema.decodeEffect(Schema.fromJsonString(SessionCancelNotification))(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("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" });       assert.deepEqual(events, [        {          direction: "outgoing",          stage: "decoded",          payload: {            _tag: "Request",            id: "",            tag: "session/cancel",            payload: {              sessionId: "session-1",            },            headers: [],          },        },        {          direction: "outgoing",          stage: "raw",          payload:            '{"jsonrpc":"2.0","method":"session/cancel","params":{"sessionId":"session-1"},"id":"","headers":[]}\n',        },      ]);    }),  );   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(),      });       const bigintError = yield* transport.notify("x/test", 1n).pipe(Effect.flip);      assert.instanceOf(bigintError, AcpError.AcpProtocolParseError);      assert.equal(bigintError.detail, "Failed to encode ACP message");       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.detail, "Failed to encode ACP message");    }),  );   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* Schema.decodeEffect(Schema.fromJsonString(ExtRequest))(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("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",            title: "Allow mock action",          },          options: [{ optionId: "allow", name: "Allow", kind: "allow_once" }],        },        headers: [],      });       yield* transport.serverProtocol.send(0, {        _tag: "Exit",        requestId: "0",        exit: {          _tag: "Success",          value: {            outcome: {              outcome: "selected",              optionId: "allow",            },          },        },      });       const outbound = yield* Queue.take(output);      assert.deepEqual(        yield* Schema.decodeEffect(Schema.fromJsonString(RequestPermissionResponse))(outbound),        {          jsonrpc: "2.0",          id: 0,          result: {            outcome: {              outcome: "selected",              optionId: "allow",            },          },        },      );    }),  );   it.effect("cleans up interrupted extension requests before a late response arrives", () =>    Effect.gen(function* () {      const { stdio, input, output } = yield* makeInMemoryStdio();      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({        stdio,        serverRequestMethods: new Set(),      });      const lateResponse = yield* Deferred.make<unknown>();       yield* transport.clientProtocol        .run(0, (message) => Deferred.succeed(lateResponse, message).pipe(Effect.asVoid))        .pipe(Effect.forkScoped);       const response = yield* transport        .request("x/test", { hello: "world" })        .pipe(Effect.forkScoped);      const outbound = yield* Queue.take(output);      assert.deepEqual(yield* Schema.decodeEffect(Schema.fromJsonString(ExtRequest))(outbound), {        jsonrpc: "2.0",        id: 1,        method: "x/test",        params: {          hello: "world",        },        headers: [],      });       yield* Fiber.interrupt(response);      yield* Queue.offer(        input,        yield* encodeJsonl(ExtResponse, {          jsonrpc: "2.0",          id: 1,          result: {            ok: true,          },        }),      );       const message = yield* Deferred.await(lateResponse);      assert.deepEqual(message, {        _tag: "Exit",        requestId: "1",        exit: {          _tag: "Success",          value: {            ok: true,          },        },      });    }),  );   it.effect("propagates the real child exit code when the input stream ends", () =>    Effect.gen(function* () {      const handle = yield* makeHandle({ ACP_MOCK_EXIT_IMMEDIATELY_CODE: "7" });      const firstMessage = yield* Deferred.make<unknown>();      const termination = yield* Deferred.make<AcpError.AcpError>();      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({        stdio: makeChildStdio(handle),        terminationError: makeTerminationError(handle),        serverRequestMethods: new Set(),        onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),      });       yield* transport.clientProtocol        .run(0, (message) => Deferred.succeed(firstMessage, message).pipe(Effect.asVoid))        .pipe(Effect.forkScoped);       const message = yield* Deferred.await(firstMessage);      const exitError = yield* Deferred.await(termination);      assert.instanceOf(exitError, AcpError.AcpProcessExitedError);      assert.equal((exitError as AcpError.AcpProcessExitedError).code, 7);      assert.equal((message as { readonly _tag?: string })._tag, "ClientProtocolError");      const defect = (message as { readonly error: { readonly reason: unknown } }).error.reason as {        readonly _tag: string;        readonly cause: unknown;      };      assert.equal(defect._tag, "RpcClientDefect");      assert.instanceOf(defect.cause, AcpError.AcpProcessExitedError);      assert.equal((defect.cause as AcpError.AcpProcessExitedError).code, 7);    }),  );   it.effect("does not emit a second process-exit error after a decode failure", () =>    Effect.gen(function* () {      const handle = yield* makeHandle({        ACP_MOCK_MALFORMED_OUTPUT: "1",        ACP_MOCK_MALFORMED_OUTPUT_EXIT_CODE: "23",      });      const terminationCalls = yield* Ref.make(0);      const firstMessage = yield* Deferred.make<unknown>();      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({        stdio: makeChildStdio(handle),        terminationError: makeTerminationError(handle),        serverRequestMethods: new Set(),        onTermination: () => Ref.update(terminationCalls, (count) => count + 1),      });       yield* transport.clientProtocol        .run(0, (message) => Deferred.succeed(firstMessage, message).pipe(Effect.asVoid))        .pipe(Effect.forkScoped);       const message = yield* Deferred.await(firstMessage);      assert.equal(yield* Ref.get(terminationCalls), 1);      assert.equal((message as { readonly _tag?: string })._tag, "ClientProtocolError");      const defect = (message as { readonly error: { readonly reason: unknown } }).error.reason as {        readonly _tag: string;        readonly cause: unknown;      };      assert.equal(defect._tag, "RpcClientDefect");      assert.instanceOf(defect.cause, AcpError.AcpProtocolParseError);    }),  );   it.effect("fails pending extension requests with the propagated exit code", () =>    Effect.gen(function* () {      const { stdio, input, output } = yield* makeInMemoryStdio();      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({        stdio,        terminationError: Effect.succeed(new AcpError.AcpProcessExitedError({ code: 0 })),        serverRequestMethods: new Set(),      });       const response = yield* transport        .request("x/test", { hello: "world" })        .pipe(Effect.forkScoped);      yield* Queue.take(output);      yield* Queue.end(input);       const error = yield* Fiber.join(response).pipe(        Effect.match({          onFailure: (error) => error,          onSuccess: () => assert.fail("Expected request to fail after process exit"),        }),      );      assert.instanceOf(error, AcpError.AcpProcessExitedError);      assert.equal(error.code, 0);    }),  );});