From 6137b67811c756c993967e19fd7e788d32515f2d Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Thu, 21 May 2026 16:31:25 +0530 Subject: [PATCH] fix(memory_tree,sync_status,scripts): IMMEDIATE-tx ingest, reembed skip-persistence, sidecar-based sync-status accounting, Windows dev-script PATH (#2349) Co-authored-by: Claude Opus 4.7 (1M context) Co-authored-by: sanil-23 --- scripts/run-dev-win.sh | 42 ++ src/openhuman/memory/sync_status/rpc.rs | 450 ++++++++++++++---- src/openhuman/memory/tree/ingest.rs | 18 +- .../memory/tree/jobs/handlers/mod.rs | 260 +++++++++- src/openhuman/memory/tree/jobs/mod.rs | 16 +- src/openhuman/memory/tree/store.rs | 85 +++- .../memory/tree/tree_source/store.rs | 25 + 7 files changed, 786 insertions(+), 110 deletions(-) diff --git a/scripts/run-dev-win.sh b/scripts/run-dev-win.sh index 8492a24da..aa2bdcef1 100644 --- a/scripts/run-dev-win.sh +++ b/scripts/run-dev-win.sh @@ -488,6 +488,29 @@ if [[ -z "$PNPM_EXE" ]]; then exit 1 fi echo "[run-dev-win] pnpm resolved to: $PNPM_EXE" + +# `cargo tauri dev` runs its beforeDevCommand (`pnpm run dev`) via a native +# `cmd /S /C` that resolves bare `pnpm` off PATH. This script otherwise only +# ever calls pnpm by absolute path, so its dir was never on PATH and Tauri +# dies with "'pnpm' is not recognized". Prepend the resolved pnpm's dir — it +# ships pnpm.CMD alongside the bash shim, which cmd.exe uses. +# Split the dirname computation out of the export so a `dirname` failure +# surfaces with a non-zero exit (SC2155) instead of being swallowed by the +# enclosing `export`. `dirname` on a validated absolute path is reliable +# in practice, but the strict-mode posture is worth the extra line. +PNPM_DIR="$(dirname "$PNPM_EXE")" +# `dirname` returns `.` for a bare filename (e.g. if PNPM_EXE somehow +# resolved to just "pnpm" without a path component). Prepending `.` would +# inject the current working directory into PATH on a Windows dev machine +# — a privilege-escalation-flavoured surprise. Skip the prepend in that +# case (and on the also-degenerate empty result); the absolute-path call +# sites elsewhere in this script still work. +if [[ -n "$PNPM_DIR" && "$PNPM_DIR" != "." ]]; then + export PATH="$PNPM_DIR:$PATH" + echo "[run-dev-win] pnpm dir prepended to PATH: $PNPM_DIR" +else + echo "[run-dev-win] pnpm dir not prepended to PATH (PNPM_EXE has no path component: $PNPM_EXE)" +fi echo "[run-dev-win] node on bash PATH: $(command -v node 2>/dev/null || echo '')" echo "[run-dev-win] node.exe on bash PATH: $(command -v node.exe 2>/dev/null || echo '')" @@ -576,6 +599,25 @@ else DEV_PORT=1420 fi +# Tauri spawns beforeDevCommand (`pnpm run dev`) via a native `cmd /S /C` +# inheriting THIS process's env. By here PATH has the full system PATH stacked +# several times over (vcvars rebuild + Git-Bash /etc/profile re-runs + pnpm +# .bin layering); the MSYS→Windows conversion overflows the process +# environment-block limit, so the child inherits an EMPTY PATH and Tauri dies +# with "'pnpm' is not recognized" (even `where` is gone). Collapse PATH to +# first-seen entries (clean POSIX `/c/...` entries, so ':' split is safe). +_dedup_seen=":" +_dedup_new="" +IFS=':' read -ra _dedup_parts <<< "$PATH" +for _dp in "${_dedup_parts[@]}"; do + [[ -z "$_dp" ]] && continue + case "$_dedup_seen" in *":$_dp:"*) continue ;; esac + _dedup_seen="${_dedup_seen}${_dp}:" + _dedup_new="${_dedup_new:+$_dedup_new:}$_dp" +done +export PATH="$_dedup_new" +echo "[run-dev-win] PATH de-duplicated: ${#_dedup_parts[@]} → $(awk -v RS=: 'END{print NR}' <<< "$_dedup_new") entries" + if (( DEV_PORT != 1420 )); then echo "[run-dev-win] OPENHUMAN_DEV_PORT=$DEV_PORT — overriding tauri devUrl" "$PNPM_EXE" tauri dev -c "{\"build\":{\"devUrl\":\"http://localhost:$DEV_PORT\"}}" diff --git a/src/openhuman/memory/sync_status/rpc.rs b/src/openhuman/memory/sync_status/rpc.rs index 9c2a4f225..9afac862e 100644 --- a/src/openhuman/memory/sync_status/rpc.rs +++ b/src/openhuman/memory/sync_status/rpc.rs @@ -3,8 +3,27 @@ //! Single SQL query against `mem_tree_chunks`. Two layers of metrics: //! //! * **Lifetime** — `chunks_synced` (total ingested), `chunks_pending` -//! (`embedding IS NULL` = still in the extract+embed queue, not -//! yet appended to the source-tree buffer). +//! (not yet *resolved* = still in the extract+embed queue, not yet +//! appended to the source-tree buffer). +//! +//! A chunk is "resolved" (i.e. NOT pending) when ANY of: +//! - it has a row in the per-(chunk,model) sidecar +//! `mem_tree_chunk_embeddings` (#1574) — embedded under some model; +//! - `lifecycle_status = 'dropped'` — the admission gate rejected it, +//! so it is intentionally never embedded (terminal, not waiting); +//! - it has a `mem_tree_chunk_reembed_skipped` tombstone (#1574 §6) — +//! embedding failed terminally (missing body / wrong dim / embed +//! error) and will not be retried (terminal, not waiting). +//! +//! NOTE: "embedded" is keyed off the sidecar table, NOT the legacy +//! inline `mem_tree_chunks.embedding` column. The #1574 §7 migration +//! copied every vector into the sidecar and stopped writing the inline +//! column, so it now reads back NULL for every chunk. Keying pending / +//! processed off the inline column made this RPC report 100% of chunks +//! as pending and `0` processed forever, regardless of real progress. +//! Dropped / terminally-skipped chunks have no sidecar row either, so +//! without the extra terminal predicates they would read as pending +//! forever and could pin a provider's progress bar below 100%. //! //! * **Active sync wave** — `batch_total` / `batch_processed`. The //! wave is identified by a *time-cluster anchor*: the earliest @@ -27,6 +46,7 @@ use crate::openhuman::config::Config; use crate::openhuman::memory::tree::store::with_connection; use crate::rpc::RpcOutcome; +use rusqlite::Connection; use super::types::{FreshnessLabel, MemorySyncStatus, StatusListResponse}; @@ -44,89 +64,8 @@ pub async fn status_list_rpc(config: &Config) -> Result = match tokio::task::spawn_blocking(move || { with_connection(&config, |conn| -> anyhow::Result> { - // Provider parsed from `source_id` prefix (substring before - // first ':'); falls back to `source_kind` when no prefix. - // - // `provider_chunks` projects per-row provider + the columns - // we need. `provider_pending` flags providers that still - // have at least one chunk waiting for an embedding — - // `wave_anchors` is gated on this so a fully-drained - // provider gets `batch_total = batch_processed = 0` (the - // UI then hides the progress bar instead of rendering a - // completed one for an idle connection). `wave_anchors` - // finds the earliest chunk within WAVE_WINDOW_MS of the - // most recent — the wave's start. The outer SELECT joins - // back to count both lifetime and in-wave totals. - let mut stmt = conn.prepare( - "WITH provider_chunks AS ( \ - SELECT \ - CASE \ - WHEN INSTR(source_id, ':') > 0 \ - THEN SUBSTR(source_id, 1, INSTR(source_id, ':') - 1) \ - ELSE source_kind \ - END AS provider, \ - created_at_ms, \ - embedding, \ - timestamp_ms \ - FROM mem_tree_chunks \ - ), \ - provider_max AS ( \ - SELECT provider, MAX(created_at_ms) AS max_created \ - FROM provider_chunks \ - GROUP BY provider \ - ), \ - provider_pending AS ( \ - SELECT provider, \ - SUM(CASE WHEN embedding IS NULL THEN 1 ELSE 0 END) AS pending \ - FROM provider_chunks \ - GROUP BY provider \ - ), \ - wave_anchors AS ( \ - SELECT p.provider, MIN(p.created_at_ms) AS anchor \ - FROM provider_chunks p \ - JOIN provider_max m ON p.provider = m.provider \ - JOIN provider_pending pp ON p.provider = pp.provider \ - WHERE pp.pending > 0 \ - AND p.created_at_ms >= m.max_created - ?1 \ - GROUP BY p.provider \ - ) \ - SELECT \ - p.provider, \ - COUNT(*) AS chunks_synced, \ - SUM(CASE WHEN p.embedding IS NULL THEN 1 ELSE 0 END) AS chunks_pending, \ - SUM(CASE WHEN w.anchor IS NOT NULL \ - AND p.created_at_ms >= w.anchor \ - THEN 1 ELSE 0 END) AS batch_total, \ - SUM(CASE WHEN w.anchor IS NOT NULL \ - AND p.created_at_ms >= w.anchor \ - AND p.embedding IS NOT NULL \ - THEN 1 ELSE 0 END) AS batch_processed, \ - MAX(p.timestamp_ms) AS last_chunk_at_ms \ - FROM provider_chunks p \ - LEFT JOIN wave_anchors w ON p.provider = w.provider \ - GROUP BY p.provider \ - ORDER BY last_chunk_at_ms DESC", - )?; let now_ms = chrono::Utc::now().timestamp_millis(); - let iter = stmt.query_map([WAVE_WINDOW_MS], |row| { - let provider: String = row.get(0)?; - let chunks_synced: i64 = row.get(1)?; - let chunks_pending: i64 = row.get(2)?; - let batch_total: i64 = row.get(3)?; - let batch_processed: i64 = row.get(4)?; - let last_chunk_at_ms: Option = row.get(5)?; - Ok(MemorySyncStatus { - provider, - chunks_synced: chunks_synced.max(0) as u64, - chunks_pending: chunks_pending.max(0) as u64, - batch_total: batch_total.max(0) as u64, - batch_processed: batch_processed.max(0) as u64, - last_chunk_at_ms, - freshness: FreshnessLabel::from_age_ms(last_chunk_at_ms, now_ms), - }) - })?; - let out = iter.collect::, _>>()?; - Ok(out) + Ok(query_sync_statuses(conn, now_ms)?) }) }) .await @@ -159,6 +98,111 @@ pub async fn status_list_rpc(config: &Config) -> Result rusqlite::Result> { + // Provider parsed from `source_id` prefix (substring before first ':'); + // falls back to `source_kind` when no prefix. + // + // `provider_chunks` projects per-row provider + a `resolved` flag (embedded + // OR dropped OR terminally skipped). `provider_pending` flags providers with + // at least one unresolved chunk *inside the wave window* (within + // WAVE_WINDOW_MS of the provider's most recent chunk) — `wave_anchors` is + // gated on this, so a stale unresolved chunk from an older wave can't + // resurrect an "active" wave when the recent chunks are all resolved, and a + // fully-drained provider gets `batch_total = batch_processed = 0` (the UI + // then hides the progress bar instead of rendering a completed one for an + // idle connection). `wave_anchors` finds the earliest chunk within + // WAVE_WINDOW_MS of the most recent — the wave's start. The outer SELECT + // joins back to count both lifetime and in-wave totals. + let mut stmt = conn.prepare( + "WITH provider_chunks AS ( \ + SELECT \ + CASE \ + WHEN INSTR(source_id, ':') > 0 \ + THEN SUBSTR(source_id, 1, INSTR(source_id, ':') - 1) \ + ELSE source_kind \ + END AS provider, \ + created_at_ms, \ + CASE WHEN EXISTS ( \ + SELECT 1 FROM mem_tree_chunk_embeddings e \ + WHERE e.chunk_id = c.id \ + ) \ + OR c.lifecycle_status = 'dropped' \ + OR EXISTS ( \ + SELECT 1 FROM mem_tree_chunk_reembed_skipped s \ + WHERE s.chunk_id = c.id \ + ) THEN 1 ELSE 0 END AS resolved, \ + timestamp_ms \ + FROM mem_tree_chunks c \ + ), \ + provider_max AS ( \ + SELECT provider, MAX(created_at_ms) AS max_created \ + FROM provider_chunks \ + GROUP BY provider \ + ), \ + provider_pending AS ( \ + SELECT p.provider, \ + SUM(CASE WHEN p.resolved = 0 \ + AND p.created_at_ms >= m.max_created - ?1 \ + THEN 1 ELSE 0 END) AS pending \ + FROM provider_chunks p \ + JOIN provider_max m ON p.provider = m.provider \ + GROUP BY p.provider \ + ), \ + wave_anchors AS ( \ + SELECT p.provider, MIN(p.created_at_ms) AS anchor \ + FROM provider_chunks p \ + JOIN provider_max m ON p.provider = m.provider \ + JOIN provider_pending pp ON p.provider = pp.provider \ + WHERE pp.pending > 0 \ + AND p.created_at_ms >= m.max_created - ?1 \ + GROUP BY p.provider \ + ) \ + SELECT \ + p.provider, \ + COUNT(*) AS chunks_synced, \ + SUM(CASE WHEN p.resolved = 0 THEN 1 ELSE 0 END) AS chunks_pending, \ + SUM(CASE WHEN w.anchor IS NOT NULL \ + AND p.created_at_ms >= w.anchor \ + THEN 1 ELSE 0 END) AS batch_total, \ + SUM(CASE WHEN w.anchor IS NOT NULL \ + AND p.created_at_ms >= w.anchor \ + AND p.resolved = 1 \ + THEN 1 ELSE 0 END) AS batch_processed, \ + MAX(p.timestamp_ms) AS last_chunk_at_ms \ + FROM provider_chunks p \ + LEFT JOIN wave_anchors w ON p.provider = w.provider \ + GROUP BY p.provider \ + ORDER BY last_chunk_at_ms DESC", + )?; + let iter = stmt.query_map([WAVE_WINDOW_MS], |row| { + let provider: String = row.get(0)?; + let chunks_synced: i64 = row.get(1)?; + let chunks_pending: i64 = row.get(2)?; + let batch_total: i64 = row.get(3)?; + let batch_processed: i64 = row.get(4)?; + let last_chunk_at_ms: Option = row.get(5)?; + Ok(MemorySyncStatus { + provider, + chunks_synced: chunks_synced.max(0) as u64, + chunks_pending: chunks_pending.max(0) as u64, + batch_total: batch_total.max(0) as u64, + batch_processed: batch_processed.max(0) as u64, + last_chunk_at_ms, + freshness: FreshnessLabel::from_age_ms(last_chunk_at_ms, now_ms), + }) + })?; + iter.collect() +} + #[cfg(test)] mod tests { use super::*; @@ -195,4 +239,242 @@ mod tests { assert!(json.get("result").is_none(), "must not be double-wrapped"); assert!(json.get("logs").is_none(), "must not be double-wrapped"); } + + /// Regression for the legacy-column bug: pending / processed must be + /// derived from the `mem_tree_chunk_embeddings` sidecar, not the inline + /// `mem_tree_chunks.embedding` column (which is always NULL post-#1574). + /// A chunk with a sidecar row counts as processed even though its inline + /// column is NULL. + #[test] + fn pending_and_processed_key_off_sidecar_not_inline_column() { + use crate::openhuman::memory::tree::store::with_connection; + use rusqlite::params; + use tempfile::TempDir; + + let tmp = TempDir::new().expect("tempdir"); + let mut cfg = Config::default(); + cfg.workspace_dir = tmp.path().to_path_buf(); + + let now = chrono::Utc::now().timestamp_millis(); + + with_connection(&cfg, |conn| { + let insert_chunk = |id: &str, source_id: &str, created: i64| { + conn.execute( + "INSERT INTO mem_tree_chunks \ + (id, source_kind, source_id, owner, timestamp_ms, \ + time_range_start_ms, time_range_end_ms, content, \ + token_count, seq_in_source, created_at_ms) \ + VALUES (?1, 'email', ?2, 'me@x.com', ?3, ?3, ?3, 'body', 10, 0, ?3)", + params![id, source_id, created], + ) + .unwrap(); + }; + let embed = |id: &str| { + conn.execute( + "INSERT INTO mem_tree_chunk_embeddings \ + (chunk_id, model_signature, vector, dim, created_at) \ + VALUES (?1, 'sig', X'00000000', 1, 0.0)", + params![id], + ) + .unwrap(); + }; + + // gmail: 3 chunks inside the active wave; 2 embedded (sidecar), 1 not. + insert_chunk("g1", "gmail:acct", now - 1_000); + insert_chunk("g2", "gmail:acct", now - 2_000); + insert_chunk("g3", "gmail:acct", now - 3_000); + embed("g1"); + embed("g2"); + + let statuses = query_sync_statuses(conn, now).unwrap(); + let gmail = statuses + .iter() + .find(|s| s.provider == "gmail") + .expect("gmail provider row"); + + assert_eq!(gmail.chunks_synced, 3, "all three ingested"); + assert_eq!( + gmail.chunks_pending, 1, + "only g3 lacks a sidecar embedding (inline column is NULL for all)" + ); + assert_eq!(gmail.batch_total, 3, "all three are within the wave window"); + assert_eq!( + gmail.batch_processed, 2, + "g1 and g2 have sidecar rows, so they count as processed" + ); + Ok(()) + }) + .unwrap(); + } + + /// A provider with every chunk embedded must report zero wave (the UI + /// hides the progress bar): `batch_total = batch_processed = 0`. + #[test] + fn fully_embedded_provider_reports_no_active_wave() { + use crate::openhuman::memory::tree::store::with_connection; + use rusqlite::params; + use tempfile::TempDir; + + let tmp = TempDir::new().expect("tempdir"); + let mut cfg = Config::default(); + cfg.workspace_dir = tmp.path().to_path_buf(); + let now = chrono::Utc::now().timestamp_millis(); + + with_connection(&cfg, |conn| { + conn.execute( + "INSERT INTO mem_tree_chunks \ + (id, source_kind, source_id, owner, timestamp_ms, \ + time_range_start_ms, time_range_end_ms, content, \ + token_count, seq_in_source, created_at_ms) \ + VALUES ('s1', 'slack', 'slack:eng', 'me@x.com', ?1, ?1, ?1, 'b', 10, 0, ?1)", + params![now - 5_000], + ) + .unwrap(); + conn.execute( + "INSERT INTO mem_tree_chunk_embeddings \ + (chunk_id, model_signature, vector, dim, created_at) \ + VALUES ('s1', 'sig', X'00000000', 1, 0.0)", + [], + ) + .unwrap(); + + let statuses = query_sync_statuses(conn, now).unwrap(); + let slack = statuses + .iter() + .find(|s| s.provider == "slack") + .expect("slack provider row"); + assert_eq!(slack.chunks_pending, 0); + assert_eq!(slack.batch_total, 0, "no pending chunks ⇒ no active wave"); + assert_eq!(slack.batch_processed, 0); + Ok(()) + }) + .unwrap(); + } + + /// Terminal-but-unembedded chunks must not read as perpetually pending: + /// a `dropped` chunk (admission-rejected) and a `reembed_skipped` + /// tombstoned chunk both count as resolved even with no sidecar row, so a + /// provider whose only leftovers are terminal drains to 0 pending / no wave. + #[test] + fn dropped_and_skipped_chunks_count_as_resolved_not_pending() { + use crate::openhuman::memory::tree::store::with_connection; + use rusqlite::params; + use tempfile::TempDir; + + let tmp = TempDir::new().expect("tempdir"); + let mut cfg = Config::default(); + cfg.workspace_dir = tmp.path().to_path_buf(); + let now = chrono::Utc::now().timestamp_millis(); + + with_connection(&cfg, |conn| { + let insert = |id: &str, lifecycle: &str, created: i64| { + conn.execute( + "INSERT INTO mem_tree_chunks \ + (id, source_kind, source_id, owner, timestamp_ms, \ + time_range_start_ms, time_range_end_ms, content, \ + token_count, seq_in_source, created_at_ms, lifecycle_status) \ + VALUES (?1, 'slack', 'slack:eng', 'me@x.com', ?2, ?2, ?2, 'b', 10, 0, ?2, ?3)", + params![id, created, lifecycle], + ) + .unwrap(); + }; + + // d1: gate-dropped (no embedding, never will be). + insert("d1", "dropped", now - 4_000); + // sk1: pending_extraction but terminally tombstoned (e.g. body missing). + insert("sk1", "pending_extraction", now - 3_000); + conn.execute( + "INSERT INTO mem_tree_chunk_reembed_skipped \ + (chunk_id, model_signature, reason, skipped_at_ms) \ + VALUES ('sk1', 'sig', 'body read failed', ?1)", + params![now - 2_000], + ) + .unwrap(); + // p1: genuinely still in the queue (no embedding, no terminal marker). + insert("p1", "pending_extraction", now - 1_000); + + let statuses = query_sync_statuses(conn, now).unwrap(); + let slack = statuses + .iter() + .find(|s| s.provider == "slack") + .expect("slack provider row"); + + assert_eq!(slack.chunks_synced, 3, "all three ingested"); + assert_eq!( + slack.chunks_pending, 1, + "only p1 is genuinely pending; d1 (dropped) and sk1 (skipped) are terminal" + ); + // p1 keeps the wave alive; d1+sk1 are in-window but resolved. + assert_eq!(slack.batch_total, 3, "all within the wave window"); + assert_eq!( + slack.batch_processed, 2, + "d1 and sk1 count as resolved; p1 does not" + ); + Ok(()) + }) + .unwrap(); + } + + /// The active wave must be gated on an unresolved chunk *inside the window*. + /// A stale unresolved chunk from an older wave plus a fully-resolved recent + /// chunk must NOT resurrect an active wave (no bogus 100%-complete bar): + /// `batch_total = batch_processed = 0`, while lifetime `chunks_pending` + /// still reflects the old straggler. + #[test] + fn stale_out_of_window_pending_does_not_open_a_wave() { + use crate::openhuman::memory::tree::store::with_connection; + use rusqlite::params; + use tempfile::TempDir; + + let tmp = TempDir::new().expect("tempdir"); + let mut cfg = Config::default(); + cfg.workspace_dir = tmp.path().to_path_buf(); + let now = chrono::Utc::now().timestamp_millis(); + // WAVE_WINDOW_MS is 10 min; place the straggler well outside it. + let old = now - 30 * 60 * 1000; + + with_connection(&cfg, |conn| { + let insert = |id: &str, created: i64| { + conn.execute( + "INSERT INTO mem_tree_chunks \ + (id, source_kind, source_id, owner, timestamp_ms, \ + time_range_start_ms, time_range_end_ms, content, \ + token_count, seq_in_source, created_at_ms) \ + VALUES (?1, 'gmail', 'gmail:acct', 'me@x.com', ?2, ?2, ?2, 'b', 10, 0, ?2)", + params![id, created], + ) + .unwrap(); + }; + + // old straggler: unresolved, 30 min ago (outside the wave window). + insert("old1", old); + // recent: resolved (embedded), inside the window. + insert("new1", now - 1_000); + conn.execute( + "INSERT INTO mem_tree_chunk_embeddings \ + (chunk_id, model_signature, vector, dim, created_at) \ + VALUES ('new1', 'sig', X'00000000', 1, 0.0)", + [], + ) + .unwrap(); + + let statuses = query_sync_statuses(conn, now).unwrap(); + let gmail = statuses + .iter() + .find(|s| s.provider == "gmail") + .expect("gmail provider row"); + + assert_eq!( + gmail.chunks_pending, 1, + "the old straggler is still pending lifetime-wise" + ); + assert_eq!( + gmail.batch_total, 0, + "no unresolved chunk inside the window ⇒ no active wave" + ); + assert_eq!(gmail.batch_processed, 0); + Ok(()) + }) + .unwrap(); + } } diff --git a/src/openhuman/memory/tree/ingest.rs b/src/openhuman/memory/tree/ingest.rs index 5bd08be50..e026c06ca 100644 --- a/src/openhuman/memory/tree/ingest.rs +++ b/src/openhuman/memory/tree/ingest.rs @@ -227,7 +227,23 @@ async fn persist( let written = tokio::task::spawn_blocking(move || -> Result> { use std::collections::{HashMap, HashSet}; store::with_connection(&config_owned, |conn| { - let tx = conn.unchecked_transaction()?; + // IMMEDIATE, not the default DEFERRED: this transaction reads + // (get_chunk_lifecycle_status_tx) before it writes + // (upsert_staged_chunks_tx). A DEFERRED tx takes only a read + // lock at BEGIN and tries to upgrade to a write lock on the + // first write; under contention with the memory_tree worker + // pool SQLite returns SQLITE_BUSY *immediately* for that + // upgrade and does NOT invoke the busy handler (deadlock + // avoidance), so the connection's 15s busy_timeout is bypassed + // and Gmail/Composio ingest fails every message with "database + // is locked", stalling composio_sync past its 30s RPC cap. + // IMMEDIATE acquires the write lock at BEGIN, where the busy + // handler / busy_timeout DOES apply, so writers serialise and + // wait instead of failing fast. + let tx = rusqlite::Transaction::new_unchecked( + conn, + rusqlite::TransactionBehavior::Immediate, + )?; // Authoritative source-level gate (documents only). // diff --git a/src/openhuman/memory/tree/jobs/handlers/mod.rs b/src/openhuman/memory/tree/jobs/handlers/mod.rs index a27b7fe20..23b3ba59d 100644 --- a/src/openhuman/memory/tree/jobs/handlers/mod.rs +++ b/src/openhuman/memory/tree/jobs/handlers/mod.rs @@ -27,6 +27,7 @@ use crate::openhuman::memory::tree::score::extract::build_summary_extractor; use crate::openhuman::memory::tree::score::store as score_store; use crate::openhuman::memory::tree::store as chunk_store; use crate::openhuman::memory::tree::tree_global::digest::{self, DigestOutcome}; +use crate::openhuman::memory::tree::tree_source::store as summary_store; use crate::openhuman::memory::tree::tree_source::{ build_summariser, get_or_create_source_tree, LabelStrategy, LeafRef, }; @@ -586,10 +587,22 @@ async fn handle_reembed_backfill(config: &Config, job: &Job) -> Result = { let mut stmt = conn.prepare( + // The second NOT EXISTS — `mem_tree_chunk_reembed_skipped` — + // is the runaway-loop fix (#1574 §6): without it, rows whose + // body file is missing on disk (or whose embed failed + // terminally) keep matching the worklist on every batch + // because the failure path only LOG-skipped, never wrote + // anything persistent. The handler below now marks such + // rows in `mem_tree_chunk_reembed_skipped` so they're + // excluded here on the next batch and the chain can + // actually reach "fully covered". "SELECT id FROM mem_tree_chunks c WHERE NOT EXISTS ( SELECT 1 FROM mem_tree_chunk_embeddings e WHERE e.chunk_id = c.id AND e.model_signature = ?1) + AND NOT EXISTS ( + SELECT 1 FROM mem_tree_chunk_reembed_skipped s + WHERE s.chunk_id = c.id AND s.model_signature = ?1) LIMIT ?2", )?; let ids = stmt @@ -605,10 +618,16 @@ async fn handle_reembed_backfill(config: &Config, job: &Job) -> Result Result)> = Vec::new(); @@ -639,16 +668,40 @@ async fn handle_reembed_backfill(config: &Config, job: &Job) -> Result match embedder.embed(&body).await { Ok(v) if pack_checked(&v).is_ok() => chunk_vecs.push((id.clone(), v)), - Ok(_) => log::warn!( - "[memory_tree::jobs] reembed_backfill: chunk {id} embed wrong dim, skipping" - ), - Err(e) => log::warn!( - "[memory_tree::jobs] reembed_backfill: chunk {id} embed failed: {e}; skipping" - ), + Ok(_) => { + log::warn!( + "[memory_tree::jobs] reembed_backfill: chunk {id} embed wrong dim, skipping (sig={active_sig})" + ); + let _ = chunk_store::mark_chunk_reembed_skipped( + config, + id, + &active_sig, + "embed wrong dim", + ); + } + Err(e) => { + log::warn!( + "[memory_tree::jobs] reembed_backfill: chunk {id} embed failed: {e}; skipping (sig={active_sig})" + ); + let _ = chunk_store::mark_chunk_reembed_skipped( + config, + id, + &active_sig, + &format!("embed failed: {e}"), + ); + } }, - Err(e) => log::warn!( - "[memory_tree::jobs] reembed_backfill: chunk {id} body read failed: {e}; skipping" - ), + Err(e) => { + log::warn!( + "[memory_tree::jobs] reembed_backfill: chunk {id} body read failed: {e}; skipping (sig={active_sig})" + ); + let _ = chunk_store::mark_chunk_reembed_skipped( + config, + id, + &active_sig, + &format!("body read failed: {e}"), + ); + } } } let mut summary_vecs: Vec<(String, Vec)> = Vec::new(); @@ -656,16 +709,40 @@ async fn handle_reembed_backfill(config: &Config, job: &Job) -> Result match embedder.embed(&body).await { Ok(v) if pack_checked(&v).is_ok() => summary_vecs.push((id.clone(), v)), - Ok(_) => log::warn!( - "[memory_tree::jobs] reembed_backfill: summary {id} embed wrong dim, skipping" - ), - Err(e) => log::warn!( - "[memory_tree::jobs] reembed_backfill: summary {id} embed failed: {e}; skipping" - ), + Ok(_) => { + log::warn!( + "[memory_tree::jobs] reembed_backfill: summary {id} embed wrong dim, skipping (sig={active_sig})" + ); + let _ = summary_store::mark_summary_reembed_skipped( + config, + id, + &active_sig, + "embed wrong dim", + ); + } + Err(e) => { + log::warn!( + "[memory_tree::jobs] reembed_backfill: summary {id} embed failed: {e}; skipping (sig={active_sig})" + ); + let _ = summary_store::mark_summary_reembed_skipped( + config, + id, + &active_sig, + &format!("embed failed: {e}"), + ); + } }, - Err(e) => log::warn!( - "[memory_tree::jobs] reembed_backfill: summary {id} body read failed: {e}; skipping" - ), + Err(e) => { + log::warn!( + "[memory_tree::jobs] reembed_backfill: summary {id} body read failed: {e}; skipping (sig={active_sig})" + ); + let _ = summary_store::mark_summary_reembed_skipped( + config, + id, + &active_sig, + &format!("body read failed: {e}"), + ); + } } } @@ -1154,6 +1231,151 @@ mod tests { ); } + /// #1574 §6 regression gate: a terminal-failure chunk (its body file is + /// missing on disk, despite the metadata row staying staged) is + /// persistently tombstoned by `mark_chunk_reembed_skipped` on the first + /// pass, then excluded from the next batch's worklist so the chain + /// terminates (`Done`) instead of looping forever. Without this guard + /// the §6 runaway-loop fix would silently regress — the same 16 orphans + /// → ~8k defers → ~128k warns symptom observed in the wild before the + /// fix landed (see PR body and store.rs:1195). + /// + /// What the test pins: + /// 1. Tombstone row is written for the failing chunk (exactly one). + /// 2. The next-batch worklist `NOT EXISTS … reembed_skipped` clause + /// excludes the tombstoned row — the handler returns `Done`. + /// 3. The `ensure_reembed_backfill` migration probe agrees the space + /// is covered (or the chain would re-arm on every config save). + #[tokio::test] + async fn reembed_backfill_tombstones_orphan_and_terminates() { + use crate::openhuman::memory::tree::store::{ + get_chunk_content_path, get_chunk_embedding_for_signature, tree_active_signature, + upsert_chunks, upsert_staged_chunks_tx, + }; + use crate::openhuman::memory::tree::types::{ + chunk_id, Chunk, Metadata, SourceKind, SourceRef, + }; + + let (_tmp, cfg) = test_config(); + let ts = chrono::Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(); + let chunk = Chunk { + id: chunk_id(SourceKind::Chat, "slack:#eng", 0, "orphan-seed"), + content: "memory content about the orphaned phoenix project".into(), + metadata: Metadata { + source_kind: SourceKind::Chat, + source_id: "slack:#eng".into(), + owner: "alice".into(), + timestamp: ts, + time_range: (ts, ts), + tags: vec![], + source_ref: Some(SourceRef::new("slack://x")), + }, + token_count: 12, + seq_in_source: 0, + created_at: ts, + partial_message: false, + }; + upsert_chunks(&cfg, &[chunk.clone()]).unwrap(); + + // Stage the body file + metadata, then DELETE the body file from + // disk while leaving the staged DB rows intact. Reproduces the + // in-wild failure mode: chunk row + path hash both present, but + // the body content was lost (user moved workspace dirs, partial + // backup restore, manual file cleanup). `stage_chunks` returns + // paths relative to `content_root`; resolve absolute before unlink. + let content_root = cfg.memory_tree_content_root(); + std::fs::create_dir_all(&content_root).unwrap(); + let staged = content_store::stage_chunks(&content_root, &[chunk.clone()]).unwrap(); + with_connection(&cfg, |conn| { + let tx = conn.unchecked_transaction()?; + upsert_staged_chunks_tx(&tx, &staged)?; + tx.commit()?; + Ok(()) + }) + .unwrap(); + let staged_rel = get_chunk_content_path(&cfg, &chunk.id) + .unwrap() + .expect("staged body path"); + let body_abs = content_root.join(&staged_rel); + std::fs::remove_file(&body_abs).unwrap(); + + let sig = tree_active_signature(&cfg); + let job = mk_running_job( + JobKind::ReembedBackfill, + serde_json::to_string(&ReembedBackfillPayload { + signature: sig.clone(), + }) + .unwrap(), + ); + + // Pass 1: worklist picks up the orphan, body read fails, tombstone + // written, `Defer` to revisit (the handler doesn't distinguish + // "all rows tombstoned" from "more rows pending" inside this batch). + let out1 = handle_reembed_backfill(&cfg, &job).await.unwrap(); + assert!( + matches!(out1, JobOutcome::Defer { .. }), + "first pass should Defer after failing to read body, got {out1:?}" + ); + assert!( + get_chunk_embedding_for_signature(&cfg, &chunk.id, &sig) + .unwrap() + .is_none(), + "orphan chunk must not have a sidecar vector after failure" + ); + + // (1) Tombstone row exists for exactly this (chunk, sig). + let tombstone_count: i64 = with_connection(&cfg, |conn| { + Ok(conn.query_row( + "SELECT COUNT(*) FROM mem_tree_chunk_reembed_skipped + WHERE chunk_id = ?1 AND model_signature = ?2", + params![chunk.id, sig], + |r| r.get(0), + )?) + }) + .unwrap(); + assert_eq!( + tombstone_count, 1, + "orphan chunk must be tombstoned exactly once" + ); + + // (2) Pass 2: worklist NOT EXISTS clause excludes the tombstoned + // row; both worklists empty; chain completes. + let out2 = handle_reembed_backfill(&cfg, &job).await.unwrap(); + assert_eq!( + out2, + JobOutcome::Done, + "tombstoned-only state must complete the chain" + ); + + // (3) Migration probe in `ensure_reembed_backfill` must agree the + // space is covered, otherwise the chain re-arms on every config + // save and we're back to the original infinite-loop bug. + let probe_uncovered: bool = with_connection(&cfg, |conn| { + Ok(conn.query_row( + "SELECT EXISTS( + SELECT 1 FROM mem_tree_chunks c + WHERE NOT EXISTS (SELECT 1 FROM mem_tree_chunk_embeddings e + WHERE e.chunk_id = c.id AND e.model_signature = ?1) + AND NOT EXISTS (SELECT 1 FROM mem_tree_chunk_reembed_skipped sk + WHERE sk.chunk_id = c.id AND sk.model_signature = ?1)) + OR EXISTS( + SELECT 1 FROM mem_tree_summaries s + WHERE s.deleted = 0 + AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_embeddings e + WHERE e.summary_id = s.id AND e.model_signature = ?1) + AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_reembed_skipped sk + WHERE sk.summary_id = s.id AND sk.model_signature = ?1))", + params![sig], + |r| r.get(0), + )?) + }) + .unwrap(); + assert!( + !probe_uncovered, + "after tombstoning the only orphan, the ensure_reembed_backfill probe must report covered" + ); + } + /// #1574 §4: `ensure_reembed_backfill` (the switch-path trigger) enqueues /// exactly one chain when there is uncovered work, is idempotent on /// re-call (per-signature dedupe), and enqueues nothing for an diff --git a/src/openhuman/memory/tree/jobs/mod.rs b/src/openhuman/memory/tree/jobs/mod.rs index 7550e98ce..e384e7a90 100644 --- a/src/openhuman/memory/tree/jobs/mod.rs +++ b/src/openhuman/memory/tree/jobs/mod.rs @@ -74,14 +74,24 @@ pub fn ensure_reembed_backfill(config: &crate::openhuman::config::Config) { let sig = crate::openhuman::memory::tree::store::tree_active_signature(config); let result = crate::openhuman::memory::tree::store::with_connection(config, |conn| { let has_uncovered: bool = conn.query_row( + // The `NOT EXISTS … reembed_skipped` clauses match the worklist in + // `handle_reembed_backfill`: terminally-failed rows are sentinel- + // marked there and must NOT count as "uncovered" here, otherwise + // this probe keeps reporting "uncovered" → keeps re-enqueueing the + // backfill chain → infinite re-arming (#1574 §6 runaway-loop fix). "SELECT EXISTS( SELECT 1 FROM mem_tree_chunks c WHERE NOT EXISTS (SELECT 1 FROM mem_tree_chunk_embeddings e - WHERE e.chunk_id = c.id AND e.model_signature = ?1)) + WHERE e.chunk_id = c.id AND e.model_signature = ?1) + AND NOT EXISTS (SELECT 1 FROM mem_tree_chunk_reembed_skipped sk + WHERE sk.chunk_id = c.id AND sk.model_signature = ?1)) OR EXISTS( SELECT 1 FROM mem_tree_summaries s - WHERE s.deleted = 0 AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_embeddings e - WHERE e.summary_id = s.id AND e.model_signature = ?1))", + WHERE s.deleted = 0 + AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_embeddings e + WHERE e.summary_id = s.id AND e.model_signature = ?1) + AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_reembed_skipped sk + WHERE sk.summary_id = s.id AND sk.model_signature = ?1))", rusqlite::params![sig], |r| r.get(0), )?; diff --git a/src/openhuman/memory/tree/store.rs b/src/openhuman/memory/tree/store.rs index c505e487d..cf2bb4c97 100644 --- a/src/openhuman/memory/tree/store.rs +++ b/src/openhuman/memory/tree/store.rs @@ -117,6 +117,30 @@ CREATE TABLE IF NOT EXISTS mem_tree_chunk_embeddings ( CREATE INDEX IF NOT EXISTS idx_mem_tree_chunk_embeddings_model ON mem_tree_chunk_embeddings(model_signature); +-- #1574 §6 reembed-backfill terminal-skip tombstone. +-- +-- A row here means: 'this (chunk, signature) pair was attempted and failed +-- terminally (body file missing on disk, embed returned wrong dim, embedder +-- erred unrecoverably) — DO NOT re-enqueue it on the next backfill batch.' +-- +-- Without this table, the reembed worklist's `NOT EXISTS embeddings` predicate +-- keeps re-selecting any chunk that failed read/embed (since no sidecar row +-- was ever written), and `handle_reembed_backfill` loops on the same rows +-- forever — observed in the wild as 16 orphan chunk_ids generating ~128k +-- 'body read failed; skipping' warns across ~8k batch defers. The handler +-- now writes a row here on terminal failure, and the worklist excludes them. +-- Idempotent: the table is created here, and `chrono::Utc` is already imported. +CREATE TABLE IF NOT EXISTS mem_tree_chunk_reembed_skipped ( + chunk_id TEXT NOT NULL REFERENCES mem_tree_chunks(id) ON DELETE CASCADE, + model_signature TEXT NOT NULL, + reason TEXT NOT NULL, + skipped_at_ms INTEGER NOT NULL, + PRIMARY KEY (chunk_id, model_signature) +); + +CREATE INDEX IF NOT EXISTS idx_mem_tree_chunk_reembed_skipped_model + ON mem_tree_chunk_reembed_skipped(model_signature); + -- Phase 2 (#708): per-chunk score rationale for admission debugging. CREATE TABLE IF NOT EXISTS mem_tree_score ( chunk_id TEXT PRIMARY KEY, @@ -224,6 +248,20 @@ CREATE TABLE IF NOT EXISTS mem_tree_summary_embeddings ( CREATE INDEX IF NOT EXISTS idx_mem_tree_summary_embeddings_model ON mem_tree_summary_embeddings(model_signature); +-- #1574 §6 reembed-backfill terminal-skip tombstone (summary side). Mirrors +-- `mem_tree_chunk_reembed_skipped` for the summary worklist. See that table's +-- comment for the full rationale. +CREATE TABLE IF NOT EXISTS mem_tree_summary_reembed_skipped ( + summary_id TEXT NOT NULL REFERENCES mem_tree_summaries(id) ON DELETE CASCADE, + model_signature TEXT NOT NULL, + reason TEXT NOT NULL, + skipped_at_ms INTEGER NOT NULL, + PRIMARY KEY (summary_id, model_signature) +); + +CREATE INDEX IF NOT EXISTS idx_mem_tree_summary_reembed_skipped_model + ON mem_tree_summary_reembed_skipped(model_signature); + -- `mem_tree_buffers` holds the unsealed frontier per (tree, level). One row -- per active level per tree; deleted when the buffer seals (clears) in the -- same transaction as the new summary node row. @@ -1265,14 +1303,25 @@ fn migrate_legacy_embeddings_to_sidecar(conn: &Connection, config: &Config) -> R // table for unrelated callers/tests. Enqueued atomically with the // migration; dedupe key = signature, so exactly one chain per space. let has_uncovered: bool = tx.query_row( + // The `NOT EXISTS … reembed_skipped` clauses match the worklist in + // `handle_reembed_backfill`: terminally-failed rows (body missing, + // embed wrong dim / err) are sentinel-marked there and must NOT count + // as "uncovered" here, otherwise this migration probe keeps reporting + // "uncovered" → keeps enqueueing the backfill chain on every DB open → + // infinite re-arming (#1574 §6 runaway-loop fix). "SELECT EXISTS( SELECT 1 FROM mem_tree_chunks c WHERE NOT EXISTS (SELECT 1 FROM mem_tree_chunk_embeddings e - WHERE e.chunk_id = c.id AND e.model_signature = ?1)) + WHERE e.chunk_id = c.id AND e.model_signature = ?1) + AND NOT EXISTS (SELECT 1 FROM mem_tree_chunk_reembed_skipped sk + WHERE sk.chunk_id = c.id AND sk.model_signature = ?1)) OR EXISTS( SELECT 1 FROM mem_tree_summaries s - WHERE s.deleted = 0 AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_embeddings e - WHERE e.summary_id = s.id AND e.model_signature = ?1))", + WHERE s.deleted = 0 + AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_embeddings e + WHERE e.summary_id = s.id AND e.model_signature = ?1) + AND NOT EXISTS (SELECT 1 FROM mem_tree_summary_reembed_skipped sk + WHERE sk.summary_id = s.id AND sk.model_signature = ?1))", rusqlite::params![sig], |r| r.get(0), )?; @@ -1537,6 +1586,36 @@ pub fn set_chunk_embedding_for_signature( }) } +/// Persistently record that `(chunk_id, signature)` cannot be re-embedded. +/// +/// Called by `handle_reembed_backfill` when the per-chunk body file is +/// missing on disk (orphan) or the embedder rejects the row terminally +/// (wrong dim / unrecoverable embed error). Inserting a row here causes +/// the next backfill batch's worklist query to exclude this chunk via the +/// `NOT EXISTS … mem_tree_chunk_reembed_skipped …` predicate, so the +/// runaway "skipping" loop terminates instead of revisiting the same row +/// every 5 s forever (#1574 §6 fix). +pub fn mark_chunk_reembed_skipped( + config: &Config, + chunk_id: &str, + model_signature: &str, + reason: &str, +) -> Result<()> { + with_connection(config, |conn| { + let now_ms = Utc::now().timestamp_millis(); + conn.execute( + "INSERT INTO mem_tree_chunk_reembed_skipped + (chunk_id, model_signature, reason, skipped_at_ms) + VALUES (?1, ?2, ?3, ?4) + ON CONFLICT(chunk_id, model_signature) DO UPDATE SET + reason = excluded.reason, + skipped_at_ms = excluded.skipped_at_ms", + rusqlite::params![chunk_id, model_signature, reason, now_ms], + )?; + Ok(()) + }) +} + /// Transaction-scoped variant of [`set_chunk_embedding_for_signature`]. /// /// For callers that already hold a `Transaction` (e.g. the chunk-admission diff --git a/src/openhuman/memory/tree/tree_source/store.rs b/src/openhuman/memory/tree/tree_source/store.rs index 41a572519..f8e6f8bf2 100644 --- a/src/openhuman/memory/tree/tree_source/store.rs +++ b/src/openhuman/memory/tree/tree_source/store.rs @@ -338,6 +338,31 @@ pub fn set_summary_embedding_for_signature( }) } +/// Persistently record that `(summary_id, signature)` cannot be re-embedded. +/// Mirror of `tree::store::mark_chunk_reembed_skipped` for the summary side +/// of the reembed worklist (#1574 §6 fix). See that function's doc for the +/// full rationale. +pub fn mark_summary_reembed_skipped( + config: &Config, + summary_id: &str, + model_signature: &str, + reason: &str, +) -> Result<()> { + with_connection(config, |conn| { + let now_ms = Utc::now().timestamp_millis(); + conn.execute( + "INSERT INTO mem_tree_summary_reembed_skipped + (summary_id, model_signature, reason, skipped_at_ms) + VALUES (?1, ?2, ?3, ?4) + ON CONFLICT(summary_id, model_signature) DO UPDATE SET + reason = excluded.reason, + skipped_at_ms = excluded.skipped_at_ms", + params![summary_id, model_signature, reason, now_ms], + )?; + Ok(()) + }) +} + /// Transaction-scoped variant of [`set_summary_embedding_for_signature`], for /// the seal path which inserts the summary row and its embedding in one tx /// (#1574 write-side cutover). Opening a fresh connection there would break