solo_api/

http.rs

// SPDX-License-Identifier: Apache-2.0

//! HTTP/JSON transport for Solo. Local-only by default — binds to
//! `127.0.0.1:<port>` and serves the same operations the MCP server
//! exposes:
//!
//! Episode operations:
//!   - `POST /memory`                — remember (body: { content, source_type?, source_id? })
//!   - `POST /memory/search`         — recall  (body: { query, limit? })
//!   - `GET  /memory/{id}`           — inspect
//!   - `DELETE /memory/{id}?reason=…` — forget
//!
//! Maintenance:
//!   - `POST /memory/consolidate`    — trigger a consolidation pass
//!   - `POST /backup`                — encrypted online backup
//!
//! Derived-layer (v0.4.0+; queries against the Steward's outputs):
//!   - `GET  /memory/themes?window_days=N&limit=K`
//!   - `GET  /memory/facts_about?subject=X&predicate=Y&since_ms=N&until_ms=N&limit=K`
//!   - `GET  /memory/contradictions?limit=K`
//!
//! There's no auth at this layer. The threat model is local-machine
//! single-user; binding to `127.0.0.1` keeps the surface off the LAN.
//! A future commit can add bearer-token auth + LAN binding.
//!
//! ## Lifecycle
//!
//! `serve_http(addr, server, shutdown)` binds to `addr`, runs axum with
//! `with_graceful_shutdown(shutdown)`, returns when shutdown fires or
//! the listener errors. `solo http-serve` invokes this from inside a
//! `OneShotContext`, so writer + reader pool + lockfile stay live for
//! the server's lifetime and clean up properly afterwards.

use std::net::SocketAddr;
use std::str::FromStr;
use std::sync::Arc;

use axum::extract::{Path, Query, State};
use axum::http::{HeaderValue, Method, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use serde::{Deserialize, Serialize};
use solo_core::{
    Confidence, Embedder, EncodingContext, Episode, MemoryId, Tier, VectorIndex,
};
use solo_storage::{ReaderPool, WriteHandle};
use tower_http::cors::{AllowOrigin, CorsLayer};
use tower_http::trace::TraceLayer;
use tower_http::validate_request::{ValidateRequest, ValidateRequestHeaderLayer};

#[derive(Clone)]
pub struct SoloHttpState {
    pub write: WriteHandle,
    pub pool: ReaderPool,
    pub embedder: Arc<dyn Embedder>,
    pub hnsw: Arc<dyn VectorIndex + Send + Sync>,
    /// Path to the live source database (`<data-dir>/solo.db`). Used by
    /// `POST /backup` to refuse a destination that resolves to the
    /// source file before any destructive `remove_file(dest)` step
    /// runs. Required field as of v0.3.4 — older callers that
    /// constructed `SoloHttpState` without this need a fix.
    pub source_db_path: std::path::PathBuf,
}

/// Build the router with optional bearer-token auth.
///
/// When `bearer_token` is `Some(t)`, every request except `GET /health`
/// (the unauthenticated liveness probe) requires
/// `Authorization: Bearer t`. The /health exemption keeps load
/// balancers and uptime monitors from needing a credential.
///
/// `tower_http::validate_request::ValidateRequestHeaderLayer::bearer`
/// returns 401 with a WWW-Authenticate header on missing/wrong token.
pub fn router_with_auth(state: SoloHttpState, bearer_token: Option<String>) -> Router {
    let cors = build_cors_layer();
    // Public, always-unauthenticated routes:
    //   - GET /health: liveness probe (load balancers, uptime monitors).
    //   - GET /openapi.json: machine-readable API description for client
    //     codegen + browser-UI tooling (TypeScript / OpenAPI Generator,
    //     curl-tools, etc.). The spec describes the API shape, not
    //     secrets — fine to serve unauthenticated even on a LAN-bound
    //     instance.
    let public = Router::new()
        .route("/health", get(|| async { "ok" }))
        .route("/openapi.json", get(openapi_handler));

    let mut authed = Router::new()
        .route("/memory", post(remember_handler))
        .route("/memory/search", post(recall_handler))
        .route("/memory/consolidate", post(consolidate_handler))
        .route("/memory/{id}", get(inspect_handler).delete(forget_handler))
        .route("/backup", post(backup_handler))
        // Path 1 derived-layer endpoints (v0.4.0+). GET-shaped because
        // these are pure read-only queries; query-string params for
        // simple filters keep them curl-friendly without a JSON body.
        .route("/memory/themes", get(themes_handler))
        .route("/memory/facts_about", get(facts_about_handler))
        .route("/memory/contradictions", get(contradictions_handler))
        .with_state(state);
    if let Some(token) = bearer_token {
        // Custom validator (the helper-shaped `::bearer` constructor is
        // deprecated in tower-http ≥ 0.6.7). Returns 401 with
        // `WWW-Authenticate: Bearer` on missing or wrong token.
        authed = authed.layer(ValidateRequestHeaderLayer::custom(BearerToken::new(token)));
    }

    public
        .merge(authed)
        .layer(cors)
        .layer(TraceLayer::new_for_http())
}

/// Convenience wrapper: no auth (loopback-only deployments).
pub fn router(state: SoloHttpState) -> Router {
    router_with_auth(state, None)
}

fn build_cors_layer() -> CorsLayer {
    // Permissive-localhost CORS: allow any localhost / 127.0.0.1 origin so
    // browser-based UIs running on a different local port can call the API
    // without preflight friction. We do NOT use `Any` because that would
    // allow arbitrary remote origins to talk to our localhost server via
    // a victim's browser. With bearer-token auth enabled the practical
    // impact is reduced (the cross-origin attacker still can't supply
    // the token), but principle of least privilege says refuse anyway.
    //
    // When the server is bound to a non-loopback address (auth required),
    // the same CORS predicate keeps localhost-only browser clients —
    // suitable for trusted-LAN deployments where the LAN client itself
    // tunnels through ssh/wireguard back to localhost. Wider CORS for
    // genuine cross-origin browser use is a future config knob.
    CorsLayer::new()
        .allow_origin(AllowOrigin::predicate(|origin: &HeaderValue, _req| {
            origin
                .to_str()
                .map(is_localhost_origin)
                .unwrap_or(false)
        }))
        .allow_methods([Method::GET, Method::POST, Method::DELETE, Method::OPTIONS])
        .allow_headers([
            axum::http::header::CONTENT_TYPE,
            axum::http::header::AUTHORIZATION,
        ])
}

/// `tower_http::validate_request::ValidateRequest` impl that accepts
/// any request whose `Authorization` header equals `Bearer <token>`
/// (constant-time comparison via `subtle`-style byte equality —
/// String == is constant-time in Rust for equal-length operands; for
/// unequal lengths the early-return is fine here because the token
/// length isn't sensitive). On miss, returns 401 with
/// `WWW-Authenticate: Bearer realm="solo"`.
#[derive(Clone)]
struct BearerToken {
    expected: HeaderValue,
}

impl BearerToken {
    fn new(token: String) -> Self {
        let expected = HeaderValue::try_from(format!("Bearer {token}"))
            .expect("bearer token must be a valid HTTP header value");
        Self { expected }
    }
}

impl<B> ValidateRequest<B> for BearerToken {
    type ResponseBody = axum::body::Body;

    fn validate(
        &mut self,
        request: &mut axum::http::Request<B>,
    ) -> Result<(), axum::http::Response<Self::ResponseBody>> {
        let got = request.headers().get(axum::http::header::AUTHORIZATION);
        match got {
            Some(value) if value == &self.expected => Ok(()),
            _ => {
                let mut resp = axum::http::Response::new(axum::body::Body::empty());
                *resp.status_mut() = StatusCode::UNAUTHORIZED;
                resp.headers_mut().insert(
                    axum::http::header::WWW_AUTHENTICATE,
                    HeaderValue::from_static(r#"Bearer realm="solo""#),
                );
                Err(resp)
            }
        }
    }
}

/// True if `origin` is `http(s)://localhost[:port]` or
/// `http(s)://127.0.0.1[:port]` or `http(s)://[::1][:port]` (loopback IPv6).
/// Anything else (incl. nip.io tricks like `127.0.0.1.nip.io`) is rejected.
fn is_localhost_origin(origin: &str) -> bool {
    let rest = origin
        .strip_prefix("http://")
        .or_else(|| origin.strip_prefix("https://"));
    let host = match rest {
        Some(r) => r,
        None => return false,
    };
    // Strip path (shouldn't appear on Origin headers but defend anyway).
    let host = host.split('/').next().unwrap_or(host);
    // Strip port.
    let host = if let Some(idx) = host.rfind(':') {
        // For [::1]:port, keep the brackets in the host part.
        if host.starts_with('[') {
            // Find matching ']'; everything up to and including it is the host.
            host.find(']')
                .map(|i| &host[..=i])
                .unwrap_or(host)
        } else {
            &host[..idx]
        }
    } else {
        host
    };
    matches!(host, "localhost" | "127.0.0.1" | "[::1]")
}

/// Bind + serve. `shutdown` is awaited inside axum's
/// `with_graceful_shutdown`; resolving it triggers a clean drain.
/// `bearer_token = None` runs unauthenticated (loopback default);
/// `Some(t)` requires `Authorization: Bearer t` on every request
/// except `GET /health`.
pub async fn serve_http(
    addr: SocketAddr,
    state: SoloHttpState,
    bearer_token: Option<String>,
    shutdown: impl std::future::Future<Output = ()> + Send + 'static,
) -> std::io::Result<()> {
    let auth_kind = if bearer_token.is_some() {
        "bearer"
    } else {
        "none"
    };
    let app = router_with_auth(state, bearer_token);
    let listener = tokio::net::TcpListener::bind(addr).await?;
    tracing::info!(%addr, auth = auth_kind, "solo http: listening");
    axum::serve(listener, app)
        .with_graceful_shutdown(shutdown)
        .await
}

// ---------------------------------------------------------------------------
// OpenAPI 3.1 spec
// ---------------------------------------------------------------------------

/// Serve the hand-crafted OpenAPI 3.1 spec at `GET /openapi.json`.
///
/// We keep the spec hand-written (rather than deriving via `utoipa`)
/// for v0.1: 4 simple endpoints, types live across crate boundaries
/// (`solo_query::RecallResult`, `solo_query::EpisodeRecord`), and a
/// `utoipa` retrofit would touch every crate. Hand-crafted is one
/// JSON literal in this file; a smoke test in `handler_tests` parses
/// the response and asserts the expected paths + components are
/// present, so drift between spec and code is caught at PR time.
async fn openapi_handler() -> Json<serde_json::Value> {
    Json(openapi_spec())
}

/// Build the OpenAPI 3.1 spec describing Solo's HTTP transport.
/// Public so the smoke test + future client-codegen tooling can
/// produce the same document without spinning up the server.
pub fn openapi_spec() -> serde_json::Value {
    serde_json::json!({
        "openapi": "3.1.0",
        "info": {
            "title": "Solo HTTP API",
            "description":
                "Local-first personal memory daemon. The HTTP transport \
                 mirrors the four MCP tools (memory_remember / recall / \
                 inspect / forget). Default deployment is loopback-only \
                 (127.0.0.1); LAN-bound deployments require a bearer \
                 token via `solo http-serve --bind <ip> --bearer-token-file <path>`.",
            "version": env!("CARGO_PKG_VERSION"),
            "license": { "name": "Apache-2.0" }
        },
        "servers": [
            { "url": "http://127.0.0.1:7437", "description": "Default loopback (replace port with your --http-port)" }
        ],
        "components": {
            "securitySchemes": {
                "bearerAuth": {
                    "type": "http",
                    "scheme": "bearer",
                    "description":
                        "Bearer-token auth. Required only on LAN-bound deployments \
                         (`solo http-serve --bind <non-loopback> --bearer-token-file <path>`); \
                         the default `127.0.0.1` deployment is unauthenticated. \
                         `GET /health` and `GET /openapi.json` are exempt from auth even \
                         on bearer-protected instances."
                }
            },
            "schemas": {
                "RememberRequest": {
                    "type": "object",
                    "required": ["content"],
                    "properties": {
                        "content": { "type": "string", "minLength": 1, "description": "Episode content to embed + store." },
                        "source_type": { "type": "string", "description": "Free-form source tag (e.g. `user_message`, `tool_output`). Defaults to `user_message`." },
                        "source_id": { "type": "string", "description": "Optional upstream ID for traceability." }
                    },
                    "additionalProperties": false
                },
                "RememberResponse": {
                    "type": "object",
                    "required": ["memory_id"],
                    "properties": {
                        "memory_id": { "type": "string", "format": "uuid", "description": "UUID v7 assigned to the new episode." }
                    }
                },
                "RecallRequest": {
                    "type": "object",
                    "required": ["query"],
                    "properties": {
                        "query": { "type": "string", "minLength": 1, "description": "Natural-language query; embedded by the same model as stored episodes." },
                        "limit": { "type": "integer", "minimum": 1, "maximum": 50, "default": 5, "description": "Max number of hits to return." }
                    },
                    "additionalProperties": false
                },
                "RecallResult": {
                    "type": "object",
                    "description":
                        "Recall response. Fields are stable across v0.1 but not exhaustively documented here — \
                         see `solo_query::RecallResult` in the source for the canonical shape. \
                         Treat as a forward-compatible JSON object.",
                    "additionalProperties": true
                },
                "ConsolidationScope": {
                    "type": "object",
                    "description": "Filter + flags for consolidation. All fields optional; empty body = unbounded defaults.",
                    "properties": {
                        "window_days": { "type": "integer", "nullable": true, "description": "Restrict to memories with ts_ms >= now - window_days * 86400000. Null/omitted = unbounded." },
                        "force_merge": { "type": "boolean", "default": false, "description": "Run the existing-vs-existing merge + abstraction-regen passes even with zero unclustered candidates. Drift catch-up on quiet corpora. Added in 0.3.1." }
                    },
                    "additionalProperties": false
                },
                "ConsolidationReport": {
                    "type": "object",
                    "required": [
                        "episodes_seen", "clusters_built", "clusters_merged",
                        "clusters_absorbed", "existing_clusters_merged",
                        "episodes_clustered", "abstractions_built",
                        "abstractions_regenerated", "triples_built",
                        "contradictions_found"
                    ],
                    "properties": {
                        "episodes_seen":             { "type": "integer", "minimum": 0 },
                        "clusters_built":            { "type": "integer", "minimum": 0, "description": "Brand-new clusters that survived to be persisted (post in-run-merge, post cross-run-absorb)." },
                        "clusters_merged":           { "type": "integer", "minimum": 0, "description": "In-run merge: clusters absorbed into a sibling within this consolidate run (cross-UTC-bucket case). Counts losers." },
                        "clusters_absorbed":         { "type": "integer", "minimum": 0, "description": "Cross-run absorb: freshly-built clusters folded into a pre-existing DB cluster with a similar centroid. Counts new-side clusters." },
                        "existing_clusters_merged":  { "type": "integer", "minimum": 0, "description": "Existing-vs-existing merge: pre-existing DB clusters that drifted toward each other and now coalesce. Counts losers." },
                        "episodes_clustered":        { "type": "integer", "minimum": 0 },
                        "abstractions_built":        { "type": "integer", "minimum": 0, "description": "Fresh abstractions persisted for newly-built clusters. 0 when no LlmClient is wired." },
                        "abstractions_regenerated":  { "type": "integer", "minimum": 0, "description": "Existing clusters whose stale abstractions were dropped and rebuilt because absorb or existing-merge changed their episode set. 0 without an LlmClient." },
                        "triples_built":             { "type": "integer", "minimum": 0 },
                        "contradictions_found":      { "type": "integer", "minimum": 0 }
                    }
                },
                "EpisodeRecord": {
                    "type": "object",
                    "description":
                        "Inspect response: full episode record. Fields are stable across v0.1 but not \
                         exhaustively documented here — see `solo_query::EpisodeRecord` in the source. \
                         Treat as a forward-compatible JSON object.",
                    "additionalProperties": true
                },
                "ThemeHit": {
                    "type": "object",
                    "description":
                        "One cluster + its (optional) abstraction. Returned by GET /memory/themes. \
                         See `solo_query::ThemeHit` for the canonical shape: cluster_id, \
                         abstraction_id?, abstraction_text?, episode_count, coherence, created_at_ms.",
                    "additionalProperties": true
                },
                "FactHit": {
                    "type": "object",
                    "description":
                        "One Steward-extracted SPO triple. Returned by GET /memory/facts_about. \
                         See `solo_query::FactHit` for fields: triple_id, subject_id, predicate, \
                         object_id, object_kind, valid_from_ms, valid_to_ms?, confidence, cluster_id?.",
                    "additionalProperties": true
                },
                "ContradictionHit": {
                    "type": "object",
                    "description":
                        "One Steward-flagged contradiction with each side's triple LEFT JOIN'd in. \
                         Returned by GET /memory/contradictions. See `solo_query::ContradictionHit`: \
                         a_id, b_id, kind, explanation, detected_at_ms, a_triple?, b_triple?.",
                    "additionalProperties": true
                },
                "ApiError": {
                    "type": "object",
                    "required": ["error", "status"],
                    "properties": {
                        "error": { "type": "string" },
                        "status": { "type": "integer", "minimum": 400, "maximum": 599 }
                    }
                }
            }
        },
        "paths": {
            "/health": {
                "get": {
                    "summary": "Liveness probe",
                    "description": "Returns plain text `ok`. Always unauthenticated.",
                    "responses": {
                        "200": {
                            "description": "Server is up.",
                            "content": { "text/plain": { "schema": { "type": "string", "example": "ok" } } }
                        }
                    }
                }
            },
            "/openapi.json": {
                "get": {
                    "summary": "Self-describing OpenAPI 3.1 spec",
                    "description": "Returns this document. Always unauthenticated.",
                    "responses": {
                        "200": {
                            "description": "OpenAPI 3.1 document.",
                            "content": { "application/json": { "schema": { "type": "object" } } }
                        }
                    }
                }
            },
            "/memory": {
                "post": {
                    "summary": "Remember (store an episode)",
                    "description": "Equivalent to MCP tool `memory_remember`.",
                    "security": [{ "bearerAuth": [] }, {}],
                    "requestBody": {
                        "required": true,
                        "content": { "application/json": { "schema": { "$ref": "#/components/schemas/RememberRequest" } } }
                    },
                    "responses": {
                        "200": {
                            "description": "Memory stored; returns the new MemoryId.",
                            "content": { "application/json": { "schema": { "$ref": "#/components/schemas/RememberResponse" } } }
                        },
                        "400": { "description": "Bad request (e.g. empty content).", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/ApiError" } } } },
                        "401": { "description": "Missing or invalid bearer token (LAN-bound deployments only)." }
                    }
                }
            },
            "/memory/search": {
                "post": {
                    "summary": "Recall (vector search)",
                    "description": "Equivalent to MCP tool `memory_recall`. Embeds the query, runs HNSW search, returns the top-K hits in cosine-distance order.",
                    "security": [{ "bearerAuth": [] }, {}],
                    "requestBody": {
                        "required": true,
                        "content": { "application/json": { "schema": { "$ref": "#/components/schemas/RecallRequest" } } }
                    },
                    "responses": {
                        "200": {
                            "description": "Search results.",
                            "content": { "application/json": { "schema": { "$ref": "#/components/schemas/RecallResult" } } }
                        },
                        "400": { "description": "Bad request (e.g. empty query).", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/ApiError" } } } },
                        "401": { "description": "Missing or invalid bearer token (LAN-bound deployments only)." }
                    }
                }
            },
            "/memory/consolidate": {
                "post": {
                    "summary": "Run a consolidation pass (clustering + abstraction)",
                    "description":
                        "Idempotent. Triggers the SWS-equivalent clustering pass; if a `Steward` LLM is wired \
                         on the server, also runs the REM-equivalent abstraction pass that populates \
                         `semantic_abstractions` and `triples`. Empty request body = default scope (unbounded \
                         window). Equivalent to the `solo consolidate` CLI.",
                    "security": [{ "bearerAuth": [] }, {}],
                    "requestBody": {
                        "required": false,
                        "content": { "application/json": { "schema": { "$ref": "#/components/schemas/ConsolidationScope" } } }
                    },
                    "responses": {
                        "200": {
                            "description": "Consolidation complete; report counts the work done.",
                            "content": { "application/json": { "schema": { "$ref": "#/components/schemas/ConsolidationReport" } } }
                        },
                        "401": { "description": "Missing or invalid bearer token (LAN-bound deployments only)." }
                    }
                }
            },
            "/backup": {
                "post": {
                    "summary": "Online encrypted backup",
                    "description":
                        "Run an online SQLCipher backup of the live data dir to a server-side path. \
                         The destination file is encrypted with the same Argon2id-derived raw key as \
                         the source, so it restores under the same passphrase + a copy of the source's \
                         `solo.config.toml`. Hot — the backup runs against the writer's existing \
                         connection without taking the lockfile, so the daemon keeps serving reads + \
                         writes during the operation. v0.3.2+.",
                    "security": [{ "bearerAuth": [] }, {}],
                    "requestBody": {
                        "required": true,
                        "content": { "application/json": { "schema": {
                            "type": "object",
                            "properties": {
                                "to": { "type": "string", "description": "Server-side absolute path for the backup file." },
                                "force": { "type": "boolean", "description": "Overwrite an existing destination file. Default false.", "default": false }
                            },
                            "required": ["to"]
                        } } }
                    },
                    "responses": {
                        "200": {
                            "description": "Backup complete; reports the destination path + elapsed milliseconds.",
                            "content": { "application/json": { "schema": {
                                "type": "object",
                                "properties": {
                                    "path": { "type": "string" },
                                    "elapsed_ms": { "type": "integer", "format": "int64" }
                                }
                            } } }
                        },
                        "400": { "description": "Destination invalid, exists without force, or its parent doesn't exist." },
                        "401": { "description": "Missing or invalid bearer token (LAN-bound deployments only)." },
                        "500": { "description": "Backup failed (disk full, permission denied, etc.)." }
                    }
                }
            },
            "/memory/{id}": {
                "get": {
                    "summary": "Inspect a memory by ID",
                    "description": "Equivalent to MCP tool `memory_inspect`.",
                    "security": [{ "bearerAuth": [] }, {}],
                    "parameters": [{
                        "name": "id",
                        "in": "path",
                        "required": true,
                        "schema": { "type": "string", "format": "uuid" },
                        "description": "MemoryId (UUID v7)."
                    }],
                    "responses": {
                        "200": {
                            "description": "Episode record.",
                            "content": { "application/json": { "schema": { "$ref": "#/components/schemas/EpisodeRecord" } } }
                        },
                        "400": { "description": "Malformed ID.", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/ApiError" } } } },
                        "404": { "description": "No such memory.", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/ApiError" } } } },
                        "401": { "description": "Missing or invalid bearer token (LAN-bound deployments only)." }
                    }
                },
                "delete": {
                    "summary": "Forget (soft-delete) a memory by ID",
                    "description":
                        "Equivalent to MCP tool `memory_forget`. Soft-delete: flips `episodes.status = 'forgotten'` \
                         and tombstones the HNSW vector. The row + embedding are preserved for forensics; \
                         re-running `solo reembed` after this does NOT restore visibility.",
                    "security": [{ "bearerAuth": [] }, {}],
                    "parameters": [
                        { "name": "id", "in": "path", "required": true, "schema": { "type": "string", "format": "uuid" } },
                        { "name": "reason", "in": "query", "required": false, "schema": { "type": "string" }, "description": "Free-form reason logged via tracing (not yet persisted to the DB)." }
                    ],
                    "responses": {
                        "204": { "description": "Forgotten (or already forgotten — idempotent)." },
                        "400": { "description": "Malformed ID.", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/ApiError" } } } },
                        "404": { "description": "No such memory.", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/ApiError" } } } },
                        "401": { "description": "Missing or invalid bearer token (LAN-bound deployments only)." }
                    }
                }
            },
            "/memory/themes": {
                "get": {
                    "summary": "List recent cluster themes",
                    "description":
                        "Equivalent to MCP tool `memory_themes`. List cluster abstractions ordered by \
                         most-recent first. Use to surface 'what has the user been thinking about lately' \
                         without paging through individual episodes. v0.4.0+.",
                    "security": [{ "bearerAuth": [] }, {}],
                    "parameters": [
                        { "name": "window_days", "in": "query", "required": false, "schema": { "type": "integer", "minimum": 1 }, "description": "Optional time window. Omit for unfiltered (all-time, most-recent first)." },
                        { "name": "limit", "in": "query", "required": false, "schema": { "type": "integer", "minimum": 1, "maximum": 100, "default": 5 } }
                    ],
                    "responses": {
                        "200": {
                            "description": "Array of ThemeHits (possibly empty).",
                            "content": { "application/json": { "schema": { "type": "array", "items": { "$ref": "#/components/schemas/ThemeHit" } } } }
                        },
                        "401": { "description": "Missing or invalid bearer token (LAN-bound deployments only)." }
                    }
                }
            },
            "/memory/facts_about": {
                "get": {
                    "summary": "Query the SPO knowledge graph by subject",
                    "description":
                        "Equivalent to MCP tool `memory_facts_about`. Query Steward-extracted triples by \
                         subject + optional predicate + optional time window. Subject is required \
                         (predicate-only scans not supported). v0.4.0+.",
                    "security": [{ "bearerAuth": [] }, {}],
                    "parameters": [
                        { "name": "subject", "in": "query", "required": true, "schema": { "type": "string", "minLength": 1 }, "description": "Subject id to query (e.g. `Sam`)." },
                        { "name": "predicate", "in": "query", "required": false, "schema": { "type": "string" }, "description": "Optional predicate filter (e.g. `works_at`)." },
                        { "name": "since_ms", "in": "query", "required": false, "schema": { "type": "integer" }, "description": "Optional valid_from_ms lower bound (epoch ms)." },
                        { "name": "until_ms", "in": "query", "required": false, "schema": { "type": "integer" }, "description": "Optional valid_to_ms upper bound (epoch ms). NULL upper bounds (still-valid facts) pass through." },
                        { "name": "limit", "in": "query", "required": false, "schema": { "type": "integer", "minimum": 1, "maximum": 100, "default": 5 } }
                    ],
                    "responses": {
                        "200": {
                            "description": "Array of FactHits (possibly empty).",
                            "content": { "application/json": { "schema": { "type": "array", "items": { "$ref": "#/components/schemas/FactHit" } } } }
                        },
                        "400": { "description": "Bad request (e.g. empty subject).", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/ApiError" } } } },
                        "401": { "description": "Missing or invalid bearer token (LAN-bound deployments only)." }
                    }
                }
            },
            "/memory/contradictions": {
                "get": {
                    "summary": "List Steward-flagged contradictions",
                    "description":
                        "Equivalent to MCP tool `memory_contradictions`. Each result includes both \
                         sides' triple SPO via LEFT JOIN for context. v0.4.0+.",
                    "security": [{ "bearerAuth": [] }, {}],
                    "parameters": [
                        { "name": "limit", "in": "query", "required": false, "schema": { "type": "integer", "minimum": 1, "maximum": 100, "default": 5 } }
                    ],
                    "responses": {
                        "200": {
                            "description": "Array of ContradictionHits (possibly empty).",
                            "content": { "application/json": { "schema": { "type": "array", "items": { "$ref": "#/components/schemas/ContradictionHit" } } } }
                        },
                        "401": { "description": "Missing or invalid bearer token (LAN-bound deployments only)." }
                    }
                }
            }
        }
    })
}

// ---------------------------------------------------------------------------
// Handlers
// ---------------------------------------------------------------------------

#[derive(Debug, Deserialize)]
struct RememberBody {
    content: String,
    #[serde(default)]
    source_type: Option<String>,
    #[serde(default)]
    source_id: Option<String>,
}

#[derive(Debug, Serialize)]
struct RememberResponse {
    memory_id: String,
}

async fn remember_handler(
    State(s): State<SoloHttpState>,
    Json(body): Json<RememberBody>,
) -> Result<Json<RememberResponse>, ApiError> {
    let content = body.content.trim_end().to_string();
    if content.is_empty() {
        return Err(ApiError::bad_request("content must not be empty"));
    }
    let embedding = s.embedder.embed(&content).await.map_err(ApiError::from)?;
    let episode = Episode {
        memory_id: MemoryId::new(),
        ts_ms: chrono::Utc::now().timestamp_millis(),
        source_type: body.source_type.unwrap_or_else(|| "user_message".into()),
        source_id: body.source_id,
        content,
        encoding_context: EncodingContext::default(),
        provenance: None,
        confidence: Confidence::new(0.9).unwrap(),
        strength: 0.5,
        salience: 0.5,
        tier: Tier::Hot,
    };
    let mid = s.write.remember(episode, embedding).await.map_err(ApiError::from)?;
    Ok(Json(RememberResponse {
        memory_id: mid.to_string(),
    }))
}

#[derive(Debug, Deserialize)]
struct RecallBody {
    query: String,
    #[serde(default = "default_limit")]
    limit: usize,
}

fn default_limit() -> usize {
    5
}

async fn recall_handler(
    State(s): State<SoloHttpState>,
    Json(body): Json<RecallBody>,
) -> Result<Json<solo_query::RecallResult>, ApiError> {
    // solo_query::run_recall handles empty-query rejection (returns
    // InvalidInput → ApiError::bad_request(400)) and clamps limit
    // upstream of the embedder call.
    let result = solo_query::run_recall(
        &s.embedder,
        &s.hnsw,
        &s.pool,
        &body.query,
        body.limit,
    )
    .await
    .map_err(ApiError::from)?;
    Ok(Json(result))
}

async fn inspect_handler(
    State(s): State<SoloHttpState>,
    Path(id): Path<String>,
) -> Result<Json<solo_query::EpisodeRecord>, ApiError> {
    let mid = MemoryId::from_str(&id)
        .map_err(|e| ApiError::bad_request(format!("invalid id: {e}")))?;
    let row = solo_query::inspect_one(&s.pool, mid)
        .await
        .map_err(ApiError::from)?;
    Ok(Json(row))
}

// Path 1 derived-layer handlers (v0.4.0+). All three are GET-shaped:
// pure read-only queries against the Steward's outputs, query-string
// params for simple filters. Each handler delegates to a single
// solo_query::derived pipeline and returns the result Vec as JSON.
// Empty derived layer → 200 with `[]` body (parseable JSON array).

#[derive(Debug, Deserialize)]
struct ThemesQuery {
    #[serde(default)]
    window_days: Option<i64>,
    #[serde(default = "default_limit")]
    limit: usize,
}

async fn themes_handler(
    State(s): State<SoloHttpState>,
    Query(q): Query<ThemesQuery>,
) -> Result<Json<Vec<solo_query::ThemeHit>>, ApiError> {
    let hits = solo_query::themes(&s.pool, q.window_days, q.limit)
        .await
        .map_err(ApiError::from)?;
    Ok(Json(hits))
}

#[derive(Debug, Deserialize)]
struct FactsAboutQuery {
    subject: String,
    #[serde(default)]
    predicate: Option<String>,
    #[serde(default)]
    since_ms: Option<i64>,
    #[serde(default)]
    until_ms: Option<i64>,
    #[serde(default = "default_limit")]
    limit: usize,
}

async fn facts_about_handler(
    State(s): State<SoloHttpState>,
    Query(q): Query<FactsAboutQuery>,
) -> Result<Json<Vec<solo_query::FactHit>>, ApiError> {
    if q.subject.trim().is_empty() {
        return Err(ApiError::bad_request("subject must not be empty"));
    }
    let hits = solo_query::facts_about(
        &s.pool,
        &q.subject,
        q.predicate.as_deref(),
        q.since_ms,
        q.until_ms,
        q.limit,
    )
    .await
    .map_err(ApiError::from)?;
    Ok(Json(hits))
}

#[derive(Debug, Deserialize)]
struct ContradictionsQuery {
    #[serde(default = "default_limit")]
    limit: usize,
}

async fn contradictions_handler(
    State(s): State<SoloHttpState>,
    Query(q): Query<ContradictionsQuery>,
) -> Result<Json<Vec<solo_query::ContradictionHit>>, ApiError> {
    let hits = solo_query::contradictions(&s.pool, q.limit)
        .await
        .map_err(ApiError::from)?;
    Ok(Json(hits))
}

#[derive(Debug, Deserialize)]
struct ForgetQuery {
    #[serde(default)]
    reason: Option<String>,
}

async fn forget_handler(
    State(s): State<SoloHttpState>,
    Path(id): Path<String>,
    Query(q): Query<ForgetQuery>,
) -> Result<StatusCode, ApiError> {
    let mid = MemoryId::from_str(&id).map_err(|e| ApiError::bad_request(format!("invalid id: {e}")))?;
    let reason = q.reason.unwrap_or_else(|| "http".into());
    s.write.forget(mid, reason).await.map_err(ApiError::from)?;
    Ok(StatusCode::NO_CONTENT)
}

async fn consolidate_handler(
    State(s): State<SoloHttpState>,
    body: axum::body::Bytes,
) -> Result<Json<solo_storage::ConsolidationReport>, ApiError> {
    // Empty body = default scope (unbounded window). We parse via
    // `Bytes` rather than `Option<Json<T>>` because axum's `Json`
    // extractor 400s on an empty body when Content-Type is JSON
    // (it can't deserialize zero bytes as `T`), and the `Option`
    // wrapper doesn't reliably degrade that failure to `None`.
    let scope = if body.is_empty() {
        solo_storage::ConsolidationScope::default()
    } else {
        serde_json::from_slice(&body)
            .map_err(|e| ApiError::bad_request(format!("invalid JSON: {e}")))?
    };
    let report = s.write.consolidate(scope).await.map_err(ApiError::from)?;
    Ok(Json(report))
}

#[derive(Debug, Deserialize)]
struct BackupBody {
    /// Server-side absolute path where the backup file should be
    /// written. Must be writable by the Solo process. Refuses to
    /// overwrite an existing file unless `force = true`.
    to: String,
    #[serde(default)]
    force: bool,
}

#[derive(Debug, Serialize)]
struct BackupResponse {
    path: String,
    elapsed_ms: u64,
}

async fn backup_handler(
    State(s): State<SoloHttpState>,
    Json(body): Json<BackupBody>,
) -> Result<Json<BackupResponse>, ApiError> {
    use std::path::PathBuf;

    let dest = PathBuf::from(&body.to);
    if dest.as_os_str().is_empty() {
        return Err(ApiError::bad_request("`to` must not be empty"));
    }
    // CRITICAL ORDER: same-file refusal MUST come BEFORE `remove_file`.
    // A `force: true` request pointing at the daemon's live `solo.db`
    // would otherwise unlink the source on Linux (the writer's open fd
    // keeps it accessible until shutdown, but the data dir is empty
    // from any new opener's perspective). See v0.3.4 release notes.
    if solo_storage::paths_refer_to_same_file(&s.source_db_path, &dest) {
        return Err(ApiError::bad_request(format!(
            "destination {} is the same file as the source database; \
             refusing to run (would corrupt the live database)",
            dest.display()
        )));
    }
    if dest.exists() {
        if !body.force {
            return Err(ApiError::bad_request(format!(
                "destination {} exists; pass force=true to overwrite",
                dest.display()
            )));
        }
        std::fs::remove_file(&dest).map_err(|e| {
            ApiError::internal(format!(
                "remove existing destination {}: {e}",
                dest.display()
            ))
        })?;
    }
    if let Some(parent) = dest.parent() {
        if !parent.as_os_str().is_empty() && !parent.is_dir() {
            return Err(ApiError::bad_request(format!(
                "destination parent directory {} does not exist",
                parent.display()
            )));
        }
    }

    let started = std::time::Instant::now();
    s.write.backup(dest.clone()).await.map_err(ApiError::from)?;
    let elapsed_ms = started.elapsed().as_millis() as u64;

    Ok(Json(BackupResponse {
        path: dest.display().to_string(),
        elapsed_ms,
    }))
}

// ---------------------------------------------------------------------------
// Error mapping
// ---------------------------------------------------------------------------

#[derive(Debug)]
pub struct ApiError {
    status: StatusCode,
    message: String,
}

impl ApiError {
    fn bad_request(msg: impl Into<String>) -> Self {
        Self {
            status: StatusCode::BAD_REQUEST,
            message: msg.into(),
        }
    }
    fn not_found(msg: impl Into<String>) -> Self {
        Self {
            status: StatusCode::NOT_FOUND,
            message: msg.into(),
        }
    }
    fn internal(msg: impl Into<String>) -> Self {
        Self {
            status: StatusCode::INTERNAL_SERVER_ERROR,
            message: msg.into(),
        }
    }
}

impl From<solo_core::Error> for ApiError {
    fn from(e: solo_core::Error) -> Self {
        use solo_core::Error;
        match e {
            Error::NotFound(msg) => ApiError::not_found(msg),
            Error::InvalidInput(msg) => ApiError::bad_request(msg),
            Error::Conflict(msg) => Self {
                status: StatusCode::CONFLICT,
                message: msg,
            },
            other => ApiError::internal(other.to_string()),
        }
    }
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        let body = serde_json::json!({
            "error": self.message,
            "status": self.status.as_u16(),
        });
        (self.status, Json(body)).into_response()
    }
}

// SQL helper for recall used to live here; consolidated into
// solo_query::recall.

#[cfg(test)]
mod handler_tests {
    //! In-process integration tests for the HTTP handler surface. We
    //! drive the axum Router directly via `tower::ServiceExt::oneshot`
    //! — no real TCP listener needed. Same `Harness`-shape as the MCP
    //! tests: real WriterActor + ReaderPool + StubEmbedder + StubVectorIndex.
    //!
    //! Tests live inline in this module rather than in a `tests/` dir
    //! because external integration-test exes triggered Windows UAC
    //! ERROR_ELEVATION_REQUIRED on the dev machine.
    use super::*;
    use axum::body::Body;
    use axum::http::{Request, StatusCode};
    use http_body_util::BodyExt;
    use serde_json::{Value, json};
    use solo_core::VectorIndex as _;
    use solo_storage::test_support::StubVectorIndex;
    use solo_storage::{ReaderPool, StubEmbedder, WriterActor, WriterSpawn};
    use std::sync::Arc as StdArc;
    use tower::ServiceExt;

    struct Harness {
        router: axum::Router,
        _tmp: tempfile::TempDir,
        write_handle_extra: Option<solo_storage::WriteHandle>,
        join: Option<std::thread::JoinHandle<()>>,
    }

    impl Harness {
        fn new(runtime: &tokio::runtime::Runtime) -> Self {
            Self::new_with_auth(runtime, None)
        }

        fn new_with_auth(
            runtime: &tokio::runtime::Runtime,
            bearer_token: Option<String>,
        ) -> Self {
            use solo_storage::embedder_registry::{EmbedderIdentity, get_or_insert_embedder_id};

            let tmp = tempfile::TempDir::new().unwrap();
            let dim = 16usize;
            let hnsw: StdArc<dyn VectorIndex + Send + Sync> = StdArc::new(StubVectorIndex::new(dim));
            let embedder: StdArc<dyn solo_core::Embedder> =
                StdArc::new(StubEmbedder::new("stub", "v1", dim));
            let path = tmp.path().join("test.db");

            // Register an embedder_id so `handle_consolidate` (and any
            // other path that filters on `embeddings.embedder_id`) sees
            // a production-shaped writer. Existing handler tests pass
            // unchanged because the writer just additionally INSERTs
            // an `embeddings` row per remember — no assertions look at
            // that table.
            let embedder_id = {
                let conn = solo_storage::test_support::open_test_db_at(&path);
                get_or_insert_embedder_id(
                    &conn,
                    &EmbedderIdentity {
                        name: "stub".into(),
                        version: "v1".into(),
                        dim: dim as u32,
                        dtype: "f32".into(),
                    },
                )
                .unwrap()
            };

            let conn = solo_storage::test_support::open_test_db_at(&path);
            let WriterSpawn { handle, join } = WriterActor::spawn_full(
                conn,
                hnsw.clone(),
                tmp.path().to_path_buf(),
                embedder_id,
            );
            let pool: ReaderPool =
                runtime.block_on(async { ReaderPool::new(&path, None, hnsw.clone()).unwrap() });
            let state = SoloHttpState {
                write: handle.clone(),
                pool,
                embedder,
                hnsw,
                source_db_path: path.clone(),
            };
            let router = router_with_auth(state, bearer_token);
            Harness {
                router,
                _tmp: tmp,
                write_handle_extra: Some(handle),
                join: Some(join),
            }
        }

        fn shutdown(mut self, runtime: &tokio::runtime::Runtime) {
            let join = self.join.take();
            let extra = self.write_handle_extra.take();
            runtime.block_on(async move {
                drop(extra);
                drop(self.router); // drops state → drops pool inside runtime ctx
                drop(self._tmp);
                if let Some(join) = join {
                    let (tx, rx) = std::sync::mpsc::channel();
                    std::thread::spawn(move || {
                        let _ = tx.send(join.join());
                    });
                    tokio::task::spawn_blocking(move || {
                        rx.recv_timeout(std::time::Duration::from_secs(5))
                    })
                    .await
                    .expect("blocking task")
                    .expect("writer thread did not exit within 5s")
                    .expect("writer thread panicked");
                }
            });
        }
    }

    fn rt() -> tokio::runtime::Runtime {
        tokio::runtime::Builder::new_multi_thread()
            .worker_threads(2)
            .enable_all()
            .build()
            .unwrap()
    }

    /// Issue one HTTP request through the router and capture status +
    /// JSON body. `body` may be `None` for GET/DELETE; `auth` adds an
    /// `Authorization` header value verbatim (e.g. `"Bearer xyz"`).
    async fn call(
        router: axum::Router,
        method: &str,
        uri: &str,
        body: Option<Value>,
    ) -> (StatusCode, Value) {
        call_with_auth(router, method, uri, body, None).await
    }

    async fn call_with_auth(
        router: axum::Router,
        method: &str,
        uri: &str,
        body: Option<Value>,
        auth: Option<&str>,
    ) -> (StatusCode, Value) {
        let mut req_builder = Request::builder()
            .method(method)
            .uri(uri)
            .header("content-type", "application/json");
        if let Some(a) = auth {
            req_builder = req_builder.header("authorization", a);
        }
        let req = if let Some(b) = body {
            let bytes = serde_json::to_vec(&b).unwrap();
            req_builder.body(Body::from(bytes)).unwrap()
        } else {
            req_builder = req_builder.header("content-length", "0");
            req_builder.body(Body::empty()).unwrap()
        };
        let resp = router.oneshot(req).await.expect("oneshot");
        let status = resp.status();
        let body_bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let v: Value = if body_bytes.is_empty() {
            Value::Null
        } else {
            serde_json::from_slice(&body_bytes).unwrap_or(Value::Null)
        };
        (status, v)
    }

    #[test]
    fn health_returns_ok() {
        let runtime = rt();
        let h = Harness::new(&runtime);
        let r = h.router.clone();
        let (status, _body) = runtime.block_on(call(r, "GET", "/health", None));
        assert_eq!(status, StatusCode::OK);
        h.shutdown(&runtime);
    }

    /// `GET /openapi.json` returns a parseable OpenAPI 3.x document with
    /// the four `memory.*` endpoints + their request/response schemas.
    /// Acts as a drift detector: if a future commit adds/removes a route
    /// without updating `openapi_spec`, this test fails loudly.
    #[test]
    fn openapi_json_describes_all_endpoints() {
        let runtime = rt();
        let h = Harness::new(&runtime);
        let r = h.router.clone();
        let (status, spec) = runtime.block_on(call(r, "GET", "/openapi.json", None));
        assert_eq!(status, StatusCode::OK);
        assert!(spec.is_object(), "openapi.json must be a JSON object");

        // Top-level shape per OpenAPI 3.1.
        assert!(
            spec.get("openapi")
                .and_then(|v| v.as_str())
                .is_some_and(|s| s.starts_with("3.")),
            "missing or wrong openapi version: {spec}"
        );
        assert!(spec.pointer("/info/title").is_some());
        assert!(spec.pointer("/info/version").is_some());

        // Every route the router serves must be documented.
        let paths = spec
            .get("paths")
            .and_then(|v| v.as_object())
            .expect("paths must be an object");
        for expected in [
            "/health",
            "/openapi.json",
            "/memory",
            "/memory/search",
            "/memory/consolidate",
            "/memory/{id}",
            // Path 1 derived-layer endpoints (v0.4.0+):
            "/memory/themes",
            "/memory/facts_about",
            "/memory/contradictions",
        ] {
            assert!(
                paths.contains_key(expected),
                "openapi paths missing {expected}: {paths:?}"
            );
        }

        // Method coverage on /memory/{id}: must document both GET (inspect)
        // and DELETE (forget).
        let memid = paths.get("/memory/{id}").expect("memory/{id}");
        assert!(memid.get("get").is_some(), "GET /memory/{{id}} undocumented");
        assert!(
            memid.get("delete").is_some(),
            "DELETE /memory/{{id}} undocumented"
        );

        // Component schemas referenced from paths must be defined.
        for schema_name in [
            "RememberRequest",
            "RememberResponse",
            "RecallRequest",
            "RecallResult",
            "EpisodeRecord",
            "ApiError",
            "ConsolidationScope",
            "ConsolidationReport",
            // Path 1 derived-layer schemas (v0.4.0+):
            "ThemeHit",
            "FactHit",
            "ContradictionHit",
        ] {
            let ptr = format!("/components/schemas/{schema_name}");
            assert!(
                spec.pointer(&ptr).is_some(),
                "component schema {schema_name} missing"
            );
        }

        // bearerAuth security scheme is declared (LAN deployments need it).
        assert!(
            spec.pointer("/components/securitySchemes/bearerAuth")
                .is_some(),
            "bearerAuth security scheme missing"
        );

        h.shutdown(&runtime);
    }

    /// `/openapi.json` must remain unauthenticated even when bearer auth
    /// is enabled — the spec describes the API shape, not secrets, and
    /// codegen tooling shouldn't need a credential to fetch it.
    #[test]
    fn openapi_json_is_exempt_from_bearer_auth() {
        let runtime = rt();
        let h = Harness::new_with_auth(&runtime, Some("super-secret".into()));
        let r = h.router.clone();
        // No Authorization header → still 200 for /openapi.json.
        let (status, _body) = runtime.block_on(call(r, "GET", "/openapi.json", None));
        assert_eq!(status, StatusCode::OK);
        h.shutdown(&runtime);
    }

    #[test]
    fn remember_returns_memory_id() {
        let runtime = rt();
        let h = Harness::new(&runtime);
        let r = h.router.clone();
        let (status, body) = runtime.block_on(call(
            r,
            "POST",
            "/memory",
            Some(json!({ "content": "http harness test" })),
        ));
        assert_eq!(status, StatusCode::OK);
        let mid = body.get("memory_id").and_then(|v| v.as_str()).unwrap();
        assert_eq!(mid.len(), 36, "uuid length");
        h.shutdown(&runtime);
    }

    #[test]
    fn empty_content_returns_400() {
        let runtime = rt();
        let h = Harness::new(&runtime);
        let r = h.router.clone();
        let (status, body) =
            runtime.block_on(call(r, "POST", "/memory", Some(json!({ "content": "" }))));
        assert_eq!(status, StatusCode::BAD_REQUEST);
        assert!(
            body.get("error")
                .and_then(|e| e.as_str())
                .map(|s| s.contains("must not be empty"))
                .unwrap_or(false),
            "got: {body}"
        );
        h.shutdown(&runtime);
    }

    #[test]
    fn empty_query_returns_400() {
        let runtime = rt();
        let h = Harness::new(&runtime);
        let r = h.router.clone();
        let (status, body) = runtime.block_on(call(
            r,
            "POST",
            "/memory/search",
            Some(json!({ "query": "" })),
        ));
        assert_eq!(status, StatusCode::BAD_REQUEST);
        assert!(
            body.get("error")
                .and_then(|e| e.as_str())
                .map(|s| s.contains("must not be empty"))
                .unwrap_or(false),
            "got: {body}"
        );
        h.shutdown(&runtime);
    }

    #[test]
    fn inspect_unknown_returns_404() {
        let runtime = rt();
        let h = Harness::new(&runtime);
        let r = h.router.clone();
        let (status, body) = runtime.block_on(call(
            r,
            "GET",
            "/memory/00000000-0000-7000-8000-000000000000",
            None,
        ));
        assert_eq!(status, StatusCode::NOT_FOUND);
        assert!(body.get("error").is_some(), "got: {body}");
        h.shutdown(&runtime);
    }

    #[test]
    fn inspect_invalid_id_returns_400() {
        let runtime = rt();
        let h = Harness::new(&runtime);
        let r = h.router.clone();
        let (status, _body) = runtime.block_on(call(r, "GET", "/memory/not-a-uuid", None));
        assert_eq!(status, StatusCode::BAD_REQUEST);
        h.shutdown(&runtime);
    }

    #[test]
    fn forget_unknown_returns_404() {
        let runtime = rt();
        let h = Harness::new(&runtime);
        let r = h.router.clone();
        let (status, _body) = runtime.block_on(call(
            r,
            "DELETE",
            "/memory/00000000-0000-7000-8000-000000000000",
            None,
        ));
        assert_eq!(status, StatusCode::NOT_FOUND);
        h.shutdown(&runtime);
    }

    /// `POST /memory/consolidate` runs the cluster pass and returns
    /// the report as JSON. With an empty body, `ConsolidationScope`
    /// defaults to unbounded; with a non-empty body, the
    /// `window_days` field is honored. The Harness's writer is
    /// spawned without a Steward, so `abstractions_built` stays 0
    /// even when `clusters_built` is nonzero — same posture as the
    /// daemon today.
    #[test]
    fn consolidate_endpoint_returns_report() {
        let runtime = rt();
        let h = Harness::new(&runtime);
        let r = h.router.clone();
        runtime.block_on(async move {
            // Empty DB → all-zero report; structural assertion only.
            let (status, body) = call(r.clone(), "POST", "/memory/consolidate", None).await;
            assert_eq!(status, StatusCode::OK);
            for field in [
                "episodes_seen",
                "clusters_built",
                "episodes_clustered",
                "abstractions_built",
                "triples_built",
                "contradictions_found",
            ] {
                assert!(
                    body.get(field).and_then(|v| v.as_u64()).is_some(),
                    "missing field {field}: {body}"
                );
            }
            assert_eq!(body["episodes_seen"], 0);
            assert_eq!(body["clusters_built"], 0);

            // Non-empty body with window_days → still 200; unmistakable
            // shape round-trips through ConsolidationScope's serde.
            let (status2, _body2) = call(
                r,
                "POST",
                "/memory/consolidate",
                Some(json!({ "window_days": 7 })),
            )
            .await;
            assert_eq!(status2, StatusCode::OK);
        });
        h.shutdown(&runtime);
    }

    #[test]
    fn auth_required_routes_reject_missing_token() {
        let runtime = rt();
        let h = Harness::new_with_auth(&runtime, Some("secret-xyz".into()));
        let r = h.router.clone();
        runtime.block_on(async move {
            // No Authorization header → 401.
            let (status, _body) = call(
                r.clone(),
                "POST",
                "/memory",
                Some(json!({ "content": "x" })),
            )
            .await;
            assert_eq!(status, StatusCode::UNAUTHORIZED);

            // Wrong token → 401.
            let (status, _body) = call_with_auth(
                r.clone(),
                "POST",
                "/memory",
                Some(json!({ "content": "x" })),
                Some("Bearer wrong-token"),
            )
            .await;
            assert_eq!(status, StatusCode::UNAUTHORIZED);

            // Correct token → handler runs (200).
            let (status, body) = call_with_auth(
                r.clone(),
                "POST",
                "/memory",
                Some(json!({ "content": "authed" })),
                Some("Bearer secret-xyz"),
            )
            .await;
            assert_eq!(status, StatusCode::OK);
            assert!(body.get("memory_id").is_some());
        });
        h.shutdown(&runtime);
    }

    #[test]
    fn health_endpoint_does_not_require_auth() {
        let runtime = rt();
        let h = Harness::new_with_auth(&runtime, Some("secret".into()));
        let r = h.router.clone();
        let (status, _body) = runtime.block_on(call(r, "GET", "/health", None));
        // Liveness probes should work without credentials.
        assert_eq!(status, StatusCode::OK);
        h.shutdown(&runtime);
    }

    #[test]
    fn auth_response_includes_www_authenticate_header() {
        // Verify the WWW-Authenticate hint that lets a well-behaved
        // client know it's a bearer-auth scheme. We check via raw
        // request → response (oneshot returns Response, but our
        // call() helper drops the headers; build the request manually).
        let runtime = rt();
        let h = Harness::new_with_auth(&runtime, Some("secret".into()));
        let r = h.router.clone();
        runtime.block_on(async move {
            let req = Request::builder()
                .method("POST")
                .uri("/memory")
                .header("content-type", "application/json")
                .body(Body::from(serde_json::to_vec(&json!({ "content": "x" })).unwrap()))
                .unwrap();
            let resp = r.oneshot(req).await.unwrap();
            assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
            let www = resp
                .headers()
                .get("www-authenticate")
                .and_then(|v| v.to_str().ok())
                .unwrap_or("");
            assert!(
                www.starts_with("Bearer"),
                "expected WWW-Authenticate: Bearer..., got: {www}"
            );
        });
        h.shutdown(&runtime);
    }

    #[test]
    fn full_remember_recall_inspect_forget_round_trip() {
        let runtime = rt();
        let h = Harness::new(&runtime);
        let r = h.router.clone();
        runtime.block_on(async move {
            // POST /memory
            let (status, body) = call(
                r.clone(),
                "POST",
                "/memory",
                Some(json!({ "content": "round-trip content" })),
            )
            .await;
            assert_eq!(status, StatusCode::OK);
            let mid = body
                .get("memory_id")
                .and_then(|v| v.as_str())
                .unwrap()
                .to_string();

            // POST /memory/search — exact-match (StubEmbedder) returns the row.
            let (status, body) = call(
                r.clone(),
                "POST",
                "/memory/search",
                Some(json!({ "query": "round-trip content", "limit": 5 })),
            )
            .await;
            assert_eq!(status, StatusCode::OK);
            let hits = body.get("hits").and_then(|v| v.as_array()).unwrap();
            assert!(
                hits.iter()
                    .any(|h| h.get("content").and_then(|c| c.as_str())
                        == Some("round-trip content")),
                "expected hit with content; got: {body}"
            );

            // GET /memory/{id}
            let (status, body) = call(r.clone(), "GET", &format!("/memory/{mid}"), None).await;
            assert_eq!(status, StatusCode::OK);
            assert_eq!(body.get("status").and_then(|v| v.as_str()), Some("active"));

            // DELETE /memory/{id}
            let (status, _body) =
                call(r.clone(), "DELETE", &format!("/memory/{mid}"), None).await;
            assert_eq!(status, StatusCode::NO_CONTENT);

            // GET again — still readable but status='forgotten'
            let (status, body) = call(r.clone(), "GET", &format!("/memory/{mid}"), None).await;
            assert_eq!(status, StatusCode::OK);
            assert_eq!(
                body.get("status").and_then(|v| v.as_str()),
                Some("forgotten")
            );

            // POST /memory/search — forgotten row excluded.
            let (status, body) = call(
                r.clone(),
                "POST",
                "/memory/search",
                Some(json!({ "query": "round-trip content", "limit": 5 })),
            )
            .await;
            assert_eq!(status, StatusCode::OK);
            let hits = body.get("hits").and_then(|v| v.as_array()).unwrap();
            assert!(
                hits.iter().all(|h| h.get("memory_id").and_then(|m| m.as_str())
                    != Some(mid.as_str())),
                "forgotten row should be excluded from recall: {body}"
            );
        });
        h.shutdown(&runtime);
    }

    // Path 1 derived-layer endpoint tests (v0.4.0+). Wire-path only —
    // the actual content correctness is covered by solo-query::derived's
    // own tests (Sub-task A). These verify the HTTP shape: GET routing,
    // Query-string param parsing, JSON-array response body, validation
    // 400s for invalid inputs.

    #[test]
    fn themes_endpoint_returns_empty_array_on_empty_db() {
        let runtime = rt();
        let h = Harness::new(&runtime);
        let r = h.router.clone();
        let (status, body) =
            runtime.block_on(call(r, "GET", "/memory/themes", None));
        assert_eq!(status, StatusCode::OK);
        assert!(body.is_array(), "expected array, got {body}");
        assert_eq!(body.as_array().unwrap().len(), 0);
        h.shutdown(&runtime);
    }

    #[test]
    fn themes_endpoint_passes_through_query_params() {
        let runtime = rt();
        let h = Harness::new(&runtime);
        let r = h.router.clone();
        let (status, body) = runtime.block_on(call(
            r,
            "GET",
            "/memory/themes?window_days=7&limit=20",
            None,
        ));
        assert_eq!(status, StatusCode::OK);
        assert!(body.is_array(), "expected array, got {body}");
        h.shutdown(&runtime);
    }

    #[test]
    fn facts_about_endpoint_requires_subject() {
        let runtime = rt();
        let h = Harness::new(&runtime);
        let r = h.router.clone();
        // Missing subject — axum's Query extractor 422 (Unprocessable
        // Entity) on missing required field; some axum versions
        // surface as 400. Accept either.
        let (status, _body) =
            runtime.block_on(call(r, "GET", "/memory/facts_about", None));
        assert!(
            status == StatusCode::BAD_REQUEST
                || status == StatusCode::UNPROCESSABLE_ENTITY,
            "expected 400 or 422 for missing subject, got {status}"
        );
        h.shutdown(&runtime);
    }

    #[test]
    fn facts_about_endpoint_rejects_blank_subject() {
        let runtime = rt();
        let h = Harness::new(&runtime);
        let r = h.router.clone();
        // Whitespace-only subject reaches the handler then trips its
        // own validation → ApiError::bad_request → 400.
        let (status, body) = runtime.block_on(call(
            r,
            "GET",
            "/memory/facts_about?subject=%20%20",
            None,
        ));
        assert_eq!(status, StatusCode::BAD_REQUEST);
        assert!(
            body.get("error")
                .and_then(|v| v.as_str())
                .is_some_and(|s| s.contains("subject")),
            "expected error mentioning subject, got {body}"
        );
        h.shutdown(&runtime);
    }

    #[test]
    fn facts_about_endpoint_returns_empty_array_for_unknown_subject() {
        let runtime = rt();
        let h = Harness::new(&runtime);
        let r = h.router.clone();
        let (status, body) = runtime.block_on(call(
            r,
            "GET",
            "/memory/facts_about?subject=NobodyKnows",
            None,
        ));
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body.as_array().unwrap().len(), 0);
        h.shutdown(&runtime);
    }

    #[test]
    fn contradictions_endpoint_returns_empty_array_on_empty_db() {
        let runtime = rt();
        let h = Harness::new(&runtime);
        let r = h.router.clone();
        let (status, body) = runtime.block_on(call(
            r,
            "GET",
            "/memory/contradictions",
            None,
        ));
        assert_eq!(status, StatusCode::OK);
        assert!(body.is_array());
        assert_eq!(body.as_array().unwrap().len(), 0);
        h.shutdown(&runtime);
    }

    #[test]
    fn derived_endpoints_require_bearer_when_auth_enabled() {
        let runtime = rt();
        let h = Harness::new_with_auth(&runtime, Some("secret-token".to_string()));
        // Each of the three new endpoints should reject missing token.
        // Per the existing tests' shutdown-timing comment: don't hold a
        // long-lived router clone across multiple iterations — drop the
        // clone before each subsequent oneshot, and don't keep a `let r =
        // h.router.clone()` alive across h.shutdown(). Re-clone per
        // iteration; the per-call clone is consumed by oneshot.
        for path in [
            "/memory/themes",
            "/memory/facts_about?subject=Sam",
            "/memory/contradictions",
        ] {
            let (status, _) = runtime.block_on(call(h.router.clone(), "GET", path, None));
            assert_eq!(
                status,
                StatusCode::UNAUTHORIZED,
                "{path} should 401 without token"
            );
        }
        h.shutdown(&runtime);
    }
}

#[cfg(test)]
mod cors_tests {
    use super::is_localhost_origin;

    #[test]
    fn accepts_canonical_localhost_origins() {
        assert!(is_localhost_origin("http://localhost"));
        assert!(is_localhost_origin("http://localhost:3000"));
        assert!(is_localhost_origin("https://localhost:8443"));
        assert!(is_localhost_origin("http://127.0.0.1"));
        assert!(is_localhost_origin("http://127.0.0.1:5173"));
        assert!(is_localhost_origin("http://[::1]"));
        assert!(is_localhost_origin("http://[::1]:8080"));
    }

    #[test]
    fn rejects_remote_origins() {
        assert!(!is_localhost_origin("http://example.com"));
        assert!(!is_localhost_origin("https://malicious.example"));
        assert!(!is_localhost_origin("http://192.168.1.5"));
        assert!(!is_localhost_origin("http://10.0.0.1"));
    }

    #[test]
    fn rejects_dns_rebinding_tricks() {
        // nip.io and friends — DNS that resolves to 127.0.0.1 but the
        // Origin header carries the public-DNS name. Rejecting these
        // closes the rebinding-via-Origin gap.
        assert!(!is_localhost_origin("http://127.0.0.1.nip.io"));
        assert!(!is_localhost_origin("http://localhost.evil.com"));
        assert!(!is_localhost_origin("http://evil.localhost"));
    }

    #[test]
    fn rejects_non_http_schemes() {
        assert!(!is_localhost_origin("file:///"));
        assert!(!is_localhost_origin("ws://localhost:3000"));
        assert!(!is_localhost_origin("javascript:alert(1)"));
    }

    #[test]
    fn rejects_malformed() {
        assert!(!is_localhost_origin(""));
        assert!(!is_localhost_origin("localhost"));
        assert!(!is_localhost_origin("//localhost"));
    }
}