sqlite

func (*Accounts) Close ¶

func (a *Accounts) Close() error

Close lets go of the file's pools, closing them if nothing else holds them.

func (*Accounts) CreateSession ¶

func (a *Accounts) CreateSession(ctx context.Context, session store.Session) error

CreateSession records a signed-in browser by its token's digest, and sweeps sessions that ended, which is the table's only bound: what is left is one row per browser still signed in.

func (*Accounts) FindClient ¶

func (a *Accounts) FindClient(ctx context.Context, id string) (store.OAuthClient, bool, error)

FindClient is a registered agent by its id.

func (*Accounts) FindSession ¶

func (a *Accounts) FindSession(ctx context.Context, tokenHash string) (store.Session, bool, error)

FindSession is the live session with this digest.

func (*Accounts) FindToken ¶

func (a *Accounts) FindToken(ctx context.Context, tokenHash string) (store.OAuthToken, bool, error)

FindToken is the live token with this digest.

func (*Accounts) FindUser ¶

func (a *Accounts) FindUser(ctx context.Context, who string) (store.User, bool, error)

FindUser is the person an actor names.

func (*Accounts) Grants ¶

func (a *Accounts) Grants(ctx context.Context, filter store.Grant) ([]store.Connection, error)

Grants are the live connections, newest first.

func (*Accounts) IssueToken ¶

func (a *Accounts) IssueToken(ctx context.Context, token store.OAuthToken) error

IssueToken records a token by its digest, sweeping ones that ended.

func (*Accounts) RegisterClient ¶

func (a *Accounts) RegisterClient(ctx context.Context, client store.OAuthClient, limit int) error

RegisterClient records a self-registered agent, refusing one past limit. The count and the insert share a transaction, so two registrations racing for the last place cannot both have it.

func (*Accounts) RevokeGrant ¶

func (a *Accounts) RevokeGrant(ctx context.Context, grant store.Grant) error

RevokeGrant ends every token under one grant.

func (*Accounts) RevokeSession ¶

func (a *Accounts) RevokeSession(ctx context.Context, tokenHash string) error

RevokeSession ends a session; revoking one that is already gone is not an error, since signing out twice signs out once.

func (*Accounts) RevokeToken ¶

func (a *Accounts) RevokeToken(ctx context.Context, tokenHash string) (bool, error)

RevokeToken ends one token.

func (*Accounts) SaveUser ¶

func (a *Accounts) SaveUser(ctx context.Context, user store.User) error

SaveUser replaces what was last recorded about a person.

func (*Reader) Activity ¶

func (r *Reader) Activity(ctx context.Context, filter store.ActivityFilter) ([]store.ActivityEntry, error)

Activity reads the log newest first. A pipeline filter matches the name an entry was recorded under or, through the id, the name that pipeline has now.

func (*Reader) Close ¶

func (r *Reader) Close() error

Close releases the connection, if this Reader is the one that opened it.

func (*Reader) Pipelines ¶

func (r *Reader) Pipelines(ctx context.Context) ([]store.PipelineRow, error)

Pipelines lists every pipeline recorded in the file, by name.

Ordered by name so a listing is stable rather than in the order rows happened to be created — a page whose entries move between reloads reads as a bug even when nothing changed.

func (*Reader) RecentRuns ¶

func (r *Reader) RecentRuns(ctx context.Context, pipelines []string, limit int) ([]store.CrossRunRow, error)

RecentRuns returns the newest runs across the NAMED pipelines, newest first.

The names are a filter in SQL, not applied to the result, and the difference is load-bearing: a state file may hold a pipeline this process does not serve, and fetching a limit and then dropping those rows would show fewer runs the busier the unserved pipeline is. Naming none asks for none — an empty served list is a configuration worth reporting, never a wildcard.

func (*Store) AbortQueuedJob ¶

func (s *Store) AbortQueuedJob(ctx context.Context, jobName string) (bool, error)

AbortQueuedJob keeps the row rather than deleting it, so the queue still says somebody stopped it.

func (*Store) AnswerQuestion ¶

func (s *Store) AnswerQuestion(ctx context.Context, id int64, answer, by string) error

AnswerQuestion records an answer against a pending question. It is the one write every ANSWERING channel shares — a seeded answer, a responder agent, a person at a TTY, `steps questions answer`, the web UI — so the options fence and the already-resolved race are decided once, here, rather than five times.

The gate is the method. Answering is the only outcome those two checks apply to, so they cannot move down into closeQuestion, which serves the outcomes nobody chose — see CloseQuestion on why these stay two entry points.

func (*Store) AppendRunEvent ¶

func (s *Store) AppendRunEvent(ctx context.Context, row store.RunEventRow) error

AppendRunEvent persists one run event. Called from the bus's sink goroutine (internal/events), so writes are already serialized. The EXISTS guard is one more primary-key probe of runs, the same one the foreign key already pays.

func (*Store) ApprovalStatus ¶

func (s *Store) ApprovalStatus(ctx context.Context, id int64) (store.Approval, error)

ApprovalStatus reads one approval's current state.

func (*Store) Approvals ¶

func (s *Store) Approvals(ctx context.Context, pendingOnly bool, limit int) ([]store.Approval, error)

Approvals lists what a pipeline has asked a person to decide. pendingOnly narrows it to the requests still waiting; limit <= 0 means no limit, the convention this repo uses everywhere.

Two orders, for the reason Store.Questions has two. Waiting requests read oldest-first, the order somebody should work through them in, and are never capped — a job is parked behind each one. The audit listing reads newest-first and is capped, because it is history.

func (*Store) AskQuestion ¶

func (s *Store) AskQuestion(ctx context.Context, question store.Question) (store.Question, bool, error)

AskQuestion records a pending question, or returns the one this run already asked.

The memo is the row, not a map beside it. Within one run a question is identified by its memo key, and the unique index is what makes that true under concurrency: an across: matrix whose twelve cells each ask the same thing, or two in_parallel: branches racing, all reach the SAME row — the first insert wins and every other caller gets that row back to wait on. Twelve identical prompts to one person is the behavior that ends adoption of a feature, and a map in the asking process would not survive the second process anyway.

existing reports whether the row was already there, which is what lets the caller tell "I am the one who asked this" from "I am waiting on somebody else's ask" — they are answered by the same row but reported to the model differently.

func (*Store) CheckErrors ¶

func (s *Store) CheckErrors(ctx context.Context) ([]store.CheckError, error)

CheckErrors is every resource this pipeline cannot currently check.

func (*Store) CheckedResources ¶

func (s *Store) CheckedResources(ctx context.Context) ([]store.CheckedResource, error)

CheckedResources lists every resource version the watcher has recorded.

func (*Store) ClaimNextJob ¶

func (s *Store) ClaimNextJob(ctx context.Context) (store.ClaimedJob, bool, error)

ClaimNextJob atomically transitions the oldest claimable pending row to running and returns it; found=false when nothing is claimable.

Two rows are not claimable. One whose job already has max_in_flight builds running — see config.Job.EffectiveMaxInFlight, which has already folded serial:/serial_groups: down to 1 by the time the row is synced. And one whose job shares a serial_groups: entry with a DIFFERENT job that is currently running, which is what stops two jobs mutating the same deploy target at once.

A job with no row defaults to 1, not to unlimited. A missing row means the job left the pipeline between enqueue and claim, and serializing something nobody can describe is the conservative reading.

Both conditions are inside the single UPDATE...RETURNING rather than checked beforehand, so two workers can never claim conflicting rows — a read-then-claim would have a race exactly where the lock is supposed to be.

func (*Store) Close ¶

func (s *Store) Close() error

Close reclaims what retention freed and closes the underlying connection.

Both reclaims are best-effort and deliberately ignored: a database that could not be compacted is still a valid database, and failing Close over it would turn a housekeeping problem into a failed command. Deleting rows only marks pages free, so without this a pruned file never shrinks — it just stops growing.

func (*Store) CloseQuestion ¶

func (s *Store) CloseQuestion(ctx context.Context, id int64, status store.QuestionStatus, answer, by string) error

CloseQuestion resolves a question nobody answered: 'expired' when the wait ran out (carrying the declared default, if there was one, so the row says what the model was actually told) or 'aborted' when the step ended first.

The aborted case is why this exists separately from AnswerQuestion. A question left `pending` after its step is gone is unanswerable, and showing it in `steps questions` as though somebody could still answer it is the same class of lie as presenting a default as a person's decision.

It reads like AnswerQuestion with the status fixed, and is not: the shared half is closeQuestion, already. What AnswerQuestion adds is the options fence, and the fence must NOT reach here. An expiry carries the `default:` declared on the grant, in the YAML, while the options come from the model's own tool call — nothing has ever compared the two, and nothing can, since the options do not exist until the call. An abort carries no answer at all.

Fold the two and that fence becomes conditional on a caller-supplied string the four answering sites each have to spell right to be checked at all. No constraint covers this column, so one misspelling writes a status no reader knows AND skips the fence, in the same silent step.

func (*Store) CompareAndSetCheckedVersion ¶

func (s *Store) CompareAndSetCheckedVersion(ctx context.Context, swap store.CheckedSwap) (bool, error)

CompareAndSetCheckedVersion is one statement either way, so nothing lands between the comparison and the write: an UPDATE guarded by the expected value, or an INSERT that does nothing if a row appeared.

func (*Store) CompleteJob ¶

func (s *Store) CompleteJob(ctx context.Context, id int64, status store.QueueStatus, runErr error) error

CompleteJob marks a claimed row done or failed. Callers must not call this for a run that stopped because ctx was canceled (SIGINT/SIGTERM) — that row should stay running so ResetStaleRunning re-queues it on the next daemon startup, since a graceful shutdown isn't a real failure and nothing else will re-trigger the job.

func (*Store) CompletedRunSteps ¶

func (s *Store) CompletedRunSteps(ctx context.Context, runID string) ([]store.RunStep, error)

CompletedRunSteps returns the steps a run already finished, in the order they finished — by rowid, the order they were recorded in.

func (*Store) ConsumedMark ¶

func (s *Store) ConsumedMark(ctx context.Context, jobName, resourceName string) (int64, error)

ConsumedMark is how far a job has fanned out over a resource: the highest check_order it has taken. Zero means it has taken nothing, and since check_order starts at 1 that reads as "everything is still to do".

func (*Store) Credential ¶ added in v0.2.0

func (s *Store) Credential(ctx context.Context, name string) ([]byte, bool, error)

Credential is the sealed bytes saved under name for this pipeline.

func (*Store) CredentialScope ¶ added in v0.2.0

func (s *Store) CredentialScope() string

CredentialScope is the pipeline's id, which a rename keeps and AUTOINCREMENT never hands out twice.

func (*Store) CurrentRevision ¶

func (s *Store) CurrentRevision(ctx context.Context) (store.Revision, bool, error)

CurrentRevision reads the revision a set made current, includes and all.

func (*Store) DecideApproval ¶

func (s *Store) DecideApproval(ctx context.Context, id int64, status store.ApprovalStatus, by, reason string) error

DecideApproval records a decision, refusing to overwrite one already made.

func (*Store) Delete ¶

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

Delete clears the current revision pointer first because it RESTRICTs, and the pipeline's own cascade would otherwise be refused by the row being deleted.

func (*Store) DeleteStepEvents ¶

func (s *Store) DeleteStepEvents(ctx context.Context, runID string, stepID int64, eventType string) error

DeleteStepEvents removes a step's events of one type. Called from the sink goroutine ahead of the record that replaces them, so the two never interleave with a chunk still queued.

Scoped by EXISTS on the one run rather than IN over the pipeline's runs: this runs once per step output, and the IN form built the pipeline's whole run list to answer for one id.

func (*Store) Delivery ¶

func (s *Store) Delivery(ctx context.Context, resourceName string, versionJSON store.VersionKey) (store.Delivery, bool, error)

Delivery reads back the payload recorded with a version.

func (*Store) Description ¶

func (s *Store) Description() string

Description is the state file behind this handle. Two handles reporting the same description share a file — which is what a shared state file makes possible, and what lets a caller ask one Reader about several pipelines at once instead of querying each and merging.

A path is already safe to print, so this driver has nothing to redact; a network driver returns a credential-free URL for the same string's purpose.

func (*Store) EnqueueJob ¶

func (s *Store) EnqueueJob(ctx context.Context, jobName, reason, by string) error

EnqueueJob inserts a pending trigger_queue row for jobName, unless one already exists — idx_trigger_queue_pending_job makes that a no-op dedup rather than an error, so a resource going dirty twice before a worker claims the row doesn't queue jobName twice.

The row carries a job NAME and a reason, and deliberately nothing about versions. It used to carry the versions the poll resolved, because a cursor-driven check could not be asked twice — which made a dropped enqueue data loss, and needed a merge, a restart fold and a hand-queued clear rule to be safe. A job now reads what it should build from resource_versions, so dropping a duplicate enqueue is once again what it looks like: the job is already queued.

func (*Store) EnqueueManualJob ¶

func (s *Store) EnqueueManualJob(ctx context.Context, jobName, reason, by string) error

EnqueueManualJob upserts rather than deduplicates: a pending row an automatic trigger left must still become manual, or the person's click is merged into a row the breaker then skips.

func (*Store) EnqueueRerunJob ¶

func (s *Store) EnqueueRerunJob(ctx context.Context, jobName, reason, runID, by string) error

EnqueueRerunJob keeps the row even when the job has an ordinary one pending: they build different versions.

func (*Store) FindRevision ¶

func (s *Store) FindRevision(ctx context.Context, sha string) (store.Revision, bool, error)

FindRevision returns a configuration this pipeline has run, by its hash.

Scoped to this pipeline like everything else here, even though the hash alone would find the row: a state file may hold several pipelines, and one answering for another's configuration would be a page showing a file the reader's pipeline never ran.

func (*Store) FindRunRow ¶

func (s *Store) FindRunRow(ctx context.Context, id string) (store.RunRow, bool, error)

FindRunRow reads one run by id, with ok reporting whether this pipeline has it — the single lookup every caller shares, from a run page to --resume.

Scoped like everything else, and here that is load-bearing rather than uniform: run ids are globally unique (pipeline.NewRunID is random), so in a shared state file `steps run a.yml --resume <id>` would otherwise happily continue a run of b.yml — reusing another pipeline's workspace and step indexes against this pipeline's plan. Not found is the right answer.

func (*Store) FinishRun ¶

func (s *Store) FinishRun(ctx context.Context, id string, status store.RunStatus) error

FinishRun records how a run ended, and when. The timestamp is what makes a run's duration answerable at all: started_at alone leaves every finished run looking like it is still going.

workspace is deliberately NOT cleared here, though it looks like dead weight once a run ends — an absolute path to a temporary directory, about ninety bytes a run, kept forever. It was tried and reverted: the column is read AFTER a run finishes by both of the features built on that, and neither can be told apart from here. --resume continues a FAILED run from the tree deliberately left on disk, and --replay forks a SUCCEEDED one that was kept with --keep-workspace. Clearing it on failure breaks resume; clearing it on success breaks replay.

func (*Store) FirstRunSince ¶

func (s *Store) FirstRunSince(ctx context.Context, jobName string, since time.Time) (store.RunRow, bool, error)

FirstRunSince returns the oldest run of a job started at or after `since`, with ok reporting whether one exists yet.

It backs the trigger-and-follow handoff: a queued job has no run id until the worker claims it and RunJob mints one, so the browser asks this until the run it caused appears. Oldest-first rather than newest, so a burst of triggers hands each caller the run its own click produced.

func (*Store) Forget ¶

func (s *Store) Forget(ctx context.Context, scope string, id int64) (bool, error)

Forget deletes one entry, and only when it is scope's.

func (*Store) ForgetChain ¶

func (s *Store) ForgetChain(ctx context.Context, jobName, rootHash string) error

ForgetChain removes (jobName, rootHash) from the skip index; a chain never recorded is a no-op.

func (*Store) ForgetScope ¶

func (s *Store) ForgetScope(ctx context.Context, scope string) (int, error)

ForgetScope deletes every entry of scope.

func (*Store) GreenVersions ¶

func (s *Store) GreenVersions(ctx context.Context, resourceName string, upstreamJobs []string) ([]map[string]any, error)

GreenVersions returns the versions of a resource that EVERY named upstream job has gone green against, oldest first by discovery order.

This is what a passed:-constrained get chooses among. The constraint is enforced at RESOLUTION rather than only at trigger time, because a gate checked at enqueue and forgotten by the build is checked against a world that can change in between: a poll validates v5, a newer v6 lands before a worker claims the job, and a build resolving plain "latest" ships the version nothing tested. It is also what makes `steps run` honor passed: at all — a manual run resolves through the same path a triggered one does.

Considered regardless of from_check: a version's green record proves a build fetched it, which is better evidence of existence than a check listing. Concourse's model, and the reason a job whose head keeps failing upstream still deploys the newest version that DID pass.

func (*Store) HasNodeSucceeded ¶

func (s *Store) HasNodeSucceeded(ctx context.Context, jobName, hash string) (bool, error)

HasNodeSucceeded reports whether a node with this exact hash has already been recorded as succeeded for this job.

It is per-NODE memoization, distinct from HasSucceededBatch's per-CHAIN check. The chain form asks "did this whole path succeed", which is right for a sequence: a changed step invalidates everything after it. An across: cell has no such sequence — cells are siblings, and one cell changing says nothing about another — so a cell asks about itself alone.

func (*Store) HasNodesSucceeded ¶

func (s *Store) HasNodesSucceeded(ctx context.Context, jobName string, hashes []string) (map[string]bool, error)

HasNodesSucceeded is HasNodeSucceeded for many hashes in one query.

func (*Store) HasPassedVersionSet ¶

func (s *Store) HasPassedVersionSet(ctx context.Context, jobName string, want map[string]store.VersionKey) (bool, error)

HasPassedVersionSet reports whether jobName has one build in which EVERY (resource, version) pair in want was green at the same time.

This is the correlated question, and the difference from asking per-resource is the whole point. Two versions that each passed upstream in different builds have never been proven to work TOGETHER, and a downstream fan-in that accepts them is running a combination nothing validated. Concourse resolves passed: across a whole plan at once for this reason; see docs/conformance.md. The per-resource form is deliberately not kept alongside — an available answer to the wrong question is how the bug comes back.

Rows written before build_id existed carry ” and are excluded, so they can never satisfy a correlated match. Conservative on purpose: passed: is a gate.

An empty want is vacuously true — there is no constraint to satisfy.

func (*Store) HasSucceededBatch ¶

func (s *Store) HasSucceededBatch(ctx context.Context, jobName string, rootHashes []string) (map[string]bool, error)

HasSucceededBatch reports which of rootHashes have a prior succeeded run recorded for jobName, in one round trip instead of one query per hash.

func (*Store) HeldJobs ¶

func (s *Store) HeldJobs(ctx context.Context) ([]store.HeldJob, error)

HeldJobs lists every job currently out of the rotation, oldest hold first. It backs `steps jobs`, which is how an operator finds out a nightly job stopped without reading a weekend of logs.

func (*Store) IsJobHeld ¶

func (s *Store) IsJobHeld(ctx context.Context, jobName string) (bool, error)

IsJobHeld reports whether the breaker has taken a job out of the rotation.

func (*Store) LastChecked ¶

func (s *Store) LastChecked(ctx context.Context, resourceName string) (store.CheckedResource, bool, error)

LastChecked returns one resource's last-checked row, with found=false when nothing has checked it yet — the single-resource counterpart to CheckedResources, for a caller that wants one row rather than every resource's just to look one up.

The version JSON on it is the check cursor: what the next check is handed to say "anything after this", and still the resource's current version when a check reports nothing new.

func (*Store) LatestFinishedRunByJob ¶

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

LatestFinishedRunByJob returns every job's newest run that is no longer running, keyed by job name — one query for a board rather than one per busy job. Reruns count, as they do in ListRuns.

func (*Store) LatestRunByJob ¶

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

LatestRunByJob returns the most recent run for every job that has one, keyed by job name — one query for a jobs board rather than one per job.

A rerun counts only when it reruns the job's latest other run, which is Concourse's rule (updateLatestCompletedBuildForJob): a green retry of an old build must not turn a job that is red today green.

func (*Store) ListMemories ¶

func (s *Store) ListMemories(ctx context.Context, scope string, limit int) ([]store.Memory, error)

ListMemories is scope's entries, newest first.

func (*Store) ListNodes ¶

func (s *Store) ListNodes(ctx context.Context, jobName string, limit int) ([]store.NodeRow, error)

ListNodes returns the most recently recorded steps, newest first. An empty jobName covers every job.

func (*Store) ListRuns ¶

func (s *Store) ListRuns(ctx context.Context, jobName string, limit int) ([]store.RunRow, error)

ListRuns returns run invocations, newest first. An empty jobName covers every job.

func (*Store) ListTriggerQueue ¶

func (s *Store) ListTriggerQueue(ctx context.Context, limit int) ([]store.QueueRow, error)

ListTriggerQueue returns the most recent trigger-queue entries, newest first — what `steps web` has queued, run, or failed to run.

func (*Store) MaxStepID ¶

func (s *Store) MaxStepID(ctx context.Context, runID string) (int64, error)

MaxStepID is joined to runs for the pipeline, as RunEvents is.

func (*Store) MemoryScopes ¶

func (s *Store) MemoryScopes(ctx context.Context) ([]store.MemoryScope, error)

MemoryScopes is every scope holding an entry, by name.

func (*Store) NodeTranscript ¶

func (s *Store) NodeTranscript(ctx context.Context, hash string) (string, bool, error)

NodeTranscript returns the stored transcript JSON for a node hash, with ok reporting whether one exists — mirroring LastChecked's shape rather than inventing a sentinel error for the common "never recorded" case.

func (*Store) NodesByHash ¶

func (s *Store) NodesByHash(ctx context.Context, hashes []string) (map[string]store.NodeRow, error)

NodesByHash returns the recorded nodes for the given hashes, keyed by hash — one query for a whole run transcript rather than one per step.

func (*Store) PassedVersions ¶

func (s *Store) PassedVersions(ctx context.Context, jobName string, limit int) ([]store.PassedVersion, error)

PassedVersions lists the resource versions a job has recorded as green — what a downstream passed: constraint is satisfied by.

func (*Store) Pause ¶

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

Pause stamps paused_at; a second Pause keeps the first stamp, since when it was paused is the fact a reader wants.

func (*Store) Paused ¶

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

Paused reports whether the pipeline-level breaker is set.

func (*Store) Pipeline ¶

func (s *Store) Pipeline() string

Pipeline is the name this handle is scoped to.

func (*Store) Prune ¶

func (s *Store) Prune(ctx context.Context, policy store.Retention, keepRunID string) error

Prune applies one build's retention policy: its job's run history, the caches that job has accumulated, and its finished trigger-queue rows.

One method because it is one decision made once, at the end of a build, and because the footprint contract is a statement about the whole sweep rather than about whichever half a caller remembered to invoke.

Both passes always run, and their failures are joined rather than short-circuited: they bound different tables for different reasons, and the first one failing is no reason to leave the second table growing.

keepRunID is the run that must survive whatever the cap says — the one whose build is calling this. Without it, retention could delete the run it was invoked from: a RESUMED run keeps its original started_at (StartRun inserts and ResumeRun leaves it alone, which is right — it is when the run started), so a resume of an older run is the OLDEST row, and it reaped itself at the end of its own build. That left its run_steps and agent_usage cascaded away and the run lookup answering "no such run", making the run permanently unresumable. Pass "" when no run is in play.

func (*Store) QuestionStatus ¶

func (s *Store) QuestionStatus(ctx context.Context, id int64) (store.Question, error)

QuestionStatus reads one question's current state.

func (*Store) Questions ¶

func (s *Store) Questions(ctx context.Context, pendingOnly bool, limit int) ([]store.Question, error)

Questions lists what a pipeline has asked. pendingOnly narrows it to the questions still waiting; limit <= 0 means no limit, the convention this repo uses everywhere.

The two orders are not cosmetic, and are why this is one method rather than a filter bolted onto one query. Waiting questions read oldest-first, because that is the order somebody should answer them in. The full listing puts pending FIRST and then the rest newest-first: it is capped, and the only route the UI offers for ANSWERING is this page, so ordered purely by recency a pipeline that asked `limit` more questions would push a parked one off the page while the nav badge still counted it — a run parked with nothing on screen to unpark it.

func (*Store) Reader ¶

func (s *Store) Reader() store.Reader

Reader returns an unscoped reader over the same file this handle is scoped to. Reading through it crosses pipelines by construction, so every method on it names which ones it wants.

func (*Store) RecordActivity ¶

func (s *Store) RecordActivity(ctx context.Context, entry store.ActivityEntry, limit int) error

RecordActivity appends an entry under this pipeline, copying the pipeline's current name so the row still reads after a destroy nulls the id.

func (*Store) RecordAgentUsage ¶

func (s *Store) RecordAgentUsage(ctx context.Context, usage store.AgentUsage) error

RecordAgentUsage stores what one agent step spent.

ACCUMULATES the token counts on conflict, and replaces the descriptive fields.

A step can execute more than once against the same node hash: a to: route sending the plan back over it, or a resumed run re-running the step its predecessor died on. You paid for every one of those attempts, so the counts are a running total — replacing them reported the last attempt as though it were the whole bill, and made this table unusable as the ledger a resumed run reads its prior spend from (RunTokensSpent).

The descriptive columns still take the newest value: model_served, finish_reason and raw_meta describe ONE response and cannot be summed, and the last attempt is the one whose result the run actually used.

The conflict is on the node hash, so two DIFFERENT steps sharing a plan index — the cells of a matrix, the members of an ensemble — never collide. A hook or fix has no node, and a NULL never conflicts, so each execution of one is a row of its own.

func (*Store) RecordChainSucceeded ¶

func (s *Store) RecordChainSucceeded(ctx context.Context, jobName, rootHash string) error

RecordChainSucceeded adds (jobName, rootHash) to the skip index.

func (*Store) RecordCheckError ¶

func (s *Store) RecordCheckError(ctx context.Context, resourceName, message string) error

RecordCheckError files why a check failed, or clears the filing when the message is empty.

The clear runs once per resource per poll for the whole life of a healthy pipeline and matches nothing almost every time, which is cheap: a DELETE opens a write transaction either way — an EXISTS guard in front of it would not change that — and one that dirties no page commits without appending a WAL frame the live stream has to read.

func (*Store) RecordCheckedVersion ¶

func (s *Store) RecordCheckedVersion(ctx context.Context, resourceName string, versionJSON store.VersionKey) error

RecordCheckedVersion upserts the latest observed version JSON for resourceName, independent of whether any job triggered by the change succeeds — version-checking and build outcomes are tracked separately, mirroring how job_runs only ever records succeeded chains.

func (*Store) RecordConsumedMark ¶

func (s *Store) RecordConsumedMark(ctx context.Context, jobName, resourceName string, order int64) error

RecordConsumedMark advances a job's cursor to include order.

Only ever forward. A run that takes an older version than one already taken -- a backfill reaching a job that has moved past it -- must not rewind the mark and offer everything in between a second time.

No pruning and no cap, which is the point of a mark: there are no members to forget, so it cannot develop the holes a capped set does. See the table comment in schema.go.

func (*Store) RecordDelivery ¶

func (s *Store) RecordDelivery(ctx context.Context, resourceName string, delivery store.Delivery, dispatch store.Dispatch, limit int) (bool, error)

RecordDelivery files the version, its payload and — unless held — the resource's current version and the jobs it triggers, in one transaction, which is what lets the handler promise a sender that a 2xx lost nothing.

func (*Store) RecordJobOutcome ¶

func (s *Store) RecordJobOutcome(ctx context.Context, jobName string, succeeded bool, maxFailures int) (held bool, consecutive int, err error)

RecordJobOutcome updates a job's consecutive-failure count and reports whether the job is now held.

It counts triggered RUNS, not the retries inside one — conflating the two would trip the breaker on ordinary flakiness that a retry would have absorbed, which is the opposite of the intent. maxFailures of 0 means the job has no breaker: the count is still kept (so turning one on later starts from a real number rather than zero), but it never holds.

func (*Store) RecordNode ¶

func (s *Store) RecordNode(ctx context.Context, node store.NodeRecord, jobName string, status store.NodeStatus, result map[string]any, execErr error) error

RecordNode upserts a node's execution outcome, keyed by its content hash.

Two statements in one transaction because content is interned: the preimage goes into node_content under a hash of itself, and the node points at it. A single statement cannot do that, and doing it without the transaction would leave a window where the node's reference has no parent row — which the RESTRICT on that reference correctly refuses.

func (*Store) RecordPassedVersion ¶

func (s *Store) RecordPassedVersion(ctx context.Context, pass store.Pass) error

RecordPassedVersion records that jobName completed successfully against this exact version of a resource. It is what a downstream job's passed: reads.

func (*Store) RecordPassedVersions ¶

func (s *Store) RecordPassedVersions(ctx context.Context, passes []store.Pass) error

RecordPassedVersions records every pass in one transaction.

func (*Store) RecordPlacement ¶

func (s *Store) RecordPlacement(ctx context.Context, placement store.Placement) error

RecordPlacement stores where one placed step ran.

REPLACES on conflict, unlike agent usage, which accumulates. Tokens are spend and every attempt was paid for; a placement is a description of the machine, and a step that ran twice against the same node ran on whichever machine answered last. Summing "which machine" has no meaning.

func (*Store) RecordRevision ¶

func (s *Store) RecordRevision(ctx context.Context, sha, source string, includes map[string]string) error

RecordRevision interns the configuration a run was started from.

Interning only: it does NOT decide what the next run records. That was the original design — the id lived on this handle and StartRun read it — and it was wrong in a way the handle could not see. A run takes its configuration as a pointer and writes its row minutes later, after placement, leases, image pulls and preflight; a reload in that window made the run name a configuration it never executed. The sha travels WITH the config instead (see revisionBySHA).

Upsert by (pipeline_id, sha), so an unchanged configuration loaded a thousand times is one row. The conflict refreshes loaded_at and nothing else: source is the same bytes by definition — the sha is over them, so rewriting a multi-KB column would be a WAL page per load for no change — while loaded_at is what the sweep reads to know which row is the one being served, and a conflict is precisely a re-load.

func (*Store) RecordRunInput ¶

func (s *Store) RecordRunInput(ctx context.Context, runID string, input store.RunInput) error

RecordRunInput remembers that a run was created with this version, bound under this get — every get of the run, which is what a resume rebuilds it against. See the run_inputs DDL for why the cursor cannot answer it. A run is created once, so a second record of the same get is the same row.

func (*Store) RecordRunParent ¶

func (s *Store) RecordRunParent(ctx context.Context, runID, parentID string) error

RecordRunParent notes that a run was forked from another by a replay.

Kept as its own statement rather than a StartRun parameter: every existing caller of StartRun records an ordinary run, and threading an almost-always empty argument through them would put replay's vocabulary in the path of code that has nothing to do with it.

func (*Store) RecordRunRerun ¶

func (s *Store) RecordRunRerun(ctx context.Context, runID, originalID string) error

RecordRunRerun is RecordRunParent's twin, and a statement of its own for the same reason.

func (*Store) RecordRunStep ¶

func (s *Store) RecordRunStep(ctx context.Context, runID string, step store.RunStep) error

RecordRunStep marks one step of one build of a run as done, so a resume skips it.

func (*Store) RecordStepBlobs ¶

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

RecordStepBlobs files the content digests of a step's outputs under its action key, replacing whatever the key held. outputs maps a DECLARED output name to the digest of the artifact it produced.

func (*Store) RecordVersionOrder ¶

func (s *Store) RecordVersionOrder(ctx context.Context, resourceName string, versionJSON store.VersionKey) (int64, error)

RecordVersionOrder files a version if it is not already known and returns the order it sits at, so a caller that resolved its own versions can advance a cursor over them.

func (*Store) RecordVersions ¶

func (s *Store) RecordVersions(ctx context.Context, resourceName string, versions []map[string]any, limit int) (int, error)

RecordVersions files what a check reported, assigning each genuinely new version the next check_order, and prunes beyond the cap. It returns how many versions were new to check-history, which is what decides whether anything is worth triggering for.

A version a check already filed is left ALONE — its check_order is the order it was discovered in, and revising it would reorder history under a job midway through walking it. A check re-reporting its whole window every poll therefore writes nothing, which is the common case.

A version only a RUN had filed (from_check = 0) is the one exception: the check is discovering it now, so it takes a fresh order at the top. Keeping its stale order would be worse than it sounds — "latest" resolves by highest order, so a run-filed newest version would sort below everything a later check reported, and the prune would treat the newest version as the OLDEST and delete it first.

Ordering within one call follows the slice, which is the order the check returned: oldest first, by the convention a check owes.

func (*Store) Release ¶

func (s *Store) Release() error

Release closes the connection without Close's reclaim, which holds the file's write lock as long as it takes: measured 31s after a destroy freed a gigabyte, with other pipelines' writes to the file failing SQLITE_BUSY.

func (*Store) Remember ¶

func (s *Store) Remember(ctx context.Context, memory store.Memory, limit int) (store.Memory, bool, error)

Remember files the memory, or finds it already filed, and applies the cap, in one transaction.

The run id goes through a scoped subquery rather than straight in: a run of a SIBLING pipeline satisfies the foreign key, and a memory should not claim a run this pipeline cannot show. A run that is not this pipeline's files NULL, the same as one retention has reaped.

func (*Store) Rename ¶

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

Rename is one UPDATE: every scoped row reaches the pipeline by id, so history, cache and queue follow the row rather than the name.

func (*Store) RequestApproval ¶

func (s *Store) RequestApproval(ctx context.Context, jobName, message string) (int64, error)

RequestApproval records a pending approval and returns its id.

func (*Store) RequestApprovalFor ¶

func (s *Store) RequestApprovalFor(ctx context.Context, runID, stepKey, jobName, message string) (int64, error)

RequestApprovalFor finds the live approval this step of this run already made, or makes one. The insert does nothing on a conflict with the live one, and the read after it is what answers either way.

func (*Store) ResetJobFailures ¶

func (s *Store) ResetJobFailures(ctx context.Context, jobName string) error

ResetJobFailures clears a job's consecutive-failure count and releases it. Any success resets the breaker — including a manual `steps run`, which is the natural way to confirm a fix.

func (*Store) ResetStaleRunning ¶

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

ResetStaleRunning flips this pipeline's running rows back to pending, and closes its running runs as aborted — called once at daemon startup so a killed (or gracefully but incompletely shut down) process doesn't strand claimed work, or a run page, forever.

It assumes ONE steps process owns a pipeline's queue at a time, and that assumption is now the whole guard. A file lock used to enforce it, so that a second daemon would give way rather than treat a live build's row as an abandoned leftover — flip it, let the job be claimed twice, and silently defeat serial:/max_in_flight. The lock is gone deliberately, and so is the second daemon it arbitrated: `steps web` is the only long-running mode, so a running row found at startup belongs to a process that is gone. Two of them against one state file is a deployment mistake rather than a case to support — see CLAUDE.md's one-process-per-state-database rule, which is what replaced the lock.

The rows are scoped to this pipeline because a state file may hold several, and one pipeline's watcher starting up must not reach into another's in-flight builds.

A running row may coexist with a pending row for the same job (a version change enqueued while that job was mid-run). Flipping the running row to pending would then create two pending rows for one job, violating idx_trigger_queue_pending_job. So any running row whose job already has a pending successor is deleted instead of flipped — that later pending row already covers re-running the job, and the interrupted build's own outputs were never captured, so there is nothing to preserve. Called once at startup before any worker/poller goroutine exists. One transaction, so a failure part way leaves the queue and the runs as they were rather than re-queued jobs whose runs still read as running.

func (*Store) ResourceVersionsJSON ¶

func (s *Store) ResourceVersionsJSON(ctx context.Context, resourceName string) ([]store.VersionKey, error)

ResourceVersionsJSON returns the versions a CHECK has reported for a resource, oldest first — the same order and the same contract a check's own output has, so a caller can treat it as the check's answer without knowing where it came from.

Rows that exist only because something referenced them are excluded, and an empty result means "nothing has checked this resource, go and check". Returning them would be worse than useless: a `steps run` records the one version it took, and treating that lone row as the resource's history would hide every other version from the next run.

Stored JSON rather than map[string]any, because half the callers only DISPLAY a version and decoding every row only to re-encode it a moment later buys nothing. A caller that inspects fields runs DecodeVersion, which is where the UseNumber that keeps an id out of exponent notation lives.

func (*Store) ResumableRuns ¶

func (s *Store) ResumableRuns(ctx context.Context) ([]store.RunRow, error)

ResumableRuns is read before the restart closes them, so the sweep knows whose tree to keep.

func (*Store) ResumeRun ¶

func (s *Store) ResumeRun(ctx context.Context, id, workspaceDir, configSHA string) error

ResumeRun puts an existing run back in flight and points it at the build this attempt is using.

The legitimate half of what the old upsert did. It updates and never inserts, so a resume of a run that is not there is an error instead of a silently invented row — and it is scoped to this pipeline, because a run id names a row in ONE pipeline and reaching another one would put a foreign run back in flight under this pipeline name.

job_name is deliberately not written: the job a run belongs to was decided when it was minted, and a resume that could rewrite it would make the run history disagree with the events already recorded against it.

The REVISION is written, and for the mirror of that reason: a resume continues a failed run under the configuration it is being resumed with, which is usually the one that fixed it. Leaving the original would make the run claim it executed a pipeline that nothing in it ever ran.

COALESCEd rather than assigned, because a subselect that matches nothing is not an answer: an empty sha (a caller that loaded no file) or a row this pipeline no longer has would ERASE a configuration the run had correctly recorded, turning "it ran that" into "it ran none" — the one thing this column exists to deny. An update that cannot name a new configuration keeps the old one.

func (*Store) RunCostTotals ¶

func (s *Store) RunCostTotals(ctx context.Context, limit int) ([]store.RunTotals, error)

RunCostTotals rolls agent_usage up per run, newest first.

Unpriced counts the steps with no reported cost, so a partial dollar total can be shown AS partial rather than presented as the whole bill.

func (*Store) RunEvents ¶

func (s *Store) RunEvents(ctx context.Context, runID string, afterSeq int64, limit int) ([]store.RunEventRow, error)

RunEvents replays a run's events in order, from afterSeq exclusive. Pass 0 for the whole run — which is also how a reconnecting live view catches up on what it missed without re-reading what it already has.

Joined to runs for the pipeline, which run_events has no column of its own for: a holder of this facet alone has no FindRunRow to ask first, so a run id another pipeline minted must read as nothing here rather than as that pipeline's events.

func (*Store) RunInputs ¶

func (s *Store) RunInputs(ctx context.Context, runID string) ([]store.RunInput, error)

RunInputs reports the versions a run was created with.

Joined to runs for the pipeline, which run_inputs has no column of its own for — the same shape as CompletedRunSteps, and for the same reason.

func (*Store) RunPlacements ¶

func (s *Store) RunPlacements(ctx context.Context, runID string) ([]store.Placement, error)

RunPlacements returns where every placed step of one run ran, in plan order.

func (*Store) RunTokensSpent ¶

func (s *Store) RunTokensSpent(ctx context.Context, runID string) (int, error)

RunTokensSpent is the total tokens already recorded against a run.

Read when RESUMING, so a job budget continues from what earlier attempts of the same run spent instead of restarting at zero. Summed in SQL rather than by walking RunUsage: the caller wants one number, and a resumed run may carry hundreds of step rows.

A run with no agent steps yet returns 0, which is also what a fresh run's id yields — both mean "nothing spent", so there is nothing to distinguish.

func (*Store) RunUsage ¶

func (s *Store) RunUsage(ctx context.Context, runID string) ([]store.AgentUsage, error)

RunUsage is every agent step's spend for one run, in step order.

func (*Store) RunsUsingNode ¶

func (s *Store) RunsUsingNode(ctx context.Context, hash string, limit int) ([]store.RunRow, error)

RunsUsingNode lists the runs whose events reference a node hash — the "which runs reused this cached step" answer a node page is built on.

func (*Store) SaveCredential ¶ added in v0.2.0

func (s *Store) SaveCredential(ctx context.Context, name string, sealed []byte) error

SaveCredential replaces the sealed bytes saved under name.

func (*Store) SaveNodeTranscript ¶

func (s *Store) SaveNodeTranscript(ctx context.Context, hash, transcript string) error

SaveNodeTranscript stores (or replaces) an agent node's full conversation transcript, a JSON array of events. Kept in its own table so nodes.result — which rides along on every node listing — stays bounded; a transcript is read on demand instead.

Truncated at MaxTranscriptBytes, keeping the HEAD: the task, the first tool calls and the early decisions are what explain what a step was doing, and a conversation that ran long enough to hit the cap is one whose middle is mostly repetition.

func (*Store) SerialGroupHolder ¶

func (s *Store) SerialGroupHolder(ctx context.Context, jobName string) (string, error)

SerialGroupHolder names a running job that shares a serial group with jobName, or "" when nothing is holding the lock. It exists so a waiting job can say WHO it is waiting for — "queued" and "blocked on a lock" are different states, and a reader who cannot tell them apart cannot tell a stuck pipeline from a busy one.

func (*Store) SetCurrentRevision ¶

func (s *Store) SetCurrentRevision(ctx context.Context, sha, from string) error

SetCurrentRevision refuses a sha nothing recorded rather than pointing at nothing: a daemon restarting against a NULL would silently serve no pipeline. The path rides in the same UPDATE, because a second write after it could fail with the switch already made.

func (*Store) SetSourcePath ¶

func (s *Store) SetSourcePath(ctx context.Context, source string) error

SetSourcePath records where this pipeline's YAML lives, for whoever reads the file back — `steps runs --db shared.db` lists it, and a name alone does not say which of two checkouts `pipeline` is.

Written on every open by the commands that load a pipeline, so a checkout that moves stops reporting where it used to live. A read-only command has nothing to correct and does not call this: it would be recording a path it never resolved.

func (*Store) StartRun ¶

func (s *Store) StartRun(ctx context.Context, run store.NewRun) error

StartRun records a NEW run and the workspace its steps will share.

An insert, never an upsert: minting an id and continuing a run are different acts, and the single statement that served both could not tell them apart. Continuing one is ResumeRun.

DO NOTHING plus a row count rather than letting the constraint violation out, so the answer does not depend on how a driver spells its unique-index error.

func (*Store) StepBlobs ¶

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

StepBlobs returns the digests recorded under an action key, keyed by declared output name — empty when the key is unknown, which a caller reads as an ordinary miss.

func (*Store) SyncJobLimits ¶

func (s *Store) SyncJobLimits(ctx context.Context, groups map[string][]string, limits map[string]int) error

SyncJobLimits replaces the recorded admission rules — serial-group membership and per-job concurrency — with what the pipeline currently declares.

Both live in SQL rather than in the Config because ClaimNextJob admits in one atomic statement, so who may run has to be readable from the claim itself rather than consulted in Go afterwards.

Replaced wholesale rather than merged: a group or a limit removed from the YAML must stop applying, and a stale row would hold two jobs apart, or throttle one, with nothing in the pipeline to explain why.

One transaction over both tables, because they are one answer. They are always synced together, from one config, at the two moments a configuration is ADOPTED (startup and every reload) — and a swap that landed the serial groups and then failed would serve a `max_in_flight:` the queue goes on ignoring, which is exactly the half-applied state a declarative mirror must never be left in.

The delete is scoped, since a shared state file holds other pipelines' mirrors of their own YAML.

func (*Store) Unpause ¶

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

Unpause clears the stamp.

func (*Store) VersionOrders ¶

func (s *Store) VersionOrders(ctx context.Context, resourceName string) (map[store.VersionKey]int64, error)

VersionOrders maps every recorded version of a resource to its check_order, INCLUDING the rows a check did not file.

Deliberately wider than ResourceVersionsJSON, which answers "what exists" and therefore reports only what a check saw. This answers "where does this version sit", and a job that resolved its own versions needs an order for them or its cursor could never advance past them — a `steps run` against an unpolled resource would repeat its whole fan-out every time.

func (*Store) VersionRuns ¶

func (s *Store) VersionRuns(ctx context.Context, resourceName string, versions []store.VersionKey) ([]store.VersionRun, error)

VersionRuns reports every run that took one of these versions of this resource as an input, newest first. DISTINCT because two gets of one resource in a run may bind the same version, and that is one run of it, not two.