From 0f160f4330825d0b143a060be0d04b4116650155 Mon Sep 17 00:00:00 2001 From: dj-oyu <68707227+dj-oyu@users.noreply.github.com> Date: Sat, 28 Feb 2026 22:59:53 +0900 Subject: [PATCH] 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 --- pkg/agent/loop.go | 85 ++++++++++++++-- pkg/agent/loop_test.go | 44 +++++++- pkg/providers/factory_provider.go | 33 +++--- pkg/providers/http_provider.go | 16 +-- pkg/providers/openai_compat/provider.go | 101 ++++++++++++------- pkg/providers/openai_compat/provider_test.go | 22 ++-- pkg/providers/protocoltypes/types.go | 1 + pkg/tools/subagent.go | 6 +- 8 files changed, 225 insertions(+), 83 deletions(-) diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index b166f811a..8b6c10471 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -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... +// ━━━━━━━━━━ +// +// ━━━━━━━━━━ +// (or blank if content is empty) +// █ +// +// Layout when no reasoning (content only): +// +// +// █ +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. // consumeStreamWithRepetitionDetection reads StreamEvents from ch, accumulates // content and tool calls, and runs repetition detection every checkInterval runes. @@ -1701,9 +1756,10 @@ func consumeStreamWithRepetitionDetection( ch <-chan protocoltypes.StreamEvent, cancelFn context.CancelFunc, checkInterval int, - onChunk func(accumulated string), + onChunk func(content, reasoning string), ) (*providers.LLMResponse, bool, error) { var content strings.Builder + var reasoning strings.Builder var toolCalls []streamToolCallAcc var finishReason string var usage *providers.UsageInfo @@ -1713,12 +1769,18 @@ func consumeStreamWithRepetitionDetection( if ev.Err != nil { return nil, false, ev.Err } + updated := false if ev.ContentDelta != "" { content.WriteString(ev.ContentDelta) runesSinceLastCheck += utf8.RuneCountInString(ev.ContentDelta) - if onChunk != nil { - onChunk(content.String()) - } + updated = true + } + if ev.ReasoningDelta != "" { + reasoning.WriteString(ev.ReasoningDelta) + updated = true + } + if updated && onChunk != nil { + onChunk(content.String(), reasoning.String()) } if ev.FinishReason != "" { finishReason = ev.FinishReason @@ -1747,13 +1809,13 @@ func consumeStreamWithRepetitionDetection( // Drain remaining events so the producer goroutine can exit. for range ch { } - resp := buildAccumulatedResponse(content.String(), toolCalls, finishReason, usage) + resp := buildAccumulatedResponse(content.String(), reasoning.String(), toolCalls, finishReason, usage) return resp, true, nil } } } - resp := buildAccumulatedResponse(content.String(), toolCalls, finishReason, usage) + resp := buildAccumulatedResponse(content.String(), reasoning.String(), toolCalls, finishReason, usage) return resp, false, nil } @@ -1765,9 +1827,10 @@ type streamToolCallAcc struct { } // 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{ Content: content, + Reasoning: reasoning, FinishReason: finishReason, Usage: usage, } @@ -1880,19 +1943,19 @@ func (al *AgentLoop) runLLMIteration( // Build onChunk callback for streaming preview. // When sending responses to a real (non-internal) channel, publish // 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) { lastPublish := time.Time{} - onChunk = func(accumulated string) { + onChunk = func(accumulated, reasoning string) { if time.Since(lastPublish) < 500*time.Millisecond { return } lastPublish = time.Now() - display := utils.TailPad(accumulated, streamingDisplayLines, maxEntryLineWidth) + display := buildStreamingDisplay(accumulated, reasoning) _ = al.bus.PublishOutbound(ctx, bus.OutboundMessage{ Channel: opts.Channel, ChatID: opts.ChatID, - Content: display + " \u2589", + Content: display, IsStatus: true, }) } diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index 20bae5a4b..b75d32a45 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -2359,7 +2359,7 @@ func TestConsumeStream_OnChunkCallback(t *testing.T) { defer cancel() var chunks []string - onChunk := func(accumulated string) { + onChunk := func(accumulated, _ string) { chunks = append(chunks, accumulated) } @@ -2410,7 +2410,7 @@ func TestConsumeStream_OnChunkWithRepetitionDetection(t *testing.T) { }() var chunkCount int - onChunk := func(accumulated string) { + onChunk := func(_, _ string) { 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) { newLoop := func(t *testing.T) (*AgentLoop, *bus.MessageBus) { t.Helper() diff --git a/pkg/providers/factory_provider.go b/pkg/providers/factory_provider.go index 65a321eba..cc5d905ad 100644 --- a/pkg/providers/factory_provider.go +++ b/pkg/providers/factory_provider.go @@ -8,6 +8,7 @@ package providers import ( "fmt" "strings" + "time" "github.com/sipeed/picoclaw/pkg/config" "github.com/sipeed/picoclaw/pkg/providers/openai_compat" @@ -85,11 +86,11 @@ func CreateProviderFromConfig(cfg *config.ModelConfig) (LLMProvider, string, err if apiBase == "" { apiBase = getDefaultAPIBase(protocol) } - return NewHTTPProviderWithOptions(cfg.APIKey, apiBase, cfg.Proxy, openai_compat.Options{ - MaxTokensField: cfg.MaxTokensField, - Stream: boolDefault(cfg.Stream, false), - RequestTimeout: cfg.RequestTimeout, - }), modelID, nil + return NewHTTPProviderWithOptions(cfg.APIKey, apiBase, cfg.Proxy, + openai_compat.WithMaxTokensField(cfg.MaxTokensField), + openai_compat.WithStream(boolDefault(cfg.Stream, false)), + openai_compat.WithRequestTimeout(time.Duration(cfg.RequestTimeout)*time.Second), + ), modelID, nil case "minimax": // 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 == "" { apiBase = getDefaultAPIBase(protocol) } - return NewHTTPProviderWithOptions(cfg.APIKey, apiBase, cfg.Proxy, openai_compat.Options{ - EndpointPath: "/text/chatcompletion_v2", - MaxTokensField: cfg.MaxTokensField, - Stream: boolDefault(cfg.Stream, true), - RequestTimeout: cfg.RequestTimeout, - }), modelID, nil + return NewHTTPProviderWithOptions(cfg.APIKey, apiBase, cfg.Proxy, + openai_compat.WithEndpointPath("/text/chatcompletion_v2"), + openai_compat.WithMaxTokensField(cfg.MaxTokensField), + openai_compat.WithStream(boolDefault(cfg.Stream, true)), + openai_compat.WithRequestTimeout(time.Duration(cfg.RequestTimeout)*time.Second), + ), modelID, nil case "openrouter", "groq", "zhipu", "gemini", "nvidia", "ollama", "moonshot", "shengsuanyun", "deepseek", "cerebras", @@ -118,11 +119,11 @@ func CreateProviderFromConfig(cfg *config.ModelConfig) (LLMProvider, string, err if apiBase == "" { apiBase = getDefaultAPIBase(protocol) } - return NewHTTPProviderWithOptions(cfg.APIKey, apiBase, cfg.Proxy, openai_compat.Options{ - MaxTokensField: cfg.MaxTokensField, - Stream: boolDefault(cfg.Stream, false), - RequestTimeout: cfg.RequestTimeout, - }), modelID, nil + return NewHTTPProviderWithOptions(cfg.APIKey, apiBase, cfg.Proxy, + openai_compat.WithMaxTokensField(cfg.MaxTokensField), + openai_compat.WithStream(boolDefault(cfg.Stream, false)), + openai_compat.WithRequestTimeout(time.Duration(cfg.RequestTimeout)*time.Second), + ), modelID, nil case "anthropic": if cfg.AuthMethod == "oauth" || cfg.AuthMethod == "token" { diff --git a/pkg/providers/http_provider.go b/pkg/providers/http_provider.go index f3c6d6a2e..1f02e9e1b 100644 --- a/pkg/providers/http_provider.go +++ b/pkg/providers/http_provider.go @@ -8,6 +8,7 @@ package providers import ( "context" + "time" "github.com/sipeed/picoclaw/pkg/providers/openai_compat" ) @@ -31,16 +32,19 @@ func NewHTTPProviderWithMaxTokensFieldAndRequestTimeout( requestTimeoutSeconds int, ) *HTTPProvider { return &HTTPProvider{ - delegate: openai_compat.NewProviderWithOptions(apiKey, apiBase, proxy, openai_compat.Options{ - MaxTokensField: maxTokensField, - RequestTimeout: requestTimeoutSeconds, - }), + delegate: openai_compat.NewProvider( + apiKey, + 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{ - delegate: openai_compat.NewProviderWithOptions(apiKey, apiBase, proxy, opts), + delegate: openai_compat.NewProvider(apiKey, apiBase, proxy, opts...), } } diff --git a/pkg/providers/openai_compat/provider.go b/pkg/providers/openai_compat/provider.go index 3e9b889b6..b2200a787 100644 --- a/pkg/providers/openai_compat/provider.go +++ b/pkg/providers/openai_compat/provider.go @@ -38,25 +38,49 @@ type Provider struct { httpClient *http.Client } -// Options configures optional behaviour for the provider. -type Options struct { - 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) -} +// Option is a functional option for configuring a Provider. +type Option func(*Provider) const defaultRequestTimeout = 120 * time.Second -func NewProviderWithOptions(apiKey, apiBase, proxy string, opts Options) *Provider { - timeout := defaultRequestTimeout - if opts.RequestTimeout > 0 { - timeout = time.Duration(opts.RequestTimeout) * time.Second - } else if opts.Stream { - timeout = 5 * time.Minute +// WithMaxTokensField sets the field name for max tokens (e.g., "max_completion_tokens"). +func WithMaxTokensField(maxTokensField string) Option { + return func(p *Provider) { + p.maxTokensField = maxTokensField } +} + +// 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{ - Timeout: timeout, + Timeout: defaultRequestTimeout, } if proxy != "" { @@ -70,39 +94,37 @@ func NewProviderWithOptions(apiKey, apiBase, proxy string, opts Options) *Provid } } - endpointPath := opts.EndpointPath - if endpointPath == "" { - endpointPath = "/chat/completions" + p := &Provider{ + apiKey: apiKey, + apiBase: strings.TrimRight(apiBase, "/"), + endpointPath: "/chat/completions", + httpClient: client, } - return &Provider{ - apiKey: apiKey, - apiBase: strings.TrimRight(apiBase, "/"), - endpointPath: endpointPath, - maxTokensField: opts.MaxTokensField, - stream: opts.Stream, - httpClient: client, + for _, opt := range opts { + if opt != nil { + opt(p) + } } -} -func NewProvider(apiKey, apiBase, proxy string) *Provider { - return NewProviderWithOptions(apiKey, apiBase, proxy, Options{}) + return p } func NewProviderWithMaxTokensField(apiKey, apiBase, proxy, maxTokensField string) *Provider { - return NewProviderWithOptions(apiKey, apiBase, proxy, Options{ - MaxTokensField: maxTokensField, - }) + return NewProvider(apiKey, apiBase, proxy, WithMaxTokensField(maxTokensField)) } func NewProviderWithMaxTokensFieldAndTimeout( apiKey, apiBase, proxy, maxTokensField string, requestTimeoutSeconds int, ) *Provider { - return NewProviderWithOptions(apiKey, apiBase, proxy, Options{ - MaxTokensField: maxTokensField, - RequestTimeout: requestTimeoutSeconds, - }) + return NewProvider( + apiKey, + apiBase, + proxy, + WithMaxTokensField(maxTokensField), + WithRequestTimeout(time.Duration(requestTimeoutSeconds)*time.Second), + ) } // 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 { choice := chunk.Choices[0] ev.ContentDelta = choice.Delta.Content + ev.ReasoningDelta = choice.Delta.ReasoningContent if 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. func AccumulateStream(ch <-chan protocoltypes.StreamEvent) (*LLMResponse, error) { var content strings.Builder + var reasoning strings.Builder var toolCalls []streamToolCallAcc var finishReason string var usage *UsageInfo @@ -353,6 +377,9 @@ func AccumulateStream(ch <-chan protocoltypes.StreamEvent) (*LLMResponse, error) if ev.ContentDelta != "" { content.WriteString(ev.ContentDelta) } + if ev.ReasoningDelta != "" { + reasoning.WriteString(ev.ReasoningDelta) + } if ev.FinishReason != "" { finishReason = ev.FinishReason } @@ -375,6 +402,7 @@ func AccumulateStream(ch <-chan protocoltypes.StreamEvent) (*LLMResponse, error) result := &LLMResponse{ Content: content.String(), + Reasoning: reasoning.String(), FinishReason: finishReason, Usage: usage, } @@ -576,8 +604,9 @@ type streamChoice struct { } type streamDelta struct { - Content string `json:"content"` - ToolCalls []streamDeltaTC `json:"tool_calls"` + Content string `json:"content"` + ReasoningContent string `json:"reasoning_content"` + ToolCalls []streamDeltaTC `json:"tool_calls"` } type streamDeltaTC struct { diff --git a/pkg/providers/openai_compat/provider_test.go b/pkg/providers/openai_compat/provider_test.go index e3c5df086..93f71458b 100644 --- a/pkg/providers/openai_compat/provider_test.go +++ b/pkg/providers/openai_compat/provider_test.go @@ -373,10 +373,10 @@ func TestProviderChat_StreamingTextResponse(t *testing.T) { })) defer server.Close() - p := NewProviderWithOptions("key", server.URL, "", Options{ - EndpointPath: "/text/chatcompletion_v2", - Stream: true, - }) + p := NewProvider("key", server.URL, "", + WithEndpointPath("/text/chatcompletion_v2"), + WithStream(true), + ) out, err := p.Chat(t.Context(), []Message{{Role: "user", Content: "hi"}}, nil, "MiniMax-M1", nil) if err != nil { t.Fatalf("Chat() error = %v", err) @@ -415,7 +415,7 @@ func TestProviderChat_StreamingToolCalls(t *testing.T) { })) 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) if err != nil { t.Fatalf("Chat() error = %v", err) @@ -449,9 +449,9 @@ func TestProviderChat_CustomEndpointPath(t *testing.T) { })) defer server.Close() - p := NewProviderWithOptions("key", server.URL, "", Options{ - EndpointPath: "/text/chatcompletion_v2", - }) + p := NewProvider("key", server.URL, "", + WithEndpointPath("/text/chatcompletion_v2"), + ) _, err := p.Chat(t.Context(), []Message{{Role: "user", Content: "hi"}}, nil, "test", nil) if err != nil { t.Fatalf("Chat() error = %v", err) @@ -633,7 +633,7 @@ func TestChatStream_EndToEnd(t *testing.T) { })) 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) if err != nil { @@ -679,7 +679,7 @@ func TestChatStream_EarlyCancel(t *testing.T) { })) defer server.Close() - p := NewProviderWithOptions("key", server.URL, "", Options{Stream: true}) + p := NewProvider("key", server.URL, "", WithStream(true)) ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -714,7 +714,7 @@ func TestCanStream(t *testing.T) { 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() { t.Error("CanStream() = false for stream provider") } diff --git a/pkg/providers/protocoltypes/types.go b/pkg/providers/protocoltypes/types.go index 1c11359f9..867078185 100644 --- a/pkg/providers/protocoltypes/types.go +++ b/pkg/providers/protocoltypes/types.go @@ -85,6 +85,7 @@ type ToolFunctionDefinition struct { // StreamEvent represents a single chunk from an SSE streaming response. type StreamEvent struct { ContentDelta string + ReasoningDelta string // incremental reasoning/thinking content ToolCallDeltas []StreamToolCallDelta FinishReason string // set only on the final event Usage *UsageInfo // set only on the final event diff --git a/pkg/tools/subagent.go b/pkg/tools/subagent.go index 4b951013d..08cbb0622 100644 --- a/pkg/tools/subagent.go +++ b/pkg/tools/subagent.go @@ -3,6 +3,7 @@ package tools import ( "context" "fmt" + "log" "sync" "time" @@ -53,7 +54,10 @@ func NewSubagentManager( reporter = orch.Noop } // 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{ tasks: make(map[string]*SubagentTask), provider: provider,