lash_lashlang_runtime/

process.rs

use std::collections::BTreeMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

use lash_core::ToolChildExecutionTraceHook;
use lash_trace::{
    TraceBranchSelection, TraceContext, TraceEvent, TraceLabelMetadata,
    TraceLashlangChildExecution, TraceLashlangExecutionEvent, TraceLashlangExecutionIdentity,
    TraceLashlangMap, TraceLashlangMapEdge, TraceLashlangMapNode, TraceLashlangStatus, TraceRecord,
    TraceRuntimeScope, TraceRuntimeSubject, TraceSink,
};
use lashlang::{ExecutionHost, ExecutionHostError};
use sha2::{Digest, Sha256};
use tokio_util::sync::CancellationToken;

use crate::{
    LASHLANG_ENGINE_KIND, LashlangProcessEngine, LashlangProcessInput,
    bridge::{
        lashlang_value_to_json, process_event_payload, protocol_tool_reply_to_lashlang_value,
        sleep_duration_ms,
    },
    lashlang_host_environment_satisfies_requirements, prepare_lashlang_process_start,
    resolve_lashlang_module_operation,
};

static SEGMENT_BOUNDARY_DECLINED_TOTAL: AtomicU64 = AtomicU64::new(0);

fn record_segment_boundary_decline(error: &dyn std::fmt::Display, message: &'static str) {
    let declined_total = SEGMENT_BOUNDARY_DECLINED_TOTAL
        .fetch_add(1, Ordering::Relaxed)
        .saturating_add(1);
    tracing::warn!(error = %error, declined_total, "{message}");
}

fn trace_lifecycle_for_segment(
    is_initial_segment: bool,
    output: &lash_core::ProcessRunOutcome,
) -> (bool, bool) {
    (
        is_initial_segment,
        matches!(output, lash_core::ProcessRunOutcome::Terminal(_)),
    )
}

#[derive(serde::Serialize, serde::Deserialize)]
struct LashlangSegmentState {
    vm: lashlang::VmContinuation,
    sleep_sequence: u64,
    event_sequence: u64,
    signal_send_sequence: u64,
    signal_wait_ordinals: BTreeMap<String, u64>,
}

fn lashlang_program_hash(input: &LashlangProcessInput) -> String {
    let identity = serde_json::to_vec(&(
        "lashlang-bytecode",
        lashlang::BYTECODE_FORMAT_VERSION,
        &input.module_ref,
        &input.process_ref,
        &input.host_requirements_ref,
        &input.process_name,
    ))
    .expect("lashlang program identity should serialize");
    format!("sha256:{:x}", Sha256::digest(identity))
}

fn validate_lashlang_program_hash(
    persisted: Option<&str>,
    current: &str,
) -> Result<(), Box<lash_core::ProcessAwaitOutput>> {
    if let Some(persisted) = persisted
        && persisted != current
    {
        return Err(Box::new(process_lashlang_failure(
            "restate_segment_program_hash_mismatch",
            format!(
                "lashlang segment program identity mismatch: persisted {persisted}, current {current}"
            ),
            None,
        )));
    }
    Ok(())
}

pub async fn run_lashlang_process(
    engine: LashlangProcessEngine,
    mut context: lash_core::ProcessEngineRunContext<'_>,
    payload: serde_json::Value,
) -> lash_core::ProcessRunOutcome {
    let handover = context.take_handover();
    let is_initial_segment = handover.is_none();
    let persisted_program_hash = handover
        .as_ref()
        .and_then(|handover| handover.program_hash.clone());
    let segment_controller = context.scoped_effect_controller();
    let phase_probe = context.turn_phase_probe();
    let input = match LashlangProcessInput::from_payload(payload) {
        Ok(input) => input,
        Err(err) => {
            return process_lashlang_failure(
                "process_payload_invalid",
                format!("invalid lashlang process payload: {err}"),
                None,
            )
            .into();
        }
    };
    let artifact = {
        let _phase = context.named_phase("rlm_process.load_artifact");
        match engine
            .artifact_store
            .get_module_artifact(&input.module_ref)
            .await
        {
            Ok(Some(artifact)) => artifact,
            Ok(None) => {
                return process_lashlang_failure(
                    "process_module_artifact_missing",
                    format!("missing lashlang module artifact `{}`", input.module_ref),
                    None,
                )
                .into();
            }
            Err(err) => {
                return process_lashlang_failure(
                    "process_module_artifact_load_failed",
                    format!(
                        "failed to load lashlang module artifact `{}`: {err}",
                        input.module_ref
                    ),
                    None,
                )
                .into();
            }
        }
    };
    if artifact.host_requirements_ref != input.host_requirements_ref {
        return process_lashlang_failure(
            "process_host_requirements_mismatch",
            format!(
                "lashlang process `{}` requested surface {}, artifact has {}",
                input.process_name, input.host_requirements_ref, artifact.host_requirements_ref
            ),
            None,
        )
        .into();
    }
    if artifact.process_ref(&input.process_name) != Some(&input.process_ref) {
        return process_lashlang_failure(
            "process_ref_mismatch",
            format!(
                "lashlang module `{}` does not export process `{}` as requested ref {:?}",
                input.module_ref, input.process_name, input.process_ref
            ),
            None,
        )
        .into();
    }
    let (tool_catalog, host_environment) = {
        let _phase = context.named_phase("rlm_process.resolve_environment");
        let tool_catalog = match context.resolved_tool_catalog() {
            Ok(tool_catalog) => tool_catalog,
            Err(err) => {
                return process_lashlang_failure(
                    "process_tool_catalog_failed",
                    err.to_string(),
                    None,
                )
                .into();
            }
        };
        let surface = engine
            .surface
            .clone()
            .for_process_registry(context.process_registry_available());
        let host_environment = match surface.host_environment(&tool_catalog) {
            Ok(host_environment) => host_environment,
            Err(err) => {
                return process_lashlang_failure("process_host_environment_invalid", err, None)
                    .into();
            }
        };
        if let Err(err) = lashlang_host_environment_satisfies_requirements(
            &artifact.host_requirements,
            &host_environment,
        ) {
            return process_lashlang_failure(
                "process_host_environment_incompatible",
                format!(
                    "lashlang process `{}` is incompatible with this host surface: {err}",
                    input.process_name
                ),
                None,
            )
            .into();
        }
        (tool_catalog, host_environment)
    };
    let compiled = {
        let _phase = context.named_phase("rlm_process.compile");
        let compiled = match engine.process_cache.lock() {
            Ok(mut cache) => {
                cache.get_or_compile(&artifact, &input.process_ref, &input.host_requirements_ref)
            }
            Err(_) => Err(lashlang::RuntimeError::ValueError {
                message: "lashlang compiled process cache lock poisoned".to_string(),
            }),
        };
        match compiled {
            Ok(compiled) => compiled,
            Err(err) => {
                return process_lashlang_failure(
                    "process_compile_failed",
                    format!("failed to compile process `{}`: {err}", input.process_name),
                    None,
                )
                .into();
            }
        }
    };
    let segment_state: Option<LashlangSegmentState> = match handover {
        Some(handover) => match serde_json::from_slice(&handover.engine_state) {
            Ok(state) => Some(state),
            Err(err) => {
                return process_lashlang_failure(
                    "process_segment_handover_invalid",
                    format!("invalid lashlang segment handover: {err}"),
                    None,
                )
                .into();
            }
        },
        None => None,
    };
    let current_program_hash = lashlang_program_hash(&input);
    if let Err(output) =
        validate_lashlang_program_hash(persisted_program_hash.as_deref(), &current_program_hash)
    {
        return (*output).into();
    }
    let process_id = context.registration().id.clone();
    let session_id = context.session_id().to_string();
    let lashlang_execution_trace = LashlangProcessExecutionTrace::new(
        engine.execution_sink.clone(),
        engine.trace_context.clone(),
        session_id,
        process_id.clone(),
        artifact.module_ref.clone(),
        input.process_ref.clone(),
        input.process_name.clone(),
    );
    if is_initial_segment {
        lashlang_execution_trace.emit_started(&artifact);
    }
    let registry = context.registry();
    let cancellation = context.cancellation_token();
    let (ctx, guard, mut state) = {
        let _phase = context.named_phase("rlm_process.build_context");
        let runtime_context = match context.into_runtime_context(tool_catalog) {
            Ok(runtime_context) => runtime_context,
            Err(err) => {
                return process_lashlang_failure(
                    "process_run_context_failed",
                    err.to_string(),
                    None,
                )
                .into();
            }
        };
        let (ctx, guard) = runtime_context.into_parts();
        let mut globals = lashlang::Record::with_capacity(input.args.len());
        for (name, value) in input.args {
            globals.insert(name, lashlang::from_json(value));
        }
        let state = lashlang::State::from_snapshot(lashlang::Snapshot { globals });
        (ctx, guard, state)
    };
    let sleep_sequence = segment_state
        .as_ref()
        .map_or(0, |state| state.sleep_sequence);
    let event_sequence = segment_state
        .as_ref()
        .map_or(0, |state| state.event_sequence);
    let signal_send_sequence = segment_state
        .as_ref()
        .map_or(0, |state| state.signal_send_sequence);
    let signal_wait_ordinals = segment_state
        .as_ref()
        .map_or_else(BTreeMap::new, |state| state.signal_wait_ordinals.clone());
    let host = LashlangProcessHost {
        ctx,
        host_environment,
        artifact_store: engine.artifact_store(),
        registry,
        process_id: process_id.clone(),
        lashlang_execution_trace: lashlang_execution_trace.clone(),
        sleep_sequence: AtomicU64::new(sleep_sequence),
        event_sequence: AtomicU64::new(event_sequence),
        signal_send_sequence: AtomicU64::new(signal_send_sequence),
        signal_wait_ordinals: tokio::sync::Mutex::new(signal_wait_ordinals),
    };
    let env = lashlang::ExecutionEnvironment::new(&host).process();
    let output = {
        let _phase = host.ctx.named_phase("rlm_process.execute");
        execute_lashlang(
            compiled,
            &mut state,
            &env,
            cancellation.clone(),
            segment_controller.controller(),
            &host,
            (segment_state, current_program_hash),
        )
        .await
    };
    drop(env);
    drop(host);
    {
        let _phase =
            lash_core::runtime::RuntimeNamedPhase::begin(phase_probe, "rlm_process.shutdown");
        guard.shutdown().await;
    }
    if trace_lifecycle_for_segment(is_initial_segment, &output).1
        && let lash_core::ProcessRunOutcome::Terminal(output) = &output
    {
        lashlang_execution_trace.emit_finished(output);
    }
    output
}

async fn execute_lashlang(
    compiled: Arc<lashlang::CompiledProgram>,
    state: &mut lashlang::State,
    env: &lashlang::ExecutionEnvironment<'_, LashlangProcessHost<'_>>,
    cancellation: CancellationToken,
    controller: &dyn lash_core::RuntimeEffectController,
    host: &LashlangProcessHost<'_>,
    segment: (Option<LashlangSegmentState>, String),
) -> lash_core::ProcessRunOutcome {
    let (segment_state, program_hash) = segment;
    let mut vm = if let Some(segment_state) = segment_state {
        match lashlang::Vm::resume_from(segment_state.vm, compiled.as_ref(), env) {
            Ok(vm) => vm,
            Err(err) => {
                return process_lashlang_failure(
                    "process_segment_resume_failed",
                    format!("failed to resume lashlang segment: {err}"),
                    None,
                )
                .into();
            }
        }
    } else {
        lashlang::Vm::from_state(compiled.as_ref(), state, env)
    };
    let mut progress = lash_core::SegmentProgress::default();
    loop {
        let execution = if env.trace_runtime_errors() {
            tokio::select! {
                _ = cancellation.cancelled() => {
                    return process_lashlang_cancelled("lashlang process was cancelled").into();
                }
                result = vm.run_process_traced_until_effect() => {
                    result.map_err(|failure| {
                        let error = failure.error.clone();
                        env.observe_runtime_failure(failure);
                        error
                    })
                }
            }
        } else {
            tokio::select! {
                _ = cancellation.cancelled() => {
                    return process_lashlang_cancelled("lashlang process was cancelled").into();
                }
                result = vm.run_process_until_effect() => result,
            }
        };
        if cancellation.is_cancelled() {
            return process_lashlang_cancelled("lashlang process was cancelled").into();
        }
        match execution {
            Ok(lashlang::VmRunOutcome::Complete(output)) => {
                vm.flush_profile(compiled.as_ref(), env);
                return process_lashlang_execution_result(Ok(output)).into();
            }
            Err(err) => {
                vm.flush_profile(compiled.as_ref(), env);
                return process_lashlang_execution_result(Err(err)).into();
            }
            Ok(lashlang::VmRunOutcome::EffectCompleted) => {
                progress.effects_executed += 1;
                let Some(reason) = controller.wants_segment_boundary(&progress) else {
                    continue;
                };
                match vm.suspend() {
                    Ok(continuation) => {
                        let segment_state = LashlangSegmentState {
                            vm: continuation,
                            sleep_sequence: host.sleep_sequence.load(Ordering::Relaxed),
                            event_sequence: host.event_sequence.load(Ordering::Relaxed),
                            signal_send_sequence: host.signal_send_sequence.load(Ordering::Relaxed),
                            signal_wait_ordinals: host.signal_wait_ordinals.lock().await.clone(),
                        };
                        match serde_json::to_vec(&segment_state) {
                            Ok(engine_state) => {
                                return lash_core::ProcessRunOutcome::SegmentBoundary(
                                    lash_core::SegmentHandover {
                                        reason,
                                        program_hash: Some(program_hash.clone()),
                                        engine_state,
                                    },
                                );
                            }
                            Err(err) => {
                                record_segment_boundary_decline(
                                    &err,
                                    "lashlang segment continuation was not serializable; continuing",
                                );
                            }
                        }
                    }
                    Err(err) => {
                        record_segment_boundary_decline(
                            &err,
                            "lashlang segment boundary declined at non-capturable point",
                        );
                    }
                }
            }
        }
    }
}

struct LashlangProcessHost<'run> {
    ctx: lash_core::RuntimeExecutionContext<'run>,
    host_environment: lashlang::LashlangHostEnvironment,
    artifact_store: Arc<dyn lashlang::LashlangArtifactStore>,
    registry: Arc<dyn lash_core::ProcessRegistry>,
    process_id: String,
    lashlang_execution_trace: LashlangProcessExecutionTrace,
    sleep_sequence: AtomicU64,
    event_sequence: AtomicU64,
    signal_send_sequence: AtomicU64,
    signal_wait_ordinals: tokio::sync::Mutex<BTreeMap<String, u64>>,
}

type ProcessHostAbilityFuture<'a> =
    Pin<Box<dyn Future<Output = Result<lashlang::AbilityResult, ExecutionHostError>> + Send + 'a>>;

impl LashlangProcessHost<'_> {
    fn resource_payload(
        &self,
        args: &[lashlang::Value],
    ) -> Result<serde_json::Value, ExecutionHostError> {
        let mut payload = if let [lashlang::Value::Record(record)] = args {
            lashlang_value_to_json(&lashlang::Value::Record(Arc::clone(record)))?
        } else {
            serde_json::json!({
                "args": args
                    .iter()
                    .map(lashlang_value_to_json)
                    .collect::<Result<Vec<_>, _>>()?,
            })
        };
        payload
            .as_object_mut()
            .ok_or_else(|| ExecutionHostError::new("module operation payload must be an object"))?;
        Ok(payload)
    }

    fn resource_tool_call_id(
        &self,
        host_operation: &str,
        call_site: &lashlang::LashlangExecutionCallSite,
        batch_index: Option<usize>,
    ) -> String {
        let mut call_id = format!(
            "lashlang:{}:resource:{}:{}:{}",
            self.process_id, host_operation, call_site.site.node_id, call_site.occurrence
        );
        if let Some(batch_index) = batch_index {
            call_id.push_str(&format!(":child:{batch_index}"));
        }
        call_id
    }

    fn prepare_resource_invocation(
        &self,
        operation: String,
        receiver: lashlang::Value,
        args: Vec<lashlang::Value>,
        call_site: Option<lashlang::LashlangExecutionCallSite>,
        batch_index: Option<usize>,
    ) -> Result<(String, lash_core::ToolInvocation), ExecutionHostError> {
        let receiver = match &receiver {
            lashlang::Value::Resource(receiver) => receiver,
            _ => {
                return Err(ExecutionHostError::new(format!(
                    "module operation `{operation}` requires a module authority receiver"
                )));
            }
        };
        let host_operation =
            resolve_lashlang_module_operation(&self.host_environment, receiver, &operation)?;
        let tool_id = lash_core::ToolId::from(host_operation.as_str());
        let manifest = self.ctx.callable_tool_manifest_by_id(&tool_id).ok_or_else(|| {
            ExecutionHostError::new(format!(
                "module operation `{}` resolved to unavailable host operation `{host_operation}`",
                operation
            ))
        })?;
        let payload = self.resource_payload(&args)?;
        let call_site = call_site.ok_or_else(|| {
            ExecutionHostError::new(format!(
                "module operation `{operation}` resolved to host operation `{host_operation}` but has no deterministic lashlang execution call site"
            ))
        })?;
        let call_id = self.resource_tool_call_id(&host_operation, &call_site, batch_index);
        let mut invocation = lash_core::ToolInvocation::new(call_id, manifest.id.clone(), payload);
        if let Some(hook) = self
            .lashlang_execution_trace
            .tool_child_execution_trace_hook(call_site)
        {
            invocation = invocation.with_child_execution_trace_hook(hook);
        }
        Ok((host_operation, invocation))
    }

    async fn resource_operation(
        &self,
        operation: String,
        receiver: lashlang::Value,
        args: Vec<lashlang::Value>,
        call_site: Option<lashlang::LashlangExecutionCallSite>,
    ) -> Result<lashlang::Value, ExecutionHostError> {
        let (_, invocation) =
            self.prepare_resource_invocation(operation, receiver, args, call_site, None)?;
        let lash_core::ToolInvocation {
            id,
            tool_id,
            args,
            execution_grant: _,
            child_execution_trace_hook,
        } = invocation;
        let reply = if let Some(call_site) = child_execution_trace_hook {
            self.ctx
                .call_tool_by_id_with_child_execution_trace_hook(id, tool_id, args, 0, call_site)
                .await
        } else {
            self.ctx.call_tool_by_id(id, tool_id, args, 0).await
        };
        protocol_tool_reply_to_lashlang_value(reply)
    }

    async fn resource_operation_batch(
        &self,
        batch: lashlang::ResourceOperationBatch,
    ) -> lashlang::ResourceOperationBatchResult {
        let mut results = vec![None; batch.operations.len()];
        let mut positions = Vec::new();
        let mut invocations = Vec::new();
        for (index, operation) in batch.operations.into_iter().enumerate() {
            match self.prepare_resource_invocation(
                operation.operation,
                operation.receiver,
                operation.args,
                operation.call_site,
                Some(index),
            ) {
                Ok((_, invocation)) => {
                    positions.push(index);
                    invocations.push(invocation);
                }
                Err(error) => {
                    results[index] = Some(lashlang::ResourceOperationResult::Error(error));
                }
            }
        }

        for (index, reply) in positions
            .into_iter()
            .zip(self.ctx.call_tool_batch(invocations).await)
        {
            results[index] = Some(lashlang::ResourceOperationResult::from_result(
                protocol_tool_reply_to_lashlang_value(reply),
            ));
        }

        lashlang::ResourceOperationBatchResult {
            results: results
                .into_iter()
                .map(|result| result.expect("every batch result slot should be filled"))
                .collect(),
        }
    }

    async fn await_handle(
        &self,
        handle: lashlang::Value,
    ) -> Result<lashlang::Value, ExecutionHostError> {
        let reply = {
            let _phase = self.ctx.named_phase("rlm_process.await_handle");
            self.ctx
                .await_tool_handle(
                    uuid::Uuid::new_v4().to_string(),
                    lashlang_value_to_json(&handle)?,
                )
                .await
        };
        protocol_tool_reply_to_lashlang_value(reply)
    }

    async fn cancel_handle(
        &self,
        handle: lashlang::Value,
    ) -> Result<lashlang::Value, ExecutionHostError> {
        let reply = self
            .ctx
            .cancel_tool_handle(
                uuid::Uuid::new_v4().to_string(),
                lashlang_value_to_json(&handle)?,
            )
            .await;
        protocol_tool_reply_to_lashlang_value(reply)
    }

    async fn start_process(
        &self,
        start: lashlang::ProcessStart,
    ) -> Result<lashlang::Value, ExecutionHostError> {
        let prepared = {
            let _phase = self.ctx.named_phase("rlm_process.prepare_start");
            let parent_start_seed = format!("parent-process:{}", self.process_id);
            prepare_lashlang_process_start(
                Arc::clone(&self.artifact_store),
                &parent_start_seed,
                start,
            )
            .await
            .map_err(ExecutionHostError::new)?
        };
        let reply = {
            let _phase = self.ctx.named_phase("rlm_process.start");
            self.ctx
                .start_child_process(prepared.registration, LASHLANG_ENGINE_KIND, prepared.label)
                .await
        };
        protocol_tool_reply_to_lashlang_value(reply)
    }

    async fn process_event(&self, event: lashlang::ProcessEvent) -> Result<(), ExecutionHostError> {
        let event_type = match event.kind {
            lashlang::ProcessEventKind::Yield => "process.yield",
            lashlang::ProcessEventKind::Wake => "process.wake",
        };
        let ordinal = self.event_sequence.fetch_add(1, Ordering::Relaxed);
        self.ctx
            .append_process_event(
                Arc::clone(&self.registry),
                &self.process_id,
                lash_core::ProcessEventAppendRequest::new(
                    event_type,
                    process_event_payload(&event.value)?,
                )
                .with_replay_key(format!("process:{}:event:{ordinal}", self.process_id)),
            )
            .await
            .map_err(|err| ExecutionHostError::new(err.to_string()))?;
        Ok(())
    }

    async fn sleep(&self, sleep: lashlang::Sleep) -> Result<lashlang::Value, ExecutionHostError> {
        let duration_ms = sleep_duration_ms(sleep.kind, &sleep.value)?;
        let sequence = self.sleep_sequence.fetch_add(1, Ordering::Relaxed);
        let scope = format!("process:{}", self.process_id);
        self.ctx
            .sleep_process(&scope, sequence, duration_ms)
            .await
            .map_err(|err| ExecutionHostError::new(err.to_string()))?;
        Ok(lashlang::Value::Null)
    }

    async fn wait_signal(&self, name: String) -> Result<lashlang::Value, ExecutionHostError> {
        let event_type = lash_core::process_signal_event_type(&name)
            .map_err(|err| ExecutionHostError::new(err.to_string()))?;
        let event_ordinal = {
            let mut ordinals = self.signal_wait_ordinals.lock().await;
            let ordinal = ordinals.entry(name.clone()).or_insert(0);
            *ordinal += 1;
            *ordinal
        };
        let key = lash_core::process_signal_wait_key(&self.process_id, &name, event_ordinal);
        let waiting_replay_key = format!(
            "process:{}:waiting:signal.{}:{event_ordinal}",
            self.process_id, name
        );
        let since_ms = self
            .wait_since_ms(&key, &waiting_replay_key)
            .await
            .map_err(|err| ExecutionHostError::new(err.to_string()))?;
        let wait = lash_core::WaitState {
            since_ms,
            kind: lash_core::WaitKind::Signal {
                name: name.clone(),
                event_type: event_type.clone(),
                key: key.clone(),
                ordinal: event_ordinal,
            },
        };
        self.registry
            .set_process_wait(&self.process_id, wait.clone())
            .await
            .map_err(|err| ExecutionHostError::new(err.to_string()))?;
        self.registry
            .append_event(
                &self.process_id,
                lash_core::ProcessEventAppendRequest::new(
                    "process.waiting",
                    serde_json::json!({ "wait": wait }),
                )
                .with_replay_key(waiting_replay_key),
            )
            .await
            .map_err(|err| ExecutionHostError::new(err.to_string()))?;
        let payload = self
            .ctx
            .await_process_signal_event(&self.process_id, &name, event_ordinal)
            .await
            .map_err(|err| ExecutionHostError::new(err.to_string()))?;
        self.registry
            .clear_process_wait(&self.process_id)
            .await
            .map_err(|err| ExecutionHostError::new(err.to_string()))?;
        self.registry
            .append_event(
                &self.process_id,
                lash_core::ProcessEventAppendRequest::new(
                    "process.resumed",
                    serde_json::json!({
                        "signal": name,
                        "key": key,
                        "ordinal": event_ordinal,
                    }),
                )
                .with_replay_key(format!(
                    "process:{}:resumed:signal.{}:{event_ordinal}",
                    self.process_id, name
                )),
            )
            .await
            .map_err(|err| ExecutionHostError::new(err.to_string()))?;
        Ok(lashlang::from_json(payload))
    }

    async fn wait_since_ms(
        &self,
        key: &str,
        waiting_replay_key: &str,
    ) -> Result<u64, lash_core::PluginError> {
        if let Some(since_ms) =
            self.registry
                .get_process(&self.process_id)
                .await
                .and_then(|record| {
                    let wait = record.wait?;
                    match &wait.kind {
                        lash_core::WaitKind::Signal { key: wait_key, .. } if wait_key == key => {
                            Some(wait.since_ms)
                        }
                        _ => None,
                    }
                })
        {
            return Ok(since_ms);
        }

        for event in self
            .registry
            .events_after(&self.process_id, 0)
            .await?
            .into_iter()
            .rev()
        {
            if event.event_type != "process.waiting"
                || event.invocation.replay_key() != Some(waiting_replay_key)
            {
                continue;
            }
            let Some(wait_value) = event.payload.get("wait") else {
                continue;
            };
            if let Ok(wait) = serde_json::from_value::<lash_core::WaitState>(wait_value.clone()) {
                return Ok(wait.since_ms);
            }
        }
        Ok(lash_core::current_epoch_ms())
    }

    async fn signal_run(
        &self,
        signal: lashlang::ProcessSignal,
    ) -> Result<lashlang::Value, ExecutionHostError> {
        let target = process_id_from_lashlang_handle(&signal.run)?;
        let payload = lashlang_value_to_json(&signal.payload)?;
        let sequence = self.signal_send_sequence.fetch_add(1, Ordering::Relaxed);
        let signal_id = format!(
            "lashlang:{}:signal.{}:{sequence}",
            self.process_id, signal.name
        );
        self.ctx
            .signal_process_by_id(
                Arc::clone(&self.registry),
                &target,
                &signal.name,
                signal_id,
                payload,
            )
            .await
            .map_err(|err| ExecutionHostError::new(err.to_string()))?;
        Ok(lashlang::Value::Null)
    }

    fn perform_selected_ability<'a>(
        &'a self,
        op: lashlang::AbilityOp,
    ) -> ProcessHostAbilityFuture<'a> {
        match op {
            lashlang::AbilityOp::ResourceOperation(operation) => Box::pin(async move {
                self.resource_operation(
                    operation.operation,
                    operation.receiver,
                    operation.args,
                    operation.call_site,
                )
                .await
                .map(lashlang::AbilityResult::Value)
            }),
            lashlang::AbilityOp::ResourceOperationBatch(batch) => Box::pin(async move {
                Ok(lashlang::AbilityResult::ResourceOperationBatch(
                    self.resource_operation_batch(batch).await,
                ))
            }),
            lashlang::AbilityOp::Await(handle) => Box::pin(async move {
                self.await_handle(handle)
                    .await
                    .map(lashlang::AbilityResult::Value)
            }),
            lashlang::AbilityOp::Cancel(handle) => Box::pin(async move {
                self.cancel_handle(handle)
                    .await
                    .map(lashlang::AbilityResult::Value)
            }),
            lashlang::AbilityOp::StartProcess(start) => Box::pin(async move {
                self.start_process(*start)
                    .await
                    .map(lashlang::AbilityResult::Value)
            }),
            lashlang::AbilityOp::ProcessEvent(event) => Box::pin(async move {
                self.process_event(event).await?;
                Ok(lashlang::AbilityResult::Unit)
            }),
            lashlang::AbilityOp::Sleep(sleep) => {
                Box::pin(async move { self.sleep(sleep).await.map(lashlang::AbilityResult::Value) })
            }
            lashlang::AbilityOp::WaitSignal { name } => Box::pin(async move {
                self.wait_signal(name)
                    .await
                    .map(lashlang::AbilityResult::Value)
            }),
            lashlang::AbilityOp::SignalRun(signal) => Box::pin(async move {
                self.signal_run(signal)
                    .await
                    .map(lashlang::AbilityResult::Value)
            }),
            lashlang::AbilityOp::Print(_) => Box::pin(async {
                Err(ExecutionHostError::new(
                    "`print` is not available inside lashlang process bodies",
                ))
            }),
            lashlang::AbilityOp::Finish(value) | lashlang::AbilityOp::Fail(value) => {
                Box::pin(async move { Ok(lashlang::AbilityResult::Value(value)) })
            }
        }
    }
}

impl lashlang::ExecutionHost for LashlangProcessHost<'_> {
    fn perform(
        &self,
        op: lashlang::AbilityOp,
    ) -> impl Future<Output = Result<lashlang::AbilityResult, ExecutionHostError>> + Send {
        self.perform_selected_ability(op)
    }

    fn observe_lashlang_execution(&self, observation: lashlang::LashlangExecutionObservation) {
        self.lashlang_execution_trace.emit_observation(observation);
    }
}

#[derive(Clone)]
struct LashlangProcessExecutionTrace {
    sink: Option<Arc<dyn TraceSink>>,
    base_context: TraceContext,
    session_id: String,
    process_id: String,
    module_ref: lashlang::ModuleRef,
    process_ref: lashlang::ProcessRef,
    process_name: String,
}

impl LashlangProcessExecutionTrace {
    fn new(
        sink: Option<Arc<dyn TraceSink>>,
        base_context: TraceContext,
        session_id: String,
        process_id: String,
        module_ref: lashlang::ModuleRef,
        process_ref: lashlang::ProcessRef,
        process_name: String,
    ) -> Self {
        Self {
            sink,
            base_context,
            session_id,
            process_id,
            module_ref,
            process_ref,
            process_name,
        }
    }

    fn identity(&self) -> TraceLashlangExecutionIdentity {
        TraceLashlangExecutionIdentity {
            scope: TraceRuntimeScope::new(self.session_id.clone()),
            subject: TraceRuntimeSubject::Process {
                process_id: self.process_id.clone(),
            },
            module_ref: self.module_ref.to_string(),
            entry_kind: "process".to_string(),
            entry_ref: Some(lashlang::process_ref_key(&self.process_ref)),
            entry_name: self.process_name.clone(),
        }
    }

    fn event_key(&self, suffix: impl std::fmt::Display) -> String {
        format!("lashlang_execution:{}:{suffix}", self.process_id)
    }

    fn emit_started(&self, artifact: &lashlang::ModuleArtifact) {
        self.emit(TraceLashlangExecutionEvent::ExecutionStarted {
            event_key: self.event_key("started"),
            identity: self.identity(),
            execution_map: trace_lashlang_process_map(
                artifact,
                &self.process_ref,
                &self.process_name,
            ),
        });
    }

    fn emit_finished(&self, output: &lash_core::ProcessAwaitOutput) {
        let (status, error) = match output {
            lash_core::ProcessAwaitOutput::Success { .. } => (TraceLashlangStatus::Completed, None),
            lash_core::ProcessAwaitOutput::Failure { message, .. } => {
                (TraceLashlangStatus::Failed, Some(message.clone()))
            }
            lash_core::ProcessAwaitOutput::Cancelled { message, .. } => {
                (TraceLashlangStatus::Cancelled, Some(message.clone()))
            }
            // `emit_finished` fires after an actual execution, whose outcome is
            // Success/Failure/Cancelled — abandonment is written out-of-band by the
            // sweep, never returned by a run. Map it defensively to Failed.
            lash_core::ProcessAwaitOutput::Abandoned { .. } => (
                TraceLashlangStatus::Failed,
                Some("process abandoned".to_string()),
            ),
        };
        self.emit(TraceLashlangExecutionEvent::ExecutionFinished {
            event_key: self.event_key("finished"),
            identity: self.identity(),
            status,
            error,
        });
    }

    fn emit_observation(&self, observation: lashlang::LashlangExecutionObservation) {
        if self.sink.is_none() {
            return;
        }
        let identity = self.identity();
        let event = match observation {
            lashlang::LashlangExecutionObservation::NodeStarted { site, occurrence } => {
                TraceLashlangExecutionEvent::NodeStarted {
                    event_key: self
                        .event_key(format!("node:{}:{occurrence}:started", site.node_id)),
                    identity,
                    node_id: site.node_id,
                    node_kind: site.node_kind,
                    label: site.label,
                    occurrence,
                }
            }
            lashlang::LashlangExecutionObservation::NodeCompleted { site, occurrence } => {
                TraceLashlangExecutionEvent::NodeCompleted {
                    event_key: self
                        .event_key(format!("node:{}:{occurrence}:completed", site.node_id)),
                    identity,
                    node_id: site.node_id,
                    node_kind: site.node_kind,
                    label: site.label,
                    occurrence,
                }
            }
            lashlang::LashlangExecutionObservation::NodeFailed {
                site,
                occurrence,
                error,
            } => TraceLashlangExecutionEvent::NodeFailed {
                event_key: self.event_key(format!("node:{}:{occurrence}:failed", site.node_id)),
                identity,
                node_id: site.node_id,
                node_kind: site.node_kind,
                label: site.label,
                occurrence,
                error,
            },
            lashlang::LashlangExecutionObservation::BranchSelected {
                site,
                occurrence,
                edge_id,
                selected,
            } => TraceLashlangExecutionEvent::BranchSelected {
                event_key: self
                    .event_key(format!("branch:{}:{occurrence}:{edge_id}", site.node_id)),
                identity,
                node_id: site.node_id,
                occurrence,
                edge_id,
                selected: match selected {
                    lashlang::ProcessBranchSelection::Then => TraceBranchSelection::Then,
                    lashlang::ProcessBranchSelection::Else => TraceBranchSelection::Else,
                },
            },
            lashlang::LashlangExecutionObservation::ChildStarted {
                site,
                occurrence,
                child,
            } => TraceLashlangExecutionEvent::ChildStarted {
                event_key: self.event_key(format!(
                    "child:{}:{occurrence}:{}",
                    site.node_id, child.process_id
                )),
                identity,
                parent_node_id: site.node_id,
                occurrence,
                child: TraceLashlangChildExecution {
                    scope: TraceRuntimeScope::new(self.session_id.clone()),
                    subject: TraceRuntimeSubject::Process {
                        process_id: child.process_id,
                    },
                    module_ref: Some(child.module_ref.to_string()),
                    entry_ref: Some(lashlang::process_ref_key(&child.process_ref)),
                    entry_name: Some(child.process_name),
                },
            },
        };
        self.emit(event);
    }

    fn tool_child_execution_trace_hook(
        &self,
        call_site: lashlang::LashlangExecutionCallSite,
    ) -> Option<ToolChildExecutionTraceHook> {
        self.sink.as_ref()?;
        let trace = self.clone();
        let parent_node_id = call_site.site.node_id;
        let occurrence = call_site.occurrence;
        Some(ToolChildExecutionTraceHook::new(move |started| {
            let child = TraceLashlangChildExecution {
                scope: TraceRuntimeScope::new(trace.session_id.clone()),
                subject: TraceRuntimeSubject::Process {
                    process_id: started.process_id,
                },
                module_ref: None,
                entry_ref: None,
                entry_name: started.child_entry_name,
            };
            let child_graph_key = child.graph_key();
            trace.emit(TraceLashlangExecutionEvent::ChildStarted {
                event_key: trace.event_key(format!(
                    "child:{parent_node_id}:{occurrence}:{child_graph_key}"
                )),
                identity: trace.identity(),
                parent_node_id: parent_node_id.clone(),
                occurrence,
                child,
            });
        }))
    }

    fn emit(&self, event: TraceLashlangExecutionEvent) {
        let Some(sink) = &self.sink else {
            return;
        };
        let mut context = self.base_context.clone();
        context.session_id = Some(self.session_id.clone());
        let _ = sink.append(&TraceRecord::new(
            context,
            TraceEvent::LashlangExecution { event },
        ));
    }
}

fn trace_lashlang_process_map(
    artifact: &lashlang::ModuleArtifact,
    process_ref: &lashlang::ProcessRef,
    process_name: &str,
) -> TraceLashlangMap {
    let source = lashlang::canonical_program_source_with_requirements(
        &artifact.canonical_ir,
        &artifact.host_requirements,
    );
    let graph = source
        .ok()
        .and_then(|source| lashlang::workflow_graph_from_source(&source).ok());
    let Some(process) = graph.as_ref().and_then(|graph| graph.process(process_name)) else {
        return TraceLashlangMap {
            module_ref: artifact.module_ref.to_string(),
            entry_kind: "process".to_string(),
            entry_ref: Some(lashlang::process_ref_key(process_ref)),
            entry_name: process_name.to_string(),
            nodes: Vec::new(),
            edges: Vec::new(),
        };
    };
    let mut nodes = Vec::new();
    let mut edges = Vec::new();
    let mut primary_runtime_ids = BTreeMap::new();
    append_trace_workflow_subgraph(
        artifact,
        &process.body,
        &mut nodes,
        &mut edges,
        &mut primary_runtime_ids,
    );
    TraceLashlangMap {
        module_ref: artifact.module_ref.to_string(),
        entry_kind: "process".to_string(),
        entry_ref: Some(lashlang::process_ref_key(process_ref)),
        entry_name: process_name.to_string(),
        nodes,
        edges,
    }
}

/// Builds the trace runtime's read-only foreground skeleton from the workflow graph.
pub fn trace_lashlang_main_map(artifact: &lashlang::ModuleArtifact) -> TraceLashlangMap {
    let graph = lashlang::canonical_program_source_with_requirements(
        &artifact.canonical_ir,
        &artifact.host_requirements,
    )
    .ok()
    .and_then(|source| lashlang::workflow_graph_from_source(&source).ok());
    let Some(graph) = graph else {
        return TraceLashlangMap {
            module_ref: artifact.module_ref.to_string(),
            entry_kind: "main".to_string(),
            entry_ref: None,
            entry_name: "main".to_string(),
            nodes: Vec::new(),
            edges: Vec::new(),
        };
    };
    let mut nodes = Vec::new();
    let mut edges = Vec::new();
    let mut primary_runtime_ids = BTreeMap::new();
    append_trace_workflow_subgraph(
        artifact,
        &graph.main,
        &mut nodes,
        &mut edges,
        &mut primary_runtime_ids,
    );
    TraceLashlangMap {
        module_ref: artifact.module_ref.to_string(),
        entry_kind: "main".to_string(),
        entry_ref: None,
        entry_name: "main".to_string(),
        nodes,
        edges,
    }
}

fn append_trace_workflow_subgraph(
    artifact: &lashlang::ModuleArtifact,
    graph: &lashlang::WorkflowSubgraph,
    nodes: &mut Vec<TraceLashlangMapNode>,
    edges: &mut Vec<TraceLashlangMapEdge>,
    primary_runtime_ids: &mut BTreeMap<String, String>,
) {
    for node in &graph.nodes {
        let label_metadata =
            (node.name_source == lashlang::WorkflowNodeNameSource::Label).then(|| {
                TraceLabelMetadata {
                    title: node.name.clone(),
                    description: node.description.clone(),
                }
            });
        for site in &node.execution_sites {
            let Some(runtime_site) =
                lashlang::runtime_execution_site_for_workflow_site(artifact, site)
            else {
                continue;
            };
            primary_runtime_ids
                .entry(node.id.to_string())
                .or_insert_with(|| runtime_site.node_id.clone());
            if nodes.iter().any(|node| node.id == runtime_site.node_id) {
                continue;
            }
            nodes.push(TraceLashlangMapNode {
                id: runtime_site.node_id,
                kind: runtime_site.node_kind,
                label: runtime_site.label,
                label_metadata: label_metadata.clone(),
            });
        }
        match &node.kind {
            lashlang::WorkflowNodeKind::Container(lashlang::WorkflowContainer::If {
                then_graph,
                else_graph,
                ..
            }) => {
                if let Some(graph) = then_graph {
                    append_trace_workflow_subgraph(
                        artifact,
                        graph,
                        nodes,
                        edges,
                        primary_runtime_ids,
                    );
                }
                if let Some(graph) = else_graph {
                    append_trace_workflow_subgraph(
                        artifact,
                        graph,
                        nodes,
                        edges,
                        primary_runtime_ids,
                    );
                }
            }
            lashlang::WorkflowNodeKind::Container(lashlang::WorkflowContainer::For {
                body: Some(graph),
                ..
            }) => {
                append_trace_workflow_subgraph(artifact, graph, nodes, edges, primary_runtime_ids);
            }
            lashlang::WorkflowNodeKind::Container(lashlang::WorkflowContainer::While {
                body: Some(graph),
                ..
            }) => {
                append_trace_workflow_subgraph(artifact, graph, nodes, edges, primary_runtime_ids);
            }
            lashlang::WorkflowNodeKind::Container(
                lashlang::WorkflowContainer::ListComprehension {
                    element: Some(graph),
                    ..
                },
            ) => {
                append_trace_workflow_subgraph(artifact, graph, nodes, edges, primary_runtime_ids);
            }
            _ => {}
        }
    }
    for edge in &graph.edges {
        let (Some(from), Some(to)) = (
            primary_runtime_ids.get(edge.from.as_str()),
            primary_runtime_ids.get(edge.to.as_str()),
        ) else {
            continue;
        };
        let label = match &edge.kind {
            lashlang::WorkflowEdgeKind::Sequence => "sequence".to_string(),
            lashlang::WorkflowEdgeKind::DataDependency { variable, version } => {
                format!("{variable}@{version}")
            }
        };
        edges.push(TraceLashlangMapEdge {
            id: edge.id.clone(),
            from: from.clone(),
            to: to.clone(),
            label,
        });
    }
}

fn process_lashlang_execution_result(
    result: Result<lashlang::ExecutionOutcome, lashlang::RuntimeError>,
) -> lash_core::ProcessAwaitOutput {
    match result {
        Ok(lashlang::ExecutionOutcome::Finished(value)) => lash_core::ProcessAwaitOutput::Success {
            value: lashlang_value_to_json(&value)
                .unwrap_or_else(|err| serde_json::json!({ "error": err.to_string() })),
            control: None,
        },
        Ok(lashlang::ExecutionOutcome::Failed(value)) => process_lashlang_failure(
            "process_failed",
            value.to_string(),
            Some(
                lashlang_value_to_json(&value)
                    .unwrap_or_else(|err| serde_json::json!({ "error": err.to_string() })),
            ),
        ),
        Ok(lashlang::ExecutionOutcome::Continued) => lash_core::ProcessAwaitOutput::Success {
            value: serde_json::Value::Null,
            control: None,
        },
        Err(err) => process_lashlang_failure("process_runtime_error", err.to_string(), None),
    }
}

fn process_lashlang_failure(
    code: &str,
    message: impl Into<String>,
    raw: Option<serde_json::Value>,
) -> lash_core::ProcessAwaitOutput {
    lash_core::ProcessAwaitOutput::Failure {
        class: lash_core::ToolFailureClass::Execution,
        code: code.to_string(),
        message: message.into(),
        raw,
        control: None,
    }
}

fn process_lashlang_cancelled(message: impl Into<String>) -> lash_core::ProcessAwaitOutput {
    lash_core::ProcessAwaitOutput::Cancelled {
        message: message.into(),
        raw: None,
        control: None,
    }
}

fn process_id_from_lashlang_handle(handle: &lashlang::Value) -> Result<String, ExecutionHostError> {
    let value = lashlang_value_to_json(handle)?;
    let Some(object) = value.as_object() else {
        return Err(ExecutionHostError::new(
            "signal_run expects a process handle",
        ));
    };
    if object.get("__handle__").and_then(serde_json::Value::as_str) != Some("process") {
        return Err(ExecutionHostError::new(
            "signal_run expects a process handle",
        ));
    }
    object
        .get("id")
        .and_then(serde_json::Value::as_str)
        .map(ToOwned::to_owned)
        .ok_or_else(|| ExecutionHostError::new("signal_run process handle is missing `id`"))
}

pub fn lashlang_process_event_types() -> Vec<lash_core::ProcessEventType> {
    vec![
        lash_core::ProcessEventType {
            name: "process.yield".to_string(),
            payload_schema: lash_core::LashSchema::any(),
            semantics: lash_core::ProcessEventSemanticsSpec::default(),
        },
        lash_core::ProcessEventType {
            name: "process.wake".to_string(),
            payload_schema: lash_core::LashSchema::any(),
            semantics: lash_core::ProcessEventSemanticsSpec {
                wake: Some(lash_core::ProcessWakeSpec {
                    when: None,
                    input: lash_core::ProcessValueSelector::Pointer("/text".to_string()),
                    dedupe_key: lash_core::ProcessWakeDedupeKey::EventIdentity,
                }),
                ..lash_core::ProcessEventSemanticsSpec::default()
            },
        },
    ]
}

pub fn lashlang_process_signal_event_types(
    process: &lashlang::ProcessDecl,
) -> Vec<lash_core::ProcessEventType> {
    process
        .signals
        .iter()
        .map(|signal| lash_core::ProcessEventType {
            name: lash_core::process_signal_event_type(signal.name.as_str())
                .expect("lashlang process signal declarations use parser-validated names"),
            payload_schema: lash_core::LashSchema::new(lashlang_type_expr_schema(&signal.ty)),
            semantics: lash_core::ProcessEventSemanticsSpec::default(),
        })
        .collect()
}

pub fn lashlang_type_expr_schema(ty: &lashlang::TypeExpr) -> serde_json::Value {
    match ty {
        lashlang::TypeExpr::Any
        | lashlang::TypeExpr::Dict
        | lashlang::TypeExpr::Ref(_)
        | lashlang::TypeExpr::Process { .. }
        | lashlang::TypeExpr::TriggerHandle(_) => serde_json::json!({}),
        lashlang::TypeExpr::Str => serde_json::json!({ "type": "string" }),
        lashlang::TypeExpr::Int => serde_json::json!({ "type": "integer" }),
        lashlang::TypeExpr::Float => serde_json::json!({ "type": "number" }),
        lashlang::TypeExpr::Bool => serde_json::json!({ "type": "boolean" }),
        lashlang::TypeExpr::Null => serde_json::json!({ "type": "null" }),
        lashlang::TypeExpr::Enum(values) => serde_json::json!({
            "enum": values.iter().map(|value| value.as_str()).collect::<Vec<_>>()
        }),
        lashlang::TypeExpr::List(item) => serde_json::json!({
            "type": "array",
            "items": lashlang_type_expr_schema(item),
        }),
        lashlang::TypeExpr::Object(fields) => {
            let mut properties = serde_json::Map::new();
            let mut required = Vec::new();
            for field in fields {
                properties.insert(field.name.to_string(), lashlang_type_expr_schema(&field.ty));
                if !field.optional {
                    required.push(serde_json::Value::String(field.name.to_string()));
                }
            }
            let mut schema = serde_json::Map::new();
            schema.insert(
                "type".to_string(),
                serde_json::Value::String("object".to_string()),
            );
            schema.insert(
                "properties".to_string(),
                serde_json::Value::Object(properties),
            );
            if !required.is_empty() {
                schema.insert("required".to_string(), serde_json::Value::Array(required));
            }
            schema.insert(
                "additionalProperties".to_string(),
                serde_json::Value::Bool(true),
            );
            serde_json::Value::Object(schema)
        }
        lashlang::TypeExpr::Union(variants) => serde_json::json!({
            "anyOf": variants.iter().map(lashlang_type_expr_schema).collect::<Vec<_>>()
        }),
    }
}

#[cfg(test)]
mod segment_trace_tests {
    use super::{
        SEGMENT_BOUNDARY_DECLINED_TOTAL, record_segment_boundary_decline,
        trace_lifecycle_for_segment, validate_lashlang_program_hash,
    };
    use std::sync::atomic::Ordering;

    #[test]
    fn multi_segment_trace_emits_one_started_and_one_finished() {
        let boundary = || {
            lash_core::ProcessRunOutcome::SegmentBoundary(lash_core::SegmentHandover {
                reason: lash_core::BoundaryReason::JournalBudget,
                program_hash: Some("program-v1".to_string()),
                engine_state: Vec::new(),
            })
        };
        let terminal = lash_core::ProcessRunOutcome::from(lash_core::ProcessAwaitOutput::Success {
            value: serde_json::Value::Null,
            control: None,
        });
        let lifecycle = [
            trace_lifecycle_for_segment(true, &boundary()),
            trace_lifecycle_for_segment(false, &boundary()),
            trace_lifecycle_for_segment(false, &terminal),
        ];
        assert_eq!(lifecycle.iter().filter(|(started, _)| *started).count(), 1);
        assert_eq!(
            lifecycle.iter().filter(|(_, finished)| *finished).count(),
            1
        );
    }

    #[test]
    fn resume_rejects_changed_bytecode_program_hash_with_typed_failure() {
        let output = validate_lashlang_program_hash(Some("sha256:old"), "sha256:current")
            .expect_err("changed bytecode identity must fail closed");
        assert!(matches!(
            *output,
            lash_core::ProcessAwaitOutput::Failure { code, .. }
                if code == "restate_segment_program_hash_mismatch"
        ));
    }

    #[test]
    fn declined_boundary_is_warned_and_counted() {
        let before = SEGMENT_BOUNDARY_DECLINED_TOTAL.load(Ordering::Relaxed);
        record_segment_boundary_decline(&"projected state", "test boundary decline");
        assert_eq!(
            SEGMENT_BOUNDARY_DECLINED_TOTAL.load(Ordering::Relaxed),
            before + 1
        );
    }
}