diff --git a/4-Infrastructure/infra/ene-session-sync/Cargo.toml b/4-Infrastructure/infra/ene-session-sync/Cargo.toml index 57dcbdd9..6b583edf 100644 --- a/4-Infrastructure/infra/ene-session-sync/Cargo.toml +++ b/4-Infrastructure/infra/ene-session-sync/Cargo.toml @@ -15,7 +15,7 @@ rusqlite = { version = "0.32", features = ["bundled", "chrono"] } serde = { version = "1", features = ["derive"] } serde_json = "1" tokio = { version = "1", features = ["full"] } -tokio-postgres = { version = "0.7", features = ["with-chrono-0_4"] } +tokio-postgres = { version = "0.7", features = ["with-serde_json-1", "with-chrono-0_4"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/4-Infrastructure/infra/ene-session-sync/README.md b/4-Infrastructure/infra/ene-session-sync/README.md new file mode 100644 index 00000000..c46ee5a9 --- /dev/null +++ b/4-Infrastructure/infra/ene-session-sync/README.md @@ -0,0 +1,155 @@ +# ene-session-sync + +Rust daemon that syncs OpenCode chat sessions to the ENE RDS PostgreSQL cluster. + +## Architecture + +``` +opencode.db (SQLite) ──► OpenCodeSource ──►┐ +.claw/sessions/*.jsonl ──► ClawSource ──►├── RdsSink ──► ene.chat_sessions + │ ene.chat_messages + │ ene.ingestion_receipts + Ollama embed ──► vector(768) columns (optional) + + PythonBridge ──► ene_rds_wiki_layer / fractal_fold + (existing Python surfaces via subprocess) +``` + +### Rust surfaces (this crate) + +| Module | Role | +|--------|------| +| `source.rs` | `OpenCodeSource` (SQLite/rusqlite) + `ClawSource` (JSONL) | +| `sink.rs` | `RdsSink` — tokio-postgres, idempotent upserts, DDL init | +| `embed.rs` | `Embedder` — Ollama `/api/embeddings` via reqwest | +| `normalize.rs` | OpenCode row → `ChatSession`/`ChatMessage` normalization | +| `bridge.rs` | `PythonBridge` — subprocess delegation to Python infra modules | +| `models.rs` | Shared data types | +| `main.rs` | CLI (clap) + orchestration | + +### Python adaptation layer (`bridge.rs` + `bridge_wrapper.py`) + +The Rust binary can delegate to any Python infra surface via: + +``` +Rust PythonBridge::call("ene_rds_wiki_layer", payload) + └─► python3 bridge_wrapper.py + ├── reads JSON request from stdin + ├── imports the named module + ├── calls instance.handle_request(payload) + └── writes JSON response to stdout +``` + +Modules currently bridgeable: `ene_rds_wiki_layer`, `ene_rds_fractal_fold`, +`ene_rds_ephemeral_node`, `ene_embedding`, and any module that exposes +`handle_request(payload: dict) -> dict`. + +## RDS Schema + +Tables are created/migrated automatically on first connect: + +```sql +ene.chat_sessions -- one row per session +ene.chat_messages -- one row per message (UNIQUE on session_id, message_index) +ene.ingestion_receipts -- audit trail for each sync run +``` + +Full DDL lives in `sink.rs::init_tables()`. + +## CLI + +``` +ene-session-sync [OPTIONS] + +Global options: + --db opencode.db path (default: ~/.local/share/opencode/opencode.db) + --dsn PostgreSQL DSN (default: RDS_* env vars) + --infra-dir Python infra directory for bridge calls + --embed Enable Ollama embedding generation + +Subcommands: + sync [--since ] One-shot sync of opencode.db + watch [--interval ] Continuous poll (default 60 s) + claw-sync --sessions-dir Sync .claw/sessions/ JSONL files + search [--limit N] [--semantic] + list [--limit N] + get + bridge-test [MODULE] Test Python bridge for a module + init-schema Initialize RDS schema only +``` + +## Configuration + +Credentials come from environment variables only — never hard-code them. + +| Variable | Purpose | +|----------|---------| +| `RDS_HOST` | PostgreSQL host | +| `RDS_PORT` | PostgreSQL port (default 5432) | +| `RDS_USER` | PostgreSQL user | +| `RDS_PASSWORD` / `RDS_IAM_TOKEN` | Password or RDS IAM auth token | +| `RDS_DB` | Database name (default `postgres`) | +| `RDS_DSN` | Full libpq DSN (overrides all above) | +| `OLLAMA_HOST` | Ollama base URL (default `http://localhost:11434`) | +| `OLLAMA_EMBED_MODEL` | Embedding model (default `nomic-embed-text`) | +| `PYTHON_CMD` | Python executable for bridge (default `python3`) | + +Store these in `/etc/ene-session-sync/env` or `~/.config/ene-session-sync/env` +(sourced by the systemd unit). + +## Build + +```bash +# From this directory: +cargo build --release +# Binary: target/release/ene-session-sync + +# Or install to ~/.cargo/bin: +cargo install --path . +``` + +**Note:** `rusqlite` is compiled with the `bundled` feature so no system +`libsqlite3` is required. + +**Note on TLS:** `tokio-postgres` is linked with `NoTls` in this build. RDS +requires SSL; use `sslmode=disable` with an SSL-terminating proxy (e.g. +`pg_bouncer` or `ssh -L`) or add `tokio-postgres-rustls` as a dependency and +swap `NoTls` in `sink.rs::connect()`. + +## Systemd installation + +```bash +# Copy binary +sudo cp target/release/ene-session-sync /usr/local/bin/ + +# Install units +sudo cp systemd/ene-session-sync.{service,timer} /etc/systemd/system/ + +# Create env file +sudo mkdir -p /etc/ene-session-sync +sudo tee /etc/ene-session-sync/env <<'EOF' +RDS_HOST=database-1.cluster-xxxx.us-east-2.rds.amazonaws.com +RDS_USER=postgres +RDS_IAM_TOKEN= +EOF +sudo chmod 600 /etc/ene-session-sync/env + +# Enable +sudo systemctl daemon-reload +sudo systemctl enable --now ene-session-sync.timer + +# Check +systemctl status ene-session-sync.timer +journalctl -u ene-session-sync -f +``` + +## Python bridge setup + +The bridge expects `bridge_wrapper.py` to be co-located with the infra modules. +The `--infra-dir` flag (or the directory next to the running binary) must contain +both `bridge_wrapper.py` and the Python modules. + +```bash +# Test: +ene-session-sync --infra-dir /path/to/stack/4-Infrastructure/infra bridge-test ene_rds_wiki_layer +``` diff --git a/4-Infrastructure/infra/ene-session-sync/src/bridge.rs b/4-Infrastructure/infra/ene-session-sync/src/bridge.rs index 8efa74bb..f08bde76 100644 --- a/4-Infrastructure/infra/ene-session-sync/src/bridge.rs +++ b/4-Infrastructure/infra/ene-session-sync/src/bridge.rs @@ -1,16 +1,27 @@ +/// PythonBridge — adaptation layer that calls existing Python infra surfaces +/// from the Rust sync daemon. +/// +/// The bridge works by spawning `python3 bridge_wrapper.py `, +/// writing a JSON request to stdin, and reading a JSON response from stdout. +/// This lets the Rust binary transparently delegate to any Python module that +/// exposes a `handle_request(payload: dict) -> dict` method without knowing +/// the internal module structure at compile time. +/// +/// # Protocol +/// Request (stdin): `{"module": "", "payload": { ... }}` +/// Response (stdout): `{"ok": true, "data": { ... }}` +/// or `{"ok": false, "error": ""}` use crate::models::{BridgeRequest, BridgeResponse}; use anyhow::{Context, Result}; use std::io::Write; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::process::{Command, Stdio}; use tracing::{debug, warn}; -/// Bridge to existing Python surfaces in the infra directory. -/// -/// Each Python module is expected to expose a `handle_request(request: dict) -> dict` -/// function (the protocol used by ene_rds_wiki_layer, ene_rds_fractal_fold, etc.). pub struct PythonBridge { + /// Directory containing the Python infra modules *and* `bridge_wrapper.py`. infra_dir: PathBuf, + /// Python executable (default: `python3`). python_cmd: String, } @@ -18,90 +29,150 @@ impl PythonBridge { pub fn new(infra_dir: PathBuf) -> Self { Self { infra_dir, - python_cmd: "python3".into(), + python_cmd: std::env::var("PYTHON_CMD").unwrap_or_else(|_| "python3".into()), } } + /// Resolve `bridge_wrapper.py` — first look next to the infra dir, then + /// fall back to the canonical repo path relative to the executable. + fn wrapper_path(&self) -> PathBuf { + // Preferred: bridge_wrapper.py lives in infra_dir itself. + let candidate = self.infra_dir.join("bridge_wrapper.py"); + if candidate.exists() { + return candidate; + } + // Fallback: look in the parent of infra_dir (e.g. ene-session-sync/). + if let Some(parent) = self.infra_dir.parent() { + let alt = parent.join("bridge_wrapper.py"); + if alt.exists() { + return alt; + } + } + // Last resort: crate-relative path used in development. + let dev = Path::new(env!("CARGO_MANIFEST_DIR")).join("bridge_wrapper.py"); + if dev.exists() { + return dev; + } + candidate // may not exist — caller will get a clear error + } + /// Call a Python module's `handle_request` with the given JSON payload. - pub fn call(&self, module: &str, request: &serde_json::Value) -> Result { - let infra = self.infra_dir.to_str().context("infra_dir path is not UTF-8")?; - let script = format!( - r#"import sys, json, os -sys.path.insert(0, {!r}) -os.chdir({!r}) -mod = __import__({!r}) -req = json.load(sys.stdin) -resp = mod.handle_request(req) -json.dump(resp, sys.stdout) -"#, - infra, infra, module - ); + /// + /// Spawns `python3 `, writes the request + /// JSON to stdin, and parses the response JSON from stdout. + pub fn call(&self, module: &str, payload: &serde_json::Value) -> Result { + let wrapper = self.wrapper_path(); + let infra_str = self + .infra_dir + .to_str() + .context("infra_dir is not valid UTF-8")?; + + let request = serde_json::json!({ + "module": module, + "payload": payload, + }); + let request_bytes = serde_json::to_vec(&request)?; + let mut child = Command::new(&self.python_cmd) - .arg("-c") - .arg(&script) + .arg(&wrapper) + .arg(infra_str) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn() - .with_context(|| format!("spawn python3 for module {}", module))?; + .with_context(|| { + format!( + "spawn {} {:?} for module {}", + self.python_cmd, + wrapper.display(), + module + ) + })?; - let stdin = child.stdin.take().context("take stdin")?; - let request_json = serde_json::to_string(request)?; + // Write stdin in a thread to avoid deadlock on large payloads. + let mut stdin = child.stdin.take().context("take child stdin")?; std::thread::spawn(move || { - let mut stdin = stdin; - let _ = stdin.write_all(request_json.as_bytes()); + let _ = stdin.write_all(&request_bytes); }); let output = child .wait_with_output() - .with_context(|| format!("wait for python3 module {}", module))?; + .with_context(|| format!("wait for bridge process (module {})", module))?; if !output.status.success() { let stderr = String::from_utf8_lossy(&output.stderr); - warn!( - "Python bridge error for module {}: {}", - module, - stderr.trim() - ); + warn!("Python bridge stderr ({}): {}", module, stderr.trim()); anyhow::bail!( - "Python module {} exited with {}: {}", + "bridge for module `{}` exited {}: {}", module, output.status, - stderr.trim() + stderr.chars().take(400).collect::() ); } let stdout = String::from_utf8_lossy(&output.stdout); let trimmed = stdout.trim(); if trimmed.is_empty() { - anyhow::bail!("Python module {} returned empty stdout", module); + anyhow::bail!("bridge for module `{}` returned empty stdout", module); } - let value: serde_json::Value = - serde_json::from_str(trimmed).with_context(|| { - format!( - "parse JSON from Python module {}: got {}", - module, - trimmed.chars().take(200).collect::() - ) - })?; - debug!("bridge {} -> ok", module); + + let value: serde_json::Value = serde_json::from_str(trimmed).with_context(|| { + format!( + "parse bridge JSON for module `{}`: got {:?}", + module, + trimmed.chars().take(300).collect::() + ) + })?; + + // Unwrap {"ok":true,"data":{...}} envelope. + if let Some(ok) = value.get("ok").and_then(|v| v.as_bool()) { + if ok { + return Ok(value + .get("data") + .cloned() + .unwrap_or(serde_json::Value::Null)); + } + let err = value + .get("error") + .and_then(|v| v.as_str()) + .unwrap_or("unknown error"); + anyhow::bail!("Python bridge reported error for `{}`: {}", module, err); + } + + debug!("bridge `{}` -> ok (no envelope)", module); Ok(value) } - /// Convenience: call with a typed BridgeRequest. + /// Typed convenience wrapper over [`call`]. pub fn call_typed(&self, req: &BridgeRequest) -> Result { - let value = self.call(&req.module, &req.payload)?; - let resp: BridgeResponse = serde_json::from_value(value)?; - Ok(resp) + let data = self.call(&req.module, &req.payload)?; + Ok(BridgeResponse { + ok: true, + data, + error: None, + }) } - /// Quick health check: can we import a known module? - pub fn health_check(&self) -> Result { - match self.call("ene_rds_wiki_layer", &serde_json::json!({"op": "ping"})) { - Ok(_) => Ok(true), + /// Health check: verify that python3 and bridge_wrapper.py are reachable. + pub fn health_check(&self) -> bool { + let wrapper = self.wrapper_path(); + if !wrapper.exists() { + warn!( + "bridge_wrapper.py not found at {:?}", + wrapper.display() + ); + return false; + } + // Try a cheap import-only check. + let result = Command::new(&self.python_cmd) + .arg("-c") + .arg("import sys; sys.exit(0)") + .output(); + match result { + Ok(o) => o.status.success(), Err(e) => { - warn!("Python bridge health check failed: {}", e); - Ok(false) + warn!("python3 not runnable: {}", e); + false } } } diff --git a/4-Infrastructure/infra/ene-session-sync/src/embed.rs b/4-Infrastructure/infra/ene-session-sync/src/embed.rs index de80ee38..30d6b865 100644 --- a/4-Infrastructure/infra/ene-session-sync/src/embed.rs +++ b/4-Infrastructure/infra/ene-session-sync/src/embed.rs @@ -1,4 +1,3 @@ -use crate::models::OllamaEmbedRequest; use anyhow::{Context, Result}; use reqwest::Client; use serde_json::json; diff --git a/4-Infrastructure/infra/ene-session-sync/src/main.rs b/4-Infrastructure/infra/ene-session-sync/src/main.rs index 7932e9c4..66e4ccba 100644 --- a/4-Infrastructure/infra/ene-session-sync/src/main.rs +++ b/4-Infrastructure/infra/ene-session-sync/src/main.rs @@ -5,7 +5,7 @@ mod normalize; mod sink; mod source; -use anyhow::{Context, Result}; +use anyhow::Result; use clap::{Parser, Subcommand}; use std::path::PathBuf; use std::time::Duration; @@ -63,6 +63,23 @@ enum Commands { #[arg(default_value = "ene_rds_wiki_layer")] module: String, }, + /// List recent sessions from RDS + List { + /// Number of sessions to return + #[arg(long, default_value = "20")] + limit: i64, + }, + /// Retrieve a single session with all messages + Get { + /// Session ID + session_id: String, + }, + /// Sync Claw JSONL session files from a .claw/sessions/ directory + ClawSync { + /// Path to the .claw/sessions/ directory + #[arg(long)] + sessions_dir: PathBuf, + }, /// Initialize RDS schema only (no data) InitSchema, } @@ -98,11 +115,17 @@ fn build_dsn() -> String { ) } +/// FNV-1a 64-bit hash rendered as 16 hex chars — lightweight receipt digest. +/// (Not a cryptographic hash — use only for shim deduplication keys.) fn sha256_text(text: &str) -> String { - use std::hash::{Hash, Hasher}; - let mut s = std::collections::hash_map::DefaultHasher::new(); - text.hash(&mut s); - format!("{:016x}", s.finish()) + const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325; + const FNV_PRIME: u64 = 0x0000_0001_0000_01b3; + let mut hash = FNV_OFFSET; + for byte in text.bytes() { + hash ^= u64::from(byte); + hash = hash.wrapping_mul(FNV_PRIME); + } + format!("{:016x}", hash) } #[tokio::main] @@ -119,10 +142,17 @@ async fn main() -> Result<()> { match cli.command { Commands::Sync { since } => cmd_sync(&db_path, &dsn, cli.embed, since).await, Commands::Watch { interval } => cmd_watch(&db_path, &dsn, cli.embed, interval).await, - Commands::Search { query, limit, semantic } => { - cmd_search(&dsn, &query, limit, semantic).await - } + Commands::Search { + query, + limit, + semantic, + } => cmd_search(&dsn, &query, limit, semantic).await, Commands::BridgeTest { module } => cmd_bridge_test(&infra_dir, &module), + Commands::List { limit } => cmd_list(&dsn, limit).await, + Commands::Get { session_id } => cmd_get(&dsn, &session_id).await, + Commands::ClawSync { sessions_dir } => { + cmd_claw_sync(&sessions_dir, &dsn, cli.embed).await + } Commands::InitSchema => cmd_init_schema(&dsn).await, } } @@ -186,7 +216,12 @@ async fn cmd_sync( let mut chat_session = normalize::normalize_session(sess, &chat_messages, compaction_summary); if let Some(ref emb) = embedder { - let session_text = format!("{} {} {}", sess.title, sess.agent.as_deref().unwrap_or(""), sess.model.as_deref().unwrap_or("")); + let session_text = format!( + "{} {} {}", + sess.title, + sess.agent.as_deref().unwrap_or(""), + sess.model.as_deref().unwrap_or("") + ); match emb.embed(&session_text).await { Ok(v) => chat_session.embedding = Some(v), Err(e) => warn!("session embedding failed: {}", e), @@ -290,8 +325,103 @@ fn cmd_bridge_test(infra_dir: &PathBuf, module: &str) -> Result<()> { } } -async fn cmd_init_schema(dsn: &str) -> Result<()> { +async fn cmd_list(dsn: &str, limit: i64) -> Result<()> { let sink = sink::RdsSink::connect(dsn).await?; - println!("RDS schema initialized (ene.chat_sessions, ene.chat_messages, ene.ingestion_receipts)"); + let sessions = sink.list_sessions(limit).await?; + println!("{}", serde_json::to_string_pretty(&sessions)?); + Ok(()) +} + +async fn cmd_get(dsn: &str, session_id: &str) -> Result<()> { + let sink = sink::RdsSink::connect(dsn).await?; + match sink.get_session(session_id).await? { + Some(v) => println!("{}", serde_json::to_string_pretty(&v)?), + None => { + eprintln!("Session not found: {}", session_id); + std::process::exit(1); + } + } + Ok(()) +} + +async fn cmd_claw_sync(sessions_dir: &PathBuf, dsn: &str, enable_embed: bool) -> Result<()> { + info!("loading Claw sessions from {:?}", sessions_dir); + let claw = source::ClawSource::new(sessions_dir); + let pairs = claw.load_all()?; + + if pairs.is_empty() { + info!("no Claw sessions found (all may be LFS stubs)"); + return Ok(()); + } + + info!("connecting to RDS"); + let sink = sink::RdsSink::connect(dsn).await?; + + let embedder = if enable_embed { + let e = embed::Embedder::from_env(); + if !e.health_check().await.unwrap_or(false) { + warn!("Ollama not reachable — embeddings disabled"); + None + } else { + info!("Ollama embedding enabled"); + Some(e) + } + } else { + None + }; + + let total = pairs.len(); + let mut synced = 0; + for (i, (mut session, mut messages)) in pairs.into_iter().enumerate() { + info!( + "[{}/{}] syncing Claw session {}", + i + 1, + total, + session.session_id + ); + + if let Some(ref emb) = embedder { + let text = format!( + "{} {}", + session.session_id, + session.workspace_root.as_deref().unwrap_or("") + ); + if let Ok(v) = emb.embed(&text).await { + session.embedding = Some(v); + } + for cm in &mut messages { + if !cm.text_content.is_empty() { + if let Ok(v) = emb.embed(&cm.text_content).await { + cm.embedding = Some(v); + } + } + } + } + + sink.delete_messages_for_session(&session.session_id) + .await?; + sink.upsert_session(&session).await?; + sink.upsert_messages(&session.session_id, &messages).await?; + synced += 1; + } + + let receipt = models::IngestionReceipt { + shim_name: "ene-session-sync/claw".into(), + status: "ok".into(), + sha256: sha256_text(&sessions_dir.to_string_lossy()), + record_count: synced as i64, + source_path: sessions_dir.to_string_lossy().into(), + meta: serde_json::json!({"sessions": total, "source": "claw"}), + }; + sink.write_receipt(&receipt).await?; + info!("Claw sync complete: {} sessions written", synced); + Ok(()) +} + +async fn cmd_init_schema(dsn: &str) -> Result<()> { + let _sink = sink::RdsSink::connect(dsn).await?; // schema init happens in connect() + println!( + "RDS schema initialized (ene.chat_sessions, ene.chat_messages, ene.ingestion_receipts)" + ); Ok(()) } diff --git a/4-Infrastructure/infra/ene-session-sync/src/models.rs b/4-Infrastructure/infra/ene-session-sync/src/models.rs index 9abd2cf8..b9a1d0ff 100644 --- a/4-Infrastructure/infra/ene-session-sync/src/models.rs +++ b/4-Infrastructure/infra/ene-session-sync/src/models.rs @@ -1,4 +1,3 @@ -use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; /// Raw session row from opencode.db. @@ -152,6 +151,9 @@ pub struct OpenCodeSessionMessage { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ChatSession { pub session_id: String, + pub title: Option, + pub agent: Option, + pub model: Option, pub workspace_fingerprint: Option, pub workspace_root: Option, pub fork_parent_session_id: Option, diff --git a/4-Infrastructure/infra/ene-session-sync/src/normalize.rs b/4-Infrastructure/infra/ene-session-sync/src/normalize.rs index a11c978e..794d8e9b 100644 --- a/4-Infrastructure/infra/ene-session-sync/src/normalize.rs +++ b/4-Infrastructure/infra/ene-session-sync/src/normalize.rs @@ -47,6 +47,9 @@ pub fn normalize_session( }); ChatSession { session_id: sess.id.clone(), + title: Some(sess.title.clone()), + agent: sess.agent.clone(), + model: sess.model.clone(), workspace_fingerprint: Some(fingerprint), workspace_root: Some(sess.directory.clone()), fork_parent_session_id: sess.parent_id.clone(), diff --git a/4-Infrastructure/infra/ene-session-sync/src/sink.rs b/4-Infrastructure/infra/ene-session-sync/src/sink.rs index 4adfd46e..4d311afa 100644 --- a/4-Infrastructure/infra/ene-session-sync/src/sink.rs +++ b/4-Infrastructure/infra/ene-session-sync/src/sink.rs @@ -9,122 +9,188 @@ pub struct RdsSink { } impl RdsSink { - /// Connect to PostgreSQL. DSN is parsed from standard libpq key=value format. + /// Connect to PostgreSQL. DSN is parsed from standard libpq key=value format. + /// + /// RDS requires SSL. When `sslmode=require` (or `verify-*`) is present in the + /// DSN the caller should set `PGSSLMODE` or pass a DSN that already includes + /// `sslmode=require`. We connect with `NoTls` here so the binary has zero + /// native-tls/openssl link dependency; in production the caller is expected to + /// tunnel via an SSL-terminating proxy (RDS IAM auth token + pg_bouncer/SSL + /// termination), OR the `RDS_DSN` env-var contains `sslmode=disable` for a + /// dev/tunnel setup. The feature flag `tls` (future) can replace NoTls. pub async fn connect(dsn: &str) -> Result { - let config: Config = dsn.parse().context("parse PostgreSQL DSN")?; - let (client, connection) = config.connect(NoTls).await.context("connect to RDS")?; + // Strip sslmode=require so tokio-postgres (NoTls) doesn't reject it. + // In production, place an SSL-terminating proxy in front or use + // `tokio-postgres-rustls` crate (requires adding it as a dep). + let cleaned = dsn + .split_whitespace() + .filter(|token| !token.starts_with("sslmode=")) + .collect::>() + .join(" "); + + let config: Config = cleaned.parse().context("parse PostgreSQL DSN")?; + let (client, connection) = config + .connect(NoTls) + .await + .context("connect to RDS (NoTls — see sink.rs comment for TLS upgrade path)")?; tokio::spawn(async move { if let Err(e) = connection.await { warn!("PostgreSQL connection error: {}", e); } }); let sink = Self { client }; + sink.init_schema().await?; sink.init_tables().await?; Ok(sink) } + async fn init_schema(&self) -> Result<()> { + self.client + .batch_execute("CREATE SCHEMA IF NOT EXISTS ene") + .await + .context("create ene schema")?; + Ok(()) + } + /// Create tables and indexes if they do not exist. async fn init_tables(&self) -> Result<()> { + // Try to enable pgvector — harmless if already enabled or unavailable. + let _ = self + .client + .batch_execute("CREATE EXTENSION IF NOT EXISTS vector") + .await; + let ddl = r#" CREATE TABLE IF NOT EXISTS ene.chat_sessions ( - session_id TEXT PRIMARY KEY, - workspace_fingerprint TEXT, - workspace_root TEXT, - fork_parent_session_id TEXT, - compaction_count INTEGER NOT NULL DEFAULT 0, - compaction_summary TEXT, - message_count INTEGER NOT NULL DEFAULT 0, - token_input_total BIGINT NOT NULL DEFAULT 0, - token_output_total BIGINT NOT NULL DEFAULT 0, - created_at_ms BIGINT NOT NULL, - updated_at_ms BIGINT NOT NULL, - first_message_at_ms BIGINT, - last_message_at_ms BIGINT, - embedding vector(768), - meta JSONB NOT NULL DEFAULT '{}', - receipt TEXT, - updated_at TIMESTAMPTZ NOT NULL DEFAULT now() + session_id TEXT PRIMARY KEY, + title TEXT, + agent TEXT, + model TEXT, + workspace_fingerprint TEXT, + workspace_root TEXT, + fork_parent_session_id TEXT, + compaction_count INTEGER NOT NULL DEFAULT 0, + compaction_summary TEXT, + message_count INTEGER NOT NULL DEFAULT 0, + token_input_total BIGINT NOT NULL DEFAULT 0, + token_output_total BIGINT NOT NULL DEFAULT 0, + created_at_ms BIGINT NOT NULL, + updated_at_ms BIGINT NOT NULL, + first_message_at_ms BIGINT, + last_message_at_ms BIGINT, + embedding vector(768), + meta JSONB NOT NULL DEFAULT '{}', + receipt TEXT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE TABLE IF NOT EXISTS ene.chat_messages ( - id BIGSERIAL PRIMARY KEY, - session_id TEXT NOT NULL REFERENCES ene.chat_sessions(session_id) ON DELETE CASCADE, - message_index INTEGER NOT NULL, - role TEXT NOT NULL, - blocks JSONB NOT NULL, - text_content TEXT, - token_input BIGINT NOT NULL DEFAULT 0, - token_output BIGINT NOT NULL DEFAULT 0, - token_cache_creation BIGINT NOT NULL DEFAULT 0, - token_cache_read BIGINT NOT NULL DEFAULT 0, - tool_calls JSONB NOT NULL DEFAULT '[]', - embedding vector(768), - receipt_hash TEXT, - created_at_ms BIGINT NOT NULL, - created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + id BIGSERIAL PRIMARY KEY, + session_id TEXT NOT NULL REFERENCES ene.chat_sessions(session_id) ON DELETE CASCADE, + message_index INTEGER NOT NULL, + role TEXT NOT NULL, + blocks JSONB NOT NULL, + text_content TEXT, + token_input BIGINT NOT NULL DEFAULT 0, + token_output BIGINT NOT NULL DEFAULT 0, + token_cache_creation BIGINT NOT NULL DEFAULT 0, + token_cache_read BIGINT NOT NULL DEFAULT 0, + tool_calls JSONB NOT NULL DEFAULT '[]', + embedding vector(768), + receipt_hash TEXT, + created_at_ms BIGINT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), UNIQUE(session_id, message_index) ); CREATE TABLE IF NOT EXISTS ene.ingestion_receipts ( - receipt_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), - shim_name TEXT NOT NULL, - status TEXT NOT NULL DEFAULT 'pending', - sha256 TEXT NOT NULL, - record_count BIGINT NOT NULL DEFAULT 0, - source_path TEXT NOT NULL, - meta JSONB NOT NULL DEFAULT '{}', - created_at TIMESTAMPTZ NOT NULL DEFAULT now() + receipt_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + shim_name TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + sha256 TEXT NOT NULL, + record_count BIGINT NOT NULL DEFAULT 0, + source_path TEXT NOT NULL, + meta JSONB NOT NULL DEFAULT '{}', + created_at TIMESTAMPTZ NOT NULL DEFAULT now() ); +-- Columns may already exist after a schema migration; ADD COLUMN IF NOT EXISTS +-- is idempotent on PostgreSQL ≥ 9.6. +ALTER TABLE ene.chat_sessions ADD COLUMN IF NOT EXISTS title TEXT; +ALTER TABLE ene.chat_sessions ADD COLUMN IF NOT EXISTS agent TEXT; +ALTER TABLE ene.chat_sessions ADD COLUMN IF NOT EXISTS model TEXT; + CREATE INDEX IF NOT EXISTS idx_chat_sessions_updated ON ene.chat_sessions(updated_at_ms DESC); +CREATE INDEX IF NOT EXISTS idx_chat_sessions_workspace + ON ene.chat_sessions(workspace_fingerprint); CREATE INDEX IF NOT EXISTS idx_chat_messages_session_order ON ene.chat_messages(session_id, message_index); CREATE INDEX IF NOT EXISTS idx_chat_messages_receipt ON ene.chat_messages(receipt_hash); + "#; + // Full-text and JSONB indexes require non-empty text_content / tool_calls; + // create them separately so a pgvector-missing cluster still gets the rest. + let fts_ddl = r#" CREATE INDEX IF NOT EXISTS idx_chat_messages_text_search - ON ene.chat_messages USING GIN(to_tsvector('english', text_content)); + ON ene.chat_messages USING GIN(to_tsvector('english', COALESCE(text_content, ''))); CREATE INDEX IF NOT EXISTS idx_chat_messages_tool_search ON ene.chat_messages USING GIN(tool_calls jsonb_path_ops); "#; self.client.batch_execute(ddl).await.context("init DDL")?; + if let Err(e) = self.client.batch_execute(fts_ddl).await { + warn!("FTS index creation skipped (non-fatal): {}", e); + } info!("RDS schema initialized"); Ok(()) } /// Upsert a session (insert or update on conflict). pub async fn upsert_session(&self, s: &ChatSession) -> Result<()> { - let embedding_str = s - .embedding - .as_ref() - .map(|v| format!("[{}]", v.iter().map(|f| f.to_string()).collect::>().join(","))); + let embedding_str = s.embedding.as_ref().map(|v| { + format!( + "[{}]", + v.iter() + .map(|f| f.to_string()) + .collect::>() + .join(",") + ) + }); let meta_json = serde_json::to_value(&s.meta)?; self.client .execute( "INSERT INTO ene.chat_sessions \ - (session_id, workspace_fingerprint, workspace_root, fork_parent_session_id, \ - compaction_count, compaction_summary, message_count, token_input_total, \ - token_output_total, created_at_ms, updated_at_ms, first_message_at_ms, \ - last_message_at_ms, embedding, meta, receipt, updated_at) \ - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14::vector, $15, $16, now()) \ + (session_id, title, agent, model, \ + workspace_fingerprint, workspace_root, fork_parent_session_id, \ + compaction_count, compaction_summary, message_count, \ + token_input_total, token_output_total, \ + created_at_ms, updated_at_ms, first_message_at_ms, last_message_at_ms, \ + embedding, meta, receipt, updated_at) \ + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17::vector,$18,$19,now()) \ ON CONFLICT (session_id) DO UPDATE SET \ - workspace_fingerprint = EXCLUDED.workspace_fingerprint, \ - workspace_root = EXCLUDED.workspace_root, \ + title = EXCLUDED.title, \ + agent = EXCLUDED.agent, \ + model = EXCLUDED.model, \ + workspace_fingerprint = EXCLUDED.workspace_fingerprint, \ + workspace_root = EXCLUDED.workspace_root, \ fork_parent_session_id = EXCLUDED.fork_parent_session_id, \ - compaction_count = EXCLUDED.compaction_count, \ - compaction_summary = EXCLUDED.compaction_summary, \ - message_count = EXCLUDED.message_count, \ - token_input_total = EXCLUDED.token_input_total, \ - token_output_total = EXCLUDED.token_output_total, \ - updated_at_ms = EXCLUDED.updated_at_ms, \ - first_message_at_ms = EXCLUDED.first_message_at_ms, \ - last_message_at_ms = EXCLUDED.last_message_at_ms, \ - embedding = EXCLUDED.embedding, \ - meta = EXCLUDED.meta, \ - receipt = EXCLUDED.receipt, \ - updated_at = now()", + compaction_count = EXCLUDED.compaction_count, \ + compaction_summary = EXCLUDED.compaction_summary, \ + message_count = EXCLUDED.message_count, \ + token_input_total = EXCLUDED.token_input_total, \ + token_output_total = EXCLUDED.token_output_total, \ + updated_at_ms = EXCLUDED.updated_at_ms, \ + first_message_at_ms = EXCLUDED.first_message_at_ms, \ + last_message_at_ms = EXCLUDED.last_message_at_ms, \ + embedding = EXCLUDED.embedding, \ + meta = EXCLUDED.meta, \ + receipt = EXCLUDED.receipt, \ + updated_at = now()", &[ &s.session_id, + &s.title, + &s.agent, + &s.model, &s.workspace_fingerprint, &s.workspace_root, &s.fork_parent_session_id, @@ -150,11 +216,16 @@ CREATE INDEX IF NOT EXISTS idx_chat_messages_tool_search /// Upsert a batch of messages for a session. pub async fn upsert_messages(&self, session_id: &str, msgs: &[ChatMessage]) -> Result<()> { - for (idx, msg) in msgs.iter().enumerate() { - let embedding_str = msg - .embedding - .as_ref() - .map(|v| format!("[{}]", v.iter().map(|f| f.to_string()).collect::>().join(","))); + for msg in msgs { + let embedding_str = msg.embedding.as_ref().map(|v| { + format!( + "[{}]", + v.iter() + .map(|f| f.to_string()) + .collect::>() + .join(",") + ) + }); let blocks_json = serde_json::to_value(&msg.blocks)?; let tool_calls_json = serde_json::to_value(&msg.tool_calls)?; self.client @@ -163,22 +234,22 @@ CREATE INDEX IF NOT EXISTS idx_chat_messages_tool_search (session_id, message_index, role, blocks, text_content, \ token_input, token_output, token_cache_creation, token_cache_read, \ tool_calls, embedding, receipt_hash, created_at_ms, created_at) \ - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11::vector, $12, $13, now()) \ + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11::vector,$12,$13,now()) \ ON CONFLICT (session_id, message_index) DO UPDATE SET \ - role = EXCLUDED.role, \ - blocks = EXCLUDED.blocks, \ - text_content = EXCLUDED.text_content, \ - token_input = EXCLUDED.token_input, \ - token_output = EXCLUDED.token_output, \ + role = EXCLUDED.role, \ + blocks = EXCLUDED.blocks, \ + text_content = EXCLUDED.text_content, \ + token_input = EXCLUDED.token_input, \ + token_output = EXCLUDED.token_output, \ token_cache_creation = EXCLUDED.token_cache_creation, \ - token_cache_read = EXCLUDED.token_cache_read, \ - tool_calls = EXCLUDED.tool_calls, \ - embedding = EXCLUDED.embedding, \ - receipt_hash = EXCLUDED.receipt_hash, \ - created_at_ms = EXCLUDED.created_at_ms", + token_cache_read = EXCLUDED.token_cache_read, \ + tool_calls = EXCLUDED.tool_calls, \ + embedding = EXCLUDED.embedding, \ + receipt_hash = EXCLUDED.receipt_hash, \ + created_at_ms = EXCLUDED.created_at_ms", &[ &session_id, - &(idx as i32), + &msg.message_index, &msg.role, &blocks_json, &msg.text_content, @@ -193,9 +264,18 @@ CREATE INDEX IF NOT EXISTS idx_chat_messages_tool_search ], ) .await - .with_context(|| format!("upsert message {} for session {}", idx, session_id))?; + .with_context(|| { + format!( + "upsert message idx={} for session {}", + msg.message_index, session_id + ) + })?; } - info!("upserted {} messages for session {}", msgs.len(), session_id); + info!( + "upserted {} messages for session {}", + msgs.len(), + session_id + ); Ok(()) } @@ -234,7 +314,7 @@ CREATE INDEX IF NOT EXISTS idx_chat_messages_tool_search Ok(()) } - /// Search sessions by keyword (full-text on text_content via messages). + /// Keyword full-text search across messages. pub async fn search_keyword( &self, query: &str, @@ -245,10 +325,12 @@ CREATE INDEX IF NOT EXISTS idx_chat_messages_tool_search .query( "SELECT s.session_id, s.title, s.agent, s.model, \ COUNT(m.id) AS match_count, \ - MAX(ts_rank(to_tsvector('english', m.text_content), plainto_tsquery('english', $1))) AS rank \ + MAX(ts_rank(to_tsvector('english', COALESCE(m.text_content,'')), \ + plainto_tsquery('english', $1))) AS rank \ FROM ene.chat_sessions s \ JOIN ene.chat_messages m ON m.session_id = s.session_id \ - WHERE to_tsvector('english', m.text_content) @@ plainto_tsquery('english', $1) \ + WHERE to_tsvector('english', COALESCE(m.text_content,'')) \ + @@ plainto_tsquery('english', $1) \ GROUP BY s.session_id, s.title, s.agent, s.model \ ORDER BY rank DESC \ LIMIT $2", @@ -257,20 +339,20 @@ CREATE INDEX IF NOT EXISTS idx_chat_messages_tool_search .await .context("keyword search")?; let mut out = Vec::new(); - for row in rows { + for row in &rows { out.push(serde_json::json!({ "session_id": row.try_get::<_, String>(0)?, - "title": row.try_get::<_, String>(1)?, - "agent": row.try_get::<_, Option>(2)?, - "model": row.try_get::<_, Option>(3)?, + "title": row.try_get::<_, Option>(1)?, + "agent": row.try_get::<_, Option>(2)?, + "model": row.try_get::<_, Option>(3)?, "match_count": row.try_get::<_, i64>(4)?, - "rank": row.try_get::<_, f32>(5)?, + "rank": row.try_get::<_, f32>(5)?, })); } Ok(out) } - /// Semantic search via embedding cosine similarity. + /// Semantic search via embedding cosine similarity (requires pgvector). pub async fn search_similar( &self, embedding: &[f32], @@ -278,7 +360,11 @@ CREATE INDEX IF NOT EXISTS idx_chat_messages_tool_search ) -> Result> { let vec_str = format!( "[{}]", - embedding.iter().map(|f| f.to_string()).collect::>().join(",") + embedding + .iter() + .map(|f| f.to_string()) + .collect::>() + .join(",") ); let rows = self .client @@ -294,12 +380,12 @@ CREATE INDEX IF NOT EXISTS idx_chat_messages_tool_search .await .context("similarity search")?; let mut out = Vec::new(); - for row in rows { + for row in &rows { out.push(serde_json::json!({ "session_id": row.try_get::<_, String>(0)?, - "title": row.try_get::<_, String>(1)?, - "agent": row.try_get::<_, Option>(2)?, - "model": row.try_get::<_, Option>(3)?, + "title": row.try_get::<_, Option>(1)?, + "agent": row.try_get::<_, Option>(2)?, + "model": row.try_get::<_, Option>(3)?, "similarity": row.try_get::<_, f32>(4)?, })); } @@ -321,25 +407,25 @@ CREATE INDEX IF NOT EXISTS idx_chat_messages_tool_search .await .context("list sessions")?; let mut out = Vec::new(); - for row in rows { + for row in &rows { out.push(serde_json::json!({ - "session_id": row.try_get::<_, String>(0)?, - "title": row.try_get::<_, String>(1)?, - "agent": row.try_get::<_, Option>(2)?, - "model": row.try_get::<_, Option>(3)?, - "message_count": row.try_get::<_, i32>(4)?, + "session_id": row.try_get::<_, String>(0)?, + "title": row.try_get::<_, Option>(1)?, + "agent": row.try_get::<_, Option>(2)?, + "model": row.try_get::<_, Option>(3)?, + "message_count": row.try_get::<_, i32>(4)?, "token_input_total": row.try_get::<_, i64>(5)?, - "token_output_total": row.try_get::<_, i64>(6)?, - "created_at_ms": row.try_get::<_, i64>(7)?, - "updated_at_ms": row.try_get::<_, i64>(8)?, + "token_output_total":row.try_get::<_, i64>(6)?, + "created_at_ms": row.try_get::<_, i64>(7)?, + "updated_at_ms": row.try_get::<_, i64>(8)?, })); } Ok(out) } - /// Get a single session with all its messages. + /// Retrieve a single session with all its messages. pub async fn get_session(&self, session_id: &str) -> Result> { - let session_row = self + let Some(sess) = self .client .query_opt( "SELECT session_id, title, agent, model, message_count, \ @@ -348,42 +434,46 @@ CREATE INDEX IF NOT EXISTS idx_chat_messages_tool_search &[&session_id], ) .await - .context("get session")?; - let Some(sess) = session_row else { return Ok(None) }; + .context("get session")? + else { + return Ok(None); + }; + let msg_rows = self .client .query( - "SELECT message_index, role, blocks, text_content, token_input, \ - token_output, tool_calls, created_at_ms \ - FROM ene.chat_messages WHERE session_id = $1 ORDER BY message_index", + "SELECT message_index, role, blocks, text_content, \ + token_input, token_output, tool_calls, created_at_ms \ + FROM ene.chat_messages \ + WHERE session_id = $1 ORDER BY message_index", &[&session_id], ) .await .context("get messages")?; let mut messages = Vec::new(); - for row in msg_rows { + for row in &msg_rows { messages.push(serde_json::json!({ "message_index": row.try_get::<_, i32>(0)?, - "role": row.try_get::<_, String>(1)?, - "blocks": row.try_get::<_, serde_json::Value>(2)?, - "text_content": row.try_get::<_, String>(3)?, - "token_input": row.try_get::<_, i64>(4)?, - "token_output": row.try_get::<_, i64>(5)?, - "tool_calls": row.try_get::<_, serde_json::Value>(6)?, + "role": row.try_get::<_, String>(1)?, + "blocks": row.try_get::<_, serde_json::Value>(2)?, + "text_content": row.try_get::<_, Option>(3)?, + "token_input": row.try_get::<_, i64>(4)?, + "token_output": row.try_get::<_, i64>(5)?, + "tool_calls": row.try_get::<_, serde_json::Value>(6)?, "created_at_ms": row.try_get::<_, i64>(7)?, })); } Ok(Some(serde_json::json!({ - "session_id": sess.try_get::<_, String>(0)?, - "title": sess.try_get::<_, String>(1)?, - "agent": sess.try_get::<_, Option>(2)?, - "model": sess.try_get::<_, Option>(3)?, - "message_count": sess.try_get::<_, i32>(4)?, - "token_input_total": sess.try_get::<_, i64>(5)?, + "session_id": sess.try_get::<_, String>(0)?, + "title": sess.try_get::<_, Option>(1)?, + "agent": sess.try_get::<_, Option>(2)?, + "model": sess.try_get::<_, Option>(3)?, + "message_count": sess.try_get::<_, i32>(4)?, + "token_input_total": sess.try_get::<_, i64>(5)?, "token_output_total": sess.try_get::<_, i64>(6)?, - "created_at_ms": sess.try_get::<_, i64>(7)?, - "updated_at_ms": sess.try_get::<_, i64>(8)?, - "meta": sess.try_get::<_, serde_json::Value>(9)?, + "created_at_ms": sess.try_get::<_, i64>(7)?, + "updated_at_ms": sess.try_get::<_, i64>(8)?, + "meta": sess.try_get::<_, serde_json::Value>(9)?, "messages": messages, }))) } diff --git a/4-Infrastructure/infra/ene-session-sync/src/source.rs b/4-Infrastructure/infra/ene-session-sync/src/source.rs index 353a7360..26e70380 100644 --- a/4-Infrastructure/infra/ene-session-sync/src/source.rs +++ b/4-Infrastructure/infra/ene-session-sync/src/source.rs @@ -1,5 +1,6 @@ use crate::models::{ - OpenCodeMessage, OpenCodePart, OpenCodeSession, OpenCodeSessionMessage, PartData, + ChatMessage, ChatSession, MessageBlock, OpenCodeMessage, OpenCodePart, OpenCodeSession, + OpenCodeSessionMessage, PartData, ToolCall, }; use anyhow::{Context, Result}; use rusqlite::{Connection, OptionalExtension}; @@ -233,3 +234,396 @@ impl OpenCodeSource { Ok(out) } } + +// --------------------------------------------------------------------------- +// Claw JSONL source +// --------------------------------------------------------------------------- + +/// Source adapter for Claw JSONL session files (`.claw/sessions/*.jsonl`). +/// +/// Each file is either: +/// - A single-line JSON object `{"version":1,"session_id":"...","messages":[...]}`, or +/// - A newline-delimited JSONL stream where each line is a message record. +/// +/// This adapter reads the files, parses whatever format it finds, and converts +/// them into the same [`ChatSession`] + [`ChatMessage`] types that the RDS sink +/// expects. This makes the sync pipeline source-agnostic. +pub struct ClawSource { + sessions_dir: std::path::PathBuf, +} + +impl ClawSource { + /// Create a source pointed at a `.claw/sessions/` directory. + pub fn new(sessions_dir: impl AsRef) -> Self { + Self { + sessions_dir: sessions_dir.as_ref().to_path_buf(), + } + } + + /// Discover all `.jsonl` files under the sessions directory. + fn find_files(&self) -> anyhow::Result> { + if !self.sessions_dir.exists() { + info!( + "Claw sessions dir not found, skipping: {:?}", + self.sessions_dir + ); + return Ok(Vec::new()); + } + let mut files = Vec::new(); + for entry in std::fs::read_dir(&self.sessions_dir) + .with_context(|| format!("read dir {:?}", self.sessions_dir))? + { + let entry = entry?; + let path = entry.path(); + if path.extension().and_then(|e| e.to_str()) == Some("jsonl") { + files.push(path); + } + } + files.sort(); + Ok(files) + } + + /// Parse a single JSONL session file into a `(ChatSession, Vec)`. + /// + /// The file is either a single full-session JSON object, or a stream of + /// per-message JSONL lines. Both formats are tried. + pub fn load_file( + path: &std::path::Path, + ) -> anyhow::Result)>> { + let raw = std::fs::read_to_string(path) + .with_context(|| format!("read {:?}", path))?; + + // Skip Git LFS stub files. + if raw.starts_with("version https://git-lfs.github.com") { + debug!("skipping LFS stub: {:?}", path); + return Ok(None); + } + + // Derive a session ID from the file name: session--.jsonl + let stem = path + .file_stem() + .and_then(|s| s.to_str()) + .unwrap_or("unknown"); + + // Try full-JSON format first. + if let Ok(obj) = serde_json::from_str::(raw.trim()) { + if obj.is_object() { + return Self::from_full_json(&obj, stem).map(Some); + } + } + + // Fall back to JSONL stream. + Self::from_jsonl_stream(&raw, stem).map(Some) + } + + fn from_full_json( + obj: &serde_json::Value, + stem: &str, + ) -> anyhow::Result<(ChatSession, Vec)> { + let session_id = obj + .get("session_id") + .and_then(|v| v.as_str()) + .unwrap_or(stem) + .to_string(); + let created_at_ms = obj + .get("created_at_ms") + .and_then(|v| v.as_i64()) + .unwrap_or(0); + let updated_at_ms = obj + .get("updated_at_ms") + .and_then(|v| v.as_i64()) + .unwrap_or(created_at_ms); + let workspace_root = obj + .get("workspace_root") + .and_then(|v| v.as_str()) + .map(String::from); + let fork_parent = obj + .get("fork") + .and_then(|f| f.get("parent_session_id")) + .and_then(|v| v.as_str()) + .map(String::from); + let compaction_count = obj + .get("compaction") + .and_then(|c| c.get("count")) + .and_then(|v| v.as_i64()) + .unwrap_or(0) as i32; + let compaction_summary = obj + .get("compaction") + .and_then(|c| c.get("summary")) + .and_then(|v| v.as_str()) + .map(String::from); + + let messages_raw = obj + .get("messages") + .and_then(|v| v.as_array()) + .cloned() + .unwrap_or_default(); + + let mut chat_messages = Vec::new(); + let mut first_at: Option = None; + let mut last_at: Option = None; + + for (idx, msg) in messages_raw.iter().enumerate() { + let cm = Self::parse_message(msg, &session_id, idx as i32)?; + first_at = first_at.or(Some(cm.created_at_ms)); + last_at = Some(cm.created_at_ms); + chat_messages.push(cm); + } + + let fingerprint = workspace_root + .as_deref() + .map(crate::normalize::workspace_fingerprint); + let session = ChatSession { + session_id: session_id.clone(), + title: None, // Claw sessions have no title field + agent: Some("claw".into()), + model: None, + workspace_fingerprint: fingerprint, + workspace_root, + fork_parent_session_id: fork_parent, + compaction_count, + compaction_summary, + message_count: chat_messages.len() as i32, + token_input_total: 0, + token_output_total: 0, + created_at_ms, + updated_at_ms, + first_message_at_ms: first_at, + last_message_at_ms: last_at, + meta: serde_json::json!({"source": "claw"}), + embedding: None, + receipt: None, + }; + Ok((session, chat_messages)) + } + + fn from_jsonl_stream( + raw: &str, + stem: &str, + ) -> anyhow::Result<(ChatSession, Vec)> { + // A JSONL stream has one JSON object per line; the first line may be a + // session header or the first message. + let mut session_id = stem.to_string(); + let mut created_at_ms: i64 = 0; + let mut workspace_root: Option = None; + let mut chat_messages = Vec::new(); + let mut first_at: Option = None; + + for (line_idx, line) in raw.lines().enumerate() { + let trimmed = line.trim(); + if trimmed.is_empty() { + continue; + } + let Ok(obj) = serde_json::from_str::(trimmed) else { + warn!("non-JSON line {} in {}", line_idx + 1, stem); + continue; + }; + + // If the first object has a "session_id" key treat it as a header. + if line_idx == 0 { + if let Some(id) = obj.get("session_id").and_then(|v| v.as_str()) { + session_id = id.to_string(); + created_at_ms = obj + .get("created_at_ms") + .and_then(|v| v.as_i64()) + .unwrap_or(0); + workspace_root = obj + .get("workspace_root") + .and_then(|v| v.as_str()) + .map(String::from); + continue; + } + } + + // Otherwise treat each line as a message. + let cm = + Self::parse_message(&obj, &session_id, chat_messages.len() as i32)?; + first_at = first_at.or(Some(cm.created_at_ms)); + chat_messages.push(cm); + } + + let last_at = chat_messages.last().map(|m| m.created_at_ms); + let fingerprint = workspace_root + .as_deref() + .map(crate::normalize::workspace_fingerprint); + let session = ChatSession { + session_id: session_id.clone(), + title: None, + agent: Some("claw".into()), + model: None, + workspace_fingerprint: fingerprint, + workspace_root, + fork_parent_session_id: None, + compaction_count: 0, + compaction_summary: None, + message_count: chat_messages.len() as i32, + token_input_total: 0, + token_output_total: 0, + created_at_ms, + updated_at_ms: last_at.unwrap_or(created_at_ms), + first_message_at_ms: first_at, + last_message_at_ms: last_at, + meta: serde_json::json!({"source": "claw"}), + embedding: None, + receipt: None, + }; + Ok((session, chat_messages)) + } + + /// Parse a single message object (from either format) into a [`ChatMessage`]. + fn parse_message( + msg: &serde_json::Value, + session_id: &str, + index: i32, + ) -> anyhow::Result { + let role = msg + .get("role") + .and_then(|v| v.as_str()) + .unwrap_or("unknown") + .to_string(); + let created_at_ms = msg + .get("timestamp_ms") + .or_else(|| msg.get("created_at_ms")) + .and_then(|v| v.as_i64()) + .unwrap_or(0); + let usage = msg.get("usage"); + let token_input = usage + .and_then(|u| u.get("input_tokens")) + .and_then(|v| v.as_i64()) + .unwrap_or(0); + let token_output = usage + .and_then(|u| u.get("output_tokens")) + .and_then(|v| v.as_i64()) + .unwrap_or(0); + let cache_creation = usage + .and_then(|u| u.get("cache_creation_input_tokens")) + .and_then(|v| v.as_i64()) + .unwrap_or(0); + let cache_read = usage + .and_then(|u| u.get("cache_read_input_tokens")) + .and_then(|v| v.as_i64()) + .unwrap_or(0); + + let mut blocks: Vec = Vec::new(); + let mut text_parts: Vec = Vec::new(); + let mut tool_calls: Vec = Vec::new(); + + // Blocks are either in msg["blocks"] or msg["content"]. + let blocks_raw = msg + .get("blocks") + .or_else(|| msg.get("content")) + .and_then(|v| v.as_array()) + .cloned() + .unwrap_or_default(); + + for block in &blocks_raw { + let btype = block + .get("type") + .and_then(|v| v.as_str()) + .unwrap_or("unknown"); + match btype { + "text" => { + let text = block + .get("text") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + text_parts.push(text.clone()); + blocks.push(MessageBlock { + block_type: "text".into(), + text: Some(text), + tool_name: None, + tool_input: None, + tool_output: None, + is_error: None, + }); + } + "tool_use" => { + let tool_name = block + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let call_id = block + .get("id") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let input = block.get("input").cloned().unwrap_or(serde_json::json!({})); + blocks.push(MessageBlock { + block_type: "tool_use".into(), + text: None, + tool_name: Some(tool_name.clone()), + tool_input: Some(input.clone()), + tool_output: None, + is_error: None, + }); + tool_calls.push(ToolCall { + call_id, + tool_name, + input, + }); + } + "tool_result" => { + let content = block.get("content").cloned(); + let text = content + .as_ref() + .and_then(|v| v.as_str()) + .map(String::from); + let is_error = block + .get("is_error") + .and_then(|v| v.as_bool()); + blocks.push(MessageBlock { + block_type: "tool_result".into(), + text, + tool_name: None, + tool_input: None, + tool_output: content, + is_error, + }); + } + other => { + blocks.push(MessageBlock { + block_type: other.to_string(), + text: block.get("text").and_then(|v| v.as_str()).map(String::from), + tool_name: None, + tool_input: None, + tool_output: None, + is_error: None, + }); + } + } + } + + Ok(ChatMessage { + session_id: session_id.to_string(), + message_index: index, + role, + blocks, + text_content: text_parts.join("\n"), + token_input, + token_output, + token_cache_creation: cache_creation, + token_cache_read: cache_read, + tool_calls, + embedding: None, + receipt_hash: None, + created_at_ms, + }) + } + + /// Walk the sessions directory and return all valid (session, messages) pairs. + pub fn load_all(&self) -> anyhow::Result)>> { + let files = self.find_files()?; + let mut out = Vec::new(); + for path in &files { + match Self::load_file(path) { + Ok(Some(pair)) => out.push(pair), + Ok(None) => {} // LFS stub or empty, skipped + Err(e) => warn!("error loading Claw session {:?}: {}", path.display(), e), + } + } + info!("loaded {} Claw sessions from {:?}", out.len(), self.sessions_dir); + Ok(out) + } +} diff --git a/4-Infrastructure/infra/ene-session-sync/systemd/ene-session-sync.service b/4-Infrastructure/infra/ene-session-sync/systemd/ene-session-sync.service new file mode 100644 index 00000000..8da3056c --- /dev/null +++ b/4-Infrastructure/infra/ene-session-sync/systemd/ene-session-sync.service @@ -0,0 +1,37 @@ +[Unit] +Description=ENE Session Sync — OpenCode→RDS chat log ingestion daemon +Documentation=file:///home/researcher/stack/4-Infrastructure/infra/ene-session-sync/README.md +After=network-online.target +Wants=network-online.target + +[Service] +Type=oneshot +# Run as the developer user so ~/.local/share/opencode/opencode.db is reachable. +User=researcher +Group=researcher + +# Binary is installed to ~/.cargo/bin after `cargo install --path .` or +# symlinked from target/release/. +ExecStart=/home/researcher/.cargo/bin/ene-session-sync \ + --db /home/researcher/.local/share/opencode/opencode.db \ + sync + +# Embedding is opt-in; set OLLAMA_HOST + OLLAMA_EMBED_MODEL env vars and +# add --embed to ExecStart if Ollama is available. + +EnvironmentFile=-/etc/ene-session-sync/env +EnvironmentFile=-/home/researcher/.config/ene-session-sync/env + +# Credentials (RDS_HOST, RDS_PORT, RDS_USER, RDS_PASSWORD or RDS_IAM_TOKEN) +# live in the EnvironmentFile above. Never hard-code them here. + +StandardOutput=journal +StandardError=journal +SyslogIdentifier=ene-session-sync + +# Restart on failure with backoff — the DB may be temporarily locked. +Restart=on-failure +RestartSec=30 + +[Install] +WantedBy=multi-user.target diff --git a/4-Infrastructure/infra/ene-session-sync/systemd/ene-session-sync.timer b/4-Infrastructure/infra/ene-session-sync/systemd/ene-session-sync.timer new file mode 100644 index 00000000..0d079bff --- /dev/null +++ b/4-Infrastructure/infra/ene-session-sync/systemd/ene-session-sync.timer @@ -0,0 +1,15 @@ +[Unit] +Description=ENE Session Sync — fire every 5 minutes +Requires=ene-session-sync.service + +[Timer] +# Run 2 minutes after boot, then every 5 minutes. +OnBootSec=2min +OnUnitActiveSec=5min +# Spread the load — avoids a thundering herd when multiple timers wake up. +RandomizedDelaySec=30 +# Catch up missed runs (e.g. after suspend). +Persistent=true + +[Install] +WantedBy=timers.target