From 6df61b210c440b81277e796af1d5901166957a5b Mon Sep 17 00:00:00 2001 From: Shanu Date: Thu, 10 Sep 2026 14:49:37 +0530 Subject: [PATCH 1/8] fix(archivist): stop persisting, embedding and enriching from a heuristic recap `on_segment_closed` computed `summarize_entries`' `produced_by_llm` flag and bound it to `_from_llm`, then ran three consumers unconditionally: it wrote the heuristic `fallback_summary` bookend as the segment's durable summary, embedded that text into `segment_embeddings`, and handed it to goals enrichment as "Recent conversation recap". When the inference provider was briefly unreachable, a transient network condition therefore produced a permanent degradation of stored memory, invisible to the user. All three consumers now honour the flag. A recap that did not come from the model is logged and dropped. Nothing is persisted, deliberately. `segment_set_summary` also flips the row to `status='summarised'`, and that flip is the one thing that removes the segment from the `segments_pending_summary` query a later re-summarisation pass would select on. Leaving the row `'closed'` with a NULL summary is the provenance marker the issue asks for, at no schema cost and with no change to the tinymemory contract. The bookend itself is derived from the segment's first and last turn, both of which stay in the episodic store, so it can be recomputed at any time. The WARN is gated on `summariser_available`: with no summariser configured for the workspace this is the steady state rather than a degradation, and warning per segment close would bury the case this is about. Also deletes `archivist_tests_part_01_tests.rs`. It has been dead since cc99ba9c6 (#6161) deleted its parent module `archivist_tests.rs` and left the part file behind with no `mod` declaration anywhere in the tree. It references helpers and engine symbols that no longer exist in the test build, so it cannot be revived as-is. Closes #6156 --- .../agent/harness/archivist/lifecycle.rs | 128 +++- .../harness/archivist/lifecycle_tests.rs | 121 ++++ .../agent/harness/archivist/recap.rs | 10 +- .../harness/archivist_tests_part_01_tests.rs | 599 ------------------ .../memory/guard/test_support_part_01.rs | 12 +- .../memory/guard/test_support_part_02.rs | 2 +- 6 files changed, 242 insertions(+), 630 deletions(-) create mode 100644 src/openhuman/agent/harness/archivist/lifecycle_tests.rs delete mode 100644 src/openhuman/agent/harness/archivist_tests_part_01_tests.rs diff --git a/src/openhuman/agent/harness/archivist/lifecycle.rs b/src/openhuman/agent/harness/archivist/lifecycle.rs index 8817a0a1e2..aceb0dafda 100644 --- a/src/openhuman/agent/harness/archivist/lifecycle.rs +++ b/src/openhuman/agent/harness/archivist/lifecycle.rs @@ -21,6 +21,23 @@ use std::time::{SystemTime, UNIX_EPOCH}; /// question of the same role. const RECAP_INFERENCE_ROLE: &str = "summarization"; +/// Whether a finalize-time recap may be handed to a consumer that will treat +/// it as real conversation content. +/// +/// `from_llm == false` means [`ArchivistHook::summarize_entries`] returned the +/// `boundary::fallback_summary` bookend — the segment's first and last 200 +/// characters joined by a pipe. That string is not a summary of anything; it +/// is a placeholder that reads like one. +/// +/// The emptiness term is a belt: a successful LLM recap is already non-empty +/// by construction, because `summarize_entries` only reports `true` from the +/// `Ok(output) if !output.content.is_empty()` arm. It fires on a shape that +/// cannot occur today, and exists so a future arm cannot quietly reintroduce +/// one. +pub(super) fn recap_is_usable(from_llm: bool, summary: &str) -> bool { + from_llm && !summary.trim().is_empty() +} + impl ArchivistHook { /// Create an Archivist hook over the workspace's bound memory driver. /// @@ -301,9 +318,16 @@ impl ArchivistHook { /// Called when a segment is closed. /// - /// Produces a segment recap (LLM if a chat provider is configured, - /// otherwise the heuristic fallback), embeds the recap, extracts - /// heuristic events, and updates the user profile. + /// Produces a segment recap, extracts heuristic events, updates the user + /// profile, and pipes the segment's raw turns into the memory tree. + /// + /// What happens to the recap depends on where it came from (#6156). An LLM + /// recap is persisted with `set_segment_summary`, embedded, and handed to + /// goals enrichment. The heuristic `boundary::fallback_summary` bookend is + /// none of those things — it is logged and dropped, leaving the segment + /// unsummarised on purpose. Everything downstream of the recap (events, + /// profile facets, tree ingest) runs either way, because none of it reads + /// the summary. /// /// Soft-fallback contract (mirrors `LlmSummariser`): this function /// never returns `Err`; all failures are logged and ignored. @@ -344,33 +368,74 @@ impl ArchivistHook { .join(". "); // ── Segment recap (LLM or heuristic fallback) ──────────────────── - let (summary, _from_llm) = self + let (summary, from_llm) = self .summarize_entries(&segment_entries, &segment.segment_id, segment.turn_count) .await; - // Persist the recap. - let set_summary = match self.episodic() { - Some(episodic) => { - episodic - .set_segment_summary(&segment.segment_id, &summary, now) - .await - } - None => return, + // Hoisted above the two recap arms because what is missing here is the + // driver, not the summary: with no episodic family there is nothing to + // write on either arm, and this is the exit that path has always taken. + let Some(episodic) = self.episodic() else { + return; }; - if let Err(e) = set_summary { - tracing::warn!("[archivist] failed to set segment summary: {e}"); + + if recap_is_usable(from_llm, &summary) { + // Persist the recap. + let set_summary = episodic + .set_segment_summary(&segment.segment_id, &summary, now) + .await; + if let Err(e) = set_summary { + tracing::warn!("[archivist] failed to set segment summary: {e}"); + } else { + tracing::debug!( + "[archivist] recap persisted segment={} summary_chars={}", + segment.segment_id, + summary.len() + ); + } + + // ── Finalize-time embedding ─────────────────────────────────── + self.embed_segment_recap(&segment.segment_id, &summary, now) + .await; } else { - tracing::debug!( - "[archivist] recap persisted segment={} summary_chars={}", - segment.segment_id, - summary.len() - ); + // #6156. The bookend is not written, not embedded, and not handed + // to goals enrichment. + // + // Persisting it would be worse than storing nothing, because + // `segment_set_summary` also flips the row to `status='summarised'` + // — and that flip is the one thing that removes the segment from + // the `segments_pending_summary` query a later re-summarisation + // pass selects on. Leaving the row `'closed'` with a NULL summary + // IS the provenance marker, at no schema cost. + // + // Nothing is lost by dropping the bookend either: it is derived + // from the segment's first and last turn, both of which stay in the + // episodic store, so it can be recomputed at any time. + // + // WARN only when a summariser was actually expected. With none + // configured for the workspace at all this is the steady state + // rather than a degradation, and warning once per segment close + // would bury the case #6156 is about — a summariser that exists + // and did not answer — in noise from the case that is working as + // intended. + if self.summariser_available { + tracing::warn!( + "[archivist] no LLM recap for segment={} ({} turns) — summary NOT persisted, \ + NOT embedded, NOT sent to goals enrichment; the segment stays unsummarised \ + so a later pass can recap it once a summariser answers", + segment.segment_id, + segment.turn_count, + ); + } else { + tracing::debug!( + "[archivist] no summariser for this workspace — segment={} ({} turns) left \ + unsummarised", + segment.segment_id, + segment.turn_count, + ); + } } - // ── Finalize-time embedding ─────────────────────────────────────── - self.embed_segment_recap(&segment.segment_id, &summary, now) - .await; - // ── Heuristic event extraction ──────────────────────────────────── if !segment_text.is_empty() { let extracted = extract_events_heuristic(&segment_text); @@ -469,7 +534,13 @@ impl ArchivistHook { // the user's durable goals list stays fresh. Feed it the fresh recap // as context. Detached + non-fatal: never blocks segment close. if let Some(ref cfg) = self.config { - if cfg.learning.goals_enrichment_enabled && !summary.trim().is_empty() { + // #6156: the recap term leads deliberately. Handing the bookend to + // the goals agent means an LLM call whose entire context is two + // truncated utterances, and it will invent durable goals out of + // that noise. Ordering it ahead of the config flag also keeps it + // evaluated whenever a config is attached, rather than being + // short-circuited away by an unrelated toggle. + if recap_is_usable(from_llm, &summary) && cfg.learning.goals_enrichment_enabled { tracing::debug!( "[memory_goals] segment closed — spawning goals enrichment \ session={session_id} segment={}", @@ -499,6 +570,11 @@ impl ArchivistHook { /// zero entries) and an empty embed input is guaranteed to 400 from /// the upstream embedding API (#13021). The segment is sealed without /// an embedding row; subsequent recap edits can re-embed. + /// + /// Since #6156 the finalize caller also declines to call this at all for a + /// heuristic recap, so this guard is no longer the only thing standing + /// between a bookend stub and the embedder. It now covers direct callers + /// and any future one that has not made that decision for itself. pub(super) async fn embed_segment_recap(&self, segment_id: &str, summary: &str, now: f64) { if summary.trim().is_empty() { tracing::warn!( @@ -573,3 +649,7 @@ impl ArchivistHook { } } } + +#[cfg(test)] +#[path = "lifecycle_tests.rs"] +mod tests; diff --git a/src/openhuman/agent/harness/archivist/lifecycle_tests.rs b/src/openhuman/agent/harness/archivist/lifecycle_tests.rs new file mode 100644 index 0000000000..76693426ff --- /dev/null +++ b/src/openhuman/agent/harness/archivist/lifecycle_tests.rs @@ -0,0 +1,121 @@ +//! Finalize-path tests for `on_segment_closed` (#6156). +//! +//! These drive the hook through the **contract**, not an engine: a +//! `RecordingProvider` serves every family and records what actually reached +//! it, so "the write did not happen" is asserted against the driver's call log +//! rather than inferred from an empty store. +//! +//! `ArchivistHook::new` leaves `summariser_available == false`, which is what +//! makes the heuristic arm of `summarize_entries` deterministic here — no chat +//! model is built, resolved or called anywhere in these tests. + +use super::*; +use crate::openhuman::memory::guard::test_support::RecordingProvider; + +const SESSION: &str = "lifecycle-6156"; + +fn turn(id: i64, role: &str, content: &str) -> EpisodicTurn { + EpisodicTurn { + id: Some(id), + session_id: SESSION.into(), + timestamp: 100.0 + id as f64, + role: role.into(), + content: content.into(), + lesson: None, + tool_calls_json: None, + cost_microdollars: 0, + } +} + +/// A closed segment spanning both turns above. +/// +/// `start_seq`/`end_seq` are deliberately `None`: the episodic read path builds +/// its `SessionEntry`s with `sequence: None`, so membership is decided on +/// `turn.id` against the episodic-id bounds. +fn segment() -> ConversationSegment { + ConversationSegment { + segment_id: "seg-6156".into(), + session_id: SESSION.into(), + namespace: "global".into(), + start_episodic_id: 1, + end_episodic_id: Some(2), + start_timestamp: 100.0, + end_timestamp: Some(103.0), + turn_count: 2, + summary: None, + embedding: None, + open: false, + start_seq: None, + end_seq: None, + } +} + +fn methods(recording: &RecordingProvider) -> Vec { + recording.calls().into_iter().map(|c| c.method).collect() +} + +/// The bug in #6156: a heuristic bookend was persisted as the segment's durable +/// summary and embedded, both unconditionally. +/// +/// Asserting `episodic.session_turns` is present matters as much as the +/// absences — without it the test would also pass if `on_segment_closed` had +/// bailed out before reaching the recap at all. +#[tokio::test] +async fn heuristic_recap_is_not_persisted_or_embedded() { + let recording = Arc::new(RecordingProvider::new().with_session_turns(vec![ + turn(1, "user", "How do I pin a submodule?"), + turn(2, "assistant", "Record the gitlink at the commit you want."), + ])); + let provider: Arc = recording.clone(); + let hook = ArchivistHook::new(provider, true); + + hook.on_segment_closed(&segment(), SESSION, 200.0).await; + + let methods = methods(&recording); + assert!( + methods.iter().any(|m| m == "episodic.session_turns"), + "the finalize path must have read the segment's turns; got {methods:?}" + ); + for forbidden in [ + "episodic.set_segment_summary", + "scoring.embedder_slug", + "scoring.embed_text", + "episodic.upsert_segment_embedding", + ] { + assert!( + !methods.iter().any(|m| m == forbidden), + "{forbidden} must not run for a heuristic recap; got {methods:?}" + ); + } +} + +/// A segment whose turns are all outside its bounds short-circuits before the +/// recap — the pre-existing empty-entries exit, not the new heuristic arm. +/// +/// Pinned so the new branch cannot quietly become the thing that handles an +/// empty segment. +#[tokio::test] +async fn empty_segment_still_short_circuits() { + let recording = Arc::new(RecordingProvider::new()); + let provider: Arc = recording.clone(); + let hook = ArchivistHook::new(provider, true); + + hook.on_segment_closed(&segment(), SESSION, 200.0).await; + + let methods = methods(&recording); + assert!( + !methods.iter().any(|m| m == "episodic.set_segment_summary"), + "an entryless segment must not be summarised; got {methods:?}" + ); +} + +/// The predicate both consumers gate on. The emptiness cases cannot be produced +/// by `summarize_entries` today; they are pinned so a future arm that could +/// produce one still lands on "not usable". +#[test] +fn recap_is_usable_truth_table() { + assert!(recap_is_usable(true, "a real recap")); + assert!(!recap_is_usable(false, "a heuristic bookend")); + assert!(!recap_is_usable(true, "")); + assert!(!recap_is_usable(true, " \n\t")); +} diff --git a/src/openhuman/agent/harness/archivist/recap.rs b/src/openhuman/agent/harness/archivist/recap.rs index f23fdd1269..0474e21685 100644 --- a/src/openhuman/agent/harness/archivist/recap.rs +++ b/src/openhuman/agent/harness/archivist/recap.rs @@ -194,11 +194,11 @@ impl ArchivistHook { /// /// Returns `(text, produced_by_llm)`. `produced_by_llm == false` means the /// LLM was unavailable / failed / returned empty and `text` is the shallow - /// heuristic `fallback_summary` bookend stub. That stub is an acceptable - /// durable last-resort on the *finalize* path, but callers driving the - /// **live prompt** (rolling recap → compaction) must treat - /// `produced_by_llm == false` as "no real recap" and fall back to their - /// own strategy — the stub must never become live compaction text. + /// heuristic `fallback_summary` bookend stub. Both callers treat that as + /// "no real recap" (#6156): the rolling recap keeps it out of the live + /// prompt so it can never become compaction text, and the finalize path + /// writes nothing at all — no summary row, no embedding, no goals + /// enrichment — rather than sealing the segment around a placeholder. pub(super) async fn summarize_entries( &self, entries: &[&EpisodicTurn], diff --git a/src/openhuman/agent/harness/archivist_tests_part_01_tests.rs b/src/openhuman/agent/harness/archivist_tests_part_01_tests.rs deleted file mode 100644 index f3949976f9..0000000000 --- a/src/openhuman/agent/harness/archivist_tests_part_01_tests.rs +++ /dev/null @@ -1,599 +0,0 @@ -use super::*; - -#[tokio::test] -async fn archivist_indexes_turn() { - let (_tmp, client, provider) = setup_provider(); - let conn = client.profile_conn(); - let hook = ArchivistHook::new(provider.clone(), true); - - let ctx = TurnContext { - user_message: "What is Rust?".into(), - assistant_response: "Rust is a systems programming language.".into(), - tool_calls: vec![], - turn_duration_ms: 500, - session_id: Some("test-session".into()), - agent_id: None, - entrypoint: None, - iteration_count: 1, - }; - - hook.on_turn_complete(&ctx).await.unwrap(); - - let entries = fts5::episodic_session_entries(&conn, "test-session").unwrap(); - assert_eq!(entries.len(), 2); - assert_eq!(entries[0].role, "user"); - assert_eq!(entries[1].role, "assistant"); -} - -#[tokio::test] -async fn archivist_creates_segment_on_first_turn() { - let (_tmp, client, provider) = setup_provider(); - let conn = client.profile_conn(); - let hook = ArchivistHook::new(provider.clone(), true); - - let ctx = TurnContext { - user_message: "Hello world".into(), - assistant_response: "Hi there!".into(), - tool_calls: vec![], - turn_duration_ms: 100, - session_id: Some("seg-test".into()), - agent_id: None, - entrypoint: None, - iteration_count: 1, - }; - - hook.on_turn_complete(&ctx).await.unwrap(); - - let open = seg::open_segment_for_session(&conn, "seg-test").unwrap(); - assert!(open.is_some()); - assert_eq!(open.unwrap().turn_count, 1); -} - -#[tokio::test] -async fn archivist_detects_topic_change_boundary() { - let (_tmp, client, provider) = setup_provider(); - let conn = client.profile_conn(); - let hook = ArchivistHook::new(provider.clone(), true); - - hook.on_turn_complete(&TurnContext { - user_message: "Tell me about Rust".into(), - assistant_response: "Rust is great.".into(), - tool_calls: vec![], - turn_duration_ms: 100, - session_id: Some("boundary-test".into()), - agent_id: None, - entrypoint: None, - iteration_count: 1, - }) - .await - .unwrap(); - - hook.on_turn_complete(&TurnContext { - user_message: "How about its memory safety?".into(), - assistant_response: "It uses ownership.".into(), - tool_calls: vec![], - turn_duration_ms: 100, - session_id: Some("boundary-test".into()), - agent_id: None, - entrypoint: None, - iteration_count: 2, - }) - .await - .unwrap(); - - hook.on_turn_complete(&TurnContext { - user_message: "Switching to a different topic now. I prefer dark mode.".into(), - assistant_response: "Noted about dark mode.".into(), - tool_calls: vec![], - turn_duration_ms: 100, - session_id: Some("boundary-test".into()), - agent_id: None, - entrypoint: None, - iteration_count: 3, - }) - .await - .unwrap(); - - let segments = seg::segments_by_namespace(&conn, "global", 10).unwrap(); - assert!( - segments.len() >= 2, - "Expected at least 2 segments, got {}", - segments.len() - ); -} - -#[tokio::test] -async fn archivist_extracts_failure_lesson() { - let (_tmp, client, provider) = setup_provider(); - let conn = client.profile_conn(); - let hook = ArchivistHook::new(provider.clone(), true); - - let ctx = TurnContext { - user_message: "Run tests".into(), - assistant_response: "Tests failed.".into(), - tool_calls: vec![ToolCallRecord { - name: "shell".into(), - arguments: serde_json::json!({"command": "cargo test"}), - success: false, - output_summary: "shell: failed (error)".into(), - duration_ms: 3000, - }], - turn_duration_ms: 3500, - session_id: Some("test-session-2".into()), - agent_id: None, - entrypoint: None, - iteration_count: 2, - }; - - hook.on_turn_complete(&ctx).await.unwrap(); - - let entries = fts5::episodic_session_entries(&conn, "test-session-2").unwrap(); - let assistant_entry = entries.iter().find(|e| e.role == "assistant").unwrap(); - assert!(assistant_entry.lesson.as_ref().unwrap().contains("shell")); -} - -#[tokio::test] -async fn disabled_archivist_is_noop() { - let hook = ArchivistHook::disabled(); - let ctx = TurnContext { - user_message: "test".into(), - assistant_response: "test".into(), - tool_calls: vec![], - turn_duration_ms: 0, - session_id: None, - agent_id: None, - entrypoint: None, - iteration_count: 0, - }; - hook.on_turn_complete(&ctx).await.unwrap(); -} - -#[test] -fn extract_profile_key_works() { - let key = extract_profile_key("I prefer dark mode for coding", "preference"); - assert!(key.starts_with("preference_")); - assert!(key.contains("prefer")); -} - -#[tokio::test] -async fn archivist_accumulates_turns_in_segment() { - let (_tmp, client, provider) = setup_provider(); - let conn = client.profile_conn(); - let hook = ArchivistHook::new(provider.clone(), true); - - let session = "accum-session"; - - for i in 1..=3 { - hook.on_turn_complete(&TurnContext { - user_message: format!("Turn number {i}"), - assistant_response: format!("Response {i}"), - tool_calls: vec![], - turn_duration_ms: 50, - session_id: Some(session.into()), - agent_id: None, - entrypoint: None, - iteration_count: i, - }) - .await - .unwrap(); - } - - let open_seg = seg::open_segment_for_session(&conn, session) - .unwrap() - .expect("Expected an open segment after 3 turns"); - - assert_eq!( - open_seg.turn_count, 3, - "Segment should have accumulated 3 turns, got {}", - open_seg.turn_count - ); -} - -#[tokio::test] -async fn archivist_extracts_preference_event_on_boundary() { - let (_tmp, client, provider) = setup_provider(); - let conn = client.profile_conn(); - let hook = ArchivistHook::new(provider.clone(), true); - - let session = "pref-boundary-session"; - - hook.on_turn_complete(&TurnContext { - user_message: "Tell me about Rust ownership".into(), - assistant_response: "Ownership is a key concept in Rust.".into(), - tool_calls: vec![], - turn_duration_ms: 100, - session_id: Some(session.into()), - agent_id: None, - entrypoint: None, - iteration_count: 1, - }) - .await - .unwrap(); - - hook.on_turn_complete(&TurnContext { - user_message: "I prefer dark mode for all my editors".into(), - assistant_response: "Good to know! Dark mode is easier on the eyes.".into(), - tool_calls: vec![], - turn_duration_ms: 100, - session_id: Some(session.into()), - agent_id: None, - entrypoint: None, - iteration_count: 2, - }) - .await - .unwrap(); - - hook.on_turn_complete(&TurnContext { - user_message: "Switching to a different topic — how does Tokio work?".into(), - assistant_response: "Tokio is an async runtime.".into(), - tool_calls: vec![], - turn_duration_ms: 100, - session_id: Some(session.into()), - agent_id: None, - entrypoint: None, - iteration_count: 3, - }) - .await - .unwrap(); - - let events = ev::events_by_type(&conn, "global", "preference", 20).unwrap(); - assert!( - !events.is_empty(), - "Expected at least one preference event after segment close; got 0." - ); - let has_dark_mode = events - .iter() - .any(|e| e.content.to_lowercase().contains("prefer")); - assert!( - has_dark_mode, - "Expected a preference event mentioning 'prefer', found: {:?}", - events.iter().map(|e| &e.content).collect::>() - ); -} - -// ── Phase 0: episodic_capture_enabled independent of learning.enabled ──────── - -/// When `learning.enabled = false` but `episodic_capture_enabled = true`, -/// the ArchivistHook (constructed directly, as builder.rs would produce) -/// must still write 2 episodic_log rows (user + assistant) and create/advance -/// a segment. This verifies the core contract: episodic capture runs -/// regardless of the learning inference stack toggle. -#[tokio::test] -async fn phase0_episodic_rows_and_segment_without_learning_enabled() { - let (_tmp, client, provider) = setup_provider(); - let conn = client.profile_conn(); - // Simulate what builder.rs does when learning.enabled=false but - // episodic_capture_enabled=true: construct the hook directly with - // the SQLite conn, enabled=true. No config attached (no LLM recap - // or tree ingest — those are gated by learning.enabled / chat_to_tree_enabled). - let hook = ArchivistHook::new(provider.clone(), true); - - let session = "phase0-test-session"; - - hook.on_turn_complete(&TurnContext { - user_message: "Hello, what is Rust?".into(), - assistant_response: "Rust is a systems language.".into(), - tool_calls: vec![], - turn_duration_ms: 100, - session_id: Some(session.into()), - agent_id: None, - entrypoint: None, - iteration_count: 1, - }) - .await - .unwrap(); - - // Verify 2 episodic rows were written. - let entries = fts5::episodic_session_entries(&conn, session).unwrap(); - assert_eq!( - entries.len(), - 2, - "Expected 2 episodic rows (user + assistant), got {}", - entries.len() - ); - assert_eq!(entries[0].role, "user"); - assert_eq!(entries[1].role, "assistant"); - - // Verify a segment was created. - let open_seg = seg::open_segment_for_session(&conn, session) - .unwrap() - .expect("Expected an open segment after first turn"); - assert_eq!(open_seg.turn_count, 1); - - // Add a second turn to verify segment advances. - hook.on_turn_complete(&TurnContext { - user_message: "Tell me more about ownership.".into(), - assistant_response: "Ownership prevents data races.".into(), - tool_calls: vec![], - turn_duration_ms: 100, - session_id: Some(session.into()), - agent_id: None, - entrypoint: None, - iteration_count: 2, - }) - .await - .unwrap(); - - let entries2 = fts5::episodic_session_entries(&conn, session).unwrap(); - assert_eq!( - entries2.len(), - 4, - "Expected 4 episodic rows after 2 turns, got {}", - entries2.len() - ); - let open_seg2 = seg::open_segment_for_session(&conn, session) - .unwrap() - .expect("Expected an open segment after 2 turns"); - assert_eq!( - open_seg2.turn_count, 2, - "Segment should have 2 turns, got {}", - open_seg2.turn_count - ); -} - -/// When a segment closes, the LLM chat provider recap is used (verified by -/// a non-empty segment summary) and an embedding row is written to -/// `segment_embeddings`. -#[tokio::test] -async fn phase1_llm_recap_and_embedding_on_segment_close() { - let (_tmp, client, provider) = setup_provider(); - let conn = client.profile_conn(); - let hook = hook_with_stubs(provider.clone()); - - let session = "phase1-recap-test"; - - // Turn 1 — opens first segment. - hook.on_turn_complete(&TurnContext { - user_message: "Tell me about Rust ownership".into(), - assistant_response: "Rust's ownership model prevents data races.".into(), - tool_calls: vec![], - turn_duration_ms: 100, - session_id: Some(session.into()), - agent_id: None, - entrypoint: None, - iteration_count: 1, - }) - .await - .unwrap(); - - // Turn 2 — continues same segment. - hook.on_turn_complete(&TurnContext { - user_message: "What about the borrow checker?".into(), - assistant_response: "The borrow checker enforces ownership rules at compile time.".into(), - tool_calls: vec![], - turn_duration_ms: 100, - session_id: Some(session.into()), - agent_id: None, - entrypoint: None, - iteration_count: 2, - }) - .await - .unwrap(); - - // Turn 3 — topic change triggers a boundary → closes first segment → recap + embed fire. - hook.on_turn_complete(&TurnContext { - user_message: "Completely different topic: what is async/await in Python?".into(), - assistant_response: "Python asyncio enables concurrent programming.".into(), - tool_calls: vec![], - turn_duration_ms: 100, - session_id: Some(session.into()), - agent_id: None, - entrypoint: None, - iteration_count: 3, - }) - .await - .unwrap(); - - // Verify segments exist. - let segments = seg::segments_by_namespace(&conn, "global", 10).unwrap(); - assert!( - segments.len() >= 2, - "Expected at least 2 segments (closed + open), got {}", - segments.len() - ); - - // Find the closed segment (has a summary). - let closed = segments - .iter() - .find(|s| s.summary.as_ref().map(|s| !s.is_empty()).unwrap_or(false)); - assert!( - closed.is_some(), - "Expected at least one closed segment with a non-empty summary" - ); - - let closed_seg = closed.unwrap(); - let summary = closed_seg.summary.as_ref().unwrap(); - // The stub provider returns a fixed string — verify it was persisted. - assert!( - summary.contains("stub recap"), - "Expected summary to contain 'stub recap', got: {:?}", - summary - ); -} - -/// `flush_open_segment` must force-close the trailing open segment and -/// trigger recap + embedding even without a boundary-triggering turn. -#[tokio::test] -async fn phase1_flush_open_segment_finalizes_trailing_segment() { - let (_tmp, client, provider) = setup_provider(); - let conn = client.profile_conn(); - let hook = hook_with_stubs(provider.clone()); - - let session = "phase1-flush-test"; - - // Write 2 turns — stays in one open segment (no topic boundary fires). - for i in 1..=2 { - hook.on_turn_complete(&TurnContext { - user_message: format!("Question about Rust turn {i}"), - assistant_response: format!("Answer about Rust turn {i}"), - tool_calls: vec![], - turn_duration_ms: 50, - session_id: Some(session.into()), - agent_id: None, - entrypoint: None, - iteration_count: i, - }) - .await - .unwrap(); - } - - // Confirm the segment is still open (no boundary fired). - let open_seg_before = seg::open_segment_for_session(&conn, session).unwrap(); - assert!( - open_seg_before.is_some(), - "Expected an open segment before flush" - ); - - // Flush — should force-close, recap, and embed. - hook.flush_open_segment(session).await; - - // Segment should now be closed (no open segment for this session). - let open_seg_after = seg::open_segment_for_session(&conn, session).unwrap(); - assert!( - open_seg_after.is_none(), - "Expected no open segment after flush_open_segment" - ); - - // The formerly-open segment should now have a summary. - let segments = seg::segments_by_namespace(&conn, "global", 10).unwrap(); - let flushed = segments.iter().find(|s| { - s.session_id == session && s.summary.as_ref().map(|s| !s.is_empty()).unwrap_or(false) - }); - assert!( - flushed.is_some(), - "Expected flushed segment to have a non-empty summary" - ); -} - -/// After a single turn (no segment boundary), the tree must have ZERO chunks — -/// the per-turn pipe_turn_to_tree path no longer exists. -#[tokio::test] -async fn phase2_no_per_turn_tree_write() { - with_stub_chat_provider(phase2_no_per_turn_tree_write_inner()).await -} - -/// When a segment closes (boundary triggered), exactly ONE tree ingest fires -/// for that segment containing all its turns — not one ingest per turn. -#[tokio::test] -async fn phase2_exactly_one_tree_ingest_per_segment_close() { - with_stub_chat_provider(phase2_exactly_one_tree_ingest_per_segment_close_inner()).await -} - -/// The ingested leaf messages must carry the episodic-provenance `source_ref` -/// in the expected format: -/// `agent://session/{session_id}/segment/{segment_id}#ep{start}-{end}`. -/// -/// Also verifies that `source_id` is the constant `"conversations:agent"`. -#[tokio::test] -async fn phase2_provenance_stamped_on_leaf_and_source_id_is_constant() { - with_stub_chat_provider(phase2_provenance_stamped_on_leaf_and_source_id_is_constant_inner()) - .await -} - -/// The ingested content must be the raw prose turns (user + assistant text), -/// NOT equal to the LLM recap text. The recap lives only in the STM segment -/// layer; the tree must ingest raw evidence so it can build its own summaries. -#[tokio::test] -async fn phase2_ingested_content_is_raw_prose_not_recap() { - with_stub_chat_provider(phase2_ingested_content_is_raw_prose_not_recap_inner()).await -} - -/// `flush_open_segment` must also trigger the tree ingest for the trailing -/// open segment (same as on_segment_closed at a topic boundary). -#[tokio::test] -async fn phase2_flush_also_triggers_tree_ingest() { - with_stub_chat_provider(phase2_flush_also_triggers_tree_ingest_inner()).await -} - -// ── #13021 empty/whitespace recap embed-skip guard ─────────────────────────── -// -// `on_segment_closed` defends against ever passing an empty or whitespace -// recap into the embedder by calling `embed_segment_recap`, which short- -// circuits before `Embedder::embed` runs. The skip is unreachable through -// the current `summarize_entries` call graph today (the heuristic -// `fallback_summary` always returns non-empty text), so these tests drive -// `embed_segment_recap` directly to lock the guard against future -// regressions where `summarize_entries` could return `""`. - -/// An empty recap must short-circuit before any scoring call. -/// -/// Uses `RecordingProvider` so we can assert that `scoring.embed_text` is -/// absent — confirming the guard fired before the scoring path, not merely -/// that no row landed in a DB queried with the wrong model_signature key. -#[tokio::test] -async fn embed_segment_recap_skips_empty_summary() { - use crate::openhuman::memory::guard::test_support::RecordingProvider; - let recording = Arc::new(RecordingProvider::new()); - let provider: Arc = recording.clone(); - let hook = hook_with_stubs(provider); - - hook.embed_segment_recap("seg-empty-recap", "", 3.0).await; - - let calls = recording.calls(); - let methods: Vec<&str> = calls.iter().map(|c| c.method.as_str()).collect(); - assert!( - !methods.contains(&"scoring.embed_text"), - "scoring.embed_text must not be called for an empty recap; got {methods:?}" - ); -} - -/// Whitespace-only recaps (newlines, tabs, spaces) must also short-circuit -/// — the upstream provider rejects whitespace inputs the same way it -/// rejects empty inputs (#13021). -/// -/// Uses `RecordingProvider` so we can assert that `scoring.embed_text` is -/// absent — confirming the guard fired before the scoring path, not merely -/// that no row landed in a DB queried with the wrong model_signature key. -#[tokio::test] -async fn embed_segment_recap_skips_whitespace_summary() { - use crate::openhuman::memory::guard::test_support::RecordingProvider; - let recording = Arc::new(RecordingProvider::new()); - let provider: Arc = recording.clone(); - let hook = hook_with_stubs(provider); - - hook.embed_segment_recap("seg-ws-recap", " \n\t ", 3.0) - .await; - - let calls = recording.calls(); - let methods: Vec<&str> = calls.iter().map(|c| c.method.as_str()).collect(); - assert!( - !methods.contains(&"scoring.embed_text"), - "scoring.embed_text must not be called for a whitespace-only recap; got {methods:?}" - ); -} - -/// Positive control: a non-empty recap must reach `embedder_slug` then -/// `embed_text` on the scoring family, in that order, and pass the recap text -/// verbatim to `embed_text`. -#[tokio::test] -async fn embed_segment_recap_reaches_scoring_for_non_empty_summary() { - use crate::openhuman::memory::guard::test_support::RecordingProvider; - let recording = Arc::new(RecordingProvider::new()); - let provider: Arc = recording.clone(); - let hook = hook_with_stubs(provider); - - hook.embed_segment_recap("seg-ok-recap", "real recap text", 3.0) - .await; - - let calls = recording.calls(); - let methods: Vec<&str> = calls.iter().map(|c| c.method.as_str()).collect(); - - let slug_pos = methods - .iter() - .position(|&m| m == "scoring.embedder_slug") - .unwrap_or_else(|| panic!("scoring.embedder_slug must be called; got {methods:?}")); - let embed_pos = methods - .iter() - .position(|&m| m == "scoring.embed_text") - .expect("scoring.embed_text must be called for a non-empty recap"); - - assert!( - slug_pos < embed_pos, - "scoring.embedder_slug ({slug_pos}) must be called before scoring.embed_text ({embed_pos})" - ); - assert_eq!( - calls[embed_pos].content.as_deref(), - Some("real recap text"), - "embed_text must receive the recap text verbatim" - ); -} diff --git a/src/openhuman/memory/guard/test_support_part_01.rs b/src/openhuman/memory/guard/test_support_part_01.rs index 870e065395..6e43ccd9e5 100644 --- a/src/openhuman/memory/guard/test_support_part_01.rs +++ b/src/openhuman/memory/guard/test_support_part_01.rs @@ -22,7 +22,8 @@ use crate::openhuman::memory::api::provider::types::{ }; use crate::openhuman::memory::api::provider::{ AddressBookSeedOutcome, ChunkDetail, ChunkEmbedding, ChunkQuery, CoverWindowQuery, EntityMatch, - EpisodicEvent, FacetType, FastRetrieveQuery, MemoryChunks, MemoryCodingSessions, MemoryCore, + EpisodicEvent, EpisodicTurn, FacetType, FastRetrieveQuery, MemoryChunks, MemoryCodingSessions, + MemoryCore, MemoryDiff, MemoryDocuments, MemoryEntities, MemoryEpisodic, MemoryGoals, MemoryGraph, MemoryIngest, MemoryMaintenance, MemoryPeople, MemoryPortability, MemoryProfile, MemoryProvider, MemoryRecall, MemoryRetrieval, MemoryScoring, MemorySourceSink, @@ -86,6 +87,9 @@ pub struct RecordingProvider { /// What `namespaces` returns, so a namespace can look populated (Lane B /// asks for the count before it pays for an embed) without a real store. namespace_summaries: Mutex>, + /// What `session_turns` returns, so the archivist's finalize path can be + /// driven past its empty-entries early return without an engine behind it. + session_turns: Mutex>, } impl Default for RecordingProvider { @@ -102,6 +106,7 @@ impl RecordingProvider { fast_retrieve_result: Mutex::new(RetrievalResponse::default()), namespace_hits: Mutex::new(Vec::new()), namespace_summaries: Mutex::new(Vec::new()), + session_turns: Mutex::new(Vec::new()), } } @@ -125,6 +130,11 @@ impl RecordingProvider { self } + pub fn with_session_turns(self, turns: Vec) -> Self { + *self.session_turns.lock().unwrap() = turns; + self + } + fn record(&self, call: Call) { self.calls.lock().unwrap().push(call); } diff --git a/src/openhuman/memory/guard/test_support_part_02.rs b/src/openhuman/memory/guard/test_support_part_02.rs index 20d024cc10..08240b2506 100644 --- a/src/openhuman/memory/guard/test_support_part_02.rs +++ b/src/openhuman/memory/guard/test_support_part_02.rs @@ -247,7 +247,7 @@ impl MemoryEpisodic for RecordingProvider { ) -> Result, MemoryError> { self.record(Call::plain("episodic.session_turns")); - Ok(vec![]) + Ok(self.session_turns.lock().unwrap().clone()) } async fn open_segment( From b6c97e28f31d8de94e953dfa18eac435a5fab670 Mon Sep 17 00:00:00 2001 From: Shanu Date: Thu, 10 Sep 2026 15:08:42 +0530 Subject: [PATCH 2/8] chore(memory): move RecordingProvider's inherent impl out of part 01 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `scripts/ci/check-openhuman-rust-layout.mjs` caps files under `src/openhuman` at 750 lines, and `test_support_part_01.rs` sat at 747 — the `session_turns` seeder pushed it to 759. The inherent `impl RecordingProvider` block is the one self-contained unit that relocates without splitting a trait implementation across files, so it moves to part 03 whole. The parts are `include!`d into one module, so nothing about resolution changes. part 01 is now 698 lines, part 03 is 124. Documents the accessors while they are in hand — the move pulls them into this changeset, and every other item in these files carries its rationale. Same for the two helpers in `lifecycle_tests.rs`. --- .../harness/archivist/lifecycle_tests.rs | 4 + .../memory/guard/test_support_part_01.rs | 61 -------------- .../memory/guard/test_support_part_03.rs | 83 +++++++++++++++++++ 3 files changed, 87 insertions(+), 61 deletions(-) diff --git a/src/openhuman/agent/harness/archivist/lifecycle_tests.rs b/src/openhuman/agent/harness/archivist/lifecycle_tests.rs index 76693426ff..6604b56dbc 100644 --- a/src/openhuman/agent/harness/archivist/lifecycle_tests.rs +++ b/src/openhuman/agent/harness/archivist/lifecycle_tests.rs @@ -14,6 +14,9 @@ use crate::openhuman::memory::guard::test_support::RecordingProvider; const SESSION: &str = "lifecycle-6156"; +/// One episodic turn. `id` is load-bearing: the episodic read path builds its +/// `SessionEntry`s with `sequence: None`, so segment membership is decided on +/// `turn.id` against the segment's episodic-id bounds. fn turn(id: i64, role: &str, content: &str) -> EpisodicTurn { EpisodicTurn { id: Some(id), @@ -50,6 +53,7 @@ fn segment() -> ConversationSegment { } } +/// The driver's call log reduced to method names — what these tests assert on. fn methods(recording: &RecordingProvider) -> Vec { recording.calls().into_iter().map(|c| c.method).collect() } diff --git a/src/openhuman/memory/guard/test_support_part_01.rs b/src/openhuman/memory/guard/test_support_part_01.rs index 6e43ccd9e5..4b0f4fc355 100644 --- a/src/openhuman/memory/guard/test_support_part_01.rs +++ b/src/openhuman/memory/guard/test_support_part_01.rs @@ -98,67 +98,6 @@ impl Default for RecordingProvider { } } -impl RecordingProvider { - pub fn new() -> Self { - Self { - calls: Mutex::new(Vec::new()), - recall_result: Mutex::new(Vec::new()), - fast_retrieve_result: Mutex::new(RetrievalResponse::default()), - namespace_hits: Mutex::new(Vec::new()), - namespace_summaries: Mutex::new(Vec::new()), - session_turns: Mutex::new(Vec::new()), - } - } - - pub fn with_recall_result(self, entries: Vec) -> Self { - *self.recall_result.lock().unwrap() = entries; - self - } - - pub fn with_fast_retrieve_result(self, response: RetrievalResponse) -> Self { - *self.fast_retrieve_result.lock().unwrap() = response; - self - } - - pub fn with_namespace_hits(self, hits: Vec) -> Self { - *self.namespace_hits.lock().unwrap() = hits; - self - } - - pub fn with_namespace_summaries(self, summaries: Vec) -> Self { - *self.namespace_summaries.lock().unwrap() = summaries; - self - } - - pub fn with_session_turns(self, turns: Vec) -> Self { - *self.session_turns.lock().unwrap() = turns; - self - } - - fn record(&self, call: Call) { - self.calls.lock().unwrap().push(call); - } - - pub fn calls(&self) -> Vec { - self.calls.lock().unwrap().clone() - } - - pub fn call_count(&self) -> usize { - self.calls.lock().unwrap().len() - } - - /// The single recorded call, panicking when there is not exactly one. - pub fn only_call(&self) -> Call { - let calls = self.calls(); - assert_eq!( - calls.len(), - 1, - "expected exactly one driver call: {calls:?}" - ); - calls.into_iter().next().unwrap() - } -} - /// A [`GuardPolicy`](super::GuardPolicy) over an embedded driver with default /// budgets — the shipped configuration. pub fn embedded_policy() -> super::GuardPolicy { diff --git a/src/openhuman/memory/guard/test_support_part_03.rs b/src/openhuman/memory/guard/test_support_part_03.rs index 1f4304b907..43a6d867ad 100644 --- a/src/openhuman/memory/guard/test_support_part_03.rs +++ b/src/openhuman/memory/guard/test_support_part_03.rs @@ -1,3 +1,86 @@ +// `RecordingProvider`'s own constructor, seeders and call accessors. They +// live here rather than beside the struct in part 01 because that file is at +// the repo's 750-line ceiling for `src/openhuman` (scripts/ci/check-openhuman- +// rust-layout.mjs); the inherent impl is the one self-contained block that +// moves without splitting a trait implementation across files. + +impl RecordingProvider { + /// A provider with an empty call log and every seeded answer at its + /// default — the shape most tests want before layering a `with_*` on top. + pub fn new() -> Self { + Self { + calls: Mutex::new(Vec::new()), + recall_result: Mutex::new(Vec::new()), + fast_retrieve_result: Mutex::new(RetrievalResponse::default()), + namespace_hits: Mutex::new(Vec::new()), + namespace_summaries: Mutex::new(Vec::new()), + session_turns: Mutex::new(Vec::new()), + } + } + + /// Seed what `recall` answers, so a budget test can drive a known set. + pub fn with_recall_result(self, entries: Vec) -> Self { + *self.recall_result.lock().unwrap() = entries; + self + } + + /// Seed what `fast_retrieve` answers, so the auto-recall lane can run + /// through a real guard with known hits. + pub fn with_fast_retrieve_result(self, response: RetrievalResponse) -> Self { + *self.fast_retrieve_result.lock().unwrap() = response; + self + } + + /// Seed what `recall_namespace_scored` answers, so the vector-floored + /// paths can be driven with known scores. + pub fn with_namespace_hits(self, hits: Vec) -> Self { + *self.namespace_hits.lock().unwrap() = hits; + self + } + + /// Seed what `namespaces` answers, so a namespace can look populated + /// without a real store behind it. + pub fn with_namespace_summaries(self, summaries: Vec) -> Self { + *self.namespace_summaries.lock().unwrap() = summaries; + self + } + + /// Seed what `session_turns` answers. Without this the archivist's + /// finalize path stops at its empty-entries early return and never + /// reaches the recap it is being tested for. + pub fn with_session_turns(self, turns: Vec) -> Self { + *self.session_turns.lock().unwrap() = turns; + self + } + + /// Append one call to the log. Every family impl funnels through this. + fn record(&self, call: Call) { + self.calls.lock().unwrap().push(call); + } + + /// Every call the driver saw, in order. + pub fn calls(&self) -> Vec { + self.calls.lock().unwrap().clone() + } + + /// How many calls reached the driver — for tests that only care that a + /// path was or was not taken. + pub fn call_count(&self) -> usize { + self.calls.lock().unwrap().len() + } + + /// The single recorded call, panicking when there is not exactly one. + pub fn only_call(&self) -> Call { + let calls = self.calls(); + assert_eq!( + calls.len(), + 1, + "expected exactly one driver call: {calls:?}" + ); + calls.into_iter().next().unwrap() + } +} + // Fixtures for the retrieval family's scored answers. Included into // `test_support.rs` after the provider parts, so the imports there are in scope. From 9047c55514577ee43ab05ef015001650c4e74059 Mon Sep 17 00:00:00 2001 From: Shanu Date: Thu, 10 Sep 2026 15:49:13 +0530 Subject: [PATCH 3/8] fix(ci): drop the deleted golden-workspace targets from the coverage gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `domain_integration_targets` mapped every `src/openhuman/memory/**` change onto `memory_golden_fixture_e2e` and `memory_golden_parity_e2e`, and the `tests/fixtures/memory_golden/*` arm named the first of those directly. Both targets were deleted in cc99ba9c6 ("build(memory): cut the engine out of the test build") along with `tests/support/memory_golden.rs`, and neither was replaced. A mapping that names a target Cargo no longer has does not degrade to a weaker gate — it is a hard failure: error: no test target named `memory_golden_fixture_e2e` in `openhuman` so Rust Core Coverage has been red on every PR touching that domain since, after the lib tests have already passed. This branch hit it by moving `RecordingProvider`'s inherent impl between `memory/guard/test_support_*` files. Remove both entries rather than repointing them at a substitute: choosing a different suite to stand in for the schema gates is a coverage decision that belongs with whoever restores them. `src/openhuman/memory/**` now scopes to its `--lib` filter, and a golden-fixture change falls through to the full-suite arm. --- scripts/ci/rust-coverage-changed.sh | 22 ++++++++-------------- 1 file changed, 8 insertions(+), 14 deletions(-) diff --git a/scripts/ci/rust-coverage-changed.sh b/scripts/ci/rust-coverage-changed.sh index 9c0e8ea195..10f4a6b7c1 100755 --- a/scripts/ci/rust-coverage-changed.sh +++ b/scripts/ci/rust-coverage-changed.sh @@ -107,9 +107,14 @@ integration_test_targets() { # tested, and wrong for domains whose contract lives in an integration target: # such a gate never runs on a PR that touches only the domain's `src/`. # -# src/openhuman/memory/** → the golden-workspace schema gates. They stand -# between a memory-store schema change and a corrupted user workspace, and -# they are `tests/` targets, so `--lib` scoping alone skips them entirely. +# `src/openhuman/memory/**` used to sit here, naming the golden-workspace +# schema gates. Both of those targets — `memory_golden_fixture_e2e` and +# `memory_golden_parity_e2e` — were deleted in cc99ba9c6, which cut the +# engine out of the test build. A mapping that names a target Cargo no longer +# has is not a weaker gate: it is a hard `error: no test target named …` on +# every PR that touches the domain, so the entry is gone rather than pointed +# at a substitute. The domain scopes to its `--lib` filter alone until there +# is a live gate to name again. # # src/openhuman/agent/harness/session/** and src/openhuman/threads/goals/** # → `agent_turn_overrides_e2e`. Per-turn `TurnOverrides` (`session/types.rs`) @@ -124,9 +129,6 @@ integration_test_targets() { # empty result. domain_integration_targets() { case "$1" in - src/openhuman/memory/*) - printf '%s\n' memory_golden_fixture_e2e memory_golden_parity_e2e - ;; src/openhuman/agent/harness/session/* | src/openhuman/threads/goals/*) printf '%s\n' agent_turn_overrides_e2e ;; @@ -338,14 +340,6 @@ for f in "${files[@]}"; do log "${f} → integration gate '--test ${extra_target}'" done < <(domain_integration_targets "${f}") ;; - tests/fixtures/memory_golden/*) - # The golden memory-workspace fixture (committed .db blobs + the derived - # manifest). A change here IS the schema-gate re-baseline, so run the - # gates rather than falling through to the `*)` full-suite arm. - test_targets_raw="${test_targets_raw}memory_golden_fixture_e2e -" - log "${f} → integration gate '--test memory_golden_fixture_e2e'" - ;; tests/raw_coverage/*.rs) # The ~76 *_raw_coverage_e2e.rs suites are aggregated into the single # `raw_coverage_all` target (see tests/raw_coverage_all.rs + build.rs), so From 27b6f7078ccbc3a3c87d53660fcae4e8d31a8c76 Mon Sep 17 00:00:00 2001 From: Shanu Date: Thu, 10 Sep 2026 15:49:13 +0530 Subject: [PATCH 4/8] docs(memory): say plainly that the driver never substitutes a fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `summarise` forwarder's doc read "`summarise` documents a deterministic fallback as the expected response to a model that errors or times out", which parses as "the driver answers with a fallback". The contract says the opposite: the *caller* owns the deterministic fallback and the driver never substitutes one, "because a caller cannot tell a fallback summary from a model's own work once it is in the tree" (tinymemory-api/src/provider/content.rs). tinycortex's engine propagates the provider error for exactly that reason rather than calling its own `fallback_summary`. That distinction is load-bearing for the recap gate this PR adds — it is what makes a non-empty `Ok` proof the text came from the model — so state it directly instead of leaving it to be inferred. --- src/openhuman/modules/memory_part_02.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/openhuman/modules/memory_part_02.rs b/src/openhuman/modules/memory_part_02.rs index 877547f886..a38d288022 100644 --- a/src/openhuman/modules/memory_part_02.rs +++ b/src/openhuman/modules/memory_part_02.rs @@ -146,8 +146,12 @@ impl MemoryTree for ModuleMemoryProvider { /// so it is also the one whose bus deadline could bind. It rides the /// default: the module clamps the fold to the `token_budget` this caller /// supplied, and a summariser that outruns the deadline is the same failure - /// a caller must already handle — `summarise` documents a deterministic - /// fallback as the expected response to a model that errors or times out. + /// a caller must already handle — the contract puts the deterministic + /// fallback on the *caller* and states that the driver never substitutes + /// one, precisely so a fallback cannot be mistaken for a model's own work + /// once it is in the tree. An `Ok` here is therefore always the model's + /// text, or empty when there was nothing to fold; a model that errors, + /// times out or refuses arrives as `Err`, never as a filled-in summary. async fn summarise( &self, inputs: &[SummaryInput], From 4aad97005662ff7a173dd2c3a079078e7391701c Mon Sep 17 00:00:00 2001 From: Shanu Date: Thu, 10 Sep 2026 18:11:53 +0530 Subject: [PATCH 5/8] feat(archivist): recover segments a failed recap left unsummarised MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #6183 stopped the archivist persisting a heuristic bookend when the summariser fails, which leaves the segment `status='closed'` with a NULL summary. That is deliberately the `needs_resummary` marker. Nothing read it, so a transient outage still cost that stretch of history its summary — just without corrupting anything on the way. Two things were missing, both now in tinymemory v1.16.0: - the contract DTO carried only `open: bool`, collapsing `closed` and `summarised` onto `false`, so a host could not tell them apart; - no contract member listed segments at all — `open_segment` returns only the currently-open one. `ConversationSegment` now carries `status`, and `SegmentsPendingSummary` returns the pending queue. Re-pinned to v1.16.0 across all five sites. **The pass.** `resummarise_pending` re-runs `summarize_entries` — the same summariser the finalize path uses, not a second one; a differently-prompted summary sitting beside the originals would be indistinguishable from them and impossible to audit. It writes only on a usable recap, by the same `recap_is_usable` rule, so a segment whose recap fails again is left exactly as it was, still carrying the marker. **What triggers it.** Not a timer: the mid-session close path, right after a recap that succeeded. That success is the only first-hand evidence the app gets that the summariser is answering now. A scheduler would have to guess and would spend its budget against a provider that is still down; this cannot run at all while recaps are failing, which is what a backoff would be approximating. Deliberately NOT driven from `flush_open_segment`, the other caller of `on_segment_closed`: that one is awaited unbounded at session wind-down, and opportunistic recovery must not be charged to how long the app takes to close. **Bounded twice.** The driver orders oldest-first, so a segment nothing can ever summarise sits at the head forever. A batch cap alone would re-attempt that same head after every close and never reach the rest — the head-500 shape from #6051. So there is a per-process attempt ledger as well: a segment tried twice is skipped and the queue behind it drains. Per-process rather than persisted, because a restart usually means new config or a new build, which is the one thing likely to change the answer. Also classifies `segments_pending_summary` as a bounded read in `BOUNDED_READ_OPERATIONS` — it selects rows and writes none. Without it `every_operation_label_is_classified_and_no_mutation_is_a_read` fails. --- .github/workflows/ci-full.yml | 4 +- .github/workflows/ci-lite.yml | 4 +- .github/workflows/e2e-reusable.yml | 8 +- .../agent/harness/archivist/hook_impl.rs | 12 + .../harness/archivist/lifecycle_tests.rs | 4 + src/openhuman/agent/harness/archivist/mod.rs | 1 + .../agent/harness/archivist/recap_tests.rs | 4 + .../agent/harness/archivist/resummarise.rs | 190 +++++++++++++++ .../harness/archivist/resummarise_tests.rs | 216 ++++++++++++++++++ .../memory/guard/families_part_03.rs | 17 ++ .../memory/guard/test_support_part_01.rs | 8 +- .../memory/guard/test_support_part_02.rs | 11 + .../memory/guard/test_support_part_03.rs | 9 + src/openhuman/modules/memory_part_01.rs | 5 +- src/openhuman/modules/memory_part_03.rs | 14 ++ src/openhuman/modules/registry_part_01.rs | 48 ++-- vendor/tinymemory | 2 +- 17 files changed, 521 insertions(+), 36 deletions(-) create mode 100644 src/openhuman/agent/harness/archivist/resummarise.rs create mode 100644 src/openhuman/agent/harness/archivist/resummarise_tests.rs diff --git a/.github/workflows/ci-full.yml b/.github/workflows/ci-full.yml index 2a648f6926..772c5dc03f 100644 --- a/.github/workflows/ci-full.yml +++ b/.github/workflows/ci-full.yml @@ -132,8 +132,8 @@ jobs: # and leave a stale digest that only fails after the download. Both values # are the ubuntu-22.04-x86_64 row of `src/openhuman/modules/registry.rs`, # which is the authoritative pin — copy them from there, never recompute. - memory_version="1.15.3" - memory_sha256="5f50715abdc1b48745282db4718027159a292b5336d4bbd258cc471bb037b78e" + memory_version="1.16.0" + memory_sha256="f3ba06867ec89b8374a405f8a8569f5cecf88490609354ae4a6c1faa6e55b425" module_root="/opt/openhuman-test-modules/${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}" memory_dir="$module_root/tinymemory" juice_dir="$module_root/tinyjuice" diff --git a/.github/workflows/ci-lite.yml b/.github/workflows/ci-lite.yml index e15253c654..640ff34f7f 100644 --- a/.github/workflows/ci-lite.yml +++ b/.github/workflows/ci-lite.yml @@ -831,8 +831,8 @@ jobs: # and leave a stale digest that only fails after the download. Both values # are the ubuntu-22.04-x86_64 row of `src/openhuman/modules/registry.rs`, # which is the authoritative pin — copy them from there, never recompute. - memory_version="1.15.3" - memory_sha256="5f50715abdc1b48745282db4718027159a292b5336d4bbd258cc471bb037b78e" + memory_version="1.16.0" + memory_sha256="f3ba06867ec89b8374a405f8a8569f5cecf88490609354ae4a6c1faa6e55b425" module_root="/opt/openhuman-test-modules/${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}" memory_dir="$module_root/tinymemory" juice_dir="$module_root/tinyjuice" diff --git a/.github/workflows/e2e-reusable.yml b/.github/workflows/e2e-reusable.yml index 1b440b9513..70831b260f 100644 --- a/.github/workflows/e2e-reusable.yml +++ b/.github/workflows/e2e-reusable.yml @@ -168,8 +168,8 @@ jobs: # and leave a stale digest that only fails after the download. Both values # are the ubuntu-22.04-x86_64 row of `src/openhuman/modules/registry.rs`, # which is the authoritative pin — copy them from there, never recompute. - memory_version="1.15.3" - memory_sha256="5f50715abdc1b48745282db4718027159a292b5336d4bbd258cc471bb037b78e" + memory_version="1.16.0" + memory_sha256="f3ba06867ec89b8374a405f8a8569f5cecf88490609354ae4a6c1faa6e55b425" module_root="/opt/openhuman-test-modules/${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}" memory_dir="$module_root/tinymemory" memory_archive="$memory_dir/tinymemory-module-${memory_version}-ubuntu-22.04-x86_64.tar.gz" @@ -374,8 +374,8 @@ jobs: # and leave a stale digest that only fails after the download. Both values # are the ubuntu-22.04-x86_64 row of `src/openhuman/modules/registry.rs`, # which is the authoritative pin — copy them from there, never recompute. - memory_version="1.15.3" - memory_sha256="5f50715abdc1b48745282db4718027159a292b5336d4bbd258cc471bb037b78e" + memory_version="1.16.0" + memory_sha256="f3ba06867ec89b8374a405f8a8569f5cecf88490609354ae4a6c1faa6e55b425" module_root="/opt/openhuman-test-modules/${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}" memory_dir="$module_root/tinymemory" memory_archive="$memory_dir/tinymemory-module-${memory_version}-ubuntu-22.04-x86_64.tar.gz" diff --git a/src/openhuman/agent/harness/archivist/hook_impl.rs b/src/openhuman/agent/harness/archivist/hook_impl.rs index 6b3e0a01ea..4c7e1bf23c 100644 --- a/src/openhuman/agent/harness/archivist/hook_impl.rs +++ b/src/openhuman/agent/harness/archivist/hook_impl.rs @@ -159,6 +159,18 @@ impl PostTurnHook for ArchivistHook { if let Some(ref segment) = closed_segment { let now = Self::now_timestamp(); self.on_segment_closed(segment, session_id, now).await; + // Recover segments an earlier failed recap left unsummarised + // (#6186). Driven from here rather than from a timer because a + // close that just happened is first-hand evidence that the + // summariser is answering *now* — a scheduler would have to guess, + // and would spend its budget against a provider that is still down. + // + // Deliberately not called from `flush_open_segment`, the other + // caller of `on_segment_closed`: that one is awaited unbounded at + // session wind-down, and opportunistic recovery must never be + // charged to how long the app takes to close. This path is a + // detached post-turn hook, so the time is invisible. + self.resummarise_pending(now).await; } tracing::debug!("[archivist] turn indexed successfully: session={session_id}"); diff --git a/src/openhuman/agent/harness/archivist/lifecycle_tests.rs b/src/openhuman/agent/harness/archivist/lifecycle_tests.rs index 6604b56dbc..e9fc0e3477 100644 --- a/src/openhuman/agent/harness/archivist/lifecycle_tests.rs +++ b/src/openhuman/agent/harness/archivist/lifecycle_tests.rs @@ -10,6 +10,7 @@ //! model is built, resolved or called anywhere in these tests. use super::*; +use crate::openhuman::memory::api::provider::SegmentStatus; use crate::openhuman::memory::guard::test_support::RecordingProvider; const SESSION: &str = "lifecycle-6156"; @@ -48,6 +49,9 @@ fn segment() -> ConversationSegment { summary: None, embedding: None, open: false, + // #6186: the lifecycle marker the contract now carries. `Closed` + // is what a segment whose recap failed is left as. + status: Some(SegmentStatus::Closed), start_seq: None, end_seq: None, } diff --git a/src/openhuman/agent/harness/archivist/mod.rs b/src/openhuman/agent/harness/archivist/mod.rs index 4499cbd8a0..19b9c10924 100644 --- a/src/openhuman/agent/harness/archivist/mod.rs +++ b/src/openhuman/agent/harness/archivist/mod.rs @@ -22,6 +22,7 @@ pub(crate) mod helpers; mod hook_impl; mod lifecycle; mod recap; +mod resummarise; // The md-backed per-turn capture store the hook dual-writes into. It lives // here rather than behind the memory engine because nothing but this hook ever // called it — see the module's own docs for the round trip and for why the diff --git a/src/openhuman/agent/harness/archivist/recap_tests.rs b/src/openhuman/agent/harness/archivist/recap_tests.rs index 50551d1e25..6f842dfba8 100644 --- a/src/openhuman/agent/harness/archivist/recap_tests.rs +++ b/src/openhuman/agent/harness/archivist/recap_tests.rs @@ -1,4 +1,5 @@ use super::*; +use crate::openhuman::memory::api::provider::SegmentStatus; fn segment() -> ConversationSegment { ConversationSegment { @@ -13,6 +14,9 @@ fn segment() -> ConversationSegment { summary: None, embedding: None, open: false, + // #6186: the lifecycle marker the contract now carries. `Closed` + // is what a segment whose recap failed is left as. + status: Some(SegmentStatus::Closed), start_seq: Some(10), end_seq: Some(14), } diff --git a/src/openhuman/agent/harness/archivist/resummarise.rs b/src/openhuman/agent/harness/archivist/resummarise.rs new file mode 100644 index 0000000000..e3755be913 --- /dev/null +++ b/src/openhuman/agent/harness/archivist/resummarise.rs @@ -0,0 +1,190 @@ +//! Recovering segments a failed recap left unsummarised (#6186). +//! +//! #6156 stopped the archivist persisting a heuristic bookend when the +//! summariser fails, which leaves the segment `status='closed'` with a `NULL` +//! summary. That is deliberately the `needs_resummary` marker — it is what +//! `SegmentsPendingSummary` selects on — but a marker nothing reads is only a +//! tidier way to lose the summary. This is the pass that reads it. +//! +//! **What triggers it.** Not a timer. The pass runs immediately after a recap +//! that *succeeded*, because that success is the only first-hand evidence the +//! app ever gets that the summariser is answering right now. A scheduler would +//! have to guess, and would spend its budget hammering a provider that is still +//! down; this cannot run at all while recaps are failing, which is exactly the +//! behaviour a backoff would be approximating. +//! +//! **Why it is bounded twice.** `SegmentsPendingSummary` orders oldest-first, +//! so a segment nothing can ever summarise sits at the head of the queue +//! forever. Draining `limit` rows every time would re-attempt that same head +//! on every segment close and never reach the rest — the head-of-queue trap. +//! So there is a per-process attempt ledger as well as a batch cap: a segment +//! that has already been tried [`MAX_ATTEMPTS_PER_SEGMENT`] times in this +//! process is skipped, and the pass moves on to segments behind it. + +use std::collections::HashMap; +use std::sync::Mutex; + +use super::lifecycle::recap_is_usable; +use super::ArchivistHook; +use crate::openhuman::memory::api::provider::episodic::EpisodicTurn; + +/// Segments re-summarised in one pass. +/// +/// Small on purpose. The pass runs after *every* successful segment close, so +/// the queue drains across many passes rather than in one long stall; a large +/// batch would put a run of inference calls behind a single close, and +/// `flush_open_segment` is awaited at session wind-down. +const RESUMMARISE_BATCH: u32 = 3; + +/// How many times one segment may be attempted before this process gives up on +/// it. +/// +/// Bounds the head-of-queue trap described in the module docs. Deliberately +/// per-process rather than persisted: a restart usually means new config or a +/// new build, which is the one thing likely to change the answer, so a fresh +/// process earns each segment one more look. +const MAX_ATTEMPTS_PER_SEGMENT: u32 = 2; + +/// Attempts spent per segment id, for this process. +static ATTEMPTS: Mutex>> = Mutex::new(None); + +/// Record an attempt against `segment_id` and report whether it should be +/// skipped as already exhausted. +fn exhausted(segment_id: &str) -> bool { + let mut guard = match ATTEMPTS.lock() { + Ok(guard) => guard, + // A poisoned ledger means some other pass panicked mid-update. The + // ledger is an optimisation, not a correctness invariant, so the honest + // response is to let this segment through rather than to stop + // recovering summaries for the rest of the process's life. + Err(poisoned) => poisoned.into_inner(), + }; + let ledger = guard.get_or_insert_with(HashMap::new); + let seen = ledger.entry(segment_id.to_string()).or_insert(0); + if *seen >= MAX_ATTEMPTS_PER_SEGMENT { + return true; + } + *seen += 1; + false +} + +impl ArchivistHook { + /// Re-summarise up to [`RESUMMARISE_BATCH`] segments whose recap failed + /// earlier. + /// + /// Uses the same [`ArchivistHook::summarize_entries`] the finalize path + /// uses — not a second summariser. That is the whole reason the pass lives + /// here rather than inside the engine: a differently-prompted summary + /// sitting beside the originals would be indistinguishable from them and + /// impossible to audit later. + /// + /// Writes only on a usable recap, by the same [`recap_is_usable`] rule as + /// finalize. A segment whose recap fails again is left exactly as it was, + /// still carrying the marker. + /// + /// Never returns `Err`: every failure is logged and the pass moves to the + /// next segment. It is opportunistic work, and a failure here must not + /// affect the close that triggered it. + pub(super) async fn resummarise_pending(&self, now: f64) { + let Some(episodic) = self.episodic() else { + return; + }; + + let pending = match episodic.segments_pending_summary(RESUMMARISE_BATCH).await { + Ok(pending) => pending, + Err(e) => { + tracing::debug!("[archivist] resummarise: cannot read the pending queue: {e}"); + return; + } + }; + if pending.is_empty() { + return; + } + + tracing::debug!( + "[archivist] resummarise: {} segment(s) pending", + pending.len() + ); + + for segment in pending { + if exhausted(&segment.segment_id) { + tracing::debug!( + "[archivist] resummarise: segment={} already attempted {} times in this \ + process — skipping so the queue behind it can drain", + segment.segment_id, + MAX_ATTEMPTS_PER_SEGMENT + ); + continue; + } + + // Read per segment rather than once for the batch: turns are + // addressed by session, and two pending segments are usually from + // different sessions. + // + // `read_session_entries` + `is_in_segment` is the same pair the + // finalize path uses, and reusing it is not incidental — a turn + // belongs to a segment by stable per-session sequence or row id, + // not by timestamp, because the md store rounds to milliseconds and + // can sort a fast turn just before its own segment's start. + let entries = self.read_session_entries(&segment.session_id).await; + let segment_entries: Vec<&EpisodicTurn> = entries + .iter() + .filter(|record| record.is_in_segment(&segment)) + .map(|record| &record.turn) + .collect(); + if segment_entries.is_empty() { + // Nothing to fold. Not an error and not worth a warning: the + // segment is a real row whose turns have since been pruned, and + // there is no recap that could be produced for it. + tracing::debug!( + "[archivist] resummarise: segment={} has no turns left — nothing to fold", + segment.segment_id + ); + continue; + } + + let (summary, from_llm) = self + .summarize_entries(&segment_entries, &segment.segment_id, segment.turn_count) + .await; + if !recap_is_usable(from_llm, &summary) { + tracing::debug!( + "[archivist] resummarise: segment={} still has no LLM recap — left \ + unsummarised", + segment.segment_id + ); + continue; + } + + match episodic + .set_segment_summary(&segment.segment_id, &summary, now) + .await + { + Ok(()) => { + tracing::info!( + "[archivist] resummarise: recovered segment={} ({} turns, {} chars)", + segment.segment_id, + segment.turn_count, + summary.len() + ); + // Embedded here for the same reason finalize embeds: the + // summary is only in the index if something puts it there, + // and a recovered segment that is durable but unindexed is + // a second, quieter version of the same gap. + self.embed_segment_recap(&segment.segment_id, &summary, now) + .await; + } + Err(e) => { + tracing::warn!( + "[archivist] resummarise: failed to persist the recovered recap for \ + segment={}: {e}", + segment.segment_id + ); + } + } + } + } +} + +#[cfg(test)] +#[path = "resummarise_tests.rs"] +mod tests; diff --git a/src/openhuman/agent/harness/archivist/resummarise_tests.rs b/src/openhuman/agent/harness/archivist/resummarise_tests.rs new file mode 100644 index 0000000000..c3fe803086 --- /dev/null +++ b/src/openhuman/agent/harness/archivist/resummarise_tests.rs @@ -0,0 +1,216 @@ +//! Recovery-pass tests (#6186). +//! +//! Driven through the **contract** like the finalize-path tests beside them: a +//! `RecordingProvider` serves every family and records what reached it, so +//! "nothing was written" is asserted against the driver's call log rather than +//! inferred from an empty store. +//! +//! `ArchivistHook::new` leaves `summariser_available == false`, so +//! `summarize_entries` takes its heuristic arm deterministically — no chat +//! model is built, resolved or called anywhere here. That makes the *write* +//! assertions negative by construction, which is the right shape: the pass must +//! be provably incapable of persisting a recap it did not get from a model, +//! and that is the same rule finalize is held to. + +use std::sync::Arc; + +use super::*; +use crate::openhuman::memory::api::provider::{ConversationSegment, MemoryProvider, SegmentStatus}; +use crate::openhuman::memory::guard::test_support::RecordingProvider; + +const SESSION: &str = "resummarise-6186"; + +fn turn(id: i64, role: &str, content: &str) -> EpisodicTurn { + EpisodicTurn { + id: Some(id), + session_id: SESSION.into(), + timestamp: 100.0 + id as f64, + role: role.into(), + content: content.into(), + lesson: None, + tool_calls_json: None, + cost_microdollars: 0, + } +} + +/// A pending segment. `segment_id` is a parameter because the attempt ledger +/// that bounds the head-of-queue trap is **per process**, so two tests sharing +/// an id would share its budget and run order would decide which one passed. +fn pending(segment_id: &str) -> ConversationSegment { + ConversationSegment { + segment_id: segment_id.into(), + session_id: SESSION.into(), + namespace: "global".into(), + start_episodic_id: 1, + end_episodic_id: Some(2), + start_timestamp: 100.0, + end_timestamp: Some(103.0), + turn_count: 2, + summary: None, + embedding: None, + open: false, + status: Some(SegmentStatus::Closed), + start_seq: None, + end_seq: None, + } +} + +fn turns() -> Vec { + vec![ + turn(1, "user", "How do I pin a submodule?"), + turn(2, "assistant", "Record the gitlink at the commit you want."), + ] +} + +fn methods(recording: &RecordingProvider) -> Vec { + recording.calls().into_iter().map(|c| c.method).collect() +} + +fn hook_over(recording: &Arc) -> ArchivistHook { + let provider: Arc = recording.clone(); + ArchivistHook::new(provider, true) +} + +/// The pass reads the queue and the pending segment's turns — it does not stop +/// at the query. +/// +/// The positive half of the assertion is the point. Without it this would also +/// pass if `resummarise_pending` had returned straight after +/// `segments_pending_summary`, which is the failure mode a purely negative test +/// cannot see. +#[tokio::test] +async fn a_pending_segment_is_read_and_recapped() { + let recording = Arc::new( + RecordingProvider::new() + .with_session_turns(turns()) + .with_pending_segments(vec![pending("seg-read")]), + ); + let hook = hook_over(&recording); + + hook.resummarise_pending(300.0).await; + + let methods = methods(&recording); + assert!( + methods + .iter() + .any(|m| m == "episodic.segments_pending_summary"), + "the pending queue was never read: {methods:?}" + ); + assert!( + methods.iter().any(|m| m == "episodic.session_turns"), + "the segment's turns were never read: {methods:?}" + ); +} + +/// A recap that fails again writes nothing, and leaves the marker in place. +/// +/// Same rule as finalize (#6156): the heuristic bookend is not a summary, and +/// persisting it would flip the row to `summarised` and remove it from the very +/// queue this pass selects on — turning a recoverable segment into a +/// permanently degraded one. A recovery pass that did that would be worse than +/// no recovery pass. +#[tokio::test] +async fn a_still_failing_recap_is_not_persisted_or_embedded() { + let recording = Arc::new( + RecordingProvider::new() + .with_session_turns(turns()) + .with_pending_segments(vec![pending("seg-still-failing")]), + ); + let hook = hook_over(&recording); + + hook.resummarise_pending(300.0).await; + + let methods = methods(&recording); + for forbidden in [ + "episodic.set_segment_summary", + "episodic.upsert_segment_embedding", + "scoring.embed_text", + "scoring.embedder_slug", + ] { + assert!( + !methods.iter().any(|m| m == forbidden), + "`{forbidden}` ran on a heuristic recap: {methods:?}" + ); + } +} + +/// An empty queue costs one call and stops. +/// +/// Pins the early return: the pass runs after every mid-session segment close, +/// so the healthy steady state must not read turns or build a corpus for +/// segments that do not exist. +#[tokio::test] +async fn an_empty_queue_reads_nothing_else() { + let recording = Arc::new(RecordingProvider::new().with_session_turns(turns())); + let hook = hook_over(&recording); + + hook.resummarise_pending(300.0).await; + + let methods = methods(&recording); + assert_eq!( + methods, + vec!["episodic.segments_pending_summary".to_string()], + "an empty queue did more than query: {methods:?}" + ); +} + +/// A segment that keeps failing is dropped from the pass, so the queue behind +/// it can drain. +/// +/// This is the head-of-queue trap, and it is not hypothetical: the driver +/// orders oldest-first, and a segment nothing can ever summarise stays at the +/// head. Without the ledger the pass would re-attempt that same segment after +/// every close and never reach the ones behind it — the same shape as the +/// head-500 loop in #6051. +#[tokio::test] +async fn a_repeatedly_failing_segment_stops_being_attempted() { + let recording = Arc::new( + RecordingProvider::new() + .with_session_turns(turns()) + .with_pending_segments(vec![pending("seg-exhausts")]), + ); + let hook = hook_over(&recording); + + // Two attempts are the budget; the third pass must skip it. + hook.resummarise_pending(300.0).await; + hook.resummarise_pending(301.0).await; + let before = methods(&recording) + .iter() + .filter(|m| *m == "episodic.session_turns") + .count(); + + hook.resummarise_pending(302.0).await; + + let after = methods(&recording) + .iter() + .filter(|m| *m == "episodic.session_turns") + .count(); + assert_eq!( + before, 2, + "the first two passes should each have attempted the segment" + ); + assert_eq!( + after, before, + "the third pass attempted an exhausted segment instead of skipping it" + ); +} + +/// A segment whose turns are gone is skipped without a write. +/// +/// The row can outlive its turns, and the honest answer is to leave it alone: +/// there is no recap that could be produced, and writing an empty summary would +/// seal it as if there were. +#[tokio::test] +async fn a_segment_with_no_turns_left_is_skipped() { + let recording = + Arc::new(RecordingProvider::new().with_pending_segments(vec![pending("seg-no-turns")])); + let hook = hook_over(&recording); + + hook.resummarise_pending(300.0).await; + + let methods = methods(&recording); + assert!( + !methods.iter().any(|m| m == "episodic.set_segment_summary"), + "a segment with no turns was summarised anyway: {methods:?}" + ); +} diff --git a/src/openhuman/memory/guard/families_part_03.rs b/src/openhuman/memory/guard/families_part_03.rs index b3dee6411d..0c2bc19279 100644 --- a/src/openhuman/memory/guard/families_part_03.rs +++ b/src/openhuman/memory/guard/families_part_03.rs @@ -298,6 +298,23 @@ impl MemoryEpisodic for GuardedEpisodic { .await } + /// A read: it selects rows, it changes none. Admitted as one so a + /// read-only policy can still drive the re-summarisation pass (#6186) — + /// the write it leads to is `set_segment_summary`, which is admitted + /// separately on its own terms. + async fn segments_pending_summary( + &self, + limit: u32, + ) -> Result, MemoryError> { + self.policy.admit_read( + Capability::Episodic, + "episodic.segments_pending_summary", + NO_NAMESPACE, + false, + )?; + self.family()?.segments_pending_summary(limit).await + } + async fn upsert_segment_embedding( &self, segment_id: &str, diff --git a/src/openhuman/memory/guard/test_support_part_01.rs b/src/openhuman/memory/guard/test_support_part_01.rs index 4b0f4fc355..26b567a8f2 100644 --- a/src/openhuman/memory/guard/test_support_part_01.rs +++ b/src/openhuman/memory/guard/test_support_part_01.rs @@ -21,8 +21,9 @@ use crate::openhuman::memory::api::provider::types::{ MaintenanceReport, SnapshotRef, SourceItem, SourceScope, }; use crate::openhuman::memory::api::provider::{ - AddressBookSeedOutcome, ChunkDetail, ChunkEmbedding, ChunkQuery, CoverWindowQuery, EntityMatch, - EpisodicEvent, EpisodicTurn, FacetType, FastRetrieveQuery, MemoryChunks, MemoryCodingSessions, + AddressBookSeedOutcome, ChunkDetail, ChunkEmbedding, ChunkQuery, ConversationSegment, + CoverWindowQuery, EntityMatch, EpisodicEvent, EpisodicTurn, FacetType, FastRetrieveQuery, + MemoryChunks, MemoryCodingSessions, MemoryCore, MemoryDiff, MemoryDocuments, MemoryEntities, MemoryEpisodic, MemoryGoals, MemoryGraph, MemoryIngest, MemoryMaintenance, MemoryPeople, MemoryPortability, MemoryProfile, @@ -90,6 +91,9 @@ pub struct RecordingProvider { /// What `session_turns` returns, so the archivist's finalize path can be /// driven past its empty-entries early return without an engine behind it. session_turns: Mutex>, + /// What `segments_pending_summary` returns, so the re-summarisation pass + /// (#6186) can be driven over a known queue. + pending_segments: Mutex>, } impl Default for RecordingProvider { diff --git a/src/openhuman/memory/guard/test_support_part_02.rs b/src/openhuman/memory/guard/test_support_part_02.rs index 08240b2506..8d67770a18 100644 --- a/src/openhuman/memory/guard/test_support_part_02.rs +++ b/src/openhuman/memory/guard/test_support_part_02.rs @@ -261,6 +261,17 @@ impl MemoryEpisodic for RecordingProvider { Ok(None) } + async fn segments_pending_summary( + &self, + _limit: u32, + ) -> Result< + Vec, + MemoryError, + > { + self.record(Call::plain("episodic.segments_pending_summary")); + Ok(self.pending_segments.lock().unwrap().clone()) + } + async fn create_segment( &self, _segment_id: &str, diff --git a/src/openhuman/memory/guard/test_support_part_03.rs b/src/openhuman/memory/guard/test_support_part_03.rs index 43a6d867ad..3c14e179c0 100644 --- a/src/openhuman/memory/guard/test_support_part_03.rs +++ b/src/openhuman/memory/guard/test_support_part_03.rs @@ -15,6 +15,7 @@ impl RecordingProvider { namespace_hits: Mutex::new(Vec::new()), namespace_summaries: Mutex::new(Vec::new()), session_turns: Mutex::new(Vec::new()), + pending_segments: Mutex::new(Vec::new()), } } @@ -53,6 +54,14 @@ impl RecordingProvider { self } + /// Seed the re-summarisation queue (#6186). The default is empty, so a + /// test that does not set this drives the pass over nothing — which is + /// the state a healthy store is in. + pub fn with_pending_segments(self, segments: Vec) -> Self { + *self.pending_segments.lock().unwrap() = segments; + self + } + /// Append one call to the log. Every family impl funnels through this. fn record(&self, call: Call) { self.calls.lock().unwrap().push(call); diff --git a/src/openhuman/modules/memory_part_01.rs b/src/openhuman/modules/memory_part_01.rs index b923388a81..dce546c206 100644 --- a/src/openhuman/modules/memory_part_01.rs +++ b/src/openhuman/modules/memory_part_01.rs @@ -8,7 +8,7 @@ use tinymemory_api::capabilities::{Capabilities, Capability}; /// Checked against the registry pin by `the_capability_list_matches_the_pinned_release`, /// so bumping the pin without re-reading the list is a red test rather than a /// silent over-claim. -pub(crate) const ARTIFACT_CAPABILITIES_PIN: &str = "1.15.3"; +pub(crate) const ARTIFACT_CAPABILITIES_PIN: &str = "1.16.0"; /// The capability families the **pinned artifact** actually serves. /// @@ -362,6 +362,9 @@ const BOUNDED_READ_OPERATIONS: &[&str] = &[ "runtime_tree_status", "score_person", "search_entities", + // #6186. Selects closed segments with no summary; it writes nothing. + // The write it leads to is `set_segment_summary`, classified separately. + "segments_pending_summary", "session_turns", "snapshots", "source_ingest_status", diff --git a/src/openhuman/modules/memory_part_03.rs b/src/openhuman/modules/memory_part_03.rs index 593c07cf9d..421ad4b6ee 100644 --- a/src/openhuman/modules/memory_part_03.rs +++ b/src/openhuman/modules/memory_part_03.rs @@ -265,6 +265,20 @@ impl MemoryEpisodic for ModuleMemoryProvider { (segment_id, summary, now) ) } + /// Forwarded rather than left to the trait default (#6186). The default + /// answers an empty list, which here would read as "no segment needs + /// re-summarising" — indistinguishable from a healthy store, and silent. + async fn segments_pending_summary( + &self, + limit: u32, + ) -> Result, MemoryError> { + module_call!( + self, + "segments_pending_summary", + methods::SEGMENTS_PENDING_SUMMARY, + (limit,) + ) + } async fn upsert_segment_embedding( &self, segment_id: &str, diff --git a/src/openhuman/modules/registry_part_01.rs b/src/openhuman/modules/registry_part_01.rs index ee9eaebd6e..7f9c26b4e6 100644 --- a/src/openhuman/modules/registry_part_01.rs +++ b/src/openhuman/modules/registry_part_01.rs @@ -175,63 +175,63 @@ const TINYMEMORY: ModuleRecord = ModuleRecord { description: "Local memory engine: store, ranked recall, and portable export", bus_name: "ai.tinyhumans.tinymemory.Memory", object_path: "/ai/tinyhumans/tinymemory/Memory", - version: "1.15.3", - release_url: "https://github.com/tinyhumansai/tinymemory/releases/tag/v1.15.3", + version: "1.16.0", + release_url: "https://github.com/tinyhumansai/tinymemory/releases/tag/v1.16.0", assets: &[ PlatformAsset { host_key: "ubuntu-24.04-x86_64", - archive: "tinymemory-module-1.15.3-ubuntu-24.04-x86_64.tar.gz", - sha256: "cdf1bc2f1deb32f7d52d5c0caa235ee28be458ce10a1dc69f2c0e611bed15b8d", + archive: "tinymemory-module-1.16.0-ubuntu-24.04-x86_64.tar.gz", + sha256: "fc65fce075b0d286b0d1cce48fc8952c2936e4b945f2af84b6a20d800c2a10d7", }, PlatformAsset { host_key: "ubuntu-24.04-arm64", - archive: "tinymemory-module-1.15.3-ubuntu-24.04-arm64.tar.gz", - sha256: "cb4e135a4be90953f277fba311e056d46d44d15db5fb027cbe83bee6247f0995", + archive: "tinymemory-module-1.16.0-ubuntu-24.04-arm64.tar.gz", + sha256: "1735efb7b0b6a56c85da1b62fbfa2d2a68995a04203ec2d79b4e1b389935fd92", }, PlatformAsset { host_key: "ubuntu-22.04-x86_64", - archive: "tinymemory-module-1.15.3-ubuntu-22.04-x86_64.tar.gz", - sha256: "5f50715abdc1b48745282db4718027159a292b5336d4bbd258cc471bb037b78e", + archive: "tinymemory-module-1.16.0-ubuntu-22.04-x86_64.tar.gz", + sha256: "f3ba06867ec89b8374a405f8a8569f5cecf88490609354ae4a6c1faa6e55b425", }, PlatformAsset { host_key: "ubuntu-22.04-arm64", - archive: "tinymemory-module-1.15.3-ubuntu-22.04-arm64.tar.gz", - sha256: "1992869bdea71ca7a96d5a7e0e32e8abc659a8bbb16d0cf34b2ba4440ca6892d", + archive: "tinymemory-module-1.16.0-ubuntu-22.04-arm64.tar.gz", + sha256: "b9f6794806e9463cffbfc3ce0c4b4b39cb8c1f3403a2ba956b0984cc35e9354a", }, PlatformAsset { host_key: "macos-26-arm64", - archive: "tinymemory-module-1.15.3-macos-26-arm64.tar.gz", - sha256: "63c64b35aa9364f2fbc7d83eb8dbab2895352441223f30a4e6226aa479c7563d", + archive: "tinymemory-module-1.16.0-macos-26-arm64.tar.gz", + sha256: "097ab5fdc352f84f34b770302db73c5c1468e7ad48709e1767e74d61cc55646c", }, PlatformAsset { host_key: "macos-26-x86_64", - archive: "tinymemory-module-1.15.3-macos-26-x86_64.tar.gz", - sha256: "61cc6b44627d11344957ff47a5047f2e25068d1d9fa158a08dbc37fbc78bc1ef", + archive: "tinymemory-module-1.16.0-macos-26-x86_64.tar.gz", + sha256: "3128eaff5e820c86fabfbe6a759976150313ca3c2c38e00bb2bb67dd2dfa76ba", }, PlatformAsset { host_key: "macos-15-arm64", - archive: "tinymemory-module-1.15.3-macos-15-arm64.tar.gz", - sha256: "4feed28d2bd4b8025ce975d3518014c85ff002bce4ca5ab7c9cfd923dc5f7686", + archive: "tinymemory-module-1.16.0-macos-15-arm64.tar.gz", + sha256: "fa9d65f31b7ece3eed0d118795ce06b66f0bd271556b91ccad09b0c4b41d94f9", }, PlatformAsset { host_key: "macos-15-x86_64", - archive: "tinymemory-module-1.15.3-macos-15-x86_64.tar.gz", - sha256: "bf567c16b0d26c8b00b6d5e4b44e37bfeec820dd5dcb7e941505859a87a3fa8c", + archive: "tinymemory-module-1.16.0-macos-15-x86_64.tar.gz", + sha256: "ac1343f128cdd4b299b43318019d1b26c0c8fc28abf62bf24ff8fb6d9894730f", }, PlatformAsset { host_key: "windows-2025-x86_64", - archive: "tinymemory-module-1.15.3-windows-2025-x86_64.zip", - sha256: "e741da2d02ee3b2bd99b7bbada298b92bfa3b233d903e05290a4a25cc3b68d62", + archive: "tinymemory-module-1.16.0-windows-2025-x86_64.zip", + sha256: "dcf5ec88583a02b284f0b430a037230a3d974703d39f11be53a4e36688c8d7db", }, PlatformAsset { host_key: "windows-2022-x86_64", - archive: "tinymemory-module-1.15.3-windows-2022-x86_64.zip", - sha256: "ad93120dcfeab2dcbf2d90bdadab220a2de46e087461e39d2203d483d868fb85", + archive: "tinymemory-module-1.16.0-windows-2022-x86_64.zip", + sha256: "cf56453fba522a075e569d60c39d7e56e938ff2d4fbb8270debaffd7e6b39819", }, PlatformAsset { host_key: "windows-11-arm64", - archive: "tinymemory-module-1.15.3-windows-11-arm64.zip", - sha256: "0242470cde6e802e93eedeb55ed8722da9483740fee8c0e5e60555a67d7b6da3", + archive: "tinymemory-module-1.16.0-windows-11-arm64.zip", + sha256: "9d0ca43cc3c7cac29e631e9870fc30d3fd4971e272e393ed2a91adb7ca20fdf3", }, ], // Eager, unlike the two codecs above. A codec that is never asked for should diff --git a/vendor/tinymemory b/vendor/tinymemory index e8b6f36739..cc09032395 160000 --- a/vendor/tinymemory +++ b/vendor/tinymemory @@ -1 +1 @@ -Subproject commit e8b6f36739a55698582a130ca0cdf917701710a8 +Subproject commit cc0903239563b7448bbacc759867524507545c9d From 094dd829018c70ff3ceff51b60ccd4d3c1623f7c Mon Sep 17 00:00:00 2001 From: Shanu Date: Thu, 10 Sep 2026 18:30:31 +0530 Subject: [PATCH 6/8] fix(archivist): only recover segments when the summariser just answered MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `resummarise_pending` was called after every mid-session segment close, regardless of whether that close produced a recap. The comment beside it claimed the opposite — "a close that just happened is first-hand evidence that the summariser is answering" — but a close happens whether the recap succeeded or failed, so the gate the comment described did not exist. That inverts what the attempt ledger is for. During an outage: a segment closes, its recap fails, the segment is flagged, and the pass then runs anyway and retries the flagged segments against the same unreachable provider. Two closes during one outage exhaust every pending segment's two-attempt budget, and each is skipped for the rest of the process — including after the provider comes back. The outage consumes the recovery it is supposed to trigger. `on_segment_closed` now returns whether a usable LLM recap was produced, and the caller gates on it. The value is a liveness signal rather than a success code: `false` covers no driver, no entries, no summariser and a summariser that failed alike, because none of them is a moment to spend budget against that provider. `heuristic_recap_is_not_persisted_or_embedded` asserts the `false`, which is the regression that matters — the `true` arm needs a live summariser and the contract-level harness deliberately has none. --- .../agent/harness/archivist/hook_impl.rs | 19 ++++++++++++++----- .../agent/harness/archivist/lifecycle.rs | 16 +++++++++++++--- .../harness/archivist/lifecycle_tests.rs | 11 ++++++++++- 3 files changed, 37 insertions(+), 9 deletions(-) diff --git a/src/openhuman/agent/harness/archivist/hook_impl.rs b/src/openhuman/agent/harness/archivist/hook_impl.rs index 4c7e1bf23c..10b628c0a0 100644 --- a/src/openhuman/agent/harness/archivist/hook_impl.rs +++ b/src/openhuman/agent/harness/archivist/hook_impl.rs @@ -158,19 +158,28 @@ impl PostTurnHook for ArchivistHook { // moves the tree write to segment granularity inside on_segment_closed. if let Some(ref segment) = closed_segment { let now = Self::now_timestamp(); - self.on_segment_closed(segment, session_id, now).await; + let recap_succeeded = self.on_segment_closed(segment, session_id, now).await; // Recover segments an earlier failed recap left unsummarised // (#6186). Driven from here rather than from a timer because a - // close that just happened is first-hand evidence that the - // summariser is answering *now* — a scheduler would have to guess, - // and would spend its budget against a provider that is still down. + // recap that just succeeded is first-hand evidence that the + // summariser is answering *now* — a scheduler would have to guess. + // + // The gate is load-bearing, not a nicety. `resummarise_pending` + // spends a per-segment attempt budget, and running it while the + // provider is still down would burn that budget on calls that + // cannot succeed: two segment closes during one outage would + // exhaust every pending segment's retries and skip them for the + // rest of the process — including after the provider came back. + // The outage would consume the recovery it is supposed to trigger. // // Deliberately not called from `flush_open_segment`, the other // caller of `on_segment_closed`: that one is awaited unbounded at // session wind-down, and opportunistic recovery must never be // charged to how long the app takes to close. This path is a // detached post-turn hook, so the time is invisible. - self.resummarise_pending(now).await; + if recap_succeeded { + self.resummarise_pending(now).await; + } } tracing::debug!("[archivist] turn indexed successfully: session={session_id}"); diff --git a/src/openhuman/agent/harness/archivist/lifecycle.rs b/src/openhuman/agent/harness/archivist/lifecycle.rs index aceb0dafda..600b293ee8 100644 --- a/src/openhuman/agent/harness/archivist/lifecycle.rs +++ b/src/openhuman/agent/harness/archivist/lifecycle.rs @@ -331,12 +331,20 @@ impl ArchivistHook { /// /// Soft-fallback contract (mirrors `LlmSummariser`): this function /// never returns `Err`; all failures are logged and ignored. + /// Returns whether a usable LLM recap was produced for this segment. + /// + /// The caller uses it as a liveness signal, not as a success code: a `true` + /// means the summariser answered *just now*, which is the only first-hand + /// evidence the app gets that it is reachable. `false` covers every reason + /// a recap did not happen — no driver, no entries, no summariser, or a + /// summariser that failed — because none of them are a moment to spend + /// budget re-trying older segments against the same provider (#6186). pub(super) async fn on_segment_closed( &self, segment: &ConversationSegment, session_id: &str, now: f64, - ) { + ) -> bool { // Gather the conversation text for this segment. Prefer the // md-backed memory_archivist read when config is available; fall // back to the driver's episodic family otherwise. @@ -356,7 +364,7 @@ impl ArchivistHook { "[archivist] segment={} has no entries — skipping recap", segment.segment_id ); - return; + return false; } // Build segment text from user messages (for event extraction). @@ -376,7 +384,7 @@ impl ArchivistHook { // driver, not the summary: with no episodic family there is nothing to // write on either arm, and this is the exit that path has always taken. let Some(episodic) = self.episodic() else { - return; + return false; }; if recap_is_usable(from_llm, &summary) { @@ -557,6 +565,8 @@ impl ArchivistHook { ); } } + + recap_is_usable(from_llm, &summary) } /// Embed `summary` for `segment_id` and write the per-model embedding row. diff --git a/src/openhuman/agent/harness/archivist/lifecycle_tests.rs b/src/openhuman/agent/harness/archivist/lifecycle_tests.rs index e9fc0e3477..858aa2194c 100644 --- a/src/openhuman/agent/harness/archivist/lifecycle_tests.rs +++ b/src/openhuman/agent/harness/archivist/lifecycle_tests.rs @@ -77,7 +77,16 @@ async fn heuristic_recap_is_not_persisted_or_embedded() { let provider: Arc = recording.clone(); let hook = ArchivistHook::new(provider, true); - hook.on_segment_closed(&segment(), SESSION, 200.0).await; + let recap_succeeded = hook.on_segment_closed(&segment(), SESSION, 200.0).await; + + // #6186: the caller gates the re-summarisation pass on this. Reporting + // `true` here would make an outage spend every pending segment's retry + // budget against the provider that is still down, and skip them for the + // rest of the process — including after it recovers. + assert!( + !recap_succeeded, + "a heuristic recap must not report the summariser as answering" + ); let methods = methods(&recording); assert!( From 708404e29120fbffb80fef7fabf07e0a1137a492 Mon Sep 17 00:00:00 2001 From: Shanu Date: Thu, 10 Sep 2026 18:44:18 +0530 Subject: [PATCH 7/8] fix(prompt): key the memory-access rule on the delegate too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `MEMORY_READ_TOOLS` listed only `memory_recall` and `memory_search`, so `MemoryAccessSection` was dropped for an agent whose memory arrives through `retrieve_memory` — the delegate synthesised from the memory sub-agent's `delegate_name`, and how the orchestrator is actually configured. That agent got the tool and no rule about using it, which is the exact shape of the bug #6040 and #6048 were filed for: claiming absence without a retrieval. `any_tool_offered` already takes the delegation tools as a separate argument, so this needed no new plumbing — the list was simply the wrong list. This is also what `memory_access_instruction_is_present_with_learning_disabled` had been failing on. It passed alone and failed in the full run, which read as flakiness; it was not. The orchestrator holds both the direct tool and the delegate, and whichever of the two a given run ended up with decided whether the assertion held. Keying on both makes the outcome the same either way. The write-side twin is deliberately untouched. `MemoryWriteSection` gates on `SAVE_PREFERENCE_TOOL` and `MEMORY_STORE_TOOL` individually rather than on `MEMORY_WRITE_TOOLS`, and it names the specific tools it found, so admitting `manage_profile_memory` there is a question about which claim the section makes — not the same one-line change. Worth its own look. Also moves `impl MemoryAnswer for RecordingProvider` from `test_support_part_02` to `part_03`: the episodic family growing `segments_pending_summary` pushed part 02 to 755 lines against the repo's 750 ceiling, which is what `check-openhuman-rust-layout.mjs` was failing on. Same reason, and the same choice of a self-contained tail block, as the inherent impl that already lives in part 03. --- .../agent/learning/prompt_sections.rs | 14 +++++++++- .../memory/guard/test_support_part_02.rs | 22 ---------------- .../memory/guard/test_support_part_03.rs | 26 +++++++++++++++++++ 3 files changed, 39 insertions(+), 23 deletions(-) diff --git a/src/openhuman/agent/learning/prompt_sections.rs b/src/openhuman/agent/learning/prompt_sections.rs index b559824761..9c6aa17d44 100644 --- a/src/openhuman/agent/learning/prompt_sections.rs +++ b/src/openhuman/agent/learning/prompt_sections.rs @@ -200,7 +200,19 @@ impl PromptSection for MemoryWriteSection { } /// The retrieval tools [`MemoryAccessSection`] is keyed on. -pub const MEMORY_READ_TOOLS: [&str; 2] = ["memory_recall", "memory_search"]; +/// +/// `retrieve_memory` is the **delegate** to the memory sub-agent, synthesised +/// from `delegate_name` in `memory/agent/agent/agent.toml`, and it belongs here +/// for the same reason the two direct tools do: the section is a rule about +/// what the model must do before claiming something is not stored, and an agent +/// holding the delegate can retrieve. Keying only on the direct names meant an +/// orchestrator whose memory arrives by delegation — which is how the +/// orchestrator is configured — got the tool and no rule about using it. That +/// is the shape of the bug #6040 and #6048 were both filed for. +/// +/// [`any_tool_offered`] already receives the delegation tools separately, so +/// this needed no new plumbing; the list was simply the wrong list. +pub const MEMORY_READ_TOOLS: [&str; 3] = ["memory_recall", "memory_search", "retrieve_memory"]; /// The tool a preference is written through. pub const SAVE_PREFERENCE_TOOL: &str = "save_preference"; diff --git a/src/openhuman/memory/guard/test_support_part_02.rs b/src/openhuman/memory/guard/test_support_part_02.rs index 8d67770a18..15b902f797 100644 --- a/src/openhuman/memory/guard/test_support_part_02.rs +++ b/src/openhuman/memory/guard/test_support_part_02.rs @@ -731,25 +731,3 @@ impl MemoryEventIngest for RecordingProvider { Ok(IngestOutcome::default()) } } - -#[async_trait] -impl MemoryAnswer for RecordingProvider { - async fn answer( - &self, - _request: crate::openhuman::memory::api::provider::operations::AnswerRequest, - ) -> Result - { - self.record(Call { - method: "answer.answer".into(), - content: None, - taint: None, - scoped: None, - }); - Ok(crate::openhuman::memory::api::provider::operations::AnswerResponse { - answer: String::new(), - model: None, - citations: Vec::new(), - steps: Vec::new(), - }) - } -} diff --git a/src/openhuman/memory/guard/test_support_part_03.rs b/src/openhuman/memory/guard/test_support_part_03.rs index 3c14e179c0..c6ca15b23f 100644 --- a/src/openhuman/memory/guard/test_support_part_03.rs +++ b/src/openhuman/memory/guard/test_support_part_03.rs @@ -131,3 +131,29 @@ pub fn namespace_hit( taint: MemoryTaint::default(), } } + +// Moved from part 02 for the same reason the inherent impl above moved: that +// file reached the 750-line ceiling when the episodic family grew +// `segments_pending_summary` (#6186). `MemoryAnswer` is the tail block and is +// self-contained, so it relocates without splitting a trait impl across files. +#[async_trait] +impl MemoryAnswer for RecordingProvider { + async fn answer( + &self, + _request: crate::openhuman::memory::api::provider::operations::AnswerRequest, + ) -> Result + { + self.record(Call { + method: "answer.answer".into(), + content: None, + taint: None, + scoped: None, + }); + Ok(crate::openhuman::memory::api::provider::operations::AnswerResponse { + answer: String::new(), + model: None, + citations: Vec::new(), + steps: Vec::new(), + }) + } +} From d15a3a5c09c00d1b11c15a2264a7e8f2ce9f1304 Mon Sep 17 00:00:00 2001 From: Shanu Date: Thu, 10 Sep 2026 18:50:59 +0530 Subject: [PATCH 8/8] fix(archivist): scan past an exhausted queue head MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The attempt ledger stopped a stuck segment being *re-attempted*, but not being *in the way*. `segments_pending_summary` orders oldest-first and takes a limit, and the pass asked for exactly `RESUMMARISE_BATCH`, so the three oldest rows were returned before the ledger filtered them. Once those three exhausted their attempts they occupied every later result and a fourth pending segment could never be recovered until the process restarted — the head-of-queue trap the ledger was added to prevent, moved one layer out rather than solved. Raised by CodeRabbit on #6183. The pass now reads a bounded window (`RESUMMARISE_SCAN`) and counts only **eligible** segments against the batch, so a skipped exhausted segment does not consume it. Bounded rather than unbounded on purpose: the point is to get past a stuck head, not to walk an arbitrarily long backlog inside one segment close — a deeper queue drains across passes, which is what the batch cap is for. No contract change, so no release and re-pin. `RecordingProvider::segments_pending_summary` now honours `limit`. It returned every seeded segment, which would have let a test drive more than the caller asked for and hidden exactly this bug — the new regression test only fails against the old code because the fake is now truthful. `an_exhausted_queue_head_does_not_block_the_segments_behind_it` seeds four pending segments, spends the three heads' budget over two passes, and asserts the third pass still reaches the fourth. It asserts on the count of `episodic.session_turns` calls because that is the call every attempted segment makes and the recording driver does not carry arguments; against the old code the third pass records zero. --- .../agent/harness/archivist/resummarise.rs | 29 +++++++++- .../harness/archivist/resummarise_tests.rs | 53 +++++++++++++++++++ .../memory/guard/test_support_part_02.rs | 13 ++++- 3 files changed, 91 insertions(+), 4 deletions(-) diff --git a/src/openhuman/agent/harness/archivist/resummarise.rs b/src/openhuman/agent/harness/archivist/resummarise.rs index e3755be913..ca09826978 100644 --- a/src/openhuman/agent/harness/archivist/resummarise.rs +++ b/src/openhuman/agent/harness/archivist/resummarise.rs @@ -34,7 +34,24 @@ use crate::openhuman::memory::api::provider::episodic::EpisodicTurn; /// the queue drains across many passes rather than in one long stall; a large /// batch would put a run of inference calls behind a single close, and /// `flush_open_segment` is awaited at session wind-down. -const RESUMMARISE_BATCH: u32 = 3; +const RESUMMARISE_BATCH: usize = 3; + +/// How far down the pending queue one pass will look to find those segments. +/// +/// The batch cap alone is not enough, and this is the correction to the first +/// version of this pass. `segments_pending_summary` orders oldest-first and +/// takes a `limit`, so asking for exactly [`RESUMMARISE_BATCH`] returns the +/// three oldest rows *before* the attempt ledger filters them. Once those three +/// exhaust their attempts they still occupy every result, and a fourth pending +/// segment can never be reached until the process restarts — the head-of-queue +/// trap, moved one layer out rather than solved. +/// +/// So the pass reads a window and stops after [`RESUMMARISE_BATCH`] **eligible** +/// segments. Bounded rather than unbounded because the point is to make +/// progress past a stuck head, not to walk an arbitrarily long backlog inside +/// one segment close; a queue deeper than this drains across passes, which is +/// what the batch cap is for. +const RESUMMARISE_SCAN: u32 = 25; /// How many times one segment may be attempted before this process gives up on /// it. @@ -90,7 +107,7 @@ impl ArchivistHook { return; }; - let pending = match episodic.segments_pending_summary(RESUMMARISE_BATCH).await { + let pending = match episodic.segments_pending_summary(RESUMMARISE_SCAN).await { Ok(pending) => pending, Err(e) => { tracing::debug!("[archivist] resummarise: cannot read the pending queue: {e}"); @@ -106,7 +123,11 @@ impl ArchivistHook { pending.len() ); + let mut attempted = 0_usize; for segment in pending { + if attempted >= RESUMMARISE_BATCH { + break; + } if exhausted(&segment.segment_id) { tracing::debug!( "[archivist] resummarise: segment={} already attempted {} times in this \ @@ -126,6 +147,10 @@ impl ArchivistHook { // belongs to a segment by stable per-session sequence or row id, // not by timestamp, because the md store rounds to milliseconds and // can sort a fast turn just before its own segment's start. + // Counted here rather than at the top of the loop: a skipped + // exhausted segment must not consume the batch, which is the whole + // point of scanning past it. + attempted += 1; let entries = self.read_session_entries(&segment.session_id).await; let segment_entries: Vec<&EpisodicTurn> = entries .iter() diff --git a/src/openhuman/agent/harness/archivist/resummarise_tests.rs b/src/openhuman/agent/harness/archivist/resummarise_tests.rs index c3fe803086..b5a706566a 100644 --- a/src/openhuman/agent/harness/archivist/resummarise_tests.rs +++ b/src/openhuman/agent/harness/archivist/resummarise_tests.rs @@ -214,3 +214,56 @@ async fn a_segment_with_no_turns_left_is_skipped() { "a segment with no turns was summarised anyway: {methods:?}" ); } + +/// A queue whose head is exhausted still lets the segments behind it recover. +/// +/// The batch cap alone did not do this, and that was a real bug in the first +/// version of this pass (caught in review of #6183). `segments_pending_summary` +/// orders oldest-first and takes a `limit`, so asking for exactly the batch size +/// returns the oldest rows *before* the ledger filters them: once those exhaust +/// their attempts they occupy every result forever and nothing behind them is +/// ever reached. The pass reads a window and counts only eligible segments +/// against the batch. +/// +/// Asserted on the **count** of `episodic.session_turns` calls, because that is +/// the call every attempted segment makes and the recording driver does not +/// carry arguments. Against the old code the third pass records zero. +#[tokio::test] +async fn an_exhausted_queue_head_does_not_block_the_segments_behind_it() { + let recording = Arc::new( + RecordingProvider::new() + .with_session_turns(turns()) + .with_pending_segments(vec![ + pending("seg-head-a"), + pending("seg-head-b"), + pending("seg-head-c"), + pending("seg-behind"), + ]), + ); + let hook = hook_over(&recording); + + let turn_reads = |recording: &RecordingProvider| { + methods(recording) + .iter() + .filter(|m| *m == "episodic.session_turns") + .count() + }; + + // Two passes spend the three heads' whole two-attempt budget. + hook.resummarise_pending(300.0).await; + hook.resummarise_pending(301.0).await; + let after_two = turn_reads(&recording); + assert_eq!( + after_two, 6, + "the first two passes should each have attempted the three heads" + ); + + // The third finds them exhausted and must scan past them. + hook.resummarise_pending(302.0).await; + + assert_eq!( + turn_reads(&recording) - after_two, + 1, + "the segment behind three exhausted heads was never reached" + ); +} diff --git a/src/openhuman/memory/guard/test_support_part_02.rs b/src/openhuman/memory/guard/test_support_part_02.rs index 15b902f797..779d648fab 100644 --- a/src/openhuman/memory/guard/test_support_part_02.rs +++ b/src/openhuman/memory/guard/test_support_part_02.rs @@ -263,13 +263,22 @@ impl MemoryEpisodic for RecordingProvider { async fn segments_pending_summary( &self, - _limit: u32, + limit: u32, ) -> Result< Vec, MemoryError, > { self.record(Call::plain("episodic.segments_pending_summary")); - Ok(self.pending_segments.lock().unwrap().clone()) + // Honour `limit` — a fake that ignored it would let a test drive more + // segments than the caller asked for and hide a bounded-recovery bug. + Ok(self + .pending_segments + .lock() + .unwrap() + .iter() + .take(limit as usize) + .cloned() + .collect()) } async fn create_segment(