aion-server 0.31.0

//! Liminal registration contract admission tests on the real wire DTO.

use crate::namespace::NamespaceGuard;
use std::sync::Arc;

use aion_package::{ExtractionLimits, Package};
use liminal::protocol::{WorkerActivityDescriptor, WorkerRegistration};

use super::{ConnectedWorkerRegistry, LiminalConnectionNotifier};
use crate::test_support::EngineUnderTest;

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

const SOURCE: &str = r"//! Liminal worker contract fixture.
workflow liminal_contract_test
  input amount: Int
  outcome completed: type Result, route success

type Result { approved: Bool }

worker payments
  action charge(amount: Int) -> Result

step run
  charge(amount: amount) -> result
  route completed(approved: result.approved)
";

async fn notifier_with_contract() -> TestResult<(LiminalConnectionNotifier, String, EngineUnderTest)>
{
    let root = tempfile::tempdir()?;
    let prepared = aion_awl_package::compile_and_assemble_awl(
        SOURCE,
        root.path(),
        "liminal_contract_test.awl",
    )?;
    let package = Package::load_from_bytes(prepared.archive, ExtractionLimits::unbounded())?;
    let version = package.content_hash().to_string();
    // Guarded on the line that creates it: the package load below is
    // fallible, and an engine that outlives an `?` with no owner is the leak
    // this fixture exists not to have.
    let engine = EngineUnderTest::new(Arc::new(
        aion::EngineBuilder::new()
            .stop_drain_timeout(std::time::Duration::from_secs(5))
            .store(aion_store::InMemoryStore::default())
            .in_memory_visibility()
            .build()
            .await?,
    ));
    engine.load_package(package).await?;
    let notifier = LiminalConnectionNotifier::new(ConnectedWorkerRegistry::default())
        .with_admission(
            NamespaceGuard::shared_engine(),
            false,
            tokio::runtime::Handle::current(),
        )
        .with_contract_catalog(engine.handle());
    Ok((notifier, version, engine))
}

fn registration(input_type: &str) -> WorkerRegistration {
    WorkerRegistration {
        namespaces: vec!["default".to_owned()],
        task_queue: "payments".to_owned(),
        node: None,
        activity_types: vec!["charge".to_owned()],
        identity: "liminal-contract-test-worker".to_owned(),
        activities: vec![WorkerActivityDescriptor {
            name: "charge".to_owned(),
            input_schema_json: serde_json::json!({
                "type": "object",
                "properties": {"amount": {"type": input_type}},
                "required": ["amount"]
            })
            .to_string(),
            output_schema_json: serde_json::json!({
                "type": "object",
                "properties": {"approved": {"type": "boolean"}},
                "required": ["approved"]
            })
            .to_string(),
        }],
    }
}

#[tokio::test]
async fn matching_liminal_registration_passes_pre_insertion_admission() -> TestResult {
    let (notifier, _, engine) = notifier_with_contract().await?;
    notifier.validate_registration_contract(&registration("integer"))?;
    engine.shutdown()?;
    Ok(())
}

#[tokio::test]
async fn mismatched_liminal_registration_names_field_and_exact_version() -> TestResult {
    let (notifier, version, engine) = notifier_with_contract().await?;
    let Err(error) = notifier.validate_registration_contract(&registration("string")) else {
        return Err("a narrowed incompatible input must be refused".into());
    };
    let message = error.to_string();

    assert!(message.contains("WORKER_CONTRACT_MISMATCH"));
    assert!(message.contains("input_schema.properties.amount.type"));
    assert!(message.contains(&version));
    engine.shutdown()?;
    Ok(())
}

// ---- Namespace admission on the liminal path (E1) --------------------------
//
// Before 2026-08-28 a worker dialling the liminal listener was inserted into
// the registry straight from the connection callback: any namespace it named,
// no guard, no mint-or-gate, no placement admission — while the identical
// worker over gRPC was refused. These pins hold the liminal path to the ONE
// admission (`ConnectedWorkerRegistry::admit_delivery`) in both directions:
// a notifier that cannot judge refuses by name, and one that can judge
// refuses and admits exactly as the gRPC path does.

use crate::config::AutoCreate;
use aion_store::{NamespaceOrigin, NamespacePlacement, NamespaceStore};
use liminal_server::server::connection::{ConnectionNotifier, ConnectionSupervisor};

/// An engine with the fixture package loaded, so the contract gate — which
/// runs BEFORE admission — passes and the refusal under test is admission's own.
async fn contract_engine() -> TestResult<EngineUnderTest> {
    let root = tempfile::tempdir()?;
    let prepared = aion_awl_package::compile_and_assemble_awl(
        SOURCE,
        root.path(),
        "liminal_contract_test.awl",
    )?;
    let package = Package::load_from_bytes(prepared.archive, ExtractionLimits::unbounded())?;
    // Guarded on the line that creates it, for the same reason as above.
    let engine = EngineUnderTest::new(Arc::new(
        aion::EngineBuilder::new()
            .stop_drain_timeout(std::time::Duration::from_secs(5))
            .store(aion_store::InMemoryStore::default())
            .in_memory_visibility()
            .build()
            .await?,
    ));
    engine.load_package(package).await?;
    Ok(engine)
}

#[tokio::test]
async fn a_listener_with_no_admission_bound_refuses_every_registration_by_name() -> TestResult {
    // The shape a listener has when a commissioning path forgets
    // `with_admission`: it must refuse, never admit unjudged.
    let engine = contract_engine().await?;
    let notifier = LiminalConnectionNotifier::new(ConnectedWorkerRegistry::default())
        .with_contract_catalog(engine.handle());
    let Err(error) = notifier.on_worker_registered(41, &registration("integer")) else {
        return Err("a listener with no admission bound must refuse, never admit unjudged".into());
    };
    let message = error.to_string();
    assert!(
        message.contains("no namespace admission bound"),
        "the refusal must name the missing admission, got: {message}"
    );
    assert!(
        message.contains("connection 41"),
        "the refusal must name the connection, got: {message}"
    );
    Ok(())
}

#[tokio::test]
async fn an_authenticating_server_refuses_the_credential_less_liminal_frame() -> TestResult {
    let engine = contract_engine().await?;
    let notifier = LiminalConnectionNotifier::new(ConnectedWorkerRegistry::default())
        .with_admission(
            NamespaceGuard::shared_engine(),
            true,
            tokio::runtime::Handle::current(),
        )
        .with_contract_catalog(engine.handle());

    let Err(error) = notifier.on_worker_registered(42, &registration("integer")) else {
        return Err("an auth-on server must refuse a frame that carries no credential".into());
    };
    let message = error.to_string();
    assert!(
        message.contains("carries no credential"),
        "the refusal must say why the frame cannot be scoped, got: {message}"
    );
    assert!(
        message.contains("liminal-contract-test-worker"),
        "the refusal must name the worker identity, got: {message}"
    );
    Ok(())
}

/// A minting registry over `store` with `namespace` pre-minted and pinned to
/// `nodes` — the registry-side fixture the gRPC placement pins use, so the
/// liminal path is judged against the SAME durable record.
async fn pinned_registry(
    store: &Arc<dyn NamespaceStore>,
    namespace: &str,
    nodes: &[&str],
) -> TestResult<ConnectedWorkerRegistry> {
    store
        .register_namespace(namespace, NamespaceOrigin::Explicit)
        .await?;
    store
        .set_namespace_placement(
            namespace,
            NamespacePlacement::Pinned {
                nodes: nodes.iter().map(|n| (*n).to_owned()).collect(),
            },
        )
        .await?;
    Ok(ConnectedWorkerRegistry::default()
        .with_namespace_minting(Arc::clone(store), AutoCreate::Open))
}

/// A fully wired notifier over `registry`: admission bound INSIDE the runtime
/// (so the bridge handle exists), the contract catalog loaded, and a
/// supervisor bound the way `outbox_commission` binds one.
async fn admitting_notifier(
    registry: ConnectedWorkerRegistry,
) -> TestResult<(Arc<LiminalConnectionNotifier>, EngineUnderTest)> {
    let engine = contract_engine().await?;
    let notifier = LiminalConnectionNotifier::new(registry)
        .with_admission(
            NamespaceGuard::shared_engine(),
            false,
            tokio::runtime::Handle::current(),
        )
        .with_contract_catalog(engine.handle());
    if !notifier.bind_supervisor(ConnectionSupervisor::new()?) {
        return Err("the supervisor must bind exactly once on a fresh notifier".into());
    }
    Ok((Arc::new(notifier), engine))
}

fn pinned_registration(node: &str) -> WorkerRegistration {
    let mut registration = registration("integer");
    registration.namespaces = vec!["iso".to_owned()];
    registration.node = Some(node.to_owned());
    registration
}

/// Drive the registration the way liminal does in production: from a plain
/// thread on NO runtime, so the admission's `block_on` bridge runs exactly
/// as it does under liminal's connection-process thread.
fn register_off_runtime(
    notifier: &Arc<LiminalConnectionNotifier>,
    pid: u64,
    registration: WorkerRegistration,
) -> TestResult<Result<(), liminal_server::ServerError>> {
    let notifier = Arc::clone(notifier);
    std::thread::spawn(move || notifier.on_worker_registered(pid, &registration))
        .join()
        .map_err(|_| "the registration thread panicked".into())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_liminal_worker_on_the_wrong_node_is_refused_by_the_pinned_namespace() -> TestResult {
    let store: Arc<dyn NamespaceStore> = Arc::new(aion_store::InMemoryStore::default());
    let registry = pinned_registry(&store, "iso", &["n1"]).await?;
    let (notifier, engine) = admitting_notifier(registry.clone()).await?;

    let Err(error) = register_off_runtime(&notifier, 43, pinned_registration("n2"))? else {
        return Err("an n2 worker must be refused by a namespace Pinned to n1".into());
    };
    let message = error.to_string();
    assert!(
        message.contains("Pinned") && message.contains("n1") && message.contains("n2"),
        "the refusal must name the pinned namespace, the required set and the worker's node, \
         got: {message}"
    );
    assert!(
        registry
            .workers_for("iso", "payments", "charge", None)?
            .is_empty(),
        "a refused liminal worker must not be in the pool — that is the bypass this pins"
    );
    engine.shutdown()?;
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_liminal_worker_on_the_required_node_is_admitted_into_the_pinned_namespace() -> TestResult
{
    let store: Arc<dyn NamespaceStore> = Arc::new(aion_store::InMemoryStore::default());
    let registry = pinned_registry(&store, "iso", &["n1"]).await?;
    let (notifier, engine) = admitting_notifier(registry.clone()).await?;

    register_off_runtime(&notifier, 44, pinned_registration("n1"))??;

    // 🔴 `pool_census`, not `workers_for` like its neighbours: this asserts a
    // REGISTRATION fact, and `workers_for` answers a DISPATCH-eligibility one.
    //
    // ADMISSION is a registration fact, so it is read from the census rather
    // than from `workers_for`. `workers_for` answers "who could take a dispatch
    // right now", and a liminal worker is deliberately not dispatchable between
    // registering and announcing its capacity — the liminal registration frame
    // cannot carry one. Using the dispatch-eligibility list as a proxy for
    // admission would make this pinned-placement test fail for a reason that
    // has nothing to do with placement.
    assert_eq!(
        registry
            .pool_census("iso", "payments", "charge", Some("n1"))?
            .compatible_workers,
        1,
        "an n1 worker must be admitted into the Pinned{{n1}} namespace's pool over liminal"
    );
    engine.shutdown()?;
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_registration_driven_from_a_multi_thread_runtime_worker_is_bridged() -> TestResult {
    // A test that drives the notifier from its own async body sits on a
    // runtime worker; the bridge hands the worker back (`block_in_place`)
    // rather than refusing or stalling the scheduler.
    let store: Arc<dyn NamespaceStore> = Arc::new(aion_store::InMemoryStore::default());
    let registry = pinned_registry(&store, "iso", &["n1"]).await?;
    let (notifier, engine) = admitting_notifier(registry.clone()).await?;

    notifier.on_worker_registered(45, &pinned_registration("n1"))?;

    // 🔴 `pool_census`, not `workers_for` like its neighbours: this asserts a
    // REGISTRATION fact, and `workers_for` answers a DISPATCH-eligibility one.
    //
    // The subject is the BRIDGE (the worker was handed back rather than the
    // scheduler stalled), so what has to be true is that the registration
    // landed. Read from the census for the same reason as the test above: a
    // liminal worker is registered before it is dispatchable, and this test is
    // about the former.
    assert_eq!(
        registry
            .pool_census("iso", "payments", "charge", Some("n1"))?
            .compatible_workers,
        1
    );
    engine.shutdown()?;
    Ok(())
}

#[tokio::test]
async fn a_registration_driven_from_a_current_thread_runtime_worker_is_refused_by_name()
-> TestResult {
    // A current-thread runtime worker has nothing to hand back to: blocking
    // it would deadlock the runtime the admission runs on. Refused, named.
    let store: Arc<dyn NamespaceStore> = Arc::new(aion_store::InMemoryStore::default());
    let registry = pinned_registry(&store, "iso", &["n1"]).await?;
    let (notifier, engine) = admitting_notifier(registry.clone()).await?;

    let Err(error) = notifier.on_worker_registered(46, &pinned_registration("n1")) else {
        return Err(
            "a current-thread runtime worker cannot be blocked; the call must refuse".into(),
        );
    };
    let message = error.to_string();
    assert!(
        message.contains("current-thread Tokio runtime worker"),
        "the refusal must name the runtime flavour, got: {message}"
    );
    assert!(
        registry
            .workers_for("iso", "payments", "charge", None)?
            .is_empty(),
        "a refused registration must leave no worker behind"
    );
    engine.shutdown()?;
    Ok(())
}