From 79111425b0aa7e4f3f9623e3d4b82fac614c4aff Mon Sep 17 00:00:00 2001 From: Amir Mamaghani Date: Wed, 4 Mar 2026 19:16:57 +0100 Subject: [PATCH] refactor(streaming): remove dead code and simplify streaming wiring MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Delete unused Anthropic ChatStream/parseStream (-131 lines) — factory creates HTTPProvider for all OpenAI-compat providers including OpenRouter - Simplify runLLMIteration from 4 to 3 return values (remove unused streamed bool) - Replace managerStreamer struct with finalizeHookStreamer using embedding (Update/Cancel promoted, only Finalize overridden) --- pkg/agent/loop.go | 29 +++--- pkg/channels/manager.go | 33 +++---- pkg/providers/anthropic/provider.go | 132 ---------------------------- 3 files changed, 23 insertions(+), 171 deletions(-) diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 1c3af0552..1889ba322 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -686,7 +686,7 @@ func (al *AgentLoop) runAgentLoop( agent.Sessions.AddMessage(opts.SessionKey, "user", opts.UserMessage) // 4. Run LLM iteration loop - finalContent, iteration, streamSent, err := al.runLLMIteration(ctx, agent, messages, opts) + finalContent, iteration, err := al.runLLMIteration(ctx, agent, messages, opts) if err != nil { return "", err } @@ -708,8 +708,11 @@ func (al *AgentLoop) runAgentLoop( al.maybeSummarize(agent, opts.SessionKey, opts.Channel, opts.ChatID) } - // 8. Optional: send response via bus (skip if streaming already delivered it) - if opts.SendResponse && !streamSent { + // 8. Optional: send response via bus. + // Even when streaming already delivered the message, we still publish so that + // preSend can clean up typing/reaction/placeholder. The streamActive flag in + // the manager's preSend will skip the actual Send (no duplicate). + if opts.SendResponse { al.bus.PublishOutbound(ctx, bus.OutboundMessage{ Channel: opts.Channel, ChatID: opts.ChatID, @@ -727,11 +730,6 @@ func (al *AgentLoop) runAgentLoop( "final_length": len(finalContent), }) - // When streaming already delivered the message, return empty so the caller - // (Run loop) doesn't publish a duplicate via PublishOutbound. - if streamSent { - return "", nil - } return finalContent, nil } @@ -792,18 +790,15 @@ func (al *AgentLoop) handleReasoning( } // runLLMIteration executes the LLM call loop with tool handling. -// Returns (finalContent, iteration, streamed, error). -// When streamed is true, the response was already delivered to the user via -// streaming (sendMessageDraft + sendMessage) and PublishOutbound should be skipped. +// Returns (finalContent, iteration, error). func (al *AgentLoop) runLLMIteration( ctx context.Context, agent *AgentInstance, messages []providers.Message, opts processOptions, -) (string, int, bool, error) { +) (string, int, error) { iteration := 0 var finalContent string - var streamed bool // Check if both the provider and channel support streaming streamProvider, providerCanStream := agent.Provider.(providers.StreamingProvider) @@ -969,7 +964,7 @@ func (al *AgentLoop) runLLMIteration( "iteration": iteration, "error": err.Error(), }) - return "", iteration, false, fmt.Errorf("LLM call failed after retries: %w", err) + return "", iteration, fmt.Errorf("LLM call failed after retries: %w", err) } go al.handleReasoning( @@ -999,8 +994,6 @@ func (al *AgentLoop) runLLMIteration( logger.WarnCF("agent", "Stream finalize failed", map[string]any{ "error": err.Error(), }) - } else { - streamed = true } } @@ -1009,7 +1002,7 @@ func (al *AgentLoop) runLLMIteration( "agent_id": agent.ID, "iteration": iteration, "content_chars": len(finalContent), - "streamed": streamed, + "streamed": streamer != nil, }) break } @@ -1174,7 +1167,7 @@ func (al *AgentLoop) runLLMIteration( } } - return finalContent, iteration, streamed, nil + return finalContent, iteration, nil } // updateToolContexts updates the context for tools that need channel/chatID info. diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index c259e439a..805707de4 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -207,32 +207,23 @@ func (m *Manager) GetStreamer(ctx context.Context, channelName, chatID string) ( return nil, false } - // Wrap the channels.Streamer to track finalization for preSend coordination - return &managerStreamer{ - inner: streamer, - manager: m, - key: channelName + ":" + chatID, + // Mark streamActive on Finalize so preSend knows to clean up the placeholder + key := channelName + ":" + chatID + return &finalizeHookStreamer{ + Streamer: streamer, + onFinalize: func() { m.streamActive.Store(key, true) }, }, true } -// managerStreamer wraps a channels.Streamer to mark streamActive on Finalize. -type managerStreamer struct { - inner Streamer - manager *Manager - key string +// finalizeHookStreamer wraps a Streamer to run a hook on Finalize. +type finalizeHookStreamer struct { + Streamer + onFinalize func() } -func (s *managerStreamer) Update(ctx context.Context, content string) error { - return s.inner.Update(ctx, content) -} - -func (s *managerStreamer) Finalize(ctx context.Context, content string) error { - s.manager.streamActive.Store(s.key, true) - return s.inner.Finalize(ctx, content) -} - -func (s *managerStreamer) Cancel(ctx context.Context) { - s.inner.Cancel(ctx) +func (s *finalizeHookStreamer) Finalize(ctx context.Context, content string) error { + s.onFinalize() + return s.Streamer.Finalize(ctx, content) } // initChannel is a helper that looks up a factory by name and creates the channel. diff --git a/pkg/providers/anthropic/provider.go b/pkg/providers/anthropic/provider.go index 23b1f9561..2b73b33f3 100644 --- a/pkg/providers/anthropic/provider.go +++ b/pkg/providers/anthropic/provider.go @@ -9,8 +9,6 @@ import ( "github.com/anthropics/anthropic-sdk-go" "github.com/anthropics/anthropic-sdk-go/option" - "github.com/anthropics/anthropic-sdk-go/packages/ssestream" - "github.com/sipeed/picoclaw/pkg/providers/protocoltypes" ) @@ -94,136 +92,6 @@ func (p *Provider) Chat( return parseResponse(resp), nil } -// ChatStream implements providers.StreamingProvider. -// It streams text deltas to onChunk (with accumulated text) while building -// the same LLMResponse returned by Chat for tool-call compatibility. -func (p *Provider) ChatStream( - ctx context.Context, - messages []Message, - tools []ToolDefinition, - model string, - options map[string]any, - onChunk func(accumulated string), -) (*LLMResponse, error) { - var opts []option.RequestOption - if p.tokenSource != nil { - tok, err := p.tokenSource() - if err != nil { - return nil, fmt.Errorf("refreshing token: %w", err) - } - opts = append(opts, option.WithAuthToken(tok)) - } - - params, err := buildParams(messages, tools, model, options) - if err != nil { - return nil, err - } - - stream := p.client.Messages.NewStreaming(ctx, params, opts...) - return parseStream(stream, onChunk) -} - -// parseStream consumes a streaming response and builds an LLMResponse. -func parseStream( - stream *ssestream.Stream[anthropic.MessageStreamEventUnion], - onChunk func(accumulated string), -) (*LLMResponse, error) { - defer stream.Close() - - var textContent strings.Builder - var toolCalls []ToolCall - var stopReason anthropic.StopReason - var inputTokens, outputTokens int64 - - // Track tool_use blocks being assembled (by content block index) - type toolBlock struct { - id string - name string - inputJSON strings.Builder - } - activeTools := map[int64]*toolBlock{} - - for stream.Next() { - event := stream.Current() - - switch event.Type { - case "message_start": - if event.Message.Usage.InputTokens > 0 { - inputTokens = event.Message.Usage.InputTokens - } - - case "content_block_start": - cb := event.ContentBlock - if cb.Type == "tool_use" { - activeTools[event.Index] = &toolBlock{ - id: cb.ID, - name: cb.Name, - } - } - - case "content_block_delta": - delta := event.Delta - switch delta.Type { - case "text_delta": - textContent.WriteString(delta.Text) - if onChunk != nil { - onChunk(textContent.String()) - } - case "input_json_delta": - if tb, ok := activeTools[event.Index]; ok { - tb.inputJSON.WriteString(delta.PartialJSON) - } - } - - case "content_block_stop": - if tb, ok := activeTools[event.Index]; ok { - var args map[string]any - if err := json.Unmarshal([]byte(tb.inputJSON.String()), &args); err != nil { - log.Printf("anthropic stream: failed to decode tool call input for %q: %v", tb.name, err) - args = map[string]any{"raw": tb.inputJSON.String()} - } - toolCalls = append(toolCalls, ToolCall{ - ID: tb.id, - Name: tb.name, - Arguments: args, - }) - delete(activeTools, event.Index) - } - - case "message_delta": - stopReason = event.Delta.StopReason - if event.Usage.OutputTokens > 0 { - outputTokens = event.Usage.OutputTokens - } - } - } - - if err := stream.Err(); err != nil { - return nil, fmt.Errorf("claude streaming API call: %w", err) - } - - finishReason := "stop" - switch stopReason { - case anthropic.StopReasonToolUse: - finishReason = "tool_calls" - case anthropic.StopReasonMaxTokens: - finishReason = "length" - case anthropic.StopReasonEndTurn: - finishReason = "stop" - } - - return &LLMResponse{ - Content: textContent.String(), - ToolCalls: toolCalls, - FinishReason: finishReason, - Usage: &UsageInfo{ - PromptTokens: int(inputTokens), - CompletionTokens: int(outputTokens), - TotalTokens: int(inputTokens + outputTokens), - }, - }, nil -} - func (p *Provider) GetDefaultModel() string { return "claude-sonnet-4.6" }