fleetforge_runtime_api/

lib.rs

//! FleetForge gRPC API surface definitions and server wiring.

mod auth;
mod http;
mod transparency;
mod transparency_writer;

use std::collections::{HashMap, HashSet, VecDeque};
use std::convert::{Infallible, TryFrom};
use std::env;
use std::net::SocketAddr;
use std::sync::Arc;

use ::http::{header, HeaderValue, Method};
use anyhow::{anyhow, Context, Result};
use auth::{AuthError, AuthHandle, Role};
use bytes::Bytes;
use chrono::{DateTime, TimeZone, Utc};
use clickhouse::{Client as ClickhouseClient, Row};
use fleetforge_changeops::{
    BudgetSignal as GateBudgetSignal, ChangeDecisionEffect as GateEffect, ChangeDiff as GateDiff,
    ChangeFollowupRecord, ChangeFollowupRequest as GateFollowupRequest,
    ChangeGateDecision as GateDecision, ChangeGateEngine, ChangeGateRequest as GateRequest,
    CoverageComponent as GateCoverageComponent, CoverageReport as GateCoverageReport,
    DecisionScorecard as GateScorecard, EvalMetric as GateEvalMetric,
    FollowupAction as GateFollowupAction, FollowupOutcome as GateFollowupOutcome,
    MetricDirection as GateMetricDirection,
};
use fleetforge_common::{
    labels::{labels_map_to_value, labels_value_to_map},
    licensing::{feature_allowed, LicensedFeature},
    prost_json::{json_to_prost_struct, json_to_prost_value, prost_struct_to_json},
    validation::{validate_run_spec, validate_step_spec, SchemaValidationError},
};
use fleetforge_contracts::{
    runtime,
    tap::{
        self, subscribe_run_request, EventSeverity, PolicyDecisionEffect, QueryEventSource,
        QueryEventsRequest, QueryEventsResponse, RunEventEnvelope, RunTraceContext,
        SubscribeRunAck, SubscribeRunOpen, SubscribeRunRequest, SubscribeRunResponse,
    },
    RuntimeService, RuntimeServiceClient, RuntimeServiceServer, TapService, TapServiceClient,
    TapServiceServer,
};
use fleetforge_storage::{models, models::NewOutboxEvent, Storage};
use fleetforge_telemetry::audit::Audit;
use fleetforge_trust::{
    Attestation, AttestationVault, InMemoryAttestationVault, ObjectStoreAttestationVault,
    TrustSubject,
};
use futures::Stream;
use http_body::Body as _;
use hyper::service::{make_service_fn, service_fn};
use hyper::{Body, Request as HyperRequest, Server as HyperServer};
use serde::Deserialize;
use serde_json::{json, Map, Number, Value};
use std::pin::Pin;
use tokio::net::TcpListener;
use tokio::sync::mpsc;
use tokio::time::{interval, sleep, Duration, MissedTickBehavior};
use tokio_stream::wrappers::ReceiverStream;
use tonic::transport::Server;
use tonic::{Request, Response, Status};
use tonic_health::pb::health_check_response::ServingStatus;
use tonic_health::server::health_reporter;
use tonic_reflection::server::Builder as ReflectionBuilder;
use tower::util::BoxCloneService;
use tower::{Service, ServiceExt};
use tower_http::cors::CorsLayer;
use tracing::{error, info, instrument, warn};
use transparency::{transparency_log_from_env, TransparencyContext, TransparencyLog};
use transparency_writer::TransparencyWriter;
use uuid::Uuid;

#[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
        )
    }
}
#[derive(Clone)]
struct RuntimeApiService {
    storage: Arc<Storage>,
    auth: Arc<AuthHandle>,
    audit: Arc<Audit>,
    attestation_vault: Arc<dyn AttestationVault>,
    changeops: ChangeGateEngine,
    transparency: Arc<dyn TransparencyLog>,
    transparency_writer_enabled: bool,
    transparency_scope: TransparencyScope,
}

impl RuntimeApiService {
    fn new(
        storage: Arc<Storage>,
        auth: Arc<AuthHandle>,
        audit: Arc<Audit>,
        attestation_vault: Arc<dyn AttestationVault>,
        transparency: Arc<dyn TransparencyLog>,
        transparency_writer_enabled: bool,
        transparency_scope: TransparencyScope,
    ) -> Self {
        Self {
            storage,
            auth,
            audit,
            attestation_vault,
            changeops: ChangeGateEngine::default(),
            transparency,
            transparency_writer_enabled,
            transparency_scope,
        }
    }

    async fn require_role(
        &self,
        metadata: &tonic::metadata::MetadataMap,
        required: Role,
    ) -> Result<Role, Status> {
        self.auth
            .require_grpc(metadata, required)
            .await
            .map_err(|err| match err {
                AuthError::MissingToken | AuthError::InvalidToken => {
                    Status::unauthenticated(err.to_string())
                }
                AuthError::InsufficientRole { .. }
                | AuthError::MissingRoleClaim(_)
                | AuthError::UnsupportedRole { .. } => Status::permission_denied(err.to_string()),
                AuthError::NotConfigured => Status::unauthenticated(err.to_string()),
                AuthError::Backend(_) => Status::internal(err.to_string()),
            })
    }

    fn actor_from_metadata(metadata: &tonic::metadata::MetadataMap, role: Role) -> String {
        metadata
            .get("x-actor")
            .and_then(|value| value.to_str().ok())
            .map(|value| value.to_string())
            .unwrap_or_else(|| format!("role:{role:?}"))
    }
}

fn invalid_argument(message: impl Into<String>) -> Status {
    Status::invalid_argument(message.into())
}

fn internal_error(message: impl Into<String>) -> Status {
    Status::internal(message.into())
}

const QUERY_DEFAULT_PAGE_SIZE: u32 = 100;
const QUERY_MAX_PAGE_SIZE: u32 = 500;
const QUERY_FETCH_MULTIPLIER: u32 = 4;
const TRANSPARENCY_WRITER_FLAG: &str = "FLEETFORGE_TRANSPARENCY_WRITER";
const TRANSPARENCY_WRITER_INTERVAL_ENV: &str = "FLEETFORGE_TRANSPARENCY_WRITER_INTERVAL_SECS";
const TRANSPARENCY_WRITER_INTERVAL_MIN_SECS: u64 = 5;
const TRANSPARENCY_WRITER_DEFAULT_INTERVAL_SECS: u64 = 30;

fn to_timestamp(time: DateTime<Utc>) -> prost_types::Timestamp {
    prost_types::Timestamp {
        seconds: time.timestamp(),
        nanos: time.timestamp_subsec_nanos() as i32,
    }
}

fn value_to_struct(value: &Value) -> Option<prost_types::Struct> {
    if value.is_null() {
        return None;
    }
    match json_to_prost_struct(value) {
        Ok(struct_value) => Some(struct_value),
        Err(err) => {
            warn!(error = %err, "failed to convert JSON value to protobuf Struct");
            None
        }
    }
}

impl RuntimeApiService {
    async fn publish_scitt_entry(
        &self,
        change_id: &str,
        gate_id: Uuid,
        artifact_sha: &str,
        attestation_refs: &[Uuid],
        metadata: &mut Value,
    ) -> Result<(), Status> {
        let signer = fleetforge_trust::scitt_signer()
            .map_err(|err| internal_error(format!("SCITT signer unavailable: {err}")))?;
        let scitt_entry = fleetforge_trust::build_scitt_entry(
            change_id,
            attestation_refs,
            artifact_sha,
            metadata,
            signer.as_ref(),
        )
        .map_err(|err| internal_error(format!("failed to build SCITT entry: {err}")))?;

        let transparency_ctx = TransparencyContext {
            change_id,
            gate_id: &gate_id,
            artifact_sha256: artifact_sha,
            attestation_ids: attestation_refs,
        };
        let record = self
            .transparency
            .append(&scitt_entry, &transparency_ctx)
            .await
            .map_err(|err| internal_error(format!("failed to publish SCITT entry: {err}")))?;

        let scitt_metadata = json!({
            "id": record.entry_id,
            "sha256": hex::encode(record.artifact.sha256),
            "media_type": record.artifact.media_type.clone(),
            "size": record.artifact.size,
            "attestation_ids": attestation_refs.iter().map(|id| id.to_string()).collect::<Vec<_>>(),
            "transparency": record.receipt,
        });

        if let Value::Object(mut meta_obj) = metadata.take() {
            meta_obj.insert("scitt_entry".to_string(), scitt_metadata);
            *metadata = Value::Object(meta_obj);
        } else {
            *metadata = json!({ "scitt_entry": scitt_metadata });
        }
        Ok(())
    }
}

fn struct_to_value(data: Option<prost_types::Struct>) -> Value {
    data.as_ref()
        .map(prost_struct_to_json)
        .unwrap_or(Value::Null)
}

fn attestation_to_proto(att: Attestation) -> Result<runtime::Attestation, Status> {
    let subject = att
        .subject
        .as_ref()
        .map(attestation_subject_to_proto)
        .transpose()?;

    let claims_value = Value::Object(att.claims.into_iter().collect());
    let claims_struct = json_to_prost_struct(&claims_value)
        .map_err(|err| internal_error(format!("failed to serialise attestation claims: {err}")))?;

    Ok(runtime::Attestation {
        attestation_id: att.id.to_string(),
        issued_at: Some(to_timestamp(att.issued_at)),
        subject,
        artifact_hash: att.artifact_hash.map(|value| value.into()),
        claims: Some(claims_struct),
        signature: att.signature.map(|value| value.into()),
    })
}

fn attestation_subject_to_proto(
    subject: &TrustSubject,
) -> Result<runtime::AttestationSubject, Status> {
    use runtime::attestation_subject::Kind;
    let kind = match subject {
        TrustSubject::Run { run_id } => Kind::Run(runtime::AttestationRunSubject {
            run_id: run_id.to_string(),
        }),
        TrustSubject::Step { run_id, step_id } => Kind::Step(runtime::AttestationStepSubject {
            run_id: run_id.to_string(),
            step_id: step_id.to_string(),
        }),
        TrustSubject::Tool { name } => {
            Kind::Tool(runtime::AttestationToolSubject { name: name.clone() })
        }
        TrustSubject::Artifact { uri } => {
            Kind::Artifact(runtime::AttestationArtifactSubject { uri: uri.clone() })
        }
        TrustSubject::CapabilityToken { token_id } => {
            Kind::CapabilityToken(runtime::AttestationCapabilitySubject {
                token_id: token_id.to_string(),
            })
        }
        TrustSubject::Custom { label } => Kind::Custom(runtime::AttestationCustomSubject {
            label: label.clone(),
        }),
    };
    Ok(runtime::AttestationSubject { kind: Some(kind) })
}

fn gate_effect_to_proto(effect: GateEffect) -> runtime::ChangeDecisionEffect {
    match effect {
        GateEffect::Allow => runtime::ChangeDecisionEffect::Allow,
        GateEffect::FollowUp => runtime::ChangeDecisionEffect::FollowUp,
        GateEffect::Deny => runtime::ChangeDecisionEffect::Deny,
    }
}

fn gate_effect_from_str(value: &str) -> GateEffect {
    match value {
        "allow" => GateEffect::Allow,
        "deny" => GateEffect::Deny,
        "follow_up" => GateEffect::FollowUp,
        other => {
            warn!(effect = %other, "unknown change gate effect stored; defaulting to follow_up");
            GateEffect::FollowUp
        }
    }
}

fn gate_followup_action_to_proto(action: &GateFollowupAction) -> runtime::FollowupAction {
    runtime::FollowupAction {
        slug: action.slug.clone(),
        description: action.description.clone(),
        recommended: action.recommended,
    }
}

fn gate_scorecard_to_proto(scorecard: &GateScorecard) -> Option<runtime::DecisionScorecard> {
    Some(runtime::DecisionScorecard {
        novelty: Some(runtime::NoveltySummary {
            max_novelty: scorecard.novelty.max_novelty,
            avg_novelty: scorecard.novelty.avg_novelty,
            high_risk_paths: scorecard.novelty.high_risk_paths.clone(),
        }),
        coverage: Some(runtime::CoverageSummary {
            overall_ratio: scorecard.coverage.overall_ratio,
            components_below_threshold: scorecard.coverage.components_below_threshold.clone(),
        }),
        evals: Some(runtime::EvalSummary {
            failing_metrics: scorecard.evals.failing_metrics.clone(),
            attention_metrics: scorecard.evals.attention_metrics.clone(),
            score_per_token: scorecard.evals.score_per_token,
        }),
        budgets: Some(runtime::BudgetSummary {
            breaches: scorecard.budgets.breaches.clone(),
            near_limits: scorecard.budgets.near_limits.clone(),
        }),
        telemetry: value_to_struct(&scorecard.telemetry),
    })
}

fn proto_followup_outcome(outcome: runtime::FollowupOutcome) -> GateFollowupOutcome {
    match outcome {
        runtime::FollowupOutcome::Approved => GateFollowupOutcome::Approved,
        runtime::FollowupOutcome::NeedsChanges => GateFollowupOutcome::NeedsChanges,
        runtime::FollowupOutcome::Deferred => GateFollowupOutcome::Deferred,
        runtime::FollowupOutcome::Unspecified => GateFollowupOutcome::Approved,
    }
}

fn followup_outcome_to_proto(outcome: GateFollowupOutcome) -> runtime::FollowupOutcome {
    match outcome {
        GateFollowupOutcome::Approved => runtime::FollowupOutcome::Approved,
        GateFollowupOutcome::NeedsChanges => runtime::FollowupOutcome::NeedsChanges,
        GateFollowupOutcome::Deferred => runtime::FollowupOutcome::Deferred,
    }
}

fn parse_reasons(value: &Value) -> Vec<String> {
    serde_json::from_value::<Vec<String>>(value.clone()).unwrap_or_default()
}

fn parse_followup_actions(value: &Value) -> Vec<GateFollowupAction> {
    serde_json::from_value::<Vec<GateFollowupAction>>(value.clone()).unwrap_or_default()
}

fn parse_scorecard(value: &Value) -> GateScorecard {
    serde_json::from_value::<GateScorecard>(value.clone()).unwrap_or_default()
}

fn gate_followup_record_to_proto(
    record: &models::ChangeGateFollowup,
) -> runtime::ChangeGateFollowup {
    let outcome =
        GateFollowupOutcome::from_str(&record.outcome).unwrap_or(GateFollowupOutcome::Approved);
    let details_struct = value_to_struct(&record.details);

    runtime::ChangeGateFollowup {
        followup_id: record.followup_id.to_string(),
        gate_id: record.gate_id.to_string(),
        actor: record.actor.clone(),
        outcome: followup_outcome_to_proto(outcome) as i32,
        note: record.note.clone(),
        details: details_struct,
        recorded_at: Some(to_timestamp(record.recorded_at)),
    }
}

fn build_proto_change_gate(
    gate_id: Uuid,
    change_id: &str,
    revision: Option<&str>,
    decision: &GateDecision,
    telemetry: &Value,
    acknowledgements: &[models::ChangeGateFollowup],
) -> Result<runtime::ChangeGateDecision, Status> {
    let effect = gate_effect_to_proto(decision.effect) as i32;
    let scorecard = gate_scorecard_to_proto(&decision.scorecard);
    let metadata_struct = value_to_struct(&decision.metadata);
    let telemetry_struct = value_to_struct(telemetry);
    let ack_proto = acknowledgements
        .iter()
        .map(gate_followup_record_to_proto)
        .collect();

    Ok(runtime::ChangeGateDecision {
        gate_id: gate_id.to_string(),
        change_id: change_id.to_string(),
        revision: revision.unwrap_or_default().to_string(),
        effect,
        reasons: decision.reasons.clone(),
        followups: decision
            .followups
            .iter()
            .map(gate_followup_action_to_proto)
            .collect(),
        scorecard,
        decided_at: Some(to_timestamp(decision.decided_at)),
        metadata: metadata_struct,
        telemetry: telemetry_struct,
        acknowledgements: ack_proto,
    })
}

fn attestation_ids_from_value(value: &Value) -> Vec<Uuid> {
    fn recurse(value: &Value, acc: &mut HashSet<Uuid>) {
        match value {
            Value::Object(map) => {
                if let Some(ids) = map.get("attestation_ids").and_then(Value::as_array) {
                    for id in ids {
                        if let Some(s) = id.as_str() {
                            if let Ok(uuid) = Uuid::parse_str(s) {
                                acc.insert(uuid);
                            }
                        }
                    }
                }
                for v in map.values() {
                    recurse(v, acc);
                }
            }
            Value::Array(items) => {
                for item in items {
                    recurse(item, acc);
                }
            }
            _ => {}
        }
    }
    let mut set = HashSet::new();
    recurse(value, &mut set);
    set.into_iter().collect()
}

#[derive(Clone)]
struct TapApiService {
    storage: Arc<Storage>,
    auth: Arc<AuthHandle>,
    clickhouse: Option<ClickhouseEventQuery>,
}

impl TapApiService {
    fn new(storage: Arc<Storage>, auth: Arc<AuthHandle>) -> Self {
        let clickhouse = ClickhouseEventQuery::from_env().ok();
        Self {
            storage,
            auth,
            clickhouse,
        }
    }

    async fn require_role(
        &self,
        metadata: &tonic::metadata::MetadataMap,
        required: Role,
    ) -> Result<Role, Status> {
        self.auth
            .require_grpc(metadata, required)
            .await
            .map_err(|err| match err {
                AuthError::MissingToken | AuthError::InvalidToken => {
                    Status::unauthenticated(err.to_string())
                }
                AuthError::InsufficientRole { .. }
                | AuthError::MissingRoleClaim(_)
                | AuthError::UnsupportedRole { .. } => Status::permission_denied(err.to_string()),
                AuthError::NotConfigured => Status::unauthenticated(err.to_string()),
                AuthError::Backend(_) => Status::internal(err.to_string()),
            })
    }
}

#[tonic::async_trait]
impl TapService for TapApiService {
    type SubscribeRunStream =
        Pin<Box<dyn Stream<Item = Result<SubscribeRunResponse, Status>> + Send>>;

    async fn subscribe_run(
        &self,
        request: Request<tonic::Streaming<SubscribeRunRequest>>,
    ) -> Result<Response<Self::SubscribeRunStream>, Status> {
        let metadata = request.metadata().clone();
        self.require_role(&metadata, Role::Reader).await?;

        let mut inbound = request.into_inner();
        let first = inbound
            .message()
            .await
            .map_err(|status| internal_error(status.to_string()))?
            .ok_or_else(|| invalid_argument("missing subscribe request"))?;

        let open = match first.request {
            Some(subscribe_run_request::Request::Open(open)) => open,
            Some(subscribe_run_request::Request::Ack(_)) => {
                return Err(invalid_argument(
                    "first SubscribeRun message must be `open`",
                ))
            }
            None => return Err(invalid_argument("first SubscribeRun message must be set")),
        };

        let mut config = SubscriptionConfig::try_from_open(open)?;
        if let Some(cursor) = config.cursor.clone() {
            if config.since_offset == 0 {
                match self.storage.fetch_tap_offset(config.run_id, &cursor).await {
                    Ok(Some(stored)) => {
                        config.since_offset = stored;
                    }
                    Ok(None) => {}
                    Err(err) => {
                        error!(
                            run = %config.run_id,
                            cursor = %cursor,
                            error = %err,
                            "failed to load stored tap cursor"
                        );
                        return Err(internal_error("failed to load stored tap cursor"));
                    }
                }
            }
        }
        let (tx, rx) =
            mpsc::channel::<Result<SubscribeRunResponse, Status>>(config.max_inflight * 2);

        let storage = Arc::clone(&self.storage);
        tokio::spawn(async move {
            if let Err(err) = run_subscription(storage, config, inbound, tx.clone()).await {
                let _ = tx.send(Err(err)).await;
            }
        });

        Ok(Response::new(
            Box::pin(ReceiverStream::new(rx)) as Self::SubscribeRunStream
        ))
    }

    async fn query_events(
        &self,
        request: Request<QueryEventsRequest>,
    ) -> Result<Response<QueryEventsResponse>, Status> {
        let metadata = request.metadata().clone();
        self.require_role(&metadata, Role::Reader).await?;

        let clickhouse = self.clickhouse.as_ref().ok_or_else(|| {
            Status::failed_precondition("QueryEvents requires ClickHouse configuration")
        })?;

        let inner = request.into_inner();
        let filters = QueryFiltersOwned::from_request(&inner)?;

        let requested_page = if inner.page_size == 0 {
            QUERY_DEFAULT_PAGE_SIZE
        } else {
            inner.page_size
        };
        let page_size = requested_page.clamp(1, QUERY_MAX_PAGE_SIZE);
        let offset = parse_page_token(inner.page_token.as_str())?;

        let QueryBatch {
            events,
            next_offset,
        } = clickhouse
            .fetch(&filters, offset, page_size)
            .await
            .map_err(|err| {
                error!(error = %err, "failed to query ClickHouse events");
                internal_error("failed to query ClickHouse events")
            })?;

        let next_page_token = next_offset
            .map(|value| value.to_string())
            .unwrap_or_default();

        Ok(Response::new(QueryEventsResponse {
            events,
            next_page_token,
        }))
    }
}

struct SubscriptionConfig {
    run_id: Uuid,
    step_ids: Option<HashSet<Uuid>>,
    kinds: Option<HashSet<String>>,
    severities: Option<HashSet<EventSeverity>>,
    tags: Option<HashSet<String>>,
    cursor: Option<String>,
    since_offset: i64,
    batch_size: i64,
    max_inflight: usize,
    heartbeat: Duration,
}

impl SubscriptionConfig {
    fn try_from_open(open: SubscribeRunOpen) -> Result<Self, Status> {
        let SubscribeRunOpen {
            run_ids,
            step_ids,
            kinds,
            severities,
            tags,
            since_offset,
            batch_size,
            max_inflight,
            heartbeat_interval_ms,
            cursor,
            ..
        } = open;

        if run_ids.is_empty() {
            return Err(invalid_argument(
                "subscribe_run requires at least one run_id",
            ));
        }
        if run_ids.len() > 1 {
            return Err(invalid_argument(
                "subscribe_run currently supports a single run_id",
            ));
        }
        let run_id = Uuid::parse_str(&run_ids[0])
            .map_err(|err| invalid_argument(format!("invalid run_id '{}' : {err}", run_ids[0])))?;

        let step_ids = if step_ids.is_empty() {
            None
        } else {
            let mut ids = HashSet::with_capacity(step_ids.len());
            for raw in step_ids {
                let parsed = Uuid::parse_str(&raw)
                    .map_err(|err| invalid_argument(format!("invalid step_id '{raw}': {err}")))?;
                ids.insert(parsed);
            }
            Some(ids)
        };

        let kinds = if kinds.is_empty() {
            None
        } else {
            Some(
                kinds
                    .into_iter()
                    .map(|kind| kind.to_ascii_lowercase())
                    .collect::<HashSet<_>>(),
            )
        };

        let severities = if severities.is_empty() {
            None
        } else {
            let mut set = HashSet::with_capacity(severities.len());
            for raw in severities {
                let sev = EventSeverity::try_from(raw)
                    .map_err(|_| invalid_argument(format!("unknown severity enum value {raw}")))?;
                set.insert(sev);
            }
            Some(set)
        };

        let tags = if tags.is_empty() {
            None
        } else {
            Some(
                tags.into_iter()
                    .map(|tag| tag.to_ascii_lowercase())
                    .collect::<HashSet<_>>(),
            )
        };

        if since_offset < 0 {
            return Err(invalid_argument("since_offset must be >= 0"));
        }

        let batch_raw = if batch_size == 0 { 256 } else { batch_size };
        let batch_size = batch_raw.max(1).min(2048) as i64;

        let inflight_raw = if max_inflight == 0 { 128 } else { max_inflight };
        let max_inflight = inflight_raw.max(1).min(4096) as usize;

        let heartbeat_ms = if heartbeat_interval_ms == 0 {
            1000u64
        } else {
            heartbeat_interval_ms as u64
        }
        .clamp(250, 60_000);

        let cursor = {
            let trimmed = cursor.trim();
            if trimmed.is_empty() {
                None
            } else {
                Some(trimmed.to_string())
            }
        };

        Ok(Self {
            run_id,
            step_ids,
            kinds,
            severities,
            tags,
            cursor,
            since_offset,
            batch_size,
            max_inflight,
            heartbeat: Duration::from_millis(heartbeat_ms),
        })
    }
}

async fn run_subscription(
    storage: Arc<Storage>,
    config: SubscriptionConfig,
    mut inbound: tonic::Streaming<SubscribeRunRequest>,
    tx: mpsc::Sender<Result<SubscribeRunResponse, Status>>,
) -> Result<(), Status> {
    let mut last_seen = config.since_offset;
    let mut last_ack = config.since_offset;
    let mut inflight: VecDeque<i64> = VecDeque::new();
    let mut ticker = interval(config.heartbeat);
    ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);

    loop {
        while inflight.len() < config.max_inflight {
            let events = storage
                .fetch_outbox_for_run_since(config.run_id, last_seen, config.batch_size)
                .await
                .map_err(|err| internal_error(format!("failed to fetch outbox events: {err}")))?;

            if events.is_empty() {
                break;
            }

            last_seen = events.last().map(|event| event.id).unwrap_or(last_seen);

            for event in events {
                last_seen = event.id;
                let severity = derive_severity(&event.kind, &event.payload);
                let tags = derive_tags(&event.kind, event.step_id, &event.payload, severity);
                if !passes_filters(&config, &event, severity, &tags) {
                    continue;
                }

                let response = build_response(&config, &event, severity, &tags, last_seen)?;
                if tx.send(Ok(response)).await.is_err() {
                    return Ok(());
                }
                inflight.push_back(event.id);
            }
        }

        tokio::select! {
            maybe_req = inbound.message() => {
                match maybe_req {
                    Ok(Some(SubscribeRunRequest { request: Some(subscribe_run_request::Request::Ack(ack)) })) => {
                        handle_ack(&mut inflight, &mut last_ack, ack.offset);
                        let ack_cursor = if !ack.cursor.trim().is_empty() {
                            Some(ack.cursor)
                        } else {
                            config.cursor.clone()
                        };
                        if let Some(cursor_id) = ack_cursor {
                            if ack.offset < 0 {
                                return Err(invalid_argument("ack offset must be >= 0"));
                            }
                            if let Err(err) = storage
                                .upsert_tap_offset(config.run_id, &cursor_id, ack.offset)
                                .await
                            {
                                error!(
                                    run = %config.run_id,
                                    cursor = %cursor_id,
                                    error = %err,
                                    "failed to persist tap cursor offset"
                                );
                                return Err(internal_error("failed to persist tap cursor offset"));
                            }
                        }
                    }
                    Ok(Some(SubscribeRunRequest { request: Some(subscribe_run_request::Request::Open(_)) })) => {
                        return Err(invalid_argument("subscribe_run received multiple open messages"));
                    }
                    Ok(Some(SubscribeRunRequest { request: None })) => {}
                    Ok(None) => {
                        return Ok(());
                    }
                    Err(status) => {
                        return Err(internal_error(status.to_string()));
                    }
                }
            }
            _ = ticker.tick() => {
                let heartbeat = SubscribeRunResponse {
                    offset: last_ack,
                    watermark: last_seen,
                    heartbeat: true,
                    event: None,
                };
                if tx.send(Ok(heartbeat)).await.is_err() {
                    return Ok(());
                }
            }
        }
    }
}

#[derive(Debug, Clone)]
struct QueryFiltersOwned {
    source: QueryEventSource,
    run_ids: Vec<Uuid>,
    step_ids: Vec<Uuid>,
    workspace_ids: Vec<String>,
    kinds: Vec<String>,
    policy_effects: Vec<PolicyDecisionEffect>,
    severities: Vec<EventSeverity>,
    start_time: Option<DateTime<Utc>>,
    end_time: Option<DateTime<Utc>>,
    ascending: bool,
}

impl QueryFiltersOwned {
    fn from_request(request: &QueryEventsRequest) -> Result<Self, Status> {
        let mut run_ids = Vec::with_capacity(request.run_ids.len());
        for raw in &request.run_ids {
            let parsed = Uuid::parse_str(raw)
                .map_err(|err| invalid_argument(format!("invalid run_id '{raw}': {err}")))?;
            run_ids.push(parsed);
        }

        let mut step_ids = Vec::with_capacity(request.step_ids.len());
        for raw in &request.step_ids {
            let parsed = Uuid::parse_str(raw)
                .map_err(|err| invalid_argument(format!("invalid step_id '{raw}': {err}")))?;
            step_ids.push(parsed);
        }

        let source = QueryEventSource::try_from(request.source).unwrap_or(QueryEventSource::Run);

        let start_time = request
            .start_time
            .as_ref()
            .map(timestamp_to_datetime)
            .transpose()?;
        let end_time = request
            .end_time
            .as_ref()
            .map(timestamp_to_datetime)
            .transpose()?;

        let kinds = request
            .kinds
            .iter()
            .filter(|kind| !kind.is_empty())
            .map(|kind| kind.to_ascii_lowercase())
            .collect::<Vec<_>>();

        let workspace_ids = request
            .workspace_ids
            .iter()
            .filter(|id| !id.is_empty())
            .map(|id| id.to_ascii_lowercase())
            .collect::<Vec<_>>();

        let mut policy_effects = Vec::with_capacity(request.policy_effects.len());
        for raw in &request.policy_effects {
            if let Ok(effect) = PolicyDecisionEffect::try_from(*raw) {
                if effect != PolicyDecisionEffect::Unspecified {
                    policy_effects.push(effect);
                }
            }
        }

        let mut severities = Vec::with_capacity(request.severities.len());
        for raw in &request.severities {
            if let Ok(severity) = EventSeverity::try_from(*raw) {
                if severity != EventSeverity::Unspecified {
                    severities.push(severity);
                }
            }
        }

        Ok(Self {
            source,
            run_ids,
            step_ids,
            workspace_ids,
            kinds,
            policy_effects,
            severities,
            start_time,
            end_time,
            ascending: request.ascending,
        })
    }
}

struct QueryBatch {
    events: Vec<RunEventEnvelope>,
    next_offset: Option<u64>,
}

#[derive(Clone)]
struct ClickhouseEventQuery {
    client: ClickhouseClient,
    run_table: String,
    step_table: String,
}

#[derive(Debug, Deserialize, Row)]
struct RunEventRow {
    run_id: Uuid,
    created_at: DateTime<Utc>,
    status: String,
    attributes: Value,
}

#[derive(Debug, Deserialize, Row)]
struct StepEventRow {
    run_id: Uuid,
    step_id: Uuid,
    status: String,
    attempt: i32,
    recorded_at: DateTime<Utc>,
    attributes: Value,
}

impl ClickhouseEventQuery {
    fn from_env() -> Result<Self> {
        let url = env::var("CLICKHOUSE_DSN")
            .context("CLICKHOUSE_DSN is not configured for Tap QueryEvents")?;
        let database = env::var("CLICKHOUSE_DATABASE").unwrap_or_else(|_| "telemetry".to_string());
        let run_table =
            env::var("CLICKHOUSE_RUN_TABLE").unwrap_or_else(|_| "run_events".to_string());
        let step_table =
            env::var("CLICKHOUSE_STEP_TABLE").unwrap_or_else(|_| "step_events".to_string());

        let client = ClickhouseClient::default()
            .with_url(url)
            .with_database(database);

        Ok(Self {
            client,
            run_table,
            step_table,
        })
    }

    async fn fetch(
        &self,
        filters: &QueryFiltersOwned,
        offset: u64,
        limit: u32,
    ) -> Result<QueryBatch> {
        match filters.source {
            QueryEventSource::Step => self.query_step_events(filters, offset, limit).await,
            _ => self.query_run_events(filters, offset, limit).await,
        }
    }

    async fn query_run_events(
        &self,
        filters: &QueryFiltersOwned,
        offset: u64,
        limit: u32,
    ) -> Result<QueryBatch> {
        let chunk = std::cmp::max(limit, 1)
            .saturating_mul(QUERY_FETCH_MULTIPLIER)
            .max(limit.max(1));
        let mut collected = Vec::with_capacity(limit as usize);
        let mut scan_offset = offset;
        let mut next_offset = None;

        while collected.len() < limit as usize {
            let rows = self.fetch_run_rows(filters, scan_offset, chunk).await?;
            if rows.is_empty() {
                break;
            }

            let mut consumed = 0usize;
            for row in rows {
                let raw_offset = scan_offset + consumed as u64;
                consumed += 1;
                let payload = normalize_payload(row.attributes);
                let kind = format!("run_{}", row.status.to_ascii_lowercase());
                let severity = derive_severity(&kind, &payload);
                if !matches_filters(filters, &payload, &kind, severity, None) {
                    continue;
                }
                let envelope = make_envelope(
                    row.run_id,
                    None,
                    row.created_at,
                    &payload,
                    &kind,
                    severity,
                    raw_offset,
                )?;
                collected.push(envelope);
                if collected.len() >= limit as usize {
                    next_offset = Some(raw_offset.saturating_add(1));
                    break;
                }
            }

            scan_offset = scan_offset.saturating_add(consumed as u64);
            if consumed < chunk as usize {
                break;
            }
        }

        Ok(QueryBatch {
            events: collected,
            next_offset,
        })
    }

    async fn query_step_events(
        &self,
        filters: &QueryFiltersOwned,
        offset: u64,
        limit: u32,
    ) -> Result<QueryBatch> {
        let chunk = std::cmp::max(limit, 1)
            .saturating_mul(QUERY_FETCH_MULTIPLIER)
            .max(limit.max(1));
        let mut collected = Vec::with_capacity(limit as usize);
        let mut scan_offset = offset;
        let mut next_offset = None;

        while collected.len() < limit as usize {
            let rows = self.fetch_step_rows(filters, scan_offset, chunk).await?;
            if rows.is_empty() {
                break;
            }

            let mut consumed = 0usize;
            for row in rows {
                let raw_offset = scan_offset + consumed as u64;
                consumed += 1;
                let payload = normalize_payload(row.attributes);
                let kind = format!("step_{}", row.status.to_ascii_lowercase());
                let severity = derive_severity(&kind, &payload);
                if !matches_filters(filters, &payload, &kind, severity, Some(row.step_id)) {
                    continue;
                }
                let envelope = make_envelope(
                    row.run_id,
                    Some(row.step_id),
                    row.recorded_at,
                    &payload,
                    &kind,
                    severity,
                    raw_offset,
                )?;
                collected.push(envelope);
                if collected.len() >= limit as usize {
                    next_offset = Some(raw_offset.saturating_add(1));
                    break;
                }
            }

            scan_offset = scan_offset.saturating_add(consumed as u64);
            if consumed < chunk as usize {
                break;
            }
        }

        Ok(QueryBatch {
            events: collected,
            next_offset,
        })
    }

    async fn fetch_run_rows(
        &self,
        filters: &QueryFiltersOwned,
        offset: u64,
        limit: u32,
    ) -> Result<Vec<RunEventRow>> {
        let mut where_clauses = Vec::new();
        if !filters.run_ids.is_empty() {
            where_clauses.push(format!(
                "run_id IN ({})",
                format_uuid_list(&filters.run_ids)
            ));
        }
        if let Some(start) = filters.start_time {
            where_clauses.push(format!(
                "created_at >= toDateTime64('{}', 3, 'UTC')",
                start.format("%Y-%m-%d %H:%M:%S%.3f")
            ));
        }
        if let Some(end) = filters.end_time {
            where_clauses.push(format!(
                "created_at <= toDateTime64('{}', 3, 'UTC')",
                end.format("%Y-%m-%d %H:%M:%S%.3f")
            ));
        }

        let mut sql = format!(
            "SELECT run_id, created_at, status, attributes FROM {}",
            self.run_table
        );
        if !where_clauses.is_empty() {
            sql.push_str(" WHERE ");
            sql.push_str(&where_clauses.join(" AND "));
        }
        sql.push_str(" ORDER BY created_at ");
        sql.push_str(if filters.ascending { "ASC" } else { "DESC" });
        sql.push_str(", run_id ");
        sql.push_str(if filters.ascending { "ASC" } else { "DESC" });
        sql.push_str(&format!(" LIMIT {} OFFSET {}", limit, offset));

        let rows = self
            .client
            .query(&sql)
            .fetch_all::<RunEventRow>()
            .await
            .context("failed to fetch run events from ClickHouse")?;
        Ok(rows)
    }

    async fn fetch_step_rows(
        &self,
        filters: &QueryFiltersOwned,
        offset: u64,
        limit: u32,
    ) -> Result<Vec<StepEventRow>> {
        let mut where_clauses = Vec::new();
        if !filters.run_ids.is_empty() {
            where_clauses.push(format!(
                "run_id IN ({})",
                format_uuid_list(&filters.run_ids)
            ));
        }
        if !filters.step_ids.is_empty() {
            where_clauses.push(format!(
                "step_id IN ({})",
                format_uuid_list(&filters.step_ids)
            ));
        }

        let time_expr = "coalesce(finished_at, started_at)";
        if let Some(start) = filters.start_time {
            where_clauses.push(format!(
                "{time_expr} >= toDateTime64('{}', 3, 'UTC')",
                start.format("%Y-%m-%d %H:%M:%S%.3f")
            ));
        }
        if let Some(end) = filters.end_time {
            where_clauses.push(format!(
                "{time_expr} <= toDateTime64('{}', 3, 'UTC')",
                end.format("%Y-%m-%d %H:%M:%S%.3f")
            ));
        }

        let mut sql = format!(
            "SELECT run_id, step_id, status, attempt, {time_expr} AS recorded_at, attributes FROM {}",
            self.step_table
        );
        if !where_clauses.is_empty() {
            sql.push_str(" WHERE ");
            sql.push_str(&where_clauses.join(" AND "));
        }
        sql.push_str(&format!(" ORDER BY {time_expr} "));
        sql.push_str(if filters.ascending { "ASC" } else { "DESC" });
        sql.push_str(", step_id ");
        sql.push_str(if filters.ascending { "ASC" } else { "DESC" });
        sql.push_str(&format!(" LIMIT {} OFFSET {}", limit, offset));

        let rows = self
            .client
            .query(&sql)
            .fetch_all::<StepEventRow>()
            .await
            .context("failed to fetch step events from ClickHouse")?;
        Ok(rows)
    }
}

fn parse_page_token(raw: &str) -> Result<u64, Status> {
    if raw.trim().is_empty() {
        return Ok(0);
    }
    raw.parse::<u64>()
        .map_err(|_| invalid_argument("page_token must be a positive integer"))
}

fn timestamp_to_datetime(ts: &prost_types::Timestamp) -> Result<DateTime<Utc>, Status> {
    Utc.timestamp_opt(ts.seconds, ts.nanos as u32)
        .single()
        .ok_or_else(|| invalid_argument("timestamp out of range"))
}

fn normalize_payload(value: Value) -> Value {
    match value {
        Value::Object(_) => value,
        Value::Null => Value::Object(Map::new()),
        other => {
            let mut map = Map::new();
            map.insert("value".to_string(), other);
            Value::Object(map)
        }
    }
}

fn matches_filters(
    filters: &QueryFiltersOwned,
    payload: &Value,
    kind: &str,
    severity: EventSeverity,
    step_id: Option<Uuid>,
) -> bool {
    if !filters.kinds.is_empty() {
        let kind_value = kind.to_ascii_lowercase();
        if !filters.kinds.iter().any(|expected| expected == &kind_value) {
            return false;
        }
    }

    if !filters.workspace_ids.is_empty() {
        let workspace = json_get_string(payload, &["trace", "run", "workspace_id"])
            .map(|value| value.to_ascii_lowercase());
        let Some(workspace) = workspace else {
            return false;
        };
        if !filters
            .workspace_ids
            .iter()
            .any(|expected| expected == &workspace)
        {
            return false;
        }
    }

    if !filters.policy_effects.is_empty() {
        let effect = payload
            .get("effect")
            .and_then(Value::as_str)
            .map(|value| value.to_ascii_lowercase());
        if let Some(effect) = effect {
            if !filters
                .policy_effects
                .iter()
                .any(|expected| policy_effect_label(*expected) == effect)
            {
                return false;
            }
        } else {
            return false;
        }
    }

    if !filters.severities.is_empty() && !filters.severities.contains(&severity) {
        return false;
    }

    if filters.source == QueryEventSource::Step && !filters.step_ids.is_empty() {
        if let Some(step_id) = step_id {
            if !filters.step_ids.contains(&step_id) {
                return false;
            }
        } else {
            return false;
        }
    }

    true
}

fn policy_effect_label(effect: PolicyDecisionEffect) -> &'static str {
    match effect {
        PolicyDecisionEffect::Allow => "allow",
        PolicyDecisionEffect::Deny => "deny",
        PolicyDecisionEffect::Escalate => "escalate",
        PolicyDecisionEffect::Modify => "modify",
        _ => "",
    }
}

fn json_get_string<'a>(value: &'a Value, path: &[&str]) -> Option<&'a str> {
    let mut current = value;
    for segment in path {
        current = current.get(*segment)?;
    }
    current.as_str()
}

fn format_uuid_list(values: &[Uuid]) -> String {
    values
        .iter()
        .map(|value| format!("'{}'", value))
        .collect::<Vec<_>>()
        .join(", ")
}

fn make_envelope(
    run_id: Uuid,
    step_id: Option<Uuid>,
    recorded_at: DateTime<Utc>,
    payload: &Value,
    kind: &str,
    severity: EventSeverity,
    raw_offset: u64,
) -> Result<RunEventEnvelope> {
    let trace = build_trace_context(run_id, step_id, payload);
    let tags = derive_tags(kind, step_id, payload, severity);
    let raw_struct = json_to_prost_struct(payload)
        .map_err(|err| anyhow!("failed to encode ClickHouse payload into Struct: {err}"))?;
    let offset =
        i64::try_from(raw_offset).map_err(|_| anyhow!("query offset exceeds i64 range"))?;

    Ok(RunEventEnvelope {
        trace: Some(trace),
        recorded_at: Some(to_timestamp(recorded_at)),
        offset,
        severity: severity as i32,
        tags,
        raw: Some(raw_struct),
        body: None,
    })
}

fn handle_ack(inflight: &mut VecDeque<i64>, last_ack: &mut i64, ack_offset: i64) {
    if ack_offset <= *last_ack {
        return;
    }
    *last_ack = ack_offset;
    while let Some(front) = inflight.front() {
        if *front <= ack_offset {
            inflight.pop_front();
        } else {
            break;
        }
    }
}

fn passes_filters(
    config: &SubscriptionConfig,
    event: &models::OutboxEvent,
    severity: EventSeverity,
    tags: &[String],
) -> bool {
    if let Some(ref kinds) = config.kinds {
        if !kinds.contains(&event.kind.to_ascii_lowercase()) {
            return false;
        }
    }

    if let Some(ref severities) = config.severities {
        if !severities.contains(&severity) {
            return false;
        }
    }

    if let Some(ref step_filter) = config.step_ids {
        match event.step_id {
            Some(step_id) if step_filter.contains(&step_id) => {}
            _ => return false,
        }
    }

    if let Some(ref tag_filter) = config.tags {
        if !tags
            .iter()
            .any(|tag| tag_filter.contains(&tag.to_ascii_lowercase()))
        {
            return false;
        }
    }

    true
}

fn derive_severity(kind: &str, payload: &Value) -> EventSeverity {
    let kind_lower = kind.to_ascii_lowercase();
    if kind_lower.contains("failed") || kind_lower.contains("error") {
        EventSeverity::Error
    } else if kind_lower.contains("denied") || kind_lower.contains("skipped") {
        EventSeverity::Warn
    } else if kind_lower.contains("debug") {
        EventSeverity::Debug
    } else if payload
        .get("severity")
        .and_then(Value::as_str)
        .map(|value| value.eq_ignore_ascii_case("warn"))
        .unwrap_or(false)
    {
        EventSeverity::Warn
    } else {
        EventSeverity::Info
    }
}

fn severity_label(severity: EventSeverity) -> &'static str {
    match severity {
        EventSeverity::Trace => "trace",
        EventSeverity::Debug => "debug",
        EventSeverity::Info => "info",
        EventSeverity::Warn => "warn",
        EventSeverity::Error => "error",
        EventSeverity::Fatal => "fatal",
        EventSeverity::Unspecified => "unspecified",
    }
}

fn derive_tags(
    kind: &str,
    step_id: Option<Uuid>,
    payload: &Value,
    severity: EventSeverity,
) -> Vec<String> {
    let mut tags = Vec::with_capacity(6);
    tags.push("tap".to_string());
    tags.push(format!("kind:{}", kind.to_ascii_lowercase()));
    tags.push(format!("severity:{}", severity_label(severity)));

    if let Some(step_id) = step_id {
        tags.push("step".to_string());
        tags.push(format!("step_id:{step_id}"));
    } else {
        tags.push("run".to_string());
    }

    if let Some(status) = payload.get("status").and_then(Value::as_str) {
        tags.push(format!("status:{}", status.to_ascii_lowercase()));
    }
    if let Some(effect) = payload.get("effect").and_then(Value::as_str) {
        tags.push(format!("effect:{}", effect.to_ascii_lowercase()));
    }
    tags
}

fn build_response(
    config: &SubscriptionConfig,
    event: &models::OutboxEvent,
    severity: EventSeverity,
    tags: &[String],
    watermark: i64,
) -> Result<SubscribeRunResponse, Status> {
    let raw_struct = json_to_prost_struct(&event.payload).map_err(|err| {
        internal_error(format!(
            "failed to encode outbox payload {}: {err}",
            event.id
        ))
    })?;

    let trace = build_trace_context(config.run_id, event.step_id, &event.payload);

    let envelope = RunEventEnvelope {
        trace: Some(trace),
        recorded_at: Some(to_timestamp(event.created_at)),
        offset: event.id,
        severity: severity as i32,
        tags: tags.to_vec(),
        raw: Some(raw_struct),
        body: None,
    };

    Ok(SubscribeRunResponse {
        offset: event.id,
        watermark,
        heartbeat: false,
        event: Some(envelope),
    })
}

fn build_trace_context(run_id: Uuid, step_id: Option<Uuid>, payload: &Value) -> RunTraceContext {
    let mut ctx = RunTraceContext {
        trace_id: String::new(),
        span_id: String::new(),
        parent_span_id: String::new(),
        run: Some(tap::RunScope {
            run_id: run_id.to_string(),
            ..Default::default()
        }),
        step: None,
        tool_call: None,
        policy_decision: None,
        budget: None,
        seed: None,
        baggage: Vec::new(),
    };

    if let Some(trace_value) = payload.get("trace").and_then(Value::as_object) {
        if let Some(trace_id) = trace_value.get("trace_id").and_then(Value::as_str) {
            ctx.trace_id = trace_id.to_string();
        }
        if let Some(span_id) = trace_value.get("span_id").and_then(Value::as_str) {
            ctx.span_id = span_id.to_string();
        }
        if let Some(parent_span_id) = trace_value.get("parent_span_id").and_then(Value::as_str) {
            ctx.parent_span_id = parent_span_id.to_string();
        }
        if let Some(seed) = trace_value.get("seed").and_then(Value::as_i64) {
            ctx.seed = Some(seed);
        }
        if let Some(run_obj) = trace_value.get("run").and_then(Value::as_object) {
            let run_id_value = run_obj
                .get("run_id")
                .and_then(Value::as_str)
                .map(str::to_owned)
                .unwrap_or_else(|| run_id.to_string());
            let mut run_scope = tap::RunScope {
                run_id: run_id_value,
                workspace_id: run_obj
                    .get("workspace_id")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
                app_id: run_obj
                    .get("app_id")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
                attempt_id: run_obj
                    .get("attempt_id")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
                labels: Default::default(),
            };
            if let Some(labels) = run_obj.get("labels").and_then(Value::as_object) {
                for (key, value) in labels {
                    if let Some(string) = value.as_str() {
                        run_scope.labels.insert(key.clone(), string.to_string());
                    }
                }
            }
            ctx.run = Some(run_scope);
        }
        if let Some(step_obj) = trace_value.get("step").and_then(Value::as_object) {
            let mut scope = tap::StepScope {
                step_id: step_obj
                    .get("step_id")
                    .and_then(Value::as_str)
                    .map(str::to_owned)
                    .unwrap_or_else(|| step_id.map(|id| id.to_string()).unwrap_or_default()),
                index: step_obj
                    .get("index")
                    .and_then(Value::as_u64)
                    .map(|value| value as u32)
                    .unwrap_or_default(),
                attempt: step_obj
                    .get("attempt")
                    .and_then(Value::as_u64)
                    .map(|value| value as u32)
                    .unwrap_or_default(),
                kind: step_obj
                    .get("kind")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
                scheduler: step_obj
                    .get("scheduler")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
            };
            if scope.step_id.is_empty() {
                if let Some(id) = step_id {
                    scope.step_id = id.to_string();
                }
            }
            ctx.step = Some(scope);
        } else if let Some(id) = step_id {
            ctx.step = Some(tap::StepScope {
                step_id: id.to_string(),
                ..Default::default()
            });
        }
        if let Some(tool_obj) = trace_value.get("tool_call").and_then(Value::as_object) {
            ctx.tool_call = Some(tap::ToolCallScope {
                tool_call_id: tool_obj
                    .get("tool_call_id")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
                tool_name: tool_obj
                    .get("tool_name")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
                tool_variant: tool_obj
                    .get("tool_variant")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
                provider: tool_obj
                    .get("provider")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
            });
        }
        if let Some(policy_obj) = trace_value
            .get("policy_decision")
            .and_then(Value::as_object)
        {
            ctx.policy_decision = Some(tap::PolicyDecisionScope {
                policy_decision_id: policy_obj
                    .get("policy_decision_id")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
                policy_pack_id: policy_obj
                    .get("policy_pack_id")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
                policy_name: policy_obj
                    .get("policy_name")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
                policy_version: policy_obj
                    .get("policy_version")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
            });
        }
        if let Some(budget_obj) = trace_value.get("budget").and_then(Value::as_object) {
            let mut budget_scope = tap::BudgetScope {
                budget_id: budget_obj
                    .get("budget_id")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
                budget_kind: budget_obj
                    .get("budget_kind")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
                amount_remaining: None,
                amount_allocated: None,
            };
            if let Some(value) = budget_obj.get("amount_remaining").and_then(Value::as_f64) {
                budget_scope.amount_remaining = Some(value);
            }
            if let Some(value) = budget_obj.get("amount_allocated").and_then(Value::as_f64) {
                budget_scope.amount_allocated = Some(value);
            }
            ctx.budget = Some(budget_scope);
        }
        if let Some(baggage_value) = trace_value.get("baggage") {
            ctx.baggage = parse_baggage(baggage_value);
        }
    } else if ctx.step.is_none() {
        if let Some(step_id) = step_id {
            ctx.step = Some(tap::StepScope {
                step_id: step_id.to_string(),
                ..Default::default()
            });
        }
    }

    if ctx.run.is_none() {
        ctx.run = Some(tap::RunScope {
            run_id: run_id.to_string(),
            ..Default::default()
        });
    }

    ctx
}

fn parse_baggage(value: &Value) -> Vec<tap::BaggageItem> {
    if let Some(array) = value.as_array() {
        let mut items = Vec::with_capacity(array.len());
        for entry in array {
            if let Some(obj) = entry.as_object() {
                if let (Some(key), Some(value)) = (
                    obj.get("key").and_then(Value::as_str),
                    obj.get("value").and_then(Value::as_str),
                ) {
                    items.push(tap::BaggageItem {
                        key: key.to_string(),
                        value: value.to_string(),
                    });
                }
            }
        }
        return items;
    }

    if let Some(header) = value.as_str() {
        return header
            .split(',')
            .filter_map(|entry| {
                let mut parts = entry.splitn(2, '=');
                let key = parts.next()?.trim();
                let remainder = parts.next()?.trim();
                let value = remainder
                    .split_once(';')
                    .map(|(val, _)| val.trim())
                    .unwrap_or(remainder);
                if key.is_empty() || value.is_empty() {
                    None
                } else {
                    Some(tap::BaggageItem {
                        key: key.to_string(),
                        value: value.to_string(),
                    })
                }
            })
            .collect();
    }

    Vec::new()
}

#[derive(Debug)]
struct SubmitRunSpec {
    dag_json: Value,
    inputs: Value,
    labels: HashMap<String, String>,
    seed: i64,
    traceparent: Option<String>,
    tracestate: Option<String>,
    baggage: Option<String>,
    debug: Option<Value>,
    breakpoints: Vec<String>,
}

#[derive(Debug)]
enum SubmitRunOutcome {
    Created(Uuid),
    Reused(Uuid),
}

#[derive(Debug)]
enum SubmitRunError {
    Invalid(String),
    Internal(String),
}

impl std::fmt::Display for SubmitRunError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            SubmitRunError::Invalid(msg) => write!(f, "invalid run submission: {msg}"),
            SubmitRunError::Internal(msg) => write!(f, "internal run submission error: {msg}"),
        }
    }
}

impl SubmitRunError {
    fn invalid(message: impl Into<String>) -> Self {
        Self::Invalid(message.into())
    }

    fn internal(message: impl Into<String>) -> Self {
        Self::Internal(message.into())
    }
}

fn format_schema_validation_error(err: SchemaValidationError) -> String {
    if err.details.is_empty() {
        err.schema.to_string()
    } else {
        err.details
            .iter()
            .map(|detail| detail.to_string())
            .collect::<Vec<_>>()
            .join("; ")
    }
}

fn normalize_idempotency_key(raw: Option<&str>) -> Option<String> {
    raw.and_then(|value| {
        let trimmed = value.trim();
        if trimmed.is_empty() {
            None
        } else {
            Some(trimmed.to_owned())
        }
    })
}

fn resolve_breakpoint(
    spec: &str,
    slug_map: &HashMap<String, Uuid>,
    step_ids: &HashSet<Uuid>,
) -> Result<Uuid, SubmitRunError> {
    let trimmed = spec.trim();
    if trimmed.is_empty() {
        return Err(SubmitRunError::invalid("breakpoint spec cannot be empty"));
    }

    let (kind, target) = match trimmed.split_once(':') {
        Some((kind, rest)) => (kind.trim(), rest.trim()),
        None => ("step", trimmed),
    };

    if target.is_empty() {
        return Err(SubmitRunError::invalid(format!(
            "breakpoint '{}' missing target",
            spec
        )));
    }

    match kind.to_ascii_lowercase().as_str() {
        "step" => {
            if let Ok(uuid) = Uuid::parse_str(target) {
                if !step_ids.contains(&uuid) {
                    return Err(SubmitRunError::invalid(format!(
                        "breakpoint references unknown step id '{}'",
                        target
                    )));
                }
                return Ok(uuid);
            }

            let key = target.to_ascii_lowercase();
            if let Some(uuid) = slug_map.get(&key) {
                return Ok(*uuid);
            }

            Err(SubmitRunError::invalid(format!(
                "breakpoint references unknown step slug '{}'",
                target
            )))
        }
        other => Err(SubmitRunError::invalid(format!(
            "unsupported breakpoint kind '{}'",
            other
        ))),
    }
}

async fn submit_run_impl(
    storage: &Storage,
    audit: &Audit,
    spec: SubmitRunSpec,
    idempotency_key: Option<String>,
    actor: &str,
) -> Result<SubmitRunOutcome, SubmitRunError> {
    if let Some(ref key) = idempotency_key {
        if let Some(existing_run) =
            storage
                .fetch_run_by_idempotency_key(key)
                .await
                .map_err(|err| {
                    SubmitRunError::internal(format!("failed to check idempotency key: {err}"))
                })?
        {
            info!(
                run_id = %existing_run.run_id,
                idempotency_key = %key,
                "reusing existing run for idempotent submission"
            );
            return Ok(SubmitRunOutcome::Reused(existing_run.run_id));
        }
    }

    let SubmitRunSpec {
        dag_json,
        inputs,
        labels,
        seed,
        traceparent,
        tracestate,
        baggage,
        debug,
        breakpoints,
    } = spec;

    if seed <= 0 {
        return Err(SubmitRunError::invalid(
            "seed must be a positive integer to guarantee deterministic replay",
        ));
    }

    let steps_array = dag_json
        .get("steps")
        .and_then(Value::as_array)
        .ok_or_else(|| SubmitRunError::invalid("dag_json.steps must be an array"))?;

    let inputs_value = match inputs {
        Value::Null => Value::Object(Map::new()),
        Value::Object(map) => Value::Object(map),
        _ => return Err(SubmitRunError::invalid("inputs must be a JSON object")),
    };

    let labels_value = labels_map_to_value(&labels);

    let run_spec_instance = json!({
        "dag_json": dag_json.clone(),
        "inputs": inputs_value.clone(),
        "labels": labels_value.clone(),
        "seed": seed,
        "breakpoints": breakpoints.clone(),
    });

    if let Err(err) = validate_run_spec(&run_spec_instance) {
        return Err(SubmitRunError::invalid(format!(
            "run spec schema validation failed: {}",
            format_schema_validation_error(err)
        )));
    }

    let mut debug_map = match debug {
        Some(Value::Object(map)) => map,
        Some(_) => {
            return Err(SubmitRunError::invalid(
                "debug metadata must be a JSON object",
            ))
        }
        None => Map::new(),
    };

    let mut ctx_map = Map::new();
    ctx_map.insert("inputs".to_string(), inputs_value.clone());
    ctx_map.insert("labels".to_string(), labels_value.clone());
    if traceparent.is_some() || tracestate.is_some() || baggage.is_some() {
        let mut trace_obj = Map::new();
        if let Some(tp) = traceparent {
            trace_obj.insert("traceparent".to_string(), Value::String(tp));
        }
        if let Some(ts) = tracestate {
            trace_obj.insert("tracestate".to_string(), Value::String(ts));
        }
        if let Some(bg) = baggage {
            trace_obj.insert("baggage".to_string(), Value::String(bg));
        }
        ctx_map.insert("trace".to_string(), Value::Object(trace_obj));
    }

    let submission_time = Utc::now();

    struct PendingStep {
        idx: i32,
        step_id: Uuid,
        priority: i16,
        spec: Value,
        max_attempts: i32,
        retry_backoff_ms: i64,
        retry_backoff_factor: f64,
        deadline_at: Option<DateTime<Utc>>,
        estimated_tokens: i64,
        estimated_cost: f64,
        compensation_target: Option<String>,
    }

    let mut pending_steps = Vec::with_capacity(steps_array.len());
    let mut seen_step_ids = HashSet::with_capacity(steps_array.len());
    let mut slug_to_step: HashMap<String, Uuid> = HashMap::new();
    let mut slug_entries: Vec<(String, Uuid)> = Vec::new();

    for (idx, step_value) in steps_array.iter().enumerate() {
        if let Err(err) = validate_step_spec(step_value) {
            let detail = err
                .details
                .iter()
                .map(|err| err.to_string())
                .collect::<Vec<_>>()
                .join("; ");
            return Err(SubmitRunError::invalid(format!(
                "step schema validation failed: {detail}"
            )));
        }

        let step_obj = step_value
            .as_object()
            .ok_or_else(|| SubmitRunError::invalid("each step must be a JSON object"))?;

        let step_id_str = step_obj
            .get("id")
            .and_then(Value::as_str)
            .ok_or_else(|| SubmitRunError::invalid("step.id must be a UUID string"))?;
        let step_id = Uuid::parse_str(step_id_str)
            .map_err(|err| SubmitRunError::invalid(format!("invalid step.id: {err}")))?;

        if !seen_step_ids.insert(step_id) {
            return Err(SubmitRunError::invalid(format!(
                "duplicate step.id '{}' detected in dag_json.steps",
                step_id
            )));
        }

        if let Some(slug_value) = step_obj.get("slug").and_then(Value::as_str) {
            let trimmed = slug_value.trim();
            if trimmed.is_empty() {
                return Err(SubmitRunError::invalid("step.slug cannot be empty"));
            }
            let key = trimmed.to_ascii_lowercase();
            if slug_to_step.insert(key.clone(), step_id).is_some() {
                return Err(SubmitRunError::invalid(format!(
                    "duplicate step.slug '{}' detected in dag_json.steps",
                    trimmed
                )));
            }
            slug_entries.push((trimmed.to_string(), step_id));
        }

        let priority = extract_priority(step_obj);

        let execution_obj = step_obj.get("execution").and_then(Value::as_object);
        let max_attempts = execution_obj
            .and_then(|exec| exec.get("max_attempts").and_then(Value::as_i64))
            .and_then(|v| i32::try_from(v).ok())
            .filter(|v| *v >= 1)
            .unwrap_or(1);
        let retry_backoff_ms = execution_obj
            .and_then(|exec| exec.get("retry_backoff_ms").and_then(Value::as_i64))
            .map(|v| v.max(0))
            .unwrap_or(0);
        let retry_backoff_factor = execution_obj
            .and_then(|exec| exec.get("retry_backoff_factor").and_then(Value::as_f64))
            .filter(|v| *v >= 1.0)
            .unwrap_or(1.0);
        let deadline_at = execution_obj
            .and_then(|exec| exec.get("deadline_ms").and_then(Value::as_i64))
            .filter(|ms| *ms > 0)
            .map(|ms| submission_time + chrono::Duration::milliseconds(ms));

        let cost_hint =
            execution_obj.and_then(|exec| exec.get("cost_hint").and_then(Value::as_object));
        let estimated_tokens = cost_hint
            .and_then(|hint| hint.get("tokens").and_then(Value::as_i64))
            .map(|v| v.max(0))
            .unwrap_or(0);
        let estimated_cost = cost_hint
            .and_then(|hint| hint.get("usd").and_then(Value::as_f64))
            .map(|v| v.max(0.0))
            .unwrap_or(0.0);

        let compensation_target = execution_obj
            .and_then(|exec| exec.get("compensation").and_then(Value::as_object))
            .and_then(|comp| comp.get("step").and_then(Value::as_str))
            .map(|s| s.to_string());

        pending_steps.push(PendingStep {
            idx: idx as i32,
            step_id,
            priority,
            spec: step_value.clone(),
            max_attempts,
            retry_backoff_ms,
            retry_backoff_factor,
            deadline_at,
            estimated_tokens,
            estimated_cost,
            compensation_target,
        });
    }
    let mut incoming_counts: HashMap<Uuid, i32> =
        pending_steps.iter().map(|step| (step.step_id, 0)).collect();

    let mut edges: Vec<(Uuid, Uuid)> = Vec::new();
    if let Some(edges_array) = dag_json.get("edges").and_then(Value::as_array) {
        let mut seen_edges = HashSet::with_capacity(edges_array.len());
        for edge_value in edges_array {
            let pair = edge_value.as_array().ok_or_else(|| {
                SubmitRunError::invalid("dag_json.edges entries must be [from, to] arrays")
            })?;
            if pair.len() != 2 {
                return Err(SubmitRunError::invalid(
                    "dag_json.edges entries must contain exactly two elements",
                ));
            }

            let from_str = pair[0]
                .as_str()
                .ok_or_else(|| SubmitRunError::invalid("edge source must be a UUID string"))?;
            let to_str = pair[1]
                .as_str()
                .ok_or_else(|| SubmitRunError::invalid("edge target must be a UUID string"))?;

            let from = Uuid::parse_str(from_str).map_err(|err| {
                SubmitRunError::invalid(format!("invalid edge source UUID: {err}"))
            })?;
            let to = Uuid::parse_str(to_str).map_err(|err| {
                SubmitRunError::invalid(format!("invalid edge target UUID: {err}"))
            })?;

            if from == to {
                return Err(SubmitRunError::invalid(
                    "edge cannot reference the same step twice",
                ));
            }
            if !seen_step_ids.contains(&from) || !seen_step_ids.contains(&to) {
                return Err(SubmitRunError::invalid("edge references unknown step id"));
            }
            if !seen_edges.insert((from, to)) {
                return Err(SubmitRunError::invalid(
                    "duplicate edge detected in dag_json.edges",
                ));
            }

            let entry = incoming_counts.get_mut(&to).ok_or_else(|| {
                SubmitRunError::invalid("edge target not present in dag_json.steps")
            })?;
            let next_count = (*entry).checked_add(1).ok_or_else(|| {
                SubmitRunError::invalid("too many incoming edges for a single step")
            })?;
            *entry = next_count;

            edges.push((from, to));
        }
    }

    let mut breakpoint_payloads: Vec<Value> = Vec::new();
    let mut breakpoint_specs: Vec<String> = Vec::new();
    let mut breakpoint_ids: HashSet<Uuid> = HashSet::new();
    for bp in &breakpoints {
        let resolved = resolve_breakpoint(bp, &slug_to_step, &seen_step_ids)?;
        if !breakpoint_ids.insert(resolved) {
            return Err(SubmitRunError::invalid(format!(
                "duplicate breakpoint targeting step {}",
                resolved
            )));
        }

        let mut record = Map::new();
        record.insert("spec".to_string(), Value::String(bp.clone()));
        record.insert("step_id".to_string(), Value::String(resolved.to_string()));
        breakpoint_payloads.push(Value::Object(record));
        breakpoint_specs.push(bp.clone());
    }

    if !breakpoint_payloads.is_empty() {
        debug_map.insert(
            "breakpoints".to_string(),
            Value::Array(breakpoint_payloads.clone()),
        );
        let specs = breakpoint_specs
            .iter()
            .cloned()
            .map(Value::String)
            .collect::<Vec<_>>();
        debug_map.insert("breakpoint_specs".to_string(), Value::Array(specs));
    }

    if !slug_entries.is_empty() {
        let mut slugs = Map::new();
        for (slug, step_id) in &slug_entries {
            slugs.insert(slug.clone(), Value::String(step_id.to_string()));
        }
        debug_map.insert("slugs".to_string(), Value::Object(slugs));
    }

    debug_map
        .entry("active_pause".to_string())
        .or_insert(Value::Null);

    if !debug_map.is_empty() {
        ctx_map.insert("debug".to_string(), Value::Object(debug_map.clone()));
    }

    let input_ctx = Value::Object(ctx_map);

    let run_id = Uuid::new_v4();
    let mut steps = Vec::with_capacity(pending_steps.len());
    for pending in pending_steps {
        let PendingStep {
            idx,
            step_id,
            priority,
            spec,
            max_attempts,
            retry_backoff_ms,
            retry_backoff_factor,
            deadline_at,
            estimated_tokens,
            estimated_cost,
            compensation_target,
        } = pending;

        let dep_count = *incoming_counts.get(&step_id).unwrap_or(&0);
        let status = if dep_count == 0 {
            models::StepStatus::Queued
        } else {
            models::StepStatus::Blocked
        };

        let compensation_step = if let Some(target) = compensation_target {
            let trimmed = target.trim();
            if trimmed.is_empty() {
                return Err(SubmitRunError::invalid(format!(
                    "compensation step reference cannot be empty for step {}",
                    step_id
                )));
            }

            let resolved = if let Ok(uuid) = Uuid::parse_str(trimmed) {
                if !seen_step_ids.contains(&uuid) {
                    return Err(SubmitRunError::invalid(format!(
                        "compensation step '{}' not found in dag_json.steps",
                        trimmed
                    )));
                }
                uuid
            } else {
                let key = trimmed.to_ascii_lowercase();
                *slug_to_step.get(&key).ok_or_else(|| {
                    SubmitRunError::invalid(format!(
                        "compensation step '{}' not found in dag_json.steps",
                        trimmed
                    ))
                })?
            };

            if resolved == step_id {
                return Err(SubmitRunError::invalid(format!(
                    "step {} cannot reference itself as a compensation step",
                    step_id
                )));
            }

            Some(resolved)
        } else {
            None
        };

        steps.push(models::NewStep {
            run_id,
            step_id,
            idx,
            priority,
            spec_json: spec,
            status,
            attempt: 0,
            input_snapshot: None,
            provider: None,
            provider_version: None,
            max_attempts,
            retry_backoff_ms,
            retry_backoff_factor,
            not_before: None,
            deadline_at,
            pending_dependencies: dep_count,
            total_dependencies: dep_count,
            estimated_tokens,
            estimated_cost,
            actual_tokens: None,
            actual_cost: None,
            checkpoint: None,
            compensation_step,
            compensation_scheduled: false,
        });
    }

    let new_run = models::NewRun {
        run_id,
        status: models::RunStatus::Pending,
        dag_json: dag_json.clone(),
        input_ctx,
        seed,
        idempotency_key: idempotency_key.clone(),
    };

    let stored_run_id = storage
        .insert_run_with_steps(new_run, &steps, &edges)
        .await
        .map_err(|err| {
            error!(error = ?err, "failed to persist run");
            SubmitRunError::internal(format!("failed to persist run: {err}"))
        })?;

    info!(run_id = %stored_run_id, "accepted run submission");

    let tenant_id = labels
        .get("tenant_id")
        .or_else(|| labels.get("tenant"))
        .and_then(|value| Uuid::parse_str(value).ok());
    let audit_resource = json!({
        "run_id": stored_run_id.to_string(),
        "idempotency_key": idempotency_key,
        "labels": labels,
    });

    if let Err(err) = audit
        .append(actor, tenant_id, "run.submit", &audit_resource)
        .await
    {
        warn!(
            error = %err,
            run_id = %stored_run_id,
            "failed to append audit log for run submission"
        );
    }

    Ok(SubmitRunOutcome::Created(stored_run_id))
}

#[tonic::async_trait]
impl RuntimeService for RuntimeApiService {
    async fn submit_run(
        &self,
        request: Request<runtime::SubmitRunRequest>,
    ) -> Result<Response<runtime::SubmitRunResponse>, Status> {
        let metadata = request.metadata().clone();
        let role = self.require_role(&metadata, Role::Writer).await?;
        let req = request.into_inner();
        let runtime::SubmitRunRequest {
            spec,
            idempotency_key: raw_idempotency_key,
        } = req;

        let idempotency_key = normalize_idempotency_key(Some(raw_idempotency_key.as_str()));

        let spec = spec.ok_or_else(|| invalid_argument("spec is required"))?;

        let dag_json = parse_dag_json(&spec.dag_json)?;
        let inputs_value = spec
            .inputs
            .as_ref()
            .map(prost_struct_to_json)
            .unwrap_or_else(|| Value::Object(Map::new()));

        let traceparent = metadata
            .get("traceparent")
            .and_then(|value| value.to_str().ok())
            .map(|s| s.to_string());
        let tracestate = metadata
            .get("tracestate")
            .and_then(|value| value.to_str().ok())
            .map(|s| s.to_string());
        let baggage = metadata
            .get("baggage")
            .and_then(|value| value.to_str().ok())
            .map(|s| s.to_string());

        let submission = SubmitRunSpec {
            dag_json,
            inputs: inputs_value,
            labels: spec.labels.clone(),
            seed: spec.seed,
            traceparent,
            tracestate,
            baggage,
            debug: None,
            breakpoints: spec.breakpoints.clone(),
        };

        let actor = Self::actor_from_metadata(&metadata, role);

        let outcome = submit_run_impl(
            self.storage.as_ref(),
            self.audit.as_ref(),
            submission,
            idempotency_key,
            &actor,
        )
        .await
        .map_err(|err| match err {
            SubmitRunError::Invalid(message) => invalid_argument(message),
            SubmitRunError::Internal(message) => internal_error(message),
        })?;

        let run_id = match outcome {
            SubmitRunOutcome::Created(run_id) | SubmitRunOutcome::Reused(run_id) => run_id,
        };

        Ok(Response::new(runtime::SubmitRunResponse {
            run_id: run_id.to_string(),
        }))
    }

    async fn get_run(
        &self,
        request: Request<runtime::GetRunRequest>,
    ) -> Result<Response<runtime::GetRunResponse>, Status> {
        let metadata = request.metadata().clone();
        self.require_role(&metadata, Role::Reader).await?;
        let req = request.into_inner();
        let run_id = Uuid::parse_str(&req.run_id)
            .map_err(|err| invalid_argument(format!("invalid run_id: {err}")))?;

        let run = self
            .storage
            .fetch_run(run_id)
            .await
            .map_err(|err| internal_error(format!("failed to fetch run: {err}")))?
            .ok_or_else(|| Status::not_found("run not found"))?;

        let steps = self
            .storage
            .fetch_steps_for_run(run_id)
            .await
            .map_err(|err| internal_error(format!("failed to fetch steps: {err}")))?;

        let run_detail = build_run_detail(run, steps)?;

        Ok(Response::new(runtime::GetRunResponse {
            run: Some(run_detail),
        }))
    }

    type TapRunStream = Pin<Box<dyn Stream<Item = Result<runtime::RunEvent, Status>> + Send>>;

    async fn tap_run(
        &self,
        request: Request<runtime::TapRunRequest>,
    ) -> Result<Response<Self::TapRunStream>, Status> {
        let metadata = request.metadata().clone();
        self.require_role(&metadata, Role::Reader).await?;
        let req = request.into_inner();
        let run_uuid = Uuid::parse_str(&req.run_id)
            .map_err(|err| invalid_argument(format!("invalid run_id: {err}")))?;

        // Ensure the run exists before entering the stream loop.
        if self
            .storage
            .fetch_run(run_uuid)
            .await
            .map_err(|err| internal_error(format!("failed to verify run: {err}")))?
            .is_none()
        {
            return Err(Status::not_found("run not found"));
        }

        let storage = Arc::clone(&self.storage);
        let (tx, rx) = mpsc::channel::<Result<runtime::RunEvent, Status>>(32);

        tokio::spawn(async move {
            let mut last_id: i64 = 0;

            loop {
                if tx.is_closed() {
                    break;
                }

                match storage
                    .fetch_outbox_for_run_since(run_uuid, last_id, 256)
                    .await
                {
                    Ok(events) => {
                        if events.is_empty() {
                            sleep(Duration::from_millis(500)).await;
                            continue;
                        }

                        for event in events {
                            last_id = event.id;
                            match run_event_from_outbox(event) {
                                Ok(run_event) => {
                                    if tx.send(Ok(run_event)).await.is_err() {
                                        return;
                                    }
                                }
                                Err(status) => {
                                    let _ = tx.send(Err(status)).await;
                                    return;
                                }
                            }
                        }
                    }
                    Err(err) => {
                        warn!(run_id = %run_uuid, error = %err, "tap stream failed to read outbox events");
                        let status = internal_error(format!("failed to fetch run events: {err}"));
                        let _ = tx.send(Err(status)).await;
                        break;
                    }
                }
            }
        });

        let stream = ReceiverStream::new(rx);
        Ok(Response::new(Box::pin(stream) as Self::TapRunStream))
    }

    async fn replay_run(
        &self,
        request: Request<runtime::ReplayRunRequest>,
    ) -> Result<Response<runtime::ReplayRunResponse>, Status> {
        let metadata = request.metadata().clone();
        self.require_role(&metadata, Role::Writer).await?;
        let req = request.into_inner();
        let run_uuid = Uuid::parse_str(&req.run_id)
            .map_err(|err| invalid_argument(format!("invalid run_id: {err}")))?;

        if self
            .storage
            .fetch_run(run_uuid)
            .await
            .map_err(|err| internal_error(format!("failed to load run: {err}")))?
            .is_none()
        {
            return Err(Status::not_found("run not found"));
        }

        let strategy = match runtime::ReplayMode::try_from(req.mode).ok() {
            Some(runtime::ReplayMode::Unspecified)
            | Some(runtime::ReplayMode::Full)
            | Some(runtime::ReplayMode::Diff) => runtime::ReplayStrategy::Mocked,
            Some(runtime::ReplayMode::StepOnly) => {
                return Err(Status::unimplemented(
                    "Replay mode STEP_ONLY is not yet supported",
                ));
            }
            None => {
                return Err(invalid_argument("unrecognized replay mode"));
            }
        };

        let steps = self
            .storage
            .fetch_steps_for_run(run_uuid)
            .await
            .map_err(|err| internal_error(format!("failed to load steps for replay: {err}")))?;
        let attempts = self
            .storage
            .fetch_step_attempts_for_run(run_uuid)
            .await
            .map_err(|err| {
                internal_error(format!("failed to load step attempts for replay: {err}"))
            })?;

        let mut attempts_by_step: HashMap<Uuid, Vec<models::StepAttempt>> = HashMap::new();
        for attempt in attempts {
            attempts_by_step
                .entry(attempt.step_id)
                .or_default()
                .push(attempt);
        }
        for entries in attempts_by_step.values_mut() {
            entries.sort_by_key(|attempt| attempt.attempt);
        }

        let replay_mode = runtime::ReplayMode::try_from(req.mode).unwrap_or(runtime::ReplayMode::Full);
        let diff_mode = matches!(replay_mode, runtime::ReplayMode::Diff);

        let mut proto_steps = Vec::with_capacity(steps.len());
        let mut all_within = true;

        for step in steps {
            let attempts_for_step = attempts_by_step.get(&step.step_id);
            let latest_attempt = attempts_for_step.and_then(|entries| entries.last());

            let recorded_snapshot = latest_attempt
                .and_then(|attempt| attempt.output_snapshot.clone())
                .or_else(|| step.output_snapshot.clone())
                .ok_or_else(|| {
                    Status::failed_precondition(format!(
                        "step {} missing output_snapshot; cannot replay",
                        step.step_id
                    ))
                })?;

            let baseline_attempt = if diff_mode {
                attempts_for_step.and_then(|entries| {
                    if entries.len() >= 2 {
                        entries.get(entries.len() - 2)
                    } else {
                        None
                    }
                })
            } else {
                latest_attempt
            };

            let baseline_output = baseline_attempt
                .and_then(|attempt| attempt.output_snapshot.clone())
                .unwrap_or_else(|| recorded_snapshot.clone());

            let current_output = if diff_mode {
                recorded_snapshot.clone()
            } else {
                step.output_json
                    .clone()
                    .unwrap_or_else(|| recorded_snapshot.clone())
            };

            let drift_result = compute_drift(
                &baseline_output,
                &current_output,
                DEFAULT_TOKEN_TOLERANCE,
                DEFAULT_COST_TOLERANCE,
            );
            if !drift_result.within_tolerance {
                all_within = false;
            }

            let snapshot_value = json_to_prost_value(&recorded_snapshot)
                .map_err(|err| internal_error(format!("failed to encode snapshot: {err}")))?;

            let recorded_meta = wrap_attempt_output(baseline_attempt, &baseline_output);
            let current_meta = wrap_attempt_output(latest_attempt, &current_output);

            let drift_proto = runtime::ReplayStepDrift {
                recorded: Some(json_to_prost_value(&recorded_meta).map_err(|err| {
                    internal_error(format!("failed to encode recorded drift snapshot: {err}"))
                })?),
                current: Some(json_to_prost_value(&current_meta).map_err(|err| {
                    internal_error(format!("failed to encode current drift snapshot: {err}"))
                })?),
                token_delta: drift_result.token_delta,
                cost_delta: drift_result.cost_delta,
                within_tolerance: drift_result.within_tolerance,
            };

            proto_steps.push(runtime::ReplayStep {
                step_id: step.step_id.to_string(),
                snapshot: Some(snapshot_value),
                drift: Some(drift_proto),
            });
        }

        let response = runtime::ReplayRunResponse {
            run_id: req.run_id,
            all_within_tolerance: all_within,
            strategy: strategy as i32,
            steps: proto_steps,
        };

        Ok(Response::new(response))
    }

    async fn update_breakpoints(
        &self,
        request: Request<runtime::UpdateBreakpointsRequest>,
    ) -> Result<Response<runtime::UpdateBreakpointsResponse>, Status> {
        let metadata = request.metadata().clone();
        let role = self.require_role(&metadata, Role::Writer).await?;
        let actor = Self::actor_from_metadata(&metadata, role);
        let req = request.into_inner();

        let run_uuid = Uuid::parse_str(&req.run_id)
            .map_err(|err| invalid_argument(format!("invalid run_id: {err}")))?;

        if !req.clear_all && req.add.is_empty() && req.remove.is_empty() {
            return Err(invalid_argument(
                "update_breakpoints requires add/remove entries or clear_all",
            ));
        }

        let run = self
            .storage
            .fetch_run(run_uuid)
            .await
            .map_err(|err| internal_error(format!("failed to fetch run: {err}")))?
            .ok_or_else(|| Status::not_found("run not found"))?;

        let step_ids = collect_step_ids(&run);

        let ctx_value = run.input_ctx.clone();
        let mut ctx_map = match ctx_value {
            Value::Object(map) => map,
            Value::Null => Map::new(),
            other => {
                return Err(internal_error(format!(
                    "run input context must be an object, found {other}"
                )))
            }
        };

        let debug_entry = ctx_map
            .entry("debug".to_string())
            .or_insert_with(|| Value::Object(Map::new()));
        let debug_map = debug_entry.as_object_mut().ok_or_else(|| {
            internal_error("run debug context is not an object; unable to update breakpoints")
        })?;

        let mut breakpoints = load_debug_breakpoints(debug_map);
        let existing_slug_entries = load_slug_map(debug_map);

        let mut slug_lookup: HashMap<String, Uuid> = existing_slug_entries
            .iter()
            .map(|(slug, id)| (slug.to_ascii_lowercase(), *id))
            .collect();
        if slug_lookup.is_empty() {
            for (slug, id) in collect_slug_entries(&run) {
                slug_lookup.insert(slug.to_ascii_lowercase(), id);
            }
        }

        let slug_entries_store: Vec<(String, Uuid)> = if existing_slug_entries.is_empty() {
            collect_slug_entries(&run)
        } else {
            existing_slug_entries
                .iter()
                .map(|(slug, id)| (slug.clone(), *id))
                .collect()
        };

        if req.clear_all {
            breakpoints.clear();
        }

        let removal_specs: HashSet<String> = req
            .remove
            .iter()
            .map(|s| s.trim().to_string())
            .filter(|s| !s.is_empty())
            .collect();

        let mut removal_ids: HashSet<Uuid> = HashSet::new();
        for spec in &req.remove {
            if let Ok(step_id) = resolve_breakpoint(spec, &slug_lookup, &step_ids) {
                removal_ids.insert(step_id);
            }
        }

        if !removal_specs.is_empty() || !removal_ids.is_empty() {
            breakpoints.retain(|(spec, step_id)| {
                let spec_trim = spec.trim();
                let matches_spec = removal_specs.contains(spec_trim);
                let matches_id = removal_ids.contains(step_id);
                !(matches_spec || matches_id)
            });
        }

        for spec in &req.add {
            let step_id = resolve_breakpoint(spec, &slug_lookup, &step_ids)
                .map_err(|err| invalid_argument(err.to_string()))?;
            breakpoints.retain(|(_, existing)| existing != &step_id);
            breakpoints.push((spec.trim().to_string(), step_id));
        }

        breakpoints.sort_by(|a, b| a.0.cmp(&b.0));
        store_debug_breakpoints(debug_map, &breakpoints);
        if !slug_entries_store.is_empty() {
            store_slug_map(debug_map, &slug_entries_store);
        }

        let mut tx = self
            .storage
            .pool()
            .begin()
            .await
            .map_err(|err| internal_error(format!("failed to begin transaction: {err}")))?;

        self.storage
            .update_run_input_ctx_tx(&mut tx, run_uuid, Value::Object(ctx_map.clone()))
            .await
            .map_err(|err| {
                internal_error(format!("failed to persist breakpoint changes: {err}"))
            })?;

        let payload_breakpoints = breakpoints
            .iter()
            .map(|(spec, step_id)| {
                json!({
                    "spec": spec,
                    "step_id": step_id.to_string(),
                })
            })
            .collect::<Vec<_>>();

        self.storage
            .enqueue_outbox_event(
                &mut tx,
                NewOutboxEvent {
                    run_id: run_uuid,
                    step_id: None,
                    kind: "run_breakpoints_updated".to_string(),
                    payload: json!({
                        "run_id": run_uuid.to_string(),
                        "breakpoints": payload_breakpoints,
                        "added": req.add,
                        "removed": req.remove,
                        "cleared": req.clear_all,
                    }),
                },
            )
            .await
            .map_err(|err| internal_error(format!("failed to enqueue outbox event: {err}")))?;

        tx.commit()
            .await
            .map_err(|err| internal_error(format!("failed to commit breakpoint update: {err}")))?;

        if let Err(err) = self
            .audit
            .append(
                &actor,
                self.storage
                    .fetch_run_tenant(run_uuid)
                    .await
                    .ok()
                    .flatten()
                    .and_then(|tenant| Uuid::parse_str(&tenant).ok()),
                "debug.breakpoints.update",
                &json!({
                    "run_id": run_uuid.to_string(),
                    "added": req.add,
                    "removed": req.remove,
                    "cleared": req.clear_all,
                }),
            )
            .await
        {
            warn!(error = %err, run = %run_uuid, "failed to append audit record for breakpoint update");
        }

        Ok(Response::new(runtime::UpdateBreakpointsResponse {
            breakpoints: breakpoints.into_iter().map(|(spec, _)| spec).collect(),
        }))
    }

    async fn resume_run(
        &self,
        request: Request<runtime::ResumeRunRequest>,
    ) -> Result<Response<runtime::ResumeRunResponse>, Status> {
        let metadata = request.metadata().clone();
        let role = self.require_role(&metadata, Role::Writer).await?;
        let actor = Self::actor_from_metadata(&metadata, role);
        let req = request.into_inner();

        let run_uuid = Uuid::parse_str(&req.run_id)
            .map_err(|err| invalid_argument(format!("invalid run_id: {err}")))?;

        let run = self
            .storage
            .fetch_run(run_uuid)
            .await
            .map_err(|err| internal_error(format!("failed to fetch run: {err}")))?
            .ok_or_else(|| Status::not_found("run not found"))?;

        if run.status != models::RunStatus::PausedAtStep {
            return Err(Status::failed_precondition(
                "run is not paused at a breakpoint",
            ));
        }

        let ctx_value = run.input_ctx.clone();
        let mut ctx_map = match ctx_value {
            Value::Object(map) => map,
            Value::Null => Map::new(),
            other => {
                return Err(internal_error(format!(
                    "run input context must be an object, found {other}"
                )))
            }
        };

        let debug_entry = ctx_map
            .get_mut("debug")
            .and_then(Value::as_object_mut)
            .ok_or_else(|| internal_error("run debug context missing; cannot resume paused run"))?;

        let paused_step = extract_active_pause_step_id(debug_entry).ok_or_else(|| {
            Status::failed_precondition("run is paused but no active step recorded")
        })?;

        debug_entry.insert("active_pause".to_string(), Value::Null);

        let mut tx = self
            .storage
            .pool()
            .begin()
            .await
            .map_err(|err| internal_error(format!("failed to begin transaction: {err}")))?;

        self.storage
            .update_run_input_ctx_tx(&mut tx, run_uuid, Value::Object(ctx_map.clone()))
            .await
            .map_err(|err| internal_error(format!("failed to persist pause state: {err}")))?;

        self.storage
            .set_run_status_tx(&mut tx, run_uuid, models::RunStatus::Running)
            .await
            .map_err(|err| internal_error(format!("failed to update run status: {err}")))?;

        self.storage
            .enqueue_outbox_event(
                &mut tx,
                NewOutboxEvent {
                    run_id: run_uuid,
                    step_id: Some(paused_step),
                    kind: "run_resumed".to_string(),
                    payload: json!({
                        "run_id": run_uuid.to_string(),
                        "step_id": paused_step.to_string(),
                    }),
                },
            )
            .await
            .map_err(|err| internal_error(format!("failed to enqueue resume event: {err}")))?;

        tx.commit()
            .await
            .map_err(|err| internal_error(format!("failed to commit resume transaction: {err}")))?;

        if let Err(err) = self
            .audit
            .append(
                &actor,
                self.storage
                    .fetch_run_tenant(run_uuid)
                    .await
                    .ok()
                    .flatten()
                    .and_then(|tenant| Uuid::parse_str(&tenant).ok()),
                "debug.breakpoints.resume",
                &json!({
                    "run_id": run_uuid.to_string(),
                    "step_id": paused_step.to_string(),
                }),
            )
            .await
        {
            warn!(error = %err, run = %run_uuid, "failed to append audit record for resume");
        }

        Ok(Response::new(runtime::ResumeRunResponse {
            status: runtime::RunStatus::Running as i32,
            resumed_step_id: paused_step.to_string(),
        }))
    }

    async fn patch_step(
        &self,
        request: Request<runtime::PatchStepRequest>,
    ) -> Result<Response<runtime::PatchStepResponse>, Status> {
        let metadata = request.metadata().clone();
        let role = self.require_role(&metadata, Role::Writer).await?;
        let actor = Self::actor_from_metadata(&metadata, role);
        let req = request.into_inner();

        let run_uuid = Uuid::parse_str(&req.run_id)
            .map_err(|err| invalid_argument(format!("invalid run_id: {err}")))?;
        let step_uuid = Uuid::parse_str(&req.step_id)
            .map_err(|err| invalid_argument(format!("invalid step_id: {err}")))?;

        let run = self
            .storage
            .fetch_run(run_uuid)
            .await
            .map_err(|err| internal_error(format!("failed to fetch run: {err}")))?
            .ok_or_else(|| Status::not_found("run not found"))?;

        if run.status != models::RunStatus::PausedAtStep {
            return Err(Status::failed_precondition(
                "run must be paused before patching step",
            ));
        }

        let ctx_value = run.input_ctx.clone();
        let paused_step = match ctx_value {
            Value::Object(ref map) => map
                .get("debug")
                .and_then(Value::as_object)
                .and_then(extract_active_pause_step_id),
            _ => None,
        };

        if let Some(active_step) = paused_step {
            if active_step != step_uuid {
                return Err(Status::failed_precondition(
                    "only the paused step can be patched",
                ));
            }
        }

        let inputs_patch_value = req.inputs_patch.as_ref().map(prost_struct_to_json);
        let env_override = req.env.as_ref().map(prost_struct_to_json);

        let model_override = req.model.clone();

        if inputs_patch_value.is_none() && env_override.is_none() && model_override.is_none() {
            return Err(invalid_argument(
                "patch_step requires inputs_patch, env, or model to be set",
            ));
        }

        let steps = self
            .storage
            .fetch_steps_for_run(run_uuid)
            .await
            .map_err(|err| internal_error(format!("failed to fetch steps: {err}")))?;

        let step = steps
            .into_iter()
            .find(|step| step.step_id == step_uuid)
            .ok_or_else(|| Status::not_found("step not found"))?;

        let mut spec_map = match step.spec_json.clone() {
            Value::Object(map) => map,
            _ => {
                return Err(internal_error(
                    "step spec must be a JSON object to apply patches",
                ))
            }
        };

        let inputs_entry = spec_map
            .entry("inputs".to_string())
            .or_insert_with(|| Value::Object(Map::new()));
        let inputs_map = inputs_entry
            .as_object_mut()
            .ok_or_else(|| internal_error("step inputs is not an object; cannot apply patch"))?;

        if let Some(patch_value) = inputs_patch_value.as_ref() {
            let patch_obj = patch_value
                .as_object()
                .ok_or_else(|| invalid_argument("inputs_patch must be a JSON object"))?;
            merge_json_object(inputs_map, patch_obj);
        }

        if let Some(env_value) = env_override.as_ref() {
            if !env_value.is_object() {
                return Err(invalid_argument("env override must be a JSON object"));
            }
            inputs_map.insert("env".to_string(), env_value.clone());
        }

        if let Some(model) = model_override.as_ref() {
            inputs_map.insert("model".to_string(), Value::String(model.clone()));
        }

        let updated_spec = Value::Object(spec_map.clone());

        let mut tx = self
            .storage
            .pool()
            .begin()
            .await
            .map_err(|err| internal_error(format!("failed to begin transaction: {err}")))?;

        self.storage
            .update_step_spec_tx(&mut tx, run_uuid, step_uuid, updated_spec.clone())
            .await
            .map_err(|err| internal_error(format!("failed to persist step patch: {err}")))?;

        self.storage
            .enqueue_outbox_event(
                &mut tx,
                NewOutboxEvent {
                    run_id: run_uuid,
                    step_id: Some(step_uuid),
                    kind: "step_patched".to_string(),
                    payload: json!({
                        "run_id": run_uuid.to_string(),
                        "step_id": step_uuid.to_string(),
                        "inputs_patch": inputs_patch_value,
                        "env_override": env_override,
                        "model_override": model_override,
                        "note": req.note,
                    }),
                },
            )
            .await
            .map_err(|err| internal_error(format!("failed to enqueue patch event: {err}")))?;

        tx.commit()
            .await
            .map_err(|err| internal_error(format!("failed to commit patch transaction: {err}")))?;

        if let Err(err) = self
            .audit
            .append(
                &actor,
                self.storage
                    .fetch_run_tenant(run_uuid)
                    .await
                    .ok()
                    .flatten()
                    .and_then(|tenant| Uuid::parse_str(&tenant).ok()),
                "debug.step.patch",
                &json!({
                    "run_id": run_uuid.to_string(),
                    "step_id": step_uuid.to_string(),
                    "note": req.note,
                }),
            )
            .await
        {
            warn!(error = %err, run = %run_uuid, step = %step_uuid, "failed to append audit record for step patch");
        }

        Ok(Response::new(runtime::PatchStepResponse {
            step_id: step_uuid.to_string(),
        }))
    }

    async fn create_branch_run(
        &self,
        request: Request<runtime::CreateBranchRunRequest>,
    ) -> Result<Response<runtime::CreateBranchRunResponse>, Status> {
        let metadata = request.metadata().clone();
        let role = self.require_role(&metadata, Role::Writer).await?;
        let actor = Self::actor_from_metadata(&metadata, role);
        let req = request.into_inner();

        let run_uuid = Uuid::parse_str(&req.run_id)
            .map_err(|err| invalid_argument(format!("invalid run_id: {err}")))?;

        let from_step_spec = req.from_step.trim().to_string();
        if from_step_spec.is_empty() {
            return Err(invalid_argument("from_step is required"));
        }

        let seed_mode = runtime::BranchSeedMode::try_from(req.seed_mode)
            .unwrap_or(runtime::BranchSeedMode::Same);

        let run = self
            .storage
            .fetch_run(run_uuid)
            .await
            .map_err(|err| internal_error(format!("failed to fetch run: {err}")))?
            .ok_or_else(|| Status::not_found("run not found"))?;

        let (inputs_value, mut labels_map) = unpack_input_ctx(&run.input_ctx);

        let debug_entry = run
            .input_ctx
            .get("debug")
            .and_then(Value::as_object)
            .cloned()
            .unwrap_or_default();
        let slug_lookup = load_slug_map(&debug_entry);
        let step_ids = collect_step_ids(&run);
        let target_step = resolve_breakpoint(&from_step_spec, &slug_lookup, &step_ids)
            .map_err(|err| invalid_argument(err.to_string()))?;

        let branch_dag = build_branch_dag(&run.dag_json, target_step)
            .map_err(|err| invalid_argument(err.to_string()))?;

        let mut inputs_map = match inputs_value {
            Value::Object(map) => map,
            Value::Null => Map::new(),
            _ => Map::new(),
        };

        let mut inputs_patch_debug: Option<Value> = None;
        if let Some(patch) = req.inputs_patch.as_ref() {
            let patch_value = prost_struct_to_json(patch);
            let patch_obj = patch_value
                .as_object()
                .ok_or_else(|| invalid_argument("inputs_patch must be a JSON object"))?;
            merge_json_object(&mut inputs_map, patch_obj);
            inputs_patch_debug = Some(Value::Object(patch_obj.clone()));
        }

        let mut labels_patch_debug: Option<Value> = None;
        if let Some(labels_patch) = req.labels_patch.as_ref() {
            let patch_value = prost_struct_to_json(labels_patch);
            let patch_obj = patch_value
                .as_object()
                .ok_or_else(|| invalid_argument("labels_patch must be a JSON object"))?;
            for (key, value) in patch_obj {
                let rendered = value
                    .as_str()
                    .map(|s| s.to_string())
                    .unwrap_or_else(|| value.to_string());
                labels_map.insert(key.clone(), rendered);
            }
            labels_patch_debug = Some(Value::Object(patch_obj.clone()));
        }

        labels_map.insert("branch_parent".to_string(), run_uuid.to_string());
        labels_map.insert("branch_from_step".to_string(), target_step.to_string());

        let branch_seed = match seed_mode {
            runtime::BranchSeedMode::Same | runtime::BranchSeedMode::Unspecified => run.seed,
            runtime::BranchSeedMode::New => Utc::now().timestamp_nanos_opt().unwrap_or(run.seed),
            runtime::BranchSeedMode::Custom => req
                .seed
                .as_ref()
                .copied()
                .ok_or_else(|| invalid_argument("seed value required for custom mode"))?,
        };

        let prompt_packs = collect_prompt_packs_json(&run.dag_json);

        let mut branch_meta = Map::new();
        branch_meta.insert(
            "parent_run_id".to_string(),
            Value::String(run_uuid.to_string()),
        );
        branch_meta.insert(
            "parent_step_id".to_string(),
            Value::String(target_step.to_string()),
        );
        branch_meta.insert(
            "from_step_spec".to_string(),
            Value::String(from_step_spec.clone()),
        );
        branch_meta.insert(
            "parent_seed".to_string(),
            Value::Number(Number::from(run.seed)),
        );
        branch_meta.insert(
            "branch_seed".to_string(),
            Value::Number(Number::from(branch_seed)),
        );
        let seed_mode_str = match seed_mode {
            runtime::BranchSeedMode::Same | runtime::BranchSeedMode::Unspecified => "same",
            runtime::BranchSeedMode::New => "new",
            runtime::BranchSeedMode::Custom => "custom",
        };
        branch_meta.insert(
            "seed_mode".to_string(),
            Value::String(seed_mode_str.to_string()),
        );
        branch_meta.insert(
            "created_at".to_string(),
            Value::String(Utc::now().to_rfc3339()),
        );
        branch_meta.insert(
            "prompt_packs".to_string(),
            Value::Array(prompt_packs.into_iter().map(Value::String).collect()),
        );
        let note = req.note.trim();
        if !note.is_empty() {
            branch_meta.insert("note".to_string(), Value::String(note.to_string()));
        }
        if let Some(patch) = inputs_patch_debug {
            branch_meta.insert("inputs_patch".to_string(), patch);
        }
        if let Some(patch) = labels_patch_debug {
            branch_meta.insert("labels_patch".to_string(), patch);
        }

        let mut debug_extra = Map::new();
        debug_extra.insert("branch".to_string(), Value::Object(branch_meta));

        let submission = SubmitRunSpec {
            dag_json: branch_dag,
            inputs: Value::Object(inputs_map.clone()),
            labels: labels_map.clone(),
            seed: branch_seed,
            traceparent: None,
            tracestate: None,
            baggage: None,
            debug: Some(Value::Object(debug_extra.clone())),
            breakpoints: Vec::new(),
        };

        let outcome = submit_run_impl(
            self.storage.as_ref(),
            self.audit.as_ref(),
            submission,
            None,
            &actor,
        )
        .await
        .map_err(|err| match err {
            SubmitRunError::Invalid(message) => invalid_argument(message),
            SubmitRunError::Internal(message) => internal_error(message),
        })?;

        let new_run_uuid = match outcome {
            SubmitRunOutcome::Created(run_id) | SubmitRunOutcome::Reused(run_id) => run_id,
        };

        let mut tx = self.storage.pool().begin().await.map_err(|err| {
            internal_error(format!("failed to begin branch event transaction: {err}"))
        })?;

        self.storage
            .enqueue_outbox_event(
                &mut tx,
                NewOutboxEvent {
                    run_id: run_uuid,
                    step_id: Some(target_step),
                    kind: "run_branch_created".to_string(),
                    payload: json!({
                        "run_id": run_uuid.to_string(),
                        "child_run_id": new_run_uuid.to_string(),
                        "from_step_spec": from_step_spec,
                        "seed": branch_seed,
                    }),
                },
            )
            .await
            .map_err(|err| internal_error(format!("failed to enqueue branch event: {err}")))?;

        tx.commit().await.map_err(|err| {
            internal_error(format!("failed to commit branch event transaction: {err}"))
        })?;

        if let Err(err) = self
            .audit
            .append(
                &actor,
                self.storage
                    .fetch_run_tenant(run_uuid)
                    .await
                    .ok()
                    .flatten()
                    .and_then(|tenant| Uuid::parse_str(&tenant).ok()),
                "debug.branch.create",
                &json!({
                    "run_id": run_uuid.to_string(),
                    "child_run_id": new_run_uuid.to_string(),
                    "step_id": target_step.to_string(),
                }),
            )
            .await
        {
            warn!(error = %err, run = %run_uuid, "failed to append audit record for branch");
        }

        Ok(Response::new(runtime::CreateBranchRunResponse {
            run_id: new_run_uuid.to_string(),
        }))
    }

    async fn get_attestations(
        &self,
        request: Request<runtime::GetAttestationsRequest>,
    ) -> Result<Response<runtime::GetAttestationsResponse>, Status> {
        let metadata = request.metadata().clone();
        self.require_role(&metadata, Role::Reader).await?;
        let req = request.into_inner();
        let mut seen: HashSet<Uuid> = HashSet::new();
        let mut results = Vec::new();

        for raw in req.attestation_ids {
            let trimmed = raw.trim();
            if trimmed.is_empty() {
                continue;
            }
            let id = Uuid::parse_str(trimmed).map_err(|err| {
                invalid_argument(format!("invalid attestation id '{trimmed}': {err}"))
            })?;
            if !seen.insert(id) {
                continue;
            }
            match self.attestation_vault.fetch(id).await {
                Ok(Some(attestation)) => {
                    results.push(attestation_to_proto(attestation)?);
                }
                Ok(None) => {
                    // Ignore missing entries so clients can compare counts.
                }
                Err(err) => {
                    return Err(internal_error(format!(
                        "failed to load attestation {id}: {err}"
                    )));
                }
            }
        }

        Ok(Response::new(runtime::GetAttestationsResponse {
            attestations: results,
        }))
    }

    async fn sync_eval_pack(
        &self,
        request: Request<runtime::SyncEvalPackRequest>,
    ) -> Result<Response<runtime::SyncEvalPackResponse>, Status> {
        let metadata = request.metadata().clone();
        self.require_role(&metadata, Role::Writer).await?;

        let req = request.into_inner();
        let pack = req.pack.trim();
        if pack.is_empty() {
            return Err(invalid_argument("pack is required"));
        }
        let pack_name = pack.to_string();

        let mut scenario_ids = Vec::with_capacity(req.scenarios.len());
        for scenario in req.scenarios {
            let slug = scenario.slug.trim();
            if slug.is_empty() {
                return Err(invalid_argument("scenario slug cannot be empty"));
            }
            let run_spec_struct = scenario
                .run_spec
                .ok_or_else(|| invalid_argument("scenario run_spec must be provided"))?;
            let expected_struct = scenario
                .expected
                .ok_or_else(|| invalid_argument("scenario expected payload must be provided"))?;

            // prost_struct_to_json yields a serde_json::Value directly.
            let run_spec_json = prost_struct_to_json(&run_spec_struct);
            let expected_json = prost_struct_to_json(&expected_struct);

            let title = scenario.title.trim();
            let description = scenario.description.trim();

            let mut spec_map = Map::new();
            spec_map.insert("pack".to_string(), Value::String(pack_name.clone()));
            spec_map.insert("slug".to_string(), Value::String(slug.to_string()));
            if !title.is_empty() {
                spec_map.insert("title".to_string(), Value::String(title.to_string()));
            }
            if !description.is_empty() {
                spec_map.insert(
                    "description".to_string(),
                    Value::String(description.to_string()),
                );
            }
            spec_map.insert("run_spec".to_string(), run_spec_json);
            spec_map.insert("expected".to_string(), expected_json);
            let spec_value = Value::Object(spec_map);

            let scenario_name = format!("{pack}/{slug}");
            let scenario_id = self
                .storage
                .upsert_eval_scenario(
                    &scenario_name,
                    if description.is_empty() {
                        None
                    } else {
                        Some(description)
                    },
                    spec_value,
                )
                .await
                .map_err(|err| {
                    internal_error(format!(
                        "failed to upsert eval scenario '{scenario_name}': {err}"
                    ))
                })?;
            scenario_ids.push(scenario_id.to_string());
        }

        Ok(Response::new(runtime::SyncEvalPackResponse {
            scenario_ids,
        }))
    }

    async fn list_eval_scenarios(
        &self,
        request: Request<runtime::ListEvalScenariosRequest>,
    ) -> Result<Response<runtime::ListEvalScenariosResponse>, Status> {
        let metadata = request.metadata().clone();
        self.require_role(&metadata, Role::Reader).await?;

        let scenarios = self
            .storage
            .list_eval_scenarios()
            .await
            .map_err(|err| internal_error(format!("failed to list eval scenarios: {err}")))?;

        let mut proto_scenarios = Vec::with_capacity(scenarios.len());
        for scenario in scenarios {
            let spec = parse_eval_spec(&scenario.spec).map_err(|err| {
                internal_error(format!(
                    "invalid eval scenario '{}' spec: {err}",
                    scenario.name
                ))
            })?;
            let history_models = self
                .storage
                .list_eval_results(scenario.scenario_id, 20)
                .await
                .map_err(|err| internal_error(format!("failed to fetch eval results: {err}")))?;
            let mut history_proto = Vec::with_capacity(history_models.len());
            for result in history_models {
                history_proto.push(build_eval_result_proto(result)?);
            }
            proto_scenarios.push(build_eval_scenario_proto(&scenario, &spec, history_proto)?);
        }

        Ok(Response::new(runtime::ListEvalScenariosResponse {
            scenarios: proto_scenarios,
        }))
    }

    async fn record_eval_result(
        &self,
        request: Request<runtime::RecordEvalResultRequest>,
    ) -> Result<Response<runtime::RecordEvalResultResponse>, Status> {
        let metadata = request.metadata().clone();
        self.require_role(&metadata, Role::Writer).await?;

        let req = request.into_inner();
        let pack = req.pack.trim();
        let slug = req.slug.trim();
        if pack.is_empty() {
            return Err(invalid_argument("pack is required"));
        }
        if slug.is_empty() {
            return Err(invalid_argument("slug is required"));
        }

        let scenario_name = format!("{pack}/{slug}");
        let scenario = self
            .storage
            .fetch_eval_scenario_by_name(&scenario_name)
            .await
            .map_err(|err| internal_error(format!("failed to fetch eval scenario: {err}")))?
            .ok_or_else(|| {
                Status::not_found(format!("eval scenario '{scenario_name}' not found"))
            })?;

        let spec = parse_eval_spec(&scenario.spec).map_err(|err| {
            internal_error(format!(
                "invalid eval scenario '{}' spec: {err}",
                scenario.name
            ))
        })?;

        let run_uuid = Uuid::parse_str(&req.run_id)
            .map_err(|err| invalid_argument(format!("invalid run_id: {err}")))?;
        let run = self
            .storage
            .fetch_run(run_uuid)
            .await
            .map_err(|err| internal_error(format!("failed to fetch run: {err}")))?
            .ok_or_else(|| Status::not_found("run not found"))?;
        let steps = self
            .storage
            .fetch_steps_for_run(run_uuid)
            .await
            .map_err(|err| internal_error(format!("failed to fetch steps: {err}")))?;
        let last_event = self
            .storage
            .fetch_last_outbox_event(run_uuid)
            .await
            .map_err(|err| internal_error(format!("failed to fetch run events: {err}")))?;

        let previous = self
            .storage
            .latest_eval_result(scenario.scenario_id)
            .await
            .map_err(|err| {
                internal_error(format!("failed to fetch previous eval result: {err}"))
            })?;
        let previous_metrics = previous
            .as_ref()
            .filter(|result| result.run_id != run_uuid)
            .map(|result| &result.metrics);

        let commit = req.commit.trim();
        let commit = if commit.is_empty() {
            None
        } else {
            Some(commit.to_string())
        };

        let (outcome, metrics) = compute_eval_metrics(
            &spec,
            &run,
            &steps,
            last_event.as_ref().map(|event| event.created_at),
            previous_metrics,
            commit.clone(),
        )
        .map_err(|err| internal_error(format!("failed to compute eval metrics: {err}")))?;

        let stored = self
            .storage
            .insert_eval_result(scenario.scenario_id, run_uuid, &outcome, metrics)
            .await
            .map_err(|err| internal_error(format!("failed to persist eval result: {err}")))?;

        let proto = build_eval_result_proto(stored)?;
        Ok(Response::new(runtime::RecordEvalResultResponse {
            result: Some(proto),
        }))
    }

    async fn check_change_gate(
        &self,
        request: Request<runtime::CheckChangeGateRequest>,
    ) -> Result<Response<runtime::CheckChangeGateResponse>, Status> {
        let metadata = request.metadata().clone();
        let role = self.require_role(&metadata, Role::Writer).await?;
        let req = request.into_inner();

        let change_id = req.change_id.trim();
        if change_id.is_empty() {
            return Err(invalid_argument("change_id is required"));
        }

        let revision = req.revision.trim();
        let revision = if revision.is_empty() {
            None
        } else {
            Some(revision.to_string())
        };

        let actor_header = Self::actor_from_metadata(&metadata, role);
        let actor = if req.actor.trim().is_empty() {
            actor_header
        } else {
            req.actor.trim().to_string()
        };

        let diffs = req
            .diffs
            .into_iter()
            .map(|diff| GateDiff {
                path: diff.path,
                lines_added: diff.lines_added,
                lines_deleted: diff.lines_deleted,
                novelty_score: diff.novelty_score,
                risk_tags: diff.risk_tags,
                component: if diff.component.is_empty() {
                    None
                } else {
                    Some(diff.component)
                },
            })
            .collect();

        let coverage = req
            .coverage
            .map(|report| GateCoverageReport {
                overall_ratio: report.overall_ratio,
                minimum_ratio: report.minimum_ratio,
                components: report
                    .components
                    .into_iter()
                    .map(|component| GateCoverageComponent {
                        name: component.name,
                        coverage_ratio: component.coverage_ratio,
                        required_ratio: component.required_ratio,
                    })
                    .collect(),
            })
            .unwrap_or_default();

        let evals = req
            .evals
            .into_iter()
            .map(|metric| GateEvalMetric {
                name: metric.name,
                slug: if metric.slug.is_empty() {
                    None
                } else {
                    Some(metric.slug)
                },
                score: metric.score,
                threshold: metric.threshold,
                direction: match runtime::MetricDirection::try_from(metric.direction) {
                    Ok(runtime::MetricDirection::LowerIsBetter) => {
                        GateMetricDirection::LowerIsBetter
                    }
                    _ => GateMetricDirection::HigherIsBetter,
                },
                weight: metric.weight,
            })
            .collect();

        let budgets = req
            .budgets
            .into_iter()
            .map(|budget| GateBudgetSignal {
                name: budget.name,
                actual: budget.actual,
                limit: budget.limit,
            })
            .collect();

        let telemetry_value = struct_to_value(req.telemetry);
        let metadata_value = struct_to_value(req.metadata);

        let gate_request = GateRequest {
            change_id: change_id.to_string(),
            gate_id: None,
            revision: revision.clone(),
            diffs,
            coverage,
            evals,
            budgets,
            telemetry: telemetry_value.clone(),
            metadata: metadata_value.clone(),
            actor: Some(actor.clone()),
        };

        let mut decision = self
            .changeops
            .evaluate(&gate_request)
            .map_err(|err| invalid_argument(err.to_string()))?;

        let gate_id = Uuid::new_v4();
        let mut stored_request = gate_request.clone();
        stored_request.gate_id = Some(gate_id);

        let decision_payload = decision.to_value(&stored_request);
        let decision_bytes = serde_json::to_vec(&decision_payload).map_err(|err| {
            internal_error(format!("failed to serialise change gate decision: {err}"))
        })?;

        let artifact_meta = self
            .storage
            .artifacts()
            .put(
                Bytes::from(decision_bytes),
                "application/json",
                json!({
                    "kind": "change_gate_decision",
                    "change_id": change_id,
                    "effect": decision.effect.as_str(),
                    "gate_id": gate_id.to_string(),
                }),
            )
            .await
            .map_err(|err| {
                internal_error(format!("failed to persist change gate artifact: {err}"))
            })?;

        let mut attestation_set: HashSet<Uuid> =
            attestation_ids_from_value(&stored_request.telemetry)
                .into_iter()
                .collect();
        attestation_set.extend(attestation_ids_from_value(&decision_payload));
        let attestation_refs: Vec<Uuid> = attestation_set.into_iter().collect();

        if !attestation_refs.is_empty() && self.transparency_scope.includes_gates() {
            let artifact_sha = hex::encode(artifact_meta.sha256);
            if self.transparency_writer_enabled {
                let payload =
                    models::TransparencyJobPayload::ReleaseBundle(models::ReleaseBundlePayload {
                        change_id: change_id.to_string(),
                        gate_id,
                        artifact_sha256: artifact_sha.clone(),
                        attestation_ids: attestation_refs.clone(),
                        metadata: decision.metadata.clone(),
                    });
                match self.storage.enqueue_transparency_job(payload).await {
                    Ok(job_id) => {
                        let queued = json!({
                            "status": "queued",
                            "job_id": job_id,
                            "artifact_sha256": artifact_sha,
                            "attestation_ids": attestation_refs.iter().map(|id| id.to_string()).collect::<Vec<_>>(),
                        });
                        if let Value::Object(mut meta_obj) = decision.metadata.take() {
                            meta_obj.insert("scitt_entry".to_string(), queued);
                            decision.metadata = Value::Object(meta_obj);
                        } else {
                            decision.metadata = json!({ "scitt_entry": queued });
                        }
                    }
                    Err(err) => {
                        warn!(
                            change = change_id,
                            gate = %gate_id,
                            error = %err,
                            "failed to enqueue transparency job; falling back to synchronous publish"
                        );
                        self.publish_scitt_entry(
                            change_id,
                            gate_id,
                            &artifact_sha,
                            &attestation_refs,
                            &mut decision.metadata,
                        )
                        .await?;
                    }
                }
            } else {
                self.publish_scitt_entry(
                    change_id,
                    gate_id,
                    &artifact_sha,
                    &attestation_refs,
                    &mut decision.metadata,
                )
                .await?;
            }
        }

        let stored_gate = self
            .storage
            .insert_change_gate(models::NewChangeGate {
                gate_id,
                change_id: change_id.to_string(),
                revision: revision.clone(),
                effect: decision.effect.as_str().to_string(),
                reasons: serde_json::to_value(&decision.reasons).map_err(|err| {
                    internal_error(format!("failed to serialise change gate reasons: {err}"))
                })?,
                followups: serde_json::to_value(&decision.followups).map_err(|err| {
                    internal_error(format!("failed to serialise change gate followups: {err}"))
                })?,
                scorecard: serde_json::to_value(&decision.scorecard).map_err(|err| {
                    internal_error(format!("failed to serialise change gate scorecard: {err}"))
                })?,
                decided_at: decision.decided_at,
                decided_by: Some(actor.clone()),
                metadata: decision.metadata.clone(),
                input: serde_json::to_value(&stored_request).map_err(|err| {
                    internal_error(format!("failed to serialise change gate request: {err}"))
                })?,
                telemetry: stored_request.telemetry.clone(),
                artifact_sha256: Some(artifact_meta.sha256),
            })
            .await
            .map_err(|err| {
                internal_error(format!("failed to persist change gate decision: {err}"))
            })?;

        let audit_payload = decision_payload.clone();
        self.audit
            .append(&actor, None, "change_gate.check", &audit_payload)
            .await
            .map_err(|err| {
                internal_error(format!(
                    "failed to append audit trail for change gate: {err}"
                ))
            })?;

        tracing::info!(
            target: "fleetforge.changeops",
            gate = %gate_id,
            change = %change_id,
            effect = decision.effect.as_str(),
            reasons = ?decision.reasons,
            "change gate evaluated"
        );

        let proto_decision = build_proto_change_gate(
            stored_gate.gate_id,
            change_id,
            revision.as_deref(),
            &decision,
            &stored_gate.telemetry,
            &[],
        )?;

        Ok(Response::new(runtime::CheckChangeGateResponse {
            decision: Some(proto_decision),
        }))
    }

    async fn record_gate_followup(
        &self,
        request: Request<runtime::RecordGateFollowupRequest>,
    ) -> Result<Response<runtime::RecordGateFollowupResponse>, Status> {
        let metadata = request.metadata().clone();
        let role = self.require_role(&metadata, Role::Writer).await?;
        let req = request.into_inner();

        let gate_id = Uuid::parse_str(&req.gate_id)
            .map_err(|err| invalid_argument(format!("invalid gate_id: {err}")))?;

        let gate = self
            .storage
            .fetch_change_gate(gate_id)
            .await
            .map_err(|err| internal_error(format!("failed to fetch change gate: {err}")))?
            .ok_or_else(|| Status::not_found("change gate not found"))?;

        let actor_header = Self::actor_from_metadata(&metadata, role);
        let actor = if req.actor.trim().is_empty() {
            actor_header
        } else {
            req.actor.trim().to_string()
        };

        let outcome = runtime::FollowupOutcome::try_from(req.outcome)
            .unwrap_or(runtime::FollowupOutcome::Approved);
        let outcome_gate = proto_followup_outcome(outcome);

        let followup_id = Uuid::new_v4();
        let note = req.note;
        let details_value = struct_to_value(req.details);

        let stored_followup = self
            .storage
            .insert_change_gate_followup(models::NewChangeGateFollowup {
                followup_id,
                gate_id,
                actor: actor.clone(),
                outcome: outcome_gate.as_str().to_string(),
                note: note.clone(),
                details: details_value.clone(),
            })
            .await
            .map_err(|err| {
                internal_error(format!("failed to persist change gate follow-up: {err}"))
            })?;

        let record_payload = ChangeFollowupRecord {
            gate_id,
            actor: actor.clone(),
            outcome: outcome_gate,
            note: note.clone(),
            details: stored_followup.details.clone(),
            recorded_at: stored_followup.recorded_at,
        };

        self.audit
            .append(
                &actor,
                None,
                "change_gate.followup",
                &record_payload.to_value(),
            )
            .await
            .map_err(|err| {
                internal_error(format!(
                    "failed to append change gate follow-up audit: {err}"
                ))
            })?;

        tracing::info!(
            target: "fleetforge.changeops",
            gate = %gate_id,
            followup = %followup_id,
            outcome = %stored_followup.outcome,
            "change gate follow-up recorded"
        );

        let proto_followup = gate_followup_record_to_proto(&stored_followup);
        Ok(Response::new(runtime::RecordGateFollowupResponse {
            followup: Some(proto_followup),
        }))
    }

    async fn list_change_gates(
        &self,
        request: Request<runtime::ListChangeGatesRequest>,
    ) -> Result<Response<runtime::ListChangeGatesResponse>, Status> {
        let metadata = request.metadata().clone();
        self.require_role(&metadata, Role::Reader).await?;
        let req = request.into_inner();

        let change_id = req.change_id.trim();
        if change_id.is_empty() {
            return Err(invalid_argument("change_id is required"));
        }

        let limit = if req.limit == 0 { 5 } else { req.limit.min(50) } as i64;

        let gates = self
            .storage
            .list_change_gates_for_change(change_id, limit)
            .await
            .map_err(|err| {
                internal_error(format!("failed to fetch change gate decisions: {err}"))
            })?;

        let mut decisions = Vec::with_capacity(gates.len());
        for gate in gates {
            let followups = self
                .storage
                .fetch_change_gate_followups(gate.gate_id)
                .await
                .map_err(|err| {
                    internal_error(format!("failed to fetch change gate follow-ups: {err}"))
                })?;

            let effect = gate_effect_from_str(&gate.effect);
            let reasons = parse_reasons(&gate.reasons);
            let followup_actions = parse_followup_actions(&gate.followups);
            let scorecard = parse_scorecard(&gate.scorecard);

            let gate_decision = GateDecision {
                effect,
                reasons,
                followups: followup_actions,
                scorecard,
                decided_at: gate.decided_at,
                metadata: gate.metadata.clone(),
            };

            let proto_decision = build_proto_change_gate(
                gate.gate_id,
                &gate.change_id,
                gate.revision.as_deref(),
                &gate_decision,
                &gate.telemetry,
                &followups,
            )?;
            decisions.push(proto_decision);
        }

        Ok(Response::new(runtime::ListChangeGatesResponse {
            decisions,
        }))
    }
}

/// Starts the API servers (gRPC + HTTP) backed by the provided storage handle.
#[instrument(skip_all)]
pub async fn start_api_server(storage: Arc<Storage>, audit: Arc<Audit>) -> Result<()> {
    let auth = Arc::new(AuthHandle::from_env().await?);

    let endpoints = resolve_endpoint_config()?;
    let cors_config = cors_config_from_env();
    let transparency = transparency_log_from_env(Arc::clone(&storage))?;
    let transparency_scope = transparency_scope_from_env();
    let transparency_writer_enabled = transparency_writer_enabled_flag();
    let writer_interval = transparency_writer_interval();

    if transparency_writer_enabled && !matches!(transparency_scope, TransparencyScope::Disabled) {
        info!(
            "transparency writer enabled (interval={}s)",
            writer_interval.as_secs()
        );
        let writer_storage = Arc::clone(&storage);
        let writer_log = Arc::clone(&transparency);
        tokio::spawn(async move {
            let writer = TransparencyWriter::new(writer_storage, writer_log, writer_interval);
            writer.run().await;
        });
    }

    let cors_summary = cors_config
        .as_ref()
        .map(|cfg| {
            if cfg.display.is_empty() {
                "[]".to_string()
            } else {
                cfg.display.join(", ")
            }
        })
        .unwrap_or_else(|| "disabled".to_string());

    info!(
        grpc_endpoint = %format!("http://{}", endpoints.grpc_addr),
        http_endpoint = %format!("http://{}", endpoints.http_addr),
        cors_allowlist = %cors_summary,
        "runtime API configuration ready"
    );

    if endpoints.combined {
        run_combined_server(
            endpoints.grpc_addr,
            storage,
            auth,
            audit,
            transparency,
            transparency_writer_enabled,
            transparency_scope,
            cors_config,
        )
        .await
    } else {
        let grpc_storage = Arc::clone(&storage);
        let http_storage = Arc::clone(&storage);
        let http_auth = Arc::clone(&auth);
        let grpc_transparency = Arc::clone(&transparency);
        tokio::try_join!(
            run_grpc_server(
                endpoints.grpc_addr,
                grpc_storage,
                Arc::clone(&auth),
                Arc::clone(&audit),
                grpc_transparency,
                transparency_writer_enabled,
                transparency_scope,
                cors_config.clone(),
            ),
            run_http_server(
                endpoints.http_addr,
                http_storage,
                http_auth,
                Arc::clone(&audit),
                cors_config,
            ),
        )?;
        Ok(())
    }
}

#[derive(Clone, Copy, Debug)]
struct EndpointConfig {
    grpc_addr: SocketAddr,
    http_addr: SocketAddr,
    combined: bool,
}

fn resolve_endpoint_config() -> Result<EndpointConfig> {
    if let Ok(port_raw) = std::env::var("PORT") {
        let port: u16 = port_raw.parse().context("invalid PORT value")?;
        let addr = SocketAddr::from(([0, 0, 0, 0], port));
        return Ok(EndpointConfig {
            grpc_addr: addr,
            http_addr: addr,
            combined: true,
        });
    }

    let grpc_addr: SocketAddr = std::env::var("FLEETFORGE_API_ADDR")
        .unwrap_or_else(|_| "0.0.0.0:50051".to_string())
        .parse()
        .context("invalid FLEETFORGE_API_ADDR")?;
    let http_addr: SocketAddr = std::env::var("FLEETFORGE_HTTP_ADDR")
        .unwrap_or_else(|_| "0.0.0.0:8080".to_string())
        .parse()
        .context("invalid FLEETFORGE_HTTP_ADDR")?;

    Ok(EndpointConfig {
        grpc_addr,
        http_addr,
        combined: grpc_addr == http_addr,
    })
}

#[instrument(skip_all)]
async fn run_combined_server(
    addr: SocketAddr,
    storage: Arc<Storage>,
    auth: Arc<AuthHandle>,
    audit: Arc<Audit>,
    transparency: Arc<dyn TransparencyLog>,
    transparency_writer_enabled: bool,
    transparency_scope: TransparencyScope,
    cors: Option<CorsConfig>,
) -> Result<()> {
    let listener = TcpListener::bind(addr)
        .await
        .with_context(|| format!("failed to bind listener on {addr}"))?;

    let attestation_vault: Arc<dyn AttestationVault> = Arc::new(ObjectStoreAttestationVault::new(
        storage.object_store(),
        storage.pool().clone(),
    ));
    let runtime_api = RuntimeApiService::new(
        Arc::clone(&storage),
        Arc::clone(&auth),
        Arc::clone(&audit),
        Arc::clone(&attestation_vault),
        Arc::clone(&transparency),
        transparency_writer_enabled,
        transparency_scope,
    );
    let tap_api = TapApiService::new(Arc::clone(&storage), Arc::clone(&auth));

    let (mut health_reporter, health_service) = health_reporter();
    health_reporter
        .set_service_status(
            "fleetforge.runtime.v1.RuntimeService",
            tonic_health::ServingStatus::Serving,
        )
        .await; // API returns () so there is nothing to convert via map_err.
    health_reporter
        .set_service_status(
            "fleetforge.tap.v2.TapService",
            tonic_health::ServingStatus::Serving,
        )
        .await;
    health_reporter
        .set_service_status("", tonic_health::ServingStatus::Serving)
        .await;

    let runtime_service = RuntimeServiceServer::new(runtime_api);
    let runtime_service = tonic_web::enable(runtime_service);
    let tap_service = TapServiceServer::new(tap_api);
    let tap_service = tonic_web::enable(tap_service);

    let reflection = ReflectionBuilder::configure()
        .register_encoded_file_descriptor_set(runtime::FILE_DESCRIPTOR_SET)
        .build()
        .context("failed to build reflection service")?;

    let cors_layer = cors.as_ref().map(|cfg| cfg.to_layer());
    let http_router = {
        let router = http::router(Arc::clone(&storage), Arc::clone(&auth), Arc::clone(&audit));
        match cors_layer.clone() {
            Some(layer) => router.layer(layer.clone()),
            None => router,
        }
    };
    let mut builder = Server::builder().accept_http1(true);

    let grpc_service = builder
        .add_service(health_service)
        .add_service(reflection)
        .add_service(runtime_service)
        .add_service(tap_service)
        .into_service();

    let grpc_service = BoxCloneService::new(grpc_service);

    info!(%addr, "starting runtime API server (single listener)");

    let make_service = make_service_fn(move |_conn| {
        let http_router = http_router.clone();
        let grpc_service = grpc_service.clone();

        async move {
            Ok::<_, Infallible>(service_fn(move |req: HyperRequest<Body>| {
                let http_router = http_router.clone();
                let mut grpc_service = grpc_service.clone();

                async move {
                    if is_grpc_request(&req) {
                        tower::Service::call(&mut grpc_service, req).await
                    } else {
                        let router = http_router.clone();
                        let response = match router.oneshot(req).await {
                            Ok(resp) => resp,
                            Err(err) => match err {},
                        }
                        .map(|body| {
                            body.map_err(|err| Status::internal(err.to_string()))
                                .boxed_unsync()
                        });
                        Ok::<_, Box<dyn std::error::Error + Send + Sync>>(response)
                    }
                }
            }))
        }
    });

    let std_listener = listener
        .into_std()
        .context("failed to convert listener for hyper server")?;
    HyperServer::from_tcp(std_listener)
        .context("failed to build runtime API server")?
        .serve(make_service)
        .await
        .context("runtime API server exited unexpectedly")
}

fn is_grpc_request(req: &HyperRequest<Body>) -> bool {
    if let Some(content_type) = req
        .headers()
        .get(header::CONTENT_TYPE)
        .and_then(|value| value.to_str().ok())
    {
        let lower = content_type.to_ascii_lowercase();
        if lower.starts_with("application/grpc")
            || lower.starts_with("application/grpc-web")
            || lower.starts_with("application/connect+")
        {
            return true;
        }
    }

    let path = req.uri().path();
    path.starts_with("/fleetforge.runtime.v1.")
        || path.starts_with("/fleetforge.tap.v2.")
        || path.starts_with("/grpc.health.v1.")
        || path.starts_with("/grpc.reflection.v1alpha.")
}

#[instrument(skip_all)]
async fn run_grpc_server(
    addr: SocketAddr,
    storage: Arc<Storage>,
    auth: Arc<AuthHandle>,
    audit: Arc<Audit>,
    transparency: Arc<dyn TransparencyLog>,
    transparency_writer_enabled: bool,
    transparency_scope: TransparencyScope,
    cors: Option<CorsConfig>,
) -> Result<()> {
    let attestation_vault: Arc<dyn AttestationVault> = Arc::new(ObjectStoreAttestationVault::new(
        storage.object_store(),
        storage.pool().clone(),
    ));
    let runtime_api = RuntimeApiService::new(
        Arc::clone(&storage),
        Arc::clone(&auth),
        Arc::clone(&audit),
        Arc::clone(&attestation_vault),
        transparency,
        transparency_writer_enabled,
        transparency_scope,
    );
    let tap_api = TapApiService::new(storage, auth);

    let (mut health_reporter, health_service) = health_reporter();
    let reflection = ReflectionBuilder::configure()
        .register_encoded_file_descriptor_set(runtime::FILE_DESCRIPTOR_SET)
        .build()
        .context("failed to build reflection service")?;

    health_reporter
        .set_service_status(
            "fleetforge.runtime.v1.RuntimeService",
            tonic_health::ServingStatus::Serving,
        )
        .await; // set_service_status returns ()
    health_reporter
        .set_service_status(
            "fleetforge.tap.v2.TapService",
            tonic_health::ServingStatus::Serving,
        )
        .await;
    health_reporter
        .set_service_status("", tonic_health::ServingStatus::Serving)
        .await;

    info!(%addr, "starting runtime gRPC API server");

    let runtime_service = RuntimeServiceServer::new(runtime_api);
    let runtime_service = tonic_web::enable(runtime_service);
    let tap_service = TapServiceServer::new(tap_api);
    let tap_service = tonic_web::enable(tap_service);
    let cors_layer = cors.as_ref().map(|cfg| cfg.to_layer());
    Server::builder()
        .accept_http1(true)
        .add_service(health_service)
        .add_service(reflection)
        .add_service(runtime_service)
        .add_service(tap_service)
        .serve(addr)
        .await
        .context("runtime gRPC API server exited unexpectedly")
}

#[instrument(skip_all)]
async fn run_http_server(
    addr: SocketAddr,
    storage: Arc<Storage>,
    auth: Arc<AuthHandle>,
    audit: Arc<Audit>,
    cors: Option<CorsConfig>,
) -> Result<()> {
    let listener = TcpListener::bind(addr)
        .await
        .with_context(|| format!("failed to bind HTTP listener on {addr}"))?;

    // Allow explicit opt-in for local UI origins; default is no cross-origin access.
    let cors_layer = cors.as_ref().map(|cfg| cfg.to_layer());
    let router = http::router(storage, auth, audit);
    let app = match cors_layer {
        Some(layer) => router.layer(layer),
        None => router,
    };

    info!(%addr, "starting runtime HTTP API server");

    let std_listener = listener
        .into_std()
        .context("failed to convert HTTP listener for hyper server")?;
    HyperServer::from_tcp(std_listener)
        .context("failed to build runtime HTTP server")?
        .serve(app.into_make_service())
        .await
        .context("runtime HTTP API server exited unexpectedly")
}

#[derive(Clone, Debug)]
struct CorsConfig {
    origins: Vec<HeaderValue>,
    display: Vec<String>,
}

impl CorsConfig {
    fn to_layer(&self) -> CorsLayer {
        CorsLayer::new()
            .allow_origin(self.origins.clone())
            .allow_methods([Method::GET, Method::POST, Method::OPTIONS])
            .allow_headers([header::CONTENT_TYPE, header::AUTHORIZATION])
    }
}

fn cors_config_from_env() -> Option<CorsConfig> {
    let raw = std::env::var("FLEETFORGE_CORS_ALLOW_ORIGINS").ok()?;
    let mut origins = Vec::new();
    let mut display = Vec::new();
    for origin in raw.split(',') {
        let trimmed = origin.trim();
        if trimmed.is_empty() {
            continue;
        }
        match HeaderValue::from_str(trimmed) {
            Ok(value) => {
                origins.push(value);
                display.push(trimmed.to_string());
            }
            Err(err) => {
                warn!(origin = %trimmed, %err, "invalid origin in FLEETFORGE_CORS_ALLOW_ORIGINS")
            }
        }
    }

    if origins.is_empty() {
        warn!(
            "no valid origins parsed from FLEETFORGE_CORS_ALLOW_ORIGINS; CORS will remain disabled"
        );
        return None;
    }

    Some(CorsConfig { origins, display })
}

fn transparency_writer_enabled_flag() -> bool {
    let enabled = matches!(
        std::env::var(TRANSPARENCY_WRITER_FLAG)
            .ok()
            .map(|value| value.trim().to_ascii_lowercase())
            .unwrap_or_default()
            .as_str(),
        "1" | "true" | "yes" | "on"
    );
    if enabled && !feature_allowed(LicensedFeature::ScittTransparency) {
        warn!(
            "FLEETFORGE_TRANSPARENCY_WRITER=1 but the current license tier does not include SCITT publishing; 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,
    }
}

fn transparency_writer_interval() -> Duration {
    let interval = std::env::var(TRANSPARENCY_WRITER_INTERVAL_ENV)
        .ok()
        .and_then(|value| value.parse::<u64>().ok())
        .unwrap_or(TRANSPARENCY_WRITER_DEFAULT_INTERVAL_SECS);
    Duration::from_secs(interval.max(TRANSPARENCY_WRITER_INTERVAL_MIN_SECS))
}

fn parse_dag_json(input: &str) -> Result<Value, Status> {
    serde_json::from_str(input)
        .map_err(|err| invalid_argument(format!("invalid dag_json payload: {err}")))
}

fn extract_priority(step_obj: &Map<String, Value>) -> i16 {
    step_obj
        .get("policy")
        .and_then(Value::as_object)
        .and_then(|policy| policy.get("priority"))
        .and_then(Value::as_i64)
        .unwrap_or(0)
        .clamp(i16::MIN as i64, i16::MAX as i64) as i16
}

fn build_run_detail(
    run: models::Run,
    steps: Vec<models::Step>,
) -> Result<runtime::RunDetail, Status> {
    let (inputs_value, labels_map) = unpack_input_ctx(&run.input_ctx);
    let breakpoint_specs = extract_breakpoint_specs(&run.input_ctx);
    let inputs_struct = match &inputs_value {
        Value::Null => None,
        Value::Object(map) if map.is_empty() => None,
        Value::Object(_) => Some(json_to_prost_struct(&inputs_value).map_err(|err| {
            internal_error(format!("failed to encode inputs for response: {err}"))
        })?),
        _ => None,
    };

    let dag_json = serde_json::to_string(&run.dag_json).map_err(|err| {
        internal_error(format!("failed to serialise dag_json for response: {err}"))
    })?;

    let run_spec = runtime::RunSpec {
        dag_json,
        inputs: inputs_struct,
        seed: run.seed,
        labels: labels_map,
        breakpoints: breakpoint_specs,
    };

    let step_summaries = steps
        .into_iter()
        .map(step_to_summary)
        .collect::<Result<Vec<_>, Status>>()?;

    Ok(runtime::RunDetail {
        run_id: run.run_id.to_string(),
        status: map_run_status(run.status) as i32,
        spec: Some(run_spec),
        created_at: Some(to_timestamp(run.created_at)),
        updated_at: None,
        steps: step_summaries,
    })
}

fn unpack_input_ctx(ctx: &Value) -> (Value, HashMap<String, String>) {
    match ctx {
        Value::Object(map) => {
            let inputs = map
                .get("inputs")
                .cloned()
                .unwrap_or_else(|| Value::Object(Map::new()));
            let labels = map
                .get("labels")
                .map(labels_value_to_map)
                .unwrap_or_default();
            (inputs, labels)
        }
        other => (other.clone(), HashMap::new()),
    }
}

fn extract_breakpoint_specs(ctx: &Value) -> Vec<String> {
    let Value::Object(map) = ctx else {
        return Vec::new();
    };
    let Some(Value::Object(debug)) = map.get("debug") else {
        return Vec::new();
    };

    if let Some(Value::Array(specs)) = debug.get("breakpoint_specs") {
        return specs
            .iter()
            .filter_map(|value| value.as_str().map(|s| s.to_string()))
            .collect();
    }

    if let Some(Value::Array(entries)) = debug.get("breakpoints") {
        return entries
            .iter()
            .filter_map(|entry| {
                entry
                    .get("spec")
                    .and_then(Value::as_str)
                    .map(|s| s.to_string())
            })
            .collect();
    }

    Vec::new()
}

fn load_debug_breakpoints(debug: &Map<String, Value>) -> Vec<(String, Uuid)> {
    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_str = obj.get("step_id")?.as_str()?;
                    let step_id = Uuid::parse_str(step_id_str).ok()?;
                    Some((spec, step_id))
                })
                .collect()
        })
        .unwrap_or_default()
}

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

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

fn load_slug_map(debug: &Map<String, Value>) -> HashMap<String, Uuid> {
    debug
        .get("slugs")
        .and_then(Value::as_object)
        .map(|map| {
            map.iter()
                .filter_map(|(slug, value)| {
                    value
                        .as_str()
                        .and_then(|id| Uuid::parse_str(id).ok())
                        .map(|uuid| (slug.clone(), uuid))
                })
                .collect()
        })
        .unwrap_or_default()
}

fn store_slug_map(debug: &mut Map<String, Value>, entries: &[(String, Uuid)]) {
    let mut slugs = Map::new();
    for (slug, step_id) in entries {
        slugs.insert(slug.clone(), Value::String(step_id.to_string()));
    }
    debug.insert("slugs".to_string(), Value::Object(slugs));
}

fn extract_active_pause_step_id(debug: &Map<String, Value>) -> Option<Uuid> {
    debug
        .get("active_pause")
        .and_then(|value| value.as_object())
        .and_then(|obj| obj.get("step_id"))
        .and_then(Value::as_str)
        .and_then(|s| Uuid::parse_str(s).ok())
}

fn merge_json_object(target: &mut Map<String, Value>, patch: &Map<String, Value>) {
    for (key, value) in patch {
        target.insert(key.clone(), value.clone());
    }
}

fn collect_step_ids(run: &models::Run) -> HashSet<Uuid> {
    run.dag_json
        .get("steps")
        .and_then(Value::as_array)
        .map(|steps| {
            steps
                .iter()
                .filter_map(|step| {
                    step.get("id")
                        .and_then(Value::as_str)
                        .and_then(|s| Uuid::parse_str(s).ok())
                })
                .collect()
        })
        .unwrap_or_default()
}

fn collect_slug_entries(run: &models::Run) -> Vec<(String, Uuid)> {
    run.dag_json
        .get("steps")
        .and_then(Value::as_array)
        .map(|steps| {
            steps
                .iter()
                .filter_map(|step| {
                    let id = step
                        .get("id")
                        .and_then(Value::as_str)
                        .and_then(|s| Uuid::parse_str(s).ok())?;
                    let slug = step
                        .get("slug")
                        .and_then(Value::as_str)
                        .map(|s| s.trim().to_string())
                        .filter(|s| !s.is_empty())?;
                    Some((slug, id))
                })
                .collect()
        })
        .unwrap_or_default()
}

fn build_branch_dag(dag_json: &Value, start: Uuid) -> Result<Value, SubmitRunError> {
    let dag_map = dag_json
        .as_object()
        .cloned()
        .ok_or_else(|| SubmitRunError::invalid("dag_json must be a JSON object"))?;
    let steps_array = dag_map
        .get("steps")
        .and_then(Value::as_array)
        .ok_or_else(|| SubmitRunError::invalid("dag_json.steps must be an array"))?;
    let edges_array = dag_map
        .get("edges")
        .and_then(Value::as_array)
        .ok_or_else(|| SubmitRunError::invalid("dag_json.edges must be an array"))?;

    let mut step_defs: HashMap<Uuid, Value> = HashMap::new();
    for step in steps_array {
        let step_obj = step
            .as_object()
            .ok_or_else(|| SubmitRunError::invalid("each step must be an object"))?;
        let id = step_obj
            .get("id")
            .and_then(Value::as_str)
            .ok_or_else(|| SubmitRunError::invalid("step.id missing"))?;
        let step_id = Uuid::parse_str(id)
            .map_err(|err| SubmitRunError::invalid(format!("invalid step.id: {err}")))?;
        step_defs.insert(step_id, step.clone());
    }

    if !step_defs.contains_key(&start) {
        return Err(SubmitRunError::invalid("from_step references unknown step"));
    }

    let mut adjacency: HashMap<Uuid, Vec<Uuid>> = HashMap::new();
    for edge in edges_array {
        let pair = edge
            .as_array()
            .ok_or_else(|| SubmitRunError::invalid("edges entries must be arrays"))?;
        if pair.len() != 2 {
            return Err(SubmitRunError::invalid(
                "edges entries must contain two members",
            ));
        }
        let from = pair[0]
            .as_str()
            .ok_or_else(|| SubmitRunError::invalid("edge source must be a string"))?;
        let to = pair[1]
            .as_str()
            .ok_or_else(|| SubmitRunError::invalid("edge target must be a string"))?;
        let from_id = Uuid::parse_str(from)
            .map_err(|err| SubmitRunError::invalid(format!("invalid edge source: {err}")))?;
        let to_id = Uuid::parse_str(to)
            .map_err(|err| SubmitRunError::invalid(format!("invalid edge target: {err}")))?;
        adjacency.entry(from_id).or_default().push(to_id);
    }

    let mut queue = VecDeque::new();
    let mut visited: HashSet<Uuid> = HashSet::new();
    queue.push_back(start);
    while let Some(node) = queue.pop_front() {
        if !visited.insert(node) {
            continue;
        }
        if let Some(neighbors) = adjacency.get(&node) {
            for next in neighbors {
                queue.push_back(*next);
            }
        }
    }

    let filtered_steps = steps_array
        .iter()
        .filter_map(|step| {
            let id = step
                .get("id")
                .and_then(Value::as_str)
                .and_then(|s| Uuid::parse_str(s).ok());
            match id {
                Some(step_id) if visited.contains(&step_id) => Some(step.clone()),
                _ => None,
            }
        })
        .collect::<Vec<_>>();

    let filtered_edges = edges_array
        .iter()
        .filter(|edge| {
            if let Some(pair) = edge.as_array() {
                if pair.len() == 2 {
                    if let (Some(from), Some(to)) = (
                        pair[0].as_str().and_then(|s| Uuid::parse_str(s).ok()),
                        pair[1].as_str().and_then(|s| Uuid::parse_str(s).ok()),
                    ) {
                        return visited.contains(&from) && visited.contains(&to);
                    }
                }
            }
            false
        })
        .cloned()
        .collect::<Vec<_>>();

    let mut new_dag = dag_json.as_object().cloned().unwrap_or_default();
    new_dag.insert("steps".to_string(), Value::Array(filtered_steps));
    new_dag.insert("edges".to_string(), Value::Array(filtered_edges));
    Ok(Value::Object(new_dag))
}

fn collect_prompt_packs_json(dag_json: &Value) -> Vec<String> {
    match dag_json {
        Value::Object(map) => {
            let packs = map.get("prompt_packs").or_else(|| map.get("promptPacks"));
            match packs {
                Some(Value::Array(items)) => items
                    .iter()
                    .filter_map(|item| item.as_str().map(|s| s.to_string()))
                    .collect(),
                Some(Value::String(value)) => vec![value.clone()],
                _ => Vec::new(),
            }
        }
        _ => Vec::new(),
    }
}

struct DriftComputation {
    token_delta: Option<f64>,
    cost_delta: Option<f64>,
    within_tolerance: bool,
}

const DEFAULT_TOKEN_TOLERANCE: f64 = 0.0;
const DEFAULT_COST_TOLERANCE: f64 = 0.0;
const TOKEN_KEYS: &[&str] = &["tokens", "token_count", "total_tokens"];
const COST_KEYS: &[&str] = &["cost", "cost_usd", "total_cost"];
const SANITIZE_KEYS: &[&str] = &[
    "tokens",
    "token_count",
    "total_tokens",
    "cost",
    "cost_usd",
    "total_cost",
];

fn wrap_attempt_output(attempt: Option<&models::StepAttempt>, output: &Value) -> Value {
    match attempt {
        Some(record) => json!({
            "attempt": record.attempt,
            "status": record.status.as_str(),
            "recorded_at": record.recorded_at,
            "output": output,
        }),
        None => output.clone(),
    }
}

fn compute_drift(
    recorded: &Value,
    current: &Value,
    token_tolerance: f64,
    cost_tolerance: f64,
) -> DriftComputation {
    let token_delta = diff_numeric(recorded, current, TOKEN_KEYS);
    let cost_delta = diff_numeric(recorded, current, COST_KEYS);

    let sanitized_recorded = strip_keys(recorded, SANITIZE_KEYS);
    let sanitized_current = strip_keys(current, SANITIZE_KEYS);

    let tokens_within = token_delta
        .map(|delta| delta.abs() <= token_tolerance)
        .unwrap_or(true);
    let cost_within = cost_delta
        .map(|delta| delta.abs() <= cost_tolerance)
        .unwrap_or(true);
    let structure_equal = sanitized_recorded == sanitized_current;

    DriftComputation {
        token_delta,
        cost_delta,
        within_tolerance: tokens_within && cost_within && structure_equal,
    }
}

fn diff_numeric(recorded: &Value, current: &Value, keys: &[&str]) -> Option<f64> {
    let recorded_value = find_numeric(recorded, keys)?;
    let current_value = find_numeric(current, keys)?;
    Some(current_value - recorded_value)
}

fn find_numeric(value: &Value, keys: &[&str]) -> Option<f64> {
    match value {
        Value::Object(map) => {
            for key in keys {
                if let Some(v) = map.get(*key) {
                    if let Some(num) = v.as_f64() {
                        return Some(num);
                    }
                }
            }
            for v in map.values() {
                if let Some(found) = find_numeric(v, keys) {
                    return Some(found);
                }
            }
            None
        }
        Value::Array(items) => {
            for item in items {
                if let Some(found) = find_numeric(item, keys) {
                    return Some(found);
                }
            }
            None
        }
        _ => None,
    }
}

fn strip_keys(value: &Value, keys: &[&str]) -> Value {
    match value {
        Value::Object(map) => {
            let mut new_map = serde_json::Map::new();
            for (k, v) in map {
                if keys.iter().any(|key| key == k) {
                    continue;
                }
                new_map.insert(k.clone(), strip_keys(v, keys));
            }
            Value::Object(new_map)
        }
        Value::Array(items) => Value::Array(items.iter().map(|v| strip_keys(v, keys)).collect()),
        _ => value.clone(),
    }
}

fn step_to_summary(step: models::Step) -> Result<runtime::StepSummary, Status> {
    let step_type = step
        .spec_json
        .get("type")
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_string();

    let output = match step.output_json {
        Some(value) => Some(json_to_prost_value(&value).map_err(|err| {
            internal_error(format!(
                "invalid output payload for step {}: {err}",
                step.step_id
            ))
        })?),
        None => None,
    };

    let error = match step.error_json {
        Some(value) => Some(json_to_prost_value(&value).map_err(|err| {
            internal_error(format!(
                "invalid error payload for step {}: {err}",
                step.step_id
            ))
        })?),
        None => None,
    };

    Ok(runtime::StepSummary {
        step_id: step.step_id.to_string(),
        idx: step.idx,
        r#type: step_type,
        status: step.status.as_str().to_string(),
        attempt: step.attempt,
        leased_at: None,
        completed_at: None,
        output,
        error,
    })
}

fn run_event_from_outbox(event: models::OutboxEvent) -> Result<runtime::RunEvent, Status> {
    let payload = json_to_prost_value(&event.payload).map_err(|err| {
        internal_error(format!(
            "failed to encode outbox payload for event {}: {err}",
            event.id
        ))
    })?;

    Ok(runtime::RunEvent {
        run_id: event.run_id.to_string(),
        step_id: event.step_id.map(|id| id.to_string()).unwrap_or_default(),
        kind: event.kind,
        payload: Some(payload),
        occurred_at: Some(to_timestamp(event.created_at)),
    })
}

fn map_run_status(status: models::RunStatus) -> runtime::RunStatus {
    match status {
        models::RunStatus::Pending => runtime::RunStatus::Pending,
        models::RunStatus::Running => runtime::RunStatus::Running,
        models::RunStatus::Succeeded => runtime::RunStatus::Succeeded,
        models::RunStatus::Failed => runtime::RunStatus::Failed,
        models::RunStatus::Canceled => runtime::RunStatus::Canceled,
        models::RunStatus::PausedAtStep => runtime::RunStatus::PausedAtStep,
    }
}

#[derive(Clone, Debug)]
pub(crate) struct EvalScenarioSpecData {
    pack: String,
    slug: String,
    title: Option<String>,
    description: Option<String>,
    run_spec: Value,
    expected: Value,
}

pub(crate) fn parse_eval_spec(spec: &Value) -> anyhow::Result<EvalScenarioSpecData> {
    let pack = spec
        .get("pack")
        .and_then(Value::as_str)
        .ok_or_else(|| anyhow!("scenario spec missing 'pack' field"))?
        .to_string();
    let slug = spec
        .get("slug")
        .and_then(Value::as_str)
        .ok_or_else(|| anyhow!("scenario spec missing 'slug' field"))?
        .to_string();
    let title = spec
        .get("title")
        .and_then(Value::as_str)
        .map(|value| value.to_string());
    let description = spec
        .get("description")
        .and_then(Value::as_str)
        .map(|value| value.to_string());
    let run_spec = spec
        .get("run_spec")
        .cloned()
        .ok_or_else(|| anyhow!("scenario spec missing 'run_spec' payload"))?;
    if !run_spec.is_object() {
        return Err(anyhow!("scenario spec 'run_spec' must be a JSON object"));
    }
    let expected = spec
        .get("expected")
        .cloned()
        .ok_or_else(|| anyhow!("scenario spec missing 'expected' payload"))?;
    if !expected.is_object() {
        return Err(anyhow!("scenario spec 'expected' must be a JSON object"));
    }

    Ok(EvalScenarioSpecData {
        pack,
        slug,
        title,
        description,
        run_spec,
        expected,
    })
}

fn metric_summary(current: f64, previous: Option<f64>) -> Value {
    let delta = previous.map(|p| current - p);
    json!({
        "value": current,
        "previous": previous,
        "delta": delta,
    })
}

fn extract_metric(metrics: &Value, key: &str) -> Option<f64> {
    metrics
        .as_object()
        .and_then(|map| map.get(key))
        .and_then(|entry| entry.get("value"))
        .and_then(Value::as_f64)
}

fn compute_eval_metrics(
    spec: &EvalScenarioSpecData,
    run: &models::Run,
    steps: &[models::Step],
    last_event_time: Option<DateTime<Utc>>,
    previous_metrics: Option<&Value>,
    commit: Option<String>,
) -> anyhow::Result<(String, Value)> {
    let tokens_actual: f64 = steps
        .iter()
        .map(|step| {
            step.actual_tokens
                .map(|value| value as f64)
                .unwrap_or_else(|| step.estimated_tokens as f64)
        })
        .sum();
    let cost_actual: f64 = steps
        .iter()
        .map(|step| step.actual_cost.unwrap_or(step.estimated_cost))
        .sum();

    let run_end = last_event_time.unwrap_or(run.created_at);
    let latency_ms = run_end
        .signed_duration_since(run.created_at)
        .num_milliseconds()
        .max(0) as f64;

    let previous_tokens = previous_metrics.and_then(|metrics| extract_metric(metrics, "tokens"));
    let previous_cost = previous_metrics.and_then(|metrics| extract_metric(metrics, "cost_usd"));
    let previous_latency =
        previous_metrics.and_then(|metrics| extract_metric(metrics, "latency_ms"));

    let mut differences: Vec<Value> = Vec::new();
    if let Some(expected_status) = spec.expected.get("status").and_then(Value::as_str) {
        let actual_status = run.status.as_str();
        if !expected_status.eq_ignore_ascii_case(actual_status) {
            differences.push(json!({
                "type": "status",
                "expected": expected_status,
                "actual": actual_status,
            }));
        }
    }

    let mut actual_steps: HashMap<String, Value> = HashMap::with_capacity(steps.len());
    for step in steps {
        if let Some(output) = step.output_json.clone() {
            actual_steps.insert(step.step_id.to_string(), output);
        }
    }

    if let Some(expected_steps) = spec.expected.get("steps").and_then(Value::as_object) {
        for (step_id, expected_output) in expected_steps {
            match actual_steps.get(step_id) {
                Some(actual_output) if actual_output == expected_output => {}
                Some(actual_output) => {
                    differences.push(json!({
                        "type": "step",
                        "step_id": step_id,
                        "expected": expected_output,
                        "actual": actual_output,
                    }));
                }
                None => {
                    differences.push(json!({
                        "type": "step_missing",
                        "step_id": step_id,
                        "expected": expected_output,
                    }));
                }
            }
        }
    }

    let quality_matched = differences.is_empty();
    let outcome = if quality_matched { "passed" } else { "failed" }.to_string();

    let mut metrics_map = Map::new();
    if let Some(commit) = commit {
        if !commit.is_empty() {
            metrics_map.insert("commit".to_string(), Value::String(commit));
        }
    }
    metrics_map.insert(
        "tokens".to_string(),
        metric_summary(tokens_actual, previous_tokens),
    );
    metrics_map.insert(
        "cost_usd".to_string(),
        metric_summary(cost_actual, previous_cost),
    );
    metrics_map.insert(
        "latency_ms".to_string(),
        metric_summary(latency_ms, previous_latency),
    );
    metrics_map.insert(
        "quality".to_string(),
        Value::Object({
            let mut quality = Map::new();
            quality.insert("matched".to_string(), Value::Bool(quality_matched));
            quality.insert("differences".to_string(), Value::Array(differences));
            quality
        }),
    );

    Ok((outcome, Value::Object(metrics_map)))
}

fn build_eval_result_proto(result: models::EvalResult) -> Result<runtime::EvalResult, Status> {
    let metrics_struct = json_to_prost_struct(&result.metrics)
        .map_err(|err| internal_error(format!("invalid eval metrics payload: {err}")))?;
    let commit = result
        .metrics
        .get("commit")
        .and_then(Value::as_str)
        .unwrap_or("")
        .to_string();

    Ok(runtime::EvalResult {
        result_id: result.result_id.to_string(),
        run_id: result.run_id.to_string(),
        scenario_id: result.scenario_id.to_string(),
        outcome: result.outcome,
        commit,
        executed_at: Some(to_timestamp(result.executed_at)),
        metrics: Some(metrics_struct),
    })
}

fn build_eval_scenario_proto(
    scenario: &models::EvalScenario,
    spec: &EvalScenarioSpecData,
    history: Vec<runtime::EvalResult>,
) -> Result<runtime::EvalScenario, Status> {
    Ok(runtime::EvalScenario {
        scenario_id: scenario.scenario_id.to_string(),
        pack: spec.pack.clone(),
        slug: spec.slug.clone(),
        title: spec.title.clone().unwrap_or_else(|| spec.slug.clone()),
        description: spec.description.clone().unwrap_or_default(),
        created_at: Some(to_timestamp(scenario.created_at)),
        history,
    })
}

#[cfg(test)]
mod unit_tests {
    use super::*;
    use chrono::Utc;
    use fleetforge_common::prost_json::prost_value_to_json;
    use fleetforge_storage::models;
    use serde_json::{json, Map, Value};
    use std::collections::{HashMap, VecDeque};
    use std::time::Duration;
    use tonic::Code;
    use uuid::Uuid;

    #[test]
    fn parse_page_token_supports_defaults() {
        assert_eq!(parse_page_token("").expect("empty token"), 0);
        assert_eq!(parse_page_token("0").expect("zero token"), 0);
        assert!(parse_page_token("abc").is_err());
    }

    #[test]
    fn query_filters_normalizes_inputs() {
        let run_id = Uuid::new_v4();
        let mut request = QueryEventsRequest::default();
        request.run_ids.push(run_id.to_string());
        request.kinds.push("Step_Succeeded".to_string());
        request.workspace_ids.push("Workspace-1".to_string());
        request
            .policy_effects
            .push(PolicyDecisionEffect::PolicyDecisionEffectDeny as i32);
        request.severities.push(EventSeverity::Error as i32);

        let filters = QueryFiltersOwned::from_request(&request).expect("filters");
        assert_eq!(filters.run_ids, vec![run_id]);
        assert_eq!(filters.kinds, vec!["step_succeeded"]);
        assert_eq!(filters.workspace_ids, vec!["workspace-1"]);
        assert_eq!(
            filters.policy_effects,
            vec![PolicyDecisionEffect::PolicyDecisionEffectDeny]
        );
        assert_eq!(filters.severities, vec![EventSeverity::Error]);
    }

    #[test]
    fn matches_filters_respects_workspace_and_kind() {
        let run_id = Uuid::new_v4();
        let mut request = QueryEventsRequest::default();
        request.run_ids.push(run_id.to_string());
        request.workspace_ids.push("tenant-x".to_string());
        request.kinds.push("run_succeeded".to_string());
        let filters = QueryFiltersOwned::from_request(&request).expect("filters");

        let payload = json!({"trace": {"run": {"workspace_id": "Tenant-X"}}});
        let kind = "run_succeeded";
        let severity = EventSeverity::Info;
        assert!(
            matches_filters(&filters, &payload, kind, severity, None),
            "matching workspace should pass filter"
        );

        let other_payload = json!({"trace": {"run": {"workspace_id": "tenant-y"}}});
        assert!(
            !matches_filters(&filters, &other_payload, kind, severity, None),
            "non-matching workspace should fail filter"
        );
    }

    #[test]
    fn parse_dag_json_round_trip() {
        let dag = r#"{"steps": []}"#;
        let parsed = parse_dag_json(dag).expect("valid JSON");
        assert_eq!(parsed["steps"], Value::Array(vec![]));

        let err = parse_dag_json("not valid json").expect_err("should reject invalid JSON");
        assert_eq!(err.code(), Code::InvalidArgument);
    }

    #[test]
    fn extract_priority_clamps_to_i16() {
        let mut policy = Map::new();
        policy.insert("priority".to_string(), Value::from(40_000));
        let mut step = Map::new();
        step.insert("policy".to_string(), Value::Object(policy));
        let priority = extract_priority(&step);
        assert_eq!(priority, i16::MAX);
    }

    #[test]
    fn extract_priority_defaults_to_zero() {
        let step = Map::new();
        let priority = extract_priority(&step);
        assert_eq!(priority, 0);
    }

    #[test]
    fn extract_priority_handles_negative_overflow() {
        let mut policy = Map::new();
        policy.insert("priority".to_string(), Value::from(-50_000));
        let mut step = Map::new();
        step.insert("policy".to_string(), Value::Object(policy));
        let priority = extract_priority(&step);
        assert_eq!(priority, i16::MIN);
    }

    #[test]
    fn labels_map_round_trip() {
        let mut map = HashMap::new();
        map.insert("team".to_string(), "infra".to_string());
        map.insert("env".to_string(), "dev".to_string());
        let value = labels_map_to_value(&map);
        let round_trip = labels_value_to_map(&value);
        assert_eq!(map, round_trip);
    }

    #[test]
    fn build_run_detail_converts_payloads() {
        let run_id = Uuid::new_v4();
        let step_id = Uuid::new_v4();
        let now = Utc::now();

        let run = models::Run {
            run_id,
            created_at: now,
            status: models::RunStatus::Pending,
            dag_json: json!({"steps": []}),
            input_ctx: json!({
                "inputs": {"foo": "bar"},
                "labels": {"team": "infra"}
            }),
            seed: 7,
            idempotency_key: Some("abc".into()),
        };

        let step = models::Step {
            run_id,
            step_id,
            idx: 0,
            priority: 0,
            spec_json: json!({"id": step_id, "type": "tool", "inputs": {}, "policy": {}}),
            status: models::StepStatus::Succeeded,
            attempt: 1,
            created_at: now,
            pending_dependencies: 0,
            total_dependencies: 0,
            estimated_tokens: 0,
            estimated_cost: 0.0,
            actual_tokens: Some(10),
            actual_cost: Some(0.01),
            leased_by: None,
            lease_expires_at: None,
            output_json: Some(json!({"answer": 42})),
            error_json: Some(json!({"message": "none"})),
            input_snapshot: None,
            output_snapshot: None,
            provider: Some("test".into()),
            provider_version: Some("1.0".into()),
        };

        let detail = build_run_detail(run, vec![step]).expect("build run detail");
        let spec = detail.spec.expect("spec present");
        assert_eq!(spec.dag_json, r#"{"steps":[]}"#);
        assert_eq!(spec.seed, 7);
        assert_eq!(spec.labels.get("team"), Some(&"infra".to_string()));

        let step_summary = &detail.steps[0];
        assert_eq!(step_summary.step_id, step_id.to_string());
        assert_eq!(step_summary.status, "succeeded");
        let output_value = prost_value_to_json(step_summary.output.as_ref().unwrap());
        assert_eq!(output_value, json!({"answer": 42}));
    }

    #[test]
    fn subscription_config_parses_filters() {
        let run_id = Uuid::new_v4().to_string();
        let step_id = Uuid::new_v4().to_string();
        let open = SubscribeRunOpen {
            run_ids: vec![run_id.clone()],
            step_ids: vec![step_id.clone()],
            kinds: vec!["step_succeeded".to_string()],
            severities: vec![EventSeverity::Warn as i32],
            tags: vec!["team:core".to_string()],
            since_offset: 10,
            batch_size: 32,
            max_inflight: 20,
            heartbeat_interval_ms: 5_000,
        };

        let config = SubscriptionConfig::try_from_open(open).expect("valid config");
        assert_eq!(config.run_id.to_string(), run_id);
        assert!(config
            .step_ids
            .as_ref()
            .expect("step filter")
            .contains(&Uuid::parse_str(&step_id).unwrap()));
        assert!(config
            .kinds
            .as_ref()
            .expect("kind filter")
            .contains("step_succeeded"));
        assert!(config
            .severities
            .as_ref()
            .expect("severity filter")
            .contains(&EventSeverity::Warn));
        assert!(config
            .tags
            .as_ref()
            .expect("tag filter")
            .contains("team:core"));
        assert_eq!(config.since_offset, 10);
        assert_eq!(config.batch_size, 32);
        assert_eq!(config.max_inflight, 20);
        assert_eq!(config.heartbeat.as_millis(), 5_000);
    }

    #[test]
    fn handle_ack_evicts_inflight_offsets() {
        let mut inflight = VecDeque::from(vec![5, 9, 12]);
        let mut last_ack = 0;
        handle_ack(&mut inflight, &mut last_ack, 9);
        assert_eq!(last_ack, 9);
        assert_eq!(inflight, VecDeque::from(vec![12]));

        handle_ack(&mut inflight, &mut last_ack, 8);
        assert_eq!(last_ack, 9, "ack lower than last should be ignored");
        assert_eq!(inflight, VecDeque::from(vec![12]));

        handle_ack(&mut inflight, &mut last_ack, 15);
        assert_eq!(last_ack, 15);
        assert!(inflight.is_empty());
    }

    #[test]
    fn passes_filters_applies_tags_and_severities() {
        let run_id = Uuid::new_v4();
        let step_id = Uuid::new_v4();
        let base_config = SubscriptionConfig {
            run_id,
            step_ids: Some(vec![step_id].into_iter().collect()),
            kinds: Some(vec!["step_succeeded".to_string()].into_iter().collect()),
            severities: Some(vec![EventSeverity::Info].into_iter().collect()),
            tags: Some(vec!["severity:info".to_string()].into_iter().collect()),
            cursor: None,
            since_offset: 0,
            batch_size: 16,
            max_inflight: 8,
            heartbeat: Duration::from_millis(1_000),
        };

        let event = models::OutboxEvent {
            id: 1,
            run_id,
            step_id: Some(step_id),
            kind: "STEP_SUCCEEDED".to_string(),
            payload: json!({
                "status": "succeeded"
            }),
            created_at: Utc::now(),
            published_at: None,
        };

        let severity = derive_severity(&event.kind, &event.payload);
        assert_eq!(severity, EventSeverity::Info);
        let tags = derive_tags(&event.kind, event.step_id, &event.payload, severity);
        assert!(passes_filters(&base_config, &event, severity, &tags));

        let bad_config = SubscriptionConfig {
            tags: Some(vec!["severity:error".to_string()].into_iter().collect()),
            ..base_config
        };
        assert!(!passes_filters(&bad_config, &event, severity, &tags));
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::Utc;
    use fleetforge_storage::{ObjectStoreConfig, StorageConfig};
    use fleetforge_telemetry::audit::Audit;
    use serde_json::json;
    use std::collections::HashMap;
    use testcontainers::clients::Cli;
    use testcontainers::images::postgres::Postgres;
    use tokio::time::{sleep, Duration};
    use uuid::Uuid;

    fn build_request(dag: serde_json::Value) -> runtime::SubmitRunRequest {
        build_request_with_key(dag, "")
    }

    fn build_request_with_key(
        dag: serde_json::Value,
        idempotency_key: impl Into<String>,
    ) -> runtime::SubmitRunRequest {
        runtime::SubmitRunRequest {
            spec: Some(runtime::RunSpec {
                dag_json: serde_json::to_string(&dag).unwrap(),
                inputs: None,
                seed: 123,
                labels: HashMap::new(),
            }),
            idempotency_key: idempotency_key.into(),
        }
    }

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

        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 = Arc::new(
            Storage::connect(StorageConfig {
                database_url: db_url,
                max_connections: 2,
                connect_timeout: Duration::from_secs(10),
                object_store: ObjectStoreConfig::InMemory,
            })
            .await
            .expect("storage connect"),
        );
        let auth = Arc::new(AuthHandle::disabled());
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());
        let transparency = transparency::local_transparency_log(Arc::clone(&storage));
        let service = RuntimeApiService::new(
            storage,
            auth,
            audit,
            attestation_vault,
            transparency,
            false,
            TransparencyScope::Disabled,
        );

        let dag = serde_json::json!({
            "steps": [
                {
                    "type": "tool",
                    "inputs": {}
                }
            ]
        });

        let request = build_request(dag);
        let response = service.submit_run(Request::new(request)).await;
        assert!(response.is_err(), "expected schema validation failure");
        let status = response.err().unwrap();
        assert_eq!(status.code(), tonic::Code::InvalidArgument);
        assert!(status.message().contains("step schema validation failed"));
        drop(container);
    }

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

        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 = Arc::new(
            Storage::connect(StorageConfig {
                database_url: db_url,
                max_connections: 2,
                connect_timeout: Duration::from_secs(10),
                object_store: ObjectStoreConfig::InMemory,
            })
            .await
            .expect("storage connect"),
        );
        let auth = Arc::new(AuthHandle::disabled());
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());
        let transparency = transparency::local_transparency_log(Arc::clone(&storage));
        let service = RuntimeApiService::new(
            storage,
            auth,
            audit,
            attestation_vault,
            transparency,
            false,
            TransparencyScope::Disabled,
        );

        let step_id = Uuid::new_v4();
        let dag = json!({
            "steps": [
                {
                    "id": step_id,
                    "type": "tool",
                    "inputs": {
                        "command": ["echo", "ok"]
                    }
                }
            ]
        });

        let mut request = build_request(dag);
        if let Some(spec) = request.spec.as_mut() {
            spec.seed = 0;
        }

        let response = service.submit_run(Request::new(request)).await;
        assert!(response.is_err(), "expected seed validation failure");
        let status = response.err().unwrap();
        assert_eq!(status.code(), tonic::Code::InvalidArgument);
        assert!(status.message().contains("seed must be a positive integer"));

        drop(container);
    }

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

        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 = Arc::new(
            Storage::connect(StorageConfig {
                database_url: db_url,
                max_connections: 2,
                connect_timeout: Duration::from_secs(10),
                object_store: ObjectStoreConfig::InMemory,
            })
            .await
            .expect("storage connect"),
        );
        let auth = Arc::new(AuthHandle::disabled());
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());
        let transparency = transparency::local_transparency_log(Arc::clone(&storage));
        let service = RuntimeApiService::new(
            Arc::clone(&storage),
            Arc::clone(&auth),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            transparency,
            false,
            TransparencyScope::Disabled,
        );

        let step_a = Uuid::new_v4();
        let step_b = Uuid::new_v4();
        let valid_step = json!({
            "id": step_a,
            "type": "tool",
            "inputs": {}
        });

        let cases: Vec<(&str, serde_json::Value, &str)> = vec![
            (
                "missing_steps",
                json!({ "edges": [] }),
                "dag_json.steps must be an array",
            ),
            (
                "unknown_step_type",
                json!({ "steps": [ { "id": step_a, "type": "unknown", "inputs": {} } ] }),
                "step schema validation failed",
            ),
            (
                "duplicate_step_id",
                json!({ "steps": [valid_step.clone(), valid_step.clone()] }),
                "duplicate step.id",
            ),
            (
                "self_edge",
                json!({ "steps": [valid_step.clone()], "edges": [[step_a, step_a]] }),
                "edge cannot reference the same step twice",
            ),
            (
                "edge_unknown_target",
                json!({ "steps": [valid_step.clone()], "edges": [[step_a, step_b]] }),
                "edge references unknown step id",
            ),
            (
                "duplicate_edge",
                json!({ "steps": [valid_step.clone(), {"id": step_b, "type": "tool", "inputs": {}}], "edges": [[step_a, step_b], [step_a, step_b]] }),
                "duplicate edge detected",
            ),
        ];

        for (label, dag, expected) in cases {
            let request = build_request(dag);
            let response = service.submit_run(Request::new(request)).await;
            assert!(response.is_err(), "case {label} unexpectedly succeeded");
            let status = response.err().unwrap();
            assert_eq!(status.code(), tonic::Code::InvalidArgument, "case {label}");
            assert!(
                status.message().contains(expected),
                "case {label} expected '{expected}' but got '{}'",
                status.message()
            );
        }

        drop(container);
    }

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

        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 = Arc::new(
            Storage::connect(StorageConfig {
                database_url: db_url,
                max_connections: 2,
                connect_timeout: Duration::from_secs(10),
                object_store: ObjectStoreConfig::InMemory,
            })
            .await
            .expect("storage connect"),
        );
        let auth = Arc::new(AuthHandle::disabled());
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());
        let transparency = transparency::local_transparency_log(Arc::clone(&storage));
        let service = RuntimeApiService::new(
            Arc::clone(&storage),
            Arc::clone(&auth),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            transparency,
            false,
            TransparencyScope::Disabled,
        );

        let step_a = Uuid::new_v4();
        let step_b = Uuid::new_v4();
        let step_c = Uuid::new_v4();

        let dag = serde_json::json!({
            "steps": [
                {
                    "id": step_a,
                    "type": "tool",
                    "inputs": {},
                    "policy": {"priority": 5}
                },
                {
                    "id": step_b,
                    "type": "llm",
                    "inputs": {},
                    "policy": {"priority": 3}
                },
                {
                    "id": step_c,
                    "type": "reduce",
                    "inputs": {},
                    "policy": {"priority": 1}
                }
            ],
            "edges": [
                [step_a, step_b],
                [step_b, step_c]
            ]
        });

        let request = build_request(dag);
        let response = service
            .submit_run(Request::new(request))
            .await
            .expect("valid run should succeed")
            .into_inner();

        let run_uuid = Uuid::parse_str(&response.run_id).expect("run_id must be UUID");
        let persisted_run = storage
            .fetch_run(run_uuid)
            .await
            .expect("fetch run")
            .expect("run should exist");
        assert_eq!(persisted_run.status, models::RunStatus::Pending);

        let steps = storage
            .fetch_steps_for_run(run_uuid)
            .await
            .expect("fetch steps");
        assert_eq!(steps.len(), 3);

        let mut by_idx = steps;
        by_idx.sort_by_key(|s| s.idx);

        assert_eq!(by_idx[0].step_id, step_a);
        assert_eq!(by_idx[0].status, models::StepStatus::Queued);
        assert_eq!(by_idx[0].priority, 5);
        assert_eq!(by_idx[0].pending_dependencies, 0);

        assert_eq!(by_idx[1].step_id, step_b);
        assert_eq!(by_idx[1].status, models::StepStatus::Blocked);
        assert_eq!(by_idx[1].priority, 3);
        assert_eq!(by_idx[1].pending_dependencies, 1);

        assert_eq!(by_idx[2].step_id, step_c);
        assert_eq!(by_idx[2].status, models::StepStatus::Blocked);
        assert_eq!(by_idx[2].priority, 1);
        assert_eq!(by_idx[2].pending_dependencies, 1);
        assert!(
            by_idx
                .iter()
                .all(|s| s.total_dependencies >= s.pending_dependencies),
            "dependency counters should be consistent"
        );
        drop(container);
    }

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

        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 = Arc::new(
            Storage::connect(StorageConfig {
                database_url: db_url,
                max_connections: 2,
                connect_timeout: Duration::from_secs(10),
                object_store: ObjectStoreConfig::InMemory,
            })
            .await
            .expect("storage connect"),
        );
        let auth = Arc::new(AuthHandle::disabled());
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());
        let transparency = transparency::local_transparency_log(Arc::clone(&storage));
        let service = RuntimeApiService::new(
            Arc::clone(&storage),
            Arc::clone(&auth),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            transparency,
            false,
            TransparencyScope::Disabled,
        );

        let step_id = Uuid::new_v4();
        let dag = serde_json::json!({
            "steps": [{
                "id": step_id,
                "type": "tool",
                "inputs": {}
            }]
        });
        let raw_key = "  idempotent-key  ";
        let normalized_key = "idempotent-key";

        let first = service
            .submit_run(Request::new(build_request_with_key(dag.clone(), raw_key)))
            .await
            .expect("first submission")
            .into_inner();

        let second = service
            .submit_run(Request::new(build_request_with_key(dag, raw_key)))
            .await
            .expect("second submission")
            .into_inner();

        assert_eq!(
            first.run_id, second.run_id,
            "idempotent submissions should reuse run_id"
        );

        let stored = storage
            .fetch_run_by_idempotency_key(normalized_key)
            .await
            .expect("lookup idempotent run")
            .expect("run should exist for idempotency key");
        assert_eq!(stored.run_id.to_string(), first.run_id);

        drop(container);
    }

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

        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 = Arc::new(
            Storage::connect(StorageConfig {
                database_url: db_url,
                max_connections: 2,
                connect_timeout: Duration::from_secs(10),
                object_store: ObjectStoreConfig::InMemory,
            })
            .await
            .expect("storage connect"),
        );
        let auth = Arc::new(AuthHandle::disabled());
        let audit = Arc::new(Audit::new(storage.pool().clone()));
        let attestation_vault: Arc<dyn AttestationVault> =
            Arc::new(InMemoryAttestationVault::new());
        let transparency = transparency::local_transparency_log(Arc::clone(&storage));
        let service = RuntimeApiService::new(
            Arc::clone(&storage),
            Arc::clone(&auth),
            Arc::clone(&audit),
            Arc::clone(&attestation_vault),
            transparency,
            false,
            TransparencyScope::Disabled,
        );

        let run_id = Uuid::new_v4();
        let attestation =
            Attestation::new(Uuid::new_v4(), Utc::now()).with_subject(TrustSubject::Run { run_id });
        attestation_vault
            .record(attestation.clone())
            .await
            .expect("record attestation");

        let response = service
            .get_attestations(Request::new(runtime::GetAttestationsRequest {
                attestation_ids: vec![
                    attestation.id.to_string(),
                    attestation.id.to_string(),
                    Uuid::new_v4().to_string(),
                ],
            }))
            .await
            .expect("get_attestations RPC")
            .into_inner();

        assert_eq!(response.attestations.len(), 1);
        let returned = &response.attestations[0];
        assert_eq!(returned.attestation_id, attestation.id.to_string());
        match returned
            .subject
            .as_ref()
            .and_then(|subject| subject.kind.as_ref())
        {
            Some(runtime::attestation_subject::Kind::Run(run_subject)) => {
                assert_eq!(run_subject.run_id, run_id.to_string());
            }
            other => panic!("unexpected subject {other:?}"),
        }

        drop(container);
    }

    #[test]
    fn compute_drift_reports_within_tolerance() {
        let recorded = json!({
            "tokens": 10,
            "nested": { "value": "ok" }
        });
        let current = recorded.clone();

        let result = compute_drift(
            &recorded,
            &current,
            DEFAULT_TOKEN_TOLERANCE,
            DEFAULT_COST_TOLERANCE,
        );
        assert!(result.within_tolerance);
        assert_eq!(result.token_delta, Some(0.0));
        assert_eq!(result.cost_delta, None);
    }

    #[test]
    fn compute_drift_flags_token_delta() {
        let recorded = json!({
            "tokens": 10,
            "cost": 1.0,
            "payload": { "message": "hello" }
        });
        let current = json!({
            "tokens": 15,
            "cost": 1.0,
            "payload": { "message": "hello" }
        });

        let result = compute_drift(
            &recorded,
            &current,
            DEFAULT_TOKEN_TOLERANCE,
            DEFAULT_COST_TOLERANCE,
        );
        assert!(!result.within_tolerance);
        assert_eq!(result.token_delta, Some(5.0));
        assert_eq!(result.cost_delta, Some(0.0));
    }
}