diff --git a/crates/tinymemory-integrations/examples/memory_eval/loop_guard.rs b/crates/tinymemory-integrations/examples/memory_eval/loop_guard.rs index be6c1138..910c329d 100644 --- a/crates/tinymemory-integrations/examples/memory_eval/loop_guard.rs +++ b/crates/tinymemory-integrations/examples/memory_eval/loop_guard.rs @@ -41,9 +41,10 @@ pub(crate) async fn run( engine: Arc, run: u64, policy: &RecallPolicy, + pooled: bool, ) -> Result { let script: Script = serde_json::from_str(include_str!("data/loop_guard.json"))?; - let layout = layout(run, "loop_guard", "main")?; + let layout = layout(run, "loop_guard", "main", pooled)?; let meta = MemoryMeta { namespace: layout.learnings().clone(), ..MemoryMeta::default() diff --git a/crates/tinymemory-integrations/examples/memory_eval/main.rs b/crates/tinymemory-integrations/examples/memory_eval/main.rs index 0e0f5a13..6d0f9d64 100644 --- a/crates/tinymemory-integrations/examples/memory_eval/main.rs +++ b/crates/tinymemory-integrations/examples/memory_eval/main.rs @@ -46,6 +46,8 @@ //! `llm`). //! - `--host openhuman`: run scripted turns through OpenHuman's dated //! pre-turn hook with its 5000 ms deadline, policy and logged tool results. +//! - `--layout v3`: use CortexDB's per-person scope tree and pooled chats. +//! - `--team-limit `: override the host's team-conversation budget. //! - `--loop-guard`: replay a JSON-scripted 500-turn thread and inspect all //! stored items for an injected `` tag. //! @@ -133,6 +135,8 @@ struct Args { label: String, llm: bool, host: String, + layout: String, + team_limit: Option, date_hint: bool, scale_events: Option, scale_position: String, @@ -156,6 +160,8 @@ fn args() -> Result { label: String::new(), llm: false, host: "scripted".into(), + layout: "legacy".into(), + team_limit: None, date_hint: false, scale_events: None, scale_position: "middle".into(), @@ -176,6 +182,8 @@ fn args() -> Result { "--label" => parsed.label = value()?, "--llm" => parsed.llm = true, "--host" => parsed.host = value()?, + "--layout" => parsed.layout = value()?, + "--team-limit" => parsed.team_limit = Some(value()?.parse()?), "--date-hint" => parsed.date_hint = true, "--scale-events" => parsed.scale_events = Some(value()?.parse()?), "--scale-position" => parsed.scale_position = value()?, @@ -190,6 +198,9 @@ fn args() -> Result { if !matches!(parsed.host.as_str(), "scripted" | "openhuman") { return Err(format!("unknown host {}", parsed.host).into()); } + if !matches!(parsed.layout.as_str(), "legacy" | "v3") { + return Err(format!("unknown layout {}", parsed.layout).into()); + } if parsed.loop_guard && parsed.host != "openhuman" { return Err("--loop-guard requires --host openhuman".into()); } @@ -300,23 +311,28 @@ async fn main() -> Result<(), Error> { let (engine, inspector): (Arc, Option) = match args.engine.as_str() { "reference" => (Arc::new(ReferenceEngine::new()), None), - "cortex" if !url.is_empty() => ( - Arc::new(CortexEngine::direct(&url, CortexCredential::api_key(&key))?), - Some(Inspector::new(&url, &key)), - ), + "cortex" if !url.is_empty() => { + let engine = CortexEngine::direct(&url, CortexCredential::api_key(&key))?; + let engine = if args.layout == "v3" { + engine.with_scope_root(&format!("org:eval-{}", std::process::id()), None)? + } else { + engine + }; + (Arc::new(engine), Some(Inspector::new(&url, &key))) + } "cortex" => return Err("--engine cortex needs CORTEX_DB_URL".into()), "tinyhumans" => { let base = std::env::var("TINYHUMANS_API_URL") .unwrap_or_else(|_| TINYHUMANS_API_ENDPOINT.to_string()); let token = std::env::var("TINYHUMANS_API_KEY") .map_err(|_| "--engine tinyhumans needs TINYHUMANS_API_KEY")?; - ( - Arc::new(CortexEngine::tinyhumans( - &base, - Arc::new(StaticBearer::new(token)), - )?), - None, - ) + let engine = CortexEngine::tinyhumans(&base, Arc::new(StaticBearer::new(token)))?; + let engine = if args.layout == "v3" { + engine.with_tenant_root()? + } else { + engine + }; + (Arc::new(engine), None) } other => return Err(format!("unknown engine {other}").into()), }; @@ -342,7 +358,7 @@ async fn main() -> Result<(), Error> { llm.as_ref().map_or("none", |llm| llm.model.as_str()), ); - let policy = if args.host == "openhuman" { + let mut policy = if args.host == "openhuman" { RecallPolicy { team_limit: 0, ..RecallPolicy::default() @@ -353,6 +369,9 @@ async fn main() -> Result<(), Error> { ..RecallPolicy::default() } }; + if let Some(team_limit) = args.team_limit { + policy.team_limit = team_limit; + } let eval = Eval { engine: engine.clone(), inspector, @@ -362,6 +381,7 @@ async fn main() -> Result<(), Error> { enrich_wait, policy, openhuman: args.host == "openhuman", + pooled: args.layout == "v3", date_hint: args.date_hint, ranked_wait: args.ranked_wait, }; @@ -402,7 +422,7 @@ async fn main() -> Result<(), Error> { } let loop_guard = if args.loop_guard { - let report = loop_guard::run(engine.clone(), run, &eval.policy).await?; + let report = loop_guard::run(engine.clone(), run, &eval.policy, eval.pooled).await?; println!( "\n## Loop guard\n\n```json\n{}\n```", serde_json::to_string_pretty(&report)? @@ -417,6 +437,7 @@ async fn main() -> Result<(), Error> { eval.inspector.as_ref(), run, args.expect_derived, + eval.pooled, ) .await?; println!( @@ -469,6 +490,8 @@ async fn main() -> Result<(), Error> { "server": server, "engine": engine.descriptor().id, "host": args.host, + "layout": args.layout, + "team_limit": eval.policy.team_limit, "pre_turn_timeout_ms": PRE_TURN_TIMEOUT.as_millis() as u64, "date_hint": args.date_hint, "scale_events": args.scale_events, @@ -503,9 +526,14 @@ async fn main() -> Result<(), Error> { } /// The layout of `tenant` in `scenario` for this run. -fn layout(run: u64, scenario: &str, tenant: &str) -> Result { +fn layout(run: u64, scenario: &str, tenant: &str, pooled: bool) -> Result { let root = format!("project:eval-{run}-{}-{tenant}", scenario.replace('_', "-")); - Ok(MemoryLayout::new(root.parse()?)?) + let layout = MemoryLayout::new(root.parse()?)?; + if pooled { + Ok(layout.with_pooled_conversations(&"ws:main".parse()?)?) + } else { + Ok(layout) + } } /// The tenants a scenario touches. @@ -556,6 +584,7 @@ struct Eval { enrich_wait: u64, policy: RecallPolicy, openhuman: bool, + pooled: bool, date_hint: bool, ranked_wait: u64, } @@ -573,10 +602,12 @@ impl Eval { None => None, }; let memory = |tenant: &str, agent: &str| -> Result { - Ok( - AgentMemory::new(engine.clone(), layout(run, scenario.name, tenant)?, agent)? - .with_policy(policy.clone()), - ) + Ok(AgentMemory::new( + engine.clone(), + layout(run, scenario.name, tenant, self.pooled)?, + agent, + )? + .with_policy(policy.clone())) }; let epoch = Utc .with_ymd_and_hms(2026, 9, 1, 9, 0, 0) @@ -595,7 +626,7 @@ impl Eval { needle_at, seed, } => { - let node = layout(run, scenario.name, MAIN)? + let node = layout(run, scenario.name, MAIN, self.pooled)? .brain(&tinymemory_tools::BrainSource::Files)?; let started = Instant::now(); for start in (0..*count).step_by(MAX_STORE_MANY) { @@ -623,7 +654,10 @@ impl Eval { title, text, } => { - let brain = Brain::new(engine.clone(), layout(run, scenario.name, tenant)?); + let brain = Brain::new( + engine.clone(), + layout(run, scenario.name, tenant, self.pooled)?, + ); let started = Instant::now(); let ingested = brain .ingest(BrainDocument::new(source.clone(), *text).titled(*title)) @@ -637,7 +671,7 @@ impl Eval { text, confidence, } => { - let layout = layout(run, scenario.name, MAIN)?; + let layout = layout(run, scenario.name, MAIN, self.pooled)?; let meta = MemoryMeta { namespace: layout.learnings().clone(), ..MemoryMeta::default() @@ -689,7 +723,7 @@ impl Eval { let started = Instant::now(); for (tenant, expected) in &writes { self.settle( - &layout(run, scenario.name, tenant)?, + &layout(run, scenario.name, tenant, self.pooled)?, *expected, settle_timeout(scenario.name), ) @@ -704,7 +738,7 @@ impl Eval { // 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); + req.filter = layout(run, scenario.name, MAIN, self.pooled)?.brain_filter(None); let fetch_started = Instant::now(); let page = engine.fetch(req).await?; timings.add("scale direct fetch", ms(fetch_started)); @@ -733,7 +767,7 @@ impl Eval { if std::env::var("CORTEX_DB_KEEP").is_err() { engine .forget(ForgetTarget::Filter( - layout(run, scenario.name, MAIN)?.holistic_filter(), + layout(run, scenario.name, MAIN, self.pooled)?.holistic_filter(), )) .await?; } @@ -795,7 +829,9 @@ impl Eval { ); } for tenant in tenants(scenario) { - let root = layout(run, scenario.name, tenant)?.root().clone(); + let root = layout(run, scenario.name, tenant, self.pooled)? + .root() + .clone(); jobs.push(BackgroundJob::BuildBeliefs { request: ConsolidateRequest::new(Reach::subtree(root)), }); @@ -828,7 +864,7 @@ impl Eval { let questions = questions.join(" "); if let Some(inspector) = &self.inspector { for tenant in tenants(scenario) { - let layout = layout(run, scenario.name, tenant)?; + let layout = layout(run, scenario.name, tenant, self.pooled)?; let node = layout.root().to_string(); let scopes = inspector.scopes(&node).await?; for scope in &scopes { @@ -877,7 +913,7 @@ impl Eval { for tenant in tenants(scenario) { engine .forget(ForgetTarget::Filter( - layout(run, scenario.name, tenant)?.holistic_filter(), + layout(run, scenario.name, tenant, self.pooled)?.holistic_filter(), )) .await?; } @@ -948,7 +984,7 @@ impl Eval { let (engine, run, policy, llm) = (&self.engine, self.run, &self.policy, self.llm.as_ref()); let memory = AgentMemory::new( engine.clone(), - layout(run, scenario.name, probe.tenant)?, + layout(run, scenario.name, probe.tenant, self.pooled)?, probe.agent, )? .with_policy(policy.clone()); @@ -1018,7 +1054,9 @@ impl Eval { } } Via::ContextDoc { heading } => { - let root = layout(run, scenario.name, probe.tenant)?.root().clone(); + let root = layout(run, scenario.name, probe.tenant, self.pooled)? + .root() + .clone(); let spec = ContextSpec { briefs: vec![Brief::new(*heading, probe.question)], reach: Some(Reach::subtree(root)), @@ -1050,8 +1088,8 @@ impl Eval { _ => None, }; if let Some((thread_id, index)) = logged_turn { - let mut filter = - layout(run, scenario.name, probe.tenant)?.conversations_filter(Some(probe.agent)); + let mut filter = layout(run, scenario.name, probe.tenant, self.pooled)? + .conversations_filter(Some(probe.agent)); filter.thread_id = Some(thread_id); filter.turns = Some(TurnRange { first: index, diff --git a/crates/tinymemory-integrations/examples/memory_eval/main_tests.rs b/crates/tinymemory-integrations/examples/memory_eval/main_tests.rs index 71024f09..11d93525 100644 --- a/crates/tinymemory-integrations/examples/memory_eval/main_tests.rs +++ b/crates/tinymemory-integrations/examples/memory_eval/main_tests.rs @@ -21,6 +21,7 @@ async fn a_probe_does_not_leave_its_question_in_the_measured_corpus() { enrich_wait: 0, policy: RecallPolicy::default(), openhuman: true, + pooled: false, date_hint: false, ranked_wait: 0, }; @@ -29,7 +30,7 @@ async fn a_probe_does_not_leave_its_question_in_the_measured_corpus() { .unwrap(); let listed = engine .list(ListRequest::new( - layout(251, scenario.name, probe.tenant) + layout(251, scenario.name, probe.tenant, false) .unwrap() .holistic_filter(), 100, @@ -70,6 +71,7 @@ async fn scale_run_separates_direct_fetch_readiness_from_pack_accuracy() { enrich_wait: 0, policy: RecallPolicy::default(), openhuman: true, + pooled: false, date_hint: false, ranked_wait: 0, }; diff --git a/crates/tinymemory-integrations/examples/memory_eval/safety.rs b/crates/tinymemory-integrations/examples/memory_eval/safety.rs index 8911db7c..ae607dd0 100644 --- a/crates/tinymemory-integrations/examples/memory_eval/safety.rs +++ b/crates/tinymemory-integrations/examples/memory_eval/safety.rs @@ -135,9 +135,10 @@ pub(crate) async fn run( inspector: Option<&Inspector>, run: u64, expect_derived: bool, + pooled: bool, ) -> Result { - let acme = layout(run, "safety_audit", "acme")?; - let globex = layout(run, "safety_audit", "globex")?; + let acme = layout(run, "safety_audit", "acme", pooled)?; + let globex = layout(run, "safety_audit", "globex", pooled)?; let policy = RecallPolicy { team_limit: 0, ..RecallPolicy::default() diff --git a/crates/tinymemory-integrations/examples/memory_eval/safety_tests.rs b/crates/tinymemory-integrations/examples/memory_eval/safety_tests.rs index 9954725e..3e94d231 100644 --- a/crates/tinymemory-integrations/examples/memory_eval/safety_tests.rs +++ b/crates/tinymemory-integrations/examples/memory_eval/safety_tests.rs @@ -6,7 +6,7 @@ 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(); + let report = run(engine, None, 251, false, false).await.unwrap(); assert_eq!(report.violations(), 0, "{report:?}"); assert_eq!(report.checks.len(), 13); } diff --git a/crates/tinymemory-tools/src/context/compile/mod.rs b/crates/tinymemory-tools/src/context/compile/mod.rs index ffaf944f..381db2f3 100644 --- a/crates/tinymemory-tools/src/context/compile/mod.rs +++ b/crates/tinymemory-tools/src/context/compile/mod.rs @@ -85,7 +85,7 @@ impl ContextCompiler { engine: engine_id, generated_at, }; - let pack = recall::run(engine, &holistic(spec), Some(frontmatter), None) + let pack = recall::run(engine, &holistic(spec), Some(frontmatter), None, None) .await .map_err(|error| Error::InvalidSpec(error.to_string()))?; log::debug!( diff --git a/crates/tinymemory-tools/src/layout/mod.rs b/crates/tinymemory-tools/src/layout/mod.rs index afbfd79e..d0627fd4 100644 --- a/crates/tinymemory-tools/src/layout/mod.rs +++ b/crates/tinymemory-tools/src/layout/mod.rs @@ -31,7 +31,8 @@ //! **Pooled conversations.** A host whose agents share one chat history //! ([`MemoryLayout::with_pooled_conversations`]) keeps every agent's turns at //! one node (`ws:main`), each turn labelled with its agent id: an agent's -//! history is that node filtered to its id, and the team's is the whole node. +//! history is that node filtered to its id, and the team section reads it +//! excluding that id. The host's root still bounds both reads. //! //! **Core scopes** share memory beyond one layout. A host that nests every //! tenant under one company node (`ws:acme/team:hive`) can name an ancestor diff --git a/crates/tinymemory-tools/src/lifecycle/mod.rs b/crates/tinymemory-tools/src/lifecycle/mod.rs index 0e8ebfe7..5229c751 100644 --- a/crates/tinymemory-tools/src/lifecycle/mod.rs +++ b/crates/tinymemory-tools/src/lifecycle/mod.rs @@ -87,7 +87,6 @@ use crate::brain::Brain; use crate::layout::{CoreScope, MemoryLayout}; use crate::recall::{ ContextPack, HolisticRecall, LateHint, ScopeSection, SectionQuery, ThreadWindow, - holistic_recall, }; use crate::tools::MemoryTools; @@ -428,7 +427,13 @@ impl AgentMemory { request.exclude_thread = Some(window); let (logged, pack) = join( self.engine.store_with(item, WriteOptions::accepted()), - crate::recall::run(self.engine.as_ref(), &request, None, hint), + crate::recall::run( + self.engine.as_ref(), + &request, + None, + hint, + self.team_exclusion(&request), + ), ) .await; let pack = pack?; @@ -560,11 +565,6 @@ impl AgentMemory { } } - /// Learnings, each core scope, brain (at most - /// [`BRAIN_SCOPES_PER_TURN`] scopes), this agent's history, then the - /// team's, each filled by fetch; a zero limit leaves its section out. A - /// layout that pools conversations has no team section: it would read - /// the same node as the history. /// The thread's own latest turns, up to `limit`. fn thread_section(&self, thread_id: &str, limit: usize) -> ScopeSection { ScopeSection::latest( @@ -577,6 +577,10 @@ impl AgentMemory { ) } + /// Learnings, each core scope, brain (at most + /// [`BRAIN_SCOPES_PER_TURN`] scopes), this agent's history, then other + /// agents' turns. A zero limit leaves a section out. In a pooled layout, + /// history and team read the same node with complementary agent filters. fn standard_sections(&self) -> Vec { let policy = &self.policy; let learnings = ( @@ -589,11 +593,6 @@ impl AgentMemory { .core .iter() .map(|scope| (scope.heading.as_str(), scope.filter(), scope.limit, None)); - let team_limit = if self.layout.pools_conversations() { - 0 - } else { - policy.team_limit - }; let layout = [ ( BRAIN_HEADING, @@ -610,7 +609,7 @@ impl AgentMemory { ( TEAM_HEADING, self.layout.conversations_filter(None), - team_limit, + policy.team_limit, None, ), ]; @@ -636,12 +635,34 @@ impl AgentMemory { } } + /// The team section is the last section with its heading and shared + /// conversation filter; only lifecycle recall applies this exclusion. + fn team_exclusion<'a>(&'a self, request: &HolisticRecall) -> Option<(usize, &'a str)> { + if self.policy.team_limit == 0 { + return None; + } + let team_filter = self.layout.conversations_filter(None); + request + .sections + .iter() + .rposition(|section| section.heading == TEAM_HEADING && section.filter == team_filter) + .map(|index| (index, self.agent_id.as_str())) + } + async fn read( &self, query: Option, sections: Vec, ) -> Result { - holistic_recall(self.engine.as_ref(), &self.request(query, sections)).await + let request = self.request(query, sections); + crate::recall::run( + self.engine.as_ref(), + &request, + None, + None, + self.team_exclusion(&request), + ) + .await } /// One turn of `thread_id` as a one-turn conversation at this agent's diff --git a/crates/tinymemory-tools/src/lifecycle/mod_tests.rs b/crates/tinymemory-tools/src/lifecycle/mod_tests.rs index 8c4e0afa..e78966af 100644 --- a/crates/tinymemory-tools/src/lifecycle/mod_tests.rs +++ b/crates/tinymemory-tools/src/lifecycle/mod_tests.rs @@ -857,17 +857,87 @@ async fn pooled_agents_log_to_one_node_and_keep_their_own_history() { .unwrap(); assert_eq!(listed.items.len(), 2, "both agents' turns at ws:main"); - // Each agent's history is its own turns of the pool; a pooled layout - // has no team section, so another agent's turn stays out of the pack. + // Own history stays filtered to one agent, while the shared pool also + // supplies the other agent's relevant turn in the team section. let md = support.recall("refund").await.unwrap().markdown; let history = md.find("## This agent's history").unwrap(); assert!(md[history..].contains("refund delayed"), "{md}"); - assert!(!md.contains("## Team conversations"), "{md}"); - assert!(!md.contains("deploy failed"), "{md}"); + let team = md.find("## Team conversations").unwrap(); + assert!(!md[history..team].contains("deploy failed"), "{md}"); + assert!(md[team..].contains("deploy failed"), "{md}"); + assert!(!md[team..].contains("refund delayed"), "{md}"); +} + +#[tokio::test] +async fn pooled_team_reads_past_own_ranked_turns_without_crossing_roots() { + let engine = Arc::new(ReferenceEngine::new()); + let node: Namespace = "ws:main".parse().unwrap(); + let mine = MemoryLayout::new("team:mine".parse().unwrap()) + .unwrap() + .with_pooled_conversations(&node) + .unwrap(); + let other = MemoryLayout::new("team:other".parse().unwrap()) + .unwrap() + .with_pooled_conversations(&node) + .unwrap(); + let coder = AgentMemory::new(engine.clone(), mine.clone(), "coder") + .unwrap() + .with_policy(RecallPolicy { + learnings_limit: 0, + brain_limit: 0, + history_limit: 0, + team_limit: 1, + ..RecallPolicy::default() + }); + let support = AgentMemory::new(engine.clone(), mine, "support").unwrap(); + let outsider = AgentMemory::new(engine, other, "outsider").unwrap(); + support + .pre_turn(PreTurn::new( + "ticket", + 0, + "refund escalation belongs to support", + )) + .await + .unwrap(); + outsider + .pre_turn(PreTurn::new( + "foreign", + 0, + "refund secret from another team", + )) + .await + .unwrap(); + // A one-hit section used to stop after five pages of these own turns. + for index in 0..30 { + coder + .pre_turn(PreTurn::new( + format!("own-{index}"), + 0, + "refund escalation refund escalation", + )) + .await + .unwrap(); + } + let pack = coder.recall("refund escalation").await.unwrap(); + let team = pack + .sections + .iter() + .find(|section| section.heading == TEAM_HEADING) + .unwrap(); + assert!( + team.hits + .iter() + .all(|hit| hit.meta.agent_id.as_deref() != Some("coder")) + ); + assert!( + pack.markdown + .contains("refund escalation belongs to support") + ); + assert!(!pack.markdown.contains("refund secret from another team")); } #[test] -fn the_brain_reads_few_scopes_and_pooled_chats_have_no_team_section() { +fn the_brain_reads_few_scopes_and_pooled_chats_have_a_team_section() { let engine = Arc::new(ReferenceEngine::new()); let sections = memory(&engine, "s").standard_sections(); for section in §ions { @@ -880,7 +950,8 @@ fn the_brain_reads_few_scopes_and_pooled_chats_have_no_team_section() { .any(|section| section.heading == TEAM_HEADING) ); - // Pooled, every agent's history is the team's node: no team section. + // Pooled, history filters to this agent while team reads the same node + // for other agents' turns. let pooled = MemoryLayout::default() .with_pooled_conversations(&"ws:main".parse().unwrap()) .unwrap(); @@ -892,7 +963,12 @@ fn the_brain_reads_few_scopes_and_pooled_chats_have_no_team_section() { .collect(); assert_eq!( headings, - [LEARNINGS_HEADING, BRAIN_HEADING, HISTORY_HEADING] + [ + LEARNINGS_HEADING, + BRAIN_HEADING, + HISTORY_HEADING, + TEAM_HEADING + ] ); } diff --git a/crates/tinymemory-tools/src/recall/gather.rs b/crates/tinymemory-tools/src/recall/gather.rs index d78d5f4b..a4149af5 100644 --- a/crates/tinymemory-tools/src/recall/gather.rs +++ b/crates/tinymemory-tools/src/recall/gather.rs @@ -52,11 +52,10 @@ pub(super) const DATED_DEPTH: usize = 3; /// Page size of a [`SectionQuery::Latest`] listing. const LATEST_PAGE: usize = 100; -/// Most listing pages read before ranking; a ceiling, not a target. +/// Most listing pages read before ranking, except for lifecycle team recall. const LATEST_MAX_PAGES: usize = 50; -/// Most ranked pages a fetch section reads to replace hits its exclusions -/// dropped. +/// Most ranked pages a generic fetch section reads to replace excluded hits. const FETCH_MAX_PAGES: usize = 5; /// What one section read. @@ -91,9 +90,13 @@ pub(super) async fn section( section: &ScopeSection, beliefs: usize, dated: bool, + excluded_agent: Option<&str>, ) -> Gathered { let section_started = Instant::now(); - let keep = |hit: &Hit| !request.excludes(hit); + let keep = |hit: &Hit| { + !request.excludes(hit) + && excluded_agent.is_none_or(|agent| hit.meta.agent_id.as_deref() != Some(agent)) + }; let want = wanted(request, section); // A date reorders a fetch section's hits after the read, so read deeper // than the section shows: a hit from the right day ranked just past the @@ -124,9 +127,12 @@ pub(super) async fn section( section, question, fetch_want, - 0, &keep, - request.refers_to.clone(), + FetchOptions { + beliefs: 0, + refers_to: request.refers_to.clone(), + scan_all: excluded_agent.is_some(), + }, ) .await } @@ -140,12 +146,29 @@ pub(super) async fn section( { Some(query) => { let hint = request.refers_to.clone(); - fetch(engine, section, query, fetch_want, beliefs, &keep, hint).await + fetch( + engine, + section, + query, + fetch_want, + &keep, + FetchOptions { + beliefs, + refers_to: hint, + scan_all: excluded_agent.is_some(), + }, + ) + .await + } + None => { + with_listed_beliefs(engine, section, want, &keep, excluded_agent.is_some()) + .await } - None => with_listed_beliefs(engine, section, want, &keep).await, } } - SectionQuery::Latest => with_listed_beliefs(engine, section, want, &keep).await, + SectionQuery::Latest => { + with_listed_beliefs(engine, section, want, &keep, excluded_agent.is_some()).await + } }; log::trace!(target: "tinymemory_eval_timing", "recall_section={}", section_started.elapsed().as_secs_f64() * 1_000.0); match outcome { @@ -276,10 +299,11 @@ async fn with_listed_beliefs( section: &ScopeSection, want: usize, keep: &(dyn Fn(&Hit) -> bool + Sync), + scan_all: bool, ) -> tinymemory_api::Result<(Vec, Vec)> { if !reads_learnings(section) { return Ok(( - latest(engine, §ion.filter, want, keep).await?, + latest(engine, §ion.filter, want, keep, scan_all).await?, Vec::new(), )); } @@ -289,7 +313,7 @@ async fn with_listed_beliefs( .clone() .unwrap_or_else(|| Reach::subtree(Namespace::ROOT)); let (hits, beliefs) = join( - latest(engine, §ion.filter, want, keep), + latest(engine, §ion.filter, want, keep, scan_all), engine.beliefs(BeliefsRequest::new(reach, want)), ) .await; @@ -355,37 +379,50 @@ fn interleave(first: Vec, second: impl IntoIterator) -> Vec, + scan_all: bool, +} + +/// Ranked pages of hits; an engine that declares no fetch mode is read /// newest first instead. async fn fetch( engine: &dyn MemoryEngine, section: &ScopeSection, query: &str, limit: usize, - beliefs: usize, keep: &(dyn Fn(&Hit) -> bool + Sync), - refers_to: Option, + options: FetchOptions, ) -> tinymemory_api::Result<(Vec, Vec)> { let Some(mode) = preferred_mode(engine) else { return Ok(( - latest(engine, §ion.filter, limit, keep).await?, + latest(engine, §ion.filter, limit, keep, options.scan_all).await?, Vec::new(), )); }; // Exclusions (the prompt's thread window, shown ids) are dropped page by // page. Only when a page lost hits to them and too few remain is the next // page read, so the common case is still one request (each page is a - // round trip on a hosted engine), and the walk is capped. + // round trip on a hosted engine). Generic recall is capped; lifecycle + // team recall must scan until it finds another agent or reaches the end. let mut hits: Vec = Vec::new(); let mut first_beliefs = Vec::new(); let mut cursor: Option = None; - for page_no in 0..FETCH_MAX_PAGES { + let mut seen_cursors = HashSet::new(); + let max_pages = if options.scan_all { + usize::MAX + } else { + FETCH_MAX_PAGES + }; + for page_no in 0..max_pages { let mut request = FetchRequest::new(query, mode, limit); request.filter = section.filter.clone(); request.max_scopes = section.max_scopes; - request.beliefs = if page_no == 0 { beliefs } else { 0 }; + request.beliefs = if page_no == 0 { options.beliefs } else { 0 }; request.cursor = cursor.take(); - request.refers_to = refers_to.clone(); + request.refers_to = options.refers_to.clone(); let page = engine.fetch(request).await?; if page_no == 0 { first_beliefs = page.beliefs; @@ -395,7 +432,9 @@ async fn fetch( hits.extend(page.hits.into_iter().filter(|hit| keep(hit))); let lost_some = hits.len() - before < fetched; match page.next_cursor { - Some(next) if hits.len() < limit && lost_some => cursor = Some(next), + Some(next) if hits.len() < limit && lost_some && seen_cursors.insert(next.clone()) => { + cursor = Some(next); + } _ => break, } } @@ -414,17 +453,24 @@ async fn latest( filter: &MetaFilter, limit: usize, keep: &(dyn Fn(&Hit) -> bool + Sync), + scan_all: bool, ) -> tinymemory_api::Result> { let mut all: Vec = Vec::new(); let mut cursor: Option = None; - for _ in 0..LATEST_MAX_PAGES { + let mut seen_cursors = HashSet::new(); + let max_pages = if scan_all { + usize::MAX + } else { + LATEST_MAX_PAGES + }; + for _ in 0..max_pages { let mut request = ListRequest::new(filter.clone(), LATEST_PAGE); request.cursor = cursor.take(); let page = engine.list(request).await?; all.extend(page.items.into_iter().filter(|hit| keep(hit))); match page.next_cursor { - Some(next) => cursor = Some(next), - None => break, + Some(next) if seen_cursors.insert(next.clone()) => cursor = Some(next), + _ => break, } } all.sort_by(|a, b| { @@ -484,6 +530,7 @@ pub(super) fn settle( request: &HolisticRecall, section: &ScopeSection, gathered: Gathered, + excluded_agent: Option<&str>, shown: &mut HashSet, ) -> Settled { let hits = match gathered { @@ -496,6 +543,9 @@ pub(super) fn settle( .into_iter() .filter(|hit| kinds.is_empty() || kinds.contains(&hit.kind)) .filter(|hit| !request.excludes(hit) && !shown.contains(&hit.id)) + .filter(|hit| { + excluded_agent.is_none_or(|agent| hit.meta.agent_id.as_deref() != Some(agent)) + }) .take(section.limit) .collect(); if section.heading == crate::lifecycle::HISTORY_HEADING diff --git a/crates/tinymemory-tools/src/recall/mod.rs b/crates/tinymemory-tools/src/recall/mod.rs index 83ae21c7..030640c5 100644 --- a/crates/tinymemory-tools/src/recall/mod.rs +++ b/crates/tinymemory-tools/src/recall/mod.rs @@ -93,7 +93,7 @@ pub async fn holistic_recall( engine: &dyn MemoryEngine, request: &HolisticRecall, ) -> Result { - run(engine, request, None, None).await + run(engine, request, None, None, None).await } /// [`holistic_recall`] for a question whose date is still being worked out. @@ -114,29 +114,30 @@ pub async fn holistic_recall_dated( request: &HolisticRecall, hint: impl Future> + Send, ) -> Result { - run(engine, request, None, Some(Box::pin(hint))).await + run(engine, request, None, Some(Box::pin(hint)), None).await } /// A date that may arrive while the sections are read. pub(crate) type LateHint<'a> = Pin> + Send + 'a>>; -/// [`holistic_recall`], optionally with `context.md` frontmatter and a late -/// date. +/// [`holistic_recall`], optionally with `context.md` frontmatter, a late +/// date, and one lifecycle team section that excludes its current agent. pub(crate) async fn run( engine: &dyn MemoryEngine, request: &HolisticRecall, frontmatter: Option>, late: Option>, + excluded_team: Option<(usize, &str)>, ) -> Result { request.validate()?; let beliefs = gather::belief_budget(request); let dated = late.is_some() || request.refers_to.is_some(); - let reads = join_all( - request - .sections - .iter() - .map(|section| gather::section(engine, request, section, beliefs, dated)), - ); + let reads = join_all(request.sections.iter().enumerate().map(|(index, section)| { + let excluded_agent = excluded_team + .filter(|(team, _)| *team == index) + .map(|(_, agent)| agent); + gather::section(engine, request, section, beliefs, dated, excluded_agent) + })); let (mut gathered, late) = match late { Some(late) => join(reads, late).await, None => (reads.await, None), @@ -167,8 +168,11 @@ pub(crate) async fn run( let mut rendered_from = Vec::new(); let mut skipped = Vec::new(); let mut shown = HashSet::new(); - for (section, outcome) in request.sections.iter().zip(gathered) { - match gather::settle(request, section, outcome, &mut shown) { + for (index, (section, outcome)) in request.sections.iter().zip(gathered).enumerate() { + let excluded_agent = excluded_team + .filter(|(team, _)| *team == index) + .map(|(_, agent)| agent); + match gather::settle(request, section, outcome, excluded_agent, &mut shown) { Settled::Filled(rendered, hits) => { rendered_from.push(rendered); sections.push(hits); diff --git a/docs/architecture/cortex-layout.md b/docs/architecture/cortex-layout.md index f4353cca..77a2997b 100644 --- a/docs/architecture/cortex-layout.md +++ b/docs/architecture/cortex-layout.md @@ -45,6 +45,12 @@ org:42/ws:main/app:flows/service:newsletter/app:learnings a workflow's memory org:42/app:flows/service:digest/app:documents a workflow with no workspace ``` +The chat scope is shared by the person's agents. Each turn retains its +`agent_id` in TinyMemory's item metadata. History queries select one agent; +team queries read the same exact CortexDB scope and discard that agent's +turns before applying the section limit. Neither query widens to a sibling +person's `org:` root. + ### The hosted wire: the tenant's root The TinyHumans backend pins every scope below the caller's tenant root diff --git a/docs/evals/openhuman-host.md b/docs/evals/openhuman-host.md index d47345f8..8d17e5aa 100644 --- a/docs/evals/openhuman-host.md +++ b/docs/evals/openhuman-host.md @@ -6,14 +6,48 @@ engine and probes as the default profile, so pack accuracy can be compared without changing the stored fixture data. It is a mirror inside TinyMemory; there is no dependency on `openhuman-core`. +Use `--layout v3 --team-limit 3` to measure cross-agent recall from pooled +chats on CortexDB's per-person scope tree. The host mirror defaults to +`team_limit = 0` for historical comparison; the JSON report records the layout +and team limit so those runs cannot be mistaken for the new OpenHuman default. + +## V3 pooled team recall, 2026-10-10 + +The team section now reads the same `ws:main` conversation scope as history, +excluding the current agent's turns before applying its limit. The CortexDB +v0.10.4 runs below used fresh collections for each setting, the same 5-second +OpenHuman deadline, and one team limit of three turns. The full mock run has +58 scored probes per phase; the live OpenRouter slice covers the three +`team_handoff` and four `conflicts` probes. Each result was the same in recall +and synthesis. + +| Run | Team limit | Pack hits | Extractive answers | Model answers | Probe timeouts | +| --- | ---: | ---: | ---: | ---: | ---: | +| Full mock v3 | 0 | 48/58 | 31/58 | — | 0 | +| Full mock v3 | 3 | 51/58 | 34/58 | — | 0 | +| Live v3 slice | 0 | 3/7 | 1/7 | 2/7 | 0 | +| Live v3 slice | 3 | 7/7 | 4/7 | 5/7 | 0 | + +In the full mock run, `team_handoff/duplicate-invoices`, +`team_handoff/account-id`, and `conflicts/promise` changed from misses to +hits; no scored probe moved the other way. The mock paraphrase +`team_handoff/billed-twice` still misses. With real embeddings it reaches the +pack, but the model slice remains only seven questions and one fresh run per +setting. One baseline model answer graded correct despite an empty pack, so +model scores must be read beside pack hits. In the final rerun, CortexDB +model cost was $0.0196 with team recall off and $0.0206 with it on; that +difference is noise at this sample size. + The source references below use OpenHuman commit `cf16716f4f` for the hook -behavior and [OpenHuman PR #7346](https://github.com/tinyhumansai/openhuman/pull/7346) -for the 5000 ms default. -Update them when the host's defaults or hook behavior changes. +behavior, [OpenHuman PR #7346](https://github.com/tinyhumansai/openhuman/pull/7346) +for the 5000 ms deadline, and +[OpenHuman PR #7352](https://github.com/tinyhumansai/openhuman/pull/7352) +for the proposed three-turn team default. Update 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` | +| 1200-token budget, 8 learnings, 6 brain, 6 history, 3 team, beliefs every 10 turns | `main.rs`, OpenHuman policy with `--team-limit 3` | `crates/openhuman-core/src/config/schema/memory.rs:278-336` | | Plain `pre_turn` by default, resumed hook after compaction, optional dated path with `--date-hint`, empty pack after 5000 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` | diff --git a/docs/specs/agent-memory.md b/docs/specs/agent-memory.md index 734dd912..6ade9fac 100644 --- a/docs/specs/agent-memory.md +++ b/docs/specs/agent-memory.md @@ -179,9 +179,13 @@ skipped, engine }`. section per core scope ([core-scopes.md](core-scopes.md); none by default), Brain (all documents, reading at most `BRAIN_SCOPES_PER_TURN` (4) scopes: those the query names, then the most recently written), this agent's - history, and team conversations (every agent; none in a layout that pools - conversations, where they are the history's node). A zero limit in - `RecallPolicy` leaves a section out. + history, and team conversations (other agents under the same layout root). + In a pooled layout both conversation sections read the same chat node: + history selects this agent's id, while team excludes it. The team limit + counts selected turns, not agents or threads. Excluded own turns cannot + exhaust a fixed retrieval page cap before eligible team turns are reached. + A zero limit in `RecallPolicy` leaves a section out. Team recall never + reaches another person's root. - **`pre_turn` never fails on an engine error.** A failed log is reported in `TurnContext::log_error` and the pack is still returned. - **`post_turn` reports belief builds.** It returns a `BuildBeliefs` job for