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/4] =?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/4] =?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, }) From 55491d3f2ea4397f9d7c2d98ca1585b552a19457 Mon Sep 17 00:00:00 2001 From: phantom5099 <1011668688@qq.com> Date: Wed, 8 Apr 2026 21:04:21 +0800 Subject: [PATCH 3/4] =?UTF-8?q?pref(provider):=E4=BF=AE=E6=94=B9=E5=85=AC?= =?UTF-8?q?=E5=85=B1=E5=A5=91=E7=BA=A6=E5=B9=B6=E5=A2=9E=E5=8A=A0=E8=83=BD?= =?UTF-8?q?=E5=8A=9B=E6=A8=A1=E5=9E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/session-persistence-design.md | 44 ----------- internal/config/config_test.go | 16 ++-- internal/config/model.go | 2 +- internal/provider/capabilities.go | 43 ++++++++++ internal/provider/catalog/service_test.go | 2 +- internal/provider/openai/driver.go | 6 ++ internal/provider/openai/openai_test.go | 36 ++++----- internal/provider/openai/provider.go | 2 +- internal/provider/openai/request.go | 4 +- internal/provider/openai/response.go | 12 --- internal/provider/provider.go | 4 +- internal/provider/registry.go | 16 +++- internal/provider/registry_test.go | 23 +++++- internal/provider/types/request.go | 4 +- internal/provider/types/types_test.go | 4 +- internal/runtime/compact_generator.go | 5 +- internal/runtime/compact_generator_test.go | 39 +++++++++ internal/runtime/events.go | 6 +- internal/runtime/runtime.go | 36 +++++++-- internal/runtime/runtime_test.go | 92 ++++++++++++++++------ 20 files changed, 268 insertions(+), 128 deletions(-) delete mode 100644 docs/session-persistence-design.md create mode 100644 internal/provider/capabilities.go diff --git a/docs/session-persistence-design.md b/docs/session-persistence-design.md deleted file mode 100644 index fb22d26d..00000000 --- a/docs/session-persistence-design.md +++ /dev/null @@ -1,44 +0,0 @@ -# Session 持久化设计 - -## 模块职责与收口边界 -- `internal/session`：承载会话领域模型、存储抽象与 JSON 持久化实现，是唯一的会话持久化实现归属层 -- `internal/runtime`：只依赖 `internal/session` 提供的抽象与模型，负责会话保存时机与主循环编排，不再维护会话存储实现细节 -- `internal/tui`：仅消费 runtime 暴露的会话数据，不直接执行会话持久化 - -## 存储策略 -NeoCode 在 MVP 阶段使用 JSON 文件持久化 Session，以保持本地优先、易于调试和跨平台可移植。 - -## 数据模型 -- `Session`：完整消息历史以及 `id`、`title`、`updated_at`、`token_input_total`、`token_output_total` 等元信息 -- `Summary`：用于侧边栏的轻量摘要结构（原 `SessionSummary` 命名已统一收口为 `Summary`） - -### Token 持久化 -- `token_input_total` 和 `token_output_total` 分别记录会话累计输入和输出 token。 -- 使用 `omitempty` 标签，确保旧版 JSON 文件正常加载（零值不序列化）。 -- runtime 在每次 provider 调用后更新 session 的 token 字段，随 session save 一起持久化。 -- 会话加载时，runtime 从 session 恢复 token 计数器；新建会话时计数器清零。 -- 自动压缩成功后 token 计数器重置为零，并持久化到 session。 - -## 加载策略 -- `ListSummaries` 只读取渲染侧边栏所需的基础信息 -- `Load` 仅在用户真正进入某个会话时读取完整消息历史 -- `Save` 通过临时文件原子写入完整 Session - -## 命名策略 -- 新会话默认展示为 `New Session` -- 一旦持久化，runtime 会根据首轮用户消息生成简短标题 - -## 并发约束 -- `internal/session` 中的 Store 实现必须自行保护共享访问 -- 真正的保存时机由 runtime 决定，TUI 不负责直接触发磁盘写入 - -## 兼容性与演进说明 -- 会话持久化能力已从 runtime 侧实现中彻底收口到 `internal/session` -- 新增会话存储实现时，应优先在 `internal/session` 内扩展并通过接口注入 runtime，避免跨层实现 - -## 工作区隔离 - -- session 现按工作区隔离存储，目录规则为 `~/.neocode/projects//sessions/` -- 工作区哈希基于启动时确定的工作区根目录生成，而不是基于 `session.Workdir` -- `session.Workdir` 仍表示该会话当前实际执行命令时使用的目录，可被 `/cwd` 修改 -- 旧的全局 `~/.neocode/sessions/` 开发期数据不迁移、不回读 diff --git a/internal/config/config_test.go b/internal/config/config_test.go index d2d37542..20dc6c24 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -1231,10 +1231,10 @@ func TestDescriptorFromRawModel(t *testing.T) { t.Parallel() tests := []struct { - name string - raw map[string]any - want ModelDescriptor - wantOK bool + name string + raw map[string]any + want ModelDescriptor + wantOK bool }{ { name: "empty map returns false", @@ -1252,10 +1252,10 @@ func TestDescriptorFromRawModel(t *testing.T) { { name: "full descriptor", raw: map[string]any{ - "id": "gpt-4.1", - "display_name": "GPT-4.1", - "description": "desc", - "context_window": 128000, + "id": "gpt-4.1", + "display_name": "GPT-4.1", + "description": "desc", + "context_window": 128000, "max_output_tokens": 16384, }, want: ModelDescriptor{ diff --git a/internal/config/model.go b/internal/config/model.go index 9b8d80f1..4281b4b1 100644 --- a/internal/config/model.go +++ b/internal/config/model.go @@ -17,7 +17,7 @@ const ( DefaultWebFetchMaxResponseBytes int64 = 256 * 1024 DefaultCompactManualKeepRecentMessages = 10 DefaultCompactMaxSummaryChars = 1200 - DefaultAutoCompactInputTokenThreshold = 100000 + DefaultAutoCompactInputTokenThreshold = 100000 ) const ( diff --git a/internal/provider/capabilities.go b/internal/provider/capabilities.go new file mode 100644 index 00000000..2218d227 --- /dev/null +++ b/internal/provider/capabilities.go @@ -0,0 +1,43 @@ +package provider + +// CapabilitySupport 表示模型能力的确认状态。 +type CapabilitySupport string + +const ( + CapabilitySupportUnknown CapabilitySupport = "unknown" + CapabilitySupportUnsupported CapabilitySupport = "unsupported" + CapabilitySupportSupported CapabilitySupport = "supported" +) + +// ReasoningMode 表示模型推理能力的形态。 +type ReasoningMode string + +const ( + ReasoningModeUnknown ReasoningMode = "unknown" + ReasoningModeNone ReasoningMode = "none" + ReasoningModeNative ReasoningMode = "native" + ReasoningModeConfigurable ReasoningMode = "configurable" +) + +// DriverCapabilities 描述 driver 层本身是否能传输某类能力。 +type DriverCapabilities struct { + Streaming bool + ToolTransport bool + ModelDiscovery bool + ImageInputTransport bool +} + +// ModelCapabilities 描述具体模型的能力声明。 +type ModelCapabilities struct { + ToolCalling CapabilitySupport + ImageInput CapabilitySupport + ReasoningMode ReasoningMode + MaxContextTokens int + MaxOutputTokens int +} + +// ResolvedCapabilities 是 driver 能力与模型能力的合并视图。 +type ResolvedCapabilities struct { + Driver DriverCapabilities + Model ModelCapabilities +} diff --git a/internal/provider/catalog/service_test.go b/internal/provider/catalog/service_test.go index da3218dc..476a1f59 100644 --- a/internal/provider/catalog/service_test.go +++ b/internal/provider/catalog/service_test.go @@ -235,7 +235,7 @@ func containsModelDescriptorID(models []config.ModelDescriptor, modelID string) type catalogTestProvider struct{} -func (catalogTestProvider) Chat(ctx context.Context, req providertypes.ChatRequest, events chan<- providertypes.StreamEvent) error { +func (catalogTestProvider) Generate(ctx context.Context, req providertypes.GenerateRequest, events chan<- providertypes.StreamEvent) error { return nil } diff --git a/internal/provider/openai/driver.go b/internal/provider/openai/driver.go index fbde6820..5b41a4d9 100644 --- a/internal/provider/openai/driver.go +++ b/internal/provider/openai/driver.go @@ -31,5 +31,11 @@ func Driver() provider.DriverDefinition { } return p.DiscoverModels(ctx) }, + Capabilities: provider.DriverCapabilities{ + Streaming: true, + ToolTransport: true, + ModelDiscovery: true, + ImageInputTransport: false, + }, } } diff --git a/internal/provider/openai/openai_test.go b/internal/provider/openai/openai_test.go index d16edc56..a55c370f 100644 --- a/internal/provider/openai/openai_test.go +++ b/internal/provider/openai/openai_test.go @@ -287,7 +287,7 @@ func TestBuildRequest_EmptyModelReturnsError(t *testing.T) { client: &http.Client{}, } - _, buildErr := p.buildRequest(providertypes.ChatRequest{}) + _, buildErr := p.buildRequest(providertypes.GenerateRequest{}) if buildErr == nil { t.Fatal("expected error for empty model") } @@ -304,7 +304,7 @@ func TestBuildRequest_FallsBackToConfigModel(t *testing.T) { t.Fatalf("New() error = %v", err) } - payload, err := p.buildRequest(providertypes.ChatRequest{Messages: []providertypes.Message{{Role: "user", Content: "hi"}}}) + payload, err := p.buildRequest(providertypes.GenerateRequest{Messages: []providertypes.Message{{Role: "user", Content: "hi"}}}) if err != nil { t.Fatalf("buildRequest() error = %v", err) } @@ -321,7 +321,7 @@ func TestBuildRequest_RequestModelTakesPrecedence(t *testing.T) { t.Fatalf("New() error = %v", err) } - payload, err := p.buildRequest(providertypes.ChatRequest{Model: "gpt-4-custom", Messages: []providertypes.Message{{Role: "user", Content: "hi"}}}) + payload, err := p.buildRequest(providertypes.GenerateRequest{Model: "gpt-4-custom", Messages: []providertypes.Message{{Role: "user", Content: "hi"}}}) if err != nil { t.Fatalf("buildRequest() error = %v", err) } @@ -338,7 +338,7 @@ func TestBuildRequest_NoSystemPrompt(t *testing.T) { t.Fatalf("New() error = %v", err) } - payload, err := p.buildRequest(providertypes.ChatRequest{SystemPrompt: "", Messages: []providertypes.Message{{Role: "user", Content: "hi"}}}) + payload, err := p.buildRequest(providertypes.GenerateRequest{SystemPrompt: "", Messages: []providertypes.Message{{Role: "user", Content: "hi"}}}) if err != nil { t.Fatalf("buildRequest() error = %v", err) } @@ -357,7 +357,7 @@ func TestBuildRequest_NoTools(t *testing.T) { t.Fatalf("New() error = %v", err) } - payload, err := p.buildRequest(providertypes.ChatRequest{Messages: []providertypes.Message{{Role: "user", Content: "hi"}}, Tools: nil}) + payload, err := p.buildRequest(providertypes.GenerateRequest{Messages: []providertypes.Message{{Role: "user", Content: "hi"}}, Tools: nil}) if err != nil { t.Fatalf("buildRequest() error = %v", err) } @@ -374,7 +374,7 @@ func TestBuildRequest_EmptyToolsSlice(t *testing.T) { t.Fatalf("New() error = %v", err) } - payload, err := p.buildRequest(providertypes.ChatRequest{Messages: []providertypes.Message{{Role: "user", Content: "hi"}}, Tools: []providertypes.ToolSpec{}}) + payload, err := p.buildRequest(providertypes.GenerateRequest{Messages: []providertypes.Message{{Role: "user", Content: "hi"}}, Tools: []providertypes.ToolSpec{}}) if err != nil { t.Fatalf("buildRequest() error = %v", err) } @@ -391,7 +391,7 @@ func TestBuildRequest_MultipleTools(t *testing.T) { t.Fatalf("New() error = %v", err) } - payload, err := p.buildRequest(providertypes.ChatRequest{ + payload, err := p.buildRequest(providertypes.GenerateRequest{ Messages: []providertypes.Message{{Role: "user", Content: "use tools"}}, Tools: []providertypes.ToolSpec{ {Name: "tool_a", Description: "Tool A", Schema: map[string]any{"type": "object"}}, @@ -417,7 +417,7 @@ func TestBuildRequest_WhitespaceSystemPromptSkipped(t *testing.T) { t.Fatalf("New() error = %v", err) } - payload, err := p.buildRequest(providertypes.ChatRequest{SystemPrompt: " ", Messages: []providertypes.Message{{Role: "user", Content: "hi"}}}) + payload, err := p.buildRequest(providertypes.GenerateRequest{SystemPrompt: " ", Messages: []providertypes.Message{{Role: "user", Content: "hi"}}}) if err != nil { t.Fatalf("buildRequest() error = %v", err) } @@ -821,12 +821,12 @@ data: [DONE] p.client = server.Client() events := make(chan providertypes.StreamEvent, 4) - err = p.Chat(context.Background(), providertypes.ChatRequest{ + err = p.Generate(context.Background(), providertypes.GenerateRequest{ Model: config.OpenAIDefaultModel, Messages: []providertypes.Message{{Role: "user", Content: "hi"}}, }, events) if err != nil { - t.Fatalf("Chat() error = %v", err) + t.Fatalf("Generate() error = %v", err) } } @@ -908,7 +908,7 @@ func TestProviderChatConsumesSSEAndMergesToolCalls(t *testing.T) { p.client = server.Client() events := make(chan providertypes.StreamEvent, 8) - err = p.Chat(context.Background(), providertypes.ChatRequest{ + err = p.Generate(context.Background(), providertypes.GenerateRequest{ Model: "gpt-5.4", Messages: []providertypes.Message{ {Role: "user", Content: "please edit the file"}, @@ -918,7 +918,7 @@ func TestProviderChatConsumesSSEAndMergesToolCalls(t *testing.T) { Tools: []providertypes.ToolSpec{{Name: "filesystem_edit", Description: "Edit one matching block in a file", Schema: map[string]any{"type": "object"}}}, }, events) if err != nil { - t.Fatalf("Chat() error = %v", err) + t.Fatalf("Generate() error = %v", err) } streamEvents := drainStreamEvents(events) @@ -1003,7 +1003,7 @@ func TestProviderChatHTTPErrorResponses(t *testing.T) { } p.client = server.Client() - err = p.Chat(context.Background(), providertypes.ChatRequest{Model: config.OpenAIDefaultModel}, make(chan providertypes.StreamEvent, 1)) + err = p.Generate(context.Background(), providertypes.GenerateRequest{Model: config.OpenAIDefaultModel}, make(chan providertypes.StreamEvent, 1)) if err == nil || !strings.Contains(err.Error(), tt.expectErr) { t.Fatalf("expected error containing %q, got %v", tt.expectErr, err) } @@ -1019,7 +1019,7 @@ func TestBuildRequestIncludesSystemPromptToolsAndToolMessages(t *testing.T) { t.Fatalf("New() error = %v", err) } - payload, err := p.buildRequest(providertypes.ChatRequest{ + payload, err := p.buildRequest(providertypes.GenerateRequest{ SystemPrompt: "system prompt", Messages: []providertypes.Message{ {Role: "user", Content: "hello"}, @@ -1368,12 +1368,12 @@ func TestProviderChatEmitsToolCallStartEvent(t *testing.T) { p.client = server.Client() events := make(chan providertypes.StreamEvent, 8) - err = p.Chat(context.Background(), providertypes.ChatRequest{ + err = p.Generate(context.Background(), providertypes.GenerateRequest{ Model: config.OpenAIDefaultModel, Messages: []providertypes.Message{{Role: "user", Content: "edit"}}, Tools: []providertypes.ToolSpec{{Name: "filesystem_edit", Description: "edit", Schema: map[string]any{"type": "object"}}}, }, events) if err != nil { - t.Fatalf("Chat() error = %v", err) + t.Fatalf("Generate() error = %v", err) } var foundToolCallStart bool @@ -1413,9 +1413,9 @@ func TestProviderChatEmitsFullEventStream(t *testing.T) { p.client = server.Client() events := make(chan providertypes.StreamEvent, 16) - err = p.Chat(context.Background(), providertypes.ChatRequest{Model: config.OpenAIDefaultModel, Messages: []providertypes.Message{{Role: "user", Content: "test"}}}, events) + err = p.Generate(context.Background(), providertypes.GenerateRequest{Model: config.OpenAIDefaultModel, Messages: []providertypes.Message{{Role: "user", Content: "test"}}}, events) if err != nil { - t.Fatalf("Chat() error = %v", err) + t.Fatalf("Generate() error = %v", err) } var foundTextDelta, foundToolCallStart, foundToolCallDelta, foundMessageDone bool diff --git a/internal/provider/openai/provider.go b/internal/provider/openai/provider.go index f6db0d37..acf48233 100644 --- a/internal/provider/openai/provider.go +++ b/internal/provider/openai/provider.go @@ -82,7 +82,7 @@ func (p *Provider) DiscoverModels(ctx context.Context) ([]config.ModelDescriptor // Chat 发起 SSE 流式对话请求。 // 流中途断连或协议错误时直接返回错误，由上层调用方决定重试策略。 -func (p *Provider) Chat(ctx context.Context, req providertypes.ChatRequest, events chan<- providertypes.StreamEvent) error { +func (p *Provider) Generate(ctx context.Context, req providertypes.GenerateRequest, events chan<- providertypes.StreamEvent) error { payload, err := p.buildRequest(req) if err != nil { return err diff --git a/internal/provider/openai/request.go b/internal/provider/openai/request.go index 02f5398b..9a9b8d2b 100644 --- a/internal/provider/openai/request.go +++ b/internal/provider/openai/request.go @@ -12,9 +12,9 @@ import ( providertypes "neo-code/internal/provider/types" ) -// buildRequest 将 provider.ChatRequest 转换为 OpenAI API 请求结构。 +// buildRequest 将 provider.GenerateRequest 转换为 OpenAI API 请求结构。 // 模型优先取 req.Model，其次使用配置中的默认模型。 -func (p *Provider) buildRequest(req providertypes.ChatRequest) (chatCompletionRequest, error) { +func (p *Provider) buildRequest(req providertypes.GenerateRequest) (chatCompletionRequest, error) { model := strings.TrimSpace(req.Model) if model == "" { model = strings.TrimSpace(p.cfg.Model) diff --git a/internal/provider/openai/response.go b/internal/provider/openai/response.go index 5026a15d..02b7b9b5 100644 --- a/internal/provider/openai/response.go +++ b/internal/provider/openai/response.go @@ -6,7 +6,6 @@ import ( "errors" "fmt" "io" - "log" "strings" "neo-code/internal/provider" @@ -68,7 +67,6 @@ 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) } @@ -116,16 +114,6 @@ 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/provider/provider.go b/internal/provider/provider.go index f8aff036..0d84bc00 100644 --- a/internal/provider/provider.go +++ b/internal/provider/provider.go @@ -6,7 +6,7 @@ import ( "neo-code/internal/provider/types" ) -// Provider 定义模型对话能力，通过 channel 推送流式事件给上层消费。 +// Provider 定义模型生成能力，通过 channel 推送流式事件给上层消费。 type Provider interface { - Chat(ctx context.Context, req types.ChatRequest, events chan<- types.StreamEvent) error + Generate(ctx context.Context, req types.GenerateRequest, events chan<- types.StreamEvent) error } diff --git a/internal/provider/registry.go b/internal/provider/registry.go index 8a2d4e36..8a33edb4 100644 --- a/internal/provider/registry.go +++ b/internal/provider/registry.go @@ -13,9 +13,10 @@ type Builder func(ctx context.Context, cfg config.ResolvedProviderConfig) (Provi type DiscoveryFunc func(ctx context.Context, cfg config.ResolvedProviderConfig) ([]config.ModelDescriptor, error) type DriverDefinition struct { - Name string - Build Builder - Discover DiscoveryFunc + Name string + Build Builder + Discover DiscoveryFunc + Capabilities DriverCapabilities } type Registry struct { @@ -72,6 +73,15 @@ func (r *Registry) Supports(driverType string) bool { return err == nil } +// DriverCapabilities 返回指定 driver 的能力声明，不存在时返回零值。 +func (r *Registry) DriverCapabilities(driverType string) DriverCapabilities { + driver, err := r.driver(driverType) + if err != nil { + return DriverCapabilities{} + } + return driver.Capabilities +} + func (r *Registry) driver(driverType string) (DriverDefinition, error) { if r == nil { return DriverDefinition{}, ErrDriverNotFound diff --git a/internal/provider/registry_test.go b/internal/provider/registry_test.go index 5ddc5b31..f8f19bf6 100644 --- a/internal/provider/registry_test.go +++ b/internal/provider/registry_test.go @@ -13,7 +13,7 @@ import ( type stubProvider struct{} -func (stubProvider) Chat(ctx context.Context, req providertypes.ChatRequest, events chan<- providertypes.StreamEvent) error { +func (stubProvider) Generate(ctx context.Context, req providertypes.GenerateRequest, events chan<- providertypes.StreamEvent) error { return nil } @@ -187,6 +187,27 @@ func TestRegistrySupports(t *testing.T) { } } +func TestRegistryDriverCapabilities(t *testing.T) { + t.Parallel() + + registry := newTestRegistry(t) + got := registry.DriverCapabilities("OPENAI") + if !got.Streaming { + t.Fatalf("expected openai driver to support streaming, got %+v", got) + } + if !got.ToolTransport { + t.Fatalf("expected openai driver to support tool transport, got %+v", got) + } + if !got.ModelDiscovery { + t.Fatalf("expected openai driver to support model discovery, got %+v", got) + } + + missing := registry.DriverCapabilities("missing") + if missing != (provider.DriverCapabilities{}) { + t.Fatalf("expected zero-value capabilities for missing driver, got %+v", missing) + } +} + func TestRegistryRegisterErrors(t *testing.T) { t.Parallel() diff --git a/internal/provider/types/request.go b/internal/provider/types/request.go index 1171f6b0..79ae4ba5 100644 --- a/internal/provider/types/request.go +++ b/internal/provider/types/request.go @@ -1,7 +1,7 @@ package types -// ChatRequest 是 provider.Chat() 的请求参数。 -type ChatRequest struct { +// GenerateRequest 是 provider.Generate() 的请求参数。 +type GenerateRequest struct { Model string `json:"model"` SystemPrompt string `json:"system_prompt"` Messages []Message `json:"messages"` diff --git a/internal/provider/types/types_test.go b/internal/provider/types/types_test.go index 958ebb16..e9a07c15 100644 --- a/internal/provider/types/types_test.go +++ b/internal/provider/types/types_test.go @@ -195,10 +195,10 @@ func TestToolSpecStructFields(t *testing.T) { } } -func TestChatRequestStructFields(t *testing.T) { +func TestGenerateRequestStructFields(t *testing.T) { t.Parallel() - req := ChatRequest{ + req := GenerateRequest{ Model: "gpt-4", SystemPrompt: "you are helpful", Messages: []Message{{Role: RoleUser}}, diff --git a/internal/runtime/compact_generator.go b/internal/runtime/compact_generator.go index ed6ce60e..beb346bb 100644 --- a/internal/runtime/compact_generator.go +++ b/internal/runtime/compact_generator.go @@ -41,6 +41,9 @@ func (g *compactSummaryGenerator) Generate(ctx context.Context, input contextcom strings.TrimSpace(g.providerConfig.APIKey) == "" { return "", errors.New("runtime: compact summary generator provider config is incomplete") } + if err := ensureProviderDriverCapabilities(g.providerFactory, g.providerConfig, true, false); err != nil { + return "", err + } prompt := agentcontext.BuildCompactPrompt(agentcontext.CompactPromptInput{ Mode: string(input.Mode), @@ -84,7 +87,7 @@ func (g *compactSummaryGenerator) Generate(ctx context.Context, input contextcom } }() - err = modelProvider.Chat(ctx, providertypes.ChatRequest{ + err = modelProvider.Generate(ctx, providertypes.GenerateRequest{ Model: g.model, SystemPrompt: prompt.SystemPrompt, Messages: []providertypes.Message{{ diff --git a/internal/runtime/compact_generator_test.go b/internal/runtime/compact_generator_test.go index 00a96ec9..bb63190e 100644 --- a/internal/runtime/compact_generator_test.go +++ b/internal/runtime/compact_generator_test.go @@ -8,6 +8,7 @@ import ( "neo-code/internal/config" contextcompact "neo-code/internal/context/compact" + "neo-code/internal/provider" providertypes "neo-code/internal/provider/types" ) @@ -198,3 +199,41 @@ func TestCompactSummaryGeneratorMalformedStreamEventDoesNotDeadlock(t *testing.T t.Fatal("expected compact generation to fail instead of deadlocking on malformed stream event") } } + +func TestCompactSummaryGeneratorRejectsDriverWithoutStreaming(t *testing.T) { + t.Parallel() + + manager := newRuntimeConfigManager(t) + resolvedProvider, err := resolvedProviderForTests(manager.Get(), config.OpenAIName) + if err != nil { + t.Fatalf("resolve provider: %v", err) + } + + scripted := &scriptedProvider{ + streams: [][]providertypes.StreamEvent{ + {providertypes.NewTextDeltaStreamEvent("should not run")}, + }, + } + factory := &scriptedProviderFactory{ + provider: scripted, + capabilities: provider.DriverCapabilities{ + Streaming: false, + ToolTransport: true, + }, + } + generator := newCompactSummaryGenerator(factory, resolvedProvider, "session-model") + + _, err = generator.Generate(context.Background(), contextcompact.SummaryInput{ + Mode: contextcompact.ModeManual, + Config: manager.Get().Context.Compact, + }) + if err == nil || !strings.Contains(err.Error(), "does not support streaming") { + t.Fatalf("expected streaming capability rejection, got %v", err) + } + if factory.calls != 0 { + t.Fatalf("expected provider build to be skipped, got %d", factory.calls) + } + if scripted.callCount != 0 { + t.Fatalf("expected provider Generate() to be skipped, got %d", scripted.callCount) + } +} diff --git a/internal/runtime/events.go b/internal/runtime/events.go index 80117aec..97eadc6a 100644 --- a/internal/runtime/events.go +++ b/internal/runtime/events.go @@ -85,8 +85,8 @@ const ( // TokenUsagePayload carries token usage statistics for a single provider turn. type TokenUsagePayload struct { - InputTokens int `json:"input_tokens"` - OutputTokens int `json:"output_tokens"` + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` SessionInputTokens int `json:"session_input_tokens"` SessionOutputTokens int `json:"session_output_tokens"` -} \ No newline at end of file +} diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index 2e13466d..24fedd75 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -23,7 +23,7 @@ import ( ) const ( - // defaultProviderRetryMax 是 runtime 层对单次 provider.Chat() 的最大重试次数（不含首次调用）。 + // defaultProviderRetryMax 是 runtime 层对单次 provider.Generate() 的最大重试次数（不含首次调用）。 // 与 RetryTransport 的 HTTP 层重试互补：Transport 耗尽后 runtime 仍可重试整个 Chat 调用。 defaultProviderRetryMax = 2 // providerRetryBaseWait 是 runtime 层重试的初始等待时间。 @@ -130,6 +130,7 @@ type UserInput struct { type ProviderFactory interface { Build(ctx context.Context, cfg config.ResolvedProviderConfig) (provider.Provider, error) + DriverCapabilities(driverType string) provider.DriverCapabilities } type Service struct { @@ -267,7 +268,7 @@ func (s *Service) Run(ctx context.Context, input UserInput) error { return s.handleRunError(ctx, input.RunID, session.ID, err) } - acc, err := s.callProviderWithRetry(ctx, input.RunID, session.ID, providertypes.ChatRequest{ + acc, err := s.callProviderWithRetry(ctx, input.RunID, session.ID, providertypes.GenerateRequest{ Model: cfg.CurrentModel, SystemPrompt: builtContext.SystemPrompt, Messages: builtContext.Messages, @@ -666,7 +667,29 @@ func isRetryableProviderError(err error) bool { return pErr.Retryable } -// callProviderWithRetry 在可重试的 ProviderError 上自动重试 provider.Chat() 调用。 +// ensureProviderDriverCapabilities 校验当前 driver 是否满足指定运行场景的基础能力要求。 +func ensureProviderDriverCapabilities( + factory ProviderFactory, + cfg config.ResolvedProviderConfig, + requireStreaming bool, + requireToolTransport bool, +) error { + if factory == nil { + return errors.New("runtime: provider factory is nil") + } + + driverType := strings.TrimSpace(cfg.Driver) + caps := factory.DriverCapabilities(driverType) + if requireStreaming && !caps.Streaming { + return fmt.Errorf("runtime: provider driver %q does not support streaming", driverType) + } + if requireToolTransport && !caps.ToolTransport { + return fmt.Errorf("runtime: provider driver %q does not support tool transport", driverType) + } + return nil +} + +// callProviderWithRetry 在可重试的 ProviderError 上自动重试 provider.Generate() 调用。 // 每次重试都会重新创建 provider 实例、流式事件转发管道和累积器。 // 非可重试错误、context 取消、重试耗尽时直接返回错误。 // 返回值 acc 保存了本轮流式事件的完整累积状态，供调用方构建 assistant Message 使用。 @@ -674,7 +697,7 @@ func (s *Service) callProviderWithRetry( ctx context.Context, runID string, sessionID string, - req providertypes.ChatRequest, + req providertypes.GenerateRequest, ) (*streamAccumulator, error) { acc := newStreamAccumulator() var lastErr error @@ -699,6 +722,9 @@ func (s *Service) callProviderWithRetry( if err != nil { return nil, err } + if err := ensureProviderDriverCapabilities(s.providerFactory, resolvedProvider, true, true); err != nil { + return nil, err + } modelProvider, err := s.providerFactory.Build(ctx, resolvedProvider) if err != nil { @@ -709,7 +735,7 @@ func (s *Service) callProviderWithRetry( streamDone := make(chan error, 1) go s.forwardProviderEvents(ctx, runID, sessionID, streamEvents, streamDone, acc) - err = modelProvider.Chat(ctx, req, streamEvents) + err = modelProvider.Generate(ctx, req, streamEvents) close(streamEvents) forwardErr := <-streamDone if forwardErr != nil { diff --git a/internal/runtime/runtime_test.go b/internal/runtime/runtime_test.go index ac40bf8a..1bad5069 100644 --- a/internal/runtime/runtime_test.go +++ b/internal/runtime/runtime_test.go @@ -94,9 +94,9 @@ type scriptedProvider struct { name string streams [][]providertypes.StreamEvent responses []scriptedResponse - requests []providertypes.ChatRequest + requests []providertypes.GenerateRequest callCount int - chatFn func(ctx context.Context, req providertypes.ChatRequest, events chan<- providertypes.StreamEvent) error + chatFn func(ctx context.Context, req providertypes.GenerateRequest, events chan<- providertypes.StreamEvent) error } type scriptedResponse struct { @@ -104,8 +104,8 @@ type scriptedResponse struct { FinishReason string } -func (p *scriptedProvider) Chat(ctx context.Context, req providertypes.ChatRequest, events chan<- providertypes.StreamEvent) error { - p.requests = append(p.requests, cloneChatRequest(req)) +func (p *scriptedProvider) Generate(ctx context.Context, req providertypes.GenerateRequest, events chan<- providertypes.StreamEvent) error { + p.requests = append(p.requests, cloneGenerateRequest(req)) callIndex := p.callCount p.callCount++ @@ -172,10 +172,11 @@ func streamContainsMessageDone(events []providertypes.StreamEvent) bool { } type scriptedProviderFactory struct { - provider provider.Provider - calls int - configs []config.ResolvedProviderConfig - err error + provider provider.Provider + calls int + configs []config.ResolvedProviderConfig + err error + capabilities provider.DriverCapabilities } func (f *scriptedProviderFactory) Build(ctx context.Context, cfg config.ResolvedProviderConfig) (provider.Provider, error) { @@ -187,6 +188,16 @@ func (f *scriptedProviderFactory) Build(ctx context.Context, cfg config.Resolved return f.provider, nil } +func (f *scriptedProviderFactory) DriverCapabilities(driverType string) provider.DriverCapabilities { + if f.capabilities == (provider.DriverCapabilities{}) { + return provider.DriverCapabilities{ + Streaming: true, + ToolTransport: true, + } + } + return f.capabilities +} + type stubTool struct { name string content string @@ -580,7 +591,7 @@ func TestServiceRunRejectsProviderCompletionWithoutMessageDone(t *testing.T) { registry.Register(&stubTool{name: "filesystem_read_file", content: "default"}) scripted := &scriptedProvider{ - chatFn: func(ctx context.Context, req providertypes.ChatRequest, events chan<- providertypes.StreamEvent) error { + chatFn: func(ctx context.Context, req providertypes.GenerateRequest, events chan<- providertypes.StreamEvent) error { select { case events <- providertypes.NewTextDeltaStreamEvent("partial"): case <-ctx.Done(): @@ -1456,7 +1467,7 @@ func TestServiceRunErrorPaths(t *testing.T) { callIdx := 0 return &scriptedProvider{ name: "retry-then-success", - chatFn: func(ctx context.Context, req providertypes.ChatRequest, events chan<- providertypes.StreamEvent) error { + chatFn: func(ctx context.Context, req providertypes.GenerateRequest, events chan<- providertypes.StreamEvent) error { callIdx++ if callIdx == 1 { return &provider.ProviderError{ @@ -1492,7 +1503,7 @@ func TestServiceRunErrorPaths(t *testing.T) { input: UserInput{RunID: "run-no-retry", Content: "hello"}, provider: &scriptedProvider{ name: "auth-error-no-retry", - chatFn: func(ctx context.Context, req providertypes.ChatRequest, events chan<- providertypes.StreamEvent) error { + chatFn: func(ctx context.Context, req providertypes.GenerateRequest, events chan<- providertypes.StreamEvent) error { return &provider.ProviderError{ StatusCode: 401, Code: provider.ErrorCodeAuthFailed, @@ -1515,7 +1526,7 @@ func TestServiceRunErrorPaths(t *testing.T) { input: UserInput{RunID: "run-retry-exhausted", Content: "hello"}, provider: &scriptedProvider{ name: "always-500", - chatFn: func(ctx context.Context, req providertypes.ChatRequest, events chan<- providertypes.StreamEvent) error { + chatFn: func(ctx context.Context, req providertypes.GenerateRequest, events chan<- providertypes.StreamEvent) error { return &provider.ProviderError{ StatusCode: 500, Code: provider.ErrorCodeServer, @@ -1599,7 +1610,7 @@ func TestServiceCancelActiveRun(t *testing.T) { started := make(chan struct{}) scripted := &scriptedProvider{ name: "cancel-active-run-provider", - chatFn: func(ctx context.Context, req providertypes.ChatRequest, events chan<- providertypes.StreamEvent) error { + chatFn: func(ctx context.Context, req providertypes.GenerateRequest, events chan<- providertypes.StreamEvent) error { close(started) <-ctx.Done() return ctx.Err() @@ -1642,7 +1653,7 @@ func TestServiceRunCanceledByProvider(t *testing.T) { started := make(chan struct{}) scripted := &scriptedProvider{ name: "blocking-provider", - chatFn: func(ctx context.Context, req providertypes.ChatRequest, events chan<- providertypes.StreamEvent) error { + chatFn: func(ctx context.Context, req providertypes.GenerateRequest, events chan<- providertypes.StreamEvent) error { close(started) <-ctx.Done() return ctx.Err() @@ -1687,7 +1698,7 @@ func TestServiceRunPreservesProviderErrorAfterCancel(t *testing.T) { providerErr := errors.New("provider failed after cancel") scripted := &scriptedProvider{ name: "provider-error-after-cancel", - chatFn: func(ctx context.Context, req providertypes.ChatRequest, events chan<- providertypes.StreamEvent) error { + chatFn: func(ctx context.Context, req providertypes.GenerateRequest, events chan<- providertypes.StreamEvent) error { close(started) <-ctx.Done() return providerErr @@ -2071,7 +2082,7 @@ func TestServiceCompactUsesSessionProviderAndModelWhenPresent(t *testing.T) { }, } factory := &scriptedProviderFactory{provider: scripted} - service := NewWithFactory(manager, registry, store, factory, nil) + service := NewWithFactory(manager, registry, store, factory, &stubContextBuilder{}) result, err := service.Compact(context.Background(), CompactInput{ SessionID: session.ID, @@ -2143,7 +2154,7 @@ func TestServiceCompactFallsBackToCurrentProviderWhenSessionMetadataMissing(t *t }, } factory := &scriptedProviderFactory{provider: scripted} - service := NewWithFactory(manager, registry, store, factory, nil) + service := NewWithFactory(manager, registry, store, factory, &stubContextBuilder{}) result, err := service.Compact(context.Background(), CompactInput{ SessionID: session.ID, @@ -2259,7 +2270,7 @@ func TestServiceSerializesRunAndCompact(t *testing.T) { providerStarted := make(chan struct{}) unblockProvider := make(chan struct{}) scripted := &scriptedProvider{ - chatFn: func(ctx context.Context, req providertypes.ChatRequest, events chan<- providertypes.StreamEvent) error { + chatFn: func(ctx context.Context, req providertypes.GenerateRequest, events chan<- providertypes.StreamEvent) error { select { case <-providerStarted: default: @@ -2676,7 +2687,7 @@ func cloneSession(session agentsession.Session) agentsession.Session { return cloned } -func cloneChatRequest(req providertypes.ChatRequest) providertypes.ChatRequest { +func cloneGenerateRequest(req providertypes.GenerateRequest) providertypes.GenerateRequest { cloned := req cloned.Messages = append([]providertypes.Message(nil), req.Messages...) cloned.Tools = append([]providertypes.ToolSpec(nil), req.Tools...) @@ -3073,7 +3084,7 @@ func TestCallProviderWithRetryReturnsCombinedForwardError(t *testing.T) { store := newMemoryStore() scripted := &scriptedProvider{ - chatFn: func(ctx context.Context, req providertypes.ChatRequest, events chan<- providertypes.StreamEvent) error { + chatFn: func(ctx context.Context, req providertypes.GenerateRequest, events chan<- providertypes.StreamEvent) error { events <- providertypes.StreamEvent{Type: providertypes.StreamEventTextDelta} return errors.New("provider chat failed") }, @@ -3084,7 +3095,7 @@ func TestCallProviderWithRetryReturnsCombinedForwardError(t *testing.T) { context.Background(), "run-forward-error", "session-forward-error", - providertypes.ChatRequest{ + providertypes.GenerateRequest{ Model: "test-model", SystemPrompt: "prompt", Messages: []providertypes.Message{{Role: providertypes.RoleUser, Content: "hello"}}, @@ -3095,6 +3106,43 @@ func TestCallProviderWithRetryReturnsCombinedForwardError(t *testing.T) { } } +func TestServiceRunRejectsDriverWithoutToolTransport(t *testing.T) { + t.Parallel() + + manager := newRuntimeConfigManager(t) + store := newMemoryStore() + registry := tools.NewRegistry() + registry.Register(&stubTool{name: "filesystem_read_file", content: "default"}) + + scripted := &scriptedProvider{ + streams: [][]providertypes.StreamEvent{ + {providertypes.NewTextDeltaStreamEvent("should not run")}, + }, + } + factory := &scriptedProviderFactory{ + provider: scripted, + capabilities: provider.DriverCapabilities{ + Streaming: true, + ToolTransport: false, + }, + } + + service := NewWithFactory(manager, registry, store, factory, &stubContextBuilder{}) + err := service.Run(context.Background(), UserInput{ + RunID: "run-driver-no-tools", + Content: "hello", + }) + if err == nil || !containsError(err, "does not support tool transport") { + t.Fatalf("expected tool transport capability error, got %v", err) + } + if factory.calls != 0 { + t.Fatalf("expected provider build to be skipped, got %d", factory.calls) + } + if scripted.callCount != 0 { + t.Fatalf("expected provider Generate() to be skipped, got %d", scripted.callCount) + } +} + func TestServiceRunPersistsAndRestoresTokenUsage(t *testing.T) { t.Parallel() @@ -3105,7 +3153,7 @@ func TestServiceRunPersistsAndRestoresTokenUsage(t *testing.T) { builder := &stubContextBuilder{} scripted := &scriptedProvider{} - scripted.chatFn = func(ctx context.Context, req providertypes.ChatRequest, events chan<- providertypes.StreamEvent) error { + scripted.chatFn = func(ctx context.Context, req providertypes.GenerateRequest, events chan<- providertypes.StreamEvent) error { usage := &providertypes.Usage{} if scripted.callCount == 1 { usage.InputTokens = 100 From 707049c6e56b5ce72351dafb899bf13c44395a4e Mon Sep 17 00:00:00 2001 From: phantom5099 <1011668688@qq.com> Date: Wed, 8 Apr 2026 21:48:27 +0800 Subject: [PATCH 4/4] =?UTF-8?q?fix(provider):=E4=BF=AE=E5=A4=8D=E5=BD=93?= =?UTF-8?q?=E5=89=8D=E5=AD=98=E5=9C=A8EOF=E8=A2=AB=E8=AF=AF=E5=BD=93?= =?UTF-8?q?=E4=BD=9C=E6=AD=A3=E5=B8=B8=E5=AE=8C=E6=88=90=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/provider/catalog/service_test.go | 186 +++++++++++++++++++++ internal/provider/catalog/store_test.go | 119 +++++++++++++ internal/provider/openai/openai_test.go | 79 +++++++++ internal/provider/openai/response.go | 8 +- internal/provider/registry.go | 8 +- internal/provider/registry_test.go | 11 +- internal/provider/types/types_test.go | 85 ++++++++++ internal/runtime/compact_generator_test.go | 28 ++++ internal/runtime/runtime.go | 7 +- internal/runtime/runtime_test.go | 46 ++++- 10 files changed, 558 insertions(+), 19 deletions(-) diff --git a/internal/provider/catalog/service_test.go b/internal/provider/catalog/service_test.go index 476a1f59..cbf3a854 100644 --- a/internal/provider/catalog/service_test.go +++ b/internal/provider/catalog/service_test.go @@ -2,7 +2,9 @@ package catalog import ( "context" + "errors" "sync" + "sync/atomic" "testing" "time" @@ -203,6 +205,190 @@ func TestDescriptorsFromIDsHelper(t *testing.T) { } } +func TestServiceValidateErrors(t *testing.T) { + t.Parallel() + + t.Run("nil service", func(t *testing.T) { + t.Parallel() + + var service *Service + _, err := service.ListProviderModels(context.Background(), config.OpenAIProvider()) + if err == nil || err.Error() != "provider catalog: service is nil" { + t.Fatalf("expected nil service error, got %v", err) + } + }) + + t.Run("nil registry", func(t *testing.T) { + t.Parallel() + + service := NewService("", nil, newMemoryStore()) + _, err := service.ListProviderModels(context.Background(), config.OpenAIProvider()) + if err == nil || err.Error() != "provider catalog: registry is nil" { + t.Fatalf("expected nil registry error, got %v", err) + } + }) +} + +func TestListProviderModelsHonorsContextError(t *testing.T) { + t.Parallel() + + service := NewService("", provider.NewRegistry(), newMemoryStore()) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err := service.ListProviderModels(ctx, config.OpenAIProvider()) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } +} + +func TestListProviderModelsCachedUsesFreshCatalogWithoutDiscovery(t *testing.T) { + t.Setenv(testAPIKeyEnv, "test-key") + + identity, err := config.OpenAIProvider().Identity() + if err != nil { + t.Fatalf("Identity() error = %v", err) + } + + store := newMemoryStore() + now := time.Date(2026, 4, 8, 12, 0, 0, 0, time.UTC) + if err := store.Save(context.Background(), ModelCatalog{ + SchemaVersion: schemaVersion, + Identity: identity, + FetchedAt: now.Add(-time.Hour), + ExpiresAt: now.Add(time.Hour), + Models: []config.ModelDescriptor{ + {ID: "cached-model", Name: "Cached Model"}, + }, + }); err != nil { + t.Fatalf("seed fresh catalog: %v", err) + } + + var discoverCalls int32 + registry := newRegistry(t, config.OpenAIName, func(ctx context.Context, cfg config.ResolvedProviderConfig) ([]config.ModelDescriptor, error) { + atomic.AddInt32(&discoverCalls, 1) + return []config.ModelDescriptor{{ID: "fresh-model", Name: "Fresh Model"}}, nil + }) + + service := NewService("", registry, store) + service.now = func() time.Time { return now } + + models, err := service.ListProviderModelsCached(context.Background(), config.OpenAIProvider()) + if err != nil { + t.Fatalf("ListProviderModelsCached() error = %v", err) + } + if !containsModelDescriptorID(models, "cached-model") { + t.Fatalf("expected cached model to be returned, got %+v", models) + } + if atomic.LoadInt32(&discoverCalls) != 0 { + t.Fatalf("expected no discovery for fresh cache, got %d", discoverCalls) + } +} + +func TestDiscoverAndPersistFailurePaths(t *testing.T) { + t.Run("unsupported driver", func(t *testing.T) { + service := NewService("", provider.NewRegistry(), newMemoryStore()) + if discovered, ok := service.discoverAndPersist(context.Background(), config.OpenAIProvider()); ok || discovered != nil { + t.Fatalf("expected unsupported driver to skip discovery, got ok=%v models=%+v", ok, discovered) + } + }) + + t.Run("resolve provider config failure", func(t *testing.T) { + service := NewService("", newRegistry(t, config.OpenAIName, func(ctx context.Context, cfg config.ResolvedProviderConfig) ([]config.ModelDescriptor, error) { + return nil, nil + }), newMemoryStore()) + + providerCfg := config.ProviderConfig{ + Name: "broken-openai", + Driver: config.OpenAIName, + BaseURL: config.OpenAIDefaultBaseURL, + Model: config.OpenAIDefaultModel, + APIKeyEnv: "", + } + + if discovered, ok := service.discoverAndPersist(context.Background(), providerCfg); ok || discovered != nil { + t.Fatalf("expected resolve failure to skip discovery, got ok=%v models=%+v", ok, discovered) + } + }) + + t.Run("discovery error", func(t *testing.T) { + t.Setenv(testAPIKeyEnv, "test-key") + service := NewService("", newRegistry(t, config.OpenAIName, func(ctx context.Context, cfg config.ResolvedProviderConfig) ([]config.ModelDescriptor, error) { + return nil, errors.New("discover failed") + }), newMemoryStore()) + + if discovered, ok := service.discoverAndPersist(context.Background(), config.OpenAIProvider()); ok || discovered != nil { + t.Fatalf("expected discovery error to skip persistence, got ok=%v models=%+v", ok, discovered) + } + }) + + t.Run("store nil still returns discovered models", func(t *testing.T) { + t.Setenv(testAPIKeyEnv, "test-key") + service := NewService("", newRegistry(t, config.OpenAIName, func(ctx context.Context, cfg config.ResolvedProviderConfig) ([]config.ModelDescriptor, error) { + return []config.ModelDescriptor{{ID: "gpt-4.1", Name: "GPT-4.1"}}, nil + }), nil) + + discovered, ok := service.discoverAndPersist(context.Background(), config.OpenAIProvider()) + if !ok { + t.Fatal("expected discovery without store to succeed") + } + if !containsModelDescriptorID(discovered, "gpt-4.1") { + t.Fatalf("expected discovered model to be returned, got %+v", discovered) + } + }) +} + +func TestQueueRefreshDeduplicatesInFlightRequests(t *testing.T) { + t.Setenv(testAPIKeyEnv, "test-key") + + identity, err := config.OpenAIProvider().Identity() + if err != nil { + t.Fatalf("Identity() error = %v", err) + } + + started := make(chan struct{}, 1) + release := make(chan struct{}) + var discoverCalls int32 + registry := newRegistry(t, config.OpenAIName, func(ctx context.Context, cfg config.ResolvedProviderConfig) ([]config.ModelDescriptor, error) { + atomic.AddInt32(&discoverCalls, 1) + select { + case started <- struct{}{}: + default: + } + select { + case <-release: + case <-ctx.Done(): + } + return []config.ModelDescriptor{{ID: "gpt-4o", Name: "GPT-4o"}}, nil + }) + + service := NewService("", registry, newMemoryStore()) + service.backgroundTimeout = time.Second + + service.queueRefresh(config.OpenAIProvider(), identity) + <-started + service.queueRefresh(config.OpenAIProvider(), identity) + + time.Sleep(50 * time.Millisecond) + if calls := atomic.LoadInt32(&discoverCalls); calls != 1 { + t.Fatalf("expected exactly one in-flight refresh, got %d", calls) + } + + close(release) + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + service.refreshMu.Lock() + _, exists := service.inFlightByID[identity.Key()] + service.refreshMu.Unlock() + if !exists { + return + } + time.Sleep(20 * time.Millisecond) + } + + t.Fatal("expected in-flight refresh marker to be cleared") +} + func newRegistry(t *testing.T, name string, discover provider.DiscoveryFunc) *provider.Registry { t.Helper() diff --git a/internal/provider/catalog/store_test.go b/internal/provider/catalog/store_test.go index 86c7c098..82024d28 100644 --- a/internal/provider/catalog/store_test.go +++ b/internal/provider/catalog/store_test.go @@ -5,6 +5,7 @@ import ( "errors" "os" "path/filepath" + "strings" "testing" "time" @@ -142,3 +143,121 @@ func TestJSONStoreSaveReplacesExistingCatalogWithoutTempLeak(t *testing.T) { t.Fatalf("expected persisted catalog to end with newline, got %q", string(data)) } } + +func TestModelCatalogExpired(t *testing.T) { + t.Parallel() + + now := time.Date(2026, 4, 8, 12, 0, 0, 0, time.UTC) + if (ModelCatalog{}).Expired(now) { + t.Fatal("expected zero-value catalog to be treated as not expired") + } + if !(ModelCatalog{ExpiresAt: now}).Expired(now) { + t.Fatal("expected catalog expiring at now to be expired") + } + if (ModelCatalog{ExpiresAt: now.Add(time.Minute)}).Expired(now) { + t.Fatal("expected future expiry to be treated as fresh") + } +} + +func TestJSONStoreLoadHonorsContextError(t *testing.T) { + t.Parallel() + + store := newJSONStore(t.TempDir()) + identity, err := config.NewProviderIdentity("openai", "https://api.openai.com/v1") + if err != nil { + t.Fatalf("NewProviderIdentity() error = %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err = store.Load(ctx, identity) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } +} + +func TestJSONStoreLoadRejectsInvalidIdentity(t *testing.T) { + t.Parallel() + + store := newJSONStore(t.TempDir()) + _, err := store.Load(context.Background(), config.ProviderIdentity{ + Driver: "openai", + BaseURL: "://bad", + }) + if err == nil || !strings.Contains(err.Error(), "normalize model catalog key") { + t.Fatalf("expected identity normalization error, got %v", err) + } +} + +func TestJSONStoreLoadRejectsInvalidJSON(t *testing.T) { + t.Parallel() + + store := newJSONStore(t.TempDir()) + identity, err := config.NewProviderIdentity("openai", "https://api.openai.com/v1") + if err != nil { + t.Fatalf("NewProviderIdentity() error = %v", err) + } + + if err := os.MkdirAll(filepath.Dir(store.catalogPath(identity)), 0o755); err != nil { + t.Fatalf("MkdirAll() error = %v", err) + } + if err := os.WriteFile(store.catalogPath(identity), []byte("{not-json"), 0o644); err != nil { + t.Fatalf("WriteFile() error = %v", err) + } + + _, err = store.Load(context.Background(), identity) + if err == nil || !strings.Contains(err.Error(), "decode model catalog") { + t.Fatalf("expected decode error, got %v", err) + } +} + +func TestJSONStoreSaveHonorsContextError(t *testing.T) { + t.Parallel() + + store := newJSONStore(t.TempDir()) + identity, err := config.NewProviderIdentity("openai", "https://api.openai.com/v1") + if err != nil { + t.Fatalf("NewProviderIdentity() error = %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err = store.Save(ctx, ModelCatalog{Identity: identity}) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } +} + +func TestJSONStoreSaveRejectsInvalidIdentity(t *testing.T) { + t.Parallel() + + store := newJSONStore(t.TempDir()) + err := store.Save(context.Background(), ModelCatalog{ + Identity: config.ProviderIdentity{ + Driver: "", + BaseURL: "https://api.openai.com/v1", + }, + }) + if err == nil || !strings.Contains(err.Error(), "normalize model catalog key") { + t.Fatalf("expected identity normalization error, got %v", err) + } +} + +func TestNormalizeCatalogDefaultsSchemaAndDedupesModels(t *testing.T) { + t.Parallel() + + modelCatalog := normalizeCatalog(ModelCatalog{ + Models: []config.ModelDescriptor{ + {ID: "gpt-4o", Name: "GPT-4o"}, + {ID: "gpt-4o", Name: "GPT-4o Duplicate"}, + }, + }) + if modelCatalog.SchemaVersion != schemaVersion { + t.Fatalf("expected schema version %d, got %d", schemaVersion, modelCatalog.SchemaVersion) + } + if len(modelCatalog.Models) != 1 { + t.Fatalf("expected duplicate models to be merged, got %+v", modelCatalog.Models) + } +} diff --git a/internal/provider/openai/openai_test.go b/internal/provider/openai/openai_test.go index a55c370f..aa3d4414 100644 --- a/internal/provider/openai/openai_test.go +++ b/internal/provider/openai/openai_test.go @@ -737,6 +737,85 @@ func TestDiscoverModels_NetworkError(t *testing.T) { } } +func TestFetchModelsSetsAuthorizationHeader(t *testing.T) { + t.Parallel() + + var authorization string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + authorization = r.Header.Get("Authorization") + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"data": []map[string]any{}}) + })) + defer server.Close() + + p, err := New(resolvedConfig(server.URL, "header-key")) + if err != nil { + t.Fatalf("New() error = %v", err) + } + p.client = server.Client() + + if _, err := p.fetchModels(context.Background()); err != nil { + t.Fatalf("fetchModels() error = %v", err) + } + if authorization != "Bearer test-key" { + t.Fatalf("expected bearer authorization header, got %q", authorization) + } +} + +func TestFetchModelsDecodeError(t *testing.T) { + t.Parallel() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte("{invalid-json")) + })) + defer server.Close() + + p, err := New(resolvedConfig(server.URL, "decode-key")) + if err != nil { + t.Fatalf("New() error = %v", err) + } + p.client = server.Client() + + _, err = p.fetchModels(context.Background()) + if err == nil || !strings.Contains(err.Error(), "decode models response") { + t.Fatalf("expected decode error, got %v", err) + } +} + +func TestDiscoverModelsSkipsInvalidEntriesAndDedupes(t *testing.T) { + t.Parallel() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "data": []map[string]any{ + {"id": "gpt-4.1", "name": "GPT-4.1"}, + {"foo": "bar"}, + {"id": "gpt-4.1", "name": "GPT-4.1 Duplicate"}, + }, + }) + })) + defer server.Close() + + p, err := New(resolvedConfig(server.URL, "discover-key")) + if err != nil { + t.Fatalf("New() error = %v", err) + } + p.client = server.Client() + + models, err := p.DiscoverModels(context.Background()) + if err != nil { + t.Fatalf("DiscoverModels() error = %v", err) + } + if len(models) != 1 { + t.Fatalf("expected invalid and duplicate models to be filtered, got %+v", models) + } + if models[0].ID != "gpt-4.1" { + t.Fatalf("expected remaining model to be gpt-4.1, got %+v", models[0]) + } +} + // --- mergeToolCallDelta 边界测试 --- func TestMergeToolCallDelta_MultipleIndices(t *testing.T) { diff --git a/internal/provider/openai/response.go b/internal/provider/openai/response.go index 02b7b9b5..3b59d880 100644 --- a/internal/provider/openai/response.go +++ b/internal/provider/openai/response.go @@ -117,7 +117,13 @@ func (p *Provider) consumeStream( if flushErr := flushPendingData(); flushErr != nil { return flushErr } - return finishStream() + if done { + return finishStream() + } + if ctx.Err() != nil { + return ctx.Err() + } + return provider.ErrStreamInterrupted } } } diff --git a/internal/provider/registry.go b/internal/provider/registry.go index 8a33edb4..b49344a1 100644 --- a/internal/provider/registry.go +++ b/internal/provider/registry.go @@ -73,13 +73,13 @@ func (r *Registry) Supports(driverType string) bool { return err == nil } -// DriverCapabilities 返回指定 driver 的能力声明，不存在时返回零值。 -func (r *Registry) DriverCapabilities(driverType string) DriverCapabilities { +// DriverCapabilities 返回指定 driver 的能力声明；driver 不存在时返回对应错误。 +func (r *Registry) DriverCapabilities(driverType string) (DriverCapabilities, error) { driver, err := r.driver(driverType) if err != nil { - return DriverCapabilities{} + return DriverCapabilities{}, err } - return driver.Capabilities + return driver.Capabilities, nil } func (r *Registry) driver(driverType string) (DriverDefinition, error) { diff --git a/internal/provider/registry_test.go b/internal/provider/registry_test.go index f8f19bf6..5d597823 100644 --- a/internal/provider/registry_test.go +++ b/internal/provider/registry_test.go @@ -191,7 +191,10 @@ func TestRegistryDriverCapabilities(t *testing.T) { t.Parallel() registry := newTestRegistry(t) - got := registry.DriverCapabilities("OPENAI") + got, err := registry.DriverCapabilities("OPENAI") + if err != nil { + t.Fatalf("DriverCapabilities() error = %v", err) + } if !got.Streaming { t.Fatalf("expected openai driver to support streaming, got %+v", got) } @@ -202,9 +205,9 @@ func TestRegistryDriverCapabilities(t *testing.T) { t.Fatalf("expected openai driver to support model discovery, got %+v", got) } - missing := registry.DriverCapabilities("missing") - if missing != (provider.DriverCapabilities{}) { - t.Fatalf("expected zero-value capabilities for missing driver, got %+v", missing) + _, err = registry.DriverCapabilities("missing") + if !errors.Is(err, provider.ErrDriverNotFound) { + t.Fatalf("expected ErrDriverNotFound for missing driver, got %v", err) } } diff --git a/internal/provider/types/types_test.go b/internal/provider/types/types_test.go index e9a07c15..a30d41da 100644 --- a/internal/provider/types/types_test.go +++ b/internal/provider/types/types_test.go @@ -265,3 +265,88 @@ func TestStreamEventValueAccessorsRejectMissingPayload(t *testing.T) { t.Fatal("expected TextDeltaValue() to reject missing payload") } } + +func TestStreamEventValueAccessorsRejectWrongType(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + run func() error + }{ + { + name: "text delta accessor", + run: func() error { + _, err := NewToolCallStartStreamEvent(0, "call-1", "edit").TextDeltaValue() + return err + }, + }, + { + name: "tool call start accessor", + run: func() error { + _, err := NewTextDeltaStreamEvent("hello").ToolCallStartValue() + return err + }, + }, + { + name: "tool call delta accessor", + run: func() error { + _, err := NewMessageDoneStreamEvent("stop", nil).ToolCallDeltaValue() + return err + }, + }, + { + name: "message done accessor", + run: func() error { + _, err := NewToolCallDeltaStreamEvent(1, "call-2", "{}").MessageDoneValue() + return err + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if err := tt.run(); err == nil { + t.Fatal("expected wrong-type accessor to return error") + } + }) + } +} + +func TestStreamEventValueAccessorsRejectMissingPayloadForAllTypes(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + run func() error + }{ + { + name: "tool call start payload missing", + run: func() error { + _, err := (StreamEvent{Type: StreamEventToolCallStart}).ToolCallStartValue() + return err + }, + }, + { + name: "tool call delta payload missing", + run: func() error { + _, err := (StreamEvent{Type: StreamEventToolCallDelta}).ToolCallDeltaValue() + return err + }, + }, + { + name: "message done payload missing", + run: func() error { + _, err := (StreamEvent{Type: StreamEventMessageDone}).MessageDoneValue() + return err + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if err := tt.run(); err == nil { + t.Fatal("expected missing payload to return error") + } + }) + } +} diff --git a/internal/runtime/compact_generator_test.go b/internal/runtime/compact_generator_test.go index bb63190e..2e643f04 100644 --- a/internal/runtime/compact_generator_test.go +++ b/internal/runtime/compact_generator_test.go @@ -2,6 +2,7 @@ package runtime import ( "context" + "errors" "strings" "testing" "time" @@ -237,3 +238,30 @@ func TestCompactSummaryGeneratorRejectsDriverWithoutStreaming(t *testing.T) { t.Fatalf("expected provider Generate() to be skipped, got %d", scripted.callCount) } } + +func TestCompactSummaryGeneratorPropagatesDriverNotFoundFromCapabilities(t *testing.T) { + t.Parallel() + + manager := newRuntimeConfigManager(t) + resolvedProvider, err := resolvedProviderForTests(manager.Get(), config.OpenAIName) + if err != nil { + t.Fatalf("resolve provider: %v", err) + } + + factory := &scriptedProviderFactory{ + provider: &scriptedProvider{}, + capabilitiesErr: provider.ErrDriverNotFound, + } + generator := newCompactSummaryGenerator(factory, resolvedProvider, "session-model") + + _, err = generator.Generate(context.Background(), contextcompact.SummaryInput{ + Mode: contextcompact.ModeManual, + Config: manager.Get().Context.Compact, + }) + if !errors.Is(err, provider.ErrDriverNotFound) { + t.Fatalf("expected ErrDriverNotFound, got %v", err) + } + if factory.calls != 0 { + t.Fatalf("expected provider build to be skipped, got %d", factory.calls) + } +} diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index 24fedd75..8efef734 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -130,7 +130,7 @@ type UserInput struct { type ProviderFactory interface { Build(ctx context.Context, cfg config.ResolvedProviderConfig) (provider.Provider, error) - DriverCapabilities(driverType string) provider.DriverCapabilities + DriverCapabilities(driverType string) (provider.DriverCapabilities, error) } type Service struct { @@ -679,7 +679,10 @@ func ensureProviderDriverCapabilities( } driverType := strings.TrimSpace(cfg.Driver) - caps := factory.DriverCapabilities(driverType) + caps, err := factory.DriverCapabilities(driverType) + if err != nil { + return err + } if requireStreaming && !caps.Streaming { return fmt.Errorf("runtime: provider driver %q does not support streaming", driverType) } diff --git a/internal/runtime/runtime_test.go b/internal/runtime/runtime_test.go index 1bad5069..5a8771e0 100644 --- a/internal/runtime/runtime_test.go +++ b/internal/runtime/runtime_test.go @@ -172,11 +172,12 @@ func streamContainsMessageDone(events []providertypes.StreamEvent) bool { } type scriptedProviderFactory struct { - provider provider.Provider - calls int - configs []config.ResolvedProviderConfig - err error - capabilities provider.DriverCapabilities + provider provider.Provider + calls int + configs []config.ResolvedProviderConfig + err error + capabilities provider.DriverCapabilities + capabilitiesErr error } func (f *scriptedProviderFactory) Build(ctx context.Context, cfg config.ResolvedProviderConfig) (provider.Provider, error) { @@ -188,14 +189,17 @@ func (f *scriptedProviderFactory) Build(ctx context.Context, cfg config.Resolved return f.provider, nil } -func (f *scriptedProviderFactory) DriverCapabilities(driverType string) provider.DriverCapabilities { +func (f *scriptedProviderFactory) DriverCapabilities(driverType string) (provider.DriverCapabilities, error) { + if f.capabilitiesErr != nil { + return provider.DriverCapabilities{}, f.capabilitiesErr + } if f.capabilities == (provider.DriverCapabilities{}) { return provider.DriverCapabilities{ Streaming: true, ToolTransport: true, - } + }, nil } - return f.capabilities + return f.capabilities, nil } type stubTool struct { @@ -3143,6 +3147,32 @@ func TestServiceRunRejectsDriverWithoutToolTransport(t *testing.T) { } } +func TestServiceRunPropagatesDriverNotFoundFromCapabilities(t *testing.T) { + t.Parallel() + + manager := newRuntimeConfigManager(t) + store := newMemoryStore() + registry := tools.NewRegistry() + registry.Register(&stubTool{name: "filesystem_read_file", content: "default"}) + + factory := &scriptedProviderFactory{ + provider: &scriptedProvider{}, + capabilitiesErr: fmt.Errorf("%w: missing", provider.ErrDriverNotFound), + } + + service := NewWithFactory(manager, registry, store, factory, &stubContextBuilder{}) + err := service.Run(context.Background(), UserInput{ + RunID: "run-driver-missing", + Content: "hello", + }) + if !errors.Is(err, provider.ErrDriverNotFound) { + t.Fatalf("expected ErrDriverNotFound, got %v", err) + } + if factory.calls != 0 { + t.Fatalf("expected provider build to be skipped, got %d", factory.calls) + } +} + func TestServiceRunPersistsAndRestoresTokenUsage(t *testing.T) { t.Parallel()