inbox

Documentation

Overview

Package inbox is the agent's unified inbound boundary. Every message or event the agent receives, from any source, is recorded as an Entry resource and then triaged into an action: a reply, a goal, a memory write, a notification, or nothing. Separating where input arrives (a Source) from what the agent does about it (the triage reconciler) means a new platform is a small adapter rather than new agent logic, and every inbound item is durable, replayable, audited, and governed at one waist.

Entry is the inbound envelope expressed on the same resource + reconcile foundation as Goal: declarative and level-triggered, so triage is crash-resumable and a lost change hint is recovered by resync rather than dropping the input.

Index

• Constants
• func ReceiveLoop(ctx context.Context, backoff time.Duration, clk clock.Timing, ...)
• func RegisterKind(reg *resource.Registry) error
• type Condition
• type Disposition
  • func DefaultPolicy(s Spec) Disposition
• type Enqueuer
• type Ingest
  • func NewIngest(store resource.Store, queue Enqueuer, clk clock.Timing, sources []Source, ...) *Ingest
  • func (in *Ingest) Run(ctx context.Context) error
• type IngestOption
  • func WithIngestErrorHandler(fn func(error)) IngestOption
• type Phase
• type Policy
• type Sink
• type Sinks
  • func NewSinks(sinks ...Sink) *Sinks
  • func (s *Sinks) Send(ctx context.Context, source, conversation, text string) error
• type Source
• type Spec
  • func DecodeSpec(r resource.Resource) (Spec, error)
• type Status
  • func DecodeStatus(r resource.Resource) (Status, error)
  • func (s Status) Encode() (json.RawMessage, error)
  • func (s *Status) SetCondition(c Condition, now time.Time)
• type Triage
  • func NewTriage(store resource.Store, worker Worker, sinks *Sinks, clk clock.Timing, ...) *Triage
  • func (t *Triage) Reconcile(ctx context.Context, ref reconcile.Ref) (reconcile.Result, error)
• type TriageOption
  • func WithPolicy(p Policy) TriageOption
  • func WithPollInterval(d time.Duration) TriageOption
• type Worker

Constants

const (
	// GroupVersion is the Entry kind's API group and version.
	GroupVersion = "inbox.ionagent.io/v1alpha1"
	// Kind is the resource kind name.
	Kind = "Entry"
)

const (
	CondTriaged = "Triaged" // True once a disposition has been chosen
	CondActed   = "Acted"   // True once the disposition has completed
	CondFailed  = "Failed"  // True when acting on the entry failed terminally
)

Standard condition types (kstatus-style: present and True only when noteworthy).

Functions

func ReceiveLoop

func ReceiveLoop(ctx context.Context, backoff time.Duration, clk clock.Timing, attempt func(context.Context) error)

ReceiveLoop is the reconnect skeleton every long-lived Source shares: it runs attempt repeatedly until ctx is cancelled, pausing for backoff after a failed attempt so a dropped connection is retried without a busy loop. A nil error means the attempt ended cleanly (a long poll returned, a stream closed on cancellation) and the next attempt runs immediately; a non-nil error triggers the backoff wait. The wait is clock-driven, so a Manual clock makes reconnect timing deterministic under test instead of sleeping on the wall clock. Every source (Telegram, Signal, and future Discord/Slack/email adapters) drives its Receive goroutine through this rather than hand-rolling the same loop.

func RegisterKind

func RegisterKind(reg *resource.Registry) error

RegisterKind registers the Entry kind so entries can be stored and admitted like any other resource.

Types

type Condition

type Condition = resource.Condition

Condition is one standard status condition (the shared resource.Condition).

type Disposition

type Disposition string

Disposition is the action triage chose for an entry. Empty means not yet triaged.

const (
	DispositionReply  Disposition = "Reply"  // answer in the originating conversation
	DispositionGoal   Disposition = "Goal"   // run it as a goal
	DispositionStore  Disposition = "Store"  // record it (e.g. to memory) without replying
	DispositionNotify Disposition = "Notify" // push a message out, not a reply to input
	DispositionDrop   Disposition = "Drop"   // intentionally ignore
)

The dispositions triage can choose for an entry.

func DefaultPolicy

func DefaultPolicy(s Spec) Disposition

DefaultPolicy replies to an entry that names a conversation and drops one that does not (an entry with nowhere to answer is not actioned by default).

type Enqueuer

type Enqueuer interface {
	Enqueue(resource.Key)
}

Enqueuer hands a newly recorded entry's key to the triage controller's work queue so it is reconciled promptly. reconcile.Manager.Enqueue satisfies it; the resync sweep is the safety net if a hint is ever missed.

type Ingest

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

Ingest records inbound entries from a set of Sources as Entry resources and enqueues each for triage. It is the inbound half of the boundary: it decides only that an entry is durably recorded, never what to do with it. That decision is the triage controller's, which keeps every disposition behind one governed waist rather than scattered across adapters.

func NewIngest

func NewIngest(store resource.Store, queue Enqueuer, clk clock.Timing, sources []Source, opts ...IngestOption) *Ingest

NewIngest builds an ingester that records entries from sources onto store and enqueues each for triage on queue. The clock stamps the receipt time when a source leaves it unset.

func (*Ingest) Run

func (in *Ingest) Run(ctx context.Context) error

Run starts every source and records entries until ctx is cancelled, then waits for the source readers to stop. A source that fails to start is reported and skipped; if none start, Run returns an error rather than blocking forever.

type IngestOption

type IngestOption func(*Ingest)

IngestOption configures an Ingest.

func WithIngestErrorHandler

func WithIngestErrorHandler(fn func(error)) IngestOption

WithIngestErrorHandler registers a callback for non-fatal errors (a source that will not start, a record that fails to persist). The default discards them. It may be called from several goroutines and must be safe for that.

type Phase

type Phase string

Phase is a coarse lifecycle summary of an entry: received, triaged into a disposition, then a terminal acted or dropped.

const (
	PhaseReceived Phase = "Received" // recorded, not yet triaged
	PhaseTriaged  Phase = "Triaged"  // a disposition was chosen, action in progress
	PhaseActed    Phase = "Acted"    // the disposition completed
	PhaseDropped  Phase = "Dropped"  // triage chose to take no action
)

The phases an entry moves through, from received to a terminal acted or dropped.

type Policy

type Policy func(Spec) Disposition

Policy chooses what to do with an entry from its content alone. The default replies to any conversational entry and drops the rest. A host can supply its own to add routing, allow-lists, or autonomy rules.

type Sink

type Sink interface {
	Name() string
	Send(ctx context.Context, conversation, text string) error
}

Sink delivers an outbound message to a conversation on a source's platform, so a disposition can reply or notify on the channel an entry arrived on. Name matches the Source it pairs with, which is how the triage controller finds the right Sink for an entry's source.

type Sinks

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

Sinks routes a reply to the Sink registered for an entry's source.

func NewSinks

func NewSinks(sinks ...Sink) *Sinks

NewSinks indexes sinks by their Name.

func (*Sinks) Send

func (s *Sinks) Send(ctx context.Context, source, conversation, text string) error

Send delivers text on the sink registered for source. It errors when no sink is registered for the source, so a dropped reply is surfaced rather than silent.

type Source

type Source interface {
	Name() string
	Receive(ctx context.Context) (<-chan Spec, error)
}

Source produces inbound entries from one origin: a chat platform, an email mailbox, a webhook listener, a monitor. Receive streams entries until ctx is cancelled, at which point the source closes the returned channel. Name identifies the source for routing replies back through the matching Sink. A Source need not set Spec.Source; the ingester stamps it from Name, so routing never depends on the adapter filling it in.

type Spec

type Spec struct {
	// Source names the adapter the entry arrived on, e.g. "telegram". Replies are
	// routed back to the Sink registered under this name.
	Source string `json:"source"`
	// Conversation identifies the thread on the source platform and is the reply
	// address scope. Empty for sources that are not conversational (a monitor).
	Conversation string `json:"conversation,omitempty"`
	// Sender is the platform user id or handle, for routing and audit. Optional.
	Sender string `json:"sender,omitempty"`
	// Type is the nature of the entry; defaults to "message" when unset.
	Type string `json:"type,omitempty"`
	// Content is the message body or event payload.
	Content string `json:"content,omitempty"`
	// Metadata carries source-specific extras without widening the schema.
	Metadata map[string]string `json:"metadata,omitempty"`
	// ReceivedAt is when the source observed the entry.
	ReceivedAt time.Time `json:"receivedAt,omitempty"`
}

Spec is an inbound entry's content: where it came from and what it carries. It is immutable input, set once when the entry is received.

func DecodeSpec

func DecodeSpec(r resource.Resource) (Spec, error)

DecodeSpec reads the typed spec from a resource.

type Status

type Status struct {
	Phase Phase `json:"phase,omitempty"`
	// Disposition is the action triage chose; empty until triaged.
	Disposition Disposition `json:"disposition,omitempty"`
	// ObservedSpecHash is the resource.SpecHash triage last acted on, so a
	// re-reconcile is a no-op once the entry has settled.
	ObservedSpecHash string `json:"observedSpecHash,omitempty"`
	// GoalName is the goal an entry was routed to, set when Disposition is Goal.
	GoalName   string      `json:"goalName,omitempty"`
	Conditions []Condition `json:"conditions,omitempty"`
	Message    string      `json:"message,omitempty"`
}

Status is an entry's observed triage state.

func DecodeStatus

func DecodeStatus(r resource.Resource) (Status, error)

DecodeStatus reads the typed status from a resource.

func (Status) Encode

func (s Status) Encode() (json.RawMessage, error)

Encode marshals the status for writing back onto a resource.

func (*Status) SetCondition

func (s *Status) SetCondition(c Condition, now time.Time)

SetCondition upserts c by type, stamping LastTransitionTime only when the status value actually changes, so a no-op reconcile does not churn the time.

type Triage

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

Triage is the reconciler that turns a recorded Entry into an action. It is level-triggered: each call re-reads the live entry and advances it one step, from received to a disposition, then (for a reply) waits on the work and sends the answer back. It never runs the work itself; that is the Worker's job, so a long task does not block the reconcile loop.

func NewTriage

func NewTriage(store resource.Store, worker Worker, sinks *Sinks, clk clock.Timing, opts ...TriageOption) *Triage

NewTriage builds the triage reconciler over store, dispatching Reply/Goal work to worker and routing replies through sinks.

func (*Triage) Reconcile

func (t *Triage) Reconcile(ctx context.Context, ref reconcile.Ref) (reconcile.Result, error)

Reconcile advances one entry. Only Reply, Goal, and Drop are acted on today; Store and Notify are reserved and currently treated as Drop.

type TriageOption

type TriageOption func(*Triage)

TriageOption configures a Triage.

func WithPolicy

func WithPolicy(p Policy) TriageOption

WithPolicy sets the disposition policy (default DefaultPolicy).

func WithPollInterval

func WithPollInterval(d time.Duration) TriageOption

WithPollInterval sets how often in-flight work is re-checked (default defaultTriagePoll). A non-positive value is ignored.

type Worker

type Worker interface {
	Start(ctx context.Context, entry, objective string) (handle string, err error)
	Poll(ctx context.Context, handle string) (done bool, answer string, failed bool, err error)
}

Worker carries out the action a Reply or Goal disposition implies: it starts work for an entry and reports the outcome. Triage depends on this rather than the goal engine, so the inbound boundary stays independent of the execution model. The entry id is passed so an implementation can make Start idempotent per entry.