Fallback when streaming starts fail

Author: SqlRush

docs/claude-code-go-rewrite-plan.md

@@ -135,6 +135,7 @@ test/parity/                 # golden tests against TS/official behavior
 - 本轮补充：Anthropic client 构造请求时会把显式 beta headers 与请求内容推导出的 beta headers 合并去重；当 messages/system content 或 tool definitions 中出现 `cache_control`/`cache_reference`，以及 messages 中出现 `cache_edits` payload 时，会自动带上 prompt-caching scope 与 cache-editing beta，避免 prompt-cache/cache-editing 请求依赖外部环境变量手动配置。
 - 本轮补充：动态 beta header 推导继续覆盖 strict tool structured-output 请求和 `[1m]` context model 请求；当 request tools 中存在 `strict:true` 时自动追加 structured-outputs beta，当 request model 以 `[1m]` 结尾时自动追加 context-1m beta，并继续和显式 beta 去重合并。
 - 本轮补充：Anthropic client 新增可选 access-token provider，非 streaming 与 streaming request 在 401/403 authentication/permission error 时会强制 refresh OAuth access token 并重试一次当前 payload；headless CLI 的 OAuth credentials 路径会把 file/env refresh token provider 传入 client，刷新后复用既有 credential store 持久化。
+- 本轮补充：conversation runner 的 streaming request 现在在未收到任何 stream event 前失败时，会用同一请求安全回退到 non-streaming `CreateMessage`；一旦已有 stream event 对外发出，错误会原样返回，避免重复生成半截输出。
 
 ### M4: Tool framework、permissions、sandbox
 

docs/first-second-parity-audit.md

@@ -93,7 +93,7 @@ Anthropic API 和 conversation：
 - Model registry now uses current Claude model IDs and aliases from the source snapshot, including Sonnet 4.6, Opus 4.6, Haiku 4.5, canonical-name rendering, and `[1m]` context variants.
 - OAuth support now includes production OAuth config, scope parsing, Claude.ai scope detection, auth URL construction, PKCE verifier/challenge, state generation, and expiry checks.
 - Session/history support now includes CC-compatible prompt history references, pasted text/image placeholder parsing, paste-cache hashing and retrieval, `history.jsonl` append/load, current-session-first up-arrow ordering, ctrl+r-style deduped timestamped history, `CLAUDE_CODE_SKIP_PROMPT_HISTORY`, remote session event pagination helpers, lenient transcript loading, legacy progress parent-bridge recovery, compact-boundary pruning, snip removal/relink replay, metadata entry collection, leaf UUID calculation, conversation-chain reconstruction, orphaned parallel tool-result recovery, content-replacement record loading/reconstruction, tombstone metadata delete/relink replay, and tombstone-style transcript message removal with a size guard.
-- Anthropic API layer now covers streaming accumulation, usage update/accumulation semantics, non-streaming max token cap, thinking-budget adjustment, retry/backoff with `Retry-After` and `x-should-retry`, context-overflow `max_tokens` retry adjustment, OAuth access-token refresh and retry on unauthorized non-streaming/streaming requests when a refresh provider is configured, beta-header dedupe, request-content beta latching for prompt-cache/cache-editing payloads, strict tool structured-output requests, and `[1m]` context models, custom request headers, basic prompt cache breakpoint/cache-reference/cache-edits placement, prompt dump JSONL capture for init/new user messages/non-streaming responses/stream chunks, and CC-compatible USD cost calculation for known Claude models including cache read/write, web search requests, and Opus 4.6 fast-tier pricing.
+- Anthropic API layer now covers streaming accumulation, safe pre-event streaming-to-non-streaming fallback in the conversation runner, usage update/accumulation semantics, non-streaming max token cap, thinking-budget adjustment, retry/backoff with `Retry-After` and `x-should-retry`, context-overflow `max_tokens` retry adjustment, OAuth access-token refresh and retry on unauthorized non-streaming/streaming requests when a refresh provider is configured, beta-header dedupe, request-content beta latching for prompt-cache/cache-editing payloads, strict tool structured-output requests, and `[1m]` context models, custom request headers, basic prompt cache breakpoint/cache-reference/cache-edits placement, prompt dump JSONL capture for init/new user messages/non-streaming responses/stream chunks, and CC-compatible USD cost calculation for known Claude models including cache read/write, web search requests, and Opus 4.6 fast-tier pricing.
 - Tool runtime now includes concurrency partitioning, ordered concurrent execution, interrupt behavior/defaults, max result size metadata, oversized result persistence, pre/post/permission-denied/permission-request hook dispatch, settings and local-plugin command-backed/HTTP hook execution for synchronous tool and conversation lifecycle hooks, hook-driven input updates/blocking/permission-request allow-deny, prompt/compact context injection, executor hook phase progress events, lifecycle progress events, retry/fallback model breadcrumbs, final-result/structured-error model attempt lists, structured API error status/type/request-id fields, headless stream-json progress events, snake_case token warning NDJSON payloads, lightweight compact NDJSON metadata, and pre-call cancellation checks.
 - Conversation runner can now use streaming clients, aggregate stream events into assistant messages, run tool calls through the orchestrator, preserve transcript append behavior, and apply CC-style per-message aggregate tool-result budget replacement before API requests with persisted replacement records for resume.
 
@@ -106,7 +106,7 @@ The following items remain incomplete and must not be treated as done:
 - Interactive permission prompt flow: REPL dialogs, bridge/channel/swarm permission relays, user feedback images, prompt race handling, cancellation.
 - Full filesystem permission parity gaps that remain: skill-scope allow suggestions, complete local/local-jsx slash-command execution wiring, command permission UI/SDK surfacing, forked/remote/MCP/plugin/bundled SkillTool behavior, skill prompt shell injection, complete auto-memory override policy, and deeper platform-specific Windows/WSL bypass handling.
 - Full tool execution parity gaps that remain: complete hook runtime policy beyond the current settings command hook path, MCP elicitation, complete SDK control event surface beyond current progress NDJSON, mid-call cancellation for concrete tools, background task behavior, telemetry, schema-not-sent hints, and concrete tool-specific semantics.
-- Complete Anthropic API parity gaps that remain: dynamic beta-header latching for any remaining provider-specific feature gates beyond prompt-cache/cache-editing/strict-output/1M-context requests, ant-only dump gating and `/issue` integration, cost tracker/session restore integration, streaming-to-non-streaming fallback, gateway/proxy-specific headers, first-party/Bedrock/Vertex/Foundry client setup, deeper OAuth account/profile lifecycle beyond request-time refresh retry, fast-mode retry/cooldown semantics, persistent unattended retry heartbeats, full prompt-cache editing lifecycle, and provider-specific cache behavior.
+- Complete Anthropic API parity gaps that remain: dynamic beta-header latching for any remaining provider-specific feature gates beyond prompt-cache/cache-editing/strict-output/1M-context requests, ant-only dump gating and `/issue` integration, cost tracker/session restore integration, post-handshake streaming failure semantics, gateway/proxy-specific headers, first-party/Bedrock/Vertex/Foundry client setup, deeper OAuth account/profile lifecycle beyond request-time refresh retry, fast-mode retry/cooldown semantics, persistent unattended retry heartbeats, full prompt-cache editing lifecycle, and provider-specific cache behavior.
 - Full conversation/query loop: stop hooks, compact/auto-compact, token budget escalation, resume, SDK JSON/NDJSON control events, status updates, rate-limit handling, model switch breadcrumbs, side questions.
 - Full settings parity gaps that remain: complete Zod-equivalent validation messages, MDM/HKCU settings, managed-settings drop-in loading, settings cache/change detector, schema generation, plugin-only customization enforcement, all strict marketplace validations, and live reload/app-state sync.
 - Full session/history parity gaps that remain: large-file optimized transcript loading, preserved-segment edge cases beyond current relink/prune support, content-replacement feature-flag/runtime override and inherited subagent gap-fill details, complete async prompt-history lifecycle parity beyond current lock/buffer/undo paths, full pasted-image processing/runtime integration beyond current prompt image-cache/image-block/metadata path, remaining remote-history edge cases, sidechain/subagent transcript layout, and all session metadata entry types.

internal/conversation/run.go

@@ -5475,13 +5475,21 @@ func (r Runner) applyToolResultBudget(history []contracts.Message) ([]contracts.
 func (r Runner) createMessage(ctx context.Context, request anthropic.Request) (*anthropic.Response, error) {
 	if r.UseStreaming {
 		if streamer, ok := r.Client.(StreamingMessageClient); ok {
-			request.Stream = true
+			streamRequest := request
+			streamRequest.Stream = true
 			acc := anthropic.NewStreamAccumulator()
-			if err := streamer.StreamMessages(ctx, request, func(event anthropic.StreamEvent) error {
+			seenStreamEvent := false
+			if err := streamer.StreamMessages(ctx, streamRequest, func(event anthropic.StreamEvent) error {
+				seenStreamEvent = true
 				eventCopy := event
-				r.emit(Event{Type: EventStreamEvent, StreamEvent: &eventCopy, Model: request.Model})
+				r.emit(Event{Type: EventStreamEvent, StreamEvent: &eventCopy, Model: streamRequest.Model})
 				return acc.Add(event)
 			}); err != nil {
+				if !seenStreamEvent {
+					fallbackRequest := request
+					fallbackRequest.Stream = false
+					return r.Client.CreateMessage(ctx, fallbackRequest)
+				}
 				return nil, err
 			}
 			return acc.Finish(), nil

internal/conversation/run_test.go

@@ -43,9 +43,10 @@ type fakeCall struct {
 }
 
 type fakeClient struct {
-	calls    []fakeCall
-	requests []anthropic.Request
-	streams  [][]anthropic.StreamEvent
+	calls      []fakeCall
+	requests   []anthropic.Request
+	streams    [][]anthropic.StreamEvent
+	streamErrs []error
 }
 
 type fakeRunnerMCPClient struct {
@@ -72,16 +73,24 @@ func (f *fakeClient) CreateMessage(ctx context.Context, req anthropic.Request) (
 
 func (f *fakeClient) StreamMessages(ctx context.Context, req anthropic.Request, handle func(anthropic.StreamEvent) error) error {
 	f.requests = append(f.requests, req)
-	if len(f.streams) == 0 {
+	if len(f.streams) == 0 && len(f.streamErrs) == 0 {
 		return anthropic.APIError{StatusCode: http.StatusInternalServerError, Type: "test_error", Message: "no fake stream configured"}
 	}
-	events := f.streams[0]
-	f.streams = f.streams[1:]
+	var events []anthropic.StreamEvent
+	if len(f.streams) > 0 {
+		events = f.streams[0]
+		f.streams = f.streams[1:]
+	}
 	for _, event := range events {
 		if err := handle(event); err != nil {
 			return err
 		}
 	}
+	if len(f.streamErrs) > 0 {
+		err := f.streamErrs[0]
+		f.streamErrs = f.streamErrs[1:]
+		return err
+	}
 	return nil
 }
 
@@ -7075,6 +7084,62 @@ func TestRunnerCanUseStreamingClient(t *testing.T) {
 	}
 }
 
+func TestRunnerFallsBackToNonStreamingWhenStreamFailsBeforeEvents(t *testing.T) {
+	client := &fakeClient{
+		streamErrs: []error{anthropic.APIError{StatusCode: http.StatusServiceUnavailable, Type: "overloaded_error", Message: "stream unavailable"}},
+		calls: []fakeCall{{
+			response: &anthropic.Response{ID: "msg_fallback", Type: "message", Role: "assistant", Model: "sonnet", Content: []contracts.ContentBlock{contracts.NewTextBlock("fallback ok")}},
+		}},
+	}
+	runner := Runner{
+		Client:       client,
+		Model:        "sonnet",
+		MaxTokens:    64,
+		UseStreaming: true,
+	}
+	result, err := runner.RunTurn(context.Background(), nil, messages.UserText("hello"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got := result.Assistant.Content[0].Text; got != "fallback ok" {
+		t.Fatalf("assistant text = %q", got)
+	}
+	if len(client.requests) != 2 {
+		t.Fatalf("requests = %#v", client.requests)
+	}
+	if !client.requests[0].Stream {
+		t.Fatalf("first request should be streaming: %#v", client.requests[0])
+	}
+	if client.requests[1].Stream {
+		t.Fatalf("fallback request should be non-streaming: %#v", client.requests[1])
+	}
+}
+
+func TestRunnerDoesNotFallbackAfterStreamEvents(t *testing.T) {
+	client := &fakeClient{
+		streams: [][]anthropic.StreamEvent{{
+			{Type: "message_start", Message: &anthropic.Response{ID: "msg_partial", Type: "message", Role: "assistant", Model: "sonnet"}},
+		}},
+		streamErrs: []error{anthropic.APIError{StatusCode: http.StatusServiceUnavailable, Type: "overloaded_error", Message: "stream interrupted"}},
+		calls: []fakeCall{{
+			response: &anthropic.Response{ID: "msg_fallback", Type: "message", Role: "assistant", Model: "sonnet", Content: []contracts.ContentBlock{contracts.NewTextBlock("should not run")}},
+		}},
+	}
+	runner := Runner{
+		Client:       client,
+		Model:        "sonnet",
+		MaxTokens:    64,
+		UseStreaming: true,
+	}
+	_, err := runner.RunTurn(context.Background(), nil, messages.UserText("hello"))
+	if err == nil {
+		t.Fatal("expected stream error")
+	}
+	if len(client.requests) != 1 {
+		t.Fatalf("requests = %#v", client.requests)
+	}
+}
+
 func conversationLSPHelperDefinition() lsppkg.ServerDefinition {
 	return lsppkg.ServerDefinition{
 		Name:           "conversation-lsp-helper",