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;
use crate::rbac::{AccessGuard, Claims as RbacClaims, builtin_role_provider, ResourceMeta, ResourceType, Verb, Visibility};

/// 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>>,
    /// RBAC Access Guard (optional, for fine-grained access control)
    pub access_guard: Option<Arc<AccessGuard>>,
}

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

/// Auth extractor — validates JWT or fallback to apikey
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),
    }
}

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

    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()
}

/// Convert JWT claims to RBAC claims for AccessGuard
fn to_rbac_claims(jwt: &JwtClaims) -> RbacClaims {
    RbacClaims::new(&jwt.sub)
        .with_groups(jwt.groups.clone().unwrap_or_default().iter().map(|s| s.as_str()).collect())
        .with_permissions(jwt.permissions.clone().unwrap_or_default().iter().map(|s| s.as_str()).collect())
}

/// Convert QueryResult to ResourceMeta for RBAC filtering
fn query_result_to_resource_meta(result: &crate::query_worker::QueryResult, project: &str) -> ResourceMeta {
    let source = result.source.as_deref().unwrap_or("unknown");
    
    // Determine resource type from source path
    let resource_type = if source.contains("SKILL-") || source.contains("/skills/") {
        ResourceType::Skill
    } else if result.level == "corpus" || result.level == "R" {
        ResourceType::Wiki  // Reference docs are wiki-like
    } else {
        ResourceType::Embedding  // L0, L1, L2 are learned embeddings
    };
    
    // Determine visibility - private if source path suggests it
    let visibility = if source.contains("/private/") || source.contains("-private") {
        Visibility::Private
    } else {
        Visibility::Public
    };
    
    ResourceMeta::new(source, resource_type, project)
        .with_visibility(visibility)
}

/// 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
    init_schema(&pool).await?;
    tracing::info!("Schema initialized");

    // 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,
        _ => {
            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)");
    }

    // Initialize RBAC AccessGuard with built-in roles
    let access_guard = {
        let role_provider = Arc::new(builtin_role_provider());
        Some(Arc::new(AccessGuard::new(role_provider)))
    };

    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,
        access_guard,
    });

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

    HttpServer::new(move || {
        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/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))
    })
    .bind(("0.0.0.0", port))?
    .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 {
    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"
        }));
    }

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

    let project = body.project.clone();
    let ingest_id = body.ingest_id.clone();
    let records: Vec<(String, String)> = body
        .records
        .iter()
        .map(|r| (r.text.clone(), body.source.clone()))
        .collect();

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

    // Create ingest job in DB
    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(&project)
    .bind(&ingest_id)
    .fetch_optional(&state.pool)
    .await;

    match job_result {
        Ok(Some(_)) => {
            // Spawn async ingest task
            let worker = state.ingest_worker.clone();
            let proj = project.clone();
            let id = ingest_id.clone();
            tokio::spawn(async move {
                if let Err(e) = worker.process_ingest(&proj, &id, records).await {
                    tracing::error!("Ingest failed: {}", e);
                }
            });

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

            // Cache the response for idempotency
            state.idempotency_store.set(ingest_id.clone(), response.clone());

            HttpResponse::Accepted().json(response)
        }
        Ok(None) => {
            // Already exists in DB (was inserted concurrently)
            let response = json!({
                "ingest_id": ingest_id,
                "status": "pending",
                "status_url": format!("/memory/ingest/{}", ingest_id)
            });
            state.idempotency_store.set(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 {
    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;
    }

    let project = body["project"].as_str().unwrap_or("knowledge").to_string();
    let text = match body["text"].as_str() {
        Some(t) => t.to_string(),
        None => return HttpResponse::BadRequest().json(json!({
            "error": "bad_request",
            "reason": "missing required field: text"
        })),
    };
    let question = body["query"].as_str()
        .unwrap_or("What are the key facts, patterns, and practices in this knowledge?")
        .to_string();
    let memory_budget = body["memory_budget"].as_u64().unwrap_or(4096) as u32;
    let chunk_size = body["chunk_size"].as_u64().unwrap_or(2000) as usize;
    let model = body["model"].as_str().unwrap_or("qwen2.5:3b-instruct").to_string();

    // Chunk the markdown
    let chunks = chunk_markdown_text(&text, 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://{}:{}", 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,
        exit_gate: false,
    };

    let config = mem_core::gated_loop::LoopConfig {
        level: mem_core::Level::L1,
        query,
        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, &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 in pgvector if non-empty
    let mut stored = false;
    if !outcome.final_memory.is_empty() {
        match state.embeddings.embed_one(&outcome.final_memory).await {
            Ok(embedding) => {
                let chunk_id = uuid::Uuid::new_v4();
                let sha = {
                    use sha2::{Digest, Sha256};
                    let mut h = Sha256::new();
                    h.update(outcome.final_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(chunk_id)
                .bind(&project)
                .bind(&outcome.final_memory)
                .bind(embedding.to_vec())
                .bind(&sha)
                .bind(format!("learn://{}", project))
                .fetch_optional(&state.pool)
                .await;

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

    HttpResponse::Ok().json(json!({
        "project": project,
        "status": "completed",
        "chunks_seen": outcome.chunks_seen,
        "chunks_used": outcome.chunks_used,
        "memory": outcome.final_memory,
        "memory_tokens": outcome.final_memory.len() / 4,
        "stored": stored,
        "model": model,
    }))
}

/// 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 {
    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/query") {
        return e;
    }

    let project = match query.get("project") {
        Some(p) => p.clone(),
        None => {
            return HttpResponse::BadRequest().json(json!({"error": "missing project parameter"}))
        }
    };

    let question = match query.get("query") {
        Some(q) => q.clone(),
        None => {
            return HttpResponse::BadRequest().json(json!({"error": "missing query parameter"}))
        }
    };

    let limit = query
        .get("limit")
        .and_then(|l| l.parse::<i64>().ok())
        .unwrap_or(10);

    let search_method = query.get("method").map(|s| s.as_str()).unwrap_or("hybrid");

    // Get semantic results from pgvector (always run)
    let mut semantic_results = match state.query_worker.query(&project, &question, Some(50)).await {
        Ok(results) => results,
        Err(e) => {
            tracing::error!("Semantic search failed: {}", e);
            return HttpResponse::InternalServerError().json(json!({"error": "semantic_search_failed"}));
        }
    };

    // M3.8: Optimize search results if optimizer is available
    semantic_results = optimize_search_results(semantic_results, state.optimizer_service.as_ref()).await;

    // RBAC: Filter results by access control
    if let Some(guard) = &state.access_guard {
        let rbac_claims = to_rbac_claims(&claims);
        let resources: Vec<ResourceMeta> = semantic_results
            .iter()
            .map(|r| query_result_to_resource_meta(r, &project))
            .collect();
        
        let decisions = guard.check_access_batch(&rbac_claims, &resources, Verb::Read).await;
        
        // Keep only allowed results
        semantic_results = semantic_results
            .into_iter()
            .zip(decisions.iter())
            .filter(|(_, decision)| decision.is_allowed())
            .map(|(result, _)| result)
            .collect();
        
        tracing::debug!(
            "RBAC filtered {} results for user {}",
            decisions.iter().filter(|d| d.is_denied()).count(),
            claims.sub
        );
    }

    // Handle different search methods
    match search_method {
        "semantic" => {
            // Return only semantic results
            let top_results = semantic_results
                .into_iter()
                .take(limit as usize)
                .collect::<Vec<_>>();
            
            HttpResponse::Ok().json(json!({
                "query": question,
                "project": project,
                "method": "semantic",
                "results": top_results
            }))
        }
        "hybrid" => {
            // If OpenSearch is available, run hybrid search
            if let Some(os_client) = &state.opensearch_client {
                // Convert semantic results to format expected by hybrid_search
                let sem_results: Vec<(String, f32, String, String, Vec<String>)> = semantic_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(&question, sem_results, &token, limit as usize, &weights).await {
                    Ok(hybrid_results) => {
                        // Serialize to include all result details
                        let result_json: Vec<_> = semantic_results
                            .into_iter()
                            .take(limit as usize)
                            .map(|r| json!({
                                "level": r.level,
                                "score": r.score,
                                "text": r.text,
                                "source": r.source,
                                "provenance": r.provenance
                            }))
                            .collect();
                        
                        HttpResponse::Ok().json(json!({
                            "query": question,
                            "project": project,
                            "method": "hybrid",
                            "results": result_json
                        }))
                    }
                    Err(e) => {
                        tracing::warn!("Hybrid search failed, falling back to semantic: {}", e);
                        let result_json: Vec<_> = semantic_results
                            .into_iter()
                            .take(limit as usize)
                            .map(|r| json!({
                                "level": r.level,
                                "score": r.score,
                                "text": r.text,
                                "source": r.source,
                                "provenance": r.provenance
                            }))
                            .collect();
                        
                        HttpResponse::Ok().json(json!({
                            "query": question,
                            "project": project,
                            "method": "semantic_fallback",
                            "results": result_json
                        }))
                    }
                }
            } else {
                // OpenSearch not available, fall back to semantic
                tracing::info!("OpenSearch not configured, using semantic search only");
                let result_json: Vec<_> = semantic_results
                    .into_iter()
                    .take(limit as usize)
                    .map(|r| json!({
                        "level": r.level,
                        "score": r.score,
                        "text": r.text,
                        "source": r.source,
                        "provenance": r.provenance
                    }))
                    .collect();
                
                HttpResponse::Ok().json(json!({
                    "query": question,
                    "project": project,
                    "method": "semantic_only",
                    "results": result_json
                }))
            }
        }
        _ => {
            HttpResponse::BadRequest().json(json!({
                "error": "invalid_search_method",
                "valid_methods": ["semantic", "hybrid"]
            }))
        }
    }
}

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

    // RBAC: Check project-level access
    if let Some(guard) = &state.access_guard {
        let rbac_claims = to_rbac_claims(&claims);
        let project_resource = ResourceMeta::new(&project, ResourceType::Project, &project);
        
        if !guard.can_read(&rbac_claims, &project_resource).await {
            tracing::warn!(
                "RBAC denied access to project '{}' for user '{}'",
                project, claims.sub
            );
            return HttpResponse::Forbidden().json(json!({
                "error": "forbidden",
                "reason": format!("access denied to project '{}'", project)
            }));
        }
    }

    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
            }))
        }
    }
}