af_workflow/

spec_driver.rs

//! Bridge from a validated product `Spec` to the durable supervisor.
//!
//! Product specs get durable execution without a host-side state machine.
//! Action steps emit data-only requests; this driver pins and persists them as
//! `ActionIntent`s before the product provider can run.

use std::collections::BTreeMap;
use std::sync::Arc;

use async_trait::async_trait;
use serde_json::{json, Map, Value};

use crate::{
    ingress_plans, schedule_decision, ActionIntent, ActionState, CapabilityKind,
    CapabilityManifest, CatchUpPolicy, EvaluationOutcome, Event, HostError, MemoryState,
    NodeRegistry, ScheduleCadence, SchedulePolicy, Spec, Terminal, WorkDisposition, WorkItem,
    WorkflowDriver, WorkflowHost, WorkflowTransitionCommand,
};

/// Seconds an event-driven instance sleeps between wakeups when no timer or
/// trigger delivery is pending. Deliveries pull `next_run_at` forward, so this
/// only bounds how long a stale row stays idle.
const IDLE_WAIT_SECS: i64 = 86_400;

/// Durable driver compiled from a product spec; see the module docs.
pub struct SpecDriver {
    spec_id: String,
    host: WorkflowHost,
    schedule: Option<SchedulePolicy>,
    actions: BTreeMap<String, CapabilityManifest>,
}

impl SpecDriver {
    /// Compile `spec`; rejects action nodes without an `Action` manifest and funds actions.
    pub fn new(spec: &Spec, registry: &NodeRegistry) -> Result<Self, HostError> {
        let host = WorkflowHost::from_spec(spec, registry)?;
        let schedule = ingress_plans(spec, registry)
            .into_iter()
            .find(|plan| plan.branch_id == host.branch_id())
            .map(|plan| {
                let required_u64 = |name: &str| {
                    plan.ingress_config[name]
                        .as_u64()
                        .ok_or_else(|| HostError::Schedule {
                            spec_id: spec.spec_id.clone(),
                            reason: format!("{} requires integer '{name}'", plan.ingress_type),
                        })
                };
                let cadence = match plan.ingress_type.as_str() {
                    "ingress.cron" => ScheduleCadence::Cron {
                        expression: plan.ingress_config["expression"]
                            .as_str()
                            .ok_or_else(|| HostError::Schedule {
                                spec_id: spec.spec_id.clone(),
                                reason: "ingress.cron requires a string 'expression'".into(),
                            })?
                            .to_owned(),
                    },
                    "ingress.fixed_rate" => ScheduleCadence::FixedRate {
                        milliseconds: required_u64("milliseconds")?,
                    },
                    "ingress.fixed_delay" => ScheduleCadence::FixedDelay {
                        milliseconds: required_u64("milliseconds")?,
                    },
                    _ => return Ok(None),
                };
                let catch_up = match plan.ingress_config["catch_up"].as_str().unwrap_or("once") {
                    "skip" => CatchUpPolicy::Skip,
                    "once" => CatchUpPolicy::CatchUpOnce,
                    "all" => CatchUpPolicy::CatchUpAll {
                        limit: plan.ingress_config["catch_up_limit"]
                            .as_u64()
                            .and_then(|value| u32::try_from(value).ok())
                            .ok_or_else(|| HostError::Schedule {
                                spec_id: spec.spec_id.clone(),
                                reason: "catch_up=all requires positive integer catch_up_limit"
                                    .into(),
                            })?,
                    },
                    value => {
                        return Err(HostError::Schedule {
                            spec_id: spec.spec_id.clone(),
                            reason: format!("unknown catch_up policy '{value}'"),
                        })
                    }
                };
                let policy = SchedulePolicy {
                    cadence,
                    timezone: plan.ingress_config["timezone"]
                        .as_str()
                        .unwrap_or("UTC")
                        .to_owned(),
                    catch_up,
                };
                policy.validate().map_err(|error| HostError::Schedule {
                    spec_id: spec.spec_id.clone(),
                    reason: error.to_string(),
                })?;
                Ok::<_, HostError>(Some(policy))
            })
            .transpose()?
            .flatten();
        let actions = registry
            .capability_manifests()
            .filter(|manifest| manifest.kind == CapabilityKind::Action)
            .map(|manifest| (manifest.id.clone(), manifest.clone()))
            .collect();
        Ok(Self {
            spec_id: spec.spec_id.clone(),
            host,
            schedule,
            actions,
        })
    }

    fn action_intents(
        &self,
        item: &WorkItem,
        requests: Vec<crate::PreparedAction>,
    ) -> Result<Vec<ActionIntent>, String> {
        requests
            .into_iter()
            .enumerate()
            .map(|(index, request)| {
                request.validate()?;
                let manifest = self.actions.get(&request.capability_id).ok_or_else(|| {
                    format!(
                        "action step emitted unknown capability '{}'",
                        request.capability_id
                    )
                })?;
                let capability = item
                    .capability_pins
                    .iter()
                    .find(|pin| {
                        pin.id == manifest.id
                            && pin.contract_version == manifest.contract_version
                            && pin.content_digest == manifest.content_digest
                    })
                    .cloned()
                    .ok_or_else(|| {
                        format!(
                            "workflow revision does not pin action capability '{}'",
                            manifest.id
                        )
                    })?;
                Ok(ActionIntent {
                    id: format!("{}:{}:action:{index}", item.id, item.state_version),
                    tenant_id: item.tenant_id.clone(),
                    instance_id: item
                        .id
                        .parse()
                        .map_err(|error: af_context::EmptyId| error.to_string())?,
                    run_id: item.run_id.clone(),
                    capability,
                    idempotency_key: format!(
                        "{}:{}:{}",
                        item.id, item.state_version, request.idempotency_key
                    ),
                    state: ActionState::Prepared,
                    input: request.input,
                    effect: manifest.effect,
                    retry_class: manifest.idempotency_mode,
                    control_epochs: item.control_epochs,
                    resource_scope_id: request.resource_scope_id,
                    lease_epoch: item.lease_version,
                    action_epoch: item.state_version,
                    // Provider timeout is per attempt. A product may set a
                    // separate whole-intent deadline explicitly.
                    deadline: request.deadline,
                    reservation: request.reservation,
                    created_at: item.claimed_at,
                })
            })
            .collect()
    }

    fn terminal_action(
        &self,
        item: &WorkItem,
        next_state: &mut Map<String, Value>,
    ) -> Result<Option<WorkflowTransitionCommand>, String> {
        let Some(wakeup) = item
            .wakeups
            .iter()
            .find(|wakeup| wakeup.payload["kind"] == "terminal_action")
        else {
            return Ok(None);
        };
        let observation: crate::ActionObservation =
            serde_json::from_value(wakeup.payload["observation"].clone())
                .map_err(|error| format!("terminal action fact: {error}"))?;
        let Some(internal) = next_state
            .get_mut("__workflow")
            .and_then(Value::as_object_mut)
        else {
            return Err("terminal action has no durable pending-action state".into());
        };
        let pending = internal
            .get_mut("pending_actions")
            .and_then(Value::as_array_mut)
            .ok_or_else(|| "terminal action has no pending action list".to_string())?;
        let before = pending.len();
        pending.retain(|id| id.as_str() != Some(&observation.action_intent_id));
        if pending.len() == before {
            return Err("terminal action does not belong to this workflow instance".into());
        }
        let disposition = if pending.is_empty() {
            internal
                .remove("resume_at")
                .map(serde_json::from_value)
                .transpose()
                .map_err(|error| format!("stored workflow resume_at: {error}"))?
                .map(|at| WorkDisposition::Reschedule { at })
                .unwrap_or_else(|| {
                    if internal
                        .get("static_event_consumed")
                        .and_then(Value::as_bool)
                        .unwrap_or(false)
                    {
                        WorkDisposition::Complete
                    } else {
                        Self::idle_disposition()
                    }
                })
        } else {
            WorkDisposition::Continue {
                delay_secs: IDLE_WAIT_SECS,
            }
        };
        let succeeded = matches!(
            observation.state.as_str(),
            "completed" | "max_steps_reached" | "succeeded"
        );
        Ok(Some(command(
            item,
            Value::Object(next_state.clone()),
            Vec::new(),
            EvaluationOutcome {
                triggered: true,
                matched: true,
                succeeded,
                action_terminal: true,
            },
            disposition,
            "workflow.action_observed",
        )))
    }

    fn catch_up_count(state: &Map<String, Value>) -> u32 {
        state
            .get("__workflow")
            .and_then(Value::as_object)
            .and_then(|internal| internal.get("catch_up_count"))
            .and_then(Value::as_u64)
            .and_then(|value| u32::try_from(value).ok())
            .unwrap_or(0)
    }

    fn set_internal(state: &mut Map<String, Value>, key: &str, value: Value) -> Result<(), String> {
        let internal = state
            .entry("__workflow")
            .or_insert_with(|| json!({}))
            .as_object_mut()
            .ok_or_else(|| "workflow internal state must be an object".to_string())?;
        internal.insert(key.into(), value);
        Ok(())
    }

    fn schedule_decision(
        &self,
        item: &WorkItem,
        state: &Map<String, Value>,
    ) -> Result<Option<crate::ScheduleDecision>, String> {
        if self.schedule.is_some() && !item.wakeups.is_empty() {
            return Ok(Some(crate::ScheduleDecision {
                tick_at: None,
                next_at: item.scheduled_at,
                catch_up_count: Self::catch_up_count(state),
            }));
        }
        self.schedule
            .as_ref()
            .map(|schedule| {
                schedule_decision(
                    schedule,
                    item.scheduled_at,
                    item.claimed_at,
                    Self::catch_up_count(state),
                )
            })
            .transpose()
    }

    fn idle_disposition() -> WorkDisposition {
        WorkDisposition::Continue {
            delay_secs: IDLE_WAIT_SECS,
        }
    }
}

#[async_trait]
impl<Context: Send + Sync> WorkflowDriver<Context> for SpecDriver {
    fn name(&self) -> &'static str {
        "spec"
    }

    fn spec_ids(&self) -> Vec<&str> {
        vec![&self.spec_id]
    }

    fn validate_specs(&self) -> Result<(), String> {
        Ok(())
    }

    async fn evaluate(
        &self,
        _: &Context,
        item: &WorkItem,
    ) -> Result<WorkflowTransitionCommand, String> {
        let mut next_state = match &item.config {
            Value::Object(map) => map.clone(),
            Value::Null => Map::new(),
            _ => return Err("spec instance config must be a JSON object".into()),
        };
        if item.cancel_requested {
            return Ok(command(
                item,
                Value::Object(next_state),
                Vec::new(),
                EvaluationOutcome::default(),
                WorkDisposition::Complete,
                "workflow.cancelled",
            ));
        }
        if let Some(command) = self.terminal_action(item, &mut next_state)? {
            return Ok(command);
        }
        if next_state
            .get("__workflow")
            .and_then(Value::as_object)
            .and_then(|internal| internal.get("pending_actions"))
            .and_then(Value::as_array)
            .is_some_and(|pending| !pending.is_empty())
        {
            return Ok(command(
                item,
                Value::Object(next_state),
                Vec::new(),
                EvaluationOutcome::default(),
                Self::idle_disposition(),
                "workflow.action_waiting",
            ));
        }
        let schedule = self.schedule_decision(item, &next_state)?;
        if let Some(decision) = schedule {
            Self::set_internal(
                &mut next_state,
                "catch_up_count",
                json!(decision.catch_up_count),
            )?;
            if decision.tick_at.is_none() && item.wakeups.is_empty() {
                return Ok(command(
                    item,
                    Value::Object(next_state),
                    Vec::new(),
                    EvaluationOutcome::default(),
                    WorkDisposition::Reschedule {
                        at: decision.next_at,
                    },
                    "workflow.schedule_skipped",
                ));
            }
        }
        let state = Arc::new(MemoryState::from_snapshot(
            next_state
                .get("state")
                .and_then(Value::as_object)
                .cloned()
                .unwrap_or_default(),
        ));
        let wakeup_payload = item.wakeups.first().map(|wakeup| wakeup.payload.clone());
        let static_event = next_state.get("event").cloned().filter(|_| {
            !next_state
                .get("__workflow")
                .and_then(Value::as_object)
                .and_then(|internal| internal.get("static_event_consumed"))
                .and_then(Value::as_bool)
                .unwrap_or(false)
        });
        let consumed_static_event = wakeup_payload.is_none() && static_event.is_some();
        let payload = wakeup_payload
            .or(static_event)
            .or_else(|| {
                schedule
                    .and_then(|decision| decision.tick_at)
                    .map(|tick_at| json!({ "tick_at": tick_at }))
            })
            .ok_or_else(|| "event workflow requires a trigger delivery".to_string())?;
        if consumed_static_event {
            Self::set_internal(&mut next_state, "static_event_consumed", json!(true))?;
        }
        let context = self.host.context(state.clone());
        let outcome = self
            .host
            .run_event(&context, Event::from_json(payload))
            .await
            .ok_or_else(|| "spec has no root branch".to_string())?;
        let (terminal, exit_reason) = match &outcome.terminal {
            Terminal::Completed => ("completed", Value::Null),
            Terminal::Dropped { node_id, reason } => {
                ("dropped", json!({ "node_id": node_id, "reason": reason }))
            }
        };
        next_state.insert("state".into(), Value::Object(state.snapshot()));
        next_state.insert(
            "last_run".into(),
            json!({
                "terminal": terminal,
                "exit": exit_reason,
                "steps_run": outcome.steps_run,
                "survivors": outcome.survivors.len(),
                "at": item.claimed_at,
            }),
        );
        let action_intents = self.action_intents(item, outcome.actions)?;
        let disposition = match schedule {
            Some(decision) if action_intents.is_empty() => WorkDisposition::Reschedule {
                at: decision.next_at,
            },
            Some(decision) => {
                Self::set_internal(&mut next_state, "resume_at", json!(decision.next_at))?;
                Self::idle_disposition()
            }
            None if action_intents.is_empty() && consumed_static_event => WorkDisposition::Complete,
            None => Self::idle_disposition(),
        };
        if !action_intents.is_empty() {
            let internal = next_state
                .entry("__workflow")
                .or_insert_with(|| json!({}))
                .as_object_mut()
                .ok_or_else(|| "workflow internal state must be an object".to_string())?;
            let pending = internal
                .entry("pending_actions")
                .or_insert_with(|| json!([]))
                .as_array_mut()
                .ok_or_else(|| "workflow pending actions must be an array".to_string())?;
            pending.extend(action_intents.iter().map(|intent| json!(intent.id)));
        }
        let evaluation = EvaluationOutcome {
            triggered: true,
            matched: outcome.matched,
            succeeded: outcome.succeeded && action_intents.is_empty(),
            action_terminal: false,
        };
        Ok(command(
            item,
            Value::Object(next_state),
            action_intents,
            evaluation,
            disposition,
            "workflow.spec_evaluated",
        ))
    }
}

fn command(
    item: &WorkItem,
    next_state: Value,
    action_intents: Vec<ActionIntent>,
    outcome: EvaluationOutcome,
    disposition: WorkDisposition,
    event_type: &str,
) -> WorkflowTransitionCommand {
    WorkflowTransitionCommand {
        delivery_key: format!("spec:{}:{}", item.id, item.state_version),
        delivery_digest: format!(
            "{}:{}:{}",
            item.workflow_revision_digest, item.execution_profile_digest, item.state_version
        ),
        event_type: event_type.into(),
        event_digest: format!("{event_type}:{}:{}", item.id, item.state_version),
        event_payload: json!({ "spec_id": item.spec_id, "wakeups": item.wakeups.len() }),
        next_state,
        action_intents,
        outcome,
        disposition,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{ControlEpochs, DriverRegistry, PreparedAction, StepNode, StepResult, Wakeup};

    struct PassNode;

    #[async_trait::async_trait]
    impl StepNode for PassNode {
        async fn process(&self, event: &Event, _: &crate::WorkflowContext) -> StepResult {
            StepResult::Pass(event.clone())
        }
    }

    struct ActionNode;

    #[async_trait::async_trait]
    impl StepNode for ActionNode {
        async fn process(&self, event: &Event, _: &crate::WorkflowContext) -> StepResult {
            StepResult::Action {
                event: event.clone(),
                action: Box::new(PreparedAction::new(
                    "execute.demo",
                    "request-1",
                    json!({"value": 1}),
                )),
            }
        }
    }

    fn item(spec_id: &str, config: Value) -> WorkItem {
        WorkItem {
            id: "instance".into(),
            run_id: uuid::Uuid::new_v4().to_string().parse().unwrap(),
            tenant_id: "tenant".parse().unwrap(),
            subject_id: "subject".parse().unwrap(),
            spec_id: spec_id.into(),
            definition_id: spec_id.into(),
            workflow_revision: 1,
            workflow_revision_digest: "digest".into(),
            execution_profile_id: "profile".into(),
            execution_profile_revision: 1,
            execution_profile_digest: "profile-digest".into(),
            kernel_abi_version: "1".into(),
            capability_pins: Vec::new(),
            lifecycle: crate::LifecyclePolicy::run_once(),
            scheduled_at: chrono::Utc::now(),
            claimed_at: chrono::Utc::now(),
            config,
            state_version: 0,
            control_epochs: ControlEpochs::default(),
            cancel_requested: false,
            lease_version: 1,
            wakeups: Vec::new(),
        }
    }

    fn spec(ingress: &str, ingress_config: Value) -> Spec {
        Spec::from_json(
            &json!({
                "spec_id": "counter", "version": "1",
                "branches": [{
                    "branch_id": "__root__",
                    "nodes": [
                        {"id": "in", "type": ingress, "config": ingress_config},
                        {"id": "count", "type": "transform.state_append",
                         "config": {"key": "seen", "path": "value", "max_len": 3}}
                    ],
                    "edges": [{"source": "in", "target": "count"}]
                }]
            })
            .to_string(),
        )
        .unwrap()
    }

    #[tokio::test]
    async fn cron_spec_reschedules_and_persists_branch_state() {
        let registry = NodeRegistry::with_builtins();
        let driver = SpecDriver::new(
            &spec("ingress.cron", json!({"expression": "0 0 * * * *"})),
            &registry,
        )
        .unwrap();
        let first = WorkflowDriver::<()>::evaluate(
            &driver,
            &(),
            &item("counter", json!({"event": {"value": 1}})),
        )
        .await
        .unwrap();
        let WorkDisposition::Reschedule { at } = first.disposition else {
            panic!("cron spec must reschedule");
        };
        assert_eq!(first.next_state["last_run"]["terminal"], "completed");
        let mut next = item("counter", first.next_state);
        next.scheduled_at = at;
        next.claimed_at = at;
        let second = WorkflowDriver::<()>::evaluate(&driver, &(), &next)
            .await
            .unwrap();
        assert_eq!(
            second.next_state["state"]["__root__.seen"],
            json!([1]),
            "the consumed bootstrap event must not be replayed on a cron tick"
        );
        assert_eq!(second.next_state["last_run"]["at"], json!(at));
    }

    #[tokio::test]
    async fn event_spec_consumes_a_wakeup_then_waits() {
        let registry = NodeRegistry::with_builtins();
        let driver = SpecDriver::new(&spec("ingress.event", json!({})), &registry).unwrap();
        assert_eq!(
            WorkflowDriver::<()>::evaluate(&driver, &(), &item("counter", json!({})))
                .await
                .unwrap_err(),
            "event workflow requires a trigger delivery"
        );
        let mut work = item("counter", json!({}));
        work.wakeups.push(Wakeup {
            id: "delivery".into(),
            kind: "delivery".into(),
            payload: json!({"value": 7}),
        });
        let command = WorkflowDriver::<()>::evaluate(&driver, &(), &work)
            .await
            .unwrap();
        assert_eq!(command.next_state["state"]["__root__.seen"], json!([7]));
        assert!(matches!(
            command.disposition,
            WorkDisposition::Continue { .. }
        ));
        let mut registry = DriverRegistry::<()>::new();
        registry.register(Arc::new(driver)).unwrap();
        assert_eq!(registry.spec_ids(), ["counter"]);
    }

    #[tokio::test]
    async fn effectful_steps_are_lifted_into_pinned_intents() {
        let mut registry = NodeRegistry::with_builtins();
        registry.register_step("guard.auth", |_| Ok(Box::new(PassNode)));
        registry.register_side_effect_guard("guard.auth");
        registry.register_step("execute.demo", |_| Ok(Box::new(ActionNode)));
        let manifest = CapabilityManifest::action(
            "execute.demo",
            "1",
            "demo-digest",
            crate::Effect::ExternalWrite,
            crate::IdempotencyMode::Native,
            true,
        );
        registry.register_capability(manifest.clone()).unwrap();
        let effect = Spec::from_json(
            &json!({
                "spec_id": "effect", "version": "1",
                "branches": [{
                    "branch_id": "__root__",
                        "nodes": [
                        {"id": "in", "type": "ingress.event", "config": {}},
                        {"id": "auth", "type": "guard.auth", "config": {}},
                        {"id": "do", "type": "execute.demo", "config": {}}
                    ],
                    "edges": [
                        {"source": "in", "target": "auth"},
                        {"source": "auth", "target": "do"}
                    ]
                }]
            })
            .to_string(),
        )
        .unwrap();
        let driver = SpecDriver::new(&effect, &registry).unwrap();
        let mut work = item("effect", json!({"event": {"value": 1}}));
        work.capability_pins.push(crate::CapabilityPin {
            id: manifest.id,
            contract_version: manifest.contract_version,
            content_digest: manifest.content_digest,
        });
        let command = WorkflowDriver::<()>::evaluate(&driver, &(), &work)
            .await
            .unwrap();
        assert_eq!(command.action_intents.len(), 1);
        assert_eq!(command.action_intents[0].state, ActionState::Prepared);
        assert_eq!(command.action_intents[0].capability.id, "execute.demo");
        assert_eq!(command.action_intents[0].deadline, None);

        let mut waiting = item("effect", command.next_state.clone());
        waiting.capability_pins = work.capability_pins.clone();
        let waiting_command = WorkflowDriver::<()>::evaluate(&driver, &(), &waiting)
            .await
            .unwrap();
        assert!(waiting_command.action_intents.is_empty());
        assert_eq!(waiting_command.event_type, "workflow.action_waiting");

        waiting.wakeups.push(Wakeup {
            id: "terminal".into(),
            kind: "timer".into(),
            payload: json!({
                "kind": "terminal_action",
                "observation": {
                    "id": "observation",
                    "action_intent_id": command.action_intents[0].id,
                    "provider_version": "1",
                    "observed_at": chrono::Utc::now(),
                    "state": "rejected",
                    "resource_ref": null,
                    "raw_receipt_digest": "receipt",
                    "terminal": true,
                    "retry_authorized": false
                }
            }),
        });
        let terminal = WorkflowDriver::<()>::evaluate(&driver, &(), &waiting)
            .await
            .unwrap();
        assert!(terminal.action_intents.is_empty());
        assert!(matches!(terminal.disposition, WorkDisposition::Complete));
        assert!(matches!(
            SpecDriver::new(&spec("ingress.cron", json!({})), &registry),
            Err(HostError::Schedule { .. })
        ));
    }

    #[tokio::test]
    async fn delivery_does_not_advance_the_cron_cursor() {
        let registry = NodeRegistry::with_builtins();
        let driver = SpecDriver::new(
            &spec("ingress.cron", json!({"expression": "0 0 * * * *"})),
            &registry,
        )
        .unwrap();
        let mut work = item("counter", json!({}));
        work.scheduled_at = work.claimed_at + chrono::Duration::hours(1);
        work.wakeups.push(Wakeup {
            id: "delivery".into(),
            kind: "delivery".into(),
            payload: json!({"value": 7}),
        });
        let command = WorkflowDriver::<()>::evaluate(&driver, &(), &work)
            .await
            .unwrap();
        assert_eq!(command.next_state["state"]["__root__.seen"], json!([7]));
        assert!(matches!(
            command.disposition,
            WorkDisposition::Reschedule { at } if at == work.scheduled_at
        ));
    }
}