Documentation
¶
Overview ¶
Package chat implements the core of the pigo chat gateway (D27): platform-agnostic messages, the synchronous turn processor with its durable turn ledger, a preview coalescer, and a local single-process runner with a durable spool.
Platform adapters (Telegram, WhatsApp) live in subpackages and plug in via the Adapter and Delivery interfaces; agent sessions are supplied by a SessionProvider such as NewLocalProvider.
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ErrRejected = errors.New("chat: message rejected")
ErrRejected marks a permanently unprocessable message (unauthorized sender, unknown platform): redelivery can never succeed, so durable runners must ack and drop instead of retrying. Matched with errors.Is.
Functions ¶
Types ¶
type Adapter ¶
type Adapter interface {
// Platform returns the platform name matched against [Message.Platform].
Platform() string
// Account returns the bot/business account identity matched against
// [Message.Account]. An empty string makes the adapter the wildcard for
// its platform: it receives every message no more specific adapter
// claims. Registering several adapters for one platform requires
// distinct accounts.
Account() string
// NewDelivery creates the output surface for one turn. A non-empty
// resumePreviewID signals crash recovery: Finalize must edit that
// message instead of sending a new one.
NewDelivery(key ConversationKey, replyTo string, resumePreviewID string) Delivery
// Download resolves an attachment reference to its content and MIME type.
Download(ctx context.Context, ref AttachmentRef) (io.ReadCloser, string, error)
}
Adapter binds one platform (Telegram, WhatsApp, ...) to the processor.
type AttachmentRef ¶
type AttachmentRef struct {
// Kind is one of "photo", "document", "audio", "video", "voice".
Kind string `json:"kind"`
// ID is the platform handle (telegram file_id / whatsapp media id).
ID string `json:"id"`
Name string `json:"name,omitempty"`
MIME string `json:"mime,omitempty"`
Size int64 `json:"size,omitempty"`
}
AttachmentRef is an opaque platform reference to an inbound media object. It never carries credentials or open handles; adapters resolve it on demand via Adapter.Download.
type Conversation ¶
type Conversation struct {
Session *codingagent.AgentSession
// Manager receives ledger writes and serves raw entry reads.
Manager *sessionstore.SessionManager
// Close persists and releases the conversation; it must be called
// exactly once.
Close func(ctx context.Context) error
}
Conversation is exclusive ownership of one hydrated agent session.
type ConversationKey ¶
type ConversationKey struct {
Tenant string
Platform string
Account string
ChatID string
ThreadID string
}
ConversationKey identifies one conversation across platforms. All fields are optional except Platform, Account, and ChatID.
func (ConversationKey) String ¶
func (k ConversationKey) String() string
String returns a stable, filesystem- and partition-safe join of the key segments. The encoding is injective: distinct keys never collide.
type Delivery ¶
type Delivery interface {
// Typing signals a best-effort, repeatable typing indicator.
Typing(ctx context.Context) error
// Preview creates or edits the persistent preview message.
Preview(ctx context.Context, text string) error
// PreviewID returns the platform id of the preview message, or "" until
// a preview exists.
PreviewID() string
// Finalize delivers the final text: it edits the preview when possible
// and chunks long text into follow-up messages.
Finalize(ctx context.Context, text string) (Receipt, error)
// Notify sends a small out-of-band notice (/status output, errors).
Notify(ctx context.Context, text string) error
}
Delivery is one turn's output surface, created by the Adapter. Calls are serialized — never concurrent, with happens-before edges between them — but not single-goroutine: Preview and PreviewID arrive from the per-turn preview renderer goroutine, while Typing, Finalize, and Notify run on the turn goroutine. Adapters must not assume goroutine identity across calls.
type Handler ¶
Handler consumes one message synchronously. *Processor implements it.
type Local ¶
type Local struct {
// contains filtered or unexported fields
}
Local is a durable single-process spool plus a keyed FIFO dispatcher: per-key FIFO order, at most one in-flight Handle per key, a global worker pool, and replay-with-compaction on boot. /stop messages bypass the keyed queue so they can preempt the in-flight turn they target, and messages the handler rejects permanently (ErrRejected) are acked and dropped instead of retried.
ponytail: single-process spool; swap Publish for a broker in clustered deployments.
func NewLocal ¶
NewLocal opens (or creates) the spool at spoolPath, replays unacked messages, compacts the file, and starts the dispatcher.
type LocalProvider ¶
type LocalProvider struct {
// contains filtered or unexported fields
}
LocalProvider maps conversation keys to per-conversation session directories under a root, resuming the most recent session file for a key or creating a new one. One ModelRegistry and one SettingsManager are shared across all sessions.
func NewLocalProvider ¶
func NewLocalProvider(root string, opts ...LocalProviderOption) (*LocalProvider, error)
NewLocalProvider creates a single-process session provider rooted at root. Sessions are constructed with tools disabled unless a WithSessionOptions hook overrides that explicitly.
func (*LocalProvider) Acquire ¶
func (p *LocalProvider) Acquire(_ context.Context, key ConversationKey) (*Conversation, error)
Acquire implements SessionProvider. It errors when the conversation is already held; release goes through the returned Conversation's Close.
func (*LocalProvider) SessionDir ¶
func (p *LocalProvider) SessionDir(key ConversationKey) string
SessionDir returns the sanitized per-conversation session directory for key.
type LocalProviderOption ¶
type LocalProviderOption func(*LocalProvider)
LocalProviderOption configures NewLocalProvider.
func WithAgentDir ¶
func WithAgentDir(dir string) LocalProviderOption
WithAgentDir overrides the global agent config directory used for the shared model registry and settings. Defaults to ~/.pi/agent.
func WithSessionOptions ¶
func WithSessionOptions(hook func(key ConversationKey, o *codingagent.AgentSessionOptions)) LocalProviderOption
WithSessionOptions installs a hook that can adjust the agent session options per conversation before construction — the sanctioned way to wire tools, models, or stream backends. Without it, sessions are created with all tools disabled (NoTools "all").
type Message ¶
type Message struct {
// EventID is platform-unique and stable across redelivery; it is the
// dedupe key for the turn ledger.
EventID string `json:"eventId"`
Tenant string `json:"tenant,omitempty"`
Platform string `json:"platform"`
Account string `json:"account"`
ChatID string `json:"chatId"`
ThreadID string `json:"threadId,omitempty"`
// ChatType classifies the conversation: "dm" (one-to-one), "group"
// (multi-party), or "channel" (broadcast). Group messages carry sender
// attribution in the prompt.
ChatType string `json:"chatType,omitempty"`
SenderID string `json:"senderId,omitempty"`
SenderName string `json:"senderName,omitempty"`
Text string `json:"text,omitempty"`
ReplyToID string `json:"replyToId,omitempty"`
Attachments []AttachmentRef `json:"attachments,omitempty"`
SentAt time.Time `json:"sentAt,omitzero"`
}
Message is one normalized inbound platform message. It is fully JSON-serializable so it can cross a durable queue.
func (Message) Key ¶
func (m Message) Key() ConversationKey
Key returns the conversation key this message belongs to.
type Options ¶
type Options struct {
// Sessions provides exclusive conversation ownership. Required.
Sessions SessionProvider
// Adapters lists one adapter per platform. Required, unique platforms.
Adapters []Adapter
// Authorize gates every inbound message. Required; pass [AllowAll] to
// opt out explicitly.
Authorize func(Message) error
// MaxConcurrent bounds concurrent turns globally. Default 64.
MaxConcurrent int
// PreviewInterval is the preview edit cadence. Default 1s.
PreviewInterval time.Duration
// TurnTimeout guards each turn's context. Default 10m.
TurnTimeout time.Duration
// Logger receives non-fatal diagnostics. Optional.
Logger *slog.Logger
}
Options configures New.
type Processor ¶
type Processor struct {
// contains filtered or unexported fields
}
Processor drives synchronous, per-conversation-serialized turns. Handle returns only once the turn is delivered (or failed), so the external queue acks after it returns and the queue itself is the buffer.
ponytail: FIFO turns per key; Steer-into-active-run if users demand mid-turn injection.
type SessionProvider ¶
type SessionProvider interface {
Acquire(ctx context.Context, key ConversationKey) (*Conversation, error)
}
SessionProvider hands out exclusive conversation ownership. The local JSONL provider is single-process; cluster providers must fence externally.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package discord implements the Discord adapter for the pigo chat gateway: Gateway (WebSocket) ingress over the internal RFC 6455 client — hello/heartbeat/identify, READY capture, resume-first reconnects — plus REST delivery with a refreshed typing indicator, streamed preview edits, chunked finalization with mass-mention suppression, and pre-signed attachment download.
|
Package discord implements the Discord adapter for the pigo chat gateway: Gateway (WebSocket) ingress over the internal RFC 6455 client — hello/heartbeat/identify, READY capture, resume-first reconnects — plus REST delivery with a refreshed typing indicator, streamed preview edits, chunked finalization with mass-mention suppression, and pre-signed attachment download. |
|
examples
|
|
|
localbot
command
Command localbot runs a minimal Telegram chat gateway: long-poll ingress, the durable local spool, and per-conversation agent sessions under a local data directory.
|
Command localbot runs a minimal Telegram chat gateway: long-poll ingress, the durable local spool, and per-conversation agent sessions under a local data directory. |
|
Package googlechat implements the Google Chat adapter for the pigo chat gateway: an HTTP-endpoint Chat app with bearer-JWT-verified ingress, service-account authenticated delivery, and authenticated media download.
|
Package googlechat implements the Google Chat adapter for the pigo chat gateway: an HTTP-endpoint Chat app with bearer-JWT-verified ingress, service-account authenticated delivery, and authenticated media download. |
|
internal
|
|
|
graphhook
Package graphhook holds the Meta Graph webhook plumbing shared by the Cloud API chat adapters (WhatsApp today, Messenger next): the one-time hub.challenge subscribe handshake and the X-Hub-Signature-256 raw-body HMAC check.
|
Package graphhook holds the Meta Graph webhook plumbing shared by the Cloud API chat adapters (WhatsApp today, Messenger next): the one-time hub.challenge subscribe handshake and the X-Hub-Signature-256 raw-body HMAC check. |
|
runechunk
Package runechunk splits platform text at readable rune boundaries.
|
Package runechunk splits platform text at readable rune boundaries. |
|
wsclient
Package wsclient implements a minimal RFC 6455 WebSocket client on the standard library alone: client role only, no extensions, no compression, no subprotocol negotiation.
|
Package wsclient implements a minimal RFC 6455 WebSocket client on the standard library alone: client role only, no extensions, no compression, no subprotocol negotiation. |
|
Package messenger implements the Facebook Messenger (Meta Messenger Platform for Pages) adapter for the pigo chat gateway: webhook ingress over the Graph "page" object (hub handshake + HMAC-signed entry[].messaging[] events), final-message delivery via the Send API, sender-action typing, and direct-URL attachment download.
|
Package messenger implements the Facebook Messenger (Meta Messenger Platform for Pages) adapter for the pigo chat gateway: webhook ingress over the Graph "page" object (hub handshake + HMAC-signed entry[].messaging[] events), final-message delivery via the Send API, sender-action typing, and direct-URL attachment download. |
|
Package slack implements the Slack adapter for the pigo chat gateway: Events API ingress over HTTP (v0 request signing, url_verification handshake, echo filtering), Web API delivery with a bot token (streamed preview edits via chat.update, mrkdwn finalization with chunking), and authenticated file download.
|
Package slack implements the Slack adapter for the pigo chat gateway: Events API ingress over HTTP (v0 request signing, url_verification handshake, echo filtering), Web API delivery with a bot token (streamed preview edits via chat.update, mrkdwn finalization with chunking), and authenticated file download. |
|
Package teams implements the Microsoft Teams (Bot Framework) adapter for the pigo chat gateway, speaking the raw connector REST API with no SDK: JWT-validated webhook ingress, typing plus final-only delivery, markdown-subset formatting with UTF-16 chunking, and authenticated attachment download.
|
Package teams implements the Microsoft Teams (Bot Framework) adapter for the pigo chat gateway, speaking the raw connector REST API with no SDK: JWT-validated webhook ingress, typing plus final-only delivery, markdown-subset formatting with UTF-16 chunking, and authenticated attachment download. |
|
Package telegram implements the Telegram Bot API adapter for the pigo chat gateway: ingress (webhook and long poll), delivery (typing indicator, streamed preview edits, HTML finalization with chunking), and media download.
|
Package telegram implements the Telegram Bot API adapter for the pigo chat gateway: ingress (webhook and long poll), delivery (typing indicator, streamed preview edits, HTML finalization with chunking), and media download. |
|
Package whatsapp implements the WhatsApp Business Cloud API adapter for the pigo chat gateway: webhook ingress (hub handshake + HMAC-signed events), final-message delivery with reply threading, and authenticated media download.
|
Package whatsapp implements the WhatsApp Business Cloud API adapter for the pigo chat gateway: webhook ingress (hub handshake + HMAC-signed events), final-message delivery with reply threading, and authenticated media download. |