1import * as Deferred from "effect/Deferred";
2import * as Effect from "effect/Effect";
3import * as Exit from "effect/Exit";
4import * as Fiber from "effect/Fiber";
5import * as Layer from "effect/Layer";
6import * as Queue from "effect/Queue";
7import * as Ref from "effect/Ref";
8import * as Schema from "effect/Schema";
9import * as Scope from "effect/Scope";
10
11import { assert, it } from "@effect/vitest";
12
13import * as AcpAgent from "./agent.ts";
14import * as AcpSchema from "./_generated/schema.gen.ts";
15import type * as AcpProtocol from "./protocol.ts";
16import {
17 encodeJsonl,
18 jsonRpcNotification,
19 jsonRpcRequest,
20 jsonRpcResponse,
21} from "./_internal/shared.ts";
22import { makeInMemoryStdio } from "./_internal/stdio.ts";
23
24const RequestPermissionRequest = jsonRpcRequest(
25 "session/request_permission",
26 AcpSchema.RequestPermissionRequest,
27);
28const InitializeRequest = jsonRpcRequest("initialize", AcpSchema.InitializeRequest);
29const InitializeResponse = jsonRpcResponse(AcpSchema.InitializeResponse);
30const RequestPermissionResponse = jsonRpcResponse(AcpSchema.RequestPermissionResponse);
31const SessionCancelNotification = jsonRpcNotification(
32 "session/cancel",
33 AcpSchema.CancelSessionNotification,
34);
35const ExtPingNotification = jsonRpcNotification("x/ping", Schema.Struct({ count: Schema.Number }));
36const ExtRequest = jsonRpcRequest("x/test", Schema.Struct({ hello: Schema.String }));
37const ExtResponse = jsonRpcResponse(Schema.Struct({ ok: Schema.Boolean }));
38/** A response whose cause is a handler's defect, as RpcServer encodes it. */
39const DieResponse = Schema.Struct({
40 id: Schema.Number,
41 error: Schema.Struct({
42 _tag: Schema.Literal("Cause"),
43 data: Schema.Tuple([Schema.Struct({ _tag: Schema.Literal("Die") })]),
44 }),
45});
46const decodeRequestPermissionRequest = Schema.decodeEffect(
47 Schema.fromJsonString(RequestPermissionRequest),
48);
49const decodeInitializeResponse = Schema.decodeEffect(Schema.fromJsonString(InitializeResponse));
50const decodeExtResponse = Schema.decodeEffect(Schema.fromJsonString(ExtResponse));
51
52it.effect("effect-acp agent handles core agent requests and outbound client requests", () =>
53 Effect.gen(function* () {
54 const { stdio, input, output } = yield* makeInMemoryStdio();
55 const cancelNotifications = yield* Ref.make<Array<string>>([]);
56 const extNotifications = yield* Ref.make<Array<number>>([]);
57 const requestContexts = yield* Ref.make<Array<AcpProtocol.AcpRequestContext>>([]);
58 const cancelReceived = yield* Deferred.make<void>();
59 const extReceived = yield* Deferred.make<void>();
60 const scope = yield* Scope.make();
61 const context = yield* Layer.buildWithScope(AcpAgent.layer(stdio), scope);
62
63 yield* Effect.gen(function* () {
64 const agent = yield* AcpAgent.AcpAgent;
65
66 yield* agent.handleInitialize((_request, requestContext) =>
67 Ref.update(requestContexts, (current) => [...current, requestContext]).pipe(
68 Effect.as({
69 protocolVersion: 2,
70 capabilities: {},
71 info: {
72 name: "mock-agent",
73 version: "0.0.0",
74 },
75 }),
76 ),
77 );
78 yield* agent.handleExtRequest(
79 "x/test",
80 Schema.Struct({ hello: Schema.String }),
81 (_payload, requestContext) =>
82 Ref.update(requestContexts, (current) => [...current, requestContext]).pipe(
83 Effect.as({ ok: true }),
84 ),
85 );
86 yield* agent.handleCancel((notification) =>
87 Ref.update(cancelNotifications, (current) => [...current, notification.sessionId]).pipe(
88 Effect.andThen(Deferred.succeed(cancelReceived, undefined)),
89 ),
90 );
91 yield* agent.handleExtNotification(
92 "x/ping",
93 Schema.Struct({ count: Schema.Number }),
94 (payload) =>
95 Ref.update(extNotifications, (current) => [...current, payload.count]).pipe(
96 Effect.andThen(Deferred.succeed(extReceived, undefined)),
97 ),
98 );
99
100 const permissionFiber = yield* agent.client
101 .requestPermission({
102 sessionId: "session-1",
103 title: "Allow mock action",
104 subject: {
105 type: "tool_call",
106 toolCall: {
107 toolCallId: "tool-1",
108 title: "Allow mock action",
109 },
110 },
111 options: [{ optionId: "allow", name: "Allow", kind: "allow_once" }],
112 })
113 .pipe(Effect.forkScoped);
114
115 const permissionRequest = yield* decodeRequestPermissionRequest(yield* Queue.take(output));
116 assert.equal(permissionRequest.jsonrpc, "2.0");
117 assert.equal(permissionRequest.method, "session/request_permission");
118 assert.deepEqual(permissionRequest.params, {
119 sessionId: "session-1",
120 title: "Allow mock action",
121 subject: {
122 type: "tool_call",
123 toolCall: {
124 toolCallId: "tool-1",
125 title: "Allow mock action",
126 },
127 },
128 options: [{ optionId: "allow", name: "Allow", kind: "allow_once" }],
129 });
130 assert.deepEqual(permissionRequest.headers, []);
131
132 yield* Queue.offer(
133 input,
134 yield* encodeJsonl(RequestPermissionResponse, {
135 jsonrpc: "2.0",
136 id: permissionRequest.id,
137 result: {
138 outcome: {
139 outcome: "selected",
140 optionId: "allow",
141 },
142 },
143 }),
144 );
145
146 const permission = yield* Fiber.join(permissionFiber);
147 assert.equal(permission.outcome.outcome, "selected");
148
149 yield* Queue.offer(
150 input,
151 yield* encodeJsonl(InitializeRequest, {
152 jsonrpc: "2.0",
153 id: 2,
154 method: "initialize",
155 params: {
156 protocolVersion: 2,
157 capabilities: {},
158 info: {
159 name: "effect-acp-test",
160 version: "0.0.0",
161 },
162 },
163 headers: [],
164 }),
165 );
166
167 const initResponse = yield* decodeInitializeResponse(yield* Queue.take(output));
168 assert.deepEqual(initResponse, {
169 jsonrpc: "2.0",
170 id: 2,
171 result: {
172 protocolVersion: 2,
173 capabilities: {},
174 info: {
175 name: "mock-agent",
176 version: "0.0.0",
177 },
178 },
179 });
180
181 yield* Queue.offer(
182 input,
183 yield* encodeJsonl(ExtRequest, {
184 jsonrpc: "2.0",
185 id: "extension-3",
186 method: "x/test",
187 params: { hello: "world" },
188 headers: [],
189 }),
190 );
191 const extResponse = yield* decodeExtResponse(yield* Queue.take(output));
192 assert.deepEqual(extResponse, {
193 jsonrpc: "2.0",
194 id: "extension-3",
195 result: { ok: true },
196 });
197 assert.deepEqual(yield* Ref.get(requestContexts), [
198 { requestId: "$t3:jsonrpc:number:2", method: "initialize" },
199 { requestId: "extension-3", method: "x/test" },
200 ]);
201
202 yield* Queue.offer(
203 input,
204 yield* encodeJsonl(SessionCancelNotification, {
205 jsonrpc: "2.0",
206 method: "session/cancel",
207 params: {
208 sessionId: "session-1",
209 },
210 }),
211 );
212 yield* Queue.offer(
213 input,
214 yield* encodeJsonl(ExtPingNotification, {
215 jsonrpc: "2.0",
216 method: "x/ping",
217 params: { count: 2 },
218 }),
219 );
220
221 yield* Deferred.await(cancelReceived);
222 yield* Deferred.await(extReceived);
223 assert.deepEqual(yield* Ref.get(cancelNotifications), ["session-1"]);
224 assert.deepEqual(yield* Ref.get(extNotifications), [2]);
225 }).pipe(Effect.provide(context), Effect.ensuring(Scope.close(scope, Exit.void)));
226 }),
227);
228
229it.effect("effect-acp agent uses distinct ids for RPC calls and extension requests", () =>
230 Effect.gen(function* () {
231 const { stdio, input, output } = yield* makeInMemoryStdio();
232 const scope = yield* Scope.make();
233 const context = yield* Layer.buildWithScope(AcpAgent.layer(stdio), scope);
234
235 yield* Effect.gen(function* () {
236 const agent = yield* AcpAgent.AcpAgent;
237
238 const permissionFiber = yield* agent.client
239 .requestPermission({
240 sessionId: "session-1",
241 title: "Allow mock action",
242 subject: {
243 type: "tool_call",
244 toolCall: {
245 toolCallId: "tool-1",
246 title: "Allow mock action",
247 },
248 },
249 options: [{ optionId: "allow", name: "Allow", kind: "allow_once" }],
250 })
251 .pipe(Effect.forkScoped);
252 const extFiber = yield* agent.client
253 .extRequest("x/test", { hello: "world" })
254 .pipe(Effect.forkScoped);
255
256 const firstOutbound = yield* Queue.take(output);
257 const secondOutbound = yield* Queue.take(output);
258
259 const decodedPermission = Schema.decodeEffect(
260 Schema.fromJsonString(RequestPermissionRequest),
261 );
262 const decodedExt = Schema.decodeEffect(Schema.fromJsonString(ExtRequest));
263 const firstIsPermission = yield* decodedPermission(firstOutbound).pipe(
264 Effect.match({
265 onFailure: () => false,
266 onSuccess: () => true,
267 }),
268 );
269
270 const permissionRequest = firstIsPermission
271 ? yield* decodedPermission(firstOutbound)
272 : yield* decodedPermission(secondOutbound);
273 const extRequest = firstIsPermission
274 ? yield* decodedExt(secondOutbound)
275 : yield* decodedExt(firstOutbound);
276
277 assert.notEqual(permissionRequest.id, extRequest.id);
278
279 yield* Queue.offer(
280 input,
281 yield* encodeJsonl(RequestPermissionResponse, {
282 jsonrpc: "2.0",
283 id: permissionRequest.id,
284 result: {
285 outcome: {
286 outcome: "selected",
287 optionId: "allow",
288 },
289 },
290 }),
291 );
292 yield* Queue.offer(
293 input,
294 yield* encodeJsonl(ExtResponse, {
295 jsonrpc: "2.0",
296 id: extRequest.id,
297 result: { ok: true },
298 }),
299 );
300
301 const permission = yield* Fiber.join(permissionFiber);
302 assert.equal(permission.outcome.outcome, "selected");
303 assert.deepEqual(yield* Fiber.join(extFiber), { ok: true });
304 }).pipe(Effect.provide(context), Effect.ensuring(Scope.close(scope, Exit.void)));
305 }),
306);
307
308it.effect("effect-acp agent answers a request whose handler dies with an error for it", () =>
309 Effect.gen(function* () {
310 const { stdio, input, output } = yield* makeInMemoryStdio();
311 const scope = yield* Scope.make();
312 const context = yield* Layer.buildWithScope(AcpAgent.layer(stdio), scope);
313 const agent = yield* Effect.service(AcpAgent.AcpAgent).pipe(Effect.provide(context));
314 yield* agent.handleInitialize(() => Effect.die(new Error("handler bug")));
315
316 yield* Queue.offer(
317 input,
318 yield* encodeJsonl(InitializeRequest, {
319 jsonrpc: "2.0",
320 id: 7,
321 method: "initialize",
322 params: {
323 protocolVersion: 2,
324 capabilities: {},
325 info: { name: "effect-acp-test", version: "0.0.0" },
326 },
327 headers: [],
328 }),
329 );
330 const response = yield* Queue.take(output).pipe(
331 Effect.flatMap(Schema.decodeEffect(Schema.fromJsonString(DieResponse))),
332 );
333 assert.equal(response.id, 7);
334 yield* Scope.close(scope, Exit.void);
335 }),
336);