aion-server 0.31.0

//! `run_server` durable-outbox bootstrap integration tests.
//!
//! The tests launch the public `aion_server::run::run` entrypoint in a child
//! copy of this integration-test executable. That keeps the production
//! bootstrap intact (config load, `ServerState::build`, outbox dispatcher gate,
//! HTTP/gRPC transports) without adding public test APIs or manually spawning an
//! `OutboxDispatcher`.
//!
//! The end-to-end boots a real server over the durable haematite backend and
//! cold-opens the same data directory for independent assertions.

#[path = "run_server_outbox_support/helpers.rs"]
mod helpers;
#[path = "run_server_outbox_support/worker.rs"]
mod worker;

use std::io::Write;
use std::path::PathBuf;
use std::process::ExitCode;
use std::time::{Duration, Instant};

/// Opens the runtime gate on the live-reconciliation test once the store
/// grants a concurrent observation seam (see that test's doc for the exact
/// capability). Absent, the test announces its skip and proves nothing.
const CONCURRENT_SEAM_ENV: &str = "AION_E2E_CONCURRENT_STORE_SEAM";

use aion_awl_package::compile_and_assemble_awl;
use aion_core::Event;
use aion_server::config::CliOverrides;
use aion_store::{OutboxRow, OutboxStatus, ReadableEventStore};
use aion_store_haematite::HaematiteStore;
use chrono::Utc;
use helpers::{
    FAN_OUT, NAMESPACE, TestError, assert_fan_out_settled, assert_task_set, count_completed,
    count_completed_for, count_kind, fetch_history_over_http, row_states, run_server_harness,
    run_server_harness_with_reconciliation, start_over_http, task_ordinal, test_error,
    unique_temp_dir, wait_for_history, wait_for_history_over_http, wait_for_rows, worker_result,
    write_package_archive,
};
use serde_json::json;
use worker::WorkerSession;

/// The outbox — not the live in-process path — carries a fan-out, and every
/// member is dispatched and completed exactly once.
///
/// Split across two boots for one reason: haematite holds an EXCLUSIVE writer
/// lock on the data directory for the life of the server process, so the durable
/// state can only be read once that process has exited. Every assertion the
/// single-boot version made survives; only the moment of looking moved. The
/// first boot drives the system to "staged, nothing dispatched" and is then
/// stopped so the bytes can say so; the second dispatches and completes.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn run_server_outbox_happy_path_fan_out_completes_once() -> Result<(), TestError> {
    let dir = unique_temp_dir("happy")?;
    let db_path = dir.path().join("aion.db");
    let package_path = write_package_archive(dir.path())?;

    // Boot 1: start the fan-out with NO worker in existence, anywhere.
    let (server, http, _grpc) = run_server_harness(dir.path(), &db_path, &package_path).await?;
    let (workflow_id, run_id) = start_over_http(http).await?;
    wait_for_history_over_http(http, &workflow_id, "fan-out scheduled", |events| {
        count_kind(events, |event| {
            matches!(event, Event::ActivityScheduled { .. })
        }) == FAN_OUT
    })
    .await?;
    // The drain is the barrier: when it returns, the process is gone and the
    // writer lock with it. No sleep is involved in that argument.
    server.stop_gracefully()?;

    // At the bytes, with nobody holding the directory: the rows were staged by
    // the outbox path before any worker could exist, and the dispatcher worked
    // on them while waiting for one.
    {
        let reader =
            HaematiteStore::open_or_create(db_path.clone(), haematite::NodeCacheBudget::Unlimited)
                .await?;
        let history = reader.read_history(&workflow_id).await?;
        assert_eq!(
            count_kind(&history, |event| matches!(
                event,
                Event::ActivityScheduled { .. }
            )),
            FAN_OUT,
            "every fan-out member must be scheduled durably"
        );
        assert_eq!(
            count_completed(&history),
            0,
            "no member may be completed before any worker has ever registered"
        );

        let states = row_states(&reader, &workflow_id, &[0, 1, 2, 3]).await?;
        // `record_fan_out_dispatch` stages these atomically with the scheduling
        // events; the live in-process path would never write them at all. That
        // they exist is the cutover proof.
        assert_eq!(
            states.len(),
            FAN_OUT,
            "every ordinal must have a staged row"
        );
        assert!(
            states
                .iter()
                .all(|state| state.status != OutboxStatus::Done),
            "no row may be Done — a row settles on DISPATCH, and no worker ever \
             registered to be dispatched to: {states:?}"
        );
        // The single-boot version also watched, live, for the dispatcher to CLAIM
        // one of these rows while no worker existed. That claim leaves no trace
        // an outside process can read after the fact: claiming does not consume
        // an attempt, and the drain re-arms a claimed row to `Pending` with the
        // attempt budget preserved. Waiting for it here would be a race against
        // the dispatcher's own poll with no observable barrier to close it, so
        // the assertion lives where the sweep can be DRIVEN rather than awaited:
        // `worker::outbox_dispatcher::tests::failed_dispatch_retries_with_backoff_and_bumps_attempt`
        // pins that the sweep claims a staged row and offers it to the dispatch
        // seam even when that seam cannot deliver — which is how the production
        // no-worker path reports itself, since `OutboxRowDispatch::dispatch`
        // returns no-worker as an ordinary dispatch error.
    }

    // Boot 2: a worker appears, and the staged work is dispatched and completed.
    let (server, http, grpc) = run_server_harness(dir.path(), &db_path, &package_path).await?;
    let mut worker = WorkerSession::connect(grpc).await?;
    let mut tasks = Vec::with_capacity(FAN_OUT);
    for _ in 0..FAN_OUT {
        tasks.push(worker.next_task().await?);
    }
    assert_task_set(&tasks, &[0, 1, 2, 3])?;
    for task in &tasks {
        let ordinal = task_ordinal(task)?;
        worker
            .complete(task, worker_result(ordinal).as_bytes())
            .await?;
    }
    wait_for_history_over_http(http, &workflow_id, "fan-out settled", |events| {
        count_completed(events) == FAN_OUT
            && count_kind(events, |event| {
                matches!(event, Event::WorkflowCompleted { .. })
            }) == 1
    })
    .await?;
    drop(worker);
    server.stop_gracefully()?;

    let reader =
        HaematiteStore::open_or_create(db_path, haematite::NodeCacheBudget::Unlimited).await?;
    let history = assert_fan_out_settled(&reader, &workflow_id).await?;
    assert_eq!(count_completed(&history), FAN_OUT);
    std::hint::black_box(run_id);
    Ok(())
}

/// A server that dies with dispatches in flight strands their outbox rows; the
/// next boot re-arms them, re-dispatches, and the fan-out still settles exactly
/// once despite a duplicate completion.
///
/// The stranded state is asserted at the BYTES, in the window between the kill
/// and the re-boot — the one moment nothing holds haematite's writer lock. That
/// window is also where this file proves a property no other test here states:
/// an outbox row settles `Done` on DISPATCH, not on completion. All four rows
/// are `Done` while ordinals 2 and 3 have no completion recorded at all.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn run_server_outbox_restart_rearms_stranded_rows() -> Result<(), TestError> {
    let dir = unique_temp_dir("restart")?;
    let db_path = dir.path().join("aion.db");
    let package_path = write_package_archive(dir.path())?;
    let (server1, http1, grpc1) = run_server_harness(dir.path(), &db_path, &package_path).await?;
    let mut worker1 = WorkerSession::connect(grpc1).await?;

    let (workflow_id, run_id) = start_over_http(http1).await?;
    let mut tasks = Vec::with_capacity(FAN_OUT);
    for _ in 0..FAN_OUT {
        tasks.push(worker1.next_task().await?);
    }
    tasks.sort_by_key(|task| task_ordinal(task).unwrap_or(u64::MAX));
    assert_task_set(&tasks, &[0, 1, 2, 3])?;

    complete_recorded_prefix(&worker1, &tasks).await?;
    wait_for_history_over_http(http1, &workflow_id, "ordinals 0 and 1 recorded", |events| {
        count_completed_for(events, 0) == 1 && count_completed_for(events, 1) == 1
    })
    .await?;
    drop(worker1);
    // An abrupt death with ordinals 2 and 3 in flight — the production way a
    // dispatch is stranded. `stop` waits for the process, so the lock is gone.
    server1.stop()?;

    // Between the stop and the re-boot, at the bytes.
    {
        let reader =
            HaematiteStore::open_or_create(db_path.clone(), haematite::NodeCacheBudget::Unlimited)
                .await?;
        let states = row_states(&reader, &workflow_id, &[0, 1, 2, 3]).await?;
        assert!(
            states
                .iter()
                .all(|state| state.status == OutboxStatus::Done),
            "every row was dispatched, so every row is Done — a row settles on \
             dispatch, not on completion: {states:?}"
        );
        let stranded = reader.read_history(&workflow_id).await?;
        assert_eq!(count_completed_for(&stranded, 0), 1);
        assert_eq!(count_completed_for(&stranded, 1), 1);
        assert_eq!(
            count_completed_for(&stranded, 2),
            0,
            "ordinal 2's dispatch died in flight and must have recorded nothing"
        );
        assert_eq!(
            count_completed_for(&stranded, 3),
            0,
            "ordinal 3's dispatch died in flight and must have recorded nothing"
        );
    }

    // The re-boot re-arms the two stranded rows and re-dispatches them. Their
    // arrival at a fresh worker IS the re-arm: rows 2 and 3 were `Done` at the
    // bytes a moment ago, so nothing could reach a worker again without one.
    // Reading the intermediate `Pending`/`Claimed` status directly would mean
    // stopping the server inside the window between re-arm and re-dispatch,
    // which the dispatcher's own poll can close at any instant.
    let (server2, http2, grpc2) = run_server_harness(dir.path(), &db_path, &package_path).await?;
    let mut worker2 = WorkerSession::connect(grpc2).await?;
    let revived = collect_revived_tasks(&mut worker2).await?;
    assert_task_set(&revived, &[2, 3])?;
    complete_with_duplicate_first(&worker2, &revived).await?;
    wait_for_history_over_http(http2, &workflow_id, "fan-out settled", |events| {
        count_completed(events) == FAN_OUT
            && count_kind(events, |event| {
                matches!(event, Event::WorkflowCompleted { .. })
            }) == 1
    })
    .await?;
    drop(worker2);
    server2.stop_gracefully()?;

    let reader =
        HaematiteStore::open_or_create(db_path, haematite::NodeCacheBudget::Unlimited).await?;
    let history = assert_fan_out_settled(&reader, &workflow_id).await?;
    assert_eq!(count_completed(&history), FAN_OUT);
    std::hint::black_box(run_id);
    Ok(())
}

/// LIVE reconciliation re-arms a stale claim exactly once — **RED, and it must
/// stay red until haematite can serve a concurrent reader.**
///
/// ## Why this one could not be relocated like its siblings
///
/// The other two tests in this file were rewritten to drive the system with the
/// server up and assert at the bytes once it has exited. That works because
/// their subjects — the outbox carrying a fan-out, and a restart re-arming a
/// stranded row — leave durable state that outlives the process.
///
/// This test's subject does not. A stale claim is a row left `Claimed` with an
/// old claim stamp, and the LIVE reconciler is the thing under test: it must
/// find that row and re-arm it *while its own server serves*. Two facts close
/// every door:
///
/// 1. **The fixture cannot be planted from outside.** `set_outbox_claimed_at`
///    needs a writer handle, and haematite grants exactly one per data
///    directory — held by the server process for its whole life. libSQL allowed
///    the second handle this test was written against; haematite refuses it,
///    and the refusal is correct (a second writer would corrupt the shard WALs).
/// 2. **Planting it BEFORE the boot makes the test vacuous.** Measured: with the
///    stale rows planted in the window between two servers, the test passes with
///    the reconciler's guard mutated to `false`, and passes again with
///    reconciliation left entirely unconfigured. The boot-time re-arm settles
///    those rows before the reconciler ever sweeps, so a "green" would be
///    attributable to the wrong mechanism — and would duplicate
///    `run_server_outbox_restart_rearms_stranded_rows`, which already owns it.
///
/// A stale claim cannot arise naturally on a live server either: with a worker
/// connected the dispatch settles the row `Done`; with none it fails and retries
/// to `Pending`. Only a death between claim and settle leaves `Claimed`, and
/// that death ends the server this test needs alive.
///
/// ## What would make it green
///
/// Either a haematite **concurrent reader/writer seam** — a second handle that
/// can perform ROUTED reads (aion co-locates an outbox row on its workflow's
/// shard, and `ReadOnlyDatabase` routes by the key itself, so today's observer
/// cannot read these rows at all, let alone write one) — or a server-side test
/// seam that can age a claim in place. Both are product decisions, not
/// something this lane may invent; tracked as aion#114 (the citation once read
/// "the aion#35 report" — a report attached to the libsql-retirement umbrella,
/// which closed 2026-08-18 and took the pointer with it).
///
/// Left in its ORIGINAL single-boot form deliberately: that form is the honest
/// statement of the contract. Runtime-gated (house convention) rather than
/// permanently red: a known-red on main would make every future battery's
/// extraction non-empty and cost the "empty = green" reading estate-wide. The
/// full assertion body is kept armed; the skip line names the missing
/// capability and the switch that opens the gate the day it ships.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn run_server_outbox_live_reconciliation_rearms_stranded_claims_once() -> Result<(), TestError>
{
    if std::env::var_os(CONCURRENT_SEAM_ENV).is_none() {
        let mut stderr = std::io::stderr();
        let _announced = writeln!(
            stderr,
            "SKIP (PROVES NOTHING): live reconciliation of a stale claim — not \
             established by this run. Needs a haematite concurrent seam with ROUTED \
             reads, or a server-side claim-aging seam (aion#114). No pinnable \
             surface exists today — the absent capability IS the absence — so this \
             announcement stands unpinned until #114 ships a surface. Set \
             {CONCURRENT_SEAM_ENV} once the capability exists."
        );
        return Ok(());
    }
    let dir = unique_temp_dir("live-reconcile")?;
    let db_path = dir.path().join("aion.db");
    let package_path = write_package_archive(dir.path())?;
    let (server, http, grpc) = run_server_harness_with_reconciliation(
        dir.path(),
        &db_path,
        &package_path,
        Some((50, 100)),
    )
    .await?;
    let reader =
        HaematiteStore::open_or_create(db_path.clone(), haematite::NodeCacheBudget::Unlimited)
            .await?;
    let mut worker = WorkerSession::connect(grpc).await?;

    let (workflow_id, run_id) = start_over_http(http).await?;
    let mut tasks = Vec::with_capacity(FAN_OUT);
    for _ in 0..FAN_OUT {
        tasks.push(worker.next_task().await?);
    }
    tasks.sort_by_key(|task| task_ordinal(task).unwrap_or(u64::MAX));
    assert_task_set(&tasks, &[0, 1, 2, 3])?;
    wait_for_rows(
        &reader,
        &workflow_id,
        &[0, 1, 2, 3],
        "initial rows done before simulated stale claim",
        |statuses| statuses.iter().all(|status| *status == OutboxStatus::Done),
    )
    .await?;

    complete_recorded_prefix(&worker, &tasks).await?;
    wait_for_history(
        &reader,
        &workflow_id,
        "ordinals 0 and 1 recorded before live reconciliation",
        |events| count_completed_for(events, 0) == 1 && count_completed_for(events, 1) == 1,
    )
    .await?;

    force_rows_to_stale_claimed(&db_path, &workflow_id, &[2, 3]).await?;

    let revived = collect_revived_tasks(&mut worker).await?;
    complete_with_duplicate_first(&worker, &revived).await?;
    complete_original_late(&worker, &tasks[2..]).await?;

    let history = assert_fan_out_settled(&reader, &workflow_id).await?;
    assert_eq!(count_completed(&history), FAN_OUT);
    assert_eq!(
        count_kind(&history, |event| matches!(
            event,
            Event::WorkflowCompleted { .. }
        )),
        1
    );
    std::hint::black_box(run_id);
    server.stop()?;
    Ok(())
}

async fn complete_recorded_prefix(
    worker: &WorkerSession,
    tasks: &[aion_proto::generated::ActivityTask],
) -> Result<(), TestError> {
    for task in tasks.iter().take(2) {
        let ordinal = task_ordinal(task)?;
        worker
            .complete(task, worker_result(ordinal).as_bytes())
            .await?;
    }
    Ok(())
}

async fn collect_revived_tasks(
    worker: &mut WorkerSession,
) -> Result<Vec<aion_proto::generated::ActivityTask>, TestError> {
    let mut revived = Vec::with_capacity(2);
    for _ in 0..2 {
        revived.push(worker.next_task().await?);
    }
    revived.sort_by_key(|task| task_ordinal(task).unwrap_or(u64::MAX));
    assert_task_set(&revived, &[2, 3])?;
    Ok(revived)
}

async fn complete_with_duplicate_first(
    worker: &WorkerSession,
    revived: &[aion_proto::generated::ActivityTask],
) -> Result<(), TestError> {
    let first = revived
        .first()
        .ok_or_else(|| test_error("missing first revived task"))?;
    let first_ordinal = task_ordinal(first)?;
    worker
        .complete(first, worker_result(first_ordinal).as_bytes())
        .await?;
    worker
        .complete(first, worker_result(first_ordinal).as_bytes())
        .await?;
    let second = revived
        .get(1)
        .ok_or_else(|| test_error("missing second revived task"))?;
    let second_ordinal = task_ordinal(second)?;
    worker
        .complete(second, worker_result(second_ordinal).as_bytes())
        .await?;
    Ok(())
}

async fn complete_original_late(
    worker: &WorkerSession,
    tasks: &[aion_proto::generated::ActivityTask],
) -> Result<(), TestError> {
    for task in tasks {
        let ordinal = task_ordinal(task)?;
        worker
            .complete(task, worker_result(ordinal).as_bytes())
            .await?;
    }
    Ok(())
}

/// Back-date `ordinals`' claim stamps so the LIVE reconciler sees them as stale.
///
/// Opens its own writer on a directory the running server already holds, which
/// is exactly the capability haematite does not grant — see the doc on
/// `run_server_outbox_live_reconciliation_rearms_stranded_claims_once`.
async fn force_rows_to_stale_claimed(
    data_dir: &std::path::Path,
    workflow_id: &aion_core::WorkflowId,
    ordinals: &[u64],
) -> Result<(), TestError> {
    let store = HaematiteStore::open(data_dir, haematite::NodeCacheBudget::Unlimited, |_, _| {})?;
    let claimed_at = Utc::now() - chrono::Duration::seconds(60);
    for ordinal in ordinals {
        let dispatch_key = OutboxRow::dispatch_key_for(workflow_id, *ordinal);
        store
            .set_outbox_claimed_at(&dispatch_key, claimed_at)
            .await?;
    }
    Ok(())
}

/// The child's Aion home is the harness directory, and nothing else.
///
/// Every `run_server` child is a REAL server: it claims a pid file at birth and
/// arms a death note under its Aion home. Before 2026-08-28 the harness spawned
/// it with `AION_HOME` stripped, so that home was `$HOME/.aion` — the operator's
/// live server's home on the machine that runs this suite. Each child wrote its
/// death note into the operator's logs and, finding the operator's pid file,
/// booted UNCLAIMED. This pin reads the child's own records back from the
/// harness directory: the pid file names THIS child, the death note is armed
/// there, and after a graceful stop the note records the disarm. A child that
/// went to the operator's home leaves nothing here and the pin is red.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn run_server_child_claims_the_harness_home_not_the_operators() -> Result<(), TestError> {
    let dir = unique_temp_dir("home")?;
    let db_path = dir.path().join("aion.db");
    let package_path = write_package_archive(dir.path())?;
    let (server, _http, _grpc) = run_server_harness(dir.path(), &db_path, &package_path).await?;

    let home = server.home().to_path_buf();
    assert_eq!(
        home,
        dir.path(),
        "the harness directory is the child's home"
    );
    let record = aion_server::control::pid_file::read(&home)?.ok_or_else(|| {
        test_error(format!(
            "no pid record under the harness home {} — the child claimed some other home",
            aion_server::control::pid_file::pid_file_path(&home).display()
        ))
    })?;
    assert_eq!(
        record.pid,
        server.pid(),
        "the pid record under the harness home must name this child"
    );
    let note_path = aion_server::death_note::note_path(&home);
    let armed = std::fs::read_to_string(&note_path).map_err(|error| {
        test_error(format!(
            "no death note under the harness home at {}: {error}",
            note_path.display()
        ))
    })?;
    assert!(
        armed.contains(" ARMED "),
        "the death note under the harness home must record this child's arming: {armed}"
    );

    server.stop_gracefully()?;

    let disarmed = std::fs::read_to_string(&note_path)?;
    assert!(
        disarmed.contains(" DISARMED "),
        "after a graceful stop the harness-home death note must record the disarm: {disarmed}"
    );
    assert!(
        aion_server::control::pid_file::read(&home)?.is_none(),
        "a cleanly stopped child must release the pid file it claimed under the harness home"
    );
    Ok(())
}

// ─── aion#193: a fork over DECLARED bodies, no worker ────────────────────────
//
// Before `DeclaredBodyOutboxDispatch` existed, every fork member whose action
// carried a declared body parked as NO_LIVE_POLLERS on this very path and
// dead-lettered after the retry budget; a one-element fork failed identically
// (field case: examples/json-between-steps/json_fanout.awl). The plain statement
// before the fork is the control arm: it always executed at the server.

const FANOUT_DECLARED_WORKFLOW_TYPE: &str = "fanout_declared";
/// The run must reach a terminal state within this bound; a regression
/// dead-letters well inside it (`max_attempts` 5 at `backoff_max` 1 s above).
const FANOUT_DECLARED_DEADLINE: Duration = Duration::from_secs(120);

/// One plain statement (the control arm) then a fork of one call per element,
/// every action a declared `run` body on a queue NO worker serves.
const FANOUT_DECLARED_DOCUMENT: &str = r#"//! aion#193 pin: a fork over declared bodies, no worker.
workflow fanout_declared
  input shas: [String]
  outcome sized: type [RunOutcome], route success

type RunOutcome { exit_code: Int, stdout: String, stderr: String }

worker json_box
  action tally(shas: [String]) -> RunOutcome
    run "printf %s counted"
  action size_of(sha: String) -> RunOutcome
    run "printf sized:%s {{sha}}"

step gather
  tally(shas: shas) -> counted

step measure
  fork sha in shas
    size_of(sha: sha)
  join -> sizes
  route sized(sizes)
"#;

async fn start_fanout(
    address: std::net::SocketAddr,
    shas: &[&str],
) -> Result<(aion_core::WorkflowId, aion_core::RunId), TestError> {
    let client = reqwest::Client::new();
    let response = client
        .post(format!("http://{address}/workflows/start"))
        .header("content-type", "application/json")
        .header("x-aion-subject", "ci")
        .header("x-aion-namespaces", NAMESPACE)
        .json(&json!({
            "namespace": NAMESPACE,
            "workflow_type": FANOUT_DECLARED_WORKFLOW_TYPE,
            "input": { "shas": shas },
        }))
        .send()
        .await?;
    let status = response.status();
    let bytes = response.bytes().await?;
    if !status.is_success() {
        return Err(test_error(format!(
            "start must succeed, got {status}: {}",
            String::from_utf8_lossy(&bytes)
        )));
    }
    let body: serde_json::Value = serde_json::from_slice(&bytes)?;
    let workflow_id = body["workflow_id"]
        .as_str()
        .ok_or_else(|| test_error("start reply has no workflow id"))?
        .parse::<uuid::Uuid>()?;
    let run_id = body["run_id"]
        .as_str()
        .ok_or_else(|| test_error("start reply has no run id"))?
        .parse::<uuid::Uuid>()?;
    Ok((
        aion_core::WorkflowId::new(workflow_id),
        aion_core::RunId::new(run_id),
    ))
}

/// The pin. Two elements, so the join has an order to keep; the outbox
/// commissioned by the server's own boot; no worker anywhere. Green only when
/// every fork member's declared body ran at the server and completed through
/// the fan-out route.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_fork_over_declared_bodies_completes_with_no_worker_on_the_outbox_path()
-> Result<(), TestError> {
    let dir = unique_temp_dir("declared-fanout")?;
    let db_path = dir.path().join("aion.db");
    let prepared =
        compile_and_assemble_awl(FANOUT_DECLARED_DOCUMENT, dir.path(), "fanout_declared.awl")?;
    let package_path = dir.path().join("fanout_declared.aion");
    std::fs::write(&package_path, prepared.archive)?;
    let (server, http, _grpc) = run_server_harness(dir.path(), &db_path, &package_path).await?;

    let (workflow_id, _run_id) = start_fanout(http, &["aaa111", "bbb222"]).await?;
    let deadline = Instant::now() + FANOUT_DECLARED_DEADLINE;
    let history = loop {
        let history = fetch_history_over_http(http, &workflow_id).await?;
        if history.iter().any(|event| {
            matches!(
                event,
                Event::WorkflowCompleted { .. } | Event::WorkflowFailed { .. }
            )
        }) {
            break history;
        }
        if Instant::now() > deadline {
            return Err(test_error(format!(
                "the fork over declared bodies reached no terminal state in {FANOUT_DECLARED_DEADLINE:?} — \
                 the members parked (NO_LIVE_POLLERS) or are still retrying: {history:#?}"
            )));
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    };
    let result = history
        .iter()
        .find_map(|event| match event {
            Event::WorkflowCompleted { result, .. } => Some(result.clone()),
            _ => None,
        })
        .ok_or_else(|| {
            test_error(format!(
                "the run ended without completing — the fork members dead-lettered: {history:#?}"
            ))
        })?;
    // The recorded result is the AWL outcome envelope: `{outcome, payload}`.
    let decoded: serde_json::Value = serde_json::from_slice(result.bytes())?;
    assert_eq!(
        decoded["outcome"], "sized",
        "the run must route its declared outcome: {decoded}"
    );
    let sizes = decoded["payload"]
        .as_array()
        .ok_or_else(|| test_error(format!("the outcome payload is not a list: {decoded}")))?;
    let stdouts: Vec<&str> = sizes
        .iter()
        .map(|size| size["stdout"].as_str().unwrap_or("<not a string>"))
        .collect();
    assert_eq!(
        stdouts,
        vec!["sized:aaa111", "sized:bbb222"],
        "every member's body ran with its own element bound, joined in input order"
    );
    assert!(sizes.iter().all(|size| size["exit_code"] == 0));
    assert_eq!(
        count_completed(&history),
        3,
        "the control statement and both fork members completed: {history:#?}"
    );
    server.stop_gracefully()?;

    // The members WERE outbox rows, and they are settled: the fork's two
    // ordinals have rows and every row is Done.
    let reader =
        HaematiteStore::open_or_create(db_path, haematite::NodeCacheBudget::Unlimited).await?;
    let states = row_states(&reader, &workflow_id, &[1, 2]).await?;
    assert!(
        states
            .iter()
            .all(|state| state.status == OutboxStatus::Done),
        "every fork member's row must be Done after its body completed: {states:?}"
    );
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn run_server_child_process() -> Result<(), TestError> {
    if std::env::var_os("AION_RUN_SERVER_CHILD").is_none() {
        return Ok(());
    }
    let config_path = std::env::var_os("AION_RUN_SERVER_CONFIG")
        .map(PathBuf::from)
        .ok_or_else(|| test_error("AION_RUN_SERVER_CONFIG is required"))?;
    let code = aion_server::run::run(CliOverrides {
        config_path: Some(config_path),
        ..CliOverrides::default()
    })
    .await;
    if code == ExitCode::SUCCESS {
        Ok(())
    } else {
        Err(test_error(format!("run_server exited with {code:?}")))
    }
}