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    /// Split thinking/output stream events (mirrors [`YamlWorkflowExecutionFlags::split_stream_deltas`]).
223    pub split_stream_deltas: bool,
224}
225
226#[derive(Debug, Clone)]
227pub struct YamlResolvedTool {
228    pub definition: super::ToolDefinition,
229    pub output_schema: Option<Value>,
230}
231
232#[async_trait]
233pub trait YamlWorkflowLlmExecutor: Send + Sync {
234    async fn complete_structured(
235        &self,
236        request: YamlLlmExecutionRequest,
237        event_sink: Option<&dyn YamlWorkflowEventSink>,
238    ) -> Result<super::YamlLlmExecutionResult, String>;
239}
240
241#[async_trait]
242pub trait YamlWorkflowCustomWorkerExecutor: Send + Sync {
243    async fn execute(
244        &self,
245        handler: &str,
246        handler_file: Option<&str>,
247        payload: &Value,
248        context: &Value,
249    ) -> Result<Value, String>;
250}
251
252#[derive(Debug, Clone, Deserialize)]
253#[serde(deny_unknown_fields)]
254pub struct YamlGlobalUpdate {
255    pub op: String,
256    pub from: Option<String>,
257    pub by: Option<f64>,
258}
259
260#[derive(Debug, Clone, Deserialize)]
261#[serde(deny_unknown_fields)]
262pub struct YamlNodeConfig {
263    pub prompt: Option<String>,
264    #[serde(default, alias = "schema")]
265    pub output_schema: Option<Value>,
266    pub payload: Option<Value>,
267    pub set_globals: Option<HashMap<String, String>>,
268    pub update_globals: Option<HashMap<String, YamlGlobalUpdate>>,
269}
270
271#[derive(Debug, Clone, Deserialize)]
272#[serde(deny_unknown_fields)]
273pub struct YamlCustomWorker {
274    pub handler: String,
275    pub handler_file: Option<String>,
276}
277
278#[derive(Debug, Clone, Deserialize)]
279#[serde(deny_unknown_fields)]
280pub struct YamlSwitchBranch {
281    pub condition: String,
282    pub target: String,
283}
284
285#[derive(Debug, Clone, Deserialize)]
286#[serde(deny_unknown_fields)]
287pub struct YamlSwitch {
288    #[serde(default)]
289    pub branches: Vec<YamlSwitchBranch>,
290    pub default: String,
291}
292
293#[derive(Debug, Clone, Deserialize)]
294#[serde(untagged)]
295pub enum YamlToolChoiceConfig {
296    Mode(super::ToolChoiceMode),
297    Function(YamlToolChoiceFunction),
298    OpenAi(super::ToolChoiceTool),
299}
300
301#[derive(Debug, Clone, Deserialize)]
302#[serde(deny_unknown_fields)]
303pub struct YamlToolChoiceFunction {
304    pub function: String,
305}
306
307#[derive(Debug, Clone, Deserialize)]
308#[serde(deny_unknown_fields)]
309pub struct YamlSimplifiedToolDeclaration {
310    pub name: String,
311    pub description: Option<String>,
312    pub input_schema: Value,
313    pub output_schema: Option<Value>,
314}
315
316#[derive(Debug, Clone, Deserialize)]
317#[serde(deny_unknown_fields)]
318pub struct YamlOpenAiToolFunction {
319    pub name: String,
320    pub description: Option<String>,
321    pub parameters: Option<Value>,
322    pub output_schema: Option<Value>,
323}
324
325#[derive(Debug, Clone, Deserialize)]
326#[serde(deny_unknown_fields)]
327pub struct YamlOpenAiToolDeclaration {
328    #[serde(rename = "type")]
329    pub tool_type: Option<ToolType>,
330    pub function: YamlOpenAiToolFunction,
331}
332
333#[derive(Debug, Clone, Deserialize)]
334#[serde(untagged)]
335pub enum YamlToolDeclaration {
336    OpenAi(YamlOpenAiToolDeclaration),
337    Simplified(YamlSimplifiedToolDeclaration),
338}
339
340#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize, Default)]
341#[serde(rename_all = "snake_case")]
342pub enum YamlToolFormat {
343    #[default]
344    Openai,
345    Simplified,
346}
347
348#[derive(Debug, Clone, Deserialize)]
349#[serde(deny_unknown_fields)]
350pub struct YamlLlmCall {
351    pub model: String,
352    pub max_tokens: Option<u32>,
353    pub temperature: Option<f32>,
354    pub top_p: Option<f32>,
355    pub stream: Option<bool>,
356    pub stream_json_as_text: Option<bool>,
357    pub heal: Option<bool>,
358    pub messages_path: Option<String>,
359    pub append_prompt_as_user: Option<bool>,
360    #[serde(default)]
361    pub tools_format: YamlToolFormat,
362    #[serde(default)]
363    pub tools: Vec<YamlToolDeclaration>,
364    pub tool_choice: Option<YamlToolChoiceConfig>,
365    pub max_tool_roundtrips: Option<u8>,
366    pub tool_calls_global_key: Option<String>,
367}
368
369#[derive(Debug, Clone, Deserialize)]
370#[serde(deny_unknown_fields)]
371pub struct YamlNodeType {
372    pub llm_call: Option<YamlLlmCall>,
373    pub switch: Option<YamlSwitch>,
374    pub custom_worker: Option<YamlCustomWorker>,
375    pub end: Option<Value>,
376}
377
378#[derive(Debug, Clone, Deserialize)]
379#[serde(deny_unknown_fields)]
380pub struct YamlNode {
381    pub id: String,
382    #[serde(default)]
383    pub name: Option<String>,
384    pub node_type: YamlNodeType,
385    pub config: Option<YamlNodeConfig>,
386}
387
388impl YamlNode {
389    pub(super) fn kind_name(&self) -> &'static str {
390        if self.node_type.llm_call.is_some() {
391            "llm_call"
392        } else if self.node_type.switch.is_some() {
393            "switch"
394        } else if self.node_type.custom_worker.is_some() {
395            "custom_worker"
396        } else if self.node_type.end.is_some() {
397            "end"
398        } else {
399            "unknown"
400        }
401    }
402}
403
404#[derive(Debug, Clone, Deserialize)]
405#[serde(deny_unknown_fields)]
406pub struct YamlEdge {
407    pub from: String,
408    pub to: String,
409}
410
411#[derive(Debug, Clone, Deserialize)]
412#[serde(deny_unknown_fields)]
413pub struct YamlWorkflow {
414    pub id: String,
415    #[serde(default)]
416    pub version: Option<String>,
417    #[serde(default)]
418    pub metadata: Option<HashMap<String, Value>>,
419    pub entry_node: String,
420    #[serde(default)]
421    pub nodes: Vec<YamlNode>,
422    #[serde(default)]
423    pub edges: Vec<YamlEdge>,
424}
simple_agents_workflow/yaml_runner/contracts.rs

simple_agents_workflow/yaml_runner/
contracts.rs
