From 2e7ab61c929b3708570290fc50ddd8081f67ab90 Mon Sep 17 00:00:00 2001 From: Shinsuke Kagawa Date: Sun, 9 Aug 2026 06:51:01 +0900 Subject: [PATCH 1/2] fix: make stream parsing backend-aware --- .../e2e/critical-user-journeys.test.ts | 8 +- .../RunAgentTool.integration.test.ts | 31 ++ .../integration/execution.integration.test.ts | 4 +- src/__tests__/security/validation.test.ts | 2 +- src/execution/AgentExecutor.ts | 49 ++- src/execution/StreamProcessor.ts | 375 +++++++++--------- src/execution/__tests__/AgentExecutor.test.ts | 43 +- .../__tests__/StreamProcessor.test.ts | 107 +++-- src/tools/RunAgentTool.ts | 7 +- 9 files changed, 373 insertions(+), 253 deletions(-) diff --git a/src/__tests__/e2e/critical-user-journeys.test.ts b/src/__tests__/e2e/critical-user-journeys.test.ts index 3a8164c..fc5c7b1 100644 --- a/src/__tests__/e2e/critical-user-journeys.test.ts +++ b/src/__tests__/e2e/critical-user-journeys.test.ts @@ -52,11 +52,13 @@ describe('Critical User Journeys - E2E Tests', () => { if (event === 'data') { // Synchronous for test stability if (isTestAgent) { - callback(Buffer.from('{"type": "result", "data": "E2E test successful"}\n')) + callback(Buffer.from('{"type": "result", "result": "E2E test successful"}\n')) } else if (isPerformanceAgent) { - callback(Buffer.from('{"type": "result", "data": "Performance test complete"}\n')) + callback(Buffer.from('{"type": "result", "result": "Performance test complete"}\n')) } else { - callback(Buffer.from('{"type": "result", "data": "Agent executed successfully"}\n')) + callback( + Buffer.from('{"type": "result", "result": "Agent executed successfully"}\n') + ) } } }), diff --git a/src/__tests__/integration/RunAgentTool.integration.test.ts b/src/__tests__/integration/RunAgentTool.integration.test.ts index dcdd72b..b2cbc87 100644 --- a/src/__tests__/integration/RunAgentTool.integration.test.ts +++ b/src/__tests__/integration/RunAgentTool.integration.test.ts @@ -363,6 +363,37 @@ describe('RunAgentTool', () => { expect(parsedContent.result).not.toContain('{"type":"error"}') }) + it('should not report an agent error as a successful structured result', async () => { + const params = { + agent: 'failing-agent', + prompt: 'Test prompt', + cwd: process.cwd(), + } + + vi.spyOn(mockAgentExecutor, 'executeAgent').mockResolvedValue({ + stdout: '{"type":"result","is_error":true}', + stderr: '', + exitCode: 143, + executionTime: 50, + hasResult: true, + resultJson: { + type: 'result', + subtype: 'error_during_execution', + is_error: true, + status: 'error', + error: 'Authentication required', + }, + }) + + const result = await runAgentTool.execute(params) + + expect(result.isError).toBe(true) + expect(result.structuredContent).toMatchObject({ + result: 'Authentication required', + status: 'error', + }) + }) + it('should include execution metadata in response', async () => { const params = { agent: 'test-agent', diff --git a/src/__tests__/integration/execution.integration.test.ts b/src/__tests__/integration/execution.integration.test.ts index 2ac3cbf..ac00d2e 100644 --- a/src/__tests__/integration/execution.integration.test.ts +++ b/src/__tests__/integration/execution.integration.test.ts @@ -41,7 +41,7 @@ describe('AgentExecutor Integration', () => { Buffer.from( `${JSON.stringify({ type: 'result', - data: 'Integration test execution success', + result: 'Integration test execution success', })}\n` ) ) @@ -53,7 +53,7 @@ describe('AgentExecutor Integration', () => { Buffer.from( `${JSON.stringify({ type: 'result', - data: 'Default integration execution', + result: 'Default integration execution', })}\n` ) ) diff --git a/src/__tests__/security/validation.test.ts b/src/__tests__/security/validation.test.ts index 0c10a2b..8884f76 100644 --- a/src/__tests__/security/validation.test.ts +++ b/src/__tests__/security/validation.test.ts @@ -54,7 +54,7 @@ describe('Security Validation Tests', () => { Buffer.from( `${JSON.stringify({ type: 'result', - data: 'Security test execution', + result: 'Security test execution', })}\n` ) ) diff --git a/src/execution/AgentExecutor.ts b/src/execution/AgentExecutor.ts index a7cf94d..e5280a2 100644 --- a/src/execution/AgentExecutor.ts +++ b/src/execution/AgentExecutor.ts @@ -808,10 +808,10 @@ export class AgentExecutor { return } - const streamProcessor = new StreamProcessor() - let stdout = '' - let stderr = '' - let stdoutBuffer = '' + const streamProcessor = new StreamProcessor(this.config.agentType) + const stdoutParts: string[] = [] + const stderrParts: string[] = [] + let stdoutLineParts: string[] = [] const stdoutDecoder = new StringDecoder('utf8') const stderrDecoder = new StringDecoder('utf8') let stdoutTruncated = false @@ -843,17 +843,21 @@ export class AgentExecutor { if (!stdoutTruncated) { const tail = stdoutDecoder.end() - stdout += tail - stdoutBuffer += tail + stdoutParts.push(tail) + stdoutLineParts.push(tail) } if (!stderrTruncated) { - stderr += stderrDecoder.end() + stderrParts.push(stderrDecoder.end()) } - if (stdoutBuffer.trim()) { - streamProcessor.processLine(stdoutBuffer) - stdoutBuffer = '' + const trailingLine = stdoutLineParts.join('') + if (trailingLine.trim()) { + streamProcessor.processLine(trailingLine) } + stdoutLineParts = [] + + const stdout = stdoutParts.join('') + const stderr = stderrParts.join('') let result = streamProcessor.getResult() if (result === null) { @@ -925,13 +929,20 @@ export class AgentExecutor { const chunk = captureChunk(data, stdoutDecoder, () => { stdoutTruncated = true }) - stdout += chunk - stdoutBuffer += chunk + stdoutParts.push(chunk) - const lines = stdoutBuffer.split('\n') - stdoutBuffer = lines.pop() || '' + let chunkOffset = 0 + while (chunkOffset < chunk.length) { + const newlineIndex = chunk.indexOf('\n', chunkOffset) + if (newlineIndex < 0) { + stdoutLineParts.push(chunk.slice(chunkOffset)) + break + } - for (const line of lines) { + stdoutLineParts.push(chunk.slice(chunkOffset, newlineIndex)) + const line = stdoutLineParts.join('') + stdoutLineParts = [] + chunkOffset = newlineIndex + 1 if (streamProcessor.processLine(line)) { requestTermination() break @@ -940,9 +951,11 @@ export class AgentExecutor { }) childProcess.stderr?.on('data', (data: Buffer) => { - stderr += captureChunk(data, stderrDecoder, () => { - stderrTruncated = true - }) + stderrParts.push( + captureChunk(data, stderrDecoder, () => { + stderrTruncated = true + }) + ) }) childProcess.on('close', (code: number | null, signal?: NodeJS.Signals | null) => { diff --git a/src/execution/StreamProcessor.ts b/src/execution/StreamProcessor.ts index 4f4cb59..d4a7528 100644 --- a/src/execution/StreamProcessor.ts +++ b/src/execution/StreamProcessor.ts @@ -1,229 +1,235 @@ +import type { AgentType } from './AgentExecutor.js' + /** - * StreamProcessor - Simplified stream processing for agent output - * - * Handles cursor, Claude-compatible, gemini, codex, grok, and OpenCode output in JSON format. - * - Cursor/Claude-compatible: Return JSON events ending with type: "result" - * - Gemini: Uses --output-format stream-json, returns multiple JSON lines, - * assistant messages contain the response, type: "result" signals completion - * - Codex: Uses --json flag with exec subcommand, returns stream of JSON events, - * agent_message items contain the response, turn.completed signals completion - * - Grok: Uses --output-format json, returns a complete JSON object after exit - * - OpenCode: Uses --format json, returns step and text events as NDJSON + * Parses one backend's documented output protocol and stores its terminal result. */ export class StreamProcessor { private resultJson: unknown = null private geminiResponseParts: string[] = [] - private isGeminiStreamJson = false - private isCodexFormat = false private codexAgentMessages: string[] = [] - private codexUsage: unknown = null - private isOpenCodeFormat = false private openCodeResponseParts: string[] = [] + constructor(private readonly agentType: AgentType) {} + /** - * Process a single line from the agent output stream. - * Returns true when a valid result JSON is detected, false otherwise. - * - * For Cursor/Claude-compatible: The first JSON line with type: "result" is the result. - * For Gemini stream-json: Accumulate assistant messages, return when type: "result" is seen. - * For Codex: Accumulate agent_message items, return when turn.completed is seen. - * - * @param line - Raw line from stdout - * @returns true if processing is complete, false to continue + * Process one line from stdout. Returns true only for a terminal event from + * the configured backend. */ processLine(line: string): boolean { const trimmedLine = line.trim() - - // Empty lines are ignored - if (!trimmedLine) { + if (!trimmedLine || this.resultJson !== null) { return false } - // If we already have a result, ignore subsequent lines - if (this.resultJson !== null) { + let json: unknown + try { + json = JSON.parse(trimmedLine) as unknown + } catch { return false } - // Try to parse as JSON - try { - const json = JSON.parse(trimmedLine) as Record + if (!this.isRecord(json)) { + return false + } - // Detect Gemini stream-json format by init message - if (json['type'] === 'init') { - this.isGeminiStreamJson = true - return false + const agentType = this.agentType + switch (agentType) { + case 'cursor': + case 'claude': + case 'glm': + case 'kimi': + return this.processClaudeCompatibleLine(json) + case 'gemini': + return this.processGeminiLine(json) + case 'codex': + return this.processCodexLine(json) + case 'grok': + return this.processGrokLine(json) + case 'opencode': + return this.processOpenCodeLine(json) + default: { + const unsupportedAgentType: never = agentType + throw new Error(`Unsupported agent type: ${String(unsupportedAgentType)}`) } + } + } - // Detect Codex format by thread.started message - if (json['type'] === 'thread.started') { - this.isCodexFormat = true - return false - } + /** + * Process a complete non-NDJSON payload after process exit. + */ + processCompleteOutput(output: string): boolean { + if (this.resultJson !== null) { + return false + } - const part = json['part'] - if ( - ['step_start', 'tool_use', 'text', 'step_finish'].includes(String(json['type'])) && - this.isRecord(part) - ) { - this.isOpenCodeFormat = true + if (this.agentType === 'opencode' && this.openCodeResponseParts.length > 0) { + this.resultJson = { + type: 'result', + result: this.openCodeResponseParts.join(''), + status: 'partial', + stop_reason: 'process-exit', } + return true + } - const normalizedError = this.normalizeFatalError(json) - if (normalizedError) { - this.resultJson = normalizedError - return true - } + if (this.agentType !== 'grok') { + return false + } - if (this.isOpenCodeFormat && json['type'] === 'text' && this.isRecord(part)) { - if (typeof part['text'] === 'string') { - this.openCodeResponseParts.push(part['text']) - } - return false - } + try { + const json = JSON.parse(output.trim()) as unknown + return this.isRecord(json) && this.processGrokLine(json) + } catch { + return false + } + } - if (this.isOpenCodeFormat && json['type'] === 'step_finish' && this.isRecord(part)) { - const reason = part['reason'] - if (reason === 'tool-calls' || reason === undefined || reason === null) { - return false - } - this.resultJson = { - type: 'result', - result: this.openCodeResponseParts.join(''), - status: reason === 'stop' ? 'success' : 'partial', - stop_reason: reason, - } - return true - } + private processClaudeCompatibleLine(json: Record): boolean { + if (json['type'] !== 'result') { + return false + } - // For Gemini: accumulate assistant message content - if ( - this.isGeminiStreamJson && - json['type'] === 'message' && - json['role'] === 'assistant' && - typeof json['content'] === 'string' - ) { - this.geminiResponseParts.push(json['content']) - return false + const subtype = json['subtype'] + const isError = + json['is_error'] === true || + json['status'] === 'error' || + (typeof subtype === 'string' && subtype.startsWith('error_')) + + if (isError) { + const normalizedError = this.normalizeError(json) + const errorMessage = + (typeof json['error'] === 'string' && json['error']) || + (typeof json['result'] === 'string' && json['result']) || + normalizedError['error'] + this.resultJson = { + ...json, + ...normalizedError, + subtype: typeof subtype === 'string' ? subtype : 'error', + status: 'error', + error: errorMessage, } + return true + } - // For Codex: accumulate agent_message content from item.completed events - if (this.isCodexFormat && json['type'] === 'item.completed') { - const item = json['item'] - if ( - this.isCodexItem(item) && - item['type'] === 'agent_message' && - typeof item['text'] === 'string' - ) { - this.codexAgentMessages.push(item['text']) - } - return false - } + if (typeof json['result'] !== 'string') { + return false + } - // For Codex: turn.completed signals end of response - if (this.isCodexFormat && json['type'] === 'turn.completed') { - this.codexUsage = json['usage'] - this.resultJson = { - type: 'result', - result: this.codexAgentMessages.join('\n'), - usage: this.codexUsage, - status: 'success', - } - return true // Processing complete - } + this.resultJson = json + return true + } - // Check if this is a result JSON - if (json['type'] === 'result') { - // For Gemini: construct result with accumulated response - if (this.isGeminiStreamJson) { - this.resultJson = { - type: 'result', - result: this.geminiResponseParts.join(''), - stats: json['stats'], - status: json['status'], - } - } else { - // Cursor/Claude-compatible: use as-is - this.resultJson = json - } - return true // Processing complete - } + private processGeminiLine(json: Record): boolean { + if ( + json['type'] === 'message' && + json['role'] === 'assistant' && + typeof json['content'] === 'string' + ) { + this.geminiResponseParts.push(json['content']) + return false + } - const normalizedCompleteOutput = this.normalizeCompleteOutput(json) - if (normalizedCompleteOutput) { - this.resultJson = normalizedCompleteOutput - return true - } + if (json['type'] !== 'result') { + return false + } - // For backwards compatibility: store first valid JSON if no type field - // This handles any CLI that doesn't use the type field - if (!('type' in json)) { - this.resultJson = json - return true + if (json['status'] === 'error') { + this.resultJson = this.normalizeError(json) + return true + } + + this.resultJson = { + type: 'result', + result: this.geminiResponseParts.join(''), + stats: json['stats'], + status: json['status'], + } + return true + } + + private processCodexLine(json: Record): boolean { + if (json['type'] === 'error' || json['type'] === 'turn.failed') { + this.resultJson = this.normalizeError(json) + return true + } + + if (json['type'] === 'item.completed') { + const item = json['item'] + if ( + this.isRecord(item) && + item['type'] === 'agent_message' && + typeof item['text'] === 'string' + ) { + this.codexAgentMessages.push(item['text']) } + return false + } - return false // Continue processing (not a result type) - } catch { - // Not valid JSON, ignore + if (json['type'] !== 'turn.completed') { return false } + + this.resultJson = { + type: 'result', + result: this.codexAgentMessages.join('\n'), + usage: json['usage'], + status: 'success', + } + return true } - /** - * Process a complete non-NDJSON payload after process exit. - * - * Grok's `--output-format json` can emit a pretty-printed JSON object, which - * cannot be parsed by the line-oriented stream path. - * - * @param output - Complete stdout captured from the agent process - * @returns true if processing is complete, false otherwise - */ - processCompleteOutput(output: string): boolean { - if (this.resultJson !== null) { + private processGrokLine(json: Record): boolean { + if (json['type'] === 'error') { + this.resultJson = this.normalizeError(json) + return true + } + + const result = this.normalizeGrokOutput(json) + if (!result) { return false } - try { - const json = JSON.parse(output.trim()) as unknown - if (this.isRecord(json)) { - const normalizedError = this.normalizeFatalError(json) - if (normalizedError) { - this.resultJson = normalizedError - return true - } - - const normalizedCompleteOutput = this.normalizeCompleteOutput(json) - if (normalizedCompleteOutput) { - this.resultJson = normalizedCompleteOutput - return true - } - } - } catch { - // NDJSON streams are expected to fail whole-output JSON parsing. + this.resultJson = result + return true + } + + private processOpenCodeLine(json: Record): boolean { + if (json['type'] === 'error') { + this.resultJson = this.normalizeError(json) + return true } - if (this.isOpenCodeFormat && this.openCodeResponseParts.length > 0) { - this.resultJson = { - type: 'result', - result: this.openCodeResponseParts.join(''), - status: 'partial', - stop_reason: 'process-exit', + const part = json['part'] + if (!this.isRecord(part)) { + return false + } + + if (json['type'] === 'text') { + if (typeof part['text'] === 'string') { + this.openCodeResponseParts.push(part['text']) } - return true + return false } - return false - } + if (json['type'] !== 'step_finish') { + return false + } - private normalizeFatalError(json: Record): Record | null { - const isFatalEvent = - json['type'] === 'error' || - json['type'] === 'turn.failed' || - (json['type'] === 'result' && json['status'] === 'error') + const reason = part['reason'] + if (reason === 'tool-calls' || reason === undefined || reason === null) { + return false + } - if (!isFatalEvent) { - return null + this.resultJson = { + type: 'result', + result: this.openCodeResponseParts.join(''), + status: reason === 'stop' ? 'success' : 'partial', + stop_reason: reason, } + return true + } + private normalizeError(json: Record): Record { const error = this.isRecord(json['error']) ? json['error'] : undefined const errorData = error && this.isRecord(error['data']) ? error['data'] : undefined const message = @@ -231,6 +237,8 @@ export class StreamProcessor { (error && typeof error['message'] === 'string' && error['message']) || (errorData && typeof errorData['message'] === 'string' && errorData['message']) || (typeof json['error'] === 'string' && json['error']) || + (typeof json['result'] === 'string' && json['result']) || + (typeof json['subtype'] === 'string' && json['subtype']) || 'Agent execution failed' const errorType = (error && typeof error['name'] === 'string' && error['name']) || @@ -262,8 +270,8 @@ export class StreamProcessor { } } - private normalizeCompleteOutput(json: Record): Record | null { - if (typeof json['text'] !== 'string' || !('stopReason' in json)) { + private normalizeGrokOutput(json: Record): Record | null { + if (typeof json['text'] !== 'string') { return null } @@ -286,23 +294,10 @@ export class StreamProcessor { return result } - /** - * Type guard for Codex item structure - * @param item - The item to check - * @returns true if item is a valid Codex item object - */ - private isCodexItem(item: unknown): item is Record { - return this.isRecord(item) - } - private isRecord(value: unknown): value is Record { return typeof value === 'object' && value !== null && !Array.isArray(value) } - /** - * Get the final result JSON. - * @returns The stored result JSON or null if not yet available - */ getResult(): unknown { return this.resultJson } diff --git a/src/execution/__tests__/AgentExecutor.test.ts b/src/execution/__tests__/AgentExecutor.test.ts index 022ddea..3763e91 100644 --- a/src/execution/__tests__/AgentExecutor.test.ts +++ b/src/execution/__tests__/AgentExecutor.test.ts @@ -92,7 +92,7 @@ function createMockProcess(options: { /** Creates a success mock process that emits a JSON result */ function createSuccessMock(data = 'Test execution successful') { return createMockProcess({ - stdoutData: `${JSON.stringify({ type: 'result', data })}\n`, + stdoutData: `${JSON.stringify({ type: 'result', result: data })}\n`, }) } @@ -845,7 +845,7 @@ describe('AgentExecutor', () => { expect(result.hasResult).toBe(true) expect(result.resultJson).toEqual({ type: 'result', - data: 'Test execution successful', + result: 'Test execution successful', }) expect(result.executionTime).toBeGreaterThan(0) }) @@ -925,6 +925,29 @@ describe('AgentExecutor', () => { }) }) + it('should ignore a non-terminal Claude event and keep reading until the result', async () => { + mockSpawn.mockImplementationOnce(() => + createMockProcess({ + stdoutData: + '{"message":"Background operation completed"}\n' + + '{"type":"result","result":"actual response"}\n', + }) + ) + const executor = new AgentExecutor(createExecutionConfig('claude')) + + const result = await executor.executeAgent({ + agent: 'test-agent', + prompt: 'Help me', + cwd: '/tmp', + }) + + expect(result.hasResult).toBe(true) + expect(result.resultJson).toMatchObject({ + type: 'result', + result: 'actual response', + }) + }) + it('should return partial OpenCode text when the finish event is missing', async () => { mockSpawn.mockImplementationOnce(() => createMockProcess({ @@ -1221,7 +1244,7 @@ describe('AgentExecutor', () => { it('should handle SIGTERM (exit code 143) as normal when hasResult is true', async () => { mockSpawn.mockImplementationOnce(() => createMockProcess({ - stdoutData: '{"type": "result", "data": "Success"}\n', + stdoutData: '{"type": "result", "result": "Success"}\n', exitCode: 143, }) ) @@ -1242,7 +1265,7 @@ describe('AgentExecutor', () => { it('should preserve a result when SIGTERM escalates to SIGKILL', async () => { mockSpawn.mockImplementationOnce(() => createMockProcess({ - stdoutData: '{"type": "result", "data": "Success"}\n', + stdoutData: '{"type": "result", "result": "Success"}\n', noClose: true, }) ) @@ -1263,7 +1286,7 @@ describe('AgentExecutor', () => { it('should distinguish timeout with partial result from complete timeout', async () => { mockSpawn.mockImplementationOnce(() => createMockProcess({ - stdoutData: '{"type": "result", "partial": true}\n', + stdoutData: '{"type": "result", "result": "Partial", "status": "partial"}\n', stdoutDelay: 50, noClose: true, // Let timeout handler fire }) @@ -1279,7 +1302,11 @@ describe('AgentExecutor', () => { expect(result.exitCode).toBe(124) expect(result.hasResult).toBe(true) - expect(result.resultJson).toEqual({ type: 'result', partial: true }) + expect(result.resultJson).toEqual({ + type: 'result', + result: 'Partial', + status: 'partial', + }) }) }) @@ -1305,7 +1332,7 @@ describe('AgentExecutor', () => { }) it('should decode UTF-8 characters split across stdout chunks', async () => { - const output = Buffer.from('{"type":"result","data":"日本語"}\n') + const output = Buffer.from('{"type":"result","result":"日本語"}\n') const splitAt = output.indexOf(Buffer.from('日')) + 1 mockSpawn.mockImplementationOnce(() => createMockProcess({ @@ -1320,7 +1347,7 @@ describe('AgentExecutor', () => { cwd: '/tmp', }) - expect(result.resultJson).toEqual({ type: 'result', data: '日本語' }) + expect(result.resultJson).toEqual({ type: 'result', result: '日本語' }) }) it('should not emit a replacement character when the byte cap splits UTF-8', async () => { diff --git a/src/execution/__tests__/StreamProcessor.test.ts b/src/execution/__tests__/StreamProcessor.test.ts index 2afed07..e14d09b 100644 --- a/src/execution/__tests__/StreamProcessor.test.ts +++ b/src/execution/__tests__/StreamProcessor.test.ts @@ -4,24 +4,24 @@ describe('StreamProcessor', () => { let processor: StreamProcessor beforeEach(() => { - processor = new StreamProcessor() + processor = new StreamProcessor('cursor') }) describe('JSON processing', () => { it('should detect and store the first valid JSON', () => { - const json = '{"type": "result", "data": "output", "status": "complete"}' + const json = '{"type": "result", "result": "output", "status": "complete"}' expect(processor.processLine(json)).toBe(true) expect(processor.getResult()).toEqual({ type: 'result', - data: 'output', + result: 'output', status: 'complete', }) }) it('should ignore subsequent JSONs after finding result type', () => { - const json1 = '{"type": "result", "response": "First JSON", "status": "success"}' - const json2 = '{"type": "result", "response": "Second JSON", "status": "also success"}' + const json1 = '{"type": "result", "result": "First JSON", "status": "success"}' + const json2 = '{"type": "result", "result": "Second JSON", "status": "also success"}' expect(processor.processLine(json1)).toBe(true) expect(processor.processLine(json2)).toBe(false) @@ -29,7 +29,7 @@ describe('StreamProcessor', () => { // Should still have the first JSON expect(processor.getResult()).toEqual({ type: 'result', - response: 'First JSON', + result: 'First JSON', status: 'success', }) }) @@ -59,6 +59,7 @@ describe('StreamProcessor', () => { type: 'result', subtype: 'error', is_error: true, + status: 'error', duration_ms: 1234, error: 'An error occurred', error_type: 'execution_error', @@ -68,6 +69,7 @@ describe('StreamProcessor', () => { }) it('should handle claude JSON format', () => { + processor = new StreamProcessor('claude') const claudeJson = '{"type":"result","subtype":"success","is_error":false,"duration_ms":2856,"result":"Hi!","session_id":"711419a4-3a19-4448-aa4a-31de7c4fa7a5"}' @@ -83,6 +85,7 @@ describe('StreamProcessor', () => { }) it('should handle gemini stream-json format by accumulating assistant messages', () => { + processor = new StreamProcessor('gemini') // Gemini stream-json outputs multiple JSON lines: // - init: signals stream-json mode // - message with role: "user": user prompt (ignored) @@ -117,17 +120,43 @@ describe('StreamProcessor', () => { }) }) - it('should handle JSON without type field for backwards compatibility', () => { + it('should not treat typeless cursor JSON as terminal', () => { const legacyJson = '{"response": "Legacy output", "status": "complete"}' - expect(processor.processLine(legacyJson)).toBe(true) - expect(processor.getResult()).toEqual({ - response: 'Legacy output', - status: 'complete', + expect(processor.processLine(legacyJson)).toBe(false) + expect(processor.getResult()).toBeNull() + }) + + it('should not apply Grok result parsing to Claude events', () => { + processor = new StreamProcessor('claude') + + expect( + processor.processLine( + '{"type":"system","text":"Background operation completed","stopReason":"EndTurn"}' + ) + ).toBe(false) + expect(processor.getResult()).toBeNull() + }) + + it('should preserve Claude terminal errors as errors', () => { + processor = new StreamProcessor('claude') + + expect( + processor.processLine( + '{"type":"result","subtype":"error_during_execution","is_error":true,"result":"Authentication required"}' + ) + ).toBe(true) + expect(processor.getResult()).toMatchObject({ + type: 'result', + subtype: 'error_during_execution', + is_error: true, + status: 'error', + error: 'Authentication required', }) }) it('should normalize single-line grok JSON output', () => { + processor = new StreamProcessor('grok') const grokJson = '{"text":"Grok output","stopReason":"EndTurn","sessionId":"session-123","requestId":"request-123"}' @@ -142,7 +171,19 @@ describe('StreamProcessor', () => { }) }) + it('should mark Grok text without a stop reason as partial', () => { + processor = new StreamProcessor('grok') + + expect(processor.processLine('{"text":"Work in progress"}')).toBe(true) + expect(processor.getResult()).toEqual({ + type: 'result', + result: 'Work in progress', + status: 'partial', + }) + }) + it('should extract agent_message text from codex output stream', () => { + processor = new StreamProcessor('codex') // Given: A complete Codex output stream with reasoning and agent_message const codexOutputStream = [ '{"type":"thread.started","thread_id":"019b1291-a763-74a1-bffe-39670dad4b6b"}', @@ -167,6 +208,7 @@ describe('StreamProcessor', () => { }) it('should collect OpenCode text across tool steps until stop', () => { + processor = new StreamProcessor('opencode') expect(processor.processLine('{"type":"step_start","part":{}}')).toBe(false) expect(processor.processLine('{"type":"text","part":{"text":"part 1"}}')).toBe(false) expect(processor.processLine('{"type":"step_finish","part":{"reason":"tool-calls"}}')).toBe( @@ -184,6 +226,7 @@ describe('StreamProcessor', () => { }) it('should mark a non-stop OpenCode finish as partial', () => { + processor = new StreamProcessor('opencode') expect(processor.processLine('{"type":"text","part":{"text":"truncated"}}')).toBe(false) expect(processor.processLine('{"type":"step_finish","part":{"reason":"length"}}')).toBe(true) @@ -196,6 +239,7 @@ describe('StreamProcessor', () => { }) it('should preserve OpenCode text as partial when the process exits without a finish event', () => { + processor = new StreamProcessor('opencode') expect(processor.processLine('{"type":"text","part":{"text":"work in progress"}}')).toBe( false ) @@ -212,6 +256,7 @@ describe('StreamProcessor', () => { }) it('should normalize OpenCode error events with diagnostic context', () => { + processor = new StreamProcessor('opencode') const openCodeError = '{"type":"error","timestamp":1784020603139,"sessionID":"ses_0a015ef55ffePMBqnSUUKOWwRG","error":{"name":"UnknownError","data":{"message":"Unexpected server error. Check server logs for details.","ref":"err_c481aeec"}}}' @@ -229,6 +274,7 @@ describe('StreamProcessor', () => { }) it('should normalize Codex turn.failed events', () => { + processor = new StreamProcessor('codex') expect( processor.processLine( '{"type":"turn.failed","error":{"message":"The model is unavailable"}}' @@ -242,7 +288,11 @@ describe('StreamProcessor', () => { }) }) - it('should normalize top-level Codex and Grok error events', () => { + it.each([ + 'codex', + 'grok', + ] as const)('should normalize top-level %s error events', (agentType) => { + processor = new StreamProcessor(agentType) expect(processor.processLine('{"type":"error","message":"Unknown model id"}')).toBe(true) expect(processor.getResult()).toEqual({ type: 'result', @@ -253,6 +303,7 @@ describe('StreamProcessor', () => { }) it('should normalize Gemini result errors', () => { + processor = new StreamProcessor('gemini') expect( processor.processLine( '{"type":"result","status":"error","error":{"type":"unknown","message":"Model not found"},"stats":{"total_tokens":0}}' @@ -269,6 +320,7 @@ describe('StreamProcessor', () => { }) it('should ignore non-fatal Codex error items and allow the turn to complete', () => { + processor = new StreamProcessor('codex') expect(processor.processLine('{"type":"thread.started","thread_id":"thread-1"}')).toBe(false) expect( processor.processLine( @@ -294,6 +346,7 @@ describe('StreamProcessor', () => { }) it('should concatenate multiple agent_messages with newlines', () => { + processor = new StreamProcessor('codex') // Given: Codex output with multiple agent_message items const codexOutputStream = [ '{"type":"thread.started","thread_id":"019b1292-47b5-7bf3-8f7a-ef0986d5b982"}', @@ -317,6 +370,7 @@ describe('StreamProcessor', () => { }) it('should ignore command_execution items and only include agent_message', () => { + processor = new StreamProcessor('codex') // Given: Codex output with command execution (reasoning, command, then summary) const codexOutputStream = [ '{"type":"thread.started","thread_id":"019b1292-e66c-7c61-bcc5-4262b08f3535"}', @@ -368,6 +422,7 @@ describe('StreamProcessor', () => { describe('Complete output handling', () => { it('should normalize pretty-printed grok JSON output after process exit', () => { + processor = new StreamProcessor('grok') expect( processor.processCompleteOutput( '{\n' + @@ -390,6 +445,7 @@ describe('StreamProcessor', () => { }) it('should mark grok non-EndTurn output as partial', () => { + processor = new StreamProcessor('grok') expect( processor.processCompleteOutput('{"text":"Progress only","stopReason":"Cancelled"}') ).toBe(true) @@ -403,6 +459,7 @@ describe('StreamProcessor', () => { }) it('should ignore complete output when it is not grok JSON', () => { + processor = new StreamProcessor('grok') expect(processor.processCompleteOutput('plain text output')).toBe(false) expect(processor.processCompleteOutput('{"response":"legacy"}')).toBe(false) expect(processor.getResult()).toBeNull() @@ -410,34 +467,28 @@ describe('StreamProcessor', () => { }) describe('Edge cases', () => { - it('should handle complex nested JSON structures', () => { + it('should ignore arbitrary structured JSON from cursor', () => { const complexJson = '{"foo": "bar", "nested": {"deep": {"value": "test"}}, "array": [1, 2, 3]}' - expect(processor.processLine(complexJson)).toBe(true) - expect(processor.getResult()).toEqual({ - foo: 'bar', - nested: { deep: { value: 'test' } }, - array: [1, 2, 3], - }) + expect(processor.processLine(complexJson)).toBe(false) + expect(processor.getResult()).toBeNull() }) - it('should handle JSON with special characters', () => { + it('should ignore typeless JSON with special characters from cursor', () => { const jsonWithSpecialChars = '{"text": "Line 1\\nLine 2\\tTabbed", "emoji": "🎉"}' - expect(processor.processLine(jsonWithSpecialChars)).toBe(true) - expect(processor.getResult()).toEqual({ - text: 'Line 1\nLine 2\tTabbed', - emoji: '🎉', - }) + expect(processor.processLine(jsonWithSpecialChars)).toBe(false) + expect(processor.getResult()).toBeNull() }) - it('should process lines with leading/trailing whitespace', () => { - const jsonWithWhitespace = ' {"data": "value"} ' + it('should process terminal lines with leading/trailing whitespace', () => { + const jsonWithWhitespace = ' {"type":"result","result":"value"} ' expect(processor.processLine(jsonWithWhitespace)).toBe(true) expect(processor.getResult()).toEqual({ - data: 'value', + type: 'result', + result: 'value', }) }) }) diff --git a/src/tools/RunAgentTool.ts b/src/tools/RunAgentTool.ts index c459180..0dbdf61 100644 --- a/src/tools/RunAgentTool.ts +++ b/src/tools/RunAgentTool.ts @@ -631,8 +631,9 @@ export class RunAgentTool { (result.exitCode === 124 && result.hasResult === true) // Timeout with partial result const isSuccess = - (!isPartialSuccess && result.exitCode === 0) || // Normal completion - (!isPartialSuccess && + (!isError && !isPartialSuccess && result.exitCode === 0) || // Normal completion + (!isError && + !isPartialSuccess && (result.exitCode === 143 || result.exitCode === 137) && result.hasResult === true) // Terminated after receiving a result @@ -643,7 +644,7 @@ export class RunAgentTool { agent: agentName, exit_code: result.exitCode, execution_time: result.executionTime, - status: isSuccess ? 'success' : isPartialSuccess ? 'partial' : 'error', + status: isError ? 'error' : isSuccess ? 'success' : isPartialSuccess ? 'partial' : 'error', ...(sessionId && { session_id: sessionId }), ...(requestId && { request_id: requestId }), } From 9208bb064c06e9b7978cb910dc674440ccc41f4f Mon Sep 17 00:00:00 2001 From: Shinsuke Kagawa Date: Sun, 9 Aug 2026 07:11:40 +0900 Subject: [PATCH 2/2] chore: bump version to 0.12.1 --- package-lock.json | 4 ++-- package.json | 2 +- server.json | 4 ++-- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/package-lock.json b/package-lock.json index 261921a..9fc6b67 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "sub-agents-mcp", - "version": "0.12.0", + "version": "0.12.1", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "sub-agents-mcp", - "version": "0.12.0", + "version": "0.12.1", "license": "MIT", "dependencies": { "@modelcontextprotocol/sdk": "^1.30.0" diff --git a/package.json b/package.json index f4c39f8..628984e 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "sub-agents-mcp", - "version": "0.12.0", + "version": "0.12.1", "mcpName": "io.github.shinpr/sub-agents-mcp", "description": "MCP server for delegating tasks to specialized AI assistants in Cursor, Claude Code, Codex, Gemini, GLM, Kimi, Grok, and OpenCode", "type": "module", diff --git a/server.json b/server.json index 6979b74..14c5034 100644 --- a/server.json +++ b/server.json @@ -8,13 +8,13 @@ "url": "https://github.com/shinpr/sub-agents-mcp", "source": "github" }, - "version": "0.12.0", + "version": "0.12.1", "packages": [ { "registryType": "npm", "registryBaseUrl": "https://registry.npmjs.org", "identifier": "sub-agents-mcp", - "version": "0.12.0", + "version": "0.12.1", "transport": { "type": "stdio" },