pharos_native/runtime/

dynamic.rs

use std::cmp::Ordering;
use std::collections::BTreeMap;
use std::env;
use std::fs;
use std::path::Path;
use std::time::{SystemTime, UNIX_EPOCH};

use serde_json::{Map, Value};
use sha2::{Digest, Sha256};

use crate::config::AppRuntime;

use super::connection::NativeTransportType;
use super::http::HttpRequest;
use super::json_commands;

/// Public constant `DYNAMIC_HOST_PROFILE` exposed by this module.
///
/// # Examples
///
/// ```rust,ignore
/// let _ = pharos_native::runtime::dynamic::DYNAMIC_HOST_PROFILE;
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for the surrounding module surface.
pub const DYNAMIC_HOST_PROFILE: &str = "dynamic-app";

/// Performs dynamic runtime host profile.
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::dynamic_runtime_host_profile`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::dynamic_runtime_host_profile();
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn dynamic_runtime_host_profile() -> &'static str {
    DYNAMIC_HOST_PROFILE
}

/// Performs request has session cookie.
///
/// # Parameters
///
/// - `cookie_header`: Input consumed by [`crate::runtime::dynamic::request_has_session_cookie`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::request_has_session_cookie`] holds and `false` otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::request_has_session_cookie("session=abc");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn request_has_session_cookie(cookie_header: &str) -> bool {
    pharos_runtime_security::request_has_session_cookie(cookie_header)
}

/// Performs mut JSON obj get string.
///
/// # Parameters
///
/// - `object`: Input consumed by [`crate::runtime::dynamic::mut_json_obj_get_string`].
/// - `key`: Input consumed by [`crate::runtime::dynamic::mut_json_obj_get_string`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::mut_json_obj_get_string`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::mut_json_obj_get_string(todo!(), "key");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn mut_json_obj_get_string(object: &Value, key: &str) -> Option<String> {
    object
        .as_object()
        .and_then(|map| map.get(key))
        .and_then(Value::as_str)
        .map(ToOwned::to_owned)
}

/// Performs generated surface label.
///
/// # Parameters
///
/// - `field_name`: Input consumed by [`crate::runtime::dynamic::generated_surface_label`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::generated_surface_label`] can produce a value and [`None`] otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::generated_surface_label`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::generated_surface_label(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn generated_surface_label(field_name: Option<&str>) -> Option<String> {
    let field_name = field_name.unwrap_or("value");
    let mut next_word = true;
    Some(
        field_name
            .chars()
            .map(|ch| {
                let mut out = if ch == '_' { ' ' } else { ch };
                if next_word && out.is_ascii_lowercase() {
                    out = out.to_ascii_uppercase();
                }
                next_word = out == ' ';
                out
            })
            .collect(),
    )
}

#[derive(Debug, Default, Clone, Eq, PartialEq)]
/// Public struct `RuntimeJsonValueList` used by this module.
///
/// # Fields
///
/// - `items`: Stored field on `RuntimeJsonValueList`.
///
/// # Examples
///
/// ```rust,ignore
/// let _ = std::mem::size_of::<pharos_native::runtime::dynamic::RuntimeJsonValueList>();
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for constructors, helpers, and neighboring runtime types.
pub struct RuntimeJsonValueList {
    pub items: Vec<Value>,
}

/// Performs runtime JSON value list free.
///
/// # Parameters
///
/// - `list`: Input consumed by [`crate::runtime::dynamic::runtime_json_value_list_free`].
///
/// # Returns
///
/// Returns after [`crate::runtime::dynamic::runtime_json_value_list_free`] applies its in-place effect.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::runtime_json_value_list_free(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn runtime_json_value_list_free(list: &mut RuntimeJsonValueList) {
    list.items.clear();
}

/// Performs runtime JSON value list append.
///
/// # Parameters
///
/// - `list`: Input consumed by [`crate::runtime::dynamic::runtime_json_value_list_append`].
/// - `item`: Input consumed by [`crate::runtime::dynamic::runtime_json_value_list_append`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::runtime_json_value_list_append`] holds and `false` otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::runtime_json_value_list_append(todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn runtime_json_value_list_append(list: &mut RuntimeJsonValueList, item: &Value) -> bool {
    list.items.push(item.clone());
    true
}

/// Performs runtime text buffer append format.
///
/// # Parameters
///
/// - `buffer`: Input consumed by [`crate::runtime::dynamic::runtime_text_buffer_append_format`].
/// - `formatted_text`: Input consumed by [`crate::runtime::dynamic::runtime_text_buffer_append_format`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::runtime_text_buffer_append_format`] holds and `false` otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::runtime_text_buffer_append_format(&mut [], todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn runtime_text_buffer_append_format(buffer: &mut String, formatted_text: &str) -> bool {
    buffer.push_str(formatted_text);
    true
}

/// Performs HTML escape.
fn html_escape(input: &str) -> String {
    input
        .replace('&', "&amp;")
        .replace('<', "&lt;")
        .replace('>', "&gt;")
        .replace('"', "&quot;")
}

/// Performs runtime text buffer append HTML escaped.
///
/// # Parameters
///
/// - `buffer`: Input consumed by [`crate::runtime::dynamic::runtime_text_buffer_append_html_escaped`].
/// - `text`: Input consumed by [`crate::runtime::dynamic::runtime_text_buffer_append_html_escaped`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::runtime_text_buffer_append_html_escaped`] holds and `false` otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::runtime_text_buffer_append_html_escaped(&mut [], "{}");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn runtime_text_buffer_append_html_escaped(buffer: &mut String, text: Option<&str>) -> bool {
    buffer.push_str(&html_escape(text.unwrap_or("")));
    true
}

/// Normalizes a public origin and rejects control characters or unsupported schemes.
///
/// # Parameters
///
/// - `configured_origin`: Input consumed by [`crate::runtime::dynamic::normalized_public_origin`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::normalized_public_origin`] can produce a value and [`None`] otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::normalized_public_origin`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::normalized_public_origin(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn normalized_public_origin(configured_origin: &str) -> Option<String> {
    pharos_runtime_dynamic::normalized_public_origin(configured_origin)
}

/// Returns the scheme trusted for one request transport.
///
/// # Parameters
///
/// - `request`: Input consumed by [`crate::runtime::dynamic::request_origin_scheme`].
/// - `transport_type`: Input consumed by [`crate::runtime::dynamic::request_origin_scheme`].
/// - `direct_client_ip`: Input consumed by [`crate::runtime::dynamic::request_origin_scheme`].
/// - `trusted_proxy_cidrs`: Input consumed by [`crate::runtime::dynamic::request_origin_scheme`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::request_origin_scheme`].
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::request_origin_scheme`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::request_origin_scheme(todo!(), todo!(), todo!(), "127.0.0.1/32");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn request_origin_scheme(
    request: Option<&HttpRequest>,
    transport_type: NativeTransportType,
    direct_client_ip: Option<&str>,
    trusted_proxy_cidrs: Option<&str>,
) -> &'static str {
    let Some(request) = request else {
        return "http";
    };
    let forwarded_https = request.headers.lines().any(|line| {
        let (name, value) = line.split_once(':').unwrap_or(("", ""));
        name.trim().eq_ignore_ascii_case("x-forwarded-proto")
            && value.trim().eq_ignore_ascii_case("https")
    });
    let origin_https = request.headers.lines().any(|line| {
        let (name, value) = line.split_once(':').unwrap_or(("", ""));
        name.trim().eq_ignore_ascii_case("origin") && value.trim_start().starts_with("https://")
    });
    let trusted_forward = match transport_type {
        NativeTransportType::Unix => forwarded_https,
        NativeTransportType::Tcp => {
            forwarded_https
                && direct_client_ip.is_some()
                && trusted_proxy_cidrs.is_some_and(|value| !value.trim().is_empty())
        }
        NativeTransportType::Unknown => false,
    };
    if trusted_forward || origin_https {
        "https"
    } else {
        "http"
    }
}

/// Builds the app-relative verification route after validating the token.
///
/// # Parameters
///
/// - `app`: Input consumed by [`crate::runtime::dynamic::verification_token_path`].
/// - `token`: Input consumed by [`crate::runtime::dynamic::verification_token_path`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::verification_token_path`] can produce a value and [`None`] otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::verification_token_path`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::verification_token_path(todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn verification_token_path(app: &AppRuntime, token: &str) -> Option<String> {
    if token.is_empty()
        || token.len() > 512
        || token.bytes().any(|byte| !byte.is_ascii_alphanumeric())
    {
        return None;
    }
    let base = app
        .base_path
        .as_deref()
        .filter(|value| !value.is_empty() && *value != "/")
        .unwrap_or("");
    Some(format!(
        "{base}/accounts/verify-email/{token}/?app_id={}",
        app.app_id.as_deref().unwrap_or_default()
    ))
}

/// Builds a verification URL from configured origin or a validated Host header.
///
/// # Parameters
///
/// - `app`: Input consumed by [`crate::runtime::dynamic::verification_url`].
/// - `request`: Input consumed by [`crate::runtime::dynamic::verification_url`].
/// - `token`: Input consumed by [`crate::runtime::dynamic::verification_url`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::verification_url`] can produce a value and [`None`] otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::verification_url`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::verification_url(todo!(), todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn verification_url(app: &AppRuntime, request: &HttpRequest, token: &str) -> Option<String> {
    let path = verification_token_path(app, token)?;
    let origin = app
        .runtime_conf_path
        .as_deref()
        .and_then(|path| {
            json_commands::dynamic_config_value(
                Path::new(app.app_root.as_deref()?),
                Some(Path::new(path)),
                "base_url",
            )
        })
        .and_then(|value| normalized_public_origin(&value));
    let origin = origin.or_else(|| {
        let host = request.headers.lines().find_map(|line| {
            let (name, value) = line.split_once(':')?;
            name.trim()
                .eq_ignore_ascii_case("host")
                .then_some(value.trim())
        })?;
        let scheme = request_origin_scheme(Some(request), NativeTransportType::Tcp, None, None);
        normalized_public_origin(&format!("{scheme}://{host}"))
    })?;
    Some(format!("{origin}{path}"))
}

/// Extracts one form value from the request body with multipart support delegated to the bounded parser.
///
/// # Parameters
///
/// - `request`: Input consumed by [`crate::runtime::dynamic::request_form_value`].
/// - `field`: Input consumed by [`crate::runtime::dynamic::request_form_value`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::request_form_value`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::request_form_value(todo!(), "field");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn request_form_value(request: &HttpRequest, field: &str) -> Option<String> {
    if field.is_empty() {
        return None;
    }
    let body = String::from_utf8_lossy(&request.body);
    if request.body_file_path.is_none()
        && request
            .content_type
            .as_deref()
            .is_some_and(|value| value.starts_with("multipart/form-data"))
    {
        return request.content_type.as_deref().and_then(|content_type| {
            json_commands::request_multipart_field_value(content_type, &request.body, field)
        });
    }
    json_commands::request_form_field_value(
        request.content_type.as_deref(),
        Some(body.as_ref()),
        request.body_file_path.as_deref().map(Path::new),
        field,
    )
}

/// Extracts one decoded query parameter from the request target.
///
/// # Parameters
///
/// - `request`: Input consumed by [`crate::runtime::dynamic::request_query_value`].
/// - `field`: Input consumed by [`crate::runtime::dynamic::request_query_value`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::request_query_value`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::request_query_value(todo!(), "field");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn request_query_value(request: &HttpRequest, field: &str) -> Option<String> {
    request
        .target
        .split_once('?')
        .and_then(|(_, query)| json_commands::request_query_value(query, field))
}

/// Resolves one bounded derived input from the request envelope.
fn dynamic_derived_input_value(spec: &Value, request: &HttpRequest) -> Option<String> {
    let source = spec
        .get("source")
        .and_then(Value::as_str)
        .unwrap_or_default();
    match source {
        "literal" => spec
            .get("value")
            .map(value_to_text)
            .filter(|value| !value.is_empty()),
        "header" => spec
            .get("header")
            .and_then(Value::as_str)
            .and_then(|header| request_header_value(request, header))
            .filter(|value| !value.is_empty()),
        "query" => spec
            .get("param")
            .or_else(|| spec.get("query_param"))
            .and_then(Value::as_str)
            .and_then(|field| request_query_value(request, field))
            .filter(|value| !value.is_empty()),
        "form" => spec
            .get("from_field")
            .or_else(|| spec.get("input_field"))
            .and_then(Value::as_str)
            .and_then(|field| request_form_value(request, field))
            .filter(|value| !value.is_empty()),
        "context_method" => (!request.method.is_empty()).then(|| request.method.clone()),
        "context_path" => {
            let path = request_path(Some(request));
            (!path.is_empty()).then_some(path)
        }
        _ => None,
    }
}

/// Performs dynamic apply default input hooks.
fn dynamic_apply_default_input_hooks(
    overrides: &mut BTreeMap<String, Option<String>>,
    action: &Value,
    request: &HttpRequest,
) -> Result<(), i32> {
    let Some(rows) = action
        .get("hook_contract")
        .and_then(|value| value.get("default_inputs"))
        .and_then(Value::as_array)
    else {
        return Ok(());
    };
    if rows.len() > 32 {
        return Err(422);
    }
    for row in rows {
        let field = row
            .get("field")
            .and_then(Value::as_str)
            .filter(|value| !value.is_empty() && value.len() <= 128)
            .ok_or(422)?;
        if dynamic_action_input_value(request, overrides, field)
            .is_some_and(|value| !value.is_empty())
        {
            continue;
        }
        overrides.insert(field.to_owned(), dynamic_derived_input_value(row, request));
    }
    Ok(())
}

/// Returns one stable action-owned input override map after applying declared normalization hooks.
fn dynamic_action_input_overrides(
    action: &Value,
    request: &HttpRequest,
) -> Result<BTreeMap<String, Option<String>>, i32> {
    let mut overrides = BTreeMap::new();
    let Some(hook_contract) = action.get("hook_contract").and_then(Value::as_object) else {
        return Ok(overrides);
    };
    if let Some(rows) = hook_contract
        .get("derived_inputs")
        .and_then(Value::as_array)
    {
        if rows.len() > 32 {
            return Err(422);
        }
        for row in rows {
            let field = row
                .get("field")
                .and_then(Value::as_str)
                .filter(|value| !value.is_empty() && value.len() <= 128)
                .ok_or(422)?;
            if row
                .get("preserve_existing")
                .and_then(Value::as_bool)
                .unwrap_or(false)
                && dynamic_action_input_value(request, &overrides, field)
                    .is_some_and(|value| !value.is_empty())
            {
                continue;
            }
            overrides.insert(field.to_owned(), dynamic_derived_input_value(row, request));
        }
    }
    dynamic_apply_default_input_hooks(&mut overrides, action, request)?;
    let Some(hooks) = hook_contract
        .get("normalize_inputs")
        .and_then(Value::as_array)
    else {
        return Ok(overrides);
    };
    for hook in hooks {
        let kind = hook.get("kind").and_then(Value::as_str).ok_or(422)?;
        let fields = hook.get("fields").and_then(Value::as_array).ok_or(422)?;
        for field in fields {
            let field = field.as_str().ok_or(422)?;
            if field.is_empty() || field.len() > 128 {
                return Err(422);
            }
            let current = overrides
                .get(field)
                .cloned()
                .unwrap_or_else(|| request_form_value(request, field));
            let next = match kind {
                "trim" => current.map(|value| value.trim().to_owned()),
                "collapse_whitespace" => current.map(|value| {
                    value
                        .split_whitespace()
                        .filter(|segment| !segment.is_empty())
                        .collect::<Vec<_>>()
                        .join(" ")
                }),
                "lowercase" => current.map(|value| value.to_ascii_lowercase()),
                "uppercase" => current.map(|value| value.to_ascii_uppercase()),
                "empty_to_null" => {
                    current.and_then(|value| (!value.trim().is_empty()).then_some(value))
                }
                _ => return Err(422),
            };
            overrides.insert(field.to_owned(), next);
        }
    }
    Ok(overrides)
}

/// Resolves one request input value after applying any declared action normalization hooks.
fn dynamic_action_input_value(
    request: &HttpRequest,
    overrides: &BTreeMap<String, Option<String>>,
    field: &str,
) -> Option<String> {
    if field.is_empty() {
        return None;
    }
    overrides
        .get(field)
        .cloned()
        .unwrap_or_else(|| request_form_value(request, field))
}

/// Builds one bounded action-report annotation map from declared success hooks.
fn dynamic_action_report_annotations(
    action: &Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    action_id: &str,
    entity_id: &str,
    event_kind: &str,
    anchor_record_id: i64,
) -> Result<Map<String, Value>, i32> {
    dynamic_action_annotations(
        action,
        "report_annotations",
        request,
        input_overrides,
        action_id,
        entity_id,
        event_kind,
        anchor_record_id,
    )
}

/// Builds one bounded action-report annotation map from a declared hook key.
fn dynamic_action_annotations(
    action: &Value,
    hook_key: &str,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    action_id: &str,
    entity_id: &str,
    event_kind: &str,
    anchor_record_id: i64,
) -> Result<Map<String, Value>, i32> {
    let mut annotations = Map::new();
    let Some(rows) = action
        .get("hook_contract")
        .and_then(|value| value.get(hook_key))
        .and_then(Value::as_array)
    else {
        return Ok(annotations);
    };
    if rows.len() > 16 {
        return Err(422);
    }
    for row in rows {
        let field = row
            .get("field")
            .and_then(Value::as_str)
            .filter(|value| !value.is_empty() && value.len() <= 128)
            .ok_or(422)?;
        let value = dynamic_notification_parameter_value(
            row.get("value").unwrap_or(&Value::Null),
            request,
            input_overrides,
            action_id,
            entity_id,
            event_kind,
            "",
            "",
            anchor_record_id,
        )
        .unwrap_or(Value::Null);
        annotations.insert(field.to_owned(), value);
    }
    Ok(annotations)
}

/// Builds one bounded action-report annotation map from declared failure hooks.
fn dynamic_action_failure_annotations(
    action: &Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    action_id: &str,
    entity_id: &str,
    event_kind: &str,
    anchor_record_id: i64,
) -> Result<Map<String, Value>, i32> {
    dynamic_action_annotations(
        action,
        "failure_annotations",
        request,
        input_overrides,
        action_id,
        entity_id,
        event_kind,
        anchor_record_id,
    )
}

/// Performs dynamic replace records from input list primitive.
fn dynamic_replace_records_from_input_list_primitive(
    root: &mut Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    primitive: &Value,
    matched_record_count: &mut i64,
    created_record_count: &mut i64,
    updated_record_count: &mut i64,
    skipped_record_count: &mut i64,
    removed_record_count: &mut i64,
) -> Result<(), i32> {
    let collection = primitive
        .get("collection")
        .and_then(Value::as_str)
        .ok_or(422)?;
    let scope_field = primitive
        .get("scope_match")
        .and_then(Value::as_object)
        .and_then(|map| map.get("field"))
        .and_then(Value::as_str)
        .ok_or(422)?;
    let scope_input = primitive
        .get("scope_match")
        .and_then(Value::as_object)
        .and_then(|map| map.get("from_input"))
        .and_then(Value::as_str)
        .ok_or(422)?;
    let wanted =
        dynamic_action_input_value(request, input_overrides, scope_input).unwrap_or_default();
    let delimiter = primitive
        .get("list_source")
        .and_then(Value::as_object)
        .and_then(|map| map.get("delimiter"))
        .and_then(Value::as_str)
        .unwrap_or(",");
    let source_field = primitive
        .get("list_source")
        .and_then(Value::as_object)
        .and_then(|map| map.get("from_input"))
        .and_then(Value::as_str)
        .ok_or(422)?;
    let raw_input =
        dynamic_action_input_value(request, input_overrides, source_field).unwrap_or_default();
    let parsed_items = raw_input
        .split(delimiter)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .collect::<Vec<_>>();
    let max_records = primitive
        .get("max_records")
        .and_then(Value::as_i64)
        .unwrap_or(parsed_items.len() as i64)
        .max(0) as usize;
    let overflow_policy = primitive
        .get("overflow_policy")
        .and_then(Value::as_str)
        .unwrap_or("truncate");
    if parsed_items.len() > max_records && overflow_policy == "fail" {
        return Err(422);
    }
    let items = parsed_items
        .into_iter()
        .take(max_records)
        .collect::<Vec<_>>();
    *skipped_record_count += raw_input
        .split(delimiter)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .count()
        .saturating_sub(items.len()) as i64;
    let order_field = primitive
        .get("order_field")
        .and_then(Value::as_str)
        .unwrap_or("id");
    let updates = primitive
        .get("updates")
        .and_then(Value::as_object)
        .ok_or(422)?;
    let record_contract = primitive
        .get("record_contract")
        .and_then(Value::as_object)
        .ok_or(422)?;
    let existing_match_count = root
        .get(collection)
        .and_then(Value::as_array)
        .map(|rows| {
            rows.iter()
                .filter(|row| row.get(scope_field).map(value_to_text).unwrap_or_default() == wanted)
                .count()
        })
        .unwrap_or(0);
    let create_count = items.len().saturating_sub(existing_match_count);
    let mut reserved_ids = (0..create_count)
        .map(|_| dynamic_take_next_global_id(root))
        .collect::<Vec<_>>();
    reserved_ids.reverse();
    let Some(rows) = dynamic_state_rows_mut(root, collection) else {
        return Err(500);
    };
    let mut matching_indexes = rows
        .iter()
        .enumerate()
        .filter(|(_, row)| row.get(scope_field).map(value_to_text).unwrap_or_default() == wanted)
        .map(|(index, _)| index)
        .collect::<Vec<_>>();
    matching_indexes.sort_by(|left, right| {
        dynamic_value_order_cmp(rows[*left].get(order_field), rows[*right].get(order_field))
    });
    *matched_record_count += matching_indexes.len() as i64;
    for (position, item) in items.iter().enumerate() {
        if let Some(index) = matching_indexes.get(position).copied() {
            let Some(object) = rows.get_mut(index).and_then(Value::as_object_mut) else {
                return Err(500);
            };
            for (field_name, update_spec) in updates {
                let source = update_spec
                    .get("source")
                    .and_then(Value::as_str)
                    .unwrap_or_default();
                let value = match source {
                    "input" => Value::String(
                        dynamic_action_input_value(
                            request,
                            input_overrides,
                            update_spec
                                .get("field")
                                .and_then(Value::as_str)
                                .unwrap_or_default(),
                        )
                        .unwrap_or_default(),
                    ),
                    "literal" => update_spec.get("value").cloned().unwrap_or(Value::Null),
                    "list_item" => Value::String((*item).to_owned()),
                    "list_index" => Value::from((position + 1) as i64),
                    _ => Value::Null,
                };
                object.insert(field_name.to_owned(), value);
            }
            *updated_record_count += 1;
        } else {
            let mut record = Map::new();
            let mut reserved_next_id = None;
            for (field_name, field_spec) in record_contract {
                let source = field_spec
                    .get("source")
                    .and_then(Value::as_str)
                    .unwrap_or_default();
                let value = match source {
                    "next_id" => {
                        let next_id = *reserved_next_id
                            .get_or_insert_with(|| reserved_ids.pop().unwrap_or(0));
                        Value::from(next_id)
                    }
                    "input" => Value::String(
                        dynamic_action_input_value(
                            request,
                            input_overrides,
                            field_spec
                                .get("field")
                                .and_then(Value::as_str)
                                .unwrap_or_default(),
                        )
                        .unwrap_or_default(),
                    ),
                    "literal" => field_spec.get("value").cloned().unwrap_or(Value::Null),
                    "context_anchor_created_id" => Value::from(0),
                    "list_item" => Value::String((*item).to_owned()),
                    "list_index" => Value::from((position + 1) as i64),
                    _ => Value::Null,
                };
                record.insert(field_name.to_owned(), value);
            }
            rows.push(Value::Object(record));
            *created_record_count += 1;
        }
    }
    let mut remove_indexes = matching_indexes
        .into_iter()
        .skip(items.len())
        .collect::<Vec<_>>();
    remove_indexes.sort_unstable_by(|left, right| right.cmp(left));
    for index in remove_indexes {
        rows.remove(index);
        *removed_record_count += 1;
    }
    Ok(())
}

/// Performs dynamic append record primitive.
fn dynamic_append_record_primitive(
    root: &mut Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    primitive: &Value,
    anchor_record_id: &mut i64,
    created_record_count: &mut i64,
) -> Result<(), i32> {
    let collection = primitive
        .get("collection")
        .and_then(Value::as_str)
        .ok_or(422)?;
    let mut record = Map::new();
    let record_contract = primitive
        .get("record_contract")
        .and_then(Value::as_object)
        .ok_or(422)?;
    for (field_name, field_spec) in record_contract {
        let source = field_spec
            .get("source")
            .and_then(Value::as_str)
            .unwrap_or_default();
        let value = match source {
            "next_id" => Value::from(dynamic_take_next_global_id(root)),
            "input" => Value::String(
                dynamic_action_input_value(
                    request,
                    input_overrides,
                    field_spec
                        .get("field")
                        .and_then(Value::as_str)
                        .unwrap_or_default(),
                )
                .unwrap_or_default(),
            ),
            "literal" => field_spec.get("value").cloned().unwrap_or(Value::Null),
            "context_anchor_created_id" => Value::from(*anchor_record_id),
            _ => Value::Null,
        };
        record.insert(field_name.to_owned(), value);
    }
    let created_id = record.get("id").and_then(Value::as_i64).unwrap_or(0);
    dynamic_state_rows_mut(root, collection)
        .ok_or(500)?
        .push(Value::Object(record));
    *created_record_count += 1;
    if *anchor_record_id <= 0 {
        *anchor_record_id = created_id;
    }
    Ok(())
}

/// Performs dynamic set root fields primitive.
fn dynamic_set_root_fields_primitive(
    app: &AppRuntime,
    root: &mut Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    primitive: &Value,
) -> Result<(), i32> {
    let ref_text = primitive.get("fields").and_then(Value::as_str).ok_or(422)?;
    let fields_root = dynamic_resolve_ref_target(app, ref_text).ok_or(422)?;
    let fields = fields_root
        .as_object()
        .or_else(|| fields_root.get("fields").and_then(Value::as_object))
        .ok_or(422)?;
    let Some(root_object) = root.as_object_mut() else {
        return Err(500);
    };
    for (field_name, field_spec) in fields {
        let source = field_spec
            .get("source")
            .and_then(Value::as_str)
            .unwrap_or_default();
        let value = match source {
            "literal" => field_spec.get("value").cloned().unwrap_or(Value::Null),
            "input" => Value::String(
                dynamic_action_input_value(
                    request,
                    input_overrides,
                    field_spec
                        .get("field")
                        .and_then(Value::as_str)
                        .unwrap_or_default(),
                )
                .unwrap_or_default(),
            ),
            _ => Value::Null,
        };
        root_object.insert(field_name.to_owned(), value);
    }
    Ok(())
}

/// Performs dynamic update record primitive.
fn dynamic_update_record_primitive(
    root: &mut Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    primitive: &Value,
    matched_record_count: &mut i64,
    updated_record_count: &mut i64,
) -> Result<(), i32> {
    let collection = primitive
        .get("collection")
        .and_then(Value::as_str)
        .ok_or(422)?;
    let match_field = primitive
        .get("match")
        .and_then(Value::as_object)
        .and_then(|map| map.get("field"))
        .and_then(Value::as_str)
        .ok_or(422)?;
    let match_input = primitive
        .get("match")
        .and_then(Value::as_object)
        .and_then(|map| map.get("from_input"))
        .and_then(Value::as_str)
        .ok_or(422)?;
    let wanted =
        dynamic_action_input_value(request, input_overrides, match_input).unwrap_or_default();
    let updates = primitive
        .get("updates")
        .and_then(Value::as_object)
        .ok_or(422)?;
    let Some(row) = dynamic_state_rows_mut(root, collection).and_then(|rows| {
        rows.iter_mut()
            .find(|row| row.get(match_field).map(value_to_text).unwrap_or_default() == wanted)
    }) else {
        return Err(404);
    };
    let Some(object) = row.as_object_mut() else {
        return Err(500);
    };
    for (field_name, update_spec) in updates {
        let source = update_spec
            .get("source")
            .and_then(Value::as_str)
            .unwrap_or_default();
        let value = match source {
            "literal" => update_spec.get("value").cloned().unwrap_or(Value::Null),
            "input" => Value::String(
                dynamic_action_input_value(
                    request,
                    input_overrides,
                    update_spec
                        .get("field")
                        .and_then(Value::as_str)
                        .unwrap_or_default(),
                )
                .unwrap_or_default(),
            ),
            "toggle_boolean" => Value::Bool(
                !object
                    .get(
                        update_spec
                            .get("field")
                            .and_then(Value::as_str)
                            .unwrap_or_default(),
                    )
                    .and_then(Value::as_bool)
                    .unwrap_or(false),
            ),
            _ => Value::Null,
        };
        object.insert(field_name.to_owned(), value);
    }
    *updated_record_count += 1;
    *matched_record_count += 1;
    Ok(())
}

/// Performs dynamic append records from input list primitive.
fn dynamic_append_records_from_input_list_primitive(
    root: &mut Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    primitive: &Value,
    anchor_record_id: i64,
    matched_record_count: i64,
    updated_record_count: i64,
    skipped_record_count: &mut i64,
    created_record_count: &mut i64,
) -> Result<(), i32> {
    let collection = primitive
        .get("collection")
        .and_then(Value::as_str)
        .ok_or(422)?;
    let delimiter = primitive
        .get("list_source")
        .and_then(Value::as_object)
        .and_then(|map| map.get("delimiter"))
        .and_then(Value::as_str)
        .unwrap_or(",");
    let source_field = primitive
        .get("list_source")
        .and_then(Value::as_object)
        .and_then(|map| map.get("from_input"))
        .and_then(Value::as_str)
        .ok_or(422)?;
    let input_value =
        dynamic_action_input_value(request, input_overrides, source_field).unwrap_or_default();
    let items = input_value
        .split(delimiter)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .collect::<Vec<_>>();
    let max_records = primitive
        .get("max_records")
        .and_then(Value::as_i64)
        .unwrap_or(items.len() as i64)
        .max(0) as usize;
    let overflow_policy = primitive
        .get("overflow_policy")
        .and_then(Value::as_str)
        .unwrap_or("truncate");
    if items.len() > max_records && overflow_policy == "fail" {
        return Err(422);
    }
    *skipped_record_count += items.len().saturating_sub(max_records) as i64;
    let record_contract = primitive
        .get("record_contract")
        .and_then(Value::as_object)
        .ok_or(422)?;
    for (position, item) in items.into_iter().take(max_records).enumerate() {
        let mut record = Map::new();
        for (field_name, field_spec) in record_contract {
            let source = field_spec
                .get("source")
                .and_then(Value::as_str)
                .unwrap_or_default();
            let value = match source {
                "next_id"
                | "input"
                | "literal"
                | "context_anchor_created_id"
                | "list_item"
                | "list_index" => dynamic_record_contract_value(
                    root,
                    request,
                    input_overrides,
                    field_spec,
                    anchor_record_id,
                    Some(item),
                    Some(position),
                    matched_record_count,
                    updated_record_count,
                    *skipped_record_count,
                ),
                _ => Value::Null,
            };
            record.insert(field_name.to_owned(), value);
        }
        dynamic_state_rows_mut(root, collection)
            .ok_or(500)?
            .push(Value::Object(record));
        *created_record_count += 1;
    }
    Ok(())
}

/// Performs dynamic append records from related collection primitive.
fn dynamic_append_records_from_related_collection_primitive(
    root: &mut Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    primitive: &Value,
    matched_record_count: &mut i64,
    skipped_record_count: &mut i64,
    created_record_count: &mut i64,
) -> Result<(), i32> {
    let collection = primitive
        .get("collection")
        .and_then(Value::as_str)
        .ok_or(422)?;
    let source_collection = primitive
        .get("source_collection")
        .and_then(Value::as_str)
        .ok_or(422)?;
    let scope_field = primitive
        .get("scope_match")
        .and_then(Value::as_object)
        .and_then(|map| map.get("field"))
        .and_then(Value::as_str)
        .ok_or(422)?;
    let scope_input = primitive
        .get("scope_match")
        .and_then(Value::as_object)
        .and_then(|map| map.get("from_input"))
        .and_then(Value::as_str)
        .ok_or(422)?;
    let wanted =
        dynamic_action_input_value(request, input_overrides, scope_input).unwrap_or_default();
    let source_rows = root
        .get(source_collection)
        .and_then(Value::as_array)
        .cloned()
        .unwrap_or_default();
    let matched = source_rows
        .into_iter()
        .filter(|row| row.get(scope_field).map(value_to_text).unwrap_or_default() == wanted)
        .collect::<Vec<_>>();
    let max_records = primitive
        .get("max_records")
        .and_then(Value::as_i64)
        .unwrap_or(matched.len() as i64)
        .max(0) as usize;
    *matched_record_count += matched.len() as i64;
    *skipped_record_count += matched.len().saturating_sub(max_records) as i64;
    let record_contract = primitive
        .get("record_contract")
        .and_then(Value::as_object)
        .ok_or(422)?;
    for source_row in matched.into_iter().take(max_records) {
        let mut record = Map::new();
        for (field_name, field_spec) in record_contract {
            let source = field_spec
                .get("source")
                .and_then(Value::as_str)
                .unwrap_or_default();
            let value = match source {
                "next_id" => Value::from(dynamic_take_next_global_id(root)),
                "source_record_field" => source_row
                    .get(
                        field_spec
                            .get("field")
                            .and_then(Value::as_str)
                            .unwrap_or_default(),
                    )
                    .cloned()
                    .unwrap_or(Value::Null),
                "input" => Value::String(
                    dynamic_action_input_value(
                        request,
                        input_overrides,
                        field_spec
                            .get("field")
                            .and_then(Value::as_str)
                            .unwrap_or_default(),
                    )
                    .unwrap_or_default(),
                ),
                "literal" => field_spec.get("value").cloned().unwrap_or(Value::Null),
                _ => Value::Null,
            };
            record.insert(field_name.to_owned(), value);
        }
        dynamic_state_rows_mut(root, collection)
            .ok_or(500)?
            .push(Value::Object(record));
        *created_record_count += 1;
    }
    Ok(())
}

/// Performs dynamic apply scoped assignment or bulk update primitive.
fn dynamic_apply_scoped_assignment_or_bulk_update_primitive(
    root: &mut Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    primitive: &Value,
    primitive_kind: &str,
    matched_record_count: &mut i64,
    updated_record_count: &mut i64,
    skipped_record_count: &mut i64,
) -> Result<(), i32> {
    let collection = primitive
        .get("collection")
        .and_then(Value::as_str)
        .ok_or(422)?;
    let scope_field = primitive
        .get("scope_match")
        .and_then(Value::as_object)
        .and_then(|map| map.get("field"))
        .and_then(Value::as_str)
        .ok_or(422)?;
    let scope_input = primitive
        .get("scope_match")
        .and_then(Value::as_object)
        .and_then(|map| map.get("from_input"))
        .and_then(Value::as_str)
        .ok_or(422)?;
    let wanted =
        dynamic_action_input_value(request, input_overrides, scope_input).unwrap_or_default();
    let max_records = primitive
        .get("max_records")
        .and_then(Value::as_i64)
        .unwrap_or(i64::MAX)
        .max(0) as usize;
    let overflow_policy = primitive
        .get("overflow_policy")
        .and_then(Value::as_str)
        .unwrap_or("truncate");
    let Some(rows) = dynamic_state_rows_mut(root, collection) else {
        return Err(500);
    };
    let matching_indexes = rows
        .iter()
        .enumerate()
        .filter(|(_, row)| row.get(scope_field).map(value_to_text).unwrap_or_default() == wanted)
        .map(|(index, _)| index)
        .collect::<Vec<_>>();
    *matched_record_count += matching_indexes.len() as i64;
    if matching_indexes.len() > max_records && overflow_policy == "fail" {
        return Err(422);
    }
    *skipped_record_count += matching_indexes.len().saturating_sub(max_records) as i64;
    for index in matching_indexes.into_iter().take(max_records) {
        let Some(object) = rows.get_mut(index).and_then(Value::as_object_mut) else {
            return Err(500);
        };
        if primitive_kind == "scoped_reassignment" {
            let target_field = primitive
                .get("target_field")
                .and_then(Value::as_str)
                .ok_or(422)?;
            let value = dynamic_action_input_value(
                request,
                input_overrides,
                primitive
                    .get("value_contract")
                    .and_then(Value::as_object)
                    .and_then(|map| map.get("field"))
                    .and_then(Value::as_str)
                    .unwrap_or_default(),
            )
            .unwrap_or_default();
            object.insert(target_field.to_owned(), Value::String(value));
        } else {
            let updates = primitive
                .get("updates")
                .and_then(Value::as_object)
                .ok_or(422)?;
            for (field_name, update_spec) in updates {
                let source = update_spec
                    .get("source")
                    .and_then(Value::as_str)
                    .unwrap_or_default();
                let value = match source {
                    "literal" => update_spec.get("value").cloned().unwrap_or(Value::Null),
                    _ => Value::Null,
                };
                object.insert(field_name.to_owned(), value);
            }
        }
        *updated_record_count += 1;
    }
    Ok(())
}

/// Performs dynamic apply scoped status transition primitive.
fn dynamic_apply_scoped_status_transition_primitive(
    root: &mut Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    primitive: &Value,
    matched_record_count: &mut i64,
    updated_record_count: &mut i64,
    skipped_record_count: &mut i64,
) -> Result<(), i32> {
    let collection = primitive
        .get("collection")
        .and_then(Value::as_str)
        .ok_or(422)?;
    let scope_field = primitive
        .get("scope_match")
        .and_then(Value::as_object)
        .and_then(|map| map.get("field"))
        .and_then(Value::as_str)
        .ok_or(422)?;
    let scope_input = primitive
        .get("scope_match")
        .and_then(Value::as_object)
        .and_then(|map| map.get("from_input"))
        .and_then(Value::as_str)
        .ok_or(422)?;
    let wanted =
        dynamic_action_input_value(request, input_overrides, scope_input).unwrap_or_default();
    let status_field = primitive
        .get("status_field")
        .and_then(Value::as_str)
        .filter(|value| !value.is_empty())
        .unwrap_or("status");
    let workflow_state_field = primitive
        .get("workflow_state_field")
        .and_then(Value::as_str)
        .filter(|value| !value.is_empty())
        .unwrap_or("workflow_state");
    let to_status = primitive
        .get("to_status")
        .and_then(Value::as_str)
        .filter(|value| !value.is_empty())
        .ok_or(422)?;
    let to_workflow_state = primitive
        .get("to_workflow_state")
        .and_then(Value::as_str)
        .filter(|value| !value.is_empty())
        .unwrap_or(to_status);
    let allowed_from_statuses = primitive
        .get("from_statuses")
        .and_then(Value::as_array)
        .map(|rows| {
            rows.iter()
                .filter_map(Value::as_str)
                .filter(|value| !value.is_empty())
                .collect::<Vec<_>>()
        })
        .unwrap_or_default();
    let max_records = primitive
        .get("max_records")
        .and_then(Value::as_i64)
        .unwrap_or(i64::MAX)
        .max(0) as usize;
    let overflow_policy = primitive
        .get("overflow_policy")
        .and_then(Value::as_str)
        .unwrap_or("truncate");
    let now = now_iso_utc();
    let actor_value = primitive
        .get("actor_input")
        .and_then(Value::as_str)
        .and_then(|field| dynamic_action_input_value(request, input_overrides, field));
    let Some(rows) = dynamic_state_rows_mut(root, collection) else {
        return Err(500);
    };
    let matching_indexes = rows
        .iter()
        .enumerate()
        .filter(|(_, row)| row.get(scope_field).map(value_to_text).unwrap_or_default() == wanted)
        .map(|(index, _)| index)
        .collect::<Vec<_>>();
    *matched_record_count += matching_indexes.len() as i64;
    if matching_indexes.len() > max_records && overflow_policy == "fail" {
        return Err(422);
    }
    *skipped_record_count += matching_indexes.len().saturating_sub(max_records) as i64;
    for index in matching_indexes.into_iter().take(max_records) {
        let Some(object) = rows.get_mut(index).and_then(Value::as_object_mut) else {
            return Err(500);
        };
        let current_status = object
            .get(status_field)
            .and_then(Value::as_str)
            .or_else(|| object.get(workflow_state_field).and_then(Value::as_str))
            .unwrap_or_default();
        if !allowed_from_statuses.is_empty() && !allowed_from_statuses.contains(&current_status) {
            *skipped_record_count += 1;
            continue;
        }
        object.insert(status_field.to_owned(), Value::String(to_status.to_owned()));
        object.insert(
            workflow_state_field.to_owned(),
            Value::String(to_workflow_state.to_owned()),
        );
        if let Some(field) = primitive
            .get("timestamp_field")
            .and_then(Value::as_str)
            .filter(|value| !value.is_empty())
        {
            object.insert(field.to_owned(), Value::String(now.clone()));
        }
        if let Some(field) = primitive
            .get("actor_field")
            .and_then(Value::as_str)
            .filter(|value| !value.is_empty())
        {
            object.insert(
                field.to_owned(),
                Value::String(actor_value.clone().unwrap_or_default()),
            );
        }
        if let Some(updates) = primitive.get("updates").and_then(Value::as_object) {
            for (field_name, update_spec) in updates {
                let source = update_spec
                    .get("source")
                    .and_then(Value::as_str)
                    .unwrap_or_default();
                let value = match source {
                    "literal" => update_spec.get("value").cloned().unwrap_or(Value::Null),
                    "input" => Value::String(
                        dynamic_action_input_value(
                            request,
                            input_overrides,
                            update_spec
                                .get("field")
                                .and_then(Value::as_str)
                                .unwrap_or_default(),
                        )
                        .unwrap_or_default(),
                    ),
                    _ => Value::Null,
                };
                object.insert(field_name.to_owned(), value);
            }
        }
        *updated_record_count += 1;
    }
    Ok(())
}

/// Performs dynamic upsert record primitive.
fn dynamic_upsert_record_primitive(
    root: &mut Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    primitive: &Value,
    anchor_record_id: i64,
    matched_record_count: &mut i64,
    updated_record_count: &mut i64,
    skipped_record_count: i64,
    created_record_count: &mut i64,
) -> Result<(), i32> {
    let collection = primitive
        .get("collection")
        .and_then(Value::as_str)
        .ok_or(422)?;
    let match_field = primitive
        .get("match")
        .and_then(Value::as_object)
        .and_then(|map| map.get("field"))
        .and_then(Value::as_str)
        .ok_or(422)?;
    let wanted = if let Some(input_name) = primitive
        .get("match")
        .and_then(Value::as_object)
        .and_then(|map| map.get("from_input"))
        .and_then(Value::as_str)
    {
        dynamic_action_input_value(request, input_overrides, input_name).unwrap_or_default()
    } else if let Some(value) = primitive
        .get("match")
        .and_then(Value::as_object)
        .and_then(|map| map.get("value"))
    {
        value_to_text(value)
    } else {
        return Err(422);
    };
    let updates = primitive
        .get("updates")
        .and_then(Value::as_object)
        .ok_or(422)?;
    let record_contract = primitive
        .get("record_contract")
        .and_then(Value::as_object)
        .ok_or(422)?;
    let existing_index = root
        .get(collection)
        .and_then(Value::as_array)
        .and_then(|rows| {
            rows.iter().position(|row| {
                row.get(match_field).map(value_to_text).unwrap_or_default() == wanted
            })
        });
    if let Some(index) = existing_index {
        *matched_record_count += 1;
        let resolved_updates = updates
            .iter()
            .map(|(field_name, update_spec)| {
                (
                    field_name.to_owned(),
                    dynamic_record_contract_value(
                        root,
                        request,
                        input_overrides,
                        update_spec,
                        anchor_record_id,
                        None,
                        None,
                        *matched_record_count,
                        *updated_record_count,
                        skipped_record_count,
                    ),
                )
            })
            .collect::<Vec<_>>();
        let Some(object) = dynamic_state_rows_mut(root, collection)
            .and_then(|rows| rows.get_mut(index))
            .and_then(Value::as_object_mut)
        else {
            return Err(500);
        };
        for (field_name, value) in resolved_updates {
            object.insert(field_name.to_owned(), value);
        }
        *updated_record_count += 1;
    } else {
        let mut record = Map::new();
        for (field_name, field_spec) in record_contract {
            let value = dynamic_record_contract_value(
                root,
                request,
                input_overrides,
                field_spec,
                anchor_record_id,
                None,
                None,
                *matched_record_count,
                *updated_record_count,
                skipped_record_count,
            );
            record.insert(field_name.to_owned(), value);
        }
        dynamic_state_rows_mut(root, collection)
            .ok_or(500)?
            .push(Value::Object(record));
        *created_record_count += 1;
    }
    Ok(())
}

/// Resolves the best available resource identity for one declarative action report or effect.
fn dynamic_action_resource_id(
    action: &Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    anchor_record_id: i64,
) -> String {
    let entity_id = action
        .get("entity_id")
        .and_then(Value::as_str)
        .unwrap_or("record");
    if anchor_record_id > 0 {
        return format!("{entity_id}:{anchor_record_id}");
    }
    if let Some(input_name) = action
        .get("mutation_primitives")
        .and_then(Value::as_array)
        .and_then(|rows| {
            rows.iter().find_map(|primitive| {
                primitive
                    .get("match")
                    .and_then(Value::as_object)
                    .and_then(|map| map.get("from_input"))
                    .and_then(Value::as_str)
                    .or_else(|| {
                        primitive
                            .get("scope_match")
                            .and_then(Value::as_object)
                            .and_then(|map| map.get("from_input"))
                            .and_then(Value::as_str)
                    })
            })
        })
        .filter(|value| !value.is_empty())
        && let Some(value) = dynamic_action_input_value(request, input_overrides, input_name)
            .filter(|value| !value.is_empty())
    {
        return format!("{entity_id}:{value}");
    }
    if let Some(value) =
        dynamic_action_input_value(request, input_overrides, "id").filter(|value| !value.is_empty())
    {
        return format!("{entity_id}:{value}");
    }
    format!("{entity_id}:0")
}

/// Prepares an app-declared file effect without touching the filesystem.
fn prepare_file_record(
    integrations_root: &Value,
    action: &Value,
    file_record: &Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    action_id: &str,
    entity_id: &str,
    event_kind: &str,
    anchor_record_id: i64,
) -> Option<(String, String, Value)> {
    let adapter_id = file_record
        .get("adapter_id")
        .and_then(Value::as_str)
        .filter(|value| !value.is_empty())
        .unwrap_or_default();
    let adapter = dynamic_find_integration_adapter(integrations_root, adapter_id)?;
    let adapter_kind = adapter
        .get("adapter_kind")
        .and_then(Value::as_str)
        .unwrap_or_default();
    if !dynamic_integration_file_adapter_kind(adapter_kind) {
        return None;
    }
    let relative_path = adapter
        .get("relative_path")
        .and_then(Value::as_str)
        .filter(|value| !value.is_empty())
        .unwrap_or_default();
    if relative_path.is_empty()
        || Path::new(relative_path)
            .components()
            .any(|part| !matches!(part, std::path::Component::Normal(_)))
    {
        return None;
    }
    let mut record = serde_json::Map::from_iter([
        (
            "adapter_id".to_owned(),
            Value::String(adapter_id.to_owned()),
        ),
        (
            "adapter_kind".to_owned(),
            Value::String(adapter_kind.to_owned()),
        ),
        ("action_id".to_owned(), Value::String(action_id.to_owned())),
        ("entity_id".to_owned(), Value::String(entity_id.to_owned())),
        (
            "event_kind".to_owned(),
            Value::String(event_kind.to_owned()),
        ),
        (
            "resource_id".to_owned(),
            Value::String(dynamic_action_resource_id(
                action,
                request,
                input_overrides,
                anchor_record_id,
            )),
        ),
        ("created_at".to_owned(), Value::String(now_iso_utc())),
    ]);
    if let Some(fields) = file_record.get("record").and_then(Value::as_object) {
        if fields.len() > 32 {
            return None;
        }
        for (field_name, field_spec) in fields {
            if field_name.is_empty() || field_name.len() > 128 {
                return None;
            }
            let value = dynamic_notification_parameter_value(
                field_spec,
                request,
                input_overrides,
                action_id,
                entity_id,
                event_kind,
                "",
                "",
                anchor_record_id,
            )
            .unwrap_or(Value::Null);
            record.insert(field_name.to_owned(), value);
        }
    }
    Some((
        adapter_kind.to_owned(),
        relative_path.to_owned(),
        Value::Object(record),
    ))
}

/// Queues the local effect in the same document commit as its domain mutation.
fn enqueue_file_record(root: &mut Value, prepared: (String, String, Value)) -> bool {
    let (kind, path, record) = prepared;
    let persisted = match root.get("_local_file_delivery_sequence") {
        None => 0,
        Some(value) => match value.as_i64().filter(|value| *value >= 0) {
            Some(value) => value,
            None => return false,
        },
    };
    let retained_maximum = root
        .get("local_file_deliveries")
        .and_then(Value::as_array)
        .into_iter()
        .flatten()
        .filter_map(|row| row.get("id").and_then(Value::as_i64))
        .max()
        .unwrap_or(0);
    let Some(id) = persisted.max(retained_maximum).checked_add(1) else {
        return false;
    };

    let Ok(encoded) = serde_json::to_vec(&record) else {
        return false;
    };
    let key = pharos_runtime_database::operations::bytes_fingerprint(&encoded);
    let Some(rows) = root.as_object_mut().and_then(|object| {
        object
            .entry("local_file_deliveries")
            .or_insert_with(|| Value::Array(Vec::new()))
            .as_array_mut()
    }) else {
        return false;
    };
    let mut terminal_count = rows
        .iter()
        .filter(|row| {
            matches!(
                row.get("delivery_status").and_then(Value::as_str),
                Some("delivered" | "failed")
            )
        })
        .count();
    rows.retain(|row| {
        if terminal_count > 64
            && matches!(
                row.get("delivery_status").and_then(Value::as_str),
                Some("delivered" | "failed")
            )
        {
            terminal_count -= 1;
            false
        } else {
            true
        }
    });
    if rows.len() >= 1024 {
        return false;
    }

    rows.push(serde_json::json!({"id": id, "adapter_kind": kind, "relative_path": path, "record": record, "delivery_key": format!("{id}:{key}"), "delivery_status": "queued", "attempt_count": 0, "next_attempt_at": ""}));
    let Some(object) = root.as_object_mut() else {
        return false;
    };
    object.insert("_local_file_delivery_sequence".to_owned(), Value::from(id));

    true
}

/// Delivers one committed local outbox record. Replayed JSONL writes use the
/// persisted delivery key, and JSON replacement remains naturally idempotent.
pub(super) fn process_one_due_file_delivery(app: &AppRuntime, root: &mut Value) -> bool {
    let now = current_epoch_seconds();
    let Some(rows) = root
        .get_mut("local_file_deliveries")
        .and_then(Value::as_array_mut)
    else {
        return false;
    };
    let mut pending_paths = std::collections::BTreeSet::new();
    let Some(row) = rows.iter_mut().find(|row| {
        let status = row.get("delivery_status").and_then(Value::as_str);
        if !matches!(status, Some("queued" | "retryable")) {
            return false;
        }
        let Some(path) = row.get("relative_path").and_then(Value::as_str) else {
            return false;
        };
        if !pending_paths.insert(path.to_owned()) {
            return false;
        }
        status == Some("queued")
            || (row.get("delivery_status").and_then(Value::as_str) == Some("retryable")
                && row
                    .get("next_attempt_at")
                    .and_then(Value::as_str)
                    .and_then(|text| text.parse::<u64>().ok())
                    .is_some_and(|due| due <= now))
    }) else {
        return false;
    };
    let state_root = app
        .state_path
        .as_deref()
        .and_then(|path| Path::new(path).parent());
    let delivered = state_root.is_some_and(|state_root| deliver_file_record(state_root, row));
    let attempts = row
        .get("attempt_count")
        .and_then(Value::as_u64)
        .unwrap_or(0)
        .saturating_add(1);
    row["attempt_count"] = Value::from(attempts);
    row["delivery_status"] = Value::String(
        if delivered {
            "delivered"
        } else if attempts >= 5 {
            "failed"
        } else {
            "retryable"
        }
        .to_owned(),
    );
    row["next_attempt_at"] = Value::String(if delivered || attempts >= 5 {
        String::new()
    } else {
        now.saturating_add(30 * attempts).to_string()
    });
    true
}

/// Executes only a validated, committed file-record envelope below its data root.
fn deliver_file_record(state_root: &Path, envelope: &Value) -> bool {
    let Some(relative) = envelope.get("relative_path").and_then(Value::as_str) else {
        return false;
    };
    if relative.is_empty()
        || Path::new(relative)
            .components()
            .any(|part| !matches!(part, std::path::Component::Normal(_)))
    {
        return false;
    }
    let Some(record) = envelope.get("record") else {
        return false;
    };
    let output_path = state_root.join(relative);
    match envelope.get("adapter_kind").and_then(Value::as_str) {
        Some("append_jsonl_local") => envelope
            .get("delivery_key")
            .and_then(Value::as_str)
            .is_some_and(|key| {
                pharos_runtime_core::fs::append_json_record_once(
                    state_root,
                    &output_path,
                    key,
                    record,
                    0o640,
                )
                .is_ok()
            }),
        Some("write_json_local") => serde_json::to_vec_pretty(record).ok().is_some_and(|bytes| {
            pharos_runtime_core::fs::write_file_within(state_root, &output_path, &bytes, 0o640)
                .is_ok()
        }),
        _ => false,
    }
}

/// Exercises preparation and delivery independently in focused adapter tests.
#[cfg(test)]
fn dynamic_append_file_record(
    state_path: &Path,
    integrations_root: &Value,
    action: &Value,
    file_record: &Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    action_id: &str,
    entity_id: &str,
    event_kind: &str,
    anchor_record_id: i64,
) -> bool {
    let Some(prepared) = prepare_file_record(
        integrations_root,
        action,
        file_record,
        request,
        input_overrides,
        action_id,
        entity_id,
        event_kind,
        anchor_record_id,
    ) else {
        return false;
    };
    let mut root = serde_json::json!({});
    if !enqueue_file_record(&mut root, prepared) {
        return false;
    }
    state_path
        .parent()
        .is_some_and(|parent| deliver_file_record(parent, &root["local_file_deliveries"][0]))
}

/// Performs dynamic apply approval transition.
fn dynamic_apply_approval_transition(
    object: &mut Map<String, Value>,
    primitive: &Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
) -> Result<(), i32> {
    let current_status = object
        .get("status")
        .and_then(Value::as_str)
        .or_else(|| object.get("workflow_state").and_then(Value::as_str))
        .unwrap_or_default();
    let transition_kind = primitive
        .get("transition_kind")
        .and_then(Value::as_str)
        .unwrap_or_default();
    let now = now_iso_utc();
    match transition_kind {
        "submit" => {
            if current_status != "draft" {
                return Err(422);
            }
            object.insert("status".to_owned(), Value::String("submitted".to_owned()));
            object.insert(
                "workflow_state".to_owned(),
                Value::String("submitted".to_owned()),
            );
            if let Some(field) = primitive
                .get("submitted_at_field")
                .and_then(Value::as_str)
                .filter(|value| !value.is_empty())
            {
                object.insert(field.to_owned(), Value::String(now));
            }
        }
        "approve" => {
            if current_status != "submitted" {
                return Err(422);
            }
            let approver = dynamic_action_input_value(
                request,
                input_overrides,
                primitive
                    .get("approver_input")
                    .and_then(Value::as_str)
                    .unwrap_or("approver"),
            )
            .unwrap_or_default();
            if approver.trim().is_empty() {
                return Err(422);
            }
            let approved_by_field = primitive
                .get("approved_by_field")
                .and_then(Value::as_str)
                .filter(|value| !value.is_empty())
                .unwrap_or("approved_by");
            object.insert(approved_by_field.to_owned(), Value::String(approver));
            if let Some(field) = primitive
                .get("approved_at_field")
                .and_then(Value::as_str)
                .filter(|value| !value.is_empty())
            {
                object.insert(field.to_owned(), Value::String(now.clone()));
            }
            object.insert("status".to_owned(), Value::String("approved".to_owned()));
            object.insert(
                "workflow_state".to_owned(),
                Value::String("approved".to_owned()),
            );
            let rejected_by_field = primitive
                .get("rejected_by_field")
                .and_then(Value::as_str)
                .filter(|value| !value.is_empty())
                .unwrap_or("rejected_by");
            let rejection_reason_field = primitive
                .get("rejection_reason_field")
                .and_then(Value::as_str)
                .filter(|value| !value.is_empty())
                .unwrap_or("rejection_reason");
            object.insert(rejected_by_field.to_owned(), Value::String(String::new()));
            object.insert(
                rejection_reason_field.to_owned(),
                Value::String(String::new()),
            );
            if let Some(field) = primitive
                .get("rejected_at_field")
                .and_then(Value::as_str)
                .filter(|value| !value.is_empty())
            {
                object.insert(field.to_owned(), Value::String(String::new()));
            }
        }
        "reject" => {
            if current_status != "submitted" {
                return Err(422);
            }
            let approver = dynamic_action_input_value(
                request,
                input_overrides,
                primitive
                    .get("approver_input")
                    .and_then(Value::as_str)
                    .unwrap_or("approver"),
            )
            .unwrap_or_default();
            let reason = dynamic_action_input_value(
                request,
                input_overrides,
                primitive
                    .get("reason_input")
                    .and_then(Value::as_str)
                    .unwrap_or("reason"),
            )
            .unwrap_or_default();
            if approver.trim().is_empty() || reason.trim().is_empty() {
                return Err(422);
            }
            let rejected_by_field = primitive
                .get("rejected_by_field")
                .and_then(Value::as_str)
                .filter(|value| !value.is_empty())
                .unwrap_or("approved_by");
            let rejection_reason_field = primitive
                .get("rejection_reason_field")
                .and_then(Value::as_str)
                .filter(|value| !value.is_empty())
                .unwrap_or("rejection_reason");
            let approved_by_field = primitive
                .get("approved_by_field")
                .and_then(Value::as_str)
                .filter(|value| !value.is_empty())
                .unwrap_or("approved_by");
            object.insert(rejected_by_field.to_owned(), Value::String(approver));
            object.insert(rejection_reason_field.to_owned(), Value::String(reason));
            object.insert(approved_by_field.to_owned(), Value::String(String::new()));
            if let Some(field) = primitive
                .get("approved_at_field")
                .and_then(Value::as_str)
                .filter(|value| !value.is_empty())
            {
                object.insert(field.to_owned(), Value::String(String::new()));
            }
            if let Some(field) = primitive
                .get("rejected_at_field")
                .and_then(Value::as_str)
                .filter(|value| !value.is_empty())
            {
                object.insert(field.to_owned(), Value::String(now));
            }
            object.insert("status".to_owned(), Value::String("rejected".to_owned()));
            object.insert(
                "workflow_state".to_owned(),
                Value::String("rejected".to_owned()),
            );
        }
        _ => return Err(422),
    }
    Ok(())
}

/// Performs dynamic append audit record.
fn dynamic_append_audit_record(
    root: &mut Value,
    integrations_root: &Value,
    action: &Value,
    audit: &Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    action_id: &str,
    entity_id: &str,
    event_kind: &str,
    anchor_record_id: i64,
) -> bool {
    let adapter_id = audit
        .get("adapter_id")
        .and_then(Value::as_str)
        .filter(|value| !value.is_empty())
        .unwrap_or_default();
    let Some(adapter) = dynamic_find_integration_adapter(integrations_root, adapter_id) else {
        return false;
    };
    let adapter_kind = adapter
        .get("adapter_kind")
        .and_then(Value::as_str)
        .unwrap_or_default();
    if !dynamic_integration_audit_adapter_kind(adapter_kind) {
        return false;
    }
    let event = dynamic_notification_parameter_value(
        audit.get("event").unwrap_or(&Value::Null),
        request,
        input_overrides,
        action_id,
        entity_id,
        event_kind,
        "",
        "",
        anchor_record_id,
    )
    .map(|value| value_to_text(&value))
    .filter(|value| !value.is_empty())
    .unwrap_or_else(|| event_kind.to_owned());
    let level = dynamic_notification_parameter_value(
        audit.get("level").unwrap_or(&Value::Null),
        request,
        input_overrides,
        action_id,
        entity_id,
        event_kind,
        "",
        "",
        anchor_record_id,
    )
    .map(|value| value_to_text(&value))
    .filter(|value| !value.is_empty())
    .unwrap_or_else(|| "info".to_owned());
    let message = dynamic_notification_parameter_value(
        audit.get("message").unwrap_or(&Value::Null),
        request,
        input_overrides,
        action_id,
        entity_id,
        event_kind,
        "",
        "",
        anchor_record_id,
    )
    .map(|value| value_to_text(&value))
    .unwrap_or_default();
    let mut annotations = Map::new();
    if let Some(rows) = audit.get("annotations").and_then(Value::as_array) {
        if rows.len() > 16 {
            return false;
        }
        for row in rows {
            let Some(field) = row
                .get("field")
                .and_then(Value::as_str)
                .filter(|value| !value.is_empty() && value.len() <= 128)
            else {
                return false;
            };
            let value = dynamic_notification_parameter_value(
                row.get("value").unwrap_or(&Value::Null),
                request,
                input_overrides,
                action_id,
                entity_id,
                event_kind,
                "",
                "",
                anchor_record_id,
            )
            .unwrap_or(Value::Null);
            annotations.insert(field.to_owned(), value);
        }
    }
    let audit_id = dynamic_take_next_section_id(root, "audit_log");
    let resource_id =
        dynamic_action_resource_id(action, request, input_overrides, anchor_record_id);
    let Some(rows) = root
        .as_object_mut()
        .map(|object| {
            object
                .entry("audit_log".to_owned())
                .or_insert_with(|| Value::Array(Vec::new()))
        })
        .and_then(Value::as_array_mut)
    else {
        return false;
    };
    let mut record = serde_json::Map::from_iter([
        ("id".to_owned(), Value::from(audit_id)),
        (
            "adapter_id".to_owned(),
            Value::String(adapter_id.to_owned()),
        ),
        (
            "adapter_kind".to_owned(),
            Value::String(adapter_kind.to_owned()),
        ),
        ("event".to_owned(), Value::String(event)),
        ("level".to_owned(), Value::String(level)),
        ("message".to_owned(), Value::String(message)),
        ("action_id".to_owned(), Value::String(action_id.to_owned())),
        ("entity_id".to_owned(), Value::String(entity_id.to_owned())),
        (
            "event_kind".to_owned(),
            Value::String(event_kind.to_owned()),
        ),
        ("resource_id".to_owned(), Value::String(resource_id)),
        ("created_at".to_owned(), Value::String(now_iso_utc())),
    ]);
    if !annotations.is_empty() {
        record.insert("annotations".to_owned(), Value::Object(annotations));
    }
    rows.push(Value::Object(record));
    true
}

/// Maps one action failure status to a stable bounded failure code.
fn dynamic_action_failure_code(status: i32) -> &'static str {
    match status {
        400 => "bad_request",
        403 => "forbidden",
        404 => "not_found",
        409 => "conflict",
        415 => "unsupported_media_type",
        422 => "validation_failed",
        500 => "internal_error",
        _ => "action_failed",
    }
}

/// Returns the request path without a query string.
///
/// # Parameters
///
/// - `request`: Input consumed by [`crate::runtime::dynamic::request_path`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::request_path`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::request_path(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn request_path(request: Option<&HttpRequest>) -> String {
    request
        .and_then(|request| super::http::request_path_without_query(&request.target))
        .unwrap_or_else(|| "/".to_owned())
}

/// Detects the HTMX request header using case-insensitive line parsing.
///
/// # Parameters
///
/// - `request`: Input consumed by [`crate::runtime::dynamic::request_is_hx`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::request_is_hx`] holds and `false` otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::request_is_hx`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::request_is_hx(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn request_is_hx(request: Option<&HttpRequest>) -> bool {
    request.is_some_and(|request| {
        request.headers.lines().any(|line| {
            let (name, value) = line.split_once(':').unwrap_or(("", ""));
            name.trim().eq_ignore_ascii_case("hx-request")
                && value.trim().eq_ignore_ascii_case("true")
        })
    })
}

/// Performs cookie value.
fn cookie_value(cookie_header: Option<&str>, name: &str) -> Option<String> {
    pharos_runtime_security::cookie_value(cookie_header, name)
}

/// Performs constant time equal.
fn constant_time_equal(left: &[u8], right: &[u8]) -> bool {
    pharos_runtime_security::constant_time_equal(left, right)
}

/// Performs hex digest.
///
/// # Parameters
///
/// - `bytes`: Raw bytes to hash into a lowercase hexadecimal SHA-256 digest.
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::hex_digest`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::hex_digest(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub(crate) fn hex_digest(bytes: &[u8]) -> String {
    pharos_runtime_dynamic::hex_digest(bytes)
}

/// Computes HMAC-SHA256 with a fixed-size key block and no external process.
///
/// # Parameters
///
/// - `key`: Input consumed by [`crate::runtime::dynamic::hmac_sha256`].
/// - `data`: Input consumed by [`crate::runtime::dynamic::hmac_sha256`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::hmac_sha256`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::hmac_sha256("key", b"");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn hmac_sha256(key: &[u8], data: &[u8]) -> [u8; 32] {
    pharos_runtime_dynamic::hmac_sha256(key, data)
}

/// Verifies the native PBKDF2 or development password format without leaking timing information.
///
/// # Parameters
///
/// - `stored_hash`: Input consumed by [`crate::runtime::dynamic::verify_password_hash`].
/// - `password`: Input consumed by [`crate::runtime::dynamic::verify_password_hash`].
/// - `username`: Input consumed by [`crate::runtime::dynamic::verify_password_hash`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::verify_password_hash`] holds and `false` otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::verify_password_hash(todo!(), todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn verify_password_hash(stored_hash: &str, password: &str, username: Option<&str>) -> bool {
    pharos_runtime_dynamic::verify_password_hash(stored_hash, password, username)
}

/// Generates a cryptographically random secret encoded as lowercase hexadecimal text.
///
/// # Parameters
///
/// - `byte_count`: Input consumed by [`crate::runtime::dynamic::random_secret_hex`].
///
/// # Returns
///
/// Returns the successful value produced by [`crate::runtime::dynamic::random_secret_hex`] or an error when the operation fails.
///
/// # Errors
///
/// Returns an error value when [`crate::runtime::dynamic::random_secret_hex`] cannot complete successfully.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::random_secret_hex(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn random_secret_hex(byte_count: usize) -> Result<String, i32> {
    pharos_runtime_dynamic::random_secret_hex(byte_count).map_err(|_| 69)
}

/// Creates a salted production PBKDF2-SHA256 password hash.
///
/// # Parameters
///
/// - `password`: Input consumed by [`crate::runtime::dynamic::hash_password`].
///
/// # Returns
///
/// Returns the successful value produced by [`crate::runtime::dynamic::hash_password`] or an error when the operation fails.
///
/// # Errors
///
/// Returns an error value when [`crate::runtime::dynamic::hash_password`] cannot complete successfully.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::hash_password(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn hash_password(password: &str) -> Result<String, i32> {
    pharos_runtime_dynamic::hash_password(password).map_err(|_| 69)
}

/// Returns the active user id for a session cookie, rejecting expired or revoked rows.
///
/// # Parameters
///
/// - `state_path`: Input consumed by [`crate::runtime::dynamic::active_user_id_from_cookie`].
/// - `cookie_header`: Input consumed by [`crate::runtime::dynamic::active_user_id_from_cookie`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::active_user_id_from_cookie`].
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::active_user_id_from_cookie`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::active_user_id_from_cookie(todo!(), "session=abc");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn active_user_id_from_cookie(state_path: &Path, cookie_header: Option<&str>) -> i64 {
    let Some(token) = cookie_value(cookie_header, pharos_runtime_security::SESSION_COOKIE_NAME)
        .filter(|value| !value.is_empty())
    else {
        return 0;
    };
    let token_hash = hex_digest(&Sha256::digest(token.as_bytes()));
    let Some(root) = fs::read_to_string(state_path)
        .ok()
        .and_then(|text| serde_json::from_str::<Value>(&text).ok())
    else {
        return 0;
    };
    let now_epoch = current_epoch_seconds();
    let now_timestamp = super::current_timestamp_utc();
    root.get("sessions")
        .and_then(Value::as_array)
        .and_then(|rows| {
            rows.iter().find(|row| {
                row.get("session_token_hash").and_then(Value::as_str) == Some(token_hash.as_str())
                    && row
                        .get("revoked_at")
                        .is_none_or(|value| value.is_null() || value.as_str() == Some(""))
                    && row
                        .get("expires_at")
                        .and_then(Value::as_str)
                        .is_some_and(|value| {
                            value
                                .parse::<u64>()
                                .map_or(value > now_timestamp.as_str(), |expires| {
                                    expires > now_epoch
                                })
                        })
            })
        })
        .and_then(|row| row.get("user_id"))
        .and_then(|value| value.as_i64().or_else(|| value.as_str()?.parse().ok()))
        .unwrap_or(0)
}

/// Performs current epoch seconds.
fn current_epoch_seconds() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|value| value.as_secs())
        .unwrap_or(0)
}

/// Performs now iso utc.
fn now_iso_utc() -> String {
    current_epoch_seconds().to_string()
}

/// Parses one bounded runtime timestamp from either epoch seconds or UTC text.
///
/// # Parameters
///
/// - `request`: Input consumed by [`crate::runtime::dynamic::csrf_request_valid`].
/// - `form_body`: Input consumed by [`crate::runtime::dynamic::csrf_request_valid`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::csrf_request_valid`] holds and `false` otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::csrf_request_valid`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::csrf_request_valid(todo!(), "name=value");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn csrf_request_valid(request: &HttpRequest, form_body: Option<&str>) -> bool {
    let cookie = cookie_value(request.cookie.as_deref(), "pharos_csrf");
    let form = request_form_value(request, "csrf_token").or_else(|| {
        form_body.and_then(|body| json_commands::request_form_value(body, "csrf_token"))
    });
    let header = request.headers.lines().find_map(|line| {
        let (name, value) = line.split_once(':')?;
        name.trim()
            .eq_ignore_ascii_case("x-csrf-token")
            .then_some(value.trim().to_owned())
    });
    let request_token = form.or(header);
    cookie
        .as_deref()
        .zip(request_token.as_deref())
        .is_some_and(|(left, right)| constant_time_equal(left.as_bytes(), right.as_bytes()))
}

/// Formats a scoped session cookie with secure defaults and an optional clear operation.
///
/// # Parameters
///
/// - `app`: Input consumed by [`crate::runtime::dynamic::session_cookie_header`].
/// - `request`: Input consumed by [`crate::runtime::dynamic::session_cookie_header`].
/// - `token`: Input consumed by [`crate::runtime::dynamic::session_cookie_header`].
/// - `clear`: Input consumed by [`crate::runtime::dynamic::session_cookie_header`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::session_cookie_header`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::session_cookie_header(todo!(), todo!(), todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn session_cookie_header(
    app: &AppRuntime,
    request: Option<&HttpRequest>,
    token: Option<&str>,
    clear: bool,
) -> String {
    let path = app
        .base_path
        .as_deref()
        .filter(|value| !value.is_empty())
        .unwrap_or("/");
    let secure = request_origin_scheme(request, NativeTransportType::Tcp, None, None) == "https";
    pharos_runtime_security::session_cookie_header(path, token, clear, secure)
}

/// Returns a page title from declared metadata or a body parameter fallback.
///
/// # Parameters
///
/// - `page_spec`: Input consumed by [`crate::runtime::dynamic::page_title`].
/// - `body_params`: Input consumed by [`crate::runtime::dynamic::page_title`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::page_title`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::page_title(todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn page_title(page_spec: &Value, body_params: &Value) -> String {
    page_spec
        .get("title")
        .and_then(Value::as_str)
        .filter(|value| !value.is_empty())
        .map(ToOwned::to_owned)
        .or_else(|| {
            page_spec
                .get("title_mode")
                .and_then(Value::as_str)
                .and_then(|key| body_params.get(key))
                .map(value_to_text)
        })
        .unwrap_or_default()
}

/// Performs value to text.
fn value_to_text(value: &Value) -> String {
    match value {
        Value::Null => String::new(),
        Value::String(value) => value.clone(),
        Value::Bool(value) => value.to_string(),
        Value::Number(value) => value.to_string(),
        _ => serde_json::to_string(value).unwrap_or_default(),
    }
}

/// Performs dynamic value order cmp.
fn dynamic_value_order_cmp(left: Option<&Value>, right: Option<&Value>) -> Ordering {
    match (
        left.and_then(Value::as_i64),
        right.and_then(Value::as_i64),
        left.map(value_to_text),
        right.map(value_to_text),
    ) {
        (Some(left_num), Some(right_num), _, _) => left_num.cmp(&right_num),
        (_, _, Some(left_text), Some(right_text)) => left_text.cmp(&right_text),
        (Some(_), None, _, _) => Ordering::Less,
        (None, Some(_), _, _) => Ordering::Greater,
        _ => Ordering::Equal,
    }
}

/// Performs dynamic record contract value.
fn dynamic_record_contract_value(
    root: &mut Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    field_spec: &Value,
    anchor_record_id: i64,
    list_item: Option<&str>,
    list_index: Option<usize>,
    matched_record_count: i64,
    updated_record_count: i64,
    skipped_record_count: i64,
) -> Value {
    let source = field_spec
        .get("source")
        .and_then(Value::as_str)
        .unwrap_or_default();
    match source {
        "next_id" => Value::from(dynamic_take_next_global_id(root)),
        "input" => Value::String(
            dynamic_action_input_value(
                request,
                input_overrides,
                field_spec
                    .get("field")
                    .and_then(Value::as_str)
                    .unwrap_or_default(),
            )
            .unwrap_or_default(),
        ),
        "literal" => field_spec.get("value").cloned().unwrap_or(Value::Null),
        "context_anchor_created_id" => Value::from(anchor_record_id),
        "list_item" => Value::String(list_item.unwrap_or_default().to_owned()),
        "list_index" => Value::from(list_index.unwrap_or(0).saturating_add(1) as i64),
        "context_matched_count" => Value::from(matched_record_count.max(0)),
        "context_updated_count" => Value::from(updated_record_count.max(0)),
        "context_skipped_count" => Value::from(skipped_record_count.max(0)),
        "now" => Value::String(now_iso_utc()),
        _ => Value::Null,
    }
}

/// Persists a complete state JSON document through the bounded filesystem helper.
///
/// # Parameters
///
/// - `app`: Input consumed by [`crate::runtime::dynamic::persist_app_state_text`].
/// - `state_json`: Input consumed by [`crate::runtime::dynamic::persist_app_state_text`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::persist_app_state_text`].
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::persist_app_state_text`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::persist_app_state_text(todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn persist_app_state_text(app: &AppRuntime, state_json: &str) -> i32 {
    let Some(path) = app.state_path.as_deref().filter(|value| !value.is_empty()) else {
        return 69;
    };
    if state_json.len() > 16 * 1024 * 1024 || serde_json::from_str::<Value>(state_json).is_err() {
        return 64;
    }
    super::fs::write_text_file(Path::new(path), state_json)
}

/// Appends one rate-limit event with a monotonic state-local identifier.
///
/// # Parameters
///
/// - `app`: Input consumed by [`crate::runtime::dynamic::append_rate_limit_event_runtime_local`].
/// - `policy_key`: Input consumed by [`crate::runtime::dynamic::append_rate_limit_event_runtime_local`].
/// - `bucket_key`: Input consumed by [`crate::runtime::dynamic::append_rate_limit_event_runtime_local`].
/// - `outcome`: Input consumed by [`crate::runtime::dynamic::append_rate_limit_event_runtime_local`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::append_rate_limit_event_runtime_local`].
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::append_rate_limit_event_runtime_local`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::append_rate_limit_event_runtime_local(todo!(), todo!(), todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn append_rate_limit_event_runtime_local(
    app: &AppRuntime,
    policy_key: &str,
    bucket_key: &str,
    outcome: &str,
) -> i32 {
    if policy_key.is_empty() || bucket_key.len() > 512 || outcome.len() > 128 {
        return 64;
    }
    let Some(path) = app.state_path.as_deref() else {
        return 69;
    };
    let Ok(mut root) = fs::read_to_string(path)
        .ok()
        .and_then(|text| serde_json::from_str::<Value>(&text).ok())
        .ok_or(66)
    else {
        return 66;
    };
    let id = root
        .get("rate_limit_events")
        .and_then(Value::as_array)
        .into_iter()
        .flatten()
        .filter_map(|row| row.get("id").and_then(Value::as_i64))
        .max()
        .unwrap_or(0)
        + 1;
    let events = root.as_object_mut().and_then(|object| {
        let value = object
            .entry("rate_limit_events".to_owned())
            .or_insert_with(|| Value::Array(Vec::new()));
        if !value.is_array() {
            *value = Value::Array(Vec::new());
        }
        value.as_array_mut()
    });
    let Some(events) = events else {
        return 69;
    };
    let created_epoch = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|value| value.as_secs())
        .unwrap_or(0);
    events.push(serde_json::json!({"id": id, "policy_key": policy_key, "bucket_key": bucket_key, "outcome": outcome, "created_at": now_iso_utc(), "created_epoch": created_epoch}));
    persist_app_state_text(app, &serde_json::to_string(&root).unwrap_or_default())
}

/// Atomically admits one request under a bounded state-backed rate-limit window.
///
/// # Parameters
///
/// - `app`: Input consumed by [`crate::runtime::dynamic::admit_rate_limited_request_runtime_local`].
/// - `policy_key`: Input consumed by [`crate::runtime::dynamic::admit_rate_limited_request_runtime_local`].
/// - `bucket_key`: Input consumed by [`crate::runtime::dynamic::admit_rate_limited_request_runtime_local`].
/// - `maximum`: Input consumed by [`crate::runtime::dynamic::admit_rate_limited_request_runtime_local`].
/// - `window_seconds`: Input consumed by [`crate::runtime::dynamic::admit_rate_limited_request_runtime_local`].
///
/// # Returns
///
/// Returns the successful value produced by [`crate::runtime::dynamic::admit_rate_limited_request_runtime_local`] or an error when the operation fails.
///
/// # Errors
///
/// Returns an error value when [`crate::runtime::dynamic::admit_rate_limited_request_runtime_local`] cannot complete successfully.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::admit_rate_limited_request_runtime_local`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::admit_rate_limited_request_runtime_local(todo!(), todo!(), todo!(), todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn admit_rate_limited_request_runtime_local(
    app: &AppRuntime,
    policy_key: &str,
    bucket_key: &str,
    maximum: usize,
    window_seconds: u64,
) -> Result<bool, i32> {
    if policy_key.is_empty()
        || policy_key.len() > 128
        || bucket_key.is_empty()
        || bucket_key.len() > 512
        || maximum == 0
        || maximum > 100_000
        || window_seconds == 0
        || window_seconds > 86_400
    {
        return Err(64);
    }
    let path = app
        .state_path
        .as_deref()
        .filter(|value| !value.is_empty())
        .map(Path::new)
        .ok_or(69)?;
    let mut root = fs::read_to_string(path)
        .ok()
        .and_then(|text| serde_json::from_str::<Value>(&text).ok())
        .filter(Value::is_object)
        .ok_or(66)?;
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_err(|_| 70)?
        .as_secs();
    let cutoff = now.saturating_sub(window_seconds);
    let events = root
        .as_object_mut()
        .ok_or(66)?
        .entry("rate_limit_events".to_owned())
        .or_insert_with(|| Value::Array(Vec::new()));
    if !events.is_array() {
        *events = Value::Array(Vec::new());
    }
    let events = events.as_array_mut().ok_or(66)?;
    events.retain(|event| {
        event
            .get("created_epoch")
            .and_then(Value::as_u64)
            .is_some_and(|created| created >= cutoff)
    });
    let observed = events
        .iter()
        .filter(|event| {
            event.get("policy_key").and_then(Value::as_str) == Some(policy_key)
                && event.get("bucket_key").and_then(Value::as_str) == Some(bucket_key)
                && event
                    .get("created_epoch")
                    .and_then(Value::as_u64)
                    .is_some_and(|created| created >= cutoff)
        })
        .count();
    if observed >= maximum {
        return Ok(false);
    }
    if events.len() >= app.resource_budgets.max_state_section_items {
        return Err(69);
    }
    let next_id = events
        .iter()
        .filter_map(|event| event.get("id").and_then(Value::as_i64))
        .max()
        .unwrap_or(0)
        .saturating_add(1);
    events.push(serde_json::json!({
        "id": next_id,
        "policy_key": policy_key,
        "bucket_key": bucket_key,
        "outcome": "admitted",
        "created_at": now_iso_utc(),
        "created_epoch": now,
    }));
    let encoded = serde_json::to_string(&root).map_err(|_| 70)?;
    (super::fs::write_text_file(path, &encoded) == 0)
        .then_some(true)
        .ok_or(69)
}

/// Updates one notification delivery row and persists the validated state document.
///
/// # Parameters
///
/// - `app`: Input consumed by [`crate::runtime::dynamic::update_notification_delivery_status`].
/// - `delivery_id`: Input consumed by [`crate::runtime::dynamic::update_notification_delivery_status`].
/// - `status`: Input consumed by [`crate::runtime::dynamic::update_notification_delivery_status`].
/// - `delivered_at`: Input consumed by [`crate::runtime::dynamic::update_notification_delivery_status`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::update_notification_delivery_status`].
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::update_notification_delivery_status`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::update_notification_delivery_status(todo!(), todo!(), 400, todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn update_notification_delivery_status(
    app: &AppRuntime,
    delivery_id: i64,
    status: &str,
    delivered_at: Option<&str>,
) -> i32 {
    if delivery_id <= 0 || status.is_empty() || status.len() > 64 {
        return 64;
    }
    let Some(path) = app.state_path.as_deref() else {
        return 69;
    };
    let Ok(mut root) = fs::read_to_string(path)
        .ok()
        .and_then(|text| serde_json::from_str::<Value>(&text).ok())
        .ok_or(66)
    else {
        return 66;
    };
    let Some(row) = root
        .get_mut("notification_deliveries")
        .and_then(Value::as_array_mut)
        .and_then(|rows| {
            rows.iter_mut()
                .find(|row| row.get("id").and_then(Value::as_i64) == Some(delivery_id))
        })
    else {
        return 66;
    };
    let Some(object) = row.as_object_mut() else {
        return 69;
    };
    object.insert(
        "delivery_status".to_owned(),
        Value::String(status.to_owned()),
    );
    object.insert("updated_at".to_owned(), Value::String(now_iso_utc()));
    if let Some(delivered_at) = delivered_at {
        object.insert(
            "delivered_at".to_owned(),
            Value::String(delivered_at.to_owned()),
        );
    }
    persist_app_state_text(app, &serde_json::to_string(&root).unwrap_or_default())
}

/// Renders a bounded JSON error HTTP response without interpolating unescaped data.
///
/// # Parameters
///
/// - `status`: Input consumed by [`crate::runtime::dynamic::write_json_error`].
/// - `error_type`: Input consumed by [`crate::runtime::dynamic::write_json_error`].
/// - `message`: Input consumed by [`crate::runtime::dynamic::write_json_error`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::write_json_error`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::write_json_error(400, todo!(), "example");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn write_json_error(status: u16, error_type: &str, message: &str) -> String {
    let code = pharos_runtime_core::error::ErrorCode::from_http_status(status);
    let mut envelope = pharos_runtime_core::error::envelope(code, message, None);
    if let Some(error) = envelope.get_mut("error").and_then(Value::as_object_mut) {
        error.insert("context".to_owned(), Value::String(error_type.to_owned()));
    }
    let body = serde_json::to_string(&envelope).unwrap_or_else(|_| {
        "{\"schema\":\"pharos.error.v1\",\"error\":{\"code\":\"internal_failure\",\"message\":\"response encoding failed\",\"retryable\":false,\"correlation_id\":\"\"}}".to_owned()
    });
    format!(
        "HTTP/1.1 {}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nCache-Control: no-store\r\nConnection: close\r\n\r\n{}",
        super::http::http_status_text(status),
        body.len(),
        body
    )
}

/// Builds an HTMX fragment response with a scoped base path and CSRF cookie.
///
/// # Parameters
///
/// - `app`: Input consumed by [`crate::runtime::dynamic::write_hx_fragment_response_with_csrf`].
/// - `request`: Input consumed by [`crate::runtime::dynamic::write_hx_fragment_response_with_csrf`].
/// - `fragment_html`: Input consumed by [`crate::runtime::dynamic::write_hx_fragment_response_with_csrf`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::write_hx_fragment_response_with_csrf`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::write_hx_fragment_response_with_csrf(todo!(), todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn write_hx_fragment_response_with_csrf(
    app: &AppRuntime,
    request: Option<&HttpRequest>,
    fragment_html: &str,
) -> String {
    let scoped = json_commands::template_apply_base_path(fragment_html, app.base_path.as_deref())
        .unwrap_or_default();
    let token = cookie_value(
        request.and_then(|request| request.cookie.as_deref()),
        "pharos_csrf",
    )
    .unwrap_or_default();
    format!(
        "HTTP/1.1 200 OK\r\nContent-Type: text/html; charset=utf-8\r\nContent-Length: {}\r\nSet-Cookie: pharos_csrf={}; Path=/; SameSite=Lax\r\nConnection: close\r\n\r\n{}",
        scoped.len(),
        token,
        scoped
    )
}

/// Renders one declared page into a complete response string with bounded headers.
///
/// # Parameters
///
/// - `app_root`: Input consumed by [`crate::runtime::dynamic::render_declared_page_response_with_headers`].
/// - `page_id`: Input consumed by [`crate::runtime::dynamic::render_declared_page_response_with_headers`].
/// - `body_params_json`: Input consumed by [`crate::runtime::dynamic::render_declared_page_response_with_headers`].
/// - `extra_headers`: Input consumed by [`crate::runtime::dynamic::render_declared_page_response_with_headers`].
///
/// # Returns
///
/// Returns the successful value produced by [`crate::runtime::dynamic::render_declared_page_response_with_headers`] or an error when the operation fails.
///
/// # Errors
///
/// Returns an error value when [`crate::runtime::dynamic::render_declared_page_response_with_headers`] cannot complete successfully.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::render_declared_page_response_with_headers`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::render_declared_page_response_with_headers(std::path::Path::new("apps/dev_docs"), "home", todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn render_declared_page_response_with_headers(
    app_root: &Path,
    page_id: &str,
    body_params_json: &str,
    extra_headers: Option<&str>,
) -> Result<String, i32> {
    let page = json_commands::template_declared_page_contract(app_root, page_id).ok_or(404)?;
    let page_spec: Value = serde_json::from_str(&page).map_err(|_| 500)?;
    let template_path = page_spec
        .get("template_path")
        .and_then(Value::as_str)
        .ok_or(500)?;
    let params: Value = serde_json::from_str(body_params_json).map_err(|_| 400)?;
    let html = json_commands::render_file_template(Path::new(template_path), &params).ok_or(500)?;
    let headers = extra_headers.unwrap_or("").replace(['\r', '\n'], " ");
    Ok(format!(
        "HTTP/1.1 200 OK\r\nContent-Type: text/html; charset=utf-8\r\nContent-Length: {}\r\n{}\r\nConnection: close\r\n\r\n{}",
        html.len(),
        headers,
        html
    ))
}

/// Renders a declared page without additional response headers.
///
/// # Parameters
///
/// - `app_root`: Input consumed by [`crate::runtime::dynamic::render_declared_page_response`].
/// - `page_id`: Input consumed by [`crate::runtime::dynamic::render_declared_page_response`].
/// - `body_params_json`: Input consumed by [`crate::runtime::dynamic::render_declared_page_response`].
///
/// # Returns
///
/// Returns the successful value produced by [`crate::runtime::dynamic::render_declared_page_response`] or an error when the operation fails.
///
/// # Errors
///
/// Returns an error value when [`crate::runtime::dynamic::render_declared_page_response`] cannot complete successfully.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::render_declared_page_response(std::path::Path::new("apps/dev_docs"), "home", todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn render_declared_page_response(
    app_root: &Path,
    page_id: &str,
    body_params_json: &str,
) -> Result<String, i32> {
    render_declared_page_response_with_headers(app_root, page_id, body_params_json, None)
}

/// Resolves the small page parameter contract needed by the final Rust renderer.
///
/// # Parameters
///
/// - `page_id`: Input consumed by [`crate::runtime::dynamic::resolve_declared_page_params_json`].
/// - `request_path`: Input consumed by [`crate::runtime::dynamic::resolve_declared_page_params_json`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::resolve_declared_page_params_json`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::resolve_declared_page_params_json("home", "/");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn resolve_declared_page_params_json(page_id: &str, request_path: &str) -> String {
    serde_json::to_string(&serde_json::json!({"page_id": page_id, "request_path": request_path, "error_html": "", "next_input": ""})).unwrap_or_else(|_| "{}".to_owned())
}

/// Handles registration by validating form input and appending a password-hash user row.
///
/// # Parameters
///
/// - `state_path`: Input consumed by [`crate::runtime::dynamic::handle_register_account_action`].
/// - `request`: Input consumed by [`crate::runtime::dynamic::handle_register_account_action`].
///
/// # Returns
///
/// Returns the successful value produced by [`crate::runtime::dynamic::handle_register_account_action`] or an error when the operation fails.
///
/// # Errors
///
/// Returns an error value when [`crate::runtime::dynamic::handle_register_account_action`] cannot complete successfully.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::handle_register_account_action`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::handle_register_account_action(todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn handle_register_account_action(
    state_path: &Path,
    request: &HttpRequest,
) -> Result<String, i32> {
    let username = request_form_value(request, "username").ok_or(422)?;
    let email = request_form_value(request, "email").ok_or(422)?;
    let password = request_form_value(request, "password").ok_or(422)?;
    let confirmation = request_form_value(request, "password_confirm").unwrap_or_default();
    if username.is_empty()
        || !json_commands::generated_surface_text_is_email(&email)
        || password.len() < 8
        || password != confirmation
    {
        return Err(422);
    }
    let mut root: Value =
        serde_json::from_str(&fs::read_to_string(state_path).map_err(|_| 66)?).map_err(|_| 66)?;
    let users = root
        .as_object_mut()
        .ok_or(66)?
        .entry("users".to_owned())
        .or_insert_with(|| Value::Array(Vec::new()));
    let rows = users.as_array_mut().ok_or(69)?;
    if rows.iter().any(|row| {
        row.get("username").and_then(Value::as_str) == Some(username.as_str())
            || row
                .get("email")
                .and_then(Value::as_str)
                .is_some_and(|value| value.eq_ignore_ascii_case(&email))
    }) {
        return Err(409);
    }
    let id = rows
        .iter()
        .filter_map(|row| row.get("id").and_then(Value::as_i64))
        .max()
        .unwrap_or(0)
        + 1;
    let password_hash = hash_password(&password)?;
    rows.push(serde_json::json!({"id": id, "username": username, "email": email, "password_hash": password_hash, "is_active": true, "email_verified": false}));
    super::fs::write_text_file(state_path, &serde_json::to_string(&root).map_err(|_| 75)?)
        .eq(&0)
        .then_some(format!("{{\"user_id\":{id}}}"))
        .ok_or(69)
}

/// Claims an active email verification row and marks its user verified.
///
/// # Parameters
///
/// - `state_path`: Input consumed by [`crate::runtime::dynamic::handle_verify_email_action`].
/// - `token_hash`: Input consumed by [`crate::runtime::dynamic::handle_verify_email_action`].
/// - `now`: Input consumed by [`crate::runtime::dynamic::handle_verify_email_action`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::handle_verify_email_action`].
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::handle_verify_email_action`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::handle_verify_email_action(todo!(), todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn handle_verify_email_action(state_path: &Path, token_hash: &str, now: &str) -> i32 {
    let Ok(mut root) = fs::read_to_string(state_path)
        .ok()
        .and_then(|text| serde_json::from_str::<Value>(&text).ok())
        .ok_or(66)
    else {
        return 66;
    };
    let Some(reset) = root
        .get("email_verifications")
        .and_then(Value::as_array)
        .and_then(|rows| {
            rows.iter().find(|row| {
                row.get("token_hash").and_then(Value::as_str) == Some(token_hash)
                    && row
                        .get("claimed_at")
                        .is_none_or(|value| value.is_null() || value.as_str() == Some(""))
                    && row
                        .get("expires_at")
                        .and_then(Value::as_str)
                        .is_some_and(|value| value > now)
            })
        })
        .cloned()
    else {
        return 404;
    };
    let user_id = reset.get("user_id").and_then(Value::as_i64).unwrap_or(0);
    let Some(object) = root.as_object_mut() else {
        return 66;
    };
    if let Some(row) = object
        .get_mut("email_verifications")
        .and_then(Value::as_array_mut)
        .and_then(|rows| {
            rows.iter_mut()
                .find(|row| row.get("token_hash").and_then(Value::as_str) == Some(token_hash))
        })
    {
        row.as_object_mut()
            .map(|row| row.insert("claimed_at".to_owned(), Value::String(now.to_owned())));
    }
    if let Some(row) = object
        .get_mut("users")
        .and_then(Value::as_array_mut)
        .and_then(|rows| {
            rows.iter_mut()
                .find(|row| row.get("id").and_then(Value::as_i64) == Some(user_id))
        })
    {
        row.as_object_mut()
            .map(|row| row.insert("email_verified".to_owned(), Value::Bool(true)));
    }
    super::fs::write_text_file(
        state_path,
        &serde_json::to_string(&root).unwrap_or_default(),
    )
}

/// Returns a password-reset request outcome without sending email or invoking scripts.
///
/// # Parameters
///
/// - `_state_path`: Input consumed by [`crate::runtime::dynamic::handle_password_reset_request_action`].
/// - `email`: Input consumed by [`crate::runtime::dynamic::handle_password_reset_request_action`].
///
/// # Returns
///
/// Returns the successful value produced by [`crate::runtime::dynamic::handle_password_reset_request_action`] or an error when the operation fails.
///
/// # Errors
///
/// Returns an error value when [`crate::runtime::dynamic::handle_password_reset_request_action`] cannot complete successfully.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::handle_password_reset_request_action(todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn handle_password_reset_request_action(
    _state_path: &Path,
    email: &str,
) -> Result<String, i32> {
    if !json_commands::generated_surface_text_is_email(email) {
        return Err(422);
    }
    random_secret_hex(32)
}

/// Updates the password for a reset token in a persisted state document.
///
/// # Parameters
///
/// - `state_path`: Input consumed by [`crate::runtime::dynamic::handle_password_reset_complete_action`].
/// - `presented_token`: Input consumed by [`crate::runtime::dynamic::handle_password_reset_complete_action`].
/// - `password`: Input consumed by [`crate::runtime::dynamic::handle_password_reset_complete_action`].
/// - `now`: Input consumed by [`crate::runtime::dynamic::handle_password_reset_complete_action`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::handle_password_reset_complete_action`].
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::handle_password_reset_complete_action`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::handle_password_reset_complete_action(todo!(), todo!(), todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn handle_password_reset_complete_action(
    state_path: &Path,
    presented_token: &str,
    password: &str,
    now: &str,
) -> i32 {
    if password.len() < 8 || presented_token.is_empty() {
        return 422;
    }
    let token_hash = hex_digest(&Sha256::digest(presented_token.as_bytes()));
    let Ok(mut root) = fs::read_to_string(state_path)
        .ok()
        .and_then(|text| serde_json::from_str::<Value>(&text).ok())
        .ok_or(66)
    else {
        return 66;
    };
    let user_id = root
        .get("password_resets")
        .and_then(Value::as_array)
        .and_then(|rows| {
            rows.iter().find(|row| {
                row.get("token_hash").and_then(Value::as_str) == Some(token_hash.as_str())
                    && row
                        .get("claimed_at")
                        .is_none_or(|value| value.is_null() || value.as_str() == Some(""))
                    && row
                        .get("expires_at")
                        .and_then(Value::as_str)
                        .is_some_and(|value| value > now)
            })
        })
        .and_then(|row| row.get("user_id"))
        .and_then(Value::as_i64)
        .unwrap_or(0);
    if user_id <= 0 {
        return 404;
    }
    let Ok(password_hash) = hash_password(password) else {
        return 69;
    };
    if let Some(reset) = root
        .get_mut("password_resets")
        .and_then(Value::as_array_mut)
        .and_then(|rows| {
            rows.iter_mut().find(|row| {
                row.get("token_hash").and_then(Value::as_str) == Some(token_hash.as_str())
            })
        })
    {
        reset["claimed_at"] = Value::String(now.to_owned());
    }
    if let Some(row) = root
        .get_mut("users")
        .and_then(Value::as_array_mut)
        .and_then(|rows| {
            rows.iter_mut()
                .find(|row| row.get("id").and_then(Value::as_i64) == Some(user_id))
        })
    {
        row.as_object_mut()
            .map(|row| row.insert("password_hash".to_owned(), Value::String(password_hash)));
    } else {
        return 404;
    }
    if let Some(rows) = root.get_mut("sessions").and_then(Value::as_array_mut) {
        for row in rows {
            if row.get("user_id").and_then(Value::as_i64) == Some(user_id)
                && row
                    .get("revoked_at")
                    .is_none_or(|value| value.is_null() || value.as_str() == Some(""))
                && let Some(object) = row.as_object_mut()
            {
                object.insert("revoked_at".to_owned(), Value::String(now.to_owned()));
            }
        }
    }
    super::fs::write_text_file(
        state_path,
        &serde_json::to_string(&root).unwrap_or_default(),
    )
}

/// Returns a bounded numeric resource identifier from a job resource key.
///
/// # Parameters
///
/// - `resource_id`: Input consumed by [`crate::runtime::dynamic::resource_numeric_id_jobs_logic_local`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::resource_numeric_id_jobs_logic_local`].
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::resource_numeric_id_jobs_logic_local`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::resource_numeric_id_jobs_logic_local(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn resource_numeric_id_jobs_logic_local(resource_id: &str) -> i64 {
    resource_id
        .rsplit([':', '/', '-'])
        .next()
        .and_then(|value| value.parse::<i64>().ok())
        .filter(|value| *value > 0)
        .unwrap_or(0)
}

/// Builds a deterministic cancellation report without a shell or external process.
///
/// # Parameters
///
/// - `app`: Input consumed by [`crate::runtime::dynamic::dynamic_job_cancel_report_json`].
/// - `job_id`: Input consumed by [`crate::runtime::dynamic::dynamic_job_cancel_report_json`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::dynamic_job_cancel_report_json`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::dynamic_job_cancel_report_json(todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn dynamic_job_cancel_report_json(app: &AppRuntime, job_id: i64) -> Option<String> {
    (job_id > 0).then(|| serde_json::json!({"report_id":"pharos-job-cancel-v1","app_id":app.app_id,"job_id":job_id,"status":"cancel_requested"}).to_string())
}

/// Builds a deterministic retry report without a shell or external process.
///
/// # Parameters
///
/// - `app`: Input consumed by [`crate::runtime::dynamic::dynamic_job_retry_report_json`].
/// - `job_id`: Input consumed by [`crate::runtime::dynamic::dynamic_job_retry_report_json`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::dynamic_job_retry_report_json`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::dynamic_job_retry_report_json(todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn dynamic_job_retry_report_json(app: &AppRuntime, job_id: i64) -> Option<String> {
    (job_id > 0).then(|| serde_json::json!({"report_id":"pharos-job-retry-v1","app_id":app.app_id,"job_id":job_id,"status":"retry_requested"}).to_string())
}

/// Loads one app contract document from the active app root.
pub(super) fn dynamic_contract_doc(app: &AppRuntime, contract_key: &str) -> Option<Value> {
    let app_root = Path::new(app.app_root.as_deref()?);
    let manifest_path = if app_root.join("pharos.app.json").is_file() {
        app_root.join("pharos.app.json")
    } else {
        app_root.join("pharos.app.yml")
    };
    let manifest = json_commands::json_doc_load_file(&manifest_path)?;
    let relative = manifest
        .get("contracts")
        .and_then(Value::as_object)?
        .get(contract_key)
        .and_then(Value::as_str)?;
    if relative.is_empty() || relative.contains("..") || Path::new(relative).is_absolute() {
        return None;
    }
    json_commands::json_doc_load_file(&app_root.join(relative))
}

/// Loads the active app integration contract from the app root.
///
/// # Parameters
///
/// - `app`: Input consumed by [`crate::runtime::dynamic::integration_contract_doc`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::integration_contract_doc`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::integration_contract_doc(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn integration_contract_doc(app: &AppRuntime) -> Option<Value> {
    dynamic_contract_doc(app, "integration_contract")
}

/// Finds one declarative action contract by id.
fn dynamic_find_action_contract<'a>(actions_root: &'a Value, action_id: &str) -> Option<&'a Value> {
    actions_root
        .get("actions")
        .and_then(Value::as_array)?
        .iter()
        .find(|action| action.get("action_id").and_then(Value::as_str) == Some(action_id))
}

/// Finds one workflow transition contract by action id.
fn dynamic_find_workflow_transition<'a>(
    workflow_root: &'a Value,
    action_id: &str,
) -> Option<&'a Value> {
    workflow_root
        .get("workflows")
        .and_then(Value::as_array)?
        .iter()
        .find_map(|workflow| {
            workflow
                .get("transitions")
                .and_then(Value::as_array)?
                .iter()
                .find(|transition| {
                    transition.get("action_id").and_then(Value::as_str) == Some(action_id)
                })
        })
}

/// Returns one mutable array section, creating it when absent.
fn dynamic_state_rows_mut<'a>(root: &'a mut Value, section: &str) -> Option<&'a mut Vec<Value>> {
    root.as_object_mut()?
        .entry(section.to_owned())
        .or_insert_with(|| Value::Array(Vec::new()))
        .as_array_mut()
}

/// Returns the next global state id for domain records and advances the root counter.
fn dynamic_take_next_global_id(root: &mut Value) -> i64 {
    let next = root
        .get("next_id")
        .and_then(Value::as_i64)
        .filter(|value| *value > 0)
        .unwrap_or(1);
    if let Some(object) = root.as_object_mut() {
        object.insert("next_id".to_owned(), Value::from(next.saturating_add(1)));
    }
    next
}

/// Returns the next per-section identifier without consuming the root domain record counter.
fn dynamic_take_next_section_id(root: &mut Value, section: &str) -> i64 {
    root.get(section)
        .and_then(Value::as_array)
        .into_iter()
        .flatten()
        .filter_map(|row| row.get("id").and_then(Value::as_i64))
        .max()
        .unwrap_or(0)
        .saturating_add(1)
}

/// Reads one text-valued presentation field from a contract node.
fn dynamic_presentation_text(node: &Value, key: &str) -> String {
    node.get("presentation")
        .and_then(Value::as_object)
        .and_then(|presentation| presentation.get(key))
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_owned()
}

/// Reads one text field from a workflow transition presentation node.
fn dynamic_transition_presentation_text(node: Option<&Value>, key: &str) -> String {
    node.and_then(|transition| transition.get("presentation"))
        .and_then(Value::as_object)
        .and_then(|presentation| presentation.get(key))
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_owned()
}

/// Resolves one entity-field reference from the built app contract tree.
fn dynamic_resolve_ref_target(app: &AppRuntime, ref_text: &str) -> Option<Value> {
    let app_root = Path::new(app.app_root.as_deref()?);
    super::declarative::declarative_source_load_ref_target(app_root, ref_text)
}

/// Percent-encodes one bounded form or query component.
fn dynamic_url_encode_component(value: &str) -> String {
    const HEX: &[u8; 16] = b"0123456789ABCDEF";
    let mut encoded = String::with_capacity(value.len());
    for byte in value.bytes() {
        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~') {
            encoded.push(char::from(byte));
        } else {
            encoded.push('%');
            encoded.push(char::from(HEX[usize::from(byte >> 4)]));
            encoded.push(char::from(HEX[usize::from(byte & 0x0f)]));
        }
    }
    encoded
}

/// Builds one HTTP webhook payload for integration delivery.
fn dynamic_webhook_payload_json(adapter_id: &str, action_id: &str, resource_id: &str) -> String {
    serde_json::json!({
        "adapter_id": adapter_id,
        "action_id": action_id,
        "resource_id": resource_id,
    })
    .to_string()
}

/// Builds one bounded form-encoded HTTP payload from a flat JSON object.
fn dynamic_http_form_payload(payload_json: &str) -> Option<String> {
    let payload = serde_json::from_str::<Value>(payload_json).ok()?;
    let object = payload.as_object()?;
    let mut encoded = String::new();
    for (index, (key, value)) in object.iter().enumerate() {
        if key.is_empty() {
            return None;
        }
        let value_text = match value {
            Value::Null => String::new(),
            Value::Bool(boolean) => boolean.to_string(),
            Value::Number(number) => number.to_string(),
            Value::String(text) => text.clone(),
            _ => return None,
        };
        if index > 0 {
            encoded.push('&');
        }
        encoded.push_str(&dynamic_url_encode_component(key));
        encoded.push('=');
        encoded.push_str(&dynamic_url_encode_component(&value_text));
    }
    Some(encoded)
}

/// Resolves the outbound HTTP request body for one declared adapter kind.
fn dynamic_delivery_payload_body(adapter_kind: &str, payload_json: &str) -> Option<String> {
    match adapter_kind {
        "webhook" | "http_json" => Some(payload_json.to_owned()),
        "http_form" => dynamic_http_form_payload(payload_json),
        _ => None,
    }
}

/// Sends one bounded, policy-validated webhook POST.
fn dynamic_http_post_status(
    url: &str,
    content_type: &str,
    payload_body: &str,
    extra_headers: &[(&str, String)],
    timeout_seconds: u64,
    max_response_bytes: usize,
    allow_plaintext_local: bool,
) -> Option<u16> {
    let timeout = std::time::Duration::from_secs(timeout_seconds.clamp(1, 300));
    pharos_runtime_integrations::webhook::post_status(
        url,
        content_type,
        payload_body,
        extra_headers,
        pharos_runtime_integrations::webhook::WebhookLimits {
            connect_timeout: timeout.min(std::time::Duration::from_secs(30)),
            operation_timeout: timeout,
            max_response_bytes: max_response_bytes.clamp(64, 16 * 1024 * 1024),
            allow_plaintext_local,
        },
    )
}

/// Appends one workflow execution row and returns its id.
fn dynamic_append_workflow_execution(
    root: &mut Value,
    action_id: &str,
    transition: Option<&Value>,
) -> Option<i64> {
    let execution_id = dynamic_take_next_section_id(root, "workflow_executions");
    let workflow_id = transition
        .and_then(|value| value.get("action_id"))
        .and_then(Value::as_str)
        .unwrap_or(action_id)
        .to_owned();
    dynamic_state_rows_mut(root, "workflow_executions")?.push(serde_json::json!({
        "id": execution_id,
        "workflow_id": workflow_id,
        "transition_id": transition.and_then(|value| value.get("transition_id")).and_then(Value::as_str).unwrap_or_default(),
        "from_state": dynamic_transition_presentation_text(transition, "from_label"),
        "to_state": transition.and_then(|value| value.get("to_state")).and_then(Value::as_str).unwrap_or_default(),
        "execution_status": "applied",
        "queued_job_count": transition.and_then(|value| value.get("scheduled_jobs")).and_then(Value::as_array).map(|rows| rows.len()).unwrap_or(0),
        "created_at": now_iso_utc(),
        "completed_at": now_iso_utc(),
    }));
    Some(execution_id)
}

/// Appends two minimal workflow events for the applied transition.
fn dynamic_append_workflow_events(
    root: &mut Value,
    workflow_execution_id: i64,
    transition: Option<&Value>,
) -> bool {
    let first_id = dynamic_take_next_section_id(root, "workflow_execution_events");
    let second_id = first_id.saturating_add(1);
    let Some(rows) = dynamic_state_rows_mut(root, "workflow_execution_events") else {
        return false;
    };
    rows.push(serde_json::json!({
        "id": first_id,
        "workflow_execution_id": workflow_execution_id,
        "event_kind": "action_applied",
        "event_key": transition.and_then(|value| value.get("action_id")).and_then(Value::as_str).unwrap_or_default(),
        "created_at": now_iso_utc(),
    }));
    rows.push(serde_json::json!({
        "id": second_id,
        "workflow_execution_id": workflow_execution_id,
        "event_kind": transition.and_then(|value| value.get("emits_event")).and_then(Value::as_str).unwrap_or("action_event"),
        "event_key": transition.and_then(|value| value.get("transition_id")).and_then(Value::as_str).unwrap_or_default(),
        "created_at": now_iso_utc(),
    }));
    true
}

/// Appends one bounded workflow execution event row tied to a queued or processed job.
fn dynamic_append_workflow_execution_event(
    root: &mut Value,
    workflow_execution_id: i64,
    event_kind: &str,
    event_key: &str,
    payload: Value,
) -> bool {
    if workflow_execution_id <= 0
        || event_kind.is_empty()
        || event_kind.len() > 128
        || event_key.len() > 128
    {
        return false;
    }
    let event_id = dynamic_take_next_section_id(root, "workflow_execution_events");
    let Some(rows) = dynamic_state_rows_mut(root, "workflow_execution_events") else {
        return false;
    };
    rows.push(serde_json::json!({
        "id": event_id,
        "workflow_execution_id": workflow_execution_id,
        "event_kind": event_kind,
        "event_key": event_key,
        "payload_json": payload.to_string(),
        "created_at": now_iso_utc(),
    }));
    true
}

/// Appends one durable queued scheduled-job execution for the background worker.
fn dynamic_append_scheduled_job(
    root: &mut Value,
    workflow_execution_id: i64,
    job_id: &str,
    queue_name: &str,
    delay_seconds: i64,
    max_attempts: i64,
    backoff_seconds: i64,
    lease_seconds: i64,
) -> bool {
    if job_id.is_empty() || queue_name.is_empty() || queue_name.len() > 128 {
        return false;
    }
    let now_epoch = current_epoch_seconds();
    let scheduled_epoch = now_epoch.saturating_add(delay_seconds.max(0) as u64);
    let execution_id = dynamic_take_next_section_id(root, "job_executions");
    let Some(job_rows) = dynamic_state_rows_mut(root, "job_executions") else {
        return false;
    };
    job_rows.push(serde_json::json!({
        "id": execution_id,
        "workflow_execution_id": workflow_execution_id,
        "queue_name": queue_name,
        "job_id": job_id,
        "handler_id": job_id,
        "execution_status": "queued",
        "attempt_count": 0,
        "max_attempts": max_attempts.clamp(1, 100),
        "backoff_seconds": backoff_seconds.max(0),
        "lease_seconds": lease_seconds.clamp(1, 3600),
        "created_at": now_epoch.to_string(),
        "scheduled_for": scheduled_epoch.to_string(),
        "available_at": scheduled_epoch.to_string(),
        "idempotency_key": format!("{workflow_execution_id}:{job_id}"),
    }));
    let _ = dynamic_append_workflow_execution_event(
        root,
        workflow_execution_id,
        "job_queued",
        job_id,
        serde_json::json!({
            "job_id": job_id,
            "queue_name": queue_name,
            "delay_seconds": delay_seconds.max(0),
            "available_at": scheduled_epoch.to_string(),
        }),
    );
    true
}

/// Appends one local notification delivery row.
fn dynamic_append_notification(
    root: &mut Value,
    integrations_root: &Value,
    adapter_id: &str,
    channel: &str,
    recipient: &str,
    template_id: &str,
    template_params: &Value,
) -> bool {
    let adapter = integrations_root
        .get("adapters")
        .and_then(Value::as_array)
        .into_iter()
        .flatten()
        .find(|row| row.get("adapter_id").and_then(Value::as_str) == Some(adapter_id));
    let adapter_kind = adapter
        .and_then(|row| row.get("adapter_kind").and_then(Value::as_str))
        .unwrap_or_default();
    if adapter_id.is_empty()
        || recipient.is_empty()
        || template_id.is_empty()
        || !template_params.is_object()
        || !(adapter_kind == "notification_local" || adapter_kind == "notification_email")
    {
        return false;
    }
    let notification_id = dynamic_take_next_section_id(root, "notification_deliveries");
    let Some(rows) = dynamic_state_rows_mut(root, "notification_deliveries") else {
        return false;
    };
    rows.push(serde_json::json!({
        "id": notification_id,
        "adapter_id": adapter_id,
        "adapter_kind": adapter_kind,
        "delivery_kind": "action_notification",
        "channel": if channel.is_empty() { "email" } else { channel },
        "recipient": recipient,
        "template_id": template_id,
        "template_params": template_params,
        "delivery_status": "queued",
        "attempt_count": 0,
        "max_attempts": adapter.and_then(|row| row.get("max_attempts").and_then(Value::as_i64)).unwrap_or(if adapter_kind == "notification_email" { 2 } else { 1 }).clamp(1, 100),
        "backoff_seconds": adapter.and_then(|row| row.get("backoff_seconds").and_then(Value::as_i64)).unwrap_or(0).max(0),
        "relay": adapter.and_then(|row| row.get("relay").and_then(Value::as_str)).unwrap_or("primary"),
        "last_smtp_code": 0,
        "last_error": "",
        "next_attempt_at": "",
        "created_at": now_iso_utc(),
        "updated_at": now_iso_utc(),
        "completed_at": "",
    }));
    true
}

/// Performs dynamic integration HTTP adapter kind.
fn dynamic_integration_http_adapter_kind(adapter_kind: &str) -> bool {
    matches!(adapter_kind, "webhook" | "http_json" | "http_form")
}

/// Performs dynamic integration audit adapter kind.
fn dynamic_integration_audit_adapter_kind(adapter_kind: &str) -> bool {
    adapter_kind == "audit_local"
}

/// Performs dynamic integration file adapter kind.
fn dynamic_integration_file_adapter_kind(adapter_kind: &str) -> bool {
    adapter_kind == "append_jsonl_local" || adapter_kind == "write_json_local"
}

/// Performs dynamic delete records by scope primitive.
fn dynamic_delete_records_by_scope_primitive(
    root: &mut Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    primitive: &Value,
    matched_record_count: &mut i64,
    skipped_record_count: &mut i64,
    removed_record_count: &mut i64,
) -> Result<(), i32> {
    let collection = primitive
        .get("collection")
        .and_then(Value::as_str)
        .ok_or(422)?;
    let scope_field = primitive
        .get("scope_match")
        .and_then(Value::as_object)
        .and_then(|map| map.get("field"))
        .and_then(Value::as_str)
        .ok_or(422)?;
    let scope_input = primitive
        .get("scope_match")
        .and_then(Value::as_object)
        .and_then(|map| map.get("from_input"))
        .and_then(Value::as_str)
        .ok_or(422)?;
    let wanted =
        dynamic_action_input_value(request, input_overrides, scope_input).unwrap_or_default();
    let max_records = primitive
        .get("max_records")
        .and_then(Value::as_i64)
        .unwrap_or(i64::MAX)
        .max(0) as usize;
    let overflow_policy = primitive
        .get("overflow_policy")
        .and_then(Value::as_str)
        .unwrap_or("truncate");
    let Some(rows) = dynamic_state_rows_mut(root, collection) else {
        return Err(500);
    };
    let matching_indexes = rows
        .iter()
        .enumerate()
        .filter(|(_, row)| row.get(scope_field).map(value_to_text).unwrap_or_default() == wanted)
        .map(|(index, _)| index)
        .collect::<Vec<_>>();
    *matched_record_count += matching_indexes.len() as i64;
    if matching_indexes.len() > max_records && overflow_policy == "fail" {
        return Err(422);
    }
    *skipped_record_count += matching_indexes.len().saturating_sub(max_records) as i64;
    let mut removed = 0usize;
    let mut keep = Vec::with_capacity(rows.len());
    for (index, row) in rows.iter().cloned().enumerate() {
        let selected = matching_indexes
            .iter()
            .take(max_records)
            .any(|candidate| *candidate == index);
        if selected {
            removed += 1;
        } else {
            keep.push(row);
        }
    }
    *rows = keep;
    *removed_record_count += removed as i64;
    Ok(())
}

/// Finds one declared outbound adapter by adapter id.
fn dynamic_find_integration_adapter<'a>(
    integrations_root: &'a Value,
    adapter_id: &str,
) -> Option<&'a Value> {
    integrations_root
        .get("adapters")
        .and_then(Value::as_array)?
        .iter()
        .find(|row| row.get("adapter_id").and_then(Value::as_str) == Some(adapter_id))
}

/// Performs dynamic state rows.
fn dynamic_state_rows<'a>(root: &'a Value, section: &str) -> &'a [Value] {
    root.get(section)
        .and_then(Value::as_array)
        .map(Vec::as_slice)
        .unwrap_or(&[])
}

/// Performs dynamic row i64.
fn dynamic_row_i64(row: &Value, field: &str) -> i64 {
    row.get(field)
        .and_then(Value::as_i64)
        .or_else(|| {
            row.get(field)
                .and_then(Value::as_str)
                .and_then(|value| value.parse::<i64>().ok())
        })
        .unwrap_or(0)
}

/// Resolves one bounded notification parameter value from action input or runtime context.
fn dynamic_notification_parameter_value(
    spec: &Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    action_id: &str,
    entity_id: &str,
    event_kind: &str,
    recipient: &str,
    template_id: &str,
    anchor_record_id: i64,
) -> Option<Value> {
    let source = spec
        .get("source")
        .and_then(Value::as_str)
        .unwrap_or("literal");
    match source {
        "literal" => spec.get("value").cloned(),
        "input" => spec.get("field").and_then(Value::as_str).map(|field| {
            Value::String(
                dynamic_action_input_value(request, input_overrides, field).unwrap_or_default(),
            )
        }),
        "context_anchor_id" => Some(Value::from(anchor_record_id.max(0))),
        "action_id" => Some(Value::String(action_id.to_owned())),
        "entity_id" => Some(Value::String(entity_id.to_owned())),
        "event_kind" => Some(Value::String(event_kind.to_owned())),
        "recipient" => Some(Value::String(recipient.to_owned())),
        "template_id" => Some(Value::String(template_id.to_owned())),
        _ => None,
    }
}

/// Builds the flat template-parameter map for one declarative notification row.
fn dynamic_notification_template_params(
    notification: &Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    action_id: &str,
    entity_id: &str,
    event_kind: &str,
    recipient: &str,
    template_id: &str,
    anchor_record_id: i64,
) -> Value {
    let mut params = Map::new();
    params.insert("action_id".to_owned(), Value::String(action_id.to_owned()));
    params.insert("entity_id".to_owned(), Value::String(entity_id.to_owned()));
    params.insert(
        "event_kind".to_owned(),
        Value::String(event_kind.to_owned()),
    );
    params.insert("recipient".to_owned(), Value::String(recipient.to_owned()));
    params.insert(
        "template_id".to_owned(),
        Value::String(template_id.to_owned()),
    );
    params.insert(
        "anchor_record_id".to_owned(),
        Value::from(anchor_record_id.max(0)),
    );
    if let Some(explicit) = notification.get("params").and_then(Value::as_object) {
        for (key, spec) in explicit {
            if key.is_empty() || key.len() > 128 {
                continue;
            }
            if let Some(value) = dynamic_notification_parameter_value(
                spec,
                request,
                input_overrides,
                action_id,
                entity_id,
                event_kind,
                recipient,
                template_id,
                anchor_record_id,
            ) {
                params.insert(key.to_owned(), value);
            }
        }
    }
    Value::Object(params)
}

/// Performs dynamic notification template ID from row.
fn dynamic_notification_template_id_from_row(row: &Value) -> Option<&str> {
    if let Some(template_id) = row.get("template_id").and_then(Value::as_str) {
        return (!template_id.is_empty()).then_some(template_id);
    }
    match row
        .get("delivery_kind")
        .and_then(Value::as_str)
        .unwrap_or_default()
    {
        "business_invite" => Some("business_invite"),
        "password_reset" => Some("password_reset"),
        _ => None,
    }
}

/// Performs dynamic notification origin.
fn dynamic_notification_origin(app: &AppRuntime) -> Option<String> {
    let app_root = Path::new(app.app_root.as_deref()?);
    let runtime_conf = app.runtime_conf_path.as_deref().map(Path::new);
    json_commands::dynamic_config_value(app_root, runtime_conf, "base_url")
        .and_then(|value| normalized_public_origin(&value))
}

/// Performs dynamic notification absolute URL.
fn dynamic_notification_absolute_url(app: &AppRuntime, relative_path: &str) -> String {
    let scoped_relative = format!(
        "{}{}",
        app.base_path
            .as_deref()
            .filter(|value| !value.is_empty() && *value != "/")
            .unwrap_or(""),
        relative_path
    );
    dynamic_notification_origin(app)
        .map(|origin| format!("{origin}{scoped_relative}"))
        .unwrap_or(scoped_relative)
}

/// Performs dynamic framework notification params.
fn dynamic_framework_notification_params(app: &AppRuntime, root: &Value, row: &Value) -> Value {
    let mut params = Map::new();
    params.insert(
        "app_id".to_owned(),
        Value::String(app.app_id.clone().unwrap_or_default()),
    );
    let recipient = row
        .get("recipient")
        .and_then(Value::as_str)
        .unwrap_or_default();
    params.insert("recipient".to_owned(), Value::String(recipient.to_owned()));
    match row
        .get("delivery_kind")
        .and_then(Value::as_str)
        .unwrap_or_default()
    {
        "business_invite" => {
            let secret = row
                .get("secret")
                .and_then(Value::as_str)
                .unwrap_or_default();
            params.insert(
                "invite_url".to_owned(),
                Value::String(dynamic_notification_absolute_url(
                    app,
                    &format!("/accounts/invite/{secret}/"),
                )),
            );
            let token_hash = hex_digest(&Sha256::digest(secret.as_bytes()));
            if let Some(invite) =
                dynamic_state_rows(root, "business_invites")
                    .iter()
                    .find(|invite| {
                        invite.get("token_hash").and_then(Value::as_str)
                            == Some(token_hash.as_str())
                    })
            {
                let business_id = dynamic_row_i64(invite, "business_id");
                let business_name = dynamic_state_rows(root, "businesses")
                    .iter()
                    .find(|business| dynamic_row_i64(business, "id") == business_id)
                    .and_then(|business| business.get("name").and_then(Value::as_str))
                    .unwrap_or("the business");
                let account_type = invite
                    .get("account_type")
                    .and_then(Value::as_str)
                    .map(super::business::dynamic_invite_account_type_label)
                    .unwrap_or("member");
                params.insert(
                    "business_name".to_owned(),
                    Value::String(business_name.to_owned()),
                );
                params.insert(
                    "account_type".to_owned(),
                    Value::String(account_type.to_owned()),
                );
            }
        }
        "password_reset" => {
            let secret = row
                .get("secret")
                .and_then(Value::as_str)
                .unwrap_or_default();
            let token_hash = hex_digest(&Sha256::digest(secret.as_bytes()));
            let password_reset_user_id = dynamic_state_rows(root, "password_resets")
                .iter()
                .find(|reset| {
                    reset.get("token_hash").and_then(Value::as_str) == Some(token_hash.as_str())
                })
                .map(|reset| dynamic_row_i64(reset, "user_id"))
                .unwrap_or(0);
            params.insert(
                "reset_url".to_owned(),
                Value::String(dynamic_notification_absolute_url(
                    app,
                    &format!("/accounts/password-reset/complete/{secret}/"),
                )),
            );
            let full_name = dynamic_state_rows(root, "users")
                .iter()
                .find(|user| {
                    user.get("email").and_then(Value::as_str) == Some(recipient)
                        || dynamic_row_i64(user, "id") == password_reset_user_id
                })
                .and_then(|user| {
                    user.get("display_name")
                        .and_then(Value::as_str)
                        .filter(|value| !value.is_empty())
                        .or_else(|| user.get("username").and_then(Value::as_str))
                })
                .unwrap_or("there");
            params.insert("full_name".to_owned(), Value::String(full_name.to_owned()));
        }
        _ => {}
    }
    Value::Object(params)
}

/// Performs dynamic update notification delivery.
fn dynamic_update_notification_delivery(
    delivery: &mut Value,
    attempt_number: i64,
    status: &str,
    last_error: &str,
    next_attempt_at: Option<u64>,
    last_smtp_code: i64,
    completed: bool,
    rendered_subject: Option<&str>,
    rendered_body: Option<&str>,
) {
    let now = now_iso_utc();
    if let Some(object) = delivery.as_object_mut() {
        object.insert(
            "attempt_count".to_owned(),
            Value::from(attempt_number.max(0)),
        );
        object.insert(
            "delivery_status".to_owned(),
            Value::String(status.to_owned()),
        );
        object.insert("updated_at".to_owned(), Value::String(now.clone()));
        object.insert(
            "last_error".to_owned(),
            Value::String(last_error.to_owned()),
        );
        object.insert(
            "last_smtp_code".to_owned(),
            Value::from(last_smtp_code.max(0)),
        );
        object.insert(
            "next_attempt_at".to_owned(),
            Value::String(
                next_attempt_at
                    .map(|value| value.to_string())
                    .unwrap_or_default(),
            ),
        );
        if completed {
            object.insert("completed_at".to_owned(), Value::String(now));
        }
        if let Some(subject) = rendered_subject {
            object.insert(
                "rendered_subject".to_owned(),
                Value::String(subject.to_owned()),
            );
        }
        if let Some(body) = rendered_body {
            object.insert("rendered_body".to_owned(), Value::String(body.to_owned()));
        }
    }
}

/// Processes one due queued or retryable notification delivery in a worker-owned state transaction.
///
/// # Parameters
///
/// - `app`: Input consumed by [`crate::runtime::dynamic::process_one_due_notification_delivery`].
/// - `root`: Input consumed by [`crate::runtime::dynamic::process_one_due_notification_delivery`].
/// - `integrations_root`: Input consumed by [`crate::runtime::dynamic::process_one_due_notification_delivery`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::process_one_due_notification_delivery`] holds and `false` otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::process_one_due_notification_delivery`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::process_one_due_notification_delivery(todo!(), std::path::Path::new("."), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn process_one_due_notification_delivery(
    app: &AppRuntime,
    root: &mut Value,
    integrations_root: &Value,
) -> bool {
    let now = current_epoch_seconds();
    let delivery_id = root
        .get("notification_deliveries")
        .and_then(Value::as_array)
        .and_then(|rows| {
            rows.iter().find(|row| {
                row.get("delivery_status").and_then(Value::as_str) == Some("queued")
                    || (row.get("delivery_status").and_then(Value::as_str) == Some("retryable")
                        && row
                            .get("next_attempt_at")
                            .and_then(Value::as_str)
                            .and_then(|value| value.parse::<u64>().ok())
                            .is_some_and(|available| available <= now))
            })
        })
        .and_then(|row| row.get("id").and_then(Value::as_i64));
    let Some(delivery_id) = delivery_id else {
        return false;
    };
    let delivery_snapshot = root
        .get("notification_deliveries")
        .and_then(Value::as_array)
        .and_then(|rows| {
            rows.iter()
                .find(|row| row.get("id").and_then(Value::as_i64) == Some(delivery_id))
        })
        .cloned();
    let framework_params = delivery_snapshot
        .as_ref()
        .map(|row| dynamic_framework_notification_params(app, root, row));
    let Some(rows) = root
        .get_mut("notification_deliveries")
        .and_then(Value::as_array_mut)
    else {
        return false;
    };
    let Some(delivery) = rows
        .iter_mut()
        .find(|row| row.get("id").and_then(Value::as_i64) == Some(delivery_id))
    else {
        return false;
    };
    let adapter_id = delivery
        .get("adapter_id")
        .and_then(Value::as_str)
        .unwrap_or_default();
    let adapter_kind = delivery
        .get("adapter_kind")
        .and_then(Value::as_str)
        .filter(|value| !value.is_empty())
        .map(ToOwned::to_owned)
        .or_else(|| {
            dynamic_find_integration_adapter(integrations_root, adapter_id)
                .and_then(|adapter| adapter.get("adapter_kind").and_then(Value::as_str))
                .map(ToOwned::to_owned)
        })
        .unwrap_or_else(|| {
            if delivery
                .get("delivery_kind")
                .and_then(Value::as_str)
                .is_some()
            {
                "notification_email".to_owned()
            } else {
                "notification_local".to_owned()
            }
        });
    let recipient = delivery
        .get("recipient")
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_owned();
    let template_id = dynamic_notification_template_id_from_row(delivery)
        .unwrap_or_default()
        .to_owned();
    let params = delivery
        .get("template_params")
        .filter(|value| value.is_object())
        .cloned()
        .or(framework_params)
        .unwrap_or_else(|| serde_json::json!({}));
    let attempt_number = delivery
        .get("attempt_count")
        .and_then(Value::as_i64)
        .unwrap_or(0)
        .saturating_add(1);
    let max_attempts = delivery
        .get("max_attempts")
        .and_then(Value::as_i64)
        .unwrap_or(if adapter_kind == "notification_email" {
            2
        } else {
            1
        })
        .max(1);
    let backoff_seconds = delivery
        .get("backoff_seconds")
        .and_then(Value::as_i64)
        .unwrap_or(0)
        .max(0) as u64;
    if recipient.is_empty() || template_id.is_empty() {
        dynamic_update_notification_delivery(
            delivery,
            attempt_number,
            "failed",
            "notification delivery is incomplete",
            None,
            0,
            true,
            None,
            None,
        );
        return true;
    }
    let Some(message) =
        super::business::load_notification_email_message(app, &recipient, &template_id, &params)
    else {
        dynamic_update_notification_delivery(
            delivery,
            attempt_number,
            "failed",
            "notification template is unavailable",
            None,
            0,
            true,
            None,
            None,
        );
        return true;
    };
    match adapter_kind.as_str() {
        "notification_local" => {
            dynamic_update_notification_delivery(
                delivery,
                attempt_number,
                "succeeded",
                "",
                None,
                0,
                true,
                Some(&message.subject),
                Some(&message.text_body),
            );
            true
        }
        "notification_email" => {
            let relay_prefix = if delivery.get("relay").and_then(Value::as_str) == Some("secondary")
            {
                Some("smtp2_")
            } else {
                Some("smtp_")
            };
            let config = super::business::email_load_relay_config(
                Path::new(app.app_root.as_deref().unwrap_or_default()),
                app.runtime_conf_path.as_deref().map(Path::new),
                relay_prefix,
            );
            let Ok(config) = config else {
                dynamic_update_notification_delivery(
                    delivery,
                    attempt_number,
                    "skipped",
                    "smtp configuration unavailable",
                    None,
                    0,
                    true,
                    Some(&message.subject),
                    Some(&message.text_body),
                );
                return true;
            };
            let mut result = super::business::EmailResult::default();
            let _ = super::business::email_send_text(&config, &message, &mut result);
            if result.delivered {
                dynamic_update_notification_delivery(
                    delivery,
                    attempt_number,
                    "succeeded",
                    "",
                    None,
                    result.smtp_code,
                    true,
                    Some(&message.subject),
                    Some(&message.text_body),
                );
            } else if result.retryable && attempt_number < max_attempts {
                dynamic_update_notification_delivery(
                    delivery,
                    attempt_number,
                    "retryable",
                    result
                        .error_message
                        .as_deref()
                        .unwrap_or("smtp delivery failed"),
                    Some(now.saturating_add(backoff_seconds.max(1))),
                    result.smtp_code,
                    false,
                    Some(&message.subject),
                    Some(&message.text_body),
                );
            } else {
                let status = if result.skipped { "skipped" } else { "failed" };
                dynamic_update_notification_delivery(
                    delivery,
                    attempt_number,
                    status,
                    result
                        .error_message
                        .as_deref()
                        .unwrap_or("smtp delivery failed"),
                    None,
                    result.smtp_code,
                    true,
                    Some(&message.subject),
                    Some(&message.text_body),
                );
            }
            true
        }
        _ => {
            dynamic_update_notification_delivery(
                delivery,
                attempt_number,
                "failed",
                "unsupported notification adapter kind",
                None,
                0,
                true,
                None,
                None,
            );
            true
        }
    }
}

/// Appends one webhook delivery row for the durable background worker.
fn dynamic_append_webhook_delivery(
    root: &mut Value,
    integrations_root: &Value,
    adapter_id: &str,
    action_id: &str,
    resource_id: &str,
) -> Option<()> {
    let adapter = integrations_root
        .get("adapters")
        .and_then(Value::as_array)?
        .iter()
        .find(|row| row.get("adapter_id").and_then(Value::as_str) == Some(adapter_id))?;
    let adapter_kind = adapter
        .get("adapter_kind")
        .and_then(Value::as_str)
        .unwrap_or_default();
    if !dynamic_integration_http_adapter_kind(adapter_kind) {
        return None;
    }
    let target_url = adapter
        .get("target_url")
        .and_then(Value::as_str)
        .filter(|value| !value.is_empty())
        .map(ToOwned::to_owned)
        .or_else(|| {
            adapter
                .get("target_url_env")
                .and_then(Value::as_str)
                .and_then(|name| env::var(name).ok())
                .filter(|value| !value.is_empty())
        })
        .unwrap_or_default();
    let content_type = adapter
        .get("content_type")
        .and_then(Value::as_str)
        .unwrap_or("application/json");
    let payload_json = dynamic_webhook_payload_json(adapter_id, action_id, resource_id);
    let delivery_id = dynamic_take_next_section_id(root, "integration_deliveries");
    dynamic_state_rows_mut(root, "integration_deliveries")?.push(serde_json::json!({
        "id": delivery_id,
        "adapter_id": adapter_id,
        "adapter_kind": adapter_kind,
        "event_kind": action_id,
        "resource_id": resource_id,
        "subject_id": "",
        "tenant_id": "",
        "target_url": target_url,
        "content_type": content_type,
        "payload_json": payload_json,
        "event_header": adapter.get("event_header").and_then(Value::as_str).unwrap_or_default(),
        "auth_header": adapter.get("auth_header").and_then(Value::as_str).unwrap_or_default(),
        "auth_scheme": adapter.get("auth_scheme").and_then(Value::as_str).unwrap_or_default(),
        "auth_token_env": adapter.get("auth_token_env").and_then(Value::as_str).unwrap_or_default(),
        "delivery_status": "queued",
        "attempt_count": 0,
        "max_attempts": adapter.get("max_attempts").and_then(Value::as_i64).unwrap_or(1),
        "backoff_seconds": adapter.get("backoff_seconds").and_then(Value::as_i64).unwrap_or(0),
        "timeout_seconds": adapter.get("timeout_seconds").and_then(Value::as_i64).unwrap_or(0),
        "max_response_bytes": adapter.get("max_response_bytes").and_then(Value::as_i64).unwrap_or(8192),
        "allow_plaintext_local": adapter.get("allow_plaintext_local").and_then(Value::as_bool).unwrap_or(false),
        "last_http_status": 0,
        "last_error": "",
        "next_attempt_at": "",
        "completed_at": "",
        "created_at": now_iso_utc(),
        "updated_at": now_iso_utc(),
    }));
    Some(())
}

/// Processes one queued or retryable webhook delivery in place.
fn dynamic_process_webhook_delivery(root: &mut Value, delivery_id: i64) -> bool {
    let Some(delivery) = dynamic_state_rows_mut(root, "integration_deliveries").and_then(|rows| {
        rows.iter_mut()
            .find(|row| row.get("id").and_then(Value::as_i64) == Some(delivery_id))
    }) else {
        return false;
    };
    let target_url = delivery
        .get("target_url")
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_owned();
    let content_type = delivery
        .get("content_type")
        .and_then(Value::as_str)
        .unwrap_or("application/json")
        .to_owned();
    let event_kind = delivery
        .get("event_kind")
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_owned();
    let adapter_kind = delivery
        .get("adapter_kind")
        .and_then(Value::as_str)
        .unwrap_or("webhook")
        .to_owned();
    let payload_json = delivery
        .get("payload_json")
        .map(|payload| {
            payload.as_str().map(ToOwned::to_owned).unwrap_or_else(|| {
                serde_json::to_string(payload).unwrap_or_else(|_| "{}".to_owned())
            })
        })
        .unwrap_or_else(|| "{}".to_owned());
    let attempt_number = delivery
        .get("attempt_count")
        .and_then(Value::as_i64)
        .unwrap_or(0)
        .saturating_add(1);
    let max_attempts = delivery
        .get("max_attempts")
        .and_then(Value::as_i64)
        .unwrap_or(1)
        .max(1);
    let backoff_seconds = delivery
        .get("backoff_seconds")
        .and_then(Value::as_i64)
        .unwrap_or(0)
        .max(0);
    let now = now_iso_utc();
    let timestamp = now_iso_utc();
    let delivery_id_text = delivery_id.to_string();
    let signature = delivery
        .get("signature_secret_env")
        .and_then(Value::as_str)
        .and_then(|secret_env| {
            webhook_signature_header_value_dup_logic_local(secret_env, &timestamp, &payload_json)
        });
    let signature_header = delivery
        .get("signature_header")
        .and_then(Value::as_str)
        .unwrap_or("X-Pharos-Signature")
        .to_owned();
    let timestamp_header = delivery
        .get("timestamp_header")
        .and_then(Value::as_str)
        .unwrap_or("X-Pharos-Timestamp")
        .to_owned();
    let mut headers = vec![
        ("X-Pharos-Delivery-Id", delivery_id_text),
        (timestamp_header.as_str(), timestamp.clone()),
    ];
    let event_header = delivery
        .get("event_header")
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_owned();
    if !event_header.is_empty() && !event_kind.is_empty() {
        headers.push((event_header.as_str(), event_kind.clone()));
    }
    let auth_header = delivery
        .get("auth_header")
        .and_then(Value::as_str)
        .filter(|value| !value.is_empty())
        .unwrap_or("Authorization")
        .to_owned();
    let auth_token_env = delivery
        .get("auth_token_env")
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_owned();
    let auth_scheme = delivery
        .get("auth_scheme")
        .and_then(Value::as_str)
        .unwrap_or("Bearer")
        .to_owned();
    let auth_required = !auth_token_env.is_empty();
    let auth_header_value = (!auth_token_env.is_empty())
        .then(|| env::var(&auth_token_env).ok())
        .flatten()
        .filter(|value| !value.is_empty())
        .map(|token| {
            if auth_scheme.is_empty() {
                token
            } else {
                format!("{auth_scheme} {token}")
            }
        });
    if let Some(auth_value) = auth_header_value.as_ref() {
        headers.push((auth_header.as_str(), auth_value.clone()));
    }
    if let Some(signature_value) = signature.as_ref() {
        headers.push((signature_header.as_str(), signature_value.clone()));
    }
    let signature_required = delivery
        .get("signature_secret_env")
        .and_then(Value::as_str)
        .is_some_and(|value| !value.is_empty());
    let payload_body = dynamic_delivery_payload_body(&adapter_kind, &payload_json);
    let delivery_error = if !dynamic_integration_http_adapter_kind(&adapter_kind) {
        "unsupported integration adapter kind"
    } else if payload_body.is_none() {
        "integration adapter payload is incompatible"
    } else if auth_required && auth_header_value.is_none() {
        "integration adapter auth unavailable"
    } else if signature_required && signature.is_none() {
        "webhook signature secret unavailable"
    } else {
        ""
    };
    let status_code = if !delivery_error.is_empty() {
        None
    } else {
        dynamic_http_post_status(
            &target_url,
            &content_type,
            payload_body.as_deref().unwrap_or_default(),
            &headers,
            delivery
                .get("timeout_seconds")
                .and_then(Value::as_u64)
                .unwrap_or(15),
            delivery
                .get("max_response_bytes")
                .and_then(Value::as_u64)
                .unwrap_or(8 * 1024) as usize,
            delivery
                .get("allow_plaintext_local")
                .and_then(Value::as_bool)
                .unwrap_or(false),
        )
    };
    let success = status_code.is_some_and(|status| (200..300).contains(&status));
    if let Some(object) = delivery.as_object_mut() {
        object.insert("attempt_count".to_owned(), Value::from(attempt_number));
        object.insert("updated_at".to_owned(), Value::String(now.clone()));
        object.insert(
            "last_http_status".to_owned(),
            Value::from(status_code.unwrap_or(0) as i64),
        );
        if success {
            object.insert(
                "delivery_status".to_owned(),
                Value::String("succeeded".to_owned()),
            );
            object.insert("completed_at".to_owned(), Value::String(now.clone()));
            object.insert("next_attempt_at".to_owned(), Value::String(String::new()));
            object.insert("last_error".to_owned(), Value::String(String::new()));
        } else if attempt_number < max_attempts {
            object.insert(
                "delivery_status".to_owned(),
                Value::String("retryable".to_owned()),
            );
            object.insert(
                "next_attempt_at".to_owned(),
                Value::String((now.parse::<i64>().unwrap_or(0) + backoff_seconds).to_string()),
            );
            object.insert(
                "last_error".to_owned(),
                Value::String(if delivery_error.is_empty() {
                    "outbound webhook transport unavailable".to_owned()
                } else {
                    delivery_error.to_owned()
                }),
            );
        } else {
            object.insert(
                "delivery_status".to_owned(),
                Value::String("failed".to_owned()),
            );
            object.insert("next_attempt_at".to_owned(), Value::String(String::new()));
            object.insert(
                "last_error".to_owned(),
                Value::String(if delivery_error.is_empty() {
                    "outbound webhook transport unavailable".to_owned()
                } else {
                    delivery_error.to_owned()
                }),
            );
        }
    }
    let attempt_id = dynamic_take_next_section_id(root, "integration_delivery_attempts");
    if let Some(rows) = dynamic_state_rows_mut(root, "integration_delivery_attempts") {
        rows.push(serde_json::json!({
            "id": attempt_id,
            "delivery_id": delivery_id,
            "attempt_status": if success { "succeeded" } else { "failed" },
            "http_status": status_code.unwrap_or(0),
            "error_message": if success {
                ""
            } else if delivery_error.is_empty() {
                "outbound webhook transport unavailable"
            } else {
                delivery_error
            },
            "started_at": now,
            "finished_at": now_iso_utc(),
        }));
    }
    success
}

/// Queues one declaratively configured webhook delivery for background execution.
///
/// # Parameters
///
/// - `root`: Input consumed by [`crate::runtime::dynamic::deliver_integration_webhook`].
/// - `integrations_root`: Input consumed by [`crate::runtime::dynamic::deliver_integration_webhook`].
/// - `adapter_id`: Input consumed by [`crate::runtime::dynamic::deliver_integration_webhook`].
/// - `event_kind`: Input consumed by [`crate::runtime::dynamic::deliver_integration_webhook`].
/// - `resource_id`: Input consumed by [`crate::runtime::dynamic::deliver_integration_webhook`].
/// - `tenant_id`: Input consumed by [`crate::runtime::dynamic::deliver_integration_webhook`].
/// - `subject_id`: Input consumed by [`crate::runtime::dynamic::deliver_integration_webhook`].
/// - `payload_json`: Input consumed by [`crate::runtime::dynamic::deliver_integration_webhook`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::deliver_integration_webhook`] can produce a value and [`None`] otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::deliver_integration_webhook`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::deliver_integration_webhook(std::path::Path::new("."), todo!(), todo!(), todo!(), todo!(), todo!(), todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn deliver_integration_webhook(
    root: &mut Value,
    integrations_root: &Value,
    adapter_id: &str,
    event_kind: Option<&str>,
    resource_id: Option<&str>,
    tenant_id: Option<&str>,
    subject_id: Option<&str>,
    payload_json: &str,
) -> Option<i64> {
    if adapter_id.is_empty()
        || payload_json.len() > 256 * 1024
        || !serde_json::from_str::<Value>(payload_json)
            .ok()?
            .is_object()
    {
        return None;
    }
    let adapter = integrations_root
        .get("adapters")
        .and_then(Value::as_array)?
        .iter()
        .find(|row| row.get("adapter_id").and_then(Value::as_str) == Some(adapter_id))?;
    let adapter_kind = adapter
        .get("adapter_kind")
        .and_then(Value::as_str)
        .unwrap_or_default();
    if !dynamic_integration_http_adapter_kind(adapter_kind) {
        return None;
    }
    let target_url = adapter
        .get("target_url")
        .and_then(Value::as_str)
        .filter(|value| !value.is_empty())
        .map(ToOwned::to_owned)
        .or_else(|| {
            adapter
                .get("target_url_env")
                .and_then(Value::as_str)
                .and_then(|name| env::var(name).ok())
                .filter(|value| !value.is_empty())
        })?;
    let signature = adapter.get("signature").and_then(Value::as_object);
    let delivery_id = dynamic_take_next_section_id(root, "integration_deliveries");
    dynamic_state_rows_mut(root, "integration_deliveries")?.push(serde_json::json!({
        "id": delivery_id,
        "adapter_id": adapter_id,
        "adapter_kind": adapter_kind,
        "event_kind": event_kind.filter(|value| !value.is_empty()).unwrap_or("integration_event"),
        "resource_id": resource_id.unwrap_or_default(),
        "tenant_id": tenant_id.unwrap_or_default(),
        "subject_id": subject_id.unwrap_or_default(),
        "target_url": target_url,
        "content_type": adapter.get("content_type").and_then(Value::as_str).unwrap_or("application/json"),
        "payload_json": payload_json,
        "event_header": adapter.get("event_header").and_then(Value::as_str).unwrap_or_default(),
        "auth_header": adapter.get("auth_header").and_then(Value::as_str).unwrap_or_default(),
        "auth_scheme": adapter.get("auth_scheme").and_then(Value::as_str).unwrap_or_default(),
        "auth_token_env": adapter.get("auth_token_env").and_then(Value::as_str).unwrap_or_default(),
        "signature_header": signature.and_then(|value| value.get("header")).and_then(Value::as_str).unwrap_or("X-Pharos-Signature"),
        "timestamp_header": signature.and_then(|value| value.get("timestamp_header")).and_then(Value::as_str).unwrap_or("X-Pharos-Timestamp"),
        "signature_secret_env": signature.and_then(|value| value.get("secret_env")).and_then(Value::as_str).unwrap_or_default(),
        "delivery_status": "queued",
        "attempt_count": 0,
        "max_attempts": adapter.get("max_attempts").and_then(Value::as_i64).unwrap_or(1).max(1),
        "backoff_seconds": adapter.get("backoff_seconds").and_then(Value::as_i64).unwrap_or(0).max(0),
        "timeout_seconds": adapter.get("timeout_seconds").and_then(Value::as_i64).unwrap_or(0).max(0),
        "max_response_bytes": adapter.get("max_response_bytes").and_then(Value::as_u64).unwrap_or(8 * 1024).clamp(64, 16 * 1024 * 1024),
        "allow_plaintext_local": adapter.get("allow_plaintext_local").and_then(Value::as_bool).unwrap_or(false),
        "last_http_status": 0,
        "last_error": "",
        "next_attempt_at": "",
        "completed_at": "",
        "created_at": now_iso_utc(),
        "updated_at": now_iso_utc(),
    }));
    Some(delivery_id)
}

/// Processes one due queued or retryable integration delivery in a worker-owned state transaction.
///
/// # Parameters
///
/// - `root`: Input consumed by [`crate::runtime::dynamic::process_one_due_integration_delivery`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::process_one_due_integration_delivery`] holds and `false` otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::process_one_due_integration_delivery`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::process_one_due_integration_delivery(std::path::Path::new("."));
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn process_one_due_integration_delivery(root: &mut Value) -> bool {
    let now = current_epoch_seconds();
    let delivery_id = root
        .get("integration_deliveries")
        .and_then(Value::as_array)
        .and_then(|rows| {
            rows.iter().find(|row| {
                row.get("delivery_status").and_then(Value::as_str) == Some("queued")
                    || (row.get("delivery_status").and_then(Value::as_str) == Some("retryable")
                        && row
                            .get("next_attempt_at")
                            .and_then(Value::as_str)
                            .and_then(|value| value.parse::<u64>().ok())
                            .is_some_and(|available| available <= now))
            })
        })
        .and_then(|row| row.get("id").and_then(Value::as_i64));
    delivery_id.is_some_and(|delivery_id| {
        let _ = dynamic_process_webhook_delivery(root, delivery_id);
        true
    })
}

/// Appends the last action report and mirrors it into the action report list.
fn dynamic_store_last_action_report(root: &mut Value, mut report: Value) -> bool {
    let report_id = dynamic_take_next_section_id(root, "action_reports");
    if let Some(object) = report.as_object_mut() {
        object.insert("id".to_owned(), Value::from(report_id));
    }
    let Some(object) = root.as_object_mut() else {
        return false;
    };
    object.insert("last_action_report".to_owned(), report.clone());
    object.insert(
        "last_action".to_owned(),
        report
            .get("event_kind")
            .cloned()
            .unwrap_or(Value::String(String::new())),
    );
    dynamic_state_rows_mut(root, "action_reports").is_some_and(|rows| {
        rows.push(report);
        true
    })
}

/// Builds the failure report without selecting or writing a persistence medium.
fn dynamic_action_failure_report_root(
    mut root: Value,
    action: &Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    action_id: &str,
    entity_id: &str,
    transition: Option<&Value>,
    anchor_record_id: i64,
    status: i32,
) -> Value {
    let event_kind = format!("{action_id}_failed");
    let mut report = serde_json::json!({
        "action_id": action_id,
        "event_kind": event_kind,
        "execution_status": "failed",
        "failure_status": status,
        "failure_code": dynamic_action_failure_code(status),
        "created_record_count": 0,
        "updated_record_count": 0,
        "matched_record_count": 0,
        "skipped_record_count": 0,
        "removed_record_count": 0,
        "queued_notification_count": 0,
        "queued_webhook_count": 0,
        "emitted_audit_count": 0,
        "emitted_file_record_count": 0,
        "delivered_webhook_count": 0,
        "retryable_webhook_count": 0,
        "failed_webhook_count": 0,
        "retried_delivery_count": 0,
        "anchor_record_id": anchor_record_id,
        "operator_message": dynamic_presentation_text(action, "failure_message"),
        "operator_summary": dynamic_presentation_text(action, "operator_summary"),
        "transition_label": dynamic_transition_presentation_text(transition, "transition_label"),
        "to_state": transition.and_then(|value| value.get("to_state")).and_then(Value::as_str).unwrap_or_default(),
        "overflow_policy": action
            .get("mutation_primitives")
            .and_then(Value::as_array)
            .and_then(|rows| rows.iter().find_map(|row| row.get("overflow_policy").and_then(Value::as_str)))
            .unwrap_or_default(),
    });
    let annotations = dynamic_action_failure_annotations(
        action,
        request,
        input_overrides,
        action_id,
        entity_id,
        &event_kind,
        anchor_record_id,
    )
    .unwrap_or_default();
    if let Some(object) = report.as_object_mut()
        && !annotations.is_empty()
    {
        object.insert("annotations".to_owned(), Value::Object(annotations));
    }
    let _ = dynamic_store_last_action_report(&mut root, report);
    root
}

/// Declared document-action response captured from the successfully committed view.
pub enum DeclarativeActionResponse {
    Redirect(String),
    Json(u16, Value),
}

/// Executes the bounded declarative dynamic-app action contract against persisted state.
///
/// # Parameters
///
/// - `app`: Input consumed by [`crate::runtime::dynamic::dynamic_execute_declarative_action`].
/// - `state_path`: Input consumed by [`crate::runtime::dynamic::dynamic_execute_declarative_action`].
/// - `request`: Input consumed by [`crate::runtime::dynamic::dynamic_execute_declarative_action`].
/// - `action_id`: Input consumed by [`crate::runtime::dynamic::dynamic_execute_declarative_action`].
///
/// # Returns
///
/// Returns the successful value produced by [`crate::runtime::dynamic::dynamic_execute_declarative_action`] or an error when the operation fails.
///
/// # Errors
///
/// Returns an error value when [`crate::runtime::dynamic::dynamic_execute_declarative_action`] cannot complete successfully.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::dynamic_execute_declarative_action`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::dynamic_execute_declarative_action(todo!(), todo!(), todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn dynamic_execute_declarative_action(
    app: &AppRuntime,
    state_path: Option<&Path>,
    request: &HttpRequest,
    action_id: &str,
) -> Result<DeclarativeActionResponse, i32> {
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(3);
    for attempt in 0..32 {
        let mut rejected_write = false;
        let result = dynamic_execute_declarative_action_attempt(
            app,
            state_path,
            request,
            action_id,
            &mut rejected_write,
        );
        if !rejected_write || attempt == 31 || std::time::Instant::now() >= deadline {
            return result;
        }
        // The compare-and-write was definitely rejected. Re-read and re-plan
        // without delivering external effects or replaying uncertain writes.
        std::thread::sleep(std::time::Duration::from_millis(25 + attempt * 5));
    }
    Err(409)
}

/// Plans one document action; marks only definitely rejected commits for retry.
fn dynamic_execute_declarative_action_attempt(
    app: &AppRuntime,
    state_path: Option<&Path>,
    request: &HttpRequest,
    action_id: &str,
    rejected_write: &mut bool,
) -> Result<DeclarativeActionResponse, i32> {
    if action_id.is_empty() || app.app_id.as_deref().is_none_or(str::is_empty) {
        return Err(400);
    }
    let actions_root = dynamic_contract_doc(app, "action_contract").ok_or(404)?;
    let workflows_root = dynamic_contract_doc(app, "workflow_contract").ok_or(404)?;
    let integrations_root = dynamic_contract_doc(app, "integration_contract")
        .unwrap_or_else(|| serde_json::json!({"adapters": []}));
    let action = dynamic_find_action_contract(&actions_root, action_id).ok_or(404)?;
    let input_overrides = dynamic_action_input_overrides(action, request)?;
    let response = action.get("response_contract").ok_or(422)?;
    let response_kind = response
        .get("response_kind")
        .and_then(Value::as_str)
        .unwrap_or("redirect");
    if !matches!(response_kind, "redirect" | "json") {
        return Err(422);
    }
    let response_status = if response_kind == "json" {
        response
            .get("status")
            .and_then(Value::as_u64)
            .and_then(|status| u16::try_from(status).ok())
            .filter(|status| (200..=299).contains(status))
            .ok_or(422)?
    } else {
        303
    };

    let transition = dynamic_find_workflow_transition(&workflows_root, action_id);
    let compiled_effects = app
        .application_ir
        .as_ref()
        .and_then(|ir| {
            ir.actions
                .values()
                .find(|candidate| candidate.id.as_str() == action_id)
        })
        .map(|action| &action.effects)
        .ok_or(403)?;
    let executable_effects =
        pharos_runtime_core::application_ir::compile_action_effects_for_runtime(action, transition)
            .map_err(|_| 422)?;
    if compiled_effects != &executable_effects {
        return Err(403);
    }
    let document_store = if super::app_uses_file_backed_runtime_state(app) {
        None
    } else {
        let store = super::database::runtime_logical_app_store(app).map_err(|_| 503)?;
        let (entity, field, identity) = declared_document_storage(app).ok_or(422)?;
        let snapshot = pharos_runtime_database::document::read_document(
            store.adapter.as_ref(),
            &store.config,
            &store.schema,
            &entity,
            identity,
            &field,
        )
        .map_err(|error| match error {
            pharos_runtime_database::operations::DatabaseError::MissingRecord => 404,
            pharos_runtime_database::operations::DatabaseError::Conflict => 409,
            _ => 503,
        })?;
        Some((store, snapshot))
    };
    let mut root = match &document_store {
        Some((_, snapshot)) => snapshot.value.clone(),
        None => fs::read_to_string(state_path.ok_or(503)?)
            .ok()
            .and_then(|text| serde_json::from_str::<Value>(&text).ok())
            .ok_or(404)?,
    };
    let original_root = root.clone();
    let result = (|| {
        let retried_delivery_count = 0;
        let failed_retried_delivery_count = 0;
        let mut created_record_count = 0_i64;
        let mut updated_record_count = 0_i64;
        let mut matched_record_count = 0_i64;
        let mut skipped_record_count = 0_i64;
        let mut removed_record_count = 0_i64;
        let mut anchor_record_id = 0_i64;
        for primitive in action
            .get("mutation_primitives")
            .and_then(Value::as_array)
            .into_iter()
            .flatten()
        {
            let primitive_kind = primitive
                .get("primitive_kind")
                .and_then(Value::as_str)
                .unwrap_or_default();
            match primitive_kind {
                "append_record" => {
                    dynamic_append_record_primitive(
                        &mut root,
                        request,
                        &input_overrides,
                        primitive,
                        &mut anchor_record_id,
                        &mut created_record_count,
                    )?;
                }
                "set_root_fields" => {
                    dynamic_set_root_fields_primitive(
                        app,
                        &mut root,
                        request,
                        &input_overrides,
                        primitive,
                    )?;
                }
                "update_record" => {
                    dynamic_update_record_primitive(
                        &mut root,
                        request,
                        &input_overrides,
                        primitive,
                        &mut matched_record_count,
                        &mut updated_record_count,
                    )?;
                }
                "approval_transition" => {
                    let collection = primitive
                        .get("collection")
                        .and_then(Value::as_str)
                        .unwrap_or("approval_requests");
                    let match_field = primitive
                        .get("match")
                        .and_then(Value::as_object)
                        .and_then(|map| map.get("field"))
                        .and_then(Value::as_str)
                        .unwrap_or("id");
                    let match_input = primitive
                        .get("match")
                        .and_then(Value::as_object)
                        .and_then(|map| map.get("from_input"))
                        .and_then(Value::as_str)
                        .unwrap_or("id");
                    let wanted = dynamic_action_input_value(request, &input_overrides, match_input)
                        .unwrap_or_default();
                    let transition_kind = primitive
                        .get("transition_kind")
                        .and_then(Value::as_str)
                        .unwrap_or_default();
                    let Some(row) =
                        dynamic_state_rows_mut(&mut root, collection).and_then(|rows| {
                            rows.iter_mut().find(|row| {
                                row.get(match_field).map(value_to_text).unwrap_or_default()
                                    == wanted
                            })
                        })
                    else {
                        return Err(404);
                    };
                    let Some(object) = row.as_object_mut() else {
                        return Err(500);
                    };
                    if transition_kind.is_empty() {
                        return Err(422);
                    }
                    dynamic_apply_approval_transition(
                        object,
                        primitive,
                        request,
                        &input_overrides,
                    )?;
                    updated_record_count += 1;
                    matched_record_count += 1;
                }
                "reservation_transition" => {
                    let collection = primitive
                        .get("collection")
                        .and_then(Value::as_str)
                        .ok_or(422)?;
                    let scope_field = primitive
                        .get("scope_match")
                        .and_then(Value::as_object)
                        .and_then(|map| map.get("field"))
                        .and_then(Value::as_str)
                        .ok_or(422)?;
                    let scope_input = primitive
                        .get("scope_match")
                        .and_then(Value::as_object)
                        .and_then(|map| map.get("from_input"))
                        .and_then(Value::as_str)
                        .ok_or(422)?;
                    let wanted = dynamic_action_input_value(request, &input_overrides, scope_input)
                        .unwrap_or_default();
                    let reservation_state_field = primitive
                        .get("reservation_state_field")
                        .and_then(Value::as_str)
                        .filter(|value| !value.is_empty())
                        .unwrap_or("reservation_state");
                    let reserved_by_field = primitive
                        .get("reserved_by_field")
                        .and_then(Value::as_str)
                        .filter(|value| !value.is_empty())
                        .unwrap_or("reserved_by");
                    let reservation_key_field = primitive
                        .get("reservation_key_field")
                        .and_then(Value::as_str)
                        .filter(|value| !value.is_empty())
                        .unwrap_or("reservation_key");
                    let reserved_at_field = primitive
                        .get("reserved_at_field")
                        .and_then(Value::as_str)
                        .filter(|value| !value.is_empty())
                        .unwrap_or("reserved_at");
                    let released_at_field = primitive
                        .get("released_at_field")
                        .and_then(Value::as_str)
                        .filter(|value| !value.is_empty())
                        .unwrap_or("released_at");
                    let released_by_field = primitive
                        .get("released_by_field")
                        .and_then(Value::as_str)
                        .filter(|value| !value.is_empty());
                    let transition_kind = primitive
                        .get("transition_kind")
                        .and_then(Value::as_str)
                        .unwrap_or_default();
                    let max_records = primitive
                        .get("max_records")
                        .and_then(Value::as_i64)
                        .unwrap_or(i64::MAX)
                        .max(0) as usize;
                    let overflow_policy = primitive
                        .get("overflow_policy")
                        .and_then(Value::as_str)
                        .unwrap_or("truncate");
                    let actor_value = primitive
                        .get("actor_input")
                        .and_then(Value::as_str)
                        .and_then(|field| {
                            dynamic_action_input_value(request, &input_overrides, field)
                        })
                        .unwrap_or_default();
                    let reservation_key = primitive
                        .get("key_input")
                        .and_then(Value::as_str)
                        .and_then(|field| {
                            dynamic_action_input_value(request, &input_overrides, field)
                        })
                        .unwrap_or_default();
                    let released_state = primitive
                        .get("released_state")
                        .and_then(Value::as_str)
                        .filter(|value| !value.is_empty())
                        .unwrap_or("available");
                    let Some(rows) = dynamic_state_rows_mut(&mut root, collection) else {
                        return Err(500);
                    };
                    let matching_indexes = rows
                        .iter()
                        .enumerate()
                        .filter(|(_, row)| {
                            row.get(scope_field).map(value_to_text).unwrap_or_default() == wanted
                        })
                        .map(|(index, _)| index)
                        .collect::<Vec<_>>();
                    matched_record_count += matching_indexes.len() as i64;
                    if matching_indexes.len() > max_records && overflow_policy == "fail" {
                        return Err(422);
                    }
                    skipped_record_count +=
                        matching_indexes.len().saturating_sub(max_records) as i64;
                    if transition_kind == "reserve" && actor_value.trim().is_empty() {
                        return Err(422);
                    }
                    let now = now_iso_utc();
                    for index in matching_indexes.into_iter().take(max_records) {
                        let Some(object) = rows.get_mut(index).and_then(Value::as_object_mut)
                        else {
                            return Err(500);
                        };
                        let current_state = object
                            .get(reservation_state_field)
                            .and_then(Value::as_str)
                            .unwrap_or_default();
                        let current_key = object
                            .get(reservation_key_field)
                            .map(value_to_text)
                            .unwrap_or_default();
                        match transition_kind {
                            "reserve" => {
                                if current_state == "reserved"
                                    && !current_key.is_empty()
                                    && current_key != reservation_key
                                {
                                    return Err(409);
                                }
                                object.insert(
                                    reservation_state_field.to_owned(),
                                    Value::String("reserved".to_owned()),
                                );
                                object.insert(
                                    reserved_by_field.to_owned(),
                                    Value::String(actor_value.clone()),
                                );
                                object.insert(
                                    reservation_key_field.to_owned(),
                                    Value::String(reservation_key.clone()),
                                );
                                object.insert(
                                    reserved_at_field.to_owned(),
                                    Value::String(now.clone()),
                                );
                                object.insert(
                                    released_at_field.to_owned(),
                                    Value::String(String::new()),
                                );
                                if let Some(field) = released_by_field {
                                    object.insert(field.to_owned(), Value::String(String::new()));
                                }
                            }
                            "release" => {
                                if current_state == "reserved"
                                    && !reservation_key.is_empty()
                                    && !current_key.is_empty()
                                    && current_key != reservation_key
                                {
                                    return Err(409);
                                }
                                object.insert(
                                    reservation_state_field.to_owned(),
                                    Value::String(released_state.to_owned()),
                                );
                                object.insert(
                                    reserved_by_field.to_owned(),
                                    Value::String(String::new()),
                                );
                                object.insert(
                                    reservation_key_field.to_owned(),
                                    Value::String(String::new()),
                                );
                                object.insert(
                                    released_at_field.to_owned(),
                                    Value::String(now.clone()),
                                );
                                if let Some(field) = released_by_field {
                                    object.insert(
                                        field.to_owned(),
                                        Value::String(actor_value.clone()),
                                    );
                                }
                            }
                            _ => return Err(422),
                        }
                        updated_record_count += 1;
                    }
                }
                "scoped_status_transition" => {
                    dynamic_apply_scoped_status_transition_primitive(
                        &mut root,
                        request,
                        &input_overrides,
                        primitive,
                        &mut matched_record_count,
                        &mut updated_record_count,
                        &mut skipped_record_count,
                    )?;
                }
                "append_records_from_input_list" => {
                    dynamic_append_records_from_input_list_primitive(
                        &mut root,
                        request,
                        &input_overrides,
                        primitive,
                        anchor_record_id,
                        matched_record_count,
                        updated_record_count,
                        &mut skipped_record_count,
                        &mut created_record_count,
                    )?;
                }
                "replace_records_from_input_list" => {
                    dynamic_replace_records_from_input_list_primitive(
                        &mut root,
                        request,
                        &input_overrides,
                        primitive,
                        &mut matched_record_count,
                        &mut created_record_count,
                        &mut updated_record_count,
                        &mut skipped_record_count,
                        &mut removed_record_count,
                    )?;
                }
                "delete_records_by_scope" => {
                    dynamic_delete_records_by_scope_primitive(
                        &mut root,
                        request,
                        &input_overrides,
                        primitive,
                        &mut matched_record_count,
                        &mut skipped_record_count,
                        &mut removed_record_count,
                    )?;
                }
                "append_records_from_related_collection" => {
                    dynamic_append_records_from_related_collection_primitive(
                        &mut root,
                        request,
                        &input_overrides,
                        primitive,
                        &mut matched_record_count,
                        &mut skipped_record_count,
                        &mut created_record_count,
                    )?;
                }
                "scoped_reassignment" | "bulk_update" => {
                    dynamic_apply_scoped_assignment_or_bulk_update_primitive(
                        &mut root,
                        request,
                        &input_overrides,
                        primitive,
                        primitive_kind,
                        &mut matched_record_count,
                        &mut updated_record_count,
                        &mut skipped_record_count,
                    )?;
                }
                "upsert_record" => {
                    dynamic_upsert_record_primitive(
                        &mut root,
                        request,
                        &input_overrides,
                        primitive,
                        anchor_record_id,
                        &mut matched_record_count,
                        &mut updated_record_count,
                        skipped_record_count,
                        &mut created_record_count,
                    )?;
                }
                _ => return Err(422),
            }
        }
        let (
            event_kind,
            queued_notification_count,
            queued_webhook_count,
            emitted_audit_count,
            queued_file_record_count,
            failed_effect_count,
        ) = prepare_action_effects(
            app,
            &mut root,
            action,
            transition,
            &integrations_root,
            request,
            &input_overrides,
            action_id,
            anchor_record_id,
        )?;
        let entity_id = action
            .get("entity_id")
            .and_then(Value::as_str)
            .unwrap_or("record");
        let delivered_webhook_count = 0_i64;
        let retryable_webhook_count = 0_i64;
        let failed_webhook_count = failed_retried_delivery_count + failed_effect_count;
        let report = serde_json::json!({
            "action_id": action_id,
            "event_kind": event_kind,
            "created_record_count": created_record_count,
            "updated_record_count": updated_record_count,
            "matched_record_count": matched_record_count,
            "skipped_record_count": skipped_record_count,
            "removed_record_count": removed_record_count,
            "queued_notification_count": queued_notification_count,
            "queued_webhook_count": queued_webhook_count,
            "emitted_audit_count": emitted_audit_count,
            "emitted_file_record_count": 0,
            "queued_file_record_count": queued_file_record_count,
            "delivered_webhook_count": delivered_webhook_count,
            "retryable_webhook_count": retryable_webhook_count,
            "failed_webhook_count": failed_webhook_count,
            "retried_delivery_count": retried_delivery_count,
            "anchor_record_id": anchor_record_id,
            "operator_message": dynamic_presentation_text(action, "success_message"),
            "operator_summary": dynamic_presentation_text(action, "operator_summary"),
            "transition_label": dynamic_transition_presentation_text(transition, "transition_label"),
            "to_state": transition.and_then(|value| value.get("to_state")).and_then(Value::as_str).unwrap_or_default(),
            "overflow_policy": action
                .get("mutation_primitives")
                .and_then(Value::as_array)
                .and_then(|rows| rows.iter().find_map(|row| row.get("overflow_policy").and_then(Value::as_str)))
                .unwrap_or_default(),
        });
        let report_annotations = dynamic_action_report_annotations(
            action,
            request,
            &input_overrides,
            action_id,
            entity_id,
            &event_kind,
            anchor_record_id,
        )?;
        let mut report = report;
        if let Some(object) = report.as_object_mut()
            && !report_annotations.is_empty()
        {
            object.insert("annotations".to_owned(), Value::Object(report_annotations));
        }
        if !dynamic_store_last_action_report(&mut root, report) {
            return Err(500);
        }
        if let Some((store, snapshot)) = &document_store {
            pharos_runtime_database::document::write_document(
                store.adapter.as_ref(),
                &store.config,
                &store.schema,
                snapshot,
                &root,
            )
            .map_err(|error| match error {
                pharos_runtime_database::operations::DatabaseError::Conflict => {
                    *rejected_write = true;
                    409
                }
                pharos_runtime_database::operations::DatabaseError::InvalidInput => 422,
                _ => 503,
            })?;
        }
        if document_store.is_none() {
            let encoded = serde_json::to_string(&root).map_err(|_| 500)?;
            if super::fs::write_text_file(state_path.ok_or(503)?, &encoded) != 0 {
                return Err(500);
            }
        }
        if response_kind == "json" {
            let mut body = serde_json::json!({
                "schema": "pharos.action.result.v1",
                "action_id": action_id,
                "report": root.get("last_action_report").cloned().unwrap_or(Value::Null),
            });
            if response.get("include_state").and_then(Value::as_bool) == Some(true) {
                if let Some(object) = root.as_object_mut() {
                    object.remove("_database_worker_lease");
                }
                body["state"] = root;
            }
            Ok(DeclarativeActionResponse::Json(response_status, body))
        } else {
            Ok(DeclarativeActionResponse::Redirect(
                response
                    .get("location")
                    .and_then(Value::as_str)
                    .unwrap_or("/")
                    .to_owned(),
            ))
        }
    })();
    if let Err(status) = result
        && !*rejected_write
    {
        let entity_id = action
            .get("entity_id")
            .and_then(Value::as_str)
            .unwrap_or("record");
        let failure_root = dynamic_action_failure_report_root(
            original_root,
            action,
            request,
            &input_overrides,
            action_id,
            entity_id,
            transition,
            0,
            status,
        );
        if let Some((store, snapshot)) = &document_store {
            // The original compare-and-swap token prevents a failure report
            // from overwriting another action that committed concurrently.
            match pharos_runtime_database::document::write_document(
                store.adapter.as_ref(),
                &store.config,
                &store.schema,
                snapshot,
                &failure_root,
            ) {
                Ok(()) => {}
                Err(pharos_runtime_database::operations::DatabaseError::Conflict) => {
                    *rejected_write = true;
                }
                Err(_) => return Err(503),
            }
        } else if let Some(state_path) = state_path
            && let Ok(encoded) = serde_json::to_string(&failure_root)
        {
            let _ = super::fs::write_text_file(state_path, &encoded);
        }
    }
    result
}

/// Prepares declared effects in memory so their persistence shares the domain
/// transaction. It performs no network request or local-file write.
fn prepare_action_effects(
    app: &AppRuntime,
    root: &mut Value,
    action: &Value,
    transition: Option<&Value>,
    integrations_root: &Value,
    request: &HttpRequest,
    input_overrides: &BTreeMap<String, Option<String>>,
    action_id: &str,
    anchor_record_id: i64,
) -> Result<(String, i64, i64, i64, i64, i64), i32> {
    let workflow_execution_id =
        dynamic_append_workflow_execution(root, action_id, transition).ok_or(500)?;
    if !dynamic_append_workflow_events(root, workflow_execution_id, transition) {
        return Err(422);
    }
    let scheduled_jobs = transition
        .and_then(|value| value.get("scheduled_jobs"))
        .and_then(Value::as_array)
        .cloned()
        .unwrap_or_default();
    for job in &scheduled_jobs {
        let job_id = job
            .as_str()
            .or_else(|| job.get("job_id").and_then(Value::as_str))
            .unwrap_or_default();
        let queue_name = job
            .get("queue_name")
            .and_then(Value::as_str)
            .unwrap_or("default");
        let delay_seconds = job
            .get("delay_seconds")
            .and_then(Value::as_i64)
            .unwrap_or(0);
        let max_attempts = job
            .get("max_attempts")
            .and_then(Value::as_i64)
            .unwrap_or(app.queue_max_attempts.max(1));
        let backoff_seconds = job
            .get("backoff_seconds")
            .and_then(Value::as_i64)
            .unwrap_or(app.queue_retry_seconds.max(0));
        let lease_seconds = job
            .get("lease_seconds")
            .and_then(Value::as_i64)
            .unwrap_or(30);
        if !dynamic_append_scheduled_job(
            root,
            workflow_execution_id,
            job_id,
            queue_name,
            delay_seconds,
            max_attempts,
            backoff_seconds,
            lease_seconds,
        ) {
            return Err(422);
        }
    }
    let mut queued_notification_count = 0_i64;
    let mut queued_webhook_count = 0_i64;
    let mut emitted_audit_count = 0_i64;
    let mut queued_file_record_count = 0_i64;

    let mut failed_webhook_count = 0_i64;
    let event_kind = transition
        .and_then(|value| value.get("emits_event"))
        .and_then(Value::as_str)
        .map(ToOwned::to_owned)
        .or_else(|| {
            root.get("last_action")
                .and_then(Value::as_str)
                .filter(|value| !value.is_empty())
                .map(ToOwned::to_owned)
        })
        .unwrap_or_else(|| action_id.to_owned());
    let entity_id = action
        .get("entity_id")
        .and_then(Value::as_str)
        .unwrap_or("record");
    for notification in action
        .get("effect_hooks")
        .and_then(Value::as_object)
        .and_then(|hooks| hooks.get("notifications"))
        .and_then(Value::as_array)
        .into_iter()
        .flatten()
    {
        let adapter_id = notification
            .get("adapter_id")
            .and_then(Value::as_str)
            .unwrap_or("local-mail");
        let channel = notification
            .get("channel")
            .and_then(Value::as_str)
            .unwrap_or("email");
        let recipient = notification
            .get("recipient")
            .and_then(Value::as_object)
            .and_then(|spec| {
                dynamic_notification_parameter_value(
                    &Value::Object(spec.clone()),
                    request,
                    input_overrides,
                    action_id,
                    entity_id,
                    &event_kind,
                    "",
                    "",
                    anchor_record_id,
                )
            })
            .and_then(|value| value.as_str().map(ToOwned::to_owned))
            .unwrap_or_default();
        let template_id = notification
            .get("template_id")
            .and_then(Value::as_object)
            .and_then(|spec| {
                dynamic_notification_parameter_value(
                    &Value::Object(spec.clone()),
                    request,
                    input_overrides,
                    action_id,
                    entity_id,
                    &event_kind,
                    &recipient,
                    "",
                    anchor_record_id,
                )
            })
            .and_then(|value| value.as_str().map(ToOwned::to_owned))
            .unwrap_or_default();
        let template_params = dynamic_notification_template_params(
            notification,
            request,
            input_overrides,
            action_id,
            entity_id,
            &event_kind,
            &recipient,
            &template_id,
            anchor_record_id,
        );
        if dynamic_append_notification(
            root,
            integrations_root,
            adapter_id,
            channel,
            &recipient,
            &template_id,
            &template_params,
        ) {
            queued_notification_count += 1;
        } else {
            return Err(422);
        }
    }
    let resource_id =
        dynamic_action_resource_id(action, request, input_overrides, anchor_record_id);
    for adapter_id in action
        .get("effect_hooks")
        .and_then(Value::as_object)
        .and_then(|hooks| hooks.get("webhook_dispatch"))
        .and_then(Value::as_array)
        .into_iter()
        .flatten()
        .filter_map(Value::as_str)
    {
        if dynamic_append_webhook_delivery(
            root,
            integrations_root,
            adapter_id,
            action_id,
            &resource_id,
        )
        .is_some()
        {
            queued_webhook_count += 1;
        } else {
            failed_webhook_count += 1;
        }
    }
    for audit in action
        .get("effect_hooks")
        .and_then(Value::as_object)
        .and_then(|hooks| hooks.get("audit_records"))
        .and_then(Value::as_array)
        .into_iter()
        .flatten()
    {
        if dynamic_append_audit_record(
            root,
            integrations_root,
            action,
            audit,
            request,
            input_overrides,
            action_id,
            entity_id,
            &event_kind,
            anchor_record_id,
        ) {
            emitted_audit_count += 1;
        } else {
            return Err(422);
        }
    }
    for file_record in action
        .get("effect_hooks")
        .and_then(Value::as_object)
        .and_then(|hooks| hooks.get("file_records"))
        .and_then(Value::as_array)
        .into_iter()
        .flatten()
    {
        let prepared = prepare_file_record(
            integrations_root,
            action,
            file_record,
            request,
            input_overrides,
            action_id,
            entity_id,
            &event_kind,
            anchor_record_id,
        )
        .ok_or(422)?;
        if !enqueue_file_record(root, prepared) {
            return Err(422);
        }
        queued_file_record_count += 1;
    }
    Ok((
        event_kind,
        queued_notification_count,
        queued_webhook_count,
        emitted_audit_count,
        queued_file_record_count,
        failed_webhook_count,
    ))
}

/// Resolves a declared action's resource label without a vendor or table guess.
pub(super) fn action_resource_binding(
    app: &AppRuntime,
    schema: &pharos_runtime_database::operations::AppSchema,
    action_id: &str,
) -> Option<pharos_runtime_database::effects::ResourceBinding> {
    let actions = dynamic_contract_doc(app, "action_contract")?;
    let action = dynamic_find_action_contract(&actions, action_id)?;
    let label = action.get("entity_id")?.as_str()?;
    let entities = dynamic_contract_doc(app, "entity_contract");
    let name = entities
        .as_ref()
        .and_then(|root| root.get("entities"))
        .and_then(Value::as_array)
        .and_then(|rows| {
            rows.iter()
                .find(|row| row.get("entity_id").and_then(Value::as_str) == Some(label))
        })
        .and_then(|row| row.get("table"))
        .and_then(Value::as_str)
        .unwrap_or(label);
    schema.entities.iter().find(|entity| entity.name == name)?;
    Some(pharos_runtime_database::effects::ResourceBinding {
        entity: name.to_owned(),
        prefix: format!("{label}:"),
    })
}

/// Converts the shared action-effect preparation into insertion-only logical
/// operations. This reads declarations, never an app-state projection.
pub(super) fn declared_effect_steps(
    app: &AppRuntime,
    schema: &pharos_runtime_database::operations::AppSchema,
    request: &HttpRequest,
    action_id: &str,
    input_overrides: &BTreeMap<String, Option<String>>,
    anchor: i64,
    actor_id: i64,
) -> Result<Vec<pharos_runtime_database::operations::Mutation>, i32> {
    use pharos_runtime_database::operations::{FieldType, Mutation, WriteValue};
    let Some(actions) = dynamic_contract_doc(app, "action_contract") else {
        return Ok(Vec::new());
    };
    let Some(action) = dynamic_find_action_contract(&actions, action_id) else {
        return Ok(Vec::new());
    };
    let workflows = dynamic_contract_doc(app, "workflow_contract");
    let transition = workflows
        .as_ref()
        .and_then(|root| dynamic_find_workflow_transition(root, action_id));
    if transition.is_none() && action.get("effect_hooks").is_none() {
        return Ok(Vec::new());
    }
    let integrations =
        integration_contract_doc(app).unwrap_or_else(|| serde_json::json!({"adapters": []}));
    let mut prepared = serde_json::json!({});
    let (_, _, _, _, _, failures) = prepare_action_effects(
        app,
        &mut prepared,
        action,
        transition,
        &integrations,
        request,
        input_overrides,
        action_id,
        anchor,
    )?;
    if failures != 0 {
        return Err(422);
    }
    static EFFECT_SEQUENCE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
    let sequence = EFFECT_SEQUENCE.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    let token = hex_digest(&Sha256::digest(
        format!(
            "{}:{}:{}:{}",
            action_id,
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map_err(|_| 503)?
                .as_nanos(),
            sequence
        )
        .as_bytes(),
    ));
    let mut steps = Vec::new();
    let mut workflow_step = None;
    for (section, entity_name) in [
        ("workflow_executions", "workflow_execution"),
        ("workflow_execution_events", "workflow_execution_event"),
        ("job_executions", "job_execution"),
        ("notification_deliveries", "notification_delivery"),
        ("integration_deliveries", "integration_delivery"),
        ("audit_log", "audit_event"),
        ("local_file_deliveries", "local_file_delivery"),
    ] {
        let Some(rows) = prepared.get(section).and_then(Value::as_array) else {
            continue;
        };
        if rows.is_empty() {
            continue;
        }
        let entity = schema
            .entities
            .iter()
            .find(|entity| entity.name == entity_name)
            .ok_or(422)?;
        for row in rows {
            let mut row = row.as_object().cloned().ok_or(422)?;
            if entity_name == "workflow_execution" {
                row.insert("idempotency_key".to_owned(), Value::String(token.clone()));
                row.insert("replay_trace_id".to_owned(), Value::String(token.clone()));
            }
            if entity_name == "audit_event" {
                row.insert("payload_json".to_owned(), Value::Object(row.clone()));
                row.insert(
                    "entity_type".to_owned(),
                    action.get("entity_id").cloned().unwrap_or(Value::Null),
                );
                row.insert(
                    "entity_id".to_owned(),
                    if anchor > 0 {
                        Value::from(anchor)
                    } else {
                        Value::Null
                    },
                );
                row.insert(
                    "subject_user_id".to_owned(),
                    if actor_id > 0 {
                        Value::from(actor_id)
                    } else {
                        Value::Null
                    },
                );
            }
            if entity_name == "workflow_execution" {
                row.insert(
                    "resource_id".to_owned(),
                    Value::String(dynamic_action_resource_id(
                        action,
                        request,
                        input_overrides,
                        anchor,
                    )),
                );
                row.insert(
                    "executed_by_user_id".to_owned(),
                    if actor_id > 0 {
                        Value::from(actor_id)
                    } else {
                        Value::Null
                    },
                );
            }
            if entity_name == "integration_delivery"
                && let Some(header) = row.get("timestamp_header").cloned()
            {
                row.insert("signature_timestamp_header".to_owned(), header);
            }
            if entity_name == "job_execution" {
                row.insert(
                    "idempotency_key".to_owned(),
                    Value::String(format!("{}:{}", token, steps.len())),
                );
                row.insert("replay_trace_id".to_owned(), Value::String(token.clone()));
            }
            let mut values = BTreeMap::new();
            for field in &entity.fields {
                if entity.identity.generated_field() == Some(field.name.as_str()) {
                    continue;
                }
                let Some(mut value) = row.get(&field.name).cloned() else {
                    continue;
                };
                if field.name == "workflow_execution_id" {
                    values.insert(
                        field.name.clone(),
                        WriteValue::GeneratedId(workflow_step.ok_or(422)?),
                    );
                    continue;
                }
                if field.nullable && value.as_str() == Some("") {
                    value = Value::Null;
                }
                if field.kind == FieldType::Timestamp
                    && let Some(text) = value.as_str()
                    && let Ok(epoch) = text.parse::<i64>()
                {
                    let seconds = u64::try_from(epoch)
                        .ok()
                        .filter(|seconds| *seconds <= 253_402_300_799)
                        .ok_or(422)?;
                    value = Value::String(pharos_runtime_core::time::timestamp_from_epoch_seconds(
                        seconds,
                    ));
                }
                if field.kind == FieldType::Json
                    && let Some(text) = value.as_str()
                {
                    value = serde_json::from_str(text).map_err(|_| 422)?;
                }
                values.insert(field.name.clone(), WriteValue::Literal(value));
            }
            if entity_name == "workflow_execution" {
                workflow_step = Some(steps.len());
            }
            steps.push(Mutation::Insert {
                entity: entity_name.to_owned(),
                values,
            });
        }
    }
    Ok(steps)
}

/// Identifies an app-declared document mutation without choosing a database vendor.
pub fn is_declarative_document_action(app: &AppRuntime, action_id: &str) -> bool {
    if !super::app_uses_file_backed_runtime_state(app) && declared_document_storage(app).is_none() {
        return false;
    }
    dynamic_contract_doc(app, "action_contract")
        .and_then(|document| dynamic_find_action_contract(&document, action_id).cloned())
        .is_some_and(|action| {
            action
                .get("mutation_primitives")
                .and_then(Value::as_array)
                .is_some_and(|steps| !steps.is_empty())
        })
}

/// Document persistence requires an explicit app declaration; relational actions cannot opt in implicitly.
pub(super) fn declared_document_storage(
    app: &AppRuntime,
) -> Option<(
    String,
    String,
    pharos_runtime_database::operations::Predicate,
)> {
    let root = app.app_root.as_deref().map(Path::new)?;
    let manifest = super::declarative::declarative_source_load_fixture(root, "pharos.app.yml")?;
    let declaration = manifest.get("database")?.get("document_storage")?;
    let entity = declaration.get("entity")?.as_str()?;
    let field = declaration.get("field")?.as_str()?;
    let identity = declaration.get("identity")?.as_object()?;
    if entity.is_empty() || field.is_empty() || identity.is_empty() || identity.len() > 16 {
        return None;
    }
    let predicates = identity
        .iter()
        .map(|(name, value)| {
            if name.is_empty() || value.is_null() || value.is_array() || value.is_object() {
                return None;
            }
            Some(pharos_runtime_database::operations::Predicate::Equal(
                name.clone(),
                value.clone(),
            ))
        })
        .collect::<Option<Vec<_>>>()?;
    Some((
        entity.to_owned(),
        field.to_owned(),
        pharos_runtime_database::operations::Predicate::All(predicates),
    ))
}

/// Returns whether a route pattern matches a request path using bounded segments.
///
/// # Parameters
///
/// - `pattern`: Input consumed by [`crate::runtime::dynamic::pattern_matches`].
/// - `path`: Input consumed by [`crate::runtime::dynamic::pattern_matches`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::pattern_matches`] holds and `false` otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::pattern_matches`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::pattern_matches(todo!(), std::path::Path::new("fixture.json"));
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn pattern_matches(pattern: &str, path: &str) -> bool {
    if pattern.is_empty() || path.is_empty() {
        return false;
    }
    let pattern = pattern.trim_matches('/').split('/').collect::<Vec<_>>();
    let path = path
        .split('?')
        .next()
        .unwrap_or(path)
        .trim_matches('/')
        .split('/')
        .collect::<Vec<_>>();
    pattern.len() == path.len()
        && pattern.iter().zip(path).all(|(expected, actual)| {
            expected.starts_with(':') || *expected == "*" || *expected == actual
        })
}

/// Scores a route pattern by its number of literal segments.
///
/// # Parameters
///
/// - `pattern`: Input consumed by [`crate::runtime::dynamic::route_pattern_specificity_score`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::route_pattern_specificity_score`].
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::route_pattern_specificity_score`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::route_pattern_specificity_score(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn route_pattern_specificity_score(pattern: &str) -> usize {
    pattern
        .trim_matches('/')
        .split('/')
        .filter(|segment| !segment.is_empty() && !segment.starts_with(':') && *segment != "*")
        .count()
}

/// Resolves a route fixture below an app root with traversal protection.
///
/// # Parameters
///
/// - `artifact_dir`: Input consumed by [`crate::runtime::dynamic::resolve_route_fixture_path`].
/// - `app_id`: Input consumed by [`crate::runtime::dynamic::resolve_route_fixture_path`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::resolve_route_fixture_path`] can produce a value and [`None`] otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::resolve_route_fixture_path`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::resolve_route_fixture_path(std::path::Path::new("build/dev_docs"), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn resolve_route_fixture_path(artifact_dir: &Path, app_id: &str) -> Option<String> {
    if artifact_dir.as_os_str().is_empty() || app_id.is_empty() || app_id.contains(['/', '\\']) {
        return None;
    }
    let candidates = [
        artifact_dir.join("routes.json"),
        artifact_dir.join("data/contracts/routes.json"),
        artifact_dir.join("apps").join(app_id).join("routes.json"),
    ];
    candidates
        .into_iter()
        .find(|path| path.is_file())
        .map(|path| path.to_string_lossy().into_owned())
}

/// Releases an owned dynamic route match value.
///
/// # Parameters
///
/// - `_value`: Input consumed by [`crate::runtime::dynamic::dynamic_dispatch_free_match`].
///
/// # Returns
///
/// Returns after [`crate::runtime::dynamic::dynamic_dispatch_free_match`] applies its in-place effect.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::dynamic_dispatch_free_match(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn dynamic_dispatch_free_match<T>(_value: Option<T>) {}

/// Validates a webhook endpoint identifier against the framework-owned route shape.
///
/// # Parameters
///
/// - `path`: Input consumed by [`crate::runtime::dynamic::framework_webhook_endpoint_id`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::framework_webhook_endpoint_id`] can produce a value and [`None`] otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::framework_webhook_endpoint_id`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::framework_webhook_endpoint_id(std::path::Path::new("fixture.json"));
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn framework_webhook_endpoint_id(path: &str) -> Option<String> {
    path.strip_prefix("/_pharos/webhooks/")
        .map(|rest| rest.strip_suffix('/').unwrap_or(rest))
        .filter(|value| {
            !value.is_empty()
                && value.len() <= 128
                && !value.contains('/')
                && value
                    .bytes()
                    .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_' || byte == b'-')
        })
        .map(ToOwned::to_owned)
}

/// Resolves the app-owned integration fixture without leaving the app root.
///
/// # Parameters
///
/// - `app`: Input consumed by [`crate::runtime::dynamic::integration_fixture_path`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::integration_fixture_path`] can produce a value and [`None`] otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::integration_fixture_path`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::integration_fixture_path(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn integration_fixture_path(app: &AppRuntime) -> Option<String> {
    let root = Path::new(app.app_root.as_deref()?);
    let value = fs::read_to_string(root.join("pharos.app.json"))
        .ok()
        .and_then(|text| serde_json::from_str::<Value>(&text).ok())?;
    let relative = value
        .get("contracts")
        .and_then(Value::as_object)?
        .get("integration_contract")
        .and_then(Value::as_str)?;
    if relative.is_empty()
        || Path::new(relative).is_absolute()
        || relative.split('/').any(|segment| segment == "..")
    {
        return None;
    }
    Some(root.join(relative).to_string_lossy().into_owned())
}

/// Validates a state-backed capability for the current user and tenant.
///
/// # Parameters
///
/// - `state_root`: Input consumed by [`crate::runtime::dynamic::generated_surface_user_has_capability`].
/// - `_app`: Input consumed by [`crate::runtime::dynamic::generated_surface_user_has_capability`].
/// - `user_id`: Input consumed by [`crate::runtime::dynamic::generated_surface_user_has_capability`].
/// - `tenant_id`: Input consumed by [`crate::runtime::dynamic::generated_surface_user_has_capability`].
/// - `capability`: Input consumed by [`crate::runtime::dynamic::generated_surface_user_has_capability`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::generated_surface_user_has_capability`] holds and `false` otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::generated_surface_user_has_capability`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::generated_surface_user_has_capability(todo!(), todo!(), 1, todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn generated_surface_user_has_capability(
    state_root: &Value,
    _app: &AppRuntime,
    user_id: i64,
    tenant_id: i64,
    capability: &str,
) -> bool {
    user_id > 0
        && tenant_id > 0
        && !capability.is_empty()
        && state_root
            .get("capabilities")
            .and_then(Value::as_array)
            .is_some_and(|rows| {
                rows.iter().any(|row| {
                    row.get("user_id").and_then(Value::as_i64) == Some(user_id)
                        && row.get("tenant_id").and_then(Value::as_i64) == Some(tenant_id)
                        && row.get("capability").and_then(Value::as_str) == Some(capability)
                })
            })
}

/// Produces a bounded JSON error envelope for direct HTTP handling.
///
/// # Parameters
///
/// - `status`: Input consumed by [`crate::runtime::dynamic::write_error_envelope`].
/// - `status_text`: Input consumed by [`crate::runtime::dynamic::write_error_envelope`].
/// - `error_type`: Input consumed by [`crate::runtime::dynamic::write_error_envelope`].
/// - `error_code`: Input consumed by [`crate::runtime::dynamic::write_error_envelope`].
/// - `message`: Input consumed by [`crate::runtime::dynamic::write_error_envelope`].
/// - `actor_scope`: Input consumed by [`crate::runtime::dynamic::write_error_envelope`].
/// - `tenant_scope`: Input consumed by [`crate::runtime::dynamic::write_error_envelope`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::write_error_envelope`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::write_error_envelope(400, "400 Bad Request", todo!(), todo!(), "example", todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn write_error_envelope(
    status: u16,
    status_text: &str,
    error_type: &str,
    error_code: &str,
    message: &str,
    actor_scope: Option<&str>,
    tenant_scope: Option<&str>,
) -> String {
    serde_json::json!({"status":status,"status_text":status_text,"error_type":error_type,"error_code":error_code,"message":message,"actor_scope":actor_scope.unwrap_or_default(),"tenant_scope":tenant_scope.unwrap_or_default()}).to_string()
}

/// Applies a route operation response directly as a JSON document.
///
/// # Parameters
///
/// - `surface_root`: Input consumed by [`crate::runtime::dynamic::write_generated_surface_api_schema_response`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::write_generated_surface_api_schema_response`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::write_generated_surface_api_schema_response(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn write_generated_surface_api_schema_response(surface_root: &Value) -> Option<String> {
    serde_json::to_string(surface_root).ok()
}

/// Builds one generated API item response with a bounded status envelope.
///
/// # Parameters
///
/// - `status`: Input consumed by [`crate::runtime::dynamic::write_generated_surface_api_item_response`].
/// - `status_text`: Input consumed by [`crate::runtime::dynamic::write_generated_surface_api_item_response`].
/// - `surface_root`: Input consumed by [`crate::runtime::dynamic::write_generated_surface_api_item_response`].
/// - `item_root`: Input consumed by [`crate::runtime::dynamic::write_generated_surface_api_item_response`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::write_generated_surface_api_item_response`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::write_generated_surface_api_item_response(400, "400 Bad Request", todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn write_generated_surface_api_item_response(
    status: u16,
    status_text: &str,
    surface_root: &Value,
    item_root: &Value,
) -> Option<String> {
    serde_json::to_string(&serde_json::json!({"status":status,"status_text":status_text,"surface":surface_root,"item":item_root})).ok()
}

/// Builds a generated API list response from an array-valued source document.
///
/// # Parameters
///
/// - `items`: Input consumed by [`crate::runtime::dynamic::write_generated_surface_api_list_response`].
/// - `raw_query`: Input consumed by [`crate::runtime::dynamic::write_generated_surface_api_list_response`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::write_generated_surface_api_list_response`] can produce a value and [`None`] otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::write_generated_surface_api_list_response`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::write_generated_surface_api_list_response(todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn write_generated_surface_api_list_response(
    items: &[Value],
    raw_query: Option<&str>,
) -> Option<String> {
    let query = raw_query.unwrap_or_default().to_ascii_lowercase();
    let filtered = items
        .iter()
        .filter(|item| query.is_empty() || item.to_string().to_ascii_lowercase().contains(&query))
        .cloned()
        .collect::<Vec<_>>();
    serde_json::to_string(&serde_json::json!({"items":filtered,"count":filtered.len()})).ok()
}

/// Reads a numeric object field with a bounded fallback.
///
/// # Parameters
///
/// - `object`: Input consumed by [`crate::runtime::dynamic::mut_json_obj_get_long_default`].
/// - `key`: Input consumed by [`crate::runtime::dynamic::mut_json_obj_get_long_default`].
/// - `default_value`: Input consumed by [`crate::runtime::dynamic::mut_json_obj_get_long_default`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::mut_json_obj_get_long_default`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::mut_json_obj_get_long_default(todo!(), "key", todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn mut_json_obj_get_long_default(object: &Value, key: &str, default_value: i64) -> i64 {
    object
        .get(key)
        .and_then(Value::as_i64)
        .or_else(|| {
            object
                .get(key)
                .and_then(Value::as_str)
                .and_then(|value| value.parse().ok())
        })
        .unwrap_or(default_value)
}

/// Compares two numeric JSON values without coercing arbitrary objects.
///
/// # Parameters
///
/// - `value`: Input consumed by [`crate::runtime::dynamic::value_long_equals_runtime_local`].
/// - `expected`: Input consumed by [`crate::runtime::dynamic::value_long_equals_runtime_local`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::value_long_equals_runtime_local`] holds and `false` otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::value_long_equals_runtime_local("value", "value");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn value_long_equals_runtime_local(value: &Value, expected: i64) -> bool {
    mut_json_obj_get_long_default(value, "value", i64::MIN) == expected
        || value.as_i64() == Some(expected)
}

/// Revokes a matching session cookie from a persisted state document.
///
/// # Parameters
///
/// - `state_path`: Input consumed by [`crate::runtime::dynamic::revoke_session_by_cookie`].
/// - `cookie`: Input consumed by [`crate::runtime::dynamic::revoke_session_by_cookie`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::revoke_session_by_cookie`].
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::revoke_session_by_cookie`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::revoke_session_by_cookie(todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn revoke_session_by_cookie(state_path: &Path, cookie: &str) -> i32 {
    if cookie.is_empty() {
        return 64;
    }
    let Ok(mut root) = fs::read_to_string(state_path)
        .ok()
        .and_then(|text| serde_json::from_str::<Value>(&text).ok())
        .ok_or(66)
    else {
        return 66;
    };
    let Some(rows) = root.get_mut("sessions").and_then(Value::as_array_mut) else {
        return 69;
    };
    for row in rows {
        if row.get("cookie").and_then(Value::as_str) == Some(cookie)
            && let Some(object) = row.as_object_mut()
        {
            object.insert("revoked_at".to_owned(), Value::String(now_iso_utc()));
        }
    }
    super::fs::write_text_file(
        state_path,
        &serde_json::to_string(&root).unwrap_or_default(),
    )
}

/// Serves a bounded route fixture file directly as response bytes.
///
/// # Parameters
///
/// - `root`: Input consumed by [`crate::runtime::dynamic::dispatch_response_file_for_request`].
/// - `method`: Input consumed by [`crate::runtime::dynamic::dispatch_response_file_for_request`].
/// - `path`: Input consumed by [`crate::runtime::dynamic::dispatch_response_file_for_request`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::dispatch_response_file_for_request`] can produce a value and [`None`] otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::dispatch_response_file_for_request`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::dispatch_response_file_for_request(std::path::Path::new("."), todo!(), std::path::Path::new("fixture.json"));
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn dispatch_response_file_for_request(root: &Path, method: &str, path: &str) -> Option<String> {
    if !method.eq_ignore_ascii_case("GET") && !method.eq_ignore_ascii_case("HEAD") {
        return None;
    }
    let relative = path.trim_start_matches('/');
    if relative.is_empty() || relative.split('/').any(|segment| segment == "..") {
        return None;
    }
    let target = root.join(relative);
    let text = fs::read_to_string(target).ok()?;
    (text.len() <= 16 * 1024 * 1024).then_some(text)
}

/// Finds one framework webhook endpoint in a declarative integration document.
///
/// # Parameters
///
/// - `integration_root`: Input consumed by [`crate::runtime::dynamic::find_webhook_endpoint`].
/// - `endpoint_id`: Input consumed by [`crate::runtime::dynamic::find_webhook_endpoint`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::find_webhook_endpoint`] can produce a value and [`None`] otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::find_webhook_endpoint`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::find_webhook_endpoint(todo!(), "webhook");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn find_webhook_endpoint<'a>(
    integration_root: &'a Value,
    endpoint_id: &str,
) -> Option<&'a Value> {
    if endpoint_id.is_empty() {
        return None;
    }
    integration_root
        .get("webhook_endpoints")
        .and_then(Value::as_array)
        .and_then(|rows| {
            rows.iter()
                .find(|row| row.get("endpoint_id").and_then(Value::as_str) == Some(endpoint_id))
        })
}

/// Verifies a declarative webhook signature and timestamp without exposing secret material.
///
/// # Parameters
///
/// - `endpoint`: Input consumed by [`crate::runtime::dynamic::webhook_request_signature_valid`].
/// - `request`: Input consumed by [`crate::runtime::dynamic::webhook_request_signature_valid`].
/// - `payload_json`: Input consumed by [`crate::runtime::dynamic::webhook_request_signature_valid`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::webhook_request_signature_valid`] holds and `false` otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::webhook_request_signature_valid("127.0.0.1:8080", todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn webhook_request_signature_valid(
    endpoint: &Value,
    request: &HttpRequest,
    payload_json: &str,
) -> bool {
    let Some(signature) = endpoint.get("signature").and_then(Value::as_object) else {
        return false;
    };
    if signature.get("mode").and_then(Value::as_str) != Some("hmac_sha256") {
        return false;
    }
    let Some(signature_header) = signature.get("header").and_then(Value::as_str) else {
        return false;
    };
    let Some(timestamp_header) = signature.get("timestamp_header").and_then(Value::as_str) else {
        return false;
    };
    let Some(secret_env) = signature.get("secret_env").and_then(Value::as_str) else {
        return false;
    };
    let max_age = signature
        .get("max_age_seconds")
        .and_then(Value::as_u64)
        .filter(|value| *value > 0)
        .unwrap_or(300);
    let Some(timestamp_text) = request_header_value(request, timestamp_header) else {
        return false;
    };
    let Some(timestamp) = timestamp_text.parse::<u64>().ok() else {
        return false;
    };
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|value| value.as_secs())
        .unwrap_or(0);
    if now.abs_diff(timestamp) > max_age {
        return false;
    }
    let Some(provided) = request_header_value(request, signature_header) else {
        return false;
    };
    let Some(secret) = env::var(secret_env).ok().filter(|value| !value.is_empty()) else {
        return false;
    };
    let message = format!("{timestamp_text}.{payload_json}");
    let expected = format!(
        "sha256={}",
        hex_digest(&hmac_sha256(secret.as_bytes(), message.as_bytes()))
    );
    constant_time_equal(provided.as_bytes(), expected.as_bytes())
}

/// Records one webhook receipt in the app state document with bounded payload size.
///
/// # Parameters
///
/// - `state_path`: Input consumed by [`crate::runtime::dynamic::append_webhook_receipt`].
/// - `endpoint_id`: Input consumed by [`crate::runtime::dynamic::append_webhook_receipt`].
/// - `verification_status`: Input consumed by [`crate::runtime::dynamic::append_webhook_receipt`].
/// - `payload_json`: Input consumed by [`crate::runtime::dynamic::append_webhook_receipt`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::append_webhook_receipt`].
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::append_webhook_receipt`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::append_webhook_receipt(todo!(), "webhook", todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn append_webhook_receipt(
    state_path: &Path,
    endpoint_id: &str,
    verification_status: &str,
    payload_json: &str,
) -> i32 {
    if endpoint_id.is_empty() || verification_status.is_empty() || payload_json.len() > 1_048_576 {
        return 64;
    }
    let Ok(mut root) = fs::read_to_string(state_path)
        .ok()
        .and_then(|text| serde_json::from_str::<Value>(&text).ok())
        .ok_or(66)
    else {
        return 66;
    };
    let Some(rows) = root
        .as_object_mut()
        .map(|object| {
            object
                .entry("webhook_receipts".to_owned())
                .or_insert_with(|| Value::Array(Vec::new()))
        })
        .and_then(Value::as_array_mut)
    else {
        return 69;
    };
    rows.push(serde_json::json!({"endpoint_id":endpoint_id,"verification_status":verification_status,"payload":payload_json,"created_at":now_iso_utc()}));
    super::fs::write_text_file(
        state_path,
        &serde_json::to_string(&root).unwrap_or_default(),
    )
}

/// Builds the HMAC signature header for an outbound webhook payload.
///
/// # Parameters
///
/// - `secret_env`: Input consumed by [`crate::runtime::dynamic::webhook_signature_header_value_dup_logic_local`].
/// - `timestamp_text`: Input consumed by [`crate::runtime::dynamic::webhook_signature_header_value_dup_logic_local`].
/// - `payload_json`: Input consumed by [`crate::runtime::dynamic::webhook_signature_header_value_dup_logic_local`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::webhook_signature_header_value_dup_logic_local`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::webhook_signature_header_value_dup_logic_local(todo!(), todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn webhook_signature_header_value_dup_logic_local(
    secret_env: &str,
    timestamp_text: &str,
    payload_json: &str,
) -> Option<String> {
    if secret_env.is_empty() || timestamp_text.is_empty() || payload_json.is_empty() {
        return None;
    }
    let secret = env::var(secret_env)
        .ok()
        .filter(|value| !value.is_empty())?;
    let message = format!("{timestamp_text}.{payload_json}");
    Some(format!(
        "sha256={}",
        hex_digest(&hmac_sha256(secret.as_bytes(), message.as_bytes()))
    ))
}

/// Appends bytes to a bounded mutable byte buffer.
///
/// # Parameters
///
/// - `output`: Input consumed by [`crate::runtime::dynamic::append_bytes_logic_local`].
/// - `bytes`: Input consumed by [`crate::runtime::dynamic::append_bytes_logic_local`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::append_bytes_logic_local`] holds and `false` otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::append_bytes_logic_local(todo!(), b"");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn append_bytes_logic_local(output: &mut Vec<u8>, bytes: &[u8]) -> bool {
    if output.len().saturating_add(bytes.len()) > 16 * 1024 * 1024 {
        return false;
    }
    output.extend_from_slice(bytes);
    true
}

/// Reads one named scalar from a JSON object as owned text.
///
/// # Parameters
///
/// - `object`: Input consumed by [`crate::runtime::dynamic::input_field_value_dup_logic_local`].
/// - `field_name`: Input consumed by [`crate::runtime::dynamic::input_field_value_dup_logic_local`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::input_field_value_dup_logic_local`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::input_field_value_dup_logic_local(todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn input_field_value_dup_logic_local(object: &Value, field_name: &str) -> Option<String> {
    object
        .get(field_name)
        .map(value_to_text)
        .filter(|value| !value.is_empty())
}

/// Inserts one validated field value into a mutable JSON object.
///
/// # Parameters
///
/// - `object`: Input consumed by [`crate::runtime::dynamic::mut_object_field_logic_local`].
/// - `field_name`: Input consumed by [`crate::runtime::dynamic::mut_object_field_logic_local`].
/// - `value`: Input consumed by [`crate::runtime::dynamic::mut_object_field_logic_local`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::mut_object_field_logic_local`] holds and `false` otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::mut_object_field_logic_local(todo!(), todo!(), "value");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn mut_object_field_logic_local(
    object: &mut Map<String, Value>,
    field_name: &str,
    value: Value,
) -> bool {
    if field_name.is_empty() || field_name.len() > 256 || field_name.contains('\0') {
        return false;
    }
    object.insert(field_name.to_owned(), value);
    true
}

/// Adds the generated-surface context binding to a mutable object.
///
/// # Parameters
///
/// - `object`: Input consumed by [`crate::runtime::dynamic::dynamic_generated_admin_context_binding_mut`].
/// - `binding`: Input consumed by [`crate::runtime::dynamic::dynamic_generated_admin_context_binding_mut`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::dynamic_generated_admin_context_binding_mut`] holds and `false` otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::dynamic_generated_admin_context_binding_mut(todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn dynamic_generated_admin_context_binding_mut(
    object: &mut Map<String, Value>,
    binding: &Value,
) -> bool {
    let Some(binding) = binding.as_object() else {
        return false;
    };
    if binding.get("param").and_then(Value::as_str).is_none() {
        return false;
    }
    object.insert("context_binding".to_owned(), Value::Object(binding.clone()));
    true
}

/// Returns the current Unix epoch as decimal text.
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::now_unix_epoch_text_logic_local`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::now_unix_epoch_text_logic_local();
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn now_unix_epoch_text_logic_local() -> String {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|value| value.as_secs().to_string())
        .unwrap_or_else(|_| "0".to_owned())
}

/// Parses and validates an HTTP URL into scheme, host, and optional port.
///
/// # Parameters
///
/// - `url`: Input consumed by [`crate::runtime::dynamic::parse_http_url_logic_local`].
/// - `String`: Input consumed by [`crate::runtime::dynamic::parse_http_url_logic_local`].
/// - `Option<u16>`: Input consumed by [`crate::runtime::dynamic::parse_http_url_logic_local`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::parse_http_url_logic_local`] can produce a value and [`None`] otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::parse_http_url_logic_local`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::parse_http_url_logic_local(todo!(), todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn parse_http_url_logic_local(url: &str) -> Option<(String, String, Option<u16>)> {
    let (scheme, rest) = url.split_once("://")?;
    if !matches!(scheme.to_ascii_lowercase().as_str(), "http" | "https")
        || rest.is_empty()
        || rest.contains(['/', '?', '#', '\\', '\0'])
    {
        return None;
    }
    let (host, port) = if let Some((host, port)) = rest.split_once(':') {
        (host, Some(port.parse::<u16>().ok()?))
    } else {
        (rest, None)
    };
    if host.is_empty() || host.bytes().any(|byte| byte < 0x20 || byte == 0x7f) {
        return None;
    }
    Some((scheme.to_ascii_lowercase(), host.to_owned(), port))
}

/// Compares a JSON record field with a bounded text value.
///
/// # Parameters
///
/// - `record`: Input consumed by [`crate::runtime::dynamic::record_field_matches_text_value_logic_local`].
/// - `field_name`: Input consumed by [`crate::runtime::dynamic::record_field_matches_text_value_logic_local`].
/// - `expected`: Input consumed by [`crate::runtime::dynamic::record_field_matches_text_value_logic_local`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::record_field_matches_text_value_logic_local`] holds and `false` otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::record_field_matches_text_value_logic_local(todo!(), todo!(), "value");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn record_field_matches_text_value_logic_local(
    record: &Value,
    field_name: &str,
    expected: &str,
) -> bool {
    !field_name.is_empty() && record.get(field_name).map(value_to_text).as_deref() == Some(expected)
}

/// Performs an ASCII case-insensitive string comparison.
///
/// # Parameters
///
/// - `left`: Input consumed by [`crate::runtime::dynamic::streq_casefold_management_local`].
/// - `right`: Input consumed by [`crate::runtime::dynamic::streq_casefold_management_local`].
///
/// # Returns
///
/// Returns `true` when the condition checked by [`crate::runtime::dynamic::streq_casefold_management_local`] holds and `false` otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::streq_casefold_management_local(todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn streq_casefold_management_local(left: &str, right: &str) -> bool {
    left.eq_ignore_ascii_case(right)
}

/// Ensures the managed media subtree exists below an app root.
///
/// # Parameters
///
/// - `app_root`: Input consumed by [`crate::runtime::dynamic::ensure_media_subtree`].
/// - `business_slug`: Input consumed by [`crate::runtime::dynamic::ensure_media_subtree`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::ensure_media_subtree`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::ensure_media_subtree(std::path::Path::new("apps/dev_docs"), "demo");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn ensure_media_subtree(app_root: &Path, business_slug: &str) -> Option<String> {
    if app_root.as_os_str().is_empty()
        || business_slug.is_empty()
        || business_slug.contains(['/', '\\', '\0'])
    {
        return None;
    }
    let path = super::config::app_media_root(app_root)?
        .join("businesses")
        .join(business_slug);
    super::fs::ensure_directory(&path)
        .eq(&0)
        .then(|| path.to_string_lossy().into_owned())
}

/// Resolves a validated media URL to a file below the app media tree.
///
/// # Parameters
///
/// - `app_root`: Input consumed by [`crate::runtime::dynamic::media_disk_path`].
/// - `business_slug`: Input consumed by [`crate::runtime::dynamic::media_disk_path`].
/// - `media_url`: Input consumed by [`crate::runtime::dynamic::media_disk_path`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::media_disk_path`] can produce a value and [`None`] otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::media_disk_path`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::media_disk_path(std::path::Path::new("apps/dev_docs"), "demo", todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn media_disk_path(app_root: &Path, business_slug: &str, media_url: &str) -> Option<String> {
    if app_root.as_os_str().is_empty() || business_slug.is_empty() || media_url.is_empty() {
        return None;
    }
    let relative = media_url
        .strip_prefix("/media/")
        .or_else(|| media_url.strip_prefix("media/"))?;
    if relative.is_empty()
        || relative
            .split('/')
            .any(|segment| segment.is_empty() || segment == "..")
    {
        return None;
    }
    Some(
        super::config::app_media_root(app_root)?
            .join(business_slug)
            .join(relative)
            .to_string_lossy()
            .into_owned(),
    )
}

/// Normalizes an uploaded filename to one safe basename.
///
/// # Parameters
///
/// - `filename`: Input consumed by [`crate::runtime::dynamic::safe_upload_filename`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::safe_upload_filename`] can produce a value and [`None`] otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::safe_upload_filename`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::safe_upload_filename(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn safe_upload_filename(filename: &str) -> Option<String> {
    let basename = filename.rsplit(['/', '\\']).next()?.trim();
    if basename.is_empty()
        || basename.len() > 255
        || basename == "."
        || basename == ".."
        || basename
            .bytes()
            .any(|byte| byte < 0x20 || byte == 0x7f || byte == 0)
    {
        return None;
    }
    Some(
        basename
            .chars()
            .filter(|character| {
                character.is_ascii_alphanumeric() || matches!(character, '.' | '-' | '_')
            })
            .collect(),
    )
}

/// Returns the request form value from the already buffered request body.
///
/// # Parameters
///
/// - `request`: Input consumed by [`crate::runtime::dynamic::dynamic_form_value`].
/// - `field_name`: Input consumed by [`crate::runtime::dynamic::dynamic_form_value`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::dynamic_form_value`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::dynamic_form_value(todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn dynamic_form_value(request: &HttpRequest, field_name: &str) -> Option<String> {
    request_form_value(request, field_name)
}

/// Returns the value of a cookie field using bounded case-sensitive parsing.
///
/// # Parameters
///
/// - `cookie_header`: Input consumed by [`crate::runtime::dynamic::dynamic_cookie_value`].
/// - `name`: Input consumed by [`crate::runtime::dynamic::dynamic_cookie_value`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::dynamic_cookie_value`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::dynamic_cookie_value("session=abc", "name");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn dynamic_cookie_value(cookie_header: &str, name: &str) -> Option<String> {
    cookie_value(Some(cookie_header), name)
}

/// Extracts the CSRF token from the cookie header or creates a fresh token.
///
/// # Parameters
///
/// - `cookie_header`: Input consumed by [`crate::runtime::dynamic::csrf_token_for_response`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::csrf_token_for_response`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::csrf_token_for_response("session=abc");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn csrf_token_for_response(cookie_header: Option<&str>) -> String {
    pharos_runtime_security::csrf_token_for_response(cookie_header)
}

/// Generates a cryptographically sourced hexadecimal token, failing closed on entropy errors.
///
/// # Parameters
///
/// - `byte_count`: Input consumed by [`crate::runtime::dynamic::csrf_generate_random_hex`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::csrf_generate_random_hex`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::csrf_generate_random_hex(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn csrf_generate_random_hex(byte_count: usize) -> String {
    pharos_runtime_security::csrf_generate_random_hex(byte_count)
}

/// Returns the CSRF cookie token when present.
///
/// # Parameters
///
/// - `cookie_header`: Input consumed by [`crate::runtime::dynamic::csrf_token_from_cookie`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::csrf_token_from_cookie`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::csrf_token_from_cookie("session=abc");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn csrf_token_from_cookie(cookie_header: Option<&str>) -> Option<String> {
    pharos_runtime_security::csrf_token_from_cookie(cookie_header)
}

/// Extracts an address-like value from peer metadata cookie text.
///
/// # Parameters
///
/// - `cookie_header`: Input consumed by [`crate::runtime::dynamic::extract_ip_from_peer_info_cookie`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::extract_ip_from_peer_info_cookie`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::extract_ip_from_peer_info_cookie("session=abc");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn extract_ip_from_peer_info_cookie(cookie_header: &str) -> Option<String> {
    cookie_value(Some(cookie_header), "pharos_peer_ip")
        .filter(|value| value.parse::<std::net::IpAddr>().is_ok())
}

/// Returns a bounded copy of the request body.
///
/// # Parameters
///
/// - `request`: Input consumed by [`crate::runtime::dynamic::request_body_bytes`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::request_body_bytes`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::request_body_bytes(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn request_body_bytes(request: &HttpRequest) -> Option<Vec<u8>> {
    (request.body.len() <= super::http::PHAROS_HTTP_MAX_MULTIPART_BODY_BYTES)
        .then(|| request.body.clone())
}

/// Returns the request body as UTF-8 text when valid.
///
/// # Parameters
///
/// - `request`: Input consumed by [`crate::runtime::dynamic::request_body_text`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::request_body_text`] can produce a value and [`None`] otherwise.
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::request_body_text(todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn request_body_text(request: &HttpRequest) -> Option<String> {
    String::from_utf8(request_body_bytes(request)?).ok()
}

/// Resolves one request header without accepting folded lines.
///
/// # Parameters
///
/// - `request`: Input consumed by [`crate::runtime::dynamic::request_header_value`].
/// - `header_name`: Input consumed by [`crate::runtime::dynamic::request_header_value`].
///
/// # Returns
///
/// Returns [`Some`] when [`crate::runtime::dynamic::request_header_value`] can produce a value and [`None`] otherwise.
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::request_header_value`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::request_header_value(todo!(), "content-type");
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn request_header_value(request: &HttpRequest, header_name: &str) -> Option<String> {
    request.headers.lines().find_map(|line| {
        let (name, value) = line.split_once(':')?;
        name.trim()
            .eq_ignore_ascii_case(header_name)
            .then(|| value.trim().to_owned())
    })
}

/// Persists a validated state document at the app-owned state path.
///
/// # Parameters
///
/// - `app`: Input consumed by [`crate::runtime::dynamic::write_app_state_text_file_runtime_local`].
/// - `state_text`: Input consumed by [`crate::runtime::dynamic::write_app_state_text_file_runtime_local`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::write_app_state_text_file_runtime_local`].
///
/// # Performance
///
/// Performance is proportional to the size of the supplied inputs and any parsed or loaded data consumed by [`crate::runtime::dynamic::write_app_state_text_file_runtime_local`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::write_app_state_text_file_runtime_local(todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn write_app_state_text_file_runtime_local(app: &AppRuntime, state_text: &str) -> i32 {
    let path = app
        .state_path
        .as_deref()
        .filter(|value| !value.is_empty())
        .map(Path::new);
    let Some(path) = path else {
        return 69;
    };
    if state_text.len() > 16 * 1024 * 1024 || serde_json::from_str::<Value>(state_text).is_err() {
        return 64;
    }
    super::fs::write_text_file(path, state_text)
}

/// Builds a complete bounded HTML response envelope.
///
/// # Parameters
///
/// - `status`: Input consumed by [`crate::runtime::dynamic::write_html_response`].
/// - `status_text`: Input consumed by [`crate::runtime::dynamic::write_html_response`].
/// - `html`: Input consumed by [`crate::runtime::dynamic::write_html_response`].
/// - `csrf_token`: Input consumed by [`crate::runtime::dynamic::write_html_response`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::write_html_response`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::write_html_response(400, "400 Bad Request", todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn write_html_response(
    status: u16,
    status_text: &str,
    html: &str,
    csrf_token: Option<&str>,
) -> String {
    let status_text = if status_text.is_empty() {
        super::http::http_status_text(status)
    } else {
        status_text
    };
    let cookie = csrf_token
        .map(|token| format!("Set-Cookie: pharos_csrf={token}; Path=/; SameSite=Lax\r\n"))
        .unwrap_or_default();
    format!(
        "HTTP/1.1 {status} {status_text}\r\nContent-Type: text/html; charset=utf-8\r\nContent-Length: {}\r\nCache-Control: no-store\r\nConnection: close\r\n{cookie}\r\n{html}",
        html.len()
    )
}

/// Adds the CSRF hidden-input fragment to POST forms without altering other markup.
///
/// # Parameters
///
/// - `_app_root`: Input consumed by [`crate::runtime::dynamic::csrf_inject_hidden_inputs`].
/// - `html`: Input consumed by [`crate::runtime::dynamic::csrf_inject_hidden_inputs`].
/// - `csrf_token`: Input consumed by [`crate::runtime::dynamic::csrf_inject_hidden_inputs`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::csrf_inject_hidden_inputs`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::csrf_inject_hidden_inputs(todo!(), todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn csrf_inject_hidden_inputs(_app_root: &Path, html: &str, csrf_token: &str) -> String {
    let escaped_token = super::json_commands::html_escape_text(csrf_token);
    pharos_runtime_dynamic::inject_csrf_hidden_inputs(html, &escaped_token)
}

/// Builds an HTML response with caller-supplied validated header lines.
///
/// # Parameters
///
/// - `status`: Input consumed by [`crate::runtime::dynamic::write_html_response_with_headers`].
/// - `status_text`: Input consumed by [`crate::runtime::dynamic::write_html_response_with_headers`].
/// - `html`: Input consumed by [`crate::runtime::dynamic::write_html_response_with_headers`].
/// - `csrf_token`: Input consumed by [`crate::runtime::dynamic::write_html_response_with_headers`].
/// - `extra_headers`: Input consumed by [`crate::runtime::dynamic::write_html_response_with_headers`].
///
/// # Returns
///
/// Returns the computed value produced by [`crate::runtime::dynamic::write_html_response_with_headers`].
///
/// # Examples
///
/// ```rust,ignore
/// let _result = pharos_native::runtime::dynamic::write_html_response_with_headers(400, "400 Bad Request", todo!(), todo!(), todo!());
/// ```
///
/// # Related
///
/// - [`crate::runtime::dynamic`] for adjacent helpers that share the same runtime surface.
pub fn write_html_response_with_headers(
    status: u16,
    status_text: &str,
    html: &str,
    csrf_token: Option<&str>,
    extra_headers: &str,
) -> String {
    let safe_headers = extra_headers.replace(['\r', '\n'], " ");
    let mut response = write_html_response(status, status_text, html, csrf_token);
    if !safe_headers.is_empty() {
        response = response.replacen("\r\n\r\n", &format!("\r\n{safe_headers}\r\n\r\n"), 1);
    }
    response
}

#[cfg(test)]
mod tests {
    /// Pruning completed work preserves durable delivery identity and active work.
    #[test]
    fn local_outbox_retention_preserves_delivery_sequence() {
        let mut root = json!({});
        for index in 0..1100 {
            assert!(super::enqueue_file_record(
                &mut root,
                (
                    "write_json_local".to_owned(),
                    "test.json".to_owned(),
                    json!({"revision": index})
                )
            ));
            let last = root["local_file_deliveries"]
                .as_array_mut()
                .expect("deliveries")
                .last_mut()
                .expect("new delivery");
            last["delivery_status"] = json!("delivered");
        }
        assert_eq!(root["_local_file_delivery_sequence"], json!(1100));
        assert!(
            root["local_file_deliveries"]
                .as_array()
                .expect("history")
                .len()
                <= 65
        );
        assert!(super::enqueue_file_record(
            &mut root,
            (
                "write_json_local".to_owned(),
                "test.json".to_owned(),
                json!({"revision": 1100})
            )
        ));
        assert_eq!(
            root["local_file_deliveries"]
                .as_array()
                .expect("history")
                .last()
                .expect("new delivery")["id"],
            json!(1101)
        );
        root["_local_file_delivery_sequence"] = json!("invalid");
        assert!(!super::enqueue_file_record(
            &mut root,
            (
                "write_json_local".to_owned(),
                "test.json".to_owned(),
                json!({})
            )
        ));
    }

    /// A delayed earlier replacement blocks a later replacement of the same
    /// path, so a retry cannot overwrite newer output after it becomes visible.
    #[test]
    fn local_outbox_replacements_do_not_overtake_retries() {
        let mut root = json!({"local_file_deliveries": [
            {"id": 1, "relative_path": "same.json", "delivery_status": "retryable", "next_attempt_at": "99999999999"},
            {"id": 2, "relative_path": "same.json", "delivery_status": "queued"}
        ]});
        assert!(!super::process_one_due_file_delivery(
            &AppRuntime::default(),
            &mut root
        ));
        assert_eq!(
            root["local_file_deliveries"][1]["delivery_status"],
            json!("queued")
        );
    }

    use super::{
        append_webhook_receipt, deliver_integration_webhook, dynamic_action_failure_annotations,
        dynamic_action_failure_report_root, dynamic_action_input_overrides,
        dynamic_action_report_annotations, dynamic_append_file_record, dynamic_append_notification,
        dynamic_append_record_primitive, dynamic_append_records_from_input_list_primitive,
        dynamic_append_records_from_related_collection_primitive,
        dynamic_apply_approval_transition,
        dynamic_apply_scoped_assignment_or_bulk_update_primitive,
        dynamic_apply_scoped_status_transition_primitive,
        dynamic_delete_records_by_scope_primitive,
        dynamic_replace_records_from_input_list_primitive, dynamic_set_root_fields_primitive,
        dynamic_store_last_action_report, dynamic_update_record_primitive,
        dynamic_upsert_record_primitive, process_one_due_notification_delivery,
        webhook_request_signature_valid, webhook_signature_header_value_dup_logic_local,
    };
    use crate::config::AppRuntime;
    use crate::runtime::http::HttpRequest;
    use serde_json::json;
    use std::collections::BTreeMap;
    use std::fs;
    use std::path::PathBuf;
    use std::time::{SystemTime, UNIX_EPOCH};

    /// Performs unique test dir.
    fn unique_test_dir(label: &str) -> PathBuf {
        let stamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("test clock")
            .as_nanos();
        std::env::temp_dir().join(format!("pharos-{label}-{stamp}"))
    }

    /// Performs request with form.
    fn request_with_form(headers: &str, body: &str) -> HttpRequest {
        HttpRequest {
            method: "POST".to_owned(),
            target: "/actions/test".to_owned(),
            cookie: None,
            content_type: Some("application/x-www-form-urlencoded".to_owned()),
            headers: headers.to_owned(),
            body: body.as_bytes().to_vec(),
            body_file_path: None,
            body_len: body.len(),
            is_multipart: false,
        }
    }

    #[test]
    /// Performs webhook request signature validation accepts valid current signature and rejects bad one.
    fn webhook_request_signature_validation_accepts_valid_current_signature_and_rejects_bad_one() {
        let payload = r#"{"event":"local_echo"}"#;
        let timestamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("clock")
            .as_secs()
            .to_string();
        let secret_env = "PATH";
        assert!(
            std::env::var(secret_env)
                .ok()
                .is_some_and(|value| !value.is_empty())
        );
        let signature =
            webhook_signature_header_value_dup_logic_local(secret_env, &timestamp, payload)
                .expect("signature");
        let endpoint = json!({
            "endpoint_id": "local-echo",
            "signature": {
                "mode": "hmac_sha256",
                "header": "X-Pharos-Signature",
                "timestamp_header": "X-Pharos-Timestamp",
                "secret_env": secret_env,
                "max_age_seconds": 300
            }
        });
        let valid_request = HttpRequest {
            method: "POST".to_owned(),
            target: "/_pharos/webhooks/local-echo/".to_owned(),
            cookie: None,
            content_type: Some("application/json".to_owned()),
            headers: format!(
                "X-Pharos-Timestamp: {timestamp}\r\nX-Pharos-Signature: {signature}\r\n"
            ),
            body: payload.as_bytes().to_vec(),
            body_file_path: None,
            body_len: payload.len(),
            is_multipart: false,
        };
        assert!(webhook_request_signature_valid(
            &endpoint,
            &valid_request,
            payload
        ));
        let invalid_request = HttpRequest {
            headers: format!(
                "X-Pharos-Timestamp: {timestamp}\r\nX-Pharos-Signature: sha256=deadbeef\r\n"
            ),
            ..valid_request
        };
        assert!(!webhook_request_signature_valid(
            &endpoint,
            &invalid_request,
            payload
        ));
    }

    #[test]
    /// Appends webhook receipt persists verified payload.
    fn append_webhook_receipt_persists_verified_payload() {
        let root = unique_test_dir("dynamic-webhook-receipt");
        fs::create_dir_all(&root).expect("create receipt root");
        let state_path = root.join("runtime-state.json");
        fs::write(&state_path, r#"{"webhook_receipts":[]}"#).expect("write state");
        assert_eq!(
            append_webhook_receipt(&state_path, "local-echo", "verified", r#"{"ok":true}"#),
            0
        );
        let persisted: serde_json::Value =
            serde_json::from_str(&fs::read_to_string(&state_path).expect("read state"))
                .expect("parse state");
        let receipts = persisted["webhook_receipts"].as_array().expect("receipts");
        assert_eq!(receipts.len(), 1);
        assert_eq!(receipts[0]["endpoint_id"], "local-echo");
        assert_eq!(receipts[0]["verification_status"], "verified");
        assert_eq!(receipts[0]["payload"], r#"{"ok":true}"#);
        assert!(
            receipts[0]["created_at"]
                .as_str()
                .is_some_and(|value| !value.is_empty())
        );
        fs::remove_dir_all(root).expect("remove receipt root");
    }

    #[test]
    /// Performs notification email delivery marks missing SMTP config as skipped.
    fn notification_email_delivery_marks_missing_smtp_config_as_skipped() {
        let root = unique_test_dir("dynamic-notification-email");
        fs::create_dir_all(root.join("templates/email")).expect("create template dir");
        fs::write(
            root.join("templates/email/account-proof.yml"),
            "subject: Account proof for {{recipient}}\nbody: Hello {{recipient}}, this is a proof email.\n",
        )
        .expect("write email template");
        let app = AppRuntime {
            app_root: Some(root.to_string_lossy().into_owned()),
            app_id: Some("ucal".to_owned()),
            ..AppRuntime::default()
        };
        let integrations = json!({
            "adapters": [{
                "adapter_id": "account-mail",
                "adapter_kind": "notification_email",
                "relay": "primary",
                "max_attempts": 2,
                "backoff_seconds": 1
            }]
        });
        let mut state = json!({"notification_deliveries": []});
        assert!(dynamic_append_notification(
            &mut state,
            &integrations,
            "account-mail",
            "email",
            "ops@llight.io",
            "account-proof",
            &json!({"recipient":"ops@llight.io"})
        ));
        assert!(process_one_due_notification_delivery(
            &app,
            &mut state,
            &integrations
        ));
        let delivery = &state["notification_deliveries"][0];
        assert_eq!(delivery["adapter_kind"], "notification_email");
        assert_eq!(delivery["delivery_status"], "skipped");
        assert_eq!(delivery["attempt_count"], 1);
        assert_eq!(delivery["last_error"], "email disabled");
        assert_eq!(
            delivery["rendered_subject"],
            "Account proof for ops@llight.io"
        );
        assert_eq!(
            delivery["rendered_body"],
            "Hello ops@llight.io, this is a proof email."
        );
        assert!(
            delivery["completed_at"]
                .as_str()
                .is_some_and(|value| !value.is_empty())
        );
        fs::remove_dir_all(root).expect("remove notification root");
    }

    #[test]
    /// Returns the default inputs fill missing values without overwriting submitted ones.
    fn default_inputs_fill_missing_values_without_overwriting_submitted_ones() {
        let action = json!({
            "hook_contract": {
                "default_inputs": [
                    { "field": "snapshot_label", "source": "literal", "value": "initial-proof" },
                    { "field": "actor", "source": "header", "header": "X-Pharos-Actor" }
                ]
            }
        });
        let request_missing = request_with_form("X-Pharos-Actor: casey", "board_id=4");
        let overrides_missing =
            dynamic_action_input_overrides(&action, &request_missing).expect("default overrides");
        assert_eq!(
            super::dynamic_action_input_value(
                &request_missing,
                &overrides_missing,
                "snapshot_label"
            )
            .as_deref(),
            Some("initial-proof")
        );
        assert_eq!(
            super::dynamic_action_input_value(&request_missing, &overrides_missing, "actor")
                .as_deref(),
            Some("casey")
        );

        let request_present = request_with_form("X-Pharos-Actor: casey", "snapshot_label=manual");
        let overrides_present =
            dynamic_action_input_overrides(&action, &request_present).expect("submitted value");
        assert_eq!(
            super::dynamic_action_input_value(
                &request_present,
                &overrides_present,
                "snapshot_label"
            )
            .as_deref(),
            Some("manual")
        );
    }

    #[test]
    /// Deletes records by scope removes matching rows and counts skips.
    fn delete_records_by_scope_removes_matching_rows_and_counts_skips() {
        let mut root = json!({
            "board_task_snapshots": [
                { "id": 8, "board_id": 4, "snapshot_label": "initial-proof" },
                { "id": 9, "board_id": 4, "snapshot_label": "initial-proof" },
                { "id": 10, "board_id": 4, "snapshot_label": "initial-proof" },
                { "id": 11, "board_id": 8, "snapshot_label": "keep" }
            ]
        });
        let primitive = json!({
            "collection": "board_task_snapshots",
            "scope_match": { "field": "board_id", "from_input": "board_id" },
            "max_records": 2,
            "overflow_policy": "truncate"
        });
        let request = request_with_form("", "board_id=4");
        let mut matched = 0;
        let mut skipped = 0;
        let mut removed = 0;
        dynamic_delete_records_by_scope_primitive(
            &mut root,
            &request,
            &BTreeMap::new(),
            &primitive,
            &mut matched,
            &mut skipped,
            &mut removed,
        )
        .expect("delete primitive");
        let rows = root["board_task_snapshots"]
            .as_array()
            .expect("snapshot rows");
        assert_eq!(matched, 3);
        assert_eq!(skipped, 1);
        assert_eq!(removed, 2);
        assert_eq!(rows.len(), 2);
        assert!(rows.iter().any(|row| row["board_id"] == 8));
    }

    #[test]
    /// Writes JSON local overwrites one bounded snapshot file.
    fn write_json_local_overwrites_one_bounded_snapshot_file() {
        let root = unique_test_dir("dynamic-json-file-record");
        let state_root = root.join("state");
        fs::create_dir_all(state_root.join("exports")).expect("create exports");
        let state_path = state_root.join("runtime-state.json");
        fs::write(&state_path, "{}").expect("write state");
        let integrations = json!({
            "adapters": [{
                "adapter_id": "local-board-snapshot",
                "adapter_kind": "write_json_local",
                "relative_path": "exports/board-snapshot.json"
            }]
        });
        let action = json!({ "entity_id": "board_task_snapshot" });
        let file_record = json!({
            "adapter_id": "local-board-snapshot",
            "record": {
                "event": { "source": "literal", "value": "board_task_snapshots_purged" },
                "board_id": { "source": "input", "field": "board_id" }
            }
        });
        let request = request_with_form("", "board_id=4");
        assert!(dynamic_append_file_record(
            &state_path,
            &integrations,
            &action,
            &file_record,
            &request,
            &BTreeMap::new(),
            "purge_board_task_snapshots",
            "board_task_snapshot",
            "board_task_snapshots_purged",
            0,
        ));
        let output = fs::read_to_string(state_root.join("exports/board-snapshot.json"))
            .expect("read snapshot json");
        let persisted: serde_json::Value =
            serde_json::from_str(&output).expect("parse snapshot json");
        assert_eq!(persisted["adapter_kind"], "write_json_local");
        assert_eq!(persisted["board_id"], "4");
        assert_eq!(persisted["event"], "board_task_snapshots_purged");
        fs::remove_dir_all(root).expect("remove test root");
    }

    /// Preserves explicit local-transport and response-budget policy on queued deliveries.
    #[test]
    fn webhook_delivery_carries_bounded_transport_policy() {
        let mut state = json!({"integration_deliveries": []});
        let integrations = json!({
            "adapters": [{
                "adapter_id": "local_webhook",
                "adapter_kind": "webhook",
                "target_url": "http://127.0.0.1:18781/capture",
                "allow_plaintext_local": true,
                "max_response_bytes": 4096
            }]
        });

        assert_eq!(
            deliver_integration_webhook(
                &mut state,
                &integrations,
                "local_webhook",
                Some("test"),
                Some("appointment:1"),
                None,
                None,
                r#"{"event":"probe"}"#,
            ),
            Some(1)
        );
        let delivery = &state["integration_deliveries"][0];
        assert_eq!(delivery["allow_plaintext_local"], true);
        assert_eq!(delivery["max_response_bytes"], 4096);
    }

    /// Keeps plaintext local delivery disabled when the adapter does not opt in.
    #[test]
    fn webhook_delivery_defaults_to_secure_transport_policy() {
        let mut state = json!({"integration_deliveries": []});
        let integrations = json!({
            "adapters": [{
                "adapter_id": "secure_webhook",
                "adapter_kind": "webhook",
                "target_url": "https://example.test/capture"
            }]
        });

        assert_eq!(
            deliver_integration_webhook(
                &mut state,
                &integrations,
                "secure_webhook",
                None,
                None,
                None,
                None,
                r#"{"event":"probe"}"#,
            ),
            Some(1)
        );
        let delivery = &state["integration_deliveries"][0];
        assert_eq!(delivery["allow_plaintext_local"], false);
        assert_eq!(delivery["max_response_bytes"], 8 * 1024);
    }

    /// Prefers the first duplicate cookie so mounted apps keep the scoped session token.
    #[test]
    fn cookie_lookup_prefers_first_duplicate_name() {
        let cookie_header =
            "pharos_session=scoped-token; other=value; pharos_session=stale-root-token";
        assert_eq!(
            super::dynamic_cookie_value(cookie_header, "pharos_session").as_deref(),
            Some("scoped-token")
        );
    }

    /// Carries explicit auth and event-header policy for the broader HTTP JSON adapter family.
    #[test]
    fn http_json_delivery_carries_auth_and_event_header_policy() {
        let mut state = json!({"integration_deliveries": []});
        let integrations = json!({
            "adapters": [{
                "adapter_id": "board_api",
                "adapter_kind": "http_json",
                "target_url": "http://127.0.0.1:18782/capture",
                "auth_header": "Authorization",
                "auth_scheme": "Bearer",
                "auth_token_env": "PHAROS_BOARD_API_TOKEN",
                "event_header": "X-Pharos-Event",
                "allow_plaintext_local": true
            }]
        });

        assert_eq!(
            deliver_integration_webhook(
                &mut state,
                &integrations,
                "board_api",
                Some("board_created"),
                Some("board:4"),
                None,
                None,
                r#"{"event":"probe"}"#,
            ),
            Some(1)
        );
        let delivery = &state["integration_deliveries"][0];
        assert_eq!(delivery["adapter_kind"], "http_json");
        assert_eq!(delivery["auth_header"], "Authorization");
        assert_eq!(delivery["auth_scheme"], "Bearer");
        assert_eq!(delivery["auth_token_env"], "PHAROS_BOARD_API_TOKEN");
        assert_eq!(delivery["event_header"], "X-Pharos-Event");
    }

    /// Carries the declared form-payload adapter policy for flat object delivery bodies.
    #[test]
    fn http_form_delivery_carries_form_adapter_policy() {
        let mut state = json!({"integration_deliveries": []});
        let integrations = json!({
            "adapters": [{
                "adapter_id": "board_form",
                "adapter_kind": "http_form",
                "target_url": "http://127.0.0.1:18783/capture",
                "content_type": "application/x-www-form-urlencoded",
                "auth_header": "X-Api-Key",
                "auth_scheme": "",
                "auth_token_env": "PHAROS_BOARD_FORM_KEY",
                "event_header": "X-Pharos-Event",
                "allow_plaintext_local": true
            }]
        });

        assert_eq!(
            deliver_integration_webhook(
                &mut state,
                &integrations,
                "board_form",
                Some("board_completed"),
                Some("board:4"),
                None,
                None,
                r#"{"event":"probe","attempt":1}"#,
            ),
            Some(1)
        );
        let delivery = &state["integration_deliveries"][0];
        assert_eq!(delivery["adapter_kind"], "http_form");
        assert_eq!(
            delivery["content_type"],
            "application/x-www-form-urlencoded"
        );
        assert_eq!(delivery["auth_header"], "X-Api-Key");
        assert_eq!(delivery["auth_scheme"], "");
        assert_eq!(delivery["auth_token_env"], "PHAROS_BOARD_FORM_KEY");
        assert_eq!(delivery["event_header"], "X-Pharos-Event");
    }

    #[test]
    /// Performs action report annotations are persisted in last action report.
    fn action_report_annotations_are_persisted_in_last_action_report() {
        let action = json!({
            "hook_contract": {
                "normalize_inputs": [
                    { "kind": "trim", "fields": ["title"] },
                    { "kind": "collapse_whitespace", "fields": ["title"] }
                ],
                "report_annotations": [
                    { "field": "normalized_title", "value": { "source": "input", "field": "title" } },
                    { "field": "request_transport", "value": { "source": "literal", "value": "form" } }
                ]
            }
        });
        let request = request_with_form("", "title=%20%20Native%20%20Proof%20Task%20%20");
        let overrides = dynamic_action_input_overrides(&action, &request).expect("input overrides");
        let annotations = dynamic_action_report_annotations(
            &action,
            &request,
            &overrides,
            "create_task",
            "task",
            "create_task_succeeded",
            41,
        )
        .expect("report annotations");
        let mut root = json!({"action_reports": []});
        assert!(dynamic_store_last_action_report(
            &mut root,
            json!({
                "action_id": "create_task",
                "event_kind": "create_task_succeeded",
                "annotations": annotations
            }),
        ));
        assert_eq!(
            root["last_action_report"]["annotations"]["normalized_title"],
            "Native Proof Task"
        );
        assert_eq!(
            root["last_action_report"]["annotations"]["request_transport"],
            "form"
        );
        assert_eq!(
            root["action_reports"]
                .as_array()
                .expect("action reports")
                .len(),
            1
        );
    }

    #[test]
    /// Performs action failure annotations are persisted in failure report.
    fn action_failure_annotations_are_preserved_in_failure_report() {
        let action = json!({
            "hook_contract": {
                "failure_annotations": [
                    { "field": "attempted_transition", "value": { "source": "literal", "value": "reject" } },
                    { "field": "approval_request_id", "value": { "source": "input", "field": "id" } },
                    { "field": "approver", "value": { "source": "input", "field": "approver" } }
                ]
            }
        });
        let request = request_with_form("", "id=17&approver=owner1");
        let overrides = dynamic_action_input_overrides(&action, &request).expect("input overrides");
        let annotations = dynamic_action_failure_annotations(
            &action,
            &request,
            &overrides,
            "reject_approval",
            "approval_request",
            "reject_approval_failed",
            17,
        )
        .expect("failure annotations");
        assert_eq!(annotations["attempted_transition"], "reject");
        let persisted = dynamic_action_failure_report_root(
            json!({"action_reports": []}),
            &action,
            &request,
            &overrides,
            "reject_approval",
            "approval_request",
            None,
            17,
            422,
        );
        assert_eq!(
            persisted["last_action_report"]["annotations"]["attempted_transition"],
            "reject"
        );
        assert_eq!(
            persisted["last_action_report"]["annotations"]["approval_request_id"],
            "17"
        );
        assert_eq!(
            persisted["last_action_report"]["annotations"]["approver"],
            "owner1"
        );
    }

    #[test]
    /// Appends record creates anchor row.
    fn append_record_creates_anchor_row() {
        let mut root = json!({
            "next_id": 7,
            "boards": []
        });
        let primitive = json!({
            "collection": "boards",
            "record_contract": {
                "id": { "source": "next_id" },
                "name": { "source": "input", "field": "board_name" },
                "workflow_state": { "source": "literal", "value": "created" }
            }
        });
        let request = request_with_form("", "board_name=Roadmap");
        let mut anchor_id = 0;
        let mut created = 0;
        dynamic_append_record_primitive(
            &mut root,
            &request,
            &BTreeMap::new(),
            &primitive,
            &mut anchor_id,
            &mut created,
        )
        .expect("append record");
        assert_eq!(created, 1);
        assert_eq!(anchor_id, 7);
        assert_eq!(root["boards"][0]["id"], 7);
        assert_eq!(root["boards"][0]["name"], "Roadmap");
        assert_eq!(root["boards"][0]["workflow_state"], "created");
    }

    #[test]
    /// Sets root fields applies declared literal and input values.
    fn set_root_fields_applies_declared_literal_and_input_values() {
        let root_dir = unique_test_dir("dynamic-set-root-fields");
        fs::create_dir_all(root_dir.join("data/contracts")).expect("create contract dir");
        fs::write(
            root_dir.join("data/contracts/test-fields.json"),
            r#"{
              "fields": {
                "last_action": { "source": "literal", "value": "board_created" },
                "last_board_name": { "source": "input", "field": "board_name" }
              }
            }"#,
        )
        .expect("write field contract");
        let app = AppRuntime {
            app_root: Some(root_dir.to_string_lossy().into_owned()),
            ..AppRuntime::default()
        };
        let primitive = json!({
            "fields": "test-fields.json#fields"
        });
        let request = request_with_form("", "board_name=Roadmap");
        let mut root = json!({});
        dynamic_set_root_fields_primitive(&app, &mut root, &request, &BTreeMap::new(), &primitive)
            .expect("set root fields");
        assert_eq!(root["last_action"], "board_created");
        assert_eq!(root["last_board_name"], "Roadmap");
        fs::remove_dir_all(root_dir).expect("remove test root");
    }

    #[test]
    /// Updates record updates matching row and missing match fails closed.
    fn update_record_updates_matching_row_and_missing_match_fails_closed() {
        let primitive = json!({
            "collection": "tasks",
            "match": { "field": "id", "from_input": "id" },
            "updates": {
                "done": { "source": "toggle_boolean", "field": "done" },
                "title": { "source": "input", "field": "title" }
            }
        });
        let request = request_with_form("", "id=2&title=Retitled");
        let mut root = json!({
            "tasks": [
                { "id": "1", "title": "One", "done": false },
                { "id": "2", "title": "Two", "done": false }
            ]
        });
        let mut matched = 0;
        let mut updated = 0;
        dynamic_update_record_primitive(
            &mut root,
            &request,
            &BTreeMap::new(),
            &primitive,
            &mut matched,
            &mut updated,
        )
        .expect("update matching row");
        assert_eq!(matched, 1);
        assert_eq!(updated, 1);
        assert_eq!(root["tasks"][1]["title"], "Retitled");
        assert_eq!(root["tasks"][1]["done"], true);
        let missing_request = request_with_form("", "id=99&title=Missing");
        let missing = dynamic_update_record_primitive(
            &mut root,
            &missing_request,
            &BTreeMap::new(),
            &primitive,
            &mut matched,
            &mut updated,
        );
        assert_eq!(missing, Err(404));
    }

    #[test]
    /// Appends records from input list creates rows with anchor and skip counts.
    fn append_records_from_input_list_creates_rows_with_anchor_and_skip_counts() {
        let mut root = json!({
            "next_id": 9,
            "board_tasks": []
        });
        let primitive = json!({
            "collection": "board_tasks",
            "list_source": { "from_input": "task_titles", "delimiter": "," },
            "max_records": 2,
            "overflow_policy": "truncate",
            "record_contract": {
                "id": { "source": "next_id" },
                "board_id": { "source": "context_anchor_created_id" },
                "title": { "source": "list_item" },
                "sort_order": { "source": "list_index" },
                "workflow_state": { "source": "literal", "value": "created" }
            }
        });
        let request = request_with_form("", "task_titles=One,Two,Three");
        let mut skipped = 0;
        let mut created = 0;
        dynamic_append_records_from_input_list_primitive(
            &mut root,
            &request,
            &BTreeMap::new(),
            &primitive,
            44,
            0,
            0,
            &mut skipped,
            &mut created,
        )
        .expect("append records from input list");
        let rows = root["board_tasks"].as_array().expect("board tasks array");
        assert_eq!(created, 2);
        assert_eq!(skipped, 1);
        assert_eq!(rows.len(), 2);
        assert_eq!(rows[0]["id"], 9);
        assert_eq!(rows[0]["board_id"], 44);
        assert_eq!(rows[0]["title"], "One");
        assert_eq!(rows[0]["sort_order"], 1);
        assert_eq!(rows[1]["id"], 10);
        assert_eq!(rows[1]["title"], "Two");
        assert_eq!(rows[1]["sort_order"], 2);
    }

    #[test]
    /// Appends records from related collection copies matched rows and tracks overflow.
    fn append_records_from_related_collection_copies_matched_rows_and_tracks_overflow() {
        let mut root = json!({
            "next_id": 20,
            "board_tasks": [
                { "id": 1, "board_id": 4, "title": "One", "assignee": "sam", "status": "pending" },
                { "id": 2, "board_id": 4, "title": "Two", "assignee": "lee", "status": "pending" },
                { "id": 3, "board_id": 8, "title": "Other", "assignee": "ivy", "status": "done" }
            ],
            "board_task_snapshots": []
        });
        let primitive = json!({
            "collection": "board_task_snapshots",
            "source_collection": "board_tasks",
            "scope_match": { "field": "board_id", "from_input": "board_id" },
            "max_records": 1,
            "record_contract": {
                "id": { "source": "next_id" },
                "board_id": { "source": "source_record_field", "field": "board_id" },
                "source_task_id": { "source": "source_record_field", "field": "id" },
                "title": { "source": "source_record_field", "field": "title" },
                "snapshot_label": { "source": "input", "field": "snapshot_label" }
            }
        });
        let request = request_with_form("", "board_id=4&snapshot_label=Nightly");
        let mut matched = 0;
        let mut skipped = 0;
        let mut created = 0;
        dynamic_append_records_from_related_collection_primitive(
            &mut root,
            &request,
            &BTreeMap::new(),
            &primitive,
            &mut matched,
            &mut skipped,
            &mut created,
        )
        .expect("append records from related collection");
        let rows = root["board_task_snapshots"]
            .as_array()
            .expect("snapshot rows array");
        assert_eq!(matched, 2);
        assert_eq!(skipped, 1);
        assert_eq!(created, 1);
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0]["id"], 20);
        assert_eq!(rows[0]["board_id"], 4);
        assert_eq!(rows[0]["source_task_id"], 1);
        assert_eq!(rows[0]["title"], "One");
        assert_eq!(rows[0]["snapshot_label"], "Nightly");
    }

    #[test]
    /// Performs scoped reassignment updates matching rows and preserves others.
    fn scoped_reassignment_updates_matching_rows_and_preserves_others() {
        let mut root = json!({
            "board_tasks": [
                { "id": 1, "board_id": 4, "assignee": "sam" },
                { "id": 2, "board_id": 4, "assignee": "lee" },
                { "id": 3, "board_id": 8, "assignee": "ivy" }
            ]
        });
        let primitive = json!({
            "collection": "board_tasks",
            "scope_match": { "field": "board_id", "from_input": "board_id" },
            "target_field": "assignee",
            "value_contract": { "source": "input", "field": "new_assignee" },
            "max_records": 1,
            "overflow_policy": "truncate"
        });
        let request = request_with_form("", "board_id=4&new_assignee=alex");
        let mut matched = 0;
        let mut updated = 0;
        let mut skipped = 0;
        dynamic_apply_scoped_assignment_or_bulk_update_primitive(
            &mut root,
            &request,
            &BTreeMap::new(),
            &primitive,
            "scoped_reassignment",
            &mut matched,
            &mut updated,
            &mut skipped,
        )
        .expect("scoped reassignment");
        let rows = root["board_tasks"].as_array().expect("board tasks");
        assert_eq!(matched, 2);
        assert_eq!(updated, 1);
        assert_eq!(skipped, 1);
        assert_eq!(rows[0]["assignee"], "alex");
        assert_eq!(rows[1]["assignee"], "lee");
        assert_eq!(rows[2]["assignee"], "ivy");
    }

    #[test]
    /// Performs bulk update fails closed when scope exceeds strict bound.
    fn bulk_update_fails_closed_when_scope_exceeds_strict_bound() {
        let mut root = json!({
            "board_tasks": [
                { "id": 1, "board_id": 4, "done": false, "status": "pending" },
                { "id": 2, "board_id": 4, "done": false, "status": "pending" }
            ]
        });
        let primitive = json!({
            "collection": "board_tasks",
            "scope_match": { "field": "board_id", "from_input": "board_id" },
            "updates": {
                "done": { "source": "literal", "value": true },
                "status": { "source": "literal", "value": "completed" }
            },
            "max_records": 1,
            "overflow_policy": "fail"
        });
        let request = request_with_form("", "board_id=4");
        let mut matched = 0;
        let mut updated = 0;
        let mut skipped = 0;
        let result = dynamic_apply_scoped_assignment_or_bulk_update_primitive(
            &mut root,
            &request,
            &BTreeMap::new(),
            &primitive,
            "bulk_update",
            &mut matched,
            &mut updated,
            &mut skipped,
        );
        assert_eq!(result, Err(422));
        assert_eq!(matched, 2);
        assert_eq!(updated, 0);
        assert_eq!(skipped, 0);
        assert_eq!(root["board_tasks"][0]["done"], false);
        assert_eq!(root["board_tasks"][1]["status"], "pending");
    }

    #[test]
    /// Performs scoped status transition updates allowed rows and reports skips.
    fn scoped_status_transition_updates_allowed_rows_and_reports_skips() {
        let mut root = json!({
            "board_tasks": [
                { "id": 1, "board_id": 4, "status": "pending", "workflow_state": "created", "done": false },
                { "id": 2, "board_id": 4, "status": "assigned", "workflow_state": "assigned", "done": false },
                { "id": 3, "board_id": 4, "status": "completed", "workflow_state": "done", "done": true },
                { "id": 4, "board_id": 8, "status": "pending", "workflow_state": "created", "done": false }
            ]
        });
        let primitive = json!({
            "collection": "board_tasks",
            "scope_match": { "field": "board_id", "from_input": "board_id" },
            "from_statuses": ["pending", "assigned"],
            "to_status": "completed",
            "to_workflow_state": "done",
            "timestamp_field": "completed_at",
            "actor_field": "completed_by",
            "actor_input": "actor",
            "updates": {
                "done": { "source": "literal", "value": true }
            }
        });
        let request = request_with_form("", "board_id=4&actor=ops-user");
        let mut matched = 0;
        let mut updated = 0;
        let mut skipped = 0;
        dynamic_apply_scoped_status_transition_primitive(
            &mut root,
            &request,
            &BTreeMap::new(),
            &primitive,
            &mut matched,
            &mut updated,
            &mut skipped,
        )
        .expect("scoped status transition");
        let rows = root["board_tasks"].as_array().expect("board tasks");
        assert_eq!(matched, 3);
        assert_eq!(updated, 2);
        assert_eq!(skipped, 1);
        assert_eq!(rows[0]["status"], "completed");
        assert_eq!(rows[0]["workflow_state"], "done");
        assert_eq!(rows[0]["done"], true);
        assert_eq!(rows[0]["completed_by"], "ops-user");
        assert!(
            rows[0]["completed_at"]
                .as_str()
                .is_some_and(|value| !value.is_empty())
        );
        assert_eq!(rows[2]["status"], "completed");
        assert_eq!(rows[3]["status"], "pending");
    }

    #[test]
    /// Performs upsert record updates existing row and creates missing row.
    fn upsert_record_updates_existing_row_and_creates_missing_row() {
        let primitive = json!({
            "collection": "board_status_summaries",
            "match": { "field": "board_id", "from_input": "board_id" },
            "updates": {
                "board_id": { "source": "input", "field": "board_id" },
                "board_status": { "source": "literal", "value": "completed" },
                "completed_task_count": { "source": "context_updated_count" },
                "updated_at": { "source": "now" }
            },
            "record_contract": {
                "id": { "source": "next_id" },
                "board_id": { "source": "input", "field": "board_id" },
                "board_status": { "source": "literal", "value": "completed" },
                "completed_task_count": { "source": "context_updated_count" },
                "created_at": { "source": "now" },
                "updated_at": { "source": "now" }
            }
        });

        let mut existing_root = json!({
            "next_id": 12,
            "board_status_summaries": [
                { "id": 3, "board_id": "4", "board_status": "pending", "completed_task_count": 1, "updated_at": "stale" }
            ]
        });
        let request = request_with_form("", "board_id=4");
        let mut matched = 0;
        let mut updated = 2;
        let mut created = 0;
        dynamic_upsert_record_primitive(
            &mut existing_root,
            &request,
            &BTreeMap::new(),
            &primitive,
            0,
            &mut matched,
            &mut updated,
            0,
            &mut created,
        )
        .expect("upsert existing row");
        let existing = &existing_root["board_status_summaries"][0];
        assert_eq!(matched, 1);
        assert_eq!(updated, 3);
        assert_eq!(created, 0);
        assert_eq!(existing["board_status"], "completed");
        assert_eq!(existing["completed_task_count"], 2);
        assert!(
            existing["updated_at"]
                .as_str()
                .is_some_and(|value| !value.is_empty() && value != "stale")
        );

        let mut missing_root = json!({
            "next_id": 20,
            "board_status_summaries": []
        });
        let mut matched_missing = 0;
        let mut updated_missing = 4;
        let mut created_missing = 0;
        dynamic_upsert_record_primitive(
            &mut missing_root,
            &request,
            &BTreeMap::new(),
            &primitive,
            0,
            &mut matched_missing,
            &mut updated_missing,
            1,
            &mut created_missing,
        )
        .expect("create missing upsert row");
        let created_row = &missing_root["board_status_summaries"][0];
        assert_eq!(matched_missing, 0);
        assert_eq!(updated_missing, 4);
        assert_eq!(created_missing, 1);
        assert_eq!(created_row["id"], 20);
        assert_eq!(created_row["board_id"], "4");
        assert_eq!(created_row["board_status"], "completed");
        assert_eq!(created_row["completed_task_count"], 4);
        assert!(
            created_row["created_at"]
                .as_str()
                .is_some_and(|value| !value.is_empty())
        );
    }

    #[test]
    /// Performs replace records from input list replaces updates and removes in order.
    fn replace_records_from_input_list_replaces_updates_and_removes_in_order() {
        let mut root = json!({
            "next_id": 8,
            "board_tasks": [
                { "id": 3, "board_id": 4, "title": "Legacy 2", "sort_order": 2, "status": "old", "workflow_state": "old" },
                { "id": 1, "board_id": 4, "title": "Legacy 1", "sort_order": 1, "status": "old", "workflow_state": "old" },
                { "id": 9, "board_id": 8, "title": "Other Board", "sort_order": 1, "status": "keep", "workflow_state": "keep" },
                { "id": 2, "board_id": 4, "title": "Legacy 3", "sort_order": 3, "status": "old", "workflow_state": "old" }
            ]
        });
        let primitive = json!({
            "collection": "board_tasks",
            "scope_match": { "field": "board_id", "from_input": "board_id" },
            "list_source": { "from_input": "task_titles", "delimiter": "," },
            "order_field": "sort_order",
            "updates": {
                "title": { "source": "list_item" },
                "sort_order": { "source": "list_index" },
                "status": { "source": "literal", "value": "pending" },
                "workflow_state": { "source": "literal", "value": "created" }
            },
            "record_contract": {
                "id": { "source": "next_id" },
                "board_id": { "source": "input", "field": "board_id" },
                "title": { "source": "list_item" },
                "sort_order": { "source": "list_index" },
                "status": { "source": "literal", "value": "pending" },
                "workflow_state": { "source": "literal", "value": "created" }
            }
        });
        let request = request_with_form("", "board_id=4&task_titles=Replaced+One,Replaced+Two");
        let mut matched = 0;
        let mut created = 0;
        let mut updated = 0;
        let mut skipped = 0;
        let mut removed = 0;
        dynamic_replace_records_from_input_list_primitive(
            &mut root,
            &request,
            &BTreeMap::new(),
            &primitive,
            &mut matched,
            &mut created,
            &mut updated,
            &mut skipped,
            &mut removed,
        )
        .expect("replace primitive should succeed");
        let board_rows = root["board_tasks"]
            .as_array()
            .expect("board tasks should remain an array")
            .iter()
            .filter(|row| row["board_id"] == json!(4))
            .cloned()
            .collect::<Vec<_>>();
        assert_eq!(matched, 3);
        assert_eq!(created, 0);
        assert_eq!(updated, 2);
        assert_eq!(skipped, 0);
        assert_eq!(removed, 1);
        assert_eq!(board_rows.len(), 2);
        assert_eq!(board_rows[0]["id"], 3);
        assert_eq!(board_rows[0]["title"], "Replaced Two");
        assert_eq!(board_rows[0]["sort_order"], 2);
        assert_eq!(board_rows[1]["id"], 1);
        assert_eq!(board_rows[1]["title"], "Replaced One");
        assert_eq!(board_rows[1]["sort_order"], 1);
        assert!(
            root["board_tasks"]
                .as_array()
                .expect("board tasks array")
                .iter()
                .any(|row| row["board_id"] == json!(8) && row["title"] == "Other Board")
        );
    }

    #[test]
    /// Performs replace records from input list counts truncated items as skipped.
    fn replace_records_from_input_list_counts_truncated_items_as_skipped() {
        let mut root = json!({
            "next_id": 4,
            "board_tasks": [
                { "id": 1, "board_id": 4, "title": "Legacy 1", "sort_order": 1 },
                { "id": 2, "board_id": 4, "title": "Legacy 2", "sort_order": 2 }
            ]
        });
        let primitive = json!({
            "collection": "board_tasks",
            "scope_match": { "field": "board_id", "from_input": "board_id" },
            "list_source": { "from_input": "task_titles", "delimiter": "," },
            "order_field": "sort_order",
            "max_records": 2,
            "overflow_policy": "truncate",
            "updates": {
                "title": { "source": "list_item" },
                "sort_order": { "source": "list_index" }
            },
            "record_contract": {
                "id": { "source": "next_id" },
                "board_id": { "source": "input", "field": "board_id" },
                "title": { "source": "list_item" },
                "sort_order": { "source": "list_index" }
            }
        });
        let request = request_with_form("", "board_id=4&task_titles=One,Two,Three,Four");
        let mut matched = 0;
        let mut created = 0;
        let mut updated = 0;
        let mut skipped = 0;
        let mut removed = 0;
        dynamic_replace_records_from_input_list_primitive(
            &mut root,
            &request,
            &BTreeMap::new(),
            &primitive,
            &mut matched,
            &mut created,
            &mut updated,
            &mut skipped,
            &mut removed,
        )
        .expect("replace primitive should truncate within bounds");
        assert_eq!(matched, 2);
        assert_eq!(created, 0);
        assert_eq!(updated, 2);
        assert_eq!(skipped, 2);
        assert_eq!(removed, 0);
    }

    #[test]
    /// Performs approval transition sets expected actor reason and timestamp fields.
    fn approval_transition_sets_expected_actor_reason_and_timestamp_fields() {
        let mut approval = serde_json::Map::from_iter([
            ("id".to_owned(), json!(17)),
            ("status".to_owned(), json!("submitted")),
            ("workflow_state".to_owned(), json!("submitted")),
            ("approved_by".to_owned(), json!("")),
            ("approved_at".to_owned(), json!("")),
            ("rejected_by".to_owned(), json!("stale")),
            ("rejected_at".to_owned(), json!("stale")),
            ("rejection_reason".to_owned(), json!("stale")),
        ]);
        let approve_primitive = json!({
            "transition_kind": "approve",
            "approver_input": "approver",
            "approved_by_field": "approved_by",
            "approved_at_field": "approved_at",
            "rejected_by_field": "rejected_by",
            "rejected_at_field": "rejected_at",
            "rejection_reason_field": "rejection_reason"
        });
        let approve_request = request_with_form("", "approver=owner1");
        dynamic_apply_approval_transition(
            &mut approval,
            &approve_primitive,
            &approve_request,
            &BTreeMap::new(),
        )
        .expect("approve transition");
        assert_eq!(approval.get("status"), Some(&json!("approved")));
        assert_eq!(approval.get("workflow_state"), Some(&json!("approved")));
        assert_eq!(approval.get("approved_by"), Some(&json!("owner1")));
        assert_eq!(approval.get("rejected_by"), Some(&json!("")));
        assert_eq!(approval.get("rejected_at"), Some(&json!("")));
        assert_eq!(approval.get("rejection_reason"), Some(&json!("")));
        assert!(
            approval
                .get("approved_at")
                .and_then(|value| value.as_str())
                .is_some_and(|value| !value.is_empty())
        );

        let mut rejection = serde_json::Map::from_iter([
            ("id".to_owned(), json!(18)),
            ("status".to_owned(), json!("submitted")),
            ("workflow_state".to_owned(), json!("submitted")),
            ("approved_by".to_owned(), json!("owner1")),
            ("approved_at".to_owned(), json!("stale")),
            ("rejected_by".to_owned(), json!("")),
            ("rejected_at".to_owned(), json!("")),
            ("rejection_reason".to_owned(), json!("")),
        ]);
        let reject_primitive = json!({
            "transition_kind": "reject",
            "approver_input": "approver",
            "reason_input": "reason",
            "approved_by_field": "approved_by",
            "approved_at_field": "approved_at",
            "rejected_by_field": "rejected_by",
            "rejected_at_field": "rejected_at",
            "rejection_reason_field": "rejection_reason"
        });
        let reject_request = request_with_form("", "approver=owner2&reason=missing+context");
        dynamic_apply_approval_transition(
            &mut rejection,
            &reject_primitive,
            &reject_request,
            &BTreeMap::new(),
        )
        .expect("reject transition");
        assert_eq!(rejection.get("status"), Some(&json!("rejected")));
        assert_eq!(rejection.get("workflow_state"), Some(&json!("rejected")));
        assert_eq!(rejection.get("approved_by"), Some(&json!("")));
        assert_eq!(rejection.get("rejected_by"), Some(&json!("owner2")));
        assert_eq!(
            rejection.get("rejection_reason"),
            Some(&json!("missing context"))
        );
        assert!(
            rejection
                .get("rejected_at")
                .and_then(|value| value.as_str())
                .is_some_and(|value| !value.is_empty())
        );
    }

    #[test]
    /// Performs local file record adapter returns false when append path is not writable file.
    fn local_file_record_adapter_returns_false_when_append_path_is_not_writable_file() {
        let root = unique_test_dir("dynamic-file-record");
        let state_root = root.join("state");
        fs::create_dir_all(state_root.join("exports")).expect("create exports directory");
        let state_path = state_root.join("runtime-state.json");
        fs::write(&state_path, "{}").expect("write state");
        let integrations = json!({
            "adapters": [{
                "adapter_id": "local-board-history",
                "adapter_kind": "append_jsonl_local",
                "relative_path": "exports"
            }]
        });
        let action = json!({ "entity_id": "board_task" });
        let file_record = json!({
            "adapter_id": "local-board-history",
            "record": {
                "board_id": { "source": "literal", "value": 4 }
            }
        });
        let request = request_with_form("", "");
        assert!(!dynamic_append_file_record(
            &state_path,
            &integrations,
            &action,
            &file_record,
            &request,
            &BTreeMap::new(),
            "append_history",
            "board_task",
            "append_history_succeeded",
            0,
        ));
        fs::remove_dir_all(root).expect("remove test root");
    }
}