sessionstore

type BlobIntegrityError ¶

type BlobIntegrityError struct {
	Seq  uint64
	Key  string
	Want string // the sha256 the pointer named (expected)
	Got  string // sha256 of the bytes actually fetched
}

BlobIntegrityError reports an offloaded record whose fetched blob bytes do not hash to the sha256 its ledger pointer named: sha256(bytes) != pointer.SHA256, so the blob has been corrupted or substituted. It fails secure — replay surfaces it rather than decoding tampered bytes — and carries the record's ledger sequence, the blob key, and both the expected (pointer) and actual hashes.

func (e *BlobIntegrityError) Error() string

type BlobPointerIDMismatchError ¶

type BlobPointerIDMismatchError struct {
	Seq     uint64
	Key     string
	OuterID string
	InnerID string
}

BlobPointerIDMismatchError reports an offloaded record whose OUTER blobptr envelope's idempotency id does not match the id embedded in the RESOLVED inner envelope. The writer always stamps the exact same id on both halves of an offload (see sessionJournal.offload/frame — both the inline pre-offload envelope and its blobptr stand-in carry rec.IdempotencyID()), so a mismatch means the pointer and the blob it names have drifted apart. Replay fails closed rather than trusting either id blindly — this is the id-integrity counterpart to BlobIntegrityError's content hash check, and matters because a durable idempotency index is keyed by this id.

func (e *BlobPointerIDMismatchError) Error() string

type BlobUnavailableError ¶

type BlobUnavailableError struct {
	Seq   uint64
	Key   string
	Cause error
}

BlobUnavailableError reports that an offloaded record's backing blob could not be fetched: a dangling pointer (the blob is absent — Cause is a *storage.BlobNotFoundError) or any other Blobs.Get / read failure. It fails closed — replay surfaces it rather than yielding a zero-valued record — so a missing blob can never be mistaken for a drained backlog. It carries the record's ledger sequence and the blob key, and unwraps to the underlying cause.

func (e *BlobUnavailableError) Error() string

func (e *BlobUnavailableError) Unwrap() error

type Catalog ¶

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

Catalog maintains the derived session catalog in storage.KV: one SessionMeta per session, keyed by the session's ledger name. It has one reason to change: how the catalog is indexed. UpdateOnEvent folds a single event into the keyed entry (best-effort, post-append); ListSessions reads the KV only (no ledger cursor); RepairCatalog rebuilds an entry from the authoritative ledger.

func (c *Catalog) ListSessions(ctx context.Context) ([]SessionMeta, error)

func (c *Catalog) ReadMeta(ctx context.Context, id uuid.UUID) (SessionMeta, bool, error)

func (c *Catalog) RepairCatalog(ctx context.Context, sessionID uuid.UUID) (SessionMeta, error)

func (c *Catalog) UpdateOnEvent(ctx context.Context, ev event.Event, seq uint64) error

type CatalogClock ¶

type CatalogClock func() time.Time

CatalogClock is the time seam for the catalog: it stamps LastActiveAt at update time. Injecting it makes activity-bump assertions deterministic in tests.

type CatalogCompactionError ¶

type CatalogCompactionError struct {
	Kind      CatalogCompactionErrorKind
	AttemptID event.CompactAttemptID
}

CatalogCompactionError reports contradictory canonical compaction history encountered while rebuilding the derived catalog from the durable journal.

func (e *CatalogCompactionError) Error() string

type CatalogCompactionErrorKind ¶

type CatalogCompactionErrorKind string

const (
	CatalogCompactionDuplicateTerminal CatalogCompactionErrorKind = "duplicate_terminal"
)

type CatalogConflictError ¶

type CatalogConflictError struct {
	SessionID uuid.UUID
	Attempts  int
}

CatalogConflictError reports that a catalog update could not win the KV revision-CAS within catalogMaxCASRetries attempts: a persistently contended key. It has no storage analog in the NATS catalog (JetStream KV Put was unconditional last-write-wins); it exists because storage.KV is CAS-only. UpdateOnEvent logs+swallows it (best-effort); RepairCatalog surfaces it.

func (e *CatalogConflictError) Error() string

type CatalogDecodeError ¶

type CatalogDecodeError struct{ Cause error }

CatalogDecodeError identifies a malformed or semantically invalid catalog value. CatalogReadError wraps it with the affected session identity.

func (e *CatalogDecodeError) Error() string

func (e *CatalogDecodeError) Unwrap() error

type CatalogDuplicateFieldError ¶

type CatalogDuplicateFieldError struct {
	Path  string
	Field string
}

CatalogDuplicateFieldError reports duplicate JSON object members, including case aliases that encoding/json would otherwise accept with last-value wins.

func (e *CatalogDuplicateFieldError) Error() string

type CatalogEncodeError ¶

type CatalogEncodeError struct{ Cause error }

CatalogEncodeError wraps a failure to marshal a SessionMeta to JSON. A SessionMeta is value-typed, so this is effectively unreachable, but the codec returns a typed error rather than dropping the json.Marshal error to satisfy errors-are-typed.

func (e *CatalogEncodeError) Error() string

func (e *CatalogEncodeError) Unwrap() error

type CatalogHustleError ¶

type CatalogHustleError struct {
	Kind  CatalogHustleErrorKind
	RunID hustle.RunID
	Cause error
}

CatalogHustleError reports a malformed or overflowing privileged lifecycle fold. RunID identifies the offending durable run without exposing its input or output.

func (e *CatalogHustleError) Error() string

func (e *CatalogHustleError) Unwrap() error

type CatalogHustleErrorKind ¶

type CatalogHustleErrorKind string

const (
	CatalogHustleDuplicateStart       CatalogHustleErrorKind = "duplicate_start"
	CatalogHustleTerminalWithoutStart CatalogHustleErrorKind = "terminal_without_start"
	CatalogHustleAttributionMismatch  CatalogHustleErrorKind = "attribution_mismatch"
	CatalogHustleInvalidLifecycle     CatalogHustleErrorKind = "invalid_lifecycle"
	CatalogHustleRuntimeMismatch      CatalogHustleErrorKind = "runtime_mismatch"
	CatalogHustleUsageOverflow        CatalogHustleErrorKind = "usage_overflow"
	CatalogHustleRunCountOverflow     CatalogHustleErrorKind = "run_count_overflow"
)

type CatalogHustleMetaValidationError ¶

type CatalogHustleMetaValidationError struct {
	Index int
	Rule  CatalogMetaRule
	Cause error
}

func (e *CatalogHustleMetaValidationError) Error() string

func (e *CatalogHustleMetaValidationError) Unwrap() error

type CatalogLogger ¶

type CatalogLogger interface {
	// CatalogUpdateFailed is called with the typed error when a best-effort catalog update
	// could not read or write its KV entry. The implementation must not panic and must not
	// re-raise — it is the end of the error's life.
	CatalogUpdateFailed(err error)
}

CatalogLogger is the narrow logging seam the best-effort catalog update writes to when a KV read/write fails: the catalog is derivable, so a failure is logged and swallowed, NEVER surfaced to the append path. It is a single-method interface (Interface Segregation); a nop default keeps existing wiring unchanged.

type CatalogMetaField ¶

type CatalogMetaField string

CatalogMetaField identifies one semantic SessionMeta projection field.

const (
	CatalogMetaFieldLoopID          CatalogMetaField = "Loops.LoopID"
	CatalogMetaFieldLoopOrder       CatalogMetaField = "Loops"
	CatalogMetaFieldRuntime         CatalogMetaField = "Loops.Runtime"
	CatalogMetaFieldRuntimeSeq      CatalogMetaField = "Loops.RuntimeValueSeq"
	CatalogMetaFieldCumulativeUsage CatalogMetaField = "Loops.CumulativeUsage"
	CatalogMetaFieldCurrentContext  CatalogMetaField = "Loops.CurrentContext"
	CatalogMetaFieldContextSeq      CatalogMetaField = "Loops.ContextSeq"
	CatalogMetaFieldContextValueSeq CatalogMetaField = "Loops.ContextValueSeq"
)

type CatalogMetaRule ¶

type CatalogMetaRule string

CatalogMetaRule identifies a semantic catalog invariant.

const (
	CatalogMetaRuleRequired        CatalogMetaRule = "must be set"
	CatalogMetaRuleSortedUnique    CatalogMetaRule = "must be sorted and unique"
	CatalogMetaRuleInvalid         CatalogMetaRule = "is invalid"
	CatalogMetaRuleExceedsRuntime  CatalogMetaRule = "must not exceed RuntimeSeq"
	CatalogMetaRuleLegacyValue     CatalogMetaRule = "must be zero when Runtime is absent"
	CatalogMetaRuleExceedsContext  CatalogMetaRule = "must not exceed ContextSeq"
	CatalogMetaRuleNotAfterRuntime CatalogMetaRule = "must be newer than RuntimeSeq"
	CatalogMetaRuleContextAbsent   CatalogMetaRule = "must be zero when CurrentContext is absent"
	CatalogMetaRuleContextCurrent  CatalogMetaRule = "must equal ContextSeq when CurrentContext is set"
)

type CatalogMetaValidationError ¶

type CatalogMetaValidationError struct {
	LoopIndex int
	Field     CatalogMetaField
	Rule      CatalogMetaRule
	Cause     error
}

CatalogMetaValidationError reports an invalid bounded loop projection. The index makes corrupt records diagnosable without parsing an error string.

func (e *CatalogMetaValidationError) Error() string

func (e *CatalogMetaValidationError) Unwrap() error

type CatalogOption ¶

type CatalogOption func(*catalogOptions)

CatalogOption configures a Catalog at OpenCatalog time. Applied in order over a defaults struct, so a later option overrides an earlier one.

func WithCatalogClock(now CatalogClock) CatalogOption

func WithCatalogLogger(log CatalogLogger) CatalogOption

func WithCatalogReplayer(opener EventReplayerOpener) CatalogOption

type CatalogOrderingError ¶

type CatalogOrderingError struct {
	EventType string
	Sequence  uint64
	Last      uint64
}

CatalogOrderingError marks an online delivery whose sequence is behind the catalog cursor and whose additive effect therefore cannot be classified as a duplicate or a delayed unique record from bounded metadata alone. The online updater repairs from the authoritative journal instead of guessing.

func (e *CatalogOrderingError) Error() string

type CatalogReadError ¶

type CatalogReadError struct {
	SessionID uuid.UUID
	Cause     error
}

CatalogReadError wraps a failure to read or decode a catalog entry (a KV Get/Keys error that is not "not found", or a malformed stored SessionMeta). It carries (when known) the session and unwraps to the cause. ListSessions and RepairCatalog surface it; the best-effort UpdateOnEvent logs+swallows it (it must never fail the append).

func (e *CatalogReadError) Error() string

func (e *CatalogReadError) Unwrap() error

type CatalogUsageError ¶

type CatalogUsageError struct {
	LoopID uuid.UUID
	Cause  error
}

CatalogUsageError reports invalid or overflowing usage encountered while building the repairable catalog projection.

func (e *CatalogUsageError) Error() string

func (e *CatalogUsageError) Unwrap() error

type CatalogWriteError ¶

type CatalogWriteError struct {
	SessionID uuid.UUID
	Cause     error
}

CatalogWriteError wraps a failure to write a catalog entry (a KV Put/encode error). It carries the session and unwraps to the cause. The best-effort UpdateOnEvent logs+swallows it; RepairCatalog surfaces it (a repair the caller asked for that could not persist is a real failure).

func (e *CatalogWriteError) Error() string

func (e *CatalogWriteError) Unwrap() error

type CheckpointSummary ¶

type CheckpointSummary struct {
	Ref         workspacestore.Ref        `json:"ref"`
	EventID     uuid.UUID                 `json:"event_id"`
	Seq         uint64                    `json:"seq"`
	Consistency event.SnapshotConsistency `json:"consistency,omitempty"`
}

CheckpointSummary identifies the newest checkpoint independently from later rewinds.

type CommandApplicationNotFoundError ¶ added in v0.31.0

type CommandApplicationNotFoundError struct {
	SessionID uuid.UUID
	Seq       uint64
	Cause     error
}

CommandApplicationNotFoundError reports that the record at the named sequence is not an application prefix. It fails closed rather than reporting an absent correlation as "not applied", because that answer would let an already-applied command be applied a second time.

func (e *CommandApplicationNotFoundError) Error() string

func (e *CommandApplicationNotFoundError) Unwrap() error

type DurableBodyTooLargeError ¶ added in v0.31.0

type DurableBodyTooLargeError struct {
	Seq           uint64
	ObjectID      string
	DeclaredBytes uint64
	MaxBytes      int
}

DurableBodyTooLargeError reports an object-backed runtime body whose DECLARED size exceeds maxRuntimeBodyBytes, the ceiling replay can admit. It is a SIZE refusal, not an integrity finding: nothing is known to be corrupt or substituted, and the object may hash exactly as its reference names. It is deliberately NOT a *BlobIntegrityError, because a caller matching on that type would conclude tampering — and might raise a security response — for a faithfully written record. Replay fails closed on it (the body is never fetched, so an oversized declared size cannot drive a large read), and it carries the record's ledger sequence, the object id, the declared size, and the ceiling that refused it.

The write side refuses to create such a record (see sessionJournal.frame), so this is reachable only for a record written by some other producer or by an older writer predating that guard.

func (e *DurableBodyTooLargeError) Error() string

type EmptySessionError ¶

type EmptySessionError struct{ SessionID uuid.UUID }

EmptySessionError reports that RepairCatalog could not rebuild a session's entry because its ledger carries no SessionStarted (nothing to index). It carries the session and unwraps to errEmptyRepair.

func (e *EmptySessionError) Error() string

func (e *EmptySessionError) Unwrap() error

type EnvelopeError ¶

type EnvelopeError struct {
	Reason string
	Cause  error
}

EnvelopeError reports a failure to encode or decode a frame (or a blobPointer body): a malformed JSON payload, an unknown kind, or an unsupported version. Reason carries the human-readable context; Cause, when non-nil, is the underlying encoding/json error reachable via errors.As / errors.Unwrap. A semantic rejection (unknown kind or unsupported version) has no underlying cause and leaves Cause nil.

func (e *EnvelopeError) Error() string

func (e *EnvelopeError) Unwrap() error

type EventReplayerOpener ¶

type EventReplayerOpener interface {
	OpenInternalEventReplayer(id uuid.UUID, req ReplayRequest) (journal.EventReplayer, error)
}

EventReplayerOpener is the narrow seam RepairCatalog folds a session's ledger through: it opens a privileged read-side event replayer for one session. *Store satisfies it via OpenInternalEventReplayer (Dependency Inversion — the catalog depends on this method alone, not the whole Store). A nil opener disables repair (RepairCatalog fails with a typed error).

type GCDeleteError ¶

type GCDeleteError struct {
	Key   string
	Cause error
}

GCDeleteError reports a failure to delete one orphaned blob. GC surfaces it rather than silently swallowing the failure, so a caller learns the session's blobs could not be fully reclaimed. It carries the blob key and unwraps to the underlying cause.

func (e *GCDeleteError) Error() string

func (e *GCDeleteError) Unwrap() error

type GCLeaseNotHeldError ¶

type GCLeaseNotHeldError struct {
	SessionID uuid.UUID
	Epoch     uint64
}

GCLeaseNotHeldError reports that GC was refused because the session's single-writer lease is not held (released, or overtaken by a higher epoch). GC deletes blobs, so it must run only as the single writer; running unguarded could reap a blob a live owner is still offloading (its pointer append is mid-flight). It fails closed with this typed error and deletes nothing. It carries the session and the (stale) epoch the refused lease held, and unwraps to a *journal.LeaseLostError for errors.As — mirroring pkg/journal's ObjectGC.

func (e *GCLeaseNotHeldError) Error() string

func (e *GCLeaseNotHeldError) Unwrap() error

type GCListError ¶

type GCListError struct {
	Prefix string
	Cause  error
}

GCListError reports a failure to list the session's blob prefix. GC fails closed: without the blob inventory it cannot decide what to reap, so it deletes nothing. It carries the prefix and unwraps to the underlying cause.

func (e *GCListError) Error() string

func (e *GCListError) Unwrap() error

type GCResult ¶

type GCResult struct {
	// Scanned is the number of canonical legacy blobs selected under the session prefix.
	Scanned int
	// Referenced is the number of listed blobs still referenced by an in-ledger
	// pointer (kept).
	Referenced int
	// Deleted is the number of orphaned blobs reaped this pass; it always equals
	// len(DeletedKeys).
	Deleted int
	// DeletedKeys enumerates the reaped blob keys in lexicographic order (Blobs.List
	// returns sorted keys and the sweep preserves that order). It lets a caller log
	// exactly what was reclaimed without re-deriving it.
	DeletedKeys []string
	// Unreclaimable is the number of keys under the session's blob prefix this
	// pass deliberately did not consider: everything outside the legacy offload
	// shape, which today means released SessionStore objects. It mixes LIVE and
	// ORPHANED objects — this pass cannot tell them apart without SessionStore's
	// private layout — so it is a coverage measure, not an orphan count. Nonzero
	// means the pass did not cover the whole prefix and reclamation of that class
	// is still owed. See the README's "Object reclamation boundary" for the
	// released API this waits on and for the orphans Harness itself can create.
	Unreclaimable int
}

GCResult summarizes one legacy-object GC pass. On a fully successful pass Scanned == Referenced + Deleted — but read that identity together with Unreclaimable, without which it is vacuously true at 0 == 0 + 0 for a session whose blob prefix holds nothing BUT released objects.

This pass covers exactly one class: the legacy Harness offload shape. Released SessionStore v0.1.0 objects are outside it, because that version exposes no safe public enumeration or deletion API and Harness will not reconstruct its private physical layout. Unreclaimable counts what was therefore left untouched, so a caller logging a GCResult can distinguish "nothing to reclaim" (Scanned == 0, Unreclaimable == 0) from "I cannot see this class" (Unreclaimable > 0). A zero-valued GCResult returned alongside a non-nil error means the pass failed closed before it could measure anything.

type GCScanError ¶

type GCScanError struct {
	Name  string
	Cause error
}

GCScanError reports a failure to scan the session's ledger for the set of blob keys referenced by a live pointer: a ledger read/cursor failure, or an undecodable envelope or blob pointer. GC fails closed — without a COMPLETE live set it cannot safely decide which blobs are orphans, so it deletes nothing rather than risk reaping a still-referenced blob. It carries the ledger name and unwraps to the underlying cause.

func (e *GCScanError) Error() string

func (e *GCScanError) Unwrap() error

type HustleUsageAggregate ¶

type HustleUsageAggregate struct {
	Name            hustle.Name           `json:"name"`
	ModelSource     hustle.ModelSource    `json:"model_source"`
	NamedModelKey   model.ModelKey        `json:"named_model_key,omitzero"`
	Runtime         event.ModelRuntime    `json:"runtime,omitzero"`
	Status          hustle.TerminalStatus `json:"status"`
	Runs            uint64                `json:"runs"`
	CumulativeUsage content.Usage         `json:"cumulative_usage,omitzero"`
}

HustleUsageAggregate is one canonical terminal bucket. Current-loop work has a zero NamedModelKey so arbitrarily many resolved runtime keys cannot grow the catalog; named work uses the immutable key from its definition descriptor.

type InvalidBackendError ¶

type InvalidBackendError struct {
	Missing string
}

InvalidBackendError preserves Harness's legacy classification for a nil composite or one of its original four primitive fields. The released store subsequently validates the complete five-primitive backend and required capabilities before provider I/O.

func (e *InvalidBackendError) Error() string

type LoopUsageMeta ¶

type LoopUsageMeta struct {
	LoopID  uuid.UUID          `json:"loop_id"`
	Runtime event.ModelRuntime `json:"runtime,omitzero"`
	// RuntimeSeq is the latest lifecycle sequence observed for runtime selection.
	// A legacy event without Runtime advances this watermark while preserving the
	// last known value. One bounded scalar per loop prevents delayed lifecycle
	// notifications from regressing selection without an unbounded event set.
	RuntimeSeq uint64 `json:"runtime_seq,omitempty"`
	// RuntimeValueSeq is the sequence that supplied Runtime. It can trail
	// RuntimeSeq when a newer legacy event carries no resolved runtime, allowing
	// delayed known values to converge to the highest known sequence boundedly.
	RuntimeValueSeq uint64        `json:"runtime_value_seq,omitempty"`
	CumulativeUsage content.Usage `json:"cumulative_usage,omitzero"`
	// ContextSeq is the highest context-relevant lifecycle, mutation, or
	// measurement sequence observed for this loop. ContextValueSeq identifies the
	// event supplying CurrentContext. Invalidation preserves the former watermark
	// while clearing only the value sequence and measurement.
	ContextSeq      uint64                   `json:"context_seq,omitempty"`
	ContextValueSeq uint64                   `json:"context_value_seq,omitempty"`
	CurrentContext  event.ContextMeasurement `json:"current_context,omitzero"`
}

LoopUsageMeta is the catalog's bounded projection for one durable loop. CumulativeUsage folds authoritative StepDone request usage only; TurnDone's convenience projection is deliberately excluded.

type NilLeaseError ¶

type NilLeaseError struct {
	SessionID uuid.UUID
}

NilLeaseError reports that a Store constructor (OpenJournal or OpenObjectGC) was handed a nil lease. The lease is a required dependency (DIP): the composition root acquires it via AcquireLease and passes it in. The constructor fails closed with this typed error rather than deferring a nil dereference to first use (stamping the epoch into the opening fence, or the GC lease guard).

func (e *NilLeaseError) Error() string

type ObjectGC ¶

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

ObjectGC reaps orphaned legacy Harness offload blobs from one session's content-addressed blob prefix. It deliberately retains released SessionStore objects until that module publishes a safe retention/reaping API, and reports how many keys that left untouched in GCResult.Unreclaimable so a pass over a prefix it cannot fully see never reads as a clean one.

It is lease-guarded: it deletes, so it runs only while holding a valid single-writer lease and is therefore the single deleter. That lease guard is also the whole of its concurrency safety. GC MUST NOT run concurrently with active appends/offloads to the same session: a blob whose pointer append is still in flight would be observed as unreferenced by the scan and wrongly reaped. The caller serializes GC with the writer — typically running it while holding the session lease (as the single writer) or when the session is idle. Unlike pkg/journal's ObjectGC there is no grace window backstop: storage's Blobs.List surfaces no ModTime, so an in-flight upload cannot be protected by age; the serialization is load-bearing, not advisory.

It is the GC analogue of the sessionstore journal (write) and replayers (read), wired at the composition root via Store.OpenObjectGC.

func (g *ObjectGC) GC(ctx context.Context) (GCResult, error)

type OpeningFenceConflictError ¶ added in v0.31.0

type OpeningFenceConflictError struct {
	SessionID uuid.UUID
	Epoch     uint64 // the spent grant's epoch, already released
	Cause     error  // the underlying *journal.AppendError
}

OpeningFenceConflictError reports that a journal's OPENING fence lost the ownership race: the ledger tip moved between this Open's tip read and its fence CAS, so some other writer legitimately owns the stream now. It is deliberately NOT a transient failure. A lease grant is never rebased: this Open does not refresh the tip and try again under the same epoch, because a fence planted at a refreshed tip can land AFTER a higher-epoch owner's fence, and every record the rebased writer then appends reads, to any later replayer, as though it belonged to that higher epoch — the ledger's epoch high-water would fall and a fenced-out writer's state would be folded in as the current owner's. The grant that hit this error is spent and has been released; the caller must acquire a FRESH lease (which yields a strictly higher epoch) and construct a new writer.

func (e *OpeningFenceConflictError) Error() string

func (e *OpeningFenceConflictError) Unwrap() error

type Option ¶

type Option func(*Options)

Option overrides a single field of Options at Open time. Options are applied in order over the defaults, so a later Option wins over an earlier one.

func WithOffloadThreshold(n int) Option

func WithTenant(tenant coresessionwire.TenantID) Option

type Options ¶

type Options struct {
	// OffloadThreshold is the payload size (bytes) above which a record is
	// stored as an out-of-line blob instead of inline in the ledger.
	OffloadThreshold int

	// TenantID is the tenant every record this Store files belongs to. It is
	// the identity a counterparty reading the same backend addresses the
	// session by, so it is an INPUT rather than a constant: see WithTenant.
	TenantID coresessionwire.TenantID
}

Options are the resolved knobs a Store operates under. It is populated by Open from the defaults plus any Option overrides; callers never construct it directly.

type PersistencePathError ¶

type PersistencePathError struct {
	Path  string
	Cause error
}

PersistencePathError reports a local persistence path that could not be canonicalized without ambiguity.

func (e *PersistencePathError) Error() string

func (e *PersistencePathError) Unwrap() error

type ReplayDecodeError ¶

type ReplayDecodeError struct {
	Seq   uint64
	Cause error
}

ReplayDecodeError reports a failure to decode a replayed ledger record into its typed form: an undecodable envelope, an undecodable blob pointer, an unexpected (post-resolution) envelope kind, or a codec unmarshal failure on the record's body. It fails secure — replay surfaces it rather than skipping or zero-valuing the record — and carries the offending record's ledger sequence and the underlying cause (a *EnvelopeError, an event/command codec error, etc.).

func (e *ReplayDecodeError) Error() string

func (e *ReplayDecodeError) Unwrap() error

type ReplayReadError ¶

type ReplayReadError struct {
	Name  string
	Cause error
}

ReplayReadError reports a failure to read the next record from the ledger cursor (a backend Ledger.Read or Cursor.Next failure). It fails closed: replay surfaces it rather than guessing the backlog is drained. It carries the ledger name and unwraps to the underlying cause.

func (e *ReplayReadError) Error() string

func (e *ReplayReadError) Unwrap() error

type ReplayRequest ¶

type ReplayRequest struct {
	// FromSeq is the inclusive ledger sequence to begin at. Storekit sequences are
	// 1-based and Ledger.Read(from) yields the record at Seq==from first; 0 (and 1)
	// both begin at the first record.
	FromSeq uint64
}

ReplayRequest positions a sessionstore replay. It carries an exported inclusive start sequence because journal.ReplayRequest hides its start behind a package-private journal.StartPos that an out-of-package replayer cannot read: the storage replayer's positioning must therefore flow through this request, set when the replayer is opened. Subject/loop narrowing is not part of storage replay — a session is one ledger, walked whole and filtered by envelope kind — so this request needs only the start position.

type RuntimeCommandLog ¶ added in v0.31.0

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

RuntimeCommandLog is the session-bound seam an applier uses to make an application prefix durable and to read one back. It pairs the strict append (through the session's own lease-fenced journal) with the privileged read (through the store), because the duplicate-delivery rule needs both: the append detects the duplicate, and the read says what the ORIGINAL delivery mapped to.

func (l *RuntimeCommandLog) AppendCommandApplication(ctx context.Context, app runtimecommand.Application) (journal.AppendResult, error)

func (l *RuntimeCommandLog) ReadCommandApplicationAt(ctx context.Context, seq uint64) (runtimecommand.Application, error)

type SessionMeta ¶

type SessionMeta struct {
	// SessionID is the session this entry describes.
	SessionID uuid.UUID `json:"session_id"`
	// Title is a short, human-readable label derived from the MOST RECENT turn's user
	// message (its first line, truncated), so a picker shows what a session is doing now
	// rather than its stale opening line. It updates on every TurnStarted that carries
	// derivable text, retaining the prior value when a turn's message has none (e.g. a
	// tool-continuation with no user-authored text). Empty until a first TurnStarted with
	// derivable text is seen.
	Title string `json:"title,omitempty"`
	// CreatedAt is when the session started (SessionStarted's CreatedAt).
	CreatedAt time.Time `json:"created_at,omitzero"`
	// LastActiveAt is the most recent activity instant (bumped by TurnStarted, StepDone,
	// RestoreDone), stamped from the catalog's injected clock at update time.
	LastActiveAt time.Time `json:"last_active_at,omitzero"`
	// Status is the session's lifecycle phase (active until SessionStopped -> stopped).
	Status SessionStatus `json:"status,omitempty"`
	// AgentKind names the agent role (from SessionStarted's ConfigFingerprint). It is
	// passthrough: empty until the agent threads its kind through loop.Definition.
	AgentKind string `json:"agent_kind,omitempty"`
	// LoopCount is the number of loops registered in the session: the primary plus one
	// per LoopStarted.
	LoopCount int `json:"loop_count,omitempty"`
	// ConfigFingerprint is the config identity the session started under, for the picker
	// to surface a config change on restore.
	ConfigFingerprint event.ConfigFingerprint `json:"config_fingerprint,omitzero"`
	// Residency records whether a process holds this session's runtime. Empty until
	// the fold sees its first residency-bearing event, so an entry written before the
	// field existed reads as the empty residency rather than claiming either value.
	Residency SessionResidency `json:"residency,omitempty"`
	// State is the status-fold lifecycle state (running/waiting_on_gate/idle/failed/
	// interrupted/stopped). It supersedes Status for richer callers; Status is retained
	// for back-compat. Empty until the fold sees its first state-bearing event.
	State SessionState `json:"state,omitempty"`
	// LastJournalSeq is the highest journal sequence folded into this entry (a monotonic
	// max over the events the projection has consumed): a status reader's resume cursor.
	LastJournalSeq uint64 `json:"last_journal_seq,omitempty"`
	// ActiveTurnID is the turn currently running (set by TurnStarted, cleared by TurnDone).
	// Zero when no turn is active.
	ActiveTurnID uuid.UUID `json:"active_turn_id,omitzero"`
	// WaitingGateID is the open gate blocking progress (set by GateOpened, cleared by
	// GateResolved). Zero when no gate is open.
	WaitingGateID uuid.UUID `json:"waiting_gate_id,omitzero"`
	// LastTurn is the codec-safe summary of the most recent terminal turn event
	// (TurnDone/TurnFailed). Nil until a turn ends.
	LastTurn *eventSummary `json:"last_turn,omitempty"`
	// LastStep is the codec-safe summary of the most recent StepDone. Nil until a step
	// completes.
	LastStep         *eventSummary     `json:"last_step,omitempty"`
	LastCheckpoint   CheckpointSummary `json:"last_checkpoint,omitzero"`
	CurrentWorkspace WorkspacePointer  `json:"current_workspace,omitzero"`
	// Loops is the deterministic, per-loop usage/runtime projection. It is sorted
	// by LoopID bytes and rebuilt from lifecycle + StepDone events.
	Loops []LoopUsageMeta `json:"loops,omitempty"`
	// Hustles is a bounded terminal-only aggregate. Detailed runs and unmatched
	// starts remain exclusively in the privileged journal.
	Hustles []HustleUsageAggregate `json:"hustles,omitempty"`
}

SessionMeta is the derived per-session catalog entry: the small, replay-free record the session picker reads to list sessions without opening a single ledger cursor. It is JSON (snake_case) stored one-per-session in storage.KV, keyed by the session's ledger name ("sessions/<uuid>"). It is a cache rebuilt from the authoritative ledger when missing or stale (RepairCatalog) — never the source of truth.

type SessionResidency ¶ added in v0.31.0

type SessionResidency string

SessionResidency is the catalog's record of whether some process currently holds this session's runtime. It is an axis INDEPENDENT of SessionStatus/SessionState: an idle session may be resident or cold, and a cold session may be either restorable or terminal. Reading residency as terminality is the exact confusion SessionResidencyReleased exists to prevent.

const (
	// ResidencyResident marks a session some process holds the runtime for (set by
	// SessionStarted and by RestoreDone).
	ResidencyResident SessionResidency = "resident"
	// ResidencyCold marks a session no process is resident for. BOTH a nonterminal
	// SessionResidencyReleased and a terminal SessionStopped set it, so residency
	// alone never distinguishes a released session from a stopped one — Status/State
	// carry that.
	ResidencyCold SessionResidency = "cold"
)

type SessionState ¶

type SessionState string

SessionState is the richer, status-fold lifecycle state the catalog projects from the event stream. It is a closed typed enum a status reader (the serve session API) switches on. It SUPERSEDES SessionStatus for callers that need the running/waiting/idle/terminal distinction, but Status is kept for back-compat (see SessionMeta.Status): State is additive, so an old entry decoded without it simply reads as the empty state and is rebuildable via RepairCatalog.

const (
	// StateRunning: a turn is actively executing (set by TurnStarted, restored after a
	// gate resolves while a turn is active).
	StateRunning SessionState = "running"
	// StateWaitingOnGate: a gate is open and blocking progress (set by GateOpened).
	StateWaitingOnGate SessionState = "waiting_on_gate"
	// StateIdle: the session is up but no turn is running (set by SessionStarted, TurnDone,
	// or a gate resolving with no active turn).
	StateIdle SessionState = "idle"
	// StateFailed: the last turn ended in a non-cancellation failure (set by TurnFailed).
	StateFailed SessionState = "failed"
	// StateInterrupted: the last turn was interrupted/cancelled (set by TurnInterrupted).
	StateInterrupted SessionState = "interrupted"
	// StateStopped: the session emitted SessionStopped — the terminal state that wins over
	// every other (set by SessionStopped).
	StateStopped SessionState = "stopped"
)

type SessionStatus ¶

type SessionStatus string

SessionStatus is the lifecycle phase the catalog records for a session. It is a closed typed enum (not a free-form string) so a picker can switch on it and a typo cannot silently mislabel a session.

const (
	// StatusActive marks a session whose primary loop is running (the SessionStarted
	// default until a SessionStopped flips it).
	StatusActive SessionStatus = "active"
	// StatusStopped marks a session that emitted SessionStopped (a clean shutdown). It
	// survives on disk and is brought back by restore — Stopped is a phase, not a delete.
	StatusStopped SessionStatus = "stopped"
)

type Store ¶

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

Store is the session-scoped facade over a storage backend. It holds the assembled *storage.Composite (whose five primitives it addresses by field; their methods overlap, so there is no flattened backend interface) plus the resolved Options. Construct it only via Open.

func Open(b *storage.Composite, opts ...Option) (*Store, error)

func (s *Store) AcquireLease(ctx context.Context, id uuid.UUID) (journal.Lease, error)

func (s *Store) OpenCatalog(opts ...CatalogOption) *Catalog

func (s *Store) OpenEventReplayer(id uuid.UUID, req ReplayRequest) (journal.EventReplayer, error)

func (s *Store) OpenInternalEventReplayer(id uuid.UUID, req ReplayRequest) (journal.EventReplayer, error)

func (s *Store) OpenInternalRecordReplayer(id uuid.UUID, req ReplayRequest) (journal.RecordReplayer, error)

func (s *Store) OpenJournal(ctx context.Context, id uuid.UUID, lease journal.Lease) (journal.SessionJournal, error)

func (s *Store) OpenJournalWithOpeningAppend(
	ctx context.Context,
	id uuid.UUID,
	lease journal.Lease,
	middleware journal.AppendMiddleware,
) (journal.SessionJournal, error)

func (s *Store) OpenObjectGC(id uuid.UUID, lease journal.Lease) (*ObjectGC, error)

func (s *Store) OpenRuntimeCommandLog(id uuid.UUID, j journal.SessionJournal) (*RuntimeCommandLog, error)

func (s *Store) PersistencePaths() ([]string, error)

func (s *Store) ReadCommandApplicationAt(ctx context.Context, id uuid.UUID, seq uint64) (runtimecommand.Application, error)

func (s *Store) WorkspaceCheckpointBySeq(ctx context.Context, id uuid.UUID, seq uint64) (CheckpointSummary, bool, error)

func (s *Store) WorkspaceCheckpointByTurn(ctx context.Context, id, turnID uuid.UUID) (CheckpointSummary, bool, error)

func (s *Store) WorkspaceLiveRefs(ctx context.Context, retainedSessionIDs []uuid.UUID) (map[workspacestore.Ref]struct{}, error)

type WorkspaceJournalScanError ¶

type WorkspaceJournalScanError struct {
	SessionID uuid.UUID
	Cause     error
}

WorkspaceJournalScanError reports that a retained session journal could not be scanned completely for workspace references. Manual workspace GC must fail closed on this error: an incomplete live set is unsafe to collect against.

func (e *WorkspaceJournalScanError) Error() string

func (e *WorkspaceJournalScanError) Unwrap() error

type WorkspacePointer ¶

type WorkspacePointer struct {
	Ref     workspacestore.Ref     `json:"ref"`
	EventID uuid.UUID              `json:"event_id"`
	Seq     uint64                 `json:"seq"`
	Source  WorkspacePointerSource `json:"source,omitempty"`
}

WorkspacePointer identifies one durable workspace transition. Ref is content identity; Seq and EventID are the journal transition identity.

type WorkspacePointerSource ¶

type WorkspacePointerSource string

WorkspacePointerSource identifies the transition that selected CurrentWorkspace. Unknown decodes catalog records written before this discriminator existed.

const (
	WorkspacePointerSourceUnknown    WorkspacePointerSource = ""
	WorkspacePointerSourceCheckpoint WorkspacePointerSource = "checkpoint"
	WorkspacePointerSourceRestore    WorkspacePointerSource = "restore"
)