aion-server 0.13.0

//! Pins for the read-only deployed-AWL surface.
//!
//! Every read pin is paired with the absence it is supposed to preserve: the
//! byte-identity pin also proves the archive it read is unchanged, and the
//! projection pin also proves its staging directory is gone. A read test that
//! only proved the read would pass just as well over a surface that rewrote
//! everything it touched.

use std::collections::BTreeMap;

use aion_package::{
    AdditionalWorkflowContract, AwlSource, BeamModule, BeamSet, ManifestVersion, PackageContract,
    SignalContract, WorkflowEntry, content_hash,
};
use aion_store::PackageRecord;
use chrono::{TimeZone, Utc};
use serde_json::json;

use super::document::read_document;
use super::fixtures::{
    DOCUMENT, SCHEMA_BYTES, SCHEMA_DOCUMENT, SCHEMA_PATH, catalog_entry, manifest, record,
    record_with_contract,
};
use super::list::project_versions;
use super::projection;
use super::types::{DeployedError, DeployedSourceState};
// Through the module's public re-export deliberately: this is the path a
// consumer of the surface names the type by.
use super::DeployedSignal;

/// 🔴 THE READ-ONLY PIN, both halves. The archived document is served
/// BYTE-IDENTICAL to what `Package::awl()` carries — and the archive bytes the
/// read consumed are unchanged afterwards. Without the second half this would
/// pass over a surface that re-serialised, re-stamped, or normalised the
/// package it read.
#[test]
fn the_archived_document_is_served_byte_identical_and_the_archive_is_unchanged()
-> Result<(), Box<dyn std::error::Error>> {
    let awl = AwlSource::new(
        "deployed_probe.awl",
        DOCUMENT,
        Vec::<(String, Vec<u8>)>::new(),
    );
    let row = record(manifest("deployed_probe"), Some(awl), 1_700_000_000)?;
    let before = row.archive.clone();
    let archives = vec![row];

    let document = read_document(&archives, "deployed_probe", &archives[0].content_hash)?;

    assert_eq!(
        document.source, DOCUMENT,
        "the served source must be the archived bytes verbatim"
    );
    assert_eq!(document.document_name, "deployed_probe.awl");
    assert_eq!(document.workflow_type, "deployed_probe");
    assert_eq!(document.content_hash, archives[0].content_hash);
    assert!(document.schemas.is_empty());
    assert!(
        document.projection.ok,
        "the deployed document must project: {:?}",
        document.projection.diagnostics
    );
    assert_eq!(document.projection.steps, Some(0));
    assert_eq!(
        archives[0].archive, before,
        "reading a deployed document rewrote the archive it read"
    );
    Ok(())
}

/// 🔴 THE ABSENCE PIN. A package with no archived AWL source is a first-class
/// state, not an error class of its own invention and not an empty document:
/// the refusal names BOTH reasons a version can lack source, because an
/// operator told only "no source" cannot tell a Gleam workflow from a broken
/// one.
#[test]
fn a_package_without_archived_source_refuses_and_states_both_reasons()
-> Result<(), Box<dyn std::error::Error>> {
    let row = record(manifest("gleam_authored"), None, 1_700_000_000)?;
    let archives = vec![row];

    let refusal = read_document(&archives, "gleam_authored", &archives[0].content_hash)
        .err()
        .ok_or("a package with no archived AWL source must refuse")?;

    assert!(
        matches!(refusal, DeployedError::NoArchivedSource { .. }),
        "expected the absence class, got {refusal:?}"
    );
    let message = refusal.to_string();
    assert!(message.contains("Gleam"), "{message}");
    assert!(
        message.contains("before deploys archived their source"),
        "{message}"
    );
    Ok(())
}

/// The pair is verified, not just the hash: a right hash under a workflow type
/// the archive never declared must not resolve. Otherwise the surface would
/// serve a document under a type that has nothing to do with it.
#[test]
fn a_workflow_type_the_archive_does_not_declare_is_not_found()
-> Result<(), Box<dyn std::error::Error>> {
    let awl = AwlSource::new(
        "deployed_probe.awl",
        DOCUMENT,
        Vec::<(String, Vec<u8>)>::new(),
    );
    let archives = vec![record(
        manifest("deployed_probe"),
        Some(awl),
        1_700_000_000,
    )?];

    let refusal = read_document(&archives, "someone_elses_type", &archives[0].content_hash)
        .err()
        .ok_or("a mismatched type/version pair must not resolve")?;
    assert!(
        matches!(refusal, DeployedError::NotFound { .. }),
        "expected not-found, got {refusal:?}"
    );

    let unknown = read_document(&archives, "deployed_probe", &"f".repeat(64))
        .err()
        .ok_or("an unknown version must not resolve")?;
    assert!(
        matches!(unknown, DeployedError::NotFound { .. }),
        "expected not-found, got {unknown:?}"
    );
    Ok(())
}

/// 🔴 THE FORM-SURFACE PIN. The served `input_schema` and `signals` are the
/// archive's identity-committed contract — read through the same
/// `Package::contract` seam the `/assistant` descriptor uses — and each entry
/// of a multi-entry archive resolves to ITS OWN commitments: its own start
/// schema, never the primary's, and NO inherited signal set — the contract
/// commits signals to the primary entry only, so an additional entry answers
/// `signals: None` (absent), not the primary's list and not a false "empty".
/// A console form built from this response is a form for exactly the version
/// and entry the operator asked about.
#[test]
fn the_deployed_document_serves_the_committed_start_schema_and_signals()
-> Result<(), Box<dyn std::error::Error>> {
    let parent_schema = json!({
        "type": "object",
        "properties": { "objective": { "type": "string" } },
        "required": ["objective"],
    });
    let child_schema = json!({
        "type": "object",
        "properties": { "count": { "type": "integer" } },
    });
    let signal_schema = json!({
        "type": "object",
        "properties": { "approved": { "type": "boolean" } },
    });
    let contract = PackageContract {
        input_schema: parent_schema.clone(),
        output_schema: json!({ "type": "object" }),
        workers: Vec::new(),
        children: Vec::new(),
        signals: vec![SignalContract {
            name: "approve".to_owned(),
            input_schema: signal_schema.clone(),
        }],
        additional_workflows: vec![AdditionalWorkflowContract {
            workflow_type: "child_flow".to_owned(),
            input_schema: child_schema.clone(),
            output_schema: json!({ "type": "object" }),
        }],
        unscoped_activities: Vec::new(),
    };
    let mut declared = manifest("parent_flow");
    declared.additional_workflows = vec![WorkflowEntry {
        workflow_type: "child_flow".to_owned(),
        entry_module: "parent_flow".to_owned(),
        entry_function: "run".to_owned(),
        input_schema: child_schema.clone(),
        output_schema: json!({ "type": "object" }),
        timeout: None,
        internal: false,
    }];
    let awl = AwlSource::new("parent_flow.awl", DOCUMENT, Vec::<(String, Vec<u8>)>::new());
    let archives = vec![record_with_contract(
        declared,
        contract,
        Some(awl),
        1_700_000_000,
    )?];

    let parent = read_document(&archives, "parent_flow", &archives[0].content_hash)?;
    assert_eq!(parent.input_schema, Some(parent_schema));
    assert_eq!(
        parent.signals,
        Some(vec![DeployedSignal {
            name: "approve".to_owned(),
            input_schema: signal_schema,
        }])
    );

    let child = read_document(&archives, "child_flow", &archives[0].content_hash)?;
    assert_eq!(
        child.input_schema,
        Some(child_schema),
        "an additional entry must serve ITS schema, not the primary's"
    );
    assert_eq!(
        child.signals, None,
        "the contract commits no signal set to an additional entry, so it \
         must not inherit the primary's"
    );
    Ok(())
}

/// The `None` case, stated by the field's own doc: a package whose stored
/// identity predates contract commitment carries NO committed contract, so
/// `input_schema` is `None` and `signals` is empty — while the document read
/// itself is still served whole. A schema invented from the un-committed
/// manifest here would be a claim the version hash never made.
#[test]
fn a_pre_contract_package_serves_no_schema_but_still_serves_its_document()
-> Result<(), Box<dyn std::error::Error>> {
    let archives = vec![legacy_record(1_700_000_000)?];

    let document = read_document(&archives, "legacy_probe", &archives[0].content_hash)?;

    assert_eq!(
        document.input_schema, None,
        "a pre-contract identity commits to no start schema"
    );
    assert_eq!(
        document.signals, None,
        "a pre-contract identity commits to no signal set — absent, not empty"
    );
    assert_eq!(
        document.source, DOCUMENT,
        "the schema absence must not degrade the document read itself"
    );
    Ok(())
}

/// A persisted archive whose stored identity is the beams-only LEGACY hash
/// and which carries no `contract.json` — the pre-contract deploy shape.
/// Written by hand with the `zip` crate because the current builder
/// (correctly) refuses to mint a contract-less identity.
fn legacy_record(deployed_at_seconds: i64) -> Result<PackageRecord, Box<dyn std::error::Error>> {
    use std::io::Write as _;

    let beam_bytes = vec![1_u8, 2, 3];
    let beams = BeamSet::new(vec![BeamModule::new("legacy_probe", beam_bytes.clone())])?;
    let mut legacy_manifest = manifest("legacy_probe");
    legacy_manifest.version = ManifestVersion::new(content_hash(&beams).to_string());

    let mut archive = zip::ZipWriter::new(std::io::Cursor::new(Vec::new()));
    let options = zip::write::SimpleFileOptions::default();
    archive.start_file("manifest.json", options)?;
    archive.write_all(&serde_json::to_vec(&legacy_manifest)?)?;
    archive.start_file("beam/legacy_probe.beam", options)?;
    archive.write_all(&beam_bytes)?;
    archive.start_file("awl/document/legacy_probe.awl", options)?;
    archive.write_all(DOCUMENT.as_bytes())?;
    let bytes = archive.finish()?.into_inner();

    Ok(PackageRecord {
        workflow_type: "legacy_probe".to_owned(),
        content_hash: legacy_manifest.version.as_str().to_owned(),
        archive: bytes,
        deployed_at: Utc
            .timestamp_opt(deployed_at_seconds, 0)
            .single()
            .ok_or("fixture instant is not representable")?,
    })
}

/// A multi-entry archive persists ONE row under its primary type while the
/// catalog registers every additional entry. An additional entry must resolve
/// to its own archive's document rather than reporting no source.
#[test]
fn an_additional_entry_resolves_to_its_archives_document() -> Result<(), Box<dyn std::error::Error>>
{
    let mut declared = manifest("parent_flow");
    declared.additional_workflows = vec![WorkflowEntry {
        workflow_type: "child_flow".to_owned(),
        entry_module: "parent_flow".to_owned(),
        entry_function: "run".to_owned(),
        input_schema: json!({ "type": "object" }),
        output_schema: json!({ "type": "object" }),
        timeout: None,
        internal: false,
    }];
    let awl = AwlSource::new("parent_flow.awl", DOCUMENT, Vec::<(String, Vec<u8>)>::new());
    let archives = vec![record(declared, Some(awl), 1_700_000_000)?];

    let child = read_document(&archives, "child_flow", &archives[0].content_hash)?;
    assert_eq!(child.workflow_type, "child_flow");
    assert_eq!(child.source, DOCUMENT);
    Ok(())
}

/// The four source states are distinguishable in one listing, and the listing
/// is the union of the two sets: a loaded version with no archive, and an
/// archived version the engine never loaded, both appear.
#[test]
fn the_listing_states_every_source_condition_distinctly() -> Result<(), Box<dyn std::error::Error>>
{
    let with_source = record(
        manifest("with_source"),
        Some(AwlSource::new(
            "with_source.awl",
            DOCUMENT,
            [(SCHEMA_PATH.to_owned(), SCHEMA_BYTES.to_vec())],
        )),
        1_700_000_100,
    )?;
    let without_source = record(manifest("without_source"), None, 1_700_000_200)?;
    let operator_file = record(manifest("operator_file"), None, 1_700_000_300)?;
    let corrupt = PackageRecord {
        workflow_type: "corrupt".to_owned(),
        content_hash: "c".repeat(64),
        archive: b"this is not a zip archive".to_vec(),
        deployed_at: Utc
            .timestamp_opt(1_700_000_400, 0)
            .single()
            .ok_or("fixture instant is not representable")?,
    };

    // `operator_file` is loaded but NOT persisted; `without_source` is
    // persisted but NOT loaded.
    let catalog = vec![
        catalog_entry(&with_source, true)?,
        catalog_entry(&operator_file, true)?,
        catalog_entry(&corrupt, false)?,
    ];
    let archives = vec![with_source.clone(), without_source.clone(), corrupt.clone()];

    let listing = project_versions(catalog, &archives);
    let state = |workflow_type: &str| {
        listing
            .iter()
            .find(|version| version.workflow_type == workflow_type)
            .map(|version| version.source.clone())
    };

    assert_eq!(
        state("with_source"),
        Some(DeployedSourceState::Available {
            document_name: "with_source.awl".to_owned(),
            schema_count: 1,
        })
    );
    assert_eq!(state("without_source"), Some(DeployedSourceState::Absent));
    assert_eq!(
        state("operator_file"),
        Some(DeployedSourceState::NotPersisted)
    );
    assert!(
        matches!(
            state("corrupt"),
            Some(DeployedSourceState::Unreadable { .. })
        ),
        "a corrupt archive must be named unreadable, never rendered as no source"
    );

    let loaded = |workflow_type: &str| {
        listing
            .iter()
            .find(|version| version.workflow_type == workflow_type)
            .map(|version| (version.loaded, version.deployed_at.is_some()))
    };
    assert_eq!(loaded("operator_file"), Some((true, false)));
    assert_eq!(loaded("without_source"), Some((false, true)));
    assert_eq!(loaded("with_source"), Some((true, true)));
    assert_eq!(
        listing.len(),
        4,
        "the listing must be the union, not either side"
    );
    Ok(())
}

/// 🔴 THE STAGING PIN, both halves. Schema imports RESOLVE from the archive's
/// own schema bytes (survival) — and the directory those bytes were staged in
/// is gone when the call returns (absence). The staging parent is a directory
/// this test owns, so an empty listing afterwards is this call's leavings and
/// nobody else's.
#[test]
fn deployed_projection_resolves_archived_schemas_and_leaves_nothing_behind()
-> Result<(), Box<dyn std::error::Error>> {
    let staging_parent = crate::test_support::private_tempdir()?;
    let mut schemas = BTreeMap::new();
    schemas.insert(SCHEMA_PATH.to_owned(), SCHEMA_BYTES.to_vec());

    let projected = projection::project_in(SCHEMA_DOCUMENT, &schemas, staging_parent.path())?;

    assert!(
        projected.ok,
        "the archived schema must resolve: {:?}",
        projected.diagnostics
    );
    assert!(projected.semantic.is_some());
    assert_eq!(
        std::fs::read_dir(staging_parent.path())?.count(),
        0,
        "the staging directory outlived the call that made it"
    );
    Ok(())
}

/// The inverted control for the pin above: WITHOUT the archived schemas the
/// same document does not check. Without this, the staging pin would pass just
/// as well over a projection that ignored schemas entirely.
#[test]
fn the_same_document_without_its_archived_schemas_does_not_resolve()
-> Result<(), Box<dyn std::error::Error>> {
    let staging_parent = crate::test_support::private_tempdir()?;
    let projected =
        projection::project_in(SCHEMA_DOCUMENT, &BTreeMap::new(), staging_parent.path())?;

    assert!(
        !projected.ok,
        "a schema import cannot resolve with no schemas staged; the staging pin would be vacuous"
    );
    Ok(())
}

/// Defence in depth where archived names become filesystem paths: a schema
/// entry that would escape the staging directory refuses the whole document
/// rather than being written, skipped, or sanitised into something else.
#[test]
fn a_schema_path_that_would_escape_staging_refuses_the_document()
-> Result<(), Box<dyn std::error::Error>> {
    let staging_parent = crate::test_support::private_tempdir()?;
    for path in ["../escape.json", "/absolute.json", ""] {
        let mut schemas = BTreeMap::new();
        schemas.insert(path.to_owned(), SCHEMA_BYTES.to_vec());
        let refusal = projection::project_in(SCHEMA_DOCUMENT, &schemas, staging_parent.path())
            .err()
            .ok_or_else(|| format!("schema path `{path}` was staged instead of refused"))?;
        assert!(
            matches!(refusal, DeployedError::UnsafeSchemaPath { .. }),
            "expected an unsafe-path refusal for `{path}`, got {refusal:?}"
        );
    }
    assert_eq!(
        std::fs::read_dir(staging_parent.path())?.count(),
        0,
        "a refused document left a staging directory behind"
    );
    Ok(())
}

/// Archived schemas reach the caller as text alongside the document, so a
/// consumer can show what the deployed version was actually checked against.
#[test]
fn archived_schemas_are_served_with_the_document() -> Result<(), Box<dyn std::error::Error>> {
    let awl = AwlSource::new(
        "schema_probe.awl",
        SCHEMA_DOCUMENT,
        [(SCHEMA_PATH.to_owned(), SCHEMA_BYTES.to_vec())],
    );
    let archives = vec![record(manifest("schema_probe"), Some(awl), 1_700_000_000)?];

    let document = read_document(&archives, "schema_probe", &archives[0].content_hash)?;

    assert_eq!(document.schemas.len(), 1);
    assert_eq!(document.schemas[0].path, SCHEMA_PATH);
    assert_eq!(
        document.schemas[0].text.as_deref(),
        Some(std::str::from_utf8(SCHEMA_BYTES)?)
    );
    assert_eq!(document.schemas[0].byte_length, SCHEMA_BYTES.len());
    assert!(
        document.projection.ok,
        "a schema-importing deployed document must project from its own archived schemas: {:?}",
        document.projection.diagnostics
    );
    Ok(())
}