diff --git a/docs/STREAM_JSON_PROTOCOL.md b/docs/STREAM_JSON_PROTOCOL.md index 525eecf00..bdf2ddb39 100644 --- a/docs/STREAM_JSON_PROTOCOL.md +++ b/docs/STREAM_JSON_PROTOCOL.md @@ -41,6 +41,7 @@ events when present. ```json { "schemaVersion": 2, "type": "run_start", "runId": "run_20260603_abc123", "sessionId": "zero_20260603100000_abc123", "cwd": "/repo", "provider": "openai", "model": "gpt-4.1", "apiModel": "gpt-4.1" } +{ "schemaVersion": 2, "type": "reasoning", "runId": "run_20260603_abc123", "delta": "Thinking..." } { "schemaVersion": 2, "type": "text", "runId": "run_20260603_abc123", "delta": "..." } { "schemaVersion": 2, "type": "tool_call", "runId": "run_20260603_abc123", "id": "call_1", "name": "read_file", "args": { "path": "README.md" }, "sideEffect": "read" } { "schemaVersion": 2, "type": "permission_request", "runId": "run_20260603_abc123", "id": "call_2", "name": "write_file", "action": "prompt", "permission": "prompt", "permissionMode": "ask", "sideEffect": "write", "reason": "Creates or overwrites files." } @@ -51,6 +52,10 @@ events when present. { "schemaVersion": 2, "type": "run_end", "runId": "run_20260603_abc123", "status": "success", "exitCode": 0 } ``` +`reasoning` events carry live model reasoning/status deltas for providers that +stream them separately from answer text. They are liveness/progress events only: +they are not folded into `text` or the final answer. + Permission events may include structured sandbox metadata: ```json diff --git a/internal/agent/guardrails.go b/internal/agent/guardrails.go index 089dc818e..1db614c72 100644 --- a/internal/agent/guardrails.go +++ b/internal/agent/guardrails.go @@ -414,8 +414,9 @@ func (state *guardState) observeToolResult(name string, failed bool, output stri func (state *guardState) observeTurn(collected zeroruntime.CollectedStream) (stop bool) { hasToolCalls := len(collected.ToolCalls) > 0 hasVisibleText := strings.TrimSpace(collected.Text) != "" + hasReasoning := collected.HasReasoning || len(collected.ReasoningBlocks) > 0 - if hasToolCalls || hasVisibleText { + if hasToolCalls || hasVisibleText || hasReasoning { state.emptyTurns = 0 } else { state.emptyTurns++ diff --git a/internal/agent/guardrails_test.go b/internal/agent/guardrails_test.go index fd7758df0..bc48e43be 100644 --- a/internal/agent/guardrails_test.go +++ b/internal/agent/guardrails_test.go @@ -22,6 +22,14 @@ func textTurn(content string) []zeroruntime.StreamEvent { } } +// reasoningTurn produces live reasoning without visible assistant text. +func reasoningTurn(content string) []zeroruntime.StreamEvent { + return []zeroruntime.StreamEvent{ + {Type: zeroruntime.StreamEventReasoning, Content: content}, + {Type: zeroruntime.StreamEventDone}, + } +} + // toolTurn produces a turn that calls a named tool with the given args JSON. func toolTurn(callID string, toolName string, args string) []zeroruntime.StreamEvent { return []zeroruntime.StreamEvent{ @@ -101,6 +109,31 @@ func TestRunResetsEmptyTurnCounterOnVisibleOutput(t *testing.T) { } } +func TestRunResetsEmptyTurnCounterOnReasoning(t *testing.T) { + provider := &mockProvider{ + turns: [][]zeroruntime.StreamEvent{ + reasoningTurn("thinking 1"), + reasoningTurn("thinking 2"), + reasoningTurn("thinking 3"), + textTurn("done"), + }, + } + + result, err := Run(context.Background(), "go", provider, Options{ + Registry: tools.NewRegistry(), + MaxTurns: 12, + }) + if err != nil { + t.Fatal(err) + } + if result.FinalAnswer != "done" { + t.Fatalf("expected reasoning-only turns to keep the run live until final answer, got %q", result.FinalAnswer) + } + if len(provider.requests) != 4 { + t.Fatalf("expected 4 turns, got %d", len(provider.requests)) + } +} + func TestRunResetsEmptyTurnCounterOnToolCall(t *testing.T) { root := t.TempDir() writeAgentTestFile(t, root+"/notes.txt", "alpha") diff --git a/internal/cli/exec.go b/internal/cli/exec.go index 069d9e795..b430a4e2b 100644 --- a/internal/cli/exec.go +++ b/internal/cli/exec.go @@ -535,6 +535,7 @@ func runExec(args []string, stdout io.Writer, stderr io.Writer, deps appDeps) in EnabledTools: options.enabledTools, DisabledTools: options.disabledTools, OnText: writer.text, + OnReasoning: writer.reasoning, OnToolCall: func(call agent.ToolCall) { writer.toolCall(call, registry) sessionRecorder.append(sessions.EventToolCall, map[string]any{ diff --git a/internal/cli/exec_protocol_test.go b/internal/cli/exec_protocol_test.go index d804e41e9..6ca670a0b 100644 --- a/internal/cli/exec_protocol_test.go +++ b/internal/cli/exec_protocol_test.go @@ -529,6 +529,45 @@ func TestRunExecStreamJSONRunStartUsesResolvedAPIModel(t *testing.T) { } } +func TestRunExecStreamJSONEmitsReasoningEvents(t *testing.T) { + t.Setenv("XDG_DATA_HOME", t.TempDir()) + cwd := t.TempDir() + var stdout bytes.Buffer + var stderr bytes.Buffer + + exitCode := runWithDeps([]string{"exec", "--output-format", "stream-json", "think then answer"}, &stdout, &stderr, appDeps{ + getwd: func() (string, error) { + return cwd, nil + }, + resolveConfig: func(_ string, _ config.Overrides) (config.ResolvedConfig, error) { + return execResolvedConfig(), nil + }, + newProvider: func(config.ProviderProfile) (zeroruntime.Provider, error) { + return reasoningExecProvider{}, nil + }, + }) + + if exitCode != exitSuccess { + t.Fatalf("expected exit code %d, got %d: %s", exitSuccess, exitCode, stderr.String()) + } + events := decodeJSONLines(t, stdout.String()) + reasoningIdx, reasoning := findJSONEventIndex(t, events, "reasoning") + if reasoning["delta"] != "Thinking. " { + t.Fatalf("unexpected reasoning event: %#v", reasoning) + } + textIdx, text := findJSONEventIndex(t, events, "text") + if text["delta"] != "done" { + t.Fatalf("unexpected text event: %#v", text) + } + if reasoningIdx >= textIdx { + t.Fatalf("expected reasoning event before text event, got indices %d and %d", reasoningIdx, textIdx) + } + final := findJSONEvent(t, events, "final") + if final["text"] != "done" { + t.Fatalf("reasoning must not be folded into final answer: %#v", final) + } +} + func TestExecEventWriterTruncatesStreamJSONToolResults(t *testing.T) { var stdout bytes.Buffer writer := execEventWriter{ @@ -751,6 +790,17 @@ type toolCallingExecProvider struct { answer string } +type reasoningExecProvider struct{} + +func (reasoningExecProvider) StreamCompletion(context.Context, zeroruntime.CompletionRequest) (<-chan zeroruntime.StreamEvent, error) { + ch := make(chan zeroruntime.StreamEvent, 3) + ch <- zeroruntime.StreamEvent{Type: zeroruntime.StreamEventReasoning, Content: "Thinking. "} + ch <- zeroruntime.StreamEvent{Type: zeroruntime.StreamEventText, Content: "done"} + ch <- zeroruntime.StreamEvent{Type: zeroruntime.StreamEventDone} + close(ch) + return ch, nil +} + func (provider toolCallingExecProvider) StreamCompletion(ctx context.Context, request zeroruntime.CompletionRequest) (<-chan zeroruntime.StreamEvent, error) { for _, message := range request.Messages { if message.Role == zeroruntime.MessageRoleTool { @@ -777,13 +827,19 @@ func (provider toolCallingExecProvider) StreamCompletion(ctx context.Context, re func findJSONEvent(t *testing.T, events []map[string]any, eventType string) map[string]any { t.Helper() - for _, event := range events { + _, event := findJSONEventIndex(t, events, eventType) + return event +} + +func findJSONEventIndex(t *testing.T, events []map[string]any, eventType string) (int, map[string]any) { + t.Helper() + for idx, event := range events { if event["type"] == eventType { - return event + return idx, event } } t.Fatalf("event %q not found in %#v", eventType, events) - return nil + return -1, nil } func findSessionEvent(t *testing.T, events []sessions.Event, eventType sessions.EventType) sessions.Event { diff --git a/internal/cli/exec_writer.go b/internal/cli/exec_writer.go index a85393eef..e98d526a6 100644 --- a/internal/cli/exec_writer.go +++ b/internal/cli/exec_writer.go @@ -80,6 +80,16 @@ func (writer *execEventWriter) text(delta string) { writer.writeStdout(delta) } +func (writer *execEventWriter) reasoning(delta string) { + if writer.format == execOutputJSON { + writer.writeJSON(map[string]any{"type": "reasoning", "delta": delta}) + return + } + if writer.format == execOutputStreamJSON { + writer.writeStreamJSON(streamjson.Event{Type: streamjson.EventReasoning, RunID: writer.runID, Delta: delta}) + } +} + func (writer *execEventWriter) toolCall(call agent.ToolCall, registry *tools.Registry) { if writer.format == execOutputJSON { writer.writeJSON(map[string]any{ diff --git a/internal/providers/openai/provider.go b/internal/providers/openai/provider.go index 49de6a0a4..bc738bb23 100644 --- a/internal/providers/openai/provider.go +++ b/internal/providers/openai/provider.go @@ -283,10 +283,10 @@ func (provider *Provider) emitChunk( events chan<- zeroruntime.StreamEvent, ) { for _, choice := range chunk.Choices { - if choice.Delta.ReasoningContent != "" { + if reasoning := choice.Delta.reasoningText(); reasoning != "" { sendEvent(ctx, events, zeroruntime.StreamEvent{ Type: zeroruntime.StreamEventReasoning, - Content: choice.Delta.ReasoningContent, + Content: reasoning, }) } if choice.Delta.Content != "" { @@ -317,6 +317,13 @@ func (provider *Provider) emitChunk( } } +func (delta streamDelta) reasoningText() string { + if delta.ReasoningContent != "" { + return delta.ReasoningContent + } + return delta.Reasoning +} + // mapFinishReason maps OpenAI's finish_reason onto the runtime's normalized // terminal reasons. A normal finish ("stop"/"tool_calls"/"") returns "". func mapFinishReason(reason string) string { diff --git a/internal/providers/openai/provider_test.go b/internal/providers/openai/provider_test.go index 69d7e9027..5c76a8d8e 100644 --- a/internal/providers/openai/provider_test.go +++ b/internal/providers/openai/provider_test.go @@ -227,6 +227,39 @@ func TestStreamCompletionEmitsReasoningContentDeltas(t *testing.T) { } } +func TestStreamCompletionEmitsReasoningAliasDeltas(t *testing.T) { + provider := newTestProvider(t, func(w http.ResponseWriter, r *http.Request) { + writeSSE(w, `{"choices":[{"delta":{"reasoning":"Thinking. "}}]}`) + writeSSE(w, `{"choices":[{"delta":{"reasoning":"Answering now."}}]}`) + writeSSE(w, `[DONE]`) + }) + + events := collectProviderEvents(t, provider) + reasoning := eventsOfType(events, zeroruntime.StreamEventReasoning) + if len(reasoning) != 2 { + t.Fatalf("reasoning events = %#v, want two reasoning deltas", reasoning) + } + if reasoning[0].Content != "Thinking. " || reasoning[1].Content != "Answering now." { + t.Fatalf("unexpected reasoning events: %#v", reasoning) + } + if text := eventsOfType(events, zeroruntime.StreamEventText); len(text) != 0 { + t.Fatalf("reasoning must not emit text events, got %#v", text) + } +} + +func TestStreamCompletionPrefersReasoningContentOverAlias(t *testing.T) { + provider := newTestProvider(t, func(w http.ResponseWriter, r *http.Request) { + writeSSE(w, `{"choices":[{"delta":{"reasoning_content":"standard","reasoning":"alias"}}]}`) + writeSSE(w, `[DONE]`) + }) + + events := collectProviderEvents(t, provider) + reasoning := eventsOfType(events, zeroruntime.StreamEventReasoning) + if len(reasoning) != 1 || reasoning[0].Content != "standard" { + t.Fatalf("reasoning events = %#v, want standard reasoning_content", reasoning) + } +} + func TestStreamCompletionEmitsReasoningBeforeRegularContent(t *testing.T) { provider := newTestProvider(t, func(w http.ResponseWriter, r *http.Request) { writeSSE(w, `{"choices":[{"delta":{"reasoning_content":"Thinking. ","content":"Answer."}}]}`) diff --git a/internal/providers/openai/types.go b/internal/providers/openai/types.go index f093ce30b..54511a6f0 100644 --- a/internal/providers/openai/types.go +++ b/internal/providers/openai/types.go @@ -78,6 +78,7 @@ type streamChoice struct { type streamDelta struct { Content string `json:"content"` ReasoningContent string `json:"reasoning_content"` + Reasoning string `json:"reasoning"` ToolCalls []streamToolCallDelta `json:"tool_calls"` } diff --git a/internal/streamjson/streamjson.go b/internal/streamjson/streamjson.go index 9f904cccd..fe5bae6ad 100644 --- a/internal/streamjson/streamjson.go +++ b/internal/streamjson/streamjson.go @@ -21,6 +21,7 @@ type EventType string const ( EventRunStart EventType = "run_start" EventText EventType = "text" + EventReasoning EventType = "reasoning" EventToolCall EventType = "tool_call" EventPermission EventType = "permission" EventPermissionRequest EventType = "permission_request" diff --git a/internal/zeroruntime/helpers.go b/internal/zeroruntime/helpers.go index 01e9188a1..0a64f8d7e 100644 --- a/internal/zeroruntime/helpers.go +++ b/internal/zeroruntime/helpers.go @@ -21,6 +21,9 @@ type CollectedStream struct { // thinking blocks) that must be replayed on the next turn. Empty for providers // or runs without extended thinking. ReasoningBlocks []ReasoningBlock + // HasReasoning records whether the provider streamed reasoning deltas. The + // deltas remain non-answer content, but they still prove the turn was live. + HasReasoning bool } // Truncated reports whether the response ended for a non-normal reason (the @@ -132,6 +135,9 @@ func CollectStreamWithOptions(ctx context.Context, events <-chan StreamEvent, op options.OnText(event.Content) } case StreamEventReasoning: + if strings.TrimSpace(event.Content) != "" { + collected.HasReasoning = true + } if options.OnReasoning != nil { options.OnReasoning(event.Content) } diff --git a/internal/zeroruntime/provider_test.go b/internal/zeroruntime/provider_test.go index c90badd15..95a991c66 100644 --- a/internal/zeroruntime/provider_test.go +++ b/internal/zeroruntime/provider_test.go @@ -312,6 +312,9 @@ func TestCollectStreamWithOptionsEmitsTextReasoningAndUsageCallbacks(t *testing. if collected.Text != "Hello zero" { t.Fatalf("text = %q, want Hello zero", collected.Text) } + if !collected.HasReasoning { + t.Fatal("expected reasoning stream to mark collected turn as reasoning-bearing") + } if len(textDeltas) != 2 || textDeltas[0] != "Hello " || textDeltas[1] != "zero" { t.Fatalf("unexpected text callbacks: %#v", textDeltas) }