Documentation
¶
Overview ¶
Package workflow executes MVM source that coordinates durable agent tasks.
Index ¶
- type AgentOptions
- type AgentRequest
- type Controller
- type Dispatcher
- func (dispatcher *Dispatcher) Await(ctx context.Context, runID string) (*database.WorkflowRunEntity, error)
- func (dispatcher *Dispatcher) Shutdown(ctx context.Context) error
- func (dispatcher *Dispatcher) Start(ctx context.Context) error
- func (dispatcher *Dispatcher) Submit(ctx context.Context, request *ServiceRequest) (*database.WorkflowRunEntity, error)
- type DispatcherOptions
- type Event
- type EventKind
- type EventSink
- type PipelineResult
- type RunRequest
- type RunResult
- type Runner
- type Service
- func (service *Service) AgentTask(ctx context.Context, taskID string) (*database.AgentTaskEntity, bool, error)
- func (service *Service) AgentTaskDetails(ctx context.Context, runIDs []string) ([]database.WorkflowAgentTaskDetail, error)
- func (service *Service) AgentTasks(ctx context.Context, runID string) ([]database.WorkflowAgentTaskEntity, error)
- func (service *Service) Await(ctx context.Context, runID string) (*database.WorkflowRunEntity, error)
- func (service *Service) Cancel(ctx context.Context, ownerSessionID, runID string) (bool, error)
- func (service *Service) Events(ctx context.Context, runID string, after int64, limit int) ([]database.TaskEventEntity, error)
- func (service *Service) ExecuteQueued(ctx context.Context, runID string) (bool, error)
- func (service *Service) Get(ctx context.Context, runID string) (*database.WorkflowRunEntity, bool, error)
- func (service *Service) List(ctx context.Context, ownerSessionID string, limit int) ([]database.WorkflowRunEntity, error)
- func (service *Service) ListActive(ctx context.Context, ownerSessionID string, limit int) ([]database.WorkflowRunEntity, error)
- func (service *Service) RecoverInterrupted(ctx context.Context) ([]string, error)
- func (service *Service) Resume(ctx context.Context, runID string) (*database.WorkflowRunEntity, *RunResult, error)
- func (service *Service) Run(ctx context.Context, request *ServiceRequest) (*database.WorkflowRunEntity, *RunResult, error)
- func (service *Service) Submit(ctx context.Context, request *ServiceRequest) (*database.WorkflowRunEntity, error)
- type ServiceRequest
- type TaskResult
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type AgentOptions ¶
type AgentOptions struct {
NodeKey string `json:"node_key"`
AgentName string `json:"agent_name"`
Model string `json:"model"`
Provider string `json:"provider"`
ConcurrencyKey string `json:"concurrency_key"`
Depth int `json:"depth"`
}
AgentOptions configures one durable agent launch. The controller supplies any product-specific defaults not provided by the workflow.
type AgentRequest ¶
type AgentRequest struct {
ParentTaskID string
OwnerSessionID string
NodeKey string
Prompt string
Options AgentOptions
InvocationIndex int
}
AgentRequest describes a durable agent operation requested by a workflow.
type Controller ¶
type Controller interface {
Submit(context.Context, *AgentRequest) (*database.AgentTaskEntity, error)
Get(context.Context, string) (*database.AgentTaskEntity, bool, error)
Await(context.Context, string) (*database.AgentTaskEntity, error)
Cancel(context.Context, string, string) (*database.TaskEntity, bool, error)
}
Controller is the narrow durable-agent boundary used by workflow runs.
type Dispatcher ¶
type Dispatcher struct {
// contains filtered or unexported fields
}
Dispatcher polls durable queued workflows and executes atomically claimed runs.
func NewDispatcher ¶
func NewDispatcher(ctx context.Context, options DispatcherOptions) (*Dispatcher, error)
NewDispatcher creates and starts durable workflow workers.
func NewStoppedDispatcher ¶
func NewStoppedDispatcher(ctx context.Context, options DispatcherOptions) (*Dispatcher, error)
NewStoppedDispatcher creates a dispatcher without starting its workers.
func (*Dispatcher) Await ¶
func (dispatcher *Dispatcher) Await(ctx context.Context, runID string) (*database.WorkflowRunEntity, error)
Await delegates durable completion waiting to the workflow service.
func (*Dispatcher) Shutdown ¶
func (dispatcher *Dispatcher) Shutdown(ctx context.Context) error
Shutdown stops polling and waits for workers.
func (*Dispatcher) Start ¶
func (dispatcher *Dispatcher) Start(ctx context.Context) error
Start launches workflow workers and polling.
func (*Dispatcher) Submit ¶
func (dispatcher *Dispatcher) Submit( ctx context.Context, request *ServiceRequest, ) (*database.WorkflowRunEntity, error)
Submit persists and schedules a workflow run.
type DispatcherOptions ¶
type DispatcherOptions struct {
Service *Service
Tasks *database.TaskRepository
Logger *slog.Logger
Concurrency int
Buffer int
Interval time.Duration
}
DispatcherOptions configures durable workflow queue polling and execution.
type Event ¶
type Event struct {
Task TaskResult
Kind EventKind
TaskID string
NodeKey string
InvocationIndex int
}
Event is an observable workflow progress update.
type PipelineResult ¶
PipelineResult is one callback outcome. Results retain input order.
type RunRequest ¶
type RunRequest struct {
RunID string
Name string
Source string
OwnerSessionID string
OnEvent EventSink
Arguments map[string]any
PersistedLinks []database.WorkflowAgentTaskEntity
}
RunRequest describes one isolated workflow evaluation.
type RunResult ¶
type RunResult struct {
Value any
Stdout string
Stderr string
LaunchedTaskIDs []string
TaskResults []TaskResult
}
RunResult contains script output and the tasks launched by this run.
type Runner ¶
type Runner struct {
// contains filtered or unexported fields
}
Runner evaluates workflow source against a durable agent controller.
func NewRunner ¶
func NewRunner(controller Controller) (*Runner, error)
NewRunner creates a workflow runner.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service persists workflow lifecycle and delegates source evaluation to a Runner.
func NewService ¶
func NewService(runs *database.WorkflowRepository, runner *Runner) (*Service, error)
NewService creates the durable workflow execution boundary.
func (*Service) AgentTask ¶
func (service *Service) AgentTask(ctx context.Context, taskID string) (*database.AgentTaskEntity, bool, error)
AgentTask returns one agent task linked from a workflow.
func (*Service) AgentTaskDetails ¶
func (service *Service) AgentTaskDetails( ctx context.Context, runIDs []string, ) ([]database.WorkflowAgentTaskDetail, error)
AgentTaskDetails returns complete child tasks for multiple workflows in one query.
func (*Service) AgentTasks ¶
func (service *Service) AgentTasks( ctx context.Context, runID string, ) ([]database.WorkflowAgentTaskEntity, error)
AgentTasks returns workflow child links in launch order.
func (*Service) Await ¶
func (service *Service) Await(ctx context.Context, runID string) (*database.WorkflowRunEntity, error)
Await waits for a workflow run to reach a terminal state.
func (*Service) Events ¶
func (service *Service) Events( ctx context.Context, runID string, after int64, limit int, ) ([]database.TaskEventEntity, error)
Events returns durable workflow events in replay order.
func (*Service) ExecuteQueued ¶
ExecuteQueued claims and executes a previously submitted workflow run. A false return means another worker claimed the run first.
func (*Service) Get ¶
func (service *Service) Get(ctx context.Context, runID string) (*database.WorkflowRunEntity, bool, error)
Get returns one workflow run.
func (*Service) List ¶
func (service *Service) List( ctx context.Context, ownerSessionID string, limit int, ) ([]database.WorkflowRunEntity, error)
List returns workflow runs owned by a session.
func (*Service) ListActive ¶
func (service *Service) ListActive( ctx context.Context, ownerSessionID string, limit int, ) ([]database.WorkflowRunEntity, error)
ListActive returns nonterminal runs and terminal runs with active directly linked agents.
func (*Service) RecoverInterrupted ¶
RecoverInterrupted marks abandoned in-process runs interrupted after restart. Workflow source is not replayed because interpreter memory is intentionally not persisted.
func (*Service) Resume ¶
func (service *Service) Resume( ctx context.Context, runID string, ) (*database.WorkflowRunEntity, *RunResult, error)
Resume replays an interrupted workflow while reusing its persisted child invocations.
func (*Service) Run ¶
func (service *Service) Run( ctx context.Context, request *ServiceRequest, ) (*database.WorkflowRunEntity, *RunResult, error)
Run creates and executes one durable workflow run.
func (*Service) Submit ¶
func (service *Service) Submit( ctx context.Context, request *ServiceRequest, ) (*database.WorkflowRunEntity, error)
Submit durably queues a workflow run without executing it.