From 9f3ae4f7e2d44d87f15333f2160499d287dbcf17 Mon Sep 17 00:00:00 2001 From: phantom5099 <1011668688@qq.com> Date: Sat, 4 Apr 2026 23:31:46 +0800 Subject: [PATCH 1/4] =?UTF-8?q?pref(provider):=E5=A2=9E=E5=8A=A0=E5=BC=BA?= =?UTF-8?q?=E6=9E=84=E9=80=A0=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/provider/openai/openai.go | 33 +++------- internal/provider/openai/openai_test.go | 10 +-- internal/provider/types.go | 81 ++++++++++++++++++++++++- 3 files changed, 94 insertions(+), 30 deletions(-) diff --git a/internal/provider/openai/openai.go b/internal/provider/openai/openai.go index b1b3599c..d67ee294 100644 --- a/internal/provider/openai/openai.go +++ b/internal/provider/openai/openai.go @@ -53,11 +53,11 @@ func Driver() provider.DriverDefinition { return New(cfg, WithTransport(defaultRetryTransport())) }, Discover: func(ctx context.Context, cfg config.ResolvedProviderConfig) ([]config.ModelDescriptor, error) { - provider, err := New(cfg, WithTransport(defaultRetryTransport())) + p, err := New(cfg, WithTransport(defaultRetryTransport())) if err != nil { return nil, err } - return provider.DiscoverModels(ctx) + return p.DiscoverModels(ctx) }, } } @@ -331,34 +331,23 @@ func emitTextDelta(ctx context.Context, events chan<- provider.StreamEvent, text if text == "" { return nil } - return emitStreamEvent(ctx, events, provider.StreamEvent{ - Type: provider.StreamEventTextDelta, - Text: text, - }) + return emitStreamEvent(ctx, events, provider.NewTextDeltaStreamEvent(text)) } func emitToolCallStart(ctx context.Context, events chan<- provider.StreamEvent, index int, id, name string) error { if name == "" { return nil } - return emitStreamEvent(ctx, events, provider.StreamEvent{ - Type: provider.StreamEventToolCallStart, - ToolCallID: id, - ToolName: name, - ToolCallIndex: index, - }) + return emitStreamEvent(ctx, events, provider.NewToolCallStartStreamEvent(index, id, name)) } // emitToolCallDelta 发送工具调用参数增量事件。 -func emitToolCallDelta(ctx context.Context, events chan<- provider.StreamEvent, index int, argumentsDelta string) error { +// id 为工具调用 ID，由上游 mergeToolCallDelta 从累积状态中传入。 +func emitToolCallDelta(ctx context.Context, events chan<- provider.StreamEvent, index int, id, argumentsDelta string) error { if argumentsDelta == "" { return nil } - return emitStreamEvent(ctx, events, provider.StreamEvent{ - Type: provider.StreamEventToolCallDelta, - ToolCallIndex: index, - ToolArgumentsDelta: argumentsDelta, - }) + return emitStreamEvent(ctx, events, provider.NewToolCallDeltaStreamEvent(index, id, argumentsDelta)) } // emitMessageDone 发送消息完成事件。 @@ -366,11 +355,7 @@ func emitMessageDone(ctx context.Context, events chan<- provider.StreamEvent, fi if events == nil { return nil } - return emitStreamEvent(ctx, events, provider.StreamEvent{ - Type: provider.StreamEventMessageDone, - FinishReason: finishReason, - Usage: usage, - }) + return emitStreamEvent(ctx, events, provider.NewMessageDoneStreamEvent(finishReason, usage)) } // extractStreamUsage 从 OpenAI usage 响应提取并覆盖累积的 token 统计。 @@ -413,7 +398,7 @@ func mergeToolCallDelta(ctx context.Context, events chan<- provider.StreamEvent, // 发送参数增量事件（同一 chunk 可能同时携带 name 和 arguments） if args := delta.Function.Arguments; args != "" { call.Arguments += args - if err := emitToolCallDelta(ctx, events, delta.Index, args); err != nil { + if err := emitToolCallDelta(ctx, events, delta.Index, call.ID, args); err != nil { return err } } diff --git a/internal/provider/openai/openai_test.go b/internal/provider/openai/openai_test.go index 413e5b44..cdf62bac 100644 --- a/internal/provider/openai/openai_test.go +++ b/internal/provider/openai/openai_test.go @@ -92,7 +92,7 @@ func TestEmitToolCallDelta(t *testing.T) { t.Run("nil events guard", func(t *testing.T) { t.Parallel() - if err := emitToolCallDelta(context.Background(), nil, 0, "args"); err != nil { + if err := emitToolCallDelta(context.Background(), nil, 0, "", "args"); err != nil { t.Fatalf("expected nil events guard to return nil, got %v", err) } }) @@ -100,7 +100,7 @@ func TestEmitToolCallDelta(t *testing.T) { t.Run("empty arguments guard", func(t *testing.T) { t.Parallel() events := make(chan domain.StreamEvent, 1) - if err := emitToolCallDelta(context.Background(), events, 0, ""); err != nil { + if err := emitToolCallDelta(context.Background(), events, 0, "", ""); err != nil { t.Fatalf("expected empty arguments guard to return nil, got %v", err) } select { @@ -113,11 +113,11 @@ func TestEmitToolCallDelta(t *testing.T) { t.Run("normal send", func(t *testing.T) { t.Parallel() events := make(chan domain.StreamEvent, 1) - if err := emitToolCallDelta(context.Background(), events, 3, `{"path":"main.go"}`); err != nil { + if err := emitToolCallDelta(context.Background(), events, 3, "call_123", `{"path":"main.go"}`); err != nil { t.Fatalf("emitToolCallDelta() error = %v", err) } got := <-events - if got.Type != domain.StreamEventToolCallDelta || got.ToolCallIndex != 3 || got.ToolArgumentsDelta != `{"path":"main.go"}` { + if got.Type != domain.StreamEventToolCallDelta || got.ToolCallIndex != 3 || got.ToolArgumentsDelta != `{"path":"main.go"}` || got.ToolCallID != "call_123" { t.Fatalf("unexpected event: %+v", got) } }) @@ -126,7 +126,7 @@ func TestEmitToolCallDelta(t *testing.T) { t.Parallel() cancelledCtx, cancel := context.WithCancel(context.Background()) cancel() - if err := emitToolCallDelta(cancelledCtx, make(chan domain.StreamEvent), 0, "args"); err == nil { + if err := emitToolCallDelta(cancelledCtx, make(chan domain.StreamEvent), 0, "", "args"); err == nil { t.Fatal("expected cancellation error") } }) diff --git a/internal/provider/types.go b/internal/provider/types.go index eba95f19..294d31b1 100644 --- a/internal/provider/types.go +++ b/internal/provider/types.go @@ -68,8 +68,13 @@ const ( ) // StreamEvent 表示 provider 驱动层向 runtime 推送的流式事件。 +// 强制使用 NewXxxStreamEvent 构造器创建实例，禁止直接构造。 type StreamEvent struct { - Type StreamEventType + Type StreamEventType `json:"type"` + Payload interface{} `json:"payload,omitempty"` // 强类型载荷，使用类型断言访问 + + // --- 以下字段已弃用，保留仅用于 Phase 1 向后兼容 --- + // Phase 2 将由 Runtime 负责人移除，届时所有消费方应通过 Payload 类型断言访问。 // text_delta Text string `json:"text,omitempty"` // 文本片段 @@ -84,3 +89,77 @@ type StreamEvent struct { FinishReason string `json:"finish_reason,omitempty"` // 结束原因（仅 message_done 时有效） Usage *Usage `json:"usage,omitempty"` // 使用统计（仅 message_done 时有效） } + +// --- Payload 强类型定义 --- + +// TextDeltaPayload 表示文本增量事件的载荷。 +type TextDeltaPayload struct { + Text string `json:"text"` +} + +// ToolCallStartPayload 表示工具调用开始事件的载荷。 +type ToolCallStartPayload struct { + Index int `json:"index"` + ID string `json:"id"` + Name string `json:"name"` +} + +// ToolCallDeltaPayload 表示工具调用参数增量事件的载荷。 +type ToolCallDeltaPayload struct { + Index int `json:"index"` + ID string `json:"id"` + ArgumentsDelta string `json:"arguments_delta"` +} + +// MessageDonePayload 表示消息完成事件的载荷。 +type MessageDonePayload struct { + FinishReason string `json:"finish_reason"` + Usage *Usage `json:"usage"` +} + +// --- 构造器 --- + +// NewTextDeltaStreamEvent 创建文本增量流事件。 +func NewTextDeltaStreamEvent(text string) StreamEvent { + return StreamEvent{ + Type: StreamEventTextDelta, + Payload: TextDeltaPayload{Text: text}, + // 兼容层：同步填充弃用字段 + Text: text, + } +} + +// NewToolCallStartStreamEvent 创建工具调用开始流事件。 +func NewToolCallStartStreamEvent(index int, id, name string) StreamEvent { + return StreamEvent{ + Type: StreamEventToolCallStart, + Payload: ToolCallStartPayload{Index: index, ID: id, Name: name}, + // 兼容层：同步填充弃用字段 + ToolCallIndex: index, + ToolCallID: id, + ToolName: name, + } +} + +// NewToolCallDeltaStreamEvent 创建工具调用参数增量流事件。 +func NewToolCallDeltaStreamEvent(index int, id, argumentsDelta string) StreamEvent { + return StreamEvent{ + Type: StreamEventToolCallDelta, + Payload: ToolCallDeltaPayload{Index: index, ID: id, ArgumentsDelta: argumentsDelta}, + // 兼容层：同步填充弃用字段 + ToolCallIndex: index, + ToolCallID: id, + ToolArgumentsDelta: argumentsDelta, + } +} + +// NewMessageDoneStreamEvent 创建消息完成流事件。 +func NewMessageDoneStreamEvent(finishReason string, usage *Usage) StreamEvent { + return StreamEvent{ + Type: StreamEventMessageDone, + Payload: MessageDonePayload{FinishReason: finishReason, Usage: usage}, + // 兼容层：同步填充弃用字段 + FinishReason: finishReason, + Usage: usage, + } +} From 87e9752a96a9fc733a4aac31305b6a524b30014a Mon Sep 17 00:00:00 2001 From: phantom5099 <1011668688@qq.com> Date: Sun, 5 Apr 2026 14:53:18 +0800 Subject: [PATCH 2/4] =?UTF-8?q?pref(provider):=E5=8E=BB=E6=8E=89=E4=BA=86p?= =?UTF-8?q?rovider=E4=B8=BAruntime=E8=BF=AD=E4=BB=A3=E7=95=99=E4=B8=8B?= =?UTF-8?q?=E7=9A=84=E5=85=BC=E5=AE=B9=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/provider/builtin/builtin.go | 4 +- internal/provider/builtin/builtin_test.go | 4 +- internal/provider/catalog/service.go | 4 +- internal/provider/catalog/service_test.go | 6 +- internal/provider/catalog/store.go | 18 +- internal/provider/catalog/store_test.go | 12 +- internal/provider/openai/openai.go | 81 ++--- internal/provider/openai/openai_test.go | 200 ++++++++---- internal/provider/provider.go | 2 +- internal/provider/registry_test.go | 4 +- internal/provider/types.go | 60 +--- internal/runtime/compact_generator.go | 47 ++- internal/runtime/compact_generator_test.go | 53 ++- internal/runtime/runtime.go | 119 ++++++- internal/runtime/runtime_test.go | 354 +++++++-------------- internal/tui/update_test.go | 4 +- 16 files changed, 481 insertions(+), 491 deletions(-) diff --git a/internal/provider/builtin/builtin.go b/internal/provider/builtin/builtin.go index 0c10abe1..33ab5b5e 100644 --- a/internal/provider/builtin/builtin.go +++ b/internal/provider/builtin/builtin.go @@ -9,13 +9,13 @@ import ( func NewRegistry() (*provider.Registry, error) { registry := provider.NewRegistry() - if err := Register(registry); err != nil { + if err := register(registry); err != nil { return nil, err } return registry, nil } -func Register(registry *provider.Registry) error { +func register(registry *provider.Registry) error { if registry == nil { return errors.New("builtin provider registry is nil") } diff --git a/internal/provider/builtin/builtin_test.go b/internal/provider/builtin/builtin_test.go index fb230a0c..a078b3da 100644 --- a/internal/provider/builtin/builtin_test.go +++ b/internal/provider/builtin/builtin_test.go @@ -27,7 +27,7 @@ func TestRegister(t *testing.T) { t.Run("nil registry", func(t *testing.T) { t.Parallel() - err := Register(nil) + err := register(nil) if err == nil { t.Fatal("expected error for nil registry") } @@ -39,7 +39,7 @@ func TestRegister(t *testing.T) { t.Run("valid registry", func(t *testing.T) { t.Parallel() registry := provider.NewRegistry() - err := Register(registry) + err := register(registry) if err != nil { t.Fatalf("Register() error = %v", err) } diff --git a/internal/provider/catalog/service.go b/internal/provider/catalog/service.go index cb229990..6fd9a2ee 100644 --- a/internal/provider/catalog/service.go +++ b/internal/provider/catalog/service.go @@ -29,7 +29,7 @@ type Service struct { func NewService(baseDir string, registry *provider.Registry, store Store) *Service { if store == nil && strings.TrimSpace(baseDir) != "" { - store = NewJSONStore(baseDir) + store = newJSONStore(baseDir) } return &Service{ @@ -174,7 +174,7 @@ func (s *Service) discoverAndPersist(ctx context.Context, providerCfg config.Pro now := s.now() _ = s.store.Save(ctx, ModelCatalog{ - SchemaVersion: SchemaVersion, + SchemaVersion: schemaVersion, Identity: identity, FetchedAt: now, ExpiresAt: now.Add(s.catalogTTL), diff --git a/internal/provider/catalog/service_test.go b/internal/provider/catalog/service_test.go index 7aaa5c34..3d5d0b05 100644 --- a/internal/provider/catalog/service_test.go +++ b/internal/provider/catalog/service_test.go @@ -136,7 +136,7 @@ func TestListProviderModelsReturnsStaleCacheAndRefreshesInBackground(t *testing. store := newMemoryStore() now := time.Date(2026, 4, 2, 12, 0, 0, 0, time.UTC) if err := store.Save(context.Background(), ModelCatalog{ - SchemaVersion: SchemaVersion, + SchemaVersion: schemaVersion, Identity: identity, FetchedAt: now.Add(-48 * time.Hour), ExpiresAt: now.Add(-24 * time.Hour), @@ -234,8 +234,8 @@ func containsModelDescriptorID(models []config.ModelDescriptor, modelID string) type catalogTestProvider struct{} -func (catalogTestProvider) Chat(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) (provider.ChatResponse, error) { - return provider.ChatResponse{}, nil +func (catalogTestProvider) Chat(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) error { + return nil } type memoryStore struct { diff --git a/internal/provider/catalog/store.go b/internal/provider/catalog/store.go index 522dce7c..96ce6f7b 100644 --- a/internal/provider/catalog/store.go +++ b/internal/provider/catalog/store.go @@ -16,7 +16,7 @@ import ( "neo-code/internal/config" ) -const SchemaVersion = 1 +const schemaVersion = 1 var ErrCatalogNotFound = errors.New("provider: model catalog not found") @@ -39,18 +39,18 @@ type Store interface { Save(ctx context.Context, catalog ModelCatalog) error } -type JSONStore struct { +type jsonStore struct { dir string mu sync.RWMutex } -func NewJSONStore(baseDir string) *JSONStore { - return &JSONStore{ +func newJSONStore(baseDir string) *jsonStore { + return &jsonStore{ dir: filepath.Join(strings.TrimSpace(baseDir), "cache", "models"), } } -func (s *JSONStore) Load(ctx context.Context, identity config.ProviderIdentity) (ModelCatalog, error) { +func (s *jsonStore) Load(ctx context.Context, identity config.ProviderIdentity) (ModelCatalog, error) { if err := ctx.Err(); err != nil { return ModelCatalog{}, err } @@ -80,7 +80,7 @@ func (s *JSONStore) Load(ctx context.Context, identity config.ProviderIdentity) return normalizeCatalog(modelCatalog), nil } -func (s *JSONStore) Save(ctx context.Context, catalog ModelCatalog) error { +func (s *jsonStore) Save(ctx context.Context, catalog ModelCatalog) error { if err := ctx.Err(); err != nil { return err } @@ -111,18 +111,18 @@ func (s *JSONStore) Save(ctx context.Context, catalog ModelCatalog) error { func normalizeCatalog(modelCatalog ModelCatalog) ModelCatalog { if modelCatalog.SchemaVersion == 0 { - modelCatalog.SchemaVersion = SchemaVersion + modelCatalog.SchemaVersion = schemaVersion } modelCatalog.Models = config.MergeModelDescriptors(modelCatalog.Models) return modelCatalog } -func (s *JSONStore) catalogPath(identity config.ProviderIdentity) string { +func (s *jsonStore) catalogPath(identity config.ProviderIdentity) string { sum := sha256.Sum256([]byte(identity.Key())) return filepath.Join(s.dir, hex.EncodeToString(sum[:])+".json") } -func (s *JSONStore) writeCatalogFile(path string, data []byte) error { +func (s *jsonStore) writeCatalogFile(path string, data []byte) error { if err := os.MkdirAll(s.dir, 0o755); err != nil { return fmt.Errorf("provider: create model catalog dir: %w", err) } diff --git a/internal/provider/catalog/store_test.go b/internal/provider/catalog/store_test.go index ed20e435..86c7c098 100644 --- a/internal/provider/catalog/store_test.go +++ b/internal/provider/catalog/store_test.go @@ -14,14 +14,14 @@ import ( func TestJSONStoreRoundTrip(t *testing.T) { t.Parallel() - store := NewJSONStore(t.TempDir()) + store := newJSONStore(t.TempDir()) identity, err := config.NewProviderIdentity("openai", "https://api.openai.com/v1") if err != nil { t.Fatalf("NewProviderIdentity() error = %v", err) } expected := ModelCatalog{ - SchemaVersion: SchemaVersion, + SchemaVersion: schemaVersion, Identity: identity, FetchedAt: time.Date(2026, 4, 2, 10, 0, 0, 0, time.UTC), ExpiresAt: time.Date(2026, 4, 3, 10, 0, 0, 0, time.UTC), @@ -70,7 +70,7 @@ func TestJSONStoreRoundTrip(t *testing.T) { func TestJSONStoreMissingCatalog(t *testing.T) { t.Parallel() - store := NewJSONStore(t.TempDir()) + store := newJSONStore(t.TempDir()) identity, err := config.NewProviderIdentity("openai", "https://api.openai.com/v1") if err != nil { t.Fatalf("NewProviderIdentity() error = %v", err) @@ -86,14 +86,14 @@ func TestJSONStoreSaveReplacesExistingCatalogWithoutTempLeak(t *testing.T) { t.Parallel() baseDir := t.TempDir() - store := NewJSONStore(baseDir) + store := newJSONStore(baseDir) identity, err := config.NewProviderIdentity("openai", "https://api.openai.com/v1") if err != nil { t.Fatalf("NewProviderIdentity() error = %v", err) } first := ModelCatalog{ - SchemaVersion: SchemaVersion, + SchemaVersion: schemaVersion, Identity: identity, FetchedAt: time.Date(2026, 4, 2, 10, 0, 0, 0, time.UTC), ExpiresAt: time.Date(2026, 4, 3, 10, 0, 0, 0, time.UTC), @@ -102,7 +102,7 @@ func TestJSONStoreSaveReplacesExistingCatalogWithoutTempLeak(t *testing.T) { }, } second := ModelCatalog{ - SchemaVersion: SchemaVersion, + SchemaVersion: schemaVersion, Identity: identity, FetchedAt: time.Date(2026, 4, 4, 10, 0, 0, 0, time.UTC), ExpiresAt: time.Date(2026, 4, 5, 10, 0, 0, 0, time.UTC), diff --git a/internal/provider/openai/openai.go b/internal/provider/openai/openai.go index d67ee294..f688943d 100644 --- a/internal/provider/openai/openai.go +++ b/internal/provider/openai/openai.go @@ -10,7 +10,6 @@ import ( "io" "log" "net/http" - "sort" "strings" "time" @@ -29,10 +28,10 @@ type buildOptions struct { transport http.RoundTripper } -type BuildOption func(*buildOptions) +type buildOption func(*buildOptions) -// WithTransport 注入自定义 HTTP Transport（如 RetryTransport）。 -func WithTransport(rt http.RoundTripper) BuildOption { +// withTransport 注入自定义 HTTP Transport（如 RetryTransport）。 +func withTransport(rt http.RoundTripper) buildOption { return func(o *buildOptions) { o.transport = rt } @@ -50,10 +49,10 @@ func Driver() provider.DriverDefinition { return provider.DriverDefinition{ Name: DriverName, Build: func(ctx context.Context, cfg config.ResolvedProviderConfig) (provider.Provider, error) { - return New(cfg, WithTransport(defaultRetryTransport())) + return New(cfg, withTransport(defaultRetryTransport())) }, Discover: func(ctx context.Context, cfg config.ResolvedProviderConfig) ([]config.ModelDescriptor, error) { - p, err := New(cfg, WithTransport(defaultRetryTransport())) + p, err := New(cfg, withTransport(defaultRetryTransport())) if err != nil { return nil, err } @@ -62,7 +61,7 @@ func Driver() provider.DriverDefinition { } } -func New(cfg config.ResolvedProviderConfig, opts ...BuildOption) (*Provider, error) { +func New(cfg config.ResolvedProviderConfig, opts ...buildOption) (*Provider, error) { if err := cfg.Validate(); err != nil { return nil, fmt.Errorf("openai provider: %w", err) } @@ -103,21 +102,21 @@ func (p *Provider) DiscoverModels(ctx context.Context) ([]config.ModelDescriptor return config.MergeModelDescriptors(descriptors), nil } -func (p *Provider) Chat(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) (provider.ChatResponse, error) { +func (p *Provider) Chat(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) error { payload, err := p.buildRequest(req) if err != nil { - return provider.ChatResponse{}, err + return err } body, err := json.Marshal(payload) if err != nil { - return provider.ChatResponse{}, fmt.Errorf("openai provider: marshal request: %w", err) + return fmt.Errorf("openai provider: marshal request: %w", err) } endpoint := strings.TrimRight(p.cfg.BaseURL, "/") + "/chat/completions" httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) if err != nil { - return provider.ChatResponse{}, fmt.Errorf("openai provider: build request: %w", err) + return fmt.Errorf("openai provider: build request: %w", err) } httpReq.Header.Set("Authorization", "Bearer "+p.cfg.APIKey) httpReq.Header.Set("Content-Type", "application/json") @@ -125,7 +124,7 @@ func (p *Provider) Chat(ctx context.Context, req provider.ChatRequest, events ch resp, err := p.client.Do(httpReq) if err != nil { - return provider.ChatResponse{}, fmt.Errorf("openai provider: send request: %w", err) + return fmt.Errorf("openai provider: send request: %w", err) } defer func(Body io.ReadCloser) { err := Body.Close() @@ -135,7 +134,7 @@ func (p *Provider) Chat(ctx context.Context, req provider.ChatRequest, events ch }(resp.Body) if resp.StatusCode >= http.StatusBadRequest { - return provider.ChatResponse{}, p.parseError(resp) + return p.parseError(resp) } return p.consumeStream(ctx, resp.Body, events) @@ -185,14 +184,13 @@ func (p *Provider) buildRequest(req provider.ChatRequest) (chatCompletionRequest return payload, nil } -func (p *Provider) consumeStream(ctx context.Context, body io.Reader, events chan<- provider.StreamEvent) (provider.ChatResponse, error) { +func (p *Provider) consumeStream(ctx context.Context, body io.Reader, events chan<- provider.StreamEvent) error { reader := bufio.NewReader(body) var ( - contentBuilder strings.Builder - finishReason string - usage provider.Usage - done bool + finishReason string + usage provider.Usage + done bool ) toolCalls := make(map[int]*provider.ToolCall) @@ -222,7 +220,6 @@ func (p *Provider) consumeStream(ctx context.Context, body io.Reader, events cha finishReason = choice.FinishReason } if choice.Delta.Content != "" { - contentBuilder.WriteString(choice.Delta.Content) if err := emitTextDelta(ctx, events, choice.Delta.Content); err != nil { return err } @@ -236,12 +233,12 @@ func (p *Provider) consumeStream(ctx context.Context, body io.Reader, events cha return nil } - // finishStream 统一的流结束处理：发送 message_done 事件并组装最终响应。 - finishStream := func() (provider.ChatResponse, error) { + // finishStream 统一的流结束处理：发送 message_done 事件。 + finishStream := func() error { if err := emitMessageDone(ctx, events, finishReason, &usage); err != nil { - return provider.ChatResponse{}, err + return err } - return finalizeResponse(contentBuilder.String(), toolCalls, finishReason, usage), nil + return nil } flushPendingData := func() error { @@ -254,7 +251,7 @@ func (p *Provider) consumeStream(ctx context.Context, body io.Reader, events cha for { line, err := reader.ReadString('\n') if err != nil && !errors.Is(err, io.EOF) { - return provider.ChatResponse{}, fmt.Errorf("openai provider: read stream: %w", err) + return fmt.Errorf("openai provider: read stream: %w", err) } line = strings.TrimRight(line, "\r\n") @@ -265,7 +262,7 @@ func (p *Provider) consumeStream(ctx context.Context, body io.Reader, events cha dataLines = append(dataLines, strings.TrimSpace(strings.TrimPrefix(trimmed, "data:"))) case trimmed == "": if flushErr := flushPendingData(); flushErr != nil { - return provider.ChatResponse{}, flushErr + return flushErr } if done { return finishStream() @@ -276,7 +273,7 @@ func (p *Provider) consumeStream(ctx context.Context, body io.Reader, events cha if errors.Is(err, io.EOF) { if flushErr := flushPendingData(); flushErr != nil { - return provider.ChatResponse{}, flushErr + return flushErr } return finishStream() } @@ -405,38 +402,6 @@ func mergeToolCallDelta(ctx context.Context, events chan<- provider.StreamEvent, return nil } -// finalizeResponse 将累积的内容、tool calls 和元数据组装为最终 ChatResponse。 -func finalizeResponse(content string, toolCalls map[int]*provider.ToolCall, finishReason string, usage provider.Usage) provider.ChatResponse { - ordered := make([]int, 0, len(toolCalls)) - for index := range toolCalls { - ordered = append(ordered, index) - } - sort.Ints(ordered) - - message := provider.Message{ - Role: provider.RoleAssistant, - Content: content, - } - - for _, index := range ordered { - call := toolCalls[index] - if call == nil { - continue - } - message.ToolCalls = append(message.ToolCalls, *call) - } - - if finishReason == "" && len(message.ToolCalls) > 0 { - finishReason = "tool_calls" - } - - return provider.ChatResponse{ - Message: message, - FinishReason: finishReason, - Usage: usage, - } -} - func emitStreamEvent(ctx context.Context, events chan<- provider.StreamEvent, event provider.StreamEvent) error { if events == nil { return nil diff --git a/internal/provider/openai/openai_test.go b/internal/provider/openai/openai_test.go index cdf62bac..9b0fc108 100644 --- a/internal/provider/openai/openai_test.go +++ b/internal/provider/openai/openai_test.go @@ -33,7 +33,7 @@ func TestWithTransport(t *testing.T) { customTransport := &http.Transport{} cfg := resolvedConfig("", "") - provider, err := New(cfg, WithTransport(customTransport)) + provider, err := New(cfg, withTransport(customTransport)) if err != nil { t.Fatalf("New() error = %v", err) } @@ -117,7 +117,8 @@ func TestEmitToolCallDelta(t *testing.T) { t.Fatalf("emitToolCallDelta() error = %v", err) } got := <-events - if got.Type != domain.StreamEventToolCallDelta || got.ToolCallIndex != 3 || got.ToolArgumentsDelta != `{"path":"main.go"}` || got.ToolCallID != "call_123" { + payload := requireToolCallDeltaPayload(t, got) + if got.Type != domain.StreamEventToolCallDelta || payload.Index != 3 || payload.ArgumentsDelta != `{"path":"main.go"}` || payload.ID != "call_123" { t.Fatalf("unexpected event: %+v", got) } }) @@ -150,7 +151,8 @@ func TestEmitMessageDone(t *testing.T) { t.Fatalf("emitMessageDone() error = %v", err) } got := <-events - if got.Type != domain.StreamEventMessageDone || got.FinishReason != "stop" || got.Usage.TotalTokens != 100 { + payload := requireMessageDonePayload(t, got) + if got.Type != domain.StreamEventMessageDone || payload.FinishReason != "stop" || payload.Usage == nil || payload.Usage.TotalTokens != 100 { t.Fatalf("unexpected event: %+v", got) } }) @@ -290,7 +292,7 @@ func TestProviderChatConsumesSSEAndMergesToolCalls(t *testing.T) { provider.client = server.Client() events := make(chan domain.StreamEvent, 8) - response, err := provider.Chat(context.Background(), domain.ChatRequest{ + err = provider.Chat(context.Background(), domain.ChatRequest{ Model: "gpt-5.4", Messages: []domain.Message{ {Role: "user", Content: "please edit the file"}, @@ -320,38 +322,57 @@ func TestProviderChatConsumesSSEAndMergesToolCalls(t *testing.T) { t.Fatalf("Chat() error = %v", err) } - if response.Message.Content != "Hello world" { - t.Fatalf("expected content %q, got %q", "Hello world", response.Message.Content) + streamEvents := drainStreamEvents(events) + if len(streamEvents) == 0 { + t.Fatal("expected streamed events") } - if response.FinishReason != "tool_calls" { - t.Fatalf("expected finish reason tool_calls, got %q", response.FinishReason) + + var ( + chunks []string + toolCallStartSeen bool + toolCallArgs strings.Builder + messageDone *domain.MessageDonePayload + ) + + for _, event := range streamEvents { + switch event.Type { + case domain.StreamEventTextDelta: + chunks = append(chunks, requireTextDeltaPayload(t, event).Text) + case domain.StreamEventToolCallStart: + payload := requireToolCallStartPayload(t, event) + toolCallStartSeen = true + if payload.Index != 0 || payload.ID != "call_1" || payload.Name != "filesystem_edit" { + t.Fatalf("unexpected tool_call_start payload: %+v", payload) + } + case domain.StreamEventToolCallDelta: + payload := requireToolCallDeltaPayload(t, event) + if payload.Index != 0 || payload.ID != "call_1" { + t.Fatalf("unexpected tool_call_delta payload: %+v", payload) + } + toolCallArgs.WriteString(payload.ArgumentsDelta) + case domain.StreamEventMessageDone: + payload := requireMessageDonePayload(t, event) + messageDone = &payload + } } - if response.Usage.TotalTokens != 15 { - t.Fatalf("expected total tokens 15, got %d", response.Usage.TotalTokens) + + if strings.Join(chunks, "") != "Hello world" { + t.Fatalf("expected streamed chunks to form %q, got %q", "Hello world", strings.Join(chunks, "")) } - if len(response.Message.ToolCalls) != 1 { - t.Fatalf("expected 1 tool call, got %d", len(response.Message.ToolCalls)) + if !toolCallStartSeen { + t.Fatal("expected tool_call_start event") } - - call := response.Message.ToolCalls[0] - if call.ID != "call_1" || call.Name != "filesystem_edit" { - t.Fatalf("unexpected tool call: %+v", call) + if toolCallArgs.String() != `{"path":"main.go","search_string":"old","replace_string":"new"}` { + t.Fatalf("unexpected merged tool arguments: %q", toolCallArgs.String()) } - if call.Arguments != `{"path":"main.go","search_string":"old","replace_string":"new"}` { - t.Fatalf("unexpected merged arguments: %q", call.Arguments) + if messageDone == nil { + t.Fatal("expected message_done event") } - - var chunks []string - for { - select { - case event := <-events: - chunks = append(chunks, event.Text) - default: - if strings.Join(chunks, "") != "Hello world" { - t.Fatalf("expected streamed chunks to form %q, got %q", "Hello world", strings.Join(chunks, "")) - } - return - } + if messageDone.FinishReason != "tool_calls" { + t.Fatalf("expected finish reason %q, got %q", "tool_calls", messageDone.FinishReason) + } + if messageDone.Usage == nil || messageDone.Usage.TotalTokens != 15 { + t.Fatalf("expected total tokens 15, got %+v", messageDone.Usage) } } @@ -395,7 +416,7 @@ func TestProviderChatHTTPErrorResponses(t *testing.T) { } provider.client = server.Client() - _, err = provider.Chat(context.Background(), domain.ChatRequest{ + err = provider.Chat(context.Background(), domain.ChatRequest{ Model: config.OpenAIDefaultModel, }, make(chan domain.StreamEvent, 1)) if err == nil || !strings.Contains(err.Error(), tt.expectErr) { @@ -513,7 +534,7 @@ func TestParseErrorAndEmitTextDelta(t *testing.T) { if err := emitTextDelta(context.Background(), eventCh, "chunk"); err != nil { t.Fatalf("emitTextDelta() error = %v", err) } - if got := <-eventCh; got.Text != "chunk" || got.Type != domain.StreamEventTextDelta { + if got := <-eventCh; got.Type != domain.StreamEventTextDelta || requireTextDeltaPayload(t, got).Text != "chunk" { t.Fatalf("unexpected stream event: %+v", got) } @@ -532,12 +553,63 @@ func TestProviderConsumeStreamRejectsDirtyJSON(t *testing.T) { t.Fatalf("New() error = %v", err) } - _, err = provider.consumeStream(context.Background(), strings.NewReader("data: {not-json}\n\n"), make(chan domain.StreamEvent, 1)) + err = provider.consumeStream(context.Background(), strings.NewReader("data: {not-json}\n\n"), make(chan domain.StreamEvent, 1)) if err == nil || !strings.Contains(err.Error(), "decode stream chunk") { t.Fatalf("expected dirty JSON decode error, got %v", err) } } +func drainStreamEvents(events <-chan domain.StreamEvent) []domain.StreamEvent { + drained := make([]domain.StreamEvent, 0) + for { + select { + case evt, ok := <-events: + if !ok { + return drained + } + drained = append(drained, evt) + default: + return drained + } + } +} + +func requireTextDeltaPayload(t *testing.T, event domain.StreamEvent) domain.TextDeltaPayload { + t.Helper() + payload, ok := event.Payload.(domain.TextDeltaPayload) + if !ok { + t.Fatalf("expected TextDeltaPayload, got %T", event.Payload) + } + return payload +} + +func requireToolCallStartPayload(t *testing.T, event domain.StreamEvent) domain.ToolCallStartPayload { + t.Helper() + payload, ok := event.Payload.(domain.ToolCallStartPayload) + if !ok { + t.Fatalf("expected ToolCallStartPayload, got %T", event.Payload) + } + return payload +} + +func requireToolCallDeltaPayload(t *testing.T, event domain.StreamEvent) domain.ToolCallDeltaPayload { + t.Helper() + payload, ok := event.Payload.(domain.ToolCallDeltaPayload) + if !ok { + t.Fatalf("expected ToolCallDeltaPayload, got %T", event.Payload) + } + return payload +} + +func requireMessageDonePayload(t *testing.T, event domain.StreamEvent) domain.MessageDonePayload { + t.Helper() + payload, ok := event.Payload.(domain.MessageDonePayload) + if !ok { + t.Fatalf("expected MessageDonePayload, got %T", event.Payload) + } + return payload +} + func containsToolRoleMessage(messages []openAIMessage, toolCallID string, content string) bool { for _, message := range messages { if message.Role == "tool" && message.ToolCallID == toolCallID && message.Content == content { @@ -601,7 +673,8 @@ func TestEmitToolCallStartGuards(t *testing.T) { t.Fatalf("emitToolCallStart() error = %v", err) } got := <-events - if got.Type != domain.StreamEventToolCallStart || got.ToolName != "filesystem_edit" || got.ToolCallID != "call-1" || got.ToolCallIndex != 2 { + payload := requireToolCallStartPayload(t, got) + if got.Type != domain.StreamEventToolCallStart || payload.Name != "filesystem_edit" || payload.ID != "call-1" || payload.Index != 2 { t.Fatalf("unexpected event: %+v", got) } @@ -646,16 +719,18 @@ func TestMergeToolCallDeltaEmitsStartWhenNameArrivesLater(t *testing.T) { if start.Type != domain.StreamEventToolCallStart { t.Fatalf("expected tool_call_start event, got %+v", start) } - if start.ToolCallID != "call_late_name" || start.ToolName != "filesystem_edit" { - t.Fatalf("unexpected tool_call_start payload: %+v", start) + startPayload := requireToolCallStartPayload(t, start) + if startPayload.ID != "call_late_name" || startPayload.Name != "filesystem_edit" { + t.Fatalf("unexpected tool_call_start payload: %+v", startPayload) } delta := <-events if delta.Type != domain.StreamEventToolCallDelta { t.Fatalf("expected tool_call_delta event, got %+v", delta) } - if delta.ToolArgumentsDelta != `{"path":"main.go"}` { - t.Fatalf("unexpected tool arguments delta: %+v", delta) + deltaPayload := requireToolCallDeltaPayload(t, delta) + if deltaPayload.ArgumentsDelta != `{"path":"main.go"}` { + t.Fatalf("unexpected tool arguments delta: %+v", deltaPayload) } call := toolCalls[0] @@ -703,7 +778,7 @@ func TestProviderChatEmitsToolCallStartEvent(t *testing.T) { provider.client = server.Client() events := make(chan domain.StreamEvent, 8) - _, err = provider.Chat(context.Background(), domain.ChatRequest{ + err = provider.Chat(context.Background(), domain.ChatRequest{ Model: config.OpenAIDefaultModel, Messages: []domain.Message{{Role: "user", Content: "edit"}}, Tools: []domain.ToolSpec{ @@ -714,17 +789,16 @@ func TestProviderChatEmitsToolCallStartEvent(t *testing.T) { t.Fatalf("Chat() error = %v", err) } - close(events) - var foundToolCallStart bool - for evt := range events { + for _, evt := range drainStreamEvents(events) { if evt.Type == domain.StreamEventToolCallStart { foundToolCallStart = true - if evt.ToolName != "filesystem_edit" { - t.Fatalf("expected ToolName %q, got %q", "filesystem_edit", evt.ToolName) + payload := requireToolCallStartPayload(t, evt) + if payload.Name != "filesystem_edit" { + t.Fatalf("expected ToolName %q, got %q", "filesystem_edit", payload.Name) } - if evt.ToolCallID != "call_tool" { - t.Fatalf("expected ToolCallID %q, got %q", "call_tool", evt.ToolCallID) + if payload.ID != "call_tool" { + t.Fatalf("expected ToolCallID %q, got %q", "call_tool", payload.ID) } } } @@ -819,7 +893,7 @@ func TestProviderChatEmitsFullEventStream(t *testing.T) { provider.client = server.Client() events := make(chan domain.StreamEvent, 16) - _, err = provider.Chat(context.Background(), domain.ChatRequest{ + err = provider.Chat(context.Background(), domain.ChatRequest{ Model: config.OpenAIDefaultModel, Messages: []domain.Message{{Role: "user", Content: "test"}}, }, events) @@ -827,35 +901,35 @@ func TestProviderChatEmitsFullEventStream(t *testing.T) { t.Fatalf("Chat() error = %v", err) } - close(events) - var ( foundTextDelta bool foundToolCallStart bool foundToolCallDelta bool foundMessageDone bool toolCallDeltaContent string - messageDoneEvt *domain.StreamEvent + messageDonePayload *domain.MessageDonePayload ) - for evt := range events { + for _, evt := range drainStreamEvents(events) { switch evt.Type { case domain.StreamEventTextDelta: foundTextDelta = true case domain.StreamEventToolCallStart: foundToolCallStart = true - if evt.ToolName != "filesystem_edit" { - t.Fatalf("expected ToolName %q, got %q", "filesystem_edit", evt.ToolName) + payload := requireToolCallStartPayload(t, evt) + if payload.Name != "filesystem_edit" { + t.Fatalf("expected ToolName %q, got %q", "filesystem_edit", payload.Name) } - if evt.ToolCallIndex != 0 { - t.Fatalf("expected ToolCallIndex %d for tool_call_start, got %d", 0, evt.ToolCallIndex) + if payload.Index != 0 { + t.Fatalf("expected ToolCallIndex %d for tool_call_start, got %d", 0, payload.Index) } case domain.StreamEventToolCallDelta: foundToolCallDelta = true - toolCallDeltaContent += evt.ToolArgumentsDelta + toolCallDeltaContent += requireToolCallDeltaPayload(t, evt).ArgumentsDelta case domain.StreamEventMessageDone: foundMessageDone = true - messageDoneEvt = &evt + payload := requireMessageDonePayload(t, evt) + messageDonePayload = &payload } } @@ -879,16 +953,16 @@ func TestProviderChatEmitsFullEventStream(t *testing.T) { } // 验证 message_done 事件包含正确的字段 - if messageDoneEvt == nil { + if messageDonePayload == nil { t.Fatal("message_done event is nil") } - if messageDoneEvt.FinishReason != "tool_calls" { - t.Fatalf("expected FinishReason %q, got %q", "tool_calls", messageDoneEvt.FinishReason) + if messageDonePayload.FinishReason != "tool_calls" { + t.Fatalf("expected FinishReason %q, got %q", "tool_calls", messageDonePayload.FinishReason) } - if messageDoneEvt.Usage == nil { + if messageDonePayload.Usage == nil { t.Fatal("expected Usage in message_done event") } - if messageDoneEvt.Usage.TotalTokens != 150 { - t.Fatalf("expected TotalTokens %d, got %d", 150, messageDoneEvt.Usage.TotalTokens) + if messageDonePayload.Usage.TotalTokens != 150 { + t.Fatalf("expected TotalTokens %d, got %d", 150, messageDonePayload.Usage.TotalTokens) } } diff --git a/internal/provider/provider.go b/internal/provider/provider.go index 40a838b9..91a99b52 100644 --- a/internal/provider/provider.go +++ b/internal/provider/provider.go @@ -3,5 +3,5 @@ package provider import "context" type Provider interface { - Chat(ctx context.Context, req ChatRequest, events chan<- StreamEvent) (ChatResponse, error) + Chat(ctx context.Context, req ChatRequest, events chan<- StreamEvent) error } diff --git a/internal/provider/registry_test.go b/internal/provider/registry_test.go index f650253d..df57c9da 100644 --- a/internal/provider/registry_test.go +++ b/internal/provider/registry_test.go @@ -12,8 +12,8 @@ import ( type stubProvider struct{} -func (stubProvider) Chat(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) (provider.ChatResponse, error) { - return provider.ChatResponse{}, nil +func (stubProvider) Chat(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) error { + return nil } func stubDriver(driverType string) provider.DriverDefinition { diff --git a/internal/provider/types.go b/internal/provider/types.go index 294d31b1..278a3a93 100644 --- a/internal/provider/types.go +++ b/internal/provider/types.go @@ -1,11 +1,14 @@ package provider -// Role 常量定义消息角色标识。 const ( - RoleSystem = "system" - RoleUser = "user" + // RoleSystem 标识系统消息。 + RoleSystem = "system" + // RoleUser 标识用户消息。 + RoleUser = "user" + // RoleAssistant 标识助手消息。 RoleAssistant = "assistant" - RoleTool = "tool" + // RoleTool 标识工具结果消息。 + RoleTool = "tool" ) // Message 表示对话中的单条消息。 @@ -39,13 +42,6 @@ type ChatRequest struct { Tools []ToolSpec `json:"tools,omitempty"` } -// ChatResponse 是 provider.Chat() 的返回结果。 -type ChatResponse struct { - Message Message `json:"message"` - FinishReason string `json:"finish_reason"` - Usage Usage `json:"usage"` -} - // Usage 记录本次请求的 token 使用统计。 type Usage struct { InputTokens int `json:"input_tokens"` @@ -59,39 +55,20 @@ type StreamEventType string const ( // StreamEventTextDelta 表示模型输出的文本片段。 StreamEventTextDelta StreamEventType = "text_delta" - // StreamEventToolCallStart 表示模型开始请求工具调用，TUI 可据此展示过渡提示。 + // StreamEventToolCallStart 表示模型开始请求工具调用。 StreamEventToolCallStart StreamEventType = "tool_call_start" // StreamEventToolCallDelta 表示工具调用参数的增量片段。 StreamEventToolCallDelta StreamEventType = "tool_call_delta" - // StreamEventMessageDone 表示本轮消息完成，包含最终统计信息。 + // StreamEventMessageDone 表示本轮消息完成，并携带最终统计信息。 StreamEventMessageDone StreamEventType = "message_done" ) -// StreamEvent 表示 provider 驱动层向 runtime 推送的流式事件。 -// 强制使用 NewXxxStreamEvent 构造器创建实例，禁止直接构造。 +// StreamEvent 表示 provider 向 runtime 推送的流式事件。 type StreamEvent struct { Type StreamEventType `json:"type"` - Payload interface{} `json:"payload,omitempty"` // 强类型载荷，使用类型断言访问 - - // --- 以下字段已弃用，保留仅用于 Phase 1 向后兼容 --- - // Phase 2 将由 Runtime 负责人移除，届时所有消费方应通过 Payload 类型断言访问。 - - // text_delta - Text string `json:"text,omitempty"` // 文本片段 - - // tool_call_start / tool_call_delta - ToolCallIndex int `json:"tool_call_index,omitempty"` // 工具调用索引 - ToolCallID string `json:"tool_call_id,omitempty"` // 工具调用 ID（tool_call_start 时使用） - ToolName string `json:"tool_name,omitempty"` // 工具名称（tool_call_start 时使用） - ToolArgumentsDelta string `json:"tool_arguments_delta,omitempty"` // 参数增量片段（tool_call_delta 时使用） - - // message_done - FinishReason string `json:"finish_reason,omitempty"` // 结束原因（仅 message_done 时有效） - Usage *Usage `json:"usage,omitempty"` // 使用统计（仅 message_done 时有效） + Payload interface{} `json:"payload,omitempty"` } -// --- Payload 强类型定义 --- - // TextDeltaPayload 表示文本增量事件的载荷。 type TextDeltaPayload struct { Text string `json:"text"` @@ -117,15 +94,11 @@ type MessageDonePayload struct { Usage *Usage `json:"usage"` } -// --- 构造器 --- - // NewTextDeltaStreamEvent 创建文本增量流事件。 func NewTextDeltaStreamEvent(text string) StreamEvent { return StreamEvent{ Type: StreamEventTextDelta, Payload: TextDeltaPayload{Text: text}, - // 兼容层：同步填充弃用字段 - Text: text, } } @@ -134,10 +107,6 @@ func NewToolCallStartStreamEvent(index int, id, name string) StreamEvent { return StreamEvent{ Type: StreamEventToolCallStart, Payload: ToolCallStartPayload{Index: index, ID: id, Name: name}, - // 兼容层：同步填充弃用字段 - ToolCallIndex: index, - ToolCallID: id, - ToolName: name, } } @@ -146,10 +115,6 @@ func NewToolCallDeltaStreamEvent(index int, id, argumentsDelta string) StreamEve return StreamEvent{ Type: StreamEventToolCallDelta, Payload: ToolCallDeltaPayload{Index: index, ID: id, ArgumentsDelta: argumentsDelta}, - // 兼容层：同步填充弃用字段 - ToolCallIndex: index, - ToolCallID: id, - ToolArgumentsDelta: argumentsDelta, } } @@ -158,8 +123,5 @@ func NewMessageDoneStreamEvent(finishReason string, usage *Usage) StreamEvent { return StreamEvent{ Type: StreamEventMessageDone, Payload: MessageDonePayload{FinishReason: finishReason, Usage: usage}, - // 兼容层：同步填充弃用字段 - FinishReason: finishReason, - Usage: usage, } } diff --git a/internal/runtime/compact_generator.go b/internal/runtime/compact_generator.go index 87d9d658..49e48755 100644 --- a/internal/runtime/compact_generator.go +++ b/internal/runtime/compact_generator.go @@ -56,22 +56,61 @@ func (g *compactSummaryGenerator) Generate(ctx context.Context, input contextcom if err != nil { return "", err } - resp, err := modelProvider.Chat(ctx, provider.ChatRequest{ + + // 使用流式事件通道收集 compact 摘要响应。 + streamEvents := make(chan provider.StreamEvent, 32) + streamDone := make(chan struct{}) + acc := newStreamAccumulator() + + go func() { + defer close(streamDone) + for { + select { + case event, ok := <-streamEvents: + if !ok { + return + } + switch event.Type { + case provider.StreamEventTextDelta: + if payload, ok := event.Payload.(provider.TextDeltaPayload); ok { + acc.accumulateTextDelta(payload.Text) + } + case provider.StreamEventToolCallStart: + if payload, ok := event.Payload.(provider.ToolCallStartPayload); ok { + acc.accumulateToolCallStart(payload.Index, payload.ID, payload.Name) + } + case provider.StreamEventToolCallDelta: + if payload, ok := event.Payload.(provider.ToolCallDeltaPayload); ok { + acc.accumulateToolCallDelta(payload.Index, payload.ID, payload.ArgumentsDelta) + } + } + case <-ctx.Done(): + return + } + } + }() + + err = modelProvider.Chat(ctx, provider.ChatRequest{ Model: g.model, SystemPrompt: prompt.SystemPrompt, Messages: []provider.Message{{ Role: provider.RoleUser, Content: prompt.UserPrompt, }}, - }, nil) + }, streamEvents) + close(streamEvents) + <-streamDone + if err != nil { return "", err } - if len(resp.Message.ToolCalls) > 0 { + + message := acc.buildMessage() + if len(message.ToolCalls) > 0 { return "", errors.New("runtime: compact summary response must not contain tool calls") } - summary := strings.TrimSpace(resp.Message.Content) + summary := strings.TrimSpace(message.Content) if summary == "" { return "", errors.New("runtime: compact summary response is empty") } diff --git a/internal/runtime/compact_generator_test.go b/internal/runtime/compact_generator_test.go index def57458..e0c348e7 100644 --- a/internal/runtime/compact_generator_test.go +++ b/internal/runtime/compact_generator_test.go @@ -20,28 +20,25 @@ func TestCompactSummaryGeneratorBuildsProviderRequestWithoutTools(t *testing.T) } scripted := &scriptedProvider{ - responses: []provider.ChatResponse{{ - Message: provider.Message{ - Role: provider.RoleAssistant, - Content: strings.Join([]string{ - "[compact_summary]", - "done:", - "- Completed the historical task and kept the final result.", - "", - "in_progress:", - "- Continue from the retained recent window.", - "", - "decisions:", - "- Keep the existing section layout for compatibility.", - "", - "code_changes:", - "- Updated compact summary generation behavior.", - "", - "constraints:", - "- Preserve only the minimum information needed to continue the work.", - }, "\n"), - }, - }}, + streams: [][]provider.StreamEvent{ + {provider.NewTextDeltaStreamEvent(strings.Join([]string{ + "[compact_summary]", + "done:", + "- Completed the historical task and kept the final result.", + "", + "in_progress:", + "- Continue from the retained recent window.", + "", + "decisions:", + "- Keep the existing section layout for compatibility.", + "", + "code_changes:", + "- Updated compact summary generation behavior.", + "", + "constraints:", + "- Preserve only the minimum information needed to continue the work.", + }, "\n"))}, + }, } factory := &scriptedProviderFactory{provider: scripted} generator := newCompactSummaryGenerator(factory, resolvedProvider, "session-model") @@ -116,14 +113,12 @@ func TestCompactSummaryGeneratorRejectsToolCalls(t *testing.T) { } scripted := &scriptedProvider{ - responses: []provider.ChatResponse{{ - Message: provider.Message{ - Role: provider.RoleAssistant, - ToolCalls: []provider.ToolCall{ - {ID: "call-1", Name: "filesystem_read_file", Arguments: "{}"}, - }, + streams: [][]provider.StreamEvent{ + { + provider.NewToolCallStartStreamEvent(0, "call-1", "filesystem_read_file"), + provider.NewToolCallDeltaStreamEvent(0, "call-1", "{}"), }, - }}, + }, } generator := newCompactSummaryGenerator(&scriptedProviderFactory{provider: scripted}, resolvedProvider, "session-model") diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index c95bbcd3..d3b3e29f 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -8,6 +8,7 @@ import ( "math/rand/v2" "os" "path/filepath" + "sort" "strings" "sync" "time" @@ -29,6 +30,67 @@ const ( providerRetryMaxWait = 5 * time.Second ) +// streamAccumulator 在流式事件处理过程中累积本轮对话需要持久化的助手消息状态， +// 包括文本内容和工具调用列表。 +type streamAccumulator struct { + content strings.Builder + toolCalls map[int]*provider.ToolCall +} + +// newStreamAccumulator 创建并初始化一个空的流式事件累积器。 +func newStreamAccumulator() *streamAccumulator { + return &streamAccumulator{ + toolCalls: make(map[int]*provider.ToolCall), + } +} + +// accumulateTextDelta 累积文本增量片段。 +func (a *streamAccumulator) accumulateTextDelta(text string) { + a.content.WriteString(text) +} + +// accumulateToolCallStart 记录新发现的工具调用（首次出现时创建条目）。 +func (a *streamAccumulator) accumulateToolCallStart(index int, id, name string) { + if _, exists := a.toolCalls[index]; !exists { + a.toolCalls[index] = &provider.ToolCall{ID: id, Name: name} + } +} + +// accumulateToolCallDelta 累积工具调用参数增量。 +func (a *streamAccumulator) accumulateToolCallDelta(index int, id, argumentsDelta string) { + call, exists := a.toolCalls[index] + if !exists { + call = &provider.ToolCall{ID: id} + a.toolCalls[index] = call + } + if name := call.Name; strings.TrimSpace(name) == "" && call.ID != "" { + // 首次出现 delta 时可能还未收到 start 事件，仅记录 ID + } + call.Arguments += argumentsDelta +} + +// buildMessage 从累积状态构建最终的 assistant Message 对象。 +func (a *streamAccumulator) buildMessage() provider.Message { + ordered := make([]int, 0, len(a.toolCalls)) + for index := range a.toolCalls { + ordered = append(ordered, index) + } + sort.Ints(ordered) + + message := provider.Message{ + Role: provider.RoleAssistant, + Content: a.content.String(), + } + for _, index := range ordered { + call := a.toolCalls[index] + if call == nil { + continue + } + message.ToolCalls = append(message.ToolCalls, *call) + } + return message +} + var runtimeSessionWorkdirs = struct { mu sync.RWMutex data map[string]string @@ -169,7 +231,7 @@ func (s *Service) Run(ctx context.Context, input UserInput) error { return s.handleRunError(ctx, input.RunID, session.ID, err) } - resp, err := s.callProviderWithRetry(ctx, input.RunID, session.ID, provider.ChatRequest{ + acc, err := s.callProviderWithRetry(ctx, input.RunID, session.ID, provider.ChatRequest{ Model: cfg.CurrentModel, SystemPrompt: builtContext.SystemPrompt, Messages: builtContext.Messages, @@ -186,7 +248,7 @@ func (s *Service) Run(ctx context.Context, input UserInput) error { session.Provider = cfg.SelectedProvider session.Model = cfg.CurrentModel - assistant := resp.Message + assistant := acc.buildMessage() if strings.TrimSpace(assistant.Role) == "" { assistant.Role = provider.RoleAssistant } @@ -417,9 +479,9 @@ func (s *Service) emit(ctx context.Context, kind EventType, runID string, sessio } } -// forwardProviderEvents 将 provider 流式事件转发为 runtime 事件。 +// forwardProviderEvents 将 provider 流式事件转发为 runtime 事件，同时向 accumulator 累积消息状态。 // 使用 select 同时监听输入通道和 context 取消信号，确保 goroutine 不会因通道阻塞而泄漏。 -func (s *Service) forwardProviderEvents(ctx context.Context, runID string, sessionID string, input <-chan provider.StreamEvent, done chan<- struct{}) { +func (s *Service) forwardProviderEvents(ctx context.Context, runID string, sessionID string, input <-chan provider.StreamEvent, done chan<- struct{}, acc *streamAccumulator) { defer close(done) for { select { @@ -429,9 +491,25 @@ func (s *Service) forwardProviderEvents(ctx context.Context, runID string, sessi } switch event.Type { case provider.StreamEventTextDelta: - s.emit(ctx, EventAgentChunk, runID, sessionID, event.Text) + if payload, ok := event.Payload.(provider.TextDeltaPayload); ok { + s.emit(ctx, EventAgentChunk, runID, sessionID, payload.Text) + if acc != nil { + acc.accumulateTextDelta(payload.Text) + } + } case provider.StreamEventToolCallStart: - s.emit(ctx, EventToolCallThinking, runID, sessionID, event.ToolName) + if payload, ok := event.Payload.(provider.ToolCallStartPayload); ok { + s.emit(ctx, EventToolCallThinking, runID, sessionID, payload.Name) + if acc != nil { + acc.accumulateToolCallStart(payload.Index, payload.ID, payload.Name) + } + } + case provider.StreamEventToolCallDelta: + if payload, ok := event.Payload.(provider.ToolCallDeltaPayload); ok { + if acc != nil { + acc.accumulateToolCallDelta(payload.Index, payload.ID, payload.ArgumentsDelta) + } + } } case <-ctx.Done(): return @@ -489,18 +567,22 @@ func isRetryableProviderError(err error) bool { } // callProviderWithRetry 在可重试的 ProviderError 上自动重试 provider.Chat() 调用。 -// 每次重试都会重新构建 provider 实例和流式事件转发管道。 +// 每次重试都会重新创建 provider 实例、流式事件转发管道和累积器。 // 非可重试错误、context 取消、重试耗尽时直接返回错误。 +// 返回值 acc 保存了本轮流式事件的完整累积状态，供调用方构建 assistant Message 使用。 func (s *Service) callProviderWithRetry( ctx context.Context, runID string, sessionID string, req provider.ChatRequest, -) (provider.ChatResponse, error) { +) (*streamAccumulator, error) { + acc := newStreamAccumulator() var lastErr error for retryAttempt := 0; retryAttempt <= defaultProviderRetryMax; retryAttempt++ { if retryAttempt > 0 { + // 重试时重置累积器，避免混入上轮数据 + acc = newStreamAccumulator() wait := providerRetryBackoff(retryAttempt) s.emit(ctx, EventProviderRetry, runID, sessionID, fmt.Sprintf("retrying provider call (attempt %d/%d, wait=%.1fs)...", @@ -508,45 +590,48 @@ func (s *Service) callProviderWithRetry( select { case <-ctx.Done(): - return provider.ChatResponse{}, ctx.Err() + return nil, ctx.Err() case <-time.After(wait): } } resolvedProvider, err := s.configManager.ResolvedSelectedProvider() if err != nil { - return provider.ChatResponse{}, err + return nil, err } modelProvider, err := s.providerFactory.Build(ctx, resolvedProvider) if err != nil { - return provider.ChatResponse{}, err + return nil, err } streamEvents := make(chan provider.StreamEvent, 32) streamDone := make(chan struct{}) - go s.forwardProviderEvents(ctx, runID, sessionID, streamEvents, streamDone) + go s.forwardProviderEvents(ctx, runID, sessionID, streamEvents, streamDone, acc) - resp, err := modelProvider.Chat(ctx, req, streamEvents) + err = modelProvider.Chat(ctx, req, streamEvents) close(streamEvents) <-streamDone if err == nil { - return resp, nil + return acc, nil } lastErr = err // 非可重试错误或 context 已取消，立即返回。 if !isRetryableProviderError(err) { - return provider.ChatResponse{}, err + return nil, lastErr } if ctx.Err() != nil { - return provider.ChatResponse{}, ctx.Err() + return nil, ctx.Err() } } - return provider.ChatResponse{}, lastErr + if lastErr == nil { + lastErr = errors.New("max retries exceeded") + } + return nil, fmt.Errorf("runtime: max retries exhausted, last error: %w", lastErr) } // providerRetryBackoff 计算指数退避 + 随机抖动的等待时间。 diff --git a/internal/runtime/runtime_test.go b/internal/runtime/runtime_test.go index ff24909e..101a91ad 100644 --- a/internal/runtime/runtime_test.go +++ b/internal/runtime/runtime_test.go @@ -90,14 +90,13 @@ func (s *memoryStore) ListSummaries(ctx context.Context) ([]SessionSummary, erro type scriptedProvider struct { name string - responses []provider.ChatResponse streams [][]provider.StreamEvent requests []provider.ChatRequest callCount int - chatFn func(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) (provider.ChatResponse, error) + chatFn func(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) error } -func (p *scriptedProvider) Chat(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) (provider.ChatResponse, error) { +func (p *scriptedProvider) Chat(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) error { p.requests = append(p.requests, cloneChatRequest(req)) callIndex := p.callCount @@ -112,15 +111,12 @@ func (p *scriptedProvider) Chat(ctx context.Context, req provider.ChatRequest, e select { case events <- event: case <-ctx.Done(): - return provider.ChatResponse{}, ctx.Err() + return ctx.Err() } } } - if callIndex >= len(p.responses) { - return provider.ChatResponse{}, fmt.Errorf("unexpected provider call %d", callIndex) - } - return p.responses[callIndex], nil + return nil } type scriptedProviderFactory struct { @@ -232,7 +228,6 @@ func TestServiceRun(t *testing.T) { tests := []struct { name string input UserInput - providerResponses []provider.ChatResponse providerStreams [][]provider.StreamEvent registerTool tools.Tool contextBuilder agentcontext.Builder @@ -245,19 +240,10 @@ func TestServiceRun(t *testing.T) { { name: "normal dialogue exits after final assistant reply", input: UserInput{RunID: "run-normal", Content: "hello"}, - providerResponses: []provider.ChatResponse{ - { - Message: provider.Message{ - Role: "assistant", - Content: "plain answer", - }, - FinishReason: "stop", - }, - }, providerStreams: [][]provider.StreamEvent{ { - {Type: provider.StreamEventTextDelta, Text: "plain "}, - {Type: provider.StreamEventTextDelta, Text: "answer"}, + provider.NewTextDeltaStreamEvent("plain "), + provider.NewTextDeltaStreamEvent("answer"), }, }, contextBuilder: &stubContextBuilder{ @@ -293,26 +279,15 @@ func TestServiceRun(t *testing.T) { { name: "tool call triggers execute and follow-up provider round", input: UserInput{RunID: "run-tool", Content: "edit file"}, - providerResponses: []provider.ChatResponse{ + // 第一轮：工具调用事件流（tool_call_start + tool_call_delta） + // 第二轮：普通文本回复 + providerStreams: [][]provider.StreamEvent{ { - Message: provider.Message{ - Role: "assistant", - ToolCalls: []provider.ToolCall{ - { - ID: "call-1", - Name: "filesystem_edit", - Arguments: `{"path":"main.go"}`, - }, - }, - }, - FinishReason: "tool_calls", + provider.NewToolCallStartStreamEvent(0, "call-1", "filesystem_edit"), + provider.NewToolCallDeltaStreamEvent(0, "call-1", `{"path":"main.go"}`), }, { - Message: provider.Message{ - Role: "assistant", - Content: "done", - }, - FinishReason: "stop", + provider.NewTextDeltaStreamEvent("done"), }, }, registerTool: &stubTool{ @@ -371,8 +346,7 @@ func TestServiceRun(t *testing.T) { } scripted := &scriptedProvider{ - responses: tt.providerResponses, - streams: tt.providerStreams, + streams: tt.providerStreams, } factory := &scriptedProviderFactory{provider: scripted} @@ -446,14 +420,8 @@ func TestServiceRunDelegatesToContextBuilder(t *testing.T) { } scripted := &scriptedProvider{ - responses: []provider.ChatResponse{ - { - Message: provider.Message{ - Role: "assistant", - Content: "done", - }, - FinishReason: "stop", - }, + streams: [][]provider.StreamEvent{ + {provider.NewTextDeltaStreamEvent("done")}, }, } @@ -501,10 +469,9 @@ func TestServiceRunPersistsSessionProviderAndModel(t *testing.T) { registry.Register(&stubTool{name: "filesystem_read_file", content: "default"}) scripted := &scriptedProvider{ - responses: []provider.ChatResponse{{ - Message: provider.Message{Role: provider.RoleAssistant, Content: "done"}, - FinishReason: "stop", - }}, + streams: [][]provider.StreamEvent{ + {provider.NewTextDeltaStreamEvent("done")}, + }, } service := NewWithFactory(manager, registry, store, &scriptedProviderFactory{provider: scripted}, nil) @@ -594,23 +561,12 @@ func TestServiceRunUsesToolManager(t *testing.T) { } scripted := &scriptedProvider{ - responses: []provider.ChatResponse{ + streams: [][]provider.StreamEvent{ { - Message: provider.Message{ - Role: "assistant", - ToolCalls: []provider.ToolCall{ - {ID: "call-manager", Name: "filesystem_edit", Arguments: `{"path":"main.go"}`}, - }, - }, - FinishReason: "tool_calls", - }, - { - Message: provider.Message{ - Role: "assistant", - Content: "done", - }, - FinishReason: "stop", + provider.NewToolCallStartStreamEvent(0, "call-manager", "filesystem_edit"), + provider.NewToolCallDeltaStreamEvent(0, "call-manager", `{"path":"main.go"}`), }, + {provider.NewTextDeltaStreamEvent("done")}, }, } @@ -672,20 +628,12 @@ func TestServiceRunEmitsPermissionRequestAndResolvedForAsk(t *testing.T) { } scripted := &scriptedProvider{ - responses: []provider.ChatResponse{ - { - Message: provider.Message{ - Role: "assistant", - ToolCalls: []provider.ToolCall{ - {ID: "call-ask", Name: "webfetch", Arguments: `{"url":"https://example.com/private"}`}, - }, - }, - FinishReason: "tool_calls", - }, + streams: [][]provider.StreamEvent{ { - Message: provider.Message{Role: "assistant", Content: "done"}, - FinishReason: "stop", + provider.NewToolCallStartStreamEvent(0, "call-ask", "webfetch"), + provider.NewToolCallDeltaStreamEvent(0, "call-ask", `{"url":"https://example.com/private"}`), }, + {provider.NewTextDeltaStreamEvent("done")}, }, } @@ -768,20 +716,12 @@ func TestServiceRunEmitsPermissionResolvedForDeny(t *testing.T) { } scripted := &scriptedProvider{ - responses: []provider.ChatResponse{ + streams: [][]provider.StreamEvent{ { - Message: provider.Message{ - Role: "assistant", - ToolCalls: []provider.ToolCall{ - {ID: "call-deny", Name: "bash", Arguments: `{"command":"echo hi"}`}, - }, - }, - FinishReason: "tool_calls", - }, - { - Message: provider.Message{Role: "assistant", Content: "done"}, - FinishReason: "stop", + provider.NewToolCallStartStreamEvent(0, "call-deny", "bash"), + provider.NewToolCallDeltaStreamEvent(0, "call-deny", `{"command":"echo hi"}`), }, + {provider.NewTextDeltaStreamEvent("done")}, }, } @@ -853,14 +793,8 @@ func TestServiceNewWithFactoryDefaultsToolManager(t *testing.T) { store := newMemoryStore() service := NewWithFactory(manager, nil, store, &scriptedProviderFactory{ provider: &scriptedProvider{ - responses: []provider.ChatResponse{ - { - Message: provider.Message{ - Role: provider.RoleAssistant, - Content: "done", - }, - FinishReason: "stop", - }, + streams: [][]provider.StreamEvent{ + {provider.NewTextDeltaStreamEvent("done")}, }, }, }, nil) @@ -902,15 +836,10 @@ func TestServiceRunErrorPaths(t *testing.T) { input: UserInput{RunID: "run-max-loops", Content: "loop"}, maxLoops: 1, provider: &scriptedProvider{ - responses: []provider.ChatResponse{ + streams: [][]provider.StreamEvent{ { - Message: provider.Message{ - Role: "assistant", - ToolCalls: []provider.ToolCall{ - {ID: "loop-call", Name: "filesystem_edit", Arguments: `{"path":"x"}`}, - }, - }, - FinishReason: "tool_calls", + provider.NewToolCallStartStreamEvent(0, "loop-call", "filesystem_edit"), + provider.NewToolCallDeltaStreamEvent(0, "loop-call", `{"path":"x"}`), }, }, }, @@ -946,14 +875,8 @@ func TestServiceRunErrorPaths(t *testing.T) { Content: "continue", }, provider: &scriptedProvider{ - responses: []provider.ChatResponse{ - { - Message: provider.Message{ - Role: "assistant", - Content: "resumed", - }, - FinishReason: "stop", - }, + streams: [][]provider.StreamEvent{ + {provider.NewTextDeltaStreamEvent("resumed")}, }, }, seedSession: &Session{ @@ -984,23 +907,18 @@ func TestServiceRunErrorPaths(t *testing.T) { callIdx := 0 return &scriptedProvider{ name: "retry-then-success", - chatFn: func(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) (provider.ChatResponse, error) { + chatFn: func(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) error { callIdx++ if callIdx == 1 { - return provider.ChatResponse{}, &provider.ProviderError{ + return &provider.ProviderError{ StatusCode: 500, Code: provider.ErrorCodeServer, Message: "internal server error", Retryable: true, } } - return provider.ChatResponse{ - Message: provider.Message{ - Role: "assistant", - Content: "recovered", - }, - FinishReason: "stop", - }, nil + events <- provider.NewTextDeltaStreamEvent("recovered") + return nil }, } }(), @@ -1024,8 +942,8 @@ 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 provider.ChatRequest, events chan<- provider.StreamEvent) (provider.ChatResponse, error) { - return provider.ChatResponse{}, &provider.ProviderError{ + chatFn: func(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) error { + return &provider.ProviderError{ StatusCode: 401, Code: provider.ErrorCodeAuthFailed, Message: "invalid api key", @@ -1047,8 +965,8 @@ func TestServiceRunErrorPaths(t *testing.T) { input: UserInput{RunID: "run-retry-exhausted", Content: "hello"}, provider: &scriptedProvider{ name: "always-500", - chatFn: func(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) (provider.ChatResponse, error) { - return provider.ChatResponse{}, &provider.ProviderError{ + chatFn: func(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) error { + return &provider.ProviderError{ StatusCode: 500, Code: provider.ErrorCodeServer, Message: "internal server error", @@ -1131,10 +1049,10 @@ func TestServiceCancelActiveRun(t *testing.T) { started := make(chan struct{}) scripted := &scriptedProvider{ name: "cancel-active-run-provider", - chatFn: func(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) (provider.ChatResponse, error) { + chatFn: func(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) error { close(started) <-ctx.Done() - return provider.ChatResponse{}, ctx.Err() + return ctx.Err() }, } @@ -1174,10 +1092,10 @@ func TestServiceRunCanceledByProvider(t *testing.T) { started := make(chan struct{}) scripted := &scriptedProvider{ name: "blocking-provider", - chatFn: func(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) (provider.ChatResponse, error) { + chatFn: func(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) error { close(started) <-ctx.Done() - return provider.ChatResponse{}, ctx.Err() + return ctx.Err() }, } @@ -1219,10 +1137,10 @@ 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 provider.ChatRequest, events chan<- provider.StreamEvent) (provider.ChatResponse, error) { + chatFn: func(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) error { close(started) <-ctx.Done() - return provider.ChatResponse{}, providerErr + return providerErr }, } @@ -1271,15 +1189,10 @@ func TestServiceRunCanceledDuringToolExecution(t *testing.T) { scripted := &scriptedProvider{ name: "tool-cancel-provider", - responses: []provider.ChatResponse{ + streams: [][]provider.StreamEvent{ { - Message: provider.Message{ - Role: "assistant", - ToolCalls: []provider.ToolCall{ - {ID: "cancel-call", Name: "filesystem_edit", Arguments: `{"path":"main.go"}`}, - }, - }, - FinishReason: "tool_calls", + provider.NewToolCallStartStreamEvent(0, "cancel-call", "filesystem_edit"), + provider.NewToolCallDeltaStreamEvent(0, "cancel-call", `{"path":"main.go"}`), }, }, } @@ -1336,15 +1249,10 @@ func TestServiceRunPreservesToolErrorAfterCancel(t *testing.T) { scripted := &scriptedProvider{ name: "tool-error-after-cancel-provider", - responses: []provider.ChatResponse{ + streams: [][]provider.StreamEvent{ { - Message: provider.Message{ - Role: "assistant", - ToolCalls: []provider.ToolCall{ - {ID: "tool-error-call", Name: "filesystem_edit", Arguments: `{"path":"main.go"}`}, - }, - }, - FinishReason: "tool_calls", + provider.NewToolCallStartStreamEvent(0, "tool-error-call", "filesystem_edit"), + provider.NewToolCallDeltaStreamEvent(0, "tool-error-call", `{"path":"main.go"}`), }, }, } @@ -1433,23 +1341,12 @@ func TestServiceRunToolTimeoutIsNotCancellation(t *testing.T) { scripted := &scriptedProvider{ name: "timeout-provider", - responses: []provider.ChatResponse{ + streams: [][]provider.StreamEvent{ { - Message: provider.Message{ - Role: "assistant", - ToolCalls: []provider.ToolCall{ - {ID: "timeout-call", Name: "filesystem_edit", Arguments: `{"path":"main.go"}`}, - }, - }, - FinishReason: "tool_calls", - }, - { - Message: provider.Message{ - Role: "assistant", - Content: "done after timeout", - }, - FinishReason: "stop", + provider.NewToolCallStartStreamEvent(0, "timeout-call", "filesystem_edit"), + provider.NewToolCallDeltaStreamEvent(0, "timeout-call", `{"path":"main.go"}`), }, + {provider.NewTextDeltaStreamEvent("done after timeout")}, }, } @@ -1603,28 +1500,25 @@ func TestServiceCompactUsesSessionProviderAndModelWhenPresent(t *testing.T) { registry.Register(&stubTool{name: "filesystem_read_file", content: "ok"}) scripted := &scriptedProvider{ - responses: []provider.ChatResponse{{ - Message: provider.Message{ - Role: provider.RoleAssistant, - Content: strings.Join([]string{ - "[compact_summary]", - "done:", - "- ok", - "", - "in_progress:", - "- continue", - "", - "decisions:", - "- kept existing provider and model", - "", - "code_changes:", - "- none", - "", - "constraints:", - "- none", - }, "\n"), - }, - }}, + streams: [][]provider.StreamEvent{ + {provider.NewTextDeltaStreamEvent(strings.Join([]string{ + "[compact_summary]", + "done:", + "- ok", + "", + "in_progress:", + "- continue", + "", + "decisions:", + "- kept existing provider and model", + "", + "code_changes:", + "- none", + "", + "constraints:", + "- none", + }, "\n"))}, + }, } factory := &scriptedProviderFactory{provider: scripted} service := NewWithFactory(manager, registry, store, factory, nil) @@ -1678,28 +1572,25 @@ func TestServiceCompactFallsBackToCurrentProviderWhenSessionMetadataMissing(t *t registry.Register(&stubTool{name: "filesystem_read_file", content: "ok"}) scripted := &scriptedProvider{ - responses: []provider.ChatResponse{{ - Message: provider.Message{ - Role: provider.RoleAssistant, - Content: strings.Join([]string{ - "[compact_summary]", - "done:", - "- ok", - "", - "in_progress:", - "- continue", - "", - "decisions:", - "- fallback to current selection", - "", - "code_changes:", - "- none", - "", - "constraints:", - "- none", - }, "\n"), - }, - }}, + streams: [][]provider.StreamEvent{ + {provider.NewTextDeltaStreamEvent(strings.Join([]string{ + "[compact_summary]", + "done:", + "- ok", + "", + "in_progress:", + "- continue", + "", + "decisions:", + "- fallback to current selection", + "", + "code_changes:", + "- none", + "", + "constraints:", + "- none", + }, "\n"))}, + }, } factory := &scriptedProviderFactory{provider: scripted} service := NewWithFactory(manager, registry, store, factory, nil) @@ -1738,20 +1629,12 @@ func TestServiceManualCompactThenRunContinuesToolRound(t *testing.T) { registry.Register(tool) scripted := &scriptedProvider{ - responses: []provider.ChatResponse{ - { - Message: provider.Message{ - Role: "assistant", - ToolCalls: []provider.ToolCall{ - {ID: "call-1", Name: "filesystem_read_file", Arguments: `{"path":"main.go"}`}, - }, - }, - FinishReason: "tool_calls", - }, + streams: [][]provider.StreamEvent{ { - Message: provider.Message{Role: "assistant", Content: "done"}, - FinishReason: "stop", + provider.NewToolCallStartStreamEvent(0, "call-1", "filesystem_read_file"), + provider.NewToolCallDeltaStreamEvent(0, "call-1", `{"path":"main.go"}`), }, + {provider.NewTextDeltaStreamEvent("done")}, }, } @@ -1826,17 +1709,15 @@ func TestServiceSerializesRunAndCompact(t *testing.T) { providerStarted := make(chan struct{}) unblockProvider := make(chan struct{}) scripted := &scriptedProvider{ - chatFn: func(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) (provider.ChatResponse, error) { + chatFn: func(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) error { select { case <-providerStarted: default: close(providerStarted) } <-unblockProvider - return provider.ChatResponse{ - Message: provider.Message{Role: provider.RoleAssistant, Content: "done"}, - FinishReason: "stop", - }, nil + events <- provider.NewTextDeltaStreamEvent("done") + return nil }, } @@ -1960,20 +1841,12 @@ func TestServiceRunUsesSessionWorkdirForContextAndTools(t *testing.T) { builder := &stubContextBuilder{} scripted := &scriptedProvider{ - responses: []provider.ChatResponse{ + streams: [][]provider.StreamEvent{ { - Message: provider.Message{ - Role: "assistant", - ToolCalls: []provider.ToolCall{ - {ID: "call-session-workdir", Name: "filesystem_edit", Arguments: `{"path":"main.go"}`}, - }, - }, - FinishReason: "tool_calls", - }, - { - Message: provider.Message{Role: "assistant", Content: "done"}, - FinishReason: "stop", + provider.NewToolCallStartStreamEvent(0, "call-session-workdir", "filesystem_edit"), + provider.NewToolCallDeltaStreamEvent(0, "call-session-workdir", `{"path":"main.go"}`), }, + {provider.NewTextDeltaStreamEvent("done")}, }, } @@ -2010,11 +1883,8 @@ func TestServiceRunUsesInputWorkdirForNewSession(t *testing.T) { registry.Register(&stubTool{name: "filesystem_read_file", content: "default"}) builder := &stubContextBuilder{} scripted := &scriptedProvider{ - responses: []provider.ChatResponse{ - { - Message: provider.Message{Role: "assistant", Content: "done"}, - FinishReason: "stop", - }, + streams: [][]provider.StreamEvent{ + {provider.NewTextDeltaStreamEvent("done")}, }, } diff --git a/internal/tui/update_test.go b/internal/tui/update_test.go index ed307cc0..9e28c76c 100644 --- a/internal/tui/update_test.go +++ b/internal/tui/update_test.go @@ -2603,8 +2603,8 @@ func newTestProviderService(t *testing.T, manager *config.Manager) *config.Selec type tUItestProvider struct{} -func (tUItestProvider) Chat(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) (provider.ChatResponse, error) { - return provider.ChatResponse{}, nil +func (tUItestProvider) Chat(ctx context.Context, req provider.ChatRequest, events chan<- provider.StreamEvent) error { + return nil } type tUItestCatalogStore struct { From 7b80117eb58f01b8dffbd87173c33a4d5d34b38d Mon Sep 17 00:00:00 2001 From: phantom5099 <1011668688@qq.com> Date: Sun, 5 Apr 2026 14:57:23 +0800 Subject: [PATCH 3/4] =?UTF-8?q?test(provider):=E8=A1=A5=E9=BD=90=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E8=A6=86=E7=9B=96=E7=8E=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/provider/openai/openai_test.go | 61 +++++++ internal/provider/types_test.go | 229 ++++++++++++++++++++++++ 2 files changed, 290 insertions(+) create mode 100644 internal/provider/types_test.go diff --git a/internal/provider/openai/openai_test.go b/internal/provider/openai/openai_test.go index 9b0fc108..ae0b6b9d 100644 --- a/internal/provider/openai/openai_test.go +++ b/internal/provider/openai/openai_test.go @@ -43,6 +43,67 @@ func TestWithTransport(t *testing.T) { } } +func TestNewValidationErrors(t *testing.T) { + t.Parallel() + + t.Run("empty api key returns error", func(t *testing.T) { + t.Parallel() + cfg := resolvedConfig("", "") + cfg.APIKey = "" + _, err := New(cfg) + if err == nil { + t.Fatal("expected error for empty api key") + } + if !strings.Contains(err.Error(), "api key is empty") { + t.Fatalf("expected api key error, got: %v", err) + } + }) + + t.Run("whitespace-only api key returns error", func(t *testing.T) { + t.Parallel() + cfg := resolvedConfig("", "") + cfg.APIKey = " " + _, err := New(cfg) + if err == nil { + t.Fatal("expected error for whitespace-only api key") + } + }) + + t.Run("invalid config validate fails", func(t *testing.T) { + t.Parallel() + // 空字符串的 BaseURL 和 Model 会导致 Validate 失败（取决于 config 实现） + cfg := config.ResolvedProviderConfig{ + ProviderConfig: config.ProviderConfig{ + Driver: DriverName, + BaseURL: "", + Model: "", + APIKeyEnv: "NONEXISTENT_ENV_VAR_" + t.Name(), + }, + APIKey: "test-key", + } + _, err := New(cfg) + // 验证失败时应该返回错误 + if err != nil { + // 预期行为：config 校验不通过 + return + } + // 如果校验通过了，也接受（取决于具体实现） + }) +} + +func TestNewDefaultTransportWhenNoOption(t *testing.T) { + t.Parallel() + + cfg := resolvedConfig("", "") + provider, err := New(cfg) // 不传任何 buildOption + if err != nil { + t.Fatalf("New() error = %v", err) + } + if provider.client.Transport == nil { + t.Fatal("expected default transport to be set") + } +} + func TestDefaultRetryTransport(t *testing.T) { t.Parallel() diff --git a/internal/provider/types_test.go b/internal/provider/types_test.go new file mode 100644 index 00000000..e5f1c5aa --- /dev/null +++ b/internal/provider/types_test.go @@ -0,0 +1,229 @@ +package provider + +import "testing" + +// --- Role 常量 --- + +func TestRoleConstants(t *testing.T) { + tests := []struct { + name string + got string + expect string + }{ + {"system", RoleSystem, "system"}, + {"user", RoleUser, "user"}, + {"assistant", RoleAssistant, "assistant"}, + {"tool", RoleTool, "tool"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if tt.got != tt.expect { + t.Fatalf("expected %q, got %q", tt.expect, tt.got) + } + }) + } +} + +// --- StreamEventType 常量 --- + +func TestStreamEventConstants(t *testing.T) { + tests := []struct { + name string + got StreamEventType + expect string + }{ + {"text_delta", StreamEventTextDelta, "text_delta"}, + {"tool_call_start", StreamEventToolCallStart, "tool_call_start"}, + {"tool_call_delta", StreamEventToolCallDelta, "tool_call_delta"}, + {"message_done", StreamEventMessageDone, "message_done"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if string(tt.got) != tt.expect { + t.Fatalf("expected %q, got %q", tt.expect, string(tt.got)) + } + }) + } +} + +// --- NewTextDeltaStreamEvent --- + +func TestNewTextDeltaStreamEvent(t *testing.T) { + t.Parallel() + + event := NewTextDeltaStreamEvent("hello") + if event.Type != StreamEventTextDelta { + t.Fatalf("expected type %q, got %q", StreamEventTextDelta, event.Type) + } + + payload, ok := event.Payload.(TextDeltaPayload) + if !ok { + t.Fatal("expected TextDeltaPayload type") + } + if payload.Text != "hello" { + t.Fatalf("expected text %q, got %q", "hello", payload.Text) + } +} + +// --- NewToolCallStartStreamEvent --- + +func TestNewToolCallStartStreamEvent(t *testing.T) { + t.Parallel() + + event := NewToolCallStartStreamEvent(3, "call_1", "edit_file") + if event.Type != StreamEventToolCallStart { + t.Fatalf("expected type %q, got %q", StreamEventToolCallStart, event.Type) + } + + payload, ok := event.Payload.(ToolCallStartPayload) + if !ok { + t.Fatal("expected ToolCallStartPayload type") + } + if payload.Index != 3 || payload.ID != "call_1" || payload.Name != "edit_file" { + t.Fatalf("unexpected payload: %+v", payload) + } +} + +// --- NewToolCallDeltaStreamEvent --- + +func TestNewToolCallDeltaStreamEvent(t *testing.T) { + t.Parallel() + + event := NewToolCallDeltaStreamEvent(1, "call_2", `{"path":"main.go"}`) + if event.Type != StreamEventToolCallDelta { + t.Fatalf("expected type %q, got %q", StreamEventToolCallDelta, event.Type) + } + + payload, ok := event.Payload.(ToolCallDeltaPayload) + if !ok { + t.Fatal("expected ToolCallDeltaPayload type") + } + if payload.Index != 1 || payload.ID != "call_2" || payload.ArgumentsDelta != `{"path":"main.go"}` { + t.Fatalf("unexpected payload: %+v", payload) + } +} + +// --- NewMessageDoneStreamEvent --- + +func TestNewMessageDoneStreamEvent(t *testing.T) { + t.Parallel() + + t.Run("with usage", func(t *testing.T) { + usage := &Usage{TotalTokens: 42} + event := NewMessageDoneStreamEvent("stop", usage) + + if event.Type != StreamEventMessageDone { + t.Fatalf("expected type %q, got %q", StreamEventMessageDone, event.Type) + } + + payload, ok := event.Payload.(MessageDonePayload) + if !ok { + t.Fatal("expected MessageDonePayload type") + } + if payload.FinishReason != "stop" { + t.Fatalf("expected finish reason %q, got %q", "stop", payload.FinishReason) + } + if payload.Usage == nil || payload.Usage.TotalTokens != 42 { + t.Fatalf("unexpected usage: %+v", payload.Usage) + } + }) + + t.Run("nil usage", func(t *testing.T) { + event := NewMessageDoneStreamEvent("tool_calls", nil) + + payload, ok := event.Payload.(MessageDonePayload) + if !ok { + t.Fatal("expected MessageDonePayload type") + } + if payload.FinishReason != "tool_calls" { + t.Fatalf("expected finish reason %q, got %q", "tool_calls", payload.FinishReason) + } + if payload.Usage != nil { + t.Fatal("expected nil usage") + } + }) + + t.Run("empty finish reason", func(t *testing.T) { + event := NewMessageDoneStreamEvent("", nil) + + payload, ok := event.Payload.(MessageDonePayload) + if !ok { + t.Fatal("expected MessageDonePayload type") + } + if payload.FinishReason != "" { + t.Fatalf("expected empty finish reason, got %q", payload.FinishReason) + } + }) +} + +// --- 结构体字段覆盖验证 --- + +func TestMessageStructFields(t *testing.T) { + t.Parallel() + + msg := Message{ + Role: RoleUser, + Content: "hello", + ToolCalls: []ToolCall{{ID: "t1"}}, + ToolCallID: "tc_1", + IsError: true, + } + if msg.Role != RoleUser || msg.Content != "hello" || len(msg.ToolCalls) != 1 || + msg.ToolCallID != "tc_1" || !msg.IsError { + t.Fatalf("message fields not as expected: %+v", msg) + } +} + +func TestToolCallStructFields(t *testing.T) { + t.Parallel() + + tc := ToolCall{ID: "c1", Name: "fn", Arguments: "{}"} + if tc.ID != "c1" || tc.Name != "fn" || tc.Arguments != "{}" { + t.Fatalf("tool call fields not as expected: %+v", tc) + } +} + +func TestToolSpecStructFields(t *testing.T) { + t.Parallel() + + spec := ToolSpec{Name: "read", Description: "read file", Schema: map[string]any{"type": "object"}} + if spec.Name != "read" || spec.Description != "read file" || spec.Schema == nil { + t.Fatalf("tool spec fields not as expected: %+v", spec) + } +} + +func TestChatRequestStructFields(t *testing.T) { + t.Parallel() + + req := ChatRequest{ + Model: "gpt-4", + SystemPrompt: "you are helpful", + Messages: []Message{{Role: RoleUser}}, + Tools: []ToolSpec{{Name: "bash"}}, + } + if req.Model != "gpt-4" || req.SystemPrompt != "you are helpful" || + len(req.Messages) != 1 || len(req.Tools) != 1 { + t.Fatalf("chat request fields not as expected: %+v", req) + } +} + +func TestUsageStructFields(t *testing.T) { + t.Parallel() + + usage := Usage{InputTokens: 10, OutputTokens: 20, TotalTokens: 30} + if usage.InputTokens != 10 || usage.OutputTokens != 20 || usage.TotalTokens != 30 { + t.Fatalf("usage fields not as expected: %+v", usage) + } +} + +func TestStreamEventStructFields(t *testing.T) { + t.Parallel() + + event := StreamEvent{ + Type: StreamEventTextDelta, + Payload: TextDeltaPayload{Text: "hi"}, + } + if event.Type != StreamEventTextDelta { + t.Fatalf("event type not as expected: %s", event.Type) + } +} From 4627929a8ceeba7152eaab1d99f06633cccee490 Mon Sep 17 00:00:00 2001 From: phantom5099 <1011668688@qq.com> Date: Sun, 5 Apr 2026 15:48:15 +0800 Subject: [PATCH 4/4] =?UTF-8?q?fix(provider):=E4=BF=AE=E5=A4=8D=E2=80=9C?= =?UTF-8?q?=E6=98=BE=E5=BC=8F=E6=8A=A5=E9=94=99=E2=80=9D=E6=B5=81=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E5=A4=84=E7=90=86=E5=AF=BC=E8=87=B4=E5=8D=A1=E6=AD=BB?= =?UTF-8?q?=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/provider/openai/openai_test.go | 24 ++-- internal/provider/types.go | 69 ++++++++-- internal/provider/types_test.go | 80 ++++++++--- internal/runtime/compact_generator.go | 34 ++--- internal/runtime/compact_generator_test.go | 65 +++++++++ internal/runtime/runtime.go | 150 ++++++++++++++++----- internal/runtime/runtime_test.go | 136 +++++++++++++++++++ 7 files changed, 461 insertions(+), 97 deletions(-) diff --git a/internal/provider/openai/openai_test.go b/internal/provider/openai/openai_test.go index ae0b6b9d..29e99654 100644 --- a/internal/provider/openai/openai_test.go +++ b/internal/provider/openai/openai_test.go @@ -637,36 +637,36 @@ func drainStreamEvents(events <-chan domain.StreamEvent) []domain.StreamEvent { func requireTextDeltaPayload(t *testing.T, event domain.StreamEvent) domain.TextDeltaPayload { t.Helper() - payload, ok := event.Payload.(domain.TextDeltaPayload) - if !ok { - t.Fatalf("expected TextDeltaPayload, got %T", event.Payload) + payload, err := event.TextDeltaValue() + if err != nil { + t.Fatalf("TextDeltaValue() error = %v", err) } return payload } func requireToolCallStartPayload(t *testing.T, event domain.StreamEvent) domain.ToolCallStartPayload { t.Helper() - payload, ok := event.Payload.(domain.ToolCallStartPayload) - if !ok { - t.Fatalf("expected ToolCallStartPayload, got %T", event.Payload) + payload, err := event.ToolCallStartValue() + if err != nil { + t.Fatalf("ToolCallStartValue() error = %v", err) } return payload } func requireToolCallDeltaPayload(t *testing.T, event domain.StreamEvent) domain.ToolCallDeltaPayload { t.Helper() - payload, ok := event.Payload.(domain.ToolCallDeltaPayload) - if !ok { - t.Fatalf("expected ToolCallDeltaPayload, got %T", event.Payload) + payload, err := event.ToolCallDeltaValue() + if err != nil { + t.Fatalf("ToolCallDeltaValue() error = %v", err) } return payload } func requireMessageDonePayload(t *testing.T, event domain.StreamEvent) domain.MessageDonePayload { t.Helper() - payload, ok := event.Payload.(domain.MessageDonePayload) - if !ok { - t.Fatalf("expected MessageDonePayload, got %T", event.Payload) + payload, err := event.MessageDoneValue() + if err != nil { + t.Fatalf("MessageDoneValue() error = %v", err) } return payload } diff --git a/internal/provider/types.go b/internal/provider/types.go index 278a3a93..1f689c12 100644 --- a/internal/provider/types.go +++ b/internal/provider/types.go @@ -1,5 +1,7 @@ package provider +import "fmt" + const ( // RoleSystem 标识系统消息。 RoleSystem = "system" @@ -65,8 +67,11 @@ const ( // StreamEvent 表示 provider 向 runtime 推送的流式事件。 type StreamEvent struct { - Type StreamEventType `json:"type"` - Payload interface{} `json:"payload,omitempty"` + Type StreamEventType `json:"type"` + TextDelta *TextDeltaPayload `json:"text_delta,omitempty"` + ToolCallStart *ToolCallStartPayload `json:"tool_call_start,omitempty"` + ToolCallDelta *ToolCallDeltaPayload `json:"tool_call_delta,omitempty"` + MessageDone *MessageDonePayload `json:"message_done,omitempty"` } // TextDeltaPayload 表示文本增量事件的载荷。 @@ -97,31 +102,75 @@ type MessageDonePayload struct { // NewTextDeltaStreamEvent 创建文本增量流事件。 func NewTextDeltaStreamEvent(text string) StreamEvent { return StreamEvent{ - Type: StreamEventTextDelta, - Payload: TextDeltaPayload{Text: text}, + Type: StreamEventTextDelta, + TextDelta: &TextDeltaPayload{Text: text}, } } // NewToolCallStartStreamEvent 创建工具调用开始流事件。 func NewToolCallStartStreamEvent(index int, id, name string) StreamEvent { return StreamEvent{ - Type: StreamEventToolCallStart, - Payload: ToolCallStartPayload{Index: index, ID: id, Name: name}, + Type: StreamEventToolCallStart, + ToolCallStart: &ToolCallStartPayload{Index: index, ID: id, Name: name}, } } // NewToolCallDeltaStreamEvent 创建工具调用参数增量流事件。 func NewToolCallDeltaStreamEvent(index int, id, argumentsDelta string) StreamEvent { return StreamEvent{ - Type: StreamEventToolCallDelta, - Payload: ToolCallDeltaPayload{Index: index, ID: id, ArgumentsDelta: argumentsDelta}, + Type: StreamEventToolCallDelta, + ToolCallDelta: &ToolCallDeltaPayload{Index: index, ID: id, ArgumentsDelta: argumentsDelta}, } } // NewMessageDoneStreamEvent 创建消息完成流事件。 func NewMessageDoneStreamEvent(finishReason string, usage *Usage) StreamEvent { return StreamEvent{ - Type: StreamEventMessageDone, - Payload: MessageDonePayload{FinishReason: finishReason, Usage: usage}, + Type: StreamEventMessageDone, + MessageDone: &MessageDonePayload{FinishReason: finishReason, Usage: usage}, + } +} + +// TextDeltaValue 返回 text_delta 事件的载荷，并在结构缺失时显式报错。 +func (e StreamEvent) TextDeltaValue() (TextDeltaPayload, error) { + if e.Type != StreamEventTextDelta { + return TextDeltaPayload{}, fmt.Errorf("provider: stream event type %q is not text_delta", e.Type) + } + if e.TextDelta == nil { + return TextDeltaPayload{}, fmt.Errorf("provider: text_delta event payload is nil") + } + return *e.TextDelta, nil +} + +// ToolCallStartValue 返回 tool_call_start 事件的载荷，并在结构缺失时显式报错。 +func (e StreamEvent) ToolCallStartValue() (ToolCallStartPayload, error) { + if e.Type != StreamEventToolCallStart { + return ToolCallStartPayload{}, fmt.Errorf("provider: stream event type %q is not tool_call_start", e.Type) + } + if e.ToolCallStart == nil { + return ToolCallStartPayload{}, fmt.Errorf("provider: tool_call_start event payload is nil") + } + return *e.ToolCallStart, nil +} + +// ToolCallDeltaValue 返回 tool_call_delta 事件的载荷，并在结构缺失时显式报错。 +func (e StreamEvent) ToolCallDeltaValue() (ToolCallDeltaPayload, error) { + if e.Type != StreamEventToolCallDelta { + return ToolCallDeltaPayload{}, fmt.Errorf("provider: stream event type %q is not tool_call_delta", e.Type) + } + if e.ToolCallDelta == nil { + return ToolCallDeltaPayload{}, fmt.Errorf("provider: tool_call_delta event payload is nil") + } + return *e.ToolCallDelta, nil +} + +// MessageDoneValue 返回 message_done 事件的载荷，并在结构缺失时显式报错。 +func (e StreamEvent) MessageDoneValue() (MessageDonePayload, error) { + if e.Type != StreamEventMessageDone { + return MessageDonePayload{}, fmt.Errorf("provider: stream event type %q is not message_done", e.Type) + } + if e.MessageDone == nil { + return MessageDonePayload{}, fmt.Errorf("provider: message_done event payload is nil") } + return *e.MessageDone, nil } diff --git a/internal/provider/types_test.go b/internal/provider/types_test.go index e5f1c5aa..5c822ee8 100644 --- a/internal/provider/types_test.go +++ b/internal/provider/types_test.go @@ -1,6 +1,9 @@ package provider -import "testing" +import ( + "encoding/json" + "testing" +) // --- Role 常量 --- @@ -56,9 +59,9 @@ func TestNewTextDeltaStreamEvent(t *testing.T) { t.Fatalf("expected type %q, got %q", StreamEventTextDelta, event.Type) } - payload, ok := event.Payload.(TextDeltaPayload) - if !ok { - t.Fatal("expected TextDeltaPayload type") + payload, err := event.TextDeltaValue() + if err != nil { + t.Fatalf("TextDeltaValue() error = %v", err) } if payload.Text != "hello" { t.Fatalf("expected text %q, got %q", "hello", payload.Text) @@ -75,9 +78,9 @@ func TestNewToolCallStartStreamEvent(t *testing.T) { t.Fatalf("expected type %q, got %q", StreamEventToolCallStart, event.Type) } - payload, ok := event.Payload.(ToolCallStartPayload) - if !ok { - t.Fatal("expected ToolCallStartPayload type") + payload, err := event.ToolCallStartValue() + if err != nil { + t.Fatalf("ToolCallStartValue() error = %v", err) } if payload.Index != 3 || payload.ID != "call_1" || payload.Name != "edit_file" { t.Fatalf("unexpected payload: %+v", payload) @@ -94,9 +97,9 @@ func TestNewToolCallDeltaStreamEvent(t *testing.T) { t.Fatalf("expected type %q, got %q", StreamEventToolCallDelta, event.Type) } - payload, ok := event.Payload.(ToolCallDeltaPayload) - if !ok { - t.Fatal("expected ToolCallDeltaPayload type") + payload, err := event.ToolCallDeltaValue() + if err != nil { + t.Fatalf("ToolCallDeltaValue() error = %v", err) } if payload.Index != 1 || payload.ID != "call_2" || payload.ArgumentsDelta != `{"path":"main.go"}` { t.Fatalf("unexpected payload: %+v", payload) @@ -116,9 +119,9 @@ func TestNewMessageDoneStreamEvent(t *testing.T) { t.Fatalf("expected type %q, got %q", StreamEventMessageDone, event.Type) } - payload, ok := event.Payload.(MessageDonePayload) - if !ok { - t.Fatal("expected MessageDonePayload type") + payload, err := event.MessageDoneValue() + if err != nil { + t.Fatalf("MessageDoneValue() error = %v", err) } if payload.FinishReason != "stop" { t.Fatalf("expected finish reason %q, got %q", "stop", payload.FinishReason) @@ -131,9 +134,9 @@ func TestNewMessageDoneStreamEvent(t *testing.T) { t.Run("nil usage", func(t *testing.T) { event := NewMessageDoneStreamEvent("tool_calls", nil) - payload, ok := event.Payload.(MessageDonePayload) - if !ok { - t.Fatal("expected MessageDonePayload type") + payload, err := event.MessageDoneValue() + if err != nil { + t.Fatalf("MessageDoneValue() error = %v", err) } if payload.FinishReason != "tool_calls" { t.Fatalf("expected finish reason %q, got %q", "tool_calls", payload.FinishReason) @@ -146,9 +149,9 @@ func TestNewMessageDoneStreamEvent(t *testing.T) { t.Run("empty finish reason", func(t *testing.T) { event := NewMessageDoneStreamEvent("", nil) - payload, ok := event.Payload.(MessageDonePayload) - if !ok { - t.Fatal("expected MessageDonePayload type") + payload, err := event.MessageDoneValue() + if err != nil { + t.Fatalf("MessageDoneValue() error = %v", err) } if payload.FinishReason != "" { t.Fatalf("expected empty finish reason, got %q", payload.FinishReason) @@ -220,10 +223,45 @@ func TestStreamEventStructFields(t *testing.T) { t.Parallel() event := StreamEvent{ - Type: StreamEventTextDelta, - Payload: TextDeltaPayload{Text: "hi"}, + Type: StreamEventTextDelta, + TextDelta: &TextDeltaPayload{Text: "hi"}, } if event.Type != StreamEventTextDelta { t.Fatalf("event type not as expected: %s", event.Type) } + if event.TextDelta == nil || event.TextDelta.Text != "hi" { + t.Fatalf("event text_delta not as expected: %+v", event.TextDelta) + } +} + +func TestStreamEventJSONRoundTrip(t *testing.T) { + t.Parallel() + + original := NewToolCallDeltaStreamEvent(2, "call-7", `{"path":"main.go"}`) + data, err := json.Marshal(original) + if err != nil { + t.Fatalf("json.Marshal() error = %v", err) + } + + var decoded StreamEvent + if err := json.Unmarshal(data, &decoded); err != nil { + t.Fatalf("json.Unmarshal() error = %v", err) + } + + payload, err := decoded.ToolCallDeltaValue() + if err != nil { + t.Fatalf("ToolCallDeltaValue() error = %v", err) + } + if payload.Index != 2 || payload.ID != "call-7" || payload.ArgumentsDelta != `{"path":"main.go"}` { + t.Fatalf("unexpected round-trip payload: %+v", payload) + } +} + +func TestStreamEventValueAccessorsRejectMissingPayload(t *testing.T) { + t.Parallel() + + event := StreamEvent{Type: StreamEventTextDelta} + if _, err := event.TextDeltaValue(); err == nil { + t.Fatal("expected TextDeltaValue() to reject missing payload") + } } diff --git a/internal/runtime/compact_generator.go b/internal/runtime/compact_generator.go index 49e48755..20be11a1 100644 --- a/internal/runtime/compact_generator.go +++ b/internal/runtime/compact_generator.go @@ -59,30 +59,24 @@ func (g *compactSummaryGenerator) Generate(ctx context.Context, input contextcom // 使用流式事件通道收集 compact 摘要响应。 streamEvents := make(chan provider.StreamEvent, 32) - streamDone := make(chan struct{}) + streamDone := make(chan error, 1) acc := newStreamAccumulator() go func() { - defer close(streamDone) + var streamErr error + defer func() { + streamDone <- streamErr + }() + for { select { case event, ok := <-streamEvents: if !ok { return } - switch event.Type { - case provider.StreamEventTextDelta: - if payload, ok := event.Payload.(provider.TextDeltaPayload); ok { - acc.accumulateTextDelta(payload.Text) - } - case provider.StreamEventToolCallStart: - if payload, ok := event.Payload.(provider.ToolCallStartPayload); ok { - acc.accumulateToolCallStart(payload.Index, payload.ID, payload.Name) - } - case provider.StreamEventToolCallDelta: - if payload, ok := event.Payload.(provider.ToolCallDeltaPayload); ok { - acc.accumulateToolCallDelta(payload.Index, payload.ID, payload.ArgumentsDelta) - } + if err := handleProviderStreamEvent(event, acc, nil, nil); err != nil && streamErr == nil { + // 记录首个协议错误后继续排空事件通道，避免 provider 在后续发送时阻塞。 + streamErr = err } case <-ctx.Done(): return @@ -99,13 +93,19 @@ func (g *compactSummaryGenerator) Generate(ctx context.Context, input contextcom }}, }, streamEvents) close(streamEvents) - <-streamDone + streamErr := <-streamDone if err != nil { return "", err } + if streamErr != nil { + return "", streamErr + } - message := acc.buildMessage() + message, err := acc.buildMessage() + if err != nil { + return "", err + } if len(message.ToolCalls) > 0 { return "", errors.New("runtime: compact summary response must not contain tool calls") } diff --git a/internal/runtime/compact_generator_test.go b/internal/runtime/compact_generator_test.go index e0c348e7..e82c625f 100644 --- a/internal/runtime/compact_generator_test.go +++ b/internal/runtime/compact_generator_test.go @@ -4,6 +4,7 @@ import ( "context" "strings" "testing" + "time" "neo-code/internal/config" contextcompact "neo-code/internal/context/compact" @@ -133,3 +134,67 @@ func TestCompactSummaryGeneratorRejectsToolCalls(t *testing.T) { t.Fatalf("expected tool call rejection, got %v", err) } } + +func TestCompactSummaryGeneratorRejectsMalformedStreamEvent(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: [][]provider.StreamEvent{ + { + {Type: provider.StreamEventTextDelta}, + }, + }, + } + generator := newCompactSummaryGenerator(&scriptedProviderFactory{provider: scripted}, 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(), "text_delta event payload is nil") { + t.Fatalf("expected malformed stream event rejection, got %v", err) + } +} + +func TestCompactSummaryGeneratorMalformedStreamEventDoesNotDeadlock(t *testing.T) { + t.Parallel() + + manager := newRuntimeConfigManager(t) + resolvedProvider, err := resolvedProviderForTests(manager.Get(), config.OpenAIName) + if err != nil { + t.Fatalf("resolve provider: %v", err) + } + + stream := []provider.StreamEvent{{Type: provider.StreamEventTextDelta}} + for i := 0; i < 40; i++ { + stream = append(stream, provider.NewTextDeltaStreamEvent("ignored")) + } + scripted := &scriptedProvider{ + streams: [][]provider.StreamEvent{stream}, + } + generator := newCompactSummaryGenerator(&scriptedProviderFactory{provider: scripted}, resolvedProvider, "session-model") + + errCh := make(chan error, 1) + go func() { + _, genErr := generator.Generate(context.Background(), contextcompact.SummaryInput{ + Mode: contextcompact.ModeManual, + Config: manager.Get().Context.Compact, + }) + errCh <- genErr + }() + + select { + case genErr := <-errCh: + if genErr == nil || !strings.Contains(genErr.Error(), "text_delta event payload is nil") { + t.Fatalf("expected malformed stream event rejection, got %v", genErr) + } + case <-time.After(2 * time.Second): + t.Fatal("expected compact generation to fail instead of deadlocking on malformed stream event") + } +} diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index d3b3e29f..03d6d556 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -49,28 +49,38 @@ func (a *streamAccumulator) accumulateTextDelta(text string) { a.content.WriteString(text) } +// ensureToolCall 返回指定索引的工具调用条目，不存在时会先创建占位对象。 +func (a *streamAccumulator) ensureToolCall(index int) *provider.ToolCall { + call, exists := a.toolCalls[index] + if !exists { + call = &provider.ToolCall{} + a.toolCalls[index] = call + } + return call +} + // accumulateToolCallStart 记录新发现的工具调用（首次出现时创建条目）。 func (a *streamAccumulator) accumulateToolCallStart(index int, id, name string) { - if _, exists := a.toolCalls[index]; !exists { - a.toolCalls[index] = &provider.ToolCall{ID: id, Name: name} + call := a.ensureToolCall(index) + if strings.TrimSpace(id) != "" { + call.ID = id + } + if strings.TrimSpace(name) != "" { + call.Name = name } } // accumulateToolCallDelta 累积工具调用参数增量。 func (a *streamAccumulator) accumulateToolCallDelta(index int, id, argumentsDelta string) { - call, exists := a.toolCalls[index] - if !exists { - call = &provider.ToolCall{ID: id} - a.toolCalls[index] = call - } - if name := call.Name; strings.TrimSpace(name) == "" && call.ID != "" { - // 首次出现 delta 时可能还未收到 start 事件，仅记录 ID + call := a.ensureToolCall(index) + if strings.TrimSpace(id) != "" { + call.ID = id } call.Arguments += argumentsDelta } -// buildMessage 从累积状态构建最终的 assistant Message 对象。 -func (a *streamAccumulator) buildMessage() provider.Message { +// buildMessage 从累积状态构建最终的 assistant Message 对象，并校验工具调用元数据是否完整。 +func (a *streamAccumulator) buildMessage() (provider.Message, error) { ordered := make([]int, 0, len(a.toolCalls)) for index := range a.toolCalls { ordered = append(ordered, index) @@ -86,9 +96,15 @@ func (a *streamAccumulator) buildMessage() provider.Message { if call == nil { continue } + if strings.TrimSpace(call.ID) == "" { + return provider.Message{}, fmt.Errorf("runtime: provider emitted tool call %d without id", index) + } + if strings.TrimSpace(call.Name) == "" { + return provider.Message{}, fmt.Errorf("runtime: provider emitted tool call %d without name", index) + } message.ToolCalls = append(message.ToolCalls, *call) } - return message + return message, nil } var runtimeSessionWorkdirs = struct { @@ -248,7 +264,10 @@ func (s *Service) Run(ctx context.Context, input UserInput) error { session.Provider = cfg.SelectedProvider session.Model = cfg.CurrentModel - assistant := acc.buildMessage() + assistant, err := acc.buildMessage() + if err != nil { + return s.handleRunError(ctx, input.RunID, session.ID, err) + } if strings.TrimSpace(assistant.Role) == "" { assistant.Role = provider.RoleAssistant } @@ -479,37 +498,88 @@ func (s *Service) emit(ctx context.Context, kind EventType, runID string, sessio } } +// handleProviderStreamEvent 解析并应用单条 provider 流式事件，缺失载荷或未知类型时返回错误。 +func handleProviderStreamEvent( + event provider.StreamEvent, + acc *streamAccumulator, + onTextDelta func(string), + onToolCallStart func(provider.ToolCallStartPayload), +) error { + switch event.Type { + case provider.StreamEventTextDelta: + payload, err := event.TextDeltaValue() + if err != nil { + return err + } + if onTextDelta != nil { + onTextDelta(payload.Text) + } + if acc != nil { + acc.accumulateTextDelta(payload.Text) + } + case provider.StreamEventToolCallStart: + payload, err := event.ToolCallStartValue() + if err != nil { + return err + } + if onToolCallStart != nil { + onToolCallStart(payload) + } + if acc != nil { + acc.accumulateToolCallStart(payload.Index, payload.ID, payload.Name) + } + case provider.StreamEventToolCallDelta: + payload, err := event.ToolCallDeltaValue() + if err != nil { + return err + } + if acc != nil { + acc.accumulateToolCallDelta(payload.Index, payload.ID, payload.ArgumentsDelta) + } + case provider.StreamEventMessageDone: + if _, err := event.MessageDoneValue(); err != nil { + return err + } + default: + return fmt.Errorf("runtime: unsupported provider stream event type %q", event.Type) + } + return nil +} + // forwardProviderEvents 将 provider 流式事件转发为 runtime 事件，同时向 accumulator 累积消息状态。 // 使用 select 同时监听输入通道和 context 取消信号，确保 goroutine 不会因通道阻塞而泄漏。 -func (s *Service) forwardProviderEvents(ctx context.Context, runID string, sessionID string, input <-chan provider.StreamEvent, done chan<- struct{}, acc *streamAccumulator) { - defer close(done) +func (s *Service) forwardProviderEvents( + ctx context.Context, + runID string, + sessionID string, + input <-chan provider.StreamEvent, + done chan<- error, + acc *streamAccumulator, +) { + var forwardErr error + defer func() { + done <- forwardErr + }() + for { select { case event, ok := <-input: if !ok { return } - switch event.Type { - case provider.StreamEventTextDelta: - if payload, ok := event.Payload.(provider.TextDeltaPayload); ok { - s.emit(ctx, EventAgentChunk, runID, sessionID, payload.Text) - if acc != nil { - acc.accumulateTextDelta(payload.Text) - } - } - case provider.StreamEventToolCallStart: - if payload, ok := event.Payload.(provider.ToolCallStartPayload); ok { + err := handleProviderStreamEvent( + event, + acc, + func(text string) { + s.emit(ctx, EventAgentChunk, runID, sessionID, text) + }, + func(payload provider.ToolCallStartPayload) { s.emit(ctx, EventToolCallThinking, runID, sessionID, payload.Name) - if acc != nil { - acc.accumulateToolCallStart(payload.Index, payload.ID, payload.Name) - } - } - case provider.StreamEventToolCallDelta: - if payload, ok := event.Payload.(provider.ToolCallDeltaPayload); ok { - if acc != nil { - acc.accumulateToolCallDelta(payload.Index, payload.ID, payload.ArgumentsDelta) - } - } + }, + ) + if err != nil && forwardErr == nil { + // 记录首个协议错误后继续排空事件通道，避免 provider 在后续发送时阻塞。 + forwardErr = err } case <-ctx.Done(): return @@ -606,12 +676,18 @@ func (s *Service) callProviderWithRetry( } streamEvents := make(chan provider.StreamEvent, 32) - streamDone := make(chan struct{}) + streamDone := make(chan error, 1) go s.forwardProviderEvents(ctx, runID, sessionID, streamEvents, streamDone, acc) err = modelProvider.Chat(ctx, req, streamEvents) close(streamEvents) - <-streamDone + forwardErr := <-streamDone + if forwardErr != nil { + if err != nil { + return nil, fmt.Errorf("runtime: provider stream handling failed after provider error: %v: %w", err, forwardErr) + } + return nil, forwardErr + } if err == nil { return acc, nil diff --git a/internal/runtime/runtime_test.go b/internal/runtime/runtime_test.go index 101a91ad..aac914a8 100644 --- a/internal/runtime/runtime_test.go +++ b/internal/runtime/runtime_test.go @@ -383,6 +383,142 @@ func TestServiceRun(t *testing.T) { } } +func TestServiceRunMergesLateToolCallMetadata(t *testing.T) { + t.Parallel() + + manager := newRuntimeConfigManager(t) + store := newMemoryStore() + tool := &stubTool{name: "filesystem_edit", content: "tool output"} + registry := tools.NewRegistry() + registry.Register(tool) + + scripted := &scriptedProvider{ + streams: [][]provider.StreamEvent{ + { + provider.NewToolCallDeltaStreamEvent(0, "", `{"path":"main.go"`), + provider.NewToolCallStartStreamEvent(0, "call-late", "filesystem_edit"), + provider.NewToolCallDeltaStreamEvent(0, "call-late", `}`), + }, + {provider.NewTextDeltaStreamEvent("done")}, + }, + } + + service := NewWithFactory(manager, registry, store, &scriptedProviderFactory{provider: scripted}, nil) + if err := service.Run(context.Background(), UserInput{RunID: "run-late-tool-metadata", Content: "edit"}); err != nil { + t.Fatalf("Run() error = %v", err) + } + + if tool.callCount != 1 { + t.Fatalf("expected tool to execute once, got %d", tool.callCount) + } + if tool.lastInput.ID != "call-late" { + t.Fatalf("expected merged tool call id %q, got %q", "call-late", tool.lastInput.ID) + } + if tool.lastInput.Name != "filesystem_edit" { + t.Fatalf("expected merged tool name %q, got %q", "filesystem_edit", tool.lastInput.Name) + } + if got := string(tool.lastInput.Arguments); got != `{"path":"main.go"}` { + t.Fatalf("expected merged tool arguments %q, got %q", `{"path":"main.go"}`, got) + } + + session := onlySession(t, store) + if len(session.Messages) < 3 { + t.Fatalf("expected assistant/tool follow-up messages, got %+v", session.Messages) + } + if len(session.Messages[1].ToolCalls) != 1 { + t.Fatalf("expected persisted assistant tool call, got %+v", session.Messages[1]) + } + if session.Messages[1].ToolCalls[0].ID != "call-late" || session.Messages[1].ToolCalls[0].Name != "filesystem_edit" { + t.Fatalf("expected merged assistant tool call metadata, got %+v", session.Messages[1].ToolCalls[0]) + } + if session.Messages[2].ToolCallID != "call-late" { + t.Fatalf("expected tool result to reference merged tool call id, got %+v", session.Messages[2]) + } +} + +func TestServiceRunRejectsToolCallWithoutID(t *testing.T) { + t.Parallel() + + manager := newRuntimeConfigManager(t) + store := newMemoryStore() + tool := &stubTool{name: "filesystem_edit", content: "tool output"} + registry := tools.NewRegistry() + registry.Register(tool) + + scripted := &scriptedProvider{ + streams: [][]provider.StreamEvent{ + { + provider.NewToolCallStartStreamEvent(0, "", "filesystem_edit"), + provider.NewToolCallDeltaStreamEvent(0, "", `{}`), + }, + }, + } + + service := NewWithFactory(manager, registry, store, &scriptedProviderFactory{provider: scripted}, nil) + 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) + } + if tool.callCount != 0 { + t.Fatalf("expected tool execution to be blocked, got %d calls", tool.callCount) + } +} + +func TestServiceRunRejectsMalformedProviderStreamEvent(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: [][]provider.StreamEvent{ + { + {Type: provider.StreamEventTextDelta}, + }, + }, + } + + service := NewWithFactory(manager, registry, store, &scriptedProviderFactory{provider: scripted}, nil) + err := service.Run(context.Background(), UserInput{RunID: "run-malformed-stream-event", Content: "hello"}) + if err == nil || !containsError(err, "text_delta event payload is nil") { + t.Fatalf("expected malformed stream event error, got %v", err) + } +} + +func TestServiceRunMalformedProviderStreamEventDoesNotDeadlock(t *testing.T) { + t.Parallel() + + manager := newRuntimeConfigManager(t) + store := newMemoryStore() + registry := tools.NewRegistry() + registry.Register(&stubTool{name: "filesystem_read_file", content: "default"}) + + stream := []provider.StreamEvent{{Type: provider.StreamEventTextDelta}} + for i := 0; i < 40; i++ { + stream = append(stream, provider.NewTextDeltaStreamEvent("ignored")) + } + scripted := &scriptedProvider{ + streams: [][]provider.StreamEvent{stream}, + } + + service := NewWithFactory(manager, registry, store, &scriptedProviderFactory{provider: scripted}, nil) + errCh := make(chan error, 1) + go func() { + errCh <- service.Run(context.Background(), UserInput{RunID: "run-malformed-stream-no-deadlock", Content: "hello"}) + }() + + select { + case err := <-errCh: + if err == nil || !containsError(err, "text_delta event payload is nil") { + t.Fatalf("expected malformed stream event error, got %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("expected run to fail instead of deadlocking on malformed stream event") + } +} + type stubCompactRunner struct { runFn func(ctx context.Context, input contextcompact.Input) (contextcompact.Result, error) calls []contextcompact.Input