Documentation
¶
Index ¶
- Variables
- type ClaimHeldError
- type Option
- type StepTemplate
- type TaskCompletedCallback
- type TaskManager
- func (tm *TaskManager) ClaimTask(ctx context.Context, taskID, holder string) error
- func (tm *TaskManager) CompleteTaskStep(ctx context.Context, taskID, stepID string, payload map[string]any) error
- func (tm *TaskManager) CompleteTaskStepByToken(ctx context.Context, token string, payload map[string]any) error
- func (tm *TaskManager) GetAllTasks(ctx context.Context, caseID string) []TaskView
- func (tm *TaskManager) GetTaskRenderInfo(context context.Context, taskID string) (TaskView, error)
- func (tm *TaskManager) HandleTaskCompletion(ctx context.Context, completion engine.WorkflowCompletion) error
- func (tm *TaskManager) ReleaseTask(ctx context.Context, taskID, holder string) error
- func (tm *TaskManager) StartTask(ctx context.Context, payload engine.TaskPayload) (map[string]any, error)
- func (tm *TaskManager) StartTaskStep(ctx context.Context, payload engine.TaskPayload) (map[string]any, error)
- type TaskTemplate
- type TaskView
Constants ¶
This section is empty.
Variables ¶
var ErrClaimHeld = errors.New("task is claimed by another holder")
ErrClaimHeld means another holder has the task's claim. ClaimTask returns it as a *ClaimHeldError, which names the holder. A caller serving HTTP should answer it with 409.
var ErrHolderRequired = errors.New("claim holder is required")
ErrHolderRequired is returned by ClaimTask and ReleaseTask when the holder is empty. A caller serving HTTP should answer it with 400; it usually means the caller has no identity to claim as.
var ErrNotClaimHolder = errors.New("task is not claimed by the caller")
ErrNotClaimHolder is returned by ReleaseTask when the task is claimed by someone else. A caller serving HTTP should answer it with 403.
var ErrStaleStep = errors.New("step is no longer the active step of the task")
ErrStaleStep means the call named a step that is no longer active. Nothing was written. Callers serving HTTP should answer 409 and have the client refetch.
var ErrStepIDRequired = errors.New("step ID is required")
ErrStepIDRequired is returned by CompleteTaskStep when the caller did not say which step it is completing. A caller serving HTTP should answer it with 400.
var ErrTaskCompleted = errors.New("task is completed")
ErrTaskCompleted is returned by ClaimTask when the task has completed: there is nothing left to work on. A caller serving HTTP should answer it with 409.
var ErrTaskNotFound = errors.New("task not found")
ErrTaskNotFound means no task has the given ID. A caller serving HTTP should answer it with 404.
Functions ¶
This section is empty.
Types ¶
type ClaimHeldError ¶
type ClaimHeldError struct {
// Holder is who has the claim.
Holder string
// ClaimedAt is when Holder claimed the task, if known.
ClaimedAt *time.Time
}
ClaimHeldError is the error ClaimTask returns when another holder has the claim. It matches ErrClaimHeld with errors.Is.
func (*ClaimHeldError) Error ¶
func (e *ClaimHeldError) Error() string
func (*ClaimHeldError) Is ¶
func (e *ClaimHeldError) Is(target error) bool
type Option ¶
type Option func(*TaskManager)
Option configures a TaskManager.
func WithLogger ¶
WithLogger overrides the structured logger used by TaskManager. Defaults to slog.Default() when not provided.
type StepTemplate ¶
type StepTemplate = types.StepTemplate
TaskTemplate and StepTemplate are defined in artifact/adapter/types so that artifact adapters and orchestrator can both reference them without a circular import. These aliases preserve the orchestrator.TaskTemplate / orchestrator.StepTemplate API.
type TaskCompletedCallback ¶
type TaskCompletedCallback func(parentWorkflowID string, parentStepID string, finalVariables map[string]any) error
TaskCompletedCallback is a callback function invoked when a Task workflow completes. It is typically used to wake up the parent workflow with the final task output variables, by calling CompleteActivation with the parent's step ID (parentStepID) and "" for the run (the parent workflow never has more than one run — see TaskRecord.ParentStepID). It must return CompleteActivation's error as it is (or wrapped): HandleTaskCompletion treats engine.ErrActivationNotPending as "already resumed".
type TaskManager ¶
type TaskManager struct {
// contains filtered or unexported fields
}
TaskManager orchestrates decoupled tasks and interactions under parent workflows. It bridges macro-level workflows and micro-level interactive tasks via a single DB entry per task.
func NewTaskManager ¶
func NewTaskManager( db store.TaskStore, registry *artifact.Registry, pluginsRegistry *plugins.Registry, extensionsRegistry *extensions.Registry, taskWorkflowManager engine.TemporalManager, onTaskCompleted TaskCompletedCallback, renderer renderer.Renderer, opts ...Option, ) *TaskManager
NewTaskManager creates a TaskManager instance.
- db — the persistence/in-memory task store.
- registry — artifact registry holding task templates, step templates, workflow definitions, and render configs.
- pluginsRegistry — registry containing task execution plugin handlers.
- taskWorkflowManager — the TemporalManager used to start and complete Task sub-workflows.
- onTaskCompleted — callback invoked when a Task workflow finishes; typically invokes Parent.CompleteActivation to resume the parent workflow using stored coordinates.
func (*TaskManager) ClaimTask ¶
func (tm *TaskManager) ClaimTask(ctx context.Context, taskID, holder string) error
ClaimTask claims the task for holder. It succeeds when the task is unclaimed or already claimed by holder; a repeat claim keeps the original claim time. The task's steps and version are not changed, so an open step stays current.
Errors: ErrHolderRequired, ErrTaskNotFound, ErrTaskCompleted (the task has completed, even if holder had claimed it), or a *ClaimHeldError (another holder has the claim).
func (*TaskManager) CompleteTaskStep ¶
func (tm *TaskManager) CompleteTaskStep(ctx context.Context, taskID, stepID string, payload map[string]any) error
CompleteTaskStep is the public API for external clients or portals to submit form/interaction data and resume the given step in the corresponding Task workflow. stepID is the step the caller is acting on, as the task view reported it; it is required, and the call only ever completes that step.
Sequence:
- Pre-check, from the task row: the task is not completed and stepID is its active step. This only turns away obviously stale calls early; the authority is step 4.
- Pre-resume extensions (authz, validation) run on deep copies and can reject the call. They must have no side effects.
- The submission is prepared: it is confined to the step's OutputNamespace and stripped of system keys.
- The step's Temporal Activity is completed. Exactly one caller wins; a step that is not pending (already completed, or an earlier run of a looping node) fails with ErrStaleStep and nothing is written.
- The submission is persisted with the state ADVANCING, guarded by the step and the seq read in step 1. If the workflow has already started the next step this matches no row and is dropped; the submission reached the workflow through the Activity result either way.
- Post-resume extensions start in the background (best effort, not awaited).
Returning nil means the workflow accepted the step; the row shows ADVANCING (or the next step) from then on. Clients should refetch the task rather than assume the state.
func (*TaskManager) CompleteTaskStepByToken ¶
func (tm *TaskManager) CompleteTaskStepByToken(ctx context.Context, token string, payload map[string]any) error
CompleteTaskStepByToken is CompleteTaskStep for a caller that holds the callback token the dispatch carried (see plugins.CallbackToken), such as an external system calling back. It decodes the token to the task and step it names. A malformed token returns an error wrapping callbacktoken.ErrInvalid, which a caller serving HTTP should answer with 400.
func (*TaskManager) GetAllTasks ¶
func (tm *TaskManager) GetAllTasks(ctx context.Context, caseID string) []TaskView
GetAllTasks returns a lightweight summary of tasks for listing purposes. The View field is intentionally not populated — callers should use GetTaskRenderInfo to fetch the full rendered view for a specific task.
If caseID is non-empty, the result is narrowed to the case's tasks (TaskRecord.CaseID), across every workflow of the case; an empty string returns all tasks.
func (*TaskManager) GetTaskRenderInfo ¶
GetTaskRenderInfo retrieves a task record and dynamically decorates it with rich render metadata (like JSON schemas) fetched on-the-fly from its executing plugin.
func (*TaskManager) HandleTaskCompletion ¶
func (tm *TaskManager) HandleTaskCompletion(ctx context.Context, completion engine.WorkflowCompletion) error
HandleTaskCompletion is called when a Task workflow hits its END node (wire it to the task workflow's completion handler). It marks the task complete and fires the onTaskCompleted callback to resume the parent workflow.
The completion is written before the parent is woken, and each half is safe to repeat, so a crash between them is repaired by Temporal retrying the whole call:
- The task row is set to COMPLETED at completion.Seq, guarded so it only matches if no later step exists. A repeat matches again.
- The parent's Activity is completed through onTaskCompleted. If the parent step is no longer pending, an earlier attempt already woke it, which is success.
Writing COMPLETED first means it cannot serve as the "already done" marker for step 2; that is why a completed row does not short-circuit a retry.
func (*TaskManager) ReleaseTask ¶
func (tm *TaskManager) ReleaseTask(ctx context.Context, taskID, holder string) error
ReleaseTask clears holder's claim on the task. Releasing a task that is not claimed succeeds, so a repeated release is harmless. A completed task's claim can still be released.
Errors: ErrHolderRequired, ErrTaskNotFound, or ErrNotClaimHolder (someone else has the claim).
func (*TaskManager) StartTask ¶
func (tm *TaskManager) StartTask(ctx context.Context, payload engine.TaskPayload) (map[string]any, error)
StartTask is called by the parent workflow engine when it activates a TASK node. It looks up the template registry, creates a single DB record with parent coordinates, and kicks off the Task's internal workflow.
func (*TaskManager) StartTaskStep ¶
func (tm *TaskManager) StartTaskStep(ctx context.Context, payload engine.TaskPayload) (map[string]any, error)
StartTaskStep is called by the Task's workflow engine when it activates an interaction step. It routes to the correct capability handler dynamically from the plugin registry.
The step is claimed in the DB before the plugin runs, and the plugin's render state is written only if the step is still the active one:
- Claim: make this step (payload.ActivationID) the task's active step, at payload.Seq. The claim matches only if no later step has started. A retry of this same step matches again. A late attempt for a step that is already over matches nothing, and returns ErrStaleStep before the plugin runs, so it makes no outbound call and writes nothing.
- The plugin runs on an in-memory copy of the record.
- Render write: store the plugin's state and data, guarded by the active step and seq. A fast callback can already have completed this step and started the next one, in which case the write matches nothing and is dropped. That is success, not an error.
type TaskTemplate ¶
type TaskTemplate = types.TaskTemplate
TaskTemplate and StepTemplate are defined in artifact/adapter/types so that artifact adapters and orchestrator can both reference them without a circular import. These aliases preserve the orchestrator.TaskTemplate / orchestrator.StepTemplate API.
type TaskView ¶
type TaskView struct {
TaskID string `json:"task_id"`
TaskType string `json:"task_type"`
State string `json:"state"`
// StepID is the step the task is currently on. A caller acting on the task (CompleteTaskStep)
// echoes it back, so an action taken on a view the task has since moved past is rejected as
// stale instead of being applied to the step that is active now. Empty until the first step
// starts.
StepID string `json:"step_id,omitempty"`
// Version increases every time the task moves on. It is read-only: clients use it to keep the
// newest of several views they fetched, and it is never accepted as input.
Version int64 `json:"version"`
// ClaimedBy is who holds the task's claim, as the host recorded it (see TaskManager.ClaimTask);
// empty when the task is unclaimed. ClaimedAt is when they claimed it.
ClaimedBy string `json:"claimed_by,omitempty"`
ClaimedAt *time.Time `json:"claimed_at,omitempty"`
View json.RawMessage `json:"view,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
TaskView represents a purely presentational view of a task for the frontend. It removes all internal Temporal coordinates and exposes only what the UI needs.
Source Files