use af_workflow::{
ingress_plans, next_cron_at, Event, HostError, MemoryState, NodeError, NodeRegistry, RunHandle,
RunRecorder, RunStatus, Spec, State, StepNode, StepResult, StepStatus, WorkflowContext,
WorkflowHost,
};
use async_trait::async_trait;
use serde_json::{json, Value};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
#[tokio::test]
async fn consensus_supports_quorum_weighting_and_ties() {
async fn vote(config: &str, signals: Value) -> Option<Value> {
let spec = Spec::from_json(&format!(
r#"{{"spec_id":"vote","version":"1","branches":[{{"branch_id":"__root__","nodes":[
{{"id":"in","type":"ingress.event","config":{{}}}},
{{"id":"vote","type":"decision.consensus_voting","config":{config}}},
{{"id":"save","type":"transform.state_set","config":{{"key":"out","value":"$event.consensus"}}}}
],"edges":[{{"source":"in","target":"vote"}},{{"source":"vote","target":"save"}}]}}]}}"#
))
.unwrap();
let registry = NodeRegistry::with_builtins();
let host = WorkflowHost::from_spec(&spec, ®istry).unwrap();
let state = Arc::new(MemoryState::new());
let context = host.context(state.clone());
host.run_event(&context, Event::from_json(json!({"signals": signals})))
.await;
state.get("__root__.out").await
}
assert_eq!(
vote(
r#"{"signals_path":"signals","quorum":2}"#,
json!([{"choice":"a"},{"choice":"a"},{"choice":"b"}]),
)
.await
.unwrap()["choice"],
"a"
);
assert_eq!(
vote(
r#"{"signals_path":"signals","mode":"weighted","threshold":0.5}"#,
json!([{"choice":"a","weight":1.0},{"choice":"b","weight":2.0}]),
)
.await
.unwrap()["choice"],
"b"
);
assert!(vote(
r#"{"signals_path":"signals"}"#,
json!([{"choice":"a"},{"choice":"b"}]),
)
.await
.is_none());
}
const MULTI_SPEC: &str = r#"{"spec_id":"multi","version":"1","branches":[
{"branch_id":"publish","nodes":[
{"id":"p-in","type":"ingress.event","config":{}},
{"id":"publish","type":"transform.state_publish_cross_branch","config":{"path":"value","key":"shared"}}
],"edges":[{"source":"p-in","target":"publish"}]},
{"branch_id":"read","nodes":[
{"id":"r-in","type":"ingress.event","config":{}},
{"id":"read","type":"transform.state_read_cross_branch","config":{"key":"shared","into":"value"}},
{"id":"save","type":"transform.state_set","config":{"key":"out","value":"$event.value"}}
],"edges":[{"source":"r-in","target":"read"},{"source":"read","target":"save"}]}
]}"#;
#[tokio::test]
async fn host_validates_and_drives_multiple_branches() {
let registry = NodeRegistry::with_builtins();
let host = WorkflowHost::assemble(MULTI_SPEC, ®istry).unwrap();
assert_eq!(host.spec_id(), "multi");
assert_eq!(host.branch_count(), 2);
assert_eq!(host.branch_ids().collect::<Vec<_>>(), ["publish", "read"]);
let state = Arc::new(MemoryState::new());
host.run_event_on(
"publish",
&host.context_for("publish", state.clone()),
Event::from_json(json!({"value": 7})),
)
.await
.unwrap();
host.run_event_on(
"read",
&host.context_for("read", state.clone()),
Event::from_json(json!({})),
)
.await;
assert_eq!(state.get("read.out").await, Some(json!(7)));
assert!(host
.run_event_on(
"missing",
&host.context_for("missing", state),
Event::from_json(json!({})),
)
.await
.is_none());
}
#[test]
fn host_and_ingress_fail_closed() {
let registry = NodeRegistry::with_builtins();
assert!(matches!(
WorkflowHost::assemble("not json", ®istry),
Err(HostError::Parse(_))
));
let unsafe_spec = r#"{"spec_id":"unsafe","version":"1","branches":[{"branch_id":"b","nodes":[{"id":"in","type":"ingress.event","config":{}},{"id":"x","type":"execute.x","config":{}}],"edges":[{"source":"in","target":"x"}]}]}"#;
assert!(matches!(
WorkflowHost::assemble(unsafe_spec, ®istry),
Err(HostError::Safety { .. })
));
let spec = Spec::from_json(MULTI_SPEC).unwrap();
let plans = ingress_plans(&spec, ®istry);
assert_eq!(plans.len(), 2);
assert!(next_cron_at("bad cron", chrono::Utc::now()).is_err());
assert!(!registry.is_step("transform.tool_invoke"));
assert!(!registry.is_step("transform.ask_llm"));
}
#[derive(Default)]
struct RecorderCounts {
steps: AtomicUsize,
filtered: AtomicUsize,
statuses: Mutex<Vec<RunStatus>>,
}
struct CountingRecorder(Arc<RecorderCounts>);
struct CountingHandle(Arc<RecorderCounts>);
#[async_trait]
impl RunRecorder for CountingRecorder {
async fn start(&self, _trigger_kind: &str, _trigger: Value) -> Box<dyn RunHandle> {
Box::new(CountingHandle(self.0.clone()))
}
}
#[async_trait]
impl RunHandle for CountingHandle {
async fn record_step(
&mut self,
_node_id: &str,
_node_type: &str,
_status: StepStatus,
_exit_reason: Option<&str>,
_detail: Value,
) {
self.0.steps.fetch_add(1, Ordering::SeqCst);
}
async fn mark_filtered(&mut self, _node_id: &str, _node_type: &str, _reason: &str) {
self.0.filtered.fetch_add(1, Ordering::SeqCst);
}
async fn end(self: Box<Self>, status: RunStatus, _exit_reason: Option<&str>) {
self.0.statuses.lock().unwrap().push(status);
}
}
struct Pass;
struct Gate;
fn pass_factory(_: &Value) -> Result<Box<dyn StepNode>, NodeError> {
Ok(Box::new(Pass))
}
fn gate_factory(_: &Value) -> Result<Box<dyn StepNode>, NodeError> {
Ok(Box::new(Gate))
}
#[async_trait]
impl StepNode for Pass {
async fn process(&self, event: &Event, _context: &WorkflowContext) -> StepResult {
StepResult::Pass(event.clone())
}
}
#[async_trait]
impl StepNode for Gate {
async fn process(&self, _event: &Event, _context: &WorkflowContext) -> StepResult {
StepResult::gate("blocked")
}
}
async fn run_recorded(node_type: &str, factory: af_workflow::StepFactory) -> Arc<RecorderCounts> {
let mut registry = NodeRegistry::with_builtins();
registry.register_step(node_type, factory);
let spec = Spec::from_json(&format!(
r#"{{"spec_id":"record","version":"1","branches":[{{"branch_id":"b","nodes":[{{"id":"in","type":"ingress.event","config":{{}}}},{{"id":"node","type":"{node_type}","config":{{}}}}],"edges":[{{"source":"in","target":"node"}}]}}]}}"#
))
.unwrap();
let host = WorkflowHost::from_spec(&spec, ®istry).unwrap();
let counts = Arc::new(RecorderCounts::default());
let context = WorkflowContext::new("b", Arc::new(MemoryState::new()))
.with_recorder(Arc::new(CountingRecorder(counts.clone())));
host.run_event(&context, Event::from_json(json!({"x": 1})))
.await;
counts
}
#[tokio::test]
async fn recorder_distinguishes_actions_gates_and_filters() {
let action = run_recorded("notify.test", pass_factory).await;
assert_eq!(action.steps.load(Ordering::SeqCst), 1);
assert_eq!(action.statuses.lock().unwrap().as_slice(), [RunStatus::Ok]);
let gate = run_recorded("filter.gate", gate_factory).await;
assert_eq!(gate.steps.load(Ordering::SeqCst), 1);
assert_eq!(
gate.statuses.lock().unwrap().as_slice(),
[RunStatus::Skipped]
);
assert_eq!(StepStatus::Error.as_str(), "error");
assert_eq!(RunStatus::Ok.as_str(), "ok");
}