use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use std::process::Stdio;
use std::str::FromStr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, LazyLock, Mutex, RwLock};
use std::time::{Duration, Instant};

use anyhow::Context as _;
use axum::body::Body;
#[cfg(test)]
use axum::body::to_bytes;
use axum::extract::{self as axum_extract, DefaultBodyLimit, Path, Query, State};
use axum::http::{HeaderMap, Method, StatusCode, header};
use axum::middleware::{self, Next};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use axum_extra::extract::cookie::Key;
use base64::Engine as _;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use bytes::Bytes;
use chrono::{DateTime, Utc};
pub use fabro_api::types::{
    AggregateBilling, AggregateBillingTotals, ApiQuestion, AppendEventResponse, ArtifactEntry,
    ArtifactListResponse, BatchDeleteRunsRequest, BatchDeleteRunsResponse, BatchDeleteRunsResult,
    BatchDeleteRunsResultOutcome, BatchDeleteRunsSummary, BatchRunLifecycleRequest,
    BatchRunLifecycleResponse, BatchRunLifecycleResult, BatchRunLifecycleResultOutcome,
    BatchRunLifecycleSummary, BillingByModel, BillingStageRef, CloseRunPullRequestResponse,
    CompletionResponse, CompletionUsage, CreateCompletionRequest, CreateRunPullRequestRequest,
    CreateSecretRequest, CreateVariableRequest, DeleteRunResponse, DeleteRunSandbox,
    DeleteSecretRequest, DenyRunRequest, DiskUsageResponse, DiskUsageRunRow, DiskUsageSummaryRow,
    ErrorResponseEntry, ForkRequest, ForkResponse, IntegrationConnectionKind,
    IntegrationConnectionState, IntegrationConnectionStatus, IntegrationProvider,
    IntegrationStatus, LinkRunPullRequestRequest, MergeRunPullRequestRequest,
    MergeRunPullRequestResponse, ModelReference, PaginatedEventList, PaginatedRunList,
    PaginationMeta, PreflightResponse, PreviewUrlRequest, PreviewUrlResponse, Provider,
    ProviderCredentialTestRequest, ProviderCredentialTestResponse, ProviderList, PruneRunEntry,
    PruneRunsRequest, PruneRunsResponse, RenderWorkflowGraphDirection, RenderWorkflowGraphRequest,
    RewindRequest, RewindResponse, Run, RunArtifactEntry, RunArtifactListResponse, RunBilling,
    RunBillingStage, RunBillingTotals, RunError, RunManifest, RunStage, SandboxDetails,
    SandboxFileEntry, SandboxFileListResponse, SandboxService, SandboxServiceListResponse,
    SshAccessRequest, SshAccessResponse, StageHandler, StageState, StartRunRequest,
    SubmitAnswerRequest, SystemCpuResourceScope, SystemCpuResources, SystemDiskResourceScope,
    SystemDiskResources, SystemInfoResponse, SystemIntegrationStatus, SystemIntegrationsResponse,
    SystemMemoryResourceScope, SystemMemoryResources, SystemRepairRunIssue,
    SystemRepairRunsResponse, SystemResourcesResponse, SystemRunCounts, TimelineEntryResponse,
    UpdateVariableRequest, VariableListResponse, VncPreviewResponse, WriteBlobResponse,
};
use fabro_auth::SqlVaultCredentialSource;
use fabro_automation::{self, AutomationStore};
use fabro_config::daemon::ServerDaemon;
use fabro_config::{LlmLayer, RunLayer, Storage, WorkflowSettingsBuilder};
use fabro_db::DbPool;
use fabro_environment::EnvironmentStore;
use fabro_interview::{
    Answer, AnswerSubmission, ControlInterviewer, Interviewer, Question, WorkerControlEnvelope,
};
use fabro_llm::credentials::CredentialProvider;
use fabro_llm::lithos_catalog::Catalog;
use fabro_llm::{ClientOptions, FabroClient};
use fabro_mcp_store::McpServerStore;
use fabro_redact::redact_jsonl_line;
use fabro_sandbox::details::sandbox_details;
use fabro_sandbox::driver::{DaytonaCredentials, ProviderAccess, ProviderConnectOptions};
use fabro_sandbox::reconnect::reconnect_for_run;
use fabro_sandbox::{SandboxInventory, daytona};
use fabro_slack::client::{PostedMessage as SlackPostedMessage, SlackClient};
use fabro_slack::config::{
    SlackCredentialResolution,
    resolve_credentials_status_with_lookup as resolve_slack_credentials_status_with_lookup,
};
use fabro_slack::payload::SlackAnswerSubmission;
use fabro_slack::threads::ThreadRegistry;
use fabro_slack::{blocks as slack_blocks, connection as slack_connection};
use fabro_static::EnvVars;
use fabro_store::{
    ArtifactKey, ArtifactStore, AuthCodeStore, AuthSessionStore, Database, EventEnvelope,
    EventPayload, KeyedMutex, NodeArtifact, PendingInterviewRecord, RunSessionRecordStore,
    RunSummaryStore, StageArtifactEntry, StageId,
};
#[cfg(test)]
use fabro_types::BlockedReason;
use fabro_types::settings::RunNamespace;
use fabro_types::settings::run::{NotificationRouteSettings, RunMode};
use fabro_types::settings::server::{
    GithubIntegrationSettings, GithubIntegrationStrategy, LogDestination,
};
use fabro_types::{
    AgentBackend, AskFabro, AskFabroUnavailableReason, BilledTokenCounts, BlobHash, EventBody,
    InterviewQuestionRecord, ModelRef, ModelTestMode, PairId, PairMessageId, PairTarget,
    PendingReason, Principal, PullRequestLink, QuestionType, RunControlAction, RunEvent, RunId,
    RunRunnableSource, RunStatusKind, SandboxProviderKind, ServerSettings, SessionCapability,
};
use fabro_util::error::{
    SharedError, collect_causes, render_compact_with_causes, render_with_causes,
};
use fabro_util::version::FABRO_VERSION;
use fabro_variable::{Error as VariableError, VariableStore};
use fabro_vault::{SecretStore, SecretStoreError, SecretType, Vault};
use fabro_workflow::artifact_upload::ArtifactSink;
#[cfg(test)]
use fabro_workflow::command_log::command_log_path;
use fabro_workflow::event::{self as workflow_event, Emitter};
use fabro_workflow::handler::HandlerRegistry;
use fabro_workflow::pipeline::Persisted;
use fabro_workflow::records::Checkpoint;
use fabro_workflow::run_lookup::{
    RunInfo, StatusFilter, filter_runs, scan_runs_with_summaries, scratch_base,
};
use fabro_workflow::run_status::{FailureReason, RunStatus, SuccessReason};
use fabro_workflow::{Error as WorkflowError, operations, pull_request};
use futures_util::future::join_all;
use lithos_llm::catalog::ProviderId;
use sha2::{Digest, Sha256};
use tempfile::NamedTempFile;
use tokio::fs;
use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWriteExt, BufReader};
use tokio::process::Command;
use tokio::runtime::Builder as TokioRuntimeBuilder;
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::{
    Mutex as AsyncMutex, Notify, RwLock as AsyncRwLock, Semaphore, broadcast, mpsc, oneshot,
};
use tokio::task::spawn_blocking;
use tokio::time::{sleep, timeout};
use tokio_stream::StreamExt;
use tokio_stream::wrappers::{BroadcastStream, UnboundedReceiverStream};
use tokio_util::sync::CancellationToken;
use tower::{ServiceExt, service_fn};
use tower_http::compression::predicate::{DefaultPredicate, NotForContentType, Predicate};
use tower_http::compression::{CompressionLayer, CompressionLevel};
use tracing::{Instrument, debug, error, info, warn};

use crate::auth::{self, GithubEndpoints, auth_translation_middleware, demo_routing_middleware};
use crate::automation_materializer::{
    AutomationRunMaterializeInput, AutomationRunMaterialized, AutomationRunMaterializer,
    ProductionAutomationRunMaterializer, RunMaterializeError,
};
use crate::canonical_origin::{canonical_origin_from_effective_web_url, effective_web_url};
use crate::error::ApiError;
use crate::git_checkout::GitRepoCache;
use crate::github_webhooks::{
    WEBHOOK_ROUTE, WEBHOOK_SECRET_ENV, parse_event_metadata, verify_signature,
};
use crate::jwt_auth::{self, AuthMode};
use crate::principal_middleware::{
    AuthContextSlot, RequestAuth, RequestAuthContext, RequireRunBlob, RequireRunManagementTarget,
    RequireRunScoped, RequireRunStageScoped, RequireStageArtifact, RequireWorkerRunScoped,
    RequiredUser, principal_middleware,
};
use crate::request_id::{self, RequestId};
use crate::run_files::{FilesInFlight, new_files_in_flight};
use crate::server_secrets::ServerSecrets;
use crate::spawn_env::apply_render_graph_env;
use crate::worker_control::{LocalWorkerControlBus, WorkerControlBus, WorkerControlBusError};
use crate::worker_runtime::{
    LocalWorkerRuntime, WorkerExit, WorkerLaunchSpec, WorkerRef, WorkerRuntime,
};
use crate::worker_token::{WorkerScopeSet, WorkerTokenKeys, issue_worker_token_with_scopes};
use crate::{
    canonical_host, demo, diagnostics, run_manifest, security_headers, static_files, web_auth,
};

mod automation_scheduler;
mod handler;
mod pull_request_supervisor;
pub(crate) mod resource_sampler;
mod session_runtime;

pub(crate) use automation_scheduler::spawn_automation_scheduler;
pub(crate) use handler::events::EventListParams;
#[cfg(test)]
pub(in crate::server) use handler::events::filtered_global_events;
pub(crate) use handler::graph::render_graph_bytes;
#[cfg(test)]
pub(in crate::server) use handler::graph::{
    RenderSubprocessError, render_dot_subprocess, render_graph_bytes_with_exe_override,
};
#[cfg(test)]
pub(in crate::server) use handler::system::validate_github_slug;
pub(crate) use pull_request_supervisor::spawn_pull_request_creation_supervisor;
use session_runtime::SessionRuntimeManager;

pub(crate) type EnvLookup = Arc<dyn Fn(&str) -> Option<String> + Send + Sync>;

pub fn default_page_limit() -> u32 {
    20
}

#[derive(serde::Deserialize)]
pub struct PaginationParams {
    #[serde(rename = "page[limit]", default = "default_page_limit")]
    pub limit:  u32,
    #[serde(rename = "page[offset]", default)]
    pub offset: u32,
}

pub(crate) fn clamp_page_limit(limit: u32) -> u32 {
    limit.clamp(1, 100)
}

pub(crate) fn clamp_page_offset(offset: u32) -> u32 {
    offset.min(MAX_PAGE_OFFSET)
}

pub(crate) fn paginate_items<T>(items: Vec<T>, pagination: &PaginationParams) -> (Vec<T>, bool) {
    let limit = clamp_page_limit(pagination.limit) as usize;
    let offset = clamp_page_offset(pagination.offset) as usize;
    let mut data: Vec<_> = items.into_iter().skip(offset).take(limit + 1).collect();
    let has_more = data.len() > limit;
    data.truncate(limit);
    (data, has_more)
}

#[derive(serde::Deserialize)]
pub(crate) struct DfParams {
    #[serde(default)]
    pub(crate) verbose: bool,
}

/// List response envelope with pagination metadata.
#[derive(serde::Serialize)]
pub struct ListResponse<T: serde::Serialize> {
    data: T,
    meta: PaginationMeta,
}

impl<T: serde::Serialize> ListResponse<T> {
    /// Non-paginated response with `has_more: false`.
    pub fn new(data: T) -> Self {
        Self {
            data,
            meta: PaginationMeta {
                has_more: false,
                total:    None,
            },
        }
    }

    pub fn paginated(data: T, has_more: bool, total: u64) -> Self {
        Self {
            data,
            meta: PaginationMeta {
                has_more,
                total: i64::try_from(total).ok(),
            },
        }
    }
}

/// Snapshot of a managed run.
struct ManagedRun {
    dot_source: String,
    status: RunStatus,
    error: Option<String>,
    created_at: chrono::DateTime<chrono::Utc>,
    // Populated when running:
    answer_transport: Option<RunAnswerTransport>,
    accepted_questions: HashSet<String>,
    /// Stage IDs of currently steerable live agent sessions, keyed to the
    /// session id that owns the active lease. Used by the steerability
    /// predicate for steer/interrupt controls.
    active_steerable_stages: HashMap<StageId, String>,
    /// API-mode session targets eligible for live pair control. ACP sessions
    /// can be steerable but are intentionally excluded from pairing.
    active_api_targets: HashMap<StageId, PairTarget>,
    /// Stage IDs of currently running agent sessions that have no live
    /// steering capability, keyed to the session id that owns the marker.
    active_non_steerable_stages: HashMap<StageId, String>,
    event_tx: Option<broadcast::Sender<RunEvent>>,
    checkpoint: Option<Checkpoint>,
    cancel_tx: Option<oneshot::Sender<()>>,
    cancel_token: Option<CancellationToken>,
    worker_ref: Option<WorkerRef>,
    /// Exact worker currently covered by a cancellation escalation task.
    /// Prevents repeated cancel requests from arming duplicate watchdogs.
    cancel_escalation_worker: Option<WorkerRef>,
    run_dir: Option<std::path::PathBuf>,
    execution_mode: RunExecutionMode,
}

impl ManagedRun {
    /// True if cancellation should still escalate to `worker_ref`; clears a
    /// stale escalation marker as a side effect.
    fn escalation_still_current(&mut self, worker_ref: &WorkerRef) -> bool {
        let matches_watchdog = self.cancel_escalation_worker.as_ref() == Some(worker_ref);
        let still_current = matches_watchdog
            && !self.status.is_terminal()
            && self.worker_ref.as_ref() == Some(worker_ref);
        if matches_watchdog && !still_current {
            self.cancel_escalation_worker = None;
        }
        still_current
    }

    /// Clears the escalation marker if it is still owned by `worker_ref`.
    fn clear_escalation_for(&mut self, worker_ref: &WorkerRef) {
        if self.cancel_escalation_worker.as_ref() == Some(worker_ref) {
            self.cancel_escalation_worker = None;
        }
    }
}

#[derive(Clone, Copy)]
enum RunExecutionMode {
    Start,
    Resume,
}

enum ExecutionResult {
    Completed(Box<Result<operations::Started, WorkflowError>>),
    CancelledBySignal,
}

const WORKER_CANCEL_GRACE: Duration = Duration::from_secs(5);
const TERMINAL_DELETE_WORKER_GRACE: Duration = Duration::from_millis(50);
const WORKER_CONTROL_ENQUEUE_TIMEOUT: Duration = Duration::from_secs(1);
/// Per-model billing totals.
#[derive(Default)]
struct ModelBillingTotals {
    stages:  i64,
    billing: BilledTokenCounts,
}

/// In-memory aggregate billing counters, reset on server restart.
#[derive(Default)]
struct BillingAccumulator {
    total_runs:   i64,
    total_timing: fabro_types::RunTiming,
    by_model:     HashMap<ModelRef, ModelBillingTotals>,
}

pub(crate) type RegistryFactoryOverride =
    dyn Fn(Arc<dyn Interviewer>) -> HandlerRegistry + Send + Sync;

#[derive(Clone)]
enum RunAnswerTransport {
    Worker {
        run_id: RunId,
        bus:    Arc<dyn WorkerControlBus>,
    },
    InProcess {
        interviewer:  Arc<ControlInterviewer>,
        steering_hub: Arc<fabro_workflow::SteeringHub>,
    },
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AnswerTransportError {
    Closed,
    Timeout,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PairTransportError {
    Closed,
    Timeout,
    Control(fabro_workflow::PairControlError),
}

impl RunAnswerTransport {
    async fn publish_worker_control(
        run_id: RunId,
        bus: &Arc<dyn WorkerControlBus>,
        message: WorkerControlEnvelope,
    ) -> Result<(), WorkerControlBusError> {
        timeout(WORKER_CONTROL_ENQUEUE_TIMEOUT, bus.publish(run_id, message))
            .await
            .map_err(|_| WorkerControlBusError::PublishTimeout)?
            .map(|_| ())
    }

    fn answer_error_from_bus(error: &WorkerControlBusError) -> AnswerTransportError {
        match error {
            WorkerControlBusError::PublishTimeout => AnswerTransportError::Timeout,
            WorkerControlBusError::Closed
            | WorkerControlBusError::Unavailable
            | WorkerControlBusError::InvalidCursor { .. } => AnswerTransportError::Closed,
        }
    }

    fn pair_error_from_bus(error: &WorkerControlBusError) -> PairTransportError {
        match error {
            WorkerControlBusError::PublishTimeout => PairTransportError::Timeout,
            WorkerControlBusError::Closed
            | WorkerControlBusError::Unavailable
            | WorkerControlBusError::InvalidCursor { .. } => PairTransportError::Closed,
        }
    }

    async fn submit(
        &self,
        qid: &str,
        submission: AnswerSubmission,
    ) -> Result<(), AnswerTransportError> {
        match self {
            Self::Worker { run_id, bus } => {
                let message = WorkerControlEnvelope::interview_answer(qid.to_string(), submission);
                Self::publish_worker_control(*run_id, bus, message)
                    .await
                    .map_err(|err| Self::answer_error_from_bus(&err))
            }
            Self::InProcess { interviewer, .. } => interviewer
                .submit(qid, submission)
                .await
                .map_err(|_| AnswerTransportError::Closed),
        }
    }

    async fn cancel_run(&self) -> Result<(), AnswerTransportError> {
        match self {
            Self::Worker { run_id, bus } => {
                let message = WorkerControlEnvelope::cancel_run();
                Self::publish_worker_control(*run_id, bus, message)
                    .await
                    .map_err(|err| Self::answer_error_from_bus(&err))
            }
            Self::InProcess { interviewer, .. } => {
                interviewer.cancel_all().await;
                Ok(())
            }
        }
    }

    /// Forward a steer to the worker (subprocess) or directly into the
    /// in-process steering hub.
    async fn steer(&self, text: String, actor: Principal) -> Result<(), AnswerTransportError> {
        match self {
            Self::Worker { run_id, bus } => {
                let message = WorkerControlEnvelope::steer(text, actor);
                Self::publish_worker_control(*run_id, bus, message)
                    .await
                    .map_err(|err| Self::answer_error_from_bus(&err))
            }
            Self::InProcess { steering_hub, .. } => {
                steering_hub.deliver_steer(text, Some(actor));
                Ok(())
            }
        }
    }

    async fn interrupt(&self, actor: Principal) -> Result<(), AnswerTransportError> {
        match self {
            Self::Worker { run_id, bus } => {
                let message = WorkerControlEnvelope::interrupt(actor);
                Self::publish_worker_control(*run_id, bus, message)
                    .await
                    .map_err(|err| Self::answer_error_from_bus(&err))
            }
            Self::InProcess { steering_hub, .. } => {
                steering_hub.interrupt(Some(&actor));
                Ok(())
            }
        }
    }

    async fn interrupt_then_steer(
        &self,
        text: String,
        actor: Principal,
    ) -> Result<(), AnswerTransportError> {
        match self {
            Self::Worker { run_id, bus } => {
                let message = WorkerControlEnvelope::interrupt_then_steer(text, actor);
                Self::publish_worker_control(*run_id, bus, message)
                    .await
                    .map_err(|err| Self::answer_error_from_bus(&err))
            }
            Self::InProcess { steering_hub, .. } => {
                steering_hub.interrupt_then_steer(&text, Some(&actor));
                Ok(())
            }
        }
    }

    async fn start_pair(
        &self,
        run_id: RunId,
        pair_id: PairId,
        target: PairTarget,
        actor: Principal,
    ) -> Result<(), PairTransportError> {
        match self {
            Self::Worker {
                run_id: worker_run_id,
                bus,
            } => {
                let message = WorkerControlEnvelope::start_pair(run_id, pair_id, target, actor);
                Self::publish_worker_control(*worker_run_id, bus, message)
                    .await
                    .map_err(|err| Self::pair_error_from_bus(&err))
            }
            Self::InProcess { steering_hub, .. } => steering_hub
                .start_pair(run_id, pair_id, target, Some(actor))
                .map(|_| ())
                .map_err(PairTransportError::Control),
        }
    }

    async fn send_pair_message(
        &self,
        pair_id: PairId,
        message_id: PairMessageId,
        text: String,
        client_message_id: Option<String>,
        actor: Principal,
    ) -> Result<(), PairTransportError> {
        match self {
            Self::Worker { run_id, bus } => {
                let message = WorkerControlEnvelope::pair_message(
                    pair_id,
                    message_id,
                    text.clone(),
                    client_message_id.clone(),
                    actor,
                );
                Self::publish_worker_control(*run_id, bus, message)
                    .await
                    .map_err(|err| Self::pair_error_from_bus(&err))
            }
            Self::InProcess { steering_hub, .. } => steering_hub
                .send_pair_message(pair_id, message_id, text, client_message_id, Some(actor))
                .map(|_| ())
                .map_err(PairTransportError::Control),
        }
    }

    async fn end_pair(&self, pair_id: PairId, actor: Principal) -> Result<(), PairTransportError> {
        match self {
            Self::Worker { run_id, bus } => {
                let message = WorkerControlEnvelope::end_pair(pair_id, actor);
                Self::publish_worker_control(*run_id, bus, message)
                    .await
                    .map_err(|err| Self::pair_error_from_bus(&err))
            }
            Self::InProcess { steering_hub, .. } => steering_hub
                .end_pair(pair_id, Some(actor))
                .map(|_| ())
                .map_err(PairTransportError::Control),
        }
    }

    async fn pause_run(&self) -> Result<(), AnswerTransportError> {
        match self {
            Self::Worker { run_id, bus } => {
                let message = WorkerControlEnvelope::pause_run();
                Self::publish_worker_control(*run_id, bus, message)
                    .await
                    .map_err(|err| Self::answer_error_from_bus(&err))
            }
            Self::InProcess { .. } => Err(AnswerTransportError::Closed),
        }
    }

    async fn unpause_run(&self) -> Result<(), AnswerTransportError> {
        match self {
            Self::Worker { run_id, bus } => {
                let message = WorkerControlEnvelope::unpause_run();
                Self::publish_worker_control(*run_id, bus, message)
                    .await
                    .map_err(|err| Self::answer_error_from_bus(&err))
            }
            Self::InProcess { .. } => Err(AnswerTransportError::Closed),
        }
    }
}

#[derive(Debug, Clone)]
struct LoadedPendingInterview {
    run_id:   RunId,
    qid:      String,
    question: InterviewQuestionRecord,
}

#[derive(Debug, Clone)]
struct SlackLifecycleDetails {
    kind:               slack_blocks::RunLifecycleKind,
    started_event_name: Option<String>,
    result:             Option<String>,
    duration_ms:        Option<u64>,
}

#[derive(Debug, Clone, Default)]
struct PriorSlackLifecycleEventDetails {
    started_event_name: Option<String>,
    pull_request:       Option<SlackLifecyclePullRequest>,
}

#[derive(Debug, Clone)]
struct SlackLifecyclePullRequest {
    number: u64,
    title:  Option<String>,
    url:    Option<String>,
}

#[derive(Debug, Clone)]
struct SlackConnectionRuntimeState {
    status:            IntegrationConnectionState,
    last_connected_at: Option<DateTime<Utc>>,
    last_error:        Option<String>,
}

impl Default for SlackConnectionRuntimeState {
    fn default() -> Self {
        Self {
            status:            IntegrationConnectionState::Connecting,
            last_connected_at: None,
            last_error:        None,
        }
    }
}

fn sanitize_integration_error(error: &str) -> String {
    const MAX_ERROR_CHARS: usize = 240;
    let sanitized = error.replace(['\r', '\n'], " ");
    sanitized.chars().take(MAX_ERROR_CHARS).collect()
}

#[derive(Clone)]
struct SlackService {
    client:          SlackClient,
    app_token:       String,
    default_channel: Option<String>,
    posted_messages: Arc<Mutex<HashMap<(RunId, String), SlackPostedMessage>>>,
    thread_registry: Arc<ThreadRegistry>,
    connection:      Arc<Mutex<SlackConnectionRuntimeState>>,
}

impl SlackService {
    fn new(bot_token: String, app_token: String, default_channel: Option<String>) -> Self {
        Self {
            client: SlackClient::new(bot_token),
            app_token,
            default_channel,
            posted_messages: Arc::new(Mutex::new(HashMap::new())),
            thread_registry: Arc::new(ThreadRegistry::new()),
            connection: Arc::new(Mutex::new(SlackConnectionRuntimeState::default())),
        }
    }

    fn connection_status(&self) -> IntegrationConnectionStatus {
        let state = self
            .connection
            .lock()
            .expect("slack connection state lock poisoned")
            .clone();
        IntegrationConnectionStatus {
            kind:              IntegrationConnectionKind::SocketMode,
            status:            state.status,
            last_connected_at: state.last_connected_at,
            last_error:        state.last_error,
        }
    }

    fn status_sink(&self) -> slack_connection::ConnectionStatusSink {
        let connection = Arc::clone(&self.connection);
        Arc::new(move |update| {
            let mut state = connection
                .lock()
                .expect("slack connection state lock poisoned");
            match update {
                slack_connection::ConnectionStatusUpdate::Connecting => {
                    state.status = IntegrationConnectionState::Connecting;
                    state.last_error = None;
                }
                slack_connection::ConnectionStatusUpdate::Connected => {
                    state.status = IntegrationConnectionState::Connected;
                    state.last_connected_at = Some(Utc::now());
                    state.last_error = None;
                }
                slack_connection::ConnectionStatusUpdate::Error(error) => {
                    state.status = IntegrationConnectionState::Error;
                    state.last_error = Some(sanitize_integration_error(&error));
                }
            }
        })
    }

    async fn handle_event(
        &self,
        state: &AppState,
        envelope: &EventEnvelope,
        run_web_url: Option<&str>,
    ) {
        let event = &envelope.event;
        match &event.body {
            EventBody::InterviewStarted(props) => {
                if props.question_id.is_empty() {
                    return;
                }
                let Some(default_channel) = self.default_channel.as_deref() else {
                    return;
                };
                let key = (event.run_id, props.question_id.clone());
                if self
                    .posted_messages
                    .lock()
                    .expect("slack posted messages lock poisoned")
                    .contains_key(&key)
                {
                    return;
                }

                let question = runtime_question_from_interview_record(&InterviewQuestionRecord {
                    id:              props.question_id.clone(),
                    text:            props.question.clone(),
                    stage:           props.stage.clone(),
                    question_type:   props.question_type.parse().unwrap_or_default(),
                    options:         props.options.clone(),
                    allow_freeform:  props.allow_freeform,
                    timeout_seconds: props.timeout_seconds,
                    context_display: props.context_display.clone(),
                    review_target:   props.review_target.clone(),
                });
                let blocks = slack_blocks::question_to_blocks(
                    &event.run_id.to_string(),
                    &props.question_id,
                    &question,
                    run_web_url,
                );

                if let Ok(posted) = self
                    .client
                    .post_message(default_channel, &blocks, None)
                    .await
                {
                    if question.allow_freeform || question.question_type == QuestionType::Freeform {
                        self.thread_registry.register(
                            &posted.ts,
                            &event.run_id.to_string(),
                            &props.question_id,
                        );
                    }
                    self.posted_messages
                        .lock()
                        .expect("slack posted messages lock poisoned")
                        .insert(key, posted);
                }
            }
            EventBody::InterviewCompleted(props) => {
                self.finish_interview(
                    event.run_id,
                    &props.question_id,
                    &props.question,
                    &props.answer,
                )
                .await;
            }
            EventBody::InterviewTimeout(props) => {
                self.finish_interview(
                    event.run_id,
                    &props.question_id,
                    &props.question,
                    "Timed out",
                )
                .await;
            }
            EventBody::InterviewInterrupted(props) => {
                self.finish_interview(
                    event.run_id,
                    &props.question_id,
                    &props.question,
                    "Interrupted",
                )
                .await;
            }
            EventBody::RunStarted(_) | EventBody::RunCompleted(_) | EventBody::RunFailed(_) => {
                self.handle_lifecycle_event(state, envelope, run_web_url)
                    .await;
            }
            _ => {}
        }
    }

    async fn handle_lifecycle_event(
        &self,
        state: &AppState,
        envelope: &EventEnvelope,
        run_web_url: Option<&str>,
    ) {
        let event = &envelope.event;
        let Some(details) = slack_lifecycle_details(event) else {
            return;
        };
        let event_name = event.body.event_name();
        let projection = match state.stores.runs.load_run_projection(&event.run_id).await {
            Ok(Some(projection)) => projection,
            Ok(None) => {
                warn!(
                    run_id = %event.run_id,
                    event = event_name,
                    "Skipping Slack lifecycle notification because run projection is missing"
                );
                return;
            }
            Err(err) => {
                warn!(
                    run_id = %event.run_id,
                    event = event_name,
                    error = %err,
                    "Skipping Slack lifecycle notification because run projection could not be loaded"
                );
                return;
            }
        };

        // Filter routes first; bail out before any further work if none match.
        let mut routes: Vec<_> = projection
            .spec
            .settings
            .run
            .notifications
            .iter()
            .filter(|(_, route)| {
                route.enabled
                    && route.provider.as_deref() == Some("slack")
                    && route.events.iter().any(|event| event == event_name)
            })
            .collect();
        if routes.is_empty() {
            return;
        }
        routes.sort_by_key(|(route_name, _)| *route_name);

        // Only completed/failed events need to recover prior PR details (a
        // run.started event cannot have a prior PullRequestCreated).
        let prior = if matches!(details.kind, slack_blocks::RunLifecycleKind::Started) {
            PriorSlackLifecycleEventDetails::default()
        } else {
            load_prior_slack_lifecycle_event_details(state, event.run_id, envelope.seq).await
        };
        let workflow_label = slack_lifecycle_workflow_label(
            projection.as_ref(),
            details
                .started_event_name
                .as_deref()
                .or(prior.started_event_name.as_deref()),
            event_name,
        );
        let pull_request = prior.pull_request.or_else(|| {
            projection
                .pull_request
                .as_ref()
                .map(slack_lifecycle_pull_request_from_link)
        });
        let run_id = event.run_id.to_string();
        let run_url = run_web_url.or(projection.web_url.as_deref());
        let pull_request_blocks =
            pull_request
                .as_ref()
                .map(|pull_request| slack_blocks::RunLifecyclePullRequest {
                    number: pull_request.number,
                    title:  pull_request.title.as_deref(),
                    url:    pull_request.url.as_deref(),
                });
        let blocks =
            slack_blocks::run_lifecycle_blocks(details.kind, &slack_blocks::RunLifecycleBlocks {
                run_id: &run_id,
                run_url,
                workflow_label: &workflow_label,
                result: details.result.as_deref(),
                duration_ms: details.duration_ms,
                pull_request: pull_request_blocks,
            });

        let blocks = &blocks;
        let posts = routes.into_iter().filter_map(|(route_name, route)| {
            let channel =
                resolve_slack_lifecycle_route_channel(event.run_id, route_name, route, event_name)?;
            Some(async move {
                if let Err(err) = self.client.post_message(&channel, blocks, None).await {
                    warn!(
                        run_id = %event.run_id,
                        event = event_name,
                        notification_route = route_name.as_str(),
                        error = %err,
                        "Failed to post Slack lifecycle notification"
                    );
                }
            })
        });
        join_all(posts).await;
    }

    async fn finish_interview(
        &self,
        run_id: RunId,
        qid: &str,
        question_text: &str,
        answer_text: &str,
    ) {
        let key = (run_id, qid.to_string());
        let posted = self
            .posted_messages
            .lock()
            .expect("slack posted messages lock poisoned")
            .remove(&key);
        let Some(posted) = posted else {
            return;
        };

        self.thread_registry.remove(&posted.ts);
        let blocks = slack_blocks::answered_blocks(question_text, answer_text);
        let _ = self
            .client
            .update_message(&posted.channel_id, &posted.ts, &blocks)
            .await;
    }

    async fn submit_answer(&self, state: Arc<AppState>, submission: SlackAnswerSubmission) {
        let Ok(run_id) = RunId::from_str(&submission.run_id) else {
            return;
        };

        let Ok(pending) = load_pending_interview(state.as_ref(), run_id, &submission.qid).await
        else {
            return;
        };
        let answer_submission = AnswerSubmission::new(submission.answer, submission.actor);
        let _ = submit_pending_interview_answer(state.as_ref(), &pending, answer_submission).await;
    }
}

fn slack_lifecycle_details(event: &RunEvent) -> Option<SlackLifecycleDetails> {
    match &event.body {
        EventBody::RunStarted(props) => Some(SlackLifecycleDetails {
            kind:               slack_blocks::RunLifecycleKind::Started,
            started_event_name: Some(props.name.clone()),
            result:             None,
            duration_ms:        None,
        }),
        EventBody::RunCompleted(props) => Some(SlackLifecycleDetails {
            kind:               slack_blocks::RunLifecycleKind::Completed,
            started_event_name: None,
            result:             Some(slack_lifecycle_completed_result(
                &props.status,
                props.reason,
            )),
            duration_ms:        Some(props.timing.wall_time_ms),
        }),
        EventBody::RunFailed(props) => Some(SlackLifecycleDetails {
            kind:               slack_blocks::RunLifecycleKind::Failed,
            started_event_name: None,
            result:             Some(slack_lifecycle_failed_result(&props.failure)),
            duration_ms:        Some(props.timing.wall_time_ms),
        }),
        _ => None,
    }
}

fn slack_lifecycle_completed_result(status: &str, reason: SuccessReason) -> String {
    let status = status.trim();
    let reason = reason.to_string();
    if status.is_empty() || status == reason {
        reason
    } else {
        format!("{status} — {reason}")
    }
}

fn slack_lifecycle_failed_result(failure: &fabro_types::RunFailure) -> String {
    let reason = failure.reason.to_string();
    let message = failure.detail.message.trim();
    if message.is_empty() {
        reason
    } else {
        format!("{reason} — {message}")
    }
}

async fn load_prior_slack_lifecycle_event_details(
    state: &AppState,
    run_id: RunId,
    before_seq: u32,
) -> PriorSlackLifecycleEventDetails {
    let run_store = match state.stores.runs.open_run_reader(&run_id).await {
        Ok(run_store) => run_store,
        Err(err) => {
            warn!(
                run_id = %run_id,
                error = %err,
                "Unable to inspect prior run events for Slack lifecycle notification"
            );
            return PriorSlackLifecycleEventDetails::default();
        }
    };
    let events = match run_store.list_events().await {
        Ok(events) => events,
        Err(err) => {
            warn!(
                run_id = %run_id,
                error = %err,
                "Unable to load prior run events for Slack lifecycle notification"
            );
            return PriorSlackLifecycleEventDetails::default();
        }
    };

    let mut details = PriorSlackLifecycleEventDetails::default();
    for envelope in events {
        if envelope.seq >= before_seq {
            break;
        }
        match envelope.event.body {
            EventBody::RunStarted(props) if !props.name.trim().is_empty() => {
                details.started_event_name = Some(props.name);
            }
            EventBody::PullRequestCreated(props) => {
                details.pull_request = Some(SlackLifecyclePullRequest {
                    number: props.pr_number,
                    title:  Some(props.title),
                    url:    Some(props.pr_url),
                });
            }
            _ => {}
        }
    }
    details
}

fn slack_lifecycle_workflow_label(
    projection: &fabro_store::RunProjection,
    started_event_name: Option<&str>,
    event_name: &str,
) -> String {
    [
        projection.spec.workflow_name(),
        projection.spec.workflow_slug(),
        projection.spec.graph_name(),
        started_event_name,
    ]
    .into_iter()
    .flatten()
    .map(str::trim)
    .find(|value| !value.is_empty())
    .unwrap_or(event_name)
    .to_string()
}

fn slack_lifecycle_pull_request_from_link(link: &PullRequestLink) -> SlackLifecyclePullRequest {
    SlackLifecyclePullRequest {
        number: link.number,
        title:  None,
        url:    Some(link.html_url()),
    }
}

fn resolve_slack_lifecycle_route_channel(
    run_id: RunId,
    route_name: &str,
    route: &NotificationRouteSettings,
    event_name: &str,
) -> Option<String> {
    let Some(channel) = route
        .slack
        .as_ref()
        .and_then(|slack| slack.channel.as_ref())
    else {
        warn!(
            run_id = %run_id,
            notification_route = route_name,
            event = event_name,
            "Skipping Slack lifecycle notification route without channel"
        );
        return None;
    };

    // `{{ vars.* }}` is substituted at run creation, so the channel is literal
    // here; anything still unresolved skips the route rather than sending to a
    // half-rendered channel name.
    let resolved = match channel.resolve_with(&mut fabro_types::settings::ResolveCtx::new()) {
        Ok(resolved) => resolved,
        Err(err) => {
            warn!(
                run_id = %run_id,
                notification_route = route_name,
                event = event_name,
                error = %err,
                "Skipping Slack lifecycle notification route with unresolved channel"
            );
            return None;
        }
    };
    if resolved.trim().is_empty() {
        warn!(
            run_id = %run_id,
            notification_route = route_name,
            event = event_name,
            "Skipping Slack lifecycle notification route with empty channel"
        );
        return None;
    }
    Some(resolved)
}

/// Shared application state for the server.
pub struct AppState {
    runs: Mutex<HashMap<RunId, ManagedRun>>,
    aggregate_billing: Mutex<BillingAccumulator>,
    pub(crate) stores: AppStores,
    session_runtimes: SessionRuntimeManager,
    artifact_store: ArtifactStore,
    automation_repo_cache: Arc<GitRepoCache>,
    #[cfg(any(test, feature = "test-support"))]
    automation_materializer_override: Option<Arc<dyn AutomationRunMaterializer>>,
    worker_tokens: WorkerTokenKeys,
    started_at: Instant,
    resource_sampler: resource_sampler::ResourceSampler,
    max_concurrent_runs: usize,
    pub(crate) worker_control_bus: Arc<dyn WorkerControlBus>,
    pub(crate) worker_runtime: Arc<dyn WorkerRuntime>,
    scheduler_notify: Notify,
    automation_scheduler_notify: Notify,
    pull_request_scheduler_notify: Notify,
    pull_request_creation_queue: Mutex<pull_request_supervisor::PendingPullRequestCreationQueue>,
    global_event_tx: broadcast::Sender<EventEnvelope>,
    /// Per-run coalescing registry for `GET /runs/{id}/files`. Concurrent
    /// callers for the same run share one materialization; different runs
    /// proceed in parallel. See `crate::run_files` for semantics.
    pub(crate) files_in_flight: FilesInFlight,
    pull_request_create_locks: KeyedMutex<RunId>,
    parent_link_lock: AsyncMutex<()>,

    pub(super) server_secrets: ServerSecrets,
    pub(crate) llm_source: Arc<dyn CredentialProvider>,
    manifest_run_defaults: RwLock<Arc<RunLayer>>,
    manifest_run_settings: RwLock<std::result::Result<RunNamespace, SharedError>>,
    pub(crate) server_settings: RwLock<Arc<ServerSettings>>,
    effective_web_url: RwLock<String>,
    catalog: RwLock<Arc<Catalog>>,
    pub(crate) env_lookup: EnvLookup,
    pub(crate) github_api_base_url: String,
    active_config_path: PathBuf,
    http_client: Option<fabro_http::HttpClient>,
    sandbox_inventory: SandboxInventory,
    shutdown: CancellationToken,
    shutting_down: AtomicBool,
    registry_factory_override: Option<Box<RegistryFactoryOverride>>,
    slack_service: Option<Arc<SlackService>>,
    slack_started: AtomicBool,
    github_webhook_secret: Option<String>,
}

pub(crate) struct AppStores {
    pub(crate) runs:            Arc<Database>,
    pub(crate) run_summaries:   Arc<RunSummaryStore>,
    /// Ask Fabro conversations, keyed by session id.
    pub(crate) session_records: Arc<RunSessionRecordStore>,
    pub(crate) auth_codes:      Arc<AuthCodeStore>,
    pub(crate) auth_sessions:   Arc<AuthSessionStore>,
    pub(crate) automations:     Arc<AutomationStore>,
    pub(crate) environments:    Arc<EnvironmentStore>,
    pub(crate) mcp_servers:     Arc<McpServerStore>,
    pub(crate) vault:           Arc<SecretStore>,
    pub(crate) variables:       Arc<VariableStore>,
}

#[cfg(any(test, feature = "test-support"))]
impl AppState {
    /// Access the auth session store so tests can seed CLI sessions against
    /// the same SQLite pool the router reads from.
    #[must_use]
    pub fn test_auth_session_store(&self) -> &Arc<AuthSessionStore> {
        &self.stores.auth_sessions
    }

    /// Access the auth-code store used by this router.
    #[must_use]
    pub fn test_auth_code_store(&self) -> &Arc<AuthCodeStore> {
        &self.stores.auth_codes
    }
}

impl AppState {
    pub(crate) fn automation_store(&self) -> &AutomationStore {
        &self.stores.automations
    }

    pub(crate) fn environment_store(&self) -> &EnvironmentStore {
        &self.stores.environments
    }

    pub(crate) fn mcp_server_store(&self) -> &McpServerStore {
        &self.stores.mcp_servers
    }

    pub(crate) async fn materialize_automation_run(
        &self,
        input: AutomationRunMaterializeInput,
    ) -> Result<AutomationRunMaterialized, RunMaterializeError> {
        #[cfg(any(test, feature = "test-support"))]
        if let Some(materializer) = self.automation_materializer_override.as_ref() {
            return materializer.materialize(input).await;
        }

        let settings = self.server_settings();
        let credentials = self
            .github_credentials(&settings.server.integrations.github)
            .await
            .map_err(|source| RunMaterializeError::LoadCredentials { source })?;
        ProductionAutomationRunMaterializer::new(
            credentials,
            self.github_api_base_url.clone(),
            self.http_client.clone(),
            Arc::clone(&self.automation_repo_cache),
            fabro_workflow_version::WorkflowVersionStore::new(self.store_ref().blobs()),
        )
        .materialize(input)
        .await
    }

    pub(crate) fn notify_automation_scheduler(&self) {
        self.automation_scheduler_notify.notify_one();
    }

    pub(crate) fn automation_scheduler_notified(
        &self,
    ) -> impl std::future::Future<Output = ()> + '_ {
        self.automation_scheduler_notify.notified()
    }

    pub(crate) fn notify_pull_request_scheduler(&self) {
        self.pull_request_scheduler_notify.notify_one();
    }

    pub(crate) fn pull_request_scheduler_notified(
        &self,
    ) -> impl std::future::Future<Output = ()> + '_ {
        self.pull_request_scheduler_notify.notified()
    }
}

pub(crate) struct AskFabroReadiness {
    default_model: Option<String>,
}

impl AskFabroReadiness {
    pub(crate) fn decorate(&self, mut run: fabro_types::Run) -> fabro_types::Run {
        run.ask_fabro = self.ask_fabro_for(&run);
        run
    }

    fn ask_fabro_for(&self, run: &fabro_types::Run) -> AskFabro {
        let unavailable_reason = if run.sandbox.is_none() {
            Some(AskFabroUnavailableReason::NoSandbox)
        } else if run
            .sandbox
            .as_ref()
            .and_then(fabro_types::RunSandbox::instance)
            .is_none()
        {
            Some(AskFabroUnavailableReason::SandboxNotReady)
        } else if self.default_model.is_none() {
            Some(AskFabroUnavailableReason::LlmUnconfigured)
        } else {
            None
        };

        AskFabro {
            available: unavailable_reason.is_none(),
            unavailable_reason,
            default_model: self.default_model.clone(),
        }
    }
}

pub(crate) struct AppStateConfig {
    pub(crate) resolved_settings: ResolvedAppStateSettings,
    pub(crate) registry_factory_override: Option<Box<RegistryFactoryOverride>>,
    pub(crate) max_concurrent_runs: usize,
    pub(crate) store: Arc<Database>,
    pub(crate) artifact_store: ArtifactStore,
    pub(crate) db_pool: DbPool,
    pub(crate) preloaded_vault: Vault,
    pub(crate) server_secrets: ServerSecrets,
    pub(crate) env_lookup: EnvLookup,
    pub(crate) github_api_base_url: Option<String>,
    pub(crate) active_config_path: PathBuf,
    pub(crate) http_client: Option<fabro_http::HttpClient>,
    pub(crate) sandbox_inventory: Option<SandboxInventory>,
    pub(crate) shutdown: CancellationToken,
    #[cfg(test)]
    pub(crate) worker_control_bus: Option<Arc<dyn WorkerControlBus>>,
    #[cfg(test)]
    pub(crate) worker_runtime: Option<Arc<dyn WorkerRuntime>>,
    #[cfg(any(test, feature = "test-support"))]
    pub(crate) automation_materializer_override: Option<Arc<dyn AutomationRunMaterializer>>,
}

#[derive(Clone)]
pub(crate) struct ResolvedAppStateSettings {
    pub(crate) server_settings:       ServerSettings,
    pub(crate) manifest_run_defaults: RunLayer,
    pub(crate) llm_overlay:           LlmLayer,
}

fn accumulate_billing_rollup(
    accumulator: &mut BillingAccumulator,
    rollup: &fabro_workflow::ProjectionBillingRollup,
) {
    accumulator.total_runs += 1;
    accumulator.total_timing = accumulator.total_timing.saturating_add(&rollup.timing);
    for model in &rollup.by_model {
        let entry = accumulator.by_model.entry(model.model.clone()).or_default();
        entry.stages += model.stages;
        entry.billing.add_counts(&model.billing);
    }
}

impl AppState {
    pub(crate) fn manifest_run_defaults(&self) -> Arc<RunLayer> {
        Arc::clone(
            &self
                .manifest_run_defaults
                .read()
                .expect("manifest run defaults lock poisoned"),
        )
    }

    pub(crate) fn server_settings(&self) -> Arc<ServerSettings> {
        Arc::clone(
            &self
                .server_settings
                .read()
                .expect("server settings lock poisoned"),
        )
    }

    pub(crate) fn catalog(&self) -> Arc<Catalog> {
        Arc::clone(&self.catalog.read().expect("catalog lock poisoned"))
    }

    pub(crate) fn active_config_path(&self) -> &std::path::Path {
        &self.active_config_path
    }

    pub(crate) fn manifest_run_settings(&self) -> std::result::Result<RunNamespace, SharedError> {
        self.manifest_run_settings
            .read()
            .expect("manifest run settings lock poisoned")
            .clone()
    }

    pub(crate) fn refresh_manifest_run_settings_from_catalogs(&self) {
        let manifest_run_defaults = self.manifest_run_defaults();
        let manifest_run_settings = resolve_manifest_run_settings_with_catalog(
            manifest_run_defaults.as_ref(),
            &self.stores.environments,
            &self.stores.mcp_servers,
        );
        *self
            .manifest_run_settings
            .write()
            .expect("manifest run settings lock poisoned") = manifest_run_settings;
    }

    pub(crate) fn refresh_manifest_run_settings_from_environment_catalog(&self) {
        self.refresh_manifest_run_settings_from_catalogs();
    }

    fn http_client(&self) -> Result<fabro_http::HttpClient, fabro_http::HttpClientBuildError> {
        match &self.http_client {
            Some(client) => Ok(client.clone()),
            None => fabro_http::http_client(),
        }
    }

    pub(crate) fn server_storage_dir(&self) -> PathBuf {
        PathBuf::from(&self.server_settings().server.storage.root)
    }

    /// Scratch directory used by the automation materializer when staging
    /// per-run manifests. Shared by API-triggered and scheduled fires.
    pub(crate) fn automation_temp_root(&self) -> PathBuf {
        Storage::new(self.server_storage_dir())
            .scratch_dir()
            .join("automations")
    }

    /// Snapshotted at create-time so attach replays surface the same link
    /// even if `server.web.url` is later changed. `None` when the UI is
    /// turned off or `server.web.url` is unset/invalid.
    pub(crate) fn run_web_url(&self, run_id: &fabro_types::RunId) -> Option<String> {
        if !self.server_settings().server.web.enabled {
            return None;
        }
        let base = self.canonical_origin().ok()?;
        Some(format!("{}/runs/{run_id}", base.trim_end_matches('/')))
    }

    pub(crate) async fn resolve_llm_client(&self) -> anyhow::Result<FabroClient> {
        resolve_llm_client_from_source(
            Arc::clone(&self.llm_source),
            self.catalog(),
            self.http_client.clone(),
        )
        .await
    }

    pub(crate) async fn configured_llm_provider_ids(&self) -> Vec<ProviderId> {
        let catalog = self.catalog();
        fabro_llm::configured_providers(catalog.as_ref(), self.llm_source.as_ref()).await
    }

    /// Resolve the LLM client once and derive the ready provider IDs from it,
    /// logging a warning when resolution fails. Callers that need both values
    /// must use this instead of `ready_llm_provider_ids` so the client is not
    /// resolved twice.
    pub(crate) async fn resolve_llm_client_with_ready_ids(
        &self,
    ) -> (anyhow::Result<FabroClient>, Vec<ProviderId>) {
        let llm_result = self.resolve_llm_client().await;
        if let Err(err) = &llm_result {
            warn!(error = ?err, "Failed to resolve LLM client while checking ready providers");
        }
        let ready_provider_ids = llm_result
            .as_ref()
            .map(FabroClient::provider_ids)
            .unwrap_or_default();
        (llm_result, ready_provider_ids)
    }

    pub(crate) async fn ready_llm_provider_ids(&self) -> Vec<ProviderId> {
        self.resolve_llm_client_with_ready_ids().await.1
    }

    pub(crate) async fn decorate_run_summary(&self, run: fabro_types::Run) -> fabro_types::Run {
        self.ask_fabro_readiness().await.decorate(run)
    }

    pub(crate) async fn decorate_run_summaries(
        &self,
        runs: Vec<fabro_types::Run>,
    ) -> Vec<fabro_types::Run> {
        let readiness = self.ask_fabro_readiness().await;
        runs.into_iter()
            .map(|run| readiness.decorate(run))
            .collect()
    }

    pub(crate) async fn ask_fabro_readiness(&self) -> AskFabroReadiness {
        let provider_ids = self.ready_llm_provider_ids().await;
        let default_model = if provider_ids.is_empty() {
            None
        } else {
            self.catalog()
                .default_offering_for(&provider_ids)
                .map(|entry| entry.model.id().to_string())
        };
        AskFabroReadiness { default_model }
    }

    pub(crate) async fn vault_secret(
        &self,
        name: &str,
    ) -> Result<Option<String>, SecretStoreError> {
        self.stores
            .vault
            .get(name)
            .await
            .map(|entry| entry.map(|entry| entry.value))
    }

    pub(crate) fn config_env_lookup(&self, name: &str) -> Option<String> {
        (self.env_lookup)(name)
    }

    /// Daytona credentials for `api_key`: the key from the vault, the
    /// control-plane URL and organization from server configuration, and
    /// the server's HTTP client. The process environment is consulted only
    /// through the configured lookup.
    pub(crate) fn daytona_credentials(&self, api_key: String) -> DaytonaCredentials {
        DaytonaCredentials::from_api_key(api_key, |name| self.config_env_lookup(name))
            .with_http_client(self.http_client().ok())
    }

    /// Everything a reconnect needs to reach a run's provider: the server's
    /// provider settings and the Daytona credentials from the vault (`None`
    /// when no key is stored).
    pub(crate) async fn provider_access(&self) -> Result<ProviderAccess, SecretStoreError> {
        Ok(ProviderAccess {
            providers: self.server_settings().server.sandbox.providers.clone(),
            daytona:   self
                .vault_secret(EnvVars::DAYTONA_API_KEY)
                .await?
                .map(|api_key| self.daytona_credentials(api_key)),
        })
    }

    pub(crate) async fn check_daytona_api_key(
        &self,
        api_key: String,
    ) -> anyhow::Result<daytona::DaytonaKeyCheck> {
        self.check_daytona_api_key_with_timeout(api_key, daytona::DAYTONA_CREDENTIAL_PROBE_TIMEOUT)
            .await
    }

    pub(crate) async fn check_daytona_api_key_with_timeout(
        &self,
        api_key: String,
        probe_timeout: Duration,
    ) -> anyhow::Result<daytona::DaytonaKeyCheck> {
        daytona::check_daytona_api_key(&self.daytona_credentials(api_key), probe_timeout).await
    }

    /// Borrow the persistent store so sibling modules can open run readers
    /// without cross-module state coupling on the `AppState` field layout.
    pub(crate) fn store_ref(&self) -> &Arc<Database> {
        &self.stores.runs
    }

    /// Loads the current projection for `run_id`, with the standard HTTP error
    /// mapping: storage failures become 500s and a missing run becomes the
    /// canonical 404.
    pub(crate) async fn load_run_projection(
        &self,
        run_id: &RunId,
    ) -> Result<Arc<fabro_store::RunProjection>, ApiError> {
        self.stores
            .runs
            .load_run_projection(run_id)
            .await
            .map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?
            .ok_or_else(|| ApiError::not_found("Run not found."))
    }

    pub(crate) fn session_runtimes(&self) -> &SessionRuntimeManager {
        &self.session_runtimes
    }

    pub(crate) fn sandbox_inventory(&self) -> &SandboxInventory {
        &self.sandbox_inventory
    }

    pub(crate) fn server_secret(&self, name: &str) -> Option<String> {
        self.server_secrets.get(name)
    }

    pub(crate) fn worker_token_keys(&self) -> &WorkerTokenKeys {
        &self.worker_tokens
    }

    /// Loopback target this server is bound to, derived from the runtime
    /// daemon record. Used by in-process Ask Fabro sessions to call the local
    /// API over the normal HTTP path (authed with a same-run worker token).
    pub(crate) fn self_server_target(&self) -> anyhow::Result<fabro_client::ServerTarget> {
        let storage_dir = self.server_storage_dir();
        let runtime_directory = Storage::new(&storage_dir).runtime_directory();
        let daemon = ServerDaemon::read(&runtime_directory)?.with_context(|| {
            format!(
                "server record {} is missing",
                runtime_directory.record_path().display()
            )
        })?;
        // `Bind::to_target()` already produces the http(s)-URL-or-absolute-
        // socket-path form that `ServerTarget`'s FromStr understands.
        daemon.bind.to_target().parse()
    }

    pub(crate) fn effective_web_url(&self) -> String {
        self.effective_web_url
            .read()
            .expect("effective web url lock poisoned")
            .clone()
    }

    pub(crate) fn canonical_origin(&self) -> Result<String, String> {
        canonical_origin_from_effective_web_url(&self.effective_web_url())
    }

    pub(crate) fn session_key(&self) -> Option<Key> {
        self.server_secret(EnvVars::SESSION_SECRET)
            .and_then(|value| auth::derive_cookie_key(value.as_bytes()).ok())
    }

    pub(crate) async fn github_credentials(
        &self,
        settings: &GithubIntegrationSettings,
    ) -> anyhow::Result<Option<fabro_github::GitHubCredentials>> {
        match settings.strategy {
            GithubIntegrationStrategy::App => {
                let Some(app_id) = settings.app_id.clone() else {
                    return Ok(None);
                };
                let raw = self
                    .vault_secret(EnvVars::GITHUB_APP_PRIVATE_KEY)
                    .await
                    .map_err(anyhow::Error::new)?;
                let Some(raw) = raw else {
                    return Ok(None);
                };
                let private_key_pem = decode_secret_pem(EnvVars::GITHUB_APP_PRIVATE_KEY, &raw)
                    .map_err(anyhow::Error::msg)?;
                Ok(Some(fabro_github::GitHubCredentials::App(
                    fabro_github::GitHubAppCredentials {
                        app_id,
                        private_key_pem,
                        slug: settings.slug.clone(),
                    },
                )))
            }
            GithubIntegrationStrategy::Token => {
                let token = self
                    .vault_secret(EnvVars::GITHUB_TOKEN)
                    .await
                    .map_err(anyhow::Error::new)?
                    .as_deref()
                    .map(str::trim)
                    .filter(|token| !token.is_empty())
                    .map(str::to_string);
                match token {
                    Some(token) => {
                        fabro_github::validate_static_github_token(&token)
                            .map_err(anyhow::Error::msg)?;
                        Ok(Some(fabro_github::GitHubCredentials::Pat(token)))
                    }
                    None => anyhow::bail!(
                        "GITHUB_TOKEN not configured -- run fabro install or run fabro secret set GITHUB_TOKEN"
                    ),
                }
            }
        }
    }

    fn begin_shutdown(&self) {
        self.shutting_down.store(true, Ordering::Relaxed);
        self.scheduler_notify.notify_waiters();
        self.automation_scheduler_notify.notify_waiters();
        self.pull_request_scheduler_notify.notify_waiters();
    }

    pub(crate) fn shutdown_token(&self) -> CancellationToken {
        self.shutdown.clone()
    }

    fn is_shutting_down(&self) -> bool {
        self.shutting_down.load(Ordering::Relaxed)
    }

    pub(crate) fn replace_runtime_settings(
        &self,
        resolved_settings: ResolvedAppStateSettings,
    ) -> anyhow::Result<()> {
        let ResolvedAppStateSettings {
            server_settings,
            manifest_run_defaults,
            llm_overlay,
        } = resolved_settings;
        let server_settings = Arc::new(server_settings);
        let manifest_run_defaults = Arc::new(manifest_run_defaults);
        let effective_web_url =
            effective_web_url(&server_settings.server, |name| (self.env_lookup)(name));
        let manifest_run_settings = resolve_manifest_run_settings_with_catalog(
            manifest_run_defaults.as_ref(),
            &self.stores.environments,
            &self.stores.mcp_servers,
        );
        let catalog = Arc::new(
            fabro_llm::build_catalog(&llm_overlay, &|name| (self.env_lookup)(name))
                .context("building LLM model catalog")?,
        );
        canonical_origin_from_effective_web_url(&effective_web_url).map_err(anyhow::Error::msg)?;

        *self
            .manifest_run_defaults
            .write()
            .expect("manifest run defaults lock poisoned") = manifest_run_defaults;
        *self
            .manifest_run_settings
            .write()
            .expect("manifest run settings lock poisoned") = manifest_run_settings;
        *self
            .server_settings
            .write()
            .expect("server settings lock poisoned") = server_settings;
        *self
            .effective_web_url
            .write()
            .expect("effective web url lock poisoned") = effective_web_url;
        *self.catalog.write().expect("catalog lock poisoned") = catalog;
        Ok(())
    }
}

/// Builds the server's LLM client: retries and attachment inlining on, the
/// server's HTTP client for provider requests when one is configured.
async fn resolve_llm_client_from_source(
    source: Arc<dyn CredentialProvider>,
    catalog: Arc<Catalog>,
    http_client: Option<fabro_http::HttpClient>,
) -> anyhow::Result<FabroClient> {
    let mut options = ClientOptions::standard();
    options.http = http_client;
    fabro_llm::build_client(Catalog::clone(&catalog), source, options)
        .await
        .context("building the LLM client")
}

fn decode_secret_pem(name: &str, raw: &str) -> Result<String, String> {
    if raw.starts_with("-----") {
        return Ok(raw.to_string());
    }
    let pem_bytes = BASE64_STANDARD
        .decode(raw)
        .map_err(|err| format!("{name} is not valid PEM or base64: {err}"))?;
    String::from_utf8(pem_bytes)
        .map_err(|err| format!("{name} base64 decoded to invalid UTF-8: {err}"))
}

fn start_optional_slack_service(state: &Arc<AppState>) {
    let Some(service) = state.slack_service.clone() else {
        return;
    };
    if state.slack_started.swap(true, Ordering::SeqCst) {
        return;
    }

    let event_state = Arc::clone(state);
    let event_service = Arc::clone(&service);
    tokio::spawn(async move {
        let mut rx = event_state.global_event_tx.subscribe();
        loop {
            match rx.recv().await {
                Ok(envelope) => {
                    // Resolve the run's web URL once per event so the Slack
                    // message can deep-link back to Fabro. Returns None when
                    // the web UI is disabled or `server.web.url` is unset, in
                    // which case `question_to_blocks` simply omits the link.
                    let run_web_url = event_state.run_web_url(&envelope.event.run_id);
                    event_service
                        .handle_event(event_state.as_ref(), &envelope, run_web_url.as_deref())
                        .await;
                }
                Err(RecvError::Lagged(_)) => {}
                Err(RecvError::Closed) => break,
            }
        }
    });

    let socket_state = Arc::clone(state);
    tokio::spawn(async move {
        let submit_service = Arc::clone(&service);
        let on_submit: Arc<dyn Fn(SlackAnswerSubmission) + Send + Sync> =
            Arc::new(move |submission| {
                let state = Arc::clone(&socket_state);
                let service = Arc::clone(&submit_service);
                tokio::spawn(async move {
                    service.submit_answer(state, submission).await;
                });
            });
        slack_connection::run_with_status(
            &service.client,
            &service.app_token,
            &service.thread_registry,
            on_submit,
            service.status_sink(),
        )
        .await;
    });
}

/// Build the axum Router with all run endpoints and embedded static assets.
#[allow(
    clippy::needless_pass_by_value,
    reason = "Public router helper keeps the existing ergonomic API and forwards by reference."
)]
pub fn build_router(state: Arc<AppState>, auth_mode: AuthMode) -> Router {
    build_router_with_options(state, &auth_mode, RouterOptions::default())
}

#[derive(Clone, Debug)]
pub struct RouterOptions {
    pub web_enabled:       bool,
    pub static_asset_root: Option<PathBuf>,
    pub github_endpoints:  Option<Arc<GithubEndpoints>>,
    /// Set when serving with the `--watch-web` dev flag. The static-file
    /// handler then refuses to fall back to the embedded SPA snapshot and
    /// returns a 503 "build in progress" page on miss, so developers see
    /// their edits or a clear signal — never stale embedded bytes.
    pub watch_web:         bool,
}

impl Default for RouterOptions {
    fn default() -> Self {
        Self {
            web_enabled:       true,
            static_asset_root: None,
            github_endpoints:  None,
            watch_web:         false,
        }
    }
}

fn removed_web_route(path: &str) -> bool {
    matches!(path, "/setup/complete") || path.starts_with("/install")
}

/// Build the axum Router with configurable web surface routing.
pub fn build_router_with_options(
    state: Arc<AppState>,
    auth_mode: &AuthMode,
    options: RouterOptions,
) -> Router {
    start_optional_slack_service(&state);
    let RouterOptions {
        web_enabled,
        static_asset_root,
        github_endpoints,
        watch_web,
    } = options;
    let translation_state = Arc::clone(&state);
    let state_for_canonical_host = Arc::clone(&state);
    let github_endpoints =
        github_endpoints.unwrap_or_else(|| Arc::new(GithubEndpoints::production_defaults()));
    let webhook_secret = state.github_webhook_secret.clone();
    let principal_layer = middleware::from_fn_with_state(Arc::clone(&state), principal_middleware);
    let api_common = if web_enabled {
        Router::new()
            .route("/openapi.json", get(handler::openapi_spec))
            .merge(web_auth::api_routes())
    } else {
        Router::new().route("/openapi.json", get(handler::openapi_spec))
    };

    let demo_router = Router::new()
        .nest(
            "/api/v1",
            api_common
                .clone()
                .merge(handler::demo_routes())
                .layer(principal_layer.clone()),
        )
        .layer(axum::Extension(auth_mode.clone()))
        .layer(axum::Extension(Arc::clone(&github_endpoints)))
        .with_state(state.clone());

    let mut real_router = Router::new().nest(
        "/api/v1",
        api_common
            .merge(handler::real_routes())
            .layer(principal_layer),
    );
    if web_enabled {
        real_router = real_router.nest("/auth", web_auth::routes().merge(auth::web_routes()));
    }
    let real_router = real_router
        .layer(axum::Extension(github_endpoints))
        .with_state(state);

    let dispatch = service_fn(move |req: axum_extract::Request| {
        let demo = demo_router.clone();
        let real = real_router.clone();
        async move {
            let demo_active = web_enabled
                && req.uri().path().starts_with("/api/")
                && req.headers().get("x-fabro-demo").is_some_and(|v| v == "1");
            if demo_active {
                demo.oneshot(req).await
            } else {
                real.oneshot(req).await
            }
        }
    });

    let mut app_router = Router::new()
        .route("/health", get(handler::health))
        .fallback_service(service_fn(move |req: axum_extract::Request| {
            let dispatch = dispatch.clone();
            let static_asset_root = static_asset_root.clone();
            async move {
                let path = req.uri().path().to_string();
                let dispatch_path = path.starts_with("/api/")
                    || path == "/health"
                    || (web_enabled && path.starts_with("/auth/"));
                if dispatch_path {
                    dispatch.oneshot(req).await
                } else if web_enabled && removed_web_route(&path) {
                    Ok::<_, std::convert::Infallible>(StatusCode::NOT_FOUND.into_response())
                } else if web_enabled && matches!(req.method(), &Method::GET | &Method::HEAD) {
                    let headers = req.headers().clone();
                    Ok::<_, std::convert::Infallible>(
                        static_files::serve_with_asset_root(
                            &path,
                            &headers,
                            static_asset_root.as_deref(),
                            watch_web,
                        )
                        .await,
                    )
                } else {
                    Ok::<_, std::convert::Infallible>(StatusCode::NOT_FOUND.into_response())
                }
            }
        }));

    app_router = app_router.layer(middleware::from_fn_with_state(
        translation_state,
        auth_translation_middleware,
    ));
    app_router = app_router.layer(middleware::from_fn(demo_routing_middleware));
    app_router = app_router.layer(axum::Extension(auth_mode.clone()));

    let mut router = app_router;
    if let Some(secret) = webhook_secret {
        let secret: Arc<[u8]> = Arc::from(secret.into_bytes().into_boxed_slice());
        router = github_webhook_routes(secret).merge(router);
    }

    router
        // Innermost of the outer layers so every response body — static SPA
        // assets and JSON API alike — is compressed before the header/log
        // middlewares see it.
        .layer(compression_layer())
        .layer(middleware::from_fn_with_state(
            canonical_host::Config {
                state: state_for_canonical_host,
                web_enabled,
            },
            canonical_host::redirect_middleware,
        ))
        .layer(middleware::from_fn(security_headers::layer))
        .layer(middleware::from_fn(http_log_middleware))
        .layer(middleware::from_fn(request_id::layer))
}

/// Response-compression layer shared by the main and install-mode routers.
///
/// The default predicate skips streaming SSE (`text/event-stream`), gRPC,
/// images, ZIP archives, and tiny bodies. The quality is pinned because
/// tower-http's default defers to each codec's own default, and brotli's is
/// quality 11 — seconds of CPU on a multi-megabyte asset. Level 4 keeps both
/// codecs fast at a near-optimal ratio.
pub(crate) fn compression_layer() -> CompressionLayer<impl Predicate> {
    CompressionLayer::new()
        .quality(CompressionLevel::Precise(4))
        .compress_when(DefaultPredicate::new().and(NotForContentType::const_new("application/zip")))
}

async fn http_log_middleware(mut req: axum_extract::Request, next: Next) -> Response {
    let path = req.uri().path();
    if path.starts_with("/assets/") || path.starts_with("/images/") {
        return next.run(req).await;
    }
    let method = req.method().clone();
    let path = req.uri().path().to_string();
    let request_id = req
        .extensions()
        .get::<RequestId>()
        .copied()
        .map(RequestId::render)
        .unwrap_or_default();
    let auth_slot = AuthContextSlot::initial();
    req.extensions_mut().insert(auth_slot.clone());
    let start = std::time::Instant::now();
    let response = next.run(req).await;
    let status = response.status().as_u16();
    let latency_ms = start.elapsed().as_millis();
    let auth_context = auth_slot.log_snapshot();
    let principal_kind = auth_context
        .principal
        .as_ref()
        .map_or("none", Principal::kind);
    let auth_status = auth_context.auth_status.as_str();

    macro_rules! emit_http_log {
        ($level:ident $(, $field:ident = $value:expr)* $(,)?) => {{
            if let Some(auth_error_code) = auth_context.auth_error_code {
                let auth_error_code = auth_error_code.as_str();
                $level!(
                    %method,
                    %path,
                    status,
                    latency_ms,
                    request_id = %request_id,
                    principal_kind,
                    auth_status,
                    auth_error_code,
                    $($field = $value,)*
                    "HTTP response"
                );
            } else {
                $level!(
                    %method,
                    %path,
                    status,
                    latency_ms,
                    request_id = %request_id,
                    principal_kind,
                    auth_status,
                    $($field = $value,)*
                    "HTTP response"
                );
            }
        }};
    }

    macro_rules! emit_principal_http_log {
        ($level:ident) => {{
            match &auth_context.principal {
                Some(Principal::User(user)) => emit_http_log!(
                    $level,
                    user_auth_method = user.auth_method.as_str(),
                    idp_issuer = user.identity.issuer(),
                    idp_subject = user.identity.subject(),
                    login = user.login.as_str(),
                ),
                Some(Principal::Worker { run_id }) => {
                    emit_http_log!($level, run_id = run_id.to_string().as_str(),)
                }
                Some(Principal::Webhook { delivery_id }) => {
                    emit_http_log!($level, delivery_id = delivery_id.as_str(),)
                }
                Some(Principal::Slack {
                    team_id, user_id, ..
                }) => emit_http_log!(
                    $level,
                    team_id = team_id.as_str(),
                    user_id = user_id.as_str(),
                ),
                None | Some(Principal::Agent { .. } | Principal::System { .. }) => {
                    emit_http_log!($level)
                }
            }
        }};
    }

    if status >= 500 {
        emit_principal_http_log!(error);
    } else {
        emit_principal_http_log!(info);
    }
    response
}

fn github_webhook_routes(secret: Arc<[u8]>) -> Router {
    Router::new()
        .route(WEBHOOK_ROUTE, post(github_webhook))
        .with_state(secret)
}

async fn github_webhook(
    State(secret): State<Arc<[u8]>>,
    RequestAuth(auth_slot): RequestAuth,
    headers: HeaderMap,
    body: Bytes,
) -> StatusCode {
    let delivery_id = headers
        .get("x-github-delivery")
        .and_then(|value| value.to_str().ok())
        .unwrap_or("unknown");

    let Some(signature) = headers
        .get("x-hub-signature-256")
        .and_then(|value| value.to_str().ok())
    else {
        auth_slot.replace(RequestAuthContext::invalid());
        warn!(delivery = %delivery_id, "Webhook missing X-Hub-Signature-256 header");
        return StatusCode::UNAUTHORIZED;
    };

    if !verify_signature(&secret, &body, signature) {
        auth_slot.replace(RequestAuthContext::invalid());
        warn!(delivery = %delivery_id, "Webhook HMAC signature mismatch");
        return StatusCode::UNAUTHORIZED;
    }

    auth_slot.replace(RequestAuthContext::authenticated(
        Principal::Webhook {
            delivery_id: delivery_id.to_string(),
        },
        None,
    ));

    let event_type = headers
        .get("x-github-event")
        .and_then(|value| value.to_str().ok())
        .unwrap_or("unknown");

    if tracing::enabled!(tracing::Level::DEBUG) {
        let (repo, action) = parse_event_metadata(&body);
        debug!(
            event = %event_type,
            delivery = %delivery_id,
            repo = %repo,
            action = %action,
            "Webhook received"
        );
    } else {
        info!(
            event = %event_type,
            delivery = %delivery_id,
            "Webhook received"
        );
    }

    StatusCode::OK
}

struct PrunePlan {
    run_ids:          Vec<RunId>,
    rows:             Vec<PruneRunEntry>,
    total_size_bytes: u64,
}

#[expect(
    clippy::disallowed_methods,
    reason = "sync helper invoked from async handler via spawn_blocking (see callers at :1301 / :1341)"
)]
fn build_disk_usage_response(
    summaries: &[fabro_types::Run],
    storage_dir: &std::path::Path,
    verbose: bool,
) -> anyhow::Result<DiskUsageResponse> {
    let scratch_base_dir = scratch_base(storage_dir);
    let logs_base_dir = Storage::new(storage_dir).runtime_directory().logs_dir();
    let runs = scan_runs_with_summaries(summaries, &scratch_base_dir)?;

    let mut active_count = 0u64;
    let mut total_run_size = 0u64;
    let mut reclaimable_run_size = 0u64;
    let mut run_rows = Vec::new();

    for run in &runs {
        let size = dir_size(&run.path);
        total_run_size += size;
        if run.status().is_active() {
            active_count += 1;
        } else {
            reclaimable_run_size += size;
        }
        if verbose {
            run_rows.push(DiskUsageRunRow {
                run_id:        Some(run.run_id().to_string()),
                workflow_name: Some(run.workflow_display_name()),
                status:        Some(run.status().to_string()),
                start_time:    Some(run.start_time()),
                size_bytes:    Some(to_i64(size)),
                reclaimable:   Some(!run.status().is_active()),
            });
        }
    }

    let mut log_count = 0u64;
    let mut total_log_size = 0u64;
    if let Ok(entries) = std::fs::read_dir(logs_base_dir) {
        for entry in entries.flatten() {
            let path = entry.path();
            if !path.is_file() || path.extension().is_none_or(|ext| ext != "log") {
                continue;
            }
            if let Ok(metadata) = path.metadata() {
                log_count += 1;
                total_log_size += metadata.len();
            }
        }
    }

    // Measure the whole storage tree so the managed total can't drift as new
    // subdirectories are added. "other" is the residual (database, artifacts,
    // sessions, vaults) — everything that isn't an enumerated run or log file.
    let managed_size = dir_size(storage_dir);
    let other_size = managed_size.saturating_sub(total_run_size + total_log_size);

    Ok(DiskUsageResponse {
        summary:                 vec![
            DiskUsageSummaryRow {
                type_:             Some("runs".to_string()),
                count:             Some(to_i64(runs.len())),
                active:            Some(to_i64(active_count)),
                size_bytes:        Some(to_i64(total_run_size)),
                reclaimable_bytes: Some(to_i64(reclaimable_run_size)),
            },
            DiskUsageSummaryRow {
                type_:             Some("logs".to_string()),
                count:             Some(to_i64(log_count)),
                active:            None,
                size_bytes:        Some(to_i64(total_log_size)),
                reclaimable_bytes: Some(to_i64(total_log_size)),
            },
            DiskUsageSummaryRow {
                type_:             Some("other".to_string()),
                count:             None,
                active:            None,
                size_bytes:        Some(to_i64(other_size)),
                reclaimable_bytes: Some(0),
            },
        ],
        total_size_bytes:        Some(to_i64(managed_size)),
        total_reclaimable_bytes: Some(to_i64(reclaimable_run_size + total_log_size)),
        runs:                    verbose.then_some(run_rows),
    })
}

fn build_prune_plan(
    request: &PruneRunsRequest,
    summaries: &[fabro_types::Run],
    storage_dir: &std::path::Path,
) -> anyhow::Result<PrunePlan> {
    let scratch_base_dir = scratch_base(storage_dir);
    let runs = scan_runs_with_summaries(summaries, &scratch_base_dir)?;
    let label_filters = request
        .labels
        .iter()
        .map(|(key, value)| (key.clone(), value.clone()))
        .collect::<Vec<_>>();

    let mut filtered = filter_runs(
        &runs,
        request.before.as_deref(),
        request.workflow.as_deref(),
        &label_filters,
        request.orphans,
        StatusFilter::All,
    );

    let has_explicit_filters =
        request.before.is_some() || request.workflow.is_some() || !label_filters.is_empty();
    let staleness_threshold = if let Some(duration) = request.older_than.as_deref() {
        Some(parse_system_duration(duration)?)
    } else if !has_explicit_filters {
        Some(chrono::Duration::hours(24))
    } else {
        None
    };

    if let Some(threshold) = staleness_threshold {
        let cutoff = chrono::Utc::now() - threshold;
        filtered.retain(|run| {
            run.end_time
                .or(run.start_time_dt)
                .is_some_and(|time| time < cutoff)
        });
    }

    filtered.retain(|run| !run.status().is_active());

    let rows = filtered
        .iter()
        .map(|run| PruneRunEntry {
            run_id:        Some(run.run_id().to_string()),
            dir_name:      Some(run.dir_name.clone()),
            workflow_name: Some(run.workflow_display_name()),
            size_bytes:    Some(to_i64(dir_size(&run.path))),
        })
        .collect::<Vec<_>>();
    let total_size_bytes = rows
        .iter()
        .map(|row| row.size_bytes.unwrap_or_default())
        .sum::<i64>()
        .max(0)
        .try_into()
        .unwrap_or_default();

    Ok(PrunePlan {
        run_ids: filtered.iter().map(RunInfo::run_id).collect(),
        rows,
        total_size_bytes,
    })
}

fn resolve_manifest_run_settings_with_catalog(
    manifest_run_defaults: &RunLayer,
    environment_store: &EnvironmentStore,
    mcp_server_store: &McpServerStore,
) -> std::result::Result<RunNamespace, SharedError> {
    WorkflowSettingsBuilder::new()
        .server_manifest_defaults(
            manifest_run_defaults.clone(),
            (*environment_store.catalog_layer()).clone(),
        )
        .server_mcp_catalog(mcp_server_store.catalog_settings())
        .build()
        .map(|settings| settings.run)
        .map_err(|err| SharedError::new(anyhow::Error::msg(err.to_string())))
}

fn system_sandbox_provider(
    manifest_run_settings: &std::result::Result<RunNamespace, SharedError>,
) -> String {
    manifest_run_settings.as_ref().map_or_else(
        |_| SandboxProviderKind::default().to_string(),
        |settings| settings.environment.provider.to_string(),
    )
}

fn parse_system_duration(raw: &str) -> anyhow::Result<chrono::Duration> {
    let raw = raw.trim();
    anyhow::ensure!(!raw.is_empty(), "empty duration string");
    let (num_str, unit) = raw.split_at(raw.len().saturating_sub(1));
    let amount = num_str.parse::<u64>()?;
    match unit {
        "h" => Ok(chrono::Duration::hours(
            i64::try_from(amount).unwrap_or(i64::MAX),
        )),
        "d" => Ok(chrono::Duration::days(
            i64::try_from(amount).unwrap_or(i64::MAX),
        )),
        _ => anyhow::bail!("invalid duration unit '{unit}' in '{raw}' (expected 'h' or 'd')"),
    }
}

fn dir_size(path: &std::path::Path) -> u64 {
    walkdir::WalkDir::new(path)
        .into_iter()
        .filter_map(std::result::Result::ok)
        .filter_map(|entry| entry.metadata().ok())
        .filter(std::fs::Metadata::is_file)
        .map(|metadata| metadata.len())
        .sum()
}

fn to_i64<T>(value: T) -> i64
where
    i64: TryFrom<T>,
{
    i64::try_from(value).unwrap_or(i64::MAX)
}

fn worker_token_keys_from_server_secrets(
    server_secrets: &ServerSecrets,
) -> anyhow::Result<WorkerTokenKeys> {
    let session_secret = server_secrets
        .get(EnvVars::SESSION_SECRET)
        .ok_or_else(|| jwt_auth::session_secret_key_error(&auth::KeyDeriveError::Empty))?;
    WorkerTokenKeys::from_master_secret(session_secret.as_bytes())
        .map_err(|err| jwt_auth::session_secret_key_error(&err))
}

fn build_sandbox_inventory(
    server_settings: &ServerSettings,
    daytona_api_key: Option<String>,
    env_lookup: &EnvLookup,
    http_client: Option<fabro_http::HttpClient>,
) -> SandboxInventory {
    let provider_settings = &server_settings.server.sandbox.providers;
    let mut inventory = SandboxInventory::empty();

    if provider_settings.is_enabled(&SandboxProviderKind::LOCAL) {
        inventory = inventory.with_host_directories(SandboxProviderKind::LOCAL);
    }

    if let Some(docker) = provider_settings.get(&SandboxProviderKind::DOCKER) {
        if docker.enabled {
            inventory = inventory.with_lazy(
                SandboxProviderKind::DOCKER,
                docker.clone(),
                ProviderConnectOptions::default(),
            );
        }
    }

    if let Some(daytona) = provider_settings.get(&SandboxProviderKind::DAYTONA) {
        if let Some(api_key) = daytona_api_key.filter(|_| daytona.enabled) {
            let credentials = DaytonaCredentials::from_api_key(api_key, |name| env_lookup(name))
                .with_http_client(http_client);
            inventory = inventory.with_lazy(
                SandboxProviderKind::DAYTONA,
                daytona.clone(),
                ProviderConnectOptions {
                    host_registry_root: None,
                    daytona:            Some(credentials),
                },
            );
        }
    }

    inventory
}

pub(crate) fn automation_dir_for_active_config(active_config_path: &std::path::Path) -> PathBuf {
    active_config_path
        .parent()
        .unwrap_or_else(|| std::path::Path::new("."))
        .join("automations")
}

fn mcp_server_dir_for_active_config(active_config_path: &std::path::Path) -> PathBuf {
    active_config_path
        .parent()
        .unwrap_or_else(|| std::path::Path::new("."))
        .join("mcps")
}

#[expect(
    clippy::disallowed_methods,
    reason = "synchronous app-state assembly may run inside an async runtime; a short-lived OS \
              thread avoids nested Tokio runtimes"
)]
fn load_store_blocking<T, F, Fut>(description: &'static str, load: F) -> anyhow::Result<T>
where
    T: Send + 'static,
    F: FnOnce() -> Fut + Send + 'static,
    Fut: std::future::Future<Output = anyhow::Result<T>>,
{
    std::thread::spawn(move || {
        let runtime = TokioRuntimeBuilder::new_current_thread()
            .enable_all()
            .build()
            .with_context(|| format!("build {description} runtime"))?;
        runtime.block_on(load())
    })
    .join()
    .expect("store load thread should not panic")
}

pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result<Arc<AppState>> {
    let AppStateConfig {
        resolved_settings,
        registry_factory_override,
        max_concurrent_runs,
        store,
        artifact_store,
        db_pool,
        preloaded_vault,
        server_secrets,
        env_lookup,
        github_api_base_url,
        active_config_path,
        http_client,
        sandbox_inventory,
        shutdown,
        #[cfg(test)]
        worker_control_bus,
        #[cfg(test)]
        worker_runtime,
        #[cfg(any(test, feature = "test-support"))]
        automation_materializer_override,
    } = config;

    let automation_migration_pool = db_pool.clone();
    load_store_blocking("automation environment migration", move || async move {
        fabro_automation::backfill_environment_selectors(&automation_migration_pool)
            .await
            .map_err(anyhow::Error::new)
    })
    .context("backfill automation environment selectors")?;
    let automation_store = Arc::new(AutomationStore::new(db_pool.clone()));
    let local_provider_enabled = resolved_settings
        .server_settings
        .server
        .sandbox
        .providers
        .is_enabled(&SandboxProviderKind::LOCAL);
    let environment_pool = db_pool.clone();
    let environment_store = Arc::new(
        load_store_blocking("environment store", move || async move {
            EnvironmentStore::load(environment_pool, local_provider_enabled)
                .await
                .map_err(anyhow::Error::new)
        })
        .context("load environments")?,
    );
    let run_summaries = store.run_summary_store();
    let auth_codes = Arc::new(AuthCodeStore::new(db_pool.clone()));
    let auth_sessions = Arc::new(AuthSessionStore::new(db_pool.clone()));
    let mcp_server_dir = mcp_server_dir_for_active_config(&active_config_path);
    let mcp_server_pool = db_pool.clone();
    let mcp_server_store = Arc::new(
        load_store_blocking("MCP server store", move || async move {
            McpServerStore::open(mcp_server_pool, mcp_server_dir)
                .await
                .map_err(anyhow::Error::new)
        })
        .context("load mcp servers")?,
    );
    let variables = Arc::new(VariableStore::new(db_pool.clone()));
    let session_records = Arc::new(RunSessionRecordStore::new(db_pool.clone()));
    let secret_store = Arc::new(SecretStore::new(db_pool));
    let vault = preloaded_vault;
    // Read vault secrets needed for synchronous setup before we wrap the vault in
    // an async lock for the rest of AppState.
    let daytona_api_key = vault.get(EnvVars::DAYTONA_API_KEY).map(str::to_string);
    let llm_source: Arc<dyn CredentialProvider> = Arc::new(SqlVaultCredentialSource::vault_only(
        Arc::clone(&secret_store),
    ));
    let (global_event_tx, _) = broadcast::channel(4096);
    let current_server_settings = Arc::new(resolved_settings.server_settings);
    let current_effective_web_url =
        effective_web_url(&current_server_settings.server, |name| env_lookup(name));
    let current_manifest_run_defaults = Arc::new(resolved_settings.manifest_run_defaults);
    let current_manifest_run_settings = resolve_manifest_run_settings_with_catalog(
        current_manifest_run_defaults.as_ref(),
        &environment_store,
        &mcp_server_store,
    );
    let current_catalog = Arc::new(
        fabro_llm::build_catalog(&resolved_settings.llm_overlay, &|name| env_lookup(name))
            .context("building LLM model catalog")?,
    );
    let sandbox_inventory = sandbox_inventory.unwrap_or_else(|| {
        build_sandbox_inventory(
            current_server_settings.as_ref(),
            daytona_api_key,
            &env_lookup,
            http_client.clone(),
        )
    });
    let slack_service = {
        let slack_settings = &current_server_settings.server.integrations.slack;
        if slack_settings.enabled {
            let default_channel = slack_settings.default_channel.clone();
            match resolve_slack_credentials_status_with_lookup(|name| {
                vault.get(name).map(str::to_string)
            }) {
                SlackCredentialResolution::Configured(credentials) => {
                    info!(
                        default_channel_configured = default_channel.is_some(),
                        "Slack integration enabled"
                    );
                    Some(Arc::new(SlackService::new(
                        credentials.bot_token,
                        credentials.app_token,
                        default_channel,
                    )))
                }
                SlackCredentialResolution::Missing { env_vars } => {
                    info!(
                        missing_env_vars = %env_vars.join(","),
                        "Slack integration disabled; missing credentials"
                    );
                    None
                }
            }
        } else {
            info!("Slack integration disabled by server configuration");
            None
        }
    };
    let worker_tokens = worker_token_keys_from_server_secrets(&server_secrets)?;
    let github_api_base_url = github_api_base_url.unwrap_or_else(fabro_github::github_api_base_url);
    let storage_root = PathBuf::from(&current_server_settings.server.storage.root);
    let automation_repo_cache = Arc::new(GitRepoCache::new(
        Storage::new(&storage_root)
            .cache_dir()
            .join("automation-repos"),
    ));
    let worker_control_bus: Arc<dyn WorkerControlBus> = {
        #[cfg(test)]
        {
            worker_control_bus.unwrap_or_else(|| Arc::new(LocalWorkerControlBus::new()))
        }
        #[cfg(not(test))]
        {
            Arc::new(LocalWorkerControlBus::new())
        }
    };
    let worker_runtime: Arc<dyn WorkerRuntime> = {
        #[cfg(test)]
        {
            worker_runtime.unwrap_or_else(|| Arc::new(LocalWorkerRuntime::new()))
        }
        #[cfg(not(test))]
        {
            Arc::new(LocalWorkerRuntime::new())
        }
    };
    Ok(Arc::new(AppState {
        runs: Mutex::new(HashMap::new()),
        aggregate_billing: Mutex::new(BillingAccumulator::default()),
        stores: AppStores {
            runs: store,
            run_summaries,
            session_records,
            auth_codes,
            auth_sessions,
            automations: automation_store,
            environments: environment_store,
            mcp_servers: mcp_server_store,
            vault: secret_store,
            variables,
        },
        session_runtimes: SessionRuntimeManager::new(),
        artifact_store,
        automation_repo_cache,
        #[cfg(any(test, feature = "test-support"))]
        automation_materializer_override,
        worker_tokens,
        started_at: Instant::now(),
        resource_sampler: resource_sampler::ResourceSampler::new(),
        max_concurrent_runs,
        worker_control_bus,
        worker_runtime,
        scheduler_notify: Notify::new(),
        automation_scheduler_notify: Notify::new(),
        pull_request_scheduler_notify: Notify::new(),
        pull_request_creation_queue: Mutex::new(
            pull_request_supervisor::PendingPullRequestCreationQueue::default(),
        ),
        global_event_tx,
        files_in_flight: new_files_in_flight(),
        pull_request_create_locks: KeyedMutex::new(),
        parent_link_lock: AsyncMutex::new(()),
        server_secrets,
        llm_source,
        manifest_run_defaults: RwLock::new(current_manifest_run_defaults),
        manifest_run_settings: RwLock::new(current_manifest_run_settings),
        server_settings: RwLock::new(current_server_settings),
        effective_web_url: RwLock::new(current_effective_web_url),
        catalog: RwLock::new(current_catalog),
        env_lookup: Arc::clone(&env_lookup),
        github_api_base_url,
        active_config_path,
        http_client,
        sandbox_inventory,
        shutdown,
        shutting_down: AtomicBool::new(false),
        registry_factory_override,
        slack_service,
        slack_started: AtomicBool::new(false),
        // Startup snapshot for the sync router build; rotating the webhook
        // secret requires a server restart.
        github_webhook_secret: vault.get(WEBHOOK_SECRET_ENV).map(str::to_string),
    }))
}

const MAX_PAGE_OFFSET: u32 = 1_000_000;

enum DeleteRunOutcome {
    Deleted,
    AlreadyAbsent,
    Preserved(DeleteRunResponse),
}

enum SandboxDeleteOutcome {
    /// The durable run store did not exist; nothing to delete.
    Absent,
    /// The sandbox resource was cleaned up (or there was none to clean).
    Cleaned,
    /// Sandbox is being handed off to the operator instead of deleted.
    Preserved(DeleteRunResponse),
}

async fn delete_run_internal(
    state: &AppState,
    id: RunId,
    force: bool,
) -> Result<DeleteRunOutcome, ApiError> {
    if !force {
        reject_active_delete_without_force(state, &id).await?;
    }

    let mut managed_run = if let Ok(mut runs) = state.runs.lock() {
        runs.remove(&id)
    } else {
        None
    };
    let had_managed_run = managed_run.is_some();
    let durable_status = if managed_run.is_some() {
        durable_run_status(state, id).await.ok().flatten()
    } else {
        None
    };
    let should_signal_cancel = !durable_status.is_some_and(RunStatus::is_terminal);

    if let Some(managed_run) = managed_run.as_mut() {
        if should_signal_cancel {
            if let Some(token) = &managed_run.cancel_token {
                token.cancel();
            }
            if let Some(answer_transport) = managed_run.answer_transport.clone() {
                let _ = answer_transport.cancel_run().await;
            }
            if let Some(cancel_tx) = managed_run.cancel_tx.take() {
                let _ = cancel_tx.send(());
            }
        }
        // Terminal runs can still carry a stale worker ref briefly after their
        // completion events land, so avoid paying the full cancellation grace.
        let delete_grace = if should_signal_cancel && managed_run.status.requires_force_to_delete()
        {
            WORKER_CANCEL_GRACE
        } else {
            TERMINAL_DELETE_WORKER_GRACE
        };
        terminate_worker_for_deletion(
            &state.worker_runtime,
            managed_run.worker_ref.clone(),
            delete_grace,
        )
        .await;
    }

    let delete_outcome = delete_run_sandbox_resource(state, id, force).await?;

    if let Some(mut managed_run) = managed_run {
        if let Some(run_dir) = managed_run.run_dir.take() {
            remove_run_dir(&run_dir)
                .map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
        }
    } else {
        let storage = Storage::new(state.server_storage_dir());
        let run_dir = storage.run_scratch(&id).root().to_path_buf();
        remove_run_dir(&run_dir)
            .map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
    }

    state
        .stores
        .runs
        .delete_run(&id)
        .await
        .map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
    state
        .artifact_store
        .delete_for_run(&id)
        .await
        .map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
    match delete_outcome {
        SandboxDeleteOutcome::Preserved(response) => Ok(DeleteRunOutcome::Preserved(response)),
        SandboxDeleteOutcome::Cleaned => Ok(DeleteRunOutcome::Deleted),
        SandboxDeleteOutcome::Absent if had_managed_run => Ok(DeleteRunOutcome::Deleted),
        SandboxDeleteOutcome::Absent => Ok(DeleteRunOutcome::AlreadyAbsent),
    }
}

async fn delete_run_sandbox_resource(
    state: &AppState,
    id: RunId,
    force: bool,
) -> Result<SandboxDeleteOutcome, ApiError> {
    let Ok(run_store) = state.stores.runs.open_run(&id).await else {
        return Ok(SandboxDeleteOutcome::Absent);
    };
    let projection = match run_store.state().await {
        Ok(projection) => projection,
        Err(err) if force => {
            tracing::warn!(
                run_id = %id,
                error = %render_with_causes(&err.to_string(), &collect_causes(&err)),
                "Skipping sandbox provider delete because run projection cannot be loaded"
            );
            return Ok(SandboxDeleteOutcome::Cleaned);
        }
        Err(err) => {
            return Err(ApiError::new(
                StatusCode::INTERNAL_SERVER_ERROR,
                err.to_string(),
            ));
        }
    };
    let delete_started = matches!(projection.status, RunStatus::Removing);
    let can_mark_removing = projection.status.can_transition_to(RunStatus::Removing);
    if !delete_started && can_mark_removing {
        workflow_event::append_event(&run_store, &id, &workflow_event::Event::RunRemoving)
            .await
            .map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
    }

    let preserve = projection
        .spec()
        .settings
        .run
        .environment
        .lifecycle
        .preserve;
    let Some(record) = projection
        .sandbox
        .as_ref()
        .and_then(fabro_types::RunSandbox::instance)
        .cloned()
    else {
        return Ok(SandboxDeleteOutcome::Cleaned);
    };
    let runtime = &record.runtime;
    if preserve {
        return Ok(SandboxDeleteOutcome::Preserved(DeleteRunResponse {
            deleted:           true,
            sandbox_preserved: true,
            sandbox:           DeleteRunSandbox {
                provider: record.provider,
                id:       runtime.id.clone(),
            },
        }));
    }

    let access = state
        .provider_access()
        .await
        .map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
    let sandbox = match reconnect_for_run(&record, &access, Some(id), None).await {
        Ok(sandbox) => sandbox,
        Err(err) if force || delete_started => {
            tracing::warn!(
                run_id = %id,
                error = %render_with_causes(&err.to_string(), &collect_causes(err.as_ref())),
                "Skipping sandbox provider delete during run deletion"
            );
            return Ok(SandboxDeleteOutcome::Cleaned);
        }
        Err(err) => {
            let detail = render_with_causes(&err.to_string(), &collect_causes(err.as_ref()));
            return Err(ApiError::new(StatusCode::CONFLICT, detail));
        }
    };
    if let Err(err) = sandbox.delete().await {
        if force || delete_started {
            tracing::warn!(
                run_id = %id,
                error = %err.display_with_causes(),
                "Skipping failed sandbox provider delete during run deletion"
            );
            return Ok(SandboxDeleteOutcome::Cleaned);
        }
        return Err(ApiError::new(
            StatusCode::CONFLICT,
            err.display_with_causes(),
        ));
    }

    Ok(SandboxDeleteOutcome::Cleaned)
}

async fn reject_active_delete_without_force(
    state: &AppState,
    run_id: &RunId,
) -> Result<(), ApiError> {
    let managed_status = state
        .runs
        .lock()
        .ok()
        .and_then(|runs| runs.get(run_id).map(|managed_run| managed_run.status));
    if let Some(status) = managed_status {
        if status.requires_force_to_delete() {
            return Err(ApiError::new(
                StatusCode::CONFLICT,
                active_run_delete_message(*run_id, status),
            ));
        }
        return Ok(());
    }

    match state.stores.run_summaries.get(run_id, Utc::now()).await {
        Ok(Some(summary)) if summary.lifecycle.status.requires_force_to_delete() => {
            Err(ApiError::new(
                StatusCode::CONFLICT,
                active_run_delete_message(*run_id, summary.lifecycle.status),
            ))
        }
        Ok(_) => Ok(()),
        Err(err) => Err(ApiError::new(
            StatusCode::INTERNAL_SERVER_ERROR,
            err.to_string(),
        )),
    }
}

fn active_run_delete_message(run_id: RunId, status: impl std::fmt::Display) -> String {
    let run_id = run_id.to_string();
    let short_run_id = &run_id[..12.min(run_id.len())];
    format!(
        "cannot remove active run {short_run_id} (status: {status}, use force=true or --force to force)"
    )
}

async fn terminate_worker_for_deletion(
    worker_runtime: &Arc<dyn WorkerRuntime>,
    worker_ref: Option<WorkerRef>,
    grace: Duration,
) {
    let Some(worker_ref) = worker_ref else {
        return;
    };

    worker_runtime.request_stop(&worker_ref).await;

    let deadline = Instant::now() + grace;
    while Instant::now() < deadline && worker_runtime.is_alive(&worker_ref).await {
        sleep(Duration::from_millis(50)).await;
    }

    if worker_runtime.is_alive(&worker_ref).await {
        worker_runtime.force_stop(&worker_ref).await;
        let kill_deadline = Instant::now() + Duration::from_secs(1);
        while Instant::now() < kill_deadline && worker_runtime.is_alive(&worker_ref).await {
            sleep(Duration::from_millis(50)).await;
        }
    }
}

fn remove_run_dir(run_dir: &std::path::Path) -> std::io::Result<()> {
    match std::fs::remove_dir_all(run_dir) {
        Ok(()) => Ok(()),
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(err) => Err(err),
    }
}

#[cfg(test)]
fn compute_queue_positions(runs: &HashMap<RunId, ManagedRun>) -> HashMap<RunId, i64> {
    let mut runnable: Vec<(&RunId, &ManagedRun)> = runs
        .iter()
        .filter(|(_, r)| r.status == RunStatus::Runnable)
        .collect();
    runnable.sort_by_key(|(_, r)| r.created_at);
    runnable
        .into_iter()
        .enumerate()
        .map(|(i, (id, _))| (*id, i64::try_from(i + 1).unwrap()))
        .collect()
}

pub(in crate::server) fn counts_toward_scheduler_capacity(status: RunStatus) -> bool {
    matches!(
        status,
        RunStatus::Starting
            | RunStatus::Running
            | RunStatus::Blocked { .. }
            | RunStatus::Paused { .. }
    )
}

#[allow(
    clippy::result_large_err,
    reason = "Run ID parsing returns HTTP 400 responses directly."
)]
pub(crate) fn parse_run_id_path(id: &str) -> Result<RunId, Response> {
    id.parse::<RunId>()
        .map_err(|_| ApiError::bad_request("Invalid run ID.").into_response())
}

#[allow(
    clippy::result_large_err,
    reason = "Stage ID parsing returns HTTP 400 responses directly."
)]
pub(crate) fn parse_stage_id_path(stage_id: &str) -> Result<StageId, Response> {
    StageId::from_str(stage_id)
        .map_err(|_| ApiError::bad_request("Invalid stage ID.").into_response())
}

#[allow(
    clippy::result_large_err,
    reason = "Blob hash parsing returns HTTP 400 responses directly."
)]
pub(crate) fn parse_blob_hash_path(blob_hash: &str) -> Result<BlobHash, Response> {
    BlobHash::from_str(blob_hash)
        .map_err(|_| ApiError::bad_request("Invalid blob hash.").into_response())
}

#[allow(
    clippy::result_large_err,
    reason = "Missing query parameter validation returns HTTP 400 responses directly."
)]
fn required_query_param<T: Clone>(value: Option<&T>, name: &str) -> Result<T, Response> {
    value.cloned().ok_or_else(|| {
        ApiError::bad_request(format!("Missing {name} query parameter.")).into_response()
    })
}

#[allow(
    clippy::result_large_err,
    reason = "Artifact path validation returns HTTP 400 responses directly."
)]
fn validate_relative_artifact_path(kind: &str, value: &str) -> Result<String, Response> {
    if value.is_empty() {
        return Err(ApiError::bad_request(format!("{kind} must not be empty")).into_response());
    }

    if value.contains('\\') {
        return Err(
            ApiError::bad_request(format!("{kind} must not contain backslashes")).into_response(),
        );
    }

    let segments = value.split('/').collect::<Vec<_>>();
    if segments.iter().any(|segment| segment.is_empty()) {
        return Err(
            ApiError::bad_request(format!("{kind} must not contain empty path segments"))
                .into_response(),
        );
    }
    if segments
        .iter()
        .any(|segment| matches!(*segment, "." | ".."))
    {
        return Err(ApiError::bad_request(format!(
            "{kind} must be a relative path without '.' or '..' segments"
        ))
        .into_response());
    }

    Ok(segments.join("/"))
}

fn bad_request_response(detail: impl Into<String>) -> Response {
    ApiError::bad_request(detail.into()).into_response()
}

fn payload_too_large_response(detail: impl Into<String>) -> Response {
    ApiError::new(StatusCode::PAYLOAD_TOO_LARGE, detail.into()).into_response()
}

fn octet_stream_response(bytes: Bytes) -> Response {
    (
        StatusCode::OK,
        [("content-type", "application/octet-stream")],
        bytes,
    )
        .into_response()
}

fn clear_live_run_state(run: &mut ManagedRun) {
    run.answer_transport = None;
    run.accepted_questions.clear();
    run.active_api_targets.clear();
    run.active_steerable_stages.clear();
    run.active_non_steerable_stages.clear();
    run.event_tx = None;
    run.cancel_tx = None;
    run.cancel_token = None;
    run.worker_ref = None;
    run.cancel_escalation_worker = None;
}

fn cleanup_worker_control_bus_for_run(state: &AppState, run_id: RunId) {
    let bus = Arc::clone(&state.worker_control_bus);
    tokio::spawn(async move {
        bus.cleanup_run(run_id).await;
    });
}

fn reconcile_live_interview_state_for_event(run: &mut ManagedRun, event: &RunEvent) {
    match &event.body {
        EventBody::InterviewCompleted(props) => {
            run.accepted_questions.remove(&props.question_id);
        }
        EventBody::InterviewTimeout(props) => {
            run.accepted_questions.remove(&props.question_id);
        }
        EventBody::InterviewInterrupted(props) => {
            run.accepted_questions.remove(&props.question_id);
        }
        EventBody::RunCompleted(_) | EventBody::RunFailed(_) => {
            run.accepted_questions.clear();
        }
        _ => {}
    }
}

fn claim_run_answer_transport(
    state: &AppState,
    run_id: RunId,
    qid: &str,
) -> Result<RunAnswerTransport, StatusCode> {
    let mut runs = state.runs.lock().expect("runs lock poisoned");
    let managed_run = runs.get_mut(&run_id).ok_or(StatusCode::NOT_FOUND)?;
    let transport = managed_run
        .answer_transport
        .clone()
        .ok_or(StatusCode::CONFLICT)?;

    if !managed_run.accepted_questions.insert(qid.to_string()) {
        return Err(StatusCode::CONFLICT);
    }

    Ok(transport)
}

fn release_run_answer_claim(state: &AppState, run_id: RunId, qid: &str) {
    let mut runs = state.runs.lock().expect("runs lock poisoned");
    if let Some(managed_run) = runs.get_mut(&run_id) {
        managed_run.accepted_questions.remove(qid);
    }
}

#[derive(Clone)]
struct LiveWorkerProcess {
    run_id:     RunId,
    worker_ref: WorkerRef,
}

/// Pick the terminal failure for a run that never produced its own terminal
/// event. A pending cancel wins over whatever failure the caller observed, so a
/// run that was cancelled while its worker was launching or dying is recorded
/// as cancelled rather than as broken.
fn failure_honoring_pending_cancel(
    pending_control: Option<RunControlAction>,
    otherwise: impl FnOnce() -> (WorkflowError, FailureReason),
) -> (WorkflowError, FailureReason) {
    if pending_control == Some(RunControlAction::Cancel) {
        (WorkflowError::Cancelled, FailureReason::Cancelled)
    } else {
        otherwise()
    }
}

fn failure_for_incomplete_run(
    pending_control: Option<RunControlAction>,
    terminated_message: String,
) -> (WorkflowError, FailureReason) {
    failure_honoring_pending_cancel(pending_control, || {
        (
            WorkflowError::engine(terminated_message),
            FailureReason::Terminated,
        )
    })
}

pub(crate) async fn reconcile_incomplete_runs_on_startup(
    state: &Arc<AppState>,
) -> anyhow::Result<usize> {
    const RECONCILABLE_STATUSES: &[RunStatusKind] = &[
        RunStatusKind::Runnable,
        RunStatusKind::Starting,
        RunStatusKind::Running,
        RunStatusKind::Blocked,
        RunStatusKind::Paused,
        RunStatusKind::Removing,
    ];
    let summaries = state
        .stores
        .run_summaries
        .list_by_statuses(RECONCILABLE_STATUSES, chrono::Utc::now())
        .await?;
    let mut reconciled = 0usize;

    for summary in summaries {
        let run_store = state.stores.runs.open_run(&summary.id).await?;
        let (error, reason) = failure_for_incomplete_run(
            summary.lifecycle.pending_control,
            "Fabro server restarted before the run reached a terminal state.".to_string(),
        );
        let failure_event = workflow_event::Event::workflow_run_failed_from_error(
            &error,
            fabro_types::RunTiming::default(),
            reason,
            None,
            None,
            None,
            None,
        );
        workflow_event::append_event(&run_store, &summary.id, &failure_event).await?;
        reconciled += 1;
    }

    Ok(reconciled)
}

fn live_worker_processes(state: &AppState) -> Vec<LiveWorkerProcess> {
    let runs = state.runs.lock().expect("runs lock poisoned");
    runs.iter()
        .filter_map(|(run_id, managed_run)| {
            managed_run
                .worker_ref
                .clone()
                .map(|worker_ref| LiveWorkerProcess {
                    run_id: *run_id,
                    worker_ref,
                })
        })
        .collect()
}

async fn persist_shutdown_run_failures(
    state: &Arc<AppState>,
    workers: &[LiveWorkerProcess],
) -> anyhow::Result<()> {
    let run_ids = workers
        .iter()
        .map(|worker| worker.run_id)
        .collect::<HashSet<_>>();

    for run_id in run_ids {
        let run_store = state.stores.runs.open_run(&run_id).await?;
        let run_state = run_store.state().await?;
        if run_state.status.is_terminal() {
            continue;
        }

        let (error, reason) = failure_for_incomplete_run(
            run_state.pending_control,
            "Fabro server shut down before the run reached a terminal state.".to_string(),
        );
        let failure_event = workflow_event::Event::workflow_run_failed_from_error(
            &error,
            fabro_types::RunTiming::default(),
            reason,
            None,
            None,
            None,
            None,
        );
        workflow_event::append_event(&run_store, &run_id, &failure_event).await?;
    }

    Ok(())
}

pub(crate) async fn shutdown_active_workers(state: &Arc<AppState>) -> anyhow::Result<usize> {
    shutdown_active_workers_with_grace(state, WORKER_CANCEL_GRACE, Duration::from_millis(50)).await
}

async fn shutdown_active_workers_with_grace(
    state: &Arc<AppState>,
    grace: Duration,
    poll_interval: Duration,
) -> anyhow::Result<usize> {
    state.begin_shutdown();
    let workers = live_worker_processes(state.as_ref());

    join_all(
        workers
            .iter()
            .map(|worker| state.worker_runtime.request_stop(&worker.worker_ref)),
    )
    .await;

    let survivors = poll_until_dead(state.as_ref(), &workers, grace, poll_interval).await;

    if !survivors.is_empty() {
        join_all(
            survivors
                .iter()
                .map(|worker_ref| state.worker_runtime.force_stop(worker_ref)),
        )
        .await;
        // Wait for the kernel to reap the killed workers so callers can
        // assume the processes are actually gone when shutdown returns.
        let kill_deadline = Instant::now() + Duration::from_secs(1);
        while Instant::now() < kill_deadline
            && !alive_refs(state.as_ref(), &survivors).await.is_empty()
        {
            sleep(poll_interval).await;
        }
    }

    persist_shutdown_run_failures(state, &workers).await?;
    Ok(workers.len())
}

/// Poll until either the deadline expires or every worker is dead, returning
/// the set of workers still alive when polling stopped.
async fn poll_until_dead(
    state: &AppState,
    workers: &[LiveWorkerProcess],
    grace: Duration,
    poll_interval: Duration,
) -> Vec<WorkerRef> {
    let refs: Vec<WorkerRef> = workers.iter().map(|w| w.worker_ref.clone()).collect();
    let deadline = Instant::now() + grace;
    loop {
        let alive = alive_refs(state, &refs).await;
        if alive.is_empty() || Instant::now() >= deadline {
            return alive;
        }
        sleep(poll_interval).await;
    }
}

async fn alive_refs(state: &AppState, refs: &[WorkerRef]) -> Vec<WorkerRef> {
    let liveness = join_all(refs.iter().map(|r| state.worker_runtime.is_alive(r))).await;
    refs.iter()
        .zip(liveness)
        .filter(|(_, alive)| *alive)
        .map(|(r, _)| r.clone())
        .collect()
}

async fn persist_cancelled_run_status(state: &AppState, run_id: RunId) -> anyhow::Result<()> {
    let run_store = state.stores.runs.open_run(&run_id).await?;
    let run_state = run_store.state().await?;
    if run_state.status.is_terminal() {
        return Ok(());
    }

    let failure_event = workflow_event::Event::workflow_run_failed_from_error(
        &WorkflowError::Cancelled,
        fabro_types::RunTiming::default(),
        FailureReason::Cancelled,
        None,
        None,
        None,
        None,
    );
    workflow_event::append_event(&run_store, &run_id, &failure_event).await
}

async fn finish_cancelled_run_before_execution(state: &Arc<AppState>, run_id: RunId) {
    if let Err(err) = persist_cancelled_run_status(state.as_ref(), run_id).await {
        error!(run_id = %run_id, error = %err, "Failed to persist cancelled run status");
    }

    let mut runs = state.runs.lock().expect("runs lock poisoned");
    if let Some(managed_run) = runs.get_mut(&run_id) {
        managed_run.status = RunStatus::Failed {
            reason: FailureReason::Cancelled,
        };
        clear_live_run_state(managed_run);
    }
    drop(runs);
    cleanup_worker_control_bus_for_run(state.as_ref(), run_id);
    state.scheduler_notify.notify_one();
}

/// Reject the run before execution if its effective sandbox provider is
/// disabled by server policy. Returns `true` when the run was rejected.
async fn reject_run_if_sandbox_provider_disabled(
    state: &Arc<AppState>,
    server_settings: &ServerSettings,
    run_id: RunId,
    settings: &RunNamespace,
) -> bool {
    let provider = run_manifest::effective_sandbox_provider(settings);
    let Some(error) = run_manifest::sandbox_provider_policy_error(server_settings, &provider)
    else {
        return false;
    };
    tracing::warn!(run_id = %run_id, error = %error, "Sandbox provider disabled by server policy");
    fail_run_before_execution(state, run_id, FailureReason::LaunchFailed, error).await;
    true
}

async fn fail_run_before_execution(
    state: &Arc<AppState>,
    run_id: RunId,
    reason: FailureReason,
    message: String,
) {
    match state.stores.runs.open_run(&run_id).await {
        Ok(run_store) => {
            let failure_event = workflow_event::Event::workflow_run_failed_from_error(
                &WorkflowError::engine(message.clone()),
                fabro_types::RunTiming::default(),
                reason,
                None,
                None,
                None,
                None,
            );
            if let Err(err) =
                workflow_event::append_event(&run_store, &run_id, &failure_event).await
            {
                error!(run_id = %run_id, error = %err, "Failed to persist run failure status");
            }
        }
        Err(err) => {
            error!(run_id = %run_id, error = %err, "Failed to open run store while persisting run failure");
        }
    }

    fail_managed_run(state, run_id, reason, message);
    state.scheduler_notify.notify_one();
}

async fn forward_run_events_to_global(
    state: Arc<AppState>,
    run_id: RunId,
    mut run_events: broadcast::Receiver<EventEnvelope>,
) {
    loop {
        match run_events.recv().await {
            Ok(event) => {
                let mut runs = state.runs.lock().expect("runs lock poisoned");
                if let Some(managed_run) = runs.get_mut(&run_id) {
                    reconcile_live_interview_state_for_event(managed_run, &event.event);
                }
                let _ = state.global_event_tx.send(event);
            }
            Err(RecvError::Lagged(_)) => {}
            Err(RecvError::Closed) => break,
        }
    }
}

fn managed_run(
    dot_source: String,
    status: RunStatus,
    created_at: chrono::DateTime<chrono::Utc>,
    run_dir: std::path::PathBuf,
    execution_mode: RunExecutionMode,
) -> ManagedRun {
    ManagedRun {
        dot_source,
        status,
        error: None,
        created_at,
        answer_transport: None,
        accepted_questions: HashSet::new(),
        active_api_targets: HashMap::new(),
        active_steerable_stages: HashMap::new(),
        active_non_steerable_stages: HashMap::new(),
        event_tx: None,
        checkpoint: None,
        cancel_tx: None,
        cancel_token: None,
        worker_ref: None,
        cancel_escalation_worker: None,
        run_dir: Some(run_dir),
        execution_mode,
    }
}

fn worker_mode_arg(mode: RunExecutionMode) -> &'static str {
    match mode {
        RunExecutionMode::Start => "start",
        RunExecutionMode::Resume => "resume",
    }
}

async fn load_pending_control(
    state: &AppState,
    run_id: RunId,
) -> anyhow::Result<Option<RunControlAction>> {
    Ok(state
        .stores
        .run_summaries
        .get(&run_id, Utc::now())
        .await?
        .and_then(|summary| summary.lifecycle.pending_control))
}

async fn durable_run_status(state: &AppState, run_id: RunId) -> anyhow::Result<Option<RunStatus>> {
    Ok(state
        .stores
        .run_summaries
        .get(&run_id, Utc::now())
        .await?
        .map(|summary| summary.lifecycle.status))
}

fn fail_managed_run(state: &Arc<AppState>, run_id: RunId, reason: FailureReason, message: String) {
    let mut runs = state.runs.lock().expect("runs lock poisoned");
    if let Some(managed_run) = runs.get_mut(&run_id) {
        managed_run.status = RunStatus::Failed { reason };
        managed_run.error = Some(message);
        clear_live_run_state(managed_run);
    }
    cleanup_worker_control_bus_for_run(state.as_ref(), run_id);
}

fn update_live_run_from_event(state: &AppState, run_id: RunId, event: &RunEvent) {
    use fabro_types::EventBody;

    let mut runs = state.runs.lock().expect("runs lock poisoned");
    let Some(managed_run) = runs.get_mut(&run_id) else {
        return;
    };

    if matches!(&event.body, EventBody::RunRunnable(_)) {
        // Scheduling is owned by the start/approve lifecycle handlers, which
        // set the live status and notify the scheduler explicitly. Direct
        // event ingestion still records durable history, but must not make
        // externally injected events schedulable.
        return;
    }

    match &event.body {
        EventBody::RunSubmitted(_) => managed_run.status = RunStatus::Submitted,
        EventBody::RunPending(props) => {
            managed_run.status = RunStatus::Pending {
                reason: props.reason,
            };
        }
        EventBody::RunStarting(_) => managed_run.status = RunStatus::Starting,
        EventBody::RunRunning(_) => managed_run.status = RunStatus::Running,
        EventBody::RunBlocked(props) => {
            managed_run.status = match managed_run.status {
                RunStatus::Paused { .. } => RunStatus::Paused {
                    prior_block: Some(props.blocked_reason),
                },
                _ => RunStatus::Blocked {
                    blocked_reason: props.blocked_reason,
                },
            };
        }
        EventBody::RunUnblocked(_) => {
            managed_run.status = match managed_run.status {
                RunStatus::Paused {
                    prior_block: Some(_) | None,
                } => RunStatus::Paused { prior_block: None },
                _ => RunStatus::Running,
            };
        }
        EventBody::RunPaused(_) => {
            let prior_block = match managed_run.status {
                RunStatus::Blocked { blocked_reason } => Some(blocked_reason),
                RunStatus::Paused { prior_block } => prior_block,
                _ => None,
            };
            managed_run.status = RunStatus::Paused { prior_block };
        }
        EventBody::RunUnpaused(_) => {
            managed_run.status = match managed_run.status {
                RunStatus::Paused {
                    prior_block: Some(blocked_reason),
                } => RunStatus::Blocked { blocked_reason },
                _ => RunStatus::Running,
            };
        }
        EventBody::RunRemoving(_) => managed_run.status = RunStatus::Removing,
        EventBody::RunCompleted(_) => {
            let EventBody::RunCompleted(props) = &event.body else {
                unreachable!(
                    "outer match arm already verified event.body is EventBody::RunCompleted"
                )
            };
            managed_run.status = RunStatus::Succeeded {
                reason: props.reason,
            };
            managed_run.error = None;
            managed_run.active_api_targets.clear();
            managed_run.active_steerable_stages.clear();
            managed_run.active_non_steerable_stages.clear();
            cleanup_worker_control_bus_for_run(state, run_id);
        }
        EventBody::RunFailed(props) => {
            managed_run.status = RunStatus::Failed {
                reason: props.failure.reason,
            };
            managed_run.error = Some(render_compact_with_causes(
                &props.failure.detail.message,
                &props.failure.detail.causes,
            ));
            managed_run.active_api_targets.clear();
            managed_run.active_steerable_stages.clear();
            managed_run.active_non_steerable_stages.clear();
            cleanup_worker_control_bus_for_run(state, run_id);
        }
        // Track active agent sessions by steerability. Activated/deactivated
        // are leased by session id so stale deactivations cannot clear a newer
        // binding for the same stage.
        EventBody::AgentSessionActivated(props) => {
            if let (Some(stage_id), Some(session_id)) =
                (event.stage_id.as_ref(), event.session_id.as_ref())
            {
                if props.capabilities.contains(&SessionCapability::Steer) {
                    managed_run
                        .active_steerable_stages
                        .insert(stage_id.clone(), session_id.clone());
                    managed_run.active_non_steerable_stages.remove(stage_id);
                    let acp_provider: &'static str = AgentBackend::Acp.into();
                    if props.provider.as_deref() == Some(acp_provider) {
                        managed_run.active_api_targets.remove(stage_id);
                    } else {
                        managed_run
                            .active_api_targets
                            .insert(stage_id.clone(), PairTarget {
                                stage_id:   stage_id.clone(),
                                node_label: event
                                    .node_label
                                    .clone()
                                    .unwrap_or_else(|| stage_id.node_id().to_string()),
                            });
                    }
                } else {
                    managed_run
                        .active_non_steerable_stages
                        .insert(stage_id.clone(), session_id.clone());
                    managed_run.active_steerable_stages.remove(stage_id);
                    managed_run.active_api_targets.remove(stage_id);
                }
            }
        }
        EventBody::AgentSessionDeactivated(_) => {
            if let (Some(stage_id), Some(session_id)) =
                (event.stage_id.as_ref(), event.session_id.as_ref())
            {
                if managed_run
                    .active_steerable_stages
                    .get(stage_id)
                    .is_some_and(|current| current == session_id)
                {
                    managed_run.active_steerable_stages.remove(stage_id);
                    managed_run.active_api_targets.remove(stage_id);
                }
                if managed_run
                    .active_non_steerable_stages
                    .get(stage_id)
                    .is_some_and(|current| current == session_id)
                {
                    managed_run.active_non_steerable_stages.remove(stage_id);
                }
            }
        }
        // ACP sessions are steerable via `agent.session.activated`; terminal
        // ACP events and stage lifecycle events are still backstops for cleanup.
        EventBody::AgentAcpCompleted(_)
        | EventBody::AgentAcpCancelled(_)
        | EventBody::AgentAcpTimedOut(_)
        | EventBody::StageCompleted(_)
        | EventBody::StageFailed(_) => {
            if let Some(stage_id) = &event.stage_id {
                managed_run.active_api_targets.remove(stage_id);
                managed_run.active_steerable_stages.remove(stage_id);
                managed_run.active_non_steerable_stages.remove(stage_id);
            }
        }
        _ => {}
    }
}

async fn drain_worker_stderr(
    run_id: RunId,
    stderr: std::pin::Pin<Box<dyn AsyncRead + Send + 'static>>,
) -> anyhow::Result<()> {
    let mut lines = BufReader::new(stderr).lines();

    while let Some(line) = lines.next_line().await? {
        tracing::warn!(run_id = %run_id, "Worker stderr: {line}");
    }

    Ok(())
}

async fn fail_worker_launch(
    state: &Arc<AppState>,
    run_store: &fabro_store::RunDatabase,
    run_id: RunId,
    err: anyhow::Error,
) {
    tracing::error!(run_id = %run_id, error = %err, "Failed to spawn worker");
    let pending_control = match run_store.state().await {
        Ok(run_state) => run_state.pending_control,
        Err(state_err) => {
            tracing::warn!(
                run_id = %run_id,
                error = %state_err,
                "Failed to load run state after worker launch failure"
            );
            None
        }
    };
    let launch_message = format!("Failed to spawn worker: {err}");
    let (error, reason) = failure_honoring_pending_cancel(pending_control, || {
        (
            WorkflowError::engine_with_anyhow("Failed to spawn worker", err),
            FailureReason::LaunchFailed,
        )
    });
    let message = if reason == FailureReason::Cancelled {
        "Run cancelled before worker launch completed".to_string()
    } else {
        launch_message
    };
    let failure_event = workflow_event::Event::workflow_run_failed_from_error(
        &error,
        fabro_types::RunTiming::default(),
        reason,
        None,
        None,
        None,
        None,
    );
    let _ = workflow_event::append_event(run_store, &run_id, &failure_event).await;
    fail_managed_run(state, run_id, reason, message);
    state.scheduler_notify.notify_one();
}

async fn append_worker_exit_failure(
    run_store: &fabro_store::RunDatabase,
    run_id: RunId,
    worker_exit: &WorkerExit,
) {
    let state = match run_store.state().await {
        Ok(state) => state,
        Err(err) => {
            tracing::warn!(run_id = %run_id, error = %err, "Failed to load run state after worker exit");
            return;
        }
    };

    let terminal = state.status.is_terminal();
    if terminal {
        return;
    }

    let (error, reason) = failure_for_incomplete_run(
        state.pending_control,
        format!(
            "Worker exited before emitting a terminal run event: {}",
            worker_exit.detail
        ),
    );
    let failure_event = workflow_event::Event::workflow_run_failed_from_error(
        &error,
        fabro_types::RunTiming::default(),
        reason,
        None,
        None,
        None,
        None,
    );

    if let Err(err) = workflow_event::append_event(run_store, &run_id, &failure_event).await {
        tracing::warn!(run_id = %run_id, error = %err, "Failed to append worker exit failure");
    }
}

#[expect(
    clippy::disallowed_methods,
    reason = "Worker subprocess startup resolves Cargo's test binary env override when present."
)]
fn worker_launch_spec(
    state: &AppState,
    run_id: RunId,
    mode: RunExecutionMode,
    run_dir: &std::path::Path,
    agent_fabro_tools_enabled: bool,
    github_app_private_key: Option<String>,
) -> anyhow::Result<WorkerLaunchSpec> {
    let current_exe = std::env::current_exe().context("reading current executable path")?;
    let executable =
        std::env::var_os(EnvVars::CARGO_BIN_EXE_FABRO).map_or(current_exe, PathBuf::from);
    let storage_dir = state.server_storage_dir();
    let runtime_directory = Storage::new(&storage_dir).runtime_directory();
    let daemon = ServerDaemon::read(&runtime_directory)?.with_context(|| {
        format!(
            "server record {} is missing",
            runtime_directory.record_path().display()
        )
    })?;
    let scopes = if agent_fabro_tools_enabled {
        WorkerScopeSet::run_worker_with_agent_run_tools()
    } else {
        WorkerScopeSet::run_worker()
    };
    let worker_token = issue_worker_token_with_scopes(state.worker_token_keys(), &run_id, scopes)
        .map_err(|_| anyhow::anyhow!("failed to sign worker token"))?;
    let log_destination = resolved_log_destination(state)?;
    let fabro_log = if (state.env_lookup)(EnvVars::FABRO_LOG).is_none() {
        state.server_settings().server.logging.level.clone()
    } else {
        None
    };

    Ok(WorkerLaunchSpec {
        executable,
        server_target: daemon.bind.to_target(),
        storage_dir,
        run_dir: run_dir.to_path_buf(),
        run_id,
        mode: worker_mode_arg(mode),
        worker_token,
        log_destination,
        fabro_log,
        active_config_path: state.active_config_path().to_path_buf(),
        github_app_private_key,
    })
}

fn resolved_log_destination(state: &AppState) -> anyhow::Result<LogDestination> {
    let env_value = (state.env_lookup)(EnvVars::FABRO_LOG_DESTINATION);
    fabro_config::resolve_log_destination_with_env(
        state.server_settings().server.logging.destination,
        env_value.as_deref(),
    )
}

fn runtime_question_from_interview_record(question: &InterviewQuestionRecord) -> Question {
    Question {
        id:              question.id.clone(),
        text:            question.text.clone(),
        question_type:   question.question_type,
        options:         question.options.clone(),
        allow_freeform:  question.allow_freeform,
        default:         None,
        timeout_seconds: question.timeout_seconds,
        stage:           question.stage.clone(),
        metadata:        HashMap::new(),
        context_display: question.context_display.clone(),
        review_target:   question.review_target.clone(),
    }
}

fn api_question_from_interview_record(question: &InterviewQuestionRecord) -> ApiQuestion {
    ApiQuestion {
        id:              question.id.clone(),
        text:            question.text.clone(),
        stage:           question.stage.clone(),
        question_type:   question.question_type,
        options:         question.options.clone(),
        allow_freeform:  question.allow_freeform,
        timeout_seconds: question.timeout_seconds,
        context_display: question.context_display.clone(),
        review_target:   question.review_target.clone(),
    }
}

fn api_question_from_pending_interview(record: &PendingInterviewRecord) -> ApiQuestion {
    api_question_from_interview_record(&record.question)
}

#[allow(
    clippy::result_large_err,
    reason = "Pending-interview lookup maps storage failures to HTTP responses."
)]
async fn load_pending_interview(
    state: &AppState,
    run_id: RunId,
    qid: &str,
) -> Result<LoadedPendingInterview, Response> {
    let projection = state
        .load_run_projection(&run_id)
        .await
        .map_err(IntoResponse::into_response)?;
    let Some(record) = projection.pending_interviews.get(qid) else {
        return Err(ApiError::new(
            StatusCode::CONFLICT,
            "Question no longer exists or was already answered.",
        )
        .into_response());
    };

    Ok(LoadedPendingInterview {
        run_id,
        qid: qid.to_string(),
        question: record.question.clone(),
    })
}

#[allow(
    clippy::result_large_err,
    reason = "Interview answer validation returns HTTP 400 responses directly."
)]
fn validate_answer_for_question(
    question: &InterviewQuestionRecord,
    answer: &Answer,
) -> Result<(), Response> {
    match (&question.question_type, &answer.value) {
        (
            QuestionType::YesNo | QuestionType::Confirmation,
            fabro_interview::AnswerValue::Yes | fabro_interview::AnswerValue::No,
        )
        | (
            _,
            fabro_interview::AnswerValue::Interrupted
            | fabro_interview::AnswerValue::Skipped
            | fabro_interview::AnswerValue::Timeout,
        ) => Ok(()),
        (QuestionType::MultipleChoice, fabro_interview::AnswerValue::Selected(key)) => {
            if question.options.iter().any(|option| option.key == *key) {
                Ok(())
            } else {
                Err(ApiError::bad_request("Invalid option key.").into_response())
            }
        }
        (QuestionType::MultiSelect, fabro_interview::AnswerValue::MultiSelected(keys)) => {
            if keys
                .iter()
                .all(|key| question.options.iter().any(|option| option.key == *key))
            {
                Ok(())
            } else {
                Err(ApiError::bad_request("Invalid option key.").into_response())
            }
        }
        (QuestionType::Freeform, fabro_interview::AnswerValue::Text(text))
            if !text.trim().is_empty() =>
        {
            Ok(())
        }
        (_, fabro_interview::AnswerValue::Text(text))
            if question.allow_freeform && !text.trim().is_empty() =>
        {
            Ok(())
        }
        _ => Err(ApiError::bad_request("Answer does not match question type.").into_response()),
    }
}

#[allow(
    clippy::result_large_err,
    reason = "Interview submission maps validation failures to HTTP responses."
)]
async fn submit_pending_interview_answer(
    state: &AppState,
    pending: &LoadedPendingInterview,
    submission: AnswerSubmission,
) -> Result<(), Response> {
    validate_answer_for_question(&pending.question, &submission.answer)?;
    deliver_answer_to_run(state, pending.run_id, &pending.qid, submission).await
}

#[allow(
    clippy::result_large_err,
    reason = "Interview delivery maps run-state failures to HTTP responses."
)]
async fn deliver_answer_to_run(
    state: &AppState,
    run_id: RunId,
    qid: &str,
    submission: AnswerSubmission,
) -> Result<(), Response> {
    let transport = match claim_run_answer_transport(state, run_id, qid) {
        Ok(transport) => transport,
        Err(StatusCode::NOT_FOUND) => {
            return Err(ApiError::not_found("Run not found.").into_response());
        }
        Err(StatusCode::CONFLICT) => {
            return Err(ApiError::new(
                StatusCode::CONFLICT,
                "Question no longer exists or was already answered.",
            )
            .into_response());
        }
        Err(status) => {
            return Err(
                ApiError::new(status, "Run is not ready to accept answers.").into_response()
            );
        }
    };

    if let Ok(()) = transport.submit(qid, submission).await {
        Ok(())
    } else {
        release_run_answer_claim(state, run_id, qid);
        Err(ApiError::new(
            StatusCode::SERVICE_UNAVAILABLE,
            "Failed to deliver answer to the active run.",
        )
        .into_response())
    }
}

#[allow(
    clippy::result_large_err,
    reason = "Answer request parsing returns HTTP 400 responses directly."
)]
fn answer_from_request(
    req: SubmitAnswerRequest,
    question: &InterviewQuestionRecord,
) -> Result<Answer, Response> {
    match req {
        SubmitAnswerRequest::YesRequest(_) => Ok(Answer::yes()),
        SubmitAnswerRequest::NoRequest(_) => Ok(Answer::no()),
        SubmitAnswerRequest::SelectedRequest(req) => {
            let key = req.option_key;
            let option = question
                .options
                .iter()
                .find(|option| option.key == key)
                .cloned();
            match option {
                Some(option) => Ok(Answer::selected(key, option)),
                None => Err(ApiError::bad_request("Invalid option key.").into_response()),
            }
        }
        SubmitAnswerRequest::MultiSelectedRequest(req) => {
            for key in &req.option_keys {
                let valid = question.options.iter().any(|option| option.key == *key);
                if !valid {
                    return Err(ApiError::bad_request("Invalid option key.").into_response());
                }
            }
            Ok(Answer::multi_selected(req.option_keys))
        }
        SubmitAnswerRequest::TextRequest(req) => Ok(Answer::text(req.text)),
    }
}

/// Execute a single run: transitions runnable → starting → running →
/// completed/failed/cancelled.
async fn execute_run(state: Arc<AppState>, run_id: RunId) {
    if state.is_shutting_down() {
        return;
    }

    if state.registry_factory_override.is_some() {
        Box::pin(execute_run_in_process(state, run_id)).await;
        return;
    }

    Box::pin(execute_run_subprocess(state, run_id)).await;
}

async fn execute_run_in_process(state: Arc<AppState>, run_id: RunId) {
    // Transition to Starting and set up cancel infrastructure
    let (cancel_rx, run_dir, event_tx, cancel_token, execution_mode) = {
        let mut runs = state.runs.lock().expect("runs lock poisoned");
        let managed_run = match runs.get_mut(&run_id) {
            Some(r) if r.status == RunStatus::Runnable => r,
            _ => return,
        };
        let Some(run_dir) = managed_run.run_dir.clone() else {
            return;
        };

        let (cancel_tx, cancel_rx) = oneshot::channel::<()>();
        let cancel_token = CancellationToken::new();
        let (event_tx, _) = broadcast::channel(256);

        managed_run.status = RunStatus::Starting;
        managed_run.cancel_tx = Some(cancel_tx);
        managed_run.cancel_token = Some(cancel_token.clone());
        managed_run.event_tx = Some(event_tx);

        (
            cancel_rx,
            run_dir,
            managed_run.event_tx.clone(),
            cancel_token,
            managed_run.execution_mode,
        )
    };

    // Create interviewer and event plumbing (this is the "provisioning" phase)
    let interviewer = Arc::new(ControlInterviewer::new());
    let interview_runtime: Arc<dyn Interviewer> = interviewer.clone();
    let emitter = Emitter::new(run_id);
    if let Some(tx_clone) = event_tx {
        emitter.on_event(move |event| {
            let _ = tx_clone.send(event.clone());
        });
    }
    let registry_override = state
        .registry_factory_override
        .as_ref()
        .map(|factory| Arc::new(factory(Arc::clone(&interview_runtime))));
    let emitter = Arc::new(emitter);
    let steering_hub = Arc::new(fabro_workflow::SteeringHub::new(Arc::clone(&emitter)));

    // Transition to Running, populate interviewer
    let cancelled_during_setup = {
        let mut runs = state.runs.lock().expect("runs lock poisoned");
        if let Some(managed_run) = runs.get_mut(&run_id) {
            if managed_run.status == RunStatus::Starting {
                managed_run.status = RunStatus::Running;
                managed_run.answer_transport = Some(RunAnswerTransport::InProcess {
                    interviewer:  Arc::clone(&interviewer),
                    steering_hub: Arc::clone(&steering_hub),
                });
                false
            } else {
                // Was cancelled during setup
                clear_live_run_state(managed_run);
                state.scheduler_notify.notify_one();
                true
            }
        } else {
            false
        }
    };
    if cancelled_during_setup {
        if let Err(err) = persist_cancelled_run_status(state.as_ref(), run_id).await {
            error!(run_id = %run_id, error = %err, "Failed to persist cancelled run status");
        }
        return;
    }

    let run_store = match state.stores.runs.open_run(&run_id).await {
        Ok(run_store) => run_store,
        Err(e) => {
            tracing::error!(run_id = %run_id, error = %e, "Failed to open run store");
            let mut runs = state.runs.lock().expect("runs lock poisoned");
            if let Some(managed_run) = runs.get_mut(&run_id) {
                managed_run.status = RunStatus::Failed {
                    reason: FailureReason::WorkflowError,
                };
                managed_run.error = Some(format!("Failed to open run store: {e}"));
                clear_live_run_state(managed_run);
            }
            state.scheduler_notify.notify_one();
            return;
        }
    };
    tokio::spawn(forward_run_events_to_global(
        Arc::clone(&state),
        run_id,
        run_store.subscribe(),
    ));
    let persisted = match Persisted::load_from_store(&run_store.clone().into(), &run_dir).await {
        Ok(persisted) => persisted,
        Err(e) => {
            tracing::error!(run_id = %run_id, error = %e, "Failed to load persisted run");
            fail_run_before_execution(
                &state,
                run_id,
                FailureReason::WorkflowError,
                format!("Failed to load persisted run: {e}"),
            )
            .await;
            return;
        }
    };
    let server_settings = state.server_settings();
    let github_settings = &server_settings.server.integrations.github;
    if cancel_token.is_cancelled() {
        finish_cancelled_run_before_execution(&state, run_id).await;
        return;
    }
    if reject_run_if_sandbox_provider_disabled(
        &state,
        &server_settings,
        run_id,
        &persisted.run_spec().settings.run,
    )
    .await
    {
        return;
    }
    let github_app_result = {
        let run_spec = persisted.run_spec();
        let settings = &run_spec.settings.run;
        let clone_can_use_github_credentials = settings.execution.mode != RunMode::DryRun
            && settings.environment.provider.clones_workspace()
            && run_spec
                .repo_origin_url()
                .is_some_and(|origin| !origin.trim().is_empty());
        let pull_request_can_use_github_credentials =
            settings.execution.mode != RunMode::DryRun && settings.pull_request.is_some();
        if settings.integrations.github.is_token_requested() {
            state.github_credentials(github_settings).await
        } else if clone_can_use_github_credentials || pull_request_can_use_github_credentials {
            match state.github_credentials(github_settings).await {
                Ok(github_app) => Ok(github_app),
                Err(err) => {
                    tracing::warn!(
                        run_id = %run_id,
                        error = %err,
                        "GitHub credentials unavailable; pull request creation will be skipped"
                    );
                    Ok(None)
                }
            }
        } else {
            Ok(None)
        }
    };
    let github_app = match github_app_result {
        Ok(github_app) => github_app,
        Err(e) => {
            if cancel_token.is_cancelled() {
                finish_cancelled_run_before_execution(&state, run_id).await;
                return;
            }
            tracing::error!(run_id = %run_id, error = %e, "Invalid GitHub credentials");
            fail_run_before_execution(
                &state,
                run_id,
                FailureReason::WorkflowError,
                format!("Invalid GitHub credentials: {e}"),
            )
            .await;
            return;
        }
    };
    let github_integration = match persisted
        .run_spec()
        .settings
        .run
        .integrations
        .github
        .resolve_integration()
    {
        Ok(integration) => integration,
        Err(err) => {
            tracing::error!(
                run_id = %run_id,
                error = %err,
                "GitHub permission interpolation failed"
            );
            fail_run_before_execution(
                &state,
                run_id,
                FailureReason::WorkflowError,
                format!("Failed to resolve GitHub permissions: {err}"),
            )
            .await;
            return;
        }
    };
    let vault = match state.stores.vault.snapshot().await {
        Ok(vault) => vault,
        Err(err) => {
            tracing::error!(run_id = %run_id, error = ?err, "Loading run secrets failed");
            fail_run_before_execution(
                &state,
                run_id,
                FailureReason::WorkflowError,
                "Loading run secrets failed".to_string(),
            )
            .await;
            return;
        }
    };
    let services = operations::StartServices {
        run_id,
        cancel_token: cancel_token.clone(),
        emitter: Arc::clone(&emitter),
        interviewer: Arc::clone(&interview_runtime),
        steering_hub: Arc::clone(&steering_hub),
        run_store: run_store.clone().into(),
        event_sink: workflow_event::RunEventSink::store(run_store.clone()),
        artifact_sink: Some(ArtifactSink::Store(state.artifact_store.clone())),
        run_control: None,
        github_app,
        github_integration,
        vault: Arc::new(AsyncRwLock::new(vault.into_vault())),
        sandbox_providers: state.server_settings().server.sandbox.providers.clone(),
        catalog: state.catalog(),
        on_node: None,
        registry_override,
        fabro_run_tools: None,
    };

    let execution = async {
        match execution_mode {
            RunExecutionMode::Start => operations::start(&run_dir, services).await,
            RunExecutionMode::Resume => operations::resume(&run_dir, services).await,
        }
    };

    let result = tokio::select! {
        result = execution => ExecutionResult::Completed(Box::new(result)),
        _ = cancel_rx => {
            cancel_token.cancel();
            ExecutionResult::CancelledBySignal
        }
    };

    if matches!(&result, ExecutionResult::CancelledBySignal) {
        if let Err(err) = persist_cancelled_run_status(state.as_ref(), run_id).await {
            error!(run_id = %run_id, error = %err, "Failed to persist cancelled run status");
        }
    }

    // Save final projection
    let final_projection = match run_store.state().await {
        Ok(state) => Some(state),
        Err(err) => {
            tracing::warn!(run_id = %run_id, error = %err, "Failed to load run state from store");
            None
        }
    };

    // Accumulate aggregate usage after execution completes.
    if let Some(ref projection) = final_projection {
        if projection.current_checkpoint().is_some() {
            let mut agg = state
                .aggregate_billing
                .lock()
                .expect("aggregate_billing lock poisoned");
            accumulate_billing_rollup(
                &mut agg,
                &fabro_workflow::billing_rollup_from_projection(projection),
            );
        }
    }

    let mut runs = state.runs.lock().expect("runs lock poisoned");
    if let Some(managed_run) = runs.get_mut(&run_id) {
        match &result {
            ExecutionResult::Completed(result) => {
                // A run can fail either before it produces a `Started` or in
                // its own outcome; both carry the same `WorkflowError`.
                let outcome = match result.as_ref() {
                    Ok(started) => started.finalized.outcome.as_ref().map(|_| ()),
                    Err(e) => Err(e),
                };
                match outcome {
                    Ok(()) => {
                        info!(run_id = %run_id, "Run completed");
                        managed_run.status = RunStatus::Succeeded {
                            reason: SuccessReason::Completed,
                        };
                    }
                    Err(WorkflowError::Cancelled) => {
                        info!(run_id = %run_id, "Run cancelled");
                        managed_run.status = RunStatus::Failed {
                            reason: FailureReason::Cancelled,
                        };
                    }
                    Err(e) => {
                        let detail = e.display_with_causes();
                        error!(run_id = %run_id, error = %detail, "Run failed");
                        managed_run.status = RunStatus::Failed {
                            reason: e.failure_reason(),
                        };
                        managed_run.error = Some(detail);
                    }
                }
            }
            ExecutionResult::CancelledBySignal => {
                info!(run_id = %run_id, "Run cancelled");
                managed_run.status = RunStatus::Failed {
                    reason: FailureReason::Cancelled,
                };
            }
        }
        managed_run.checkpoint = final_projection
            .as_ref()
            .and_then(|projection| projection.current_checkpoint().cloned());
        managed_run.run_dir = Some(run_dir);
        clear_live_run_state(managed_run);
    }
    drop(runs);
    state.scheduler_notify.notify_one();
}

async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
    let (run_dir, execution_mode) = {
        let mut runs = state.runs.lock().expect("runs lock poisoned");
        if state.is_shutting_down() {
            return;
        }
        let managed_run = match runs.get_mut(&run_id) {
            Some(run) if run.status == RunStatus::Runnable => run,
            _ => return,
        };
        let Some(run_dir) = managed_run.run_dir.clone() else {
            return;
        };
        managed_run.status = RunStatus::Starting;
        (run_dir, managed_run.execution_mode)
    };

    let run_store = match state.stores.runs.open_run(&run_id).await {
        Ok(run_store) => run_store,
        Err(err) => {
            tracing::error!(run_id = %run_id, error = %err, "Failed to open run store");
            fail_managed_run(
                &state,
                run_id,
                FailureReason::WorkflowError,
                format!("Failed to open run store: {err}"),
            );
            state.scheduler_notify.notify_one();
            return;
        }
    };
    tokio::spawn(forward_run_events_to_global(
        Arc::clone(&state),
        run_id,
        run_store.subscribe(),
    ));

    let run_state = match run_store.state().await {
        Ok(run_state) => run_state,
        Err(err) => {
            tracing::error!(run_id = %run_id, error = %err, "Failed to load run state");
            fail_managed_run(
                &state,
                run_id,
                FailureReason::WorkflowError,
                format!("Failed to load run state: {err}"),
            );
            state.scheduler_notify.notify_one();
            return;
        }
    };
    let agent_fabro_tools_enabled = run_state.spec.settings.run.agent.fabro_tools;
    if reject_run_if_sandbox_provider_disabled(
        &state,
        &state.server_settings(),
        run_id,
        &run_state.spec.settings.run,
    )
    .await
    {
        return;
    }

    let github_app_private_key = match state.vault_secret(EnvVars::GITHUB_APP_PRIVATE_KEY).await {
        Ok(value) => value,
        Err(err) => {
            fail_run_before_execution(
                &state,
                run_id,
                FailureReason::WorkflowError,
                "Loading worker secrets failed".to_string(),
            )
            .await;
            tracing::error!(run_id = %run_id, error = ?err, "Loading worker secrets failed");
            return;
        }
    };
    let state_for_build = Arc::clone(&state);
    let run_dir_for_build = run_dir.clone();
    let start_result = spawn_blocking(move || {
        worker_launch_spec(
            state_for_build.as_ref(),
            run_id,
            execution_mode,
            &run_dir_for_build,
            agent_fabro_tools_enabled,
            github_app_private_key,
        )
    })
    .await
    .context("worker_launch_spec task failed")
    .and_then(|inner| inner);

    let launch_result = match start_result {
        Ok(spec) => state.worker_runtime.start(spec).await,
        Err(err) => Err(err),
    };
    let started_worker = match launch_result {
        Ok(worker) => worker,
        Err(err) => {
            fail_worker_launch(&state, &run_store, run_id, err).await;
            return;
        }
    };
    let worker_ref = started_worker.worker_ref.clone();

    {
        let mut runs = state.runs.lock().expect("runs lock poisoned");
        if let Some(managed_run) = runs.get_mut(&run_id) {
            managed_run.worker_ref = Some(worker_ref.clone());
            managed_run.run_dir = Some(run_dir.clone());
            managed_run.answer_transport = Some(RunAnswerTransport::Worker {
                run_id,
                bus: Arc::clone(&state.worker_control_bus),
            });
        }
    }

    let stderr_task = tokio::spawn(drain_worker_stderr(run_id, started_worker.stderr));

    let worker_exit = match started_worker.wait.await {
        Ok(exit) => exit,
        Err(err) => {
            tracing::error!(run_id = %run_id, error = %err, "Failed while waiting on worker");
            let message = format!("Worker wait failed: {err}");
            state.worker_runtime.force_stop(&worker_ref).await;
            let failure_event = workflow_event::Event::workflow_run_failed_from_error(
                &WorkflowError::engine_with_source("Worker wait failed", err),
                fabro_types::RunTiming::default(),
                FailureReason::Terminated,
                None,
                None,
                None,
                None,
            );
            let _ = workflow_event::append_event(&run_store, &run_id, &failure_event).await;
            fail_managed_run(&state, run_id, FailureReason::Terminated, message);
            state.scheduler_notify.notify_one();
            return;
        }
    };

    match stderr_task.await {
        Ok(Ok(())) => {}
        Ok(Err(err)) => {
            tracing::warn!(run_id = %run_id, error = %err, "Worker stderr drain failed");
        }
        Err(err) => {
            tracing::warn!(run_id = %run_id, error = %err, "Worker stderr task panicked");
        }
    }

    let superseded = {
        let runs = state.runs.lock().expect("runs lock poisoned");
        runs.get(&run_id)
            .is_some_and(|managed_run| managed_run.worker_ref.as_ref() != Some(&worker_ref))
    };
    if superseded {
        tracing::info!(
            run_id = %run_id,
            worker_ref = ?worker_ref,
            "Skipping stale worker cleanup for superseded run execution"
        );
        return;
    }

    append_worker_exit_failure(&run_store, run_id, &worker_exit).await;

    let final_state = match run_store.state().await {
        Ok(state) => state,
        Err(err) => {
            tracing::warn!(run_id = %run_id, error = %err, "Failed to load final run state from store");
            fail_managed_run(
                &state,
                run_id,
                FailureReason::WorkflowError,
                format!("Failed to load final run state: {err}"),
            );
            state.scheduler_notify.notify_one();
            return;
        }
    };

    if final_state.current_checkpoint().is_some() {
        let mut agg = state
            .aggregate_billing
            .lock()
            .expect("aggregate_billing lock poisoned");
        accumulate_billing_rollup(
            &mut agg,
            &fabro_workflow::billing_rollup_from_projection(&final_state),
        );
    }

    let mut runs = state.runs.lock().expect("runs lock poisoned");
    if let Some(managed_run) = runs.get_mut(&run_id) {
        if final_state.status != managed_run.status {
            managed_run.status = final_state.status;
        } else if !worker_exit.success {
            managed_run.status = RunStatus::Failed {
                reason: FailureReason::Terminated,
            };
        }
        managed_run.error = final_state
            .conclusion
            .as_ref()
            .and_then(|conclusion| {
                conclusion.failure.as_ref().map(|failure| {
                    render_compact_with_causes(&failure.detail.message, &failure.detail.causes)
                })
            })
            .or_else(|| managed_run.error.clone());
        managed_run.checkpoint = final_state.current_checkpoint().cloned();
        managed_run.run_dir = Some(run_dir);
        clear_live_run_state(managed_run);
    }
    drop(runs);
    state.scheduler_notify.notify_one();
}

/// Background task that promotes runnable runs when capacity is available.
pub fn spawn_scheduler(state: Arc<AppState>) {
    tokio::spawn(async move {
        loop {
            tokio::select! {
                () = state.scheduler_notify.notified() => {},
                () = sleep(std::time::Duration::from_secs(1)) => {},
            }
            if state.is_shutting_down() {
                break;
            }
            let runs_to_start = {
                let runs = state.runs.lock().expect("runs lock poisoned");
                let active = runs
                    .values()
                    .filter(|r| counts_toward_scheduler_capacity(r.status))
                    .count();
                let available = state.max_concurrent_runs.saturating_sub(active);
                if available == 0 {
                    Vec::new()
                } else {
                    let mut runnable: Vec<_> = runs
                        .iter()
                        .filter(|(_, r)| r.status == RunStatus::Runnable)
                        .map(|(id, r)| (*id, r.created_at))
                        .collect();
                    runnable.sort_by_key(|(_, created_at)| *created_at);
                    runnable
                        .into_iter()
                        .take(available)
                        .map(|(id, _)| id)
                        .collect::<Vec<_>>()
                }
            };
            for id in runs_to_start {
                if state.is_shutting_down() {
                    break;
                }
                let state_clone = Arc::clone(&state);
                tokio::spawn(
                    execute_run(state_clone, id).instrument(tracing::info_span!("run", id = %id)),
                );
            }
        }
    });
}

async fn append_control_request(
    state: &AppState,
    run_id: RunId,
    action: RunControlAction,
    actor: Option<Principal>,
) -> anyhow::Result<()> {
    let run_store = state.stores.runs.open_run(&run_id).await?;
    let event = match action {
        RunControlAction::Cancel => workflow_event::Event::RunCancelRequested { actor },
        RunControlAction::Pause => workflow_event::Event::RunPauseRequested { actor },
        RunControlAction::Unpause => workflow_event::Event::RunUnpauseRequested { actor },
    };
    if action == RunControlAction::Cancel {
        workflow_event::append_event_if(&run_store, &run_id, &event, |projection| {
            projection.pending_control != Some(RunControlAction::Cancel)
        })
        .await
        .map(|_| ())
    } else {
        workflow_event::append_event(&run_store, &run_id, &event).await
    }
}

/// Returns a 409 response with an actionable "unarchive first" message if the
/// run is currently archived. Returns `None` otherwise (including when the run
/// doesn't exist — the caller's own not-found handling will surface that).
async fn reject_if_archived(state: &AppState, run_id: &RunId) -> Option<Response> {
    let summary = state
        .stores
        .run_summaries
        .get(run_id, Utc::now())
        .await
        .ok()
        .flatten()?;
    summary.lifecycle.archived_at.is_some().then(|| {
        ApiError::new(
            StatusCode::CONFLICT,
            operations::archived_rejection_message(run_id),
        )
        .into_response()
    })
}

#[cfg(test)]
#[expect(
    clippy::disallowed_methods,
    reason = "server unit tests stage fixtures with sync std::fs writes"
)]
mod tests;