runifold_workflow/
step.rs

1use std::{fmt, future::Future, pin::Pin, sync::Arc};
2
3use runifold_agent::{Agent, AgentOutcome};
4use runifold_core::RunContext;
5use runifold_model::ContentPart;
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8
9use crate::WorkflowStepError;
10
11/// Stable workflow node identity used by checkpoints and event streams.
12#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
13#[serde(transparent)]
14pub struct StepId(String);
15
16impl StepId {
17    pub(crate) fn parse(value: impl Into<String>) -> Result<Self, String> {
18        let value = value.into();
19        let valid = !value.is_empty()
20            && value.len() <= 128
21            && value
22                .bytes()
23                .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.'));
24        valid.then_some(Self(value.clone())).ok_or(value)
25    }
26
27    /// Returns the stable identifier text.
28    pub fn as_str(&self) -> &str {
29        &self.0
30    }
31}
32
33impl fmt::Display for StepId {
34    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
35        self.0.fmt(formatter)
36    }
37}
38
39/// Boxed asynchronous workflow-step result.
40#[cfg(not(target_arch = "wasm32"))]
41pub type WorkflowStepFuture<'a> =
42    Pin<Box<dyn Future<Output = Result<Value, WorkflowStepError>> + Send + 'a>>;
43
44/// Boxed workflow-step result on single-threaded WASM.
45#[cfg(target_arch = "wasm32")]
46pub type WorkflowStepFuture<'a> =
47    Pin<Box<dyn Future<Output = Result<Value, WorkflowStepError>> + 'a>>;
48
49/// One executable, provider-neutral workflow node.
50///
51/// External work must be represented by capabilities passed to the workflow
52/// builder. The scheduler relies on that declaration to attenuate authority
53/// and to reject write-capable branches from first-success races.
54pub trait WorkflowStep: Send + Sync {
55    /// Executes from canonical JSON input to canonical JSON output.
56    fn execute<'a>(&'a self, input: Value, run: &'a RunContext) -> WorkflowStepFuture<'a>;
57}
58
59/// Pure, synchronous branch decision over canonical workflow data.
60pub trait WorkflowCondition: Send + Sync {
61    /// Selects the true or false branch.
62    ///
63    /// # Errors
64    ///
65    /// Returns [`WorkflowStepError`] when the canonical input cannot be
66    /// evaluated safely.
67    fn evaluate(&self, input: &Value) -> Result<bool, WorkflowStepError>;
68}
69
70/// Adapts a closure into a durable workflow branch condition.
71pub struct PredicateCondition<F> {
72    predicate: F,
73}
74
75impl<F> PredicateCondition<F> {
76    /// Creates a condition from a pure predicate.
77    pub const fn new(predicate: F) -> Self {
78        Self { predicate }
79    }
80}
81
82impl<F> WorkflowCondition for PredicateCondition<F>
83where
84    F: Fn(&Value) -> Result<bool, WorkflowStepError> + Send + Sync,
85{
86    fn evaluate(&self, input: &Value) -> Result<bool, WorkflowStepError> {
87        (self.predicate)(input)
88    }
89}
90
91impl<F> fmt::Debug for PredicateCondition<F> {
92    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
93        formatter
94            .debug_struct("PredicateCondition")
95            .finish_non_exhaustive()
96    }
97}
98
99/// Adapts a Runifold Agent to the canonical workflow step boundary.
100#[derive(Clone, Debug)]
101pub struct AgentStep {
102    agent: Arc<Agent>,
103}
104
105impl AgentStep {
106    /// Creates an Agent-backed step.
107    pub const fn new(agent: Arc<Agent>) -> Self {
108        Self { agent }
109    }
110
111    /// Returns the wrapped Agent.
112    pub const fn agent(&self) -> &Arc<Agent> {
113        &self.agent
114    }
115}
116
117/// Canonical Agent workflow value.
118///
119/// `input` is the concatenated terminal text consumed automatically by a
120/// following Agent step. `outcome` retains the full response and transcript.
121#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
122pub struct AgentStepOutput {
123    /// Text forwarded to a following Agent step.
124    pub input: String,
125    /// Complete canonical Agent result.
126    pub outcome: AgentOutcome,
127}
128
129impl WorkflowStep for AgentStep {
130    fn execute<'a>(&'a self, input: Value, run: &'a RunContext) -> WorkflowStepFuture<'a> {
131        Box::pin(async move {
132            let prompt = match input {
133                Value::String(prompt) => prompt,
134                Value::Object(mut object) => object
135                    .remove("input")
136                    .and_then(|value| value.as_str().map(ToOwned::to_owned))
137                    .ok_or_else(|| {
138                        WorkflowStepError::InvalidInput(
139                            "Agent steps require a string or an object containing string `input`"
140                                .into(),
141                        )
142                    })?,
143                _ => {
144                    return Err(WorkflowStepError::InvalidInput(
145                        "Agent steps require a string or an object containing string `input`"
146                            .into(),
147                    ));
148                }
149            };
150            let outcome = self.agent.run(prompt, run).await?;
151            let input = agent_text(&outcome)?;
152            Ok(serde_json::to_value(AgentStepOutput { input, outcome })?)
153        })
154    }
155}
156
157fn agent_text(outcome: &AgentOutcome) -> Result<String, WorkflowStepError> {
158    if outcome
159        .response
160        .content
161        .iter()
162        .any(|part| matches!(part, ContentPart::Refusal { .. }))
163    {
164        return Err(WorkflowStepError::InvalidOutput(
165            "Agent returned a refusal that cannot be forwarded automatically".into(),
166        ));
167    }
168    let text = outcome
169        .response
170        .content
171        .iter()
172        .filter_map(|part| match part {
173            ContentPart::Text { text } => Some(text.as_str()),
174            _ => None,
175        })
176        .collect::<String>();
177    if text.is_empty() {
178        return Err(WorkflowStepError::InvalidOutput(
179            "Agent returned no terminal text to forward".into(),
180        ));
181    }
182    Ok(text)
183}
runifold_workflow/step.rs

runifold_workflow/
step.rs
