fix(executor): streaming after tool calls

[Sg312]

Summary

Fixes streaming api for agents with tool calls

Type of Change

• Bug fix

Testing

Manual

Checklist

• Code follows project style guidelines
• Self-reviewed my changes
• Tests added/updated and passing
• No new warnings introduced
• I confirm that I have read and agree to the terms outlined in the Contributor License Agreement (CLA)

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

1 Skipped Deployment

Project	Deployment	Preview	Comments	Updated (UTC)
docs	Skipped			Nov 13, 2025 6:40pm

[greptile-apps]

Greptile Overview

Greptile Summary

Fixed streaming responses to work correctly after tool call processing by removing the iterationCount > 0 condition across all LLM providers and refactoring the executor's stream handling.

Key Changes:

• Removed conditional check that prevented streaming on first response (when no tool calls occurred)
• Refactored block-executor.ts to properly handle streaming with stream.tee() for parallel consumption
• Added handleStreamingExecution() method that splits streams into client and executor copies
• Executor now consumes stream to populate execution.output.content while client receives processed stream
• Consistent fix applied across 10 providers (Anthropic, Azure OpenAI, Cerebras, DeepSeek, Groq, Mistral, Ollama, OpenAI, OpenRouter, xAI)

Impact:
Enables streaming responses for all LLM requests when stream=true, not just after tool calls. This improves user experience by providing real-time feedback even for simple queries without tool usage.

Confidence Score: 4/5

• Safe to merge with minor attention to stream handling edge cases
• The changes are well-structured and consistent across all providers. The refactoring properly handles stream tee/consumption pattern. However, the consumeExecutorStream method has potential edge cases around error handling and the assumption that streams support .tee() (though there's a fallback for when they don't).
• Pay attention to apps/sim/executor/execution/block-executor.ts - ensure the stream consumption logic handles all edge cases correctly in production

Important Files Changed

File Analysis

Filename	Score	Overview
apps/sim/executor/execution/block-executor.ts	4/5	Refactored streaming execution to handle tool calls and response format processing with proper stream tee and consumption
apps/sim/providers/anthropic/index.ts	5/5	Removed iterationCount > 0 condition to enable streaming even on first response
apps/sim/providers/openai/index.ts	5/5	Removed iterationCount > 0 condition to enable streaming even on first response

Sequence Diagram

sequenceDiagram
    participant Executor as BlockExecutor
    participant Handler as Block Handler
    participant Provider as LLM Provider
    participant StreamProc as StreamingResponseFormatProcessor
    participant Client as onStream Callback

    Note over Executor,Client: Streaming After Tool Calls Flow

    Executor->>Handler: executeWithNode(ctx, block, inputs)
    Handler->>Provider: execute LLM request with stream=true
    
    alt Tool calls required
        Provider->>Handler: response with tool_calls
        Handler->>Handler: process tool calls (iterationCount++)
        Handler->>Provider: request again with tool results
    end
    
    Note over Provider: After tool processing complete
    
    alt stream=true (NEW: no iterationCount check)
        Provider->>Handler: streaming response
        Handler->>Executor: {stream, execution}
        
        alt stream.tee available
            Executor->>Executor: stream.tee() → [clientStream, executorStream]
            
            par Process Client Stream
                Executor->>StreamProc: processStream(clientStream)
                StreamProc->>StreamProc: filter by selectedOutputs
                StreamProc-->>Executor: processedClientStream
                Executor->>Client: onStream(processedClientStream)
            and Consume Executor Stream
                Executor->>Executor: consumeExecutorStream(executorStream)
                Executor->>Executor: read all chunks, build fullContent
                Executor->>Executor: update execution.output.content
            end
        else stream.tee not available
            Executor->>StreamProc: processStream(stream)
            StreamProc-->>Executor: processedStream
            Executor->>Client: onStream(processedStream)
        end
    end
    
    Executor->>Executor: normalizeOutput(execution.output)
    Executor->>Executor: setBlockOutput()

Loading

[greptile-apps]

_{11 files reviewed, no comments}

_{Edit Code Review Agent Settings | Greptile}