store

package
v0.2.1 Latest Latest
Warning

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

 Go to latest Published: Jul 2, 2026 License: AGPL-3.0 Imports: 18 Imported by: 0

Documentation

Overview

Package store is the akari-server data layer: a Postgres connection pool, the startup migration runner, and the query methods the rest of the server uses.

Index

Constants

View Source 
const DefaultSort = "updated"

DefaultSort is the global session list's order when none is requested: most recently active first, matching the feed's "find a recent run" purpose.

Variables

View Source 
var ErrBlobHashMismatch = errors.New("uploaded blob bytes do not match the declared hash")

ErrBlobHashMismatch reports that the uploaded bytes did not hash to the declared key, so the body was not stored. The handler maps it to a 400: it is the client's error, not a server fault.

View Source 
var ErrBlobNotUploaded = errors.New("referenced tool body is not present in the CAS")

ErrBlobNotUploaded reports that a tool body the transcript references by sha256 is not in the CAS. Under the client-CAS protocol the client uploads every body before the transcript that references it, so this means an out-of-order or dropped upload; the parse leaves the cursor for a retry rather than inventing a dangling reference.

View Source 
var ErrChunkNotLineAligned = errors.New("chunk must be non-empty and end on a newline")

ErrChunkNotLineAligned reports a chunk that is empty or does not end on a newline. The ingest protocol requires every stored byte to rest on a JSONL line boundary so the server only ever parses complete lines; the server enforces it rather than trusting the client.

View Source 
var ErrInvalidGrant = errors.New("invalid or expired grant")

ErrInvalidGrant is returned when an authorization code or refresh token is unknown, already used, expired, or revoked. It maps to the OAuth "invalid_grant" error at the token endpoint.

View Source 
var ErrInvalidInvite = errors.New("invalid or already used invite token")

ErrInvalidInvite is returned when registration presents an invite token that is unknown, already redeemed, or expired.

View Source 
var ErrNotFound = errors.New("not found")

ErrNotFound is returned by lookups that match no row.

View Source 
var ErrParserVersionStale = errors.New("parser version changed since last parse: reparse required")

ErrParserVersionStale reports that a session was partially parsed by a different parser version than the caller's, so incremental parsing cannot safely continue from the stored cursor. The fix is a reparse, which resets the projection and cursor and replays from zero. The raw bytes are untouched.

Functions

func HashBytes 

func HashBytes(content []byte) string

HashBytes returns the lowercase hex sha256 of content, the key the CAS uses.

func HashString 

func HashString(content string) string

HashString returns the lowercase hex sha256 of content. It hashes in place (the digest consumes the string in 64-byte blocks), so a large body is never copied into a byte slice just to be hashed.

func IsSortKey 

func IsSortKey(key string) bool

IsSortKey reports whether key names a sortable column of the global session list, so the handler can reject an unknown or tampered sort param before it reaches the query builder.

Types

type APIToken 

type APIToken struct {
	ID         int64
	UserID     int64
	Name       string
	Scope      string
	CreatedAt  time.Time
	LastUsedAt *time.Time
	RevokedAt  *time.Time
}

APIToken is a stored client token (the secret itself is never stored, only its hash).

type Analytics 

type Analytics struct {
	Series          []DayPoint
	Models          []Breakdown
	Agents          []Breakdown
	TotalCost       float64
	TotalIn         int64
	TotalOut        int64
	TotalCacheRead  int64
	TotalCacheWrite int64
	// TotalReasoning is the window's reasoning-token volume, summed from the by-agent split
	// like the other totals. It sits beside TotalTokens rather than inside it (see
	// Breakdown.Reasoning for why), so the Tokens tile shows it as a distinct class without
	// disturbing the headline-equals-sum-of-series reconciliation the four billed classes hold.
	TotalReasoning int64
	Sessions       int
	// CostIncomplete is true when any usage event in the window carried token
	// volume but no price, so TotalCost is a lower bound. The headline Cost tile
	// shows the "$X+" marker when set, matching how a single session flags an
	// incomplete cost. It is the OR of the by-agent slices, the same rows the
	// headline totals are summed from.
	CostIncomplete bool
	// Cache is the prompt-cache effectiveness over the same scope: hit rate, the
	// dollars caching saved, and the prompt-token split. It reads from the same dated
	// usage_events base as the totals (see CacheStats), so the Cache tile reconciles
	// with the Tokens tile beside it rather than counting usage the panel drops.
	Cache CacheStats
}

Analytics is everything the inline charts render from, scoped by an AnalyticsFilter (a project or the whole instance, a trailing window, and an optional user/agent/machine narrowing).

func (Analytics) HasData 

func (a Analytics) HasData() bool

HasData reports whether there is anything worth charting, so the view can show an empty state instead of an axis with no line.

func (Analytics) TotalTokens 

func (a Analytics) TotalTokens() int64

TotalTokens is the combined token volume across all four classes (input, output, cache read, cache write). The overview's Tokens readout shows this one figure and keeps the per-class split behind its tooltip.

type AnalyticsFilter 

type AnalyticsFilter struct {
	ProjectID int64
	Since     time.Time
	// Until is an exclusive upper bound on occurred_at; the zero value means no upper
	// bound (every figure through the latest event). The OG card sets it to the end
	// of the current day so its headline and caption cover exactly the days its
	// heatmap draws (the grid stops at today), rather than folding a future-dated
	// event into the total that no visible cell accounts for. The live pages leave it
	// zero.
	Until   time.Time
	UserIDs []int64
	// Username scopes by a single account name, the form the project page's filter
	// carries. It is independent of UserIDs (the overview's multi-select by id): an
	// unknown name resolves to no session, the same empty result the session list's
	// u.username = $ predicate gives, so the panel and the table stay in lockstep
	// even for a stale or mistyped user rather than the panel falling back to every
	// user while the table shows nothing.
	Username string
	Agent    string
	Machine  string
}

AnalyticsFilter scopes an Analytics query. The zero value is the whole instance, all of history, every user, every agent and machine. ProjectID 0 means all projects; a zero Since means all of history; an empty UserIDs/Agent/Machine means no scoping on that dimension. The project page sets Agent/User/Machine so its usage panel reflects the same filter as its session table, which keeps the panel headline and the rows beneath it reconciled under a filter rather than letting the headline stay instance-wide while the rows narrow.

type AnnounceParams 

type AnnounceParams struct {
	UserID          int64
	Agent           string
	SourceSessionID string
	ProjectID       int64
	Kind            string
	GitBranch       string
	Cwd             string
	Machine         string
}

AnnounceParams carries what a client reports when announcing a session. Kind is the session's classification ("remote", "standalone", or "orphaned"); it gates the downgrade guard in Announce.

type AnnounceResult 

type AnnounceResult struct {
	SessionID    int64
	StoredBytes  int64
	PrefixSHA256 string
}

AnnounceResult is the server's authoritative view of a session's raw store.

type AttachmentDelta 

type AttachmentDelta struct {
	MessageOrdinal int
	SHA256         string
	Body           string
	Bytes          int64
	MediaType      string
	Filename       string
}

AttachmentDelta is one attachments insert (today a lifted image). Like a tool body it reaches the CAS by one of two paths: when the client lifted the image, SHA256 names the already-uploaded blob and applyDelta records the reference with no blob write; otherwise Body holds the decoded bytes inline for the server to store. Bytes and MediaType describe the decoded image so the row carries its size and type without fetching the blob.

type AttachmentView 

type AttachmentView struct {
	MessageOrdinal int
	SHA256         string
	MediaType      string
	ByteLen        int64
	Filename       string
}

AttachmentView is one attachment (today a lifted image) rendered under its message: the blob key plus enough metadata to show or link the image without fetching it. The bytes are served on demand through the session-scoped blob route.

type AuthCode 

type AuthCode struct {
	ClientID      string
	UserID        int64
	RedirectURI   string
	CodeChallenge string
	Scope         string
	Resource      string
}

AuthCode is the data an authorization code carries from the authorize step to the token exchange: who approved it, for which client and redirect, the PKCE challenge the exchange must answer, and the scope and audience it grants.

type Blob 

type Blob struct {
	SHA256      string
	ByteLen     int64
	MediaType   string
	ContentType string
}

Blob is a stored content-addressed body: its key, stored size, semantic media type, and storage content type. The bytes themselves live in a Postgres large object referenced by lo_oid and are stored exactly as the client uploaded them. ContentType names how those bytes are encoded (application/octet-stream verbatim, or application/zstd compressed): the server never (de)compresses, so it serves this as the response's Content-Encoding and lets the client decode. ByteLen is the stored (possibly compressed) size; the raw body size lives on the tool_calls row.

type Breakdown 

type Breakdown struct {
	Label      string
	CostUSD    float64
	Input      int64
	Output     int64
	CacheRead  int64
	CacheWrite int64
	// Reasoning is the slice's reasoning-token volume, the class some agents (codex, pi)
	// report for the model's hidden deliberation. It is tracked and shown on its own but
	// deliberately excluded from Tokens(): reasoning is neither a prompt nor a cache class,
	// so folding it into the bar-sizing total would double-count against the billed classes
	// and unsettle the headline-equals-sum reconciliation. It surfaces as its own figure.
	Reasoning int64
	Sessions  int
	// CostIncomplete is true when this slice folded in a usage event that carried
	// real token volume but no price (an unpriced model), so the slice's cost is a
	// lower bound. It lets a by-model or by-agent row show the same "$X+" marker the
	// per-session figures use rather than an exact cost that understates the slice.
	CostIncomplete bool
}

Breakdown is one slice of a by-model or by-agent split: a label with its rolled cost, its token volume broken out by class, and how many sessions it touched. The per-class split lets a slice both size its bar on the full token volume and reproduce the same hover card every other token figure carries.

func (Breakdown) Tokens 

func (b Breakdown) Tokens() int64

Tokens is the all-class token volume (input, output, cache read, cache write) for the slice. The breakdown bars size and label on this, so a model's share reflects everything it was billed for. Sizing on uncached in/out alone (the old behavior) made cache-heavy models like Claude read as mispriced: a bar a third the width of its cost, because the cache tokens that drove the cost were absent from the figure beside it.

type CacheStats 

type CacheStats struct {
	Input      int64 // uncached prompt tokens, billed at the input rate
	Output     int64
	CacheRead  int64 // prompt tokens served from cache (the discounted read)
	CacheWrite int64 // prompt tokens written to cache (creation)
	SavingsUSD float64
	// SavingsIncomplete is true when some cached read or write volume rode an unpriced
	// model, so that model's saving is omitted and SavingsUSD is partial. Unlike cost,
	// this is NOT a clean lower bound: an omitted model's saving can be negative (a
	// Claude cache write is priced above input, a cost paid up front), so the true
	// figure could be lower OR higher than what is shown. The Cache readout flags it
	// "partial" rather than the cost figures' "$X+" lower-bound marker.
	SavingsIncomplete bool
}

CacheStats summarizes prompt-cache effectiveness over a scope: the prompt token volume split into uncached input, cached reads, and cache writes (creation), the output volume alongside, and the USD caching saved versus paying the uncached input rate for the same prompt tokens. It backs the Cache readout on the overview, project, and session views, the cache counterpart to the cost and token figures beside it.

func (CacheStats) HasData 

func (c CacheStats) HasData() bool

HasData reports whether any prompt tokens were seen, so a view can show an empty state instead of a 0% hit rate and a $0 saving on a scope with no usage.

func (CacheStats) HitRate 

func (c CacheStats) HitRate() float64

HitRate is the share of prompt tokens served from cache, 0..1. Cache writes count as misses: a token written to cache was read fresh on that turn, and only a later read of it is a hit. Zero when there are no prompt tokens, so a no-usage scope reads 0% rather than dividing by zero.

func (CacheStats) PromptTokens 

func (c CacheStats) PromptTokens() int64

PromptTokens is the total prompt-side token volume: uncached input plus cached reads plus cache writes. Output is excluded; caching is a prompt-side economy, so the hit rate measures the prompt, not the completion.

type ChurnFile 

type ChurnFile struct {
	Path     string
	Edits    int
	Sessions int
}

ChurnFile is one file's edit thrash over a scope: how many times it was edited and across how many sessions. The edits are deduped (a replayed transcript re-emits prior edits, so the raw rows over-count), the same dedup the tool analytics and the per-session signals apply.

type ConcurrencyStats 

type ConcurrencyStats struct {
	FleetPeak       int       // most sessions active simultaneously, anywhere in scope
	FleetPeakAt     time.Time // when the fleet peak was first reached
	BusiestUser     string    // the user whose own sessions overlapped the most
	BusiestUserPeak int       // that user's peak simultaneous sessions
	AvgConcurrent   float64   // total active session-time over the span the sessions cover
	Sessions        int       // sessions with a measurable span in scope
}

ConcurrencyStats answers "how many sessions ran at once" over a scope: the fleet's peak overlap and when it happened, the single busiest user's peak, and the average concurrency across the active span. It reads from session start/end spans, so it sees the same scoped sessions the rest of the Insights page does. A session with no parsed start or end (or an end before its start) carries no measurable span and sits out.

func (ConcurrencyStats) HasData 

func (c ConcurrencyStats) HasData() bool

HasData reports whether any scoped session had a measurable span, so the panel can show an empty state rather than a peak of zero.

type ContextHealthStats 

type ContextHealthStats struct {
	Sessions          int   // scoped sessions with a measured peak, the rate denominator
	PeakTokensP50     int64 // median session peak context, in tokens
	PeakTokensP90     int64 // 90th-percentile session peak context
	PeakTokensMax     int64 // heaviest single session peak
	TotalResets       int64 // inferred context resets summed over the cohort
	SessionsWithReset int   // sessions that reset context at least once
}

ContextHealthStats is the cohort's context-load picture over a scope: how heavy the scoped sessions' contexts got and how often they shed that context. The peak percentiles read straight off the stored per-session peaks (quality.ContextHealthFolder, materialized by the settle pass or a reparse); the reset figures sum the inferred compaction/clear counts. The cohort is the scoped sessions carrying a current-version signals row whose peak is measured (a session with no usage stores NULL and is left out), so every rate divides by the same measured set. A stale or missing signals row contributes nothing, the same way the quality distribution folds it into unknown, so the panel never mixes a half-rebuilt view.

func (ContextHealthStats) HasData 

func (h ContextHealthStats) HasData() bool

HasData reports whether the scope carried any measured session, so the panel can show a note rather than a row of zeroes for a window with no context-measured sessions.

type DayPoint 

type DayPoint struct {
	Day        time.Time
	CostUSD    float64
	Input      int64
	Output     int64
	CacheRead  int64
	CacheWrite int64
}

DayPoint is one day's aggregated usage, the unit of the analytics time series. Days with no priced usage events are absent rather than zero-filled; the chart layer decides how to render gaps.

type FacetCount 

type FacetCount struct {
	Value string
	Count int
}

FacetCount is one filter value and how many sessions carry it, for a faceted filter rail that shows counts beside each option.

type FacetValues 

type FacetValues struct {
	Agents   []string
	Machines []string
	Users    []string
}

FacetValues holds the distinct filter values available within a project's sessions, for populating the session-list filter dropdowns.

type FileChurn 

type FileChurn struct {
	Files   []ChurnFile
	Clipped int // churned files beyond the shown list
}

FileChurn is the cohort's edit-thrash picture: the files edited more than once in the window, the most-edited first. A file touched once is not churn and never appears. It is the fleet counterpart to the per-session edit_churn signal, pointing at the paths a fleet kept returning to (a sign of a hard spot, a flaky change, or a moving target).

func (FileChurn) HasData 

func (c FileChurn) HasData() bool

HasData reports whether any file was edited more than once, so the panel can show a note rather than an empty list for a window with no repeated edits.

type GlobalFacetValues 

type GlobalFacetValues struct {
	Agents   []FacetCount
	Machines []FacetCount
	Users    []FacetCount
	Projects []ProjectFacet
}

GlobalFacetValues holds the filter options for the cross-project session view, each with a count so the rail reads like an instrument.

type Insights 

type Insights struct {
	Quality     QualityDistribution
	Archetypes  []LabeledCount
	Concurrency ConcurrencyStats
	Velocity    VelocityStats
	Tools       ToolStats
	Hygiene     PromptHygiene
	Churn       FileChurn
	Context     ContextHealthStats
}

Insights is everything the Insights page renders for a scope: the quality distribution (grades and outcomes), the archetype mix, the concurrency figures, the velocity cadence, the tool reliability and mix, the prompt-hygiene rates, the file churn, and the context-load figures. It is the cross-cutting counterpart to Analytics (which is about cost and tokens), scoped by the same AnalyticsFilter so a window or a per-user narrowing applies to both surfaces alike.

func (Insights) HasData 

func (i Insights) HasData() bool

HasData reports whether any scoped session carried signals, so the page can show an empty state instead of a row of zero bars on a scope with no sessions.

type LabeledCount 

type LabeledCount struct {
	Key   string
	Count int
}

LabeledCount is one bar in a distribution: a canonical key (the stored value, or "" for the unscored grade bucket) and how many sessions fell in it. The view maps the key to a display label and a colour; the store keeps the raw key so the two layers stay decoupled.

type Message 

type Message struct {
	Ordinal      int
	Role         string
	Content      string
	ThinkingText string
	Model        string
	HasThinking  bool
	HasToolUse   bool
	Timestamp    *time.Time
}

Message is one transcript row for rendering.

type MessageDelta 

type MessageDelta struct {
	Ordinal      int
	Role         string
	Content      string
	ThinkingText string
	Model        string
	HasThinking  bool
	HasToolUse   bool
	Timestamp    time.Time
}

MessageDelta is one message write. Each ordinal is written exactly once: the ingest protocol keeps a whole turn inside one chunk, so Content and ThinkingText are the complete text of the message, never a fragment to append.

type OAuthClient 

type OAuthClient struct {
	ID           string
	ClientName   string
	RedirectURIs []string
	CreatedAt    time.Time
}

OAuthClient is a dynamically registered MCP client (a coding agent). Clients are public: they authenticate with PKCE and hold no secret, so this is identity and the redirect allowlist only.

type OAuthGrant 

type OAuthGrant struct {
	ClientID    string
	ClientName  string
	Scope       string
	ConnectedAt time.Time
	LastUsedAt  time.Time
}

OAuthGrant summarizes a client a user has connected, for the account page's "connected apps" list. One row per client the user has live (unrevoked) tokens for, with when it was first connected and when it last authenticated a request or redeemed a refresh token.

type OAuthTokenParams 

type OAuthTokenParams struct {
	AccessHash       string
	RefreshHash      string // empty for no refresh token
	ClientID         string
	UserID           int64
	Scope            string
	Resource         string
	AccessExpiresAt  time.Time
	RefreshExpiresAt *time.Time
}

OAuthTokenParams is one issued token pair and its bindings.

type OffsetMismatchError 

type OffsetMismatchError struct{ StoredBytes int64 }

OffsetMismatchError reports that an append was attempted at the wrong offset; StoredBytes is the server's current cursor, which the client should resume at.

func (OffsetMismatchError) Error 

func (e OffsetMismatchError) Error() string

type OverviewOGImage 

type OverviewOGImage struct {
	PNG         []byte
	GeneratedAt time.Time
}

OverviewOGImage is a cached Open Graph preview card: the rendered PNG bytes and when they were generated. The generated_at stamp drives the TTL (a request past the cache window re-renders) and the cleanup sweep (an expired card is pruned).

type ProjToolCall 

type ProjToolCall struct {
	MessageOrdinal int
	CallIndex      int
	ToolName       string
	Category       string
	FilePath       string
	InputBody      string
	InputSHA256    string
	InputBytes     int64
	InputMediaType string
	CallUID        string
}

ProjToolCall is one tool_calls insert. The input body lives in the CAS, by one of two paths: InputBody holds the bulky input inline and AdvanceProjection writes it and records the sha256; or InputSHA256 is already set because the client lifted the body to the CAS at upload time and left a sentinel, so the reference is recorded with no blob write. Exactly one of InputBody / InputSHA256 is set when there is an input. CallUID is the agent's call id, used to back-patch the result that arrives on a later line (and possibly a later region, for Claude).

type ProjUsage 

type ProjUsage struct {
	MessageOrdinal *int
	Model          string
	Input          int
	Output         int
	CacheWrite     int
	CacheRead      int
	Reasoning      int
	CostUSD        *float64
	OccurredAt     time.Time
	DedupKey       string
	SourceOffset   int64
	SourceIndex    int
}

ProjUsage is one usage_events insert. SourceOffset and SourceIndex make the insert idempotent (the unique index absorbs a replay via ON CONFLICT).

type ProjectFacet 

type ProjectFacet struct {
	ID    int64
	Key   string
	Name  string
	Kind  string
	Count int
}

ProjectFacet is one project option in the global session filter: enough to label, color, and link it, plus its session count.

type ProjectParams 

type ProjectParams struct {
	RemoteKey   string
	Host        string
	Owner       string
	Repo        string
	DisplayName string
	Kind        string
}

ProjectParams is the project identity carried by an announce request before the server knows the row id. Keeping this with Announce lets the downgrade guard run before a local project is inserted, so an old client cannot recreate an unused orphaned project row for a session already stuck to a remote.

type ProjectSummary 

type ProjectSummary struct {
	ID           int64
	RemoteKey    string
	Host         string
	Owner        string
	Repo         string
	DisplayName  string
	Kind         string
	SessionCount int
	TotalCostUSD float64
	TotalInput   int64
	TotalOutput  int64
	// Cache token rollups back the projects-index tokens column, whose hover
	// detail breaks the total into in, out, cache read, and cache write (the same
	// four classes the overview heatmap surfaces per day).
	TotalCacheRead  int64
	TotalCacheWrite int64
	// CostIncomplete is true when any session folded into this project's totals
	// carries an unpriced usage event, so the rolled-up cost is a lower bound. It
	// is the OR of the per-session cost_incomplete flags, letting the index render
	// the same "$X+" marker the per-session rows show instead of an exact figure
	// that silently understates an aggregate built from incomplete sessions.
	//
	// Like every other figure on this index, it is rollup-scoped (every surviving
	// usage row), so it can read true for a project whose only unpriced usage is
	// undated while the all-time analytics panel, which drops undated rows off its
	// time axis, reads exact. That is the one documented rollup-vs-analytics gap,
	// the same one the token and cost totals carry (see, in package store's tests,
	// TestUndatedUsageIsTheOnlyRollupAnalyticsGap): the flag tracks each surface's
	// own displayed total rather than diverging from it.
	CostIncomplete bool
	LastActivity   *time.Time
}

ProjectSummary is one row of the projects index: a project plus rolled-up session counts and token/cost totals.

func (ProjectSummary) TotalTokens 

func (p ProjectSummary) TotalTokens() int64

TotalTokens is the sum of every token class for a project: input, output, and both cache directions. It is the headline figure for the projects-index tokens column, matching how the overview heatmap totals a day.

type ProjectionDelta 

type ProjectionDelta struct {
	Messages    []MessageDelta
	ToolCalls   []ProjToolCall
	ToolResults []ToolResultDelta
	Usage       []ProjUsage
	Attachments []AttachmentDelta

	Started time.Time
	Ended   time.Time
}

ProjectionDelta is the incremental projection write for one parsed region: the rows to add and the region's timestamp span. The session rollups are not folded from precomputed counters carried here. They are derived from the rows that actually persist (see appliedDelta), because the row inserts dedup on conflict and the rollups must count exactly the surviving set. Claude streams one assistant message across several transcript lines that share its message id, so a region can carry the same usage block several times while the ledger keeps one; folding precomputed per-region deltas over-counted those duplicates.

type PromptHygiene 

type PromptHygiene struct {
	Prompts            int // human prompts across the scoped sessions, the rate denominator
	Short              int // prompts under the terse-word threshold
	Duplicate          int // prompts repeating an earlier one verbatim
	NoCodeContext      int // change requests that pointed at no code
	Sessions           int // scoped sessions carrying a current-version signals row
	UnstructuredStarts int // of those sessions, how many opened with an unstructured prompt
}

PromptHygiene is the cohort's input-quality picture over a scope: how many of the window's human prompts were terse, repeated, or asked for a change without pointing at code, and how many sessions opened with an unstructured prompt. The counts come from the stored per-session signals (aggregated by the settle pass or a reparse from the per-message hygiene columns quality.ClassifyPrompt materializes at insert), summed over the sessions carrying a current-version row so the numerators and the Prompts denominator cover the same set. A stale or missing signals row contributes nothing to either, the same way the quality distribution folds it into the unknown bucket, so the panel never mixes a half-rebuilt view.

func (PromptHygiene) HasData 

func (h PromptHygiene) HasData() bool

HasData reports whether the scope carried any measured prompt, so the panel can show a note rather than a row of zero-over-zero rates for a window with no signalled sessions.

type QualityDistribution 

type QualityDistribution struct {
	Grades   []LabeledCount // canonical order: A, B, C, D, F, then "" (unscored)
	Outcomes []LabeledCount // canonical order: completed, errored, abandoned, unknown
	Sessions int            // total scoped sessions (every session falls in one bucket)
}

QualityDistribution is the Insights page's quality summary over a scope: how the scoped sessions split across letter grades and across outcomes, plus the total. Every scoped session contributes to exactly one bucket of each split: a session whose signals row is missing or was written by an older scoring version (before its backfill reparse) reads as unscored and unknown rather than vanishing, so the splits cover the same session set the archetype distribution and the session count do, and the three reconcile. The parsed views are gated during a reparse, so a reader never sees a half-rebuilt distribution.

type ReduceFunc 

type ReduceFunc func(state, region []byte, baseOffset int64) (newState []byte, d ProjectionDelta, err error)

ReduceFunc parses a raw region beginning at baseOffset, given the prior serialized parser state, and returns the new state plus the projection delta. It is pure CPU: AdvanceProjection runs it inside the parse transaction, so it must not perform I/O.

type ReparseLock 

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

ReparseLock holds the session-scoped advisory lock that serializes a reparse across processes. It pins a dedicated pooled connection for the lock's lifetime, because a Postgres advisory lock is tied to the session that took it; Release unlocks and returns the connection to the pool.

func (*ReparseLock) Release 

func (l *ReparseLock) Release(ctx context.Context)

Release unlocks the advisory lock and returns the pinned connection to the pool. It is safe to call once. Pass a bounded context (the caller derives one that is detached from request/shutdown cancellation but capped) so a stuck unlock cannot block shutdown indefinitely.

If the unlock fails, the session may still hold the lock, so the connection is destroyed (hijacked out of the pool and closed) rather than returned: ending the session is what frees the advisory lock, and it stops a future pool user from inheriting a held reparse lock and wedging all later reparses.

type ReparseTarget 

type ReparseTarget struct {
	ID    int64
	Agent string
}

ReparseTarget identifies a session to re-parse. The reparse loop fetches targets a bounded page at a time (see SessionsForReparsePage), so the server never holds the whole session list resident at once.

type SessionDetail 

type SessionDetail struct {
	SessionSummary
	OwnerID       int64
	ProjectID     int64
	ProjectKey    string
	ProjectName   string
	ProjectKind   string
	Cwd           string
	ParentID      *int64
	ParserVersion int
	// TotalCacheSavingsUSD is the session's rolled-up prompt-cache saving (folded at parse
	// time beside total_cost_usd), so the Cache tile reads it in O(1) instead of scanning
	// usage_events on every live refresh. A session whose rollup is not yet backfilled is
	// priced and persisted once on read (see scanDetail), so the tile is never served the
	// seeded value and later reads stay O(1). CacheSavingsIncomplete flags that some cached
	// volume rode an unpriced model, so the figure is partial (and, unlike cost, not a clean
	// lower bound: an omitted model's saving can be either sign).
	TotalCacheSavingsUSD   float64
	CacheSavingsIncomplete bool
}

SessionDetail adds the owning project to a session summary, plus the rolled-up prompt-cache saving the session header's Cache tile renders. The saving lives only on the detail (not the list summary): the session list shows no cache tile, so the extra per-model-priced rollup rides only the single-session read that needs it.

type SessionFeedCursor 

type SessionFeedCursor struct {
	ID int64
}

SessionFeedCursor marks a position in the session feed: the id of the last row a page returned. The feed pages on the session id, which is immutable, rather than on updated_at. That matters for a client paging the whole feed across separate requests: a session re-activated mid-walk (its updated_at bumped by ingest or a re-parse) keeps the same id, so it never jumps past the cursor and drops out of a later page the way an updated_at keyset would silently skip it. Paging stays O(N), not the O(N^2) an OFFSET scan would cost.

type SessionFilter 

type SessionFilter struct {
	ProjectID int64
	Agent     string
	Machine   string
	Username  string
	// Since bounds the list to sessions last active at or after this instant,
	// matching the analytics window so a project page's session list and its usage
	// panel cover the same range. The zero time means no lower bound.
	Since time.Time
	// Sort names the column the global session list is ordered by (see
	// sessionSortColumns). The empty string means DefaultSort. Desc selects
	// descending order. Together they back the click-to-sort table headers; an
	// unknown Sort falls back to DefaultSort in the query builder.
	Sort   string
	Desc   bool
	Limit  int
	Offset int
}

SessionFilter narrows a session list. Empty fields are ignored.

type SessionPage 

type SessionPage struct {
	Sessions  []SessionSummary
	Remainder SessionRemainder
}

SessionPage is a project's windowed session table: the capped rows the page shows, newest-active first, plus the aggregate of every windowed session that did not fit (Remainder). Shown rows plus Remainder reproduce the usage panel's headline, since both the rows and the remainder derive from the one dated-usage base the panel sums.

type SessionRemainder 

type SessionRemainder struct {
	Sessions       int
	Input          int64
	Output         int64
	CacheRead      int64
	CacheWrite     int64
	CostUSD        float64
	CostIncomplete bool
}

SessionRemainder is the aggregate of the windowed sessions the capped table did not show: how many, their per-class token volume, their summed cost, and whether any of them carried unpriced usage. The project page renders it as a footer so the visible rows plus this line reconcile with the usage panel headline even when more sessions match than the table caps at. It carries all four token classes (not just a total) so the footer can show the same breakdown card every other token figure does, and its CostIncomplete is a bool_or over the hidden sessions alone, so the footer flags "$X+" only when a hidden session is the unpriced one, never because a visible row was.

func (SessionRemainder) Has 

func (r SessionRemainder) Has() bool

Has reports whether any windowed sessions fell outside the capped table, so the project page shows the reconciling footer only when rows were actually withheld.

func (SessionRemainder) Tokens 

func (r SessionRemainder) Tokens() int64

Tokens is the hidden tail's all-class token volume, the figure the footer's total shows with the per-class split behind its card, matching every other token readout.

type SessionRow 

type SessionRow struct {
	SessionSummary
	ProjectID   int64
	ProjectKey  string
	ProjectName string
	ProjectKind string
}

SessionRow is one row of the global (cross-project) session list: a session summary plus the project it ran in, so a reader can scan and filter every session in one place without first choosing a project.

type SessionSignals 

type SessionSignals struct {
	SessionID            int64
	Version              int
	Outcome              string
	OutcomeConfidence    string
	Score                *int    // nil when the session is unscored
	Grade                *string // nil when the session is unscored
	ToolCalls            int
	ToolFailures         int
	ToolRetries          int
	EditChurn            int
	LongestFailureStreak int
	// Prompt-hygiene counts describe the human's input, not the agent's work, so they
	// ride alongside the tool-health counts but never feed the score. PromptCount is the
	// classifier's base (non-empty human prompts), the denominator the counts are read
	// against.
	PromptCount          int
	ShortPromptCount     int
	DuplicatePromptCount int
	NoCodeContextCount   int
	UnstructuredStart    bool
	// HygieneMeasured is true only when the row's stored prompt-hygiene counts were derived at
	// the running quality.PromptFactsVersion. The classifier version is deliberately separate
	// from the scoring quality.Version (see quality.PromptFactsVersion), so a row still at the
	// current signals_version can carry hygiene counts from a superseded classifier until the
	// reparse re-derives them. The read sets this from session_signals.prompt_facts_version, and
	// HasHygieneSignal gates on it, so a stale-classifier count never surfaces as a signal.
	HygieneMeasured bool
	// Context-health figures describe resource load, not the agent's work, so like the
	// hygiene counts they ride alongside the score without feeding it. Both are nil when
	// the session had no usage to measure, so the UI can tell "unmeasured" apart from a
	// measured zero. PeakContextTokens is the heaviest single-turn context the session
	// reached; ContextResetCount is how many inferred context resets (compactions or
	// clears) it went through.
	PeakContextTokens *int64
	ContextResetCount *int
}

SessionSignals is a session's stored behavioral signals: its outcome, its quality score and grade (nil when unscored), and the tool-health counts the score is built from. It is the read shape of the session_signals row, derived from the session's own projection and materialized by the settle pass once the session settles, or re-derived on reparse.

func (SessionSignals) HasContextHealth 

func (s SessionSignals) HasContextHealth() bool

HasContextHealth reports whether the session had usage to measure, so the UI can show the context readout only when there is a real figure rather than a blank stand-in. Peak and reset count are populated together, so testing the peak is enough.

func (SessionSignals) HasHygieneSignal 

func (s SessionSignals) HasHygieneSignal() bool

HasHygieneSignal reports whether any prompt-hygiene signal fired, so the UI can omit the input readout for a session whose prompts were all clean. It is false when the row's hygiene is not measured at the current classifier version (HygieneMeasured), so the session page never shows a count a superseded classifier produced: the block reads as unmeasured until the reparse re-derives the facts, the same way the fleet hygiene aggregate excludes the row.

func (SessionSignals) HasToolActivity 

func (s SessionSignals) HasToolActivity() bool

HasToolActivity reports whether the session ran any tools, so the UI can omit the tool-health detail for a pure-conversation session that has none.

func (SessionSignals) Scored 

func (s SessionSignals) Scored() bool

Scored reports whether the session carries a score and grade, so the UI can show a grade tile or fall back to the outcome alone for an unscored (unknown, no-signal) session.

type SessionSummary 

type SessionSummary struct {
	ID               int64
	Agent            string
	Machine          string
	GitBranch        string
	Username         string
	MessageCount     int
	UserMessageCount int
	TotalInput       int64
	TotalOutput      int64
	TotalCacheWrite  int64
	TotalCacheRead   int64
	TotalCostUSD     float64
	CostIncomplete   bool
	Visibility       string
	PublicID         *string
	StartedAt        *time.Time
	EndedAt          *time.Time
	UpdatedAt        *time.Time
}

SessionSummary is one row of a session list (project view, search results).

type Store 

type Store struct {
	Pool *pgxpool.Pool
}

Store wraps a Postgres connection pool.

func Open 

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

Open connects to Postgres and verifies the connection.

func (*Store) AcquireReparseLock 

func (s *Store) AcquireReparseLock(ctx context.Context) (*ReparseLock, bool, error)

AcquireReparseLock tries to take the fleet-wide reparse advisory lock without blocking. It returns (lock, true, nil) when the lock was acquired and the caller owns it until Release, or (nil, false, nil) when another instance already holds it, in which case the caller should skip its reparse. The lock is advisory and session-scoped, so a crashed holder releases it automatically when its connection drops.

func (*Store) AdvanceProjection 

func (s *Store) AdvanceProjection(ctx context.Context, sessionID int64, parserVersion int, reduce ReduceFunc) (parsedTo int64, caughtUp bool, err error)

AdvanceProjection parses the next unparsed region of a session and applies it incrementally. It locks the session_raw row, reads up to parseBatchBytes of raw content past the parse cursor, runs reduce, applies the delta, and advances the cursor and parser state, all in one transaction. It returns the new parse cursor and whether the session is now fully parsed.

It is a no-op (caughtUp=true) when the cursor already equals the stored length. It returns ErrParserVersionStale when a partially parsed session was last touched by a different parser version, leaving the projection for a reparse to rebuild. The raw bytes are never modified here.

func (*Store) Analytics 

func (s *Store) Analytics(ctx context.Context, f AnalyticsFilter) (Analytics, error)

Analytics aggregates usage for the charts, scoped by f (see AnalyticsFilter): project or whole instance, a trailing window, and an optional user/agent/machine narrowing applied uniformly to every base.

Every figure derives from one base set, the scoped dated usage_events, so the headline totals, the daily series, the by-model split, and the by-agent split all reconcile by construction: sum the per-day cells, or the by-model tokens, or the by-agent tokens, and you get the headline tokens, every time. This is deliberate. The figures used to come from three different sources (tokens from the daily series, cost from the session rollups, the by-model split from a separate usage_events query that dropped unnamed models), so the headline and the rows beneath it could disagree by an order of magnitude. They are the same base now, grouped three ways (by day, by model, by agent), with the headline summed from one of them.

Every base shares the one filter the time axis forces: occurred_at IS NOT NULL. An undated usage event has no day to plot, so counting it in the headline but not in the daily cells would make the total exceed the sum of the chart, the exact drift this view exists to avoid. So the overview counts dated usage only, uniformly. In practice that excludes nothing: Claude, Codex, and pi all stamp the turn a usage line belongs to, so a NULL occurred_at is a malformed transcript to fix at ingest, not usage to scatter across the dashboard.

The headline totals are summed from the by-agent split rather than queried again: a session has exactly one agent, so the per-agent rows partition the usage cleanly and their sum is the grand total with no double counting. It reads inside one read-only REPEATABLE READ transaction so every grouped query and the Cache tile share a single MVCC snapshot: the headline token classes and the Cache split are the same sums regrouped, so a concurrent ingest landing between two pooled reads would let them disagree by the usage that arrived mid-render. Unlike AnalyticsSnapshot this takes no reparse-lock gate (a live panel tolerates a snapshot that falls during a reparse, where each session is atomically old or new), and being read-only it takes no locks that could block ingest.

func (*Store) AnalyticsSnapshot 

func (s *Store) AnalyticsSnapshot(ctx context.Context, f AnalyticsFilter) (a Analytics, ok bool, err error)

AnalyticsSnapshot reads Analytics as a single consistent snapshot that is guaranteed not to straddle a reparse, for the OG card render. It runs the three grouped queries inside one REPEATABLE READ transaction, so they all read one MVCC snapshot rather than three independently-timed reads. The first statement in that transaction is the reparse-lock check, which both establishes the snapshot and reads the lock state at that instant: if no reparse holds the lock at the snapshot point, none is mid-flight, so every session in the snapshot is in a settled state (fully reparsed or untouched) rather than a half-rebuilt mix, and a reparse that starts later cannot alter the frozen snapshot. When a reparse does hold the lock at that instant, it returns ok=false and no analytics, so the caller skips the render rather than caching a mixed aggregate. Unlike taking the reparse advisory lock itself, this holds no lock, so it never makes the fleet read as "reparsing".

func (*Store) Announce 

func (s *Store) Announce(ctx context.Context, p AnnounceParams) (AnnounceResult, error)

Announce upserts the session row (latest announce wins for mutable metadata), ensures its raw-store row exists, and returns the current cursor and hash.

Remote attribution is sticky: once a session resolves to a git-remote project, a later announce that can no longer find a remote (standalone or orphaned, because the folder lost its origin or was deleted) does not move it to a local project. Backed-up work keeps its repo grouping rather than sliding into an orphaned bucket the moment its checkout is removed. An upgrade in the other direction (a local session that gains a remote) is allowed and re-homes it.

func (*Store) AnnounceWithProject 

func (s *Store) AnnounceWithProject(ctx context.Context, p AnnounceParams, project ProjectParams) (AnnounceResult, error)

AnnounceWithProject upserts the project and the session in one transaction. For non-remote announces it first applies the sticky remote guard; when the guard wins, the local project is never inserted. The HTTP ingest path uses this form because it receives a project identity rather than a project id.

func (*Store) AppendChunk 

func (s *Store) AppendChunk(ctx context.Context, sessionID, offset int64, data []byte) (newStoredBytes int64, err error)

AppendChunk appends data at the given offset as a new raw chunk row. If offset does not match the server's current byte_len it returns OffsetMismatchError with the truth and makes no change. The prefix hash is advanced by resuming the stored sha256 state and folding in only the new bytes, so appending is work proportional to the chunk, not to the whole session.

func (*Store) ApplyProjectionDelta 

func (s *Store) ApplyProjectionDelta(ctx context.Context, sessionID int64, d ProjectionDelta) error

ApplyProjectionDelta applies a projection delta to a session in one transaction (message upserts, tool-call inserts with their CAS bodies, tool-result back-patches, and usage inserts) without advancing the parse cursor or the session aggregates. AdvanceProjection wraps applyDelta with that bookkeeping; this exposes just the row writes, which is the seam tests use to exercise the projection and CAS directly.

func (*Store) ArchetypeDistribution 

func (s *Store) ArchetypeDistribution(ctx context.Context, f AnalyticsFilter) ([]LabeledCount, error)

ArchetypeDistribution buckets the scoped sessions by shape on its own pooled connection for the Insights page. The snapshot path threads archetypeDistributionFrom so every panel reads one MVCC snapshot.

func (*Store) Attachments 

func (s *Store) Attachments(ctx context.Context, sessionID int64) ([]AttachmentView, error)

Attachments returns all of a session's attachments, ordered by the message they hang on, for the web renderer. Bounded readers pass an ordinal range to AttachmentsInRange.

func (*Store) AttachmentsInRange 

func (s *Store) AttachmentsInRange(ctx context.Context, sessionID int64, minOrdinal, maxOrdinal int) ([]AttachmentView, error)

AttachmentsInRange returns the attachments hanging on messages in the inclusive ordinal window [minOrdinal, maxOrdinal], so a bounded transcript read fetches only the attachments for the messages it returned.

func (*Store) BackfillCacheSavings 

func (s *Store) BackfillCacheSavings(ctx context.Context) (int, error)

BackfillCacheSavings prices each not-yet-backfilled session's cached usage into the total_cache_savings_usd rollup. The rollup is normally folded at parse time (applyDelta), and the epoch reparse fills it for the corpus, but a session that fails to reparse (a malformed transcript the parser cannot rebuild) keeps its old usage_events while the reparse rolls back, so its rollup would stay at a suspect value forever. This pass closes that gap by pricing the existing ledger directly, independent of the parse: the saving is a pure function of usage_events, so it is correct even when the transcript is not. It runs at startup after migrations.

A candidate is a cache-bearing session whose cache_savings_backfilled flag is false: the migration marks every pre-existing cache-bearing session that way and defaults sessions ingested afterward to true (their fold starts from a correct empty base, so they are authoritative from creation). The flag, not the stored number, is the "needs backfill" signal on purpose: a session seeded at 0 that took a live append folds only the new rows, leaving a partial nonzero total, so "total is zero" would wrongly pass it over. Pricing recomputes the full value from usage_events and sets the flag, so the session stops being a candidate and the pass converges to a no-op that is safe to run every startup. Each session is priced under a row lock (see backfillCacheSavingsForSession) so a write can never clobber the live parse fold. It keyset-pages by id so peak memory is one batch of ids, and returns how many sessions it corrected.

It first runs the pricing reconcile (reconcileCacheSavingsPricingIfNeeded): a rate change re-prices every cache-bearing session, not just never-folded ones, and a failed-reparse session would keep its old-priced rollup flagged backfilled=true out of the candidate set. The reconcile clears that flag across the cache-bearing corpus once per pricing.Version bump, so the drain below re-prices them at the current rates. On a steady-state startup (marker current) the reconcile is one O(1) read and the drain finds no candidates.

func (*Store) BlobMeta 

func (s *Store) BlobMeta(ctx context.Context, sha256hex string) (Blob, error)

BlobMeta returns a blob's stored size, media type, and storage content type without reading its body. The content type lets the serve path set Content-Encoding so the client decodes a compressed blob, while the server never touches the bytes.

func (*Store) CacheStats 

func (s *Store) CacheStats(ctx context.Context, f AnalyticsFilter) (CacheStats, error)

CacheStats aggregates prompt-cache effectiveness over the analytics scope. It shares the analytics base exactly: the scoped dated usage_events (occurred_at IS NOT NULL), grouped by model, so the cache figures reconcile with the usage panel they sit beside rather than counting undated usage the panel drops off its time axis. Savings is folded per model in Go because pricing is compiled into the binary, not in the database, so the rate gap that defines a saving is not a column to sum.

It reads on its own pooled connection. The snapshot path (AnalyticsSnapshot) instead threads its transaction through cacheStats, so the Cache tile and the token totals come from one MVCC snapshot and one connection rather than two.

func (*Store) Close 

func (s *Store) Close()

Close releases the connection pool.

func (*Store) ConcurrencyStats 

func (s *Store) ConcurrencyStats(ctx context.Context, f AnalyticsFilter) (ConcurrencyStats, error)

ConcurrencyStats computes the scope's overlap figures for the Insights page. Fleet peak, busiest user, and the average plus session count are three separate reads over the same span set, so it wraps them in one repeatable-read, read-only snapshot: a concurrent ingest between the reads could otherwise return a peak from one cohort with a session count and average from another. Insights threads its own snapshot through concurrencyStatsFrom; this is the standalone equivalent. The snapshot takes no row locks, so it never blocks ingest.

func (*Store) ConsumeAuthCode 

func (s *Store) ConsumeAuthCode(ctx context.Context, codeHash string) (AuthCode, error)

ConsumeAuthCode redeems an authorization code exactly once. It marks the code consumed and returns its bound data only if it was unconsumed and unexpired; otherwise it returns ErrInvalidGrant. The UPDATE ... RETURNING is the atomic single-use gate: a replayed code finds consumed_at already set and matches no row, so two token requests racing on the same code cannot both succeed.

func (*Store) ContextHealth 

func (s *Store) ContextHealth(ctx context.Context, f AnalyticsFilter) (ContextHealthStats, error)

ContextHealth aggregates the scoped sessions' context-load figures on its own pooled connection. Insights instead threads its snapshot transaction through contextHealthFrom, so the measured cohort shares one MVCC snapshot with the other panels.

func (*Store) CreateAPIToken 

func (s *Store) CreateAPIToken(ctx context.Context, userID int64, name, scope, tokenHash string) (int64, error)

CreateAPIToken stores a token's hash with a scope and returns its row id.

func (*Store) CreateAuthCode 

func (s *Store) CreateAuthCode(ctx context.Context, codeHash string, c AuthCode, expiresAt time.Time) error

CreateAuthCode stores an authorization code's hash with everything the token exchange will need to validate and honor it.

func (*Store) CreateInvite 

func (s *Store) CreateInvite(ctx context.Context, tokenHash string, createdBy int64, note string, expiresAt *time.Time) (int64, error)

CreateInvite stores an invite token hash issued by an admin.

func (*Store) CreateOAuthClient 

func (s *Store) CreateOAuthClient(ctx context.Context, id, name string, redirectURIs []string) error

CreateOAuthClient stores a dynamic client registration and returns nothing but an error: the caller already holds the generated id.

func (*Store) CreateOAuthToken 

func (s *Store) CreateOAuthToken(ctx context.Context, p OAuthTokenParams) error

CreateOAuthToken stores an issued access/refresh token pair.

func (*Store) CreateWebSession 

func (s *Store) CreateWebSession(ctx context.Context, id string, userID int64, expiresAt time.Time) error

CreateWebSession persists a browser session.

func (*Store) DeleteExpiredOGImages 

func (s *Store) DeleteExpiredOGImages(ctx context.Context, olderThan time.Time) (int64, error)

DeleteExpiredOGImages removes cached preview cards stamped before the cutoff, the housekeeping the cleanup loop runs. A card for a shared overview re-renders on demand, so pruning a stale one only discards bytes nobody is serving. It returns how many rows it removed.

func (*Store) DeleteSession 

func (s *Store) DeleteSession(ctx context.Context, sessionID int64) error

DeleteSession removes a session and everything derived from it. The foreign keys cascade messages, tool calls, usage events, attachments, and the raw bytes; child sessions have their parent pointer nulled. Any CAS blobs the session referenced are left for a later SweepBlobs to reclaim. Authorization (owner or admin) is enforced by the caller, so this deletes unconditionally by id and returns ErrNotFound when nothing matched.

func (*Store) DeleteWebSession 

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

DeleteWebSession removes a browser session (logout).

func (*Store) DuplicateCallUIDCount 

func (s *Store) DuplicateCallUIDCount(ctx context.Context, sessionID int64) (int, error)

DuplicateCallUIDCount returns how many of a session's tool-call ids appear on more than one row. The GROUP BY runs in the database against the (session_id, call_uid) index, so the result is a bounded scalar and the session view can flag a repeated id without loading or grouping the calls in process memory. It is normally zero; a non-zero count means the transcript replayed a turn (a resumed or compacted Claude session repeats a tool_use id), which the view surfaces as a chip so a genuinely malformed id reuse is visible rather than silent.

func (*Store) FileChurn 

func (s *Store) FileChurn(ctx context.Context, f AnalyticsFilter) (FileChurn, error)

FileChurn computes which files a scope edited repeatedly on its own pooled connection for the Insights page. The snapshot path threads fileChurnFrom so every panel reads one MVCC snapshot.

func (*Store) GlobalFacets 

func (s *Store) GlobalFacets(ctx context.Context) (GlobalFacetValues, error)

GlobalFacets returns the busiest agents, machines, usernames, and projects across all sessions, each with its session count, ordered busiest first. The counts are read from the session_facets rollup (maintained incrementally by a trigger on the sessions table, see migration 0005), so each category is a bounded top-N index read rather than a GROUP BY over the whole sessions table. It backs the global Sessions view's filter rail.

func (*Store) Insights 

func (s *Store) Insights(ctx context.Context, f AnalyticsFilter) (Insights, error)

Insights gathers the page's panels in one repeatable-read snapshot. The panels overlap: the quality split's session total, the archetype split, and the cohort denominators of the hygiene, context, and tool panels all describe the same scoped session set, so reading them on separate pooled connections would let a concurrent ingest land between two panels and make the page disagree with itself (the quality total and the archetype total, say, off by the one session that arrived mid-render). One read-only REPEATABLE READ transaction pins every panel to the same MVCC snapshot, so the overlapping totals reconcile exactly.

Unlike AnalyticsSnapshot it takes no reparse-lock gate: a live page tolerates a snapshot that falls during a reparse (each session is atomically old or new in it, since ReparseSession commits per session), it just must not straddle two snapshots within one render. The panels still fail fast on the first error, and the read-only transaction takes no row locks, so it never blocks ingest.

func (*Store) ListAPITokens 

func (s *Store) ListAPITokens(ctx context.Context, userID int64) ([]APIToken, error)

ListAPITokens returns a user's tokens, newest first.

func (*Store) ListAllSessions 

func (s *Store) ListAllSessions(ctx context.Context, f SessionFilter) ([]SessionRow, error)

ListAllSessions returns sessions across every project matching the filter, newest first. A zero ProjectID means "all projects"; the other fields narrow the set exactly as ListSessions does. This backs the global Sessions view and the Overview's recent-activity feed.

func (*Store) ListOAuthGrants 

func (s *Store) ListOAuthGrants(ctx context.Context, userID int64) ([]OAuthGrant, error)

ListOAuthGrants returns the clients a user has live tokens for, one row per client, newest connection first. It backs the account page's "connected apps" list. Revoked and fully expired tokens are excluded, so a disconnected client drops off the list.

func (*Store) ListProjects 

func (s *Store) ListProjects(ctx context.Context) ([]ProjectSummary, error)

ListProjects returns every project with rolled-up stats, most recently active first.

func (*Store) ListSessions 

func (s *Store) ListSessions(ctx context.Context, f SessionFilter) ([]SessionSummary, error)

ListSessions returns sessions matching the filter, newest first.

func (*Store) ListUsers 

func (s *Store) ListUsers(ctx context.Context) ([]User, error)

ListUsers returns every account, id and username only, ordered by username, to populate the overview's per-user activity filter. The password hash is left zero: this list names identities for a scope control, it does not carry the credential.

func (*Store) MessageCount 

func (s *Store) MessageCount(ctx context.Context, sessionID int64) (int, error)

MessageCount returns a session's current message count from its rollup.

func (*Store) Messages 

func (s *Store) Messages(ctx context.Context, sessionID int64) ([]Message, error)

Messages returns a session's whole transcript in order. The web renderer wants the full session in one pass; bounded readers (the MCP transcript window) use MessagesPage instead so peak memory does not scale with session size.

func (*Store) MessagesAfter 

func (s *Store) MessagesAfter(ctx context.Context, sessionID int64, after *int, limit int) ([]Message, error)

MessagesAfter returns the next window of a session's transcript ordered by ordinal, starting strictly after the given ordinal (after == nil for the first window). It pages by keyset on ordinal rather than OFFSET: each call walks the messages primary key (session_id, ordinal) straight to the resume point and reads only the next `limit` rows, so reading a whole session window by window costs O(N), not the O(N^2/limit) an OFFSET walk would (Postgres re-skips the already returned prefix on every page). limit is clamped to [1, 2000].

func (*Store) Migrate 

func (s *Store) Migrate(ctx context.Context, migrationFS embed.FS) error

Migrate applies every embedded migration not yet recorded, in lexical order, each inside its own transaction. It is safe to run on every startup.

func (*Store) MissingBlobs 

func (s *Store) MissingBlobs(ctx context.Context, shas []string) ([]string, error)

MissingBlobs reports which of a set of candidate hashes the CAS does not hold, and atomically (re)pins every hash it does hold. The client calls this before uploading tool bodies: a body the server already has (from an earlier sync, or any other session, since the CAS dedupes globally) is reported absent from the missing set and so not re-sent, but it is pinned here so it survives the sweep until the transcript chunk that references it commits. Without the pin a present but unreferenced, unpinned blob could be reclaimed in the window between this check and the transcript append, stranding a sentinel with no body.

The whole check-and-pin runs in one transaction so the pin is durable before the client is told a body is present. Pinning takes the blob rows FOR KEY SHARE (via the upsert's FK validation) which conflicts with the sweep's FOR UPDATE, so a body cannot be both reported-present and swept.

func (*Store) OAuthAccessAuth 

func (s *Store) OAuthAccessAuth(ctx context.Context, accessHash string) (userID int64, scope string, expiresAt time.Time, err error)

OAuthAccessAuth resolves a presented access-token hash to its owner, scope, and expiry, rejecting expired and revoked tokens. It is the MCP endpoint's bearer check; the expiry it returns lets the caller mirror the real token lifetime.

func (*Store) OAuthClient 

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

OAuthClient looks up a registered client by id.

func (*Store) OverviewOGImage 

func (s *Store) OverviewOGImage(ctx context.Context, userID int64) (OverviewOGImage, error)

OverviewOGImage loads the cached preview card for a user, addressed by id, or ErrNotFound when none is cached yet. It is a plain by-id read with no visibility join: the public serve path reads through PublicOverviewCard, which folds in the overview_public gate atomically. This by-id form backs the render path's own reconciliation (Generate reloads the canonical card after a skipped guarded write) and the tests, where the visibility gate is not the property under test.

func (*Store) Project 

func (s *Store) Project(ctx context.Context, id int64) (ProjectSummary, error)

Project returns one project's identity (without rollups).

func (*Store) ProjectSparklines 

func (s *Store) ProjectSparklines(ctx context.Context, days int) (map[int64][]float64, error)

ProjectSparklines returns, per project id, a fixed-length slice of daily cost over the last `days` days (index 0 oldest, last index today, UTC). Projects with no recent usage are simply absent from the map; the index view renders a flat line for them. One query buckets in Go so the index stays a single round trip regardless of project count.

func (*Store) PromptHygiene 

func (s *Store) PromptHygiene(ctx context.Context, f AnalyticsFilter) (PromptHygiene, error)

PromptHygiene aggregates the scoped sessions' prompt-hygiene counts on its own pooled connection. Insights instead threads its snapshot transaction through promptHygieneFrom, so the measured cohort shares one MVCC snapshot with the other panels.

func (*Store) PublicOverviewCard 

func (s *Store) PublicOverviewCard(ctx context.Context, username string) (User, OverviewOGImage, bool, error)

PublicOverviewCard resolves a username to its account and reads that account's cached Open Graph card in one query, gated on overview_public = TRUE. Folding the public check, the user lookup, and the card read into a single statement is what keeps the /u/<username>/og.png serve atomic: a split (resolve the user, then read the card) leaves a window where a concurrent unpublish between the two steps could serve a card for an overview that just went private. found is false when the name is unknown or the overview is not public, so the caller 404s the link. When found is true the account is public; card.PNG is nil when no card is cached yet (the LEFT JOIN yields NULLs), which the caller renders on demand. Only the fields the card path needs are loaded; the password hash is left zero.

func (*Store) PublicOverviewUser 

func (s *Store) PublicOverviewUser(ctx context.Context, username string) (User, error)

PublicOverviewUser resolves a username to its account for the logged-out overview page, only while that account's overview is public. The flag is folded into the WHERE clause, so an unknown or unpublished name yields ErrNotFound and the link 404s. Only the fields the public page needs are loaded; the password hash is left zero.

func (*Store) PublishOverview 

func (s *Store) PublishOverview(ctx context.Context, userID int64) error

PublishOverview marks a user's own usage overview public, so /u/<username> resolves for logged-out viewers. The address is the username, so there is no capability id to mint: the call only flips the gate.

func (*Store) PublishSession 

func (s *Store) PublishSession(ctx context.Context, sessionID, userID int64, candidateID string) (string, error)

PublishSession marks a session public and returns its public id, minting a new one only if it had none. The owner check is folded into the WHERE clause, so a session that does not belong to the user returns ErrNotFound and is untouched. Re-publishing an already-public session keeps the existing id, so a shared link stays valid across repeated publishes.

func (*Store) PutBlob 

func (s *Store) PutBlob(ctx context.Context, sha, mediaType, contentType string, r io.Reader) error

PutBlob stores a content-addressed body uploaded directly by the client and pins it against the sweep for blobPinTTL. The bytes the client sends are the STORED bytes (raw or zstd-compressed, as contentType declares); the server stores them verbatim and never (de)compresses, so it stays off the compression CPU path. The body streams in from r in bounded slices so neither side holds the whole body in memory: a 500 MiB tool result lands as a large object without a 500 MiB buffer. The stored bytes are verified against the claimed sha256 (which is the hash of the stored bytes), so a corrupt upload cannot poison the CAS; the server does not validate that a zstd-declared body actually decompresses, since that would cost the CPU this design avoids and the key already pins the exact bytes.

No database lock is held across the network read. An already-present body is pinned and committed in a short transaction before its (redundant) body is drained, so a slow duplicate upload cannot block the sweep's FOR UPDATE behind a FOR KEY SHARE held across a client-controlled read. A new body is written inside one transaction (Postgres large objects require it), but that transaction holds no lock on any existing row until it inserts the new blobs row at the end, so it does not block the sweep either.

func (*Store) PutOverviewOGImage 

func (s *Store) PutOverviewOGImage(ctx context.Context, userID int64, png []byte, generatedAt time.Time) (bool, error)

PutOverviewOGImage stores the rendered preview card for a user's published overview, stamped with the instant the card's analytics were taken (generatedAt), not the write time. It upserts on the one-per-user key, but only overwrites a card that is not newer than this one: the DO UPDATE is guarded on EXCLUDED.generated_at >= the stored generated_at. So when concurrent requests race to regenerate an expired card, a render that read an older analytics snapshot but finishes last cannot clobber a newer card and make stale content look fresh for a whole TTL. Ties (equal timestamps) win harmlessly, since the render is deterministic for a given analytics window.

It reports whether this card became the cached one: true when the row was inserted or the guarded update fired, false when a newer card was already present and the write was skipped. The caller uses that to avoid serving bytes it rendered but did not store (see ogimage.Generate), so the served image never diverges from the cache.

func (*Store) QualityDistribution 

func (s *Store) QualityDistribution(ctx context.Context, f AnalyticsFilter) (QualityDistribution, error)

QualityDistribution aggregates the scoped sessions' grades and outcomes. The grade split and the outcome split are separate scans, so it wraps them in one repeatable-read, read-only snapshot: the session total, the grade buckets, and the outcome buckets then all describe the same scoped cohort, where a concurrent session insert or signals_stale change between the two scans could otherwise pair a grade split from one cohort with an outcome split from another. Insights threads its own snapshot through qualityDistributionFrom so its panels reconcile against each other; this is the standalone equivalent. The snapshot takes no row locks, so it never blocks ingest.

func (*Store) RefreshSessionSignals 

func (s *Store) RefreshSessionSignals(ctx context.Context, sessionID int64) error

RefreshSessionSignals recomputes one session's signals in its own transaction. It is the standalone form the settle pass (RefreshSettledSignals) and the tests use; the reparse path calls refreshSignalsTx inside its existing transaction instead, so the signals commit with the projection rather than in a second round trip.

func (*Store) RefreshSettledSignals 

func (s *Store) RefreshSettledSignals(ctx context.Context) (int, error)

RefreshSettledSignals recomputes signals for every settled session marked stale. It is the production path that materializes signals: the append path no longer refreshes on catch-up (see AdvanceProjection), so a session's signals are computed once here, after it has been idle past the abandoned threshold, off the ingest hot path.

A session is due when it is settled (ended_at at least abandonedIdleMinutes in the past) AND signals_stale is set. The flag is the single-table marker that replaces a cross-table due predicate: applyAggregates and the reparse reset set it whenever the projection moves, and refreshSignalsTx clears it only when it grades a settled session. So it captures every way a stored grade can fall behind its source, without a join the settle scan would have to evaluate per row:

  • Never graded (a fresh ingest, or a session that predates signals): the column defaults true, so it is due from creation until the first settle grades it.
  • Graded before the projection last changed (a later chunk of a multi-upload historical session, whose ended_at stays far in the past so it looks long settled): the appended region set the flag again, so the stale partial grade is re-derived.
  • Graded before the session settled (a reparse that ran refreshSignalsTx while the session was still live, so its outcome was not yet stable): that refresh left the flag set because the session was not idleLongEnough, so the settle pass re-grades it once settled.

A stale signals_version is the one case the flag does not cover on its own (a version bump changes no projection), so reconcileStaleVersionsIfNeeded marks those rows before the drain. It runs the inequality scan once per quality.Version change, gated on a parse_meta marker, so a steady-state wake never pays it.

It drains the whole due backlog in bounded batches, keyset-paging the settled-and-stale tail once in (ended_at, id) order: each batch resumes strictly after the last row of the previous one, and a session drops out of the settle index the moment it is graded, so the pass reads only the due rows via the partial index (idx_sessions_signals_stale), O(D_due) per wake rather than O(settled history). Each session is refreshed in its own transaction so one slow session never holds a broad lock, cancellation stops the drain between sessions, and it returns how many it refreshed.

func (*Store) Register 

func (s *Store) Register(ctx context.Context, username, passwordHash, inviteHash string) (User, error)

Register creates a user. The first account ever created becomes admin and needs no invite; every later account must present an unredeemed, unexpired invite token (by its hash), which is redeemed atomically with the insert.

func (*Store) ReparseLockHeld 

func (s *Store) ReparseLockHeld(ctx context.Context) (bool, error)

ReparseLockHeld reports whether any session currently holds the reparse advisory lock for this database. It is how a server instance that is not itself reparsing learns that another instance is, so it can gate its parsed UI for the duration: the advisory lock is the authoritative cross-process "a reparse is running" signal, and it clears automatically if the holder crashes. The two-key advisory lock records classid = first key, objid = second key, objsubid = 2; classid is an oid, so the first key is compared as one (the cast also carries a negative hashtext result through as its unsigned bit pattern, the way Postgres stores it).

func (*Store) ReparseScope 

func (s *Store) ReparseScope(ctx context.Context, agent string) (total int, maxID int64, err error)

ReparseScope reports how many sessions a reparse will cover and the largest id among them, optionally filtered to one agent. The reparse loop pages through ids up to maxID, so a session ingested after the scope is read is left to the live parse path rather than swelling the count. total drives the progress bar's denominator; maxID bounds the keyset paging.

func (*Store) ReparseSession 

func (s *Store) ReparseSession(ctx context.Context, sessionID int64, parserVersion int, reduce ReduceFunc) error

ReparseSession rebuilds a session's projection from its stored raw bytes in a single transaction: it clears the derived rows and rewinds the cursor, then replays the whole session through reduce, all atomically. Because the clear and the replay commit together, two correctness properties hold that the older reset-then-advance path did not provide:

  • A concurrent reader never sees the session empty or half rebuilt. It sees the prior projection until this transaction commits, then the new one; there is no window of cleared-but-not-yet-replayed rows.
  • Any failure rolls the whole rebuild back. A parser error on malformed bytes or an operational store/CAS error leaves the prior projection intact rather than a cleared session, so a per-session parser failure loses no data and an operational failure is safe to retry.

The raw bytes are never modified. The replay is bounded to one region at a time (parseBatchBytes), so peak memory does not scale with session size even though the whole session is rebuilt in one transaction.

func (*Store) ReparsedEpoch 

func (s *Store) ReparsedEpoch(ctx context.Context) (int, error)

ReparsedEpoch reads the parser epoch the stored projection was last rebuilt under. A fresh database returns 0 (the column default), which differs from parse.Epoch so the server reparses on first start and converges.

func (*Store) ResetProjectionForReparse 

func (s *Store) ResetProjectionForReparse(ctx context.Context, sessionID int64, parserVersion int) error

ResetProjectionForReparse clears a session's parser-owned projection rows and its aggregates, and rewinds the parse cursor to zero at the given version, keeping the raw bytes and their hash. It is the standalone clear without the replay: the server reparses through ReparseSession, which composes this clear with an in-transaction replay so the rebuild is atomic. This form is kept for store tests that exercise the clear and its blob pin on their own. Attachments are parser-owned (the reducer emits them from the transcript's image events), so they are cleared here too; a reparse rewrites them, and the orphan sweep reclaims any blob left unreferenced.

func (*Store) ResetRaw 

func (s *Store) ResetRaw(ctx context.Context, sessionID int64) error

ResetRaw clears a session's raw store and its derived rows so the next chunk re-parses from zero. Dropping the tool_calls and attachments can orphan CAS blobs; like any deletion or re-parse, those are reclaimed by a later SweepBlobs rather than synchronously here, so a client reset stays cheap.

It takes the parent session row lock and then the session_raw lock, the locks AppendChunk and AdvanceProjection serialize on, so a reset cannot interleave with an in-flight append or parse and leave behind a chunk row or projection rows for a session it just zeroed. Taking the session row before session_raw matches DeleteSession's order, so the two cannot deadlock.

func (*Store) RevokeAPIToken 

func (s *Store) RevokeAPIToken(ctx context.Context, userID, tokenID int64) error

RevokeAPIToken marks a user's token revoked. It is a no-op if the token does not belong to the user.

func (*Store) RevokeOAuthGrant 

func (s *Store) RevokeOAuthGrant(ctx context.Context, userID int64, clientID string) error

RevokeOAuthGrant revokes every live token a user holds for one client, disconnecting it. It is scoped to the user, so a request cannot revoke another account's grant.

func (*Store) RotateOAuthToken 

func (s *Store) RotateOAuthToken(ctx context.Context, oldRefreshHash string, p OAuthTokenParams) (clientID string, userID int64, scope, resource string, err error)

RotateOAuthToken redeems a refresh token for a fresh access/refresh pair. It rewrites the existing row in place (so the grant stays one row per connection) only if the presented refresh token is live, returning the bindings the new access token inherits. A refresh token that is unknown, revoked, or past its expiry matches no row and yields ErrInvalidGrant. Rotating the refresh hash on every use makes refresh tokens single-use, so a leaked-and-replayed refresh token is caught the next time the legitimate client refreshes.

func (*Store) SessionCacheStats 

func (s *Store) SessionCacheStats(ctx context.Context, sessionID int64) (CacheStats, error)

SessionCacheStats recomputes one session's cache effectiveness by scanning its usage rows and pricing per model. Unlike the scoped CacheStats it counts ALL the session's usage, dated or not, so its prompt totals match the session's token rollups (sessions.total_*); the scoped path keeps the dated guard to match the time-bounded panel instead. That mirrors the one documented rollup-versus-analytics gap the cost and token figures already carry: a per-session figure counts every usage row, an analytics figure counts only the dated rows it can plot.

The session header no longer calls this on the hot path: it reads the same figure off the total_cache_savings_usd rollup (folded per row at parse time), which is O(1) rather than a per-refresh scan. This full recompute stays as the independent oracle the reconciliation test prices the rollup against, so a drift between the parse-time fold and a from-scratch per-model recompute fails a test rather than shipping a wrong tile.

func (*Store) SessionDetailByID 

func (s *Store) SessionDetailByID(ctx context.Context, id int64) (SessionDetail, error)

SessionDetailByID loads a session by numeric id.

func (*Store) SessionDetailByPublicID 

func (s *Store) SessionDetailByPublicID(ctx context.Context, publicID string) (SessionDetail, error)

SessionDetailByPublicID loads a published session by its public id.

func (*Store) SessionFacets 

func (s *Store) SessionFacets(ctx context.Context, projectID int64) (FacetValues, error)

SessionFacets returns the distinct agents, machines, and usernames present in a project's sessions, each sorted, for the project view's filter controls.

func (*Store) SessionFeed 

func (s *Store) SessionFeed(ctx context.Context, f SessionFilter, limit int, cursor *SessionFeedCursor) ([]SessionRow, *SessionFeedCursor, error)

SessionFeed returns one page of the cross-project feed (newest session first) and the cursor for the page after it (nil when this page is the last). It applies the same filters as ListAllSessions but pages by keyset on the immutable session id descending rather than OFFSET: each page resumes from the prior page's last id and reads only the next `limit` rows, so a client paging the whole feed never re-skips the rows it already saw and a concurrent updated_at bump cannot drop a row from the walk. Each row still carries its updated_at, so a caller can order by recency within a page. limit is clamped to [1, 500] (default 100).

func (*Store) SessionMeta 

func (s *Store) SessionMeta(ctx context.Context, sessionID int64) (userID int64, agent string, err error)

SessionMeta returns the owning user and agent of a session, or ErrNotFound.

func (*Store) SessionRawTo 

func (s *Store) SessionRawTo(ctx context.Context, w io.Writer, sessionID, limit int64) (written int64, truncated bool, total int64, err error)

SessionRawTo streams a session's raw uploaded bytes (the lossless JSONL the client sent, the source every projection is rebuilt from) to w in upload order, writing at most limit bytes. It returns the number of bytes written, whether the session held more than was written (so the caller can flag a truncated read), and the session's full raw length. A limit of zero or less means no cap. This is the raw underlying data behind the parsed transcript, exposed so an agent can inspect exactly what was ingested rather than only the projection. A missing session returns ErrNotFound.

func (*Store) SessionReferencesBlob 

func (s *Store) SessionReferencesBlob(ctx context.Context, sessionID int64, sha256hex string) (bool, error)

SessionReferencesBlob reports whether a session points at a blob, through a tool call's input or result or through an attachment. Blob serving is gated on this so a session can never read a blob it does not reference, even though the CAS dedupes content across sessions.

Lifting Codex images to the CAS means a session view fetches a blob per rendered image as well as per tool body, so this runs once per blob on every open and live refresh: it must stay logarithmic in the session's references, not scan them. Each arm is a hash-leading index lookup: tool_calls by (input_sha256, session_id) and (result_sha256, session_id), attachments by (sha256, session_id). The parameter is cast to char(64) so it matches the bpchar columns and their indexes; a bare text parameter would compare bpchar against text, cast the columns, and fall back to scanning the session's whole slice of tool calls and attachments.

func (*Store) SessionSignalsByID 

func (s *Store) SessionSignalsByID(ctx context.Context, sessionID int64) (SessionSignals, error)

SessionSignalsByID reads a session's current-version, up-to-date stored signals. A session with no usable row reads as an unknown, unscored result rather than an error, so the session page renders a neutral state instead of a stale or missing grade. A row is usable only when it is at the current signals_version AND the session is not flagged signals_stale, so the header and the fleet aggregates gate on the same flag and agree on exactly which grades count. signals_stale is set whenever the projection moves (applyAggregates and the reparse reset), so a session that gained an appended region after its last grade reads as unmeasured until the settle pass re-grades it, rather than showing a grade for an earlier, smaller session. The flag also covers the pre-settle case: refreshSignalsTx leaves it set when it grades a still-live session, so a not-yet-stable outcome (abandoned versus unknown turns on the idle gap) never reaches a reader before the settle pass pins it.

Gating on the flag rather than a refreshed_at >= updated_at comparison is deliberate. updated_at also moves on metadata-only writes (an announce re-announce, an owner reassignment) that leave the grade valid, so keying reads on it would strand those grades unread while the settle pass, which keys on the flag, never revisits them. The flag is set at exactly the projection-change sites, so it is the precise "grade is behind its source" signal that updated_at is not.

func (*Store) SessionsForReparsePage 

func (s *Store) SessionsForReparsePage(ctx context.Context, agent string, afterID, maxID int64, limit int) ([]ReparseTarget, error)

SessionsForReparsePage returns the next page of reparse targets: the sessions with id in (afterID, maxID], ordered by id, capped at limit. Keyset paging on the primary key keeps each query bounded and lets the reparse loop hold only one page (plus the current session) in memory, rather than the whole corpus. An empty result means the scope is exhausted.

func (*Store) SetReparsedEpoch 

func (s *Store) SetReparsedEpoch(ctx context.Context, epoch int) error

SetReparsedEpoch records that the whole corpus has been reparsed under the given epoch. It is written only after a full (all-agents) reparse completes, so the next startup sees the epochs match and does not reparse again.

func (*Store) Subagents 

func (s *Store) Subagents(ctx context.Context, parentID int64) ([]SessionSummary, error)

Subagents returns sessions whose parent is the given session.

func (*Store) SweepBlobs 

func (s *Store) SweepBlobs(ctx context.Context) (int, error)

SweepBlobs deletes every blob no live row references, unlinking its large object. Liveness is computed, not refcounted, so the sweep is self-healing: it is only needed after a delete or re-parse, the only events that can orphan a blob. It returns the number of blobs removed.

A freshly uploaded body the client has not yet referenced from a transcript is protected by an unexpired pin (see PutBlob): the orphan predicate excludes any blob with a live blob_pins row, so the gap between uploading a body and uploading the transcript that references it cannot lose the body. Expired pins are cleared first so a body whose transcript never arrived is eventually reclaimable.

func (*Store) TokenAuth 

func (s *Store) TokenAuth(ctx context.Context, tokenHash string) (userID int64, scope string, err error)

TokenAuth resolves a presented token hash to its owner and scope, rejecting revoked tokens, and stamps last_used_at.

func (*Store) ToolCalls 

func (s *Store) ToolCalls(ctx context.Context, sessionID int64) ([]ToolCallView, error)

ToolCalls returns all of a session's tool calls as metadata, for the web renderer. Bounded readers pass a message-ordinal range to ToolCallsInRange.

func (*Store) ToolCallsInRange 

func (s *Store) ToolCallsInRange(ctx context.Context, sessionID int64, minOrdinal, maxOrdinal int) ([]ToolCallView, error)

ToolCallsInRange returns the tool calls hanging on messages in the inclusive ordinal window [minOrdinal, maxOrdinal], so a bounded transcript read fetches only the calls for the messages it returned rather than the whole session.

func (*Store) ToolStats 

func (s *Store) ToolStats(ctx context.Context, f AnalyticsFilter) (ToolStats, error)

ToolStats computes the scope's tool volume, reliability, and mix for the Insights page. The deduped tool-call numerator and the turn denominator are separate reads, so it wraps them in one repeatable-read, read-only snapshot: a concurrent projection update between them could otherwise pair TotalCalls from one cohort with Turns from another, making ToolsPerTurn a mixed-snapshot figure. Insights threads its own snapshot through toolStatsFrom; this is the standalone equivalent. The snapshot takes no row locks, so it never blocks ingest.

func (*Store) UnpublishOverview 

func (s *Store) UnpublishOverview(ctx context.Context, userID int64) error

UnpublishOverview hides a user's public overview by clearing the gate flag. The URL is the username and never changes, so re-publishing later brings the same /u/<username> back.

func (*Store) UnpublishSession 

func (s *Store) UnpublishSession(ctx context.Context, sessionID, userID int64) error

UnpublishSession returns a session to internal visibility and clears its public id, so the old link stops resolving rather than merely flipping a flag. It is owner-scoped; a session the user does not own yields ErrNotFound.

func (*Store) UpsertProject 

func (s *Store) UpsertProject(ctx context.Context, remoteKey, host, owner, repo, displayName, kind string) (int64, error)

UpsertProject inserts the project keyed by its remote/synthetic key, or refreshes last_seen on an existing one, returning the project id. The kind is updated on conflict so a standalone folder that is later deleted transitions to orphaned in place (its key, machine + path, is unchanged), and one that gains a remote is never re-resolved here: a remote session carries its own remote key.

func (*Store) UserByID 

func (s *Store) UserByID(ctx context.Context, id int64) (User, error)

UserByID looks up a user by id, including its overview publicity state, which the account page and the overview badge read for the signed-in owner.

func (*Store) UserByUsername 

func (s *Store) UserByUsername(ctx context.Context, username string) (User, error)

UserByUsername looks up a user by username.

func (*Store) VelocityStats 

func (s *Store) VelocityStats(ctx context.Context, f AnalyticsFilter) (VelocityStats, error)

VelocityStats computes the scope's cadence figures for the Insights page. Latency, active time with message count, and tool count are three separate reads over the message timeline, so it wraps them in one repeatable-read, read-only snapshot: a concurrent projection update between them could otherwise combine active minutes from one message timeline with a tool count from another, making ToolsPerActiveMin and the headline cadence internally inconsistent. Insights threads its own snapshot through velocityStatsFrom; this is the standalone equivalent. The snapshot takes no row locks, so it never blocks ingest.

func (*Store) WebSession 

func (s *Store) WebSession(ctx context.Context, id string) (userID int64, err error)

WebSession resolves a session cookie id to its user, rejecting expired ones.

func (*Store) WindowSessionPage 

func (s *Store) WindowSessionPage(ctx context.Context, f SessionFilter) (SessionPage, error)

WindowSessionPage returns the project page's session table: the capped rows that contributed dated usage inside the filter's window, each carrying its in-window token and cost sums rather than its all-time rollup, plus the aggregate of the windowed sessions beyond the cap. It shares the analytics base exactly (the same dated usage_events under the same project, window, agent, user, and machine scope, grouped per session instead of summed whole), which is what lets the rows be a partition of the usage panel's headline where the lifetime rollups ListSessions returns would overcount a session whose usage predates the window.

The cap keeps a project with thousands of windowed sessions from rendering an unbounded table. So the visible rows alone need not sum to the headline; the remainder closes the gap. It is queried, not subtracted from the panel: a boolean OR like cost_incomplete cannot be undone by subtraction (a visible unpriced row would wrongly mark the priced tail incomplete), so the tail is aggregated directly over the hidden sessions, carrying its own per-class sums, cost, and bool_or flag. The remainder query runs only when the cap actually engaged.

func (*Store) WriteBlobPrefixTo 

func (s *Store) WriteBlobPrefixTo(ctx context.Context, w io.Writer, sha256hex string, limit int64) (mediaType string, err error)

WriteBlobPrefixTo streams at most limit bytes of a blob's body to w and returns its media type; limit <= 0 means the whole body. The large-object reader pulls only the bytes it copies, so a small limit transfers a small prefix rather than the whole object: a capped preview of a bulky CAS body is O(limit), not O(blob). A caller that needs to flag truncation compares limit against the stored byte_len from BlobMeta.

func (*Store) WriteBlobTo 

func (s *Store) WriteBlobTo(ctx context.Context, w io.Writer, sha256hex string) (mediaType string, err error)

WriteBlobTo streams a blob's whole body to w and returns its media type. Large object reads must run in a transaction, so the copy happens inside one.

type ToolCallView 

type ToolCallView struct {
	MessageOrdinal  int
	CallIndex       int
	ToolName        string
	Category        string
	FilePath        string
	InputSHA        string
	InputBytes      int64
	InputMediaType  string
	ResultSHA       string
	ResultBytes     int64
	ResultMediaType string
	ResultStatus    string
}

ToolCallView is one tool call rendered as metadata (the body lives in the CAS, fetched on demand by its sha256).

type ToolResultDelta 

type ToolResultDelta struct {
	CallUID    string
	Body       string
	BodySHA256 string
	Bytes      int64
	MediaType  string
	Status     string
}

ToolResultDelta back-patches a tool call's result, matched by call id. The result body reaches the CAS by one of two paths, mirroring ProjToolCall: Body holds it inline for the server to write, or BodySHA256 is the reference the client already uploaded. Both are empty when the result carries no body.

type ToolStat 

type ToolStat struct {
	Name     string
	Calls    int
	Failures int
}

ToolStat is one tool's volume and reliability over a scope: how many times it ran and how many of those runs failed. Calls and Failures are deduped (a replayed transcript re-emits prior tool calls, so the raw rows over-count), the same dedup the per-session signals apply, so the fleet figures reconcile with the session tiles.

func (ToolStat) ErrorRate 

func (t ToolStat) ErrorRate() float64

ErrorRate is the tool's failure share, 0 when it never ran.

type ToolStats 

type ToolStats struct {
	TotalCalls    int
	TotalFailures int
	Turns         int        // human prompts across the cohort (sum of user_message_count)
	Tools         []ToolStat // the busiest tools, most calls first (see maxToolBars)
	Clipped       int        // tools beyond the shown list, folded into the totals but not the bars
}

ToolStats is the fleet tool picture for a scope: the overall call and failure volume, the prompt count the tools-per-turn rate divides by, and the busiest tools with their own reliability. It answers "how much tool work happens, how reliable is it, and which tools carry it" over the same cohort the rest of the Insights page reads.

func (ToolStats) ErrorRate 

func (t ToolStats) ErrorRate() float64

ErrorRate is the fleet failure share across every tool, 0 when nothing ran.

func (ToolStats) HasData 

func (t ToolStats) HasData() bool

HasData reports whether the scope ran any tools, so the panel can show a note rather than an empty bar list for a pure-conversation window.

func (ToolStats) ToolsPerTurn 

func (t ToolStats) ToolsPerTurn() float64

ToolsPerTurn is how many tool calls the fleet ran per human prompt, 0 when the cohort has no prompts (a window of pure automation). The denominator is every prompt in the cohort, so a conversation that needed no tools pulls the rate down rather than sitting outside the average.

type User 

type User struct {
	ID           int64
	Username     string
	PasswordHash string
	IsAdmin      bool
	CreatedAt    time.Time
	// OverviewPublic reports whether this account has published its own usage
	// overview to logged-out viewers at /u/<username>. It is loaded by UserByID
	// (the account page and the overview badge read the owner's own state through
	// it); ListUsers leaves it zero, since the per-user filter only needs
	// identities.
	OverviewPublic bool
}

User is an akari account.

type VelocityStats 

type VelocityStats struct {
	ResponseP50       time.Duration // median prompt-to-first-reply latency, over every turn
	ResponseP90       time.Duration // 90th percentile, the slow-turn tail
	FirstResponseP50  time.Duration // median opening-turn latency (the prompt that pays the context load)
	MsgsPerActiveMin  float64       // messages per active minute (throughput, idle time excluded)
	ToolsPerActiveMin float64       // tool calls per active minute
	ActiveSeconds     float64       // total active time the rates divide by (the dead air between bursts removed)
	Turns             int           // prompt-to-reply pairs measured
	Sessions          int           // sessions that contributed at least one measured turn
}

VelocityStats answers "how fast does work move" over a scope: how long the agent takes to start replying after a prompt (the turn-cycle latency, at the median and the long tail), how slow the opening reply is on its own, and how densely messages and tool calls land during the time the session is actively working. It reads message timestamps, so it sees the same scoped cohort the rest of the Insights page does: sessions that started in the window.

func (VelocityStats) HasData 

func (v VelocityStats) HasData() bool

HasData reports whether the scope carried any measurable cadence, so the panel can show a note rather than a row of dashes and zeros on a window with no timed turns.

func (VelocityStats) HasThroughput 

func (v VelocityStats) HasThroughput() bool

HasThroughput reports whether there was any active time to divide by, so the panel can dash the per-minute rates rather than print a 0.0 that reads as a real measurement when the denominator is undefined (a single-message scope, or one whose every gap exceeds the idle cap).

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.