runtime

type AwaitController ¶

type AwaitController interface {
	// Begin 为 async 节点登记一个等待（提交外部请求、写 AwaitBinding）。
	// 返回后节点进入 NodeAwaiting，由外部事件/poll 通过 Scheduler.Resume 唤醒。
	Begin(ctx context.Context, node *PlanNode, input map[string]any) (bindingID int64, err error)
}

AwaitController 异步原语：提交外部任务 + 建 AwaitBinding，等待外部唤醒。 现有 engine 的 executeAwaitNode / CompleteAwaitNode + AwaitBindingRepository 适配它。

type Emitter ¶

type Emitter interface {
	Emit(event Event)
}

Emitter 事件端口（现有 nodes.Context.EmitToolEvent 适配它）。

type Event ¶

type Event struct {
	Node     string
	Type     string
	Message  string
	Progress float64
	Data     map[string]any
}

type ExecState ¶

type ExecState interface {
	Input() map[string]any // 任务级输入
	Output(node string) map[string]any
	SetOutput(node string, out map[string]any)

	State(node string) NodeState
	Transition(node string, to NodeState)

	// Nodes 返回当前已知节点名（动态前沿下会随 planner 增长）
	Nodes() []string
}

ExecState 是底座读写的"活"执行上下文。 现有的 workflow/nodes.Context（持 Output/Runtime + expr 环境）将实现这个接口， 从而老路径无需改数据结构即可跑在新底座上。

type FileAwait ¶

type FileAwait struct {
	Dir string
	// contains filtered or unexported fields
}

FileAwait 实现 AwaitController 端口：在内存中管理 binding，同时持久化到文件。 B-M1b 验证用：crash 后从 binding 文件恢复。

func NewFileAwait(dir string) *FileAwait

func (fa *FileAwait) Begin(_ context.Context, node *PlanNode, input map[string]any) (int64, error)

func (fa *FileAwait) BindingIDs() []int64

func (fa *FileAwait) Complete(bindingID int64) (nodeName string, input map[string]any, ok bool)

type FileStore ¶

type FileStore struct {
	Dir   string
	State *MemState // 共享引用——Scheduler 和 Store 操作同一份 MemState
	// contains filtered or unexported fields
}

FileStore 实现 Store 端口，每次 PersistNode 时把完整 ExecState 快照写到文件。 B-M1b 验证用：crash 后从快照恢复状态。

func NewFileStore(dir string, state *MemState) *FileStore

func (fs *FileStore) PersistNode(_ context.Context, _ string, _ NodeState, _ map[string]any) error

type InputResolver ¶

type InputResolver func(ctx context.Context, state ExecState) (map[string]any, error)

InputResolver：给定当前执行状态，产出本节点入参。失败即节点失败。

type Invoker ¶

type Invoker interface {
	Invoke(ctx context.Context, toolName string, input map[string]any, emit Emitter) (map[string]any, error)
}

Invoker 执行一次工具调用。tool.Registry 适配它（Tool-First）。

type JoinKind ¶

type JoinKind uint8

const (
	JoinAll JoinKind = iota // 所有依赖 success 才就绪（默认）
	JoinAny                 // 任一依赖 success 即就绪（条件分支 / merge）
)

type MemState ¶

type MemState struct {
	// contains filtered or unexported fields
}

MemState 是 ExecState 的最小内存实现，供 demo / planner / 测试复用（纯 stdlib）。 生产环境中 nodes.Context 也实现 ExecState（见 workflow 包的适配）。

func LoadSnapshot(path string) (*MemState, error)

func NewMemState(input map[string]any) *MemState

func (m *MemState) Input() map[string]any

func (m *MemState) Nodes() []string

func (m *MemState) Output(node string) map[string]any

func (m *MemState) SaveSnapshot(path string) error

func (m *MemState) SetOutput(node string, o map[string]any)

func (m *MemState) State(node string) NodeState

func (m *MemState) Transition(node string, to NodeState)

type NodeState ¶

type NodeState uint8

NodeState 运行时节点状态（自包含，不依赖 domain）。 适配器层负责与 domain.NodeRuntime 的状态互转。

const (
	NodePending NodeState = iota
	NodeRunning
	NodeAwaiting // 已挂起等外部事件（对应 domain.NodeAwaiting）
	NodeSuccess
	NodeFailed
	NodeSkipped
)

func (s NodeState) Terminal() bool

type NopEmitter ¶

type NopEmitter struct{}

NopEmitter 是无操作的事件发射器（B-M1b 这类同步验证不需要 trace）。

func (NopEmitter) Emit(Event)

type Plan ¶

type Plan struct {
	Nodes map[string]*PlanNode
}

Plan 是一张"已解析的可执行 DAG"。谁产出的不重要：

• workflow 编译器把人写的 WorkflowDefinition 编译成 Plan
• planner 把 Goal 增量生成成 Plan
• SDK/代码直接拼 Plan

func FanoutPlan(n int) *Plan

DAG: [async_0, async_1, ..., async_N-1] → join
所有 async 节点无依赖（并行提交），join 依赖所有 async（JoinAll）。
join 的 Resolve 收集所有分支的输出（包括失败分支——通过 state 检查）。

func SimplePlan() *Plan

type PlanNode ¶

type PlanNode struct {
	Name      string
	ToolName  string   // 节点要调用的工具（registry 里查）
	DependsOn []string // 纯拓扑依赖边（不含条件）

	// Async 取代 definition 的 await/async node type。
	// true → 走"挂起 + AwaitBinding + 外部唤醒"路径（对应 executor.go:133）。
	Async bool

	// Join 决定依赖满足语义：map/loop 的并行汇聚下沉成的原语。
	Join JoinKind

	// Resolve 在执行到本节点时、用实时上游产出算出本节点入参。
	// workflow 编译器 → expr 求值实现；planner → 直接给值。expr-lang 因此被关在前端。
	Resolve InputResolver

	Retry RetryPolicy
}

PlanNode 一个可执行单元 = 一次工具调用（Tool-First）。 注意这里没有 input_mapping 表达式、没有 edge condition —— 那些是编排语义， 已在前端编译期解析掉，只留下纯依赖边 + 一个 InputResolver 回调。

type PlanSource ¶

type PlanSource interface {
	// Next 返回新增（追加到计划中）的节点。底座负责依赖求解与调度，
	// 因此 Source 只需"产出节点"，不需要自己算就绪顺序。
	Next(ctx context.Context, state ExecState) (added []*PlanNode, done bool, err error)
}

PlanSource 是底座的"计划来源"。这是让 workflow 与 planner 可互换的关键抽象：

• 静态 workflow：首次返回整张 DAG，done=true（退化情形）
• planner/agent：每次根据已完成结果返回下一步，done 由 LLM 判定（增量规划）

func FanoutPlanSource(n int) PlanSource

func SimplePlanSource() PlanSource

type ProviderAwait ¶

type ProviderAwait struct {
	Dir         string
	ProviderURL string
	HTTPClient  *http.Client
	// contains filtered or unexported fields
}

ProviderAwait 实现 AwaitController，对真实 HTTP Provider 做 submit。 Begin() → POST /submit → 拿到 provider_task_id → 存入 binding。 crash 恢复时从文件重建 binding 映射。

func NewProviderAwait(dir, providerURL string) *ProviderAwait

func (pa *ProviderAwait) Begin(_ context.Context, node *PlanNode, input map[string]any) (int64, error)

func (pa *ProviderAwait) BindingIDs() []int64

func (pa *ProviderAwait) NodeForBinding(bindingID int64) (string, bool)

func (pa *ProviderAwait) PollProvider(taskID string) (status string, result map[string]any, err error)

func (pa *ProviderAwait) ProviderTaskID(bindingID int64) (string, bool)

type Result ¶

type Result struct {
	Status RunStatus
	Err    error
}

type RetryPolicy ¶

type RetryPolicy struct {
	MaxRetries int
	Interval   time.Duration
}

type RunStatus ¶

type RunStatus uint8

const (
	StatusCompleted RunStatus = iota
	StatusSuspended           // 有节点在 NodeAwaiting，等外部唤醒后再 Resume
	StatusFailed
)

type Scheduler ¶

type Scheduler struct {
	// contains filtered or unexported fields
}

Scheduler 是底座的执行器：消费 PlanSource，做依赖求解，跑就绪节点， 处理 sync 内联 / async 挂起。它取代 engine.runDAG 的拓扑循环，但不认识 definition。

func NewScheduler(inv Invoker, aw AwaitController, st Store, em Emitter) *Scheduler

func (s *Scheduler) Resume(ctx context.Context, src PlanSource, state ExecState, node string, out map[string]any) (Result, error)

func (s *Scheduler) Run(ctx context.Context, src PlanSource, state ExecState) (Result, error)

func (s *Scheduler) WithMaxSteps(n int) *Scheduler

func (s *Scheduler) WithPlan(p *Plan) *Scheduler

func (s *Scheduler) WithTrace(sink TraceSink, runID string) *Scheduler

type StaticSource ¶

type StaticSource struct {
	// contains filtered or unexported fields
}

StaticSource 把一张完整的 Plan 包成"一次性给完"的 Source —— 这就是让 老 workflow 路径在新底座上原样跑通的垫片（shim）。

用法：plan := workflow.Compile(def); scheduler.Run(ctx, runtime.NewStaticSource(plan), state)

func NewStaticSource(plan *Plan) *StaticSource

func (s *StaticSource) Next(_ context.Context, _ ExecState) ([]*PlanNode, bool, error)

type Store ¶

type Store interface {
	PersistNode(ctx context.Context, node string, state NodeState, out map[string]any) error
}

Store 持久化端口：节点状态/输出落库（现有 NodeRuntime 持久化适配它）。

type TraceClass ¶

type TraceClass uint8

TraceClass 区分两条语义流，但二者共享同一条 Seq 全序。

const (
	// ClassExecution 确定性流：节点生命周期 + tool I/O。
	ClassExecution TraceClass = iota
	// ClassControl 非确定性流：planner/编译器的计划产出（plan_extend）。
	// "agent 的自由"是内核外部性，一旦记录就被固化为确定性事实。
	ClassControl
)

type TraceEvent ¶

type TraceEvent struct {
	RunID string
	// Seq 是全序真相：execution 与 control 共用同一单调序列，
	// 跨流因果（"先有 Y.output，planner 才产出 X"）才可回放。回放靠 Seq，不靠 wall-clock。
	Seq     int64
	Class   TraceClass
	Node    string // plan_extend 时为空
	Type    TraceType
	Payload map[string]any
}

TraceEvent 一条不可变的执行真相记录。

type TraceSink ¶

type TraceSink interface {
	EmitExecution(e TraceEvent)
	EmitControl(e TraceEvent)
}

TraceSink 双写协议。两个方法只是语义分类；Seq 由唯一写入者（Scheduler）统一盖章， 保证两流共享单序列。nil sink ⇒ 完全不记录（sidecar 默认关闭，决策无副作用）。

type TraceType ¶

type TraceType string

const (
	TraceNodeStart  TraceType = "node_start"
	TraceInput      TraceType = "input"  // resolved input（回放时的确定性入参）
	TraceOutput     TraceType = "output" // tool 产出（回放时要注入的外部不确定性）
	TraceAwait      TraceType = "await"
	TraceResume     TraceType = "resume"
	TraceFail       TraceType = "fail"
	TracePlanExtend TraceType = "plan_extend" // control 流：planner 产出了哪些节点
)