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,
};
const IDLE_WAIT_SECS: i64 = 86_400;
pub struct SpecDriver {
spec_id: String,
host: WorkflowHost,
schedule: Option<SchedulePolicy>,
actions: BTreeMap<String, CapabilityManifest>,
}
impl SpecDriver {
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,
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 * * * *"})),
®istry,
)
.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!({})), ®istry).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, ®istry).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!({})), ®istry),
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 * * * *"})),
®istry,
)
.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
));
}
}