channels

package
v1.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 16, 2026 License: MIT Imports: 31 Imported by: 0

Documentation

Index

Constants

View Source 
const MessageSplitMarker = "<|[SPLIT]|>"

MessageSplitMarker is the delimiter used to split a message into multiple outbound messages. When SplitOnMarker is enabled in config, the Manager will split messages on this marker and send each part as a separate message.

Variables

View Source 
var (
	// ErrNotRunning indicates the channel is not running.
	// Manager will not retry.
	ErrNotRunning = errors.New("channel not running")

	// ErrRateLimit indicates the platform returned a rate-limit response (e.g. HTTP 429).
	// Manager will wait a fixed delay and retry.
	ErrRateLimit = errors.New("rate limited")

	// ErrTemporary indicates a transient failure (e.g. network timeout, 5xx).
	// Manager will use exponential backoff and retry.
	ErrTemporary = errors.New("temporary failure")

	// ErrSendFailed indicates a permanent failure (e.g. invalid chat ID, 4xx non-429).
	// Manager will not retry.
	ErrSendFailed = errors.New("send failed")
)

Functions

func BuildMediaScope 

func BuildMediaScope(channel, chatID, messageID string) string

BuildMediaScope constructs a scope key for media lifecycle tracking.

func ClassifyNetError 

func ClassifyNetError(err error) error

ClassifyNetError wraps a network/timeout error as ErrTemporary.

func ClassifySendError 

func ClassifySendError(statusCode int, rawErr error) error

ClassifySendError wraps a raw error with the appropriate sentinel based on an HTTP status code. Channels that perform HTTP API calls should use this in their Send path.

func GetRegisteredFactoryNames 

func GetRegisteredFactoryNames() []string

GetRegisteredFactoryNames returns a slice of all registered channel factory names.

func InitialAnimatedToolFeedbackContent 

func InitialAnimatedToolFeedbackContent(baseContent string) string

func MaxToolFeedbackAnimationFrameLength 

func MaxToolFeedbackAnimationFrameLength() int

MaxToolFeedbackAnimationFrameLength returns the largest frame suffix length so callers can reserve room before sending messages to length-limited APIs.

func RegisterFactory 

func RegisterFactory(name string, f ChannelFactory)

RegisterFactory registers a named channel factory. Called from subpackage init() functions.

func RegisterSafeFactory 

func RegisterSafeFactory[S any](
	channelType string,
	ctor func(bc *config.Channel, settings *S, bus *bus.MessageBus) (Channel, error),
)

RegisterSafeFactory is a convenience wrapper that handles GetDecoded() error checking and type assertion, reducing boilerplate in channel init() functions.

Usage: 

func init() {
    channels.RegisterSafeFactory(config.ChannelTelegram,
        func(bc *config.Channel, c *config.TelegramSettings, b *bus.MessageBus) (channels.Channel, error) {
            return NewTelegramChannel(bc, c, b)
        })
}

func SplitByMarker 

func SplitByMarker(content string) []string

SplitByMarker splits a message by the MessageSplitMarker and returns the parts. Empty parts (including from consecutive markers) are filtered out. If no marker is found, returns a single-element slice containing the original content.

func SplitMessage 

func SplitMessage(content string, maxLen int) []string

SplitMessage splits long messages into chunks, preserving code block integrity. The maxLen parameter is measured in runes (Unicode characters), not bytes. The function reserves a buffer (10% of maxLen, min 50) to leave room for closing code blocks, but may extend to maxLen when needed. Call SplitMessage with the full text content and the maximum allowed length of a single message; it returns a slice of message chunks that each respect maxLen and avoid splitting fenced code blocks.

Types

type BaseChannel 

type BaseChannel struct {
	// contains filtered or unexported fields
}

func NewBaseChannel 

func NewBaseChannel(
	name string,
	config any,
	bus *bus.MessageBus,
	allowList []string,
	opts ...BaseChannelOption,
) *BaseChannel

func (*BaseChannel) GetMediaStore 

func (c *BaseChannel) GetMediaStore() media.MediaStore

GetMediaStore returns the injected MediaStore (may be nil).

func (*BaseChannel) GetPlaceholderRecorder 

func (c *BaseChannel) GetPlaceholderRecorder() PlaceholderRecorder

GetPlaceholderRecorder returns the injected PlaceholderRecorder (may be nil).

func (*BaseChannel) HandleInboundContext 

func (c *BaseChannel) HandleInboundContext(
	ctx context.Context,
	deliveryChatID, content string,
	media []string,
	inboundCtx bus.InboundContext,
	senderOpts ...bus.SenderInfo,
) error

HandleInboundContext publishes a normalized inbound message using only the structured context.

func (*BaseChannel) HandleMessageWithContext 

func (c *BaseChannel) HandleMessageWithContext(
	ctx context.Context,
	deliveryChatID, content string,
	media []string,
	inboundCtx bus.InboundContext,
	senderOpts ...bus.SenderInfo,
) error

func (*BaseChannel) IsAllowed 

func (c *BaseChannel) IsAllowed(senderID string) bool

func (*BaseChannel) IsAllowedSender 

func (c *BaseChannel) IsAllowedSender(sender bus.SenderInfo) bool

IsAllowedSender checks whether a structured SenderInfo is permitted by the allow-list. It delegates to identity.MatchAllowed for each entry, providing unified matching across all legacy formats and the new canonical "platform:id" format.

func (*BaseChannel) IsRunning 

func (c *BaseChannel) IsRunning() bool

func (*BaseChannel) MaxMessageLength 

func (c *BaseChannel) MaxMessageLength() int

MaxMessageLength returns the maximum message length (in runes) for this channel. A value of 0 means no limit.

func (*BaseChannel) Name 

func (c *BaseChannel) Name() string

func (*BaseChannel) ReasoningChannelID 

func (c *BaseChannel) ReasoningChannelID() string

func (*BaseChannel) SetMediaStore 

func (c *BaseChannel) SetMediaStore(s media.MediaStore)

SetMediaStore injects a MediaStore into the channel.

func (*BaseChannel) SetName 

func (c *BaseChannel) SetName(name string)

SetName updates the channel name. Used by the manager after channel creation to ensure the name matches the config key (which may differ from the type).

func (*BaseChannel) SetOwner 

func (c *BaseChannel) SetOwner(ch Channel)

SetOwner injects the concrete channel that embeds this BaseChannel. This allows HandleMessage to auto-trigger TypingCapable / ReactionCapable / PlaceholderCapable.

func (*BaseChannel) SetPlaceholderRecorder 

func (c *BaseChannel) SetPlaceholderRecorder(r PlaceholderRecorder)

SetPlaceholderRecorder injects a PlaceholderRecorder into the channel.

func (*BaseChannel) SetRunning 

func (c *BaseChannel) SetRunning(running bool)

func (*BaseChannel) ShouldRespondInGroup 

func (c *BaseChannel) ShouldRespondInGroup(isMentioned bool, content string) (bool, string)

ShouldRespondInGroup determines whether the bot should respond in a group chat. Each channel is responsible for:

  1. Detecting isMentioned (platform-specific)
  2. Stripping bot mention from content (platform-specific)
  3. Calling this method to get the group response decision

Logic:

  • If isMentioned → always respond
  • If mention_only configured and not mentioned → ignore
  • If prefixes configured → respond if content starts with any prefix (strip it)
  • If prefixes configured but no match and not mentioned → ignore
  • Otherwise (no group_trigger configured) → respond to all (permissive default)

type BaseChannelOption 

type BaseChannelOption func(*BaseChannel)

BaseChannelOption is a functional option for configuring a BaseChannel.

func WithGroupTrigger 

func WithGroupTrigger(gt config.GroupTriggerConfig) BaseChannelOption

WithGroupTrigger sets the group trigger configuration for a channel.

func WithMaxMessageLength 

func WithMaxMessageLength(n int) BaseChannelOption

WithMaxMessageLength sets the maximum message length (in runes) for a channel. Messages exceeding this limit will be automatically split by the Manager. A value of 0 means no limit.

func WithReasoningChannelID 

func WithReasoningChannelID(id string) BaseChannelOption

WithReasoningChannelID sets the reasoning channel ID where thoughts should be sent.

type Channel 

type Channel interface {
	Name() string
	Start(ctx context.Context) error
	Stop(ctx context.Context) error
	Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error)
	IsRunning() bool
	IsAllowed(senderID string) bool
	IsAllowedSender(sender bus.SenderInfo) bool
	ReasoningChannelID() string
}

type ChannelFactory 

type ChannelFactory func(channelName, channelType string, cfg *config.Config, bus *bus.MessageBus) (Channel, error)

ChannelFactory is a constructor function that creates a Channel from config and message bus. Each channel subpackage registers one or more factories via init(). channelName is the config map key for this channel instance (may differ from the channel type). channelType is the channel type string used to look up the Channel config.

type ChannelLifecyclePayload 

type ChannelLifecyclePayload struct {
	Type  string `json:"type,omitempty"`
	Error string `json:"error,omitempty"`
}

ChannelLifecyclePayload describes channel lifecycle runtime events.

type ChannelOutboundPayload 

type ChannelOutboundPayload struct {
	Media            bool     `json:"media,omitempty"`
	ContentLen       int      `json:"content_len,omitempty"`
	MessageIDs       []string `json:"message_ids,omitempty"`
	ReplyToMessageID string   `json:"reply_to_message_id,omitempty"`
	Error            string   `json:"error,omitempty"`
	Retries          int      `json:"retries,omitempty"`
}

ChannelOutboundPayload describes channel outbound message runtime events.

type CommandRegistrarCapable 

type CommandRegistrarCapable interface {
	RegisterCommands(ctx context.Context, defs []commands.Definition) error
}

CommandRegistrarCapable is implemented by channels that can register command menus with their upstream platform (e.g. Telegram BotCommand). Channels that do not support platform-level command menus can ignore it.

type HealthChecker 

type HealthChecker interface {
	HealthPath() string
	HealthHandler(w http.ResponseWriter, r *http.Request)
}

HealthChecker is an optional interface for channels that expose a health check endpoint on the shared HTTP server.

type Manager 

type Manager struct {
	// contains filtered or unexported fields
}

func NewManager 

func NewManager(
	cfg *config.Config,
	messageBus *bus.MessageBus,
	store media.MediaStore,
	opts ...ManagerOption,
) (*Manager, error)

func (*Manager) DismissToolFeedback 

func (m *Manager) DismissToolFeedback(
	ctx context.Context, channelName, chatID string, outboundCtx *bus.InboundContext,
)

DismissToolFeedback clears any tracked tool feedback animation for the given channel/chat. This is called when a turn ends without a final response (e.g., ResponseHandled tools) to stop orphaned animation goroutines. outboundCtx carries topic/thread info for channels that use scoped tracker keys (e.g., Telegram forum topics); may be nil for non-topic channels.

func (*Manager) GetChannel 

func (m *Manager) GetChannel(name string) (Channel, bool)

func (*Manager) GetEnabledChannels 

func (m *Manager) GetEnabledChannels() []string

func (*Manager) GetStatus 

func (m *Manager) GetStatus() map[string]any

func (*Manager) GetStreamer 

func (m *Manager) GetStreamer(ctx context.Context, channelName, chatID, sessionKey string) (bus.Streamer, bool)

GetStreamer implements bus.StreamDelegate. It checks if the named channel supports streaming and returns a Streamer.

func (*Manager) InvokeTypingStop 

func (m *Manager) InvokeTypingStop(channel, chatID string)

InvokeTypingStop invokes the registered typing stop function for the given channel and chatID. It is safe to call even when no typing indicator is active (no-op). Used by the agent loop to stop typing when processing completes (success, error, or panic), regardless of whether an outbound message is published.

func (*Manager) RecordPlaceholder 

func (m *Manager) RecordPlaceholder(channel, chatID, placeholderID string)

RecordPlaceholder registers a placeholder message for later editing. Implements PlaceholderRecorder.

func (*Manager) RecordReactionUndo 

func (m *Manager) RecordReactionUndo(channel, chatID string, undo func())

RecordReactionUndo registers a reaction undo function for later invocation. Implements PlaceholderRecorder.

func (*Manager) RecordTypingStop 

func (m *Manager) RecordTypingStop(channel, chatID string, stop func())

RecordTypingStop registers a typing stop function for later invocation. Implements PlaceholderRecorder.

func (*Manager) RegisterChannel 

func (m *Manager) RegisterChannel(name string, channel Channel)

func (*Manager) Reload 

func (m *Manager) Reload(ctx context.Context, cfg *config.Config) error

Reload updates the config reference without restarting channels. This is used when channel config hasn't changed but other parts of the config have.

func (*Manager) SendMedia 

func (m *Manager) SendMedia(ctx context.Context, msg bus.OutboundMediaMessage) error

SendMedia sends outbound media synchronously through the channel worker's rate limiter and retry logic. It blocks until the media is delivered (or all retries are exhausted), which preserves ordering when later agent behavior depends on actual media delivery.

func (*Manager) SendMessage 

func (m *Manager) SendMessage(ctx context.Context, msg bus.OutboundMessage) error

SendMessage sends an outbound message synchronously through the channel worker's rate limiter and retry logic. It blocks until the message is delivered (or all retries are exhausted), which preserves ordering when a subsequent operation depends on the message having been sent.

func (*Manager) SendPlaceholder 

func (m *Manager) SendPlaceholder(ctx context.Context, channel, chatID string) bool

SendPlaceholder sends a "Thinking…" placeholder for the given channel/chatID and records it for later editing. Returns true if a placeholder was sent.

func (*Manager) SendToChannel 

func (m *Manager) SendToChannel(ctx context.Context, channelName, chatID, content string) error

func (*Manager) SetMediaStore 

func (m *Manager) SetMediaStore(store media.MediaStore)

SetMediaStore updates the store used by the manager and every channel that accepts media store injection. Gateway reload creates a fresh store, so keeping existing channels on the same store as the agent is required for inbound media refs to remain resolvable after reload.

func (*Manager) SetupHTTPServer 

func (m *Manager) SetupHTTPServer(addr string, healthServer *health.Server)

SetupHTTPServer creates a shared HTTP server with the given listen address. It registers health endpoints from the health server and discovers channels that implement WebhookHandler and/or HealthChecker to register their handlers.

func (*Manager) SetupHTTPServerListeners 

func (m *Manager) SetupHTTPServerListeners(listeners []net.Listener, addr string, healthServer *health.Server)

SetupHTTPServerListeners creates a shared HTTP server on pre-opened listeners. When listeners is empty it falls back to Addr-based ListenAndServe behavior.

func (*Manager) StartAll 

func (m *Manager) StartAll(ctx context.Context) error

func (*Manager) StopAll 

func (m *Manager) StopAll(ctx context.Context) error

func (*Manager) UnregisterChannel 

func (m *Manager) UnregisterChannel(name string)

type ManagerOption 

type ManagerOption func(*Manager)

ManagerOption configures a channel Manager.

func WithRuntimeEvents 

func WithRuntimeEvents(eventBus runtimeevents.Bus) ManagerOption

WithRuntimeEvents injects the runtime event bus used for channel observations.

type MediaSender 

type MediaSender interface {
	SendMedia(ctx context.Context, msg bus.OutboundMediaMessage) ([]string, error)
}

MediaSender is an optional interface for channels that can send media attachments (images, files, audio, video). Manager discovers channels implementing this interface via type assertion and routes OutboundMediaMessage to them.

type MessageDeleter 

type MessageDeleter interface {
	DeleteMessage(ctx context.Context, chatID string, messageID string) error
}

MessageDeleter — channels that can delete a message by ID.

type MessageEditor 

type MessageEditor interface {
	EditMessage(ctx context.Context, chatID string, messageID string, content string) error
}

MessageEditor — channels that can edit an existing message. messageID is always string; channels convert platform-specific types internally.

type MessageEditorWithPayload 

type MessageEditorWithPayload interface {
	EditMessageWithPayload(
		ctx context.Context,
		chatID string,
		messageID string,
		payload map[string]any,
	) error
}

MessageEditorWithPayload extends MessageEditor for channels that can update structured message metadata in addition to plain text content.

type MessageLengthProvider 

type MessageLengthProvider interface {
	MaxMessageLength() int
}

MessageLengthProvider is an opt-in interface that channels implement to advertise their maximum message length. The Manager uses this via type assertion to decide whether to split outbound messages.

type PlaceholderCapable 

type PlaceholderCapable interface {
	SendPlaceholder(ctx context.Context, chatID string) (messageID string, err error)
}

PlaceholderCapable — channels that can send a placeholder message (e.g. "Thinking... 💭") that will later be edited to the actual response. The channel MUST also implement MessageEditor for the placeholder to be useful. SendPlaceholder returns the platform message ID of the placeholder so that Manager.preSend can later edit it via MessageEditor.EditMessage.

type PlaceholderRecorder 

type PlaceholderRecorder interface {
	RecordPlaceholder(channel, chatID, placeholderID string)
	RecordTypingStop(channel, chatID string, stop func())
	RecordReactionUndo(channel, chatID string, undo func())
}

PlaceholderRecorder is injected into channels by Manager. Channels call these methods on inbound to register typing/placeholder state. Manager uses the registered state on outbound to stop typing and edit placeholders.

type ReactionCapable 

type ReactionCapable interface {
	ReactToMessage(ctx context.Context, chatID, messageID string) (undo func(), err error)
}

ReactionCapable — channels that can add a reaction (e.g. 👀) to an inbound message. ReactToMessage adds a reaction and returns an undo function to remove it. The undo function MUST be idempotent and safe to call multiple times.

type Streamer 

type Streamer = bus.Streamer

Streamer is defined in pkg/bus to avoid circular imports. This alias keeps channel implementations using channels.Streamer unchanged.

type StreamingCapable 

type StreamingCapable interface {
	BeginStream(ctx context.Context, chatID string) (Streamer, error)
}

StreamingCapable — channels that can show partial LLM output in real-time. The channel SHOULD gracefully degrade if the platform rejects streaming (e.g. Telegram bot without forum mode). In that case, Update becomes a no-op and Finalize still delivers the final message.

type ToolFeedbackAnimator 

type ToolFeedbackAnimator struct {
	// contains filtered or unexported fields
}

func NewToolFeedbackAnimator 

func NewToolFeedbackAnimator(
	editFn func(ctx context.Context, chatID, messageID, content string) error,
) *ToolFeedbackAnimator

func (*ToolFeedbackAnimator) Clear 

func (a *ToolFeedbackAnimator) Clear(chatID string)

func (*ToolFeedbackAnimator) Current 

func (a *ToolFeedbackAnimator) Current(chatID string) (string, bool)

func (*ToolFeedbackAnimator) Record 

func (a *ToolFeedbackAnimator) Record(chatID, messageID, content string)

func (*ToolFeedbackAnimator) StopAll 

func (a *ToolFeedbackAnimator) StopAll()

func (*ToolFeedbackAnimator) Take 

func (a *ToolFeedbackAnimator) Take(chatID string) (string, string, bool)

func (*ToolFeedbackAnimator) Update 

func (a *ToolFeedbackAnimator) Update(ctx context.Context, chatID, content string) (string, bool, error)

Update edits an existing tracked feedback message. If the edit fails, the previous feedback state is restored so callers can retry without orphaning the old progress message.

type TypingCapable 

type TypingCapable interface {
	StartTyping(ctx context.Context, chatID string) (stop func(), err error)
}

TypingCapable — channels that can show a typing/thinking indicator. StartTyping begins the indicator and returns a stop function. The stop function MUST be idempotent and safe to call multiple times.

type VoiceCapabilities 

type VoiceCapabilities struct {
	ASR bool
	TTS bool
}

VoiceCapabilities describes whether ASR (speech-to-text) and TTS (text-to-speech) are available for a channel under the current configuration.

func DetectVoiceCapabilities 

func DetectVoiceCapabilities(channelName string, ch Channel, asrAvailable bool, ttsAvailable bool) VoiceCapabilities

DetectVoiceCapabilities returns ASR/TTS availability for a channel, gated by whether providers are configured.

type VoiceCapabilityProvider 

type VoiceCapabilityProvider interface {
	VoiceCapabilities() VoiceCapabilities
}

VoiceCapabilityProvider is an optional interface for channels that want to explicitly declare their ASR/TTS support.

type WebhookHandler 

type WebhookHandler interface {
	// WebhookPath returns the path to mount this handler on the shared server.
	// Examples: "/webhook/line", "/webhook/wecom"
	WebhookPath() string
	http.Handler // ServeHTTP(w http.ResponseWriter, r *http.Request)
}

WebhookHandler is an optional interface for channels that receive messages via HTTP webhooks. Manager discovers channels implementing this interface and registers them on the shared HTTP server.

Source Files

View all Source files

Directories

Path Synopsis
Package deltachat implements a Facet Studio channel for Delta Chat, an email-based, end-to-end encrypted messenger.
 Package deltachat implements a Facet Studio channel for Delta Chat, an email-based, end-to-end encrypted messenger.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.