fix(provider):模型流式输出中途截断但界面无错误提示

internal/provider/openai/openai_test.go

@@ -37,12 +37,12 @@ func TestWithTransport(t *testing.T) {
 	customTransport := &http.Transport{}
 	cfg := resolvedConfig("", "")
 
-	provider, err := New(cfg, withTransport(customTransport))
+	p, err := New(cfg, withTransport(customTransport))
 	if err != nil {
 		t.Fatalf("New() error = %v", err)
 	}
 
-	if provider.client.Transport != customTransport {
+	if p.client.Transport != customTransport {
 		t.Fatal("expected custom transport to be set")
 	}
 }
@@ -95,11 +95,11 @@ func TestNewDefaultTransportWhenNoOption(t *testing.T) {
 	t.Parallel()
 
 	cfg := resolvedConfig("", "")
-	provider, err := New(cfg)
+	p, err := New(cfg)
 	if err != nil {
 		t.Fatalf("New() error = %v", err)
 	}
-	if provider.client.Transport == nil {
+	if p.client.Transport == nil {
 		t.Fatal("expected default transport to be set")
 	}
 }
@@ -130,13 +130,13 @@ func TestDiscoverModels(t *testing.T) {
 	}))
 	defer server.Close()
 
-	provider, err := New(resolvedConfig(server.URL, ""))
+	p, err := New(resolvedConfig(server.URL, ""))
 	if err != nil {
 		t.Fatalf("New() error = %v", err)
 	}
-	provider.client = server.Client()
+	p.client = server.Client()
 
-	models, err := provider.DiscoverModels(context.Background())
+	models, err := p.DiscoverModels(context.Background())
 	if err != nil {
 		t.Fatalf("DiscoverModels() error = %v", err)
 	}
@@ -490,6 +490,7 @@ func TestConsumeStream_MultiLineDataPayload(t *testing.T) {
 
 	sseData := `data: {"id":"a","choices":[{"delta":{"content":"part1"},"finish_reason":""}]}
 data: {"id":"b","choices":[{"delta":{"content":"part2"},"finish_reason":"stop"}]}
+data: [DONE]
 
 `
 	events := make(chan providertypes.StreamEvent, 8)
@@ -503,7 +504,7 @@ data: {"id":"b","choices":[{"delta":{"content":"part2"},"finish_reason":"stop"}]
 	}
 }
 
-func TestConsumeStream_EOFWithoutDone(t *testing.T) {
+func TestConsumeStream_EOFWithoutDoneReturnsInterrupted(t *testing.T) {
 	t.Setenv(config.OpenAIDefaultAPIKeyEnv, "test-key")
 
 	p, err := New(resolvedConfig("", ""))
@@ -515,8 +516,11 @@ func TestConsumeStream_EOFWithoutDone(t *testing.T) {
 `
 	events := make(chan providertypes.StreamEvent, 4)
 	err = p.consumeStream(context.Background(), strings.NewReader(sseData), events)
-	if err != nil {
-		t.Fatalf("consumeStream() error = %v", err)
+	if err == nil {
+		t.Fatal("expected interrupted error for EOF without [DONE]")
+	}
+	if !errors.Is(err, provider.ErrStreamInterrupted) {
+		t.Fatalf("expected ErrStreamInterrupted, got %v", err)
 	}
 	drained := drainStreamEvents(events)
 	var foundText bool

internal/provider/openai/response.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log"
 	"strings"
 
 	"neo-code/internal/provider"
@@ -67,6 +68,7 @@ func (p *Provider) consumeStream(
 
 	// finishStream 统一的流结束处理：发送 message_done 事件。
 	finishStream := func() error {
+		log.Printf("[DEBUG-STREAM] finishStream called: finishReason=%q, done=%v", finishReason, done)
 		return emitMessageDone(ctx, events, finishReason, &usage)
 	}
 
@@ -114,6 +116,16 @@ func (p *Provider) consumeStream(
 		}
 
 		if errors.Is(err, io.EOF) {
+			// [DEBUG] 流 EOF 时打印关键状态，用于诊断截断原因
+			log.Printf("[DEBUG-STREAM] EOF reached: done=%v, finishReason=%q, totalRead=%d, toolCallCount=%d",
+				done, finishReason, reader.totalRead, len(toolCalls))
+			if !done {
+				log.Printf("[DEBUG-STREAM] WARNING: stream ended WITHOUT [DONE] marker — treating as interruption")
+				if flushErr := flushPendingData(); flushErr != nil {
+					return flushErr
+				}
+				return fmt.Errorf("%w: missing [DONE] marker before EOF", provider.ErrStreamInterrupted)
+			}
 			if flushErr := flushPendingData(); flushErr != nil {
 				return flushErr
 			}

internal/runtime/runtime.go

@@ -35,8 +35,9 @@ const (
 // streamAccumulator 在流式事件处理过程中累积本轮对话需要持久化的助手消息状态，
 // 包括文本内容和工具调用列表。
 type streamAccumulator struct {
-	content   strings.Builder
-	toolCalls map[int]*providertypes.ToolCall
+	content     strings.Builder
+	toolCalls   map[int]*providertypes.ToolCall
+	messageDone bool
 }
 
 // newStreamAccumulator 创建并初始化一个空的流式事件累积器。
@@ -132,17 +133,17 @@ type ProviderFactory interface {
 }
 
 type Service struct {
-	configManager   *config.Manager      // 配置管理器，提供当前选中的 provider、model、workdir 等配置读取能力。
-	sessionStore    agentsession.Store   // 会话持久化接口，负责保存和加载聊天会话。
-	toolManager     tools.Manager        // 工具管理器，统一工具 schema 暴露与执行入口。
-	providerFactory ProviderFactory      // Provider 工厂接口，根据配置动态创建具体的 provider 实例。
-	contextBuilder  agentcontext.Builder // 上下文构建器，负责组装 system prompt 与本轮发给模型的消息上下文。
-	compactRunner   contextcompact.Runner
-	events          chan RuntimeEvent
-	operationMu     sync.Mutex         // 运行级互斥：串行化 Run 与 Compact，避免并发写同一会话。
-	runMu           sync.Mutex         // 仅保护 activeRun* 字段的并发读写。
-	activeRunToken  uint64             // 当前活跃运行的令牌标识，用于标记正在执行的 Run 实例。
-	nextRunToken    uint64             // 下一个运行令牌的递增计数器，用于区分不同 Run 的生命周期。
+	configManager       *config.Manager      // 配置管理器，提供当前选中的 provider、model、workdir 等配置读取能力。
+	sessionStore        agentsession.Store   // 会话持久化接口，负责保存和加载聊天会话。
+	toolManager         tools.Manager        // 工具管理器，统一工具 schema 暴露与执行入口。
+	providerFactory     ProviderFactory      // Provider 工厂接口，根据配置动态创建具体的 provider 实例。
+	contextBuilder      agentcontext.Builder // 上下文构建器，负责组装 system prompt 与本轮发给模型的消息上下文。
+	compactRunner       contextcompact.Runner
+	events              chan RuntimeEvent
+	operationMu         sync.Mutex         // 运行级互斥：串行化 Run 与 Compact，避免并发写同一会话。
+	runMu               sync.Mutex         // 仅保护 activeRun* 字段的并发读写。
+	activeRunToken      uint64             // 当前活跃运行的令牌标识，用于标记正在执行的 Run 实例。
+	nextRunToken        uint64             // 下一个运行令牌的递增计数器，用于区分不同 Run 的生命周期。
 	activeRunCancel     context.CancelFunc // 当前活跃 Run 的取消函数。
 	sessionInputTokens  int                // 当前会话累计输入 token。
 	sessionOutputTokens int                // 当前会话累计输出 token。
@@ -535,6 +536,9 @@ func handleProviderStreamEvent(
 		if err != nil {
 			return err
 		}
+		if acc != nil {
+			acc.messageDone = true
+		}
 		if onMessageDone != nil {
 			onMessageDone(payload)
 		}
@@ -579,8 +583,8 @@ func (s *Service) forwardProviderEvents(
 						s.sessionInputTokens += done.Usage.InputTokens
 						s.sessionOutputTokens += done.Usage.OutputTokens
 						s.emit(ctx, EventTokenUsage, runID, sessionID, TokenUsagePayload{
-							InputTokens:        done.Usage.InputTokens,
-							OutputTokens:       done.Usage.OutputTokens,
+							InputTokens:         done.Usage.InputTokens,
+							OutputTokens:        done.Usage.OutputTokens,
 							SessionInputTokens:  s.sessionInputTokens,
 							SessionOutputTokens: s.sessionOutputTokens,
 						})
@@ -716,6 +720,9 @@ func (s *Service) callProviderWithRetry(
 		}
 
 		if err == nil {
+			if !acc.messageDone {
+				return nil, fmt.Errorf("%w: provider stream ended without message_done event", provider.ErrStreamInterrupted)
+			}
 			return acc, nil
 		}
 

internal/runtime/runtime_test.go

@@ -122,6 +122,13 @@ func (p *scriptedProvider) Chat(ctx context.Context, req providertypes.ChatReque
 				return ctx.Err()
 			}
 		}
+		if callIndex >= len(p.responses) && !streamContainsMessageDone(p.streams[callIndex]) {
+			select {
+			case events <- providertypes.NewMessageDoneStreamEvent("", nil):
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		}
 	}
 	if callIndex < len(p.responses) {
 		response := p.responses[callIndex]
@@ -154,6 +161,16 @@ func (p *scriptedProvider) Chat(ctx context.Context, req providertypes.ChatReque
 	return nil
 }
 
+// streamContainsMessageDone 判断测试流中是否已显式包含结束事件，避免辅助 provider 重复补发 message_done。
+func streamContainsMessageDone(events []providertypes.StreamEvent) bool {
+	for _, event := range events {
+		if event.Type == providertypes.StreamEventMessageDone {
+			return true
+		}
+	}
+	return false
+}
+
 type scriptedProviderFactory struct {
 	provider provider.Provider
 	calls    int
@@ -470,7 +487,7 @@ func TestServiceRunMergesLateToolCallMetadata(t *testing.T) {
 		},
 	}
 
-	service := NewWithFactory(manager, registry, store, &scriptedProviderFactory{provider: scripted}, nil)
+	service := NewWithFactory(manager, registry, store, &scriptedProviderFactory{provider: scripted}, &stubContextBuilder{})
 	if err := service.Run(context.Background(), UserInput{RunID: "run-late-tool-metadata", Content: "edit"}); err != nil {
 		t.Fatalf("Run() error = %v", err)
 	}
@@ -521,7 +538,7 @@ func TestServiceRunRejectsToolCallWithoutID(t *testing.T) {
 		},
 	}
 
-	service := NewWithFactory(manager, registry, store, &scriptedProviderFactory{provider: scripted}, nil)
+	service := NewWithFactory(manager, registry, store, &scriptedProviderFactory{provider: scripted}, &stubContextBuilder{})
 	err := service.Run(context.Background(), UserInput{RunID: "run-missing-tool-id", Content: "edit"})
 	if err == nil || !containsError(err, "without id") {
 		t.Fatalf("expected missing tool id error, got %v", err)
@@ -554,6 +571,41 @@ func TestServiceRunRejectsMalformedProviderStreamEvent(t *testing.T) {
 	}
 }
 
+func TestServiceRunRejectsProviderCompletionWithoutMessageDone(t *testing.T) {
+	t.Parallel()
+
+	manager := newRuntimeConfigManager(t)
+	store := newMemoryStore()
+	registry := tools.NewRegistry()
+	registry.Register(&stubTool{name: "filesystem_read_file", content: "default"})
+
+	scripted := &scriptedProvider{
+		chatFn: func(ctx context.Context, req providertypes.ChatRequest, events chan<- providertypes.StreamEvent) error {
+			select {
+			case events <- providertypes.NewTextDeltaStreamEvent("partial"):
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+			return nil
+		},
+	}
+
+	service := NewWithFactory(manager, registry, store, &scriptedProviderFactory{provider: scripted}, &stubContextBuilder{})
+	err := service.Run(context.Background(), UserInput{RunID: "run-missing-message-done", Content: "hello"})
+	if err == nil || !containsError(err, "without message_done") {
+		t.Fatalf("expected missing message_done error, got %v", err)
+	}
+
+	events := collectRuntimeEvents(service.Events())
+	assertEventSequence(t, events, []EventType{EventUserMessage, EventAgentChunk, EventError})
+	assertNoEventType(t, events, EventAgentDone)
+
+	session := onlySession(t, store)
+	if len(session.Messages) != 1 || session.Messages[0].Role != providertypes.RoleUser {
+		t.Fatalf("expected only user message to persist after missing message_done, got %+v", session.Messages)
+	}
+}
+
 func TestServiceRunMalformedProviderStreamEventDoesNotDeadlock(t *testing.T) {
 	t.Parallel()
 
@@ -1415,6 +1467,7 @@ func TestServiceRunErrorPaths(t *testing.T) {
 							}
 						}
 						events <- providertypes.NewTextDeltaStreamEvent("recovered")
+						events <- providertypes.NewMessageDoneStreamEvent("stop", nil)
 						return nil
 					},
 				}
@@ -2214,6 +2267,7 @@ func TestServiceSerializesRunAndCompact(t *testing.T) {
 			}
 			<-unblockProvider
 			events <- providertypes.NewTextDeltaStreamEvent("done")
+			events <- providertypes.NewMessageDoneStreamEvent("stop", nil)
 			return nil
 		},
 	}
@@ -2463,6 +2517,7 @@ func newRuntimeConfigManagerWithProviderEnvs(t *testing.T, providerEnvs map[stri
 	t.Helper()
 
 	apiKeyEnv := runtimeTestAPIKeyEnv(t)
+	defaultWorkdir := t.TempDir()
 	restoreRuntimeEnv(t, apiKeyEnv)
 	if err := os.Setenv(apiKeyEnv, "test-key"); err != nil {
 		t.Fatalf("set env: %v", err)
@@ -2492,6 +2547,7 @@ func newRuntimeConfigManagerWithProviderEnvs(t *testing.T, providerEnvs map[stri
 	if err := manager.Update(context.Background(), func(cfg *config.Config) error {
 		cfg.ToolTimeoutSec = 1
 		cfg.MaxLoops = 4
+		cfg.Workdir = defaultWorkdir
 		return nil
 	}); err != nil {
 		t.Fatalf("update config: %v", err)
@@ -2731,6 +2787,7 @@ func TestIsRetryableProviderError(t *testing.T) {
 		{"non-retryable provider error", &provider.ProviderError{Retryable: false}, false},
 		{"plain error", errors.New("something failed"), false},
 		{"wrapped retryable", fmt.Errorf("wrapped: %w", &provider.ProviderError{Retryable: true}), true},
+		{"stream interrupted sentinel", provider.ErrStreamInterrupted, false},
 	}
 
 	for _, tt := range tests {