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