zebflow 0.11.14

//! `logic.retry` — bounded retry control, fed two ways.
//!
//! **The error road.** The engine routes a node failure into `:error` edges
//! when they exist: the payload the failing node received, kept, plus that
//! node's key answering `{ ok: false, error: { code, message } }`
//! (`node-conventions.md` §6), with the retry state under `__zf_retry`
//! (`error_key` names the key). `logic.retry` either emits `retry` with the
//! failing node's input given back plus updated retry metadata, or `failed`
//! with the failure when the attempt budget is exhausted. A `refused`-class
//! error goes straight to `failed`: running it again cannot help.
//!
//! **The verdict road.** A payload arriving on `in` from an ordinary edge is a
//! verdict, not a failure: `retry: true` on it (or `--when "<expr>"` true —
//! the same JavaScript over `input` that `logic.if --when` reads) fires
//! `retry` with that payload, as many times as `--max-attempts` allows, then
//! `failed`; a false verdict passes the payload through on `done`. This is
//! what lets a poll loop wait without throwing — a wait is not an error, and
//! the canvas should never paint it red.
//!
//! Attempts are counted from `__zf_retry.attempt`: on the payload when the
//! author (or the engine's failure delivery) carried it, and from this node's own last
//! output through `$nodes.<self>` — the engine keeps a retry node in its own
//! scope for exactly this — so a loop whose payload is replaced on the way
//! round (`http.response.fetch` answers with a fresh body) still counts 1, 2, 3.
//!
//! **Two budgets, whichever runs out first.** `--max-attempts` is the count;
//! `--max-elapsed` is wall time from the first attempt this node saw
//! (`__zf_retry.first_at`, ms epoch). The wait between attempts is
//! `--delay`, grown by `--backoff` each attempt (a factor: 1 is a fixed
//! delay, 2 doubles it) and capped by `--max-delay`; the wait just taken
//! is stamped as `__zf_retry.next_delay_ms`. The three times are durations
//! (`500ms`, `30s`, `2m`), read by the shared units parser. A wait that would end past the
//! elapsed budget is not taken: `failed` fires at once, and
//! `__zf_retry.reason` / `__zf_retry.message` say which budget ran out.

use async_trait::async_trait;
use serde::{Deserialize, Deserializer, Serialize};
use serde_json::{Value, json};

use crate::language::{
    COMPILE_TARGET_BACKEND, CompileOptions, CompiledProgram, LanguageEngine, ModuleSource,
    SourceKind,
};
use crate::pipeline::expr::build_expression_scope_input;
use crate::pipeline::model::{DslFlag, DslFlagKind, LayoutItem};
use crate::pipeline::{
    NodeDefinition, PipelineError,
    nodes::{NodeExecutionInput, NodeExecutionOutput, NodeHandler},
};

pub const NODE_KIND: &str = "logic.retry";
pub const INPUT_PIN_IN: &str = "in";
pub const OUTPUT_PIN_RETRY: &str = "retry";
pub const OUTPUT_PIN_FAILED: &str = "failed";
pub const OUTPUT_PIN_DONE: &str = "done";
const RETRY_STATE_KEY: &str = "__zf_retry";
/// The verdict key read when `--when` is not given.
const VERDICT_KEY: &str = "retry";

pub fn definition() -> NodeDefinition {
    NodeDefinition {
        kind: NODE_KIND.to_string(),
        title: "Retry".to_string(),
        description: "Runs a failed or not-yet-ready node again, with a delay and a budget. It is fed either way. **From an `:error` pin** (`[b]:error -> [r]`): it receives the failure — \
             the payload kept, the failing node's key answering `{ ok: false, error: { code, message } }` — and fires `retry` with \
             the failing node's input given back (wire `[r]:retry -> [b]`) until `--max-attempts` is spent, then `failed` with the \
             last failure (`input.<noun>.error.code`); a `refused`-class error goes straight to `failed`. **From an ordinary edge** \
             (`[check] -> [r]`): the payload is a verdict — `retry: true` on it, or `--when \"<expr>\"` true (JavaScript over \
             `input`, as `logic.if --when`), fires `retry` with that payload; false passes it through on `done`; the budget \
             spent fires `failed`. A poll loop waits this way without throwing, so the canvas shows a wait, never a failure. \
             `--delay` waits between attempts (a duration: `500ms`, `5s`); `--backoff 2` doubles that wait each attempt, `--max-delay` \
             caps the grown wait, and `--max-elapsed` is a wall-time budget from the first attempt — whichever budget runs out \
             first fires `failed`, and `__zf_retry.reason` names it. Attempts are counted in `__zf_retry.attempt` \
             (`$nodes.<r>.__zf_retry`); a retry loop without a `failed` edge swallows the give-up."
            .to_string(),
        input_schema: serde_json::json!({ "type": "object" }),
        output_schema: serde_json::json!({ "type": "object" }),
        input_pins: vec![INPUT_PIN_IN.to_string()],
        output_pins: vec![
            OUTPUT_PIN_RETRY.to_string(),
            OUTPUT_PIN_FAILED.to_string(),
            OUTPUT_PIN_DONE.to_string(),
        ],
        script_available: false,
        script_bridge: None,
        config_schema: Default::default(),
        dsl_flags: vec![
            DslFlag {
                flag: "--max-attempts".to_string(),
                config_key: "max_attempts".to_string(),
                description: "Maximum total attempts before routing to failed.".to_string(),
                kind: DslFlagKind::Scalar,
                required: true,
                value: "number".to_string(),
                ..Default::default()
            },
            DslFlag {
                flag: "--delay".to_string(),
                config_key: "delay".to_string(),
                description: "Optional wait before retrying (`500ms`, `5s`); the first wait when `--backoff` grows it.".to_string(),
                kind: DslFlagKind::Scalar,
                required: false,
                value: "duration".to_string(),
                ..Default::default()
            },
            DslFlag {
                flag: "--backoff".to_string(),
                config_key: "backoff".to_string(),
                description: "Factor the delay grows by each attempt: 1 (default) is a fixed delay, 2 doubles it (500ms, 1s, 2s, …)."
                    .to_string(),
                kind: DslFlagKind::Scalar,
                required: false,
                value: "number".to_string(),
                ..Default::default()
            },
            DslFlag {
                flag: "--max-delay".to_string(),
                config_key: "max_delay".to_string(),
                description: "Cap for a delay grown by `--backoff` (`8s`).".to_string(),
                kind: DslFlagKind::Scalar,
                required: false,
                value: "duration".to_string(),
                ..Default::default()
            },
            DslFlag {
                flag: "--max-elapsed".to_string(),
                config_key: "max_elapsed".to_string(),
                description: "Wall-time budget from the first attempt (`30s`, `5m`); when spent (or the next wait would overrun it) `failed` fires even with attempts left."
                    .to_string(),
                kind: DslFlagKind::Scalar,
                required: false,
                value: "duration".to_string(),
                ..Default::default()
            },
            DslFlag {
                flag: "--when".to_string(),
                config_key: "when".to_string(),
                description: "Verdict expression for a payload from an ordinary edge (JavaScript over `input`, like `logic.if --when`); true retries, false is `done`. Without it the payload's `retry: true` is the verdict."
                    .to_string(),
                kind: DslFlagKind::Scalar,
                required: false,
                value: "expression".to_string(),
                ..Default::default()
            },
        ],
        fields: {
            use crate::pipeline::model::{NodeFieldDef, NodeFieldType};
            vec![
                NodeFieldDef {
                    name: "max_attempts".to_string(),
                    label: "Max Attempts".to_string(),
                    field_type: NodeFieldType::Text,
                    help: Some(
                        "Maximum total attempts including the original failed try.".to_string(),
                    ),
                    ..Default::default()
                },
                NodeFieldDef {
                    name: "delay".to_string(),
                    label: "Delay".to_string(),
                    field_type: NodeFieldType::Text,
                    placeholder: Some("500ms".to_string()),
                    help: Some("Optional wait before emitting the retry path, e.g. 500ms or 5s; the first wait when a backoff grows it.".to_string()),
                    ..Default::default()
                },
                NodeFieldDef {
                    name: "backoff".to_string(),
                    label: "Backoff factor".to_string(),
                    field_type: NodeFieldType::Text,
                    help: Some("The delay grows by this factor each attempt. 1 (default) is fixed; 2 doubles it.".to_string()),
                    ..Default::default()
                },
                NodeFieldDef {
                    name: "max_delay".to_string(),
                    label: "Max delay".to_string(),
                    field_type: NodeFieldType::Text,
                    placeholder: Some("8s".to_string()),
                    help: Some("Cap for a delay grown by the backoff, e.g. 8s.".to_string()),
                    ..Default::default()
                },
                NodeFieldDef {
                    name: "max_elapsed".to_string(),
                    label: "Max elapsed".to_string(),
                    field_type: NodeFieldType::Text,
                    placeholder: Some("30s".to_string()),
                    help: Some("Wall-time budget from the first attempt, e.g. 30s or 5m; when spent, failed fires even with attempts left.".to_string()),
                    ..Default::default()
                },
                NodeFieldDef {
                    name: "when".to_string(),
                    label: "Retry when".to_string(),
                    field_type: NodeFieldType::Textarea,
                    rows: Some(3),
                    help: Some(
                        "Verdict for a payload from an ordinary edge: JS over `input`, true retries, false is done. Empty: the payload's `retry: true`."
                            .to_string(),
                    ),
                    ..Default::default()
                },
            ]
        },
        layout: vec![
            LayoutItem::Row {
                row: vec![
                    LayoutItem::Field("max_attempts".to_string()),
                    LayoutItem::Field("delay".to_string()),
                ],
            },
            LayoutItem::Row {
                row: vec![
                    LayoutItem::Field("backoff".to_string()),
                    LayoutItem::Field("max_delay".to_string()),
                    LayoutItem::Field("max_elapsed".to_string()),
                ],
            },
            LayoutItem::Field("when".to_string()),
        ],
        ai_tool: Default::default(),
        examples: vec![
            crate::pipeline::model::NodeExample::dsl("Three attempts at a flaky API", "logic.retry --max-attempts 3 --delay 500ms")
                .note("Graph: `[b] http.response.fetch …`, `[b]:error -> [r]`, `[r]:retry -> [b]`, `[r]:failed -> [e] web.response.send --status 502`."),
            crate::pipeline::model::NodeExample::dsl("A rate-limited API", "logic.retry --max-attempts 6 --delay 500ms --backoff 2 --max-delay 8s")
                .note("The waits grow 500ms, 1s, 2s, 4s, 8s — the cap holds the last one. Add `--max-elapsed 30s` to give up on wall time instead of on the count; `__zf_retry.reason` on the `failed` payload says which budget ran out."),
            crate::pipeline::model::NodeExample::dsl("Poll until a job is ready", r#"logic.retry --max-attempts 40 --delay 5s --when "input.script.retry""#)
                .input(serde_json::json!({ "script": { "retry": true, "url": null } }))
                .note("Graph: `[poll] http.response.fetch …`, `[check] javascript.script.run -- return { retry: input.response.body.status !== 'success', url: input.response.body.videoURL }`, `[poll] -> [check]`, `[check] -> [wait]`, `[wait]:retry -> [poll]`, `[wait]:done -> [download]`, `[wait]:failed -> [gaveup]`. The check never throws; `--when` reads its answer: true is the wait, false goes on through `done`."),
        ],
        ..Default::default()
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Config {
    #[serde(deserialize_with = "count_or_string")]
    pub max_attempts: usize,
    /// `--delay`: a duration (`500ms`, `5s`); unset is no wait.
    #[serde(default)]
    pub delay: Value,
    /// Factor the delay grows by each attempt; 1 (or absent) is a fixed delay.
    #[serde(default, deserialize_with = "factor_or_string")]
    pub backoff: Option<f64>,
    /// `--max-delay`: cap for a delay grown by `backoff`.
    #[serde(default)]
    pub max_delay: Value,
    /// `--max-elapsed`: wall-time budget from the first attempt.
    #[serde(default)]
    pub max_elapsed: Value,
    /// The verdict expression for the verdict road; `retry: true` when empty.
    #[serde(default, deserialize_with = "super::expression_text")]
    pub when: String,
}

const CONFIG_CODE: &str = "FW_NODE_LOGIC_RETRY_CONFIG";

fn factor_or_string<'de, D: Deserializer<'de>>(de: D) -> Result<Option<f64>, D::Error> {
    let raw = Value::deserialize(de)?;
    Ok(match raw {
        Value::Number(n) => n.as_f64(),
        Value::String(s) => {
            let s = s.trim();
            if s.is_empty() { None } else { s.parse::<f64>().ok() }
        }
        _ => None,
    })
}

fn count_or_string<'de, D: Deserializer<'de>>(de: D) -> Result<usize, D::Error> {
    let raw = Value::deserialize(de)?;
    let parsed = match &raw {
        Value::Number(n) => n.as_u64().map(|n| n as usize),
        Value::String(s) => s.trim().parse::<usize>().ok(),
        _ => None,
    };
    parsed.ok_or_else(|| serde::de::Error::custom(format!("max_attempts must be a whole number, got {raw}")))
}

/// A duration flag as milliseconds: unset or empty is `None`; anything that
/// is not a duration (a bare number names no unit) is refused.
fn duration_ms(value: &Value, flag: &str) -> Result<Option<u64>, PipelineError> {
    let text = match value {
        Value::Null => return Ok(None),
        Value::String(text) if text.trim().is_empty() => return Ok(None),
        Value::String(text) => text.trim().to_string(),
        other => other.to_string(),
    };
    let duration = crate::pipeline::nodes::shared::units::duration(&text, flag, CONFIG_CODE)?;
    Ok(Some(duration.as_millis().min(u128::from(u64::MAX)) as u64))
}

/// A duration in milliseconds the way a flag writes it (`30s`, `1500ms`).
fn shown(ms: u64) -> String {
    crate::pipeline::nodes::shared::units::describe_duration(std::time::Duration::from_millis(ms))
}

/// The budgets, read from the config once at build: the times in ms.
#[derive(Debug, Clone, Default)]
pub struct Budgets {
    pub max_attempts: usize,
    pub delay_ms: Option<u64>,
    pub backoff: Option<f64>,
    pub max_delay_ms: Option<u64>,
    pub max_elapsed_ms: Option<u64>,
}

impl Budgets {
    /// Every budget checked: a count of at least 1, a backoff of at least 1,
    /// and each time a duration.
    pub fn of(config: &Config) -> Result<Self, PipelineError> {
        if config.max_attempts == 0 {
            return Err(PipelineError::new(CONFIG_CODE, "--max-attempts must be greater than 0"));
        }
        if let Some(factor) = config.backoff {
            if !factor.is_finite() || factor < 1.0 {
                return Err(PipelineError::new(
                    CONFIG_CODE,
                    format!("--backoff must be a factor of 1 or more (got {factor}); 1 is a fixed delay, 2 doubles it"),
                ));
            }
        }
        Ok(Self {
            max_attempts: config.max_attempts,
            delay_ms: duration_ms(&config.delay, "--delay")?,
            backoff: config.backoff,
            max_delay_ms: duration_ms(&config.max_delay, "--max-delay")?,
            max_elapsed_ms: duration_ms(&config.max_elapsed, "--max-elapsed")?,
        })
    }

    /// The wait after attempt `attempt` (1-based): `delay × backoff^(attempt-1)`,
    /// capped by `--max-delay`. 0 without a delay.
    pub fn delay_after(&self, attempt: usize) -> u64 {
        let Some(base) = self.delay_ms.filter(|d| *d > 0) else {
            return 0;
        };
        let factor = self.backoff.filter(|f| f.is_finite() && *f > 1.0);
        let grown = match factor {
            Some(f) => {
                let exp = attempt.saturating_sub(1) as i32;
                let v = (base as f64) * f.powi(exp);
                if v.is_finite() && v < u64::MAX as f64 { v.round() as u64 } else { u64::MAX }
            }
            None => base,
        };
        match self.max_delay_ms {
            Some(cap) => grown.min(cap),
            None => grown,
        }
    }
}

fn now_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

/// Which budget ended a loop, for the `failed` payload and the trace.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum GaveUp {
    Attempts,
    Elapsed,
}

impl GaveUp {
    fn reason(self) -> &'static str {
        match self {
            GaveUp::Attempts => "max_attempts",
            GaveUp::Elapsed => "max_elapsed",
        }
    }
}

pub struct Node {
    budgets: Budgets,
    when: Option<CompiledProgram>,
    language: std::sync::Arc<dyn LanguageEngine>,
}

impl Node {
    pub fn new(
        node_id: &str,
        config: Config,
        language: std::sync::Arc<dyn LanguageEngine>,
    ) -> Result<Self, PipelineError> {
        let budgets = Budgets::of(&config)?;
        let when = match config.when.trim() {
            "" => None,
            expression => Some(compile_when(node_id, expression, language.as_ref())?),
        };
        Ok(Self {
            budgets,
            when,
            language,
        })
    }
}

/// `--when` compiled the way `logic.if --when` is: `input` is the payload,
/// `$trigger` and `$nodes` are in scope, the answer is `Boolean(...)`.
fn compile_when(
    node_id: &str,
    expression: &str,
    language: &dyn LanguageEngine,
) -> Result<CompiledProgram, PipelineError> {
    let expression = &crate::pipeline::expr::optional_nodes_paths(expression);
    let source = format!(
        "var __scope = input;\n\
         var input = __scope.$input;\n\
         var $input = input;\n\
         var $item = __scope.$item;\n\
         var $index = __scope.$index;\n\
         var $count = __scope.$count;\n\
         var $trigger = __scope.$trigger || null;\n\
         var $nodes = __scope.$nodes || {{}};\n\
         return Boolean({expression});"
    );
    let module = ModuleSource {
        id: format!("logic.retry:{node_id}"),
        source_path: None,
        kind: SourceKind::Tsx,
        code: source,
    };
    let ir = language.parse(&module).map_err(|e| {
        PipelineError::new(
            "FW_NODE_LOGIC_RETRY_WHEN_PARSE",
            format!("node '{node_id}': --when: {e}"),
        )
    })?;
    language
        .compile(
            &ir,
            &CompileOptions {
                target: COMPILE_TARGET_BACKEND.to_string(),
                optimize_level: 1,
                emit_trace_hints: false,
            },
        )
        .map_err(|e| {
            PipelineError::new(
                "FW_NODE_LOGIC_RETRY_WHEN_COMPILE",
                format!("node '{node_id}': --when: {e}"),
            )
        })
}

fn retry_state(payload: &Value) -> Option<&Value> {
    payload.get(RETRY_STATE_KEY)
}

fn attempt_of(state: Option<&Value>) -> Option<usize> {
    state
        .and_then(|value| value.get("attempt"))
        .and_then(Value::as_u64)
        .map(|n| n as usize)
}

/// Where the engine put the failure it delivered: the key its retry state
/// names, when that key answers `ok: false` with an error `code`. Only the
/// engine stamps `error_key` (this node's own stamp never keeps it), so a
/// verdict payload carrying its own `error` (a provider's answer, say), or
/// the state left from an earlier round, is not this.
fn failure_key_of(payload: &Value) -> Option<&str> {
    let key = retry_state(payload)?.get("error_key")?.as_str()?;
    let answer = payload.get(key)?;
    (answer.get("ok") == Some(&Value::Bool(false))
        && answer.get("error").and_then(|e| e.get("code")).and_then(Value::as_str).is_some())
    .then_some(key)
}

/// The retry state carried into this run, wherever it rode: on the payload
/// itself (both roads — on the error road the engine kept what this node
/// stamped last round and counted the failure), or on this node's own last
/// output in `$nodes.<self>`.
struct PriorState {
    attempt: usize,
    first_at: Option<u64>,
}

fn prior_state(input: &NodeExecutionInput) -> PriorState {
    let payload = &input.payload;
    let candidates = [
        retry_state(payload),
        input
            .metadata
            .get("nodes")
            .and_then(|nodes| nodes.get(&input.node_id))
            .and_then(retry_state),
    ];
    let attempt = candidates.iter().filter_map(|s| attempt_of(*s)).max().unwrap_or(0);
    let first_at = candidates
        .iter()
        .filter_map(|s| s.and_then(|v| v.get("first_at")).and_then(Value::as_u64))
        .min();
    PriorState { attempt, first_at }
}

/// The state this node stamps on what it emits. `failing_node_id` and kind
/// are kept from the engine's envelope on the error road, so the record
/// still says whose failure was retried.
struct StateStamp<'a> {
    attempt: usize,
    max_attempts: usize,
    first_at: u64,
    next_delay_ms: Option<u64>,
    gave_up: Option<(GaveUp, &'a str)>,
    engine: Option<&'a Value>,
}

impl StateStamp<'_> {
    fn to_value(&self) -> Value {
        let mut state = serde_json::Map::new();
        if let Some(Value::Object(engine)) = self.engine {
            for key in ["failing_node_id", "failing_node_kind"] {
                if let Some(v) = engine.get(key) {
                    state.insert(key.to_string(), v.clone());
                }
            }
        }
        state.insert("attempt".to_string(), json!(self.attempt));
        state.insert("max_attempts".to_string(), json!(self.max_attempts));
        state.insert("first_at".to_string(), json!(self.first_at));
        if let Some(delay) = self.next_delay_ms {
            state.insert("next_delay_ms".to_string(), json!(delay));
        }
        if let Some((why, message)) = self.gave_up {
            state.insert("reason".to_string(), json!(why.reason()));
            state.insert("message".to_string(), json!(message));
        }
        Value::Object(state)
    }

    /// The payload with the state on it; a non-object payload is wrapped.
    fn stamp(&self, payload: Value) -> Value {
        let state = self.to_value();
        match payload {
            Value::Object(mut map) => {
                map.insert(RETRY_STATE_KEY.to_string(), state);
                Value::Object(map)
            }
            other => json!({ "input": other, RETRY_STATE_KEY: state }),
        }
    }
}

/// The failing node's input given back, for the `retry` pin: the payload
/// with the failure taken off its key, and what that key held before put
/// back (`shadowed`). The retry state is restamped by the caller.
fn original_input(payload: &Value, key: &str) -> Value {
    let mut original = payload.as_object().cloned().unwrap_or_default();
    original.remove(key);
    if let Some(previous) = retry_state(payload).and_then(|s| s.get("shadowed")) {
        original.insert(key.to_string(), previous.clone());
    }
    Value::Object(original)
}

/// What the node decided for attempt `attempt`: wait `Some(delay)` and go
/// round again, or give up for the budget named.
enum Decision {
    Retry { delay_ms: u64 },
    GiveUp(GaveUp, String),
}

fn decide(budgets: &Budgets, attempt: usize, first_at: u64, now: u64) -> Decision {
    let elapsed = now.saturating_sub(first_at);
    if let Some(budget) = budgets.max_elapsed_ms {
        if elapsed >= budget {
            return Decision::GiveUp(
                GaveUp::Elapsed,
                format!(
                    "retry gave up: --max-elapsed {} spent after {attempt} attempt(s) ({elapsed} ms elapsed)",
                    shown(budget)
                ),
            );
        }
    }
    if attempt >= budgets.max_attempts {
        return Decision::GiveUp(
            GaveUp::Attempts,
            format!(
                "retry gave up: --max-attempts {} spent ({elapsed} ms elapsed)",
                budgets.max_attempts
            ),
        );
    }
    let delay_ms = budgets.delay_after(attempt);
    if let Some(budget) = budgets.max_elapsed_ms {
        if elapsed.saturating_add(delay_ms) > budget {
            return Decision::GiveUp(
                GaveUp::Elapsed,
                format!(
                    "retry gave up: --max-elapsed {} would be overrun by the next wait of {} after {attempt} attempt(s) ({elapsed} ms elapsed)",
                    shown(budget),
                    shown(delay_ms)
                ),
            );
        }
    }
    Decision::Retry { delay_ms }
}

impl Node {
    async fn verdict(&self, input: &NodeExecutionInput) -> Result<bool, PipelineError> {
        let Some(compiled) = &self.when else {
            return Ok(input.payload.get(VERDICT_KEY) == Some(&Value::Bool(true)));
        };
        let metadata = &input.metadata;
        let text = |key: &str| {
            metadata
                .get(key)
                .and_then(Value::as_str)
                .unwrap_or_default()
                .to_string()
        };
        let out = self
            .language
            .run(
                compiled,
                build_expression_scope_input(&input.payload, metadata),
                &crate::language::ExecutionContext {
                    project: text("project"),
                    pipeline: text("pipeline"),
                    request_id: text("request_id"),
                    trigger: metadata.get("trigger").cloned().unwrap_or(Value::Null),
                    metadata: metadata.clone(),
                },
            )
            .map_err(|e| {
                PipelineError::new(
                    "FW_NODE_LOGIC_RETRY_WHEN_RUN",
                    format!("node '{}': --when: {e}", input.node_id),
                )
            })?;
        Ok(match &out.value {
            Value::Bool(b) => *b,
            Value::Null => false,
            Value::Number(n) => n.as_f64().map(|f| f != 0.0).unwrap_or(false),
            Value::String(s) => !s.is_empty(),
            _ => true,
        })
    }

    /// One decision for both roads: wait and go round, or give up naming the
    /// budget. `payload` is what `retry` carries (the original input on the
    /// error road, the verdict payload itself on the verdict road);
    /// `failed_payload` is what `failed` carries.
    async fn retry_or_fail(
        &self,
        attempt: usize,
        prior: &PriorState,
        engine_state: Option<&Value>,
        payload: Value,
        failed_payload: Value,
        mut trace: Vec<String>,
    ) -> Result<NodeExecutionOutput, PipelineError> {
        let now = now_ms();
        let first_at = prior.first_at.unwrap_or(now);
        let mut stamp = StateStamp {
            attempt,
            max_attempts: self.budgets.max_attempts,
            first_at,
            next_delay_ms: None,
            gave_up: None,
            engine: engine_state,
        };
        match decide(&self.budgets, attempt, first_at, now) {
            Decision::Retry { delay_ms } => {
                trace.push(format!("next attempt in {delay_ms} ms"));
                stamp.next_delay_ms = Some(delay_ms);
                if delay_ms > 0 {
                    tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
                }
                Ok(NodeExecutionOutput {
                    output_pins: vec![OUTPUT_PIN_RETRY.to_string()],
                    payload: stamp.stamp(payload),
                    trace,
                })
            }
            Decision::GiveUp(why, message) => {
                trace.push(message.clone());
                stamp.gave_up = Some((why, &message));
                Ok(NodeExecutionOutput {
                    output_pins: vec![OUTPUT_PIN_FAILED.to_string()],
                    payload: stamp.stamp(failed_payload),
                    trace,
                })
            }
        }
    }

    /// The error road: the engine's delivery after a node failed, the
    /// failure under `key`.
    async fn after_error(
        &self,
        input: NodeExecutionInput,
        key: String,
        mut trace: Vec<String>,
    ) -> Result<NodeExecutionOutput, PipelineError> {
        let prior = prior_state(&input);
        // The engine already counted this failure; its stamp wins over what
        // this node remembers, which is one behind.
        let attempt = attempt_of(retry_state(&input.payload))
            .unwrap_or(1)
            .max(prior.attempt);
        trace.push("road=error".to_string());
        trace.push(format!("attempt={attempt}"));

        // A refusal is the caller's fault — a bad address, a wrong credential
        // kind, config that cannot work. Running it again changes nothing, so
        // the attempt budget is not spent on it: straight to the failed pin,
        // with the reason in the trace. Only `failed`-class errors (the world's
        // fault — a relay down, a timeout) are worth another try.
        let error = input.payload.get(&key).and_then(|answer| answer.get("error"));
        let error_code = error.and_then(|e| e.get("code")).and_then(Value::as_str);
        let error_message = error.and_then(|e| e.get("message")).and_then(Value::as_str).unwrap_or_default();
        let engine_state = retry_state(&input.payload).cloned();
        if let Some(code) = error_code {
            use crate::pipeline::error_class::{ErrorClass, class_of_error};
            if class_of_error(code, error_message) == ErrorClass::Refused {
                trace.push(format!("refused: {code} — retrying cannot help"));
                let stamp = StateStamp {
                    attempt,
                    max_attempts: self.budgets.max_attempts,
                    first_at: prior.first_at.unwrap_or_else(now_ms),
                    next_delay_ms: None,
                    gave_up: None,
                    engine: engine_state.as_ref(),
                };
                return Ok(NodeExecutionOutput {
                    output_pins: vec![OUTPUT_PIN_FAILED.to_string()],
                    payload: stamp.stamp(input.payload),
                    trace,
                });
            }
        }

        let original = original_input(&input.payload, &key);
        self.retry_or_fail(
            attempt,
            &prior,
            engine_state.as_ref(),
            original,
            input.payload,
            trace,
        )
        .await
    }

    /// The verdict road: a payload from an ordinary edge that says whether
    /// to go round again.
    async fn after_verdict(
        &self,
        input: NodeExecutionInput,
        mut trace: Vec<String>,
    ) -> Result<NodeExecutionOutput, PipelineError> {
        // Counted from the payload when it carried the state, and from this
        // node's own last output otherwise — the higher wins, so a payload
        // replaced on the way round does not reset the loop.
        let prior = prior_state(&input);
        let attempt = prior.attempt + 1;
        trace.push("road=verdict".to_string());
        trace.push(format!("attempt={attempt}"));

        let again = self.verdict(&input).await?;
        trace.push(format!("verdict={}", if again { "retry" } else { "done" }));
        if !again {
            let stamp = StateStamp {
                attempt: attempt.saturating_sub(1),
                max_attempts: self.budgets.max_attempts,
                first_at: prior.first_at.unwrap_or_else(now_ms),
                next_delay_ms: None,
                gave_up: None,
                engine: None,
            };
            return Ok(NodeExecutionOutput {
                output_pins: vec![OUTPUT_PIN_DONE.to_string()],
                payload: stamp.stamp(input.payload),
                trace,
            });
        }
        let payload = input.payload;
        self.retry_or_fail(attempt, &prior, None, payload.clone(), payload, trace)
            .await
    }
}

#[async_trait]
impl NodeHandler for Node {
    fn kind(&self) -> &'static str {
        NODE_KIND
    }
    fn input_pins(&self) -> &'static [&'static str] {
        &[INPUT_PIN_IN]
    }
    fn output_pins(&self) -> &'static [&'static str] {
        &[OUTPUT_PIN_RETRY, OUTPUT_PIN_FAILED, OUTPUT_PIN_DONE]
    }

    async fn execute_async(
        &self,
        input: NodeExecutionInput,
    ) -> Result<NodeExecutionOutput, PipelineError> {
        let trace = vec![
            format!("node_kind={NODE_KIND}"),
            format!("max_attempts={}", self.budgets.max_attempts),
        ];
        if let Some(key) = failure_key_of(&input.payload).map(str::to_string) {
            self.after_error(input, key, trace).await
        } else {
            self.after_verdict(input, trace).await
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    fn node(max_attempts: usize, when: Option<&str>) -> Node {
        node_with(Config {
            max_attempts,
            delay: Value::Null,
            backoff: None,
            max_delay: Value::Null,
            max_elapsed: Value::Null,
            when: when.unwrap_or_default().to_string(),
        })
    }

    fn node_with(config: Config) -> Node {
        Node::new(
            "r",
            config,
            std::sync::Arc::new(crate::language::DenoSandboxEngine::default()),
        )
        .expect("node")
    }

    /// A config with its times as a flag writes them (`500ms`, `8s`).
    fn budgets(max_attempts: usize, delay: &str, backoff: f64, max_delay: Option<&str>, max_elapsed: Option<&str>) -> Config {
        Config {
            max_attempts,
            delay: json!(delay),
            backoff: Some(backoff),
            max_delay: max_delay.map_or(Value::Null, |d| json!(d)),
            max_elapsed: max_elapsed.map_or(Value::Null, |d| json!(d)),
            when: String::new(),
        }
    }

    fn read(config: &Config) -> Budgets {
        Budgets::of(config).expect("budgets")
    }

    fn run(node: &Node, payload: serde_json::Value) -> NodeExecutionOutput {
        run_with(node, payload, json!({}))
    }

    fn run_with(node: &Node, payload: serde_json::Value, metadata: serde_json::Value) -> NodeExecutionOutput {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("runtime")
            .block_on(node.execute_async(NodeExecutionInput {
                node_id: "r".to_string(),
                input_pin: INPUT_PIN_IN.to_string(),
                payload,
                metadata,
                bus: None,
            }))
            .expect("retry node never errors")
    }

    /// A relay being down is worth another try.
    #[test]
    fn a_failed_class_error_is_retried() {
        let out = run(&node(3, None), json!({
            "email": "sari@example.test",
            "message": { "ok": false, "error": { "code": "FW_NODE_MAIL_MESSAGE_SEND", "message": "relay unreachable" } },
            "__zf_retry": { "attempt": 1, "failing_node_id": "m", "error_key": "message" }
        }));
        assert_eq!(out.output_pins, vec![OUTPUT_PIN_RETRY.to_string()]);
        assert_eq!(out.payload["email"], "sari@example.test");
        assert!(out.payload.get("message").is_none(), "the failure is taken off the input given back: {}", out.payload);
        assert_eq!(out.payload["__zf_retry"]["attempt"], 1);
        assert!(out.payload["__zf_retry"].get("error_key").is_none(), "only the engine stamps error_key");
    }

    /// What the failing node's key held before the failure covered it is
    /// what the re-run receives there.
    #[test]
    fn a_retry_gives_back_what_the_failure_covered() {
        let out = run(&node(3, None), json!({
            "query": { "ok": false, "error": { "code": "FW_NODE_POSTGRES_QUERY_RUN", "message": "connection reset" } },
            "__zf_retry": { "attempt": 1, "failing_node_id": "q", "error_key": "query", "shadowed": { "rows": [1] } }
        }));
        assert_eq!(out.output_pins, vec![OUTPUT_PIN_RETRY.to_string()]);
        assert_eq!(out.payload["query"], json!({ "rows": [1] }));
        assert!(out.payload["__zf_retry"].get("shadowed").is_none());
    }

    /// A bad address is the caller's fault. Retrying cannot help, so the
    /// attempt budget is not spent discovering that three times.
    #[test]
    fn a_refused_class_error_goes_straight_to_failed() {
        let out = run(&node(3, None), json!({
            "email": "not an address",
            "message": { "ok": false, "error": { "code": "FW_NODE_MAIL_MESSAGE_SEND_ADDRESS", "message": "not a mailbox" } },
            "__zf_retry": { "attempt": 1, "failing_node_id": "m", "error_key": "message" }
        }));
        assert_eq!(out.output_pins, vec![OUTPUT_PIN_FAILED.to_string()]);
        assert_eq!(out.payload["message"]["error"]["code"], "FW_NODE_MAIL_MESSAGE_SEND_ADDRESS", "the failure rides on to `failed`");
        assert!(
            out.trace.iter().any(|t| t.contains("retrying cannot help")),
            "{:?}",
            out.trace
        );
    }

    /// The verdict road: `retry: true` fires `retry` with the payload as it
    /// came, the attempt counted on it.
    #[test]
    fn a_true_verdict_retries_with_the_original_payload_and_counts() {
        let out = run(&node(5, None), json!({ "retry": true, "state": "processing", "url": null }));
        assert_eq!(out.output_pins, vec![OUTPUT_PIN_RETRY.to_string()]);
        assert_eq!(out.payload["state"], "processing");
        assert_eq!(out.payload["retry"], true);
        assert_eq!(out.payload["__zf_retry"]["attempt"], 1);
        assert_eq!(out.payload["__zf_retry"]["max_attempts"], 5);
        assert!(out.trace.iter().any(|t| t == "road=verdict"), "{:?}", out.trace);
    }

    /// The count survives a payload that forgot it: the node's own last
    /// output is in `$nodes.<self>`, and the higher count wins.
    #[test]
    fn the_count_comes_from_the_nodes_own_last_output_when_the_payload_lost_it() {
        let out = run_with(
            &node(5, None),
            json!({ "retry": true }),
            json!({ "nodes": { "r": { "retry": true, "__zf_retry": { "attempt": 3, "max_attempts": 5 } } } }),
        );
        assert_eq!(out.output_pins, vec![OUTPUT_PIN_RETRY.to_string()]);
        assert_eq!(out.payload["__zf_retry"]["attempt"], 4);
    }

    /// A false verdict is `done`, with the payload passed through.
    #[test]
    fn a_false_verdict_is_done_with_the_payload() {
        let out = run(&node(5, None), json!({ "retry": false, "state": "success", "url": "https://x/v.mp4" }));
        assert_eq!(out.output_pins, vec![OUTPUT_PIN_DONE.to_string()]);
        assert_eq!(out.payload["url"], "https://x/v.mp4");
        assert_eq!(out.payload["state"], "success");
    }

    /// The budget spent on a true verdict is `failed`, the payload kept.
    #[test]
    fn a_true_verdict_with_the_budget_spent_is_failed() {
        let out = run(&node(3, None), json!({ "retry": true, "state": "processing", "__zf_retry": { "attempt": 2 } }));
        assert_eq!(out.output_pins, vec![OUTPUT_PIN_FAILED.to_string()]);
        assert_eq!(out.payload["state"], "processing");
        assert_eq!(out.payload["__zf_retry"]["attempt"], 3);
        assert_eq!(out.payload["__zf_retry"]["reason"], "max_attempts");
        assert!(
            out.trace.iter().any(|t| t.contains("--max-attempts 3 spent")),
            "{:?}",
            out.trace
        );
    }

    /// `--when` reads the payload as `input`, the way `logic.if --when` does.
    #[test]
    fn when_is_a_logic_if_expression_over_input() {
        let node = node(5, Some("input.status !== 'success'"));
        let waiting = run(&node, json!({ "status": "processing" }));
        assert_eq!(waiting.output_pins, vec![OUTPUT_PIN_RETRY.to_string()]);
        let ready = run(&node, json!({ "status": "success" }));
        assert_eq!(ready.output_pins, vec![OUTPUT_PIN_DONE.to_string()]);
    }

    /// A verdict payload that happens to carry an `ok: false` answer (a
    /// provider's), or the state an earlier round left, is still a verdict:
    /// only the engine's `error_key` marks a delivered failure.
    #[test]
    fn a_verdict_payload_with_its_own_error_key_is_not_an_envelope() {
        let out = run(&node(5, None), json!({ "retry": false, "state": "error", "error": { "message": "provider said no" } }));
        assert_eq!(out.output_pins, vec![OUTPUT_PIN_DONE.to_string()]);
        let out = run(&node(5, None), json!({
            "retry": false,
            "response": { "ok": false, "error": { "code": "X", "message": "no" } },
            "__zf_retry": { "attempt": 1, "failing_node_id": "m" }
        }));
        assert_eq!(out.output_pins, vec![OUTPUT_PIN_DONE.to_string()]);
    }
    /// The rate-limited-API example: 500 ms doubled each attempt, held at
    /// 8000 — the waits after attempts 1..6.
    #[test]
    fn the_backoff_sequence_doubles_and_the_cap_holds() {
        let config = read(&budgets(6, "500ms", 2.0, Some("8s"), None));
        let waits: Vec<u64> = (1..=6).map(|attempt| config.delay_after(attempt)).collect();
        assert_eq!(waits, vec![500, 1000, 2000, 4000, 8000, 8000]);
        // No backoff (or 1) is the fixed delay it always was; no delay is 0.
        assert_eq!(read(&budgets(6, "500ms", 1.0, None, None)).delay_after(4), 500);
        assert_eq!(read(&Config { delay: Value::Null, ..budgets(6, "500ms", 2.0, None, None) }).delay_after(4), 0);
    }

    /// The wait the node took is stamped on the payload as `next_delay_ms`,
    /// and `first_at` is set once and carried round unchanged.
    #[test]
    fn the_stamp_carries_first_at_and_the_delay_taken() {
        let node = node_with(budgets(6, "10ms", 2.0, Some("30ms"), None));
        let first = run(&node, json!({ "retry": true }));
        assert_eq!(first.output_pins, vec![OUTPUT_PIN_RETRY.to_string()]);
        let first_at = first.payload["__zf_retry"]["first_at"].as_u64().expect("first_at");
        assert!(first_at > 0);
        assert_eq!(first.payload["__zf_retry"]["next_delay_ms"], 10);

        let second = run(&node, first.payload.clone());
        assert_eq!(second.payload["__zf_retry"]["attempt"], 2);
        assert_eq!(second.payload["__zf_retry"]["first_at"], first_at);
        assert_eq!(second.payload["__zf_retry"]["next_delay_ms"], 20);

        let third = run(&node, second.payload.clone());
        assert_eq!(third.payload["__zf_retry"]["next_delay_ms"], 30, "capped by --max-delay");
    }

    /// The error road carries `first_at` too: the engine keeps the state
    /// this node stamped last round and counts the failure on it.
    #[test]
    fn the_error_road_reads_first_at_from_the_kept_state_and_keeps_the_failing_node() {
        let node = node_with(budgets(5, "0ms", 1.0, None, None));
        let out = run(&node, json!({
            "email": "x@example.test",
            "message": { "ok": false, "error": { "code": "FW_NODE_MAIL_MESSAGE_SEND", "message": "relay unreachable" } },
            "__zf_retry": { "attempt": 2, "first_at": 1000, "failing_node_id": "m", "failing_node_kind": "mail.message.send", "error_key": "message" }
        }));
        assert_eq!(out.output_pins, vec![OUTPUT_PIN_RETRY.to_string()]);
        assert_eq!(out.payload["__zf_retry"]["attempt"], 2);
        assert_eq!(out.payload["__zf_retry"]["first_at"], 1000);
        assert_eq!(out.payload["__zf_retry"]["failing_node_id"], "m");
        assert_eq!(out.payload["__zf_retry"]["max_attempts"], 5);
    }

    /// The elapsed budget spent fires `failed` with the elapsed message even
    /// though attempts remain.
    #[test]
    fn the_elapsed_budget_spent_is_failed_with_the_elapsed_message() {
        let node = node_with(budgets(40, "0ms", 1.0, None, Some("5s")));
        let long_ago = now_ms() - 10_000;
        let out = run(&node, json!({ "retry": true, "__zf_retry": { "attempt": 2, "first_at": long_ago } }));
        assert_eq!(out.output_pins, vec![OUTPUT_PIN_FAILED.to_string()]);
        assert_eq!(out.payload["__zf_retry"]["reason"], "max_elapsed");
        let message = out.payload["__zf_retry"]["message"].as_str().unwrap_or_default();
        assert!(message.contains("--max-elapsed 5s spent"), "{message}");
        assert!(message.contains("3 attempt(s)"), "{message}");
        assert!(out.trace.iter().any(|t| t == message), "{:?}", out.trace);
    }

    /// A wait that would end past the elapsed budget is not taken: `failed`
    /// at once, naming the wait.
    #[test]
    fn a_wait_that_would_overrun_the_elapsed_budget_is_not_taken() {
        // 5s budget, 4s wait, and 2s already gone.
        let node = node_with(budgets(40, "4s", 1.0, None, Some("5s")));
        let out = run(&node, json!({ "retry": true, "__zf_retry": { "attempt": 1, "first_at": now_ms() - 2000 } }));
        assert_eq!(out.output_pins, vec![OUTPUT_PIN_FAILED.to_string()]);
        assert_eq!(out.payload["__zf_retry"]["reason"], "max_elapsed");
        let message = out.payload["__zf_retry"]["message"].as_str().unwrap_or_default();
        assert!(message.contains("overrun by the next wait of 4s"), "{message}");
    }

    /// Both budgets together: whichever runs out first names itself.
    #[test]
    fn both_budgets_together_whichever_first() {
        // Attempts run out first: elapsed is fine, the count is spent.
        let node = node_with(budgets(3, "0ms", 1.0, None, Some("1m")));
        let out = run(&node, json!({ "retry": true, "__zf_retry": { "attempt": 2, "first_at": now_ms() } }));
        assert_eq!(out.output_pins, vec![OUTPUT_PIN_FAILED.to_string()]);
        assert_eq!(out.payload["__zf_retry"]["reason"], "max_attempts");

        // Elapsed runs out first: attempts remain, the clock is spent.
        let node = node_with(budgets(3, "0ms", 1.0, None, Some("100ms")));
        let out = run(&node, json!({ "retry": true, "__zf_retry": { "attempt": 1, "first_at": now_ms() - 500 } }));
        assert_eq!(out.output_pins, vec![OUTPUT_PIN_FAILED.to_string()]);
        assert_eq!(out.payload["__zf_retry"]["reason"], "max_elapsed");

        // Neither: round again.
        let node = node_with(budgets(3, "0ms", 1.0, None, Some("1m")));
        let out = run(&node, json!({ "retry": true, "__zf_retry": { "attempt": 1, "first_at": now_ms() } }));
        assert_eq!(out.output_pins, vec![OUTPUT_PIN_RETRY.to_string()]);
    }

    /// A backoff under 1 would shrink the wait; it is refused at build time.
    #[test]
    fn a_backoff_under_one_is_refused() {
        let err = Node::new(
            "r",
            budgets(3, "100ms", 0.5, None, None),
            std::sync::Arc::new(crate::language::DenoSandboxEngine::default()),
        )
        .err()
        .expect("refused");
        assert_eq!(err.code, "FW_NODE_LOGIC_RETRY_CONFIG");
        assert!(err.message.contains("--backoff"), "{}", err.message);
    }
    /// The dialog saves every value as text; the DSL sends the count as a
    /// number. Both spell the same budgets, and the times are durations.
    #[test]
    fn the_budgets_read_the_dialog_spelling_too() {
        let config: Config = serde_json::from_value(json!({
            "max_attempts": "6", "delay": "500ms", "backoff": "2", "max_delay": "8s", "max_elapsed": ""
        }))
        .expect("dialog spelling");
        let budgets = read(&config);
        assert_eq!(budgets.max_attempts, 6);
        assert_eq!(budgets.delay_ms, Some(500));
        assert_eq!(budgets.backoff, Some(2.0));
        assert_eq!(budgets.max_delay_ms, Some(8000));
        assert_eq!(budgets.max_elapsed_ms, None);
        let config: Config = serde_json::from_value(json!({ "max_attempts": 6, "delay": "500ms", "backoff": 2 })).expect("dsl spelling");
        assert_eq!(read(&config).delay_after(3), 2000);
    }

    /// A time without a unit names no duration and is refused at build,
    /// naming the flag.
    #[test]
    fn a_time_without_a_unit_is_refused() {
        for (key, flag) in [("delay", "--delay"), ("max_delay", "--max-delay"), ("max_elapsed", "--max-elapsed")] {
            let mut raw = json!({ "max_attempts": 3 });
            raw[key] = json!(500);
            let config: Config = serde_json::from_value(raw).expect("config");
            let err = Budgets::of(&config).expect_err("a bare number is refused");
            assert_eq!(err.code, CONFIG_CODE);
            assert!(err.message.contains(flag), "{}", err.message);
        }
    }

    /// The flags are the 0.11 words, typed: durations, a number, an expression.
    #[test]
    fn the_flags_are_durations_and_an_expression() {
        let def = definition();
        let value_of = |flag: &str| def.dsl_flags.iter().find(|f| f.flag == flag).map(|f| f.value.clone());
        assert_eq!(value_of("--delay").as_deref(), Some("duration"));
        assert_eq!(value_of("--max-delay").as_deref(), Some("duration"));
        assert_eq!(value_of("--max-elapsed").as_deref(), Some("duration"));
        assert_eq!(value_of("--when").as_deref(), Some("expression"));
        assert_eq!(value_of("--max-attempts").as_deref(), Some("number"));
        assert!(def.dsl_flags.iter().all(|f| !f.flag.ends_with("-ms")));
    }
}