1import * as Path from "effect/Path";
2import * as AcpError from "./errors.ts";
3import * as Effect from "effect/Effect";
4import * as Deferred from "effect/Deferred";
5import * as Fiber from "effect/Fiber";
6import * as Exit from "effect/Exit";
7import * as Queue from "effect/Queue";
8import * as Schema from "effect/Schema";
9import * as Scope from "effect/Scope";
10import * as Sink from "effect/Sink";
11import * as Stdio from "effect/Stdio";
12import * as Stream from "effect/Stream";
13import * as Ref from "effect/Ref";
14import * as TestClock from "effect/testing/TestClock";
15import { ChildProcess, ChildProcessSpawner } from "effect/process";
16
17import { it, assert } from "@effect/vitest";
18import * as NodeServices from "@effect/platform-node/NodeServices";
19
20import * as AcpSchema from "./schema.ts";
21import * as AcpProtocol from "./protocol.ts";
22import {
23 encodeJsonl,
24 jsonRpcNotification,
25 jsonRpcRequest,
26 jsonRpcResponse,
27} from "./_internal/shared.ts";
28import { makeInMemoryStdio, makeTerminationError, makeChildStdio } from "./_internal/stdio.ts";
29
30const SessionCancelNotification = jsonRpcNotification(
31 "session/cancel",
32 AcpSchema.CancelNotification,
33);
34const SessionUpdateNotification = jsonRpcNotification(
35 "session/update",
36 AcpSchema.SessionNotification,
37);
38const ElicitationCompleteNotification = jsonRpcNotification(
39 "elicitation/complete",
40 AcpSchema.CompleteElicitationNotification,
41);
42const RequestPermissionRequest = jsonRpcRequest(
43 "session/request_permission",
44 AcpSchema.RequestPermissionRequest,
45);
46const RequestPermissionResponse = jsonRpcResponse(AcpSchema.RequestPermissionResponse);
47const ExtRequest = jsonRpcRequest("x/test", Schema.Struct({ hello: Schema.String }));
48const ExtResponse = jsonRpcResponse(Schema.Struct({ ok: Schema.Boolean }));
49const decodeSessionCancelNotification = Schema.decodeEffect(
50 Schema.fromJsonString(SessionCancelNotification),
51);
52const decodeExtRequest = Schema.decodeEffect(Schema.fromJsonString(ExtRequest));
53const decodeExtResponse = Schema.decodeEffect(Schema.fromJsonString(ExtResponse));
54const decodeRequestPermissionResponse = Schema.decodeEffect(
55 Schema.fromJsonString(RequestPermissionResponse),
56);
57const decodeJsonRpcRequestId = Schema.decodeEffect(
58 Schema.fromJsonString(Schema.Struct({ id: Schema.Number })),
59);
60const JsonRpcErrorResponse = Schema.Struct({
61 jsonrpc: Schema.Literal("2.0"),
62 id: Schema.Number,
63 error: Schema.Struct({ code: Schema.Number, message: Schema.String }),
64});
65const encodeUnknownJsonString = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
66const encoder = new TextEncoder();
67const mockPeerPath = Effect.map(Effect.service(Path.Path), (path) =>
68 path.join(import.meta.dirname, "../test/fixtures/acp-mock-peer.ts"),
69);
70const mockPeerArgs = (path: string) => [path];
71
72const makeHandle = (env?: Record<string, string>) =>
73 Effect.gen(function* () {
74 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
75 const path = yield* Path.Path;
76 const command = ChildProcess.make(process.execPath, mockPeerArgs(yield* mockPeerPath), {
77 cwd: path.join(import.meta.dirname, ".."),
78 ...(env ? { env: { ...process.env, ...env } } : {}),
79 });
80 return yield* spawner.spawn(command);
81 });
82
83it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
84 it.effect(
85 "emits exact JSON-RPC notifications and decodes inbound session/update and elicitation completion",
86 () =>
87 Effect.gen(function* () {
88 const { stdio, input, output } = yield* makeInMemoryStdio();
89 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
90 stdio,
91 serverRequestMethods: new Set(),
92 });
93
94 const notifications =
95 yield* Deferred.make<ReadonlyArray<AcpProtocol.AcpIncomingNotification>>();
96 yield* transport.incoming.pipe(
97 Stream.take(2),
98 Stream.runCollect,
99 Effect.flatMap((notificationChunk) => Deferred.succeed(notifications, notificationChunk)),
100 Effect.forkScoped,
101 );
102
103 yield* transport.notify("session/cancel", { sessionId: "session-1" });
104 const outbound = yield* Queue.take(output);
105 assert.deepEqual(yield* decodeSessionCancelNotification(outbound), {
106 jsonrpc: "2.0",
107 method: "session/cancel",
108 params: {
109 sessionId: "session-1",
110 },
111 });
112
113 yield* Queue.offer(
114 input,
115 yield* encodeJsonl(SessionUpdateNotification, {
116 jsonrpc: "2.0",
117 method: "session/update",
118 params: {
119 sessionId: "session-1",
120 update: {
121 sessionUpdate: "plan",
122 entries: [
123 {
124 content: "Inspect repository",
125 priority: "high",
126 status: "in_progress",
127 },
128 ],
129 },
130 },
131 }),
132 );
133
134 yield* Queue.offer(
135 input,
136 yield* encodeJsonl(ElicitationCompleteNotification, {
137 jsonrpc: "2.0",
138 method: "elicitation/complete",
139 params: {
140 elicitationId: "elicitation-1",
141 },
142 }),
143 );
144
145 const [update, completion] = yield* Deferred.await(notifications);
146 assert.equal(update?._tag, "SessionUpdate");
147 assert.equal(completion?._tag, "ElicitationComplete");
148 }),
149 );
150
151 it.effect("keeps only recent raw notifications after their callbacks run", () =>
152 Effect.gen(function* () {
153 const { stdio, input } = yield* makeInMemoryStdio();
154 const handled = yield* Deferred.make<void>();
155 let handledCount = 0;
156 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
157 stdio,
158 serverRequestMethods: new Set(),
159 onNotification: () =>
160 Effect.sync(() => ++handledCount).pipe(
161 Effect.flatMap((count) =>
162 count === 64 ? Deferred.succeed(handled, undefined).pipe(Effect.asVoid) : Effect.void,
163 ),
164 ),
165 });
166
167 const messages = Array.from({ length: 64 }, (_, index) =>
168 encodeUnknownJsonString({
169 jsonrpc: "2.0",
170 method: "x/performance",
171 params: { index },
172 }),
173 );
174 yield* Queue.offer(input, encoder.encode(`${messages.join("\n")}\n`));
175 yield* Deferred.await(handled);
176
177 const retained = yield* transport.incoming.pipe(Stream.take(32), Stream.runCollect);
178
179 assert.equal(handledCount, 64);
180 assert.equal(retained.length, 32);
181 assert.deepEqual(retained[0]?.params, { index: 32 });
182 assert.deepEqual(retained[31]?.params, { index: 63 });
183 }),
184 );
185
186 it.effect("decodes standard JSON-RPC errors into typed ACP request failures", () =>
187 Effect.gen(function* () {
188 const { stdio, input, output } = yield* makeInMemoryStdio();
189 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
190 stdio,
191 serverRequestMethods: new Set(),
192 });
193
194 const request = yield* transport.request("x/auth", {}).pipe(Effect.forkScoped);
195 const outbound = yield* decodeJsonRpcRequestId(yield* Queue.take(output));
196 yield* Queue.offer(
197 input,
198 yield* encodeJsonl(JsonRpcErrorResponse, {
199 jsonrpc: "2.0",
200 id: outbound.id,
201 error: { code: -32000, message: "Authentication required" },
202 }),
203 );
204
205 const error = yield* Fiber.join(request).pipe(Effect.flip, Effect.orDie);
206 assert.instanceOf(error, AcpError.AcpRequestError);
207 assert.equal(error.code, -32000);
208 assert.equal(error.message, "Authentication required");
209 }),
210 );
211
212 it.effect("keeps invalid core notification values only in the schema cause", () =>
213 Effect.gen(function* () {
214 const secret = "acp-core-notification-secret-sentinel";
215 const { stdio, input } = yield* makeInMemoryStdio();
216 const termination = yield* Deferred.make<AcpError.AcpError>();
217 let transformCalls = 0;
218 yield* AcpProtocol.makeAcpPatchedProtocol({
219 stdio,
220 serverRequestMethods: new Set(),
221 transformSessionUpdate: (notification) => {
222 transformCalls += 1;
223 return notification;
224 },
225 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
226 });
227
228 yield* Queue.offer(
229 input,
230 encoder.encode(
231 `${encodeUnknownJsonString({
232 jsonrpc: "2.0",
233 method: "session/update",
234 params: {
235 sessionId: { secret },
236 update: {
237 sessionUpdate: "plan",
238 entries: [],
239 },
240 },
241 })}\n`,
242 ),
243 );
244
245 const error = yield* Deferred.await(termination);
246 assert.equal(transformCalls, 0);
247 assert.instanceOf(error, AcpError.AcpProtocolParseError);
248 const parseError = error as AcpError.AcpProtocolParseError;
249 const { cause, ...directDiagnostics } = parseError;
250 assert.equal(parseError.operation, "decode-notification-payload");
251 assert.equal(parseError.method, "session/update");
252 assert.isAbove(parseError.issueCount ?? 0, 0);
253 assert.include(parseError.issueKinds ?? [], "Pointer");
254 assert.isAbove(parseError.maximumPathDepth ?? 0, 0);
255 assert.isTrue(Schema.isSchemaError(cause));
256 assert.notInclude(parseError.message, secret);
257 assert.notInclude(encodeUnknownJsonString(directDiagnostics), secret);
258 }),
259 );
260
261 it.effect("terminates when a callback on the reader dies", () =>
262 Effect.gen(function* () {
263 const { stdio, input } = yield* makeInMemoryStdio();
264 const termination = yield* Deferred.make<AcpError.AcpError>();
265 yield* AcpProtocol.makeAcpPatchedProtocol({
266 stdio,
267 serverRequestMethods: new Set(),
268 transformSessionUpdate: () => {
269 throw new Error("normalizer bug");
270 },
271 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
272 });
273
274 yield* Queue.offer(
275 input,
276 encoder.encode(
277 `${encodeUnknownJsonString({
278 jsonrpc: "2.0",
279 method: "session/update",
280 params: {
281 sessionId: "session-1",
282 update: { sessionUpdate: "plan", entries: [] },
283 },
284 })}\n`,
285 ),
286 );
287
288 // Pending requests are failed through termination instead of hanging.
289 const error = yield* TestClock.withLive(
290 Deferred.await(termination).pipe(Effect.timeout("2 seconds")),
291 );
292 assert.instanceOf(error, AcpError.AcpTransportError);
293 assert.equal((error as AcpError.AcpTransportError).operation, "read-input-stream");
294 }),
295 );
296
297 it.effect("logs outgoing notifications when logOutgoing is enabled", () =>
298 Effect.gen(function* () {
299 const { stdio } = yield* makeInMemoryStdio();
300 const events: Array<AcpProtocol.AcpProtocolLogEvent> = [];
301 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
302 stdio,
303 serverRequestMethods: new Set(),
304 logOutgoing: true,
305 logger: (event) =>
306 Effect.sync(() => {
307 events.push(event);
308 }),
309 });
310
311 yield* transport.notify("session/cancel", { sessionId: "session-1" });
312
313 // A notification must not carry `id` or `headers`. Grok CLI drops frames that do.
314 assert.deepEqual(events, [
315 {
316 direction: "outgoing",
317 stage: "decoded",
318 payload: {
319 _tag: "Notification",
320 tag: "session/cancel",
321 payload: {
322 sessionId: "session-1",
323 },
324 },
325 },
326 {
327 direction: "outgoing",
328 stage: "raw",
329 payload:
330 '{"jsonrpc":"2.0","method":"session/cancel","params":{"sessionId":"session-1"}}\n',
331 },
332 ]);
333 }),
334 );
335
336 it.effect("logs decode failures without copying the cause or wire payload", () =>
337 Effect.gen(function* () {
338 const secret = "acp-wire-secret-sentinel";
339 const { stdio, input } = yield* makeInMemoryStdio();
340 const events: Array<AcpProtocol.AcpProtocolLogEvent> = [];
341 const termination = yield* Deferred.make<AcpError.AcpError>();
342 yield* AcpProtocol.makeAcpPatchedProtocol({
343 stdio,
344 serverRequestMethods: new Set(),
345 logIncoming: true,
346 logger: (event) =>
347 Effect.sync(() => {
348 events.push(event);
349 }),
350 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
351 });
352
353 yield* Queue.offer(input, encoder.encode(`{"secret":"${secret}"\n`));
354 yield* Deferred.await(termination);
355
356 const event = events.find(({ stage }) => stage === "decode_failed");
357 assert.deepEqual(event, {
358 direction: "incoming",
359 stage: "decode_failed",
360 payload: {
361 operation: "decode-wire-message",
362 },
363 });
364 assert.notInclude(encodeUnknownJsonString(event), secret);
365 }),
366 );
367
368 it.effect("fails notification encoding through the declared ACP error channel", () =>
369 Effect.gen(function* () {
370 const { stdio } = yield* makeInMemoryStdio();
371 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
372 stdio,
373 serverRequestMethods: new Set(),
374 });
375
376 // Notifications encode through Schema, so the cause is the schema failure rather
377 // than the raw TypeError JSON.stringify throws. The ACP error shape is what callers see.
378 const bigintError = yield* transport.notify("x/test", 1n).pipe(Effect.flip);
379 assert.instanceOf(bigintError, AcpError.AcpProtocolParseError);
380 assert.equal(bigintError.operation, "encode-message");
381 assert.equal(bigintError.method, "x/test");
382 assert.isDefined(bigintError.cause);
383 assert.equal(
384 bigintError.message,
385 "ACP protocol operation 'encode-message' failed for method 'x/test'.",
386 );
387
388 const circular: Record<string, unknown> = {};
389 circular.self = circular;
390 const circularError = yield* transport.notify("x/test", circular).pipe(Effect.flip);
391 assert.instanceOf(circularError, AcpError.AcpProtocolParseError);
392 assert.equal(circularError.operation, "encode-message");
393 assert.equal(circularError.method, "x/test");
394 assert.isDefined(circularError.cause);
395
396 const requestError = yield* transport.request("x/request", 1n).pipe(
397 Effect.match({
398 onFailure: (error) => error,
399 onSuccess: () => assert.fail("Expected request encoding to fail"),
400 }),
401 );
402 assert.instanceOf(requestError, AcpError.AcpProtocolParseError);
403 assert.deepInclude(requestError, {
404 operation: "encode-message",
405 method: "x/request",
406 requestId: 1,
407 });
408 }),
409 );
410
411 it.effect("supports generic extension requests over the patched transport", () =>
412 Effect.gen(function* () {
413 const { stdio, input, output } = yield* makeInMemoryStdio();
414 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
415 stdio,
416 serverRequestMethods: new Set(),
417 });
418
419 const response = yield* transport
420 .request("x/test", { hello: "world" })
421 .pipe(Effect.forkScoped);
422 const outbound = yield* Queue.take(output);
423 assert.deepEqual(yield* decodeExtRequest(outbound), {
424 jsonrpc: "2.0",
425 id: 1,
426 method: "x/test",
427 params: {
428 hello: "world",
429 },
430 headers: [],
431 });
432
433 yield* Queue.offer(
434 input,
435 yield* encodeJsonl(ExtResponse, {
436 jsonrpc: "2.0",
437 id: 1,
438 result: {
439 ok: true,
440 },
441 }),
442 );
443
444 const resolved = yield* Fiber.join(response);
445 assert.deepEqual(resolved, { ok: true });
446 }),
447 );
448
449 it.effect("does not correlate a string response id with a numeric extension request id", () =>
450 Effect.gen(function* () {
451 const { stdio, input, output } = yield* makeInMemoryStdio();
452 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
453 stdio,
454 serverRequestMethods: new Set(),
455 });
456
457 const response = yield* transport
458 .request("x/test", { hello: "world" })
459 .pipe(Effect.forkScoped);
460 yield* Queue.take(output);
461 yield* Queue.offer(
462 input,
463 encoder.encode(
464 `${encodeUnknownJsonString({
465 jsonrpc: "2.0",
466 id: "1",
467 result: { ok: false },
468 })}\n`,
469 ),
470 );
471 yield* Effect.yieldNow;
472 assert.isUndefined(response.pollUnsafe());
473
474 yield* Queue.offer(
475 input,
476 yield* encodeJsonl(ExtResponse, {
477 jsonrpc: "2.0",
478 id: 1,
479 result: { ok: true },
480 }),
481 );
482 assert.deepEqual(yield* Fiber.join(response), { ok: true });
483 }),
484 );
485
486 it.effect("correlates extension response errors with the originating request", () =>
487 Effect.gen(function* () {
488 const { stdio, input, output } = yield* makeInMemoryStdio();
489 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
490 stdio,
491 serverRequestMethods: new Set(),
492 });
493
494 const response = yield* transport
495 .request("x/private", { hello: "world" })
496 .pipe(Effect.forkScoped);
497 yield* Queue.take(output);
498 yield* Queue.offer(
499 input,
500 encoder.encode(
501 `${encodeUnknownJsonString({
502 jsonrpc: "2.0",
503 id: 1,
504 error: {
505 _tag: "Cause",
506 code: -32602,
507 message: "Invalid params",
508 data: [
509 {
510 _tag: "Fail",
511 error: {
512 code: -32602,
513 message: "Invalid params",
514 data: { field: "hello" },
515 },
516 },
517 ],
518 },
519 })}\n`,
520 ),
521 );
522
523 const error = yield* Fiber.join(response).pipe(
524 Effect.match({
525 onFailure: (error) => error,
526 onSuccess: () => assert.fail("Expected extension request to fail"),
527 }),
528 );
529 assert.instanceOf(error, AcpError.AcpRequestError);
530 assert.deepInclude(error, {
531 code: -32602,
532 errorMessage: "Invalid params",
533 method: "x/private",
534 requestId: 1,
535 operation: "receive-response",
536 });
537 }),
538 );
539
540 it.effect("answers an extension request whose handler also dies as an internal error", () =>
541 Effect.gen(function* () {
542 const { stdio, input, output } = yield* makeInMemoryStdio();
543 yield* AcpProtocol.makeAcpPatchedProtocol({
544 stdio,
545 serverRequestMethods: new Set(),
546 // The typed failure must not hide the defect from its cleanup.
547 onExtRequest: () =>
548 Effect.fail(AcpError.AcpRequestError.invalidParams("bad params")).pipe(
549 Effect.ensuring(Effect.die(new Error("cleanup bug"))),
550 ),
551 });
552
553 yield* Queue.offer(
554 input,
555 yield* encodeJsonl(ExtRequest, {
556 jsonrpc: "2.0",
557 id: 9,
558 method: "x/test",
559 params: { hello: "world" },
560 headers: [],
561 }),
562 );
563
564 const response = yield* Schema.decodeUnknownEffect(
565 Schema.fromJsonString(JsonRpcErrorResponse),
566 )(yield* Queue.take(output));
567 assert.equal(response.id, 9);
568 assert.equal(response.error.code, -32603);
569 }),
570 );
571
572 it.effect("preserves numeric ids for inbound extension requests", () =>
573 Effect.gen(function* () {
574 const { stdio, input, output } = yield* makeInMemoryStdio();
575 yield* AcpProtocol.makeAcpPatchedProtocol({
576 stdio,
577 serverRequestMethods: new Set(),
578 onExtRequest: () => Effect.succeed({ ok: true }),
579 });
580
581 yield* Queue.offer(
582 input,
583 yield* encodeJsonl(ExtRequest, {
584 jsonrpc: "2.0",
585 id: 7,
586 method: "x/test",
587 params: {
588 hello: "world",
589 },
590 headers: [],
591 }),
592 );
593
594 const outbound = yield* Queue.take(output);
595 assert.deepEqual(yield* decodeExtResponse(outbound), {
596 jsonrpc: "2.0",
597 id: 7,
598 result: {
599 ok: true,
600 },
601 });
602 }),
603 );
604
605 it.effect("keeps numeric and string extension request ids distinct in handler context", () =>
606 Effect.gen(function* () {
607 const { stdio, input } = yield* makeInMemoryStdio();
608 const contexts = yield* Queue.unbounded<string>();
609 yield* AcpProtocol.makeAcpPatchedProtocol({
610 stdio,
611 serverRequestMethods: new Set(),
612 onExtRequest: (_method, _params, context) =>
613 Queue.offer(contexts, context.requestId).pipe(Effect.as({ ok: true })),
614 });
615
616 for (const id of [1, "1"] as const) {
617 yield* Queue.offer(
618 input,
619 yield* encodeJsonl(ExtRequest, {
620 jsonrpc: "2.0",
621 id,
622 method: "x/test",
623 params: { hello: "world" },
624 headers: [],
625 }),
626 );
627 }
628
629 assert.deepEqual(yield* Queue.takeAll(contexts), ["$t3:jsonrpc:number:1", "1"]);
630 }),
631 );
632
633 it.effect("preserves zero-valued ids for inbound core client requests", () =>
634 Effect.gen(function* () {
635 const { stdio, input, output } = yield* makeInMemoryStdio();
636 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
637 stdio,
638 serverRequestMethods: new Set(["session/request_permission"]),
639 });
640 const inboundRequest = yield* Deferred.make<unknown>();
641
642 yield* transport.serverProtocol
643 .run((_clientId, message) => Deferred.succeed(inboundRequest, message).pipe(Effect.asVoid))
644 .pipe(Effect.forkScoped);
645
646 yield* Queue.offer(
647 input,
648 yield* encodeJsonl(RequestPermissionRequest, {
649 jsonrpc: "2.0",
650 id: 0,
651 method: "session/request_permission",
652 params: {
653 sessionId: "session-1",
654 title: "Allow mock action",
655 subject: {
656 type: "tool_call",
657 toolCall: {
658 toolCallId: "tool-1",
659 title: "Allow mock action",
660 },
661 },
662 options: [{ optionId: "allow", name: "Allow", kind: "allow_once" }],
663 },
664 headers: [],
665 }),
666 );
667
668 const message = yield* Deferred.await(inboundRequest);
669 assert.deepEqual(message, {
670 _tag: "Request",
671 id: 0,
672 tag: "session/request_permission",
673 payload: {
674 sessionId: "session-1",
675 title: "Allow mock action",
676 subject: {
677 type: "tool_call",
678 toolCall: {
679 toolCallId: "tool-1",
680 title: "Allow mock action",
681 },
682 },
683 options: [{ optionId: "allow", name: "Allow", kind: "allow_once" }],
684 },
685 headers: [],
686 });
687
688 yield* transport.serverProtocol.send(0, {
689 _tag: "Exit",
690 requestId: 0,
691 exit: {
692 _tag: "Success",
693 value: {
694 outcome: {
695 outcome: "selected",
696 optionId: "allow",
697 },
698 },
699 },
700 });
701
702 const outbound = yield* Queue.take(output);
703 assert.deepEqual(yield* decodeRequestPermissionResponse(outbound), {
704 jsonrpc: "2.0",
705 id: 0,
706 result: {
707 outcome: {
708 outcome: "selected",
709 optionId: "allow",
710 },
711 },
712 });
713 }),
714 );
715
716 it.effect("acknowledges outgoing responses and reports encoding failures by request id", () =>
717 Effect.gen(function* () {
718 const { stdio } = yield* makeInMemoryStdio();
719 const acknowledged = yield* Deferred.make<string>();
720 const failed = yield* Deferred.make<readonly [string, AcpError.AcpError]>();
721 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
722 stdio,
723 serverRequestMethods: new Set(),
724 onOutgoingResponse: (requestId) =>
725 Deferred.succeed(acknowledged, requestId).pipe(Effect.asVoid),
726 onOutgoingResponseFailure: (requestId, error) =>
727 Deferred.succeed(failed, [requestId, error]).pipe(Effect.asVoid),
728 });
729
730 yield* transport.serverProtocol.send(0, {
731 _tag: "Exit",
732 requestId: "success-1",
733 exit: { _tag: "Success", value: { ok: true } },
734 });
735 assert.equal(yield* Deferred.await(acknowledged), "success-1");
736
737 const circular: { self?: unknown } = {};
738 circular.self = circular;
739 yield* transport.serverProtocol
740 .send(0, {
741 _tag: "Exit",
742 requestId: "failure-1",
743 exit: { _tag: "Success", value: circular },
744 })
745 .pipe(Effect.exit);
746 const [requestId, error] = yield* Deferred.await(failed);
747 assert.equal(requestId, "failure-1");
748 assert.instanceOf(error, AcpError.AcpProtocolParseError);
749 }),
750 );
751
752 it.effect("acknowledges a response only after the stdout write completes", () =>
753 Effect.gen(function* () {
754 const writeStarted = yield* Deferred.make<void>();
755 const releaseWrite = yield* Deferred.make<void>();
756 const acknowledged = yield* Deferred.make<string>();
757 const stdio = Stdio.make({
758 args: Effect.succeed([]),
759 stdin: Stream.never,
760 stdout: () =>
761 Sink.forEach(() =>
762 Deferred.succeed(writeStarted, undefined).pipe(
763 Effect.andThen(Deferred.await(releaseWrite)),
764 ),
765 ),
766 stderr: () => Sink.drain,
767 });
768 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
769 stdio,
770 serverRequestMethods: new Set(),
771 onOutgoingResponse: (requestId) =>
772 Deferred.succeed(acknowledged, requestId).pipe(Effect.asVoid),
773 });
774 const sendFiber = yield* transport.serverProtocol
775 .send(0, {
776 _tag: "Exit",
777 requestId: "held-write",
778 exit: { _tag: "Success", value: { ok: true } },
779 })
780 .pipe(Effect.forkScoped);
781
782 yield* Deferred.await(writeStarted);
783 assert.isUndefined(sendFiber.pollUnsafe());
784 assert.isFalse(yield* Deferred.isDone(acknowledged));
785 yield* Deferred.succeed(releaseWrite, undefined);
786 yield* Fiber.join(sendFiber);
787 assert.equal(yield* Deferred.await(acknowledged), "held-write");
788 }),
789 );
790
791 it.effect("fails current, queued, and future responses when the stdout writer fails", () =>
792 Effect.gen(function* () {
793 const writeStarted = yield* Deferred.make<void>();
794 const releaseFailure = yield* Deferred.make<void>();
795 const failures = yield* Queue.unbounded<string>();
796 const terminated = yield* Deferred.make<AcpError.AcpError>();
797 const stdio = Stdio.make({
798 args: Effect.succeed([]),
799 stdin: Stream.never,
800 stdout: () =>
801 Sink.forEach(() =>
802 Deferred.succeed(writeStarted, undefined).pipe(
803 Effect.andThen(Deferred.await(releaseFailure)),
804 Effect.andThen(Effect.die("stdout failed")),
805 ),
806 ),
807 stderr: () => Sink.drain,
808 });
809 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
810 stdio,
811 serverRequestMethods: new Set(),
812 onOutgoingResponseFailure: (requestId) => Queue.offer(failures, requestId),
813 onTermination: (error) => Deferred.succeed(terminated, error).pipe(Effect.asVoid),
814 });
815 const send = (requestId: string) =>
816 transport.serverProtocol
817 .send(0, {
818 _tag: "Exit",
819 requestId,
820 exit: { _tag: "Success", value: { ok: true } },
821 })
822 .pipe(Effect.exit);
823 const current = yield* send("current").pipe(Effect.forkScoped);
824 yield* Deferred.await(writeStarted);
825 const queued = yield* send("queued").pipe(Effect.forkScoped);
826 yield* Effect.yieldNow;
827 assert.isUndefined(current.pollUnsafe());
828 assert.isUndefined(queued.pollUnsafe());
829 yield* Deferred.succeed(releaseFailure, undefined);
830 assert.instanceOf(yield* Deferred.await(terminated), AcpError.AcpTransportError);
831 yield* Fiber.join(current);
832 yield* Fiber.join(queued);
833 yield* send("future");
834
835 assert.deepEqual(
836 new Set(yield* Queue.takeN(failures, 3)),
837 new Set(["current", "queued", "future"]),
838 );
839 }),
840 );
841
842 it.effect(
843 "atomically rejects an admitted response when the writer closes before queue offer",
844 () =>
845 Effect.gen(function* () {
846 const admitted = yield* Deferred.make<void>();
847 const releaseAdmission = yield* Deferred.make<void>();
848 const failures = yield* Queue.unbounded<string>();
849 const protocolScope = yield* Scope.make();
850 const stdio = Stdio.make({
851 args: Effect.succeed([]),
852 stdin: Stream.never,
853 stdout: () => Sink.drain,
854 stderr: () => Sink.drain,
855 });
856 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
857 stdio,
858 serverRequestMethods: new Set(),
859 onOutgoingResponseFailure: (requestId) => Queue.offer(failures, requestId),
860 testHooks: {
861 onOutgoingWriteAdmitted: () =>
862 Deferred.succeed(admitted, undefined).pipe(
863 Effect.andThen(Deferred.await(releaseAdmission)),
864 ),
865 },
866 }).pipe(Effect.provideService(Scope.Scope, protocolScope));
867 const send = (requestId: string) =>
868 transport.serverProtocol
869 .send(0, {
870 _tag: "Exit",
871 requestId,
872 exit: { _tag: "Success", value: { ok: true } },
873 })
874 .pipe(Effect.exit);
875 const admittedSend = yield* send("admitted").pipe(Effect.forkScoped);
876
877 yield* Deferred.await(admitted);
878 yield* Scope.close(protocolScope, Exit.void);
879 yield* Deferred.succeed(releaseAdmission, undefined);
880 yield* Fiber.join(admittedSend);
881 yield* send("future");
882
883 assert.deepEqual(yield* Queue.takeN(failures, 2), ["admitted", "future"]);
884 }),
885 );
886
887 it.effect("drops agent-internal JSON-RPC responses with non-numeric request ids", () =>
888 Effect.gen(function* () {
889 const { stdio, input } = yield* makeInMemoryStdio();
890 const notifications =
891 yield* Deferred.make<ReadonlyArray<AcpProtocol.AcpIncomingNotification>>();
892 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
893 stdio,
894 serverRequestMethods: new Set(),
895 });
896
897 yield* transport.incoming.pipe(
898 Stream.take(1),
899 Stream.runCollect,
900 Effect.flatMap((notificationChunk) => Deferred.succeed(notifications, notificationChunk)),
901 Effect.forkScoped,
902 );
903
904 yield* Queue.offer(
905 input,
906 encoder.encode(
907 `${encodeUnknownJsonString({
908 jsonrpc: "2.0",
909 id: "skills-reload",
910 result: {
911 reloaded: true,
912 },
913 })}\n`,
914 ),
915 );
916 yield* Queue.offer(
917 input,
918 yield* encodeJsonl(SessionUpdateNotification, {
919 jsonrpc: "2.0",
920 method: "session/update",
921 params: {
922 sessionId: "session-1",
923 update: {
924 sessionUpdate: "plan",
925 entries: [
926 {
927 content: "Reload skills",
928 priority: "high",
929 status: "in_progress",
930 },
931 ],
932 },
933 },
934 }),
935 );
936
937 const [update] = yield* Deferred.await(notifications);
938 assert.equal(update?._tag, "SessionUpdate");
939 }),
940 );
941
942 it.effect("cleans up interrupted extension requests before a late response arrives", () =>
943 Effect.gen(function* () {
944 const { stdio, input, output } = yield* makeInMemoryStdio();
945 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
946 stdio,
947 serverRequestMethods: new Set(),
948 });
949 const lateResponse = yield* Deferred.make<unknown>();
950
951 yield* transport.clientProtocol
952 .run(0, (message) => Deferred.succeed(lateResponse, message).pipe(Effect.asVoid))
953 .pipe(Effect.forkScoped);
954
955 const response = yield* transport
956 .request("x/test", { hello: "world" })
957 .pipe(Effect.forkScoped);
958 const outbound = yield* Queue.take(output);
959 assert.deepEqual(yield* decodeExtRequest(outbound), {
960 jsonrpc: "2.0",
961 id: 1,
962 method: "x/test",
963 params: {
964 hello: "world",
965 },
966 headers: [],
967 });
968
969 yield* Fiber.interrupt(response);
970 yield* Queue.offer(
971 input,
972 yield* encodeJsonl(ExtResponse, {
973 jsonrpc: "2.0",
974 id: 1,
975 result: {
976 ok: true,
977 },
978 }),
979 );
980
981 const message = yield* Deferred.await(lateResponse);
982 assert.deepEqual(message, {
983 _tag: "Exit",
984 requestId: 1,
985 exit: {
986 _tag: "Success",
987 value: {
988 ok: true,
989 },
990 },
991 });
992 }),
993 );
994
995 it.effect("propagates the real child exit code when the input stream ends", () =>
996 Effect.gen(function* () {
997 const handle = yield* makeHandle({ ACP_MOCK_EXIT_IMMEDIATELY_CODE: "7" });
998 const firstMessage = yield* Deferred.make<unknown>();
999 const termination = yield* Deferred.make<AcpError.AcpError>();
1000 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
1001 stdio: makeChildStdio(handle),
1002 terminationError: makeTerminationError(handle),
1003 serverRequestMethods: new Set(),
1004 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
1005 });
1006
1007 yield* transport.clientProtocol
1008 .run(0, (message) => Deferred.succeed(firstMessage, message).pipe(Effect.asVoid))
1009 .pipe(Effect.forkScoped);
1010
1011 const message = yield* Deferred.await(firstMessage);
1012 const exitError = yield* Deferred.await(termination);
1013 assert.instanceOf(exitError, AcpError.AcpProcessExitedError);
1014 assert.equal((exitError as AcpError.AcpProcessExitedError).code, 7);
1015 assert.equal((message as { readonly _tag?: string })._tag, "ClientProtocolError");
1016 const defect = (message as { readonly error: { readonly reason: unknown } }).error.reason as {
1017 readonly _tag: string;
1018 readonly message: string;
1019 readonly cause: unknown;
1020 };
1021 assert.equal(defect._tag, "RpcClientDefect");
1022 assert.equal(defect.message, "ACP protocol terminated.");
1023 assert.instanceOf(defect.cause, AcpError.AcpProcessExitedError);
1024 assert.equal((defect.cause as AcpError.AcpProcessExitedError).code, 7);
1025 }),
1026 );
1027
1028 it.effect("classifies an input stream ending without inventing a cause", () =>
1029 Effect.gen(function* () {
1030 const { stdio, input } = yield* makeInMemoryStdio();
1031 const termination = yield* Deferred.make<AcpError.AcpError>();
1032 yield* AcpProtocol.makeAcpPatchedProtocol({
1033 stdio,
1034 serverRequestMethods: new Set(),
1035 onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
1036 });
1037
1038 yield* Queue.end(input);
1039
1040 const error = yield* Deferred.await(termination);
1041 assert.instanceOf(error, AcpError.AcpInputStreamEndedError);
1042 assert.equal(error.message, "ACP input stream ended.");
1043 assert.equal("cause" in error, false);
1044 }),
1045 );
1046
1047 it.effect("does not treat intentional serverProtocol.end as a transport failure", () =>
1048 Effect.gen(function* () {
1049 const { stdio, input } = yield* makeInMemoryStdio();
1050 const terminations = yield* Ref.make(0);
1051 const writerExit = yield* Deferred.make<{ readonly intentionalEnd: boolean }>();
1052 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
1053 stdio,
1054 serverRequestMethods: new Set(),
1055 onTermination: () => Ref.update(terminations, (count) => count + 1),
1056 testHooks: {
1057 onOutgoingWriterExit: (detail) =>
1058 Deferred.succeed(writerExit, detail).pipe(Effect.asVoid),
1059 },
1060 });
1061
1062 yield* transport.serverProtocol.end(0);
1063 const exitDetail = yield* Deferred.await(writerExit);
1064 assert.isTrue(exitDetail.intentionalEnd);
1065 assert.equal(yield* Ref.get(terminations), 0);
1066
1067 // Tear down the forever input reader so the scoped test can finish.
1068 yield* Queue.end(input);
1069 }),
1070 );
1071
1072 it.effect("does not emit a second process-exit error after a decode failure", () =>
1073 Effect.gen(function* () {
1074 const handle = yield* makeHandle({
1075 ACP_MOCK_MALFORMED_OUTPUT: "1",
1076 ACP_MOCK_MALFORMED_OUTPUT_EXIT_CODE: "23",
1077 });
1078 const terminationCalls = yield* Ref.make(0);
1079 const firstMessage = yield* Deferred.make<unknown>();
1080 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
1081 stdio: makeChildStdio(handle),
1082 terminationError: makeTerminationError(handle),
1083 serverRequestMethods: new Set(),
1084 onTermination: () => Ref.update(terminationCalls, (count) => count + 1),
1085 });
1086
1087 yield* transport.clientProtocol
1088 .run(0, (message) => Deferred.succeed(firstMessage, message).pipe(Effect.asVoid))
1089 .pipe(Effect.forkScoped);
1090
1091 const message = yield* Deferred.await(firstMessage);
1092 assert.equal(yield* Ref.get(terminationCalls), 1);
1093 assert.equal((message as { readonly _tag?: string })._tag, "ClientProtocolError");
1094 const defect = (message as { readonly error: { readonly reason: unknown } }).error.reason as {
1095 readonly _tag: string;
1096 readonly message: string;
1097 readonly cause: unknown;
1098 };
1099 assert.equal(defect._tag, "RpcClientDefect");
1100 assert.equal(defect.message, "ACP protocol terminated.");
1101 assert.instanceOf(defect.cause, AcpError.AcpProtocolParseError);
1102 }),
1103 );
1104
1105 it.effect("keeps client send failure messages independent from the cause", () =>
1106 Effect.gen(function* () {
1107 const { stdio } = yield* makeInMemoryStdio();
1108 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
1109 stdio,
1110 serverRequestMethods: new Set(),
1111 });
1112
1113 const failure = yield* transport.clientProtocol
1114 .send(0, {
1115 _tag: "Request",
1116 id: "request-1",
1117 tag: "x/test",
1118 payload: 1n,
1119 headers: [],
1120 })
1121 .pipe(Effect.flip);
1122 const defect = failure.reason as {
1123 readonly _tag: string;
1124 readonly message: string;
1125 readonly cause: unknown;
1126 };
1127
1128 assert.equal(defect._tag, "RpcClientDefect");
1129 assert.equal(defect.message, "Failed to send ACP protocol message.");
1130 assert.instanceOf(defect.cause, AcpError.AcpProtocolParseError);
1131 }),
1132 );
1133
1134 it.effect("fails pending extension requests with the propagated exit code", () =>
1135 Effect.gen(function* () {
1136 const { stdio, input, output } = yield* makeInMemoryStdio();
1137 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
1138 stdio,
1139 terminationError: Effect.succeed(new AcpError.AcpProcessExitedError({ code: 0 })),
1140 serverRequestMethods: new Set(),
1141 });
1142
1143 const response = yield* transport
1144 .request("x/test", { hello: "world" })
1145 .pipe(Effect.forkScoped);
1146 yield* Queue.take(output);
1147 yield* Queue.end(input);
1148
1149 const error = yield* Fiber.join(response).pipe(
1150 Effect.match({
1151 onFailure: (error) => error,
1152 onSuccess: () => assert.fail("Expected request to fail after process exit"),
1153 }),
1154 );
1155 assert.instanceOf(error, AcpError.AcpProcessExitedError);
1156 assert.equal(error.code, 0);
1157 }),
1158 );
1159
1160 it.effect.each(["request", "notification"] as const)(
1161 "rejects a %s if the connection ends while its logger is running",
1162 (operation) =>
1163 Effect.gen(function* () {
1164 const { stdio, input, output } = yield* makeInMemoryStdio();
1165 const writeStarted = yield* Deferred.make<void>();
1166 const releaseWrite = yield* Deferred.make<void>();
1167 const terminated = yield* Deferred.make<AcpError.AcpError>();
1168 const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
1169 stdio,
1170 serverRequestMethods: new Set(),
1171 logOutgoing: true,
1172 logger: (event) =>
1173 event.stage === "raw"
1174 ? Deferred.succeed(writeStarted, undefined).pipe(
1175 Effect.andThen(Deferred.await(releaseWrite)),
1176 )
1177 : Effect.void,
1178 onTermination: (error) => Deferred.succeed(terminated, error).pipe(Effect.asVoid),
1179 });
1180 const send = yield* (
1181 operation === "request"
1182 ? transport.request("x/test", { hello: "world" })
1183 : transport.notify("session/cancel", { sessionId: "session-1" })
1184 ).pipe(Effect.forkScoped);
1185
1186 yield* Deferred.await(writeStarted);
1187 yield* Queue.end(input);
1188 const error = yield* Deferred.await(terminated);
1189 yield* Deferred.succeed(releaseWrite, undefined);
1190
1191 const failure = yield* Fiber.join(send).pipe(
1192 Effect.match({
1193 onFailure: (failure) => failure,
1194 onSuccess: () => assert.fail("Expected the send to fail after termination"),
1195 }),
1196 );
1197 assert.strictEqual(failure, error);
1198 assert.equal(yield* Queue.size(output), 0);
1199 }),
1200 );
1201});