relux-runtime 0.8.0

pub mod registry;

use std::collections::HashMap;
use std::collections::HashSet;
use std::sync::Arc;

use tokio::sync::Mutex as TokioMutex;
use tokio::sync::watch;

use futures::future::join_all;

use crate::RuntimeContext;
use crate::cancel::CancelToken;
use crate::effect::registry::AcquiredEffect;
use crate::effect::registry::EffectGuard;
use crate::effect::registry::EffectHandle;
use crate::effect::registry::EffectInstanceKey;
use crate::effect::registry::EffectRegistry;
use crate::effect::registry::EffectSlot;
use crate::effect::registry::ExportedEffect;
use crate::effect::registry::ReleaseOutcome;
use crate::effect::registry::ShellInstanceKey;
use crate::effect::registry::ShellMap;
use crate::effect::registry::VarMap;
use crate::observe::structured::MatchContext;
use crate::observe::structured::SpanId;
use crate::observe::structured::SpanKind;
use crate::report::result::ExecError;
use crate::report::result::Failure;
use crate::report::result::FailureContext;
use crate::report::result::pure_eval_failure;
use crate::vm::Vm;
use crate::vm::context::ExecutionContext;
use crate::vm::context::Scope;
use crate::vm::context::ShellState;
use relux_core::pure::Env;
use relux_core::pure::LayeredEnv;
use relux_core::pure::LayeredEnvSource;
use relux_core::pure::VarScope;
use relux_ir::IrCleanupBlock;
use relux_ir::IrEffectItem;
use relux_ir::IrEffectStart;
use relux_ir::IrNode;

// --- Warning / CleanupSource -----------------------------

#[derive(Debug, Clone)]
pub enum CleanupSource {
    Test,
    Effect { name: String },
}

#[derive(Debug, Clone)]
pub enum Warning {
    CleanupFailed {
        source: CleanupSource,
        failure: ExecError,
    },
}

// --- Start scheduling -----------------------------------

/// Completion signal a start publishes for its dependents. Sent over a
/// per-start `watch` channel, which is multi-consumer and race-free: a
/// dependent that subscribes AFTER its dependency published still
/// observes the current value, so the publish-before-await window cannot
/// strand a waiter. Modeled as an enum so the illegal "succeeded but no
/// vars" / "failed but has vars" states are unrepresentable.
#[derive(Clone)]
enum StartSignal {
    /// The dependency is ready; carries its exposed vars (keyed by
    /// exposed name) for injection into dependents under `Alias.var`.
    Ready(VarMap),
    /// The dependency failed (or short-circuited on its own failed dep).
    Failed,
}

/// Per-start acquire outcome. Each driver future evaluates to one of
/// these; `join_all` collects them in input order for partitioning.
enum StartOutcome {
    /// The start acquired successfully.
    Ready {
        export: ExportedEffect,
        guard: EffectGuard,
    },
    /// The start hit a real overlay-eval or acquire error.
    Failed(ExecError),
    /// The start short-circuited because a dependency failed; it never
    /// evaluated its overlay. Carries no error of its own - the root
    /// cause is a `Failed` elsewhere in the same batch.
    DepFailed,
}

// --- EffectManager ---------------------------------------

pub struct EffectManager {
    registry: Arc<EffectRegistry>,
    pub(crate) rt_ctx: RuntimeContext,
    /// Guards for the test's direct effect acquires (`start E as a`
    /// at the top of a test). Drained by `cleanup_all`. Per-test by
    /// construction: each test instantiates its own `EffectManager`.
    top_level_guards: TokioMutex<Vec<EffectGuard>>,
    /// Root anchor for every `EffectCleanup` span emitted during this
    /// test's lifetime - happy-path teardown, mid-setup rollback, and
    /// `try_guards!`-driven partial teardown alike. Per-test by
    /// construction (same lifetime as `top_level_guards`). Threading
    /// this from `lib.rs` once at construction lets failure paths
    /// reach `test_span` directly without having to thread it through
    /// every recursive call alongside the setup-hierarchy parent.
    test_span: SpanId,
}

impl EffectManager {
    pub fn new(registry: Arc<EffectRegistry>, rt_ctx: RuntimeContext, test_span: SpanId) -> Self {
        Self {
            registry,
            rt_ctx,
            top_level_guards: TokioMutex::new(Vec::new()),
            test_span,
        }
    }

    /// Acquire all starts, honoring the wait-set edges enrichment
    /// derived from each start's overlay (`IrEffectStart::deps()`).
    /// `caller_vars` contains the caller's accumulated variable scope,
    /// allowing overlay expressions to reference the caller's `let` bindings.
    /// `caller_env` is the layered environment visible to the caller.
    /// Returns one `(ExportedEffect, EffectGuard)` per start declaration,
    /// in input order.
    ///
    /// Scheduling: each start runs as its own future. A start whose
    /// `deps()` is non-empty first awaits those sibling starts' completion
    /// signals (published over per-start `watch` channels); the ready
    /// deps' EXPOSED VARS are injected into the start's scope under
    /// `Alias.var` keys BEFORE its overlay is evaluated, so the overlay's
    /// `QualifiedVar` refs resolve. Independent starts (empty `deps()`)
    /// still bootstrap concurrently, exactly as before. Enrichment proved
    /// the dep graph acyclic, so the wait-set awaits cannot deadlock.
    ///
    /// The slot lock in `acquire` (see `registry::EffectSlot`) serialises
    /// bootstrap-vs-reuse for the same dedup key, so parallel acquires of
    /// overlapping keys are safe - one bootstraps, the others wait on the
    /// slot's `Notify`. Independent effects (different keys) bootstrap
    /// truly in parallel. Overlay evaluation runs inside each start's
    /// future (concurrently): a start's scope is fully materialized
    /// (caller scope + resolved dep vars) BEFORE its pure overlay eval
    /// runs, and pure eval has no cross-start side effects, so concurrent
    /// evaluation cannot change any overlay's value or its derived dedup
    /// key.
    ///
    /// Rollback stays all-or-nothing, generalized along dep edges: a
    /// failed dep fails its dependents (recorded as `DepFailed`), and on
    /// ANY failure every successful acquire is released concurrently
    /// before the first root error propagates.
    #[allow(clippy::type_complexity)]
    pub fn instantiate<'a>(
        &'a self,
        starts: &'a [IrEffectStart],
        caller_vars: &'a VarScope,
        caller_env: &'a Arc<LayeredEnv>,
        parent_span: SpanId,
        caller_captures: &'a std::collections::HashMap<String, String>,
    ) -> std::pin::Pin<
        Box<
            dyn std::future::Future<Output = Result<Vec<(ExportedEffect, EffectGuard)>, ExecError>>
                + Send
                + 'a,
        >,
    > {
        Box::pin(async move {
            let n = starts.len();

            // One `watch` channel per start carries that start's
            // completion signal to its dependents. `watch` is
            // multi-consumer and race-free: a dependent that subscribes
            // AFTER its dep published still sees the current value, so a
            // publish-before-await race cannot strand a waiter.
            let mut senders: Vec<watch::Sender<Option<StartSignal>>> = Vec::with_capacity(n);
            let mut receivers: Vec<watch::Receiver<Option<StartSignal>>> = Vec::with_capacity(n);
            for _ in 0..n {
                let (tx, rx) = watch::channel(None);
                senders.push(tx);
                receivers.push(rx);
            }

            let senders_ref = &senders;
            let receivers_ref = &receivers;

            // One driving future per start. A start awaits its `deps()`
            // wait-set, then evaluates its overlay against a scope
            // augmented with the ready deps' exposed vars, then acquires.
            // Each future EVALUATES TO its `StartOutcome`; `join_all`
            // returns them in input order, so no side-channel collector is
            // needed. A start that short-circuits on dep failure STILL
            // publishes a `Failed` signal so ITS dependents unblock instead
            // of hanging.
            let drivers = starts.iter().enumerate().map(|(i, start)| async move {
                // 1. Await every dependency's completion signal.
                //    Enrichment proved the graph acyclic, so this cannot
                //    deadlock.
                let mut dep_failed = false;
                let mut dep_vars: Vec<(usize, VarMap)> = Vec::with_capacity(start.deps().len());
                for &j in start.deps() {
                    let mut rx = receivers_ref[j].clone();
                    match rx.wait_for(Option::is_some).await {
                        Ok(v) => match v.clone().expect("wait_for predicate guarantees Some") {
                            StartSignal::Ready(vars) => dep_vars.push((j, vars)),
                            StartSignal::Failed => dep_failed = true,
                        },
                        // Unreachable in practice: senders live until this
                        // batch's `join_all` returns. Treat a dropped
                        // sender as a failed dep rather than panicking.
                        Err(_) => dep_failed = true,
                    }
                }

                // 2. A failed dependency means this start can never
                //    evaluate its overlay. Publish a failure signal (so
                //    transitive dependents short-circuit too) and yield a
                //    `DepFailed` outcome without acquiring.
                if dep_failed {
                    let _ = senders_ref[i].send(Some(StartSignal::Failed));
                    return StartOutcome::DepFailed;
                }

                // 3. Build this start's scope: the caller's scope
                //    augmented with each ready dep's exposed vars, injected
                //    under `Alias.var` keys - the exact flat-key shape the
                //    overlay's `QualifiedVar` refs resolve against (and the
                //    same shape `bootstrap_effect` injects for an effect's
                //    own deps). Only aliased deps contribute; deps only
                //    ever point at aliased starts by construction. With no
                //    deps we reuse the caller's scope untouched, preserving
                //    today's independent-start path (no clone).
                let augmented;
                let scope: &VarScope = if dep_vars.is_empty() {
                    caller_vars
                } else {
                    let mut aug = caller_vars.clone();
                    for (j, vars) in &dep_vars {
                        if let Some(alias) = starts[*j].alias() {
                            for (var_name, value) in vars {
                                aug.insert(format!("{alias}.{var_name}"), value.clone());
                            }
                        }
                    }
                    augmented = aug;
                    &augmented
                };

                // 4. Evaluate the overlay against the augmented scope and
                //    derive the dedup key.
                let evaluated = match self
                    .eval_overlay(start, scope, caller_env, parent_span, caller_captures)
                    .await
                {
                    Ok(e) => e,
                    Err(err) => {
                        let _ = senders_ref[i].send(Some(StartSignal::Failed));
                        return StartOutcome::Failed(err);
                    }
                };
                let expect_names: Vec<&str> = self
                    .rt_ctx
                    .tables
                    .effects
                    .get(start.effect())
                    .and_then(|r| r.as_ref().ok())
                    .map(|eff| eff.expects().iter().map(|e| e.name()).collect())
                    .unwrap_or_default();
                let key = EffectInstanceKey::from_expects(
                    start.effect().clone(),
                    &expect_names,
                    &evaluated,
                );

                // 5. Acquire, then publish this start's completion signal:
                //    its exposed vars (for dependents to inject). On failure
                //    publish `Failed`.
                match self
                    .acquire(&key, start, scope, caller_env, evaluated, parent_span)
                    .await
                {
                    Ok((acquired, guard)) => {
                        let _ =
                            senders_ref[i].send(Some(StartSignal::Ready(acquired.vars.clone())));
                        StartOutcome::Ready {
                            export: ExportedEffect {
                                key,
                                shells: acquired.shells,
                                vars: acquired.vars,
                            },
                            guard,
                        }
                    }
                    Err(err) => {
                        let _ = senders_ref[i].send(Some(StartSignal::Failed));
                        StartOutcome::Failed(err)
                    }
                }
            });

            // Drive ALL start futures to completion - never drop an
            // in-flight `acquire`, which would strand its slot in
            // `Loading` and stall future acquirers. `join_all` preserves
            // input order, so `outcomes` is already position-aligned.
            let outcomes: Vec<StartOutcome> = join_all(drivers).await;

            // Partition in input order. On ANY failure, release every
            // successful acquire (concurrently) and propagate the first
            // root error. Cleanup spans anchor under `self.test_span`,
            // never `parent_span` - on the recursive path `parent_span`
            // is the caller effect's open `EffectSetup` span, and nesting
            // cleanups inside it violates the invariant from 85eef51.
            let mut results: Vec<(ExportedEffect, EffectGuard)> = Vec::with_capacity(n);
            let mut first_error: Option<ExecError> = None;
            let mut failed = false;
            for outcome in outcomes {
                match outcome {
                    StartOutcome::Ready { export, guard } => results.push((export, guard)),
                    StartOutcome::Failed(err) => {
                        failed = true;
                        if first_error.is_none() {
                            first_error = Some(err);
                        }
                    }
                    // A `DepFailed` start's root cause is a `Failed`
                    // elsewhere in this same batch (a dep can only fail
                    // because it - or transitively ITS dep - hit a real
                    // acquire/overlay error), so `first_error` is always
                    // populated whenever any `DepFailed` is present.
                    StartOutcome::DepFailed => failed = true,
                }
            }

            if failed {
                // `.expect` guards the (unreachable) case of a failed
                // batch with no root error: panicking beats silently
                // returning a truncated success `Vec` that a caller's
                // `zip(starts)` would misalign.
                let failure = first_error
                    .expect("a failed start batch always carries a root acquire/overlay error");
                let releases = results
                    .into_iter()
                    .map(|(_export, guard)| self.release_and_teardown(guard, self.test_span));
                let _ = join_all(releases).await;
                return Err(failure);
            }

            Ok(results)
        })
    }

    /// Public top-level entry point used by the test runner. Acquires every
    /// `start` in `starts`, stashes the resulting guards on the
    /// `EffectManager` so `cleanup_all` can drain them, and returns the
    /// shells/vars exports for the caller's shell map.
    pub async fn instantiate_top_level(
        &self,
        starts: &[IrEffectStart],
        caller_vars: &VarScope,
        caller_env: &Arc<LayeredEnv>,
        caller_captures: &std::collections::HashMap<String, String>,
    ) -> Result<Vec<ExportedEffect>, ExecError> {
        let pairs = self
            .instantiate(
                starts,
                caller_vars,
                caller_env,
                self.test_span,
                caller_captures,
            )
            .await?;
        let mut top = self.top_level_guards.lock().await;
        let mut exported = Vec::with_capacity(pairs.len());
        for (ex, guard) in pairs {
            top.push(guard);
            exported.push(ex);
        }
        Ok(exported)
    }

    /// Drain the test's top-level guards and release each concurrently.
    /// The slot mutex + refcount guarantee that for each dedup'd slot,
    /// exactly one releaser sees `refcount == 0` and runs the cleanup
    /// body; other releasers return `None` and short-circuit.
    ///
    /// Every `EffectCleanup` span opened here is parented under
    /// `self.test_span`. Cleanups are operationally test-level activity
    /// (scheduled at test teardown), and the `EffectSetup` span has
    /// long since closed.
    pub async fn cleanup_all(&self) -> Vec<Warning> {
        let guards: Vec<EffectGuard> = std::mem::take(&mut *self.top_level_guards.lock().await);
        let futures = guards
            .into_iter()
            .map(|g| self.release_and_teardown(g, self.test_span));
        join_all(futures).await.into_iter().flatten().collect()
    }

    async fn acquire(
        &self,
        key: &EffectInstanceKey,
        start: &IrEffectStart,
        caller_vars: &VarScope,
        caller_env: &Arc<LayeredEnv>,
        evaluated_overlay: Env,
        parent_span: SpanId,
    ) -> Result<(AcquiredEffect, EffectGuard), ExecError> {
        let slot = self.registry.slot(key);
        // The slot lock is held only across state inspection and transitions
        // (`Empty -> Loading`, `Loading -> Ready/Failed`). `bootstrap_effect`
        // runs WITHOUT the slot lock, so concurrent acquirers that hit
        // `Loading` can wait without blocking the bootstrap task. Per-test
        // serial use means this lock-free window is dead code today, but
        // removing it would re-introduce a deadlock surface if instantiation
        // ever runs concurrently.
        let mut evaluated_overlay = Some(evaluated_overlay);
        loop {
            let mut guard = slot.lock().await;
            match &mut *guard {
                EffectSlot::Ready { refcount, handle } => {
                    *refcount += 1;
                    let acquired = AcquiredEffect {
                        shells: handle.exposed_shells(),
                        vars: handle.exposed_vars.clone(),
                    };
                    let marker = handle.marker.clone();
                    drop(guard);

                    // Emit a zero-duration reuse span under the caller's
                    // parent so the dedup hit is visible in the viewer.
                    // The marker matches the bootstrap span's marker -
                    // the viewer hops back by marker on pill click.
                    let overlay = evaluated_overlay
                        .take()
                        .expect("Ready slot reachable only once per acquire");
                    let reuse_span = self.rt_ctx.log.open_span(
                        SpanKind::EffectSetup {
                            effect: start.effect().name.to_string(),
                            overlay: Self::evaluated_overlay_pairs(&overlay),
                            alias: start.alias().map(String::from),
                            dep_sources: relux_ir::overlay_dep_sources(start),
                            marker,
                            is_reuse: true,
                        },
                        Some(parent_span),
                        Some(start.span()),
                    );
                    reuse_span.close();
                    return Ok((acquired, EffectGuard::new(slot.clone())));
                }
                EffectSlot::Failed(failure) => return Err(failure.clone()),
                EffectSlot::Loading(notify) => {
                    let notify = notify.clone();
                    drop(guard);
                    notify.notified().await;
                    // Slot is now Ready, Failed, or (rarely, on bootstrap
                    // panic in another task) still Loading. Loop and re-check.
                    continue;
                }
                EffectSlot::Empty => {
                    let notify = Arc::new(tokio::sync::Notify::new());
                    *guard = EffectSlot::Loading(notify.clone());
                    drop(guard);

                    // `Some` on the first iteration; the loop only continues
                    // through `Loading`, which doesn't consume the overlay.
                    let overlay = evaluated_overlay
                        .take()
                        .expect("Empty slot reachable only once per acquire");
                    let bootstrap_result = self
                        .bootstrap_effect(key, start, caller_vars, caller_env, overlay, parent_span)
                        .await;

                    let mut guard = slot.lock().await;
                    match bootstrap_result {
                        Ok(handle) => {
                            let acquired = AcquiredEffect {
                                shells: handle.exposed_shells(),
                                vars: handle.exposed_vars.clone(),
                            };
                            *guard = EffectSlot::Ready {
                                refcount: 1,
                                handle: Box::new(handle),
                            };
                            drop(guard);
                            notify.notify_waiters();
                            return Ok((acquired, EffectGuard::new(slot.clone())));
                        }
                        Err(failure) => {
                            self.rt_ctx.log.emit_error(
                                parent_span,
                                "",
                                "",
                                &failure.summary(),
                                None,
                            );
                            *guard = EffectSlot::Failed(failure.clone());
                            drop(guard);
                            notify.notify_waiters();
                            return Err(failure);
                        }
                    }
                }
            }
        }
    }

    async fn bootstrap_effect(
        &self,
        key: &EffectInstanceKey,
        start: &IrEffectStart,
        _caller_vars: &VarScope,
        caller_env: &Arc<LayeredEnv>,
        evaluated_overlay: Env,
        parent_span: SpanId,
    ) -> Result<EffectHandle, ExecError> {
        let marker = key.marker();
        let overlay_pairs = Self::evaluated_overlay_pairs(&evaluated_overlay);
        let setup_span = self.rt_ctx.log.open_span(
            SpanKind::EffectSetup {
                effect: start.effect().name.to_string(),
                overlay: overlay_pairs,
                alias: start.alias().map(String::from),
                dep_sources: relux_ir::overlay_dep_sources(start),
                marker: marker.clone(),
                is_reuse: false,
            },
            Some(parent_span),
            Some(start.span()),
        );
        self.rt_ctx.log.push_effect_setup(&start.effect().name.0);

        let effect_result = self
            .rt_ctx
            .tables
            .effects
            .get(start.effect())
            .ok_or_else(|| Failure::Runtime {
                message: format!("effect {:?} not found in table", start.effect()),
                span: start.effect_span().clone(),
                shell: None,
                context: FailureContext::pre_vm_with_span(setup_span.id()),
            })?;
        let effect = effect_result.as_ref().map_err(|e| Failure::Runtime {
            message: format!("effect resolution failed: {e:?}"),
            span: start.effect_span().clone(),
            shell: None,
            context: FailureContext::pre_vm_with_span(setup_span.id()),
        })?;
        let setup_span_id = setup_span.id();

        // 1. Create layered env from pre-evaluated overlay (inherits caller's env)
        let effect_env = Arc::new(LayeredEnv::child_with_source(
            caller_env.clone(),
            evaluated_overlay,
            LayeredEnvSource::EffectOverlay(marker.clone()),
        ));

        // 2. Create effect scope
        let scope = Scope::Effect {
            name: effect.name().name().to_string(),
            vars: Arc::new(TokioMutex::new(VarScope::new())),
            _timeout: None,
            env: effect_env.clone(),
        };

        // 3. Evaluate effect-level preamble (lets + pure-matches) into scope
        //    (parser enforces these come before starts). `body_captures`
        //    is hoisted across the whole preamble so a regex pure-match's
        //    `$n` captures flow into later lets, pure-matches, and the
        //    sub-dependency overlays instantiated in step 4.
        let mut body_captures: HashMap<String, String> = HashMap::new();
        let ec = MatchContext::EffectPreamble {
            name: start.effect().name.to_string(),
        };
        for item in effect.body() {
            match item {
                IrEffectItem::Let { stmt, span } => {
                    crate::preamble::eval_preamble_let(
                        &self.rt_ctx.log,
                        &effect_env,
                        &self.rt_ctx.tables.pure_fns,
                        &scope,
                        setup_span_id,
                        &ec,
                        stmt,
                        span,
                        &body_captures,
                    )
                    .await?;
                }
                IrEffectItem::PureMatch {
                    lhs,
                    pattern,
                    is_regex,
                    span,
                } => {
                    crate::preamble::eval_preamble_pure_match(
                        &self.rt_ctx.log,
                        &effect_env,
                        &self.rt_ctx.tables.pure_fns,
                        &scope,
                        setup_span_id,
                        &ec,
                        lhs,
                        pattern,
                        *is_regex,
                        span,
                        &mut body_captures,
                    )
                    .await?;
                }
                // Non-preamble items run in the body walk below.
                IrEffectItem::Comment { .. }
                | IrEffectItem::Expect { .. }
                | IrEffectItem::Start { .. }
                | IrEffectItem::Expose { .. }
                | IrEffectItem::Shell { .. }
                | IrEffectItem::Cleanup { .. } => {}
            }
        }

        // 4. Recursively instantiate sub-dependencies. Each pair = (export, guard).
        //    The `?` below is safe without guard release: `dep_guards` hasn't
        //    been populated yet, and `instantiate`'s own partial-batch
        //    rollback handles anything it acquired before failing. The
        //    effect's own cleanup body is also not invoked here - it can
        //    reference dep-exposed vars, and at this point deps don't exist,
        //    so there is nothing for cleanup to act on.
        let effect_vars = scope.vars().lock().await.clone();
        let exported_deps = self
            .instantiate(
                effect.starts(),
                &effect_vars,
                &effect_env,
                setup_span_id,
                &body_captures,
            )
            .await?;

        // From here on, `dep_guards` accumulates the guards for the
        // successfully-instantiated deps. Every fallible step between this
        // point and the final `Ok(EffectHandle { ... dep_guards ... })` is
        // wrapped in `try_guards!`, which runs this effect's own cleanup
        // body (if declared) and releases the accumulated dep guards via
        // `run_effect_cleanup` before propagating the error - matching the
        // success-path teardown order (effect's own cleanup before its
        // deps').

        let mut dep_shells: HashMap<String, ShellMap> = HashMap::new();
        let mut dep_vars: HashMap<String, VarMap> = HashMap::new();
        let mut dep_guards: Vec<EffectGuard> = Vec::with_capacity(exported_deps.len());
        let mut alias_to_effect_name: HashMap<String, String> = HashMap::new();
        for (sub_start, (exported, guard)) in effect.starts().iter().zip(exported_deps) {
            dep_guards.push(guard);
            if let Some(alias) = sub_start.alias() {
                dep_shells.insert(alias.to_string(), exported.shells);
                dep_vars.insert(alias.to_string(), exported.vars);
                alias_to_effect_name.insert(alias.to_string(), sub_start.effect().name.0.clone());
            }
        }

        // Pre-extract the cleanup block so it is available to `try_guards!`
        // failures that fire inside the body walk below (a shell-block
        // statement that fails before the body walk reaches the
        // `IrEffectItem::Cleanup` arm). The body walk no longer captures
        // this - see step 6.
        let cleanup_block: Option<IrCleanupBlock> = effect.body().iter().find_map(|item| {
            if let IrEffectItem::Cleanup { block, .. } = item {
                Some(block.clone())
            } else {
                None
            }
        });

        // 5b. Reset imported VMs into this scope's POV.
        let mut reset_seen = HashSet::new();
        for (alias, shells_map) in &dep_shells {
            let source_effect_name = alias_to_effect_name.get(alias).cloned();
            for (shell_local_name, vm_arc) in shells_map.iter() {
                let ptr = Arc::as_ptr(vm_arc) as usize;
                if reset_seen.insert(ptr) {
                    vm_arc.lock().await.reset_for_export(
                        scope.clone(),
                        Some(alias.clone()),
                        source_effect_name.clone(),
                        shell_local_name.clone(),
                    );
                }
            }
        }

        // Build local shells map, pre-populated with aliased dependency shells.
        // When a dependency is aliased (e.g. `start SetupDb as db`), its exported
        // shells are accessible by alias in the effect body (`shell db { ... }`
        // reuses the dependency's shell).
        let mut shells: HashMap<String, Arc<TokioMutex<Vm>>> = HashMap::new();
        for (alias, dep_exported) in &dep_shells {
            if dep_exported.len() == 1 {
                let vm_arc = dep_exported.values().next().unwrap().clone();
                shells.insert(alias.clone(), vm_arc);
            }
        }

        // Local helper: on any failure below, run this effect's own cleanup
        // body (best-effort) and release dep guards under that cleanup
        // span, then propagate. Warnings from the partial-teardown are
        // discarded - the test is failing anyway; surfacing extra cleanup
        // noise on top would obscure the root failure. Defined after the
        // `shells` binding because macro hygiene resolves `&shells`
        // against the binding visible at the macro's definition site.
        //
        // The final argument is `self.test_span`, NOT the local
        // `parent_span` - on the recursive sub-effect path, `parent_span`
        // is the grandparent effect's open EffectSetup span. Cleanup
        // spans must always anchor under the test span (85eef51).
        macro_rules! try_guards {
            ($e:expr) => {{
                match $e {
                    Ok(v) => v,
                    Err(failure) => {
                        let guards_taken = std::mem::take(&mut dep_guards);
                        // Pin the setup span's end_ts to the moment of
                        // failure, before awaiting cleanup. The `setup_span`
                        // SpanGuard local would otherwise stay alive through
                        // `run_effect_cleanup`'s shells-shutdown + cleanup-
                        // block phase and only drop when this function
                        // unwinds, leaving end_ts near test-end instead of
                        // setup-end.
                        self.rt_ctx.log.close_span(setup_span_id);
                        self.run_effect_cleanup(
                            effect.name().name(),
                            start.alias().map(String::from),
                            setup_span_id,
                            &marker,
                            key,
                            &scope,
                            &shells,
                            cleanup_block.as_ref(),
                            guards_taken,
                            self.test_span,
                        )
                        .await;
                        return Err(failure.into());
                    }
                }
            }};
        }

        // 5c. Inject dependency-exposed variables into the effect scope so
        //      they're accessible via ${Alias.var_name} in shell blocks.
        {
            let mut vars = scope.vars().lock().await;
            for (alias, var_map) in &dep_vars {
                for (var_name, value) in var_map {
                    vars.insert(format!("{alias}.{var_name}"), value.clone());
                }
            }
        }

        // 6. Walk IrEffectItems (lets already evaluated, starts already
        //    instantiated, cleanup block already extracted above).
        for item in effect.body() {
            match item {
                IrEffectItem::Comment { .. }
                | IrEffectItem::Expect { .. }
                | IrEffectItem::Start { .. }
                | IrEffectItem::Expose { .. }
                | IrEffectItem::Let { .. }
                | IrEffectItem::PureMatch { .. }
                | IrEffectItem::Cleanup { .. } => continue,
                IrEffectItem::Shell { block, .. } => {
                    let switch_span = block.name().span();
                    if let Some(qualifier) = block.qualifier() {
                        // Qualified: alias.shell { ... }
                        let alias = qualifier.name();
                        let shell_name = block.name().name();
                        let display = format!("{alias}.{shell_name}");
                        let block_span = self.rt_ctx.log.open_span(
                            SpanKind::ShellBlock {
                                shell: display.clone(),
                            },
                            Some(setup_span_id),
                            Some(switch_span),
                        );
                        let block_span_id = block_span.id();
                        let dep =
                            try_guards!(dep_shells.get(alias).ok_or_else(|| Failure::Runtime {
                                message: format!("unknown effect alias `{alias}`"),
                                span: qualifier.span().clone(),
                                shell: None,
                                context: FailureContext::pre_vm_with_span(block_span_id),
                            }));
                        let vm_arc =
                            try_guards!(dep.get(shell_name).ok_or_else(|| Failure::Runtime {
                                message: format!(
                                    "effect alias `{alias}` does not expose shell `{shell_name}`"
                                ),
                                span: block.name().span().clone(),
                                shell: None,
                                context: FailureContext::pre_vm_with_span(block_span_id),
                            }));
                        let exec_result = {
                            let mut vm = vm_arc.lock().await;
                            let vm_name = vm.current_name();
                            let vm_marker = vm.shell_marker().to_string();
                            self.rt_ctx.log.emit_shell_switch(
                                block_span_id,
                                &vm_name,
                                &vm_marker,
                                None,
                            );
                            vm.set_block_span(block_span_id);
                            vm.exec_stmts(block.body()).await
                            // vm lock drops at end of this block, BEFORE try_guards! awaits any
                            // release_and_teardown that would re-lock the same vm via
                            // teardown_effect::shutdown.
                        };
                        // Pin the shell-block's end_ts to "body done" before
                        // `try_guards!` may await `run_effect_cleanup`. The
                        // local `block_span` guard would otherwise stay on the
                        // stack through the entire cleanup phase and only
                        // close when this function unwinds, leaving the viewer
                        // with a shell-block that appears active alongside
                        // cleanup operations.
                        self.rt_ctx.log.close_span(block_span_id);
                        try_guards!(exec_result);
                        // block_span drops here as a no-op (already closed).
                    } else {
                        // Unqualified: shell name { ... }
                        let name = block.name().name().to_string();
                        let block_span = self.rt_ctx.log.open_span(
                            SpanKind::ShellBlock {
                                shell: name.clone(),
                            },
                            Some(setup_span_id),
                            Some(switch_span),
                        );
                        let block_span_id = block_span.id();
                        if !shells.contains_key(&name) {
                            let shell_state = ShellState::new(name.clone());
                            let ctx = ExecutionContext::new(
                                scope.clone(),
                                shell_state,
                                self.rt_ctx.shell.default_timeout.clone(),
                                self.rt_ctx.env.clone(),
                                block_span_id,
                            );
                            let shell_key = ShellInstanceKey::Effect {
                                effect: key.clone(),
                                shell_name: name.clone(),
                            };
                            let vm = try_guards!(
                                Vm::new(
                                    name.clone(),
                                    shell_key.marker(),
                                    ctx,
                                    &self.rt_ctx,
                                    block.span().clone(),
                                )
                                .await
                            );
                            shells.insert(name.clone(), Arc::new(TokioMutex::new(vm)));
                        }
                        let exec_result = {
                            let vm_arc = shells.get(&name).expect("shell just inserted above");
                            let mut vm = vm_arc.lock().await;
                            let display_name = vm.current_name();
                            let display_marker = vm.shell_marker().to_string();
                            self.rt_ctx.log.emit_shell_switch(
                                block_span_id,
                                &display_name,
                                &display_marker,
                                None,
                            );
                            vm.set_block_span(block_span_id);
                            vm.exec_stmts(block.body()).await
                            // vm lock drops at end of this block, BEFORE try_guards! awaits any
                            // release_and_teardown that would re-lock the same vm.
                        };
                        // Pin the shell-block's end_ts to "body done" before
                        // `try_guards!` may await `run_effect_cleanup`. See the
                        // matching comment in the qualified arm above.
                        self.rt_ctx.log.close_span(block_span_id);
                        try_guards!(exec_result);
                        // block_span drops here as a no-op (already closed).
                    }
                }
            }
        }

        // 7. Resolve expose declarations - mark which shells/vars are exposed
        let mut exposed: HashSet<String> = HashSet::new();
        let mut exposed_vars: HashMap<String, String> = HashMap::new();

        let effect_vars = scope.vars().lock().await;
        for expose in effect.exposes() {
            let exposed_name = expose.exposed_name().to_string();
            match expose.kind() {
                relux_ir::IrExposeKind::Shell => {
                    if let Some(qualifier) = expose.qualifier() {
                        let dep = try_guards!(dep_shells.get(qualifier).ok_or_else(|| {
                            Failure::Runtime {
                                message: format!(
                                    "effect `{}` expose references unknown alias `{}`",
                                    effect.name().name(),
                                    qualifier,
                                ),
                                span: expose
                                    .qualifier_span()
                                    .expect("qualified expose has a qualifier span")
                                    .clone(),
                                shell: None,
                                context: FailureContext::pre_vm_with_span(setup_span_id),
                            }
                        }));
                        let vm_arc = try_guards!(dep.get(expose.target()).ok_or_else(|| {
                            Failure::Runtime {
                                message: format!(
                                    "effect `{}` expose references shell `{}` not exposed by `{}`",
                                    effect.name().name(),
                                    expose.target(),
                                    qualifier,
                                ),
                                span: expose.target_span().clone(),
                                shell: None,
                                context: FailureContext::pre_vm_with_span(setup_span_id),
                            }
                        }));
                        shells.insert(exposed_name.clone(), vm_arc.clone());
                        exposed.insert(exposed_name.clone());
                    } else {
                        if !shells.contains_key(expose.target()) {
                            try_guards!(Err::<(), _>(Failure::Runtime {
                                message: format!(
                                    "effect `{}` expose references unknown shell `{}`",
                                    effect.name().name(),
                                    expose.target(),
                                ),
                                span: expose.target_span().clone(),
                                shell: None,
                                context: FailureContext::pre_vm_with_span(setup_span_id),
                            }));
                        }
                        if exposed_name != expose.target() {
                            let vm_arc = shells.get(expose.target()).unwrap().clone();
                            shells.insert(exposed_name.clone(), vm_arc);
                        }
                        exposed.insert(exposed_name.clone());
                    }
                    self.rt_ctx.log.emit_effect_expose_shell(
                        setup_span_id,
                        &exposed_name,
                        expose.target(),
                        expose.qualifier(),
                        None,
                    );
                }
                relux_ir::IrExposeKind::Var => {
                    let value = if let Some(qualifier) = expose.qualifier() {
                        // Re-expose a variable from a dependency
                        let qualifier_vars =
                            try_guards!(dep_vars.get(qualifier).ok_or_else(|| {
                                Failure::Runtime {
                                    message: format!(
                                        "effect `{}` expose references unknown alias `{}`",
                                        effect.name().name(),
                                        qualifier,
                                    ),
                                    span: expose
                                        .qualifier_span()
                                        .expect("qualified expose has a qualifier span")
                                        .clone(),
                                    shell: None,
                                    context: FailureContext::pre_vm_with_span(setup_span_id),
                                }
                            }));
                        try_guards!(qualifier_vars.get(expose.target()).ok_or_else(|| {
                            Failure::Runtime {
                                message: format!(
                                    "effect `{}` expose references var `{}` not exposed by `{}`",
                                    effect.name().name(),
                                    expose.target(),
                                    qualifier,
                                ),
                                span: expose.target_span().clone(),
                                shell: None,
                                context: FailureContext::pre_vm_with_span(setup_span_id),
                            }
                        }))
                        .clone()
                    } else {
                        // Expose a local let-bound variable
                        effect_vars.get(expose.target()).unwrap_or("").to_string()
                    };
                    exposed_vars.insert(exposed_name.clone(), value.clone());
                    self.rt_ctx.log.emit_effect_expose_var(
                        setup_span_id,
                        &exposed_name,
                        expose.target(),
                        expose.qualifier(),
                        &value,
                        None,
                    );
                }
            }
        }
        drop(effect_vars);

        // 8. Terminate non-exposed local shells (deduplicate by Arc pointer).
        //    Collect pointers of exposed VMs first - a non-exposed key may alias
        //    the same Arc as an exposed key (e.g. backwards-compat single-shell alias),
        //    so we must not shut those down.
        let exposed_ptrs: HashSet<usize> = shells
            .iter()
            .filter(|(k, _)| exposed.contains(k.as_str()))
            .map(|(_, v)| Arc::as_ptr(v) as usize)
            .collect();
        let non_exposed_keys: Vec<String> = shells
            .keys()
            .filter(|k| !exposed.contains(k.as_str()))
            .cloned()
            .collect();
        for key in non_exposed_keys {
            if let Some(vm_arc) = shells.remove(&key) {
                let ptr = Arc::as_ptr(&vm_arc) as usize;
                if !exposed_ptrs.contains(&ptr) {
                    vm_arc.lock().await.shutdown().await;
                }
            }
        }

        // setup_span drops here, closing the span.

        Ok(EffectHandle {
            scope,
            shells,
            exposed,
            exposed_vars,
            dep_guards,
            cleanup: cleanup_block,
            setup_span: setup_span_id,
            key: key.clone(),
            marker,
            alias: start.alias().map(String::from),
        })
    }

    /// Surface form of an evaluated overlay, used wherever a structured
    /// `EffectSetup` span needs the overlay as `(key, value)` pairs.
    /// Same conversion used by bootstrap and reuse paths so dedup'd
    /// acquires render identically to bootstraps.
    fn evaluated_overlay_pairs(overlay: &Env) -> Vec<(String, String)> {
        overlay
            .iter()
            .map(|(k, v)| (k.to_string(), v.to_string()))
            .collect()
    }

    async fn eval_overlay(
        &self,
        start: &IrEffectStart,
        caller_vars: &VarScope,
        caller_env: &Arc<LayeredEnv>,
        caller_span: SpanId,
        caller_captures: &std::collections::HashMap<String, String>,
    ) -> Result<Env, ExecError> {
        let mut overlay = Env::new();
        let mut sink =
            crate::observe::structured::log_sink::LogSink::new(&self.rt_ctx.log, caller_span);
        for entry in start.overlay() {
            let value = match relux_ir::evaluator::eval_pure_expr(
                entry.value(),
                caller_vars,
                caller_captures,
                caller_env,
                &self.rt_ctx.tables.pure_fns,
                &mut sink,
            ) {
                Ok(v) => v,
                Err(err) => {
                    let vars_in_scope = caller_vars.snapshot();
                    return Err(pure_eval_failure(
                        err,
                        caller_span,
                        MatchContext::EffectPreamble {
                            name: start.effect().name.to_string(),
                        },
                        vars_in_scope,
                        &sink,
                        &self.rt_ctx.log,
                    ));
                }
            };
            overlay.insert(entry.key().name().to_string(), value);
        }
        Ok(overlay)
    }

    /// Glue: release one guard, then either run its cleanup body (when
    /// this caller was the last holder) or open a zero-duration
    /// deferred-cleanup span (otherwise).
    fn release_and_teardown<'a>(
        &'a self,
        guard: EffectGuard,
        parent_span: SpanId,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Vec<Warning>> + Send + 'a>> {
        Box::pin(async move {
            match guard.release().await {
                ReleaseOutcome::LastHolder { handle } => {
                    self.teardown_effect(*handle, parent_span).await
                }
                ReleaseOutcome::Deferred {
                    effect,
                    alias,
                    setup_span,
                    marker,
                } => {
                    let deferred = self.rt_ctx.log.open_span(
                        SpanKind::EffectCleanup {
                            effect,
                            alias,
                            setup_span,
                            marker,
                            is_deferred: true,
                        },
                        Some(parent_span),
                        None,
                    );
                    deferred.close();
                    Vec::new()
                }
                ReleaseOutcome::Drift => Vec::new(),
            }
        })
    }

    /// Run cleanup for one effect we now exclusively own.
    ///
    /// Thin wrapper around `run_effect_cleanup`: destructures the handle
    /// and forwards the components. The partial-setup-failure path in
    /// `bootstrap_effect` calls `run_effect_cleanup` directly with the
    /// same fields collected from local state.
    async fn teardown_effect(&self, handle: EffectHandle, parent_span: SpanId) -> Vec<Warning> {
        let effect_name = handle.scope.name().to_string();
        self.run_effect_cleanup(
            &effect_name,
            handle.alias,
            handle.setup_span,
            &handle.marker,
            &handle.key,
            &handle.scope,
            &handle.shells,
            handle.cleanup.as_ref(),
            handle.dep_guards,
            parent_span,
        )
        .await
    }

    /// Run an effect's cleanup sequence.
    ///
    /// Sequence:
    ///   1. Open `EffectCleanup` span (parent = `parent_span`).
    ///   2. Shut down all owned VMs (deduplicated by Arc pointer).
    ///   3. If a cleanup block exists, run it inside a `CleanupBlock`
    ///      span; collect `Warning::CleanupFailed` on error.
    ///   4. Concurrently `release_and_teardown` every dep guard the
    ///      handle was holding (parented under the cleanup span).
    ///   5. Close cleanup span (after step 4 so deferred-cleanup spans
    ///      emitted by dep releases are well-ordered children).
    ///
    /// Used by both the success path (`teardown_effect`, via an
    /// `EffectHandle`) and the partial-setup-failure path
    /// (`bootstrap_effect`'s `try_guards!` macro, with the in-flight
    /// local state). The two call sites pass the same kind of
    /// information; collecting it once here keeps the cleanup
    /// semantics identical regardless of how the effect's lifecycle
    /// ended.
    #[allow(clippy::too_many_arguments)]
    async fn run_effect_cleanup(
        &self,
        effect_name: &str,
        alias: Option<String>,
        setup_span: SpanId,
        marker: &str,
        key: &EffectInstanceKey,
        scope: &Scope,
        shells: &HashMap<String, Arc<TokioMutex<Vm>>>,
        cleanup_block: Option<&IrCleanupBlock>,
        dep_guards: Vec<EffectGuard>,
        parent_span: SpanId,
    ) -> Vec<Warning> {
        let mut warnings = Vec::new();

        let cleanup_span = self.rt_ctx.log.open_span(
            SpanKind::EffectCleanup {
                effect: effect_name.to_string(),
                alias,
                setup_span,
                marker: marker.to_string(),
                is_deferred: false,
            },
            Some(parent_span),
            None,
        );
        let cleanup_span_id = cleanup_span.id();

        // Shut down all VMs (exposed and non-exposed, deduplicated).
        let mut seen = HashSet::new();
        for vm_arc in shells.values() {
            let ptr = Arc::as_ptr(vm_arc) as usize;
            if seen.insert(ptr) {
                vm_arc.lock().await.shutdown().await;
            }
        }

        // Run cleanup block in fresh shell (best-effort).
        if let Some(cleanup_block) = cleanup_block {
            let block_loc = cleanup_block.span();
            let block_span = self.rt_ctx.log.open_span(
                SpanKind::CleanupBlock,
                Some(cleanup_span_id),
                Some(block_loc),
            );
            let block_span_id = block_span.id();
            let cleanup_shell_key = ShellInstanceKey::Effect {
                effect: key.clone(),
                shell_name: "__cleanup".into(),
            };
            let cleanup_marker = cleanup_shell_key.marker();
            let cleanup_result = self
                .run_cleanup_block(cleanup_block, scope, &cleanup_marker, block_span_id)
                .await;
            if let Err(failure) = cleanup_result {
                self.rt_ctx.log.emit_warning(
                    block_span_id,
                    "__cleanup",
                    &cleanup_marker,
                    &format!("effect {effect_name} cleanup failed"),
                    None,
                );
                warnings.push(Warning::CleanupFailed {
                    source: CleanupSource::Effect {
                        name: effect_name.to_string(),
                    },
                    failure,
                });
            }
            // block_span drops here, closing the span.
        }

        // Concurrently release dep guards under our own cleanup span:
        // dep cleanups (including deferred-release spans for the
        // diamond's non-last holder) parent under this cleanup, not
        // under our caller. The diamond serialization still happens
        // inside `release` (atomic decrement under slot mutex);
        // join_all lets independent branches make progress.
        let dep_futures = dep_guards
            .into_iter()
            .map(|g| self.release_and_teardown(g, cleanup_span_id));
        let dep_warnings: Vec<Warning> =
            join_all(dep_futures).await.into_iter().flatten().collect();
        warnings.extend(dep_warnings);

        // Close cleanup_span AFTER the recursion so deferred-cleanup
        // spans emitted by dep releases (and nested final-cleanup spans
        // from dep release-to-zero) are well-ordered children.
        cleanup_span.close();
        self.rt_ctx.log.push_effect_teardown();

        warnings
    }

    async fn run_cleanup_block(
        &self,
        cleanup_block: &IrCleanupBlock,
        scope: &Scope,
        cleanup_marker: &str,
        block_span: SpanId,
    ) -> Result<(), ExecError> {
        let shell_state = ShellState::new("__cleanup".to_string());
        let ctx = ExecutionContext::new(
            scope.clone(),
            shell_state,
            self.rt_ctx.shell.default_timeout.clone(),
            self.rt_ctx.env.clone(),
            block_span,
        );
        // Cleanup uses its own uncancellable token
        let mut cleanup_rt_ctx = self.rt_ctx.clone();
        cleanup_rt_ctx.cancel = CancelToken::new();
        let mut vm = Vm::new(
            "__cleanup".to_string(),
            cleanup_marker.to_string(),
            ctx,
            &cleanup_rt_ctx,
            cleanup_block.span().clone(),
        )
        .await?;
        vm.exec_stmts(cleanup_block.body()).await?;
        vm.shutdown().await;
        Ok(())
    }
}