liminal-server 0.25.4

//! One startup pull and one shared owner for explicit administrative signals.
//!
//! The startup pull runs behind the listening door (SRV-041 R4): a start
//! never waits on Lys, and a pull Lys refuses never stops the server.

use std::sync::{Arc, Weak};

use super::{LiminalConnectionServices, ServerError};
use crate::auth_lys::drafts::admin::{AdminDrafts, AdminLimits, AdminRequest, AdminResponse};
use crate::auth_lys::drafts::{
    AgentDraftStore, AgentProjection, AtomicDraftStore, DraftDecisionHook, DraftError, DraftPuller,
    PullReport, PullSignal, TargetDraftStore,
};
use crate::health::endpoint::admin::DraftEndpoint;

struct Projection(Weak<LiminalConnectionServices>);

impl AgentProjection for Projection {
    fn apply(&self, draft: &lys_pass::drafts::ApprovedDraft) -> Result<(), DraftError> {
        self.0
            .upgrade()
            .ok_or(DraftError::Store)?
            .execute_agent_draft(draft)
            .map(|_| ())
            .map_err(|_| DraftError::Store)
    }
}

pub struct ApprovalRuntime {
    route: tokio::sync::Mutex<AdminDrafts<TargetDraftStore>>,
    services: Weak<LiminalConnectionServices>,
    connector: zeroize::Zeroizing<String>,
    handle: tokio::runtime::Handle,
    limits: AdminLimits,
}

impl std::fmt::Debug for ApprovalRuntime {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ApprovalRuntime").finish_non_exhaustive()
    }
}

fn clock() -> Result<u64, DraftError> {
    let elapsed = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_err(|_| DraftError::Contract)?;
    u64::try_from(elapsed.as_millis()).map_err(|_| DraftError::Bound)
}

impl ApprovalRuntime {
    async fn signal(&self, signal: PullSignal) -> Result<PullReport, DraftError> {
        self.route
            .lock()
            .await
            .signal(&self.connector, signal, clock()?)
            .await
    }

    fn block<T: Send>(
        &self,
        future: impl Future<Output = Result<T, DraftError>> + Send,
    ) -> Result<T, ServerError> {
        std::thread::scope(|scope| scope.spawn(|| self.handle.block_on(future)).join())
            .map_err(|_| failure("draft pull worker failed"))?
            .map_err(|error| failure(&error.to_string()))
    }
}

impl DraftEndpoint for ApprovalRuntime {
    fn limits(&self) -> AdminLimits {
        self.limits
    }

    fn handle(&self, request: AdminRequest<'_>) -> Result<AdminResponse, ServerError> {
        if request.path == crate::auth_lys::drafts::admin::CREATE_PATH {
            return self.create(&request);
        }
        self.block(async {
            self.route
                .lock()
                .await
                .handle(request, &self.connector)
                .await
        })
    }
}

#[derive(serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct CreateBody {
    operation: String,
    name: String,
    spec: crate::agents::records::AgentSpec,
}

impl ApprovalRuntime {
    /// A person's outright `create_agent`: the same executor an approved draft
    /// reaches, with the words liminal prepares from the verified pass and the
    /// effective outright grant as the evidence; nothing is read from a header
    /// or the body as authority.
    fn create(&self, request: &AdminRequest<'_>) -> Result<AdminResponse, ServerError> {
        let bad = |name: &str| AdminResponse {
            status: 400,
            body: serde_json::json!({ "refusal": name }),
        };
        if request.method != "POST" {
            return Ok(bad("agent_create_method_refused"));
        }
        let Ok(body) = serde_json::from_slice::<CreateBody>(request.body) else {
            return Ok(bad("agent_create_body_invalid"));
        };
        let Ok(target) = lys_pass::Target::new("liminal.agent", &body.name, "create_agent") else {
            return Ok(bad("agent_create_name_invalid"));
        };
        let admitted = self.block(async {
            let asked = self
                .route
                .lock()
                .await
                .outright(request.authorization, request.at_ms)?;
            match asked {
                Ok(ask) => ask.check(&target, &self.connector).await,
                Err(refusal) => Ok(Err(refusal)),
            }
        })?;
        let (pass, grant) = match admitted {
            Ok(admitted) => admitted,
            Err(refusal) => return Ok(refusal),
        };
        let act = crate::agents::records::AgentAct::prepare(
            body.name.clone(),
            crate::agents::records::AgentEvent::Create(body.spec),
            pass.verified(),
            request.at_ms,
        );
        let act = match act {
            Ok(act) => act,
            Err(error) => return Ok(named(409, &error.to_string())),
        };
        let words = serde_json::to_string(&act).map_err(|_| failure("agent create words"))?;
        let draft = lys_pass::drafts::ApprovedDraft {
            id: body.operation,
            app: "liminal".to_owned(),
            grant,
            target,
            request_digest: lys_pass::drafts::request_digest(&words),
            approved_by: vec![pass.verified().claims().holder.id.clone()],
            mode: lys_pass::Mode::Outright,
            words,
        };
        let services = self
            .services
            .upgrade()
            .ok_or_else(|| failure("server stopped"))?;
        let owners = services
            .approval_agents
            .clone()
            .ok_or_else(|| failure("agent approval owners missing"))?;
        // The registry's writes land through the draft journal, which admits
        // exactly the act it has begun: the same order a pulled draft takes.
        let approval = crate::auth_lys::drafts::ApprovalTransition {
            draft: draft.id.clone(),
            digest: draft.request_digest.clone(),
            approved_by: draft.approved_by.clone(),
            executed_at_ms: request.at_ms,
        };
        if let Err(error) = owners.journal().begin(&draft, &approval) {
            return Ok(named(409, &error.to_string()));
        }
        // Stamped with the journal's own time, as a pulled draft's execution is.
        let executed = owners
            .agents()
            .registry()
            .execute_approved_at(&draft, request.at_ms)
            .map_err(ServerError::from);
        owners
            .journal()
            .finish()
            .map_err(|error| failure(&error.to_string()))?;
        let receipt = match executed {
            Ok(receipt) => receipt,
            Err(error) => return Ok(named(409, &error.to_string())),
        };
        // An outright act has no Lys draft to report back to: its outcome is
        // acknowledged here, so no later pull asks Lys to close a draft it never held.
        owners
            .journal()
            .update(|snapshot| {
                if let Some(outcome) = snapshot.outcomes.get_mut(&draft.id) {
                    outcome.acknowledged = true;
                }
                Ok(())
            })
            .map_err(|error| failure(&error.to_string()))?;
        services.agent_created(&draft.target.id)?;
        Ok(AdminResponse {
            status: 200,
            body: serde_json::to_value(receipt).map_err(|_| failure("agent create receipt"))?,
        })
    }
}

/// A refusal answered by name, so the caller reads why instead of a closed connection.
fn named(status: u16, reason: &str) -> AdminResponse {
    AdminResponse {
        status,
        body: serde_json::json!({ "refusal": "agent_create_refused", "reason": reason }),
    }
}

#[async_trait::async_trait]
impl DraftDecisionHook for ApprovalRuntime {
    async fn draft_decided(&self) -> Result<PullReport, DraftError> {
        self.signal(PullSignal::Decision).await
    }
}

impl LiminalConnectionServices {
    pub(in crate::server) fn draft_decision_hook(&self) -> Option<Arc<dyn DraftDecisionHook>> {
        self.draft_hook.get().and_then(Weak::upgrade)
    }

    pub(in crate::server) fn install_approval_runtime(
        self: &Arc<Self>,
        lys: &crate::config::types::LysConfig,
        lifecycle: Arc<crate::services::lifecycle::Lifecycle>,
        http: Arc<dyn crate::services::lys_http::LysHttp>,
        credential: String,
        handle: tokio::runtime::Handle,
        health: &crate::health::HealthServerHandle,
    ) -> Result<Box<dyn FnOnce() + Send>, ServerError> {
        let owners = self
            .approval_agents
            .as_ref()
            .ok_or_else(|| failure("construction requires configured agent approval owners"))?;
        let delivery = self
            .agent_delivery
            .as_ref()
            .ok_or_else(|| failure("agent outcome owner missing"))?;
        let limits = lys
            .drafts
            .ok_or_else(|| failure("auth.lys.drafts is required"))?;
        let holder = Arc::new(crate::auth_lys::prepared::SiteHolder {
            lifecycle: lifecycle.clone(),
            owners: owners.clone(),
            config: lys.clone(),
            handle: handle.clone(),
        });
        self.host_config
            .set((lys.clone(), handle.clone()))
            .map_err(|_| failure("host configuration already installed"))?;
        if lys.construction.is_some() {
            self.install_construction(lifecycle.clone(), lys, http, holder)
                .map_err(|error| failure(&error.to_string()))?;
        } else {
            self.site_lifecycle
                .set(lifecycle.clone())
                .map_err(|_| failure("lifecycle already installed"))?;
        }
        for record in owners.agents().registry().records()? {
            if record.active() {
                self.agent_created(&record.name)?;
            }
        }
        let agents = AgentDraftStore::new(owners, delivery.clone())
            .with_projection(Arc::new(Projection(Arc::downgrade(self))));
        let sites = crate::services::lifecycle::site::executor::SiteDraftStore::new(
            lifecycle,
            owners.journal().clone(),
            delivery.clone(),
            0,
        );
        let store = TargetDraftStore { agents, sites };
        let puller = DraftPuller::new(
            lys.api_url.as_deref().unwrap_or(&lys.issuer),
            store,
            limits.pull,
        )
        .map_err(|error| failure(&error.to_string()))?;
        let keys = std::fs::read_to_string(&lys.keys_path)
            .map_err(|_| failure("draft keyset unreadable"))?;
        let adapter = Arc::new(
            crate::auth_lys::lys_pass_adapter::LysPassAdapter::from_keys_json(
                &keys,
                lys.issuer.clone(),
            )
            .map_err(|error| failure(error.name()))?,
        );
        let route = AdminDrafts::new(puller, adapter, lys.network_id.clone(), limits.http)
            .map_err(|error| failure(&error.to_string()))?
            .asking_at(lys.api_url.clone());
        let runtime = Arc::new(ApprovalRuntime {
            route: tokio::sync::Mutex::new(route),
            services: Arc::downgrade(self),
            connector: zeroize::Zeroizing::new(credential),
            handle,
            limits: limits.http,
        });
        let hook: Arc<dyn DraftDecisionHook> = runtime.clone();
        self.draft_hook
            .set(Arc::downgrade(&hook))
            .map_err(|_| failure("draft decision hook already installed"))?;
        health.install_drafts(runtime.clone())?;
        // SRV-041 R4: the caller begins this once the listener binds.
        Ok(Box::new(move || {
            pull_at_start(&runtime.route, &runtime.connector, &runtime.handle);
        }))
    }
}

/// Pulls the drafts Lys approved while this exchange was down, on the Lys
/// ask runtime behind the route's lock, so a console or decision pull that
/// comes first is answered before it and one that comes during waits. A
/// failure is logged by its cause and the start goes on serving: the drafts
/// wait in Lys until acknowledged, and the next signal pulls them. Answers
/// whether the pull completed.
pub(super) fn pull_at_start<S: AtomicDraftStore>(
    route: &tokio::sync::Mutex<AdminDrafts<S>>,
    connector: &str,
    handle: &tokio::runtime::Handle,
) -> bool {
    let started = std::time::Instant::now();
    let pulled = handle.block_on(async {
        let at_ms = clock()?;
        route
            .lock()
            .await
            .signal(connector, PullSignal::Startup, at_ms)
            .await
    });
    crate::server::boot_steps::finished("draft_startup_pull", started);
    match pulled {
        Ok(report) => {
            tracing::info!(?report, "drafts pulled at start behind the listening door");
            true
        }
        Err(error) => {
            tracing::error!(
                %error,
                "DRAFT_STARTUP_PULL_FAILED: the server serves; the next console or decision signal pulls"
            );
            false
        }
    }
}

fn failure(message: &str) -> ServerError {
    ServerError::ConfigValidation {
        message: message.into(),
    }
}

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