refactor(streaming): remove dead code and simplify streaming wiring

- 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

@@ -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)
+	// 8. Optional: send response via bus.
-	if opts.SendResponse && !streamSent {
+	// 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).
+// Returns (finalContent, iteration, error).
-// When streamed is true, the response was already delivered to the user via
-// streaming (sendMessageDraft + sendMessage) and PublishOutbound should be skipped.
 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.

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
+	// Mark streamActive on Finalize so preSend knows to clean up the placeholder
-	return &managerStreamer{
+	key := channelName + ":" + chatID
-		inner:   streamer,
+	return &finalizeHookStreamer{
-		manager: m,
+		Streamer:   streamer,
-		key:     channelName + ":" + chatID,
+		onFinalize: func() { m.streamActive.Store(key, true) },
 	}, true
 }
 
-// managerStreamer wraps a channels.Streamer to mark streamActive on Finalize.
+// finalizeHookStreamer wraps a Streamer to run a hook on Finalize.
-type managerStreamer struct {
+type finalizeHookStreamer struct {
-	inner   Streamer
+	Streamer
-	manager *Manager
+	onFinalize func()
-	key     string
 }
 
-func (s *managerStreamer) Update(ctx context.Context, content string) error {
+func (s *finalizeHookStreamer) Finalize(ctx context.Context, content string) error {
-	return s.inner.Update(ctx, content)
+	s.onFinalize()
-}
+	return s.Streamer.Finalize(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)
 }
 
 // initChannel is a helper that looks up a factory by name and creates the channel.

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"
 }