aion-server 0.31.0

//! Standing a built-in agent worker up because a deployed document asked for
//! one — and saying so, on every arm, including the arms that mint nothing.
//!
//! # The rules, in order
//!
//! 1. **No `harness` section, no record.** Auto-provision keys on the section,
//!    never on deploys generally — and the rule is enforced in BOTH directions:
//!    a redeploy that removed the section (or renamed the queue) retires the
//!    record this server minted for it, rather than leaving a worker alive on
//!    a launch the document no longer declares.
//! 2. **An operator's record wins.** If a record whose name is not `auto/…`
//!    could serve the queue on this node in this namespace, nothing is minted,
//!    nothing is replaced, and the skip says which record won.
//! 3. **A dark outbox is a REFUSAL, not a mint.** Without a liminal worker
//!    listener a record would dial nothing forever; the refusal names the
//!    `[outbox]` settings that change it.
//! 4. **A redeploy RE-MINTS.** A record replays its argv verbatim, so a new
//!    document is a new argv, and the worker is restarted onto it. A redeploy
//!    of identical bytes resolves to the identical argv and the running worker
//!    is left alone.
//! 5. **An operator's STOP is preserved.** A record they stopped stays stopped
//!    across a redeploy: the desired state carries forward, and the outcome
//!    says the queue is unserved by the operator's own decision.
//!
//! # It never claims a worker it did not start
//!
//! Every arm that reports a running worker does so only after a convergence
//! RETURNED success. A record written durably onto a server with no
//! `[worker_supervision]` policy is [`AutoWorkerDecision::RecordedNotRunning`]
//! — the durable record and the live process disagreeing while the record
//! looks healthy is the whole defect this lane exists to end, and a status
//! surface that claimed otherwise would rebuild it one layer up.

use std::path::Path;

use aion_awl::{CompiledHarness, CompiledHarnessKind, CwdSource};
use aion_core::{ClusterEvent, PutOutcome};
use aion_package::AwlSource;
use aion_store::{DesiredState, WorkerDeployment, WorkerDeploymentListing};

use crate::ServerState;
use crate::worker::supervisor::{Convergence, converge_and_report};

use super::documents;
use super::listener::worker_dial_address;
use super::narrate;
use super::outcome::{AutoWorkerDecision, AutoWorkerOutcome};
use super::queues::{self, HarnessQueue};
use super::record;
use super::retire;

/// The operation label every auto-provision log line carries.
pub(super) const OPERATION: &str = "worker.auto_provision";

/// Provision a built-in agent worker for every queue `awl` declares one on,
/// and retire the records this workflow type's earlier deploys left behind.
///
/// Never fails: a server whose auto-provision could not run is a server whose
/// queue is unserved, not a server that must refuse the deploy that already
/// succeeded. Every arm returns an outcome carrying its own account, and the
/// outcomes are also recorded on the supervisor so the managed-worker surface
/// can answer "why is nothing serving this queue" later, not just at deploy.
pub async fn provision(
    state: &ServerState,
    root: &Path,
    awl: &AwlSource,
    workflow_type: &str,
) -> Vec<AutoWorkerOutcome> {
    let queues = match queues::harness_queues(awl.document()) {
        Ok(queues) => queues,
        Err(error) => return finish(state, vec![document_failure(&error, workflow_type)]),
    };
    let listing = match state
        .worker_deployment_store()
        .list_worker_deployments()
        .await
    {
        Ok(listing) => listing,
        Err(error) => {
            return finish(
                state,
                vec![AutoWorkerOutcome::document_level(
                    workflow_type,
                    format!(
                        "the durable worker deployments could not be listed, so this server \
                         cannot tell what already serves the queues `{workflow_type}` declares, \
                         and provisioned nothing: {error}"
                    ),
                )],
            );
        }
    };

    let mut outcomes = Vec::with_capacity(queues.len().saturating_add(1));
    for queue in &queues {
        outcomes.push(provision_queue(state, root, awl, workflow_type, queue, &listing).await);
    }
    // Rule 1, the other direction. Run even when `queues` is empty — that IS
    // the "the harness section was removed" case, and it is the one where a
    // worker is otherwise left alive replaying a launch nobody declares.
    outcomes.extend(retire::withdraw_undeclared(state, workflow_type, &queues, &listing).await);
    retire::prune_snapshots(state, root).await;
    finish(state, outcomes)
}

/// Log every outcome, record it on the supervisor, and hand it back.
fn finish(state: &ServerState, outcomes: Vec<AutoWorkerOutcome>) -> Vec<AutoWorkerOutcome> {
    for outcome in &outcomes {
        log_outcome(outcome);
    }
    state.worker_supervisor().record_auto_provision(&outcomes);
    outcomes
}

/// A harness-queue read failure, attributed to the queue when it named one.
fn document_failure(error: &queues::HarnessQueueError, workflow_type: &str) -> AutoWorkerOutcome {
    match error {
        queues::HarnessQueueError::Parse { .. } => {
            AutoWorkerOutcome::document_level(workflow_type, error.to_string())
        }
        queues::HarnessQueueError::Harness { task_queue, .. }
        | queues::HarnessQueueError::NoServiceableAction { task_queue, .. } => {
            AutoWorkerOutcome::new(
                task_queue.clone(),
                workflow_type,
                AutoWorkerDecision::Failed,
                None,
                error.to_string(),
            )
        }
    }
}

/// The decision for one declaring queue.
async fn provision_queue(
    state: &ServerState,
    root: &Path,
    awl: &AwlSource,
    workflow_type: &str,
    queue: &HarnessQueue,
    listing: &WorkerDeploymentListing,
) -> AutoWorkerOutcome {
    let task_queue = queue.task_queue.as_str();
    let name = record::auto_name(task_queue);
    let namespace = state.runtime_config().default_namespace.clone();
    let node = state.cluster_self_node();

    if let Some(skip) = operator_record_skip(listing, task_queue, &namespace, node, workflow_type) {
        return skip;
    }

    // The most specific refusal first: a harness naming paths this host does
    // not have would mint a worker that fails every dispatch routed to it
    // (measured 2026-08-26 — a fleet document deployed cross-host stood two
    // such workers up on the server's own machine, and one in three dispatches
    // on those queues burned an attempt on a spawn that could only fail).
    if let Some(why) = harness_unrunnable_here(&queue.harness) {
        return narrate::unrunnable_harness(task_queue, workflow_type, &why);
    }

    let dial = match worker_dial_address(&state.runtime_config().outbox) {
        Ok(dial) => dial,
        Err(dark) => return narrate::dark_outbox(task_queue, workflow_type, &dark),
    };

    let staged = match documents::stage(root, awl, workflow_type) {
        Ok(staged) => staged,
        Err(error) => {
            return narrate::failure(
                task_queue,
                workflow_type,
                format!(
                    "the deployed document could not be staged for a built-in agent worker on \
                     `{task_queue}`, so nothing was minted: {error}"
                ),
            );
        }
    };
    let binary = match crate::worker::capture_binary_identity() {
        Ok(binary) => binary,
        Err(error) => {
            return narrate::failure(
                task_queue,
                workflow_type,
                format!(
                    "the running server executable could not be identified, so no built-in agent \
                     worker was minted for `{task_queue}`: {error}"
                ),
            );
        }
    };

    let current = listing
        .deployments
        .iter()
        .find(|record| record.name == name);
    // An operator's STOP is durable intent and survives a redeploy. Only a
    // record that does not yet exist is minted `Running`.
    let desired = current.map_or(DesiredState::Running, |record| record.desired);
    let mut requested = record::new_deployment(task_queue, &staged.path, &dial, &namespace, binary);
    requested.desired = desired;

    // An identical redeploy must not restart a healthy worker. The document is
    // content-addressed, so "same document" is exactly "same argv", and that is
    // the whole test — no durable write, no cluster event, no replacement.
    if let Some(current) = current
        && current.artifact == requested.artifact
        && current.task_queue == requested.task_queue
        && current.namespaces == requested.namespaces
    {
        let converged = converge(state, &name, Convergence::Idempotent).await;
        return narrate::unchanged(
            &narrate::Subject {
                task_queue,
                workflow_type,
                name: &name,
                namespace: &namespace,
                document: &staged.path,
            },
            desired,
            converged,
        );
    }

    mint(
        state,
        MintRequest {
            subject: narrate::Subject {
                task_queue,
                workflow_type,
                name: &name,
                namespace: &namespace,
                document: &staged.path,
            },
            dial: &dial,
        },
        requested,
    )
    .await
}

/// Everything the mint's own reporting needs, held together so the write and
/// the sentence describing it cannot disagree about what was written.
struct MintRequest<'a> {
    subject: narrate::Subject<'a>,
    dial: &'a str,
}

/// Write the record and converge this node onto it.
async fn mint(
    state: &ServerState,
    request: MintRequest<'_>,
    requested: aion_store::NewWorkerDeployment,
) -> AutoWorkerOutcome {
    let stopped = requested.desired == DesiredState::Stopped;
    let record = match WorkerDeployment::new(requested, chrono::Utc::now()) {
        Ok(record) => record,
        Err(error) => {
            return narrate::failure(
                request.subject.task_queue,
                request.subject.workflow_type,
                format!(
                    "the auto worker-deployment record for `{}` is invalid: {error}",
                    request.subject.task_queue
                ),
            );
        }
    };
    let result = match state
        .worker_deployment_store()
        .put_worker_deployment(record)
        .await
    {
        Ok(result) => result,
        Err(error) => {
            return narrate::failure(
                request.subject.task_queue,
                request.subject.workflow_type,
                format!(
                    "the auto worker-deployment record for `{}` could not be persisted: {error}",
                    request.subject.task_queue
                ),
            );
        }
    };
    publish_put(state, &result);

    // A replaced record rewrote the argv a live worker replays verbatim, so
    // the running process is serving the SUPERSEDED document and must be
    // replaced — that is what "re-minting is the fix" means mechanically.
    let mode = match result.outcome {
        PutOutcome::Created => Convergence::Idempotent,
        PutOutcome::Replaced => Convergence::Replacing,
    };
    let converged = converge(state, request.subject.name, mode).await;
    // The decision is chosen HERE, from the put outcome joined with whether the
    // convergence actually succeeded, so the sentence below can only repeat a
    // fact rather than assert one.
    let decision = match (stopped, result.outcome, converged) {
        (true, _, _) => AutoWorkerDecision::OperatorStopped,
        (false, _, false) => AutoWorkerDecision::RecordedNotRunning,
        (false, PutOutcome::Created, true) => AutoWorkerDecision::Minted,
        (false, PutOutcome::Replaced, true) => AutoWorkerDecision::Reminted,
    };
    narrate::written(&request.subject, decision, request.dial)
}

/// The operator's own record for this queue, if one could actually serve it
/// HERE — the whole of rule 2, held apart so the skip and the mint cannot grow
/// into one another.
///
/// Node and namespace are part of the question, not decoration. A record pinned
/// to another node, or bound to namespaces this server's default is not in,
/// serves this queue somewhere else; treating it as a winner would leave the
/// queue unserved on this box with an outcome saying it was served.
fn operator_record_skip(
    listing: &WorkerDeploymentListing,
    task_queue: &str,
    namespace: &str,
    node: Option<&str>,
    workflow_type: &str,
) -> Option<AutoWorkerOutcome> {
    let operator = listing.deployments.iter().find(|record| {
        record.task_queue == task_queue
            && !record::is_auto_name(&record.name)
            && record.namespaces.contains(namespace)
            && match record.node.as_deref() {
                None => true,
                Some(pinned) => node == Some(pinned),
            }
    })?;
    Some(AutoWorkerOutcome::new(
        task_queue,
        workflow_type,
        AutoWorkerDecision::OperatorRecord,
        Some(operator.name.clone()),
        format!(
            "worker deployment `{}` already serves task queue `{task_queue}` in namespace \
             `{namespace}` and was authored by an operator, so nothing was minted or replaced: an \
             explicit record wins over the document's declaration. Delete or rename it if you want \
             this server to provision the queue itself",
            operator.name
        ),
    ))
}

/// Bring the freshly written record's live state to its durable desired state
/// on this node, answering whether it actually happened.
///
/// The boolean is the whole point: a caller that reported "started it" without
/// reading it would be making the claim this lane exists to stop anyone making.
async fn converge(state: &ServerState, name: &str, mode: Convergence) -> bool {
    converge_and_report(state.worker_supervisor(), name, mode, OPERATION)
        .await
        .is_ok()
}

/// Publish the same put event the worker-deployment transport publishes, so the
/// console's live feed cannot tell an auto-provisioned record from a typed one.
fn publish_put(state: &ServerState, result: &aion_store::WorkerDeploymentPutResult) {
    let name = result.deployment.name.clone();
    let outcome = result.outcome;
    let desired_state = result.deployment.desired;
    let binary_version = result.deployment.binary.version.clone();
    let binary_content_hash = result.deployment.binary.content_hash.clone();
    drop(
        state
            .cluster_publisher()
            .emit(|meta| ClusterEvent::WorkerDeploymentPut {
                meta,
                name,
                outcome,
                desired_state,
                binary_version,
                binary_content_hash,
            }),
    );
}

/// One structured line per queue, at the severity the decision earns: a refusal
/// an operator has to act on is never an `info` they scroll past.
pub(super) fn log_outcome(outcome: &AutoWorkerOutcome) {
    let task_queue = outcome.task_queue.as_str();
    let workflow_type = outcome.workflow_type.as_str();
    let decision = outcome.decision.token();
    let detail = outcome.detail.as_str();
    if outcome.decision.is_refusal() {
        tracing::error!(
            operation = OPERATION,
            workflow_type,
            task_queue,
            decision,
            "{detail}"
        );
    } else {
        tracing::info!(
            operation = OPERATION,
            workflow_type,
            task_queue,
            decision,
            "{detail}"
        );
    }
}

#[cfg(test)]
#[path = "provision_tests.rs"]
mod tests;

/// Why this host cannot launch the document's harness, when it cannot.
///
/// The probe is deliberately shallow — existence and the executable bit, never
/// a spawn: presence is not proof the agent WORKS, and no answer here claims
/// it is. What the probe refuses is the one face that can never succeed: an
/// agent binary or a literal working directory that does not exist on this
/// machine. A `{workspace_root}`-templated or dispatch-bound cwd is resolvable
/// only later and is not judged here.
fn harness_unrunnable_here(harness: &CompiledHarness) -> Option<String> {
    let (command, cwd) = match &harness.kind {
        CompiledHarnessKind::Acp { command, cwd, .. } => (Some(command.clone()), cwd),
        CompiledHarnessKind::Norn { binary, cwd } => (binary.clone(), cwd),
    };
    if let Some(command) = command {
        if let Some(why) = executable_missing(&command) {
            return Some(why);
        }
    } else if let Some(why) = resolves_nowhere_on_path("norn") {
        return Some(why);
    }
    if let CwdSource::Literal(path) = cwd
        && path.is_absolute()
        && !path.is_dir()
    {
        return Some(format!(
            "the harness working directory `{}` is not a directory on this host",
            path.display()
        ));
    }
    None
}

/// The named agent executable's absence, as a sentence, or `None` when it is
/// present and executable.
fn executable_missing(command: &Path) -> Option<String> {
    let Ok(metadata) = std::fs::metadata(command) else {
        return Some(format!(
            "the harness agent executable `{}` does not exist on this host",
            command.display()
        ));
    };
    if !metadata.is_file() {
        return Some(format!(
            "the harness agent path `{}` is not a file on this host",
            command.display()
        ));
    }
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt as _;
        if metadata.permissions().mode() & 0o111 == 0 {
            return Some(format!(
                "the harness agent `{}` exists on this host but is not executable",
                command.display()
            ));
        }
    }
    None
}

/// Whether `name` resolves to an executable file on this process's own PATH —
/// the environment the managed worker inherits — reported as the refusal
/// sentence when it does not.
fn resolves_nowhere_on_path(name: &str) -> Option<String> {
    let path = std::env::var_os("PATH")?;
    let resolves =
        std::env::split_paths(&path).any(|dir| executable_missing(&dir.join(name)).is_none());
    if resolves {
        None
    } else {
        Some(format!(
            "the harness names no binary and `{name}` resolves nowhere on this server's PATH"
        ))
    }
}