pipeline

package
v1.55.0 Latest Latest
Warning

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

 Go to latest Published: Sep 11, 2026 License: MIT Imports: 28 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrFatalGateReconciliation = errors.New("fatal gate reconciliation")

Functions

func BindUncertifiedPipelineRange 

func BindUncertifiedPipelineRange(sctx *StepContext) error

BindUncertifiedPipelineRange copies a persisted uncertified recovery boundary onto the review step context when this run's head is that range's tip or a descendant of it. Unreadable commit ancestry or persisted review truth blocks replacement review.

func FilterDeferredPipelineOwnedDeliveryFindings 

func FilterDeferredPipelineOwnedDeliveryFindings(findings types.Findings) (types.Findings, int)

func PersistUncertifiedPipelineRange 

func PersistUncertifiedPipelineRange(sctx *StepContext, fromSHA, toSHA string) error

PersistUncertifiedPipelineRange records a post-review commit span until a review of the new head completes.

func PersistUncertifiedPipelineRangeWithRollback 

func PersistUncertifiedPipelineRangeWithRollback(sctx *StepContext, fromSHA, toSHA string) (func() error, error)

func ReconcileReviewFindings 

func ReconcileReviewFindings(findings types.Findings, priorRaw string) (types.Findings, int, error)

func RemapUncertifiedPipelineRangeAfterRebase 

func RemapUncertifiedPipelineRangeAfterRebase(sctx *StepContext, oldHead, newHead string) (func() error, error)

RemapUncertifiedPipelineRangeAfterRebase rewrites a persisted uncertified range onto the new head when rebase replaced a head that contained it.

func ValidateRecoveredRun 

func ValidateRecoveredRun(database *db.DB, run *db.Run, steps []Step) error

Types

type ApprovalGateReconciler 

type ApprovalGateReconciler interface {
	ReconcileApprovalGate(sctx *StepContext) (resolved bool, err error)
}

ApprovalGateReconciler is implemented by a step whose parked approval gate can become obsolete when an external source of truth changes. The executor invokes it with a bounded context while also waiting for an approval. A true result completes the step through the normal success path; false or an error leaves the gate parked. Implementations must be read-only and fail closed.

type Capacity added in v1.55.0 

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

Capacity is shared by every executor in a host daemon. Review and suite execution occupy independent pools; approval waits occupy neither. Release closes a notification channel, so admission never depends on a polling timer.

func NewCapacity added in v1.55.0 

func NewCapacity(limits config.Concurrency) *Capacity

func (*Capacity) Acquire added in v1.55.0 

func (c *Capacity) Acquire(ctx context.Context, step types.StepName) (func(), error)

func (*Capacity) Configure added in v1.55.0 

func (c *Capacity) Configure(limits config.Concurrency)

Configure applies the operator config on run admission, including recovery. Lowering a cap drains existing holders; it never interrupts their work.

type EventFunc 

type EventFunc func(ipc.Event)

EventFunc is called when a pipeline event occurs, for streaming to subscribers.

type Executor 

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

Executor runs pipeline steps sequentially and coordinates approval interactions.

func NewExecutor 

func NewExecutor(database *db.DB, p *paths.Paths, cfg *config.Config, ag agent.Agent, steps []Step, onEvent EventFunc) *Executor

NewExecutor creates a pipeline executor.

func (*Executor) Execute 

func (e *Executor) Execute(ctx context.Context, run *db.Run, repo *db.Repo, workDir string) error

Execute runs the pipeline steps sequentially for a given run. The workDir is the directory where steps execute (typically a git worktree). If the context is cancelled with a cause (via context.WithCancelCause), the cause message is preserved as the run's error in the DB.

func (*Executor) Respond 

func (e *Executor) Respond(step types.StepName, action types.ApprovalAction, findingIDs []string) error

Respond sends a user approval action to the currently waiting step. The step parameter must match the step currently awaiting approval. Returns an error if no step is awaiting approval or if the step name doesn't match.

func (*Executor) RespondWithOverrides 

func (e *Executor) RespondWithOverrides(step types.StepName, action types.ApprovalAction, findingIDs []string, instructions map[string]string, addedFindings []types.Finding) error

RespondWithOverrides is like Respond but also carries per-finding user instructions and user-authored findings. Both are merged into the round's findings on a fix action before the fix agent runs.

func (*Executor) Resume 

func (e *Executor) Resume(ctx context.Context, run *db.Run, repo *db.Repo, workDir string) error

Resume restores a run that was durably parked at an approval gate when the daemon stopped. It only accepts a fully recorded gate and otherwise returns an error so startup recovery can fail the run rather than guessing.

func (*Executor) SetCapacity added in v1.55.0 

func (e *Executor) SetCapacity(c *Capacity)

func (*Executor) SetGateDir 

func (e *Executor) SetGateDir(gateDir string)

SetGateDir identifies the managed bare mirror where reviewed-tree evidence refs are retained. It is separate from the step worktree so developer refs are never polluted by gate custody metadata.

func (*Executor) SetGateReconcileTimings 

func (e *Executor) SetGateReconcileTimings(interval, timeout time.Duration)

SetGateReconcileTimings overrides the interval between approval-gate reconciliation checks and the deadline for each check. It is primarily used by deterministic tests and specialized embeddings; non-positive values keep the production defaults.

func (*Executor) SetOnPRMerged 

func (e *Executor) SetOnPRMerged(fn func(context.Context, string))

SetOnPRMerged registers a best-effort hook invoked after a merged PR state is persisted. The pipeline never fails the run if the hook errors.

func (*Executor) SetSkippedSteps 

func (e *Executor) SetSkippedSteps(steps []types.StepName)

SetSkippedSteps configures steps that should be marked skipped without running.

type HousekeepingLintResult 

type HousekeepingLintResult struct {
	// FindingsJSON holds the lint-category findings (possibly an empty set)
	// in the same JSON shape the lint step produces itself.
	FindingsJSON string
	// Summary is the housekeeping pass's one-line lint summary.
	Summary string
}

HousekeepingLintResult is the lint assessment produced by the combined document+lint housekeeping pass: the document step performs both duties in one agent invocation and hands the lint half to the lint step so it does not pay a second cold agent pass.

type RunSessions 

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

RunSessions manages the per-run, per-role durable agent sessions of the review loop. It is strictly scoped to one run: identities are keyed by (run, role), persisted as minimum resume metadata (run, role, agent, session id - never prompts or transcripts), and never shared across runs, branches, repositories, or roles.

Correctness always wins over reuse: adapters without session support run cold, a failed resume drops the identity and re-runs the same turn in a fresh same-role session, and any persistence failure degrades to cold invocations. A nil *RunSessions runs everything cold, preserving the pre-session behavior for steps outside the review loop and for tests.

func NewRunSessions 

func NewRunSessions(database *db.DB, runID string, sessionAgent agent.Agent, enabled bool) *RunSessions

NewRunSessions creates the manager for one run, loading any persisted session identities recorded by a previous process for the same run and agent. Identities stored for a different adapter are ignored: a session id is only meaningful to the adapter that minted it.

func (*RunSessions) Run 

func (rs *RunSessions) Run(ctx context.Context, a agent.Agent, role SessionRole, opts agent.RunOpts, logf func(string)) (*agent.Result, error)

Run executes one turn of the given role, reusing the role's durable session when the adapter supports it. logf (optional) receives operator- visible notes about session reuse and fallbacks.

type RunShared 

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

RunShared carries in-memory run-scoped results one step hands to a later step in the same run. It lives on the executor for the run's lifetime and is never persisted: on any process boundary the consuming step simply falls back to doing its own work.

func (*RunShared) ClearHousekeepingLint 

func (s *RunShared) ClearHousekeepingLint()

ClearHousekeepingLint discards a previous combined-pass lint assessment before a document pass starts, so a later lint step never consumes stale findings.

func (*RunShared) SetHousekeepingLint 

func (s *RunShared) SetHousekeepingLint(result HousekeepingLintResult)

SetHousekeepingLint records the combined pass's lint assessment for the lint step. It replaces any previous assessment (a document fix round re-runs the combined pass and re-stashes a fresh result).

func (*RunShared) TakeHousekeepingLint 

func (s *RunShared) TakeHousekeepingLint() (HousekeepingLintResult, bool)

TakeHousekeepingLint returns and consumes the combined pass's lint assessment. The second call returns false so a lint fix round re-assesses with its own agent pass instead of trusting a stale result.

type ScopeLimitedFindingsStep 

type ScopeLimitedFindingsStep interface {
	FindingsMayBeScopeLimited() bool
}

ScopeLimitedFindingsStep marks a step whose later rounds may reassess only the work selected for that round. For such a step, silence in a later round cannot retract an unresolved finding that the operator was already shown.

type SessionRole 

type SessionRole string

SessionRole identifies which durable review-loop session an invocation belongs to. The fixer role spans every review-fix turn of a run and is the only role that resumes a session. Review turns deliberately run session-free: a rereview certifies fixes implementing the previous review turn's findings, so resuming any review session would seat the prescriber of those fixes as their certifier and degrade the rereview into checking that its own prescription was implemented. 

const (
	// SessionRoleReviewer is legacy: review turns no longer create or resume
	// sessions. The constant remains so crash recovery keeps accepting
	// persisted rows written by earlier versions (see
	// validateRecoveredSessionProviders); such rows are never resumed.
	SessionRoleReviewer SessionRole = "reviewer"
	SessionRoleFixer    SessionRole = "review-fixer"
)

type Step 

type Step interface {
	// Name returns the step's identity in the fixed pipeline sequence.
	Name() types.StepName

	// Execute runs the step logic and returns an outcome.
	// A step that returns NeedsApproval=true will pause the pipeline
	// until the user responds with an approval action.
	Execute(sctx *StepContext) (*StepOutcome, error)
}

Step is the interface that each pipeline step implements.

type StepContext 

type StepContext struct {
	Ctx                   context.Context
	Run                   *db.Run
	Repo                  *db.Repo
	WorkDir               string
	Agent                 agent.Agent
	Config                *config.Config
	DB                    *db.DB
	Log                   func(string) // discrete log line (newline-terminated, user-visible + file)
	LogChunk              func(string) // raw streaming chunk (user-visible + file)
	LogFile               func(string) // file-only log callback (not shown to user)
	Fixing                bool         // true when re-executing after a "fix" action
	SkipFixExecution      bool         // replay an already-completed fix round's review turn only
	ReviewStartingHeadSHA string
	PreviousFindings      string // JSON findings from the previous execution (set during fix loop)
	KnownReviewLineages   string
	// StepResultID is the DB row ID of the current step's step_results record.
	// Steps use it to query their own round history for multi-round prompts.
	StepResultID string
	// EvidenceDir is where this run's test-evidence artifacts belong, always
	// outside the worktree. The executor resolves it once from the app root
	// (honoring test.evidence.local_root) so every consumer - the test step's
	// prompt and the PR step's publisher - names the same directory. Empty only
	// in embeddings that never gather evidence.
	EvidenceDir string
	Env         []string // extra environment variables for subprocesses (used in tests)
	// UserIntent is a short, possibly-empty summary of what the change author
	// was trying to accomplish. It's surfaced in step prompts so agents have
	// context beyond the diff. Its authority depends on IntentSource: an
	// explicit `--intent` is the author's own goal statement, while an
	// inferred summary comes from a local agent transcript.
	UserIntent string
	// IntentSource records the provenance of UserIntent so steps can weigh
	// its authority. db.RunIntentSourceAgent ("agent") means the driving
	// agent supplied it explicitly via `axi run --intent`; db.RunIntentSourceRerun
	// ("rerun") means that authoritative intent was inherited. Both are
	// authoritative acceptance criteria; an agent name ("claude", "codex", ...)
	// means it was inferred from a transcript (a hint). Empty when no intent exists.
	IntentSource string
	// UncertifiedFromSHA/ToSHA/SourceRunID name a previous run's fixer
	// commits on this branch whose re-review did not complete. They are set
	// on a later run's initial review (Fixing==false) so that review still
	// receives fix-round provenance. Empty when no such range applies.
	UncertifiedFromSHA     string
	UncertifiedToSHA       string
	UncertifiedSourceRunID string
	// UncertifiedPriorRounds are review rounds from the source run that left
	// the uncertified range. Nil when none apply.
	UncertifiedPriorRounds   []*db.StepRound
	UncertifiedPriorFindings string
	UncertifiedPriorLineages string
	// UncertifiedSelectedFindings is the selected subset a recovered
	// review-only round must verify. It is kept separate from carried findings:
	// silence may resolve it after a real review, but an ignored-only delta must
	// return it to the gate instead of silently approving.
	UncertifiedSelectedFindings string
	// Sessions manages the run's durable review-fixer session. The session
	// machinery remains role-generic for legacy recovery; nil runs every
	// invocation cold.
	Sessions *RunSessions
	// Shared carries in-memory run-scoped results one step hands to a later
	// step in the same run (e.g. the combined document+lint pass).
	Shared             *RunShared
	CIReadinessChanged func(ready, declaredNoCI bool)
	// OnPRMerged is a best-effort hook after a merged PR state is persisted.
	// Eval uses it to relabel auto-fix/shipped-unfixed gold; nil is a no-op.
	OnPRMerged func(ctx context.Context, runID string)
}

StepContext provides shared resources to pipeline steps during execution.

func (*StepContext) RunAgentSession 

func (sctx *StepContext) RunAgentSession(role SessionRole, opts agent.RunOpts) (*agent.Result, error)

RunAgentSession executes one turn of a durable review-loop role session, running cold when sessions are unavailable. Only the review step's fixer turns use this; every other agent invocation - including every review turn, which must stay independent of the session that prescribed the fixes under review - goes through sctx.Agent.Run directly and stays session-isolated.

type StepOutcome 

type StepOutcome struct {
	NeedsApproval      bool // whether the step pauses for user action
	AutoFixable        bool
	Findings           string // JSON findings for TUI display (optional)
	FindingsNormalized bool
	ExitCode           int    // process exit code (0 = success)
	PRURL              string // PR/MR URL if this step created or found one
	Skipped            bool   // mark the step as skipped without failing the run
	SkipRemaining      bool   // skip all subsequent steps (e.g. empty diff after rebase)
	// RestartFrom asks the executor to re-run validation from this earlier step.
	// CI repairs use it to send the new local head back through review before push.
	RestartFrom types.StepName
	// FixSummary, when non-empty, is the agent's one-line commit summary for
	// the fix attempt performed during this round. Steps populate it in fix
	// mode so the executor can persist it on the round record and later
	// rounds can reference what was previously attempted.
	FixSummary string
	// ReviewApprovedHeadSHA is set only by a successfully executed full review
	// round. The executor durably records it only when the review step actually
	// completes, never while that outcome is parked or after a failed round.
	ReviewApprovedHeadSHA string

	// DurationOverrideMS, when positive, replaces the wall-clock duration
	// reported for this step. Used by demo mode to show realistic durations
	// without actually waiting.
	DurationOverrideMS int64
}

StepOutcome is the result of executing a pipeline step.

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.