chatapps

package
v0.20.0 Latest Latest
Warning

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

 Go to latest Published: Mar 5, 2026 License: MIT Imports: 32 Imported by: 0

README

ChatApps: Multi-Platform Connector

The chatapps package provides the bridge between HotPlex's core engine and various chat platforms (Slack, Feishu/Lark, Telegram, Discord, etc.). It normalizes platform-specific events and messages into a unified format for the engine to process.

🏛 Architecture Overview

HotPlex uses an Adapter-based Pipeline architecture to achieve platform neutrality and consistent behavior across different IM apps.

🔄 End-to-End Bidirectional Flow

From the client terminal's perspective, HotPlex operates as a reactive streaming system.

sequenceDiagram
    participant User as 👤 User Terminal
    participant Platform as ☁️ Chat Platform (Slack/Feishu)
    participant Adapter as 🔌 Platform Adapter
    participant Pipeline as ⛓️ Processor Pipeline
    participant Handler as 🧠 Engine Handler
    participant Engine as 🔥 HotPlex Engine

    Note over User, Engine: 🟢 Phase 1: Request (C2S)
    User->>Platform: Sends message/command
    Platform->>Adapter: Webhook Event (JSON)
    Adapter->>Adapter: Parse & Normalize to base.ChatMessage
    Adapter->>Handler: Handle(ctx, msg)
    Handler->>Engine: Execute(ctx, prompt)

    Note over User, Engine: 🔵 Phase 2: Response Streaming (S2C)
    Engine-->>Handler: Stream Token/Event
    Handler->>Pipeline: Process(msg)
    Pipeline->>Pipeline: Filter -> RateLimit -> Format -> Chunk
    Pipeline->>Adapter: SendMessage/UpdateMessage
    Adapter->>Platform: Platform API Call (e.g., chat.update)
    Platform-->>User: Visual Update (Real-time)
    
    Note over User, Engine: ⚪ Phase 3: Finalization
    Engine-->>Handler: Execution Complete
    Handler->>Adapter: Send Session Stats/UI Controls
    Adapter->>Platform: Final Message Render
    Platform-->>User: Session Result
Data Normalization (Event to Message Mapping)

The chatapps layer normalizes raw provider events into a standard "Chat Language" using the base.MessageType Go type. While the underlying values often share names with Engine events, they represent UI Rendering Intents documented in base/types.go.

Provider/Engine Event base.MessageType Constant UI Presentation
thinking MessageTypeThinking [Status Only] Thinking indicators / Bubbles
tool_use MessageTypeToolUse [Status Only] "Executing tool..." indicator
tool_result MessageTypeToolResult [Silent Success] Status only; Errors show blocks
answer MessageTypeAnswer Standard Markdown text / Streaming output
error MessageTypeError Red-themed alert block
plan_mode MessageTypePlanMode Planning phase indicator (Status + Text)
permission_request MessageTypePermissionRequest Interactive Allow/Deny buttons
session_stats MessageTypeSessionStats Usage summary (Tokens, Duration)
danger_block MessageTypeDangerBlock Critical warning with confirmation

[!NOTE] Slack Free Plan Compatibility: Some advanced features (Streaming, Status Bar) require a paid plan or Developer Sandbox. See docs/plans/slack_free_plan_compatibility.md for current tracking of fallback optimizations.

Key Architectural Concepts
  • ProcessorChain: A middleware-style pipeline that processes messages before they are sent or after they are received. Standard processors include:
    • Filter: [Black Hole] Silently drops noise events, unparsed raw outputs, and redundant user reflections within the integration layer.
    • Thread: Manages thread state and caching for multi-step responses to maintain context.
    • FormatConversion: Converts Standard Markdown to platform-specific formats (e.g., Slack Block Kit, Feishu Card).
    • Chunking: Splits long messages to respect platform API limits.
  • Space Folding: A policy where high-volume tool outputs (>2KB) are automatically diverted to thread replies or collapsed, preventing main channel pollution while preserving "Geek Transparency".
  • EngineMessageHandler: The main business logic that connects normalized chat events to the HotPlex Engine.

🛠 Developer Guide

1. Implementing a New Platform Adapter

To add a platform (e.g., whatsapp), create a new package chatapps/whatsapp and follow these patterns:

Phase A: Core Interface (base.ChatAdapter)

The foundation of any adapter is the ChatAdapter interface.

type WhatsAppAdapter struct {
    config  Config
    handler base.MessageHandler // Injected via SetHandler
    logger  *slog.Logger
}

func (a *WhatsAppAdapter) Platform() string { return "whatsapp" }

// Handle incoming platform events (e.g., from webhook)
func (a *WhatsAppAdapter) HandleMessage(ctx context.Context, msg *base.ChatMessage) error {
    if a.handler != nil {
        return a.handler(ctx, msg) // Delegate to Engine
    }
    return nil
}

// Send outgoing formatted messages
func (a *WhatsAppAdapter) SendMessage(ctx context.Context, sessionID string, msg *base.ChatMessage) error {
    // 1. Convert msg (Markdown/Blocks) to WhatsApp format
    // 2. Call WhatsApp Cloud API
}

func (a *WhatsAppAdapter) SetHandler(h base.MessageHandler) { a.handler = h }
Phase B: Webhook Support (base.WebhookProvider)

If the platform uses webhooks, implement this to register routes automatically under /webhook/myplatform/.

func (a *WhatsAppAdapter) WebhookHandler() http.Handler {
    mux := http.NewServeMux()
    mux.HandleFunc("/events", func(w http.ResponseWriter, r *http.Request) {
        // 1. Parse signature & event
        // 2. Convert to base.ChatMessage
        // 3. Call a.HandleMessage(r.Context(), normalizedMsg)
    })
    return mux
}
Phase C: Advanced UI & Resiliency

Implement these optional interfaces to provide a premium experience:

  • base.StatusProvider: Handles "Thinking..." or "Running tool X..." visual indicators.
  • base.MessageOperations: Supports updating/deleting existing messages (critical for streaming and UI updates).
  • base.StreamWriter: A standard io.Writer interface for real-time token streaming.
2. Message Processor Pipeline

The ProcessorChain is a middleware system. You can add global or platform-specific processors.

// Example: Custom Privacy Masking Processor
type PrivacyMasker struct {}
func (p *PrivacyMasker) Name() string { return "privacy_mask" }
func (p *PrivacyMasker) Order() int { return 12 } // Run between RateLimit and Aggregation

func (p *PrivacyMasker) Process(ctx context.Context, msg *base.ChatMessage) (*base.ChatMessage, error) {
    msg.Content = strings.ReplaceAll(msg.Content, "SECRET", "****")
    return msg, nil
}
3. Integration into setup.go

Once implemented, register your adapter in chatapps/setup.go:

// 1. Add to setupPlatform calls
setupPlatform(ctx, "whatsapp", loader, manager, logger, func(pc *PlatformConfig) ChatAdapter {
    // return whatsapp.NewAdapter(...)
})

// 2. Ensure credentials are in .env or config.yaml

🏗 Interaction & Button Handling

For platforms supporting buttons (Slack Blocks, Feishu Cards), use the InteractionManager:

  1. Define Action IDs: Use a structured format {action_name}:{session_id} in buttons.
  2. Handle Callbacks: The adapter should capture button clicks and route them to InteractionManager.HandleAction.
  3. UI Feedback: Use StatusProvider to show "Processing..." while the action executes.

📊 Observability & Metrics

HOTPLEX_CHATAPPS_CONFIG_DIR: Path to platform-specific YAML configs.

Status: Active / Modular
Maintainer: HotPlex Core Team

Documentation

Index

Constants

View Source 
const (
	StatusThinkingLabel       = "🧠 深度推演规划中..."
	StatusToolExecutingLabel  = "🛠️ 正在执行工具 [%s]..."
	StatusResultParsingLabel  = "🧠 正在解析执行结果..."
	StatusErrorLabel          = "❌ 执行过程中发生错误"
	StatusDangerBlockLabel    = "⚠️ 拦截到高危操作，等待确认..."
	StatusProgressLabel       = "⏳ 正在执行后台任务..."
	StatusProgressDoneLabel   = "✅ 后台任务执行完毕"
	StatusStepStartLabel      = "🔍 正在分析执行轨迹..."
	StatusStepFinishLabel     = "✅ 当前任务阶段构建完成"
	StatusPlanModeLabel       = "📝 正在制定作战计划..."
	StatusExitPlanModeLabel   = "📝 作战计划就绪，等待您的批准..."
	StatusAskUserLabel        = "⏳ 等待您提供更多信息..."
	StatusEngineStartingLabel = "🚀 正在唤醒推演引擎..."
	StatusPermissionLabel     = "🛡️ 拦截到高危操作，等待提权审批..."

	// Contextual labels
	StatusToolResultThinkingLabel = "🧠 正在解析执行结果 (耗时: %dms)..."
	StatusSessionStartColdLabel   = "🚀 正在初始化上下文..."
	StatusSessionStartHotLabel    = "🚀 重新连接并恢复上下文..."
	StatusAnswerLabel             = "✍️ 正在生成回答..."
)

UI Status indicator labels for AI Native experience

View Source 
const (
	ParseModeNone     = base.ParseModeNone
	ParseModeMarkdown = base.ParseModeMarkdown
	ParseModeHTML     = base.ParseModeHTML
)
View Source 
const (
	// Metadata keys
	MetadataKeyTransient = "is_transient"
)

Variables

View Source 
var (
	// MessagesAggregatedTotal counts total messages aggregated
	MessagesAggregatedTotal = promauto.NewCounterVec(
		prometheus.CounterOpts{
			Name: "hotplex_aggregator_messages_aggregated_total",
			Help: "Total number of messages that have been aggregated",
		},
		[]string{"event_type", "platform"},
	)

	// MessagesFlushedTotal counts total flushes by reason
	MessagesFlushedTotal = promauto.NewCounterVec(
		prometheus.CounterOpts{
			Name: "hotplex_aggregator_messages_flushed_total",
			Help: "Total number of times the buffer was flushed",
		},
		[]string{"event_type", "platform", "reason"},
	)

	// MessagesDroppedTotal counts dropped messages
	MessagesDroppedTotal = promauto.NewCounterVec(
		prometheus.CounterOpts{
			Name: "hotplex_aggregator_messages_dropped_total",
			Help: "Total number of messages dropped due to buffer overflow",
		},
		[]string{"event_type", "platform", "reason"},
	)

	// BufferSizeGauge tracks current buffer size per platform
	BufferSizeGauge = promauto.NewGaugeVec(
		prometheus.GaugeOpts{
			Name: "hotplex_aggregator_buffer_size",
			Help: "Current number of messages in the buffer per platform",
		},
		[]string{"platform"},
	)

	// BufferDurationHistogram tracks time from first message to flush
	BufferDurationHistogram = promauto.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:    "hotplex_aggregator_buffer_duration_seconds",
			Help:    "Time in seconds from first message arrival to buffer flush",
			Buckets: []float64{0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0},
		},
		[]string{"platform"},
	)

	// MessageSizeHistogram tracks message size distribution
	MessageSizeHistogram = promauto.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:    "hotplex_aggregator_message_size_bytes",
			Help:    "Size distribution of aggregated messages in bytes",
			Buckets: []float64{64, 256, 512, 1024, 2048, 4096, 8192, 16384, 32768},
		},
		[]string{"event_type", "platform"},
	)
)
View Source 
var ErrQueueFull = &QueueError{Message: "queue is full"}

Functions

func GenerateInteractionID added in v0.12.0 

func GenerateInteractionID(sessionID string, timestamp time.Time) string

GenerateInteractionID generates a unique interaction ID.

func GetExtraChunks added in v0.12.0 

func GetExtraChunks(msg *base.ChatMessage) []string

GetExtraChunks returns any extra chunks stored in message metadata. Returns nil if no extra chunks exist.

func IsEnabled added in v0.20.0 

func IsEnabled(configDir string) bool

IsEnabled returns true if ChatApps should be activated based on environment variables or flags. It returns true if any of the following is true: 1. HOTPLEX_CHATAPPS_ENABLED environment variable is "true" 2. configDir parameter is not empty (explicitly set via --config flag) 3. HOTPLEX_CHATAPPS_CONFIG_DIR environment variable is not empty

func IsRetryableError 

func IsRetryableError(err error) bool

IsRetryableError classifies errors as retryable or non-retryable

func RetryWithBackoff 

func RetryWithBackoff(ctx context.Context, config RetryConfig, fn func() error) error

Types

type AdapterManager 

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

func NewAdapterManager 

func NewAdapterManager(logger *slog.Logger) *AdapterManager

func Setup added in v0.11.0 

func Setup(ctx context.Context, logger *slog.Logger, configDir ...string) (http.Handler, *AdapterManager, error)

Setup initializes all enabled ChatApps and their dedicated Engines. It returns an http.Handler that handles all webhook routes. The configDir parameter takes priority over HOTPLEX_CHATAPPS_CONFIG_DIR environment variable.

func (*AdapterManager) GetAdapter 

func (m *AdapterManager) GetAdapter(platform string) (ChatAdapter, bool)

func (*AdapterManager) GetMessageOperations added in v0.15.5 

func (m *AdapterManager) GetMessageOperations(platform string) MessageOperations

GetMessageOperations returns platform-specific message operations interface Returns nil if the platform doesn't support message operations

func (*AdapterManager) GetSessionOperations added in v0.15.5 

func (m *AdapterManager) GetSessionOperations(platform string) SessionOperations

GetSessionOperations returns platform-specific session operations interface Returns nil if the platform doesn't support session operations

func (*AdapterManager) GetStatusProvider added in v0.18.0 

func (m *AdapterManager) GetStatusProvider(platform string) base.StatusProvider

GetStatusProvider returns platform-specific status provider interface Returns nil if the platform doesn't support status operations

func (*AdapterManager) Handler added in v0.11.0 

func (m *AdapterManager) Handler() http.Handler

Handler returns an http.Handler with all adapter webhooks mounted This is a convenience method when you don't need gorilla/mux

func (*AdapterManager) ListPlatforms 

func (m *AdapterManager) ListPlatforms() []string

func (*AdapterManager) NewStreamWriter added in v0.18.0 

func (m *AdapterManager) NewStreamWriter(ctx context.Context, platform, channelID, threadTS string) base.StreamWriter

NewStreamWriter creates a platform-agnostic streaming writer for the given platform Returns nil if the platform doesn't support streaming or adapter not found

func (*AdapterManager) Register 

func (m *AdapterManager) Register(adapter ChatAdapter) error

func (*AdapterManager) RegisterEngine added in v0.11.0 

func (m *AdapterManager) RegisterEngine(eng *engine.Engine)

func (*AdapterManager) RegisterRoutes added in v0.11.0 

func (m *AdapterManager) RegisterRoutes(router *mux.Router)

RegisterRoutes registers all adapter webhooks to a unified router Path format: /webhook/{platform} (e.g., /webhook/telegram, /webhook/discord)

func (*AdapterManager) SendMessage 

func (m *AdapterManager) SendMessage(ctx context.Context, platform, sessionID string, msg *ChatMessage) error

SendMessage sends a message to a specific platform

func (*AdapterManager) StartAll 

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

func (*AdapterManager) StopAll 

func (m *AdapterManager) StopAll() error

func (*AdapterManager) Unregister 

func (m *AdapterManager) Unregister(platform string) error

type AdapterMetrics 

type AdapterMetrics struct {
	MessagesReceived atomic.Int64
	MessagesSent     atomic.Int64
	MessagesFailed   atomic.Int64
	LastMessageAt    atomic.Int64
	Uptime           atomic.Int64
}

func NewAdapterMetrics 

func NewAdapterMetrics() *AdapterMetrics

func (*AdapterMetrics) GetStats 

func (m *AdapterMetrics) GetStats() map[string]int64

func (*AdapterMetrics) RecordFailure 

func (m *AdapterMetrics) RecordFailure()

func (*AdapterMetrics) RecordReceive 

func (m *AdapterMetrics) RecordReceive()

func (*AdapterMetrics) RecordSend 

func (m *AdapterMetrics) RecordSend()

type Attachment 

type Attachment = base.Attachment

type ChatAdapter 

type ChatAdapter = base.ChatAdapter

type ChatMessage 

type ChatMessage = base.ChatMessage

func NewChatMessage added in v0.11.0 

func NewChatMessage(platform, sessionID, userID, content string) *ChatMessage

type ChunkInfo added in v0.12.0 

type ChunkInfo struct {
	TotalChunks  int
	CurrentChunk int
	ThreadTS     string
	ChannelID    string
}

ChunkInfo holds metadata about chunked messages.

func GetChunkInfo added in v0.12.0 

func GetChunkInfo(msg *base.ChatMessage) *ChunkInfo

GetChunkInfo returns the ChunkInfo from message metadata.

type ChunkProcessor added in v0.12.0 

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

ChunkProcessor splits long messages into chunks that fit within platform limits. It uses the Slack chunker for Markdown-aware splitting.

func NewChunkProcessor added in v0.12.0 

func NewChunkProcessor(logger *slog.Logger, opts ChunkProcessorOptions) *ChunkProcessor

NewChunkProcessor creates a new ChunkProcessor.

func (*ChunkProcessor) Name added in v0.12.0 

func (p *ChunkProcessor) Name() string

Name returns the processor name.

func (*ChunkProcessor) Order added in v0.12.0 

func (p *ChunkProcessor) Order() int

Order returns the processor order (runs after format conversion).

func (*ChunkProcessor) Process added in v0.12.0 

func (p *ChunkProcessor) Process(ctx context.Context, msg *base.ChatMessage) (*base.ChatMessage, error)

Process splits the message into chunks if it exceeds the character limit. Returns either a single message or a slice of messages.

type ChunkProcessorOptions added in v0.12.0 

type ChunkProcessorOptions struct {
	MaxChars int // Maximum characters per chunk (default: 4000 for Slack)
}

ChunkProcessorOptions configures the ChunkProcessor.

type CommandCallback added in v0.13.0 

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

CommandCallback handles slash command progress events

func NewCommandCallback added in v0.13.0 

func NewCommandCallback(ctx context.Context, platform, sessionID string, adapters *AdapterManager, logger *slog.Logger, metadata map[string]any) *CommandCallback

NewCommandCallback creates a new command callback

func (*CommandCallback) Handle added in v0.13.0 

func (c *CommandCallback) Handle(eventType string, data any) error

Handle implements event.Callback interface

type ConfigLoader 

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

func NewConfigLoader 

func NewConfigLoader(configDir string, logger *slog.Logger) (*ConfigLoader, error)

func (*ConfigLoader) Close added in v0.11.0 

func (c *ConfigLoader) Close() error

Close stops all hot reload watchers and releases resources.

func (*ConfigLoader) GetConfig 

func (c *ConfigLoader) GetConfig(platform string) *PlatformConfig

func (*ConfigLoader) GetOptions added in v0.11.0 

func (c *ConfigLoader) GetOptions(platform string) map[string]any

func (*ConfigLoader) GetSystemPrompt 

func (c *ConfigLoader) GetSystemPrompt(platform string) string

func (*ConfigLoader) GetTaskInstructions 

func (c *ConfigLoader) GetTaskInstructions(platform string) string

func (*ConfigLoader) HasPlatform 

func (c *ConfigLoader) HasPlatform(platform string) bool

func (*ConfigLoader) Load 

func (c *ConfigLoader) Load(configDir string) error

func (*ConfigLoader) Platforms 

func (c *ConfigLoader) Platforms() []string

func (*ConfigLoader) StartHotReload added in v0.11.0 

func (c *ConfigLoader) StartHotReload(ctx context.Context, configDir string, onReload func(platform string, cfg *PlatformConfig)) error

StartHotReload starts watching all config files for changes and automatically reloads them. The onReload callback is called with the updated PlatformConfig for each platform.

type DingTalkConfig 

type DingTalkConfig struct {
	AppID         string `yaml:"app_id"`
	AppSecret     string `yaml:"app_secret"`
	CallbackToken string `yaml:"callback_token"`
	CallbackKey   string `yaml:"callback_key"`
	MaxMessageLen int    `yaml:"max_message_len"`
}

type DiscordEmbed 

type DiscordEmbed struct {
	Title       string                 `json:"title,omitempty"`
	Description string                 `json:"description,omitempty"`
	URL         string                 `json:"url,omitempty"`
	Color       int                    `json:"color,omitempty"`
	Fields      []DiscordEmbedField    `json:"fields,omitempty"`
	Footer      *DiscordEmbedFooter    `json:"footer,omitempty"`
	Thumbnail   *DiscordEmbedThumbnail `json:"thumbnail,omitempty"`
	Image       *DiscordEmbedImage     `json:"image,omitempty"`
	Timestamp   string                 `json:"timestamp,omitempty"`
}

type DiscordEmbedField 

type DiscordEmbedField struct {
	Name   string `json:"name"`
	Value  string `json:"value"`
	Inline bool   `json:"inline,omitempty"`
}

type DiscordEmbedFooter 

type DiscordEmbedFooter struct {
	Text    string `json:"text"`
	IconURL string `json:"icon_url,omitempty"`
}

type DiscordEmbedImage 

type DiscordEmbedImage struct {
	URL string `json:"url"`
}

type DiscordEmbedThumbnail 

type DiscordEmbedThumbnail struct {
	URL string `json:"url"`
}

type Engine added in v0.15.5 

type Engine interface {
	Execute(ctx context.Context, cfg *types.Config, prompt string, callback event.Callback) error
	CheckDanger(prompt string) (blocked bool, operation, reason string)
	GetSession(sessionID string) (Session, bool)
	Close() error
	GetSessionStats(sessionID string) *SessionStats
	ValidateConfig(cfg *types.Config) error
	StopSession(sessionID string, reason string) error
	ResetSessionProvider(sessionID string)
	SetDangerAllowPaths(paths []string)
	SetDangerBypassEnabled(token string, enabled bool) error
	SetAllowedTools(tools []string)
	SetDisallowedTools(tools []string)
	GetAllowedTools() []string
	GetDisallowedTools() []string
}

Engine abstracts the engine functionality for dependency inversion

type EngineConfig added in v0.11.0 

type EngineConfig struct {
	Timeout         time.Duration `yaml:"timeout"`
	IdleTimeout     time.Duration `yaml:"idle_timeout"`
	WorkDir         string        `yaml:"work_dir"`
	AllowedTools    []string      `yaml:"allowed_tools"`
	DisallowedTools []string      `yaml:"disallowed_tools"`
}

type EngineHolder 

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

EngineHolder holds the Engine instance and configuration for ChatApps integration

func NewEngineHolder 

func NewEngineHolder(opts EngineHolderOptions) (*EngineHolder, error)

NewEngineHolder creates a new EngineHolder with the given options

func (*EngineHolder) GetAdapterManager 

func (h *EngineHolder) GetAdapterManager() *AdapterManager

GetAdapterManager returns the AdapterManager for sending messages

func (*EngineHolder) GetEngine 

func (h *EngineHolder) GetEngine() Engine

GetEngine returns the underlying Engine instance

type EngineHolderOptions 

type EngineHolderOptions struct {
	Logger                     *slog.Logger
	Adapters                   *AdapterManager
	Timeout                    time.Duration
	IdleTimeout                time.Duration
	Namespace                  string
	PermissionMode             string
	DangerouslySkipPermissions bool
	AllowedTools               []string
	DisallowedTools            []string
	DefaultWorkDir             string
	DefaultTaskInstr           string
}

EngineHolderOptions configures the EngineHolder

type EngineMessageHandler 

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

EngineMessageHandler implements MessageHandler and integrates with Engine

func NewEngineMessageHandler 

func NewEngineMessageHandler(engine Engine, adapters *AdapterManager, opts ...EngineMessageHandlerOption) *EngineMessageHandler

NewEngineMessageHandler creates a new EngineMessageHandler

func (*EngineMessageHandler) Close added in v0.17.0 

func (h *EngineMessageHandler) Close()

Close releases resources held by the handler

func (*EngineMessageHandler) Handle 

func (h *EngineMessageHandler) Handle(ctx context.Context, msg *ChatMessage) error

Handle implements EngineMessageHandler

type EngineMessageHandlerOption 

type EngineMessageHandlerOption func(*EngineMessageHandler)

EngineMessageHandlerOption configures the EngineMessageHandler

func WithConfigLoader 

func WithConfigLoader(loader *ConfigLoader) EngineMessageHandlerOption

func WithLogger 

func WithLogger(logger *slog.Logger) EngineMessageHandlerOption

func WithTaskInstrFn 

func WithTaskInstrFn(fn func(sessionID string) string) EngineMessageHandlerOption

func WithWorkDirFn 

func WithWorkDirFn(fn func(sessionID string) string) EngineMessageHandlerOption

type FormatConversionProcessor added in v0.12.0 

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

FormatConversionProcessor converts message content to platform-specific formats

func NewFormatConversionProcessor added in v0.12.0 

func NewFormatConversionProcessor(logger *slog.Logger) *FormatConversionProcessor

NewFormatConversionProcessor creates a new FormatConversionProcessor

func (*FormatConversionProcessor) Name added in v0.12.0 

func (p *FormatConversionProcessor) Name() string

Name returns the processor name

func (*FormatConversionProcessor) Order added in v0.12.0 

func (p *FormatConversionProcessor) Order() int

Order returns the processor order

func (*FormatConversionProcessor) Process added in v0.12.0 

func (p *FormatConversionProcessor) Process(ctx context.Context, msg *base.ChatMessage) (*base.ChatMessage, error)

Process converts message content based on platform

type HealthCheck 

type HealthCheck struct {
	Name      string
	Status    string
	LastCheck time.Time
	LastError string
	Interval  time.Duration
	CheckFunc func() error
}

type HealthChecker 

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

func NewHealthChecker 

func NewHealthChecker(interval time.Duration) *HealthChecker

func (*HealthChecker) GetStatus 

func (h *HealthChecker) GetStatus() map[string]HealthCheck

func (*HealthChecker) Register 

func (h *HealthChecker) Register(check HealthCheck)

func (*HealthChecker) Start 

func (h *HealthChecker) Start()

func (*HealthChecker) Stop 

func (h *HealthChecker) Stop()

type InlineKeyboardButton 

type InlineKeyboardButton struct {
	Text         string `json:"text"`
	URL          string `json:"url,omitempty"`
	CallbackData string `json:"callback_data,omitempty"`
}

type InlineKeyboardMarkup 

type InlineKeyboardMarkup struct {
	InlineKeyboard [][]InlineKeyboardButton `json:"inline_keyboard"`
}

type InteractionCallback added in v0.12.0 

type InteractionCallback func(interaction *PendingInteraction) error

InteractionCallback is a function that handles an interaction callback.

type InteractionManager added in v0.12.0 

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

InteractionManager manages pending interactions for Slack interactive components.

func NewInteractionManager added in v0.12.0 

func NewInteractionManager(logger *slog.Logger, opts InteractionManagerOptions) *InteractionManager

NewInteractionManager creates a new InteractionManager.

func (*InteractionManager) Complete added in v0.12.0 

func (m *InteractionManager) Complete(id string, response *InteractionResponse) error

Complete marks an interaction as completed with a response.

func (*InteractionManager) Count added in v0.12.0 

func (m *InteractionManager) Count() int

Count returns the number of pending interactions.

func (*InteractionManager) Delete added in v0.12.0 

func (m *InteractionManager) Delete(id string)

Delete removes a pending interaction.

func (*InteractionManager) Expire added in v0.12.0 

func (m *InteractionManager) Expire(id string) error

Expire marks an interaction as expired.

func (*InteractionManager) Get added in v0.12.0 

func (m *InteractionManager) Get(id string) (*PendingInteraction, bool)

Get retrieves a pending interaction by ID. Returns nil if not found or expired.

func (*InteractionManager) GetBySession added in v0.12.0 

func (m *InteractionManager) GetBySession(sessionID string) []*PendingInteraction

GetBySession retrieves all pending interactions for a session.

func (*InteractionManager) HandleCallback added in v0.12.0 

func (m *InteractionManager) HandleCallback(interactionID, userID, actionID, callbackData string) error

HandleCallback processes an interaction callback. It looks up the interaction, calls the callback, and removes the interaction.

func (*InteractionManager) PendingCount added in v0.12.0 

func (m *InteractionManager) PendingCount() int

PendingCount returns the number of pending (non-expired) interactions.

func (*InteractionManager) Stop added in v0.12.0 

func (m *InteractionManager) Stop()

Stop stops the cleanup goroutine.

func (*InteractionManager) Store added in v0.12.0 

func (m *InteractionManager) Store(interaction *PendingInteraction) string

Store adds a new pending interaction and returns its ID.

func (*InteractionManager) TotalCount added in v0.12.0 

func (m *InteractionManager) TotalCount() int

TotalCount returns the total number of interactions (including expired).

type InteractionManagerOptions added in v0.12.0 

type InteractionManagerOptions struct {
	CleanupInterval time.Duration // How often to run cleanup (default: 1 min)
	TTL             time.Duration // How long to keep pending interactions (default: 10 min)
}

InteractionManagerOptions configures the InteractionManager.

type InteractionResponse added in v0.12.0 

type InteractionResponse struct {
	ActionID    string    `json:"action_id"`
	Value       string    `json:"value"`
	UserID      string    `json:"user_id"`
	RespondedAt time.Time `json:"responded_at"`
}

InteractionResponse represents the user's response to an interaction.

type InteractionStatus added in v0.12.0 

type InteractionStatus string

InteractionStatus is the status of a pending interaction. 

const (
	InteractionStatusPending   InteractionStatus = "pending"
	InteractionStatusCompleted InteractionStatus = "completed"
	InteractionStatusExpired   InteractionStatus = "expired"
	InteractionStatusCancelled InteractionStatus = "cancelled"
)

type InteractionType added in v0.12.0 

type InteractionType string

InteractionType defines the type of interactive message. 

const (
	// InteractionTypePermission is for Claude Code permission requests (Issue #39).
	InteractionTypePermission InteractionType = "permission"
	// InteractionTypeApproval is for general approval requests (Issue #37).
	InteractionTypeApproval InteractionType = "approval"
	// InteractionTypeSelection is for selection/choice requests.
	InteractionTypeSelection InteractionType = "selection"
)

type LifecycleManager 

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

func NewLifecycleManager 

func NewLifecycleManager() *LifecycleManager

func (*LifecycleManager) OnStart 

func (m *LifecycleManager) OnStart(hook func(ChatAdapter) error)

func (*LifecycleManager) OnStop 

func (m *LifecycleManager) OnStop(hook func(ChatAdapter) error)

func (*LifecycleManager) RegisterAdapter 

func (m *LifecycleManager) RegisterAdapter(adapter ChatAdapter, startPriority int)

func (*LifecycleManager) StartAll 

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

func (*LifecycleManager) StopAll 

func (m *LifecycleManager) StopAll() error

type Logger 

type Logger = slog.Logger

type MessageFilterProcessor added in v0.15.1 

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

MessageFilterProcessor drops noise events before they enter the rest of the chain.

func NewMessageFilterProcessor added in v0.15.1 

func NewMessageFilterProcessor(logger *slog.Logger) *MessageFilterProcessor

NewMessageFilterProcessor creates a new MessageFilterProcessor.

func (*MessageFilterProcessor) Name added in v0.15.1 

func (p *MessageFilterProcessor) Name() string

Name returns the processor name.

func (*MessageFilterProcessor) Order added in v0.15.1 

func (p *MessageFilterProcessor) Order() int

Order returns the processor order – must run before all others.

func (*MessageFilterProcessor) Process added in v0.15.1 

func (p *MessageFilterProcessor) Process(_ context.Context, msg *base.ChatMessage) (*base.ChatMessage, error)

Process drops hidden events by returning nil.

type MessageHandler 

type MessageHandler = base.MessageHandler

type MessageOperations added in v0.15.5 

type MessageOperations = base.MessageOperations

Re-export interfaces from base for convenience

type MessageProcessor added in v0.12.0 

type MessageProcessor interface {
	// Process processes a message and returns the processed message
	// Can return the same message pointer if no modification needed
	// Can return a new message pointer if modification needed
	// Can return an error to stop processing
	Process(ctx context.Context, msg *base.ChatMessage) (*base.ChatMessage, error)

	// Name returns the processor name for logging and debugging
	Name() string

	// Order returns the processor order in the chain (lower = earlier)
	Order() int
}

MessageProcessor defines the interface for processing messages before sending

type MessageQueue 

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

func NewMessageQueue 

func NewMessageQueue(logger *slog.Logger, maxSize, dlqSize, workers int) *MessageQueue

func (*MessageQueue) AddToDLQ 

func (q *MessageQueue) AddToDLQ(msg *QueuedMessage)

AddToDLQ stores a failed message to the Dead Letter Queue

func (*MessageQueue) DLQLen 

func (q *MessageQueue) DLQLen() int

DLQLen returns the number of messages in the Dead Letter Queue

func (*MessageQueue) Dequeue 

func (q *MessageQueue) Dequeue() (*QueuedMessage, bool)

func (*MessageQueue) Enqueue 

func (q *MessageQueue) Enqueue(platform, sessionID string, msg *ChatMessage) error

func (*MessageQueue) GetDLQ 

func (q *MessageQueue) GetDLQ() []*QueuedMessage

GetDLQ returns all messages in the Dead Letter Queue

func (*MessageQueue) Requeue 

func (q *MessageQueue) Requeue(msg *QueuedMessage) error

Requeue adds a message back to the queue for retry

func (*MessageQueue) Size 

func (q *MessageQueue) Size() int

func (*MessageQueue) Start 

func (q *MessageQueue) Start(adapterGetter func(string) (ChatAdapter, bool), sendFn func(context.Context, string, string, *ChatMessage) error)

func (*MessageQueue) Stop 

func (q *MessageQueue) Stop()

type ParseMode 

type ParseMode = base.ParseMode

type PendingInteraction added in v0.12.0 

type PendingInteraction struct {
	ID           string
	SessionID    string
	ChannelID    string
	MessageTS    string
	ActionID     string
	UserID       string
	CallbackData string
	Callback     InteractionCallback
	CreatedAt    time.Time
	ExpiresAt    time.Time
	Type         InteractionType
	ThreadTS     string
	Status       InteractionStatus
	Response     *InteractionResponse
	Metadata     map[string]any
}

PendingInteraction represents a pending interactive action (e.g., button click).

func CreatePendingInteraction added in v0.12.0 

func CreatePendingInteraction(
	sessionID string,
	userID string,
	channelID string,
	interactionType InteractionType,
	metadata map[string]any,
	ttl time.Duration,
) *PendingInteraction

CreatePendingInteraction creates a new PendingInteraction with default values.

func (*PendingInteraction) IsExpired added in v0.12.0 

func (p *PendingInteraction) IsExpired() bool

IsExpired returns true if the interaction has expired.

func (*PendingInteraction) TimeUntilExpiry added in v0.12.0 

func (p *PendingInteraction) TimeUntilExpiry() time.Duration

TimeUntilExpiry returns the duration until the interaction expires.

type PermissionConfig added in v0.17.0 

type PermissionConfig struct {
	DMPolicy              string   `yaml:"dm_policy"`
	GroupPolicy           string   `yaml:"group_policy"`
	BotUserID             string   `yaml:"bot_user_id"`
	AllowedUsers          []string `yaml:"allowed_users"`
	BlockedUsers          []string `yaml:"blocked_users"`
	SlashCommandRateLimit float64  `yaml:"slash_command_rate_limit"`
}

type PlatformConfig 

type PlatformConfig struct {
	Platform         string                  `yaml:"platform"`
	SystemPrompt     string                  `yaml:"system_prompt"`
	TaskInstructions string                  `yaml:"task_instructions"`
	Engine           EngineConfig            `yaml:"engine"`
	Provider         provider.ProviderConfig `yaml:"provider"`
	Security         SecurityConfig          `yaml:"security"`
	DingTalk         DingTalkConfig          `yaml:"dingtalk"`
	WhatsApp         WhatsAppConfig          `yaml:"whatsapp"`
	Options          map[string]any          `yaml:"options,omitempty"`
	SourceFile       string                  `yaml:"-"` // Tracks which file this config was loaded from
}

type ProcessorChain added in v0.12.0 

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

ProcessorChain executes a chain of message processors

func NewDefaultProcessorChain added in v0.12.0 

func NewDefaultProcessorChain(ctx context.Context, logger *slog.Logger) *ProcessorChain

NewDefaultProcessorChain creates a default processor chain with all standard processors

func NewProcessorChain added in v0.12.0 

func NewProcessorChain(processors ...MessageProcessor) *ProcessorChain

NewProcessorChain creates a new processor chain with the given processors Processors are automatically sorted by Order()

func (*ProcessorChain) AddProcessor added in v0.12.0 

func (c *ProcessorChain) AddProcessor(processor MessageProcessor)

AddProcessor adds a processor to the chain and re-sorts Note: This method is thread-safe but should preferably be called during initialization

func (*ProcessorChain) Close added in v0.17.0 

func (c *ProcessorChain) Close()

Close stops all processors that have background goroutines

func (*ProcessorChain) Name added in v0.16.0 

func (c *ProcessorChain) Name() string

Name returns the processor chain name

func (*ProcessorChain) Order added in v0.16.0 

func (c *ProcessorChain) Order() int

Order returns the processor chain order

func (*ProcessorChain) Process added in v0.12.0 

func (c *ProcessorChain) Process(ctx context.Context, msg *base.ChatMessage) (*base.ChatMessage, error)

Process executes all processors in order

func (*ProcessorChain) ResetSession added in v0.15.1 

func (c *ProcessorChain) ResetSession(platform, sessionID string)

ResetSession propagate session reset to all processors that support it

type ProcessorOrder added in v0.12.0 

type ProcessorOrder int

ProcessorOrder defines standard processor ordering 

const (
	// OrderFilter drops noise events before anything else
	OrderFilter ProcessorOrder = 10
	// OrderThread manages thread_ts caching for message chunking
	OrderThread ProcessorOrder = 15
	// OrderFormatConversion converts markdown to platform-specific format
	OrderFormatConversion ProcessorOrder = 40
	// OrderChunk splits long messages for Slack API limits
	OrderChunk ProcessorOrder = 50
)

type QueueError 

type QueueError struct {
	Message string
}

func (*QueueError) Error 

func (e *QueueError) Error() string

type QueuedMessage 

type QueuedMessage struct {
	Platform  string
	SessionID string
	Message   *ChatMessage
	Retries   int
	CreatedAt time.Time
}

type RateLimiter 

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

func NewRateLimiter 

func NewRateLimiter(maxTokens float64, refillRate float64) *RateLimiter

func (*RateLimiter) Allow 

func (r *RateLimiter) Allow() bool

func (*RateLimiter) Wait 

func (r *RateLimiter) Wait(ctx context.Context) error

type RetryConfig 

type RetryConfig struct {
	MaxAttempts int
	BaseDelay   time.Duration
	MaxDelay    time.Duration
}

type RichContent 

type RichContent = base.RichContent

type SecurityConfig added in v0.17.0 

type SecurityConfig struct {
	VerifySignature bool             `yaml:"verify_signature"`
	Permission      PermissionConfig `yaml:"permission"`
}

type Session added in v0.15.5 

type Session interface {
	ID() string
	Status() string
	CreatedAt() time.Time
	IsResumed() bool
}

Session abstracts session state and operations

type SessionOperations added in v0.15.5 

type SessionOperations = base.SessionOperations

Re-export interfaces from base for convenience

type SessionStats added in v0.15.5 

type SessionStats struct {
	SessionID     string
	Status        string
	TotalTokens   int64
	InputTokens   int64
	OutputTokens  int64
	CacheRead     int64
	CacheWrite    int64
	TotalCost     float64
	Duration      time.Duration
	ToolCallCount int
	ErrorCount    int
}

SessionStats holds session statistics

type SlackBlock 

type SlackBlock map[string]any

type StreamAdapter 

type StreamAdapter interface {
	ChatAdapter
	SendStreamMessage(ctx context.Context, sessionID string, msg *ChatMessage) (StreamHandler, error)
	UpdateMessage(ctx context.Context, sessionID, messageID string, msg *ChatMessage) error
}

type StreamCallback 

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

StreamCallback implements event.Callback to receive Engine events and forward to ChatApp

func NewStreamCallback 

func NewStreamCallback(
	ctx context.Context,
	sessionID, platform string,
	adapters *AdapterManager,
	logger *slog.Logger,
	isHot bool,
	metadata map[string]any,
	messageOps MessageOperations,
	sessionOps SessionOperations,
) *StreamCallback

NewStreamCallback creates a new StreamCallback with injected platform-specific operations

func (*StreamCallback) Close added in v0.17.0 

func (c *StreamCallback) Close()

Close releases resources held by the callback

func (*StreamCallback) Handle 

func (c *StreamCallback) Handle(eventType string, data any) error

Handle implements event.Callback

type StreamHandler 

type StreamHandler func(ctx context.Context, sessionID string, chunk string, isFinal bool) error

type ThreadInfo added in v0.12.0 

type ThreadInfo struct {
	ThreadTS     string
	ChannelID    string
	LastActivity time.Time
}

ThreadInfo holds thread-related metadata for a session.

type ThreadProcessor added in v0.12.0 

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

ThreadProcessor manages thread_ts caching for message chunking. It tracks the first message's timestamp for each session to associate subsequent chunked messages in the same thread.

func NewThreadProcessor added in v0.12.0 

func NewThreadProcessor(logger *slog.Logger, opts ThreadProcessorOptions) *ThreadProcessor

NewThreadProcessor creates a new ThreadProcessor.

func (*ThreadProcessor) Delete added in v0.12.0 

func (p *ThreadProcessor) Delete(sessionID string)

Delete removes thread info for a session.

func (*ThreadProcessor) GetThreadTS added in v0.12.0 

func (p *ThreadProcessor) GetThreadTS(sessionID string) string

GetThreadTS returns the stored thread_ts for a session. Returns empty string if no thread info exists.

func (*ThreadProcessor) Name added in v0.12.0 

func (p *ThreadProcessor) Name() string

Name returns the processor name.

func (*ThreadProcessor) Order added in v0.12.0 

func (p *ThreadProcessor) Order() int

Order returns the processor order (runs after rate limit, before aggregation).

func (*ThreadProcessor) Process added in v0.12.0 

func (p *ThreadProcessor) Process(ctx context.Context, msg *base.ChatMessage) (*base.ChatMessage, error)

Process manages thread_ts for the message. For the first message: stores thread_ts from metadata if present. For subsequent messages: attaches stored thread_ts to metadata.

func (*ThreadProcessor) SetThreadTS added in v0.12.0 

func (p *ThreadProcessor) SetThreadTS(sessionID, threadTS, channelID string)

SetThreadTS explicitly sets the thread_ts for a session. This is useful when the first message response provides the thread_ts.

func (*ThreadProcessor) Stop added in v0.12.0 

func (p *ThreadProcessor) Stop()

Stop stops the cleanup goroutine.

type ThreadProcessorOptions added in v0.12.0 

type ThreadProcessorOptions struct {
	CleanupInterval time.Duration // How often to run cleanup (default: 5 min)
	TTL             time.Duration // How long to keep thread info (default: 30 min)
}

ThreadProcessorOptions configures the ThreadProcessor.

type WhatsAppConfig 

type WhatsAppConfig struct {
	PhoneNumberID string `yaml:"phone_number_id"`
	AccessToken   string `yaml:"access_token"`
	VerifyToken   string `yaml:"verify_token"`
	APIVersion    string `yaml:"api_version"`
}

Source Files

View all Source files

Directories

Path Synopsis
Package slack provides a high-performance, AI-native Slack adapter for the HotPlex engine.
 Package slack provides a high-performance, AI-native Slack adapter for the HotPlex engine.

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.