kyleandClaude Opus 5.5
de1032157b
memoryd (files, the check, edits, journal, index, recall, serving), the memory tools and built-in grants in brokerd, the memory block in loopd, bxctl reindex and memory, and the device checks with memoryd. Each task was built as a reference and gated. Each was replayed on its own at its end state (854 to 936 tests) and mutation-checked. Each skeleton compiles and fails its tests. Every todo comment was read beside its signature. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
195 lines
8.9 KiB
Rust
195 lines
8.9 KiB
Rust
//! The index: SQLite FTS5 over the session logs and `notes.md` (M5a spec, section 6). Disposable:
|
|
//! it is rebuilt from those files. What is indexed: the owner's messages, the assistant's
|
|
//! replies, and each tool call by name and arguments; never a tool's result.
|
|
|
|
use std::os::unix::fs::PermissionsExt;
|
|
use std::path::{Path, PathBuf};
|
|
|
|
use proto::{DataClass, LogRecord, Timestamp};
|
|
use rusqlite::{Connection, params};
|
|
|
|
use crate::notes::{self, Kind};
|
|
|
|
const SCHEMA: &str = "
|
|
CREATE VIRTUAL TABLE IF NOT EXISTS hits USING fts5(text, tokenize = 'unicode61');
|
|
CREATE TABLE IF NOT EXISTS sources (
|
|
rowid INTEGER PRIMARY KEY, kind TEXT NOT NULL, session TEXT NOT NULL, epoch INTEGER NOT NULL,
|
|
record INTEGER NOT NULL, time TEXT NOT NULL, role TEXT NOT NULL, class TEXT NOT NULL,
|
|
untrusted INTEGER NOT NULL);
|
|
CREATE TABLE IF NOT EXISTS progress (log TEXT PRIMARY KEY, records INTEGER NOT NULL);
|
|
";
|
|
|
|
/// One indexed item.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct Source {
|
|
/// `session`, `noted`, `moved` or `dropped`.
|
|
pub kind: String,
|
|
pub session: String,
|
|
pub epoch: i64,
|
|
/// The record's line in its log, or the note's place in `notes.md` (from 0).
|
|
pub record: i64,
|
|
pub time: String,
|
|
/// `owner`, `assistant` or `call` for a session record; the file (`memory.md`, `user.md`) for
|
|
/// a moved or dropped note; empty for a noted one.
|
|
pub role: String,
|
|
pub class: DataClass,
|
|
pub untrusted: bool,
|
|
}
|
|
|
|
/// How much a reindex read.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
|
|
pub struct Counts {
|
|
pub logs: u64,
|
|
pub records: u64,
|
|
pub notes: u64,
|
|
}
|
|
|
|
pub struct Index {
|
|
conn: Connection,
|
|
max_class: DataClass,
|
|
}
|
|
|
|
fn db(e: rusqlite::Error) -> String {
|
|
format!("index: {e}")
|
|
}
|
|
|
|
pub fn class_name(class: DataClass) -> &'static str {
|
|
match class {
|
|
DataClass::Public => "public",
|
|
DataClass::Private => "private",
|
|
DataClass::Secret => "secret",
|
|
}
|
|
}
|
|
|
|
pub fn class_of(name: &str) -> DataClass {
|
|
match name {
|
|
"public" => DataClass::Public,
|
|
"secret" => DataClass::Secret,
|
|
_ => DataClass::Private,
|
|
}
|
|
}
|
|
|
|
fn int(n: usize) -> i64 {
|
|
i64::try_from(n).unwrap_or(i64::MAX)
|
|
}
|
|
|
|
impl Index {
|
|
/// Open or create the index at `path` (its directory 0700, the file 0600).
|
|
pub fn open(path: &Path, max_class: DataClass) -> Result<Index, String> {
|
|
// 1. When `path.parent()` is Some(dir): `create_dir_all(dir)`, then `set_permissions(dir,
|
|
// Permissions::from_mode(0o700))`.
|
|
// 2. `Connection::open(path)`, then `set_permissions(path, Permissions::from_mode(0o600))`,
|
|
// then `conn.execute_batch(SCHEMA)`. Every error is format!("{}: {e}", the path).
|
|
// Ok(Index { conn, max_class }).
|
|
todo!()
|
|
}
|
|
|
|
/// The connection, for `recall`.
|
|
pub fn conn(&self) -> &Connection {
|
|
&self.conn
|
|
}
|
|
|
|
fn progress(&self, log: &str) -> Result<usize, String> {
|
|
// `self.conn.query_row("SELECT records FROM progress WHERE log = ?1", params![log], |row|
|
|
// row.get(0))` as an Option<i64>: Err(rusqlite::Error::QueryReturnedNoRows) means None,
|
|
// another error goes through `db`. Return it as a usize (`usize::try_from`, 0 when it does
|
|
// not fit), 0 for None.
|
|
todo!()
|
|
}
|
|
|
|
fn add(&self, text: &str, source: &Source) -> Result<(), String> {
|
|
// 1. `execute("INSERT INTO hits (text) VALUES (?1)", params![text])`.
|
|
// 2. `let rowid = self.conn.last_insert_rowid();`
|
|
// 3. `execute("INSERT INTO sources (rowid, kind, session, epoch, record, time, role, class,
|
|
// untrusted) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", params![rowid, kind, session,
|
|
// epoch, record, time, role, class_name(class), untrusted])`. Errors through `db`.
|
|
todo!()
|
|
}
|
|
|
|
fn set_progress(&self, log: &str, records: usize) -> Result<(), String> {
|
|
// `execute("INSERT INTO progress (log, records) VALUES (?1, ?2) ON CONFLICT(log) DO UPDATE
|
|
// SET records = excluded.records", params![log, int(records)])`, errors through `db`.
|
|
todo!()
|
|
}
|
|
|
|
/// Index the records of one session log past what was indexed before; how many were read.
|
|
/// The label is walked from the log's first record: the highest class of any tool result
|
|
/// before a record (at least `private`), and whether any was untrusted.
|
|
pub fn index_log(&self, path: &Path, session: &str, epoch: i64) -> Result<usize, String> {
|
|
// 1. `read_to_string(path)` (an error names the path). `key` =
|
|
// format!("{session}/{epoch}"); `done = self.progress(&key)?`.
|
|
// 2. The label starts at class Private, untrusted false; `read` at 0.
|
|
// 3. For each line with its place n (from 0): parse it as a LogRecord (an error ->
|
|
// Err(format!("{}: line {}: {e}", path.display(), n + 1))). When n >= done: read += 1,
|
|
// and `self.index_record(&record, &base)?` where base is a Source with kind "session",
|
|
// session, epoch, record int(n), time and role empty, and the label as it stands. Then,
|
|
// for every record (indexed or not): a ToolResult raises the label: class =
|
|
// class.max(its class), untrusted |= its untrusted.
|
|
// 4. `self.set_progress(&key, done + read)?`; Ok(read).
|
|
todo!()
|
|
}
|
|
|
|
/// Index what one log record holds: the owner's message, the assistant's reply, and each tool
|
|
/// call by name and arguments. `base` carries the kind, session, epoch, line and label; each item
|
|
/// gets its own role and time. Nothing when the label's class is above `max_class`, and nothing
|
|
/// for an item with no text.
|
|
fn index_record(&self, record: &LogRecord, base: &Source) -> Result<(), String> {
|
|
// When base.class > self.max_class: Ok(()) and nothing indexed. Otherwise each item is
|
|
// `self.add(text, &Source { time: time.to_rfc3339(), role, ..base.clone() })`, skipped when
|
|
// text.trim() is empty. User -> its content, role "owner". Assistant -> its content when
|
|
// Some, role "assistant"; then each tool call as format!("{} {}", name, arguments), role
|
|
// "call"; all with the record's time. Every other record: nothing.
|
|
todo!()
|
|
}
|
|
|
|
/// Index the entries of `notes.md` past what was indexed before; how many were read.
|
|
pub fn index_notes(&self, text: &str) -> Result<usize, String> {
|
|
// `done = self.progress("notes.md")?`. Parse `text` with notes::parse and skip the first
|
|
// `done`. For each remaining note (with its place n): read += 1; its kind, role, class and
|
|
// untrusted: Noted -> ("noted", "", its class, its untrusted); Moved(file) -> ("moved",
|
|
// file.name(), Private, false); Dropped(file) -> ("dropped", file.name(), Private, false).
|
|
// Skip it when class > self.max_class. Otherwise `self.add(¬e.text, &Source { kind,
|
|
// session: note.session, epoch: 0, record: int(n), time: note.time, role, class, untrusted
|
|
// })?`. Then `self.set_progress("notes.md", done + read)?`; Ok(read).
|
|
todo!()
|
|
}
|
|
|
|
/// Index everything new: every `sessions/<id>/<epoch>.jsonl`, then `notes.md`. A log that
|
|
/// cannot be read or parsed is skipped and named in the returned problems; it is `loopd`'s.
|
|
pub fn catch_up(
|
|
&self,
|
|
sessions: &Path,
|
|
notes_md: &Path,
|
|
) -> Result<(Counts, Vec<String>), String> {
|
|
// For each (session, epoch, path) of `logs(sessions)?`: `self.index_log(&path, &session,
|
|
// epoch)`; Ok(n) -> counts.logs += 1 and counts.records += n (u64::try_from, u64::MAX when
|
|
// it does not fit); Err(e) -> push e onto `problems` and go on. Then read `notes_md`:
|
|
// NotFound is an empty text, another error -> Err(format!("{}: {e}", notes_md.display()));
|
|
// counts.notes = `self.index_notes(&text)?` as u64 (u64::try_from). Ok((counts, problems)).
|
|
todo!()
|
|
}
|
|
}
|
|
|
|
/// Every session log under `sessions`: `(session, epoch, path)`, sorted.
|
|
pub fn logs(sessions: &Path) -> Result<Vec<(String, i64, PathBuf)>, String> {
|
|
// Read the directory `sessions` (NotFound -> Ok(empty); another error -> Err naming it). For
|
|
// each entry, its name is the session; read that directory (skip it when that fails); every
|
|
// file named `<n>.jsonl` where n parses as an i64 gives (session, n, the file's path). Sort the
|
|
// list and return it.
|
|
todo!()
|
|
}
|
|
|
|
/// Build a fresh index at `path` from `sessions` and `notes.md`: into a new file beside it,
|
|
/// renamed over the old one, so a running `recall` keeps its index until the new one is whole.
|
|
pub fn rebuild(
|
|
path: &Path,
|
|
max_class: DataClass,
|
|
sessions: &Path,
|
|
notes_md: &Path,
|
|
) -> Result<(Counts, Vec<String>), String> {
|
|
// `fresh = path.with_extension("rebuilding")`; remove it if it is there (ignore the result).
|
|
// `Index::open(&fresh, max_class)?`, then `catch_up(sessions, notes_md)?` on it, and drop that
|
|
// Index. `std::fs::rename(&fresh, path)` (an error names path). Return the result of catch_up.
|
|
todo!()
|
|
}
|