Bump version to 0.2.18-fork in package.json. Refactor SubprocessCLITransport to handle large JSON responses and improve JSON parsing by manually processing data chunks. Enhance tests to validate handling of large JSON messages and responses split across multiple data chunks, ensuring robustness in streaming scenarios.

Author: tobymarsden

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@botanicastudios/claude-code-sdk-ts",
-  "version": "0.2.17-fork",
+  "version": "0.2.18-fork",
   "description": "Unofficial TypeScript port of the official Python Claude Code SDK",
   "type": "module",
   "main": "dist/index.cjs",

src/_internal/transport/subprocess-cli.ts

@@ -432,49 +432,163 @@ export class SubprocessCLITransport {
       });
     }
 
-    const rl = createInterface({
-      input: this.process.stdout,
-      crlfDelay: Infinity
-    });
-
     try {
-      // Process stream-json format - each line is a JSON object
-      for await (const line of rl) {
-        const trimmedLine = line.trim();
-        if (!trimmedLine) continue;
+      // Handle large JSON responses that may exceed readline buffer limits
+      // by manually parsing JSON from raw data instead of relying on line-by-line reading
+      let buffer = '';
+      const messages: CLIOutput[] = [];
+      let processComplete = false;
+      let parseError: Error | null = null;
+
+      // Set up data handler for manual JSON parsing
+      const onData = (chunk: Buffer) => {
+        buffer += chunk.toString();
+
+        // Try to parse complete JSON objects from the buffer
+        let startIndex = 0;
+        while (startIndex < buffer.length) {
+          const openBrace = buffer.indexOf('{', startIndex);
+          const openBracket = buffer.indexOf('[', startIndex);
+
+          // Find the first JSON object/array start
+          let jsonStart = -1;
+          if (openBrace !== -1 && openBracket !== -1) {
+            jsonStart = Math.min(openBrace, openBracket);
+          } else if (openBrace !== -1) {
+            jsonStart = openBrace;
+          } else if (openBracket !== -1) {
+            jsonStart = openBracket;
+          }
 
-        this.debugLog('DEBUG stdout:', trimmedLine);
+          if (jsonStart === -1) break;
+
+          // Try to parse JSON starting from this position
+          let depth = 0;
+          let inString = false;
+          let escaped = false;
+          let jsonEnd = -1;
+
+          for (let i = jsonStart; i < buffer.length; i++) {
+            const char = buffer[i];
+
+            if (escaped) {
+              escaped = false;
+              continue;
+            }
+
+            if (char === '\\') {
+              escaped = true;
+              continue;
+            }
+
+            if (char === '"') {
+              inString = !inString;
+              continue;
+            }
+
+            if (!inString) {
+              if (char === '{' || char === '[') {
+                depth++;
+              } else if (char === '}' || char === ']') {
+                depth--;
+                if (depth === 0) {
+                  jsonEnd = i + 1;
+                  break;
+                }
+              }
+            }
+          }
+
+          if (jsonEnd !== -1) {
+            // Found complete JSON object
+            const jsonStr = buffer.substring(jsonStart, jsonEnd);
 
-        try {
-          const parsed = JSON.parse(trimmedLine) as CLIOutput;
-
-          // For non-keepAlive mode, close stdin when we receive a result message
-          // This allows the CLI process to exit gracefully after completing the response
-          if (
-            !this.keepAlive &&
-            (parsed as any).type === 'result' &&
-            this.process?.stdin &&
-            !this.process.stdin.destroyed
-          ) {
             this.debugLog(
-              'DEBUG: [Transport] Received result message, closing stdin for non-keepAlive mode'
+              'DEBUG stdout:',
+              jsonStr.substring(0, 200) + (jsonStr.length > 200 ? '...' : '')
             );
-            this.process.stdin.end();
+
+            try {
+              const parsed = JSON.parse(jsonStr) as CLIOutput;
+
+              // For non-keepAlive mode, close stdin when we receive a result message
+              if (
+                !this.keepAlive &&
+                (parsed as any).type === 'result' &&
+                this.process?.stdin &&
+                !this.process.stdin.destroyed
+              ) {
+                this.debugLog(
+                  'DEBUG: [Transport] Received result message, closing stdin for non-keepAlive mode'
+                );
+                this.process.stdin.end();
+              }
+
+              messages.push(parsed);
+            } catch (error) {
+              // If JSON parsing fails but it looks like JSON, capture error
+              if (
+                jsonStr.trim().startsWith('{') ||
+                jsonStr.trim().startsWith('[')
+              ) {
+                parseError = new CLIJSONDecodeError(
+                  `Failed to parse CLI output: ${error}`,
+                  jsonStr
+                );
+                return; // Stop processing more data
+              }
+              this.debugLog(
+                'DEBUG: Skipping non-JSON data:',
+                jsonStr.substring(0, 100)
+              );
+            }
+
+            // Remove processed JSON from buffer
+            buffer = buffer.substring(jsonEnd);
+            startIndex = 0;
+          } else {
+            // No complete JSON found, wait for more data
+            break;
           }
+        }
+      };
 
-          yield parsed;
-        } catch (error) {
-          // Skip non-JSON lines (like Python SDK does)
-          if (trimmedLine.startsWith('{') || trimmedLine.startsWith('[')) {
-            throw new CLIJSONDecodeError(
-              `Failed to parse CLI output: ${error}`,
-              trimmedLine
-            );
+      const onEnd = () => {
+        processComplete = true;
+      };
+
+      this.process.stdout.on('data', onData);
+      this.process.stdout.on('end', onEnd);
+
+      // Yield messages as they become available
+      let messageIndex = 0;
+      while (!processComplete || messageIndex < messages.length) {
+        // Check for parse errors first
+        if (parseError) {
+          throw parseError;
+        }
+
+        if (messageIndex < messages.length) {
+          const message = messages[messageIndex];
+          if (message) {
+            yield message;
           }
-          continue;
+          messageIndex++;
+        } else {
+          // Wait a bit for more messages
+          await new Promise((resolve) => setTimeout(resolve, 10));
         }
       }
 
+      // Clean up event listeners
+      this.process.stdout.removeListener('data', onData);
+      this.process.stdout.removeListener('end', onEnd);
+
+      // Final check for parse errors
+      if (parseError) {
+        throw parseError;
+      }
+
       // After all messages are processed, wait for process to exit
       try {
         await this.process;
@@ -514,8 +628,6 @@ export class SubprocessCLITransport {
       // Ensure cleanup on any error
       await this.cleanup();
       throw error;
-    } finally {
-      rl.close();
     }
   }
 

tests/subprocess-cli.test.ts

@@ -26,12 +26,17 @@ describe('SubprocessCLITransport', () => {
     const stdinStream = new Readable({ read() {} });
     (stdinStream as any).write = vi.fn();
     (stdinStream as any).end = vi.fn();
+    (stdinStream as any).destroyed = false;
 
     mockProcess = {
       stdout: stdoutStream,
       stderr: new Readable({ read() {} }),
       stdin: stdinStream,
       cancel: vi.fn(),
+      killed: false,
+      exitCode: null,
+      on: vi.fn(),
+      removeListener: vi.fn(),
       then: vi.fn((onfulfilled) => {
         // Simulate successful process completion
         if (onfulfilled) onfulfilled({ exitCode: 0 });
@@ -89,7 +94,7 @@ describe('SubprocessCLITransport', () => {
 
       expect(execa).toHaveBeenCalledWith(
         '/usr/local/bin/claude-code',
-        ['test prompt', '--output-format', 'json'],
+        ['--output-format', 'stream-json', '--verbose', '--print'],
         expect.any(Object)
       );
     });
@@ -352,7 +357,8 @@ describe('SubprocessCLITransport', () => {
       await transport.connect();
 
       setTimeout(() => {
-        stdoutStream.push('invalid json\n');
+        // Push invalid JSON that looks like JSON (starts with {) to trigger the error
+        stdoutStream.push('{"invalid": json}');
         stdoutStream.push(null);
       }, 10);
 
@@ -396,6 +402,94 @@ describe('SubprocessCLITransport', () => {
       // Skip this test for now - it's complex to test async stream + promise rejection timing
       // The actual functionality is tested in integration tests
     });
+
+    it('should handle large JSON responses exceeding 8000 bytes', async () => {
+      vi.mocked(which as any).mockResolvedValue('/usr/local/bin/claude-code');
+      vi.mocked(execa).mockReturnValue(mockProcess as any);
+
+      const transport = new SubprocessCLITransport('test prompt');
+      await transport.connect();
+
+      // Create a large JSON object that exceeds 8000 bytes
+      const largeContent = 'A'.repeat(8500); // Create content > 8000 bytes
+      const largeMessage = {
+        type: 'message',
+        data: {
+          type: 'assistant',
+          content: [{ type: 'text', text: largeContent }]
+        }
+      };
+
+      const largeJsonString = JSON.stringify(largeMessage);
+      expect(largeJsonString.length).toBeGreaterThan(8000); // Verify it's actually large
+
+      const messages = [largeMessage, { type: 'end' }];
+
+      // Emit the large JSON message to stdout
+      setTimeout(() => {
+        // Send the large JSON as a single chunk (no newlines, simulating the truncation scenario)
+        stdoutStream.push(largeJsonString);
+        stdoutStream.push(JSON.stringify({ type: 'end' }));
+        stdoutStream.push(null); // End stream
+      }, 10);
+
+      const received = [];
+      for await (const msg of transport.receiveMessages()) {
+        received.push(msg);
+      }
+
+      expect(received).toHaveLength(2);
+      expect(received[0]).toEqual(largeMessage);
+      expect(received[1]).toEqual({ type: 'end' });
+
+      // Verify the large content was received intact
+      expect((received[0] as any).data.content[0].text).toEqual(largeContent);
+      expect((received[0] as any).data.content[0].text.length).toBe(8500);
+    });
+
+    it('should handle JSON responses split across multiple data chunks', async () => {
+      vi.mocked(which as any).mockResolvedValue('/usr/local/bin/claude-code');
+      vi.mocked(execa).mockReturnValue(mockProcess as any);
+
+      const transport = new SubprocessCLITransport('test prompt');
+      await transport.connect();
+
+      // Create a large JSON that will be split across chunks
+      const largeContent = 'B'.repeat(9000);
+      const largeMessage = {
+        type: 'message',
+        data: {
+          type: 'assistant',
+          content: [{ type: 'text', text: largeContent }]
+        }
+      };
+
+      const largeJsonString = JSON.stringify(largeMessage);
+      const midpoint = Math.floor(largeJsonString.length / 2);
+
+      // Split the JSON into two chunks to simulate data arriving in parts
+      const chunk1 = largeJsonString.substring(0, midpoint);
+      const chunk2 = largeJsonString.substring(midpoint);
+
+      setTimeout(() => {
+        // Send JSON split across multiple chunks
+        stdoutStream.push(chunk1);
+        setTimeout(() => {
+          stdoutStream.push(chunk2);
+          stdoutStream.push(JSON.stringify({ type: 'end' }));
+          stdoutStream.push(null);
+        }, 5);
+      }, 10);
+
+      const received = [];
+      for await (const msg of transport.receiveMessages()) {
+        received.push(msg);
+      }
+
+      expect(received).toHaveLength(2);
+      expect(received[0]).toEqual(largeMessage);
+      expect((received[0] as any).data.content[0].text.length).toBe(9000);
+    });
   });
 
   describe('disconnect', () => {