lash_core/runtime/process_worker/

mod.rs

use std::collections::BTreeSet;
use std::sync::Arc;

use tokio_util::sync::CancellationToken;

use super::effect::ProcessRunner;
use super::session_manager::RuntimeSessionServices;
use super::{EmbeddedRuntimeBuilder, ProcessWorkDriver, QueuedWorkDriver, RuntimeHostConfig};
use crate::{
    AbandonEvidence, AbandonWriter, InMemorySessionStore, LashRuntime, PluginError, PluginFactory,
    PluginHost, PluginStack, ProcessAwaitOutput, ProcessExecutionContext, ProcessInput,
    ProcessLease, ProcessLeaseCompletion, ProcessRecord, ProcessRegistration, ProcessRegistry,
    RecoveryDisposition, SessionStoreFactory,
};

/// Deployment-local configuration for rebuilding durable process executions.
///
/// Process rows intentionally carry only portable process input and provenance.
/// Workers provide plugins, providers, stores, secrets, and host capabilities
/// for the deployment that owns those rows.
#[derive(Clone)]
pub struct DurableProcessWorkerConfig {
    pub plugin_host: Arc<PluginHost>,
    pub runtime_host: RuntimeHostConfig,
    pub session_policy: crate::SessionPolicy,
    pub session_store_factory: Arc<dyn SessionStoreFactory>,
    pub process_registry: Arc<dyn ProcessRegistry>,
    pub process_change_hub: Option<crate::ProcessChangeHub>,
    pub trigger_store: Arc<dyn crate::TriggerStore>,
    pub process_work_driver: Option<ProcessWorkDriver>,
    pub queued_work_driver: Option<QueuedWorkDriver>,
    #[doc(hidden)]
    pub turn_phase_probe_slot: crate::runtime::RuntimeTurnPhaseProbeSlot,
    /// Residency for sessions the worker rebuilds to run a process. Defaults to
    /// [`Residency::KeepAll`]; a host running [`Residency::ActivePathOnly`] wires
    /// it here so the worker's rebuilt sessions trim to the active path too,
    /// instead of silently diverging from the live runtime by keeping the full
    /// graph resident.
    pub residency: crate::Residency,
    /// Owner identity stem this worker derives per-recovery lease owners from.
    ///
    /// Each recovery attempt claims with a unique `(owner_id, incarnation_id)`
    /// derived from this identity — a live lease held by an earlier attempt
    /// must fence a later sweep pass rather than be re-entered as the same
    /// incarnation — while the liveness metadata is inherited as-is. Defaults
    /// to a fresh opaque identity per config. Hosts that run one worker per OS
    /// process should wire a
    /// [`LeaseOwnerIdentity::local_process`](crate::LeaseOwnerIdentity::local_process)
    /// identity so peers on the same host can prove a crashed worker dead and
    /// reclaim its process leases before the TTL — mirroring the session
    /// execution lane's runtime lease owner.
    pub lease_owner: crate::LeaseOwnerIdentity,
}

impl DurableProcessWorkerConfig {
    pub fn new(
        plugin_host: Arc<PluginHost>,
        runtime_host: RuntimeHostConfig,
        session_store_factory: Arc<dyn SessionStoreFactory>,
        process_registry: Arc<dyn ProcessRegistry>,
    ) -> Self {
        let clock = Arc::clone(&runtime_host.clock);
        Self {
            plugin_host,
            runtime_host,
            session_policy: crate::SessionPolicy::default(),
            session_store_factory,
            process_registry,
            process_change_hub: None,
            trigger_store: Arc::new(crate::InMemoryTriggerStore::with_clock(clock)),
            process_work_driver: None,
            queued_work_driver: None,
            turn_phase_probe_slot: crate::runtime::RuntimeTurnPhaseProbeSlot::default(),
            residency: crate::Residency::default(),
            lease_owner: crate::LeaseOwnerIdentity::opaque(
                format!("durable-process-worker:{}", uuid::Uuid::new_v4()),
                uuid::Uuid::new_v4().to_string(),
            ),
        }
    }

    pub fn with_trigger_store(mut self, store: Arc<dyn crate::TriggerStore>) -> Self {
        self.trigger_store = store;
        self
    }

    pub fn with_session_policy(mut self, policy: crate::SessionPolicy) -> Self {
        self.session_policy = policy;
        self
    }

    pub fn with_residency(mut self, residency: crate::Residency) -> Self {
        self.residency = residency;
        self
    }

    pub fn with_process_work_driver(mut self, driver: ProcessWorkDriver) -> Self {
        self.process_work_driver = Some(driver);
        self
    }

    pub fn with_change_hub(mut self, hub: crate::ProcessChangeHub) -> Self {
        self.process_change_hub = Some(hub);
        self
    }

    /// Set the owner identity this worker presents when claiming process
    /// leases. See [`DurableProcessWorkerConfig::lease_owner`].
    pub fn with_lease_owner(mut self, lease_owner: crate::LeaseOwnerIdentity) -> Self {
        self.lease_owner = lease_owner;
        self
    }

    pub fn with_queued_work_driver(mut self, driver: QueuedWorkDriver) -> Self {
        self.queued_work_driver = Some(driver);
        self
    }

    #[doc(hidden)]
    pub fn with_turn_phase_probe_slot(
        mut self,
        slot: crate::runtime::RuntimeTurnPhaseProbeSlot,
    ) -> Self {
        self.turn_phase_probe_slot = slot;
        self
    }

    pub fn from_plugin_factories(
        plugin_factories: impl IntoIterator<Item = Arc<dyn PluginFactory>>,
        runtime_host: RuntimeHostConfig,
        session_store_factory: Arc<dyn SessionStoreFactory>,
        process_registry: Arc<dyn ProcessRegistry>,
    ) -> Self {
        Self::new(
            Arc::new(PluginHost::new(plugin_factories.into_iter().collect())),
            runtime_host,
            session_store_factory,
            process_registry,
        )
    }

    pub fn from_plugin_stack(
        plugin_stack: PluginStack,
        runtime_host: RuntimeHostConfig,
        session_store_factory: Arc<dyn SessionStoreFactory>,
        process_registry: Arc<dyn ProcessRegistry>,
    ) -> Self {
        Self::from_plugin_factories(
            plugin_stack.into_factories(),
            runtime_host,
            session_store_factory,
            process_registry,
        )
    }
}

/// Reconstructable background-process worker.
#[derive(Clone)]
pub struct DurableProcessWorker {
    config: Arc<DurableProcessWorkerConfig>,
}

/// Report from a graceful [owner drain](DurableProcessWorker::drain_owner_bound_work).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ProcessDrainReport {
    /// Process ids this host's own started `OwnerBound` work was terminalized as
    /// `Abandoned{OwnerDrain}` on, in the order they were drained.
    pub abandoned: Vec<String>,
}

/// Why a recovery run did not produce a terminal outcome under the lease.
enum RecoverFailure {
    /// The lease was lost mid-run (another owner reclaimed an expired lease).
    /// The losing worker must not write a terminal outcome — the new owner is
    /// now the single writer.
    LeaseLost(PluginError),
    /// The process could not be run (rebuild/store-facet failure). The lease is
    /// still held, so this worker terminalizes the row.
    Run(PluginError),
}

impl DurableProcessWorker {
    pub fn new(config: DurableProcessWorkerConfig) -> Self {
        Self {
            config: Arc::new(config),
        }
    }

    pub fn from_shared_config(config: Arc<DurableProcessWorkerConfig>) -> Self {
        Self { config }
    }

    pub fn config(&self) -> &DurableProcessWorkerConfig {
        &self.config
    }

    pub async fn run_process(
        &self,
        registration: ProcessRegistration,
        execution_context: ProcessExecutionContext,
        cancellation: CancellationToken,
    ) -> Result<ProcessAwaitOutput, PluginError> {
        let scoped_effect_controller = self
            .config
            .runtime_host
            .control
            .effect_host
            .scoped_static(crate::ExecutionScope::process(registration.id.clone()))
            .map_err(|err| PluginError::Session(err.to_string()))?
            .ok_or_else(|| {
                PluginError::Session(
                    "process worker effect host must provide a static process scope".to_string(),
                )
            })?;
        self.run_process_with_scoped_effect_controller(
            registration,
            execution_context,
            scoped_effect_controller,
            cancellation,
        )
        .await
    }

    pub async fn run_process_with_scoped_effect_controller(
        &self,
        registration: ProcessRegistration,
        execution_context: ProcessExecutionContext,
        scoped_effect_controller: crate::ScopedEffectController<'_>,
        cancellation: CancellationToken,
    ) -> Result<ProcessAwaitOutput, PluginError> {
        self.ensure_stable_process_id(&registration)?;
        self.ensure_durable_store_facets()?;
        // Externally-owned rows are never executed by lash (ADR 0019). Reject the
        // disposition before touching a runtime — the old fabricated-success path
        // for External inputs is deleted.
        if registration.disposition == RecoveryDisposition::ExternallyOwned {
            return Err(PluginError::Session(format!(
                "process `{}` is externally-owned and must not be executed by lash",
                registration.id
            )));
        }
        // Durable, lease-fenced "execution started" fact: recorded immediately
        // before executing so a later sweep can distinguish a started OwnerBound
        // row (never re-run) from an unstarted one (runnable by anyone). This is
        // the shared run path both the inline sweep and the Restate run handler
        // funnel through. First-writer-wins, so a re-invocation is a no-op.
        self.config
            .process_registry
            .record_first_started(
                &registration.id,
                crate::ProcessStarted {
                    owner: self.config.lease_owner.clone(),
                    started_at_ms: self.now_ms(),
                },
            )
            .await?;
        let mut runtime = self.runtime_for_registration(&registration).await?;
        let originator_scope = if let crate::ProcessOriginator::Session { scope } =
            &registration.provenance.originator
        {
            Some(scope)
        } else {
            None
        };
        let probe_scope = registration.wake_target.as_ref().or(originator_scope);
        if let Some(probe) =
            probe_scope.and_then(|scope| self.config.turn_phase_probe_slot.get_for_scope(scope))
        {
            runtime.set_turn_phase_probe(probe);
        }
        let manager = RuntimeSessionServices::new(&runtime, true, None).map_err(|err| {
            PluginError::Session(format!(
                "failed to build runtime env for process `{}`: {err}",
                registration.id
            ))
        })?;
        Ok(manager
            .run_process(
                registration,
                execution_context,
                Arc::clone(&self.config.process_registry),
                scoped_effect_controller,
                cancellation,
            )
            .await)
    }

    /// Sweep the registry for non-terminal processes and re-execute the ones
    /// this worker can claim, driving each to a terminal state.
    ///
    /// This is the crash-recovery counterpart to a worker that ran a process
    /// from a live turn: a trigger/trigger-started process whose worker
    /// died mid-flight is left non-terminal in the registry, and a subsequent
    /// worker reopening that registry must finish it. The sweep:
    ///
    /// 1. lists every non-terminal process ([`ProcessRegistry::list_non_terminal`]);
    /// 2. claims the durable single-owner [`ProcessLease`] over each — a process
    ///    already leased live by *another* owner is skipped (it is being run by
    ///    that owner right now) unless persisted liveness metadata proves that
    ///    owner definitely dead, in which case the lease is reclaimed with the
    ///    fenced CAS discipline of
    ///    [`ProcessRegistry::reclaim_process_lease`]; either way a non-terminal
    ///    process is re-run by exactly one owner (lease fencing);
    /// 3. runs the claimed process on this worker's wired controller, renewing
    ///    the lease across the long-running execution so a healthy recovery is
    ///    not swept out from under itself;
    /// 4. atomically writes the terminal outcome and releases the validated lease.
    ///
    /// Idempotent by `process_id`: terminal processes are never in the worklist,
    /// and a process that became terminal between the list and the claim is
    /// detected after claiming and skipped, so re-running a recovery sweep does
    /// not double-execute completed work.
    pub async fn drive_pending_processes(&self) -> Result<(), PluginError> {
        self.reconcile_trigger_deliveries().await?;
        let records = self.config.process_registry.list_non_terminal().await?;
        for record in records {
            // Run each claimed process on its OWN lease-fenced task. A sequential
            // drive that awaited each process to terminal would deadlock a process
            // that blocks awaiting a nested child (`start child` then `await`, or a
            // subagent fan-out): the one drive task would park inside the parent's
            // await and never claim the child. Spawning frees the loop so a
            // subsequent host-driven pass claims and runs the child, and the
            // per-process `ProcessLease` fences concurrent owners — so spawning a
            // task per pending row on every drive is idempotent (a row already
            // running is skipped on claim conflict) and one failing row never
            // aborts the rest of the sweep.
            let worker = self.clone();
            tokio::spawn(async move { worker.recover_process(record).await });
        }
        Ok(())
    }

    async fn reconcile_trigger_deliveries(&self) -> Result<(), PluginError> {
        let subscriptions = self
            .config
            .trigger_store
            .list_subscriptions(crate::TriggerSubscriptionFilter::default())
            .await?;
        if subscriptions.is_empty() {
            return Ok(());
        }
        let router = crate::TriggerRouter::new(
            Arc::clone(&self.config.trigger_store),
            Some(Arc::clone(&self.config.process_registry)),
            self.config.process_work_driver.clone(),
        );
        let mut seen = BTreeSet::new();
        let mut started_any = false;
        for subscription in subscriptions {
            let deliveries = self
                .config
                .trigger_store
                .list_deliveries_by_subscription_id(&subscription.subscription_id)
                .await?;
            for delivery in deliveries {
                let delivery_key = (
                    delivery.occurrence.occurrence_id.clone(),
                    delivery.subscription.subscription_id.clone(),
                );
                if !seen.insert(delivery_key) {
                    continue;
                }
                if self
                    .config
                    .process_registry
                    .get_process(&delivery.process_id)
                    .await
                    .is_some()
                {
                    continue;
                }
                let Some(scoped_effect_controller) = self
                    .config
                    .runtime_host
                    .control
                    .effect_host
                    .scoped_static(crate::ExecutionScope::runtime_operation(format!(
                        "trigger-delivery-reconcile:{}",
                        delivery.process_id
                    )))
                    .map_err(|err| PluginError::Session(err.to_string()))?
                else {
                    return Err(PluginError::Session(
                        "process worker effect host must provide a static trigger delivery reconcile scope"
                            .to_string(),
                    ));
                };
                match router
                    .start_delivery(
                        &delivery,
                        Arc::clone(&self.config.process_registry),
                        scoped_effect_controller.controller(),
                    )
                    .await
                {
                    Ok(()) => started_any = true,
                    Err(err) => tracing::warn!(
                        process_id = %delivery.process_id,
                        occurrence_id = %delivery.occurrence.occurrence_id,
                        subscription_id = %delivery.subscription.subscription_id,
                        error = %err,
                        "failed to reconcile trigger delivery",
                    ),
                }
            }
        }
        if started_any && let Some(driver) = self.config.process_work_driver.as_ref() {
            driver
                .claim_and_run_pending("trigger_delivery_reconcile")
                .await?;
        }
        Ok(())
    }

    /// Graceful owner drain: terminalize this host's own started `OwnerBound`
    /// work as `Abandoned{OwnerDrain}` at close (ADR 0019).
    ///
    /// This is an explicit **host lever on the worker**, never an implicit
    /// consequence of closing a session. Processes are global and outlive any
    /// one session ([ADR 0011]), so `LashSession::close`/`park` must not touch
    /// them; a host that wants its in-flight owner-bound work terminalized at
    /// shutdown calls this on the worker it is tearing down.
    ///
    /// Drain sequence (the operations runbook owns the surrounding steps; this
    /// is the terminal-writing step):
    /// 1. stop admitting new work to this worker;
    /// 2. cancel or await the worker's in-flight run tasks so they release their
    ///    per-run leases — for **Rerunnable** in-flight work that is the whole
    ///    story: stopping the local run task without any terminal write leaves
    ///    the row non-terminal so the next worker re-runs it (its contract);
    /// 3. call this lever: for every non-terminal **OwnerBound** row this exact
    ///    worker started (`first_started.owner == self.config.lease_owner`),
    ///    claim a fresh drain lease and, being the owner completing its own
    ///    work, write `Abandoned{OwnerDrain}` under it — the ordinary graceful
    ///    completion path, respecting the single-writer rule.
    ///
    /// A row still held by a live foreign lease (an in-flight run under one of
    /// this worker's own recovery incarnations that step 2 has not yet released)
    /// is deferred rather than reclaimed, so the drain never races a still-live
    /// run; such a row reaches `Abandoned` on the next drain pass or at a peer's
    /// recovery sweep. Rows started by a different owner, not-yet-started
    /// OwnerBound rows (still claimable by anyone), Rerunnable rows, and
    /// Externally-Owned rows are all left untouched.
    ///
    /// [ADR 0011]: durable process registration is session-independent.
    pub async fn drain_owner_bound_work(&self) -> Result<ProcessDrainReport, PluginError> {
        let mut abandoned = Vec::new();
        for record in self.config.process_registry.list_non_terminal().await? {
            if record.disposition != RecoveryDisposition::OwnerBound {
                continue;
            }
            let Some(first_started) = record.first_started.as_ref() else {
                // Never started: first execution is not re-execution, so any
                // worker may still claim it. Draining it would strand runnable
                // work as Abandoned.
                continue;
            };
            if first_started.owner != self.config.lease_owner {
                // Started by a different owner; not this host's to drain.
                continue;
            }
            let owner = first_started.owner.clone();
            if self.drain_one_owner_bound(&record.id, owner).await {
                abandoned.push(record.id);
            }
        }
        Ok(ProcessDrainReport { abandoned })
    }

    /// Terminalize one of this host's started OwnerBound rows as
    /// `Abandoned{OwnerDrain}` under a freshly claimed drain lease. Returns
    /// whether the terminal was written (`false` when the row is held by a live
    /// foreign lease, already terminal, or the claim failed).
    async fn drain_one_owner_bound(
        &self,
        process_id: &str,
        owner: crate::LeaseOwnerIdentity,
    ) -> bool {
        let lease_ttl_ms = self.lease_timings().ttl_ms();
        let drain_owner = self.recovery_lease_owner();
        let lease = match self
            .config
            .process_registry
            .claim_process_lease(process_id, &drain_owner, lease_ttl_ms)
            .await
        {
            Ok(crate::ProcessLeaseClaimOutcome::Acquired(lease)) => lease,
            // A live run still holds the lease, or the claim failed: defer.
            Ok(crate::ProcessLeaseClaimOutcome::Busy { .. }) | Err(_) => return false,
        };
        if self
            .config
            .process_registry
            .get_process(process_id)
            .await
            .is_some_and(|current| current.is_terminal())
        {
            self.release_or_log(&lease).await;
            return false;
        }
        let evidence = AbandonEvidence {
            writer: AbandonWriter::OwnerDrain,
            owner: Some(owner),
            epoch_ms: self.now_ms(),
        };
        self.complete_and_release(
            &lease,
            process_id,
            ProcessAwaitOutput::Abandoned {
                evidence: Box::new(evidence),
                control: None,
            },
        )
        .await;
        true
    }

    /// Unique lease owner for one recovery attempt.
    ///
    /// Derived from [`DurableProcessWorkerConfig::lease_owner`]: a fresh
    /// `(owner_id, incarnation_id)` per attempt keeps sweeps idempotent (a
    /// still-running attempt's live lease fences later passes instead of being
    /// re-entered as "own lease"), while the configured liveness metadata is
    /// inherited so peers can prove a crashed worker dead and reclaim.
    fn recovery_lease_owner(&self) -> crate::LeaseOwnerIdentity {
        let attempt = uuid::Uuid::new_v4();
        crate::LeaseOwnerIdentity {
            owner_id: format!("{}:recovery:{attempt}", self.config.lease_owner.owner_id),
            incarnation_id: attempt.to_string(),
            liveness: self.config.lease_owner.liveness.clone(),
        }
    }

    /// Recover one non-terminal row, obeying its declared recovery disposition
    /// (ADR 0019). The verdict per disposition:
    ///
    /// - **ExternallyOwned**: never claimed, never run. If a pending Abandon
    ///   Request is present it is reconciled into `Abandoned{reconciled_request}`.
    /// - **Rerunnable**: exactly today's behavior — claim, (re-)run, complete.
    /// - **OwnerBound, never started**: any worker may run it (first execution is
    ///   not re-execution); the runner records `first_started` before executing.
    /// - **OwnerBound, started**: never re-run. A provably-dead holder yields
    ///   `Abandoned{sweep}`; a merely silent/expired holder is left non-terminal
    ///   unless an Abandon Request is present and the lease has lapsed, which
    ///   yields `Abandoned{reconciled_request}`. Elapsed time alone never
    ///   terminalizes.
    ///
    /// Every Abandoned write goes through `complete_process_with_lease`, which
    /// atomically validates this sweep's fence, appends the terminal, and clears
    /// the lease so a revenant's stale token is rejected.
    async fn recover_process(&self, record: ProcessRecord) {
        let process_id = record.id.clone();
        // ExternallyOwned: lash never executes the row. The only recovery action
        // is reconciling a pending Abandon Request; there is no owner lease to
        // wait out.
        if record.disposition == RecoveryDisposition::ExternallyOwned {
            if record.abandon_request.is_some() {
                self.reconcile_externally_owned_abandon(&process_id).await;
            }
            return;
        }

        let lease_ttl_ms = self.lease_timings().ttl_ms();
        let owner = self.recovery_lease_owner();
        // Claim the single-owner lease, distinguishing a fenced reclaim of a
        // provably-dead holder (death evidence) from acquiring a free/expired
        // lease (no death evidence). A live, not-provably-dead holder or a claim
        // error leaves the row to its owner.
        let Some((lease, dead_holder)) = self
            .claim_for_recovery(&process_id, &owner, lease_ttl_ms)
            .await
        else {
            return;
        };
        // Terminal between the list and the claim. Idempotent by process_id: do
        // not re-execute or re-terminalize a finished process.
        if self
            .config
            .process_registry
            .get_process(&process_id)
            .await
            .is_some_and(|current| current.is_terminal())
        {
            self.release_or_log(&lease).await;
            return;
        }

        match record.disposition {
            // Rerunnable: claim, (re-)run, complete — exactly today's behavior.
            RecoveryDisposition::Rerunnable => self.run_and_complete(record, lease).await,
            RecoveryDisposition::OwnerBound if record.first_started.is_some() => {
                // Started OwnerBound work is NEVER re-run — abandonment is the
                // only recovery. `first_started`'s owner is the lapsed owner the
                // reconciled-request evidence names.
                let lapsed_owner = record
                    .first_started
                    .as_ref()
                    .map(|started| started.owner.clone());
                let evidence = if let Some(dead_holder) = dead_holder {
                    // Holder provably dead ⇒ Abandoned{sweep}.
                    Some(AbandonEvidence {
                        writer: AbandonWriter::Sweep,
                        owner: Some(dead_holder.owner),
                        epoch_ms: self.now_ms(),
                    })
                } else if record.abandon_request.is_some() {
                    // Silent/expired holder without death evidence, but an
                    // operator authorized abandonment and the lease has lapsed
                    // (we acquired a free/expired lease) ⇒ Abandoned{reconciled}.
                    Some(AbandonEvidence {
                        writer: AbandonWriter::ReconciledRequest,
                        owner: lapsed_owner,
                        epoch_ms: self.now_ms(),
                    })
                } else {
                    // No death evidence and no authorization: elapsed time alone
                    // never terminalizes. Leave the row non-terminal.
                    None
                };
                match evidence {
                    Some(evidence) => {
                        self.complete_and_release(
                            &lease,
                            &process_id,
                            ProcessAwaitOutput::Abandoned {
                                evidence: Box::new(evidence),
                                control: None,
                            },
                        )
                        .await;
                    }
                    None => self.release_or_log(&lease).await,
                }
            }
            // OwnerBound, never started: first execution is not re-execution, so
            // any worker may run it; the runner records first_started first.
            RecoveryDisposition::OwnerBound => self.run_and_complete(record, lease).await,
            // Filtered above; releasing keeps the lease honest if reached.
            RecoveryDisposition::ExternallyOwned => self.release_or_log(&lease).await,
        }
    }

    /// Wall-clock epoch ms from the worker's configured clock.
    fn now_ms(&self) -> u64 {
        self.config.runtime_host.clock.timestamp_ms()
    }

    /// Claim the recovery lease. Returns the acquired lease plus, when the claim
    /// fenced out a provably-dead holder, that holder as death evidence. Returns
    /// `None` when the row is held by a live (not provably-dead) owner or the
    /// claim fails — either way this pass leaves the row to its owner.
    async fn claim_for_recovery(
        &self,
        process_id: &str,
        owner: &crate::LeaseOwnerIdentity,
        lease_ttl_ms: u64,
    ) -> Option<(ProcessLease, Option<ProcessLease>)> {
        match self
            .config
            .process_registry
            .claim_process_lease(process_id, owner, lease_ttl_ms)
            .await
        {
            Ok(crate::ProcessLeaseClaimOutcome::Acquired(lease)) => Some((lease, None)),
            Ok(crate::ProcessLeaseClaimOutcome::Busy { holder })
                if holder.owner.is_definitely_dead_for_claimant(owner) =>
            {
                match self
                    .config
                    .process_registry
                    .reclaim_process_lease(process_id, owner, &holder, lease_ttl_ms)
                    .await
                {
                    Ok(crate::ProcessLeaseClaimOutcome::Acquired(lease)) => {
                        Some((lease, Some(holder)))
                    }
                    Ok(crate::ProcessLeaseClaimOutcome::Busy { .. }) | Err(_) => None,
                }
            }
            Ok(crate::ProcessLeaseClaimOutcome::Busy { .. }) | Err(_) => None,
        }
    }

    /// Reconcile a pending Abandon Request on an externally-owned row into an
    /// `Abandoned{reconciled_request}` terminal. Lash never executed the row, so
    /// there is no owner lease to wait out — but the sweep claims its own lease
    /// and completes through the atomic fenced path so it stays the single writer.
    async fn reconcile_externally_owned_abandon(&self, process_id: &str) {
        let lease_ttl_ms = self.lease_timings().ttl_ms();
        let owner = self.recovery_lease_owner();
        let lease = match self
            .config
            .process_registry
            .claim_process_lease(process_id, &owner, lease_ttl_ms)
            .await
        {
            Ok(crate::ProcessLeaseClaimOutcome::Acquired(lease)) => lease,
            // A concurrent writer holds the lease; let it finish.
            Ok(crate::ProcessLeaseClaimOutcome::Busy { .. }) | Err(_) => return,
        };
        if self
            .config
            .process_registry
            .get_process(process_id)
            .await
            .is_some_and(|current| current.is_terminal())
        {
            self.release_or_log(&lease).await;
            return;
        }
        let evidence = AbandonEvidence {
            writer: AbandonWriter::ReconciledRequest,
            // Externally-owned work has no lash execution owner to name.
            owner: None,
            epoch_ms: self.now_ms(),
        };
        self.complete_and_release(
            &lease,
            process_id,
            ProcessAwaitOutput::Abandoned {
                evidence: Box::new(evidence),
                control: None,
            },
        )
        .await;
    }

    /// (Re-)run a claimed row under its renewed lease and write the terminal
    /// outcome, the same live-owner-is-single-writer path used before ADR 0019.
    async fn run_and_complete(&self, record: ProcessRecord, lease: ProcessLease) {
        let process_id = record.id.clone();
        let registration = registration_from_record(record);
        let execution_context = ProcessExecutionContext::default();
        match self
            .run_process_with_lease_renewal(registration, execution_context, lease.clone())
            .await
        {
            // Ran to a terminal outcome (success or a process-level failure) while
            // holding the lease: this owner is the single writer of the terminal.
            Ok(output) => self.complete_and_release(&lease, &process_id, output).await,
            // The lease was lost mid-run — another owner reclaimed the expired
            // lease and is now running this process. Do NOT write a terminal
            // outcome or release the lease: that would race the new owner and
            // could record a succeeded process as Failed. Leave the row to the
            // lease holder; it will finish (or another sweep retries it).
            Err(RecoverFailure::LeaseLost(err)) => {
                tracing::warn!(
                    process_id = %process_id,
                    error = %err,
                    "process recovery lost its lease mid-run; deferring to the new owner",
                );
            }
            // The process could not be run at all (rebuild/store-facet failure):
            // terminalize as a recovery failure so the row leaves the worklist.
            Err(RecoverFailure::Run(err)) => {
                let output = ProcessAwaitOutput::Failure {
                    class: crate::ToolFailureClass::Execution,
                    code: "process_recovery_failed".to_string(),
                    message: err.to_string(),
                    raw: None,
                    control: None,
                };
                self.complete_and_release(&lease, &process_id, output).await;
            }
        }
    }

    /// Write a recovered process's terminal outcome and release its lease in one
    /// atomic fenced registry operation.
    async fn complete_and_release(
        &self,
        lease: &ProcessLease,
        process_id: &str,
        output: ProcessAwaitOutput,
    ) {
        // Refresh first so a healthy long-running worker has a full completion
        // window. The registry then validates this exact fence, appends the
        // terminal event, and clears the lease in one transaction. There is no
        // renew-then-unfenced-write gap for a stalled worker to cross.
        let fenced = match self
            .config
            .process_registry
            .renew_process_lease(lease, self.lease_timings().ttl_ms())
            .await
        {
            Ok(renewed) => renewed,
            Err(err) => {
                tracing::warn!(
                    process_id = %process_id,
                    error = %err,
                    "lost process lease before terminal write; deferring to the new owner",
                );
                return;
            }
        };
        if let Err(err) = self
            .config
            .process_registry
            .complete_process_with_lease(&fenced, output)
            .await
        {
            tracing::warn!(
                process_id = %process_id,
                error = %err,
                "failed to write recovered process terminal outcome",
            );
        }
    }

    async fn release_or_log(&self, lease: &ProcessLease) {
        if let Err(err) = self.release_process_lease(lease).await {
            tracing::warn!(
                process_id = %lease.process_id,
                error = %err,
                "failed to release recovered process lease",
            );
        }
    }

    /// Run a recovered process while renewing its lease across the execution,
    /// mirroring the turn-lease renewal that keeps a long-running effect's lease
    /// from expiring under the live owner.
    async fn run_process_with_lease_renewal(
        &self,
        registration: ProcessRegistration,
        execution_context: ProcessExecutionContext,
        mut lease: ProcessLease,
    ) -> Result<ProcessAwaitOutput, RecoverFailure> {
        let process_id = registration.id.clone();
        let cancellation = CancellationToken::new();
        let cancel_watcher = {
            let awaiter = self
                .config
                .process_change_hub
                .clone()
                .map(|hub| {
                    crate::ProcessAwaiter::new(Arc::clone(&self.config.process_registry), hub)
                })
                .unwrap_or_else(|| {
                    crate::ProcessAwaiter::polling(Arc::clone(&self.config.process_registry))
                });
            let process_id = process_id.clone();
            let cancellation = cancellation.clone();
            tokio::spawn(async move {
                match awaiter
                    .await_event(&process_id, "process.cancel_requested", 0)
                    .await
                {
                    Ok(_) => cancellation.cancel(),
                    Err(err) => tracing::warn!(
                        process_id = %process_id,
                        error = %err,
                        "process cancel watcher stopped before observing cancellation",
                    ),
                }
            })
        };
        let pending = self.run_process(registration, execution_context, cancellation.clone());
        tokio::pin!(pending);
        loop {
            tokio::select! {
                outcome = &mut pending => {
                    cancel_watcher.abort();
                    return outcome.map_err(RecoverFailure::Run);
                }
                _ = self.config.runtime_host.clock.sleep(self.lease_timings().renew_interval()) => {
                    match self
                        .config
                        .process_registry
                        .renew_process_lease(&lease, self.lease_timings().ttl_ms())
                        .await
                    {
                        Ok(renewed) => lease = renewed,
                        Err(err) => {
                            cancellation.cancel();
                            cancel_watcher.abort();
                            return Err(RecoverFailure::LeaseLost(err));
                        }
                    }
                }
            }
        }
    }

    fn lease_timings(&self) -> crate::LeaseTimings {
        self.config.runtime_host.control.lease_timings
    }

    async fn release_process_lease(&self, lease: &ProcessLease) -> Result<(), PluginError> {
        self.config
            .process_registry
            .complete_process_lease(&ProcessLeaseCompletion::from_lease(lease))
            .await
    }

    pub async fn request_process_cancel(
        &self,
        process_id: &str,
        reason: Option<String>,
    ) -> Result<(), PluginError> {
        self.config
            .process_registry
            .append_event(
                process_id,
                crate::ProcessEventAppendRequest::cancel_requested(process_id, reason),
            )
            .await
            .map(|_| ())
    }

    async fn runtime_for_registration(
        &self,
        registration: &ProcessRegistration,
    ) -> Result<LashRuntime, PluginError> {
        match registration.input.as_ref() {
            ProcessInput::SessionTurn { create_request, .. } => {
                self.runtime_for_session_turn(registration, create_request.as_ref())
                    .await
            }
            ProcessInput::ToolCall { .. } | ProcessInput::Engine { .. } => {
                self.runtime_for_process_env(registration).await
            }
            // Externally-owned rows are rejected before dispatch (ADR 0019), so an
            // External input has no execution runtime; fail loudly rather than
            // fabricate one.
            ProcessInput::External { .. } => Err(PluginError::Session(format!(
                "process `{}` is externally-owned and has no execution runtime",
                registration.id
            ))),
        }
    }

    async fn runtime_for_session_turn(
        &self,
        registration: &ProcessRegistration,
        create_request: &crate::SessionCreateRequest,
    ) -> Result<LashRuntime, PluginError> {
        let mut policy = create_request
            .policy
            .clone()
            .unwrap_or_else(|| self.config.session_policy.clone());
        if policy.recorded_provider_id().is_empty() {
            policy.provider_id = self.config.session_policy.provider_id.clone();
        }
        self.build_ephemeral_runtime(
            format!("process-session-turn:{}", registration.id),
            policy,
            create_request.plugin_options.clone(),
            "session turn request",
        )
        .await
    }

    async fn runtime_for_process_env(
        &self,
        registration: &ProcessRegistration,
    ) -> Result<LashRuntime, PluginError> {
        let Some(env_ref) = registration.env_ref.as_ref() else {
            return Err(PluginError::Session(format!(
                "process `{}` is missing a captured execution env",
                registration.id
            )));
        };
        let env = crate::load_process_execution_env(
            self.config
                .runtime_host
                .durability
                .process_env_store
                .as_ref(),
            env_ref,
        )
        .await?;
        self.build_ephemeral_runtime(
            format!("process-env:{}", registration.id),
            env.policy,
            env.plugin_options,
            env_ref.as_str(),
        )
        .await
    }

    async fn build_ephemeral_runtime(
        &self,
        session_id: String,
        policy: crate::SessionPolicy,
        plugin_options: crate::PluginOptions,
        source_label: &str,
    ) -> Result<LashRuntime, PluginError> {
        let store = Arc::new(InMemorySessionStore::default());
        let process_work_driver = self.config.process_work_driver.clone().unwrap_or_else(|| {
            if let Some(hub) = self.config.process_change_hub.clone() {
                ProcessWorkDriver::from_watched(
                    Arc::clone(&self.config.process_registry),
                    hub,
                    Arc::new(crate::InlineProcessRunHandle::new(self.clone())),
                )
            } else {
                ProcessWorkDriver::inline(Arc::clone(&self.config.process_registry), self.clone())
            }
        });
        let mut builder = EmbeddedRuntimeBuilder::new()
            .with_session_id(session_id.to_string())
            .with_plugin_host(self.config.plugin_host.as_ref().clone())
            .with_runtime_host(self.config.runtime_host.clone())
            .with_policy(policy)
            .with_plugin_options(plugin_options)
            .with_session_store_factory(Arc::clone(&self.config.session_store_factory))
            .with_trigger_store(Arc::clone(&self.config.trigger_store))
            .with_process_registry(Arc::clone(&self.config.process_registry))
            .with_process_work_driver(process_work_driver)
            .with_residency(self.config.residency)
            .with_store(store);
        if let Some(driver) = self.config.queued_work_driver.clone() {
            builder = builder.with_queued_work_driver(driver);
        }
        builder.build().await.map_err(|err| {
            PluginError::Session(format!(
                "failed to build process worker runtime for {source_label}: {err}"
            ))
        })
    }

    /// Enforce the durable-first wiring invariant at the worker process-run
    /// boundary: when the worker was wired with a durable effect host, every
    /// store it will execute against must also be durable. A durable host
    /// running against any ephemeral store fails loudly here rather than
    /// silently re-executing a process against non-durable state.
    ///
    /// Inline controllers (the default tier) impose no requirement, so
    /// inline/in-memory workers pass unchanged.
    fn ensure_durable_store_facets(&self) -> Result<(), PluginError> {
        if self
            .config
            .runtime_host
            .control
            .effect_host
            .durability_tier()
            != crate::DurabilityTier::Durable
        {
            return Ok(());
        }
        let require = |facet: crate::DurableStoreFacet| {
            PluginError::Session(crate::RuntimeError::durable_store_required(facet).to_string())
        };
        if self
            .config
            .runtime_host
            .durability
            .attachment_store
            .persistence()
            .durability_tier()
            != crate::DurabilityTier::Durable
        {
            return Err(require(crate::DurableStoreFacet::AttachmentStore));
        }
        if self
            .config
            .runtime_host
            .durability
            .process_env_store
            .durability_tier()
            != crate::DurabilityTier::Durable
        {
            return Err(require(crate::DurableStoreFacet::ProcessEnvStore));
        }
        if self.config.session_store_factory.durability_tier() != crate::DurabilityTier::Durable {
            return Err(require(crate::DurableStoreFacet::SessionStore));
        }
        if self.config.process_registry.durability_tier() != crate::DurabilityTier::Durable {
            return Err(require(crate::DurableStoreFacet::ProcessRegistry));
        }
        if self.config.trigger_store.durability_tier() != crate::DurabilityTier::Durable {
            return Err(require(crate::DurableStoreFacet::TriggerStore));
        }
        Ok(())
    }

    /// Enforce the stable-process-id invariant at every (re-)execution: process
    /// execution identity is the persisted `process_id`, so a retry — a Restate
    /// `run` re-invocation (keyed `LashProcessWorkflow/{process_id}`) or a
    /// recovery sweep re-running a non-terminal row — must present that stable
    /// id. An empty/fresh id has lost its idempotency anchor and is rejected
    /// loudly here, mirroring how `ExecutionScope` rejects an
    /// empty turn id at the durable-effect boundary.
    fn ensure_stable_process_id(
        &self,
        registration: &ProcessRegistration,
    ) -> Result<(), PluginError> {
        if registration.id.trim().is_empty() {
            return Err(PluginError::Session(
                crate::RuntimeError::missing_process_execution_id().to_string(),
            ));
        }
        Ok(())
    }
}

/// Rebuild a runnable registration from a persisted row, preserving its declared
/// disposition (ADR 0019).
fn registration_from_record(record: ProcessRecord) -> ProcessRegistration {
    ProcessRegistration {
        id: record.id,
        input: record.input,
        disposition: record.disposition,
        identity: record.identity,
        event_types: record.event_types,
        provenance: record.provenance,
        env_ref: record.env_ref,
        wake_target: record.wake_target,
    }
}

#[cfg(test)]
mod boundary_tests;
#[cfg(test)]
mod recovery_tests;