Improve OpenHuman benchmark parity and safety coverage

crates/tinymemory-api/src/conformance/reference/mod.rs

@@ -11,6 +11,7 @@
 mod distil;
 mod score;
 
+use std::collections::HashSet;
 use std::sync::Mutex;
 
 pub use distil::CONSOLIDATED_TAG;
@@ -213,6 +214,26 @@ impl MemoryEngine for ReferenceEngine {
         Ok(StoreReceipt { id, replayed })
     }
 
+    async fn store_many(&self, batch: Vec<StoreItem>) -> Result<Vec<StoreReceipt>> {
+        crate::validate_many(&batch)?;
+        let mut items = self.items()?;
+        // Compute held fingerprints once per batch. Calling `store` for each
+        // item rehashes every preceding item, which makes scale sweeps
+        // quadratic in the number of documents.
+        let mut seen: HashSet<String> = items.iter().map(StoreItem::fingerprint).collect();
+        let mut receipts = Vec::with_capacity(batch.len());
+        for item in batch {
+            item.validate()?;
+            let id = ItemId(item.fingerprint());
+            let replayed = !seen.insert(id.0.clone());
+            if !replayed {
+                items.push(item);
+            }
+            receipts.push(StoreReceipt { id, replayed });
+        }
+        Ok(receipts)
+    }
+
     async fn forget(&self, target: ForgetTarget) -> Result<ForgetReport> {
         target.validate()?;
         let mut items = self.items()?;

crates/tinymemory-api/src/conformance/reference/mod_tests.rs

@@ -4,6 +4,37 @@ use crate::{ItemKind, LearningKind, MemoryMeta};
 
 use super::*;
 
+#[tokio::test]
+async fn bulk_store_preserves_order_and_marks_existing_and_in_batch_replays() {
+    let engine = ReferenceEngine::new();
+    let a = StoreItem::document("alpha", MemoryMeta::default());
+    let b = StoreItem::document("beta", MemoryMeta::default());
+    engine.store(a.clone()).await.unwrap();
+    let receipts = engine
+        .store_many(vec![a.clone(), b.clone(), b.clone()])
+        .await
+        .unwrap();
+    assert_eq!(
+        receipts
+            .iter()
+            .map(|receipt| receipt.replayed)
+            .collect::<Vec<_>>(),
+        vec![true, false, true]
+    );
+    assert_eq!(engine.len(), 2);
+    let page = engine
+        .export(ListRequest::new(MetaFilter::default(), 10))
+        .await
+        .unwrap();
+    assert_eq!(
+        page.items
+            .into_iter()
+            .map(|entry| entry.item)
+            .collect::<Vec<_>>(),
+        vec![a, b]
+    );
+}
+
 #[tokio::test]
 async fn a_bad_cursor_is_an_invalid_request() {
     let engine = ReferenceEngine::new();

crates/tinymemory-integrations/examples/memory_eval/agent.rs

@@ -24,6 +24,36 @@ use tinymemory_tools::{AgentMemory, BackgroundJob, PostTurn, PreTurn, TurnContex
 
 /// OpenHuman's default deadline for the pack before a model turn.
 pub(crate) const PRE_TURN_TIMEOUT: StdDuration = StdDuration::from_millis(1_500);
+
+/// The host selects one lifecycle hook from its recall configuration and
+/// whether the thread is resuming after compaction.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum HostHook {
+    Plain,
+    Resumed,
+    Dated,
+    DatedResumed,
+}
+
+impl HostHook {
+    pub(crate) fn for_turn(resumed: bool, date_hint: bool) -> Self {
+        match (resumed, date_hint) {
+            (false, false) => Self::Plain,
+            (true, false) => Self::Resumed,
+            (false, true) => Self::Dated,
+            (true, true) => Self::DatedResumed,
+        }
+    }
+
+    pub(crate) async fn run(self, memory: AgentMemory, pre: PreTurn) -> Result<TurnContext, Error> {
+        match self {
+            Self::Plain => memory.pre_turn(pre).await,
+            Self::Resumed => memory.pre_turn_resumed(pre).await,
+            Self::Dated => memory.pre_turn_dated(pre, false, async { None }).await,
+            Self::DatedResumed => memory.pre_turn_dated(pre, true, async { None }).await,
+        }
+    }
+}
 /// OpenHuman's maximum length of one logged tool result.
 const MAX_TOOL_LINE_CHARS: usize = 240;
 
@@ -59,6 +89,11 @@ pub(crate) struct TurnRecord {
     pub(crate) tool_calls: usize,
 }
 
+type PendingPreTurn = (
+    Instant,
+    tokio::task::JoinHandle<(Result<TurnContext, Error>, Instant)>,
+);
+
 /// One conversation thread driven by the script.
 pub(crate) struct ScriptedAgent {
     memory: AgentMemory,
@@ -68,7 +103,14 @@ pub(crate) struct ScriptedAgent {
     window: u32,
     clock: Option<DateTime<Utc>>,
     openhuman: bool,
-    pending: Vec<tokio::task::JoinHandle<Result<TurnContext, Error>>>,
+    date_hint: bool,
+    pending: Vec<PendingPreTurn>,
+}
+
+/// User turns that completed after the simulated host stopped waiting.
+pub(crate) struct FlushReport {
+    pub(crate) logged: usize,
+    pub(crate) completion_ms: Vec<f64>,
 }
 
 impl ScriptedAgent {
@@ -82,27 +124,40 @@ impl ScriptedAgent {
             window,
             clock: None,
             openhuman: false,
+            date_hint: false,
             pending: Vec::new(),
         }
     }
 
-    /// Mirror OpenHuman's dated hook, deadline, and reply logging.
+    /// Mirror OpenHuman's default hook, deadline, and reply logging.
     pub(crate) fn openhuman(mut self) -> Self {
         self.openhuman = true;
         self
     }
 
+    /// Exercise the host's optional dated-recall path.
+    pub(crate) fn date_hint(mut self) -> Self {
+        self.date_hint = true;
+        self
+    }
+
     /// Wait for pre-turn tasks whose host deadline expired. OpenHuman leaves
     /// those tasks running, so their accepted user turns may still land.
-    pub(crate) async fn flush(&mut self) -> Result<usize, Error> {
+    pub(crate) async fn flush(&mut self) -> Result<FlushReport, Error> {
         let mut logged = 0;
-        for task in self.pending.drain(..) {
-            let context = task
+        let mut completion_ms = Vec::new();
+        for (started, task) in self.pending.drain(..) {
+            let (context, completed) = task
                 .await
-                .map_err(|error| Error::Unavailable(format!("pre-turn task failed: {error}")))??;
+                .map_err(|error| Error::Unavailable(format!("pre-turn task failed: {error}")))?;
+            let context = context?;
             logged += usize::from(context.logged.is_some());
+            completion_ms.push(completed.duration_since(started).as_secs_f64() * 1_000.0);
         }
-        Ok(logged)
+        Ok(FlushReport {
+            logged,
+            completion_ms,
+        })
     }
 
     /// Timestamps the thread's turns from `at`, a minute apart.
@@ -136,16 +191,21 @@ impl ScriptedAgent {
         let started = Instant::now();
         let context = if self.openhuman {
             let memory = self.memory.clone();
-            let mut task =
-                tokio::spawn(
-                    async move { memory.pre_turn_dated(pre, false, async { None }).await },
-                );
+            let hook = HostHook::for_turn(false, self.date_hint);
+            let mut task = tokio::spawn(async move {
+                let result = hook.run(memory, pre).await;
+                (result, Instant::now())
+            });
             match tokio::time::timeout(PRE_TURN_TIMEOUT, &mut task).await {
-                Ok(result) => Some(result.map_err(|error| {
-                    Error::Unavailable(format!("pre-turn task failed: {error}"))
-                })??),
+                Ok(result) => Some(
+                    result
+                        .map_err(|error| {
+                            Error::Unavailable(format!("pre-turn task failed: {error}"))
+                        })?
+                        .0?,
+                ),
                 Err(_) => {
-                    self.pending.push(task);
+                    self.pending.push((started, task));
                     None
                 }
             }

crates/tinymemory-integrations/examples/memory_eval/agent_tests.rs

@@ -6,6 +6,14 @@ use tinymemory_api::conformance::ReferenceEngine;
 use tinymemory_api::{LearningKind, ListRequest, MemoryEngine, MemoryMeta, StoreItem};
 use tinymemory_tools::MemoryLayout;
 
+#[test]
+fn openhuman_defaults_to_plain_recall_and_resumes_after_compaction() {
+    assert_eq!(HostHook::for_turn(false, false), HostHook::Plain);
+    assert_eq!(HostHook::for_turn(true, false), HostHook::Resumed);
+    assert_eq!(HostHook::for_turn(false, true), HostHook::Dated);
+    assert_eq!(HostHook::for_turn(true, true), HostHook::DatedResumed);
+}
+
 #[test]
 fn logged_reply_keeps_bounded_tool_results() {
     let long = "a".repeat(MAX_TOOL_LINE_CHARS + 20);

crates/tinymemory-integrations/examples/memory_eval/instrument.rs

@@ -0,0 +1,49 @@
+//! Collect narrow library timing traces in the eval JSON report.
+
+use std::sync::Mutex;
+
+use crate::Timings;
+
+const TARGET: &str = "tinymemory_eval_timing";
+
+struct TimingLogger(Mutex<Vec<(String, f64)>>);
+
+static LOGGER: TimingLogger = TimingLogger(Mutex::new(Vec::new()));
+
+impl log::Log for TimingLogger {
+    fn enabled(&self, metadata: &log::Metadata<'_>) -> bool {
+        metadata.target() == TARGET
+    }
+
+    fn log(&self, record: &log::Record<'_>) {
+        if !self.enabled(record.metadata()) {
+            return;
+        }
+        let line = record.args().to_string();
+        let Some((stage, value)) = line.split_once('=') else {
+            return;
+        };
+        let Ok(ms) = value.parse::<f64>() else {
+            return;
+        };
+        if let Ok(mut samples) = self.0.lock() {
+            samples.push((stage.replace('_', " "), ms));
+        }
+    }
+
+    fn flush(&self) {}
+}
+
+pub(crate) fn install() {
+    if log::set_logger(&LOGGER).is_ok() {
+        log::set_max_level(log::LevelFilter::Trace);
+    }
+}
+
+pub(crate) fn drain(timings: &mut Timings) {
+    if let Ok(mut samples) = LOGGER.0.lock() {
+        for (stage, ms) in samples.drain(..) {
+            timings.add(&stage, ms);
+        }
+    }
+}

crates/tinymemory-integrations/examples/memory_eval/kpi_tests.rs

@@ -28,6 +28,8 @@ fn report(name: &'static str, probes: Vec<ProbeResult>, conflicts: &[&str]) -> S
         writes: 0,
         tool_calls: 0,
         pre_turn_timeouts: 0,
+        ranked_ready: None,
+        ranked_wait_ms: None,
         settle_ms: 0.0,
         synthesis: Synthesis {
             captured: Captured {

crates/tinymemory-integrations/examples/memory_eval/loop_guard.rs

@@ -74,7 +74,7 @@ pub(crate) async fn run(
             runner.run(job).await?;
         }
     }
-    agent.flush().await?;
+    let _ = agent.flush().await?;
     let mut echoed_items = 0;
     let mut cursor = None;
     loop {