From b4e8be9afcb786cc60ff1b040a87cfc20b57301f Mon Sep 17 00:00:00 2001 From: rrader2890 Date: Fri, 7 Aug 2026 20:26:42 -0400 Subject: [PATCH 1/4] feat(console): local web console (status/memories/logs/config) + Postgres backend switch - memmesh console: single-binary web UI served from the REST API, opens browser - embedded SPA (crates/server/src/ui/index.html) with MemMesh logo - REST endpoints: /stats, GET /memory (list), /config (get/put), /logs, /consolidate - unified log file (~/.memmesh/logs/memmesh.log) tailed by the console - selectable storage backend via [database] config; generic run dispatch - /database (get/put), /database/test, /database/copy for SQLite->Postgres switch + data copy --- Cargo.lock | 30 +++ Cargo.toml | 1 + crates/cli/Cargo.toml | 1 + crates/cli/src/main.rs | 169 ++++++++++++-- crates/core/src/config.rs | 64 ++++++ crates/server/Cargo.toml | 1 + crates/server/src/http.rs | 350 ++++++++++++++++++++++++++-- crates/server/src/ui/index.html | 393 ++++++++++++++++++++++++++++++++ 8 files changed, 973 insertions(+), 36 deletions(-) create mode 100644 crates/server/src/ui/index.html diff --git a/Cargo.lock b/Cargo.lock index 7af4c46..70c9f31 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -706,6 +706,15 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "crossbeam-channel" +version = "0.5.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d85363c37faeca707aef026efa9f3b34d077bce547e48f770770625c6013679e" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-deque" version = "0.8.6" @@ -2099,6 +2108,7 @@ dependencies = [ "toml", "toml_edit", "tracing", + "tracing-appender", "tracing-subscriber", "uuid", ] @@ -2217,6 +2227,7 @@ dependencies = [ "serde", "serde_json", "tokio", + "toml", "tonic", "tower-http 0.5.2", "tracing", @@ -3915,6 +3926,12 @@ version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" +[[package]] +name = "symlink" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7973cce6668464ea31f176d85b13c7ab3bba2cb3b77a2ed26abd7801688010a" + [[package]] name = "syn" version = "2.0.118" @@ -4394,6 +4411,19 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-appender" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "050686193eb999b4bb3bc2acfa891a13da00f79734704c4b8b4ef1a10b368a3c" +dependencies = [ + "crossbeam-channel", + "symlink", + "thiserror 2.0.18", + "time", + "tracing-subscriber", +] + [[package]] name = "tracing-attributes" version = "0.1.31" diff --git a/Cargo.toml b/Cargo.toml index 1df5bd8..8058298 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,6 +36,7 @@ futures = "0.3" # Logging / tracing tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } +tracing-appender = "0.2" # Serialization serde = { version = "1", features = ["derive"] } diff --git a/crates/cli/Cargo.toml b/crates/cli/Cargo.toml index fd01618..2050dbc 100644 --- a/crates/cli/Cargo.toml +++ b/crates/cli/Cargo.toml @@ -29,6 +29,7 @@ tokio.workspace = true clap.workspace = true tracing.workspace = true tracing-subscriber.workspace = true +tracing-appender.workspace = true serde.workspace = true serde_json.workspace = true serde_yaml.workspace = true diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index 21d2bd9..d2a5b6d 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -17,9 +17,10 @@ use anyhow::{anyhow, Context, Result}; use chrono::{DateTime, Utc}; use clap::{Parser, Subcommand}; use installer::{Action, SkillBundle, Tool}; +use memory_core::config::DatabaseBackend; use memory_core::{MemoryItem, MemoryScope}; use memory_license::License; -use memory_storage::{sqlite::SqliteStore, MemoryFilter, Storage}; +use memory_storage::{postgres::PostgresStore, sqlite::SqliteStore, MemoryFilter, Storage}; use std::sync::Arc; /// Default DB path under $XDG_DATA_HOME / ~/.local/share / fallback CWD. @@ -205,6 +206,18 @@ enum Cmd { http: String, }, + /// Launch the local web console: starts the REST API, serves the + /// memory management UI, and opens it in your browser. One command to + /// see engine status, browse memories, tail logs, and tweak config. + Console { + /// Address to bind. Default 127.0.0.1:7878. + #[arg(long, default_value = "127.0.0.1:7878")] + http: String, + /// Don't auto-open a browser; just print the URL. + #[arg(long)] + no_open: bool, + }, + /// Manage the agent teaching skill (markdown that tells the AI when /// and how to use the memory tools). Skill { @@ -448,21 +461,49 @@ const SKILL_MD: &str = include_str!("../../../skills/memmesh/SKILL.md"); async fn main() -> Result<()> { let cli = Cli::parse(); - init_tracing(&cli.log)?; - - let db_url = format!("sqlite://{}?mode=rwc", cli.db); - let store = SqliteStore::connect(&db_url) - .await - .with_context(|| format!("opening sqlite store at {}", cli.db))?; - store.migrate().await.context("running migrations")?; - let store = Arc::new(store); + // Hold the log-file guard for the whole process so buffered log lines + // flush on exit. Dropping it early would truncate the tail of the log. + let _log_guard = init_tracing(&cli.log)?; // Resolve license once per CLI invocation. `tflk_...` license keys // trigger an HTTP exchange against the SaaS to receive a JWT; raw - // JWTs verify locally without network. Kept immutable for the rest - // of main(). Phase 2 will add runtime refresh. + // JWTs verify locally without network. let license = License::load_from_env(Utc::now()).await; + // Pick the storage backend from config, then dispatch into the generic + // `run`. Both arms monomorphize `run` — the whole command surface works + // identically on SQLite or Postgres. + let cfg = memory_core::config::Config::load_or_default(); + match cfg.database.backend { + DatabaseBackend::Sqlite => { + let db_url = format!("sqlite://{}?mode=rwc", cli.db); + let store = SqliteStore::connect(&db_url) + .await + .with_context(|| format!("opening sqlite store at {}", cli.db))?; + store.migrate().await.context("running migrations")?; + run(Arc::new(store), cli, license).await + } + DatabaseBackend::Postgres => { + let url = cfg.database.url.clone().ok_or_else(|| { + anyhow!( + "[database] backend is \"postgres\" but no url is set. Add a url \ + to ~/.memmesh/config.toml (or set THINKFLEET_DATABASE_URL), or \ + switch back to sqlite in the console." + ) + })?; + let store = PostgresStore::connect(&url) + .await + .with_context(|| "connecting to postgres")?; + store.migrate().await.context("running postgres migrations")?; + tracing::info!("storage backend: postgres"); + run(Arc::new(store), cli, license).await + } + } +} + +/// Generic command dispatch — runs against whichever `Storage` backend +/// `main` selected. Monomorphized once per backend. +async fn run(store: Arc, cli: Cli, license: License) -> Result<()> { match cli.cmd { Cmd::Migrate => { let version = store.migrate().await?; @@ -711,6 +752,42 @@ async fn main() -> Result<()> { memory_server::serve_http(store, &http).await?; } + Cmd::Console { http, no_open } => { + // Same server as `serve` (REST + embedded UI), plus a friendly + // banner and an auto-opened browser. Kick off background sync if + // SaaS is configured, exactly like `serve`. + let cfg = memory_core::config::Config::load_or_default(); + if let Some(sync_cfg) = cfg.sync.clone() { + let sync_store = store.clone(); + tokio::spawn(async move { + run_sync_daemon(sync_store, sync_cfg).await; + }); + } + let url = format!("http://{http}"); + println!(); + println!(" MemMesh console"); + println!(" ───────────────"); + println!(" engine : {}", cli.db); + println!(" logs : {}", memory_core::config::Config::log_file().display()); + println!(" url : {url}"); + println!(); + if no_open { + println!(" Open {url} in your browser."); + } else { + // Delay the launch briefly so the listener is bound before the + // browser requests the page (avoids a first-load failure). + let url_for_open = url.clone(); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(1500)).await; + open_browser(&url_for_open); + }); + println!(" Opening browser… (Ctrl-C to stop the console)"); + } + println!(); + tracing::info!(db = %cli.db, addr = %http, "starting web console"); + memory_server::serve_http(store, &http).await?; + } + Cmd::Sync { skip_token_check } => { let cfg = memory_core::config::Config::load_or_default(); let sync_cfg = cfg.sync.clone().ok_or_else(|| { @@ -1079,8 +1156,8 @@ fn format_claude_context(rows: &[MemoryItem], project_hint: Option<&str>) -> Str /// once at startup, then loops on `interval_seconds`, calling /// `run_cycle` each tick. Errors are logged but never panic the loop; /// the next tick retries. -async fn run_sync_daemon( - store: std::sync::Arc, +async fn run_sync_daemon( + store: std::sync::Arc, sync_cfg: memory_core::config::SyncConfig, ) { let client = match memory_sync::SyncClient::new(sync_cfg.url.clone(), sync_cfg.token.clone()) { @@ -1273,15 +1350,63 @@ fn parse_scope(s: &str) -> Result { }) } -fn init_tracing(level: &str) -> Result<()> { - // The MCP stdio server uses stdout for protocol traffic, so logs MUST - // go to stderr. tracing_subscriber::fmt defaults to stdout — switch - // explicitly. Honor RUST_LOG if set. - use tracing_subscriber::EnvFilter; +/// Initialize logging. Logs go to **stderr** (the MCP stdio server uses +/// stdout for protocol traffic, so logs must never touch it) **and** are +/// appended to a unified log file at `~/.memmesh/logs/memmesh.log` that the +/// web console tails. Returns the file-writer guard, which the caller must +/// hold for the process lifetime so buffered lines flush on exit. +fn init_tracing(level: &str) -> Result> { + use tracing_subscriber::{fmt, prelude::*, EnvFilter}; let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(level)); - tracing_subscriber::fmt() - .with_writer(std::io::stderr) - .with_env_filter(filter) + + let stderr_layer = fmt::layer().with_writer(std::io::stderr); + + // Best-effort file layer — if the log dir can't be created we just log to + // stderr and carry on rather than failing to start the engine. + let (file_layer, guard) = match open_log_writer() { + Some((writer, guard)) => ( + Some(fmt::layer().with_ansi(false).with_writer(writer)), + Some(guard), + ), + None => (None, None), + }; + + tracing_subscriber::registry() + .with(filter) + .with(stderr_layer) + .with(file_layer) .init(); - Ok(()) + Ok(guard) +} + +/// Open the unified log file for appending, wrapped in a non-blocking writer. +/// Every `memmesh` process appends to the same file so the console shows one +/// timeline across `mcp`, `serve`, and `console`. +fn open_log_writer() -> Option<( + tracing_appender::non_blocking::NonBlocking, + tracing_appender::non_blocking::WorkerGuard, +)> { + let dir = memory_core::config::Config::log_dir(); + std::fs::create_dir_all(&dir).ok()?; + // `never` = no rotation; a single append-only file, which is what makes + // concurrent appends from multiple processes behave predictably. + let appender = tracing_appender::rolling::never(&dir, "memmesh.log"); + Some(tracing_appender::non_blocking(appender)) +} + +/// Best-effort: open `url` in the user's default browser. Never fails the +/// command — if it can't spawn, the URL was already printed for manual use. +fn open_browser(url: &str) { + #[cfg(target_os = "macos")] + let (bin, args): (&str, Vec<&str>) = ("open", vec![url]); + #[cfg(target_os = "linux")] + let (bin, args): (&str, Vec<&str>) = ("xdg-open", vec![url]); + #[cfg(target_os = "windows")] + let (bin, args): (&str, Vec<&str>) = ("cmd", vec!["/C", "start", "", url]); + #[cfg(not(any(target_os = "macos", target_os = "linux", target_os = "windows")))] + let (bin, args): (&str, Vec<&str>) = ("true", vec![]); + + if let Err(e) = std::process::Command::new(bin).args(&args).spawn() { + tracing::warn!(error = %e, "could not launch browser; open the URL manually"); + } } diff --git a/crates/core/src/config.rs b/crates/core/src/config.rs index f9533a2..7a19fb2 100644 --- a/crates/core/src/config.rs +++ b/crates/core/src/config.rs @@ -46,6 +46,42 @@ pub struct Config { /// the user's SaaS/on-prem engine. #[serde(default)] pub embeddings: memory_embed::EmbeddingConfig, + /// Which storage backend the engine binds to at startup. Defaults to + /// SQLite (local file). Set `backend = "postgres"` + a `url` to run + /// against a Postgres instance instead. Changing this takes effect the + /// next time each `memmesh` process starts (console, `serve`, and every + /// MCP server an AI tool spawns). + #[serde(default)] + pub database: DatabaseConfig, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct DatabaseConfig { + /// `sqlite` (default) or `postgres`. + #[serde(default)] + pub backend: DatabaseBackend, + /// Postgres connection URL (`postgres://user:pass@host:5432/dbname`). + /// Required when `backend = "postgres"`; ignored for sqlite, which uses + /// the `--db` path / `THINKFLEET_MEMORY_DB`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub url: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum DatabaseBackend { + #[default] + Sqlite, + Postgres, +} + +impl DatabaseBackend { + pub fn as_str(&self) -> &'static str { + match self { + DatabaseBackend::Sqlite => "sqlite", + DatabaseBackend::Postgres => "postgres", + } + } } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -140,6 +176,23 @@ impl Config { .join("config.toml") } + /// Directory where the engine writes rotating log files. `~/.memmesh/logs` + /// unless `THINKFLEET_MEMORY_LOG_DIR` overrides it. Every `memmesh` + /// process (mcp / serve / console) appends to the same file here, so the + /// web console can tail a single unified log across all engine activity. + pub fn log_dir() -> PathBuf { + if let Ok(p) = std::env::var("THINKFLEET_MEMORY_LOG_DIR") { + return PathBuf::from(p); + } + let home = std::env::var_os("HOME").unwrap_or_default(); + PathBuf::from(home).join(".memmesh").join("logs") + } + + /// The unified engine log file (`/memmesh.log`). + pub fn log_file() -> PathBuf { + Self::log_dir().join("memmesh.log") + } + /// True iff sync section is fully populated. The engine treats this as /// "attempt SaaS-connected mode" — the token still has to validate /// successfully on its first sync attempt before paid features unlock. @@ -229,6 +282,17 @@ impl Config { self.embeddings = c; } } + + // Backend selection via env. `THINKFLEET_DATABASE_URL` (or the sqlx + // convention `DATABASE_URL`) switches the engine to Postgres and + // supplies the DSN — handy for CI, containers, and one-off overrides + // without editing the TOML. + if let Ok(url) = std::env::var("THINKFLEET_DATABASE_URL").or_else(|_| std::env::var("DATABASE_URL")) { + if !url.is_empty() { + self.database.backend = DatabaseBackend::Postgres; + self.database.url = Some(url); + } + } } } diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index 152e934..0d988d0 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -25,5 +25,6 @@ tracing-subscriber.workspace = true config.workspace = true serde.workspace = true serde_json.workspace = true +toml.workspace = true axum.workspace = true tower-http.workspace = true diff --git a/crates/server/src/http.rs b/crates/server/src/http.rs index 51b32f0..4a29fca 100644 --- a/crates/server/src/http.rs +++ b/crates/server/src/http.rs @@ -1,14 +1,18 @@ // Copyright 2026 Thinkfleet AI, LLC Licensed under the Apache License, Version 2.0. -//! HTTP REST API for the GUI (thinkfleet-desktop). +//! HTTP REST API + web console for memmesh. //! //! Bound to loopback by default — this is a local-only management API, -//! not an internet-facing service. The GUI process talks to it via -//! `127.0.0.1:`. +//! not an internet-facing service. The `memmesh console` command serves +//! the embedded UI (see `ui/index.html`) from the same origin, so the +//! page talks to these endpoints with no CORS/auth friction. //! //! Endpoints: //! -//! GET /health — liveness + version + counts +//! GET / — embedded web console (HTML) +//! GET /health — liveness + version + backend +//! GET /stats — counts by scope / status / type +//! GET /memory — list/search (query params) //! GET /memory/:id — fetch one memory item //! POST /memory — upsert //! DELETE /memory/:id — soft delete (status=rejected) @@ -16,27 +20,52 @@ //! POST /search — body = MemoryFilter+limit //! POST /memory/:id/touch — bump lastAccessedAt //! POST /memory/:id/supersede — body = { byId } +//! POST /consolidate — dedup (dryRun previews) +//! GET /config — read ~/.memmesh/config.toml +//! PUT /config — write config.toml (validated) +//! GET /logs?lines=N — tail the unified engine log //! -//! Response shape is JSON; errors carry a stable `code` field for the GUI +//! Response shape is JSON; errors carry a stable `code` field for the UI //! to switch on. No auth in v1 — loopback only. Token auth lands when the //! API gets exposed beyond localhost. use axum::{ extract::{Path, Query, State}, http::StatusCode, - response::{IntoResponse, Json}, - routing::{delete, get, post}, + response::{Html, IntoResponse, Json}, + routing::{get, post}, Router, }; -use memory_core::MemoryItem; +use memory_core::{ + config::{Config, DatabaseBackend}, + MemoryItem, MemoryScope, +}; use memory_storage::{ observe::{ObserveRequest, ObserveResponse}, MemoryFilter, MemoryQuery, Storage, StorageError, }; use serde::{Deserialize, Serialize}; +use serde_json::json; use std::sync::Arc; use tower_http::cors::{Any, CorsLayer}; +/// The web console single-page app, baked into the binary at build time. +const INDEX_HTML: &str = include_str!("ui/index.html"); + +/// Parse a scope string to `MemoryScope`, returning `None` for unknown +/// values (treated as "no scope filter" by the query layer). +fn parse_scope_opt(s: Option) -> Option { + match s.as_deref()? { + "platform" => Some(MemoryScope::Platform), + "project" => Some(MemoryScope::Project), + "location" => Some(MemoryScope::Location), + "agent" => Some(MemoryScope::Agent), + "user" => Some(MemoryScope::User), + "session" => Some(MemoryScope::Session), + _ => None, + } +} + struct AppState { storage: Arc, } @@ -227,6 +256,296 @@ async fn supersede_memory( } } +// ── Web console (embedded SPA) ────────────────────────────── + +async fn index() -> Html<&'static str> { + Html(INDEX_HTML) +} + +// ── /stats ────────────────────────────────────────────────── + +async fn stats(State(s): State>) -> axum::response::Response { + match s.storage.memory_stats(&MemoryFilter::default()).await { + Ok(st) => Json(json!({ + "total": st.total, + "byScope": st.by_scope, + "byStatus": st.by_status, + "byType": st.by_type, + "withEmbedding": st.with_embedding, + "withoutEmbedding": st.without_embedding, + "patternCount": st.pattern_count, + })) + .into_response(), + Err(e) => map_err(e), + } +} + +// ── GET /memory (list / text search) ──────────────────────── + +#[derive(Deserialize)] +struct ListQuery { + query: Option, + project: Option, + scope: Option, + #[serde(rename = "type")] + kind: Option, + limit: Option, + offset: Option, +} + +async fn list_memory( + State(s): State>, + Query(q): Query, +) -> axum::response::Response { + let filter = MemoryFilter { + project_id: q.project, + scope: parse_scope_opt(q.scope), + kind: q.kind, + ..Default::default() + }; + // Hybrid semantic + lexical + recency ranking; empty query returns the + // newest items ranked by recency (falls back to lexical when embeddings + // are off). Same path the CLI `search` uses. + match memory_storage::search::search( + s.storage.as_ref(), + &filter, + q.query.as_deref(), + q.limit.unwrap_or(50), + q.offset.unwrap_or(0), + ) + .await + { + Ok(rows) => Json(rows).into_response(), + Err(e) => err("search", StatusCode::INTERNAL_SERVER_ERROR, e.to_string()), + } +} + +// ── /consolidate ──────────────────────────────────────────── + +#[derive(Deserialize)] +struct ConsolidateBody { + project: Option, + scope: Option, + #[serde(rename = "type")] + kind: Option, + threshold: Option, + // The web console sends camelCase `dryRun`; keep `dry_run` as an alias so + // scripts/CLI-style callers work too. Getting this wrong is dangerous: a + // missed `dryRun` silently turns a "preview" into a real collapse. + #[serde(default, rename = "dryRun", alias = "dry_run")] + dry_run: bool, +} + +async fn consolidate_handler( + State(s): State>, + Json(body): Json, +) -> axum::response::Response { + let filter = MemoryFilter { + project_id: body.project, + scope: parse_scope_opt(body.scope), + kind: body.kind, + ..Default::default() + }; + let threshold = body.threshold.unwrap_or(0.95); + match memory_storage::consolidate::consolidate(s.storage.as_ref(), &filter, threshold, body.dry_run) + .await + { + Ok(r) => Json(json!({ + "scanned": r.scanned, + "threshold": r.threshold, + "semantic": r.semantic, + "dryRun": r.dry_run, + "collapsed": r.collapses.len(), + "collapses": r.collapses.iter().map(|c| json!({ + "loserId": c.loser_id, + "survivorId": c.survivor_id, + "similarity": c.similarity, + })).collect::>(), + })) + .into_response(), + Err(e) => err("consolidate", StatusCode::INTERNAL_SERVER_ERROR, e.to_string()), + } +} + +// ── /config ───────────────────────────────────────────────── + +async fn get_config() -> axum::response::Response { + let path = Config::resolve_path(); + let text = std::fs::read_to_string(&path).unwrap_or_default(); + // Empty/absent file → hand back the serialized effective defaults so the + // editor shows a real starting point instead of a blank box. + let toml_text = if text.trim().is_empty() { + toml::to_string_pretty(&Config::load_or_default()).unwrap_or_default() + } else { + text + }; + Json(json!({ "path": path.display().to_string(), "toml": toml_text })).into_response() +} + +async fn put_config(body: String) -> axum::response::Response { + // Validate before persisting so a typo can't leave the engine unable to + // start. We write the user's raw text (comments preserved), not a + // re-serialized struct. + if let Err(e) = toml::from_str::(&body) { + return err("config_invalid", StatusCode::BAD_REQUEST, format!("invalid config TOML: {e}")); + } + let path = Config::resolve_path(); + if let Some(parent) = path.parent() { + let _ = std::fs::create_dir_all(parent); + } + match std::fs::write(&path, &body) { + Ok(()) => Json(json!({ "ok": true, "path": path.display().to_string() })).into_response(), + Err(e) => err("config_write", StatusCode::INTERNAL_SERVER_ERROR, e.to_string()), + } +} + +// ── /logs ─────────────────────────────────────────────────── + +#[derive(Deserialize)] +struct LogsQuery { + lines: Option, +} + +async fn logs_handler(Query(q): Query) -> axum::response::Response { + let path = Config::log_file(); + let n = q.lines.unwrap_or(200).min(5000); + let text = std::fs::read_to_string(&path).unwrap_or_default(); + let all: Vec<&str> = text.lines().collect(); + let start = all.len().saturating_sub(n); + let tail: Vec = all[start..].iter().map(|s| s.to_string()).collect(); + Json(json!({ + "path": path.display().to_string(), + "total": all.len(), + "lines": tail, + })) + .into_response() +} + +// ── /database (backend selection + migration) ─────────────── + +/// Report the storage backend this process is actually running on +/// (`current`) versus what config would use on next start (`configured`). +/// They differ after a switch until the engine is restarted. +async fn get_database(State(s): State>) -> axum::response::Response { + let cfg = Config::load_or_default(); + Json(json!({ + "current": s.storage.backend(), + "configured": cfg.database.backend.as_str(), + "url": cfg.database.url, + })) + .into_response() +} + +#[derive(Deserialize)] +struct DbUrlBody { + url: String, +} + +/// Probe a Postgres URL without saving anything — validates host, creds, and +/// reachability so the user gets a clear yes/no before committing. Bounded by +/// a short timeout: sqlx's default acquire timeout is ~30s, which would make a +/// wrong host or port feel like a hang in the UI. +async fn test_database(Json(b): Json) -> axum::response::Response { + let connect = memory_storage::postgres::PostgresStore::connect(&b.url); + let body = match tokio::time::timeout(std::time::Duration::from_secs(6), connect).await { + Ok(Ok(_)) => json!({ "ok": true, "message": "connected" }), + Ok(Err(e)) => json!({ "ok": false, "message": e.to_string() }), + Err(_) => json!({ "ok": false, "message": "connection timed out after 6s — check host/port" }), + }; + Json(body).into_response() +} + +/// One-time copy of memory items from the CURRENT backend into a Postgres +/// target: connect, run migrations, then page through and re-`save` each row. +/// Idempotent (save is an upsert), so re-running is safe. +async fn copy_database( + State(s): State>, + Json(b): Json, +) -> axum::response::Response { + let dest = match memory_storage::postgres::PostgresStore::connect(&b.url).await { + Ok(d) => d, + Err(e) => return err("pg_connect", StatusCode::BAD_GATEWAY, e.to_string()), + }; + if let Err(e) = dest.migrate().await { + return err("pg_migrate", StatusCode::INTERNAL_SERVER_ERROR, e.to_string()); + } + let page: u32 = 500; + let mut offset: u32 = 0; + let mut copied: u64 = 0; + loop { + let rows = match s + .storage + .query(&MemoryQuery { + filter: MemoryFilter::default(), + limit: Some(page), + offset: Some(offset), + }) + .await + { + Ok(r) => r, + Err(e) => return map_err(e), + }; + if rows.is_empty() { + break; + } + let n = rows.len() as u32; + for item in &rows { + if let Err(e) = dest.save(item).await { + return err( + "pg_write", + StatusCode::INTERNAL_SERVER_ERROR, + format!("copied {copied} item(s) then failed: {e}"), + ); + } + copied += 1; + } + if n < page { + break; + } + offset += page; + } + let dest_total = dest.count_items().await.unwrap_or(-1); + Json(json!({ "copied": copied, "destTotal": dest_total })).into_response() +} + +#[derive(Deserialize)] +struct SetDbBody { + backend: String, + #[serde(default)] + url: Option, +} + +/// Persist the backend choice to `~/.memmesh/config.toml`. Takes effect on +/// the next start of each `memmesh` process. +async fn put_database(Json(b): Json) -> axum::response::Response { + let backend = match b.backend.as_str() { + "sqlite" => DatabaseBackend::Sqlite, + "postgres" => DatabaseBackend::Postgres, + other => { + return err( + "bad_backend", + StatusCode::BAD_REQUEST, + format!("unknown backend '{other}'"), + ) + } + }; + let url = b.url.filter(|u| !u.trim().is_empty()); + if backend == DatabaseBackend::Postgres && url.is_none() { + return err( + "missing_url", + StatusCode::BAD_REQUEST, + "postgres backend requires a connection url", + ); + } + let mut cfg = Config::load_or_default(); + cfg.database.backend = backend; + cfg.database.url = url; + match cfg.save() { + Ok(()) => Json(json!({ "ok": true, "backend": backend.as_str() })).into_response(), + Err(e) => err("config_write", StatusCode::INTERNAL_SERVER_ERROR, e.to_string()), + } +} + // ── Public entrypoint ─────────────────────────────────────── /// Build the HTTP router. Caller owns binding + serving so the same router @@ -234,13 +553,21 @@ async fn supersede_memory( pub fn router(storage: Arc) -> Router { let state = AppState { storage }; Router::new() + .route("/", get(index)) .route("/health", get(health)) + .route("/stats", get(stats)) .route("/memory/:id", get(get_memory).delete(delete_memory)) - .route("/memory", post(save_memory)) + .route("/memory", post(save_memory).get(list_memory)) .route("/memory/:id/touch", post(touch_memory)) .route("/memory/:id/supersede", post(supersede_memory)) .route("/search", post(search_memory)) .route("/observe", post(observe_handler)) + .route("/consolidate", post(consolidate_handler)) + .route("/config", get(get_config).put(put_config)) + .route("/database", get(get_database).put(put_database)) + .route("/database/test", post(test_database)) + .route("/database/copy", post(copy_database)) + .route("/logs", get(logs_handler)) // CORS open for any origin so the desktop renderer (file://) can hit // the localhost endpoint without preflight rejections. Bind to // 127.0.0.1 keeps the surface local-only. @@ -260,8 +587,3 @@ pub async fn serve(storage: Arc, addr: &str) -> anyhow::Result<() axum::serve(listener, router(storage)).await?; Ok(()) } - -// `delete` route name was being shadowed by axum's `delete` import — keep -// the import explicit so the warning stays quiet on stable. -#[allow(unused_imports)] -use delete as _delete; diff --git a/crates/server/src/ui/index.html b/crates/server/src/ui/index.html new file mode 100644 index 0000000..6340120 --- /dev/null +++ b/crates/server/src/ui/index.html @@ -0,0 +1,393 @@ + + + + + +MemMesh Console + + + +
+ MemMesh +
connecting…
+
+
+
+ +
+ +
+
+
+
Maintenance — consolidate duplicates
+
Finds near-duplicate memories and non-destructively collapses them (history kept via supersede). Preview first.
+
+ + + +
+
+
+
+ + + + + + + + + +
+
+ + + + From 1a2edb55e35a42f563b8a59df75194a8c843bcef Mon Sep 17 00:00:00 2001 From: rrader2890 Date: Fri, 7 Aug 2026 20:33:24 -0400 Subject: [PATCH 2/4] feat(console): knowledge-graph explorer, context preview, provenance - Graph tab: force-directed KG of entities + typed edges (zero-LLM extractor), hover-highlight, legend by entity type; GET /graph (edges deduped on read) - /graph/rebuild: backfill the graph from all stored memories (idempotent) - Context tab: shows the exact claude-context block injected for a query - Memories: provenance line (source, confidence, learned date, supersession) --- crates/server/src/http.rs | 113 ++++++++++++++++++- crates/server/src/ui/index.html | 189 +++++++++++++++++++++++++++++++- 2 files changed, 300 insertions(+), 2 deletions(-) diff --git a/crates/server/src/http.rs b/crates/server/src/http.rs index 4a29fca..ee5ebbb 100644 --- a/crates/server/src/http.rs +++ b/crates/server/src/http.rs @@ -41,8 +41,9 @@ use memory_core::{ MemoryItem, MemoryScope, }; use memory_storage::{ + graph_extractor::{extract_and_wire, GraphContext}, observe::{ObserveRequest, ObserveResponse}, - MemoryFilter, MemoryQuery, Storage, StorageError, + EdgeFilter, EntityFilter, MemoryFilter, MemoryQuery, Storage, StorageError, }; use serde::{Deserialize, Serialize}; use serde_json::json; @@ -546,6 +547,114 @@ async fn put_database(Json(b): Json) -> axum::response::Response { } } +// ── /graph (knowledge-graph explorer) ─────────────────────── + +/// Return the full entity + edge set for the graph view. Entities carry +/// canonical name + type; edges carry subject/predicate/object (object is +/// either another entity id or a literal string). +async fn get_graph(State(s): State>) -> axum::response::Response { + let entities = match s.storage.query_entities(&EntityFilter::default()).await { + Ok(e) => e, + Err(e) => return map_err(e), + }; + let edges = match s.storage.query_edges(&EdgeFilter::default()).await { + Ok(e) => e, + Err(e) => return map_err(e), + }; + // Collapse duplicate edges (same subject→predicate→object) for display — + // the extractor can write the same relationship more than once (e.g. after + // a graph rebuild re-scans already-ingested memories), and the graph view + // wants one line per distinct relationship. + let mut seen = std::collections::HashSet::new(); + let edges_json: Vec<_> = edges + .iter() + .filter(|e| { + let key = format!( + "{}|{}|{}", + e.subject_id, + e.predicate, + e.object_id.as_deref().or(e.object_literal.as_deref()).unwrap_or("") + ); + seen.insert(key) + }) + .map(|e| json!({ + "id": e.id, + "subject": e.subject_id, + "predicate": e.predicate, + "object": e.object_id, + "objectLiteral": e.object_literal, + })) + .collect(); + Json(json!({ + "entities": entities.iter().map(|e| json!({ + "id": e.id, + "name": e.canonical_name, + "type": e.type_, + "aliases": e.aliases, + })).collect::>(), + "edges": edges_json, + })) + .into_response() +} + +/// Backfill the graph by re-running the zero-LLM extractor over every stored +/// memory. Idempotent — the entity resolver dedupes and edges upsert — so +/// this is safe to run repeatedly (e.g. after importing memories that were +/// saved before graph extraction existed). +async fn rebuild_graph(State(s): State>) -> axum::response::Response { + let page: u32 = 500; + let mut offset: u32 = 0; + let (mut scanned, mut ents, mut edges) = (0u64, 0u64, 0u64); + loop { + let rows = match s + .storage + .query(&MemoryQuery { + filter: MemoryFilter::default(), + limit: Some(page), + offset: Some(offset), + }) + .await + { + Ok(r) => r, + Err(e) => return map_err(e), + }; + if rows.is_empty() { + break; + } + let n = rows.len() as u32; + for item in &rows { + let ctx = GraphContext { + platform_id: item.platform_id.clone(), + project_id: item.project_id.clone(), + scope: item.scope, + source_memory_id: item.id.clone(), + }; + if let Ok(st) = extract_and_wire(s.storage.as_ref(), &item.content, &ctx).await { + ents += st.entities_resolved as u64; + edges += st.edges_written as u64; + } + scanned += 1; + } + if n < page { + break; + } + offset += page; + } + let total_entities = s + .storage + .query_entities(&EntityFilter::default()) + .await + .map(|v| v.len()) + .unwrap_or(0); + Json(json!({ + "scanned": scanned, + "entitiesResolved": ents, + "edgesWritten": edges, + "totalEntities": total_entities, + })) + .into_response() +} + // ── Public entrypoint ─────────────────────────────────────── /// Build the HTTP router. Caller owns binding + serving so the same router @@ -564,6 +673,8 @@ pub fn router(storage: Arc) -> Router { .route("/observe", post(observe_handler)) .route("/consolidate", post(consolidate_handler)) .route("/config", get(get_config).put(put_config)) + .route("/graph", get(get_graph)) + .route("/graph/rebuild", post(rebuild_graph)) .route("/database", get(get_database).put(put_database)) .route("/database/test", post(test_database)) .route("/database/copy", post(copy_database)) diff --git a/crates/server/src/ui/index.html b/crates/server/src/ui/index.html index 6340120..e94586d 100644 --- a/crates/server/src/ui/index.html +++ b/crates/server/src/ui/index.html @@ -77,6 +77,20 @@ label.check { display: inline-flex; align-items: center; gap: 6px; color: var(--muted); font-size: 13px; } .hidden { display: none; } .hint { font-size: 12px; color: var(--muted); margin: 4px 0 12px; } + .hint code { font-family: var(--mono); background: var(--panel-2); padding: 1px 5px; border-radius: 4px; } + .graphwrap { background: var(--panel); border: 1px solid var(--border); border-radius: 12px; height: 62vh; overflow: hidden; } + #graphSvg { width: 100%; height: 100%; display: block; cursor: grab; } + .gedge { stroke: var(--border); stroke-width: 1.5; } + .gedge.hi { stroke: var(--accent); stroke-width: 2.5; } + .gedge-label { fill: var(--muted); font-size: 10px; font-family: var(--mono); } + .gnode circle { stroke: var(--panel); stroke-width: 2; cursor: pointer; transition: r .1s; } + .gnode text { fill: var(--fg); font-size: 11px; pointer-events: none; } + .gnode.dim { opacity: .25; } + .legend { display: flex; gap: 12px; flex-wrap: wrap; } + .legend span { display: inline-flex; align-items: center; gap: 5px; font-size: 11px; color: var(--muted); } + .legend i { width: 10px; height: 10px; border-radius: 50%; display: inline-block; } + .prov { font-size: 11px; color: var(--muted); margin-top: 8px; font-family: var(--mono); display: flex; gap: 10px; flex-wrap: wrap; } + .prov .sup { color: var(--warn); } @@ -89,6 +103,8 @@ @@ -125,6 +141,32 @@
+ + + + + + + + + + + + + + + @@ -328,8 +328,7 @@ $('#verText').textContent = `${h.name} v${h.version} · ${h.backend}`; const st = await api('/stats'); $('#statCards').innerHTML = [ - ['memories', st.total], ['indexed', st.withEmbedding], - ['un-indexed', st.withoutEmbedding], ['behavior patterns', st.patternCount], + ['memories', st.total], ['indexed', st.withEmbedding], ['un-indexed', st.withoutEmbedding], ].map(([l, n]) => `
${n}
${esc(l)}
`).join(''); renderBars('#barsScope', st.byScope); renderBars('#barsType', st.byType); renderBars('#barsStatus', st.byStatus); const cov = st.total ? Math.round(st.withEmbedding / st.total * 100) : 0;