spine

Documentation

Overview

Package spine is the agent's canonical event log: one durable, ordered, replayable stream of events per run. It is the single source of truth from which state, audit, metrics, replay, and live progress are all derived - the "event spine" the architecture is built around.

Generalising the orchestration event spine into this foundation is the single biggest refactor-avoider: checkpoint/replay, rollback, monitoring, and human-in-the-loop approvals all project from one log instead of parallel, drifting ones. The durable SQLite Log (storage/sqlite) implements these semantics; MemoryLog is the in-memory reference they are matched against.

Index

• Constants
• Variables
• func Fold[S any](events []Event, initial S, reduce func(S, Event) S) S
• type ActorType
• type AppendInput
  • func (in AppendInput) Materialize(clk clock.Clock, seq int64) (Event, json.RawMessage, error)
• type Event
• type Log
• type MemoryLog
  • func NewMemoryLog(opts ...MemoryOption) *MemoryLog
  • func (l *MemoryLog) Append(_ context.Context, in AppendInput) (Event, error)
  • func (l *MemoryLog) LatestSnapshot(_ context.Context, stream string, upToSeq int64) (Snapshot, bool, error)
  • func (l *MemoryLog) Read(_ context.Context, q Query) ([]Event, error)
  • func (l *MemoryLog) SaveSnapshot(_ context.Context, s Snapshot) error
• type MemoryOption
  • func WithClock(c clock.Clock) MemoryOption
• type Query
• type Snapshot
• type SnapshotCodec
• type UpcastRegistry
  • func (r UpcastRegistry) Apply(e Event) Event
• type Upcaster

Constants

const DefaultSchemaVersion = 1

DefaultSchemaVersion is the version stamped on an event whose SchemaVersion is left unset (0) at append time. It is the baseline shape every event type starts at, so existing logs and callers that do not yet care about versioning all read as version 1.

Variables

var ErrPayloadConflict = errors.New("spine: AppendInput sets both Payload and RawPayload")

ErrPayloadConflict is returned by Append when an AppendInput carries both the decoded Payload and the pre-encoded RawPayload; a producer must pick one form.

Functions

func Fold

func Fold[S any](events []Event, initial S, reduce func(S, Event) S) S

Fold replays events through a reducer to project state: state is a function of the log. This is the core of the event-sourcing foundation - every view (progress, metrics, the current session) is a Fold over the spine.

Types

type ActorType

type ActorType string

ActorType identifies who produced an event.

const (
	// ActorAgent is the agent itself (a worker or coordinator run).
	ActorAgent ActorType = "agent"
	// ActorHuman is a person (an approval, a correction, an instruction).
	ActorHuman ActorType = "human"
	// ActorSystem is the runtime (scheduling, heartbeats, lifecycle).
	ActorSystem ActorType = "system"
)

type AppendInput

type AppendInput struct {
	Stream           string
	Type             string
	Actor            ActorType
	Payload          map[string]any
	RawPayload       json.RawMessage
	SchemaVersion    int
	Time             time.Time
	TraceID          string
	SpanID           string
	CausationID      string
	OriginInstanceID string
	Principal        string
}

AppendInput appends one event to a stream. The Log assigns Seq, assigns Time from its clock when Time is zero, and stamps SchemaVersion to DefaultSchemaVersion when it is left unset (0).

The payload is supplied in exactly one of two forms. Payload is the decoded form. RawPayload is its pre-encoded form: the JSON encoding of a single object, produced by a caller that already serialized its record once (a stamper) so the log does not re-encode it. A log stores the same bytes either way and always returns the decoded form on Event.Payload, so which form a producer used is unobservable downstream (readers, folds, and the canonical chain encoding see one shape). Setting both is a contract violation and Append rejects it.

func (AppendInput) Materialize

func (in AppendInput) Materialize(clk clock.Clock, seq int64) (Event, json.RawMessage, error)

Materialize builds the stored Event for this input at seq, applying the Log's defaulting rules - an unset Time is stamped from clk (as UTC), an unset SchemaVersion becomes DefaultSchemaVersion - and returns the payload in its canonical stored-bytes form alongside it: RawPayload verbatim when the producer pre-encoded it, else the JSON encoding of Payload. A durable backend stores those bytes and assigns Seq itself (the in-memory log counts under its lock; the SQLite log passes 0 and stamps Event.Seq from the row the database assigns), so both logs build their returned Event through this one method and a new Event field is threaded here once instead of in every backend. Setting both Payload and RawPayload is a contract violation and returns ErrPayloadConflict.

type Event

type Event struct {
	Stream  string
	Seq     int64
	Time    time.Time
	Type    string
	Actor   ActorType
	Payload map[string]any

	// SchemaVersion is the version of this event's payload shape for its Type.
	// Because Payload is a map, adding a key is backward compatible, but renaming,
	// removing, or retyping one is not. The version lets newer code recognise an
	// older payload and migrate it (see UpcastRegistry) instead of misreading it.
	// The log stamps an unset (0) version to DefaultSchemaVersion on append, so
	// every stored event carries an explicit version.
	SchemaVersion int

	// Trace linkage cross-references the OpenTelemetry span layer so ops traces
	// and the semantic spine line up. May be empty until the tracing adapter
	// populates them. CausationID is the event (or span) that caused this one,
	// enabling exact causal replay.
	TraceID     string
	SpanID      string
	CausationID string

	// OriginInstanceID is the instance that first produced this event. Set from
	// the start so multi-instance (fleet/P2P) sync never forces an ID refactor.
	OriginInstanceID string

	// Principal is the identity on whose authority the event was produced: which
	// agent in a fan-out, or which human in a multi-user host. It is the audit
	// "who", distinct from the coarse Actor kind (agent/human/system). Empty means
	// the standalone agent itself. Carried from the start so multi-agent and
	// multi-user attribution never forces an event-schema migration.
	Principal string
}

Event is one immutable, ordered record on a stream. Seq is monotonic within a stream; events are never mutated or deleted.

type Log

type Log interface {
	Append(ctx context.Context, in AppendInput) (Event, error)
	Read(ctx context.Context, q Query) ([]Event, error)
	// SaveSnapshot stores s, replacing any existing snapshot for (Stream, Seq).
	SaveSnapshot(ctx context.Context, s Snapshot) error
	// LatestSnapshot returns the newest snapshot for stream with Seq <= upToSeq and
	// true, or false when none exists. A non-positive upToSeq returns the newest
	// snapshot at any Seq.
	LatestSnapshot(ctx context.Context, stream string, upToSeq int64) (Snapshot, bool, error)
}

Log is the append-only event store - the spine port. Append assigns a monotonic Seq within a stream; Read returns events in Seq order. It also keeps stream snapshots: materialized checkpoints a Fold resumes from so reading state stays fast as a stream grows without bound (see Snapshot). A host supplies a durable implementation; the agent ships MemoryLog.

type MemoryLog

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

MemoryLog is an in-memory Log and SnapshotStore: ordered, per-stream monotonic Seq, safe for concurrent use. Its semantics are the contract the durable (SQLite) implementation must match.

func NewMemoryLog

func NewMemoryLog(opts ...MemoryOption) *MemoryLog

NewMemoryLog returns an empty in-memory Log.

func (*MemoryLog) Append

func (l *MemoryLog) Append(_ context.Context, in AppendInput) (Event, error)

Append implements Log. It builds the stored event through AppendInput.Materialize (the one place event defaulting and shape live) under the lock, so the assigned Seq stays monotonic within the stream.

func (*MemoryLog) LatestSnapshot

func (l *MemoryLog) LatestSnapshot(_ context.Context, stream string, upToSeq int64) (Snapshot, bool, error)

LatestSnapshot implements SnapshotStore. Returned payloads are cloned, so a caller may retain or mutate them freely.

func (*MemoryLog) Read

func (l *MemoryLog) Read(_ context.Context, q Query) ([]Event, error)

Read implements Log. Returned events share the stored payload maps; callers must treat Payload as read-only.

func (*MemoryLog) SaveSnapshot

func (l *MemoryLog) SaveSnapshot(_ context.Context, s Snapshot) error

SaveSnapshot implements SnapshotStore. It clones the payload so the stored snapshot is decoupled from the caller's slice, and replaces any snapshot already at (Stream, Seq).

type MemoryOption

type MemoryOption func(*MemoryLog)

MemoryOption configures a MemoryLog.

func WithClock

func WithClock(c clock.Clock) MemoryOption

WithClock sets the time source used to stamp events whose Time is unset (default: clock.System). Tests and replay pass a clock.Manual.

type Query

type Query struct {
	Stream   string
	AfterSeq int64 // exclusive lower bound; 0 returns from the start
	Limit    int   // <= 0 means no limit
}

Query reads a contiguous slice of a stream in Seq order.

type Snapshot

type Snapshot struct {
	Stream  string
	Seq     int64
	Payload []byte
}

Snapshot is a materialized projection of a stream up to and including Seq: a checkpoint a Fold resumes from instead of replaying the stream from the start, so reading state stays fast as a stream grows without bound. Payload is the domain's serialized state, opaque to the spine, so every projection (the resource store, a session) snapshots through one mechanism.

A snapshot never replaces events: the log stays the immutable source of truth, and a snapshot is a derived cache that can always be rebuilt by folding from an earlier point. So a missing snapshot is only slower, never wrong, and the spine Log keeps the snapshots alongside the events it checkpoints.

type SnapshotCodec

type SnapshotCodec interface {
	Seal(ctx context.Context, log Log, s Snapshot) (Snapshot, error)
	Open(ctx context.Context, s Snapshot) (Snapshot, error)
}

SnapshotCodec transforms snapshots between their projection form and their stored form, so a store can persist verified snapshots without the spine depending on any cryptography. Seal wraps a projection payload before it is saved (binding it to the log prefix it derives from); Open verifies and unwraps a stored payload before it is restored. A store with no codec stores payloads as-is. The chain package provides the signing implementation; an Open failure means the snapshot must not be trusted and the reader folds from the start instead - a rejected snapshot is only slower, never wrong.

type UpcastRegistry

type UpcastRegistry map[string]Upcaster

UpcastRegistry maps an event Type to the Upcaster that brings older payloads of that Type up to current. A consumer that folds historical events applies the registry first, so the reducer only ever sees current-shaped payloads. Holding the registry as an explicit value (rather than package-global mutable state) keeps upcasting deterministic across runs and replays.

func (UpcastRegistry) Apply

func (r UpcastRegistry) Apply(e Event) Event

Apply upcasts e if its Type has a registered Upcaster, otherwise returns it unchanged. The Upcaster itself decides, from e.SchemaVersion, whether any migration is needed.

type Upcaster

type Upcaster func(Event) Event

Upcaster migrates one event's payload from the shape it was stored in up to the shape current code expects for its Type. It must be deterministic: replay feeds the same stored events through the same upcasters and must reproduce identical results (see ARCHITECTURE.md, the determinism invariant). An Upcaster should branch on the event's SchemaVersion and return it unchanged when already current.