Documentation
¶
Overview ¶
Package runs is the run manager: it persists workflow versions, runs and their steps, and orchestrates execution on top of internal/workflow's engine and an ActionExecutor (typically internal/plugins.Supervisor).
Index ¶
- Constants
- Variables
- func CancelRun(ctx context.Context, db *sql.DB, id string) error
- func Continue(ctx context.Context, db *sql.DB, executor ActionExecutor, ...) error
- func GetRun(ctx context.Context, db *sql.DB, id string) (*Run, []Step, error)
- func InstallWorkflow(ctx context.Context, db *sql.DB, source []byte, ...) (*workflow.Definition, error)
- func InstallWorkflowAtVersion(ctx context.Context, db *sql.DB, source []byte, version int, ...) (*workflow.Definition, error)
- func LatestWorkflow(ctx context.Context, db *sql.DB, workflowID string) (*workflow.Definition, error)
- func NextWorkflowVersion(ctx context.Context, db *sql.DB, workflowID string) (int, error)
- func WatchRun(ctx context.Context, db *sql.DB, runID string) (<-chan Event, error)
- func WorkflowSource(ctx context.Context, db *sql.DB, workflowID string, version int) (string, error)
- type ActionExecutor
- type Event
- type ExecuteOptions
- type Run
- func Execute(ctx context.Context, db *sql.DB, executor ActionExecutor, workflowID string, ...) (*Run, error)
- func ListRuns(ctx context.Context, db *sql.DB, workflowID string) ([]Run, error)
- func Start(ctx context.Context, db *sql.DB, workflowID string, inputs map[string]any) (*workflow.Definition, *Run, map[string]any, error)
- type Step
- type WorkflowVersionSummary
Constants ¶
const DefaultStepTimeout = 30 * time.Second
DefaultStepTimeout bounds how long a single step's action call may run when ExecuteOptions.StepTimeout is left zero.
Variables ¶
var ErrRunNotCancellable = errors.New("run is not in a cancellable state")
ErrRunNotCancellable is returned by CancelRun when the run has already reached a terminal state.
var ErrRunNotFound = errors.New("run not found")
ErrRunNotFound is returned when no run with the given id has been recorded.
var ErrWorkflowNotFound = errors.New("workflow not found")
ErrWorkflowNotFound is returned when no workflow (or no such version of it) has been installed.
Functions ¶
func CancelRun ¶
CancelRun marks a run still in the queued or running state as cancelled, along with any of its steps that never reached a terminal state.
patchcord workflow run executes synchronously within its own process, so this cannot interrupt a run actively in progress elsewhere — it is meant for a run left behind non-terminal by a crashed process. It returns ErrRunNotCancellable if the run has already reached a terminal state.
func Continue ¶
func Continue(ctx context.Context, db *sql.DB, executor ActionExecutor, def *workflow.Definition, run *Run, inputs map[string]any, bindings map[string]string, opts ExecuteOptions) error
Continue runs def's steps for run — already created and transitioned to Running by Start — against executor, persisting progress and the final status as it happens. inputs and bindings are the same values passed to Start (bindings maps a logical binding name, as referenced by a step's ${{ bindings.<name> }} connector expression, to the id of the connector to use — see workflow.ResolveConnector).
Steps run sequentially; the first one to fail — including timing out, or ctx being cancelled — stops the run. Every step that never got a chance to run is recorded as skipped, so no step is left dangling in "pending". A step whose If resolves to false is also recorded as skipped, but does not stop the run — the loop moves on to the next step exactly as if this one had succeeded, only without an entry in stepOutputs (see workflow.ResolveIf) — unless that step also sets StopIfFalse, in which case every following step is recorded skipped too and the run ends Succeeded, not Failed: a guard clause's early return, not an error. A step whose ElseOf names an earlier step that actually ran is skipped before its own If is even evaluated (see ranSteps below) — chaining ElseOf onto consecutive steps builds an if/elseif/else without nesting. A foreach step calls its action once per resolved item, sequentially, sharing one StepTimeout budget across every iteration; the first item to fail stops the run exactly like a regular step failure would, and the step's recorded output is the per-item outputs collected into lists under the action's own output keys (see workflow.ResolveForeach). Continue only returns a non-nil error for a genuine persistence failure — a step's own failure (or the run's ctx being cancelled) is captured in the run's final status instead, never returned as an error here, exactly as Execute's callers already expect.
func GetRun ¶
GetRun returns a run and its steps by id. It returns ErrRunNotFound if no such run has been recorded.
func InstallWorkflow ¶
func InstallWorkflow(ctx context.Context, db *sql.DB, source []byte, knownActions map[string]workflow.KnownAction) (*workflow.Definition, error)
InstallWorkflow validates def against knownActions, then records it as a new, immutable version (ADR-0008): publishing never overwrites an existing (workflow_id, version) row.
func InstallWorkflowAtVersion ¶
func InstallWorkflowAtVersion(ctx context.Context, db *sql.DB, source []byte, version int, knownActions map[string]workflow.KnownAction) (*workflow.Definition, error)
InstallWorkflowAtVersion validates source against knownActions like InstallWorkflow, but records it under version rather than source's own declared `version:` field. It exists solely for internal/bundles' dev-mode install path (`bundle dev`/`patchcord dev`, see installWorkflowForDev): editing an embedded workflow's body without bumping its version is installed under the next unused version instead of being rejected, so the source file on disk is never rewritten. workflow install and InstallPackage (`bundle install`/`update`) never call this — they stay on InstallWorkflow, strict under ADR-0008 with no exception.
func LatestWorkflow ¶
func LatestWorkflow(ctx context.Context, db *sql.DB, workflowID string) (*workflow.Definition, error)
LatestWorkflow returns the highest installed version of workflowID.
func NextWorkflowVersion ¶
NextWorkflowVersion returns the next unused version number for workflowID: one past the highest version currently installed, or 1 if none is. Used by internal/bundles' dev-mode install path to auto-assign a version when a workflow's content changed without bumping its declared `version:` field (see InstallWorkflowAtVersion).
func WatchRun ¶
WatchRun returns a channel delivering a status Event each time runID or one of its steps moves to a new status, starting from an empty baseline so a client connecting mid-run still gets the status each entity currently holds instead of only the ones still to come. It closes the channel once the run reaches a terminal status or ctx is cancelled.
Because the database only ever holds the current status (there is no event log to replay — see the vision document, section 14, "Event log"), a client that connects after a fast run has already finished only observes each entity's single final status, not the intermediate ones it passed through; a client watching a run already in flight observes every transition from the moment it connects onward.
Patchcord runs a workflow synchronously within a single process from start to finish (ADR-0018), so there is no in-process event bus another process — such as the agent's HTTP server — could subscribe to. WatchRun polls the database instead, the only channel shared between a `workflow run` process and anyone watching it (see ADR-0019).
It returns ErrRunNotFound immediately if no such run exists.
func WorkflowSource ¶
func WorkflowSource(ctx context.Context, db *sql.DB, workflowID string, version int) (string, error)
WorkflowSource returns the raw YAML source of one workflow version. version 0 means the latest installed version. It returns ErrWorkflowNotFound if no such (workflow, version) has been installed.
Types ¶
type ActionExecutor ¶
type ActionExecutor interface {
ExecuteAction(ctx context.Context, actionID string, input map[string]any, connector *connectors.ResolvedConnector) (map[string]any, error)
}
ActionExecutor runs one action and returns its output. It is the only thing the runner needs to actually execute a step, which keeps this package free of any dependency on how plugins are launched or supervised — internal/plugins.Supervisor satisfies this interface. connector is nil unless the step bound one (see workflow.Step.Connector).
type Event ¶
type Event struct {
RunID string
StepID string // empty for a run-level event
Status string
Error string
Time time.Time
}
Event is one observed status change for a run or one of its steps, delivered by WatchRun. It mirrors the event log sketched in the vision document (section 14).
type ExecuteOptions ¶
type ExecuteOptions struct {
// StepTimeout bounds each individual step's action call. Defaults to
// DefaultStepTimeout when zero. A step that times out fails the run
// (it is not treated as a user-requested cancellation). For a foreach
// step, this budget covers the whole step — every item's action call
// combined, not one budget per item — so a long list needs a StepTimeout
// sized for all of its iterations together.
StepTimeout time.Duration
// Secrets resolves a step's bound connector's secret references.
// Defaults to secrets.EnvStore{} when nil, so existing callers that
// build ExecuteOptions{} directly (or only set StepTimeout) keep
// resolving "env" references exactly as before secrets.MultiStore
// existed.
Secrets secrets.Store
}
ExecuteOptions controls how Execute runs a workflow.
type Run ¶
type Run struct {
ID string
WorkflowID string
WorkflowVersion int
Status workflow.RunStatus
Inputs map[string]any
Outputs map[string]any
Error string
CreatedAt time.Time
StartedAt *time.Time
FinishedAt *time.Time
}
Run is one execution of a workflow version, as persisted.
func Execute ¶
func Execute(ctx context.Context, db *sql.DB, executor ActionExecutor, workflowID string, inputs map[string]any, bindings map[string]string, opts ExecuteOptions) (*Run, error)
Execute runs the latest installed version of workflowID to completion — Start followed by Continue — persisting the run and every step's progress as it happens, and returns the completed run. See Start and Continue for what each half does; most callers (the CLI's `workflow run`, most tests) want this blocking, all-in-one behavior.
func ListRuns ¶
ListRuns returns every recorded run, most recently created first. workflowID, if non-empty, restricts the list to runs of that workflow.
func Start ¶
func Start(ctx context.Context, db *sql.DB, workflowID string, inputs map[string]any) (*workflow.Definition, *Run, map[string]any, error)
Start creates a new run of workflowID's latest installed version and transitions it straight to Running, then returns immediately — it does not execute any step. Call Continue next to actually run them.
inputs is resolved against def.Inputs (workflow.PrepareInputs) before anything is persisted: defaults are filled in, values coming from a string-only source (the CLI's --input flags) are coerced to their declared type, and a missing required input or an undeclared key fails fast, before a run row even exists. The returned map is this resolved result, not the caller's original inputs — callers must pass it, not their own inputs, to Continue, so step expression resolution (${{ workflow.inputs.<key> }}) sees the same coerced/defaulted values that were persisted as the run's inputs.
Split from the step-running loop (Continue) so a caller that must not block for the run's entire duration — the HTTP API's POST /v1/workflows/{id}/run, which needs to answer with the new run's id right away so a client can start watching /v1/runs/{id}/events — can call Start synchronously and Continue in a background goroutine (see internal/api's handleRunWorkflow). Execute composes both for callers (the CLI, most tests) that do want to block until completion.
Like Continue's own bookkeeping writes, Start's persistence is bounded by persistTimeout but deliberately not derived from ctx: a caller whose ctx is cancelled the instant it calls Start (e.g. an HTTP client that disconnects immediately) still gets a consistently created, Running run row rather than an ambiguous half-created one — the step loop in Continue is what turns a cancelled ctx into a properly recorded RunCancelled.
type Step ¶
type Step struct {
RunID string
StepID string
Status workflow.StepStatus
Input map[string]any
Output map[string]any
Error string
StartedAt *time.Time
FinishedAt *time.Time
}
Step is one step of a Run, as persisted.
type WorkflowVersionSummary ¶
WorkflowVersionSummary describes one installed workflow version, without its full source (see WorkflowSource for that).
func ListWorkflows ¶
ListWorkflows returns every installed workflow version, most recently installed first.
Source Files