obelisk 0.41.5

use crate::{
    command::server::{self, PreparedDirs, ServerVerified, SubmitError, SubmitOutcome},
    server::web_api_server::{
        backtrace::{execution_backtrace, execution_backtrace_source},
        components::{component_wit, components_list},
        deployment::{
            gc_orphan_files, get_current_deployment_id, get_deployment, get_file, list_deployments,
            submit_deployment, switch_deployment,
        },
        functions::{function_wit, functions_list},
    },
};
use axum::{
    Json, Router,
    body::{Body, Bytes},
    extract::{Path, State},
    response::{IntoResponse, Response},
    routing,
};
use axum_accept::AcceptExtractor;
use axum_extra::extract::Query;
use chrono::{DateTime, Utc};
use concepts::{
    ComponentType, ExecutionId, FinishedExecutionFailure, FunctionFqn, JoinSetId, JoinSetKind,
    StrVariant, SupportedFunctionReturnValue,
    component_id::ComponentDigest,
    prefixed_ulid::{DelayId, DeploymentId, ExecutionIdDerived},
    storage::{
        self, BacktraceFilter, CancelOutcome, DbErrorGeneric, DbErrorRead, DbErrorReadWithTimeout,
        DbErrorWrite, DbErrorWriteNonRetriable, DbPool, DelayCancelOutcome, ExecutionEvent,
        ExecutionListPagination, ExecutionRequest, ExecutionWithState, FunctionNameFilter,
        ListExecutionsFilter, Pagination, PendingState, PendingStateFinishedError,
        PendingStateFinishedResultKind, ResponseCursor, ResponseWithCursor, TimeoutOutcome,
        Version, VersionType,
    },
    time::{ClockFn as _, Now, Sleep as _},
};
use http::{StatusCode, header};
use serde::{Deserialize, Serialize};
use serde_json::json;
use std::num::NonZeroU16;
use std::sync::Arc;
use std::{fmt::Write as _, time::Duration};
use tokio::{
    select,
    sync::{mpsc, watch},
};
use tokio_stream::wrappers::ReceiverStream;
use tracing::{Instrument as _, Span, debug, info, info_span, instrument, trace, warn};
use utoipa::{IntoParams, OpenApi, ToSchema};
use val_json::{wast_val::WastVal, wast_val_ser::deserialize_value};
use wasm_workers::{
    activity::cancel_registry::CancelRegistry,
    registry::ReplayWorker,
    workflow::workflow_worker::{AdvanceError, BacktraceCapture},
};

#[derive(Clone)]
pub(crate) struct WebApiState {
    pub(crate) server_verified: ServerVerified,
    pub(crate) deployment_ctx: crate::command::server::DeploymentContextHandle,
    pub(crate) db_pool: Arc<dyn DbPool>,
    pub(crate) cancel_registry: CancelRegistry,
    pub(crate) termination_watcher: watch::Receiver<()>,
    pub(crate) subscription_interruption: Option<Duration>,
    pub(crate) prepared_dirs: PreparedDirs,
    pub(crate) deployment_switch_manager: crate::command::server::DeploymentSwitchManagerHandle,
}

/// `OpenAPI` documentation for the Obelisk REST API
#[derive(OpenApi)]
#[openapi(
    info(
        title = "Obelisk REST API",
        description = "REST API for the Obelisk deterministic workflow engine. JSON is the stable representation for structured resources and actions. Text responses are stable only for WIT documents, backtrace source files, and the execution and deployment ID endpoints; all other text/plain representations are deprecated.",
        version = "1.0.0"
    ),
    tags(
        (name = "executions", description = "Execution management"),
        (name = "components", description = "Component management"),
        (name = "functions", description = "Function management"),
        (name = "deployments", description = "Deployment management"),
        (name = "delays", description = "Delay management")
    ),
    paths(
        execution_id_generate,
        delay_cancel,
        delay_pause,
        delay_unpause,
        executions_list,
        execution_cancel,
        execution_pause,
        execution_unpause,
        execution_events,
        logs::execution_logs,
        execution_responses,
        execution_status_get,
        execution_stub,
        execution_get_retval,
        execution_submit_put,
        execution_submit_post,
        execution_replay,
        execution_advance,
        execution_persist_backtraces,
        execution_upgrade,
        backtrace::execution_backtrace,
        backtrace::execution_backtrace_source,
        components::component_wit,
        components::components_list,
        functions::functions_list,
        functions::function_wit,
        deployment::list_deployments,
        deployment::get_current_deployment_id,
        deployment::get_deployment,
        deployment::submit_deployment,
        deployment::switch_deployment,
        deployment::get_file,
        deployment::gc_orphan_files,
    ),
    components(schemas(
        PaginationDirectionSortedFromLatest,
        PaginationDirectionSortedFromOldest,
        ExecutionWithStateSer,
        ExecutionEventsResponse,
        ExecutionResponsesResponse,
        ExecutionStubPayload,
        RetVal,
        ReplayResponseSer,
        AdvanceRequestSer,
        AdvanceResponseSer,
        PersistBacktracesResponseSer,
        ExecutionSubmitPayload,
        ExecutionUpgradePayload,
        logs::LogEntryRowSer,
        logs::LogEntrySer,
        logs::LogLevelSer,
        logs::LogStreamTypeSer,
        logs::LogLevelParam,
        logs::LogStreamTypeParam,
        components::ComponentConfig,
        components::FunctionMetadataLite,
        components::ParameterTypeLite,
        functions::FunctionOutput,
        deployment::DeploymentStateSer,
        deployment::DeploymentSubmitResponse,
        deployment::GcOrphanFilesResponseSer,
        deployment::SubmitPackageErrorBody,
        deployment::FileIssue,
        deployment::DigestMismatch,
        backtrace::BacktraceInfoSer,
    ))
)]
pub(crate) struct ApiDoc;

/// Return the `OpenAPI` JSON schema
#[utoipa::path(
    get,
    path = "/openapi.json",
    responses(
        (status = 200, description = "OpenAPI JSON schema")
    )
)]
async fn openapi_json() -> impl IntoResponse {
    pretty_json_response(StatusCode::OK, &ApiDoc::openapi())
}

fn pretty_json_response<T: Serialize>(status: StatusCode, value: &T) -> Response {
    let body = serde_json::to_vec_pretty(value).expect("JSON response serialization must succeed");
    let mut response = (status, body).into_response();
    response.headers_mut().insert(
        header::CONTENT_TYPE,
        header::HeaderValue::from_static("application/json"),
    );
    response
}

fn deprecated_text_response(response: impl IntoResponse) -> Response {
    let mut response = response.into_response();
    response.headers_mut().insert(
        header::HeaderName::from_static("deprecation"),
        header::HeaderValue::from_static("@1787443200"),
    );
    response.headers_mut().append(
        header::LINK,
        header::HeaderValue::from_static("</openapi.json>; rel=\"deprecation\""),
    );
    response
}

pub(crate) fn app_router(state: WebApiState) -> Router {
    Router::new()
        .route("/openapi.json", routing::get(openapi_json))
        .nest("/v1", v1_router())
        .with_state(Arc::new(state))
}

fn v1_router() -> Router<Arc<WebApiState>> {
    Router::new()
        .route("/components", routing::get(components_list))
        .route("/components/{digest}/wit", routing::get(component_wit))
        .route("/functions", routing::get(functions_list))
        .route("/functions/wit", routing::get(function_wit))
        .route("/delays/{delay-id}/cancel", routing::put(delay_cancel))
        .route("/delays/{delay-id}/pause", routing::put(delay_pause))
        .route("/delays/{delay-id}/unpause", routing::put(delay_unpause))
        .route("/execution-id", routing::get(execution_id_generate))
        .route("/executions", routing::get(executions_list))
        .route("/executions", routing::post(execution_submit_post))
        .route(
            "/executions/{execution-id}/cancel",
            routing::put(execution_cancel),
        )
        .route(
            "/executions/{execution-id}/pause",
            routing::put(execution_pause),
        )
        .route(
            "/executions/{execution-id}/unpause",
            routing::put(execution_unpause),
        )
        .route(
            "/executions/{execution-id}/events",
            routing::get(execution_events),
        )
        .route(
            "/executions/{execution-id}/logs",
            routing::get(logs::execution_logs),
        )
        .route(
            "/executions/{execution-id}/replay",
            routing::put(execution_replay),
        )
        .route(
            "/executions/{execution-id}/advance",
            routing::put(execution_advance),
        )
        .route(
            "/executions/{execution-id}/backtrace/persist",
            routing::put(execution_persist_backtraces),
        )
        .route(
            "/executions/{execution-id}/responses",
            routing::get(execution_responses),
        )
        .route(
            "/executions/{execution-id}/status",
            routing::get(execution_status_get),
        )
        .route(
            "/executions/{execution-id}/stub",
            routing::put(execution_stub),
        )
        .route(
            "/executions/{execution-id}",
            routing::get(execution_get_retval),
        )
        .route(
            "/executions/{execution-id}",
            routing::put(execution_submit_put),
        )
        .route(
            "/executions/{execution-id}/upgrade",
            routing::put(execution_upgrade),
        )
        .route("/deployments", routing::get(list_deployments))
        .route("/deployments", routing::post(submit_deployment))
        .route("/deployments/{deployment-id}", routing::get(get_deployment))
        .route("/files/orphans", routing::delete(gc_orphan_files))
        .route("/files/{digest}", routing::get(get_file))
        .route(
            "/deployments/{deployment-id}/switch",
            routing::put(switch_deployment),
        )
        .route("/deployment-id", routing::get(get_current_deployment_id))
        .route(
            "/executions/{execution-id}/backtrace",
            routing::get(execution_backtrace),
        )
        .route(
            "/executions/{execution-id}/backtrace/source",
            routing::get(execution_backtrace_source),
        )
}

/// Generate a new execution ID
#[utoipa::path(
    get,
    path = "/v1/execution-id",
    tag = "executions",
    responses(
        (status = 200, description = "Generated execution ID", body = String)
    )
)]
async fn execution_id_generate(
    _: State<Arc<WebApiState>>,
    accept: TextDefaultAcceptHeader,
) -> Response {
    let accept = accept.into();
    let id = ExecutionId::generate();
    match accept {
        AcceptHeader::Json => pretty_json_response(StatusCode::OK, &id),
        AcceptHeader::Text => id.to_string().into_response(),
    }
}

/// Cancel a delay
#[utoipa::path(
    put,
    path = "/v1/delays/{delay_id}/cancel",
    tag = "delays",
    params(
        ("delay_id" = String, Path, description = "Delay ID to cancel")
    ),
    responses(
        (status = 200, description = "Delay cancelled"),
        (status = 409, description = "Already finished")
    )
)]
#[instrument(skip_all, fields(delay_id))]
async fn delay_cancel(
    Path(delay_id): Path<DelayId>,
    state: State<Arc<WebApiState>>,
    accept: AcceptHeader,
) -> Result<Response, HttpResponse> {
    let conn = state
        .db_pool
        .external_api_conn()
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    let executed_at = Now.now();
    let outcome = storage::cancel_delay(conn.as_ref(), delay_id, executed_at)
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    Ok(HttpResponse::from_delay_cancel_outcome(outcome, accept).into_response())
}

/// Pause a delay
#[utoipa::path(
    put,
    path = "/v1/delays/{delay_id}/pause",
    tag = "delays",
    params(
        ("delay_id" = String, Path, description = "Delay ID to pause")
    ),
    responses(
        (status = 200, description = "Delay paused"),
        (status = 404, description = "Delay not found")
    )
)]
#[instrument(skip_all, fields(delay_id))]
async fn delay_pause(
    Path(delay_id): Path<DelayId>,
    state: State<Arc<WebApiState>>,
    accept: AcceptHeader,
) -> Result<Response, HttpResponse> {
    let conn = state
        .db_pool
        .external_api_conn()
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    conn.pause_delay(&delay_id)
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    Ok(HttpResponse {
        status: StatusCode::OK,
        message: "paused".to_string(),
        accept,
    }
    .into_response())
}

/// Unpause a delay
#[utoipa::path(
    put,
    path = "/v1/delays/{delay_id}/unpause",
    tag = "delays",
    params(
        ("delay_id" = String, Path, description = "Delay ID to unpause")
    ),
    responses(
        (status = 200, description = "Delay unpaused"),
        (status = 404, description = "Delay not found")
    )
)]
#[instrument(skip_all, fields(delay_id))]
async fn delay_unpause(
    Path(delay_id): Path<DelayId>,
    state: State<Arc<WebApiState>>,
    accept: AcceptHeader,
) -> Result<Response, HttpResponse> {
    let conn = state
        .db_pool
        .external_api_conn()
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    conn.unpause_delay(&delay_id)
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    Ok(HttpResponse {
        status: StatusCode::OK,
        message: "unpaused".to_string(),
        accept,
    }
    .into_response())
}

#[derive(Deserialize, Debug, IntoParams)]
#[into_params(parameter_in = Query)]
struct ExecutionsListParams {
    /// Filter by function FQN prefix
    ffqn_prefix: Option<String>,
    /// Show derived executions (child executions spawned by workflows)
    #[serde(default)]
    show_derived: bool,
    /// Hide finished executions
    #[serde(default)]
    hide_finished: bool,
    /// Filter by execution ID prefix
    execution_id_prefix: Option<String>,
    /// Filter by component digest
    #[serde(default)]
    #[param(value_type = Option<String>)]
    component_digest: Option<ComponentDigest>,
    /// Filter by deployment ID
    #[serde(default)]
    #[param(value_type = Option<String>)]
    deployment_id: Option<DeploymentId>,
    // pagination
    /// Pagination cursor (`DateTime` or `ExecutionId`)
    #[param(value_type = Option<String>)]
    cursor: Option<ExecutionListCursorDeser>,
    /// Number of items to return
    #[param(minimum = 1)]
    length: Option<u16>,
    /// Include the cursor item in results
    #[serde(default)]
    including_cursor: bool,
    /// Pagination direction
    #[serde(default)]
    direction: PaginationDirectionSortedFromLatest,
}

#[derive(Debug, Clone, Copy, Deserialize, Default, ToSchema)]
#[serde(rename_all = "snake_case")]
enum PaginationDirectionSortedFromLatest {
    /// Fetch items older than cursor
    #[default] // Default = last few items from newest
    Older,
    /// Fetch items newer than cursor
    Newer,
}

#[derive(Debug, Clone, Copy, Deserialize, Default, ToSchema)]
#[serde(rename_all = "snake_case")]
enum PaginationDirectionSortedFromOldest {
    /// Fetch items older than cursor
    Older,
    /// Fetch items newer than cursor
    #[default] // Default = start from 0
    Newer,
}

fn nonzero_page_length(length: u16, accept: AcceptHeader) -> Result<NonZeroU16, HttpResponse> {
    NonZeroU16::new(length).ok_or_else(|| {
        HttpResponse::bad_request(accept, "`length` must be greater than zero".to_string())
    })
}

#[derive(Deserialize, Debug)]
#[serde(untagged)]
enum ExecutionListCursorDeser {
    CreatedBy(DateTime<Utc>),
    ExecutionId(ExecutionId),
}
/// Execution with current state information
#[derive(Serialize, Deserialize, ToSchema)]
pub struct ExecutionWithStateSer {
    /// Unique execution identifier
    #[schema(value_type = String, example = "E_01JKXYZ123456789ABCDEFGHIJ")]
    pub execution_id: ExecutionId,
    /// Fully qualified function name
    #[schema(value_type = String, example = "my-pkg:my-ifc/my-fn")]
    pub ffqn: FunctionFqn,
    /// Current pending state of the execution
    #[schema(value_type = Object)]
    pub pending_state: PendingState,
    /// When the execution was created
    pub created_at: DateTime<Utc>,
    /// When the execution was first scheduled
    pub first_scheduled_at: DateTime<Utc>,
    /// Component content digest
    #[schema(value_type = String)]
    pub component_digest: ComponentDigest,
    /// Type of component (workflow, activity, webhook)
    #[schema(value_type = String)]
    pub component_type: ComponentType,
    /// Deployment that created this execution
    #[schema(value_type = String, example = "Dep_01JKXYZ123456789ABCDEFGHIJ")]
    pub deployment_id: DeploymentId,
}
impl From<ExecutionWithState> for ExecutionWithStateSer {
    fn from(value: ExecutionWithState) -> Self {
        let ExecutionWithState {
            execution_id,
            ffqn,
            pending_state,
            created_at,
            first_scheduled_at,
            component_digest,
            component_type,
            deployment_id,
        } = value;
        ExecutionWithStateSer {
            execution_id,
            ffqn,
            pending_state,
            created_at,
            first_scheduled_at,
            component_digest,
            component_type,
            deployment_id,
        }
    }
}

/// List executions with filtering and pagination
#[utoipa::path(
    get,
    path = "/v1/executions",
    tag = "executions",
    params(ExecutionsListParams),
    responses(
        (status = 200, description = "List of executions", body = Vec<ExecutionWithStateSer>)
    )
)]
#[instrument(skip_all)]
async fn executions_list(
    state: State<Arc<WebApiState>>,
    Query(params): Query<ExecutionsListParams>,
    accept: AcceptHeader,
) -> Result<Response, HttpResponse> {
    let default_pagination = ExecutionListPagination::default();
    let pagination = {
        let ExecutionsListParams {
            cursor,
            length,
            including_cursor,
            direction,
            ..
        } = params;
        let length = nonzero_page_length(length.unwrap_or(default_pagination.length()), accept)?;
        match cursor {
            Some(ExecutionListCursorDeser::CreatedBy(cursor)) => {
                ExecutionListPagination::CreatedBy(match direction {
                    PaginationDirectionSortedFromLatest::Older => Pagination::OlderThan {
                        length,
                        cursor: Some(cursor),
                        including_cursor,
                    },
                    PaginationDirectionSortedFromLatest::Newer => Pagination::NewerThan {
                        length,
                        cursor: Some(cursor),
                        including_cursor,
                    },
                })
            }
            Some(ExecutionListCursorDeser::ExecutionId(cursor)) => {
                ExecutionListPagination::ExecutionId(match direction {
                    PaginationDirectionSortedFromLatest::Older => Pagination::OlderThan {
                        length,
                        cursor: Some(cursor),
                        including_cursor,
                    },
                    PaginationDirectionSortedFromLatest::Newer => Pagination::NewerThan {
                        length,
                        cursor: Some(cursor),
                        including_cursor,
                    },
                })
            }
            None => ExecutionListPagination::CreatedBy(
                // CreatedBy because it is the current default
                match direction {
                    PaginationDirectionSortedFromLatest::Older => Pagination::OlderThan {
                        length,
                        cursor: None,
                        including_cursor, // does not matter
                    },

                    PaginationDirectionSortedFromLatest::Newer => Pagination::NewerThan {
                        length,
                        cursor: None,
                        including_cursor, // does not matter
                    },
                },
            ),
        }
    };

    let conn = state
        .db_pool
        .external_api_conn()
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;

    let filter = ListExecutionsFilter {
        function_name_filter: {
            // Map `ffqn_prefix` to a FunctionName.
            // If this is a package name with a version, the search will not find anything as the
            // FFQN contains the version behind the interface.
            params.ffqn_prefix.map(FunctionNameFilter::FunctionName)
        },
        show_derived: params.show_derived,
        hide_finished: params.hide_finished,
        execution_id_prefix: params.execution_id_prefix,
        component_digest: params.component_digest,
        deployment_id: params.deployment_id,
        state_filters: Vec::new(),
    };

    let executions = conn
        .list_executions(filter, pagination)
        .await
        .map_err(|err| ErrorWrapper(err, accept))?;
    Ok(match accept {
        AcceptHeader::Text => {
            let mut output = String::new();
            for execution in executions {
                writeln!(
                    &mut output,
                    "{id} `{pending_state}` {ffqn} `{first_scheduled_at}`",
                    id = execution.execution_id,
                    ffqn = execution.ffqn,
                    pending_state = execution.pending_state,
                    first_scheduled_at = execution.first_scheduled_at,
                )
                .expect("writing to string");
            }
            deprecated_text_response(output)
        }
        AcceptHeader::Json => {
            let executions: Vec<_> = executions
                .into_iter()
                .map(ExecutionWithStateSer::from)
                .collect();
            pretty_json_response(StatusCode::OK, &executions)
        }
    })
}

/// Cancel an activity or cancellable workflow execution
#[utoipa::path(
    put,
    path = "/v1/executions/{execution_id}/cancel",
    tag = "executions",
    params(
        ("execution_id" = String, Path, description = "Execution ID to cancel")
    ),
    responses(
        (status = 200, description = "Cancellation requested"),
        (status = 409, description = "Already finished or already cancelling"),
        (status = 422, description = "Not an activity or cancellable workflow")
    )
)]
#[instrument(skip_all, fields(execution_id))]
async fn execution_cancel(
    Path(execution_id): Path<ExecutionId>,
    state: State<Arc<WebApiState>>,
    accept: AcceptHeader,
) -> Result<Response, HttpResponse> {
    let conn = state
        .db_pool
        .external_api_conn()
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    let create_req = conn
        .get_create_request(&execution_id)
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    let executed_at = Now.now();
    let outcome = match create_req.component_id.component_type {
        component_type if component_type.is_activity() => {
            state
                .cancel_registry
                .cancel_activity(conn.as_ref(), &execution_id, executed_at)
                .await
        }
        ComponentType::Workflow => {
            let outcome = conn
                .cancel_workflow_with_retries(&execution_id, executed_at)
                .await;
            if outcome.is_ok() {
                // Best-effort CPU saver: the write above already cancels it; a locally-running
                // workflow would otherwise burn CPU until its next db append fails on that write.
                state
                    .cancel_registry
                    .signal_workflow_interrupt(&execution_id);
            }
            outcome
        }
        _ => {
            return Err(HttpResponse {
                status: StatusCode::UNPROCESSABLE_ENTITY,
                message: "cancelled execution must be an activity or cancellable workflow"
                    .to_string(),
                accept,
            });
        }
    }
    .map_err(|e| ErrorWrapper(e, accept))?;
    Ok(HttpResponse::from_cancel_execution_outcome(outcome, accept).into_response())
}

#[instrument(skip_all, fields(execution_id))]
/// Pause an execution
#[utoipa::path(
    put,
    path = "/v1/executions/{execution_id}/pause",
    tag = "executions",
    params(
        ("execution_id" = String, Path, description = "Execution ID to pause")
    ),
    responses(
        (status = 200, description = "Execution paused")
    )
)]
async fn execution_pause(
    Path(execution_id): Path<ExecutionId>,
    state: State<Arc<WebApiState>>,
    accept: AcceptHeader,
) -> Result<Response, HttpResponse> {
    info!("Pausing execution");
    let conn = state
        .db_pool
        .external_api_conn()
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    let paused_at = Now.now();
    conn.pause_execution(&execution_id, paused_at)
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    // Best-effort CPU saver: the write above already pauses it; a locally-running
    // workflow would otherwise burn CPU until its next db append fails on that write.
    state
        .cancel_registry
        .signal_workflow_interrupt(&execution_id);
    Ok(HttpResponse {
        status: StatusCode::OK,
        message: "paused".to_string(),
        accept,
    }
    .into_response())
}

/// Unpause an execution
#[utoipa::path(
    put,
    path = "/v1/executions/{execution_id}/unpause",
    tag = "executions",
    params(
        ("execution_id" = String, Path, description = "Execution ID to unpause")
    ),
    responses(
        (status = 200, description = "Execution unpaused")
    )
)]
#[instrument(skip_all, fields(execution_id))]
async fn execution_unpause(
    Path(execution_id): Path<ExecutionId>,
    state: State<Arc<WebApiState>>,
    accept: AcceptHeader,
) -> Result<Response, HttpResponse> {
    info!("Unpausing execution");
    let conn = state
        .db_pool
        .external_api_conn()
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    let unpaused_at = Now.now();
    conn.unpause_execution(&execution_id, unpaused_at)
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    Ok(HttpResponse {
        status: StatusCode::OK,
        message: "unpaused".to_string(),
        accept,
    }
    .into_response())
}

#[derive(Debug, Deserialize, IntoParams)]
#[into_params(parameter_in = Query)]
struct ExecutionEventsParams {
    /// Version cursor for pagination
    version: Option<VersionType>,
    /// Number of events to return
    #[param(minimum = 1)]
    length: Option<u16>,
    /// Include the cursor item in results
    #[serde(default)]
    including_cursor: bool,
    /// Pagination direction
    #[serde(default)]
    direction: PaginationDirectionSortedFromOldest,
    /// Include backtrace IDs in events
    #[serde(default)]
    include_backtrace_id: bool,
}

/// Response containing execution events
#[derive(Serialize, Deserialize, ToSchema)]
pub(crate) struct ExecutionEventsResponse {
    /// List of execution events
    #[schema(value_type = Vec<Object>)]
    pub(crate) events: Vec<ExecutionEvent>,
    /// Maximum version in the response
    #[schema(value_type = u32)]
    pub(crate) max_version: Version,
}
/// Get execution events (history)
#[utoipa::path(
    get,
    path = "/v1/executions/{execution_id}/events",
    tag = "executions",
    params(
        ("execution_id" = String, Path, description = "Execution ID"),
        ExecutionEventsParams
    ),
    responses(
        (status = 200, description = "Execution events", body = ExecutionEventsResponse)
    )
)]
#[instrument(skip_all, fields(execution_id))]
async fn execution_events(
    Path(execution_id): Path<ExecutionId>,
    state: State<Arc<WebApiState>>,
    Query(params): Query<ExecutionEventsParams>,
    accept: AcceptHeader,
) -> Result<Response, HttpResponse> {
    const DEFAULT_LENGTH: u16 = 20;
    let length = nonzero_page_length(params.length.unwrap_or(DEFAULT_LENGTH), accept)?;
    let conn = state
        .db_pool
        .external_api_conn()
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    let pagination = match params.direction {
        PaginationDirectionSortedFromOldest::Older => Pagination::OlderThan {
            length,
            cursor: params.version.unwrap_or(VersionType::MAX),
            including_cursor: params.including_cursor,
        },
        PaginationDirectionSortedFromOldest::Newer => Pagination::NewerThan {
            length,
            cursor: params.version.unwrap_or(0),
            including_cursor: params.including_cursor,
        },
    };
    let result = conn
        .list_execution_events(&execution_id, pagination, params.include_backtrace_id)
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    Ok(match accept {
        AcceptHeader::Json => pretty_json_response(
            StatusCode::OK,
            &ExecutionEventsResponse {
                events: result.events,
                max_version: result.max_version,
            },
        ),
        AcceptHeader::Text => {
            let mut output = String::new();
            for event in result.events {
                writeln!(
                    &mut output,
                    "{version} `{created_at}` {event}",
                    version = event.version,
                    created_at = event.created_at,
                    event = event.event,
                )
                .expect("writing to string");
            }
            deprecated_text_response(output)
        }
    })
}

pub(crate) mod logs {
    use super::*;
    use base64::{Engine as _, prelude::BASE64_STANDARD};
    use chrono::{DateTime, Utc};
    use concepts::{
        prefixed_ulid::RunId,
        storage::{LogCursor, LogEntry, LogEntryRow, LogFilter, LogLevel, LogStreamType},
    };
    use std::fmt::Display;

    #[derive(Deserialize, Debug, IntoParams)]
    #[into_params(parameter_in = Query)]
    #[expect(clippy::struct_excessive_bools)]
    pub(crate) struct ExecutionLogsParams {
        /// Filter by log levels
        #[serde(default)]
        #[param(value_type = Vec<String>)]
        level: Vec<LogLevelParam>,
        /// Filter by stream types (stdout, stderr)
        #[serde(default)]
        #[param(value_type = Vec<String>)]
        stream_type: Vec<LogStreamTypeParam>,
        /// Include log entries
        #[serde(default = "default_true")]
        show_logs: bool,
        /// Include stream entries (stdout/stderr)
        #[serde(default = "default_true")]
        show_streams: bool,
        /// Include logs from all derived executions
        #[serde(default)]
        show_derived: bool,
        /// Include the run ID in text output
        #[serde(default)]
        show_run_id: bool,

        // pagination
        /// Opaque cursor for pagination. RFC 3339 timestamps are accepted for compatibility.
        cursor: Option<String>,
        /// Only include entries created after this timestamp.
        after: Option<DateTime<Utc>>,
        /// Number of entries to return
        #[param(minimum = 1)]
        length: Option<u16>,
        /// Include the cursor item in results
        #[serde(default)]
        including_cursor: bool,
        /// Pagination direction
        #[serde(default)]
        direction: PaginationDirectionSortedFromOldest,
    }

    fn default_true() -> bool {
        true
    }

    #[derive(Debug, Deserialize, Clone, Copy, ToSchema)]
    #[serde(rename_all = "snake_case")]
    pub(crate) enum LogLevelParam {
        Trace,
        Debug,
        Info,
        Warn,
        Error,
    }

    impl From<LogLevelParam> for LogLevel {
        fn from(value: LogLevelParam) -> Self {
            match value {
                LogLevelParam::Trace => LogLevel::Trace,
                LogLevelParam::Debug => LogLevel::Debug,
                LogLevelParam::Info => LogLevel::Info,
                LogLevelParam::Warn => LogLevel::Warn,
                LogLevelParam::Error => LogLevel::Error,
            }
        }
    }

    #[derive(Debug, Deserialize, Clone, Copy, ToSchema)]
    #[serde(rename_all = "snake_case")]
    pub(crate) enum LogStreamTypeParam {
        Stdout,
        Stderr,
    }

    impl From<LogStreamTypeParam> for LogStreamType {
        fn from(value: LogStreamTypeParam) -> Self {
            match value {
                LogStreamTypeParam::Stdout => LogStreamType::StdOut,
                LogStreamTypeParam::Stderr => LogStreamType::StdErr,
            }
        }
    }

    /// Log entry row with cursor
    #[derive(Serialize, Deserialize, ToSchema)]
    pub(crate) struct LogEntryRowSer {
        /// Opaque cursor position
        pub cursor: String,
        /// Run ID that produced this log
        #[schema(value_type = String)]
        pub run_id: RunId,
        /// Execution ID that produced this log
        pub execution_id: String,
        /// Log entry details
        #[serde(flatten)]
        pub info: LogEntrySer,
    }

    /// Log entry content
    #[derive(Serialize, Deserialize, ToSchema)]
    #[serde(tag = "type", rename_all = "snake_case")]
    pub(crate) enum LogEntrySer {
        /// Structured log message
        Log {
            created_at: DateTime<Utc>,
            level: LogLevelSer,
            message: String,
        },
        /// Raw stream output (stdout/stderr)
        Stream {
            created_at: DateTime<Utc>,
            /// Base64 encoded payload
            payload: String,
            stream_type: LogStreamTypeSer,
        },
    }

    impl From<LogEntryRow> for LogEntryRowSer {
        fn from(row: LogEntryRow) -> Self {
            Self {
                cursor: BASE64_STANDARD
                    .encode(serde_json::to_vec(&row.cursor).expect("log cursor must serialize")),
                run_id: row.run_id,
                execution_id: row.execution_id.to_string(),
                info: match row.log_entry {
                    LogEntry::Log {
                        created_at,
                        level,
                        message,
                    } => LogEntrySer::Log {
                        created_at,
                        level: level.into(),
                        message,
                    },
                    LogEntry::Stream {
                        created_at,
                        payload,
                        stream_type,
                    } => LogEntrySer::Stream {
                        created_at,
                        payload: BASE64_STANDARD.encode(payload),
                        stream_type: stream_type.into(),
                    },
                },
            }
        }
    }

    #[derive(serde::Serialize, serde::Deserialize, ToSchema)]
    #[serde(rename_all = "snake_case")]
    pub(crate) enum LogLevelSer {
        Trace,
        Debug,
        Info,
        Warn,
        Error,
    }

    impl Display for LogLevelSer {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            f.pad(match self {
                Self::Trace => "TRACE",
                Self::Debug => "DEBUG",
                Self::Info => "INFO",
                Self::Warn => "WARN",
                Self::Error => "ERROR",
            })
        }
    }

    impl From<LogLevel> for LogLevelSer {
        fn from(value: LogLevel) -> Self {
            match value {
                LogLevel::Trace => Self::Trace,
                LogLevel::Debug => Self::Debug,
                LogLevel::Info => Self::Info,
                LogLevel::Warn => Self::Warn,
                LogLevel::Error => Self::Error,
            }
        }
    }

    #[derive(serde::Serialize, serde::Deserialize, ToSchema)]
    #[serde(rename_all = "snake_case")]
    pub(crate) enum LogStreamTypeSer {
        Stdout,
        Stderr,
    }

    impl Display for LogStreamTypeSer {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            f.pad(match self {
                Self::Stdout => "STDOUT",
                Self::Stderr => "STDERR",
            })
        }
    }

    impl From<LogStreamType> for LogStreamTypeSer {
        fn from(value: LogStreamType) -> Self {
            match value {
                LogStreamType::StdOut => Self::Stdout,
                LogStreamType::StdErr => Self::Stderr,
            }
        }
    }

    /// Get execution logs
    #[utoipa::path(
        get,
        path = "/v1/executions/{execution_id}/logs",
        tag = "executions",
        params(
            ("execution_id" = String, Path, description = "Execution ID"),
            ExecutionLogsParams
        ),
        responses(
            (status = 200, description = "Execution logs", body = Vec<LogEntryRowSer>)
        )
    )]
    #[instrument(skip_all, fields(execution_id))]
    pub(crate) async fn execution_logs(
        Path(execution_id): Path<ExecutionId>,
        state: State<Arc<WebApiState>>,
        Query(params): Query<ExecutionLogsParams>,
        accept: AcceptHeader,
    ) -> Result<Response, HttpResponse> {
        const DEFAULT_LENGTH: u16 = 20;
        const MAX_LENGTH_INCLUSIVE: u16 = 200;

        let (cursor, legacy_after) = match params.cursor.as_deref() {
            Some(cursor) => {
                let opaque = BASE64_STANDARD
                    .decode(cursor)
                    .ok()
                    .and_then(|decoded| serde_json::from_slice::<LogCursor>(&decoded).ok());
                let legacy_after = DateTime::parse_from_rfc3339(cursor)
                    .ok()
                    .map(|created_at| created_at.with_timezone(&Utc));
                if opaque.is_none() && legacy_after.is_none() {
                    return Err(HttpResponse {
                        status: StatusCode::BAD_REQUEST,
                        message: "invalid log cursor".to_string(),
                        accept,
                    });
                }
                (opaque, legacy_after)
            }
            None => (None, None),
        };

        let (legacy_after, legacy_before) = match params.direction {
            PaginationDirectionSortedFromOldest::Newer => (legacy_after, None),
            PaginationDirectionSortedFromOldest::Older => (None, legacy_after),
        };
        let filter = match (params.show_logs, params.show_streams) {
            (true, true) => LogFilter::show_combined(
                params.level.into_iter().map(Into::into).collect(),
                params.stream_type.into_iter().map(Into::into).collect(),
            ),
            (true, false) => {
                LogFilter::show_logs(params.level.into_iter().map(Into::into).collect())
            }
            (false, true) => {
                LogFilter::show_streams(params.stream_type.into_iter().map(Into::into).collect())
            }
            (false, false) => {
                return Err(HttpResponse {
                    status: StatusCode::BAD_REQUEST,
                    message: "at least one of `show_logs`, `show_streams` must be set".to_string(),
                    accept,
                });
            }
        }
        .with_created_bounds(params.after.or(legacy_after), legacy_before);

        let length = nonzero_page_length(
            MAX_LENGTH_INCLUSIVE.min(params.length.unwrap_or(DEFAULT_LENGTH)),
            accept,
        )?;

        let pagination = match params.direction {
            PaginationDirectionSortedFromOldest::Older => Pagination::OlderThan {
                length,
                cursor: cursor.unwrap_or(LogCursor(i64::MAX)),
                including_cursor: params.including_cursor,
            },
            PaginationDirectionSortedFromOldest::Newer => Pagination::NewerThan {
                length,
                cursor: cursor.unwrap_or(LogCursor(i64::MIN)),
                including_cursor: params.including_cursor,
            },
        };

        let conn = state
            .db_pool
            .external_api_conn()
            .await
            .map_err(|e| ErrorWrapper(e, accept))?;

        let result = conn
            .list_logs(&execution_id, params.show_derived, filter, pagination)
            .await
            .map_err(|e| ErrorWrapper(e, accept))?;

        Ok(match accept {
            AcceptHeader::Json => {
                let items: Vec<LogEntryRowSer> =
                    result.items.into_iter().map(LogEntryRowSer::from).collect();
                pretty_json_response(StatusCode::OK, &items)
            }
            AcceptHeader::Text => {
                let mut output = String::new();
                struct PrefixId<'a>(bool, &'a dyn Display);
                impl Display for PrefixId<'_> {
                    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
                        if self.0 {
                            write!(f, "{} ", self.1)
                        } else {
                            Ok(())
                        }
                    }
                }
                for log in result.items {
                    match log.log_entry {
                        LogEntry::Log {
                            created_at,
                            level,
                            message,
                        } => {
                            let level = LogLevelSer::from(level);
                            writeln!(
                                &mut output,
                                "{created_at} [{level:<6}] {run_id}{exec_id}{message}",
                                created_at = created_at.format("%Y-%m-%dT%H:%M:%S%.9fZ"),
                                run_id = PrefixId(params.show_run_id, &log.run_id),
                                exec_id = PrefixId(params.show_derived, &log.execution_id),
                            )
                            .expect("writing to string");
                        }
                        LogEntry::Stream {
                            created_at,
                            payload,
                            stream_type,
                        } => {
                            let stream_type = LogStreamTypeSer::from(stream_type);
                            let payload_utf8 = String::from_utf8_lossy(&payload);
                            writeln!(
                                &mut output,
                                "{created_at} [{stream_type:<6}] {run_id}{exec_id}{payload_utf8}",
                                created_at = created_at.format("%Y-%m-%dT%H:%M:%S%.9fZ"),
                                run_id = PrefixId(params.show_run_id, &log.run_id),
                                exec_id = PrefixId(params.show_derived, &log.execution_id),
                            )
                            .expect("writing to string");
                        }
                    }
                }
                deprecated_text_response(output)
            }
        })
    }
}

#[derive(Debug, Deserialize, IntoParams)]
#[into_params(parameter_in = Query)]
struct ExecutionResponsesParams {
    /// Filter by an exact join-set ID (bare names select named join sets)
    join_set: Option<String>,
    /// Cursor for pagination
    cursor: Option<u32>,
    /// Maximum number of responses to return
    #[param(minimum = 1)]
    length: Option<u16>,
    /// Include the cursor item in results
    #[serde(default)]
    including_cursor: bool,
    /// Pagination direction
    #[serde(default)]
    direction: PaginationDirectionSortedFromOldest,
}

fn parse_join_set_filter(join_set: String) -> Result<JoinSetId, String> {
    if join_set.contains(':') {
        join_set
            .parse()
            .map_err(|err: concepts::JoinSetIdParseError| err.to_string())
    } else {
        // backcompat: 0.41.4 accepted bare names as named join-set IDs.
        JoinSetId::new(JoinSetKind::Named, StrVariant::from(join_set))
            .map_err(|err| err.to_string())
    }
}

/// Response containing execution responses
#[derive(Serialize, Deserialize, ToSchema)]
pub(crate) struct ExecutionResponsesResponse {
    /// List of responses with cursors
    #[schema(value_type = Vec<Object>)]
    pub(crate) responses: Vec<ResponseWithCursor>,
    /// Maximum cursor in the response
    #[schema(value_type = u32)]
    pub(crate) max_cursor: ResponseCursor,
    /// Cursor reached in the requested direction
    #[schema(value_type = u32)]
    pub(crate) scan_cursor: ResponseCursor,
}
/// Get execution responses (join set responses)
#[utoipa::path(
    get,
    path = "/v1/executions/{execution_id}/responses",
    tag = "executions",
    params(
        ("execution_id" = String, Path, description = "Execution ID"),
        ExecutionResponsesParams
    ),
    responses(
        (status = 200, description = "Execution responses", body = ExecutionResponsesResponse)
    )
)]
#[instrument(skip_all, fields(execution_id))]
async fn execution_responses(
    Path(execution_id): Path<ExecutionId>,
    state: State<Arc<WebApiState>>,
    Query(params): Query<ExecutionResponsesParams>,
    accept: AcceptHeader,
) -> Result<Response, HttpResponse> {
    const DEFAULT_LENGTH: u16 = 20;
    let length = nonzero_page_length(params.length.unwrap_or(DEFAULT_LENGTH), accept)?;
    let conn = state
        .db_pool
        .external_api_conn()
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    let join_set = params
        .join_set
        .map(parse_join_set_filter)
        .transpose()
        .map_err(|err| HttpResponse::bad_request(accept, err))?;
    let pagination = match params.direction {
        PaginationDirectionSortedFromOldest::Older => Pagination::OlderThan {
            length,
            cursor: params.cursor.unwrap_or(u32::MAX),
            including_cursor: params.including_cursor,
        },
        PaginationDirectionSortedFromOldest::Newer => Pagination::NewerThan {
            length,
            cursor: params.cursor.unwrap_or(0),
            including_cursor: params.including_cursor,
        },
    };
    let result = conn
        .list_responses_filtered(&execution_id, pagination, join_set.as_ref())
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;

    Ok(match accept {
        AcceptHeader::Json => pretty_json_response(
            StatusCode::OK,
            &ExecutionResponsesResponse {
                responses: result.responses,
                max_cursor: result.max_cursor,
                scan_cursor: result.scan_cursor,
            },
        ),
        AcceptHeader::Text => {
            let mut output = String::new();
            for response in result.responses {
                writeln!(
                    &mut output,
                    "{cursor} `{created_at}` {join_set_id} {resp}",
                    cursor = response.cursor,
                    created_at = response.event.created_at,
                    join_set_id = response.event.event.join_set_id,
                    resp = response.event.event.event,
                )
                .expect("writing to string");
            }
            deprecated_text_response(output)
        }
    })
}

/// Get execution status
#[utoipa::path(
    get,
    path = "/v1/executions/{execution_id}/status",
    tag = "executions",
    params(
        ("execution_id" = String, Path, description = "Execution ID")
    ),
    responses(
        (status = 200, description = "Execution status", body = ExecutionWithStateSer)
    )
)]
#[instrument(skip_all, fields(execution_id))]
async fn execution_status_get(
    Path(execution_id): Path<ExecutionId>,
    state: State<Arc<WebApiState>>,
    accept: AcceptHeader,
) -> Result<Response, HttpResponse> {
    let execution_with_state = state
        .db_pool
        .external_api_conn()
        .await
        .map_err(|e| ErrorWrapper(e, accept))?
        .get_pending_state(&execution_id)
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    Ok(match accept {
        AcceptHeader::Json => pretty_json_response(
            StatusCode::OK,
            &ExecutionWithStateSer::from(execution_with_state),
        ),
        AcceptHeader::Text => deprecated_text_response(format_execution_status_text(
            &execution_with_state.pending_state,
        )),
    })
}

pub(crate) fn format_execution_status_text(pending_state: &PendingState) -> String {
    match pending_state {
        PendingState::Locked(_) => "Locked".to_string(),
        PendingState::PendingAt(pending) => format!("Pending at {}", pending.scheduled_at),
        PendingState::BlockedByJoinSet(blocked) => format!(
            "Blocked by {}{}",
            blocked.join_set_id,
            if blocked.closing { " (closing)" } else { "" }
        ),
        PendingState::Paused(_) => "Paused".to_string(),
        PendingState::Cancelling(_) => "Cancelling".to_string(),
        PendingState::Finished(finished) => match finished.result_kind {
            PendingStateFinishedResultKind::Ok => "Finished: OK".to_string(),
            PendingStateFinishedResultKind::Err(PendingStateFinishedError::Error) => {
                "Finished: Error".to_string()
            }
            PendingStateFinishedResultKind::Err(PendingStateFinishedError::ExecutionFailure(
                kind,
            )) => format!("Finished: Execution failure ({kind})"),
        },
    }
}

/// Payload for stubbing an execution result
#[derive(Deserialize, ToSchema)]
struct ExecutionStubPayload(
    /// The return value to stub
    serde_json::Value,
);

/// Stub an execution result (for testing/debugging)
#[utoipa::path(
    put,
    path = "/v1/executions/{execution_id}/stub",
    tag = "executions",
    params(
        ("execution_id" = String, Path, description = "Derived execution ID to stub")
    ),
    request_body = ExecutionStubPayload,
    responses(
        (status = 200, description = "Execution stubbed"),
        (status = 422, description = "Invalid stub")
    )
)]
#[instrument(skip_all, fields(execution_id))]
async fn execution_stub(
    Path(execution_id): Path<ExecutionIdDerived>,
    state: State<Arc<WebApiState>>,
    Json(ExecutionStubPayload(return_value)): Json<ExecutionStubPayload>,
) -> Result<Response, HttpResponse> {
    let accept = AcceptHeader::Json;
    let (parent_execution_id, join_set_id) = execution_id.split_to_parts();
    let component_registry_ro = {
        let ctx = state.deployment_ctx.read().await;
        ctx.component_registry_ro.clone()
    };
    // Get FFQN
    let db_connection = state
        .db_pool
        .external_api_conn()
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    let ffqn = db_connection
        .get_create_request(&ExecutionId::Derived(execution_id.clone()))
        .await
        .map_err(|err| ErrorWrapper(err, accept))?
        .ffqn;

    // Check that ffqn exists
    let Some((_component_id, fn_metadata)) =
        component_registry_ro.find_by_exported_ffqn_stub(&ffqn)
    else {
        return Err(HttpResponse {
            status: StatusCode::NOT_FOUND,
            message: "function not found".to_string(),
            accept,
        });
    };
    let created_at = Now.now();

    // Type check `return_value`
    let return_value = {
        let type_wrapper = fn_metadata.return_type.type_wrapper();
        let return_value = match deserialize_value(&return_value, type_wrapper) {
            Ok(wast_val_with_type) => wast_val_with_type,
            Err(err) => {
                return Err(HttpResponse {
                    status: StatusCode::UNPROCESSABLE_ENTITY,
                    message: format!(
                        "cannot deserialize return value according to its type - {err}"
                    ),
                    accept,
                });
            }
        };
        SupportedFunctionReturnValue::from_wast_val_with_type(return_value)
            .expect("checked that ffqn is no-ext, return type must be Compatible")
    };
    storage::stub_execution(
        db_connection.as_ref(),
        execution_id,
        parent_execution_id,
        join_set_id,
        created_at,
        return_value,
    )
    .await
    .map_err(|err| ErrorWrapper(err, accept))?;

    Ok(HttpResponse {
        status: StatusCode::OK,
        message: "stubbed".to_string(),
        accept,
    }
    .into_response())
}

/// Lossy representation of `SupportedFunctionReturnValue`. Does not contain WIT type.
#[derive(Debug, Serialize, ToSchema)]
#[serde(rename_all = "snake_case")]
enum RetVal {
    /// Successful result value
    #[schema(value_type = Option<Object>)]
    Ok(Option<WastVal>),
    /// Error result (WIT result's error variant)
    #[schema(value_type = Option<Object>)]
    Err(Option<WastVal>),
    /// Execution failed
    #[schema(value_type = Object)]
    ExecutionFailed(FinishedExecutionFailure),
}
impl From<SupportedFunctionReturnValue> for RetVal {
    fn from(value: SupportedFunctionReturnValue) -> RetVal {
        match value {
            SupportedFunctionReturnValue::Ok(val_with_type) => {
                RetVal::Ok(val_with_type.map(|it| it.value))
            }
            SupportedFunctionReturnValue::Err(val_with_type) => {
                RetVal::Err(val_with_type.map(|it| it.value))
            }
            SupportedFunctionReturnValue::ExecutionFailure(err) => RetVal::ExecutionFailed(err),
        }
    }
}

/// A serializable mirror of `CreateRequest` from concepts.
#[derive(Debug, Serialize, Deserialize, ToSchema)]
pub(crate) struct CreateRequestSer {
    pub(crate) created_at: DateTime<Utc>,
    pub(crate) execution_id: String,
    pub(crate) ffqn: String,
    #[schema(value_type = Vec<Object>)]
    pub(crate) params: concepts::Params,
    pub(crate) parent_execution_id: Option<String>,
    pub(crate) parent_join_set_id: Option<String>,
    pub(crate) scheduled_at: DateTime<Utc>,
    #[schema(value_type = Object)]
    pub(crate) component_id: concepts::ComponentId,
    pub(crate) deployment_id: String,
    #[schema(value_type = Object)]
    pub(crate) metadata: concepts::ExecutionMetadata,
    pub(crate) scheduled_by: Option<String>,
    pub(crate) paused: bool,
}

impl From<concepts::storage::CreateRequest> for CreateRequestSer {
    fn from(r: concepts::storage::CreateRequest) -> Self {
        let (parent_execution_id, parent_join_set_id) = r
            .parent
            .map(|(execution_id, join_set_id)| {
                (
                    Some(execution_id.to_string()),
                    Some(join_set_id.to_string()),
                )
            })
            .unwrap_or((None, None));
        Self {
            created_at: r.created_at,
            execution_id: r.execution_id.to_string(),
            ffqn: r.ffqn.to_string(),
            params: r.params,
            parent_execution_id,
            parent_join_set_id,
            scheduled_at: r.scheduled_at,
            component_id: r.component_id,
            deployment_id: r.deployment_id.to_string(),
            metadata: r.metadata,
            scheduled_by: r.scheduled_by.map(|execution_id| execution_id.to_string()),
            paused: r.paused,
        }
    }
}

impl TryFrom<CreateRequestSer> for concepts::storage::CreateRequest {
    type Error = String;

    fn try_from(value: CreateRequestSer) -> Result<Self, Self::Error> {
        let parent = match (value.parent_execution_id, value.parent_join_set_id) {
            (Some(execution_id), Some(join_set_id)) => Some((
                execution_id
                    .parse()
                    .map_err(|err| format!("invalid parent_execution_id - {err}"))?,
                join_set_id
                    .parse::<JoinSetId>()
                    .map_err(|err| format!("invalid parent_join_set_id - {err}"))?,
            )),
            (None, None) => None,
            (Some(_), None) | (None, Some(_)) => {
                return Err(
                    "parent_execution_id and parent_join_set_id must be both set or both omitted"
                        .to_string(),
                );
            }
        };

        Ok(Self {
            created_at: value.created_at,
            execution_id: value
                .execution_id
                .parse()
                .map_err(|err| format!("invalid execution_id - {err}"))?,
            ffqn: value
                .ffqn
                .parse()
                .map_err(|err| format!("invalid ffqn - {err}"))?,
            params: value.params,
            parent,
            scheduled_at: value.scheduled_at,
            component_id: value.component_id,
            deployment_id: value
                .deployment_id
                .parse()
                .map_err(|err| format!("invalid deployment_id - {err}"))?,
            metadata: value.metadata,
            scheduled_by: value
                .scheduled_by
                .map(|execution_id| {
                    execution_id
                        .parse()
                        .map_err(|err| format!("invalid scheduled_by - {err}"))
                })
                .transpose()?,
            paused: value.paused,
        })
    }
}

/// A serializable mirror of `AppendResponseToExecution` from concepts.
#[derive(Debug, Serialize, Deserialize, ToSchema)]
pub(crate) struct StubResponseSer {
    pub(crate) parent_execution_id: String,
    pub(crate) created_at: DateTime<Utc>,
    pub(crate) join_set_id: String,
    pub(crate) child_execution_id: String,
    pub(crate) finished_version: u32,
    #[schema(value_type = Object)]
    pub(crate) result: SupportedFunctionReturnValue,
}

impl From<concepts::storage::AppendResponseToExecution> for StubResponseSer {
    fn from(r: concepts::storage::AppendResponseToExecution) -> Self {
        Self {
            parent_execution_id: r.parent_execution_id.to_string(),
            created_at: r.created_at,
            join_set_id: r.join_set_id.to_string(),
            child_execution_id: r.child_execution_id.to_string(),
            finished_version: r.finished_version.0,
            result: r.result,
        }
    }
}

impl TryFrom<StubResponseSer> for concepts::storage::AppendResponseToExecution {
    type Error = String;

    fn try_from(value: StubResponseSer) -> Result<Self, Self::Error> {
        let child_execution_id = value
            .child_execution_id
            .parse::<ExecutionId>()
            .map_err(|err| format!("invalid child_execution_id - {err}"))?;
        let ExecutionId::Derived(child_execution_id) = child_execution_id else {
            return Err("child_execution_id must be a derived execution id".to_string());
        };

        Ok(Self {
            parent_execution_id: value
                .parent_execution_id
                .parse()
                .map_err(|err| format!("invalid parent_execution_id - {err}"))?,
            created_at: value.created_at,
            join_set_id: value
                .join_set_id
                .parse()
                .map_err(|err| format!("invalid join_set_id - {err}"))?,
            child_execution_id,
            finished_version: Version::new(value.finished_version),
            result: value.result,
        })
    }
}

/// A write captured during replay, produced by the replay endpoint and applied by the advance endpoint.
#[derive(Debug, Serialize, Deserialize, ToSchema)]
#[serde(tag = "type", rename_all = "snake_case")]
pub(crate) enum CapturedWriteSer {
    Append {
        execution_id: String,
        version: u32,
        #[schema(value_type = Object)]
        event: concepts::storage::AppendRequest,
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        backtraces: Vec<backtrace::BacktraceInfoSer>,
    },
    AppendBatch {
        #[schema(value_type = Vec<Object>)]
        events: Vec<concepts::storage::AppendRequest>,
        execution_id: String,
        version: u32,
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        backtraces: Vec<backtrace::BacktraceInfoSer>,
    },
    AppendBatchCreateNewExecution {
        #[schema(value_type = Vec<Object>)]
        events: Vec<concepts::storage::AppendRequest>,
        execution_id: String,
        version: u32,
        child_requests: Vec<CreateRequestSer>,
        backtraces: Vec<backtrace::BacktraceInfoSer>,
    },
    AppendBatchWithDelayResponse {
        #[schema(value_type = Vec<Object>)]
        events: Vec<concepts::storage::AppendRequest>,
        execution_id: String,
        version: u32,
        join_set_id: String,
        delay_id: String,
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        backtraces: Vec<backtrace::BacktraceInfoSer>,
    },
    AppendStubResponse {
        execution_id: String,
        version: u32,
        #[schema(value_type = Vec<Object>)]
        events: Vec<concepts::storage::AppendRequest>,
        response: StubResponseSer,
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        backtraces: Vec<backtrace::BacktraceInfoSer>,
    },
    AppendFinished {
        execution_id: String,
        version: u32,
        #[schema(value_type = Object)]
        retval: SupportedFunctionReturnValue,
        #[serde(skip_serializing_if = "Option::is_none")]
        parent_execution_id: Option<String>,
        #[serde(skip_serializing_if = "Option::is_none")]
        parent_join_set_id: Option<String>,
    },
}

impl From<concepts::storage::CapturedDbWrite> for CapturedWriteSer {
    fn from(w: concepts::storage::CapturedDbWrite) -> Self {
        use concepts::storage::CapturedDbWrite;
        match w {
            CapturedDbWrite::Append {
                execution_id,
                version,
                req,
                backtraces,
            } => CapturedWriteSer::Append {
                execution_id: execution_id.to_string(),
                version: version.0,
                event: req,
                backtraces: backtraces
                    .into_iter()
                    .map(backtrace::BacktraceInfoSer::from)
                    .collect(),
            },
            CapturedDbWrite::AppendBatch {
                current_time: _,
                batch,
                execution_id,
                version,
                backtraces,
            } => CapturedWriteSer::AppendBatch {
                events: batch,
                execution_id: execution_id.to_string(),
                version: version.0,
                backtraces: backtraces
                    .into_iter()
                    .map(backtrace::BacktraceInfoSer::from)
                    .collect(),
            },
            CapturedDbWrite::AppendBatchCreateNewExecution {
                current_time: _,
                batch,
                execution_id,
                version,
                child_req,
                backtraces,
            } => CapturedWriteSer::AppendBatchCreateNewExecution {
                events: batch,
                execution_id: execution_id.to_string(),
                version: version.0,
                child_requests: child_req.into_iter().map(CreateRequestSer::from).collect(),
                backtraces: backtraces
                    .into_iter()
                    .map(backtrace::BacktraceInfoSer::from)
                    .collect(),
            },
            CapturedDbWrite::AppendBatchWithDelayResponse {
                current_time: _,
                batch,
                execution_id,
                version,
                join_set_id,
                delay_id,
                backtraces,
            } => CapturedWriteSer::AppendBatchWithDelayResponse {
                events: batch,
                execution_id: execution_id.to_string(),
                version: version.0,
                join_set_id: join_set_id.to_string(),
                delay_id: delay_id.to_string(),
                backtraces: backtraces
                    .into_iter()
                    .map(backtrace::BacktraceInfoSer::from)
                    .collect(),
            },
            CapturedDbWrite::AppendStubResponse {
                events,
                response,
                current_time: _,
                backtraces,
            } => CapturedWriteSer::AppendStubResponse {
                execution_id: events.execution_id.to_string(),
                version: events.version.0,
                events: events.batch,
                response: StubResponseSer::from(response),
                backtraces: backtraces
                    .into_iter()
                    .map(backtrace::BacktraceInfoSer::from)
                    .collect(),
            },
            CapturedDbWrite::AppendFinished {
                execution_id,
                version,
                retval,
                current_time: _,
                parent,
            } => CapturedWriteSer::AppendFinished {
                execution_id: execution_id.to_string(),
                version: version.0,
                retval,
                parent_execution_id: parent.as_ref().map(|(id, _)| id.to_string()),
                parent_join_set_id: parent.map(|(_, js)| js.to_string()),
            },
        }
    }
}

impl TryFrom<CapturedWriteSer> for concepts::storage::CapturedDbWrite {
    type Error = String;

    fn try_from(value: CapturedWriteSer) -> Result<Self, Self::Error> {
        match value {
            CapturedWriteSer::Append {
                execution_id,
                version,
                event,
                backtraces,
            } => Ok(Self::Append {
                execution_id: execution_id
                    .parse()
                    .map_err(|err| format!("invalid execution_id - {err}"))?,
                version: Version::new(version),
                req: event,
                backtraces: backtraces
                    .into_iter()
                    .map(concepts::storage::BacktraceInfo::try_from)
                    .collect::<Result<Vec<_>, _>>()?,
            }),
            CapturedWriteSer::AppendBatch {
                events,
                execution_id,
                version,
                backtraces,
            } => Ok(Self::AppendBatch {
                current_time: DateTime::UNIX_EPOCH, // will be replaced in `advance`
                batch: events,
                execution_id: execution_id
                    .parse()
                    .map_err(|err| format!("invalid execution_id - {err}"))?,
                version: Version::new(version),
                backtraces: backtraces
                    .into_iter()
                    .map(concepts::storage::BacktraceInfo::try_from)
                    .collect::<Result<Vec<_>, _>>()?,
            }),
            CapturedWriteSer::AppendBatchCreateNewExecution {
                events,
                execution_id,
                version,
                child_requests,
                backtraces,
            } => Ok(Self::AppendBatchCreateNewExecution {
                current_time: DateTime::UNIX_EPOCH, // will be replaced in `advance`
                batch: events,
                execution_id: execution_id
                    .parse()
                    .map_err(|err| format!("invalid execution_id - {err}"))?,
                version: Version::new(version),
                child_req: child_requests
                    .into_iter()
                    .map(concepts::storage::CreateRequest::try_from)
                    .collect::<Result<Vec<_>, _>>()?,
                backtraces: backtraces
                    .into_iter()
                    .map(concepts::storage::BacktraceInfo::try_from)
                    .collect::<Result<Vec<_>, _>>()?,
            }),
            CapturedWriteSer::AppendBatchWithDelayResponse {
                events,
                execution_id,
                version,
                join_set_id,
                delay_id,
                backtraces,
            } => Ok(Self::AppendBatchWithDelayResponse {
                current_time: DateTime::UNIX_EPOCH, // will be replaced in `advance`
                batch: events,
                execution_id: execution_id
                    .parse()
                    .map_err(|err| format!("invalid execution_id - {err}"))?,
                version: Version::new(version),
                join_set_id: join_set_id
                    .parse()
                    .map_err(|err| format!("invalid join_set_id - {err}"))?,
                delay_id: delay_id
                    .parse()
                    .map_err(|err| format!("invalid delay_id - {err}"))?,
                backtraces: backtraces
                    .into_iter()
                    .map(concepts::storage::BacktraceInfo::try_from)
                    .collect::<Result<Vec<_>, _>>()?,
            }),
            CapturedWriteSer::AppendStubResponse {
                execution_id,
                version,
                events,
                response,
                backtraces,
            } => Ok(Self::AppendStubResponse {
                events: concepts::storage::AppendEventsToExecution {
                    execution_id: execution_id
                        .parse()
                        .map_err(|err| format!("invalid execution_id - {err}"))?,
                    version: Version::new(version),
                    batch: events,
                },
                response: response.try_into()?,
                current_time: DateTime::UNIX_EPOCH, // will be replaced in `advance`
                backtraces: backtraces
                    .into_iter()
                    .map(concepts::storage::BacktraceInfo::try_from)
                    .collect::<Result<Vec<_>, _>>()?,
            }),
            CapturedWriteSer::AppendFinished {
                execution_id,
                version,
                retval,
                parent_execution_id,
                parent_join_set_id,
            } => {
                let parent = match (parent_execution_id, parent_join_set_id) {
                    (Some(exec_id), Some(js_id)) => Some((
                        exec_id
                            .parse()
                            .map_err(|err| format!("invalid parent_execution_id - {err}"))?,
                        js_id
                            .parse()
                            .map_err(|err| format!("invalid parent_join_set_id - {err}"))?,
                    )),
                    (None, None) => None,
                    _ => {
                        return Err(
                            "parent_execution_id and parent_join_set_id must both be set or both be unset"
                                .to_string(),
                        );
                    }
                };
                Ok(Self::AppendFinished {
                    execution_id: execution_id
                        .parse()
                        .map_err(|err| format!("invalid execution_id - {err}"))?,
                    version: Version::new(version),
                    retval,
                    current_time: DateTime::UNIX_EPOCH, // will be replaced in `advance`
                    parent,
                })
            }
        }
    }
}

#[derive(Debug, Serialize, Deserialize, ToSchema)]
#[serde(tag = "type", rename_all = "snake_case")]
pub(crate) enum ReplayResponseSer {
    Advanceable {
        captured_writes: Vec<CapturedWriteSer>,
    },
    Finished {
        retval: serde_json::Value, // SupportedFunctionReturnValue -> RetVal
    },
    Blocked,
    ReplayFailed {
        error: String,
        captured_writes: Vec<CapturedWriteSer>,
    },
}

impl From<wasm_workers::workflow::workflow_worker::ReplayResponse> for ReplayResponseSer {
    fn from(value: wasm_workers::workflow::workflow_worker::ReplayResponse) -> Self {
        match value {
            wasm_workers::workflow::workflow_worker::ReplayResponse::Advanceable(replay) => {
                Self::Advanceable {
                    captured_writes: replay
                        .captured_writes
                        .into_iter()
                        .map(CapturedWriteSer::from)
                        .collect(),
                }
            }
            wasm_workers::workflow::workflow_worker::ReplayResponse::Finished { result } => {
                Self::Finished {
                    retval: serde_json::to_value(RetVal::from(result))
                        .expect("supported retval must be JSON serializable"),
                }
            }
            wasm_workers::workflow::workflow_worker::ReplayResponse::Blocked => Self::Blocked,
        }
    }
}

#[derive(Debug, Serialize, Deserialize, ToSchema)]
pub(crate) struct AdvanceRequestSer {
    pub(crate) captured_writes: Vec<CapturedWriteSer>,
    /// Capture and persist fresh call-site backtraces while advancing.
    #[serde(default)]
    pub(crate) persist_backtrace: bool,
}

#[derive(Debug, Serialize, ToSchema)]
#[serde(tag = "type", rename_all = "snake_case")]
enum AdvanceResponseSer {
    Finished {
        value: RetVal,
    },
    InProgress {
        #[schema(value_type = Object)]
        pending_state: PendingState,
    },
}

#[derive(Debug, Serialize, ToSchema)]
#[serde(tag = "type", rename_all = "snake_case")]
enum AdvanceErrorSer {
    VersionMismatch { expected: u32 },
    ReplayMismatch,
    Transient(String),
}

#[derive(Debug, Serialize, Deserialize, ToSchema)]
pub(crate) struct PersistBacktracesResponseSer {
    /// Number of newly persisted backtraces (already-persisted ones are skipped).
    pub(crate) persisted_backtrace_count: u32,
}

/// Get execution return value
#[utoipa::path(
    get,
    path = "/v1/executions/{execution_id}",
    tag = "executions",
    params(
        ("execution_id" = String, Path, description = "Execution ID"),
        ExecutionFollowParam
    ),
    responses(
        (status = 200, description = "Execution result", body = RetVal),
        (status = 425, description = "Not finished yet")
    )
)]
#[instrument(skip_all, fields(execution_id))]
async fn execution_get_retval(
    Path(execution_id): Path<ExecutionId>,
    Query(params): Query<ExecutionFollowParam>,
    state: State<Arc<WebApiState>>,
) -> Result<http::Response<Body>, HttpResponse> {
    let last_event = state
        .db_pool
        .external_api_conn()
        .await
        .map_err(|e| ErrorWrapper(e, AcceptHeader::Json))?
        .get_last_execution_event(&execution_id)
        .await
        .map_err(|e| ErrorWrapper(e, AcceptHeader::Json))?;

    if let ExecutionRequest::Finished { retval, .. } = last_event.event {
        let retval = RetVal::from(retval);

        Ok(pretty_json_response(StatusCode::OK, &retval))
    } else if params.follow {
        Ok(stream_execution_response(
            execution_id,
            &state,
            StatusCode::OK,
            state.subscription_interruption,
        ))
    } else {
        Ok(HttpResponse {
            status: StatusCode::TOO_EARLY,
            message: "not finished yet".to_string(),
            accept: AcceptHeader::Json,
        }
        .into_response())
    }
}

/// Payload for submitting a new execution
#[derive(Serialize, Deserialize, Debug, ToSchema)]
pub(crate) struct ExecutionSubmitPayload {
    /// Fully qualified function name to execute
    #[schema(value_type = String, example = "my-pkg:my-ifc/my-fn")]
    pub(crate) ffqn: FunctionFqn,
    /// Function parameters as JSON values
    pub(crate) params: Vec<serde_json::Value>,
    /// If true, create the execution in paused state.
    #[serde(default)]
    pub(crate) paused: bool,
}

#[derive(Deserialize, Debug, IntoParams)]
#[into_params(parameter_in = Query)]
struct ExecutionFollowParam {
    /// If true, stream the result when execution finishes
    #[serde(default)]
    follow: bool,
}

/// Submit an execution with a specific ID
#[utoipa::path(
    put,
    path = "/v1/executions/{execution_id}",
    tag = "executions",
    params(
        ("execution_id" = String, Path, description = "Execution ID"),
        ExecutionFollowParam
    ),
    request_body = ExecutionSubmitPayload,
    responses(
        (status = 200, description = "Execution submitted", body = RetVal),
        (status = 409, description = "Conflict")
    )
)]
async fn execution_submit_put(
    Path(execution_id): Path<ExecutionId>,
    state: State<Arc<WebApiState>>,
    Query(params): Query<ExecutionFollowParam>,
    accept: AcceptHeader,
    Json(payload): Json<ExecutionSubmitPayload>,
) -> Result<http::Response<Body>, HttpResponse> {
    execution_submit(execution_id, state, payload, params.follow, accept).await
}

/// Submit a new execution (auto-generated ID)
#[utoipa::path(
    post,
    path = "/v1/executions",
    tag = "executions",
    params(ExecutionFollowParam),
    request_body = ExecutionSubmitPayload,
    responses(
        (status = 200, description = "Execution submitted", body = RetVal)
    )
)]
async fn execution_submit_post(
    state: State<Arc<WebApiState>>,
    accept: AcceptHeader,
    Query(params): Query<ExecutionFollowParam>,
    Json(payload): Json<ExecutionSubmitPayload>,
) -> Result<http::Response<Body>, HttpResponse> {
    let execution_id = ExecutionId::generate();
    execution_submit(execution_id, state, payload, params.follow, accept).await
}

#[instrument(skip_all, fields(execution_id))]
async fn execution_submit(
    execution_id: ExecutionId,
    state: State<Arc<WebApiState>>,
    payload: ExecutionSubmitPayload,
    follow: bool,
    accept: AcceptHeader,
) -> Result<http::Response<Body>, HttpResponse> {
    let (deployment_id, component_registry_ro) = {
        let ctx = state.deployment_ctx.read().await;
        (ctx.deployment_id, ctx.component_registry_ro.clone())
    };
    let conn = state
        .db_pool
        .external_api_conn()
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    let res = server::submit(
        deployment_id,
        conn.as_ref(),
        execution_id.clone(),
        payload.ffqn,
        payload.params,
        payload.paused,
        &component_registry_ro,
    )
    .await
    .map_err(|err| ErrorWrapper(err, accept))?;
    let status = match res {
        SubmitOutcome::Created => StatusCode::CREATED,
        SubmitOutcome::ExistsWithSameParameters => StatusCode::OK,
    };
    if follow {
        Ok(stream_execution_response(
            execution_id,
            &state,
            status,
            state.subscription_interruption,
        ))
    } else {
        Ok(HttpResponse {
            status,
            message: execution_id.to_string(),
            accept,
        }
        .into_response())
    }
}

/// Wait until the execution finishes, return `RetVal` as JSON.
fn stream_execution_response(
    execution_id: ExecutionId,
    state: &WebApiState,
    status: StatusCode,
    subscription_interruption: Option<Duration>,
) -> http::Response<Body> {
    let (tx, rx) = mpsc::channel::<Result<Bytes, std::io::Error>>(1);
    let trace_id = server::gen_trace_id();
    let span = info_span!("stream_execution_response", trace_id, %execution_id);
    utils::spawn::spawn_named(
        "stream_execution_response",
        stream_execution_response_task(
            execution_id,
            state.db_pool.clone(),
            tx,
            state.termination_watcher.clone(),
            subscription_interruption,
        )
        .instrument(span),
    );
    // Send response headers immediately.
    let stream = ReceiverStream::new(rx);
    let mut response = (status, Body::from_stream(stream)).into_response();
    response.headers_mut().insert(
        header::CONTENT_TYPE,
        header::HeaderValue::from_static("application/json"),
    );
    response
}

async fn stream_execution_response_task(
    execution_id: ExecutionId,
    db_pool: Arc<dyn DbPool>,
    tx: mpsc::Sender<Result<Bytes, std::io::Error>>,
    server_termination_watcher: watch::Receiver<()>,
    subscription_interruption: Option<Duration>,
) {
    debug!("Started streaming execution response");
    let db_connection = match db_pool.connection().await {
        Ok(ok) => ok,
        Err(err) => {
            warn!("Cannot obtain connection - {err:?}");
            return;
        }
    };
    let sleep = concepts::time::TokioSleep;
    let timeout_factory = {
        let tx = tx.clone();
        move || {
            let subscription_interruption = subscription_interruption.unwrap_or(Duration::MAX);
            let sleep = sleep.clone();
            let tx = tx.clone();
            let mut server_termination_watcher = server_termination_watcher.clone();
            Box::pin(async move {
                select! {
                    () = tx.closed() => {
                        debug!("Client disconnected");
                        TimeoutOutcome::Cancel
                    }
                    () = sleep.sleep(subscription_interruption) => TimeoutOutcome::Timeout,
                    _ = server_termination_watcher.changed() => TimeoutOutcome::Cancel,
                }
            })
        }
    };

    loop {
        let timeout = timeout_factory();
        let res = db_connection
            .wait_for_finished_result(&execution_id, Some(timeout))
            .await;
        match res {
            Ok(result) => {
                trace!("Finished ok");
                let result = RetVal::from(result);
                let result = serde_json::to_vec_pretty(&result)
                    .expect("serialization of already stored retval cannot fail");
                let _ = tx.try_send(Ok(Bytes::from(result))); // Ignore if the remote side is closed.
                debug!("Sent execution result");
                return;
            }
            Err(DbErrorReadWithTimeout::Timeout(TimeoutOutcome::Timeout)) => {
                trace!("Timeout triggers resubscribing");
            }
            Err(DbErrorReadWithTimeout::Timeout(TimeoutOutcome::Cancel)) => {
                debug!("Connection closed, not waiting for result");
                return;
            }
            Err(DbErrorReadWithTimeout::DbErrorRead(err)) => {
                warn!("Database error: {err:?}");
                return;
            }
        }
    }
}

/// Replay an execution. An advanceable (unfinished and unblocked) execution produces captured writes for the advance endpoint; backtraces on captured writes are display-only.
#[utoipa::path(
    put,
    path = "/v1/executions/{execution_id}/replay",
    tag = "executions",
    params(
        ("execution_id" = String, Path, description = "Execution ID to replay")
    ),
    responses(
        (status = 200, description = "Execution replayed", body = ReplayResponseSer),
        (status = 404, description = "Not found"),
        (status = StatusCode::CONFLICT, description = "Replay failed", body = ReplayResponseSer)
    )
)]
#[instrument(skip_all, fields(execution_id))]
async fn execution_replay(
    Path(execution_id): Path<ExecutionId>,
    state: State<Arc<WebApiState>>,
    accept: AcceptHeader,
) -> Result<Response, HttpResponse> {
    let ser = replay_execution_internal(&state, &execution_id, accept).await?;
    let status = if matches!(ser, ReplayResponseSer::ReplayFailed { .. }) {
        StatusCode::CONFLICT
    } else {
        StatusCode::OK
    };
    Ok(match accept {
        AcceptHeader::Json => pretty_json_response(status, &ser),
        AcceptHeader::Text => {
            let body = match ser {
                ReplayResponseSer::Advanceable { captured_writes } => {
                    format!("outcome: advanceable, {} writes", captured_writes.len())
                }
                ReplayResponseSer::Finished { retval } => {
                    format!("outcome: finished\nresult: {retval}")
                }
                ReplayResponseSer::Blocked => "outcome: blocked".to_string(),
                ReplayResponseSer::ReplayFailed {
                    error,
                    captured_writes,
                } => {
                    format!(
                        "outcome: replay_failed, error: {error}, {} writes",
                        captured_writes.len()
                    )
                }
            };
            deprecated_text_response((status, body))
        }
    })
}

/// Advance a paused execution by applying captured writes from a prior replay; set `persist_backtrace` to persist fresh backtraces.
#[utoipa::path(
    put,
    path = "/v1/executions/{execution_id}/advance",
    tag = "executions",
    params(
        ("execution_id" = String, Path, description = "Execution ID to advance")
    ),
    request_body = AdvanceRequestSer,
    responses(
        (status = 200, description = "Execution advance outcome", body = AdvanceResponseSer),
        (status = 400, description = "Bad request"),
        (status = 404, description = "Not found"),
        (status = 422, description = "Advance failed", body = AdvanceErrorSer)
    )
)]
#[instrument(skip_all, fields(execution_id))]
async fn execution_advance(
    Path(execution_id): Path<ExecutionId>,
    state: State<Arc<WebApiState>>,
    accept: AcceptHeader,
    Json(payload): Json<AdvanceRequestSer>,
) -> Result<Response, HttpResponse> {
    let replay_worker = get_replay_target(&state, &execution_id, accept).await?;

    let captured_writes = wasm_workers::workflow::workflow_worker::ReplayAdvanceable {
        captured_writes: payload
            .captured_writes
            .into_iter()
            .map(concepts::storage::CapturedDbWrite::try_from)
            .collect::<Result<Vec<_>, _>>()
            .map_err(|message| HttpResponse {
                status: StatusCode::UNPROCESSABLE_ENTITY,
                message,
                accept,
            })?,
    };
    if captured_writes.is_empty() {
        return Err(HttpResponse::bad_request(
            accept,
            "`captured_writes` must not be empty".to_string(),
        ));
    }

    let backtrace_capture = if payload.persist_backtrace {
        BacktraceCapture::NewEventsOnly
    } else {
        BacktraceCapture::Disabled
    };
    let advance_res = replay_worker
        .advance(execution_id.clone(), captured_writes, backtrace_capture)
        .await;

    let advance_response = match advance_res {
        Ok(ok) => ok,
        Err(err) => {
            info!("Advance failed: {err:?}");
            let error = match err {
                AdvanceError::NoWrites => {
                    unreachable!("sent 401")
                }
                AdvanceError::VersionMismatch { expected } => AdvanceErrorSer::VersionMismatch {
                    expected: expected.0,
                },
                AdvanceError::ReplayMismatch => AdvanceErrorSer::ReplayMismatch,
                AdvanceError::DbError(db_err) => {
                    return Err(ErrorWrapper(db_err, accept).into());
                }
                err @ (AdvanceError::ExecutorClosing | AdvanceError::LimitReached { .. }) => {
                    AdvanceErrorSer::Transient(err.to_string())
                }
            };
            let response = match accept {
                AcceptHeader::Json => {
                    pretty_json_response(StatusCode::UNPROCESSABLE_ENTITY, &error)
                }
                AcceptHeader::Text => {
                    let text = match error {
                        AdvanceErrorSer::VersionMismatch { expected } => {
                            format!("error: version_mismatch\nexpected: {expected}")
                        }
                        AdvanceErrorSer::ReplayMismatch => "error: replay_mismatch".to_string(),
                        AdvanceErrorSer::Transient(err) => format!("transient error: {err}"),
                    };
                    deprecated_text_response((StatusCode::UNPROCESSABLE_ENTITY, text))
                }
            };
            return Ok(response);
        }
    };

    let response = if let Some(finished) = advance_response.finished {
        AdvanceResponseSer::Finished {
            value: RetVal::from(finished),
        }
    } else {
        let execution_with_state = state
            .db_pool
            .external_api_conn()
            .await
            .map_err(|e| ErrorWrapper(e, accept))?
            .get_pending_state(&execution_id)
            .await
            .map_err(|e| ErrorWrapper(e, accept))?;

        AdvanceResponseSer::InProgress {
            pending_state: execution_with_state.pending_state,
        }
    };
    match &response {
        AdvanceResponseSer::Finished { value } => Ok(match accept {
            AcceptHeader::Json => pretty_json_response(StatusCode::OK, &response),
            AcceptHeader::Text => deprecated_text_response(format!(
                "success:\n{}",
                serde_json::to_string_pretty(&value).expect("retval must be JSON serializable")
            )),
        }),
        AdvanceResponseSer::InProgress { pending_state } => Ok(match accept {
            AcceptHeader::Json => pretty_json_response(StatusCode::OK, &response),
            AcceptHeader::Text => {
                deprecated_text_response(format!("success, current state: {pending_state}"))
            }
        }),
    }
}

/// Replay an execution and persist its call-site backtraces (idempotent).
#[utoipa::path(
    put,
    path = "/v1/executions/{execution_id}/backtrace/persist",
    tag = "executions",
    params(
        ("execution_id" = String, Path, description = "Execution ID whose backtraces to persist")
    ),
    responses(
        (status = 200, description = "Backtraces persisted", body = PersistBacktracesResponseSer),
        (status = 404, description = "Not found"),
        (status = 422, description = "Replay failed")
    )
)]
#[instrument(skip_all, fields(execution_id))]
async fn execution_persist_backtraces(
    Path(execution_id): Path<ExecutionId>,
    state: State<Arc<WebApiState>>,
    accept: AcceptHeader,
) -> Result<Response, HttpResponse> {
    let replay_worker = get_replay_target(&state, &execution_id, accept).await?;
    let persisted_backtrace_count = replay_worker
        .persist_backtraces(execution_id.clone())
        .await
        .map_err(|err| HttpResponse {
            status: StatusCode::UNPROCESSABLE_ENTITY,
            message: format!("Replay error: {err}"),
            accept,
        })?;
    let persisted_backtrace_count =
        u32::try_from(persisted_backtrace_count).map_err(|_| HttpResponse {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            message: "too many backtraces persisted".to_string(),
            accept,
        })?;
    let ser = PersistBacktracesResponseSer {
        persisted_backtrace_count,
    };
    Ok(match accept {
        AcceptHeader::Json => pretty_json_response(StatusCode::OK, &ser),
        AcceptHeader::Text => {
            deprecated_text_response(format!("persisted {persisted_backtrace_count} backtraces"))
        }
    })
}

async fn get_replay_target(
    state: &Arc<WebApiState>,
    execution_id: &ExecutionId,
    accept: AcceptHeader,
) -> Result<ReplayWorker, HttpResponse> {
    let (component_registry_ro, replay_workers) = {
        let ctx = state.deployment_ctx.read().await;
        (
            ctx.component_registry_ro.clone(),
            ctx.replay_workers.clone(),
        )
    };
    let conn = state
        .db_pool
        .connection()
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    let create_req = conn.get_create_request(execution_id).await.map_err(|e| {
        if e == DbErrorRead::NotFound {
            HttpResponse::not_found(accept, "execution")
        } else {
            ErrorWrapper(e, accept).into()
        }
    })?;
    let Some((component_id, fn_metadata)) =
        component_registry_ro.find_by_exported_ffqn_submittable(&create_req.ffqn)
    else {
        return Err(HttpResponse::not_found(accept, "component"));
    };
    if fn_metadata.extension.is_some() {
        return Err(HttpResponse::bad_request(
            accept,
            "function must not be an extension".to_string(),
        ));
    }
    Span::current().record("component_id", tracing::field::display(&component_id));
    let (_component_id, replay_worker) = replay_workers
        .get(&component_id.component_digest)
        .ok_or_else(|| HttpResponse::not_found(accept, "replay worker"))?;
    Ok(replay_worker.clone())
}

async fn replay_execution_internal(
    state: &Arc<WebApiState>,
    execution_id: &ExecutionId,
    accept: AcceptHeader,
) -> Result<ReplayResponseSer, HttpResponse> {
    let replay_worker = get_replay_target(state, execution_id, accept).await?;

    let map_replay_err =
        |err: wasm_workers::workflow::workflow_worker::ReplayError| -> Result<ReplayResponseSer, HttpResponse> {
            use wasm_workers::workflow::workflow_worker::ReplayError;
            match err {
                ReplayError::ReplayFailed {
                    err,
                    captured_writes,
                } => Ok(ReplayResponseSer::ReplayFailed {
                    error: err.to_string(),
                    captured_writes: captured_writes
                        .into_iter()
                        .map(CapturedWriteSer::from)
                        .collect(),
                }),
                other => Err(HttpResponse {
                    status: StatusCode::UNPROCESSABLE_ENTITY,
                    message: format!("Replay error: {other}"),
                    accept,
                }),
            }
        };

    // Capture backtraces for the newly produced writes only; the already-persisted prefix
    // does not need them (and re-deriving them for a long log is expensive).
    let replay_res = replay_worker
        .replay(execution_id.clone(), BacktraceCapture::NewEventsOnly)
        .await;
    match replay_res {
        Ok(response) => Ok(ReplayResponseSer::from(response)),
        Err(err) => map_replay_err(err),
    }
}

/// Payload for upgrading an execution to a new component version
#[derive(Serialize, Deserialize, ToSchema)]
pub(crate) struct ExecutionUpgradePayload {
    /// Old component digest to upgrade from
    #[schema(value_type = String)]
    pub(crate) old: ComponentDigest,
    /// New component digest to upgrade to
    #[schema(value_type = String)]
    pub(crate) new: ComponentDigest,
    /// Skip determinism check during upgrade
    #[serde(default)]
    pub(crate) skip_determinism_check: bool,
}

/// Upgrade an execution to a new component version
#[utoipa::path(
    put,
    path = "/v1/executions/{execution_id}/upgrade",
    tag = "executions",
    params(
        ("execution_id" = String, Path, description = "Execution ID to upgrade")
    ),
    request_body = ExecutionUpgradePayload,
    responses(
        (status = 200, description = "Execution upgraded"),
        (status = 404, description = "Not found"),
        (status = 422, description = "Upgrade failed")
    )
)]
#[instrument(skip_all, fields(execution_id))]
async fn execution_upgrade(
    Path(execution_id): Path<ExecutionId>,
    state: State<Arc<WebApiState>>,
    accept: AcceptHeader,
    Json(payload): Json<ExecutionUpgradePayload>,
) -> Result<Response, HttpResponse> {
    if !payload.skip_determinism_check {
        let replay_workers = {
            let ctx = state.deployment_ctx.read().await;
            ctx.replay_workers.clone()
        };
        let (_component_id, replay_worker) = replay_workers
            .get(&payload.new)
            .ok_or_else(|| HttpResponse::not_found(accept, Some("new component")))?;
        // no backtrace capture on upgrade
        let replay_res = replay_worker
            .replay(execution_id.clone(), BacktraceCapture::Disabled)
            .await;
        if let Err(err) = replay_res {
            info!("Replay failed: {err:?}");
            return Err(HttpResponse {
                status: StatusCode::UNPROCESSABLE_ENTITY,
                message: format!("Replay failed: {err}"),
                accept,
            });
        }
    }
    state
        .db_pool
        .external_api_conn()
        .await
        .map_err(|e| ErrorWrapper(e, accept))?
        .upgrade_execution_component(
            &execution_id,
            &payload.old,
            &payload.new,
            concepts::storage::ComponentUpgradeReason::Manual {
                force: payload.skip_determinism_check,
            },
        )
        .await
        .map_err(|e| ErrorWrapper(e, accept))?;
    Ok(HttpResponse {
        status: StatusCode::OK,
        message: "upgraded".to_string(),
        accept,
    }
    .into_response())
}

pub(crate) mod components {
    use crate::server::web_api_server::{
        HttpResponse, deprecated_text_response, pretty_json_response,
    };

    use super::{
        AcceptHeader, Arc, Deserialize, FunctionFqn, IntoParams, IntoResponse, Query, Response,
        Serialize, State, StatusCode, ToSchema, WebApiState,
    };
    use axum::extract::Path;
    use concepts::{
        ComponentId, ComponentType, FunctionExtension, FunctionMetadata, ParameterType,
        component_id::ComponentDigest,
        prefixed_ulid::DeploymentId,
        storage::{
            ComponentFileRole, DeploymentComponentDetail, DeploymentComponentFileDetail,
            PersistedFunctionMetadata, PersistedParameterType,
        },
    };
    use itertools::Itertools;
    use std::fmt::{Debug, Write as _};

    #[derive(Deserialize, Debug, IntoParams)]
    #[into_params(parameter_in = Query)]
    pub(crate) struct ComponentsListParams {
        /// Filter by deployment ID
        #[param(value_type = Option<String>)]
        deployment_id: Option<DeploymentId>,
        /// Filter by component type (workflow, activity, webhook)
        #[param(value_type = Option<String>)]
        r#type: Option<ComponentType>,
        /// Filter by component name
        name: Option<String>,
        /// Filter by component digest
        #[param(value_type = Option<String>)]
        digest: Option<ComponentDigest>,
        /// Filter by function
        #[param(value_type = Option<String>)]
        ffqn: Option<FunctionFqn>,
        /// Include exports in response
        #[serde(default)]
        exports: bool,
        /// Include imports in response
        #[serde(default)]
        imports: bool,
        /// Include extension functions
        #[serde(default)]
        extensions: bool,
        /// If set to true, show only submittable exports
        submittable: Option<bool>,
    }

    #[derive(Debug, serde::Deserialize, utoipa::IntoParams)]
    #[into_params(parameter_in = Query)]
    pub(crate) struct ComponentWitParams {
        /// Filter by deployment ID
        #[param(value_type = Option<String>)]
        deployment_id: Option<DeploymentId>,
    }

    /// Get WIT definition for a component
    #[utoipa::path(
        get,
        path = "/v1/components/{digest}/wit",
        tag = "components",
        params(
            ("digest" = String, Path, description = "Component content digest"),
            ComponentWitParams,
        ),
        responses(
            (status = 200, description = "WIT definition", body = String),
            (status = 204, description = "No WIT available"),
            (status = 404, description = "Component not found")
        )
    )]
    pub(crate) async fn component_wit(
        Path(digest): Path<ComponentDigest>,
        state: State<Arc<WebApiState>>,
        Query(params): Query<ComponentWitParams>,
    ) -> Result<Response, HttpResponse> {
        let deployment_id = if let Some(deployment_id) = params.deployment_id {
            deployment_id
        } else {
            state.deployment_ctx.read().await.deployment_id
        };
        let conn = state
            .db_pool
            .external_api_conn()
            .await
            .map_err(|err| HttpResponse {
                status: StatusCode::INTERNAL_SERVER_ERROR,
                message: err.to_string(),
                accept: AcceptHeader::Text,
            })?;
        let wit = conn
            .get_deployment_component_wit(deployment_id, &digest)
            .await
            .map_err(|err| HttpResponse {
                status: StatusCode::INTERNAL_SERVER_ERROR,
                message: err.to_string(),
                accept: AcceptHeader::Text,
            })?;
        match wit {
            Some(wit) => Ok(wit.into_response()),
            None => Err(HttpResponse::not_found(
                AcceptHeader::Text,
                Some("component"),
            )),
        }
    }

    /// List components
    #[utoipa::path(
        get,
        path = "/v1/components",
        tag = "components",
        params(ComponentsListParams),
        responses(
            (status = 200, description = "List of components", body = Vec<ComponentConfig>)
        )
    )]
    pub(crate) async fn components_list(
        state: State<Arc<WebApiState>>,
        Query(params): Query<ComponentsListParams>,
        accept: AcceptHeader,
    ) -> Response {
        let deployment_id = if let Some(deployment_id) = params.deployment_id {
            deployment_id
        } else {
            state.deployment_ctx.read().await.deployment_id
        };
        let mut components = match state.db_pool.external_api_conn().await {
            Ok(conn) => match conn.list_deployment_components(deployment_id).await {
                Ok(components) => components,
                Err(err) => {
                    return HttpResponse {
                        status: StatusCode::INTERNAL_SERVER_ERROR,
                        message: err.to_string(),
                        accept,
                    }
                    .into_response();
                }
            },
            Err(err) => {
                return HttpResponse {
                    status: StatusCode::INTERNAL_SERVER_ERROR,
                    message: err.to_string(),
                    accept,
                }
                .into_response();
            }
        };

        if let Some(name) = params.name {
            components.retain(|c| c.component_id.name.as_ref() == name);
        }
        if let Some(digest) = params.digest {
            components.retain(|c| c.component_id.component_digest == digest);
        }
        if let Some(ffqn) = params.ffqn {
            components.retain(|c| {
                filtered_exports(c, params.extensions)
                    .iter()
                    .find(|fn_meta| fn_meta.ffqn == ffqn)
                    .is_some()
            });
        }
        if let Some(ty) = params.r#type {
            components.retain(|c| c.component_id.component_type == ty);
        }
        let components: Vec<_> = components
            .into_iter()
            .map(|c| {
                let exports = if params.exports {
                    let mut exports = Vec::new();
                    for export in filtered_exports(&c, params.extensions)
                        .into_iter()
                        .filter(|e| {
                            if let Some(submittable) = params.submittable {
                                e.submittable == submittable
                            } else {
                                true
                            }
                        })
                    {
                        exports.push(FunctionMetadataLite::from(export));
                    }

                    Some(exports)
                } else {
                    None
                };

                ComponentConfig {
                    component_id: c.component_id,
                    files: c.files.into_iter().map(Into::into).collect(),
                    imports: if params.imports {
                        Some(
                            c.imports
                                .into_iter()
                                .map(FunctionMetadataLite::from)
                                .collect(),
                        )
                    } else {
                        None
                    },
                    exports,
                }
            })
            .collect();

        match accept {
            AcceptHeader::Json => pretty_json_response(StatusCode::OK, &components),
            AcceptHeader::Text => {
                let mut output = String::new();
                for component in components {
                    writeln!(
                        output,
                        "{} {}",
                        component.component_id, component.component_id.component_digest
                    )
                    .expect("writing to string");
                    if let Some(fns) = component.exports {
                        writeln!(output, " exports:").expect("writing to string");
                        for func in fns {
                            writeln!(output, "  {func}").expect("writing to string");
                        }
                    }
                    if let Some(fns) = component.imports {
                        writeln!(output, " imports:").expect("writing to string");
                        for func in fns {
                            writeln!(output, "  {func}").expect("writing to string");
                        }
                    }
                    if !component.files.is_empty() {
                        writeln!(output, " files:").expect("writing to string");
                        for entry in component.files {
                            writeln!(
                                output,
                                "  {} {} {} {}",
                                entry.file.path, entry.file.digest, entry.file.size, entry.role
                            )
                            .expect("writing to string");
                        }
                    }
                }
                deprecated_text_response(output)
            }
        }
    }

    /// Component configuration with optional imports/exports
    #[derive(Serialize, Deserialize, ToSchema)]
    pub(crate) struct ComponentConfig {
        /// Component identifier
        #[schema(value_type = Object)]
        pub(crate) component_id: ComponentId,
        /// Deployment-owned files used by this component.
        pub(crate) files: Vec<ComponentFileRef>,
        /// Imported functions
        #[serde(skip_serializing_if = "Option::is_none")]
        pub(crate) imports: Option<Vec<FunctionMetadataLite>>,
        /// Exported functions
        #[serde(skip_serializing_if = "Option::is_none")]
        pub(crate) exports: Option<Vec<FunctionMetadataLite>>,
    }

    #[derive(Serialize, Deserialize, ToSchema)]
    pub(crate) struct ComponentFileRef {
        pub(crate) file: ComponentFile,
        #[schema(value_type = String)]
        pub(crate) role: ComponentFileRole,
    }

    #[derive(Serialize, Deserialize, ToSchema)]
    pub(crate) struct ComponentFile {
        pub(crate) path: String,
        pub(crate) digest: String,
        pub(crate) size: u64,
    }

    impl From<DeploymentComponentFileDetail> for ComponentFileRef {
        fn from(value: DeploymentComponentFileDetail) -> Self {
            Self {
                file: ComponentFile {
                    path: value.file.path,
                    digest: value.file.digest.to_string(),
                    size: value.file.size,
                },
                role: value.role,
            }
        }
    }

    /// Lightweight function metadata
    #[derive(serde::Serialize, serde::Deserialize, derive_more::Display, ToSchema)]
    #[display("{ffqn}: func({}) -> {return_type}", parameter_types.iter().join(", "))]
    pub(crate) struct FunctionMetadataLite {
        /// Fully qualified function name
        #[schema(value_type = String)]
        pub(crate) ffqn: FunctionFqn,
        /// Parameter types
        pub(crate) parameter_types: Vec<ParameterTypeLite>,
        /// Return type as WIT string
        pub(crate) return_type: String,
        /// Function extension type if any
        #[serde(skip_serializing_if = "Option::is_none")]
        #[schema(value_type = Option<String>)]
        pub(crate) extension: Option<FunctionExtension>,
        /// Externally submittable: primary functions + `-schedule` extended, but no activity stubs
        pub(crate) submittable: bool,
    }
    impl From<FunctionMetadata> for FunctionMetadataLite {
        fn from(value: FunctionMetadata) -> Self {
            FunctionMetadataLite {
                ffqn: value.ffqn,
                parameter_types: value
                    .parameter_types
                    .0
                    .into_iter()
                    .map(ParameterTypeLite::from)
                    .collect(),
                return_type: value.return_type.wit_type().to_string(),
                extension: value.extension,
                submittable: value.submittable,
            }
        }
    }
    impl From<PersistedFunctionMetadata> for FunctionMetadataLite {
        fn from(value: PersistedFunctionMetadata) -> Self {
            FunctionMetadataLite {
                ffqn: value.ffqn,
                parameter_types: value
                    .parameter_types
                    .into_iter()
                    .map(ParameterTypeLite::from)
                    .collect(),
                return_type: value.return_type,
                extension: value.extension,
                submittable: value.submittable,
            }
        }
    }

    /// Parameter type information
    #[derive(serde::Serialize, serde::Deserialize, derive_more::Display, ToSchema)]
    #[display("{name}: {wit_type}")]
    pub(crate) struct ParameterTypeLite {
        /// Parameter name
        pub(crate) name: String,
        /// WIT type representation
        pub(crate) wit_type: String,
    }

    impl From<ParameterType> for ParameterTypeLite {
        fn from(value: ParameterType) -> Self {
            ParameterTypeLite {
                name: value.name.to_string(),
                wit_type: value.wit_type.to_string(),
            }
        }
    }
    impl From<PersistedParameterType> for ParameterTypeLite {
        fn from(value: PersistedParameterType) -> Self {
            ParameterTypeLite {
                name: value.name,
                wit_type: value.wit_type,
            }
        }
    }

    fn filtered_exports(
        component: &DeploymentComponentDetail,
        extensions: bool,
    ) -> Vec<PersistedFunctionMetadata> {
        let mut exports = component.exports.clone();
        if !extensions {
            exports.retain(|fn_metadata| !fn_metadata.ffqn.ifc_fqn.is_extension());
        }
        exports
    }
}

mod functions {
    use super::{
        AcceptHeader, Arc, Deserialize, IntoParams, IntoResponse, Query, Response, State,
        StatusCode, ToSchema, WebApiState, deprecated_text_response, pretty_json_response,
    };
    use concepts::{FunctionExtension, FunctionFqn, FunctionRegistry};
    use std::fmt::Write as _;

    #[derive(Deserialize, Debug, IntoParams)]
    #[into_params(parameter_in = Query)]
    pub(crate) struct FunctionsListParams {
        /// Include extension functions
        #[serde(default)]
        extensions: bool,
    }

    /// List all functions
    #[utoipa::path(
        get,
        path = "/v1/functions",
        tag = "functions",
        params(FunctionsListParams),
        responses(
            (status = 200, description = "List of functions", body = Vec<FunctionOutput>)
        )
    )]
    pub(crate) async fn functions_list(
        state: State<Arc<WebApiState>>,
        Query(params): Query<FunctionsListParams>,
        accept: AcceptHeader,
    ) -> Response {
        let component_registry_ro = state
            .deployment_ctx
            .read()
            .await
            .component_registry_ro
            .clone();
        let all_exports = component_registry_ro.all_exports();

        let functions: Vec<FunctionOutput> = all_exports
            .iter()
            .filter(|pkg_ifc| params.extensions || !pkg_ifc.extension)
            .flat_map(|pkg_ifc| pkg_ifc.fns.values())
            .map(|fn_metadata| FunctionOutput {
                ffqn: fn_metadata.ffqn.clone(),
                extension: fn_metadata.extension,
            })
            .collect();

        match accept {
            AcceptHeader::Json => pretty_json_response(StatusCode::OK, &functions),
            AcceptHeader::Text => {
                let mut output = String::new();
                for func in functions {
                    writeln!(output, "{}", func.ffqn).expect("writing to string");
                }
                deprecated_text_response(output)
            }
        }
    }

    /// Function output information
    #[derive(serde::Serialize, ToSchema)]
    pub(crate) struct FunctionOutput {
        /// Fully qualified function name
        #[schema(value_type = String)]
        ffqn: FunctionFqn,
        /// Extension type if any
        #[serde(skip_serializing_if = "Option::is_none")]
        #[schema(value_type = Option<String>)]
        extension: Option<FunctionExtension>,
    }

    use super::HttpResponse;

    #[derive(Deserialize, Debug, IntoParams)]
    #[into_params(parameter_in = Query)]
    pub(crate) struct FunctionWitParams {
        /// Fully qualified function name
        #[param(value_type = String)]
        ffqn: FunctionFqn,
    }

    /// Get WIT definition for a function
    #[utoipa::path(
        get,
        path = "/v1/functions/wit",
        tag = "functions",
        params(FunctionWitParams),
        responses(
            (status = 200, description = "WIT definition", body = String),
            (status = 404, description = "Function not found")
        )
    )]
    pub(crate) async fn function_wit(
        Query(params): Query<FunctionWitParams>,
        state: State<Arc<WebApiState>>,
    ) -> Result<Response, HttpResponse> {
        let ffqn = params.ffqn;
        let component_registry_ro = state
            .deployment_ctx
            .read()
            .await
            .component_registry_ro
            .clone();
        // Find the component that exports this function
        let Some((component_id, _fn_metadata)) = component_registry_ro.find_by_exported_ffqn(&ffqn)
        else {
            return Err(HttpResponse::not_found(
                AcceptHeader::Text,
                Some("function"),
            ));
        };

        // Get the WIT for this component
        let wit = component_registry_ro
            .get_wit(&component_id.component_digest)
            .expect("if function is found, component must be found");

        // Print just the interface with the single function
        match crate::wit_printer::print_interface_with_single_fn(wit, &ffqn) {
            Ok(output) => Ok(output.into_response()),
            Err(e) => Err(HttpResponse {
                status: http::StatusCode::INTERNAL_SERVER_ERROR,
                message: format!("failed to print WIT: {e}"),
                accept: AcceptHeader::Text,
            }),
        }
    }
}

pub(crate) mod deployment {
    use crate::{
        command::server::{self, RuntimeConfigAvailability, SwitchDeploymentAction},
        config::deployment::strip_generated_deployment_metadata,
        server::{
            deployment_summary,
            web_api_server::{
                AcceptHeader, ErrorWrapper, HttpResponse, TextDefaultAcceptHeader, WebApiState,
                deprecated_text_response, nonzero_page_length, pretty_json_response,
            },
        },
    };
    use http::header;

    /// Map the REST `allow_unavailable_runtime_config` flag to the runtime config policy.
    fn runtime_config_availability_from_bool(allow_unavailable: bool) -> RuntimeConfigAvailability {
        if allow_unavailable {
            RuntimeConfigAvailability::AllowUnavailable
        } else {
            RuntimeConfigAvailability::Strict
        }
    }
    use axum::{
        Json,
        extract::{Path, Query, State},
        response::{IntoResponse, Response},
    };
    use chrono::{DateTime, Utc};
    use concepts::prefixed_ulid::DeploymentId;
    use concepts::storage::Pagination;
    use concepts::storage::{
        DeploymentExecutionCounts, DeploymentFileRecord, DeploymentRecord, DeploymentState,
        DeploymentStatus, LIST_DEPLOYMENT_STATES_DEFAULT_LENGTH,
    };
    use http::StatusCode;
    use serde::{Deserialize, Serialize};
    use std::fmt::Write as _;
    use std::sync::Arc;
    use tracing::{info, instrument};
    use utoipa::{IntoParams, ToSchema};

    #[derive(Debug, Serialize, Deserialize, ToSchema)]
    #[serde(rename_all = "snake_case")]
    pub enum DeploymentStatusSer {
        Inactive,
        Enqueued,
        Active,
    }

    impl From<&DeploymentStatus> for DeploymentStatusSer {
        fn from(s: &DeploymentStatus) -> Self {
            match s {
                DeploymentStatus::Inactive => Self::Inactive,
                DeploymentStatus::Enqueued => Self::Enqueued,
                DeploymentStatus::Active => Self::Active,
            }
        }
    }
    /// Deployment state with execution counts
    #[derive(Debug, Serialize, Deserialize, ToSchema)]
    pub struct DeploymentStateSer {
        /// Deployment identifier
        #[schema(value_type = String, example = "Dep_01JKXYZ123456789ABCDEFGHIJ")]
        pub deployment_id: DeploymentId,
        /// Optional human-readable deployment description
        pub description: Option<String>,
        /// Content digest = `sha256(deployment_toml)`
        pub digest: String,
        /// Deployment lifecycle status
        pub status: DeploymentStatusSer,
        /// When this deployment was submitted
        pub created_at: DateTime<Utc>,
        /// When this deployment was last active; None if never active
        pub last_active_at: Option<DateTime<Utc>>,
        /// Number of locked executions
        pub locked: u32,
        /// Number of pending executions
        pub pending: u32,
        /// Number of scheduled executions
        pub scheduled: u32,
        /// Number of blocked executions
        pub blocked: u32,
        /// Number of paused executions, regardless of the underlying state
        pub paused: u32,
        /// Number of executions with cancellation requested; teardown in progress
        pub cancelling: u32,
        /// Number of executions finished successfully
        pub finished_ok: u32,
        /// Number of executions finished with the `err` variant of the result type
        pub finished_error: u32,
        /// Number of executions finished with an execution failure
        pub finished_execution_failure: u32,
        /// Component counts by manifest section type. Absent when not requested or unavailable.
        pub component_summary: Option<deployment_summary::DeploymentComponentSummary>,
    }

    impl DeploymentStateSer {
        fn from(deployment_state: &DeploymentState, include_component_summary: bool) -> Self {
            let component_summary = include_component_summary
                .then(|| {
                    deployment_state
                        .deployment_toml
                        .as_deref()
                        .and_then(deployment_summary::deployment_component_summary)
                })
                .flatten();
            Self {
                deployment_id: deployment_state.deployment_id,
                description: deployment_state.description.clone(),
                digest: deployment_state.digest.to_string(),
                status: DeploymentStatusSer::from(&deployment_state.status),
                created_at: deployment_state.created_at,
                last_active_at: deployment_state.last_active_at,
                locked: deployment_state.locked,
                pending: deployment_state.pending,
                scheduled: deployment_state.scheduled,
                blocked: deployment_state.blocked,
                paused: deployment_state.paused,
                cancelling: deployment_state.cancelling,
                finished_ok: deployment_state.finished_ok,
                finished_error: deployment_state.finished_error,
                finished_execution_failure: deployment_state.finished_execution_failure,
                component_summary,
            }
        }
    }
    #[derive(Debug, Deserialize, IntoParams)]
    #[into_params(parameter_in = Query)]
    pub(crate) struct ListDeploymentsParams {
        /// Cursor for pagination (deployment ID)
        #[serde(default)]
        #[param(value_type = Option<String>)]
        cursor_from: Option<DeploymentId>,
        /// Number of items to return
        #[param(minimum = 1)]
        length: Option<u16>,
        /// Include the cursor item in results
        #[serde(default)]
        including_cursor: bool,
        /// Count child executions in the summary buckets as well
        #[serde(default)]
        include_derived: bool,
        /// Include component counts by manifest section type
        #[serde(default)]
        include_component_summary: bool,
    }

    /// List deployments
    #[utoipa::path(
        get,
        path = "/v1/deployments",
        tag = "deployments",
        params(ListDeploymentsParams),
        responses(
            (status = 200, description = "List of deployments", body = Vec<DeploymentStateSer>)
        )
    )]
    #[instrument(skip_all)]
    pub(crate) async fn list_deployments(
        state: State<Arc<WebApiState>>,
        Query(params): Query<ListDeploymentsParams>,
        accept: AcceptHeader,
    ) -> Result<Response, HttpResponse> {
        let length = nonzero_page_length(
            params
                .length
                .unwrap_or(LIST_DEPLOYMENT_STATES_DEFAULT_LENGTH),
            accept,
        )?;
        let conn = state
            .db_pool
            .external_api_conn()
            .await
            .map_err(|e| ErrorWrapper(e, accept))?;
        let pagination = Pagination::OlderThan {
            length,
            cursor: params.cursor_from,
            including_cursor: params.including_cursor,
        };
        let states = conn
            .list_deployment_states(
                Utc::now(),
                pagination,
                params.include_component_summary,
                DeploymentExecutionCounts::Count {
                    include_derived: params.include_derived,
                },
            )
            .await
            .map_err(|e| ErrorWrapper(e, accept))?;

        let states: Vec<DeploymentStateSer> = states
            .into_iter()
            .map(|dep| DeploymentStateSer::from(&dep, params.include_component_summary))
            .collect();

        Ok(match accept {
            AcceptHeader::Json => pretty_json_response(StatusCode::OK, &states),
            AcceptHeader::Text => {
                let mut output = String::new();
                for s in states {
                    writeln!(
                        &mut output,
                        "{} locked={} pending={} scheduled={} blocked={} paused={} cancelling={} \
                        finished_ok={} finished_error={} finished_execution_failure={}",
                        s.deployment_id,
                        s.locked,
                        s.pending,
                        s.scheduled,
                        s.blocked,
                        s.paused,
                        s.cancelling,
                        s.finished_ok,
                        s.finished_error,
                        s.finished_execution_failure,
                    )
                    .expect("writing to string");
                }
                deprecated_text_response(output)
            }
        })
    }

    /// Get current deployment ID
    #[utoipa::path(
        get,
        path = "/v1/deployment-id",
        tag = "deployments",
        responses(
            (status = 200, description = "Current deployment ID", body = String)
        )
    )]
    pub(crate) async fn get_current_deployment_id(
        state: State<Arc<WebApiState>>,
        accept: TextDefaultAcceptHeader,
    ) -> Result<Response, HttpResponse> {
        let accept = accept.into();
        let deployment_id = state.deployment_ctx.read().await.deployment_id;
        Ok(match accept {
            AcceptHeader::Json => pretty_json_response(StatusCode::OK, &deployment_id),
            AcceptHeader::Text => deployment_id.to_string().into_response(),
        })
    }

    /// A deployment-owned file the manifest references.
    #[derive(Debug, Serialize, Deserialize, ToSchema)]
    pub struct FileRefSer {
        /// Deployment-relative path.
        pub path: String,
        /// Content digest ("sha256:...").
        pub digest: String,
        /// Byte length of the referenced blob.
        pub size: u64,
    }

    impl From<&DeploymentFileRecord> for FileRefSer {
        fn from(f: &DeploymentFileRecord) -> Self {
            Self {
                path: f.path.clone(),
                digest: f.digest.to_string(),
                size: f.size,
            }
        }
    }

    /// Deployment details with config
    #[derive(Debug, Serialize, Deserialize, ToSchema)]
    pub struct DeploymentRecordSer {
        #[schema(value_type = String)]
        pub deployment_id: DeploymentId,
        pub description: Option<String>,
        /// Content digest = `sha256(deployment_toml)`
        pub digest: String,
        pub status: DeploymentStatusSer,
        pub created_at: DateTime<Utc>,
        pub last_active_at: Option<DateTime<Utc>>,
        /// Verbatim deployment manifest (`deployment.toml`)
        pub deployment_toml: String,
        /// Files the manifest references.
        pub files: Vec<FileRefSer>,
    }

    #[derive(Debug, Serialize, Deserialize, ToSchema)]
    pub(crate) struct GcOrphanFilesResponseSer {
        pub(crate) deleted_count: u64,
    }

    impl From<&DeploymentRecord> for DeploymentRecordSer {
        fn from(r: &DeploymentRecord) -> Self {
            Self {
                deployment_id: r.deployment_id,
                description: r.description.clone(),
                digest: r.digest.to_string(),
                status: DeploymentStatusSer::from(&r.status),
                created_at: r.created_at,
                last_active_at: r.last_active_at,
                deployment_toml: r.deployment_toml.clone(),
                files: r.files.iter().map(FileRefSer::from).collect(),
            }
        }
    }

    #[derive(Debug, Default, Deserialize, IntoParams)]
    #[into_params(parameter_in = Query)]
    pub(crate) struct GetDeploymentParams {
        /// Include server-generated metadata in `deployment_toml`. Omission preserves the stored server view.
        include_generated_metadata: Option<bool>,
    }

    /// Get a specific deployment by ID
    #[utoipa::path(
        get,
        path = "/v1/deployments/{deployment_id}",
        tag = "deployments",
        params(
            ("deployment_id" = String, Path, description = "Deployment ID"),
            GetDeploymentParams
        ),
        responses(
            (status = 200, description = "Deployment details", body = DeploymentRecordSer),
            (status = 404, description = "Deployment not found")
        )
    )]
    #[instrument(skip_all, fields(deployment_id))]
    pub(crate) async fn get_deployment(
        Path(deployment_id): Path<DeploymentId>,
        Query(params): Query<GetDeploymentParams>,
        state: State<Arc<WebApiState>>,
        accept: AcceptHeader,
    ) -> Result<Response, HttpResponse> {
        let conn = state
            .db_pool
            .external_api_conn()
            .await
            .map_err(|e| ErrorWrapper(e, accept))?;
        let mut record = conn
            .get_deployment(deployment_id)
            .await
            .map_err(|e| ErrorWrapper(e, accept))?
            .ok_or_else(|| HttpResponse::not_found(accept, Some("deployment")))?;
        if params.include_generated_metadata == Some(false) {
            record.deployment_toml = strip_generated_deployment_metadata(&record.deployment_toml)
                .map_err(|err| HttpResponse {
                status: StatusCode::INTERNAL_SERVER_ERROR,
                message: format!("cannot clean deployment manifest: {err:#}"),
                accept,
            })?;
        }

        let ser = DeploymentRecordSer::from(&record);
        Ok(match accept {
            AcceptHeader::Json => pretty_json_response(StatusCode::OK, &ser),
            AcceptHeader::Text => {
                let mut output = String::new();
                writeln!(
                    &mut output,
                    "{} status={} created={} last_active={} description={} digest={} manifest={}",
                    ser.deployment_id,
                    match ser.status {
                        DeploymentStatusSer::Inactive => "inactive",
                        DeploymentStatusSer::Enqueued => "enqueued",
                        DeploymentStatusSer::Active => "active",
                    },
                    ser.created_at.to_rfc3339(),
                    ser.last_active_at
                        .map(|t| t.to_rfc3339())
                        .unwrap_or_default(),
                    ser.description.unwrap_or_default(),
                    ser.digest,
                    ser.deployment_toml,
                )
                .expect("writing to string");
                for file in &ser.files {
                    writeln!(
                        &mut output,
                        "file path={} digest={} size={}",
                        file.path, file.digest, file.size,
                    )
                    .expect("writing to string");
                }
                deprecated_text_response(output)
            }
        })
    }

    /// Request payload for submitting a new deployment
    #[derive(Serialize, Deserialize, ToSchema)]
    pub struct DeploymentSubmitPayload {
        /// Verbatim deployment manifest (`deployment.toml`). Referenced file blobs must
        /// be uploaded separately after the server returns `missing_digests`.
        pub deployment_toml: String,
        /// Optional human-readable deployment description
        pub description: Option<String>,
        /// Tolerate runtime requirements unavailable on this server while verifying
        /// the deployment before persisting it.
        #[serde(default)]
        pub allow_unavailable_runtime_config: bool,
        /// Optional client-supplied deployment ID for idempotent submission.
        /// If a deployment with this ID already exists and its content digest
        /// matches, the submission is a no-op; a digest mismatch is rejected.
        pub deployment_id: Option<String>,
    }

    #[derive(Debug, Serialize, Deserialize, ToSchema)]
    pub struct DeploymentSubmitResponse {
        pub deployment_id: String,
    }

    /// A single file-set problem found while validating a submit package.
    #[derive(Debug, Serialize, Deserialize, ToSchema)]
    pub struct FileIssue {
        pub section: String,
        pub component_name: Option<String>,
        pub field_path: String,
        pub path: Option<String>,
        pub digest: Option<String>,
        pub message: String,
    }

    #[derive(Debug, Serialize, Deserialize, ToSchema)]
    pub struct DigestMismatch {
        pub file: FileIssue,
        pub supplied_digest: String,
        pub actual_digest: String,
    }

    /// Structured `409` body describing why a submit package was rejected. No
    /// deployment is stored when this is returned. A client retries the same submit
    /// with the `missing_files` blobs attached.
    #[derive(Debug, Serialize, Deserialize, ToSchema)]
    pub struct SubmitPackageErrorBody {
        pub missing_digest_fields: Vec<FileIssue>,
        pub missing_files: Vec<FileIssue>,
        pub unexpected_files: Vec<FileIssue>,
        pub digest_mismatches: Vec<DigestMismatch>,
        pub oversized_files: Vec<FileIssue>,
    }

    impl From<&server::SubmitFileIssue> for FileIssue {
        fn from(issue: &server::SubmitFileIssue) -> Self {
            FileIssue {
                section: issue.section.clone(),
                component_name: issue.component_name.clone(),
                field_path: issue.field_path.clone(),
                path: issue.path.clone(),
                digest: issue.digest.clone(),
                message: issue.message.clone(),
            }
        }
    }

    impl From<&server::SubmitPackageError> for SubmitPackageErrorBody {
        fn from(pkg: &server::SubmitPackageError) -> Self {
            SubmitPackageErrorBody {
                missing_digest_fields: pkg.missing_digest_fields.iter().map(Into::into).collect(),
                missing_files: pkg.missing_files.iter().map(Into::into).collect(),
                unexpected_files: pkg.unexpected_files.iter().map(Into::into).collect(),
                digest_mismatches: pkg
                    .digest_mismatches
                    .iter()
                    .map(|m| DigestMismatch {
                        file: (&m.file).into(),
                        supplied_digest: m.supplied_digest.clone(),
                        actual_digest: m.actual_digest.clone(),
                    })
                    .collect(),
                oversized_files: pkg.oversized_files.iter().map(Into::into).collect(),
            }
        }
    }

    /// Parsed inputs for a submit request, from either a JSON body or a multipart package.
    struct SubmitInputs {
        deployment_toml: String,
        description: Option<String>,
        allow_unavailable_runtime_config: bool,
        deployment_id: Option<String>,
        files: Vec<server::SuppliedFile>,
    }

    /// Parse a `multipart/form-data` submit package. Text fields `deployment_toml`,
    /// `description`, `allow_unavailable_runtime_config`, and `deployment_id` carry the
    /// manifest and options. Every other part is a deployment-owned file blob: its
    /// `filename` is the deployment-relative path and its form-field `name` is the
    /// claimed `sha256:...` content digest (or `file` when unknown).
    async fn parse_multipart_submit(
        mut multipart: axum::extract::Multipart,
        accept: AcceptHeader,
    ) -> Result<SubmitInputs, HttpResponse> {
        let bad = |message: String| HttpResponse {
            status: StatusCode::BAD_REQUEST,
            message,
            accept,
        };
        let mut deployment_toml = None;
        let mut description = None;
        let mut allow_unavailable_runtime_config = false;
        let mut deployment_id = None;
        let mut files = Vec::new();
        while let Some(field) = multipart
            .next_field()
            .await
            .map_err(|err| bad(format!("invalid multipart body: {err}")))?
        {
            let name = field.name().unwrap_or_default().to_string();
            let file_name = field.file_name().map(str::to_string);
            match name.as_str() {
                "deployment_toml" => {
                    deployment_toml =
                        Some(field.text().await.map_err(|err| {
                            bad(format!("invalid `deployment_toml` field: {err}"))
                        })?);
                }
                "description" => {
                    description = Some(
                        field
                            .text()
                            .await
                            .map_err(|err| bad(format!("invalid `description` field: {err}")))?,
                    );
                }
                "allow_unavailable_runtime_config" => {
                    let value = field.text().await.map_err(|err| {
                        bad(format!(
                            "invalid `allow_unavailable_runtime_config` field: {err}"
                        ))
                    })?;
                    allow_unavailable_runtime_config = value.trim() == "true";
                }
                "deployment_id" => {
                    deployment_id = Some(
                        field
                            .text()
                            .await
                            .map_err(|err| bad(format!("invalid `deployment_id` field: {err}")))?,
                    );
                }
                // Any other part is a file blob. `name` is the claimed digest (or `file`).
                _ => {
                    let supplied_digest = (name != "file").then_some(name);
                    let path = file_name.unwrap_or_default();
                    let content = field
                        .bytes()
                        .await
                        .map_err(|err| bad(format!("cannot read file blob `{path}`: {err}")))?
                        .to_vec();
                    files.push(server::SuppliedFile {
                        path,
                        supplied_digest,
                        content,
                    });
                }
            }
        }
        Ok(SubmitInputs {
            deployment_toml: deployment_toml
                .ok_or_else(|| bad("missing `deployment_toml` field".to_string()))?,
            description,
            allow_unavailable_runtime_config,
            deployment_id,
            files,
        })
    }

    /// Submit a new deployment.
    ///
    /// Accepts either an `application/json` body (no attached file blobs; referenced
    /// files must already exist in the CAS) or a `multipart/form-data` package whose
    /// parts carry the manifest and the deployment-owned file blobs. On an incomplete
    /// package the server stores nothing and returns a `409` `SubmitPackageErrorBody`
    /// listing the missing files to attach and retry.
    #[utoipa::path(
        post,
        path = "/v1/deployments",
        tag = "deployments",
        request_body = DeploymentSubmitPayload,
        responses(
            (status = 200, description = "Deployment submitted", body = DeploymentSubmitResponse),
            (status = 400, description = "Invalid config"),
            (status = 409, description = "Incomplete or invalid package", body = SubmitPackageErrorBody)
        )
    )]
    #[instrument(skip_all)]
    pub(crate) async fn submit_deployment(
        state: State<Arc<WebApiState>>,
        accept: AcceptHeader,
        request: axum::extract::Request,
    ) -> Result<Response, HttpResponse> {
        use axum::extract::FromRequest as _;

        let is_multipart = request
            .headers()
            .get(header::CONTENT_TYPE)
            .and_then(|v| v.to_str().ok())
            .is_some_and(|ct| ct.starts_with("multipart/form-data"));

        let inputs = if is_multipart {
            let multipart = axum::extract::Multipart::from_request(request, &*state)
                .await
                .map_err(|err| HttpResponse {
                    status: StatusCode::BAD_REQUEST,
                    message: format!("invalid multipart request: {err}"),
                    accept,
                })?;
            parse_multipart_submit(multipart, accept).await?
        } else {
            let Json(payload) = Json::<DeploymentSubmitPayload>::from_request(request, &*state)
                .await
                .map_err(|err| HttpResponse {
                    status: StatusCode::BAD_REQUEST,
                    message: format!("invalid JSON body: {err}"),
                    accept,
                })?;
            SubmitInputs {
                deployment_toml: payload.deployment_toml,
                description: payload.description,
                allow_unavailable_runtime_config: payload.allow_unavailable_runtime_config,
                deployment_id: payload.deployment_id,
                files: Vec::new(),
            }
        };

        let requested_deployment_id = inputs
            .deployment_id
            .as_deref()
            .map(str::parse::<DeploymentId>)
            .transpose()
            .map_err(|err| HttpResponse {
                status: StatusCode::BAD_REQUEST,
                message: format!("invalid deployment_id: {err}"),
                accept,
            })?;
        let mut termination_watcher = state.termination_watcher.clone();
        let result = Box::pin(crate::command::server::submit_deployment(
            state.server_verified.clone(),
            &inputs.deployment_toml,
            runtime_config_availability_from_bool(inputs.allow_unavailable_runtime_config),
            Some("web-api".to_string()),
            inputs.description,
            requested_deployment_id,
            &state.prepared_dirs,
            inputs.files,
            state.db_pool.clone(),
            &mut termination_watcher,
            state.deployment_switch_manager.clone(),
        ))
        .await;

        let deployment_id = match result {
            Ok(deployment_id) => deployment_id,
            Err(server::SubmitDeploymentError::Busy) => {
                return Err(HttpResponse {
                    status: StatusCode::SERVICE_UNAVAILABLE,
                    message: "another deployment submit or switch is already running".to_string(),
                    accept,
                });
            }
            Err(server::SubmitDeploymentError::Other(err)) => {
                return Err(HttpResponse {
                    status: StatusCode::BAD_REQUEST,
                    message: format!("{err:#}"),
                    accept,
                });
            }
            Err(server::SubmitDeploymentError::Package(pkg)) => {
                let body = SubmitPackageErrorBody::from(&pkg);
                return Ok(match accept {
                    AcceptHeader::Json => pretty_json_response(StatusCode::CONFLICT, &body),
                    AcceptHeader::Text => HttpResponse {
                        status: StatusCode::CONFLICT,
                        message: format!("{pkg:?}"),
                        accept,
                    }
                    .into_response(),
                });
            }
        };
        Ok(match accept {
            AcceptHeader::Json => pretty_json_response(
                StatusCode::OK,
                &DeploymentSubmitResponse {
                    deployment_id: deployment_id.to_string(),
                },
            ),
            AcceptHeader::Text => HttpResponse {
                status: StatusCode::OK,
                message: deployment_id.to_string(),
                accept,
            }
            .into_response(),
        })
    }

    /// Fetch a deployment file blob by content digest.
    #[utoipa::path(
        get,
        path = "/v1/files/{digest}",
        tag = "deployments",
        params(("digest" = String, Path, description = "Content digest (sha256:...)")),
        responses(
            (status = 200, description = "Blob bytes", body = Vec<u8>),
            (status = 404, description = "Blob not found")
        )
    )]
    #[instrument(skip_all)]
    pub(crate) async fn get_file(
        Path(digest): Path<String>,
        state: State<Arc<WebApiState>>,
        accept: AcceptHeader,
    ) -> Result<Response, HttpResponse> {
        let digest: concepts::ContentDigest = digest.parse().map_err(|err| HttpResponse {
            status: StatusCode::BAD_REQUEST,
            message: format!("invalid digest: {err}"),
            accept,
        })?;
        let cas = state
            .db_pool
            .cas_conn()
            .await
            .map_err(|e| ErrorWrapper(e, accept))?;
        let content = cas
            .read_blob(&digest)
            .await
            .map_err(|err| HttpResponse {
                status: StatusCode::INTERNAL_SERVER_ERROR,
                message: format!("cannot read file: {err}"),
                accept,
            })?
            .ok_or_else(|| HttpResponse::not_found(accept, Some("file")))?;
        Ok(content.into_response())
    }

    /// Delete file blobs that are not referenced by a deployment.
    #[utoipa::path(
        delete,
        path = "/v1/files/orphans",
        tag = "deployments",
        responses(
            (status = 200, description = "Orphan files deleted", body = GcOrphanFilesResponseSer)
        )
    )]
    pub(crate) async fn gc_orphan_files(
        state: State<Arc<WebApiState>>,
    ) -> Result<Response, HttpResponse> {
        let accept = AcceptHeader::Json;
        let deleted_count = state
            .db_pool
            .external_api_conn()
            .await
            .map_err(|e| ErrorWrapper(e, accept))?
            .gc_orphan_files()
            .await
            .map_err(|e| ErrorWrapper(e, accept))?;
        Ok(pretty_json_response(
            StatusCode::OK,
            &GcOrphanFilesResponseSer { deleted_count },
        ))
    }

    /// Request payload for switching deployment
    #[derive(Serialize, Deserialize, ToSchema)]
    pub struct DeploymentSwitchPayload {
        /// Tolerate runtime requirements unavailable on this server while verifying.
        /// Rejected when applying the deployment without restart.
        #[serde(default)]
        pub allow_unavailable_runtime_config: bool,
        /// Apply deployment without restart
        #[serde(default, alias = "hot_redeploy")] // backcompat: 0.41.0
        pub apply: bool,
    }

    /// Switch the active deployment
    #[utoipa::path(
        put,
        path = "/v1/deployments/{deployment_id}/switch",
        tag = "deployments",
        params(
            ("deployment_id" = String, Path, description = "Deployment ID to switch to")
        ),
        request_body = DeploymentSwitchPayload,
        responses(
            (status = 200, description = "Deployment switched or enqueued", body = String),
            (status = 404, description = "Deployment not found"),
            (status = 409, description = "Validation or switch failed")
        )
    )]
    #[instrument(skip_all, fields(deployment_id))]
    pub(crate) async fn switch_deployment(
        Path(deployment_id): Path<DeploymentId>,
        state: State<Arc<WebApiState>>,
        accept: AcceptHeader,
        Json(payload): Json<DeploymentSwitchPayload>,
    ) -> Result<Response, HttpResponse> {
        tracing::Span::current().record("deployment_id", tracing::field::display(&deployment_id));
        let action = if payload.apply {
            if payload.allow_unavailable_runtime_config {
                return Err(HttpResponse::bad_request(
                    accept,
                    "`allow_unavailable_runtime_config = true` not allowed with `apply = true`"
                        .to_string(),
                ));
            }
            SwitchDeploymentAction::Activate
        } else {
            SwitchDeploymentAction::Enqueue(runtime_config_availability_from_bool(
                payload.allow_unavailable_runtime_config,
            ))
        };
        let outcome = crate::command::server::switch_deployment(
            state.deployment_switch_manager.clone(),
            deployment_id,
            action,
        )
        .await
        .map_err(|err| match err {
            crate::command::server::SwitchError::Busy => HttpResponse {
                status: StatusCode::SERVICE_UNAVAILABLE,
                message: "another deployment submit or switch is already running".to_string(),
                accept,
            },
            crate::command::server::SwitchError::NotFound => {
                HttpResponse::not_found(accept, Some("deployment"))
            }
            crate::command::server::SwitchError::Other(e) => HttpResponse {
                status: StatusCode::BAD_REQUEST,
                message: format!("{e:#}"),
                accept,
            },
        })?;
        info!(%deployment_id, "Deployment switch outcome: {outcome}");
        let message = match outcome {
            crate::command::server::SwitchOutcome::Switched => "switched",
            crate::command::server::SwitchOutcome::RestartRequired => "restart_required",
        };
        Ok(HttpResponse {
            status: StatusCode::OK,
            message: message.to_string(),
            accept,
        }
        .into_response())
    }
}

mod backtrace {
    use super::*;

    /// Selector for which backtrace version to retrieve: "first", "last" (default), or a version number
    #[derive(Debug, Clone)]
    pub(crate) enum BacktraceVersionQuery {
        First,
        Last,
        Specific(Version),
    }

    impl TryFrom<String> for BacktraceVersionQuery {
        type Error = String;
        fn try_from(s: String) -> Result<Self, String> {
            match s.as_str() {
                "first" => Ok(BacktraceVersionQuery::First),
                "last" => Ok(BacktraceVersionQuery::Last),
                v => {
                    let n: VersionType = v
                        .parse()
                        .map_err(|_| format!("invalid version value `{v}`"))?;
                    Ok(BacktraceVersionQuery::Specific(Version(n)))
                }
            }
        }
    }

    impl<'de> serde::Deserialize<'de> for BacktraceVersionQuery {
        fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
            let s = String::deserialize(d)?;
            BacktraceVersionQuery::try_from(s).map_err(serde::de::Error::custom)
        }
    }

    fn parse_version(
        version: Option<String>,
        accept: AcceptHeader,
    ) -> Result<BacktraceFilter, HttpResponse> {
        match version {
            None => Ok(BacktraceFilter::Last),
            Some(s) => match BacktraceVersionQuery::try_from(s) {
                Ok(BacktraceVersionQuery::First) => Ok(BacktraceFilter::First),
                Ok(BacktraceVersionQuery::Last) => Ok(BacktraceFilter::Last),
                Ok(BacktraceVersionQuery::Specific(v)) => Ok(BacktraceFilter::Specific(v)),
                Err(msg) => Err(HttpResponse {
                    status: StatusCode::BAD_REQUEST,
                    message: msg,
                    accept,
                }),
            },
        }
    }

    #[derive(Deserialize, Debug, IntoParams)]
    #[into_params(parameter_in = Query)]
    pub(crate) struct BacktraceParams {
        /// Which backtrace version to retrieve: "first", "last" (default), or a version number
        version: Option<String>,
    }

    /// Serializable version of `BacktraceInfo`
    #[derive(Debug, Serialize, Deserialize, ToSchema)]
    pub(crate) struct BacktraceInfoSer {
        #[schema(value_type = String)]
        pub execution_id: ExecutionId,
        #[schema(value_type = Object)]
        pub component_id: concepts::ComponentId,
        #[schema(value_type = String)]
        pub version_min_including: VersionType,
        #[schema(value_type = String)]
        pub version_max_excluding: VersionType,
        #[schema(value_type = Object)]
        pub wasm_backtrace: concepts::storage::WasmBacktrace,
    }

    impl From<concepts::storage::BacktraceInfo> for BacktraceInfoSer {
        fn from(value: concepts::storage::BacktraceInfo) -> Self {
            BacktraceInfoSer {
                execution_id: value.execution_id,
                component_id: value.component_id,
                version_min_including: value.version_min_including.0,
                version_max_excluding: value.version_max_excluding.0,
                wasm_backtrace: value.wasm_backtrace,
            }
        }
    }

    impl TryFrom<BacktraceInfoSer> for concepts::storage::BacktraceInfo {
        type Error = String;

        fn try_from(value: BacktraceInfoSer) -> Result<Self, Self::Error> {
            Ok(Self {
                execution_id: value.execution_id,
                component_id: value.component_id,
                version_min_including: Version::new(value.version_min_including),
                version_max_excluding: Version::new(value.version_max_excluding),
                wasm_backtrace: value.wasm_backtrace,
            })
        }
    }

    /// Get execution backtrace
    #[utoipa::path(
        get,
        path = "/v1/executions/{execution_id}/backtrace",
        tag = "executions",
        params(
            ("execution_id" = String, Path, description = "Execution ID"),
            BacktraceParams
        ),
        responses(
            (status = 200, description = "Execution backtrace", body = BacktraceInfoSer),
            (status = 404, description = "Not found")
        )
    )]
    #[instrument(skip_all, fields(execution_id))]
    pub(crate) async fn execution_backtrace(
        Path(execution_id): Path<ExecutionId>,
        state: State<Arc<WebApiState>>,
        Query(params): Query<BacktraceParams>,
        accept: AcceptHeader,
    ) -> Result<Response, HttpResponse> {
        let filter = parse_version(params.version, accept)?;

        let conn = state
            .db_pool
            .external_api_conn()
            .await
            .map_err(|e| ErrorWrapper(e, accept))?;

        let info = conn
            .get_backtrace(&execution_id, filter)
            .await
            .map_err(|e| ErrorWrapper(e, accept))?;

        let info_ser = BacktraceInfoSer::from(info);
        Ok(match accept {
            AcceptHeader::Json => pretty_json_response(StatusCode::OK, &info_ser),
            AcceptHeader::Text => {
                let mut output = String::new();
                writeln!(&mut output, "execution_id: {}", info_ser.execution_id)
                    .expect("writing to string");
                writeln!(&mut output, "component_id: {}", info_ser.component_id)
                    .expect("writing to string");
                writeln!(
                    &mut output,
                    "version: {}..{}",
                    info_ser.version_min_including, info_ser.version_max_excluding
                )
                .expect("writing to string");
                for frame in &info_ser.wasm_backtrace.frames {
                    writeln!(&mut output, "  {}:{}", frame.module, frame.func_name)
                        .expect("writing to string");
                    for sym in &frame.symbols {
                        if let (Some(file), Some(line), Some(col)) = (&sym.file, sym.line, sym.col)
                        {
                            writeln!(
                                &mut output,
                                "    {} ({}:{}:{})",
                                sym.func_name.as_deref().unwrap_or("??"),
                                file,
                                line,
                                col
                            )
                            .expect("writing to string");
                        } else if let Some(func) = &sym.func_name {
                            writeln!(&mut output, "    {func}").expect("writing to string");
                        }
                    }
                }
                deprecated_text_response(output)
            }
        })
    }

    #[derive(Deserialize, Debug, IntoParams)]
    #[into_params(parameter_in = Query)]
    pub(crate) struct BacktraceSourceParams {
        /// File path to retrieve (supports suffix matching if not exact)
        file: String,
        /// Which backtrace version to use for component lookup: "first", "last" (default), or a version number
        version: Option<String>,
    }

    /// Get source file for a backtrace frame
    #[utoipa::path(
        get,
        path = "/v1/executions/{execution_id}/backtrace/source",
        tag = "executions",
        params(
            ("execution_id" = String, Path, description = "Execution ID"),
            BacktraceSourceParams
        ),
        responses(
            (status = 200, description = "Source file content", body = String),
            (status = 404, description = "Not found")
        )
    )]
    #[instrument(skip_all, fields(execution_id))]
    pub(crate) async fn execution_backtrace_source(
        Path(execution_id): Path<ExecutionId>,
        state: State<Arc<WebApiState>>,
        Query(params): Query<BacktraceSourceParams>,
        accept: TextDefaultAcceptHeader,
    ) -> Result<Response, HttpResponse> {
        let accept = accept.into();
        let filter = parse_version(params.version, accept)?;

        let conn = state
            .db_pool
            .external_api_conn()
            .await
            .map_err(|e| ErrorWrapper(e, accept))?;

        let backtrace_info = conn
            .get_backtrace(&execution_id, filter)
            .await
            .map_err(|e| ErrorWrapper(e, accept))?;

        let digest = conn
            .resolve_source_digest(&backtrace_info.component_id.component_digest, &params.file)
            .await
            .map_err(|e| ErrorWrapper(e, accept))?;

        let Some(digest) = digest else {
            return Err(HttpResponse::not_found(accept, "source file"));
        };

        let cas = state
            .db_pool
            .cas_conn()
            .await
            .map_err(|e| ErrorWrapper(e, accept))?;
        let content = cas
            .read_blob(&digest)
            .await
            .map_err(|err| HttpResponse {
                status: StatusCode::INTERNAL_SERVER_ERROR,
                message: format!("cannot read source file: {err}"),
                accept,
            })?
            .ok_or_else(|| HttpResponse::not_found(accept, "source file"))?;
        let content = String::from_utf8(content).map_err(|_| HttpResponse {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            message: "backtrace source is not valid UTF-8".to_string(),
            accept,
        })?;

        Ok(match accept {
            AcceptHeader::Json => pretty_json_response(StatusCode::OK, &content),
            AcceptHeader::Text => content.into_response(),
        })
    }
}

#[derive(AcceptExtractor, Clone, Copy, Default)]
pub(crate) enum AcceptHeader {
    #[accept(mediatype = "application/json")]
    #[default]
    Json,
    #[accept(mediatype = "text/plain")]
    Text,
}

// backcompat: 0.41 exposed structured text; after 0.43 remove it for resources and actions but retain WIT, backtrace sources, and generated IDs.

#[derive(AcceptExtractor, Clone, Copy, Default)]
pub(crate) enum TextDefaultAcceptHeader {
    #[accept(mediatype = "application/json")]
    Json,
    #[accept(mediatype = "text/plain")]
    #[default]
    Text,
}

impl From<TextDefaultAcceptHeader> for AcceptHeader {
    fn from(value: TextDefaultAcceptHeader) -> Self {
        match value {
            TextDefaultAcceptHeader::Json => Self::Json,
            TextDefaultAcceptHeader::Text => Self::Text,
        }
    }
}

struct ErrorWrapper<E>(E, AcceptHeader);

pub(crate) struct HttpResponse {
    status: StatusCode,
    message: String,
    accept: AcceptHeader,
}
impl HttpResponse {
    fn from_cancel_execution_outcome(outcome: CancelOutcome, accept: AcceptHeader) -> Self {
        match outcome {
            CancelOutcome::CancelRequested => HttpResponse {
                status: StatusCode::OK,
                message: "cancellation requested".to_string(),
                accept,
            },
            CancelOutcome::AlreadyFinished => HttpResponse {
                status: StatusCode::CONFLICT,
                message: "already finished".to_string(),
                accept,
            },
            CancelOutcome::AlreadyCancelling => HttpResponse {
                status: StatusCode::CONFLICT,
                message: "already cancelling".to_string(),
                accept,
            },
        }
    }

    fn from_delay_cancel_outcome(outcome: DelayCancelOutcome, accept: AcceptHeader) -> Self {
        match outcome {
            DelayCancelOutcome::Cancelled => HttpResponse {
                status: StatusCode::OK,
                message: "cancelled".to_string(),
                accept,
            },
            DelayCancelOutcome::AlreadyFinished => HttpResponse {
                status: StatusCode::CONFLICT,
                message: "already finished".to_string(),
                accept,
            },
        }
    }

    fn not_found(accept: AcceptHeader, what: impl Into<Option<&'static str>>) -> Self {
        HttpResponse {
            status: StatusCode::NOT_FOUND,
            message: if let Some(what) = what.into() {
                format!("{what} not found")
            } else {
                "not found".to_string()
            },
            accept,
        }
    }

    fn bad_request(accept: AcceptHeader, message: String) -> Self {
        HttpResponse {
            status: StatusCode::BAD_REQUEST,
            message,
            accept,
        }
    }
}

impl IntoResponse for HttpResponse {
    fn into_response(self) -> Response {
        match self.accept {
            AcceptHeader::Json => pretty_json_response(
                self.status,
                &if self.status.is_success() {
                    json!({ "ok": self.message })
                } else {
                    json!({ "err": self.message })
                },
            ),
            AcceptHeader::Text => deprecated_text_response((self.status, self.message)),
        }
    }
}
impl From<ErrorWrapper<DbErrorGeneric>> for HttpResponse {
    #[track_caller]
    fn from(value: ErrorWrapper<DbErrorGeneric>) -> Self {
        let err = value.0;
        let accept = value.1;
        warn!("{err:?}");
        HttpResponse {
            status: StatusCode::SERVICE_UNAVAILABLE,
            message: "database error".to_string(),
            accept,
        }
    }
}
impl From<ErrorWrapper<DbErrorRead>> for HttpResponse {
    #[track_caller]
    fn from(value: ErrorWrapper<DbErrorRead>) -> Self {
        let accept = value.1;
        match value.0 {
            DbErrorRead::NotFound => HttpResponse::not_found(accept, None),
            DbErrorRead::Generic(err) => HttpResponse::from(ErrorWrapper(err, accept)),
        }
    }
}
impl From<ErrorWrapper<DbErrorWriteNonRetriable>> for HttpResponse {
    #[track_caller]
    fn from(value: ErrorWrapper<DbErrorWriteNonRetriable>) -> Self {
        let err = value.0;
        let accept = value.1;
        match err {
            DbErrorWriteNonRetriable::ValidationFailed(reason)
            | DbErrorWriteNonRetriable::IllegalState { reason, .. } => HttpResponse {
                status: StatusCode::UNPROCESSABLE_ENTITY,
                message: reason.to_string(),
                accept,
            },
            DbErrorWriteNonRetriable::Conflict => HttpResponse {
                status: StatusCode::CONFLICT,
                message: "conflict".to_string(),
                accept,
            },
            err => {
                let loc = std::panic::Location::caller();
                let (loc_file, loc_line) = (loc.file(), loc.line());

                warn!(loc_file, loc_line, "{err:?}");
                HttpResponse {
                    status: StatusCode::INTERNAL_SERVER_ERROR,
                    message: "database error".to_string(),
                    accept,
                }
            }
        }
    }
}
impl From<ErrorWrapper<DbErrorWrite>> for HttpResponse {
    #[track_caller]
    fn from(value: ErrorWrapper<DbErrorWrite>) -> Self {
        let accept = value.1;
        match value.0 {
            DbErrorWrite::NotFound => HttpResponse::not_found(accept, None),
            DbErrorWrite::Generic(err) => HttpResponse::from(ErrorWrapper(err, accept)),
            DbErrorWrite::NonRetriable(err) => HttpResponse::from(ErrorWrapper(err, accept)),
        }
    }
}
impl From<ErrorWrapper<SubmitError>> for HttpResponse {
    #[track_caller]
    fn from(value: ErrorWrapper<SubmitError>) -> Self {
        let accept = value.1;
        match value.0 {
            err @ SubmitError::Conflict => HttpResponse {
                status: StatusCode::CONFLICT,
                message: err.to_string(),
                accept,
            },
            SubmitError::FunctionNotFound => HttpResponse::not_found(accept, Some("ffqn")),
            SubmitError::DbErrorWrite(db_error_write) => {
                HttpResponse::from(ErrorWrapper(db_error_write, accept))
            }
            err @ (SubmitError::ExecutionIdMustBeTopLevel | SubmitError::ParamsInvalid(_)) => {
                HttpResponse {
                    status: StatusCode::BAD_REQUEST,
                    message: err.to_string(),
                    accept,
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{
        AcceptHeader, RetVal, format_execution_status_text, nonzero_page_length,
        parse_join_set_filter,
    };
    use chrono::{DateTime, Utc};
    use concepts::{
        ExecutionFailureKind, SupportedFunctionReturnValue,
        storage::{
            ExecutionRequest, PendingState, PendingStateFinished, PendingStateFinishedError,
            PendingStateFinishedResultKind,
        },
    };

    #[test]
    fn pagination_length_must_be_nonzero() {
        let Ok(length) = nonzero_page_length(1, AcceptHeader::Json) else {
            panic!("positive page length must be accepted");
        };
        assert_eq!(1, length.get());
        assert_eq!(
            http::StatusCode::BAD_REQUEST,
            nonzero_page_length(0, AcceptHeader::Json)
                .unwrap_err()
                .status
        );
    }

    #[test]
    fn text_status_for_finished_ok_and_error_is_explicit() {
        assert_eq!(
            format_execution_status_text(&PendingState::Finished(PendingStateFinished {
                version: 1,
                finished_at: parse_dt("2026-01-01T00:00:00Z"),
                result_kind: PendingStateFinishedResultKind::Ok,
            })),
            "Finished: OK"
        );
        assert_eq!(
            format_execution_status_text(&PendingState::Finished(PendingStateFinished {
                version: 1,
                finished_at: parse_dt("2026-01-01T00:00:00Z"),
                result_kind: PendingStateFinishedResultKind::Err(PendingStateFinishedError::Error),
            })),
            "Finished: Error"
        );
    }

    #[test]
    fn text_status_for_finished_execution_failure_is_explicit() {
        assert_eq!(
            format_execution_status_text(&PendingState::Finished(PendingStateFinished {
                version: 1,
                finished_at: parse_dt("2026-01-01T00:00:00Z"),
                result_kind: PendingStateFinishedResultKind::Err(
                    PendingStateFinishedError::ExecutionFailure(
                        ExecutionFailureKind::Uncategorized,
                    ),
                ),
            })),
            "Finished: Execution failure (Uncategorized)"
        );
    }

    #[test]
    fn execution_failure_retval_has_canonical_field() {
        let failure = concepts::FinishedExecutionFailure {
            kind: ExecutionFailureKind::TimedOut,
            reason: Some("timed out".to_string()),
            detail: None,
        };

        assert_eq!(
            serde_json::to_value(RetVal::from(
                SupportedFunctionReturnValue::ExecutionFailure(failure.clone())
            ))
            .unwrap(),
            serde_json::json!({
                "execution_failed": failure,
            })
        );
    }

    #[test]
    fn finished_event_display_is_explicit() {
        assert_eq!(
            ExecutionRequest::Finished {
                retval: SupportedFunctionReturnValue::Err(None),
                http_client_traces: None,
            }
            .to_string(),
            "Finished: Error"
        );
        assert_eq!(
            ExecutionRequest::Finished {
                retval: SupportedFunctionReturnValue::ExecutionFailure(
                    concepts::FinishedExecutionFailure {
                        kind: ExecutionFailureKind::Uncategorized,
                        reason: None,
                        detail: None,
                    },
                ),
                http_client_traces: None,
            }
            .to_string(),
            "Finished: Execution failure (Uncategorized)"
        );
    }

    #[test]
    fn response_join_set_filter_accepts_canonical_id_and_bare_named_id() {
        assert_eq!(
            parse_join_set_filter("n:session-name".to_string()).unwrap(),
            parse_join_set_filter("session-name".to_string()).unwrap(),
        );
    }

    fn parse_dt(value: &str) -> DateTime<Utc> {
        value.parse().unwrap()
    }
}