aion-rs 0.31.0

//! Workloop engine integration (workloop brief Leg 2): registration and the
//! listing kind, the engine-side cadence dead-man, the atomic iteration
//! boundary with deferred successor, declared retirement, hatch dedupe — and
//! the R13.3 acceptance MEASUREMENT: a thousand sleeping workloops cost the
//! engine nothing it can be charged for.
//!
//! These tests run against a real built engine (beamr runtime, in-memory
//! stores) with histories seeded through Recorders — no deployed package, so
//! the wake path's process respawn is exercised up to its catalog resolution
//! and reported as a sweep fault; the durable halves (fires, alarms,
//! boundaries, retirement) are asserted at the history bytes.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};

use aion::durability::{Recorder, WorkflowStartRecord};
use aion::workloop::{HatchOutcome, WorkloopIterationClose};
use aion::{Engine, EngineBuilder};
use aion_core::{
    AlarmCause, ContentType, Event, InvariantSpec, PackageVersion, Payload, RunId, SortDirection,
    ToleranceSpec, WorkflowId, WorkflowListFilter, WorkflowListRequest, WorkflowSort,
    WorkflowSortField, WorkflowStatus, WorkloopArming, WorkloopSpec,
};
use aion_store::workloop::WorkloopStore;
use aion_store::{EventStore, InMemoryStore, ReadableEventStore};
use chrono::Utc;

type TestResult = Result<(), Box<dyn std::error::Error>>;

/// The background sweeper is kept quiet (one-hour interval) so every test
/// drives ticks deterministically through the service handle.
const QUIET_SWEEP: Duration = Duration::from_secs(3600);

async fn build_engine(store: &Arc<InMemoryStore>) -> Result<Engine, Box<dyn std::error::Error>> {
    Ok(EngineBuilder::new()
        .stop_drain_timeout(std::time::Duration::from_secs(5))
        .store_arc(Arc::clone(store) as Arc<dyn EventStore>)
        .in_memory_visibility()
        .scheduler_threads(1)
        .with_workloop_service(Arc::clone(store) as Arc<dyn WorkloopStore>, QUIET_SWEEP)
        .build()
        .await?)
}

async fn seed_started_workflow(
    store: &Arc<InMemoryStore>,
    loop_id: &WorkflowId,
) -> Result<RunId, Box<dyn std::error::Error>> {
    let run_id = RunId::new_v4();
    let mut recorder = Recorder::new(loop_id.clone(), Arc::clone(store) as Arc<dyn EventStore>);
    recorder
        .record_workflow_started(
            Utc::now(),
            WorkflowStartRecord {
                workflow_type: String::from("queue_watch"),
                input: Payload::new(ContentType::Json, b"{}".to_vec()),
                run_id: run_id.clone(),
                parent_run_id: None,
                parent_workflow_id: None,
                package_version: PackageVersion::new("a".repeat(64)),
            },
        )
        .await?;
    Ok(run_id)
}

/// Every row the engine lists in the default namespace, oldest start first.
async fn list_default_namespace(
    engine: &Engine,
) -> Result<Vec<aion_core::WorkflowSummary>, Box<dyn std::error::Error>> {
    let page = engine
        .list_workflows(&WorkflowListRequest {
            namespace: String::from(aion_core::DEFAULT_NAMESPACE),
            filter: WorkflowListFilter::default(),
            sort: WorkflowSort {
                field: WorkflowSortField::StartedAt,
                direction: SortDirection::Asc,
            },
            cursor: None,
            limit: 100,
        })
        .await?;
    Ok(page.items)
}

fn cadence_spec(
    period: Duration,
    tolerance: ToleranceSpec,
) -> Result<WorkloopSpec, Box<dyn std::error::Error>> {
    Ok(WorkloopSpec::new(
        WorkloopArming::every(period)?,
        vec![InvariantSpec {
            name: String::from("serving"),
            record_type: String::from("ServeState"),
            tolerance,
            confirms: vec![String::from("sweep")],
        }],
        Duration::from_secs(14 * 86_400),
    )?)
}

fn event_kinds(history: &[Event]) -> Vec<&'static str> {
    history
        .iter()
        .map(|event| match event {
            Event::WorkflowStarted { .. } => "WorkflowStarted",
            Event::WorkflowCompleted { .. } => "WorkflowCompleted",
            Event::WorkflowContinuedAsNew { .. } => "WorkflowContinuedAsNew",
            Event::SearchAttributesUpdated { .. } => "SearchAttributesUpdated",
            Event::CadenceFired { .. } => "CadenceFired",
            Event::IterationClosed { .. } => "IterationClosed",
            Event::LoopRetired { .. } => "LoopRetired",
            Event::InvariantUnconfirmed { .. } => "InvariantUnconfirmed",
            _ => "other",
        })
        .collect()
}

/// 🔴 THE R13.3 ACCEPTANCE, MEASURED: a thousand sleeping workloops are store
/// bytes plus sweep-set rows. Structurally: registration spawns no process,
/// no per-loop task, and a sweep over 1000 sleeping loops evaluates ZERO of
/// them and appends NOTHING. Measured: the per-tick sweep cost with 1000
/// sleeping loops, against the same engine shape with zero loops, printed and
/// bounded — the sweeper's duty cycle stays indistinguishable from idle.
#[tokio::test(flavor = "multi_thread")]
async fn a_thousand_sleeping_workloops_cost_the_engine_nothing_measured() -> TestResult {
    // World A: 1000 sleeping loops (one-hour cadence: never due in this test).
    let store_a = Arc::new(InMemoryStore::default());
    let engine_a = build_engine(&store_a).await?;
    let spec_period = Duration::from_secs(3600);
    let mut loop_ids = Vec::with_capacity(1000);
    for _ in 0..1000 {
        let loop_id = WorkflowId::new_v4();
        seed_started_workflow(&store_a, &loop_id).await?;
        engine_a
            .register_workloop(
                &loop_id,
                String::from("default"),
                cadence_spec(spec_period, ToleranceSpec::count(3))?,
            )
            .await?;
        loop_ids.push(loop_id);
    }
    let service_a = engine_a
        .workloop_service()
        .ok_or("workloop service must be configured")?;

    // World B: the identical engine shape with ZERO loops — the control arm.
    let store_b = Arc::new(InMemoryStore::default());
    let engine_b = build_engine(&store_b).await?;
    let service_b = engine_b
        .workloop_service()
        .ok_or("workloop service must be configured")?;

    // Structural half: a sweep sees no due loop, fires nothing, wakes nothing.
    let report = service_a.tick().await?;
    assert_eq!(report.swept, 0, "sleeping loops must not be swept");
    assert!(report.fired.is_empty());
    assert!(report.alarms.is_empty());
    assert!(report.faults.is_empty());
    // No per-loop history mutation happened after registration: exactly the
    // seeded start plus the kind stamp, for every loop.
    for loop_id in loop_ids.iter().take(10) {
        let history = store_a.read_history(loop_id).await?;
        assert_eq!(
            event_kinds(&history),
            vec!["WorkflowStarted", "SearchAttributesUpdated"],
            "a sleeping loop's history must not move"
        );
    }

    // Measured half: per-tick sweep cost, 1000 sleeping loops vs zero.
    let ticks: u32 = 200;
    let started = Instant::now();
    for _ in 0..ticks {
        let report = service_a.tick().await?;
        assert_eq!(report.swept, 0);
    }
    let with_thousand = started.elapsed() / ticks;

    let started = Instant::now();
    for _ in 0..ticks {
        service_b.tick().await?;
    }
    let with_zero = started.elapsed() / ticks;

    println!(
        "R13.3 MEASURED: sweep tick with 1000 sleeping loops = {with_thousand:?}, \
         with 0 loops = {with_zero:?} (sweep interval floor 1s)"
    );
    // The bound is generous and absolute: even at a 1-second production sweep
    // interval, 1000 sleeping loops must keep the sweeper's duty cycle under
    // half a percent — engine load indistinguishable from idle. (Measured
    // values on this rig are microseconds; the bound only catches a per-loop
    // resident process or per-loop task sneaking back in.)
    assert!(
        with_thousand < Duration::from_millis(5),
        "1000 sleeping loops cost {with_thousand:?} per sweep tick — not indistinguishable from idle"
    );

    engine_a.shutdown()?;
    engine_b.shutdown()?;
    Ok(())
}

#[tokio::test(flavor = "multi_thread")]
async fn registration_stamps_the_listing_kind_additively() -> TestResult {
    let store = Arc::new(InMemoryStore::default());
    let engine = build_engine(&store).await?;
    let loop_id = WorkflowId::new_v4();
    seed_started_workflow(&store, &loop_id).await?;
    engine
        .register_workloop(
            &loop_id,
            String::from("default"),
            cadence_spec(Duration::from_secs(3600), ToleranceSpec::count(3))?,
        )
        .await?;

    let summaries = list_default_namespace(&engine).await?;
    let summary = summaries
        .iter()
        .find(|summary| summary.workflow_id == loop_id)
        .ok_or("registered loop must list")?;
    assert_eq!(summary.kind.as_deref(), Some("workloop"));
    assert_eq!(summary.status, WorkflowStatus::Running);

    // Fixture pair: an unregistered workflow keeps NO kind. The engine's start
    // path projects a run's row the moment its start is durable; a seeded
    // history is written around that path, so the fixture projects it the
    // same way rather than listing a run production would never leave rowless.
    let plain_id = WorkflowId::new_v4();
    let plain_run = seed_started_workflow(&store, &plain_id).await?;
    aion::lifecycle::visibility::upsert_workflow_visibility(
        Arc::clone(&store) as Arc<dyn EventStore>,
        engine.visibility_store(),
        &plain_id,
        &plain_run,
    )
    .await?;
    let summaries = list_default_namespace(&engine).await?;
    let plain = summaries
        .iter()
        .find(|summary| summary.workflow_id == plain_id)
        .ok_or("plain workflow must list")?;
    assert_eq!(plain.kind, None);

    // Refusal pair: duplicate registration and unknown workflow.
    let duplicate = engine
        .register_workloop(
            &loop_id,
            String::from("default"),
            cadence_spec(Duration::from_secs(3600), ToleranceSpec::count(3))?,
        )
        .await;
    assert!(duplicate.is_err(), "duplicate registration must refuse");
    let unknown = engine
        .register_workloop(
            &WorkflowId::new_v4(),
            String::from("default"),
            cadence_spec(Duration::from_secs(3600), ToleranceSpec::count(3))?,
        )
        .await;
    assert!(
        unknown.is_err(),
        "registering an unstarted workflow must refuse"
    );

    engine.shutdown()?;
    Ok(())
}

#[tokio::test(flavor = "multi_thread")]
async fn cadence_fires_and_the_deadman_alarms_in_recorded_history() -> TestResult {
    let store = Arc::new(InMemoryStore::default());
    let engine = build_engine(&store).await?;
    let loop_id = WorkflowId::new_v4();
    seed_started_workflow(&store, &loop_id).await?;
    // Tolerance zero: the first missed window exceeds it.
    engine
        .register_workloop(
            &loop_id,
            String::from("default"),
            cadence_spec(Duration::from_millis(250), ToleranceSpec::count(0))?,
        )
        .await?;
    let service = engine
        .workloop_service()
        .ok_or("workloop service must be configured")?;

    // Window 1 elapses; the sweep records the fire through the loop's
    // Recorder. The wake tries to respawn the generation, and with no
    // deployed package that respawn faults — reported, never silent, and the
    // recorded fire stands.
    tokio::time::sleep(Duration::from_millis(350)).await;
    let report = service.tick().await?;
    assert_eq!(report.fired.len(), 1, "window 1 must fire: {report:?}");

    // Window 2 elapses; iteration 1 never closed (nothing ran it): the
    // dead-man counts the miss engine-side and tolerance zero alarms with
    // cause window-missed, on the one alarm path, in recorded history.
    tokio::time::sleep(Duration::from_millis(350)).await;
    let report = service.tick().await?;
    assert_eq!(report.fired.len(), 1, "window 2 must fire: {report:?}");
    assert_eq!(report.alarms.len(), 1, "the miss must alarm: {report:?}");

    let history = store.read_history(&loop_id).await?;
    let fires: Vec<u64> = history
        .iter()
        .filter_map(|event| match event {
            Event::CadenceFired { window_seq, .. } => Some(*window_seq),
            _ => None,
        })
        .collect();
    assert_eq!(fires, vec![1, 2], "both fires are recorded events");
    let alarm = history
        .iter()
        .find_map(|event| match event {
            Event::InvariantUnconfirmed {
                invariant,
                cause,
                window_seq,
                consecutive_unconfirmed,
                ..
            } => Some((
                invariant.clone(),
                *cause,
                *window_seq,
                *consecutive_unconfirmed,
            )),
            _ => None,
        })
        .ok_or("the alarm must be a recorded event")?;
    assert_eq!(alarm.0, "serving");
    assert_eq!(alarm.1, AlarmCause::WindowMissed);
    assert_eq!(alarm.2, Some(2));
    assert_eq!(alarm.3, 1);

    engine.shutdown()?;
    Ok(())
}

#[tokio::test(flavor = "multi_thread")]
async fn the_iteration_boundary_is_one_atomic_batch_with_a_deferred_successor() -> TestResult {
    let store = Arc::new(InMemoryStore::default());
    let engine = build_engine(&store).await?;
    let loop_id = WorkflowId::new_v4();
    let first_run = seed_started_workflow(&store, &loop_id).await?;
    engine
        .register_workloop(
            &loop_id,
            String::from("default"),
            cadence_spec(Duration::from_secs(3600), ToleranceSpec::count(3))?,
        )
        .await?;

    let carry = Payload::new(ContentType::Json, b"{\"seen\":[\"t1\"]}".to_vec());
    let state = Payload::new(ContentType::Json, b"{\"connected\":2}".to_vec());
    let next_run = engine
        .close_workloop_iteration(
            &loop_id,
            WorkloopIterationClose {
                routes: vec![String::from("sweep"), String::from("start")],
                carry: carry.clone(),
                invariant_states: vec![(String::from("serving"), state.clone())],
            },
        )
        .await?;
    assert_ne!(next_run, first_run);

    let history = store.read_history(&loop_id).await?;
    assert_eq!(
        event_kinds(&history),
        vec![
            "WorkflowStarted",
            "SearchAttributesUpdated",
            "IterationClosed",
            "WorkflowContinuedAsNew",
            "WorkflowStarted",
        ],
        "the boundary is IterationClosed + terminal + successor, in order"
    );
    // The successor generation carries the carry as its input and chains to
    // its predecessor; the loop projects Running with NO resident process.
    let successor = history
        .iter()
        .rev()
        .find_map(|event| match event {
            Event::WorkflowStarted {
                input,
                run_id,
                parent_run_id,
                ..
            } => Some((input.clone(), run_id.clone(), parent_run_id.clone())),
            _ => None,
        })
        .ok_or("successor start must exist")?;
    assert_eq!(successor.0, carry);
    assert_eq!(successor.1, next_run);
    assert_eq!(successor.2, Some(first_run));
    assert_eq!(
        aion_core::status_from_events(&history),
        WorkflowStatus::Running
    );

    // The iteration's samples landed: the confirming route confirmed the
    // invariant (no alarm), and the current-state record is queryable (R7).
    let samples = history
        .iter()
        .find_map(|event| match event {
            Event::IterationClosed { health_samples, .. } => Some(health_samples.clone()),
            _ => None,
        })
        .ok_or("IterationClosed must carry samples")?;
    assert_eq!(samples.len(), 1);
    assert_eq!(samples[0].status, aion_core::HealthStatus::Confirmed);
    let current = store
        .current_invariant_record(&loop_id, "serving")
        .await?
        .ok_or("invariant current-state record must exist")?;
    assert_eq!(current.payload, state);
    assert_eq!(current.record_type, "ServeState");

    engine.shutdown()?;
    Ok(())
}

/// #214, the query half: between iterations a workloop's successor is
/// durably started and deferred — no registry handle until its wake. A query
/// aimed at it is answered `NotRunning` (the run exists, nothing is resident
/// to answer), never `WorkflowNotFound`: the operator can see the loop in the
/// listing, and "not found" for a listed loop was the reported lie. A run no
/// history names stays `WorkflowNotFound`, pinned beside it so the fix cannot
/// widen into "everything is not-running".
#[tokio::test(flavor = "multi_thread")]
async fn a_query_at_the_deferred_successor_is_not_running_never_not_found() -> TestResult {
    let store = Arc::new(InMemoryStore::default());
    let engine = build_engine(&store).await?;
    let loop_id = WorkflowId::new_v4();
    let first_run = seed_started_workflow(&store, &loop_id).await?;
    engine
        .register_workloop(
            &loop_id,
            String::from("default"),
            cadence_spec(Duration::from_secs(3600), ToleranceSpec::count(3))?,
        )
        .await?;
    let next_run = engine
        .close_workloop_iteration(
            &loop_id,
            WorkloopIterationClose {
                routes: vec![String::from("sweep")],
                carry: Payload::new(ContentType::Json, b"{}".to_vec()),
                invariant_states: Vec::new(),
            },
        )
        .await?;
    assert_ne!(next_run, first_run);

    let arguments = Payload::new(ContentType::Json, b"null".to_vec());
    let deferred = engine
        .query(&loop_id, &next_run, "state", arguments.clone())
        .await;
    assert!(
        matches!(
            deferred,
            Err(aion::EngineError::Query(aion::query::QueryError::NotRunning(ref id))) if *id == loop_id
        ),
        "the deferred successor exists and is not running: {deferred:?}"
    );

    let closed = engine
        .query(&loop_id, &first_run, "state", arguments.clone())
        .await;
    assert!(
        matches!(
            closed,
            Err(aion::EngineError::Query(
                aion::query::QueryError::NotRunning(_)
            ))
        ),
        "the closed predecessor is terminal, so not running: {closed:?}"
    );

    let unknown_run = RunId::new_v4();
    let unknown = engine
        .query(&loop_id, &unknown_run, "state", arguments)
        .await;
    assert!(
        matches!(unknown, Err(aion::EngineError::WorkflowNotFound { .. })),
        "a run no history names is still not found: {unknown:?}"
    );

    engine.shutdown()?;
    Ok(())
}

/// Tom's one-row order (#214 collapsed): a workloop is ONE listing row for
/// its whole life. After N closed windows the listing holds exactly one row
/// for the loop — its current generation, Running, with no end — and no row
/// anywhere in the list reads `ContinuedAsNew`. The hand-over is the loop's
/// heartbeat, never its obituary; the generation chain stays in history.
#[tokio::test(flavor = "multi_thread")]
async fn a_workloop_lists_as_one_row_across_generations() -> TestResult {
    let store = Arc::new(InMemoryStore::default());
    let engine = build_engine(&store).await?;
    let loop_id = WorkflowId::new_v4();
    let first_run = seed_started_workflow(&store, &loop_id).await?;
    engine
        .register_workloop(
            &loop_id,
            String::from("default"),
            cadence_spec(Duration::from_secs(3600), ToleranceSpec::count(3))?,
        )
        .await?;
    let close = || WorkloopIterationClose {
        routes: vec![String::from("sweep")],
        carry: Payload::new(ContentType::Json, b"{}".to_vec()),
        invariant_states: Vec::new(),
    };
    let second_run = engine.close_workloop_iteration(&loop_id, close()).await?;
    let third_run = engine.close_workloop_iteration(&loop_id, close()).await?;
    assert_ne!(second_run, first_run);
    assert_ne!(third_run, second_run);

    let rows = list_default_namespace(&engine).await?;
    let loop_rows: Vec<&aion_core::WorkflowSummary> = rows
        .iter()
        .filter(|row| row.workflow_id == loop_id)
        .collect();
    assert_eq!(
        loop_rows.len(),
        1,
        "one row for the loop, never one per window: {loop_rows:?}"
    );
    let row = loop_rows[0];
    assert_eq!(row.status, WorkflowStatus::Running, "{row:?}");
    assert_eq!(row.run_id, third_run, "the row is the current generation");
    assert_eq!(row.ended_at, None, "a live loop has no end: {row:?}");
    assert!(
        rows.iter()
            .all(|row| row.status != WorkflowStatus::ContinuedAsNew),
        "no listing row ever reads ContinuedAsNew: {rows:?}"
    );

    engine.shutdown()?;
    Ok(())
}

#[tokio::test(flavor = "multi_thread")]
async fn retirement_is_recorded_with_its_terminal_and_leaves_records_standing() -> TestResult {
    let store = Arc::new(InMemoryStore::default());
    let engine = build_engine(&store).await?;
    let loop_id = WorkflowId::new_v4();
    seed_started_workflow(&store, &loop_id).await?;
    engine
        .register_workloop(
            &loop_id,
            String::from("default"),
            cadence_spec(Duration::from_secs(3600), ToleranceSpec::count(3))?,
        )
        .await?;
    // Give the loop a current-state record that must OUTLIVE retirement.
    engine
        .close_workloop_iteration(
            &loop_id,
            WorkloopIterationClose {
                routes: vec![String::from("sweep"), String::from("start")],
                carry: Payload::new(ContentType::Json, b"{}".to_vec()),
                invariant_states: vec![(
                    String::from("serving"),
                    Payload::new(ContentType::Json, b"{\"connected\":1}".to_vec()),
                )],
            },
        )
        .await?;

    // This fixture's loop is a synthesized history with no compiled module,
    // so it declares no `retire` block — the bodyless verb is the one that
    // matches the declaration. The body-invoking verb is covered against a
    // real compiled retire entry in `workloop_retire_e2e.rs`.
    engine
        .retire_workloop_without_body(
            &loop_id,
            String::from("queue decommissioned"),
            Payload::new(ContentType::Json, b"{\"drained\":true}".to_vec()),
        )
        .await?;

    let history = store.read_history(&loop_id).await?;
    let tail: Vec<&'static str> = event_kinds(&history)
        .into_iter()
        .rev()
        .take(2)
        .collect::<Vec<_>>()
        .into_iter()
        .rev()
        .collect();
    assert_eq!(
        tail,
        vec!["LoopRetired", "WorkflowCompleted"],
        "retirement is the declared marker plus its terminal, atomically"
    );
    assert_eq!(
        aion_core::status_from_events(&history),
        WorkflowStatus::Completed,
        "a retired loop reads as an intentional stop, never an outage"
    );
    // A genuine terminal on the last generation is the row's own: the one
    // row stays, Completed, with its end — the collapse hides hand-overs,
    // never endings.
    let rows = list_default_namespace(&engine).await?;
    let retired_rows: Vec<&aion_core::WorkflowSummary> = rows
        .iter()
        .filter(|row| row.workflow_id == loop_id)
        .collect();
    assert_eq!(retired_rows.len(), 1, "one row, retired: {retired_rows:?}");
    assert_eq!(retired_rows[0].status, WorkflowStatus::Completed);
    assert!(
        retired_rows[0].ended_at.is_some(),
        "a retired loop's row carries its end: {retired_rows:?}"
    );
    // Deregistered from the sweep set; the current-state record survives
    // indefinitely (R8.1).
    assert!(store.get_workloop(&loop_id).await?.is_none());
    assert!(
        store
            .current_invariant_record(&loop_id, "serving")
            .await?
            .is_some()
    );
    // Refusal pair: retiring an already-terminal loop refuses.
    let again = engine
        .retire_workloop_without_body(
            &loop_id,
            String::from("twice"),
            Payload::new(ContentType::Json, b"{}".to_vec()),
        )
        .await;
    assert!(again.is_err(), "retiring a terminal loop must refuse");

    engine.shutdown()?;
    Ok(())
}

#[tokio::test(flavor = "multi_thread")]
async fn hatch_dedupe_returns_the_existing_workflow_as_a_no_op() -> TestResult {
    let store = Arc::new(InMemoryStore::default());
    let engine = build_engine(&store).await?;

    // The identity's workflow already exists (a prior hatch, here seeded):
    // hatching the same (namespace, type, key) is a recorded no-op returning
    // the existing id — never a second workflow, never an error.
    let existing = aion_core::hatch_workflow_id("default", "process_task", "task-42")?;
    seed_started_workflow(&store, &existing).await?;
    let outcome = engine
        .hatch_workflow(
            "default",
            "process_task",
            "task-42",
            Payload::new(ContentType::Json, b"{}".to_vec()),
            HashMap::new(),
        )
        .await?;
    assert_eq!(outcome, HatchOutcome::Existing(existing.clone()));
    let history = store.read_history(&existing).await?;
    assert_eq!(
        history.len(),
        1,
        "a duplicate hatch must append nothing to the existing workflow"
    );

    // A DIFFERENT key derives a different identity, and with no deployed
    // package the start refuses loudly — a first hatch is a real start, never
    // a silent no-op.
    let fresh = engine
        .hatch_workflow(
            "default",
            "process_task",
            "task-43",
            Payload::new(ContentType::Json, b"{}".to_vec()),
            HashMap::new(),
        )
        .await;
    assert!(fresh.is_err(), "a first hatch with no package must refuse");

    // Identity refusals are typed.
    let refused = engine
        .hatch_workflow(
            "default",
            "process_task",
            "",
            Payload::new(ContentType::Json, b"{}".to_vec()),
            HashMap::new(),
        )
        .await;
    assert!(refused.is_err(), "an empty hatch key must refuse");

    engine.shutdown()?;
    Ok(())
}

/// 🔴 A SHUT-DOWN ENGINE RELEASES ITS STORE.
///
/// The `close_iteration/3` NIF bridge holds the whole iteration-close
/// component set — the workloop store, the event store, the visibility store
/// and the registry — and it lives in the NIF state, which OUTLIVES the engine
/// that installed it. So an engine that shut down left its stores alive for
/// the life of the process. On the in-memory store that is invisible; on a
/// durable backend that takes a file lock it is fatal, and it was: the moment
/// the server wired the workloop service unconditionally,
/// `recovery_declared_body_e2e` hung INDEFINITELY on the successor server's
/// attempt to open the same data directory. A restart is exactly what that
/// test performs, and a server that cannot restart is a server that cannot be
/// upgraded.
///
/// Asserted as REFERENCE COUNTS on the store the engine was built from,
/// because that is the property — "nothing of this engine outlives it" —
/// rather than any one holder. The world it comes out differently in is the
/// one before the fix: `stop()` aborted the sweep task and left the bridge in
/// place, and the count stayed above the test's own single handle forever.
#[tokio::test(flavor = "multi_thread")]
async fn a_shutdown_engine_leaves_no_handle_on_the_store_it_was_built_from()
-> Result<(), Box<dyn std::error::Error>> {
    let store = Arc::new(InMemoryStore::default());
    // The control on the measurement itself: while the engine is UP it must
    // hold handles, or the assertion below would pass against an engine that
    // never took one and would prove nothing.
    let engine = build_engine(&store).await?;
    let while_running = Arc::strong_count(&store);
    assert!(
        while_running > 1,
        "a running engine must hold the store it was built from; count {while_running}"
    );

    engine.shutdown()?;
    drop(engine);
    // The sweep task is aborted, not awaited, so give the runtime a moment to
    // reap it before counting. A bounded wait, never an unbounded one: if the
    // handle is genuinely leaked, waiting longer will not free it.
    for _ in 0..100u32 {
        if Arc::strong_count(&store) == 1 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    assert_eq!(
        Arc::strong_count(&store),
        1,
        "a shut-down engine must release every handle on its store — on a durable \
         backend a retained handle keeps the data directory's file lock and the next \
         process waits on it forever"
    );
    Ok(())
}