rollups

type Batch ¶

type Batch struct {
	// Checkpoint is the sequence of the last event the batch covers (the
	// existing local durable sequence — the bus Sequence). Must be
	// strictly greater than the stored checkpoint or the batch is a
	// no-op.
	Checkpoint uint64
	// Deltas are the row updates derived from the batch's events.
	Deltas []Delta
}

Batch is one atomic write unit: the row deltas derived from a contiguous run of consumed events plus the checkpoint they advance to. The Store applies the deltas AND moves the checkpoint to Batch.Checkpoint in ONE atomic step, which is what makes replay idempotent: a crash between applying deltas and checkpointing is impossible, and re-applying a batch whose checkpoint does not advance the stored checkpoint is a no-op.

type BucketSize ¶

type BucketSize string

BucketSize is the closed set of fixed UTC bucket sizes. Every bucket grid is anchored to UTC: minute buckets start at MM:00Z, hour buckets at HH:00:00Z, day buckets at 00:00:00Z — never at a local-time or DST-adjusted boundary. A bucket boundary is a pure function of (instant, BucketSize), so two runs at two different instants that fall in the same bucket compute IDENTICAL boundaries, including across a runtime restart.

const (
	// BucketMinute is the UTC minute grid: buckets [MM:00Z, MM+1:00Z).
	// This is the STORAGE granularity — every stored row is keyed on the
	// minute grid (see StoreGranularity in extract.go).
	BucketMinute BucketSize = "minute"
	// BucketHour is the UTC hour grid: buckets [HH:00:00Z, HH+1:00:00Z).
	BucketHour BucketSize = "hour"
	// BucketDay is the UTC day grid: buckets [00:00:00Z, next 00:00:00Z).
	BucketDay BucketSize = "day"
)

func (b BucketSize) Duration() time.Duration

func (b BucketSize) Validate() error

type Clock ¶

type Clock interface {
	Now() time.Time
}

Clock abstracts the projector's "now" so tests do not depend on the wall clock. Production passes nil and the real UTC clock is used.

type Delta ¶

type Delta struct {
	Key Key
	Add MeasureSet
}

Delta is one row update derived from a single canonical event: the bucket row to touch and the exact integer measures to accumulate.

func Extract(ev events.Event) ([]Delta, error)

type Dimension ¶

type Dimension string

Dimension is a closed rollup dimension. The set is EXACTLY tenant / user / session / model — no other axis exists in this release. All values are AUTHORITATIVE: the tenant / user / session axes come from the event's identity triple (the isolation principal — never from payload fields a producer could choose), and model comes from the source event's model field. Agent is NOT a rollup dimension: none of the V1 canonical payloads carry an authoritative agent id, and an empty agent axis would fabricate a dimension with no source. A group_by of "agent" is rejected loudly. agent_id is NOT an isolation principal — it is not a scoping key at all in this domain (CLAUDE.md §6).

const (
	// DimensionTenant is the event's tenant_id.
	DimensionTenant Dimension = "tenant"
	// DimensionUser is the event's user_id.
	DimensionUser Dimension = "user"
	// DimensionSession is the event's session_id.
	DimensionSession Dimension = "session"
	// DimensionModel is the source payload's model (LLM completions only;
	// empty for events with no authoritative model).
	DimensionModel Dimension = "model"
)

func (d Dimension) Validate() error

type DimensionValues ¶

type DimensionValues map[Dimension]string

DimensionValues maps a dimension to its value for a grouped result row. The map carries exactly the query's GroupBy dimensions; a missing entry means the empty value (a grouped row that aggregates un-attributed events, e.g. model "" for task outcomes).

func (v DimensionValues) Less(w DimensionValues) bool

type Filter ¶

type Filter struct {
	// TenantIDs restricts to rows of these tenants ("" matches all).
	TenantIDs []string
	// UserIDs restricts to rows of these users ("" matches all).
	UserIDs []string
	// SessionIDs restricts to rows of these sessions ("" matches all).
	SessionIDs []string
	// Models restricts to rows with these model values. An empty Models
	// slice matches BOTH un-attributed (model "") and attributed rows;
	// to see only model-attributed rows, name the models explicitly.
	Models []string
}

Filter constrains a query over the closed dimensions. Each slice has set semantics: an empty slice matches every value on that axis; a non-empty slice matches exactly the listed values. All axes are ANDed.

func (f Filter) Matches(k Key) bool

type Key ¶

type Key struct {
	// BucketStart is the start instant of the fixed UTC bucket the row
	// belongs to (UTC, on the BucketMinute grid).
	BucketStart time.Time
	// TenantID is the event's tenant_id (authoritative — the isolation
	// principal).
	TenantID string
	// UserID is the event's user_id.
	UserID string
	// SessionID is the event's session_id.
	SessionID string
	// Model is the source payload's model; empty when the event carried
	// no authoritative model (e.g. task outcome events).
	Model string
}

Key identifies one rollup row: a fixed UTC bucket start (on the minute grid — the storage granularity) plus the authoritative dimension values. Model is the empty string when the source event carried no authoritative value for that dimension — the row is then the un-attributed aggregate for the triple it shares. A Key is comparable, so it can be a map key directly.

func (k Key) DimensionValue(d Dimension) string

type Measure ¶

type Measure string

Measure is a closed, additive rollup measure. Every measure is sourced ONLY from existing canonical event payloads — there are no derived, estimated, or sampled values. Measures that have no canonical source are ABSENT from the set: they cannot be requested (the query fails loud) and no row ever carries them. In particular, attempts, failed LLM calls, and user-message counts have no canonical payload backing and are absent.

All measures accumulate in exact integer form (int64): counts and token counts are plain integers, cost is integer micro-units of USD (see CostScaleMicros), latency is integer milliseconds. Nothing is normalised to float64 anywhere in accumulation, storage, or query — see MeasureValue.

Most measures are sums; the latency min/max measures are folds (the per-group minimum / maximum of the per-event latencies), merged by MeasureSet.Add — which range-checks every additive field and fails loudly (ErrMeasureOverflow) rather than letting a sum wrap.

const (
	// MeasureLLMCompletions is the count of successfully-recorded LLM
	// completions (`llm.cost.recorded` events). "Successful" here means the
	// provider returned a completion AND the runtime emitted the cost
	// record for it — the only successful-completion signal the canonical
	// payloads carry.
	MeasureLLMCompletions Measure = "llm_completions"
	// MeasureLLMTokensPrompt is the sum of Usage.PromptTokens.
	MeasureLLMTokensPrompt Measure = "llm_tokens_prompt" //nolint:gosec // G101 false positive: closed rollup measure-name wire constant, not a credential
	// MeasureLLMTokensCompletion is the sum of Usage.CompletionTokens.
	MeasureLLMTokensCompletion Measure = "llm_tokens_completion" //nolint:gosec // G101 false positive: closed rollup measure-name wire constant, not a credential
	// MeasureLLMTokensReasoning is the sum of Usage.ReasoningTokens.
	MeasureLLMTokensReasoning Measure = "llm_tokens_reasoning" //nolint:gosec // G101 false positive: closed rollup measure-name wire constant, not a credential
	// MeasureLLMTokensCacheRead is the sum of Usage.CacheReadTokens.
	MeasureLLMTokensCacheRead Measure = "llm_tokens_cache_read" //nolint:gosec // G101 false positive: closed rollup measure-name wire constant, not a credential
	// MeasureLLMTokensCacheWrite is the sum of Usage.CacheWriteTokens.
	MeasureLLMTokensCacheWrite Measure = "llm_tokens_cache_write" //nolint:gosec // G101 false positive: closed rollup measure-name wire constant, not a credential
	// MeasureLLMTokensTotal is the sum of Usage.TotalTokens.
	MeasureLLMTokensTotal Measure = "llm_tokens_total" //nolint:gosec // G101 false positive: closed rollup measure-name wire constant, not a credential
	// MeasureLLMCostMicros is the sum of provider-reported TotalCost in
	// exact integer micro-units of USD (USD * CostScaleMicros), across
	// successful LLM completions. The source float is converted once per
	// canonical event in Extract; no float is accumulated or exposed.
	MeasureLLMCostMicros Measure = "llm_cost_micros"
	// MeasureLLMLatencyCount is the count of latency-bearing completions
	// (the same population as MeasureLLMCompletions).
	MeasureLLMLatencyCount Measure = "llm_latency_count"
	// MeasureLLMLatencySumMS is the sum of Usage.LatencyMS. Average latency
	// for a group is SumMS / LatencyCount (exact integer arithmetic when
	// the quotient is exact).
	MeasureLLMLatencySumMS Measure = "llm_latency_sum_ms"
	// MeasureLLMLatencyMinMS is the minimum Usage.LatencyMS in the group.
	// Defined exactly when MeasureLLMLatencyCount > 0.
	MeasureLLMLatencyMinMS Measure = "llm_latency_min_ms"
	// MeasureLLMLatencyMaxMS is the maximum Usage.LatencyMS in the group.
	// Defined exactly when MeasureLLMLatencyCount > 0.
	MeasureLLMLatencyMaxMS Measure = "llm_latency_max_ms"
	// MeasureTasksCompleted is the count of `task.completed` events.
	MeasureTasksCompleted Measure = "tasks_completed"
	// MeasureTasksFailed is the count of `task.failed` events.
	MeasureTasksFailed Measure = "tasks_failed"
	// MeasureTasksCancelled is the count of `task.cancelled` events.
	MeasureTasksCancelled Measure = "tasks_cancelled"
)

func (m Measure) Validate() error

type MeasureSet ¶

type MeasureSet struct {
	LLMCompletions      int64
	LLMTokensPrompt     int64
	LLMTokensCompletion int64
	LLMTokensReasoning  int64
	LLMTokensCacheRead  int64
	LLMTokensCacheWrite int64
	LLMTokensTotal      int64
	LLMCostMicros       int64
	LLMLatencyCount     int64
	LLMLatencySumMS     int64
	LLMLatencyMinMS     int64 // fold-min; defined when LLMLatencyCount > 0
	LLMLatencyMaxMS     int64 // fold-max; defined when LLMLatencyCount > 0
	TasksCompleted      int64
	TasksFailed         int64
	TasksCancelled      int64
	// contains filtered or unexported fields
}

MeasureSet is the fixed-width set of all measures for one bucket row. Each field is an exact integer accumulation of the source payload values: int64 for tokens, latency, counts, and cost micro-units — there is NO float64 field. A zero MeasureSet is the additive identity.

Latency min/max are folds, not sums: Add merges them by taking the group-wise minimum / maximum. LLMLatencyMinMS / LLMLatencyMaxMS are defined exactly when LLMLatencyCount > 0 (the hasLatency flag tracks whether any latency-bearing record has been folded in).

Accumulation is CHECKED: every additive field is verified against the domain's nonnegative + exact int64 bounds before any write — a negative delta is refused with ErrNegativeMeasure (a counter never shrinks) and a sum that would overflow fails loudly with ErrMeasureOverflow instead of wrapping into negative or corrupt data. The receiver is never partially mutated by a refused merge.

func (m *MeasureSet) Add(other MeasureSet) error

func (m MeasureSet) Get(measure Measure) MeasureValue

func (m MeasureSet) IsZero() bool

type MeasureValue ¶

type MeasureValue struct {
	// N is the exact accumulated integer.
	N int64
	// Scale is the decimal denominator of the measure's unit; constant
	// per measure.
	Scale uint32
}

MeasureValue is the exact, wire-ready value of one measure for one row. Every measure accumulates in integer form only (see measure.go): counts, tokens, latency ms, and cost micro-units. N is the exact accumulated integer — counters above 2^53 stay exact because nothing is ever normalised to float64. Scale is the measure's fixed decimal denominator (1 for integer measures; CostScaleMicros for cost), so a consumer formats decimal USD exactly as N / Scale at the edge. A MeasureValue is JSON-safe and comparable on N for a fixed measure.

type PageCursor ¶

type PageCursor struct {
	// ShapeVersion is CursorShapeVersion of the producing query. A cursor
	// with a different version is rejected with ErrBadCursor.
	ShapeVersion int
	// Fingerprint is the deterministic fingerprint of the producing
	// query's canonical shape (see QueryShapeFingerprint). A query whose
	// fingerprint differs is rejected with ErrBadCursor before paging.
	Fingerprint string
	// BucketNano is the row's bucket start in unix nanoseconds (exact
	// int64 — never float-converted).
	BucketNano int64
	// MeasureVal is the row's SortMeasure sum in its exact integer form
	// (the MeasureValue.N — same scale as SortMeasure, so comparison is
	// exact; never float-converted).
	MeasureVal int64
	// Group is the row's grouped dimension values.
	Group DimensionValues
}

PageCursor is the opaque deterministic pagination position: the primary sort value plus the exact bucket start and grouped dimension values of the last row of the page. The next page starts strictly after it. Encode via EncodeCursor; the encoded string is what a caller passes back as Query.Cursor.

A cursor is BOUND to the canonical shape of the query that produced it: the shape version and fingerprint (see QueryShapeFingerprint) ride along, and any query whose shape differs is rejected with ErrBadCursor — a cursor is never silently re-purposed across a different window, bucket, filter, grouping, measure set, or sort.

func DecodeCursor(s string) (PageCursor, error)

type Projector ¶

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

Projector consumes successfully-persisted canonical events from a Source and applies their measure deltas to a Store, checkpointing the existing local durable sequence (the bus Sequence) after every batch.

The projector is a compiled artifact: source + store + options are set at construction and never mutated. Per-call state (the watermark, the catch-up state) is guarded by an internal mutex and is safe to read (Quality) at any time. The state-MUTATING path is serialised by an advance mutex held for the FULL Advance/CatchUp step (watermark read, source read, fence checks, ApplyBatch, and the in-memory watermark/state update) and by Rebuild, so a delayed Advance can never apply a newer/no-op batch and then overwrite p.watermark/state backwards, and a Rebuild can never interleave with an in-flight Advance.

Best-effort posture: the projector is a DOWNSTREAM consumer of an already-successful publication. The durable log persisted the event and fanned it out BEFORE the projector reads it; the projector's failures (StateUnavailable, retried on the next Advance) therefore never fail the canonical event publication path. No caller should use projector quality to fail an already-persisted event. There is no outbox, no new event id, and no exactly-once claim: replay is idempotent through the atomic checkpoint, and the projector is the single writer to its Store.

func NewProjector(source Source, store Store, opts ...ProjectorOption) (*Projector, error)

func (p *Projector) Advance(ctx context.Context) (bool, error)

func (p *Projector) CatchUp(ctx context.Context) error

func (p *Projector) Quality(ctx context.Context) (Quality, error)

func (p *Projector) Rebuild(ctx context.Context) error

type ProjectorOption ¶

type ProjectorOption func(*Projector)

ProjectorOption configures the projector at construction. The options are test/operator seams; production wiring uses the defaults.

func WithProjectorBatchSize(n int) ProjectorOption

func WithProjectorClock(c Clock) ProjectorOption

type Quality ¶

type Quality struct {
	// State is current / catching_up / unavailable.
	State State
	// Watermark is the last successfully applied sequence (the existing
	// local durable sequence — read from the Store's checkpoint, so it is
	// the durable truth across restarts).
	Watermark uint64
	// WatermarkAt is the wall-clock instant the watermark last advanced in
	// THIS projector instance (zero before the first advance — e.g. right
	// after a restart, before catch-up ran).
	WatermarkAt time.Time
	// RetentionStart is the oldest retained bucket start (zero when the
	// store holds no rows).
	RetentionStart time.Time
	// RetentionEnd is the newest retained bucket start (zero when the
	// store holds no rows).
	RetentionEnd time.Time
	// Err is the latest ingestion failure, present only when State is
	// StateUnavailable.
	Err error
}

Quality is the projector's operational snapshot: catch-up state, the watermark (the last applied local durable sequence), and the retention horizon of the rows the store holds. It is a READ-ONLY view — it never mutates the projector or the store.

type Query ¶

type Query struct {
	// From / To bound the bucket window (half-open [From, To), both
	// normalised to UTC AND aligned to the Bucket grid: each must equal
	// its own BucketStart, so a window never includes a partial bucket —
	// an unaligned edge is rejected with ErrQueryInvalid). Mandatory:
	// From must precede To, and the window may span at most MaxBuckets at
	// the requested Bucket size.
	From time.Time
	To   time.Time
	// Bucket is the (closed) query bucket size. Rows are stored at
	// StoreGranularity (the minute grid) and coarsened to Bucket at read
	// time.
	Bucket BucketSize
	// GroupBy is the closed dimension subset the rows are grouped by (may
	// be empty — then one row per bucket aggregates the whole window).
	GroupBy []Dimension
	// Filter constrains the rows before grouping (closed axes).
	Filter Filter
	// Measures selects the measures each result row carries (mandatory,
	// non-empty, closed, deduplicated).
	Measures []Measure
	// Sort is the closed sort key (default: SortKeyBucketAsc when empty).
	Sort SortKey
	// SortMeasure names the measure used by SortKeyMeasureAsc/Desc. When a
	// measure sort is requested it must be a closed measure AND a member of
	// the selected Measures: sorting by an unselected measure would read a
	// missing row value as zero and silently degenerate the order and the
	// pagination cursor, so Validate refuses it loudly.
	SortMeasure Measure
	// Limit bounds the page size (1..MaxRowsPerQuery, mandatory).
	Limit int
	// Cursor is the opaque deterministic pagination cursor returned by a
	// previous page ("" = the first page). A stale or malformed cursor is
	// rejected with ErrBadCursor — a query never silently restarts at the
	// beginning.
	Cursor string
}

Query is the typed rollup read. The window is mandatory; every other field is closed / enumerated and validated by Validate. A Query is immutable in the intended usage — Validate does not mutate it.

func (q Query) Validate() error

type Result ¶

type Result struct {
	// Rows is the page, in the query's total order (nil when empty).
	Rows []Row
	// NextCursor is the opaque cursor for the next page ("" when this is
	// the last page).
	NextCursor string
}

Result is one query page.

type Row ¶

type Row struct {
	// BucketStart is the bucket the row aggregates (coarsened to the
	// query's Bucket size, UTC).
	BucketStart time.Time
	// Dimensions carries the query's GroupBy dimension values (empty when
	// GroupBy was empty — the row aggregates the whole window).
	Dimensions DimensionValues
	// Measures carries the query's requested measures and their exact
	// integer sums / folds.
	Measures map[Measure]MeasureValue
}

Row is one grouped result row.

type SessionTriple ¶

type SessionTriple struct {
	TenantID  string
	UserID    string
	SessionID string
}

SessionTriple is the comparable, collision-free key for a session's isolation triple. It is the ONLY sanctioned way to key per-session state (the erasure fence): identity validation permits NUL characters inside ids, so NUL-joining the three strings into one key would alias distinct triples (e.g. tenant "a\x00b" vs tenant "a", user "b"). A struct of the three string fields is comparable and never aliases.

func TripleOf(id identity.Identity) SessionTriple

func (t SessionTriple) Matches(k Key) bool

type SortKey ¶

type SortKey string

SortKey is the closed set of query sort keys. Every sort is total: the primary key, then (bucket start, then the grouped dimension values in canonical order) as deterministic tie-breakers, so pagination never skips or repeats a row on a stable store.

const (
	// SortKeyBucketAsc sorts chronologically, oldest bucket first.
	SortKeyBucketAsc SortKey = "bucket_asc"
	// SortKeyBucketDesc sorts newest bucket first.
	SortKeyBucketDesc SortKey = "bucket_desc"
	// SortKeyMeasureAsc sorts by the query's SortMeasure sum, ascending.
	SortKeyMeasureAsc SortKey = "measure_asc"
	// SortKeyMeasureDesc sorts by the query's SortMeasure sum, descending.
	SortKeyMeasureDesc SortKey = "measure_desc"
)

type Source ¶

type Source interface {
	// Next returns at most limit events whose Sequence is strictly greater
	// than after, in ascending Sequence order (oldest first). A short
	// non-empty batch does NOT report exhaustion — only (nil, nil) means
	// "caught up". An error stops the projector with StateUnavailable; the
	// checkpoint is NOT advanced.
	Next(ctx context.Context, after uint64, limit int) ([]events.Event, error)
}

Source yields successfully-persisted canonical events for the projector to consume, in global bus-sequence order.

"Successfully persisted" is the contract: the durable event log is the canonical source — the runtime's StateStore-backed durable bus driver persists every event before it is fanned out, and its per-bus sequence is gap-free. A Source implementation SHOULD be backed by that log (for example by scanning its persisted entry records via the StateStore maintenance surface, in global sequence order) and MUST return each sequence at most once across calls, in strictly ascending order, without gaps.

Next promises AT MOST limit events: a batch of fewer than limit events does NOT prove the source is exhausted — more events may exist beyond the returned prefix. Only an empty read (nil slice) reports "caught up". This is the load-bearing contract behind the projector's state machine: a short non-empty batch leaves the projector in StateCatchingUp until a SUBSEQUENT read returns empty.

The projector owns the cursor: Next is called with the projector's current checkpoint and must return events strictly newer than it. (nil, nil) means the source holds nothing newer (the projector reports StateCurrent).

type State ¶

type State string

State is the projector's catch-up quality.

const (
	// StateCurrent — the last read was EMPTY: nothing newer than the
	// watermark existed at that read, so the rollups are caught up with
	// the log as of it. A live runtime may persist new events a moment
	// later; the next Advance moves back to StateCatchingUp until it
	// drains them. A short non-empty batch NEVER proves current — only an
	// empty read does.
	StateCurrent State = "current"
	// StateCatchingUp — more events remain, or the projector has not yet
	// verified the log head after a construction / rebuild, or the last
	// read returned a non-empty batch (short or full) that did not prove
	// exhaustion. Rollups may trail the live log.
	StateCatchingUp State = "catching_up"
	// StateUnavailable — the source or the store failed; the projector
	// cannot make progress. Quality.Err carries the last failure.
	StateUnavailable State = "unavailable"
)

type Store ¶

type Store interface {
	// ApplyBatch atomically applies the batch's deltas and advances the
	// checkpoint to batch.Checkpoint. A batch whose Checkpoint does not
	// advance the stored checkpoint is a no-op (idempotent replay: because
	// deltas + checkpoint are atomic, every event at or below the stored
	// checkpoint is already applied, so a non-advancing batch has nothing
	// new to apply — this is what makes concurrent advances and restart
	// replays safe). A delta for a fenced triple rejects the WHOLE batch
	// with ErrSessionFenced — the checkpoint does not advance, and the
	// projector drops the offending event and retries.
	ApplyBatch(ctx context.Context, batch Batch) error

	// Query executes a validated rollup query. The query MUST pass
	// Validate (the store re-validates and returns the wrapped
	// ErrQueryInvalid / ErrQueryBudget sentinels). The response page is
	// deterministic for a stable store: same query + same cursor ⇒ same
	// rows, and pages never skip or repeat a row. Result measure values
	// are exact integers (MeasureValue) — counters are never normalised to
	// float64, so values above 2^53 stay exact.
	Query(ctx context.Context, q Query) (Result, error)

	// FenceSession erases every row for the session triple and fences the
	// triple PERMANENTLY so no future ApplyBatch can create rows for it,
	// and no Rebuild can clear it. Idempotent. This is the rollups side of
	// the session-erasure cascade; the runtime calls it when a session is
	// erased, and the projector drops late events for fenced triples at
	// ingestion time.
	FenceSession(ctx context.Context, id identity.Identity) error

	// IsFenced reports whether the session triple is fenced (erased).
	IsFenced(ctx context.Context, id identity.Identity) (bool, error)

	// Checkpoint returns the last applied sequence (0 = nothing applied).
	Checkpoint(ctx context.Context) (uint64, error)

	// Retention returns the oldest and newest bucket start currently
	// retained, or (zero, zero) when no rows exist. The two instants are
	// the row-level (StoreGranularity — minute) boundaries; a query
	// coarsens them.
	Retention(ctx context.Context) (oldest, newest time.Time, err error)

	// Rebuild clears every row and the checkpoint (reset to 0) so the
	// projector reprocesses the full log from the beginning. Erasure
	// fences are PERMANENT and are never cleared by Rebuild: rebuilding
	// projection rows or the checkpoint cannot authorize the resurrection
	// of an erased session. The projector's restart catch-up and rebuild
	// paths both rest on it.
	Rebuild(ctx context.Context) error

	// Close releases the store's resources. Idempotent; later calls
	// return ErrClosed.
	Close(ctx context.Context) error
}

Store is the mandatory persistence surface of the rollups domain. It is deliberately driver-shaped so indexed implementations can back it — the shipped in-memory reference (memstore), plus SQLite and Postgres implementations sharing the interface and the conformancetest suite.

Intended SQL shape for the row table (single table, one row per Key, all measures as exact BIGINT — never DOUBLE PRECISION):

rollup_rows(
    bucket_start          TIMESTAMPTZ NOT NULL,  -- UTC, on the MINUTE grid
    tenant_id             TEXT NOT NULL,
    user_id               TEXT NOT NULL,
    session_id            TEXT NOT NULL,
    model                 TEXT NOT NULL DEFAULT '',
    llm_completions       BIGINT NOT NULL DEFAULT 0,
    llm_tokens_prompt     BIGINT NOT NULL DEFAULT 0,
    llm_tokens_completion BIGINT NOT NULL DEFAULT 0,
    llm_tokens_reasoning  BIGINT NOT NULL DEFAULT 0,
    llm_tokens_cache_read BIGINT NOT NULL DEFAULT 0,
    llm_tokens_cache_write BIGINT NOT NULL DEFAULT 0,
    llm_tokens_total      BIGINT NOT NULL DEFAULT 0,
    llm_cost_micros       BIGINT NOT NULL DEFAULT 0,  -- exact micro-units of USD
    llm_latency_count     BIGINT NOT NULL DEFAULT 0,
    llm_latency_sum_ms    BIGINT NOT NULL DEFAULT 0,
    llm_latency_min_ms    BIGINT NOT NULL DEFAULT 0,
    llm_latency_max_ms    BIGINT NOT NULL DEFAULT 0,
    tasks_completed       BIGINT NOT NULL DEFAULT 0,
    tasks_failed          BIGINT NOT NULL DEFAULT 0,
    tasks_cancelled       BIGINT NOT NULL DEFAULT 0,
    PRIMARY KEY (bucket_start, tenant_id, user_id, session_id, model)
)
CREATE INDEX idx_rollup_bucket_tenant ON rollup_rows (bucket_start, tenant_id);
CREATE INDEX idx_rollup_bucket_tenant_user ON rollup_rows (bucket_start, tenant_id, user_id);
CREATE INDEX idx_rollup_tenant ON rollup_rows (tenant_id);

The checkpoint and fence state are single-row tables:

rollup_checkpoint(id INTEGER PRIMARY KEY CHECK (id = 1), sequence BIGINT NOT NULL);
rollup_fence(tenant_id, user_id, session_id, PRIMARY KEY (tenant_id, user_id, session_id));

The interface contracts:

• ApplyBatch is atomic (all deltas + the checkpoint move in one transaction). A batch whose Checkpoint does not advance the stored checkpoint is a no-op — this is the replay-idempotency invariant.
• Query is a pure read; it must not mutate stored state and must be safe for concurrent use. An indexed implementation must resolve the query against its bucket/dimension indexes (the bounded window + filter), never a full-table scan.
• FenceSession erases the triple's rows AND fences it PERMANENTLY; a later ApplyBatch that touches the triple fails with ErrSessionFenced (the erasure is never resurrected by a late event). Rows for a fenced triple are never returned by Query. There is NO unfence operation: the fence outlives Rebuild — reprojection must never resurrect an erased session.
• Checkpoint / Retention / Rebuild are the projector's restart and rebuild surfaces. Rebuild resets rows and the checkpoint ONLY; the erasure fences are never cleared.

A Store MUST be safe for concurrent use by N goroutines against a single shared instance (the concurrent-reuse contract; the conformance suite pins it).