mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
Rust ene-session-sync: complete rewrite with Python adaptation layer
- source.rs: Add ClawSource adapter that reads .claw/sessions/*.jsonl (skips
LFS stubs), parses both full-JSON and JSONL-stream formats, normalises into
ChatSession/ChatMessage identical to the OpenCode path.
- models.rs: Add title/agent/model fields to ChatSession; remove unused
chrono import.
- normalize.rs: Populate title/agent/model from OpenCodeSession.
- sink.rs: Full rewrite:
- DDL adds title/agent/model columns + ALTER TABLE IF NOT EXISTS migration
path for existing clusters.
- upsert_session and upsert_messages parameter counts match placeholders
($19 and $13 respectively).
- FTS index uses COALESCE to handle NULL text_content.
- sslmode= token stripped before parse so NoTls connect doesn't reject
it; README documents the SSL upgrade path (tokio-postgres-rustls).
- with-serde_json-1 feature added to tokio-postgres dep for JSONB params.
- main.rs:
- Replace DefaultHasher sha256_text with FNV-1a 64-bit (correct hex digest).
- Add ClawSync, List, Get subcommands.
- Remove unused Context import.
- bridge.rs: Rewrite to use bridge_wrapper.py subprocess rather than inline
-c script; resolves wrapper path relative to infra_dir or CARGO_MANIFEST_DIR;
unwraps {"ok":true,"data":{}} envelope; health_check() tests python3 only.
- embed.rs: Remove unused OllamaEmbedRequest import.
- systemd/: Add ene-session-sync.service + .timer (5-min cadence, 2-min
OnBootSec, Persistent=true for catch-up after suspend).
- README.md: Architecture diagram, CLI reference, schema, build/install,
systemd install, Python bridge setup.
Generated with [Devin](https://cli.devin.ai/docs)
Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Brandon Schneider
parent
4d715908fd
commit
2cc9fc8dc7
11 changed files with 1097 additions and 201 deletions
|
|
@ -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"] }
|
||||
|
||||
|
|
|
|||
155
4-Infrastructure/infra/ene-session-sync/README.md
Normal file
155
4-Infrastructure/infra/ene-session-sync/README.md
Normal file
|
|
@ -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 <infra_dir>
|
||||
├── 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] <SUBCOMMAND>
|
||||
|
||||
Global options:
|
||||
--db <PATH> opencode.db path (default: ~/.local/share/opencode/opencode.db)
|
||||
--dsn <DSN> PostgreSQL DSN (default: RDS_* env vars)
|
||||
--infra-dir <DIR> Python infra directory for bridge calls
|
||||
--embed Enable Ollama embedding generation
|
||||
|
||||
Subcommands:
|
||||
sync [--since <MS>] One-shot sync of opencode.db
|
||||
watch [--interval <SECS>] Continuous poll (default 60 s)
|
||||
claw-sync --sessions-dir <D> Sync .claw/sessions/ JSONL files
|
||||
search <QUERY> [--limit N] [--semantic]
|
||||
list [--limit N]
|
||||
get <SESSION_ID>
|
||||
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=<generated by aws rds generate-db-auth-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
|
||||
```
|
||||
|
|
@ -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 <infra_dir>`,
|
||||
/// 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": "<name>", "payload": { ... }}`
|
||||
/// Response (stdout): `{"ok": true, "data": { ... }}`
|
||||
/// or `{"ok": false, "error": "<message>"}`
|
||||
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<serde_json::Value> {
|
||||
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 <bridge_wrapper.py> <infra_dir>`, writes the request
|
||||
/// JSON to stdin, and parses the response JSON from stdout.
|
||||
pub fn call(&self, module: &str, payload: &serde_json::Value) -> Result<serde_json::Value> {
|
||||
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::<String>()
|
||||
);
|
||||
}
|
||||
|
||||
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(|| {
|
||||
|
||||
let value: serde_json::Value = serde_json::from_str(trimmed).with_context(|| {
|
||||
format!(
|
||||
"parse JSON from Python module {}: got {}",
|
||||
"parse bridge JSON for module `{}`: got {:?}",
|
||||
module,
|
||||
trimmed.chars().take(200).collect::<String>()
|
||||
trimmed.chars().take(300).collect::<String>()
|
||||
)
|
||||
})?;
|
||||
debug!("bridge {} -> ok", module);
|
||||
|
||||
// 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<BridgeResponse> {
|
||||
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<bool> {
|
||||
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
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,4 +1,3 @@
|
|||
use crate::models::OllamaEmbedRequest;
|
||||
use anyhow::{Context, Result};
|
||||
use reqwest::Client;
|
||||
use serde_json::json;
|
||||
|
|
|
|||
|
|
@ -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(())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<String>,
|
||||
pub agent: Option<String>,
|
||||
pub model: Option<String>,
|
||||
pub workspace_fingerprint: Option<String>,
|
||||
pub workspace_root: Option<String>,
|
||||
pub fork_parent_session_id: Option<String>,
|
||||
|
|
|
|||
|
|
@ -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(),
|
||||
|
|
|
|||
|
|
@ -10,24 +10,62 @@ pub struct RdsSink {
|
|||
|
||||
impl RdsSink {
|
||||
/// 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<Self> {
|
||||
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::<Vec<_>>()
|
||||
.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,
|
||||
title TEXT,
|
||||
agent TEXT,
|
||||
model TEXT,
|
||||
workspace_fingerprint TEXT,
|
||||
workspace_root TEXT,
|
||||
fork_parent_session_id TEXT,
|
||||
|
|
@ -76,38 +114,63 @@ CREATE TABLE IF NOT EXISTS ene.ingestion_receipts (
|
|||
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::<Vec<_>>().join(",")));
|
||||
let embedding_str = s.embedding.as_ref().map(|v| {
|
||||
format!(
|
||||
"[{}]",
|
||||
v.iter()
|
||||
.map(|f| f.to_string())
|
||||
.collect::<Vec<_>>()
|
||||
.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 \
|
||||
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, \
|
||||
|
|
@ -125,6 +188,9 @@ CREATE INDEX IF NOT EXISTS idx_chat_messages_tool_search
|
|||
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::<Vec<_>>().join(",")));
|
||||
for msg in msgs {
|
||||
let embedding_str = msg.embedding.as_ref().map(|v| {
|
||||
format!(
|
||||
"[{}]",
|
||||
v.iter()
|
||||
.map(|f| f.to_string())
|
||||
.collect::<Vec<_>>()
|
||||
.join(",")
|
||||
)
|
||||
});
|
||||
let blocks_json = serde_json::to_value(&msg.blocks)?;
|
||||
let tool_calls_json = serde_json::to_value(&msg.tool_calls)?;
|
||||
self.client
|
||||
|
|
@ -178,7 +249,7 @@ CREATE INDEX IF NOT EXISTS idx_chat_messages_tool_search
|
|||
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,10 +339,10 @@ 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)?,
|
||||
"title": row.try_get::<_, Option<String>>(1)?,
|
||||
"agent": row.try_get::<_, Option<String>>(2)?,
|
||||
"model": row.try_get::<_, Option<String>>(3)?,
|
||||
"match_count": row.try_get::<_, i64>(4)?,
|
||||
|
|
@ -270,7 +352,7 @@ CREATE INDEX IF NOT EXISTS idx_chat_messages_tool_search
|
|||
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<Vec<serde_json::Value>> {
|
||||
let vec_str = format!(
|
||||
"[{}]",
|
||||
embedding.iter().map(|f| f.to_string()).collect::<Vec<_>>().join(",")
|
||||
embedding
|
||||
.iter()
|
||||
.map(|f| f.to_string())
|
||||
.collect::<Vec<_>>()
|
||||
.join(",")
|
||||
);
|
||||
let rows = self
|
||||
.client
|
||||
|
|
@ -294,10 +380,10 @@ 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)?,
|
||||
"title": row.try_get::<_, Option<String>>(1)?,
|
||||
"agent": row.try_get::<_, Option<String>>(2)?,
|
||||
"model": row.try_get::<_, Option<String>>(3)?,
|
||||
"similarity": row.try_get::<_, f32>(4)?,
|
||||
|
|
@ -321,10 +407,10 @@ 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)?,
|
||||
"title": row.try_get::<_, Option<String>>(1)?,
|
||||
"agent": row.try_get::<_, Option<String>>(2)?,
|
||||
"model": row.try_get::<_, Option<String>>(3)?,
|
||||
"message_count": row.try_get::<_, i32>(4)?,
|
||||
|
|
@ -337,9 +423,9 @@ CREATE INDEX IF NOT EXISTS idx_chat_messages_tool_search
|
|||
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<Option<serde_json::Value>> {
|
||||
let session_row = self
|
||||
let Some(sess) = self
|
||||
.client
|
||||
.query_opt(
|
||||
"SELECT session_id, title, agent, model, message_count, \
|
||||
|
|
@ -348,25 +434,29 @@ 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)?,
|
||||
"text_content": row.try_get::<_, Option<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)?,
|
||||
|
|
@ -375,7 +465,7 @@ CREATE INDEX IF NOT EXISTS idx_chat_messages_tool_search
|
|||
}
|
||||
Ok(Some(serde_json::json!({
|
||||
"session_id": sess.try_get::<_, String>(0)?,
|
||||
"title": sess.try_get::<_, String>(1)?,
|
||||
"title": sess.try_get::<_, Option<String>>(1)?,
|
||||
"agent": sess.try_get::<_, Option<String>>(2)?,
|
||||
"model": sess.try_get::<_, Option<String>>(3)?,
|
||||
"message_count": sess.try_get::<_, i32>(4)?,
|
||||
|
|
|
|||
|
|
@ -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<std::path::Path>) -> 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<Vec<std::path::PathBuf>> {
|
||||
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<ChatMessage>)`.
|
||||
///
|
||||
/// 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<Option<(ChatSession, Vec<ChatMessage>)>> {
|
||||
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-<ts>-<n>.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::<serde_json::Value>(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<ChatMessage>)> {
|
||||
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<i64> = None;
|
||||
let mut last_at: Option<i64> = 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<ChatMessage>)> {
|
||||
// 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<String> = None;
|
||||
let mut chat_messages = Vec::new();
|
||||
let mut first_at: Option<i64> = 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::<serde_json::Value>(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<ChatMessage> {
|
||||
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<MessageBlock> = Vec::new();
|
||||
let mut text_parts: Vec<String> = Vec::new();
|
||||
let mut tool_calls: Vec<ToolCall> = 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<Vec<(ChatSession, Vec<ChatMessage>)>> {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
Loading…
Add table
Reference in a new issue