feat(telegram): stream LLM responses via sendMessageDraft

[amirmamaghani]

Summary

Implements real-time LLM response streaming to Telegram using the sendMessageDraft API, replacing the current "Thinking..." placeholder pattern with live token-by-token output as it's generated.

Closes #1098

What changed

New interfaces and capabilities

• StreamingProvider (providers/types.go): opt-in ChatStream() method with onChunk callback receiving accumulated text
• StreamingCapable + Streamer (channels/interfaces.go): opt-in channel capability (like TypingCapable/PlaceholderCapable)
• StreamDelegate (bus/bus.go): decouples agent loop from channel manager

Provider streaming (works with any OpenAI-compatible API)

• openai_compat/provider.go: SSE streaming parser (stream: true) — handles text deltas, tool call assembly, and usage tracking. Uses a separate HTTP client without timeout for long streams (context cancellation provides safety).
• http_provider.go: delegates ChatStream to openai_compat
• anthropic/provider.go: native SDK streaming via Messages.NewStreaming() for direct Anthropic API connections

Telegram streaming

• telegram/telegram.go: BeginStream() returns a telegramStreamer that calls sendMessageDraft with throttling (3s / 200 chars minimum) to stay within Telegram rate limits
• Graceful degradation: first sendMessageDraft error (e.g., no forum/topics mode) sets failed=true — subsequent Update() calls become no-ops while Finalize() still delivers via SendMessage
• Config opt-out: streaming.enabled (default: true) in telegram channel config

Agent loop + manager integration

• agent/loop.go: when both provider and channel support streaming, uses ChatStream with streaming callback. Cancels stream on tool calls. Skips PublishOutbound when Finalize already delivered.
• channels/manager.go: implements StreamDelegate, tracks streamActive state per channel+chatID, coordinates with placeholder editing in preSend
• channels/base.go: always shows typing + placeholder on inbound (streaming coordination happens on the output side via streamActive map)

Fallback behavior

Scenario	Behavior
Bot without forum/topics mode	First sendMessageDraft fails → streamer degrades silently → Finalize sends via SendMessage
Non-streaming provider	Type assertion fails → falls back to Chat() → normal placeholder flow
streaming.enabled: false	BeginStream returns error → GetStreamer returns nil → normal placeholder flow
Tool calls mid-stream	Stream cancelled → tools execute → response sent via normal outbound path

Test plan

• Verified streamed=true in logs with OpenRouter provider
• Verified graceful degradation on 429 rate limit (streamer disables, Finalize delivers)
• Verified go build ./pkg/... and go vet ./pkg/... pass
• Verified full binary builds cleanly
• Manual test with forum-mode-enabled bot for full draft streaming UX
• Verify tool-call scenarios cancel stream correctly

[CLAassistant]

All committers have signed the CLA.

[amirmamaghani]

ready for merge guys @yinwm @alexhoshina @lxowalle

[yinwm]

Code Review: Telegram Streaming Implementation

This PR implements real-time LLM response streaming to Telegram using sendMessageDraft API. The overall architecture is well-designed with clean separation of concerns.

---

🔴 Critical Issues

1. Duplicate Streamer Interface Definition - Potential Type Mismatch

pkg/channels/interfaces.go and pkg/bus/bus.go both define Streamer interface:

// channels/interfaces.go
type Streamer interface {
    Update(ctx context.Context, content string) error
    Finalize(ctx context.Context, content string) error
    Cancel(ctx context.Context)
}

// bus/bus.go  
type Streamer interface {
    Update(ctx context.Context, content string) error
    Finalize(ctx context.Context, content string) error
    Cancel(ctx context.Context)
}

Problem: These are two different types in Go. manager.go returns channels.Streamer, but bus.GetStreamer declares return type as bus.Streamer. This could cause subtle bugs or compilation issues if the interfaces drift apart.

Recommendation: Keep only one definition. Define Streamer in bus/bus.go and have channels package import and use bus.Streamer. Or use type aliasing.

---

2. Scanner Buffer Size Limitation in SSE Parsing

pkg/providers/openai_compat/provider.go:247:

scanner := bufio.NewScanner(reader)
for scanner.Scan() {

bufio.Scanner has a default max token size of 64KB. If an LLM response contains a single line (without newlines) exceeding 64KB, it will trigger bufio.ErrTooLong.

Recommendation: Increase buffer size:

scanner := bufio.NewScanner(reader)
buf := make([]byte, 0, 1024*1024) // 1MB initial capacity
scanner.Buffer(buf, 10*1024*1024) // max 10MB

---

🟡 Medium Issues

3. DraftID Potential Collision Risk

pkg/channels/telegram/telegram.go:815:

draftID: rand.Intn(1<<31-1) + 1, // non-zero random draft ID

math/rand is not cryptographically secure, and without proper seeding could produce predictable/colliding values.

Recommendation: Use crypto/rand:

import "crypto/rand"
import "encoding/binary"

func randomDraftID() int {
    var b [4]byte
    rand.Read(b[:])
    return int(binary.BigEndian.Uint32(b[:])) | 1 // ensure non-zero
}

---

4. Missing Context Cancellation Check in Stream Parsing

In parseStreamResponse, if the context is cancelled mid-stream, the function continues processing until the stream ends naturally.

Recommendation: Add context check in the parsing loop:

select {
case <-ctx.Done():
    return nil, ctx.Err()
default:
    // process chunk
}

---

5. Silent Failure in Finalize

When Update fails (sets failed=true), Finalize is still called and sends the message. However, if Finalize also fails after the fallback, the user might not see any message at all.

Recommendation: Consider retry mechanism or logging to persistent storage for recovery.

---

📋 Suggestions

6. Add Observability/Metrics

Streaming success rate, fallback rate, and latency are important for operations.

Recommendation: Add Prometheus metrics or structured logging.

---

7. Configuration Could Be More Flexible

The throttle constants are hardcoded:

const (
    streamThrottleInterval = 3 * time.Second
    streamMinGrowth        = 200
)

Recommendation: Consider making these configurable for different deployment scenarios.

---

✅ Well Done

1. Graceful Degradation: The failed flag pattern provides smooth fallback
2. Interface Design: StreamingCapable as optional interface maintains backward compatibility
3. Throttling Strategy: Prevents Telegram API rate limiting
4. Tool Call Handling: Correctly cancels stream when tool calls are detected
5. Heartbeat Fix: Properly guards internal messages from triggering streaming

---

Summary

Severity	Count
🔴 Critical	2
🟡 Medium	3
🟢 Suggestion	2

Recommendation: Address at least the two critical issues before merging.

🤖 Generated with Claude Code

[amirmamaghani]

Thanks for the thorough review @yinwm! Pushed fixes for all actionable items:

Critical:

1. Duplicate Streamer interface — channels.Streamer is now a type alias for bus.Streamer, single source of truth
2. Scanner buffer size — Increased to 1MB initial / 10MB max to handle large single-line SSE payloads

Medium:
3. DraftID collision — Switched from math/rand to crypto/rand
4. Context cancellation in stream parsing — Added ctx.Err() check between chunks in the SSE loop
5. Finalize logging — Added structured error logging with chat_id and content length when both HTML and plain-text delivery fail

Skipped #6 (observability/metrics) and #7 (configurable throttle) for now — can follow up in separate PRs if needed.

EDIT: you know what, let me do the throttle config too 😁

[amirmamaghani]

forgot to write you.. ready to merge! 🙌

[amirmamaghani]

guys, just a friendly reminder that ready to merge @yinwm @alexhoshina

[alexhoshina]

guys, just a friendly reminder that ready to merge @yinwm @alexhoshina

I've briefly reviewed the changes related to the channel, and they should be fine. The other parts might need to be reviewed by @yinwm.

[amirmamaghani]

hey bro, just a reminder @yinwm

[alexhoshina]

Perhaps the CI issues need to be resolved, as all three checks have failed

[amirmamaghani]

Updated — removed the wrapper function, now just uses parseTelegramChatID directly with _ for the unused threadID. Same fix, no new functions.

[yinwm]

Review: Ready to Merge ✅

This PR is well-designed with a solid streaming pipeline implementation and robust graceful degradation.

Before merging, please address:

1. 🗑️ Remove the pico-echo-server binary file that was accidentally committed
2. 🔀 Resolve any conflicts with the main branch

Once these are addressed, this is good to merge. The remaining suggestions I had (tool call index handling, unit tests) are incremental improvements that can be addressed in follow-up PRs.

[amirmamaghani]

Resolved the two items from the latest review:

1. Removed pico-echo-server binary — accidentally committed binary has been deleted
2. Resolved merge conflicts with main — updated 5 files:
   • config/config.example.json — merged use_markdown_v2 field + streaming config
   • pkg/agent/loop.go — merged activeRequests tracking with streaming support
   • pkg/bus/bus.go — merged closeOnce/wg refactoring with streamDelegate
   • pkg/config/defaults.go — merged UseMarkdownV2 default with Streaming default
   • pkg/providers/openai_compat/provider.go — adapted ChatStream to use refactored common package (HandleErrorResponse, AsInt, AsFloat, supportsPromptCacheKey)

All checks pass locally: go build, go vet, golangci-lint, and go test ./pkg/... (all green).

[amirmamaghani]

ready for merge

[alexhoshina]

Setting streamActive to True too early may cause the message sending to not fall back to the regular path after Streamer.Finalize fails.