aion-rs 0.27.1

//! SDK-declared per-activity retry policy, honored at the dispatch seam (#197).
//!
//! The Gleam SDK has always shipped the activity's retry policy to the engine
//! inside the dispatch `config` JSON (`gleam/aion_flow/src/aion/workflow/run.gleam`,
//! `retry_config`), where — until this module — it was parsed by nothing. This
//! module is the consuming half: it decodes that policy and supplies the
//! retryable-failure classification and backoff math the completion-task retry
//! loop (`nif_activity_dispatch::spawn_completion_task`) drives.
//!
//! Classification: the worker reports a STRUCTURED `ActivityErrorKind` on the
//! wire; the server seam collapses it into the SDK's prefixed reason
//! vocabulary (`retryable:<message>` / `terminal:<message>`, see
//! `PendingActivities::complete_activity` in `aion-server`). By the time a
//! failure reaches the [`crate::activity::bridge::ActivityDispatcher`] seam it
//! is that string, so the `retryable:` prefix here IS the wire's structured
//! kind, not a heuristic over free text.
//!
//! Absence semantics match the SDK contract exactly: an activity with no
//! `retry` decorator carries `"retry": null` and runs exactly once — there is
//! no engine-imposed default policy.

use std::time::Duration;

use aion_core::{ActivityId, Event};

use crate::runtime::nif_activity_dispatch::FIRST_DELIVERY_ATTEMPT;

/// Reason prefix the server seam synthesizes from a structured
/// `ActivityErrorKind::Retryable` wire failure (and the in-VM SDK encodes for
/// a typed `error.Retryable`).
pub(super) const RETRYABLE_REASON_PREFIX: &str = "retryable:";

/// Reason prefix for a provider policy refusal. The retry loop handles this
/// class before ordinary retryability so a declared queue hop always wins.
const POLICY_REFUSED_REASON_PREFIX: &str = "policy_refused:";

/// Whether a dispatcher failure is the dedicated provider-policy refusal class.
#[must_use]
pub(super) fn is_policy_refused_reason(reason: &str) -> bool {
    reason.starts_with(POLICY_REFUSED_REASON_PREFIX)
}

/// Reason prefix of the ephemeral parked sentinel (#207). Never recorded to
/// history, never delivered to workflow code, never crossing the SDK wire —
/// it exists only to resolve a local pending waiter whose dispatch the server
/// parked for restart recovery during a graceful drain.
const PARKED_REASON_PREFIX: &str = "parked:";

/// The parked-dispatch sentinel the server's graceful drain resolves an
/// in-flight activity waiter with (#207).
///
/// Parking converges graceful shutdown onto the kill-9 recovery semantics: the
/// durable log keeps its dangling `ActivityScheduled`/`ActivityStarted` trail
/// (the proven re-dispatchable state the replay cursor exhausts into a live
/// re-dispatch), and this sentinel only unblocks the local completion wait so
/// process exit is never wedged on a blocking dispatcher thread. Defined here,
/// next to the reason-classification the retry loop consumes, and re-used by
/// `aion-server`'s drain path so both sides agree byte-for-byte.
pub const PARKED_ACTIVITY_REASON: &str = "parked:server-draining";

/// Whether a dispatcher failure reason is the parked-dispatch sentinel class
/// (#207): the `parked:` prefix, mirroring [`is_retryable_reason`]'s prefix
/// classification. Checked BEFORE the retry-policy filter — a parked dispatch
/// must never consume retry budget nor be delivered as a failure.
#[must_use]
pub fn is_parked_reason(reason: &str) -> bool {
    reason.starts_with(PARKED_REASON_PREFIX)
}

/// Reason prefix of a TRANSPORT-domain worker loss (`aion-server`'s
/// `worker::transport_loss::WORKER_LOST_REASON_PREFIX`): the worker died before
/// the activity reported any result, so the activity NEVER RAN.
///
/// It is deliberately not the `retryable:` class. An authored `retry N` prices
/// the probability that the ACTION fails; a transport death is a different
/// failure domain that produced no result at all, and charging it against the
/// action's budget would make authored semantics weather-dependent — `retry 2`
/// meaning less on a bad network day. So this class is ATTEMPT-NEUTRAL: it
/// consumes no authored budget, records nothing, and re-dispatches the SAME
/// attempt (see [`is_worker_lost_reason`]'s use in the retry loop).
///
/// The transport carries its own ceiling on that neutrality: once the server's
/// transport-loss budget is spent it stops sending this class and sends its
/// `transport-exhausted:` reason instead, which no retry path recognises, so
/// the composed system always terminates.
const WORKER_LOST_REASON_PREFIX: &str = "lost:";

/// Whether a dispatcher failure reason is a transport-domain worker loss.
///
/// True exactly when the reason carries the `lost:` prefix, mirroring
/// [`is_retryable_reason`]'s prefix classification. Checked BEFORE the retry
/// policy — a worker death must never consume an authored retry budget, and
/// must never be delivered as an activity failure while the transport still has
/// budget to deliver the activity.
#[must_use]
pub fn is_worker_lost_reason(reason: &str) -> bool {
    reason.starts_with(WORKER_LOST_REASON_PREFIX)
}

/// Whether a dispatcher failure reason is retryable-class.
///
/// True exactly when the reason carries the `retryable:` prefix — the string
/// form of the wire's structured retryability classification (see module
/// docs). The separate `policy_refused:` class is handled ahead of this check
/// so a live fallback chain wins; every other prefix (`terminal:`, `timeout:`,
/// `cancelled:`, ...) and every unprefixed engine failure is non-retryable.
pub(super) fn is_retryable_reason(reason: &str) -> bool {
    reason.starts_with(RETRYABLE_REASON_PREFIX)
}

/// The SDK-declared retry policy for one activity, decoded from the dispatch
/// `config` JSON.
///
/// Mirrors `aion/activity.RetryPolicy` in the Gleam SDK: `max_attempts` is the
/// TOTAL attempt budget (a policy of 3 means at most 2 retries after the first
/// delivery), and `backoff` is the delay strategy between attempts.
///
/// This is the SDK's convention, and it is NOT AWL's. AWL's `retry N` declares
/// N FURTHER attempts after the first; its two lowerings translate that count
/// into this total budget (`N + 1`) at emission, so nothing here needs to know
/// which surface authored the policy.
#[derive(Clone, Debug, PartialEq)]
pub(super) struct RetryPolicy {
    /// Total one-based attempt budget, including the first delivery.
    pub(super) max_attempts: u32,
    /// Delay strategy applied between a failed attempt and its retry.
    pub(super) backoff: Backoff,
}

/// Backoff strategy between attempts, mirroring `aion/activity.Backoff`.
#[derive(Clone, Debug, PartialEq)]
pub(super) enum Backoff {
    /// `initial * multiplier^(failed_attempt - 1)`, capped at `max`.
    Exponential {
        /// Delay after the first failed attempt.
        initial: Duration,
        /// Per-attempt growth factor.
        multiplier: f64,
        /// Upper bound on any single delay.
        max: Duration,
    },
    /// `initial + increment * (failed_attempt - 1)`, capped at `max`.
    Linear {
        /// Delay after the first failed attempt.
        initial: Duration,
        /// Additive per-attempt growth.
        increment: Duration,
        /// Upper bound on any single delay.
        max: Duration,
    },
    /// The same `delay` between every pair of attempts.
    Fixed {
        /// Constant inter-attempt delay.
        delay: Duration,
    },
}

impl Backoff {
    /// Delay before the retry that follows one-based `failed_attempt`.
    pub(super) fn delay_after(&self, failed_attempt: u32) -> Duration {
        let step = failed_attempt.saturating_sub(1);
        match self {
            Self::Fixed { delay } => *delay,
            Self::Linear {
                initial,
                increment,
                max,
            } => initial
                .saturating_add(increment.saturating_mul(step))
                .min(*max),
            Self::Exponential {
                initial,
                multiplier,
                max,
            } => {
                // f64 milliseconds keeps the growth math simple and saturating;
                // the cap bounds any precision loss to "capped at max".
                let factor = multiplier.powi(i32::try_from(step).unwrap_or(i32::MAX));
                let initial_ms = u64::try_from(initial.as_millis()).unwrap_or(u64::MAX);
                let scaled = precision_safe_mul(initial_ms, factor);
                Duration::from_millis(scaled).min(*max)
            }
        }
    }
}

/// `base * factor` in u64 milliseconds, saturating on overflow, NaN, or a
/// negative factor.
///
/// Precision-lossy by design (f64 mantissa < 64 bits): a backoff delay only
/// needs millisecond fidelity and the product is clamped to u64 range, so the
/// worst case of every lossy cast here is "capped at the policy max".
#[allow(
    clippy::cast_precision_loss,
    clippy::cast_possible_truncation,
    clippy::cast_sign_loss
)]
fn precision_safe_mul(base: u64, factor: f64) -> u64 {
    if !factor.is_finite() || factor <= 0.0 {
        return if factor <= 0.0 { base } else { u64::MAX };
    }
    let product = (base as f64) * factor;
    if product >= u64::MAX as f64 {
        u64::MAX
    } else {
        product as u64
    }
}

/// Decode the SDK-declared retry policy from the dispatch `config` JSON.
///
/// Returns `None` — run exactly once, today's behaviour — when the config
/// carries `"retry": null`, omits the field, or is malformed. A malformed
/// policy object is logged (never silently coerced) and treated as absent: a
/// broken declaration must not invent retry semantics the workflow author
/// never wrote.
pub(super) fn retry_policy_from_config(config: &str) -> Option<RetryPolicy> {
    let value: serde_json::Value = serde_json::from_str(config).ok()?;
    let retry = value.get("retry")?;
    if retry.is_null() {
        return None;
    }
    let policy = decode_policy(retry);
    if policy.is_none() {
        tracing::warn!(
            retry = %retry,
            "malformed SDK retry policy in dispatch config; treating the activity as \
             single-attempt (no retries)"
        );
    }
    policy
}

/// Decode the SDK-declared per-attempt activity timeout from the dispatch
/// `config` JSON (#223).
///
/// Returns `None` — unbounded, the behaviour before this reader existed — when
/// the config omits `timeout_ms`, carries `"timeout_ms": null`, or carries a
/// value that is not a positive whole number of milliseconds.
///
/// Absence is the NORMAL case here, not an edge. The field is written by the
/// Gleam SDK's `activity_dispatch.config` and by nothing else in this
/// workspace, so a dispatch that did not come through it carries no key at
/// all.
///
/// A present-but-unusable value is logged and treated as absent, never
/// coerced, exactly as [`retry_policy_from_config`] treats a malformed policy:
/// a broken declaration must not invent a bound the workflow author never
/// wrote. A non-positive value is that class, and rejecting it mirrors the
/// retry decoder rejecting `max_attempts: 0` — a `timeout 0s` would expire
/// every attempt before it began, which is not a deadline anyone authored.
///
/// Public because the authored bound has a SECOND consumer that must read it
/// identically: `aion-server`'s declared-body executor, which runs the
/// activity's command in the server's own process and applies this bound to
/// that process. Two decoders would be two answers to "what did the document
/// authorise", so there is one, and it is here.
#[must_use]
pub fn activity_timeout_from_config(config: &str) -> Option<Duration> {
    let value: serde_json::Value = serde_json::from_str(config).ok()?;
    let timeout = value.get("timeout_ms")?;
    if timeout.is_null() {
        return None;
    }
    let Some(milliseconds) = timeout.as_u64().filter(|milliseconds| *milliseconds > 0) else {
        tracing::warn!(
            timeout_ms = %timeout,
            "unusable SDK activity timeout in dispatch config; treating the activity as \
             unbounded (no per-attempt deadline)"
        );
        return None;
    };
    Some(Duration::from_millis(milliseconds))
}

/// The dispatcher failure reason recorded when an attempt outlives its
/// authored per-attempt bound (#223).
///
/// Carries the `timeout:` prefix the SDK's `activity_dispatch.parse_error`
/// maps to the typed `error.ActivityTimedOut`, so the authored word surfaces
/// to workflow code as a timeout rather than as an anonymous engine failure.
/// The bound is stated in the message because an operator reading the failure
/// needs to know WHICH deadline fired — this one is per attempt, and the
/// enclosing `with_timeout` scope has its own distinct
/// `timeout:deadline expired`.
///
/// Public for the same reason [`activity_timeout_from_config`] is: an attempt
/// the SERVER ended on this bound must report it in the same words the engine
/// uses when it ends one, or the two halves of one deadline would surface to
/// an operator as two unrelated failures.
#[must_use]
pub fn activity_timeout_reason(bound: Duration) -> String {
    format!(
        "timeout:activity exceeded its per-attempt bound of {}ms",
        bound.as_millis()
    )
}

fn decode_policy(retry: &serde_json::Value) -> Option<RetryPolicy> {
    let max_attempts = u32::try_from(retry.get("max_attempts")?.as_u64()?).ok()?;
    if max_attempts == 0 {
        return None;
    }
    let backoff = retry.get("backoff")?;
    let kind = backoff.get("kind")?.as_str()?;
    let backoff = match kind {
        "exponential" => Backoff::Exponential {
            initial: millis_field(backoff, "initial_ms")?,
            multiplier: backoff.get("multiplier")?.as_f64()?,
            max: millis_field(backoff, "max_ms")?,
        },
        "linear" => Backoff::Linear {
            initial: millis_field(backoff, "initial_ms")?,
            increment: millis_field(backoff, "increment_ms")?,
            max: millis_field(backoff, "max_ms")?,
        },
        "fixed" => Backoff::Fixed {
            delay: millis_field(backoff, "delay_ms")?,
        },
        _ => return None,
    };
    Some(RetryPolicy {
        max_attempts,
        backoff,
    })
}

fn millis_field(value: &serde_json::Value, field: &str) -> Option<Duration> {
    Some(Duration::from_millis(value.get(field)?.as_u64()?))
}

/// Whether the activity (or its whole workflow) already reached a terminal
/// recorded outcome, so an in-flight retry loop must stop recording.
///
/// Guards the completion task's durable retry records against the settle
/// races the workflow thread can win while a retry sleeps or dispatches: a
/// `with_timeout` scope expiry records a terminal `ActivityFailed` for the
/// awaited ordinal, and a workflow terminal ends the run outright. Recording
/// a retry attempt AFTER such a terminal would make the replay walk read the
/// terminal as superseded — so the loop re-checks this under the recorder
/// lock before every append and aborts when the decision was already made.
///
/// Scans NEWEST-first because reopen supersedes terminals: a
/// [`Event::WorkflowReopened`] naming this activity re-drives it live, so the
/// most recent decisive event — the activity's own terminal, a workflow
/// terminal, or the reopen that supersedes them — wins.
pub(super) fn activity_settled(history: &[Event], activity_id: &ActivityId) -> bool {
    for event in history.iter().rev() {
        match event {
            Event::ActivityCompleted {
                activity_id: id, ..
            }
            | Event::ActivityCancelled {
                activity_id: id, ..
            } if id == activity_id => return true,
            Event::ActivityFailed {
                activity_id: id,
                error,
                ..
            } if id == activity_id && error.is_settled() => return true,
            Event::WorkflowReopened { reopened, .. } if reopened.contains(activity_id) => {
                // The reopen supersedes every prior terminal for this
                // activity AND the workflow terminal it belonged to; the
                // activity is live again.
                return false;
            }
            Event::WorkflowCompleted { .. }
            | Event::WorkflowFailed { .. }
            | Event::WorkflowCancelled { .. }
            | Event::WorkflowTimedOut { .. }
            | Event::WorkflowContinuedAsNew { .. } => return true,
            _ => {}
        }
    }
    false
}

/// The recorded reason a recovery re-dispatch stamps onto an ORDINARY
/// activity's dangling attempt: the prior server died with this attempt in
/// flight and nothing else will ever settle it (#266 Defect B).
///
/// The record is written with `ActivityErrorKind::Retryable` — NON-terminal —
/// and is an HONESTY record, not a budget event (ruled 2026-08-17,
/// honesty-only): infrastructure death is attempt-neutral at the budget layer,
/// exactly as [`is_worker_lost_reason`]'s arm treats a worker dying
/// mid-attempt, so re-dispatch proceeds unconditionally and the history cursor
/// walks past this record exactly as it walks past a live retry record (#197).
/// Terminality by retry policy governs action-authored failures only.
///
/// It is NOT written for a declared AGENT action: that holder can outlive the
/// engine by construction, so the same sentence about it would be a lie about
/// live work (#36). See `super::nif_activity_agent` for the fork.
pub(super) const SUPERSEDED_BY_SERVER_DEATH_REASON: &str = "superseded:server-death";

/// The dangling attempt a recovery re-dispatch has to account for, or `None`
/// when there is nothing dangling.
///
/// ONE detector, TWO consumers, because they are asking the same question and
/// must never disagree about the answer: an ordinary activity's dangling
/// attempt is SUPERSEDED (#266 Defect B) and an agent's is OFFERED FOR
/// ADOPTION (#36). What differs is what gets recorded and whether the attempt
/// identity is retained — not which attempt is dangling.
///
/// `Some(n)` exactly when the ordinal's highest recorded attempt `n` has an
/// [`Event::ActivityStarted`] and NO settling record of its own — no
/// [`Event::ActivityFailed`], [`Event::ActivityCompleted`], or
/// [`Event::ActivityCancelled`] at attempt `n`. An already-settled attempt is
/// not adoptable; a fresh ordinal has no identity to adopt; and a
/// Scheduled-with-no-Started trail carries no attempt number at all.
pub(super) fn dangling_attempt(history: &[Event], activity_id: &ActivityId) -> Option<u32> {
    let latest = latest_recorded_attempt(history, activity_id)?;
    let mut started = false;
    for event in history {
        match event {
            Event::ActivityStarted {
                activity_id: id,
                attempt,
                ..
            } if id == activity_id && *attempt == latest => started = true,
            Event::ActivityFailed {
                activity_id: id,
                attempt,
                ..
            }
            | Event::ActivityCompleted {
                activity_id: id,
                attempt,
                ..
            }
            | Event::ActivityCancelled {
                activity_id: id,
                attempt,
                ..
            } if id == activity_id && *attempt == latest => return None,
            _ => {}
        }
    }
    started.then_some(latest)
}

/// The one-based attempt the NEXT live delivery of `activity_id` must carry,
/// derived from recorded history.
///
/// A fresh ordinal (no recorded attempts) is [`FIRST_DELIVERY_ATTEMPT`]. A
/// crash-recovery re-dispatch after a dangling retryable failure — recorded
/// attempt trail with no terminal — continues the trail instead of reusing an
/// attempt number, keeping `(workflow, activity, attempt)` a stable identity
/// across the restart. Legacy histories whose events decode the `attempt`
/// sentinel (`0`) resolve to [`FIRST_DELIVERY_ATTEMPT`] deterministically.
pub(super) fn next_delivery_attempt(history: &[Event], activity_id: &ActivityId) -> u32 {
    latest_recorded_attempt(history, activity_id).map_or(FIRST_DELIVERY_ATTEMPT, |attempt| {
        attempt.saturating_add(1).max(FIRST_DELIVERY_ATTEMPT)
    })
}

/// The highest attempt recorded for `activity_id` on any lifecycle event, or
/// `None` when the ordinal has no recorded attempt trail.
pub(super) fn latest_recorded_attempt(history: &[Event], activity_id: &ActivityId) -> Option<u32> {
    history
        .iter()
        .filter_map(|event| match event {
            Event::ActivityStarted {
                activity_id: id,
                attempt,
                ..
            }
            | Event::ActivityFailed {
                activity_id: id,
                attempt,
                ..
            }
            | Event::ActivityCompleted {
                activity_id: id,
                attempt,
                ..
            } if id == activity_id => Some(*attempt),
            _ => None,
        })
        .max()
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use aion_core::{
        ActivityError, ActivityErrorKind, ActivityId, ContentType, Event, EventEnvelope, Payload,
        WorkflowId,
    };

    use super::{
        Backoff, PARKED_ACTIVITY_REASON, RetryPolicy, activity_settled,
        activity_timeout_from_config, activity_timeout_reason, dangling_attempt, is_parked_reason,
        is_retryable_reason, latest_recorded_attempt, next_delivery_attempt,
        retry_policy_from_config,
    };

    fn config_with(retry: &str) -> String {
        format!(r#"{{"retry":{retry},"timeout_ms":null,"labels":{{}}}}"#)
    }

    /// Build the activity-dispatch config the SDK emits, with `timeout_ms`
    /// set to the supplied JSON.
    fn config_with_timeout(timeout_ms: &str) -> String {
        format!(r#"{{"retry":null,"timeout_ms":{timeout_ms},"labels":{{}}}}"#)
    }

    /// #223. Absence is the NORMAL case — the field is written by the Gleam
    /// SDK and by nothing else, so an omitted key, an explicit `null`, and
    /// unparseable config all mean UNBOUNDED. None of them may be read as a
    /// bound of any size.
    #[test]
    fn absent_null_and_unparseable_configs_decode_to_no_bound() {
        assert_eq!(activity_timeout_from_config("{}"), None);
        assert_eq!(
            activity_timeout_from_config(&config_with_timeout("null")),
            None
        );
        assert_eq!(activity_timeout_from_config("not json"), None);
    }

    /// A positive whole number of milliseconds is the bound, verbatim — no
    /// rounding, no floor, no ceiling imposed by the engine.
    #[test]
    fn a_positive_millisecond_value_is_the_authored_bound() {
        assert_eq!(
            activity_timeout_from_config(&config_with_timeout("1")),
            Some(Duration::from_millis(1)),
            "the smallest expressible bound survives the decode"
        );
        assert_eq!(
            activity_timeout_from_config(&config_with_timeout("5400000")),
            Some(Duration::from_secs(5400)),
            "a 90m action timeout survives the decode"
        );
    }

    /// Degenerate and malformed values decode to NO BOUND rather than to a
    /// bound of zero, mirroring the retry decoder's `max_attempts: 0`
    /// rejection. A zero bound would expire every attempt before it began,
    /// which is not a deadline any author wrote.
    #[test]
    fn degenerate_and_malformed_bounds_decode_to_no_bound() {
        assert_eq!(
            activity_timeout_from_config(&config_with_timeout("0")),
            None
        );
        assert_eq!(
            activity_timeout_from_config(&config_with_timeout("-1")),
            None
        );
        assert_eq!(
            activity_timeout_from_config(&config_with_timeout("1.5")),
            None
        );
        assert_eq!(
            activity_timeout_from_config(&config_with_timeout(r#""30s""#)),
            None
        );
        assert_eq!(
            activity_timeout_from_config(&config_with_timeout("{}")),
            None
        );
    }

    /// 🔴 The coupling that forces the dispatch loop to carry expiry as a
    /// FACT rather than re-read it from the reason string.
    ///
    /// The expiry reason must carry the `timeout:` prefix, because that is
    /// what the SDK's `activity_dispatch.parse_error` maps to the typed
    /// `error.ActivityTimedOut`. That prefix is NOT the retryable class, so
    /// anyone who "simplifies" the loop's `expired || is_retryable_reason(..)`
    /// down to the reason check alone silently makes an authored
    /// `timeout … retry N` mean one attempt.
    #[test]
    fn the_expiry_reason_types_a_timeout_and_is_not_retryable_class() {
        let reason = activity_timeout_reason(Duration::from_secs(30));
        assert!(
            reason.starts_with("timeout:"),
            "the SDK types ActivityTimedOut off this prefix: {reason}"
        );
        assert!(
            reason.contains("30000ms"),
            "an operator must be able to see WHICH bound fired: {reason}"
        );
        assert!(
            !is_retryable_reason(&reason),
            "the retryable classification must NOT come from this string — the loop \
             carries the expiry fact instead: {reason}"
        );
    }

    #[test]
    fn absent_null_and_malformed_policies_decode_to_no_retries() {
        assert_eq!(retry_policy_from_config("{}"), None);
        assert_eq!(retry_policy_from_config(&config_with("null")), None);
        assert_eq!(retry_policy_from_config("not json"), None);
        // Malformed object: missing backoff.
        assert_eq!(
            retry_policy_from_config(&config_with(r#"{"max_attempts":3}"#)),
            None
        );
        // Unknown backoff kind.
        assert_eq!(
            retry_policy_from_config(&config_with(
                r#"{"max_attempts":3,"backoff":{"kind":"warp","delay_ms":5}}"#
            )),
            None
        );
        // A zero attempt budget cannot mean "retry forever"; it decodes to
        // absent.
        assert_eq!(
            retry_policy_from_config(&config_with(
                r#"{"max_attempts":0,"backoff":{"kind":"fixed","delay_ms":5}}"#
            )),
            None
        );
    }

    #[test]
    fn sdk_shaped_policies_decode_exactly() {
        assert_eq!(
            retry_policy_from_config(&config_with(
                r#"{"max_attempts":3,"backoff":{"kind":"fixed","delay_ms":50}}"#
            )),
            Some(RetryPolicy {
                max_attempts: 3,
                backoff: Backoff::Fixed {
                    delay: Duration::from_millis(50)
                },
            })
        );
        assert_eq!(
            retry_policy_from_config(&config_with(
                r#"{"max_attempts":5,"backoff":{"kind":"exponential","initial_ms":100,"multiplier":2.0,"max_ms":1000}}"#
            )),
            Some(RetryPolicy {
                max_attempts: 5,
                backoff: Backoff::Exponential {
                    initial: Duration::from_millis(100),
                    multiplier: 2.0,
                    max: Duration::from_secs(1),
                },
            })
        );
        assert_eq!(
            retry_policy_from_config(&config_with(
                r#"{"max_attempts":4,"backoff":{"kind":"linear","initial_ms":10,"increment_ms":20,"max_ms":45}}"#
            )),
            Some(RetryPolicy {
                max_attempts: 4,
                backoff: Backoff::Linear {
                    initial: Duration::from_millis(10),
                    increment: Duration::from_millis(20),
                    max: Duration::from_millis(45),
                },
            })
        );
    }

    #[test]
    fn backoff_delays_grow_and_cap() {
        let exponential = Backoff::Exponential {
            initial: Duration::from_millis(100),
            multiplier: 2.0,
            max: Duration::from_millis(350),
        };
        assert_eq!(exponential.delay_after(1), Duration::from_millis(100));
        assert_eq!(exponential.delay_after(2), Duration::from_millis(200));
        assert_eq!(exponential.delay_after(3), Duration::from_millis(350));

        let linear = Backoff::Linear {
            initial: Duration::from_millis(10),
            increment: Duration::from_millis(20),
            max: Duration::from_millis(45),
        };
        assert_eq!(linear.delay_after(1), Duration::from_millis(10));
        assert_eq!(linear.delay_after(2), Duration::from_millis(30));
        assert_eq!(linear.delay_after(3), Duration::from_millis(45));

        let fixed = Backoff::Fixed {
            delay: Duration::from_millis(7),
        };
        assert_eq!(fixed.delay_after(1), Duration::from_millis(7));
        assert_eq!(fixed.delay_after(9), Duration::from_millis(7));
    }

    #[test]
    fn reason_classification_follows_the_wire_prefix_only() {
        assert!(is_retryable_reason("retryable:boom"));
        assert!(!is_retryable_reason("terminal:boom"));
        assert!(!is_retryable_reason("timeout:deadline expired"));
        assert!(!is_retryable_reason("cancelled:operator"));
        assert!(!is_retryable_reason("unprefixed engine failure"));
    }

    /// The parked sentinel (#207) is its own class: never retryable-class (it
    /// must not consume retry budget) and detected purely by the `parked:`
    /// prefix, exactly like the retryable classification.
    #[test]
    fn parked_classification_is_prefix_scoped_and_disjoint_from_retryable() {
        assert!(is_parked_reason(PARKED_ACTIVITY_REASON));
        assert!(is_parked_reason("parked:other-drain-vocabulary"));
        assert!(!is_parked_reason(
            "lost:worker lost before reporting a result"
        ));
        assert!(!is_parked_reason("terminal:boom"));
        assert!(!is_parked_reason("unprefixed engine failure"));
        assert!(!is_retryable_reason(PARKED_ACTIVITY_REASON));
    }

    fn envelope(seq: u64) -> EventEnvelope {
        EventEnvelope {
            seq,
            recorded_at: chrono::Utc::now(),
            workflow_id: WorkflowId::new_v4(),
        }
    }

    fn started(seq: u64, ordinal: u64, attempt: u32) -> Event {
        Event::ActivityStarted {
            envelope: envelope(seq),
            activity_id: ActivityId::from_sequence_position(ordinal),
            attempt,
        }
    }

    fn failed(seq: u64, ordinal: u64, attempt: u32, kind: ActivityErrorKind) -> Event {
        Event::ActivityFailed {
            envelope: envelope(seq),
            activity_id: ActivityId::from_sequence_position(ordinal),
            error: ActivityError {
                kind,
                message: "boom".to_owned(),
                details: None,
            },
            attempt,
        }
    }

    fn completed(seq: u64, ordinal: u64, attempt: u32) -> Event {
        Event::ActivityCompleted {
            envelope: envelope(seq),
            activity_id: ActivityId::from_sequence_position(ordinal),
            result: Payload::new(ContentType::Json, br#""r""#.to_vec()),
            attempt,
        }
    }

    fn cancelled(seq: u64, ordinal: u64, attempt: u32) -> Event {
        Event::ActivityCancelled {
            envelope: envelope(seq),
            activity_id: ActivityId::from_sequence_position(ordinal),
            attempt,
        }
    }

    fn scheduled(seq: u64, ordinal: u64) -> Event {
        Event::ActivityScheduled {
            envelope: envelope(seq),
            activity_id: ActivityId::from_sequence_position(ordinal),
            activity_type: "act".to_owned(),
            input: Payload::new(ContentType::Json, b"null".to_vec()),
            task_queue: "default".to_owned(),
            node: None,
        }
    }

    /// #266 Defect B. `Some(n)` names EXACTLY the shape a server death leaves
    /// behind — a Started attempt with no settling record of its own — and
    /// nothing else, because every `None` arm below is a trail where a record
    /// either already exists (#197) or would be fabricated.
    #[test]
    fn supersession_names_only_a_started_and_unsettled_attempt() {
        let target = ActivityId::from_sequence_position(0);
        // The wild shape: Started with nothing after it.
        assert_eq!(dangling_attempt(&[started(1, 0, 1)], &target), Some(1));
        // Mid-trail death: the earlier attempt's failure is recorded, the
        // latest Started dangles — the LATEST attempt is the superseded one.
        assert_eq!(
            dangling_attempt(
                &[
                    started(1, 0, 1),
                    failed(2, 0, 1, ActivityErrorKind::Retryable),
                    started(3, 0, 2),
                ],
                &target
            ),
            Some(2)
        );
        // #197 crash-mid-backoff: the failure IS recorded — no second record.
        assert_eq!(
            dangling_attempt(
                &[
                    started(1, 0, 1),
                    failed(2, 0, 1, ActivityErrorKind::Retryable),
                ],
                &target
            ),
            None
        );
        // A completed attempt is settled, not dangling.
        assert_eq!(
            dangling_attempt(&[started(1, 0, 1), completed(2, 0, 1)], &target),
            None
        );
        // A cancelled attempt is settled, not dangling.
        assert_eq!(
            dangling_attempt(&[started(1, 0, 1), cancelled(2, 0, 1)], &target),
            None
        );
        // A fresh ordinal has nothing to supersede.
        assert_eq!(dangling_attempt(&[], &target), None);
        // Scheduled-with-no-Started carries no attempt number: nothing
        // started, so nothing is fabricated for it (the park-before-start
        // shape — stated, not recorded).
        assert_eq!(dangling_attempt(&[scheduled(1, 0)], &target), None);
        // Another ordinal's dangling trail is invisible to this one.
        assert_eq!(dangling_attempt(&[started(1, 9, 4)], &target), None);
    }

    #[test]
    fn settlement_tracks_terminal_outcomes_not_retryable_attempts() {
        let target = ActivityId::from_sequence_position(0);
        // A retryable attempt trail is NOT settled.
        assert!(!activity_settled(
            &[
                started(1, 0, 1),
                failed(2, 0, 1, ActivityErrorKind::Retryable)
            ],
            &target
        ));
        // A terminal failure settles it.
        assert!(activity_settled(
            &[failed(2, 0, 1, ActivityErrorKind::Terminal)],
            &target
        ));
        // A completion settles it.
        assert!(activity_settled(&[completed(3, 0, 2)], &target));
        // Another ordinal's terminal does not.
        assert!(!activity_settled(
            &[failed(2, 7, 1, ActivityErrorKind::Terminal)],
            &target
        ));
        // A workflow terminal settles every ordinal.
        assert!(activity_settled(
            &[Event::WorkflowFailed {
                envelope: envelope(4),
                error: aion_core::WorkflowError {
                    message: "done".to_owned(),
                    details: None,
                },
            }],
            &target
        ));
        // A reopen naming this activity supersedes both its terminal and the
        // workflow terminal: the activity is live again, NOT settled.
        assert!(!activity_settled(
            &[
                failed(2, 0, 3, ActivityErrorKind::Terminal),
                Event::WorkflowFailed {
                    envelope: envelope(3),
                    error: aion_core::WorkflowError {
                        message: "exhausted".to_owned(),
                        details: None,
                    },
                },
                Event::WorkflowReopened {
                    envelope: envelope(4),
                    run_id: aion_core::RunId::new_v4(),
                    reopened: vec![target.clone()],
                },
            ],
            &target
        ));
        // A reopen naming only OTHER activities leaves this one's terminal
        // decisive.
        assert!(activity_settled(
            &[
                failed(2, 0, 3, ActivityErrorKind::Terminal),
                Event::WorkflowReopened {
                    envelope: envelope(4),
                    run_id: aion_core::RunId::new_v4(),
                    reopened: vec![ActivityId::from_sequence_position(9)],
                },
            ],
            &target
        ));
    }

    #[test]
    fn next_attempt_continues_the_recorded_trail() {
        let target = ActivityId::from_sequence_position(0);
        assert_eq!(next_delivery_attempt(&[], &target), 1);
        assert_eq!(
            next_delivery_attempt(
                &[
                    started(1, 0, 1),
                    failed(2, 0, 1, ActivityErrorKind::Retryable),
                    started(3, 0, 2),
                    failed(4, 0, 2, ActivityErrorKind::Retryable),
                ],
                &target
            ),
            3
        );
        // The legacy attempt sentinel (0) resolves to the first delivery.
        assert_eq!(next_delivery_attempt(&[started(1, 0, 0)], &target), 1);
        // Another ordinal's trail is invisible.
        assert_eq!(next_delivery_attempt(&[started(1, 9, 4)], &target), 1);
        assert_eq!(
            latest_recorded_attempt(&[started(1, 0, 2), completed(2, 0, 2)], &target),
            Some(2)
        );
    }
}