aion-server 0.29.0

//! The rules, each against a real `ServerState`, a real durable store, and a
//! real supervised process.
//!
//! The parts have their own pins; these are about the COMPOSITION, because
//! every rule the brief states lives in the orchestration and a defect in it
//! looks exactly like a correct part.
//!
//! Supervision is commissioned over `/bin/sh` so a "worker" here is a plain
//! sleep rather than a re-execution of the test binary — the same seam the
//! managed-worker HTTP tests use.

use std::collections::BTreeSet;
use std::num::NonZeroU32;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use aion_package::AwlSource;
use aion_store::{
    DesiredState, InMemoryStore, NewWorkerDeployment, WorkerArtifactRef, WorkerDeployment,
    WorkerDeploymentStore,
};

use crate::assistant::EMBEDDED_ASSISTANT_DOCUMENT;
use crate::config::{NamespaceMode, OutboxConfig, OutboxTransport};
use crate::worker::supervisor::{ManagedExecutable, SupervisionPolicy};
use crate::{NamespaceResolver, ServerState, StaticScheduleNamespaces, StaticWorkflowNamespaces};

use super::provision;
use crate::worker::auto_provision::{AutoWorkerDecision, AutoWorkerOutcome};

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

/// The document every test deploys: the SHIPPED assistant, which really does
/// carry a `harness` section on queue `assistant`. A hand-written fixture would
/// be a second document free to drift from the one the product deploys.
fn document() -> AwlSource {
    AwlSource::new(
        "assistant.awl",
        EMBEDDED_ASSISTANT_DOCUMENT,
        Vec::<(String, Vec<u8>)>::new(),
    )
}

/// The same document with its harness section REMOVED — the negative control,
/// cut from the real thing rather than written beside it.
fn document_without_harness() -> Result<AwlSource, Box<dyn std::error::Error>> {
    let start = EMBEDDED_ASSISTANT_DOCUMENT
        .find("  harness\n")
        .ok_or("the shipped assistant must carry a harness section")?;
    let end = EMBEDDED_ASSISTANT_DOCUMENT[start..]
        .find("\n  action ")
        .map(|offset| start.saturating_add(offset).saturating_add(1))
        .ok_or("the harness section must be followed by an action")?;
    let mut stripped = String::from(&EMBEDDED_ASSISTANT_DOCUMENT[..start]);
    stripped.push_str(&EMBEDDED_ASSISTANT_DOCUMENT[end..]);
    Ok(AwlSource::new(
        "assistant.awl",
        stripped,
        Vec::<(String, Vec<u8>)>::new(),
    ))
}

/// A document whose harness section differs — a harness-only edit, which is
/// exactly the change the package content hash cannot see.
fn edited_document() -> AwlSource {
    AwlSource::new(
        "assistant.awl",
        EMBEDDED_ASSISTANT_DOCUMENT.replace("    concurrency 4\n", "    concurrency 6\n"),
        Vec::<(String, Vec<u8>)>::new(),
    )
}

/// 🔴 THE NEGATIVE CONTROL, at the composition. A document with no `harness`
/// section mints nothing, reports nothing, and leaves the durable store empty.
#[tokio::test]
async fn a_document_with_no_harness_section_writes_no_record() -> TestResult {
    let fixture = Fixture::new(true)?;
    let outcomes = fixture.provision(&document_without_harness()?).await;
    assert!(outcomes.is_empty(), "{outcomes:?}");
    assert!(fixture.names().await?.is_empty());
    Ok(())
}

/// The control for the control, and rule 4's first half: the SAME document with
/// its section intact mints, starts, and says so — and only then may the empty
/// answer above be read as the section's absence.
#[tokio::test]
async fn a_declaring_document_mints_starts_and_claims_only_what_ran() -> TestResult {
    let fixture = Fixture::new(true)?;
    let outcomes = fixture.provision(&document()).await;
    let first = outcomes.first().ok_or("one queue declares a harness")?;

    assert_eq!(outcomes.len(), 1);
    assert_eq!(first.task_queue, "assistant");
    assert_eq!(first.decision, AutoWorkerDecision::Minted);
    assert_eq!(first.deployment.as_deref(), Some("auto/assistant"));
    assert!(first.decision.claims_running());
    assert!(first.detail.contains("started it"), "{}", first.detail);

    let record = fixture.record("auto/assistant").await?;
    assert_eq!(record.task_queue, "assistant");
    assert_eq!(record.desired, DesiredState::Running);
    let pid = fixture.running_pid("auto/assistant").await?;
    assert!(pid > 0);
    fixture.teardown().await;
    Ok(())
}

/// A document whose harness names paths this host does not have — the edit is
/// a `binary` line pointing into another machine's filesystem, the exact shape
/// the 2026-08-26 cross-host fleet deploy stood up as a dispatch-burning
/// worker on the server's own Mac.
fn foreign_host_document() -> AwlSource {
    AwlSource::new(
        "assistant.awl",
        EMBEDDED_ASSISTANT_DOCUMENT.replace(
            "    kind norn
",
            "    kind norn
    binary \"/nonexistent-host-root/.bun/bin/bunx\"
",
        ),
        Vec::<(String, Vec<u8>)>::new(),
    )
}

/// 🔴 A HARNESS THIS HOST CANNOT LAUNCH IS REFUSED BY NAME, NOT MINTED. The
/// refusal carries the missing path, the deploy itself lands, and no record is
/// written — a minted record would replay a spawn that can only fail, burning
/// an attempt on every dispatch the queue rotates to it.
#[tokio::test]
async fn a_harness_naming_paths_this_host_lacks_is_refused_not_minted() -> TestResult {
    let fixture = Fixture::new(true)?;
    let outcomes = fixture.provision(&foreign_host_document()).await;
    let first = outcomes.first().ok_or("one queue declares a harness")?;

    assert_eq!(outcomes.len(), 1);
    assert_eq!(first.decision, AutoWorkerDecision::UnrunnableHarness);
    assert!(first.decision.is_refusal());
    assert!(
        first
            .detail
            .contains("/nonexistent-host-root/.bun/bin/bunx"),
        "the refusal must name the missing path: {}",
        first.detail
    );
    assert_eq!(first.deployment, None);
    assert!(
        fixture.names().await?.is_empty(),
        "no record may be written for an unrunnable harness"
    );
    fixture.teardown().await;
    Ok(())
}

/// 🔴 IT NEVER CLAIMS A WORKER IT DID NOT START. The record is written durably
/// on an UNCOMMISSIONED server and the outcome says so — the same server state
/// every installation whose config predates `[worker_supervision]` boots into.
#[tokio::test]
async fn an_uncommissioned_server_records_the_worker_and_refuses_to_claim_it() -> TestResult {
    let fixture = Fixture::new(false)?;
    let outcomes = fixture.provision(&document()).await;
    let first = outcomes.first().ok_or("one queue declares a harness")?;

    assert_eq!(first.decision, AutoWorkerDecision::RecordedNotRunning);
    assert!(
        !first.decision.claims_running(),
        "an unstarted worker must never be reported as started"
    );
    assert!(first.decision.is_refusal());
    assert!(!first.detail.contains("started it"), "{}", first.detail);
    // The record IS durable: the next boot's reconcile is what honours it.
    assert_eq!(
        fixture.record("auto/assistant").await?.desired,
        DesiredState::Running
    );
    Ok(())
}

/// 🔴 AN OPERATOR'S RECORD WINS. Nothing is minted, nothing is replaced, and
/// the skip names the record that won.
#[tokio::test]
async fn an_operators_record_for_the_queue_is_never_touched_or_duplicated() -> TestResult {
    let fixture = Fixture::new(true)?;
    fixture
        .put_operator_record("my-assistant", "assistant", Some("default"), None)
        .await?;

    let outcomes = fixture.provision(&document()).await;
    let first = outcomes.first().ok_or("one queue declares a harness")?;

    assert_eq!(first.decision, AutoWorkerDecision::OperatorRecord);
    assert_eq!(first.deployment.as_deref(), Some("my-assistant"));
    assert_eq!(fixture.names().await?, vec!["my-assistant".to_owned()]);
    Ok(())
}

/// An operator's record that could NOT serve this queue here — pinned to
/// another node, or bound to namespaces this server's default is not in — is
/// not a winner: it serves the queue somewhere else, and treating it as one
/// would leave this box unserved while reporting it served.
#[tokio::test]
async fn a_record_that_cannot_serve_this_node_or_namespace_does_not_win() -> TestResult {
    for (name, namespace, node) in [
        ("elsewhere-node", Some("default"), Some("another-box")),
        ("elsewhere-namespace", Some("tenant-a"), None),
    ] {
        let fixture = Fixture::new(true)?;
        fixture
            .put_operator_record(name, "assistant", namespace, node)
            .await?;
        let outcomes = fixture.provision(&document()).await;
        let first = outcomes.first().ok_or("one queue declares a harness")?;
        assert_eq!(
            first.decision,
            AutoWorkerDecision::Minted,
            "`{name}` cannot serve this queue here and must not win the skip: {}",
            first.detail
        );
        fixture.teardown().await;
    }
    Ok(())
}

/// 🔴 A DARK OUTBOX MINTS NOTHING. The refusal names the section, and the
/// durable store is untouched.
#[tokio::test]
async fn a_dark_outbox_refuses_loudly_and_writes_no_record() -> TestResult {
    let fixture = Fixture::dark()?;
    let outcomes = fixture.provision(&document()).await;
    let first = outcomes.first().ok_or("one queue declares a harness")?;

    assert_eq!(first.decision, AutoWorkerDecision::DarkOutbox);
    assert!(first.decision.is_refusal());
    assert_eq!(first.deployment, None);
    assert!(first.detail.contains("[outbox]"), "{}", first.detail);
    assert!(fixture.names().await?.is_empty(), "nothing may be minted");
    Ok(())
}

/// 🔴 RULE 4, BOTH HALVES. Identical bytes leave the running worker alone;
/// a HARNESS-ONLY edit — which produces the same package content hash, and so
/// is invisible to any package-keyed scheme — re-mints and restarts it.
#[tokio::test]
async fn an_identical_redeploy_is_unchanged_and_an_edited_one_re_mints() -> TestResult {
    let fixture = Fixture::new(true)?;
    drop(fixture.provision(&document()).await);
    let before = fixture.running_pid("auto/assistant").await?;
    let argv_before = fixture.argv("auto/assistant").await?;

    let again = fixture.provision(&document()).await;
    let unchanged = again.first().ok_or("one queue declares a harness")?;
    assert_eq!(unchanged.decision, AutoWorkerDecision::Unchanged);
    assert_eq!(
        fixture.running_pid("auto/assistant").await?,
        before,
        "an identical redeploy restarted a healthy worker"
    );

    let edited = fixture.provision(&edited_document()).await;
    let reminted = edited.first().ok_or("one queue declares a harness")?;
    assert_eq!(reminted.decision, AutoWorkerDecision::Reminted);
    let after = fixture.running_pid("auto/assistant").await?;
    assert_ne!(after, before, "a re-mint must replace the running worker");
    assert_ne!(
        fixture.argv("auto/assistant").await?,
        argv_before,
        "a re-mint must rewrite the argv"
    );
    fixture.teardown().await;
    Ok(())
}

/// 🔴 AN OPERATOR'S STOP SURVIVES A REDEPLOY. Neither an identical redeploy nor
/// an edited one may quietly start a worker they stopped.
#[tokio::test]
async fn a_stopped_auto_record_stays_stopped_across_a_redeploy() -> TestResult {
    let fixture = Fixture::new(true)?;
    drop(fixture.provision(&document()).await);
    assert!(fixture.running_pid("auto/assistant").await? > 0);
    drop(
        fixture
            .state
            .worker_supervisor()
            .stop("auto/assistant")
            .await?,
    );

    for source in [document(), edited_document()] {
        let outcomes = fixture.provision(&source).await;
        let first = outcomes.first().ok_or("one queue declares a harness")?;
        assert_eq!(
            first.decision,
            AutoWorkerDecision::OperatorStopped,
            "{}",
            first.detail
        );
        assert!(!first.decision.claims_running());
        assert_eq!(
            fixture.record("auto/assistant").await?.desired,
            DesiredState::Stopped,
            "a redeploy reversed an operator's stop"
        );
        let report = fixture.state.worker_supervisor().report().await?;
        assert!(
            report
                .workers
                .iter()
                .all(|worker| worker.name != "auto/assistant" || worker.pid.is_none()),
            "a stopped record must have no process: {report:?}"
        );
    }
    Ok(())
}

/// 🔴 RULE 1, THE OTHER DIRECTION. Removing the `harness` section retires the
/// record this server minted and stops its worker — otherwise the rule would
/// hold only at the moment a document was first deployed, and a worker would
/// stay alive on a launch nothing declares.
#[tokio::test]
async fn removing_the_harness_section_retires_the_record_it_minted() -> TestResult {
    let fixture = Fixture::new(true)?;
    drop(fixture.provision(&document()).await);
    assert!(fixture.running_pid("auto/assistant").await? > 0);

    let outcomes = fixture.provision(&document_without_harness()?).await;
    let retired = outcomes
        .first()
        .ok_or("the stale record must be reported")?;

    assert_eq!(retired.decision, AutoWorkerDecision::Retired);
    assert_eq!(retired.deployment.as_deref(), Some("auto/assistant"));
    assert!(
        fixture.names().await?.is_empty(),
        "the withdrawn record must be gone"
    );
    let report = fixture.state.worker_supervisor().report().await?;
    assert!(
        report.workers.is_empty(),
        "the withdrawn worker must be gone too: {report:?}"
    );
    Ok(())
}

/// A retirement is attributed by the workflow type that STAGED the document, so
/// one document can never withdraw another's worker.
#[tokio::test]
async fn a_deploy_never_retires_another_workflow_types_record() -> TestResult {
    let fixture = Fixture::new(true)?;
    drop(fixture.provision(&document()).await);

    // A different workflow type, declaring no queue at all.
    let outcomes = fixture
        .provision_as(&document_without_harness()?, "some_other_workflow")
        .await;

    assert!(outcomes.is_empty(), "{outcomes:?}");
    assert_eq!(fixture.names().await?, vec!["auto/assistant".to_owned()]);
    fixture.teardown().await;
    Ok(())
}

/// The supervisor's per-queue log is what answers "why is nothing serving this"
/// after the deploy that decided it is long gone.
#[tokio::test]
async fn the_decision_is_readable_from_the_managed_worker_report_afterwards() -> TestResult {
    let fixture = Fixture::dark()?;
    drop(fixture.provision(&document()).await);

    let report = fixture.state.worker_supervisor().report().await?;
    let entry = report
        .auto_provision
        .iter()
        .find(|entry| entry.task_queue == "assistant")
        .ok_or("the decision must survive on the status surface")?;
    assert_eq!(entry.decision, AutoWorkerDecision::DarkOutbox);
    assert!(entry.detail.contains("[outbox]"), "{}", entry.detail);
    Ok(())
}

/// One `ServerState` over an in-memory store, with a scratch snapshot root.
struct Fixture {
    state: ServerState,
    root: PathBuf,
    _home: tempfile::TempDir,
}

impl Fixture {
    fn new(commissioned: bool) -> Result<Self, Box<dyn std::error::Error>> {
        Self::build(liminal_outbox(), commissioned)
    }

    fn dark() -> Result<Self, Box<dyn std::error::Error>> {
        Self::build(OutboxConfig::default(), true)
    }

    fn build(outbox: OutboxConfig, commissioned: bool) -> Result<Self, Box<dyn std::error::Error>> {
        let store = Arc::new(InMemoryStore::default());
        let resolver = NamespaceResolver::authorization_only(
            NamespaceMode::SharedEngine,
            StaticWorkflowNamespaces::default(),
            StaticScheduleNamespaces::default(),
        );
        let mut runtime = crate::api::http::test_support::runtime_config();
        runtime.auth.enabled = false;
        runtime.deploy.enabled = true;
        runtime.outbox = outbox;
        let namespace_store: Arc<dyn aion_store::NamespaceStore> = store.clone();
        let worker_store: Arc<dyn WorkerDeploymentStore> = store;
        let state = ServerState::from_parts_with_control_stores(
            resolver,
            runtime,
            namespace_store,
            worker_store,
        );
        let home = tempfile::tempdir()?;
        let root = home.path().join("workers/documents");
        if commissioned
            && !state
                .worker_supervisor()
                .commission(policy()?, stand_in_worker(home.path())?)
        {
            return Err("the supervisor was already commissioned".into());
        }
        Ok(Self {
            state,
            root,
            _home: home,
        })
    }

    async fn provision(&self, source: &AwlSource) -> Vec<AutoWorkerOutcome> {
        self.provision_as(source, "assistant").await
    }

    async fn provision_as(
        &self,
        source: &AwlSource,
        workflow_type: &str,
    ) -> Vec<AutoWorkerOutcome> {
        provision(&self.state, &self.root, source, workflow_type).await
    }

    async fn names(&self) -> Result<Vec<String>, Box<dyn std::error::Error>> {
        Ok(self
            .state
            .worker_deployment_store()
            .list_worker_deployments()
            .await?
            .deployments
            .into_iter()
            .map(|deployment| deployment.name)
            .collect())
    }

    async fn record(&self, name: &str) -> Result<WorkerDeployment, Box<dyn std::error::Error>> {
        self.state
            .worker_deployment_store()
            .get_worker_deployment(name)
            .await?
            .ok_or_else(|| format!("`{name}` must exist").into())
    }

    async fn argv(&self, name: &str) -> Result<Vec<String>, Box<dyn std::error::Error>> {
        let WorkerArtifactRef::Builtin { verb } = self.record(name).await?.artifact;
        Ok(verb)
    }

    /// The pid of a supervised worker, waited for through the server's OWN
    /// status join: a spawn reports `Starting` until a pid is observed, so one
    /// immediate read would race the spawn rather than measure it.
    async fn running_pid(&self, name: &str) -> Result<u32, Box<dyn std::error::Error>> {
        let deadline = std::time::Instant::now() + Duration::from_secs(10);
        let mut last = String::from("no report was taken");
        while std::time::Instant::now() < deadline {
            let report = self.state.worker_supervisor().report().await?;
            match report.workers.iter().find(|worker| worker.name == name) {
                Some(worker) => match worker.pid {
                    Some(pid) => return Ok(pid),
                    None => last = format!("{worker:?}"),
                },
                None => last = format!("`{name}` is not in the report"),
            }
            tokio::time::sleep(Duration::from_millis(25)).await;
        }
        Err(format!("`{name}` never reported a pid: {last}").into())
    }

    async fn put_operator_record(
        &self,
        name: &str,
        task_queue: &str,
        namespace: Option<&str>,
        node: Option<&str>,
    ) -> Result<(), Box<dyn std::error::Error>> {
        let record = WorkerDeployment::new(
            NewWorkerDeployment {
                name: name.to_owned(),
                artifact: WorkerArtifactRef::Builtin {
                    verb: vec!["-c".to_owned(), "sleep 300".to_owned()],
                },
                binary: aion_store::DeployedBinaryIdentity {
                    version: "test".to_owned(),
                    commit: "test".to_owned(),
                    dirty: "false".to_owned(),
                    content_hash: "operator".to_owned(),
                },
                namespaces: namespace
                    .map(|namespace| BTreeSet::from([namespace.to_owned()]))
                    .unwrap_or_default(),
                task_queue: task_queue.to_owned(),
                node: node.map(ToOwned::to_owned),
                // Stopped, so the operator's record never spawns anything in a
                // test whose subject is the SKIP rather than the process.
                desired: DesiredState::Stopped,
            },
            chrono::Utc::now(),
        )?;
        drop(
            self.state
                .worker_deployment_store()
                .put_worker_deployment(record)
                .await?,
        );
        Ok(())
    }

    /// Unconditional cleanup for the tests that really do spawn a process, so
    /// a red run cannot leave a sleeping child behind for its remaining minutes.
    async fn teardown(&self) {
        drop(self.state.worker_supervisor().shutdown().await);
    }
}

fn liminal_outbox() -> OutboxConfig {
    OutboxConfig {
        enabled: true,
        transport: OutboxTransport::Liminal,
        liminal_listen_address: Some("127.0.0.1:50061".to_owned()),
        ..OutboxConfig::default()
    }
}

fn policy() -> Result<SupervisionPolicy, &'static str> {
    Ok(SupervisionPolicy {
        restart_backoff_initial: Duration::from_millis(20),
        restart_backoff_max: Duration::from_millis(20),
        restart_backoff_multiplier: NonZeroU32::new(1).ok_or("multiplier")?,
        restart_window: Duration::from_secs(600),
        max_restarts_per_window: NonZeroU32::new(5).ok_or("budget")?,
        stop_grace: Duration::from_secs(2),
    })
}

/// A stand-in for this server's own executable: a script that ignores whatever
/// argv it is handed and sleeps.
///
/// It must IGNORE the argv, not merely be harmless with it. `/bin/sh` reads the
/// record's first argument (`worker`) as a script path, exits 127 at once, and
/// burns the restart budget into `Failed` — so a test whose subject is "this is
/// still the same process" would be measuring a crash loop instead. Every
/// assertion here is about the RECORD and the supervisor's own report, so a
/// process that simply stays up is the right stand-in, and it is never a
/// re-execution of the test binary.
fn stand_in_worker(home: &Path) -> Result<ManagedExecutable, Box<dyn std::error::Error>> {
    let path = home.join("stand-in-worker");
    std::fs::write(&path, "#!/bin/sh\nexec sleep 300\n")?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt as _;
        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o700))?;
    }
    Ok(ManagedExecutable::Path(path))
}

/// The staged snapshot root must be under the fixture's own home, never the
/// developer's real `~/.aion`.
#[test]
fn the_fixture_root_is_never_the_real_home() -> TestResult {
    let home = tempfile::tempdir()?;
    let root = home.path().join("workers/documents");
    assert!(root.starts_with(home.path()));
    assert!(!root.starts_with(Path::new("/Users").join("shared")));
    Ok(())
}