diff --git a/Cargo.lock b/Cargo.lock index df9404a08..72ee1df3b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4520,7 +4520,7 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" [[package]] name = "openhuman" -version = "0.53.18" +version = "0.53.19" dependencies = [ "aes-gcm", "anyhow", diff --git a/app/src-tauri/Cargo.lock b/app/src-tauri/Cargo.lock index ee198c50a..4fc23443e 100644 --- a/app/src-tauri/Cargo.lock +++ b/app/src-tauri/Cargo.lock @@ -4,7 +4,7 @@ version = 4 [[package]] name = "OpenHuman" -version = "0.53.18" +version = "0.53.19" dependencies = [ "anyhow", "async-trait", @@ -4468,7 +4468,7 @@ dependencies = [ [[package]] name = "openhuman" -version = "0.53.18" +version = "0.53.19" dependencies = [ "aes-gcm", "anyhow", diff --git a/scripts/memory-tree-progress.sh b/scripts/memory-tree-progress.sh new file mode 100755 index 000000000..2ede11b80 --- /dev/null +++ b/scripts/memory-tree-progress.sh @@ -0,0 +1,207 @@ +#!/usr/bin/env bash +# +# memory-tree-progress.sh — live progress monitor for the memory_tree pipeline. +# +# Polls the workspace SQLite DB and prints a one-line snapshot every INTERVAL +# seconds: extract jobs done/pending, summaries per level, the currently +# claimed job (if any), recent throughput, and a rolling cloud-LLM round-trip +# estimate scraped from the core log. Exits cleanly when there is nothing left +# to do (no `ready`/`running` jobs other than the daily digest dedupe). +# +# Optionally triggers a fresh `flush_now` first so the seal cascade picks up +# whatever is currently buffered without waiting for the 50k-token threshold. +# +# Usage: +# scripts/memory-tree-progress.sh # monitor only +# scripts/memory-tree-progress.sh --flush # flush_now then monitor +# scripts/memory-tree-progress.sh --interval 10 # change tick (default 5s) +# scripts/memory-tree-progress.sh --log /tmp/x.log # override log path +# scripts/memory-tree-progress.sh --once # one snapshot, then exit +# +# Env: +# OPENHUMAN_WORKSPACE — workspace dir (default: derive from active_user.toml) +# CORE_BIN — path to openhuman-core (default: target/debug/openhuman-core) +# CORE_LOG — core log to scrape for round-trip times (default: /tmp/oh-core.log) +# +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +REPO_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)" +cd "$REPO_ROOT" + +INTERVAL=5 +DO_FLUSH=0 +ONCE=0 +CORE_BIN="${CORE_BIN:-target/debug/openhuman-core}" +CORE_LOG="${CORE_LOG:-/tmp/oh-core.log}" + +while [ $# -gt 0 ]; do + case "$1" in + --flush) DO_FLUSH=1; shift ;; + --interval) + [ $# -ge 2 ] || { echo "--interval requires a value (seconds)" >&2; exit 2; } + [[ "$2" =~ ^[1-9][0-9]*$ ]] || { echo "--interval must be a positive integer" >&2; exit 2; } + INTERVAL="$2"; shift 2 ;; + --log) + [ $# -ge 2 ] || { echo "--log requires a file path" >&2; exit 2; } + CORE_LOG="$2"; shift 2 ;; + --once) ONCE=1; shift ;; + -h|--help) sed -n '2,25p' "$0"; exit 0 ;; + *) echo "unknown arg: $1" >&2; exit 2 ;; + esac +done + +# ── Resolve workspace + DB path ───────────────────────────────────────────── + +if [ -z "${OPENHUMAN_WORKSPACE:-}" ]; then + DEFAULT_DIR="$HOME/.openhuman-staging" + [ -d "$DEFAULT_DIR" ] || DEFAULT_DIR="$HOME/.openhuman" + ACTIVE_USER_FILE="$DEFAULT_DIR/active_user.toml" + if [ -f "$ACTIVE_USER_FILE" ]; then + USER_ID=$(awk -F'"' '/user_id/ {print $2; exit}' "$ACTIVE_USER_FILE") + OPENHUMAN_WORKSPACE="$DEFAULT_DIR/users/$USER_ID/workspace" + fi +fi +DB="${OPENHUMAN_WORKSPACE:-}/memory_tree/chunks.db" +if [ ! -f "$DB" ]; then + echo "memory_tree DB not found at: $DB" >&2 + echo "Set OPENHUMAN_WORKSPACE to override." >&2 + exit 1 +fi + +echo "workspace: $OPENHUMAN_WORKSPACE" +echo "db: $DB" +echo "log: $CORE_LOG" +echo + +# ── Optional initial flush ────────────────────────────────────────────────── + +if [ "$DO_FLUSH" = 1 ]; then + if [ ! -x "$CORE_BIN" ]; then + echo "core binary not found: $CORE_BIN — build with 'cargo build --bin openhuman-core'" >&2 + exit 1 + fi + echo "→ triggering memory_tree.flush_now" + # Capture the full output so we can echo it on failure (the call's + # exit code is what we gate on; the grep below is just for the + # success-path summary). + flush_out="$("$CORE_BIN" call --method openhuman.memory_tree_flush_now --params '{}' 2>&1)" || { + echo "$flush_out" >&2 + echo "flush_now failed; aborting monitor start." >&2 + exit 1 + } + echo "$flush_out" | grep -E '"enqueued"|"stale_buffers"|memory_tree::read' || true + echo +fi + +# ── Snapshot helper ───────────────────────────────────────────────────────── + +q() { sqlite3 "$DB" "$@"; } + +# Track previous done counts so we can show throughput per tick. +PREV_EXTRACT_DONE=0 +PREV_SUMMARIES=0 +START_TS=$(date +%s) + +snapshot() { + local now ts ext_done ext_ready ext_run sums_l1 sums_l2 sums_l3 sums_l0 \ + chunks_pending chunks_admitted chunks_buffered \ + running_kind running_started_ms running_age_s \ + rt_recent_avg eta_min + + now=$(date +%s) + ts=$(date '+%H:%M:%S') + + ext_done=$(q "SELECT COALESCE(COUNT(*),0) FROM mem_tree_jobs WHERE kind='extract_chunk' AND status='done';") + ext_ready=$(q "SELECT COALESCE(COUNT(*),0) FROM mem_tree_jobs WHERE kind='extract_chunk' AND status='ready';") + ext_run=$(q "SELECT COALESCE(COUNT(*),0) FROM mem_tree_jobs WHERE kind='extract_chunk' AND status='running';") + + sums_l0=$(q "SELECT COALESCE(COUNT(*),0) FROM mem_tree_summaries WHERE level=0;") + sums_l1=$(q "SELECT COALESCE(COUNT(*),0) FROM mem_tree_summaries WHERE level=1;") + sums_l2=$(q "SELECT COALESCE(COUNT(*),0) FROM mem_tree_summaries WHERE level=2;") + sums_l3=$(q "SELECT COALESCE(COUNT(*),0) FROM mem_tree_summaries WHERE level>=3;") + + chunks_pending=$(q "SELECT COALESCE(COUNT(*),0) FROM mem_tree_chunks WHERE lifecycle_status='pending_extraction';") + chunks_admitted=$(q "SELECT COALESCE(COUNT(*),0) FROM mem_tree_chunks WHERE lifecycle_status='admitted';") + chunks_buffered=$(q "SELECT COALESCE(COUNT(*),0) FROM mem_tree_chunks WHERE lifecycle_status='buffered';") + + # Currently-claimed work (any kind), with age in seconds. + local now_ms; now_ms=$((now * 1000)) + running_row=$(q "SELECT kind, COALESCE(started_at_ms,0) FROM mem_tree_jobs WHERE status='running' ORDER BY started_at_ms ASC LIMIT 1;") + if [ -n "$running_row" ]; then + running_kind=$(echo "$running_row" | cut -d'|' -f1) + running_started_ms=$(echo "$running_row" | cut -d'|' -f2) + if [ "$running_started_ms" -gt 0 ] 2>/dev/null; then + running_age_s=$(( (now_ms - running_started_ms) / 1000 )) + else + running_age_s="?" + fi + else + running_kind="-" + running_age_s="-" + fi + + # Throughput since last tick. + local d_extract=$((ext_done - PREV_EXTRACT_DONE)) + local d_sums=$(( (sums_l0 + sums_l1 + sums_l2 + sums_l3) - PREV_SUMMARIES )) + PREV_EXTRACT_DONE=$ext_done + PREV_SUMMARIES=$((sums_l0 + sums_l1 + sums_l2 + sums_l3)) + + # Rolling round-trip estimate from the last few cloud responses. + rt_recent_avg="?" + if [ -f "$CORE_LOG" ]; then + rt_recent_avg=$(awk ' + /\[memory_tree::chat::cloud\] kind=/ { + split($1, a, ":"); start = a[1]*3600 + a[2]*60 + a[3] + } + /\[memory_tree::chat::cloud\] response/ { + split($1, a, ":"); end = a[1]*3600 + a[2]*60 + a[3] + if (start > 0) { sum += (end - start); n++ } + } + END { if (n>0) printf "%.1fs", sum/n; else printf "?" } + ' "$CORE_LOG" | tail -c 16) + fi + + eta_min="?" + if [ "$d_extract" -gt 0 ] 2>/dev/null; then + # ETA based on jobs/tick * INTERVAL seconds. + local secs_per=$(( INTERVAL / d_extract )) + [ "$secs_per" -lt 1 ] && secs_per=1 + eta_min=$(( ext_ready * secs_per / 60 ))m + fi + + # NOTE: source-tree leaves are L1+ (raw chunks are the L0 leaves of the + # tree but aren't represented in `mem_tree_summaries`); the L0 row in + # `mem_tree_summaries` is only populated by global-tree daily digests. + # We surface it here as `digest=` so the bucket name doesn't mislead. + printf "%s extract: done=%d pending=%d run=%d (+%d/tick eta~%s) summaries L1=%d L2=%d L3+=%d digest=%d (+%d) chunks: pend=%d adm=%d buf=%d running=%s/%ss cloud_avg=%s\n" \ + "$ts" "$ext_done" "$ext_ready" "$ext_run" "$d_extract" "$eta_min" \ + "$sums_l1" "$sums_l2" "$sums_l3" "$sums_l0" "$d_sums" \ + "$chunks_pending" "$chunks_admitted" "$chunks_buffered" \ + "$running_kind" "$running_age_s" "$rt_recent_avg" + + # Done-condition: nothing pending or running across all kinds (digest_daily + # rows are dedupe-suppressed steady state, ignore them). + local active_other + active_other=$(q "SELECT COALESCE(COUNT(*),0) FROM mem_tree_jobs \ + WHERE status IN ('ready','running') \ + AND kind <> 'digest_daily';") + [ "$active_other" = "0" ] +} + +# ── Loop ──────────────────────────────────────────────────────────────────── + +if [ "$ONCE" = 1 ]; then + snapshot || true + exit 0 +fi + +trap 'echo; echo "interrupted."; exit 0' INT + +while true; do + if snapshot; then + echo "→ pipeline idle (no ready/running jobs). exiting." + exit 0 + fi + sleep "$INTERVAL" +done diff --git a/src/openhuman/memory/tree/chat/local.rs b/src/openhuman/memory/tree/chat/local.rs index 700958f08..18dbb7707 100644 --- a/src/openhuman/memory/tree/chat/local.rs +++ b/src/openhuman/memory/tree/chat/local.rs @@ -62,6 +62,20 @@ impl ChatProvider for OllamaChatProvider { } async fn chat_for_json(&self, prompt: &ChatPrompt) -> Result { + self.run_chat(prompt, Some("json")).await + } + + async fn chat_for_text(&self, prompt: &ChatPrompt) -> Result { + // Omit `format` entirely — Ollama's `/api/chat` only accepts + // `"json"`, a JSON-schema object, or absence-of-field for + // free-form text. Sending `format: ""` is undefined behaviour, + // so the field is dropped from the request body when None. + self.run_chat(prompt, None).await + } +} + +impl OllamaChatProvider { + async fn run_chat(&self, prompt: &ChatPrompt, format: Option<&str>) -> Result { if self.endpoint.is_empty() || self.model.is_empty() { return Err(anyhow!( "[memory_tree::chat::local] Ollama endpoint or model not configured \ @@ -84,7 +98,7 @@ impl ChatProvider for OllamaChatProvider { content: prompt.user.clone(), }, ], - format: "json".to_string(), + format: format.map(str::to_string), stream: false, options: OllamaOptions { temperature: prompt.temperature as f32, @@ -141,7 +155,11 @@ fn truncate_for_log(s: &str, max_chars: usize) -> String { struct OllamaChatRequest { model: String, messages: Vec, - format: String, + /// Omitted from the wire body when `None` (`#[serde(skip_serializing_if)]`), + /// so the JSON-mode flag is only present for the `chat_for_json` path. + /// Ollama treats absence as "free-form text". + #[serde(skip_serializing_if = "Option::is_none")] + format: Option, stream: bool, options: OllamaOptions, } diff --git a/src/openhuman/memory/tree/chat/mod.rs b/src/openhuman/memory/tree/chat/mod.rs index 505c27e66..009c62cb3 100644 --- a/src/openhuman/memory/tree/chat/mod.rs +++ b/src/openhuman/memory/tree/chat/mod.rs @@ -78,11 +78,24 @@ pub trait ChatProvider: Send + Sync { /// Stable, grep-friendly name for logs. e.g. `"cloud:summarization-v1"`. fn name(&self) -> &str; - /// Run one chat completion and return the assistant's content. + /// Run one chat completion and return the assistant's content, + /// constraining the model to JSON output where the wire format + /// supports it (Ollama's `format: "json"`). /// /// Implementations should log entry / exit at debug level under the /// `[memory_tree::chat]` prefix. async fn chat_for_json(&self, prompt: &ChatPrompt) -> Result; + + /// Run one chat completion and return the assistant's plain-text + /// content. Unlike [`chat_for_json`], implementations MUST NOT + /// enable any wire-level JSON-mode flag — used by the summariser + /// which emits prose, not a structured envelope. + /// + /// Default impl forwards to `chat_for_json`; providers that gate + /// JSON-mode at the wire (e.g. Ollama) override to skip it. + async fn chat_for_text(&self, prompt: &ChatPrompt) -> Result { + self.chat_for_json(prompt).await + } } /// Build the [`ChatProvider`] dictated by `config.memory_tree.llm_backend`. diff --git a/src/openhuman/memory/tree/chunker.rs b/src/openhuman/memory/tree/chunker.rs index 3dfe8a248..064e6a205 100644 --- a/src/openhuman/memory/tree/chunker.rs +++ b/src/openhuman/memory/tree/chunker.rs @@ -2,8 +2,9 @@ //! //! The canonicalisers produce one big canonical Markdown blob per source //! record; the chunker slices that into chunks of at most [`DEFAULT_CHUNK_MAX_TOKENS`] -//! so later phases (#709 seal budget = 10k tokens) can ingest them without -//! blowing past the summariser ceiling. +//! so later phases (L0 seal at `INPUT_TOKEN_BUDGET = 50k` tokens, or 10 +//! items via the count fallback) can ingest them without blowing past +//! the summariser ceiling. //! //! ## Dispatch by source kind (Phase B) //! @@ -20,10 +21,10 @@ use crate::openhuman::memory::tree::util::redact::redact; /// Default upper bound on per-chunk tokens. /// -/// Sized below the L0 seal budget (`tree_source::types::TOKEN_BUDGET = 4_500`) -/// so each seal accumulates roughly 1–3 chunks before firing — natural pacing -/// for the local 1B summariser, which produces noticeably better summaries -/// with smaller (≤4–5k) inputs than at the previous 10k cap. +/// Well below `tree_source::types::INPUT_TOKEN_BUDGET = 50_000` so each +/// L0 seal accumulates many chunks (~15+) before firing — the cloud +/// summariser handles large input contexts well, so we let the seal +/// fold a meaningful slice of the source rather than a single chunk. pub const DEFAULT_CHUNK_MAX_TOKENS: u32 = 3_000; /// Tunable settings for the chunker. diff --git a/src/openhuman/memory/tree/content_store/mod.rs b/src/openhuman/memory/tree/content_store/mod.rs index ca2bb5f04..9eac121cf 100644 --- a/src/openhuman/memory/tree/content_store/mod.rs +++ b/src/openhuman/memory/tree/content_store/mod.rs @@ -14,6 +14,7 @@ pub mod atomic; pub mod compose; +pub mod obsidian; pub mod paths; pub mod raw; pub mod read; diff --git a/src/openhuman/memory/tree/content_store/obsidian.rs b/src/openhuman/memory/tree/content_store/obsidian.rs new file mode 100644 index 000000000..0217765e3 --- /dev/null +++ b/src/openhuman/memory/tree/content_store/obsidian.rs @@ -0,0 +1,143 @@ +//! Obsidian vault defaults. +//! +//! When the memory_tree content root is first populated we drop a small +//! `.obsidian/` directory into it so a user opening the vault gets the +//! intended graph-view colour mapping (one colour per summary level) and +//! the front-matter type hints (`time_range_*` as `date`, `sealed_at` as +//! `datetime`) without any manual configuration. +//! +//! The bundled defaults live as static files under `obsidian_defaults/` +//! and are baked into the binary via `include_str!`. We only stage them +//! when the corresponding `.obsidian/` doesn't already exist — +//! never overwrite a file the user has tweaked. +//! +//! Callers should invoke [`ensure_obsidian_defaults`] from any code path +//! that creates files under `content_root` (summary stage, raw write, +//! etc.). The function is idempotent and cheap on the steady-state path +//! (one `Path::exists()` per file). +//! +//! Failure mode: best-effort. A failed stage logs a warn and returns +//! `Ok(())` so seal/raw-write callers don't abort persistence over a +//! cosmetic vault default. + +use std::path::Path; + +use anyhow::Result; + +const GRAPH_JSON: &str = include_str!("obsidian_defaults/graph.json"); +const TYPES_JSON: &str = include_str!("obsidian_defaults/types.json"); + +/// Write the bundled `.obsidian/` defaults into `content_root` if they +/// aren't already there. Idempotent — never overwrites existing files. +pub fn ensure_obsidian_defaults(content_root: &Path) -> Result<()> { + let obsidian_dir = content_root.join(".obsidian"); + if let Err(err) = std::fs::create_dir_all(&obsidian_dir) { + log::warn!( + "[content_store::obsidian] create .obsidian dir failed at {:?}: {err:#} — skipping defaults", + obsidian_dir + ); + return Ok(()); + } + + write_default_if_missing(&obsidian_dir, "graph.json", GRAPH_JSON); + write_default_if_missing(&obsidian_dir, "types.json", TYPES_JSON); + Ok(()) +} + +fn write_default_if_missing(obsidian_dir: &Path, name: &str, body: &str) { + use std::io::{ErrorKind, Write}; + let target = obsidian_dir.join(name); + // `create_new(true)` makes existence-check + create atomic at the + // OS level, so a concurrent staging from another process can't + // race past `target.exists()` and clobber the winner. The + // AlreadyExists branch is the steady-state idempotent no-op. + let mut file = match std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&target) + { + Ok(f) => f, + Err(err) if err.kind() == ErrorKind::AlreadyExists => return, + Err(err) => { + log::warn!( + "[content_store::obsidian] create default {} failed at {:?}: {err:#}", + name, + target + ); + return; + } + }; + match file.write_all(body.as_bytes()) { + Ok(()) => log::info!( + "[content_store::obsidian] staged default {} at {}", + name, + target.display() + ), + Err(err) => { + // `create_new` already produced an empty file at `target`; + // a write_all failure (disk full, transient I/O) leaves a + // truncated remnant. Without cleanup, the next call hits + // the AlreadyExists fast-path and never repairs the bad + // file. Remove it so the next call retries cleanly. + if let Err(cleanup_err) = std::fs::remove_file(&target) { + log::warn!( + "[content_store::obsidian] cleanup partial default {} failed at {:?}: {cleanup_err:#}", + name, + target + ); + } + log::warn!( + "[content_store::obsidian] write default {} failed at {:?}: {err:#}", + name, + target + ); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + #[test] + fn stages_defaults_into_fresh_root() { + let tmp = TempDir::new().unwrap(); + ensure_obsidian_defaults(tmp.path()).unwrap(); + let graph = tmp.path().join(".obsidian").join("graph.json"); + let types = tmp.path().join(".obsidian").join("types.json"); + assert!(graph.exists(), "graph.json should be staged"); + assert!(types.exists(), "types.json should be staged"); + // Body must be the bundled content, not empty. + let g = std::fs::read_to_string(&graph).unwrap(); + assert!(g.contains("colorGroups"), "graph.json missing colorGroups"); + } + + #[test] + fn does_not_overwrite_existing_file() { + let tmp = TempDir::new().unwrap(); + let obs = tmp.path().join(".obsidian"); + std::fs::create_dir_all(&obs).unwrap(); + let graph = obs.join("graph.json"); + std::fs::write(&graph, r#"{"user":"custom"}"#).unwrap(); + + ensure_obsidian_defaults(tmp.path()).unwrap(); + + let body = std::fs::read_to_string(&graph).unwrap(); + assert_eq!( + body, r#"{"user":"custom"}"#, + "user-customised graph.json must not be clobbered" + ); + } + + #[test] + fn idempotent_second_call_is_no_op() { + let tmp = TempDir::new().unwrap(); + ensure_obsidian_defaults(tmp.path()).unwrap(); + ensure_obsidian_defaults(tmp.path()).unwrap(); + // Second call must succeed without panicking and must not have + // duplicated or grown the file. + let g = std::fs::read_to_string(tmp.path().join(".obsidian/graph.json")).unwrap(); + assert!(g.contains("colorGroups")); + } +} diff --git a/src/openhuman/memory/tree/content_store/obsidian_defaults/graph.json b/src/openhuman/memory/tree/content_store/obsidian_defaults/graph.json new file mode 100644 index 000000000..aa4bbd923 --- /dev/null +++ b/src/openhuman/memory/tree/content_store/obsidian_defaults/graph.json @@ -0,0 +1,65 @@ +{ + "collapse-filter": false, + "search": "", + "showTags": false, + "showAttachments": false, + "hideUnresolved": true, + "showOrphans": true, + "collapse-color-groups": false, + "colorGroups": [ + { + "query": "file:summary-L1", + "color": { + "a": 1, + "rgb": 14701138 + } + }, + { + "query": "file:summary-L2", + "color": { + "a": 1, + "rgb": 14725458 + } + }, + { + "query": "file:summary-L3", + "color": { + "a": 1, + "rgb": 11657298 + } + }, + { + "query": "file:summary-L4", + "color": { + "a": 1, + "rgb": 5420768 + } + }, + { + "query": "file:summary-L5", + "color": { + "a": 1, + "rgb": 5431504 + } + }, + { + "query": "file:summary-L6", + "color": { + "a": 1, + "rgb": 14701261 + } + } + ], + "collapse-display": false, + "showArrow": false, + "textFadeMultiplier": 0.9, + "nodeSizeMultiplier": 1.34371527777778, + "lineSizeMultiplier": 1.44048177083333, + "collapse-forces": false, + "centerStrength": 0.493880208333333, + "repelStrength": 10, + "linkStrength": 1, + "linkDistance": 250, + "scale": 0.5443310539518227, + "close": false +} \ No newline at end of file diff --git a/src/openhuman/memory/tree/content_store/obsidian_defaults/types.json b/src/openhuman/memory/tree/content_store/obsidian_defaults/types.json new file mode 100644 index 000000000..34f567621 --- /dev/null +++ b/src/openhuman/memory/tree/content_store/obsidian_defaults/types.json @@ -0,0 +1,10 @@ +{ + "types": { + "aliases": "aliases", + "cssclasses": "multitext", + "tags": "tags", + "time_range_end": "date", + "time_range_start": "date", + "sealed_at": "datetime" + } +} \ No newline at end of file diff --git a/src/openhuman/memory/tree/ingest.rs b/src/openhuman/memory/tree/ingest.rs index fb2054ccc..0312223a5 100644 --- a/src/openhuman/memory/tree/ingest.rs +++ b/src/openhuman/memory/tree/ingest.rs @@ -282,7 +282,7 @@ mod tests { drain_until_idle(&cfg).await.unwrap(); // Final lifecycle is `buffered`: extract → admitted → append_buffer → buffered. - // The single packed chunk does not cross TOKEN_BUDGET so no seal fires. + // The single packed chunk does not cross INPUT_TOKEN_BUDGET so no seal fires. assert_eq!( count_chunks_by_lifecycle_status(&cfg, CHUNK_STATUS_BUFFERED).unwrap(), 1 diff --git a/src/openhuman/memory/tree/jobs/handlers/mod.rs b/src/openhuman/memory/tree/jobs/handlers/mod.rs index eef71a7dd..a0e594bca 100644 --- a/src/openhuman/memory/tree/jobs/handlers/mod.rs +++ b/src/openhuman/memory/tree/jobs/handlers/mod.rs @@ -575,7 +575,7 @@ mod tests { } /// Seed a source tree and push enough labeled leaves into its L0 buffer - /// to cross `TOKEN_BUDGET`, returning the tree. The caller can then + /// to cross `INPUT_TOKEN_BUDGET`, returning the tree. The caller can then /// fire `handle_seal` and inspect the result. async fn seed_source_tree_ready_to_seal( cfg: &Config, @@ -599,7 +599,7 @@ mod tests { source_ref: Some(SourceRef::new("slack://x")), }, // Bust budget so the L0 buffer is "ready" for seal. - token_count: 10_000, + token_count: 60_000, seq_in_source: 0, created_at: ts, partial_message: false, @@ -619,7 +619,7 @@ mod tests { .unwrap(); let leaf = LeafRef { chunk_id: chunk.id, - token_count: 10_000, + token_count: 60_000, timestamp: ts, content: chunk.content, entities: vec![], @@ -672,7 +672,12 @@ mod tests { let p: TopicRoutePayload = serde_json::from_str(&payload_json).unwrap(); match p.node { NodeRef::Summary { summary_id } => { - assert!(summary_id.starts_with("summary:L1:")); + // Format: `summary:<13-digit-ms>:L-<8hex>` — + // see `tree_source::registry::new_summary_id`. + assert!( + summary_id.starts_with("summary:") && summary_id.contains(":L1-"), + "expected summary id with L1 segment, got {summary_id}" + ); } other => panic!("expected NodeRef::Summary, got {other:?}"), } @@ -707,7 +712,7 @@ mod tests { tags: vec![], source_ref: Some(SourceRef::new("slack://x")), }, - token_count: 10_000, + token_count: 60_000, seq_in_source: 0, created_at: ts, partial_message: false, @@ -727,7 +732,7 @@ mod tests { .unwrap(); let leaf = LeafRef { chunk_id: chunk.id, - token_count: 10_000, + token_count: 60_000, timestamp: ts, content: chunk.content, entities: vec![], @@ -794,7 +799,7 @@ mod tests { tags: vec![], source_ref: Some(SourceRef::new("slack://x")), }, - token_count: 6_000, + token_count: 30_000, seq_in_source: seq, created_at: ts, partial_message: false, @@ -812,7 +817,7 @@ mod tests { .unwrap(); let leaf = LeafRef { chunk_id: chunk.id, - token_count: 6_000, + token_count: 30_000, timestamp: ts, content: chunk.content, entities: vec![], diff --git a/src/openhuman/memory/tree/jobs/worker.rs b/src/openhuman/memory/tree/jobs/worker.rs index 5fa99ec9d..da4ffc877 100644 --- a/src/openhuman/memory/tree/jobs/worker.rs +++ b/src/openhuman/memory/tree/jobs/worker.rs @@ -20,7 +20,23 @@ use crate::openhuman::memory::tree::jobs::store::{ claim_next, mark_done, mark_failed, recover_stale_locks, DEFAULT_LOCK_DURATION_MS, }; -const WORKER_COUNT: usize = 1; +/// Number of concurrent job-worker tasks. Each worker claims one job +/// at a time via `claim_next` (atomic UPDATE under SQLite WAL with +/// `locked_until_ms` + status='running'), so multiple workers +/// parallelize independent jobs without double-claim risk. +/// +/// On cloud backends, LLM-bound jobs drop the global LLM permit +/// after claim (see `run_once`) so all 4 workers can run cloud +/// extract/summarise calls in parallel. +/// +/// On local backends, the single global LLM slot still serialises +/// Ollama calls for laptop-RAM safety. Note that `wait_for_capacity` +/// is acquired **before** `claim_next`, so non-LLM jobs (AppendBuffer, +/// FlushStale, TopicRoute) also block on the gate when an LLM job +/// holds the permit — they only run in parallel with each other while +/// no LLM job is in flight. Bumping `WORKER_COUNT` therefore helps +/// throughput most when local LLM calls are sparse. +const WORKER_COUNT: usize = 4; const POLL_INTERVAL: Duration = Duration::from_secs(5); static WORKER_NOTIFY: OnceLock> = OnceLock::new(); @@ -108,7 +124,19 @@ pub async fn run_once(config: &Config) -> Result { }; let llm_permit = if job.kind.is_llm_bound() { - gate_permit + // Local Ollama loads ~1.3 GB resident per concurrent call — + // hold the gate to enforce process-wide single-slot RAM + // safety. Cloud calls are bandwidth-bound, not RAM-bound: + // drop the permit so multiple workers can run cloud + // extract/summarise calls in parallel (the worker pool + // itself, sized to `WORKER_COUNT`, is the upstream bound). + match config.memory_tree.llm_backend { + crate::openhuman::config::LlmBackend::Local => gate_permit, + crate::openhuman::config::LlmBackend::Cloud => { + drop(gate_permit); + None + } + } } else { // Non-LLM jobs don't need the global slot; release it so an // LLM-bound caller waiting elsewhere in the process can run. diff --git a/src/openhuman/memory/tree/retrieval/drill_down.rs b/src/openhuman/memory/tree/retrieval/drill_down.rs index a1e56f9ed..f5190f324 100644 --- a/src/openhuman/memory/tree/retrieval/drill_down.rs +++ b/src/openhuman/memory/tree/retrieval/drill_down.rs @@ -301,7 +301,8 @@ mod tests { tags: vec![], source_ref: Some(SourceRef::new("slack://x")), }, - token_count: crate::openhuman::memory::tree::tree_source::types::TOKEN_BUDGET * 6 + token_count: crate::openhuman::memory::tree::tree_source::types::INPUT_TOKEN_BUDGET + * 6 / 10, seq_in_source: seq, created_at: ts, @@ -324,9 +325,9 @@ mod tests { &tree, &LeafRef { chunk_id: c.id.clone(), - token_count: crate::openhuman::memory::tree::tree_source::types::TOKEN_BUDGET - * 6 - / 10, + token_count: + crate::openhuman::memory::tree::tree_source::types::INPUT_TOKEN_BUDGET * 6 + / 10, timestamp: ts, content: c.content.clone(), entities: vec![], diff --git a/src/openhuman/memory/tree/retrieval/global.rs b/src/openhuman/memory/tree/retrieval/global.rs index dce81ae75..c150f42fd 100644 --- a/src/openhuman/memory/tree/retrieval/global.rs +++ b/src/openhuman/memory/tree/retrieval/global.rs @@ -148,7 +148,7 @@ mod tests { tags: vec![], source_ref: Some(SourceRef::new("slack://x")), }, - token_count: 6_000, + token_count: 30_000, seq_in_source: seq, created_at: ts, partial_message: false, @@ -160,7 +160,7 @@ mod tests { &tree, &LeafRef { chunk_id: c.id.clone(), - token_count: 6_000, + token_count: 30_000, timestamp: ts, content: c.content.clone(), entities: vec![], diff --git a/src/openhuman/memory/tree/retrieval/integration_test.rs b/src/openhuman/memory/tree/retrieval/integration_test.rs index 52df68e1c..3fd58822e 100644 --- a/src/openhuman/memory/tree/retrieval/integration_test.rs +++ b/src/openhuman/memory/tree/retrieval/integration_test.rs @@ -225,8 +225,8 @@ async fn seal_populates_summary_embedding() { created_at: ts, partial_message: false, }; - let c1 = mk_chunk(0, 6_000); - let c2 = mk_chunk(1, 6_000); + let c1 = mk_chunk(0, 30_000); + let c2 = mk_chunk(1, 30_000); upsert_chunks(&cfg, &[c1.clone(), c2.clone()]).unwrap(); { let content_root = cfg.memory_tree_content_root(); diff --git a/src/openhuman/memory/tree/retrieval/source.rs b/src/openhuman/memory/tree/retrieval/source.rs index a282f85af..85f6aac99 100644 --- a/src/openhuman/memory/tree/retrieval/source.rs +++ b/src/openhuman/memory/tree/retrieval/source.rs @@ -346,7 +346,8 @@ mod tests { tags: vec!["eng".into()], source_ref: Some(SourceRef::new(format!("slack://{scope}/{seq}"))), }, - token_count: crate::openhuman::memory::tree::tree_source::types::TOKEN_BUDGET * 6 + token_count: crate::openhuman::memory::tree::tree_source::types::INPUT_TOKEN_BUDGET + * 6 / 10, seq_in_source: seq, created_at: ts, @@ -369,9 +370,9 @@ mod tests { &tree, &LeafRef { chunk_id: c.id.clone(), - token_count: crate::openhuman::memory::tree::tree_source::types::TOKEN_BUDGET - * 6 - / 10, + token_count: + crate::openhuman::memory::tree::tree_source::types::INPUT_TOKEN_BUDGET * 6 + / 10, timestamp: ts, content: c.content.clone(), entities: vec![], diff --git a/src/openhuman/memory/tree/tree_global/digest_tests.rs b/src/openhuman/memory/tree/tree_global/digest_tests.rs index 2e528f12d..691c9711a 100644 --- a/src/openhuman/memory/tree/tree_global/digest_tests.rs +++ b/src/openhuman/memory/tree/tree_global/digest_tests.rs @@ -61,7 +61,7 @@ async fn seed_source_tree_with_sealed_l1(cfg: &Config, scope: &str, ts: DateTime tags: vec![], source_ref: Some(SourceRef::new("slack://x")), }, - token_count: 6_000, + token_count: 30_000, seq_in_source: 0, created_at: ts, partial_message: false, @@ -78,7 +78,7 @@ async fn seed_source_tree_with_sealed_l1(cfg: &Config, scope: &str, ts: DateTime tags: vec![], source_ref: Some(SourceRef::new("slack://y")), }, - token_count: 6_000, + token_count: 30_000, seq_in_source: 1, created_at: ts, partial_message: false, @@ -88,7 +88,7 @@ async fn seed_source_tree_with_sealed_l1(cfg: &Config, scope: &str, ts: DateTime let leaf1 = LeafRef { chunk_id: c1.id.clone(), - token_count: 6_000, + token_count: 30_000, timestamp: ts, content: c1.content.clone(), entities: vec![], @@ -97,7 +97,7 @@ async fn seed_source_tree_with_sealed_l1(cfg: &Config, scope: &str, ts: DateTime }; let leaf2 = LeafRef { chunk_id: c2.id.clone(), - token_count: 6_000, + token_count: 30_000, timestamp: ts, content: c2.content.clone(), entities: vec![], @@ -281,7 +281,7 @@ async fn seed_source_tree_with_labeled_l1( tags: topics.clone(), source_ref: Some(SourceRef::new(format!("slack://{scope}/{seq}"))), }, - token_count: 6_000, + token_count: 30_000, seq_in_source: seq, created_at: ts, partial_message: false, @@ -326,7 +326,7 @@ async fn seed_source_tree_with_labeled_l1( for chunk in &chunks { let leaf = LeafRef { chunk_id: chunk.id.clone(), - token_count: 6_000, + token_count: 30_000, timestamp: ts, content: chunk.content.clone(), entities: entities.clone(), diff --git a/src/openhuman/memory/tree/tree_global/recap.rs b/src/openhuman/memory/tree/tree_global/recap.rs index d4aefe9ca..611f08f71 100644 --- a/src/openhuman/memory/tree/tree_global/recap.rs +++ b/src/openhuman/memory/tree/tree_global/recap.rs @@ -230,7 +230,7 @@ mod tests { tags: vec![], source_ref: Some(SourceRef::new("slack://x")), }, - token_count: 6_000, + token_count: 30_000, seq_in_source: 0, created_at: ts, partial_message: false, @@ -247,7 +247,7 @@ mod tests { tags: vec![], source_ref: Some(SourceRef::new("slack://y")), }, - token_count: 6_000, + token_count: 30_000, seq_in_source: 1, created_at: ts, partial_message: false, @@ -259,7 +259,7 @@ mod tests { &tree, &LeafRef { chunk_id: c1.id.clone(), - token_count: 6_000, + token_count: 30_000, timestamp: ts, content: c1.content.clone(), entities: vec![], @@ -276,7 +276,7 @@ mod tests { &tree, &LeafRef { chunk_id: c2.id.clone(), - token_count: 6_000, + token_count: 30_000, timestamp: ts, content: c2.content.clone(), entities: vec![], diff --git a/src/openhuman/memory/tree/tree_source/README.md b/src/openhuman/memory/tree/tree_source/README.md index 434dc5cb8..161917935 100644 --- a/src/openhuman/memory/tree/tree_source/README.md +++ b/src/openhuman/memory/tree/tree_source/README.md @@ -10,7 +10,7 @@ Phase 3a (#709) — per-source summary trees with bucket-seal mechanics. One tre - `pub fn build_summariser` / `pub trait Summariser` / `pub struct SummaryInput` / `pub struct SummaryContext` / `pub struct SummaryOutput` — `summariser/mod.rs` — folds N inputs into one summary. - `pub struct InertSummariser` — `summariser/inert.rs` — deterministic dependency-free fallback. - `pub struct LlmSummariser` / `pub struct LlmSummariserConfig` — `summariser/llm.rs` — Ollama-backed implementation with soft-fallback to inert. -- `pub struct Tree` / `pub struct SummaryNode` / `pub struct Buffer` / `pub enum TreeKind` / `pub enum TreeStatus` / `pub const TOKEN_BUDGET` / `pub const SUMMARY_FANOUT` — `types.rs`. +- `pub struct Tree` / `pub struct SummaryNode` / `pub struct Buffer` / `pub enum TreeKind` / `pub enum TreeStatus` / `pub const INPUT_TOKEN_BUDGET` / `pub const OUTPUT_TOKEN_BUDGET` / `pub const SUMMARY_FANOUT` — `types.rs`. - `pub fn get_summary_embedding` / `pub fn set_summary_embedding` / `pub fn insert_tree` / `pub fn get_tree_by_scope` / `pub fn get_tree` / `pub fn list_trees_by_kind` / `pub fn get_summary` / `pub fn list_summaries_at_level` / `pub fn count_summaries` / `pub fn get_buffer` / `pub fn list_stale_buffers` — `store.rs`. ## Files diff --git a/src/openhuman/memory/tree/tree_source/bucket_seal.rs b/src/openhuman/memory/tree/tree_source/bucket_seal.rs index db189b1f2..9289fd0a5 100644 --- a/src/openhuman/memory/tree/tree_source/bucket_seal.rs +++ b/src/openhuman/memory/tree/tree_source/bucket_seal.rs @@ -3,7 +3,7 @@ //! `append_leaf` pushes a persisted chunk into the L0 buffer of a tree. //! Seal gates differ by level: //! -//! - **L0 (leaves → L1)**: seal when `token_sum >= TOKEN_BUDGET`. Bounds +//! - **L0 (leaves → L1)**: seal when `token_sum >= INPUT_TOKEN_BUDGET`. Bounds //! the summariser's raw input. //! - **L≥1 (summaries → next level)**: seal when `item_ids.len() >= //! SUMMARY_FANOUT`. Per-summary token size depends on summariser @@ -47,7 +47,7 @@ use crate::openhuman::memory::tree::tree_source::summariser::{ Summariser, SummaryContext, SummaryInput, }; use crate::openhuman::memory::tree::tree_source::types::{ - Buffer, SummaryNode, Tree, TreeKind, SUMMARY_FANOUT, TOKEN_BUDGET, + Buffer, SummaryNode, Tree, TreeKind, INPUT_TOKEN_BUDGET, OUTPUT_TOKEN_BUDGET, SUMMARY_FANOUT, }; /// Hard cap on cascade depth — prevents runaway loops if token accounting @@ -298,18 +298,22 @@ pub async fn cascade_all_from( /// Level-aware seal gate. /// -/// L0 buffers (raw leaves) gate on `token_sum` so the summariser's input -/// stays bounded. L≥1 buffers gate on sibling count so the tree's +/// 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). L≥1 buffers gate on sibling count alone so the tree's /// fan-in is independent of per-summary token size — without this, /// summarisers that emit at the full token budget (e.g. the inert -/// fallback) collapse the cascade into a 1:1:1 chain instead of a real -/// tree. +/// fallback) collapse the cascade into a 1:1:1 chain instead of a +/// real tree. pub(crate) fn should_seal(buf: &Buffer) -> bool { if buf.is_empty() { return false; } if buf.level == 0 { - buf.token_sum >= TOKEN_BUDGET as i64 + buf.token_sum >= INPUT_TOKEN_BUDGET as i64 || (buf.item_ids.len() as u32) >= SUMMARY_FANOUT } else { (buf.item_ids.len() as u32) >= SUMMARY_FANOUT } @@ -378,7 +382,7 @@ pub(crate) async fn seal_one_level( tree_id: &tree.id, tree_kind: tree.kind, target_level, - token_budget: TOKEN_BUDGET, + token_budget: OUTPUT_TOKEN_BUDGET, }; let output = summariser .summarise(&inputs, &ctx) @@ -399,11 +403,10 @@ pub(crate) async fn seal_one_level( // // Embedder context-window guard: `nomic-embed-text-v1.5` accepts // up to 8192 tokens of input. Summary content is bounded by - // `ctx.token_budget = TOKEN_BUDGET = 10_000` so a worst-case - // summary overshoots the embedder. We truncate the input passed - // to `embed()` to fit (the persisted summary content stays full; - // only the embedding's "view" of it is clamped). 6000 tokens - // leaves headroom for tokenizer differences across embedders. + // `ctx.token_budget = OUTPUT_TOKEN_BUDGET = 5_000` which fits, but + // we still truncate the input passed to `embed()` to leave + // headroom for tokenizer drift (the persisted summary content + // stays full; only the embedding's "view" of it is clamped). let embedder = build_embedder_from_config(config).context("build embedder during seal")?; // Conservative cap. Slack-style chat content (URLs, mentions, // emoji) tokenizes 2-4× higher than the 4-chars/token heuristic. @@ -501,11 +504,13 @@ pub(crate) async fn seal_one_level( // the raw archive file the chunk's body lives in — the email // chunk-store path `email//.md` no longer // exists, so `[[]]` would be an unresolved Obsidian - // link. The raw file basename `_` matches what - // `raw_store::write_raw_items` writes. L≥2 children are - // summary ids whose default `sanitize_filename` resolves to - // existing `wiki/summaries/...md` files — leave overrides - // unset there. + // link. We emit the relative path under content_root (with `.md` + // stripped) so the wikilink resolves unambiguously even outside + // Obsidian's unique-basename heuristic — e.g. + // `[[raw/gmail-stevent95-at-gmail-dot-com/_]]`. + // L≥2 children are summary ids whose default `sanitize_filename` + // resolves to existing `wiki/summaries/...md` files — leave + // overrides unset there. let child_basename_overrides: Option>> = if node.level == 1 { let overrides: Vec> = node .child_ids @@ -518,15 +523,28 @@ pub(crate) async fn seal_one_level( // takes the sanitised-id fallback) but a warn log // makes the SQL error visible for diagnosis. match crate::openhuman::memory::tree::store::get_chunk_raw_refs(config, chunk_id) { - Ok(refs_opt) => { - refs_opt - .and_then(|refs| refs.into_iter().next()) - .and_then(|r| { - std::path::Path::new(&r.path) - .file_stem() - .and_then(|s| s.to_str()) - .map(|s| s.to_string()) - }) + Ok(Some(refs)) if !refs.is_empty() => { + // RawRef::path is a forward-slash relative path + // under content_root, e.g. + // "raw/gmail-…/1700000_msg-id.md". Strip `.md` + // for Obsidian's extension-less wikilink + // convention. + let r = refs.into_iter().next().expect("non-empty"); + Some(r.path.strip_suffix(".md").unwrap_or(&r.path).to_string()) + } + Ok(_) => { + // No raw_refs persisted for this chunk — most + // commonly slack chunks (we only stage raw + // archive files for gmail today). The wikilink + // falls back to `sanitize_filename(chunk_id)`, + // which produces a deliberately-unresolved + // Obsidian link. Log so the silent-degradation + // path stays visible during diagnosis. + log::debug!( + "[tree_source::bucket_seal] no raw_refs for chunk_id={chunk_id} \ + — wikilink will fall back to sanitised chunk id" + ); + None } Err(e) => { log::warn!( @@ -562,6 +580,20 @@ pub(crate) async fn seal_one_level( // content_path = NULL. The buffer stays unsealed and the job-retry path // will re-attempt the file write on next execution. let content_root = config.memory_tree_content_root(); + // Drop the bundled `.obsidian/` defaults (graph + types) so a user + // opening the vault gets the intended graph-view colour mapping + // without manual configuration. Best-effort and idempotent — never + // overwrites an existing file. + if let Err(err) = + crate::openhuman::memory::tree::content_store::obsidian::ensure_obsidian_defaults( + &content_root, + ) + { + log::warn!( + "[tree_source::bucket_seal] ensure_obsidian_defaults failed: {err:#} — \ + continuing seal without vault defaults" + ); + } let staged = stage_summary(&content_root, &compose_input, &scope_slug, None).with_context(|| { format!( diff --git a/src/openhuman/memory/tree/tree_source/bucket_seal_tests.rs b/src/openhuman/memory/tree/tree_source/bucket_seal_tests.rs index d06f30fc6..24fe942a9 100644 --- a/src/openhuman/memory/tree/tree_source/bucket_seal_tests.rs +++ b/src/openhuman/memory/tree/tree_source/bucket_seal_tests.rs @@ -101,9 +101,9 @@ async fn crossing_budget_triggers_seal() { created_at: ts, partial_message: false, }; - // Budget-relative sizes so the test stays correct as TOKEN_BUDGET shifts: + // Budget-relative sizes so the test stays correct as INPUT_TOKEN_BUDGET shifts: // each leaf is 60% of budget, so the second append crosses the threshold. - let per_leaf = TOKEN_BUDGET * 6 / 10; + let per_leaf = INPUT_TOKEN_BUDGET * 6 / 10; let c1 = mk_chunk(0, per_leaf); let c2 = mk_chunk(1, per_leaf); upsert_chunks(&cfg, &[c1.clone(), c2.clone()]).unwrap(); @@ -200,10 +200,10 @@ async fn fanout_at_l1_triggers_l2_seal() { tags: vec![], source_ref: Some(SourceRef::new("slack://x")), }, - // Each leaf alone busts TOKEN_BUDGET so the L0→L1 seal + // Each leaf alone busts INPUT_TOKEN_BUDGET so the L0→L1 seal // fires on every append. After SUMMARY_FANOUT seals, the // L1 buffer's count-based gate trips and cascades to L2. - token_count: 10_000, + token_count: INPUT_TOKEN_BUDGET + 1, seq_in_source: seq, created_at: ts, partial_message: false, @@ -292,7 +292,7 @@ async fn upper_level_does_not_seal_below_fanout() { tags: vec![], source_ref: Some(SourceRef::new("slack://x")), }, - token_count: 10_000, + token_count: INPUT_TOKEN_BUDGET + 1, seq_in_source: seq, created_at: ts, partial_message: false, @@ -369,8 +369,8 @@ fn seed_leaf( tags: topics.clone(), source_ref: Some(SourceRef::new(format!("slack://x{seq}"))), }, - // Bust TOKEN_BUDGET in one leaf so the seal fires immediately. - token_count: 10_000, + // Bust INPUT_TOKEN_BUDGET in one leaf so the seal fires immediately. + token_count: INPUT_TOKEN_BUDGET + 1, seq_in_source: seq, created_at: ts, partial_message: false, @@ -490,7 +490,7 @@ async fn seal_with_union_strategy_inherits_labels_from_children() { // a third append triggers a seal containing both. Reuse the helper // and override the leaf's token_count for this test. // Each leaf at half the budget so two together hit threshold exactly. - let per_leaf = TOKEN_BUDGET / 2; + let per_leaf = INPUT_TOKEN_BUDGET / 2; let leaf1 = LeafRef { token_count: per_leaf, ..leaf1 diff --git a/src/openhuman/memory/tree/tree_source/flush.rs b/src/openhuman/memory/tree/tree_source/flush.rs index 67daed199..4bfc7e7bc 100644 --- a/src/openhuman/memory/tree/tree_source/flush.rs +++ b/src/openhuman/memory/tree/tree_source/flush.rs @@ -1,10 +1,12 @@ //! Time-based buffer flush for source trees (#709). //! -//! The bucket-seal path only fires when a buffer crosses `TOKEN_BUDGET`. -//! Low-volume sources (e.g. an email account with two threads a week) can -//! otherwise leave leaves parked in the L0 buffer indefinitely, which -//! hurts recall. `flush_stale_buffers` force-seals any buffer whose -//! `oldest_at` is older than `max_age`, regardless of token count. +//! The bucket-seal path only fires when a buffer crosses +//! `INPUT_TOKEN_BUDGET` (token volume) or `SUMMARY_FANOUT` (item count +//! — the L0 fallback gate). Low-volume sources (e.g. an email account +//! with two threads a week) can still park a buffer below both +//! thresholds indefinitely, which hurts recall. +//! `flush_stale_buffers` force-seals any buffer whose `oldest_at` is +//! older than `max_age`, regardless of token count or item count. //! //! This is meant to run on a cadence (e.g. daily cron). Phase 3a ships //! the primitive; wiring into a scheduler is not required for merge. diff --git a/src/openhuman/memory/tree/tree_source/mod.rs b/src/openhuman/memory/tree/tree_source/mod.rs index 1f9694468..796f74c3a 100644 --- a/src/openhuman/memory/tree/tree_source/mod.rs +++ b/src/openhuman/memory/tree/tree_source/mod.rs @@ -26,4 +26,7 @@ pub use bucket_seal::{append_leaf, append_leaf_deferred, LabelStrategy, LeafRef} pub use registry::get_or_create_source_tree; pub use store::{get_summary_embedding, set_summary_embedding}; pub use summariser::{build_summariser, inert::InertSummariser, llm::LlmSummariser, Summariser}; -pub use types::{Buffer, SummaryNode, Tree, TreeKind, TreeStatus, TOKEN_BUDGET}; +pub use types::{ + Buffer, SummaryNode, Tree, TreeKind, TreeStatus, INPUT_TOKEN_BUDGET, OUTPUT_TOKEN_BUDGET, + SUMMARY_FANOUT, +}; diff --git a/src/openhuman/memory/tree/tree_source/registry.rs b/src/openhuman/memory/tree/tree_source/registry.rs index 7c0c69939..3ab59381e 100644 --- a/src/openhuman/memory/tree/tree_source/registry.rs +++ b/src/openhuman/memory/tree/tree_source/registry.rs @@ -84,10 +84,21 @@ fn new_tree_id(kind: TreeKind) -> String { } /// Public id generator for summary nodes — exported so `bucket_seal` can -/// share the same format (kept separate for readability; both use UUID v4 -/// suffixes to keep ids short but unambiguous). +/// share the same format. The Unix-ms timestamp is the leading sort +/// key so `ORDER BY id` is globally chronological across all levels +/// (a level-first layout grouped L1, L2, … together, breaking that). +/// `:013` zero-pads the millisecond field to 13 digits so the +/// lexicographic order matches numeric order through year 2286 — well +/// outside any reasonable retention window. Level is suffixed for +/// filter-by-level queries (`LIKE '%:L1-%'`). 8-hex of `u32` entropy +/// shrinks same-millisecond collision probability to ~2⁻³² per pair, +/// sized for uniqueness across the file-system and Obsidian wikilink +/// namespaces. pub fn new_summary_id(level: u32) -> String { - format!("summary:L{}:{}", level, Uuid::new_v4()) + use rand::Rng; + let ms = chrono::Utc::now().timestamp_millis() as u64; + let rand_tail: u32 = rand::thread_rng().gen(); + format!("summary:{:013}:L{}-{:08x}", ms, level, rand_tail) } #[cfg(test)] @@ -126,7 +137,43 @@ mod tests { let id = new_tree_id(TreeKind::Source); assert!(id.starts_with("source:")); let sum_id = new_summary_id(3); - assert!(sum_id.starts_with("summary:L3:")); + assert!(sum_id.starts_with("summary:")); + // Time-first layout: the segment after `summary:` is a 13-digit + // zero-padded ms timestamp, then `:L-<8hex>`. + assert!(sum_id.contains(":L3-"), "expected level suffix in {sum_id}"); + } + + #[test] + fn summary_id_format_is_lexicographically_chronological() { + // The prefix `summary:` is identical across all ids, so the + // first character that differs is in the 13-digit ms field. + // Comparing two synthesised ids built around the same ms +/- a + // step proves the format sorts by time without depending on + // wall-clock granularity in the test runner. We verify the + // generator's _format_ (the contract), not the system clock. + let earlier_ms: u64 = 1_700_000_000_000; + let later_ms: u64 = 1_700_000_000_001; + // Use a max-tail rand for the earlier id to prove the + // millisecond field dominates over the random suffix. + let earlier = format!("summary:{:013}:L1-{:08x}", earlier_ms, u32::MAX); + let later = format!("summary:{:013}:L9-{:08x}", later_ms, 0u32); + assert!( + earlier < later, + "expected {earlier} < {later} (ms must outrank level + tail)" + ); + + // Sanity: a real id from the live generator parses with the + // same prefix shape so the contract above maps onto runtime + // values, not just synthesised strings. + let live = new_summary_id(2); + assert!(live.starts_with("summary:"), "live: {live}"); + let rest = &live["summary:".len()..]; + let ms_part = rest.split(':').next().expect("ms segment"); + assert_eq!(ms_part.len(), 13, "ms must be 13 digits in {live}"); + assert!( + ms_part.chars().all(|c| c.is_ascii_digit()), + "ms must be all digits in {live}" + ); } #[test] diff --git a/src/openhuman/memory/tree/tree_source/summariser/llm.rs b/src/openhuman/memory/tree/tree_source/summariser/llm.rs index b4d927afb..06a09d45a 100644 --- a/src/openhuman/memory/tree/tree_source/summariser/llm.rs +++ b/src/openhuman/memory/tree/tree_source/summariser/llm.rs @@ -38,7 +38,6 @@ use anyhow::Result; use async_trait::async_trait; -use serde::Deserialize; use std::sync::Arc; use super::inert::InertSummariser; @@ -48,37 +47,28 @@ use crate::openhuman::memory::tree::types::approx_token_count; /// Hard cap on summariser output length (in approximate tokens). /// -/// Two constraints set this: -/// -/// 1. The downstream embedder (`nomic-embed-text-v1.5`) accepts up to -/// 8192 tokens, and Phase 4 (`tree_source::bucket_seal`) embeds the -/// summary right after we produce it. An overshoot returns HTTP 500 -/// and rolls back the whole seal transaction. -/// 2. Empirically, small instruction-tuned models running locally -/// degrade quickly past ~3500 tokens — they drift, hallucinate, or -/// produce repetitive boilerplate as they extend toward longer -/// targets. Keeping the cap below that breakeven keeps output -/// quality stable on local Ollama deployments. -/// -/// 3500 sits comfortably under the embedder ceiling AND below the local -/// LLM quality cliff. The post-generation [`clamp_to_budget`] enforces -/// this regardless of what the model produces. -const MAX_SUMMARY_OUTPUT_TOKENS: u32 = 3_500; +/// Sized to fit the downstream embedder (`nomic-embed-text-v1.5`, +/// 8192-token input ceiling) with headroom for tokenizer drift between +/// our 4-chars/token heuristic and the embedder's real tokenizer. The +/// post-generation [`clamp_to_budget`] enforces this regardless of what +/// the model produces. +const MAX_SUMMARY_OUTPUT_TOKENS: u32 = 5_000; -/// Context window assumed for the model. Used as the divisor in the -/// per-input clamp so the joined prompt body stays under this even at -/// upper-level seals where SUMMARY_FANOUT children each near -/// MAX_SUMMARY_OUTPUT_TOKENS would otherwise overflow. Conservative — -/// real cloud models have larger contexts; smaller local models may -/// truncate, but the post-generation `clamp_to_budget` ensures output -/// fits the embedder regardless. -const NUM_CTX_TOKENS: u32 = 16_384; +/// Context window assumed for the model. Sized for the cloud +/// summariser's 120k-token window with comfortable headroom — leaves +/// room for the joined L0 input batch (up to `INPUT_TOKEN_BUDGET = 50k`), +/// the requested output budget, the system prompt, and tokenizer drift. +/// Used as the divisor in the per-input clamp so the joined prompt body +/// stays under this even at upper-level seals where many children fold +/// together. +const NUM_CTX_TOKENS: u32 = 60_000; -/// Tokens reserved for the system prompt, JSON wrapper, and tokenizer -/// drift between our 4-chars/token heuristic and the model's tokenizer. -/// Trades a small loss of input capacity for a guarantee that the -/// prompt body + output budget never exceeds `num_ctx`. -const OVERHEAD_RESERVE_TOKENS: u32 = 512; +/// Tokens reserved for the system prompt, message-envelope overhead, +/// and tokenizer drift between our 4-chars/token heuristic and the +/// model's tokenizer. Trades a small loss of input capacity for a +/// guarantee that the prompt body + output budget never exceeds +/// `num_ctx`. +const OVERHEAD_RESERVE_TOKENS: u32 = 2_048; /// Configuration for [`LlmSummariser`]. Threaded down to the chat /// provider for diagnostic logging — model selection at the wire level @@ -189,7 +179,7 @@ impl Summariser for LlmSummariser { ctx.token_budget ); - let raw = match self.provider.chat_for_json(&prompt).await { + let raw = match self.provider.chat_for_text(&prompt).await { Ok(v) => v, Err(e) => { log::warn!( @@ -203,19 +193,7 @@ impl Summariser for LlmSummariser { } }; - let parsed: LlmSummaryOutput = match serde_json::from_str(&raw) { - Ok(v) => v, - Err(e) => { - log::warn!( - "[tree_source::summariser::llm] model returned non-JSON or wrong-shape \ - body: {e}; content was: {} — falling back to inert", - truncate_for_log(&raw, 400) - ); - return self.fallback.summarise(inputs, ctx).await; - } - }; - - let (content, token_count) = clamp_to_budget(&parsed.summary, effective_budget); + let (content, token_count) = clamp_to_budget(raw.trim(), effective_budget); log::debug!( "[tree_source::summariser::llm] sealed tree_id={} level={} inputs={} tokens={}", ctx.tree_id, @@ -257,7 +235,7 @@ fn build_user_prompt(inputs: &[SummaryInput], per_input_cap_tokens: u32) -> Stri out } -/// System prompt. Length isn't templated in — empirically, telling small +/// System prompt. Length isn't templated in — empirically, telling /// instruction-tuned models "stay under N tokens" makes them produce /// curt, generic output even when the input has plenty of substance. /// Output is clamped post-generation by [`clamp_to_budget`] in the @@ -267,11 +245,9 @@ fn system_prompt(_budget: u32) -> String { single cohesive passage that preserves concrete facts, decisions, \ and temporal ordering. Do not invent facts.\n\ \n\ - Return JSON only — no prose, no markdown, no commentary. Schema:\n\ - {\n\ - \x20 \"summary\": \"\"\n\ - }" - .to_string() + Return only the summary text. No commentary, no preamble, no headings, \ + no markdown wrappers, no JSON — just the prose summary." + .to_string() } /// Truncate to the caller's token budget using the same ~4 chars/token @@ -287,22 +263,6 @@ fn clamp_to_budget(text: &str, budget: u32) -> (String, u32) { (truncated, tokens) } -fn truncate_for_log(s: &str, max_chars: usize) -> String { - if s.chars().count() <= max_chars { - return s.to_string(); - } - let truncated: String = s.chars().take(max_chars).collect(); - format!("{truncated}…") -} - -// ── LLM JSON output ────────────────────────────────────────────────────── - -#[derive(Debug, Deserialize)] -struct LlmSummaryOutput { - #[serde(default)] - summary: String, -} - #[cfg(test)] mod tests { use super::*; @@ -385,16 +345,17 @@ mod tests { } #[test] - fn system_prompt_describes_schema() { - // Budget is no longer templated into the prompt — small models + fn system_prompt_describes_plain_text_output() { + // Budget is no longer templated into the prompt — models // produced overly curt output when told to "stay under N tokens". // The clamp in `clamp_to_budget` handles enforcement instead. let p = system_prompt(4096); assert!(!p.contains("4096")); assert!(!p.contains("Stay well under")); - assert!(p.contains("\"summary\"")); - assert!(!p.contains("\"entities\"")); - assert!(!p.contains("\"topics\"")); + // Output is plain prose, not JSON. + assert!(!p.contains("\"summary\"")); + assert!(p.to_lowercase().contains("no commentary")); + assert!(p.to_lowercase().contains("no json")); } #[test] @@ -412,19 +373,6 @@ mod tests { assert!(t <= 6); } - #[test] - fn truncate_for_log_short_input_unchanged() { - assert_eq!(truncate_for_log("hi", 10), "hi"); - } - - #[test] - fn truncate_for_log_long_input_appends_ellipsis() { - let long = "x".repeat(500); - let out = truncate_for_log(&long, 10); - assert_eq!(out.chars().count(), 11); - assert!(out.ends_with('…')); - } - /// Mock chat provider that lets us assert prompt shape and stub responses /// in summariser unit tests without hitting the network. struct StubProvider { @@ -493,35 +441,22 @@ mod tests { assert!(out.topics.is_empty()); } - #[tokio::test] - async fn malformed_response_falls_back_to_inert() { - // Provider returns garbage → parse fails → fallback to inert. - let provider = std::sync::Arc::new(StubProvider::ok("not json")); - let s = LlmSummariser::new(LlmSummariserConfig::default(), provider); - let inputs = vec![sample_input("a", "alice ships friday")]; - let out = s.summarise(&inputs, &test_ctx()).await.unwrap(); - // Inert fallback content includes the original input. - assert!(out.content.contains("alice")); - } - #[tokio::test] async fn provider_summary_response_is_used_and_clamped() { - // Provider returns valid JSON; summariser parses it and clamps to - // the budget. - let provider = std::sync::Arc::new(StubProvider::ok( - r#"{"summary":"alice decided to ship friday"}"#, - )); + // Provider returns plain text; summariser uses it verbatim + // (after trim) and clamps to the budget. + let provider = std::sync::Arc::new(StubProvider::ok("alice decided to ship friday\n")); let s = LlmSummariser::new(LlmSummariserConfig::default(), provider.clone()); let inputs = vec![sample_input("a", "alice ships friday")]; let out = s.summarise(&inputs, &test_ctx()).await.unwrap(); - assert!(out.content.contains("alice decided to ship")); + assert_eq!(out.content, "alice decided to ship friday"); assert!(out.token_count > 0); assert_eq!(provider.calls.load(std::sync::atomic::Ordering::SeqCst), 1); } #[test] fn build_prompt_carries_body_and_kind_tag() { - let provider = std::sync::Arc::new(StubProvider::ok("{}")); + let provider = std::sync::Arc::new(StubProvider::ok("hi")); let s = LlmSummariser::new( LlmSummariserConfig { model: "llama3.1:8b".into(), @@ -529,26 +464,10 @@ mod tests { provider, ); let prompt = s.build_prompt("body", 2048); - assert!(prompt.system.contains("\"summary\"")); - assert!(!prompt.system.contains("\"entities\"")); + assert!(prompt.system.to_lowercase().contains("no commentary")); + assert!(!prompt.system.contains("\"summary\"")); assert_eq!(prompt.user, "body"); assert_eq!(prompt.temperature, 0.0); assert_eq!(prompt.kind, "memory_tree::summarise"); } - - #[test] - fn llm_output_deserialises_with_only_summary() { - let v: LlmSummaryOutput = serde_json::from_str(r#"{"summary":"hi"}"#).unwrap(); - assert_eq!(v.summary, "hi"); - } - - #[test] - fn llm_output_ignores_extraneous_fields() { - // Prompt no longer asks for entities/topics, but if the model - // emits them anyway we should still parse `summary` cleanly. - let v: LlmSummaryOutput = - serde_json::from_str(r#"{"summary":"hi","entities":["Alice"],"topics":["x"]}"#) - .unwrap(); - assert_eq!(v.summary, "hi"); - } } diff --git a/src/openhuman/memory/tree/tree_source/types.rs b/src/openhuman/memory/tree/tree_source/types.rs index 2b3b72790..4e7c652b7 100644 --- a/src/openhuman/memory/tree/tree_source/types.rs +++ b/src/openhuman/memory/tree/tree_source/types.rs @@ -178,27 +178,31 @@ impl Buffer { } } -/// Token ceiling for one summariser invocation. +/// Input token target for one L0 → L1 seal: when an L0 buffer's +/// `token_sum` reaches this, we summarise the accumulated leaves. /// -/// Sized for the local 1B summariser (`gemma3:1b-it-qat`), which produces -/// noticeably better summaries with ≤4-5k input than at higher caps. The -/// chunker's `DEFAULT_CHUNK_MAX_TOKENS` (3_000) sits below this so each -/// L0 buffer accumulates roughly 1-3 chunks before sealing. -/// -/// Gates only the L0 → L1 seal: leaves are fan-in by raw token volume so -/// the summariser input stays bounded. Summaries above L0 use -/// [`SUMMARY_FANOUT`] instead — see `bucket_seal::should_seal`. -pub const TOKEN_BUDGET: u32 = 4_500; +/// Sized for the cloud summariser's 120k-token context with headroom for +/// the system prompt and the model's own output. With ~5k tokens emitted +/// per summary (see [`OUTPUT_TOKEN_BUDGET`]), one parent represents ~50k +/// tokens of leaf content — i.e. ~10 child summaries' worth. +pub const INPUT_TOKEN_BUDGET: u32 = 50_000; + +/// Output token budget passed to the summariser as `ctx.token_budget`. +/// The summariser may clamp lower (see `summariser/llm.rs`'s +/// `MAX_SUMMARY_OUTPUT_TOKENS`). 5k keeps the produced summary well +/// under the embedder's 8k input ceiling so the post-seal embed never +/// rejects the row. +pub const OUTPUT_TOKEN_BUDGET: u32 = 5_000; /// Sibling count that triggers a seal at level ≥ 1 (summaries → next level). /// -/// Decouples upper-level seals from per-summary token size so the tree's -/// fan-in stays stable regardless of summariser quality. With a real -/// summariser each L1 might be ~500 tokens; with the inert fallback each -/// L1 fills the full [`TOKEN_BUDGET`]. Token-based gating collapses the -/// inert case into a 1:1:1 chain — count-based gating gives a real tree -/// shape (`SUMMARY_FANOUT` children per parent) in both cases. -pub const SUMMARY_FANOUT: u32 = 4; +/// Set to match the [`INPUT_TOKEN_BUDGET`] / [`OUTPUT_TOKEN_BUDGET`] +/// ratio so each level folds roughly the same volume of content as L0: +/// 10 summaries × ~5k tokens ≈ 50k input. Decouples upper-level seals +/// from per-summary token size so the tree's fan-in stays stable +/// regardless of summariser quality (token-based gating would collapse +/// the inert-fallback case into a 1:1:1 chain). +pub const SUMMARY_FANOUT: u32 = 10; /// Default age at which a non-empty buffer is force-sealed even under the /// token budget. Keeps recent activity from stalling waiting for more diff --git a/src/openhuman/scheduler_gate/gate.rs b/src/openhuman/scheduler_gate/gate.rs index 826ca5545..157de0b62 100644 --- a/src/openhuman/scheduler_gate/gate.rs +++ b/src/openhuman/scheduler_gate/gate.rs @@ -17,15 +17,15 @@ use crate::openhuman::scheduler_gate::signals::Signals; /// Process-wide ceiling on concurrent LLM-bound work. /// /// Held at 1 to keep concurrent local-Ollama / bge-m3 calls (8K context, -/// ~1.3 GB resident each) from saturating local RAM. The cloud path -/// itself is bandwidth-bound but the gate also fronts triage and -/// reflection turns that fan out to `agent.run_turn`, where every local -/// route loads the same Ollama model — so a single global slot is the -/// safest contract until #1064 (per-toolkit triage toggle) and the -/// cloud-side rate limiter ship. -/// -/// See `feedback_local_llm_load.md` — backfills with multiple +/// ~1.3 GB resident each) from saturating local RAM. See +/// `feedback_local_llm_load.md` — backfills with multiple /// simultaneous Ollama requests have crashed the user's laptop twice. +/// +/// Cloud-backend LLM calls bypass this semaphore at the worker layer +/// (see `memory_tree::jobs::worker::run_once`) because they're +/// bandwidth-bound, not RAM-bound, and the worker pool itself bounds +/// concurrency upstream. Keeping this at 1 preserves the laptop-RAM +/// contract regardless of backend. const LLM_SLOTS: usize = 1; static LLM_PERMITS: OnceLock> = OnceLock::new();