poimen-memory/crates/mem-cli/src/http_server.rs

use actix_web::{web, App, HttpServer, HttpResponse, HttpRequest, middleware::Logger};
use anyhow::Result;
use chrono;
use mem_llm::{EmbeddingsClient, RerankClient};
use mem_store::{init_schema, VectorStore};
use serde_json::json;
use sqlx::PgPool;
use std::sync::Arc;
use std::time::Instant;
use crate::endpoints::IngestRequest;
use crate::ingest_worker::IngestWorker;
use crate::query_worker::QueryWorker;
use crate::rate_limiter::{RateLimiter, LimitConfig};
use crate::idempotency::IdempotencyStore;
use crate::jwt_validator::{JwtValidator, JwtClaims};
use crate::opensearch_client::{OpenSearchClient, HybridWeights};
use crate::dual_write_indexer::DualWriteIndexer;
use crate::gateway_queue_adapter::GatewayQueueAdapter;
use crate::queue_worker::{QueueWorker, QueueWorkerConfig};
use crate::queue_adapter::QueueAdapter;
// RBAC removed for MVP - will add after core ingest/query working
use crate::handlers::{
    QueryParams, QueryParamsError, SearchMethod, build_search_response, 
    LearnParams, LearnParamsError, build_learn_response,
    visualize_handler, visualize_stream_handler, compact_handler
};

/// Server state with database and workers
pub struct AppState {
    pub api_key: String,
    pub start_time: Instant,
    pub pool: PgPool,
    pub vector_store: Arc<VectorStore>,
    pub embeddings: Arc<EmbeddingsClient>,
    pub ingest_worker: Arc<IngestWorker>,
    pub query_worker: Arc<QueryWorker>,
    pub rate_limiter: Arc<RateLimiter>,
    pub idempotency_store: Arc<IdempotencyStore>,
    pub jwt_validator: Option<Arc<JwtValidator>>,
    pub auth_mode: AuthMode,
    pub opensearch_client: Option<Arc<OpenSearchClient>>,
    /// M3.8 Query Optimizer (optional, from environment)
    pub optimizer_service: Option<Arc<mem_core::optimizer::OptimizerService>>,
}

/// Authentication mode
#[derive(Debug, Clone, Copy)]
pub enum AuthMode {
    Jwt,      // Validate JWT from Authentik
    ApiKey,   // Fallback to static API key
    None,     // No auth (testing only)
}

/// Auth extractor — validates JWT, apikey, or disabled
async fn validate_auth(req: &HttpRequest, state: &AppState) -> Result<(JwtClaims, String), HttpResponse> {
    match state.auth_mode {
        AuthMode::Jwt => validate_jwt_token(req, state).await,
        AuthMode::ApiKey => validate_apikey(req, state),
        AuthMode::None => {
            tracing::warn!("Auth disabled - returning synthetic claims");
            let claims = JwtClaims {
                sub: "test-user".to_string(),
                iss: "test".to_string(),
                aud: "memory".to_string(),
                exp: i64::MAX,
                iat: chrono::Utc::now().timestamp(),
                nbf: None,
                permissions: Some(vec!["memory:write".to_string(), "memory:read".to_string()]),
                groups: Some(vec!["test".to_string()]),
                roles: None,
            };
            Ok((claims, "synthetic-token".to_string()))
        }
    }
}

/// Validate JWT token from Authorization header
async fn validate_jwt_token(req: &HttpRequest, state: &AppState) -> Result<(JwtClaims, String), HttpResponse> {
    let validator = state
        .jwt_validator
        .as_ref()
        .ok_or_else(|| HttpResponse::InternalServerError().json(json!({"error": "jwt_validator_not_configured"})))?;

    let auth_header = req
        .headers()
        .get("Authorization")
        .and_then(|h| h.to_str().ok())
        .ok_or_else(|| {
            HttpResponse::Unauthorized().json(json!({
                "error": "unauthorized",
                "reason": "missing Authorization header"
            }))
        })?
        .to_string();

    let token = crate::jwt_validator::JwtValidator::extract_bearer_token(&auth_header)
        .map_err(|_| {
            HttpResponse::Unauthorized().json(json!({
                "error": "unauthorized",
                "reason": "invalid Authorization header format"
            }))
        })?
        .to_string();

    let claims = validator
        .validate_token(&token)
        .await
        .map_err(|e| {
            tracing::warn!("JWT validation failed: {}", e);
            HttpResponse::Unauthorized().json(json!({
                "error": "unauthorized",
                "reason": format!("JWT validation failed: {}", e)
            }))
        })?
        .clone();

    Ok((claims, token))
}

/// Fallback: validate apikey header
fn validate_apikey(req: &HttpRequest, state: &AppState) -> Result<(JwtClaims, String), HttpResponse> {
    let api_key = req
        .headers()
        .get("apikey")
        .and_then(|h| h.to_str().ok())
        .map(|s| s.to_string());

    if api_key.as_ref() != Some(&state.api_key) {
        return Err(HttpResponse::Unauthorized().json(json!({
            "error": "unauthorized",
            "reason": "missing or invalid apikey header"
        })));
    }

    // Create a synthetic JWT claims for apikey mode (all permissions)
    let claims = JwtClaims {
        sub: "apikey-user".to_string(),
        iss: "internal".to_string(),
        aud: "memory".to_string(),
        exp: i64::MAX,
        iat: chrono::Utc::now().timestamp(),
        nbf: None,
        permissions: Some(vec!["*".to_string()]),
        groups: None,
        roles: Some(vec!["admin".to_string()]),  // API key gets admin role
    };

    Ok((claims, "apikey".to_string()))
}

/// Check if claims have required capability
fn has_capability(claims: &JwtClaims, required_capability: &str) -> bool {
    if let Some(perms) = &claims.permissions {
        // Wildcard permission grants everything
        if perms.contains(&"*".to_string()) {
            return true;
        }
        perms.contains(&required_capability.to_string())
    } else {
        false
    }
}

/// Extract client identifier from claims for rate limiting
fn extract_rate_limit_key(claims: &JwtClaims) -> String {
    // Use subject (user/service ID) as rate limit key
    claims.sub.clone()
}

/// Rate limit guard — call this in handlers to check rate limit
fn check_rate_limit(claims: &JwtClaims, state: &AppState, endpoint: &str) -> Result<(), HttpResponse> {
    let key = extract_rate_limit_key(claims);
    
    match state.rate_limiter.check(&key, endpoint) {
        Ok(_) => Ok(()),
        Err(rate_limit_err) => {
            let retry_after = rate_limit_err.retry_after_seconds.to_string();
            Err(HttpResponse::TooManyRequests()
                .insert_header(("Retry-After", retry_after))
                .json(json!({
                    "error": "rate_limit_exceeded",
                    "reason": rate_limit_err.reason.clone(),
                    "retry_after_seconds": rate_limit_err.retry_after_seconds,
                    "limit_window": format!("{}s", rate_limit_err.limit_window_secs),
                })))
        }
    }
}

/// Start HTTP server with database initialization
pub async fn start_server(port: u16, api_key: String, database_url: &str) -> Result<()> {
    // Create connection pool
    let pool = PgPool::connect(database_url).await?;
    tracing::info!("Connected to database");

    // Initialize schema
    match init_schema(&pool).await {
        Ok(_) => tracing::info!("Schema initialized"),
        Err(e) => {
            tracing::warn!("Schema init error (may be non-fatal): {}", e);
            // Continue anyway - tables might exist
        }
    }

    // Create workers
    let vector_store = Arc::new(VectorStore::new(pool.clone()));
    let embeddings = Arc::new(EmbeddingsClient::from_env()?);
    let ingest_worker = Arc::new(IngestWorker::new(pool.clone(), (*embeddings).clone()));
    let reranker = RerankClient::from_env()?;
    let query_worker = Arc::new(QueryWorker::new(VectorStore::new(pool.clone()), (*embeddings).clone(), reranker));

    // Initialize rate limiter and idempotency store
    let limit_config = LimitConfig {
        ingest_per_hour: std::env::var("MEM_RATE_LIMIT_INGEST")
            .ok()
            .and_then(|s| s.parse().ok())
            .unwrap_or(100.0),
        query_per_hour: std::env::var("MEM_RATE_LIMIT_QUERY")
            .ok()
            .and_then(|s| s.parse().ok())
            .unwrap_or(1000.0),
        projects_per_hour: std::env::var("MEM_RATE_LIMIT_PROJECTS")
            .ok()
            .and_then(|s| s.parse().ok())
            .unwrap_or(100.0),
        burst_per_second: std::env::var("MEM_RATE_LIMIT_BURST")
            .ok()
            .and_then(|s| s.parse().ok())
            .unwrap_or(10.0),
    };
    let rate_limiter = Arc::new(RateLimiter::new(limit_config));

    let idempotency_ttl = std::env::var("MEM_IDEMPOTENCY_TTL_SECS")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(86400); // 24 hours default
    let idempotency_store = Arc::new(IdempotencyStore::new(idempotency_ttl));

    // Determine auth mode
    let auth_mode = std::env::var("MEM_AUTH_MODE")
        .unwrap_or_else(|_| "apikey".to_string())
        .to_lowercase();
    let auth_mode = match auth_mode.as_str() {
        "jwt" => AuthMode::Jwt,
        "apikey" => AuthMode::ApiKey,
        "none" => AuthMode::None,
        _ => {
            tracing::warn!("Unknown auth mode: {}, defaulting to apikey", auth_mode);
            AuthMode::ApiKey
        }
    };

    // Setup JWT validator if in JWT mode
    let jwt_validator = if matches!(auth_mode, AuthMode::Jwt) {
        let issuer = std::env::var("AUTHENTIK_ISSUER").map_err(|e| {
            anyhow::anyhow!("AUTHENTIK_ISSUER env var required for JWT auth: {}", e)
        })?;
        let audience = std::env::var("AUTHENTIK_AUDIENCE").map_err(|e| {
            anyhow::anyhow!("AUTHENTIK_AUDIENCE env var required for JWT auth: {}", e)
        })?;
        let cache_ttl = std::env::var("JWT_CACHE_TTL_SECS")
            .ok()
            .and_then(|s| s.parse().ok())
            .unwrap_or(3600); // 1 hour default
        Some(Arc::new(crate::jwt_validator::JwtValidator::new(
            issuer,
            audience,
            cache_ttl,
        )))
    } else {
        None
    };

    // Initialize OpenSearch client if configured
    let opensearch_client = if let Ok(hosts_str) = std::env::var("OPENSEARCH_HOSTS") {
        let hosts: Vec<String> = hosts_str
            .split(',')
            .map(|h| h.trim().to_string())
            .collect();
        Some(Arc::new(OpenSearchClient::new(hosts)))
    } else {
        tracing::warn!("OPENSEARCH_HOSTS not set, hybrid search disabled");
        None
    };

    // Initialize M3.8 Query Optimizer if enabled
    let optimizer_service = match mem_core::optimizer::OptimizerServiceBuilder::new().build() {
        Ok(service) => {
            tracing::info!("M3.8 Query Optimizer enabled");
            Some(Arc::new(service))
        }
        Err(e) => {
            tracing::debug!("M3.8 Query Optimizer not available: {}", e);
            None
        }
    };

    // Initialize M8.2 Queue Adapter and Dual-Write Indexer
    let queue_adapter: Arc<dyn QueueAdapter> = if let Ok(gateway_url) = std::env::var("GATEWAY_URL") {
        let adapter = GatewayQueueAdapter::with_authentik(
            gateway_url,
            std::env::var("AUTHENTIK_ISSUER").unwrap_or_default(),
            std::env::var("AUTHENTIK_CLIENT_ID").unwrap_or_default(),
            std::env::var("AUTHENTIK_CLIENT_SECRET").unwrap_or_default(),
        );
        tracing::info!("M8.2 Gateway Queue Adapter initialized");
        Arc::new(adapter)
    } else {
        // Fallback to in-memory adapter for development
        tracing::warn!("GATEWAY_URL not set, using in-memory queue adapter (development only)");
        Arc::new(crate::queue_adapter::InMemoryQueueAdapter::new())
    };

    let dual_write_indexer = Arc::new(DualWriteIndexer::new(
        pool.clone(),
        opensearch_client.clone(),
        queue_adapter.clone(),
    ));

    // Start queue worker in background (only if queue operations are enabled)
    let enable_queue_worker = std::env::var("ENABLE_QUEUE_WORKER")
        .unwrap_or_else(|_| "true".to_string())
        .to_lowercase()
        == "true";

    if enable_queue_worker {
        let worker_indexer = dual_write_indexer.clone();
        let worker_embeddings = embeddings.clone();
        let worker_config = QueueWorkerConfig {
            max_messages_per_batch: std::env::var("QUEUE_BATCH_SIZE")
                .ok()
                .and_then(|s| s.parse().ok())
                .unwrap_or(10),
            visibility_timeout_secs: std::env::var("QUEUE_VISIBILITY_TIMEOUT")
                .ok()
                .and_then(|s| s.parse().ok())
                .unwrap_or(300),
            wait_time_secs: std::env::var("QUEUE_WAIT_TIME")
                .ok()
                .and_then(|s| s.parse().ok())
                .unwrap_or(20),
            project: std::env::var("QUEUE_PROJECT").ok(),
            max_retries: std::env::var("QUEUE_MAX_RETRIES")
                .ok()
                .and_then(|s| s.parse().ok())
                .unwrap_or(3),
            ..Default::default()
        };

        tokio::spawn(async move {
            let worker = QueueWorker::new(worker_indexer, worker_embeddings, worker_config);
            if let Err(e) = worker.start().await {
                tracing::error!("Queue worker error: {}", e);
            }
        });

        tracing::info!("M8.2 Queue Worker started (background task)");
    }

    let state = web::Data::new(AppState {
        api_key,
        start_time: Instant::now(),
        pool,
        vector_store,
        embeddings,
        ingest_worker,
        query_worker,
        rate_limiter,
        idempotency_store,
        jwt_validator,
        auth_mode,
        opensearch_client,
        optimizer_service,
    });

    tracing::info!("Starting HTTP server on port {}", port);
    tracing::info!("Creating HttpServer instance...");

    let server = HttpServer::new(move || {
        tracing::debug!("HttpServer::new() closure executing");
        App::new()
            .app_data(state.clone())
            .wrap(Logger::default())
            .route("/health", web::get().to(health_check))
            .route("/memory/ingest", web::post().to(ingest_handler))
            .route("/memory/ingest/{ingest_id}", web::get().to(ingest_status))
            .route("/memory/query", web::get().to(query_handler))
            .route("/memory/query", web::post().to(crate::handlers::unified_query::unified_query_handler))
            .route("/memory/query/semantic/entities", web::post().to(crate::handlers::semantic::search_entities_handler))
            .route("/memory/query/semantic/edges", web::post().to(crate::handlers::semantic::search_edges_handler))
            .route("/memory/query/hybrid", web::post().to(crate::handlers::semantic::hybrid_search_handler))
            .route("/memory/context", web::post().to(context_handler))
            .route("/memory/projects", web::get().to(projects_handler))
            .route("/memory/skills", web::get().to(skills_handler))
            .route("/memory/learn", web::post().to(learn_handler))
            .route("/memory/vault/generate", web::post().to(vault_generate_handler))
            .route("/memory/vault", web::get().to(vault_browser_handler))
            .route("/memory/vault/{project}", web::get().to(vault_project_handler))
            .route("/memory/vault/{project}/{file}", web::get().to(vault_file_handler))
            .route("/memory/visualize", web::post().to(visualize_handler))
            .route("/memory/visualize/stream", web::post().to(visualize_stream_handler))
            .route("/memory/compact", web::post().to(compact_handler))
            .route("/memory/synthesis/link-entities", web::post().to(crate::handlers::synthesis::link_entities_handler))
            .route("/memory/synthesis/detect-aliases", web::post().to(crate::handlers::synthesis::detect_aliases_handler))
            .route("/memory/synthesis/suggest-merges", web::post().to(crate::handlers::synthesis::suggest_merges_handler))
            .route("/memory/synthesis/detect-coreferences", web::post().to(crate::handlers::synthesis::detect_coreferences_handler))
            .route("/memory/synthesis/infer", web::post().to(crate::handlers::synthesis::infer_facts_handler))
            .route("/memory/synthesis/transitive-closure", web::post().to(crate::handlers::synthesis::transitive_closure_handler))
            .route("/memory/synthesis/reasoning-paths", web::post().to(crate::handlers::synthesis::reasoning_paths_handler))
            .route("/memory/synthesis/reason", web::post().to(crate::handlers::synthesis::reason_query_handler))
            .route("/memory/synthesis/summarize", web::post().to(crate::handlers::synthesis::summarize_handler))
            .route("/memory/synthesis", web::post().to(crate::handlers::unified_synthesis::unified_synthesis_handler))
            .route("/agents", web::post().to(crate::handlers::agent_handler::register_agent_handler))
            .route("/agents/{id}", web::get().to(crate::handlers::agent_handler::get_agent_handler))
            .route("/agents/{id}", web::put().to(crate::handlers::agent_handler::update_agent_handler))
            .route("/agents/{id}", web::delete().to(crate::handlers::agent_handler::delete_agent_handler))
            .route("/agents/{id}/metrics", web::get().to(crate::handlers::agent_handler::get_agent_metrics_handler))
    });
    
    tracing::info!("HttpServer instance created, binding to 0.0.0.0:{}", port);
    let server = server.bind(("0.0.0.0", port))?;
    tracing::info!("Successfully bound to port {}, about to run", port);
    
    server.run().await?;

    Ok(())
}

/// Health check (no auth)
pub async fn health_check(state: web::Data<AppState>) -> HttpResponse {
    let uptime = state.start_time.elapsed().as_secs();
    HttpResponse::Ok().json(json!({"status": "ok", "uptime_seconds": uptime}))
}

/// POST /memory/ingest — queue an ingest job
pub async fn ingest_handler(
    req: HttpRequest,
    body: web::Json<IngestRequest>,
    state: web::Data<AppState>,
) -> HttpResponse {
    // Auth + capability check
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };
    if !has_capability(&claims, "memory:write") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:write"
        }));
    }
    if let Err(e) = check_rate_limit(&claims, &state, "/memory/ingest") {
        return e;
    }

    // Check idempotency
    if let Some(cached) = state.idempotency_store.get(&body.ingest_id) {
        tracing::info!("Returning cached response for ingest_id: {}", body.ingest_id);
        return HttpResponse::Accepted().json(cached);
    }

    // Execute ingest
    execute_ingest(&state, &body).await
}

/// Execute ingest job creation and spawn worker
async fn execute_ingest(
    state: &web::Data<AppState>,
    body: &IngestRequest,
) -> HttpResponse {
    let records: Vec<(String, String)> = body.records
        .iter()
        .map(|r| (r.text.clone(), body.source.clone()))
        .collect();

    let job_result = sqlx::query(
        "INSERT INTO ingest_jobs (id, project, ingest_id, status, created_at)
         VALUES ($1, $2, $3, 'pending', NOW())
         ON CONFLICT (ingest_id) DO NOTHING
         RETURNING id",
    )
    .bind(uuid::Uuid::new_v4())
    .bind(&body.project)
    .bind(&body.ingest_id)
    .fetch_optional(&state.pool)
    .await;

    let response = json!({
        "ingest_id": body.ingest_id,
        "status": "pending",
        "status_url": format!("/memory/ingest/{}", body.ingest_id)
    });

    match job_result {
        Ok(Some(_)) => {
            // Spawn async ingest task
            let worker = state.ingest_worker.clone();
            let project = body.project.clone();
            let ingest_id = body.ingest_id.clone();
            tokio::spawn(async move {
                if let Err(e) = worker.process_ingest(&project, &ingest_id, records).await {
                    tracing::error!("Ingest failed: {}", e);
                }
            });
            state.idempotency_store.set(body.ingest_id.clone(), response.clone());
            HttpResponse::Accepted().json(response)
        }
        Ok(None) => {
            // Already exists (concurrent insert)
            state.idempotency_store.set(body.ingest_id.clone(), response.clone());
            HttpResponse::Accepted().json(response)
        }
        Err(e) => {
            tracing::error!("DB error: {}", e);
            HttpResponse::InternalServerError().json(json!({"error": "database_error"}))
        }
    }
}

/// GET /memory/ingest/{ingest_id} — check ingest status
pub async fn ingest_status(
    req: HttpRequest,
    ingest_id: web::Path<String>,
    state: web::Data<AppState>,
) -> HttpResponse {
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };

    // Check read capability
    if !has_capability(&claims, "memory:read") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:read"
        }));
    }

    let id = ingest_id.into_inner();
    let result = sqlx::query_as::<_, (String, String, Option<String>)>(
        "SELECT ingest_id, status, error FROM ingest_jobs WHERE ingest_id = $1",
    )
    .bind(&id)
    .fetch_optional(&state.pool)
    .await;

    match result {
        Ok(Some((ingest_id, status, error))) => {
            HttpResponse::Ok().json(json!({
                "ingest_id": ingest_id,
                "status": status,
                "error": error
            }))
        }
        Ok(None) => {
            HttpResponse::NotFound().json(json!({"error": "not_found"}))
        }
        Err(_) => {
            HttpResponse::InternalServerError().json(json!({"error": "database_error"}))
        }
    }
}

/// M3.8: Optimize search results using pluggable OptimizerService
/// 
/// If optimizer_service is available, optimizes chunk text before returning.
/// Gracefully falls back to original on any error.
/// 
/// For LLM integration, use build_cache_aligned_async from PromptBuilder:
/// ```ignore
/// let msgs = PromptBuilder::build_cache_aligned_async(
///     &query,
///     previous_memory.as_deref(),
///     &chunk,
///     &optimizer_service,
/// ).await?;
/// ```
async fn optimize_search_results(
    mut results: Vec<crate::query_worker::QueryResult>,
    optimizer: Option<&Arc<mem_core::optimizer::OptimizerService>>,
) -> Vec<crate::query_worker::QueryResult> {
    if optimizer.is_none() {
        return results; // Optimizer not enabled, return as-is
    }

    let svc = optimizer.unwrap();
    let mut optimized = Vec::new();

    for mut result in results {
        match svc.optimize(&result.text, "text/plain", Some("raw")).await {
            Ok(optimized_bytes) => {
                if let Ok(optimized_text) = String::from_utf8(optimized_bytes) {
                    let orig_len = result.text.len();
                    let opt_len = optimized_text.len();
                    result.text = optimized_text;
                    tracing::debug!(
                        "M3.8 optimized chunk: {} bytes → {} bytes ({:.1}% compression)",
                        orig_len,
                        opt_len,
                        (opt_len as f32 / orig_len as f32) * 100.0
                    );
                }
            }
            Err(e) => {
                // Graceful fallback: use original on optimization error
                tracing::warn!("M3.8 optimization failed, using original: {}", e);
            }
        }
        optimized.push(result);
    }

    optimized
}

/// POST /memory/learn — Ingest knowledge via gated loop (LLM evaluates + compacts)
///
/// OpenAI-compatible style endpoint. Accepts markdown text, chunks it,
/// runs each chunk through the gated loop where the LLM decides whether
/// to accept/reject and rewrites memory to stay compact.
///
/// Request:
///   POST /memory/learn
///   { "project": "knowledge", "text": "## Rust\n- ownership...", "query": "What are key Rust patterns?" }
///
/// Response:
///   { "project": "knowledge", "chunks_seen": 5, "chunks_used": 3,
///     "memory": "compacted memory text...", "status": "completed" }
pub async fn learn_handler(
    req: HttpRequest,
    body: web::Json<serde_json::Value>,
    state: web::Data<AppState>,
) -> HttpResponse {
    // Auth + capability check
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };
    if !has_capability(&claims, "memory:write") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:write"
        }));
    }
    if let Err(e) = check_rate_limit(&claims, &state, "/memory/ingest") {
        return e;
    }

    // Parse + validate params
    let params = match LearnParams::from_body(&body) {
        Ok(p) => p,
        Err(e) => return e.to_response(),
    };

    // Chunk the markdown
    let chunks = chunk_markdown_text(&params.text, params.chunk_size);
    if chunks.is_empty() {
        return HttpResponse::BadRequest().json(json!({
            "error": "bad_request",
            "reason": "text produced no chunks"
        }));
    }

    // Build domain chunks
    let domain_chunks: Vec<mem_core::domain::Chunk> = chunks
        .iter()
        .enumerate()
        .map(|(i, text)| {
            let record = mem_core::Record {
                role: mem_core::Role::User,
                text: text.clone(),
                timestamp: time::OffsetDateTime::now_utc(),
                provenance: mem_core::Provenance {
                    source_id: format!("learn://{}:{}", params.project, i),
                    offset: i as u64,
                },
            };
            mem_core::domain::Chunk::new((i + 1) as u32, vec![record], text.len() / 4)
        })
        .collect();

    // Build query for gated loop
    let query = mem_core::query::Query {
        id: format!("learn-{}", uuid::Uuid::new_v4()),
        question: params.question.clone(),
        exit_gate: false,
    };

    let config = mem_core::gated_loop::LoopConfig {
        level: mem_core::Level::L1,
        query,
        memory_budget: params.memory_budget,
        use_exit_gate: false,
    };

    // Create LLM client
    let llm_base = std::env::var("LLM_API_BASE")
        .unwrap_or_else(|_| "https://api.riotpiao.com".to_string());
    let llm_key = std::env::var("LLM_API_KEY")
        .or_else(|_| std::env::var("MEM_API_KEY"))
        .unwrap_or_default();
    let llm = match mem_llm::ChatClient::new(&llm_base, &llm_key, &params.model) {
        Ok(c) => c,
        Err(e) => return HttpResponse::InternalServerError().json(json!({
            "error": "llm_init_failed",
            "reason": e.to_string()
        })),
    };

    // Run gated loop
    let outcome = match mem_core::gated_loop::run_loop(config, domain_chunks, &llm) {
        Ok(o) => o,
        Err(e) => return HttpResponse::InternalServerError().json(json!({
            "error": "gated_loop_failed",
            "reason": e.to_string()
        })),
    };

    // Store compacted memory
    let stored = store_compacted_memory(&state, &params.project, &outcome.final_memory).await;

    build_learn_response(
        &params.project,
        &params.model,
        outcome.chunks_seen,
        outcome.chunks_used,
        &outcome.final_memory,
        stored,
    )
}

/// Store compacted memory in pgvector
async fn store_compacted_memory(
    state: &web::Data<AppState>,
    project: &str,
    memory: &str,
) -> bool {
    if memory.is_empty() {
        return false;
    }

    let embedding = match state.embeddings.embed_one(memory).await {
        Ok(e) => e,
        Err(e) => {
            tracing::error!("Failed to embed compacted memory: {}", e);
            return false;
        }
    };

    let sha = {
        use sha2::{Digest, Sha256};
        let mut h = Sha256::new();
        h.update(memory.as_bytes());
        format!("{:x}", h.finalize())
    };

    let result = sqlx::query(
        "INSERT INTO memory_chunks (id, project, level, text, embedding, sha256, source, created_at)
         VALUES ($1, $2, 'L1', $3, $4, $5, $6, NOW())
         ON CONFLICT (sha256) DO UPDATE SET text = $3, embedding = $4",
    )
    .bind(uuid::Uuid::new_v4())
    .bind(project)
    .bind(memory)
    .bind(embedding.to_vec())
    .bind(&sha)
    .bind(format!("learn://{}", project))
    .fetch_optional(&state.pool)
    .await;

    match result {
        Ok(_) => true,
        Err(e) => {
            tracing::error!("Failed to store compacted memory: {}", e);
            false
        }
    }
}

/// Split markdown on ## headings for learn endpoint.
fn chunk_markdown_text(content: &str, max_chunk: usize) -> Vec<String> {
    let mut chunks = Vec::new();
    let mut current = String::new();
    for line in content.lines() {
        if line.starts_with("## ") && !current.is_empty() {
            let trimmed = current.trim().to_string();
            if !trimmed.is_empty() { chunks.push(trimmed); }
            current = String::new();
        }
        current.push_str(line);
        current.push('\n');
        if current.len() > max_chunk {
            let trimmed = current.trim().to_string();
            if !trimmed.is_empty() { chunks.push(trimmed); }
            current = String::new();
        }
    }
    let trimmed = current.trim().to_string();
    if !trimmed.is_empty() { chunks.push(trimmed); }
    chunks
}

/// GET /memory/query — semantic search across memories
pub async fn query_handler(
    req: HttpRequest,
    query: web::Query<std::collections::HashMap<String, String>>,
    state: web::Data<AppState>,
) -> HttpResponse {
    // Auth + capability check
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };
    if !has_capability(&claims, "memory:read") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:read"
        }));
    }
    if let Err(e) = check_rate_limit(&claims, &state, "/memory/query") {
        return e;
    }

    // Parse + validate params
    let params = match QueryParams::from_query(&query) {
        Ok(p) => p,
        Err(e) => return e.to_response(),
    };

    // Execute temporal graph query
    match query_temporal_graph(&state, &params).await {
        Ok(response) => HttpResponse::Ok().json(response),
        Err(e) => {
            tracing::error!("Temporal graph query failed: {}", e);
            HttpResponse::InternalServerError().json(json!({"error": "query_failed", "reason": e.to_string()}))
        }
    }
}

/// Execute hybrid search with OpenSearch fallback
async fn execute_hybrid_search(
    state: &web::Data<AppState>,
    params: &QueryParams,
    results: Vec<crate::query_worker::QueryResult>,
    token: &str,
) -> HttpResponse {
    let Some(os_client) = &state.opensearch_client else {
        tracing::info!("OpenSearch not configured, using semantic search only");
        return build_search_response(params, results, Some("semantic_only"));
    };

    let sem_results: Vec<(String, f32, String, String, Vec<String>)> = results
        .iter()
        .enumerate()
        .map(|(i, r)| (
            format!("sem-{}", i),
            r.score,
            r.text.clone(),
            r.source.clone().unwrap_or_default(),
            r.provenance.clone(),
        ))
        .collect();

    let weights = HybridWeights { semantic: 0.6, lexical: 0.4 };

    match os_client.hybrid_search(&params.question, sem_results, token, params.limit as usize, &weights).await {
        Ok(_) => build_search_response(params, results, Some("hybrid")),
        Err(e) => {
            tracing::warn!("Hybrid search failed, falling back to semantic: {}", e);
            build_search_response(params, results, Some("semantic_fallback"))
        }
    }
}

/// GET /memory/projects — list projects with memory
pub async fn projects_handler(
    req: HttpRequest,
    state: web::Data<AppState>,
) -> HttpResponse {
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };

    // Check read capability
    if !has_capability(&claims, "memory:read") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:read"
        }));
    }

    if let Err(e) = check_rate_limit(&claims, &state, "/memory/projects") {
        return e;
    }

    let result = sqlx::query_as::<_, (String,)>(
        "SELECT DISTINCT project FROM memories_l2 ORDER BY project",
    )
    .fetch_all(&state.pool)
    .await;

    match result {
        Ok(rows) => {
            let projects: Vec<String> = rows.into_iter().map(|(p,)| p).collect();
            HttpResponse::Ok().json(json!({
                "projects": projects,
                "count": projects.len()
            }))
        }
        Err(_) => {
            HttpResponse::InternalServerError().json(json!({"error": "database_error"}))
        }
    }
}

/// GET /memory/skills — list extracted skills
pub async fn skills_handler(
    req: HttpRequest,
    state: web::Data<AppState>,
) -> HttpResponse {
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };

    // Check read capability
    if !has_capability(&claims, "memory:read") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:read"
        }));
    }

    let result = sqlx::query_as::<_, (String, String, String)>(
        "SELECT name, description, when_to_use FROM skills ORDER BY created_at DESC LIMIT 50",
    )
    .fetch_all(&state.pool)
    .await;

    match result {
        Ok(rows) => {
            let skills: Vec<serde_json::Value> = rows
                .into_iter()
                .map(|(name, desc, when_to_use)| {
                    json!({
                        "name": name,
                        "description": desc,
                        "when_to_use": when_to_use
                    })
                })
                .collect();
            HttpResponse::Ok().json(json!({
                "skills": skills,
                "count": skills.len()
            }))
        }
        Err(_) => {
            HttpResponse::InternalServerError().json(json!({"error": "database_error"}))
        }
    }
}

/// POST /memory/context — three-tier context lookup for failure diagnosis
pub async fn context_handler(
    req: HttpRequest,
    body: web::Json<crate::context_endpoint::ContextRequest>,
    state: web::Data<AppState>,
) -> HttpResponse {
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };

    // Check read capability
    if !has_capability(&claims, "memory:read") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:read"
        }));
    }

    if let Err(e) = check_rate_limit(&claims, &state, "/memory/context") {
        return e;
    }

    let project = body.project.clone().unwrap_or_else(|| "all".to_string());
    let scope = body.scope.clone().unwrap_or_else(|| "project".to_string());
    let budget = body.budget.unwrap_or(6000);

    let lookup = crate::context_endpoint::ContextLookup::new(budget, project, scope);

    match lookup.lookup(body.into_inner()).await {
        Ok(response) => {
            tracing::info!(
                tier = response.tier,
                lessons = response.lessons.len(),
                skills = response.skills.len(),
                "context lookup successful"
            );
            HttpResponse::Ok().json(response)
        }
        Err(e) => {
            tracing::error!("context lookup error: {}", e);
            HttpResponse::BadRequest().json(json!({
                "error": "lookup_failed",
                "reason": e.to_string()
            }))
        }
    }
}

/// POST /memory/vault/generate — generate Obsidian vault from memories
pub async fn vault_generate_handler(
    req: HttpRequest,
    state: web::Data<AppState>,
) -> HttpResponse {
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };

    // Check write capability
    if !has_capability(&claims, "memory:write") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:write"
        }));
    }

    let vault_dir = std::env::var("MEM_HOME").unwrap_or_else(|_| "/data".to_string());
    
    // Get all projects from database
    let projects_result = sqlx::query_as::<_, (String,)>(
        "SELECT DISTINCT project FROM memories_l1 ORDER BY project",
    )
    .fetch_all(&state.pool)
    .await;

    match projects_result {
        Ok(projects) => {
            let mut generated = 0;
            let mut errors = Vec::new();

            for (project,) in projects {
                let project_vault_dir = format!("{}/vault/{}", vault_dir, project);
                
                // Create project directory
                if let Err(e) = std::fs::create_dir_all(&project_vault_dir) {
                    errors.push(format!("Failed to create {}: {}", project_vault_dir, e));
                    continue;
                }

                // Get all L1 memories for this project
                let l1s_result = sqlx::query_as::<_, (String, String, String)>(
                    "SELECT id, query_id, content FROM memories_l1 WHERE project = $1 ORDER BY updated_at DESC",
                )
                .bind(&project)
                .fetch_all(&state.pool)
                .await;

                match l1s_result {
                    Ok(l1s) => {
                        for (id, query_id, content) in l1s {
                            let filename = format!("{}/{}.md", project_vault_dir, query_id);
                            let note = format!(
                                "---\nproject: {}\nlevel: L1\nquery_id: {}\nid: {}\nupdated: {}\n---\n\n{}",
                                project,
                                query_id,
                                id,
                                chrono::Utc::now().to_rfc3339(),
                                content
                            );
                            if std::fs::write(&filename, note).is_ok() {
                                generated += 1;
                            }
                        }
                    }
                    Err(e) => {
                        errors.push(format!("Failed to fetch L1s for {}: {}", project, e));
                    }
                }
                
                // Get L2 synthesis
                let l2_result = sqlx::query_as::<_, (String,)>(
                    "SELECT content FROM memories_l2 WHERE project = $1",
                )
                .bind(&project)
                .fetch_optional(&state.pool)
                .await;

                if let Ok(Some((content,))) = l2_result {
                    let filename = format!("{}/index.md", project_vault_dir);
                    let note = format!(
                        "---\nproject: {}\nlevel: L2\ntitle: {} Synthesis\nupdated: {}\n---\n\n{}",
                        project,
                        project,
                        chrono::Utc::now().to_rfc3339(),
                        content
                    );
                    if std::fs::write(&filename, note).is_ok() {
                        generated += 1;
                    }
                }
            }

            HttpResponse::Ok().json(json!({
                "status": "generated",
                "vault_dir": format!("{}/vault", vault_dir),
                "notes_created": generated,
                "errors": errors
            }))
        }
        Err(_) => {
            HttpResponse::InternalServerError().json(json!({"error": "database_error"}))
        }
    }
}

/// GET /memory/vault — list all projects with vault browser UI
pub async fn vault_browser_handler(
    req: HttpRequest,
    query: web::Query<std::collections::HashMap<String, String>>,
    state: web::Data<AppState>,
) -> HttpResponse {
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };

    // Check read capability
    if !has_capability(&claims, "memory:read") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:read"
        }));
    }

    // If project parameter provided, return file tree for that project
    if let Some(project) = query.get("project") {
        return vault_project_tree(project, &state).await;
    }

    // Otherwise return all projects
    let result = sqlx::query_as::<_, (String,)>(
        "SELECT DISTINCT project FROM memories_l1 ORDER BY project",
    )
    .fetch_all(&state.pool)
    .await;

    match result {
        Ok(rows) => {
            let projects: Vec<serde_json::Value> = rows
                .into_iter()
                .map(|(p,)| json!({"name": p}))
                .collect();
            HttpResponse::Ok().json(json!({
                "projects": projects
            }))
        }
        Err(e) => {
            tracing::error!("Error fetching projects: {}", e);
            HttpResponse::InternalServerError().json(json!({"error": "database_error"}))
        }
    }
}

/// Helper: Build file tree for a project
async fn vault_project_tree(
    project: &str,
    state: &web::Data<AppState>,
) -> HttpResponse {
    let vault_dir = std::env::var("MEM_HOME").unwrap_or_else(|_| "/data".to_string());
    let project_path = format!("{}/vault/{}", vault_dir, project);

    match std::fs::read_dir(&project_path) {
        Ok(entries) => {
            let mut files = Vec::new();
            
            for entry in entries {
                if let Ok(e) = entry {
                    if let Ok(metadata) = e.metadata() {
                        if let Some(name) = e.file_name().to_str() {
                            if name.ends_with(".md") && !name.starts_with(".") {
                                let path = format!("{}/{}", project, name);
                                let title = name.replace(".md", "").replace("-", " ");
                                let updated = metadata
                                    .modified()
                                    .ok()
                                    .and_then(|t| {
                                        let elapsed = t.elapsed().ok()?;
                                        Some(chrono::Local::now() - chrono::Duration::from_std(elapsed).ok()?)
                                    })
                                    .map(|t| t.to_rfc3339())
                                    .unwrap_or_else(|| "unknown".to_string());
                                
                                files.push(json!({
                                    "path": path,
                                    "name": name,
                                    "title": title,
                                    "updated_at": updated
                                }));
                            }
                        }
                    }
                }
            }
            
            files.sort_by(|a, b| {
                let a_name = a["name"].as_str().unwrap_or("");
                let b_name = b["name"].as_str().unwrap_or("");
                a_name.cmp(b_name)
            });

            HttpResponse::Ok().json(json!({
                "project": project,
                "files": files
            }))
        }
        Err(_) => {
            HttpResponse::NotFound().json(json!({
                "error": "project_not_found",
                "project": project
            }))
        }
    }
}

/// GET /memory/vault/{project} — list files in project vault
pub async fn vault_project_handler(
    req: HttpRequest,
    project: web::Path<String>,
    state: web::Data<AppState>,
) -> HttpResponse {
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };

    // Check read capability
    if !has_capability(&claims, "memory:read") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:read"
        }));
    }

    let proj = project.into_inner();
    vault_project_tree(&proj, &state).await
}

/// GET /memory/vault/{project}/{file} — view markdown file
pub async fn vault_file_handler(
    req: HttpRequest,
    path: web::Path<(String, String)>,
    state: web::Data<AppState>,
) -> HttpResponse {
    let (claims, _token) = match validate_auth(&req, &state).await {
        Ok(c) => c,
        Err(e) => return e,
    };

    // Check read capability
    if !has_capability(&claims, "memory:read") {
        return HttpResponse::Forbidden().json(json!({
            "error": "forbidden",
            "reason": "missing capability: memory:read"
        }));
    }

    let (project, file) = path.into_inner();
    let vault_dir = std::env::var("MEM_HOME").unwrap_or_else(|_| "/data".to_string());
    let file_path = format!("{}/vault/{}/{}.md", vault_dir, project, file);

    // Security: prevent path traversal
    if file.contains("..") || file.contains("/") {
        return HttpResponse::BadRequest().json(json!({
            "error": "invalid_filename"
        }));
    }

    match std::fs::read_to_string(&file_path) {
        Ok(content) => {
            // Parse frontmatter and body
            let (frontmatter, body) = if content.starts_with("---") {
                let parts: Vec<&str> = content.split("---").collect();
                if parts.len() >= 3 {
                    (parts[1].to_string(), parts[2..].join("---"))
                } else {
                    ("".to_string(), content.clone())
                }
            } else {
                ("".to_string(), content.clone())
            };

            // Parse YAML frontmatter into JSON
            let metadata = if !frontmatter.is_empty() {
                frontmatter
                    .lines()
                    .filter_map(|line| {
                        let parts: Vec<&str> = line.splitn(2, ':').collect();
                        if parts.len() == 2 {
                            Some((parts[0].trim().to_string(), parts[1].trim().to_string()))
                        } else {
                            None
                        }
                    })
                    .collect::<std::collections::HashMap<_, _>>()
            } else {
                std::collections::HashMap::new()
            };

            let mut meta_json = json!({});
            for (k, v) in metadata {
                meta_json[&k] = json!(v);
            }

            HttpResponse::Ok().json(json!({
                "project": project,
                "file": format!("{}.md", file),
                "path": format!("{}/{}.md", project, file),
                "title": file.replace("-", " "),
                "metadata": meta_json,
                "content": body.trim()
            }))
        }
        Err(_) => {
            HttpResponse::NotFound().json(json!({
                "error": "file_not_found",
                "project": project,
                "file": file
            }))
        }
    }
}

/// Query temporal knowledge graph
/// 1. Find entities via semantic search
/// 2. Traverse edges from entities
/// 3. Apply temporal filtering (t_valid/t_invalid)
/// 4. Return graph with confidence scores
async fn query_temporal_graph(
    state: &web::Data<AppState>,
    params: &QueryParams,
) -> anyhow::Result<serde_json::Value> {
    // Step 1: Find entities (order by name for deterministic results)
    let entities_rows: Vec<(String, String, String)> = sqlx::query_as(
        "SELECT id, name, entity_type FROM memory_entity WHERE project_id = $1 LIMIT $2"
    )
    .bind(&params.project)
    .bind(params.limit as i32)
    .fetch_all(&state.pool)
    .await
    .unwrap_or_default();
    
    // Step 2: Traverse edges from found entities
    // NOTE: Edges will be empty until temporal schema is migrated
    let mut edges_data: Vec<(String, String, String, String, String, f32)> = Vec::new();
    
    // Try to fetch edges (will be empty if schema not migrated yet)
    for (entity_id, _name, _type_str) in &entities_rows {
        let entity_edges: Vec<(String, String, String, String, f32, Option<chrono::DateTime<chrono::Utc>>, Option<chrono::DateTime<chrono::Utc>>)> = 
            sqlx::query_as(
                "SELECT id, target_entity_id, relation_type, fact, confidence, t_valid, t_invalid FROM memory_edge WHERE project_id = $1 AND source_entity_id = $2"
            )
            .bind(&params.project)
            .bind(entity_id)
            .fetch_all(&state.pool)
            .await
            .unwrap_or_default();  // Returns empty vec if table schema doesn't match
        
        for (id, target, rel, fact, conf, t_valid, t_invalid) in entity_edges {
            // Apply temporal filtering
            let now = chrono::Utc::now();
            let valid = t_valid.as_ref().map(|t| *t <= now).unwrap_or(true);
            let not_invalid = t_invalid.as_ref().map(|t| *t > now).unwrap_or(true);
            
            if valid && not_invalid {
                edges_data.push((id, entity_id.clone(), target, rel, fact, conf));
            }
        }
    }
    
    // Step 4: Build response
    let response = json!({
        "query": params.question,
        "project": params.project,
        "entities": entities_rows.iter().map(|(id, name, etype)| json!({
            "id": id,
            "name": name,
            "type": etype
        })).collect::<Vec<_>>(),
        "edges": edges_data.iter().map(|(id, src, tgt, rel, fact, conf)| json!({
            "id": id,
            "source": src,
            "target": tgt,
            "relation": rel,
            "fact": fact,
            "confidence": conf
        })).collect::<Vec<_>>(),
        "count": json!({
            "entities": entities_rows.len(),
            "edges": edges_data.len()
        })
    });
    
    Ok(response)
}