agent-runtime-bootstrap 0.1.0

//! Production composition root and lifecycle management for Agent Runtime.
//!
//! This crate wires validated deployment configuration to infrastructure
//! adapters, execution coordination, the Kernel, and HTTP transport. Agent
//! implementations remain outside this dependency boundary.

use std::{collections::BTreeMap, sync::Arc, time::Duration};

use agent_runtime_config::{AuthConfig, JournalConfig, RuntimeConfig};
use async_trait::async_trait;
use axum::Router;
use runtime_adapter_agent_infra::{
    AgentInfraConfig as InfraAdapterConfig, AgentInfraSessionFactory, RuntimeBinding,
    StaticRuntimeBindingResolver,
};
use runtime_adapter_execution_journal::MemoryExecutionJournal;
use runtime_agent_definition_loader::{DefinitionCatalogResolver, load_definition_report};
use runtime_extension_api::AgentDefinition;
use runtime_kernel::AgentKernel;
use runtime_kernel_api::{AgentDefinitionResolver, ContextCompactionLimits, ContextLimits};
use runtime_link::ExecutionCoordinator;
use runtime_outer_http::{
    IngressAuthenticator, ReadinessProbe, RuntimeTransportConfig, StaticTokenAuthenticator,
    TrustedGatewayAuthenticator,
};
use runtime_types::{CallerScope, CredentialHandle, RuntimeInstanceId, WorkspaceId};
use thiserror::Error;

#[derive(Clone)]
pub struct RuntimeApplication {
    pub router: Router,
    readiness: Arc<RuntimeReadiness>,
    coordinator: Arc<ExecutionCoordinator>,
}

impl RuntimeApplication {
    pub fn begin_shutdown(&self) {
        self.readiness
            .accepting
            .store(false, std::sync::atomic::Ordering::SeqCst);
        self.coordinator.begin_shutdown();
    }

    pub async fn shutdown(&self, grace: Duration) -> bool {
        self.begin_shutdown();
        self.coordinator.shutdown(grace).await
    }

    pub fn active_executions(&self) -> usize {
        self.coordinator.active_executions()
    }
}

#[derive(Debug, Error)]
pub enum BootstrapError {
    #[error("invalid runtime wiring: {0}")]
    Invalid(String),
    #[error("execution journal failed: {0}")]
    Journal(String),
}

pub struct RuntimeBootstrap<'a> {
    config: &'a RuntimeConfig,
    definition: Option<Arc<AgentDefinition>>,
    transport: Option<RuntimeTransportConfig>,
}

impl<'a> RuntimeBootstrap<'a> {
    /// Create the official composition root from typed deployment config.
    ///
    /// Without an injected Definition, Bootstrap loads `definition_path`.
    /// There is intentionally no hidden Code Agent default: Tool/Skill-focused
    /// distributions call [`Self::with_definition`] so behavior is visible in
    /// their composition root.
    pub fn new(config: &'a RuntimeConfig) -> Self {
        Self {
            config,
            definition: None,
            transport: None,
        }
    }

    /// Inject one already-built Definition.
    ///
    /// This is the narrow path for Tool + Skill and Skill + Workspace agents:
    /// Bootstrap still owns Infra SDK sessions, execution, authentication and
    /// transport, while developer code owns only the Definition.
    pub fn with_definition(mut self, definition: Arc<AgentDefinition>) -> Self {
        self.definition = Some(definition);
        self
    }

    /// Override the route selection derived from `server.transport`.
    pub fn with_transport(mut self, transport: RuntimeTransportConfig) -> Self {
        self.transport = Some(transport);
        self
    }

    pub async fn build(self) -> Result<RuntimeApplication, BootstrapError> {
        build_application(self.config, self.definition, self.transport).await
    }
}

pub async fn build(settings: &RuntimeConfig) -> Result<RuntimeApplication, BootstrapError> {
    RuntimeBootstrap::new(settings).build().await
}

async fn build_application(
    settings: &RuntimeConfig,
    injected_definition: Option<Arc<AgentDefinition>>,
    transport_override: Option<RuntimeTransportConfig>,
) -> Result<RuntimeApplication, BootstrapError> {
    settings
        .validate()
        .map_err(|error| BootstrapError::Invalid(error.to_string()))?;
    let journal = match &settings.journal {
        JournalConfig::Memory {
            max_events_per_execution,
        } => Arc::new(MemoryExecutionJournal::new(*max_events_per_execution)),
    };
    let sessions = Arc::new(
        AgentInfraSessionFactory::new(InfraAdapterConfig {
            gateway_url: settings.agent_infra.gateway_url.clone(),
            request_timeout: Duration::from_millis(settings.agent_infra.request_timeout_ms),
            max_response_bytes: settings.agent_infra.max_response_bytes,
            trusted_mesh_http: settings.agent_infra.trusted_mesh_http,
            workload_subject: settings.agent_infra.workload_subject.clone(),
            workload_credential: CredentialHandle::new(
                settings.agent_infra.workload_token.expose(),
            )
            .map_err(|error| BootstrapError::Invalid(error.to_string()))?,
            delegation_ttl: Duration::from_secs(settings.agent_infra.delegation_ttl_seconds),
        })
        .map_err(|error| BootstrapError::Invalid(error.to_string()))?,
    );
    let runtime_id = RuntimeInstanceId::new(settings.binding.runtime_id.clone())
        .map_err(|error| BootstrapError::Invalid(error.to_string()))?;
    let (definition_resolver, definition_digest, runtime_type, definition_id, definition_version): (
        Arc<dyn AgentDefinitionResolver>,
        String,
        String,
        String,
        String,
    ) = match (injected_definition, &settings.binding.definition_path) {
        (Some(definition), _) => {
            let catalog = DefinitionCatalogResolver::new([definition.clone()])
                .map_err(|error| BootstrapError::Invalid(error.to_string()))?;
            (
                Arc::new(catalog),
                definition.manifest.digest.clone(),
                definition.manifest.runtime_type.clone(),
                definition.manifest.id.to_string(),
                definition.manifest.version.clone(),
            )
        }
        (None, Some(path)) => {
            let report = load_definition_report(std::path::Path::new(path))
                .map_err(|error| BootstrapError::Invalid(error.to_string()))?;
            let definition = report.definition;
            let digest = definition.manifest.digest.clone();
            let runtime_type = definition.manifest.runtime_type.clone();
            let definition_id = definition.manifest.id.to_string();
            let definition_version = definition.manifest.version.clone();
            let catalog = DefinitionCatalogResolver::new([definition])
                .map_err(|error| BootstrapError::Invalid(error.to_string()))?;
            (
                Arc::new(catalog),
                digest,
                runtime_type,
                definition_id,
                definition_version,
            )
        }
        (None, None) => return Err(BootstrapError::Invalid(
            "an Agent Definition must be injected with RuntimeBootstrap::with_definition or selected by binding.definition_path".into(),
        )),
    };
    let binding = RuntimeBinding {
        agent_id: settings.binding.agent_id.clone(),
        runtime_type,
        model: settings.binding.model.clone(),
        workspace_id: WorkspaceId::new(settings.binding.workspace_id.clone())
            .map_err(|error| BootstrapError::Invalid(error.to_string()))?,
        definition_id,
        definition_version,
        definition_digest: Some(definition_digest),
        metadata: BTreeMap::new(),
    };
    let resolver = StaticRuntimeBindingResolver::new(BTreeMap::from([(runtime_id, binding)]))
        .map_err(|error| BootstrapError::Invalid(error.to_string()))?;
    let link = Arc::new(
        ExecutionCoordinator::new_with_context_limits(
            journal,
            Arc::new(resolver),
            sessions.clone(),
            definition_resolver,
            Arc::new(AgentKernel),
            settings.scheduler.max_concurrent_executions,
            ContextLimits {
                max_messages: settings.context.max_messages,
                max_input_tokens: settings.context.max_input_tokens,
                reserved_output_tokens: settings.context.reserved_output_tokens,
                protocol_overhead_tokens: settings.context.protocol_overhead_tokens,
                auto_compact: settings.context.auto_compact,
                compaction: ContextCompactionLimits {
                    trigger_percent: settings.context.compaction.trigger_percent,
                    target_percent: settings.context.compaction.target_percent,
                    emergency_percent: settings.context.compaction.emergency_percent,
                    min_reclaim_percent: settings.context.compaction.min_reclaim_percent,
                    min_turns_between: settings.context.compaction.min_turns_between,
                    checkpoint_max_chars: settings.context.compaction.checkpoint_max_chars,
                },
            },
        )
        .map_err(|error| BootstrapError::Invalid(error.to_string()))?,
    );
    link.recover(1_000)
        .await
        .map_err(|error| BootstrapError::Journal(error.to_string()))?;
    link.start_recovery_loop(Duration::from_secs(5), 1_000)
        .map_err(|error| BootstrapError::Invalid(error.to_string()))?;
    let authenticator: Arc<dyn IngressAuthenticator> = match &settings.auth {
        AuthConfig::LocalStatic {
            token,
            subject,
            tenant_id,
            project_id,
            capabilities,
        } => Arc::new(
            StaticTokenAuthenticator::new(
                token.expose(),
                CallerScope {
                    subject: subject.clone(),
                    tenant_id: tenant_id.clone(),
                    project_id: project_id.clone(),
                    capabilities: capabilities.clone(),
                },
            )
            .map_err(|error| BootstrapError::Invalid(error.message))?,
        ),
        AuthConfig::TrustedGateway { shared_secret } => Arc::new(
            TrustedGatewayAuthenticator::new(shared_secret.expose())
                .map_err(|error| BootstrapError::Invalid(error.message))?,
        ),
    };
    let readiness = Arc::new(RuntimeReadiness {
        sessions,
        accepting: std::sync::atomic::AtomicBool::new(true),
    });
    let configured_transport = RuntimeTransportConfig {
        agent_tasks: settings.server.transport.agent_tasks,
        openai_responses: settings.server.transport.openai_responses,
        claude_messages: settings.server.transport.claude_messages,
        sse: settings.server.transport.sse,
        websocket: settings.server.transport.websocket,
        websocket_max_message_bytes: settings.server.transport.websocket_max_message_bytes,
    };
    let router = runtime_outer_http::router(
        link.clone(),
        authenticator,
        readiness.clone(),
        transport_override.unwrap_or(configured_transport),
    );
    Ok(RuntimeApplication {
        router,
        readiness,
        coordinator: link,
    })
}

struct RuntimeReadiness {
    sessions: Arc<AgentInfraSessionFactory>,
    accepting: std::sync::atomic::AtomicBool,
}

#[async_trait]
impl ReadinessProbe for RuntimeReadiness {
    async fn ready(&self) -> Result<(), String> {
        if !self.accepting.load(std::sync::atomic::Ordering::SeqCst) {
            return Err("runtime is shutting down".into());
        }
        self.sessions
            .check_ready()
            .await
            .map_err(|error| error.to_string())
    }
}