aion/engine/

api_workloop.rs

//! Workloop verbs on the engine API (workloop brief Leg 2): register, close
//! iteration, retire, hatch.

use std::collections::HashMap;
use std::future::Future;
use std::sync::Arc;

use aion_core::{
    Event, Payload, RunId, SearchAttributeValue, WORKFLOW_KIND_ATTRIBUTE, WORKLOOP_KIND,
    WorkflowId, WorkflowStatus, WorkloopSpec, status_from_events,
};
use aion_store::StoreError;
use aion_store::workloop::WorkloopStore;
use chrono::Utc;
use tokio::task::JoinHandle;

use super::api::Engine;
use crate::durability::Recorder;
use crate::error::EngineError;
use crate::workloop::hatch::{HatchOutcome, derive_identity};
use crate::workloop::iteration::WorkloopIterationClose;
use crate::workloop::service::WorkloopService;

/// The engine's running workloop machinery: the service, its store, and the
/// sweep task with its shutdown line. Present only when the builder was given
/// a workloop store and a sweep interval.
pub(crate) struct WorkloopEngineRuntime {
    pub(crate) service: Arc<WorkloopService>,
    pub(crate) store: Arc<dyn WorkloopStore>,
    pub(crate) shutdown: tokio::sync::watch::Sender<bool>,
    pub(crate) task: JoinHandle<()>,
    /// The NIF state holding this engine's `close_iteration/3` bridge, so
    /// shutdown can EMPTY that slot. See [`WorkloopEngineRuntime::stop`].
    pub(crate) nif_state: Arc<crate::runtime::EngineNifState>,
}

impl WorkloopEngineRuntime {
    /// Signal and abort the sweep task, then EMPTY the NIF bridge slot.
    ///
    /// Called from both `Engine::shutdown` and `Drop for Engine`, because a
    /// task spawned on the HOST runtime is not reached by the engine-task
    /// epoch gate.
    ///
    /// # 🔴 THE BRIDGE MUST BE RELEASED, OR THE STORE IS NEVER RELEASED
    ///
    /// The bridge holds an `IterationCloseContext` — the workloop store, the
    /// event store, the visibility store and the registry — and it lives in
    /// the NIF state, which outlives this engine. So an engine that installed
    /// a bridge and shut down left its EVENT STORE alive forever, and the next
    /// process to want that store waited on its file lock: a server that had
    /// ever built a workloop-capable engine could not hand its data directory
    /// to a successor. Measured as an indefinite hang in
    /// `recovery_declared_body_e2e` the moment the server wired the service
    /// unconditionally — a restart is exactly what that test performs.
    ///
    /// Emptying the slot is also the CORRECT runtime behaviour and not merely
    /// a leak fix: after shutdown there is no engine to close an iteration
    /// against, and an empty slot is precisely how the NIF says so
    /// (`no workloop service is configured on this engine`) instead of driving
    /// a close through a half-torn-down engine.
    pub(crate) fn stop(&self) {
        // A closed receiver means the task already exited; nothing to signal.
        let _ = self.shutdown.send(true);
        self.task.abort();
        crate::runtime::nif_workloop::release_workloop_nif_bridge(&self.nif_state);
    }
}

impl Engine {
    pub(crate) fn workloop_runtime(&self) -> Result<&WorkloopEngineRuntime, EngineError> {
        self.workloop.as_ref().ok_or_else(|| EngineError::Runtime {
            reason: "workloop service is not configured on this engine \
                     (EngineBuilder::with_workloop_service)"
                .to_owned(),
        })
    }

    /// The cadence service, when configured — the operational surface for
    /// sweeps and registration state.
    #[must_use]
    pub fn workloop_service(&self) -> Option<Arc<WorkloopService>> {
        self.workloop
            .as_ref()
            .map(|runtime| Arc::clone(&runtime.service))
    }

    /// The workloop spec the currently ROUTED deployment of `workflow_type`
    /// declares, or `None` when that deployment is an ordinary workflow.
    ///
    /// # 🔴 THE START PATH ASKS THIS, AND IT IS WHY A `.awl` WORKLOOP RUNS
    ///
    /// Every caller-facing start surface — HTTP, gRPC, the CLI — funnels into
    /// one start verb, and that verb cannot tell a workloop from a workflow by
    /// looking at the request: nothing in a start request says "this is a
    /// loop". The DEPLOYED PACKAGE says so, in the contract the compiler bound
    /// into its identity, and this is where that is read. Without it a
    /// workloop deploys, starts, runs its first iteration and is REFUSED at
    /// `close_iteration` for not being a registered workloop — which is
    /// exactly the shape this whole integration exists to close.
    ///
    /// Answering from the ROUTED version (not an exact pin) is deliberate: a
    /// start goes to whatever version routing would run, so the declaration a
    /// start is registered under must come from the same place.
    ///
    /// # Errors
    ///
    /// Propagates catalog failures, and refuses a declaration the engine
    /// cannot act on — an unarmed loop, an invariant with no tolerance or no
    /// confirming route, a retention window of zero. Those are refused HERE,
    /// before anything is registered or started, so the loop that cannot be
    /// armed never exists rather than existing and never firing.
    pub fn declared_workloop_spec(
        &self,
        workflow_type: &str,
    ) -> Result<Option<WorkloopSpec>, EngineError> {
        let Some(loaded) = self.catalog.routed(workflow_type)? else {
            return Ok(None);
        };
        // A package with no contract at all is a pre-`.v4` identity that
        // commits to none; it cannot be a workloop, because the header a loop
        // needs did not exist under those identities.
        let Ok(contract) = loaded.contract() else {
            return Ok(None);
        };
        let Some(workloop) = contract.workloop.as_ref() else {
            return Ok(None);
        };
        crate::workloop::spec_from_contract(workflow_type, workloop)
            .map(Some)
            .map_err(EngineError::from)
    }

    /// Registers a STARTED workflow as a workloop: stamps the `aion.kind`
    /// listing attribute durably in its history and arms the declared cadence
    /// and tolerance deadlines on the sweep set. Every declared value —
    /// arming, tolerance, retention — arrives validated inside `spec`; there
    /// are no defaults to assume.
    ///
    /// # Errors
    ///
    /// Refuses an unknown or non-Running workflow, a duplicate registration,
    /// and propagates store/append failures.
    pub async fn register_workloop(
        &self,
        loop_id: &WorkflowId,
        namespace: String,
        spec: WorkloopSpec,
    ) -> Result<(), EngineError> {
        let runtime = self.workloop_runtime()?;
        let history = self.store.read_history(loop_id).await?;
        if history.is_empty() {
            return Err(EngineError::InvalidState {
                reason: format!(
                    "workloop registration requires a started workflow; {loop_id} has no history"
                ),
            });
        }
        let status = status_from_events(&history);
        if status != WorkflowStatus::Running {
            return Err(EngineError::InvalidState {
                reason: format!(
                    "workloop registration requires a Running workflow; {loop_id} is {status:?}"
                ),
            });
        }
        // The duplicate check runs BEFORE the kind stamp so a refused
        // registration leaves history byte-identical (the service re-checks
        // under its own put, so a racing duplicate still cannot register
        // twice — this ordering only keeps the refusal append-free).
        if runtime
            .store
            .get_workloop(loop_id)
            .await
            .map_err(EngineError::from)?
            .is_some()
        {
            return Err(EngineError::InvalidState {
                reason: format!("workloop {loop_id} is already registered"),
            });
        }

        // Stamp the kind attribute through the loop's one Recorder so every
        // listing surface projects it (additive; no status change).
        let attributes = HashMap::from([(
            String::from(WORKFLOW_KIND_ATTRIBUTE),
            SearchAttributeValue::String(String::from(WORKLOOP_KIND)),
        )]);
        self.with_loop_recorder(loop_id, |recorder, _history| {
            let schema = Arc::clone(&self.search_attribute_schema);
            Box::pin(async move {
                recorder
                    .record_search_attributes_updated(Utc::now(), attributes, &schema)
                    .await
            })
        })
        .await?;

        runtime
            .service
            .register(loop_id.clone(), namespace, spec)
            .await
            .map_err(EngineError::from)?;
        Ok(())
    }

    /// Starts a workflow AS A WORKLOOP: seeds generation 1's carry, registers
    /// the loop, and only then lets the body run.
    ///
    /// # 🔴 THIS EXISTS BECAUSE START-THEN-REGISTER IS A RACE
    ///
    /// `register_workloop` requires an already-RUNNING workflow, so the only
    /// way to stand a loop up was to start it and then register it — and the
    /// started body begins executing immediately. Generation 1 could reach
    /// `close_iteration` BEFORE its registration landed, and the close would
    /// then refuse with "not a registered workloop" and FAIL the run. A loop's
    /// very first iteration could lose a race with its own registration.
    ///
    /// The registration row is therefore written FIRST, before the workflow
    /// exists, so the close always finds it. A failed start removes the row
    /// again rather than leaving a registration for a loop that never ran.
    ///
    /// # 🔴 AND BECAUSE GENERATION 1 HAS NO PREVIOUS ITERATION TO CARRY FROM
    ///
    /// Every later generation gets its carry from the previous iteration's
    /// `route start` payload. Generation 1 has none, and the compiled input
    /// codec requires the carry fields regardless — so a start payload that
    /// passes schema admission would then be UNDECODABLE by the workflow it
    /// was admitted for. The declared carry defaults are merged into the
    /// generation-1 input here, filling only ABSENT fields so a caller-supplied
    /// value is never clobbered by a default.
    ///
    /// # 🔴 THE KIND STAMP IS PART OF THE START, NOT A THIRD STEP
    ///
    /// The `aion.kind` attribute travels in the START's own attribute map, so
    /// `WorkflowStarted` and `SearchAttributesUpdated` land in ONE append
    /// (`Recorder::record_workflow_started_with_attributes`). It used to be a
    /// separate durable step after the start, and the gap was not cosmetic:
    /// boot recovery reads the kind FROM HISTORY to decide whether a Running
    /// workflow with no resident process is a crashed workflow to resurrect or
    /// a parked LOOP to leave alone. A crash between the start and the stamp
    /// left a registered loop that boot recovery did not recognise as one — it
    /// resurrected the generation resident while the sweep set still carried
    /// the row and would wake it, which is two paths driving one loop, the
    /// exact thing the skip exists to prevent.
    ///
    /// # 🔴 WHAT IS STILL NOT ATOMIC, SAID PLAINLY
    ///
    /// Two durable systems are involved — the workloop registration (a KV row)
    /// and the workflow's event history — and nothing spans them. The
    /// registration is written FIRST because the alternative loses the race
    /// above, so a crash between it and the start leaves a sweep-set row for a
    /// workflow with no history. That row is not left to alarm forever: the
    /// sweep refuses to fire at a workflow with no recorded start and reports
    /// the fault, and engine boot reconciliation
    /// ([`WorkloopService::withdraw_unstarted_registrations`]) withdraws every
    /// such row. Boot is the point at which the answer is unambiguous — no
    /// start can be in flight across a process boundary — which is why the
    /// reconciliation lives there rather than in the sweep, where it would
    /// race the birth window it is supposed to tolerate.
    ///
    /// [`WorkloopService::withdraw_unstarted_registrations`]:
    ///     crate::workloop::service::WorkloopService::withdraw_unstarted_registrations
    ///
    /// # Errors
    ///
    /// Refuses an unconfigured workloop service, a start payload that is not a
    /// JSON object when carry is declared, a duplicate registration, a caller
    /// attempting to set `aion.kind` itself, and propagates start failures.
    pub async fn start_workloop(
        &self,
        workflow_type: &str,
        input: Payload,
        search_attributes: HashMap<String, SearchAttributeValue>,
        namespace: String,
        spec: WorkloopSpec,
    ) -> Result<crate::registry::WorkflowHandle, EngineError> {
        let runtime = self.workloop_runtime()?;
        let seeded = spec
            .carry()
            .seed(&input)
            .map_err(|error| EngineError::InvalidState {
                reason: format!("seeding generation 1's carry refused: {error}"),
            })?;
        let attributes = with_workloop_kind(search_attributes)?;

        // The identity is minted here so the registration can precede the
        // start. Nothing observes it until the start records `WorkflowStarted`
        // under exactly this id.
        let loop_id = WorkflowId::new_v4();
        runtime
            .service
            .register(loop_id.clone(), namespace.clone(), spec)
            .await
            .map_err(EngineError::from)?;

        let started = self
            .start_workflow_with_id(
                workflow_type,
                seeded,
                attributes,
                namespace,
                Some(loop_id.clone()),
                None,
            )
            .await;
        match started {
            Ok(handle) => Ok(handle),
            Err(error) => {
                // No loop ran, so no registration should survive. A failure to
                // withdraw it is reported rather than swallowed: the sweep set
                // would otherwise carry a row for a workflow that does not
                // exist until the next boot reconciles it away.
                if let Err(cleanup) = runtime.service.deregister(&loop_id).await {
                    tracing::error!(
                        %loop_id,
                        start_error = %error,
                        cleanup_error = %cleanup,
                        "workloop start failed AND its registration could not be withdrawn; the \
                         sweep set carries a row for a workflow that was never started until \
                         boot reconciliation withdraws it"
                    );
                }
                Err(error)
            }
        }
    }

    /// Closes the current iteration at the continue-as-new boundary (R3.1):
    /// derives one health sample per invariant from the taken routes (R3.3),
    /// records `IterationClosed` + `WorkflowContinuedAsNew` + the successor
    /// generation's `WorkflowStarted` in ONE atomic batch through the loop's
    /// Recorder — spawning NO successor process (R13.3) — then installs the
    /// produced invariant current-state records with retention pruning (R7/
    /// R8) and feeds the samples into tolerance accounting. Returns the
    /// successor generation's run id.
    ///
    /// # Errors
    ///
    /// Refuses an unregistered loop, a terminal run, pending work, and
    /// undeclared invariants; propagates store/append failures.
    pub async fn close_workloop_iteration(
        &self,
        loop_id: &WorkflowId,
        close: WorkloopIterationClose,
    ) -> Result<RunId, EngineError> {
        let context = self.iteration_close_context()?;
        crate::workloop::close::close_iteration(&context, loop_id, close).await
    }

    /// The component set one iteration close needs, assembled from this
    /// engine. Shared with the `close_iteration/3` NIF bridge so a close
    /// reached from compiled workflow code and a close reached from this API
    /// verb are the SAME close.
    ///
    /// # Errors
    ///
    /// Refuses when no workloop service is configured on this engine.
    pub(crate) fn iteration_close_context(
        &self,
    ) -> Result<crate::workloop::close::IterationCloseContext, EngineError> {
        let runtime = self.workloop_runtime()?;
        Ok(crate::workloop::close::IterationCloseContext {
            workloop_store: Arc::clone(&runtime.store),
            service: Arc::clone(&runtime.service),
            store: self.store(),
            visibility_store: self.visibility_store(),
            registry: Arc::clone(&self.registry),
        })
    }

    /// Retires a workloop (R2.5): records `LoopRetired { reason }` and its
    /// `WorkflowCompleted` terminal in ONE atomic batch — the declared,
    /// recorded way to stop that is not failure — and removes the loop from
    /// the sweep set. Invariant current-state records survive indefinitely
    /// (R8.1).
    ///
    /// # Errors
    ///
    /// Refuses an unconfigured service and a terminal run; propagates
    /// store/append failures.
    pub async fn retire_workloop(
        &self,
        loop_id: &WorkflowId,
        reason: String,
        result: Payload,
    ) -> Result<(), EngineError> {
        self.retire_workloop_inner(loop_id, reason, result, RetireBody::Invoke)
            .await
    }

    /// [`Engine::retire_workloop`] for a loop that declares NO retire body.
    ///
    /// Kept as a separate verb rather than a flag on the main one because the
    /// difference is a DECLARATION, not a caller preference: a loop whose
    /// document declares `retire` must run it, and a caller must never be able
    /// to skip a declared cleanup by passing an argument. The AWL-driven path
    /// selects between them from the compiled contract; this exists so an
    /// operator retiring a bodyless loop is not forced through an entry probe
    /// that would refuse a module which correctly exports nothing.
    ///
    /// # Errors
    ///
    /// As [`Engine::retire_workloop`], minus the retire-body refusals.
    pub async fn retire_workloop_without_body(
        &self,
        loop_id: &WorkflowId,
        reason: String,
        result: Payload,
    ) -> Result<(), EngineError> {
        self.retire_workloop_inner(loop_id, reason, result, RetireBody::None)
            .await
    }

    /// Retires a workloop, taking the retire-body decision FROM ITS DEPLOYED
    /// DECLARATION rather than from the caller.
    ///
    /// # 🔴 THE OPERATOR SURFACE, AND WHY IT HAS NO "SKIP CLEANUP" FLAG
    ///
    /// [`Engine::retire_workloop`] and [`Engine::retire_workloop_without_body`]
    /// are two verbs precisely so a caller cannot choose: a document that
    /// declares `retire` must run it, and letting an argument skip a declared
    /// cleanup is how a lease is released twice or a queue is stranded. This
    /// is the verb every operator-facing surface calls, and it reads the
    /// answer out of the package the loop's CURRENT generation is pinned to —
    /// the same package whose module the body would be spawned from.
    ///
    /// A loop whose deployment declares no workloop surface at all (registered
    /// through the Rust API rather than deployed from a `.awl` document) is
    /// retired without a body: the engine has no declaration saying there is
    /// one, and inventing a `retire/1` probe for it would refuse every such
    /// loop for lacking an entry it was never meant to export.
    ///
    /// # Errors
    ///
    /// As [`Engine::retire_workloop`], plus catalog and history failures while
    /// resolving the loop's deployed declaration.
    pub async fn retire_declared_workloop(
        &self,
        loop_id: &WorkflowId,
        reason: String,
        result: Payload,
    ) -> Result<(), EngineError> {
        let history = self.store.read_history(loop_id).await?;
        let workflow_type = history
            .iter()
            .rev()
            .find_map(|event| match event {
                Event::WorkflowStarted { workflow_type, .. } => Some(workflow_type.clone()),
                _ => None,
            })
            .ok_or_else(|| EngineError::InvalidState {
                reason: format!("workloop {loop_id} has no recorded generation to retire"),
            })?;
        let declares_body = self
            .catalog
            .routed(&workflow_type)?
            .and_then(|loaded| loaded.contract().ok().cloned())
            .and_then(|contract| contract.workloop)
            .is_some_and(|workloop| workloop.has_retire_body);
        if declares_body {
            self.retire_workloop(loop_id, reason, result).await
        } else {
            self.retire_workloop_without_body(loop_id, reason, result)
                .await
        }
    }

    async fn retire_workloop_inner(
        &self,
        loop_id: &WorkflowId,
        reason: String,
        result: Payload,
        body: RetireBody,
    ) -> Result<(), EngineError> {
        let runtime = self.workloop_runtime()?;

        // 🔴 TERMINALITY IS CHECKED BEFORE THE BODY RUNS, NOT AFTER IT.
        //
        // The check used to live only inside the terminal append below, so
        // retiring an ALREADY-RETIRED loop ran the declared cleanup a second
        // time — releasing a released lease, re-draining a drained queue,
        // appending after `WorkflowCompleted` — and only then reported the
        // refusal it had already earned. A refusal that arrives after the
        // effects is not a refusal.
        //
        // The check inside `with_loop_recorder` stays: it is the one taken
        // under the same acquisition as the append, and it is what makes the
        // decision atomic rather than merely early. This one is what makes it
        // EFFECT-FREE.
        let history = self.store.read_history(loop_id).await?;
        crate::workloop::retire::refuse_if_terminal(loop_id, &history)
            .map_err(EngineError::from)?;

        // 🔴 THE RETIRE BODY RUNS FIRST, AND A MISSING ENTRY REFUSES LOUDLY.
        //
        // Before any terminal is recorded: a run that already holds its
        // terminal cannot append, so a body invoked afterwards could record
        // nothing it did. And a declared body whose entry is absent from the
        // deployed module is REFUSED — the engine cannot tell "compiled
        // before the retire entry existed" from "declared no cleanup", and
        // silently skipping the first is how a lease is lost.
        if body == RetireBody::Invoke {
            crate::workloop::retire::run_retire_body(&crate::workloop::retire::RetireInvocation {
                loop_id,
                runtime: &self.runtime,
                catalog: self.catalog.as_ref(),
                registry: &self.registry,
                store: &self.store,
                visibility_store: &self.visibility_store(),
            })
            .await
            .map_err(EngineError::from)?;
        }
        let retire_reason = reason;
        self.with_loop_recorder(loop_id, move |recorder, history| {
            let reason = retire_reason.clone();
            let result = result.clone();
            let terminal = aion_core::current_lease_terminal(history).is_some();
            Box::pin(async move {
                if terminal {
                    return Err(crate::durability::DurabilityError::HistoryShape {
                        reason: "cannot retire a workloop whose run is already terminal".to_owned(),
                    });
                }
                recorder
                    .record_loop_retired(Utc::now(), reason, result)
                    .await
            })
        })
        .await?;

        // 🔴 DEREGISTER BEFORE THE HANDLE GOES, AND ACCOUNT FOR THE ANSWER.
        //
        // Between the terminal above and the deregistration below the loop is
        // a REGISTERED row whose run holds a terminal — precisely the shape
        // the sweep reads as loop death (`sink.rs` refuses the cadence fire on
        // a terminal run, and `declare_loop_dead` then fans an `AlarmCause::
        // LoopDead` at every declared invariant). A cleanly retired loop would
        // be declared dead, durably, in its own history. The sweep now checks
        // for `LoopRetired` before declaring death, and this ordering narrows
        // the window it has to check in.
        //
        // The `bool` is not discarded: `deregister` answering `false` means
        // the row was already gone, which on this path means a sweep removed
        // it — the operator's retirement succeeded but the loop had already
        // been swept, and that is worth a line rather than a shrug.
        let was_registered = runtime
            .service
            .deregister(loop_id)
            .await
            .map_err(EngineError::from)?;
        if !was_registered {
            tracing::warn!(
                %loop_id,
                "workloop retirement recorded its terminal but the loop was no longer on the \
                 sweep set; a sweep deregistered it first, so its invariants may already carry \
                 loop-dead alarms for a loop that was being retired on purpose"
            );
        }
        if let Some(handle) = self.registry_handle(loop_id)? {
            let run = handle.run_id().clone();
            self.registry.remove(loop_id, &run)?;
        }
        Ok(())
    }

    /// Starts a DETACHED top-level workflow under the mandatory dedupe
    /// identity (R13.1): `WorkflowId = hatch_workflow_id(namespace, type,
    /// key)`. Not a child — no lifecycle tie, no supervision edge. A
    /// duplicate hatch is a recorded no-op returning the existing workflow id;
    /// two racing first-hatches are settled by the store's optimistic append,
    /// the loser resolving to the winner's workflow.
    ///
    /// # Errors
    ///
    /// Refuses empty/NUL identity parts and propagates start failures other
    /// than the dedupe race.
    pub async fn hatch_workflow(
        &self,
        namespace: &str,
        workflow_type: &str,
        key: &str,
        input: Payload,
        search_attributes: HashMap<String, SearchAttributeValue>,
    ) -> Result<HatchOutcome, EngineError> {
        let hatch_id = derive_identity(namespace, workflow_type, key).map_err(|error| {
            EngineError::InvalidState {
                reason: format!("hatch identity refused: {error}"),
            }
        })?;
        if !self.store.read_history(&hatch_id).await?.is_empty() {
            return Ok(HatchOutcome::Existing(hatch_id));
        }
        match self
            .start_workflow_with_id(
                workflow_type,
                input,
                search_attributes,
                namespace.to_owned(),
                Some(hatch_id.clone()),
                None,
            )
            .await
        {
            Ok(_handle) => Ok(HatchOutcome::Hatched(hatch_id)),
            // The dedupe race: a concurrent hatch of the same identity won the
            // first append. The store's optimistic concurrency IS the index —
            // the loser resolves to the existing workflow, never a second one.
            Err(EngineError::Store(StoreError::SequenceConflict { .. })) => {
                Ok(HatchOutcome::Existing(hatch_id))
            }
            Err(error) => Err(error),
        }
    }

    fn registry_handle(
        &self,
        loop_id: &WorkflowId,
    ) -> Result<Option<crate::registry::WorkflowHandle>, EngineError> {
        // aion#213: `with_loop_recorder` APPENDS DURABLY through whatever this
        // returns — see the twin in `workloop/close.rs`.
        self.registry.sole_handle(loop_id)
    }

    /// Append through the loop's ONE Recorder: the live handle's recorder
    /// when registered, a one-shot `Recorder::resume_at` when suspended (the
    /// sanctioned non-resident pattern). The closure receives the history
    /// read under the same acquisition, so check-then-append is not
    /// interleaved.
    async fn with_loop_recorder<T>(
        &self,
        loop_id: &WorkflowId,
        record: impl for<'a> FnOnce(
            &'a mut Recorder,
            &'a [Event],
        ) -> std::pin::Pin<
            Box<dyn Future<Output = Result<T, crate::durability::DurabilityError>> + Send + 'a>,
        >,
    ) -> Result<T, EngineError> {
        if let Some(handle) = self.registry_handle(loop_id)? {
            let recorder = handle.recorder();
            let mut recorder = recorder.lock().await;
            let history = self.store.read_history(loop_id).await?;
            let value = record(&mut recorder, &history).await?;
            return Ok(value);
        }
        let history = self.store.read_history(loop_id).await?;
        let head = history.iter().map(Event::seq).max().unwrap_or_default();
        let mut recorder = Recorder::resume_at(loop_id.clone(), self.store(), head);
        if let Some(run_id) = active_run_id(&history) {
            recorder = recorder.with_visibility(run_id, self.visibility_store());
        }
        let value = record(&mut recorder, &history).await?;
        Ok(value)
    }
}

/// Whether a retirement invokes the loop's declared retire body.
///
/// Not a boolean: `retire(loop, reason, result, true)` at a call site says
/// nothing about what `true` means, and this decision is load-bearing enough
/// that a reader must not have to look it up.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum RetireBody {
    /// The document declares a `retire` block; run it, and refuse if the
    /// deployed module exports no entry for it.
    Invoke,
    /// The document declares no `retire` block.
    None,
}

/// The caller's attributes plus the workloop kind stamp, refusing a caller
/// that tried to set the stamp itself.
///
/// The refusal is not fussiness. `aion.kind` is what boot recovery reads to
/// decide whether a Running workflow with no process is a crash to resurrect
/// or a park to leave alone, so a caller that could write it could make an
/// ordinary workflow claim to be a loop — or, worse, silently disagree with
/// the registration this very call is about to write.
fn with_workloop_kind(
    mut attributes: HashMap<String, SearchAttributeValue>,
) -> Result<HashMap<String, SearchAttributeValue>, EngineError> {
    if let Some(existing) = attributes.get(WORKFLOW_KIND_ATTRIBUTE) {
        return Err(EngineError::InvalidState {
            reason: format!(
                "`{WORKFLOW_KIND_ATTRIBUTE}` is stamped by the engine and cannot be supplied by \
                 a caller; it was given as {existing:?}. It is the attribute boot recovery reads \
                 to tell a parked workloop from a crashed workflow, and a caller-set value could \
                 disagree with the registration this start writes"
            ),
        });
    }
    attributes.insert(
        String::from(WORKFLOW_KIND_ATTRIBUTE),
        SearchAttributeValue::String(String::from(WORKLOOP_KIND)),
    );
    Ok(attributes)
}

fn active_run_id(history: &[Event]) -> Option<RunId> {
    history.iter().rev().find_map(|event| match event {
        Event::WorkflowStarted { run_id, .. } => Some(run_id.clone()),
        _ => None,
    })
}