fleetforge_runtime/

scheduler.rs

use std::cmp::Ordering;
use std::collections::HashMap;
use std::env;
use std::sync::Arc;
use std::time::{Duration, Instant};

use crate::{event_policy, span_step};
use anyhow::{anyhow, Context, Result};
use base64::engine::general_purpose::STANDARD as BASE64;
use base64::Engine;
use bytes::Bytes;
use chrono::{Duration as ChronoDuration, Utc};
use fleetforge_telemetry::otel::{
    self as telemetry_otel, Counter, Histogram, Meter, SpanId, TraceId,
};
use serde_json::{json, to_value, Map, Value};
use tracing::{debug, error, info, warn};
use tracing_opentelemetry::OpenTelemetrySpanExt;
use uuid::Uuid;

const C2PA_PROFILE_ENV: &str = "FLEETFORGE_C2PA_PROFILE";
const DEFAULT_QUEUE_TARGET_MS: i64 = 30_000;
const MAX_INFLIGHT_TOKENS_ENV: &str = "FLEETFORGE_MAX_INFLIGHT_TOKENS";
const MAX_INFLIGHT_COST_ENV: &str = "FLEETFORGE_MAX_INFLIGHT_COST";
const BACKPRESSURE_DELAY_ENV: &str = "FLEETFORGE_QUEUE_BACKPRESSURE_MS";

use crate::aibom::build_aibom;
use crate::capability;
use crate::executor::{
    artifact_meta_to_value, guardrail_budget_view, BudgetCtx, ExecutorRegistry, StepCtx,
    StepExecutionResult, StepSlo, ToolEventRecord,
};
use crate::guardrails::{PolicyBundle, PolicyDecisionTrace, PolicyEvaluationRecord};
use crate::model::{QueuedStep, RunId, StepExecution, StepId, StepSpec, StepStatus, StepType};
use crate::otel::{record_attribute_str, record_trust_attributes};
use crate::policy::{enforce_policy, PolicyOutcome, RuntimePolicyPack};
use fleetforge_policy::DecisionEffect;
use fleetforge_common::licensing::{feature_allowed, LicensedFeature};
use fleetforge_storage::models::{self, NewOutboxEvent};
use fleetforge_storage::{ArtifactStore, BudgetResult, Storage};
use fleetforge_telemetry::audit::Audit;
use fleetforge_telemetry::context::{RunScope, StepScope, TraceContext};
use fleetforge_trust::{
    capability_signer, generate_c2pa_manifest, mint_capability_token, trust_signer,
    AttestationVault, CapabilityToken, CapabilityTokenSubject, IdentityEvidence, ManifestInput,
    ManifestProfile, PolicyEvidence, Trust, TrustBoundary, TrustOrigin, TrustSource, Untrusted,
};
use tokio::sync::{Mutex, Semaphore};

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum TransparencyScope {
    Disabled,
    Gates,
    Runs,
    Artifacts,
}

impl TransparencyScope {
    fn includes_gates(self) -> bool {
        matches!(
            self,
            TransparencyScope::Gates | TransparencyScope::Runs | TransparencyScope::Artifacts
        )
    }

    fn includes_runs(self) -> bool {
        matches!(self, TransparencyScope::Runs | TransparencyScope::Artifacts)
    }

    fn includes_artifacts(self) -> bool {
        matches!(self, TransparencyScope::Artifacts)
    }
}

#[derive(Clone, Debug)]
struct SloContract {
    tier: Option<String>,
    queue_target_ms: Option<i64>,
    priority_boost: i32,
    error_budget_ratio: Option<f64>,
}

#[derive(Clone, Debug)]
struct SloScore {
    has_contract: bool,
    slack_ms: i64,
    wait_ms: i64,
    priority: i32,
    queue_target_ms: Option<i64>,
    tier: Option<String>,
    error_budget_ratio: Option<f64>,
}

impl SloScore {
    fn compare(&self, other: &Self) -> Ordering {
        match (self.has_contract, other.has_contract) {
            (true, false) => Ordering::Less,
            (false, true) => Ordering::Greater,
            _ => {
                if self.has_contract && other.has_contract {
                    let self_breach = self.slack_ms < 0;
                    let other_breach = other.slack_ms < 0;
                    match (self_breach, other_breach) {
                        (true, false) => return Ordering::Less,
                        (false, true) => return Ordering::Greater,
                        _ => {}
                    }
                    match self.slack_ms.cmp(&other.slack_ms) {
                        Ordering::Equal => {}
                        ord => return ord,
                    }
                }

                match other.priority.cmp(&self.priority) {
                    Ordering::Equal => self.wait_ms.cmp(&other.wait_ms),
                    ord => ord,
                }
            }
        }
    }
}

fn trace_id_from_uuid(uuid: &Uuid) -> TraceId {
    TraceId::from_bytes(*uuid.as_bytes())
}

fn span_id_from_uuid(uuid: &Uuid) -> SpanId {
    let raw = uuid.as_u128();
    let mut span_bits = ((raw >> 64) as u64) ^ (raw as u64);
    if span_bits == 0 {
        span_bits = 1;
    }
    SpanId::from_bytes(span_bits.to_be_bytes())
}

fn extract_slo_contract(spec: &StepSpec) -> Option<SloContract> {
    let slo_value = spec.policy.get("slo")?;
    if !slo_value.is_object() {
        return None;
    }
    Some(SloContract {
        tier: slo_value
            .get("tier")
            .and_then(Value::as_str)
            .map(|s| s.to_string()),
        queue_target_ms: slo_value.get("queue_target_ms").and_then(Value::as_i64),
        priority_boost: slo_value
            .get("priority_boost")
            .and_then(Value::as_i64)
            .unwrap_or(0) as i32,
        error_budget_ratio: slo_value.get("error_budget_ratio").and_then(Value::as_f64),
    })
}

fn compute_slo_score(
    step: &QueuedStep,
    contract: Option<&SloContract>,
    now: chrono::DateTime<Utc>,
) -> SloScore {
    let wait_ms = now
        .signed_duration_since(step.created_at)
        .num_milliseconds()
        .max(0);

    if let Some(contract) = contract {
        let target_ms = contract.queue_target_ms.unwrap_or(DEFAULT_QUEUE_TARGET_MS);
        let slack_ms = target_ms - wait_ms;
        SloScore {
            has_contract: true,
            slack_ms,
            wait_ms,
            priority: i32::from(step.priority) + contract.priority_boost,
            queue_target_ms: contract.queue_target_ms,
            tier: contract.tier.clone(),
            error_budget_ratio: contract.error_budget_ratio,
        }
    } else {
        SloScore {
            has_contract: false,
            slack_ms: i64::MAX,
            wait_ms,
            priority: i32::from(step.priority),
            queue_target_ms: None,
            tier: None,
            error_budget_ratio: None,
        }
    }
}

fn build_step_slo(step: &QueuedStep, now: chrono::DateTime<Utc>) -> Option<StepSlo> {
    let contract = extract_slo_contract(&step.spec)?;
    let score = compute_slo_score(step, Some(&contract), now);
    Some(StepSlo {
        tier: contract.tier.clone(),
        queue_target_ms: Some(contract.queue_target_ms.unwrap_or(DEFAULT_QUEUE_TARGET_MS)),
        observed_queue_ms: Some(score.wait_ms),
        slack_ms: Some(score.slack_ms),
        priority_boost: contract.priority_boost,
        error_budget_ratio: contract.error_budget_ratio,
    })
}

fn resolve_step_providers(spec: &crate::model::StepSpec) -> (Option<String>, Option<String>) {
    let provider = spec
        .inputs
        .get("provider")
        .and_then(Value::as_str)
        .or_else(|| spec.inputs.get("model").and_then(Value::as_str))
        .or_else(|| spec.inputs.get("tool").and_then(Value::as_str))
        .map(|s| s.to_owned());
    let provider_version = spec
        .inputs
        .get("provider_version")
        .or_else(|| spec.inputs.get("model_version"))
        .and_then(Value::as_str)
        .map(|s| s.to_owned());
    (provider, provider_version)
}

fn build_scheduler_payload(
    spec: &crate::model::StepSpec,
    step: &QueuedStep,
    budget: &BudgetCtx,
    slo: Option<&StepSlo>,
) -> Value {
    let execution = serde_json::to_value(&spec.execution).unwrap_or(Value::Null);
    json!({
        "inputs": spec.inputs,
        "policy": spec.policy,
        "execution": execution,
        "step": {
            "id": step.step_id.to_string(),
            "type": spec.r#type,
            "slug": spec.slug,
            "attempt": step.attempt,
            "max_attempts": spec.execution.max_attempts,
        },
        "budget": guardrail_budget_view(budget, slo),
    })
}

#[derive(Debug, Default)]
struct InFlightBudget {
    tokens: i64,
    cost: f64,
}

impl InFlightBudget {
    fn add(&mut self, tokens: i64, cost: f64) {
        if tokens > 0 {
            self.tokens = self.tokens.saturating_add(tokens);
        }
        if cost > 0.0 {
            self.cost += cost;
        }
    }

    fn subtract(&mut self, tokens: i64, cost: f64) {
        if tokens > 0 {
            self.tokens = (self.tokens - tokens).max(0);
        }
        if cost > 0.0 {
            self.cost = (self.cost - cost).max(0.0);
        }
    }
}

#[derive(Clone, Debug)]
struct BreakpointRecord {
    spec: String,
    step_id: Uuid,
}

fn load_debug_breakpoints(debug: &Map<String, Value>) -> Vec<BreakpointRecord> {
    debug
        .get("breakpoints")
        .and_then(Value::as_array)
        .map(|items| {
            items
                .iter()
                .filter_map(|entry| {
                    let obj = entry.as_object()?;
                    let spec = obj.get("spec")?.as_str()?.to_string();
                    let step_id = obj
                        .get("step_id")
                        .and_then(Value::as_str)
                        .and_then(|s| Uuid::parse_str(s).ok())?;
                    Some(BreakpointRecord { spec, step_id })
                })
                .collect()
        })
        .unwrap_or_default()
}

fn store_debug_breakpoints(debug: &mut Map<String, Value>, entries: &[BreakpointRecord]) {
    let list = entries
        .iter()
        .map(|record| {
            let mut obj = Map::new();
            obj.insert("spec".to_string(), Value::String(record.spec.clone()));
            obj.insert(
                "step_id".to_string(),
                Value::String(record.step_id.to_string()),
            );
            Value::Object(obj)
        })
        .collect::<Vec<_>>();
    debug.insert("breakpoints".to_string(), Value::Array(list));

    let specs = entries
        .iter()
        .map(|record| Value::String(record.spec.clone()))
        .collect::<Vec<_>>();
    debug.insert("breakpoint_specs".to_string(), Value::Array(specs));
}

fn set_active_pause(debug: &mut Map<String, Value>, record: &BreakpointRecord) {
    let pause = json!({
        "step_id": record.step_id.to_string(),
        "spec": record.spec,
        "started_at": chrono::Utc::now().to_rfc3339(),
    });
    debug.insert("active_pause".to_string(), pause);
}

#[derive(Clone, Debug)]
struct TraceHeaders {
    traceparent: String,
    tracestate: Option<String>,
    baggage: Option<String>,
}

fn parse_trace_headers(value: &Value) -> Option<TraceHeaders> {
    let trace_value = value.get("trace")?;
    let trace_obj = trace_value.as_object()?;
    let traceparent = trace_obj
        .get("traceparent")
        .and_then(Value::as_str)?
        .to_string();
    let tracestate = trace_obj
        .get("tracestate")
        .and_then(Value::as_str)
        .map(|s| s.to_string());
    let baggage = trace_obj
        .get("baggage")
        .and_then(Value::as_str)
        .map(|s| s.to_string());
    Some(TraceHeaders {
        traceparent,
        tracestate,
        baggage,
    })
}

/// Tunables that shape how the scheduler polls, leases, and back-pressures work.
#[derive(Debug, Clone)]
pub struct SchedulerConfig {
    pub worker_id: String,
    pub batch_size: i64,
    pub idle_backoff: Duration,
    pub active_backoff: Duration,
    pub reaper_interval: Duration,
    pub requeue_batch_size: i64,
    pub max_inflight_tokens: Option<i64>,
    pub max_inflight_cost: Option<f64>,
    pub backpressure_delay: Duration,
    pub transparency_writer: bool,
    pub transparency_scope: TransparencyScope,
}

impl Default for SchedulerConfig {
    fn default() -> Self {
        let scope = transparency_scope_from_env();
        let max_inflight_tokens = env::var(MAX_INFLIGHT_TOKENS_ENV)
            .ok()
            .and_then(|value| value.parse::<i64>().ok())
            .filter(|value| *value > 0);
        let max_inflight_cost = env::var(MAX_INFLIGHT_COST_ENV)
            .ok()
            .and_then(|value| value.parse::<f64>().ok())
            .filter(|value| *value > 0.0);
        let backpressure_delay = env::var(BACKPRESSURE_DELAY_ENV)
            .ok()
            .and_then(|value| value.parse::<u64>().ok())
            .map(Duration::from_millis)
            .unwrap_or_else(|| Duration::from_millis(50));
        Self {
            worker_id: format!("worker-{}", uuid::Uuid::new_v4()),
            batch_size: 8,
            idle_backoff: Duration::from_millis(200),
            active_backoff: Duration::from_millis(10),
            reaper_interval: Duration::from_secs(30),
            requeue_batch_size: 128,
            max_inflight_tokens,
            max_inflight_cost,
            backpressure_delay,
            transparency_writer: transparency_writer_flag()
                && !matches!(scope, TransparencyScope::Disabled),
            transparency_scope: scope,
        }
    }
}

/// Core runtime loop that leases steps from storage and hands them to executors.
pub struct Scheduler {
    storage: Arc<Storage>,
    executors: Arc<ExecutorRegistry>,
    policy_pack: Arc<RuntimePolicyPack>,
    artifact_store: Arc<dyn ArtifactStore>,
    attestation_vault: Arc<dyn AttestationVault>,
    config: SchedulerConfig,
    metrics: QueueMetrics,
    concurrency: Arc<Semaphore>,
    audit: Arc<Audit>,
    trace_headers: Mutex<HashMap<RunId, Option<TraceHeaders>>>,
    inflight: Mutex<InFlightBudget>,
}

fn transparency_writer_flag() -> bool {
    let enabled = matches!(
        std::env::var("FLEETFORGE_TRANSPARENCY_WRITER")
            .ok()
            .map(|value| value.trim().to_ascii_lowercase())
            .unwrap_or_default()
            .as_str(),
        "1" | "true" | "yes" | "on"
    );
    if enabled && !feature_allowed(LicensedFeature::ScittTransparency) {
        warn!(
            "SCITT transparency writer requested but the current license tier does not allow it; disabling writer"
        );
        return false;
    }
    enabled
}

fn transparency_scope_from_env() -> TransparencyScope {
    match std::env::var("FLEETFORGE_TRANSPARENCY_SCOPE")
        .ok()
        .map(|value| value.trim().to_ascii_lowercase())
        .as_deref()
    {
        Some("artifacts") => TransparencyScope::Artifacts,
        Some("runs") => TransparencyScope::Runs,
        Some("gates") => TransparencyScope::Gates,
        Some("disabled") => TransparencyScope::Disabled,
        _ => TransparencyScope::Gates,
    }
}

impl Scheduler {
    pub fn new(
        storage: Arc<Storage>,
        executors: Arc<ExecutorRegistry>,
        policy_pack: Arc<RuntimePolicyPack>,
        audit: Arc<Audit>,
        attestation_vault: Arc<dyn AttestationVault>,
        config: SchedulerConfig,
    ) -> Self {
        let artifact_store = storage.artifacts();
        let metrics = QueueMetrics::new();
        let permits_i64 = std::cmp::max(config.batch_size, 1);
        let permits = permits_i64.min(usize::MAX as i64) as usize;
        Self {
            storage,
            executors,
            policy_pack,
            artifact_store,
            attestation_vault,
            config,
            metrics,
            concurrency: Arc::new(Semaphore::new(permits)),
            audit,
            trace_headers: Mutex::new(HashMap::new()),
            inflight: Mutex::new(InFlightBudget::default()),
        }
    }

    /// Starts the scheduling loop, spawning workers and the lease reaper.
    pub async fn run(self: Arc<Self>) -> Result<()> {
        let reaper = Arc::clone(&self);
        tokio::spawn(async move {
            reaper.lease_reaper_loop().await;
        });

        loop {
            let claimed = self
                .claim_ready_steps()
                .await
                .context("failed to claim ready steps")?;

            self.metrics.record_queue_depth(claimed.len());

            if claimed.is_empty() {
                tokio::time::sleep(self.config.idle_backoff).await;
                continue;
            }

            for step in claimed {
                let scheduler = Arc::clone(&self);
                let semaphore = Arc::clone(&self.concurrency);
                tokio::spawn(async move {
                    let _permit = semaphore
                        .acquire_owned()
                        .await
                        .expect("scheduler semaphore closed");
                    if let Err(err) = scheduler.execute_step(step).await {
                        error!(error = ?err, "step execution failed");
                    }
                });
            }

            tokio::time::sleep(self.config.active_backoff).await;
        }
    }

    async fn lease_reaper_loop(self: Arc<Self>) {
        loop {
            let sleep = match self
                .storage
                .requeue_expired_leases(self.config.requeue_batch_size)
                .await
            {
                Ok(requeued) => {
                    if !requeued.is_empty() {
                        info!(
                            count = requeued.len(),
                            "requeued expired leases back to the queue"
                        );
                        for (run_id, step_id) in &requeued {
                            debug!(%run_id, %step_id, "expired lease requeued");
                        }
                        // Run another sweep quickly to drain backlog.
                        self.config.active_backoff
                    } else {
                        self.config.reaper_interval
                    }
                }
                Err(err) => {
                    warn!(error = %err, "lease reaper iteration failed");
                    self.config.reaper_interval
                }
            };

            tokio::time::sleep(sleep).await;
        }
    }

    async fn trace_headers_for_run(&self, run_id: RunId) -> Result<Option<TraceHeaders>> {
        {
            let cache = self.trace_headers.lock().await;
            if let Some(entry) = cache.get(&run_id) {
                return Ok(entry.clone());
            }
        }

        let run = self
            .storage
            .fetch_run(Uuid::from(run_id))
            .await
            .context("failed to load run for trace context")?;
        let headers = run.and_then(|row| parse_trace_headers(&row.input_ctx));
        let mut cache = self.trace_headers.lock().await;
        cache.insert(run_id, headers.clone());
        Ok(headers)
    }

    async fn claim_ready_steps(&self) -> Result<Vec<QueuedStep>> {
        let db_steps = self
            .storage
            .claim_queued_steps(self.config.batch_size, &self.config.worker_id)
            .await?;

        let mut steps = db_steps
            .into_iter()
            .map(QueuedStep::try_from)
            .collect::<Result<Vec<_>>>()?;
        self.rank_steps(&mut steps);
        Ok(steps)
    }

    async fn execute_step(&self, step: QueuedStep) -> Result<()> {
        let mut spec = step.spec.clone();
        let mut input_snapshot_value = spec.inputs.clone();
        let (mut provider, mut provider_version) = resolve_step_providers(&spec);
        let mut policy_event_log: Option<(String, String, Option<String>, DecisionEffect)> = None;
        let pack_id = self.policy_pack.id().to_string();

        let observed_at = Utc::now();
        if let Ok(wait) = observed_at.signed_duration_since(step.created_at).to_std() {
            self.metrics.record_queue_lag(wait);
        }
        let step_slo = build_step_slo(&step, observed_at);
        if step.attempt > 0 {
            self.metrics.record_retry();
        }

        if self.try_pause_for_breakpoint(&step).await? {
            return Ok(());
        }

        if let Some(deadline_at) = step.deadline_at {
            if Utc::now() > deadline_at {
                let error_payload = json!({
                    "message": "step deadline exceeded",
                    "kind": "deadline_exceeded",
                });
                self.finish_step(
                    &step,
                    &spec,
                    StepStatus::Failed,
                    step.attempt + 1,
                    None,
                    Some(error_payload),
                    None,
                    &BudgetCtx::default(),
                    Some(input_snapshot_value.clone()),
                    None,
                    None,
                    None,
                    None,
                    Vec::new(),
                    Vec::new(),
                    None,
                    None,
                    step_slo.as_ref(),
                )
                .await?;

                let run_trace = self.build_run_trace_for_step(&step).await?;
                self.schedule_compensation(&step, &run_trace).await?;
                return Ok(());
            }
        }

        let mut policy_records: Vec<PolicyEvaluationRecord> = Vec::new();
        let policy_outcome = enforce_policy(self.policy_pack.as_ref(), step.run_id, &spec).await?;
        let policy_record = policy_outcome_to_record(
            &policy_outcome,
            step.run_id,
            step.step_id,
            TrustBoundary::Scheduler,
            &pack_id,
        );
        policy_records.push(policy_record.clone());

        self.attestation_vault
            .record(policy_outcome.attestation.clone())
            .await
            .context("failed to persist policy attestation")?;

        let att_id_str = policy_outcome.attestation.id.to_string();
        let subject_label = format!("run:{}:step:{}", step.run_id, step.step_id);
        record_trust_attributes(
            &[att_id_str.clone()],
            Some(subject_label.as_str()),
            Some(att_id_str.as_str()),
        );

        match policy_outcome.decision.effect {
            DecisionEffect::Deny => {
                let reason = policy_outcome
                    .decision
                    .reason
                    .clone()
                    .unwrap_or_else(|| "policy denied execution".to_string());
                warn!(
                    run = %step.run_id,
                    step = %step.step_id,
                    policy = %pack_id,
                    "policy denied execution: {reason}"
                );

                let (artifact_value, summary) = persist_policy_decisions(
                    Arc::clone(&self.artifact_store),
                    step.run_id,
                    step.step_id,
                    &policy_records,
                )
                .await
                .unwrap_or_else(|err| {
                    warn!(error = %err, "failed to persist policy decisions for denial");
                    (Value::Null, Value::Null)
                });
                let artifact_payload = if artifact_value.is_null() {
                    None
                } else {
                    Some(artifact_value.clone())
                };

                let artifact_id = artifact_value
                    .get("sha256")
                    .and_then(Value::as_str)
                    .map(|s| s.to_string())
                    .or_else(|| {
                        artifact_value
                            .get("artifact_id")
                            .and_then(Value::as_str)
                            .map(|s| s.to_string())
                    });

                let summary_str = summary
                    .get("reasons")
                    .and_then(Value::as_array)
                    .and_then(|arr| arr.first())
                    .and_then(Value::as_str)
                    .map(|s| s.to_string())
                    .unwrap_or_else(|| reason.clone());

                policy_event_log = Some((
                    pack_id.clone(),
                    summary_str.clone(),
                    artifact_id.clone(),
                    policy_outcome.decision.effect,
                ));

                let error_payload = json!({
                    "message": reason,
                    "kind": "policy_denied",
                });
                self.finish_step(
                    &step,
                    &spec,
                    StepStatus::Failed,
                    step.attempt,
                    None,
                    Some(error_payload),
                    None,
                    &BudgetCtx::default(),
                    Some(input_snapshot_value),
                    None,
                    provider.as_deref(),
                    provider_version.as_deref(),
                    None,
                    None,
                    Vec::new(),
                    policy_records,
                    None,
                    artifact_payload,
                )
                .await?;
                return Ok(());
            }
            DecisionEffect::Redact => {
                if let Some(reason) = policy_outcome.decision.reason.as_deref() {
                    debug!(
                        run = %step.run_id,
                        step = %step.step_id,
                        policy = %pack_id,
                        "policy requested redaction: {reason}"
                    );
                }
            }
            DecisionEffect::Allow => {
                if let Some(reason) = policy_outcome.decision.reason.as_deref() {
                    debug!(
                        run = %step.run_id,
                        step = %step.step_id,
                        policy = %pack_id,
                        "policy allow with advisory: {reason}"
                    );
                }
            }
        }

        let mut input_snapshot = Some(input_snapshot_value.clone());

        let budget_request = BudgetRequest::from_step(&step);
        if self.should_backpressure(&budget_request).await {
            self.defer_for_backpressure(&step).await?;
            return Ok(());
        }
        let mut budget_ctx = BudgetCtx::default();
        match self.reserve_budget(step.run_id, &budget_request).await? {
            BudgetDecision::Granted(ctx) => {
                budget_ctx = ctx;
            }
            BudgetDecision::Unlimited(ctx) => {
                budget_ctx = ctx;
            }
            BudgetDecision::Denied { reason } => {
                let error_payload = json!({
                    "message": reason,
                    "kind": "budget_exceeded",
                });
                self.finish_step(
                    &step,
                    &spec,
                    StepStatus::Failed,
                    step.attempt,
                    None,
                    Some(error_payload),
                    None,
                    &BudgetCtx::default(),
                    input_snapshot.clone(),
                    None,
                    None,
                    provider.as_deref(),
                    provider_version.as_deref(),
                    None,
                    None,
                    Vec::new(),
                    Vec::new(),
                    None,
                    None,
                    step_slo.as_ref(),
                )
                .await?;
                return Ok(());
            }
        }
        let policy_decision_id = policy_outcome.attestation.id;

        if let Err(err) = self
            .storage
            .update_step_estimate(
                Uuid::from(step.run_id),
                Uuid::from(step.step_id),
                budget_ctx.tokens_reserved,
                budget_ctx.cost_reserved,
            )
            .await
        {
            warn!(
                error = %err,
                run = %step.run_id,
                step = %step.step_id,
                "failed to persist budget estimate"
            );
        }

        self.metrics.record_budget_projection(&budget_ctx);

        let executor = self
            .executors
            .get(spec.r#type)
            .ok_or_else(|| anyhow!("no executor registered for kind {}", spec.r#type))?;

        let run_uuid = Uuid::from(step.run_id);
        let run_span_id = span_id_from_uuid(&run_uuid);
        let trace_headers = self.trace_headers_for_run(step.run_id).await?;
        let run_scope = RunScope::new(step.run_id.to_string());
        let run_trace = if let Some(headers) = trace_headers {
            match TraceContext::from_w3c(
                run_scope.clone(),
                &headers.traceparent,
                headers.tracestate.as_deref(),
                headers.baggage.as_deref(),
            ) {
                Ok(ctx) => ctx.with_span_id(run_span_id),
                Err(err) => {
                    warn!(
                        run = %step.run_id,
                        error = %err,
                        "failed to parse trace headers; using fallback trace id"
                    );
                    TraceContext::new(run_scope.clone())
                        .with_trace_id(trace_id_from_uuid(&run_uuid))
                        .with_span_id(run_span_id)
                }
            }
        } else {
            TraceContext::new(run_scope.clone())
                .with_trace_id(trace_id_from_uuid(&run_uuid))
                .with_span_id(run_span_id)
        };
        let (mut step_span, step_trace) = span_step!(
            parent: &run_trace,
            step: &step,
            spec: &spec,
            worker: &self.config.worker_id
        );
        let _span_guard = step_span.enter();

        let guardrails = PolicyBundle::for_policy(&spec.policy);

        if !guardrails.is_empty() {
            let scheduler_payload =
                build_scheduler_payload(&spec, &step, &budget_ctx, step_slo.as_ref());
            let outcome = guardrails
                .evaluate(
                    TrustBoundary::Scheduler,
                    Some(step.run_id),
                    Some(step.step_id),
                    scheduler_payload,
                )
                .await?;

            match outcome.effect {
                DecisionEffect::Deny => {
                    let mut events = guardrails.drain_events();
                    if !policy_records.is_empty() {
                        events.extend(policy_records.clone());
                    }
                    let mut policy_artifact: Option<(Value, Value)> = None;
                    let mut artifact_id: Option<String> = None;
                    if !events.is_empty() {
                        let policy_pack = policy_pack_name(&spec.policy);
                        let dominant_effect = dominant_policy_effect(&events);
                        let summary_text = summarize_policy_events(&events);
                        let effect_label = decision_effect_label(dominant_effect);
                        match persist_policy_decisions(
                            Arc::clone(&self.artifact_store),
                            step.run_id,
                            step.step_id,
                            &events,
                        )
                        .await
                        {
                            Ok((artifact_meta, summary)) => {
                                artifact_id = artifact_meta
                                    .get("sha256")
                                    .and_then(Value::as_str)
                                    .map(|s| s.to_string())
                                    .or_else(|| {
                                        artifact_meta
                                            .get("artifact_id")
                                            .and_then(Value::as_str)
                                            .map(|s| s.to_string())
                                    });
                                policy_artifact = Some((artifact_meta, summary));
                            }
                            Err(err) => {
                                warn!(
                                    error = %err,
                                    run = %step.run_id,
                                    step = %step.step_id,
                                    "failed to persist policy decisions"
                                );
                            }
                        }
                        step_span.record("fleetforge_policy_pack", policy_pack.as_str());
                        step_span.record("fleetforge_policy_effect", effect_label.as_str());
                        event_policy!(
                            pack: policy_pack.as_str(),
                            effect: effect_label.as_str(),
                            summary: summary_text.as_str(),
                            artifact: artifact_id.as_deref()
                        );
                    }

                    let mut error_payload = json!({
                        "message": outcome.summary(),
                        "kind": "policy_denied",
                    });
                    if let Some((ref artifact_meta, ref summary)) = policy_artifact {
                        attach_policy_summary(
                            &mut error_payload,
                            summary.clone(),
                            artifact_meta.clone(),
                        );
                    }
                    self.finish_step(
                        &step,
                        &spec,
                        StepStatus::Failed,
                        step.attempt,
                        None,
                        Some(error_payload),
                        None,
                        &budget_ctx,
                        Some(input_snapshot_value.clone()),
                        None,
                        provider.as_deref(),
                        provider_version.as_deref(),
                        None,
                        None,
                        Vec::new(),
                        events,
                        None,
                        None,
                        step_slo.as_ref(),
                    )
                    .await?;
                    return Ok(());
                }
                DecisionEffect::Redact => {
                    if let Some(inputs) = outcome.value.get("inputs").cloned() {
                        spec.inputs = inputs;
                    }
                    if let Some(policy_value) = outcome.value.get("policy").cloned() {
                        spec.policy = policy_value;
                    }
                    if let Some(execution_value) = outcome.value.get("execution") {
                        if let Ok(updated_execution) =
                            serde_json::from_value(execution_value.clone())
                        {
                            spec.execution = updated_execution;
                        }
                    }
                    input_snapshot_value = spec.inputs.clone();
                    input_snapshot = Some(input_snapshot_value.clone());
                    let (prov, prov_version) = resolve_step_providers(&spec);
                    provider = prov;
                    provider_version = prov_version;
                }
                DecisionEffect::Allow => {}
            }
        }

        self.storage
            .mark_step_running(
                Uuid::from(step.run_id),
                Uuid::from(step.step_id),
                &self.config.worker_id,
                Some(&input_snapshot_value),
                provider.as_deref(),
                provider_version.as_deref(),
            )
            .await
            .context("failed to mark step running")?;

        self.add_inflight(budget_ctx.tokens_reserved, budget_ctx.cost_reserved)
            .await;

        let untrusted_inputs = Untrusted(spec.inputs.clone());
        let attempt_number = step.attempt + 1;
        let capability_token = match self.issue_capability_token(
            step.run_id,
            step.step_id,
            attempt_number,
            &spec,
            &budget_ctx,
        ) {
            Ok(token) => {
                let token_id = token.token_id().to_string();
                step_span.record("trust.capability_token_id", token_id.as_str());
                record_attribute_str("trust.capability_token_id", &token_id);
                Some(token)
            }
            Err(err) => {
                warn!(
                    run = %step.run_id,
                    step = %step.step_id,
                    error = %err,
                    "failed to mint capability token"
                );
                let mut error_payload = json!({
                    "message": "capability token mint failed",
                    "kind": "capability_mint_failed",
                    "error": err.to_string(),
                });
                if let Some(obj) = error_payload.as_object_mut() {
                    obj.insert(
                        "step".into(),
                        json!({ "run_id": step.run_id.to_string(), "step_id": step.step_id.to_string() }),
                    );
                }
                self.finish_step(
                    &step,
                    &spec,
                    StepStatus::Failed,
                    step.attempt,
                    None,
                    Some(error_payload),
                    None,
                    &budget_ctx,
                    Some(input_snapshot_value.clone()),
                    None,
                    provider.as_deref(),
                    provider_version.as_deref(),
                    Some(&step_trace),
                    Some(&run_trace),
                    Vec::new(),
                    Vec::new(),
                    None,
                    None,
                    step_slo.as_ref(),
                )
                .await?;
                return Ok(());
            }
        };

        if let Some(violation) = self.validate_capability_guard(
            capability_token.as_ref(),
            &spec,
            &budget_ctx,
            step.run_id,
            step.step_id,
        )? {
            let summary_text = violation
                .record
                .reasons
                .first()
                .cloned()
                .unwrap_or_else(|| violation.reason.clone());
            step_span.record("fleetforge_policy_pack", "capability_guard");
            step_span.record("fleetforge_policy_effect", "deny");

            let records = vec![violation.record.clone()];
            let (artifact_meta, summary) = persist_policy_decisions(
                Arc::clone(&self.artifact_store),
                step.run_id,
                step.step_id,
                &records,
            )
            .await?;

            let artifact_id = artifact_meta
                .get("sha256")
                .and_then(Value::as_str)
                .map(|s| s.to_string())
                .or_else(|| {
                    artifact_meta
                        .get("artifact_id")
                        .and_then(Value::as_str)
                        .map(|s| s.to_string())
                });

            event_policy!(
                pack: "capability_guard",
                effect: "deny",
                summary: summary_text.as_str(),
                artifact: artifact_id.as_deref()
            );

            let mut error_payload = json!({
                "message": summary_text,
                "kind": "capability_denied",
            });
            if let Some(token_id) = violation.token_id.as_ref() {
                if let Some(obj) = error_payload.as_object_mut() {
                    obj.insert(
                        "capability_token_id".into(),
                        Value::String(token_id.clone()),
                    );
                }
            }
            if let Some(token) = capability_token.as_ref() {
                if let Some(obj) = error_payload.as_object_mut() {
                    obj.insert(
                        "capability_token".into(),
                        to_value(token).expect("capability token is serialisable"),
                    );
                }
            }
            attach_policy_summary(&mut error_payload, summary.clone(), artifact_meta.clone());

            warn!(
                run = %step.run_id,
                step = %step.step_id,
                reason = %violation.reason,
                "capability guard denied step execution"
            );

            self.finish_step(
                &step,
                &spec,
                StepStatus::Failed,
                step.attempt,
                None,
                Some(error_payload),
                None,
                &budget_ctx,
                Some(input_snapshot_value.clone()),
                None,
                provider.as_deref(),
                provider_version.as_deref(),
                Some(&step_trace),
                Some(&run_trace),
                Vec::new(),
                records,
                None,
                None,
                step_slo.as_ref(),
            )
            .await?;
            return Ok(());
        }
        let idempotency_key = format!("{}:{}:{}", step.run_id, step.step_id, attempt_number);
        let ctx = StepCtx {
            run_id: step.run_id,
            step_id: step.step_id,
            attempt: attempt_number,
            max_attempts: step.max_attempts,
            spec: spec.clone(),
            untrusted_inputs,
            artifacts: self.artifact_store.clone(),
            policy: Arc::clone(&self.policy_pack),
            policy_decision_id: Some(policy_decision_id),
            guardrails,
            trust: spec.trust.clone(),
            trust_origin: spec.trust_origin.clone(),
            budget: budget_ctx.clone(),
            trace: step_trace.clone(),
            checkpoint: step.checkpoint.clone(),
            idempotency_key,
            capability_token: capability_token.clone(),
            slo: step_slo.clone(),
        };

        let start = Instant::now();
        let execution = executor.execute(&ctx).await;
        let latency = start.elapsed();
        step_span.record("fleetforge_duration_ms", latency.as_secs_f64() * 1000.0);
        let mut guardrail_events = ctx.guardrails.drain_events();
        if !policy_records.is_empty() {
            guardrail_events.extend(policy_records.clone());
        }
        let mut policy_artifact = None;
        if !guardrail_events.is_empty() {
            let mut policy_pack = policy_pack_name(&spec.policy);
            if policy_pack == "baseline" && !policy_records.is_empty() {
                policy_pack = pack_id.clone();
            }
            let dominant_effect = dominant_policy_effect(&guardrail_events);
            let summary_text = summarize_policy_events(&guardrail_events);
            match persist_policy_decisions(
                ctx.artifacts.clone(),
                step.run_id,
                step.step_id,
                &guardrail_events,
            )
            .await
            {
                Ok(value) => {
                    let (artifact_meta, summary) = value;
                    let artifact_id = artifact_meta
                        .get("sha256")
                        .and_then(Value::as_str)
                        .map(|s| s.to_string())
                        .or_else(|| {
                            artifact_meta
                                .get("artifact_id")
                                .and_then(Value::as_str)
                                .map(|s| s.to_string())
                        });
                    policy_event_log = Some((
                        policy_pack.clone(),
                        summary_text.clone(),
                        artifact_id.clone(),
                        dominant_effect,
                    ));
                    policy_artifact = Some((artifact_meta, summary));
                }
                Err(err) => {
                    warn!(
                        error = %err,
                        run = %step.run_id,
                        step = %step.step_id,
                        "failed to persist policy decisions"
                    );
                    policy_event_log = Some((
                        policy_pack.clone(),
                        summary_text.clone(),
                        None,
                        dominant_effect,
                    ));
                }
            }
            if policy_event_log.is_none() {
                policy_event_log = Some((
                    policy_pack.clone(),
                    summary_text.clone(),
                    None,
                    dominant_effect,
                ));
            }
        }

        if let Some((pack, summary, artifact_id, effect)) = &policy_event_log {
            let effect_label = decision_effect_label(*effect);
            step_span.record("fleetforge_policy_pack", pack.as_str());
            step_span.record("fleetforge_policy_effect", effect_label.as_str());
            event_policy!(
                pack: pack.as_str(),
                effect: effect_label.as_str(),
                summary: summary.as_str(),
                artifact: artifact_id.as_deref()
            );
            tracing::info!(
                target: "fleetforge.policy",
                run = %step.run_id,
                step = %step.step_id,
                pack = %pack,
                effect = %effect_label,
                summary = %summary,
                artifact = %artifact_id.clone().unwrap_or_default(),
                "policy evaluation"
            );
        }

        match execution {
            Ok(mut result) => {
                if result.capability_token.is_none() {
                    if let Some(token) = ctx.capability_token.clone() {
                        result.capability_token = Some(token);
                    }
                }
                let StepExecutionResult {
                    mut output,
                    usage,
                    provider: result_provider,
                    provider_version: result_provider_version,
                    tool_events,
                    checkpoint,
                    capability_token: result_capability_token,
                } = result;

                let capability_token =
                    result_capability_token.or_else(|| ctx.capability_token.clone());

                if let Some((artifact_meta, summary)) = policy_artifact {
                    attach_policy_summary(&mut output, summary, artifact_meta);
                }

                if self.config.transparency_writer
                    && self.config.transparency_scope.includes_artifacts()
                {
                    if let Some(artifacts_value) = output.get_mut("artifacts") {
                        self.decorate_artifacts(step.run_id, step.step_id, artifacts_value)
                            .await?;
                    }
                }

                if let Some(token) = capability_token.as_ref() {
                    if let Some(map) = output.as_object_mut() {
                        map.insert(
                            "capability_token".into(),
                            to_value(token).expect("capability token serialisable"),
                        );
                        map.insert(
                            "capability_token_id".into(),
                            Value::String(token.token_id().to_string()),
                        );
                    }
                }

                if let Some(usage) = usage {
                    budget_ctx.tokens_actual = usage.tokens;
                    budget_ctx.cost_actual = usage.cost;
                    if let Some(tokens) = usage.tokens {
                        step_span.record("fleetforge_tokens_total", tokens);
                    }
                    if let Some(cost) = usage.cost {
                        step_span.record("fleetforge_cost_usd", cost);
                    }
                }
                if let Some(p) = result_provider {
                    provider = Some(p);
                }
                if let Some(v) = result_provider_version {
                    provider_version = Some(v);
                }
                step_span.record("fleetforge_status", "succeeded");

                let mut checkpoint_value = checkpoint.clone();
                if checkpoint_value.is_none() && step.spec.execution.checkpoint {
                    checkpoint_value = Some(output.clone());
                }
                let artifacts_value = output.get("artifacts").cloned();
                let output_snapshot_value = output.clone();

                self.finish_step(
                    &step,
                    &spec,
                    StepStatus::Succeeded,
                    step.attempt + 1,
                    Some(output.clone()),
                    None,
                    Some(latency),
                    &budget_ctx,
                    input_snapshot.clone(),
                    Some(output_snapshot_value),
                    provider.as_deref(),
                    provider_version.as_deref(),
                    Some(&ctx.trace),
                    Some(&run_trace),
                    tool_events,
                    guardrail_events.clone(),
                    checkpoint_value,
                    artifacts_value,
                    ctx.slo.as_ref(),
                )
                .await
            }
            Err(err) => {
                step_span.record("fleetforge_status", "failed");
                let error_message = err.to_string();
                step_span.record("error_message", error_message.as_str());
                let mut error_payload = json!({
                    "message": &error_message,
                    "kind": "execution_error"
                });
                if let Some((artifact_meta, summary)) = policy_artifact {
                    attach_policy_summary(&mut error_payload, summary, artifact_meta);
                }
                if let Some(token) = ctx.capability_token.as_ref() {
                    if let Some(map) = error_payload.as_object_mut() {
                        map.insert(
                            "capability_token".into(),
                            to_value(token).expect("capability token serialisable"),
                        );
                        map.insert(
                            "capability_token_id".into(),
                            Value::String(token.token_id().to_string()),
                        );
                    }
                }
                insert_trace(&mut error_payload, &ctx.trace);

                let attempt_number = step.attempt + 1;
                if attempt_number < step.max_attempts {
                    self.release_inflight(budget_ctx.tokens_reserved, budget_ctx.cost_reserved)
                        .await;
                    self.schedule_retry(&step, error_payload, &budget_ctx, &ctx.trace, &run_trace)
                        .await?;
                    return Ok(());
                }

                self.finish_step(
                    &step,
                    &spec,
                    StepStatus::Failed,
                    attempt_number,
                    None,
                    Some(error_payload),
                    Some(latency),
                    &budget_ctx,
                    input_snapshot,
                    None,
                    provider.as_deref(),
                    provider_version.as_deref(),
                    Some(&ctx.trace),
                    Some(&run_trace),
                    Vec::new(),
                    guardrail_events.clone(),
                    None,
                    None,
                    ctx.slo.as_ref(),
                )
                .await?;

                self.schedule_compensation(&step, &run_trace).await
            }
        }
    }

    fn rank_steps(&self, steps: &mut [QueuedStep]) {
        if steps.len() <= 1 {
            return;
        }

        let now = Utc::now();
        let mut scores: HashMap<StepId, SloScore> = HashMap::with_capacity(steps.len());
        let mut original_positions: HashMap<StepId, usize> = HashMap::with_capacity(steps.len());

        for (idx, step) in steps.iter().enumerate() {
            let contract = extract_slo_contract(&step.spec);
            let score = compute_slo_score(step, contract.as_ref(), now);
            if score.has_contract {
                self.metrics.record_slo_slack(Some(score.slack_ms));
            }
            scores.insert(step.step_id, score);
            original_positions.insert(step.step_id, idx);
        }

        steps.sort_by(|a, b| {
            let sa = scores.get(&a.step_id).expect("slo score missing");
            let sb = scores.get(&b.step_id).expect("slo score missing");
            sa.compare(sb).then_with(|| a.created_at.cmp(&b.created_at))
        });

        for (idx, step) in steps.iter().enumerate() {
            if let Some(score) = scores.get(&step.step_id) {
                if score.has_contract && score.slack_ms < 0 {
                    if let Some(original) = original_positions.get(&step.step_id) {
                        if idx < *original {
                            self.metrics.record_slo_preemption();
                        }
                    }
                }
            }
        }
    }

    async fn finish_step(
        &self,
        step: &QueuedStep,
        spec_snapshot: &StepSpec,
        status: StepStatus,
        attempt: i32,
        output: Option<Value>,
        error: Option<Value>,
        latency: Option<Duration>,
        budget: &BudgetCtx,
        input_snapshot: Option<Value>,
        output_snapshot: Option<Value>,
        provider: Option<&str>,
        provider_version: Option<&str>,
        step_trace: Option<&TraceContext>,
        run_trace: Option<&TraceContext>,
        tool_events: Vec<ToolEventRecord>,
        guardrail_events: Vec<PolicyEvaluationRecord>,
        checkpoint: Option<Value>,
        artifacts: Option<Value>,
        slo: Option<&StepSlo>,
    ) -> Result<()> {
        let mut tx = self.storage.pool().begin().await?;
        let run_uuid = Uuid::from(step.run_id);
        let step_uuid = Uuid::from(step.step_id);

        let run_trace_ctx = if let Some(trace) = run_trace {
            trace.clone()
        } else {
            self.build_run_trace_for_step(step).await?
        };
        let step_trace_ctx = if let Some(trace) = step_trace {
            trace.clone()
        } else {
            self.build_step_trace_for_step(step, &run_trace_ctx)
        };

        let artifacts_for_event = artifacts.clone();

        self.storage
            .update_step_status_tx(
                &mut tx,
                run_uuid,
                step_uuid,
                models::StepStatus::from(status),
                Some(attempt),
                output.clone(),
                error.clone(),
                input_snapshot.clone(),
                output_snapshot.clone(),
                checkpoint.clone(),
                provider,
                provider_version,
            )
            .await?;

        let actual_tokens = budget
            .tokens_actual
            .unwrap_or(budget.tokens_reserved)
            .max(0);
        let actual_cost = budget.cost_actual.unwrap_or(budget.cost_reserved).max(0.0);

        self.storage
            .update_step_usage_tx(
                &mut tx,
                run_uuid,
                step_uuid,
                Some(actual_tokens),
                Some(actual_cost),
            )
            .await?;

        let tokens_delta = actual_tokens - budget.tokens_reserved;
        let cost_delta = actual_cost - budget.cost_reserved;
        let mut tokens_total = if budget.tracked {
            budget.tokens_used
        } else {
            actual_tokens
        };
        let mut cost_total = if budget.tracked {
            budget.cost_used
        } else {
            actual_cost
        };

        if budget.tracked {
            self.storage
                .reconcile_budget_usage_tx(&mut tx, run_uuid, tokens_delta, cost_delta)
                .await?;
            tokens_total += tokens_delta;
            cost_total += cost_delta;
        }

        let latency_ms = latency.map(|d| d.as_secs_f64() * 1000.0);
        let spec_snapshot_json = serde_json::to_value(spec_snapshot)
            .context("failed to serialise step spec snapshot")?;
        let budget_snapshot_json = guardrail_budget_view(budget, slo);
        let guardrail_events_json = if guardrail_events.is_empty() {
            None
        } else {
            Some(serde_json::to_value(&guardrail_events)?)
        };
        let tool_events_json = if tool_events.is_empty() {
            None
        } else {
            Some(tool_events_to_value(&tool_events))
        };

        self.storage
            .insert_step_attempt_tx(
                &mut tx,
                models::NewStepAttempt {
                    run_id: run_uuid,
                    step_id: step_uuid,
                    attempt,
                    status: models::StepStatus::from(status),
                    latency_ms,
                    spec_snapshot: spec_snapshot_json.clone(),
                    input_snapshot: input_snapshot.clone(),
                    output_snapshot: output_snapshot.clone(),
                    error_snapshot: error.clone(),
                    checkpoint: checkpoint.clone(),
                    budget_snapshot: Some(budget_snapshot_json.clone()),
                    guardrail_events: guardrail_events_json.clone(),
                    tool_events: tool_events_json.clone(),
                    artifacts: artifacts.clone(),
                },
            )
            .await?;

        let mut newly_ready = Vec::new();
        let mut newly_skipped = Vec::new();

        match status {
            StepStatus::Succeeded => {
                newly_ready = self
                    .storage
                    .unlock_successors_tx(&mut tx, run_uuid, step_uuid)
                    .await?;
            }
            StepStatus::Failed => {
                newly_skipped = self
                    .storage
                    .skip_downstream_steps_tx(&mut tx, run_uuid, step_uuid)
                    .await?;
            }
            _ => {}
        }

        let (previous_run_status, current_run_status) = self
            .storage
            .refresh_run_status_tx(&mut tx, run_uuid)
            .await?;

        let mut payload = json!({
            "run_id": step.run_id,
            "step_id": step.step_id,
            "status": status.as_str(),
            "output": output,
            "error": error,
            "attempt": attempt,
            "priority": step.priority,
            "provider": provider,
            "provider_version": provider_version,
            "snapshots": {
                "input": input_snapshot,
                "output": output_snapshot,
                "checkpoint": checkpoint,
            },
            "budget": {
                "tracked": budget.tracked,
                "reserved": {
                    "tokens": budget.tokens_reserved,
                    "cost": budget.cost_reserved,
                },
                "actual": {
                    "tokens": actual_tokens,
                    "cost": actual_cost,
                },
                "deltas": {
                    "tokens": tokens_delta,
                    "cost": cost_delta,
                },
                "totals": {
                    "tokens_used": tokens_total,
                    "cost_used": cost_total,
                },
                "limits": {
                    "tokens": budget.token_limit,
                    "cost": budget.cost_limit,
                    "time_secs": budget.time_limit.map(|d| d.as_secs()),
                },
                "snapshot": budget_snapshot_json.clone(),
            },
            "spec_snapshot": spec_snapshot_json.clone(),
            "guardrail_events": guardrail_events_json.clone(),
            "tool_events": tool_events_json.clone(),
            "latency_ms": latency_ms,
            "artifacts": artifacts_for_event,
        });
        insert_trace(&mut payload, &step_trace_ctx);

        self.storage
            .enqueue_outbox_event(
                &mut tx,
                NewOutboxEvent {
                    run_id: run_uuid,
                    step_id: Some(step_uuid),
                    kind: format!("step_{}", status.as_str()),
                    payload,
                },
            )
            .await?;

        for event in tool_events {
            let mut payload = event.payload;
            insert_trace(&mut payload, &event.trace);
            self.storage
                .enqueue_outbox_event(
                    &mut tx,
                    NewOutboxEvent {
                        run_id: run_uuid,
                        step_id: Some(step_uuid),
                        kind: event.kind,
                        payload,
                    },
                )
                .await?;
        }

        self.storage
            .insert_cost_ledger_tx(
                &mut tx,
                run_uuid,
                step_uuid,
                attempt,
                status.as_str(),
                budget.tokens_reserved,
                budget.cost_reserved,
                Some(actual_tokens),
                Some(actual_cost),
            )
            .await?;

        for queued_step_id in &newly_ready {
            let mut payload = json!({
                "run_id": step.run_id,
                "step_id": StepId::from(*queued_step_id),
                "status": StepStatus::Queued.as_str(),
                "reason": "dependencies_satisfied",
            });
            insert_trace(&mut payload, &run_trace_ctx);

            self.storage
                .enqueue_outbox_event(
                    &mut tx,
                    NewOutboxEvent {
                        run_id: run_uuid,
                        step_id: Some(*queued_step_id),
                        kind: "step_queued".to_string(),
                        payload,
                    },
                )
                .await?;
        }

        for skipped_step_id in &newly_skipped {
            let mut payload = json!({
                "run_id": step.run_id,
                "step_id": StepId::from(*skipped_step_id),
                "status": StepStatus::Skipped.as_str(),
                "reason": "dependency_failed",
            });
            insert_trace(&mut payload, &run_trace_ctx);

            self.storage
                .enqueue_outbox_event(
                    &mut tx,
                    NewOutboxEvent {
                        run_id: run_uuid,
                        step_id: Some(*skipped_step_id),
                        kind: "step_skipped".to_string(),
                        payload,
                    },
                )
                .await?;
        }

        if previous_run_status != current_run_status {
            let mut payload = json!({
                "run_id": step.run_id,
                "previous_status": previous_run_status.as_str(),
                "status": current_run_status.as_str(),
            });
            insert_trace(&mut payload, &run_trace_ctx);

            self.storage
                .enqueue_outbox_event(
                    &mut tx,
                    NewOutboxEvent {
                        run_id: run_uuid,
                        step_id: None,
                        kind: format!("run_{}", current_run_status.as_str()),
                        payload,
                    },
                )
                .await?;

            if matches!(
                current_run_status,
                models::RunStatus::Succeeded | models::RunStatus::Failed
            ) {
                if let Err(err) = self.emit_run_bom(step.run_id, current_run_status).await {
                    warn!(
                        run = %step.run_id,
                        error = %err,
                        "failed to generate BOM for run"
                    );
                }
            }
        }

        if actual_tokens > 0 || actual_cost > 0.0 {
            self.storage
                .record_billing_usage_tx(&mut tx, run_uuid, actual_tokens, actual_cost)
                .await?;
        }

        tx.commit().await?;

        let tenant_uuid = match self.storage.fetch_run_tenant(run_uuid).await {
            Ok(value) => value.and_then(|s| Uuid::parse_str(&s).ok()),
            Err(err) => {
                warn!(error = %err, run = %step.run_id, "failed to fetch tenant for audit");
                None
            }
        };

        let audit_resource = json!({
            "run_id": step.run_id.to_string(),
            "step_id": step.step_id.to_string(),
            "status": status.as_str(),
            "attempt": attempt,
            "worker": self.config.worker_id,
        });

        if let Err(err) = self
            .audit
            .append(
                &self.config.worker_id,
                tenant_uuid,
                &format!("step.{}", status.as_str()),
                &audit_resource,
            )
            .await
        {
            warn!(error = %err, run = %step.run_id, step = %step.step_id, "failed to append audit log for step");
        }

        if let Some(latency) = latency {
            self.metrics.record_latency(latency);
        }
        self.metrics
            .record_budget_consumed(actual_tokens, actual_cost);

        if budget.tracked {
            let remaining_tokens = budget.token_limit.map(|limit| {
                let diff = limit - tokens_total;
                if diff < 0 {
                    0
                } else {
                    diff
                }
            });
            let remaining_cost = budget.cost_limit.map(|limit| (limit - cost_total).max(0.0));
            self.metrics
                .record_budget_remaining(remaining_tokens, remaining_cost);
        }

        self.release_inflight(budget.tokens_reserved, budget.cost_reserved)
            .await;

        Ok(())
    }

    fn tool_events_to_value(events: &[ToolEventRecord]) -> Value {
        Value::Array(
            events
                .iter()
                .map(|event| {
                    let mut object = Map::new();
                    object.insert("kind".to_string(), Value::String(event.kind.clone()));
                    object.insert("payload".to_string(), event.payload.clone());
                    object.insert("trace".to_string(), event.trace.to_json());
                    Value::Object(object)
                })
                .collect(),
        )
    }

    async fn build_run_trace_for_step(&self, step: &QueuedStep) -> Result<TraceContext> {
        let run_uuid = Uuid::from(step.run_id);
        let run_span_id = span_id_from_uuid(&run_uuid);
        let trace_headers = self.trace_headers_for_run(step.run_id).await?;
        let run_scope = RunScope::new(step.run_id.to_string());

        let run_trace = if let Some(headers) = trace_headers {
            match TraceContext::from_w3c(
                run_scope.clone(),
                &headers.traceparent,
                headers.tracestate.as_deref(),
                headers.baggage.as_deref(),
            ) {
                Ok(ctx) => ctx.with_span_id(run_span_id),
                Err(err) => {
                    warn!(
                        run = %step.run_id,
                        error = %err,
                        "failed to parse trace headers; using fallback trace id"
                    );
                    TraceContext::new(run_scope.clone())
                        .with_trace_id(trace_id_from_uuid(&run_uuid))
                        .with_span_id(run_span_id)
                }
            }
        } else {
            TraceContext::new(run_scope.clone())
                .with_trace_id(trace_id_from_uuid(&run_uuid))
                .with_span_id(run_span_id)
        };

        Ok(run_trace)
    }

    fn build_step_trace_for_step(
        &self,
        step: &QueuedStep,
        run_trace: &TraceContext,
    ) -> TraceContext {
        let step_index = if step.idx >= 0 { step.idx as u32 } else { 0 };
        let step_attempt = if step.attempt >= 0 {
            (step.attempt + 1) as u32
        } else {
            0
        };

        let step_scope = StepScope::new(step.step_id.to_string())
            .with_index(step_index)
            .with_attempt(step_attempt)
            .with_kind(step.spec.r#type.clone())
            .with_scheduler(self.config.worker_id.clone());

        run_trace.child_step(step_scope)
    }

    async fn try_pause_for_breakpoint(&self, step: &QueuedStep) -> Result<bool> {
        let run_uuid = Uuid::from(step.run_id);
        let step_uuid = Uuid::from(step.step_id);

        let run = match self.storage.fetch_run(run_uuid).await? {
            Some(run) => run,
            None => return Ok(false),
        };

        let mut ctx_map = match run.input_ctx.clone() {
            Value::Object(map) => map,
            _ => return Ok(false),
        };

        let debug_value = match ctx_map.get_mut("debug") {
            Some(Value::Object(map)) => map,
            _ => return Ok(false),
        };

        let mut breakpoints = load_debug_breakpoints(debug_value);
        if breakpoints.is_empty() {
            return Ok(false);
        }

        let index = match breakpoints.iter().position(|bp| bp.step_id == step_uuid) {
            Some(idx) => idx,
            None => return Ok(false),
        };

        let record = breakpoints.remove(index);
        store_debug_breakpoints(debug_value, &breakpoints);
        set_active_pause(debug_value, &record);

        let mut tx = self
            .storage
            .pool()
            .begin()
            .await
            .context("failed to begin pause transaction")?;

        self.storage
            .update_run_input_ctx_tx(&mut tx, run_uuid, Value::Object(ctx_map.clone()))
            .await
            .context("failed to persist pause context")?;

        self.storage
            .set_run_status_tx(&mut tx, run_uuid, models::RunStatus::PausedAtStep)
            .await
            .context("failed to set run status to paused")?;

        self.storage
            .revert_step_to_queue_tx(&mut tx, run_uuid, step_uuid)
            .await
            .context("failed to revert step to queue after pause")?;

        self.storage
            .enqueue_outbox_event(
                &mut tx,
                NewOutboxEvent {
                    run_id: run_uuid,
                    step_id: Some(step_uuid),
                    kind: "run_paused".to_string(),
                    payload: json!({
                        "run_id": run_uuid.to_string(),
                        "step_id": step_uuid.to_string(),
                        "spec": record.spec,
                        "reason": "breakpoint",
                    }),
                },
            )
            .await
            .context("failed to enqueue pause event")?;

        tx.commit()
            .await
            .context("failed to commit pause transaction")?;

        let tenant_uuid = match self.storage.fetch_run_tenant(run_uuid).await {
            Ok(Some(tenant)) => Uuid::parse_str(&tenant).ok(),
            _ => None,
        };

        if let Err(err) = self
            .audit
            .append(
                &self.config.worker_id,
                tenant_uuid,
                "debug.breakpoints.pause",
                &json!({
                    "run_id": run_uuid.to_string(),
                    "step_id": step_uuid.to_string(),
                    "reason": "breakpoint",
                }),
            )
            .await
        {
            warn!(error = %err, run = %step.run_id, step = %step.step_id, "failed to append audit log for breakpoint pause");
        }

        info!(run = %step.run_id, step = %step.step_id, "paused run at breakpoint");
        Ok(true)
    }

    async fn reserve_budget(
        &self,
        run_id: RunId,
        request: &BudgetRequest,
    ) -> Result<BudgetDecision> {
        if let Some(tenant) = self.storage.fetch_run_tenant(Uuid::from(run_id)).await? {
            if let Err(err) = self
                .storage
                .enforce_tenant_quota(&tenant, request.tokens, request.cost)
                .await
            {
                return Ok(BudgetDecision::Denied {
                    reason: err.to_string(),
                });
            }
        }

        match self
            .storage
            .try_consume_budget(Uuid::from(run_id), request.tokens, request.cost)
            .await?
        {
            BudgetResult::Granted(budget) => {
                let mut ctx = BudgetCtx::default();
                ctx.token_limit = budget.tokens_limit;
                ctx.tokens_used = budget.tokens_used;
                ctx.tokens_reserved = request.tokens;
                ctx.cost_limit = budget.cost_limit;
                ctx.cost_used = budget.cost_used;
                ctx.cost_reserved = request.cost;
                ctx.tracked = true;
                Ok(BudgetDecision::Granted(ctx))
            }
            BudgetResult::Denied => Ok(BudgetDecision::Denied {
                reason: format!("budget exhausted for run {}", run_id),
            }),
            BudgetResult::NoBudgetConfigured => {
                let mut ctx = BudgetCtx::default();
                ctx.tokens_reserved = request.tokens;
                ctx.cost_reserved = request.cost;
                Ok(BudgetDecision::Unlimited(ctx))
            }
        }
    }

    async fn should_backpressure(&self, request: &BudgetRequest) -> bool {
        if (request.tokens <= 0 && request.cost <= 0.0)
            || (self.config.max_inflight_tokens.is_none()
                && self.config.max_inflight_cost.is_none())
        {
            return false;
        }

        let inflight = self.inflight.lock().await;
        if let Some(limit) = self.config.max_inflight_tokens {
            if request.tokens > 0 && inflight.tokens + request.tokens > limit {
                return true;
            }
        }
        if let Some(limit) = self.config.max_inflight_cost {
            if request.cost > 0.0 && inflight.cost + request.cost > limit {
                return true;
            }
        }
        false
    }

    async fn defer_for_backpressure(&self, step: &QueuedStep) -> Result<()> {
        let delay = chrono::Duration::from_std(self.config.backpressure_delay)
            .unwrap_or_else(|_| ChronoDuration::milliseconds(50));
        let not_before = Utc::now() + delay;
        let run_trace = self.build_run_trace_for_step(step).await?;
        let mut payload = json!({
            "run_id": step.run_id,
            "step_id": step.step_id,
            "status": "queued",
            "reason": "backpressure",
            "not_before": not_before.to_rfc3339(),
        });
        insert_trace(&mut payload, &run_trace);
        self.storage
            .defer_step_to_queue(
                Uuid::from(step.run_id),
                Uuid::from(step.step_id),
                not_before,
                payload,
            )
            .await
    }

    async fn add_inflight(&self, tokens: i64, cost: f64) {
        if tokens <= 0 && cost <= 0.0 {
            return;
        }
        let mut inflight = self.inflight.lock().await;
        inflight.add(tokens, cost);
    }

    async fn release_inflight(&self, tokens: i64, cost: f64) {
        if tokens <= 0 && cost <= 0.0 {
            return;
        }
        let mut inflight = self.inflight.lock().await;
        inflight.subtract(tokens, cost);
    }

    fn compute_retry_delay(&self, step: &QueuedStep) -> ChronoDuration {
        let base = step.retry_backoff_ms.max(0);
        if base == 0 {
            return ChronoDuration::from_std(self.config.active_backoff)
                .unwrap_or_else(|_| ChronoDuration::milliseconds(10));
        }
        let exponent = step.attempt.max(0);
        let factor = step.retry_backoff_factor.max(1.0);
        let scale = if exponent > 0 {
            factor.powi(exponent)
        } else {
            1.0
        };
        let delay = (base as f64) * scale;
        let clamped = delay.min(i64::MAX as f64);
        ChronoDuration::milliseconds(clamped as i64)
    }

    fn issue_capability_token(
        &self,
        run_id: RunId,
        step_id: StepId,
        attempt: i32,
        spec: &StepSpec,
        budget: &BudgetCtx,
    ) -> Option<CapabilityToken> {
        let signer = match capability_signer() {
            Ok(signer) => signer,
            Err(err) => {
                warn!(
                    run = %run_id,
                    step = %step_id,
                    error = %err,
                    "failed to load capability signer; skipping capability token mint"
                );
                return None;
            }
        };

        let subject = CapabilityTokenSubject {
            run_id: Uuid::from(run_id),
            step_id: Some(Uuid::from(step_id)),
            attempt: Some(attempt),
        };
        let scope = capability::build_capability_scope(spec, budget);
        let ttl = spec
            .execution
            .deadline_ms
            .map(|ms| {
                let clamped = ms.min(i64::MAX as u64) as i64;
                ChronoDuration::milliseconds(clamped)
            })
            .filter(|duration| *duration > ChronoDuration::zero())
            .unwrap_or_else(|| ChronoDuration::minutes(15));

        match mint_capability_token(
            subject,
            scope,
            Some(vec!["runtime-tool-adapter".into()]),
            ttl,
            signer.as_ref(),
        ) {
            Ok(token) => Some(token),
            Err(err) => {
                warn!(
                    run = %run_id,
                    step = %step_id,
                    error = %err,
                    "failed to mint capability token"
                );
                None
            }
        }
    }

    fn validate_capability_guard(
        &self,
        token: Option<&CapabilityToken>,
        spec: &StepSpec,
        budget: &BudgetCtx,
        run_id: RunId,
        step_id: StepId,
    ) -> Result<Option<CapabilityViolation>> {
        match capability::validate_capability_token(token, spec, budget, Utc::now()) {
            Ok(()) => Ok(None),
            Err(err) => Ok(Some(capability_violation_record(
                run_id,
                step_id,
                err.reasons,
                err.detail,
                err.token_id,
            ))),
        }
    }

    async fn schedule_retry(
        &self,
        step: &QueuedStep,
        error_payload: Value,
        budget: &BudgetCtx,
        step_trace: &TraceContext,
        run_trace: &TraceContext,
    ) -> Result<()> {
        let delay = self.compute_retry_delay(step);
        let not_before = Utc::now() + delay;
        let run_uuid = Uuid::from(step.run_id);
        let step_uuid = Uuid::from(step.step_id);
        let attempt = step.attempt + 1;

        let mut tx = self.storage.pool().begin().await?;
        self.storage
            .schedule_retry_tx(
                &mut tx,
                run_uuid,
                step_uuid,
                attempt,
                not_before,
                error_payload,
            )
            .await?;

        if budget.tracked {
            self.storage
                .reconcile_budget_usage_tx(
                    &mut tx,
                    run_uuid,
                    -budget.tokens_reserved,
                    -budget.cost_reserved,
                )
                .await?;
        }

        let mut payload = json!({
            "run_id": step.run_id,
            "step_id": step.step_id,
            "status": "queued",
            "reason": "retry",
            "attempt": attempt,
            "not_before": not_before.to_rfc3339(),
        });
        insert_trace(&mut payload, run_trace);
        insert_trace(&mut payload, step_trace);

        self.storage
            .enqueue_outbox_event(
                &mut tx,
                NewOutboxEvent {
                    run_id: run_uuid,
                    step_id: Some(step_uuid),
                    kind: "step_retry_scheduled".to_string(),
                    payload,
                },
            )
            .await?;

        self.storage
            .insert_cost_ledger_tx(
                &mut tx,
                run_uuid,
                step_uuid,
                attempt,
                "retry_scheduled",
                budget.tokens_reserved,
                budget.cost_reserved,
                budget.tokens_actual,
                budget.cost_actual,
            )
            .await?;

        tx.commit().await?;

        let tenant_uuid = match self.storage.fetch_run_tenant(run_uuid).await {
            Ok(value) => value.and_then(|s| Uuid::parse_str(&s).ok()),
            Err(err) => {
                warn!(
                    error = %err,
                    run = %step.run_id,
                    step = %step.step_id,
                    "failed to fetch tenant for retry audit"
                );
                None
            }
        };

        let audit_resource = json!({
            "run_id": step.run_id.to_string(),
            "step_id": step.step_id.to_string(),
            "attempt": attempt,
            "not_before": not_before.to_rfc3339(),
        });

        if let Err(err) = self
            .audit
            .append(
                &self.config.worker_id,
                tenant_uuid,
                "step.retry",
                &audit_resource,
            )
            .await
        {
            warn!(
                error = %err,
                run = %step.run_id,
                step = %step.step_id,
                "failed to append audit log for retry"
            );
        }

        self.metrics.record_retry();
        Ok(())
    }

    async fn schedule_compensation(
        &self,
        step: &QueuedStep,
        run_trace: &TraceContext,
    ) -> Result<()> {
        let Some(target) = step.compensation_step else {
            return Ok(());
        };
        if step.compensation_scheduled {
            return Ok(());
        }

        let run_uuid = Uuid::from(step.run_id);
        let origin_uuid = Uuid::from(step.step_id);
        let target_uuid = Uuid::from(target);

        let mut tx = self.storage.pool().begin().await?;
        self.storage
            .trigger_compensation_tx(&mut tx, run_uuid, origin_uuid, target_uuid)
            .await?;

        let mut payload = json!({
            "run_id": step.run_id,
            "step_id": target,
            "reason": "compensation",
        });
        insert_trace(&mut payload, run_trace);

        self.storage
            .enqueue_outbox_event(
                &mut tx,
                NewOutboxEvent {
                    run_id: run_uuid,
                    step_id: Some(target_uuid),
                    kind: "step_compensation_queued".to_string(),
                    payload,
                },
            )
            .await?;

        self.storage
            .insert_cost_ledger_tx(
                &mut tx,
                run_uuid,
                origin_uuid,
                step.attempt + 1,
                "compensation_scheduled",
                0,
                0.0,
                None,
                None,
            )
            .await?;

        tx.commit().await?;

        let tenant_uuid = match self.storage.fetch_run_tenant(run_uuid).await {
            Ok(value) => value.and_then(|s| Uuid::parse_str(&s).ok()),
            Err(err) => {
                warn!(
                    error = %err,
                    run = %step.run_id,
                    step = %step.step_id,
                    "failed to fetch tenant for compensation audit"
                );
                None
            }
        };

        let audit_resource = json!({
            "run_id": step.run_id.to_string(),
            "step_id": step.step_id.to_string(),
            "compensation_step": target.to_string(),
        });

        if let Err(err) = self
            .audit
            .append(
                &self.config.worker_id,
                tenant_uuid,
                "step.compensation",
                &audit_resource,
            )
            .await
        {
            warn!(
                error = %err,
                run = %step.run_id,
                step = %step.step_id,
                "failed to append audit log for compensation step"
            );
        }

        Ok(())
    }
}

fn attach_policy_summary(target: &mut Value, mut summary: Value, artifact_meta: Value) {
    let Some(obj) = target.as_object_mut() else {
        return;
    };

    if let Value::Object(ref mut summary_obj) = summary {
        summary_obj.insert("artifact".to_string(), artifact_meta.clone());
    }

    obj.insert("policy_decisions".to_string(), summary);
    let entry = obj
        .entry("artifacts")
        .or_insert_with(|| Value::Object(Map::new()));
    if let Value::Object(ref mut map) = entry {
        map.insert("policy_decisions".to_string(), artifact_meta);
    } else {
        let mut map = Map::new();
        map.insert("policy_decisions".to_string(), artifact_meta);
        *entry = Value::Object(map);
    }
}

fn insert_trace(payload: &mut Value, trace: &TraceContext) {
    if let Value::Object(ref mut map) = payload {
        map.insert("trace".to_string(), trace.to_json());
    }
}

impl Scheduler {
    async fn decorate_artifacts(
        &self,
        run_id: RunId,
        step_id: StepId,
        artifacts_value: &mut Value,
    ) -> Result<()> {
        if !self.config.transparency_writer || !self.config.transparency_scope.includes_artifacts()
        {
            return Ok(());
        }

        let Value::Object(map) = artifacts_value else {
            return Ok(());
        };

        for (key, entry) in map.iter_mut() {
            self.enqueue_artifact_job(run_id, step_id, key, entry)
                .await?;
        }

        Ok(())
    }

    async fn enqueue_artifact_job(
        &self,
        run_id: RunId,
        step_id: StepId,
        artifact_key: &str,
        entry: &mut Value,
    ) -> Result<()> {
        let Value::Object(obj) = entry else {
            return Ok(());
        };
        let Some(sha256) = obj
            .get("sha256")
            .and_then(Value::as_str)
            .map(|s| s.to_string())
        else {
            return Ok(());
        };
        let metadata = obj.get("metadata").cloned().unwrap_or(Value::Null);
        let metadata_obj = match metadata.as_object() {
            Some(map) => map,
            None => return Ok(()),
        };
        let attestation_ids_value = metadata_obj
            .get("attestation_ids")
            .and_then(Value::as_array)
            .cloned()
            .unwrap_or_default();
        if attestation_ids_value.is_empty() {
            return Ok(());
        }
        let mut attestation_ids: Vec<Uuid> = Vec::new();
        for value in attestation_ids_value {
            if let Some(id) = value.as_str() {
                if let Ok(uuid) = Uuid::parse_str(id) {
                    attestation_ids.push(uuid);
                }
            }
        }
        if attestation_ids.is_empty() {
            return Ok(());
        }

        let payload = models::TransparencyJobPayload::ArtifactEntry(models::ArtifactEntryPayload {
            run_id: Some(Uuid::from(run_id)),
            step_id: Some(Uuid::from(step_id)),
            artifact_key: artifact_key.to_string(),
            artifact_sha256: sha256.clone(),
            attestation_ids: attestation_ids.clone(),
            metadata: metadata.clone(),
        });

        match self.storage.enqueue_transparency_job(payload).await {
            Ok(job_id) => {
                obj.insert(
                    "transparency".to_string(),
                    json!({
                        "status": "queued",
                        "job_id": job_id,
                        "artifact_sha256": sha256,
                        "attestation_ids": attestation_ids.iter().map(|id| id.to_string()).collect::<Vec<_>>(),
                    }),
                );
            }
            Err(err) => {
                warn!(
                    run = %run_id,
                    step = %step_id,
                    artifact = artifact_key,
                    error = %err,
                    "failed to enqueue artifact transparency job"
                );
            }
        }

        Ok(())
    }
}

fn policy_pack_name(policy: &Value) -> String {
    match policy {
        Value::Object(map) => map
            .get("pack")
            .and_then(Value::as_str)
            .map(|value| value.to_string())
            .unwrap_or_else(|| "baseline".to_string()),
        _ => "baseline".to_string(),
    }
}

fn policy_outcome_to_record(
    outcome: &PolicyOutcome,
    run_id: RunId,
    step_id: StepId,
    boundary: TrustBoundary,
    policy_id: &str,
) -> PolicyEvaluationRecord {
    let run_uuid = Uuid::from(run_id);
    let step_uuid = Uuid::from(step_id);
    let reasons = outcome
        .decision
        .reason
        .as_ref()
        .map(|reason| vec![reason.clone()])
        .unwrap_or_default();
    let decision_trace = PolicyDecisionTrace {
        policy: policy_id.to_string(),
        effect: outcome.decision.effect,
        reason: outcome.decision.reason.clone(),
        patches: outcome.decision.patches.clone(),
    };

    PolicyEvaluationRecord {
        boundary,
        run_id: Some(run_uuid),
        step_id: Some(step_uuid),
        effect: outcome.decision.effect,
        reasons,
        decisions: vec![decision_trace],
        patches: outcome.decision.patches.clone(),
        payload: None,
        timestamp: outcome.attestation.issued_at,
        origins: Vec::new(),
        preview: None,
        attestation_id: Some(outcome.attestation.id),
        attestation: serde_json::to_value(&outcome.attestation).ok(),
        trust_decision: serde_json::to_value(&outcome.trust).ok(),
    }
}

fn dominant_policy_effect(events: &[PolicyEvaluationRecord]) -> DecisionEffect {
    events
        .iter()
        .map(|event| event.effect)
        .max_by_key(effect_priority)
        .unwrap_or(DecisionEffect::Allow)
}

fn effect_priority(effect: DecisionEffect) -> u8 {
    match effect {
        DecisionEffect::Deny => 4,
        DecisionEffect::Redact => 2,
        DecisionEffect::Allow => 0,
    }
}

fn decision_effect_label(effect: DecisionEffect) -> String {
    match effect {
        DecisionEffect::Allow => "allow",
        DecisionEffect::Deny => "deny",
        DecisionEffect::Redact => "redact",
    }
    .to_string()
}

fn trust_boundary_label(boundary: &TrustBoundary) -> String {
    serde_json::to_value(boundary)
        .ok()
        .and_then(|value| value.as_str().map(|s| s.to_string()))
        .unwrap_or_else(|| format!("{boundary:?}"))
}

fn resolve_manifest_profile(
    metadata: Option<&Value>,
    fallback: ManifestProfile,
) -> ManifestProfile {
    metadata_manifest_profile(metadata)
        .or_else(env_manifest_profile)
        .unwrap_or(fallback)
}

fn metadata_manifest_profile(metadata: Option<&Value>) -> Option<ManifestProfile> {
    metadata
        .and_then(|value| match value {
            Value::Object(map) => map.get("c2pa_profile").and_then(Value::as_str),
            _ => None,
        })
        .and_then(parse_manifest_profile_label)
}

fn env_manifest_profile() -> Option<ManifestProfile> {
    env::var(C2PA_PROFILE_ENV)
        .ok()
        .and_then(|value| parse_manifest_profile_label(value.trim()))
}

fn parse_manifest_profile_label(label: &str) -> Option<ManifestProfile> {
    match label.to_ascii_lowercase().as_str() {
        "basic" => Some(ManifestProfile::Basic),
        "policy" | "policy-evidence" => Some(ManifestProfile::PolicyEvidence),
        "full" => Some(ManifestProfile::Full),
        _ => None,
    }
}

fn summarize_policy_events(events: &[PolicyEvaluationRecord]) -> String {
    let mut snippets: Vec<String> = Vec::new();
    for event in events {
        if !event.reasons.is_empty() {
            snippets.extend(event.reasons.iter().cloned());
        } else if let Some(preview) = &event.preview {
            snippets.push(preview.clone());
        }
    }

    if snippets.is_empty() {
        decision_effect_label(dominant_policy_effect(events))
    } else {
        if snippets.len() > 3 {
            snippets.truncate(3);
        }
        snippets.join("; ")
    }
}

async fn persist_policy_decisions(
    artifacts: Arc<dyn ArtifactStore>,
    run_id: RunId,
    step_id: StepId,
    events: &[PolicyEvaluationRecord],
) -> Result<(Value, Value)> {
    let events_value = serde_json::to_value(events)?;
    let payload_vec = serde_json::to_vec(&events_value)?;
    let attestation_ids: Vec<Uuid> = events
        .iter()
        .filter_map(|event| event.attestation_id)
        .collect();
    let boundaries: Vec<TrustBoundary> = events.iter().map(|e| e.boundary.clone()).collect();
    let boundary_labels: Vec<String> = boundaries.iter().map(trust_boundary_label).collect();
    let reasons = collect_unique_reasons(events);
    let denied = events.iter().any(|e| e.effect == DecisionEffect::Deny);
    let redacted = events.iter().any(|e| e.effect == DecisionEffect::Redact);
    let preview_strings: Vec<String> = events
        .iter()
        .filter_map(|e| e.preview.clone())
        .take(4)
        .collect();
    let previews: Vec<Value> = preview_strings
        .iter()
        .map(|preview| Value::String(preview.clone()))
        .collect();

    let policy_subject = format!("run:{}:step:{}:policy_decisions", run_id, step_id);
    let policy_evidence = PolicyEvidence {
        effect: decision_effect_label(dominant_policy_effect(events)),
        boundaries: boundary_labels.clone(),
        reasons: reasons.clone(),
        previews: preview_strings,
        decision_attestation_ids: attestation_ids.clone(),
    };
    let identity = IdentityEvidence::new()
        .with_run_id(Uuid::from(run_id))
        .with_step_id(Uuid::from(step_id))
        .with_subject_label(policy_subject.clone());
    let manifest = generate_c2pa_manifest(ManifestInput {
        bytes: &payload_vec,
        media_type: "application/json",
        subject: Some(&policy_subject),
        attestation_ids: &attestation_ids,
        profile: ManifestProfile::PolicyEvidence,
        policy_evidence: Some(&policy_evidence),
        identity: Some(&identity),
        capability_evidence: None,
    })?;
    let payload = Bytes::from(payload_vec);
    let effects: Vec<DecisionEffect> = events.iter().map(|e| e.effect).collect();
    let origins = aggregate_origins(events);

    let signer = trust_signer();
    let envelope = signer
        .sign_json(&events_value)
        .context("failed to sign policy decisions payload")?;
    let signature = BASE64.encode(&envelope.signature);

    let mut metadata = Map::new();
    metadata.insert(
        "kind".to_string(),
        Value::String("policy_decisions".to_string()),
    );
    metadata.insert("count".to_string(), Value::from(events.len()));
    metadata.insert("effects".to_string(), serde_json::to_value(&effects)?);
    metadata.insert("boundaries".to_string(), serde_json::to_value(&boundaries)?);
    if !reasons.is_empty() {
        metadata.insert("reasons".to_string(), serde_json::to_value(&reasons)?);
    }
    metadata.insert("denied".to_string(), Value::Bool(denied));
    metadata.insert("redacted".to_string(), Value::Bool(redacted));
    if !origins.is_empty() {
        metadata.insert("origins".to_string(), Value::Array(origins.clone()));
    }
    metadata.insert("c2pa_manifest".to_string(), manifest.clone());
    metadata.insert("c2pa_verified".to_string(), Value::Bool(true));
    if !attestation_ids.is_empty() {
        let ids = attestation_ids
            .iter()
            .map(|id| Value::String(id.to_string()))
            .collect::<Vec<_>>();
        metadata.insert("attestation_ids".to_string(), Value::Array(ids));
    }
    metadata.insert(
        "signature".to_string(),
        json!({
            "algorithm": envelope.algorithm.as_str(),
            "key_id": envelope.key_id,
            "value": signature,
        }),
    );

    let meta = artifacts
        .put(payload, "application/json", Value::Object(metadata))
        .await?;

    let mut artifact_value = artifact_meta_to_value(&meta);
    if let Value::Object(ref mut obj) = artifact_value {
        let origin = TrustOrigin::new(TrustBoundary::Artifact)
            .with_run_id(Uuid::from(run_id))
            .with_step_id(Uuid::from(step_id))
            .with_source(TrustSource::System);
        obj.insert(
            "trust".to_string(),
            serde_json::to_value(Trust::derived(origin.clone()))?,
        );
        obj.insert("trust_origin".to_string(), serde_json::to_value(origin)?);
        if !origins.is_empty() {
            obj.insert("origins".to_string(), Value::Array(origins.clone()));
        }
    }

    let mut summary = Map::new();
    summary.insert("count".to_string(), Value::from(events.len()));
    summary.insert("effects".to_string(), serde_json::to_value(&effects)?);
    summary.insert("boundaries".to_string(), serde_json::to_value(&boundaries)?);
    if !reasons.is_empty() {
        summary.insert("reasons".to_string(), serde_json::to_value(&reasons)?);
    }
    summary.insert("denied".to_string(), Value::Bool(denied));
    summary.insert("redacted".to_string(), Value::Bool(redacted));
    if !previews.is_empty() {
        summary.insert("previews".to_string(), Value::Array(previews));
    }
    if !origins.is_empty() {
        summary.insert("origins".to_string(), Value::Array(origins));
    }

    Ok((artifact_value, Value::Object(summary)))
}

fn aggregate_origins(events: &[PolicyEvaluationRecord]) -> Vec<Value> {
    let mut aggregated = Vec::new();
    for event in events {
        for origin in &event.origins {
            if !aggregated.iter().any(|existing| existing == origin) {
                aggregated.push(origin.clone());
            }
        }
    }
    aggregated
}

fn collect_unique_reasons(events: &[PolicyEvaluationRecord]) -> Vec<String> {
    let mut reasons = Vec::new();
    for event in events {
        for reason in &event.reasons {
            push_unique_string(&mut reasons, reason);
        }
        for decision in &event.decisions {
            if let Some(reason) = &decision.reason {
                push_unique_string(&mut reasons, reason);
            }
        }
    }
    reasons
}

fn capability_violation_record(
    run_id: RunId,
    step_id: StepId,
    reasons: Vec<String>,
    detail: Value,
    token_id: Option<String>,
) -> CapabilityViolation {
    let summary = reasons
        .first()
        .cloned()
        .unwrap_or_else(|| "capability_denied".to_string());
    let decision_trace = PolicyDecisionTrace {
        policy: "capability_guard".to_string(),
        effect: DecisionEffect::Deny,
        reason: Some(summary.clone()),
        patches: Vec::new(),
    };
    let record = PolicyEvaluationRecord {
        boundary: TrustBoundary::Scheduler,
        run_id: Some(Uuid::from(run_id)),
        step_id: Some(Uuid::from(step_id)),
        effect: DecisionEffect::Deny,
        reasons,
        decisions: vec![decision_trace],
        patches: Vec::new(),
        payload: Some(detail),
        timestamp: Utc::now(),
        origins: Vec::new(),
        preview: None,
        attestation_id: None,
        attestation: None,
        trust_decision: None,
    };
    CapabilityViolation {
        record,
        reason: summary,
        token_id,
    }
}

impl Scheduler {
    async fn emit_run_bom(&self, run_id: RunId, status: models::RunStatus) -> Result<()> {
        let run_uuid = Uuid::from(run_id);
        let Some(run) = self.storage.fetch_run(run_uuid).await? else {
            warn!(run = %run_id, "run missing during BOM generation");
            return Ok(());
        };
        if has_run_artifact(&run.input_ctx, "aibom") {
            return Ok(());
        }
        let steps = self.storage.fetch_steps_for_run(run_uuid).await?;
        let build = build_aibom(&run, &steps, self.policy_pack.id())
            .context("failed to assemble AIBOM payload")?;
        let payload_bytes =
            serde_json::to_vec(&build.bom).context("failed to serialise AIBOM payload")?;
        let attestation_refs = build.attestation_ids.clone();
        let manifest_subject = format!("run:{}:aibom", run_id);
        let status_label = crate::model::RunStatus::from(status).as_str().to_string();
        let attestation_values: Vec<Value> = attestation_refs
            .iter()
            .map(|id| Value::String(id.to_string()))
            .collect();
        let mut metadata = json!({
            "kind": "aibom",
            "run_id": run_id.to_string(),
            "status": status_label,
            "trust_subjects": build.trust_subjects,
            "attestation_ids": attestation_values,
            "summary": build.summary,
        });
        let manifest_identity = IdentityEvidence::new()
            .with_run_id(Uuid::from(run_id))
            .with_subject_label(manifest_subject.clone());
        let manifest_profile = resolve_manifest_profile(Some(&metadata), ManifestProfile::Full);
        let manifest = generate_c2pa_manifest(ManifestInput {
            bytes: &payload_bytes,
            media_type: "application/json",
            subject: Some(&manifest_subject),
            attestation_ids: &attestation_refs,
            profile: manifest_profile,
            policy_evidence: None,
            identity: Some(&manifest_identity),
            capability_evidence: None,
        })?;
        let signer = trust_signer();
        let envelope = signer
            .sign(&payload_bytes)
            .context("failed to sign AIBOM payload")?;
        let signature_b64 = BASE64.encode(&envelope.signature);
        if let Some(obj) = metadata.as_object_mut() {
            obj.insert("c2pa_manifest".to_string(), manifest);
            obj.insert(
                "signature".to_string(),
                json!({
                    "algorithm": envelope.algorithm.as_str(),
                    "key_id": envelope.key_id,
                    "value": signature_b64,
                }),
            );
        }
        let artifact_meta = self
            .artifact_store
            .put(Bytes::from(payload_bytes), "application/json", metadata)
            .await
            .context("failed to persist AIBOM artifact")?;
        let mut artifact_value = artifact_meta_to_value(&artifact_meta);

        if self.config.transparency_writer
            && self.config.transparency_scope.includes_runs()
            && !attestation_refs.is_empty()
        {
            let payload = models::TransparencyJobPayload::RunRollup(models::RunRollupPayload {
                run_id: run_uuid,
                artifact_sha256: hex::encode(artifact_meta.sha256),
                attestation_ids: attestation_refs.clone(),
                artifact_key: "aibom".to_string(),
                metadata: artifact_meta.metadata.clone(),
            });
            match self.storage.enqueue_transparency_job(payload).await {
                Ok(job_id) => {
                    if let Value::Object(ref mut obj) = artifact_value {
                        obj.insert(
                            "transparency".to_string(),
                            json!({
                                "status": "queued",
                                "job_id": job_id,
                            }),
                        );
                    }
                }
                Err(err) => {
                    warn!(
                        run = %run_id,
                        error = %err,
                        "failed to enqueue transparency job for AIBOM"
                    );
                }
            }
        }

        if let Value::Object(ref mut obj) = artifact_value {
            if let Some(Value::Object(ref mut meta_obj)) = obj.get_mut("metadata") {
                meta_obj.insert("kind".to_string(), Value::String("aibom".to_string()));
            }
        }
        self.attach_run_artifact(run_uuid, "aibom", artifact_value)
            .await
    }

    async fn attach_run_artifact(&self, run_uuid: Uuid, key: &str, artifact: Value) -> Result<()> {
        let Some(run) = self.storage.fetch_run(run_uuid).await? else {
            warn!(run = %run_uuid, "run missing while attaching artifact");
            return Ok(());
        };
        let mut ctx_map = match run.input_ctx {
            Value::Object(map) => map,
            other => {
                let mut map = Map::new();
                map.insert("inputs".to_string(), other);
                map
            }
        };
        let entry = ctx_map
            .entry("artifacts".to_string())
            .or_insert_with(|| Value::Object(Map::new()));
        match entry {
            Value::Object(map) => {
                map.insert(key.to_string(), artifact);
            }
            _ => {
                let mut map = Map::new();
                map.insert(key.to_string(), artifact);
                ctx_map.insert("artifacts".to_string(), Value::Object(map));
            }
        }
        self.storage
            .update_run_input_ctx(run_uuid, Value::Object(ctx_map))
            .await
    }
}

fn push_unique_string(list: &mut Vec<String>, value: &str) {
    if !list.iter().any(|existing| existing == value) {
        list.push(value.to_string());
    }
}

fn has_run_artifact(ctx: &Value, key: &str) -> bool {
    ctx.as_object()
        .and_then(|map| map.get("artifacts"))
        .and_then(Value::as_object)
        .map(|artifacts| artifacts.contains_key(key))
        .unwrap_or(false)
}

#[derive(Clone)]
struct CapabilityViolation {
    record: PolicyEvaluationRecord,
    reason: String,
    token_id: Option<String>,
}

#[derive(Debug, Clone, Copy)]
struct BudgetRequest {
    tokens: i64,
    cost: f64,
}

impl BudgetRequest {
    fn from_step(step: &QueuedStep) -> Self {
        let budget_policy = step.spec.policy.get("budget");
        let tokens = budget_policy
            .and_then(|p| p.get("tokens"))
            .and_then(Value::as_i64);
        let execution_hint = step.spec.execution.cost_hint.as_ref();
        let tokens = execution_hint
            .and_then(|hint| hint.tokens)
            .or(tokens)
            .or_else(|| {
                step.spec
                    .inputs
                    .get("token_estimate")
                    .and_then(Value::as_i64)
            })
            .unwrap_or(0)
            .max(0);
        let cost = budget_policy
            .and_then(|p| p.get("cost"))
            .and_then(Value::as_f64);
        let cost = execution_hint
            .and_then(|hint| hint.usd)
            .or(cost)
            .or_else(|| {
                step.spec
                    .inputs
                    .get("cost_estimate")
                    .and_then(Value::as_f64)
            })
            .unwrap_or(0.0)
            .max(0.0);
        Self { tokens, cost }
    }
}

enum BudgetDecision {
    Granted(BudgetCtx),
    Denied { reason: String },
    Unlimited(BudgetCtx),
}

struct QueueMetrics {
    queue_depth: Histogram<f64>,
    queue_lag: Histogram<f64>,
    step_latency: Histogram<f64>,
    budget_tokens: Counter<u64>,
    budget_cost: Counter<f64>,
    budget_tokens_remaining: Histogram<f64>,
    budget_cost_remaining: Histogram<f64>,
    budget_tokens_projected: Histogram<f64>,
    budget_cost_projected: Histogram<f64>,
    retries: Counter<u64>,
    slo_slack: Histogram<f64>,
    slo_preemptions: Counter<u64>,
}

impl QueueMetrics {
    fn new() -> Self {
        let meter: Meter = telemetry_otel::meter("fleetforge.runtime");
        let queue_depth = meter
            .f64_histogram("fleetforge.queue.depth")
            .with_description("Sampled queue depth of the step scheduler")
            .init();
        let queue_lag = meter
            .f64_histogram("fleetforge.queue.lag_seconds")
            .with_description("Time a step spent waiting in the scheduler queue before execution")
            .init();
        let step_latency = meter
            .f64_histogram("fleetforge.step.latency")
            .with_description("End-to-end step execution latency in seconds")
            .init();
        let budget_tokens = meter
            .u64_counter("fleetforge.budget.tokens")
            .with_description("Total tokens consumed by steps")
            .init();
        let budget_cost = meter
            .f64_counter("fleetforge.budget.cost")
            .with_description("Total cost consumed by steps (USD)")
            .init();
        let budget_tokens_remaining = meter
            .f64_histogram("fleetforge.budget.tokens_remaining")
            .with_description("Remaining tokens in the active run budget after each step")
            .init();
        let budget_cost_remaining = meter
            .f64_histogram("fleetforge.budget.cost_remaining")
            .with_description("Remaining cost in the active run budget after each step (USD)")
            .init();
        let budget_tokens_projected = meter
            .f64_histogram("fleetforge.budget.tokens_projected_burn")
            .with_description(
                "Projected token consumption (used + reserved) prior to executing a step",
            )
            .init();
        let budget_cost_projected = meter
            .f64_histogram("fleetforge.budget.cost_projected_burn")
            .with_description("Projected cost (used + reserved) prior to executing a step (USD)")
            .init();
        let retries = meter
            .u64_counter("fleetforge.step.retries")
            .with_description("Number of step retries executed by the scheduler")
            .init();
        let slo_slack = meter
            .f64_histogram("fleetforge.slo.queue_slack_ms")
            .with_description("Observed queue slack (target - wait) for steps with SLO contracts, in milliseconds")
            .init();
        let slo_preemptions = meter
            .u64_counter("fleetforge.slo.preemptions")
            .with_description("Count of times the scheduler moved an SLO-contracted step ahead of another step due to a breach")
            .init();
        Self {
            queue_depth,
            queue_lag,
            step_latency,
            budget_tokens,
            budget_cost,
            budget_tokens_remaining,
            budget_cost_remaining,
            budget_tokens_projected,
            budget_cost_projected,
            retries,
            slo_slack,
            slo_preemptions,
        }
    }

    fn record_slo_slack(&self, slack_ms: Option<i64>) {
        if let Some(value) = slack_ms {
            self.slo_slack.record(value as f64, &[]);
        }
    }

    fn record_slo_preemption(&self) {
        self.slo_preemptions.add(1, &[]);
    }

    fn record_queue_depth(&self, depth: usize) {
        self.queue_depth.record(depth as f64, &[]);
    }

    fn record_queue_lag(&self, lag: Duration) {
        self.queue_lag.record(lag.as_secs_f64(), &[]);
    }

    fn record_latency(&self, duration: Duration) {
        self.step_latency.record(duration.as_secs_f64(), &[]);
    }

    fn record_retry(&self) {
        self.retries.add(1, &[]);
    }

    fn record_budget_projection(&self, budget: &BudgetCtx) {
        if budget.tokens_reserved > 0 {
            let projected = if budget.tracked {
                budget.tokens_used.saturating_add(budget.tokens_reserved)
            } else {
                budget.tokens_reserved
            };
            self.budget_tokens_projected.record(projected as f64, &[]);
        }
        if budget.cost_reserved > 0.0 {
            let projected = if budget.tracked {
                budget.cost_used + budget.cost_reserved
            } else {
                budget.cost_reserved
            };
            self.budget_cost_projected.record(projected, &[]);
        }
    }

    fn record_budget_consumed(&self, tokens: i64, cost: f64) {
        if tokens > 0 {
            self.budget_tokens.add(tokens as u64, &[]);
        }
        if cost > 0.0 {
            self.budget_cost.add(cost, &[]);
        }
    }

    fn record_budget_remaining(&self, tokens: Option<i64>, cost: Option<f64>) {
        if let Some(tokens) = tokens {
            if tokens >= 0 {
                self.budget_tokens_remaining.record(tokens as f64, &[]);
            }
        }
        if let Some(cost) = cost {
            if cost >= 0.0 {
                self.budget_cost_remaining.record(cost, &[]);
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::guardrails::PolicyDecisionTrace;
    use crate::model::{RunId, StepId};
    use async_trait::async_trait;
    use bytes::Bytes;
    use chrono::Utc;
    use fleetforge_storage::ArtifactMeta;
    use fleetforge_trust::InMemoryAttestationVault;
    use serde_json::{json, Value};
    use sqlx::Row;
    use std::sync::{
        atomic::{AtomicBool, Ordering},
        Arc,
    };
    use testcontainers::clients::Cli;
    use testcontainers::images::postgres::Postgres;
    use tokio::time::{sleep, Duration};
    use url::Url;
    use uuid::Uuid;

    fn docker_available() -> bool {
        std::fs::metadata("/var/run/docker.sock").is_ok() || std::env::var("DOCKER_HOST").is_ok()
    }

    fn test_step_with_policy(
        policy: Value,
        created_at: chrono::DateTime<Utc>,
        priority: i16,
    ) -> QueuedStep {
        QueuedStep {
            run_id: RunId(Uuid::new_v4()),
            step_id: StepId(Uuid::new_v4()),
            idx: 0,
            priority,
            status: StepStatus::Queued,
            attempt: 0,
            created_at,
            max_attempts: 1,
            retry_backoff_ms: 0,
            retry_backoff_factor: 1.0,
            not_before: None,
            deadline_at: None,
            checkpoint: None,
            compensation_step: None,
            compensation_scheduled: false,
            spec: StepSpec {
                id: StepId(Uuid::new_v4()),
                r#type: StepType::Tool,
                inputs: json!({}),
                policy,
                execution: StepExecution::default(),
                slug: None,
                trust: None,
                trust_origin: None,
                llm_inputs: None,
            },
        }
    }

    #[test]
    fn build_step_slo_includes_queue_observations() {
        let now = Utc::now();
        let policy = json!({
            "slo": {
                "tier": "gold",
                "queue_target_ms": 1000,
                "priority_boost": 3,
                "error_budget_ratio": 0.25
            }
        });
        let step = test_step_with_policy(policy, now - ChronoDuration::milliseconds(1500), 5);
        let slo = build_step_slo(&step, now).expect("slo metadata expected");

        assert_eq!(slo.tier.as_deref(), Some("gold"));
        assert_eq!(slo.queue_target_ms, Some(1000));
        assert!(slo.observed_queue_ms.unwrap() >= 1500);
        assert_eq!(slo.slack_ms, Some(-500));
        assert_eq!(slo.priority_boost, 3);
        assert_eq!(slo.error_budget_ratio, Some(0.25));
    }

    #[test]
    fn slo_scores_prioritize_breaches_over_healthy_steps() {
        let now = Utc::now();
        let breached = test_step_with_policy(
            json!({ "slo": { "queue_target_ms": 2000 } }),
            now - ChronoDuration::milliseconds(4000),
            0,
        );
        let healthy = test_step_with_policy(
            json!({ "slo": { "queue_target_ms": 2000 } }),
            now - ChronoDuration::milliseconds(100),
            0,
        );

        let breached_contract = extract_slo_contract(&breached.spec);
        let healthy_contract = extract_slo_contract(&healthy.spec);

        let breached_score = compute_slo_score(&breached, breached_contract.as_ref(), now);
        let healthy_score = compute_slo_score(&healthy, healthy_contract.as_ref(), now);

        assert!(breached_score.has_contract);
        assert!(healthy_score.has_contract);
        assert!(breached_score.slack_ms < 0);
        assert!(healthy_score.slack_ms > 0);
        assert_eq!(breached_score.compare(&healthy_score), Ordering::Less);
    }

    #[derive(Default, Clone)]
    struct StubArtifactStore;

    #[async_trait]
    impl ArtifactStore for StubArtifactStore {
        async fn put(
            &self,
            bytes: Bytes,
            media_type: &str,
            metadata: Value,
        ) -> Result<ArtifactMeta> {
            Ok(ArtifactMeta {
                sha256: [0u8; 32],
                media_type: media_type.to_string(),
                size: bytes.len() as u64,
                metadata,
            })
        }

        async fn get(&self, _sha256: [u8; 32]) -> Result<Bytes> {
            unimplemented!("get not implemented for stub artifact store");
        }

        async fn presign_get(&self, _sha256: [u8; 32], _ttl: std::time::Duration) -> Result<Url> {
            unimplemented!("presign not implemented for stub artifact store");
        }
    }

    #[tokio::test]
    async fn persist_policy_decisions_emits_summary() {
        let store = Arc::new(StubArtifactStore::default());
        let event = PolicyEvaluationRecord {
            boundary: TrustBoundary::IngressPrompt,
            run_id: Some(Uuid::nil()),
            step_id: Some(Uuid::nil()),
            effect: DecisionEffect::Redact,
            reasons: vec!["mask".into()],
            decisions: vec![PolicyDecisionTrace {
                policy: "test".into(),
                effect: DecisionEffect::Redact,
                reason: Some("mask".into()),
                patches: vec![
                    json!({"op": "replace", "path": "/inputs/secret", "value": "[masked]"}),
                ],
            }],
            patches: vec![json!({"op": "replace", "path": "/inputs/secret", "value": "[masked]"})],
            payload: Some(json!({"inputs": {"secret": "[masked]"}})),
            timestamp: Utc::now(),
            origins: vec![json!({"boundary": "ingress_prompt"})],
            preview: Some("secret".into()),
            attestation_id: None,
            attestation: None,
            trust_decision: None,
        };

        let (artifact, summary) =
            persist_policy_decisions(store, RunId(Uuid::nil()), StepId(Uuid::nil()), &[event])
                .await
                .expect("persist policy decisions");

        assert_eq!(summary["count"], 1);
        assert_eq!(summary["redacted"], true);
        assert!(summary["origins"].is_array());
        assert!(summary["previews"].is_array());
        assert_eq!(artifact["metadata"]["kind"], "policy_decisions");
        assert!(artifact["metadata"]["signature"].is_object());
        assert_eq!(artifact["metadata"]["signature"]["algorithm"], "HS256");
        assert!(artifact["metadata"]["signature"]["key_id"].is_string());
        assert!(artifact["metadata"]["signature"]["value"]
            .as_str()
            .is_some());
        assert!(artifact["trust_origin"].is_object());
    }

    #[test]
    fn attach_policy_summary_inserts_artifact() {
        let mut target = json!({"existing": true});
        let summary = json!({"count": 2});
        let artifact = json!({"sha256": "abc123", "metadata": {"kind": "policy_decisions"}});
        attach_policy_summary(&mut target, summary, artifact.clone());

        assert_eq!(target["policy_decisions"]["count"], 2);
        assert_eq!(target["policy_decisions"]["artifact"]["sha256"], "abc123");
        assert_eq!(
            target["artifacts"]["policy_decisions"]["metadata"]["kind"],
            "policy_decisions"
        );
    }

    #[derive(Clone)]
    struct DenyPolicy;

    #[async_trait]
    impl PolicyEngine for DenyPolicy {
        async fn evaluate(
            &self,
            _request: &fleetforge_policy::PolicyRequest,
        ) -> Result<PolicyDecision> {
            Ok(PolicyDecision {
                effect: DecisionEffect::Deny,
                reason: Some("blocked".to_string()),
                patches: Vec::new(),
            })
        }
    }

    #[derive(Clone)]
    struct RedactPolicy;

    #[async_trait]
    impl PolicyEngine for RedactPolicy {
        async fn evaluate(
            &self,
            _request: &fleetforge_policy::PolicyRequest,
        ) -> Result<PolicyDecision> {
            Ok(PolicyDecision {
                effect: DecisionEffect::Redact,
                reason: Some("mask".to_string()),
                patches: vec![json!({"op": "remove", "path": "/secret"})],
            })
        }
    }

    struct RecordingExecutor {
        executed: Arc<AtomicBool>,
    }

    #[async_trait]
    impl StepExecutor for RecordingExecutor {
        fn kind(&self) -> &'static str {
            "tool"
        }

        async fn execute(&self, _ctx: &StepCtx) -> Result<StepExecutionResult> {
            self.executed.store(true, Ordering::SeqCst);
            Ok(StepExecutionResult::new(json!({"ok": true})))
        }
    }

    struct UsageExecutor {
        executed: Arc<AtomicBool>,
    }

    #[async_trait]
    impl StepExecutor for UsageExecutor {
        fn kind(&self) -> &'static str {
            "tool"
        }

        async fn execute(&self, _ctx: &StepCtx) -> Result<StepExecutionResult> {
            self.executed.store(true, Ordering::SeqCst);
            Ok(StepExecutionResult::with_usage(
                json!({"result": "ok"}),
                ResourceUsage {
                    tokens: Some(10),
                    cost: Some(0.0005),
                },
            ))
        }
    }

    struct FailingExecutor;

    #[async_trait]
    impl StepExecutor for FailingExecutor {
        fn kind(&self) -> &'static str {
            "tool"
        }

        async fn execute(&self, _ctx: &StepCtx) -> Result<StepExecutionResult> {
            Err(anyhow!("executor failure"))
        }
    }

    struct ConditionalExecutor;

    #[async_trait]
    impl StepExecutor for ConditionalExecutor {
        fn kind(&self) -> &'static str {
            "tool"
        }

        async fn execute(&self, ctx: &StepCtx) -> Result<StepExecutionResult> {
            let mode = ctx
                .untrusted_inputs
                .as_ref()
                .get("mode")
                .and_then(Value::as_str)
                .unwrap_or_default();
            match mode {
                "primary" => Err(anyhow!("primary failure")),
                "compensation" => Ok(StepExecutionResult::with_usage(
                    json!({"status": "compensated"}),
                    ResourceUsage {
                        tokens: Some(5),
                        cost: Some(0.25),
                    },
                )),
                _ => Ok(StepExecutionResult::new(json!({"status": mode}))),
            }
        }
    }

    async fn setup_storage() -> (Arc<Storage>, testcontainers::Container<'static, Postgres>) {
        let docker = Cli::default();
        let container = docker.run(Postgres::default());
        let port = container.get_host_port_ipv4(5432);
        let db_url = format!("postgres://postgres:postgres@127.0.0.1:{port}/postgres");
        sleep(Duration::from_millis(250)).await;

        let storage = Storage::connect(StorageConfig {
            database_url: db_url,
            max_connections: 2,
            connect_timeout: Duration::from_secs(10),
            object_store: ObjectStoreConfig::InMemory,
        })
        .await
        .expect("connect storage");

        let container: testcontainers::Container<'static, Postgres> =
            unsafe { std::mem::transmute(container) };
        (Arc::new(storage), container)
    }

    fn seed_step(run_id: Uuid, step_id: Uuid) -> models::NewStep {
        models::NewStep {
            run_id,
            step_id,
            idx: 0,
            priority: 5,
            spec_json: json!({"id": step_id, "type": "tool", "inputs": {}, "policy": {}}),
            status: models::StepStatus::Queued,
            attempt: 0,
            input_snapshot: None,
            provider: None,
            provider_version: None,
            pending_dependencies: 0,
            total_dependencies: 0,
            estimated_tokens: 50,
            estimated_cost: 0.001,
            actual_tokens: None,
            actual_cost: None,
            ..models::NewStep::default()
        }
    }

    #[tokio::test]
    async fn deny_decision_blocks_execution() {
        if std::fs::metadata("/var/run/docker.sock").is_err()
            && std::env::var("DOCKER_HOST").is_err()
        {
            eprintln!("Skipping scheduler tests because Docker is unavailable");
            return;
        }

        let (storage, container) = setup_storage().await;
        let executed = Arc::new(AtomicBool::new(false));
        let mut registry = ExecutorRegistry::new();
        registry
            .register(Arc::new(RecordingExecutor {
                executed: Arc::clone(&executed),
            }))
            .unwrap();
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());

        let scheduler = Scheduler::new(
            Arc::clone(&storage),
            Arc::new(registry),
            Arc::new(DenyPolicy),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            SchedulerConfig::default(),
        );

        let run_id = Uuid::new_v4();
        let step_id = Uuid::new_v4();

        let _ = storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({"steps": [], "edges": []}),
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: None,
                },
                &[seed_step(run_id, step_id)],
                &[],
            )
            .await
            .unwrap();

        let cfg = SchedulerConfig::default();
        let leased = storage.claim_queued_steps(1, &cfg.worker_id).await.unwrap();
        let queued = leased
            .into_iter()
            .map(QueuedStep::try_from)
            .next()
            .unwrap()
            .unwrap();

        scheduler.execute_step(queued).await.unwrap();

        assert!(!executed.load(Ordering::SeqCst));

        let steps = storage.fetch_steps_for_run(run_id).await.unwrap();
        let step = &steps[0];
        assert_eq!(step.status, models::StepStatus::Failed);
        assert_eq!(step.attempt, 0);
        assert_eq!(
            step.error_json.as_ref().unwrap()["kind"],
            json!("policy_denied")
        );

        let events = storage.fetch_unpublished_outbox(10).await.unwrap();
        let event = events.iter().find(|ev| ev.kind == "step_failed").unwrap();
        assert_eq!(event.payload["error"]["kind"], json!("policy_denied"));

        drop(container);
    }

    #[tokio::test]
    async fn retry_reschedules_until_max_attempts() {
        if !docker_available() {
            eprintln!(
                "Skipping retry_reschedules_until_max_attempts because Docker is unavailable"
            );
            return;
        }

        let (storage, container) = setup_storage().await;
        let mut registry = ExecutorRegistry::new();
        registry.register(Arc::new(FailingExecutor)).unwrap();
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());
        let scheduler = Scheduler::new(
            Arc::clone(&storage),
            Arc::new(registry),
            crate::policy::AllowAllPolicy::shared(),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            SchedulerConfig::default(),
        );

        let run_id = Uuid::new_v4();
        let step_id = Uuid::new_v4();
        let step_spec = json!({
            "id": step_id,
            "type": "tool",
            "inputs": {},
            "execution": {
                "max_attempts": 2,
                "retry_backoff_ms": 10
            }
        });

        let spec = json!({
            "steps": [step_spec.clone()],
            "edges": []
        });

        storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: spec,
                    input_ctx: json!({}),
                    seed: 7,
                    idempotency_key: None,
                },
                &[models::NewStep {
                    run_id,
                    step_id,
                    idx: 0,
                    priority: 0,
                    spec_json: step_spec,
                    status: models::StepStatus::Queued,
                    attempt: 0,
                    input_snapshot: None,
                    provider: None,
                    provider_version: None,
                    pending_dependencies: 0,
                    total_dependencies: 0,
                    estimated_tokens: 0,
                    estimated_cost: 0.0,
                    actual_tokens: None,
                    actual_cost: None,
                    ..models::NewStep::default()
                }],
                &[],
            )
            .await
            .unwrap();

        let cfg = scheduler.config.clone();
        let first = storage
            .claim_queued_steps(1, &cfg.worker_id)
            .await
            .unwrap()
            .into_iter()
            .map(QueuedStep::try_from)
            .next()
            .unwrap()
            .unwrap();

        scheduler.execute_step(first).await.unwrap();

        let steps = storage.fetch_steps_for_run(run_id).await.unwrap();
        let queued = steps.iter().find(|step| step.step_id == step_id).unwrap();
        assert_eq!(queued.status, models::StepStatus::Queued);
        assert_eq!(queued.attempt, 1);
        assert!(queued.not_before.is_some());

        sleep(Duration::from_millis(20)).await;

        let second = storage
            .claim_queued_steps(1, &cfg.worker_id)
            .await
            .unwrap()
            .into_iter()
            .map(QueuedStep::try_from)
            .next()
            .unwrap()
            .unwrap();

        scheduler.execute_step(second).await.unwrap();

        let final_steps = storage.fetch_steps_for_run(run_id).await.unwrap();
        let failed = final_steps
            .iter()
            .find(|step| step.step_id == step_id)
            .unwrap();
        assert_eq!(failed.status, models::StepStatus::Failed);
        assert!(failed.compensation_step.is_none());

        drop(container);
    }

    #[tokio::test]
    async fn backpressure_defers_steps_exceeding_inflight_cost() {
        if !docker_available() {
            eprintln!("Skipping backpressure test because Docker is unavailable");
            return;
        }

        let (storage, container) = setup_storage().await;
        let executed = Arc::new(AtomicBool::new(false));
        let mut registry = ExecutorRegistry::new();
        registry
            .register(Arc::new(RecordingExecutor {
                executed: Arc::clone(&executed),
            }))
            .unwrap();
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());
        let mut config = SchedulerConfig::default();
        config.batch_size = 2;
        config.max_inflight_cost = Some(1.0);
        config.backpressure_delay = Duration::from_millis(20);
        let scheduler = Scheduler::new(
            Arc::clone(&storage),
            Arc::new(registry),
            crate::policy::AllowAllPolicy::shared(),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            config,
        );

        let run_id = Uuid::new_v4();
        let first_id = Uuid::new_v4();
        let second_id = Uuid::new_v4();

        let first_spec = json!({
            "id": first_id,
            "type": "tool",
            "inputs": {},
            "execution": { "cost_hint": { "usd": 1.0 } }
        });
        let second_spec = json!({
            "id": second_id,
            "type": "tool",
            "inputs": {},
            "execution": { "cost_hint": { "usd": 0.5 } }
        });

        storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({ "steps": [first_spec.clone(), second_spec.clone()], "edges": [] }),
                    input_ctx: json!({}),
                    seed: 99,
                    idempotency_key: None,
                },
                &[
                    models::NewStep {
                        run_id,
                        step_id: first_id,
                        idx: 0,
                        priority: 0,
                        spec_json: first_spec,
                        status: models::StepStatus::Queued,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 0,
                        total_dependencies: 0,
                        estimated_tokens: 0,
                        estimated_cost: 1.0,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                    models::NewStep {
                        run_id,
                        step_id: second_id,
                        idx: 1,
                        priority: 0,
                        spec_json: second_spec,
                        status: models::StepStatus::Queued,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 0,
                        total_dependencies: 0,
                        estimated_tokens: 0,
                        estimated_cost: 0.5,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                ],
                &[],
            )
            .await
            .unwrap();

        let cfg = scheduler.config.clone();
        let mut leased: Vec<QueuedStep> = storage
            .claim_queued_steps(2, &cfg.worker_id)
            .await
            .unwrap()
            .into_iter()
            .map(|step| QueuedStep::try_from(step).unwrap())
            .collect();

        let first = leased.remove(0);
        scheduler.execute_step(first).await.unwrap();
        assert!(executed.load(Ordering::SeqCst));

        // Simulate inflight usage before processing the second step
        scheduler.add_inflight(0, 1.0).await;

        let second = leased.remove(0);
        scheduler.execute_step(second).await.unwrap();

        let steps = storage.fetch_steps_for_run(run_id).await.unwrap();
        let deferred = steps.iter().find(|step| step.step_id == second_id).unwrap();
        assert_eq!(deferred.status, models::StepStatus::Queued);
        assert!(deferred.not_before.is_some());
        assert_eq!(deferred.attempt, 0);

        scheduler.release_inflight(0, 1.0).await;
        drop(container);
    }

    struct CheckpointExecutor;

    #[async_trait]
    impl StepExecutor for CheckpointExecutor {
        fn kind(&self) -> &'static str {
            "tool"
        }

        async fn execute(&self, _ctx: &StepCtx) -> Result<StepExecutionResult> {
            Ok(StepExecutionResult::new(json!({"ok": true})).with_checkpoint(json!({"state": 42})))
        }
    }

    #[tokio::test]
    async fn checkpoint_is_persisted_from_executor() {
        if !docker_available() {
            eprintln!("Skipping checkpoint test because Docker is unavailable");
            return;
        }

        let (storage, container) = setup_storage().await;
        let mut registry = ExecutorRegistry::new();
        registry.register(Arc::new(CheckpointExecutor)).unwrap();
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());
        let scheduler = Scheduler::new(
            Arc::clone(&storage),
            Arc::new(registry),
            crate::policy::AllowAllPolicy::shared(),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            SchedulerConfig::default(),
        );

        let run_id = Uuid::new_v4();
        let step_id = Uuid::new_v4();

        let step_spec = json!({
            "id": step_id,
            "type": "tool",
            "inputs": {},
            "execution": { "checkpoint": true }
        });

        storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({ "steps": [step_spec.clone()], "edges": [] }),
                    input_ctx: json!({}),
                    seed: 11,
                    idempotency_key: None,
                },
                &[models::NewStep {
                    run_id,
                    step_id,
                    idx: 0,
                    priority: 0,
                    spec_json: step_spec,
                    status: models::StepStatus::Queued,
                    attempt: 0,
                    input_snapshot: None,
                    provider: None,
                    provider_version: None,
                    pending_dependencies: 0,
                    total_dependencies: 0,
                    estimated_tokens: 0,
                    estimated_cost: 0.0,
                    actual_tokens: None,
                    actual_cost: None,
                    ..models::NewStep::default()
                }],
                &[],
            )
            .await
            .unwrap();

        let worker = scheduler.config.worker_id.clone();
        let queued = storage
            .claim_queued_steps(1, &worker)
            .await
            .unwrap()
            .into_iter()
            .map(QueuedStep::try_from)
            .next()
            .unwrap()
            .unwrap();

        scheduler.execute_step(queued).await.unwrap();

        let steps = storage.fetch_steps_for_run(run_id).await.unwrap();
        let step = steps.iter().find(|item| item.step_id == step_id).unwrap();
        assert_eq!(step.status, models::StepStatus::Succeeded);
        let checkpoint = step.checkpoint.as_ref().expect("checkpoint missing");
        assert_eq!(checkpoint, &json!({"state": 42}));

        drop(container);
    }

    #[tokio::test]
    async fn compensation_step_queued_on_terminal_failure() {
        if !docker_available() {
            eprintln!("Skipping compensation test because Docker is unavailable");
            return;
        }

        let (storage, container) = setup_storage().await;
        let mut registry = ExecutorRegistry::new();
        registry.register(Arc::new(FailingExecutor)).unwrap();
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());
        let scheduler = Scheduler::new(
            Arc::clone(&storage),
            Arc::new(registry),
            crate::policy::AllowAllPolicy::shared(),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            SchedulerConfig::default(),
        );

        let run_id = Uuid::new_v4();
        let primary_id = Uuid::new_v4();
        let compensation_id = Uuid::new_v4();

        let primary_spec = json!({
            "id": primary_id,
            "type": "tool",
            "inputs": {},
            "execution": {
                "max_attempts": 1,
                "compensation": { "step": compensation_id.to_string() }
            }
        });

        let compensation_spec = json!({
            "id": compensation_id,
            "type": "tool",
            "inputs": { "action": "rollback" }
        });

        storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({ "steps": [primary_spec.clone(), compensation_spec.clone()], "edges": [] }),
                    input_ctx: json!({}),
                    seed: 13,
                    idempotency_key: None,
                },
                &[
                    models::NewStep {
                        run_id,
                        step_id: primary_id,
                        idx: 0,
                        priority: 1,
                        spec_json: primary_spec,
                        status: models::StepStatus::Queued,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 0,
                        total_dependencies: 0,
                        estimated_tokens: 0,
                        estimated_cost: 0.0,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                    models::NewStep {
                        run_id,
                        step_id: compensation_id,
                        idx: 1,
                        priority: 0,
                        spec_json: compensation_spec,
                        status: models::StepStatus::Blocked,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 1,
                        total_dependencies: 1,
                        estimated_tokens: 0,
                        estimated_cost: 0.0,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                ],
                &[],
            )
            .await
            .unwrap();

        let worker = scheduler.config.worker_id.clone();
        let queued = storage
            .claim_queued_steps(1, &worker)
            .await
            .unwrap()
            .into_iter()
            .map(QueuedStep::try_from)
            .next()
            .unwrap()
            .unwrap();

        scheduler.execute_step(queued).await.unwrap();

        let steps = storage.fetch_steps_for_run(run_id).await.unwrap();
        let primary = steps
            .iter()
            .find(|step| step.step_id == primary_id)
            .unwrap();
        assert_eq!(primary.status, models::StepStatus::Failed);
        assert!(primary.compensation_scheduled);

        let compensation = steps
            .iter()
            .find(|step| step.step_id == compensation_id)
            .unwrap();
        assert_eq!(compensation.status, models::StepStatus::Queued);
        assert_eq!(compensation.pending_dependencies, 0);

        drop(container);
    }

    #[tokio::test]
    async fn compensation_step_executes_and_records_ledger() {
        if !docker_available() {
            eprintln!("Skipping compensation ledger test because Docker is unavailable");
            return;
        }

        let (storage, container) = setup_storage().await;
        let mut registry = ExecutorRegistry::new();
        registry.register(Arc::new(ConditionalExecutor)).unwrap();
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());
        let scheduler = Scheduler::new(
            Arc::clone(&storage),
            Arc::new(registry),
            crate::policy::AllowAllPolicy::shared(),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            SchedulerConfig::default(),
        );

        let run_id = Uuid::new_v4();
        let primary_id = Uuid::new_v4();
        let compensation_id = Uuid::new_v4();

        let primary_spec = json!({
            "id": primary_id,
            "type": "tool",
            "inputs": { "mode": "primary" },
            "execution": {
                "max_attempts": 1,
                "compensation": { "step": compensation_id.to_string() },
                "cost_hint": { "tokens": 50, "usd": 1.5 }
            }
        });
        let compensation_spec = json!({
            "id": compensation_id,
            "type": "tool",
            "inputs": { "mode": "compensation" },
            "execution": { "cost_hint": { "tokens": 10, "usd": 0.5 } }
        });

        storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({
                        "steps": [primary_spec.clone(), compensation_spec.clone()],
                        "edges": []
                    }),
                    input_ctx: json!({}),
                    seed: 17,
                    idempotency_key: None,
                },
                &[
                    models::NewStep {
                        run_id,
                        step_id: primary_id,
                        idx: 0,
                        priority: 1,
                        spec_json: primary_spec,
                        status: models::StepStatus::Queued,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 0,
                        total_dependencies: 0,
                        estimated_tokens: 50,
                        estimated_cost: 1.5,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                    models::NewStep {
                        run_id,
                        step_id: compensation_id,
                        idx: 1,
                        priority: 0,
                        spec_json: compensation_spec,
                        status: models::StepStatus::Blocked,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 1,
                        total_dependencies: 1,
                        estimated_tokens: 10,
                        estimated_cost: 0.5,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                ],
                &[],
            )
            .await
            .unwrap();

        let worker = scheduler.config.worker_id.clone();
        let queued = storage
            .claim_queued_steps(1, &worker)
            .await
            .unwrap()
            .into_iter()
            .map(QueuedStep::try_from)
            .next()
            .unwrap()
            .unwrap();

        scheduler.execute_step(queued).await.unwrap();

        let steps = storage.fetch_steps_for_run(run_id).await.unwrap();
        let primary = steps
            .iter()
            .find(|step| step.step_id == primary_id)
            .unwrap();
        assert_eq!(primary.status, models::StepStatus::Failed);
        assert!(primary.compensation_scheduled);

        let compensation = storage
            .claim_queued_steps(1, &worker)
            .await
            .unwrap()
            .into_iter()
            .map(QueuedStep::try_from)
            .next()
            .unwrap()
            .unwrap();

        scheduler.execute_step(compensation).await.unwrap();

        let steps = storage.fetch_steps_for_run(run_id).await.unwrap();
        let primary = steps
            .iter()
            .find(|step| step.step_id == primary_id)
            .unwrap();
        assert_eq!(primary.status, models::StepStatus::Failed);
        assert!(primary.compensation_scheduled);
        let compensation = steps
            .iter()
            .find(|step| step.step_id == compensation_id)
            .unwrap();
        assert_eq!(compensation.status, models::StepStatus::Succeeded);
        assert_eq!(compensation.attempt, 1);

        let ledger_rows = sqlx::query(
            "select step_id, status, reserved_tokens, reserved_cost, actual_tokens, actual_cost \
             from step_cost_ledger where run_id = $1 order by recorded_at",
        )
        .bind(run_id)
        .fetch_all(storage.pool())
        .await
        .unwrap();

        let mut saw_failed = false;
        let mut saw_comp_scheduled = false;
        let mut saw_comp_success = false;

        for row in ledger_rows {
            let step_id: Uuid = row.get("step_id");
            let status: String = row.get("status");
            if step_id == primary_id && status == "failed" {
                saw_failed = true;
                assert_eq!(row.get::<i64, _>("reserved_tokens"), 50);
                assert_eq!(row.get::<f64, _>("reserved_cost"), 1.5);
            } else if step_id == primary_id && status == "compensation_scheduled" {
                saw_comp_scheduled = true;
            } else if step_id == compensation_id && status == "succeeded" {
                saw_comp_success = true;
                assert_eq!(row.get::<i64, _>("reserved_tokens"), 10);
                assert_eq!(row.get::<f64, _>("reserved_cost"), 0.5);
                assert_eq!(row.get::<Option<i64>, _>("actual_tokens"), Some(5));
                assert_eq!(row.get::<Option<f64>, _>("actual_cost"), Some(0.25));
            }
        }

        assert!(saw_failed, "missing failed ledger entry for primary step");
        assert!(
            saw_comp_scheduled,
            "missing compensation_scheduled ledger entry for primary step"
        );
        assert!(
            saw_comp_success,
            "missing succeeded ledger entry for compensation step"
        );

        drop(container);
    }

    #[tokio::test]
    async fn failing_step_skips_dependents() {
        if std::fs::metadata("/var/run/docker.sock").is_err()
            && std::env::var("DOCKER_HOST").is_err()
        {
            eprintln!("Skipping scheduler tests because Docker is unavailable");
            return;
        }

        let (storage, container) = setup_storage().await;
        let mut registry = ExecutorRegistry::new();
        registry.register(Arc::new(FailingExecutor)).unwrap();
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());

        let scheduler = Scheduler::new(
            Arc::clone(&storage),
            Arc::new(registry),
            crate::policy::AllowAllPolicy::shared(),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            SchedulerConfig::default(),
        );

        let run_id = Uuid::new_v4();
        let root_id = Uuid::new_v4();
        let child_id = Uuid::new_v4();

        let root_step = models::NewStep {
            run_id,
            step_id: root_id,
            idx: 0,
            priority: 0,
            spec_json: json!({"id": root_id, "type": "tool", "inputs": {}, "policy": {}}),
            status: models::StepStatus::Queued,
            attempt: 0,
            input_snapshot: None,
            provider: None,
            provider_version: None,
            pending_dependencies: 0,
            total_dependencies: 0,
            estimated_tokens: 0,
            estimated_cost: 0.0,
            actual_tokens: None,
            actual_cost: None,
            ..models::NewStep::default()
        };
        let child_step = models::NewStep {
            run_id,
            step_id: child_id,
            idx: 1,
            priority: 0,
            spec_json: json!({"id": child_id, "type": "tool", "inputs": {}, "policy": {}}),
            status: models::StepStatus::Blocked,
            attempt: 0,
            input_snapshot: None,
            provider: None,
            provider_version: None,
            pending_dependencies: 1,
            total_dependencies: 1,
            estimated_tokens: 0,
            estimated_cost: 0.0,
            actual_tokens: None,
            actual_cost: None,
            ..models::NewStep::default()
        };

        let _ = storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({
                        "steps": [root_step.spec_json.clone(), child_step.spec_json.clone()],
                        "edges": [[root_id, child_id]]
                    }),
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: None,
                },
                &[root_step, child_step],
                &[(root_id, child_id)],
            )
            .await
            .unwrap();

        let cfg = SchedulerConfig::default();
        let leased = storage.claim_queued_steps(1, &cfg.worker_id).await.unwrap();
        let queued = leased
            .into_iter()
            .map(QueuedStep::try_from)
            .next()
            .unwrap()
            .unwrap();

        scheduler.execute_step(queued).await.unwrap();

        let steps = storage.fetch_steps_for_run(run_id).await.unwrap();
        let root = steps.iter().find(|step| step.step_id == root_id).unwrap();
        let child = steps.iter().find(|step| step.step_id == child_id).unwrap();

        assert_eq!(root.status, models::StepStatus::Failed);
        assert_eq!(child.status, models::StepStatus::Skipped);

        let events = storage.fetch_unpublished_outbox(10).await.unwrap();
        assert!(events.iter().any(|ev| ev.kind == "step_failed"));
        assert!(events.iter().any(|ev| ev.kind == "step_skipped"));

        drop(container);
    }

    #[tokio::test]
    async fn redact_decision_logs_and_allows() {
        if std::fs::metadata("/var/run/docker.sock").is_err()
            && std::env::var("DOCKER_HOST").is_err()
        {
            eprintln!("Skipping scheduler tests because Docker is unavailable");
            return;
        }

        let (storage, container) = setup_storage().await;
        let executed = Arc::new(AtomicBool::new(false));
        let mut registry = ExecutorRegistry::new();
        registry
            .register(Arc::new(UsageExecutor {
                executed: Arc::clone(&executed),
            }))
            .unwrap();
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());

        let scheduler = Scheduler::new(
            Arc::clone(&storage),
            Arc::new(registry),
            Arc::new(RedactPolicy),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            SchedulerConfig::default(),
        );

        let run_id = Uuid::new_v4();
        let step_id = Uuid::new_v4();

        let _ = storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({"steps": [], "edges": []}),
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: None,
                },
                &[seed_step(run_id, step_id)],
                &[],
            )
            .await
            .unwrap();

        let cfg = SchedulerConfig::default();
        let leased = storage.claim_queued_steps(1, &cfg.worker_id).await.unwrap();
        let queued = leased
            .into_iter()
            .map(QueuedStep::try_from)
            .next()
            .unwrap()
            .unwrap();

        scheduler.execute_step(queued).await.unwrap();

        assert!(executed.load(Ordering::SeqCst));

        let steps = storage.fetch_steps_for_run(run_id).await.unwrap();
        let step = &steps[0];
        assert_eq!(step.status, models::StepStatus::Succeeded);
        assert_eq!(step.attempt, 1);
        assert_eq!(step.actual_tokens, Some(10));

        let events = storage.fetch_unpublished_outbox(10).await.unwrap();
        let event = events
            .iter()
            .find(|ev| ev.kind == "step_succeeded")
            .unwrap();
        assert_eq!(event.payload["budget"]["actual"]["tokens"], json!(10));

        drop(container);
    }

    #[tokio::test]
    async fn reserve_budget_granted_denied_unlimited() {
        if std::fs::metadata("/var/run/docker.sock").is_err()
            && std::env::var("DOCKER_HOST").is_err()
        {
            eprintln!("Skipping scheduler tests because Docker is unavailable");
            return;
        }

        let (storage, container) = setup_storage().await;
        let registry = ExecutorRegistry::new();
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());
        let scheduler = Scheduler::new(
            Arc::clone(&storage),
            Arc::new(registry),
            Arc::new(RedactPolicy),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            SchedulerConfig::default(),
        );

        let run_id = Uuid::new_v4();
        // insert budget row with small limits
        sqlx::query(
            "insert into budgets (budget_id, run_id, tokens_limit, tokens_used, cost_limit, cost_used) values (gen_random_uuid(), $1, 100, 0, 10.0, 0.0)"
        )
        .bind(run_id)
        .execute(&storage.pool())
        .await
        .unwrap();

        let granted = scheduler
            .reserve_budget(
                RunId(run_id),
                &BudgetRequest {
                    tokens: 10,
                    cost: 1.0,
                },
            )
            .await
            .unwrap();
        match granted {
            BudgetDecision::Granted(ctx) => {
                assert_eq!(ctx.tokens_reserved, 10);
                assert_eq!(ctx.cost_reserved, 1.0);
                assert!(ctx.tracked);
            }
            _ => panic!("expected granted"),
        }

        let denied = scheduler
            .reserve_budget(
                RunId(run_id),
                &BudgetRequest {
                    tokens: 200,
                    cost: 20.0,
                },
            )
            .await
            .unwrap();
        match denied {
            BudgetDecision::Denied { reason } => {
                assert!(reason.contains("budget exhausted"));
            }
            _ => panic!("expected denied"),
        }

        let unlimited = scheduler
            .reserve_budget(
                RunId(Uuid::new_v4()),
                &BudgetRequest {
                    tokens: 0,
                    cost: 0.0,
                },
            )
            .await
            .unwrap();
        match unlimited {
            BudgetDecision::Unlimited(ctx) => {
                assert_eq!(ctx.tokens_reserved, 0);
                assert!(!ctx.tracked);
            }
            _ => panic!("expected unlimited"),
        }

        drop(container);
    }

    #[tokio::test]
    async fn zero_request_uses_existing_budget() {
        if std::fs::metadata("/var/run/docker.sock").is_err()
            && std::env::var("DOCKER_HOST").is_err()
        {
            eprintln!("Skipping scheduler tests because Docker is unavailable");
            return;
        }

        let (storage, container) = setup_storage().await;
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());
        let scheduler = Scheduler::new(
            Arc::clone(&storage),
            Arc::new(ExecutorRegistry::new()),
            Arc::new(RedactPolicy),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            SchedulerConfig::default(),
        );

        let run_id = Uuid::new_v4();
        sqlx::query(
            "insert into budgets (budget_id, run_id, tokens_limit, tokens_used, cost_limit, cost_used) values (gen_random_uuid(), $1, 50, 20, 5.0, 2.5)"
        )
        .bind(run_id)
        .execute(&storage.pool())
        .await
        .unwrap();

        let decision = scheduler
            .reserve_budget(
                RunId(run_id),
                &BudgetRequest {
                    tokens: 0,
                    cost: 0.0,
                },
            )
            .await
            .unwrap();

        match decision {
            BudgetDecision::Granted(ctx) => {
                assert_eq!(ctx.tokens_used, 20);
                assert_eq!(ctx.token_limit, Some(50));
                assert_eq!(ctx.cost_used, 2.5);
                assert_eq!(ctx.cost_limit, Some(5.0));
            }
            other => panic!("expected granted but got {:?}", other),
        }

        drop(container);
    }

    #[cfg(feature = "insecure-shell")]
    #[tokio::test]
    async fn end_to_end_shell_execution() {
        if std::fs::metadata("/var/run/docker.sock").is_err()
            && std::env::var("DOCKER_HOST").is_err()
        {
            eprintln!("Skipping scheduler tests because Docker is unavailable");
            return;
        }

        std::env::set_var("FLEETFORGE_ALLOWED_TOOLS", "/bin/echo");
        std::env::set_var("FLEETFORGE_ALLOW_SHELL", "1");

        let docker = Cli::default();
        let container = docker.run(Postgres::default());
        let port = container.get_host_port_ipv4(5432);
        let database_url = format!("postgres://postgres:postgres@127.0.0.1:{port}/postgres");
        sleep(Duration::from_millis(250)).await;

        let storage = Arc::new(
            Storage::connect(StorageConfig {
                database_url,
                max_connections: 2,
                connect_timeout: Duration::from_secs(10),
                object_store: ObjectStoreConfig::InMemory,
            })
            .await
            .expect("connect storage"),
        );

        let mut registry = ExecutorRegistry::new();
        registry
            .register(Arc::new(
                ShellExecutor::new().expect("shell executor should be enabled for test"),
            ))
            .expect("register shell executor");
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());

        let scheduler = Arc::new(Scheduler::new(
            Arc::clone(&storage),
            Arc::new(registry),
            crate::policy::AllowAllPolicy::shared(),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            SchedulerConfig {
                batch_size: 2,
                idle_backoff: Duration::from_millis(50),
                active_backoff: Duration::from_millis(10),
                transparency_writer: false,
                ..SchedulerConfig::default()
            },
        ));

        let run_id = Uuid::new_v4();
        let root_id = Uuid::new_v4();
        let child_id = Uuid::new_v4();

        storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({
                        "steps": [
                            {
                                "id": root_id,
                                "type": "tool",
                                "inputs": {"command": ["/bin/echo", "root"]},
                                "policy": {}
                            },
                            {
                                "id": child_id,
                                "type": "tool",
                                "inputs": {"command": ["/bin/echo", "child"]},
                                "policy": {}
                            }
                        ],
                        "edges": [[root_id, child_id]]
                    }),
                    input_ctx: json!({}),
                    seed: 42,
                    idempotency_key: None,
                },
                &[
                    models::NewStep {
                        run_id,
                        step_id: root_id,
                        idx: 0,
                        priority: 1,
                        spec_json: json!({
                            "id": root_id,
                            "type": "tool",
                            "inputs": {"command": ["/bin/echo", "root"]},
                            "policy": {}
                        }),
                        status: models::StepStatus::Queued,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 0,
                        total_dependencies: 0,
                        estimated_tokens: 1,
                        estimated_cost: 0.0,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                    models::NewStep {
                        run_id,
                        step_id: child_id,
                        idx: 1,
                        priority: 1,
                        spec_json: json!({
                            "id": child_id,
                            "type": "tool",
                            "inputs": {"command": ["/bin/echo", "child"]},
                            "policy": {}
                        }),
                        status: models::StepStatus::Blocked,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 1,
                        total_dependencies: 1,
                        estimated_tokens: 1,
                        estimated_cost: 0.0,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                ],
                &[(root_id, child_id)],
            )
            .await
            .expect("seed run");

        let initial = storage.fetch_steps_for_run(run_id).await.unwrap();
        assert_eq!(initial.len(), 2);
        let root = initial.iter().find(|s| s.step_id == root_id).unwrap();
        let child = initial.iter().find(|s| s.step_id == child_id).unwrap();
        assert_eq!(root.status, models::StepStatus::Queued);
        assert_eq!(child.status, models::StepStatus::Blocked);

        let running = Arc::clone(&scheduler);
        let handle = tokio::spawn(async move {
            let _ = running.run().await;
        });

        tokio::time::sleep(Duration::from_secs(1)).await;
        handle.abort();

        let steps = storage.fetch_steps_for_run(run_id).await.unwrap();
        let root = steps.iter().find(|s| s.step_id == root_id).unwrap();
        let child = steps.iter().find(|s| s.step_id == child_id).unwrap();
        assert_eq!(root.status, models::StepStatus::Succeeded);
        assert_eq!(child.status, models::StepStatus::Succeeded);
        assert_eq!(root.attempt, 1);
        assert_eq!(child.attempt, 1);

        let run_status = storage.fetch_run(run_id).await.unwrap().unwrap().status;
        assert_eq!(run_status, models::RunStatus::Succeeded);

        drop(container);
    }

    #[cfg(feature = "insecure-shell")]
    #[tokio::test]
    async fn lease_reaper_requeues_expired() {
        if std::fs::metadata("/var/run/docker.sock").is_err()
            && std::env::var("DOCKER_HOST").is_err()
        {
            eprintln!("Skipping scheduler tests because Docker is unavailable");
            return;
        }

        std::env::set_var("FLEETFORGE_ALLOWED_TOOLS", "/bin/echo");
        std::env::set_var("FLEETFORGE_ALLOW_SHELL", "1");

        let docker = Cli::default();
        let container = docker.run(Postgres::default());
        let port = container.get_host_port_ipv4(5432);
        let database_url = format!("postgres://postgres:postgres@127.0.0.1:{port}/postgres");
        sleep(Duration::from_millis(250)).await;

        let storage = Arc::new(
            Storage::connect(StorageConfig {
                database_url,
                max_connections: 2,
                connect_timeout: Duration::from_secs(10),
                object_store: ObjectStoreConfig::InMemory,
            })
            .await
            .expect("connect storage"),
        );

        let mut registry = ExecutorRegistry::new();
        registry
            .register(Arc::new(
                ShellExecutor::new().expect("shell executor should be enabled for test"),
            ))
            .expect("register shell executor");
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());

        let scheduler = Arc::new(Scheduler::new(
            Arc::clone(&storage),
            Arc::new(registry),
            crate::policy::AllowAllPolicy::shared(),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            SchedulerConfig {
                batch_size: 1,
                idle_backoff: Duration::from_millis(50),
                active_backoff: Duration::from_millis(10),
                reaper_interval: Duration::from_millis(200),
                transparency_writer: false,
                ..SchedulerConfig::default()
            },
        ));

        let run_id = Uuid::new_v4();
        let step_id = Uuid::new_v4();

        storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({
                        "steps": [{
                            "id": step_id,
                            "type": "tool",
                            "inputs": {"command": ["/bin/echo", "late"]},
                            "policy": {}
                        }],
                        "edges": []
                    }),
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: None,
                },
                &[models::NewStep {
                    run_id,
                    step_id,
                    idx: 0,
                    priority: 1,
                    spec_json: json!({
                        "id": step_id,
                        "type": "tool",
                        "inputs": {"command": ["/bin/echo", "late"]},
                        "policy": {}
                    }),
                    status: models::StepStatus::Queued,
                    attempt: 0,
                    input_snapshot: None,
                    provider: None,
                    provider_version: None,
                    pending_dependencies: 0,
                    total_dependencies: 0,
                    estimated_tokens: 0,
                    estimated_cost: 0.0,
                    actual_tokens: None,
                    actual_cost: None,
                    ..models::NewStep::default()
                }],
                &[],
            )
            .await
            .expect("seed run");

        // Manually lease the step with a short expiry to simulate a hung worker.
        sqlx::query(
            "update steps set status = 'leased', leased_by = 'worker-a', lease_expires_at = now() - interval '1 second' where run_id = $1 and step_id = $2"
        )
        .bind(run_id)
        .bind(step_id)
        .execute(storage.pool())
        .await
        .expect("mark leased");

        let running = Arc::clone(&scheduler);
        let handle = tokio::spawn(async move {
            let _ = running.run().await;
        });

        tokio::time::sleep(Duration::from_secs(2)).await;
        handle.abort();

        let step = storage
            .fetch_steps_for_run(run_id)
            .await
            .unwrap()
            .first()
            .cloned()
            .unwrap();
        assert_eq!(step.status, models::StepStatus::Succeeded);
        assert_eq!(step.attempt, 1);

        drop(container);
    }

    #[cfg(feature = "insecure-shell")]
    #[tokio::test]
    async fn budget_denial_records_failure() {
        if std::fs::metadata("/var/run/docker.sock").is_err()
            && std::env::var("DOCKER_HOST").is_err()
        {
            eprintln!("Skipping scheduler tests because Docker is unavailable");
            return;
        }

        std::env::set_var("FLEETFORGE_ALLOWED_TOOLS", "/bin/echo");
        std::env::set_var("FLEETFORGE_ALLOW_SHELL", "1");

        let docker = Cli::default();
        let container = docker.run(Postgres::default());
        let port = container.get_host_port_ipv4(5432);
        let database_url = format!("postgres://postgres:postgres@127.0.0.1:{port}/postgres");
        sleep(Duration::from_millis(250)).await;

        let storage = Arc::new(
            Storage::connect(StorageConfig {
                database_url,
                max_connections: 2,
                connect_timeout: Duration::from_secs(10),
                object_store: ObjectStoreConfig::InMemory,
            })
            .await
            .expect("connect storage"),
        );

        let mut registry = ExecutorRegistry::new();
        registry
            .register(Arc::new(
                ShellExecutor::new().expect("shell executor should be enabled for test"),
            ))
            .expect("register shell executor");
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());

        let scheduler = Arc::new(Scheduler::new(
            Arc::clone(&storage),
            Arc::new(registry),
            crate::policy::AllowAllPolicy::shared(),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            SchedulerConfig::default(),
        ));

        let run_id = Uuid::new_v4();
        let step_id = Uuid::new_v4();

        storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({
                        "steps": [{
                            "id": step_id,
                            "type": "tool",
                            "inputs": {"command": ["/bin/echo", "over"]},
                            "policy": {"budget": {"tokens": 10, "cost": 0.0}}
                        }],
                        "edges": []
                    }),
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: None,
                },
                &[models::NewStep {
                    run_id,
                    step_id,
                    idx: 0,
                    priority: 1,
                    spec_json: json!({
                        "id": step_id,
                        "type": "tool",
                        "inputs": {"command": ["/bin/echo", "over"]},
                        "policy": {"budget": {"tokens": 10, "cost": 0.0}}
                    }),
                    status: models::StepStatus::Queued,
                    attempt: 0,
                    input_snapshot: None,
                    provider: None,
                    provider_version: None,
                    pending_dependencies: 0,
                    total_dependencies: 0,
                    estimated_tokens: 20,
                    estimated_cost: 0.0,
                    actual_tokens: None,
                    actual_cost: None,
                    ..models::NewStep::default()
                }],
                &[],
            )
            .await
            .expect("seed run");

        let running = Arc::clone(&scheduler);
        let handle = tokio::spawn(async move {
            let _ = running.run().await;
        });

        tokio::time::sleep(Duration::from_millis(500)).await;
        handle.abort();

        let steps = storage.fetch_steps_for_run(run_id).await.unwrap();
        let step = steps.first().unwrap();
        assert_eq!(step.status, models::StepStatus::Failed);
        let error = step.error_json.clone().unwrap();
        assert_eq!(error["kind"], json!("budget_exceeded"));

        let events = storage.fetch_unpublished_outbox(10).await.unwrap();
        let failed_event = events.iter().find(|ev| ev.kind == "step_failed").unwrap();
        assert_eq!(
            failed_event.payload["error"]["kind"],
            json!("budget_exceeded")
        );

        drop(container);
    }
}