af-workflow 0.5.0

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, &registry).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, &registry).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", &registry),
        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, &registry),
        Err(HostError::Safety { .. })
    ));

    let spec = Spec::from_json(MULTI_SPEC).unwrap();
    let plans = ingress_plans(&spec, &registry);
    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, &registry).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");
}