postgres

func (*Driver) AggregateSpanStats ¶ added in v0.16.0

func (d *Driver) AggregateSpanStats(ctx context.Context, orgID string, since, until *time.Time) (storage.SpanStats, error)

AggregateSpanStats sums the trace-grain rollups over a time window — the span-layer aggregate behind /v1/stats. Implements storage.SpanStatsReader.

func (*Driver) Ancestry ¶ added in v0.5.0

func (d *Driver) Ancestry(ctx context.Context, hash string) ([]*merkle.Node, error)

func (*Driver) AncestryChain ¶ added in v0.5.0

func (d *Driver) AncestryChain(ctx context.Context, hash string) (*storage.Chain, error)

func (*Driver) AncestryChains ¶ added in v0.5.0

func (d *Driver) AncestryChains(ctx context.Context, hashes []string) (map[string]*storage.Chain, error)

func (*Driver) BackfillSession ¶ added in v0.10.0

func (d *Driver) BackfillSession(ctx context.Context, req storage.SessionBackfillRequest) (storage.SessionBackfillResult, error)

BackfillSession links existing legacy nodes to a session table row. It is used by transcript backfills where the node DAG already exists but the original ingest path predated session envelopes.

func (*Driver) BackfillSessionStatus ¶ added in v0.13.0

func (d *Driver) BackfillSessionStatus(ctx context.Context) (storage.BackfillSessionStatusResult, error)

BackfillSessionStatus recomputes derived_status (plus the sticky has_git_activity flag and tool_result_count / tool_error_count) for sessions still at the default 'unknown' status — the rows that predate the ingest-time computation. It walks each session's nodes with the same signal helpers ingest uses (sessions.CountToolResults / CountToolResultErrors / BlocksHaveGitActivity over every node, sessions.DetermineStatus over the chronologically-last node as the leaf). Live ingest keeps status current on its own.

Scoping to 'unknown' keeps re-runs cheap and idempotent — already-classified rows are skipped. Re-classifying already-decided rows after a classifier change is intentionally out of scope for this endpoint.

Safe to run online: a concurrent live turn re-runs the same UpdateSessionStatus path, so the worst case is a redundant equal write. Each session's recompute is its own statement; there is no global lock.

func (*Driver) ClearDeriveDirty ¶ added in v0.16.0

func (d *Driver) ClearDeriveDirty(ctx context.Context, e storage.DeriveQueueEntry) (bool, error)

ClearDeriveDirty implements storage.DeriveQueue. The DELETE is guarded on dirtied_at equality so a raw turn landing mid-derive (which bumps dirtied_at) keeps the session queued.

func (*Driver) Close ¶ added in v0.5.0

func (d *Driver) Close() error

func (*Driver) CountRawTurns ¶ added in v0.16.0

func (d *Driver) CountRawTurns(ctx context.Context) (int64, error)

CountRawTurns implements storage.RawTurnStore.

func (*Driver) CountSessions ¶ added in v0.5.0

func (d *Driver) CountSessions(ctx context.Context, opts storage.ListOpts) (storage.SessionStats, error)

func (*Driver) CountSkills ¶ added in v0.20.0

func (d *Driver) CountSkills(ctx context.Context, orgID, query, author string) (storage.SkillCounts, error)

CountSkills returns the per-tab totals for a search (ignoring scope/cursor): every matching skill and how many the caller authored.

func (*Driver) CreateSkillVersion ¶ added in v0.20.0

func (d *Driver) CreateSkillVersion(ctx context.Context, orgID string, rec storage.SkillVersionRecord) (*storage.SkillVersionRecord, error)

CreateSkillVersion appends an immutable published snapshot and returns it.

func (*Driver) DB ¶ added in v0.5.0

func (d *Driver) DB() *pgxpool.Pool

func (*Driver) DeleteSession ¶ added in v0.22.0

func (d *Driver) DeleteSession(ctx context.Context, orgID, id string) (bool, error)

DeleteSession removes a session by its org-scoped id and returns whether a row was actually deleted (false when the id was absent). The session_id ON DELETE CASCADE foreign keys tear down the rest of the subtree in the same statement: subagent child sessions (parent_session_id), the session's derived nodes, and its spans/span_turns/span_links. A malformed id is treated as a no-op delete, matching DeleteSkill.

func (*Driver) DeleteSkill ¶ added in v0.20.0

func (d *Driver) DeleteSkill(ctx context.Context, orgID, id string) (bool, error)

DeleteSkill removes a skill and its published history by id. Returns whether a skill row was actually deleted (false when the id was already absent).

func (*Driver) Depth ¶ added in v0.5.0

func (d *Driver) Depth(ctx context.Context, hash string) (int, error)

func (*Driver) DeriveQueueStats ¶ added in v0.16.0

func (d *Driver) DeriveQueueStats(ctx context.Context) (storage.DeriveQueueStats, error)

DeriveQueueStats implements storage.DeriveQueue.

func (*Driver) Get ¶ added in v0.5.0

func (d *Driver) Get(ctx context.Context, hash string) (*merkle.Node, error)

func (*Driver) GetByParent ¶ added in v0.5.0

func (d *Driver) GetByParent(ctx context.Context, parentHash *string) ([]*merkle.Node, error)

func (*Driver) GetDeriveDirty ¶ added in v0.16.0

func (d *Driver) GetDeriveDirty(ctx context.Context, orgID, harnessID, harnessSessionID string) (*storage.DeriveQueueEntry, error)

GetDeriveDirty implements storage.DeriveQueue.

func (*Driver) GetSessionRecord ¶ added in v0.12.0

func (d *Driver) GetSessionRecord(ctx context.Context, orgID, id string) (*storage.SessionRecord, error)

GetSessionRecord returns a single session by its UUID, or nil if not found.

func (*Driver) GetSessionRecordByHarness ¶ added in v0.15.0

func (d *Driver) GetSessionRecordByHarness(
	ctx context.Context,
	orgID string,
	harnessID string,
	harnessSessionID string,
) (*storage.SessionRecord, error)

GetSessionRecordByHarness returns the single session matching the org-scoped natural key (org_id, harness_id, harness_session_id), or nil if no row matches. The lookup is an exact-match point read on the sessions_harness_uq unique index, mirroring the GetSessionRecord nil-on-no-rows contract.

func (*Driver) GetSkill ¶ added in v0.20.0

func (d *Driver) GetSkill(ctx context.Context, orgID, id string) (*storage.SkillRecord, error)

GetSkill returns a single skill by its org-scoped id, or nil if not found.

func (*Driver) GetSpanRecord ¶ added in v0.16.0

func (d *Driver) GetSpanRecord(ctx context.Context, orgID, traceID, spanID string) (*storage.SpanRecord, error)

GetSpanRecord returns one span with full payloads. Implements storage.SpanModelReader.

func (*Driver) GetTraceDetail ¶ added in v0.16.0

func (d *Driver) GetTraceDetail(ctx context.Context, orgID, traceID string) (*storage.SpanTurnRecord, []storage.SpanRecord, []storage.SpanLinkRecord, error)

GetTraceDetail returns one turn with its spans and links. Implements storage.SpanModelReader.

func (*Driver) Has ¶ added in v0.5.0

func (d *Driver) Has(ctx context.Context, hash string) (bool, error)

func (*Driver) IncrementSkillDownloads ¶ added in v0.20.0

func (d *Driver) IncrementSkillDownloads(ctx context.Context, orgID, id string) error

IncrementSkillDownloads bumps the real download counter for a skill.

func (*Driver) IngestTurn ¶ added in v0.10.0

func (d *Driver) IngestTurn(ctx context.Context, req storage.IngestTurnRequest) (storage.IngestTurnResult, error)

IngestTurn implements storage.SessionIngester for the Postgres driver. The session-tracking flow runs in a single transaction: resolve / UPSERT a sessions row, resolve the optional fork-parent FK (placeholder-inserting the parent when its own first turn hasn't landed yet), insert every node in the supplied chain, stamp session_id onto each newly-inserted node, and roll up the per-turn counters.

func (*Driver) Leaves ¶ added in v0.5.0

func (d *Driver) Leaves(ctx context.Context) ([]*merkle.Node, error)

func (*Driver) List ¶ added in v0.5.0

func (d *Driver) List(ctx context.Context) ([]*merkle.Node, error)

func (*Driver) ListDeriveDirty ¶ added in v0.16.0

func (d *Driver) ListDeriveDirty(ctx context.Context, dirtiedBefore, firstDirtiedBefore time.Time, limit int32) ([]storage.DeriveQueueEntry, error)

ListDeriveDirty implements storage.DeriveQueue.

func (*Driver) ListParentRefs ¶ added in v0.5.0

func (d *Driver) ListParentRefs(ctx context.Context) ([]storage.ParentRef, error)

func (*Driver) ListRawTurnHeaders ¶ added in v0.16.0

func (d *Driver) ListRawTurnHeaders(ctx context.Context, orgID, harnessID, harnessSessionID string) ([]storage.RawTurnHeader, error)

ListRawTurnHeaders returns the wire log for one session: capture identity and payload sizes, no blobs. Implements storage.SpanModelReader.

func (*Driver) ListRawTurns ¶ added in v0.16.0

func (d *Driver) ListRawTurns(ctx context.Context, afterID int64, pageSize int32) ([]storage.RawTurnRecord, error)

ListRawTurns implements storage.RawTurnStore.

func (*Driver) ListSessionRecords ¶ added in v0.12.0

func (d *Driver) ListSessionRecords(
	ctx context.Context,
	orgID string,
	opts storage.SessionListOpts,
) ([]storage.SessionRecord, error)

ListSessionRecords returns a page of sessions for an org ordered by the requested sort column (default last_seen_at DESC), optionally windowed by activity (last_seen_at) and narrowed to one gateway-stamped JWT subject (exact match on the indexed column; empty lists every user's sessions). Pass zero-value opts to start from the beginning, unwindowed and unfiltered.

func (*Driver) ListSessionSpanModel ¶ added in v0.16.0

func (d *Driver) ListSessionSpanModel(ctx context.Context, sessionID string) ([]storage.SpanTurnRecord, []storage.SpanRecord, []storage.SpanLinkRecord, error)

ListSessionSpanModel returns the stored span projection for one session: turns, spans, and links, each in stable presentation order. Implements storage.SpanModelReader.

func (*Driver) ListSessions ¶ added in v0.5.0

func (d *Driver) ListSessions(ctx context.Context, opts storage.ListOpts) (*storage.Page[*merkle.Node], error)

func (*Driver) ListSkillVersions ¶ added in v0.20.0

func (d *Driver) ListSkillVersions(ctx context.Context, orgID, skillID string) ([]storage.SkillVersionRecord, error)

ListSkillVersions returns a skill's published history, newest first.

func (*Driver) ListSkills ¶ added in v0.20.0

func (d *Driver) ListSkills(ctx context.Context, orgID string, opts storage.SkillListOpts) ([]storage.SkillRecord, error)

ListSkills returns one keyset page of skills for an org, honoring the optional search/scope filters, the requested sort, and the cursor in opts.

func (*Driver) ListSkillsBySession ¶ added in v0.20.0

func (d *Driver) ListSkillsBySession(ctx context.Context, orgID, sessionID string) ([]storage.SkillRecord, error)

ListSkillsBySession returns the skills generated from a given session (reverse lookup over the provenance array), newest-edited first.

func (*Driver) ListTraceSummaries ¶ added in v0.16.0

func (d *Driver) ListTraceSummaries(ctx context.Context, sessionID string) ([]storage.TraceSummaryRecord, error)

ListTraceSummaries returns a session's turn headers with span counts — the lazy session-detail rows. Implements storage.SpanModelReader.

func (*Driver) LoadDag ¶ added in v0.5.0

func (d *Driver) LoadDag(ctx context.Context, hash string) (*merkle.Dag, error)

func (*Driver) MarkDeriveDirty ¶ added in v0.16.0

func (d *Driver) MarkDeriveDirty(ctx context.Context, orgID, harnessID, harnessSessionID string) error

MarkDeriveDirty implements storage.DeriveQueue.

func (*Driver) NextSkillVersionNumber ¶ added in v0.20.0

func (d *Driver) NextSkillVersionNumber(ctx context.Context, orgID, skillID string) (int, error)

NextSkillVersionNumber returns the next monotonic version number for a skill (1 when nothing is published yet).

func (*Driver) Open ¶ added in v0.5.0

func (d *Driver) Open(ctx context.Context) error

func (*Driver) Put ¶ added in v0.5.0

func (d *Driver) Put(ctx context.Context, n *merkle.Node) (bool, error)

func (*Driver) PutRawTurn ¶ added in v0.16.0

func (d *Driver) PutRawTurn(ctx context.Context, rec storage.RawTurnRecord) (bool, error)

PutRawTurn implements storage.RawTurnStore. The row is appended verbatim; a retried POST with the same (org_id, request_id) is a no-op per the partial unique index.

Session-keyed rows also mark the session dirty in derive_queue, in the same transaction, so the derive worker picks the session up. Marking happens even when the row deduped: a re-POST of an existing turn is the natural "re-derive this session" signal, and a redundant mark only costs one idempotent derive.

func (*Driver) RederiveFromRaw ¶ added in v0.16.0

func (d *Driver) RederiveFromRaw(ctx context.Context, project string) (map[string]*derive.RederiveReport, error)

RederiveFromRaw rebuilds the derived node layer from the immutable raw-turn store: every raw turn is re-parsed, re-classified, and re-chained with the CURRENT projection, then written back in one transaction per org — upserting nodes (refreshing derived columns on rows that already exist) and pruning rows in covered sessions whose hashes the current derivation no longer produces (e.g. chains built under a superseded projection).

Memory discipline: the pass first scans a payload-free index (id + capture time), sorts per org chronologically, then streams full rows ONE AT A TIME through the deriver. Each turn's chain re-contains the whole conversation history, so holding all turns at once is O(N²) in content and OOMs a modestly-sized container; the deriver retains only the deduplicated (unique-content) node set.

The pass is idempotent: re-running against an unchanged raw layer upserts the same set and prunes nothing. Raw rows are never written.

func (*Driver) RederiveSession ¶ added in v0.16.0

func (d *Driver) RederiveSession(ctx context.Context, project, orgID, harnessID, harnessSessionID string) (*derive.RederiveReport, error)

RederiveSession is the session-scoped sibling of RederiveFromRaw: re-derive ONE harness session from its raw turns and apply the result transactionally (upsert + prune scoped to that session). This is the derive worker's unit of work — memory stays bounded by one session's unique content, and the full rows stream through the deriver one at a time exactly like the full-org pass.

Same idempotence contract: re-running against an unchanged raw layer upserts the same set and prunes nothing.

func (*Driver) Roots ¶ added in v0.5.0

func (d *Driver) Roots(ctx context.Context) ([]*merkle.Node, error)

func (*Driver) SessionIdentityByHash ¶ added in v0.10.0

func (d *Driver) SessionIdentityByHash(ctx context.Context, orgID, hash string) (*storage.SessionIdentity, error)

SessionIdentityByHash returns the harness identity for the sessions row attached to a node hash, scoped to orgID. Older rows, legacy Put writes, and non-Postgres stores may not have session tracking metadata; those return nil without error so read APIs can expose the field opportunistically.

The org_id predicate is load-bearing, not cosmetic: nodes is keyed by the composite (org_id, hash), so identical content yields one row per org. A lookup on hash alone would return an arbitrary org's row for shared content, leaking another tenant's harness_session_id. An empty orgID maps to the nil-UUID sentinel bucket (legacy / non-session writes).

func (*Driver) SetSkillVersion ¶ added in v0.20.0

func (d *Driver) SetSkillVersion(ctx context.Context, orgID, skillID, semver string, updatedAt time.Time) error

SetSkillVersion bumps a skill's current published semver and updated_at.

func (*Driver) SweepDeriveDirty ¶ added in v0.16.0

func (d *Driver) SweepDeriveDirty(ctx context.Context, activeSince time.Time) (int64, error)

SweepDeriveDirty implements storage.DeriveQueue.

func (*Driver) TryDeriveSessionLock ¶ added in v0.16.0

func (d *Driver) TryDeriveSessionLock(ctx context.Context, orgID, harnessID, harnessSessionID string) (release func(), acquired bool, err error)

TryDeriveSessionLock takes a session-scoped Postgres advisory lock so concurrent workers never double-derive one session. The lock is connection-scoped: the pooled connection is pinned for the lock's lifetime and returned by release. A false return with nil error means another holder has the session — skip it, this is not a failure.

The key hashes the harness triple to 64 bits (FNV-1a). A cross-triple collision only serializes two unrelated sessions' derives — safe, merely slower — so probabilistic keying is fine here.

func (*Driver) UpdateUsage ¶ added in v0.5.0

func (d *Driver) UpdateUsage(ctx context.Context, hash string, usage *llm.Usage) error

func (*Driver) UpsertSkill ¶ added in v0.20.0

func (d *Driver) UpsertSkill(ctx context.Context, orgID string, rec storage.SkillRecord) (*storage.SkillRecord, error)

UpsertSkill inserts or replaces a skill keyed by (org_id, id) and returns the persisted record. Create/generate/duplicate pass a freshly minted id (a plain insert); PUT/publish pass the existing id (an update). created_at is preserved.