simple_agents_workflow/yaml_runner/
contracts.rs

1use std::collections::HashMap;
2
3use async_trait::async_trait;
4use serde::{Deserialize, Serialize};
5use serde_json::Value;
6use simple_agent_type::message::Role;
7use simple_agent_type::tool::{ToolChoice, ToolType};
8use thiserror::Error;
9
10use super::{TraceContext, YamlToolTraceMode, YamlWorkflowTraceTenantContext};
11
12#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
13#[serde(rename_all = "snake_case")]
14pub enum YamlWorkflowTokenKind {
15    Output,
16    Thinking,
17}
18
19#[derive(Debug, Clone, PartialEq, Serialize)]
20pub struct YamlWorkflowEvent {
21    pub event_type: String,
22    #[serde(skip_serializing_if = "Option::is_none")]
23    pub node_id: Option<String>,
24    #[serde(skip_serializing_if = "Option::is_none")]
25    pub step_id: Option<String>,
26    #[serde(skip_serializing_if = "Option::is_none")]
27    pub node_kind: Option<String>,
28    #[serde(skip_serializing_if = "Option::is_none")]
29    pub streamable: Option<bool>,
30    #[serde(skip_serializing_if = "Option::is_none")]
31    pub message: Option<String>,
32    #[serde(skip_serializing_if = "Option::is_none")]
33    pub delta: Option<String>,
34    #[serde(skip_serializing_if = "Option::is_none")]
35    pub token_kind: Option<YamlWorkflowTokenKind>,
36    #[serde(skip_serializing_if = "Option::is_none")]
37    pub is_terminal_node_token: Option<bool>,
38    #[serde(skip_serializing_if = "Option::is_none")]
39    pub elapsed_ms: Option<u128>,
40    #[serde(skip_serializing_if = "Option::is_none")]
41    pub metadata: Option<Value>,
42}
43
44pub type WorkflowMessageRole = Role;
45
46#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
47pub struct WorkflowMessage {
48    pub role: WorkflowMessageRole,
49    pub content: String,
50    #[serde(default)]
51    pub name: Option<String>,
52    #[serde(default, alias = "toolCallId")]
53    pub tool_call_id: Option<String>,
54}
55
56#[derive(Debug, Clone, PartialEq, Serialize)]
57pub struct YamlTemplateBinding {
58    pub index: usize,
59    pub expression: String,
60    pub source_path: String,
61    pub resolved: Value,
62    pub resolved_type: String,
63    pub missing: bool,
64}
65
66#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
67pub enum YamlWorkflowDiagnosticSeverity {
68    Error,
69    Warning,
70}
71
72#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
73pub struct YamlWorkflowDiagnostic {
74    pub node_id: Option<String>,
75    pub code: String,
76    pub severity: YamlWorkflowDiagnosticSeverity,
77    pub message: String,
78}
79
80#[derive(Debug, Error)]
81pub enum YamlWorkflowRunError {
82    #[error("failed to read workflow yaml '{path}': {source}")]
83    Read {
84        path: String,
85        source: std::io::Error,
86    },
87    #[error("failed to parse workflow yaml '{path}': {source}")]
88    Parse {
89        path: String,
90        source: serde_yaml::Error,
91    },
92    #[error("rejected workflow yaml '{path}': {reason}")]
93    FileRejected { path: String, reason: String },
94    #[error("workflow '{workflow_id}' has no nodes")]
95    EmptyNodes { workflow_id: String },
96    #[error("entry node '{entry_node}' does not exist")]
97    MissingEntry { entry_node: String },
98    #[error("unknown node id '{node_id}'")]
99    MissingNode { node_id: String },
100    #[error("unsupported node type in '{node_id}'")]
101    UnsupportedNodeType { node_id: String },
102    #[error("unsupported switch condition format: {condition}")]
103    UnsupportedCondition { condition: String },
104    #[error("switch node '{node_id}' has no valid next target")]
105    InvalidSwitchTarget { node_id: String },
106    #[error("llm returned non-object payload for node '{node_id}'")]
107    LlmPayloadNotObject { node_id: String },
108    #[error("custom worker handler '{handler}' is not supported")]
109    UnsupportedCustomHandler { handler: String },
110    #[error("llm execution failed for node '{node_id}': {message}")]
111    Llm { node_id: String, message: String },
112    #[error("custom worker execution failed for node '{node_id}': {message}")]
113    CustomWorker { node_id: String, message: String },
114    #[error("workflow validation failed with {diagnostics_count} error(s)")]
115    Validation {
116        diagnostics_count: usize,
117        diagnostics: Vec<YamlWorkflowDiagnostic>,
118    },
119    #[error("invalid workflow input: {message}")]
120    InvalidInput { message: String },
121    #[error("ir runtime execution failed: {message}")]
122    IrRuntime { message: String },
123    #[error("workflow event stream cancelled: {message}")]
124    EventSinkCancelled { message: String },
125}
126
127pub trait YamlWorkflowEventSink: Send + Sync {
128    fn emit(&self, event: &YamlWorkflowEvent);
129
130    fn is_cancelled(&self) -> bool {
131        false
132    }
133}
134
135/// Returns true for token delta events that should be gated by `workflow_streaming`.
136pub fn is_workflow_stream_delta_event(event_type: &str) -> bool {
137    matches!(
138        event_type,
139        "node_stream_delta" | "node_stream_thinking_delta" | "node_stream_output_delta"
140    )
141}
142
143/// Forwards events to `inner` unless `workflow_streaming` is false and the event is a stream delta.
144pub struct YamlWorkflowStreamFilterSink<'a> {
145    inner: &'a dyn YamlWorkflowEventSink,
146    workflow_streaming: bool,
147}
148
149impl<'a> YamlWorkflowStreamFilterSink<'a> {
150    pub fn new(inner: &'a dyn YamlWorkflowEventSink, workflow_streaming: bool) -> Self {
151        Self {
152            inner,
153            workflow_streaming,
154        }
155    }
156}
157
158impl YamlWorkflowEventSink for YamlWorkflowStreamFilterSink<'_> {
159    fn emit(&self, event: &YamlWorkflowEvent) {
160        if !self.workflow_streaming && is_workflow_stream_delta_event(event.event_type.as_str()) {
161            return;
162        }
163        self.inner.emit(event);
164    }
165
166    fn is_cancelled(&self) -> bool {
167        self.inner.is_cancelled()
168    }
169}
170
171pub struct NoopYamlWorkflowEventSink;
172
173impl YamlWorkflowEventSink for NoopYamlWorkflowEventSink {
174    fn emit(&self, _event: &YamlWorkflowEvent) {}
175}
176
177pub(super) fn workflow_event_sink_cancelled_message() -> &'static str {
178    "workflow event callback cancelled"
179}
180
181pub(super) fn event_sink_is_cancelled(event_sink: Option<&dyn YamlWorkflowEventSink>) -> bool {
182    event_sink.map(|sink| sink.is_cancelled()).unwrap_or(false)
183}
184
185#[derive(Debug, Clone, PartialEq, Eq, Error)]
186pub enum YamlToIrError {
187    #[error("entry node '{entry_node}' does not exist")]
188    MissingEntry { entry_node: String },
189    #[error("node '{node_id}' has multiple outgoing edges in YAML; IR llm/tool nodes require one")]
190    MultipleOutgoingEdge { node_id: String },
191    #[error("node '{node_id}' is unsupported for IR conversion: {reason}")]
192    UnsupportedNode { node_id: String, reason: String },
193}
194
195#[derive(Debug, Clone)]
196pub struct YamlLlmExecutionRequest {
197    pub node_id: String,
198    pub is_terminal_node: bool,
199    pub stream_json_as_text: bool,
200    pub model: String,
201    pub max_tokens: Option<u32>,
202    pub temperature: Option<f32>,
203    pub top_p: Option<f32>,
204    pub messages: Option<Vec<super::Message>>,
205    pub append_prompt_as_user: bool,
206    pub prompt: String,
207    pub prompt_template: String,
208    pub prompt_bindings: Vec<YamlTemplateBinding>,
209    pub schema: Value,
210    pub stream: bool,
211    pub heal: bool,
212    pub tools: Vec<YamlResolvedTool>,
213    pub tool_choice: Option<ToolChoice>,
214    pub max_tool_roundtrips: u8,
215    pub tool_calls_global_key: Option<String>,
216    pub tool_trace_mode: YamlToolTraceMode,
217    pub execution_context: Value,
218    pub trace_id: Option<String>,
219    pub trace_context: Option<TraceContext>,
220    pub tenant_context: YamlWorkflowTraceTenantContext,
221    pub trace_sampled: bool,
222}
223
224#[derive(Debug, Clone)]
225pub struct YamlResolvedTool {
226    pub definition: super::ToolDefinition,
227    pub output_schema: Option<Value>,
228}
229
230#[async_trait]
231pub trait YamlWorkflowLlmExecutor: Send + Sync {
232    async fn complete_structured(
233        &self,
234        request: YamlLlmExecutionRequest,
235        event_sink: Option<&dyn YamlWorkflowEventSink>,
236    ) -> Result<super::YamlLlmExecutionResult, String>;
237}
238
239#[async_trait]
240pub trait YamlWorkflowCustomWorkerExecutor: Send + Sync {
241    async fn execute(
242        &self,
243        handler: &str,
244        handler_file: Option<&str>,
245        payload: &Value,
246        context: &Value,
247    ) -> Result<Value, String>;
248}
249
250#[derive(Debug, Clone, Deserialize)]
251#[serde(deny_unknown_fields)]
252pub struct YamlGlobalUpdate {
253    pub op: String,
254    pub from: Option<String>,
255    pub by: Option<f64>,
256}
257
258#[derive(Debug, Clone, Deserialize)]
259#[serde(deny_unknown_fields)]
260pub struct YamlNodeConfig {
261    pub prompt: Option<String>,
262    #[serde(default, alias = "schema")]
263    pub output_schema: Option<Value>,
264    pub payload: Option<Value>,
265    pub set_globals: Option<HashMap<String, String>>,
266    pub update_globals: Option<HashMap<String, YamlGlobalUpdate>>,
267}
268
269#[derive(Debug, Clone, Deserialize)]
270#[serde(deny_unknown_fields)]
271pub struct YamlCustomWorker {
272    pub handler: String,
273    pub handler_file: Option<String>,
274}
275
276#[derive(Debug, Clone, Deserialize)]
277#[serde(deny_unknown_fields)]
278pub struct YamlSwitchBranch {
279    pub condition: String,
280    pub target: String,
281}
282
283#[derive(Debug, Clone, Deserialize)]
284#[serde(deny_unknown_fields)]
285pub struct YamlSwitch {
286    #[serde(default)]
287    pub branches: Vec<YamlSwitchBranch>,
288    pub default: String,
289}
290
291#[derive(Debug, Clone, Deserialize)]
292#[serde(untagged)]
293pub enum YamlToolChoiceConfig {
294    Mode(super::ToolChoiceMode),
295    Function(YamlToolChoiceFunction),
296    OpenAi(super::ToolChoiceTool),
297}
298
299#[derive(Debug, Clone, Deserialize)]
300#[serde(deny_unknown_fields)]
301pub struct YamlToolChoiceFunction {
302    pub function: String,
303}
304
305#[derive(Debug, Clone, Deserialize)]
306#[serde(deny_unknown_fields)]
307pub struct YamlSimplifiedToolDeclaration {
308    pub name: String,
309    pub description: Option<String>,
310    pub input_schema: Value,
311    pub output_schema: Option<Value>,
312}
313
314#[derive(Debug, Clone, Deserialize)]
315#[serde(deny_unknown_fields)]
316pub struct YamlOpenAiToolFunction {
317    pub name: String,
318    pub description: Option<String>,
319    pub parameters: Option<Value>,
320    pub output_schema: Option<Value>,
321}
322
323#[derive(Debug, Clone, Deserialize)]
324#[serde(deny_unknown_fields)]
325pub struct YamlOpenAiToolDeclaration {
326    #[serde(rename = "type")]
327    pub tool_type: Option<ToolType>,
328    pub function: YamlOpenAiToolFunction,
329}
330
331#[derive(Debug, Clone, Deserialize)]
332#[serde(untagged)]
333pub enum YamlToolDeclaration {
334    OpenAi(YamlOpenAiToolDeclaration),
335    Simplified(YamlSimplifiedToolDeclaration),
336}
337
338#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize, Default)]
339#[serde(rename_all = "snake_case")]
340pub enum YamlToolFormat {
341    #[default]
342    Openai,
343    Simplified,
344}
345
346#[derive(Debug, Clone, Deserialize)]
347#[serde(deny_unknown_fields)]
348pub struct YamlLlmCall {
349    pub model: String,
350    pub max_tokens: Option<u32>,
351    pub temperature: Option<f32>,
352    pub top_p: Option<f32>,
353    pub stream: Option<bool>,
354    pub stream_json_as_text: Option<bool>,
355    pub heal: Option<bool>,
356    pub messages_path: Option<String>,
357    pub append_prompt_as_user: Option<bool>,
358    #[serde(default)]
359    pub tools_format: YamlToolFormat,
360    #[serde(default)]
361    pub tools: Vec<YamlToolDeclaration>,
362    pub tool_choice: Option<YamlToolChoiceConfig>,
363    pub max_tool_roundtrips: Option<u8>,
364    pub tool_calls_global_key: Option<String>,
365}
366
367#[derive(Debug, Clone, Deserialize)]
368#[serde(deny_unknown_fields)]
369pub struct YamlNodeType {
370    pub llm_call: Option<YamlLlmCall>,
371    pub switch: Option<YamlSwitch>,
372    pub custom_worker: Option<YamlCustomWorker>,
373    pub end: Option<Value>,
374}
375
376#[derive(Debug, Clone, Deserialize)]
377#[serde(deny_unknown_fields)]
378pub struct YamlNode {
379    pub id: String,
380    #[serde(default)]
381    pub name: Option<String>,
382    pub node_type: YamlNodeType,
383    pub config: Option<YamlNodeConfig>,
384}
385
386impl YamlNode {
387    pub(super) fn kind_name(&self) -> &'static str {
388        if self.node_type.llm_call.is_some() {
389            "llm_call"
390        } else if self.node_type.switch.is_some() {
391            "switch"
392        } else if self.node_type.custom_worker.is_some() {
393            "custom_worker"
394        } else if self.node_type.end.is_some() {
395            "end"
396        } else {
397            "unknown"
398        }
399    }
400}
401
402#[derive(Debug, Clone, Deserialize)]
403#[serde(deny_unknown_fields)]
404pub struct YamlEdge {
405    pub from: String,
406    pub to: String,
407}
408
409#[derive(Debug, Clone, Deserialize)]
410#[serde(deny_unknown_fields)]
411pub struct YamlWorkflow {
412    pub id: String,
413    #[serde(default)]
414    pub version: Option<String>,
415    #[serde(default)]
416    pub metadata: Option<HashMap<String, Value>>,
417    pub entry_node: String,
418    #[serde(default)]
419    pub nodes: Vec<YamlNode>,
420    #[serde(default)]
421    pub edges: Vec<YamlEdge>,
422}
simple_agents_workflow/yaml_runner/contracts.rs

simple_agents_workflow/yaml_runner/
contracts.rs
