storage

package
v0.13.0 Latest Latest
Warning

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

Go to latest
Published: Aug 30, 2026 License: MIT Imports: 6 Imported by: 0

Documentation

Overview

Package storage defines the core persistence interfaces for Chronos.

Index

Constants

View Source
const (
	// DefaultPageLimit is used when a caller passes a non-positive limit.
	DefaultPageLimit = 100
	// MaxPageLimit caps how many rows a single paged read may return.
	MaxPageLimit = 1000
)

Pagination bounds. Adapters clamp caller-supplied limits into this range so a single request can never scan an unbounded ledger.

View Source
const DefaultTenant = "default"

DefaultTenant is the tenant id used when a context carries no explicit tenant. It preserves single-tenant behavior for callers that do not opt into multi-tenancy: all such reads and writes are scoped to this tenant.

Variables

View Source
var ErrFileNotFound = errors.New("session file not found")

ErrFileNotFound is returned by SessionFileStore.GetFile when no file exists at the requested path. Callers can test for it with errors.Is.

Functions

func ClampLimit added in v0.4.0

func ClampLimit(limit int) int

ClampLimit normalizes a caller-supplied page limit into [1, MaxPageLimit], defaulting to DefaultPageLimit when limit <= 0.

func DecodeSeqCursor added in v0.4.0

func DecodeSeqCursor(cursor string) (int64, error)

DecodeSeqCursor decodes a cursor produced by EncodeSeqCursor. An empty cursor decodes to 0 with no error.

func DecodeStrCursor added in v0.4.0

func DecodeStrCursor(cursor string) (string, error)

DecodeStrCursor decodes a cursor produced by EncodeStrCursor. An empty cursor decodes to the empty string with no error.

func EncodeSeqCursor added in v0.4.0

func EncodeSeqCursor(seq int64) string

EncodeSeqCursor encodes a monotonic sequence number into an opaque cursor.

func EncodeStrCursor added in v0.4.0

func EncodeStrCursor(key string) string

EncodeStrCursor encodes a string key (e.g. a row's primary-key id) into an opaque cursor for keyset pagination.

func MatchesFilter added in v0.10.0

func MatchesFilter(md, filter map[string]any) bool

MatchesFilter reports whether md satisfies every key/value pair in filter (exact match). A nil or empty filter matches everything. It is the shared client-side implementation for in-memory stores and adapters without native metadata filtering.

func PrefixRange added in v0.10.0

func PrefixRange(prefix string) (lo, hi string, hasUpper bool)

PrefixRange returns the half-open key range [lo, hi) containing exactly the strings that begin with prefix, for a byte-wise (case-sensitive) SQL range scan of the form `path >= lo AND path < hi`. A range scan is used in place of SQL LIKE deliberately: SQLite's LIKE is ASCII-case-insensitive by default, which would make prefix listing diverge from the case-sensitive Postgres and in-memory implementations; a byte range is case-sensitive on every backend.

hasUpper is false when no finite upper bound exists — an empty prefix (lists everything) or a prefix that is entirely 0xFF bytes — in which case the caller applies only the `path >= lo` bound.

func SessionFromContext added in v0.10.0

func SessionFromContext(ctx context.Context) string

SessionFromContext returns the session id carried by ctx, or the empty string when none has been set. Unlike TenantFromContext there is no default: a caller that needs a session id must treat the empty string as "no active session".

func TenantFromContext added in v0.6.0

func TenantFromContext(ctx context.Context) string

TenantFromContext returns the tenant id carried by ctx, or DefaultTenant when no tenant has been set. It never returns an empty string, so adapters can use the result directly as a query parameter.

func WithSession added in v0.10.0

func WithSession(ctx context.Context, sessionID string) context.Context

WithSession returns a copy of ctx that carries the given session id. Harness primitives that persist per-session state (e.g. the planning tool and the virtual filesystem) read this id to scope their reads and writes to a single session. An empty id is a no-op and leaves the context unchanged.

This is intentionally shaped like WithTenant: session scoping composes with tenant scoping, so a fully scoped context carries both.

func WithTenant added in v0.6.0

func WithTenant(ctx context.Context, tenantID string) context.Context

WithTenant returns a copy of ctx that carries the given tenant id. Storage adapters scope every read and write to this tenant, making cross-tenant access impossible. An empty id is normalized to DefaultTenant.

The control plane derives the tenant from the authenticated principal (JWT or API key) and never from client-supplied object ids; see the ChronosOS server.

Types

type AuditLog

type AuditLog struct {
	ID        string         `json:"id"`
	TenantID  string         `json:"tenant_id,omitempty"`
	SessionID string         `json:"session_id"`
	Actor     string         `json:"actor"`
	Action    string         `json:"action"`
	Resource  string         `json:"resource"`
	Detail    map[string]any `json:"detail,omitempty"`
	CreatedAt time.Time      `json:"created_at"`
}

AuditLog records a security-relevant event.

type BatchIngester added in v0.4.0

type BatchIngester interface {
	// AppendEvents appends many ledger events atomically in one batch. Like
	// AppendEvent it is idempotent on the event id.
	AppendEvents(ctx context.Context, events []*Event) error
	// InsertTraces inserts many trace spans in one batch.
	InsertTraces(ctx context.Context, traces []*Trace) error
}

BatchIngester is an OPTIONAL interface a Storage adapter may implement to ingest many rows in a single round-trip (multi-row INSERT / COPY) instead of one statement per row. It is separate from Storage so that adding it never breaks existing implementations. Callers should type-assert:

if b, ok := store.(storage.BatchIngester); ok {
	err := b.AppendEvents(ctx, events)
} else {
	for _, e := range events { _ = store.AppendEvent(ctx, e) }
}

type Checkpoint

type Checkpoint struct {
	ID        string         `json:"id"`
	TenantID  string         `json:"tenant_id,omitempty"`
	SessionID string         `json:"session_id"`
	RunID     string         `json:"run_id"`
	NodeID    string         `json:"node_id"`
	State     map[string]any `json:"state"`
	SeqNum    int64          `json:"seq_num"`
	CreatedAt time.Time      `json:"created_at"`
}

Checkpoint captures the full state of a run for resume/time-travel.

type CheckpointPage added in v0.4.0

type CheckpointPage struct {
	Checkpoints []*Checkpoint `json:"checkpoints"`
	NextCursor  string        `json:"next_cursor,omitempty"`
}

CheckpointPage is a cursor-paginated slice of checkpoints.

type Config

type Config struct {
	// Storage backend: "sqlite", "postgres", "mysql", "redis", "mongo", "dynamo"
	Backend string `json:"backend" yaml:"backend"`

	// DSN / connection string for the storage backend.
	DSN string `json:"dsn" yaml:"dsn"`

	// VectorBackend: "qdrant", "redisvector", "milvus", "weaviate", "pinecone"
	VectorBackend string `json:"vector_backend" yaml:"vector_backend"`

	// VectorDSN / connection for vector store.
	VectorDSN string `json:"vector_dsn" yaml:"vector_dsn"`
}

Config selects which storage and vector adapters to use at runtime.

type Embedding

type Embedding struct {
	ID       string         `json:"id"`
	Vector   []float32      `json:"vector"`
	Metadata map[string]any `json:"metadata,omitempty"`
	Content  string         `json:"content,omitempty"`
}

Embedding represents a vector embedding with metadata.

type Event

type Event struct {
	ID        string    `json:"id"`
	TenantID  string    `json:"tenant_id,omitempty"`
	SessionID string    `json:"session_id"`
	SeqNum    int64     `json:"seq_num"`
	Type      string    `json:"type"`
	Payload   any       `json:"payload"`
	CreatedAt time.Time `json:"created_at"`
}

Event is an append-only ledger entry for replayability.

type EventPage added in v0.4.0

type EventPage struct {
	Events     []*Event `json:"events"`
	NextCursor string   `json:"next_cursor,omitempty"`
}

EventPage is a cursor-paginated slice of events. NextCursor is empty when the page is the last one.

type FileInfo added in v0.10.0

type FileInfo struct {
	Path      string    `json:"path"`
	Size      int       `json:"size"`
	UpdatedAt time.Time `json:"updated_at"`
}

FileInfo describes a stored session file without its content, for directory listings.

type MemoryRecord

type MemoryRecord struct {
	ID        string    `json:"id"`
	TenantID  string    `json:"tenant_id,omitempty"`
	SessionID string    `json:"session_id,omitempty"` // empty = long-term
	AgentID   string    `json:"agent_id"`
	UserID    string    `json:"user_id,omitempty"`
	Kind      string    `json:"kind"` // short_term, long_term
	Key       string    `json:"key"`
	Value     any       `json:"value"`
	CreatedAt time.Time `json:"created_at"`
}

MemoryRecord represents a short-term or long-term memory entry.

type Paginator added in v0.4.0

type Paginator interface {
	ListEventsPaged(ctx context.Context, sessionID string, afterSeq int64, limit int, cursor string) (*EventPage, error)
	ListTracesPaged(ctx context.Context, sessionID string, limit int, cursor string) (*TracePage, error)
	ListCheckpointsPaged(ctx context.Context, sessionID string, limit int, cursor string) (*CheckpointPage, error)
}

Paginator is an OPTIONAL interface a Storage adapter may implement to expose cursor-paginated reads of the unbounded ledgers (traces, events, checkpoints).

It is deliberately separate from Storage so that adding it never breaks existing Storage implementations. Callers should type-assert:

if p, ok := store.(storage.Paginator); ok {
	page, err := p.ListEventsPaged(ctx, sid, 0, 100, "")
}

Pass an empty cursor to fetch the first page; pass the previous page's NextCursor to fetch the following page. Limits are clamped via ClampLimit.

type Retention added in v0.4.0

type Retention interface {
	// TrimEvents deletes events created strictly before olderThan.
	TrimEvents(ctx context.Context, olderThan time.Time) (int64, error)
	// TrimTraces deletes traces started strictly before olderThan.
	TrimTraces(ctx context.Context, olderThan time.Time) (int64, error)
	// TrimAuditLogs deletes audit logs created strictly before olderThan.
	TrimAuditLogs(ctx context.Context, olderThan time.Time) (int64, error)
	// TrimCheckpoints keeps only the most recent keep checkpoints (by seq_num)
	// for the given session and deletes the rest. keep <= 0 is a no-op.
	TrimCheckpoints(ctx context.Context, sessionID string, keep int) (int64, error)
}

Retention is an OPTIONAL interface a Storage adapter may implement to trim the unbounded, append-only tables (events, traces, audit logs, checkpoints). It is separate from Storage so that adding it never breaks existing implementations.

Each Trim* method returns the number of rows removed. Callers should type-assert:

if r, ok := store.(storage.Retention); ok {
	_, _ = r.TrimEvents(ctx, time.Now().Add(-30*24*time.Hour))
}

type RetentionPolicy added in v0.4.0

type RetentionPolicy struct {
	// MaxAge deletes events, traces and audit logs older than now-MaxAge.
	// Zero disables age-based trimming.
	MaxAge time.Duration `json:"max_age,omitempty"`
	// KeepCheckpointsPerSession limits checkpoints retained per session.
	// Zero disables checkpoint trimming.
	KeepCheckpointsPerSession int `json:"keep_checkpoints_per_session,omitempty"`
}

RetentionPolicy describes a time-and-count based retention window that a caller (e.g. a background job) can apply via ApplyRetention.

type RetentionResult added in v0.4.0

type RetentionResult struct {
	Events      int64 `json:"events"`
	Traces      int64 `json:"traces"`
	AuditLogs   int64 `json:"audit_logs"`
	Checkpoints int64 `json:"checkpoints"`
}

RetentionResult reports how many rows each Trim* step removed.

func ApplyRetention added in v0.4.0

func ApplyRetention(ctx context.Context, r Retention, policy RetentionPolicy, sessionIDs []string) (RetentionResult, error)

ApplyRetention runs a RetentionPolicy against a Retention store. The sessionIDs argument scopes checkpoint trimming (checkpoint retention is per-session); pass nil to skip checkpoint trimming.

type SearchOption added in v0.10.0

type SearchOption func(*SearchOptions)

SearchOption customizes a similarity search (e.g. a metadata filter).

func WithFilter added in v0.10.0

func WithFilter(filter map[string]any) SearchOption

WithFilter restricts a search to embeddings whose metadata matches every key/value pair in filter (exact match). This is how callers scope a shared collection — e.g. a tenant token — so top-k is computed within that scope.

type SearchOptions added in v0.10.0

type SearchOptions struct {
	// Filter restricts results to embeddings whose metadata matches every
	// key/value pair (exact match, AND semantics). A nil or empty Filter matches
	// everything. Adapters that store metadata structurally (qdrant, pgvector,
	// pinecone, chromadb) apply it server-side so top-k is computed over the
	// matching subset; adapters that store metadata as an opaque blob (lancedb,
	// milvus, weaviate, redisvector) apply it client-side over the returned
	// window (which may under-return under a highly selective filter).
	Filter map[string]any
}

SearchOptions configures a similarity search. It is built from the variadic SearchOption arguments to VectorStore.Search via ApplySearchOptions.

func ApplySearchOptions added in v0.10.0

func ApplySearchOptions(opts ...SearchOption) SearchOptions

ApplySearchOptions folds opts into a SearchOptions. Adapters call this at the top of Search to read the requested filter.

type SearchResult

type SearchResult struct {
	Embedding
	Score float32 `json:"score"`
}

SearchResult is a single result from a similarity search.

func FilterSearchResults added in v0.10.0

func FilterSearchResults(results []SearchResult, filter map[string]any) []SearchResult

FilterSearchResults returns, in order, the results whose metadata matches filter. A nil or empty filter returns results unchanged. This is the shared client-side path for adapters that cannot filter on metadata server-side. It returns a new slice and never mutates the input.

type Session

type Session struct {
	ID        string         `json:"id"`
	TenantID  string         `json:"tenant_id,omitempty"`
	AgentID   string         `json:"agent_id"`
	Status    string         `json:"status"` // running, paused, completed, failed
	Metadata  map[string]any `json:"metadata,omitempty"`
	CreatedAt time.Time      `json:"created_at"`
	UpdatedAt time.Time      `json:"updated_at"`
}

Session represents an agent execution session.

type SessionFileStore added in v0.10.0

type SessionFileStore interface {
	// PutFile creates or overwrites the file at path with content.
	PutFile(ctx context.Context, sessionID, path string, content []byte) error
	// GetFile returns the content of the file at path, or an error if it does
	// not exist.
	GetFile(ctx context.Context, sessionID, path string) ([]byte, error)
	// ListFiles returns metadata for every file in the session whose path has
	// the given prefix (empty prefix lists all), ordered by path.
	ListFiles(ctx context.Context, sessionID, prefix string) ([]FileInfo, error)
	// DeleteFile removes the file at path. Deleting a missing file is not an
	// error (idempotent).
	DeleteFile(ctx context.Context, sessionID, path string) error
}

SessionFileStore is an OPTIONAL interface a Storage adapter may implement to persist per-session scratch files — the backing store for the harness virtual filesystem (engine/tool/builtins VFS) that lets an agent offload large intermediate artifacts out of its context window and page them back in by path. It is deliberately separate from Storage so that adding it never breaks existing Storage implementations; callers type-assert:

if fs, ok := store.(storage.SessionFileStore); ok {
	_ = fs.PutFile(ctx, sessionID, "notes.md", data)
}

Every method is scoped to the tenant carried by ctx (TenantFromContext) and to the given session id, so files written under one tenant/session are never visible to another — the same isolation guarantee as the rest of Storage.

type SessionStatusUpdater added in v0.13.0

type SessionStatusUpdater interface {
	// UpdateSessionStatus sets the session's Status and UpdatedAt. It returns
	// an error if the session does not exist (or is not visible to the
	// context's tenant), mirroring GetSession's not-found behavior.
	UpdateSessionStatus(ctx context.Context, sessionID, status string, updatedAt time.Time) error
}

SessionStatusUpdater is an OPTIONAL interface a Storage adapter may implement to update a session's Status (and UpdatedAt) as a narrow, targeted write instead of a whole-record GetSession -> mutate -> UpdateSession read-modify-write. Sessions already have another read-modify-write writer of their own (a plan/VFS Metadata writer scoped by storage.WithSession), so a caller that only wants to flip Status — e.g. the graph runner mirroring a run's paused/completed/failed status — would otherwise risk silently clobbering a concurrent Metadata write with the stale copy it read moments earlier. A narrow update can never do that, because it never touches Metadata.

Callers should type-assert and fall back to GetSession/UpdateSession when the store doesn't implement it (accepting the pre-existing race) so this stays additive: an adapter that doesn't implement SessionStatusUpdater is otherwise unaffected.

if su, ok := store.(storage.SessionStatusUpdater); ok {
	_ = su.UpdateSessionStatus(ctx, sessionID, status, time.Now())
}

type Storage

type Storage interface {
	// Sessions
	CreateSession(ctx context.Context, s *Session) error
	GetSession(ctx context.Context, id string) (*Session, error)
	UpdateSession(ctx context.Context, s *Session) error
	ListSessions(ctx context.Context, agentID string, limit, offset int) ([]*Session, error)

	// Memory
	PutMemory(ctx context.Context, m *MemoryRecord) error
	GetMemory(ctx context.Context, agentID, key string) (*MemoryRecord, error)
	ListMemory(ctx context.Context, agentID string, kind string) ([]*MemoryRecord, error)
	DeleteMemory(ctx context.Context, id string) error

	// Audit logs
	AppendAuditLog(ctx context.Context, log *AuditLog) error
	ListAuditLogs(ctx context.Context, sessionID string, limit, offset int) ([]*AuditLog, error)

	// Traces
	InsertTrace(ctx context.Context, t *Trace) error
	GetTrace(ctx context.Context, id string) (*Trace, error)
	ListTraces(ctx context.Context, sessionID string) ([]*Trace, error)

	// Event ledger (append-only)
	AppendEvent(ctx context.Context, e *Event) error
	ListEvents(ctx context.Context, sessionID string, afterSeq int64) ([]*Event, error)

	// Checkpoints
	SaveCheckpoint(ctx context.Context, cp *Checkpoint) error
	GetCheckpoint(ctx context.Context, id string) (*Checkpoint, error)
	GetLatestCheckpoint(ctx context.Context, sessionID string) (*Checkpoint, error)
	ListCheckpoints(ctx context.Context, sessionID string) ([]*Checkpoint, error)

	// Lifecycle
	Migrate(ctx context.Context) error
	Close() error
}

Storage is the primary persistence interface. All adapters must implement this.

Multi-tenancy: every method is scoped to the tenant carried by its context (see WithTenant / TenantFromContext). Adapters stamp writes with, and filter reads by, the context tenant so that objects created under one tenant are never visible to another — including id-addressed reads such as GetSession, GetTrace and GetCheckpoint, which return a not-found error rather than another tenant's object. Callers that do not set a tenant operate under DefaultTenant, preserving single-tenant behavior.

type Trace

type Trace struct {
	ID        string    `json:"id"`
	TenantID  string    `json:"tenant_id,omitempty"`
	SessionID string    `json:"session_id"`
	ParentID  string    `json:"parent_id,omitempty"`
	Name      string    `json:"name"`
	Kind      string    `json:"kind"` // node, tool_call, model_call, approval
	Input     any       `json:"input,omitempty"`
	Output    any       `json:"output,omitempty"`
	Error     string    `json:"error,omitempty"`
	StartedAt time.Time `json:"started_at"`
	EndedAt   time.Time `json:"ended_at,omitempty"`
}

Trace represents a single trace span for observability.

type TracePage added in v0.4.0

type TracePage struct {
	Traces     []*Trace `json:"traces"`
	NextCursor string   `json:"next_cursor,omitempty"`
}

TracePage is a cursor-paginated slice of traces.

type VectorStore

type VectorStore interface {
	// Upsert inserts or updates embeddings in the given collection.
	Upsert(ctx context.Context, collection string, embeddings []Embedding) error

	// Search performs similarity search and returns top-k results. Optional
	// SearchOptions (e.g. WithFilter) scope the search; adapters apply any filter
	// so that top-k is selected over the matching subset.
	Search(ctx context.Context, collection string, query []float32, topK int, opts ...SearchOption) ([]SearchResult, error)

	// Delete removes embeddings by IDs.
	Delete(ctx context.Context, collection string, ids []string) error

	// CreateCollection ensures a collection exists.
	CreateCollection(ctx context.Context, name string, dimension int) error

	// Close releases resources.
	Close() error
}

VectorStore abstracts vector DB operations for RAG and embeddings.

Directories

Path Synopsis
adapters
chromadb
Package chromadb provides a ChromaDB-backed VectorStore adapter for Chronos.
Package chromadb provides a ChromaDB-backed VectorStore adapter for Chronos.
dynamo
Package dynamo provides a DynamoDB-backed Storage adapter for Chronos.
Package dynamo provides a DynamoDB-backed Storage adapter for Chronos.
lancedb
Package lancedb provides a LanceDB-backed VectorStore adapter for Chronos.
Package lancedb provides a LanceDB-backed VectorStore adapter for Chronos.
memory
Package memory provides an in-memory Storage adapter for Chronos.
Package memory provides an in-memory Storage adapter for Chronos.
milvus
Package milvus provides a Milvus-backed VectorStore adapter for Chronos.
Package milvus provides a Milvus-backed VectorStore adapter for Chronos.
mongo
Package mongo provides a MongoDB-backed Storage adapter for Chronos.
Package mongo provides a MongoDB-backed Storage adapter for Chronos.
pgvector
Package pgvector provides a PostgreSQL+pgvector-backed VectorStore adapter for Chronos.
Package pgvector provides a PostgreSQL+pgvector-backed VectorStore adapter for Chronos.
pinecone
Package pinecone provides a Pinecone-backed VectorStore adapter for Chronos.
Package pinecone provides a Pinecone-backed VectorStore adapter for Chronos.
postgres
Package postgres provides a PostgreSQL-backed Storage adapter for Chronos.
Package postgres provides a PostgreSQL-backed Storage adapter for Chronos.
qdrant
Package qdrant provides a Qdrant-backed VectorStore adapter for Chronos.
Package qdrant provides a Qdrant-backed VectorStore adapter for Chronos.
redis
Package redis provides a Redis-backed Storage adapter for Chronos.
Package redis provides a Redis-backed Storage adapter for Chronos.
redisvector
Package redisvector provides a Redis-backed VectorStore adapter using the RediSearch module (Redis Stack) for KNN vector similarity search.
Package redisvector provides a Redis-backed VectorStore adapter using the RediSearch module (Redis Stack) for KNN vector similarity search.
sqlite
Package sqlite provides a SQLite-backed Storage adapter for Chronos.
Package sqlite provides a SQLite-backed Storage adapter for Chronos.
weaviate
Package weaviate provides a Weaviate-backed VectorStore adapter for Chronos.
Package weaviate provides a Weaviate-backed VectorStore adapter for Chronos.
Package migrate provides versioned database migrations for SQL backends.
Package migrate provides versioned database migrations for SQL backends.

Jump to

Keyboard shortcuts

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