From 719f828ddc8025cb47e585173149e4d5fd57d8ef Mon Sep 17 00:00:00 2001 From: KooshaPari Date: Sun, 9 Aug 2026 17:01:26 -0700 Subject: [PATCH] fix(daemon): deduplicate durable watcher facts --- crates/sl-daemon/src/etl.rs | 17 ++++++++++ .../00_SESSION_OVERVIEW.md | 5 +++ .../01_RESEARCH.md | 8 +++++ .../02_SPECIFICATIONS.md | 5 +++ .../03_DAG_WBS.md | 5 +++ .../04_IMPLEMENTATION_STRATEGY.md | 3 ++ .../05_KNOWN_ISSUES.md | 5 +++ .../06_TESTING_STRATEGY.md | 5 +++ src/ports/sqlite_memory.rs | 34 ++++++++++++------- 9 files changed, 75 insertions(+), 12 deletions(-) create mode 100644 docs/sessions/20260809-watcher-fact-idempotency/00_SESSION_OVERVIEW.md create mode 100644 docs/sessions/20260809-watcher-fact-idempotency/01_RESEARCH.md create mode 100644 docs/sessions/20260809-watcher-fact-idempotency/02_SPECIFICATIONS.md create mode 100644 docs/sessions/20260809-watcher-fact-idempotency/03_DAG_WBS.md create mode 100644 docs/sessions/20260809-watcher-fact-idempotency/04_IMPLEMENTATION_STRATEGY.md create mode 100644 docs/sessions/20260809-watcher-fact-idempotency/05_KNOWN_ISSUES.md create mode 100644 docs/sessions/20260809-watcher-fact-idempotency/06_TESTING_STRATEGY.md diff --git a/crates/sl-daemon/src/etl.rs b/crates/sl-daemon/src/etl.rs index 2863b9fd..1f1b8f0d 100644 --- a/crates/sl-daemon/src/etl.rs +++ b/crates/sl-daemon/src/etl.rs @@ -291,4 +291,21 @@ mod tests { let recalled = store.recall("pagination", 5).expect("recall distilled facts"); assert!(!recalled.is_empty(), "distilled episodic facts should persist"); } + + #[cfg(feature = "sqlite")] + #[test] + fn transform_file_does_not_duplicate_durable_facts_for_repeated_input() { + use session_ledger::SqliteMemoryStore; + + let tmp = tempfile::tempdir().expect("tempdir"); + let jsonl = write_fixture(tmp.path(), 1); + let out = tmp.path().join("out"); + let store = SqliteMemoryStore::open(tmp.path().join("memory.db")).expect("open memory db"); + + transform_file(&jsonl, &out, Some(&store)).expect("first transform"); + transform_file(&jsonl, &out, Some(&store)).expect("repeated transform"); + + let facts = store.recall("session/sess-0", 10).expect("recall durable facts"); + assert_eq!(facts.len(), 3, "repeated watcher events must not duplicate durable facts"); + } } diff --git a/docs/sessions/20260809-watcher-fact-idempotency/00_SESSION_OVERVIEW.md b/docs/sessions/20260809-watcher-fact-idempotency/00_SESSION_OVERVIEW.md new file mode 100644 index 00000000..374add2d --- /dev/null +++ b/docs/sessions/20260809-watcher-fact-idempotency/00_SESSION_OVERVIEW.md @@ -0,0 +1,5 @@ +# Watcher fact idempotency + +Goal: prevent repeated filesystem-event handling of unchanged transcript input from duplicating SQLite episodic facts. + +Success: a focused ETL regression test fails before the fix, passes after it, and the daemon's relevant test suite remains green. diff --git a/docs/sessions/20260809-watcher-fact-idempotency/01_RESEARCH.md b/docs/sessions/20260809-watcher-fact-idempotency/01_RESEARCH.md new file mode 100644 index 00000000..4904539d --- /dev/null +++ b/docs/sessions/20260809-watcher-fact-idempotency/01_RESEARCH.md @@ -0,0 +1,8 @@ +# Research + +- `crates/sl-daemon/src/watcher.rs` forwards each create or modify event for a transcript; FSEvents may report repeated events for unchanged input. +- `crates/sl-daemon/src/etl.rs::transform_file` invokes `compile_and_store` once per parsed session when SQLite memory is configured. +- `src/ports/sqlite_memory.rs::store` currently assigns an incrementing identifier, so reprocessing the same session/key/content creates another durable row. +- `DistillMemoryWriter` already constructs a stable key from session id and bundle kind. A deterministic identity over session id, key, and content can make the SQLite primary key idempotent without changing watcher timing or HTTP behavior. + +The red regression test transformed one JSONL fixture twice and recalled six SQLite facts where the three stable distilled facts were expected. This confirms the duplicate originates at the durable SQLite identity boundary. diff --git a/docs/sessions/20260809-watcher-fact-idempotency/02_SPECIFICATIONS.md b/docs/sessions/20260809-watcher-fact-idempotency/02_SPECIFICATIONS.md new file mode 100644 index 00000000..e81fc191 --- /dev/null +++ b/docs/sessions/20260809-watcher-fact-idempotency/02_SPECIFICATIONS.md @@ -0,0 +1,5 @@ +# Specifications + +Repeated handling of one unchanged transcript must leave one durable SQLite fact per distilled fact identity. Distinct content must remain independently persistable. + +Scope excludes HTTP persistence, viewer changes, daemon configuration, and live daemon control. diff --git a/docs/sessions/20260809-watcher-fact-idempotency/03_DAG_WBS.md b/docs/sessions/20260809-watcher-fact-idempotency/03_DAG_WBS.md new file mode 100644 index 00000000..b1fcfbf1 --- /dev/null +++ b/docs/sessions/20260809-watcher-fact-idempotency/03_DAG_WBS.md @@ -0,0 +1,5 @@ +# Work breakdown + +1. Trace watcher, ETL, memory writer, and SQLite store. +2. Add a focused failing repeated-input regression test. +3. Make durable identity idempotent and validate targeted tests. diff --git a/docs/sessions/20260809-watcher-fact-idempotency/04_IMPLEMENTATION_STRATEGY.md b/docs/sessions/20260809-watcher-fact-idempotency/04_IMPLEMENTATION_STRATEGY.md new file mode 100644 index 00000000..5ab18277 --- /dev/null +++ b/docs/sessions/20260809-watcher-fact-idempotency/04_IMPLEMENTATION_STRATEGY.md @@ -0,0 +1,3 @@ +# Implementation strategy + +Use a deterministic content identity for SQLite memory facts and SQLite primary-key conflict handling. This fixes the durable boundary where duplicates are created and avoids debouncing heuristics that could discard legitimate later edits. diff --git a/docs/sessions/20260809-watcher-fact-idempotency/05_KNOWN_ISSUES.md b/docs/sessions/20260809-watcher-fact-idempotency/05_KNOWN_ISSUES.md new file mode 100644 index 00000000..c00ed55d --- /dev/null +++ b/docs/sessions/20260809-watcher-fact-idempotency/05_KNOWN_ISSUES.md @@ -0,0 +1,5 @@ +# Known issues + +OS filesystem event timing is intentionally not used in the test; the deterministic repeated ETL input reproduces its downstream effect. + +The current checkout's repository-wide `cargo fmt --check` reports pre-existing formatting drift in unrelated daemon resolver, HTTP, main, and lib files. The changed Rust files pass an isolated `rustfmt --check`. `cargo clippy` is blocked before linting by the existing `clippy.toml` fields `warn` and `allow`, which this installed Clippy rejects. diff --git a/docs/sessions/20260809-watcher-fact-idempotency/06_TESTING_STRATEGY.md b/docs/sessions/20260809-watcher-fact-idempotency/06_TESTING_STRATEGY.md new file mode 100644 index 00000000..b933c6f1 --- /dev/null +++ b/docs/sessions/20260809-watcher-fact-idempotency/06_TESTING_STRATEGY.md @@ -0,0 +1,5 @@ +# Testing strategy + +Run a focused `sl-daemon` ETL unit test that transforms the same real JSONL fixture twice through a SQLite memory store, then run the daemon crate test suite with the SQLite feature. + +Observed RED: 6 recalled durable facts after two identical transforms. Expected and observed GREEN: 3 facts. diff --git a/src/ports/sqlite_memory.rs b/src/ports/sqlite_memory.rs index 6cd23c41..9c627269 100644 --- a/src/ports/sqlite_memory.rs +++ b/src/ports/sqlite_memory.rs @@ -4,10 +4,10 @@ //! persists distilled facts in the versioned `memory_facts` table. use std::path::Path; -use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Mutex; use rusqlite::{params, Connection}; +use sha2::{Digest, Sha256}; use super::{MemoryStore, PortError}; use crate::schema::migrate::{self, MigrateError}; @@ -15,7 +15,6 @@ use crate::schema::migrate::{self, MigrateError}; /// Durable [`MemoryStore`] backed by `SQLite` with forward-only migrations. pub struct SqliteMemoryStore { conn: Mutex, - next_id: AtomicU64, } impl SqliteMemoryStore { @@ -31,12 +30,7 @@ impl SqliteMemoryStore { .map_err(|error| map_sqlite(&error))?; migrate::apply_all(&conn).map_err(|error| map_migrate(&error))?; - let count: i64 = conn - .query_row("SELECT COUNT(*) FROM memory_facts", [], |row| row.get(0)) - .map_err(|error| map_sqlite(&error))?; - let next_id = u64::try_from(count).unwrap_or(0); - - Ok(Self { conn: Mutex::new(conn), next_id: AtomicU64::new(next_id) }) + Ok(Self { conn: Mutex::new(conn) }) } /// Open an in-memory database for tests. @@ -47,7 +41,7 @@ impl SqliteMemoryStore { pub fn open_in_memory() -> Result { let conn = Connection::open_in_memory().map_err(|error| map_sqlite(&error))?; migrate::apply_all(&conn).map_err(|error| map_migrate(&error))?; - Ok(Self { conn: Mutex::new(conn), next_id: AtomicU64::new(0) }) + Ok(Self { conn: Mutex::new(conn) }) } /// Lightweight readiness probe for `/readyz` dependency checks. @@ -65,8 +59,7 @@ impl SqliteMemoryStore { impl MemoryStore for SqliteMemoryStore { fn store(&self, session_id: &str, key: &str, content: &str) -> Result { - let sequence = self.next_id.fetch_add(1, Ordering::Relaxed); - let id = format!("memory-{sequence:020}"); + let id = fact_id(session_id, key, content); let payload = serde_json::json!({ "key": key, "content": content, @@ -78,7 +71,9 @@ impl MemoryStore for SqliteMemoryStore { PortError::Backend(format!("sqlite memory store lock poisoned: {error}")) })?; conn.execute( - "INSERT INTO memory_facts (id, session_id, kind, payload_json) VALUES (?1, ?2, 'EPISODIC', ?3)", + "INSERT INTO memory_facts (id, session_id, kind, payload_json) + VALUES (?1, ?2, 'EPISODIC', ?3) + ON CONFLICT(id) DO NOTHING", params![id, session_id, payload_json], ) .map_err(|error| map_sqlite(&error))?; @@ -129,6 +124,21 @@ impl MemoryStore for SqliteMemoryStore { } } +fn fact_id(session_id: &str, key: &str, content: &str) -> String { + let mut hasher = Sha256::new(); + for value in [session_id, key, content] { + hasher.update(u64::try_from(value.len()).unwrap_or(u64::MAX).to_be_bytes()); + hasher.update(value.as_bytes()); + } + let digest = hasher.finalize(); + let mut id = String::from("memory-"); + for byte in digest { + use std::fmt::Write; + let _ = write!(id, "{byte:02x}"); + } + id +} + fn map_sqlite(error: &rusqlite::Error) -> PortError { PortError::Backend(format!("sqlite memory store: {error}")) }