feat(stream): show reasoning content in real-time during streaming

- Add ReasoningDelta to StreamEvent and parse reasoning_content from SSE
- Accumulate reasoning into LLMResponse.Reasoning during streaming
- Display reasoning in a fixed-height sliding window (buildStreamingDisplay)
  with mode indicator: "Thinking..." / "Thought, now responding..."
- Refactor openai_compat to upstream's functional options pattern
  (WithMaxTokensField, WithRequestTimeout, WithStream, WithEndpointPath)
- Fix subagent exec tool error handling (log warning instead of silent _)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
dj-oyu 2026-02-28 22:59:53 +09:00
parent 0ab75d92ec
commit 0f160f4330
8 changed files with 225 additions and 83 deletions

View file

@ -1692,6 +1692,61 @@ func (al *AgentLoop) handleReasoning(ctx context.Context, reasoningContent, chan
} }
} }
// streamingReasoningLines is the number of lines reserved for reasoning
// in the streaming display. The remaining lines go to content.
const streamingReasoningLines = 6
// buildStreamingDisplay builds a fixed-height status bubble for streaming.
//
// Layout when reasoning is active (reasoning only or both):
//
// 🧠 Thinking...
// ━━━━━━━━━━
// <reasoning tail — streamingReasoningLines lines>
// ━━━━━━━━━━
// <content tail — remaining lines> (or blank if content is empty)
// █
//
// Layout when no reasoning (content only):
//
// <content tail — streamingDisplayLines lines>
// █
func buildStreamingDisplay(content, reasoning string) string {
if reasoning == "" {
// No reasoning — full window for content.
return utils.TailPad(content, streamingDisplayLines, maxEntryLineWidth) + " \u2589"
}
var sb strings.Builder
// Header
if content == "" {
sb.WriteString("\U0001f9e0 Thinking...\n")
} else {
sb.WriteString("\U0001f9e0 Thought, now responding...\n")
}
sb.WriteString(statusSeparator)
// Reasoning window
headerLines := 2 // header + separator
footerLines := 1 // separator before content
contentLines := streamingDisplayLines - headerLines - footerLines - streamingReasoningLines
if contentLines < 3 {
contentLines = 3
}
rLines := streamingDisplayLines - headerLines - footerLines - contentLines
sb.WriteString(utils.TailPad(reasoning, rLines, maxEntryLineWidth))
sb.WriteByte('\n')
sb.WriteString(statusSeparator)
// Content window (may be blank padding if content hasn't started)
sb.WriteString(utils.TailPad(content, contentLines, maxEntryLineWidth))
sb.WriteString(" \u2589")
return sb.String()
}
// runLLMIteration executes the LLM call loop with tool handling. // runLLMIteration executes the LLM call loop with tool handling.
// consumeStreamWithRepetitionDetection reads StreamEvents from ch, accumulates // consumeStreamWithRepetitionDetection reads StreamEvents from ch, accumulates
// content and tool calls, and runs repetition detection every checkInterval runes. // content and tool calls, and runs repetition detection every checkInterval runes.
@ -1701,9 +1756,10 @@ func consumeStreamWithRepetitionDetection(
ch <-chan protocoltypes.StreamEvent, ch <-chan protocoltypes.StreamEvent,
cancelFn context.CancelFunc, cancelFn context.CancelFunc,
checkInterval int, checkInterval int,
onChunk func(accumulated string), onChunk func(content, reasoning string),
) (*providers.LLMResponse, bool, error) { ) (*providers.LLMResponse, bool, error) {
var content strings.Builder var content strings.Builder
var reasoning strings.Builder
var toolCalls []streamToolCallAcc var toolCalls []streamToolCallAcc
var finishReason string var finishReason string
var usage *providers.UsageInfo var usage *providers.UsageInfo
@ -1713,12 +1769,18 @@ func consumeStreamWithRepetitionDetection(
if ev.Err != nil { if ev.Err != nil {
return nil, false, ev.Err return nil, false, ev.Err
} }
updated := false
if ev.ContentDelta != "" { if ev.ContentDelta != "" {
content.WriteString(ev.ContentDelta) content.WriteString(ev.ContentDelta)
runesSinceLastCheck += utf8.RuneCountInString(ev.ContentDelta) runesSinceLastCheck += utf8.RuneCountInString(ev.ContentDelta)
if onChunk != nil { updated = true
onChunk(content.String()) }
} if ev.ReasoningDelta != "" {
reasoning.WriteString(ev.ReasoningDelta)
updated = true
}
if updated && onChunk != nil {
onChunk(content.String(), reasoning.String())
} }
if ev.FinishReason != "" { if ev.FinishReason != "" {
finishReason = ev.FinishReason finishReason = ev.FinishReason
@ -1747,13 +1809,13 @@ func consumeStreamWithRepetitionDetection(
// Drain remaining events so the producer goroutine can exit. // Drain remaining events so the producer goroutine can exit.
for range ch { for range ch {
} }
resp := buildAccumulatedResponse(content.String(), toolCalls, finishReason, usage) resp := buildAccumulatedResponse(content.String(), reasoning.String(), toolCalls, finishReason, usage)
return resp, true, nil return resp, true, nil
} }
} }
} }
resp := buildAccumulatedResponse(content.String(), toolCalls, finishReason, usage) resp := buildAccumulatedResponse(content.String(), reasoning.String(), toolCalls, finishReason, usage)
return resp, false, nil return resp, false, nil
} }
@ -1765,9 +1827,10 @@ type streamToolCallAcc struct {
} }
// buildAccumulatedResponse constructs an LLMResponse from accumulated stream data. // buildAccumulatedResponse constructs an LLMResponse from accumulated stream data.
func buildAccumulatedResponse(content string, toolCalls []streamToolCallAcc, finishReason string, usage *providers.UsageInfo) *providers.LLMResponse { func buildAccumulatedResponse(content, reasoning string, toolCalls []streamToolCallAcc, finishReason string, usage *providers.UsageInfo) *providers.LLMResponse {
resp := &providers.LLMResponse{ resp := &providers.LLMResponse{
Content: content, Content: content,
Reasoning: reasoning,
FinishReason: finishReason, FinishReason: finishReason,
Usage: usage, Usage: usage,
} }
@ -1880,19 +1943,19 @@ func (al *AgentLoop) runLLMIteration(
// Build onChunk callback for streaming preview. // Build onChunk callback for streaming preview.
// When sending responses to a real (non-internal) channel, publish // When sending responses to a real (non-internal) channel, publish
// throttled status updates so the user sees LLM output in real time. // throttled status updates so the user sees LLM output in real time.
var onChunk func(string) var onChunk func(string, string)
if !constants.IsInternalChannel(opts.Channel) { if !constants.IsInternalChannel(opts.Channel) {
lastPublish := time.Time{} lastPublish := time.Time{}
onChunk = func(accumulated string) { onChunk = func(accumulated, reasoning string) {
if time.Since(lastPublish) < 500*time.Millisecond { if time.Since(lastPublish) < 500*time.Millisecond {
return return
} }
lastPublish = time.Now() lastPublish = time.Now()
display := utils.TailPad(accumulated, streamingDisplayLines, maxEntryLineWidth) display := buildStreamingDisplay(accumulated, reasoning)
_ = al.bus.PublishOutbound(ctx, bus.OutboundMessage{ _ = al.bus.PublishOutbound(ctx, bus.OutboundMessage{
Channel: opts.Channel, Channel: opts.Channel,
ChatID: opts.ChatID, ChatID: opts.ChatID,
Content: display + " \u2589", Content: display,
IsStatus: true, IsStatus: true,
}) })
} }

View file

@ -2359,7 +2359,7 @@ func TestConsumeStream_OnChunkCallback(t *testing.T) {
defer cancel() defer cancel()
var chunks []string var chunks []string
onChunk := func(accumulated string) { onChunk := func(accumulated, _ string) {
chunks = append(chunks, accumulated) chunks = append(chunks, accumulated)
} }
@ -2410,7 +2410,7 @@ func TestConsumeStream_OnChunkWithRepetitionDetection(t *testing.T) {
}() }()
var chunkCount int var chunkCount int
onChunk := func(accumulated string) { onChunk := func(_, _ string) {
chunkCount++ chunkCount++
} }
@ -2791,6 +2791,46 @@ func TestFilterInterviewTools(t *testing.T) {
} }
} }
func TestBuildStreamingDisplay_ContentOnly(t *testing.T) {
display := buildStreamingDisplay("hello world", "")
if !strings.HasSuffix(display, " \u2589") {
t.Error("expected cursor suffix")
}
if strings.Contains(display, "\U0001f9e0") {
t.Error("should not contain brain emoji when no reasoning")
}
lines := strings.Count(display, "\n") + 1
if lines != streamingDisplayLines+1 { // TailPad lines + cursor on last line
t.Logf("display:\n%s", display)
}
}
func TestBuildStreamingDisplay_ReasoningOnly(t *testing.T) {
display := buildStreamingDisplay("", "let me think about this")
if !strings.Contains(display, "\U0001f9e0") {
t.Error("expected brain emoji for reasoning phase")
}
if !strings.Contains(display, "Thinking...") {
t.Error("expected Thinking... header")
}
if !strings.HasSuffix(display, " \u2589") {
t.Error("expected cursor suffix")
}
}
func TestBuildStreamingDisplay_Both(t *testing.T) {
display := buildStreamingDisplay("the answer is 42", "first I considered...")
if !strings.Contains(display, "\U0001f9e0") {
t.Error("expected brain emoji")
}
if !strings.Contains(display, "responding") {
t.Error("expected responding header when both present")
}
if !strings.Contains(display, "the answer is 42") {
t.Error("expected content in display")
}
}
func TestHandleReasoning(t *testing.T) { func TestHandleReasoning(t *testing.T) {
newLoop := func(t *testing.T) (*AgentLoop, *bus.MessageBus) { newLoop := func(t *testing.T) (*AgentLoop, *bus.MessageBus) {
t.Helper() t.Helper()

View file

@ -8,6 +8,7 @@ package providers
import ( import (
"fmt" "fmt"
"strings" "strings"
"time"
"github.com/sipeed/picoclaw/pkg/config" "github.com/sipeed/picoclaw/pkg/config"
"github.com/sipeed/picoclaw/pkg/providers/openai_compat" "github.com/sipeed/picoclaw/pkg/providers/openai_compat"
@ -85,11 +86,11 @@ func CreateProviderFromConfig(cfg *config.ModelConfig) (LLMProvider, string, err
if apiBase == "" { if apiBase == "" {
apiBase = getDefaultAPIBase(protocol) apiBase = getDefaultAPIBase(protocol)
} }
return NewHTTPProviderWithOptions(cfg.APIKey, apiBase, cfg.Proxy, openai_compat.Options{ return NewHTTPProviderWithOptions(cfg.APIKey, apiBase, cfg.Proxy,
MaxTokensField: cfg.MaxTokensField, openai_compat.WithMaxTokensField(cfg.MaxTokensField),
Stream: boolDefault(cfg.Stream, false), openai_compat.WithStream(boolDefault(cfg.Stream, false)),
RequestTimeout: cfg.RequestTimeout, openai_compat.WithRequestTimeout(time.Duration(cfg.RequestTimeout)*time.Second),
}), modelID, nil ), modelID, nil
case "minimax": case "minimax":
// MiniMax uses a non-standard endpoint path and defaults to SSE streaming. // MiniMax uses a non-standard endpoint path and defaults to SSE streaming.
@ -100,12 +101,12 @@ func CreateProviderFromConfig(cfg *config.ModelConfig) (LLMProvider, string, err
if apiBase == "" { if apiBase == "" {
apiBase = getDefaultAPIBase(protocol) apiBase = getDefaultAPIBase(protocol)
} }
return NewHTTPProviderWithOptions(cfg.APIKey, apiBase, cfg.Proxy, openai_compat.Options{ return NewHTTPProviderWithOptions(cfg.APIKey, apiBase, cfg.Proxy,
EndpointPath: "/text/chatcompletion_v2", openai_compat.WithEndpointPath("/text/chatcompletion_v2"),
MaxTokensField: cfg.MaxTokensField, openai_compat.WithMaxTokensField(cfg.MaxTokensField),
Stream: boolDefault(cfg.Stream, true), openai_compat.WithStream(boolDefault(cfg.Stream, true)),
RequestTimeout: cfg.RequestTimeout, openai_compat.WithRequestTimeout(time.Duration(cfg.RequestTimeout)*time.Second),
}), modelID, nil ), modelID, nil
case "openrouter", "groq", "zhipu", "gemini", "nvidia", case "openrouter", "groq", "zhipu", "gemini", "nvidia",
"ollama", "moonshot", "shengsuanyun", "deepseek", "cerebras", "ollama", "moonshot", "shengsuanyun", "deepseek", "cerebras",
@ -118,11 +119,11 @@ func CreateProviderFromConfig(cfg *config.ModelConfig) (LLMProvider, string, err
if apiBase == "" { if apiBase == "" {
apiBase = getDefaultAPIBase(protocol) apiBase = getDefaultAPIBase(protocol)
} }
return NewHTTPProviderWithOptions(cfg.APIKey, apiBase, cfg.Proxy, openai_compat.Options{ return NewHTTPProviderWithOptions(cfg.APIKey, apiBase, cfg.Proxy,
MaxTokensField: cfg.MaxTokensField, openai_compat.WithMaxTokensField(cfg.MaxTokensField),
Stream: boolDefault(cfg.Stream, false), openai_compat.WithStream(boolDefault(cfg.Stream, false)),
RequestTimeout: cfg.RequestTimeout, openai_compat.WithRequestTimeout(time.Duration(cfg.RequestTimeout)*time.Second),
}), modelID, nil ), modelID, nil
case "anthropic": case "anthropic":
if cfg.AuthMethod == "oauth" || cfg.AuthMethod == "token" { if cfg.AuthMethod == "oauth" || cfg.AuthMethod == "token" {

View file

@ -8,6 +8,7 @@ package providers
import ( import (
"context" "context"
"time"
"github.com/sipeed/picoclaw/pkg/providers/openai_compat" "github.com/sipeed/picoclaw/pkg/providers/openai_compat"
) )
@ -31,16 +32,19 @@ func NewHTTPProviderWithMaxTokensFieldAndRequestTimeout(
requestTimeoutSeconds int, requestTimeoutSeconds int,
) *HTTPProvider { ) *HTTPProvider {
return &HTTPProvider{ return &HTTPProvider{
delegate: openai_compat.NewProviderWithOptions(apiKey, apiBase, proxy, openai_compat.Options{ delegate: openai_compat.NewProvider(
MaxTokensField: maxTokensField, apiKey,
RequestTimeout: requestTimeoutSeconds, apiBase,
}), proxy,
openai_compat.WithMaxTokensField(maxTokensField),
openai_compat.WithRequestTimeout(time.Duration(requestTimeoutSeconds)*time.Second),
),
} }
} }
func NewHTTPProviderWithOptions(apiKey, apiBase, proxy string, opts openai_compat.Options) *HTTPProvider { func NewHTTPProviderWithOptions(apiKey, apiBase, proxy string, opts ...openai_compat.Option) *HTTPProvider {
return &HTTPProvider{ return &HTTPProvider{
delegate: openai_compat.NewProviderWithOptions(apiKey, apiBase, proxy, opts), delegate: openai_compat.NewProvider(apiKey, apiBase, proxy, opts...),
} }
} }

View file

@ -38,25 +38,49 @@ type Provider struct {
httpClient *http.Client httpClient *http.Client
} }
// Options configures optional behaviour for the provider. // Option is a functional option for configuring a Provider.
type Options struct { type Option func(*Provider)
EndpointPath string // API path appended to apiBase (default: "/chat/completions")
MaxTokensField string // Field name for max tokens parameter
Stream bool // Use SSE streaming internally
RequestTimeout int // Request timeout in seconds (0 = default 120s)
}
const defaultRequestTimeout = 120 * time.Second const defaultRequestTimeout = 120 * time.Second
func NewProviderWithOptions(apiKey, apiBase, proxy string, opts Options) *Provider { // WithMaxTokensField sets the field name for max tokens (e.g., "max_completion_tokens").
timeout := defaultRequestTimeout func WithMaxTokensField(maxTokensField string) Option {
if opts.RequestTimeout > 0 { return func(p *Provider) {
timeout = time.Duration(opts.RequestTimeout) * time.Second p.maxTokensField = maxTokensField
} else if opts.Stream {
timeout = 5 * time.Minute
} }
}
// WithRequestTimeout overrides the HTTP client timeout.
func WithRequestTimeout(timeout time.Duration) Option {
return func(p *Provider) {
if timeout > 0 {
p.httpClient.Timeout = timeout
}
}
}
// WithStream enables SSE streaming mode.
func WithStream(stream bool) Option {
return func(p *Provider) {
p.stream = stream
if stream && p.httpClient.Timeout == defaultRequestTimeout {
p.httpClient.Timeout = 5 * time.Minute
}
}
}
// WithEndpointPath sets the API path appended to apiBase (default: "/chat/completions").
func WithEndpointPath(path string) Option {
return func(p *Provider) {
if path != "" {
p.endpointPath = path
}
}
}
func NewProvider(apiKey, apiBase, proxy string, opts ...Option) *Provider {
client := &http.Client{ client := &http.Client{
Timeout: timeout, Timeout: defaultRequestTimeout,
} }
if proxy != "" { if proxy != "" {
@ -70,39 +94,37 @@ func NewProviderWithOptions(apiKey, apiBase, proxy string, opts Options) *Provid
} }
} }
endpointPath := opts.EndpointPath p := &Provider{
if endpointPath == "" { apiKey: apiKey,
endpointPath = "/chat/completions" apiBase: strings.TrimRight(apiBase, "/"),
endpointPath: "/chat/completions",
httpClient: client,
} }
return &Provider{ for _, opt := range opts {
apiKey: apiKey, if opt != nil {
apiBase: strings.TrimRight(apiBase, "/"), opt(p)
endpointPath: endpointPath, }
maxTokensField: opts.MaxTokensField,
stream: opts.Stream,
httpClient: client,
} }
}
func NewProvider(apiKey, apiBase, proxy string) *Provider { return p
return NewProviderWithOptions(apiKey, apiBase, proxy, Options{})
} }
func NewProviderWithMaxTokensField(apiKey, apiBase, proxy, maxTokensField string) *Provider { func NewProviderWithMaxTokensField(apiKey, apiBase, proxy, maxTokensField string) *Provider {
return NewProviderWithOptions(apiKey, apiBase, proxy, Options{ return NewProvider(apiKey, apiBase, proxy, WithMaxTokensField(maxTokensField))
MaxTokensField: maxTokensField,
})
} }
func NewProviderWithMaxTokensFieldAndTimeout( func NewProviderWithMaxTokensFieldAndTimeout(
apiKey, apiBase, proxy, maxTokensField string, apiKey, apiBase, proxy, maxTokensField string,
requestTimeoutSeconds int, requestTimeoutSeconds int,
) *Provider { ) *Provider {
return NewProviderWithOptions(apiKey, apiBase, proxy, Options{ return NewProvider(
MaxTokensField: maxTokensField, apiKey,
RequestTimeout: requestTimeoutSeconds, apiBase,
}) proxy,
WithMaxTokensField(maxTokensField),
WithRequestTimeout(time.Duration(requestTimeoutSeconds)*time.Second),
)
} }
// streamBufferSize is the channel buffer size for ChatStream events. // streamBufferSize is the channel buffer size for ChatStream events.
@ -308,6 +330,7 @@ func readSSEIntoChannel(ctx context.Context, r io.Reader, ch chan<- protocoltype
if len(chunk.Choices) > 0 { if len(chunk.Choices) > 0 {
choice := chunk.Choices[0] choice := chunk.Choices[0]
ev.ContentDelta = choice.Delta.Content ev.ContentDelta = choice.Delta.Content
ev.ReasoningDelta = choice.Delta.ReasoningContent
if choice.FinishReason != "" { if choice.FinishReason != "" {
ev.FinishReason = choice.FinishReason ev.FinishReason = choice.FinishReason
} }
@ -342,6 +365,7 @@ func readSSEIntoChannel(ctx context.Context, r io.Reader, ch chan<- protocoltype
// AccumulateStream drains a StreamEvent channel and returns a complete LLMResponse. // AccumulateStream drains a StreamEvent channel and returns a complete LLMResponse.
func AccumulateStream(ch <-chan protocoltypes.StreamEvent) (*LLMResponse, error) { func AccumulateStream(ch <-chan protocoltypes.StreamEvent) (*LLMResponse, error) {
var content strings.Builder var content strings.Builder
var reasoning strings.Builder
var toolCalls []streamToolCallAcc var toolCalls []streamToolCallAcc
var finishReason string var finishReason string
var usage *UsageInfo var usage *UsageInfo
@ -353,6 +377,9 @@ func AccumulateStream(ch <-chan protocoltypes.StreamEvent) (*LLMResponse, error)
if ev.ContentDelta != "" { if ev.ContentDelta != "" {
content.WriteString(ev.ContentDelta) content.WriteString(ev.ContentDelta)
} }
if ev.ReasoningDelta != "" {
reasoning.WriteString(ev.ReasoningDelta)
}
if ev.FinishReason != "" { if ev.FinishReason != "" {
finishReason = ev.FinishReason finishReason = ev.FinishReason
} }
@ -375,6 +402,7 @@ func AccumulateStream(ch <-chan protocoltypes.StreamEvent) (*LLMResponse, error)
result := &LLMResponse{ result := &LLMResponse{
Content: content.String(), Content: content.String(),
Reasoning: reasoning.String(),
FinishReason: finishReason, FinishReason: finishReason,
Usage: usage, Usage: usage,
} }
@ -576,8 +604,9 @@ type streamChoice struct {
} }
type streamDelta struct { type streamDelta struct {
Content string `json:"content"` Content string `json:"content"`
ToolCalls []streamDeltaTC `json:"tool_calls"` ReasoningContent string `json:"reasoning_content"`
ToolCalls []streamDeltaTC `json:"tool_calls"`
} }
type streamDeltaTC struct { type streamDeltaTC struct {

View file

@ -373,10 +373,10 @@ func TestProviderChat_StreamingTextResponse(t *testing.T) {
})) }))
defer server.Close() defer server.Close()
p := NewProviderWithOptions("key", server.URL, "", Options{ p := NewProvider("key", server.URL, "",
EndpointPath: "/text/chatcompletion_v2", WithEndpointPath("/text/chatcompletion_v2"),
Stream: true, WithStream(true),
}) )
out, err := p.Chat(t.Context(), []Message{{Role: "user", Content: "hi"}}, nil, "MiniMax-M1", nil) out, err := p.Chat(t.Context(), []Message{{Role: "user", Content: "hi"}}, nil, "MiniMax-M1", nil)
if err != nil { if err != nil {
t.Fatalf("Chat() error = %v", err) t.Fatalf("Chat() error = %v", err)
@ -415,7 +415,7 @@ func TestProviderChat_StreamingToolCalls(t *testing.T) {
})) }))
defer server.Close() defer server.Close()
p := NewProviderWithOptions("key", server.URL, "", Options{Stream: true}) p := NewProvider("key", server.URL, "", WithStream(true))
out, err := p.Chat(t.Context(), []Message{{Role: "user", Content: "weather?"}}, nil, "test", nil) out, err := p.Chat(t.Context(), []Message{{Role: "user", Content: "weather?"}}, nil, "test", nil)
if err != nil { if err != nil {
t.Fatalf("Chat() error = %v", err) t.Fatalf("Chat() error = %v", err)
@ -449,9 +449,9 @@ func TestProviderChat_CustomEndpointPath(t *testing.T) {
})) }))
defer server.Close() defer server.Close()
p := NewProviderWithOptions("key", server.URL, "", Options{ p := NewProvider("key", server.URL, "",
EndpointPath: "/text/chatcompletion_v2", WithEndpointPath("/text/chatcompletion_v2"),
}) )
_, err := p.Chat(t.Context(), []Message{{Role: "user", Content: "hi"}}, nil, "test", nil) _, err := p.Chat(t.Context(), []Message{{Role: "user", Content: "hi"}}, nil, "test", nil)
if err != nil { if err != nil {
t.Fatalf("Chat() error = %v", err) t.Fatalf("Chat() error = %v", err)
@ -633,7 +633,7 @@ func TestChatStream_EndToEnd(t *testing.T) {
})) }))
defer server.Close() defer server.Close()
p := NewProviderWithOptions("key", server.URL, "", Options{Stream: true}) p := NewProvider("key", server.URL, "", WithStream(true))
ch, err := p.ChatStream(t.Context(), []Message{{Role: "user", Content: "hi"}}, nil, "test", nil) ch, err := p.ChatStream(t.Context(), []Message{{Role: "user", Content: "hi"}}, nil, "test", nil)
if err != nil { if err != nil {
@ -679,7 +679,7 @@ func TestChatStream_EarlyCancel(t *testing.T) {
})) }))
defer server.Close() defer server.Close()
p := NewProviderWithOptions("key", server.URL, "", Options{Stream: true}) p := NewProvider("key", server.URL, "", WithStream(true))
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
defer cancel() defer cancel()
@ -714,7 +714,7 @@ func TestCanStream(t *testing.T) {
t.Error("CanStream() = true for non-stream provider") t.Error("CanStream() = true for non-stream provider")
} }
p2 := NewProviderWithOptions("key", "https://example.com", "", Options{Stream: true}) p2 := NewProvider("key", "https://example.com", "", WithStream(true))
if !p2.CanStream() { if !p2.CanStream() {
t.Error("CanStream() = false for stream provider") t.Error("CanStream() = false for stream provider")
} }

View file

@ -85,6 +85,7 @@ type ToolFunctionDefinition struct {
// StreamEvent represents a single chunk from an SSE streaming response. // StreamEvent represents a single chunk from an SSE streaming response.
type StreamEvent struct { type StreamEvent struct {
ContentDelta string ContentDelta string
ReasoningDelta string // incremental reasoning/thinking content
ToolCallDeltas []StreamToolCallDelta ToolCallDeltas []StreamToolCallDelta
FinishReason string // set only on the final event FinishReason string // set only on the final event
Usage *UsageInfo // set only on the final event Usage *UsageInfo // set only on the final event

View file

@ -3,6 +3,7 @@ package tools
import ( import (
"context" "context"
"fmt" "fmt"
"log"
"sync" "sync"
"time" "time"
@ -53,7 +54,10 @@ func NewSubagentManager(
reporter = orch.Noop reporter = orch.Noop
} }
// Create a shared exec tool for all presets // Create a shared exec tool for all presets
execTool, _ := NewExecTool(workspace, true) execTool, err := NewExecTool(workspace, true)
if err != nil {
log.Printf("subagent: failed to create exec tool: %v (exec disabled for subagents)", err)
}
return &SubagentManager{ return &SubagentManager{
tasks: make(map[string]*SubagentTask), tasks: make(map[string]*SubagentTask),
provider: provider, provider: provider,