fakecloud_stepfunctions/service/

mod.rs

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

use async_trait::async_trait;
use chrono::Utc;
use http::StatusCode;
use serde_json::{json, Value};
use tokio::sync::Mutex as AsyncMutex;

use fakecloud_core::delivery::DeliveryBus;
use fakecloud_core::pagination::paginate_checked;
use fakecloud_core::registry::ServiceRegistry;
use fakecloud_core::service::{AwsRequest, AwsResponse, AwsService, AwsServiceError};
use fakecloud_core::validation::*;
use fakecloud_dynamodb::SharedDynamoDbState;
use fakecloud_persistence::SnapshotStore;

use crate::interpreter;
use crate::state::{
    Execution, ExecutionStatus, SharedStepFunctionsState, StateMachine, StateMachineStatus,
    StateMachineType, StepFunctionsSnapshot, StepFunctionsState,
    STEPFUNCTIONS_SNAPSHOT_SCHEMA_VERSION,
};

const SUPPORTED: &[&str] = &[
    "CreateStateMachine",
    "DescribeStateMachine",
    "ListStateMachines",
    "DeleteStateMachine",
    "UpdateStateMachine",
    "TagResource",
    "UntagResource",
    "ListTagsForResource",
    "StartExecution",
    "StopExecution",
    "DescribeExecution",
    "ListExecutions",
    "GetExecutionHistory",
    "DescribeStateMachineForExecution",
    "CreateActivity",
    "DeleteActivity",
    "DescribeActivity",
    "ListActivities",
    "GetActivityTask",
    "SendTaskFailure",
    "SendTaskHeartbeat",
    "SendTaskSuccess",
    "PublishStateMachineVersion",
    "DeleteStateMachineVersion",
    "ListStateMachineVersions",
    "CreateStateMachineAlias",
    "DeleteStateMachineAlias",
    "DescribeStateMachineAlias",
    "ListStateMachineAliases",
    "UpdateStateMachineAlias",
    "DescribeMapRun",
    "ListMapRuns",
    "UpdateMapRun",
    "RedriveExecution",
    "StartSyncExecution",
    "TestState",
    "ValidateStateMachineDefinition",
];

/// Handle to the central service registry, set by `main.rs` after every service
/// has been registered. Wrapped in `OnceLock` so `StepFunctionsService` can be
/// constructed (and registered into the very registry it later reads back) before
/// the registry itself is finalized. The interpreter snapshots the inner `Arc`
/// when it needs to dispatch generic `aws-sdk:*` Task integrations.
pub type SharedServiceRegistry = Arc<std::sync::OnceLock<Arc<ServiceRegistry>>>;

pub struct StepFunctionsService {
    state: SharedStepFunctionsState,
    delivery: Option<Arc<DeliveryBus>>,
    dynamodb_state: Option<SharedDynamoDbState>,
    registry: Option<SharedServiceRegistry>,
    snapshot_store: Option<Arc<dyn SnapshotStore>>,
    snapshot_lock: Arc<AsyncMutex<()>>,
}

mod activities;
mod executions;
mod map_runs;
mod state_machines;
mod tags;
mod tasks;
mod validation;

impl StepFunctionsService {
    pub fn new(state: SharedStepFunctionsState) -> Self {
        Self {
            state,
            delivery: None,
            dynamodb_state: None,
            registry: None,
            snapshot_store: None,
            snapshot_lock: Arc::new(AsyncMutex::new(())),
        }
    }

    pub fn with_delivery(mut self, delivery: Arc<DeliveryBus>) -> Self {
        self.delivery = Some(delivery);
        self
    }

    pub fn with_dynamodb(mut self, dynamodb_state: SharedDynamoDbState) -> Self {
        self.dynamodb_state = Some(dynamodb_state);
        self
    }

    /// Hand the service a deferred-fill handle to the central [`ServiceRegistry`].
    /// `main.rs` calls [`OnceLock::set`] on the inner cell after every service
    /// has been registered; until then the interpreter falls back to its
    /// hand-coded SDK integrations (lambda invoke, sqs sendMessage, …).
    pub fn with_registry(mut self, registry: SharedServiceRegistry) -> Self {
        self.registry = Some(registry);
        self
    }

    pub fn with_snapshot_store(mut self, store: Arc<dyn SnapshotStore>) -> Self {
        self.snapshot_store = Some(store);
        self
    }

    async fn save_snapshot(&self) {
        save_stepfunctions_snapshot(
            &self.state,
            self.snapshot_store.clone(),
            &self.snapshot_lock,
        )
        .await;
    }

    /// Build a hook that persists the current Step Functions state when invoked,
    /// or `None` in memory mode (no snapshot store). The CloudFormation
    /// provisioner mutates `state` directly and uses this to write a
    /// CFN-provisioned resource through to disk, the same way a direct mutating
    /// API call would.
    pub fn snapshot_hook(&self) -> Option<fakecloud_persistence::SnapshotHook> {
        let store = self.snapshot_store.clone()?;
        let state = self.state.clone();
        let lock = self.snapshot_lock.clone();
        Some(Arc::new(move || {
            let state = state.clone();
            let store = store.clone();
            let lock = lock.clone();
            Box::pin(async move {
                save_stepfunctions_snapshot(&state, Some(store), &lock).await;
            })
        }))
    }
}

/// Persist the current Step Functions state as a snapshot. Offloads the serde +
/// blocking file write to the Tokio blocking pool. Noop when `store` is `None`
/// (memory mode). Shared by `StepFunctionsService::save_snapshot` and the
/// CloudFormation provisioner's post-provision persist hook so both route
/// through the same serialize-and-write path.
pub async fn save_stepfunctions_snapshot(
    state: &SharedStepFunctionsState,
    store: Option<Arc<dyn SnapshotStore>>,
    lock: &AsyncMutex<()>,
) {
    let Some(store) = store else {
        return;
    };
    let _guard = lock.lock().await;
    let snapshot =
        StepFunctionsSnapshot::of(STEPFUNCTIONS_SNAPSHOT_SCHEMA_VERSION, state.read().clone());
    let join = tokio::task::spawn_blocking(move || -> std::io::Result<()> {
        let bytes = serde_json::to_vec(&snapshot)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?;
        store.save(&bytes)
    })
    .await;
    match join {
        Ok(Ok(())) => {}
        Ok(Err(err)) => tracing::error!(%err, "failed to write stepfunctions snapshot"),
        Err(err) => tracing::error!(%err, "stepfunctions snapshot task panicked"),
    }
}

/// Abort executions that were RUNNING (or PENDING_REDRIVE) when the server
/// stopped, called once after a persistence snapshot is loaded on startup.
///
/// The in-memory interpreter that was driving each one is gone after a restart,
/// so the execution can never advance. Real Step Functions resumes from durable
/// interpreter state; fakecloud can't, so leaving them RUNNING would strand them
/// forever (DescribeExecution would report RUNNING with no way to ever finish).
/// Aborting with a terminal stop date + error/cause is the honest outcome
/// (bug-audit 2026-06-20, 0.A2). Returns the number reconciled.
pub fn reconcile_interrupted_executions(state: &SharedStepFunctionsState) -> usize {
    let now = Utc::now();
    let mut count = 0;
    let mut accounts = state.write();
    for (_, _, s) in accounts.iter_regional_mut() {
        for exec in s.executions.values_mut() {
            if matches!(
                exec.status,
                ExecutionStatus::Running | ExecutionStatus::PendingRedrive
            ) {
                exec.status = ExecutionStatus::Aborted;
                exec.stop_date = Some(now);
                exec.error = Some("Fakecloud.Restart".to_string());
                exec.cause = Some(
                    "Execution was interrupted by a fakecloud restart and cannot be resumed"
                        .to_string(),
                );
                count += 1;
            }
        }
    }
    count
}

fn is_mutating_action(action: &str) -> bool {
    matches!(
        action,
        "CreateStateMachine"
            | "DeleteStateMachine"
            | "UpdateStateMachine"
            | "TagResource"
            | "UntagResource"
            | "StartExecution"
            | "StopExecution"
            | "CreateActivity"
            | "DeleteActivity"
            | "GetActivityTask"
            | "SendTaskFailure"
            | "SendTaskHeartbeat"
            | "SendTaskSuccess"
            | "PublishStateMachineVersion"
            | "DeleteStateMachineVersion"
            | "CreateStateMachineAlias"
            | "DeleteStateMachineAlias"
            | "UpdateStateMachineAlias"
            | "UpdateMapRun"
            | "RedriveExecution"
            | "StartSyncExecution"
    )
}

#[async_trait]
impl AwsService for StepFunctionsService {
    fn service_name(&self) -> &str {
        "states"
    }

    async fn handle(&self, req: AwsRequest) -> Result<AwsResponse, AwsServiceError> {
        let mutates = is_mutating_action(req.action.as_str());
        let result = match req.action.as_str() {
            "CreateStateMachine" => self.create_state_machine(&req),
            "DescribeStateMachine" => self.describe_state_machine(&req),
            "ListStateMachines" => self.list_state_machines(&req),
            "DeleteStateMachine" => self.delete_state_machine(&req),
            "UpdateStateMachine" => self.update_state_machine(&req),
            "TagResource" => self.tag_resource(&req),
            "UntagResource" => self.untag_resource(&req),
            "ListTagsForResource" => self.list_tags_for_resource(&req),
            "StartExecution" => self.start_execution(&req),
            "StopExecution" => self.stop_execution(&req),
            "DescribeExecution" => self.describe_execution(&req),
            "ListExecutions" => self.list_executions(&req),
            "GetExecutionHistory" => self.get_execution_history(&req),
            "DescribeStateMachineForExecution" => self.describe_state_machine_for_execution(&req),
            "CreateActivity" => self.create_activity(&req),
            "DeleteActivity" => self.delete_activity(&req),
            "DescribeActivity" => self.describe_activity(&req),
            "ListActivities" => self.list_activities(&req),
            "GetActivityTask" => self.get_activity_task(&req).await,
            "SendTaskFailure" => self.send_task_failure(&req),
            "SendTaskHeartbeat" => self.send_task_heartbeat(&req),
            "SendTaskSuccess" => self.send_task_success(&req),
            "PublishStateMachineVersion" => self.publish_state_machine_version(&req),
            "DeleteStateMachineVersion" => self.delete_state_machine_version(&req),
            "ListStateMachineVersions" => self.list_state_machine_versions(&req),
            "CreateStateMachineAlias" => self.create_state_machine_alias(&req),
            "DeleteStateMachineAlias" => self.delete_state_machine_alias(&req),
            "DescribeStateMachineAlias" => self.describe_state_machine_alias(&req),
            "ListStateMachineAliases" => self.list_state_machine_aliases(&req),
            "UpdateStateMachineAlias" => self.update_state_machine_alias(&req),
            "DescribeMapRun" => self.describe_map_run(&req),
            "ListMapRuns" => self.list_map_runs(&req),
            "UpdateMapRun" => self.update_map_run(&req),
            "RedriveExecution" => self.redrive_execution(&req),
            "StartSyncExecution" => self.start_sync_execution(&req).await,
            "TestState" => self.test_state(&req),
            "ValidateStateMachineDefinition" => self.validate_state_machine_definition(&req),
            _ => Err(AwsServiceError::action_not_implemented(
                "states",
                &req.action,
            )),
        };
        if mutates && matches!(result.as_ref(), Ok(resp) if resp.status.is_success()) {
            self.save_snapshot().await;
        }
        result
    }

    fn supported_actions(&self) -> &[&str] {
        SUPPORTED
    }
}

impl StepFunctionsService {
    fn list_activities(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
        let body = req.json_body();
        let raw_max_results = body["maxResults"].as_i64();
        if let Some(mr) = raw_max_results {
            validate_max_results(mr)?;
        }
        let next_token = body["nextToken"].as_str();
        if let Some(t) = next_token {
            validate_page_token(t)?;
        }
        // Smithy default: maxResults=0 means "use the service default"
        // (100 for Step Functions), which we honour by treating both
        // `None` and `0` as 100.
        let max_results = match raw_max_results.unwrap_or(0) {
            0 => 100,
            n => n as usize,
        };
        let accounts = self.state.read();
        let empty = crate::state::StepFunctionsState::new(&req.account_id, &req.region);
        let state = accounts
            .regional(&req.account_id, &req.region)
            .unwrap_or(&empty);
        let mut activities: Vec<&crate::state::Activity> = state.activities.values().collect();
        activities.sort_by(|a, b| a.name.cmp(&b.name));
        let items: Vec<Value> = activities
            .iter()
            .map(|a| {
                json!({
                    "activityArn": a.arn,
                    "name": a.name,
                    "creationDate": a.creation_date.timestamp(),
                })
            })
            .collect();
        let (page, token) =
            paginate_checked(&items, next_token, max_results).map_err(|_| invalid_token())?;
        let mut resp = json!({ "activities": page });
        if let Some(t) = token {
            resp["nextToken"] = json!(t);
        }
        Ok(AwsResponse::ok_json(resp))
    }
}

fn state_machine_alias_to_json(alias: &crate::state::StateMachineAlias) -> Value {
    json!({
        "stateMachineAliasArn": alias.arn,
        "name": alias.name,
        "description": alias.description,
        "routingConfiguration": alias.routing_configuration.iter().map(|r| json!({
            "stateMachineVersionArn": r.state_machine_version_arn,
            "weight": r.weight,
        })).collect::<Vec<_>>(),
        "creationDate": alias.creation_date.timestamp(),
        "updateDate": alias.update_date.timestamp(),
    })
}

fn map_run_to_json(mr: &crate::state::MapRun) -> Value {
    // Iterations still in flight while the map run is RUNNING; zero once it has
    // reached a terminal status.
    let accounted = mr.succeeded_count + mr.failed_count;
    let running = if mr.status == "RUNNING" {
        (mr.total_count - accounted).max(0)
    } else {
        0
    };
    // AWS reports parallel `itemCounts` and `executionCounts` maps; with one
    // child execution per item they carry identical tallies here.
    let counts = json!({
        "pending": 0,
        "running": running,
        "succeeded": mr.succeeded_count,
        "failed": mr.failed_count,
        "timedOut": 0,
        "aborted": 0,
        "results": mr.succeeded_count,
        "total": mr.total_count,
        "failuresNotRedrivable": 0,
        "pendingRedrive": 0,
    });
    json!({
        "mapRunArn": mr.map_run_arn,
        "executionArn": mr.execution_arn,
        "maxConcurrency": mr.max_concurrency,
        "toleratedFailurePercentage": mr.tolerated_failure_percentage,
        "toleratedFailureCount": mr.tolerated_failure_count,
        "status": mr.status,
        "startDate": mr.start_date.timestamp(),
        "stopDate": mr.stop_date.map(|d| d.timestamp()),
        "itemCounts": counts,
        "executionCounts": counts,
    })
}

// ─── Helpers ────────────────────────────────────────────────────────────

fn state_machine_to_json(sm: &StateMachine) -> Value {
    let mut resp = json!({
        "name": sm.name,
        "stateMachineArn": sm.arn,
        "definition": sm.definition,
        "roleArn": sm.role_arn,
        "type": sm.machine_type.as_str(),
        "status": sm.status.as_str(),
        "creationDate": sm.creation_date.timestamp() as f64,
        "updateDate": sm.update_date.timestamp() as f64,
        "revisionId": sm.revision_id,
        "label": sm.name,
    });

    if !sm.description.is_empty() {
        resp["description"] = json!(sm.description);
    }

    if let Some(ref logging) = sm.logging_configuration {
        resp["loggingConfiguration"] = logging.clone();
    } else {
        resp["loggingConfiguration"] = json!({
            "level": "OFF",
            "includeExecutionData": false,
            "destinations": [],
        });
    }

    if let Some(ref tracing) = sm.tracing_configuration {
        resp["tracingConfiguration"] = tracing.clone();
    } else {
        resp["tracingConfiguration"] = json!({
            "enabled": false,
        });
    }

    if let Some(ref enc) = sm.encryption_configuration {
        resp["encryptionConfiguration"] = enc.clone();
    }

    resp
}

fn missing(name: &str) -> AwsServiceError {
    AwsServiceError::aws_error(
        StatusCode::BAD_REQUEST,
        "ValidationException",
        format!("The request must contain the parameter {name}."),
    )
}

fn state_machine_not_found(arn: &str) -> AwsServiceError {
    AwsServiceError::aws_error(
        StatusCode::BAD_REQUEST,
        "StateMachineDoesNotExist",
        format!("State Machine Does Not Exist: '{arn}'"),
    )
}

fn activity_not_found(arn: &str) -> AwsServiceError {
    AwsServiceError::aws_error(
        StatusCode::BAD_REQUEST,
        "ActivityDoesNotExist",
        format!("Activity does not exist: {arn}"),
    )
}

fn task_does_not_exist(token: &str) -> AwsServiceError {
    AwsServiceError::aws_error(
        StatusCode::BAD_REQUEST,
        "TaskDoesNotExist",
        format!("Task does not exist: {token}"),
    )
}

fn resource_not_found(arn: &str) -> AwsServiceError {
    AwsServiceError::aws_error(
        StatusCode::BAD_REQUEST,
        "ResourceNotFound",
        format!("Resource not found: '{arn}'"),
    )
}

/// Parse + validate an alias `routingConfiguration` array.
///
/// AWS rules: 1 or 2 routes; weights are 0-100 and sum to 100; each
/// route must include `stateMachineVersionArn`.
fn parse_routing_configuration(
    routes: &[serde_json::Value],
) -> Result<Vec<crate::state::AliasRoute>, AwsServiceError> {
    if routes.is_empty() || routes.len() > 2 {
        return Err(AwsServiceError::aws_error(
            StatusCode::BAD_REQUEST,
            "ValidationException",
            "routingConfiguration must contain 1 or 2 routes.",
        ));
    }
    let parsed: Vec<crate::state::AliasRoute> = routes
        .iter()
        .map(|r| {
            let arn = r["stateMachineVersionArn"].as_str().ok_or_else(|| {
                AwsServiceError::aws_error(
                    StatusCode::BAD_REQUEST,
                    "ValidationException",
                    "routingConfiguration entries must contain stateMachineVersionArn.",
                )
            })?;
            let weight = r["weight"].as_i64().ok_or_else(|| {
                AwsServiceError::aws_error(
                    StatusCode::BAD_REQUEST,
                    "ValidationException",
                    "routingConfiguration entries must contain a numeric weight.",
                )
            })?;
            if !(0..=100).contains(&weight) {
                return Err(AwsServiceError::aws_error(
                    StatusCode::BAD_REQUEST,
                    "ValidationException",
                    format!("Invalid routing weight {weight}; must be 0-100."),
                ));
            }
            Ok(crate::state::AliasRoute {
                state_machine_version_arn: arn.to_string(),
                weight: weight as i32,
            })
        })
        .collect::<Result<_, _>>()?;
    let total: i32 = parsed.iter().map(|r| r.weight).sum();
    if total != 100 {
        return Err(AwsServiceError::aws_error(
            StatusCode::BAD_REQUEST,
            "ValidationException",
            format!("routingConfiguration weights must sum to 100, got {total}."),
        ));
    }
    Ok(parsed)
}

fn validate_name(name: &str) -> Result<(), AwsServiceError> {
    if name.is_empty() || name.len() > 80 {
        return Err(AwsServiceError::aws_error(
            StatusCode::BAD_REQUEST,
            "InvalidName",
            format!("Invalid Name: '{name}' (length must be between 1 and 80 characters)"),
        ));
    }
    // Only allow alphanumeric, hyphens, and underscores
    if !name
        .chars()
        .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
    {
        return Err(AwsServiceError::aws_error(
            StatusCode::BAD_REQUEST,
            "InvalidName",
            format!(
                "Invalid Name: '{name}' (must only contain alphanumeric characters, hyphens, and underscores)"
            ),
        ));
    }
    Ok(())
}

fn validate_definition(definition: &str) -> Result<(), AwsServiceError> {
    let mut parsed: Value = serde_json::from_str(definition).map_err(|e| {
        AwsServiceError::aws_error(
            StatusCode::BAD_REQUEST,
            "InvalidDefinition",
            format!("Invalid State Machine Definition: '{e}'"),
        )
    })?;

    // What follows depends on each state's effective query language.
    crate::jsonata::propagate_query_language(&mut parsed, None);

    if parsed.get("StartAt").and_then(|v| v.as_str()).is_none() {
        return Err(AwsServiceError::aws_error(
            StatusCode::BAD_REQUEST,
            "InvalidDefinition",
            "Invalid State Machine Definition: 'MISSING_START_AT' (StartAt field is required)"
                .to_string(),
        ));
    }

    let states_obj = parsed
        .get("States")
        .and_then(|v| v.as_object())
        .ok_or_else(|| {
            AwsServiceError::aws_error(
                StatusCode::BAD_REQUEST,
                "InvalidDefinition",
                "Invalid State Machine Definition: 'MISSING_STATES' (States field is required)"
                    .to_string(),
            )
        })?;

    let start_at = parsed["StartAt"].as_str().ok_or_else(|| {
        AwsServiceError::aws_error(
            StatusCode::BAD_REQUEST,
            "InvalidDefinition",
            "Invalid State Machine Definition: 'MISSING_START_AT' (StartAt field is required)"
                .to_string(),
        )
    })?;
    if !states_obj.contains_key(start_at) {
        return Err(AwsServiceError::aws_error(
            StatusCode::BAD_REQUEST,
            "InvalidDefinition",
            format!(
                "Invalid State Machine Definition: 'MISSING_TRANSITION_TARGET' \
                 (StartAt '{start_at}' does not reference a valid state)"
            ),
        ));
    }

    validate_states_are_objects("", &parsed)?;
    validate_state_schemas(&parsed)?;
    validate_timeout_field("<root>", &parsed, "TimeoutSeconds")?;

    // Reject malformed JSONPath reference fields. AWS rejects bad reference
    // paths at CreateStateMachine; accepting them here lets a panic-inducing
    // path reach the interpreter at execution time.
    for (state_name, state_val) in states_obj {
        validate_state_paths(state_name, state_val)?;
    }
    validate_integration_schemas("", states_obj)?;

    Ok(())
}

/// Variables one state assigns, through its own, its Choice rules' and its
/// catchers' `Assign`. A JSONPath key's `.$` isn't part of the name; a JSONata one's
/// would be.
fn assigned_variables(state: &Value) -> Vec<String> {
    let mut holders = vec![state];
    for list in ["Choices", "Catch"] {
        holders.extend(state[list].as_array().into_iter().flatten());
    }
    holders
        .into_iter()
        .filter_map(|holder| holder["Assign"].as_object())
        .flat_map(|assign| assign.keys())
        .map(|key| {
            let name = if crate::jsonata::is_jsonata(state) {
                key.as_str()
            } else {
                key.strip_suffix(".$").unwrap_or(key)
            };
            name.to_string()
        })
        .collect()
}

/// What is wrong with a variable name: an identifier of at most 80 characters.
/// Parts of a variable (`x.y`, `x[2]`) cannot be assigned.
fn variable_name_problem(name: &str) -> Option<&'static str> {
    if name.chars().count() > 80 {
        return Some("is too long");
    }
    let mut chars = name.chars();
    let valid = chars.next().is_some_and(|c| c.is_alphabetic() || c == '_')
        && chars.all(|c| c.is_alphanumeric() || c == '_');
    (!valid).then_some("contains invalid characters")
}

/// The fields of a state only one query language has.
const JSONPATH_ONLY_FIELDS: &[&str] = &[
    "InputPath",
    "Parameters",
    "ResultSelector",
    "ResultPath",
    "OutputPath",
    "ItemsPath",
    "MaxConcurrencyPath",
    "ToleratedFailureCountPath",
    "ToleratedFailurePercentagePath",
    "SecondsPath",
    "TimestampPath",
    "ErrorPath",
    "CausePath",
    "TimeoutSecondsPath",
    "HeartbeatSecondsPath",
];
const JSONATA_ONLY_FIELDS: &[&str] = &["Arguments", "Output", "Items"];

/// The fields of a JSONata state, or of a catcher or Choice rule in one, that can
/// hold `{% %}` strings.
const JSONATA_TEMPLATE_FIELDS: &[&str] = &[
    "Arguments",
    "Output",
    "Assign",
    "Condition",
    "Items",
    "ItemSelector",
    "ItemBatcher",
    "ItemReader",
    "ResultWriter",
    "Seconds",
    "Timestamp",
    "Error",
    "Cause",
    "TimeoutSeconds",
    "HeartbeatSeconds",
    "MaxConcurrency",
    "ToleratedFailureCount",
    "ToleratedFailurePercentage",
];

const WAIT_DURATION_FIELDS: &[&str] = &["SecondsPath", "Seconds", "TimestampPath", "Timestamp"];

fn schema_detail(detail: String) -> String {
    format!("SCHEMA_VALIDATION_FAILED: {detail}")
}

/// Exactly one of `fields`, worded as recorded for none and for several. A field
/// counts when it is present, even as null.
fn check_exactly_one(state: &Value, fields: &[&str], path: &str, errors: &mut Vec<String>) {
    let list = fields.join(", ");
    match fields.iter().filter(|f| state.get(**f).is_some()).count() {
        1 => {}
        0 => errors.push(schema_detail(format!(
            "This object should have only one of the following fields: [{list}] at {path}"
        ))),
        _ => errors.push(schema_detail(format!(
            "There should only be one of the following fields: [{list}] at {path}"
        ))),
    }
}

/// Every string in a JSONata template that opens `{%` without closing it. Text
/// merely containing delimiters is a literal.
fn check_jsonata_delimiters(value: &Value, path: &str, errors: &mut Vec<String>) {
    match value {
        Value::String(text) if crate::jsonata::is_unclosed_expression(text) => {
            errors.push(schema_detail(format!(
                "JSONata string must start with {{% and end with %}} at {path}"
            )));
        }
        Value::Object(members) => {
            for (key, member) in members {
                check_jsonata_delimiters(member, &format!("{path}/{key}"), errors);
            }
        }
        Value::Array(items) => {
            for (index, item) in items.iter().enumerate() {
                check_jsonata_delimiters(item, &format!("{path}/{index}"), errors);
            }
        }
        _ => {}
    }
}

fn check_jsonata_holder(holder: &Value, path: &str, errors: &mut Vec<String>) {
    for field in JSONATA_TEMPLATE_FIELDS {
        if let Some(value) = holder.get(*field) {
            check_jsonata_delimiters(value, &format!("{path}/{field}"), errors);
        }
    }
}

/// Everything a state's shape can get wrong independently of the rest, collected
/// so a definition reports them together, as AWS does. `machine_jsonata` is whether
/// the machine sets JSONata as a whole; every state then is JSONata. Under a
/// JSONPath machine a state's own language holds, which by now (the definition's
/// languages propagated) is what it declares or inherits from its enclosing
/// state, and a JSONPath state inside a JSONata Parallel or Map is allowed, as
/// recorded.
fn collect_state_schema_errors(
    at: &str,
    states: &serde_json::Map<String, Value>,
    machine_jsonata: bool,
    errors: &mut Vec<String>,
) {
    for (name, state) in states {
        let path = format!("{at}/States/{name}");
        let declared = state["QueryLanguage"].as_str();
        if machine_jsonata && declared == Some("JSONPath") {
            errors.push(schema_detail(format!(
                "'QueryLanguage' can not be 'JSONPath' if set to 'JSONata' for whole state \
                 machine at {path}"
            )));
        }
        let jsonata = machine_jsonata || declared == Some("JSONata");
        let (language, other, wrong_fields) = if jsonata {
            ("JSONata", "JSONPath", JSONPATH_ONLY_FIELDS)
        } else {
            ("JSONPath", "JSONata", JSONATA_ONLY_FIELDS)
        };
        for field in wrong_fields {
            if state.get(*field).is_some() {
                errors.push(schema_detail(format!(
                    "The QueryLanguage is set to '{language}', but field '{field}' is only \
                     supported for the '{other}' QueryLanguage at {path}"
                )));
            }
        }

        let kind = state["Type"].as_str().unwrap_or_default();
        if !jsonata && kind == "Task" && state.get("Parameters").is_none() {
            if let Some(resource) = state["Resource"]
                .as_str()
                .filter(|resource| resource.contains(":::aws-sdk:"))
            {
                errors.push(schema_detail(format!(
                    "Parameters field is required for resource ARN: {resource} at \
                     {path}/Resource"
                )));
            }
        }
        if !matches!(kind, "Task" | "Parallel" | "Map") {
            for field in ["Retry", "Catch"] {
                if state.get(field).is_some() {
                    errors.push(schema_detail(format!(
                        "Field '{field}' is not supported at {path}"
                    )));
                }
            }
        }
        if matches!(kind, "Task" | "Pass" | "Wait" | "Parallel" | "Map") {
            check_exactly_one(state, &["Next", "End"], &path, errors);
        }
        if kind == "Wait" {
            check_exactly_one(state, WAIT_DURATION_FIELDS, &path, errors);
        }
        // A literal count is an integer in range (`2.0` is not one, as recorded);
        // a JSONata expression's value is checked when it runs.
        let integer_fields: &[(&str, Option<i64>)] = match kind {
            "Wait" => &[("Seconds", Some(MAX_TIMEOUT_SECONDS as i64))],
            "Map" => &[("ToleratedFailureCount", None)],
            _ => &[],
        };
        for (field, max) in integer_fields {
            let Some(value) = state.get(*field) else {
                continue;
            };
            if jsonata && value.is_string() {
                continue;
            }
            let problem = match value.as_i64() {
                None if value.is_u64() => max.map(|max| format!("Maximum value is {max}")),
                None => Some("Expected value of type Integer".to_string()),
                Some(n) if n < 0 => Some("Minimum value is 0".to_string()),
                Some(n) => max
                    .filter(|max| n > *max)
                    .map(|max| format!("Maximum value is {max}")),
            };
            if let Some(problem) = problem {
                errors.push(schema_detail(format!("{problem} at {path}/{field}")));
            }
        }
        if kind == "Fail" && !jsonata {
            for field in ["ErrorPath", "CausePath"] {
                let Some(value) = state.get(field) else {
                    continue;
                };
                let valid = value.as_str().is_some_and(|text| {
                    (text.starts_with('$') && is_reference_or_context_path(text))
                        || crate::intrinsics::is_intrinsic_call(text)
                });
                if !valid {
                    errors.push(schema_detail(format!(
                        "The value must be a valid Reference Path or a valid intrinsic function \
                         call at {path}/{field}"
                    )));
                }
            }
        }

        if jsonata {
            check_jsonata_holder(state, &path, errors);
            for list in ["Catch", "Choices"] {
                for (index, holder) in state[list].as_array().into_iter().flatten().enumerate() {
                    check_jsonata_holder(holder, &format!("{path}/{list}/{index}"), errors);
                }
            }
        }

        for key in ["ItemProcessor", "Iterator"] {
            if let Some(nested) = state[key]["States"].as_object() {
                collect_state_schema_errors(
                    &format!("{path}/{key}"),
                    nested,
                    machine_jsonata,
                    errors,
                );
            }
        }
        for (index, branch) in state["Branches"]
            .as_array()
            .into_iter()
            .flatten()
            .enumerate()
        {
            if let Some(nested) = branch["States"].as_object() {
                collect_state_schema_errors(
                    &format!("{path}/Branches[{index}]"),
                    nested,
                    machine_jsonata,
                    errors,
                );
            }
        }
    }
}

/// Variables are valid names, and a scope can't assign one that an outer scope
/// assigns in another state: a Parallel or Map state's own `Assign` runs after its
/// nested scopes are gone, so they can reuse its names, as recorded.
fn collect_variable_errors(
    at: &str,
    states: &serde_json::Map<String, Value>,
    outer: &[String],
    errors: &mut Vec<String>,
) {
    let assigned_by: Vec<(&String, Vec<String>)> = states
        .iter()
        .map(|(name, state)| (name, assigned_variables(state)))
        .collect();
    for (name, state) in states {
        let state_path = format!("{at}/States/{name}");
        let mut holders = vec![(state_path.clone(), state)];
        for list in ["Choices", "Catch"] {
            for (index, holder) in state[list].as_array().into_iter().flatten().enumerate() {
                holders.push((format!("{state_path}/{list}/{index}"), holder));
            }
        }
        for (path, holder) in holders {
            if holder
                .get("Assign")
                .is_some_and(|assign| !assign.is_object())
            {
                errors.push(schema_detail(format!(
                    "Expected value of type [OBJECT] at {path}/Assign"
                )));
            }
        }
        for variable in assigned_variables(state) {
            if variable == "states" {
                errors.push(format!(
                    "RESERVED_VARIABLE_NAME: Cannot assign to reserved variable '$states'. at \
                     {state_path}"
                ));
            } else if let Some(problem) = variable_name_problem(&variable) {
                errors.push(format!(
                    "INVALID_VARIABLE_NAME: Invalid variable name '{variable}': the variable \
                     name {problem}. at {state_path}/Assign/{variable}"
                ));
            }
            if outer.contains(&variable) {
                errors.push(schema_detail(format!(
                    "The variable '{variable}' is already assigned in an outer scope at \
                     {state_path}/Assign"
                )));
            }
        }

        let visible: Vec<String> = outer
            .iter()
            .cloned()
            .chain(
                assigned_by
                    .iter()
                    .filter(|(other, _)| *other != name)
                    .flat_map(|(_, variables)| variables.iter().cloned()),
            )
            .collect();
        for key in ["ItemProcessor", "Iterator"] {
            if let Some(nested) = state[key]["States"].as_object() {
                collect_variable_errors(&format!("{state_path}/{key}"), nested, &visible, errors);
            }
        }
        for (index, branch) in state["Branches"]
            .as_array()
            .into_iter()
            .flatten()
            .enumerate()
        {
            if let Some(nested) = branch["States"].as_object() {
                collect_variable_errors(
                    &format!("{state_path}/Branches[{index}]"),
                    nested,
                    &visible,
                    errors,
                );
            }
        }
    }
}

/// The recorded `SCHEMA_VALIDATION_FAILED`, `RESERVED_VARIABLE_NAME` and
/// `INVALID_VARIABLE_NAME` problems of a definition, comma-joined.
fn validate_state_schemas(def: &Value) -> Result<(), AwsServiceError> {
    let Some(states) = def["States"].as_object() else {
        return Ok(());
    };
    let mut errors = Vec::new();
    collect_state_schema_errors("", states, def["QueryLanguage"] == "JSONata", &mut errors);
    collect_variable_errors("", states, &[], &mut errors);
    if errors.is_empty() {
        return Ok(());
    }
    Err(AwsServiceError::aws_error(
        StatusCode::BAD_REQUEST,
        "InvalidDefinition",
        format!("Invalid State Machine Definition: '{}'", errors.join(", ")),
    ))
}

/// The fields a `ReaderConfig` may hold.
const READER_CONFIG_FIELDS: &[&str] = &[
    "InputType",
    "Transformation",
    "ManifestType",
    "CSVDelimiter",
    "CSVHeaderLocation",
    "CSVHeaders",
    "ItemsPointer",
    "MaxItems",
    "MaxItemsPath",
];

/// Every state is a JSON object, nested ones too, worded as recorded. A
/// definition reaching this far has a valid `StartAt`.
fn validate_states_are_objects(at: &str, def: &Value) -> Result<(), AwsServiceError> {
    let Some(states) = def["States"].as_object() else {
        return Ok(());
    };
    for (name, state) in states {
        let path = format!("{at}/States/{name}");
        if !state.is_object() {
            let mut detail = format!("Expected value of type [OBJECT] at {path}");
            let targeted = def["StartAt"] == name.as_str()
                || states.values().any(|other| {
                    other["Next"] == name.as_str()
                        || other["Default"] == name.as_str()
                        || ["Choices", "Catch"].iter().any(|list| {
                            other[*list].as_array().is_some_and(|items| {
                                items.iter().any(|i| i["Next"] == name.as_str())
                            })
                        })
                });
            if !targeted {
                detail.push_str(&format!(
                    ", MISSING_TRANSITION_TARGET: State \"{name}\" is not reachable. at {path}"
                ));
            }
            return Err(schema_error(detail));
        }
        for key in ["ItemProcessor", "Iterator"] {
            validate_states_are_objects(&format!("{path}/{key}"), &state[key])?;
        }
        for (index, branch) in state["Branches"]
            .as_array()
            .into_iter()
            .flatten()
            .enumerate()
        {
            validate_states_are_objects(&format!("{path}/Branches[{index}]"), branch)?;
        }
    }
    Ok(())
}

fn schema_error(detail: String) -> AwsServiceError {
    AwsServiceError::aws_error(
        StatusCode::BAD_REQUEST,
        "InvalidDefinition",
        format!("Invalid State Machine Definition: 'SCHEMA_VALIDATION_FAILED: {detail}'"),
    )
}

/// The integration tail of a `Resource`: `s3:getObject` for
/// `arn:aws:states:::s3:getObject`.
fn integration_of(resource: &str) -> Option<&str> {
    fakecloud_aws::arn::arn_resource(resource, "states").and_then(|r| r.strip_prefix("::"))
}

/// The required fields of optimized integrations, which AWS checks when the
/// definition is created rather than when the Task runs. `at` is the JSON
/// pointer of the enclosing `States`' owner.
fn validate_integration_schemas(
    at: &str,
    states: &serde_json::Map<String, Value>,
) -> Result<(), AwsServiceError> {
    for (name, state) in states {
        let path = format!("{at}/States/{name}");
        match state["Type"].as_str() {
            Some("Task") => {
                let is_nested_start = state["Resource"]
                    .as_str()
                    .and_then(integration_of)
                    .is_some_and(|tail| tail.starts_with("states:startExecution"));
                let field = if crate::jsonata::is_jsonata(state) {
                    "Arguments"
                } else {
                    "Parameters"
                };
                // A whole-value expression can only be checked when it runs.
                let is_expression = state[field].is_string();
                let has_arn = ["StateMachineArn", "StateMachineArn.$"]
                    .iter()
                    .any(|key| state[field].get(*key).is_some());
                if is_nested_start && !is_expression && !has_arn {
                    return Err(schema_error(format!(
                        "The field 'StateMachineArn' is required but was missing at \
                         {path}/{field}"
                    )));
                }
            }
            Some("Map") => {
                let reader = &state["ItemReader"];
                if reader["Resource"].as_str().and_then(integration_of) == Some("s3:getObject")
                    && reader["ReaderConfig"].as_object().is_none()
                {
                    return Err(schema_error(format!(
                        "These fields are required: [ReaderConfig] at {path}/ItemReader"
                    )));
                }
                if let Some(config) = reader["ReaderConfig"].as_object() {
                    let at = format!("{path}/ItemReader/ReaderConfig");
                    // As recorded: `InputType: MANIFEST` already names an inventory.
                    if config.contains_key("ManifestType")
                        && config.get("InputType") == Some(&Value::from("MANIFEST"))
                    {
                        return Err(schema_error(format!(
                            "Field 'ManifestType' is not supported at {at}"
                        )));
                    }
                    if let Some(unknown) = config
                        .keys()
                        .find(|field| !READER_CONFIG_FIELDS.contains(&field.as_str()))
                    {
                        return Err(schema_error(format!(
                            "Field '{unknown}' is not supported at {at}"
                        )));
                    }
                }
                for key in ["ItemProcessor", "Iterator"] {
                    if let Some(nested) = state[key]["States"].as_object() {
                        validate_integration_schemas(&format!("{path}/{key}"), nested)?;
                    }
                }
            }
            Some("Parallel") => {
                for (index, branch) in state["Branches"]
                    .as_array()
                    .into_iter()
                    .flatten()
                    .enumerate()
                {
                    if let Some(nested) = branch["States"].as_object() {
                        validate_integration_schemas(&format!("{path}/Branches[{index}]"), nested)?;
                    }
                }
            }
            _ => {}
        }
    }
    Ok(())
}

/// Validate the JSONPath reference fields of a single state. Recurses into the
/// `Choices` array of Choice states, the `Branches` of Parallel states, and the
/// `Iterator`/`ItemProcessor` of Map states. Returns an `InvalidDefinition`
/// error for any syntactically malformed path or payload-template key.
fn validate_state_paths(state_name: &str, state: &Value) -> Result<(), AwsServiceError> {
    validate_timeout_field(state_name, state, "TimeoutSeconds")?;
    validate_timeout_field(state_name, state, "HeartbeatSeconds")?;
    for field in ["InputPath", "OutputPath", "ResultPath"] {
        if let Some(p) = state.get(field).and_then(|v| v.as_str()) {
            // `InputPath`/`OutputPath` accept the literal "null" to mean "no
            // value"; `ResultPath` accepts JSON null (handled separately) but
            // not the string "null". Both accept "$"-rooted reference paths.
            let valid = if field == "ResultPath" {
                is_valid_reference_path(p)
            } else {
                p == "null" || is_valid_path(p)
            };
            if valid {
                continue;
            }
            return Err(invalid_reference_path(state_name, field, p));
        }
    }

    // Reference-path fields that select a value from the input (or the context
    // object). Unlike Input/Output/ResultPath they do not accept the literal
    // "null", but they may reference the context object via `$$`.
    for field in [
        "SecondsPath",
        "TimestampPath",
        "ItemsPath",
        "MaxConcurrencyPath",
        "ToleratedFailureCountPath",
        "ToleratedFailurePercentagePath",
    ] {
        if let Some(p) = state.get(field).and_then(|v| v.as_str()) {
            if is_reference_or_context_path(p) {
                continue;
            }
            return Err(invalid_reference_path(state_name, field, p));
        }
    }

    // Payload templates: every key ending in `.$` must carry a string value
    // that is either a JSONPath reference (`$`/`$$`) or an intrinsic call. AWS
    // rejects non-string and bare-literal `.$` values at CreateStateMachine.
    // In a JSONata state `.$` means nothing: such a key is a literal.
    if !crate::jsonata::is_jsonata(state) {
        for field in ["Parameters", "ResultSelector", "ItemSelector", "Assign"] {
            if let Some(template) = state.get(field) {
                validate_payload_template(state_name, field, template)?;
            }
        }
    }
    match state.get("Type").and_then(|v| v.as_str()) {
        Some("Choice") => {
            if let Some(choices) = state.get("Choices").and_then(|v| v.as_array()) {
                for rule in choices {
                    validate_choice_variables(state_name, rule)?;
                }
            }
        }
        Some("Parallel") => {
            if let Some(branches) = state.get("Branches").and_then(|v| v.as_array()) {
                for branch in branches {
                    validate_nested_definition(state_name, branch)?;
                }
            }
        }
        Some("Map") => {
            // Distributed Map uses `ItemProcessor`; legacy inline Map uses
            // `Iterator`. Both carry a nested `States` block.
            for key in ["ItemProcessor", "Iterator"] {
                if let Some(processor) = state.get(key) {
                    validate_nested_definition(state_name, processor)?;
                }
            }
        }
        _ => {}
    }

    Ok(())
}

/// Largest `TimeoutSeconds` / `HeartbeatSeconds` AWS accepts in an ASL definition.
const MAX_TIMEOUT_SECONDS: u64 = 99_999_999;

/// AWS requires `TimeoutSeconds` / `HeartbeatSeconds` to be a positive integer
/// no larger than 99999999 and rejects anything else at CreateStateMachine /
/// UpdateStateMachine with `SCHEMA_VALIDATION_FAILED`.
fn validate_timeout_field(
    state_name: &str,
    state: &Value,
    field: &str,
) -> Result<(), AwsServiceError> {
    let Some(v) = state.get(field) else {
        return Ok(());
    };
    // In a JSONata state the value may be an expression, checked when it runs.
    if crate::jsonata::is_jsonata(state) && v.as_str().is_some_and(|text| text.starts_with("{%")) {
        return Ok(());
    }
    match v.as_u64() {
        Some(n) if (1..=MAX_TIMEOUT_SECONDS).contains(&n) => Ok(()),
        _ => Err(AwsServiceError::aws_error(
            StatusCode::BAD_REQUEST,
            "InvalidDefinition",
            format!(
                "Invalid State Machine Definition: 'SCHEMA_VALIDATION_FAILED' \
                 (The value of field '{field}' of state '{state_name}' must be a positive \
                 integer between 1 and {MAX_TIMEOUT_SECONDS}, got {v})"
            ),
        )),
    }
}

/// Validate every state inside a nested definition (a Parallel branch or a Map
/// processor). Errors are attributed to the enclosing state so the message
/// still points at a real state name.
fn validate_nested_definition(parent: &str, def: &Value) -> Result<(), AwsServiceError> {
    if let Some(states) = def.get("States").and_then(|v| v.as_object()) {
        for (name, state) in states {
            validate_state_paths(&format!("{parent}/{name}"), state)?;
        }
    }
    Ok(())
}

/// Recursively validate a payload template (Parameters / ResultSelector /
/// ItemSelector). Any object key ending in `.$` must map to a string that is a
/// JSONPath reference or an intrinsic-function invocation.
fn validate_payload_template(
    state_name: &str,
    field: &str,
    template: &Value,
) -> Result<(), AwsServiceError> {
    match template {
        Value::Object(map) => {
            for (key, value) in map {
                if let Some(stripped) = key.strip_suffix(".$") {
                    let expr = value.as_str().ok_or_else(|| {
                        invalid_payload_template(
                            state_name,
                            field,
                            &format!(
                                "the '{key}' field must be a JSONPath or intrinsic string, \
                                 not {value}"
                            ),
                        )
                    })?;
                    let is_ref = expr.starts_with('$');
                    if !is_ref && !crate::intrinsics::is_intrinsic_call(expr) {
                        return Err(invalid_payload_template(
                            state_name,
                            field,
                            &format!(
                                "the '{stripped}.$' field value '{expr}' is neither a JSONPath \
                                 reference nor an intrinsic function"
                            ),
                        ));
                    }
                } else {
                    validate_payload_template(state_name, field, value)?;
                }
            }
        }
        Value::Array(arr) => {
            for v in arr {
                validate_payload_template(state_name, field, v)?;
            }
        }
        _ => {}
    }
    Ok(())
}

/// A `*Path` field value: either a `$`-rooted reference path the interpreter can
/// parse, or a `$$`-rooted context-object reference.
fn is_reference_or_context_path(p: &str) -> bool {
    if let Some(rest) = p.strip_prefix("$$") {
        return is_valid_path(&format!("${rest}"));
    }
    is_valid_path(p)
}

fn invalid_payload_template(state_name: &str, field: &str, detail: &str) -> AwsServiceError {
    AwsServiceError::aws_error(
        StatusCode::BAD_REQUEST,
        "InvalidDefinition",
        format!(
            "Invalid State Machine Definition: 'SCHEMA_VALIDATION_FAILED' \
             (The {field} of state '{state_name}' is invalid: {detail})"
        ),
    )
}

/// Validate `Variable` reference paths inside a Choice rule, recursing through
/// nested `And` / `Or` / `Not` boolean combinators.
fn validate_choice_variables(state_name: &str, rule: &Value) -> Result<(), AwsServiceError> {
    if let Some(v) = rule.get("Variable").and_then(|v| v.as_str()) {
        // A Choice Variable may read the context object (`$$.Execution.Name`).
        if !is_reference_or_context_path(v) {
            return Err(invalid_reference_path(state_name, "Variable", v));
        }
    }
    for combinator in ["And", "Or"] {
        if let Some(nested) = rule.get(combinator).and_then(|v| v.as_array()) {
            for n in nested {
                validate_choice_variables(state_name, n)?;
            }
        }
    }
    if let Some(n) = rule.get("Not") {
        validate_choice_variables(state_name, n)?;
    }
    Ok(())
}

/// A reference path must start with `$` and every `[...]` index segment must be
/// a balanced, non-empty, decimal-integer index. This mirrors the subset of
/// JSONPath the interpreter understands; anything it cannot parse is rejected.
/// A Reference Path (ResultPath): a JSONPath that identifies a single node,
/// i.e. only property names and single array indexes.
fn is_valid_reference_path(path: &str) -> bool {
    crate::jsonpath::JsonPath::parse(path).is_ok_and(|p| p.is_definite())
}

/// A Path (InputPath, OutputPath, ItemsPath, Choice `Variable`, ...): any
/// JSONPath the interpreter can evaluate, including wildcards, slices and
/// filters.
fn is_valid_path(path: &str) -> bool {
    // A variable reference is a path within that variable.
    let within = crate::variables::split_reference(path).map(|(_, within)| within);
    crate::jsonpath::JsonPath::parse(within.as_deref().unwrap_or(path)).is_ok()
}

fn invalid_reference_path(state_name: &str, field: &str, path: &str) -> AwsServiceError {
    AwsServiceError::aws_error(
        StatusCode::BAD_REQUEST,
        "InvalidDefinition",
        format!(
            "Invalid State Machine Definition: 'SCHEMA_VALIDATION_FAILED' \
             (The {field} field of state '{state_name}' is not a valid reference path: '{path}')"
        ),
    )
}

fn execution_not_found(arn: &str) -> AwsServiceError {
    AwsServiceError::aws_error(
        StatusCode::BAD_REQUEST,
        "ExecutionDoesNotExist",
        format!("Execution Does Not Exist: '{arn}'"),
    )
}

fn execution_to_json(exec: &Execution) -> Value {
    let mut resp = json!({
        "executionArn": exec.execution_arn,
        "stateMachineArn": exec.state_machine_arn,
        "name": exec.name,
        "status": exec.status.as_str(),
        "startDate": exec.start_date.timestamp_millis() as f64 / 1000.0,
        "redriveCount": exec.redrive_count,
    });

    if let Some(ref input) = exec.input {
        resp["input"] = json!(input);
    }
    if let Some(ref output) = exec.output {
        resp["output"] = json!(output);
    }
    if let Some(stop) = exec.stop_date {
        resp["stopDate"] = json!(stop.timestamp_millis() as f64 / 1000.0);
    }
    if let Some(ref error) = exec.error {
        resp["error"] = json!(error);
    }
    if let Some(ref cause) = exec.cause {
        resp["cause"] = json!(cause);
    }

    resp
}

/// Convert an event type to the JSON key prefix used in the `HistoryEvent`
/// shape (the full key is `<prefix>EventDetails`).
///
/// AWS collapses every `*StateEntered` event type (Pass/Task/Choice/Wait/Map/
/// Parallel/Succeed/...) into a single `stateEnteredEventDetails` member, and
/// every `*StateExited` into `stateExitedEventDetails`. All other event types
/// map by lowercasing the first character (`TaskScheduled` ->
/// `taskScheduledEventDetails`). Without the collapse, SDK deserializers see an
/// unknown key like `passStateEnteredEventDetails` and return null name/input.
fn camel_to_details_key(event_type: &str) -> String {
    if event_type.ends_with("StateEntered") {
        return "stateEntered".to_string();
    }
    if event_type.ends_with("StateExited") {
        return "stateExited".to_string();
    }
    let mut chars = event_type.chars();
    match chars.next() {
        None => String::new(),
        Some(c) => c.to_lowercase().to_string() + chars.as_str(),
    }
}

/// Compare two StartExecution inputs for STANDARD idempotency. A missing input
/// defaults to `{}` (AWS's default execution input). Comparison is structural
/// (parsed JSON), so insignificant whitespace differences still dedup; inputs
/// that fail to parse fall back to an exact string match.
fn execution_input_matches(stored: Option<&str>, incoming: Option<&str>) -> bool {
    let stored = stored.unwrap_or("{}");
    let incoming = incoming.unwrap_or("{}");
    match (
        serde_json::from_str::<Value>(stored),
        serde_json::from_str::<Value>(incoming),
    ) {
        (Ok(a), Ok(b)) => a == b,
        _ => stored == incoming,
    }
}

fn validate_arn(arn: &str) -> Result<(), AwsServiceError> {
    if !arn.starts_with("arn:") {
        return Err(AwsServiceError::aws_error(
            StatusCode::BAD_REQUEST,
            "InvalidArn",
            format!("Invalid Arn: '{arn}'"),
        ));
    }
    Ok(())
}

/// Enforce the Smithy @length on an ARN-typed field (`Arn` = 1..=256,
/// `LongArn` = 1..=2000). Empty / oversize ARNs map to `InvalidArn`, which
/// every ARN-bearing Step Functions operation declares.
fn validate_arn_length(field: &str, value: &str, max: usize) -> Result<(), AwsServiceError> {
    if value.is_empty() || value.len() > max {
        return Err(AwsServiceError::aws_error(
            StatusCode::BAD_REQUEST,
            "InvalidArn",
            format!("Invalid Arn at '{field}': must be 1..={max} characters"),
        ));
    }
    Ok(())
}

/// Shared error for a malformed (unparseable) `nextToken` reaching the
/// pagination helper — `InvalidToken` is the only pagination error code
/// declared on every list op.
pub(super) fn invalid_token() -> AwsServiceError {
    AwsServiceError::aws_error(StatusCode::BAD_REQUEST, "InvalidToken", "Invalid nextToken")
}

/// `nextToken` is declared as `PageToken` (length 1..=1024). The only
/// error code declared on every paginated list op is `InvalidToken`, so
/// route both empty and oversize tokens through it.
fn validate_page_token(value: &str) -> Result<(), AwsServiceError> {
    if value.is_empty() || value.len() > 1024 {
        return Err(AwsServiceError::aws_error(
            StatusCode::BAD_REQUEST,
            "InvalidToken",
            "nextToken must be 1..=1024 characters",
        ));
    }
    Ok(())
}

/// `maxResults` is typed as `PageSize` (range 0..=1000). The negative
/// probes flip below-min / above-max; since the only error declared on
/// `ListActivities` is `InvalidToken`, we map page-size violations to
/// the same code (matches how AWS surfaces malformed pagination input
/// for ops that don't model a separate `ValidationException`).
fn validate_max_results(value: i64) -> Result<(), AwsServiceError> {
    if !(0..=1000).contains(&value) {
        return Err(AwsServiceError::aws_error(
            StatusCode::BAD_REQUEST,
            "InvalidToken",
            format!("maxResults '{value}' is outside 0..=1000"),
        ));
    }
    Ok(())
}

/// Start a Step Functions execution from a cross-service delivery (e.g. EventBridge).
///
/// This is the public entry point used by `StepFunctionsDeliveryImpl` in the server crate.
/// It mirrors the logic from `StartExecution` but without the AWS request/response wrapper.
/// Start a Step Functions execution from a cross-service delivery (e.g. EventBridge).
///
/// This is the public entry point used by `StepFunctionsDeliveryImpl` in the server crate.
/// It mirrors the logic from `StartExecution` but without the AWS request/response wrapper.
pub fn start_execution_from_delivery(
    state: &SharedStepFunctionsState,
    delivery: &Option<Arc<DeliveryBus>>,
    dynamodb_state: &Option<SharedDynamoDbState>,
    registry: &Option<SharedServiceRegistry>,
    state_machine_arn: &str,
    input: &str,
) {
    // Validate input is valid JSON
    if serde_json::from_str::<serde_json::Value>(input).is_err() {
        tracing::warn!(
            state_machine_arn,
            "Step Functions delivery: invalid JSON input, skipping execution"
        );
        return;
    }

    let execution_name = uuid::Uuid::new_v4().to_string();

    let mut accounts = state.write();
    // The state machine lives in the account and region its ARN names.
    let Some(st) = accounts.by_arn_mut(state_machine_arn) else {
        tracing::warn!(
            state_machine_arn,
            "Step Functions delivery: state machine not found"
        );
        return;
    };
    let sm = match st.state_machines.get(state_machine_arn) {
        Some(sm) => sm,
        None => {
            tracing::warn!(
                state_machine_arn,
                "Step Functions delivery: state machine not found"
            );
            return;
        }
    };

    let sm_name = sm.name.clone();
    let sm_role_arn = sm.role_arn.clone();
    let sm_type = sm.machine_type;
    let definition = sm.definition.clone();
    // No request in this delivery path; the execution inherits the region its
    // state machine ARN was minted in (the request region at CreateStateMachine).
    let region = state_machine_arn
        .split(':')
        .nth(3)
        .unwrap_or("us-east-1")
        .to_string();
    let exec_arn = st.new_execution_arn(&region, sm_type, &sm_name, &execution_name);

    let now = Utc::now();
    let execution = Execution {
        execution_arn: exec_arn.clone(),
        state_machine_arn: state_machine_arn.to_string(),
        state_machine_name: sm_name,
        name: execution_name,
        status: ExecutionStatus::Running,
        input: Some(input.to_string()),
        output: None,
        start_date: now,
        stop_date: None,
        error: None,
        cause: None,
        history_events: vec![],
        parent_execution_arn: None,
        is_sync: false,
        billed_duration_ms: None,
        billed_memory_mb: None,
        role_arn: sm_role_arn,
        redrive_count: 0,
    };

    st.executions.insert(exec_arn.clone(), execution);
    let logging_config = sm.logging_configuration.clone();
    drop(accounts);

    let shared_state = state.clone();
    let delivery = delivery.clone();
    let dynamodb_state = dynamodb_state.clone();
    let registry = registry.clone();
    let input = Some(input.to_string());
    tokio::spawn(async move {
        interpreter::execute_state_machine(
            shared_state,
            exec_arn,
            definition,
            input,
            delivery,
            dynamodb_state,
            registry,
            logging_config,
        )
        .await;
    });
}

#[cfg(test)]
mod integration_schema_tests {
    use super::*;

    fn message(definition: serde_json::Value) -> Option<String> {
        validate_definition(&definition.to_string())
            .err()
            .map(|e| e.message())
    }

    #[test]
    fn nested_start_requires_the_definition_casing() {
        assert_eq!(
            message(serde_json::json!({"StartAt":"T","States":{"T":{
                "Type":"Task",
                "Resource":"arn:aws:states:::states:startExecution.sync:2",
                "Parameters":{"stateMachineArn":"arn"},
                "End":true
            }}})),
            Some(
                "Invalid State Machine Definition: 'SCHEMA_VALIDATION_FAILED: The field \
                 'StateMachineArn' is required but was missing at /States/T/Parameters'"
                    .to_string()
            )
        );
        for key in ["StateMachineArn", "StateMachineArn.$"] {
            let definition = serde_json::json!({"StartAt":"T","States":{"T":{
                "Type":"Task",
                "Resource":"arn:aws:states:::states:startExecution",
                "Parameters":{key:"$.arn"},
                "End":true
            }}});
            assert_eq!(message(definition), None);
        }
    }

    #[test]
    fn get_object_item_reader_requires_reader_config() {
        let map = |reader: serde_json::Value| {
            serde_json::json!({"StartAt":"M","States":{"M":{
                "Type":"Map",
                "ItemProcessor":{"ProcessorConfig":{"Mode":"DISTRIBUTED"},"StartAt":"P",
                    "States":{"P":{"Type":"Pass","End":true}}},
                "ItemReader":reader,
                "End":true
            }}})
        };
        assert_eq!(
            message(map(serde_json::json!({
                "Resource":"arn:aws:states:::s3:getObject",
                "Parameters":{"Bucket":"b","Key":"k"}
            }))),
            Some(
                "Invalid State Machine Definition: 'SCHEMA_VALIDATION_FAILED: These fields are \
                 required: [ReaderConfig] at /States/M/ItemReader'"
                    .to_string()
            )
        );
        assert_eq!(
            message(map(serde_json::json!({
                "Resource":"arn:aws:states:::s3:listObjectsV2",
                "Parameters":{"Bucket":"b"}
            }))),
            None
        );
    }

    #[test]
    fn reader_config_fields_are_checked() {
        let map = |config: serde_json::Value| {
            serde_json::json!({"StartAt":"M","States":{"M":{
                "Type":"Map",
                "ItemProcessor":{"ProcessorConfig":{"Mode":"DISTRIBUTED"},"StartAt":"P",
                    "States":{"P":{"Type":"Pass","End":true}}},
                "ItemReader":{
                    "Resource":"arn:aws:states:::s3:getObject",
                    "ReaderConfig":config,
                    "Parameters":{"Bucket":"b","Key":"k"}
                },
                "End":true
            }}})
        };
        assert_eq!(
            message(map(serde_json::json!({"InputType":"MANIFEST","ManifestType":"S3_INVENTORY"}))),
            Some(
                "Invalid State Machine Definition: 'SCHEMA_VALIDATION_FAILED: Field 'ManifestType' \
                 is not supported at /States/M/ItemReader/ReaderConfig'"
                    .to_string()
            )
        );
        assert!(
            message(map(serde_json::json!({"InputType":"JSON","Bogus":1})))
                .unwrap()
                .contains("Field 'Bogus' is not supported")
        );
        assert_eq!(
            message(map(
                serde_json::json!({"InputType":"CSV","CSVHeaderLocation":"FIRST_ROW","MaxItems":1})
            )),
            None
        );
    }

    #[test]
    fn variable_names_and_scopes_are_checked() {
        let with_assign = |assign: serde_json::Value| serde_json::json!({"StartAt":"P","States":{"P":{"Type":"Pass","Assign":assign,"End":true}}});
        for bad in ["x.y", "x[2]", "1x"] {
            let error = message(with_assign(serde_json::json!({ bad: 1 }))).unwrap();
            assert!(
                error.contains(&format!(
                    "INVALID_VARIABLE_NAME: Invalid variable name '{bad}': the variable name \
                     contains invalid characters. at /States/P/Assign/{bad}"
                )),
                "{error}"
            );
        }
        assert_eq!(
            message(with_assign(serde_json::json!({"states": 1}))),
            Some(
                "Invalid State Machine Definition: 'RESERVED_VARIABLE_NAME: Cannot assign to \
                 reserved variable '$states'. at /States/P'"
                    .to_string()
            )
        );
        assert_eq!(
            message(with_assign(serde_json::json!({"ok_1": 1, "p.$": "$.a"}))),
            None
        );

        // A state's own `Assign` runs after its nested scopes are gone, as recorded.
        let own = serde_json::json!({"StartAt":"M","States":{"M":{
            "Type":"Map",
            "Assign":{"x":1},
            "ItemProcessor":{"StartAt":"P","States":{"P":{"Type":"Pass","Assign":{"x":2},"End":true}}},
            "End":true
        }}});
        assert_eq!(message(own), None);
        let sibling = serde_json::json!({"StartAt":"A","States":{
            "A":{"Type":"Pass","Assign":{"x":1},"Next":"M"},
            "M":{
                "Type":"Map",
                "ItemProcessor":{"StartAt":"P","States":{"P":{"Type":"Pass","Assign":{"x":2},"End":true}}},
                "End":true
            }
        }});
        assert!(message(sibling).unwrap().contains("outer scope"));
    }

    #[test]
    fn several_problems_are_reported_together_and_nested_ones_found() {
        let def = serde_json::json!({"StartAt":"P","States":{"P":{
            "Type":"Parallel",
            "Branches":[{"StartAt":"W","States":{"W":{"Type":"Wait","Next":"X","End":true}}}],
            "End":true
        }}});
        let error = message(def).unwrap();
        assert!(
            error.contains(
                "There should only be one of the following fields: [Next, End] at \
                 /States/P/Branches[0]/States/W, SCHEMA_VALIDATION_FAILED: This object should have \
                 only one of the following fields: [SecondsPath, Seconds, TimestampPath, \
                 Timestamp] at /States/P/Branches[0]/States/W"
            ),
            "{error}"
        );
    }

    #[test]
    fn jsonata_catchers_and_nested_templates_check_their_delimiters() {
        let def = serde_json::json!({"QueryLanguage":"JSONata","StartAt":"T","States":{
            "T":{"Type":"Task","Resource":"arn:aws:states:::aws-sdk:sfn:listStateMachines",
                 "Arguments":{"a":["{% 1"]},
                 "Catch":[{"ErrorEquals":["States.ALL"],"Output":"{% 2","Next":"E"}],
                 "End":true},
            "E":{"Type":"Succeed"}
        }});
        let error = message(def).unwrap();
        assert!(error.contains("at /States/T/Arguments/a/0"), "{error}");
        assert!(error.contains("at /States/T/Catch/0/Output"), "{error}");
    }

    #[test]
    fn variable_references_are_valid_paths() {
        assert!(is_valid_path("$order.items[0]"));
        assert!(is_valid_path("$x"));
        assert!(!is_valid_path("$x["));
    }

    #[test]
    fn a_non_object_state_is_refused_as_recorded() {
        let unreachable = serde_json::json!({"StartAt": "A", "States": {
            "A": {"Type": "Pass", "End": true},
            "B": 1
        }});
        assert_eq!(
            message(unreachable),
            Some(
                "Invalid State Machine Definition: 'SCHEMA_VALIDATION_FAILED: Expected value of \
                 type [OBJECT] at /States/B, MISSING_TRANSITION_TARGET: State \"B\" is not \
                 reachable. at /States/B'"
                    .to_string()
            )
        );
        let start = serde_json::json!({"StartAt": "S", "States": {"S": 1}});
        assert_eq!(
            message(start),
            Some(
                "Invalid State Machine Definition: 'SCHEMA_VALIDATION_FAILED: Expected value of \
                 type [OBJECT] at /States/S'"
                    .to_string()
            )
        );
        let nested = serde_json::json!({"StartAt": "M", "States": {"M": {
            "Type": "Map",
            "ItemProcessor": {"StartAt": "S", "States": {"S": "nope"}},
            "End": true
        }}});
        assert!(message(nested)
            .unwrap()
            .contains("at /States/M/ItemProcessor/States/S"));
    }

    #[test]
    fn nested_start_without_a_template_is_refused() {
        let definition = |language: &str, field: Option<serde_json::Value>| {
            let mut state = serde_json::json!({
                "Type": "Task",
                "Resource": "arn:aws:states:::states:startExecution",
                "End": true
            });
            if let Some(value) = field {
                state["Arguments"] = value;
            }
            serde_json::json!({"QueryLanguage": language, "StartAt": "T", "States": {"T": state}})
        };
        assert!(message(definition("JSONPath", None))
            .unwrap()
            .contains("/States/T/Parameters"));
        assert!(message(definition("JSONata", None))
            .unwrap()
            .contains("/States/T/Arguments"));
        assert_eq!(
            message(definition(
                "JSONata",
                Some(serde_json::json!("{% $args %}"))
            )),
            None
        );
    }

    #[test]
    fn timeout_expressions_need_a_jsonata_state() {
        let definition = |language: &str| {
            serde_json::json!({"QueryLanguage": language, "StartAt": "T", "States": {"T": {
                "Type": "Task",
                "Resource": "arn:aws:states:::sqs:sendMessage",
                "TimeoutSeconds": "{% 5 %}",
                "End": true
            }}})
        };
        assert_eq!(message(definition("JSONata")), None);
        assert!(message(definition("JSONPath")).is_some());
    }

    #[test]
    fn assign_and_reader_config_must_be_objects() {
        let pass = serde_json::json!({"StartAt": "P", "States": {"P": {
            "Type": "Pass", "Assign": 5, "End": true
        }}});
        assert!(message(pass)
            .unwrap()
            .contains("Expected value of type [OBJECT] at /States/P/Assign"));
        let rule = serde_json::json!({"StartAt": "C", "States": {
            "C": {"Type": "Choice", "Choices": [
                {"Variable": "$.a", "IsPresent": true, "Assign": [], "Next": "E"}
            ]},
            "E": {"Type": "Succeed"}
        }});
        assert!(message(rule)
            .unwrap()
            .contains("/States/C/Choices/0/Assign"));
        let map = serde_json::json!({"StartAt": "M", "States": {"M": {
            "Type": "Map",
            "ItemProcessor": {"ProcessorConfig": {"Mode": "DISTRIBUTED"}, "StartAt": "P",
                "States": {"P": {"Type": "Pass", "End": true}}},
            "ItemReader": {
                "Resource": "arn:aws:states:::s3:getObject",
                "ReaderConfig": "JSON",
                "Parameters": {"Bucket": "b", "Key": "k"}
            },
            "End": true
        }}});
        assert!(message(map).unwrap().contains("[ReaderConfig]"));
    }

    #[test]
    fn fail_state_paths_must_be_references_or_intrinsics() {
        let fail = |path: serde_json::Value| serde_json::json!({"StartAt": "F", "States": {"F": {"Type": "Fail", "ErrorPath": path}}});
        assert_eq!(message(fail(serde_json::json!("$.error"))), None);
        assert_eq!(
            message(fail(serde_json::json!("States.Format('{}', $.a)"))),
            None
        );
        // Recorded.
        assert_eq!(
            message(fail(serde_json::json!("notapath"))).unwrap(),
            "Invalid State Machine Definition: 'SCHEMA_VALIDATION_FAILED: The value must be a \
             valid Reference Path or a valid intrinsic function call at /States/F/ErrorPath'"
        );
        assert!(message(fail(serde_json::json!(5))).is_some());
        let cause = serde_json::json!({"StartAt": "F", "States": {"F": {"Type": "Fail", "Error": "e", "CausePath": "notapath"}}});
        assert!(message(cause)
            .unwrap()
            .contains("intrinsic function call at /States/F/CausePath"));
    }

    /// Recorded: a literal count must be an integer, `2.0` included.
    #[test]
    fn literal_counts_must_be_integers() {
        let map = |count: serde_json::Value| {
            serde_json::json!({"StartAt":"M","States":{"M":{
                "Type":"Map",
                "ToleratedFailureCount": count,
                "ItemProcessor":{"StartAt":"P","States":{"P":{"Type":"Pass","End":true}}},
                "End":true
            }}})
        };
        for bad in [serde_json::json!(2.0), serde_json::json!(2.5)] {
            assert_eq!(
                message(map(bad)).unwrap(),
                "Invalid State Machine Definition: 'SCHEMA_VALIDATION_FAILED: Expected value of \
                 type Integer at /States/M/ToleratedFailureCount'"
            );
        }
        assert_eq!(message(map(serde_json::json!(2))), None);
        assert!(message(map(serde_json::json!(-1)))
            .unwrap()
            .contains("Minimum value is 0 at /States/M/ToleratedFailureCount"));
        let literal = |seconds: i64| serde_json::json!({"StartAt":"W","States":{"W":{"Type":"Wait","Seconds":seconds,"End":true}}});
        assert_eq!(message(literal(99_999_999)), None);
        assert_eq!(
            message(literal(100_000_000)).unwrap(),
            "Invalid State Machine Definition: 'SCHEMA_VALIDATION_FAILED: Maximum value is \
             99999999 at /States/W/Seconds'"
        );
        assert!(message(literal(-1))
            .unwrap()
            .contains("Minimum value is 0 at /States/W/Seconds"));
        let wait = serde_json::json!({"QueryLanguage":"JSONata","StartAt":"W","States":{
            "W":{"Type":"Wait","Seconds":1.5,"End":true}
        }});
        assert!(message(wait)
            .unwrap()
            .contains("Expected value of type Integer at /States/W/Seconds"));
        let expression = serde_json::json!({"QueryLanguage":"JSONata","StartAt":"W","States":{
            "W":{"Type":"Wait","Seconds":"{% 1 %}","End":true}
        }});
        assert_eq!(message(expression), None);
    }

    /// Recorded: only a machine-wide JSONata refuses a JSONPath state; under a
    /// JSONata Parallel or Map in a JSONPath machine one is allowed.
    #[test]
    fn jsonpath_states_are_refused_only_under_a_jsonata_machine() {
        let parallel = |machine: Option<&str>, state: Option<&str>| {
            let mut def = serde_json::json!({"StartAt":"A","States":{"A":{
                "Type":"Parallel",
                "Branches":[{"StartAt":"X","States":{"X":{"Type":"Pass","QueryLanguage":"JSONPath","End":true}}}],
                "End":true
            }}});
            if let Some(language) = machine {
                def["QueryLanguage"] = serde_json::json!(language);
            }
            if let Some(language) = state {
                def["States"]["A"]["QueryLanguage"] = serde_json::json!(language);
            }
            def
        };
        assert_eq!(message(parallel(None, Some("JSONata"))), None);
        assert_eq!(
            message(parallel(Some("JSONata"), None)).unwrap(),
            "Invalid State Machine Definition: 'SCHEMA_VALIDATION_FAILED: 'QueryLanguage' can \
             not be 'JSONPath' if set to 'JSONata' for whole state machine at \
             /States/A/Branches[0]/States/X'"
        );
        let map = serde_json::json!({"StartAt":"A","States":{"A":{
            "Type":"Map",
            "QueryLanguage":"JSONata",
            "Items":"{% [1] %}",
            "ItemProcessor":{"StartAt":"X","States":{"X":{"Type":"Pass","QueryLanguage":"JSONPath","End":true}}},
            "End":true
        }}});
        assert_eq!(message(map), None);
    }

    #[test]
    fn nested_states_report_their_full_path() {
        let definition = serde_json::json!({"StartAt":"M","States":{"M":{
            "Type":"Map",
            "ItemProcessor":{"StartAt":"T","States":{"T":{
                "Type":"Task",
                "Resource":"arn:aws:states:::states:startExecution",
                "Parameters":{},
                "End":true
            }}},
            "End":true
        }}});
        assert!(message(definition)
            .unwrap()
            .contains("/States/M/ItemProcessor/States/T/Parameters"));
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use http::{HeaderMap, Method};
    use parking_lot::RwLock;
    use serde_json::Value;
    use std::collections::HashMap;
    use std::sync::Arc;

    fn make_state() -> SharedStepFunctionsState {
        Arc::new(RwLock::new(
            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
        ))
    }

    fn make_request(action: &str, body: &str) -> AwsRequest {
        AwsRequest {
            service: "states".to_string(),
            action: action.to_string(),
            region: "us-east-1".to_string(),
            account_id: "123456789012".to_string(),
            request_id: "test-id".to_string(),
            headers: HeaderMap::new(),
            query_params: HashMap::new(),
            body: body.as_bytes().to_vec().into(),
            body_stream: parking_lot::Mutex::new(None),
            path_segments: vec![],
            raw_path: "/".to_string(),
            raw_query: String::new(),
            method: Method::POST,
            is_query_protocol: false,
            access_key_id: None,
            principal: None,
        }
    }

    fn body_json(resp: &AwsResponse) -> Value {
        serde_json::from_slice(resp.body.expect_bytes()).unwrap()
    }

    fn expect_err(result: Result<AwsResponse, AwsServiceError>) -> AwsServiceError {
        match result {
            Err(e) => e,
            Ok(_) => panic!("expected error, got Ok"),
        }
    }

    const VALID_DEF: &str = r#"{"StartAt":"Pass","States":{"Pass":{"Type":"Pass","End":true}}}"#;

    fn create_sm(svc: &StepFunctionsService, name: &str) -> String {
        let body = json!({
            "name": name,
            "definition": VALID_DEF,
            "roleArn": "arn:aws:iam::123456789012:role/test",
        });
        let req = make_request("CreateStateMachine", &body.to_string());
        let resp = svc.create_state_machine(&req).unwrap();
        let b = body_json(&resp);
        b["stateMachineArn"].as_str().unwrap().to_string()
    }

    #[test]
    fn same_state_machine_name_coexists_in_two_regions() {
        let svc = StepFunctionsService::new(make_state());
        let in_region = |action: &str, body: Value, region: &str| {
            let mut req = make_request(action, &body.to_string());
            req.region = region.to_string();
            req
        };
        let create = |region: &str| {
            let body = json!({
                "name": "orders",
                "definition": VALID_DEF,
                "roleArn": "arn:aws:iam::123456789012:role/test",
            });
            let resp = svc
                .create_state_machine(&in_region("CreateStateMachine", body, region))
                .unwrap();
            body_json(&resp)["stateMachineArn"]
                .as_str()
                .unwrap()
                .to_string()
        };
        let east = create("us-east-1");
        let west = create("eu-west-1");
        assert_eq!(
            east,
            "arn:aws:states:us-east-1:123456789012:stateMachine:orders"
        );
        assert_eq!(
            west,
            "arn:aws:states:eu-west-1:123456789012:stateMachine:orders"
        );

        // Each region lists only its own state machine.
        for (region, arn) in [("us-east-1", &east), ("eu-west-1", &west)] {
            let resp = svc
                .list_state_machines(&in_region("ListStateMachines", json!({}), region))
                .unwrap();
            let listed: Vec<String> = body_json(&resp)["stateMachines"]
                .as_array()
                .unwrap()
                .iter()
                .map(|m| m["stateMachineArn"].as_str().unwrap().to_string())
                .collect();
            assert_eq!(listed, vec![arn.clone()]);
        }

        // A state machine ARN of another region is not found here.
        let err = svc
            .describe_state_machine(&in_region(
                "DescribeStateMachine",
                json!({ "stateMachineArn": west }),
                "us-east-1",
            ))
            .err()
            .unwrap();
        assert_eq!(err.code(), "StateMachineDoesNotExist");

        // Deleting one region's state machine leaves the other's.
        svc.delete_state_machine(&in_region(
            "DeleteStateMachine",
            json!({ "stateMachineArn": west }),
            "eu-west-1",
        ))
        .unwrap();
        assert!(svc
            .describe_state_machine(&in_region(
                "DescribeStateMachine",
                json!({ "stateMachineArn": east }),
                "us-east-1",
            ))
            .is_ok());
    }

    #[test]
    fn v2_snapshot_migrates_resources_into_their_arn_region() {
        let mut legacy: fakecloud_core::multi_account::MultiAccountState<StepFunctionsState> =
            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", "");
        {
            let s = legacy.default_mut();
            s.activities.insert(
                "arn:aws:states:eu-west-1:123456789012:activity:work".into(),
                crate::state::Activity {
                    name: "work".into(),
                    arn: "arn:aws:states:eu-west-1:123456789012:activity:work".into(),
                    creation_date: Utc::now(),
                    tags: Default::default(),
                },
            );
        }
        let bytes = serde_json::to_vec(&json!({"schema_version": 2, "accounts": legacy})).unwrap();
        let snap = crate::state::parse_stepfunctions_snapshot(&bytes).unwrap();
        assert_eq!(snap.schema_version, STEPFUNCTIONS_SNAPSHOT_SCHEMA_VERSION);
        let accounts = snap.accounts.unwrap();
        let west = accounts.regional("123456789012", "eu-west-1").unwrap();
        assert_eq!(west.region, "eu-west-1");
        assert_eq!(west.activities.len(), 1);
        assert!(accounts.regional("123456789012", "us-east-1").is_none());
    }

    #[test]
    fn china_region_arns_use_the_aws_cn_partition() {
        let svc = StepFunctionsService::new(make_state());
        let in_china = |action: &str, body: Value| {
            let mut req = make_request(action, &body.to_string());
            req.region = "cn-north-1".to_string();
            req
        };
        let resp = svc
            .create_state_machine(&in_china(
                "CreateStateMachine",
                json!({
                    "name": "cn-sm",
                    "definition": VALID_DEF,
                    "roleArn": "arn:aws-cn:iam::123456789012:role/test",
                }),
            ))
            .unwrap();
        let sm_arn = body_json(&resp)["stateMachineArn"]
            .as_str()
            .unwrap()
            .to_string();
        assert_eq!(
            sm_arn,
            "arn:aws-cn:states:cn-north-1:123456789012:stateMachine:cn-sm"
        );
        let resp = svc
            .describe_state_machine(&in_china(
                "DescribeStateMachine",
                json!({"stateMachineArn": sm_arn}),
            ))
            .unwrap();
        assert_eq!(body_json(&resp)["name"], "cn-sm");

        let resp = svc
            .create_activity(&in_china("CreateActivity", json!({"name": "cn-act"})))
            .unwrap();
        let activity_arn = body_json(&resp)["activityArn"]
            .as_str()
            .unwrap()
            .to_string();
        assert_eq!(
            activity_arn,
            "arn:aws-cn:states:cn-north-1:123456789012:activity:cn-act"
        );
        let resp = svc
            .describe_activity(&in_china(
                "DescribeActivity",
                json!({"activityArn": activity_arn}),
            ))
            .unwrap();
        assert_eq!(body_json(&resp)["name"], "cn-act");
    }

    // ── CreateStateMachine ──

    #[test]
    fn create_state_machine_basic() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_sm(&svc, "test-sm");
        assert!(arn.contains("test-sm"));
    }

    // bug-hunt 2026-07-19: CreateStateMachine with publish=true creates a real
    // version that ListStateMachineVersions returns; without publish, no version
    // ARN is returned and no version exists.
    #[test]
    fn create_state_machine_publish_creates_listable_version() {
        let svc = StepFunctionsService::new(make_state());
        let body = json!({
            "name": "pub-sm",
            "definition": VALID_DEF,
            "roleArn": "arn:aws:iam::123456789012:role/r",
            "publish": true,
            "versionDescription": "first",
        });
        let req = make_request("CreateStateMachine", &body.to_string());
        let resp = svc.create_state_machine(&req).unwrap();
        let b = body_json(&resp);
        let arn = b["stateMachineArn"].as_str().unwrap().to_string();
        let version_arn = b["stateMachineVersionArn"].as_str().unwrap().to_string();
        assert_eq!(version_arn, format!("{arn}:1"));

        let list_req = make_request(
            "ListStateMachineVersions",
            &json!({ "stateMachineArn": arn }).to_string(),
        );
        let list = body_json(&svc.list_state_machine_versions(&list_req).unwrap());
        let versions = list["stateMachineVersions"].as_array().unwrap();
        assert_eq!(versions.len(), 1);
        assert_eq!(versions[0]["stateMachineVersionArn"], version_arn);
    }

    #[test]
    fn create_state_machine_no_publish_has_no_version() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_sm(&svc, "nopub-sm");
        // No version ARN is returned when publish is absent.
        let body = json!({
            "name": "nopub-sm2",
            "definition": VALID_DEF,
            "roleArn": "arn:aws:iam::123456789012:role/r",
        });
        let req = make_request("CreateStateMachine", &body.to_string());
        let b = body_json(&svc.create_state_machine(&req).unwrap());
        assert!(b.get("stateMachineVersionArn").is_none());

        let list_req = make_request(
            "ListStateMachineVersions",
            &json!({ "stateMachineArn": arn }).to_string(),
        );
        let list = body_json(&svc.list_state_machine_versions(&list_req).unwrap());
        assert!(list["stateMachineVersions"].as_array().unwrap().is_empty());
    }

    // bug-hunt 2026-07-19: UpdateStateMachine with publish=true publishes the
    // next version, and encryptionConfiguration persists + round-trips on
    // Describe.
    #[test]
    fn update_state_machine_publish_and_encryption_round_trip() {
        let svc = StepFunctionsService::new(make_state());
        let create = json!({
            "name": "enc-sm",
            "definition": VALID_DEF,
            "roleArn": "arn:aws:iam::123456789012:role/r",
            "encryptionConfiguration": { "type": "CUSTOMER_MANAGED_KMS_KEY", "kmsKeyId": "key-123" },
        });
        let req = make_request("CreateStateMachine", &create.to_string());
        let arn = body_json(&svc.create_state_machine(&req).unwrap())["stateMachineArn"]
            .as_str()
            .unwrap()
            .to_string();

        // encryptionConfiguration round-trips on Describe.
        let desc_req = make_request(
            "DescribeStateMachine",
            &json!({ "stateMachineArn": arn }).to_string(),
        );
        let desc = body_json(&svc.describe_state_machine(&desc_req).unwrap());
        assert_eq!(
            desc["encryptionConfiguration"]["type"],
            "CUSTOMER_MANAGED_KMS_KEY"
        );
        assert_eq!(desc["encryptionConfiguration"]["kmsKeyId"], "key-123");

        // Update with publish creates version 1.
        let upd = json!({
            "stateMachineArn": arn,
            "definition": VALID_DEF,
            "publish": true,
            "versionDescription": "v via update",
        });
        let upd_req = make_request("UpdateStateMachine", &upd.to_string());
        let upd_body = body_json(&svc.update_state_machine(&upd_req).unwrap());
        assert_eq!(
            upd_body["stateMachineVersionArn"].as_str().unwrap(),
            format!("{arn}:1")
        );

        let list_req = make_request(
            "ListStateMachineVersions",
            &json!({ "stateMachineArn": arn }).to_string(),
        );
        let list = body_json(&svc.list_state_machine_versions(&list_req).unwrap());
        assert_eq!(list["stateMachineVersions"].as_array().unwrap().len(), 1);
    }

    #[test]
    fn create_state_machine_with_express_type() {
        let svc = StepFunctionsService::new(make_state());
        let body = json!({
            "name": "express-sm",
            "definition": VALID_DEF,
            "roleArn": "arn:aws:iam::123456789012:role/r",
            "type": "EXPRESS",
        });
        let req = make_request("CreateStateMachine", &body.to_string());
        let resp = svc.create_state_machine(&req).unwrap();
        let b = body_json(&resp);
        assert!(b["stateMachineArn"].as_str().is_some());
    }

    #[test]
    fn create_state_machine_duplicate_fails() {
        let svc = StepFunctionsService::new(make_state());
        create_sm(&svc, "dup-sm");
        let body = json!({
            "name": "dup-sm",
            "definition": VALID_DEF,
            "roleArn": "arn:aws:iam::123456789012:role/r",
        });
        let req = make_request("CreateStateMachine", &body.to_string());
        let err = expect_err(svc.create_state_machine(&req));
        assert!(err.to_string().contains("StateMachineAlreadyExists"));
    }

    #[test]
    fn create_state_machine_missing_name() {
        let svc = StepFunctionsService::new(make_state());
        let body = json!({
            "definition": VALID_DEF,
            "roleArn": "arn:aws:iam::123456789012:role/r",
        });
        let req = make_request("CreateStateMachine", &body.to_string());
        assert!(svc.create_state_machine(&req).is_err());
    }

    #[test]
    fn create_state_machine_invalid_definition() {
        let svc = StepFunctionsService::new(make_state());
        let body = json!({
            "name": "bad-def",
            "definition": "not json",
            "roleArn": "arn:aws:iam::123456789012:role/r",
        });
        let req = make_request("CreateStateMachine", &body.to_string());
        let err = expect_err(svc.create_state_machine(&req));
        assert!(err.to_string().contains("InvalidDefinition"));
    }

    #[test]
    fn create_state_machine_definition_missing_start_at() {
        let svc = StepFunctionsService::new(make_state());
        let body = json!({
            "name": "no-start",
            "definition": r#"{"States":{"S":{"Type":"Pass","End":true}}}"#,
            "roleArn": "arn:aws:iam::123456789012:role/r",
        });
        let req = make_request("CreateStateMachine", &body.to_string());
        let err = expect_err(svc.create_state_machine(&req));
        assert!(err.to_string().contains("InvalidDefinition"));
    }

    #[test]
    fn create_state_machine_definition_missing_states() {
        let svc = StepFunctionsService::new(make_state());
        let body = json!({
            "name": "no-states",
            "definition": r#"{"StartAt":"S"}"#,
            "roleArn": "arn:aws:iam::123456789012:role/r",
        });
        let req = make_request("CreateStateMachine", &body.to_string());
        let err = expect_err(svc.create_state_machine(&req));
        assert!(err.to_string().contains("InvalidDefinition"));
    }

    #[test]
    fn create_state_machine_definition_start_at_not_in_states() {
        let svc = StepFunctionsService::new(make_state());
        let body = json!({
            "name": "bad-start",
            "definition": r#"{"StartAt":"Missing","States":{"S":{"Type":"Pass","End":true}}}"#,
            "roleArn": "arn:aws:iam::123456789012:role/r",
        });
        let req = make_request("CreateStateMachine", &body.to_string());
        let err = expect_err(svc.create_state_machine(&req));
        assert!(err.to_string().contains("MISSING_TRANSITION_TARGET"));
    }

    #[test]
    fn create_state_machine_invalid_type() {
        let svc = StepFunctionsService::new(make_state());
        let body = json!({
            "name": "bad-type",
            "definition": VALID_DEF,
            "roleArn": "arn:aws:iam::123456789012:role/r",
            "type": "INVALID",
        });
        let req = make_request("CreateStateMachine", &body.to_string());
        assert!(svc.create_state_machine(&req).is_err());
    }

    #[test]
    fn create_state_machine_invalid_arn() {
        let svc = StepFunctionsService::new(make_state());
        let body = json!({
            "name": "bad-arn",
            "definition": VALID_DEF,
            "roleArn": "not-an-arn",
        });
        let req = make_request("CreateStateMachine", &body.to_string());
        let err = expect_err(svc.create_state_machine(&req));
        assert!(err.to_string().contains("InvalidArn"));
    }

    #[test]
    fn create_state_machine_invalid_name() {
        let svc = StepFunctionsService::new(make_state());
        let body = json!({
            "name": "has spaces!",
            "definition": VALID_DEF,
            "roleArn": "arn:aws:iam::123456789012:role/r",
        });
        let req = make_request("CreateStateMachine", &body.to_string());
        let err = expect_err(svc.create_state_machine(&req));
        assert!(err.to_string().contains("InvalidName"));
    }

    #[test]
    fn create_state_machine_name_too_long() {
        let svc = StepFunctionsService::new(make_state());
        let long_name = "a".repeat(81);
        let body = json!({
            "name": long_name,
            "definition": VALID_DEF,
            "roleArn": "arn:aws:iam::123456789012:role/r",
        });
        let req = make_request("CreateStateMachine", &body.to_string());
        let err = expect_err(svc.create_state_machine(&req));
        assert!(err.to_string().contains("InvalidName"));
    }

    // ── DescribeStateMachine ──

    #[test]
    fn describe_state_machine_found() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_sm(&svc, "desc-sm");

        let req = make_request(
            "DescribeStateMachine",
            &json!({"stateMachineArn": arn}).to_string(),
        );
        let resp = svc.describe_state_machine(&req).unwrap();
        let b = body_json(&resp);
        assert_eq!(b["name"], "desc-sm");
        assert_eq!(b["status"], "ACTIVE");
        assert!(b["definition"].as_str().is_some());
    }

    #[test]
    fn describe_state_machine_not_found() {
        let svc = StepFunctionsService::new(make_state());
        let req = make_request(
            "DescribeStateMachine",
            &json!({"stateMachineArn": "arn:aws:states:us-east-1:123456789012:stateMachine:nope"})
                .to_string(),
        );
        let err = expect_err(svc.describe_state_machine(&req));
        assert!(err.to_string().contains("StateMachineDoesNotExist"));
    }

    // ── ListStateMachines ──

    #[test]
    fn list_state_machines_empty() {
        let svc = StepFunctionsService::new(make_state());
        let req = make_request("ListStateMachines", "{}");
        let resp = svc.list_state_machines(&req).unwrap();
        let b = body_json(&resp);
        assert!(b["stateMachines"].as_array().unwrap().is_empty());
    }

    #[test]
    fn list_state_machines_returns_created() {
        let svc = StepFunctionsService::new(make_state());
        create_sm(&svc, "sm-1");
        create_sm(&svc, "sm-2");

        let req = make_request("ListStateMachines", "{}");
        let resp = svc.list_state_machines(&req).unwrap();
        let b = body_json(&resp);
        assert_eq!(b["stateMachines"].as_array().unwrap().len(), 2);
    }

    // ── DeleteStateMachine ──

    #[test]
    fn delete_state_machine() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_sm(&svc, "del-sm");

        let req = make_request(
            "DeleteStateMachine",
            &json!({"stateMachineArn": arn}).to_string(),
        );
        svc.delete_state_machine(&req).unwrap();

        // Describe should fail
        let req = make_request(
            "DescribeStateMachine",
            &json!({"stateMachineArn": arn}).to_string(),
        );
        assert!(svc.describe_state_machine(&req).is_err());
    }

    #[test]
    fn delete_state_machine_nonexistent_succeeds() {
        let svc = StepFunctionsService::new(make_state());
        let req = make_request(
            "DeleteStateMachine",
            &json!({"stateMachineArn": "arn:aws:states:us-east-1:123456789012:stateMachine:nope"})
                .to_string(),
        );
        // AWS returns success even for nonexistent
        svc.delete_state_machine(&req).unwrap();
    }

    // ── UpdateStateMachine ──

    #[test]
    fn update_state_machine() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_sm(&svc, "upd-sm");

        let new_def = r#"{"StartAt":"NewPass","States":{"NewPass":{"Type":"Pass","End":true}}}"#;
        let body = json!({
            "stateMachineArn": arn,
            "definition": new_def,
            "description": "updated",
        });
        let req = make_request("UpdateStateMachine", &body.to_string());
        let resp = svc.update_state_machine(&req).unwrap();
        let b = body_json(&resp);
        assert!(b["updateDate"].as_f64().is_some());

        // Verify
        let req = make_request(
            "DescribeStateMachine",
            &json!({"stateMachineArn": arn}).to_string(),
        );
        let resp = svc.describe_state_machine(&req).unwrap();
        let b = body_json(&resp);
        assert!(b["definition"].as_str().unwrap().contains("NewPass"));
        assert_eq!(b["description"], "updated");
    }

    #[test]
    fn update_state_machine_not_found() {
        let svc = StepFunctionsService::new(make_state());
        let body = json!({
            "stateMachineArn": "arn:aws:states:us-east-1:123456789012:stateMachine:nope",
            "definition": VALID_DEF,
        });
        let req = make_request("UpdateStateMachine", &body.to_string());
        let err = expect_err(svc.update_state_machine(&req));
        assert!(err.to_string().contains("StateMachineDoesNotExist"));
    }

    // ── StartExecution ──

    #[tokio::test]
    async fn start_execution_basic() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_sm(&svc, "exec-sm");

        let body = json!({
            "stateMachineArn": arn,
            "input": r#"{"key":"value"}"#,
        });
        let req = make_request("StartExecution", &body.to_string());
        let resp = svc.start_execution(&req).unwrap();
        let b = body_json(&resp);
        assert!(b["executionArn"].as_str().is_some());
        assert!(b["startDate"].as_f64().is_some());
    }

    #[tokio::test]
    async fn start_execution_with_name() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_sm(&svc, "named-exec");

        let body = json!({
            "stateMachineArn": arn,
            "name": "my-execution",
        });
        let req = make_request("StartExecution", &body.to_string());
        let resp = svc.start_execution(&req).unwrap();
        let b = body_json(&resp);
        assert!(b["executionArn"].as_str().unwrap().contains("my-execution"));
    }

    #[tokio::test]
    async fn start_execution_sm_not_found() {
        let svc = StepFunctionsService::new(make_state());
        let body = json!({
            "stateMachineArn": "arn:aws:states:us-east-1:123456789012:stateMachine:nope",
        });
        let req = make_request("StartExecution", &body.to_string());
        let err = expect_err(svc.start_execution(&req));
        assert!(err.to_string().contains("StateMachineDoesNotExist"));
    }

    #[tokio::test]
    async fn start_execution_invalid_input() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_sm(&svc, "bad-input");

        let body = json!({
            "stateMachineArn": arn,
            "input": "not json",
        });
        let req = make_request("StartExecution", &body.to_string());
        let err = expect_err(svc.start_execution(&req));
        assert!(err.to_string().contains("InvalidExecutionInput"));
    }

    #[tokio::test]
    async fn start_execution_same_name_same_input_is_idempotent() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_sm(&svc, "dup-exec");

        let body = json!({
            "stateMachineArn": arn,
            "name": "same-name",
            "input": "{\"a\":1}",
        });
        let req = make_request("StartExecution", &body.to_string());
        let first = body_json(&svc.start_execution(&req).unwrap());

        // Same name AND same input -> 200 with the existing executionArn.
        let req = make_request("StartExecution", &body.to_string());
        let second = body_json(&svc.start_execution(&req).unwrap());
        assert_eq!(first["executionArn"], second["executionArn"]);
        assert_eq!(first["startDate"], second["startDate"]);
    }

    #[tokio::test]
    async fn start_execution_same_name_same_input_conflicts_once_closed() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_sm(&svc, "dup-exec-closed");
        let body = json!({
            "stateMachineArn": arn,
            "name": "same-name",
            "input": "{\"a\":1}",
        });
        let req = make_request("StartExecution", &body.to_string());
        let first = body_json(&svc.start_execution(&req).unwrap());
        let exec_arn = first["executionArn"].as_str().unwrap().to_string();
        svc.state
            .write()
            .by_arn_mut(&exec_arn)
            .and_then(|s| s.executions.get_mut(&exec_arn))
            .unwrap()
            .status = ExecutionStatus::Succeeded;

        // AWS: idempotent only while RUNNING; a closed execution conflicts.
        let req = make_request("StartExecution", &body.to_string());
        let err = expect_err(svc.start_execution(&req));
        assert!(err.to_string().contains("ExecutionAlreadyExists"));
        assert_eq!(err.status(), StatusCode::BAD_REQUEST);
    }

    fn parent_of(svc: &StepFunctionsService, exec_arn: &str) -> Option<String> {
        svc.state
            .read()
            .by_arn(exec_arn)
            .and_then(|s| s.executions.get(exec_arn))
            .and_then(|e| e.parent_execution_arn.clone())
    }

    const PARENT_EXEC: &str = "arn:aws:states:us-east-1:123456789012:execution:parent:p";

    #[tokio::test]
    async fn start_execution_records_the_execution_starting_it() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_sm(&svc, "child-sm");
        let req = make_request(
            "StartExecution",
            &json!({"stateMachineArn": arn}).to_string(),
        );

        let started = crate::state::STARTED_BY_EXECUTION
            .scope(PARENT_EXEC.to_string(), async { svc.start_execution(&req) })
            .await;
        let child = body_json(&started.unwrap())["executionArn"]
            .as_str()
            .unwrap()
            .to_string();
        assert_eq!(parent_of(&svc, &child).as_deref(), Some(PARENT_EXEC));

        // Started directly, an execution has no parent.
        let direct = body_json(&svc.start_execution(&req).unwrap())["executionArn"]
            .as_str()
            .unwrap()
            .to_string();
        assert_eq!(parent_of(&svc, &direct), None);
    }

    #[tokio::test]
    async fn idempotent_start_execution_never_reparents_the_running_execution() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_sm(&svc, "idem-child-sm");
        let req = make_request(
            "StartExecution",
            &json!({"stateMachineArn": arn, "name": "fixed", "input": "{}"}).to_string(),
        );
        let first = body_json(&svc.start_execution(&req).unwrap())["executionArn"]
            .as_str()
            .unwrap()
            .to_string();

        // A Task re-issuing the call gets the running execution back, but did
        // not start it.
        let again = crate::state::STARTED_BY_EXECUTION
            .scope(PARENT_EXEC.to_string(), async { svc.start_execution(&req) })
            .await;
        assert_eq!(body_json(&again.unwrap())["executionArn"], first.as_str());
        assert_eq!(parent_of(&svc, &first), None);
    }

    #[tokio::test]
    async fn start_execution_same_name_different_input_conflicts() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_sm(&svc, "dup-exec-diff");

        let req = make_request(
            "StartExecution",
            &json!({
                "stateMachineArn": arn,
                "name": "same-name",
                "input": "{\"a\":1}",
            })
            .to_string(),
        );
        svc.start_execution(&req).unwrap();

        // Same name, DIFFERENT input -> 400 ExecutionAlreadyExists.
        let req = make_request(
            "StartExecution",
            &json!({
                "stateMachineArn": arn,
                "name": "same-name",
                "input": "{\"a\":2}",
            })
            .to_string(),
        );
        let err = expect_err(svc.start_execution(&req));
        assert!(err.to_string().contains("ExecutionAlreadyExists"));
    }

    #[tokio::test]
    async fn start_execution_express_reuses_names_as_new_executions() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_express_sm(&svc, "dup-exec-express");

        let body = json!({
            "stateMachineArn": arn,
            "name": "same-name",
            "input": "{\"a\":1}",
        });
        // AWS: EXPRESS names can be reused immediately and StartExecution is
        // never idempotent, so an identical re-issue starts a second execution.
        let req = make_request("StartExecution", &body.to_string());
        let first = body_json(&svc.start_execution(&req).unwrap());
        let req = make_request("StartExecution", &body.to_string());
        let second = body_json(&svc.start_execution(&req).unwrap());

        let (a, b) = (
            first["executionArn"].as_str().unwrap(),
            second["executionArn"].as_str().unwrap(),
        );
        assert_ne!(a, b);
        for exec_arn in [a, b] {
            assert!(
                exec_arn.starts_with(
                    "arn:aws:states:us-east-1:123456789012:express:dup-exec-express:same-name:"
                ),
                "{exec_arn}"
            );
        }
        let accounts = svc.state.read();
        let state = accounts.by_arn(a).unwrap();
        assert_eq!(state.executions[a].name, "same-name");
        assert_eq!(state.executions[b].name, "same-name");
    }

    #[tokio::test]
    async fn delivered_express_executions_get_express_arns() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_express_sm(&svc, "delivered-express");
        start_execution_from_delivery(&svc.state, &None, &None, &None, &arn, "{}");
        start_execution_from_delivery(&svc.state, &None, &None, &None, &arn, "{}");

        let accounts = svc.state.read();
        let state = accounts.by_arn(&arn).unwrap();
        let arns: Vec<_> = state
            .executions
            .keys()
            .filter(|k| k.contains("delivered-express"))
            .collect();
        assert_eq!(arns.len(), 2);
        for exec_arn in arns {
            assert!(
                exec_arn.starts_with(
                    "arn:aws:states:us-east-1:123456789012:express:delivered-express:"
                ),
                "{exec_arn}"
            );
        }
    }

    #[tokio::test]
    async fn start_sync_execution_rejects_a_malformed_arn() {
        let svc = StepFunctionsService::new(make_state());
        let req = make_request(
            "StartSyncExecution",
            &json!({"stateMachineArn": "foo"}).to_string(),
        );
        let err = expect_err(svc.start_sync_execution(&req).await);
        assert!(err.to_string().contains("InvalidArn"), "{err}");
    }

    #[tokio::test]
    async fn start_sync_execution_uses_the_given_name() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_express_sm(&svc, "sync-named");
        let body = json!({"stateMachineArn": arn, "name": "my-run"});

        let req = make_request("StartSyncExecution", &body.to_string());
        let first = body_json(&svc.start_sync_execution(&req).await.unwrap());
        let req = make_request("StartSyncExecution", &body.to_string());
        let second = body_json(&svc.start_sync_execution(&req).await.unwrap());

        assert_eq!(first["name"], "my-run");
        assert_eq!(second["name"], "my-run");
        assert_ne!(first["executionArn"], second["executionArn"]);
        assert!(first["executionArn"]
            .as_str()
            .unwrap()
            .contains(":express:sync-named:my-run:"));

        let bad = json!({"stateMachineArn": arn, "name": "has space"});
        let req = make_request("StartSyncExecution", &bad.to_string());
        assert!(svc.start_sync_execution(&req).await.is_err());
    }

    // ── DescribeExecution ──

    #[tokio::test]
    async fn describe_execution_found() {
        let svc = StepFunctionsService::new(make_state());
        let sm_arn = create_sm(&svc, "desc-exec");

        let body = json!({"stateMachineArn": sm_arn, "name": "e1"});
        let req = make_request("StartExecution", &body.to_string());
        let resp = svc.start_execution(&req).unwrap();
        let exec_arn = body_json(&resp)["executionArn"]
            .as_str()
            .unwrap()
            .to_string();

        let req = make_request(
            "DescribeExecution",
            &json!({"executionArn": exec_arn}).to_string(),
        );
        let resp = svc.describe_execution(&req).unwrap();
        let b = body_json(&resp);
        assert_eq!(b["name"], "e1");
        assert_eq!(b["status"], "RUNNING");
    }

    #[tokio::test]
    async fn describe_execution_not_found() {
        let svc = StepFunctionsService::new(make_state());
        let req = make_request(
            "DescribeExecution",
            &json!({"executionArn": "arn:aws:states:us-east-1:123456789012:execution:sm:nope"})
                .to_string(),
        );
        let err = expect_err(svc.describe_execution(&req));
        assert!(err.to_string().contains("ExecutionDoesNotExist"));
    }

    // ── StopExecution ──

    #[tokio::test]
    async fn stop_execution() {
        let svc = StepFunctionsService::new(make_state());
        let sm_arn = create_sm(&svc, "stop-sm");

        let body = json!({"stateMachineArn": sm_arn, "name": "stop-e"});
        let req = make_request("StartExecution", &body.to_string());
        let resp = svc.start_execution(&req).unwrap();
        let exec_arn = body_json(&resp)["executionArn"]
            .as_str()
            .unwrap()
            .to_string();

        let body = json!({
            "executionArn": exec_arn,
            "error": "UserAborted",
            "cause": "test stop",
        });
        let req = make_request("StopExecution", &body.to_string());
        let resp = svc.stop_execution(&req).unwrap();
        let b = body_json(&resp);
        assert!(b["stopDate"].as_f64().is_some());

        // Verify aborted
        let req = make_request(
            "DescribeExecution",
            &json!({"executionArn": exec_arn}).to_string(),
        );
        let resp = svc.describe_execution(&req).unwrap();
        let b = body_json(&resp);
        assert_eq!(b["status"], "ABORTED");
        assert_eq!(b["error"], "UserAborted");
    }

    #[tokio::test]
    async fn redrive_execution_reruns_to_terminal() {
        let svc = StepFunctionsService::new(make_state());
        let sm_arn = create_sm(&svc, "redrive-sm");

        let body = json!({"stateMachineArn": sm_arn, "name": "redrive-e"});
        let req = make_request("StartExecution", &body.to_string());
        let resp = svc.start_execution(&req).unwrap();
        let exec_arn = body_json(&resp)["executionArn"]
            .as_str()
            .unwrap()
            .to_string();

        // Stop before the spawned interpreter runs -> terminal ABORTED.
        let req = make_request(
            "StopExecution",
            &json!({"executionArn": exec_arn, "error": "UserAborted", "cause": "stop"}).to_string(),
        );
        svc.stop_execution(&req).unwrap();

        // Redrive: must reset to RUNNING and re-spawn the interpreter.
        let req = make_request(
            "RedriveExecution",
            &json!({"executionArn": exec_arn}).to_string(),
        );
        let resp = svc.redrive_execution(&req).unwrap();
        assert!(body_json(&resp)["redriveDate"].as_i64().is_some());

        // The redriven Pass-state execution must reach a terminal state
        // (SUCCEEDED) instead of hanging RUNNING forever.
        let mut status = String::new();
        for _ in 0..100 {
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
            let req = make_request(
                "DescribeExecution",
                &json!({"executionArn": exec_arn}).to_string(),
            );
            let b = body_json(&svc.describe_execution(&req).unwrap());
            status = b["status"].as_str().unwrap().to_string();
            if status != "RUNNING" {
                break;
            }
        }
        assert_eq!(
            status, "SUCCEEDED",
            "redriven execution must run to completion"
        );
        let req = make_request(
            "DescribeExecution",
            &json!({"executionArn": exec_arn}).to_string(),
        );
        let b = body_json(&svc.describe_execution(&req).unwrap());
        assert_eq!(b["redriveCount"], 1);
    }

    #[tokio::test]
    async fn redrive_execution_running_rejected() {
        let svc = StepFunctionsService::new(make_state());
        let sm_arn = create_sm(&svc, "redrive-running-sm");
        let body = json!({"stateMachineArn": sm_arn, "name": "still-running"});
        let req = make_request("StartExecution", &body.to_string());
        let exec_arn = body_json(&svc.start_execution(&req).unwrap())["executionArn"]
            .as_str()
            .unwrap()
            .to_string();

        // A still-RUNNING execution is not redrivable (it already has a driver).
        let req = make_request(
            "RedriveExecution",
            &json!({"executionArn": exec_arn}).to_string(),
        );
        let err = expect_err(svc.redrive_execution(&req));
        assert!(err.to_string().contains("ExecutionNotRedrivable"));
    }

    #[tokio::test]
    async fn stop_execution_not_found() {
        let svc = StepFunctionsService::new(make_state());
        let req = make_request(
            "StopExecution",
            &json!({"executionArn": "arn:aws:states:us-east-1:123456789012:execution:sm:nope"})
                .to_string(),
        );
        let err = expect_err(svc.stop_execution(&req));
        assert!(err.to_string().contains("ExecutionDoesNotExist"));
    }

    // ── ListExecutions ──

    #[tokio::test]
    async fn list_executions() {
        let svc = StepFunctionsService::new(make_state());
        let sm_arn = create_sm(&svc, "list-exec");

        for i in 0..3 {
            let body = json!({"stateMachineArn": sm_arn, "name": format!("e{i}")});
            let req = make_request("StartExecution", &body.to_string());
            svc.start_execution(&req).unwrap();
        }

        let req = make_request(
            "ListExecutions",
            &json!({"stateMachineArn": sm_arn}).to_string(),
        );
        let resp = svc.list_executions(&req).unwrap();
        let b = body_json(&resp);
        assert_eq!(b["executions"].as_array().unwrap().len(), 3);
    }

    #[tokio::test]
    async fn list_executions_sm_not_found() {
        let svc = StepFunctionsService::new(make_state());
        let req = make_request(
            "ListExecutions",
            &json!({"stateMachineArn": "arn:aws:states:us-east-1:123456789012:stateMachine:nope"})
                .to_string(),
        );
        let err = expect_err(svc.list_executions(&req));
        assert!(err.to_string().contains("StateMachineDoesNotExist"));
    }

    // ── GetExecutionHistory ──

    #[tokio::test]
    async fn get_execution_history_not_found() {
        let svc = StepFunctionsService::new(make_state());
        let req = make_request(
            "GetExecutionHistory",
            &json!({"executionArn": "arn:aws:states:us-east-1:123456789012:execution:sm:nope"})
                .to_string(),
        );
        let err = expect_err(svc.get_execution_history(&req));
        assert!(err.to_string().contains("ExecutionDoesNotExist"));
    }

    // ── DescribeStateMachineForExecution ──

    #[tokio::test]
    async fn describe_sm_for_execution() {
        let svc = StepFunctionsService::new(make_state());
        let sm_arn = create_sm(&svc, "sm-for-exec");

        let body = json!({"stateMachineArn": sm_arn, "name": "e1"});
        let req = make_request("StartExecution", &body.to_string());
        let resp = svc.start_execution(&req).unwrap();
        let exec_arn = body_json(&resp)["executionArn"]
            .as_str()
            .unwrap()
            .to_string();

        let req = make_request(
            "DescribeStateMachineForExecution",
            &json!({"executionArn": exec_arn}).to_string(),
        );
        let resp = svc.describe_state_machine_for_execution(&req).unwrap();
        let b = body_json(&resp);
        assert_eq!(b["name"], "sm-for-exec");
    }

    // ── Tags ──

    #[test]
    fn tag_untag_list_tags() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_sm(&svc, "tagged-sm");

        // Tag
        let body = json!({
            "resourceArn": arn,
            "tags": [{"key": "env", "value": "prod"}],
        });
        let req = make_request("TagResource", &body.to_string());
        svc.tag_resource(&req).unwrap();

        // List
        let req = make_request(
            "ListTagsForResource",
            &json!({"resourceArn": arn}).to_string(),
        );
        let resp = svc.list_tags_for_resource(&req).unwrap();
        let b = body_json(&resp);
        let tags = b["tags"].as_array().unwrap();
        assert_eq!(tags.len(), 1);
        assert_eq!(tags[0]["key"], "env");

        // Untag
        let body = json!({
            "resourceArn": arn,
            "tagKeys": ["env"],
        });
        let req = make_request("UntagResource", &body.to_string());
        svc.untag_resource(&req).unwrap();

        // Verify empty
        let req = make_request(
            "ListTagsForResource",
            &json!({"resourceArn": arn}).to_string(),
        );
        let resp = svc.list_tags_for_resource(&req).unwrap();
        let b = body_json(&resp);
        assert!(b["tags"].as_array().unwrap().is_empty());
    }

    #[test]
    fn tag_resource_not_found() {
        let svc = StepFunctionsService::new(make_state());
        let body = json!({
            "resourceArn": "arn:aws:states:us-east-1:123456789012:stateMachine:nope",
            "tags": [{"key": "k", "value": "v"}],
        });
        let req = make_request("TagResource", &body.to_string());
        let err = expect_err(svc.tag_resource(&req));
        assert!(err.to_string().contains("ResourceNotFound"));
    }

    // ── Helper function tests ──

    #[test]
    fn test_validate_name() {
        assert!(validate_name("valid-name").is_ok());
        assert!(validate_name("under_score").is_ok());
        assert!(validate_name("").is_err());
        assert!(validate_name("has spaces").is_err());
        assert!(validate_name(&"a".repeat(81)).is_err());
    }

    #[test]
    fn test_validate_definition() {
        assert!(validate_definition(VALID_DEF).is_ok());
        assert!(validate_definition("not json").is_err());
        assert!(validate_definition(r#"{"States":{}}"#).is_err()); // missing StartAt
        assert!(validate_definition(r#"{"StartAt":"S"}"#).is_err()); // missing States
    }

    #[test]
    fn test_validate_definition_timeout_range() {
        let task = |field: &str, v: &str| {
            format!(
                r#"{{"StartAt":"T","States":{{"T":{{"Type":"Task","Resource":"arn:aws:states:::lambda:invoke","{field}":{v},"End":true}}}}}}"#
            )
        };
        for field in ["TimeoutSeconds", "HeartbeatSeconds"] {
            assert!(validate_definition(&task(field, "1")).is_ok());
            assert!(validate_definition(&task(field, "99999999")).is_ok());
            for bad in [
                "0",
                "-1",
                "100000000",
                "18446744073709551615",
                "1.5",
                "\"10\"",
            ] {
                let err = validate_definition(&task(field, bad)).unwrap_err();
                assert!(
                    err.to_string().contains("InvalidDefinition"),
                    "{field}={bad}: {err}"
                );
            }
        }
        // Top-level TimeoutSeconds.
        let def = r#"{"StartAt":"P","TimeoutSeconds":18446744073709551615,"States":{"P":{"Type":"Pass","End":true}}}"#;
        assert!(validate_definition(def).is_err());
        // Nested inside a Parallel branch.
        let def = r#"{"StartAt":"Par","States":{"Par":{"Type":"Parallel","End":true,"Branches":[{"StartAt":"T","States":{"T":{"Type":"Task","Resource":"arn:aws:states:::lambda:invoke","TimeoutSeconds":100000000,"End":true}}}]}}}"#;
        assert!(validate_definition(def).is_err());
    }

    #[test]
    fn test_validate_definition_rejects_malformed_paths() {
        // Unterminated bracket in InputPath.
        let def =
            r#"{"StartAt":"P","States":{"P":{"Type":"Pass","InputPath":"$.arr[","End":true}}}"#;
        assert!(validate_definition(def).is_err());

        // Multibyte char where the close bracket would be, in OutputPath.
        let def =
            "{\"StartAt\":\"P\",\"States\":{\"P\":{\"Type\":\"Pass\",\"OutputPath\":\"$.x[\u{00e9}\",\"End\":true}}}";
        assert!(validate_definition(def).is_err());

        // Malformed Choice Variable.
        let def = r#"{"StartAt":"C","States":{"C":{"Type":"Choice","Choices":[{"Variable":"$.n[","NumericEquals":1,"Next":"P"}]},"P":{"Type":"Pass","End":true}}}"#;
        assert!(validate_definition(def).is_err());

        // Path not rooted at $.
        let def =
            r#"{"StartAt":"P","States":{"P":{"Type":"Pass","InputPath":"foo.bar","End":true}}}"#;
        assert!(validate_definition(def).is_err());

        // Empty index brackets.
        let def =
            r#"{"StartAt":"P","States":{"P":{"Type":"Pass","ResultPath":"$.x[]","End":true}}}"#;
        assert!(validate_definition(def).is_err());
    }

    #[test]
    fn test_validate_definition_accepts_well_formed_paths() {
        // Valid reference paths, the literal "null" for Input/OutputPath, and
        // nested Choice combinators must all be accepted.
        let def = r#"{"StartAt":"P","States":{
            "P":{"Type":"Pass","InputPath":"$.a.b[0].c","OutputPath":"$","ResultPath":"$.out","Next":"C"},
            "C":{"Type":"Choice","Choices":[
                {"And":[{"Variable":"$.items[2]","NumericEquals":1},{"Variable":"$.flag","BooleanEquals":true}],"Next":"S"}
            ],"Default":"S"},
            "S":{"Type":"Succeed","InputPath":"null"}
        }}"#;
        assert!(validate_definition(def).is_ok());
    }

    // M5: malformed paths nested inside Parallel branches and Map processors
    // must be rejected.
    #[test]
    fn test_validate_definition_recurses_into_parallel_and_map() {
        let par = r#"{"StartAt":"Par","States":{"Par":{"Type":"Parallel","End":true,
            "Branches":[{"StartAt":"I","States":{"I":{"Type":"Pass","InputPath":"nope","End":true}}}]}}}"#;
        assert!(validate_definition(par).is_err());

        let map = r#"{"StartAt":"M","States":{"M":{"Type":"Map","End":true,
            "ItemProcessor":{"StartAt":"E","States":{"E":{"Type":"Pass","OutputPath":"nope","End":true}}}}}}"#;
        assert!(validate_definition(map).is_err());

        // Legacy inline Map uses `Iterator`.
        let iter = r#"{"StartAt":"M","States":{"M":{"Type":"Map","End":true,
            "Iterator":{"StartAt":"E","States":{"E":{"Type":"Pass","ResultPath":"$.x[]","End":true}}}}}}"#;
        assert!(validate_definition(iter).is_err());

        // A well-formed Parallel/Map still validates.
        let ok = r#"{"StartAt":"M","States":{"M":{"Type":"Map","ItemsPath":"$.items","End":true,
            "ItemProcessor":{"StartAt":"E","States":{"E":{"Type":"Pass","InputPath":"$.a","End":true}}}}}}"#;
        assert!(validate_definition(ok).is_ok());
    }

    // L7: `.$` payload-template keys must carry a JSONPath/intrinsic string.
    #[test]
    fn test_validate_definition_payload_template_dollar_keys() {
        // Bare literal value.
        let lit = r#"{"StartAt":"P","States":{"P":{"Type":"Pass",
            "Parameters":{"v.$":"just text"},"End":true}}}"#;
        assert!(validate_definition(lit).is_err());

        // Non-string value.
        let num = r#"{"StartAt":"P","States":{"P":{"Type":"Pass",
            "Parameters":{"v.$":42},"End":true}}}"#;
        assert!(validate_definition(num).is_err());

        // Valid: JSONPath reference, context-object ref, and intrinsic.
        let ok = r#"{"StartAt":"P","States":{"P":{"Type":"Pass","Parameters":{
            "a.$":"$.input","b.$":"$$.Execution.Id","c.$":"States.Format('{}',$.n)",
            "nested":{"d.$":"$.x"},"plain":"literal-ok"},"End":true}}}"#;
        assert!(validate_definition(ok).is_ok());

        // A bad `.$` nested inside an array under a plain key is also caught.
        let arr = r#"{"StartAt":"P","States":{"P":{"Type":"Pass",
            "ResultSelector":{"items":[{"bad.$":"literal"}]},"End":true}}}"#;
        assert!(validate_definition(arr).is_err());
    }

    #[test]
    fn test_paths_accept_full_jsonpath_but_result_path_must_be_definite() {
        let def = |field: &str, path: &str| {
            json!({
                "StartAt": "P",
                "States": {"P": {"Type": "Pass", field: path, "End": true}}
            })
            .to_string()
        };
        for p in [
            "$.items[*].id",
            "$.a[?(@.x == 1)]",
            "$['a']",
            "$.a[-1]",
            "$..id",
            "$.a[1:2]",
        ] {
            assert!(validate_definition(&def("InputPath", p)).is_ok(), "{p}");
            assert!(validate_definition(&def("OutputPath", p)).is_ok(), "{p}");
        }
        assert!(validate_definition(&def("ResultPath", "$['a'][0]")).is_ok());
        let choice = |var: &str| {
            json!({
                "StartAt": "C",
                "States": {
                    "C": {"Type": "Choice", "Choices": [{"Variable": var, "IsPresent": true, "Next": "D"}], "Default": "D"},
                    "D": {"Type": "Succeed"}
                }
            })
            .to_string()
        };
        assert!(validate_definition(&choice("$$.Execution.Name")).is_ok());
        assert!(validate_definition(&choice("$$.x[")).is_err());
        for p in ["$.items[*]", "$..a", "$.a[?(@.x)]"] {
            assert!(validate_definition(&def("ResultPath", p)).is_err(), "{p}");
        }
    }

    #[test]
    fn test_is_valid_reference_path() {
        assert!(is_valid_reference_path("$"));
        assert!(is_valid_reference_path("$.foo"));
        assert!(is_valid_reference_path("$.foo.bar[3].baz"));
        assert!(is_valid_reference_path("$[0]"));
        assert!(!is_valid_reference_path("$.arr["));
        assert!(!is_valid_reference_path("$.x[\u{00e9}"));
        assert!(!is_valid_reference_path("$.x[]"));
        assert!(!is_valid_reference_path("$.x[abc]"));
        assert!(!is_valid_reference_path("foo.bar"));
        assert!(!is_valid_reference_path(""));
    }

    #[test]
    fn test_validate_arn() {
        assert!(validate_arn("arn:aws:states:us-east-1:123:sm:test").is_ok());
        assert!(validate_arn("not-an-arn").is_err());
    }

    #[test]
    fn test_camel_to_details_key() {
        // All *StateEntered / *StateExited types collapse to one key each.
        assert_eq!(camel_to_details_key("PassStateEntered"), "stateEntered");
        assert_eq!(camel_to_details_key("TaskStateEntered"), "stateEntered");
        assert_eq!(camel_to_details_key("ChoiceStateExited"), "stateExited");
        assert_eq!(camel_to_details_key("MapStateExited"), "stateExited");
        // Other event types keep first-char-lowercased prefix.
        assert_eq!(camel_to_details_key("TaskScheduled"), "taskScheduled");
        assert_eq!(camel_to_details_key("ExecutionStarted"), "executionStarted");
        assert_eq!(camel_to_details_key(""), "");
    }

    #[test]
    fn test_is_mutating_action() {
        assert!(is_mutating_action("CreateStateMachine"));
        assert!(is_mutating_action("StartExecution"));
        assert!(!is_mutating_action("DescribeStateMachine"));
        assert!(!is_mutating_action("ListStateMachines"));
    }

    // ── StartSyncExecution ──

    fn create_express_sm(svc: &StepFunctionsService, name: &str) -> String {
        let body = json!({
            "name": name,
            "definition": VALID_DEF,
            "roleArn": "arn:aws:iam::123456789012:role/test",
            "type": "EXPRESS",
        });
        let req = make_request("CreateStateMachine", &body.to_string());
        let resp = svc.create_state_machine(&req).unwrap();
        let b = body_json(&resp);
        b["stateMachineArn"].as_str().unwrap().to_string()
    }

    #[tokio::test]
    async fn start_sync_execution_basic() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_express_sm(&svc, "sync-sm");

        let body = json!({
            "stateMachineArn": arn,
            "input": r#"{"key":"value"}"#,
        });
        let req = make_request("StartSyncExecution", &body.to_string());
        let resp = svc.start_sync_execution(&req).await.unwrap();
        let b = body_json(&resp);
        assert!(b["executionArn"]
            .as_str()
            .unwrap()
            .contains("express:sync-sm"));
        assert_eq!(b["stateMachineArn"], arn);
        assert_eq!(b["status"], "SUCCEEDED");
        assert!(b["startDate"].as_f64().is_some());
        assert!(b["stopDate"].as_f64().is_some());
        assert!(b["output"].as_str().is_some());
        assert!(b["billingDetails"]["billedDurationInMilliseconds"]
            .as_i64()
            .is_some());
    }

    #[tokio::test]
    async fn start_sync_execution_not_express() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_sm(&svc, "std-sm");

        let body = json!({"stateMachineArn": arn});
        let req = make_request("StartSyncExecution", &body.to_string());
        let err = expect_err(svc.start_sync_execution(&req).await);
        assert!(err.to_string().contains("StateMachineTypeNotSupported"));
    }

    #[tokio::test]
    async fn start_sync_execution_sm_not_found() {
        let svc = StepFunctionsService::new(make_state());
        let body = json!({
            "stateMachineArn": "arn:aws:states:us-east-1:123456789012:stateMachine:nope",
        });
        let req = make_request("StartSyncExecution", &body.to_string());
        let err = expect_err(svc.start_sync_execution(&req).await);
        assert!(err.to_string().contains("StateMachineDoesNotExist"));
    }

    #[tokio::test]
    async fn start_sync_execution_records_introspection_fields() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_express_sm(&svc, "sync-introspect");

        let body = json!({"stateMachineArn": arn, "input": "{}"});
        let req = make_request("StartSyncExecution", &body.to_string());
        let resp = svc.start_sync_execution(&req).await.unwrap();
        let b = body_json(&resp);
        let exec_arn = b["executionArn"].as_str().unwrap().to_string();

        let accounts = svc.state.read();
        let state = accounts.regional("123456789012", "us-east-1").unwrap();
        let stored = state
            .executions
            .get(&exec_arn)
            .expect("sync execution should be persisted for introspection");
        assert!(stored.is_sync, "sync executions must be marked is_sync");
        assert_eq!(stored.billed_memory_mb, Some(64));
        assert!(
            stored.billed_duration_ms.is_some(),
            "billed_duration_ms must be populated after sync run"
        );
        assert!(
            stored.parent_execution_arn.is_none(),
            "top-level sync execution has no parent"
        );
    }

    #[tokio::test]
    async fn start_sync_execution_invalid_input() {
        let svc = StepFunctionsService::new(make_state());
        let arn = create_express_sm(&svc, "bad-input-sync");

        let body = json!({
            "stateMachineArn": arn,
            "input": "not json",
        });
        let req = make_request("StartSyncExecution", &body.to_string());
        let err = expect_err(svc.start_sync_execution(&req).await);
        assert!(err.to_string().contains("InvalidExecutionInput"));
    }

    /// No snapshot store (memory mode) -> no persist hook for the CFN provisioner.
    #[test]
    fn snapshot_hook_is_none_without_store() {
        let svc = StepFunctionsService::new(make_state());
        assert!(svc.snapshot_hook().is_none());
    }

    /// With a store, the hook is present and invoking it runs the whole-state
    /// persist path the CloudFormation provisioner uses after mutating Step
    /// Functions state directly.
    #[tokio::test]
    async fn snapshot_hook_fires_with_store() {
        let store: Arc<dyn fakecloud_persistence::SnapshotStore> =
            Arc::new(fakecloud_persistence::MemorySnapshotStore::new());
        let svc = StepFunctionsService::new(make_state()).with_snapshot_store(store);
        let hook = svc
            .snapshot_hook()
            .expect("hook present when a store is set");
        hook().await;
    }

    fn make_execution(arn: &str, status: ExecutionStatus) -> Execution {
        Execution {
            execution_arn: arn.to_string(),
            state_machine_arn: "arn:aws:states:us-east-1:123456789012:stateMachine:sm".to_string(),
            state_machine_name: "sm".to_string(),
            name: arn.to_string(),
            status,
            input: None,
            output: None,
            start_date: Utc::now(),
            stop_date: None,
            error: None,
            cause: None,
            history_events: vec![],
            parent_execution_arn: None,
            is_sync: false,
            billed_duration_ms: None,
            billed_memory_mb: None,
            role_arn: String::new(),
            redrive_count: 0,
        }
    }

    #[test]
    fn reconcile_aborts_running_executions_on_restart() {
        // After a restart a RUNNING execution has no interpreter driving it, so
        // it must be aborted rather than left RUNNING forever (0.A2). A
        // completed execution is untouched.
        let state = make_state();
        {
            let mut accounts = state.write();
            let s = accounts.regional_mut("123456789012", "us-east-1");
            s.executions.insert(
                "running".into(),
                make_execution("running", ExecutionStatus::Running),
            );
            s.executions.insert(
                "done".into(),
                make_execution("done", ExecutionStatus::Succeeded),
            );
        }

        let n = reconcile_interrupted_executions(&state);
        assert_eq!(n, 1, "only the RUNNING execution is reconciled");

        let accounts = state.read();
        let s = accounts.regional("123456789012", "us-east-1").unwrap();
        let running = &s.executions["running"];
        assert_eq!(running.status, ExecutionStatus::Aborted);
        assert!(running.stop_date.is_some());
        assert_eq!(running.error.as_deref(), Some("Fakecloud.Restart"));
        assert_eq!(s.executions["done"].status, ExecutionStatus::Succeeded);
    }

    // ── Distributed Map -> MapRun population (bug-hunt read-shape) ──

    #[tokio::test]
    async fn distributed_map_populates_map_run() {
        let state = make_state();
        let svc = StepFunctionsService::new(state.clone());
        let execution_arn =
            "arn:aws:states:us-east-1:123456789012:execution:dist-sm:exec-1".to_string();

        let def = json!({
            "StartAt": "M",
            "States": {
                "M": {
                    "Type": "Map",
                    "ItemsPath": "$.items",
                    "ItemProcessor": {
                        "ProcessorConfig": { "Mode": "DISTRIBUTED", "ExecutionType": "STANDARD" },
                        "StartAt": "Item",
                        "States": { "Item": { "Type": "Pass", "End": true } }
                    },
                    "End": true
                }
            }
        });

        interpreter::execute_state_machine(
            state.clone(),
            execution_arn.clone(),
            def.to_string(),
            Some(r#"{"items":[1,2,3]}"#.to_string()),
            None,
            None,
            None,
            None,
        )
        .await;

        // ListMapRuns(executionArn) surfaces the newly created MapRun.
        let req = make_request(
            "ListMapRuns",
            &json!({ "executionArn": execution_arn }).to_string(),
        );
        let listed = body_json(&svc.list_map_runs(&req).unwrap());
        let runs = listed["mapRuns"].as_array().unwrap();
        assert_eq!(
            runs.len(),
            1,
            "distributed map must create exactly one MapRun"
        );
        let map_run_arn = runs[0]["mapRunArn"].as_str().unwrap().to_string();
        assert!(map_run_arn.contains(":mapRun:"));
        assert_eq!(runs[0]["executionArn"], execution_arn);

        // DescribeMapRun(mapRunArn) returns the record with real counts.
        let req = make_request(
            "DescribeMapRun",
            &json!({ "mapRunArn": map_run_arn }).to_string(),
        );
        let mr = body_json(&svc.describe_map_run(&req).unwrap());
        assert_eq!(mr["status"], "SUCCEEDED");
        assert_eq!(mr["itemCounts"]["total"], 3);
        assert_eq!(mr["itemCounts"]["succeeded"], 3);
        assert_eq!(mr["itemCounts"]["failed"], 0);
        assert_eq!(mr["executionCounts"]["succeeded"], 3);
    }

    #[tokio::test]
    async fn inline_map_does_not_create_map_run() {
        let state = make_state();
        let svc = StepFunctionsService::new(state.clone());
        let execution_arn =
            "arn:aws:states:us-east-1:123456789012:execution:inline-sm:exec-1".to_string();

        // No ProcessorConfig.Mode => inline Map: AWS creates no MapRun.
        let def = json!({
            "StartAt": "M",
            "States": {
                "M": {
                    "Type": "Map",
                    "ItemsPath": "$.items",
                    "ItemProcessor": {
                        "StartAt": "Item",
                        "States": { "Item": { "Type": "Pass", "End": true } }
                    },
                    "End": true
                }
            }
        });

        interpreter::execute_state_machine(
            state.clone(),
            execution_arn.clone(),
            def.to_string(),
            Some(r#"{"items":[1,2]}"#.to_string()),
            None,
            None,
            None,
            None,
        )
        .await;

        let req = make_request(
            "ListMapRuns",
            &json!({ "executionArn": execution_arn }).to_string(),
        );
        let listed = body_json(&svc.list_map_runs(&req).unwrap());
        assert!(listed["mapRuns"].as_array().unwrap().is_empty());
    }
}

#[cfg(test)]
mod pagination_reject_test {
    #[test]
    fn paginate_checked_rejects_invalid_token() {
        use fakecloud_core::pagination::paginate_checked;
        let items: Vec<i32> = (0..5).collect();
        assert!(paginate_checked(&items, Some("bad"), 3).is_err());
        assert!(paginate_checked(&items, Some("2"), 3).is_ok());
    }
}