postgres

func (*Driver) AcquireDeriveSessionLock ¶ added in v0.25.0

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

AcquireDeriveSessionLock is the BLOCKING sibling of TryDeriveSessionLock: it waits until the per-session advisory lock is held rather than returning acquired=false. This is the right semantics for a manual/one-off re-derive that must serialize BEHIND the derive worker (finish deriving after the worker releases) instead of skipping. The worker itself keeps using the non-blocking probe, so it never waits on a manual re-derive — only the reverse — which keeps the two free of a lock cycle.

Like TryDeriveSessionLock the lock is connection-scoped: the pooled connection is pinned until release unlocks it (on a fresh background context, since the caller's ctx may already be canceled).

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) 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) 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) DeriveQueueStats ¶ added in v0.16.0

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

DeriveQueueStats implements storage.DeriveQueue.

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) 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 (keyed by the envelope's natural key or a synthetic harness_session_id from the turn's Merkle root), resolve the optional fork-parent FK (placeholder-inserting the parent when its own first turn hasn't landed yet), and fold the derived title. It writes session IDENTITY only: nodes are not persisted (the merkle layer is in-memory), and every derived rollup — counters, model_usage, tasks, kind_counts, and the chain-aware status/git/tool signals — is owned by the derive-time span fold.

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) 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) ListSessionLinks ¶ added in v0.25.0

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

ListSessionLinks returns a session's dataflow links alone, in the same deterministic key order ListSessionSpanModel serves them. It is the payload-free half of that read, for the per-trace streaming export. Implements storage.SpanModelReader.

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 (a turn started in the window, matching /v1/stats) 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) 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) ListTraceSpans ¶ added in v0.25.0

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

ListTraceSpans returns one trace's spans in presentation order (seq ASC, matching ListSessionSpanModel restricted to the trace) — the same per-trace read GetTraceDetail performs, without the turn/link round-trips. Implements storage.SpanModelReader.

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) 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) 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.

CONCURRENCY (PCC-687 G, partial): unlike RederiveSessionLocked, this whole-org pass does NOT hold the per-session derive lock, so running it while the derive worker is live can still race — this pass reads the whole org's raw UP FRONT, and a turn the worker writes after that read but before this pass's prune is deleted as "not in my set". A write-only lock does not fix this: the stale READ is the defect, so a correct fix must read-and-write each session under its lock — i.e. reimplement this as a loop of RederiveSessionLocked calls, trading the org-wide single deriver (and its cross-session dedup + one-tx-per-org atomicity) for the worker's per-session model. That trade is a design decision left for a human; the session-scoped race (seed, ad-hoc re-derive) is closed by RederiveSessionLocked in the meantime.

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) RederiveSessionLocked ¶ added in v0.25.0

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

RederiveSessionLocked is the externally-safe entry point for a session-scoped re-derive that may run WHILE the derive worker is live: it holds the per-session advisory lock across the whole read-derive-write pass, so it cannot interleave with the worker's derive of the same session and prune a turn the worker just wrote. The worker's own path (RederiveSession) is already called under this lock — it takes it in processEntry — so RederiveSession stays lock-free and this wrapper is the one non-worker callers use. Blocking: it waits out a concurrent worker derive rather than skipping, since a manual re-derive must actually run.

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) UpdateSessionName ¶ added in v0.25.0

func (d *Driver) UpdateSessionName(ctx context.Context, orgID, id string, name *string) (int64, error)

UpdateSessionName sets (or clears, when name is nil) the user-facing `name` column for a single session, scoped to the caller's org. The org_id predicate lives in the SQL (UpdateSessionName query), never just checked beforehand, so a cross-org id can never be updated. Returns the number of rows affected: 0 means no row matched (unknown id, or an id that exists but belongs to a different org), which the caller (the sessions handler) maps to a 404 rather than treating as success.

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.