af_workflow/
node.rs

1//! Node trait + context. Port of `platform/node_protocol.py`.
2//!
3//! A spec has two node kinds. Ingress nodes are pure event sources and are
4//! handled by the runner (next phase); this module models the **step node** —
5//! a pure transformer: one event in, one [`StepResult`] out. State is the only
6//! cross-run channel, scoped to the branch by id prefix.
7
8use std::sync::Arc;
9
10use async_trait::async_trait;
11use serde_json::Value;
12
13use crate::event::Event;
14use crate::recorder::{noop_recorder, RunRecorder};
15use crate::result::StepResult;
16use crate::state::State;
17
18/// The execution context handed to every step node.
19///
20/// Port of the subset of `WorkflowContext` a node legitimately touches:
21/// branch id (for state scoping), instance config + metadata (for template
22/// resolution), and the state handle. I/O tool calls (`ctx.do(...)`) are the
23/// next porting phase.
24pub struct WorkflowContext {
25    /// Workflow branch this record belongs to.
26    pub branch_id: String,
27    /// Configuration object validated against the declared schema.
28    pub config: Value,
29    /// Instance metadata available to templates.
30    pub instance_metadata: Value,
31    /// Branch-scoped durable state handle.
32    pub state: Arc<dyn State>,
33    /// Run recorder receiving step facts.
34    pub recorder: Arc<dyn RunRecorder>,
35    /// Kind of trigger that started the run.
36    pub trigger_kind: String,
37}
38
39impl WorkflowContext {
40    /// Context for `branch_id` over `state` with a no-op recorder.
41    pub fn new(branch_id: impl Into<String>, state: Arc<dyn State>) -> Self {
42        Self {
43            branch_id: branch_id.into(),
44            config: Value::Object(Default::default()),
45            instance_metadata: Value::Object(Default::default()),
46            state,
47            recorder: noop_recorder(),
48            trigger_kind: "event".into(),
49        }
50    }
51
52    /// Attach instance config.
53    pub fn with_config(mut self, config: Value) -> Self {
54        self.config = config;
55        self
56    }
57
58    /// Attach a recorder.
59    pub fn with_recorder(mut self, recorder: Arc<dyn RunRecorder>) -> Self {
60        self.recorder = recorder;
61        self
62    }
63
64    /// Set the trigger kind.
65    pub fn with_trigger_kind(mut self, trigger_kind: impl Into<String>) -> Self {
66        self.trigger_kind = trigger_kind.into();
67        self
68    }
69
70    /// Prefix a state key with the branch id (R6 scoping).
71    pub fn scoped(&self, key: &str) -> String {
72        format!("{}.{}", self.branch_id, key)
73    }
74}
75
76/// A pure transformer node. One event in, one result out.
77#[async_trait]
78pub trait StepNode: Send + Sync {
79    /// Whether `process` may return [`StepResult::FanOut`]. Default `false`.
80    fn produces_fan_out(&self) -> bool {
81        false
82    }
83
84    /// Transform one event.
85    async fn process(&self, event: &Event, ctx: &WorkflowContext) -> StepResult;
86}
af_workflow/node.rs

af_workflow/
node.rs
