obeli-sk-wasm-workers 0.40.0

use super::caching_db_connection::WorkflowDbConnection;
use super::deadline_tracker::DeadlineTracker;
use super::event_history::{
    ApplyError, EventHistory, JoinNextRequestingFfqn, OneOffChildExecutionRequest,
    OneOffDelayRequest, Schedule, Stub, SubmitChildExecution,
};
use super::host_exports::latest::ScheduleAtTypes;
use super::host_exports::latest::obelisk::types as typesTypes;
use super::host_exports::{
    SUFFIX_FN_AWAIT_NEXT, SUFFIX_FN_SCHEDULE, SUFFIX_FN_SUBMIT,
    history_event_schedule_at_from_wast_val,
};
use super::workflow_worker::JoinNextBlockingStrategy;
use crate::WasmFileError;
use crate::activity::cancel_registry::CancelRegistry;
use crate::component_logger::{ComponentLogger, LogStrageConfig, log_activities};
use crate::workflow::deadline_tracker::EpochCallbackError;
use crate::workflow::event_history::{
    JoinSetCreate, ScheduleIntent, StubIntent, StubIntentErr, StubParams, SubmitChildIntent,
};
use crate::workflow::host_exports::latest::{self, DelayIdTypes, ExecutionIdTypes};
use crate::workflow::host_exports::{
    self, SUFFIX_FN_GET, SUFFIX_FN_STUB, execution_id_derived_into_wast_val,
    execution_id_into_wast_val, stub_result_to_wast_val,
};
use chrono::{DateTime, Utc};
use concepts::prefixed_ulid::{DeploymentId, ExecutionIdDerived};
use concepts::storage::{
    self, DbErrorRead, DbErrorWrite, HistoryEventScheduleAt, Locked, LogEntry, LogInfoAppendRow,
    LogLevel, ResponseWithCursor, Version, WasmBacktrace,
};
use concepts::storage::{HistoryEvent, StubRetVal};
use concepts::time::ClockFn;
use concepts::{
    ComponentId, ExecutionId, FunctionMetadata, FunctionRegistry, IfcFqnName, InvalidNameError,
    ReturnType, ReturnTypeExtendable, StrVariant, SupportedFunctionReturnValue,
};
use concepts::{FunctionFqn, Params};
use concepts::{JoinSetId, JoinSetKind};
use executor::worker::FatalError;
use log_activities::obelisk::log::log::Host;
use rand::SeedableRng;
use rand::rngs::StdRng;
use std::fmt::Debug;
use std::sync::Arc;
use std::time::Duration;
use tracing::{Span, debug, error, info, instrument, trace, warn};
use val_json::wast_val::WastVal;
use wasmtime::component::{Linker, Resource, ResourceType, Val};
use wasmtime_wasi::{
    ResourceTable, ResourceTableError, WasiCtx, WasiCtxBuilder, WasiCtxView, WasiView,
};
use wasmtime_wasi_io::IoView;

/// Result that is passed from guest to host as an error, must be downcast from anyhow.
#[derive(thiserror::Error, Debug, Clone)]
pub(crate) enum WorkflowFunctionError {
    // fatal errors:
    #[error("non deterministic execution: {0}")]
    NondeterminismDetected(String),
    #[error("error calling imported function {ffqn} - {reason}")]
    ImportedFunctionCallError {
        ffqn: FunctionFqn,
        reason: StrVariant,
        detail: Option<String>,
    },
    #[error("constraint violation: {0}")]
    ConstraintViolation(StrVariant),
    // retriable errors:
    #[error("interrupt, db updated")]
    InterruptDbUpdated,
    #[error(transparent)]
    DbError(DbErrorWrite),
    #[error("lock expired")]
    LockExpired,
    #[error("executor closing")]
    ExecutorClosing,
    #[error("replay interrupt")]
    ReplayInterrupt,
}

#[derive(Debug)]
pub(crate) enum WorkerPartialResult {
    FatalError(FatalError, Version),
    ReplayWaitingForResponse,
    // retriable:
    InterruptDbUpdated,
    LockExpired,
    ExecutorClosing,
    DbError(DbErrorWrite),
}

impl WorkflowFunctionError {
    pub(crate) fn into_worker_partial_result(self, version: Version) -> WorkerPartialResult {
        match self {
            WorkflowFunctionError::InterruptDbUpdated => WorkerPartialResult::InterruptDbUpdated,
            WorkflowFunctionError::DbError(db_error) => WorkerPartialResult::DbError(db_error),
            // fatal errors:
            WorkflowFunctionError::NondeterminismDetected(detail) => {
                WorkerPartialResult::FatalError(
                    FatalError::NondeterminismDetected { detail },
                    version,
                )
            }
            WorkflowFunctionError::ImportedFunctionCallError {
                ffqn,
                reason,
                detail,
            } => WorkerPartialResult::FatalError(
                FatalError::ImportedFunctionCallError {
                    ffqn,
                    reason,
                    detail,
                },
                version,
            ),
            WorkflowFunctionError::ConstraintViolation(reason) => {
                WorkerPartialResult::FatalError(FatalError::ConstraintViolation { reason }, version)
            }
            WorkflowFunctionError::LockExpired => WorkerPartialResult::LockExpired,
            WorkflowFunctionError::ExecutorClosing => WorkerPartialResult::ExecutorClosing,
            WorkflowFunctionError::ReplayInterrupt => WorkerPartialResult::ReplayWaitingForResponse,
        }
    }
}

impl From<ApplyError> for WorkflowFunctionError {
    fn from(value: ApplyError) -> WorkflowFunctionError {
        match value {
            ApplyError::NondeterminismDetected(reason) => {
                WorkflowFunctionError::NondeterminismDetected(reason)
            }
            ApplyError::InterruptDbUpdated => WorkflowFunctionError::InterruptDbUpdated,
            ApplyError::DbError(db_error) => WorkflowFunctionError::DbError(db_error),
            ApplyError::ConstraintViolation(reason) => {
                WorkflowFunctionError::ConstraintViolation(reason)
            }
            ApplyError::ExecutorClosing => WorkflowFunctionError::ExecutorClosing,
            ApplyError::ReplayInterrupt => WorkflowFunctionError::ReplayInterrupt,
        }
    }
}

pub(crate) struct WorkflowCtx {
    pub(crate) execution_id: ExecutionId,
    event_history: EventHistory,
    rng: StdRng,
    pub(crate) clock_fn: Box<dyn ClockFn>,
    pub(crate) db_connection: Box<dyn WorkflowDbConnection>,
    component_logger: ComponentLogger,
    pub(crate) resource_table: wasmtime::component::ResourceTable,
    backtrace_persist: bool,
    wasi_ctx: WasiCtx,
    is_replay: Option<ReplayKind>,
}

#[derive(derive_more::Debug)]
pub(crate) enum ImportedFnCall<'a> {
    Direct(DirectFnCall<'a>),
    Schedule(ScheduleFnCall<'a>),
    SubmitExecution(SubmitExecutionFnCall<'a>),
    AwaitNext(AwaitNextFnCall),
    Stub(StubFnCall<'a>),
    Get(GetFnCall),
}

#[derive(derive_more::Debug)]
pub(crate) struct DirectFnCall<'a> {
    ffqn: FunctionFqn,
    fn_component_id: ComponentId,
    params: &'a [Val],
    #[debug(skip)]
    wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl DirectFnCall<'_> {
    async fn call_imported_fn(
        self,
        ctx: &mut WorkflowCtx,
        called_at: DateTime<Utc>,
    ) -> Result<wasmtime::component::Val, WorkflowFunctionError> {
        let DirectFnCall {
            ffqn,
            fn_component_id,
            params,
            wasm_backtrace,
        } = self;
        OneOffChildExecutionRequest::apply(
            ffqn,
            fn_component_id,
            Params::from_wasmtime(Arc::from(params)),
            wasm_backtrace,
            &mut ctx.event_history,
            &mut *ctx.db_connection,
            called_at,
        )
        .await
    }
}

#[derive(derive_more::Debug)]
pub(crate) struct ScheduleFnCall<'a> {
    target_ffqn: FunctionFqn,
    target_component_id: ComponentId,
    schedule_at: HistoryEventScheduleAt,
    #[debug(skip)]
    target_params: &'a [Val],
    #[debug(skip)]
    wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl ScheduleFnCall<'_> {
    fn new(
        target_ffqn: FunctionFqn,
        called_ffqn: FunctionFqn,
        params: &[Val],
        wasm_backtrace: Option<storage::WasmBacktrace>,
        target_component_id: ComponentId,
    ) -> Result<ScheduleFnCall<'_>, WorkflowFunctionError> {
        let Some((schedule_at, params)) = params.split_first() else {
            return Err(WorkflowFunctionError::ImportedFunctionCallError {
                ffqn: called_ffqn,
                reason: StrVariant::Static("exepcted at least one parameter of type `schedule-at`"),
                detail: None,
            });
        };
        let schedule_at = match WastVal::try_from(schedule_at.clone()) {
            Ok(ok) => ok,
            Err(err) => {
                return Err(WorkflowFunctionError::ImportedFunctionCallError {
                    ffqn: called_ffqn,
                    reason: format!(
                        "cannot convert `schedule-at` to internal representation - {err}"
                    )
                    .into(),
                    detail: Some(format!("{err:?}")),
                });
            }
        };
        let schedule_at = match history_event_schedule_at_from_wast_val(&schedule_at) {
            Ok(ok) => ok,
            Err(detail) => {
                return Err(WorkflowFunctionError::ImportedFunctionCallError {
                    ffqn: called_ffqn,
                    reason: "first parameter type must be `schedule-at`".into(),
                    detail: Some(detail.to_string()),
                });
            }
        };
        Ok(ScheduleFnCall {
            target_ffqn,
            target_component_id,
            schedule_at,
            target_params: params,
            wasm_backtrace,
        })
    }

    async fn call_imported_fn(
        self,
        ctx: &mut WorkflowCtx,
        called_at: DateTime<Utc>,
        called_ffqn: &FunctionFqn,
    ) -> Result<wasmtime::component::Val, WorkflowFunctionError> {
        let ScheduleFnCall {
            target_ffqn,
            target_component_id,
            schedule_at,
            target_params,
            wasm_backtrace,
        } = self;
        // Generate new ULID using this execution, changing the randomness part using this execution's seed.
        // We are ignoring the possibility of ExecutionId conflict:
        // According to birthday paradox, with 80 bits of randomness, a 50% chance
        // of a conflict will occur only after we generate 10^12 ULIDs.
        let execution_id = ExecutionId::from_parts(
            ctx.execution_id.get_top_level().timestamp_part(),
            ctx.next_u128(),
        );
        let scheduled_at_if_new = schedule_at.as_date_time(called_at).map_err(|err| {
            WorkflowFunctionError::ImportedFunctionCallError {
                ffqn: called_ffqn.clone(),
                reason: "schedule-at conversion error".into(),
                detail: Some(format!("{err:?}")),
            }
        })?;
        let execution_id_val = execution_id_into_wast_val(&execution_id).as_val();
        Ok(Schedule {
            schedule_at,
            scheduled_at_if_new,
            execution_id,
            ffqn: target_ffqn,
            intent: ScheduleIntent::Ok {
                fn_component_id: target_component_id,
                params: Params::from_wasmtime(Arc::from(target_params)),
            },
            wasm_backtrace,
        }
        .apply(&mut ctx.event_history, &mut *ctx.db_connection, called_at)
        .await?
        .map(|()| execution_id_val)
        .expect("Ok intent cannot produce ScheduleRequestError"))
    }
}

#[derive(derive_more::Debug)]
pub(crate) struct SubmitExecutionFnCall<'a> {
    target_ffqn: FunctionFqn,
    target_component_id: ComponentId,
    join_set_id: JoinSetId,
    #[debug(skip)]
    target_params: &'a [Val],
    #[debug(skip)]
    wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl SubmitExecutionFnCall<'_> {
    fn new<'a>(
        target_ffqn: FunctionFqn,
        called_ffqn: FunctionFqn,
        store_ctx: &mut wasmtime::StoreContextMut<'a, WorkflowCtx>,
        params: &'a [Val],
        wasm_backtrace: Option<storage::WasmBacktrace>,
        target_component_id: ComponentId,
    ) -> Result<SubmitExecutionFnCall<'a>, WorkflowFunctionError> {
        let (join_set_id, params) =
            ImportedFnCall::extract_join_set_id(&called_ffqn, store_ctx, params).map_err(
                |detail| WorkflowFunctionError::ImportedFunctionCallError {
                    ffqn: called_ffqn,
                    reason: StrVariant::Static("cannot extract join set id"),
                    detail: Some(detail),
                },
            )?;
        Ok(SubmitExecutionFnCall {
            target_ffqn,
            target_component_id,
            join_set_id,
            target_params: params,
            wasm_backtrace,
        })
    }

    async fn call_imported_fn(
        self,
        ctx: &mut WorkflowCtx,
        called_at: DateTime<Utc>,
    ) -> Result<wasmtime::component::Val, WorkflowFunctionError> {
        let SubmitExecutionFnCall {
            target_ffqn,
            target_component_id,
            join_set_id,
            target_params,
            wasm_backtrace,
        } = self;
        let child_execution_id = ctx.next_child_id(&join_set_id);
        let child_execution_id_val =
            execution_id_derived_into_wast_val(&child_execution_id).as_val();
        Ok(SubmitChildExecution {
            target_ffqn: target_ffqn.clone(),
            join_set_id,
            intent: SubmitChildIntent::Ok {
                fn_component_id: target_component_id,
                params: Params::from_wasmtime(Arc::from(target_params)),
            },
            child_execution_id,
            wasm_backtrace,
        }
        .apply(&mut ctx.event_history, &mut *ctx.db_connection, called_at)
        .await?
        .map(|()| child_execution_id_val)
        .expect("Ok intent cannot produce ChildExecutionRequestError"))
    }
}

#[derive(derive_more::Debug)]
pub(crate) struct AwaitNextFnCall {
    target_ffqn: FunctionFqn,
    join_set_id: JoinSetId,
    #[debug(skip)]
    wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl AwaitNextFnCall {
    fn new<'a>(
        target_ffqn: FunctionFqn,
        called_ffqn: FunctionFqn,
        store_ctx: &mut wasmtime::StoreContextMut<'a, WorkflowCtx>,
        params: &'a [Val],
        wasm_backtrace: Option<storage::WasmBacktrace>,
    ) -> Result<AwaitNextFnCall, WorkflowFunctionError> {
        let (join_set_id, params) =
            match ImportedFnCall::extract_join_set_id(&called_ffqn, store_ctx, params) {
                Ok(ok) => ok,
                Err(detail) => {
                    return Err(WorkflowFunctionError::ImportedFunctionCallError {
                        ffqn: called_ffqn,
                        reason: "cannot extract join set id".into(),
                        detail: Some(detail),
                    });
                }
            };
        if !params.is_empty() {
            return Err(WorkflowFunctionError::ImportedFunctionCallError {
                reason: StrVariant::Static("wrong parameter length"),
                detail: Some(format!(
                    "wrong parameter length, expected single string parameter containing join-set-id, got {} other parameters",
                    params.len()
                )),
                ffqn: called_ffqn,
            });
        }
        Ok(AwaitNextFnCall {
            target_ffqn,
            join_set_id,
            wasm_backtrace,
        })
    }

    async fn call_imported_fn(
        self,
        ctx: &mut WorkflowCtx,
        called_at: DateTime<Utc>,
    ) -> Result<wasmtime::component::Val, WorkflowFunctionError> {
        let AwaitNextFnCall {
            target_ffqn,
            join_set_id,
            wasm_backtrace,
        } = self;
        JoinNextRequestingFfqn {
            join_set_id,
            wasm_backtrace,
            requested_ffqn: target_ffqn,
        }
        .apply(&mut ctx.event_history, &mut *ctx.db_connection, called_at)
        .await
    }
}

#[derive(derive_more::Debug)]
pub(crate) struct StubFnCall<'a> {
    target_ffqn: FunctionFqn,
    target_execution_id: ExecutionIdDerived,
    target_return_type: ReturnTypeExtendable,
    #[debug(skip)]
    return_value: &'a wasmtime::component::Val,
    #[debug(skip)]
    wasm_backtrace: Option<storage::WasmBacktrace>,
}
impl StubFnCall<'_> {
    fn new(
        target_ffqn: FunctionFqn,
        called_ffqn: FunctionFqn,
        params: &[Val],
        wasm_backtrace: Option<storage::WasmBacktrace>,
        target_fn_metadata: FunctionMetadata,
    ) -> Result<StubFnCall<'_>, WorkflowFunctionError> {
        let Some((target_execution_id, [return_value])) = params.split_first() else {
            return Err(WorkflowFunctionError::ImportedFunctionCallError {
                ffqn: called_ffqn,
                reason: StrVariant::Static("-stub function exepcts two parameters"),
                detail: None,
            });
        };

        let target_execution_id = match ExecutionId::try_from(target_execution_id) {
            Ok(ExecutionId::Derived(derived)) => derived,
            Ok(ExecutionId::TopLevel(_)) => {
                return Err(WorkflowFunctionError::ImportedFunctionCallError {
                    ffqn: called_ffqn,
                    reason: "first parameter must not be a top-level `execution-id`".into(),
                    detail: None,
                });
            }
            Err(err) => {
                return Err(WorkflowFunctionError::ImportedFunctionCallError {
                    ffqn: called_ffqn,
                    reason: format!("cannot parse first parameter as `execution-id`: {err}").into(),
                    detail: Some(format!("{err:?}")),
                });
            }
        };

        let ReturnType::Extendable(target_return_type) = target_fn_metadata.return_type else {
            unreachable!("only functions with compatible return types are exported and extended")
        };

        Ok(StubFnCall {
            target_ffqn,
            target_execution_id,
            target_return_type,
            return_value,
            wasm_backtrace,
        })
    }

    async fn get_stub_intent(
        ctx: &mut WorkflowCtx,
        target_execution_id: ExecutionIdDerived,
        target_ffqn: &FunctionFqn,
        retval: SupportedFunctionReturnValue,
    ) -> Result<StubIntent, WorkflowFunctionError> {
        match ctx
            .db_connection
            .get_stub_create_request(&ExecutionId::Derived(target_execution_id))
            .await
        {
            Ok(create_req) if *target_ffqn == create_req.ffqn => {
                // If this is replay, retval must match the original.
                // Otherwise, we assume the parent workflow and this stub writer share the
                // same WIT definition, meaning type checking is done at server startup.
                // `stub-json` must do its own type checking before this call.
                Ok(StubIntent::StubTypeChecked(retval))
            }
            Ok(create_req) => Ok(StubIntent::Err(StubIntentErr::TypeCheckError(format!(
                "ffqn mismatch, code stubs {target_ffqn}, but execution was created with {}",
                create_req.ffqn
            )))),
            Err(DbErrorRead::NotFound) => Ok(StubIntent::Err(StubIntentErr::ExecutionNotFound)),
            Err(err) => {
                Err(WorkflowFunctionError::DbError(err.into())) // intermittent error
            }
        }
    }

    // -stub is called
    async fn call_imported_fn(
        self,
        ctx: &mut WorkflowCtx,
        called_at: DateTime<Utc>,
    ) -> Result<wasmtime::component::Val, WorkflowFunctionError> {
        let StubFnCall {
            target_ffqn,
            target_execution_id,
            target_return_type,
            return_value,
            wasm_backtrace,
        } = self;

        let retval = SupportedFunctionReturnValue::from_val_and_type_wrapper_tl(
            return_value.clone(),
            target_return_type.type_wrapper_tl,
        )
        .expect("only functions with compatible return types are exported and extended");

        let params = StubParams {
            target_execution_id: target_execution_id.clone(),
            retval_hash: StubRetVal::Typed(retval.clone()).hash(),
        };
        let intent = Self::get_stub_intent(ctx, target_execution_id, &target_ffqn, retval).await?;

        let stub_result = Stub {
            intent,
            params,
            wasm_backtrace,
        }
        .apply(&mut ctx.event_history, &mut *ctx.db_connection, called_at)
        .await?;

        Ok(stub_result_to_wast_val(stub_result).as_val())
    }
}

#[derive(derive_more::Debug)]
pub(crate) struct GetFnCall {
    target_ffqn: FunctionFqn,
    child_execution_id: ExecutionIdDerived,
}
impl GetFnCall {
    fn new(
        target_ffqn: FunctionFqn,
        called_ffqn: FunctionFqn,
        params: &[Val],
    ) -> Result<GetFnCall, WorkflowFunctionError> {
        let Some((child_execution_id, [])) = params.split_first() else {
            return Err(WorkflowFunctionError::ImportedFunctionCallError {
                ffqn: called_ffqn,
                reason: StrVariant::Static("-get function exepcts one parameter"),
                detail: Some(format!(
                    "wrong parameter length, expected single parameter of type `execution-id`, got {} parameters",
                    params.len()
                )),
            });
        };

        let child_execution_id = match ExecutionId::try_from(child_execution_id) {
            Ok(ExecutionId::Derived(derived)) => derived,
            Ok(ExecutionId::TopLevel(_)) => {
                return Err(WorkflowFunctionError::ImportedFunctionCallError {
                    ffqn: called_ffqn,
                    reason: "first parameter must not be a top-level `execution-id`".into(),
                    detail: None,
                });
            }
            Err(err) => {
                return Err(WorkflowFunctionError::ImportedFunctionCallError {
                    ffqn: called_ffqn,
                    reason: format!(
                        "cannot parse first parameter as a derived `execution-id`: {err}"
                    )
                    .into(),
                    detail: Some(format!("{err:?}")),
                });
            }
        };
        Ok(GetFnCall {
            target_ffqn,
            child_execution_id,
        })
    }

    // Stateless, never interrupts the execution
    fn call_imported_fn(self, ctx: &mut WorkflowCtx) -> wasmtime::component::Val {
        let GetFnCall {
            target_ffqn,
            child_execution_id,
        } = self;

        match ctx
            .event_history
            .get_processed_response(&child_execution_id, &target_ffqn)
        {
            // Return Ok(None) or Ok(retval)
            Ok(wast_val) => wasmtime::component::Val::Result(Ok(Some(Box::new(wast_val.as_val())))),
            Err(get_extension_err) => wasmtime::component::Val::Result(Err(Some(Box::new(
                wasmtime::component::Val::from(get_extension_err),
            )))),
        }
    }
}

impl<'a> ImportedFnCall<'a> {
    fn extract_join_set_id<'ctx>(
        called_ffqn: &FunctionFqn,
        store_ctx: &'ctx mut wasmtime::StoreContextMut<'a, WorkflowCtx>,
        params: &'a [Val],
    ) -> Result<(JoinSetId, &'a [Val]), String> {
        let Some((join_set_id, params)) = params.split_first() else {
            error!("Got empty params, expected JoinSetId");
            return Err(format!(
                "error running {called_ffqn} extension function: exepcted at least one parameter with JoinSetId, got empty parameter list"
            ));
        };
        if let Val::Resource(resource) = join_set_id {
            let resource: Resource<JoinSetId> = resource
                .try_into_resource(&mut *store_ctx)
                .inspect_err(|err| error!("Cannot turn `ResourceAny` into a `Resource` - {err:?}"))
                .map_err(|err| format!("cannot turn `ResourceAny` into a `Resource` - {err:?}"))?;
            let join_set_id = store_ctx
                .data()
                .resource_to_join_set_id(&resource)
                .map_err(|resource_err| {
                    format!("error running {called_ffqn} extension function: {resource_err:?}")
                })?;
            Ok((join_set_id.clone(), params))
        } else {
            error!("Wrong type for JoinSetId, expected join-set-id, got `{join_set_id:?}`");
            Err(format!(
                "wrong type for JoinSetId, expected join-set-id, got `{join_set_id:?}`"
            ))
        }
    }

    #[instrument(skip_all, fields(ffqn = %called_ffqn, otel.name = format!("ImportedFnCall::new {called_ffqn}")), name = "ImportedFnCall::new")]
    pub(crate) fn new<'ctx>(
        called_ffqn: FunctionFqn,
        store_ctx: &'ctx mut wasmtime::StoreContextMut<'a, WorkflowCtx>,
        params: &'a [Val],
        backtrace_persist: bool,
        fn_registry: &dyn FunctionRegistry,
    ) -> Result<ImportedFnCall<'a>, WorkflowFunctionError> {
        let wasm_backtrace = if backtrace_persist {
            let wasm_backtrace = wasmtime::WasmBacktrace::capture(&store_ctx);
            concepts::storage::WasmBacktrace::maybe_from(&wasm_backtrace)
        } else {
            None
        };
        // TODO: Instead of stripping the suffix here use registry and switch based on extension.
        if let Some(target_package_name) = called_ffqn.ifc_fqn.package_strip_obelisk_ext_suffix() {
            let target_ifc_fqn = IfcFqnName::from_parts(
                called_ffqn.ifc_fqn.namespace(),
                target_package_name,
                called_ffqn.ifc_fqn.ifc_name(),
                called_ffqn.ifc_fqn.version(),
            );
            if let Some(function_name) = called_ffqn.function_name.strip_suffix(SUFFIX_FN_SUBMIT) {
                let target_ffqn = FunctionFqn::new_arc(
                    Arc::from(target_ifc_fqn.to_string()),
                    Arc::from(function_name),
                );
                let (target_fn_metadata, target_component_id) = fn_registry
                    .get_by_exported_function(&target_ffqn)
                    .expect("function obtained from `fn_registry.all_exports()` must be found");
                assert_eq!(None, target_fn_metadata.extension);
                let submit = SubmitExecutionFnCall::new(
                    target_ffqn,
                    called_ffqn,
                    store_ctx,
                    params,
                    wasm_backtrace,
                    target_component_id,
                )?;
                Ok(ImportedFnCall::SubmitExecution(submit))
            } else if let Some(function_name) =
                called_ffqn.function_name.strip_suffix(SUFFIX_FN_AWAIT_NEXT)
            {
                let target_ffqn = FunctionFqn::new_arc(
                    Arc::from(target_ifc_fqn.to_string()),
                    Arc::from(function_name),
                );
                let await_next = AwaitNextFnCall::new(
                    target_ffqn,
                    called_ffqn,
                    store_ctx,
                    params,
                    wasm_backtrace,
                )?;

                Ok(ImportedFnCall::AwaitNext(await_next))
            } else if let Some(function_name) =
                called_ffqn.function_name.strip_suffix(SUFFIX_FN_GET)
            {
                let target_ffqn = FunctionFqn::new_arc(
                    Arc::from(target_ifc_fqn.to_string()),
                    Arc::from(function_name),
                );
                let get = GetFnCall::new(target_ffqn, called_ffqn, params)?;
                Ok(ImportedFnCall::Get(get))
            } else {
                unreachable!(
                    "all extensions were covered, no way fn_registry contains {called_ffqn}"
                );
            }
        } else if let Some(target_package_name) =
            called_ffqn.ifc_fqn.package_strip_obelisk_schedule_suffix()
        {
            let target_ifc_fqn = IfcFqnName::from_parts(
                called_ffqn.ifc_fqn.namespace(),
                target_package_name,
                called_ffqn.ifc_fqn.ifc_name(),
                called_ffqn.ifc_fqn.version(),
            );
            if let Some(function_name) = called_ffqn.function_name.strip_suffix(SUFFIX_FN_SCHEDULE)
            {
                let target_ffqn = FunctionFqn::new_arc(
                    Arc::from(target_ifc_fqn.to_string()),
                    Arc::from(function_name),
                );
                let (target_fn_metadata, target_component_id) = fn_registry
                    .get_by_exported_function(&target_ffqn)
                    .expect("function obtained from `fn_registry.all_exports()` must be found");
                assert_eq!(None, target_fn_metadata.extension);
                let schedule = ScheduleFnCall::new(
                    target_ffqn,
                    called_ffqn,
                    params,
                    wasm_backtrace,
                    target_component_id,
                )?;
                Ok(ImportedFnCall::Schedule(schedule))
            } else {
                unreachable!(
                    "all extensions were covered, no way fn_registry contains {called_ffqn}"
                );
            }
        } else if let Some(target_package_name) =
            called_ffqn.ifc_fqn.package_strip_obelisk_stub_suffix()
        {
            let target_ifc_fqn = IfcFqnName::from_parts(
                called_ffqn.ifc_fqn.namespace(),
                target_package_name,
                called_ffqn.ifc_fqn.ifc_name(),
                called_ffqn.ifc_fqn.version(),
            );
            if let Some(function_name) = called_ffqn.function_name.strip_suffix(SUFFIX_FN_STUB) {
                let target_ffqn = FunctionFqn::new_arc(
                    Arc::from(target_ifc_fqn.to_string()),
                    Arc::from(function_name),
                );
                let (target_fn_metadata, _target_component_id) = fn_registry
                    .get_by_exported_function(&target_ffqn)
                    .expect("function obtained from `fn_registry.all_exports()` must be found");
                assert_eq!(None, target_fn_metadata.extension);
                let stub = StubFnCall::new(
                    target_ffqn,
                    called_ffqn,
                    params,
                    wasm_backtrace,
                    target_fn_metadata,
                )?;
                Ok(ImportedFnCall::Stub(stub))
            } else {
                unreachable!(
                    "all extensions were covered, no way fn_registry contains {called_ffqn}"
                );
            }
        } else {
            let (fn_metadata, fn_component_id) = fn_registry
                .get_by_exported_function(&called_ffqn)
                .expect("function obtained from `fn_registry.all_exports()` must be found");
            assert_eq!(None, fn_metadata.extension);
            Ok(ImportedFnCall::Direct(DirectFnCall {
                ffqn: called_ffqn,
                params,
                wasm_backtrace,
                fn_component_id,
            }))
        }
    }

    fn ffqn(&self) -> &FunctionFqn {
        match self {
            ImportedFnCall::Direct(DirectFnCall { ffqn, .. })
            | ImportedFnCall::Schedule(ScheduleFnCall {
                target_ffqn: ffqn, ..
            })
            | ImportedFnCall::SubmitExecution(SubmitExecutionFnCall {
                target_ffqn: ffqn, ..
            })
            | ImportedFnCall::AwaitNext(AwaitNextFnCall {
                target_ffqn: ffqn, ..
            })
            | ImportedFnCall::Stub(StubFnCall {
                target_ffqn: ffqn, ..
            })
            | ImportedFnCall::Get(GetFnCall {
                target_ffqn: ffqn, ..
            }) => ffqn,
        }
    }
}

impl wasmtime::component::HasData for WorkflowCtx {
    type Data<'a> = &'a mut WorkflowCtx;
}
impl IoView for WorkflowCtx {
    fn table(&mut self) -> &mut ResourceTable {
        &mut self.resource_table
    }
}
impl WasiView for WorkflowCtx {
    fn ctx(&mut self) -> WasiCtxView<'_> {
        WasiCtxView {
            ctx: &mut self.wasi_ctx,
            table: &mut self.resource_table,
        }
    }
}

const IFC_FQN_WORKFLOW_SUPPORT: &str = "obelisk:workflow/workflow-support@6.0.0";
const IFC_FQN_WORKFLOW_SUPPORT_BACKTRACE: &str =
    "obelisk:workflow/workflow-support-backtrace@6.0.0";

#[derive(Clone, Copy, PartialEq, Eq, derive_more::Display)]
pub(crate) enum ReplayKind {
    Finished,
    Unfinished,
}

impl WorkflowCtx {
    #[expect(clippy::too_many_arguments)]
    pub(crate) fn new(
        deployment_id: DeploymentId,
        db_connection: Box<dyn WorkflowDbConnection>,
        event_history: Vec<(HistoryEvent, Version)>,
        responses: Vec<ResponseWithCursor>,
        seed: u64,
        clock_fn: Box<dyn ClockFn>,
        join_next_blocking_strategy: JoinNextBlockingStrategy,
        worker_span: Span,
        backtrace_persist: bool,
        deadline_tracker: Box<dyn DeadlineTracker>,
        fn_registry: Arc<dyn FunctionRegistry>,
        cancel_registry: CancelRegistry,
        locked_event: Locked,
        lock_extension: Option<Duration>,
        subscription_interruption: Option<Duration>,
        logs_storage_config: Option<LogStrageConfig>,
        is_replay: Option<ReplayKind>,
    ) -> Self {
        let mut wasi_ctx_builder = WasiCtxBuilder::new();
        wasi_ctx_builder.allow_tcp(false);
        wasi_ctx_builder.allow_udp(false);
        wasi_ctx_builder.insecure_random_seed(0);
        let run_id = locked_event.run_id;
        let execution_id = db_connection.execution_id().clone();
        Self {
            execution_id: execution_id.clone(),
            db_connection,
            event_history: EventHistory::new(
                deployment_id,
                event_history,
                responses,
                join_next_blocking_strategy,
                fn_registry,
                cancel_registry,
                deadline_tracker,
                locked_event,
                lock_extension,
                subscription_interruption,
                worker_span.clone(),
                is_replay == Some(ReplayKind::Unfinished),
            ),
            rng: StdRng::seed_from_u64(seed),
            clock_fn,
            component_logger: ComponentLogger {
                span: worker_span,
                execution_id,
                run_id,
                logs_storage_config,
            },
            resource_table: wasmtime::component::ResourceTable::default(),
            backtrace_persist,
            wasi_ctx: wasi_ctx_builder.build(),
            is_replay,
        }
    }

    pub(crate) fn check_epoch_callback(&self) -> Result<(), EpochCallbackError> {
        self.event_history.deadline_tracker.check_epoch_callback()
    }

    pub(crate) async fn flush(&mut self) -> Result<(), DbErrorWrite> {
        self.db_connection
            .flush_non_blocking_event_cache(self.clock_fn.now())
            .await
    }

    /// Return durable sleep expiry time.
    async fn persist_sleep(
        &mut self,
        schedule_at: HistoryEventScheduleAt,
        name: Option<String>,
        wasm_backtrace: Option<storage::WasmBacktrace>,
    ) -> Result<Result<DateTime<Utc>, ()>, WorkflowFunctionError> {
        let expires_at_if_new = schedule_at
            .as_date_time(self.clock_fn.now())
            .map_err(|err| {
                const FFQN: FunctionFqn =
                    FunctionFqn::new_static(IFC_FQN_WORKFLOW_SUPPORT, "sleep");
                WorkflowFunctionError::ImportedFunctionCallError {
                    ffqn: FFQN,
                    reason: "schedule-at conversion error".into(),
                    detail: Some(format!("{err:?}")),
                }
            })?;
        OneOffDelayRequest::apply(
            schedule_at,
            name,
            expires_at_if_new,
            wasm_backtrace,
            &mut self.event_history,
            &mut *self.db_connection,
            self.clock_fn.now(),
        )
        .await
    }

    // Must be persisted by the caller.
    fn next_u128(&mut self) -> u128 {
        rand::Rng::random(&mut self.rng)
    }

    pub(crate) fn resource_to_join_set_id(
        &self,
        resource: &Resource<JoinSetId>,
    ) -> Result<&JoinSetId, ResourceTableError> {
        self.resource_table
            .get(resource)
            .inspect_err(|err| error!("Cannot get resource - {err:?}"))
    }

    async fn persist_join_set_with_kind(
        &mut self,
        name: String,
        kind: JoinSetKind,
        wasm_backtrace: Option<storage::WasmBacktrace>,
    ) -> Result<Resource<JoinSetId>, JoinSetCreateError> {
        if !self.event_history.join_set_name_exists(&name, kind) {
            let join_set_id = JoinSetId::new(kind, StrVariant::from(name))
                .map_err(JoinSetCreateError::InvalidNameError)?;
            let join_set_id = JoinSetCreate {
                join_set_id,
                wasm_backtrace,
            }
            .apply(
                &mut self.event_history,
                &mut *self.db_connection,
                self.clock_fn.now(),
            )
            .await
            .map_err(JoinSetCreateError::ApplyError)?;
            let join_set = self
                .resource_table
                .push(join_set_id)
                .map_err(JoinSetCreateError::ResourceTableError)?;
            Ok(join_set)
        } else {
            Err(JoinSetCreateError::Conflict)
        }
    }

    fn get_host_maybe_capture_backtrace<'a>(
        caller: &'a mut wasmtime::StoreContextMut<'_, Self>,
        wit_backtrace: Option<typesTypes::backtrace::WasmBacktrace>,
    ) -> (&'a mut WorkflowCtx, Option<storage::WasmBacktrace>) {
        let backtrace = wit_backtrace
            .map(Self::wit_wasm_backtrace_to_storage)
            .or_else(|| {
                if caller.data().backtrace_persist {
                    let backtrace = wasmtime::WasmBacktrace::capture(&caller);
                    WasmBacktrace::maybe_from(&backtrace)
                } else {
                    None
                }
            });
        let host = caller.data_mut();
        (host, backtrace)
    }

    // Adding host functions using `func_wrap*` so that we can capture the backtrace.
    pub(crate) fn add_to_linker(
        linker: &mut Linker<Self>,
        stub_wasi: bool,
    ) -> Result<(), WasmFileError> {
        log_activities::obelisk::log::log::add_to_linker::<_, WorkflowCtx>(
            linker,
            |state: &mut Self| state,
        )
        .map_err(|err| WasmFileError::linking_error("cannot link obelisk::log", err))?;
        // link obelisk:types/execution@4.2.0 (has no host functions, just type definitions)
        typesTypes::execution::add_to_linker::<_, WorkflowCtx>(linker, |state: &mut Self| state)
            .map_err(|err| {
                WasmFileError::linking_error("cannot link obelisk:types/execution@4.2.0", err)
            })?;

        // link obelisk:workflow/workflow-support interface (native, no backtrace)
        Self::add_to_linker_workflow_support(linker, IFC_FQN_WORKFLOW_SUPPORT)?;

        // link obelisk:workflow/workflow-support-backtrace interface (JS runtime)
        Self::add_to_linker_workflow_support_backtrace(linker, IFC_FQN_WORKFLOW_SUPPORT_BACKTRACE)?;

        // link obelisk:types/join-set interface
        Self::add_to_linker_join_set(linker)?;

        if stub_wasi {
            super::wasi::add_to_linker_async(linker)?;
        }
        Ok(())
    }

    fn add_to_linker_join_set(linker: &mut Linker<Self>) -> Result<(), WasmFileError> {
        const IFC_FQN_JOIN_SET: &str = "obelisk:types/join-set@5.0.0";
        let mut inst_join_set_ifc = linker
            .instance(IFC_FQN_JOIN_SET)
            .map_err(|err| WasmFileError::linking_error(IFC_FQN_JOIN_SET, err))?;

        inst_join_set_ifc
            .resource_async(
                "join-set",
                ResourceType::host::<JoinSetId>(),
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>, rep: u32| {
                    Box::new(async move {
                        let (host, wasm_backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, None);
                        let resource: Resource<JoinSetId> =
                            wasmtime::component::Resource::new_own(rep);
                        host.join_set_close_resource(resource, wasm_backtrace).await
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("cannot link resource join-set", err))?;

        // id: func() -> string;
        inst_join_set_ifc
            .func_wrap(
                "[method]join-set.id",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (resource,): (Resource<JoinSetId>,)| {
                    let host = caller.data_mut();
                    let id = host.resource_to_join_set_id(&resource)?.to_string();
                    Ok((id,))
                },
            )
            .map_err(|err| WasmFileError::linking_error("cannot link function id", err))?;

        // last-id: func() -> option<response-id>;
        inst_join_set_ifc
            .func_wrap(
                "[method]join-set.last-id",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (resource,): (Resource<JoinSetId>,)| {
                    let host = caller.data_mut();
                    let join_set_id = host.resource_to_join_set_id(&resource)?.clone();
                    let last_id = host.join_set_last_id_wit(&join_set_id);
                    Ok((last_id,))
                },
            )
            .map_err(|err| WasmFileError::linking_error("cannot link function last-id", err))?;

        Ok(())
    }

    fn wit_wasm_backtrace_to_storage(
        bt: typesTypes::backtrace::WasmBacktrace,
    ) -> storage::WasmBacktrace {
        storage::WasmBacktrace {
            frames: bt
                .frames
                .into_iter()
                .map(|f| storage::FrameInfo {
                    module: f.module,
                    func_name: f.func_name,
                    symbols: f
                        .symbols
                        .into_iter()
                        .map(|s| storage::FrameSymbol {
                            func_name: s.func_name,
                            file: s.file,
                            line: s.line,
                            col: s.col,
                        })
                        .collect(),
                })
                .collect(),
        }
    }

    /// Register the plain `workflow-support` functions for 6.0.0. Native components
    /// import these; they carry no `backtrace` param. Any host-side backtrace is still
    /// captured when `backtrace_persist` is enabled (`get_host_maybe_capture_backtrace`
    /// with `None`). The backtrace-carrying variants live on
    /// `workflow-support-backtrace` for interpreted (JS) runtimes.
    fn add_to_linker_workflow_support(
        linker: &mut Linker<Self>,
        ifc_fqn: &'static str,
    ) -> Result<(), WasmFileError> {
        let mut inst_workflow_support = linker
            .instance(ifc_fqn)
            .map_err(|err| WasmFileError::linking_error(ifc_fqn, err))?;

        inst_workflow_support
            .func_wrap_async(
                "execution-id-current",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>, (): ()| {
                    Box::new(async move {
                        let host = caller.data_mut();
                        let execution_id =
                            host_exports::latest::obelisk::types::execution::ExecutionId::from(
                                &host.execution_id,
                            );
                        Ok((execution_id,))
                    })
                },
            )
            .map_err(|err| {
                WasmFileError::linking_error("cannot link function execution-id-current", err)
            })?;

        inst_workflow_support
            .func_wrap_async(
                "random-u64",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (min, max_exclusive): (u64, u64)| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, None);
                        let random_u64 = host
                            .random_u64_exclusive(min, max_exclusive, backtrace)
                            .await?;
                        Ok((random_u64,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("cannot link function random-u64", err))?;

        inst_workflow_support
            .func_wrap_async(
                "random-u64-inclusive",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (min, max_inclusive): (u64, u64)| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, None);
                        let random_u64 = host
                            .random_u64_inclusive(min, max_inclusive, backtrace)
                            .await?;
                        Ok((random_u64,))
                    })
                },
            )
            .map_err(|err| {
                WasmFileError::linking_error("linking function random-u64-inclusive", err)
            })?;

        inst_workflow_support
            .func_wrap_async(
                "random-string",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (min_length, max_length_exclusive): (u16, u16)| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, None);
                        let random_string = host
                            .random_string(min_length, max_length_exclusive, backtrace)
                            .await?;
                        Ok((random_string,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function random-string", err))?;

        inst_workflow_support
            .func_wrap_async(
                "sleep",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (schedule_at, name): (ScheduleAtTypes, Option<String>)| {
                    let schedule_at = HistoryEventScheduleAt::from(schedule_at);
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, None);
                        let expires_at = host.sleep_named(schedule_at, name, backtrace).await?;
                        Ok((expires_at,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function sleep", err))?;

        inst_workflow_support
            .func_wrap_async(
                "join-set-create",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>, (): ()| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, None);
                        let resource_js = host.join_set_create_generated(backtrace).await?;
                        Ok((resource_js,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function join-set-create", err))?;

        inst_workflow_support
            .func_wrap_async(
                "join-set-create-named",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (name,): (String,)| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, None);
                        let resource_js = host.join_set_create_named(name, backtrace).await?;
                        Ok((resource_js,))
                    })
                },
            )
            .map_err(|err| {
                WasmFileError::linking_error("linking function join-set-create-named", err)
            })?;

        inst_workflow_support
            .func_wrap_async(
                "join-set-close",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (join_set_resource,): (Resource<JoinSetId>,)| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, None);
                        host.join_set_close_resource(join_set_resource, backtrace)
                            .await?;
                        Ok(())
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function join-set-close", err))?;

        // submit-json: func(join-set, function, params) -> result<execution-id, submit-json-error>
        inst_workflow_support
            .func_wrap_async(
                "submit-json",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (join_set_resource, function, params): (
                    Resource<JoinSetId>,
                    typesTypes::execution::Function,
                    String,
                )| {
                    Box::new(async move {
                        use latest::obelisk::workflow::workflow_support::SubmitJsonError;
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, None);
                        let join_set_id = host.resource_to_join_set_id(&join_set_resource)?.clone();
                        let ffqn = match FunctionFqn::try_from_tuple(
                            &function.interface_name,
                            &function.function_name,
                        ) {
                            Ok(ffqn) => ffqn,
                            Err(err) => {
                                let wit_result =
                                    Err(SubmitJsonError::FfqnParsingError(err.to_string()));
                                return Ok((wit_result,));
                            }
                        };
                        let wit_result = host
                            .submit_json(join_set_id, ffqn, params, backtrace)
                            .await?;
                        Ok((wit_result,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function submit-json", err))?;

        // get-result-json: func(execution-id) -> result<result<option<string>, option<string>>, get-result-json-error>
        inst_workflow_support
            .func_wrap(
                "get-result-json",
                move |caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (execution_id,): (ExecutionIdTypes,)| {
                    use concepts::ExecutionId;
                    use latest::obelisk::workflow::workflow_support::GetResultJsonError;
                    let host = caller.data();
                    let execution_id = match ExecutionId::try_from(execution_id) {
                        Ok(ExecutionId::Derived(derived)) => derived,
                        Ok(ExecutionId::TopLevel(_)) => {
                            return Ok((Err(GetResultJsonError::ExecutionIdParsingError(
                                "must not be a top-level execution id".to_string(),
                            )),));
                        }
                        Err(err) => {
                            return Ok((Err(GetResultJsonError::ExecutionIdParsingError(
                                err.to_string(),
                            )),));
                        }
                    };
                    let wit_result = host.get_result_json(&execution_id);
                    Ok((wit_result,))
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function get-result-json", err))?;

        // get-execution-failure-kind: func(execution-id) -> result<option<execution-failure-kind>, get-result-json-error>
        inst_workflow_support
            .func_wrap(
                "get-execution-failure-kind",
                move |caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (execution_id,): (ExecutionIdTypes,)| {
                    use concepts::ExecutionId;
                    use latest::obelisk::workflow::workflow_support::GetResultJsonError;
                    let host = caller.data();
                    let execution_id = match ExecutionId::try_from(execution_id) {
                        Ok(ExecutionId::Derived(derived)) => derived,
                        Ok(ExecutionId::TopLevel(_)) => {
                            return Ok((Err(GetResultJsonError::ExecutionIdParsingError(
                                "must not be a top-level execution id".to_string(),
                            )),));
                        }
                        Err(err) => {
                            return Ok((Err(GetResultJsonError::ExecutionIdParsingError(
                                err.to_string(),
                            )),));
                        }
                    };
                    let wit_result = host.get_execution_failure_kind(&execution_id);
                    Ok((wit_result,))
                },
            )
            .map_err(|err| {
                WasmFileError::linking_error("linking function get-execution-failure-kind", err)
            })?;

        // last-direct-call-id: func() -> option<execution-id>
        inst_workflow_support
            .func_wrap(
                "last-direct-call-id",
                move |caller: wasmtime::StoreContextMut<'_, WorkflowCtx>, (): ()| {
                    let host = caller.data();
                    Ok((host.last_direct_call_id_wit(),))
                },
            )
            .map_err(|err| {
                WasmFileError::linking_error("linking function last-direct-call-id", err)
            })?;

        // last-oneoff-id: func() -> option<response-id>
        inst_workflow_support
            .func_wrap(
                "last-oneoff-id",
                move |caller: wasmtime::StoreContextMut<'_, WorkflowCtx>, (): ()| {
                    let host = caller.data();
                    Ok((host.last_oneoff_id_wit(),))
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function last-oneoff-id", err))?;

        // execution-id-generate: func() -> execution-id
        inst_workflow_support
            .func_wrap_async(
                "execution-id-generate",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>, (): ()| {
                    Box::new(async move {
                        let (host, wasm_backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, None);
                        let execution_id = host.execution_id_generate(wasm_backtrace).await?;
                        Ok((typesTypes::execution::ExecutionId {
                            id: execution_id.to_string(),
                        },))
                    })
                },
            )
            .map_err(|err| {
                WasmFileError::linking_error("linking function execution-id-generate", err)
            })?;

        // schedule-json: func(execution-id, schedule-at, function, params) -> result<_, schedule-json-error>
        inst_workflow_support
            .func_wrap_async(
                "schedule-json",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (execution_id, schedule_at, function, params): (
                    typesTypes::execution::ExecutionId,
                    ScheduleAtTypes,
                    typesTypes::execution::Function,
                    String,
                )| {
                    let schedule_at = HistoryEventScheduleAt::from(schedule_at);
                    Box::new(async move {
                        use latest::obelisk::workflow::workflow_support::ScheduleJsonError;
                        let (host, wasm_backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, None);
                        let execution_id = match ExecutionId::try_from(execution_id) {
                            Ok(id) => id,
                            Err(err) => {
                                let msg = format!("schedule-json: invalid execution ID: {err}");
                                host.error(msg.clone()).await;
                                return Ok((Err(ScheduleJsonError::FfqnParsingError(msg)),));
                            }
                        };
                        let ffqn = match FunctionFqn::try_from_tuple(
                            &function.interface_name,
                            &function.function_name,
                        ) {
                            Ok(ffqn) => ffqn,
                            Err(err) => {
                                let msg = format!("schedule-json: invalid function name: {err}");
                                host.error(msg.clone()).await;
                                return Ok((Err(ScheduleJsonError::FfqnParsingError(msg)),));
                            }
                        };
                        let wit_result = host
                            .schedule_json(execution_id, ffqn, params, schedule_at, wasm_backtrace)
                            .await?;
                        Ok((wit_result,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function schedule-json", err))?;

        // call-json: func(function, params) -> result<result<option<string>, option<string>>, schedule-json-error>
        inst_workflow_support
            .func_wrap_async(
                "call-json",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (function, params): (typesTypes::execution::Function, String)| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, None);
                        let ffqn = match FunctionFqn::try_from_tuple(
                            &function.interface_name,
                            &function.function_name,
                        ) {
                            Ok(ffqn) => ffqn,
                            Err(err) => {
                                use latest::obelisk::workflow::workflow_support::ScheduleJsonError;
                                let msg = format!("call-json: invalid function name: {err}");
                                host.error(msg.clone()).await;
                                return Ok((Err(ScheduleJsonError::FfqnParsingError(msg)),));
                            }
                        };
                        let wit_result = host.call_json(ffqn, params, backtrace).await?;
                        Ok((wit_result,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function call-json", err))?;

        // stub-json: func(execution-id, result-json) -> result<_, stub-json-error>
        inst_workflow_support
            .func_wrap_async(
                "stub-json",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (execution_id, result_json): (
                    typesTypes::execution::ExecutionId,
                    String,
                )| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, None);
                        let wit_result = host
                            .stub_json(execution_id, result_json, backtrace)
                            .await
                            .map_err(wasmtime::Error::new)?; // Wraps `WorkflowFunctionError` with anyhow to be unwrapped by workflow_worker
                        Ok((wit_result,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function stub-json", err))?;

        // submit-delay: func(join-set, timeout) -> delay-id
        inst_workflow_support
            .func_wrap_async(
                "submit-delay",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (join_set_resource, schedule_at): (Resource<JoinSetId>, ScheduleAtTypes)| {
                    let schedule_at = HistoryEventScheduleAt::from(schedule_at);
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, None);
                        let join_set_id = host.resource_to_join_set_id(&join_set_resource)?.clone();
                        let delay_id: DelayIdTypes = host
                            .submit_delay(join_set_id, schedule_at, backtrace)
                            .await
                            .map_err(wasmtime::Error::new)?;
                        Ok((delay_id,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function submit-delay", err))?;

        // join-next: func(join-set) -> result<result<option<string>, option<string>>, join-next-error>
        inst_workflow_support
            .func_wrap_async(
                "join-next",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (join_set_resource,): (Resource<JoinSetId>,)| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, None);
                        let join_set_id = host.resource_to_join_set_id(&join_set_resource)?.clone();
                        let res = host
                            .join_next(join_set_id, backtrace)
                            .await
                            .map_err(wasmtime::Error::new)?;
                        Ok((res,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function join-next", err))?;

        // join-next-try: func(join-set) -> result<result<option<string>, option<string>>, join-next-try-error>
        inst_workflow_support
            .func_wrap_async(
                "join-next-try",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (join_set_resource,): (Resource<JoinSetId>,)| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, None);
                        let join_set_id = host.resource_to_join_set_id(&join_set_resource)?.clone();
                        let res = host
                            .join_next_try(join_set_id, backtrace)
                            .await
                            .map_err(wasmtime::Error::new)?;
                        Ok((res,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function join-next-try", err))?;

        Ok(())
    }

    /// Register the backtrace-carrying `workflow-support-backtrace` functions for
    /// 6.0.0. Only interpreted runtimes (the JS runtime) import these; they supply a
    /// source-level `backtrace`. The event-persisting subset only (getters and id
    /// accessors live solely on the plain `workflow-support` interface).
    fn add_to_linker_workflow_support_backtrace(
        linker: &mut Linker<Self>,
        ifc_fqn: &'static str,
    ) -> Result<(), WasmFileError> {
        let mut inst_workflow_support = linker
            .instance(ifc_fqn)
            .map_err(|err| WasmFileError::linking_error(ifc_fqn, err))?;

        inst_workflow_support
            .func_wrap_async(
                "random-u64",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (min, max_exclusive, wit_backtrace): (
                    u64,
                    u64,
                    Option<typesTypes::backtrace::WasmBacktrace>,
                )| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, wit_backtrace);
                        let random_u64 = host
                            .random_u64_exclusive(min, max_exclusive, backtrace)
                            .await?;
                        Ok((random_u64,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("cannot link function random-u64", err))?;

        inst_workflow_support
            .func_wrap_async(
                "random-u64-inclusive",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (min, max_inclusive, wit_backtrace): (
                    u64,
                    u64,
                    Option<typesTypes::backtrace::WasmBacktrace>,
                )| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, wit_backtrace);
                        let random_u64 = host
                            .random_u64_inclusive(min, max_inclusive, backtrace)
                            .await?;
                        Ok((random_u64,))
                    })
                },
            )
            .map_err(|err| {
                WasmFileError::linking_error("linking function random-u64-inclusive", err)
            })?;

        inst_workflow_support
            .func_wrap_async(
                "random-string",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (min_length, max_length_exclusive, wit_backtrace): (
                    u16,
                    u16,
                    Option<typesTypes::backtrace::WasmBacktrace>,
                )| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, wit_backtrace);
                        let random_string = host
                            .random_string(min_length, max_length_exclusive, backtrace)
                            .await?;
                        Ok((random_string,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function random-string", err))?;

        inst_workflow_support
            .func_wrap_async(
                "sleep",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (schedule_at, name, wit_backtrace): (
                    ScheduleAtTypes,
                    Option<String>,
                    Option<typesTypes::backtrace::WasmBacktrace>,
                )| {
                    let schedule_at = HistoryEventScheduleAt::from(schedule_at);
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, wit_backtrace);
                        let expires_at = host.sleep_named(schedule_at, name, backtrace).await?;
                        Ok((expires_at,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function sleep", err))?;

        inst_workflow_support
            .func_wrap_async(
                "join-set-create",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (wit_backtrace,): (Option<typesTypes::backtrace::WasmBacktrace>,)| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, wit_backtrace);
                        let resource_js = host.join_set_create_generated(backtrace).await?;
                        Ok((resource_js,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function join-set-create", err))?;

        inst_workflow_support
            .func_wrap_async(
                "join-set-create-named",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (name, wit_backtrace): (
                    String,
                    Option<typesTypes::backtrace::WasmBacktrace>,
                )| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, wit_backtrace);
                        let resource_js = host.join_set_create_named(name, backtrace).await?;
                        Ok((resource_js,))
                    })
                },
            )
            .map_err(|err| {
                WasmFileError::linking_error("linking function join-set-create-named", err)
            })?;

        inst_workflow_support
            .func_wrap_async(
                "join-set-close",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (join_set_resource, wit_backtrace): (
                    Resource<JoinSetId>,
                    Option<typesTypes::backtrace::WasmBacktrace>,
                )| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, wit_backtrace);
                        host.join_set_close_resource(join_set_resource, backtrace)
                            .await?;
                        Ok(())
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function join-set-close", err))?;

        // submit-json: func(join-set, function, params, backtrace) -> result<execution-id, submit-json-error>
        inst_workflow_support
            .func_wrap_async(
                "submit-json",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (join_set_resource, function, params, wit_backtrace): (
                    Resource<JoinSetId>,
                    typesTypes::execution::Function,
                    String,
                    Option<typesTypes::backtrace::WasmBacktrace>,
                )| {
                    Box::new(async move {
                        use latest::obelisk::workflow::workflow_support::SubmitJsonError;
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, wit_backtrace);
                        let join_set_id = host.resource_to_join_set_id(&join_set_resource)?.clone();
                        let ffqn = match FunctionFqn::try_from_tuple(
                            &function.interface_name,
                            &function.function_name,
                        ) {
                            Ok(ffqn) => ffqn,
                            Err(err) => {
                                let wit_result =
                                    Err(SubmitJsonError::FfqnParsingError(err.to_string()));
                                return Ok((wit_result,));
                            }
                        };
                        let wit_result = host
                            .submit_json(join_set_id, ffqn, params, backtrace)
                            .await?;
                        Ok((wit_result,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function submit-json", err))?;

        // execution-id-generate: func(backtrace: option<wasm-backtrace>) -> execution-id
        inst_workflow_support
            .func_wrap_async(
                "execution-id-generate",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (wit_backtrace,): (Option<typesTypes::backtrace::WasmBacktrace>,)| {
                    Box::new(async move {
                        let (host, wasm_backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, wit_backtrace);
                        let execution_id = host.execution_id_generate(wasm_backtrace).await?;
                        Ok((typesTypes::execution::ExecutionId {
                            id: execution_id.to_string(),
                        },))
                    })
                },
            )
            .map_err(|err| {
                WasmFileError::linking_error("linking function execution-id-generate", err)
            })?;

        // schedule-json: func(execution-id, schedule-at, function, params, backtrace) -> result<_, schedule-json-error>
        inst_workflow_support
            .func_wrap_async(
                "schedule-json",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (execution_id, schedule_at, function, params, wit_backtrace): (
                    typesTypes::execution::ExecutionId,
                    ScheduleAtTypes,
                    typesTypes::execution::Function,
                    String,
                    Option<typesTypes::backtrace::WasmBacktrace>,
                )| {
                    let schedule_at = HistoryEventScheduleAt::from(schedule_at);
                    Box::new(async move {
                        use latest::obelisk::workflow::workflow_support::ScheduleJsonError;
                        let (host, wasm_backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, wit_backtrace);
                        let execution_id = match ExecutionId::try_from(execution_id) {
                            Ok(id) => id,
                            Err(err) => {
                                let msg = format!("schedule-json: invalid execution ID: {err}");
                                host.error(msg.clone()).await;
                                return Ok((Err(ScheduleJsonError::FfqnParsingError(msg)),));
                            }
                        };
                        let ffqn = match FunctionFqn::try_from_tuple(
                            &function.interface_name,
                            &function.function_name,
                        ) {
                            Ok(ffqn) => ffqn,
                            Err(err) => {
                                let msg = format!("schedule-json: invalid function name: {err}");
                                host.error(msg.clone()).await;
                                return Ok((Err(ScheduleJsonError::FfqnParsingError(msg)),));
                            }
                        };
                        let wit_result = host
                            .schedule_json(execution_id, ffqn, params, schedule_at, wasm_backtrace)
                            .await?;
                        Ok((wit_result,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function schedule-json", err))?;

        // call-json: func(function, params, backtrace) -> result<result<option<string>, option<string>>, schedule-json-error>
        inst_workflow_support
            .func_wrap_async(
                "call-json",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (function, params, wit_backtrace): (
                    typesTypes::execution::Function,
                    String,
                    Option<typesTypes::backtrace::WasmBacktrace>,
                )| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, wit_backtrace);
                        let ffqn = match FunctionFqn::try_from_tuple(
                            &function.interface_name,
                            &function.function_name,
                        ) {
                            Ok(ffqn) => ffqn,
                            Err(err) => {
                                use latest::obelisk::workflow::workflow_support::ScheduleJsonError;
                                let msg = format!("call-json: invalid function name: {err}");
                                host.error(msg.clone()).await;
                                return Ok((Err(ScheduleJsonError::FfqnParsingError(msg)),));
                            }
                        };
                        let wit_result = host.call_json(ffqn, params, backtrace).await?;
                        Ok((wit_result,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function call-json", err))?;

        // stub-json: func(execution-id, result-json, backtrace) -> result<_, stub-json-error>
        inst_workflow_support
            .func_wrap_async(
                "stub-json",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (execution_id, result_json, wit_backtrace): (
                    typesTypes::execution::ExecutionId,
                    String,
                    Option<typesTypes::backtrace::WasmBacktrace>,
                )| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, wit_backtrace);
                        let wit_result = host
                            .stub_json(execution_id, result_json, backtrace)
                            .await
                            .map_err(wasmtime::Error::new)?; // Wraps `WorkflowFunctionError` with anyhow to be unwrapped by workflow_worker
                        Ok((wit_result,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function stub-json", err))?;

        // submit-delay: func(join-set, timeout, backtrace) -> delay-id
        inst_workflow_support
            .func_wrap_async(
                "submit-delay",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (join_set_resource, schedule_at, wit_backtrace): (
                    Resource<JoinSetId>,
                    ScheduleAtTypes,
                    Option<typesTypes::backtrace::WasmBacktrace>,
                )| {
                    let schedule_at = HistoryEventScheduleAt::from(schedule_at);
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, wit_backtrace);
                        let join_set_id = host.resource_to_join_set_id(&join_set_resource)?.clone();
                        let delay_id: DelayIdTypes = host
                            .submit_delay(join_set_id, schedule_at, backtrace)
                            .await
                            .map_err(wasmtime::Error::new)?;
                        Ok((delay_id,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function submit-delay", err))?;

        // join-next: func(join-set, backtrace) -> result<result<option<string>, option<string>>, join-next-error>
        inst_workflow_support
            .func_wrap_async(
                "join-next",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (join_set_resource, wit_backtrace): (
                    Resource<JoinSetId>,
                    Option<typesTypes::backtrace::WasmBacktrace>,
                )| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, wit_backtrace);
                        let join_set_id = host.resource_to_join_set_id(&join_set_resource)?.clone();
                        let res = host
                            .join_next(join_set_id, backtrace)
                            .await
                            .map_err(wasmtime::Error::new)?;
                        Ok((res,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function join-next", err))?;

        // join-next-try: func(join-set, backtrace) -> result<result<option<string>, option<string>>, join-next-try-error>
        inst_workflow_support
            .func_wrap_async(
                "join-next-try",
                move |mut caller: wasmtime::StoreContextMut<'_, WorkflowCtx>,
                      (join_set_resource, wit_backtrace): (
                    Resource<JoinSetId>,
                    Option<typesTypes::backtrace::WasmBacktrace>,
                )| {
                    Box::new(async move {
                        let (host, backtrace) =
                            Self::get_host_maybe_capture_backtrace(&mut caller, wit_backtrace);
                        let join_set_id = host.resource_to_join_set_id(&join_set_resource)?.clone();
                        let res = host
                            .join_next_try(join_set_id, backtrace)
                            .await
                            .map_err(wasmtime::Error::new)?;
                        Ok((res,))
                    })
                },
            )
            .map_err(|err| WasmFileError::linking_error("linking function join-next-try", err))?;

        Ok(())
    }

    pub(crate) async fn join_sets_close_on_finish(&mut self) -> Result<(), ApplyError> {
        debug!("Closing opened join sets");
        self.event_history
            .finalize(&mut *self.db_connection, self.clock_fn.now())
            .await?;
        Ok(())
    }

    fn next_child_id(&mut self, join_set_id: &JoinSetId) -> ExecutionIdDerived {
        let count = self.event_history.execution_count(join_set_id);
        self.execution_id
            .next_level(join_set_id)
            .get_incremented_by(u64::try_from(count).unwrap())
    }

    #[instrument(level = tracing::Level::DEBUG, skip_all, fields(%called_ffqn, ffqn = %imported_fn_call.ffqn()))]
    pub(crate) async fn call_imported_fn(
        &mut self,
        imported_fn_call: ImportedFnCall<'_>,
        called_at: DateTime<Utc>,
        called_ffqn: &FunctionFqn,
    ) -> Result<Val, WorkflowFunctionError> {
        match imported_fn_call {
            ImportedFnCall::Direct(direct) => direct.call_imported_fn(self, called_at).await,
            ImportedFnCall::Schedule(schedule) => {
                schedule
                    .call_imported_fn(self, called_at, called_ffqn)
                    .await
            }
            ImportedFnCall::SubmitExecution(submit) => {
                submit.call_imported_fn(self, called_at).await
            }
            ImportedFnCall::AwaitNext(await_next) => {
                await_next.call_imported_fn(self, called_at).await
            }
            ImportedFnCall::Stub(stub) => stub.call_imported_fn(self, called_at).await,
            ImportedFnCall::Get(get) => Ok(get.call_imported_fn(self)),
        }
    }
}

#[derive(thiserror::Error, Debug)]
enum JoinSetCreateError {
    #[error(transparent)]
    InvalidNameError(InvalidNameError<JoinSetId>),
    #[error(transparent)]
    ApplyError(ApplyError),
    #[error(transparent)]
    ResourceTableError(ResourceTableError),
    #[error("join set name conflict")]
    Conflict,
}

pub(crate) mod workflow_support {
    use super::{
        Schedule, Stub, SubmitChildExecution, WorkflowCtx, WorkflowFunctionError, typesTypes,
    };
    use crate::component_logger::log_activities::obelisk::log::log::Host as LogHost;
    use crate::workflow::event_history::{
        JoinNext, JoinNextTry, Persist, ScheduleIntent, StubIntent, StubIntentErr, StubParams,
        SubmitDelay,
    };
    use crate::workflow::host_exports::latest::obelisk::types::execution::Host as ExecutionIfcHost;
    use crate::workflow::host_exports::latest::obelisk::workflow::workflow_support::JoinNextError;
    use crate::workflow::host_exports::latest::obelisk::workflow::workflow_support::JoinNextTryError as WitJoinNextTryError;
    use crate::workflow::host_exports::{self, latest};
    use crate::workflow::workflow_ctx::{IFC_FQN_WORKFLOW_SUPPORT, JoinSetCreateError};
    use concepts::prefixed_ulid::{ExecutionIdDerived, ExecutionIdTopLevel};
    use concepts::storage::{DbErrorRead, HistoryEventScheduleAt, StubRetVal};
    use concepts::{CHARSET_ALPHANUMERIC, ComponentType, JoinSetId, JoinSetKind, Params};
    use concepts::{ExecutionId, ReturnType, SupportedFunctionReturnValue};
    use concepts::{FunctionFqn, storage};
    use rand::rngs::StdRng;
    use std::sync::Arc;
    use tracing::{error, instrument, trace};
    use val_json::type_wrapper::TypeWrapper;
    use val_json::wast_val_ser::deserialize_value;
    use wasmtime::component::Resource;

    impl ExecutionIfcHost for WorkflowCtx {}

    pub(crate) fn execution_failure_kind_to_wit(
        kind: concepts::ExecutionFailureKind,
    ) -> typesTypes::execution::ExecutionFailureKind {
        use concepts::ExecutionFailureKind as K;
        use typesTypes::execution::ExecutionFailureKind as Wit;
        match kind {
            K::TimedOut => Wit::TimedOut,
            K::NondeterminismDetected => Wit::NondeterminismDetected,
            K::OutOfFuel => Wit::OutOfFuel,
            K::Cancelled => Wit::Cancelled,
            K::Uncategorized => Wit::Uncategorized,
        }
    }

    impl WorkflowCtx {
        pub(crate) async fn join_set_close_resource(
            &mut self,
            resource: Resource<JoinSetId>,
            wasm_backtrace: Option<storage::WasmBacktrace>,
        ) -> wasmtime::Result<()> {
            let join_set_id = self.resource_table.delete(resource)?;
            self.join_set_close(&join_set_id, wasm_backtrace).await
        }

        pub(crate) async fn join_set_close(
            &mut self,
            join_set_id: &JoinSetId,
            wasm_backtrace: Option<storage::WasmBacktrace>,
        ) -> wasmtime::Result<()> {
            self.event_history
                .join_set_close(
                    join_set_id,
                    &mut *self.db_connection,
                    self.clock_fn.now(),
                    wasm_backtrace,
                )
                .await?;
            Ok(())
        }

        pub(crate) async fn random_u64_exclusive(
            &mut self,
            min: u64,
            max_exclusive: u64,
            wasm_backtrace: Option<storage::WasmBacktrace>,
        ) -> wasmtime::Result<u64> {
            self.random_u64_inclusive(
                min,
                max_exclusive.saturating_sub(1), // if zero, it will be an empty range and error
                wasm_backtrace,
            )
            .await
        }

        pub(crate) async fn random_u64_inclusive(
            &mut self,
            min: u64,
            max_inclusive: u64,
            wasm_backtrace: Option<storage::WasmBacktrace>,
        ) -> wasmtime::Result<u64> {
            let range = min..max_inclusive;
            if range.is_empty() {
                // Panic in host function would kill the worker task.
                return Err(wasmtime::Error::msg("range must not be empty"));
            }
            let value = rand::Rng::random_range(&mut self.rng, range);
            Persist::apply_u64(
                value,
                min,
                max_inclusive,
                wasm_backtrace,
                &mut self.event_history,
                &mut *self.db_connection,
                self.clock_fn.now(),
            )
            .await?;
            Ok(value)
        }

        pub(crate) async fn execution_id_generate(
            &mut self,
            wasm_backtrace: Option<storage::WasmBacktrace>,
        ) -> wasmtime::Result<ExecutionIdTopLevel> {
            let execution_id = ExecutionIdTopLevel::from_parts(
                self.execution_id.get_top_level().timestamp_part(),
                self.next_u128(),
            );
            Persist::apply_execution_id(
                &execution_id,
                wasm_backtrace,
                &mut self.event_history,
                &mut *self.db_connection,
                self.clock_fn.now(),
            )
            .await?;
            Ok(execution_id)
        }

        pub(crate) fn generate_random_string(
            rng: &mut StdRng,
            min_length: u16,
            max_length_exclusive: u16,
        ) -> wasmtime::Result<String> {
            let value: String = {
                let range = min_length..max_length_exclusive;
                if range.is_empty() {
                    // Panic in host function would kill the worker task.
                    return Err(wasmtime::Error::msg("range must not be empty"));
                }
                let length = rand::Rng::random_range(rng, range);
                (0..length)
                    .map(|_| {
                        let idx = rand::Rng::random_range(rng, 0..CHARSET_ALPHANUMERIC.len());
                        CHARSET_ALPHANUMERIC
                            .chars()
                            .nth(idx)
                            .expect("idx is < charset.len()")
                    })
                    .collect()
            };
            Ok(value)
        }

        pub(crate) async fn random_string(
            &mut self,
            min_length: u16,
            max_length_exclusive: u16,
            wasm_backtrace: Option<storage::WasmBacktrace>,
        ) -> wasmtime::Result<String> {
            let value =
                Self::generate_random_string(&mut self.rng, min_length, max_length_exclusive)?;
            // Persist
            Persist::apply_string(
                &value,
                u64::from(min_length),
                u64::from(max_length_exclusive),
                wasm_backtrace,
                &mut self.event_history,
                &mut *self.db_connection,
                self.clock_fn.now(),
            )
            .await?;
            Ok(value)
        }

        pub(crate) async fn submit_delay(
            &mut self,
            join_set_id: JoinSetId,
            schedule_at: HistoryEventScheduleAt,
            wasm_backtrace: Option<storage::WasmBacktrace>,
        ) -> Result<typesTypes::execution::DelayId, WorkflowFunctionError> {
            let delay_id = self
                .event_history
                .next_delay_id(&join_set_id, self.db_connection.execution_id());
            let expires_at_if_new =
                schedule_at
                    .as_date_time(self.clock_fn.now())
                    .map_err(|err| {
                        const FFQN: FunctionFqn =
                            FunctionFqn::new_static(IFC_FQN_WORKFLOW_SUPPORT, "submit-delay");

                        WorkflowFunctionError::ImportedFunctionCallError {
                            ffqn: FFQN,
                            reason: "schedule-at conversion error".into(),
                            detail: Some(format!("{err:?}")),
                        }
                    })?;
            let delay_id = SubmitDelay {
                join_set_id,
                delay_id,
                schedule_at,
                expires_at_if_new,
                wasm_backtrace,
            }
            .apply(
                &mut self.event_history,
                &mut *self.db_connection,
                self.clock_fn.now(),
            )
            .await?;
            Ok(typesTypes::execution::DelayId::from(&delay_id))
        }

        pub(crate) async fn sleep_named(
            &mut self,
            schedule_at: HistoryEventScheduleAt,
            name: Option<String>,
            wasm_backtrace: Option<storage::WasmBacktrace>,
        ) -> wasmtime::Result<Result<host_exports::latest::obelisk::types::time::Datetime, ()>>
        {
            Ok(
                match self
                    .persist_sleep(schedule_at, name, wasm_backtrace)
                    .await
                    .map_err(wasmtime::Error::new)?
                {
                    Ok(expires_at) => Ok(
                        host_exports::latest::obelisk::types::time::Datetime::try_from(expires_at)?,
                    ),
                    Err(()) => Err(()),
                },
            )
        }

        pub(crate) async fn join_set_create_named(
            &mut self,
            name: String,
            wasm_backtrace: Option<storage::WasmBacktrace>,
        ) -> wasmtime::Result<
            Result<
                Resource<JoinSetId>,
                host_exports::latest::obelisk::workflow::workflow_support::JoinSetCreateError,
            >,
        > {
            match self
                .persist_join_set_with_kind(
                    name,
                    JoinSetKind::Named,
                    wasm_backtrace,
                )
                .await
            {
                Ok(resource) => Ok(Ok(resource)),
                Err(JoinSetCreateError::InvalidNameError(err)) => {
                    Ok(Err(host_exports::latest::obelisk::workflow::workflow_support::JoinSetCreateError::InvalidName(err.to_string())))
                }
                Err(JoinSetCreateError::Conflict) => {
                    Ok(Err(host_exports::latest::obelisk::workflow::workflow_support::JoinSetCreateError::Conflict))
                }
                Err(JoinSetCreateError::ApplyError(apply_err)) => {
                    // db errors etc
                    Err(wasmtime::Error::new(WorkflowFunctionError::from(apply_err)))
                }
                Err(JoinSetCreateError::ResourceTableError(resource_table_error)) => {
                    // trap
                    Err(wasmtime::Error::new(resource_table_error))
                }
            }
        }

        pub(crate) async fn join_set_create_generated(
            &mut self,
            wasm_backtrace: Option<storage::WasmBacktrace>,
        ) -> wasmtime::Result<Resource<JoinSetId>> {
            let name = self.event_history.next_join_set_name_generated();
            trace!("new_join_set_generated: {name}");
            self.persist_join_set_with_kind(name, JoinSetKind::Generated, wasm_backtrace)
                .await
                .map_err(|err| match err {
                    JoinSetCreateError::InvalidNameError(_) | JoinSetCreateError::Conflict => {
                        unreachable!("generated index has been incremented")
                    }
                    JoinSetCreateError::ApplyError(apply_err) => {
                        wasmtime::Error::new(WorkflowFunctionError::from(apply_err))
                    }
                    JoinSetCreateError::ResourceTableError(resource_table_error) => {
                        wasmtime::Error::new(resource_table_error)
                    }
                })
        }

        /// Turn a processed response id + ok/err bit into the JSON value that
        /// `join-next` returns. Folds in the former `get-result-json` round-trip.
        /// A child id resolves to its stored result; a delay id maps expired to
        /// `Ok(None)` and cancelled to `Err(None)`.
        fn response_outcome_to_json_value(
            &self,
            response_id: &typesTypes::execution::ResponseId,
            bit: Result<(), ()>,
        ) -> Result<Option<String>, Option<String>> {
            match response_id {
                typesTypes::execution::ResponseId::ExecutionId(exec_id) => {
                    let Ok(ExecutionId::Derived(derived)) = ExecutionId::try_from(exec_id.clone())
                    else {
                        unreachable!("join-next child response id is always a derived execution id")
                    };
                    self.get_result_json(&derived)
                        .expect("response processed by join-next must be retrievable")
                }
                typesTypes::execution::ResponseId::DelayId(_) => match bit {
                    Ok(()) => Ok(None),   // delay expired
                    Err(()) => Err(None), // delay cancelled
                },
            }
        }

        pub(crate) async fn join_next(
            &mut self,
            join_set_id: JoinSetId,
            wasm_backtrace: Option<storage::WasmBacktrace>,
        ) -> Result<
            Result<Result<Option<String>, Option<String>>, JoinNextError>,
            WorkflowFunctionError,
        > {
            let outcome = JoinNext {
                join_set_id,
                wasm_backtrace,
            }
            .apply(
                &mut self.event_history,
                &mut *self.db_connection,
                self.clock_fn.now(),
            )
            .await?;
            Ok(match outcome {
                Ok((response_id, bit)) => {
                    Ok(self.response_outcome_to_json_value(&response_id, bit))
                }
                Err(err) => Err(err),
            })
        }

        pub(crate) async fn join_next_try(
            &mut self,
            join_set_id: JoinSetId,
            wasm_backtrace: Option<storage::WasmBacktrace>,
        ) -> Result<
            Result<Result<Option<String>, Option<String>>, WitJoinNextTryError>,
            WorkflowFunctionError,
        > {
            let outcome = JoinNextTry {
                join_set_id,
                wasm_backtrace,
            }
            .apply(
                &mut self.event_history,
                &mut *self.db_connection,
                self.clock_fn.now(),
            )
            .await?;
            Ok(match outcome {
                Ok((response_id, bit)) => {
                    Ok(self.response_outcome_to_json_value(&response_id, bit))
                }
                Err(err) => Err(err),
            })
        }

        /// Failure kind of a processed child response (additive to the err value).
        pub(crate) fn get_execution_failure_kind(
            &self,
            child_execution_id: &ExecutionIdDerived,
        ) -> Result<
            Option<typesTypes::execution::ExecutionFailureKind>,
            latest::obelisk::workflow::workflow_support::GetResultJsonError,
        > {
            Ok(self
                .event_history
                .get_processed_response_failure_kind(child_execution_id)?
                .map(execution_failure_kind_to_wit))
        }

        pub(crate) fn last_direct_call_id_wit(&self) -> Option<typesTypes::execution::ExecutionId> {
            self.event_history
                .last_direct_call_id()
                .map(typesTypes::execution::ExecutionId::from)
        }

        pub(crate) fn last_oneoff_id_wit(&self) -> Option<typesTypes::execution::ResponseId> {
            self.event_history
                .last_oneoff_id()
                .cloned()
                .map(typesTypes::execution::ResponseId::from)
        }

        pub(crate) fn join_set_last_id_wit(
            &self,
            join_set_id: &JoinSetId,
        ) -> Option<typesTypes::execution::ResponseId> {
            self.event_history
                .last_response_id(join_set_id)
                .cloned()
                .map(typesTypes::execution::ResponseId::from)
        }

        /// Submit a child execution request with JSON-serialized parameters.
        /// The `fn_registry` lookup result is persisted as an intent for determinism.
        pub(crate) async fn submit_json(
            &mut self,
            join_set_id: JoinSetId,
            target_ffqn: FunctionFqn,
            params_json: String,
            wasm_backtrace: Option<storage::WasmBacktrace>,
        ) -> wasmtime::Result<
            Result<
                typesTypes::execution::ExecutionId,
                latest::obelisk::workflow::workflow_support::SubmitJsonError,
            >,
        > {
            use concepts::storage::ChildExecutionRequestError;
            use latest::obelisk::workflow::workflow_support::SubmitJsonError;

            // Return errors that do not depend on state directly without persisting
            let params_json = match serde_json::from_str(&params_json) {
                Ok(serde_json::Value::Array(params)) => params,
                Ok(_other) => {
                    return Ok(Err(SubmitJsonError::TypeCheckError(
                        "params must be a json array".to_string(),
                    )));
                }
                Err(err) => {
                    return Ok(Err(SubmitJsonError::TypeCheckError(format!(
                        "cannot parse params as JSON array: {err}"
                    ))));
                }
            };

            // Compute intent from fn_registry lookup
            let intent = self.get_submit_child_intent(&target_ffqn, params_json);

            // Get the child execution ID
            let child_execution_id = self.next_child_id(&join_set_id);

            // Submit the child execution
            let called_at = self.clock_fn.now();
            let submit = SubmitChildExecution {
                target_ffqn,
                join_set_id,
                child_execution_id: child_execution_id.clone(),
                intent,
                wasm_backtrace,
            };

            let result = submit
                .apply(&mut self.event_history, &mut *self.db_connection, called_at)
                .await
                .map_err(wasmtime::Error::new)?; // Wraps `WorkflowFunctionError` with anyhow to be unwrapped by workflow_worker

            // Return stored result from apply
            match result {
                Ok(()) => Ok(Ok(typesTypes::execution::ExecutionId {
                    id: child_execution_id.to_string(),
                })),
                Err(ChildExecutionRequestError::FunctionNotFound) => {
                    Ok(Err(SubmitJsonError::FunctionNotFound))
                }
                Err(ChildExecutionRequestError::TypeCheckError(msg)) => {
                    Ok(Err(SubmitJsonError::TypeCheckError(msg)))
                }
            }
        }

        /// Compute the intent for submit-json from `fn_registry` lookup.
        fn get_submit_child_intent(
            &self,
            target_ffqn: &FunctionFqn,
            params_json: Vec<serde_json::Value>,
        ) -> crate::workflow::event_history::SubmitChildIntent {
            use crate::workflow::event_history::SubmitChildIntent;
            use concepts::storage::ChildExecutionRequestError;

            // Look up the function in the registry
            let Some((fn_metadata, fn_component_id)) = self
                .event_history
                .fn_registry
                .get_by_exported_function(target_ffqn)
            else {
                return SubmitChildIntent::Err(ChildExecutionRequestError::FunctionNotFound);
            };
            assert_eq!(
                None, fn_metadata.extension,
                "get_by_exported_function must not return extended functions"
            );
            let params = match Params::from_json_values(
                Arc::from(params_json),
                fn_metadata
                    .parameter_types
                    .iter()
                    .map(|param_type| &param_type.type_wrapper),
            ) {
                Ok(params) => params,
                Err(err) => {
                    return SubmitChildIntent::Err(ChildExecutionRequestError::TypeCheckError(
                        format!("params type checking failed: {err}"),
                    ));
                }
            };

            SubmitChildIntent::Ok {
                fn_component_id,
                params,
            }
        }

        /// Schedule an execution request with JSON-serialized parameters.
        /// The `fn_registry` lookup result is persisted as an intent for determinism.
        pub(crate) async fn schedule_json(
            &mut self,
            execution_id: ExecutionId,
            target_ffqn: FunctionFqn,
            params_json: String,
            schedule_at: HistoryEventScheduleAt,
            wasm_backtrace: Option<storage::WasmBacktrace>,
        ) -> wasmtime::Result<
            Result<(), latest::obelisk::workflow::workflow_support::ScheduleJsonError>,
        > {
            use concepts::storage::ScheduleRequestError;
            use latest::obelisk::workflow::workflow_support::ScheduleJsonError;

            let called_at = self.clock_fn.now();
            let scheduled_at_if_new = schedule_at.as_date_time(called_at).map_err(|err| {
                wasmtime::Error::msg(format!("schedule-at conversion error: {err:?}"))
            })?;

            // Return errors that do not depend on state directly without persisting
            let params_json = match serde_json::from_str(&params_json) {
                Ok(serde_json::Value::Array(params)) => params,
                Ok(_other) => {
                    let msg = "schedule-json: params must be a json array".to_string();
                    self.error(msg.clone()).await;
                    return Ok(Err(ScheduleJsonError::TypeCheckError(msg)));
                }
                Err(err) => {
                    let msg = format!("schedule-json: cannot parse params as JSON array: {err}");
                    self.error(msg.clone()).await;
                    return Ok(Err(ScheduleJsonError::TypeCheckError(msg)));
                }
            };

            // Compute intent from fn_registry lookup
            let intent = self.get_schedule_intent(&target_ffqn, params_json);

            let result = Schedule {
                schedule_at,
                scheduled_at_if_new,
                execution_id,
                ffqn: target_ffqn,
                intent,
                wasm_backtrace,
            }
            .apply(&mut self.event_history, &mut *self.db_connection, called_at)
            .await
            .map_err(wasmtime::Error::new)?; // Wraps `WorkflowFunctionError` with anyhow to be unwrapped by workflow_worker

            match result {
                Ok(()) => Ok(Ok(())),
                Err(ScheduleRequestError::FunctionNotFound) => {
                    self.error("schedule-json: function not found".to_string())
                        .await;
                    Ok(Err(ScheduleJsonError::FunctionNotFound))
                }
                Err(ScheduleRequestError::TypeCheckError(msg)) => {
                    self.error(format!("schedule-json: type check error: {msg}"))
                        .await;
                    Ok(Err(ScheduleJsonError::TypeCheckError(msg)))
                }
            }
        }

        /// Compute the intent for schedule-json from `fn_registry` lookup.
        fn get_schedule_intent(
            &self,
            target_ffqn: &FunctionFqn,
            params: Vec<serde_json::Value>,
        ) -> ScheduleIntent {
            use concepts::storage::ScheduleRequestError;

            // Look up the function in the registry
            let Some((fn_metadata, fn_component_id)) = self
                .event_history
                .fn_registry
                .get_by_exported_function(target_ffqn)
            else {
                return ScheduleIntent::Err(ScheduleRequestError::FunctionNotFound);
            };
            assert_eq!(
                None, fn_metadata.extension,
                "get_by_exported_function must not return extended functions"
            );
            let params = match Params::from_json_values(
                Arc::from(params),
                fn_metadata
                    .parameter_types
                    .iter()
                    .map(|param_type| &param_type.type_wrapper),
            ) {
                Ok(params) => params,
                Err(err) => {
                    return ScheduleIntent::Err(ScheduleRequestError::TypeCheckError(format!(
                        "params type checking failed: {err}"
                    )));
                }
            };
            ScheduleIntent::Ok {
                fn_component_id,
                params,
            }
        }

        #[instrument(skip_all, fields(%target_execution_id))]
        async fn get_stub_intent_and_params(
            &mut self,
            target_execution_id: ExecutionIdDerived,
            retval: String,
        ) -> Result<(StubIntent, StubParams), DbErrorRead> {
            // Look up the target function's FFQN
            let target_ffqn = match self
                .db_connection
                .get_stub_create_request(&ExecutionId::Derived(target_execution_id.clone()))
                .await
            {
                Ok(create_req) => create_req.ffqn.clone(),
                Err(DbErrorRead::NotFound) => {
                    return Ok((
                        StubIntent::Err(StubIntentErr::ExecutionNotFound),
                        StubParams {
                            target_execution_id,
                            retval_hash: StubRetVal::Untyped(retval).hash(),
                        },
                    ));
                }
                Err(db_err) => {
                    return Err(db_err); // intermittent error
                }
            };

            // Get the function metadata to determine the return type
            let Some((fn_metadata, fn_component_id)) = self
                .event_history
                .fn_registry
                .get_by_exported_function(&target_ffqn)
            else {
                return Ok((
                    StubIntent::Err(StubIntentErr::TypeCheckError(format!(
                        "function {target_ffqn} not found in registry"
                    ))),
                    StubParams {
                        target_execution_id,
                        retval_hash: StubRetVal::Untyped(retval).hash(),
                    },
                ));
            };

            if fn_component_id.component_type != ComponentType::ActivityStub {
                return Ok((
                    StubIntent::Err(StubIntentErr::TypeCheckError(format!(
                        "cannot stub an execution of type {}",
                        fn_component_id.component_type
                    ))),
                    StubParams {
                        target_execution_id,
                        retval_hash: StubRetVal::Untyped(retval).hash(),
                    },
                ));
            }

            // Get the return type
            let ReturnType::Extendable(return_type) = fn_metadata.return_type else {
                error!(
                    "Database contains execution with a non-extendable return type {fn_component_id} {fn_metadata:?}",
                );
                return Ok((
                    StubIntent::Err(StubIntentErr::TypeCheckError(
                        "target function's return type is not extendable".to_string(),
                    )),
                    StubParams {
                        target_execution_id,
                        retval_hash: StubRetVal::Untyped(retval).hash(),
                    },
                ));
            };

            // Parse the JSON result into a SupportedFunctionReturnValue
            let retval_parsed = {
                let retval_parsed = match serde_json::from_str(&retval) {
                    Ok(ok) => ok,
                    Err(err) => {
                        return Ok((
                            StubIntent::Err(StubIntentErr::TypeCheckError(format!(
                                "cannot parse stubbed return value: {err}"
                            ))),
                            StubParams {
                                target_execution_id,
                                retval_hash: StubRetVal::Untyped(retval).hash(),
                            },
                        ));
                    }
                };
                let type_wrapper = TypeWrapper::from(return_type.type_wrapper_tl);
                let retval_parsed = match deserialize_value(&retval_parsed, type_wrapper) {
                    Ok(ok) => ok,
                    Err(err) => {
                        return Ok((
                            StubIntent::Err(StubIntentErr::TypeCheckError(format!(
                                "cannot type check stubbed return value: {err}"
                            ))),
                            StubParams {
                                target_execution_id,
                                retval_hash: StubRetVal::Untyped(retval).hash(),
                            },
                        ));
                    }
                };
                SupportedFunctionReturnValue::from_wast_val_with_type(retval_parsed)
                    .expect("checked that ffqn is no-ext, return type must be compatible")
            };
            Ok((
                StubIntent::StubTypeChecked(retval_parsed),
                StubParams {
                    target_execution_id,
                    retval_hash: StubRetVal::Untyped(retval).hash(),
                },
            ))
        }

        /// Write a stub response with a JSON-serialized result for an `activity_stub` execution.
        pub(crate) async fn stub_json(
            &mut self,
            target_execution_id: typesTypes::execution::ExecutionId,
            retval: String,
            wasm_backtrace: Option<storage::WasmBacktrace>,
        ) -> Result<
            Result<(), latest::obelisk::workflow::workflow_support::StubJsonError>,
            WorkflowFunctionError,
        > {
            use latest::obelisk::workflow::workflow_support::StubJsonError;

            // Parse the execution ID
            // Error is not persisted, as it is not dependent on state in db or fn registry.
            let target_execution_id = match ExecutionId::try_from(target_execution_id) {
                Ok(ExecutionId::Derived(derived)) => derived,
                Ok(ExecutionId::TopLevel(_)) => {
                    let msg = "stub-json: execution-id must be a derived (child) execution ID, not a top-level one"
                        .to_string();
                    self.error(msg.clone()).await;
                    return Ok(Err(StubJsonError::ExecutionIdParsingError(msg)));
                }
                Err(err) => {
                    let msg = format!("stub-json: cannot parse execution-id: {err}");
                    self.error(msg.clone()).await;
                    return Ok(Err(StubJsonError::ExecutionIdParsingError(msg)));
                }
            };

            let called_at = self.clock_fn.now();
            let (intent, params) = self
                .get_stub_intent_and_params(target_execution_id, retval)
                .await
                .map_err(|err| WorkflowFunctionError::DbError(err.into()))?;

            // Apply the stub

            let stub_result = Stub {
                intent,
                params,
                wasm_backtrace,
            }
            .apply(&mut self.event_history, &mut *self.db_connection, called_at)
            .await?;

            match stub_result {
                Ok(()) => Ok(Ok(())),
                Err(err) => Ok(Err(StubJsonError::from(err))),
            }
        }

        /// Call a function and block until the result is available.
        /// Equivalent to join-set-create + submit-json + join-next + close, but with a single backtrace entry.
        pub(crate) async fn call_json(
            &mut self,
            target_ffqn: FunctionFqn,
            params_json: String,
            wasm_backtrace: Option<storage::WasmBacktrace>,
        ) -> wasmtime::Result<
            Result<
                Result<Option<String>, Option<String>>,
                latest::obelisk::workflow::workflow_support::ScheduleJsonError,
            >,
        > {
            use crate::workflow::event_history::{OneOffChildExecutionRequest, SubmitChildIntent};
            use concepts::storage::ChildExecutionRequestError;
            use latest::obelisk::workflow::workflow_support::ScheduleJsonError;

            // Parse params, returning early on malformed input without touching state
            let params_json_arr = match serde_json::from_str(&params_json) {
                Ok(serde_json::Value::Array(params)) => params,
                Ok(_other) => {
                    let msg = "call-json: params must be a json array".to_string();
                    self.error(msg.clone()).await;
                    return Ok(Err(ScheduleJsonError::TypeCheckError(msg)));
                }
                Err(err) => {
                    let msg = format!("call-json: cannot parse params as JSON array: {err}");
                    self.error(msg.clone()).await;
                    return Ok(Err(ScheduleJsonError::TypeCheckError(msg)));
                }
            };

            // Look up function and type-check params (same logic as submit_json)
            let intent = self.get_submit_child_intent(&target_ffqn, params_json_arr);
            let (fn_component_id, params) = match intent {
                SubmitChildIntent::Ok {
                    fn_component_id,
                    params,
                } => (fn_component_id, params),
                SubmitChildIntent::Err(ChildExecutionRequestError::FunctionNotFound) => {
                    self.error(format!("call-json: function not found: {target_ffqn}"))
                        .await;
                    return Ok(Err(ScheduleJsonError::FunctionNotFound));
                }
                SubmitChildIntent::Err(ChildExecutionRequestError::TypeCheckError(msg)) => {
                    self.error(format!("call-json: type check error: {msg}"))
                        .await;
                    return Ok(Err(ScheduleJsonError::TypeCheckError(msg)));
                }
            };

            // Pre-compute the child execution ID using the same logic as OneOffChildExecutionRequest::apply,
            // so we can look up the result afterwards.
            let join_set_id = self
                .event_history
                .next_join_set_one_off_named(&target_ffqn.function_name)
                .expect("function names only contain alphanumeric and dash, no illegal chars");
            let child_execution_id = self.db_connection.execution_id().next_level(&join_set_id);

            let called_at = self.clock_fn.now();
            OneOffChildExecutionRequest::apply(
                target_ffqn,
                fn_component_id,
                params,
                wasm_backtrace,
                &mut self.event_history,
                &mut *self.db_connection,
                called_at,
            )
            .await
            .map_err(wasmtime::Error::new)?;

            // The child result is now in processed responses; convert to JSON.
            let json_result = self
                .get_result_json(&child_execution_id)
                .expect("result must be in processed responses after OneOffChildExecutionRequest");
            Ok(Ok(json_result))
        }

        /// Obtain child execution result as JSON after it has been awaited.
        /// Stateless, never interrupts the execution, same as `-get` extension.
        pub(crate) fn get_result_json(
            &self,
            child_execution_id: &ExecutionIdDerived,
        ) -> Result<
            Result<Option<String>, Option<String>>,
            latest::obelisk::workflow::workflow_support::GetResultJsonError,
        > {
            // Get the result from processed responses (without function type checking)
            let result = self
                .event_history
                .get_processed_response_json(child_execution_id)?;

            Ok(result)
        }
    }
}

fn trace_on_replay(ctx: &mut WorkflowCtx, level: LogLevel, message: String) {
    if ctx.event_history.has_unprocessed_requests()  // This log message already appeared with proper level unless code changed.
    || ctx.is_replay.is_some()
    {
        ctx.component_logger
            .log(LogLevel::Trace, format!("(replay) {message}"));
    } else {
        ctx.component_logger.log(level, message);
    }
}

fn emit_application_log_to_tracing_only(ctx: &WorkflowCtx, level: LogLevel, message: &str) {
    const TARGET: &str = "app";
    ctx.component_logger.span.in_scope(|| match level {
        LogLevel::Trace => trace!(target: TARGET, "{message}"),
        LogLevel::Debug => debug!(target: TARGET, "{message}"),
        LogLevel::Info => info!(target: TARGET, "{message}"),
        LogLevel::Warn => warn!(target: TARGET, "{message}"),
        LogLevel::Error => error!(target: TARGET, "{message}"),
    });
}

fn capture_replay_application_log(ctx: &mut WorkflowCtx, level: LogLevel, message: &str) -> bool {
    if ctx.is_replay != Some(ReplayKind::Unfinished) {
        return false;
    }
    let Some(logs_storage_config) = &ctx.component_logger.logs_storage_config else {
        return false;
    };
    if ctx.event_history.has_unprocessed_requests() || ctx.event_history.has_unprocessed_responses()
    {
        return false;
    }
    if level >= logs_storage_config.min_level {
        return ctx.db_connection.capture_application_log(LogInfoAppendRow {
            execution_id: ctx.component_logger.execution_id.clone(),
            run_id: ctx.component_logger.run_id,
            log_entry: LogEntry::Log {
                created_at: ctx.clock_fn.now(),
                level,
                message: message.to_owned(),
            },
        });
    }
    false
}

impl log_activities::obelisk::log::log::Host for WorkflowCtx {
    async fn trace(&mut self, message: String) {
        if capture_replay_application_log(self, LogLevel::Trace, &message) {
            emit_application_log_to_tracing_only(self, LogLevel::Trace, &message);
        } else {
            trace_on_replay(self, LogLevel::Trace, message);
        }
    }

    async fn debug(&mut self, message: String) {
        if capture_replay_application_log(self, LogLevel::Debug, &message) {
            emit_application_log_to_tracing_only(self, LogLevel::Debug, &message);
        } else {
            trace_on_replay(self, LogLevel::Debug, message);
        }
    }

    async fn info(&mut self, message: String) {
        if capture_replay_application_log(self, LogLevel::Info, &message) {
            emit_application_log_to_tracing_only(self, LogLevel::Info, &message);
        } else {
            trace_on_replay(self, LogLevel::Info, message);
        }
    }

    async fn warn(&mut self, message: String) {
        if capture_replay_application_log(self, LogLevel::Warn, &message) {
            emit_application_log_to_tracing_only(self, LogLevel::Warn, &message);
        } else {
            trace_on_replay(self, LogLevel::Warn, message);
        }
    }

    async fn error(&mut self, message: String) {
        if capture_replay_application_log(self, LogLevel::Error, &message) {
            emit_application_log_to_tracing_only(self, LogLevel::Error, &message);
        } else {
            trace_on_replay(self, LogLevel::Error, message);
        }
    }
}

#[cfg(test)]
pub(crate) mod tests {
    use crate::activity::cancel_registry::CancelRegistry;
    use crate::cancellation_driver;
    use crate::testing_fn_registry::fn_registry_dummy;
    use crate::workflow::caching_db_connection::{CachingBuffer, CachingDbConnection};
    use crate::workflow::deadline_tracker::{
        DeadlineTrackerFactory as _, DeadlineTrackerFactoryTokio,
    };
    use crate::workflow::event_history::ApplyError;
    use crate::workflow::host_exports::SUFFIX_FN_SUBMIT;
    use crate::workflow::workflow_ctx::{
        DirectFnCall, ImportedFnCall, ReplayKind, SubmitExecutionFnCall, WorkerPartialResult,
    };
    use crate::{
        workflow::workflow_ctx::WorkflowCtx, workflow::workflow_worker::JoinNextBlockingStrategy,
    };
    use assert_matches::assert_matches;
    use async_trait::async_trait;
    use chrono::DateTime;
    use concepts::prefixed_ulid::{DEPLOYMENT_ID_DUMMY, ExecutionIdDerived, ExecutorId, RunId};
    use concepts::storage::{
        AppendEventsToExecution, AppendRequest, AppendResponseToExecution, CreateRequest, DbPool,
        ExecutionEvent, ExecutionRequest, HistoryEvent, HistoryEventScheduleAt, JoinSetRequest,
        Locked, PendingState, PendingStateBlockedByJoinSet, PendingStateFinished,
        PendingStateFinishedError, PendingStateFinishedResultKind, PendingStatePendingAt,
    };
    use concepts::storage::{DbPoolCloseable, ExecutionLog};
    use concepts::time::{ClockFn, Now};
    use concepts::{
        ComponentId, ComponentRetryConfig, ExecutionFailureKind, ExecutionMetadata,
        FinishedExecutionFailure, FunctionRegistry, IfcFqnName, JoinSetId, JoinSetKind,
        RETURN_TYPE_DUMMY, SUFFIX_PKG_EXT, SUPPORTED_RETURN_VALUE_OK_EMPTY,
    };
    use concepts::{ExecutionId, FunctionFqn, Params, SupportedFunctionReturnValue};
    use concepts::{FunctionMetadata, ParameterTypes};
    use db_tests::Database;
    use executor::executor::LockingStrategy;
    use executor::worker::{RunFinished, WorkerResultOk};
    use executor::{
        executor::{ExecConfig, ExecTask},
        expired_timers_watcher,
        worker::{Worker, WorkerContext, WorkerResult},
    };
    use hashbrown::HashSet;
    use rand::SeedableRng as _;
    use rand::rngs::StdRng;

    use std::{fmt::Debug, sync::Arc, time::Duration};
    use test_utils::get_seed;
    use test_utils::{arbitrary::UnstructuredHolder, sim_clock::SimClock};
    use tracing::{debug, info, info_span};

    const TICK_SLEEP: Duration = Duration::from_millis(1);
    pub const FFQN_MOCK: FunctionFqn = FunctionFqn::new_static("namespace:pkg/ifc", "fn");

    #[test]
    fn generate_random_string_should_respect_max_length() {
        // regression test
        test_utils::set_up();
        for seed in get_seed() {
            let mut rng = StdRng::seed_from_u64(seed);
            let string =
                WorkflowCtx::generate_random_string(&mut rng, 1, 5 /* max length exclusive */)
                    .expect("range is not empty");
            assert!(!string.is_empty());
            assert!(string.len() < 5);
        }
    }

    #[test]
    fn generate_random_string_should_fail_on_empty_range() {
        // regression test
        test_utils::set_up();
        let mut rng = StdRng::seed_from_u64(0);
        WorkflowCtx::generate_random_string(&mut rng, 1, 1 /* max length exclusive */).unwrap_err();
    }

    impl From<WorkerPartialResult> for WorkerResult {
        fn from(worker_partial_result: WorkerPartialResult) -> Self {
            match worker_partial_result {
                WorkerPartialResult::FatalError(err, version) => {
                    Err(executor::worker::WorkerError::FatalError(err, version))
                }
                WorkerPartialResult::InterruptDbUpdated
                | WorkerPartialResult::ReplayWaitingForResponse => {
                    Ok(WorkerResultOk::DbUpdatedByWorkerOrWatcher)
                }
                WorkerPartialResult::DbError(db_err) => {
                    Err(executor::worker::WorkerError::DbError(db_err))
                }
                WorkerPartialResult::LockExpired | WorkerPartialResult::ExecutorClosing => {
                    unreachable!()
                }
            }
        }
    }

    #[derive(Debug, Clone, arbitrary::Arbitrary)]
    enum WorkflowStep {
        Sleep {
            millis: u32, // Avoid ScheduleAtConversionError::OutOfRangeError
        },
        Call {
            ffqn: FunctionFqn,
        },
        JoinSetCreateNamed {
            join_set_id: JoinSetId,
        },
        SubmitExecution {
            target_ffqn: FunctionFqn,
            join_set_id: JoinSetId,
        },
        SubmitDelay {
            millis: u32, // Avoid ScheduleAtConversionError::OutOfRangeError
            join_set_id: JoinSetId,
        },
        RandomU64 {
            min: u64,
            max_inclusive: u64,
        },
        RandomString {
            min_length: u16,
            max_length_exclusive: u16,
        },
        JoinSetClose {
            join_set_id: JoinSetId,
        },
    }

    impl WorkflowStep {
        fn is_valid(&self) -> bool {
            match self {
                Self::RandomU64 { min, max_inclusive } => min <= max_inclusive,
                Self::RandomString {
                    min_length,
                    max_length_exclusive,
                } => min_length < max_length_exclusive,
                _ => true,
            }
        }
    }

    #[derive(derive_more::Debug)]
    struct WorkflowWorkerMock {
        #[expect(dead_code)]
        ffqn: FunctionFqn, // For debugging
        steps: Vec<WorkflowStep>,
        #[debug(skip)]
        clock_fn: Box<dyn ClockFn>,
        #[debug(skip)]
        db_pool: Arc<dyn DbPool>,
        #[debug(skip)]
        fn_registry: Arc<dyn FunctionRegistry>,
        #[debug(skip)]
        exports: [FunctionMetadata; 1],
    }

    impl WorkflowWorkerMock {
        fn new(
            ffqn: FunctionFqn,
            fn_registry: Arc<dyn FunctionRegistry>,
            steps: Vec<WorkflowStep>,
            clock_fn: Box<dyn ClockFn>,
            db_pool: Arc<dyn DbPool>,
        ) -> Self {
            Self {
                exports: [FunctionMetadata {
                    ffqn: ffqn.clone(),
                    parameter_types: ParameterTypes::default(),
                    return_type: RETURN_TYPE_DUMMY,
                    extension: None,
                    submittable: true,
                }],
                ffqn,
                steps,
                clock_fn,
                db_pool,
                fn_registry,
            }
        }
    }

    #[async_trait]
    impl Worker for WorkflowWorkerMock {
        async fn run(&self, ctx: WorkerContext) -> WorkerResult {
            info!("Starting");
            let seed = ctx.execution_id.random_seed();
            let is_replay = if ctx.event_history.is_empty() {
                None
            } else {
                Some(ReplayKind::Unfinished)
            };
            let join_next_blocking_strategy = JoinNextBlockingStrategy::Interrupt; // Cannot Await: when moving time forward both worker and timers watcher would race.
            let caching_db_connection = CachingDbConnection::new(
                self.db_pool.connection().await.unwrap(),
                ctx.execution_id.clone(),
                CachingBuffer::new(join_next_blocking_strategy),
                ctx.version,
            );

            let cancel_registry = CancelRegistry::new();
            let mut workflow_ctx = WorkflowCtx::new(
                DEPLOYMENT_ID_DUMMY,
                Box::new(caching_db_connection),
                ctx.event_history,
                ctx.responses,
                seed,
                self.clock_fn.clone_box(),
                join_next_blocking_strategy,
                tracing::info_span!("workflow-test"),
                false,
                DeadlineTrackerFactoryTokio::new(Duration::ZERO, self.clock_fn.clone_box())
                    .create(
                        ctx.locked_event.lock_expires_at,
                        tokio::sync::watch::channel(false).1,
                    )
                    .unwrap(),
                self.fn_registry.clone(),
                cancel_registry,
                ctx.locked_event,
                Some(Duration::from_secs(1)), // lock extension
                None,                         // subscription_interruption
                None,                         // logs_storage_config,
                is_replay,
            );
            for step in &self.steps {
                info!("Processing step {step:?}");
                let res = match step {
                    WorkflowStep::Sleep { millis } => workflow_ctx
                        .persist_sleep(
                            HistoryEventScheduleAt::In(Duration::from_millis(u64::from(*millis))),
                            None, // custom name
                            None, // backtrace
                        )
                        .await
                        .map(|_| ()),
                    WorkflowStep::Call { ffqn } => {
                        let (_fn_metadata, fn_component_id) =
                            self.fn_registry.get_by_exported_function(ffqn).expect(
                                "function obtained from `fn_registry.all_exports()` must be found",
                            );
                        workflow_ctx
                            .call_imported_fn(
                                ImportedFnCall::Direct(DirectFnCall {
                                    ffqn: ffqn.clone(),
                                    fn_component_id,
                                    params: &[],
                                    wasm_backtrace: None,
                                }),
                                self.clock_fn.now(),
                                ffqn,
                            )
                            .await
                            .map(|_| ())
                    }

                    WorkflowStep::JoinSetCreateNamed { join_set_id } => {
                        assert_eq!(JoinSetKind::Named, join_set_id.kind);
                        workflow_ctx
                            .join_set_create_named(join_set_id.name.to_string(), None)
                            .await
                            .unwrap()
                            .unwrap();
                        Ok(())
                    }

                    WorkflowStep::SubmitExecution {
                        target_ffqn,
                        join_set_id,
                    } => {
                        let target_ifc = target_ffqn.ifc_fqn.clone();
                        let submit_ffqn = FunctionFqn {
                            ifc_fqn: IfcFqnName::from_parts(
                                target_ifc.namespace(),
                                &format!("{}{SUFFIX_PKG_EXT}", target_ifc.package_name()),
                                target_ifc.ifc_name(),
                                target_ifc.version(),
                            ),
                            function_name: concepts::FnName::from(format!(
                                "{}{}",
                                target_ffqn.function_name, SUFFIX_FN_SUBMIT
                            )),
                        };
                        let (_fn_metadata, target_component_id) = self
                            .fn_registry
                            .get_by_exported_function(target_ffqn)
                            .expect(
                                "function obtained from `fn_registry.all_exports()` must be found",
                            );
                        workflow_ctx
                            .call_imported_fn(
                                ImportedFnCall::SubmitExecution(SubmitExecutionFnCall {
                                    target_ffqn: target_ffqn.clone(),
                                    target_component_id,
                                    join_set_id: join_set_id.clone(),
                                    target_params: &[],
                                    wasm_backtrace: None,
                                }),
                                self.clock_fn.now(),
                                &submit_ffqn,
                            )
                            .await
                            .map(|_| ())
                    }
                    WorkflowStep::SubmitDelay {
                        millis,
                        join_set_id,
                    } => workflow_ctx
                        .submit_delay(
                            join_set_id.clone(),
                            HistoryEventScheduleAt::In(Duration::from_millis(u64::from(*millis))),
                            None,
                        )
                        .await
                        .map(|_| ()),
                    WorkflowStep::RandomU64 { min, max_inclusive } => {
                        let value = workflow_ctx
                            .random_u64_inclusive(*min, *max_inclusive, None)
                            .await
                            .unwrap();
                        assert!(value > *min);
                        assert!(value <= *max_inclusive);
                        Ok(())
                    }
                    WorkflowStep::RandomString {
                        min_length,
                        max_length_exclusive,
                    } => {
                        let value = workflow_ctx
                            .random_string(*min_length, *max_length_exclusive, None)
                            .await
                            .unwrap();
                        assert!(value.len() > *min_length as usize);
                        assert!(value.len() < usize::from(*max_length_exclusive));
                        Ok(())
                    }
                    WorkflowStep::JoinSetClose { join_set_id } => {
                        workflow_ctx
                            .join_set_close(join_set_id, None)
                            .await
                            .unwrap();
                        Ok(())
                    }
                };
                if let Err(err) = res {
                    info!("Sending {err:?}");
                    return err
                        .into_worker_partial_result(workflow_ctx.db_connection.version().clone())
                        .into();
                }
            }
            let res = match workflow_ctx.join_sets_close_on_finish().await {
                Ok(()) => {
                    info!("Finishing");
                    WorkerResult::Ok(WorkerResultOk::RunFinished(RunFinished {
                        retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                        version: workflow_ctx.db_connection.version().clone(),
                        http_client_traces: None,
                    }))
                }
                Err(ApplyError::InterruptDbUpdated) => {
                    info!("Interrupting");
                    return WorkerResult::Ok(WorkerResultOk::DbUpdatedByWorkerOrWatcher);
                }
                other => panic!("Unexpected error: {other:?}"),
            };
            info!("Done");
            res
        }

        fn exported_functions_noext(&self) -> &[FunctionMetadata] {
            &self.exports
        }
    }

    // TODO: verify nondeterminism detection:
    // Start WorkflowWorkerMock, wait until it completes.
    // Copy its execution history to a new database
    // A. Swap two event history items
    // B. Swap two steps in WorkflowWorkerMock
    // C. Add new event history item
    // D. Add new step - needs whole execution history, must be done on another layer
    // E. Remove a step
    // F. Change the final result

    #[tokio::test]
    async fn generate_steps_execute_twice_check_determinism() {
        test_utils::set_up();
        for seed in get_seed() {
            let steps = generate_steps(seed);
            let closure = |steps, mut sim_clock, seed| async move {
                let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
                let mut seedable_rng = StdRng::seed_from_u64(seed);
                let next_u128 = || rand::Rng::random(&mut seedable_rng);
                let res = execute_steps(steps, db_pool.clone(), &mut sim_clock, next_u128).await;
                db_close.close().await;
                res
            };
            println!("Run 1");
            let res1 = closure(steps.clone(), SimClock::epoch(), seed).await;
            println!("Run 2");
            let res2 = closure(steps, SimClock::epoch(), seed).await;
            assert_eq!(res1, res2);
        }
    }

    #[tokio::test]
    async fn generate_steps_execute_keep_responses_reexecute_should_generate_same_exec_log() {
        test_utils::set_up();
        for seed in get_seed() {
            let steps = generate_steps(seed);

            println!("Run 1");
            let (execution_id, execution_log) = {
                let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
                let mut seedable_rng = StdRng::seed_from_u64(seed);
                let next_u128 = || rand::Rng::random(&mut seedable_rng);
                let (execution_id, execution_log) = execute_steps(
                    steps.clone(),
                    db_pool.clone(),
                    &mut SimClock::epoch(),
                    next_u128,
                )
                .await;
                println!("{execution_log:?}");
                db_close.close().await;
                (execution_id, execution_log)
            };
            println!("Run 2");
            let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
            let sim_clock = SimClock::epoch();
            let fn_registry = steps_to_registry(&steps);
            let worker = Arc::new(WorkflowWorkerMock::new(
                FFQN_MOCK,
                fn_registry.clone(),
                steps,
                sim_clock.clone_box(),
                db_pool.clone(),
            ));
            // Create an execution.
            let db_connection = db_pool.connection_test().await.unwrap();
            db_connection
                .create(CreateRequest {
                    created_at: sim_clock.now(),
                    execution_id: execution_id.clone(),
                    ffqn: FFQN_MOCK,
                    params: Params::empty(),
                    parent: None,
                    metadata: concepts::ExecutionMetadata::empty(),
                    scheduled_at: sim_clock.now(),
                    component_id: ComponentId::dummy_activity(),

                    deployment_id: DEPLOYMENT_ID_DUMMY,
                    scheduled_by: None,
                    paused: false,
                })
                .await
                .unwrap();
            let mut responses = execution_log.responses.clone();
            // Expect that the same steps + same responses produce same exec log (lock events and timestamps may differ)
            loop {
                let current_log = db_connection.get(&execution_id).await.unwrap();
                let worker_result = worker
                    .run(WorkerContext {
                        execution_id: execution_id.clone(),
                        metadata: ExecutionMetadata::empty(),
                        component_digest: ComponentId::dummy_activity().component_digest,
                        ffqn: FFQN_MOCK,
                        params: Params::empty(),
                        event_history: current_log.event_history().collect(),
                        responses: execution_log.responses.clone(),
                        parent: None,
                        version: current_log.next_version.clone(),
                        can_be_retried: false,
                        worker_span: info_span!("generate_steps_reexecute"),
                        locked_event: Locked {
                            component_id: ComponentId::dummy_activity(),
                            executor_id: ExecutorId::from_parts(0, 0),
                            deployment_id: DEPLOYMENT_ID_DUMMY,
                            run_id: RunId::from_parts(
                                u64::try_from(sim_clock.now().timestamp_millis()).unwrap(),
                                0,
                            ),
                            lock_expires_at: sim_clock.now() + Duration::from_secs(1),
                            retry_config: ComponentRetryConfig::ZERO,
                        },
                        executor_close_watcher: tokio::sync::watch::channel(false).1,
                    })
                    .await;
                if let WorkerResult::Ok(WorkerResultOk::RunFinished(RunFinished {
                    retval,
                    version,
                    http_client_traces,
                })) = worker_result
                {
                    db_connection
                        .append(
                            execution_id.clone(),
                            version,
                            AppendRequest {
                                created_at: sim_clock.now(),
                                event: ExecutionRequest::Finished {
                                    retval,
                                    http_client_traces,
                                },
                            },
                        )
                        .await
                        .unwrap();
                }
                let mut append_response = |join_set_id| {
                    let response_idx = responses
                        .iter()
                        .position(|response| response.event.event.join_set_id == join_set_id)
                        .expect("response must be found");
                    responses.remove(response_idx)
                };
                match db_connection
                    .get_pending_state(&execution_id)
                    .await
                    .unwrap()
                    .pending_state
                {
                    PendingState::BlockedByJoinSet(PendingStateBlockedByJoinSet {
                        join_set_id,
                        ..
                    }) => {
                        let response = append_response(join_set_id);
                        db_connection
                            .append_response(
                                response.event.created_at,
                                execution_id.clone(),
                                response.event.event,
                            )
                            .await
                            .unwrap();
                    }
                    PendingState::PendingAt(PendingStatePendingAt {
                        scheduled_at,
                        last_lock: _,
                    }) if scheduled_at <= sim_clock.now() => {}
                    PendingState::Finished { .. } => break,
                    pending_state => unreachable!("unexpected {pending_state:?}"),
                }
            }
            let execution_log2 = db_connection.get(&execution_id).await.unwrap();
            println!("{execution_log2:?}");
            assert_eq!(
                skip_locked(execution_log.events.iter()).count(),
                skip_locked(execution_log2.events.iter()).count()
            );
            let sanitize = |mut event| {
                match &mut event {
                    ExecutionRequest::HistoryEvent {
                        event: HistoryEvent::JoinNext { run_expires_at, .. },
                        ..
                    } => {
                        *run_expires_at = DateTime::UNIX_EPOCH;
                    }
                    ExecutionRequest::HistoryEvent {
                        event:
                            HistoryEvent::JoinSetRequest {
                                request: JoinSetRequest::DelayRequest { expires_at, .. },
                                ..
                            },
                    } => {
                        *expires_at = DateTime::UNIX_EPOCH;
                    }
                    _ => {}
                }
                event
            };

            for (idx, (event, event2)) in skip_locked(execution_log.events.iter())
                .cloned()
                .map(sanitize)
                .zip(
                    skip_locked(execution_log2.events.iter())
                        .cloned()
                        .map(sanitize),
                )
                .enumerate()
            {
                println!("Comparing {idx}");
                println!("{event:?}");
                println!("{event2:?}");

                assert_eq!(event, event2, "mismatch at {idx}");
            }
            db_close.close().await;
        }
    }

    #[tokio::test]
    async fn creating_oneoff_and_generated_join_sets_with_same_name_should_work() {
        test_utils::set_up();
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let join_set_id = JoinSetId::new(concepts::JoinSetKind::Named, "".into()).unwrap();
        let steps = vec![
            WorkflowStep::JoinSetCreateNamed {
                join_set_id: join_set_id.clone(),
            },
            WorkflowStep::Call {
                ffqn: FFQN_CHILD_MOCK,
            },
            WorkflowStep::SubmitExecution {
                target_ffqn: FFQN_CHILD_MOCK,
                join_set_id,
            },
        ];
        execute_steps(steps, db_pool.clone(), &mut SimClock::epoch(), || 0).await;
        db_close.close().await;
    }

    #[tokio::test]
    async fn submitting_two_delays_should_work() {
        test_utils::set_up();
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let join_set_id = JoinSetId::new(concepts::JoinSetKind::Named, "".into()).unwrap();

        let steps = vec![
            WorkflowStep::JoinSetCreateNamed {
                join_set_id: join_set_id.clone(),
            },
            WorkflowStep::SubmitDelay {
                millis: 1,
                join_set_id: join_set_id.clone(),
            },
            WorkflowStep::SubmitDelay {
                millis: 10,
                join_set_id,
            },
        ];
        execute_steps(steps, db_pool.clone(), &mut SimClock::epoch(), || 0).await;
        db_close.close().await;
    }

    #[tokio::test]
    async fn submitting_to_closed_join_set_should_be_permanent_execution_error() {
        test_utils::set_up();
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let join_set_id = JoinSetId::new(concepts::JoinSetKind::Named, "foo".into()).unwrap();
        let steps = vec![
            WorkflowStep::JoinSetCreateNamed {
                join_set_id: join_set_id.clone(),
            },
            WorkflowStep::JoinSetClose {
                join_set_id: join_set_id.clone(),
            },
            WorkflowStep::SubmitExecution {
                target_ffqn: FFQN_CHILD_MOCK,
                join_set_id: join_set_id.clone(),
            },
        ];
        let (_, log) = execute_steps(steps, db_pool.clone(), &mut SimClock::epoch(), || 0).await;
        let kind = assert_matches!(
            log.pending_state,
            PendingState::Finished(
                PendingStateFinished {
                    result_kind: PendingStateFinishedResultKind::Err(
                        PendingStateFinishedError::ExecutionFailure(kind)
                    ),
                    ..
                },
            ) => kind
        );
        assert_eq!(ExecutionFailureKind::Uncategorized, kind);
        let (reason, kind, _detail) = assert_matches!(log.as_finished_result().unwrap(),
            SupportedFunctionReturnValue::ExecutionFailure(FinishedExecutionFailure { reason, kind, detail })
            =>(reason, kind, detail));
        assert_eq!(ExecutionFailureKind::Uncategorized, kind);
        assert_eq!(
            Some(format!(
                "constraint violation: join set is not open: `{join_set_id}`"
            )),
            reason
        );
        db_close.close().await;
    }

    const FFQN_CHILD_MOCK: FunctionFqn = FunctionFqn::new_static("namespace:pkg/ifc", "fn-child");

    #[tokio::test]
    async fn check_determinism_closing_multiple_join_sets() {
        const SUBMITS: usize = 10;
        test_utils::set_up();
        let (_guard, db_pool, db_close) = Database::Sqlite.set_up().await;
        let sim_clock = SimClock::new(Now.now());
        let db_connection = db_pool.connection_test().await.unwrap();

        // Create an execution.
        let execution_id = ExecutionId::generate();
        let version = db_connection
            .create(CreateRequest {
                created_at: sim_clock.now(),
                execution_id: execution_id.clone(),
                ffqn: FFQN_MOCK,
                params: Params::empty(),
                parent: None,
                metadata: concepts::ExecutionMetadata::empty(),
                scheduled_at: sim_clock.now(),
                component_id: ComponentId::dummy_activity(),

                deployment_id: DEPLOYMENT_ID_DUMMY,
                scheduled_by: None,
                paused: false,
            })
            .await
            .unwrap();

        let mut steps = Vec::new();
        for i in 0..SUBMITS {
            let join_set_id =
                JoinSetId::new(concepts::JoinSetKind::Named, format!("{i}").into()).unwrap();
            steps.push(WorkflowStep::JoinSetCreateNamed {
                join_set_id: join_set_id.clone(),
            });
            steps.push(WorkflowStep::SubmitExecution {
                target_ffqn: FFQN_CHILD_MOCK,
                join_set_id: join_set_id.clone(),
            });
        }
        let worker = Arc::new(WorkflowWorkerMock::new(
            FFQN_MOCK,
            steps_to_registry(&steps),
            steps,
            sim_clock.clone_box(),
            db_pool.clone(),
        ));
        {
            let mut worker_result = worker
                .run(WorkerContext {
                    execution_id: execution_id.clone(),
                    metadata: ExecutionMetadata::empty(),
                    component_digest: ComponentId::dummy_activity().component_digest,
                    ffqn: FFQN_MOCK,
                    params: Params::empty(),
                    event_history: vec![],
                    responses: vec![],
                    parent: None,
                    version,
                    can_be_retried: false,
                    worker_span: info_span!("check_determinism"),
                    locked_event: Locked {
                        component_id: ComponentId::dummy_activity(),
                        executor_id: ExecutorId::generate(),
                        deployment_id: DEPLOYMENT_ID_DUMMY,
                        run_id: RunId::generate(),
                        lock_expires_at: sim_clock.now() + Duration::from_secs(1),
                        retry_config: ComponentRetryConfig::ZERO,
                    },
                    executor_close_watcher: tokio::sync::watch::channel(false).1,
                })
                .await;
            let cancel_registry = CancelRegistry::new();
            // Run it SUBMITS times to close all join sets.
            for run in 0..SUBMITS {
                info!("Run {run}");
                assert_matches!(
                    worker_result,
                    WorkerResult::Ok(WorkerResultOk::DbUpdatedByWorkerOrWatcher)
                );
                cancellation_driver::tick_test(
                    db_connection.as_ref(),
                    &cancel_registry,
                    sim_clock.now(),
                )
                .await;
                let execution_log = db_connection.get(&execution_id).await.unwrap();
                let closing_join_nexts: hashbrown::HashSet<_> = execution_log
                    .event_history()
                    .filter_map(|(event, _version)| match event {
                        HistoryEvent::JoinNext {
                            closing: true,
                            join_set_id,
                            ..
                        } => Some(join_set_id),
                        _ => None,
                    })
                    .collect();
                assert_eq!(run + 1, closing_join_nexts.len());
                // Cancelled means we block on a join set but add a response => pending now
                assert_matches!(&execution_log.pending_state, PendingState::PendingAt(..));

                info!("Advancing the worker");
                let execution_log = db_connection.get(&execution_id).await.unwrap();
                assert_eq!(run + 1, execution_log.responses.len());
                worker_result = worker
                    .run(WorkerContext {
                        execution_id: execution_id.clone(),
                        metadata: ExecutionMetadata::empty(),
                        component_digest: ComponentId::dummy_activity().component_digest,
                        ffqn: FFQN_MOCK,
                        params: Params::empty(),
                        event_history: execution_log.event_history().collect(),
                        responses: execution_log.responses,
                        parent: None,
                        version: execution_log.next_version,
                        can_be_retried: false,
                        worker_span: info_span!("check_determinism"),
                        locked_event: Locked {
                            component_id: ComponentId::dummy_activity(),
                            executor_id: ExecutorId::generate(),
                            deployment_id: DEPLOYMENT_ID_DUMMY,
                            run_id: RunId::generate(),
                            lock_expires_at: sim_clock.now() + Duration::from_secs(1),
                            retry_config: ComponentRetryConfig::ZERO,
                        },
                        executor_close_watcher: tokio::sync::watch::channel(false).1,
                    })
                    .await;
            }
            let (finished_value, version) = assert_matches!(
                worker_result,
                WorkerResult::Ok(WorkerResultOk::RunFinished(RunFinished { retval, version, http_client_traces:_ })) => (retval, version),
                "should be finished"
            );
            info!("Appending finished result");
            db_connection
                .append(
                    execution_id.clone(),
                    version,
                    AppendRequest {
                        created_at: sim_clock.now(),
                        event: concepts::storage::ExecutionRequest::Finished {
                            retval: finished_value,
                            http_client_traces: None,
                        },
                    },
                )
                .await
                .unwrap();
        }

        let execution_log = db_connection.get(&execution_id).await.unwrap();
        assert_matches!(
            execution_log.pending_state,
            PendingState::Finished(PendingStateFinished {
                result_kind: PendingStateFinishedResultKind::Ok,
                ..
            },)
        );

        // Currently only finished executions can be checked. Otherwise a noop
        // response in `find_matching_atomic` would need to be implemented.
        info!("Run again to test determinism");
        let worker_result = worker
            .run(WorkerContext {
                execution_id: execution_id.clone(),
                metadata: ExecutionMetadata::empty(),
                component_digest: ComponentId::dummy_activity().component_digest,
                ffqn: FFQN_MOCK,
                params: Params::empty(),
                event_history: execution_log.event_history().collect(),
                responses: execution_log.responses.clone(),
                parent: None,
                version: execution_log.next_version.clone(),
                can_be_retried: false,
                worker_span: info_span!("check_determinism"),
                locked_event: Locked {
                    component_id: ComponentId::dummy_activity(),
                    executor_id: ExecutorId::generate(),
                    deployment_id: DEPLOYMENT_ID_DUMMY,
                    run_id: RunId::generate(),
                    lock_expires_at: sim_clock.now() + Duration::from_secs(1),
                    retry_config: ComponentRetryConfig::ZERO,
                },
                executor_close_watcher: tokio::sync::watch::channel(false).1,
            })
            .await;
        assert_matches!(worker_result, WorkerResult::Ok(..), "should be finished");
        assert_eq!(
            execution_log,
            db_connection.get(&execution_id).await.unwrap(),
            "nothing should be written when verifying determinism"
        );
        db_close.close().await;
    }

    fn generate_steps(seed: u64) -> Vec<WorkflowStep> {
        let unstructured_holder = UnstructuredHolder::new(seed);
        let mut unstructured = unstructured_holder.unstructured();
        let mut steps = unstructured
            .arbitrary_iter()
            .unwrap()
            .map(std::result::Result::unwrap)
            .filter(|step: &WorkflowStep| step.is_valid())
            .collect::<Vec<_>>();
        // TODO: the test harness supports a single child/delay request per join set
        let mut ffqns = hashbrown::HashSet::new();
        steps.retain(|step| match step {
            WorkflowStep::Call { ffqn }
            | WorkflowStep::SubmitExecution {
                target_ffqn: ffqn, ..
            } => {
                ffqns.insert(ffqn.clone()) // Retain only the first step for a given ffqn
            }
            _ => true,
        });
        let join_sets_already_existing: HashSet<_> = steps
            .iter()
            .filter_map(|step| {
                if let WorkflowStep::JoinSetCreateNamed { join_set_id } = step {
                    Some(join_set_id.clone())
                } else {
                    None
                }
            })
            .collect();

        let join_sets_added: HashSet<_> = steps
            .iter()
            .filter_map(|step| match step {
                WorkflowStep::SubmitDelay { join_set_id, .. }
                | WorkflowStep::SubmitExecution { join_set_id, .. }
                | WorkflowStep::JoinSetClose { join_set_id } => Some(join_set_id.clone()),
                _ => None,
            })
            .filter(|join_set| {
                // There can be a collision with CreateJoinSetNamed
                !join_sets_already_existing.contains(join_set)
            })
            .collect();
        for join_set_id in join_sets_added {
            steps.insert(0, WorkflowStep::JoinSetCreateNamed { join_set_id });
        }
        // Do not allow using the join set after close
        let mut join_set_closed = HashSet::new();
        steps.retain(|step| match step {
            WorkflowStep::JoinSetClose { join_set_id } => {
                let already_present = join_set_closed.insert(join_set_id.clone());
                !already_present // only first one is retained
            }
            WorkflowStep::SubmitDelay { join_set_id, .. }
            | WorkflowStep::SubmitExecution { join_set_id, .. } => {
                !join_set_closed.contains(join_set_id)
            }
            _ => true,
        });

        println!("Generated steps: {steps:?}");
        steps
    }

    fn steps_to_registry(steps: &[WorkflowStep]) -> Arc<dyn FunctionRegistry> {
        let ffqns = steps
            .iter()
            .filter_map(|step| match step {
                WorkflowStep::Call { ffqn }
                | WorkflowStep::SubmitExecution {
                    target_ffqn: ffqn, ..
                } => Some(ffqn.clone()),
                _ => None,
            })
            .collect::<Vec<_>>();
        fn_registry_dummy(ffqns.as_slice())
    }

    async fn execute_steps(
        steps: Vec<WorkflowStep>,
        db_pool: Arc<dyn DbPool>,
        sim_clock: &mut SimClock,
        mut next_u128: impl FnMut() -> u128,
    ) -> (ExecutionId, ExecutionLog) {
        let created_at = sim_clock.now();
        info!(now = %created_at, "Steps: {steps:?}");
        let execution_id = ExecutionId::from_parts(
            u64::try_from(sim_clock.now().timestamp_millis()).unwrap(),
            next_u128(),
        );

        let mut child_execution_count = steps
            .iter()
            .filter(|step| matches!(step, WorkflowStep::Call { .. }))
            .count();
        let mut delay_request_count = steps
            .iter()
            .filter(|step| matches!(step, WorkflowStep::Sleep { .. }))
            .count();

        let created_at = sim_clock.now();
        let db_connection = db_pool.connection_test().await.unwrap();
        let fn_registry = steps_to_registry(&steps);
        let (workflow_exec, _close_tx) = {
            let worker = Arc::new(WorkflowWorkerMock::new(
                FFQN_MOCK,
                fn_registry.clone(),
                steps,
                sim_clock.clone_box(),
                db_pool.clone(),
            ));
            let exec_config = ExecConfig {
                batch_size: 1,
                lock_expiry: Duration::from_secs(1),
                tick_sleep: TICK_SLEEP,
                component_id: ComponentId::dummy_workflow(),
                task_limiter_global: None,
                task_limiter_local: None,
                executor_id: ExecutorId::from_parts(
                    u64::try_from(sim_clock.now().timestamp_millis()).unwrap(),
                    next_u128(),
                ),
                retry_config: ComponentRetryConfig::ZERO,
                locking_strategy: LockingStrategy::ByFfqns,
            };
            ExecTask::new_test(
                exec_config,
                worker,
                sim_clock.clone_box(),
                db_pool.clone(),
                Arc::new([FFQN_MOCK]),
            )
        };
        // Create an execution.
        db_connection
            .create(CreateRequest {
                created_at,
                execution_id: execution_id.clone(),
                ffqn: FFQN_MOCK,
                params: Params::empty(),
                parent: None,
                metadata: concepts::ExecutionMetadata::empty(),
                scheduled_at: created_at,
                component_id: ComponentId::dummy_activity(),

                deployment_id: DEPLOYMENT_ID_DUMMY,
                scheduled_by: None,
                paused: false,
            })
            .await
            .unwrap();

        let mut processed = Vec::new();
        let cancel_registry = CancelRegistry::new();
        loop {
            workflow_exec
                .tick_test(
                    sim_clock.now(),
                    RunId::from_parts(
                        u64::try_from(sim_clock.now().timestamp_millis()).unwrap(),
                        next_u128(),
                    ),
                )
                .await
                .wait_for_tasks()
                .await;
            cancellation_driver::tick_test(
                db_connection.as_ref(),
                &cancel_registry,
                sim_clock.now(),
            )
            .await;
            let pending_state = db_connection
                .get_pending_state(&execution_id)
                .await
                .unwrap()
                .pending_state;
            info!("Got {pending_state:?}");
            let join_set_id = match pending_state {
                PendingState::BlockedByJoinSet(PendingStateBlockedByJoinSet {
                    join_set_id,
                    ..
                }) => join_set_id,
                PendingState::PendingAt(PendingStatePendingAt {
                    scheduled_at,
                    last_lock: _,
                }) if scheduled_at <= sim_clock.now() => {
                    // Can happen when join set cancels delay requests.
                    continue;
                }
                PendingState::Finished { .. } => break,
                _ => unreachable!("unexpected {pending_state:?}"),
            };
            let execution_log = db_connection.get(&execution_id).await.unwrap();
            let join_set_req = execution_log
                .find_join_set_request(&join_set_id)
                .cloned()
                .expect("must be found");
            if !processed.contains(&join_set_id) {
                match join_set_req {
                    JoinSetRequest::DelayRequest {
                        delay_id,
                        expires_at,
                        ..
                    } => {
                        info!("DelayRequest {delay_id} moving time to {expires_at}");
                        assert!(delay_request_count > 0);
                        sim_clock.move_time_to(expires_at);
                        delay_request_count -= 1;
                        let actual_progress =
                            expired_timers_watcher::tick_test(db_connection.as_ref(), expires_at)
                                .await
                                .unwrap();
                        assert!(actual_progress.expired_async_timers > 0); // Ignore SubmitDelay-s that were fullfiled by the watcher.
                    }
                    JoinSetRequest::ChildExecutionRequest {
                        child_execution_id,
                        target_ffqn: _,
                        params: _,
                        result,
                    } => {
                        debug!("Got ChildExecutionRequest, executing child");
                        result.expect("must never end with type check error or fn not found");
                        assert!(child_execution_count > 0);
                        exec_child(child_execution_id, db_pool.clone(), sim_clock.clone()).await;
                        child_execution_count -= 1;
                    }
                }
                processed.push(join_set_id);
            }
        }
        // must be finished at this point
        let execution_log = db_connection.get(&execution_id).await.unwrap();
        assert!(execution_log.pending_state.is_finished());
        assert_eq!(0, child_execution_count);
        assert_eq!(0, delay_request_count);
        drop(db_connection);
        (execution_id, execution_log)
    }

    async fn exec_child(
        child_execution_id: ExecutionIdDerived,
        db_pool: Arc<dyn DbPool>,
        sim_clock: SimClock,
    ) {
        info!("Executing child {child_execution_id}");
        let db_connection = db_pool.connection_test().await.unwrap();
        let child_log = db_connection
            .get(&ExecutionId::Derived(child_execution_id.clone()))
            .await
            .unwrap();
        let (parent_execution_id, join_set_id) = assert_matches!(
            child_log.parent(),
            Some(v) => v
        );
        db_connection
            .append_batch_respond_to_parent(
                AppendEventsToExecution {
                    execution_id: ExecutionId::Derived(child_execution_id.clone()),
                    version: child_log.next_version.clone(),
                    batch: vec![AppendRequest {
                        created_at: sim_clock.now(),
                        event: concepts::storage::ExecutionRequest::Finished {
                            retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                            http_client_traces: None,
                        },
                    }],
                },
                AppendResponseToExecution {
                    parent_execution_id,
                    created_at: sim_clock.now(),
                    join_set_id,
                    child_execution_id: child_execution_id.clone(),
                    finished_version: child_log.next_version.clone(),
                    result: SUPPORTED_RETURN_VALUE_OK_EMPTY,
                },
                sim_clock.now(),
            )
            .await
            .unwrap();
        let child_log = db_connection
            .get(&ExecutionId::Derived(child_execution_id.clone()))
            .await
            .unwrap();
        debug!(
            "Child execution {child_execution_id} should be finished: {:?}",
            &child_log.events
        );
        let child_res = child_log.as_finished_result().unwrap();
        assert_matches!(child_res, SupportedFunctionReturnValue::Ok(None));
    }

    fn skip_locked<'a>(
        events_iter: impl Iterator<Item = &'a ExecutionEvent>,
    ) -> impl Iterator<Item = &'a ExecutionRequest> {
        events_iter.filter_map(|event| {
            if matches!(event.event, ExecutionRequest::Locked { .. }) {
                None
            } else {
                Some(&event.event)
            }
        })
    }
}