Documentation
¶
Index ¶
- Constants
- Variables
- func RunStatusTerminal(s RunStatus) bool
- func StepRunStatusTerminal(s StepRunStatus) bool
- func ValidRunStatusTransition(from, to RunStatus) bool
- func ValidStepRunTransition(from, to StepRunStatus) bool
- type ClaimLeaseInput
- type CreateRunInput
- type CreateStepRunInput
- type Definition
- type DefinitionStep
- type FinalizeFailedRunInput
- type ReleaseLeaseInput
- type RenewLeaseInput
- type Revision
- type Run
- type RunStatus
- type StepBinding
- type StepRun
- type StepRunStatus
- type Store
- type TransitionRunInput
- type TransitionStepRunInput
- type UpdateInput
- type Workflow
Constants ¶
const ( StatusDraft = "draft" StatusPublished = "published" StatusArchived = "archived" StepTypeAgentTask = "agent_task" )
Variables ¶
var ( ErrInvalidRunTransition = errors.New("invalid workflow run status transition") ErrInvalidStepRunTransition = errors.New("invalid workflow step run status transition") )
ErrInvalidRunTransition and ErrInvalidStepRunTransition are returned when a caller asks a run or step run to move between statuses that ValidRunStatusTransition / ValidStepRunTransition do not allow. They name a programming error, not a lost race: a refused-but-valid transition (the row was no longer at the expected status) is reported as a false result, not an error.
var ErrRevisionConflict = apierr.New(apierr.KindConflict, "workflow changed since it was read")
ErrRevisionConflict means the workflow advanced between the revision the service observed and the write it guarded on it: another edit already appended the next revision. The caller re-reads and retries from the current revision. It is the conflict every content edit, status transition, and restore returns so a stale write is refused rather than overwriting a newer definition or leaking the duplicate-key error the append would otherwise raise.
Functions ¶
func RunStatusTerminal ¶
RunStatusTerminal reports whether a run in this status has finished; a terminal run never changes again.
func StepRunStatusTerminal ¶
func StepRunStatusTerminal(s StepRunStatus) bool
StepRunStatusTerminal reports whether a step run has finished. Blocked is terminal alongside the three natural ends: a blocked step is never revisited.
func ValidRunStatusTransition ¶
ValidRunStatusTransition reports whether a run may move directly from one status to another. Terminal statuses are immutable. Runs are created running; pending is reserved for a run that has not yet been dispatched.
func ValidStepRunTransition ¶
func ValidStepRunTransition(from, to StepRunStatus) bool
ValidStepRunTransition reports whether a step run may move directly from one status to another. A pending step may start (running), be blocked by an earlier failure, or fail outright when its task cannot be created; a running step ends succeeded, failed, or canceled. Terminal statuses are immutable.
Types ¶
type ClaimLeaseInput ¶
type ClaimLeaseInput struct {
WorkflowRunID string
Owner string
Now time.Time
LeaseExpiresAt time.Time
}
ClaimLeaseInput acquires a reconciliation lease on a non-terminal run for Owner. The store writes nothing unless the run has no owner or its prior lease has expired at Now.
type CreateRunInput ¶
type CreateStepRunInput ¶
type Definition ¶
type Definition struct {
Steps []DefinitionStep `json:"steps"`
}
Definition is the parsed structure of a workflow definition JSON.
type DefinitionStep ¶
type DefinitionStep struct {
StepID string `json:"step_id"`
Type string `json:"type"`
TargetAgentID string `json:"target_agent_id"`
Prompt string `json:"prompt"`
// Bindings feed an earlier step's output into this step's input. Each names a
// value (Name) taken from the whole output of a prior step (FromStep). The
// bound output reaches the Task as labelled untrusted context, never the
// agent's instructions.
Bindings []StepBinding `json:"bindings,omitempty"`
}
DefinitionStep describes one step in a workflow definition.
type FinalizeFailedRunInput ¶
type FinalizeFailedRunInput struct {
WorkflowRunID string
StepRunID string
StepIndex int
StepExpected StepRunStatus
StepStatus StepRunStatus
RunExpected RunStatus
RunStatus RunStatus
TaskRunID *string
ErrorMessage *string
StartedAt *time.Time
EndedAt *time.Time
}
FinalizeFailedRunInput ends a run because one step ended badly. In one transaction the store moves the step to StepStatus (failed or canceled), blocks every later step still pending, and moves the run to RunStatus. Both moves are guarded: nothing is written unless the step is at StepExpected and both transitions are valid.
type ReleaseLeaseInput ¶
ReleaseLeaseInput clears the lease Owner holds and sets when the run next wants a pass. NextReconcileAt is nil to leave the run without a scheduled pass, waking only from a due sweep. A stale owner matches nothing.
type RenewLeaseInput ¶
RenewLeaseInput extends the lease Owner already holds. A stale owner -- one a takeover has replaced -- matches nothing and gets a false result.
type Revision ¶
type Revision struct {
WorkflowID string `json:"workflow_id"`
Revision int `json:"revision"`
Name string `json:"name"`
Description string `json:"description"`
Definition string `json:"definition"`
Status string `json:"status"`
CreatedBy string `json:"created_by"`
CreatedAt time.Time `json:"created_at"`
}
Revision is one recorded version of a workflow.
Revisions are append-only: an edit adds one, nothing rewrites or deletes one, and restoring an older revision is itself an edit that appends a new one.
type Run ¶
type Run struct {
ID string `json:"id"`
WorkflowID string `json:"workflow_id"`
// WorkflowRevision is the workflow revision this run expanded. It is 0 for
// runs started before workflows recorded revisions.
WorkflowRevision int `json:"workflow_revision,omitempty"`
IssueID *string `json:"issue_id,omitempty"`
Status string `json:"status"`
CreatedBy string `json:"created_by"`
CreatedAt time.Time `json:"created_at"`
StartedAt *time.Time `json:"started_at,omitempty"`
EndedAt *time.Time `json:"ended_at,omitempty"`
ErrorMessage *string `json:"error_message,omitempty"`
// Reconciliation scheduling and ownership. ReconcileOwner and LeaseExpiresAt
// are a bounded lease that reduces duplicate reconciliation work; they are not
// the correctness mechanism. NextReconcileAt is when this run next wants a
// reconciliation pass. All three are cleared when the run becomes terminal.
ReconcileOwner *string `json:"reconcile_owner,omitempty"`
LeaseExpiresAt *time.Time `json:"lease_expires_at,omitempty"`
NextReconcileAt *time.Time `json:"next_reconcile_at,omitempty"`
}
Run is one execution attempt of a workflow.
type RunStatus ¶
type RunStatus string
RunStatus is the lifecycle status of one workflow run. StepRunStatus is one step's status within that run. Both are the canonical execution-plane state machine for workflows, the analog of coretask.RunStatus, and every move between their values goes through the transition helpers below so an illegal or concurrent change is refused at the store rather than silently written.
func TerminalRunStatuses ¶
func TerminalRunStatuses() []RunStatus
TerminalRunStatuses lists the statuses RunStatusTerminal reports true for, so a store can build the "non-terminal" filter the due-run query and the lease guards rest on without duplicating the set.
type StepBinding ¶
StepBinding binds one earlier step's output into a downstream step's input under a name. The whole upstream output is bound; there is no selection or templating in this contract.
type StepRun ¶
type StepRun struct {
ID string `json:"id"`
WorkflowRunID string `json:"workflow_run_id"`
StepID string `json:"step_id"`
StepIndex int `json:"step_index"`
StepType string `json:"step_type"`
TargetAgentID *string `json:"target_agent_id,omitempty"`
// AgentName, AgentDescription, and AgentInstructions capture the target agent
// definition as it was when the run started, so later edits to the agent cannot
// change what a step in flight sends to the model.
AgentName string `json:"agent_name,omitempty"`
AgentDescription string `json:"agent_description,omitempty"`
AgentInstructions string `json:"agent_instructions,omitempty"`
AgentRevision int `json:"agent_revision,omitempty"`
Prompt string `json:"prompt"`
// Bindings is the run's snapshot of this step's input bindings, taken at start
// so a later definition edit cannot change what an in-flight step receives.
Bindings []StepBinding `json:"bindings,omitempty"`
Status string `json:"status"`
TaskID *string `json:"task_id,omitempty"`
TaskRunID *string `json:"task_run_id,omitempty"`
OutputSummary *string `json:"output_summary,omitempty"`
ErrorMessage *string `json:"error_message,omitempty"`
CreatedAt time.Time `json:"created_at"`
StartedAt *time.Time `json:"started_at,omitempty"`
EndedAt *time.Time `json:"ended_at,omitempty"`
}
StepRun is one durable step execution record under a workflow run.
type StepRunStatus ¶
type StepRunStatus string
const ( StepRunStatusPending StepRunStatus = "pending" StepRunStatusRunning StepRunStatus = "running" StepRunStatusSucceeded StepRunStatus = "succeeded" StepRunStatusFailed StepRunStatus = "failed" StepRunStatusCanceled StepRunStatus = "canceled" // StepRunStatusBlocked is terminal: an earlier step ended badly, so this // still-pending step will never run. StepRunStatusBlocked StepRunStatus = "blocked" )
type Store ¶
type Store interface {
ListWorkflowsBySpace(ctx context.Context, spaceID string) ([]Workflow, error)
CreateWorkflow(ctx context.Context, spaceID, createdBy, name, description, definition string) (*Workflow, error)
GetWorkflow(ctx context.Context, workflowID string) (*Workflow, error)
UpdateWorkflow(ctx context.Context, workflowID, spaceID string, in UpdateInput) (*Workflow, error)
CreateWorkflowRun(ctx context.Context, in CreateRunInput) (*Run, error)
ListWorkflowRunsByWorkflow(ctx context.Context, workflowID string, limit, offset int) ([]Run, int, error)
ListWorkflowRunsByIssue(ctx context.Context, issueID string, limit, offset int) ([]Run, int, error)
GetWorkflowRun(ctx context.Context, workflowRunID string) (*Run, error)
ListWorkflowStepRuns(ctx context.Context, workflowRunID string) ([]StepRun, error)
CreateWorkflowStepRuns(ctx context.Context, workflowRunID string, steps []CreateStepRunInput) ([]StepRun, error)
// TransitionWorkflowRun and TransitionWorkflowStepRun apply one guarded
// status change each; a false result means the row was not at the expected
// status, so another actor won the transition. FinalizeFailedWorkflowRun
// ends a run and blocks its remaining steps in one transaction.
TransitionWorkflowRun(ctx context.Context, in TransitionRunInput) (bool, error)
TransitionWorkflowStepRun(ctx context.Context, in TransitionStepRunInput) (bool, error)
FinalizeFailedWorkflowRun(ctx context.Context, in FinalizeFailedRunInput) (bool, error)
// ListDueWorkflowRuns returns non-terminal runs that need a reconciliation
// pass at now -- their scheduled time has arrived or their lease expired --
// in stable oldest-due order, bounded by limit (a documented default when
// limit <= 0). ClaimWorkflowRunLease, RenewWorkflowRunLease, and
// ReleaseWorkflowRunLease are guarded writes: a false result means the run was
// terminal, already leased to another unexpired owner, or held by someone
// else, so this caller did not win the lease.
ListDueWorkflowRuns(ctx context.Context, now time.Time, limit int) ([]Run, error)
ClaimWorkflowRunLease(ctx context.Context, in ClaimLeaseInput) (bool, error)
RenewWorkflowRunLease(ctx context.Context, in RenewLeaseInput) (bool, error)
ReleaseWorkflowRunLease(ctx context.Context, in ReleaseLeaseInput) (bool, error)
GetWorkflowStepRunByTaskID(ctx context.Context, taskID string) (*StepRun, error)
GetWorkflowStepRunByTaskRunID(ctx context.Context, taskRunID string) (*StepRun, error)
// ListWorkflowRevisions returns a workflow's revisions, newest first, with
// the total count.
ListWorkflowRevisions(ctx context.Context, workflowID string, limit, offset int) ([]Revision, int, error)
// GetWorkflowRevision returns one revision, or nil when the workflow has no
// such revision number.
GetWorkflowRevision(ctx context.Context, workflowID string, revision int) (*Revision, error)
}
Store provides workflow and workflow execution persistence.
type TransitionRunInput ¶
type TransitionRunInput struct {
WorkflowRunID string
ExpectedStatus RunStatus
NewStatus RunStatus
StartedAt *time.Time
EndedAt *time.Time
ErrorMessage *string
}
TransitionRunInput atomically moves a run from ExpectedStatus to NewStatus. The store writes nothing unless the run's current status is ExpectedStatus and the move is a ValidRunStatusTransition.
type TransitionStepRunInput ¶
type TransitionStepRunInput struct {
StepRunID string
ExpectedStatus StepRunStatus
NewStatus StepRunStatus
TaskID *string
TaskRunID *string
OutputSummary *string
ErrorMessage *string
StartedAt *time.Time
EndedAt *time.Time
}
TransitionStepRunInput atomically moves a step run from ExpectedStatus to NewStatus, carrying the fields that land with a status change. The store writes nothing unless the step's current status is ExpectedStatus and the move is a ValidStepRunTransition.
type UpdateInput ¶
type UpdateInput struct {
Name *string
Description *string
Definition *string
Status *string
// ExpectedRevision is the revision the service observed before this edit. The
// store guards the workflow row update on it and appends the next revision in
// the same transaction, so a second writer that started from the same revision
// loses the compare-and-set and gets ErrRevisionConflict rather than
// overwriting the winner or leaking the duplicate-key error.
ExpectedRevision int
// UpdatedBy is recorded as the author of the revision this update appends.
UpdatedBy string
}
type Workflow ¶
type Workflow struct {
ID string `json:"id"`
SpaceID string `json:"space_id"`
Name string `json:"name"`
Description string `json:"description"`
Definition string `json:"definition"`
Status string `json:"status"`
// Revision numbers the workflow_revision row holding this content. It
// starts at 1 and advances every time the name, description, definition,
// or status changes.
Revision int `json:"revision"`
CreatedBy string `json:"created_by"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
Workflow is a reusable space-scoped execution plan.