Struct WorkflowCtx 

pub struct WorkflowCtx<S: WorkflowStore> {
    pub binary_version: Option<&'static str>,
    /* private fields */
}

The workflow context. Owns the store, background-task handles, and per-request config. Serves as both the orchestrator (all engine methods live as impl WorkflowCtx<S>) and the axum state (Arc<WorkflowCtx<S>>).

S is the concrete store backend (SqliteStore or PostgresStore). WorkflowStore uses RPIT futures and is not dyn-compatible, so the generic parameter is retained here to avoid boxing every async call.

Fields

binary_version: Option<&'static str>

Version of the containing binary (e.g. the assay-lua CLI) — set by embedders so /api/v1/engine/workflow/version reflects the user-facing binary, not this internal engine-crate version.

Implementations

impl<S: WorkflowStore> WorkflowCtx<S>

pub async fn schedule_activity( &self, workflow_id: &str, seq: i32, name: &str, input: Option<&str>, task_queue: &str, opts: ScheduleActivityOpts, ) -> Result<WorkflowActivity>

Schedule an activity within a workflow.

Idempotent on (workflow_id, seq) — if an activity with this sequence number already exists for the workflow, returns its id without creating a duplicate row or duplicate ActivityScheduled event. This is essential for deterministic replay: a worker can re-run the workflow function and call schedule_activity(seq=1, ...) repeatedly without producing side effects.

On the first call for a seq:

• inserts a row in workflow_activities with status PENDING
• appends an ActivityScheduled event to the workflow event log
• if the workflow is still PENDING, transitions it to RUNNING

pub async fn claim_activity( &self, task_queue: &str, worker_id: &str, ) -> Result<Option<WorkflowActivity>>

pub async fn get_activity(&self, id: i64) -> Result<Option<WorkflowActivity>>

pub async fn complete_activity( &self, id: i64, result: Option<&str>, error: Option<&str>, failed: bool, ) -> Result<()>

Mark a successfully-executed activity complete and append an ActivityCompleted event to the workflow event log so a replaying workflow can pick up the cached result.

failed=true is preserved for legacy callers that go straight through complete with a non-retry path; new code should call WorkflowCtx::fail_activity instead so retry policy is honored.

pub async fn fail_activity(&self, id: i64, error: &str) -> Result<()>

Fail an activity, honoring its retry policy.

If attempt < max_attempts, the activity is re-queued with exponential backoff (initial_interval_secs * backoff_coefficient^(attempt-1)) and attempt is incremented. No event is appended — retries are an internal-engine concern, not workflow-visible.

If attempt >= max_attempts, the activity is permanently FAILED and an ActivityFailed event is appended so the workflow can react.

pub async fn heartbeat_activity( &self, id: i64, details: Option<&str>, ) -> Result<()>

pub async fn record_side_effect( &self, workflow_id: &str, value: &str, ) -> Result<()>

impl<S: WorkflowStore> WorkflowCtx<S>

pub async fn start_child_workflow( &self, namespace: &str, parent_id: &str, workflow_type: &str, workflow_id: &str, input: Option<&str>, task_queue: &str, ) -> Result<WorkflowRecord>

pub async fn list_child_workflows( &self, parent_id: &str, ) -> Result<Vec<WorkflowRecord>>

pub async fn continue_as_new( &self, workflow_id: &str, input: Option<&str>, explicit_new_id: Option<&str>, ) -> Result<WorkflowRecord>

impl<S: WorkflowStore> WorkflowCtx<S>

pub fn start(store: Arc<S>) -> Self

Start the context with all background tasks.

pub fn with_event_bus(self, bus: WorkflowEventBus) -> Self

Attach the engine-wide event bus. The API layer sets this up so the SSE stream (/events/stream) and the dispatch-wakeup loop see state transitions as they happen. Returns the context by value so callers can chain.

pub fn with_binary_version(self, version: &'static str) -> Self

Set the binary version string surfaced by /api/v1/engine/workflow/version.

pub fn store(&self) -> &S

Access the underlying store (for the API layer).

pub fn bus(&self) -> Option<&WorkflowEventBus>

Access the event bus (for SSE + scheduler wake-up).

impl<S: WorkflowStore> WorkflowCtx<S>

pub async fn start_workflow( &self, namespace: &str, workflow_type: &str, workflow_id: &str, input: Option<&str>, task_queue: &str, search_attributes: Option<&str>, ) -> Result<WorkflowRecord>

pub async fn get_workflow(&self, id: &str) -> Result<Option<WorkflowRecord>>

pub async fn list_workflows( &self, namespace: &str, status: Option<WorkflowStatus>, workflow_type: Option<&str>, search_attrs_filter: Option<&str>, limit: i64, offset: i64, ) -> Result<Vec<WorkflowRecord>>

pub async fn upsert_search_attributes( &self, workflow_id: &str, patch_json: &str, ) -> Result<()>

pub async fn cancel_workflow( &self, id: &str, reason: Option<&str>, ) -> Result<bool>

pub async fn terminate_workflow( &self, id: &str, reason: Option<&str>, ) -> Result<bool>

pub async fn finalise_cancellation(&self, workflow_id: &str) -> Result<()>

Finalise a cancellation: flips status to CANCELLED and appends the terminal WorkflowCancelled event. Called by the CancelWorkflow command handler (worker acknowledged cancel) and by cancel_workflow directly when the workflow has no worker yet.

pub async fn complete_workflow( &self, workflow_id: &str, result: Option<&str>, ) -> Result<()>

Mark a workflow COMPLETED with a result + append WorkflowCompleted event. If the workflow has a parent, also notifies the parent with a ChildWorkflowCompleted event and marks it dispatchable so it can replay past ctx:start_child_workflow and pick up the child’s result.

pub async fn fail_workflow(&self, workflow_id: &str, error: &str) -> Result<()>

Mark a workflow FAILED with an error + append WorkflowFailed event. Notifies the parent if any (ChildWorkflowFailed).

impl<S: WorkflowStore> WorkflowCtx<S>

pub async fn create_namespace(&self, name: &str) -> Result<()>

pub async fn list_namespaces(&self) -> Result<Vec<NamespaceRecord>>

pub async fn delete_namespace(&self, name: &str) -> Result<bool>

pub async fn get_namespace_stats( &self, namespace: &str, ) -> Result<NamespaceStats>

pub async fn get_queue_stats(&self, namespace: &str) -> Result<Vec<QueueStats>>

pub async fn create_snapshot( &self, workflow_id: &str, event_seq: i32, state_json: &str, ) -> Result<()>

pub async fn get_latest_snapshot( &self, workflow_id: &str, ) -> Result<Option<WorkflowSnapshot>>

impl<S: WorkflowStore> WorkflowCtx<S>

pub async fn create_schedule(&self, schedule: &WorkflowSchedule) -> Result<()>

pub async fn list_schedules( &self, namespace: &str, ) -> Result<Vec<WorkflowSchedule>>

pub async fn get_schedule( &self, namespace: &str, name: &str, ) -> Result<Option<WorkflowSchedule>>

pub async fn delete_schedule(&self, namespace: &str, name: &str) -> Result<bool>

pub async fn update_schedule( &self, namespace: &str, name: &str, patch: &SchedulePatch, ) -> Result<Option<WorkflowSchedule>>

pub async fn set_schedule_paused( &self, namespace: &str, name: &str, paused: bool, ) -> Result<Option<WorkflowSchedule>>

impl<S: WorkflowStore> WorkflowCtx<S>

pub async fn send_signal( &self, workflow_id: &str, name: &str, payload: Option<&str>, ) -> Result<()>

pub async fn get_events(&self, workflow_id: &str) -> Result<Vec<WorkflowEvent>>

impl<S: WorkflowStore> WorkflowCtx<S>

pub async fn claim_workflow_task( &self, task_queue: &str, worker_id: &str, ) -> Result<Option<(WorkflowRecord, Vec<WorkflowEvent>)>>

Claim a dispatchable workflow task on a queue. Returns the workflow record + full event history so the worker can replay the handler deterministically. Atomic — multiple workers polling the same queue will each get a different task or None.

pub async fn submit_workflow_commands( &self, workflow_id: &str, worker_id: &str, commands: &[Value], ) -> Result<()>

Submit a worker’s batch of commands for a workflow it claimed. Each command produces durable events / rows transactionally and the dispatch lease is released on return.

Supported command types:

• ScheduleActivity { seq, name, task_queue, input?, max_attempts?, … }
• CompleteWorkflow { result }
• FailWorkflow { error }

pub async fn schedule_timer( &self, workflow_id: &str, seq: i32, duration_secs: f64, ) -> Result<WorkflowTimer>

Schedule a durable timer for a workflow.

Idempotent on (workflow_id, seq) — a workflow that yields the same ScheduleTimer{seq=N} on retry will reuse the existing timer, not schedule a second one. This is the timer counterpart to schedule_activity’s replay-safe behaviour.

On the first call:

• inserts a row in workflow_timers with fire_at = now + duration
• appends a TimerScheduled event so the worker can replay and know it’s been scheduled (otherwise replays would yield it again)

impl<S: WorkflowStore> WorkflowCtx<S>

pub async fn register_worker(&self, worker: &WorkflowWorker) -> Result<()>

pub async fn heartbeat_worker(&self, id: &str) -> Result<()>

pub async fn list_workers(&self, namespace: &str) -> Result<Vec<WorkflowWorker>>