store

package
v0.27.2 Latest Latest
Warning

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

 Go to latest Published: Jun 25, 2026 License: CC0-1.0 Imports: 17 Imported by: 0

Documentation

Overview

Package store wraps the local SQLite database that holds the canonical session metadata projected from raw agent transcripts. The package owns the open/close lifecycle, applies embedded migrations, and exposes typed helpers for the importer and CLI layers.

Index

Constants

View Source 
const (
	SnippetMarkStart = "«"
	SnippetMarkEnd   = "»"
)

SnippetMarkStart and SnippetMarkEnd wrap matched terms in the snippet text. The CLI render layer recognizes them and applies Lipgloss styling.

View Source 
const MatchFieldTurnContent = "turn.content"

MatchFieldTurnContent is the only value SearchHit.MatchField takes today. Held as a constant so renderers don't sprinkle the literal across packages.

Variables

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

ErrSessionNotFound is returned by helpers that want a typed miss; thin alias so callers don't import database/sql just to compare.

View Source 
var ErrStoreNeedsMigration = errors.New("prosa store needs migration; run `prosa sync` or another write command first")

ErrStoreNeedsMigration is returned by OpenReadOnly when the embedded migrations include a version higher than anything recorded in the on-disk schema_migrations table — read-only callers can't apply migrations themselves.

View Source 
var ErrStoreNotInitialized = errors.New("prosa store not initialized; run `prosa sync` first")

ErrStoreNotInitialized is returned by OpenReadOnly when the database file does not exist. Callers should surface the message verbatim so the user knows to run `prosa sync` first.

Functions

func FormatDuration 

func FormatDuration(d time.Duration) string

FormatDuration is exposed for tests / renderers.

Types

type AnalyticsResult 

type AnalyticsResult struct {
	Headers []string
	Rows    []AnalyticsRow
}

AnalyticsResult bundles headers + rows so callers can render uniformly without per-report dispatch.

type AnalyticsRow 

type AnalyticsRow struct {
	Values []any
}

AnalyticsRow is the row emitted by every Analytics<X> query — a generic ordered slice of values keyed by Headers. The renderer (or the --json emitter) prints them as is.

type BoilerplateCandidate 

type BoilerplateCandidate struct {
	ID      string
	RawPath string
}

BoilerplateCandidate is one row returned by ListSessionsWithBoilerplatePrompt: the bits the denoise pass needs to reopen the raw and update the row.

type Device 

type Device struct {
	ID              string
	Hostname        string
	MachineID       string
	FriendlyName    string
	FingerprintedAt time.Time
}

Device is the row stored in the devices table. ID is the stable per-machine fingerprint (see internal/device).

type ManifestRow 

type ManifestRow struct {
	ID      string
	RawHash string
	RawPath string
}

ManifestRow is the minimal projection used by the catch-up reconcile path: enough to ask "does the server already have this id with this hash?" and locate the raw on disk if we need to re-push it.

type ProfileCount 

type ProfileCount struct {
	Agent   string
	Profile string
	Count   int
}

ProfileCount is one (agent, profile) group with its session count.

type SearchHit 

type SearchHit struct {
	Session session.Session
	Snippet string
	Role    string // "user" | "assistant" | "tool"
	// TurnID is the matching turn's primary key (turns.id).
	TurnID int64
	// TurnTS is the timestamp of the matching turn.
	TurnTS time.Time
	// Kind mirrors Turn.Kind for the matched turn ("message" |
	// "tool_result" | "operational"). Empty when older rows.
	Kind string
	// ToolName carries Turn.ToolName when the matched turn is a tool
	// projection; empty otherwise.
	ToolName string
	// MatchField names the document field that produced the match.
	// Currently always MatchFieldTurnContent; reserved for future
	// session-level matches (first_prompt, project name).
	MatchField string
	// Rank is SQLite FTS5's bm25() score. Lower means more relevant;
	// rows arrive sorted ascending.
	Rank float64
}

SearchHit is the per-session result of a Search call: the session metadata plus the highest-ranked snippet from any of its turns and the metadata needed to fetch the exact evidence without re-reading the raw transcript.

type SessionFilter 

type SessionFilter struct {
	Since, Until time.Time
	ProjectExact *string // exact match on sessions.project_path
	// ProjectMatch is the substring filter from --project. It matches when
	// any of project_path / project_remote / project_marker contains the
	// value as a substring, so the leading wildcard can defeat the project
	// indexes. Prefer the exact fields when the caller has a full path,
	// remote, or marker.
	ProjectMatch *string
	// ProjectRemote matches sessions.project_remote exactly. Used by
	// the git-remote-anchored auto-detect (INTENT §5 step 1).
	ProjectRemote *string
	// ProjectMarker matches sessions.project_marker exactly. Used by
	// the .prosa.yaml-anchored auto-detect (INTENT §5 step 2).
	ProjectMarker *string
	Agent         *string
	DeviceName    *string
	// Profile matches sessions.profile exactly. Drives the --profile filter.
	Profile *string
	// Kinds restricts results to sessions carrying at least one of the
	// listed session_kinds (OR semantics). Empty means no kind filter.
	Kinds []string
	// Limit caps the number of rows returned. 0 means no limit.
	Limit int
}

SessionFilter narrows ListSessions and Search. Since/Until are required; the pointer fields are optional and combine with AND semantics. ProjectExact is the cwd-anchored auto-filter (exact equality on project_path); ProjectMatch is substring (used by the --project flag) and ORs across project_path / project_remote / project_marker so `--project movaincentivo` finds sessions stored under any of the three columns. Agent matches the canonical agent string ("claude-code" | "codex"). DeviceName matches against sessions.device_id or devices.friendly_name. Limit > 0 caps the returned rows.

type Store 

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

Store is a thin handle around the underlying database. It is safe for concurrent use; modernc.org/sqlite serializes writes internally.

func Open 

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

Open creates the parent directory if needed, opens the SQLite file with WAL + foreign-keys + synchronous=NORMAL + a 5s busy_timeout, and applies any pending migrations before returning a ready Store. Use this for any code path that may write — sync, import, denoise.

func OpenReadOnly 

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

OpenReadOnly opens the SQLite file in mode=ro for the read paths used by timeline, search, show, and analytics commands. It never creates directories, never runs migrations, and never enables WAL — so it can run safely while sync holds the writer connection. busy_timeout=5s makes it ride out brief writer contention; the bounded pool keeps a single process from saturating SQLite's internal reader serialization.

Returns ErrStoreNotInitialized when path does not exist, and ErrStoreNeedsMigration when the embedded schema is newer than what is recorded on disk — both messages are user-facing.

func (*Store) AnalyticsErrors 

func (s *Store) AnalyticsErrors(ctx context.Context, f SessionFilter) (AnalyticsResult, error)

AnalyticsErrors: sessions whose assistant turns match common error signals. Heuristic — flagged in --help.

func (*Store) AnalyticsErrorsByModel 

func (s *Store) AnalyticsErrorsByModel(ctx context.Context, f SessionFilter) (AnalyticsResult, error)

AnalyticsErrorsByModel counts the sessions flagged by the errorTriggers FTS heuristic, grouped by model. Unlike AnalyticsErrors (a recent-rows list capped at 30) this is the full aggregate, so the sum across rows is the true flagged-session count an error-rate indicator needs. Heuristic, same caveat as AnalyticsErrors.

func (*Store) AnalyticsHeatmap 

func (s *Store) AnalyticsHeatmap(ctx context.Context, f SessionFilter) (AnalyticsResult, error)

AnalyticsHeatmap emits one row per (day, agent) over the selected window, matching the server's heatmap report shape exactly (DATE, AGENT, SESSIONS). Zero-session days still get a single (day, "", 0) row so callers can render a stable GitHub-style contribution graph with correct calendar positions. The CLI rolls these per-agent rows up to per-day totals for its table; the panel uses the per-agent breakdown.

func (*Store) AnalyticsHours 

func (s *Store) AnalyticsHours(ctx context.Context, f SessionFilter) (AnalyticsResult, error)

AnalyticsHours buckets sessions by their UTC start-hour ("00".."23") for a "when do I work" view. The hour is read straight off the RFC3339Nano started_at text — substr is cheaper than a date parse and mirrors the substr(started_at, 1, 10) day idiom AnalyticsHeatmap uses. The report is canonically UTC (like the heatmap); callers wanting a local-time view rotate the buckets after the fact.

func (*Store) AnalyticsModels 

func (s *Store) AnalyticsModels(ctx context.Context, f SessionFilter) (AnalyticsResult, error)

func (*Store) AnalyticsProfiles 

func (s *Store) AnalyticsProfiles(ctx context.Context, f SessionFilter) (AnalyticsResult, error)

AnalyticsProfiles breaks sessions down by device, agent, and profile. The dv alias avoids colliding with the devices join analyticsQuery adds for --device.

func (*Store) AnalyticsProjects 

func (s *Store) AnalyticsProjects(ctx context.Context, f SessionFilter) (AnalyticsResult, error)

AnalyticsProjects emits all identity-aware project/agent buckets by the best-available project key. Callers that need a top-N view should cap after any project-level rollup so one noisy agent bucket cannot hide a project.

func (*Store) AnalyticsSessions 

func (s *Store) AnalyticsSessions(ctx context.Context, f SessionFilter) (AnalyticsResult, error)

func (*Store) AnalyticsSubagents 

func (s *Store) AnalyticsSubagents(ctx context.Context, f SessionFilter) (AnalyticsResult, error)

AnalyticsSubagents aggregates subagent fan-out per parent agent, mirroring the server's subagents report. Filters apply to the children (the spawned sessions); grouping is by the parent's agent. The selectSQL ends inside the inner subquery so analyticsQuery's devices join for --device lands against the child alias s.

func (*Store) AnalyticsTools 

func (s *Store) AnalyticsTools(ctx context.Context, f SessionFilter) (AnalyticsResult, error)

func (*Store) AnalyticsUsage 

func (s *Store) AnalyticsUsage(ctx context.Context, f SessionFilter) (AnalyticsResult, error)

AnalyticsUsage aggregates measured token consumption by agent and adds an estimated USD cost where the embedded pricing table recognizes the model.

func (*Store) AnalyticsUsageByModel 

func (s *Store) AnalyticsUsageByModel(ctx context.Context, f SessionFilter) (AnalyticsResult, error)

AnalyticsUsageByModel mirrors AnalyticsUsage but groups by model instead of agent, so the panel can rank token spend per model and draw a cost donut. Each group shares one model, so cost is a straight pricing.CostUSD over the summed usage; models the table doesn't recognize emit an empty EST_COST_USD.

func (*Store) Close 

func (s *Store) Close() error

Close releases the underlying database handle.

func (*Store) DB 

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

DB exposes the raw *sql.DB for ad-hoc queries (e.g. tests and CLI verification commands). Avoid using it from package internals — prefer adding a typed method.

func (*Store) DistinctProjectPaths 

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

DistinctProjectPaths returns every non-null project_path stored. Used by the CLI to drive auto-detection of the current project from cwd.

func (*Store) DistinctProjectPathsNeedingIdentity 

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

DistinctProjectPathsNeedingIdentity returns every distinct cwd (sessions.project_path) that has at least one row missing BOTH project_remote AND project_marker. The CLI uses this to drive a one-shot resolution pass on first sync after migration 0002.

Empty paths and NULLs are excluded. Order is stable for deterministic tests but not load-bearing.

func (*Store) FillProjectIdentity 

func (s *Store) FillProjectIdentity(ctx context.Context, path, remote, marker string) (int64, int64, error)

FillProjectIdentity sets project_remote and/or project_marker on every session row whose project_path equals path AND whose target column is still NULL. Either argument may be empty — empty values are skipped so the caller doesn't have to special-case partial matches. Returns the pair (remoteRowsUpdated, markerRowsUpdated).

func (*Store) GetSession 

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

GetSession returns a single session by id, or sql.ErrNoRows if missing.

func (*Store) GetSessionTools 

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

GetSessionTools returns the (name, count) tool-usage rows for sessionID.

func (*Store) GetTurns 

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

GetTurns returns every turn for sessionID in insertion (ts) order. Used by the push pipeline to mirror the local store onto the server, and by show to render the human view.

func (*Store) InsertTurns 

func (s *Store) InsertTurns(ctx context.Context, sessionID string, turns []session.Turn) error

InsertTurns replaces the turn set for a session in one transaction. Old rows are deleted first to keep re-imports idempotent. The FTS5 virtual table is kept in sync by the AFTER INSERT / AFTER DELETE triggers defined in the migration; kind and tool_name live on the base table and are joined into search results when needed.

Empty Turn.Kind defaults to "message" so importers that haven't learned the new shape yet still insert valid rows.

func (*Store) LastHash 

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

LastHash returns the most recently recorded raw hash for a session, or (_, false, nil) if the session has never been recorded. Callers compare the freshly computed hash to short-circuit reimports of unchanged files.

func (*Store) LastImportSkip 

func (s *Store) LastImportSkip(ctx context.Context, sessionID, reason string) (string, bool, error)

LastImportSkip returns a remembered policy skip hash, if it is still valid for the current projection version.

func (*Store) ListChildren 

func (s *Store) ListChildren(ctx context.Context, parentID string) ([]session.Session, error)

ListChildren returns every session whose parent_session_id matches parentID, ordered started_at ascending so the panel can show them in the order they were spawned. Empty parentID returns an empty slice (callers shouldn't ask for "children of nothing").

func (*Store) ListDevices 

func (s *Store) ListDevices(ctx context.Context) ([]Device, error)

ListDevices returns every device row known to the store, ordered by most-recently fingerprinted first (with NULL — the seed row — last).

func (*Store) ListDevicesMap 

func (s *Store) ListDevicesMap(ctx context.Context) (map[string]string, error)

ListDevicesMap returns id → friendly_name for every device row, usable as a lookup table during render so the timeline can show human-readable device names instead of the raw fingerprint hex.

func (*Store) ListSessions 

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

ListSessions runs the configurable session query. It assembles the WHERE clause from the populated filter fields and returns sessions ordered newest first. Empty result is not an error.

func (*Store) ListSessionsByRange 

func (s *Store) ListSessionsByRange(ctx context.Context, since, until time.Time) ([]session.Session, error)

ListSessionsByRange is a thin convenience wrapper preserving the cut-1 signature for callers that don't need the additional filter dimensions.

func (*Store) ListSessionsManifest 

func (s *Store) ListSessionsManifest(ctx context.Context, deviceID, afterID string, limit int) ([]ManifestRow, error)

ListSessionsManifest paginates every session for a given device by id ASC. afterID = "" starts the scan; limit <= 0 means "all rows in one page" (used by the CLI which holds the whole local set in memory during reconcile). The composite (device_id, id) index from migration 0003 makes this an index-only walk.

func (*Store) ListSessionsWithBoilerplatePrompt 

func (s *Store) ListSessionsWithBoilerplatePrompt(ctx context.Context, limit int) ([]BoilerplateCandidate, error)

ListSessionsWithBoilerplatePrompt returns rows whose stored first_prompt starts with one of the known agent-injected meta prefixes. Used by `prosa sync` to one-shot denoise legacy data without forcing a full reimport.

The pattern list is sourced from internal/sessiontext.Prefixes so adding a new wrapper in one place automatically extends both the importer-time classifier and the SQL denoise sweep — no silent drift between Go and SQL anymore.

func (*Store) ProfileCounts 

func (s *Store) ProfileCounts(ctx context.Context) ([]ProfileCount, error)

ProfileCounts returns the session count per (agent, profile).

func (*Store) ProjectMarkerExists 

func (s *Store) ProjectMarkerExists(ctx context.Context, name string) (bool, error)

ProjectMarkerExists reports whether at least one session has a project_marker equal to name. Same role as ProjectRemoteExists but for the .prosa.yaml-anchored identity.

func (*Store) ProjectRemoteExists 

func (s *Store) ProjectRemoteExists(ctx context.Context, url string) (bool, error)

ProjectRemoteExists reports whether at least one session has a project_remote equal to url. Used by DetectProject to confirm the store already knows the current cwd's remote before asking the timeline to scope by it.

func (*Store) RebindLocalSessions 

func (s *Store) RebindLocalSessions(ctx context.Context, fingerprint string) (int64, error)

RebindLocalSessions reassigns every `device_id = 'local'` session row to the given fingerprint, in one transaction. Returns the count rewritten. Used during startup to migrate sessions imported under the seed device id to the real per-machine fingerprint.

No-op when fingerprint == "local" (defensive: avoids a self-rewrite if the resolver ever returns the seed value).

func (*Store) RecordImportSkip 

func (s *Store) RecordImportSkip(ctx context.Context, sessionID, hash, reason string) error

RecordImportSkip remembers that a source file was parsed and intentionally not projected into sessions.

func (*Store) RecordSync 

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

RecordSync upserts the hash + timestamp the importer just observed. Foreign key on session_id means the parent session row must already exist before this is called.

func (*Store) RefreshOrchestratorKinds 

func (s *Store) RefreshOrchestratorKinds(ctx context.Context) error

RefreshOrchestratorKinds reconciles the edge-dependent "orchestrator" kind across the whole store. A session is an orchestrator when at least one other session names it as parent_session_id. This cannot be decided while projecting a single session (the parent may be imported before or after its children), so the import sweep calls this once after all importers complete. Idempotent: it adds the kind to every current parent and removes it from sessions that no longer have children.

func (*Store) Search 

func (s *Store) Search(ctx context.Context, query string, f SessionFilter, limit int) ([]SearchHit, error)

Search runs an FTS5 MATCH query against turns_fts and returns at most `limit` hits, deduplicated by session (highest-ranked turn wins). The SessionFilter reuses the same filter semantics as ListSessions so `prosa search` honors --project / --agent / --device / --last.

func (*Store) UpdateFirstPrompt 

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

UpdateFirstPrompt rewrites just the first_prompt column for a session. Used by the denoise pass; everything else stays untouched.

func (*Store) UpsertDevice 

func (s *Store) UpsertDevice(ctx context.Context, d Device) error

UpsertDevice writes (or updates) a device row. Existing FriendlyName is preserved when the caller passes an empty value, because the `prosa devices rename` command edits that field independent of sync.

func (*Store) UpsertSession 

func (s *Store) UpsertSession(ctx context.Context, sess session.Session, tools []session.ToolUsage) error

UpsertSession writes (or replaces) a session row plus its normalized session_tools rows in a single transaction. Existing session_tools rows for the same session id are deleted first so the post-condition matches the input slice exactly.

func (*Store) WriteSession 

func (s *Store) WriteSession(
	ctx context.Context,
	sess session.Session,
	tools []session.ToolUsage,
	turns []session.Turn,
	hash string,
) error

WriteSession persists a complete projection — session row + session_usage + session_tools, turns, and the sync_state hash — in a single transaction. Importers call this instead of UpsertSession + InsertTurns + RecordSync so a crash mid-write can never leave a session row visible without its turns or with a stale sync_state.

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.