runtime

func (*Runtime) DB ¶

func (r *Runtime) DB() *gorm.DB

DB exposes the underlying *gorm.DB. With it a consumer can construct any query repository (via repository/query) or run its own read-side queries — the primary extension point for building a business/HTTP layer on top.

func (*Runtime) Engine ¶

func (r *Runtime) Engine() *engine.Engine

Engine exposes the underlying engine for advanced use (replay, redo, cancel, fork, and other power-user operations that are intentionally not on the facade).

func (*Runtime) EventBus ¶

func (r *Runtime) EventBus() *eventbus.EventBus

EventBus exposes the event bus, for publishing custom events or attaching additional listeners beyond Subscribe.

func (*Runtime) NodeRegistry ¶

func (r *Runtime) NodeRegistry() *nodes.NodeRegistry

NodeRegistry exposes the node registry for registering custom node types.

func (*Runtime) RegisterWorkflow ¶

func (r *Runtime) RegisterWorkflow(ctx context.Context, def *definition.WorkflowDefinition) error

RegisterWorkflow 注册工作流定义并立即持久化。 定义变更时自动发布新版本（按 def.Hash() 判定）；此后可通过 Submit(ctx, def.Name, input) 按名称提交任务。

func (*Runtime) Resume ¶

func (r *Runtime) Resume(
	ctx context.Context,
	taskID int64,
	nodeName string,
	meta map[string]any,
) (*RunResult, error)

Resume 用外部事件的结果唤醒挂起中的任务：把 meta 作为 nodeName 节点的输出 闭合该节点，然后在当前调用内同步续跑 DAG，直到下一个终态或挂起点。

典型场景：await/async 节点挂起后，外部回调（webhook、人工审批、异步任务完成） 携带结果到达。节点已被处理时返回 Status "noop"，可安全重复调用。 failed/canceled 的任务请使用 Retry。

func (*Runtime) Retry ¶

func (r *Runtime) Retry(
	ctx context.Context,
	taskID int64,
	resumeFrom string,
	patches []domain.RuntimePatch,
) error

Retry 人工恢复 failed / canceled / suspended 的任务：重置失败子树后重新入队， 由后台 worker（需已 Start）异步执行。

resumeFrom 指定从某节点重跑（空则自动收集失败根节点）；patches 可在恢复前 修正上游节点的输出。手动重试会重置自动重试计数器。

func (*Runtime) Run ¶

func (r *Runtime) Run(
	ctx context.Context,
	def *definition.WorkflowDefinition,
	input map[string]any,
) (*RunResult, error)

Run executes a workflow definition synchronously.

The definition does not need to be registered. If it is registered and unchanged (same hash), the task is linked to the persisted version so it can later be resumed/replayed by version ID.

func (*Runtime) Shutdown ¶

func (r *Runtime) Shutdown() error

Shutdown stops background workers (if started), waits for them to exit, then closes the database connection if owned by this Runtime.

func (*Runtime) Start ¶

func (r *Runtime) Start(ctx context.Context, opts ...StartOption) error

Start launches the background workers that consume Submit-ed tasks:

• task workers: 从任务队列取任务并交给 engine 执行
• recovery scanner: 扫描 crash 的 running 节点并自动恢复
• async workers: 消费异步节点 job 队列
• await poll workers: 轮询 await binding 的超时与到期

Workers run until Shutdown is called or ctx is canceled. Start 只能调用一次；未调用 Start 时 Run 仍可同步执行工作流。

func (*Runtime) Status ¶

func (r *Runtime) Status(ctx context.Context, taskID int64) (*domain.Task, error)

Status returns the current task state.

func (*Runtime) Submit ¶

func (r *Runtime) Submit(
	ctx context.Context,
	workflowName string,
	input map[string]any,
) (int64, error)

Submit enqueues a task for async execution by workflow name and returns the task ID. The workflow must have been registered via RegisterWorkflow (or otherwise persisted); the task is bound to the latest version, which workers resolve to load the definition.

func (*Runtime) Subscribe ¶

func (r *Runtime) Subscribe(eventType string) (<-chan *domain.TaskEvent, func())

Subscribe returns a channel that receives events for the given event type, plus a cancel function that unsubscribes and closes the channel.

func (*Runtime) ToolRegistry ¶

func (r *Runtime) ToolRegistry() *tool.Registry

ToolRegistry exposes the tool registry the engine's tool nodes resolve against. Register additional tools on it (they resolve lazily at node execution) and share it with any service that executes tools directly, so the engine and the business layer see the same tool set.

func (*Runtime) WorkflowRegistry ¶

func (r *Runtime) WorkflowRegistry() *registry.WorkflowRegistry