chat

package
v0.4.10 Latest Latest
Warning

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

Go to latest
Published: Jul 30, 2026 License: MIT Imports: 21 Imported by: 0

Documentation

Overview

Package chat implements the core of the orb 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

View Source
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

func AllowAll

func AllowAll(Message) error

AllowAll is an explicit opt-out authorizer accepting every message.

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

type Handler interface {
	Handle(ctx context.Context, m Message) error
}

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, running on their own bounded pool, 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

func NewLocal(handler Handler, spoolPath string) (*Local, error)

NewLocal opens (or creates) the spool at spoolPath, replays unacked messages, compacts the file, and starts the dispatcher.

func (*Local) Close

func (l *Local) Close(ctx context.Context) error

Close stops accepting work and waits for in-flight handling until ctx ends.

func (*Local) Publish

func (l *Local) Publish(m Message) error

Publish durably appends the message and hands it to the dispatcher. This is the callback given to adapter ingress.

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

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.

func New

func New(opts Options) (*Processor, error)

New validates opts and creates a Processor.

func (*Processor) Close

func (p *Processor) Close(ctx context.Context) error

Close waits for in-flight turns and rejects new ones.

func (*Processor) Handle

func (p *Processor) Handle(ctx context.Context, m Message) error

Handle processes one inbound message synchronously: it returns only when the turn is delivered, deduplicated, or failed. Redelivering the same EventID is safe — the turn ledger short-circuits duplicate work.

type Receipt

type Receipt struct {
	MessageIDs []string  `json:"messageIds"`
	At         time.Time `json:"at"`
}

Receipt records the platform message ids produced by a finalized delivery.

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.

Directories

Path Synopsis
Package discord implements the Discord adapter for the orb 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 orb 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 orb 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 orb chat gateway: an HTTP-endpoint Chat app with bearer-JWT-verified ingress, service-account authenticated delivery, and authenticated media download.
internal
ctxsleep
Package ctxsleep provides a context-aware sleep shared by the chat adapters and processor.
Package ctxsleep provides a context-aware sleep shared by the chat adapters and processor.
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 orb 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 orb 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 orb 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 orb 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 orb 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 orb 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 orb 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 orb 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 orb 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 orb chat gateway: webhook ingress (hub handshake + HMAC-signed events), final-message delivery with reply threading, and authenticated media download.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL