feat(channels/telegram): Telegram 渠道插件（Bot API 长轮询/Webhook + file_id 媒体 + 主动推送强）

plugins/channels/telegram/adapter.go

@@ -0,0 +1,244 @@
+// Package telegram implements the Telegram Bot API as a channel adapter for
+// Cyrene v2. It conforms to internal/channel/adapter.PlatformAdapter, translating
+// Telegram updates (getUpdates long polling, or a registered webhook) into the
+// normalized UnifiedMessage model and delivering outbound replies through the
+// Bot API send* methods.
+//
+// Media is carried by Telegram file_id values: inbound attachments populate
+// UnifiedMessage.Media with URL set to the file_id (plus size/name/mime when the
+// update provides them), and outbound image/voice/documents are sent back by
+// file_id via sendPhoto/sendVoice/sendDocument. The design authority is
+// DevDocs/design/08-渠道接入.md section 4; the contract is defined in sections
+// 0-2. Proactive push is enabled (design: Telegram is a strong push channel).
+//
+// The adapter is self-contained: it depends only on the standard library plus
+// the project's internal adapter, config-independent logger, and channel/actor
+// model. Tokens are never logged.
+package telegram
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"sort"
+	"sync"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
+)
+
+// Compile-time assertion that Adapter satisfies the shared contract.
+var _ adapter.PlatformAdapter = (*Adapter)(nil)
+
+// Adapter is a Telegram Bot API channel adapter. It is safe for concurrent use.
+// The zero use is via New; do not copy it after first use.
+type Adapter struct {
+	mu      sync.RWMutex
+	cfg     Config
+	client  *http.Client
+	baseURL string
+	log     *slog.Logger
+
+	connected  bool
+	closed     bool
+	webhook    bool
+	handler    adapter.MessageHandler
+	pollCancel context.CancelFunc
+	runID      uint64
+
+	offset int64
+	chats  map[string]adapter.ChannelInfo
+}
+
+// Name returns the platform identifier.
+func (a *Adapter) Name() string { return "telegram" }
+
+// Config returns a snapshot of the adapter's configuration. The BotToken is the
+// sensitive field; callers that persist this map must treat it as a secret.
+func (a *Adapter) Config() map[string]any {
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+	return map[string]any{
+		"bot_token":   a.cfg.BotToken,
+		"webhook_url": a.cfg.WebhookURL,
+	}
+}
+
+// Connect establishes the session. It is idempotent. When WebhookURL is
+// configured the webhook is registered with the Bot API and long polling is
+// disabled; otherwise the adapter prepares for getUpdates polling.
+func (a *Adapter) Connect(ctx context.Context) error {
+	a.mu.Lock()
+	if a.connected {
+		a.mu.Unlock()
+		return nil
+	}
+	if a.closed {
+		a.mu.Unlock()
+		return fmt.Errorf("telegram: %w", adapter.ErrClosed)
+	}
+	if a.cfg.BotToken == "" {
+		a.mu.Unlock()
+		return fmt.Errorf("telegram: %w", ErrMissingToken)
+	}
+	token := a.cfg.BotToken
+	webhookURL := a.cfg.WebhookURL
+	a.mu.Unlock()
+
+	webhookMode := false
+	if webhookURL != "" {
+		if err := a.setWebhook(ctx, webhookURL); err != nil {
+			return fmt.Errorf("telegram: register webhook: %w", err)
+		}
+		webhookMode = true
+	}
+	a.log.Info("connected", "webhook", webhookMode, "token_len", len(token))
+
+	a.mu.Lock()
+	a.connected = true
+	a.webhook = webhookMode
+	a.offset = 0
+	a.mu.Unlock()
+	return nil
+}
+
+// Disconnect tears down the session and stops an in-flight poll loop. It is
+// idempotent.
+func (a *Adapter) Disconnect(ctx context.Context) error {
+	a.mu.Lock()
+	if !a.connected {
+		a.mu.Unlock()
+		return nil
+	}
+	a.connected = false
+	cancel := a.pollCancel
+	a.mu.Unlock()
+	if cancel != nil {
+		cancel()
+	}
+	a.log.Info("disconnected")
+	return nil
+}
+
+// Close permanently shuts the adapter down, stopping any polling loop and
+// marking it closed. It is idempotent; subsequent operations fail with
+// adapter.ErrClosed.
+func (a *Adapter) Close() error {
+	a.mu.Lock()
+	if a.closed {
+		a.mu.Unlock()
+		return nil
+	}
+	a.closed = true
+	a.connected = false
+	cancel := a.pollCancel
+	a.mu.Unlock()
+	if cancel != nil {
+		cancel()
+	}
+	a.log.Info("closed")
+	return nil
+}
+
+// IsConnected reports whether the session is currently live.
+func (a *Adapter) IsConnected() bool {
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+	return a.connected
+}
+
+// HealthCheck verifies the session is usable by calling getMe against the Bot
+// API. It returns an error wrapping adapter.ErrNotConnected when not connected.
+func (a *Adapter) HealthCheck(ctx context.Context) error {
+	if err := a.requireConnected(); err != nil {
+		return err
+	}
+	bot, err := a.getMe(ctx)
+	if err != nil {
+		return fmt.Errorf("telegram: health check: %w", err)
+	}
+	a.log.Debug("health check ok", "bot", bot)
+	return nil
+}
+
+// Capabilities declares what the Telegram platform supports. Proactive push is
+// enabled because Telegram permits arbitrary outbound messages.
+func (a *Adapter) Capabilities() adapter.Capabilities {
+	return adapter.Capabilities{
+		MaxMessageLength:  maxMessageLen,
+		Markdown:          true,
+		Image:             true,
+		Voice:             true,
+		Reaction:          false,
+		TypingHint:        false,
+		RecommendBurstMax: 1,
+		Proactive:         true,
+	}.Normalize()
+}
+
+// SetMessageHandler installs the push-style inbound callback used by the webhook
+// delivery path (see ProcessUpdate). A nil fn clears the handler.
+func (a *Adapter) SetMessageHandler(fn adapter.MessageHandler) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	a.handler = fn
+}
+
+// ListChannels returns the conversations observed so far, ordered by ChannelID.
+func (a *Adapter) ListChannels(ctx context.Context) ([]adapter.ChannelInfo, error) {
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+	out := make([]adapter.ChannelInfo, 0, len(a.chats))
+	for _, c := range a.chats {
+		out = append(out, c)
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].ChannelID < out[j].ChannelID })
+	return out, nil
+}
+
+// ProcessUpdate decodes a raw webhook payload (a single Telegram Update) and
+// delivers the normalized message through the installed message handler. It is
+// the push-side counterpart to ReadMessages and returns the number of messages
+// delivered. Non-message updates (for example callback queries) are ignored.
+func (a *Adapter) ProcessUpdate(ctx context.Context, raw []byte) (int, error) {
+	var update telegramUpdate
+	if err := json.Unmarshal(raw, &update); err != nil {
+		return 0, fmt.Errorf("telegram: decode webhook update: %w", err)
+	}
+	msg, ok := a.toUnifiedMessage(update)
+	if !ok {
+		return 0, nil
+	}
+	if err := msg.Validate(); err != nil {
+		return 0, fmt.Errorf("telegram: validate webhook update: %w", err)
+	}
+	a.mu.RLock()
+	handler := a.handler
+	a.mu.RUnlock()
+	if handler != nil {
+		handler(ctx, msg)
+	}
+	return 1, nil
+}
+
+// requireConnected guards operations that need a live session.
+func (a *Adapter) requireConnected() error {
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+	if a.closed {
+		return fmt.Errorf("telegram: %w", adapter.ErrClosed)
+	}
+	if !a.connected {
+		return fmt.Errorf("telegram: %w", adapter.ErrNotConnected)
+	}
+	return nil
+}
+
+// isWebhookMode reports whether the adapter is serving via webhook rather than
+// polling.
+func (a *Adapter) isWebhookMode() bool {
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+	return a.webhook
+}

plugins/channels/telegram/botapi.go

@@ -0,0 +1,205 @@
+package telegram
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+)
+
+// botResponse is the shared envelope Telegram returns for every method: ok plus
+// either result or the error description/code.
+type botResponse struct {
+	OK          bool            `json:"ok"`
+	Result      json.RawMessage `json:"result"`
+	Description string          `json:"description"`
+	ErrorCode   int             `json:"error_code"`
+}
+
+// doJSON posts a JSON payload to a Bot API method and decodes the envelope. The
+// caller is responsible for decoding Result; a non-OK reply is returned as
+// ErrBadResponse wrapped with the API description.
+func (a *Adapter) doJSON(ctx context.Context, method string, payload any) (*botResponse, error) {
+	var buf bytes.Buffer
+	if payload != nil {
+		if err := json.NewEncoder(&buf).Encode(payload); err != nil {
+			return nil, fmt.Errorf("telegram: encode %s request: %w", method, err)
+		}
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, a.endpoint(method), &buf)
+	if err != nil {
+		return nil, fmt.Errorf("telegram: build %s request: %w", method, err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json")
+
+	resp, err := a.client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("telegram: %s call: %w", method, err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
+		desc := strings.TrimSpace(string(body))
+		if desc != "" {
+			return nil, fmt.Errorf("telegram: %s: status %d: %s: %w", method, resp.StatusCode, desc, ErrBadResponse)
+		}
+		return nil, fmt.Errorf("telegram: %s: status %d: %w", method, resp.StatusCode, ErrBadResponse)
+	}
+
+	var out botResponse
+	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+		return nil, fmt.Errorf("telegram: decode %s response: %w", method, err)
+	}
+	if !out.OK {
+		return nil, fmt.Errorf("telegram: %s: %w", method, ErrBadResponse)
+	}
+	return &out, nil
+}
+
+// getMe verifies the token and returns the bot's username.
+func (a *Adapter) getMe(ctx context.Context) (string, error) {
+	out, err := a.doJSON(ctx, "getMe", nil)
+	if err != nil {
+		return "", err
+	}
+	var me struct {
+		Username string `json:"username"`
+	}
+	if err := json.Unmarshal(out.Result, &me); err != nil {
+		return "", fmt.Errorf("telegram: getMe result: %w", err)
+	}
+	return me.Username, nil
+}
+
+// setWebhook registers the webhook URL and disables getUpdates polling.
+func (a *Adapter) setWebhook(ctx context.Context, url string) error {
+	_, err := a.doJSON(ctx, "setWebhook", map[string]any{"url": url})
+	return err
+}
+
+// getUpdates performs a long-poll for updates at the current offset. It returns
+// a list of updates; the caller advances a.offset after acknowledging them.
+func (a *Adapter) getUpdates(ctx context.Context) ([]telegramUpdate, error) {
+	a.mu.RLock()
+	offset := a.offset
+	a.mu.RUnlock()
+
+	payload := map[string]any{
+		"offset":  offset,
+		"limit":   maxUpdatesPerPoll,
+		"timeout": longPollTimeoutSec,
+	}
+	out, err := a.doJSON(ctx, "getUpdates", payload)
+	if err != nil {
+		return nil, err
+	}
+	var updates []telegramUpdate
+	if err := json.Unmarshal(out.Result, &updates); err != nil {
+		return nil, fmt.Errorf("telegram: getUpdates result: %w", err)
+	}
+	if updates == nil {
+		updates = []telegramUpdate{}
+	}
+	return updates, nil
+}
+
+// sendMessage sends a text message and returns the new message id.
+func (a *Adapter) sendMessage(ctx context.Context, chatID, text string, replyTo int64, parseMode string) (int64, error) {
+	payload := map[string]any{
+		"chat_id": chatID,
+		"text":    text,
+	}
+	if replyTo != 0 {
+		payload["reply_to_message_id"] = replyTo
+	}
+	if parseMode != "" {
+		payload["parse_mode"] = parseMode
+	}
+	return a.sendPayload(ctx, "sendMessage", payload)
+}
+
+// sendPhoto sends an existing photo by file_id and returns the new message id.
+func (a *Adapter) sendPhoto(ctx context.Context, chatID, fileID string, replyTo int64, caption string) (int64, error) {
+	payload := map[string]any{
+		"chat_id": chatID,
+		"photo":   fileID,
+	}
+	if caption != "" {
+		payload["caption"] = caption
+		payload["parse_mode"] = "Markdown"
+	}
+	if replyTo != 0 {
+		payload["reply_to_message_id"] = replyTo
+	}
+	return a.sendPayload(ctx, "sendPhoto", payload)
+}
+
+// sendVoice sends an existing voice clip by file_id and returns the new id.
+func (a *Adapter) sendVoice(ctx context.Context, chatID, fileID string, replyTo int64, caption string) (int64, error) {
+	payload := map[string]any{
+		"chat_id": chatID,
+		"voice":   fileID,
+	}
+	if caption != "" {
+		payload["caption"] = caption
+		payload["parse_mode"] = "Markdown"
+	}
+	if replyTo != 0 {
+		payload["reply_to_message_id"] = replyTo
+	}
+	return a.sendPayload(ctx, "sendVoice", payload)
+}
+
+// sendDocument sends an existing document by file_id and returns the new id.
+func (a *Adapter) sendDocument(ctx context.Context, chatID, fileID, fileName string, replyTo int64, caption string) (int64, error) {
+	payload := map[string]any{
+		"chat_id":  chatID,
+		"document": fileID,
+	}
+	if fileName != "" {
+		payload["file_name"] = fileName
+	}
+	if caption != "" {
+		payload["caption"] = caption
+		payload["parse_mode"] = "Markdown"
+	}
+	if replyTo != 0 {
+		payload["reply_to_message_id"] = replyTo
+	}
+	return a.sendPayload(ctx, "sendDocument", payload)
+}
+
+// sendPayload posts a media/message payload and returns the acknowledged
+// message id from the Bot API result.
+func (a *Adapter) sendPayload(ctx context.Context, method string, payload map[string]any) (int64, error) {
+	out, err := a.doJSON(ctx, method, payload)
+	if err != nil {
+		return 0, err
+	}
+	var res struct {
+		MessageID int64 `json:"message_id"`
+	}
+	if err := json.Unmarshal(out.Result, &res); err != nil {
+		return 0, fmt.Errorf("telegram: %s result: %w", method, err)
+	}
+	return res.MessageID, nil
+}
+
+// parseReplyToID converts a reply context identifier to a Telegram message id.
+// A non-numeric or empty value yields 0 (no reply).
+func parseReplyToID(s string) int64 {
+	if s == "" {
+		return 0
+	}
+	n, err := strconv.ParseInt(s, 10, 64)
+	if err != nil {
+		return 0
+	}
+	return n
+}

plugins/channels/telegram/config.go

@@ -0,0 +1,90 @@
+package telegram
+
+import (
+	"io"
+	"log/slog"
+	"net/http"
+	"time"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
+)
+
+// Defaults and knob values for the Telegram Bot API integration.
+const (
+	// defaultBaseURL is the Telegram API root for unconfigured adapters.
+	defaultBaseURL = "https://api.telegram.org"
+	// httpTimeout bounds a single Bot API request.
+	httpTimeout = 180 * time.Second
+	// longPollTimeoutSec is the getUpdates long-poll timeout in seconds.
+	longPollTimeoutSec = 50
+	// pollInterval is how long to sleep between getUpdates batches.
+	pollInterval = 1 * time.Second
+	// maxUpdatesPerPoll caps the number of updates fetched in one getUpdates call.
+	maxUpdatesPerPoll = 100
+	// maxMessageLen is Telegram's per-message text limit.
+	maxMessageLen = 4096
+)
+
+// Config holds the settings required to reach the Telegram Bot API.
+type Config struct {
+	// BotToken is the secret token issued by BotFather. It is stored and
+	// surfaced through Config() but never written to logs.
+	BotToken string `json:"bot_token"`
+	// WebhookURL is the public HTTPS endpoint Telegram delivers updates to.
+	// When empty the adapter uses getUpdates long polling instead.
+	WebhookURL string `json:"webhook_url"`
+}
+
+// Option customizes an Adapter at construction time.
+type Option func(*Adapter)
+
+// WithHTTPClient injects the *http.Client used for Bot API requests. Tests pass
+// the client of an httptest.Server so the mock endpoint answers calls.
+func WithHTTPClient(c *http.Client) Option {
+	return func(a *Adapter) { a.client = c }
+}
+
+// WithBaseURL overrides the Telegram API root (default https://api.telegram.org).
+// Tests point it at an httptest.Server URL.
+func WithBaseURL(base string) Option {
+	return func(a *Adapter) {
+		if base != "" {
+			a.baseURL = base
+		}
+	}
+}
+
+// WithLogger injects the logger used by the adapter. When nil the adapter logs
+// through a discard logger so a bare adapter stays silent; callers wire their
+// production logger via this option.
+func WithLogger(l *slog.Logger) Option {
+	return func(a *Adapter) {
+		if l == nil {
+			l = logger.New(logger.Options{Out: io.Discard})
+		}
+		a.log = logger.WithSource(l, "plugins/channels/telegram")
+	}
+}
+
+// New builds a Telegram adapter. It performs no network I/O; call Connect to
+// establish the session.
+func New(cfg Config, opts ...Option) *Adapter {
+	a := &Adapter{
+		cfg:     cfg,
+		client:  &http.Client{Timeout: httpTimeout},
+		baseURL: defaultBaseURL,
+		log:     logger.WithSource(logger.New(logger.Options{Out: io.Discard}), "plugins/channels/telegram"),
+		chats:   map[string]adapter.ChannelInfo{},
+	}
+	for _, opt := range opts {
+		opt(a)
+	}
+	return a
+}
+
+// endpoint returns the absolute Bot API URL for the given method, bundling the
+// token into the path (never into query parameters or logs).
+func (a *Adapter) endpoint(method string) string {
+	return a.baseURL + "/bot" + a.cfg.BotToken + "/" + method
+}

plugins/channels/telegram/config_test.go

@@ -0,0 +1,217 @@
+package telegram
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"sync"
+	"testing"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
+)
+
+func TestName(t *testing.T) {
+	t.Parallel()
+	a := New(Config{BotToken: "x"})
+	if got := a.Name(); got != "telegram" {
+		t.Fatalf("Name() = %q, want telegram", got)
+	}
+}
+
+func TestConfig(t *testing.T) {
+	t.Parallel()
+	a := New(Config{BotToken: "abc123", WebhookURL: "https://example.com/hook"})
+	got := a.Config()
+	if got["bot_token"] != "abc123" {
+		t.Fatalf("bot_token = %v, want abc123", got["bot_token"])
+	}
+	if got["webhook_url"] != "https://example.com/hook" {
+		t.Fatalf("webhook_url = %v, want https://example.com/hook", got["webhook_url"])
+	}
+	// Round-trip through the shared decoder to prove the map is decodable.
+	var round Config
+	if err := adapter.DecodeConfig(got, &round); err != nil {
+		t.Fatalf("DecodeConfig: %v", err)
+	}
+	if round.BotToken != "abc123" || round.WebhookURL != "https://example.com/hook" {
+		t.Fatalf("round trip mismatch: %+v", round)
+	}
+}
+
+func TestCapabilities(t *testing.T) {
+	t.Parallel()
+	a := New(Config{BotToken: "x"})
+	caps := a.Capabilities()
+	if !caps.Proactive {
+		t.Error("Proactive should be true for telegram")
+	}
+	if !caps.Markdown || !caps.Image || !caps.Voice {
+		t.Errorf("expected Markdown/Image/Voice supported, got %+v", caps)
+	}
+	if caps.Reaction || caps.TypingHint {
+		t.Errorf("expected Reaction/TypingHint unsupported, got %+v", caps)
+	}
+	if caps.MaxMessageLength != 4096 {
+		t.Errorf("MaxMessageLength = %d, want 4096", caps.MaxMessageLength)
+	}
+	if caps.RecommendBurstMax <= 0 {
+		t.Errorf("RecommendBurstMax should be positive, got %d", caps.RecommendBurstMax)
+	}
+}
+
+func TestListChannels_Empty(t *testing.T) {
+	t.Parallel()
+	a := New(Config{BotToken: "x"})
+	got, err := a.ListChannels(context.Background())
+	if err != nil {
+		t.Fatalf("ListChannels: %v", err)
+	}
+	if got == nil || len(got) != 0 {
+		t.Fatalf("expected empty non-nil slice, got %#v", got)
+	}
+}
+
+func TestConnect_MissingToken(t *testing.T) {
+	t.Parallel()
+	a := New(Config{})
+	err := a.Connect(context.Background())
+	if !errors.Is(err, ErrMissingToken) {
+		t.Fatalf("err = %v, want ErrMissingToken", err)
+	}
+	if a.IsConnected() {
+		t.Fatal("should not be connected")
+	}
+}
+
+func TestConnect_WebhookIdempotent(t *testing.T) {
+	t.Parallel()
+	var mu sync.Mutex
+	calls := 0
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if methodOf(r) != "setWebhook" {
+			http.Error(w, "unexpected method", http.StatusNotFound)
+			return
+		}
+		mu.Lock()
+		calls++
+		mu.Unlock()
+		okResponse(w, map[string]any{"webhook_url": "https://example.com/hook"})
+	}))
+	a.cfg.WebhookURL = "https://example.com/hook"
+
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	if !a.IsConnected() {
+		t.Fatal("expected connected")
+	}
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("second Connect should be idempotent: %v", err)
+	}
+	mu.Lock()
+	if calls != 1 {
+		t.Fatalf("expected 1 setWebhook call, got %d", calls)
+	}
+	mu.Unlock()
+}
+
+func TestConnect_PollingMode(t *testing.T) {
+	t.Parallel()
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "no api call expected in poll mode", http.StatusNotFound)
+	}))
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	if !a.IsConnected() {
+		t.Fatal("expected connected")
+	}
+	if a.isWebhookMode() {
+		t.Fatal("expected polling mode")
+	}
+}
+
+func TestDisconnect_Idempotent(t *testing.T) {
+	t.Parallel()
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "no api call expected", http.StatusNotFound)
+	}))
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	if err := a.Disconnect(context.Background()); err != nil {
+		t.Fatalf("Disconnect: %v", err)
+	}
+	if a.IsConnected() {
+		t.Fatal("expected disconnected")
+	}
+	if err := a.Disconnect(context.Background()); err != nil {
+		t.Fatalf("second Disconnect should be idempotent: %v", err)
+	}
+}
+
+func TestClose_Terminal(t *testing.T) {
+	t.Parallel()
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "no api call expected", http.StatusNotFound)
+	}))
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	if err := a.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	if a.IsConnected() {
+		t.Fatal("expected disconnected after close")
+	}
+	if err := a.Connect(context.Background()); !errors.Is(err, adapter.ErrClosed) {
+		t.Fatalf("Connect after close err = %v, want ErrClosed", err)
+	}
+	if err := a.HealthCheck(context.Background()); !errors.Is(err, adapter.ErrClosed) {
+		t.Fatalf("HealthCheck after close err = %v, want ErrClosed", err)
+	}
+}
+
+func TestHealthCheck_OK(t *testing.T) {
+	t.Parallel()
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if methodOf(r) != "getMe" {
+			http.Error(w, "unexpected method", http.StatusNotFound)
+			return
+		}
+		okResponse(w, map[string]any{"id": 123, "username": "cyrene_bot"})
+	}))
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	if err := a.HealthCheck(context.Background()); err != nil {
+		t.Fatalf("HealthCheck: %v", err)
+	}
+}
+
+func TestHealthCheck_NotConnected(t *testing.T) {
+	t.Parallel()
+	a := New(Config{BotToken: "x"})
+	err := a.HealthCheck(context.Background())
+	if !errors.Is(err, adapter.ErrNotConnected) {
+		t.Fatalf("err = %v, want ErrNotConnected", err)
+	}
+}
+
+func TestHealthCheck_APIError(t *testing.T) {
+	t.Parallel()
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if methodOf(r) != "getMe" {
+			http.Error(w, "unexpected method", http.StatusNotFound)
+			return
+		}
+		failResponse(w, "Unauthorized", 401)
+	}))
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	err := a.HealthCheck(context.Background())
+	if !errors.Is(err, ErrBadResponse) {
+		t.Fatalf("err = %v, want ErrBadResponse", err)
+	}
+}

plugins/channels/telegram/errors.go

@@ -0,0 +1,17 @@
+package telegram
+
+import "errors"
+
+// Sentinel errors returned by the Telegram adapter. Consumers test for them
+// with errors.Is; failures that mean "not connected" wrap adapter.ErrNotConnected
+// so the shared channel sentinel still matches.
+var (
+	// ErrMissingToken is returned when no BotToken is configured.
+	ErrMissingToken = errors.New("telegram: missing bot token")
+	// ErrBadResponse is returned when the Bot API replies with OK=false or an
+	// unexpected payload shape.
+	ErrBadResponse = errors.New("telegram: bad bot api response")
+	// ErrNoChatTarget is returned when an outbound message carries no usable
+	// conversation/group identifier.
+	ErrNoChatTarget = errors.New("telegram: message has no target chat")
+)

plugins/channels/telegram/helpers_test.go

@@ -0,0 +1,61 @@
+package telegram
+
+import (
+	"encoding/json"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
+)
+
+// discardLogger returns a silent, source-tagged logger for tests.
+func discardLogger() *slog.Logger {
+	return logger.WithSource(logger.New(logger.Options{Out: io.Discard}), "plugins/channels/telegram")
+}
+
+// newTestAdapter wires a Telegram adapter to an httptest server whose handler
+// answers the Bot API methods. The server is closed on test cleanup.
+func newTestAdapter(t *testing.T, h http.Handler) (*Adapter, *httptest.Server) {
+	t.Helper()
+	srv := httptest.NewServer(h)
+	t.Cleanup(srv.Close)
+	a := New(
+		Config{BotToken: "123456:TESTTOKEN"},
+		WithBaseURL(srv.URL),
+		WithHTTPClient(srv.Client()),
+		WithLogger(discardLogger()),
+	)
+	return a, srv
+}
+
+// methodOf extracts the Bot API method name from the request path, which takes
+// the form /bot<token>/<method>.
+func methodOf(r *http.Request) string {
+	parts := strings.Split(r.URL.Path, "/")
+	return parts[len(parts)-1]
+}
+
+// decodeBody unmarshals an incoming JSON request body into a map for assertions.
+func decodeBody(r *http.Request) map[string]any {
+	var m map[string]any
+	if err := json.NewDecoder(r.Body).Decode(&m); err != nil {
+		return nil
+	}
+	return m
+}
+
+// okResponse writes a generic Bot API success envelope with the given result.
+func okResponse(w http.ResponseWriter, result any) {
+	body := map[string]any{"ok": true, "result": result}
+	_ = json.NewEncoder(w).Encode(body)
+}
+
+// failResponse writes a Bot API error envelope.
+func failResponse(w http.ResponseWriter, desc string, code int) {
+	body := map[string]any{"ok": false, "description": desc, "error_code": code}
+	_ = json.NewEncoder(w).Encode(body)
+}

plugins/channels/telegram/poll.go

@@ -0,0 +1,93 @@
+package telegram
+
+import (
+	"context"
+	"time"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
+)
+
+// ReadMessages runs a blocking getUpdates long-poll loop, emitting each
+// normalized inbound message to emit until ctx is canceled or the adapter is
+// disconnected. When running in webhook mode it does not poll; it simply waits
+// for ctx to be canceled, since messages are delivered through the message
+// handler via ProcessUpdate instead.
+//
+// The update offset is advanced only after a batch has been fully processed, so
+// a canceled read does not lose the batch; acknowledgment is batched so the
+// getUpdates call does not re-deliver the same update_id flood.
+func (a *Adapter) ReadMessages(ctx context.Context, emit func(adapter.UnifiedMessage)) {
+	if emit == nil {
+		return
+	}
+	if a.isWebhookMode() {
+		<-ctx.Done()
+		return
+	}
+	if err := a.requireConnected(); err != nil {
+		a.log.Warn("read not started", "err", err)
+		return
+	}
+
+	pollCtx, cancel := context.WithCancel(ctx)
+	a.mu.Lock()
+	a.runID++
+	a.pollCancel = cancel
+	runID := a.runID
+	a.mu.Unlock()
+	defer func() {
+		a.mu.Lock()
+		if a.runID == runID {
+			a.pollCancel = nil
+		}
+		a.mu.Unlock()
+		cancel()
+	}()
+
+	for {
+		if err := pollCtx.Err(); err != nil {
+			return
+		}
+		updates, err := a.getUpdates(pollCtx)
+		if err != nil {
+			a.log.Warn("long poll failed", "err", err)
+			if !sleepWait(pollCtx, pollInterval) {
+				return
+			}
+			continue
+		}
+		if len(updates) == 0 {
+			if !sleepWait(pollCtx, pollInterval) {
+				return
+			}
+			continue
+		}
+		for _, u := range updates {
+			msg, ok := a.toUnifiedMessage(u)
+			if !ok {
+				continue
+			}
+			if err := msg.Validate(); err != nil {
+				a.log.Warn("skip invalid message", "err", err)
+				continue
+			}
+			emit(msg)
+		}
+		a.mu.Lock()
+		a.offset = updates[len(updates)-1].UpdateID + 1
+		a.mu.Unlock()
+	}
+}
+
+// sleepWait blocks for d or until ctx is canceled, returning false when the
+// context fires first.
+func sleepWait(ctx context.Context, d time.Duration) bool {
+	t := time.NewTimer(d)
+	defer t.Stop()
+	select {
+	case <-ctx.Done():
+		return false
+	case <-t.C:
+		return true
+	}
+}

plugins/channels/telegram/poll_test.go

@@ -0,0 +1,171 @@
+package telegram
+
+import (
+	"context"
+	"net/http"
+	"sync"
+	"testing"
+	"time"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
+)
+
+// fakeUpdate builds a minimal telegramUpdate JSON object.
+func fakeUpdate(id int, text string) map[string]any {
+	return map[string]any{
+		"update_id": id,
+		"message": map[string]any{
+			"message_id": id,
+			"from":       map[string]any{"id": 1, "first_name": "Tester"},
+			"chat":       map[string]any{"id": 10, "type": "private"},
+			"date":       1700000000,
+			"text":       text,
+		},
+	}
+}
+
+func TestReadMessages_LongPollOffset(t *testing.T) {
+	t.Parallel()
+	var mu sync.Mutex
+	first := true
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if methodOf(r) != "getUpdates" {
+			http.Error(w, "unexpected method", http.StatusNotFound)
+			return
+		}
+		mu.Lock()
+		firstCall := first
+		first = false
+		mu.Unlock()
+		if firstCall {
+			okResponse(w, []map[string]any{fakeUpdate(1, "one"), fakeUpdate(2, "two")})
+			return
+		}
+		okResponse(w, []map[string]any{})
+	}))
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	emitted := make(chan adapter.UnifiedMessage, 4)
+	done := make(chan struct{})
+	go func() {
+		a.ReadMessages(ctx, func(m adapter.UnifiedMessage) { emitted <- m })
+		close(done)
+	}()
+
+	got := make([]adapter.UnifiedMessage, 0, 2)
+	for i := 0; i < 2; i++ {
+		select {
+		case m := <-emitted:
+			got = append(got, m)
+		case <-time.After(3 * time.Second):
+			t.Fatal("timeout waiting for emitted message")
+		}
+	}
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("ReadMessages did not return after ctx cancel")
+	}
+
+	if len(got) != 2 {
+		t.Fatalf("emitted %d messages, want 2", len(got))
+	}
+	if got[0].Content != "one" || got[1].Content != "two" {
+		t.Errorf("contents = %q/%q, want one/two", got[0].Content, got[1].Content)
+	}
+	a.mu.RLock()
+	offset := a.offset
+	a.mu.RUnlock()
+	if offset != 3 {
+		t.Errorf("offset = %d, want 3", offset)
+	}
+}
+
+func TestReadMessages_CtxPreCanceled(t *testing.T) {
+	t.Parallel()
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "no api call expected", http.StatusNotFound)
+	}))
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	done := make(chan struct{})
+	go func() {
+		a.ReadMessages(ctx, func(m adapter.UnifiedMessage) {})
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("ReadMessages should return immediately on canceled ctx")
+	}
+}
+
+func TestReadMessages_WebhookWaits(t *testing.T) {
+	t.Parallel()
+	var calls int
+	var mu sync.Mutex
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if methodOf(r) == "setWebhook" {
+			okResponse(w, map[string]any{"webhook_url": "https://example.com/hook"})
+			return
+		}
+		mu.Lock()
+		calls++
+		mu.Unlock()
+		http.Error(w, "no getUpdates expected in webhook mode", http.StatusNotFound)
+	}))
+	a.cfg.WebhookURL = "https://example.com/hook"
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		a.ReadMessages(ctx, func(m adapter.UnifiedMessage) {})
+		close(done)
+	}()
+	time.Sleep(100 * time.Millisecond)
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("ReadMessages should return on cancel in webhook mode")
+	}
+	mu.Lock()
+	if calls != 0 {
+		t.Errorf("getUpdates called %d times in webhook mode", calls)
+	}
+	mu.Unlock()
+}
+
+func TestReadMessages_ErrorRetriesThenCancel(t *testing.T) {
+	t.Parallel()
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if methodOf(r) != "getUpdates" {
+			http.Error(w, "unexpected method", http.StatusNotFound)
+			return
+		}
+		failResponse(w, "Conflict", 409)
+	}))
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		a.ReadMessages(ctx, func(m adapter.UnifiedMessage) {})
+		close(done)
+	}()
+	time.Sleep(150 * time.Millisecond)
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("ReadMessages should return on cancel despite getUpdates errors")
+	}
+}

plugins/channels/telegram/send.go

@@ -0,0 +1,94 @@
+package telegram
+
+import (
+	"context"
+	"fmt"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
+)
+
+// markdownParseMode is the parse_mode sent with outgoing text/captions so the
+// Markdown capability is actually exercised.
+const markdownParseMode = "Markdown"
+
+// SendMessage delivers the reply messages contained in resp to their target
+// chats. Each message goes to its own ChannelID (or GroupID) conversation.
+func (a *Adapter) SendMessage(ctx context.Context, resp adapter.UnifiedResponse) error {
+	if err := a.requireConnected(); err != nil {
+		return err
+	}
+	for _, m := range resp.Messages {
+		if err := a.sendOne(ctx, m, resp.ReplyTo); err != nil {
+			return fmt.Errorf("telegram: send message: %w", err)
+		}
+	}
+	return nil
+}
+
+// SendProactive pushes a message not triggered by a recent inbound message.
+// Telegram permits arbitrary proactive push, so this is only gated by the
+// connection state.
+func (a *Adapter) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) error {
+	if err := a.requireConnected(); err != nil {
+		return err
+	}
+	return a.sendOne(ctx, msg, "")
+}
+
+// sendOne delivers a single message to its conversation, choosing the Bot API
+// method that matches the content type and the presence of a file_id payload.
+func (a *Adapter) sendOne(ctx context.Context, m adapter.UnifiedMessage, replyTo string) error {
+	chatID := m.ChannelID
+	if chatID == "" {
+		chatID = m.GroupID
+	}
+	if chatID == "" {
+		return fmt.Errorf("telegram: %w", ErrNoChatTarget)
+	}
+	replyToID := parseReplyToID(replyTo)
+	fileID := mediaFileID(m.Media)
+
+	switch m.ContentType {
+	case adapter.ContentImage:
+		if fileID != "" {
+			_, err := a.sendPhoto(ctx, chatID, fileID, replyToID, m.Content)
+			return err
+		}
+	case adapter.ContentVoice:
+		if fileID != "" {
+			_, err := a.sendVoice(ctx, chatID, fileID, replyToID, m.Content)
+			return err
+		}
+	case adapter.ContentFile:
+		if fileID != "" {
+			_, err := a.sendDocument(ctx, chatID, fileID, mediaFileName(m.Media), replyToID, m.Content)
+			return err
+		}
+	}
+
+	if m.Content == "" {
+		return fmt.Errorf("telegram: %w: empty message content", ErrBadResponse)
+	}
+	_, err := a.sendMessage(ctx, chatID, m.Content, replyToID, markdownParseMode)
+	return err
+}
+
+// mediaFileID returns the first non-empty file_id carried by the media list.
+func mediaFileID(media []adapter.Media) string {
+	for _, md := range media {
+		if md.URL != "" {
+			return md.URL
+		}
+	}
+	return ""
+}
+
+// mediaFileName returns the first non-empty file name carried by the media list.
+func mediaFileName(media []adapter.Media) string {
+	for _, md := range media {
+		if md.Name != "" {
+			return md.Name
+		}
+	}
+	return ""
+}

plugins/channels/telegram/send_test.go

@@ -0,0 +1,324 @@
+package telegram
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"sync"
+	"testing"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
+)
+
+func TestSendMessage_TextConstruction(t *testing.T) {
+	t.Parallel()
+	var mu sync.Mutex
+	var body map[string]any
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if methodOf(r) != "sendMessage" {
+			http.Error(w, "unexpected method: "+methodOf(r), http.StatusNotFound)
+			return
+		}
+		mu.Lock()
+		body = decodeBody(r)
+		mu.Unlock()
+		okResponse(w, map[string]any{"message_id": 100})
+	}))
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+
+	resp := adapter.UnifiedResponse{
+		ReplyTo: "42",
+		Messages: []adapter.UnifiedMessage{
+			{ChannelID: "200", ChannelType: adapter.ChannelDirect, Content: "hello", ContentType: adapter.ContentText},
+		},
+	}
+	if err := a.SendMessage(context.Background(), resp); err != nil {
+		t.Fatalf("SendMessage: %v", err)
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if body["chat_id"] != "200" {
+		t.Errorf("chat_id = %v, want 200", body["chat_id"])
+	}
+	if body["text"] != "hello" {
+		t.Errorf("text = %v, want hello", body["text"])
+	}
+	if body["parse_mode"] != "Markdown" {
+		t.Errorf("parse_mode = %v, want Markdown", body["parse_mode"])
+	}
+	if body["reply_to_message_id"] != float64(42) {
+		t.Errorf("reply_to_message_id = %v, want 42", body["reply_to_message_id"])
+	}
+}
+
+func TestSendMessage_ImageByFileID(t *testing.T) {
+	t.Parallel()
+	var mu sync.Mutex
+	var body map[string]any
+	var method string
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		m := methodOf(r)
+		if m != "sendPhoto" {
+			http.Error(w, "unexpected method: "+m, http.StatusNotFound)
+			return
+		}
+		mu.Lock()
+		method = m
+		body = decodeBody(r)
+		mu.Unlock()
+		okResponse(w, map[string]any{"message_id": 101})
+	}))
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	resp := adapter.UnifiedResponse{
+		Messages: []adapter.UnifiedMessage{
+			{
+				ChannelID:   "200",
+				ChannelType: adapter.ChannelDirect,
+				Content:     "a caption",
+				ContentType: adapter.ContentImage,
+				Media:       []adapter.Media{{Kind: adapter.MediaImage, URL: "file-abc"}},
+			},
+		},
+	}
+	if err := a.SendMessage(context.Background(), resp); err != nil {
+		t.Fatalf("SendMessage: %v", err)
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if method != "sendPhoto" {
+		t.Errorf("method = %q, want sendPhoto", method)
+	}
+	if body["chat_id"] != "200" || body["photo"] != "file-abc" {
+		t.Errorf("body = %v, want chat_id 200 and photo file-abc", body)
+	}
+	if body["caption"] != "a caption" || body["parse_mode"] != "Markdown" {
+		t.Errorf("caption/parse_mode = %v/%v", body["caption"], body["parse_mode"])
+	}
+}
+
+func TestSendMessage_ImageNoFileIDFallsBackToText(t *testing.T) {
+	t.Parallel()
+	var mu sync.Mutex
+	var method string
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		m := methodOf(r)
+		mu.Lock()
+		method = m
+		mu.Unlock()
+		if m != "sendMessage" {
+			http.Error(w, "unexpected method: "+m, http.StatusNotFound)
+			return
+		}
+		okResponse(w, map[string]any{"message_id": 102})
+	}))
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	resp := adapter.UnifiedResponse{
+		Messages: []adapter.UnifiedMessage{
+			{ChannelID: "200", ContentType: adapter.ContentImage, Content: "no file_id so text"},
+		},
+	}
+	if err := a.SendMessage(context.Background(), resp); err != nil {
+		t.Fatalf("SendMessage: %v", err)
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if method != "sendMessage" {
+		t.Errorf("method = %q, want fallback sendMessage", method)
+	}
+}
+
+func TestSendMessage_VoiceAndDocument(t *testing.T) {
+	t.Parallel()
+	for _, tc := range []struct {
+		name       string
+		ct         adapter.ContentType
+		media      adapter.Media
+		wantMethod string
+		wantField  string
+	}{
+		{name: "voice", ct: adapter.ContentVoice, media: adapter.Media{Kind: adapter.MediaVoice, URL: "voice-1"}, wantMethod: "sendVoice", wantField: "voice"},
+		{name: "document", ct: adapter.ContentFile, media: adapter.Media{Kind: adapter.MediaFile, URL: "doc-1", Name: "x.pdf"}, wantMethod: "sendDocument", wantField: "document"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			var mu sync.Mutex
+			var body map[string]any
+			var method string
+			a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				m := methodOf(r)
+				mu.Lock()
+				method = m
+				body = decodeBody(r)
+				mu.Unlock()
+				if m != tc.wantMethod {
+					http.Error(w, "unexpected method: "+m, http.StatusNotFound)
+					return
+				}
+				okResponse(w, map[string]any{"message_id": 103})
+			}))
+			if err := a.Connect(context.Background()); err != nil {
+				t.Fatalf("Connect: %v", err)
+			}
+			resp := adapter.UnifiedResponse{
+				Messages: []adapter.UnifiedMessage{
+					{ChannelID: "200", ChannelType: adapter.ChannelDirect, ContentType: tc.ct, Media: []adapter.Media{tc.media}},
+				},
+			}
+			if err := a.SendMessage(context.Background(), resp); err != nil {
+				t.Fatalf("SendMessage: %v", err)
+			}
+			mu.Lock()
+			defer mu.Unlock()
+			if method != tc.wantMethod {
+				t.Errorf("method = %q, want %q", method, tc.wantMethod)
+			}
+			field := tc.media.URL
+			if body[tc.wantField] != field {
+				t.Errorf("%s = %v, want %v", tc.wantField, body[tc.wantField], field)
+			}
+		})
+	}
+}
+
+func TestSendProactive(t *testing.T) {
+	t.Parallel()
+	var mu sync.Mutex
+	var body map[string]any
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if methodOf(r) != "sendMessage" {
+			http.Error(w, "unexpected method", http.StatusNotFound)
+			return
+		}
+		mu.Lock()
+		body = decodeBody(r)
+		mu.Unlock()
+		okResponse(w, map[string]any{"message_id": 104})
+	}))
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	msg := adapter.UnifiedMessage{
+		ChannelID: "300", ChannelType: adapter.ChannelDirect,
+		Content: "reminder", ContentType: adapter.ContentText,
+	}
+	if err := a.SendProactive(context.Background(), msg); err != nil {
+		t.Fatalf("SendProactive: %v", err)
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if body["chat_id"] != "300" || body["text"] != "reminder" {
+		t.Errorf("body = %v, want chat_id 300 text reminder", body)
+	}
+	if _, ok := body["reply_to_message_id"]; ok {
+		t.Errorf("proactive message should not carry reply_to_message_id, got %v", body)
+	}
+}
+
+func TestSendMessage_NotConnected(t *testing.T) {
+	t.Parallel()
+	a := New(Config{BotToken: "x"})
+	err := a.SendMessage(context.Background(), adapter.UnifiedResponse{})
+	if !errors.Is(err, adapter.ErrNotConnected) {
+		t.Fatalf("err = %v, want ErrNotConnected", err)
+	}
+	if err := a.SendProactive(context.Background(), adapter.UnifiedMessage{}); !errors.Is(err, adapter.ErrNotConnected) {
+		t.Fatalf("SendProactive err = %v, want ErrNotConnected", err)
+	}
+}
+
+func TestSendMessage_NoTargetChat(t *testing.T) {
+	t.Parallel()
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "unexpected method", http.StatusNotFound)
+	}))
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	err := a.SendMessage(context.Background(), adapter.UnifiedResponse{
+		Messages: []adapter.UnifiedMessage{{ContentType: adapter.ContentText, Content: "hi"}},
+	})
+	if !errors.Is(err, ErrNoChatTarget) {
+		t.Fatalf("err = %v, want ErrNoChatTarget", err)
+	}
+}
+
+func TestSendMessage_APIError(t *testing.T) {
+	t.Parallel()
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if methodOf(r) != "sendMessage" {
+			http.Error(w, "unexpected method", http.StatusNotFound)
+			return
+		}
+		failResponse(w, "Bad Request: chat not found", 400)
+	}))
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	err := a.SendMessage(context.Background(), adapter.UnifiedResponse{
+		Messages: []adapter.UnifiedMessage{{ChannelID: "999", ContentType: adapter.ContentText, Content: "hi"}},
+	})
+	if !errors.Is(err, ErrBadResponse) {
+		t.Fatalf("err = %v, want ErrBadResponse", err)
+	}
+}
+
+func TestProcessUpdate_DeliversToHandler(t *testing.T) {
+	t.Parallel()
+	a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "unexpected method", http.StatusNotFound)
+	}))
+	if err := a.Connect(context.Background()); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+
+	var mu sync.Mutex
+	var got *adapter.UnifiedMessage
+	a.SetMessageHandler(func(ctx context.Context, m adapter.UnifiedMessage) {
+		mu.Lock()
+		got = &m
+		mu.Unlock()
+	})
+	raw := []byte(`{"update_id":1,"message":{"message_id":5,"from":{"id":1,"first_name":"Ada"},"chat":{"id":10,"type":"private"},"date":1700000000,"text":"webhook hi"}}`)
+	n, err := a.ProcessUpdate(context.Background(), raw)
+	if err != nil {
+		t.Fatalf("ProcessUpdate: %v", err)
+	}
+	if n != 1 {
+		t.Errorf("delivered %d messages, want 1", n)
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if got == nil {
+		t.Fatal("handler was not called")
+	}
+	if got.Content != "webhook hi" || got.Platform != "telegram" {
+		t.Errorf("message = %+v", got)
+	}
+}
+
+func TestProcessUpdate_NonMessageIgnored(t *testing.T) {
+	t.Parallel()
+	a := New(Config{BotToken: "x"})
+	raw := []byte(`{"update_id":2,"callback_query":{"id":"cq","from":{"id":1},"data":"btn"}}`)
+	n, err := a.ProcessUpdate(context.Background(), raw)
+	if err != nil {
+		t.Fatalf("ProcessUpdate: %v", err)
+	}
+	if n != 0 {
+		t.Errorf("delivered %d messages, want 0 for callback query", n)
+	}
+}
+
+func TestProcessUpdate_InvalidJSON(t *testing.T) {
+	t.Parallel()
+	a := New(Config{BotToken: "x"})
+	if _, err := a.ProcessUpdate(context.Background(), []byte("{not json")); err == nil {
+		t.Fatal("expected error decoding invalid JSON")
+	}
+}

plugins/channels/telegram/updates.go

@@ -0,0 +1,223 @@
+package telegram
+
+import (
+	"strconv"
+	"time"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
+)
+
+// telegramUpdate is the top-level object Telegram sends for every event.
+type telegramUpdate struct {
+	UpdateID          int64            `json:"update_id"`
+	Message           *telegramMessage `json:"message"`
+	EditedMessage     *telegramMessage `json:"edited_message"`
+	ChannelPost       *telegramMessage `json:"channel_post"`
+	EditedChannelPost *telegramMessage `json:"edited_channel_post"`
+}
+
+// message returns the concrete chat message carried by the update, or nil.
+func (u telegramUpdate) message() *telegramMessage {
+	switch {
+	case u.Message != nil:
+		return u.Message
+	case u.EditedMessage != nil:
+		return u.EditedMessage
+	case u.ChannelPost != nil:
+		return u.ChannelPost
+	case u.EditedChannelPost != nil:
+		return u.EditedChannelPost
+	default:
+		return nil
+	}
+}
+
+// telegramMessage is a chat message as delivered by the Bot API.
+type telegramMessage struct {
+	MessageID       int64               `json:"message_id"`
+	From            *telegramUser       `json:"from"`
+	Chat            telegramChat        `json:"chat"`
+	Date            int64               `json:"date"`
+	Text            string              `json:"text"`
+	Caption         string              `json:"caption"`
+	Photo           []telegramPhotoSize `json:"photo"`
+	Voice           *telegramAudio      `json:"voice"`
+	Audio           *telegramAudio      `json:"audio"`
+	Document        *telegramDocument   `json:"document"`
+	Video           *telegramVideo      `json:"video"`
+	ReplyToMessage  *telegramMessage    `json:"reply_to_message"`
+	MessageThreadID int64               `json:"message_thread_id"`
+}
+
+// telegramUser is the sender of a message.
+type telegramUser struct {
+	ID        int64  `json:"id"`
+	FirstName string `json:"first_name"`
+	LastName  string `json:"last_name"`
+	Username  string `json:"username"`
+}
+
+// telegramChat is the conversation a message belongs to.
+type telegramChat struct {
+	ID    int64  `json:"id"`
+	Type  string `json:"type"` // private | group | supergroup | channel
+	Title string `json:"title"`
+}
+
+// telegramPhotoSize is one entry of a photo attachment.
+type telegramPhotoSize struct {
+	FileID   string `json:"file_id"`
+	FileSize int64  `json:"file_size"`
+}
+
+// telegramAudio describes a voice or audio attachment.
+type telegramAudio struct {
+	FileID   string `json:"file_id"`
+	FileSize int64  `json:"file_size"`
+	MIMEType string `json:"mime_type"`
+}
+
+// telegramDocument describes a document attachment.
+type telegramDocument struct {
+	FileID   string `json:"file_id"`
+	FileSize int64  `json:"file_size"`
+	FileName string `json:"file_name"`
+	MIMEType string `json:"mime_type"`
+}
+
+// telegramVideo describes a video attachment.
+type telegramVideo struct {
+	FileID   string `json:"file_id"`
+	FileSize int64  `json:"file_size"`
+	MIMEType string `json:"mime_type"`
+}
+
+// toUnifiedMessage converts an update to the normalized model. It returns ok
+// false for updates that carry no deliverable chat message.
+func (a *Adapter) toUnifiedMessage(update telegramUpdate) (adapter.UnifiedMessage, bool) {
+	m := update.message()
+	if m == nil {
+		return adapter.UnifiedMessage{}, false
+	}
+	msg := adapter.UnifiedMessage{
+		Platform:    a.Name(),
+		ChannelID:   strconv.FormatInt(m.Chat.ID, 10),
+		ChannelType: chatType(m.Chat.Type),
+		MessageID:   strconv.FormatInt(m.MessageID, 10),
+		Timestamp:   time.Unix(m.Date, 0),
+	}
+
+	if m.Chat.Type == "group" || m.Chat.Type == "supergroup" || m.Chat.Type == "channel" {
+		msg.GroupID = strconv.FormatInt(m.Chat.ID, 10)
+		msg.GroupName = m.Chat.Title
+	}
+	if m.MessageThreadID != 0 {
+		msg.ThreadID = strconv.FormatInt(m.MessageThreadID, 10)
+	}
+
+	content, contentType, media := extractPayload(m)
+	msg.Content = content
+	msg.ContentType = contentType
+	msg.Media = media
+
+	if m.From != nil {
+		msg.SenderID = strconv.FormatInt(m.From.ID, 10)
+		msg.SenderName = displayName(m.From)
+		msg.Sender = adapter.Actor{
+			Role:        adapter.RoleUnknown,
+			UserID:      msg.SenderID,
+			DisplayName: msg.SenderName,
+			GroupID:     msg.GroupID,
+			GroupName:   msg.GroupName,
+		}.Resolve()
+	}
+
+	if m.ReplyToMessage != nil {
+		msg.ReplyToID = strconv.FormatInt(m.ReplyToMessage.MessageID, 10)
+		msg.ReplyToText = m.ReplyToMessage.Text
+		if msg.ReplyToText == "" {
+			msg.ReplyToText = m.ReplyToMessage.Caption
+		}
+	}
+
+	a.recordChat(m.Chat, m.From)
+	return msg, true
+}
+
+// extractPayload picks the primary content and attachment list from a message.
+func extractPayload(m *telegramMessage) (string, adapter.ContentType, []adapter.Media) {
+	media := make([]adapter.Media, 0, 2)
+	content := m.Caption
+	if m.Text != "" {
+		content = m.Text
+	}
+	ct := adapter.ContentText
+
+	if len(m.Photo) > 0 {
+		p := m.Photo[len(m.Photo)-1]
+		media = append(media, adapter.Media{Kind: adapter.MediaImage, URL: p.FileID, Size: p.FileSize, MIMEType: "image/jpeg"})
+		ct = adapter.ContentImage
+	}
+	if m.Voice != nil {
+		media = append(media, adapter.Media{Kind: adapter.MediaVoice, URL: m.Voice.FileID, Size: m.Voice.FileSize, MIMEType: m.Voice.MIMEType})
+		ct = adapter.ContentVoice
+	} else if m.Audio != nil {
+		media = append(media, adapter.Media{Kind: adapter.MediaVoice, URL: m.Audio.FileID, Size: m.Audio.FileSize, MIMEType: m.Audio.MIMEType})
+		ct = adapter.ContentVoice
+	}
+	if m.Document != nil {
+		media = append(media, adapter.Media{Kind: adapter.MediaFile, URL: m.Document.FileID, Size: m.Document.FileSize, Name: m.Document.FileName, MIMEType: m.Document.MIMEType})
+		ct = adapter.ContentFile
+	}
+	if m.Video != nil {
+		media = append(media, adapter.Media{Kind: adapter.MediaVideo, URL: m.Video.FileID, Size: m.Video.FileSize, MIMEType: m.Video.MIMEType})
+		ct = adapter.ContentFile
+	}
+
+	return content, ct, media
+}
+
+// chatType maps a Telegram chat type to the adapter's ChannelType.
+func chatType(t string) adapter.ChannelType {
+	if t == "group" || t == "supergroup" || t == "channel" {
+		return adapter.ChannelGroup
+	}
+	return adapter.ChannelDirect
+}
+
+// displayName renders a sender's best human-readable name without leaking
+// anything sensitive.
+func displayName(u *telegramUser) string {
+	if u.FirstName != "" && u.LastName != "" {
+		return u.FirstName + " " + u.LastName
+	}
+	if u.FirstName != "" {
+		return u.FirstName
+	}
+	if u.Username != "" {
+		return u.Username
+	}
+	return strconv.FormatInt(u.ID, 10)
+}
+
+// recordChat caches a conversation so ListChannels can report it.
+func (a *Adapter) recordChat(c telegramChat, from *telegramUser) {
+	info := adapter.ChannelInfo{
+		ChannelID:   strconv.FormatInt(c.ID, 10),
+		ChannelType: chatType(c.Type),
+		ChannelName: c.Title,
+	}
+	if info.ChannelType == adapter.ChannelGroup {
+		info.GroupID = info.ChannelID
+		info.GroupName = c.Title
+	} else if from != nil {
+		info.ChannelName = displayName(from)
+	}
+	if info.ChannelName == "" {
+		info.ChannelName = info.ChannelID
+	}
+
+	a.mu.Lock()
+	a.chats[info.ChannelID] = info
+	a.mu.Unlock()
+}

plugins/channels/telegram/updates_test.go

@@ -0,0 +1,244 @@
+package telegram
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
+)
+
+func TestToUnifiedMessage_PrivateText(t *testing.T) {
+	t.Parallel()
+	a := New(Config{BotToken: "x"})
+	u := telegramUpdate{
+		UpdateID: 1,
+		Message: &telegramMessage{
+			MessageID: 42,
+			From:      &telegramUser{ID: 100, FirstName: "Alice", Username: "alice"},
+			Chat:      telegramChat{ID: 200, Type: "private"},
+			Date:      1700000000,
+			Text:      "hello world",
+		},
+	}
+	msg, ok := a.toUnifiedMessage(u)
+	if !ok {
+		t.Fatal("expected ok")
+	}
+	if msg.Platform != "telegram" {
+		t.Errorf("Platform = %q, want telegram", msg.Platform)
+	}
+	if msg.ChannelType != adapter.ChannelDirect {
+		t.Errorf("ChannelType = %q, want direct", msg.ChannelType)
+	}
+	if msg.ChannelID != "200" {
+		t.Errorf("ChannelID = %q, want 200", msg.ChannelID)
+	}
+	if msg.SenderID != "100" {
+		t.Errorf("SenderID = %q, want 100", msg.SenderID)
+	}
+	if msg.SenderName != "Alice" {
+		t.Errorf("SenderName = %q, want Alice", msg.SenderName)
+	}
+	if msg.Content != "hello world" || msg.ContentType != adapter.ContentText {
+		t.Errorf("content = %q/%q, want hello world/text", msg.Content, msg.ContentType)
+	}
+	if msg.Sender.UserID != "100" || msg.Sender.DisplayName != "Alice" {
+		t.Errorf("Sender actor = %+v", msg.Sender)
+	}
+	if msg.Sender.IsGroupChat() {
+		t.Error("private message should not be a group actor")
+	}
+	wantTS := time.Unix(1700000000, 0)
+	if !msg.Timestamp.Equal(wantTS) {
+		t.Errorf("Timestamp = %v, want %v", msg.Timestamp, wantTS)
+	}
+}
+
+func TestToUnifiedMessage_Group(t *testing.T) {
+	t.Parallel()
+	a := New(Config{BotToken: "x"})
+	u := telegramUpdate{
+		UpdateID: 2,
+		Message: &telegramMessage{
+			MessageID: 7,
+			From:      &telegramUser{ID: 300, FirstName: "Bob"},
+			Chat:      telegramChat{ID: -1000, Type: "supergroup", Title: "Team Chat"},
+			Date:      1700000001,
+			Text:      "ping",
+		},
+	}
+	msg, ok := a.toUnifiedMessage(u)
+	if !ok {
+		t.Fatal("expected ok")
+	}
+	if msg.ChannelType != adapter.ChannelGroup {
+		t.Errorf("ChannelType = %q, want group", msg.ChannelType)
+	}
+	if msg.GroupID != "-1000" || msg.GroupName != "Team Chat" {
+		t.Errorf("group = %q/%q, want -1000/Team Chat", msg.GroupID, msg.GroupName)
+	}
+	if !msg.Sender.IsGroupChat() {
+		t.Error("Sender should be a group actor")
+	}
+	if msg.Sender.GroupID != "-1000" {
+		t.Errorf("Sender.GroupID = %q, want -1000", msg.Sender.GroupID)
+	}
+}
+
+func TestToUnifiedMessage_ImageFileID(t *testing.T) {
+	t.Parallel()
+	a := New(Config{BotToken: "x"})
+	u := telegramUpdate{
+		UpdateID: 3,
+		Message: &telegramMessage{
+			MessageID: 9,
+			From:      &telegramUser{ID: 400, FirstName: "Cara"},
+			Chat:      telegramChat{ID: 500, Type: "private"},
+			Date:      1700000002,
+			Caption:   "look at this",
+			Photo:     []telegramPhotoSize{{FileID: "photo-small", FileSize: 100}, {FileID: "photo-big", FileSize: 2000}},
+		},
+	}
+	msg, ok := a.toUnifiedMessage(u)
+	if !ok {
+		t.Fatal("expected ok")
+	}
+	if msg.ContentType != adapter.ContentImage {
+		t.Errorf("ContentType = %q, want image", msg.ContentType)
+	}
+	if msg.Content != "look at this" {
+		t.Errorf("Content = %q, want caption", msg.Content)
+	}
+	if len(msg.Media) != 1 {
+		t.Fatalf("media length = %d, want 1", len(msg.Media))
+	}
+	m := msg.Media[0]
+	if m.Kind != adapter.MediaImage || m.URL != "photo-big" || m.Size != 2000 {
+		t.Errorf("media = %+v, want image/photo-big/2000", m)
+	}
+}
+
+func TestToUnifiedMessage_VoiceFileID(t *testing.T) {
+	t.Parallel()
+	a := New(Config{BotToken: "x"})
+	u := telegramUpdate{
+		UpdateID: 4,
+		Message: &telegramMessage{
+			MessageID: 10,
+			From:      &telegramUser{ID: 600, FirstName: "Dan"},
+			Chat:      telegramChat{ID: 700, Type: "private"},
+			Date:      1700000003,
+			Voice:     &telegramAudio{FileID: "voice-id", FileSize: 5000, MIMEType: "audio/ogg"},
+		},
+	}
+	msg, ok := a.toUnifiedMessage(u)
+	if !ok {
+		t.Fatal("expected ok")
+	}
+	if msg.ContentType != adapter.ContentVoice {
+		t.Errorf("ContentType = %q, want voice", msg.ContentType)
+	}
+	if len(msg.Media) != 1 || msg.Media[0].Kind != adapter.MediaVoice || msg.Media[0].URL != "voice-id" {
+		t.Errorf("media = %+v, want voice/voice-id", msg.Media)
+	}
+}
+
+func TestToUnifiedMessage_DocumentFileID(t *testing.T) {
+	t.Parallel()
+	a := New(Config{BotToken: "x"})
+	u := telegramUpdate{
+		UpdateID: 5,
+		Message: &telegramMessage{
+			MessageID: 11,
+			From:      &telegramUser{ID: 800, FirstName: "Eve"},
+			Chat:      telegramChat{ID: 900, Type: "private"},
+			Date:      1700000004,
+			Document:  &telegramDocument{FileID: "doc-id", FileSize: 9000, FileName: "report.pdf", MIMEType: "application/pdf"},
+		},
+	}
+	msg, ok := a.toUnifiedMessage(u)
+	if !ok {
+		t.Fatal("expected ok")
+	}
+	if msg.ContentType != adapter.ContentFile {
+		t.Errorf("ContentType = %q, want file", msg.ContentType)
+	}
+	m := msg.Media[0]
+	if m.Kind != adapter.MediaFile || m.URL != "doc-id" || m.Name != "report.pdf" {
+		t.Errorf("media = %+v, want file/doc-id/report.pdf", m)
+	}
+}
+
+func TestToUnifiedMessage_ReplyAndThread(t *testing.T) {
+	t.Parallel()
+	a := New(Config{BotToken: "x"})
+	u := telegramUpdate{
+		UpdateID: 6,
+		Message: &telegramMessage{
+			MessageID:       20,
+			From:            &telegramUser{ID: 1000, FirstName: "Finn"},
+			Chat:            telegramChat{ID: -2000, Type: "group", Title: "Thread Group"},
+			Date:            1700000005,
+			Text:            "a reply",
+			MessageThreadID: 33,
+			ReplyToMessage:  &telegramMessage{MessageID: 19, Text: "original"},
+		},
+	}
+	msg, ok := a.toUnifiedMessage(u)
+	if !ok {
+		t.Fatal("expected ok")
+	}
+	if msg.ThreadID != "33" {
+		t.Errorf("ThreadID = %q, want 33", msg.ThreadID)
+	}
+	if msg.ReplyToID != "19" || msg.ReplyToText != "original" {
+		t.Errorf("reply = %q/%q, want 19/original", msg.ReplyToID, msg.ReplyToText)
+	}
+}
+
+func TestToUnifiedMessage_NoMessage(t *testing.T) {
+	t.Parallel()
+	a := New(Config{BotToken: "x"})
+	u := telegramUpdate{UpdateID: 7}
+	if _, ok := a.toUnifiedMessage(u); ok {
+		t.Fatal("expected ok=false for non-message update")
+	}
+}
+
+func TestListChannels_FromConversations(t *testing.T) {
+	t.Parallel()
+	a := New(Config{BotToken: "x"})
+	a.toUnifiedMessage(telegramUpdate{
+		UpdateID: 8,
+		Message: &telegramMessage{
+			MessageID: 1,
+			From:      &telegramUser{ID: 1, FirstName: "Ada"},
+			Chat:      telegramChat{ID: 10, Type: "private"},
+			Text:      "hi",
+		},
+	})
+	a.toUnifiedMessage(telegramUpdate{
+		UpdateID: 9,
+		Message: &telegramMessage{
+			MessageID: 2,
+			From:      &telegramUser{ID: 2, FirstName: "Bob"},
+			Chat:      telegramChat{ID: -20, Type: "group", Title: "Gang"},
+			Text:      "yo",
+		},
+	})
+	got, err := a.ListChannels(context.Background())
+	if err != nil {
+		t.Fatalf("ListChannels: %v", err)
+	}
+	if len(got) != 2 {
+		t.Fatalf("channels length = %d, want 2: %+v", len(got), got)
+	}
+	// Sort puts -20 (group) before 10 (private) as strings.
+	if got[0].ChannelID != "-20" || got[0].ChannelType != adapter.ChannelGroup || got[0].GroupName != "Gang" {
+		t.Errorf("channels[0] = %+v, want group -20", got[0])
+	}
+	if got[1].ChannelID != "10" || got[1].ChannelType != adapter.ChannelDirect || got[1].ChannelName != "Ada" {
+		t.Errorf("channels[1] = %+v, want direct 10", got[1])
+	}
+}