Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion packages/effect-acp/src/agent.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,8 @@ const SessionCancelNotification = jsonRpcNotification(
const ExtPingNotification = jsonRpcNotification("x/ping", Schema.Struct({ count: Schema.Number }));
const ExtRequest = jsonRpcRequest("x/test", Schema.Struct({ hello: Schema.String }));
const ExtResponse = jsonRpcResponse(Schema.Struct({ ok: Schema.Boolean }));
/** Peers on the Kotlin ACP SDK (Junie) decode numeric request ids as a signed 32-bit Int. */
const isInt32RequestId = Schema.is(Schema.Number.check(Schema.isInt32()));
/** A response whose cause is a handler's defect, as RpcServer encodes it. */
const DieResponse = Schema.Struct({
id: Schema.Number,
Expand Down Expand Up @@ -226,7 +228,7 @@ it.effect("effect-acp agent handles core agent requests and outbound client requ
}),
);

it.effect("effect-acp agent uses distinct ids for RPC calls and extension requests", () =>
it.effect("effect-acp agent keeps RPC and extension ids distinct and within signed 32-bit", () =>
Effect.gen(function* () {
const { stdio, input, output } = yield* makeInMemoryStdio();
const scope = yield* Scope.make();
Expand Down Expand Up @@ -275,6 +277,7 @@ it.effect("effect-acp agent uses distinct ids for RPC calls and extension reques
: yield* decodedExt(firstOutbound);

assert.notEqual(permissionRequest.id, extRequest.id);
assert.isTrue(isInt32RequestId(permissionRequest.id) && isInt32RequestId(extRequest.id));

yield* Queue.offer(
input,
Expand Down
3 changes: 1 addition & 2 deletions packages/effect-acp/src/agent.ts
Original file line number Diff line number Diff line change
Expand Up @@ -436,9 +436,8 @@ export const make = Effect.fn("effect-acp/AcpAgent.make")(function* (
Effect.forkScoped,
);

let nextRpcRequestId = 2 ** 32;
const rpc = yield* RpcClient.make(AcpRpcs.ClientRpcs, {
generateRequestId: () => RpcMessage.RequestId(nextRpcRequestId++),
generateRequestId: () => RpcMessage.RequestId(transport.nextRequestId()),
}).pipe(Effect.provideService(RpcClient.Protocol, transport.clientProtocol));

return AcpAgent.of({
Expand Down
5 changes: 4 additions & 1 deletion packages/effect-acp/src/client.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,8 @@ const PromptRequestV1 = jsonRpcRequest("session/prompt", AcpSchemaV1.PromptReque
const PromptResponseV1 = jsonRpcResponse(AcpSchemaV1.PromptResponse);
const ExtRequest = jsonRpcRequest("x/test", Schema.Struct({ hello: Schema.String }));
const ExtResponse = jsonRpcResponse(Schema.Struct({ ok: Schema.Boolean }));
/** Peers on the Kotlin ACP SDK (Junie) decode numeric request ids as a signed 32-bit Int. */
const isInt32RequestId = Schema.is(Schema.Number.check(Schema.isInt32()));
const PromptRequest = jsonRpcRequest("session/prompt", AcpSchema.PromptRequest);
const PromptResponse = jsonRpcResponse(AcpSchema.PromptResponse);
const SessionUpdateNotification = jsonRpcNotification("session/update", Schema.Unknown);
Expand Down Expand Up @@ -755,7 +757,7 @@ it.layer(NodeServices.layer)("effect-acp client", (it) => {
}),
);

it.effect("uses distinct ids for RPC calls and extension requests", () =>
it.effect("keeps RPC and extension ids distinct and within signed 32-bit", () =>
Effect.gen(function* () {
const { stdio, input, output } = yield* makeInMemoryStdio();
const scope = yield* Scope.make();
Expand Down Expand Up @@ -796,6 +798,7 @@ it.layer(NodeServices.layer)("effect-acp client", (it) => {
: yield* decodedExt(firstOutbound);

assert.notEqual(initializeRequest.id, extRequest.id);
assert.isTrue(isInt32RequestId(initializeRequest.id) && isInt32RequestId(extRequest.id));

yield* Queue.offer(
input,
Expand Down
3 changes: 1 addition & 2 deletions packages/effect-acp/src/client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1113,9 +1113,8 @@ export const make = Effect.fn("effect-acp/AcpClient.make")(function* (
Effect.forkScoped,
);

let nextRpcRequestId = 2 ** 32;
const rpc = yield* RpcClient.make(AcpRpcs.CompatAgentRpcs, {
generateRequestId: () => RpcMessage.RequestId(nextRpcRequestId++),
generateRequestId: () => RpcMessage.RequestId(transport.nextRequestId()),
}).pipe(Effect.provideService(RpcClient.Protocol, transport.clientProtocol));

return AcpClient.of({
Expand Down
18 changes: 13 additions & 5 deletions packages/effect-acp/src/protocol.ts
Original file line number Diff line number Diff line change
Expand Up @@ -121,6 +121,15 @@ export interface AcpPatchedProtocol {
readonly incoming: Stream.Stream<AcpIncomingNotification>;
readonly request: (method: string, payload: unknown) => Effect.Effect<unknown, AcpError.AcpError>;
readonly notify: (method: string, payload: unknown) => Effect.Effect<void, AcpError.AcpError>;
/**
* Allocates the next outbound request id. Pass it to `RpcClient.make` as
* `generateRequestId` so typed calls and `request` share one counter and ids
* stay unique per connection. Ids count up from 1, so they start small enough
* for peers built on the Kotlin ACP SDK (Junie), which decode them as signed
* 32-bit integers. The counter is not bounded; only a connection that sends
* over 2^31 - 1 requests would leave that range.
*/
readonly nextRequestId: () => number;
}

interface AcpPendingRequest {
Expand Down Expand Up @@ -222,7 +231,8 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
intentionalEnd: false,
outstandingAcknowledgements: new Set(),
});
const nextRequestId = yield* Ref.make(1);
let lastRequestId = 0;
const nextRequestId = () => ++lastRequestId;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
const terminationHandled = yield* Ref.make(false);
const terminationFailure = yield* Deferred.make<never, AcpError.AcpError>();
const extPending = yield* Ref.make(new Map<string, AcpPendingRequest>());
Expand Down Expand Up @@ -946,10 +956,7 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi

const sendRequest = Effect.fn("sendRequest")(function* (method: string, payload: unknown) {
yield* ensureActive;
const requestId = yield* Ref.modify(
nextRequestId,
(current) => [current, current + 1] as const,
);
const requestId = nextRequestId();
const deferred = yield* Deferred.make<unknown, AcpError.AcpError>();
yield* Ref.update(extPending, (pending) =>
new Map(pending).set(acpRequestIdentity(requestId), { deferred, method }),
Expand All @@ -974,6 +981,7 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
},
request: sendRequest,
notify: sendNotification,
nextRequestId,
} satisfies AcpPatchedProtocol;
});

Expand Down
Loading