chat

package
v0.19.11 Latest Latest
Warning

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

Go to latest
Published: Oct 17, 2026 License: MIT Imports: 26 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.

func AttachmentKind added in v0.19.1

func AttachmentKind(mime string) string

AttachmentKind maps a MIME type to an AttachmentRef Kind: image/* is "photo", audio/* "audio", video/* "video", and anything else "document".

func SendFile added in v0.19.1

func SendFile(api extensions.API) error

SendFile is the send_file tool, for the conversations of a FileSender platform: it sends a file under the working directory to the person the agent is talking with.

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 Agent added in v0.17.6

type Agent struct {
	// Serve runs the agent's ACP server on one connection until it closes.
	Serve func(ctx context.Context, in io.Reader, out io.Writer) error
	// Connect returns the command line of a process that relays its stdio to
	// the ACP socket at path, for a client that starts its agent as a command.
	Connect func(socket string) ([]string, error)
	// Environ is the environment the agent's tools get; Export adds to it.
	Environ func() []string
	Export  func(name, value string)
	// Log receives the platform's diagnostics.
	Log io.Writer
}

Agent is the running agent as a front, a platform that drives its sessions itself, reaches it.

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 {
	Reset   func() error
	Session *agent.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 FileSender added in v0.19.1

type FileSender interface {
	// SendFile sends size bytes of content as the file name to the
	// conversation key, or fails with the platform's reason, a size limit
	// included.
	SendFile(ctx context.Context, key ConversationKey, name string, size int64, content io.Reader) error
}

FileSender is an Adapter whose platform takes files from the agent: the conversations of its platform get SendFile.

type Handler

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

Handler consumes one message synchronously. *Processor implements it.

type Identity added in v0.18.0

type Identity struct {
	Name, About, Avatar string
}

Identity is how an agent presents itself on its platforms.

type Inbound added in v0.17.6

type Inbound struct {
	Poll    func(ctx context.Context, publish func(Message) error) error
	Webhook func(publish func(Message) error) http.Handler
}

Inbound is how a platform's messages reach the processor: it fetches them itself (Poll), or it is called and its host serves the handler (Webhook).

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 NewLocalWithSpool added in v0.8.0

func NewLocalWithSpool(handler Handler, spool Spool) (*Local, error)

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 ~/.orb/agent.

func WithPersistence added in v0.8.0

func WithPersistence(repo func(ConversationKey) harness.SessionRepo, settings *config.SettingsManager, registry *config.ModelRegistry) LocalProviderOption

WithPersistence selects explicit native session and configuration repositories.

func WithSessionOptions

func WithSessionOptions(hook func(key ConversationKey, o *agent.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").

func WithWorkspace added in v0.17.0

func WithWorkspace(cwd string, open Open) LocalProviderOption

WithWorkspace runs conversations as full agents working in cwd, each session built by open: the way to give an isolated deployment its tools. Without it sessions have no tools (D27).

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 ParseSpool added in v0.8.0

func ParseSpool(data []byte) ([]Message, error)

ParseSpool rejects damaged migration sources instead of silently dropping work.

func (Message) Key

func (m Message) Key() ConversationKey

Key returns the conversation key this message belongs to.

type Open added in v0.17.0

type Open func(ctx context.Context, key ConversationKey, manager *sessionstore.SessionManager) (session *agent.AgentSession, close func(), err error)

Open builds the agent session of the conversation key on its manager; close disposes it.

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.

type Spool added in v0.8.0

type Spool interface {
	Pending(context.Context) ([]Message, error)
	Put(context.Context, Message) error
	Ack(context.Context, string) error
}

Spool is a durable pending-message store. The owner supplies its lifetime.

Directories

Path Synopsis
Package buzz runs an Orb agent on Buzz: buzz-acp holds the agent's relay identity and drives its sessions over ACP, and the agent's shell posts with the buzz CLI, which runs here with the Buzz key.
Package buzz runs an Orb agent on Buzz: buzz-acp holds the agent's relay identity and drives its sessions over ACP, and the agent's shell posts with the buzz CLI, which runs here with the Buzz key.
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 gateway runs an agent's chat platforms as one process: the adapters and their ingress (polling or a webhook server) feed one at-least-once processor over local sessions, beside the fronts that drive the agent themselves.
Package gateway runs an agent's chat platforms as one process: the adapters and their ingress (polling or a webhook server) feed one at-least-once processor over local sessions, beside the fronts that drive the agent themselves.
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 serves the Meta Graph webhook endpoint shared by the Cloud API chat adapters (WhatsApp, Messenger): the one-time hub.challenge subscribe handshake and the X-Hub-Signature-256 raw-body HMAC check, both compared in constant time via hmac.Equal.
Package graphhook serves the Meta Graph webhook endpoint shared by the Cloud API chat adapters (WhatsApp, Messenger): the one-time hub.challenge subscribe handshake and the X-Hub-Signature-256 raw-body HMAC check, both compared in constant time via hmac.Equal.
httpjson
Package httpjson is the JSON-over-HTTP round trip and retry loop shared by the chat adapters' platform API clients.
Package httpjson is the JSON-over-HTTP round trip and retry loop shared by the chat adapters' platform API clients.
runechunk
Package runechunk splits and cuts platform text at readable rune boundaries.
Package runechunk splits and cuts platform text at readable rune boundaries.
typing
Package typing keeps a platform typing indicator alive while a turn runs.
Package typing keeps a platform typing indicator alive while a turn runs.
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 platforms is the catalog of chat platforms an agent runs on, shared by the assemblies that run agents (orb chat, the team agent image): one row per platform maps its environment, and its section of an agent file, onto the platform package's options.
Package platforms is the catalog of chat platforms an agent runs on, shared by the assemblies that run agents (orb chat, the team agent image): one row per platform maps its environment, and its section of an agent file, onto the platform package's options.
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