import * as Cause from "effect/Cause";
import * as Effect from "effect/Effect";
import * as Deferred from "effect/Deferred";
import * as Exit from "effect/Exit";
import * as Queue from "effect/Queue";
import * as Ref from "effect/Ref";
import type * as PlatformError from "effect/PlatformError";
import * as Schema from "effect/Schema";
import * as Scope from "effect/Scope";
import * as Stream from "effect/Stream";
import * as Stdio from "effect/Stdio";
import * as RpcClient from "effect/unstable/rpc/RpcClient";
import * as RpcClientError from "effect/unstable/rpc/RpcClientError";
import * as RpcMessage from "effect/unstable/rpc/RpcMessage";
import * as RpcSerialization from "effect/unstable/rpc/RpcSerialization";
import * as RpcServer from "effect/unstable/rpc/RpcServer";

import * as AcpSchema from "./_generated/schema.gen.ts";
import { CLIENT_METHODS } from "./_generated/meta.gen.ts";
import * as AcpError from "./errors.ts";
const isAcpError = Schema.is(AcpError.AcpError);

export interface AcpProtocolLogEvent {
  readonly direction: "incoming" | "outgoing";
  readonly stage: "raw" | "decoded" | "decode_failed";
  readonly payload: unknown;
}

export type AcpIncomingNotification =
  | {
      readonly _tag: "SessionUpdate";
      readonly method: typeof CLIENT_METHODS.session_update;
      readonly params: AcpSchema.SessionNotification;
    }
  | {
      readonly _tag: "ElicitationComplete";
      readonly method: typeof CLIENT_METHODS.session_elicitation_complete;
      readonly params: AcpSchema.ElicitationCompleteNotification;
    }
  | {
      readonly _tag: "ExtNotification";
      readonly method: string;
      readonly params: unknown;
    };

/** Standard I/O whose input can report provider-specific ACP failures. */
export interface AcpStdio extends Omit<Stdio.Stdio, "stdin"> {
  readonly stdin: Stream.Stream<Uint8Array, PlatformError.PlatformError | AcpError.AcpError>;
}

export interface AcpPatchedProtocolOptions {
  readonly stdio: AcpStdio;
  readonly terminationError?: Effect.Effect<AcpError.AcpError>;
  readonly serverRequestMethods: ReadonlySet<string>;
  readonly logIncoming?: boolean;
  readonly logOutgoing?: boolean;
  readonly logger?: (event: AcpProtocolLogEvent) => Effect.Effect<void, never>;
  readonly onIncomingRequest?: (
    requestId: string,
    method: string,
    payload: unknown,
  ) => Effect.Effect<void, never>;
  readonly transformSessionUpdate?: (
    notification: AcpSchema.SessionNotification,
  ) => AcpSchema.SessionNotification;
  readonly onNotification?: (
    notification: AcpIncomingNotification,
  ) => Effect.Effect<void, AcpError.AcpError, never>;
  readonly onExtRequest?: (
    method: string,
    params: unknown,
  ) => Effect.Effect<unknown, AcpError.AcpError, never>;
  readonly onTermination?: (error: AcpError.AcpError) => Effect.Effect<void, never, never>;
  readonly onOutgoingResponseFailure?: (
    requestId: string,
    error: AcpError.AcpError,
  ) => Effect.Effect<void, never>;
  readonly onOutgoingResponse?: (requestId: string) => Effect.Effect<void, never>;
  readonly testHooks?: {
    readonly onOutgoingWriteAdmitted?: () => Effect.Effect<void>;
    /** Invoked from the stdout writer exit path (after intentional/unexpected handling). */
    readonly onOutgoingWriterExit?: (input: {
      readonly intentionalEnd: boolean;
    }) => Effect.Effect<void>;
  };
}

export interface AcpPatchedProtocol {
  readonly clientProtocol: RpcClient.Protocol["Service"];
  readonly serverProtocol: RpcServer.Protocol["Service"];
  readonly incoming: Stream.Stream<AcpIncomingNotification>;
  readonly request: (method: string, payload: unknown) => Effect.Effect<unknown, AcpError.AcpError>;
  readonly notify: (method: string, payload: unknown) => Effect.Effect<void, AcpError.AcpError>;
}

interface AcpPendingRequest {
  readonly deferred: Deferred.Deferred<unknown, AcpError.AcpError>;
  readonly method: string;
}

interface AcpOutgoingWrite {
  readonly acknowledgement?: Deferred.Deferred<void, AcpError.AcpError>;
  readonly payload: string | Uint8Array;
}

interface AcpOutgoingWriterState {
  readonly outstandingAcknowledgements: ReadonlySet<Deferred.Deferred<void, AcpError.AcpError>>;
  readonly intentionalEnd: boolean;
  readonly terminalError?: AcpError.AcpError;
}

const decodeSessionUpdate = Schema.decodeUnknownEffect(AcpSchema.SessionNotification);
const decodeElicitationComplete = Schema.decodeUnknownEffect(
  AcpSchema.ElicitationCompleteNotification,
);
const parserFactory = RpcSerialization.ndJsonRpc();
const MAX_BUFFERED_RAW_NOTIFICATIONS = 32;
// Outbound JSON-RPC notification: no `id`, so peers never treat it as a request.
const encodeJsonRpcNotification = Schema.encodeUnknownExit(
  Schema.fromJsonString(
    Schema.Struct({
      jsonrpc: Schema.Literal("2.0"),
      method: Schema.String,
      params: Schema.Unknown,
    }),
  ),
);

const isEffectRpcRequestId = (requestId: AcpError.AcpRequestId): boolean =>
  typeof requestId === "number" && Number.isSafeInteger(requestId);

export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(function* (
  options: AcpPatchedProtocolOptions,
): Effect.fn.Return<AcpPatchedProtocol, never, Scope.Scope> {
  const parser = parserFactory.makeUnsafe();
  const serverQueue = yield* Queue.unbounded<RpcMessage.FromClientEncoded>();
  const clientQueue = yield* Queue.unbounded<RpcMessage.FromServerEncoded>();
  const notificationQueue = yield* Queue.sliding<AcpIncomingNotification>(
    MAX_BUFFERED_RAW_NOTIFICATIONS,
  );
  const disconnects = yield* Queue.unbounded<number>();
  const outgoing = yield* Queue.unbounded<AcpOutgoingWrite, AcpError.AcpError | Cause.Done<void>>();
  const outgoingWriterState = yield* Ref.make<AcpOutgoingWriterState>({
    intentionalEnd: false,
    outstandingAcknowledgements: new Set(),
  });
  const nextRequestId = yield* Ref.make(1);
  const terminationHandled = yield* Ref.make(false);
  const terminationFailure = yield* Deferred.make<never, AcpError.AcpError>();
  const extPending = yield* Ref.make(new Map<string, AcpPendingRequest>());

  const ensureActive = Ref.get(terminationHandled).pipe(
    Effect.flatMap((terminated) => (terminated ? Deferred.await(terminationFailure) : Effect.void)),
  );

  const logProtocol = (event: AcpProtocolLogEvent) => {
    if (event.direction === "incoming" && !options.logIncoming) {
      return Effect.void;
    }
    if (event.direction === "outgoing" && !options.logOutgoing) {
      return Effect.void;
    }
    return (
      options.logger?.(event) ??
      Effect.logDebug("ACP protocol event").pipe(Effect.annotateLogs({ event }))
    );
  };

  const offerOutgoing = Effect.fn("offerOutgoing")(function* (
    message: RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded,
  ) {
    // RpcClient emits `@effect/rpc/Interrupt` when a pending request's fiber is interrupted.
    // ACP has no such method; agents log it as an error and cannot act on it, so drop it.
    if (message._tag === "Interrupt") {
      return;
    }
    yield* ensureActive;
    yield* logProtocol({
      direction: "outgoing",
      stage: "decoded",
      payload: message,
    });

    const method = message._tag === "Request" ? message.tag : undefined;
    const encodedRequestId =
      message._tag === "Request"
        ? message.id
        : "requestId" in message
          ? message.requestId
          : undefined;
    const requestId = encodedRequestId === "" ? undefined : encodedRequestId;
    const encoded = yield* Effect.try({
      try: () => parser.encode(message),
      catch: (cause) => AcpError.AcpProtocolParseError.fromEncodingError(method, requestId, cause),
    });

    if (encoded) {
      yield* logProtocol({
        direction: "outgoing",
        stage: "raw",
        payload: typeof encoded === "string" ? encoded : new TextDecoder().decode(encoded),
      });

      yield* ensureActive;
      const acknowledgement =
        message._tag === "Exit" ? yield* Deferred.make<void, AcpError.AcpError>() : undefined;
      const admissionError = yield* Ref.modify(outgoingWriterState, (state) => {
        if (state.terminalError !== undefined) {
          return [state.terminalError, state] as const;
        }
        if (acknowledgement === undefined) {
          return [undefined, state] as const;
        }
        return [
          undefined,
          {
            ...state,
            outstandingAcknowledgements: new Set(state.outstandingAcknowledgements).add(
              acknowledgement,
            ),
          },
        ] as const;
      });
      if (admissionError !== undefined) {
        return yield* admissionError;
      }
      if (options.testHooks?.onOutgoingWriteAdmitted !== undefined) {
        yield* options.testHooks.onOutgoingWriteAdmitted();
      }
      yield* Queue.offer(outgoing, {
        payload: encoded,
        ...(acknowledgement === undefined ? {} : { acknowledgement }),
      }).pipe(
        Effect.mapError(
          (cause) =>
            new AcpError.AcpTransportError({
              detail: "Failed to queue an outgoing ACP message",
              cause,
            }),
        ),
        Effect.filterOrFail(
          (accepted) => accepted,
          () =>
            new AcpError.AcpTransportError({
              detail: "The ACP output queue was closed before accepting a message",
              cause: "Output queue closed",
            }),
        ),
        Effect.tapError((error) =>
          acknowledgement === undefined
            ? Effect.void
            : Ref.update(outgoingWriterState, (state) => {
                const updated = new Set(state.outstandingAcknowledgements);
                updated.delete(acknowledgement);
                return { ...state, outstandingAcknowledgements: updated };
              }).pipe(Effect.andThen(Deferred.fail(acknowledgement, error)), Effect.asVoid),
        ),
      );
      if (acknowledgement !== undefined) {
        yield* Deferred.await(acknowledgement);
      }
      if (message._tag === "Exit" && options.onOutgoingResponse !== undefined) {
        yield* options.onOutgoingResponse(String(message.requestId));
      }
    }
  });

  const resolveExtPending = (
    requestId: AcpError.AcpRequestId,
    onFound: (pendingRequest: AcpPendingRequest) => Effect.Effect<void>,
  ) =>
    Ref.modify(extPending, (pending) => {
      const pendingKey = String(requestId);
      const pendingRequest = pending.get(pendingKey);
      if (!pendingRequest) {
        return [Effect.void, pending] as const;
      }
      const next = new Map(pending);
      next.delete(pendingKey);
      return [onFound(pendingRequest), next] as const;
    }).pipe(Effect.flatten);

  const removeExtPending = (requestId: AcpError.AcpRequestId) =>
    Ref.update(extPending, (pending) => {
      const pendingKey = String(requestId);
      if (!pending.has(pendingKey)) {
        return pending;
      }
      const next = new Map(pending);
      next.delete(pendingKey);
      return next;
    });

  const completeExtPendingFailure = (requestId: AcpError.AcpRequestId, error: AcpError.AcpError) =>
    resolveExtPending(requestId, ({ deferred }) => Deferred.fail(deferred, error));

  const completeExtPendingSuccess = (requestId: AcpError.AcpRequestId, value: unknown) =>
    resolveExtPending(requestId, ({ deferred }) => Deferred.succeed(deferred, value));

  const failAllExtPending = (error: AcpError.AcpError) =>
    Ref.getAndSet(extPending, new Map()).pipe(
      Effect.flatMap((pending) =>
        Effect.forEach([...pending.values()], ({ deferred }) => Deferred.fail(deferred, error), {
          discard: true,
        }),
      ),
    );

  const dispatchNotification = (notification: AcpIncomingNotification) =>
    Queue.offer(notificationQueue, notification).pipe(
      Effect.andThen(
        options.onNotification
          ? options.onNotification(notification).pipe(Effect.catch(() => Effect.void))
          : Effect.void,
      ),
      Effect.asVoid,
    );

  const emitClientProtocolError = (error: AcpError.AcpError) =>
    Queue.offer(clientQueue, {
      _tag: "ClientProtocolError",
      error: new RpcClientError.RpcClientError({
        reason: new RpcClientError.RpcClientDefect({
          message: "ACP protocol terminated.",
          cause: error,
        }),
      }),
    }).pipe(Effect.asVoid);

  const handleTermination = (classify: () => Effect.Effect<AcpError.AcpError>) =>
    Ref.modify(terminationHandled, (handled) => {
      if (handled) {
        return [Effect.void, true] as const;
      }
      return [
        Effect.gen(function* () {
          yield* Queue.offer(disconnects, 0);
          const error = yield* classify();
          yield* Deferred.fail(terminationFailure, error);
          yield* failAllExtPending(error);
          yield* emitClientProtocolError(error);
          if (options.onTermination) {
            yield* options.onTermination(error);
          }
        }),
        true,
      ] as const;
    }).pipe(Effect.flatten);

  const respondWithSuccess = (requestId: AcpError.AcpRequestId, value: unknown) =>
    offerOutgoing({
      _tag: "Exit",
      requestId,
      exit: {
        _tag: "Success",
        value,
      },
    });

  const respondWithError = (requestId: AcpError.AcpRequestId, error: AcpError.AcpRequestError) =>
    offerOutgoing({
      _tag: "Exit",
      requestId,
      exit: {
        _tag: "Failure",
        cause: [
          {
            _tag: "Fail",
            error: error.toProtocolError(),
          },
        ],
      },
    });

  const handleExtRequest = (message: RpcMessage.RequestEncoded) => {
    if (!options.onExtRequest) {
      return respondWithError(message.id, AcpError.AcpRequestError.methodNotFound(message.tag));
    }
    return options.onExtRequest(message.tag, message.payload).pipe(
      Effect.matchEffect({
        onFailure: (error) =>
          respondWithError(
            message.id,
            AcpError.AcpRequestError.fromExtensionHandlerError(error, message.tag),
          ),
        onSuccess: (value) => respondWithSuccess(message.id, value),
      }),
    );
  };

  const handleRequestEncoded = (message: RpcMessage.RequestEncoded) => {
    if (message.id === "") {
      if (message.tag === CLIENT_METHODS.session_update) {
        return decodeSessionUpdate(message.payload).pipe(
          Effect.map(
            (params) =>
              ({
                _tag: "SessionUpdate",
                method: CLIENT_METHODS.session_update,
                params: options.transformSessionUpdate?.(params) ?? params,
              }) satisfies AcpIncomingNotification,
          ),
          Effect.mapError((cause) =>
            AcpError.AcpProtocolParseError.fromSchemaError(
              "decode-notification-payload",
              CLIENT_METHODS.session_update,
              cause,
            ),
          ),
          Effect.flatMap(dispatchNotification),
        );
      }
      if (message.tag === CLIENT_METHODS.session_elicitation_complete) {
        return decodeElicitationComplete(message.payload).pipe(
          Effect.map(
            (params) =>
              ({
                _tag: "ElicitationComplete",
                method: CLIENT_METHODS.session_elicitation_complete,
                params,
              }) satisfies AcpIncomingNotification,
          ),
          Effect.mapError((cause) =>
            AcpError.AcpProtocolParseError.fromSchemaError(
              "decode-notification-payload",
              CLIENT_METHODS.session_elicitation_complete,
              cause,
            ),
          ),
          Effect.flatMap(dispatchNotification),
        );
      }
      return dispatchNotification({
        _tag: "ExtNotification",
        method: message.tag,
        params: message.payload,
      });
    }

    const observeIncoming =
      options.onIncomingRequest?.(String(message.id), message.tag, message.payload) ?? Effect.void;

    if (!options.serverRequestMethods.has(message.tag)) {
      return observeIncoming.pipe(Effect.andThen(handleExtRequest(message))).pipe(
        Effect.catchTags({
          AcpProtocolParseError: (error) =>
            Effect.logWarning(error).pipe(
              Effect.annotateLogs({
                method: message.tag,
                requestId: message.id,
                operation: error.operation,
              }),
              Effect.andThen(
                respondWithError(
                  message.id,
                  AcpError.AcpRequestError.fromExtensionResponseEncodingError(
                    message.tag,
                    message.id,
                    error,
                  ),
                ),
              ),
            ),
        }),
        Effect.asVoid,
      );
    }

    return observeIncoming.pipe(Effect.andThen(Queue.offer(serverQueue, message)), Effect.asVoid);
  };

  const forwardToRpcClient = (
    message: RpcMessage.ResponseChunkEncoded | RpcMessage.ResponseExitEncoded,
  ) =>
    isEffectRpcRequestId(message.requestId)
      ? Queue.offer(clientQueue, message).pipe(Effect.asVoid)
      : Effect.void;

  const handleExitEncoded = (message: RpcMessage.ResponseExitEncoded) =>
    Ref.get(extPending).pipe(
      Effect.flatMap((pending) => {
        const pendingRequest = pending.get(String(message.requestId));
        if (!pendingRequest) {
          return forwardToRpcClient(message);
        }
        if (message.exit._tag === "Success") {
          return completeExtPendingSuccess(message.requestId, message.exit.value);
        }
        const failure = message.exit.cause.find((entry) => entry._tag === "Fail");
        if (failure && isProtocolError(failure.error)) {
          return completeExtPendingFailure(
            message.requestId,
            AcpError.AcpRequestError.fromProtocolError(failure.error, {
              method: pendingRequest.method,
              requestId: message.requestId,
              cause: message.exit.cause,
            }),
          );
        }
        return completeExtPendingFailure(
          message.requestId,
          AcpError.AcpRequestError.fromExtensionResponseFailure(
            pendingRequest.method,
            message.requestId,
            message.exit.cause,
          ),
        );
      }),
    );

  const routeDecodedMessage = (
    message: RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded,
  ): Effect.Effect<void, AcpError.AcpError> => {
    switch (message._tag) {
      case "Request":
        return handleRequestEncoded(message);
      case "Exit":
        return handleExitEncoded(message);
      case "Chunk":
        return Ref.get(extPending).pipe(
          Effect.flatMap((pending) => {
            const pendingRequest = pending.get(String(message.requestId));
            return pendingRequest
              ? completeExtPendingFailure(
                  message.requestId,
                  AcpError.AcpRequestError.unsupportedStreamingResponse(
                    pendingRequest.method,
                    message.requestId,
                  ),
                )
              : forwardToRpcClient(message);
          }),
        );
      case "Defect":
      case "ClientProtocolError":
      case "Pong":
        return Queue.offer(clientQueue, message).pipe(Effect.asVoid);
      case "Ack":
      case "Interrupt":
      case "Ping":
      case "Eof":
        return Queue.offer(serverQueue, message).pipe(Effect.asVoid);
    }
  };

  yield* options.stdio.stdin.pipe(
    Stream.runForEach((data) =>
      (options.logIncoming
        ? logProtocol({
            direction: "incoming",
            stage: "raw",
            payload: typeof data === "string" ? data : new TextDecoder().decode(data),
          })
        : Effect.void
      ).pipe(
        Effect.flatMap(() =>
          Effect.try({
            try: () =>
              parser.decode(data) as ReadonlyArray<
                RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded
              >,
            catch: (cause) =>
              new AcpError.AcpProtocolParseError({
                operation: "decode-wire-message",
                cause,
              }),
          }),
        ),
        Effect.tap((messages) =>
          logProtocol({
            direction: "incoming",
            stage: "decoded",
            payload: messages,
          }),
        ),
        Effect.tapErrorTag("AcpProtocolParseError", (error) =>
          logProtocol({
            direction: "incoming",
            stage: "decode_failed",
            payload: {
              operation: error.operation,
              ...(error.method === undefined ? {} : { method: error.method }),
              ...(error.requestId === undefined ? {} : { requestId: error.requestId }),
              ...(error.issueCount === undefined ? {} : { issueCount: error.issueCount }),
              ...(error.issueKinds === undefined ? {} : { issueKinds: error.issueKinds }),
              ...(error.maximumPathDepth === undefined
                ? {}
                : { maximumPathDepth: error.maximumPathDepth }),
            },
          }),
        ),
        Effect.flatMap((messages) =>
          Effect.forEach(messages, routeDecodedMessage, {
            discard: true,
          }),
        ),
      ),
    ),
    Effect.matchEffect({
      onFailure: (error) => {
        const normalized: AcpError.AcpError = isAcpError(error)
          ? error
          : new AcpError.AcpTransportError({
              operation: "read-input-stream",
              cause: error,
            });
        return handleTermination(() => Effect.succeed(normalized));
      },
      onSuccess: () =>
        handleTermination(
          () =>
            options.terminationError ?? Effect.succeed(new AcpError.AcpInputStreamEndedError({})),
        ),
    }),
    Effect.forkScoped,
  );

  const failOutgoingWrites = (error: AcpError.AcpError) =>
    Ref.modify(outgoingWriterState, (state) => {
      const terminalError = state.terminalError ?? error;
      return [
        [terminalError, state.outstandingAcknowledgements] as const,
        {
          ...state,
          terminalError,
          outstandingAcknowledgements: new Set<Deferred.Deferred<void, AcpError.AcpError>>(),
        },
      ] as const;
    }).pipe(
      Effect.flatMap((acknowledgements) =>
        Effect.forEach(
          acknowledgements[1],
          (acknowledgement) => Deferred.fail(acknowledgement, acknowledgements[0]),
          { concurrency: "unbounded", discard: true },
        ),
      ),
      Effect.andThen(Queue.fail(outgoing, error)),
      Effect.asVoid,
    );
  const completeOutgoingWrites = () =>
    Ref.modify(outgoingWriterState, (state) => {
      return [
        state.outstandingAcknowledgements,
        {
          ...state,
          outstandingAcknowledgements: new Set<Deferred.Deferred<void, AcpError.AcpError>>(),
        },
      ] as const;
    }).pipe(
      Effect.flatMap((acknowledgements) =>
        Effect.forEach(
          acknowledgements,
          (acknowledgement) => Deferred.succeed(acknowledgement, undefined),
          { concurrency: "unbounded", discard: true },
        ),
      ),
      Effect.asVoid,
    );
  const acknowledgeOutgoingWrite = (acknowledgement: Deferred.Deferred<void, AcpError.AcpError>) =>
    Ref.update(outgoingWriterState, (state) => {
      const updated = new Set(state.outstandingAcknowledgements);
      updated.delete(acknowledgement);
      return { ...state, outstandingAcknowledgements: updated };
    }).pipe(Effect.andThen(Deferred.succeed(acknowledgement, undefined)), Effect.asVoid);
  yield* Stream.fromQueue(outgoing).pipe(
    Stream.flatMap((write) => {
      const acknowledgement = write.acknowledgement;
      const completion =
        acknowledgement === undefined
          ? Stream.empty
          : Stream.fromEffect(acknowledgeOutgoingWrite(acknowledgement)).pipe(Stream.drain);
      return Stream.make(write.payload).pipe(Stream.concat(completion));
    }),
    Stream.run(options.stdio.stdout()),
    Effect.onExit((exit) =>
      Exit.match(exit, {
        onFailure: (cause) => {
          const error = new AcpError.AcpTransportError({
            detail: Cause.hasInterruptsOnly(cause)
              ? "The ACP output writer was interrupted while closing"
              : "Failed to write an outgoing ACP message",
            cause,
          });
          return failOutgoingWrites(error).pipe(
            Effect.andThen(
              Cause.hasInterruptsOnly(cause)
                ? Effect.void
                : handleTermination(() => Effect.succeed(error)),
            ),
          );
        },
        onSuccess: () =>
          Ref.get(outgoingWriterState).pipe(
            Effect.flatMap((state) => {
              // Intentional Queue.end (serverProtocol.end) completes the writer
              // successfully. Do not invent a transport failure or client protocol
              // error for a clean RPC shutdown.
              const handled = state.intentionalEnd
                ? completeOutgoingWrites()
                : (() => {
                    const error = new AcpError.AcpTransportError({
                      detail: "ACP output writer ended before the protocol closed",
                      cause: "Output writer ended",
                    });
                    return failOutgoingWrites(error).pipe(
                      Effect.andThen(handleTermination(() => Effect.succeed(error))),
                    );
                  })();
              return handled.pipe(
                Effect.andThen(
                  options.testHooks?.onOutgoingWriterExit === undefined
                    ? Effect.void
                    : options.testHooks.onOutgoingWriterExit({
                        intentionalEnd: state.intentionalEnd,
                      }),
                ),
              );
            }),
          ),
      }),
    ),
    Effect.forkScoped,
  );

  const clientProtocol = RpcClient.Protocol.of({
    run: (_clientId, f) =>
      Stream.fromQueue(clientQueue).pipe(
        Stream.runForEach((message) => f(message)),
        Effect.forever,
      ),
    send: (_clientId, request) =>
      offerOutgoing(request).pipe(
        Effect.mapError(
          (error) =>
            new RpcClientError.RpcClientError({
              reason: new RpcClientError.RpcClientDefect({
                message: "Failed to send ACP protocol message.",
                cause: error,
              }),
            }),
        ),
      ),
    supportsAck: true,
    supportsTransferables: false,
  });

  const serverProtocol = RpcServer.Protocol.of({
    run: (f) =>
      Stream.fromQueue(serverQueue).pipe(
        Stream.runForEach((message) => f(0, message)),
        Effect.forever,
      ),
    disconnects,
    send: (_clientId, response) =>
      offerOutgoing(response).pipe(
        Effect.tapError((error) =>
          response._tag === "Exit" && options.onOutgoingResponseFailure !== undefined
            ? options.onOutgoingResponseFailure(String(response.requestId), error)
            : Effect.void,
        ),
        Effect.orDie,
      ),
    end: (_clientId) =>
      Ref.update(outgoingWriterState, (state) => ({ ...state, intentionalEnd: true })).pipe(
        Effect.andThen(Queue.end(outgoing)),
      ),
    clientIds: Effect.succeed(new Set([0])),
    initialMessage: Effect.succeedNone,
    supportsAck: true,
    supportsTransferables: false,
    supportsSpanPropagation: true,
  });

  // JSON-RPC notifications carry no `id`. The generic Request encoder emits `id: ""` plus
  // `headers`, which real agents (Grok CLI) parse as a malformed request and silently drop.
  // That made `session/cancel` a no-op against Grok while the lenient mock agent accepted it.
  const sendNotification = Effect.fn("sendNotification")(function* (
    method: string,
    payload: unknown,
  ) {
    yield* ensureActive;
    yield* logProtocol({
      direction: "outgoing",
      stage: "decoded",
      payload: { _tag: "Notification", tag: method, payload },
    });
    const exit = encodeJsonRpcNotification({ jsonrpc: "2.0", method, params: payload });
    if (Exit.isFailure(exit)) {
      return yield* AcpError.AcpProtocolParseError.fromEncodingError(
        method,
        undefined,
        Cause.squash(exit.cause),
      );
    }
    const encoded = `${exit.value}\n`;
    yield* logProtocol({ direction: "outgoing", stage: "raw", payload: encoded });
    yield* ensureActive;
    // FIFO queue admission preserves wire ordering against later requests
    // (a session/prompt sent after session/cancel cannot overtake it).
    yield* Queue.offer(outgoing, { payload: encoded }).pipe(
      Effect.mapError(
        (cause) =>
          new AcpError.AcpTransportError({
            detail: "Failed to queue an outgoing ACP notification",
            cause,
          }),
      ),
      Effect.filterOrFail(
        (accepted) => accepted,
        () =>
          new AcpError.AcpTransportError({
            detail: "The ACP output queue was closed before accepting a notification",
            cause: "Output queue closed",
          }),
      ),
    );
  });

  const sendRequest = Effect.fn("sendRequest")(function* (method: string, payload: unknown) {
    yield* ensureActive;
    const requestId = yield* Ref.modify(
      nextRequestId,
      (current) => [current, current + 1] as const,
    );
    const deferred = yield* Deferred.make<unknown, AcpError.AcpError>();
    yield* Ref.update(extPending, (pending) =>
      new Map(pending).set(String(requestId), { deferred, method }),
    );
    yield* offerOutgoing({
      _tag: "Request",
      id: requestId,
      tag: method,
      payload,
      headers: [],
    }).pipe(Effect.tapError(() => removeExtPending(requestId)));
    return yield* Deferred.await(deferred).pipe(
      Effect.onInterrupt(() => removeExtPending(requestId)),
    );
  });

  return {
    clientProtocol,
    serverProtocol,
    get incoming() {
      return Stream.fromQueue(notificationQueue);
    },
    request: sendRequest,
    notify: sendNotification,
  } satisfies AcpPatchedProtocol;
});

function isProtocolError(
  value: unknown,
): value is { code: number; message: string; data?: unknown } {
  return (
    typeof value === "object" &&
    value !== null &&
    "code" in value &&
    typeof value.code === "number" &&
    "message" in value &&
    typeof value.message === "string"
  );
}