diff --git a/Cargo.lock b/Cargo.lock index 9786ffa9..0922f32f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -883,6 +883,7 @@ dependencies = [ "serde", "serde_json", "tempfile", + "tiktoken-rs", "tracing", ] diff --git a/crates/cmem-eval-locomo/src/lib.rs b/crates/cmem-eval-locomo/src/lib.rs index f644e4dd..618fbdda 100644 --- a/crates/cmem-eval-locomo/src/lib.rs +++ b/crates/cmem-eval-locomo/src/lib.rs @@ -17,6 +17,9 @@ //! `:qa:`. Effective QA IDs are unique across the file. //! `answer`, `category`, `speaker`, `img_url`, `blip_caption`, `query` and dates //! remain optional annotations. Raw dates are retained alongside normalized timestamps. +//! The conventional runner uses the latest parseable source timestamp, inclusive, +//! as the scene time for every QA. Official dates without an offset mean UTC. +//! It rejects a conversation with no parseable timestamp before creating run state. //! QA `evidence` admits scalar references and arrays of scalars or objects with //! `dia_id`. References to absent turns do not reject the sample. Missing //! annotations do not determine abstention status. diff --git a/crates/cmem-eval-longmemeval/Cargo.toml b/crates/cmem-eval-longmemeval/Cargo.toml index 3e7b8c98..08cb7d5f 100644 --- a/crates/cmem-eval-longmemeval/Cargo.toml +++ b/crates/cmem-eval-longmemeval/Cargo.toml @@ -9,6 +9,7 @@ chrono.workspace = true serde.workspace = true serde_json.workspace = true tracing.workspace = true +tiktoken-rs.workspace = true cmem-eval = { path = "../cmem-eval" } [dev-dependencies] diff --git a/crates/cmem-eval-longmemeval/src/identity.rs b/crates/cmem-eval-longmemeval/src/identity.rs index 61466d6f..fbf9b182 100644 --- a/crates/cmem-eval-longmemeval/src/identity.rs +++ b/crates/cmem-eval-longmemeval/src/identity.rs @@ -38,13 +38,24 @@ impl Identities { *ordinal += 1; identities.session_ids.insert(assigned.clone(), raw.clone()); for turn in &session.turns { - identities.turn_ids.insert( - format!("{assigned}:turn:{}", turn.index), - format!("{raw}:turn:{}", turn.index), - ); + for (part, _) in turn.observation_texts().enumerate() { + identities.turn_ids.insert( + observation_id(&assigned, turn.index, part), + format!("{raw}:turn:{}", turn.index), + ); + } } identities.episodes.push(assigned); } identities } } + +pub(crate) fn observation_id(episode: &str, turn: usize, part: usize) -> String { + let original = format!("{episode}:turn:{turn}"); + if part == 0 { + original + } else { + format!("{original}:part:{}", part + 1) + } +} diff --git a/crates/cmem-eval-longmemeval/src/ingest.rs b/crates/cmem-eval-longmemeval/src/ingest.rs index 65f50b0a..83d620d4 100644 --- a/crates/cmem-eval-longmemeval/src/ingest.rs +++ b/crates/cmem-eval-longmemeval/src/ingest.rs @@ -42,19 +42,21 @@ pub fn to_memory_inputs(instance: &LongMemEvalInstance) -> LongMemEvalMemoryInpu }), }); for turn in &session.turns { - observations.push(ObservationInput { - external_id: format!("{}:turn:{}", assigned, turn.index), - episode_external_id: assigned.clone(), - namespace: namespace.clone(), - speaker: turn.speaker.clone(), - text: turn.text.clone(), - observed_at: session.date.clone(), - metadata: serde_json::json!({ - "source": "longmemeval_s", - "raw_date": session.raw_date.clone(), - "normalized_date": session.date.clone() - }), - }); + for (part, text) in turn.observation_texts().enumerate() { + observations.push(ObservationInput { + external_id: crate::identity::observation_id(assigned, turn.index, part), + episode_external_id: assigned.clone(), + namespace: namespace.clone(), + speaker: turn.speaker.clone(), + text: text.to_string(), + observed_at: session.date.clone(), + metadata: serde_json::json!({ + "source": "longmemeval_s", + "raw_date": session.raw_date.clone(), + "normalized_date": session.date.clone() + }), + }); + } } } diff --git a/crates/cmem-eval-longmemeval/src/lib.rs b/crates/cmem-eval-longmemeval/src/lib.rs index eb203ffe..03f53019 100644 --- a/crates/cmem-eval-longmemeval/src/lib.rs +++ b/crates/cmem-eval-longmemeval/src/lib.rs @@ -16,10 +16,14 @@ //! `haystack_dates` is optional: absent or null is admitted; otherwise it must be //! an array matching the session count. Each null or non-string slot means no //! date at that position, without shifting later dates. Raw dates are retained -//! alongside normalized timestamps. `question_type`, `answer`, `question_date`, -//! `role` and `answer_session_ids` remain optional. `has_answer` defaults to false -//! when absent or not boolean; references to absent sessions do not reject the -//! item. Optional annotations do not determine abstention status. +//! alongside normalized timestamps. `question_date` is optional when absent or +//! null; supplied dates must parse as an official timestamp or RFC3339 and are +//! normalized to RFC3339 UTC, treating official timestamps without an offset as +//! UTC. Invalid dates reject the item with its question identity. +//! `question_type`, `answer`, `role` and `answer_session_ids` remain optional. +//! `has_answer` defaults to false when absent or not boolean; references to absent +//! sessions do not reject the item. Optional annotations do not determine +//! abstention status. //! //! Ingest assigns each occurrence its own episode identity: the first keeps the //! raw benchmark session ID, later copies receive `#2`, `#3`, ... while skipping @@ -28,6 +32,17 @@ //! raw ID. Scoring recomputes the same identity tables from the item to map both //! session and turn rankings back to benchmark IDs, crediting each ID only at //! its first retrieved rank. Retrieved row IDs are assigned; gold IDs are raw. +//! +//! Oversized turns retain their original text/index but gain observation parts +//! at greedy sentence/blank-paragraph boundaries, each native prefixed surface +//! at most 8192 cl100k tokens. If none fits, use the last newline, then the last +//! whitespace under the limit; an unbreakable oversized span is an admission +//! error. The manifest records each part's boundary kind; source end closes its +//! final paragraph. +//! Ingest keeps the original ID for the first part and uses `:part:2`, ... for +//! later parts in the same episode/speaker/time. Scoring maps every part to the +//! original turn; full-history text and converter source text stay unchanged. +//! Conventional run headers record affected identities and the split policy. mod error; mod identity; diff --git a/crates/cmem-eval-longmemeval/src/loader.rs b/crates/cmem-eval-longmemeval/src/loader.rs index fe545ab2..4ef64b8d 100644 --- a/crates/cmem-eval-longmemeval/src/loader.rs +++ b/crates/cmem-eval-longmemeval/src/loader.rs @@ -51,12 +51,30 @@ fn parse_instance(raw: Value, index: usize) -> Result None, + Some(value) => Some( + value + .as_str() + .and_then(|value| { + let value = value.trim(); + DateTime::parse_from_rfc3339(value) + .map(|timestamp| timestamp.with_timezone(&Utc)) + .ok() + .or_else(|| parse_official_longmemeval_timestamp(value)) + }) + .map(|timestamp| timestamp.to_rfc3339_opts(SecondsFormat::AutoSi, true)) + .ok_or_else(|| { + location.error("question_date", "expected an official timestamp or RFC3339") + })?, + ), + }; Ok(LongMemEvalInstance { question_id, question_type: string_field(&raw, "question_type"), question, answer: string_field(&raw, "answer"), - question_date: string_field(&raw, "question_date"), + question_date, sessions, answer_session_ids: string_array(raw.get("answer_session_ids")), }) @@ -195,12 +213,20 @@ fn parse_turns( if !turn.is_object() { return Err(location.error(format!("{field}[{idx}]"), "expected a turn object")); } + let text = string_field(turn, "content").ok_or_else(|| { + location.error(format!("{field}[{idx}].content"), "expected a string") + })?; + let observation_parts = split_observation(&text).ok_or_else(|| { + location.error( + format!("{field}[{idx}].content"), + "a span without a usable boundary exceeds the 8192-token observation limit", + ) + })?; Ok(LongMemEvalTurn { index: idx + 1, speaker: string_field(turn, "role"), - text: string_field(turn, "content").ok_or_else(|| { - location.error(format!("{field}[{idx}].content"), "expected a string") - })?, + text, + observation_parts, has_answer: turn .get("has_answer") .and_then(Value::as_bool) @@ -210,6 +236,78 @@ fn parse_turns( .collect() } +pub(crate) const OBSERVATION_TOKEN_LIMIT: usize = 8192; + +fn observation_tokens(text: &str) -> usize { + // Include the native surface prefix in the provider's token limit. + let clean = text.split_whitespace().collect::>().join(" "); + let surface = if clean.is_empty() { + "Observation excerpt".to_string() + } else { + format!("Observation excerpt: {clean}") + }; + tiktoken_rs::cl100k_base_singleton() + .encode_ordinary(&surface) + .len() +} + +fn split_observation(text: &str) -> Option> { + use cmem_eval::ObservationBoundary::{Newline, Paragraph, Sentence, Whitespace}; + if text.len() + "Observation excerpt: ".len() <= OBSERVATION_TOKEN_LIMIT + || observation_tokens(text) <= OBSERVATION_TOKEN_LIMIT + { + return Some(Vec::new()); + } + let ends = text + .char_indices() + .filter_map(|(index, ch)| { + let end = index + ch.len_utf8(); + let sentence = matches!(ch, '.' | '!' | '?') + && text[end..].chars().next().is_none_or(char::is_whitespace); + let kind = if ch == '\n' + && (text[end..].starts_with('\n') || text[end..].starts_with("\r\n")) + { + Paragraph + } else if sentence { + Sentence + } else if ch == '\n' { + Newline + } else if ch.is_whitespace() { + Whitespace + } else { + return None; + }; + Some((end, kind)) + }) + .collect::>(); + let mut parts = Vec::new(); + let mut start = 0; + while observation_tokens(&text[start..]) > OBSERVATION_TOKEN_LIMIT { + let last_fit = |kinds: &[cmem_eval::ObservationBoundary]| { + ends.iter().rev().copied().find(|(end, kind)| { + *end > start + && kinds.contains(kind) + && !text[start..*end].trim().is_empty() + && observation_tokens(&text[start..*end]) <= OBSERVATION_TOKEN_LIMIT + }) + }; + let (end, boundary_kind) = last_fit(&[Paragraph, Sentence]) + .or_else(|| last_fit(&[Newline])) + .or_else(|| last_fit(&[Whitespace]))?; + parts.push(crate::LongMemEvalObservationPart { + text: text[start..end].to_string(), + boundary_kind, + }); + start = end; + } + // End of the source turn closes its final paragraph without another cut. + parts.push(crate::LongMemEvalObservationPart { + text: text[start..].to_string(), + boundary_kind: Paragraph, + }); + Some(parts) +} + fn string_field(value: &Value, key: &str) -> Option { value .get(key) @@ -234,6 +332,59 @@ fn string_array(value: Option<&Value>) -> Vec { mod tests { use super::*; + #[test] + fn records_preferred_boundary_then_newline_and_whitespace_fallbacks() { + use cmem_eval::ObservationBoundary::{Newline, Paragraph, Sentence, Whitespace}; + for (separator, expected) in [ + (". ", Sentence), + ("\n\n", Paragraph), + ("\n", Newline), + (" ", Whitespace), + ] { + let text = format!( + "{}{separator}{}", + "alpha ".repeat(4500), + "beta ".repeat(4500) + ); + let parts = split_observation(&text).unwrap(); + assert_eq!(parts[0].boundary_kind, expected); + assert_eq!(parts.last().unwrap().boundary_kind, Paragraph); + assert_eq!( + parts + .iter() + .map(|part| part.text.as_str()) + .collect::(), + text + ); + assert!( + parts + .iter() + .all(|part| observation_tokens(&part.text) <= OBSERVATION_TOKEN_LIMIT) + ); + } + } + + #[test] + fn paragraph_split_preserves_unicode_and_crlf_bytes_under_native_token_limit() { + let text = format!("{}\r\n\r\n{}", "猫 ".repeat(2200), "猫 ".repeat(2200)); + assert!(observation_tokens(&text) > OBSERVATION_TOKEN_LIMIT); + let parts = split_observation(&text).unwrap(); + assert_eq!(parts.len(), 2); + assert_eq!( + parts + .iter() + .map(|part| part.text.as_str()) + .collect::(), + text + ); + assert!( + parts + .iter() + .all(|part| observation_tokens(&part.text) <= OBSERVATION_TOKEN_LIMIT) + ); + assert!(split_observation(&" \n".repeat(9000)).unwrap().is_empty()); + } + #[test] fn parses_official_fixture() { let value = serde_json::json!([{ diff --git a/crates/cmem-eval-longmemeval/src/types.rs b/crates/cmem-eval-longmemeval/src/types.rs index fcecbcd8..19c16409 100644 --- a/crates/cmem-eval-longmemeval/src/types.rs +++ b/crates/cmem-eval-longmemeval/src/types.rs @@ -28,6 +28,49 @@ impl LongMemEvalInstance { }) .collect() } + + pub fn observation_splits(&self) -> Vec { + if !self + .sessions + .iter() + .flat_map(|session| &session.turns) + .any(|turn| !turn.observation_parts.is_empty()) + { + return Vec::new(); + } + let identities = crate::identity::Identities::new(self); + self.sessions + .iter() + .zip(&identities.episodes) + .flat_map(|(session, episode)| { + session + .turns + .iter() + .filter(|turn| !turn.observation_parts.is_empty()) + .map(|turn| cmem_eval::ObservationSplit { + dataset_item_id: self.question_id.clone(), + source_observation_id: crate::identity::observation_id( + episode, turn.index, 0, + ), + parts: turn + .observation_parts + .iter() + .enumerate() + .map(|(index, part)| cmem_eval::ObservationSplitPart { + observation_id: crate::identity::observation_id( + episode, turn.index, index, + ), + boundary_kind: part.boundary_kind, + }) + .collect(), + boundary_policy: "greedy_sentence_paragraph_then_newline_whitespace_v1" + .into(), + tokenizer: "cl100k_base".into(), + max_input_tokens: crate::loader::OBSERVATION_TOKEN_LIMIT, + }) + }) + .collect() + } } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -43,9 +86,26 @@ pub struct LongMemEvalTurn { pub index: usize, pub speaker: Option, pub text: String, + pub observation_parts: Vec, pub has_answer: bool, } +impl LongMemEvalTurn { + pub fn observation_texts(&self) -> impl Iterator { + self.observation_parts + .is_empty() + .then_some(self.text.as_str()) + .into_iter() + .chain(self.observation_parts.iter().map(|part| part.text.as_str())) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct LongMemEvalObservationPart { + pub text: String, + pub boundary_kind: cmem_eval::ObservationBoundary, +} + #[derive(Debug, Clone)] pub struct LongMemEvalMemoryInputs { pub episodes: Vec, diff --git a/crates/cmem-eval-longmemeval/tests/admission.rs b/crates/cmem-eval-longmemeval/tests/admission.rs index 3b68e955..47bc6300 100644 --- a/crates/cmem-eval-longmemeval/tests/admission.rs +++ b/crates/cmem-eval-longmemeval/tests/admission.rs @@ -267,12 +267,7 @@ fn admits_parallel_session_repeats_with_identical_turns_and_different_dates() { #[test] fn admits_missing_or_malformed_optional_annotations() { - for field in [ - "question_type", - "answer", - "question_date", - "answer_session_ids", - ] { + for field in ["question_type", "answer", "answer_session_ids"] { for value in [None, Some(json!(null)), Some(json!({}))] { let mut row = item(); if let Some(value) = value { @@ -298,6 +293,48 @@ fn admits_missing_or_malformed_optional_annotations() { } } +#[test] +fn normalizes_question_dates_to_rfc3339_utc() { + for (value, expected) in [ + (None, None), + (Some(json!(null)), None), + ( + Some(json!("2023/05/30 (Tue) 23:40")), + Some("2023-05-30T23:40:00Z"), + ), + ( + Some(json!("2023/05/30 23:40")), + Some("2023-05-30T23:40:00Z"), + ), + ( + Some(json!(" 2023-05-31T08:40:00.123+09:00 ")), + Some("2023-05-30T23:40:00.123Z"), + ), + ] { + let mut row = item(); + if let Some(value) = value { + row["question_date"] = value; + } + let rows = load_value(json!([row])).unwrap(); + assert_eq!(rows[0].question_date.as_deref(), expected); + } +} + +#[test] +fn rejects_invalid_question_dates_with_question_identity() { + for value in [ + json!("2023/02/30 (Thu) 23:40"), + json!("not-a-date"), + json!(" \t"), + json!({}), + json!(42), + ] { + let mut row = item(); + row["question_date"] = value; + rejected(json!([row]), "question_date", 0, Some("q1")); + } +} + #[test] fn admits_abstention_answers_and_empty_turn_text() { let mut row = item(); diff --git a/crates/cmem-eval-longmemeval/tests/observation_splits.rs b/crates/cmem-eval-longmemeval/tests/observation_splits.rs new file mode 100644 index 00000000..5213ecd9 --- /dev/null +++ b/crates/cmem-eval-longmemeval/tests/observation_splits.rs @@ -0,0 +1,156 @@ +use cmem_eval::{ObjectType, RetrievedItem}; +use cmem_eval_longmemeval::{ + full_history_text, ingest::to_memory_inputs, load_value, scoring::score, +}; + +#[test] +fn splits_oversized_turn_into_consecutive_observations_without_changing_source_or_gold() { + let text = format!( + "{}End.\n\n{}End.", + "alpha ".repeat(4500), + "beta ".repeat(4500) + ); + let input = serde_json::json!([{ + "question_id": "oversized-turn", + "question": "What happened?", + "haystack_session_ids": ["s1"], + "haystack_dates": ["2023/05/30 (Tue) 23:40"], + "haystack_sessions": [[ + {"role": "user", "content": text, "has_answer": true}, + {"role": "assistant", "content": "Next turn."} + ]], + "answer_session_ids": ["s1"] + }]); + let item = load_value(input.clone()).unwrap().remove(0); + let mapped = to_memory_inputs(&item); + assert_eq!(mapped.observations.len(), 3); + assert_eq!(mapped.observations[0].external_id, "s1:turn:1"); + assert_eq!(mapped.observations[1].external_id, "s1:turn:1:part:2"); + assert_eq!(mapped.observations[2].external_id, "s1:turn:2"); + assert_eq!( + mapped.observations[..2] + .iter() + .map(|o| o.text.as_str()) + .collect::(), + text + ); + for observation in &mapped.observations[..2] { + let surface = format!( + "Observation excerpt: {}", + observation + .text + .split_whitespace() + .collect::>() + .join(" ") + ); + assert!( + tiktoken_rs::cl100k_base_singleton() + .encode_ordinary(&surface) + .len() + <= 8192 + ); + assert_eq!(observation.episode_external_id, "s1"); + assert_eq!(observation.speaker.as_deref(), Some("user")); + assert_eq!( + observation.observed_at.as_deref(), + Some("2023-05-30T23:40:00Z") + ); + assert!(!observation.metadata.to_string().contains("has_answer")); + } + assert_eq!( + full_history_text(&item), + format!("user: {text}\nassistant: Next turn.") + ); + assert_eq!(item.gold_turn_ids(), ["s1:turn:1"]); + let retrieved = mapped.observations[..2] + .iter() + .enumerate() + .map(|(index, observation)| RetrievedItem { + kind: ObjectType::Observation, + internal_id: format!("id-{index}"), + external_id: Some(observation.external_id.clone()), + episode_external_id: Some("s1".into()), + score: None, + rank: index + 1, + rationale: vec![], + text: None, + }) + .collect::>(); + let metrics = score(&item, &retrieved[1..], &[1], &[1]); + assert_eq!(metrics["turn_recall_fraction@1"], 1.0); + assert_eq!(metrics["turn_mrr@1"], 1.0); + assert_eq!( + score(&item, &retrieved, &[1], &[2])["turn_recall_fraction@2"], + 1.0 + ); + assert_eq!( + serde_json::to_value(&item).unwrap(), + serde_json::to_value(load_value(input).unwrap().remove(0)).unwrap() + ); +} + +#[test] +fn leading_blank_lines_stay_with_the_first_content_part() { + for prefix in ["\n\n", "\r\n\r\n"] { + let text = format!("{prefix}{}", "alpha ".repeat(9000)); + let item = load_value(serde_json::json!([{ + "question_id": "leading-blank-lines", + "question": "What happened?", + "haystack_session_ids": ["s1"], + "haystack_sessions": [[ + {"content": text, "has_answer": true}, + {"content": "Second gold turn.", "has_answer": true} + ]] + }])) + .unwrap() + .remove(0); + let mapped = to_memory_inputs(&item); + assert!( + mapped + .observations + .iter() + .all(|o| !o.text.trim().is_empty()) + ); + assert_eq!(mapped.observations.len(), 3); + assert_eq!(mapped.observations[0].external_id, "s1:turn:1"); + assert!(mapped.observations[0].text.starts_with(prefix)); + assert!(mapped.observations[0].text.contains("alpha")); + assert_eq!( + mapped.observations[..2] + .iter() + .map(|o| o.text.as_str()) + .collect::(), + text + ); + for observation in &mapped.observations[..2] { + let clean = observation + .text + .split_whitespace() + .collect::>() + .join(" "); + let tokens = tiktoken_rs::cl100k_base_singleton() + .encode_ordinary(&format!("Observation excerpt: {clean}")) + .len(); + assert!(tokens <= 8192); + } + } +} + +#[test] +fn unsplittable_oversized_turn_names_the_question_before_ingest() { + let error = load_value(serde_json::json!([{ + "question_id": "unsplittable-turn", + "question": "What happened?", + "haystack_session_ids": ["s1"], + "haystack_sessions": [[{"content": "x".repeat(80000)}]] + }])) + .unwrap_err(); + assert!(matches!( + error, + cmem_eval_longmemeval::LoadError::Admission { .. } + )); + let message = error.to_string(); + assert!(message.contains("unsplittable-turn")); + assert!(message.contains("haystack_sessions[0].turns[0].content")); + assert!(message.contains("8192-token")); +} diff --git a/crates/cmem-eval-runner/src/pipeline.rs b/crates/cmem-eval-runner/src/pipeline.rs index 71c5cd46..411b72a4 100644 --- a/crates/cmem-eval-runner/src/pipeline.rs +++ b/crates/cmem-eval-runner/src/pipeline.rs @@ -3,7 +3,7 @@ use crate::commands::read_config; use crate::commands::{ContinuityRunArgs, RunArgs}; use crate::enrichment; use anyhow::{Context, Result, bail}; -use chrono::Utc; +use chrono::{DateTime, SecondsFormat, Utc}; use cmem_eval::CharacterMemoryAdapter; use cmem_eval::{ BenchmarkRunConfig, ControllableDimensionPolicy, EmbeddingBindingRecord, @@ -218,7 +218,7 @@ trait DatasetSpec { fn question_id(question: &Self::Question) -> &str; fn question_type(question: &Self::Question) -> Option; fn question_text(question: &Self::Question) -> &str; - fn query_date(question: &Self::Question) -> Option; + fn query_date(item: &Self::Item, question: &Self::Question) -> Result>; fn gold_episode_ids(item: &Self::Item, question: &Self::Question) -> Vec; fn gold_observation_ids(item: &Self::Item, question: &Self::Question) -> Vec; fn score( @@ -228,6 +228,9 @@ trait DatasetSpec { config: &BenchmarkRunConfig, ) -> Value; fn full_history_text(item: &Self::Item) -> String; + fn observation_splits(_item: &Self::Item) -> Vec { + Vec::new() + } fn enrichment( item: &Self::Item, namespace: &str, @@ -280,6 +283,10 @@ async fn run_pipeline(args: RunArgs) -> Result<()> { header .embedding_bindings .insert(config.dataset.to_string(), embedding_binding); + header.observation_splits = source_items + .iter() + .flat_map(S::observation_splits) + .collect(); adapter = if lexical { None } else { @@ -344,7 +351,7 @@ async fn run_pipeline(args: RunArgs) -> Result<()> { namespace: namespace.clone(), topic: Some(S::question_text(question).to_string()), scene: cmem_eval::MemorySceneInput { - time: S::query_date(question), + time: S::query_date(&item, question)?, ..Default::default() }, surface_policy: config.retrieval.surface_policy.clone(), @@ -737,8 +744,8 @@ impl DatasetSpec for LongMemEvalSpec { &question.question } - fn query_date(question: &Self::Question) -> Option { - question.question_date.clone() + fn query_date(_item: &Self::Item, question: &Self::Question) -> Result> { + Ok(question.question_date.clone()) } fn gold_episode_ids(_item: &Self::Item, question: &Self::Question) -> Vec { @@ -767,6 +774,10 @@ impl DatasetSpec for LongMemEvalSpec { cmem_eval_longmemeval::full_history_text(item) } + fn observation_splits(item: &Self::Item) -> Vec { + item.observation_splits() + } + fn enrichment( item: &Self::Item, namespace: &str, @@ -797,6 +808,23 @@ impl DatasetSpec for LongMemEvalSpec { struct LoCoMoSpec; +#[derive(Debug)] +struct MissingLoCoMoQueryTimestamp { + sample_id: String, +} + +impl std::fmt::Display for MissingLoCoMoQueryTimestamp { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + formatter, + "LoCoMo conversation {:?} has no parseable source timestamp", + self.sample_id + ) + } +} + +impl std::error::Error for MissingLoCoMoQueryTimestamp {} + impl DatasetSpec for LoCoMoSpec { type Item = cmem_eval_locomo::LoCoMoSample; type Question = cmem_eval_locomo::LoCoMoQa; @@ -813,7 +841,12 @@ impl DatasetSpec for LoCoMoSpec { } fn load(source: &str) -> Result> { - Ok(cmem_eval_locomo::load_value(serde_json::from_str(source)?)?) + let items = cmem_eval_locomo::load_value(serde_json::from_str(source)?)?; + for item in &items { + // The loader admits non-empty QA; reject missing time before native writes. + Self::query_date(item, &item.qa[0])?; + } + Ok(items) } fn item_id(item: &Self::Item) -> &str { @@ -858,8 +891,21 @@ impl DatasetSpec for LoCoMoSpec { &question.question } - fn query_date(_question: &Self::Question) -> Option { - None + fn query_date(item: &Self::Item, _question: &Self::Question) -> Result> { + let timestamp = item + .sessions + .iter() + .filter_map(|session| session.timestamp.as_deref()) + .filter_map(|timestamp| DateTime::parse_from_rfc3339(timestamp).ok()) + .max() + .ok_or_else(|| MissingLoCoMoQueryTimestamp { + sample_id: item.sample_id.clone(), + })?; + Ok(Some( + timestamp + .with_timezone(&Utc) + .to_rfc3339_opts(SecondsFormat::AutoSi, true), + )) } fn gold_episode_ids(item: &Self::Item, question: &Self::Question) -> Vec { @@ -1070,6 +1116,7 @@ fn run_header( run_id: config.run_id.clone(), dataset: config.dataset.clone(), input_sha256, + observation_splits: Vec::new(), embedding_bindings: BTreeMap::new(), self_notion_ids: BTreeMap::new(), harness_commit: commit(&workspace)?, @@ -1605,6 +1652,7 @@ mod tests { serde_json::to_vec(&serde_json::json!([{ "sample_id": "p1", "conversation": { + "session_1_date_time": "2024-01-01T00:00:00Z", "session_1": [{"dia_id": "d1", "speaker": "A", "text": "likes tea"}] }, "qa": [ @@ -2208,7 +2256,7 @@ mod tests { fn derived_locomo_fixture() -> Value { serde_json::json!([{ "sample_id": "p1", - "conversation": {"session_1": [{"dia_id": "d1", "text": "jasmine tea"}]}, + "conversation": {"session_1_date_time": "2024-01-01T00:00:00Z", "session_1": [{"dia_id": "d1", "text": "jasmine tea"}]}, "session_summary": {"session_1_summary": "UNIQUE_SUMMARY_TOKEN"}, "observation": {"session_1_observation": {"A": [["UNIQUE_CLAIM_TOKEN", "d1,missing"], null]}}, "qa": [{"question": "UNIQUE_SUMMARY_TOKEN", "evidence": ["d1"]}] diff --git a/crates/cmem-eval-runner/tests/locomo_query_time.rs b/crates/cmem-eval-runner/tests/locomo_query_time.rs new file mode 100644 index 00000000..ea472640 --- /dev/null +++ b/crates/cmem-eval-runner/tests/locomo_query_time.rs @@ -0,0 +1,73 @@ +use cmem_eval::{ObjectType, read_jsonl}; +use serde_json::{Value, json}; +use std::{fs, path::PathBuf, process::Command}; + +fn run_fixture(value: Value) -> (tempfile::TempDir, PathBuf, std::process::Output) { + let root = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../.agent-work/evals-worker2"); + fs::create_dir_all(&root).unwrap(); + let directory = tempfile::tempdir_in(root).unwrap(); + let dataset = directory.path().join("fixture.json"); + let config = directory.path().join("config.toml"); + let output = directory.path().join("run/results.jsonl"); + fs::write(&dataset, serde_json::to_vec(&value).unwrap()).unwrap(); + let mut run_config: cmem_eval::BenchmarkRunConfig = + toml::from_str(include_str!("../../../configs/locomo_retrieval.toml")).unwrap(); + run_config.backend.embedding.provider = cmem_eval::EmbeddingProviderConfig::Deterministic; + run_config.ingest.enrichment_snapshot_path = None; + fs::write(&config, toml::to_string(&run_config).unwrap()).unwrap(); + let result = Command::new(env!("CARGO_BIN_EXE_cmem-eval")) + .env_remove("OPENAI_API_KEY") + .args(["run", "locomo", "--dataset"]) + .arg(dataset) + .arg("--config") + .arg(config) + .arg("--out") + .arg(&output) + .output() + .unwrap(); + (directory, output, result) +} + +#[test] +fn locomo_queries_include_the_final_source_timestamp() { + let (_directory, output, result) = run_fixture(json!([{ + "sample_id": "source_time", + "conversation": { + "session_1_date_time": "2024-02-01T22:00:00+09:00", + "session_1": [{"dia_id": "D1:1", "text": "lighthouse trip"}], + "session_2_date_time": "1:00 pm on 1 January, 2024", + "session_2": [{"dia_id": "D2:1", "text": "earlier lighthouse plans"}] + }, + "qa": [{"question": "lighthouse"}, {"question": "trip"}] + }])); + assert!( + result.status.success(), + "{}", + String::from_utf8_lossy(&result.stderr) + ); + let rows = read_jsonl(&output).unwrap(); + assert_eq!(rows.len(), 2); + for row in rows { + assert!(row.retrieved.iter().any(|item| { + item.kind == ObjectType::Observation && item.external_id.as_deref() == Some("D1:1") + })); + } + assert!(!output.parent().unwrap().join("stores").exists()); +} + +#[test] +fn locomo_without_a_parseable_timestamp_names_the_conversation() { + for timestamp in [Value::Null, json!("not-a-date")] { + let (_directory, output, result) = run_fixture(json!([{ + "sample_id": "missing_source_time", + "conversation": { + "session_1_date_time": timestamp, + "session_1": [{"dia_id": "D1:1", "text": "lighthouse"}] + }, + "qa": [{"question": "lighthouse"}] + }])); + assert!(!result.status.success()); + assert!(String::from_utf8_lossy(&result.stderr).contains("missing_source_time")); + assert!(!output.parent().unwrap().exists()); + } +} diff --git a/crates/cmem-eval-runner/tests/longmemeval_observation_splits.rs b/crates/cmem-eval-runner/tests/longmemeval_observation_splits.rs new file mode 100644 index 00000000..8d181f4b --- /dev/null +++ b/crates/cmem-eval-runner/tests/longmemeval_observation_splits.rs @@ -0,0 +1,93 @@ +use serde_json::json; +use std::{fs, path::PathBuf, process::Command}; + +#[test] +fn run_manifest_records_oversized_turn_split_identities_and_policy() { + let root = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../.agent-work/evals-worker2"); + fs::create_dir_all(&root).unwrap(); + let directory = tempfile::tempdir_in(root).unwrap(); + let dataset = directory.path().join("fixture.json"); + let config = directory.path().join("config.toml"); + let output = directory.path().join("run/results.jsonl"); + let text = format!( + "{}End.\n\n{}End.", + "alpha ".repeat(4500), + "beta ".repeat(4500) + ); + fs::write( + &dataset, + serde_json::to_vec(&json!([{ + "question_id": "manifest-split", + "question": "What happened?", + "question_date": "2023/05/30 (Tue) 23:40", + "haystack_session_ids": ["s1"], + "haystack_dates": ["2023/05/30 (Tue) 23:40"], + "haystack_sessions": [[{"role": "user", "content": text, "has_answer": true}]], + "answer_session_ids": ["s1"] + }])) + .unwrap(), + ) + .unwrap(); + let mut run_config: cmem_eval::BenchmarkRunConfig = toml::from_str(include_str!( + "../../../configs/longmemeval_s_retrieval.toml" + )) + .unwrap(); + run_config.backend.embedding.provider = cmem_eval::EmbeddingProviderConfig::Deterministic; + run_config.ingest.enrichment_snapshot_path = None; + fs::write(&config, toml::to_string(&run_config).unwrap()).unwrap(); + let result = Command::new(env!("CARGO_BIN_EXE_cmem-eval")) + .env_remove("OPENAI_API_KEY") + .args(["run", "longmemeval-s", "--dataset"]) + .arg(dataset) + .arg("--config") + .arg(config) + .arg("--out") + .arg(&output) + .output() + .unwrap(); + assert!( + result.status.success(), + "{}", + String::from_utf8_lossy(&result.stderr) + ); + let header: cmem_eval::RunHeader = + serde_json::from_slice(&fs::read(output.with_file_name("header.json")).unwrap()).unwrap(); + assert_eq!(header.observation_splits.len(), 1); + let split = &header.observation_splits[0]; + assert_eq!(split.dataset_item_id, "manifest-split"); + assert_eq!(split.source_observation_id, "s1:turn:1"); + assert_eq!( + split + .parts + .iter() + .map(|part| part.observation_id.as_str()) + .collect::>(), + ["s1:turn:1", "s1:turn:1:part:2"] + ); + assert!( + split + .parts + .iter() + .all(|part| part.boundary_kind == cmem_eval::ObservationBoundary::Paragraph) + ); + assert_eq!(split.tokenizer, "cl100k_base"); + assert_eq!(split.max_input_tokens, 8192); + assert_eq!( + split.boundary_policy, + "greedy_sentence_paragraph_then_newline_whitespace_v1" + ); + let rows = cmem_eval::read_jsonl(&output).unwrap(); + assert_eq!(rows[0].question_id, "manifest-split"); + assert_eq!(rows[0].gold_observation_ids, ["s1:turn:1"]); + assert!( + rows[0] + .retrieved + .iter() + .any(|item| item.external_id.as_deref() == Some("s1:turn:1:part:2")) + ); + assert_eq!( + serde_json::to_value(&rows[0].metrics).unwrap()["turn_recall_fraction@5"], + 1.0 + ); + assert!(!output.parent().unwrap().join("stores").exists()); +} diff --git a/crates/cmem-eval/src/results.rs b/crates/cmem-eval/src/results.rs index 9778dae3..7eb7a85d 100644 --- a/crates/cmem-eval/src/results.rs +++ b/crates/cmem-eval/src/results.rs @@ -47,6 +47,7 @@ pub struct RunHeader { pub run_id: String, pub dataset: String, pub input_sha256: String, + pub observation_splits: Vec, /// Scenario id (continuity) or dataset id maps to the embedding used at runtime. pub embedding_bindings: BTreeMap, /// Executed native namespace maps to the self notion passed to its stores. @@ -64,6 +65,31 @@ pub struct RunHeader { pub retain_reason: Option, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct ObservationSplit { + pub dataset_item_id: String, + pub source_observation_id: String, + pub parts: Vec, + pub boundary_policy: String, + pub tokenizer: String, + pub max_input_tokens: usize, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct ObservationSplitPart { + pub observation_id: String, + pub boundary_kind: ObservationBoundary, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ObservationBoundary { + Paragraph, + Sentence, + Newline, + Whitespace, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct LatencySummary { pub latency_ms: NumericMetricAggregate, @@ -221,6 +247,7 @@ mod tests { run_id: "r".into(), dataset: "locomo".into(), input_sha256: crate::text_sha256("input"), + observation_splits: Vec::new(), embedding_bindings: BTreeMap::new(), self_notion_ids: BTreeMap::new(), harness_commit: "test".into(),