diff --git a/src/openhuman/agent/harness/archivist/recap.rs b/src/openhuman/agent/harness/archivist/recap.rs index 0474e21685..207d55202b 100644 --- a/src/openhuman/agent/harness/archivist/recap.rs +++ b/src/openhuman/agent/harness/archivist/recap.rs @@ -5,6 +5,7 @@ use super::types::ArchivistHook; use crate::openhuman::memory::api::provider::{ ConversationSegment, EpisodicTurn, SummaryContext, SummaryInput, }; +use std::time::Duration; // The fold itself is `MemoryTree::summarise` now, so the DTOs are the // contract's owned ones and `tree_kind` is the wire string the driver // validates rather than the engine's `TreeKind` enum (#5560). @@ -28,6 +29,34 @@ const INPUT_TOKEN_BUDGET: u32 = 50_000; /// See [`INPUT_TOKEN_BUDGET`] for why this is a copy. const SUMMARY_OVERHEAD_RESERVE_TOKENS: u32 = 2_048; +/// How long a segment recap may take before the turn stops waiting for it. +/// +/// This bound is **reasoned, not measured** — say so plainly, because the +/// number should be re-tuned the moment someone has real folds to look at. +/// The attempt to measure it (#6200) produced nothing usable: the workspace it +/// ran in had no summariser configured, so every fold short-circuited to the +/// heuristic without a network call. +/// +/// Three things set it: +/// +/// - **600 s is what it replaces.** `tinyinference`'s request default is the +/// only bound under this call today, and `ChatPrompt` carries no per-call +/// override, so a summariser that accepts a connection and then goes silent +/// holds the turn for ten minutes. +/// - **It must clear two attempts.** `tinymemory`'s retry gives up on a further +/// attempt once ~20 s have elapsed, so its realistic worst case is two folds +/// back to back. A deadline under that would cut the chain before the second +/// attempt and quietly turn the retry into dead code. +/// - **Overshooting is cheap now, undershooting is not.** Since #6156 a fold +/// that misses this writes nothing and leaves the segment `'closed'` for the +/// recovery pass, so a healthy-but-slow fold is deferred rather than lost. +/// Cutting a good fold short only churns. +/// +/// Compare `AUTO_RECALL_BUDGET` (5 s), which bounds a *retrieval* on the same +/// turn path and was set from a field measurement. A fold is an LLM generation +/// over up to `INPUT_TOKEN_BUDGET` tokens, so it is not the same kind of wait. +const RECAP_DEADLINE: Duration = Duration::from_secs(90); + /// Fold one segment's corpus through the **guarded** driver's tree family. /// /// The guard rather than the archivist's own provider handle, because @@ -268,13 +297,51 @@ impl ArchivistHook { "[archivist] summarize_entries: LLM recap segment={segment_id} entries={}", entries.len() ); - let summary_result = fold_through_driver(&corpus_inputs, &summary_ctx).await; + // #6200: bounded because this await is on the turn path — the + // caller's `flush_open_segment` runs before a turn returns — and + // the only bound under it otherwise is tinyinference's 600 s + // request default. Wrapping here rather than passing a timeout + // down keeps it host-side: `ChatPrompt` has no timeout field, so + // the alternative is a `tinymemory` contract change, a release + // and a re-pin. It also bounds the whole retry chain rather than + // a single attempt, which is the thing the turn actually waits + // on. `elapsed_ms` rides every arm so the number above can be + // re-tuned from real folds. + let started = std::time::Instant::now(); + let summary_result = match tokio::time::timeout( + RECAP_DEADLINE, + fold_through_driver(&corpus_inputs, &summary_ctx), + ) + .await + { + Ok(result) => result, + Err(_) => { + let elapsed_ms = started.elapsed().as_millis(); + // WARN, not debug: a fold that outlasts the deadline + // held the turn for the whole of it, and that is the + // symptom worth finding in a log. Falls through to + // the same `(bookend, false)` the error arm returns, + // so nothing is persisted and the segment keeps the + // marker a later pass selects on. + tracing::warn!( + "[archivist] summarize_entries: recap exceeded {:?} \ + elapsed_ms={elapsed_ms} — segment={segment_id} left \ + unsummarised for a later pass", + RECAP_DEADLINE + ); + return ( + super::boundary::fallback_summary(first, last, turn_count), + false, + ); + } + }; + let elapsed_ms = started.elapsed().as_millis(); match summary_result { Ok(output) if !output.content.is_empty() => { tracing::debug!( "[archivist] summarize_entries: LLM recap ok segment={segment_id} \ - chars={}", + chars={} elapsed_ms={elapsed_ms}", output.content.len() ); return (output.content, true); @@ -282,13 +349,14 @@ impl ArchivistHook { Ok(_) => { tracing::debug!( "[archivist] summarize_entries: LLM returned empty — \ - heuristic fallback segment={segment_id}" + heuristic fallback segment={segment_id} elapsed_ms={elapsed_ms}" ); } Err(e) => { tracing::warn!( "[archivist] summarize_entries: LLM recap failed (non-fatal) \ - segment={segment_id}: {e} — heuristic fallback" + segment={segment_id} elapsed_ms={elapsed_ms}: {e} — \ + heuristic fallback" ); } } diff --git a/src/openhuman/agent/harness/archivist/recap_tests.rs b/src/openhuman/agent/harness/archivist/recap_tests.rs index 6f842dfba8..7a16c68bd7 100644 --- a/src/openhuman/agent/harness/archivist/recap_tests.rs +++ b/src/openhuman/agent/harness/archivist/recap_tests.rs @@ -59,3 +59,28 @@ fn segment_membership_falls_back_to_episodic_id() { assert!(entry(None, Some(25), 101.0).is_in_segment(&segment)); assert!(!entry(None, Some(26), 100.001).is_in_segment(&segment)); } + +/// The recap deadline clears two of `tinymemory`'s retry attempts. +/// +/// Not a behavioural test — an arithmetic one, and it exists because the two +/// numbers live in different repositories and nothing else would catch them +/// drifting apart. `tinymemory`'s retry stops starting further attempts once +/// ~20 s have elapsed, so its realistic worst case is two folds back to back. +/// A `RECAP_DEADLINE` under that would cut the chain before the second attempt +/// and silently turn the retry into dead code — the failure this pins. +#[test] +fn the_recap_deadline_leaves_room_for_the_drivers_retry() { + const TINYMEMORY_RETRY_CEILING: Duration = Duration::from_secs(20); + + assert!( + RECAP_DEADLINE > TINYMEMORY_RETRY_CEILING * 2, + "RECAP_DEADLINE ({RECAP_DEADLINE:?}) must clear two attempts of the \ + driver's {TINYMEMORY_RETRY_CEILING:?} retry window, or the retry can \ + never fire a second attempt" + ); + assert!( + RECAP_DEADLINE < Duration::from_secs(600), + "RECAP_DEADLINE must stay well under tinyinference's 600s request \ + default, which is the bound it exists to replace" + ); +} diff --git a/src/openhuman/agent/harness/session/builder/helpers.rs b/src/openhuman/agent/harness/session/builder/helpers.rs index c99fcde2e8..13b114d53f 100644 --- a/src/openhuman/agent/harness/session/builder/helpers.rs +++ b/src/openhuman/agent/harness/session/builder/helpers.rs @@ -121,7 +121,7 @@ pub(super) fn add_memory_prompt_sections( ) -> SystemPromptBuilder { use crate::openhuman::agent::learning::{ any_tool_offered, MemoryAccessSection, MemoryWriteSection, MEMORY_READ_TOOLS, - MEMORY_STORE_TOOL, SAVE_PREFERENCE_TOOL, + MEMORY_STORE_TOOL, MEMORY_WRITE_DELEGATE_TOOL, SAVE_PREFERENCE_TOOL, }; let mut prompt_builder = prompt_builder; if any_tool_offered(&MEMORY_READ_TOOLS, tools, delegation_tools, visible) { @@ -138,17 +138,31 @@ pub(super) fn add_memory_prompt_sections( // about the other (review finding). let preferences = any_tool_offered(&[SAVE_PREFERENCE_TOOL], tools, delegation_tools, visible); let facts = any_tool_offered(&[MEMORY_STORE_TOOL], tools, delegation_tools, visible); - if preferences || facts { - prompt_builder = - prompt_builder.add_section(Box::new(MemoryWriteSection::new(preferences, facts))); + // #6200: asked for as well as the pair, not instead of it. An agent whose + // only write path is the delegate held the tool and no rule about using it + // — the write-side twin of the read-side gap #6183 closed. + let delegate = any_tool_offered( + &[MEMORY_WRITE_DELEGATE_TOOL], + tools, + delegation_tools, + visible, + ); + if preferences || facts || delegate { + prompt_builder = prompt_builder.add_section(Box::new(MemoryWriteSection::new( + preferences, + facts, + delegate, + ))); log::debug!( "[memory_write] prompt section registered for agent={agent_id} \ - save_preference={preferences} memory_store={facts}" + save_preference={preferences} memory_store={facts} \ + manage_profile_memory={delegate}" ); } else { log::debug!( - "[memory_write] skipping MemoryWriteSection — neither memory_store nor \ - save_preference is registered+visible for agent={agent_id}" + "[memory_write] skipping MemoryWriteSection — none of memory_store, \ + save_preference or manage_profile_memory is registered+visible for \ + agent={agent_id}" ); } prompt_builder diff --git a/src/openhuman/agent/harness/session/turn/core_turn.rs b/src/openhuman/agent/harness/session/turn/core_turn.rs index 3cf56aa274..dea2f2473e 100644 --- a/src/openhuman/agent/harness/session/turn/core_turn.rs +++ b/src/openhuman/agent/harness/session/turn/core_turn.rs @@ -673,13 +673,26 @@ impl Agent { // archivist sub-agent that will distil durable facts into the // workspace MEMORY.md file via the `update_memory_md` tool. // - // The spawn is fire-and-forget: the main turn returns the - // user-visible response immediately, and the archivist runs - // asynchronously on the `agentic` tier. We optimistically mark - // the extraction complete right away — if it actually fails, - // we'll just retry on the next threshold window (a few turns - // later), which is the right amount of retry behaviour for a - // librarian task that's idempotent across reruns. + // The archivist sub-agent itself is spawned and runs asynchronously + // on the `agentic` tier. We optimistically mark the extraction + // complete right away — if it actually fails, we'll just retry on the + // next threshold window (a few turns later), which is the right amount + // of retry behaviour for a librarian task that's idempotent across + // reruns. + // + // This call is NOT fire-and-forget, despite spawning one (#6200). It + // is awaited, and before it spawns anything it awaits + // `flush_open_segment`, so the trailing segment's recap runs on this + // path — `result` below is returned only afterwards. The comment here + // used to claim the turn returned immediately; it did not, and with + // `tinyinference`'s 600 s request default underneath that was up to ten + // minutes of a held-open turn. `RECAP_DEADLINE` in `archivist::recap` + // is what bounds it now. + // + // Detaching the flush instead would match the old comment, but it would + // drop the `GUARANTEE:` documented at the flush site — that the + // trailing segment always receives its recap before wind-down. Bounding + // the wait keeps that promise and removes the hazard. if result.is_ok() && self.context.should_extract_session_memory() { self.spawn_session_memory_extraction(session_memory_parent_context) .await; diff --git a/src/openhuman/agent/learning/mod.rs b/src/openhuman/agent/learning/mod.rs index 36a15ecbc7..addc33eae4 100644 --- a/src/openhuman/agent/learning/mod.rs +++ b/src/openhuman/agent/learning/mod.rs @@ -48,7 +48,8 @@ pub use profile_md_renderer::ProfileMdRenderer; pub use prompt_sections::{ any_tool_offered, load_learned_from_cache, memory_write_instruction, LearnedContextSection, MemoryAccessSection, MemoryWriteSection, UserProfileSection, MEMORY_ACCESS_INSTRUCTION, - MEMORY_READ_TOOLS, MEMORY_STORE_TOOL, MEMORY_WRITE_TOOLS, SAVE_PREFERENCE_TOOL, + MEMORY_READ_TOOLS, MEMORY_STORE_TOOL, MEMORY_WRITE_DELEGATE_TOOL, MEMORY_WRITE_TOOLS, + SAVE_PREFERENCE_TOOL, }; pub use reflection::ReflectionHook; pub use schemas::{ diff --git a/src/openhuman/agent/learning/prompt_sections.rs b/src/openhuman/agent/learning/prompt_sections.rs index 9c6aa17d44..3425c180bc 100644 --- a/src/openhuman/agent/learning/prompt_sections.rs +++ b/src/openhuman/agent/learning/prompt_sections.rs @@ -155,14 +155,19 @@ impl PromptSection for MemoryAccessSection { pub struct MemoryWriteSection { preferences: bool, facts: bool, + delegate: bool, } impl MemoryWriteSection { /// `preferences` = `save_preference` is offered here, `facts` = - /// `memory_store` is. + /// `memory_store` is, `delegate` = [`MEMORY_WRITE_DELEGATE_TOOL`] is. #[must_use] - pub fn new(preferences: bool, facts: bool) -> Self { - Self { preferences, facts } + pub fn new(preferences: bool, facts: bool, delegate: bool) -> Self { + Self { + preferences, + facts, + delegate, + } } } @@ -173,11 +178,27 @@ impl MemoryWriteSection { /// Empty when neither tool is offered — which is also when nothing registers /// the section, so the empty string is a guard, not a path in normal use. #[must_use] -pub fn memory_write_instruction(preferences: bool, facts: bool) -> String { +pub fn memory_write_instruction(preferences: bool, facts: bool, delegate: bool) -> String { let route = match (preferences, facts) { (true, true) => "— `save_preference` for preferences, `memory_store` for everything else", (true, false) => "with `save_preference`", (false, true) => "with `memory_store`", + // Only when neither direct tool is held, so an agent that has one + // renders exactly the text it rendered before this arm existed. + // + // `blocking: true` is not a stylistic detail, it is what makes the + // sentence above true (#6200 review). `ArchetypeDelegationTool` + // defaults an omitted `blocking` to `false` and dispatches + // `PreferAsync`, which hands back an immediate reference while the + // worker runs later — so a model told merely to "write with + // `manage_profile_memory`" would confirm a save that had not happened, + // which is the #6048 bug arriving by a new route. The argument is + // advertised on the tool's own schema, so this is a demand the model + // can actually satisfy. + (false, false) if delegate => { + "with `manage_profile_memory` (pass `blocking: true` so the write \ + gates your reply)" + } (false, false) => return String::new(), }; format!( @@ -195,7 +216,11 @@ impl PromptSection for MemoryWriteSection { } fn build(&self, _ctx: &PromptContext<'_>) -> Result { - Ok(memory_write_instruction(self.preferences, self.facts)) + Ok(memory_write_instruction( + self.preferences, + self.facts, + self.delegate, + )) } } @@ -220,8 +245,37 @@ pub const SAVE_PREFERENCE_TOOL: &str = "save_preference"; /// The tool every other remembered fact is written through. pub const MEMORY_STORE_TOOL: &str = "memory_store"; -/// The writing tools [`MemoryWriteSection`] is keyed on. -pub const MEMORY_WRITE_TOOLS: [&str; 2] = [MEMORY_STORE_TOOL, SAVE_PREFERENCE_TOOL]; +/// The delegate an agent writes through when it holds neither direct write +/// tool. +/// +/// Synthesised from `profile_memory_agent`'s `delegate_name`, and the write-side +/// counterpart of `retrieve_memory` in [`MEMORY_READ_TOOLS`]. The orchestrator +/// is configured this way: its visible set carries this delegate and neither +/// [`MEMORY_STORE_TOOL`] nor [`SAVE_PREFERENCE_TOOL`], so keying the section on +/// the direct pair alone dropped the rule for the agent that needed it most — +/// the #6048 case, "got it, saved" with no tool call behind it. +/// +/// The section's promise survives the indirection **only under a blocking +/// delegation**. `profile_memory_agent` holds both direct tools, so a delegated +/// write reaches the same store — but `ArchetypeDelegationTool` defaults to an +/// async dispatch that returns before the worker runs, so the route text demands +/// `blocking: true`. Without that the parent could confirm a save that had not +/// happened yet, which is exactly the bug this section exists to prevent. +pub const MEMORY_WRITE_DELEGATE_TOOL: &str = "manage_profile_memory"; + +/// The writing routes [`MemoryWriteSection`] is keyed on, delegate included. +/// +/// Mirrors [`MEMORY_READ_TOOLS`], which lists `retrieve_memory` beside its two +/// direct tools for the same reason. The live gate in `add_memory_prompt_sections` +/// asks about each of these separately rather than reading this array — the +/// section names the route it found, so it cannot treat them interchangeably — +/// but a reader reaching for "what does the write section care about" should get +/// the whole answer here (#6200 review). +pub const MEMORY_WRITE_TOOLS: [&str; 3] = [ + MEMORY_STORE_TOOL, + SAVE_PREFERENCE_TOOL, + MEMORY_WRITE_DELEGATE_TOOL, +]; /// Whether any of `names` is registered on this session **and** survives tool /// filtering. diff --git a/src/openhuman/agent/learning/prompt_sections_tests_2_tests.rs b/src/openhuman/agent/learning/prompt_sections_tests_2_tests.rs index 769001c38f..ec3e6705be 100644 --- a/src/openhuman/agent/learning/prompt_sections_tests_2_tests.rs +++ b/src/openhuman/agent/learning/prompt_sections_tests_2_tests.rs @@ -332,19 +332,22 @@ fn visible(names: &[&str]) -> HashSet { #[test] fn memory_write_section_states_the_rule_the_bug_needed() { - let section = MemoryWriteSection::new(true, true); + let section = MemoryWriteSection::new(true, true, false); assert_eq!(section.name(), "memory_write"); let rendered = section .build(&prompt_context(LearnedContextData::default())) .unwrap(); - assert_eq!(rendered.trim(), memory_write_instruction(true, true).trim()); + assert_eq!( + rendered.trim(), + memory_write_instruction(true, true, false).trim() + ); assert!(rendered.contains("## Remembering"), "{rendered}"); assert!( rendered.contains("Never say saved"), "the instruction must forbid claiming a save that did not happen: {rendered}" ); // Not context-gated: it renders for an empty learned context too. - let empty = MemoryWriteSection::new(true, true) + let empty = MemoryWriteSection::new(true, true, false) .build(&prompt_context(LearnedContextData::default())) .unwrap(); assert!(!empty.trim().is_empty()); @@ -355,13 +358,13 @@ fn memory_write_section_states_the_rule_the_bug_needed() { /// registering the section with no write tool at all. #[test] fn memory_write_instruction_names_only_the_offered_tools() { - let both = memory_write_instruction(true, true); + let both = memory_write_instruction(true, true, false); assert!( both.contains("`save_preference`") && both.contains("`memory_store`"), "{both}" ); - let preferences_only = memory_write_instruction(true, false); + let preferences_only = memory_write_instruction(true, false, false); assert!( preferences_only.contains("`save_preference`"), "{preferences_only}" @@ -371,7 +374,7 @@ fn memory_write_instruction_names_only_the_offered_tools() { "a session without memory_store must not be sent to it: {preferences_only}" ); - let facts_only = memory_write_instruction(false, true); + let facts_only = memory_write_instruction(false, true, false); assert!(facts_only.contains("`memory_store`"), "{facts_only}"); assert!( !facts_only.contains("`save_preference`"), @@ -395,8 +398,8 @@ fn memory_write_instruction_names_only_the_offered_tools() { /// a tool the session lacks. #[test] fn memory_write_instruction_is_empty_without_a_write_tool() { - assert!(memory_write_instruction(false, false).is_empty()); - let rendered = MemoryWriteSection::new(false, false) + assert!(memory_write_instruction(false, false, false).is_empty()); + let rendered = MemoryWriteSection::new(false, false, false) .build(&prompt_context(LearnedContextData::default())) .unwrap(); assert!(rendered.is_empty(), "{rendered}"); @@ -462,3 +465,105 @@ fn write_tool_gate_counts_delegation_tools_and_either_write_tool() { &visible(&[]) )); } + +// ── The write rule reaches a delegated agent (#6200) ───────────────────────── + +/// An agent whose only write path is the delegate gets the rule, and the rule +/// names the delegate. +/// +/// This is the write-side twin of the gap #6183 closed on the read side. The +/// orchestrator is configured exactly this way — its visible set carries +/// `manage_profile_memory` and neither direct tool — so before this it held a +/// write path and no rule about using it: the #6048 case, "got it, saved" with +/// no tool call behind it. +/// +/// Naming matters as much as presence. Pointing it at `memory_store`, a tool it +/// cannot see, is the failure `any_tool_offered` exists to prevent. +#[test] +fn a_delegate_only_agent_gets_the_write_rule_naming_the_delegate() { + let delegate = named(&[MEMORY_WRITE_DELEGATE_TOOL]); + let none: Vec> = Vec::new(); + + assert!( + any_tool_offered( + &[MEMORY_WRITE_DELEGATE_TOOL], + &delegate, + &none, + &visible(&[]) + ), + "the gate must see the delegate" + ); + + let rendered = memory_write_instruction(false, false, true); + assert!( + rendered.contains(MEMORY_WRITE_DELEGATE_TOOL), + "the rule must name the delegate: {rendered}" + ); + // #6200 review (Codex P1). `ArchetypeDelegationTool` defaults an omitted + // `blocking` to `false` and dispatches async, returning before the worker + // runs. Without demanding `blocking: true` this section would tell the model + // to confirm a save that has not happened — #6048 arriving by a new route, + // through the very rule meant to prevent it. + assert!( + rendered.contains("blocking: true"), + "a delegated write must be demanded as blocking, or the reply can \ + confirm before the write lands: {rendered}" + ); + for absent in [MEMORY_STORE_TOOL, SAVE_PREFERENCE_TOOL] { + assert!( + !rendered.contains(absent), + "the rule named `{absent}`, which a delegate-only agent cannot see: {rendered}" + ); + } +} + +/// An agent holding a direct write tool renders exactly what it rendered before +/// the delegate arm existed. +/// +/// The regression that would matter most here is a silent prompt change for +/// every agent that was already working, so the delegate flag is asserted to be +/// inert whenever either direct tool is present. +#[test] +fn a_direct_write_tool_renders_the_same_text_with_or_without_the_delegate() { + for (preferences, facts) in [(true, true), (true, false), (false, true)] { + assert_eq!( + memory_write_instruction(preferences, facts, false), + memory_write_instruction(preferences, facts, true), + "the delegate flag changed the text for ({preferences}, {facts})" + ); + } + // And the no-write case is still empty rather than falling into the + // delegate arm by accident. + assert!(memory_write_instruction(false, false, false).is_empty()); +} + +/// Read and write both admit their delegate, and neither list admits the +/// other's. +/// +/// #6183 fixed the read side and left the write side behind; the two drifting +/// apart is what produced a half-fixed release. Pinning both directions here +/// makes that specific mistake fail a test rather than ship. +#[test] +fn read_and_write_rules_are_symmetric_about_their_delegates() { + let none: Vec> = Vec::new(); + let readers = named(&["retrieve_memory"]); + let writers = named(&[MEMORY_WRITE_DELEGATE_TOOL]); + + assert!( + any_tool_offered(&MEMORY_READ_TOOLS, &readers, &none, &visible(&[])), + "the read list must admit its delegate" + ); + assert!( + !any_tool_offered(&MEMORY_READ_TOOLS, &writers, &none, &visible(&[])), + "the read list must not admit the write delegate" + ); + assert!( + !any_tool_offered( + &[MEMORY_WRITE_DELEGATE_TOOL], + &readers, + &none, + &visible(&[]) + ), + "the write delegate must not be satisfied by the read delegate" + ); +}