diff --git a/pkg/channels/interfaces.go b/pkg/channels/interfaces.go index 99e977079..4b5847e70 100644 --- a/pkg/channels/interfaces.go +++ b/pkg/channels/interfaces.go @@ -1,6 +1,10 @@ package channels -import "context" +import ( + "context" + + "github.com/sipeed/picoclaw/pkg/bus" +) // TypingCapable — channels that can show a typing/thinking indicator. // StartTyping begins the indicator and returns a stop function. @@ -44,15 +48,9 @@ type StreamingCapable interface { BeginStream(ctx context.Context, chatID string) (Streamer, error) } -// Streamer pushes incremental content to a streaming-capable channel. -type Streamer interface { - // Update sends accumulated partial content to the user. - Update(ctx context.Context, content string) error - // Finalize commits the final message. After this, the Streamer is done. - Finalize(ctx context.Context, content string) error - // Cancel aborts the stream (e.g. when tool calls are detected mid-stream). - Cancel(ctx context.Context) -} +// Streamer is defined in pkg/bus to avoid circular imports. +// This alias keeps channel implementations using channels.Streamer unchanged. +type Streamer = bus.Streamer // PlaceholderRecorder is injected into channels by Manager. // Channels call these methods on inbound to register typing/placeholder state. diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index fcd068e6f..a2facc505 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -3,7 +3,8 @@ package telegram import ( "context" "fmt" - "math/rand" + "crypto/rand" + "encoding/binary" "net/http" "net/url" "os" @@ -800,7 +801,7 @@ func (c *TelegramChannel) BeginStream(ctx context.Context, chatID string) (chann return &telegramStreamer{ bot: c.bot, chatID: cid, - draftID: rand.Intn(1<<31-1) + 1, // non-zero random draft ID + draftID: cryptoRandInt(), // non-zero random draft ID }, nil } @@ -868,6 +869,11 @@ func (s *telegramStreamer) Finalize(ctx context.Context, content string) error { // Fallback to plain text tgMsg.ParseMode = "" if _, err = s.bot.SendMessage(ctx, tgMsg); err != nil { + logger.ErrorCF("telegram", "Finalize failed after HTML and plain-text attempts", map[string]any{ + "chat_id": s.chatID, + "error": err.Error(), + "len": len(content), + }) return fmt.Errorf("telegram finalize: %w", err) } } @@ -877,3 +883,10 @@ func (s *telegramStreamer) Finalize(ctx context.Context, content string) error { func (s *telegramStreamer) Cancel(ctx context.Context) { // Draft auto-expires on Telegram's side; nothing to clean up. } + +// cryptoRandInt returns a non-zero random int using crypto/rand. +func cryptoRandInt() int { + var b [4]byte + _, _ = rand.Read(b[:]) + return int(binary.BigEndian.Uint32(b[:])) | 1 // ensure non-zero +} diff --git a/pkg/providers/openai_compat/provider.go b/pkg/providers/openai_compat/provider.go index 241cc8c5b..3049320a5 100644 --- a/pkg/providers/openai_compat/provider.go +++ b/pkg/providers/openai_compat/provider.go @@ -244,11 +244,11 @@ func (p *Provider) ChatStream( return nil, fmt.Errorf("API request failed:\n Status: %d\n Body: %s", resp.StatusCode, string(body)) } - return parseStreamResponse(resp.Body, onChunk) + return parseStreamResponse(ctx, resp.Body, onChunk) } // parseStreamResponse parses an OpenAI-compatible SSE stream. -func parseStreamResponse(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 finishReason string var usage *UsageInfo @@ -262,7 +262,13 @@ func parseStreamResponse(reader io.Reader, onChunk func(accumulated string)) (*L activeTools := map[int]*toolAccum{} scanner := bufio.NewScanner(reader) + scanner.Buffer(make([]byte, 0, 1024*1024), 10*1024*1024) // 1MB initial, 10MB max for scanner.Scan() { + // Check for context cancellation between chunks + if err := ctx.Err(); err != nil { + return nil, err + } + line := scanner.Text() if !strings.HasPrefix(line, "data: ") {