state

package
v0.1.2 Latest Latest
Warning

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

 Go to latest Published: Jul 7, 2026 License: MIT Imports: 13 Imported by: 0

Documentation

Overview

Package state defines the persistence boundary between the open-source agent and any host. This is the host boundary: the agent reaches all durable state - sessions, skills, memory - only through the interfaces here.

The open agent ships a local implementation (in-memory in memory.go; a durable SQLite implementation in storage/sqlite). A commercial host such as an Ion Alpha instance can supply its own Provider backed by a knowledge graph and fleet-wide learning, without this package ever importing the host.

Index

Constants

View Source 
const (
	EvSessionCreated = "session.created"
	EvTurnAppended   = "session.turn_appended"
	EvSessionDeleted = "session.deleted"
	EvSkillUpserted  = "skill.upserted"
	EvSkillDeleted   = "skill.deleted"
	EvMemoryWritten  = "memory.written"
	EvMemoryDeleted  = "memory.deleted"
)

State event types: the vocabulary of the state stream. Each event is the post-image of the affected record(s): the Stamper computes the canonical record (IDs, Seq, HLC, version, timestamps all assigned) and it is written as the event payload, so replaying the events in Seq order reproduces identical state without re-running any clock or RNG. Exported so durable backends project the same vocabulary rather than redeclaring the strings.

View Source 
const StateStream = "state"

StateStream is the spine stream every state mutation is recorded on. A single ordered stream for sessions, skills, and memory means a Replay folds one log to reconstruct the whole provider. It is exported so a host can observe or audit the state stream directly.

Variables

View Source 
var ErrConflict = errors.New("state: version conflict")

ErrConflict is returned by a write when optimistic concurrency fails: the caller passed a non-zero SyncVersion that no longer matches the stored record (someone else wrote in between). Re-read and retry.

View Source 
var ErrNotFound = errors.New("state: not found")

ErrNotFound is returned by stores when a requested record does not exist.

Functions

func MarshalSnapshot 

func MarshalSnapshot(s Snapshot) ([]byte, error)

MarshalSnapshot serializes a state projection into the payload a Replay or Rebuild restores from. The collections are sorted so the payload is deterministic for a given projection: sessions, skills, and items by id, turns by session then seq.

Types

type Envelope 

type Envelope = envelope.Envelope

Envelope is the sync/concurrency metadata carried by every persisted record, embedded into Session, Turn, Skill, and MemoryItem. It is the shared sync envelope (see the envelope package for the fields and the stamping rules): state records and resources carry the identical five fields under identical rules, which is what keeps fleet merge one discipline instead of two.

SyncVersion powers optimistic concurrency: on an update, pass the version you read and the write fails with ErrConflict if the stored version has moved (a zero SyncVersion means "unconditional").

type MemoryItem 

type MemoryItem struct {
	ID        string
	Kind      string // e.g. "fact", "preference", "observation"
	Content   string
	Scope     Scope
	Source    string // provenance: where this memory came from
	CreatedAt time.Time
	Envelope
}

MemoryItem is a durable fact the agent has learned, attributable to its source for provenance and rollback.

func DecodeMemoryItem 

func DecodeMemoryItem(payload map[string]any) (MemoryItem, error)

DecodeMemoryItem extracts the MemoryItem post-image from a state event payload.

type MemoryStore 

type MemoryStore interface {
	// Write persists a memory item, assigning an ID if one is not set.
	Write(ctx context.Context, m MemoryItem) (MemoryItem, error)
	// Recall returns memory matching the query.
	Recall(ctx context.Context, q RecallQuery) ([]MemoryItem, error)
	// Delete tombstones a memory item by ID (soft delete), or returns
	// ErrNotFound.
	Delete(ctx context.Context, id string) error
}

MemoryStore persists and recalls memory. The durable implementation combines lexical (FTS5) and vector (chromem-go) recall; the in-memory implementation does a case-insensitive substring scan, most-recent first.

type Option 

type Option func(*memProvider)

Option configures the in-memory Provider.

func WithClock 

func WithClock(c clock.Clock) Option

WithClock sets the time source for record timestamps (default clock.System), so tests and deterministic replay can supply a clock.Manual. The same clock stamps the event log when one is not injected separately.

func WithEventLog 

func WithEventLog(l spine.Log) Option

WithEventLog backs the provider with a specific spine.Log instead of a private in-memory one. Inject a shared log to observe, audit, or Replay the state stream; pass the same log to Replay to reconstruct the provider from it.

func WithIDGenerator 

func WithIDGenerator(g *ids.Generator) Option

WithIDGenerator sets the source of record IDs (default: a generator on the provider's clock with crypto/rand entropy). Supply a generator seeded with a deterministic clock and entropy so a re-run with the same seeds produces the exact same IDs - the basis of deterministic replay.

func WithInstanceID 

func WithInstanceID(id string) Option

WithInstanceID sets the origin/last-writer instance stamped onto records this provider creates (default "local"). The agent passes its own instance identity here so fleet/P2P merge can attribute records.

func WithSnapshotCodec 

func WithSnapshotCodec(c spine.SnapshotCodec) Option

WithSnapshotCodec makes the provider's snapshots verified: Snapshot seals the projection payload through the codec before saving it, and Replay opens (verifies) a stored snapshot through it before restoring. A snapshot that fails to open is skipped and the stream folds from the start, so an unsigned or tampered snapshot is never restored. With no codec the payload is stored as-is.

func WithSnapshotEvery 

func WithSnapshotEvery(k int) Option

WithSnapshotEvery checkpoints the provider automatically: after every k successful mutations it writes a snapshot, so a Replay folds at most k events past the last checkpoint. The snapshot is best effort, so a failure never fails the write. Zero or negative disables automatic snapshots (the default); Snapshot can still be called explicitly.

type Provider 

type Provider interface {
	// Name identifies the backend for diagnostics, e.g. "memory", "sqlite".
	Name() string
	// Sessions returns the durable conversation store.
	Sessions() SessionStore
	// Skills returns the scoped, searchable skill store.
	Skills() SkillStore
	// Memory returns the durable memory store.
	Memory() MemoryStore
	// Close releases any resources held by the provider.
	Close() error
}

Provider is the agent's durable backend: the single interface a host implements to back the agent with its own storage. The agent never depends on a concrete store, only on this Provider and the stores it returns.

func NewMemory 

func NewMemory(opts ...Option) Provider

NewMemory returns an empty in-memory Provider so the agent runs with zero setup. It is safe for concurrent use and intended as the standalone default and for tests. Every mutation is recorded on a spine.Log and projected, so the provider's state is always a fold of its log (see Replay).

func Replay 

func Replay(ctx context.Context, log spine.Log, opts ...Option) (Provider, error)

Replay reconstructs an in-memory Provider purely by folding a log's "state" stream: the running proof that state is a projection of the spine. The returned Provider is backed by the same log, with its projection caught up to the last recorded event, so subsequent writes continue the stream.

type RecallQuery 

type RecallQuery struct {
	Query string
	Scope Scope
	Limit int
}

RecallQuery selects memory for prefetch into context. Query is matched lexically (and, in vector-capable backends, semantically); Scope narrows the search; Limit caps results (<= 0 means no cap).

type Scope 

type Scope struct {
	Instance  string
	Project   string
	Workspace string
}

Scope locates a resource on the instance/project/workspace axis, so skills and memory can be partitioned and resolved most-specific-first. The zero Scope is the global (instance) scope. Scope is comparable.

type Session 

type Session struct {
	ID        string
	Title     string
	Model     string
	CreatedAt time.Time
	UpdatedAt time.Time
	Envelope
}

Session is a durable, resumable conversation. Sessions survive process restarts so a crashed or disconnected run can be picked back up - the agent's answer to message loss in ephemeral, file-based agents.

func DecodeSession 

func DecodeSession(payload map[string]any) (Session, error)

DecodeSession extracts the Session post-image from a state event payload. Durable backends use it to project the same records the in-memory core does.

type SessionStore 

type SessionStore interface {
	// Create persists a new session, assigning an ID if one is not set.
	Create(ctx context.Context, s Session) (Session, error)
	// Get returns the session by ID, or ErrNotFound.
	Get(ctx context.Context, id string) (Session, error)
	// List returns all sessions, oldest first.
	List(ctx context.Context) ([]Session, error)
	// AppendTurn appends a turn to its session, assigning ID and Seq, and bumps
	// the session's SyncVersion. It returns ErrNotFound if the session does not
	// exist.
	AppendTurn(ctx context.Context, t Turn) (Turn, error)
	// Turns returns a session's transcript in Seq order.
	Turns(ctx context.Context, sessionID string) ([]Turn, error)
	// Delete tombstones a session by ID (soft delete), or returns ErrNotFound.
	Delete(ctx context.Context, id string) error
}

SessionStore persists conversations and their transcripts. Turns are append-only; resuming a session means reading its turns back in Seq order.

type Skill 

type Skill struct {
	ID    string
	Slug  string
	Name  string
	Body  string
	Tags  []string
	Scope Scope
	// Uses and Wins are outcome evidence: how many runs recalled this skill, and
	// how many of those runs then succeeded. They let a skill be ranked and retired
	// by how well it has actually performed, not by recency alone.
	Uses int
	Wins int
	// Check is an optional shell command that verifies the skill still works, kept
	// so the skill can be re-graded later (re-run as the environment changes).
	Check     string
	Version   int
	CreatedAt time.Time
	UpdatedAt time.Time
	Envelope
}

Skill is a reusable, versioned unit of learned procedure. Slug is unique within a Scope; Body is the skill content. Version is the content revision (for provenance/rollback), distinct from Envelope.SyncVersion (the concurrency/sync token).

func DecodeSkill 

func DecodeSkill(payload map[string]any) (Skill, error)

DecodeSkill extracts the Skill post-image from a state event payload.

type SkillStore 

type SkillStore interface {
	// Upsert creates or updates a skill keyed by (Scope, Slug). On update the
	// content Version is incremented, CreatedAt and OriginInstanceID preserved,
	// and SyncVersion bumped. Optimistic concurrency is opt-in: a non-zero
	// SyncVersion on the passed skill must match the stored record, else
	// ErrConflict; a zero SyncVersion writes unconditionally.
	Upsert(ctx context.Context, sk Skill) (Skill, error)
	// Get returns a skill by ID or slug, or ErrNotFound.
	Get(ctx context.Context, idOrSlug string) (Skill, error)
	// List returns the skills in a scope, ordered by slug.
	List(ctx context.Context, scope Scope) ([]Skill, error)
	// Search returns skills matching query, ordered by slug, capped at limit
	// (limit <= 0 means no cap).
	Search(ctx context.Context, query string, limit int) ([]Skill, error)
	// Delete tombstones a skill by ID or slug (soft delete), or returns
	// ErrNotFound.
	Delete(ctx context.Context, idOrSlug string) error
}

SkillStore persists scoped skills and searches them. The durable implementation backs Search with full-text search (SQLite FTS5); the in-memory implementation does a case-insensitive substring scan.

type Snapshot 

type Snapshot struct {
	LastSeq  int64             `json:"lastSeq"`
	Sessions []Session         `json:"sessions"`
	Turns    []Turn            `json:"turns"`
	Skills   []Skill           `json:"skills"`
	Items    []MemoryItem      `json:"items"`
	SlugToID map[string]string `json:"slugToID,omitempty"`
}

Snapshot is a materialized projection of the state stream up to LastSeq: every session, its turns, every skill (live and tombstoned), every memory item, and the derived skill-slug index, enough to restore the read model without folding the stream from the start. It is the state analogue of the resource snapshot, opaque to the spine and sealed by the same codec when one is configured. The slug index is carried so an in-memory restore is exact; a table-backed store rebuilds it from its own rows and leaves the field nil.

func UnmarshalSnapshot 

func UnmarshalSnapshot(payload []byte) (Snapshot, error)

UnmarshalSnapshot decodes a state snapshot payload, the inverse of MarshalSnapshot, so any backend restores from the one shared format.

type Stamper 

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

Stamper computes the canonical post-image of a state mutation: it assigns the record ID, timestamps, the hybrid-logical clock, the content and sync versions, and the sync envelope, enforces optimistic concurrency (CAS), and builds the spine event to append. Every backend routes its writes through one Stamper, so the envelope/version/tombstone rules are defined exactly once and the in-memory and durable backends cannot drift (the duplication that previously lived in both state/memory.go and the SQLite adapter as Go and as SQL).

A Stamper does no persistence and reads no store: the caller supplies the existing record (looked up under its own lock or transaction) and persists the returned record and event. This keeps the rules backend-agnostic and pure, so the same stamp produces the same record regardless of which backend stores it.

func NewStamper 

func NewStamper(instanceID string, clk clock.Clock, hc *hlc.Clock, gen *ids.Generator) *Stamper

NewStamper builds a Stamper from a backend's instance identity and its injected deterministic primitives, so a re-run with the same seeds produces identical records and IDs (the basis of deterministic replay).

func (*Stamper) AppendTurn 

func (s *Stamper) AppendTurn(ses Session, t Turn, nextSeq int64) (Turn, Session, spine.AppendInput, error)

AppendTurn stamps a turn at nextSeq and the envelope bump it induces on its session, returning both plus the single event that records the pair. The caller supplies the live session (so its envelope advances under the same HLC) and the next sequence number from its own store.

func (*Stamper) CreateSession 

func (s *Stamper) CreateSession(ses Session) (Session, spine.AppendInput, error)

CreateSession stamps a new session and returns it with the event to append.

func (*Stamper) DeleteMemory 

func (s *Stamper) DeleteMemory(it MemoryItem) (MemoryItem, spine.AppendInput, error)

DeleteMemory tombstones the given live memory item and returns the event.

func (*Stamper) DeleteSession 

func (s *Stamper) DeleteSession(ses Session) (Session, spine.AppendInput, error)

DeleteSession tombstones the given live session and returns the event.

func (*Stamper) DeleteSkill 

func (s *Stamper) DeleteSkill(sk Skill) (Skill, spine.AppendInput, error)

DeleteSkill tombstones the given live skill (bumping the content version too, so a delete is itself a revision) and returns the event.

func (*Stamper) InstanceID 

func (s *Stamper) InstanceID() string

InstanceID is the instance this Stamper attributes writes to.

func (*Stamper) UpsertSkill 

func (s *Stamper) UpsertSkill(existing *Skill, sk Skill) (Skill, spine.AppendInput, error)

UpsertSkill stamps a create or an update keyed by (Scope, Slug). existing is the stored record for that key (tombstones included, so an upsert over a tombstone resurrects it) or nil if none. Optimistic concurrency is opt-in: a non-zero SyncVersion on the input must match the stored one, else ErrConflict; a non-zero SyncVersion with no existing record is also a conflict (it expected one).

func (*Stamper) WriteMemory 

func (s *Stamper) WriteMemory(it MemoryItem) (MemoryItem, spine.AppendInput, error)

WriteMemory stamps a new memory item and returns the event.

type Turn 

type Turn struct {
	ID        string
	SessionID string
	Seq       int64
	Role      string // "user", "assistant", "tool", or "system"
	Content   string
	CreatedAt time.Time
	Envelope
}

Turn is one entry in a session's ordered transcript. Seq is assigned by the store and is monotonic within a session.

func DecodeTurn 

func DecodeTurn(payload map[string]any) (Turn, error)

DecodeTurn extracts the Turn post-image from a state event payload.

Source Files

View all Source files

Directories

Path Synopsis
Package statetest is the conformance suite for state.Provider.
 Package statetest is the conformance suite for state.Provider.

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.