roboticus_api/api/routes/

mod.rs

mod admin;
mod agent;
mod channels;
mod cron;
mod health;
mod interview;
pub(crate) mod mcp;
mod memory;
mod sessions;
mod skills;
pub(crate) mod subagent_integrity;
mod themes;
pub(crate) use self::agent::execute_scheduled_agent_task;
mod observability;
mod subagents;
mod traces;

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;

use axum::extract::DefaultBodyLimit;
use axum::response::IntoResponse;
use axum::{
    Router, middleware,
    routing::{get, post, put},
};
use tokio::sync::RwLock;

use crate::config_runtime::ConfigApplyStatus;
use roboticus_agent::policy::PolicyEngine;
use roboticus_agent::subagents::SubagentRegistry;
use roboticus_browser::Browser;
use roboticus_channels::a2a::A2aProtocol;
use roboticus_channels::router::ChannelRouter;
use roboticus_channels::telegram::TelegramAdapter;
use roboticus_channels::whatsapp::WhatsAppAdapter;
use roboticus_core::RoboticusConfig;
// personality types accessed via roboticus_core::personality:: in InterviewSession
use roboticus_db::Database;
use roboticus_llm::LlmService;
use roboticus_llm::OAuthManager;
use roboticus_llm::semantic_classifier::SemanticClassifier;
use roboticus_plugin_sdk::registry::PluginRegistry;
use roboticus_wallet::WalletService;

use roboticus_agent::approvals::ApprovalManager;
use roboticus_agent::capability::CapabilityRegistry;
use roboticus_agent::obsidian::ObsidianVault;
use roboticus_agent::tools::ToolRegistry;
use roboticus_channels::discord::DiscordAdapter;
use roboticus_channels::email::EmailAdapter;
use roboticus_channels::media::MediaService;
use roboticus_channels::signal::SignalAdapter;
use roboticus_channels::voice::VoicePipeline;

use crate::ws::EventBus;

// ── RFC 9457 Problem Details error response ─────────────────
//
// All API errors are serialized as RFC 9457 "Problem Details for HTTP
// APIs" (application/problem+json). The `JsonError` type is the single
// chokepoint — all 250+ error sites construct one, and the `IntoResponse`
// impl serializes it.

/// RFC 9457 Problem Details response.
///
/// Wire format:
/// ```json
/// {
///   "type": "about:blank",
///   "title": "Bad Request",
///   "status": 400,
///   "detail": "api_key cannot be empty"
/// }
/// ```
///
/// Constructors (`bad_request()`, `not_found()`, `internal_err()`) are
/// unchanged — call sites do not need modification.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ProblemDetails {
    /// URI reference identifying the problem type. Defaults to `"about:blank"`
    /// per RFC 9457 §4.2.1 when no specific type URI is defined.
    #[serde(rename = "type")]
    pub problem_type: String,
    /// Short, human-readable summary (typically the HTTP reason phrase).
    pub title: String,
    /// HTTP status code (duplicated in body per RFC 9457 §4.2.4).
    pub status: u16,
    /// Human-readable explanation specific to this occurrence.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub detail: Option<String>,
    /// URI reference identifying this specific occurrence (optional).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub instance: Option<String>,
}

/// The API error type. Wraps a status code and message, serialized as
/// RFC 9457 Problem Details on the wire.
#[derive(Debug)]
pub(crate) struct JsonError(pub axum::http::StatusCode, pub String);

impl JsonError {
    /// Build the RFC 9457 Problem Details body.
    fn to_problem_details(&self) -> ProblemDetails {
        ProblemDetails {
            problem_type: "about:blank".into(),
            title: canonical_reason(self.0).into(),
            status: self.0.as_u16(),
            detail: if self.1.is_empty() {
                None
            } else {
                Some(self.1.clone())
            },
            instance: None,
        }
    }
}

/// Map a status code to its canonical reason phrase.
fn canonical_reason(status: axum::http::StatusCode) -> &'static str {
    status.canonical_reason().unwrap_or("Error")
}

impl axum::response::IntoResponse for JsonError {
    fn into_response(self) -> axum::response::Response {
        let status = self.0;
        let body = self.to_problem_details();
        (
            status,
            [(axum::http::header::CONTENT_TYPE, "application/problem+json")],
            axum::Json(body),
        )
            .into_response()
    }
}

impl From<(axum::http::StatusCode, String)> for JsonError {
    fn from((status, msg): (axum::http::StatusCode, String)) -> Self {
        Self(status, msg)
    }
}

/// Shorthand for a 400 Bad Request JSON error.
pub(crate) fn bad_request(msg: impl std::fmt::Display) -> JsonError {
    JsonError(axum::http::StatusCode::BAD_REQUEST, msg.to_string())
}

/// Shorthand for a 404 Not Found JSON error.
pub(crate) fn not_found(msg: impl std::fmt::Display) -> JsonError {
    JsonError(axum::http::StatusCode::NOT_FOUND, msg.to_string())
}

/// Build an RFC 9457 Problem Details JSON value for handlers that return
/// `(StatusCode, Json<Value>)` instead of `JsonError`.
///
/// Prefer using `JsonError` where possible. This helper is for response
/// paths that return a tuple directly (e.g. channel webhooks that also
/// return `{"ok": false}`).
pub(crate) fn problem_json(status: axum::http::StatusCode, detail: &str) -> serde_json::Value {
    serde_json::json!({
        "type": "about:blank",
        "title": canonical_reason(status),
        "status": status.as_u16(),
        "detail": detail,
    })
}

/// Build an RFC 9457 response tuple with `application/problem+json` content type.
pub(crate) fn problem_response(
    status: axum::http::StatusCode,
    detail: &str,
) -> axum::response::Response {
    (
        status,
        [(axum::http::header::CONTENT_TYPE, "application/problem+json")],
        axum::Json(problem_json(status, detail)),
    )
        .into_response()
}

// ── Helpers (used by submodules) ──────────────────────────────

/// Sanitizes error messages before returning to clients (strip paths, internal details, cap length).
///
/// LIMITATIONS: This is a best-effort filter that strips known wrapper
/// prefixes and truncates. It does NOT guarantee that internal details
/// (file paths, SQL fragments, stack traces) are fully redacted. If a new
/// error source leaks sensitive info, add its prefix to the stripping list
/// below or, better, ensure the call site maps the error before it reaches
/// this function.
pub(crate) fn sanitize_error_message(msg: &str) -> String {
    let sanitized = msg.lines().next().unwrap_or(msg);

    let sanitized = sanitized
        .trim_start_matches("Database(\"")
        .trim_end_matches("\")")
        .trim_start_matches("Wallet(\"")
        .trim_end_matches("\")");

    // Strip content after common internal-detail prefixes that may leak
    // implementation specifics (connection strings, file paths, etc.).
    let sensitive_prefixes = [
        "at /", // stack trace file paths
        "called `Result::unwrap()` on an `Err` value:",
        "SQLITE_",                // raw SQLite error codes
        "Connection refused",     // infra details
        "constraint failed",      // SQLite constraint errors (leaks table/column names)
        "no such table",          // SQLite schema details
        "no such column",         // SQLite schema details
        "UNIQUE constraint",      // SQLite constraint (leaks table.column)
        "FOREIGN KEY constraint", // SQLite constraint
        "NOT NULL constraint",    // SQLite constraint
    ];
    let sanitized = {
        let mut s = sanitized.to_string();
        for prefix in &sensitive_prefixes {
            if let Some(pos) = s.find(prefix) {
                s.truncate(pos);
                s.push_str("[details redacted]");
                break;
            }
        }
        s
    };

    if sanitized.len() > 200 {
        let boundary = sanitized
            .char_indices()
            .map(|(i, _)| i)
            .take_while(|&i| i <= 200)
            .last()
            .unwrap_or(0);
        format!("{}...", &sanitized[..boundary])
    } else {
        sanitized
    }
}

/// Logs the full error and returns a JSON 500 error for API responses.
pub(crate) fn internal_err(e: &impl std::fmt::Display) -> JsonError {
    tracing::error!(error = %e, "request failed");
    JsonError(
        axum::http::StatusCode::INTERNAL_SERVER_ERROR,
        sanitize_error_message(&e.to_string()),
    )
}

// ── Input validation helpers ──────────────────────────────────

/// Maximum allowed length for short identifier fields (agent_id, name, etc.).
const MAX_SHORT_FIELD: usize = 256;
/// Maximum allowed length for long text fields (description, content, etc.).
const MAX_LONG_FIELD: usize = 4096;

/// Validate a user-supplied string field: reject empty/whitespace-only, null bytes, and enforce length.
pub(crate) fn validate_field(
    field_name: &str,
    value: &str,
    max_len: usize,
) -> Result<(), JsonError> {
    if value.trim().is_empty() {
        return Err(bad_request(format!("{field_name} must not be empty")));
    }
    if value.contains('\0') {
        return Err(bad_request(format!(
            "{field_name} must not contain null bytes"
        )));
    }
    if value.len() > max_len {
        return Err(bad_request(format!(
            "{field_name} exceeds max length ({max_len})"
        )));
    }
    Ok(())
}

/// Validate a short identifier field (agent_id, name, session_id, etc.).
pub(crate) fn validate_short(field_name: &str, value: &str) -> Result<(), JsonError> {
    validate_field(field_name, value, MAX_SHORT_FIELD)
}

/// Validate a long text field (description, content, etc.).
pub(crate) fn validate_long(field_name: &str, value: &str) -> Result<(), JsonError> {
    validate_field(field_name, value, MAX_LONG_FIELD)
}

/// Strip HTML tags from a string to prevent injection in stored values.
pub(crate) fn sanitize_html(input: &str) -> String {
    input
        .replace('&', "&amp;")
        .replace('<', "&lt;")
        .replace('>', "&gt;")
        .replace('"', "&quot;")
        .replace('\'', "&#x27;")
}

// ── Pagination helpers ──────────────────────────────────────────

/// Default maximum items per page for list endpoints.
const DEFAULT_PAGE_SIZE: i64 = 200;
/// Absolute maximum items per page (prevents memory abuse via huge limits).
const MAX_PAGE_SIZE: i64 = 500;

/// Shared pagination query parameters for list endpoints.
#[derive(Debug, serde::Deserialize)]
pub(crate) struct PaginationQuery {
    pub limit: Option<i64>,
    pub offset: Option<i64>,
}

impl PaginationQuery {
    /// Returns (limit, offset) clamped to safe ranges.
    pub fn resolve(&self) -> (i64, i64) {
        let limit = self
            .limit
            .unwrap_or(DEFAULT_PAGE_SIZE)
            .clamp(1, MAX_PAGE_SIZE);
        let offset = self.offset.unwrap_or(0).max(0);
        (limit, offset)
    }
}

// ── Shared state and types ────────────────────────────────────

pub use roboticus_pipeline::personality::PersonalityState;

/// Tracks a multi-turn personality interview for a single user.
#[derive(Debug)]
pub struct InterviewSession {
    pub history: Vec<roboticus_llm::format::UnifiedMessage>,
    pub awaiting_confirmation: bool,
    pub pending_output: Option<roboticus_core::personality::InterviewOutput>,
    pub created_at: std::time::Instant,
}

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

impl InterviewSession {
    pub fn new() -> Self {
        Self {
            history: vec![roboticus_llm::format::UnifiedMessage {
                role: "system".into(),
                content: roboticus_agent::interview::build_interview_prompt(),
                parts: None,
            }],
            awaiting_confirmation: false,
            pending_output: None,
            created_at: std::time::Instant::now(),
        }
    }
}

#[derive(Clone)]
pub struct AppState {
    pub db: Database,
    pub config: Arc<RwLock<RoboticusConfig>>,
    pub llm: Arc<RwLock<LlmService>>,
    pub wallet: Arc<WalletService>,
    pub a2a: Arc<RwLock<A2aProtocol>>,
    pub personality: Arc<RwLock<PersonalityState>>,
    pub hmac_secret: Arc<Vec<u8>>,
    pub interviews: Arc<RwLock<HashMap<String, InterviewSession>>>,
    pub plugins: Arc<PluginRegistry>,
    pub policy_engine: Arc<PolicyEngine>,
    pub browser: Arc<Browser>,
    pub registry: Arc<SubagentRegistry>,
    pub event_bus: EventBus,
    pub channel_router: Arc<ChannelRouter>,
    pub telegram: Option<Arc<TelegramAdapter>>,
    pub whatsapp: Option<Arc<WhatsAppAdapter>>,
    pub retriever: Arc<roboticus_agent::retrieval::MemoryRetriever>,
    pub ann_index: roboticus_db::ann::AnnIndex,
    pub tools: Arc<ToolRegistry>,
    /// Mirror of `tools` for LLM catalog and capability-aware execution paths.
    pub capabilities: Arc<CapabilityRegistry>,
    pub approvals: Arc<ApprovalManager>,
    pub discord: Option<Arc<DiscordAdapter>>,
    pub signal: Option<Arc<SignalAdapter>>,
    pub email: Option<Arc<EmailAdapter>>,
    pub voice: Option<Arc<RwLock<VoicePipeline>>>,
    pub media_service: Option<Arc<MediaService>>,
    pub discovery: Arc<RwLock<roboticus_agent::discovery::DiscoveryRegistry>>,
    pub devices: Arc<RwLock<roboticus_agent::device::DeviceManager>>,
    pub mcp_clients: Arc<RwLock<roboticus_agent::mcp::McpClientManager>>,
    pub mcp_server: Arc<RwLock<roboticus_agent::mcp::McpServerRegistry>>,
    pub live_mcp: Arc<roboticus_agent::mcp::manager::McpConnectionManager>,
    pub oauth: Arc<OAuthManager>,
    pub keystore: Arc<roboticus_core::keystore::Keystore>,
    pub obsidian: Option<Arc<RwLock<ObsidianVault>>>,
    pub started_at: std::time::Instant,
    pub config_path: Arc<PathBuf>,
    pub config_apply_status: Arc<RwLock<ConfigApplyStatus>>,
    pub pending_specialist_proposals: Arc<RwLock<HashMap<String, serde_json::Value>>>,
    pub ws_tickets: crate::ws_ticket::TicketStore,
    pub rate_limiter: crate::rate_limit::GlobalRateLimitLayer,
    /// Shared semantic classifier backed by the embedding client.
    /// Used for guard pre-computation and specialist workflow detection.
    pub semantic_classifier: Arc<SemanticClassifier>,
}

impl AppState {
    /// Rebuild `CapabilityRegistry` from the current `ToolRegistry` (e.g. after plugin hot-load).
    pub async fn resync_capabilities_from_tools(&self) {
        if let Err(e) = self
            .capabilities
            .sync_from_tool_registry(Arc::clone(&self.tools))
            .await
        {
            tracing::warn!(error = %e, "capability resync from tools reported errors");
        }
    }

    pub async fn reload_personality(&self) {
        let workspace = {
            let config = self.config.read().await;
            config.agent.workspace.clone()
        };
        let new_state = PersonalityState::from_workspace(&workspace);
        tracing::info!(
            personality = %new_state.identity.name,
            generated_by = %new_state.identity.generated_by,
            "Hot-reloaded personality from workspace"
        );
        *self.personality.write().await = new_state;
    }
}

// ── Capability trait implementations ─────────────────────────────────────
//
impl AppState {
    /// Construct the full pipeline dependency set from AppState.
    ///
    /// This is the canonical bridge from the API-side god object to the
    /// pipeline's trait-scoped deps. Connectors call this once per request.
    pub(crate) fn pipeline_deps(&self) -> roboticus_pipeline::stage_deps::PipelineDeps<'_> {
        roboticus_pipeline::stage_deps::PipelineDeps {
            core: self,
            security: self,
            reasoning: self,
            retrieval: self,
            coordination: self,
            personality: self,
            tooling: self,
            execution: self,
            tool_executor: self,
            inference_runner: self,
        }
    }
}

// These impls bridge AppState (the API-side god object) to the pipeline
// crate's narrow capability traits. Each stage's dep struct composes
// these traits, giving it compile-time-enforced access to only the
// capabilities it declares.

impl roboticus_pipeline::capabilities::PipelineCore for AppState {
    fn db(&self) -> &Database {
        &self.db
    }
    fn config(&self) -> &Arc<RwLock<RoboticusConfig>> {
        &self.config
    }
}

impl roboticus_pipeline::capabilities::PipelineSecurity for AppState {
    fn hmac_secret(&self) -> &[u8] {
        &self.hmac_secret
    }
}

impl roboticus_pipeline::capabilities::PipelineReasoning for AppState {
    fn llm(&self) -> &Arc<RwLock<LlmService>> {
        &self.llm
    }
    fn semantic_classifier(&self) -> &Arc<SemanticClassifier> {
        &self.semantic_classifier
    }
}

impl roboticus_pipeline::capabilities::PipelineRetrieval for AppState {
    fn retriever(&self) -> &Arc<roboticus_agent::retrieval::MemoryRetriever> {
        &self.retriever
    }
    fn ann_index(&self) -> &roboticus_db::ann::AnnIndex {
        &self.ann_index
    }
}

impl roboticus_pipeline::capabilities::PipelineCoordination for AppState {
    fn channel_router(&self) -> &Arc<ChannelRouter> {
        &self.channel_router
    }
    fn event_bus(&self) -> &roboticus_pipeline::event_bus::EventBus {
        &self.event_bus
    }
}

impl roboticus_pipeline::capabilities::PipelinePersonality for AppState {
    fn personality(&self) -> &Arc<RwLock<PersonalityState>> {
        &self.personality
    }
}

impl roboticus_pipeline::capabilities::PipelineTooling for AppState {
    fn tools(&self) -> &Arc<ToolRegistry> {
        &self.tools
    }
    fn capabilities(&self) -> &Arc<CapabilityRegistry> {
        &self.capabilities
    }
    fn policy_engine(&self) -> &Arc<PolicyEngine> {
        &self.policy_engine
    }
    fn approvals(&self) -> &Arc<ApprovalManager> {
        &self.approvals
    }
    fn registry(&self) -> &Arc<SubagentRegistry> {
        &self.registry
    }
    fn plugins(&self) -> &Arc<PluginRegistry> {
        &self.plugins
    }
    fn mcp_server(&self) -> &Arc<RwLock<roboticus_agent::mcp::McpServerRegistry>> {
        &self.mcp_server
    }
    fn pending_specialist_proposals(&self) -> &Arc<RwLock<HashMap<String, serde_json::Value>>> {
        &self.pending_specialist_proposals
    }
}

impl roboticus_pipeline::capabilities::PipelineExecution for AppState {
    fn keystore(&self) -> &Arc<roboticus_core::Keystore> {
        &self.keystore
    }
    fn oauth(&self) -> &Arc<OAuthManager> {
        &self.oauth
    }
    fn live_mcp(&self) -> &Arc<roboticus_agent::mcp::manager::McpConnectionManager> {
        &self.live_mcp
    }
    fn mcp_clients(&self) -> &Arc<RwLock<roboticus_agent::mcp::McpClientManager>> {
        &self.mcp_clients
    }
    fn browser(&self) -> &Arc<Browser> {
        &self.browser
    }
}

#[async_trait::async_trait]
impl roboticus_pipeline::tool_executor::ToolExecutor for AppState {
    async fn execute_tool(
        &self,
        tool_name: &str,
        params: &serde_json::Value,
        turn_id: &str,
        authority: roboticus_core::InputAuthority,
        channel: Option<&str>,
    ) -> Result<String, String> {
        agent::execute_tool_call(self, tool_name, params, turn_id, authority, channel).await
    }

    async fn execute_tool_detailed(
        &self,
        tool_name: &str,
        params: &serde_json::Value,
        turn_id: &str,
        authority: roboticus_core::InputAuthority,
        channel: Option<&str>,
    ) -> Result<roboticus_pipeline::tool_executor::ToolExecutionDetails, String> {
        let details =
            agent::execute_tool_call_detailed(self, tool_name, params, turn_id, authority, channel)
                .await?;
        Ok(roboticus_pipeline::tool_executor::ToolExecutionDetails {
            output: details.output,
            source: details.source,
        })
    }
}

#[async_trait::async_trait]
impl roboticus_pipeline::capabilities::InferenceRunner for AppState {
    async fn select_and_audit_model(
        &self,
        user_content: &str,
        session_id: &str,
        turn_id: &str,
        channel_label: &str,
        complexity: Option<&str>,
    ) -> String {
        let audit =
            agent::select_routed_model_with_audit(self, user_content, Some(session_id)).await;
        let selected = audit.selected_model.clone();
        agent::persist_model_selection_audit(
            self,
            self,
            turn_id,
            session_id,
            channel_label,
            complexity,
            user_content,
            &audit,
        )
        .await;
        selected
    }

    async fn infer_with_fallback(
        &self,
        request: &roboticus_llm::format::UnifiedRequest,
        initial_model: &str,
    ) -> Result<roboticus_pipeline::core_types::InferenceResult, String> {
        let result = agent::infer_with_fallback(self, request, initial_model).await?;
        Ok(roboticus_pipeline::core_types::InferenceResult {
            content: result.content,
            model: result.model,
            provider: result.provider,
            tokens_in: result.tokens_in,
            tokens_out: result.tokens_out,
            cost: result.cost,
            latency_ms: result.latency_ms,
            quality_score: result.quality_score,
            escalated: result.escalated,
        })
    }

    async fn infer_stream_with_fallback(
        &self,
        request: &roboticus_llm::format::UnifiedRequest,
        initial_model: &str,
    ) -> Result<roboticus_pipeline::core_types::StreamResolvedInference, String> {
        let ctx = agent::infer_stream_with_fallback(self, request, initial_model).await?;
        Ok(roboticus_pipeline::core_types::StreamResolvedInference {
            stream: ctx.stream,
            selected_model: ctx.selected_model,
            provider_prefix: ctx.provider_prefix,
            cost_in: ctx.cost_in,
            cost_out: ctx.cost_out,
        })
    }
}

// ── JSON error normalization middleware ────────────────────────
//
// BUG-006/014/016/017: axum returns plain-text bodies for its built-in
// rejections (JSON parse errors, wrong Content-Type, 405 Method Not
// Allowed). This middleware intercepts any non-JSON error response and
// wraps it in the standard `{"error":"..."}` format.

async fn json_error_layer(
    req: axum::extract::Request,
    next: middleware::Next,
) -> axum::response::Response {
    let response = next.run(req).await;
    let status = response.status();

    if !(status.is_client_error() || status.is_server_error()) {
        return response;
    }

    let content_type = response
        .headers()
        .get(axum::http::header::CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");
    // Already RFC 9457 or structured JSON — pass through
    if content_type.contains("application/problem+json")
        || content_type.contains("application/json")
    {
        return response;
    }

    let code = response.status();
    let (_parts, body) = response.into_parts();
    let bytes = match axum::body::to_bytes(body, 8192).await {
        Ok(b) => b,
        Err(e) => {
            tracing::warn!(error = %e, "failed to read response body for JSON wrapping");
            axum::body::Bytes::new()
        }
    };
    let original_text = String::from_utf8_lossy(&bytes);

    let error_msg = if original_text.trim().is_empty() {
        match code {
            axum::http::StatusCode::METHOD_NOT_ALLOWED => "method not allowed".to_string(),
            axum::http::StatusCode::NOT_FOUND => "not found".to_string(),
            axum::http::StatusCode::UNSUPPORTED_MEDIA_TYPE => {
                "unsupported content type: expected application/json".to_string()
            }
            other => other.to_string(),
        }
    } else {
        sanitize_error_message(original_text.trim())
    };

    let json_body = problem_json(code, &error_msg);
    let body_bytes = serde_json::to_vec(&json_body)
        .unwrap_or_else(|_| br#"{"type":"about:blank","title":"Internal Server Error","status":500,"detail":"internal error"}"#.to_vec());
    let mut resp = axum::response::Response::new(axum::body::Body::from(body_bytes));
    *resp.status_mut() = code;
    resp.headers_mut().insert(
        axum::http::header::CONTENT_TYPE,
        axum::http::HeaderValue::from_static("application/problem+json"),
    );
    resp
}

// ── Security headers ─────────────────────────────────────────────
// BUG-018: Content-Security-Policy
// BUG-019: X-Frame-Options

// CSP nonce handling: the dashboard handler generates a per-request nonce,
// injects it into inline <script> tags, and sets its own CSP header with
// 'nonce-<value>'.  This middleware only sets the default (non-inline) CSP
// for responses that do not already carry a Content-Security-Policy header
// (i.e. API endpoints, redirects, etc.).

async fn security_headers_layer(
    req: axum::extract::Request,
    next: middleware::Next,
) -> axum::response::Response {
    let mut response = next.run(req).await;
    let headers = response.headers_mut();

    // Only set the default CSP if the handler did not already set one
    // (the dashboard handler sets a nonce-based CSP).
    let csp_name = axum::http::header::HeaderName::from_static("content-security-policy");
    if !headers.contains_key(&csp_name) {
        headers.insert(
            csp_name,
            axum::http::HeaderValue::from_static(
                "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; connect-src 'self' ws: wss:; frame-ancestors 'none'",
            ),
        );
    }
    headers.insert(
        axum::http::header::X_FRAME_OPTIONS,
        axum::http::HeaderValue::from_static("DENY"),
    );
    headers.insert(
        axum::http::header::X_CONTENT_TYPE_OPTIONS,
        axum::http::HeaderValue::from_static("nosniff"),
    );
    response
}

async fn dashboard_redirect() -> axum::response::Redirect {
    axum::response::Redirect::permanent("/")
}

// ── Router ──────────────────────────────────────────────────────

pub fn build_router(state: AppState) -> Router {
    use admin::{
        a2a_hello, breaker_open, breaker_reset, breaker_status, browser_action, browser_start,
        browser_status, browser_stop, change_agent_model, confirm_revenue_swap_task,
        confirm_revenue_tax_task, create_service_quote, delete_provider_key, execute_plugin_tool,
        fail_revenue_swap_task, fail_revenue_tax_task, fail_service_request,
        fulfill_revenue_opportunity, fulfill_service_request, generate_deep_analysis, get_agents,
        get_available_models, get_cache_stats, get_capacity_stats, get_config,
        get_config_apply_status, get_config_capabilities, get_config_raw, get_costs,
        get_efficiency, get_mcp_runtime, get_memory_analytics, get_overview_timeseries,
        get_plugins, get_recommendations, get_revenue_opportunity, get_routing_dataset,
        get_routing_diagnostics, get_runtime_surfaces, get_service_request, get_task_events,
        get_throttle_stats, get_transactions, get_workspace_tasks, intake_micro_bounty_opportunity,
        intake_oracle_feed_opportunity, intake_revenue_opportunity, keystore_status,
        keystore_unlock, list_discovered_agents, list_paired_devices, list_revenue_opportunities,
        list_revenue_swap_tasks, list_revenue_tax_tasks, list_services_catalog,
        mcp_client_disconnect, mcp_client_discover, pair_device, plan_revenue_opportunity,
        qualify_revenue_opportunity, reconcile_revenue_swap_task, reconcile_revenue_tax_task,
        record_revenue_opportunity_feedback, register_discovered_agent, roster, run_routing_eval,
        score_revenue_opportunity, set_provider_key, settle_revenue_opportunity, start_agent,
        start_revenue_swap_task, start_revenue_tax_task, stop_agent, submit_revenue_swap_task,
        submit_revenue_tax_task, toggle_plugin, unpair_device, update_config, update_config_raw,
        verify_discovered_agent, verify_paired_device, verify_service_payment, wallet_address,
        wallet_balance, workspace_state,
    };
    use agent::{agent_message, agent_message_stream, agent_status};
    use channels::{
        get_channels_status, get_dead_letters, get_integrations, replay_dead_letter, test_channel,
    };
    use cron::{
        create_cron_job, delete_cron_job, get_cron_job, list_cron_jobs, list_cron_runs,
        run_cron_job_now, update_cron_job,
    };
    use health::{get_logs, health};
    use memory::{
        get_episodic_memory, get_semantic_categories, get_semantic_memory, get_semantic_memory_all,
        get_working_memory, get_working_memory_all, knowledge_ingest, memory_health, memory_search,
        trigger_consolidation, trigger_reindex,
    };
    use sessions::{
        analyze_session, analyze_turn, archive_session_handler, backfill_nicknames, create_session,
        get_session, get_session_feedback, get_session_insights, get_turn, get_turn_context,
        get_turn_feedback, get_turn_model_selection, get_turn_tips, get_turn_tools, list_messages,
        list_model_selection_events, list_session_turns, list_sessions, post_message,
        post_turn_feedback, put_turn_feedback,
    };
    use skills::{
        audit_skills, catalog_activate, catalog_install, catalog_list, delete_skill, get_skill,
        list_skills, reload_skills, toggle_skill, update_skill,
    };
    use subagents::{
        create_sub_agent, delete_sub_agent, get_subagent_retirement_candidates, list_sub_agents,
        retire_unused_subagents, toggle_sub_agent, update_sub_agent,
    };
    use themes::{install_catalog_theme, list_theme_catalog, list_themes};

    Router::new()
        .route("/", get(crate::dashboard::dashboard_handler))
        .route("/dashboard", get(dashboard_redirect))
        .route("/dashboard/", get(dashboard_redirect))
        .route("/api/health", get(health))
        .route("/health", get(health))
        .route("/api/config", get(get_config).put(update_config))
        .route(
            "/api/config/raw",
            get(get_config_raw).put(update_config_raw),
        )
        .route("/api/config/capabilities", get(get_config_capabilities))
        .route("/api/config/status", get(get_config_apply_status))
        .route(
            "/api/providers/{name}/key",
            put(set_provider_key).delete(delete_provider_key),
        )
        .route("/api/keystore/status", get(keystore_status))
        .route("/api/keystore/unlock", post(keystore_unlock))
        .route("/api/logs", get(get_logs))
        .route("/api/sessions", get(list_sessions).post(create_session))
        .route("/api/sessions/backfill-nicknames", post(backfill_nicknames))
        .route("/api/sessions/{id}", get(get_session))
        .route(
            "/api/sessions/{id}/messages",
            get(list_messages).post(post_message),
        )
        .route("/api/sessions/{id}/turns", get(list_session_turns))
        .route("/api/sessions/{id}/archive", post(archive_session_handler))
        .route("/api/sessions/{id}/insights", get(get_session_insights))
        .route("/api/sessions/{id}/feedback", get(get_session_feedback))
        .route("/api/turns/{id}", get(get_turn))
        .route("/api/turns/{id}/context", get(get_turn_context))
        .route(
            "/api/turns/{id}/model-selection",
            get(get_turn_model_selection),
        )
        .route("/api/turns/{id}/tools", get(get_turn_tools))
        .route("/api/turns/{id}/tips", get(get_turn_tips))
        .route("/api/models/selections", get(list_model_selection_events))
        .route(
            "/api/turns/{id}/feedback",
            get(get_turn_feedback)
                .post(post_turn_feedback)
                .put(put_turn_feedback),
        )
        .route("/api/memory/working", get(get_working_memory_all))
        .route("/api/memory/working/{session_id}", get(get_working_memory))
        .route("/api/memory/episodic", get(get_episodic_memory))
        .route("/api/memory/semantic", get(get_semantic_memory_all))
        .route(
            "/api/memory/semantic/categories",
            get(get_semantic_categories),
        )
        .route("/api/memory/semantic/{category}", get(get_semantic_memory))
        .route("/api/memory/search", get(memory_search))
        .route("/api/memory/health", get(memory_health))
        .route("/api/memory/consolidate", post(trigger_consolidation))
        .route("/api/memory/reindex", post(trigger_reindex))
        .route("/api/knowledge/ingest", post(knowledge_ingest))
        .route("/api/cron/jobs", get(list_cron_jobs).post(create_cron_job))
        .route("/api/cron/runs", get(list_cron_runs))
        .route(
            "/api/cron/jobs/{id}",
            get(get_cron_job)
                .put(update_cron_job)
                .delete(delete_cron_job),
        )
        .route(
            "/api/cron/jobs/{id}/run",
            axum::routing::post(run_cron_job_now),
        )
        .route("/api/stats/costs", get(get_costs))
        .route("/api/stats/timeseries", get(get_overview_timeseries))
        .route("/api/stats/efficiency", get(get_efficiency))
        .route("/api/stats/memory-analytics", get(get_memory_analytics))
        .route("/api/recommendations", get(get_recommendations))
        .route("/api/stats/transactions", get(get_transactions))
        .route("/api/services/catalog", get(list_services_catalog))
        .route("/api/services/quote", post(create_service_quote))
        .route("/api/services/requests/{id}", get(get_service_request))
        .route(
            "/api/services/requests/{id}/payment/verify",
            post(verify_service_payment),
        )
        .route(
            "/api/services/requests/{id}/fulfill",
            post(fulfill_service_request),
        )
        .route(
            "/api/services/requests/{id}/fail",
            post(fail_service_request),
        )
        .route(
            "/api/services/opportunities/intake",
            get(list_revenue_opportunities).post(intake_revenue_opportunity),
        )
        .route(
            "/api/services/opportunities/adapters/micro-bounty/intake",
            post(intake_micro_bounty_opportunity),
        )
        .route(
            "/api/services/opportunities/adapters/oracle-feed/intake",
            post(intake_oracle_feed_opportunity),
        )
        .route(
            "/api/services/opportunities/{id}",
            get(get_revenue_opportunity),
        )
        .route(
            "/api/services/opportunities/{id}/score",
            post(score_revenue_opportunity),
        )
        .route(
            "/api/services/opportunities/{id}/qualify",
            post(qualify_revenue_opportunity),
        )
        .route(
            "/api/services/opportunities/{id}/feedback",
            post(record_revenue_opportunity_feedback),
        )
        .route(
            "/api/services/opportunities/{id}/plan",
            post(plan_revenue_opportunity),
        )
        .route(
            "/api/services/opportunities/{id}/fulfill",
            post(fulfill_revenue_opportunity),
        )
        .route(
            "/api/services/opportunities/{id}/settle",
            post(settle_revenue_opportunity),
        )
        .route("/api/services/swaps", get(list_revenue_swap_tasks))
        .route("/api/services/tax-payouts", get(list_revenue_tax_tasks))
        .route(
            "/api/services/swaps/{id}/start",
            post(start_revenue_swap_task),
        )
        .route(
            "/api/services/swaps/{id}/submit",
            post(submit_revenue_swap_task),
        )
        .route(
            "/api/services/swaps/{id}/reconcile",
            post(reconcile_revenue_swap_task),
        )
        .route(
            "/api/services/swaps/{id}/confirm",
            post(confirm_revenue_swap_task),
        )
        .route(
            "/api/services/swaps/{id}/fail",
            post(fail_revenue_swap_task),
        )
        .route(
            "/api/services/tax-payouts/{id}/start",
            post(start_revenue_tax_task),
        )
        .route(
            "/api/services/tax-payouts/{id}/submit",
            post(submit_revenue_tax_task),
        )
        .route(
            "/api/services/tax-payouts/{id}/reconcile",
            post(reconcile_revenue_tax_task),
        )
        .route(
            "/api/services/tax-payouts/{id}/confirm",
            post(confirm_revenue_tax_task),
        )
        .route(
            "/api/services/tax-payouts/{id}/fail",
            post(fail_revenue_tax_task),
        )
        .route("/api/stats/cache", get(get_cache_stats))
        .route("/api/stats/capacity", get(get_capacity_stats))
        .route("/api/stats/throttle", get(get_throttle_stats))
        .route("/api/models/available", get(get_available_models))
        .route(
            "/api/models/routing-diagnostics",
            get(get_routing_diagnostics),
        )
        .route("/api/models/routing-dataset", get(get_routing_dataset))
        .route("/api/models/routing-eval", post(run_routing_eval))
        .route("/api/models/reset", post(admin::reset_model_scores))
        .route("/api/breaker/status", get(breaker_status))
        .route("/api/breaker/open/{provider}", post(breaker_open))
        .route("/api/breaker/reset/{provider}", post(breaker_reset))
        .route("/api/agent/status", get(agent_status))
        .route("/api/agent/message", post(agent_message))
        .route("/api/agent/message/stream", post(agent_message_stream))
        .route("/api/wallet/balance", get(wallet_balance))
        .route("/api/wallet/address", get(wallet_address))
        .route("/api/skills", get(list_skills))
        .route("/api/skills/catalog", get(catalog_list))
        .route("/api/skills/catalog/install", post(catalog_install))
        .route("/api/skills/catalog/activate", post(catalog_activate))
        .route("/api/skills/audit", get(audit_skills))
        .route(
            "/api/skills/{id}",
            get(get_skill).put(update_skill).delete(delete_skill),
        )
        .route("/api/skills/reload", post(reload_skills))
        .route("/api/skills/{id}/toggle", put(toggle_skill))
        .route("/api/themes", get(list_themes))
        .route("/api/themes/catalog", get(list_theme_catalog))
        .route("/api/themes/catalog/install", post(install_catalog_theme))
        .route("/api/plugins/catalog/install", post(catalog_install))
        .route("/api/plugins", get(get_plugins))
        .route("/api/plugins/{name}/toggle", put(toggle_plugin))
        .route(
            "/api/plugins/{name}/execute/{tool}",
            post(execute_plugin_tool),
        )
        .route("/api/browser/status", get(browser_status))
        .route("/api/browser/start", post(browser_start))
        .route("/api/browser/stop", post(browser_stop))
        .route("/api/browser/action", post(browser_action))
        .route("/api/agents", get(get_agents))
        .route("/api/agents/{id}/start", post(start_agent))
        .route("/api/agents/{id}/stop", post(stop_agent))
        .route(
            "/api/subagents",
            get(list_sub_agents).post(create_sub_agent),
        )
        .route(
            "/api/subagents/retirement-candidates",
            get(get_subagent_retirement_candidates),
        )
        .route(
            "/api/subagents/retire-unused",
            post(retire_unused_subagents),
        )
        .route(
            "/api/subagents/{name}",
            put(update_sub_agent).delete(delete_sub_agent),
        )
        .route("/api/subagents/{name}/toggle", put(toggle_sub_agent))
        .route("/api/workspace/state", get(workspace_state))
        .route("/api/workspace/tasks", get(get_workspace_tasks))
        .route("/api/admin/task-events", get(get_task_events))
        .route("/api/roster", get(roster))
        .route("/api/roster/{name}/model", put(change_agent_model))
        .route("/api/a2a/hello", post(a2a_hello))
        .route("/api/channels/status", get(get_channels_status))
        .route("/api/integrations", get(get_integrations))
        .route("/api/channels/{platform}/test", post(test_channel))
        .route("/api/channels/dead-letter", get(get_dead_letters))
        .route(
            "/api/channels/dead-letter/{id}/replay",
            post(replay_dead_letter),
        )
        .route("/api/runtime/surfaces", get(get_runtime_surfaces))
        .route(
            "/api/runtime/discovery",
            get(list_discovered_agents).post(register_discovered_agent),
        )
        .route(
            "/api/runtime/discovery/{id}/verify",
            post(verify_discovered_agent),
        )
        .route("/api/runtime/devices", get(list_paired_devices))
        .route("/api/runtime/devices/pair", post(pair_device))
        .route(
            "/api/runtime/devices/{id}/verify",
            post(verify_paired_device),
        )
        .route(
            "/api/runtime/devices/{id}",
            axum::routing::delete(unpair_device),
        )
        .route("/api/runtime/mcp", get(get_mcp_runtime))
        .route(
            "/api/runtime/mcp/clients/{name}/discover",
            post(mcp_client_discover),
        )
        .route(
            "/api/runtime/mcp/clients/{name}/disconnect",
            post(mcp_client_disconnect),
        )
        .route("/api/mcp/servers", get(mcp::list_servers))
        .route("/api/mcp/servers/{name}", get(mcp::get_server))
        .route("/api/mcp/servers/{name}/test", post(mcp::test_server))
        .route("/api/approvals", get(admin::list_approvals))
        .route("/api/approvals/{id}/approve", post(admin::approve_request))
        .route("/api/approvals/{id}/deny", post(admin::deny_request))
        .route("/api/ws-ticket", post(admin::issue_ws_ticket))
        .route("/api/interview/start", post(interview::start_interview))
        .route("/api/interview/turn", post(interview::interview_turn))
        .route("/api/interview/finish", post(interview::finish_interview))
        .route("/api/audit/policy/{turn_id}", get(admin::get_policy_audit))
        .route("/api/audit/tools/{turn_id}", get(admin::get_tool_audit))
        .route("/api/traces/{turn_id}", get(traces::get_trace))
        .route(
            "/api/traces/{turn_id}/react",
            get(traces::get_react_trace_handler),
        )
        .route("/api/traces/{turn_id}/export", get(traces::export_trace))
        .route("/api/traces/{turn_id}/flow", get(traces::get_trace_flow))
        .route(
            "/api/traces/{turn_id}/replay",
            axum::routing::post(traces::replay_trace),
        )
        .route("/api/traces/search", get(traces::search_traces))
        .route("/api/observability/traces", get(observability::list_traces))
        .route(
            "/api/observability/traces/{turn_id}/waterfall",
            get(observability::trace_waterfall),
        )
        .route(
            "/api/observability/delegation/outcomes",
            get(observability::delegation_outcomes),
        )
        .route(
            "/api/observability/delegation/stats",
            get(observability::delegation_stats),
        )
        .route(
            "/favicon.ico",
            get(|| async { axum::http::StatusCode::NO_CONTENT }),
        )
        // LLM analysis routes have their own concurrency limit to prevent
        // expensive analysis requests from starving lightweight API calls.
        .merge(
            Router::new()
                .route("/api/sessions/{id}/analyze", post(analyze_session))
                .route("/api/turns/{id}/analyze", post(analyze_turn))
                .route(
                    "/api/recommendations/generate",
                    post(generate_deep_analysis),
                )
                .layer(tower::limit::ConcurrencyLimitLayer::new(3))
                .with_state(state.clone()),
        )
        .fallback(|| async { JsonError(axum::http::StatusCode::NOT_FOUND, "not found".into()) })
        .layer(DefaultBodyLimit::max(1024 * 1024)) // 1MB
        .layer(middleware::from_fn(json_error_layer))
        .layer(middleware::from_fn(security_headers_layer))
        // CORS is applied at the top level in lib.rs (wraps authed + ws + public).
        .with_state(state)
}

/// Routes that must be accessible without API key authentication
/// (webhooks from external services, discovery endpoints).
pub fn build_public_router(state: AppState) -> Router {
    use admin::agent_card;
    use channels::{webhook_telegram, webhook_whatsapp, webhook_whatsapp_verify};

    Router::new()
        .route("/.well-known/agent.json", get(agent_card))
        .route("/api/webhooks/telegram", post(webhook_telegram))
        .route(
            "/api/webhooks/whatsapp",
            get(webhook_whatsapp_verify).post(webhook_whatsapp),
        )
        .layer(DefaultBodyLimit::max(1024 * 1024)) // 1MB — match auth router
        .with_state(state)
}

// ── MCP Gateway (P.1) ─────────────────────────────────────────

/// Builds an axum `Router` that serves the MCP protocol endpoint.
///
/// The returned router should be merged at the top level — it handles
/// its own transport (POST for JSON-RPC, GET for SSE, DELETE for sessions)
/// under the `/mcp` prefix via rmcp's `StreamableHttpService`.
///
/// Auth: MCP clients authenticate via `Authorization: Bearer <api_key>`.
/// The same API key used for the REST API is accepted here.
pub fn build_mcp_router(state: &AppState, api_key: Option<String>) -> Router {
    use crate::auth::ApiKeyLayer;
    use rmcp::transport::streamable_http_server::{
        StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
    };
    use roboticus_agent::mcp_handler::{McpToolContext, RoboticusMcpHandler};
    use std::time::Duration;

    let mcp_ctx = {
        let (workspace_root, agent_name, tool_allowed_paths, sandbox) = state
            .config
            .try_read()
            .map(|c| {
                (
                    c.agent.workspace.clone(),
                    c.agent.name.clone(),
                    c.security.filesystem.tool_allowed_paths.clone(),
                    roboticus_agent::tools::ToolSandboxSnapshot::from_config(
                        &c.security.filesystem,
                        &c.skills,
                    ),
                )
            })
            .unwrap_or_else(|_| {
                (
                    std::path::PathBuf::from("."),
                    "roboticus".to_string(),
                    Vec::new(),
                    roboticus_agent::tools::ToolSandboxSnapshot::default(),
                )
            });
        McpToolContext {
            agent_id: "roboticus-mcp-gateway".to_string(),
            agent_name,
            workspace_root,
            tool_allowed_paths,
            sandbox,
            db: Some(state.db.clone()),
        }
    };

    let handler = RoboticusMcpHandler::new(state.tools.clone(), mcp_ctx);

    let config = StreamableHttpServerConfig {
        sse_keep_alive: Some(Duration::from_secs(15)),
        stateful_mode: true,
        ..Default::default()
    };

    let service = StreamableHttpService::new(
        move || Ok(handler.clone()),
        Arc::new(LocalSessionManager::default()),
        config,
    );

    Router::new()
        .nest_service("/mcp", service)
        .layer(ApiKeyLayer::new(api_key))
}

// ── Re-exports for api.rs and lib.rs ────────────────────────────

pub use agent::{discord_poll_loop, email_poll_loop, signal_poll_loop, telegram_poll_loop};
pub use health::LogEntry;

// ── Tests ─────────────────────────────────────────────────────

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