diff --git a/.claude/memory.md b/.claude/memory.md index 7fa1a1477..19605869a 100644 --- a/.claude/memory.md +++ b/.claude/memory.md @@ -319,3 +319,12 @@ Quick reference for anyone starting with Claude on this project. Updated by the - **`MemorySourcePatch` was missing limit fields** — `max_commits`, `max_issues`, `max_prs` were absent from `MemorySourcePatch`, `update_source`, and the update schema. Also missing from the TS `MemorySourceEntry` interface in `memorySourcesService.ts`. Both were fixed in this issue. - **Per-source settings UI** — `SourceSettingsPanel.tsx` (sibling of `MemorySourcesRegistry.tsx`) with a `KIND_FIELDS` map driving which limit fields appear per source kind. Empty input = omit from patch (use default); number = set. "All In" button uses `ConfirmationModal` (primary-500 prominent). Toasts via existing `onToast` prop. - **`pnpm i18n:english:check` is pre-failing** — Exits with `total unexpected English: 1312` on a clean base tree. Confirm your new keys aren't among the failures rather than expecting exit 0. `pnpm i18n:check` (key parity) is the real gate. + +## Memory Tree Sync & Raw-Archive Reconcile (branch fix/memory-tree-sync-reconcile) + +- **Tree scope ≠ raw-archive id for GitHub** — `github:owner/repo` (tree registry key) and `github.com/owner/repo` (raw archive id) slugify to DIFFERENT `raw/` dirs (`github-owner-repo` vs `github-com-owner-repo`). Any code resolving a source's raw dir must use `memory_sources::sync::SourceScope { tree_scope, archive_source_id }` — deriving from the tree scope scans the wrong directory and silently reports "nothing to do". +- **Raw-file coverage gate** — `mem_tree_ingested_sources` with `source_kind = "raw_file"`, `source_id = ` (`memory_store/chunks/store.rs: mark_raw_paths_ingested / filter_raw_paths_not_ingested`). Written by github sync + rebuild after each summary batch lands. `raw_coverage()` in `memory_sync/sources/rebuild.rs` backfills the gate once per scope from L1 summary child labels (`commit:` / `issue:` / `pr:` / file stems) before diffing disk vs gate. +- **Workspace periodic scheduler** — `memory_sync/workspace/periodic.rs` (started in `core/jsonrpc.rs` next to the Composio one) is the ONLY thing auto-syncing `github_repo`/`folder`/`rss_feed`/`web_page` sources; the Composio loop walks Composio connections exclusively. Cadence: `config.memory_sync_interval_secs` (`Some(0)` = manual only, default 24h), due-check via in-memory fired-at map + persisted sync-audit fallback. +- **Queue self-heal retries transient failures only** — `memory_queue/scheduler.rs` calls `store::requeue_transient_failed` every 3h (excludes `failure_class = 'unrecoverable'` so bad keys/budget don't retry-loop). Full reset stays manual via `memory_tree_retry_failed` RPC. +- **`openhuman.memory_sources_reconcile` RPC** — Reports per-scope raw coverage `{total_raw_files, covered, pending}`; `execute: true` spawns background incremental rebuild. Same reconcile auto-runs after every sync via `check_and_rebuild_tree`. +- **L0 seal gate is token-only by design** — `should_seal` has NO sibling-count fallback at L0 (docs used to lie about this); small buffers seal via the 3-hourly `flush_stale` path. L≥1 gates on `SUMMARY_FANOUT` (10) — an L1 buffer holding <10 summaries is healthy, not stuck. diff --git a/docs/TEST-COVERAGE-MATRIX.md b/docs/TEST-COVERAGE-MATRIX.md index e3a78e912..9a34744a8 100644 --- a/docs/TEST-COVERAGE-MATRIX.md +++ b/docs/TEST-COVERAGE-MATRIX.md @@ -290,6 +290,7 @@ Canonical mapping of every product feature to its test source(s). Drives gap-fil | 8.2.1 | Context Injection | RI | `tests/autocomplete_memory_e2e.rs` | ✅ | | | 8.2.2 | Memory Consistency | RI | `tests/memory_graph_sync_e2e.rs`, `tests/worker_c_modules_e2e.rs` | ✅ | Worker C RPC E2E verifies memory-tree ingest is reflected by `memory_sync_status_list` | | 8.2.3 | Memory Scaling | RU | `src/openhuman/memory/ingestion_tests.rs` | 🟡 | Soak/scale benchmark not asserted | +| 8.2.4 | Raw-archive sync reconcile | RU+RI | `src/openhuman/memory_sync/sources/rebuild.rs`, `src/openhuman/memory_sync/workspace/periodic.rs`, `tests/json_rpc_e2e.rs` (`json_rpc_memory_sources_reconcile_reports_pending_raw_files`), `tests/memory_sync_pipeline_e2e.rs` | ✅ | Coverage gate + incremental rebuild + workspace periodic scheduler + `memory_sources_reconcile` RPC | ### 8.3 Memory Retrieval Benchmarks diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index 77c253a70..f9f7e1fac 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -1917,6 +1917,11 @@ fn register_domain_subscribers( crate::openhuman::agent_meetings::calendar::register_meet_calendar_subscriber(); crate::openhuman::agent_meetings::bus::register_meeting_event_subscriber(); crate::openhuman::composio::start_periodic_sync(); + // Workspace-kind memory sources (GitHub repos, folders, RSS, web + // pages) get their own cadence loop — the Composio scheduler above + // only walks Composio connections, so without this they only sync + // on manual "Sync now" and silently go stale. + crate::openhuman::memory_sync::workspace::start_workspace_periodic_sync(); // Task-sources proactive ingestion: connection-created hook + poll. crate::openhuman::task_sources::bus::register_task_sources_subscriber(); crate::openhuman::task_sources::start_periodic_poll(); diff --git a/src/openhuman/memory_queue/scheduler.rs b/src/openhuman/memory_queue/scheduler.rs index 938bcf55b..d853a8567 100644 --- a/src/openhuman/memory_queue/scheduler.rs +++ b/src/openhuman/memory_queue/scheduler.rs @@ -26,15 +26,35 @@ pub fn start(config: Config) { tokio::spawn(async move { // Fire once on startup so new installs & restarts don't wait // up to 3 h for the first seal window. + retry_transient_failures(&cfg); enqueue_flush_stale(&cfg); loop { tokio::time::sleep(Duration::from_secs(3 * 60 * 60)).await; + retry_transient_failures(&cfg); enqueue_flush_stale(&cfg); } }); }); } +/// Self-heal the pipeline before each flush window: requeue jobs that +/// failed for transient reasons (network blips, timeouts, SQLITE_BUSY) +/// so chunks never sit unprocessed until the next manual sync. +/// Unrecoverable failures stay parked — see +/// [`store::requeue_transient_failed`]. +fn retry_transient_failures(config: &Config) { + match store::requeue_transient_failed(config) { + Ok(0) => {} + Ok(n) => { + log::info!("[memory::jobs] periodic retry requeued {n} transient-failed job(s)"); + super::worker::wake_workers(); + } + Err(err) => { + log::warn!("[memory::jobs] periodic transient-failure retry failed: {err:#}"); + } + } +} + fn enqueue_flush_stale(config: &Config) { // Take a single `Utc::now()` reading and derive both the date and // 3-hour block from it so the dedupe key can't disagree with itself diff --git a/src/openhuman/memory_queue/store.rs b/src/openhuman/memory_queue/store.rs index cba8eeb1d..83c2db1cb 100644 --- a/src/openhuman/memory_queue/store.rs +++ b/src/openhuman/memory_queue/store.rs @@ -434,10 +434,10 @@ pub fn recover_stale_locks(config: &Config) -> Result { /// clears the typed `failure_reason` / `failure_class` and `last_error`, and /// makes the row immediately available. Returns the number of jobs requeued. /// -/// NOTE: there is currently **no automatic caller**. An automatic -/// requeue-on-sync was planned, but its hook lived on the upstream-removed -/// vault sync path and has not been re-homed, so requeue is **manual-only** -/// (the `memory_tree_retry_failed` RPC) for now. +/// Automatic callers: the manual sync path (`memory_sources::sync`, +/// via [`retry_all_failed`]) and the 3-hourly queue scheduler (via +/// [`requeue_transient_failed`], which excludes `unrecoverable` +/// failures so a bad config can't 3-hourly retry-loop forever). pub fn requeue_failed(config: &Config) -> Result { with_connection(config, |conn| { let now_ms = Utc::now().timestamp_millis(); @@ -462,6 +462,44 @@ pub fn requeue_failed(config: &Config) -> Result { }) } +/// Requeue only failed jobs whose recorded failure class is NOT +/// `unrecoverable` — i.e. transient failures (network 5xx, timeouts, +/// SQLITE_BUSY) and legacy rows with no class recorded. +/// +/// This is the **automatic** self-healing variant, fired from the +/// 3-hourly queue scheduler so a crash or flaky network never leaves +/// pipeline jobs permanently stuck until the user happens to press +/// "Sync now". Unrecoverable failures (bad/missing key, budget +/// exhausted, dim mismatch) stay parked for the manual +/// `memory_tree_retry_failed` RPC after the user fixes the cause — +/// auto-retrying those would just burn a failure every 3 hours forever. +pub fn requeue_transient_failed(config: &Config) -> Result { + with_connection(config, |conn| { + let now_ms = Utc::now().timestamp_millis(); + let n = conn.execute( + "UPDATE mem_tree_jobs + SET status = 'ready', + attempts = 0, + available_at_ms = ?1, + locked_until_ms = NULL, + started_at_ms = NULL, + completed_at_ms = NULL, + last_error = NULL, + failure_reason = NULL, + failure_class = NULL + WHERE status = 'failed' + AND (failure_class IS NULL OR failure_class != 'unrecoverable')", + params![now_ms], + )?; + if n > 0 { + log::info!( + "[memory::jobs] auto-requeued {n} transient-failed job(s) from the periodic scheduler" + ); + } + Ok(n as u64) + }) +} + /// Release this process's in-flight job locks on a *graceful* shutdown: /// flip every `running` row back to `ready` so the work is immediately /// re-claimable on next launch instead of waiting out the lease and @@ -759,6 +797,67 @@ mod tests { ); } + /// The 3-hourly scheduler's self-heal must requeue jobs that failed + /// transiently (attempts exhausted on network-class errors — untyped + /// `last_error` only) while leaving `unrecoverable`-classified jobs + /// parked for the manual RPC, so a bad embeddings key can't produce a + /// retry-fail loop every 3 hours forever. + #[test] + fn requeue_transient_failed_skips_unrecoverable_jobs() { + use crate::openhuman::memory_tree::health::{FailureCode, PipelineFailure}; + let (_tmp, cfg) = test_config(); + + // Job A: terminal via exhausted retry budget, NO typed class + // (the shape an interrupted network leaves behind). + let mut a = NewJob::extract_chunk(&ExtractChunkPayload { + chunk_id: "a-transient".into(), + }) + .unwrap(); + a.max_attempts = Some(1); + let id_a = enqueue(&cfg, &a).unwrap().unwrap(); + let claim_a = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap(); + mark_failed(&cfg, &claim_a, "connection reset by peer").unwrap(); + assert_eq!( + get_job(&cfg, &id_a).unwrap().unwrap().status, + JobStatus::Failed + ); + + // Job B: terminal with an unrecoverable classification. + let mut b = NewJob::extract_chunk(&ExtractChunkPayload { + chunk_id: "b-unrecoverable".into(), + }) + .unwrap(); + b.max_attempts = Some(1); + let id_b = enqueue(&cfg, &b).unwrap().unwrap(); + let claim_b = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap(); + mark_failed_typed( + &cfg, + &claim_b, + "Insufficient budget", + Some(&PipelineFailure::new(FailureCode::BudgetExhausted)), + ) + .unwrap(); + assert_eq!( + get_job(&cfg, &id_b).unwrap().unwrap().status, + JobStatus::Failed + ); + + let requeued = requeue_transient_failed(&cfg).unwrap(); + assert_eq!(requeued, 1, "only the unclassified failure requeues"); + + let row_a = get_job(&cfg, &id_a).unwrap().unwrap(); + assert_eq!(row_a.status, JobStatus::Ready, "transient job re-runs"); + assert_eq!(row_a.attempts, 0); + + let row_b = get_job(&cfg, &id_b).unwrap().unwrap(); + assert_eq!( + row_b.status, + JobStatus::Failed, + "unrecoverable job stays parked for the manual retry RPC" + ); + assert_eq!(row_b.failure_class.as_deref(), Some("unrecoverable")); + } + #[test] fn mark_failed_typed_transient_still_retries() { use crate::openhuman::memory_tree::health::{FailureCode, PipelineFailure}; diff --git a/src/openhuman/memory_sources/rpc.rs b/src/openhuman/memory_sources/rpc.rs index cc4706ba4..f19e41606 100644 --- a/src/openhuman/memory_sources/rpc.rs +++ b/src/openhuman/memory_sources/rpc.rs @@ -354,6 +354,113 @@ pub async fn sync_rpc(req: SyncRequest) -> Result, Stri )) } +// ── Reconcile ── + +#[derive(Debug, Default, serde::Deserialize)] +pub struct ReconcileRequest { + /// Restrict to one source; omit to inspect every enabled source. + #[serde(default)] + pub source_id: Option, + /// When true, kick off background summarise+ingest for every scope + /// with pending files. When false (default), report-only. + #[serde(default)] + pub execute: bool, +} + +#[derive(Debug, serde::Serialize)] +pub struct ReconcileScopeReport { + pub source_id: String, + pub tree_scope: String, + /// Raw `.md` files on disk for this scope. + pub total_raw_files: usize, + /// Files already covered by a tree summary. + pub covered: usize, + /// Files awaiting summarisation into the tree. + pub pending: usize, + /// True when `execute` was set and a background reconcile was started. + pub started: bool, +} + +#[derive(Debug, serde::Serialize)] +pub struct ReconcileResponse { + pub scopes: Vec, +} + +/// Report (and optionally repair) raw-archive → tree coverage for memory +/// sources. The same incremental reconcile runs automatically after every +/// sync; this RPC exposes it for inspection and manual triggering. +pub async fn reconcile_rpc(req: ReconcileRequest) -> Result, String> { + use crate::openhuman::memory_sources::sync::derive_scopes; + use crate::openhuman::memory_sync::sources::rebuild::{raw_coverage, rebuild_tree_from_raw}; + + tracing::info!( + source_id = ?req.source_id, + execute = req.execute, + "[memory_sources] reconcile_rpc: entry" + ); + + let config = config_rpc::load_config_with_timeout().await?; + let sources: Vec = match &req.source_id { + Some(id) => vec![registry::get_source(id) + .await? + .ok_or_else(|| format!("source '{id}' not found"))?], + None => registry::list_sources().await?, + }; + + let mut reports: Vec = Vec::new(); + for source in sources.iter().filter(|s| s.enabled) { + for scope in derive_scopes(source, &config) { + let coverage = raw_coverage(&config, &scope.tree_scope, &scope.archive_source_id) + .map_err(|e| format!("coverage for {}: {e:#}", scope.tree_scope))?; + let pending = coverage.pending.len(); + let mut started = false; + if req.execute && pending > 0 { + let cfg = config.clone(); + let tree_scope = scope.tree_scope.clone(); + let archive = scope.archive_source_id.clone(); + tokio::spawn(async move { + match rebuild_tree_from_raw(&cfg, &tree_scope, &archive).await { + Ok(outcome) => tracing::info!( + tree_scope = %tree_scope, + files = outcome.files_read, + batches = outcome.batches, + "[memory_sources] reconcile_rpc: background reconcile complete" + ), + Err(e) => tracing::warn!( + tree_scope = %tree_scope, + error = %format!("{e:#}"), + "[memory_sources] reconcile_rpc: background reconcile failed" + ), + } + }); + started = true; + } + tracing::debug!( + source_id = %source.id, + tree_scope = %scope.tree_scope, + total = coverage.total, + covered = coverage.covered, + pending = pending, + started = started, + "[memory_sources] reconcile_rpc: scope report" + ); + reports.push(ReconcileScopeReport { + source_id: source.id.clone(), + tree_scope: scope.tree_scope, + total_raw_files: coverage.total, + covered: coverage.covered, + pending, + started, + }); + } + } + + Ok(RpcOutcome::new( + ReconcileResponse { scopes: reports }, + vec![], + )) +} + // ── Status List ── #[derive(Debug, serde::Serialize)] diff --git a/src/openhuman/memory_sources/schemas.rs b/src/openhuman/memory_sources/schemas.rs index 6b218aba3..176a733c2 100644 --- a/src/openhuman/memory_sources/schemas.rs +++ b/src/openhuman/memory_sources/schemas.rs @@ -128,6 +128,7 @@ pub fn all_controller_schemas() -> Vec { schemas("list_items"), schemas("read_item"), schemas("sync"), + schemas("reconcile"), schemas("status_list"), schemas("supported_toolkits"), schemas("sync_audit_log"), @@ -171,6 +172,10 @@ pub fn all_registered_controllers() -> Vec { schema: schemas("sync"), handler: handle_sync, }, + RegisteredController { + schema: schemas("reconcile"), + handler: handle_reconcile, + }, RegisteredController { schema: schemas("status_list"), handler: handle_status_list, @@ -395,6 +400,35 @@ pub fn schemas(function: &str) -> ControllerSchema { }, ], }, + "reconcile" => ControllerSchema { + namespace: NAMESPACE, + function: "reconcile", + description: "Report raw-archive vs memory-tree coverage per source scope; \ + with execute=true, start a background incremental reconcile \ + (summarise + ingest) for every scope with pending files. The \ + same reconcile also runs automatically after each sync.", + inputs: vec![ + FieldSchema { + name: "source_id", + ty: TypeSchema::String, + comment: "Restrict to one source id; omit for all enabled sources.", + required: false, + }, + FieldSchema { + name: "execute", + ty: TypeSchema::Bool, + comment: "Start background reconcile for scopes with pending files \ + (default false = report only).", + required: false, + }, + ], + outputs: vec![FieldSchema { + name: "scopes", + ty: TypeSchema::Array(Box::new(TypeSchema::Ref("ReconcileScopeReport"))), + comment: "Per-scope coverage: total raw files, covered, pending, started.", + required: true, + }], + }, "status_list" => ControllerSchema { namespace: NAMESPACE, function: "status_list", @@ -609,6 +643,13 @@ fn handle_sync(params: Map) -> ControllerFuture { }) } +fn handle_reconcile(params: Map) -> ControllerFuture { + Box::pin(async move { + let req = parse_value::(Value::Object(params))?; + to_json(rpc::reconcile_rpc(req).await?) + }) +} + fn handle_status_list(_params: Map) -> ControllerFuture { Box::pin(async move { to_json(rpc::status_list_rpc().await?) }) } diff --git a/src/openhuman/memory_sources/sync.rs b/src/openhuman/memory_sources/sync.rs index 4d43dc218..377b9765a 100644 --- a/src/openhuman/memory_sources/sync.rs +++ b/src/openhuman/memory_sources/sync.rs @@ -416,24 +416,27 @@ async fn sync_items_individually( Ok(ingested.load(Ordering::Relaxed)) } -/// Derive the tree scope(s) for a source and rebuild from raw if needed. -async fn check_and_rebuild_tree(source: &MemorySourceEntry, config: &Config) { +/// Derive the tree scope(s) for a source and reconcile any raw files that +/// are not yet covered by tree summaries (incremental — see +/// `memory_sync::sources::rebuild`). +pub(crate) async fn check_and_rebuild_tree(source: &MemorySourceEntry, config: &Config) { use crate::openhuman::memory_sync::sources::rebuild::{needs_rebuild, rebuild_tree_from_raw}; let scopes = derive_scopes(source, config); for scope in scopes { - if !needs_rebuild(config, &scope) { + if !needs_rebuild(config, &scope.tree_scope, &scope.archive_source_id) { continue; } tracing::info!( source_id = %source.id, - scope = %scope, - "[memory_sources:sync] auto-rebuilding tree from raw" + scope = %scope.tree_scope, + archive = %scope.archive_source_id, + "[memory_sources:sync] reconciling uncovered raw files into tree" ); - match rebuild_tree_from_raw(config, &scope).await { + match rebuild_tree_from_raw(config, &scope.tree_scope, &scope.archive_source_id).await { Ok(outcome) => { tracing::info!( - scope = %scope, + scope = %scope.tree_scope, files = outcome.files_read, batches = outcome.batches, cost = %format!( @@ -441,39 +444,58 @@ async fn check_and_rebuild_tree(source: &MemorySourceEntry, config: &Config) { outcome.actual_charged_usd.unwrap_or(outcome.estimated_cost_usd) ), cost_is_actual = outcome.actual_charged_usd.is_some(), - "[memory_sources:sync] rebuild complete" + "[memory_sources:sync] reconcile complete" ); } Err(e) => { tracing::warn!( - scope = %scope, + scope = %scope.tree_scope, error = %format!("{e:#}"), - "[memory_sources:sync] rebuild failed" + "[memory_sources:sync] reconcile failed" ); } } } } -/// Derive the tree scope string(s) that a source maps to. -fn derive_scopes(source: &MemorySourceEntry, config: &Config) -> Vec { +/// A source's tree scope paired with its raw-archive source id. The two +/// slugify to DIFFERENT directories for GitHub (`github:owner/repo` vs +/// `github.com/owner/repo`) — conflating them makes reconcile scan an +/// empty directory while the real archive sits uncovered. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct SourceScope { + /// Tree registry key, e.g. `"github:owner/repo"`. + pub tree_scope: String, + /// Raw-archive id whose slug names `raw//`, e.g. + /// `"github.com/owner/repo"`. Equal to `tree_scope` for sources that + /// archive under their scope (gmail). + pub archive_source_id: String, +} + +/// Derive the tree scope(s) + raw-archive id(s) that a source maps to. +pub(crate) fn derive_scopes(source: &MemorySourceEntry, config: &Config) -> Vec { use crate::openhuman::memory_sources::readers::github; - use crate::openhuman::memory_store::content::raw::slug_account_email; match source.kind { SourceKind::GithubRepo => { - // GitHub sync already builds its own tree — but check anyway. - source - .url - .as_deref() - .and_then(github::repo_chunk_scope) - .into_iter() - .collect() + let Some(url) = source.url.as_deref() else { + return Vec::new(); + }; + match ( + github::repo_chunk_scope(url), + github::repo_archive_source_id(url), + ) { + (Some(tree_scope), Some(archive_source_id)) => vec![SourceScope { + tree_scope, + archive_source_id, + }], + _ => Vec::new(), + } } SourceKind::Composio => { // Composio sources scope by toolkit + connection email. - // Gmail: "gmail:" - // Others: "composio::" + // Gmail: "gmail:" — archive dir shares + // the scope. Others: no raw archive to reconcile yet. let toolkit = source.toolkit.as_deref().unwrap_or("unknown"); match toolkit { "gmail" | "GMAIL" => { @@ -495,10 +517,15 @@ fn derive_scopes(source: &MemorySourceEntry, config: &Config) -> Vec { let source_md = e.path().join("_source.md"); let content = std::fs::read_to_string(&source_md).ok()?; content.lines().find(|l| l.starts_with("scope:")).map(|l| { - l.trim_start_matches("scope:") + let scope = l + .trim_start_matches("scope:") .trim() .trim_matches('"') - .to_string() + .to_string(); + SourceScope { + tree_scope: scope.clone(), + archive_source_id: scope, + } }) }) .collect() diff --git a/src/openhuman/memory_store/chunks/raw_refs.rs b/src/openhuman/memory_store/chunks/raw_refs.rs index 1eccdcb8a..f6619a227 100644 --- a/src/openhuman/memory_store/chunks/raw_refs.rs +++ b/src/openhuman/memory_store/chunks/raw_refs.rs @@ -80,6 +80,49 @@ pub fn get_chunk_raw_refs(config: &Config, chunk_id: &str) -> Result Result> { + with_connection(config, |conn| { + let mut stmt = conn.prepare( + "SELECT raw_refs_json FROM mem_tree_chunks \ + WHERE raw_refs_json IS NOT NULL AND raw_refs_json != ''", + )?; + let rows = stmt.query_map([], |r| r.get::<_, String>(0))?; + let mut out: std::collections::HashSet = std::collections::HashSet::new(); + for row in rows { + let json = row?; + // Tolerate individually-corrupt rows: skip with a warning + // rather than failing the whole coverage scan. + match serde_json::from_str::>(&json) { + Ok(refs) => { + for raw_ref in refs { + if raw_ref.path.starts_with(rel_prefix) { + out.insert(raw_ref.path); + } + } + } + Err(e) => { + log::warn!( + "[memory::chunk_store] skipping unparseable raw_refs_json during \ + coverage scan: {e}" + ); + } + } + } + Ok(out) + }) +} + /// Return both `content_path` and `content_sha256` stored in SQLite for `chunk_id`. /// /// Returns `Ok(None)` if the chunk does not exist or has no content_path recorded yet. diff --git a/src/openhuman/memory_store/chunks/store.rs b/src/openhuman/memory_store/chunks/store.rs index 5f19fdf32..06596a5ca 100644 --- a/src/openhuman/memory_store/chunks/store.rs +++ b/src/openhuman/memory_store/chunks/store.rs @@ -833,6 +833,85 @@ pub(crate) fn claim_source_ingest_tx( Ok(inserted > 0) } +/// `source_kind` value used in `mem_tree_ingested_sources` to record that a +/// raw archive file (relative path under `/`, e.g. +/// `raw/github-com-org-repo/commits/_.md`) has been covered by a +/// tree summary. Distinct from the chunk-store [`SourceKind`] values so the +/// two gate namespaces can never collide. +pub const RAW_FILE_GATE_KIND: &str = "raw_file"; + +/// Record that the given raw archive files (relative paths under +/// `/`) are covered by a tree summary. Idempotent +/// (`INSERT OR IGNORE`); returns the number of newly-recorded paths. +pub fn mark_raw_paths_ingested(config: &Config, rel_paths: &[String]) -> Result { + if rel_paths.is_empty() { + return Ok(0); + } + let now_ms = Utc::now().timestamp_millis(); + with_connection(config, |conn| { + let tx = conn.unchecked_transaction()?; + let mut inserted: u64 = 0; + { + let mut stmt = tx.prepare( + "INSERT OR IGNORE INTO mem_tree_ingested_sources \ + (source_kind, source_id, ingested_at_ms) \ + VALUES (?1, ?2, ?3)", + )?; + for path in rel_paths { + inserted += stmt.execute(params![RAW_FILE_GATE_KIND, path, now_ms])? as u64; + } + } + tx.commit()?; + log::debug!( + "[memory::chunk_store] mark_raw_paths_ingested: {} given, {} newly recorded", + rel_paths.len(), + inserted + ); + Ok(inserted) + }) +} + +/// Filter `rel_paths` down to the ones NOT yet recorded as ingested raw +/// files. Order of the surviving paths is preserved. +pub fn filter_raw_paths_not_ingested(config: &Config, rel_paths: &[String]) -> Result> { + if rel_paths.is_empty() { + return Ok(Vec::new()); + } + with_connection(config, |conn| { + let mut stmt = conn.prepare( + "SELECT COUNT(*) FROM mem_tree_ingested_sources \ + WHERE source_kind = ?1 AND source_id = ?2", + )?; + let mut out: Vec = Vec::new(); + for path in rel_paths { + let n: i64 = stmt.query_row(params![RAW_FILE_GATE_KIND, path], |r| r.get(0))?; + if n == 0 { + out.push(path.clone()); + } + } + Ok(out) + }) +} + +/// Count raw-file gate rows whose path starts with `rel_prefix` (e.g. +/// `raw/github-com-org-repo/`). Diagnostic helper for reconcile reporting. +pub fn count_raw_paths_ingested_with_prefix(config: &Config, rel_prefix: &str) -> Result { + with_connection(config, |conn| { + // Rust-side prefix filter (not SQL LIKE) so `_` / `%` in slugs are + // treated literally — same convention as delete_chunks_by_source_prefix. + let mut stmt = + conn.prepare("SELECT source_id FROM mem_tree_ingested_sources WHERE source_kind = ?1")?; + let rows = stmt.query_map(params![RAW_FILE_GATE_KIND], |r| r.get::<_, String>(0))?; + let mut n: u64 = 0; + for row in rows { + if row?.starts_with(rel_prefix) { + n += 1; + } + } + Ok(n) + }) +} + /// Delete all chunk rows for one exact `(source_kind, source_id)` and clear /// dependent source-local indexes. Returns the number of chunk rows removed. pub fn delete_chunks_by_source( @@ -1186,8 +1265,8 @@ use migrations::{migrate_legacy_embeddings_to_sidecar, purge_global_topic_trees} mod raw_refs; pub use raw_refs::{ get_chunk_content_path, get_chunk_content_pointers, get_chunk_raw_refs, - get_summary_content_pointers, list_summaries_with_content_path, set_chunk_raw_refs, - set_chunk_raw_refs_tx, RawRef, + get_summary_content_pointers, list_chunk_raw_ref_paths_with_prefix, + list_summaries_with_content_path, set_chunk_raw_refs, set_chunk_raw_refs_tx, RawRef, }; fn normalized_limit(requested: Option) -> i64 { diff --git a/src/openhuman/memory_store/content/raw.rs b/src/openhuman/memory_store/content/raw.rs index c135dd59a..e8b230b68 100644 --- a/src/openhuman/memory_store/content/raw.rs +++ b/src/openhuman/memory_store/content/raw.rs @@ -157,8 +157,10 @@ fn build_filename(created_at_ms: i64, uid: &str) -> String { /// Replace path-illegal characters in the upstream uid before splicing /// it into a filename. Mirrors `paths::sanitize_filename` but is local -/// so a future change to either side stays decoupled. -fn sanitize_uid(uid: &str) -> String { +/// so a future change to either side stays decoupled. `pub(crate)` so +/// the raw-coverage backfill (`memory_sync::sources::rebuild`) can map +/// summary child labels back to on-disk filename suffixes. +pub(crate) fn sanitize_uid(uid: &str) -> String { let cleaned: String = uid .chars() .map(|c| match c { diff --git a/src/openhuman/memory_sync/README.md b/src/openhuman/memory_sync/README.md index 0e1930bdb..8825b9842 100644 --- a/src/openhuman/memory_sync/README.md +++ b/src/openhuman/memory_sync/README.md @@ -36,9 +36,32 @@ and retry policy behind that. | [`workspace/`](workspace/) | Vault, harness, dictation pipelines. | | [`mcp/`](mcp/) | MCP-server pipelines (one per connected server). | +## Schedulers & self-healing + +Three background loops keep memory synced and compressed without manual +"Sync now" clicks: + +| Loop | Module | Cadence | Covers | +| --- | --- | --- | --- | +| Composio periodic | [`composio/periodic.rs`](composio/periodic.rs) | 20-min tick, per-connection interval | Composio connections (gmail, slack, …) | +| Workspace periodic | [`workspace/periodic.rs`](workspace/periodic.rs) | 20-min tick, `memory_sync_interval_secs` (24h default, `0` = manual only) | `github_repo`, `folder`, `rss_feed`, `web_page` sources | +| Queue scheduler | `memory_queue::scheduler` | 3 h | flush stale L0 buffers + requeue transient-failed jobs | + +**Raw-archive coverage** ([`sources/rebuild.rs`](sources/rebuild.rs)): every +summary batch records its raw files in `mem_tree_ingested_sources` +(`source_kind = "raw_file"`). After each sync, `check_and_rebuild_tree` +diffs disk vs gate and summarises only the uncovered remainder — so an +interrupted sync self-heals on the next pass instead of stranding raw +files forever. Inspect/trigger over RPC with +`openhuman.memory_sources_reconcile`. + +⚠️ A source's **tree scope** and **raw-archive id** slugify to different +directories for GitHub (`github:owner/repo` vs `github.com/owner/repo`) — +always carry both (`memory_sources::sync::SourceScope`). + ## Status -**Scaffold only.** Today's sync code still lives in: +**Pipelines: scaffold only.** Today's sync code still lives in: - `composio/providers//ingest.rs` + `bin/{slack_backfill,gmail_backfill_3d}.rs` - `vault/sync.rs`, `agent_experience/`, `dictation_hotkeys/` diff --git a/src/openhuman/memory_sync/composio/periodic.rs b/src/openhuman/memory_sync/composio/periodic.rs index 34f66465b..86b5bc09e 100644 --- a/src/openhuman/memory_sync/composio/periodic.rs +++ b/src/openhuman/memory_sync/composio/periodic.rs @@ -127,7 +127,7 @@ pub fn record_sync_success(toolkit: &str, connection_id: &str) { /// - `global == None` → `Some(max(DEFAULT, provider_default))`: no explicit /// user choice, so fall back to the 24h default cadence (also floored at the /// provider default). -fn effective_interval_secs(provider_default: u64, global: Option) -> Option { +pub(crate) fn effective_interval_secs(provider_default: u64, global: Option) -> Option { match global { Some(0) => None, Some(n) => Some(n.max(provider_default)), @@ -141,7 +141,7 @@ fn effective_interval_secs(provider_default: u64, global: Option) -> Option /// `since_last_sync == None` means we have no record of a sync this process /// lifetime, so we fire immediately (the restart-recovery path). Kept pure so /// the due-check can be simulated without driving the real `Instant` clock. -fn connection_is_due(interval_secs: u64, since_last_sync: Option) -> bool { +pub(crate) fn connection_is_due(interval_secs: u64, since_last_sync: Option) -> bool { match since_last_sync { Some(elapsed) => elapsed >= Duration::from_secs(interval_secs), None => true, @@ -336,7 +336,7 @@ async fn run_loop() { /// transitional / not-yet-resolved condition. Letting the tick proceed /// here keeps periodic sync running through brief transitions instead of /// pausing on stale unresolved state. -fn periodic_pause_reason() -> Option { +pub(crate) fn periodic_pause_reason() -> Option { // Delegate the `Policy::Paused { .. }` → `PauseReason` extraction to // the existing `Policy::pause_reason()` helper (avoids re-implementing // the same destructure twice). The allow-list below is the only thing diff --git a/src/openhuman/memory_sync/sources/github.rs b/src/openhuman/memory_sync/sources/github.rs index 828055aca..6b4d8452b 100644 --- a/src/openhuman/memory_sync/sources/github.rs +++ b/src/openhuman/memory_sync/sources/github.rs @@ -225,15 +225,45 @@ pub async fn run_github_sync( child_basenames: batch_basenames, }; + // `child_basenames` holds the raw-archive wikilink paths (rel path + // with `.md` stripped) — captured BEFORE ingest_summary moves the + // input. Re-suffix to recover the coverage-gate keys. + let batch_raw_rel_paths: Vec = ingest_input + .child_basenames + .iter() + .flatten() + .map(|basename| format!("{basename}.md")) + .collect(); + let outcome = ingest_summary(config, &tree, ingest_input).await?; + // Record raw-archive coverage so the incremental reconcile + // (`memory_sync::sources::rebuild`) knows these files are + // summarised. Marked only after the summary landed: a crash in + // between re-summarises the batch (duplicate summary, acceptable) + // instead of silently losing coverage. + if !batch_raw_rel_paths.is_empty() { + if let Err(e) = crate::openhuman::memory_store::chunks::store::mark_raw_paths_ingested( + config, + &batch_raw_rel_paths, + ) { + tracing::warn!( + source_id = %source_id, + batch = batch_idx, + error = %format!("{e:#}"), + "[memory_sync:github] failed to record raw coverage — reconcile may re-summarise this batch" + ); + } + } + tracing::info!( source_id = %source_id, batch = batch_idx, summary_id = %outcome.summary_id, path = %outcome.content_path, sealed = outcome.sealed_ids.len(), - "[memory_sync:github] batch ingested" + covered_raw_files = batch_raw_rel_paths.len(), + "[memory_sync:github] batch ingested + coverage recorded" ); } diff --git a/src/openhuman/memory_sync/sources/rebuild.rs b/src/openhuman/memory_sync/sources/rebuild.rs index db58cee55..456073efd 100644 --- a/src/openhuman/memory_sync/sources/rebuild.rs +++ b/src/openhuman/memory_sync/sources/rebuild.rs @@ -1,23 +1,49 @@ -//! Rebuild tree from raw archive files on disk. +//! Reconcile raw archive files on disk into the memory tree. //! -//! When a sync has ingested raw files but never built summaries (e.g. -//! interrupted sync, or legacy data), this module reads all `.md` files -//! from `raw///`, batches them into ~50k-token -//! groups, summarises each batch, and ingests into the tree via -//! `ingest_summary`. +//! Raw archive files are written eagerly at fetch time (one `.md` per +//! upstream item under `raw///`), but summaries only +//! land at the end of a sync's batch loop — so an interrupted sync, a +//! crash mid-summarise, or legacy data leaves raw files on disk with no +//! tree coverage. This module closes that gap **incrementally**: //! -//! Idempotent: re-running produces new L1 summaries (each has a unique -//! id), so callers should check whether the tree already has L1 nodes -//! before triggering. +//! 1. Every successful summary-batch ingest records its raw files in the +//! `mem_tree_ingested_sources` coverage gate +//! (`source_kind = "raw_file"`, `source_id = `). +//! 2. [`raw_coverage`] lists the on-disk files NOT in the gate. For +//! scopes whose summaries predate the gate (legacy data), it first +//! backfills the gate from the existing L1 summaries' child labels so +//! already-covered files are never re-summarised. +//! 3. [`rebuild_tree_from_raw`] reads only the pending files, batches +//! them into ~50k-token groups, summarises each batch, ingests via +//! `ingest_summary`, and marks the batch's files covered. +//! +//! Idempotent: re-running with full coverage is a no-op; a run that dies +//! mid-way resumes from the first unmarked batch. +//! +//! ## Tree scope vs archive id +//! +//! A source has TWO identifiers that slugify to *different* directories: +//! the tree scope (e.g. `github:owner/repo` → tree registry key) and the +//! raw-archive source id (e.g. `github.com/owner/repo` → +//! `raw/github-com-owner-repo/`). Callers must pass both — deriving the +//! raw dir from the tree scope silently scans the wrong (usually empty) +//! directory, which is exactly the bug that let thousands of GitHub raw +//! files sit unreconciled. +use std::collections::HashSet; use std::path::{Path, PathBuf}; use anyhow::Result; use crate::openhuman::config::Config; use crate::openhuman::memory::tree_source::get_or_create_source_tree; +use crate::openhuman::memory_store::chunks::store::{ + count_raw_paths_ingested_with_prefix, filter_raw_paths_not_ingested, + list_chunk_raw_ref_paths_with_prefix, mark_raw_paths_ingested, +}; use crate::openhuman::memory_store::content::paths::slugify_source_id; -use crate::openhuman::memory_store::content::raw::raw_source_dir; +use crate::openhuman::memory_store::content::raw::{raw_source_dir, sanitize_uid}; +use crate::openhuman::memory_store::trees::store::list_summaries_at_level; use crate::openhuman::memory_store::trees::types::{TreeKind, INPUT_TOKEN_BUDGET}; use crate::openhuman::memory_sync::sources::audit::{ append_audit_entry, RealCostAccumulator, SyncAuditEntry, @@ -41,80 +67,240 @@ pub struct RebuildOutcome { pub actual_charged_usd: Option, } -/// Check whether a source needs tree rebuilding: raw files exist on disk -/// but the tree has no L1+ summaries (max_level == 0). -pub fn needs_rebuild(config: &Config, scope: &str) -> bool { +/// One raw archive file pending tree coverage. +#[derive(Clone, Debug)] +pub struct RawFileRef { + /// Absolute path on disk. + pub abs: PathBuf, + /// Forward-slash relative path under `/`, with `.md` — + /// the canonical coverage-gate key (matches `raw::raw_rel_path`). + pub rel: String, +} + +/// Coverage report for one source's raw archive. +#[derive(Clone, Debug, Default)] +pub struct RawCoverage { + /// Total `.md` files on disk (excluding `_source.md` etc.). + pub total: usize, + /// Files recorded in the coverage gate. + pub covered: usize, + /// Files on disk with no coverage record, sorted chronologically + /// (filenames start with the item timestamp). + pub pending: Vec, +} + +/// Compute raw-archive coverage for a source. +/// +/// `tree_scope` keys the source tree (e.g. `"github:owner/repo"`, +/// `"gmail:user-at-gmail-dot-com"`); `archive_source_id` keys the raw +/// archive directory (e.g. `"github.com/owner/repo"` — pass the tree +/// scope again when the source writes its archive under the same id, +/// as gmail does). +/// +/// On first call for a scope with existing L1 summaries but an empty +/// gate (legacy data from before coverage tracking), the gate is +/// backfilled from the summaries' child labels so previously-summarised +/// files don't get re-ingested. +pub fn raw_coverage( + config: &Config, + tree_scope: &str, + archive_source_id: &str, +) -> Result { let content_root = config.memory_tree_content_root(); - let source_dir = raw_source_dir(&content_root, scope); + let source_dir = raw_source_dir(&content_root, archive_source_id); if !source_dir.exists() { - return false; + return Ok(RawCoverage::default()); } - let tree = match get_or_create_source_tree(config, scope) { - Ok(t) => t, - Err(_) => return false, - }; - - // Tree has no sealed summaries — raw files are sitting un-summarised. - if tree.max_level > 0 { - return false; + let mut files = collect_raw_files(&source_dir)?; + files.sort(); + if files.is_empty() { + return Ok(RawCoverage::default()); } - // Check there's actually raw content (not just `_source.md`). - match collect_raw_files(&source_dir) { - Ok(files) => !files.is_empty(), - Err(_) => false, + let refs: Vec = files + .into_iter() + .filter_map(|abs| { + let rel = abs + .strip_prefix(&content_root) + .ok()? + .to_str()? + .replace(std::path::MAIN_SEPARATOR, "/"); + Some(RawFileRef { abs, rel }) + }) + .collect(); + let total = refs.len(); + + // Legacy backfill: gate empty for this archive but the tree already + // has L1 summaries → mark the files those summaries cover. + let rel_prefix = format!("raw/{}/", slugify_source_id(archive_source_id)); + if count_raw_paths_ingested_with_prefix(config, &rel_prefix)? == 0 { + let backfilled = backfill_coverage_from_summaries(config, tree_scope, &refs)?; + if backfilled > 0 { + tracing::info!( + tree_scope = %tree_scope, + archive = %archive_source_id, + backfilled = backfilled, + "[memory_sync:rebuild] backfilled coverage gate from existing L1 summaries" + ); + } + } + + let rel_paths: Vec = refs.iter().map(|r| r.rel.clone()).collect(); + let mut pending_rels: HashSet = filter_raw_paths_not_ingested(config, &rel_paths)? + .into_iter() + .collect(); + + // A raw file referenced by a persisted chunk (`raw_refs_json`) is + // already in the tree via the chunk pipeline — gmail mirrors every + // email to raw/ AND ingests it as a chunk. Those files are covered; + // re-summarising them through the rebuild path would duplicate + // content (and burn LLM batches) for sources that never use the + // summarise-direct path at all. + let chunk_covered = list_chunk_raw_ref_paths_with_prefix(config, &rel_prefix)?; + if !chunk_covered.is_empty() { + pending_rels.retain(|rel| !chunk_covered.contains(rel)); + } + + let pending: Vec = refs + .into_iter() + .filter(|r| pending_rels.contains(&r.rel)) + .collect(); + + tracing::debug!( + tree_scope = %tree_scope, + archive = %archive_source_id, + total = total, + pending = pending.len(), + "[memory_sync:rebuild] raw coverage computed" + ); + + Ok(RawCoverage { + total, + covered: total - pending.len(), + pending, + }) +} + +/// Check whether a source has raw files on disk that the tree does not +/// cover yet. Coverage-based: a partially-covered scope (interrupted +/// sync) returns `true` even when the tree already has L1 summaries. +pub fn needs_rebuild(config: &Config, tree_scope: &str, archive_source_id: &str) -> bool { + match raw_coverage(config, tree_scope, archive_source_id) { + Ok(cov) => !cov.pending.is_empty(), + Err(e) => { + tracing::warn!( + tree_scope = %tree_scope, + archive = %archive_source_id, + error = %format!("{e:#}"), + "[memory_sync:rebuild] coverage check failed — skipping rebuild" + ); + false + } } } -/// Rebuild the tree for a given source scope by reading all raw `.md` -/// files from disk. `scope` is the tree scope (e.g. -/// `"gmail:stevent95-at-gmail-dot-com"`). +/// Mark a label set covered in the gate from existing L1 summaries. /// -/// The raw archive lives at `/raw//`. -pub async fn rebuild_tree_from_raw(config: &Config, scope: &str) -> Result { +/// Sync-written summaries label children as `commit:` / `issue:` / +/// `pr:`; rebuild-written summaries label them with the raw file stem +/// (`_`). Both shapes are matched against the on-disk files. +/// Returns the number of files marked covered. +fn backfill_coverage_from_summaries( + config: &Config, + tree_scope: &str, + files: &[RawFileRef], +) -> Result { + let tree = get_or_create_source_tree(config, tree_scope) + .map_err(|e| anyhow::anyhow!("get_or_create_source_tree: {e:#}"))?; + if tree.max_level == 0 { + return Ok(0); + } + + let summaries = list_summaries_at_level(config, &tree.id, 1)?; + // (kind_dir, sanitised uid) pairs from `:` labels. + let mut kind_uids: HashSet<(&'static str, String)> = HashSet::new(); + // Full file stems from rebuild-written labels. + let mut stems: HashSet = HashSet::new(); + for summary in &summaries { + for label in &summary.child_ids { + if let Some(uid) = label.strip_prefix("commit:") { + kind_uids.insert(("commits", sanitize_uid(uid))); + } else if let Some(uid) = label.strip_prefix("issue:") { + kind_uids.insert(("issues", sanitize_uid(uid))); + } else if let Some(uid) = label.strip_prefix("pr:") { + kind_uids.insert(("prs", sanitize_uid(uid))); + } else { + stems.insert(label.clone()); + } + } + } + if kind_uids.is_empty() && stems.is_empty() { + return Ok(0); + } + + let mut covered: Vec = Vec::new(); + for f in files { + // rel = raw///_.md + let mut parts = f.rel.rsplitn(2, '/'); + let filename = parts.next().unwrap_or_default(); + let kind_dir = parts + .next() + .and_then(|p| p.rsplit('/').next()) + .unwrap_or_default(); + let stem = filename.strip_suffix(".md").unwrap_or(filename); + let uid = stem.split_once('_').map(|(_, u)| u).unwrap_or(stem); + + let label_match = + stems.contains(stem) || kind_uids.iter().any(|(k, u)| *k == kind_dir && u == uid); + if label_match { + covered.push(f.rel.clone()); + } + } + + mark_raw_paths_ingested(config, &covered) +} + +/// Summarise + ingest every **pending** raw file for a source. `tree_scope` +/// and `archive_source_id` as in [`raw_coverage`]. +/// +/// Each batch's files are marked covered immediately after that batch's +/// summary lands, so an interrupted run resumes where it left off. +pub async fn rebuild_tree_from_raw( + config: &Config, + tree_scope: &str, + archive_source_id: &str, +) -> Result { let start = std::time::Instant::now(); let content_root = config.memory_tree_content_root(); - let source_dir = raw_source_dir(&content_root, scope); + let coverage = raw_coverage(config, tree_scope, archive_source_id)?; tracing::info!( - scope = %scope, - dir = %source_dir.display(), - "[memory_sync:rebuild] starting rebuild from raw" + tree_scope = %tree_scope, + archive = %archive_source_id, + total = coverage.total, + covered = coverage.covered, + pending = coverage.pending.len(), + "[memory_sync:rebuild] starting incremental rebuild" ); - if !source_dir.exists() { - anyhow::bail!( - "raw source directory does not exist: {}", - source_dir.display() - ); - } - - // Collect all .md files recursively (skip _source.md). - let mut files = collect_raw_files(&source_dir)?; - files.sort(); // chronological order (filename starts with timestamp) - - if files.is_empty() { + if coverage.pending.is_empty() { return Ok(RebuildOutcome::default()); } + let files = coverage.pending; - tracing::info!( - scope = %scope, - files = files.len(), - "[memory_sync:rebuild] found raw files" - ); - - // Read all files into SummaryInputs. + // Read pending files into SummaryInputs. let mut inputs: Vec = Vec::with_capacity(files.len()); let mut basenames: Vec> = Vec::with_capacity(files.len()); let mut labels: Vec = Vec::with_capacity(files.len()); + let mut rel_paths: Vec = Vec::with_capacity(files.len()); - for path in &files { - let body = match std::fs::read_to_string(path) { + for file in &files { + let body = match std::fs::read_to_string(&file.abs) { Ok(b) => b, Err(e) => { tracing::warn!( - path = %path.display(), + path = %file.abs.display(), error = %e, "[memory_sync:rebuild] skipping unreadable file" ); @@ -122,7 +308,8 @@ pub async fn rebuild_tree_from_raw(config: &Config, scope: &str) -> Result Result Result Result Result Result Result Result) -> Result<()> { Ok(()) } +/// One summarise-and-ingest batch with its parallel provenance vectors. +struct RebuildBatch { + inputs: Vec, + labels: Vec, + basenames: Vec>, + rel_paths: Vec, +} + fn batch_inputs( inputs: &[SummaryInput], labels: &[String], basenames: &[Option], + rel_paths: &[String], budget: u32, -) -> Vec<(Vec, Vec, Vec>)> { - let mut batches = Vec::new(); - let mut cur_inputs: Vec = Vec::new(); - let mut cur_labels: Vec = Vec::new(); - let mut cur_basenames: Vec> = Vec::new(); +) -> Vec { + let mut batches: Vec = Vec::new(); + let mut cur = RebuildBatch { + inputs: Vec::new(), + labels: Vec::new(), + basenames: Vec::new(), + rel_paths: Vec::new(), + }; let mut cur_tokens: u32 = 0; - for ((input, label), basename) in inputs.iter().zip(labels.iter()).zip(basenames.iter()) { - if !cur_inputs.is_empty() && cur_tokens + input.token_count > budget { - batches.push(( - std::mem::take(&mut cur_inputs), - std::mem::take(&mut cur_labels), - std::mem::take(&mut cur_basenames), + for i in 0..inputs.len() { + if !cur.inputs.is_empty() && cur_tokens + inputs[i].token_count > budget { + batches.push(std::mem::replace( + &mut cur, + RebuildBatch { + inputs: Vec::new(), + labels: Vec::new(), + basenames: Vec::new(), + rel_paths: Vec::new(), + }, )); cur_tokens = 0; } - cur_tokens += input.token_count; - cur_inputs.push(input.clone()); - cur_labels.push(label.clone()); - cur_basenames.push(basename.clone()); + cur_tokens += inputs[i].token_count; + cur.inputs.push(inputs[i].clone()); + cur.labels.push(labels[i].clone()); + cur.basenames.push(basenames[i].clone()); + cur.rel_paths.push(rel_paths[i].clone()); } - if !cur_inputs.is_empty() { - batches.push((cur_inputs, cur_labels, cur_basenames)); + if !cur.inputs.is_empty() { + batches.push(cur); } batches @@ -371,6 +593,28 @@ fn batch_inputs( #[cfg(test)] mod tests { use super::*; + use crate::openhuman::memory::chat::{test_override, ChatProvider, StaticChatProvider}; + use std::sync::Arc; + use tempfile::TempDir; + + fn test_config(tmp: &TempDir) -> Config { + let mut cfg = Config::default(); + cfg.workspace_dir = tmp.path().to_path_buf(); + cfg.memory_tree.embedding_endpoint = None; + cfg.memory_tree.embedding_model = None; + cfg.memory_tree.embedding_strict = false; + cfg + } + + fn write_raw(cfg: &Config, archive_id: &str, kind: &str, name: &str, body: &str) { + let dir = cfg + .memory_tree_content_root() + .join("raw") + .join(slugify_source_id(archive_id)) + .join(kind); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write(dir.join(name), body).unwrap(); + } #[test] fn collect_raw_files_skips_underscore_files() { @@ -399,4 +643,193 @@ mod tests { let files = collect_raw_files(tmp.path()).unwrap(); assert_eq!(files.len(), 2); } + + #[test] + fn batch_inputs_carries_rel_paths_alongside_provenance() { + let make = |tokens: u32, id: &str| SummaryInput { + id: id.to_string(), + content: String::new(), + token_count: tokens, + entities: Vec::new(), + topics: Vec::new(), + time_range_start: chrono::Utc::now(), + time_range_end: chrono::Utc::now(), + score: 0.5, + }; + let inputs = vec![make(30_000, "a"), make(30_000, "b"), make(30_000, "c")]; + let labels: Vec = vec!["a".into(), "b".into(), "c".into()]; + let basenames: Vec> = vec![None, None, None]; + let rels: Vec = vec![ + "raw/x/a.md".into(), + "raw/x/b.md".into(), + "raw/x/c.md".into(), + ]; + + let batches = batch_inputs(&inputs, &labels, &basenames, &rels, 50_000); + assert_eq!(batches.len(), 3); + assert_eq!(batches[0].rel_paths, vec!["raw/x/a.md".to_string()]); + assert_eq!(batches[2].labels, vec!["c".to_string()]); + } + + #[test] + fn raw_coverage_reports_all_pending_for_untracked_archive() { + let tmp = TempDir::new().unwrap(); + let cfg = test_config(&tmp); + write_raw(&cfg, "github.com/o/r", "commits", "100_aaa.md", "a"); + write_raw(&cfg, "github.com/o/r", "issues", "200_7.md", "b"); + + let cov = raw_coverage(&cfg, "github:o/r", "github.com/o/r").unwrap(); + assert_eq!(cov.total, 2); + assert_eq!(cov.covered, 0); + assert_eq!(cov.pending.len(), 2); + assert!(cov.pending[0].rel.starts_with("raw/github-com-o-r/")); + assert!(needs_rebuild(&cfg, "github:o/r", "github.com/o/r")); + } + + #[test] + fn raw_coverage_empty_when_archive_dir_missing() { + let tmp = TempDir::new().unwrap(); + let cfg = test_config(&tmp); + let cov = raw_coverage(&cfg, "github:o/r", "github.com/o/r").unwrap(); + assert_eq!(cov.total, 0); + assert!(cov.pending.is_empty()); + assert!(!needs_rebuild(&cfg, "github:o/r", "github.com/o/r")); + } + + #[test] + fn marked_files_drop_out_of_pending() { + let tmp = TempDir::new().unwrap(); + let cfg = test_config(&tmp); + write_raw(&cfg, "github.com/o/r", "commits", "100_aaa.md", "a"); + write_raw(&cfg, "github.com/o/r", "commits", "200_bbb.md", "b"); + + mark_raw_paths_ingested(&cfg, &["raw/github-com-o-r/commits/100_aaa.md".to_string()]) + .unwrap(); + + let cov = raw_coverage(&cfg, "github:o/r", "github.com/o/r").unwrap(); + assert_eq!(cov.total, 2); + assert_eq!(cov.covered, 1); + assert_eq!(cov.pending.len(), 1); + assert!(cov.pending[0].rel.ends_with("200_bbb.md")); + } + + #[tokio::test] + async fn backfill_marks_files_covered_by_existing_summaries() { + let tmp = TempDir::new().unwrap(); + let cfg = test_config(&tmp); + let scope = "github:o/r"; + let archive = "github.com/o/r"; + + // Two raw files on disk; an existing L1 summary covers only the + // commit (sync-written `commit:` label shape). + write_raw(&cfg, archive, "commits", "100_abc123.md", "commit body"); + write_raw(&cfg, archive, "issues", "200_42.md", "issue body"); + + let tree = get_or_create_source_tree(&cfg, scope).unwrap(); + let input = SummaryIngestInput { + content: "summary of one commit".to_string(), + token_count: 10, + entities: Vec::new(), + topics: Vec::new(), + time_range_start: chrono::Utc::now(), + time_range_end: chrono::Utc::now(), + score: 0.5, + child_labels: vec!["commit:abc123".to_string()], + child_basenames: Vec::new(), + }; + ingest_summary(&cfg, &tree, input).await.unwrap(); + + let cov = raw_coverage(&cfg, scope, archive).unwrap(); + assert_eq!(cov.total, 2); + assert_eq!(cov.covered, 1, "commit file backfilled as covered"); + assert_eq!(cov.pending.len(), 1); + assert!(cov.pending[0].rel.ends_with("200_42.md")); + } + + #[test] + fn chunk_raw_refs_count_as_covered() { + use crate::openhuman::memory_store::chunks::store::{ + set_chunk_raw_refs, upsert_chunks, RawRef, + }; + use crate::openhuman::memory_store::chunks::types::{ + chunk_id, Chunk, Metadata, SourceKind as ChunkSourceKind, SourceRef, + }; + + let tmp = TempDir::new().unwrap(); + let cfg = test_config(&tmp); + let scope = "gmail:user-at-example-dot-com"; + + // Two raw email files; one is referenced by a persisted chunk + // (the gmail pipeline shape), the other is orphaned. + write_raw(&cfg, scope, "emails", "1000_msg-a.md", "email a"); + write_raw(&cfg, scope, "emails", "2000_msg-b.md", "email b"); + + let now = chrono::Utc::now(); + let c = Chunk { + id: chunk_id(ChunkSourceKind::Email, scope, 0, "email a"), + content: "email a".into(), + metadata: Metadata { + source_kind: ChunkSourceKind::Email, + source_id: scope.into(), + owner: "user".into(), + timestamp: now, + time_range: (now, now), + tags: vec![], + source_ref: Some(SourceRef::new("gmail://msg-a")), + path_scope: None, + }, + token_count: 2, + seq_in_source: 0, + created_at: now, + partial_message: false, + }; + upsert_chunks(&cfg, &[c.clone()]).unwrap(); + set_chunk_raw_refs( + &cfg, + &c.id, + &[RawRef { + path: "raw/gmail-user-at-example-dot-com/emails/1000_msg-a.md".into(), + start: 0, + end: None, + }], + ) + .unwrap(); + + let cov = raw_coverage(&cfg, scope, scope).unwrap(); + assert_eq!(cov.total, 2); + assert_eq!(cov.covered, 1, "chunk-referenced file is covered"); + assert_eq!(cov.pending.len(), 1); + assert!(cov.pending[0].rel.ends_with("2000_msg-b.md")); + } + + #[tokio::test] + async fn rebuild_processes_only_pending_and_records_coverage() { + let tmp = TempDir::new().unwrap(); + let cfg = test_config(&tmp); + let scope = "github:o/r2"; + let archive = "github.com/o/r2"; + + write_raw(&cfg, archive, "commits", "100_aaa.md", "first commit body"); + write_raw(&cfg, archive, "commits", "200_bbb.md", "second commit body"); + mark_raw_paths_ingested( + &cfg, + &["raw/github-com-o-r2/commits/100_aaa.md".to_string()], + ) + .unwrap(); + + let provider: Arc = + Arc::new(StaticChatProvider::new("rebuilt summary content")); + let outcome = test_override::with_provider(provider, async { + rebuild_tree_from_raw(&cfg, scope, archive).await.unwrap() + }) + .await; + + assert_eq!(outcome.files_read, 1, "only the pending file is read"); + assert_eq!(outcome.batches, 1); + + // Idempotent: a second run finds nothing pending. + let cov = raw_coverage(&cfg, scope, archive).unwrap(); + assert!(cov.pending.is_empty(), "all files covered after rebuild"); + assert!(!needs_rebuild(&cfg, scope, archive)); + } } diff --git a/src/openhuman/memory_sync/workspace/mod.rs b/src/openhuman/memory_sync/workspace/mod.rs index 75057717b..ea73b4329 100644 --- a/src/openhuman/memory_sync/workspace/mod.rs +++ b/src/openhuman/memory_sync/workspace/mod.rs @@ -11,7 +11,15 @@ //! //! ## Status //! -//! Scaffold only. Today folder ingestion lives in +//! Mostly scaffold. Today folder ingestion lives in //! `memory_sources/readers/folder.rs`, harness capture in //! `agent_experience/`, and dictation in `dictation_hotkeys/`. Each will //! land here as a [`SyncPipeline`] impl in a follow-up. +//! +//! [`periodic`] is live: the background cadence driver that keeps +//! workspace-kind memory sources (GitHub repos, folders, RSS, web pages) +//! syncing without manual "Sync now" clicks. + +pub mod periodic; + +pub use periodic::start_workspace_periodic_sync; diff --git a/src/openhuman/memory_sync/workspace/periodic.rs b/src/openhuman/memory_sync/workspace/periodic.rs new file mode 100644 index 000000000..23c2837bc --- /dev/null +++ b/src/openhuman/memory_sync/workspace/periodic.rs @@ -0,0 +1,347 @@ +//! Periodic sync scheduler for workspace (non-Composio) memory sources. +//! +//! The Composio scheduler (`memory_sync::composio::periodic`) walks +//! Composio *connections* exclusively — GitHub repos, folders, RSS feeds +//! and web pages registered in `config.memory_sources` were only ever +//! synced when the user pressed "Sync now" (the `memory_sources.sync` +//! RPC). A GitHub source would sync once at setup and then silently go +//! stale forever. This loop closes that gap: it walks the registry on a +//! fixed tick and fires the existing [`sync_source`] dispatcher for every +//! enabled workspace-kind source whose cadence has elapsed. +//! +//! Cadence semantics mirror the Composio loop (#3302): +//! - `config.memory_sync_interval_secs == Some(0)` → "Manual only", the +//! loop skips every source. +//! - `Some(n)` → sync every `max(n, 24h-default)` seconds. +//! - `None` → the 24h default. +//! +//! Due-check sources, in priority order: +//! 1. the in-memory fired-at map (most accurate within this process), and +//! 2. the persisted sync-audit log — keyed by `source_id` with the +//! source's own `source_kind` — so a configured cadence survives app +//! restarts instead of re-firing on every cold start. +//! +//! `sync_source` itself owns overlap protection (per-source `ACTIVE_SYNCS` +//! mutex), audit writes, and post-sync raw-coverage reconcile +//! (`check_and_rebuild_tree`), so this loop stays a thin cadence driver. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex, OnceLock}; +use std::time::{Duration, Instant}; + +use chrono::{DateTime, Utc}; +use tokio::time::interval; + +use crate::openhuman::config::rpc as config_rpc; +use crate::openhuman::config::DEFAULT_MEMORY_SYNC_INTERVAL_SECS; +use crate::openhuman::memory_sources::sync::sync_source; +use crate::openhuman::memory_sources::types::{MemorySourceEntry, SourceKind}; +use crate::openhuman::memory_sync::composio::periodic::{ + connection_is_due, effective_interval_secs, periodic_pause_reason, +}; +use crate::openhuman::memory_sync::sources::audit::{read_audit_log, SyncAuditEntry}; +use crate::openhuman::scheduler_gate::gate::resume_notify; + +/// How often the scheduler wakes up to look for due syncs. Matches the +/// Composio loop's cadence — per-source intervals (24h default) bound the +/// actual sync frequency; this only bounds how far past due we can drift. +const TICK_SECONDS: u64 = 1200; + +/// Process-wide guard: only the first call spawns the loop. +static SCHEDULER_STARTED: OnceLock<()> = OnceLock::new(); + +/// `source_id → last fired-at instant` for this process lifetime. Recorded +/// at *fire* time (the sync runs detached in `sync_source`'s spawned task), +/// so a failing source retries on the next due boundary, not every tick. +type FiredAtMap = Arc>>; + +static LAST_FIRED_AT: OnceLock = OnceLock::new(); + +fn fired_map() -> FiredAtMap { + LAST_FIRED_AT + .get_or_init(|| Arc::new(Mutex::new(HashMap::new()))) + .clone() +} + +/// Source kinds this loop schedules. Composio is owned by the Composio +/// scheduler; Conversation/Twitter have no periodic pull semantics today. +fn is_workspace_synced_kind(kind: &SourceKind) -> bool { + matches!( + kind, + SourceKind::GithubRepo | SourceKind::Folder | SourceKind::RssFeed | SourceKind::WebPage + ) +} + +/// Index `source_id → most recent successful sync timestamp` from the +/// persisted audit log, restricted to workspace source kinds. Failed runs +/// are skipped (matching the in-memory semantics — a failure retries at +/// the next tick after the cadence elapses). +fn index_last_success_by_source_id(entries: &[SyncAuditEntry]) -> HashMap> { + let mut idx: HashMap> = HashMap::new(); + for e in entries { + if !e.success { + continue; + } + let is_workspace_kind = matches!( + e.source_kind.as_str(), + "github_repo" | "folder" | "rss_feed" | "web_page" + ); + if !is_workspace_kind { + continue; + } + idx.entry(e.source_id.clone()) + .and_modify(|t| { + if e.timestamp > *t { + *t = e.timestamp; + } + }) + .or_insert(e.timestamp); + } + idx +} + +/// Wall-clock elapsed since the persisted last success, saturating at zero +/// for clock skew. `None` when the source has never successfully synced. +fn persisted_since_last_sync( + idx: &HashMap>, + source_id: &str, + now: DateTime, +) -> Option { + idx.get(source_id).map(|ts| { + let secs = (now - *ts).num_seconds().max(0) as u64; + Duration::from_secs(secs) + }) +} + +/// Spawn the workspace-source periodic sync task. Idempotent. +pub fn start_workspace_periodic_sync() { + if SCHEDULER_STARTED.set(()).is_err() { + tracing::debug!("[memory_sync:workspace:periodic] scheduler already running"); + return; + } + tokio::spawn(async move { + tracing::info!( + tick_seconds = TICK_SECONDS, + "[memory_sync:workspace:periodic] scheduler starting" + ); + run_loop().await; + tracing::error!("[memory_sync:workspace:periodic] scheduler loop exited"); + }); +} + +/// Tick loop: wakes on the steady cadence or a scheduler-gate resume +/// (Memory Tree toggled back on / sign-in), same shape as the Composio +/// loop — resume runs a tick immediately and re-bases the ticker. +async fn run_loop() { + let mut ticker = interval(Duration::from_secs(TICK_SECONDS)); + let resume = resume_notify(); + // Skip the immediate-fire tick so startup isn't slammed before sign-in. + ticker.tick().await; + + loop { + tokio::select! { + _ = ticker.tick() => {} + _ = resume.notified() => { + ticker.reset(); + } + } + if let Err(e) = run_one_tick().await { + tracing::warn!( + error = %e, + "[memory_sync:workspace:periodic] tick failed (continuing)" + ); + } + } +} + +/// Run a single scheduler tick. `pub(crate)` so tests can drive ticks +/// without the real interval. +pub(crate) async fn run_one_tick() -> Result<(), String> { + // Honour the same pause reasons as the Composio loop: user toggled + // Memory Tree off, or signed out. + if let Some(reason) = periodic_pause_reason() { + tracing::debug!( + reason = reason.as_str(), + "[memory_sync:workspace:periodic] scheduler-gate paused — skipping tick" + ); + return Ok(()); + } + + let config = config_rpc::load_config_with_timeout() + .await + .map_err(|e| format!("load_config: {e}"))?; + + let global_interval = config.memory_sync_interval_secs; + let Some(interval_secs) = + effective_interval_secs(DEFAULT_MEMORY_SYNC_INTERVAL_SECS, global_interval) + else { + tracing::debug!( + "[memory_sync:workspace:periodic] manual-only mode — skipping all workspace sources" + ); + return Ok(()); + }; + + let audit_index = index_last_success_by_source_id(&read_audit_log(&config)); + let now = Utc::now(); + let map = fired_map(); + + let due_sources: Vec = config + .memory_sources + .iter() + .filter(|s| s.enabled && is_workspace_synced_kind(&s.kind)) + .filter(|s| { + let since = { + let guard = map.lock().unwrap_or_else(|e| e.into_inner()); + guard.get(&s.id).map(|when| when.elapsed()) + } + .or_else(|| persisted_since_last_sync(&audit_index, &s.id, now)); + connection_is_due(interval_secs, since) + }) + .cloned() + .collect(); + + if due_sources.is_empty() { + tracing::debug!("[memory_sync:workspace:periodic] tick complete — nothing due"); + return Ok(()); + } + + let mut fired = 0usize; + for source in due_sources { + let source_id = source.id.clone(); + let kind = source.kind.as_str(); + tracing::info!( + source_id = %source_id, + kind = %kind, + interval_secs, + "[memory_sync:workspace:periodic] firing sync" + ); + // sync_source spawns the actual work and returns immediately; it + // rejects overlapping syncs of the same source internally. + match sync_source(source, config.clone()).await { + Ok(()) => { + if let Ok(mut guard) = map.lock() { + guard.insert(source_id, Instant::now()); + } + fired += 1; + } + Err(e) => { + tracing::warn!( + source_id = %source_id, + kind = %kind, + error = %e, + "[memory_sync:workspace:periodic] sync dispatch failed (will retry next tick)" + ); + } + } + } + + tracing::debug!(fired, "[memory_sync:workspace:periodic] tick complete"); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn entry(source_id: &str, kind: &str, success: bool, ts: DateTime) -> SyncAuditEntry { + SyncAuditEntry { + timestamp: ts, + source_id: source_id.to_string(), + source_kind: kind.to_string(), + scope: format!("{kind}:{source_id}"), + items_fetched: 1, + batches: 0, + input_tokens: 0, + output_tokens: 0, + estimated_cost_usd: 0.0, + composio_actions_called: 0, + composio_cost_usd: 0.0, + actual_charged_usd: None, + duration_ms: 10, + success, + error: None, + } + } + + #[test] + fn workspace_kinds_are_scheduled_composio_is_not() { + assert!(is_workspace_synced_kind(&SourceKind::GithubRepo)); + assert!(is_workspace_synced_kind(&SourceKind::Folder)); + assert!(is_workspace_synced_kind(&SourceKind::RssFeed)); + assert!(is_workspace_synced_kind(&SourceKind::WebPage)); + assert!(!is_workspace_synced_kind(&SourceKind::Composio)); + assert!(!is_workspace_synced_kind(&SourceKind::Conversation)); + assert!(!is_workspace_synced_kind(&SourceKind::TwitterQuery)); + } + + #[test] + fn audit_index_keeps_latest_workspace_success_and_skips_others() { + let now = Utc::now(); + let older = now - chrono::Duration::hours(30); + let newer = now - chrono::Duration::hours(2); + let entries = vec![ + entry("src_gh", "github_repo", true, older), + entry("src_gh", "github_repo", true, newer), // newest success wins + entry("src_gh", "github_repo", false, now), // failure ignored + entry("conn_1", "composio", true, now), // composio kind ignored + ]; + let idx = index_last_success_by_source_id(&entries); + assert_eq!(idx.get("src_gh"), Some(&newer)); + assert!(!idx.contains_key("conn_1")); + } + + /// The headline regression: a GitHub source that synced once long ago + /// must read as DUE under the default 24h cadence — before this loop + /// existed, nothing ever consulted that staleness, so the source went + /// permanently dark after its first manual sync. + #[test] + fn stale_github_source_is_due_fresh_one_is_not() { + let now = Utc::now(); + let mut idx = HashMap::new(); + idx.insert("src_stale".to_string(), now - chrono::Duration::days(5)); + idx.insert("src_fresh".to_string(), now - chrono::Duration::hours(1)); + + let interval = + effective_interval_secs(DEFAULT_MEMORY_SYNC_INTERVAL_SECS, None).expect("interval"); + + let stale = persisted_since_last_sync(&idx, "src_stale", now); + assert!(connection_is_due(interval, stale), "5-day-old sync is due"); + + let fresh = persisted_since_last_sync(&idx, "src_fresh", now); + assert!( + !connection_is_due(interval, fresh), + "1h-old sync is not due" + ); + + // Never-synced source fires immediately. + let never = persisted_since_last_sync(&idx, "src_new", now); + assert!(connection_is_due(interval, never)); + } + + #[test] + fn manual_only_global_setting_disables_the_loop() { + assert_eq!( + effective_interval_secs(DEFAULT_MEMORY_SYNC_INTERVAL_SECS, Some(0)), + None + ); + } + + #[test] + fn persisted_since_last_sync_saturates_clock_skew() { + let now = Utc::now(); + let mut idx = HashMap::new(); + idx.insert("future".to_string(), now + chrono::Duration::hours(2)); + assert_eq!( + persisted_since_last_sync(&idx, "future", now), + Some(Duration::ZERO) + ); + assert_eq!(persisted_since_last_sync(&idx, "missing", now), None); + } + + #[tokio::test] + async fn start_workspace_periodic_sync_is_idempotent() { + start_workspace_periodic_sync(); + start_workspace_periodic_sync(); + assert!(SCHEDULER_STARTED.get().is_some()); + } +} diff --git a/src/openhuman/memory_tree/tree/bucket_seal.rs b/src/openhuman/memory_tree/tree/bucket_seal.rs index b43dd9456..e18a78850 100644 --- a/src/openhuman/memory_tree/tree/bucket_seal.rs +++ b/src/openhuman/memory_tree/tree/bucket_seal.rs @@ -306,17 +306,18 @@ pub async fn cascade_all_from( /// Level-aware seal gate. /// -/// L0 buffers gate on **either** `token_sum >= INPUT_TOKEN_BUDGET` -/// (so the summariser's input stays bounded) **or** sibling count -/// `>= SUMMARY_FANOUT` (so leaves form predictably for sources whose -/// chunks are individually small — without the count fallback, -/// hundreds of tiny emails can sit unsealed waiting to hit 50k -/// tokens). +/// L0 buffers gate on `token_sum >= INPUT_TOKEN_BUDGET` **only** — no +/// sibling-count fallback. Token-only gating is deliberate (see the +/// module header): small-token items (commit messages, short emails) +/// accumulate into large batches so each L1 summary covers a +/// meaningful span instead of 10 tiny items. Small buffers that never +/// reach the budget are sealed by time instead: /// /// Time-based sealing for low-volume sources is handled separately /// by `flush_stale_buffers` (see [`crate::openhuman::memory_tree:: /// tree::flush::flush_stale_buffers`]), which filters buffers -/// by `oldest_at` before calling the cascade. Keeping the time gate +/// by `oldest_at` before calling the cascade (the 3-hourly +/// `memory_queue` scheduler drives it). Keeping the time gate /// out of `should_seal` avoids prematurely sealing buffers during /// normal `append_leaf` calls when test/restored data carries older /// timestamps. diff --git a/tests/config_auth_app_state_connectivity_e2e.rs b/tests/config_auth_app_state_connectivity_e2e.rs index b992116c7..2d3041e9b 100644 --- a/tests/config_auth_app_state_connectivity_e2e.rs +++ b/tests/config_auth_app_state_connectivity_e2e.rs @@ -2867,6 +2867,7 @@ async fn worker_a_controller_schemas_are_fully_exposed() { "openhuman.memory_sources_list_items", "openhuman.memory_sources_monthly_cost_summary", "openhuman.memory_sources_read_item", + "openhuman.memory_sources_reconcile", "openhuman.memory_sources_remove", "openhuman.memory_sources_status_list", "openhuman.memory_sources_supported_toolkits", diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index a33011ddb..610ca5aa7 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -11623,6 +11623,94 @@ async fn json_rpc_memory_sources_conversation_crud() { rpc_join.abort(); } +/// Raw-archive coverage reconcile over JSON-RPC: a GitHub source whose raw +/// archive holds files that no tree summary covers must be reported as +/// pending by `openhuman.memory_sources_reconcile` (report-only mode). +/// Regression for the silent divergence where thousands of raw files sat +/// unsummarised with no way to see or repair it. +#[tokio::test] +async fn json_rpc_memory_sources_reconcile_reports_pending_raw_files() { + let _env_lock = json_rpc_e2e_env_lock(); + let tmp = tempdir().expect("tempdir"); + let home = tmp.path(); + let _home_guard = EnvVarGuard::set_to_path("HOME", home); + let _ws_guard = EnvVarGuard::set_to_path("OPENHUMAN_WORKSPACE", home); + let _action_guard = EnvVarGuard::set_to_path("OPENHUMAN_ACTION_DIR", home); + let _backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); + + let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; + let rpc_base = format!("http://{rpc_addr}"); + + // 1. Register a GitHub repo source (no sync — we plant the archive). + let add_resp = post_json_rpc( + &rpc_base, + 9601, + "openhuman.memory_sources_add", + json!({ + "kind": "github_repo", + "label": "Repo", + "enabled": true, + "url": "https://github.com/owner/repo" + }), + ) + .await; + let add_result = assert_no_jsonrpc_error(&add_resp, "memory_sources_add github_repo"); + let source_id = add_result + .get("source") + .and_then(|s| s.get("id")) + .and_then(Value::as_str) + .expect("source must have id") + .to_string(); + + // 2. Plant two uncovered raw archive files where a fetch would write + // them: /memory_tree/content/raw/github-com-owner-repo/. + // With OPENHUMAN_WORKSPACE=$home and no config.toml, the resolved + // workspace dir is $home/workspace (resolve_config_dir_for_workspace). + let commits_dir = home + .join("workspace") + .join("memory_tree") + .join("content") + .join("raw") + .join("github-com-owner-repo") + .join("commits"); + std::fs::create_dir_all(&commits_dir).expect("create raw commits dir"); + std::fs::write(commits_dir.join("1000_aaa.md"), "commit one").expect("write raw"); + std::fs::write(commits_dir.join("2000_bbb.md"), "commit two").expect("write raw"); + + // 3. Report-only reconcile must surface both files as pending. + let reconcile_resp = post_json_rpc( + &rpc_base, + 9602, + "openhuman.memory_sources_reconcile", + json!({ "source_id": source_id }), + ) + .await; + let reconcile_result = assert_no_jsonrpc_error(&reconcile_resp, "memory_sources_reconcile"); + let scopes = reconcile_result + .get("scopes") + .and_then(Value::as_array) + .expect("reconcile should return scopes array"); + assert_eq!(scopes.len(), 1, "one scope for one github source"); + let scope = &scopes[0]; + assert_eq!( + scope.get("tree_scope").and_then(Value::as_str), + Some("github:owner/repo") + ); + assert_eq!( + scope.get("total_raw_files").and_then(Value::as_u64), + Some(2) + ); + assert_eq!(scope.get("covered").and_then(Value::as_u64), Some(0)); + assert_eq!(scope.get("pending").and_then(Value::as_u64), Some(2)); + assert_eq!( + scope.get("started").and_then(Value::as_bool), + Some(false), + "report-only call must not start a reconcile" + ); + + rpc_join.abort(); +} + // ── Memory sources: active-connection filtering over JSON-RPC ──────────────── /// Mock Composio v3 `/connected_accounts`. Returns a single ACTIVE Gmail diff --git a/tests/memory_sync_pipeline_e2e.rs b/tests/memory_sync_pipeline_e2e.rs index a85cf170a..8d433ae47 100644 --- a/tests/memory_sync_pipeline_e2e.rs +++ b/tests/memory_sync_pipeline_e2e.rs @@ -311,7 +311,7 @@ async fn rebuild_tree_from_raw_builds_from_disk() { let before = get_or_create_source_tree(&cfg, scope).unwrap(); assert_eq!(before.max_level, 0, "fresh tree should be at level 0"); - let outcome = rebuild_tree_from_raw(&cfg, scope).await.unwrap(); + let outcome = rebuild_tree_from_raw(&cfg, scope, scope).await.unwrap(); assert_eq!(outcome.files_read, 3, "should read the 3 seeded emails"); assert!(outcome.batches >= 1, "should produce at least one batch"); @@ -452,12 +452,12 @@ async fn check_and_rebuild_auto_detects_raw_without_summaries() { // Before: raw exists, tree at level 0 → rebuild needed. assert!( - needs_rebuild(&cfg, scope), - "needs_rebuild must be true when raw exists and max_level == 0" + needs_rebuild(&cfg, scope, scope), + "needs_rebuild must be true when raw files exist with no coverage" ); // Drive the rebuild (what check_and_rebuild_tree calls). - rebuild_tree_from_raw(&cfg, scope).await.unwrap(); + rebuild_tree_from_raw(&cfg, scope, scope).await.unwrap(); // After: tree now has summaries → no further rebuild needed. let tree = get_or_create_source_tree(&cfg, scope).unwrap(); @@ -467,13 +467,17 @@ async fn check_and_rebuild_auto_detects_raw_without_summaries() { tree.max_level ); assert!( - !needs_rebuild(&cfg, scope), - "needs_rebuild must be false once the tree has summaries" + !needs_rebuild(&cfg, scope, scope), + "needs_rebuild must be false once every raw file is covered" ); // A scope with no raw files on disk never triggers a rebuild. assert!( - !needs_rebuild(&cfg, "gmail:empty-at-example-dot-com"), + !needs_rebuild( + &cfg, + "gmail:empty-at-example-dot-com", + "gmail:empty-at-example-dot-com" + ), "needs_rebuild must be false when no raw directory exists" ); }