kernex_runtime/

lib.rs

#![cfg_attr(test, allow(clippy::unwrap_used, clippy::expect_used))]

//! kernex-runtime: The facade crate that composes all Kernex components.
//!
//! Provides `Runtime` for configuring and running an AI agent runtime
//! with sandboxed execution, multi-provider support, persistent memory,
//! skills, and multi-agent pipeline orchestration.
//!
//! # Quick Start
//!
//! ```rust,ignore
//! use kernex_runtime::RuntimeBuilder;
//! use kernex_core::traits::Provider;
//! use kernex_core::message::Request;
//! use kernex_providers::ollama::OllamaProvider;
//!
//! #[tokio::main]
//! async fn main() -> anyhow::Result<()> {
//!     let runtime = RuntimeBuilder::new()
//!         .data_dir("~/.my-agent")
//!         .build()
//!         .await?;
//!
//!     let provider = OllamaProvider::from_config(
//!         "http://localhost:11434".into(),
//!         "llama3.2".into(),
//!         None,
//!     )?;
//!
//!     let request = Request::text("user-1", "Hello!");
//!     let response = runtime.complete(&provider, &request).await?;
//!     println!("{}", response.text);
//!
//!     Ok(())
//! }
//! ```

pub use kernex_adapter_core::{Adapter, AdapterError, AdapterId, AdapterRegistry, Capability};

#[cfg(feature = "opentelemetry")]
pub mod telemetry;

#[cfg(feature = "sqlite-store")]
use kernex_core::config::MemoryConfig;
use kernex_core::context::{CompactionStrategy, Context, ContextNeeds};
use kernex_core::error::KernexError;
use kernex_core::guardrails::{GuardrailAction, GuardrailRunner};
use kernex_core::hooks::{HookRunner, NoopHookRunner};
use kernex_core::message::{CompletionMeta, Request, Response};
use kernex_core::permissions::PermissionRules;
use kernex_core::run::{RunConfig, RunOutcome};
use kernex_core::stream::StreamEvent;
use kernex_core::traits::Provider;
use kernex_core::traits::StreamingProvider;
use kernex_core::traits::Summarizer;
#[cfg(feature = "sqlite-store")]
use kernex_memory::{Store, UsageBreakdown};
use kernex_skills::{
    build_skill_prompt, match_skill_toolboxes, match_skill_triggers, Project, Skill,
};
use std::sync::Arc;

/// Re-export sub-crates for convenience.
pub use kernex_core as core;
#[cfg(feature = "sqlite-store")]
pub use kernex_memory as memory;
pub use kernex_pipelines as pipelines;
pub use kernex_providers as providers;
pub use kernex_sandbox as sandbox;
pub use kernex_skills as skills;

/// A configured Kernex runtime with all subsystems initialized.
pub struct Runtime {
    /// Persistent memory store.
    #[cfg(feature = "sqlite-store")]
    pub store: Store,
    /// Loaded skills from the data directory.
    pub skills: Vec<Skill>,
    /// Loaded projects from the data directory.
    pub projects: Vec<Project>,
    /// Data directory path (expanded).
    pub data_dir: String,
    /// Base system prompt prepended to every request.
    pub system_prompt: String,
    /// Communication channel identifier (e.g. "cli", "api", "slack").
    pub channel: String,
    /// Active project key for scoping memory and lessons.
    pub project: Option<String>,
    /// Hook runner for tool lifecycle events.
    pub hook_runner: Arc<dyn HookRunner>,
    /// Declarative allow/deny rules applied before each tool call.
    pub permission_rules: Option<Arc<PermissionRules>>,
    /// Optional guardrail applied to input before provider call and output after.
    pub guardrail_runner: Option<Arc<dyn GuardrailRunner>>,
    /// When true, conversations whose history exceeds `max_context_messages`
    /// have their overflow summarized via the active provider instead of
    /// silently dropped. See [`RuntimeBuilder::auto_compact`].
    pub auto_compact: bool,
}

/// Adapter that lets `Store::build_context` reuse the active provider as a
/// summarizer when [`Runtime::auto_compact`] is enabled.
///
/// Wraps a `&dyn Provider` and answers [`Summarizer::summarize`] by sending a
/// fixed instruction prompt through the same provider that is handling the
/// turn. Costs one extra round-trip per overflow event (not per turn). The
/// summarizer is constructed per-call inside the runtime; it does not persist
/// state across requests, so there's no risk of summary drift.
struct ProviderSummarizer<'a> {
    provider: &'a dyn Provider,
}

#[async_trait::async_trait]
impl Summarizer for ProviderSummarizer<'_> {
    async fn summarize(&self, text: &str) -> Result<String, KernexError> {
        // Tight, role-stable prompt. The model never sees the original system
        // prompt (we deliberately call with an empty one) so its output is
        // a pure summary, not another agent reply.
        let instruction = format!(
            "You are a conversation summarizer. Summarize the following \
             exchange in 200 words or fewer. Focus on: decisions made, files \
             touched, errors encountered, and unresolved questions. Skip \
             greetings and small talk. Output the summary only — no preamble.\n\n\
             ---\n{text}\n---"
        );
        let mut ctx = Context::new(&instruction);
        ctx.system_prompt.clear();
        let response = self.provider.complete(&ctx).await?;
        Ok(response.text)
    }
}

impl Runtime {
    /// Return an `Arc<dyn MemoryStore>` over this runtime's composed
    /// memory store. Downstream consumers (a binary CLI, an HTTP API, an
    /// MCP shim) can call this to share the runtime's connection pool
    /// rather than opening a second SQLite connection against the same
    /// database file.
    ///
    /// `Store` already implements `Clone` (its `SqlitePool` is internally
    /// reference-counted), so cloning here shares the same pool.
    #[cfg(feature = "sqlite-store")]
    pub fn store_handle(&self) -> Arc<dyn kernex_memory::MemoryStore> {
        Arc::new(self.store.clone())
    }

    /// Send a request through the full runtime pipeline:
    /// build context from memory → enrich with skills → complete via provider → save exchange.
    ///
    /// This is the high-level convenience method that wires together all
    /// Kernex subsystems in a single call.
    pub async fn complete(
        &self,
        provider: &dyn Provider,
        request: &Request,
    ) -> Result<Response, KernexError> {
        self.complete_with_needs(provider, request, &ContextNeeds::default())
            .await
    }

    /// Like [`complete`](Self::complete), but with explicit control over which
    /// context blocks are loaded from memory.
    #[tracing::instrument(
        name = "kernex.complete",
        skip_all,
        fields(provider = provider.name(), sender = %request.sender_id)
    )]
    pub async fn complete_with_needs(
        &self,
        provider: &dyn Provider,
        request: &Request,
        #[allow(unused_variables)] needs: &ContextNeeds,
    ) -> Result<Response, KernexError> {
        let project_ref = self.project.as_deref();

        // Input guardrail: check (and optionally sanitize) the request text
        // before it reaches the provider or is stored in memory.
        let owned_req;
        let request = if let Some(gr) = &self.guardrail_runner {
            match gr.check_input(&request.text).await {
                GuardrailAction::Allow => request,
                GuardrailAction::Block(reason) => return Err(KernexError::Guardrail(reason)),
                GuardrailAction::Sanitize(clean) => {
                    owned_req = Request {
                        text: clean,
                        ..request.clone()
                    };
                    &owned_req
                }
            }
        } else {
            request
        };

        // Build skill context (prompt block + optional model override).
        let skill_ctx = build_skill_prompt(&self.skills);
        let full_system_prompt = if skill_ctx.prompt.is_empty() {
            self.system_prompt.clone()
        } else if self.system_prompt.is_empty() {
            skill_ctx.prompt.clone()
        } else {
            format!("{}\n\n{}", self.system_prompt, skill_ctx.prompt)
        };

        // Build context from memory (history, recall, facts, lessons, etc).
        #[cfg(feature = "sqlite-store")]
        let mut context = {
            let (effective_needs, summarizer): (
                std::borrow::Cow<'_, ContextNeeds>,
                Option<ProviderSummarizer<'_>>,
            ) = if self.auto_compact {
                let mut owned = needs.clone();
                owned.compact = CompactionStrategy::Summarize;
                (
                    std::borrow::Cow::Owned(owned),
                    Some(ProviderSummarizer { provider }),
                )
            } else {
                (std::borrow::Cow::Borrowed(needs), None)
            };
            self.store
                .build_context(
                    &self.channel,
                    request,
                    &full_system_prompt,
                    &effective_needs,
                    project_ref,
                    summarizer.as_ref().map(|s| s as &dyn Summarizer),
                )
                .await?
        };

        #[cfg(not(feature = "sqlite-store"))]
        let mut context = {
            let mut ctx = kernex_core::context::Context::new(&request.text);
            ctx.system_prompt = full_system_prompt;
            ctx
        };

        // Apply skill model override when no model was already set on context.
        if context.model.is_none() {
            context.model = skill_ctx.model;
        }

        // Enrich context with triggered MCP servers.
        let mcp_servers = match_skill_triggers(&self.skills, &request.text);
        if !mcp_servers.is_empty() {
            context.mcp_servers = mcp_servers;
        }

        // Enrich context with triggered toolboxes.
        let toolboxes = match_skill_toolboxes(&self.skills, &request.text);
        if !toolboxes.is_empty() {
            context.toolboxes = toolboxes;
        }

        // Wire hooks and permission rules into context.
        context.hook_runner = Some(self.hook_runner.clone());
        context.permission_rules = self.permission_rules.clone();

        // Send to provider.
        let raw_response = provider.complete(&context).await?;

        // Output guardrail: check (and optionally sanitize) the response text.
        let response = if let Some(gr) = &self.guardrail_runner {
            match gr.check_output(&raw_response.text).await {
                GuardrailAction::Allow => raw_response,
                GuardrailAction::Block(reason) => return Err(KernexError::Guardrail(reason)),
                GuardrailAction::Sanitize(clean) => Response {
                    text: clean,
                    metadata: raw_response.metadata,
                },
            }
        } else {
            raw_response
        };

        // Persist exchange in memory.
        #[allow(unused_variables)]
        let project_key = project_ref.unwrap_or("default");

        #[cfg(feature = "sqlite-store")]
        self.store
            .store_exchange(&self.channel, request, &response, project_key)
            .await?;

        // Record token usage if the provider reported a count.
        #[cfg(feature = "sqlite-store")]
        if let Some(tokens) = response.metadata.tokens_used {
            let model = response.metadata.model.as_deref().unwrap_or("unknown");
            let session = response.metadata.session_id.as_deref().unwrap_or("default");
            let breakdown = UsageBreakdown {
                input_tokens: response.metadata.input_tokens,
                output_tokens: response.metadata.output_tokens,
                cache_read_tokens: response.metadata.cache_read_tokens,
                cache_creation_tokens: response.metadata.cache_creation_tokens,
            };
            if let Err(e) = self
                .store
                .record_usage_full(&request.sender_id, session, tokens, model, breakdown)
                .await
            {
                tracing::warn!("failed to record token usage: {e}");
            }
        }

        Ok(response)
    }

    /// Stream a request through the runtime pipeline, returning events as they arrive.
    ///
    /// Builds context from memory, enriches with skills, opens a streaming connection
    /// to the provider, and persists the exchange to memory after the stream completes.
    /// Returns a channel receiver that yields [`StreamEvent`]s until `Done` or `Error`.
    pub async fn complete_stream(
        &self,
        provider: &dyn StreamingProvider,
        request: &Request,
    ) -> Result<tokio::sync::mpsc::Receiver<StreamEvent>, KernexError> {
        self.complete_stream_with_needs(provider, request, &ContextNeeds::default())
            .await
    }

    /// Like [`complete_stream`](Self::complete_stream), but with explicit control over which
    /// context blocks are loaded from memory.
    #[tracing::instrument(
        name = "kernex.stream",
        skip_all,
        fields(provider = provider.name(), sender = %request.sender_id)
    )]
    pub async fn complete_stream_with_needs(
        &self,
        provider: &dyn StreamingProvider,
        request: &Request,
        #[allow(unused_variables)] needs: &ContextNeeds,
    ) -> Result<tokio::sync::mpsc::Receiver<StreamEvent>, KernexError> {
        let project_ref = self.project.as_deref();

        // Input guardrail: check (and optionally sanitize) the request text
        // before the stream is opened. Block returns early; Sanitize clones the request.
        let owned_req;
        let request = if let Some(gr) = &self.guardrail_runner {
            match gr.check_input(&request.text).await {
                GuardrailAction::Allow => request,
                GuardrailAction::Block(reason) => return Err(KernexError::Guardrail(reason)),
                GuardrailAction::Sanitize(clean) => {
                    owned_req = Request {
                        text: clean,
                        ..request.clone()
                    };
                    &owned_req
                }
            }
        } else {
            request
        };

        let skill_ctx = build_skill_prompt(&self.skills);
        let full_system_prompt = if skill_ctx.prompt.is_empty() {
            self.system_prompt.clone()
        } else if self.system_prompt.is_empty() {
            skill_ctx.prompt.clone()
        } else {
            format!("{}\n\n{}", self.system_prompt, skill_ctx.prompt)
        };

        #[cfg(feature = "sqlite-store")]
        let mut context = {
            let (effective_needs, summarizer): (
                std::borrow::Cow<'_, ContextNeeds>,
                Option<ProviderSummarizer<'_>>,
            ) = if self.auto_compact {
                let mut owned = needs.clone();
                owned.compact = CompactionStrategy::Summarize;
                (
                    std::borrow::Cow::Owned(owned),
                    Some(ProviderSummarizer { provider }),
                )
            } else {
                (std::borrow::Cow::Borrowed(needs), None)
            };
            self.store
                .build_context(
                    &self.channel,
                    request,
                    &full_system_prompt,
                    &effective_needs,
                    project_ref,
                    summarizer.as_ref().map(|s| s as &dyn Summarizer),
                )
                .await?
        };

        #[cfg(not(feature = "sqlite-store"))]
        let mut context = {
            let mut ctx = kernex_core::context::Context::new(&request.text);
            ctx.system_prompt = full_system_prompt;
            ctx
        };

        if context.model.is_none() {
            context.model = skill_ctx.model;
        }

        let mcp_servers = match_skill_triggers(&self.skills, &request.text);
        if !mcp_servers.is_empty() {
            context.mcp_servers = mcp_servers;
        }
        let toolboxes = match_skill_toolboxes(&self.skills, &request.text);
        if !toolboxes.is_empty() {
            context.toolboxes = toolboxes;
        }

        context.hook_runner = Some(self.hook_runner.clone());
        context.permission_rules = self.permission_rules.clone();

        // Open streaming connection to provider.
        let provider_name = provider.name().to_string();
        let mut upstream = provider.complete_stream(&context).await?;

        // Forwarding channel returned to the caller.
        let (tx, rx) = tokio::sync::mpsc::channel::<StreamEvent>(64);

        // Background task: forward events and persist exchange when done.
        #[cfg(feature = "sqlite-store")]
        let store = self.store.clone();
        let channel = self.channel.clone();
        let request_clone = request.clone();
        #[allow(unused_variables)]
        let project_key = project_ref.unwrap_or("default").to_string();
        let guardrail_runner = self.guardrail_runner.clone();

        tokio::spawn(async move {
            use kernex_core::stream::{StreamAccumulator, StreamEvent as SE};
            let mut acc = StreamAccumulator::new();
            let started = std::time::Instant::now();

            while let Some(event) = upstream.recv().await {
                acc.push(&event);
                let is_terminal = matches!(event, SE::Done | SE::Error(_));
                // Best-effort forward; drop silently if receiver was dropped.
                let _ = tx.send(event).await;
                if is_terminal {
                    break;
                }
            }

            // Persist accumulated exchange to memory.
            // Output guardrail runs on the full accumulated text before storage.
            // The stream has already been forwarded to the caller so the guardrail
            // only affects what is persisted — it does not modify the streamed tokens.
            #[cfg(feature = "sqlite-store")]
            {
                let elapsed_ms = started.elapsed().as_millis() as u64;
                let tokens_used = acc.total_tokens();
                let stop_reason = acc.usage().and_then(|u| u.stop_reason.clone());
                let accumulated = acc.into_text();
                let persisted_text = if let Some(gr) = &guardrail_runner {
                    match gr.check_output(&accumulated).await {
                        GuardrailAction::Allow => accumulated,
                        GuardrailAction::Block(_) => String::new(),
                        GuardrailAction::Sanitize(clean) => clean,
                    }
                } else {
                    accumulated
                };
                let response = Response {
                    text: persisted_text,
                    metadata: CompletionMeta {
                        provider_used: provider_name,
                        tokens_used,
                        processing_time_ms: elapsed_ms,
                        model: None,
                        session_id: None,
                        stop_reason,
                        ..Default::default()
                    },
                };
                if let Err(e) = store
                    .store_exchange(&channel, &request_clone, &response, &project_key)
                    .await
                {
                    tracing::warn!("failed to persist streaming exchange: {e}");
                }
            }
            #[cfg(not(feature = "sqlite-store"))]
            {
                let _ = acc;
                let _ = started;
                let _ = provider_name;
                let _ = guardrail_runner;
            }
        });

        Ok(rx)
    }

    /// Run the agent with explicit lifecycle control.
    ///
    /// Sets `max_turns` in context so the provider's agentic loop respects it,
    /// wires the runtime hook runner, calls the provider, fires the `on_stop`
    /// hook, and wraps the outcome in [`RunOutcome`].
    #[tracing::instrument(
        name = "kernex.run",
        skip_all,
        fields(provider = provider.name(), sender = %request.sender_id, turns = config.max_turns)
    )]
    pub async fn run(
        &self,
        provider: &dyn Provider,
        request: &Request,
        config: &RunConfig,
    ) -> Result<RunOutcome, KernexError> {
        let needs = ContextNeeds::default();
        let project_ref = self.project.as_deref();

        // Input guardrail.
        let owned_req;
        let request = if let Some(gr) = &self.guardrail_runner {
            match gr.check_input(&request.text).await {
                GuardrailAction::Allow => request,
                GuardrailAction::Block(reason) => return Err(KernexError::Guardrail(reason)),
                GuardrailAction::Sanitize(clean) => {
                    owned_req = Request {
                        text: clean,
                        ..request.clone()
                    };
                    &owned_req
                }
            }
        } else {
            request
        };

        let skill_ctx = build_skill_prompt(&self.skills);
        let full_system_prompt = if skill_ctx.prompt.is_empty() {
            self.system_prompt.clone()
        } else if self.system_prompt.is_empty() {
            skill_ctx.prompt.clone()
        } else {
            format!("{}\n\n{}", self.system_prompt, skill_ctx.prompt)
        };

        #[cfg(feature = "sqlite-store")]
        let mut context = {
            let (effective_needs, summarizer): (
                std::borrow::Cow<'_, ContextNeeds>,
                Option<ProviderSummarizer<'_>>,
            ) = if self.auto_compact {
                let mut owned = needs.clone();
                owned.compact = CompactionStrategy::Summarize;
                (
                    std::borrow::Cow::Owned(owned),
                    Some(ProviderSummarizer { provider }),
                )
            } else {
                (std::borrow::Cow::Borrowed(&needs), None)
            };
            self.store
                .build_context(
                    &self.channel,
                    request,
                    &full_system_prompt,
                    &effective_needs,
                    project_ref,
                    summarizer.as_ref().map(|s| s as &dyn Summarizer),
                )
                .await?
        };

        #[cfg(not(feature = "sqlite-store"))]
        let mut context = {
            let mut ctx = kernex_core::context::Context::new(&request.text);
            ctx.system_prompt = full_system_prompt;
            ctx
        };

        // Apply skill model override when no model was already set on context.
        if context.model.is_none() {
            context.model = skill_ctx.model;
        }

        let mcp_servers = match_skill_triggers(&self.skills, &request.text);
        if !mcp_servers.is_empty() {
            context.mcp_servers = mcp_servers;
        }
        let toolboxes = match_skill_toolboxes(&self.skills, &request.text);
        if !toolboxes.is_empty() {
            context.toolboxes = toolboxes;
        }

        // Set max_turns, hooks, and permission rules.
        context.max_turns = Some(config.max_turns);
        context.hook_runner = Some(self.hook_runner.clone());
        context.permission_rules = self.permission_rules.clone();

        let raw_response = provider.complete(&context).await?;

        // Output guardrail.
        let response = if let Some(gr) = &self.guardrail_runner {
            match gr.check_output(&raw_response.text).await {
                GuardrailAction::Allow => raw_response,
                GuardrailAction::Block(reason) => return Err(KernexError::Guardrail(reason)),
                GuardrailAction::Sanitize(clean) => Response {
                    text: clean,
                    metadata: raw_response.metadata,
                },
            }
        } else {
            raw_response
        };

        // Record token usage if the provider reported a count. Tokens are
        // billed even when the run exhausts its turn budget, so this runs
        // before the max-turns branch below.
        #[cfg(feature = "sqlite-store")]
        if let Some(tokens) = response.metadata.tokens_used {
            let model = response.metadata.model.as_deref().unwrap_or("unknown");
            let session = response.metadata.session_id.as_deref().unwrap_or("default");
            let breakdown = UsageBreakdown {
                input_tokens: response.metadata.input_tokens,
                output_tokens: response.metadata.output_tokens,
                cache_read_tokens: response.metadata.cache_read_tokens,
                cache_creation_tokens: response.metadata.cache_creation_tokens,
            };
            if let Err(e) = self
                .store
                .record_usage_full(&request.sender_id, session, tokens, model, breakdown)
                .await
            {
                tracing::warn!("failed to record token usage: {e}");
            }
        }

        // Turn-budget exhaustion: the provider hit `max_turns` without a final
        // answer. Surface it as `RunOutcome::MaxTurns` rather than persisting an
        // empty/synthetic exchange or firing `on_stop` with a non-answer.
        if response.metadata.stop_reason.as_deref() == Some("max_turns") {
            return Ok(RunOutcome::MaxTurns);
        }

        // Fire on_stop hook.
        self.hook_runner.on_stop(&response.text).await;

        // Persist exchange.
        #[allow(unused_variables)]
        let project_key = project_ref.unwrap_or("default");
        #[cfg(feature = "sqlite-store")]
        self.store
            .store_exchange(&self.channel, request, &response, project_key)
            .await?;

        Ok(RunOutcome::EndTurn(response))
    }
}

/// Builder for constructing a `Runtime` with the desired configuration.
pub struct RuntimeBuilder {
    data_dir: String,
    #[cfg(feature = "sqlite-store")]
    db_path: Option<String>,
    system_prompt: String,
    channel: String,
    project: Option<String>,
    hook_runner: Option<Arc<dyn HookRunner>>,
    permission_rules: Option<Arc<PermissionRules>>,
    guardrail_runner: Option<Arc<dyn GuardrailRunner>>,
    auto_compact: bool,
}

impl RuntimeBuilder {
    /// Create a new builder with default settings.
    pub fn new() -> Self {
        Self {
            data_dir: "~/.kernex".to_string(),
            #[cfg(feature = "sqlite-store")]
            db_path: None,
            system_prompt: String::new(),
            channel: "cli".to_string(),
            project: None,
            hook_runner: None,
            permission_rules: None,
            guardrail_runner: None,
            // Default off for backward compatibility with v0.4.0 callers.
            // kernex-agent flips it on; future major versions may default it on.
            auto_compact: false,
        }
    }

    /// Create a new builder pre-populated from a declarative agent definition file.
    ///
    /// The file format is detected by extension: `.yaml` / `.yml` use YAML
    /// (requires the `yaml` feature on `kernex-core`); all other extensions
    /// use TOML. Missing files silently fall back to defaults.
    ///
    /// Maps `[runtime]` fields (`data_dir`, `system_prompt`, `channel`,
    /// `project`) and `[memory]` → `db_path` into the builder. Provider
    /// selection is left to the caller.
    ///
    /// # Example (agent.toml)
    ///
    /// ```toml
    /// [runtime]
    /// name       = "my-agent"
    /// data_dir   = "~/.my-agent"
    /// channel    = "api"
    /// project    = "acme"
    /// system_prompt = "You are a helpful coding assistant."
    ///
    /// [memory]
    /// db_path = "~/.my-agent/memory.db"
    /// ```
    pub fn from_file(path: &str) -> Result<Self, kernex_core::error::KernexError> {
        let config = kernex_core::config::load_file(path)?;
        Ok(Self::from_config(&config))
    }

    /// Populate a builder from a pre-parsed [`KernexConfig`].
    ///
    /// Maps `runtime.{data_dir, system_prompt, channel, project}` and
    /// `memory.db_path` into builder fields. Provider selection is left
    /// to the caller.
    ///
    /// [`KernexConfig`]: kernex_core::config::KernexConfig
    pub fn from_config(config: &kernex_core::config::KernexConfig) -> Self {
        let mut builder = Self::new()
            .data_dir(&config.runtime.data_dir)
            .system_prompt(&config.runtime.system_prompt)
            .channel(&config.runtime.channel);

        if let Some(proj) = &config.runtime.project {
            builder = builder.project(proj);
        }

        #[cfg(feature = "sqlite-store")]
        {
            builder = builder.db_path(&config.memory.db_path);
        }

        builder
    }

    /// Create a new builder configured from environment variables.
    ///
    /// Recognizes:
    /// - `KERNEX_DATA_DIR`
    /// - `KERNEX_DB_PATH` (when `sqlite-store` feature is enabled)
    /// - `KERNEX_SYSTEM_PROMPT`
    /// - `KERNEX_CHANNEL`
    /// - `KERNEX_PROJECT`
    pub fn from_env() -> Self {
        let mut builder = Self::new();

        if let Ok(dir) = std::env::var("KERNEX_DATA_DIR") {
            warn_if_data_dir_unusual(&dir);
            builder = builder.data_dir(&dir);
        }
        #[cfg(feature = "sqlite-store")]
        if let Ok(path) = std::env::var("KERNEX_DB_PATH") {
            builder = builder.db_path(&path);
        }
        if let Ok(prompt) = std::env::var("KERNEX_SYSTEM_PROMPT") {
            builder = builder.system_prompt(&prompt);
        }
        if let Ok(channel) = std::env::var("KERNEX_CHANNEL") {
            builder = builder.channel(&channel);
        }
        if let Ok(project) = std::env::var("KERNEX_PROJECT") {
            builder = builder.project(&project);
        }

        builder
    }

    /// Set the data directory (default: `~/.kernex`).
    pub fn data_dir(mut self, path: &str) -> Self {
        self.data_dir = path.to_string();
        self
    }

    /// Set a custom database path (default: `{data_dir}/memory.db`).
    #[cfg(feature = "sqlite-store")]
    pub fn db_path(mut self, path: &str) -> Self {
        self.db_path = Some(path.to_string());
        self
    }

    /// Set the base system prompt.
    pub fn system_prompt(mut self, prompt: &str) -> Self {
        self.system_prompt = prompt.to_string();
        self
    }

    /// Set the channel identifier (default: `"cli"`).
    pub fn channel(mut self, channel: &str) -> Self {
        self.channel = channel.to_string();
        self
    }

    /// Set the active project for scoping memory.
    pub fn project(mut self, project: &str) -> Self {
        self.project = Some(project.to_string());
        self
    }

    /// Set a hook runner for tool lifecycle events.
    pub fn hook_runner(mut self, runner: Arc<dyn HookRunner>) -> Self {
        self.hook_runner = Some(runner);
        self
    }

    /// Set declarative allow/deny permission rules for tool calls.
    pub fn permission_rules(mut self, rules: PermissionRules) -> Self {
        self.permission_rules = Some(Arc::new(rules));
        self
    }

    /// Set a guardrail runner that intercepts and filters input/output text.
    pub fn guardrail_runner(mut self, runner: Arc<dyn GuardrailRunner>) -> Self {
        self.guardrail_runner = Some(runner);
        self
    }

    /// Enable automatic context compaction on long conversations.
    ///
    /// When enabled, every call into the runtime that builds context (e.g.
    /// [`Runtime::complete`], [`Runtime::complete_stream`], [`Runtime::run`])
    /// will, on overflow past `max_context_messages`, summarize the dropped
    /// rows via the active provider and prepend the summary to the system
    /// prompt under `[Earlier conversation summary]`. The summarization
    /// adds one extra provider round-trip per overflow event (not per turn),
    /// uses a fixed instruction prompt that never reveals the agent's own
    /// system prompt, and falls back to the default Drop behavior if the
    /// provider call fails.
    ///
    /// Default is **off** to preserve v0.4.0 behavior. Recommended for any
    /// long-running interactive session; the default exists only so existing
    /// callers do not silently change billing characteristics.
    pub fn auto_compact(mut self, enable: bool) -> Self {
        self.auto_compact = enable;
        self
    }

    /// Build and initialize the runtime.
    pub async fn build(self) -> Result<Runtime, KernexError> {
        let expanded_dir = kernex_core::shellexpand(&self.data_dir);

        // Ensure data directory exists.
        tokio::fs::create_dir_all(&expanded_dir)
            .await
            .map_err(|e| KernexError::Config(format!("failed to create data dir: {e}")))?;

        // Initialize store.
        #[cfg(feature = "sqlite-store")]
        let store = {
            let db_path = self
                .db_path
                .unwrap_or_else(|| format!("{expanded_dir}/memory.db"));
            let mem_config = MemoryConfig {
                db_path: db_path.clone(),
                ..Default::default()
            };
            Store::new(&mem_config).await?
        };

        // Load skills and projects. These functions use synchronous std::fs
        // internally; offload to a blocking thread so we do not stall the
        // tokio executor on cold start (especially relevant for projects with
        // large skills/ trees).
        let skills_data_dir = self.data_dir.clone();
        let skills =
            tokio::task::spawn_blocking(move || kernex_skills::load_skills(&skills_data_dir))
                .await
                .map_err(|e| {
                    KernexError::skill(kernex_skills::SkillError::Logic(format!(
                        "load_skills task failed: {e}"
                    )))
                })?;
        let projects_data_dir = self.data_dir.clone();
        let projects =
            tokio::task::spawn_blocking(move || kernex_skills::load_projects(&projects_data_dir))
                .await
                .map_err(|e| {
                    KernexError::skill(kernex_skills::SkillError::Logic(format!(
                        "load_projects task failed: {e}"
                    )))
                })?;

        tracing::info!(
            "runtime initialized: {} skills, {} projects",
            skills.len(),
            projects.len()
        );

        let hook_runner: Arc<dyn HookRunner> =
            self.hook_runner.unwrap_or_else(|| Arc::new(NoopHookRunner));

        Ok(Runtime {
            #[cfg(feature = "sqlite-store")]
            store,
            skills,
            projects,
            data_dir: expanded_dir,
            system_prompt: self.system_prompt,
            channel: self.channel,
            project: self.project,
            hook_runner,
            permission_rules: self.permission_rules,
            guardrail_runner: self.guardrail_runner,
            auto_compact: self.auto_compact,
        })
    }
}

impl Default for RuntimeBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// Emit a warning when `KERNEX_DATA_DIR` resolves to a path outside the
/// usual locations. Misconfigured env on a shared host (e.g.
/// `KERNEX_DATA_DIR=/etc`) means writes happen in places the operator
/// likely didn't intend; the sandbox's blocklist policy still allows
/// writes outside the configured data dir, so the agent could end up
/// writing `/etc/cron.d/` before anyone notices.
fn warn_if_data_dir_unusual(dir: &str) {
    // Only act on absolute paths; relative paths are project-scoped and
    // resolve under cwd, which is always operator-chosen.
    let path = std::path::Path::new(dir);
    if !path.is_absolute() {
        return;
    }
    let s = dir;
    let in_home = std::env::var("HOME")
        .ok()
        .map(|h| !h.is_empty() && s.starts_with(&h))
        .unwrap_or(false);
    let usual = in_home
        || s.starts_with("/tmp/")
        || s.starts_with("/var/")
        || s.starts_with("/Users/")
        || s.starts_with("/home/")
        || s == "/tmp"
        || s == "/var";
    if !usual {
        tracing::warn!(
            data_dir = %dir,
            "KERNEX_DATA_DIR resolves outside $HOME / /tmp / /var — \
             writes may land in unexpected locations"
        );
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_runtime_builder_creates_runtime() {
        let tmp_dir = tempfile::TempDir::new().unwrap();
        let tmp = tmp_dir.path();

        let runtime = RuntimeBuilder::new()
            .data_dir(tmp.to_str().unwrap())
            .build()
            .await
            .unwrap();

        assert!(runtime.skills.is_empty());
        assert!(runtime.projects.is_empty());
        assert!(runtime.system_prompt.is_empty());
        assert_eq!(runtime.channel, "cli");
        assert!(runtime.project.is_none());
        assert!(std::path::Path::new(&runtime.data_dir).exists());
    }

    #[tokio::test]
    async fn test_runtime_builder_custom_db_path() {
        let tmp_dir = tempfile::TempDir::new().unwrap();
        let tmp = tmp_dir.path();

        let db = tmp.join("custom.db");
        let runtime = RuntimeBuilder::new()
            .data_dir(tmp.to_str().unwrap())
            .db_path(db.to_str().unwrap())
            .build()
            .await
            .unwrap();

        assert!(db.exists());
        drop(runtime);
    }

    #[tokio::test]
    async fn test_runtime_builder_with_config() {
        let tmp_dir = tempfile::TempDir::new().unwrap();
        let tmp = tmp_dir.path();

        let runtime = RuntimeBuilder::new()
            .data_dir(tmp.to_str().unwrap())
            .system_prompt("You are helpful.")
            .channel("api")
            .project("my-project")
            .build()
            .await
            .unwrap();

        assert_eq!(runtime.system_prompt, "You are helpful.");
        assert_eq!(runtime.channel, "api");
        assert_eq!(runtime.project, Some("my-project".to_string()));
    }

    #[tokio::test]
    async fn test_runtime_builder_from_config() {
        use kernex_core::config::{KernexConfig, MemoryConfig, RuntimeConfig};

        let tmp_dir = tempfile::TempDir::new().unwrap();
        let tmp = tmp_dir.path();

        // Override `memory.db_path` so the test does not share
        // `~/.kernex/data/memory.db` with sibling tests; that shared file
        // races on migration replay across parallel runs (#duplicate-column
        // / #UNIQUE-constraint regressions seen in CI).
        let cfg = KernexConfig {
            runtime: RuntimeConfig {
                name: "test-agent".to_string(),
                data_dir: tmp.to_str().unwrap().to_string(),
                channel: "slack".to_string(),
                project: Some("my-proj".to_string()),
                system_prompt: "Be concise.".to_string(),
                ..RuntimeConfig::default()
            },
            memory: MemoryConfig {
                db_path: tmp.join("memory.db").to_str().unwrap().to_string(),
                ..MemoryConfig::default()
            },
            ..KernexConfig::default()
        };

        let runtime = RuntimeBuilder::from_config(&cfg).build().await.unwrap();

        assert_eq!(runtime.channel, "slack");
        assert_eq!(runtime.project, Some("my-proj".to_string()));
        assert_eq!(runtime.system_prompt, "Be concise.");
    }

    #[tokio::test]
    async fn test_runtime_builder_from_file_toml() {
        use std::io::Write;

        let tmp_dir = tempfile::TempDir::new().unwrap();
        let tmp = tmp_dir.path();
        let escaped = tmp.to_str().unwrap().replace('\\', "\\\\");

        // Pin both `data_dir` and `[memory] db_path` inside the TempDir so
        // this test does not race against the shared `~/.kernex/data/memory.db`
        // default that other parallel tests hit.
        let cfg_path = tmp.join("agent.toml");
        let mut f = std::fs::File::create(&cfg_path).unwrap();
        writeln!(
            f,
            r#"[runtime]
name = "file-agent"
data_dir = "{escaped}"
channel = "api"
project = "file-proj"
system_prompt = "From file."

[memory]
db_path = "{escaped}/memory.db"
"#
        )
        .unwrap();

        let runtime = RuntimeBuilder::from_file(cfg_path.to_str().unwrap())
            .unwrap()
            .build()
            .await
            .unwrap();

        assert_eq!(runtime.channel, "api");
        assert_eq!(runtime.project, Some("file-proj".to_string()));
        assert_eq!(runtime.system_prompt, "From file.");
    }

    /// Mock provider returning a `Response` with a fixed `stop_reason`, to test
    /// how `Runtime::run` maps the provider's stop reason to a `RunOutcome`.
    struct StopReasonProvider(Option<&'static str>);

    #[async_trait::async_trait]
    impl Provider for StopReasonProvider {
        fn name(&self) -> &str {
            "mock"
        }
        fn requires_api_key(&self) -> bool {
            false
        }
        async fn complete(&self, _ctx: &Context) -> Result<Response, KernexError> {
            Ok(Response {
                text: "partial".to_string(),
                metadata: CompletionMeta {
                    stop_reason: self.0.map(|s| s.to_string()),
                    ..Default::default()
                },
            })
        }
        async fn is_available(&self) -> bool {
            true
        }
    }

    async fn run_with_stop_reason(stop: Option<&'static str>) -> RunOutcome {
        let tmp_dir = tempfile::TempDir::new().unwrap();
        let runtime = RuntimeBuilder::new()
            .data_dir(tmp_dir.path().to_str().unwrap())
            .build()
            .await
            .unwrap();
        runtime
            .run(
                &StopReasonProvider(stop),
                &Request::text("user-1", "hi"),
                &RunConfig::default(),
            )
            .await
            .unwrap()
    }

    #[tokio::test]
    async fn run_maps_max_turns_stop_reason_to_max_turns_outcome() {
        let outcome = run_with_stop_reason(Some("max_turns")).await;
        assert!(
            matches!(outcome, RunOutcome::MaxTurns),
            "max_turns stop_reason should yield RunOutcome::MaxTurns, got {outcome:?}"
        );
    }

    #[tokio::test]
    async fn run_returns_end_turn_for_normal_stop_reason() {
        match run_with_stop_reason(Some("end_turn")).await {
            RunOutcome::EndTurn(resp) => assert_eq!(resp.text, "partial"),
            other => panic!("expected EndTurn, got {other:?}"),
        }
    }
}