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) <noreply@anthropic.com>
Co-authored-by: sanil-23 <sanil@alphahuman.xyz>

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 '<not found>')"
 echo "[run-dev-win] node.exe on bash PATH: $(command -v node.exe 2>/dev/null || echo '<not found>')"
 
@@ -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\"}}"

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<RpcOutcome<StatusListRes
     let config = config.clone();
     let statuses: Vec<MemorySyncStatus> = match tokio::task::spawn_blocking(move || {
         with_connection(&config, |conn| -> anyhow::Result<Vec<MemorySyncStatus>> {
-            // 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<i64> = 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::<Result<Vec<_>, _>>()?;
-            Ok(out)
+            Ok(query_sync_statuses(conn, now_ms)?)
         })
     })
     .await
@@ -159,6 +98,111 @@ pub async fn status_list_rpc(config: &Config) -> Result<RpcOutcome<StatusListRes
     Ok(RpcOutcome::new(StatusListResponse { statuses }, vec![]))
 }
 
+/// Run the per-provider lifetime + active-wave aggregation against `conn`.
+///
+/// Split out from [`status_list_rpc`] so it can be unit-tested against a
+/// tempdir-backed connection without the async / spawn_blocking wrapper.
+///
+/// A chunk is "resolved" (not pending) when it has a sidecar embedding (any
+/// model signature), OR is `dropped`, OR carries a reembed-skip tombstone —
+/// see the module header. Resolution is keyed off the `mem_tree_chunk_embeddings`
+/// sidecar, NOT the legacy inline `mem_tree_chunks.embedding` column.
+fn query_sync_statuses(conn: &Connection, now_ms: i64) -> rusqlite::Result<Vec<MemorySyncStatus>> {
+    // 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<i64> = 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();
+    }
 }

src/openhuman/memory/tree/ingest.rs

@@ -227,7 +227,23 @@ async fn persist(
     let written = tokio::task::spawn_blocking(move || -> Result<Option<usize>> {
         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).
             //

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<JobOutcom
         chunk_store::with_connection(config, |conn| {
             let chunks: Vec<String> = {
                 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<JobOutcom
                 Vec::new()
             } else {
                 let mut stmt = conn.prepare(
+                    // Summary-side counterpart of the runaway-loop fix; see
+                    // the chunks worklist above for the full rationale.
                     "SELECT id FROM mem_tree_summaries s
-                      WHERE s.deleted = 0 AND NOT EXISTS (
+                      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)
                       LIMIT ?2",
                 )?;
                 let ids = stmt
@@ -632,6 +651,16 @@ async fn handle_reembed_backfill(config: &Config, job: &Job) -> Result<JobOutcom
 
     // Phase 2 (no tx held): embed each row's stored source text. Per-row
     // errors are skipped (logged) so a single bad row can't strand memory.
+    //
+    // #1574 §6 fix: terminal failures (body file missing on disk, embed
+    // wrong dim, embed unrecoverable error) are *persistently* tombstoned
+    // via `mark_chunk_reembed_skipped` / `mark_summary_reembed_skipped`.
+    // The worklist queries above exclude these tombstones, so a single
+    // unembeddable row is attempted at most ONCE per signature instead of
+    // re-selected on every batch forever (the original bug: 16 orphans
+    // generating ~128k warns across ~8k defers, observed in the wild).
+    // The mark itself is best-effort — if its own SQLite write fails the
+    // row will be retried on a later batch, which is the desired fallback.
     let embedder =
         build_embedder_from_config(config).context("build embedder in reembed_backfill")?;
     let mut chunk_vecs: Vec<(String, Vec<f32>)> = Vec::new();
@@ -639,16 +668,40 @@ async fn handle_reembed_backfill(config: &Config, job: &Job) -> Result<JobOutcom
         match content_read::read_chunk_body(config, id) {
             Ok(body) => 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<f32>)> = Vec::new();
@@ -656,16 +709,40 @@ async fn handle_reembed_backfill(config: &Config, job: &Job) -> Result<JobOutcom
         match content_read::read_summary_body(config, id) {
             Ok(body) => 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

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),
         )?;

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

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