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.
40pub type WorkflowStepFuture<'a> =
41    Pin<Box<dyn Future<Output = Result<Value, WorkflowStepError>> + Send + 'a>>;
42
43/// One executable, provider-neutral workflow node.
44///
45/// External work must be represented by capabilities passed to the workflow
46/// builder. The scheduler relies on that declaration to attenuate authority
47/// and to reject write-capable branches from first-success races.
48pub trait WorkflowStep: Send + Sync {
49    /// Executes from canonical JSON input to canonical JSON output.
50    fn execute<'a>(&'a self, input: Value, run: &'a RunContext) -> WorkflowStepFuture<'a>;
51}
52
53/// Pure, synchronous branch decision over canonical workflow data.
54pub trait WorkflowCondition: Send + Sync {
55    /// Selects the true or false branch.
56    ///
57    /// # Errors
58    ///
59    /// Returns [`WorkflowStepError`] when the canonical input cannot be
60    /// evaluated safely.
61    fn evaluate(&self, input: &Value) -> Result<bool, WorkflowStepError>;
62}
63
64/// Adapts a closure into a durable workflow branch condition.
65pub struct PredicateCondition<F> {
66    predicate: F,
67}
68
69impl<F> PredicateCondition<F> {
70    /// Creates a condition from a pure predicate.
71    pub const fn new(predicate: F) -> Self {
72        Self { predicate }
73    }
74}
75
76impl<F> WorkflowCondition for PredicateCondition<F>
77where
78    F: Fn(&Value) -> Result<bool, WorkflowStepError> + Send + Sync,
79{
80    fn evaluate(&self, input: &Value) -> Result<bool, WorkflowStepError> {
81        (self.predicate)(input)
82    }
83}
84
85impl<F> fmt::Debug for PredicateCondition<F> {
86    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
87        formatter
88            .debug_struct("PredicateCondition")
89            .finish_non_exhaustive()
90    }
91}
92
93/// Adapts a Runifold Agent to the canonical workflow step boundary.
94#[derive(Clone, Debug)]
95pub struct AgentStep {
96    agent: Arc<Agent>,
97}
98
99impl AgentStep {
100    /// Creates an Agent-backed step.
101    pub const fn new(agent: Arc<Agent>) -> Self {
102        Self { agent }
103    }
104
105    /// Returns the wrapped Agent.
106    pub const fn agent(&self) -> &Arc<Agent> {
107        &self.agent
108    }
109}
110
111/// Canonical Agent workflow value.
112///
113/// `input` is the concatenated terminal text consumed automatically by a
114/// following Agent step. `outcome` retains the full response and transcript.
115#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
116pub struct AgentStepOutput {
117    /// Text forwarded to a following Agent step.
118    pub input: String,
119    /// Complete canonical Agent result.
120    pub outcome: AgentOutcome,
121}
122
123impl WorkflowStep for AgentStep {
124    fn execute<'a>(&'a self, input: Value, run: &'a RunContext) -> WorkflowStepFuture<'a> {
125        Box::pin(async move {
126            let prompt = match input {
127                Value::String(prompt) => prompt,
128                Value::Object(mut object) => object
129                    .remove("input")
130                    .and_then(|value| value.as_str().map(ToOwned::to_owned))
131                    .ok_or_else(|| {
132                        WorkflowStepError::InvalidInput(
133                            "Agent steps require a string or an object containing string `input`"
134                                .into(),
135                        )
136                    })?,
137                _ => {
138                    return Err(WorkflowStepError::InvalidInput(
139                        "Agent steps require a string or an object containing string `input`"
140                            .into(),
141                    ));
142                }
143            };
144            let outcome = self.agent.run(prompt, run).await?;
145            let input = agent_text(&outcome)?;
146            Ok(serde_json::to_value(AgentStepOutput { input, outcome })?)
147        })
148    }
149}
150
151fn agent_text(outcome: &AgentOutcome) -> Result<String, WorkflowStepError> {
152    if outcome
153        .response
154        .content
155        .iter()
156        .any(|part| matches!(part, ContentPart::Refusal { .. }))
157    {
158        return Err(WorkflowStepError::InvalidOutput(
159            "Agent returned a refusal that cannot be forwarded automatically".into(),
160        ));
161    }
162    let text = outcome
163        .response
164        .content
165        .iter()
166        .filter_map(|part| match part {
167            ContentPart::Text { text } => Some(text.as_str()),
168            _ => None,
169        })
170        .collect::<String>();
171    if text.is_empty() {
172        return Err(WorkflowStepError::InvalidOutput(
173            "Agent returned no terminal text to forward".into(),
174        ));
175    }
176    Ok(text)
177}
runifold_workflow/step.rs

runifold_workflow/
step.rs
