diff --git a/crates/tinymemory-api/src/conformance/reference/mod.rs b/crates/tinymemory-api/src/conformance/reference/mod.rs index 59570b8c..d9783ab3 100644 --- a/crates/tinymemory-api/src/conformance/reference/mod.rs +++ b/crates/tinymemory-api/src/conformance/reference/mod.rs @@ -11,6 +11,7 @@ mod distil; mod score; +use std::collections::HashSet; use std::sync::Mutex; pub use distil::CONSOLIDATED_TAG; @@ -213,6 +214,26 @@ impl MemoryEngine for ReferenceEngine { Ok(StoreReceipt { id, replayed }) } + async fn store_many(&self, batch: Vec) -> Result> { + crate::validate_many(&batch)?; + let mut items = self.items()?; + // Compute held fingerprints once per batch. Calling `store` for each + // item rehashes every preceding item, which makes scale sweeps + // quadratic in the number of documents. + let mut seen: HashSet = items.iter().map(StoreItem::fingerprint).collect(); + let mut receipts = Vec::with_capacity(batch.len()); + for item in batch { + item.validate()?; + let id = ItemId(item.fingerprint()); + let replayed = !seen.insert(id.0.clone()); + if !replayed { + items.push(item); + } + receipts.push(StoreReceipt { id, replayed }); + } + Ok(receipts) + } + async fn forget(&self, target: ForgetTarget) -> Result { target.validate()?; let mut items = self.items()?; diff --git a/crates/tinymemory-api/src/conformance/reference/mod_tests.rs b/crates/tinymemory-api/src/conformance/reference/mod_tests.rs index 48a2cf42..b262ad3c 100644 --- a/crates/tinymemory-api/src/conformance/reference/mod_tests.rs +++ b/crates/tinymemory-api/src/conformance/reference/mod_tests.rs @@ -4,6 +4,37 @@ use crate::{ItemKind, LearningKind, MemoryMeta}; use super::*; +#[tokio::test] +async fn bulk_store_preserves_order_and_marks_existing_and_in_batch_replays() { + let engine = ReferenceEngine::new(); + let a = StoreItem::document("alpha", MemoryMeta::default()); + let b = StoreItem::document("beta", MemoryMeta::default()); + engine.store(a.clone()).await.unwrap(); + let receipts = engine + .store_many(vec![a.clone(), b.clone(), b.clone()]) + .await + .unwrap(); + assert_eq!( + receipts + .iter() + .map(|receipt| receipt.replayed) + .collect::>(), + vec![true, false, true] + ); + assert_eq!(engine.len(), 2); + let page = engine + .export(ListRequest::new(MetaFilter::default(), 10)) + .await + .unwrap(); + assert_eq!( + page.items + .into_iter() + .map(|entry| entry.item) + .collect::>(), + vec![a, b] + ); +} + #[tokio::test] async fn a_bad_cursor_is_an_invalid_request() { let engine = ReferenceEngine::new(); diff --git a/crates/tinymemory-integrations/examples/memory_eval/agent.rs b/crates/tinymemory-integrations/examples/memory_eval/agent.rs index 61ad7b3f..4c6e7b85 100644 --- a/crates/tinymemory-integrations/examples/memory_eval/agent.rs +++ b/crates/tinymemory-integrations/examples/memory_eval/agent.rs @@ -24,6 +24,36 @@ use tinymemory_tools::{AgentMemory, BackgroundJob, PostTurn, PreTurn, TurnContex /// OpenHuman's default deadline for the pack before a model turn. pub(crate) const PRE_TURN_TIMEOUT: StdDuration = StdDuration::from_millis(1_500); + +/// The host selects one lifecycle hook from its recall configuration and +/// whether the thread is resuming after compaction. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum HostHook { + Plain, + Resumed, + Dated, + DatedResumed, +} + +impl HostHook { + pub(crate) fn for_turn(resumed: bool, date_hint: bool) -> Self { + match (resumed, date_hint) { + (false, false) => Self::Plain, + (true, false) => Self::Resumed, + (false, true) => Self::Dated, + (true, true) => Self::DatedResumed, + } + } + + pub(crate) async fn run(self, memory: AgentMemory, pre: PreTurn) -> Result { + match self { + Self::Plain => memory.pre_turn(pre).await, + Self::Resumed => memory.pre_turn_resumed(pre).await, + Self::Dated => memory.pre_turn_dated(pre, false, async { None }).await, + Self::DatedResumed => memory.pre_turn_dated(pre, true, async { None }).await, + } + } +} /// OpenHuman's maximum length of one logged tool result. const MAX_TOOL_LINE_CHARS: usize = 240; @@ -59,6 +89,11 @@ pub(crate) struct TurnRecord { pub(crate) tool_calls: usize, } +type PendingPreTurn = ( + Instant, + tokio::task::JoinHandle<(Result, Instant)>, +); + /// One conversation thread driven by the script. pub(crate) struct ScriptedAgent { memory: AgentMemory, @@ -68,7 +103,14 @@ pub(crate) struct ScriptedAgent { window: u32, clock: Option>, openhuman: bool, - pending: Vec>>, + date_hint: bool, + pending: Vec, +} + +/// User turns that completed after the simulated host stopped waiting. +pub(crate) struct FlushReport { + pub(crate) logged: usize, + pub(crate) completion_ms: Vec, } impl ScriptedAgent { @@ -82,27 +124,40 @@ impl ScriptedAgent { window, clock: None, openhuman: false, + date_hint: false, pending: Vec::new(), } } - /// Mirror OpenHuman's dated hook, deadline, and reply logging. + /// Mirror OpenHuman's default hook, deadline, and reply logging. pub(crate) fn openhuman(mut self) -> Self { self.openhuman = true; self } + /// Exercise the host's optional dated-recall path. + pub(crate) fn date_hint(mut self) -> Self { + self.date_hint = true; + self + } + /// Wait for pre-turn tasks whose host deadline expired. OpenHuman leaves /// those tasks running, so their accepted user turns may still land. - pub(crate) async fn flush(&mut self) -> Result { + pub(crate) async fn flush(&mut self) -> Result { let mut logged = 0; - for task in self.pending.drain(..) { - let context = task + let mut completion_ms = Vec::new(); + for (started, task) in self.pending.drain(..) { + let (context, completed) = task .await - .map_err(|error| Error::Unavailable(format!("pre-turn task failed: {error}")))??; + .map_err(|error| Error::Unavailable(format!("pre-turn task failed: {error}")))?; + let context = context?; logged += usize::from(context.logged.is_some()); + completion_ms.push(completed.duration_since(started).as_secs_f64() * 1_000.0); } - Ok(logged) + Ok(FlushReport { + logged, + completion_ms, + }) } /// Timestamps the thread's turns from `at`, a minute apart. @@ -136,16 +191,21 @@ impl ScriptedAgent { let started = Instant::now(); let context = if self.openhuman { let memory = self.memory.clone(); - let mut task = - tokio::spawn( - async move { memory.pre_turn_dated(pre, false, async { None }).await }, - ); + let hook = HostHook::for_turn(false, self.date_hint); + let mut task = tokio::spawn(async move { + let result = hook.run(memory, pre).await; + (result, Instant::now()) + }); match tokio::time::timeout(PRE_TURN_TIMEOUT, &mut task).await { - Ok(result) => Some(result.map_err(|error| { - Error::Unavailable(format!("pre-turn task failed: {error}")) - })??), + Ok(result) => Some( + result + .map_err(|error| { + Error::Unavailable(format!("pre-turn task failed: {error}")) + })? + .0?, + ), Err(_) => { - self.pending.push(task); + self.pending.push((started, task)); None } } diff --git a/crates/tinymemory-integrations/examples/memory_eval/agent_tests.rs b/crates/tinymemory-integrations/examples/memory_eval/agent_tests.rs index 90b09b2e..2b032e4c 100644 --- a/crates/tinymemory-integrations/examples/memory_eval/agent_tests.rs +++ b/crates/tinymemory-integrations/examples/memory_eval/agent_tests.rs @@ -6,6 +6,14 @@ use tinymemory_api::conformance::ReferenceEngine; use tinymemory_api::{LearningKind, ListRequest, MemoryEngine, MemoryMeta, StoreItem}; use tinymemory_tools::MemoryLayout; +#[test] +fn openhuman_defaults_to_plain_recall_and_resumes_after_compaction() { + assert_eq!(HostHook::for_turn(false, false), HostHook::Plain); + assert_eq!(HostHook::for_turn(true, false), HostHook::Resumed); + assert_eq!(HostHook::for_turn(false, true), HostHook::Dated); + assert_eq!(HostHook::for_turn(true, true), HostHook::DatedResumed); +} + #[test] fn logged_reply_keeps_bounded_tool_results() { let long = "a".repeat(MAX_TOOL_LINE_CHARS + 20); diff --git a/crates/tinymemory-integrations/examples/memory_eval/instrument.rs b/crates/tinymemory-integrations/examples/memory_eval/instrument.rs new file mode 100644 index 00000000..b24f6875 --- /dev/null +++ b/crates/tinymemory-integrations/examples/memory_eval/instrument.rs @@ -0,0 +1,49 @@ +//! Collect narrow library timing traces in the eval JSON report. + +use std::sync::Mutex; + +use crate::Timings; + +const TARGET: &str = "tinymemory_eval_timing"; + +struct TimingLogger(Mutex>); + +static LOGGER: TimingLogger = TimingLogger(Mutex::new(Vec::new())); + +impl log::Log for TimingLogger { + fn enabled(&self, metadata: &log::Metadata<'_>) -> bool { + metadata.target() == TARGET + } + + fn log(&self, record: &log::Record<'_>) { + if !self.enabled(record.metadata()) { + return; + } + let line = record.args().to_string(); + let Some((stage, value)) = line.split_once('=') else { + return; + }; + let Ok(ms) = value.parse::() else { + return; + }; + if let Ok(mut samples) = self.0.lock() { + samples.push((stage.replace('_', " "), ms)); + } + } + + fn flush(&self) {} +} + +pub(crate) fn install() { + if log::set_logger(&LOGGER).is_ok() { + log::set_max_level(log::LevelFilter::Trace); + } +} + +pub(crate) fn drain(timings: &mut Timings) { + if let Ok(mut samples) = LOGGER.0.lock() { + for (stage, ms) in samples.drain(..) { + timings.add(&stage, ms); + } + } +} diff --git a/crates/tinymemory-integrations/examples/memory_eval/kpi_tests.rs b/crates/tinymemory-integrations/examples/memory_eval/kpi_tests.rs index 033745d8..6a64caf6 100644 --- a/crates/tinymemory-integrations/examples/memory_eval/kpi_tests.rs +++ b/crates/tinymemory-integrations/examples/memory_eval/kpi_tests.rs @@ -28,6 +28,8 @@ fn report(name: &'static str, probes: Vec, conflicts: &[&str]) -> S writes: 0, tool_calls: 0, pre_turn_timeouts: 0, + ranked_ready: None, + ranked_wait_ms: None, settle_ms: 0.0, synthesis: Synthesis { captured: Captured { diff --git a/crates/tinymemory-integrations/examples/memory_eval/loop_guard.rs b/crates/tinymemory-integrations/examples/memory_eval/loop_guard.rs index ec0ab9d2..be6c1138 100644 --- a/crates/tinymemory-integrations/examples/memory_eval/loop_guard.rs +++ b/crates/tinymemory-integrations/examples/memory_eval/loop_guard.rs @@ -74,7 +74,7 @@ pub(crate) async fn run( runner.run(job).await?; } } - agent.flush().await?; + let _ = agent.flush().await?; let mut echoed_items = 0; let mut cursor = None; loop { diff --git a/crates/tinymemory-integrations/examples/memory_eval/main.rs b/crates/tinymemory-integrations/examples/memory_eval/main.rs index f99a5dfb..320b0c43 100644 --- a/crates/tinymemory-integrations/examples/memory_eval/main.rs +++ b/crates/tinymemory-integrations/examples/memory_eval/main.rs @@ -60,9 +60,11 @@ mod agent; mod compare; mod inspect; +mod instrument; mod kpi; mod llm; mod loop_guard; +mod safety; mod scenarios; mod score; @@ -74,8 +76,8 @@ use chrono::{TimeZone, Utc}; use serde::Serialize; use tinymemory_api::conformance::ReferenceEngine; use tinymemory_api::{ - ConsolidateRequest, ForgetTarget, ListRequest, MemoryEngine, MemoryMeta, Reach, Role, - StoreItem, Turn, + ConsolidateRequest, FetchMode, FetchRequest, ForgetTarget, ListRequest, MAX_STORE_MANY, + MemoryEngine, MemoryMeta, Reach, Role, StoreItem, Turn, TurnRange, WriteOptions, }; use tinymemory_integrations::cortex::{ CortexCredential, CortexEngine, StaticBearer, TINYHUMANS_API_ENDPOINT, @@ -86,7 +88,7 @@ use tinymemory_tools::{ MemoryLayout, PreTurn, RecallPolicy, SessionStart, }; -use agent::{PRE_TURN_TIMEOUT, ScriptedAgent, ms}; +use agent::{HostHook, PRE_TURN_TIMEOUT, ScriptedAgent, ms}; use inspect::{Captured, Derived, Inspector, Usage}; use llm::Llm; use scenarios::{MAIN, Probe, Scenario, Step, Via}; @@ -103,6 +105,20 @@ const WINDOW: u32 = 8; /// The longest a scenario's writes may take to become visible. const SETTLE_TIMEOUT: Duration = Duration::from_secs(90); +/// Bulk accepted writes can take longer to become fully listable. +const SCALE_SETTLE_TIMEOUT: Duration = Duration::from_secs(300); + +fn settle_timeout(scenario: &str) -> Duration { + if scenario == "needle_scale" { + SCALE_SETTLE_TIMEOUT + } else { + SETTLE_TIMEOUT + } +} + +fn settle_page_size(expected: usize) -> usize { + expected.clamp(100, 1_000) +} /// Consecutive empty, quiet polls of the enrichment queue (3 s apart) that /// count as drained. @@ -117,6 +133,13 @@ struct Args { label: String, llm: bool, host: String, + date_hint: bool, + scale_events: Option, + scale_position: String, + ranked_wait: u64, + seed: u64, + safety_audit: bool, + expect_derived: bool, loop_guard: bool, } @@ -133,6 +156,13 @@ fn args() -> Result { label: String::new(), llm: false, host: "scripted".into(), + date_hint: false, + scale_events: None, + scale_position: "middle".into(), + ranked_wait: 0, + seed: 251, + safety_audit: false, + expect_derived: false, loop_guard: false, }; let mut raw = std::env::args().skip(1); @@ -146,6 +176,13 @@ fn args() -> Result { "--label" => parsed.label = value()?, "--llm" => parsed.llm = true, "--host" => parsed.host = value()?, + "--date-hint" => parsed.date_hint = true, + "--scale-events" => parsed.scale_events = Some(value()?.parse()?), + "--scale-position" => parsed.scale_position = value()?, + "--ranked-wait" => parsed.ranked_wait = value()?.parse()?, + "--seed" => parsed.seed = value()?.parse()?, + "--safety-audit" => parsed.safety_audit = true, + "--expect-derived" => parsed.expect_derived = true, "--loop-guard" => parsed.loop_guard = true, other => return Err(format!("unknown flag {other}").into()), } @@ -156,6 +193,15 @@ fn args() -> Result { if parsed.loop_guard && parsed.host != "openhuman" { return Err("--loop-guard requires --host openhuman".into()); } + if parsed.date_hint && parsed.host != "openhuman" { + return Err("--date-hint requires --host openhuman".into()); + } + if parsed.expect_derived && !parsed.safety_audit { + return Err("--expect-derived requires --safety-audit".into()); + } + if parsed.ranked_wait > 0 && parsed.scale_events.is_none() { + return Err("--ranked-wait requires --scale-events".into()); + } if parsed.label.is_empty() { parsed.label = parsed.engine.clone(); } @@ -184,6 +230,8 @@ struct Synthesis { derived: Vec, /// Every fact, belief and conflict CortexDB holds for the scenario. captured: Captured, + /// Whether the queue reached four quiet polls before the wait cap. + enrichment_drained: Option, } /// One scenario's results. @@ -199,6 +247,8 @@ struct ScenarioReport { probes: Vec, /// What CortexDB's models spent on this scenario. usage: Option, + ranked_ready: Option, + ranked_wait_ms: Option, } /// The CortexDB flag profile a run is under: its name, and every flag it @@ -233,7 +283,18 @@ async fn main() -> Result<(), Error> { return compare::run(&raw[1..]); } let args = args()?; + instrument::install(); let (profile, flags) = profile()?; + let models: BTreeMap<&str, String> = [ + "CORTEX_EMBEDDING_MODEL", + "CORTEX_EXTRACTION_MODEL", + "CORTEX_ENRICHMENT_MODEL", + "CORTEX_ANSWER_MODEL", + "CORTEX_VERIFIER_MODEL", + ] + .into_iter() + .filter_map(|name| std::env::var(name).ok().map(|value| (name, value))) + .collect(); let url = std::env::var("CORTEX_DB_URL").unwrap_or_default(); let key = std::env::var("CORTEX_DB_KEY").unwrap_or_else(|_| "tinymemory-cortex-test".into()); let (engine, inspector): (Arc, Option) = match args.engine.as_str() @@ -301,6 +362,8 @@ async fn main() -> Result<(), Error> { enrich_wait, policy, openhuman: args.host == "openhuman", + date_hint: args.date_hint, + ranked_wait: args.ranked_wait, }; let usage_before = match &eval.inspector { Some(inspector) => Some(inspector.usage().await?), @@ -308,7 +371,15 @@ async fn main() -> Result<(), Error> { }; let mut timings = Timings::default(); let mut reports = Vec::new(); - for scenario in scenarios::all() { + let cases = if let Some(count) = args.scale_events { + vec![scenarios::scaled(count, &args.scale_position, args.seed)?] + } else { + scenarios::all() + }; + for scenario in cases { + if scenario.name == "long_compaction" && args.only.is_none() { + continue; + } if args .only .as_deref() @@ -340,6 +411,23 @@ async fn main() -> Result<(), Error> { } else { None }; + let safety = if args.safety_audit { + let report = safety::run( + engine.clone(), + eval.inspector.as_ref(), + run, + args.expect_derived, + ) + .await?; + println!( + "\n## Safety audit\n\n```json\n{}\n```", + serde_json::to_string_pretty(&report)? + ); + Some(report) + } else { + None + }; + instrument::drain(&mut timings); print_summary(&args.label, &reports, &timings); let usage = match (&eval.inspector, usage_before) { (Some(inspector), Some(before)) => Some(inspector.usage().await?.since(&before)), @@ -376,12 +464,21 @@ async fn main() -> Result<(), Error> { "label": args.label, "profile": profile, "flags": flags, + "models": models, + "probe_answer_model": eval.llm.as_ref().map(|llm| llm.model.as_str()), "server": server, "engine": engine.descriptor().id, "host": args.host, + "date_hint": args.date_hint, + "scale_events": args.scale_events, + "scale_position": args.scale_position, + "ranked_wait": args.ranked_wait, + "seed": args.seed, "run": run, "scenarios": reports, "loop_guard": loop_guard, + "safety": safety, + "expect_derived": args.expect_derived, "timings": timings, "usage": usage, "kpis": kpis, @@ -395,6 +492,12 @@ async fn main() -> Result<(), Error> { { return Err("loop guard found injected pack text in stored items".into()); } + if safety + .as_ref() + .is_some_and(|report| report.violations() > 0) + { + return Err("safety audit found a leaked or residual item".into()); + } Ok(()) } @@ -410,6 +513,7 @@ fn tenants(scenario: &Scenario) -> Vec<&'static str> { .steps .iter() .map(|step| match step { + Step::BulkDocuments { .. } => MAIN, Step::Doc { tenant, .. } | Step::Chat { tenant, .. } => *tenant, Step::Learning { .. } => MAIN, }) @@ -420,6 +524,24 @@ fn tenants(scenario: &Scenario) -> Vec<&'static str> { tenants } +/// The host queue need only execute one pending belief build per identical +/// scope: each build reads all accepted writes when it runs. Keep ingestion +/// jobs in order because they carry distinct documents. +fn coalesce_builds(jobs: Vec) -> Vec { + let mut builds = Vec::new(); + let mut ready = Vec::new(); + for job in jobs { + if matches!(job, BackgroundJob::BuildBeliefs { .. }) { + if builds.contains(&job) { + continue; + } + builds.push(job.clone()); + } + ready.push(job); + } + ready +} + /// One eval run: the engine, the optional helpers, and the settings every /// scenario shares. struct Eval { @@ -433,6 +555,8 @@ struct Eval { enrich_wait: u64, policy: RecallPolicy, openhuman: bool, + date_hint: bool, + ranked_wait: u64, } impl Eval { @@ -465,6 +589,33 @@ impl Eval { let mut pre_turn_timeouts = 0; for step in &scenario.steps { match step { + Step::BulkDocuments { + count, + needle_at, + seed, + } => { + let node = layout(run, scenario.name, MAIN)? + .brain(&tinymemory_tools::BrainSource::Files)?; + let started = Instant::now(); + for start in (0..*count).step_by(MAX_STORE_MANY) { + let batch = (start..(*count).min(start + MAX_STORE_MANY)) + .map(|index| { + StoreItem::document( + scenarios::scale_document(index, *needle_at, *seed), + MemoryMeta { + namespace: node.clone(), + ..MemoryMeta::default() + }, + ) + }) + .collect(); + engine + .store_many_with(batch, WriteOptions::accepted()) + .await?; + } + timings.add("bulk ingest (accepted)", ms(started)); + *writes.entry(MAIN).or_default() += *count; + } Step::Doc { tenant, source, @@ -506,6 +657,9 @@ impl Eval { .at(epoch + chrono::Duration::days(*day)); if self.openhuman { scripted = scripted.openhuman(); + if self.date_hint { + scripted = scripted.date_hint(); + } } for (text, tools) in turns { let record = scripted.user(text, tools).await?; @@ -521,7 +675,11 @@ impl Eval { jobs.extend(record.jobs); *writes.entry(tenant).or_default() += 1 + usize::from(record.logged); } - *writes.entry(tenant).or_default() += scripted.flush().await?; + let flushed = scripted.flush().await?; + *writes.entry(tenant).or_default() += flushed.logged; + for elapsed in flushed.completion_ms { + timings.add("pre_turn completed after deadline", elapsed); + } } } } @@ -529,19 +687,77 @@ impl Eval { // Settle: wait until every write is listed. let started = Instant::now(); for (tenant, expected) in &writes { - self.settle(&layout(run, scenario.name, tenant)?, *expected) - .await?; + self.settle( + &layout(run, scenario.name, tenant)?, + *expected, + settle_timeout(scenario.name), + ) + .await?; } let settle_ms = ms(started); timings.add("settle (all writes listed)", settle_ms); + let (ranked_ready, ranked_wait_ms) = if scenario.name == "needle_scale" { + let started = Instant::now(); + let ready = loop { + // The unique owner name checks indexing without warming a + // scored question's exact query. + let mut req = FetchRequest::new("Mira Solis", FetchMode::Hybrid, 10); + req.filter = layout(run, scenario.name, MAIN)?.brain_filter(None); + let fetch_started = Instant::now(); + let page = engine.fetch(req).await?; + timings.add("scale direct fetch", ms(fetch_started)); + if page.hits.iter().any(|hit| hit.text.contains("Mira Solis")) { + break true; + } + if started.elapsed() >= Duration::from_secs(self.ranked_wait) { + break false; + } + tokio::time::sleep(Duration::from_secs(2)).await; + }; + (Some(ready), Some(ms(started))) + } else { + (None, None) + }; + let mut probes = Vec::new(); for probe in &scenario.probes { probes.push(self.probe(scenario, probe, "recall", timings).await?); } + // The scale sweep measures retrieval over the planted corpus. Building + // thousands of reference beliefs changes that corpus and obscures + // the position/depth comparison. + if scenario.name == "needle_scale" { + if std::env::var("CORTEX_DB_KEEP").is_err() { + engine + .forget(ForgetTarget::Filter( + layout(run, scenario.name, MAIN)?.holistic_filter(), + )) + .await?; + } + let usage = match (&self.inspector, usage_before) { + (Some(inspector), Some(before)) => Some(inspector.usage().await?.since(&before)), + _ => None, + }; + return Ok(ScenarioReport { + name: scenario.name, + about: scenario.about, + writes: writes.values().sum(), + tool_calls, + pre_turn_timeouts, + settle_ms, + synthesis: Synthesis::default(), + probes, + usage, + ranked_ready, + ranked_wait_ms, + }); + } + // Synthesis: the jobs the writes handed back, then one build per tenant // over its whole tree. + let mut enrichment_drained = None; if let Some(inspector) = &self.inspector { let started = Instant::now(); let cap = Duration::from_secs(self.enrich_wait); @@ -559,7 +775,16 @@ impl Eval { } let waited = ms(started); timings.add("enrichment (queue drained)", waited); - println!(" enrichment drained in {:.0} s", waited / 1e3); + enrichment_drained = Some(quiet >= DRAINED_POLLS); + println!( + " enrichment {} in {:.0} s", + if quiet >= DRAINED_POLLS { + "drained" + } else { + "hit wait cap" + }, + waited / 1e3 + ); } else if self.hosted && self.enrich_wait > 0 { // No queue to read (hosted): give enrichment its usual lag. tokio::time::sleep(Duration::from_secs(self.enrich_wait)).await; @@ -574,9 +799,11 @@ impl Eval { request: ConsolidateRequest::new(Reach::subtree(root)), }); } + let jobs = coalesce_builds(jobs); let runner = memory(MAIN, "eval")?.background(); let mut synthesis = Synthesis { jobs: jobs.len(), + enrichment_drained, ..Synthesis::default() }; let started = Instant::now(); @@ -660,6 +887,8 @@ impl Eval { }; Ok(ScenarioReport { usage, + ranked_ready, + ranked_wait_ms, name: scenario.name, about: scenario.about, writes: writes.values().sum(), @@ -672,13 +901,19 @@ impl Eval { } /// Waits until `layout` lists at least `expected` items. - async fn settle(&self, layout: &MemoryLayout, expected: usize) -> Result<(), Error> { + async fn settle( + &self, + layout: &MemoryLayout, + expected: usize, + timeout: Duration, + ) -> Result<(), Error> { let started = Instant::now(); loop { let mut listed = 0; let mut cursor = None; loop { - let mut req = ListRequest::new(layout.holistic_filter(), 100); + let mut req = + ListRequest::new(layout.holistic_filter(), settle_page_size(expected)); req.cursor = cursor; let page = self.engine.list(req).await?; listed += page.items.len(); @@ -690,7 +925,7 @@ impl Eval { if listed >= expected { return Ok(()); } - if started.elapsed() > SETTLE_TIMEOUT { + if started.elapsed() > timeout { return Err(format!( "only {listed} of {expected} writes visible under {}", layout.root() @@ -719,14 +954,21 @@ impl Eval { let started = Instant::now(); let mut timed_out = false; let mut host_elapsed = None; + let mut completion_elapsed = None; let (markdown, tokens) = match &probe.via { Via::Ask => { let thread = format!("probe-{}", probe.id); let pre = PreTurn::new(thread, 0, probe.question); if self.openhuman { - let (pack, timeout, elapsed) = host_probe_pack(memory.clone(), pre).await?; + let (pack, timeout, elapsed, completed) = host_probe_pack( + memory.clone(), + pre, + HostHook::for_turn(false, self.date_hint), + ) + .await?; timed_out = timeout; host_elapsed = Some(elapsed); + completion_elapsed = completed; pack } else { pack(memory.pre_turn(pre).await?.pack) @@ -760,9 +1002,15 @@ impl Eval { let mut pre = PreTurn::new(*thread, *turn_index, probe.question); pre.in_prompt_from = *in_prompt_from; if self.openhuman { - let (pack, timeout, elapsed) = host_probe_pack(memory.clone(), pre).await?; + let (pack, timeout, elapsed, completed) = host_probe_pack( + memory.clone(), + pre, + HostHook::for_turn(true, self.date_hint), + ) + .await?; timed_out = timeout; host_elapsed = Some(elapsed); + completion_elapsed = completed; pack } else { pack(memory.pre_turn(pre).await?.pack) @@ -790,6 +1038,26 @@ impl Eval { result.llm_cost_usd = answer.cost_usd; } timings.add(&format!("probe {}", result.via), elapsed); + if let Some(completed) = completion_elapsed { + timings.add("probe pre_turn completed after deadline", completed); + } + let logged_turn = match &probe.via { + Via::Ask => Some((format!("probe-{}", probe.id), 0)), + Via::Continue { + thread, turn_index, .. + } => Some(((*thread).to_string(), *turn_index)), + _ => None, + }; + if let Some((thread_id, index)) = logged_turn { + let mut filter = + layout(run, scenario.name, probe.tenant)?.conversations_filter(Some(probe.agent)); + filter.thread_id = Some(thread_id); + filter.turns = Some(TurnRange { + first: index, + last: index, + }); + engine.forget(ForgetTarget::Filter(filter)).await?; + } Ok(result) } } @@ -804,16 +1072,25 @@ fn pack(pack: ContextPack) -> (String, usize) { async fn host_probe_pack( memory: AgentMemory, pre: PreTurn, -) -> Result<((String, usize), bool, f64), Error> { + hook: HostHook, +) -> Result<((String, usize), bool, f64, Option), Error> { let started = Instant::now(); - let mut task = - tokio::spawn(async move { memory.pre_turn_dated(pre, false, async { None }).await }); + let mut task = tokio::spawn(async move { + let result = hook.run(memory, pre).await; + (result, Instant::now()) + }); match tokio::time::timeout(PRE_TURN_TIMEOUT, &mut task).await { - Ok(context) => Ok((pack(context??.pack), false, ms(started))), + Ok(context) => Ok((pack(context?.0?.pack), false, ms(started), None)), Err(_) => { let elapsed = ms(started); - task.await??; - Ok(((String::new(), 0), true, elapsed)) + let (context, completed) = task.await?; + context?; + Ok(( + (String::new(), 0), + true, + elapsed, + Some(completed.duration_since(started).as_secs_f64() * 1_000.0), + )) } } } @@ -911,3 +1188,7 @@ fn print_summary(label: &str, reports: &[ScenarioReport], timings: &Timings) { ); } } + +#[cfg(test)] +#[path = "main_tests.rs"] +mod tests; diff --git a/crates/tinymemory-integrations/examples/memory_eval/main_tests.rs b/crates/tinymemory-integrations/examples/memory_eval/main_tests.rs new file mode 100644 index 00000000..71024f09 --- /dev/null +++ b/crates/tinymemory-integrations/examples/memory_eval/main_tests.rs @@ -0,0 +1,83 @@ +//! Probe isolation checks for the benchmark runner. + +use super::*; + +#[tokio::test] +async fn a_probe_does_not_leave_its_question_in_the_measured_corpus() { + let engine: Arc = Arc::new(ReferenceEngine::new()); + let probe = scenarios::all()[0].probes[0].clone(); + let scenario = Scenario { + name: "probe_cleanup", + about: "", + steps: Vec::new(), + probes: vec![probe.clone()], + }; + let eval = Eval { + engine: engine.clone(), + inspector: None, + hosted: false, + llm: None, + run: 251, + enrich_wait: 0, + policy: RecallPolicy::default(), + openhuman: true, + date_hint: false, + ranked_wait: 0, + }; + eval.probe(&scenario, &probe, "recall", &mut Timings::default()) + .await + .unwrap(); + let listed = engine + .list(ListRequest::new( + layout(251, scenario.name, probe.tenant) + .unwrap() + .holistic_filter(), + 100, + )) + .await + .unwrap(); + assert!( + listed.items.is_empty(), + "a probe's logged question contaminated later measurements" + ); +} + +#[test] +fn repeated_build_requests_for_one_scope_run_once() { + let request = ConsolidateRequest::new(Reach::exact("agent:planner".parse().unwrap())); + let build = BackgroundJob::BuildBeliefs { request }; + let jobs = coalesce_builds(vec![build.clone(), build]); + assert_eq!(jobs.len(), 1); +} + +#[test] +fn bulk_scale_writes_get_a_longer_visibility_window() { + assert_eq!(settle_timeout("needle_scale"), Duration::from_secs(300)); + assert_eq!(settle_timeout("tool_heavy"), Duration::from_secs(90)); + assert_eq!(settle_page_size(1_000), 1_000); + assert_eq!(settle_page_size(3), 100); +} + +#[tokio::test] +async fn scale_run_separates_direct_fetch_readiness_from_pack_accuracy() { + let engine: Arc = Arc::new(ReferenceEngine::new()); + let eval = Eval { + engine, + inspector: None, + hosted: false, + llm: None, + run: 252, + enrich_wait: 0, + policy: RecallPolicy::default(), + openhuman: true, + date_hint: false, + ranked_wait: 0, + }; + let scenario = scenarios::scaled(100, "middle", 251).unwrap(); + let report = eval + .scenario(&scenario, &mut Timings::default()) + .await + .unwrap(); + assert_eq!(report.ranked_ready, Some(true)); + assert_eq!(report.probes.len(), 2); +} diff --git a/crates/tinymemory-integrations/examples/memory_eval/safety.rs b/crates/tinymemory-integrations/examples/memory_eval/safety.rs new file mode 100644 index 00000000..8911db7c --- /dev/null +++ b/crates/tinymemory-integrations/examples/memory_eval/safety.rs @@ -0,0 +1,331 @@ +//! Synthetic tenant-isolation and deletion checks across public read channels. + +use std::sync::Arc; + +use serde::Serialize; +use tinymemory_api::{ + EraseRequest, ForgetTarget, LearningKind, ListRequest, MemoryEngine, MemoryMeta, MetaFilter, + Reach, RecallRequest, StoreItem, +}; +use tinymemory_tools::context::{self, Brief, ContextSpec}; +use tinymemory_tools::{AgentMemory, PreTurn, RecallPolicy}; + +use crate::inspect::Inspector; +use crate::{Error, layout}; + +const SENTINEL: &str = "cobalt-heron-251"; + +/// Each check records whether the synthetic private fact escaped or survived. +#[derive(Debug, Serialize)] +pub(crate) struct SafetyReport { + pub(crate) checks: Vec, + pub(crate) forgotten: usize, + pub(crate) derived_after_forget: Vec, + pub(crate) derived_ready_before_forget: Option, + pub(crate) derived_ready_before_erase: Option, +} + +impl SafetyReport { + pub(crate) fn violations(&self) -> usize { + self.checks.iter().filter(|check| !check.passed).count() + } +} + +#[derive(Debug, Serialize)] +pub(crate) struct SafetyCheck { + channel: &'static str, + phase: &'static str, + passed: bool, +} + +fn check(checks: &mut Vec, channel: &'static str, phase: &'static str, exposed: bool) { + checks.push(SafetyCheck { + channel, + phase, + passed: !exposed, + }); +} + +async fn pack_contains(memory: &AgentMemory, thread: &str) -> Result { + let pack = memory + .pre_turn(PreTurn::new( + thread, + 0, + "What is the private migration token?", + )) + .await? + .pack; + Ok(pack.markdown.contains(SENTINEL)) +} + +async fn context_contains( + engine: &dyn MemoryEngine, + root: tinymemory_api::Namespace, +) -> Result { + let spec = ContextSpec { + briefs: vec![Brief::new( + "Migration", + "What is the private migration token?", + )], + reach: Some(Reach::subtree(root)), + ..ContextSpec::default() + }; + Ok(context::compile(engine, &spec) + .await? + .markdown + .contains(SENTINEL)) +} + +async fn answer_contains(engine: &dyn MemoryEngine, filter: MetaFilter) -> Result { + let mut request = RecallRequest::new("What is the private migration token?", 8); + request.filter = filter; + Ok(engine.recall(request).await?.answer.contains(SENTINEL)) +} + +async fn export_contains( + engine: &dyn MemoryEngine, + filter: tinymemory_api::MetaFilter, +) -> Result { + let mut cursor = None; + loop { + let mut req = ListRequest::new(filter.clone(), 100); + req.cursor = cursor; + let page = engine.export(req).await?; + if page + .items + .iter() + .any(|item| serde_json::to_string(&item.item).is_ok_and(|body| body.contains(SENTINEL))) + { + return Ok(true); + } + cursor = page.next_cursor; + if cursor.is_none() { + return Ok(false); + } + } +} + +async fn derived_contains( + inspector: Option<&Inspector>, + root: &str, +) -> Result, Error> { + let Some(inspector) = inspector else { + return Ok(None); + }; + let scopes = inspector.scopes(root).await?; + Ok(Some(inspector.captured(&scopes).await?.mentions(SENTINEL))) +} + +async fn wait_derived(inspector: &Inspector, root: &str) -> Result { + let started = std::time::Instant::now(); + loop { + if derived_contains(Some(inspector), root).await? == Some(true) { + return Ok(true); + } + if started.elapsed() >= std::time::Duration::from_secs(90) { + return Ok(false); + } + tokio::time::sleep(std::time::Duration::from_secs(3)).await; + } +} + +/// Audit sibling isolation, filter-forget, and subtree erase on one engine. +pub(crate) async fn run( + engine: Arc, + inspector: Option<&Inspector>, + run: u64, + expect_derived: bool, +) -> Result { + let acme = layout(run, "safety_audit", "acme")?; + let globex = layout(run, "safety_audit", "globex")?; + let policy = RecallPolicy { + team_limit: 0, + ..RecallPolicy::default() + }; + let acme_memory = + AgentMemory::new(engine.clone(), acme.clone(), "auditor")?.with_policy(policy.clone()); + let globex_memory = + AgentMemory::new(engine.clone(), globex.clone(), "auditor")?.with_policy(policy); + let item = || { + StoreItem::learning( + format!("The private migration token is {SENTINEL}."), + LearningKind::Fact, + 1.0, + MemoryMeta { + namespace: acme.learnings().clone(), + ..MemoryMeta::default() + }, + ) + }; + engine.store(item()).await?; + let mut checks = Vec::new(); + // The control proves that the fixture is retrievable before testing isolation. + checks.push(SafetyCheck { + channel: "own_pack_control", + phase: "before", + passed: pack_contains(&acme_memory, "own-control").await?, + }); + check( + &mut checks, + "sibling_pack", + "before", + pack_contains(&globex_memory, "sibling-check").await?, + ); + check( + &mut checks, + "sibling_context", + "before", + context_contains(engine.as_ref(), globex.root().clone()).await?, + ); + check( + &mut checks, + "sibling_answer", + "before", + answer_contains(engine.as_ref(), globex.holistic_filter()).await?, + ); + if let Some(exposed) = derived_contains(inspector, &globex.root().to_string()).await? { + check(&mut checks, "sibling_derived", "before", exposed); + } + + let derived_ready_before_forget = if expect_derived { + let Some(inspector) = inspector else { + return Err("--expect-derived needs a direct CortexDB inspector".into()); + }; + let ready = wait_derived(inspector, &acme.root().to_string()).await?; + checks.push(SafetyCheck { + channel: "own_derived_control", + phase: "before", + passed: ready, + }); + Some(ready) + } else { + None + }; + + let forgotten = engine + .forget(ForgetTarget::Filter(acme.holistic_filter())) + .await? + .forgotten; + check( + &mut checks, + "pack", + "after_forget", + pack_contains(&acme_memory, "forgotten-check").await?, + ); + check( + &mut checks, + "context", + "after_forget", + context_contains(engine.as_ref(), acme.root().clone()).await?, + ); + check( + &mut checks, + "answer", + "after_forget", + answer_contains(engine.as_ref(), acme.holistic_filter()).await?, + ); + check( + &mut checks, + "export", + "after_forget", + export_contains(engine.as_ref(), acme.holistic_filter()).await?, + ); + let mut derived_after_forget = Vec::new(); + if let Some(inspector) = inspector { + let scopes = inspector.scopes(&acme.root().to_string()).await?; + let captured = inspector.captured(&scopes).await?; + derived_after_forget.extend( + captured + .facts + .iter() + .chain(&captured.beliefs) + .filter(|line| line.contains(SENTINEL)) + .cloned(), + ); + } + if let Some(exposed) = derived_contains(inspector, &acme.root().to_string()).await? { + check(&mut checks, "derived", "after_forget", exposed); + } + if inspector.is_some() { + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + if let Some(exposed) = derived_contains(inspector, &acme.root().to_string()).await? { + check(&mut checks, "derived", "after_forget_5s", exposed); + } + } + + let restored = engine.store(item()).await?; + checks.push(SafetyCheck { + channel: "restore_control", + phase: "before_erase", + passed: !restored.replayed, + }); + let derived_ready_before_erase = if expect_derived { + let Some(inspector) = inspector else { + return Err("--expect-derived needs a direct CortexDB inspector".into()); + }; + let ready = wait_derived(inspector, &acme.root().to_string()).await?; + checks.push(SafetyCheck { + channel: "own_derived_control", + phase: "before_erase", + passed: ready, + }); + Some(ready) + } else { + None + }; + engine + .erase(EraseRequest::new(Reach::subtree(acme.root().clone()))) + .await?; + check( + &mut checks, + "pack", + "after_erase", + pack_contains(&acme_memory, "erased-check").await?, + ); + check( + &mut checks, + "context", + "after_erase", + context_contains(engine.as_ref(), acme.root().clone()).await?, + ); + check( + &mut checks, + "answer", + "after_erase", + answer_contains(engine.as_ref(), acme.holistic_filter()).await?, + ); + check( + &mut checks, + "export", + "after_erase", + export_contains(engine.as_ref(), acme.holistic_filter()).await?, + ); + if let Some(exposed) = derived_contains(inspector, &acme.root().to_string()).await? { + check(&mut checks, "derived", "after_erase", exposed); + } + if inspector.is_some() { + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + if let Some(exposed) = derived_contains(inspector, &acme.root().to_string()).await? { + check(&mut checks, "derived", "after_erase_5s", exposed); + } + } + if std::env::var("CORTEX_DB_KEEP").is_err() { + engine + .forget(ForgetTarget::Filter(acme.holistic_filter())) + .await?; + engine + .forget(ForgetTarget::Filter(globex.holistic_filter())) + .await?; + } + Ok(SafetyReport { + checks, + forgotten, + derived_after_forget, + derived_ready_before_forget, + derived_ready_before_erase, + }) +} + +#[cfg(test)] +#[path = "safety_tests.rs"] +mod tests; diff --git a/crates/tinymemory-integrations/examples/memory_eval/safety_tests.rs b/crates/tinymemory-integrations/examples/memory_eval/safety_tests.rs new file mode 100644 index 00000000..9954725e --- /dev/null +++ b/crates/tinymemory-integrations/examples/memory_eval/safety_tests.rs @@ -0,0 +1,12 @@ +//! The reference engine must pass the synthetic safety audit. + +use super::*; +use tinymemory_api::conformance::ReferenceEngine; + +#[tokio::test] +async fn sibling_memory_stays_isolated_and_deletions_leave_no_public_trace() { + let engine = Arc::new(ReferenceEngine::new()); + let report = run(engine, None, 251, false).await.unwrap(); + assert_eq!(report.violations(), 0, "{report:?}"); + assert_eq!(report.checks.len(), 13); +} diff --git a/crates/tinymemory-integrations/examples/memory_eval/scenarios.rs b/crates/tinymemory-integrations/examples/memory_eval/scenarios.rs index 826cee81..2adf7444 100644 --- a/crates/tinymemory-integrations/examples/memory_eval/scenarios.rs +++ b/crates/tinymemory-integrations/examples/memory_eval/scenarios.rs @@ -21,6 +21,12 @@ pub(crate) const MAIN: &str = "main"; /// Something written before the probes run. pub(crate) enum Step { + /// A seeded set of documents written in batches for the scale sweep. + BulkDocuments { + count: usize, + needle_at: usize, + seed: u64, + }, /// A brain document. Doc { tenant: &'static str, @@ -162,8 +168,11 @@ pub(crate) fn all() -> Vec { restart_recall(), contradictions(), tool_heavy(), + coding_session(), + task_drift(), team_handoff(), compaction(), + long_compaction(), isolation(), needle_in_noise(), learnings(), @@ -173,6 +182,75 @@ pub(crate) fn all() -> Vec { ] } +/// A controlled needle sweep. The same seed produces the same decoys at +/// every size, while the chosen position moves the answer-bearing item. +pub(crate) fn scaled(count: usize, position: &str, seed: u64) -> Result { + if !matches!(count, 100 | 1_000 | 10_000) { + return Err("--scale-events must be 100, 1000, or 10000".into()); + } + let needle_at = match position { + "early" => count / 10, + "middle" => count / 2, + "late" => count - count / 10 - 1, + _ => return Err("--scale-position must be early, middle, or late".into()), + }; + Ok(Scenario { + name: "needle_scale", + about: "One owner hidden among seeded near-duplicate support documents", + steps: vec![Step::BulkDocuments { + count, + needle_at, + seed, + }], + probes: vec![ + Probe::new( + "owner-lexical", + "scale-agent", + "Who owns orbital cache invalidation?", + Style::Lexical, + ) + .expect(&["Mira Solis"]), + Probe::new( + "owner-paraphrase", + "scale-agent", + "Which engineer handles the satellite cache refresh rule?", + Style::Paraphrase, + ) + .expect(&["Mira Solis"]), + ], + }) +} + +/// A stable distractor or the answer-bearing document for a scale run. +pub(crate) fn scale_document(index: usize, needle_at: usize, seed: u64) -> String { + if index == needle_at { + return format!("Case {index}: Mira Solis owns orbital cache invalidation."); + } + let shuffled = (index as u64) + .wrapping_mul(6_364_136_223_846_793_005) + .wrapping_add(seed); + let teams = [ + "billing", + "search", + "mobile", + "storage", + "checkout", + "analytics", + ]; + let owners = ["Nadia", "Owen", "Priya", "Rafael", "Tess", "Uma"]; + let n = (shuffled >> 32) as usize; + format!( + "Case {index}: {} owns {} cache review. The refresh checklist is in playbook {}.", + owners[n % owners.len()], + teams[(n / 7) % teams.len()], + n % 997 + ) +} + +#[cfg(test)] +#[path = "scenarios_tests.rs"] +mod tests; + /// User turns with no tool calls. fn said(lines: &[&str]) -> Vec<(String, Vec)> { lines @@ -602,6 +680,196 @@ fn tool_heavy() -> Scenario { } } +fn coding_session() -> Scenario { + use Style::{Lexical, Paraphrase}; + Scenario { + name: "coding_session", + about: "A coding decision, failing test, and file path found only in tool results", + steps: vec![chat( + "coder-42", + "retry-fix", + 0, + vec![ + ( + "Find why the API retries a refused connection.".into(), + vec![ + tool("read_file", "src/client/retry.rs: retry_on_connect=false"), + tool( + "run_tests", + "FAILED test_retries_refused_connection at retry_tests.rs:81", + ), + ], + ), + ( + "Try the proposed change and inspect the diff.".into(), + vec![ + tool( + "git_diff", + "retry_on_connect=true caused duplicate writes after ambiguous timeout", + ), + tool("run_tests", "FAILED test_does_not_repeat_unknown_write"), + ], + ), + ( + "Revert that change and keep the safe path.".into(), + vec![tool( + "git_diff", + "reverted retry_on_connect=true; kept writes single-attempt", + )], + ), + ], + )], + probes: vec![ + Probe::new( + "retry-file", + "coder-42", + "Which file contains the retry policy?", + Lexical, + ) + .expect(&["src/client/retry.rs"]), + Probe::new( + "failed-test", + "coder-42", + "Which test exposed the unsafe retry?", + Paraphrase, + ) + .expect(&["test_does_not_repeat_unknown_write"]), + Probe::new( + "why-revert", + "coder-42", + "Why did we back out the connection retry?", + Paraphrase, + ) + .expect(&["duplicate writes"]), + ], + } +} + +fn task_drift() -> Scenario { + use Style::{Lexical, Paraphrase}; + Scenario { + name: "task_drift", + about: "A two-week task changes owner and plan while old status remains searchable", + steps: vec![ + chat( + "planner-07", + "migration", + 0, + said(&[ + "The ledger migration is owned by Anna. Plan A is a Friday cutover.", + "The blocker is an unverified backfill checksum.", + ]), + ), + chat( + "planner-07", + "migration-update", + 7, + said(&[ + "The ledger migration moved to Ravi. Plan B is a staged Monday cutover.", + "The backfill checksum is verified; the current blocker is partner signoff.", + ]), + ), + chat( + "planner-07", + "migration-resume", + 14, + said(&[ + "Resume the ledger migration. Ravi still owns it; partner signoff remains open.", + ]), + ), + ], + probes: vec![ + Probe::new( + "current-owner", + "planner-07", + "Who owns the ledger migration now?", + Lexical, + ) + .expect(&["Ravi"]) + .stale(&["Anna"]), + Probe::new( + "current-blocker", + "planner-07", + "What still blocks the cutover?", + Paraphrase, + ) + .expect(&["partner signoff"]) + .stale(&["unverified backfill checksum"]), + Probe::new( + "changed-plan", + "planner-07", + "What is the new release approach?", + Paraphrase, + ) + .expect(&["staged Monday cutover"]) + .stale(&["Friday cutover"]), + ], + } +} + +fn long_compaction() -> Scenario { + use Style::Lexical; + let mut turns = said(&["The archive key is indigo-cedar; retain it after compaction."]); + turns.extend((1..200).map(|n| { + ( + format!("Routine planning note {n}: check the agenda."), + Vec::new(), + ) + })); + let first: Vec = turns + .iter() + .take(100) + .map(|(text, _)| text.clone()) + .collect(); + let second: Vec = turns + .iter() + .take(180) + .map(|(text, _)| text.clone()) + .collect(); + Scenario { + name: "long_compaction", + about: "Two compaction windows over a 200-exchange thread", + steps: vec![chat("planner-07", "archive", 0, turns)], + probes: vec![ + Probe::new( + "first-compaction", + "planner-07", + "What archive key must survive?", + Lexical, + ) + .via(Via::Compact { + thread: "archive", + dropped: first, + }) + .expect(&["indigo-cedar"]), + Probe::new( + "second-compaction", + "planner-07", + "What archive key must survive?", + Lexical, + ) + .via(Via::Compact { + thread: "archive", + dropped: second, + }) + .expect(&["indigo-cedar"]), + Probe::new( + "resumed-window", + "planner-07", + "What archive key must survive?", + Lexical, + ) + .via(Via::Continue { + thread: "archive", + turn_index: 400, + in_prompt_from: 392, + }) + .expect(&["indigo-cedar"]) + .forbid(&["planning note 198", "planning note 199"]), + ], + } +} + fn team_handoff() -> Scenario { use Style::{Lexical, Paraphrase}; Scenario { diff --git a/crates/tinymemory-integrations/examples/memory_eval/scenarios_tests.rs b/crates/tinymemory-integrations/examples/memory_eval/scenarios_tests.rs new file mode 100644 index 00000000..e3bbd5c0 --- /dev/null +++ b/crates/tinymemory-integrations/examples/memory_eval/scenarios_tests.rs @@ -0,0 +1,39 @@ +//! Determinism and placement checks for generated benchmark fixtures. + +use super::*; + +#[test] +fn scaled_needles_move_without_changing_the_decoys() { + let early = scaled(100, "early", 17).unwrap(); + let late = scaled(100, "late", 17).unwrap(); + let [ + Step::BulkDocuments { + needle_at: early_at, + .. + }, + ] = early.steps.as_slice() + else { + panic!("one bulk step") + }; + let [ + Step::BulkDocuments { + needle_at: late_at, .. + }, + ] = late.steps.as_slice() + else { + panic!("one bulk step") + }; + assert_eq!((*early_at, *late_at), (10, 89)); + assert_eq!( + scale_document(20, *early_at, 17), + scale_document(20, *late_at, 17) + ); + assert!(scale_document(*early_at, *early_at, 17).contains("Mira Solis")); + assert!(!scale_document(20, *early_at, 17).contains("Mira Solis")); +} + +#[test] +fn scaled_fixture_rejects_unbounded_sizes_and_unknown_positions() { + assert!(scaled(99, "early", 1).is_err()); + assert!(scaled(100, "unknown", 1).is_err()); +} diff --git a/crates/tinymemory-integrations/src/cortex/engine/fetch.rs b/crates/tinymemory-integrations/src/cortex/engine/fetch.rs index c048f88e..486e7254 100644 --- a/crates/tinymemory-integrations/src/cortex/engine/fetch.rs +++ b/crates/tinymemory-integrations/src/cortex/engine/fetch.rs @@ -27,6 +27,7 @@ //! ranking (`next_cursor: None`) when no hit beyond it was found. use std::collections::{HashMap, HashSet}; +use std::time::Instant; use futures::{StreamExt, TryStreamExt, stream}; use serde_json::{Value, json}; @@ -132,6 +133,7 @@ pub(super) fn ranked(pack: &Value, kind: Option, filter: &MetaFilter) impl CortexEngine { /// See the module docs. pub(super) async fn fetch_page(&self, req: FetchRequest) -> Result { + let fetch_started = Instant::now(); self.descriptor.ensure_mode(req.mode)?; req.validate()?; let (offset, chosen) = match &req.cursor { @@ -146,7 +148,9 @@ impl CortexEngine { .saturating_add(1) .saturating_mul(EVENTS_PER_HIT) .min(MAX_PACK_EVENTS); + let discovery_started = Instant::now(); let scopes = self.scopes_for(&req.filter).await?; + log::trace!(target: "tinymemory_eval_timing", "cortex_scope_discovery={}", discovery_started.elapsed().as_secs_f64() * 1_000.0); let (scopes, chosen) = match (req.max_scopes, chosen) { // A later page reads the scopes the first page chose. (Some(_), Some(paths)) => { @@ -181,6 +185,7 @@ impl CortexEngine { if let Some(temporal) = hint.as_ref().filter(|_| !self.refers_off()) { body["temporal"] = temporal.clone(); } + let pack_started = Instant::now(); let pack = match self.log.recall(&body).await { Err(error @ Error::InvalidRequest(_)) if hint.is_some() => { body.as_object_mut().map(|body| body.remove("temporal")); @@ -190,6 +195,7 @@ impl CortexEngine { } pack => pack?, }; + log::trace!(target: "tinymemory_eval_timing", "cortex_recall_pack={}", pack_started.elapsed().as_secs_f64() * 1_000.0); let beliefs = if wanted_beliefs > 0 { beliefs_in(&self.layout, &pack, "/layers/beliefs") } else { @@ -257,6 +263,7 @@ impl CortexEngine { } else { None }; + log::trace!(target: "tinymemory_eval_timing", "cortex_fetch_total={}", fetch_started.elapsed().as_secs_f64() * 1_000.0); Ok(FetchPage { hits, next_cursor, diff --git a/crates/tinymemory-tools/src/recall/gather.rs b/crates/tinymemory-tools/src/recall/gather.rs index ac30142e..b2c25ad5 100644 --- a/crates/tinymemory-tools/src/recall/gather.rs +++ b/crates/tinymemory-tools/src/recall/gather.rs @@ -28,6 +28,7 @@ //! itself. use std::collections::HashSet; +use std::time::Instant; use futures::future::join; use tinymemory_api::{ @@ -89,6 +90,7 @@ pub(super) async fn section( beliefs: usize, dated: bool, ) -> Gathered { + let section_started = Instant::now(); let keep = |hit: &Hit| !request.excludes(hit); let want = wanted(request, section); // A date reorders a fetch section's hits after the read, so read deeper @@ -105,7 +107,10 @@ pub(super) async fn section( instructions, fallback_to_fetch, } => match answer(engine, request, section, question, instructions.clone()).await { - Ok(Some(filled)) => return filled, + Ok(Some(filled)) => { + log::trace!(target: "tinymemory_eval_timing", "recall_section={}", section_started.elapsed().as_secs_f64() * 1_000.0); + return filled; + } Ok(None) => Ok((Vec::new(), Vec::new())), Err(error) if *fallback_to_fetch => { log::debug!( @@ -140,6 +145,7 @@ pub(super) async fn section( } SectionQuery::Latest => with_listed_beliefs(engine, section, want, &keep).await, }; + log::trace!(target: "tinymemory_eval_timing", "recall_section={}", section_started.elapsed().as_secs_f64() * 1_000.0); match outcome { Ok((hits, beliefs)) => Gathered::Hits { hits, beliefs }, Err(error) => { diff --git a/docs/evals/openhuman-host.md b/docs/evals/openhuman-host.md index 946213c6..f2a367a4 100644 --- a/docs/evals/openhuman-host.md +++ b/docs/evals/openhuman-host.md @@ -12,12 +12,18 @@ them when the host's defaults or hook behavior changes. | Behavior | Eval mirror | OpenHuman source | | --- | --- | --- | | 1200-token budget, 8 learnings, 6 brain, 6 history, 0 team, beliefs every 10 turns | `main.rs`, OpenHuman policy | `crates/openhuman-core/src/config/schema/memory.rs:278-336` | -| Dated pre-turn hook, ready date hint, empty pack after 1500 ms | `agent.rs`, `ScriptedAgent::user` | `crates/openhuman-core/src/memory/lifecycle/hooks.rs:225-285` | +| Plain `pre_turn` by default, resumed hook after compaction, optional dated path with `--date-hint`, empty pack after 1500 ms | `agent.rs`, `HostHook` | `crates/openhuman-core/src/config/schema/memory.rs` and `crates/openhuman-core/src/memory/lifecycle/hooks.rs` | | Timed-out task continues in the background and can still log the turn | `agent.rs`, `ScriptedAgent::flush` | `crates/openhuman-core/src/memory/lifecycle/hooks.rs:190-225` | +| Identical pending belief builds run once | `main.rs`, `coalesce_builds` | `crates/openhuman-core/src/memory/lifecycle/jobs.rs`, `enqueue` | | User and reply indices `2n` and `2n+1` | `agent.rs`, `ScriptedAgent::user` | `crates/openhuman-core/src/memory/lifecycle/hooks.rs:245-255,414-435` | | Tool-result lines, whitespace folding, 240-character cap | `agent.rs`, `logged_reply` | `crates/openhuman-core/src/memory/lifecycle/hooks.rs:385-406` | -The profile reports pre-turn p50, p95, p99 and timeout rate. A timeout leaves +The profile reports pre-turn p50, p95, p99 and timeout rate. These host-facing +latencies stop at the deadline; separate completion samples show how long +timed-out work actually took. The report also records per-section recall, +scope discovery, CortexDB fetch, and per-scope recall-pack timings. The +recall-pack timing includes server embedding and ranking; CortexDB does not +expose those stages separately through this wire. A timeout leaves the simulated model with an empty pack. `--loop-guard` reads its script from `examples/memory_eval/data/loop_guard.json`, replays 500 turns, and exports all stored items to detect any `` tag copied into memory. It @@ -31,6 +37,12 @@ cargo run -p tinymemory-integrations --features full --example memory_eval -- \ ./scripts/memory-eval.sh --host openhuman --only tool_heavy,compaction \ --json target/memory-eval/openhuman-cortex.json + +cargo run -p tinymemory-integrations --features full --example memory_eval -- \ + --engine reference --host openhuman --scale-events 1000 \ + --scale-position middle --seed 251 --json target/memory-eval/scale.json + +./scripts/memory-eval.sh --host openhuman --only none --safety-audit ``` The 500-turn loop guard is separate from the standard scenario run because it @@ -42,6 +54,22 @@ scenario format for the full suite, and 8000 ms compaction deadline in issue implemented. The OpenHuman scrub guard is also outside this profile; the scripted fixtures contain no sensitive data. +The scale sweep accepts 100, 1,000, or 10,000 seeded documents and early, +middle, or late needle placement. It measures retrieval before synthesis so +belief builds do not change the planted corpus. A direct fetch first checks +whether the answer-bearing document is ranked at all; `--ranked-wait 30` +polls for up to 30 seconds and records that lag separately from the host +probe. Readiness uses the planted owner's name rather than either scored +question, so it does not warm their exact queries. The 200-turn +`long_compaction` case is opt-in with `--only long_compaction`; ordinary full +runs include the shorter coding and task-drift cases. Probe turns are removed +after each score so they cannot contaminate later probes or synthesis. +`--safety-audit` checks synthetic tenant isolation and forget/erase across +packs, direct answers, `context.md`, export, and the derived layers available +on direct CortexDB. Set `BUDGET_USD=5` for a sequence of OpenRouter runs in +one `OUT_DIR`; the runner tracks the key's cumulative usage and reserves +`COST_PER_RUN` (default $0.50) before starting each run. + ## Measured runs, 2026-10-10 | Run | Scope | Recall / synthesis pack hit | Synthesis model answer | Pre-turn p95 | Timeout rate | CortexDB cost | @@ -52,7 +80,7 @@ scripted fixtures contain no sensitive data. | repeat 2 | same | 0/6 · 5/6 | 5/6 | 1502 ms | 11/19 calls | $0.012 | | repeat 3 | same | 0/6 · 5/6 | 4/6 | 1508 ms | 12/19 calls | $0.030 | -The three live runs used CortexDB v0.10.4, OpenRouter embeddings and +The historical three live runs used CortexDB v0.10.4, OpenRouter embeddings and extraction, and `openai/gpt-4.1-mini` answers. The answer model cost another $0.002 per run. The timeout rate counts seven scripted turns and twelve pre-turn probes across both phases. All six recall-phase probes missed the @@ -62,9 +90,107 @@ six expected answers each time; the absent one was the paraphrase asking who to contact about the regression. CortexDB model cost varied from $0.012 to $0.050, so a single run would hide substantial cost noise. +Those runs used dated recall even though OpenHuman defaults to plain recall, +and their probes logged questions back into the measured corpus. The +corrected runs below supersede these figures for comparisons. + The reference-engine loop guard ran all 500 turns with 500 nonempty packs, zero timeouts and zero stored `` tags. Mean pack size was 114 tokens in the first 50 turns and 130 in the last 50. Repeated bullet-line rates were 82.4% and 85.7%; this fixture intentionally repeats the same acknowledgement, so the rate is a measurement of that repetition rather than a leakage count. + +## Expanded benchmark slice + +The corrected full mock run covered 14 default scenarios and 58 scored +probes per phase. Packs contained the expected text for 48/58 (83%) in both +phases; the extractive answerer answered 28/58 (48%) correctly. It recorded +zero host timeouts across 195 calls; pre-turn p50/p95/p99 were 30/76/144 ms. +The three new coding probes and three task-drift probes +each had 3/3 pack hits and 1/3 extractive answers. These separate retrieval +from answer selection. Mock inference still derives no facts or beliefs, so +its captured and synthesis-gain figures are not live model scores. The +200-turn compaction case is opt-in: on the reference engine it returned the +archive key in all three packs. Coalescing identical pending belief builds +reduced synthesis from 41 jobs and 85 seconds to 2 jobs and 4.1 seconds. It +remains opt-in because a live 200-turn run would dominate ordinary sweeps. + +The seeded reference-engine retrieval sweep used one answer-bearing document +among otherwise deterministic near duplicates. Every lexical probe hit at +rank 1. Every paraphrase probe missed; the reference engine uses a toy +keyword/vector scorer, so this is an offline calibration result, not a live +semantic-retrieval estimate. + +| Events | Early lexical / paraphrase | Middle | Late | Probe time range | +| ---: | --- | --- | --- | ---: | +| 100 | 1/1 · 0/1 | 1/1 · 0/1 | 1/1 · 0/1 | 5 ms | +| 1,000 | 1/1 · 0/1 | 1/1 · 0/1 | 1/1 · 0/1 | 40–49 ms | +| 10,000 | 1/1 · 0/1 | 1/1 · 0/1 | 1/1 · 0/1 | 457–1,079 ms | + +The 10,000-item setup exposed quadratic duplicate checking in the reference +engine's default `store_many` path. Its batch override now computes held IDs +once per batch; the 10,000-item accepted writes took 6.8–7.5 seconds. Listing +all writes for settlement took another 14–17 seconds. Neither is included in +the probe latency column. + +Three fresh CortexDB v0.10.4/OpenRouter `tool_heavy` repeats used the corrected +plain pre-turn path, deleted each probe's logged question before the next +score, and waited for enrichment. The six scored questions and model settings +were the same as the historical runs above. + +| Repeat | Initial / synthesis pack hit | Synthesis model answer | Pre-turn timeouts | CortexDB model cost | +| --- | ---: | ---: | ---: | ---: | +| 1 | 3/6 · 6/6 | 5/6 | 7/19 | $0.031 | +| 2 | 1/6 · 6/6 | 6/6 | 10/19 | $0.025 | +| 3 | 4/6 · 6/6 | 6/6 | 6/19 | $0.017 | + +Initial recall still varies substantially and misses the 1500 ms deadline +in 32–53% of calls. All three synthesis packs now include `jmiller` for the +formerly consistent `who-to-ask` miss; removing probe contamination is a +plausible contributor, but these small runs do not isolate its effect from +model variability. The third repeat's stage timings show the bottleneck: +57 scope discoveries had p95 13 ms, while 57 CortexDB recall packs had p95 +2233 ms. End-to-end Cortex fetch p95 was 2235 ms. The pack request includes +server embedding and ranking, which this wire cannot break out further. + +The live safety audit passed 20/20 sibling-tenant, forget, and erase checks +after waiting for a derived record before both deletion operations. It also +confirmed that the restore between them created a new record. An earlier +immediate forget run reported a derived residual while every public item +channel was clear; a second immediate run and the controlled runs passed. +That transient needs a dedicated extraction-versus-forget race reproduction +before claiming the derived layer is always clear at return. + +The first three live 100-document scale runs probed immediately after all +writes appeared in `list`. None of their six packs contained the needle; each +returned only `Case 0`. A controlled repeat that polled direct ranked fetch +found the needle after 5.5 seconds (three attempts). Its lexical host probe +then hit, while the paraphrase probe still missed. Three further fresh +100-document collections used the corrected readiness query. Direct fetch +found the needle after 19.8, 15.1, and 15.9 seconds, respectively. All six +host probes then reached the 1.5-second deadline and received empty packs; +their CortexDB model costs were $0.018, $0.006, and $0.006. This distinguishes +write-to-ranked lag from host-deadline and semantic-retrieval failures. The +`--ranked-wait` report records both whether the direct fetch became ready and +how long it waited; a pack miss after readiness is a separate failure. +The first 1,000-document live attempt exposed a second lag: only 645 writes +were listable at the ordinary 90-second settlement cap. Scale runs now wait +up to 300 seconds, list up to 1,000 items per page, and record that wait +separately from ranked readiness and host probe latency. Three fresh live +1,000-document repeats with a 30-second direct-fetch readiness window were +still unstable: + +| Repeat | List settlement | Ranked readiness wait | Direct fetch ready | Lexical / paraphrase pack | Probe timeouts | CortexDB cost | +| --- | ---: | ---: | --- | --- | ---: | ---: | +| 1 | 19.1 s | 8.3 s | yes | 1/1 · 0/1 | 0/2 | $0.103 | +| 2 | 31.7 s | 1.5 s | yes | 0/1 · 0/1 | 2/2 | $0.099 | +| 3 | 0.4 s | 31.6 s | no | 0/1 · 0/1 | 2/2 | $0.097 | + +These are separate fresh collections, not repeated reads of one index. The +first repeat returned the lexical answer in 1.14 seconds and the paraphrase +missed at 1.37 seconds. Both probes in repeats 2 and 3 reached OpenHuman's +1.5-second deadline. A visible write therefore does not guarantee prompt +ranked retrieval, and even direct-fetch readiness did not guarantee that a +host probe would complete within its deadline. The per-run model cost includes +background extraction and enrichment and is not a per-query price. diff --git a/scripts/memory-eval.sh b/scripts/memory-eval.sh index 29d6ec19..60872a0c 100755 --- a/scripts/memory-eval.sh +++ b/scripts/memory-eval.sh @@ -29,6 +29,8 @@ root="$(cd "$(dirname "$0")/.." && pwd)" models="${MODELS:-mock}" port="${CORTEXDB_PORT:-3145}" label="${LABEL:-cortex-$models}" +out="${OUT_DIR:-$root/target/memory-eval}" +mkdir -p "$out" export CORTEXDB_PORT="$port" url="http://127.0.0.1:$port" compose=(docker compose --project-name "tinymemory-eval-$models-$port" -f "$root/integration/cortexdb/docker-compose.yml") @@ -51,6 +53,37 @@ case "$models" in ;; esac +# A shared budget file keeps a sequence of live runs below one spending cap. +# The OpenRouter key's usage is cumulative, so unrelated concurrent spend is +# counted conservatively too. Refuse a run when there is less than the +# expected single-run reserve left. +if [ "$models" = openrouter ] && [ -n "${BUDGET_USD:-}" ]; then + usage="" + for _ in 1 2 3 4 5; do + response="$(curl --fail --silent --max-time 10 https://openrouter.ai/api/v1/key \ + -H "Authorization: Bearer $OPENROUTER_API_KEY" || true)" + usage="$(printf '%s' "$response" | python3 -c \ + 'import json,sys; print(json.load(sys.stdin)["data"]["usage"])' 2>/dev/null || true)" + [ -n "$usage" ] && break + sleep 2 + done + if [ -z "$usage" ]; then + echo "could not read OpenRouter usage for the live budget" >&2 + exit 1 + fi + start_file="$out/live-budget-start" + if [ ! -f "$start_file" ]; then + printf '%s\n' "$usage" > "$start_file" + fi + start="$(cat "$start_file")" + if ! awk -v now="$usage" -v start="$start" -v cap="$BUDGET_USD" \ + -v reserve="${COST_PER_RUN:-0.5}" \ + 'BEGIN { exit !((now - start + reserve) <= cap) }'; then + echo "live eval budget exhausted: spent since start plus reserve exceeds \$$BUDGET_USD" >&2 + exit 1 + fi +fi + # Like cortexdb-live.sh: never reuse or replace a server someone else runs. if curl --silent --max-time 2 "$url/v1/admin/health" >/dev/null 2>&1; then echo "something already serves $url; pick a free CORTEXDB_PORT" >&2 @@ -84,8 +117,6 @@ curl --fail --silent "$url/v1/admin/ready" >/dev/null || { } echo "CortexDB $(curl --silent "$url/v1/admin/health") at $url, models: $models" -out="${OUT_DIR:-$root/target/memory-eval}" -mkdir -p "$out" CORTEX_DB_URL="$url" CORTEX_DB_KEY="${TINYMEMORY_TEST_CORTEX_KEY:-tinymemory-cortex-test}" \ cargo run --quiet -p tinymemory-integrations --features full --example memory_eval -- \ --label "$label" --json "$out/$label.json" "$@" |