1import type * as AcpCompat from "./compat.ts";
2import * as Cause from "effect/Cause";
3import * as Clock from "effect/Clock";
4import * as Effect from "effect/Effect";
5import * as Deferred from "effect/Deferred";
6import * as Exit from "effect/Exit";
7import * as Option from "effect/Option";
8import * as Queue from "effect/Queue";
9import * as Ref from "effect/Ref";
10import type * as PlatformError from "effect/PlatformError";
11import * as Schema from "effect/Schema";
12import * as Scope from "effect/Scope";
13import * as Stream from "effect/Stream";
14import * as Stdio from "effect/Stdio";
15import * as RpcClient from "effect/rpc/RpcClient";
16import * as RpcClientError from "effect/rpc/RpcClientError";
17import * as RpcMessage from "effect/rpc/RpcMessage";
18import * as RpcSerialization from "effect/rpc/RpcSerialization";
19import * as RpcServer from "effect/rpc/RpcServer";
20
21import * as AcpSchema from "./schema.ts";
22import * as AcpSchemaV1 from "./_generated/schema-v1.gen.ts";
23import { CLIENT_METHODS } from "./_generated/meta.gen.ts";
24import * as AcpError from "./errors.ts";
25import { isolateNotificationHandler } from "./_internal/shared.ts";
26const isAcpError = Schema.is(AcpError.AcpError);
27
28export interface AcpProtocolLogEvent {
29 readonly direction: "incoming" | "outgoing";
30 readonly stage: "raw" | "decoded" | "decode_failed";
31 readonly payload: unknown;
32}
33
34export interface AcpRequestContext {
35 readonly requestId: string;
36 readonly method: string;
37}
38
39export function acpRequestIdentity(requestId: AcpError.AcpRequestId): string {
40 const prefix = "$t3:jsonrpc:";
41 if (typeof requestId === "number") return `${prefix}number:${requestId}`;
42 return requestId.startsWith(prefix) ? `${prefix}string:${requestId}` : requestId;
43}
44
45export type AcpRequestHandler<Request, Response> = (
46 request: Request,
47 context: AcpRequestContext,
48) => Effect.Effect<Response, AcpError.AcpError>;
49
50export type AcpIncomingNotification =
51 | {
52 readonly _tag: "SessionUpdate";
53 readonly method: typeof CLIENT_METHODS.session_update;
54 readonly params:
55 | AcpSchema.UpdateSessionNotification
56 | AcpSchemaV1.SessionNotification
57 | AcpCompat.SessionNotification;
58 }
59 | {
60 readonly _tag: "ElicitationComplete";
61 readonly method: typeof CLIENT_METHODS.elicitation_complete | "session/elicitation/complete";
62 readonly params: AcpSchema.CompleteElicitationNotification;
63 }
64 | {
65 readonly _tag: "ExtNotification";
66 readonly method: string;
67 readonly params: unknown;
68 };
69
70export interface AcpStdio extends Omit<Stdio.Stdio, "stdin"> {
71 readonly stdin: Stream.Stream<Uint8Array, PlatformError.PlatformError | AcpError.AcpError>;
72}
73
74export interface AcpPatchedProtocolOptions {
75 readonly stdio: AcpStdio;
76 readonly terminationError?: Effect.Effect<AcpError.AcpError>;
77 readonly serverRequestMethods: ReadonlySet<string>;
78 readonly logIncoming?: boolean;
79 readonly logOutgoing?: boolean;
80 readonly logger?: (event: AcpProtocolLogEvent) => Effect.Effect<void, never>;
81 readonly onIncomingRequest?: (
82 requestId: string,
83 method: string,
84 payload: unknown,
85 ) => Effect.Effect<void, never>;
86
87 readonly transformSessionUpdate?: (
88 notification:
89 | AcpSchema.UpdateSessionNotification
90 | AcpSchemaV1.SessionNotification
91 | AcpCompat.SessionNotification,
92 ) =>
93 | AcpSchema.UpdateSessionNotification
94 | AcpSchemaV1.SessionNotification
95 | AcpCompat.SessionNotification;
96 readonly onNotification?: (
97 notification: AcpIncomingNotification,
98 ) => Effect.Effect<void, AcpError.AcpError, never>;
99 readonly onExtRequest?: (
100 method: string,
101 params: unknown,
102 context: AcpRequestContext,
103 ) => Effect.Effect<unknown, AcpError.AcpError, never>;
104 readonly onTermination?: (error: AcpError.AcpError) => Effect.Effect<void, never, never>;
105 readonly onOutgoingResponseFailure?: (
106 requestId: string,
107 error: AcpError.AcpError,
108 ) => Effect.Effect<void, never>;
109 readonly onOutgoingResponse?: (requestId: string) => Effect.Effect<void, never>;
110 readonly testHooks?: {
111 readonly onOutgoingWriteAdmitted?: () => Effect.Effect<void>;
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
131/** A decoded wire message and the length of the JSON line it arrived on. */
132interface AcpDecodedWireMessage {
133 readonly message: RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded;
134 readonly lineLength: number;
135}
136
137interface AcpBufferedRawNotification {
138 readonly notification: AcpIncomingNotification;
139 readonly lineLength: number;
140}
141
142interface AcpOutgoingWrite {
143 readonly acknowledgement?: Deferred.Deferred<void, AcpError.AcpError>;
144 readonly payload: string | Uint8Array;
145}
146
147interface AcpOutgoingWriterState {
148 readonly outstandingAcknowledgements: ReadonlySet<Deferred.Deferred<void, AcpError.AcpError>>;
149 readonly intentionalEnd: boolean;
150 readonly terminalError?: AcpError.AcpError;
151}
152
153const decodeSessionUpdate = Schema.decodeUnknownEffect(
154 Schema.Union([
155 // Claude ACP v2 reports prompt failures on idle; alpha.3 omits that field.
156 Schema.Struct({
157 ...AcpSchema.UpdateSessionNotification.fields,
158 update: Schema.Struct({
159 sessionUpdate: Schema.Literal("state_update"),
160 state: Schema.Literal("idle"),
161 stopReason: Schema.optionalKey(Schema.NullOr(AcpSchema.StopReason)),
162 usage: Schema.optionalKey(Schema.NullOr(AcpSchema.Usage)),
163 error: Schema.optionalKey(Schema.NullOr(AcpSchema.Error)),
164 _meta: AcpSchema.UpdateSessionNotification.fields._meta,
165 }),
166 }),
167 AcpSchema.UpdateSessionNotification,
168 AcpSchemaV1.SessionNotification,
169 ]),
170);
171const decodeElicitationComplete = Schema.decodeUnknownEffect(
172 AcpSchema.CompleteElicitationNotification,
173);
174const parserFactory = RpcSerialization.ndJsonRpc();
175// ndJsonRpc skips lines that are not JSON. A malformed agent line has to end the
176// session, so frames are split here and each one goes through the strict codec.
177const makeStrictNdJsonRpcParser = () => {
178 const codec = RpcSerialization.jsonRpc().makeUnsafe();
179 const decoder = new TextDecoder();
180 let buffer = "";
181 const decodeLine = (line: string): ReadonlyArray<AcpDecodedWireMessage> =>
182 codec.decode(line).map((message) => ({
183 message: normalizeAcpJsonRpcError(
184 message as RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded,
185 ),
186 lineLength: line.length,
187 }));
188 return {
189 // Every message in a batch line is charged the whole line.
190 decode: (bytes: Uint8Array | string): ReadonlyArray<AcpDecodedWireMessage> => {
191 const chunk = typeof bytes === "string" ? bytes : decoder.decode(bytes, { stream: true });
192 const lastNewline = chunk.lastIndexOf("\n");
193 // Splitting the whole buffer per chunk is quadratic for multi-megabyte lines.
194 if (lastNewline === -1) {
195 buffer += chunk;
196 return [];
197 }
198 const lines = (buffer + chunk.slice(0, lastNewline)).split("\n");
199 buffer = chunk.slice(lastNewline + 1);
200 return lines.flatMap(decodeLine);
201 },
202 // A truncated final line is left to the exit status, which explains a crash better.
203 end: (): ReadonlyArray<AcpDecodedWireMessage> => {
204 const line = buffer + decoder.decode();
205 buffer = "";
206 try {
207 return line.trim() === "" ? [] : decodeLine(line);
208 } catch {
209 return [];
210 }
211 },
212 encode: (response: Parameters<typeof codec.encode>[0]) => {
213 const encoded = codec.encode(response);
214 if (encoded === undefined) return undefined;
215 const json = typeof encoded === "string" ? encoded : new TextDecoder().decode(encoded);
216 if (!json.includes('"_tag":"Cause"')) return `${json}\n`;
217 return `${JSON.stringify(normalizeOutgoingFrame(JSON.parse(json)))}\n`;
218 },
219 };
220};
221const isCauseResponse = Schema.is(
222 Schema.Struct({
223 error: Schema.TaggedStruct("Cause", { data: Schema.Array(Schema.Unknown) }),
224 }),
225);
226const isProtocolFailure = Schema.is(Schema.TaggedStruct("Fail", { error: AcpSchema.Error }));
227const isInterruptReason = Schema.is(Schema.TaggedStruct("Interrupt", {}));
228
229// Normalize only Effect's error envelope after the codec has completed batch tracking.
230function normalizeOutgoingFrame(frame: unknown): unknown {
231 if (Array.isArray(frame)) return frame.map(normalizeOutgoingFrame);
232 if (!isCauseResponse(frame)) return frame;
233 const reasons = frame.error.data.filter((reason) => !isInterruptReason(reason));
234 const [failure] = reasons;
235 const error =
236 reasons.length === 1 && isProtocolFailure(failure)
237 ? failure.error
238 : AcpError.AcpRequestError.internalError().toProtocolError();
239 return {
240 ...frame,
241 error: {
242 code: error.code,
243 message: error.message,
244 ...(error.data === undefined ? {} : { data: error.data }),
245 },
246 };
247}
248
249// Unread raw notifications are best-effort. The newest is always kept; older
250// ones are dropped once the backlog passes either limit. The length limit counts
251// JSON line characters (UTF-16 code units), not bytes or heap size.
252const MAX_BUFFERED_RAW_NOTIFICATIONS = 32;
253const MAX_BUFFERED_RAW_NOTIFICATION_JSON_LENGTH = 16 * 1024 * 1024;
254
255const decodeFailureDiagnostics = (error: AcpError.AcpProtocolParseError) => ({
256 operation: error.operation,
257 ...(error.method === undefined ? {} : { method: error.method }),
258 ...(error.requestId === undefined ? {} : { requestId: error.requestId }),
259 ...(error.issueCount === undefined ? {} : { issueCount: error.issueCount }),
260 ...(error.issueKinds === undefined ? {} : { issueKinds: error.issueKinds }),
261 ...(error.maximumPathDepth === undefined ? {} : { maximumPathDepth: error.maximumPathDepth }),
262 ...(error.issueSummary === undefined ? {} : { issueSummary: error.issueSummary }),
263});
264const encodeJsonRpcNotification = Schema.encodeUnknownExit(
265 Schema.fromJsonString(
266 Schema.Struct({
267 jsonrpc: Schema.Literal("2.0"),
268 method: Schema.String,
269 params: Schema.Unknown,
270 }),
271 ),
272);
273
274/**
275 * First id for typed Effect RPC requests. Starts within signed int32 for SDKs that reject
276 * larger numeric ids (e.g. the Kotlin ACP SDK decodes ids as `Int`), while staying far
277 * above extension ids, which count up from 1.
278 */
279export const RPC_REQUEST_ID_START = 2 ** 30;
280
281const isEffectRpcRequestId = (requestId: AcpError.AcpRequestId): boolean =>
282 typeof requestId === "number" && Number.isSafeInteger(requestId);
283
284/**
285 * Effect RPC's JSON-RPC codec treats a standard JSON-RPC error object as a
286 * defect unless it carries Effect's private `_tag: "Cause"` envelope.
287 */
288function normalizeAcpJsonRpcError(
289 message: RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded,
290): RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded {
291 if (message._tag !== "Exit" || message.exit._tag !== "Failure") return message;
292 const [failure] = message.exit.cause;
293 if (
294 message.exit.cause.length !== 1 ||
295 failure?._tag !== "Die" ||
296 !isProtocolError(failure.defect)
297 ) {
298 return message;
299 }
300 return {
301 ...message,
302 exit: {
303 _tag: "Failure",
304 cause: [{ _tag: "Fail", error: failure.defect }],
305 },
306 };
307}
308
309export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(function* (
310 options: AcpPatchedProtocolOptions,
311): Effect.fn.Return<AcpPatchedProtocol, never, Scope.Scope> {
312 const parser = makeStrictNdJsonRpcParser();
313 const serverQueue = yield* Queue.unbounded<RpcMessage.FromClientEncoded>();
314 const clientQueue = yield* Queue.unbounded<RpcMessage.FromServerEncoded>();
315 const rawNotifications = yield* Queue.unbounded<AcpBufferedRawNotification, Cause.Done<void>>();
316 // JSON length of the notifications offered since the queue was last empty.
317 // `incoming` takes the whole backlog on every pull, so an empty queue holds
318 // nothing, and a read can only make this an overestimate.
319 let rawNotificationsJsonLength = 0;
320 // Raw readers finish with the connection: termination ends the queue, and so
321 // does closing the scope, which interrupts the reader before it terminates.
322 yield* Effect.addFinalizer(() => Queue.end(rawNotifications));
323 const disconnects = yield* Queue.unbounded<number>();
324 const outgoing = yield* Queue.unbounded<AcpOutgoingWrite, AcpError.AcpError | Cause.Done<void>>();
325 const outgoingWriterState = yield* Ref.make<AcpOutgoingWriterState>({
326 intentionalEnd: false,
327 outstandingAcknowledgements: new Set(),
328 });
329 const nextRequestId = yield* Ref.make(1);
330 const warnedDecodeFailures = new Set<string>();
331 const terminationHandled = yield* Ref.make(false);
332 const terminationFailure = yield* Deferred.make<never, AcpError.AcpError>();
333 const extPending = yield* Ref.make(new Map<string, AcpPendingRequest>());
334
335 const ensureActive = Ref.get(terminationHandled).pipe(
336 Effect.flatMap((terminated) => (terminated ? Deferred.await(terminationFailure) : Effect.void)),
337 );
338
339 const logProtocol = (event: AcpProtocolLogEvent) => {
340 if (event.direction === "incoming" && !options.logIncoming) {
341 return Effect.void;
342 }
343 if (event.direction === "outgoing" && !options.logOutgoing) {
344 return Effect.void;
345 }
346 return (
347 options.logger?.(event) ??
348 Effect.logDebug("ACP protocol event").pipe(Effect.annotateLogs({ event }))
349 );
350 };
351
352 const offerOutgoing = Effect.fn("offerOutgoing")(function* (
353 message: RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded,
354 ) {
355 // RpcClient emits `@effect/rpc/Interrupt` when a pending request's fiber is interrupted.
356 // ACP has no such method; agents log it as an error and cannot act on it, so drop it.
357 if (message._tag === "Interrupt") {
358 return;
359 }
360 yield* ensureActive;
361 yield* logProtocol({
362 direction: "outgoing",
363 stage: "decoded",
364 payload: message,
365 });
366
367 const method = message._tag === "Request" ? message.tag : undefined;
368 const encodedRequestId =
369 message._tag === "Request"
370 ? message.id
371 : "requestId" in message
372 ? message.requestId
373 : undefined;
374 const requestId = encodedRequestId === "" ? undefined : encodedRequestId;
375 const encoded = yield* Effect.try({
376 try: () => parser.encode(message),
377 catch: (cause) => AcpError.AcpProtocolParseError.fromEncodingError(method, requestId, cause),
378 });
379
380 if (encoded) {
381 yield* logProtocol({
382 direction: "outgoing",
383 stage: "raw",
384 payload: typeof encoded === "string" ? encoded : new TextDecoder().decode(encoded),
385 });
386
387 yield* ensureActive;
388 const acknowledgement =
389 message._tag === "Exit" ? yield* Deferred.make<void, AcpError.AcpError>() : undefined;
390 const admissionError = yield* Ref.modify(outgoingWriterState, (state) => {
391 if (state.terminalError !== undefined) {
392 return [state.terminalError, state] as const;
393 }
394 if (acknowledgement === undefined) {
395 return [undefined, state] as const;
396 }
397 return [
398 undefined,
399 {
400 ...state,
401 outstandingAcknowledgements: new Set(state.outstandingAcknowledgements).add(
402 acknowledgement,
403 ),
404 },
405 ] as const;
406 });
407 if (admissionError !== undefined) {
408 return yield* admissionError;
409 }
410 if (options.testHooks?.onOutgoingWriteAdmitted !== undefined) {
411 yield* options.testHooks.onOutgoingWriteAdmitted();
412 }
413 yield* Queue.offer(outgoing, {
414 payload: encoded,
415 ...(acknowledgement === undefined ? {} : { acknowledgement }),
416 }).pipe(
417 Effect.mapError(
418 (cause) =>
419 new AcpError.AcpTransportError({
420 detail: "Failed to queue an outgoing ACP message",
421 cause,
422 }),
423 ),
424 Effect.filterOrFail(
425 (accepted) => accepted,
426 () =>
427 new AcpError.AcpTransportError({
428 detail: "The ACP output queue was closed before accepting a message",
429 cause: "Output queue closed",
430 }),
431 ),
432 Effect.tapError((error) =>
433 acknowledgement === undefined
434 ? Effect.void
435 : Ref.update(outgoingWriterState, (state) => {
436 const updated = new Set(state.outstandingAcknowledgements);
437 updated.delete(acknowledgement);
438 return { ...state, outstandingAcknowledgements: updated };
439 }).pipe(Effect.andThen(Deferred.fail(acknowledgement, error)), Effect.asVoid),
440 ),
441 );
442 if (acknowledgement !== undefined) {
443 yield* Deferred.await(acknowledgement);
444 }
445 if (message._tag === "Exit" && options.onOutgoingResponse !== undefined) {
446 yield* options.onOutgoingResponse(acpRequestIdentity(message.requestId));
447 }
448 }
449 });
450
451 const resolveExtPending = (
452 requestId: AcpError.AcpRequestId,
453 onFound: (pendingRequest: AcpPendingRequest) => Effect.Effect<void>,
454 ) =>
455 Ref.modify(extPending, (pending) => {
456 const pendingKey = acpRequestIdentity(requestId);
457 const pendingRequest = pending.get(pendingKey);
458 if (!pendingRequest) {
459 return [Effect.void, pending] as const;
460 }
461 const next = new Map(pending);
462 next.delete(pendingKey);
463 return [onFound(pendingRequest), next] as const;
464 }).pipe(Effect.flatten);
465
466 const removeExtPending = (requestId: AcpError.AcpRequestId) =>
467 Ref.update(extPending, (pending) => {
468 const pendingKey = acpRequestIdentity(requestId);
469 if (!pending.has(pendingKey)) {
470 return pending;
471 }
472 const next = new Map(pending);
473 next.delete(pendingKey);
474 return next;
475 });
476
477 const completeExtPendingFailure = (requestId: AcpError.AcpRequestId, error: AcpError.AcpError) =>
478 resolveExtPending(requestId, ({ deferred }) => Deferred.fail(deferred, error));
479
480 const completeExtPendingSuccess = (requestId: AcpError.AcpRequestId, value: unknown) =>
481 resolveExtPending(requestId, ({ deferred }) => Deferred.succeed(deferred, value));
482
483 const failAllExtPending = (error: AcpError.AcpError) =>
484 Ref.getAndSet(extPending, new Map()).pipe(
485 Effect.flatMap((pending) =>
486 Effect.forEach([...pending.values()], ({ deferred }) => Deferred.fail(deferred, error), {
487 discard: true,
488 }),
489 ),
490 );
491
492 // Offer first so an ended queue rejects the notification without evicting the
493 // backlog readers are still draining. Offer and eviction share one sync block,
494 // so readers never see more than the limits allow.
495 const bufferRawNotification = (notification: AcpIncomingNotification, lineLength: number) =>
496 Effect.sync(() => {
497 if (Queue.sizeUnsafe(rawNotifications) === 0) rawNotificationsJsonLength = 0;
498 if (!Queue.offerUnsafe(rawNotifications, { notification, lineLength })) return;
499 rawNotificationsJsonLength += lineLength;
500 while (
501 Queue.sizeUnsafe(rawNotifications) > MAX_BUFFERED_RAW_NOTIFICATIONS ||
502 (Queue.sizeUnsafe(rawNotifications) > 1 &&
503 rawNotificationsJsonLength > MAX_BUFFERED_RAW_NOTIFICATION_JSON_LENGTH)
504 ) {
505 const dropped = Queue.takeUnsafe(rawNotifications);
506 if (dropped === undefined || Exit.isFailure(dropped)) break;
507 rawNotificationsJsonLength -= dropped.value.lineLength;
508 }
509 });
510
511 const dispatchNotification = (notification: AcpIncomingNotification, lineLength: number) =>
512 bufferRawNotification(notification, lineLength).pipe(
513 Effect.andThen(
514 // A failing or dying handler must not stop the reader, or every later
515 // message on the connection goes unanswered.
516 options.onNotification
517 ? isolateNotificationHandler(options.onNotification(notification))
518 : Effect.void,
519 ),
520 Effect.asVoid,
521 );
522
523 const emitClientProtocolError = (error: AcpError.AcpError) =>
524 Queue.offer(clientQueue, {
525 _tag: "ClientProtocolError",
526 error: new RpcClientError.RpcClientError({
527 reason: new RpcClientError.RpcClientDefect({
528 message: "ACP protocol terminated.",
529 cause: error,
530 }),
531 }),
532 }).pipe(Effect.asVoid);
533
534 const handleTermination = (classify: () => Effect.Effect<AcpError.AcpError>) =>
535 Ref.modify(terminationHandled, (handled) => {
536 if (handled) {
537 return [Effect.void, true] as const;
538 }
539 return [
540 Effect.gen(function* () {
541 yield* Queue.offer(disconnects, 0);
542 yield* Queue.end(rawNotifications);
543 const error = yield* classify();
544 yield* Deferred.fail(terminationFailure, error);
545 yield* failAllExtPending(error);
546 yield* emitClientProtocolError(error);
547 if (options.onTermination) {
548 yield* options.onTermination(error);
549 }
550 }),
551 true,
552 ] as const;
553 }).pipe(Effect.flatten);
554
555 const respondWithSuccess = (requestId: AcpError.AcpRequestId, value: unknown) =>
556 offerOutgoing({
557 _tag: "Exit",
558 requestId,
559 exit: {
560 _tag: "Success",
561 value,
562 },
563 });
564
565 const respondWithError = (requestId: AcpError.AcpRequestId, error: AcpError.AcpRequestError) =>
566 offerOutgoing({
567 _tag: "Exit",
568 requestId,
569 exit: {
570 _tag: "Failure",
571 cause: [
572 {
573 _tag: "Fail",
574 error: error.toProtocolError(),
575 },
576 ],
577 },
578 });
579
580 const handleExtRequest = (message: RpcMessage.RequestEncoded) => {
581 if (!options.onExtRequest) {
582 return respondWithError(message.id, AcpError.AcpRequestError.methodNotFound(message.tag));
583 }
584 return options
585 .onExtRequest(message.tag, message.payload, {
586 requestId: acpRequestIdentity(message.id),
587 method: message.tag,
588 })
589 .pipe(
590 Effect.matchCauseEffect({
591 onFailure: (cause) => {
592 const failure = Cause.hasDies(cause) ? Option.none() : Cause.findErrorOption(cause);
593 if (Option.isSome(failure)) {
594 return respondWithError(
595 message.id,
596 AcpError.AcpRequestError.fromExtensionHandlerError(failure.value, message.tag),
597 );
598 }
599 return Effect.logError(
600 `ACP extension request handler failed for '${message.tag}'`,
601 cause,
602 ).pipe(
603 Effect.andThen(
604 respondWithError(
605 message.id,
606 AcpError.AcpRequestError.internalError(
607 `ACP extension request handler failed for method '${message.tag}'`,
608 undefined,
609 {
610 method: message.tag,
611 operation: "handle-extension-request",
612 cause: Cause.squash(cause),
613 },
614 ),
615 ),
616 ),
617 );
618 },
619 onSuccess: (value) => respondWithSuccess(message.id, value),
620 }),
621 );
622 };
623
624 // Drop undecodable notifications without ending working sessions; newer specs may add unknown shapes.
625 // Warn once per distinct failure per connection because agents repeat bad payloads.
626 const dropUndecodableNotification = (method: string) => (cause: Schema.SchemaError) => {
627 const error = AcpError.AcpProtocolParseError.fromSchemaError(
628 "decode-notification-payload",
629 method,
630 cause,
631 );
632 const diagnostics = decodeFailureDiagnostics(error);
633 const key = `${method}\n${error.issueSummary}`;
634 const firstOfItsKind = !warnedDecodeFailures.has(key);
635 warnedDecodeFailures.add(key);
636 return logProtocol({
637 direction: "incoming",
638 stage: "decode_failed",
639 payload: diagnostics,
640 }).pipe(
641 Effect.andThen(
642 firstOfItsKind
643 ? Effect.logWarning("Dropped an ACP notification that failed to decode").pipe(
644 Effect.annotateLogs(diagnostics),
645 )
646 : Effect.void,
647 ),
648 );
649 };
650
651 const handleRequestEncoded = (message: RpcMessage.RequestEncoded, lineLength: number) => {
652 if (message.id === "") {
653 if (message.tag === CLIENT_METHODS.session_update) {
654 return decodeSessionUpdate(message.payload).pipe(
655 Effect.matchEffect({
656 onFailure: dropUndecodableNotification(CLIENT_METHODS.session_update),
657 onSuccess: (params) =>
658 dispatchNotification(
659 {
660 _tag: "SessionUpdate",
661 method: CLIENT_METHODS.session_update,
662 params: options.transformSessionUpdate?.(params) ?? params,
663 },
664 lineLength,
665 ),
666 }),
667 );
668 }
669 if (
670 message.tag === CLIENT_METHODS.elicitation_complete ||
671 message.tag === "session/elicitation/complete"
672 ) {
673 const method = message.tag;
674 return decodeElicitationComplete(message.payload).pipe(
675 Effect.matchEffect({
676 onFailure: dropUndecodableNotification(method),
677 onSuccess: (params) =>
678 dispatchNotification({ _tag: "ElicitationComplete", method, params }, lineLength),
679 }),
680 );
681 }
682 return dispatchNotification(
683 { _tag: "ExtNotification", method: message.tag, params: message.payload },
684 lineLength,
685 );
686 }
687
688 const observeIncoming =
689 options.onIncomingRequest?.(acpRequestIdentity(message.id), message.tag, message.payload) ??
690 Effect.void;
691
692 if (!options.serverRequestMethods.has(message.tag)) {
693 return observeIncoming.pipe(
694 Effect.andThen(handleExtRequest(message)),
695 Effect.catchTags({
696 AcpProtocolParseError: (error) =>
697 Effect.logWarning(error).pipe(
698 Effect.annotateLogs({
699 method: message.tag,
700 requestId: message.id,
701 operation: error.operation,
702 }),
703 Effect.andThen(
704 respondWithError(
705 message.id,
706 AcpError.AcpRequestError.fromExtensionResponseEncodingError(
707 message.tag,
708 message.id,
709 error,
710 ),
711 ),
712 ),
713 ),
714 }),
715 Effect.asVoid,
716 );
717 }
718
719 return observeIncoming.pipe(Effect.andThen(Queue.offer(serverQueue, message)), Effect.asVoid);
720 };
721
722 const forwardToRpcClient = (
723 message: RpcMessage.ResponseChunkEncoded | RpcMessage.ResponseExitEncoded,
724 ) =>
725 isEffectRpcRequestId(message.requestId)
726 ? Queue.offer(clientQueue, message).pipe(Effect.asVoid)
727 : Effect.void;
728
729 const handleExitEncoded = (message: RpcMessage.ResponseExitEncoded) =>
730 Ref.get(extPending).pipe(
731 Effect.flatMap((pending) => {
732 const pendingRequest = pending.get(acpRequestIdentity(message.requestId));
733 if (!pendingRequest) {
734 return forwardToRpcClient(message);
735 }
736 if (message.exit._tag === "Success") {
737 return completeExtPendingSuccess(message.requestId, message.exit.value);
738 }
739 const failure = message.exit.cause.find((entry) => entry._tag === "Fail");
740 if (failure && isProtocolError(failure.error)) {
741 return completeExtPendingFailure(
742 message.requestId,
743 AcpError.AcpRequestError.fromProtocolError(failure.error, {
744 method: pendingRequest.method,
745 requestId: message.requestId,
746 cause: message.exit.cause,
747 }),
748 );
749 }
750 return completeExtPendingFailure(
751 message.requestId,
752 AcpError.AcpRequestError.fromExtensionResponseFailure(
753 pendingRequest.method,
754 message.requestId,
755 message.exit.cause,
756 ),
757 );
758 }),
759 );
760
761 const routeDecodedMessage = ({
762 message,
763 lineLength,
764 }: AcpDecodedWireMessage): Effect.Effect<void, AcpError.AcpError> => {
765 switch (message._tag) {
766 case "Request":
767 return handleRequestEncoded(message, lineLength);
768 case "Exit":
769 return handleExitEncoded(message);
770 case "Chunk":
771 return Ref.get(extPending).pipe(
772 Effect.flatMap((pending) => {
773 const pendingRequest = pending.get(acpRequestIdentity(message.requestId));
774 return pendingRequest
775 ? completeExtPendingFailure(
776 message.requestId,
777 AcpError.AcpRequestError.unsupportedStreamingResponse(
778 pendingRequest.method,
779 message.requestId,
780 ),
781 )
782 : forwardToRpcClient(message);
783 }),
784 );
785 case "Defect":
786 case "ClientProtocolError":
787 case "Pong":
788 return Queue.offer(clientQueue, message).pipe(Effect.asVoid);
789 case "Ack":
790 case "Interrupt":
791 case "Ping":
792 case "Eof":
793 return Queue.offer(serverQueue, message).pipe(Effect.asVoid);
794 }
795 };
796
797 yield* options.stdio.stdin.pipe(
798 Stream.runForEach((data) =>
799 (options.logIncoming
800 ? logProtocol({
801 direction: "incoming",
802 stage: "raw",
803 payload: typeof data === "string" ? data : new TextDecoder().decode(data),
804 })
805 : Effect.void
806 ).pipe(
807 Effect.flatMap(() =>
808 Effect.try({
809 try: () => parser.decode(data),
810 catch: (cause) =>
811 new AcpError.AcpProtocolParseError({
812 operation: "decode-wire-message",
813 cause,
814 }),
815 }),
816 ),
817 Effect.tap((decoded) =>
818 options.logIncoming
819 ? logProtocol({
820 direction: "incoming",
821 stage: "decoded",
822 payload: decoded.map(({ message }) => message),
823 })
824 : Effect.void,
825 ),
826 Effect.tapErrorTag("AcpProtocolParseError", (error) =>
827 logProtocol({
828 direction: "incoming",
829 stage: "decode_failed",
830 payload: decodeFailureDiagnostics(error),
831 }),
832 ),
833 Effect.flatMap((decoded) =>
834 Effect.forEach(decoded, routeDecodedMessage, {
835 discard: true,
836 }),
837 ),
838 ),
839 ),
840 Effect.andThen(
841 Effect.suspend(() => Effect.forEach(parser.end(), routeDecodedMessage, { discard: true })),
842 ),
843 // Anything that ends the reader, including a defect in a callback it runs,
844 // terminates the connection so pending requests fail instead of hanging.
845 Effect.matchCauseEffect({
846 onFailure: (cause) => {
847 const failure = Cause.findErrorOption(cause);
848 const error = Option.isSome(failure) ? failure.value : Cause.squash(cause);
849 const normalized: AcpError.AcpError = isAcpError(error)
850 ? error
851 : new AcpError.AcpTransportError({
852 operation: "read-input-stream",
853 cause: error,
854 });
855 return handleTermination(() => Effect.succeed(normalized));
856 },
857 onSuccess: () =>
858 handleTermination(() => {
859 const ended = new AcpError.AcpInputStreamEndedError({});
860 if (!options.terminationError) return Effect.succeed(ended);
861 // Stdout can close without process exit; use a live-clock diagnostic grace period.
862 return options.terminationError.pipe(
863 Effect.timeoutOption("250 millis"),
864 Effect.provideService(Clock.Clock, Clock.Clock.defaultValue()),
865 Effect.map(Option.getOrElse(() => ended)),
866 );
867 }),
868 }),
869 Effect.forkScoped,
870 );
871
872 const failOutgoingWrites = (error: AcpError.AcpError) =>
873 Ref.modify(outgoingWriterState, (state) => {
874 const terminalError = state.terminalError ?? error;
875 return [
876 [terminalError, state.outstandingAcknowledgements] as const,
877 {
878 ...state,
879 terminalError,
880 outstandingAcknowledgements: new Set<Deferred.Deferred<void, AcpError.AcpError>>(),
881 },
882 ] as const;
883 }).pipe(
884 Effect.flatMap((acknowledgements) =>
885 Effect.forEach(
886 acknowledgements[1],
887 (acknowledgement) => Deferred.fail(acknowledgement, acknowledgements[0]),
888 { concurrency: "unbounded", discard: true },
889 ),
890 ),
891 Effect.andThen(Queue.fail(outgoing, error)),
892 Effect.asVoid,
893 );
894 const completeOutgoingWrites = () =>
895 Ref.modify(outgoingWriterState, (state) => {
896 return [
897 state.outstandingAcknowledgements,
898 {
899 ...state,
900 outstandingAcknowledgements: new Set<Deferred.Deferred<void, AcpError.AcpError>>(),
901 },
902 ] as const;
903 }).pipe(
904 Effect.flatMap((acknowledgements) =>
905 Effect.forEach(
906 acknowledgements,
907 (acknowledgement) => Deferred.succeed(acknowledgement, undefined),
908 { concurrency: "unbounded", discard: true },
909 ),
910 ),
911 Effect.asVoid,
912 );
913 const acknowledgeOutgoingWrite = (acknowledgement: Deferred.Deferred<void, AcpError.AcpError>) =>
914 Ref.update(outgoingWriterState, (state) => {
915 const updated = new Set(state.outstandingAcknowledgements);
916 updated.delete(acknowledgement);
917 return { ...state, outstandingAcknowledgements: updated };
918 }).pipe(Effect.andThen(Deferred.succeed(acknowledgement, undefined)), Effect.asVoid);
919 yield* Stream.fromQueue(outgoing).pipe(
920 Stream.flatMap((write) => {
921 const acknowledgement = write.acknowledgement;
922 const completion =
923 acknowledgement === undefined
924 ? Stream.empty
925 : Stream.fromEffect(acknowledgeOutgoingWrite(acknowledgement)).pipe(Stream.drain);
926 return Stream.make(write.payload).pipe(Stream.concat(completion));
927 }),
928 Stream.run(options.stdio.stdout()),
929 Effect.onExit((exit) =>
930 Exit.match(exit, {
931 onFailure: (cause) => {
932 const error = new AcpError.AcpTransportError({
933 detail: Cause.hasInterruptsOnly(cause)
934 ? "The ACP output writer was interrupted while closing"
935 : "Failed to write an outgoing ACP message",
936 cause,
937 });
938 return failOutgoingWrites(error).pipe(
939 Effect.andThen(
940 Cause.hasInterruptsOnly(cause)
941 ? Effect.void
942 : handleTermination(() => Effect.succeed(error)),
943 ),
944 );
945 },
946 onSuccess: () =>
947 Ref.get(outgoingWriterState).pipe(
948 Effect.flatMap((state) => {
949 const handled = state.intentionalEnd
950 ? completeOutgoingWrites()
951 : (() => {
952 const error = new AcpError.AcpTransportError({
953 detail: "ACP output writer ended before the protocol closed",
954 cause: "Output writer ended",
955 });
956 return failOutgoingWrites(error).pipe(
957 Effect.andThen(handleTermination(() => Effect.succeed(error))),
958 );
959 })();
960 return handled.pipe(
961 Effect.andThen(
962 options.testHooks?.onOutgoingWriterExit === undefined
963 ? Effect.void
964 : options.testHooks.onOutgoingWriterExit({
965 intentionalEnd: state.intentionalEnd,
966 }),
967 ),
968 );
969 }),
970 ),
971 }),
972 ),
973 Effect.forkScoped,
974 );
975
976 // Protocol loops and per-message sends share the caller's operation span.
977 const clientProtocol = RpcClient.Protocol.of({
978 run: Effect.fnUntraced(function* (_clientId, f) {
979 return yield* Stream.fromQueue(clientQueue).pipe(
980 Stream.runForEach((message) => f(message)),
981 Effect.forever,
982 );
983 }),
984 send: Effect.fnUntraced(function* (_clientId, request) {
985 return yield* offerOutgoing(request).pipe(
986 Effect.mapError(
987 (error) =>
988 new RpcClientError.RpcClientError({
989 reason: new RpcClientError.RpcClientDefect({
990 message: "Failed to send ACP protocol message.",
991 cause: error,
992 }),
993 }),
994 ),
995 );
996 }),
997 supportsAck: true,
998 supportsTransferables: false,
999 codecFor: parserFactory.codecFor,
1000 });
1001
1002 const serverProtocol = RpcServer.Protocol.of({
1003 run: Effect.fnUntraced(function* (f) {
1004 return yield* Stream.fromQueue(serverQueue).pipe(
1005 Stream.runForEach((message) => f(0, message)),
1006 Effect.forever,
1007 );
1008 }),
1009 disconnects,
1010 send: Effect.fnUntraced(function* (_clientId, response) {
1011 return yield* offerOutgoing(response).pipe(
1012 Effect.tapError((error) =>
1013 response._tag === "Exit" && options.onOutgoingResponseFailure !== undefined
1014 ? options.onOutgoingResponseFailure(acpRequestIdentity(response.requestId), error)
1015 : Effect.void,
1016 ),
1017 Effect.orDie,
1018 );
1019 }),
1020 end: Effect.fn("RpcServer.Protocol.end")(function (_clientId) {
1021 return Ref.update(outgoingWriterState, (state) => ({ ...state, intentionalEnd: true })).pipe(
1022 Effect.andThen(Queue.end(outgoing)),
1023 );
1024 }),
1025 clientIds: Effect.succeed(new Set([0])),
1026 initialMessage: Effect.succeedNone,
1027 supportsAck: true,
1028 supportsTransferables: false,
1029 codecFor: parserFactory.codecFor,
1030 supportsSpanPropagation: true,
1031 supportsNotifications: true,
1032 });
1033
1034 // JSON-RPC notifications carry no `id`. Encoding a Request without `isNotification`
1035 // emits an `id`, which real agents (Grok CLI) parse as a malformed request and silently drop.
1036 const sendNotification = Effect.fn("sendNotification")(function* (
1037 method: string,
1038 payload: unknown,
1039 ) {
1040 yield* ensureActive;
1041 yield* logProtocol({
1042 direction: "outgoing",
1043 stage: "decoded",
1044 payload: { _tag: "Notification", tag: method, payload },
1045 });
1046 const exit = encodeJsonRpcNotification({ jsonrpc: "2.0", method, params: payload });
1047 if (Exit.isFailure(exit)) {
1048 return yield* AcpError.AcpProtocolParseError.fromEncodingError(
1049 method,
1050 undefined,
1051 Cause.squash(exit.cause),
1052 );
1053 }
1054 const encoded = `${exit.value}\n`;
1055 yield* logProtocol({ direction: "outgoing", stage: "raw", payload: encoded });
1056 yield* ensureActive;
1057 // FIFO queue admission preserves wire ordering against later requests
1058 // (a session/prompt sent after session/cancel cannot overtake it).
1059 yield* Queue.offer(outgoing, { payload: encoded }).pipe(
1060 Effect.mapError(
1061 (cause) =>
1062 new AcpError.AcpTransportError({
1063 detail: "Failed to queue an outgoing ACP notification",
1064 cause,
1065 }),
1066 ),
1067 Effect.filterOrFail(
1068 (accepted) => accepted,
1069 () =>
1070 new AcpError.AcpTransportError({
1071 detail: "The ACP output queue was closed before accepting a notification",
1072 cause: "Output queue closed",
1073 }),
1074 ),
1075 );
1076 });
1077
1078 const sendRequest = Effect.fn("sendRequest")(function* (method: string, payload: unknown) {
1079 yield* ensureActive;
1080 const requestId = yield* Ref.modify(
1081 nextRequestId,
1082 (current) => [current, current + 1] as const,
1083 );
1084 const deferred = yield* Deferred.make<unknown, AcpError.AcpError>();
1085 yield* Ref.update(extPending, (pending) =>
1086 new Map(pending).set(acpRequestIdentity(requestId), { deferred, method }),
1087 );
1088 yield* offerOutgoing({
1089 _tag: "Request",
1090 id: requestId,
1091 tag: method,
1092 payload,
1093 headers: [],
1094 }).pipe(Effect.tapError(() => removeExtPending(requestId)));
1095 return yield* Deferred.await(deferred).pipe(
1096 Effect.onInterrupt(() => removeExtPending(requestId)),
1097 );
1098 });
1099
1100 return {
1101 clientProtocol,
1102 serverProtocol,
1103 get incoming() {
1104 return Stream.fromQueue(rawNotifications).pipe(
1105 Stream.map(({ notification }) => notification),
1106 );
1107 },
1108 request: sendRequest,
1109 notify: sendNotification,
1110 } satisfies AcpPatchedProtocol;
1111});
1112
1113const isProtocolError = Schema.is(AcpSchema.Error);