@@ -0,0 +1,638 @@ +use super::*; + +#[derive(Debug, Clone, Serialize)] +struct ContinuousLoopReport { + benchmark_id: String, + passed: bool, + sessions_observed: usize, + source_ref_count: usize, + reflection_notes: Vec, + stats: MemoryStoreStats, + checks: Vec, + queries: Vec, +} + +#[derive(Debug, Clone, Serialize)] +struct ContinuousLoopNoteTrace { + note_ref: String, + lane: MemoryLane, + path: String, + source_count: usize, + chunk_refs: Vec, +} + +#[derive(Debug, Clone, Serialize)] +struct ContinuousLoopCheck { + name: String, + passed: bool, + details: String, +} + +#[derive(Debug, Clone, Serialize)] +struct ContinuousLoopQueryTrace { + query_id: String, + text: String, + route_reason: String, + candidates: Vec, + packets: Vec, + used_refs: Vec, + omission_reasons: Vec, + expected_terms: Vec, + forbidden_terms: Vec, + passed: bool, +} + +#[derive(Debug, Clone)] +struct ContinuousLoopQueryCase { + query_id: String, + text: String, + expected_terms: Vec, + forbidden_terms: Vec, + expected_omission: Option, + expect_packets: bool, + require_candidates: bool, +} + +pub(super) async fn run_continuous_loop_command(output_dir: &str) -> Result<()> { + let output_dir = PathBuf::from(output_dir); + fs::create_dir_all(&output_dir)?; + + let report = run_continuous_loop_benchmark().await?; + write_continuous_loop_outputs(&output_dir, &report)?; + + if !report.passed { + bail!( + "continuous-loop benchmark failed; see {}", + output_dir.join("report.md").display() + ); + } + + println!( + "wrote continuous-loop benchmark outputs to {}", + output_dir.display() + ); + Ok(()) +} + +async fn run_continuous_loop_benchmark() -> Result { + let tmp = NamedTempFile::new()?; + let store = MemoryStore::open( + tmp.path() + .to_str() + .context("temporary continuous-loop db path was not valid utf-8")?, + 4, + )?; + let mut runtime = ModelsConfig::default(); + runtime.embed_dim = 4; + let llm = LlmClient::new(runtime); + let mut pipeline = MemoryPipeline::new(store, llm, klbr_core::config::MemoryConfig::default()) + .with_profile("klbr-full"); + pipeline.profile.write_episode_notes = true; + pipeline.profile.write_episode_memories = false; + pipeline.profile.lexical = true; + pipeline.profile.dense = false; + pipeline.profile.graph = true; + + let sessions = continuous_loop_sessions(); + let mut writes = Vec::new(); + for session in sessions { + writes.push(pipeline.observe_session(session).await?); + } + + let reflection_notes = write_continuous_loop_reflections(&pipeline, &writes)?; + pipeline.memory.sync_reference_indexes()?; + + let budget = ContextBudget { + max_tokens: 2_000, + top_k: 6, + graph_depth: 1, + }; + let queries = run_continuous_loop_queries(&pipeline, &writes, budget).await?; + let stats = pipeline.memory.stats()?; + let source_ref_count = writes + .iter() + .map(|write| write.source_refs.len()) + .sum::(); + let checks = continuous_loop_checks( + &pipeline, + &writes, + &reflection_notes, + &queries, + &stats, + source_ref_count, + )?; + let passed = checks.iter().all(|check| check.passed); + + Ok(ContinuousLoopReport { + benchmark_id: "continuous-loop-v1".to_string(), + passed, + sessions_observed: writes.len(), + source_ref_count, + reflection_notes, + stats, + checks, + queries, + }) +} + +fn continuous_loop_sessions() -> Vec { + vec![ + BenchSession { + session_id: "voice-session".to_string(), + timestamp: Some(1_000), + turns: vec![ + BenchTurn { + role: "user".to_string(), + content: "project voice note: in klbr docs and comments, keep the lowercase repo voice and avoid officecore phrasing.".to_string(), + timestamp: Some(1_000), + }, + BenchTurn { + role: "assistant".to_string(), + content: "noted; preserve lowercase direct style in klbr work.".to_string(), + timestamp: Some(1_001), + }, + ], + }, + BenchSession { + session_id: "workflow-session".to_string(), + timestamp: Some(2_000), + turns: vec![ + BenchTurn { + role: "user".to_string(), + content: "workflow note: before changing memory retrieval, run rtk cargo test -p klbr-core and the relevant klbr-bench command.".to_string(), + timestamp: Some(2_000), + }, + BenchTurn { + role: "assistant".to_string(), + content: "i will treat that as the retrieval preflight.".to_string(), + timestamp: Some(2_001), + }, + ], + }, + BenchSession { + session_id: "coupon-session".to_string(), + timestamp: Some(3_000), + turns: vec![ + BenchTurn { + role: "user".to_string(), + content: "i redeemed the coffee creamer coupon at Target after checking the grocery budget.".to_string(), + timestamp: Some(3_000), + }, + BenchTurn { + role: "assistant".to_string(), + content: "Target is the store to remember for that coupon.".to_string(), + timestamp: Some(3_001), + }, + ], + }, + BenchSession { + session_id: "dotfiles-klbr-session".to_string(), + timestamp: Some(4_000), + turns: vec![BenchTurn { + role: "user".to_string(), + content: "klbr note: the dotfiles syncer shares the rust cli harness." + .to_string(), + timestamp: Some(4_000), + }], + }, + ] +} + +fn write_continuous_loop_reflections( + pipeline: &MemoryPipeline, + writes: &[WriteTrace], +) -> Result> { + let profile_sources = source_refs_for(writes, "voice-session")?; + let workflow_sources = source_refs_for(writes, "workflow-session")?; + let coupon_sources = source_refs_for(writes, "coupon-session")?; + + Ok(vec![ + upsert_continuous_loop_note( + pipeline, + "nbench_profile_voice", + MemoryLane::Profile, + "profile_note", + "klbr repo voice", + "profile/klbr-repo-voice.md", + "dawn prefers lowercase repo voice for klbr docs and comments. avoid officecore phrasing unless the source text explicitly requires it.", + profile_sources, + &["dawn", "klbr", "repo voice"], + )?, + upsert_continuous_loop_note( + pipeline, + "nbench_proc_retrieval_preflight", + MemoryLane::Procedural, + "procedural_note", + "memory retrieval preflight", + "procedural/memory-retrieval-preflight.md", + "before memory retrieval changes, run rtk cargo test -p klbr-core and the relevant klbr-bench continuous-loop command.", + workflow_sources, + &["klbr", "memory retrieval", "preflight"], + )?, + upsert_continuous_loop_note( + pipeline, + "nbench_semantic_coupon_target", + MemoryLane::Semantic, + "semantic_note", + "coupon store fact", + "semantic/coupon-store-fact.md", + "the coffee creamer coupon was redeemed at Target; this fact came from the coupon session and should stay source-grounded.", + coupon_sources, + &["coupon", "target", "grocery"], + )?, + ]) +} + +#[allow(clippy::too_many_arguments)] +fn upsert_continuous_loop_note( + pipeline: &MemoryPipeline, + note_ref: &str, + lane: MemoryLane, + kind: &str, + title: &str, + path: &str, + body: &str, + sources: Vec, + entities: &[&str], +) -> Result { + let entities = entities + .iter() + .map(|entity| entity.to_string()) + .collect::>(); + let record = pipeline.memory.upsert_markdown_note(&MarkdownNoteInput { + note_ref: Some(note_ref.to_string()), + lane, + kind: kind.to_string(), + title: title.to_string(), + path: Some(path.to_string()), + body: body.to_string(), + sources: sources.clone(), + follow: None, + entities: entities.clone(), + status: "active".to_string(), + frontmatter: serde_json::json!({ + "benchmark": "continuous-loop", + "entities": entities, + "source_refs": sources.clone(), + }), + })?; + + Ok(ContinuousLoopNoteTrace { + note_ref: record.note_ref, + lane, + path: record.path, + source_count: sources.len(), + chunk_refs: record.chunk_refs, + }) +} + +async fn run_continuous_loop_queries( + pipeline: &MemoryPipeline, + writes: &[WriteTrace], + budget: ContextBudget, +) -> Result> { + let coupon_ref = source_refs_for(writes, "coupon-session")? + .first() + .cloned() + .context("coupon session did not produce a source ref")?; + let cases = vec![ + ContinuousLoopQueryCase { + query_id: "profile-voice".to_string(), + text: "what style do i prefer for klbr docs and comments?".to_string(), + expected_terms: vec![ + "lowercase repo voice".to_string(), + "officecore phrasing".to_string(), + ], + forbidden_terms: Vec::new(), + expected_omission: None, + expect_packets: true, + require_candidates: true, + }, + ContinuousLoopQueryCase { + query_id: "procedural-preflight".to_string(), + text: "what workflow should we always run for memory retrieval changes?".to_string(), + expected_terms: vec![ + "rtk cargo test -p klbr-core".to_string(), + "klbr-bench continuous-loop".to_string(), + ], + forbidden_terms: Vec::new(), + expected_omission: None, + expect_packets: true, + require_candidates: true, + }, + ContinuousLoopQueryCase { + query_id: "episode-coupon".to_string(), + text: "where did i redeem the coffee creamer coupon last time?".to_string(), + expected_terms: vec!["coffee creamer coupon".to_string(), "Target".to_string()], + forbidden_terms: Vec::new(), + expected_omission: None, + expect_packets: true, + require_candidates: true, + }, + ContinuousLoopQueryCase { + query_id: "exact-coupon-ref".to_string(), + text: format!("what store is mentioned in [{coupon_ref}]?"), + expected_terms: vec!["Target".to_string()], + forbidden_terms: Vec::new(), + expected_omission: None, + expect_packets: true, + require_candidates: true, + }, + ContinuousLoopQueryCase { + query_id: "project-syncer".to_string(), + text: "dotfiles syncer rust cli harness".to_string(), + expected_terms: vec!["rust cli harness".to_string()], + forbidden_terms: Vec::new(), + expected_omission: None, + expect_packets: true, + require_candidates: true, + }, + ]; + + let mut traces = Vec::new(); + for case in cases { + traces.push(run_continuous_loop_query(pipeline, case, budget).await?); + } + Ok(traces) +} + +async fn run_continuous_loop_query( + pipeline: &MemoryPipeline, + case: ContinuousLoopQueryCase, + budget: ContextBudget, +) -> Result { + let query = BenchQuery { + query_id: case.query_id.clone(), + text: case.text.clone(), + reference_time: Some(5_000), + op_hint: None, + }; + let retrieval = pipeline.retrieve_evidence(query.clone(), budget).await?; + let context = pipeline + .assemble_context(query.clone(), &retrieval, budget) + .await?; + let content_lower = context.content.to_lowercase(); + let expected_ok = case + .expected_terms + .iter() + .all(|term| content_lower.contains(&term.to_lowercase())); + let forbidden_ok = case + .forbidden_terms + .iter() + .all(|term| !content_lower.contains(&term.to_lowercase())); + let omission_reasons = retrieval + .packet_omissions + .iter() + .map(|omission| omission.reason.clone()) + .collect::>(); + let omission_ok = case + .expected_omission + .as_ref() + .map(|expected| omission_reasons.iter().any(|reason| reason == expected)) + .unwrap_or(true); + let packets_ok = if case.expect_packets { + !retrieval.packets.is_empty() + } else { + retrieval.packets.is_empty() && context.used_refs.is_empty() + }; + let candidates_ok = !case.require_candidates || !retrieval.candidates.is_empty(); + let passed = expected_ok && forbidden_ok && omission_ok && packets_ok && candidates_ok; + + Ok(ContinuousLoopQueryTrace { + query_id: case.query_id, + text: case.text, + route_reason: retrieval.route.reason, + candidates: retrieval + .candidates + .iter() + .take(12) + .map(|candidate| format!("{}:{}", candidate.source.as_str(), candidate.ref_id)) + .collect(), + packets: retrieval + .packets + .iter() + .map(|packet| { + format!( + "{}:{}:{}", + packet.packet_kind.as_str(), + packet.anchor_ref, + packet.refs.join(",") + ) + }) + .collect(), + used_refs: context.used_refs, + omission_reasons, + expected_terms: case.expected_terms, + forbidden_terms: case.forbidden_terms, + passed, + }) +} + +fn continuous_loop_checks( + pipeline: &MemoryPipeline, + writes: &[WriteTrace], + reflection_notes: &[ContinuousLoopNoteTrace], + queries: &[ContinuousLoopQueryTrace], + stats: &MemoryStoreStats, + source_ref_count: usize, +) -> Result> { + let notes_resolved = reflection_notes + .iter() + .map(|note| { + pipeline + .memory + .get_resolved_ref(¬e.note_ref) + .map(|resolved| resolved.and_then(|data| data.body).is_some()) + }) + .collect::>>()?; + let query_pass_count = queries.iter().filter(|query| query.passed).count(); + let project_ok = queries + .iter() + .find(|query| query.query_id == "project-syncer") + .is_some_and(|query| query.passed); + let exact_ok = queries + .iter() + .find(|query| query.query_id == "exact-coupon-ref") + .is_some_and(|query| query.passed); + + Ok(vec![ + ContinuousLoopCheck { + name: "sessions_observed".to_string(), + passed: writes.len() == 4 && writes.iter().all(|write| !write.source_refs.is_empty()), + details: format!( + "{} sessions, {} source refs", + writes.len(), + source_ref_count + ), + }, + ContinuousLoopCheck { + name: "reflection_notes_resolved".to_string(), + passed: reflection_notes.len() == 3 && notes_resolved.iter().all(|resolved| *resolved), + details: format!( + "{} reflection notes, {} resolved with promptable bodies", + reflection_notes.len(), + notes_resolved.iter().filter(|resolved| **resolved).count() + ), + }, + ContinuousLoopCheck { + name: "markdown_sync".to_string(), + passed: stats.markdown_notes >= 7 && stats.fts_rows == stats.promptable_refs, + details: format!( + "{} markdown notes, {} promptable refs, {} fts rows", + stats.markdown_notes, stats.promptable_refs, stats.fts_rows + ), + }, + ContinuousLoopCheck { + name: "source_edges".to_string(), + passed: stats.active_edges >= source_ref_count as i64, + details: format!( + "{} active edges for {} source refs", + stats.active_edges, source_ref_count + ), + }, + ContinuousLoopCheck { + name: "query_alignment".to_string(), + passed: query_pass_count == queries.len(), + details: format!("{} / {} queries passed", query_pass_count, queries.len()), + }, + ContinuousLoopCheck { + name: "project_passive_recall".to_string(), + passed: project_ok, + details: "project query retrieves the matching packet without query-word suppression" + .to_string(), + }, + ContinuousLoopCheck { + name: "exact_ref_resolution".to_string(), + passed: exact_ok, + details: "turn chunk inline ref resolves through packet assembly".to_string(), + }, + ]) +} + +fn source_refs_for(writes: &[WriteTrace], session_id: &str) -> Result> { + writes + .iter() + .find(|write| write.session_id == session_id) + .map(|write| write.source_refs.clone()) + .with_context(|| format!("missing write trace for session {session_id}")) +} + +fn write_continuous_loop_outputs(output_dir: &Path, report: &ContinuousLoopReport) -> Result<()> { + fs::write( + output_dir.join("report.json"), + serde_json::to_vec_pretty(report)?, + )?; + fs::write( + output_dir.join("report.md"), + render_continuous_loop_report(report), + )?; + Ok(()) +} + +fn render_continuous_loop_report(report: &ContinuousLoopReport) -> String { + let checks = report + .checks + .iter() + .map(|check| { + format!( + "| `{}` | `{}` | {} |", + check.name, + yes_no(check.passed), + md_cell(&check.details) + ) + }) + .collect::>() + .join("\n"); + let queries = report + .queries + .iter() + .map(|query| { + format!( + "| `{}` | `{}` | `{}` | `{}` | `{}` | `{}` |", + query.query_id, + yes_no(query.passed), + query.route_reason, + query.candidates.len(), + query.packets.len(), + md_cell(&query.omission_reasons.join(", ")) + ) + }) + .collect::>() + .join("\n"); + + format!( + "# continuous-loop memory benchmark\n\n\ +benchmark: `{}`\n\ +passed: `{}`\n\ +sessions observed: `{}`\n\ +source refs: `{}`\n\ +reflection notes: `{}`\n\n\ +## store stats\n\n\ +- refs: `{}` active `{}`\n\ +- edges: `{}` active `{}`\n\ +- markdown notes: `{}`\n\ +- promptable refs / fts rows: `{}` / `{}`\n\ +- db bytes: `{}`\n\n\ +## checks\n\n\ +| check | passed | details |\n\ +|---|---|---|\n\ +{}\n\n\ +## queries\n\n\ +| query | passed | route | candidates | packets | omissions |\n\ +|---|---|---|---:|---:|---|\n\ +{}\n", + report.benchmark_id, + yes_no(report.passed), + report.sessions_observed, + report.source_ref_count, + report.reflection_notes.len(), + report.stats.refs, + report.stats.active_refs, + report.stats.edges, + report.stats.active_edges, + report.stats.markdown_notes, + report.stats.promptable_refs, + report.stats.fts_rows, + report.stats.db_bytes, + checks, + queries + ) +} + +fn yes_no(value: bool) -> &'static str { + if value { + "yes" + } else { + "no" + } +} + +fn md_cell(value: &str) -> String { + if value.is_empty() { + "(none)".to_string() + } else { + value.replace('|', "\\|").replace('\n', "
") + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn continuous_loop_benchmark_smoke_passes() { + let report = run_continuous_loop_benchmark() + .await + .expect("continuous-loop benchmark should run"); + + assert!(report.passed, "{:#?}", report.checks); + assert!(report + .queries + .iter() + .any(|query| query.query_id == "project-syncer" && query.passed)); + assert!(report + .queries + .iter() + .any(|query| query.query_id == "exact-coupon-ref" && query.passed)); + } +}