From ada183370d972dc639c833a1115c3419ec981815 Mon Sep 17 00:00:00 2001 From: phantom5099 <1011668688@qq.com> Date: Tue, 7 Apr 2026 22:59:55 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix(provider):=E6=A8=A1=E5=9E=8B=E6=B5=81?= =?UTF-8?q?=E5=BC=8F=E8=BE=93=E5=87=BA=E4=B8=AD=E9=80=94=E6=88=AA=E6=96=AD?= =?UTF-8?q?=E4=BD=86=E7=95=8C=E9=9D=A2=E6=97=A0=E9=94=99=E8=AF=AF=E6=8F=90?= =?UTF-8?q?=E7=A4=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/provider/openai/openai_test.go | 24 ++++++---- internal/provider/openai/response.go | 12 +++++ internal/runtime/runtime.go | 17 ++++++- internal/runtime/runtime_test.go | 61 ++++++++++++++++++++++++- 4 files changed, 100 insertions(+), 14 deletions(-) diff --git a/internal/provider/openai/openai_test.go b/internal/provider/openai/openai_test.go index 33179a2e..d16edc56 100644 --- a/internal/provider/openai/openai_test.go +++ b/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 diff --git a/internal/provider/openai/response.go b/internal/provider/openai/response.go index 02b7b9b5..5026a15d 100644 --- a/internal/provider/openai/response.go +++ b/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 } diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index 7b727e08..0ea4361b 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -34,8 +34,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 创建并初始化一个空的流式事件累积器。 @@ -294,6 +295,8 @@ func (s *Service) Run(ctx context.Context, input UserInput) error { return s.handleRunError(ctx, input.RunID, session.ID, err) } if len(assistant.ToolCalls) == 0 { + log.Printf("[DEBUG-RUNTIME] No tool calls in assistant response, content length=%d, emitting EventAgentDone", + len(assistant.Content)) s.emit(ctx, EventAgentDone, input.RunID, session.ID, assistant) return nil } @@ -537,6 +540,9 @@ func handleProviderStreamEvent( if _, err := event.MessageDoneValue(); err != nil { return err } + if acc != nil { + acc.messageDone = true + } default: return fmt.Errorf("runtime: unsupported provider stream event type %q", event.Type) } @@ -679,6 +685,10 @@ func (s *Service) callProviderWithRetry( err = modelProvider.Chat(ctx, req, streamEvents) close(streamEvents) forwardErr := <-streamDone + + // [DEBUG] 打印 provider.Chat() 返回状态和转发错误 + log.Printf("[DEBUG-RUNTIME] provider.Chat() returned: err=%v, forwardErr=%v", err, forwardErr) + if forwardErr != nil { if err != nil { return nil, fmt.Errorf("runtime: provider stream handling failed after provider error: %v: %w", err, forwardErr) @@ -687,6 +697,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 } diff --git a/internal/runtime/runtime_test.go b/internal/runtime/runtime_test.go index 0fe9f86a..b899eb56 100644 --- a/internal/runtime/runtime_test.go +++ b/internal/runtime/runtime_test.go @@ -121,6 +121,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] @@ -153,6 +160,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 @@ -469,7 +486,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) } @@ -520,7 +537,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) @@ -553,6 +570,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() @@ -1414,6 +1466,7 @@ func TestServiceRunErrorPaths(t *testing.T) { } } events <- providertypes.NewTextDeltaStreamEvent("recovered") + events <- providertypes.NewMessageDoneStreamEvent("stop", nil) return nil }, } @@ -2213,6 +2266,7 @@ func TestServiceSerializesRunAndCompact(t *testing.T) { } <-unblockProvider events <- providertypes.NewTextDeltaStreamEvent("done") + events <- providertypes.NewMessageDoneStreamEvent("stop", nil) return nil }, } @@ -2462,6 +2516,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) @@ -2491,6 +2546,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) @@ -2730,6 +2786,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 { From 0292f648e5adb777c429533487cfc61514832f36 Mon Sep 17 00:00:00 2001 From: phantom5099 <1011668688@qq.com> Date: Wed, 8 Apr 2026 15:35:46 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(runtime)=EF=BC=9A=E8=A7=A3=E5=86=B3?= =?UTF-8?q?=E5=86=B2=E7=AA=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/runtime/runtime.go | 32 ++++++++++++++++++-------------- 1 file changed, 18 insertions(+), 14 deletions(-) diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index 8f38a8a4..2e13466d 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -133,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。 @@ -532,12 +532,16 @@ func handleProviderStreamEvent( acc.accumulateToolCallDelta(payload.Index, payload.ID, payload.ArgumentsDelta) } case providertypes.StreamEventMessageDone: - if _, err := event.MessageDoneValue(); err != nil { + payload, err := event.MessageDoneValue() + if err != nil { return err } if acc != nil { acc.messageDone = true } + if onMessageDone != nil { + onMessageDone(payload) + } default: return fmt.Errorf("runtime: unsupported provider stream event type %q", event.Type) } @@ -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, })