storage

package
v0.1.3 Latest Latest
Warning

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

 Go to latest Published: Aug 30, 2026 License: AGPL-3.0 Imports: 26 Imported by: 0

Documentation

Overview

Package storage provides the validation seam for durable agent events.

Package storage provides the validation seam for durable agent events.

Index

Constants

View Source 
const KindRunDeleted = "run_deleted"

KindRunDeleted is the deletion-tombstone event kind written by AppendAndDeleteRun. A run whose only remaining events are tombstones of this kind is free for re-admission by AppendBatchForNewRun; any other surviving event means the run is still live and admission is refused.

Variables

View Source 
var (
	ErrDuplicate       = errors.New("duplicate event")
	ErrClaimHeld       = errors.New("run claim held by another holder")
	ErrClaimNotHeld    = errors.New("run claim not held by this holder")
	ErrContentNotFound = errors.New("content not found")
)
View Source 
var ErrQueueFull = errors.New("storage queue full")

Functions

func NewUsageWriter 

func NewUsageWriter(store *SQLite, workspaceID string) usage.UsageWriter

NewUsageWriter returns a usage.UsageWriter that records into store, scoped to workspaceID.

Types

type BatchAppender 

type BatchAppender interface {
	AppendBatch(context.Context, []Event) error
}

BatchAppender atomically appends a set of events.

type Claim 

type Claim struct {
	RunID      string
	Holder     string
	AcquiredAt string
	Fence      uint64
}

Claim represents an exclusive execution claim on a run.

type ClaimReader 

type ClaimReader interface {
	// GetClaim returns the current claim for runID. ErrClaimNotHeld when the
	// run has no claim.
	GetClaim(context.Context, string) (Claim, error)
}

ClaimReader is the optional extension for reading a run's current execution claim without mutating it. It backs read-only liveness probes (sidebar claim age, delivery_pending heartbeat); a backend that cannot expose claims simply does not implement it.

type Event 

type Event struct {
	ID       string
	RunID    string
	Sequence int
	Kind     string
	Payload  []byte
	// RowID is the event's position in the store's global append order: the
	// SQLite rowid, or the monotone append index on the memory backend. It is
	// set by the store when events are read so a reader can fold events from
	// several runs in the order they were actually appended - in particular a
	// run_deleted tombstone always precedes a later run_created that reuses
	// its idempotency key.
	RowID uint64
}

type ExistingClaimAppender 

type ExistingClaimAppender interface {
	AppendWithExistingClaim(context.Context, Event, string) error
}

ExistingClaimAppender appends only when holder owns an existing claim. Unlike Store.AppendClaimed, an unclaimed run is refused.

type FencedLeaseStore 

type FencedLeaseStore interface {
	ClaimRunFenced(context.Context, string, string) (Claim, error)
	TakeoverExpiredClaimFenced(context.Context, string, string, time.Duration) (Claim, error)
	// TakeoverClaimFenced atomically replaces any existing claim with holder,
	// bumping the claim fence so a prior holder's captured fence no longer
	// authorizes writes, and returning the new claim.
	TakeoverClaimFenced(context.Context, string, string) (Claim, error)
	// RefreshClaimFenced refreshes the claim's acquired_at ONLY when holder
	// already owns the claim row, returning that claim. It never inserts a
	// missing row: a holder whose claim is gone is reported as
	// ErrClaimNotHeld, so a displaced/expired holder cannot reclaim itself
	// through the heartbeat.
	RefreshClaimFenced(context.Context, string, string) (Claim, error)
	AppendClaimedFenced(context.Context, Event, Claim) error
	ReleaseClaimFenced(context.Context, Claim) error
}

FencedLeaseStore guards stale writes after an expired claim changes owner.

type LeaseStore 

type LeaseStore interface {
	TakeoverExpiredClaim(context.Context, string, string, time.Duration) error
}

LeaseStore is the optional extension used by workflow recovery.

type Memory 

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

func NewMemory 

func NewMemory() *Memory

func (*Memory) Append 

func (m *Memory) Append(_ context.Context, e Event) error

func (*Memory) AppendAndDeleteRun 

func (m *Memory) AppendAndDeleteRun(_ context.Context, tombstone Event, claim Claim) error

AppendAndDeleteRun appends a deletion tombstone and deletes earlier events and the claim while holding one lock.

func (*Memory) AppendBatch 

func (m *Memory) AppendBatch(_ context.Context, events []Event) error

func (*Memory) AppendBatchForNewRun 

func (m *Memory) AppendBatchForNewRun(_ context.Context, runID string, events []Event) error

func (*Memory) AppendClaimed 

func (m *Memory) AppendClaimed(_ context.Context, e Event, holder string) error

AppendClaimed atomically checks the run claim and appends the event.

func (*Memory) AppendClaimedFenced 

func (m *Memory) AppendClaimedFenced(_ context.Context, e Event, claim Claim) error

func (*Memory) AppendWithExistingClaim 

func (m *Memory) AppendWithExistingClaim(_ context.Context, e Event, holder string) error

func (*Memory) Changes 

func (m *Memory) Changes(_ context.Context, afterCursor uint64) (map[string]int, uint64, error)

func (*Memory) ClaimRun 

func (m *Memory) ClaimRun(_ context.Context, runID, holder string) error

func (*Memory) ClaimRunFenced 

func (m *Memory) ClaimRunFenced(_ context.Context, runID, holder string) (Claim, error)

func (*Memory) ClearClaim 

func (m *Memory) ClearClaim(_ context.Context, runID string) error

func (*Memory) Close 

func (m *Memory) Close() error

func (*Memory) Count 

func (m *Memory) Count(_ context.Context) (int, error)

func (*Memory) DeleteRun 

func (m *Memory) DeleteRun(_ context.Context, runID string, throughSequence int) error

func (*Memory) Events 

func (m *Memory) Events(_ context.Context, runID string) ([]Event, error)

func (*Memory) EventsSince 

func (m *Memory) EventsSince(_ context.Context, runID string, afterSequence int) ([]Event, error)

func (*Memory) GetClaim 

func (m *Memory) GetClaim(_ context.Context, runID string) (Claim, error)

func (*Memory) GetContent 

func (m *Memory) GetContent(_ context.Context, ref string) ([]byte, error)

func (*Memory) IsRunHeld 

func (m *Memory) IsRunHeld(_ context.Context, runID string) (bool, error)

IsRunHeld reports whether runID currently has an active claim. A pure liveness probe; it never acquires, refreshes, or releases a claim.

func (*Memory) IsRunTokenFenced 

func (m *Memory) IsRunTokenFenced(_ context.Context, runID, token string) (bool, error)

IsRunTokenFenced reports whether token has been fenced out of runID by a subsequent takeover. The history is durable across releases. A token that is the current holder of runID always reads false.

func (*Memory) ListRunIDs 

func (m *Memory) ListRunIDs(_ context.Context) ([]string, error)

func (*Memory) PutContent 

func (m *Memory) PutContent(_ context.Context, ref string, data []byte) error

func (*Memory) RefreshClaimFenced 

func (m *Memory) RefreshClaimFenced(_ context.Context, runID, holder string) (Claim, error)

RefreshClaimFenced refreshes the claim's acquired_at ONLY when holder already owns the claim row. A missing row (or a row owned by another holder) returns ErrClaimNotHeld, so a heartbeat can never insert itself back into a claim it lost (F2).

func (*Memory) ReleaseClaim 

func (m *Memory) ReleaseClaim(_ context.Context, runID, holder string) error

func (*Memory) ReleaseClaimFenced 

func (m *Memory) ReleaseClaimFenced(_ context.Context, claim Claim) error

func (*Memory) TakeoverClaim 

func (m *Memory) TakeoverClaim(_ context.Context, runID, holder string) error

func (*Memory) TakeoverClaimFenced 

func (m *Memory) TakeoverClaimFenced(_ context.Context, runID, holder string) (Claim, error)

func (*Memory) TakeoverExpiredClaim 

func (m *Memory) TakeoverExpiredClaim(_ context.Context, runID, holder string, maxAge time.Duration) error

func (*Memory) TakeoverExpiredClaimFenced 

func (m *Memory) TakeoverExpiredClaimFenced(_ context.Context, runID, holder string, maxAge time.Duration) (Claim, error)

type NewRunBatchAppender 

type NewRunBatchAppender interface {
	AppendBatchForNewRun(context.Context, string, []Event) error
}

NewRunBatchAppender atomically appends a new-run batch only when no event or claim exists for its run. It is the atomic admission boundary.

type Options added in v0.1.1 

type Options struct {
	// Harden marks the path as an ad-hoc, OS-temp-dir-backed store (see
	// config.TempStorePath): the directory chain is created 0700 before the
	// sqlite file exists and the file is chmod 0600 after open, failing
	// closed on error. WAL mode still creates transient -wal/-shm sidecars,
	// but only inside the hardened 0700 directory and Close folds the WAL
	// back into the main file, so the directory is the access boundary.
	Harden bool
}

Options tunes OpenSQLiteWithOptions.

type QueueMetrics 

type QueueMetrics struct {
	Submitted, Committed, Rejected uint64
	TotalWait                      time.Duration
	MaxWait                        time.Duration
}

type QueuedWriter 

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

QueuedWriter provides bounded backpressure around a Store for validation.

func NewQueuedWriter 

func NewQueuedWriter(store Store, capacity int) *QueuedWriter

func (*QueuedWriter) Close 

func (w *QueuedWriter) Close() error

func (*QueuedWriter) Metrics 

func (w *QueuedWriter) Metrics() QueueMetrics

func (*QueuedWriter) Submit 

func (w *QueuedWriter) Submit(ctx context.Context, event Event) error

type SQLite 

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

func OpenSQLite 

func OpenSQLite(path string) (*SQLite, error)

OpenSQLite opens the store with default options.

func OpenSQLiteWithOptions added in v0.1.1 

func OpenSQLiteWithOptions(path string, opts Options) (*SQLite, error)

func (*SQLite) AbandonWorktreeCreation 

func (s *SQLite) AbandonWorktreeCreation(ctx context.Context, principal contextstate.Principal, instance contextstate.WorktreeInstance) error

AbandonWorktreeCreation removes a reservation after Git creation fails.

func (*SQLite) Advance 

func (s *SQLite) Advance(ctx context.Context, request contextstate.AdvanceRequest) error

func (*SQLite) Append 

func (s *SQLite) Append(ctx context.Context, e Event) error

func (*SQLite) AppendAndDeleteRun 

func (s *SQLite) AppendAndDeleteRun(ctx context.Context, tombstone Event, claim Claim) error

AppendAndDeleteRun appends a deletion tombstone and deletes earlier events and the claim in one SQLite transaction.

func (*SQLite) AppendBatch 

func (s *SQLite) AppendBatch(ctx context.Context, events []Event) error

func (*SQLite) AppendBatchForNewRun 

func (s *SQLite) AppendBatchForNewRun(ctx context.Context, runID string, events []Event) error

func (*SQLite) AppendClaimed 

func (s *SQLite) AppendClaimed(ctx context.Context, e Event, holder string) error

AppendClaimed atomically checks the run claim and appends the event.

func (*SQLite) AppendClaimedFenced 

func (s *SQLite) AppendClaimedFenced(ctx context.Context, e Event, claim Claim) error

func (*SQLite) AppendWithExistingClaim 

func (s *SQLite) AppendWithExistingClaim(ctx context.Context, e Event, holder string) error

func (*SQLite) Backup 

func (s *SQLite) Backup(ctx context.Context, d string) error

func (*SQLite) BeginWorktreeAdoption 

func (s *SQLite) BeginWorktreeAdoption(ctx context.Context, principal contextstate.Principal, instance contextstate.WorktreeInstance, canonicalPath string) error

BeginWorktreeAdoption reserves an exact legacy route for adoption.

func (*SQLite) BeginWorktreeCreation 

func (s *SQLite) BeginWorktreeCreation(ctx context.Context, principal contextstate.Principal, instance contextstate.WorktreeInstance, canonicalPath string) error

BeginWorktreeCreation reserves one worktree name before Git creates its directory. A live record with the same name blocks same-name reuse.

func (*SQLite) BeginWorktreeDeletion 

func (s *SQLite) BeginWorktreeDeletion(ctx context.Context, principal contextstate.Principal, instance contextstate.WorktreeInstance) error

BeginWorktreeDeletion fences the exact active physical worktree instance.

func (*SQLite) CalibrationSeed 

func (s *SQLite) CalibrationSeed(ctx context.Context, workspaceID, provider, model string) (float64, bool, error)

CalibrationSeed returns the estimate-vs-actual correction ratio observed for a (provider, model) binding in this workspace, so a freshly started process can plan its FIRST request with the correction it already learned instead of starting blind at 1.0.

Without this the ratio was written to token_usage_events on every turn and never read back: every process, every session and every resume began assuming the len(s)/4 estimate was exact. For payloads that are mostly code and JSON tool schemas that estimate runs ~1.7x low, so the first request sailed past the compaction trigger and the next one repaid the whole error at once - the sequence that destroyed a real session's context.

The ratio is aggregated (sum of actual over sum of estimated) rather than averaged per row, so large requests - the ones that actually approach the budget and matter for compaction - carry proportionate weight. ok is false when the binding has no usable observation yet; the caller then keeps the uncorrected default.

func (*SQLite) Changes 

func (s *SQLite) Changes(ctx context.Context, after uint64) (map[string]int, uint64, error)

func (*SQLite) CheckSpoolGrant 

func (s *SQLite) CheckSpoolGrant(ctx context.Context, ref, principal string) (bool, error)

CheckSpoolGrant reports whether principal holds a durable read grant on ref.

func (*SQLite) ClaimRun 

func (s *SQLite) ClaimRun(ctx context.Context, id, h string) error

func (*SQLite) ClaimRunFenced 

func (s *SQLite) ClaimRunFenced(ctx context.Context, id, h string) (Claim, error)

func (*SQLite) ClearClaim 

func (s *SQLite) ClearClaim(ctx context.Context, id string) error

func (*SQLite) Close 

func (s *SQLite) Close() error

Close folds the WAL back into the main database file before closing the connection pool. Without the checkpoint, WAL-mode leaves -wal/-shm files on disk that a caller's own directory cleanup (e.g. t.TempDir()) can race against, since a bare db.Close() gives no guarantee those files are done being written to by the time it returns. Idempotent like sql.DB.Close(): a repeat call is a no-op that returns the first call's result, rather than a "database is closed" error from re-running the checkpoint query.

func (*SQLite) Commit 

func (s *SQLite) Commit(ctx context.Context, request contextstate.CommitRequest) error

func (*SQLite) Compact 

func (s *SQLite) Compact(ctx context.Context) error

Compact rewrites the database file. It does three things a running store cannot do for itself:

  • reclaims the pages a prune freed. SQLite moves deleted pages to the freelist and never shrinks the file on DELETE alone, so retention without this bounds growth without reducing size;
  • adopts compactPageSize, which is fixed when the file is created and can only change across a VACUUM;
  • switches auto_vacuum to INCREMENTAL, which needs the same full rebuild because it adds a pointer map to every page.

All of it runs on one pinned connection: PRAGMA page_size and auto_vacuum apply per connection, and a pooled VACUUM could otherwise land on a different connection that never saw them. The store leaves WAL mode for the rewrite, because SQLite refuses a page_size change in WAL, and returns to it afterwards.

VACUUM cannot run inside a transaction and rewrites the whole file, so this belongs in an explicit maintenance command, never on the open path.

func (*SQLite) Count 

func (s *SQLite) Count(ctx context.Context) (int, error)

func (*SQLite) CreatingWorktreeInstance 

func (s *SQLite) CreatingWorktreeInstance(ctx context.Context, principal contextstate.Principal, worktree string) (contextstate.WorktreeInstanceInfo, error)

CreatingWorktreeInstance returns the retained creation record for a name.

func (*SQLite) DeleteRun 

func (s *SQLite) DeleteRun(ctx context.Context, id string, through int) error

func (*SQLite) DeleteSession 

func (s *SQLite) DeleteSession(ctx context.Context, principal contextstate.Principal, sessionID string) (contextstate.DeleteResult, error)

func (*SQLite) DeleteSessionSnapshot 

func (s *SQLite) DeleteSessionSnapshot(ctx context.Context, principal contextstate.Principal, name string) error

func (*SQLite) DeleteWorktreeRoute 

func (s *SQLite) DeleteWorktreeRoute(ctx context.Context, principal contextstate.Principal, worktree string) (int64, error)

DeleteWorktreeRoute removes a launch route after its Git worktree is gone. It reports how many rows it removed.

func (*SQLite) DeleteWorktreeRoutesByName 

func (s *SQLite) DeleteWorktreeRoutesByName(ctx context.Context, principal contextstate.Principal, worktree string) (int64, error)

DeleteWorktreeRoutesByName removes every launch route for one worktree name, whether bound to an instance or legacy. It reports how many rows it removed. Call it only when no live instance owns the name, so no active route can be affected.

func (*SQLite) DeleteWorktreeSessionSnapshot 

func (s *SQLite) DeleteWorktreeSessionSnapshot(ctx context.Context, p contextstate.Principal, n string, i contextstate.WorktreeInstance) error

func (*SQLite) DeleteWorktreeSessions 

func (s *SQLite) DeleteWorktreeSessions(ctx context.Context, principal contextstate.Principal, instance contextstate.WorktreeInstance) (int, error)

DeleteWorktreeSessions completes a fenced deletion. The lifecycle row is exact-scoped so a retry for an old instance cannot change a replacement.

func (*SQLite) DeletingWorktreeInstance 

func (s *SQLite) DeletingWorktreeInstance(ctx context.Context, principal contextstate.Principal, worktree string) (contextstate.WorktreeInstance, error)

DeletingWorktreeInstance returns the retained deletion record for a name.

func (*SQLite) EnsureSession 

func (s *SQLite) EnsureSession(ctx context.Context, request contextstate.EnsureSessionRequest) error

EnsureSession creates the zero-revision context head and binds it to the owner capability. Existing heads are idempotent only for the same owner and binding.

func (*SQLite) Events 

func (s *SQLite) Events(ctx context.Context, id string) ([]Event, error)

func (*SQLite) EventsSince 

func (s *SQLite) EventsSince(ctx context.Context, id string, after int) ([]Event, error)

func (*SQLite) ExportSession 

func (s *SQLite) ExportSession(ctx context.Context, principal contextstate.Principal, sessionID string) (contextstate.ExportResult, error)

func (*SQLite) FirstUserMessage 

func (s *SQLite) FirstUserMessage(ctx context.Context, principal contextstate.Principal, sessionID string) (string, error)

FirstUserMessage returns the first user message of a live context session, derived from the oldest complete checkpoint's active context. It is used to title untitled sessions in the picker. The lookup is subject-scoped: an older run's capability digest is stale by design, but its opener text is still readable by the subject that owns the catalog row.

The whole oldest checkpoint is read: for a forked continuation the oldest checkpoint holds the full loaded history, which can exceed 64 KiB, and a byte-sliced prefix would break the JSON parse and void the title. A session with no complete checkpoint or no user message yields an empty string, never an error.

func (*SQLite) GetClaim 

func (s *SQLite) GetClaim(ctx context.Context, id string) (Claim, error)

GetClaim reads a run's current execution claim as a read-only liveness probe. It never acquires, refreshes, or releases the claim. It is split into its own file to keep sqlite.go under the line budget.

func (*SQLite) GetContent 

func (s *SQLite) GetContent(ctx context.Context, ref string) ([]byte, error)

GetContent retrieves bytes previously stored by PutContent. Returns ErrContentNotFound if the ref is unknown.

func (*SQLite) GrantSpool 

func (s *SQLite) GrantSpool(ctx context.Context, ref, principal string) error

GrantSpool durably records that principal holds a read grant on a remainder ref. INSERT OR IGNORE keeps the first grant for a (ref, principal) pair, so re-spooling the same ref for the same principal is idempotent.

func (*SQLite) ImportSource 

func (s *SQLite) ImportSource(ctx context.Context, principal contextstate.Principal, legacyID, operationKey string, events []contextstate.SourceEvent, payloads []contextstate.PayloadRecord) (contextstate.ImportResult, error)

ImportSource is the explicit all-or-nothing adapter for legacy JSONL data. It is separate from appendSourceEvents so checkpoint publication remains private to the context store transaction.

func (*SQLite) IsRunHeld 

func (s *SQLite) IsRunHeld(ctx context.Context, runID string) (bool, error)

IsRunHeld reports whether runID currently has an active claim row. It is a pure liveness probe; it never acquires, refreshes, or releases a claim, so observing a run can never disturb its holder. Returns (false, nil) when no claim row exists.

func (*SQLite) IsRunTokenFenced 

func (s *SQLite) IsRunTokenFenced(ctx context.Context, runID, token string) (bool, error)

IsRunTokenFenced reports whether token has been fenced out of runID by a subsequent takeover. The history is durable: a fenced token stays fenced across releases, so a re-issued claim by the same token reads false UNLESS the token has been fenced by an intervening takeover. A token that is the current holder of runID always reads false.

func (*SQLite) ListCreatingWorktreeInstances 

func (s *SQLite) ListCreatingWorktreeInstances(ctx context.Context, principal contextstate.Principal) ([]contextstate.WorktreeInstanceInfo, error)

func (*SQLite) ListDeletingWorktreeInstances 

func (s *SQLite) ListDeletingWorktreeInstances(ctx context.Context, principal contextstate.Principal) ([]contextstate.WorktreeInstanceInfo, error)

func (*SQLite) ListRunIDs 

func (s *SQLite) ListRunIDs(ctx context.Context) ([]string, error)

func (*SQLite) ListSessions 

func (s *SQLite) ListSessions(ctx context.Context, principal contextstate.Principal) ([]contextstate.SessionCatalogInfo, error)

func (*SQLite) ListWorktreeSessions 

func (s *SQLite) ListWorktreeSessions(ctx context.Context, principal contextstate.Principal, instance contextstate.WorktreeInstance) ([]contextstate.SessionCatalogInfo, error)

ListWorktreeSessions lists snapshots for one active worktree instance.

func (*SQLite) LiveWorktreeInstance 

func (s *SQLite) LiveWorktreeInstance(ctx context.Context, principal contextstate.Principal, worktree string) (contextstate.WorktreeInstanceInfo, error)

LiveWorktreeInstance returns the one non-deleted instance for a worktree.

func (*SQLite) Load 

func (s *SQLite) Load(ctx context.Context, principal contextstate.Principal, sessionID string) (contextstate.Snapshot, error)

func (*SQLite) LoadSession 

func (s *SQLite) LoadSession(ctx context.Context, principal contextstate.Principal, name string) ([]byte, contextstate.SessionCatalogInfo, error)

func (*SQLite) LoadSessionAdmission 

func (s *SQLite) LoadSessionAdmission(ctx context.Context, principal contextstate.Principal, name string) (contextstate.SessionAdmission, error)

LoadSessionAdmission returns the stored admission record. A session with no row yields the zero value and a nil error: no admissions is a normal state.

func (*SQLite) LoadWorktree 

func (s *SQLite) LoadWorktree(ctx context.Context, principal contextstate.Principal, sessionID string, instance contextstate.WorktreeInstance) (contextstate.Snapshot, error)

LoadWorktree loads a durable session only while its exact instance is active.

func (*SQLite) LoadWorktreeSession 

func (s *SQLite) LoadWorktreeSession(ctx context.Context, p contextstate.Principal, n string, i contextstate.WorktreeInstance) ([]byte, contextstate.SessionCatalogInfo, error)

func (*SQLite) LoadWorktreeSessionAdmission 

func (s *SQLite) LoadWorktreeSessionAdmission(ctx context.Context, p contextstate.Principal, n string, i contextstate.WorktreeInstance) (contextstate.SessionAdmission, error)

func (*SQLite) Path 

func (s *SQLite) Path() string

Path returns the database file path this store was opened with - e.g. for internal/hub, which places hub.lock/hub.sock beside it.

func (*SQLite) PruneContextPayloads 

func (s *SQLite) PruneContextPayloads(ctx context.Context, now time.Time, limit int) (int, error)

PruneContextPayloads removes only revoked payload rows whose retention expiry has elapsed. Tombstones and audit records are compliance records and are intentionally outside this maintenance operation.

func (*SQLite) PruneOrphanedContent 

func (s *SQLite) PruneOrphanedContent(ctx context.Context) (removed int, err error)

PruneOrphanedContent deletes every workflow-ledger content row ("sha256:" prefix) that no live event payload references any longer.

AppendAndDeleteRun (DeleteRun's storage layer) strips a deleted run down to a tombstone event, but never touches the content table: the run's output/error/diff blobs it referenced become permanently unreachable (no live event payload names them) yet permanently retained (nothing ever deletes them) - a live finding from a session that mass-deleted dozens of stacked-delivery runs and left their content orphaned.

Safe by construction: it computes the live set by scanning every row CURRENTLY in the events table (a deleted run's real events are already gone, only its tombstone remains, so its content refs are already unreachable and correctly excluded) and only ever deletes rows matching workflowContentRefPattern - a coordinator/subagent/chat content row (a different ref prefix) is never a match and is never considered.

func (*SQLite) PruneSessionCheckpoints 

func (s *SQLite) PruneSessionCheckpoints(ctx context.Context, now time.Time, retention time.Duration, keep, limit int) (int, error)

PruneSessionCheckpoints deletes complete checkpoint rows older than retention, keeping per session the active checkpoint, the earliest complete checkpoint, and the newest keep rows. It returns the number of rows removed.

This is the only bound on context_checkpoints. Each row carries a full active_context blob, so without a sweep the table grows with every committed turn and never shrinks; on one real store it reached 144 MB of a 311 MB database in ten days.

The sweep is idempotent and bounded by limit. A caller that wants the table fully swept loops until the result is below limit.

func (*SQLite) PruneSessionSnapshots 

func (s *SQLite) PruneSessionSnapshots(ctx context.Context, principal contextstate.Principal, names []string) error

func (*SQLite) PruneWorktreeInstances 

func (s *SQLite) PruneWorktreeInstances(ctx context.Context, now time.Time, retention time.Duration, limit int) (int, int, error)

PruneWorktreeInstances removes worktree instances in a reapable state whose last update is older than retention, the routes bound to them, and any route whose instance_id no longer resolves to an instance row.

Nothing bounded these tables before. Each managed worktree leaves rows behind permanently: on one real store they held 70,841 instances across 57,077 distinct workspace ids, against a handful of live worktrees, for 34 MB of rows plus 22 MB of indexes.

Returns the instance and route counts removed. Bounded by limit and idempotent, so a caller loops until a sweep comes back short.

func (*SQLite) PruneWorktreeSessionSnapshots 

func (s *SQLite) PruneWorktreeSessionSnapshots(ctx context.Context, p contextstate.Principal, names []string, i contextstate.WorktreeInstance) error

func (*SQLite) PutContent 

func (s *SQLite) PutContent(ctx context.Context, ref string, data []byte) error

PutContent stores raw bytes keyed by a content-addressed reference (e.g. "ref:output:xxxx"). Idempotent for the same ref.

func (*SQLite) ReactivateWorktreeInstance 

func (s *SQLite) ReactivateWorktreeInstance(ctx context.Context, principal contextstate.Principal, instance contextstate.WorktreeInstance) error

ReactivateWorktreeInstance restores an instance only when deletion did not remove its Git worktree. Callers use it after a failed external removal.

func (*SQLite) ReadPayload 

func (s *SQLite) ReadPayload(ctx context.Context, principal contextstate.Principal, ref contextstate.ContentRef) (contextstate.SanitizedPayload, error)

func (*SQLite) ReadRange 

func (s *SQLite) ReadRange(ctx context.Context, principal contextstate.Principal, sourceRange contextstate.SourceRange) ([]contextstate.SourceEvent, error)

func (*SQLite) ReclaimSession 

func (s *SQLite) ReclaimSession(ctx context.Context, principal contextstate.Principal, sessionID string) (contextstate.Snapshot, error)

ReclaimSession transfers write ownership of an existing, non-tombstoned live context session to principal's own freshly minted capability, then returns its current snapshot. It exists because Principal.capability is minted fresh and random per process and never persisted anywhere it could be recovered - a later process resuming a session by id (an id it learned through LoadSession/ListSessions, both scoped only to workspace+subject with no capability check) has no way to reconstruct the capability the original process held, so authorizing every other durable-write path on an exact capability match would make cross-process resume impossible by construction. Reclaiming is scoped the same way those reads already are: knowing the session's id, workspace and subject is what LoadSession and DeleteSessionSnapshot already treat as sufficient authority for the same session, so extending that authority to "take over its capability" adds no new trust boundary.

A managed worktree session is rejected: those are addressed by name through the chat_sessions catalog (worktree_catalog_keys), never through this capability-gated context_sessions row, so reclaiming one here would be meaningless.

The takeover deliberately does NOT stamp a fresh lease_at - only a real heartbeat tick (RenewLease) may mark a lease fresh. Stamping here (an earlier version did) meant every successful reclaim, even a totally uncontested one, poisoned the row against any other reclaim for the next sessionLeaseTTL - breaking one-shot commands (mivia compact, a quick chat -p turn) that never renew a lease at all. Tradeoff accepted instead: a THIRD process reclaiming within the sub-heartbeat-interval window right after this takeover can still succeed (benign churn, loud ErrPrincipalMismatch on the loser's next write) rather than the silent eviction this feature exists to prevent.

func (*SQLite) RecordUsageEvent 

func (s *SQLite) RecordUsageEvent(ctx context.Context, workspaceID string, record usage.UsageRecord) error

RecordUsageEvent durably records one token/cache/compaction usage measurement. One INSERT, its own transaction - mirrors every other durable write in this package (writeMu-serialized, retried on a transient busy lock, since a session's own store can be written by more than one mivia process sharing a workspace, same as every other table here).

func (*SQLite) RefreshClaimFenced 

func (s *SQLite) RefreshClaimFenced(ctx context.Context, id, h string) (Claim, error)

RefreshClaimFenced refreshes the claim's acquired_at ONLY when holder already owns the claim row. A missing row (or a row owned by another holder) refreshes nothing and returns ErrClaimNotHeld, so a heartbeat can never insert itself back into a claim it lost (F2).

func (*SQLite) RegisterAdoptedWorktreeInstance 

func (s *SQLite) RegisterAdoptedWorktreeInstance(ctx context.Context, principal contextstate.Principal, instance contextstate.WorktreeInstance, canonicalPath string) error

RegisterAdoptedWorktreeInstance activates an adoption reservation only if the legacy route remains exact and unbound.

func (*SQLite) RegisterWorktreeInstance 

func (s *SQLite) RegisterWorktreeInstance(ctx context.Context, principal contextstate.Principal, instance contextstate.WorktreeInstance, canonicalPath string) error

RegisterWorktreeInstance activates a preflighted catalog record. It adds the caller route in the same transaction.

func (*SQLite) ReleaseClaim 

func (s *SQLite) ReleaseClaim(ctx context.Context, id, h string) error

func (*SQLite) ReleaseClaimFenced 

func (s *SQLite) ReleaseClaimFenced(ctx context.Context, claim Claim) error

func (*SQLite) ReleaseLease added in v0.1.2 

func (s *SQLite) ReleaseLease(ctx context.Context, principal contextstate.Principal, sessionID string) error

ReleaseLease clears the caller's lease on a clean shutdown, so the next resume of this session id sees an immediately-stale (NULL) lease instead of waiting out sessionLeaseTTL against a process that already quit. Scoped by capability_digest exactly like RenewLease: a process whose capability was already reclaimed away matches zero rows and this is a no-op, not an error - it has nothing left to release.

func (*SQLite) RenewLease added in v0.1.2 

func (s *SQLite) RenewLease(ctx context.Context, principal contextstate.Principal, sessionID string) error

RenewLease refreshes the caller's context session lease so ReclaimSession treats this process's ownership as live. Scoped by capability_digest, not just subject: a process whose capability was already reclaimed away by a second process cannot resurrect its own stale lease and block the new owner - the UPDATE simply matches zero rows for a capability that no longer owns the row, which RenewLease treats as a no-op rather than an error, since the caller has no standing to be told about a takeover it lost with no reclaim call of its own.

func (*SQLite) RequireLegacyWorktreeRoute 

func (s *SQLite) RequireLegacyWorktreeRoute(ctx context.Context, principal contextstate.Principal, worktree, canonicalPath string) error

RequireLegacyWorktreeRoute verifies the exact unbound route needed for adoption.

func (*SQLite) SaveSession 

func (s *SQLite) SaveSession(ctx context.Context, principal contextstate.Principal, name string, messages []byte, model, provider string, turns, tokens, messageCount int, opts contextstate.SessionSaveOptions) error

func (*SQLite) SaveSessionAdmission 

func (s *SQLite) SaveSessionAdmission(ctx context.Context, principal contextstate.Principal, name string, record contextstate.SessionAdmission) error

SaveSessionAdmission persists a named session's admitted tool set. An empty name set deletes the row: resuming a session that admitted nothing must not resurrect an older set.

func (*SQLite) SaveWorktreeRoute 

func (s *SQLite) SaveWorktreeRoute(ctx context.Context, principal contextstate.Principal, worktree, dir string) error

SaveWorktreeRoute upserts the launch route for one mivia-managed worktree.

func (*SQLite) SaveWorktreeSessionAdmission 

func (s *SQLite) SaveWorktreeSessionAdmission(ctx context.Context, p contextstate.Principal, n string, r contextstate.SessionAdmission, i contextstate.WorktreeInstance) error

func (*SQLite) SetSessionTitle 

func (s *SQLite) SetSessionTitle(ctx context.Context, principal contextstate.Principal, sessionID, title string, instance contextstate.WorktreeInstance) error

SetSessionTitle updates display metadata for an authorized context session.

func (*SQLite) TableRowCounts 

func (s *SQLite) TableRowCounts(ctx context.Context) (map[string]int, error)

TableRowCounts reports the row count of every table WipeAllExceptSchema would empty - the dry-run half of a destructive reset. It never writes.

func (*SQLite) TakeoverClaim 

func (s *SQLite) TakeoverClaim(ctx context.Context, id, h string) error

func (*SQLite) TakeoverClaimFenced 

func (s *SQLite) TakeoverClaimFenced(ctx context.Context, id, h string) (Claim, error)

func (*SQLite) TakeoverExpiredClaim 

func (s *SQLite) TakeoverExpiredClaim(ctx context.Context, id, h string, maxAge time.Duration) error

func (*SQLite) TakeoverExpiredClaimFenced 

func (s *SQLite) TakeoverExpiredClaimFenced(ctx context.Context, id, h string, maxAge time.Duration) (Claim, error)

func (*SQLite) ValidateActiveWorktreeInstance 

func (s *SQLite) ValidateActiveWorktreeInstance(ctx context.Context, principal contextstate.Principal, instance contextstate.WorktreeInstance, canonicalPath string) error

ValidateActiveWorktreeInstance verifies the exact active catalog binding.

func (*SQLite) WipeAllExceptSchema 

func (s *SQLite) WipeAllExceptSchema(ctx context.Context) error

WipeAllExceptSchema deletes every row from every table this store owns, except resetPreservedTables, in one transaction. It exists for a full reset that keeps only the separate memory store (memory.db / org.db, internal/memory - a different file, never opened by this method) and the store's own migrated schema.

Several tables carry a foreign key onto context_sessions or context_payloads (context_checkpoints, context_payloads, context_source_events, context_audits, context_tombstones, context_imports - see context_schema_v1_v6.go), and that set has grown across migrations. Rather than hand-order every delete against a schema that keeps changing, this disables foreign_keys for the duration: SQLite only allows toggling that PRAGMA outside a transaction, so it is set on a single pinned connection before BEGIN and explicitly restored to ON after COMMIT, on that same connection, before the connection returns to the pool - a pooled connection some later caller reuses must never be the one still carrying foreign_keys=OFF.

type Store 

type Store interface {
	Append(context.Context, Event) error
	// AppendClaimed appends an event when its run is unclaimed or holder owns
	// the current claim. It returns ErrClaimHeld when another holder owns it.
	AppendClaimed(ctx context.Context, event Event, holder string) error
	// AppendAndDeleteRun atomically appends a deletion tombstone and removes
	// prior events and the claim for the same run. The supplied claim authorizes
	// the append when the run has an active claim.
	AppendAndDeleteRun(context.Context, Event, Claim) error
	Events(context.Context, string) ([]Event, error)
	// EventsSince returns the events of a run whose sequence is strictly
	// greater than afterSequence, ordered by ascending sequence. It is the
	// bounded tail read that lets a reader catch up on another writer's
	// appends without replaying the whole history.
	EventsSince(ctx context.Context, runID string, afterSequence int) ([]Event, error)
	// DeleteRun removes events at or below throughSequence and any claim for a
	// run. It never deletes content; a later tombstone event remains visible.
	DeleteRun(ctx context.Context, runID string, throughSequence int) error
	// Changes is the freshness probe for incremental catch-up. Given a cursor
	// previously returned by Changes (0 to start from the beginning), it
	// reports the highest sequence of every run appended to since that cursor,
	// together with the new cursor. Cost is proportional to the number of runs
	// that moved, not to the size of the history, so a caller that is already
	// up to date pays a constant-time probe.
	Changes(ctx context.Context, afterCursor uint64) (maxSequences map[string]int, cursor uint64, err error)
	// ClaimRun acquires an exclusive claim on a run for holder. Returns nil
	// if the claim was acquired. Returns ErrClaimHeld if another holder
	// already holds the claim. The same holder calling ClaimRun again
	// refreshes the claim successfully.
	ClaimRun(ctx context.Context, runID, holder string) error
	// TakeoverClaim atomically replaces any existing claim with holder.
	TakeoverClaim(ctx context.Context, runID, holder string) error
	// ReleaseClaim releases the claim on a run. Only the current holder may
	// release. Returns ErrClaimNotHeld if the caller does not hold the claim.
	ReleaseClaim(ctx context.Context, runID, holder string) error
	// ClearClaim force-releases any claim on a run, regardless of holder.
	// Returns nil if no claim existed. Used during crash recovery to clear
	// stale claims on terminal runs.
	ClearClaim(ctx context.Context, runID string) error
	// PutContent stores raw bytes keyed by a content-addressed reference
	// (e.g. "ref:output:xxxx"). Idempotent for the same ref.
	PutContent(ctx context.Context, ref string, data []byte) error
	// GetContent retrieves bytes previously stored by PutContent.
	// Returns ErrContentNotFound if the ref is unknown.
	GetContent(ctx context.Context, ref string) ([]byte, error)
	Count(context.Context) (int, error)
	ListRunIDs(context.Context) ([]string, error)
	Close() error
}

Source Files

View all Source files

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.