mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
feat(memory_tree): scale summariser for cloud — bigger batches, parallel workers, prose output (#1348)
This commit is contained in:
Steven Enamakel
Generated
+1
-1
@@ -4520,7 +4520,7 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381"
|
||||
|
||||
[[package]]
|
||||
name = "openhuman"
|
||||
version = "0.53.18"
|
||||
version = "0.53.19"
|
||||
dependencies = [
|
||||
"aes-gcm",
|
||||
"anyhow",
|
||||
|
||||
Generated
+2
-2
@@ -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",
|
||||
|
||||
Executable
+207
@@ -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
|
||||
@@ -62,6 +62,20 @@ impl ChatProvider for OllamaChatProvider {
|
||||
}
|
||||
|
||||
async fn chat_for_json(&self, prompt: &ChatPrompt) -> Result<String> {
|
||||
self.run_chat(prompt, Some("json")).await
|
||||
}
|
||||
|
||||
async fn chat_for_text(&self, prompt: &ChatPrompt) -> Result<String> {
|
||||
// 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<String> {
|
||||
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<OllamaMessage>,
|
||||
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<String>,
|
||||
stream: bool,
|
||||
options: OllamaOptions,
|
||||
}
|
||||
|
||||
@@ -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<String>;
|
||||
|
||||
/// 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<String> {
|
||||
self.chat_for_json(prompt).await
|
||||
}
|
||||
}
|
||||
|
||||
/// Build the [`ChatProvider`] dictated by `config.memory_tree.llm_backend`.
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -14,6 +14,7 @@
|
||||
|
||||
pub mod atomic;
|
||||
pub mod compose;
|
||||
pub mod obsidian;
|
||||
pub mod paths;
|
||||
pub mod raw;
|
||||
pub mod read;
|
||||
|
||||
@@ -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/<file>` 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"));
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
{
|
||||
"types": {
|
||||
"aliases": "aliases",
|
||||
"cssclasses": "multitext",
|
||||
"tags": "tags",
|
||||
"time_range_end": "date",
|
||||
"time_range_start": "date",
|
||||
"sealed_at": "datetime"
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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<level>-<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![],
|
||||
|
||||
@@ -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<Arc<Notify>> = OnceLock::new();
|
||||
@@ -108,7 +124,19 @@ pub async fn run_once(config: &Config) -> Result<bool> {
|
||||
};
|
||||
|
||||
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.
|
||||
|
||||
@@ -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![],
|
||||
|
||||
@@ -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![],
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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![],
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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![],
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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/<scope>/<chunk_id>.md` no longer
|
||||
// exists, so `[[<chunk_id>]]` would be an unresolved Obsidian
|
||||
// link. The raw file basename `<ts_ms>_<msg_id>` 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/<ts_ms>_<msg_id>]]`.
|
||||
// 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<Vec<Option<String>>> = if node.level == 1 {
|
||||
let overrides: Vec<Option<String>> = 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!(
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
};
|
||||
|
||||
@@ -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<level>-<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]
|
||||
|
||||
@@ -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\": \"<summary body>\"\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");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<Arc<Semaphore>> = OnceLock::new();
|
||||
|
||||
Reference in New Issue
Block a user