store

Documentation

Overview

Package store is the SQLite-backed implementation of session.Store.

Driver and dependencies

Uses modernc.org/sqlite, a pure-Go transpilation of SQLite. No CGo is required. ULIDs come from github.com/oklog/ulid/v2 (via internal/session).

Connection setup

Every database opened by Open runs the following PRAGMAs before any other query:

• journal_mode = WAL
• synchronous = NORMAL
• foreign_keys = ON
• busy_timeout = 5000

See pragmas.go for the implementation and the assertion that WAL actually engaged.

Migrations

Migrations are .sql files embedded from internal/store/migrations/. Each is applied inside its own transaction. Already-applied versions are tracked in schema_migrations and skipped on subsequent opens.

Fork-chain reads

MessagesForSession walks the fork chain by recursive CTE, returning the inherited prefix from each ancestor up to (and including) its fork point, concatenated with the local session's messages. See messages.go for the query. The recursion depth is capped at 8 to defend against cycles or pathological nesting; exceeding the cap returns ErrForkChainTooDeep.

Bus events

This package does NOT publish bus events. The agent layer (Task 11) observes store operations and publishes the corresponding events.

Index

• Constants
• Variables
• type Store
  • func Open(ctx context.Context, path string) (*Store, error)
  • func (s *Store) AddCompactionMarker(ctx context.Context, sessionID, beforeMessageID, summary string, ...) (*session.Marker, error)
  • func (s *Store) AppendMessage(ctx context.Context, sessionID string, in session.NewMessage) (*session.Message, error)
  • func (s *Store) Close() error
  • func (s *Store) CreateSession(ctx context.Context, in session.NewSession) (*session.Session, error)
  • func (s *Store) DB() *sql.DB
  • func (s *Store) ForgetSessionMemory(ctx context.Context, sessionID, memoryID string) (*session.Memory, error)
  • func (s *Store) ForkSession(ctx context.Context, fromSessionID, fromMessageID string, ...) (*session.Session, error)
  • func (s *Store) GetMessage(ctx context.Context, id string) (*session.Message, error)
  • func (s *Store) GetSession(ctx context.Context, id string) (*session.Session, error)
  • func (s *Store) GetSessionTodos(ctx context.Context, sessionID string) ([]session.TodoItem, session.TodoSummary, error)
  • func (s *Store) LatestMarker(ctx context.Context, sessionID string) (*session.Marker, error)
  • func (s *Store) LatestUserMessageID(ctx context.Context, sessionID string) (string, error)
  • func (s *Store) ListMarkersForSession(ctx context.Context, sessionID string) ([]*session.Marker, error)
  • func (s *Store) ListSessionMemories(ctx context.Context, sessionID string) ([]*session.Memory, error)
  • func (s *Store) ListSessions(ctx context.Context, opts session.ListOpts) ([]*session.Session, error)
  • func (s *Store) MessagesDirectForSession(ctx context.Context, sessionID string) ([]*session.Message, error)
  • func (s *Store) MessagesForSession(ctx context.Context, sessionID string) ([]*session.Message, error)
  • func (s *Store) MessagesSinceLatestMarker(ctx context.Context, sessionID string) ([]*session.Message, *session.Marker, error)
  • func (s *Store) PropagateTotals(ctx context.Context, id string, delta session.Totals) ([]string, error)
  • func (s *Store) RememberSessionMemory(ctx context.Context, sessionID string, in session.NewMemory) (*session.Memory, error)
  • func (s *Store) RenameSession(ctx context.Context, id, slug string) error
  • func (s *Store) ReplaceSessionTodos(ctx context.Context, sessionID string, items []session.TodoItem) (session.TodoSummary, error)
  • func (s *Store) SoftDeleteSession(ctx context.Context, id string) error
  • func (s *Store) UpdateSessionTotals(ctx context.Context, id string, delta session.Totals) error

Constants

const MaxForkDepth = 8

MaxForkDepth is the maximum number of ancestor sessions walked when resolving fork history. Anything deeper aborts with ErrForkChainTooDeep.

Variables

var ErrForkChainTooDeep = errors.New("store: fork chain exceeds maximum depth")

ErrForkChainTooDeep is returned when MessagesForSession's recursive CTE hits the depth cap (currently 8 ancestors).

var ErrNotFound = errors.New("store: not found")

ErrNotFound is returned when a lookup misses (no row with the given id).

Types

type Store

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

Store is a SQLite-backed session.Store.

func Open

func Open(ctx context.Context, path string) (*Store, error)

Open opens (or creates) a SQLite database at path, applies the connection PRAGMAs, and runs any pending migrations. Pass ":memory:" for an in-memory database.

The returned *Store is safe for concurrent use.

func (*Store) AddCompactionMarker

func (s *Store) AddCompactionMarker(
	ctx context.Context,
	sessionID, beforeMessageID, summary string,
	tokensSaved int64,
) (*session.Marker, error)

AddCompactionMarker records a new compaction cut-off for the session. The marker is never a deletion: the original messages remain in place, and the marker simply defines where the live context window starts.

func (*Store) AppendMessage

func (s *Store) AppendMessage(
	ctx context.Context, sessionID string, in session.NewMessage,
) (*session.Message, error)

AppendMessage inserts a new message row. The parts slice is encoded via session.MarshalParts; an invalid parts payload would be rejected by the CHECK (json_valid(parts)) constraint on the messages table.

When the message is the first user-role message for the session, the session's first_message_preview column is populated with up to 256 chars of the message's first text part. This supports fast substring filtering in ListSessions without a per-query JOIN.

func (*Store) Close

func (s *Store) Close() error

Close closes the underlying database. Safe to call multiple times.

func (*Store) CreateSession

func (s *Store) CreateSession(ctx context.Context, in session.NewSession) (*session.Session, error)

CreateSession persists a new session row. ID, CreatedAt, UpdatedAt are assigned by this method.

func (*Store) DB

func (s *Store) DB() *sql.DB

DB exposes the underlying *sql.DB for tests that need to issue raw SQL. Not part of session.Store; do not use outside of test or maintenance code.

func (*Store) ForgetSessionMemory

func (s *Store) ForgetSessionMemory(ctx context.Context, sessionID, memoryID string) (*session.Memory, error)

ForgetSessionMemory marks an active session memory as deleted.

func (*Store) ForkSession

func (s *Store) ForkSession(
	ctx context.Context,
	fromSessionID, fromMessageID string,
	model session.ModelRef,
	slug string,
) (*session.Session, error)

ForkSession creates a new session pointing at fromSessionID, inheriting history up to and including fromMessageID. The parent session need not be the message's owner — but the message must be visible to it via MessagesForSession. We perform a cheap existence check on both before inserting.

func (*Store) GetMessage

func (s *Store) GetMessage(ctx context.Context, id string) (*session.Message, error)

GetMessage returns the message row (deleted or not), or ErrNotFound.

func (*Store) GetSession

func (s *Store) GetSession(ctx context.Context, id string) (*session.Session, error)

GetSession returns the session row, or ErrNotFound.

func (*Store) GetSessionTodos

func (s *Store) GetSessionTodos(ctx context.Context, sessionID string) ([]session.TodoItem, session.TodoSummary, error)

GetSessionTodos returns the persisted todo list for sessionID.

func (*Store) LatestMarker

func (s *Store) LatestMarker(ctx context.Context, sessionID string) (*session.Marker, error)

LatestMarker returns the most recent compaction marker for the session, breaking ties by id (ULIDs are time-ordered so id-desc is the natural secondary sort). Returns (nil, nil) when no marker exists for the session.

func (*Store) LatestUserMessageID

func (s *Store) LatestUserMessageID(ctx context.Context, sessionID string) (string, error)

LatestUserMessageID returns the id of the most recent non-deleted user message in sessionID, or ("", nil) when no user message exists. Part of the extended store surface used by the session modal fork path.

func (*Store) ListMarkersForSession

func (s *Store) ListMarkersForSession(ctx context.Context, sessionID string) ([]*session.Marker, error)

ListMarkersForSession returns all compaction markers for the session in chronological order (oldest first). Returns an empty (non-nil) slice when no markers exist for the session.

func (*Store) ListSessionMemories

func (s *Store) ListSessionMemories(ctx context.Context, sessionID string) ([]*session.Memory, error)

ListSessionMemories returns active session memories in creation order.

func (*Store) ListSessions

func (s *Store) ListSessions(ctx context.Context, opts session.ListOpts) ([]*session.Session, error)

ListSessions returns sessions matching opts, newest first.

func (*Store) MessagesDirectForSession

func (s *Store) MessagesDirectForSession(ctx context.Context, sessionID string) ([]*session.Message, error)

MessagesDirectForSession returns only the messages directly owned by sessionID, without walking the fork chain. Soft-deleted messages are excluded. Used for KindSubagent sessions that have a fresh history independent of their parent.

func (*Store) MessagesForSession

func (s *Store) MessagesForSession(ctx context.Context, sessionID string) ([]*session.Message, error)

MessagesForSession returns the full conversation visible to sessionID, walking ancestor sessions and selecting each one's contribution up to its fork point. Soft-deleted messages are excluded.

The depth of the fork chain is capped at MaxForkDepth (=8); deeper chains return ErrForkChainTooDeep.

func (*Store) MessagesSinceLatestMarker

func (s *Store) MessagesSinceLatestMarker(
	ctx context.Context, sessionID string,
) ([]*session.Message, *session.Marker, error)

MessagesSinceLatestMarker returns the messages newer than the session's latest compaction marker, plus the marker itself. When the session has no marker, the full conversation history is returned with a nil marker.

We use the marker's before_message_id as the filter cursor: because every id is a ULID, "id > before_message_id" is equivalent to "created strictly after the message at the cut-off". This pushes the filter into the same SQL query that loads history, avoiding a round-trip-and-slice in Go.

func (*Store) PropagateTotals

func (s *Store) PropagateTotals(ctx context.Context, id string, delta session.Totals) ([]string, error)

PropagateTotals adds delta to every session in the parent chain starting at id (inclusive) and walking up via parent_id. The chain walk uses a recursive CTE capped at 32 hops to guard against accidental parent_id cycles (impossible by schema constraints, but the cap is a defensive belt-and-suspenders measure). All row updates happen inside a single transaction so the chain is updated atomically; a partial write is an acceptable failure mode — the next delta will catch up.

Returns the slice of session ids that were updated, ordered leaf-first (id is always index 0; the root ancestor is last).

func (*Store) RememberSessionMemory

func (s *Store) RememberSessionMemory(ctx context.Context, sessionID string, in session.NewMemory) (*session.Memory, error)

RememberSessionMemory stores a new active memory scoped to sessionID.

func (*Store) RenameSession

func (s *Store) RenameSession(ctx context.Context, id, slug string) error

RenameSession sets a new slug on an existing session and bumps UpdatedAt. An empty slug clears the slug (sets the column to NULL). Returns ErrNotFound when id is unknown. A no-op when the new slug equals the current slug (UpdatedAt is not bumped in that case).

func (*Store) ReplaceSessionTodos

func (s *Store) ReplaceSessionTodos(ctx context.Context, sessionID string, items []session.TodoItem) (session.TodoSummary, error)

ReplaceSessionTodos stores the complete todo list for sessionID.

func (*Store) SoftDeleteSession

func (s *Store) SoftDeleteSession(ctx context.Context, id string) error

SoftDeleteSession marks deleted_at and bumps updated_at. No-op if already deleted.

func (*Store) UpdateSessionTotals

func (s *Store) UpdateSessionTotals(ctx context.Context, id string, delta session.Totals) error

UpdateSessionTotals atomically adds delta to the session's running totals. Single-statement, single round-trip.