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::{
    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;
use roboticus_core::personality::{self, OsIdentity, OsVoice};
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;

// ── JSON error response type ─────────────────────────────────

/// A JSON-formatted API error response. All error paths in the API return
/// `{"error": "<message>"}` with the appropriate HTTP status code.
#[derive(Debug)]
pub(crate) struct JsonError(pub axum::http::StatusCode, pub String);

impl axum::response::IntoResponse for JsonError {
    fn into_response(self) -> axum::response::Response {
        let body = serde_json::json!({ "error": self.1 });
        (self.0, 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())
}

// ── 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 ────────────────────────────────────

/// Holds the composed personality text plus metadata for status display.
#[derive(Debug, Clone)]
pub struct PersonalityState {
    pub os_text: String,
    pub firmware_text: String,
    pub identity: OsIdentity,
    pub voice: OsVoice,
}

impl PersonalityState {
    pub fn from_workspace(workspace: &std::path::Path) -> Self {
        let os = personality::load_os(workspace);
        let fw = personality::load_firmware(workspace);
        let operator = personality::load_operator(workspace);
        let directives = personality::load_directives(workspace);

        let os_text =
            personality::compose_identity_text(os.as_ref(), operator.as_ref(), directives.as_ref());
        let firmware_text = personality::compose_firmware_text(fw.as_ref());

        let (identity, voice) = match os {
            Some(os) => (os.identity, os.voice),
            None => (
                OsIdentity {
                    name: String::new(),
                    version: "1.0".into(),
                    generated_by: "none".into(),
                },
                OsVoice::default(),
            ),
        };

        Self {
            os_text,
            firmware_text,
            identity,
            voice,
        }
    }

    pub fn empty() -> Self {
        Self {
            os_text: String::new(),
            firmware_text: String::new(),
            identity: OsIdentity {
                name: String::new(),
                version: "1.0".into(),
                generated_by: "none".into(),
            },
            voice: OsVoice::default(),
        }
    }
}

/// 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;
    }
}

// ── 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 is_json = response
        .headers()
        .get(axum::http::header::CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .is_some_and(|ct| ct.contains("application/json"));
    if is_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 = serde_json::json!({ "error": error_msg });
    let body_bytes = serde_json::to_vec(&json_body)
        .unwrap_or_else(|_| br#"{"error":"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/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, 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, 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,
    };
    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::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/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/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/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/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/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/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))
        .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;