store

package
v0.2.0 Latest Latest
Warning

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

Go to latest
Published: Oct 25, 2026 License: MIT Imports: 11 Imported by: 0

Documentation

Overview

Package store is the contract steps holds its state database to: the row types every driver returns, the errors every driver raises, and the Store interface every driver implements.

It runs no SQL and opens no connection. A driver — internal/store/sqlite or internal/store/postgres — owns its schema, its version stamp and its queries end to end, so each can use what its database is good at rather than the intersection of two. What keeps them honest with each other is this file plus the tests written against it (internal/store/storetest).

Only internal/cli names a driver; every other package takes a Store it was handed.

Index

Constants

View Source
const (
	UsageStep = "step"
	UsageHook = "hook"
	UsageFix  = "fix"
)

What spent an agent_usage row's tokens. A hook's agent and a task's fix: agent run inside the job and spend against its budget, but have no node.

View Source
const ChainsPerRetainedRun = 5

ChainsPerRetainedRun is the same multiplier for job_runs, which holds one row per whole chain rather than one per step — so a run costs a handful of rows rather than a plan's worth. Several because a version: every fan-out runs several chains under one job.

View Source
const DefaultActivityCap = 10000

DefaultActivityCap is how many entries the log keeps when nothing says otherwise.

View Source
const DefaultMemoryEntries = 100

DefaultMemoryEntries bounds a scope's entries when nothing says otherwise. A count, like every other cap here, and the oldest go first. Override with defaults.memory_entries: in the pipeline.

View Source
const DefaultResourceVersionCap = 1000

DefaultResourceVersionCap bounds resource_versions per resource when nothing says otherwise.

A count rather than an age, for the same reason every other cap here is: what matters is how far behind a job may fall, not how old a version is. A resource nobody has polled in a month has not accumulated anything, while a busy one accumulates a row per version forever.

The bound is not free of meaning. A version pruned here takes its job_versions row with it (ON DELETE CASCADE), so a `passed:` gate can no longer clear for it — correct, since a version out of history cannot be built, but it means a cap set below what a slow downstream job needs will hold that job back. Override with defaults.version_history: in the pipeline, or --version-history on the command line.

View Source
const DefaultTriggerQueueHistory = 50

DefaultTriggerQueueHistory bounds the finished rows kept in trigger_queue.

The queue is a work list, not a history: a pending row means "run this job" and a done row means nothing at all to anything that reads the table. They were kept anyway, and a failed one carries the error that stopped the job — which for a check is the whole generated script, comments included, about 1.3KB. A remote that is down at a one-minute poll interval wrote that same 1.3KB every minute, forever.

Enough rows to answer "what has this job been doing lately" from `steps runs`, and far too few to matter on disk.

View Source
const MaxEventTextBytes = 64 * 1024

MaxEventTextBytes bounds the free-text columns of one run event.

The publishers already bound what they MEAN to store — a step's output at 32,000 bytes (pipeline's maxPublishedOutputBytes), a tool result at 16,384 (agent's maxRecordedResultBytes) — but nothing bounded an event carrying an ERROR, and one error is routinely enormous: a failing check or task reports `command %q failed`, where the command is the whole generated shell script, about 1.3KB for the built-in git check. MaxStoredErrorBytes caps that for the node and queue error columns; this event went in verbatim, into the table with the most rows, on every failing step of every poll.

Set above every deliberate publisher cap so it never truncates something a publisher chose to keep — it is the backstop for the columns nobody capped, not a second opinion on the ones they did.

View Source
const MaxFreeTextBytes = 64 << 10

MaxFreeTextBytes caps the one free-text value a person stores whole: an approval's reason, a question's answer. The browser and the CLI both refuse past it.

View Source
const MaxMemoryBytes = 1024

MaxMemoryBytes caps one entry. A memory is a fact, not a transcript: one that does not fit is a model saving the conversation, which is the next run's preload filling up with one entry.

View Source
const MaxMemoryScopeBytes = 256

MaxMemoryScopeBytes caps a scope's name. A scope is an id — a Slack user id is eleven bytes — and one this long is a file that was not one.

View Source
const MaxStoredErrorBytes = 2 * 1024

MaxStoredErrorBytes bounds the error text any column here keeps.

A stored error is unbounded by default, and one of them is routinely enormous: a failing check or task reports "command %q failed", where the command is the whole generated shell script — about 1.3KB for the built-in git check, with its comments. That message is exactly right in a terminal and far too big to keep a copy of per failure, forever. A remote that is down at a one-minute poll wrote that same 1.3KB every minute.

2KB keeps the head of any real message — the part naming what failed — while making the worst case a rounding error rather than the largest thing a failing watch produces.

View Source
const MaxTranscriptBytes = 256 * 1024

MaxTranscriptBytes bounds one stored transcript.

internal/agent already caps a single tool RESULT (maxRecordedResultBytes), which bounds the widest single value in a transcript but not the transcript: a conversation has as many turns as it needs, and an agent that worked for an hour writes all of them into one row. The largest value in the schema was therefore the one with no total bound at all.

256KB is far past any transcript worth reading end to end while staying an order of magnitude below the point where one row dominates the database. A transcript is a diagnostic, not a ledger: nothing reads it to make a decision, so losing the tail of a very long one costs nothing that the per-step token counts in agent_usage do not still record exactly.

View Source
const NodesPerRetainedRun = 20

NodesPerRetainedRun is how many cached nodes a job may keep per run it retains — the multiplier turning run_history: into a node cap.

Not its own DSL knob: run_history: already expresses "how much of this job do you want kept", and a second dial would ask authors to reason about node counts per plan, which is an implementation detail. Scaling off it rather than being a flat number matters because the flat version did not bound anything at realistic scale — a generous constant meant the cache, not the history, was the term that grew, and the footprint measurement caught it.

Twenty is a plan-size allowance: enough that a job of twenty steps keeps a full run's worth of nodes for every run it retains, and the nodes of the retained runs themselves are exempt from the cap anyway, so this governs only how much EXTRA cache is carried for its hit value.

View Source
const StepBlobEntryCap = 200

StepBlobEntryCap bounds how many step entries this index keeps per pipeline, newest INSERTED first — which is not the local cache's LRU: a local hit refreshes the entry's mtime but records nothing here, so under churn past the cap the rows evicted first are the hottest keys. Known and accepted: re-recording on every hit would cost a digest walk per hit, the loss is cost-only (the other machine re-runs and re-records), and the cap exists for bound, not fidelity. It mirrors the on-disk step cache's own entry bound (workspace's defaultStepCacheMaxEntries, 200): an index entry whose local bytes could never still be cached buys little, since the action keys of evicted entries stop being computed long before they stop being stored. Count, never age — the standing rule for caches in this schema.

View Source
const TruncationEvent = `{"type":"text","text":"[transcript truncated: over the stored size limit]"}`

TruncationEvent is appended in place of the entries dropped at the cap, so a reader who reaches the end knows the conversation continued rather than stopped. Spelled as a transcript event because that is what the consumers decode (internal/agent's transcriptEvent, rendered by internal/web).

View Source
const UnusedClientTTL = 24 * time.Hour

UnusedClientTTL is how long a registered client with no token may sit before a registration is free to evict it; it is the only thing that frees a place, since registration answers anybody and nothing else deletes a client.

Variables

View Source
var (
	// ErrRunExists is a MINT against an id some run already holds.
	//
	// Loud on purpose, and the whole reason StartRun is not an upsert. A run
	// id is a single global key, and an upsert answered a collision by taking
	// the existing row OVER: a finished run flipped back to running and its
	// workspace repointed, while the row kept the old job name and the old
	// pipeline — so every child row the new run wrote hung off a record
	// describing a different run of a different job. A build that refuses to
	// start is a bad afternoon; that was silent history corruption, and
	// nothing downstream could tell it had happened.
	ErrRunExists = errors.New("a run with this id already exists")
	// ErrNoSuchRun is a RESUME of, or a run-scoped write to, a run this pipeline does not have.
	ErrNoSuchRun = errors.New("no run with this id")
)
View Source
var ErrApprovalNotPending = errors.New("approval is not pending")

ErrApprovalNotPending is a decision aimed at a row that no longer waits — a double click, a second approver, or the deadline getting there first. A sentinel so every caller can settle on the decision already recorded while still reporting a write that failed as a failure.

View Source
var ErrMemoryRefused = errors.New("memory refused")

ErrMemoryRefused is a memory no driver will file: empty, or over a cap.

View Source
var ErrNoState = errors.New("state database has nothing recorded yet")

ErrNoState is a state database with no steps schema in it yet: one a writer is in the middle of creating, or a file somebody made with touch. Nothing has been recorded, which is an answer rather than a failure, and a driver distinguishes it from a version mismatch — telling an operator to delete the database their first run is this instant creating is the worst possible advice.

View Source
var ErrNoSuchPipeline = errors.New("no such pipeline in this state file")

ErrNoSuchPipeline is a name the state database has never heard of.

It is the reason opening for reading RESOLVES a pipeline rather than registering one: a read that invents its own subject answers "no job runs recorded" to a typo, which is what a pipeline that has never run says too.

View Source
var ErrQuestionNotPending = errors.New("question is not pending")

ErrQuestionNotPending is what every answer path gets when the row it aimed at has already resolved — a race between a person and a responder agent, or between a person and the deadline. A sentinel because the CLI and the web UI both report it as "somebody got there first" rather than as a failure.

View Source
var ErrSchemaVersion = errors.New("the state database was written by a different version of steps")

ErrSchemaVersion is a database some other build of steps wrote.

There is no migration path, deliberately: a driver refuses a schema it cannot write rather than upgrading it, and the answer is to delete the database. What this exists for is that somebody is TOLD — an unrecognised schema used to open fine, lack a column, fail every insert naming it, and leave a green build that had recorded nothing.

View Source
var ErrTooManyClients = errors.New("too many registered clients")

ErrTooManyClients is a registration past the limit.

Functions

func BoundedError

func BoundedError(text string) string

BoundedError caps a message at MaxStoredErrorBytes, keeping the head for the reason ErrText gives.

func CheckMemory

func CheckMemory(memory Memory) error

CheckMemory is what every driver refuses before writing, so the refusal is the contract's rather than one driver's.

func CheckMemoryScope

func CheckMemoryScope(scope string) error

CheckMemoryScope is the scope half of CheckMemory, for a reader to refuse before it lists: Postgres cannot even compare against a scope it could not store, so a scope only sqlite accepted would fail one driver's step and not the other's.

func ContentKey

func ContentKey(content []byte) string

ContentKey is the interning key for node_content: a hash OF the content bytes.

Its own hash rather than the node's, because a node hash folds in its parent — two byte-identical steps in different chains must share one row, and under a node hash they could not.

func DecodeOptions

func DecodeOptions(encoded string) ([]string, error)

DecodeOptions reads what EncodeOptions stored.

func DecodeVersion

func DecodeVersion(encoded VersionKey) (map[string]any, error)

DecodeVersion parses one stored version, keeping numbers as exact digits.

func DecodeVersions

func DecodeVersions(encoded []VersionKey) ([]map[string]any, error)

DecodeVersions decodes a stored check history in order — what ResourceVersionsJSON hands back, for the caller that inspects fields.

func EncodeOptions

func EncodeOptions(options []string) (string, error)

EncodeOptions stores the offered list as JSON. Always a JSON array, never NULL: "the model offered nothing" and "we did not record what it offered" are different facts, and an empty array says the first one.

func ErrText

func ErrText(err error) any

ErrText renders an error for a nullable error column: nil (NULL) when there was none, and never more than MaxStoredErrorBytes of text.

Truncated at the head for the reason a transcript is: the message opens with what failed and closes with the detail, and "could not record node X: constraint failed" is worth more than the tail of a shell script.

func MintIdentity

func MintIdentity() string

MintIdentity is a fresh database identity for a driver's Identity to keep. Lowercase hex because the id rides in a GCE label value, which allows nothing else of note.

func TruncateTranscript

func TruncateTranscript(transcript string) string

TruncateTranscript cuts a transcript to at most MaxTranscriptBytes while keeping it VALID JSON, by dropping whole events off the end.

The first version sliced the string at a byte offset and appended a plain-text notice. That produced `[{"type":"text","text":"ttt` + a bare newline — invalid JSON, so json.Unmarshal in the renderer failed and every over-cap transcript displayed as NO transcript at all, the exact opposite of keeping the head. It passed a test asserting only that the stored string was non-empty. Byte slicing could also cut a multi-byte rune in half.

Events are dropped from the TAIL: the task, the first tool calls and the early decisions are what explain what a step was doing, and a conversation long enough to hit the cap is one whose middle is mostly repetition.

Anything that does not parse as a JSON array is passed through untouched under a hard byte cap. This function's job is a storage bound, not validation — a caller that hands over something else has a bug of its own, and silently rewriting it would hide that.

func TruncateUTF8

func TruncateUTF8(s string, limit int) string

TruncateUTF8 cuts s to at most a byte limit without splitting a rune.

Types

type Accounts

type Accounts interface {
	// CreateSession records a signed-in browser. The token itself is never
	// stored, only its digest, so a copy of the database signs nobody in.
	CreateSession(ctx context.Context, session Session) error
	// FindSession is the live session with this digest; found is false for
	// one that never existed, was revoked, or has expired.
	FindSession(ctx context.Context, tokenHash string) (Session, bool, error)
	// RevokeSession ends a session at once, which is what signing out is.
	RevokeSession(ctx context.Context, tokenHash string) error
	// SaveUser records what a sign-in provider said about a person, replacing
	// what it said last time: organizations and groups are read at sign-in,
	// so a person removed from a team loses it at their next.
	SaveUser(ctx context.Context, user User) error
	// FindUser is the person an actor names, as last recorded.
	FindUser(ctx context.Context, actor string) (User, bool, error)
	// RegisterClient records an agent that registered itself to connect to
	// exports (RFC 7591), refusing one past limit: registration answers
	// anybody, so its table must not grow without bound. Before counting it
	// evicts clients holding no token that are older than UnusedClientTTL.
	RegisterClient(ctx context.Context, client OAuthClient, limit int) error
	FindClient(ctx context.Context, id string) (OAuthClient, bool, error)
	// IssueToken records a token by its digest, never the token.
	IssueToken(ctx context.Context, token OAuthToken) error
	// FindToken is the live token with this digest: unexpired, unrevoked.
	FindToken(ctx context.Context, tokenHash string) (OAuthToken, bool, error)
	// RevokeToken ends one token, which is what rotating a refresh token does
	// to the old one. revoked is false for a token already ended: of two
	// refreshes racing with one token, only the one that ended it may mint.
	RevokeToken(ctx context.Context, tokenHash string) (revoked bool, err error)
	// RevokeGrant ends every token a person gave one client for one export: disconnecting it.
	RevokeGrant(ctx context.Context, grant Grant) error
	// Grants are the live connections — a grant with a token still good —
	// newest first, narrowed by filter's non-empty fields.
	Grants(ctx context.Context, filter Grant) ([]Connection, error)
	Close() error
}

Accounts is what a daemon keeps about the people who sign in to it, beside the pipelines rather than inside one: a session is somebody's standing with the daemon, and no pipeline handle is the right scope for it. One per database, opened by the driver on its own (sqlite.OpenAccounts, postgres.OpenAccounts), the way Identity is.

type Activity

type Activity interface {
	// RecordActivity appends one entry and trims the whole log, across every
	// pipeline, to the newest limit entries (zero is no limit). The log is
	// one per database because who did what is a question about the daemon,
	// not about any one pipeline.
	RecordActivity(ctx context.Context, entry ActivityEntry, limit int) error
}

Activity is the log of what people did to a pipeline: set, trigger, approve, abort, pause and the rest. Automatic work is not here — a run a check started says so on the run — so routine polling cannot push a person's action out of a log bounded by count.

A row is written through the handle of the pipeline acted on and keeps that pipeline's NAME beside its id, so destroying a pipeline leaves the record of its destruction readable.

type ActivityEntry

type ActivityEntry struct {
	ID int64
	// Pipeline is the name it was done to, as it was named then.
	Pipeline string
	// Actor is who did it, actor.Actor's stored form; empty where nobody
	// signs in.
	Actor string
	// Action is the verb: set, trigger, rerun, abort, approve, reject,
	// answer, pause, unpause, release, rename, destroy, mcp.connect, mcp.test.
	Action string
	// Target is what the verb was done to: a job, a run id, a revision sha.
	Target string
	// Detail is a JSON object of anything else worth reading, never secrets.
	Detail string
	At     time.Time
}

ActivityEntry is one thing a person did.

type ActivityFilter

type ActivityFilter struct {
	// Pipeline matches the name an entry was recorded under, or the current
	// name of the pipeline it was recorded against, so a rename keeps its
	// history.
	Pipeline string
	Actor    string
	Action   string
	// Actions keeps entries with any of these actions; empty keeps all. Beside Action because a page about one thing — an export — asks for every verb that thing has.
	Actions []string
	// Target matches what the verb was done to: a job, a run, an export.
	Target string
	// Since keeps only entries at or after it; zero keeps all.
	Since time.Time
	// Before pages: only entries older than this id.
	Before int64
	Limit  int
}

ActivityFilter narrows Reader.Activity. Empty fields match everything.

type AgentUsage

type AgentUsage struct {
	RunID     string
	StepIndex int
	StepName  string
	JobName   string
	// Kind is UsageStep, UsageHook or UsageFix; empty records as UsageStep.
	Kind string
	// NodeHash is empty for a hook or a fix, which are never hashed: their
	// rows hang off the run alone.
	NodeHash string
	// StepID is the run_events row the spend is drawn on — a hook's own, a fix's task's — or 0 when it ran outside a published step.
	StepID       int64
	ModelReq     string
	ModelServed  string
	Prompt       int
	Completion   int
	Total        int
	Cached       int
	Reasoning    int
	CostUSD      *float64
	FinishReason string
	DurationMS   int64
	RawMeta      string
}

AgentUsage is one agent step's recorded spend, as it goes in and comes back out of agent_usage.

CostUSD is a pointer because absent and zero are different answers: only a CLI-backed agent reports a dollar figure at all, and rendering an unreported cost as $0.00 would make an unpriced run look free.

func (AgentUsage) UsageKind

func (u AgentUsage) UsageKind() string

UsageKind is the kind a row records: Kind, or UsageStep when unset.

type Approval

type Approval struct {
	ID          int64
	JobName     string
	Message     string
	Status      ApprovalStatus
	RequestedAt time.Time
	// DecidedAt is zero while the approval is pending.
	DecidedAt time.Time
	DecidedBy string
	Reason    string
}

Approval is one recorded request for a human decision, and what became of it. The row IS the audit trail: who approved a deploy, when, and why a rejection was a rejection, are exactly the facts someone needs to reconstruct later — and they must not depend on external chat history.

type ApprovalStatus

type ApprovalStatus string

ApprovalStatus is where an approval is: pending until somebody, or the deadline, decides it.

const (
	ApprovalPending  ApprovalStatus = "pending"
	ApprovalApproved ApprovalStatus = "approved"
	ApprovalRejected ApprovalStatus = "rejected"
	// ApprovalExpired is the deadline deciding: nobody did.
	ApprovalExpired ApprovalStatus = "expired"
)

The statuses an approval is recorded with.

type Approvals

type Approvals interface {
	RequestApproval(ctx context.Context, jobName, message string) (int64, error)
	// RequestApprovalFor is RequestApproval for one step of one run, and asking twice for the same pair is one request: a daemon that restarts while a run waits resumes it, the step asks again, and it must find the approval it already made — pending, or already granted while nobody was looking — instead of a second row nobody is waiting on. A refused or expired one is not found: a retry asks a person afresh. An empty runID is a request with no run to tie it to, and always new.
	RequestApprovalFor(ctx context.Context, runID, stepKey, jobName, message string) (int64, error)
	DecideApproval(ctx context.Context, id int64, status ApprovalStatus, by, reason string) error
	ApprovalStatus(ctx context.Context, id int64) (Approval, error)
	// Approvals is the audit trail newest first, capped by limit (zero means
	// no limit). pendingOnly lists only what is still waiting, OLDEST first —
	// the order to work through them in — and its callers pass zero, because
	// a waiting build that scrolled off the list waits forever.
	Approvals(ctx context.Context, pendingOnly bool, limit int) ([]Approval, error)
}

Approvals is a build waiting on a person for PERMISSION.

Separate from Questions, which waits on a person for a FACT: an approval gates work and its row is the audit trail for having allowed it, and folding the two together would file "which environment?" as a decision somebody authorised.

type Blobs

type Blobs interface {
	RecordStepBlobs(ctx context.Context, actionKey string, outputs map[string]string) error
	StepBlobs(ctx context.Context, actionKey string) (map[string]string, error)
}

Blobs is the step cache's durable half: what a cached step's outputs were, by action key, so a skip can restore them instead of re-earning them.

The index is truth and the bytes live elsewhere — internal/workspace declares this same pair as StepIndex, because its depguard allow-list names neither this package nor a blob store, so both halves reach it as interfaces it declares and internal/cli wires together.

type Cache

type Cache interface {
	// RecordChainSucceeded adds a whole chain to the skip index.
	// ForgetChain removes it: a chain that ran again and failed — a --force
	// rerun, a hook — must not be skipped on the strength of the older
	// success. Failure is a removal rather than a row because the index
	// holds only what a rerun may skip; a failed run's classification lives
	// on its run and its nodes, and a job failing through fresh content
	// every time must not be able to crowd the green chains out of a
	// count-capped table.
	RecordChainSucceeded(ctx context.Context, jobName, rootHash string) error
	ForgetChain(ctx context.Context, jobName, rootHash string) error
	HasSucceededBatch(ctx context.Context, jobName string, rootHashes []string) (map[string]bool, error)
	RecordNode(ctx context.Context, node NodeRecord, jobName string, status NodeStatus, result map[string]any, execErr error) error
	HasNodeSucceeded(ctx context.Context, jobName, hash string) (bool, error)
	// HasNodesSucceeded is HasNodeSucceeded for many hashes in one read; a
	// hash that has not succeeded is absent rather than false.
	HasNodesSucceeded(ctx context.Context, jobName string, hashes []string) (map[string]bool, error)
	ListNodes(ctx context.Context, jobName string, limit int) ([]NodeRow, error)
	NodesByHash(ctx context.Context, hashes []string) (map[string]NodeRow, error)
	// SaveNodeTranscript replaces the transcript of a node, truncated to
	// MaxTranscriptBytes along whole events so it stays valid JSON.
	SaveNodeTranscript(ctx context.Context, hash, transcript string) error
	NodeTranscript(ctx context.Context, hash string) (string, bool, error)
}

Cache is the merkle skip index — the nodes and job_runs a rerun consults to decide what it does not have to do again — plus the transcript a cached agent step is still readable through.

Bounded by COUNT and never by age: a fully-cached poll records no new node and refreshes no timestamp, so an age floor sweeps past a working pipeline's cache, the faster it polls the sooner it loses it.

type CheckError

type CheckError struct {
	Name     string
	Message  string
	FailedAt time.Time
}

CheckError is a resource whose last check failed, and what it said.

Recorded because a failing check is otherwise SILENT: the poll aborts on the first resource that errors, the recorded version stays where it was, and the only trace is a log line on the machine running the daemon.

type CheckedResource

type CheckedResource struct {
	Name      string
	Version   VersionKey
	CheckedAt time.Time
}

CheckedResource is a resource's last observed version.

type CheckedSwap

type CheckedSwap struct {
	Resource      string
	Expected      VersionKey
	ExpectedFound bool
	Next          VersionKey
}

CheckedSwap moves a resource's checked version from Expected to Next. ExpectedFound false expects no checked version at all, and Expected is then ignored.

type ClaimedJob

type ClaimedJob struct {
	ID      int64
	JobName string
	// Manual is a person's trigger, which the breaker does not hold back.
	Manual bool
	// RerunOf is the run this one retries against the versions it was
	// created with, empty for any other trigger.
	RerunOf string
	// Resume is the run a restart interrupted, which this claim continues rather than starting a new one; empty for any other trigger.
	Resume string
}

ClaimedJob is a queue row a worker has just taken: its job, and what the trigger asks for beyond it.

type Connection

type Connection struct {
	Grant
	// Client is what the agent called itself, and RedirectURIs where it said to send people back.
	Client       string
	RedirectURIs []string
	// AuthorizedAt is when the person first agreed, the oldest of the grant's live tokens.
	AuthorizedAt time.Time
}

Connection is a live grant and the client it was given to.

type Control

type Control interface {
	// Pause is the circuit breaker for the whole pipeline: nothing polls, nothing is admitted, until Unpause.
	Pause(ctx context.Context) error
	Unpause(ctx context.Context) error
	Paused(ctx context.Context) (bool, error)
	// Rename is one UPDATE and keeps history; the handle goes on answering under its old name, so a caller closes it and re-opens under the new one.
	Rename(ctx context.Context, name string) error
	// Delete forgets the pipeline and everything that cascades off it; the handle is spent afterwards and must be closed.
	Delete(ctx context.Context) error
	// SetSourcePath records where this pipeline's YAML lives, so a reader of
	// the database can tell two checkouts apart. Only a command that LOADED
	// the pipeline calls it.
	SetSourcePath(ctx context.Context, source string) error
}

Control is the pipeline-level switchboard a `steps pipeline` verb throws: whether it admits work, what it is called, where its YAML lives, and whether it exists at all.

type Credentials added in v0.2.0

type Credentials interface {
	// Credential is the sealed value saved under name; found is false when there is none.
	Credential(ctx context.Context, name string) (sealed []byte, found bool, err error)
	// SaveCredential replaces whatever name held.
	SaveCredential(ctx context.Context, name string, sealed []byte) error
	// CredentialScope is what a sealed value is bound to, so a row copied onto another pipeline does not open: stable across a rename, never reused after a destroy.
	CredentialScope() string
}

Credentials is what a pipeline keeps secret: bytes SEALED above the store (internal/sealbox), which the store holds and cannot read. A copy of the database is a copy of ciphertext.

type CrossRunRow

type CrossRunRow struct {
	RunRow

	Pipeline string
}

CrossRunRow is a run plus the pipeline it belongs to. The pipeline is not optional: a feed that spans pipelines and cannot say which one a row came from is showing runs it has no route to.

type Deliveries

type Deliveries interface {
	// RecordDelivery files a delivery as the newest version of resourceName, keeps its payload, and dispatches it, all in one transaction — a sender that was answered 2xx has had all of it happen. A version already recorded (a redelivery) changes nothing and reports false.
	RecordDelivery(ctx context.Context, resourceName string, delivery Delivery, dispatch Dispatch, limit int) (bool, error)
	// Delivery is the payload recorded with a version; found is false once version_history: has pruned it.
	Delivery(ctx context.Context, resourceName string, version VersionKey) (Delivery, bool, error)
}

Deliveries is what a webhook resource receives: each delivery is a version, and its payload is what a get of that version writes out.

type Delivery

type Delivery struct {
	Version map[string]any
	Body    []byte
	Headers map[string]string
}

Delivery is one webhook delivery as recorded.

type Dispatch

type Dispatch struct {
	Jobs []string
	Held bool
	// By is who the queued rows say asked (actor.Actor's stored form).
	By string
}

Dispatch is what recording a delivery sets in motion: the jobs it queues, and the resource's current version moving to it. Held records the delivery and does neither — a paused pipeline — so the first poll after unpause finds a current version behind the newest delivery and dispatches it then.

type Events

type Events interface {
	AppendRunEvent(ctx context.Context, row RunEventRow) error
	RunEvents(ctx context.Context, runID string, afterSeq int64, limit int) ([]RunEventRow, error)
	// MaxStepID is the highest display-tree id a run's events carry, 0 for none: a resumed run mints past it, or its steps fold into the attempt before.
	MaxStepID(ctx context.Context, runID string) (int64, error)
	// DeleteStepEvents removes every event of one type a step recorded, which is how a step's output chunks make way for its one output record (events.TypeStepOutputChunk).
	DeleteStepEvents(ctx context.Context, runID string, stepID int64, eventType string) error
}

Events is what the web UI replays and streams.

type Grant

type Grant struct {
	ClientID string
	Actor    string
	Pipeline string
	Export   string
}

Grant is one person letting one client use one export.

type HeldJob

type HeldJob struct {
	Name        string
	Consecutive int
	HeldAt      time.Time
}

HeldJob is one job the breaker has stopped, with the count that stopped it.

type Memories

type Memories interface {
	// Remember files text under memory.Scope as its newest entry, then evicts
	// the oldest past limit (zero means no limit; a negative limit caps at
	// DefaultMemoryEntries). Text the scope already holds is not filed twice:
	// the entry already there comes back, reporting false.
	Remember(ctx context.Context, memory Memory, limit int) (Memory, bool, error)
	// ListMemories is scope's entries, newest first, capped by limit (zero
	// means no limit).
	ListMemories(ctx context.Context, scope string, limit int) ([]Memory, error)
	// Forget deletes one entry of scope. An id that is another scope's reports
	// false and deletes nothing — the id comes from a model.
	Forget(ctx context.Context, scope string, id int64) (bool, error)
	// ForgetScope deletes every entry of scope, reporting how many went.
	ForgetScope(ctx context.Context, scope string) (int, error)
	// MemoryScopes is every scope holding an entry, by name.
	MemoryScopes(ctx context.Context) ([]MemoryScope, error)
}

Memories is what an agent step keeps about one scope — a person, usually — from one run to the next: short facts a model chose to remember, handed to the next run that names the same scope.

The scope is the caller's, never the model's: every method takes it, and none reaches across it, which is what keeps one user's memory away from another's when a model is talked into asking.

type Memory

type Memory struct {
	ID    int64
	Scope string
	Text  string
	// RunID is the run that remembered it, "" once run_history: has reaped
	// that run — the memory outlives it, which is the point of keeping one.
	RunID     string
	CreatedAt time.Time
}

Memory is one remembered fact.

type MemoryScope

type MemoryScope struct {
	Scope   string
	Entries int
	// LastAt is when its newest entry was remembered.
	LastAt time.Time
}

MemoryScope is one scope's summary, for listing what is kept without reading all of it.

type Meta

type Meta interface {
	// Pipeline is the name this handle is scoped to.
	Pipeline() string

	// Description identifies the database behind this handle, in a form safe
	// to print: the file path for SQLite, a credential-free URL for a network
	// database. It is stable, so two handles reporting the same description
	// share a database — which is how a caller groups handles before asking
	// one Reader about several pipelines at once.
	Description() string

	// Reader reads across every pipeline in this database, unscoped by
	// construction.
	Reader() Reader

	// Close releases the connection, after whatever reclaim the driver owes
	// for what retention deleted. A handle obtained by RESOLVING a pipeline
	// rather than registering one must not write here: a command whose whole
	// contract is that it only asks changes no bytes.
	Close() error
}

Meta is the handle itself: what it is scoped to, what it is backed by, and how it is let go. Every other facet is a table; this one is the identity the tables are read through.

type NewRun

type NewRun struct {
	ID      string
	JobName string
	// Workspace is the directory the run's steps execute in, and what a
	// resume continues from.
	Workspace string
	// ConfigSHA is the configuration the run executes, empty for a caller
	// that loaded no pipeline file.
	ConfigSHA string
	// QueueID is the trigger_queue row that started this run, zero for a run no queue started; recorded on that row, so a restart knows which run to resume.
	QueueID int64
	// CreatedBy is who started a run no queue row did. A queued run takes its row's requester instead, read in the same statement that records the run, so nothing between the claim and the start can lose it.
	CreatedBy string
}

NewRun is what StartRun mints a run from.

type NodeRecord

type NodeRecord struct {
	Hash       string
	ParentHash string
	Kind       string
	StepIndex  int
	Resource   string // resource name (get/put) or task name (task); metadata only
	Content    map[string]any
}

NodeRecord is the subset of a merkle plan node's fields a driver persists. It's a plain data shape rather than an import of merkle.Node so this leaf package doesn't need to depend on the planner — callers convert their own Node type into one of these.

type NodeRow

type NodeRow struct {
	Hash      string
	Kind      string
	JobName   string
	Resource  string
	StepIndex int
	Status    NodeStatus
	Error     string
	Result    string
	CreatedAt time.Time
	// Content and ParentHash are what the hash is MADE of — populated by
	// NodesByHash (the node-detail read), left empty by the list queries,
	// whose callers want a table row rather than a whole content map.
	Content    string
	ParentHash string
}

NodeRow is one recorded step, with whatever the step produced.

type NodeStatus added in v0.2.0

type NodeStatus string

NodeStatus is how a step's node ended: running while an agent step is in flight, otherwise the outcome.Class the step earned.

const (
	NodeRunning   NodeStatus = "running"
	NodeSucceeded NodeStatus = "succeeded"
	NodeFailed    NodeStatus = "failed"
	NodeErrored   NodeStatus = "errored"
	NodeAborted   NodeStatus = "aborted"
)

The statuses a node is stored with; they equal outcome.Class's strings, which store cannot import.

type OAuthClient

type OAuthClient struct {
	ID string
	// Name is what the client calls itself, which nothing verifies.
	Name         string
	RedirectURIs []string
	CreatedAt    time.Time
}

OAuthClient is an agent that registered itself to connect to exports.

type OAuthToken

type OAuthToken struct {
	TokenHash string
	Kind      TokenKind
	Grant     Grant
	CreatedAt time.Time
	ExpiresAt time.Time
}

OAuthToken is a token issued under a grant.

type Pass

type Pass struct {
	JobName  string
	Resource string
	Version  VersionKey
	BuildID  string
}

Pass is a version a job went green against, and the build it went green in: a downstream fan-in asks whether its versions passed in ONE build.

type PassedVersion

type PassedVersion struct {
	Resource   string
	Version    VersionKey
	RecordedAt time.Time
}

PassedVersion is one resource version a job succeeded against.

type PipelineRow

type PipelineRow struct {
	Name string
	Path string
	// CurrentSHA is the revision a `steps pipeline set` made current, empty for a pipeline that only ever ran by hand; a daemon serves exactly the rows that have one.
	CurrentSHA string
	Paused     bool
}

PipelineRow is one row of the pipelines table: what a state database actually holds, which is the question a shared one created and nothing asked.

type Placement

type Placement struct {
	RunID     string
	StepIndex int
	StepName  string
	JobName   string
	// Slot is what makes this placement unique within its run: a plan step's
	// node hash, or a hook's scope label. A hook is not merkle-hashed, so it
	// has no node to be keyed on.
	Slot string
	// NodeHash is the node this placement belongs to, EMPTY for a hook. It is
	// what lets retention reaping a node take the placement with it; a hook
	// row has no node and cascades off its run instead.
	//
	// A string rather than a pointer, unlike InstanceID and UID beside it: a
	// hash is never legitimately empty, so there is no zero value to confuse
	// with absence, and a node's parent hash already spells the same
	// distinction this way.
	NodeHash   string
	Tag        string
	Address    string
	InstanceID *string
	GOOS       string
	GOARCH     string
	Workdir    string
	UID        *int
	GID        *int
	Image      string
	BytesSent  int64
	// BytesReceived is the tree that came back from that machine. Together
	// with BytesSent it is the whole cost of the placement in bytes, and
	// the pair is what says whether a produced tree ever had to leave the
	// worker at all.
	BytesReceived int64
	// VolumeBytes is what the step's own volumes held on the worker when it ended: its outputs, its overlays' upper layers, and any tree poured into the worker's cache for it — never a cached tree it only read. Nil where nothing weighed them — a worker that keeps no volumes (ssh://), a step that held no output there (the weighing rides in that output's digest run) — which is not the same answer as zero.
	VolumeBytes *int64
}

Placement is one placed step's record of the machine that ran it.

InstanceID, UID and GID are pointers because absent and zero are different answers. Only a cloud worker has an instance at all; only a worker that can answer reports an identity, and uid 0 is ROOT, so an int could not tell "ran as root" from "did not say", and those mean opposite things to a reader.

type Placements

type Placements interface {
	RecordPlacement(ctx context.Context, placement Placement) error
	// RunPlacements returns a run's placements in plan order.
	RunPlacements(ctx context.Context, runID string) ([]Placement, error)
}

Placements is which machine ran a step, and what it found there.

type Pruning

type Pruning interface {
	// Prune applies one retention policy at the end of a build: the job's runs
	// and its finished queue rows are capped by COUNT, newest kept, and a
	// reaped run takes its events, steps, usage, placements, questions and
	// inputs with it. keepRunID is never reaped, however the counts fall.
	//
	// Newest is by start time, with insertion order breaking a tie: two runs
	// can land in the same instant, and the one to keep is the one recorded
	// last.
	//
	// The merkle caches are bounded here too, and by count alone — see Cache.
	// A transcript hangs off the node cache, not the run, and goes with its
	// node. Every call also sweeps the configuration revisions nothing refers
	// to any more; a wholly zero Retention is that sweep alone.
	Prune(ctx context.Context, policy Retention, keepRunID string) error
}

Pruning is the one place anything is DELETED, named for the verb because Retention beside it is the policy the verb takes.

type Question

type Question struct {
	ID      int64
	RunID   string
	JobName string
	// AgentName is the agent as the step KNOWS it (across: cells report under
	// their labelled identity), so a parked question names something a reader
	// can find in the plan.
	AgentName string
	Question  string
	Options   []string
	// OptionsRequired is recorded rather than only enforced in the asking
	// process, because the process that ANSWERS is usually a different one:
	// `steps questions answer` and the web UI both write this row directly, and a fence
	// only the asker knew about would be no fence on either of them.
	OptionsRequired bool
	Default         string
	Status          QuestionStatus
	AskedAt         time.Time
	// AnsweredAt is zero while the question is pending.
	AnsweredAt time.Time
	AnsweredBy string
	Answer     string
}

Question is one recorded ask_user call, and what became of it. Like an approval, the row IS the audit trail — but it records a FACT rather than a permission, so it carries the things an approval has no use for: the options the model offered, the default it declared, who or what answered, and the memo key that makes one question asked by twelve matrix cells one question.

func (Question) MemoKey

func (q Question) MemoKey() string

MemoKey identifies a question WITHIN one run: the same text offering the same options is the same question, however many steps ask it.

The options are part of the key, not decoration. "Which environment?" over {staging, prod} and the same sentence over {staging, prod, canary} are two different asks, and answering the second with the first's answer would be the runtime deciding a question nobody put that way.

Hashed rather than stored raw because it is a unique-index key and a question may be a paragraph; the text itself is in the row beside it.

type QuestionStatus

type QuestionStatus string

QuestionStatus is where a question is: pending until a person answers it or one of the endings nobody answered closes it.

const (
	QuestionPending  QuestionStatus = "pending"
	QuestionAnswered QuestionStatus = "answered"
	// QuestionExpired is the deadline passing, with the default standing in.
	QuestionExpired QuestionStatus = "expired"
	// QuestionAborted is the run asking it ending first.
	QuestionAborted QuestionStatus = "aborted"
)

The statuses a question is recorded with.

type Questions

type Questions interface {
	// AskQuestion records a pending question, or returns the one this run
	// already asked under the same memo key — reporting false when it did.
	// The memo is the row, so twelve across: cells asking the same thing all
	// reach one row rather than one person twelve times.
	AskQuestion(ctx context.Context, question Question) (Question, bool, error)
	// AnswerQuestion is a person's answer, ErrQuestionNotPending if somebody
	// or something got there first.
	AnswerQuestion(ctx context.Context, id int64, answer, by string) error
	// CloseQuestion resolves a question with a status the caller chooses, for
	// the endings nobody answered.
	CloseQuestion(ctx context.Context, id int64, status QuestionStatus, answer, by string) error
	QuestionStatus(ctx context.Context, id int64) (Question, error)
	// Questions is the audit trail, what is still waiting first and then
	// newest first, capped by limit (zero means no limit). pendingOnly lists
	// only what is waiting, OLDEST first, and its callers pass zero — for the
	// same reason Approvals' do.
	Questions(ctx context.Context, pendingOnly bool, limit int) ([]Question, error)
}

Questions is a build waiting on a person for a FACT — the ask_user tool and the two ways a question ends. See Approvals for why the two stay apart.

type Queue

type Queue interface {
	// EnqueueJob and its two siblings take by, the stored form of who or what asked (actor.Actor.String), which the run the row starts inherits.
	EnqueueJob(ctx context.Context, jobName, reason, by string) error
	// EnqueueManualJob is EnqueueJob for a person's trigger: the row is marked manual, and a pending row it merges into becomes manual too, because the breaker holds back only what was triggered automatically. The merged row takes this person as its requester, since they are why it is manual.
	EnqueueManualJob(ctx context.Context, jobName, reason, by string) error
	// EnqueueRerunJob queues a retry of a recorded run. It is a person's request, so it is manual; it is not merged into the job's ordinary pending row, since it builds different versions.
	EnqueueRerunJob(ctx context.Context, jobName, reason, runID, by string) error
	// ClaimNextJob takes the oldest pending row admitted by serial: and
	// max_in_flight, atomically, and reports false when nothing is ready. The
	// row comes back from the statement that claimed it, so nothing can be
	// read between the claim and what the trigger asked for.
	ClaimNextJob(ctx context.Context) (ClaimedJob, bool, error)
	CompleteJob(ctx context.Context, id int64, status QueueStatus, runErr error) error
	// AbortQueuedJob finalizes the job's pending row as aborted so it is never claimed, reporting false when nothing was pending. A running row is not its to touch: that build is stopped through its context, and holds its serial slot until it actually ends.
	AbortQueuedJob(ctx context.Context, jobName string) (bool, error)
	// ResetStaleRunning reclaims every running row as an abandoned leftover.
	// It is only true when nothing else is alive, which is why steps runs one
	// daemon per database.
	// A row whose run is still running goes back to pending carrying that run (ClaimedJob.Resume); a row whose run already finished is closed with the run's status, so a run somebody aborted is never built again.
	ResetStaleRunning(ctx context.Context) error
	ListTriggerQueue(ctx context.Context, limit int) ([]QueueRow, error)
	// SyncJobLimits replaces both admission mirrors from one configuration, in
	// one transaction. A job in neither map has no serial group and is
	// admitted one at a time: a missing row means the job left the pipeline
	// between enqueue and claim, and serializing what nobody can describe is
	// the conservative reading.
	SyncJobLimits(ctx context.Context, groups map[string][]string, limits map[string]int) error
	SerialGroupHolder(ctx context.Context, jobName string) (string, error)
	RecordJobOutcome(ctx context.Context, jobName string, succeeded bool, maxFailures int) (held bool, consecutive int, err error)
	ResetJobFailures(ctx context.Context, jobName string) error
	IsJobHeld(ctx context.Context, jobName string) (bool, error)
	HeldJobs(ctx context.Context) ([]HeldJob, error)
}

Queue is the work list one daemon drains, and the breaker that holds a job that keeps failing. Held is the breaker's word, for one job; paused is Control's, for the whole pipeline.

The breaker is here rather than beside it because it decides what the queue may hand out: a held job is a row that will not be claimed, so splitting them would give a consumer the queue without the rule that governs it.

type QueueRow

type QueueRow struct {
	ID         int64
	JobName    string
	Reason     string
	Status     QueueStatus
	EnqueuedAt time.Time
	StartedAt  time.Time
	FinishedAt time.Time
	Error      string
}

QueueRow is one entry in the downstream-trigger queue.

type QueueStatus added in v0.2.0

type QueueStatus string

QueueStatus is where a trigger-queue row is: pending until claimed, running until CompleteJob (or ResetStaleRunning, which copies the run's status) closes it.

const (
	QueuePending   QueueStatus = "pending"
	QueueRunning   QueueStatus = "running"
	QueueSucceeded QueueStatus = "succeeded"
	QueueFailed    QueueStatus = "failed"
	QueueAborted   QueueStatus = "aborted"
	// QueueSkipped is a row the circuit breaker took out of the rotation without running it.
	QueueSkipped QueueStatus = "skipped"
)

The statuses a queue row is stored with; the SQL in both drivers spells the same strings.

type Reader

type Reader interface {
	// Pipelines is what this database actually holds.
	Pipelines(ctx context.Context) ([]PipelineRow, error)
	// RecentRuns interleaves the named pipelines' runs, newest first. A name
	// the database has never heard of contributes nothing rather than failing:
	// the caller asked about a set, not about each one.
	RecentRuns(ctx context.Context, pipelines []string, limit int) ([]CrossRunRow, error)
	// Activity is the log of what people did, newest first, across every
	// pipeline — including ones since destroyed, by the name they had.
	Activity(ctx context.Context, filter ActivityFilter) ([]ActivityEntry, error)
	// Close releases the connection, unless this Reader borrowed one from a
	// Store — in which case closing is the Store's to do.
	Close() error
}

Reader reads ACROSS the pipelines in one database.

Every read on a Store is scoped: the handle holds its pipeline and every method filters by it, which is what makes forgetting to scope impossible. That property is worth keeping exactly as it is, so a cross-pipeline read is a different TYPE rather than a nullable parameter threaded through the scoped methods — the alternative puts the footgun back, and puts it in the one place merkle hashing makes it dangerous.

A Reader can therefore answer only questions that name their pipelines out loud. It cannot be handed to code expecting a Store, and there is no method here that silently means "all of them".

type Retention

type Retention struct {
	// JobName is the job whose history is being bounded. Per job rather than
	// globally, so one busy job cannot evict a quiet one's only run — a global
	// cap makes the least active job the least inspectable, which is backwards.
	JobName string
	// Runs is how many of that job's runs to keep, newest first.
	Runs int
	// TriggerQueue is how many of its FINISHED queue rows to keep. A separate
	// number because the queue is a work list rather than history: see
	// DefaultTriggerQueueHistory, which is what the run passes.
	TriggerQueue int
}

Retention is what one build is allowed to leave behind: a policy read off the configuration, applied once, at the end.

Zero means no limit in both fields, the convention this repo documents in docs/attempts-timeout.md. A wholly zero Retention is therefore the configuration sweep on its own — what a reload wants, having just adopted a configuration and orphaned the one it replaced, with no job in play and no run to reap.

type Revision

type Revision struct {
	SHA      string
	Source   string
	Includes map[string]string
}

Revision is one recorded configuration: the substituted source a run was started from, the includes it carried, and the hash that identifies both.

type Revisions

type Revisions interface {
	// RecordRevision interns a configuration with the include contents it folded in; the same sha twice is one row.
	RecordRevision(ctx context.Context, sha, source string, includes map[string]string) error
	FindRevision(ctx context.Context, sha string) (Revision, bool, error)
	// SetCurrentRevision names the recorded revision a `steps pipeline set` made current and, in the same write, records from as the pipeline's source path, so the two cannot disagree; the current one is never reaped.
	SetCurrentRevision(ctx context.Context, sha, from string) error
	// CurrentRevision is what a daemon serves for this pipeline at startup; false for a pipeline that only ever ran by hand.
	CurrentRevision(ctx context.Context) (Revision, bool, error)
}

Revisions is the configuration a run was executed under, interned once by its hash, and which of them a daemon currently serves.

type RunEventRow

type RunEventRow struct {
	Seq       int64
	RunID     string
	Type      string
	StepIndex int
	StepName  string
	StepKind  string
	// StepID/ParentStepID carry the display tree — see events.Event, which
	// documents why it is minted rather than read off the merkle chain.
	StepID       int64
	ParentStepID int64
	Status       string
	Hash         string
	Text         string
	Name         string
	Detail       string
	DurationMS   int64
	// Worker is where a placed step ran; empty for this machine. See
	// events.Event.Worker.
	Worker string
	At     time.Time
}

RunEventRow is one persisted run event — the stored form of events.Event, which is what a finished run is replayed from.

type RunInput

type RunInput struct {
	Input    string
	Resource string
	Version  VersionKey
}

RunInput is a version a run was created with, keyed by the get it was bound under: two gets of one resource in a run each keep their own.

type RunRow

type RunRow struct {
	ID         string
	JobName    string
	Workspace  string
	Status     RunStatus
	StartedAt  time.Time
	FinishedAt time.Time
	// ParentRunID is the run a replay forked from, empty for an ordinary run.
	// It is what lets a tuning session read as a session rather than as a pile
	// of unrelated runs that happen to share a job.
	//
	// Filled by EVERY query that builds a RunRow, deliberately: it was briefly
	// selected by only one of them, which made Replayed() answer differently
	// depending on which call site had loaded the row — the jobs list said a
	// forked run was ordinary while its own page linked its parent. That is
	// what the driver's one shared column list exists to make impossible.
	ParentRunID string
	// ConfigSHA is the configuration this run executed, empty for a run
	// started by a caller that loaded no pipeline file. Selected by every
	// RunRow query for the same reason ParentRunID is.
	ConfigSHA string
	// RerunOf is the run this one re-ran against the versions it was created with, empty for any other run. Selected by every RunRow query for the same reason ParentRunID is.
	RerunOf string
	// CreatedBy is the stored form of who or what started the run (actor.Parse reads it), empty for a run recorded before anyone was. Selected by every RunRow query for the same reason ParentRunID is.
	CreatedBy string
}

RunRow is one run invocation as the history views read it: the resume record plus the finish timestamp that makes a duration answerable.

func (RunRow) Duration

func (r RunRow) Duration() time.Duration

Duration is how long the run took, or how long it has been going when it has not finished. Zero when the run never started.

func (RunRow) Replayed

func (r RunRow) Replayed() bool

Replayed reports a run forked from another by --replay.

func (RunRow) Rerun

func (r RunRow) Rerun() bool

Rerun reports a run made by retrying another.

type RunStatus

type RunStatus string

RunStatus is where a run is: running until FinishRun records how it ended. There is no errored: a run the machinery broke is recorded failed, and only a step's own status tells the two apart.

const (
	RunRunning   RunStatus = "running"
	RunSucceeded RunStatus = "succeeded"
	RunFailed    RunStatus = "failed"
	RunAborted   RunStatus = "aborted"
)

The statuses a run is recorded with.

type RunStep

type RunStep struct {
	BuildID string
	Index   int
	Name    string
}

RunStep is one step a run finished. Index is relative to the walk it ran in — the walk after a get counts from zero again — so it names a step only together with BuildID.

type RunTotals

type RunTotals struct {
	RunID    string
	Tokens   int
	Cached   int
	CostUSD  *float64
	Steps    int
	Unpriced int
}

RunTotals is the per-run rollup `steps runs cost` lists.

type Runs

type Runs interface {
	// StartRun mints a run. An id some run already holds is ErrRunExists,
	// never an overwrite.
	StartRun(ctx context.Context, run NewRun) error
	// ResumeRun continues a run this pipeline already has, ErrNoSuchRun
	// otherwise.
	ResumeRun(ctx context.Context, id, workspaceDir, configSHA string) error
	// ResumableRuns is every run of this pipeline still recorded running, or that a queue row still waits to resume (a restart before its resume started): at startup, the ones a restart may resume, whose workspaces must survive the stale-build sweep.
	ResumableRuns(ctx context.Context) ([]RunRow, error)
	FinishRun(ctx context.Context, id string, status RunStatus) error
	// RecordRunStep marks a step of a run done. buildID is "<run>#0" inside
	// the build a get triggers and the bare run id before it.
	RecordRunStep(ctx context.Context, runID string, step RunStep) error
	RecordRunParent(ctx context.Context, runID, parentID string) error
	// RecordRunRerun marks a run as a rerun of another (fly rerun-build).
	RecordRunRerun(ctx context.Context, runID, originalID string) error
	// CompletedRunSteps returns what a run finished, in completion order.
	CompletedRunSteps(ctx context.Context, runID string) ([]RunStep, error)
	// ListRuns returns a job's runs newest first. Zero means no limit, the
	// convention everywhere here.
	ListRuns(ctx context.Context, jobName string, limit int) ([]RunRow, error)
	LatestRunByJob(ctx context.Context) (map[string]RunRow, error)
	// LatestFinishedRunByJob is each job's newest run that is no longer
	// running, reruns included, as ListRuns includes them.
	LatestFinishedRunByJob(ctx context.Context) (map[string]RunRow, error)
	RunsUsingNode(ctx context.Context, hash string, limit int) ([]RunRow, error)
	FindRunRow(ctx context.Context, id string) (RunRow, bool, error)
	FirstRunSince(ctx context.Context, jobName string, since time.Time) (RunRow, bool, error)
}

Runs is one row per invocation: what --resume continues and what the history views list.

type Session

type Session struct {
	TokenHash string
	// Actor is who it signed in as, actor.Actor's stored form.
	Actor     string
	CreatedAt time.Time
	ExpiresAt time.Time
}

Session is one signed-in browser.

type Store

Store is one pipeline's view of the state database: every facet at once.

The handle IS the scope, and that is the property the interface exists to preserve: a state file (or a Postgres database) may hold several pipelines, and no method here takes a pipeline name. A caller cannot forget to pass one and cannot pass the wrong one. It matters most for the cache — a merkle hash folds in kind, content and parent but NOT the pipeline, so two pipelines each with a job named `build` over an identical task hash the same, and an unscoped skip index would let the second one skip work it never did.

`steps web` serving three pipelines from one database holds three of these. Reading ACROSS them is Reader, which is a different type for the same reason.

It is a composition and NOTHING else — facets_test.go holds it to that. The facets are the aggregates the files here are already cut along, and they exist so a consumer can name the parts of the database it touches and no more; only code that runs a whole build needs all of it. A method declared straight on Store would belong to no aggregate, so nobody could ask for it by name and a second driver would have no facet to write it under.

type TokenKind added in v0.2.0

type TokenKind string

TokenKind says what an OAuth token may do: an access token calls tools, a refresh token only mints the next access token.

const (
	TokenAccess  TokenKind = "access"
	TokenRefresh TokenKind = "refresh"
)

The kinds a token is stored with.

type Usage

type Usage interface {
	RecordAgentUsage(ctx context.Context, usage AgentUsage) error
	RunTokensSpent(ctx context.Context, runID string) (int, error)
	RunUsage(ctx context.Context, runID string) ([]AgentUsage, error)
	RunCostTotals(ctx context.Context, limit int) ([]RunTotals, error)
}

Usage is the agent token and cost ledger.

type User

type User struct {
	// Actor is who they are to every record, actor.Actor's stored form.
	Actor string
	// AccountID is the provider's own id for the account, which survives a rename where the name in Actor does not: GitHub's numeric user id, Google's subject.
	AccountID string
	// Email is the address the provider gave, empty when it gave none.
	Email string
	// Organizations and Groups are what permission rules match on: GitHub
	// organizations and "org/team" teams, a Google Workspace domain.
	Organizations []string
	Groups        []string
	RefreshedAt   time.Time
}

User is a person a sign-in provider vouched for, as it last described them.

type VersionKey

type VersionKey string

VersionKey is a version as the canonical JSON every version table keys on.

A type rather than a string because two spellings of one version are two keys: a version marshalled some other way reaches a table as a version nobody recorded, and a passed: gate waits on it forever. EncodeVersion is the only constructor outside a driver; a driver hands back what it stored.

func EncodeVersion

func EncodeVersion(version map[string]any) (VersionKey, error)

EncodeVersion renders a version as its VersionKey. json.Marshal sorts map keys, so the same version always produces the same key.

type VersionRun

type VersionRun struct {
	Version VersionKey
	Run     RunRow
}

VersionRun is one run that took a version of a resource as an input.

type Versions

type Versions interface {
	// RecordVersions files what a check reported and prunes beyond limit,
	// returning how many were new to check-history. A version already filed
	// keeps its discovery order. Zero means no limit, as Retention's fields
	// do; a negative limit prunes at DefaultResourceVersionCap.
	RecordVersions(ctx context.Context, resourceName string, versions []map[string]any, limit int) (int, error)
	ResourceVersionsJSON(ctx context.Context, resourceName string) ([]VersionKey, error)
	VersionOrders(ctx context.Context, resourceName string) (map[VersionKey]int64, error)
	RecordVersionOrder(ctx context.Context, resourceName string, version VersionKey) (int64, error)
	GreenVersions(ctx context.Context, resourceName string, upstreamJobs []string) ([]map[string]any, error)
	RecordCheckedVersion(ctx context.Context, resourceName string, version VersionKey) error
	// CompareAndSetCheckedVersion moves the checked version to next only if it is still expected — or, with expectedFound false, still absent — reporting whether it moved. It is how a poll advances a resource something else may have advanced since the poll read it.
	CompareAndSetCheckedVersion(ctx context.Context, swap CheckedSwap) (bool, error)
	LastChecked(ctx context.Context, resourceName string) (CheckedResource, bool, error)
	CheckedResources(ctx context.Context) ([]CheckedResource, error)
	// RecordCheckError files why a resource's check failed; an empty message
	// clears what was filed. A check that errors leaves the recorded version
	// exactly where it was — which is correct, and which is why a failing
	// resource is otherwise indistinguishable from a quiet one.
	RecordCheckError(ctx context.Context, resourceName, message string) error
	// CheckErrors is every resource whose last check failed, by name. It is
	// kept apart from CheckedResources because the two answer different
	// questions about different rows: a resource that has never checked
	// successfully has an error and no version at all.
	CheckErrors(ctx context.Context) ([]CheckError, error)
	RecordPassedVersion(ctx context.Context, pass Pass) error
	// RecordPassedVersions is RecordPassedVersion for a whole build's
	// versions in one transaction: all of them land or none does.
	RecordPassedVersions(ctx context.Context, passes []Pass) error
	PassedVersions(ctx context.Context, jobName string, limit int) ([]PassedVersion, error)
	HasPassedVersionSet(ctx context.Context, jobName string, want map[string]VersionKey) (bool, error)
	ConsumedMark(ctx context.Context, jobName, resourceName string) (int64, error)
	RecordConsumedMark(ctx context.Context, jobName, resourceName string, order int64) error
	RecordRunInput(ctx context.Context, runID string, input RunInput) error
	RunInputs(ctx context.Context, runID string) ([]RunInput, error)
	// VersionRuns is every run that took one of versions of the resource as
	// an input, newest first. Naming the versions bounds the read by what a
	// caller shows rather than by run history, which zero leaves unbounded;
	// none named is none read.
	VersionRuns(ctx context.Context, resourceName string, versions []VersionKey) ([]VersionRun, error)
}

Versions is resource version history and the three marks kept against it: what a check SAW, what a job PASSED, and how far a job has CONSUMED.

One facet rather than two because the marks are meaningless without the history they point into — a passed row cascades off the version it names, so a cap on one is a cap on the other.

Directories

Path Synopsis
Package postgres is the Postgres driver for the state database: its own schema and its own SQL behind internal/store's contract, proven by the same conformance suite as the sqlite driver.
Package postgres is the Postgres driver for the state database: its own schema and its own SQL behind internal/store's contract, proven by the same conformance suite as the sqlite driver.
Package sqlite is the sqlite driver for the state database: the schema, the SQL and the connection pragmas behind internal/store's contract.
Package sqlite is the sqlite driver for the state database: the schema, the SQL and the connection pragmas behind internal/store's contract.
Package storetest is the store's conformance suite: the behavior tests written against internal/store's contract rather than against any one driver, so a second driver is proven by the same tests the first one passes.
Package storetest is the store's conformance suite: the behavior tests written against internal/store's contract rather than against any one driver, so a second driver is proven by the same tests the first one passes.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL