opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

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

Browse source 
- 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 36c3345816
commit eac212cef9
8 changed files with 225 additions and 83 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

85
pkg/agent/loop.go
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.
// 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,
})
}

44
pkg/agent/loop_test.go
View file

@ -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()

33
pkg/providers/factory_provider.go
View file

@ -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" {

16
pkg/providers/http_provider.go
View file

@ -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...),
}
}

101
pkg/providers/openai_compat/provider.go
View file

@ -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 {

22
pkg/providers/openai_compat/provider_test.go
View file

@ -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")
}

1
pkg/providers/protocoltypes/types.go
View file

@ -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

6
pkg/tools/subagent.go
View file

@ -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,