1import type * as AcpCompat from "./compat.ts";
2import * as Cause from "effect/Cause";
3import * as Effect from "effect/Effect";
4import * as Deferred from "effect/Deferred";
5import * as Exit from "effect/Exit";
6import * as Option from "effect/Option";
7import * as Queue from "effect/Queue";
8import * as Ref from "effect/Ref";
9import type * as PlatformError from "effect/PlatformError";
10import * as Schema from "effect/Schema";
11import * as Scope from "effect/Scope";
12import * as Stream from "effect/Stream";
13import * as Stdio from "effect/Stdio";
14import * as RpcClient from "effect/rpc/RpcClient";
15import * as RpcClientError from "effect/rpc/RpcClientError";
16import * as RpcMessage from "effect/rpc/RpcMessage";
17import * as RpcSerialization from "effect/rpc/RpcSerialization";
18import * as RpcServer from "effect/rpc/RpcServer";
19
20import * as AcpSchema from "./schema.ts";
21import * as AcpSchemaV1 from "./_generated/schema-v1.gen.ts";
22import { CLIENT_METHODS } from "./_generated/meta.gen.ts";
23import * as AcpError from "./errors.ts";
24import { isolateNotificationHandler } from "./_internal/shared.ts";
25const isAcpError = Schema.is(AcpError.AcpError);
26
27export interface AcpProtocolLogEvent {
28 readonly direction: "incoming" | "outgoing";
29 readonly stage: "raw" | "decoded" | "decode_failed";
30 readonly payload: unknown;
31}
32
33/** Stable transport identity for an inbound ACP JSON-RPC request. */
34export interface AcpRequestContext {
35 readonly requestId: string;
36 readonly method: string;
37}
38
39/** Lossless string identity for JSON-RPC request IDs used by callback maps. */
40export function acpRequestIdentity(requestId: AcpError.AcpRequestId): string {
41 const prefix = "$t3:jsonrpc:";
42 if (typeof requestId === "number") return `${prefix}number:${requestId}`;
43 return requestId.startsWith(prefix) ? `${prefix}string:${requestId}` : requestId;
44}
45
46export type AcpRequestHandler<Request, Response> = (
47 request: Request,
48 context: AcpRequestContext,
49) => Effect.Effect<Response, AcpError.AcpError>;
50
51export type AcpIncomingNotification =
52 | {
53 readonly _tag: "SessionUpdate";
54 readonly method: typeof CLIENT_METHODS.session_update;
55 readonly params:
56 | AcpSchema.UpdateSessionNotification
57 | AcpSchemaV1.SessionNotification
58 | AcpCompat.SessionNotification;
59 }
60 | {
61 readonly _tag: "ElicitationComplete";
62 readonly method: typeof CLIENT_METHODS.elicitation_complete | "session/elicitation/complete";
63 readonly params: AcpSchema.CompleteElicitationNotification;
64 }
65 | {
66 readonly _tag: "ExtNotification";
67 readonly method: string;
68 readonly params: unknown;
69 };
70
71/** Standard I/O whose input can report provider-specific ACP failures. */
72export interface AcpStdio extends Omit<Stdio.Stdio, "stdin"> {
73 readonly stdin: Stream.Stream<Uint8Array, PlatformError.PlatformError | AcpError.AcpError>;
74}
75
76export interface AcpPatchedProtocolOptions {
77 readonly stdio: AcpStdio;
78 readonly terminationError?: Effect.Effect<AcpError.AcpError>;
79 readonly serverRequestMethods: ReadonlySet<string>;
80 readonly logIncoming?: boolean;
81 readonly logOutgoing?: boolean;
82 readonly logger?: (event: AcpProtocolLogEvent) => Effect.Effect<void, never>;
83 readonly onIncomingRequest?: (
84 requestId: string,
85 method: string,
86 payload: unknown,
87 ) => Effect.Effect<void, never>;
88
89 readonly transformSessionUpdate?: (
90 notification: AcpSchema.UpdateSessionNotification | AcpSchemaV1.SessionNotification,
91 ) =>
92 | AcpSchema.UpdateSessionNotification
93 | AcpSchemaV1.SessionNotification
94 | AcpCompat.SessionNotification;
95 readonly onNotification?: (
96 notification: AcpIncomingNotification,
97 ) => Effect.Effect<void, AcpError.AcpError, never>;
98 readonly onExtRequest?: (
99 method: string,
100 params: unknown,
101 context: AcpRequestContext,
102 ) => Effect.Effect<unknown, AcpError.AcpError, never>;
103 readonly onTermination?: (error: AcpError.AcpError) => Effect.Effect<void, never, never>;
104 readonly onOutgoingResponseFailure?: (
105 requestId: string,
106 error: AcpError.AcpError,
107 ) => Effect.Effect<void, never>;
108 readonly onOutgoingResponse?: (requestId: string) => Effect.Effect<void, never>;
109 readonly testHooks?: {
110 readonly onOutgoingWriteAdmitted?: () => Effect.Effect<void>;
111 /** Invoked from the stdout writer exit path (after intentional/unexpected handling). */
112 readonly onOutgoingWriterExit?: (input: {
113 readonly intentionalEnd: boolean;
114 }) => Effect.Effect<void>;
115 };
116}
117
118export interface AcpPatchedProtocol {
119 readonly clientProtocol: RpcClient.Protocol["Service"];
120 readonly serverProtocol: RpcServer.Protocol["Service"];
121 readonly incoming: Stream.Stream<AcpIncomingNotification>;
122 readonly request: (method: string, payload: unknown) => Effect.Effect<unknown, AcpError.AcpError>;
123 readonly notify: (method: string, payload: unknown) => Effect.Effect<void, AcpError.AcpError>;
124}
125
126interface AcpPendingRequest {
127 readonly deferred: Deferred.Deferred<unknown, AcpError.AcpError>;
128 readonly method: string;
129}
130
131interface AcpOutgoingWrite {
132 readonly acknowledgement?: Deferred.Deferred<void, AcpError.AcpError>;
133 readonly payload: string | Uint8Array;
134}
135
136interface AcpOutgoingWriterState {
137 readonly outstandingAcknowledgements: ReadonlySet<Deferred.Deferred<void, AcpError.AcpError>>;
138 readonly intentionalEnd: boolean;
139 readonly terminalError?: AcpError.AcpError;
140}
141
142const decodeSessionUpdate = Schema.decodeUnknownEffect(
143 Schema.Union([AcpSchema.UpdateSessionNotification, AcpSchemaV1.SessionNotification]),
144);
145const decodeElicitationComplete = Schema.decodeUnknownEffect(
146 AcpSchema.CompleteElicitationNotification,
147);
148const parserFactory = RpcSerialization.ndJsonRpc();
149// ndJsonRpc skips lines that are not JSON. A malformed agent line has to end the
150// session, so frames are split here and each one goes through the strict codec.
151const makeStrictNdJsonRpcParser = () => {
152 const codec = RpcSerialization.jsonRpc().makeUnsafe();
153 const decoder = new TextDecoder();
154 let buffer = "";
155 return {
156 decode: (bytes: Uint8Array | string): ReadonlyArray<unknown> => {
157 buffer += typeof bytes === "string" ? bytes : decoder.decode(bytes, { stream: true });
158 const lines = buffer.split("\n");
159 buffer = lines.pop() ?? "";
160 return lines.flatMap((line) => codec.decode(line));
161 },
162 encode: (response: Parameters<typeof codec.encode>[0]) => {
163 const encoded = codec.encode(response);
164 return encoded === undefined ? undefined : `${encoded}\n`;
165 },
166 };
167};
168const MAX_BUFFERED_RAW_NOTIFICATIONS = 32;
169// Outbound JSON-RPC notification: no `id`, so peers never treat it as a request.
170const encodeJsonRpcNotification = Schema.encodeUnknownExit(
171 Schema.fromJsonString(
172 Schema.Struct({
173 jsonrpc: Schema.Literal("2.0"),
174 method: Schema.String,
175 params: Schema.Unknown,
176 }),
177 ),
178);
179
180const isEffectRpcRequestId = (requestId: AcpError.AcpRequestId): boolean =>
181 typeof requestId === "number" && Number.isSafeInteger(requestId);
182
183/**
184 * Effect RPC's JSON-RPC codec treats a standard JSON-RPC error object as a
185 * defect unless it carries Effect's private `_tag: "Cause"` envelope. ACP
186 * agents correctly send the standard `{ code, message, data? }` shape, so
187 * restore it to the typed failure channel before handing it to RpcClient.
188 */
189function normalizeAcpJsonRpcError(
190 message: RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded,
191): RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded {
192 if (message._tag !== "Exit" || message.exit._tag !== "Failure") return message;
193 const [failure] = message.exit.cause;
194 if (
195 message.exit.cause.length !== 1 ||
196 failure?._tag !== "Die" ||
197 !isProtocolError(failure.defect)
198 ) {
199 return message;
200 }
201 return {
202 ...message,
203 exit: {
204 _tag: "Failure",
205 cause: [{ _tag: "Fail", error: failure.defect }],
206 },
207 };
208}
209
210export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(function* (
211 options: AcpPatchedProtocolOptions,
212): Effect.fn.Return<AcpPatchedProtocol, never, Scope.Scope> {
213 const parser = makeStrictNdJsonRpcParser();
214 const serverQueue = yield* Queue.unbounded<RpcMessage.FromClientEncoded>();
215 const clientQueue = yield* Queue.unbounded<RpcMessage.FromServerEncoded>();
216 const notificationQueue = yield* Queue.sliding<AcpIncomingNotification>(
217 MAX_BUFFERED_RAW_NOTIFICATIONS,
218 );
219 const disconnects = yield* Queue.unbounded<number>();
220 const outgoing = yield* Queue.unbounded<AcpOutgoingWrite, AcpError.AcpError | Cause.Done<void>>();
221 const outgoingWriterState = yield* Ref.make<AcpOutgoingWriterState>({
222 intentionalEnd: false,
223 outstandingAcknowledgements: new Set(),
224 });
225 const nextRequestId = yield* Ref.make(1);
226 const terminationHandled = yield* Ref.make(false);
227 const terminationFailure = yield* Deferred.make<never, AcpError.AcpError>();
228 const extPending = yield* Ref.make(new Map<string, AcpPendingRequest>());
229
230 const ensureActive = Ref.get(terminationHandled).pipe(
231 Effect.flatMap((terminated) => (terminated ? Deferred.await(terminationFailure) : Effect.void)),
232 );
233
234 const logProtocol = (event: AcpProtocolLogEvent) => {
235 if (event.direction === "incoming" && !options.logIncoming) {
236 return Effect.void;
237 }
238 if (event.direction === "outgoing" && !options.logOutgoing) {
239 return Effect.void;
240 }
241 return (
242 options.logger?.(event) ??
243 Effect.logDebug("ACP protocol event").pipe(Effect.annotateLogs({ event }))
244 );
245 };
246
247 const offerOutgoing = Effect.fn("offerOutgoing")(function* (
248 message: RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded,
249 ) {
250 // RpcClient emits `@effect/rpc/Interrupt` when a pending request's fiber is interrupted.
251 // ACP has no such method; agents log it as an error and cannot act on it, so drop it.
252 if (message._tag === "Interrupt") {
253 return;
254 }
255 yield* ensureActive;
256 yield* logProtocol({
257 direction: "outgoing",
258 stage: "decoded",
259 payload: message,
260 });
261
262 const method = message._tag === "Request" ? message.tag : undefined;
263 const encodedRequestId =
264 message._tag === "Request"
265 ? message.id
266 : "requestId" in message
267 ? message.requestId
268 : undefined;
269 const requestId = encodedRequestId === "" ? undefined : encodedRequestId;
270 const encoded = yield* Effect.try({
271 try: () => parser.encode(message),
272 catch: (cause) => AcpError.AcpProtocolParseError.fromEncodingError(method, requestId, cause),
273 });
274
275 if (encoded) {
276 yield* logProtocol({
277 direction: "outgoing",
278 stage: "raw",
279 payload: typeof encoded === "string" ? encoded : new TextDecoder().decode(encoded),
280 });
281
282 yield* ensureActive;
283 const acknowledgement =
284 message._tag === "Exit" ? yield* Deferred.make<void, AcpError.AcpError>() : undefined;
285 const admissionError = yield* Ref.modify(outgoingWriterState, (state) => {
286 if (state.terminalError !== undefined) {
287 return [state.terminalError, state] as const;
288 }
289 if (acknowledgement === undefined) {
290 return [undefined, state] as const;
291 }
292 return [
293 undefined,
294 {
295 ...state,
296 outstandingAcknowledgements: new Set(state.outstandingAcknowledgements).add(
297 acknowledgement,
298 ),
299 },
300 ] as const;
301 });
302 if (admissionError !== undefined) {
303 return yield* admissionError;
304 }
305 if (options.testHooks?.onOutgoingWriteAdmitted !== undefined) {
306 yield* options.testHooks.onOutgoingWriteAdmitted();
307 }
308 yield* Queue.offer(outgoing, {
309 payload: encoded,
310 ...(acknowledgement === undefined ? {} : { acknowledgement }),
311 }).pipe(
312 Effect.mapError(
313 (cause) =>
314 new AcpError.AcpTransportError({
315 detail: "Failed to queue an outgoing ACP message",
316 cause,
317 }),
318 ),
319 Effect.filterOrFail(
320 (accepted) => accepted,
321 () =>
322 new AcpError.AcpTransportError({
323 detail: "The ACP output queue was closed before accepting a message",
324 cause: "Output queue closed",
325 }),
326 ),
327 Effect.tapError((error) =>
328 acknowledgement === undefined
329 ? Effect.void
330 : Ref.update(outgoingWriterState, (state) => {
331 const updated = new Set(state.outstandingAcknowledgements);
332 updated.delete(acknowledgement);
333 return { ...state, outstandingAcknowledgements: updated };
334 }).pipe(Effect.andThen(Deferred.fail(acknowledgement, error)), Effect.asVoid),
335 ),
336 );
337 if (acknowledgement !== undefined) {
338 yield* Deferred.await(acknowledgement);
339 }
340 if (message._tag === "Exit" && options.onOutgoingResponse !== undefined) {
341 yield* options.onOutgoingResponse(acpRequestIdentity(message.requestId));
342 }
343 }
344 });
345
346 const resolveExtPending = (
347 requestId: AcpError.AcpRequestId,
348 onFound: (pendingRequest: AcpPendingRequest) => Effect.Effect<void>,
349 ) =>
350 Ref.modify(extPending, (pending) => {
351 const pendingKey = acpRequestIdentity(requestId);
352 const pendingRequest = pending.get(pendingKey);
353 if (!pendingRequest) {
354 return [Effect.void, pending] as const;
355 }
356 const next = new Map(pending);
357 next.delete(pendingKey);
358 return [onFound(pendingRequest), next] as const;
359 }).pipe(Effect.flatten);
360
361 const removeExtPending = (requestId: AcpError.AcpRequestId) =>
362 Ref.update(extPending, (pending) => {
363 const pendingKey = acpRequestIdentity(requestId);
364 if (!pending.has(pendingKey)) {
365 return pending;
366 }
367 const next = new Map(pending);
368 next.delete(pendingKey);
369 return next;
370 });
371
372 const completeExtPendingFailure = (requestId: AcpError.AcpRequestId, error: AcpError.AcpError) =>
373 resolveExtPending(requestId, ({ deferred }) => Deferred.fail(deferred, error));
374
375 const completeExtPendingSuccess = (requestId: AcpError.AcpRequestId, value: unknown) =>
376 resolveExtPending(requestId, ({ deferred }) => Deferred.succeed(deferred, value));
377
378 const failAllExtPending = (error: AcpError.AcpError) =>
379 Ref.getAndSet(extPending, new Map()).pipe(
380 Effect.flatMap((pending) =>
381 Effect.forEach([...pending.values()], ({ deferred }) => Deferred.fail(deferred, error), {
382 discard: true,
383 }),
384 ),
385 );
386
387 const dispatchNotification = (notification: AcpIncomingNotification) =>
388 Queue.offer(notificationQueue, notification).pipe(
389 Effect.andThen(
390 // A failing or dying handler must not stop the reader, or every later
391 // message on the connection goes unanswered.
392 options.onNotification
393 ? isolateNotificationHandler(options.onNotification(notification))
394 : Effect.void,
395 ),
396 Effect.asVoid,
397 );
398
399 const emitClientProtocolError = (error: AcpError.AcpError) =>
400 Queue.offer(clientQueue, {
401 _tag: "ClientProtocolError",
402 error: new RpcClientError.RpcClientError({
403 reason: new RpcClientError.RpcClientDefect({
404 message: "ACP protocol terminated.",
405 cause: error,
406 }),
407 }),
408 }).pipe(Effect.asVoid);
409
410 const handleTermination = (classify: () => Effect.Effect<AcpError.AcpError>) =>
411 Ref.modify(terminationHandled, (handled) => {
412 if (handled) {
413 return [Effect.void, true] as const;
414 }
415 return [
416 Effect.gen(function* () {
417 yield* Queue.offer(disconnects, 0);
418 const error = yield* classify();
419 yield* Deferred.fail(terminationFailure, error);
420 yield* failAllExtPending(error);
421 yield* emitClientProtocolError(error);
422 if (options.onTermination) {
423 yield* options.onTermination(error);
424 }
425 }),
426 true,
427 ] as const;
428 }).pipe(Effect.flatten);
429
430 const respondWithSuccess = (requestId: AcpError.AcpRequestId, value: unknown) =>
431 offerOutgoing({
432 _tag: "Exit",
433 requestId,
434 exit: {
435 _tag: "Success",
436 value,
437 },
438 });
439
440 const respondWithError = (requestId: AcpError.AcpRequestId, error: AcpError.AcpRequestError) =>
441 offerOutgoing({
442 _tag: "Exit",
443 requestId,
444 exit: {
445 _tag: "Failure",
446 cause: [
447 {
448 _tag: "Fail",
449 error: error.toProtocolError(),
450 },
451 ],
452 },
453 });
454
455 const handleExtRequest = (message: RpcMessage.RequestEncoded) => {
456 if (!options.onExtRequest) {
457 return respondWithError(message.id, AcpError.AcpRequestError.methodNotFound(message.tag));
458 }
459 return options
460 .onExtRequest(message.tag, message.payload, {
461 requestId: acpRequestIdentity(message.id),
462 method: message.tag,
463 })
464 .pipe(
465 Effect.matchCauseEffect({
466 // A dying handler answers its own request, like a core handler, and
467 // leaves the reader running. A defect wins over a typed failure in
468 // the same cause, so it is never hidden behind an expected error.
469 onFailure: (cause) => {
470 const failure = Cause.hasDies(cause) ? Option.none() : Cause.findErrorOption(cause);
471 if (Option.isSome(failure)) {
472 return respondWithError(
473 message.id,
474 AcpError.AcpRequestError.fromExtensionHandlerError(failure.value, message.tag),
475 );
476 }
477 return Effect.logError(
478 `ACP extension request handler failed for '${message.tag}'`,
479 cause,
480 ).pipe(
481 Effect.andThen(
482 respondWithError(
483 message.id,
484 AcpError.AcpRequestError.internalError(
485 `ACP extension request handler failed for method '${message.tag}'`,
486 undefined,
487 {
488 method: message.tag,
489 operation: "handle-extension-request",
490 cause: Cause.squash(cause),
491 },
492 ),
493 ),
494 ),
495 );
496 },
497 onSuccess: (value) => respondWithSuccess(message.id, value),
498 }),
499 );
500 };
501
502 const handleRequestEncoded = (message: RpcMessage.RequestEncoded) => {
503 if (message.id === "") {
504 if (message.tag === CLIENT_METHODS.session_update) {
505 return decodeSessionUpdate(message.payload).pipe(
506 Effect.map(
507 (params) =>
508 ({
509 _tag: "SessionUpdate",
510 method: CLIENT_METHODS.session_update,
511 params: options.transformSessionUpdate?.(params) ?? params,
512 }) satisfies AcpIncomingNotification,
513 ),
514 Effect.mapError((cause) =>
515 AcpError.AcpProtocolParseError.fromSchemaError(
516 "decode-notification-payload",
517 CLIENT_METHODS.session_update,
518 cause,
519 ),
520 ),
521 Effect.flatMap(dispatchNotification),
522 );
523 }
524 if (
525 message.tag === CLIENT_METHODS.elicitation_complete ||
526 message.tag === "session/elicitation/complete"
527 ) {
528 const method = message.tag;
529 return decodeElicitationComplete(message.payload).pipe(
530 Effect.map(
531 (params) =>
532 ({
533 _tag: "ElicitationComplete",
534 method,
535 params,
536 }) satisfies AcpIncomingNotification,
537 ),
538 Effect.mapError((cause) =>
539 AcpError.AcpProtocolParseError.fromSchemaError(
540 "decode-notification-payload",
541 method,
542 cause,
543 ),
544 ),
545 Effect.flatMap(dispatchNotification),
546 );
547 }
548 return dispatchNotification({
549 _tag: "ExtNotification",
550 method: message.tag,
551 params: message.payload,
552 });
553 }
554
555 const observeIncoming =
556 options.onIncomingRequest?.(acpRequestIdentity(message.id), message.tag, message.payload) ??
557 Effect.void;
558
559 if (!options.serverRequestMethods.has(message.tag)) {
560 return observeIncoming.pipe(
561 Effect.andThen(handleExtRequest(message)),
562 Effect.catchTags({
563 AcpProtocolParseError: (error) =>
564 Effect.logWarning(error).pipe(
565 Effect.annotateLogs({
566 method: message.tag,
567 requestId: message.id,
568 operation: error.operation,
569 }),
570 Effect.andThen(
571 respondWithError(
572 message.id,
573 AcpError.AcpRequestError.fromExtensionResponseEncodingError(
574 message.tag,
575 message.id,
576 error,
577 ),
578 ),
579 ),
580 ),
581 }),
582 Effect.asVoid,
583 );
584 }
585
586 return observeIncoming.pipe(Effect.andThen(Queue.offer(serverQueue, message)), Effect.asVoid);
587 };
588
589 const forwardToRpcClient = (
590 message: RpcMessage.ResponseChunkEncoded | RpcMessage.ResponseExitEncoded,
591 ) =>
592 isEffectRpcRequestId(message.requestId)
593 ? Queue.offer(clientQueue, message).pipe(Effect.asVoid)
594 : Effect.void;
595
596 const handleExitEncoded = (message: RpcMessage.ResponseExitEncoded) =>
597 Ref.get(extPending).pipe(
598 Effect.flatMap((pending) => {
599 const pendingRequest = pending.get(acpRequestIdentity(message.requestId));
600 if (!pendingRequest) {
601 return forwardToRpcClient(message);
602 }
603 if (message.exit._tag === "Success") {
604 return completeExtPendingSuccess(message.requestId, message.exit.value);
605 }
606 const failure = message.exit.cause.find((entry) => entry._tag === "Fail");
607 if (failure && isProtocolError(failure.error)) {
608 return completeExtPendingFailure(
609 message.requestId,
610 AcpError.AcpRequestError.fromProtocolError(failure.error, {
611 method: pendingRequest.method,
612 requestId: message.requestId,
613 cause: message.exit.cause,
614 }),
615 );
616 }
617 return completeExtPendingFailure(
618 message.requestId,
619 AcpError.AcpRequestError.fromExtensionResponseFailure(
620 pendingRequest.method,
621 message.requestId,
622 message.exit.cause,
623 ),
624 );
625 }),
626 );
627
628 const routeDecodedMessage = (
629 message: RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded,
630 ): Effect.Effect<void, AcpError.AcpError> => {
631 switch (message._tag) {
632 case "Request":
633 return handleRequestEncoded(message);
634 case "Exit":
635 return handleExitEncoded(message);
636 case "Chunk":
637 return Ref.get(extPending).pipe(
638 Effect.flatMap((pending) => {
639 const pendingRequest = pending.get(acpRequestIdentity(message.requestId));
640 return pendingRequest
641 ? completeExtPendingFailure(
642 message.requestId,
643 AcpError.AcpRequestError.unsupportedStreamingResponse(
644 pendingRequest.method,
645 message.requestId,
646 ),
647 )
648 : forwardToRpcClient(message);
649 }),
650 );
651 case "Defect":
652 case "ClientProtocolError":
653 case "Pong":
654 return Queue.offer(clientQueue, message).pipe(Effect.asVoid);
655 case "Ack":
656 case "Interrupt":
657 case "Ping":
658 case "Eof":
659 return Queue.offer(serverQueue, message).pipe(Effect.asVoid);
660 }
661 };
662
663 yield* options.stdio.stdin.pipe(
664 Stream.runForEach((data) =>
665 (options.logIncoming
666 ? logProtocol({
667 direction: "incoming",
668 stage: "raw",
669 payload: typeof data === "string" ? data : new TextDecoder().decode(data),
670 })
671 : Effect.void
672 ).pipe(
673 Effect.flatMap(() =>
674 Effect.try({
675 try: () =>
676 (
677 parser.decode(data) as ReadonlyArray<
678 RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded
679 >
680 ).map(normalizeAcpJsonRpcError),
681 catch: (cause) =>
682 new AcpError.AcpProtocolParseError({
683 operation: "decode-wire-message",
684 cause,
685 }),
686 }),
687 ),
688 Effect.tap((messages) =>
689 logProtocol({
690 direction: "incoming",
691 stage: "decoded",
692 payload: messages,
693 }),
694 ),
695 Effect.tapErrorTag("AcpProtocolParseError", (error) =>
696 logProtocol({
697 direction: "incoming",
698 stage: "decode_failed",
699 payload: {
700 operation: error.operation,
701 ...(error.method === undefined ? {} : { method: error.method }),
702 ...(error.requestId === undefined ? {} : { requestId: error.requestId }),
703 ...(error.issueCount === undefined ? {} : { issueCount: error.issueCount }),
704 ...(error.issueKinds === undefined ? {} : { issueKinds: error.issueKinds }),
705 ...(error.maximumPathDepth === undefined
706 ? {}
707 : { maximumPathDepth: error.maximumPathDepth }),
708 },
709 }),
710 ),
711 Effect.flatMap((messages) =>
712 Effect.forEach(messages, routeDecodedMessage, {
713 discard: true,
714 }),
715 ),
716 ),
717 ),
718 // Anything that ends the reader, including a defect in a callback it runs,
719 // terminates the connection so pending requests fail instead of hanging.
720 Effect.matchCauseEffect({
721 onFailure: (cause) => {
722 // The reader's own interruption never gets here: an interrupted fiber
723 // skips failure handlers. An interrupt raised inside it ends it too.
724 const failure = Cause.findErrorOption(cause);
725 const error = Option.isSome(failure) ? failure.value : Cause.squash(cause);
726 const normalized: AcpError.AcpError = isAcpError(error)
727 ? error
728 : new AcpError.AcpTransportError({
729 operation: "read-input-stream",
730 cause: error,
731 });
732 return handleTermination(() => Effect.succeed(normalized));
733 },
734 onSuccess: () =>
735 handleTermination(
736 () =>
737 options.terminationError ?? Effect.succeed(new AcpError.AcpInputStreamEndedError({})),
738 ),
739 }),
740 Effect.forkScoped,
741 );
742
743 const failOutgoingWrites = (error: AcpError.AcpError) =>
744 Ref.modify(outgoingWriterState, (state) => {
745 const terminalError = state.terminalError ?? error;
746 return [
747 [terminalError, state.outstandingAcknowledgements] as const,
748 {
749 ...state,
750 terminalError,
751 outstandingAcknowledgements: new Set<Deferred.Deferred<void, AcpError.AcpError>>(),
752 },
753 ] as const;
754 }).pipe(
755 Effect.flatMap((acknowledgements) =>
756 Effect.forEach(
757 acknowledgements[1],
758 (acknowledgement) => Deferred.fail(acknowledgement, acknowledgements[0]),
759 { concurrency: "unbounded", discard: true },
760 ),
761 ),
762 Effect.andThen(Queue.fail(outgoing, error)),
763 Effect.asVoid,
764 );
765 const completeOutgoingWrites = () =>
766 Ref.modify(outgoingWriterState, (state) => {
767 return [
768 state.outstandingAcknowledgements,
769 {
770 ...state,
771 outstandingAcknowledgements: new Set<Deferred.Deferred<void, AcpError.AcpError>>(),
772 },
773 ] as const;
774 }).pipe(
775 Effect.flatMap((acknowledgements) =>
776 Effect.forEach(
777 acknowledgements,
778 (acknowledgement) => Deferred.succeed(acknowledgement, undefined),
779 { concurrency: "unbounded", discard: true },
780 ),
781 ),
782 Effect.asVoid,
783 );
784 const acknowledgeOutgoingWrite = (acknowledgement: Deferred.Deferred<void, AcpError.AcpError>) =>
785 Ref.update(outgoingWriterState, (state) => {
786 const updated = new Set(state.outstandingAcknowledgements);
787 updated.delete(acknowledgement);
788 return { ...state, outstandingAcknowledgements: updated };
789 }).pipe(Effect.andThen(Deferred.succeed(acknowledgement, undefined)), Effect.asVoid);
790 yield* Stream.fromQueue(outgoing).pipe(
791 Stream.flatMap((write) => {
792 const acknowledgement = write.acknowledgement;
793 const completion =
794 acknowledgement === undefined
795 ? Stream.empty
796 : Stream.fromEffect(acknowledgeOutgoingWrite(acknowledgement)).pipe(Stream.drain);
797 return Stream.make(write.payload).pipe(Stream.concat(completion));
798 }),
799 Stream.run(options.stdio.stdout()),
800 Effect.onExit((exit) =>
801 Exit.match(exit, {
802 onFailure: (cause) => {
803 const error = new AcpError.AcpTransportError({
804 detail: Cause.hasInterruptsOnly(cause)
805 ? "The ACP output writer was interrupted while closing"
806 : "Failed to write an outgoing ACP message",
807 cause,
808 });
809 return failOutgoingWrites(error).pipe(
810 Effect.andThen(
811 Cause.hasInterruptsOnly(cause)
812 ? Effect.void
813 : handleTermination(() => Effect.succeed(error)),
814 ),
815 );
816 },
817 onSuccess: () =>
818 Ref.get(outgoingWriterState).pipe(
819 Effect.flatMap((state) => {
820 // Intentional Queue.end (serverProtocol.end) completes the writer
821 // successfully. Do not invent a transport failure or client protocol
822 // error for a clean RPC shutdown.
823 const handled = state.intentionalEnd
824 ? completeOutgoingWrites()
825 : (() => {
826 const error = new AcpError.AcpTransportError({
827 detail: "ACP output writer ended before the protocol closed",
828 cause: "Output writer ended",
829 });
830 return failOutgoingWrites(error).pipe(
831 Effect.andThen(handleTermination(() => Effect.succeed(error))),
832 );
833 })();
834 return handled.pipe(
835 Effect.andThen(
836 options.testHooks?.onOutgoingWriterExit === undefined
837 ? Effect.void
838 : options.testHooks.onOutgoingWriterExit({
839 intentionalEnd: state.intentionalEnd,
840 }),
841 ),
842 );
843 }),
844 ),
845 }),
846 ),
847 Effect.forkScoped,
848 );
849
850 const clientProtocol = RpcClient.Protocol.of({
851 run: (_clientId, f) =>
852 Stream.fromQueue(clientQueue).pipe(
853 Stream.runForEach((message) => f(message)),
854 Effect.forever,
855 ),
856 send: (_clientId, request) =>
857 offerOutgoing(request).pipe(
858 Effect.mapError(
859 (error) =>
860 new RpcClientError.RpcClientError({
861 reason: new RpcClientError.RpcClientDefect({
862 message: "Failed to send ACP protocol message.",
863 cause: error,
864 }),
865 }),
866 ),
867 ),
868 supportsAck: true,
869 supportsTransferables: false,
870 codecFor: parserFactory.codecFor,
871 });
872
873 const serverProtocol = RpcServer.Protocol.of({
874 run: (f) =>
875 Stream.fromQueue(serverQueue).pipe(
876 Stream.runForEach((message) => f(0, message)),
877 Effect.forever,
878 ),
879 disconnects,
880 send: (_clientId, response) =>
881 offerOutgoing(response).pipe(
882 Effect.tapError((error) =>
883 response._tag === "Exit" && options.onOutgoingResponseFailure !== undefined
884 ? options.onOutgoingResponseFailure(acpRequestIdentity(response.requestId), error)
885 : Effect.void,
886 ),
887 Effect.orDie,
888 ),
889 end: (_clientId) =>
890 Ref.update(outgoingWriterState, (state) => ({ ...state, intentionalEnd: true })).pipe(
891 Effect.andThen(Queue.end(outgoing)),
892 ),
893 clientIds: Effect.succeed(new Set([0])),
894 initialMessage: Effect.succeedNone,
895 supportsAck: true,
896 supportsTransferables: false,
897 codecFor: parserFactory.codecFor,
898 supportsSpanPropagation: true,
899 supportsNotifications: true,
900 });
901
902 // JSON-RPC notifications carry no `id`. Encoding a Request without `isNotification`
903 // emits an `id`, which real agents (Grok CLI) parse as a malformed request and silently drop.
904 // That made `session/cancel` a no-op against Grok while the lenient mock agent accepted it.
905 const sendNotification = Effect.fn("sendNotification")(function* (
906 method: string,
907 payload: unknown,
908 ) {
909 yield* ensureActive;
910 yield* logProtocol({
911 direction: "outgoing",
912 stage: "decoded",
913 payload: { _tag: "Notification", tag: method, payload },
914 });
915 const exit = encodeJsonRpcNotification({ jsonrpc: "2.0", method, params: payload });
916 if (Exit.isFailure(exit)) {
917 return yield* AcpError.AcpProtocolParseError.fromEncodingError(
918 method,
919 undefined,
920 Cause.squash(exit.cause),
921 );
922 }
923 const encoded = `${exit.value}\n`;
924 yield* logProtocol({ direction: "outgoing", stage: "raw", payload: encoded });
925 yield* ensureActive;
926 // FIFO queue admission preserves wire ordering against later requests
927 // (a session/prompt sent after session/cancel cannot overtake it).
928 yield* Queue.offer(outgoing, { payload: encoded }).pipe(
929 Effect.mapError(
930 (cause) =>
931 new AcpError.AcpTransportError({
932 detail: "Failed to queue an outgoing ACP notification",
933 cause,
934 }),
935 ),
936 Effect.filterOrFail(
937 (accepted) => accepted,
938 () =>
939 new AcpError.AcpTransportError({
940 detail: "The ACP output queue was closed before accepting a notification",
941 cause: "Output queue closed",
942 }),
943 ),
944 );
945 });
946
947 const sendRequest = Effect.fn("sendRequest")(function* (method: string, payload: unknown) {
948 yield* ensureActive;
949 const requestId = yield* Ref.modify(
950 nextRequestId,
951 (current) => [current, current + 1] as const,
952 );
953 const deferred = yield* Deferred.make<unknown, AcpError.AcpError>();
954 yield* Ref.update(extPending, (pending) =>
955 new Map(pending).set(acpRequestIdentity(requestId), { deferred, method }),
956 );
957 yield* offerOutgoing({
958 _tag: "Request",
959 id: requestId,
960 tag: method,
961 payload,
962 headers: [],
963 }).pipe(Effect.tapError(() => removeExtPending(requestId)));
964 return yield* Deferred.await(deferred).pipe(
965 Effect.onInterrupt(() => removeExtPending(requestId)),
966 );
967 });
968
969 return {
970 clientProtocol,
971 serverProtocol,
972 get incoming() {
973 return Stream.fromQueue(notificationQueue);
974 },
975 request: sendRequest,
976 notify: sendNotification,
977 } satisfies AcpPatchedProtocol;
978});
979
980const isProtocolError = Schema.is(AcpSchema.Error);