fix(acp): a final agent message without a trailing newline is no longer dropped

packages/effect-acp/src/protocol.test.ts

@@ -884,6 +884,7 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
         .request("x/test", { hello: "world" })
         .pipe(Effect.flip, Effect.orDie, Effect.forkScoped);
       yield* Queue.take(output);
+      yield* Queue.offer(input, encoder.encode('{"jsonrpc":"2.0","id":1,"result":'));
       yield* Queue.end(input);
 
       const error = yield* Deferred.await(termination);
@@ -900,6 +901,27 @@ it.layer(NodeServices.layer)("effect-acp protocol", (it) => {
     }),
   );
 
+  it.effect("answers a pending request from a final response without a trailing newline", () =>
+    Effect.gen(function* () {
+      const { stdio, input, output } = yield* makeInMemoryStdio();
+      const transport = yield* AcpProtocol.makeAcpPatchedProtocol({
+        stdio,
+        serverRequestMethods: new Set(),
+      });
+      const pending = yield* transport
+        .request("x/test", { hello: "world" })
+        .pipe(Effect.forkScoped);
+      yield* Queue.take(output);
+      yield* Queue.offer(
+        input,
+        encoder.encode(encodeUnknownJsonString({ jsonrpc: "2.0", id: 1, result: { ok: true } })),
+      );
+      yield* Queue.end(input);
+
+      assert.deepEqual(yield* Fiber.join(pending), { ok: true });
+    }),
+  );
+
   it.effect("propagates the real child exit code when the input stream ends", () =>
     Effect.gen(function* () {
       const handle = yield* makeHandle({ ACP_MOCK_EXIT_IMMEDIATELY_CODE: "7" });

packages/effect-acp/src/protocol.ts

@@ -180,6 +180,16 @@ const makeStrictNdJsonRpcParser = () => {
       buffer = chunk.slice(lastNewline + 1);
       return lines.flatMap((line) => codec.decode(line));
     },
+    // A truncated final line is left to the exit status, which explains a crash better.
+    end: (): ReadonlyArray<unknown> => {
+      const line = buffer + decoder.decode();
+      buffer = "";
+      try {
+        return line.trim() === "" ? [] : codec.decode(line);
+      } catch {
+        return [];
+      }
+    },
     encode: (response: Parameters<typeof codec.encode>[0]) => {
       const encoded = codec.encode(response);
       return encoded === undefined ? undefined : `${encoded}\n`;
@@ -748,6 +758,19 @@ export const makeAcpPatchedProtocol = Effect.fn("makeAcpPatchedProtocol")(functi
         ),
       ),
     ),
+    Effect.andThen(
+      Effect.suspend(() =>
+        Effect.forEach(
+          (
+            parser.end() as ReadonlyArray<
+              RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded
+            >
+          ).map(normalizeAcpJsonRpcError),
+          routeDecodedMessage,
+          { discard: true },
+        ),
+      ),
+    ),
     // Anything that ends the reader, including a defect in a callback it runs,
     // terminates the connection so pending requests fail instead of hanging.
     Effect.matchCauseEffect({