opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: make streaming throttle interval and min growth configurable

Browse source 
Move hardcoded streamThrottleInterval (3s) and streamMinGrowth (200)
into StreamingConfig so they can be tuned per deployment via config
or environment variables.
This commit is contained in:
Amir Mamaghani 2026-03-05 18:49:29 +01:00
parent fa3c00ad9c
commit 0213494cff
4 changed files with 26 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
pkg/channels/telegram/telegram.go
View file

@ -2,9 +2,9 @@ package telegram
import ( import (
"context" "context"
"fmt"
"crypto/rand" "crypto/rand"
"encoding/binary" "encoding/binary"
"fmt"
"net/http" "net/http"
"net/url" "net/url"
"os" "os"
@ -798,10 +798,13 @@ func (c *TelegramChannel) BeginStream(ctx context.Context, chatID string) (chann
return nil, err return nil, err
} }
streamCfg := c.config.Channels.Telegram.Streaming
return &telegramStreamer{ return &telegramStreamer{
bot: c.bot, bot: c.bot,
chatID: cid, chatID: cid,
draftID: cryptoRandInt(), // non-zero random draft ID draftID: cryptoRandInt(),
throttleInterval: time.Duration(streamCfg.ThrottleSeconds) * time.Second,
minGrowth: streamCfg.MinGrowthChars,
}, nil }, nil
} }
@ -809,20 +812,17 @@ func (c *TelegramChannel) BeginStream(ctx context.Context, chatID string) (chann
// On first API error (e.g. bot lacks forum mode), it silently degrades: Update // On first API error (e.g. bot lacks forum mode), it silently degrades: Update
// becomes a no-op, while Finalize still delivers the final message. // becomes a no-op, while Finalize still delivers the final message.
type telegramStreamer struct { type telegramStreamer struct {
bot *telego.Bot bot *telego.Bot
chatID int64 chatID int64
draftID int draftID int
lastLen int throttleInterval time.Duration
lastAt time.Time minGrowth int
failed bool lastLen int
mu sync.Mutex lastAt time.Time
failed bool
mu sync.Mutex
} }
const (
streamThrottleInterval = 3 * time.Second
streamMinGrowth = 200 // minimum character growth to send an update
)
func (s *telegramStreamer) Update(ctx context.Context, content string) error { func (s *telegramStreamer) Update(ctx context.Context, content string) error {
s.mu.Lock() s.mu.Lock()
defer s.mu.Unlock() defer s.mu.Unlock()
@ -834,7 +834,7 @@ func (s *telegramStreamer) Update(ctx context.Context, content string) error {
// Throttle: skip if not enough time or content has passed // Throttle: skip if not enough time or content has passed
now := time.Now() now := time.Now()
growth := len(content) - s.lastLen growth := len(content) - s.lastLen
if s.lastLen > 0 && now.Sub(s.lastAt) < streamThrottleInterval && growth < streamMinGrowth { if s.lastLen > 0 && now.Sub(s.lastAt) < s.throttleInterval && growth < s.minGrowth {
return nil return nil
} }

4
pkg/config/config.go
View file

@ -238,7 +238,9 @@ type PlaceholderConfig struct {
} }
type StreamingConfig struct { type StreamingConfig struct {
Enabled bool `json:"enabled,omitempty" env:"PICOCLAW_CHANNELS_TELEGRAM_STREAMING_ENABLED"` Enabled bool `json:"enabled,omitempty" env:"PICOCLAW_CHANNELS_TELEGRAM_STREAMING_ENABLED"`
ThrottleSeconds int `json:"throttle_seconds,omitempty" env:"PICOCLAW_CHANNELS_TELEGRAM_STREAMING_THROTTLE_SECONDS"`
MinGrowthChars int `json:"min_growth_chars,omitempty" env:"PICOCLAW_CHANNELS_TELEGRAM_STREAMING_MIN_GROWTH_CHARS"`
} }
type WhatsAppConfig struct { type WhatsAppConfig struct {

2
pkg/config/defaults.go
View file

@ -58,7 +58,7 @@ func DefaultConfig() *Config {
Enabled: true, Enabled: true,
Text: "Thinking... 💭", Text: "Thinking... 💭",
}, },
Streaming: StreamingConfig{Enabled: true}, Streaming: StreamingConfig{Enabled: true, ThrottleSeconds: 3, MinGrowthChars: 200},
}, },
Feishu: FeishuConfig{ Feishu: FeishuConfig{
Enabled: false, Enabled: false,

6
pkg/providers/openai_compat/provider.go
View file

@ -248,7 +248,11 @@ func (p *Provider) ChatStream(
} }
// parseStreamResponse parses an OpenAI-compatible SSE stream. // parseStreamResponse parses an OpenAI-compatible SSE stream.
func parseStreamResponse(ctx context.Context, reader io.Reader, onChunk func(accumulated string)) (*LLMResponse, error) { func parseStreamResponse(
ctx context.Context,
reader io.Reader,
onChunk func(accumulated string),
) (*LLMResponse, error) {
var textContent strings.Builder var textContent strings.Builder
var finishReason string var finishReason string
var usage *UsageInfo var usage *UsageInfo