diff --git a/src/automation/artifact_payloads.rs b/src/automation/artifact_payloads.rs index 14dc6ce7..590a32f1 100644 --- a/src/automation/artifact_payloads.rs +++ b/src/automation/artifact_payloads.rs @@ -71,6 +71,7 @@ pub(super) fn traces_payload(ctx: &ArtifactPayloadContext<'_>) -> Value { "response_schema": ctx.request.contract.response_schema, "strict_json": ctx.request.contract.strict_json, "evidence_hash": ctx.record.evidence_hash, + "evidence_mode": context_evidence_mode(&ctx.request.context), "input_hash": ctx.record.input_hash, "output_hash": ctx.record.output_hash, "context_keys": ctx.request.context.as_object() @@ -79,6 +80,22 @@ pub(super) fn traces_payload(ctx: &ArtifactPayloadContext<'_>) -> Value { }) } +/// Extracts the `evidence_mode` label from whichever evidence object the task +/// placed in the request context (e.g. `session_reflection_evidence` or +/// `skill_writer_evidence`), so traces distinguish session-replay-backed runs +/// from grep-only runs. Null for tasks without a mode label. +fn context_evidence_mode(context: &Value) -> Value { + context + .as_object() + .and_then(|object| { + object + .values() + .find_map(|value| value.get("evidence_mode").filter(|mode| mode.is_string())) + }) + .cloned() + .unwrap_or(Value::Null) +} + pub(super) fn feedback_payload(ctx: &ArtifactPayloadContext<'_>, trace_ref: &Value) -> Value { json!({ "schema_version": 1, diff --git a/src/automation/runner.rs b/src/automation/runner.rs index e6d755c1..4ee834f5 100644 --- a/src/automation/runner.rs +++ b/src/automation/runner.rs @@ -34,7 +34,9 @@ use crate::global_db::GlobalDb; use crate::sessions::cursor::{ resolve_hermes_profile_session_db_readonly, HermesProfileDbReadOnly, }; -use crate::sessions::lcm::{LcmGrepRequest, LcmGrepSort, LcmScope}; +use crate::sessions::lcm::{ + LcmGrepRequest, LcmGrepSort, LcmScope, LcmSessionReplayRequest, LcmSessionReplaySlice, +}; use crate::tracedecay::{current_timestamp, TraceDecay}; pub use super::memory_curator::{ @@ -43,6 +45,14 @@ pub use super::memory_curator::{ const SKILL_ANALYTICS_IMPORT_LIMIT: usize = 2_000; +/// Bounds for the session-replay evidence channel. Worst case per session is +/// `(4 + 4) * 500 + 3 * 700 = 6_100` snippet chars, so the default three +/// sessions stay under ~5k tokens alongside the grep hits. +const SESSION_REPLAY_HEAD_TURNS: usize = 4; +const SESSION_REPLAY_TAIL_TURNS: usize = 4; +const SESSION_REPLAY_SNIPPET_CHARS: usize = 500; +const SESSION_REPLAY_SUMMARY_NODES: usize = 3; +const SESSION_REPLAY_SUMMARY_CHARS: usize = 700; #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct SessionReflectorAutomationOptions { #[serde(default)] @@ -65,6 +75,14 @@ pub struct SessionReflectorAutomationOptions { pub include_summaries: bool, #[serde(default = "default_session_evidence_limit")] pub evidence_limit: usize, + /// When true, include bounded turn-ordered slices of recently active + /// sessions as a primary evidence channel alongside the keyword grep. + #[serde(default = "default_include_recent_sessions")] + pub include_recent_sessions: bool, + /// How many recently active sessions to replay when `session_id` is not + /// explicitly set. + #[serde(default = "default_recent_sessions_limit")] + pub recent_sessions_limit: usize, #[serde(default = "default_lcm_grep_sort")] pub sort: LcmGrepSort, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -90,6 +108,8 @@ impl Default for SessionReflectorAutomationOptions { session_id: None, include_summaries: default_include_summaries(), evidence_limit: default_session_evidence_limit(), + include_recent_sessions: default_include_recent_sessions(), + recent_sessions_limit: default_recent_sessions_limit(), sort: default_lcm_grep_sort(), source: None, role: None, @@ -124,6 +144,13 @@ pub struct SkillWriterAutomationOptions { pub query: String, #[serde(default = "default_skill_writer_evidence_limit")] pub evidence_limit: usize, + /// When true, include bounded turn-ordered slices of recently active + /// sessions as a primary evidence channel alongside the keyword grep. + #[serde(default = "default_include_recent_sessions")] + pub include_recent_sessions: bool, + /// How many recently active sessions to replay. + #[serde(default = "default_recent_sessions_limit")] + pub recent_sessions_limit: usize, #[serde(default, skip_serializing_if = "Option::is_none")] pub profile_root: Option, } @@ -138,6 +165,8 @@ impl Default for SkillWriterAutomationOptions { provider: default_skill_writer_provider(), query: default_skill_writer_query(), evidence_limit: default_skill_writer_evidence_limit(), + include_recent_sessions: default_include_recent_sessions(), + recent_sessions_limit: default_recent_sessions_limit(), profile_root: None, } } @@ -684,7 +713,28 @@ async fn build_session_reflector_evidence( .map_err(|e| TraceDecayError::Config { message: format!("failed to build session reflection evidence: {e}"), })?; + let recent_session_slices = if options.include_recent_sessions + && session_reflector_replay_allowed( + options.scope, + session_id.as_deref(), + source.as_deref(), + role.as_deref(), + options.start_time, + options.end_time, + ) { + recent_session_replay_evidence( + &lcm_db, + &provider, + session_id.as_deref(), + options.recent_sessions_limit, + "session_reflector", + ) + .await? + } else { + None + }; let evidence = json!({ + "evidence_mode": evidence_mode_label(recent_session_slices.is_some()), "storage_scope": storage_scope, "hermes_home": options.hermes_home.as_ref().map(|path| path.display().to_string()), "provider": provider, @@ -697,14 +747,19 @@ async fn build_session_reflector_evidence( "role": role, "start_time": options.start_time, "end_time": options.end_time, + "recent_session_slices": recent_session_slices, "hits": hits, }); let evidence_hash = Some(sha256_json(&evidence)); - if evidence + let has_grep_hits = evidence .get("hits") .and_then(Value::as_array) - .is_none_or(Vec::is_empty) - { + .is_some_and(|hits| !hits.is_empty()); + let has_replay_sessions = evidence + .pointer("/recent_session_slices/sessions") + .and_then(Value::as_array) + .is_some_and(|sessions| !sessions.is_empty()); + if !has_grep_hits && !has_replay_sessions { return Ok(SessionReflectorEvidenceOutcome::Skipped { reason: "no_session_evidence", evidence_hash, @@ -772,6 +827,18 @@ async fn build_skill_writer_evidence( .map_err(|e| TraceDecayError::Config { message: format!("failed to build skill writer evidence: {e}"), })?; + let recent_session_slices = if options.include_recent_sessions { + recent_session_replay_evidence( + &lcm_db, + &provider, + None, + options.recent_sessions_limit, + "skill_writer", + ) + .await? + } else { + None + }; let existing_skills = list_managed_skills(&profile_root).await?; let global_db = GlobalDb::open().await; ingest_project_analytics_events( @@ -805,10 +872,12 @@ async fn build_skill_writer_evidence( &underused_tool_families, ); let evidence = json!({ + "evidence_mode": evidence_mode_label(recent_session_slices.is_some()), "storage_scope": storage_scope, "hermes_home": options.hermes_home.as_ref().map(|path| path.display().to_string()), "provider": provider, "query": query, + "recent_session_slices": recent_session_slices, "hits": hits, "skill_usage_summaries": skill_usage_summaries, "stale_recommendations": stale_recommendations, @@ -834,11 +903,15 @@ async fn build_skill_writer_evidence( .collect::>(), }); let evidence_hash = Some(sha256_json(&evidence)); - if evidence + let has_grep_hits = evidence .get("hits") .and_then(Value::as_array) - .is_none_or(Vec::is_empty) - { + .is_some_and(|hits| !hits.is_empty()); + let has_replay_sessions = evidence + .pointer("/recent_session_slices/sessions") + .and_then(Value::as_array) + .is_some_and(|sessions| !sessions.is_empty()); + if !has_grep_hits && !has_replay_sessions { return Ok(SkillWriterEvidenceOutcome::Skipped { reason: "no_skill_writer_evidence", evidence_hash, @@ -1236,6 +1309,7 @@ fn build_combined_review_prompt(reflector_evidence: &Value, skill_evidence: &Val fn build_session_reflector_prompt(evidence: &Value) -> String { const POLICY: &str = concat!( "Review these bounded TraceDecay session snippets and propose only durable memory facts.\n", + "Evidence has two channels: recent_session_slices holds turn-ordered head/tail turns and summary nodes replayed from recently active sessions, and hits holds keyword search matches; both are citable.\n", "\n", "Signals worth capturing (any one is enough):\n", "- The user revealed durable preferences, persona, expectations, or ways they want the agent to operate.\n", @@ -1262,6 +1336,7 @@ fn build_session_reflector_prompt(evidence: &Value) -> String { fn build_skill_writer_prompt(evidence: &Value) -> String { const POLICY: &str = concat!( "Review these bounded TraceDecay session snippets and propose only reusable managed skills for repeated workflows, corrections, or tool-use patterns.\n", + "Evidence has two channels: recent_session_slices holds turn-ordered head/tail turns and summary nodes replayed from recently active sessions, and hits holds keyword search matches.\n", "\n", "Target shape of the skill library: CLASS-LEVEL umbrella skills, each with a rich body and support files for session-specific detail — not a long flat list of narrow one-session-one-skill entries. This shapes HOW you update, not WHETHER you update.\n", "\n", @@ -1302,6 +1377,121 @@ fn normalized_non_empty(value: &str) -> Option { } } +struct ReplaySessionTarget { + provider: String, + session_id: String, +} + +/// Builds the "recent completed sessions" replay evidence channel: bounded +/// turn-ordered head/tail slices plus top summary-DAG nodes for the last N +/// recently active sessions (or one explicitly requested session). +/// +/// Returns `None` when no session has any raw messages, so callers can fall +/// back to grep-only evidence. +async fn recent_session_replay_evidence( + lcm_db: &GlobalDb, + provider: &str, + explicit_session_id: Option<&str>, + sessions_limit: usize, + task_name: &str, +) -> Result> { + let sessions_limit = sessions_limit.clamp(1, 10); + let provider_filter = (provider != "all").then_some(provider); + let replay_error = |e: crate::sessions::lcm::LcmError| TraceDecayError::Config { + message: format!("failed to build {task_name} session replay evidence: {e}"), + }; + let (session_selection, targets) = if let Some(session_id) = explicit_session_id { + let providers = match provider_filter { + Some(provider) => vec![provider.to_string()], + None => lcm_db + .lcm_session_providers(session_id) + .await + .map_err(replay_error)?, + }; + let targets: Vec = providers + .into_iter() + .map(|provider| ReplaySessionTarget { + provider, + session_id: session_id.to_string(), + }) + .collect(); + ("explicit_session_id", targets) + } else { + let targets: Vec = lcm_db + .lcm_recent_sessions(provider_filter, sessions_limit) + .await + .map_err(replay_error)? + .into_iter() + .map(|session| ReplaySessionTarget { + provider: session.provider, + session_id: session.session_id, + }) + .collect(); + ("recent_activity", targets) + }; + + let mut sessions: Vec = Vec::new(); + for target in targets { + let slice = lcm_db + .lcm_session_replay_slice(&LcmSessionReplayRequest { + provider: target.provider, + session_id: target.session_id, + head_limit: SESSION_REPLAY_HEAD_TURNS, + tail_limit: SESSION_REPLAY_TAIL_TURNS, + max_snippet_chars: SESSION_REPLAY_SNIPPET_CHARS, + summary_limit: SESSION_REPLAY_SUMMARY_NODES, + max_summary_chars: SESSION_REPLAY_SUMMARY_CHARS, + }) + .await + .map_err(replay_error)?; + if slice.total_messages == 0 && slice.summary_nodes.is_empty() { + continue; + } + sessions.push(slice); + } + if sessions.is_empty() { + return Ok(None); + } + Ok(Some(json!({ + "mode": "recent_sessions", + "session_selection": session_selection, + "sessions_limit": sessions_limit, + "bounds": { + "head_turns": SESSION_REPLAY_HEAD_TURNS, + "tail_turns": SESSION_REPLAY_TAIL_TURNS, + "snippet_chars": SESSION_REPLAY_SNIPPET_CHARS, + "summary_nodes": SESSION_REPLAY_SUMMARY_NODES, + "summary_chars": SESSION_REPLAY_SUMMARY_CHARS, + }, + "sessions": sessions, + }))) +} + +/// Names the evidence channels actually present so run artifacts can +/// distinguish replay-backed runs from grep-only runs. +fn evidence_mode_label(has_replay: bool) -> &'static str { + if has_replay { + "session_replay_with_grep" + } else { + "grep_only" + } +} + +fn session_reflector_replay_allowed( + scope: LcmScope, + session_id: Option<&str>, + source: Option<&str>, + role: Option<&str>, + start_time: Option, + end_time: Option, +) -> bool { + if source.is_some() || role.is_some() || start_time.is_some() || end_time.is_some() { + return false; + } + + matches!(scope, LcmScope::All) || session_id.is_some() +} + /// Resolves the LCM sessions database for an automation task, reporting /// `NotIngested` when the store does not exist yet so callers can skip /// without re-checking the path. @@ -1372,6 +1562,14 @@ fn default_session_evidence_limit() -> usize { 20 } +fn default_include_recent_sessions() -> bool { + true +} + +fn default_recent_sessions_limit() -> usize { + 3 +} + fn default_skill_writer_query() -> String { "workflow correction repeated skill tool pattern".to_string() } diff --git a/src/automation/session_reflector.rs b/src/automation/session_reflector.rs index e6fc0374..6c7e5365 100644 --- a/src/automation/session_reflector.rs +++ b/src/automation/session_reflector.rs @@ -71,9 +71,52 @@ impl EvidenceCitationSet { } } } + citations.collect_replay_slices(evidence); citations } + /// Registers the session-replay evidence channel (`recent_session_slices`) + /// so facts can cite turns and summary nodes surfaced by replay even when + /// they did not match the keyword grep. + fn collect_replay_slices(&mut self, evidence: &Value) { + for session in evidence + .pointer("/recent_session_slices/sessions") + .and_then(Value::as_array) + .into_iter() + .flatten() + { + let Some(session_id) = session.get("session_id").and_then(Value::as_str) else { + continue; + }; + for slice_key in ["head", "tail"] { + for message in session + .get(slice_key) + .and_then(Value::as_array) + .into_iter() + .flatten() + { + if let Some(message_id) = message.get("message_id").and_then(Value::as_str) { + self.raw_messages + .insert((session_id.to_string(), message_id.to_string())); + } + if let Some(store_id) = message.get("store_id").and_then(value_as_i64) { + self.raw_store_ids.insert(store_id); + } + } + } + for node in session + .get("summary_nodes") + .and_then(Value::as_array) + .into_iter() + .flatten() + { + if let Some(node_id) = node.get("node_id").and_then(Value::as_str) { + self.summary_nodes.insert(node_id.to_string()); + } + } + } + } + fn contains(&self, source_span: &Value) -> bool { let Some(span) = source_span.as_object() else { return false; diff --git a/src/automation_cli.rs b/src/automation_cli.rs index eedf45da..a7136f23 100644 --- a/src/automation_cli.rs +++ b/src/automation_cli.rs @@ -527,6 +527,7 @@ async fn handle_automation_run_command( role, start_time, end_time, + ..SessionReflectorAutomationOptions::default() }, ) .await?; @@ -573,6 +574,7 @@ async fn handle_automation_run_command( query, evidence_limit, profile_root: None, + ..SkillWriterAutomationOptions::default() }, ) .await?; diff --git a/src/global_db.rs b/src/global_db.rs index 57216ae8..187adef5 100644 --- a/src/global_db.rs +++ b/src/global_db.rs @@ -2714,6 +2714,35 @@ impl GlobalDb { crate::sessions::lcm::query::load_session(&self.conn, request).await } + /// Lists sessions in the raw LCM store ordered by most recent activity. + /// + /// `provider = None` spans all providers. Used to select "recently + /// active" sessions for automation session-replay evidence. + pub async fn lcm_recent_sessions( + &self, + provider: Option<&str>, + limit: usize, + ) -> Result, crate::sessions::lcm::LcmError> { + crate::sessions::lcm::query::recent_sessions(&self.conn, provider, limit).await + } + + /// Lists providers that have raw LCM messages for an explicit session id. + pub async fn lcm_session_providers( + &self, + session_id: &str, + ) -> Result, crate::sessions::lcm::LcmError> { + crate::sessions::lcm::query::session_providers(&self.conn, session_id).await + } + + /// Loads a bounded turn-ordered replay slice (head/tail turns plus top + /// summary-DAG nodes) for one session. + pub async fn lcm_session_replay_slice( + &self, + request: &crate::sessions::lcm::LcmSessionReplayRequest, + ) -> Result { + crate::sessions::lcm::query::session_replay_slice(&self.conn, request).await + } + /// Searches bounded LCM raw snippets and, optionally, summary node text. pub async fn lcm_grep( &self, diff --git a/src/sessions/lcm/mod.rs b/src/sessions/lcm/mod.rs index fdf7b077..5924896a 100644 --- a/src/sessions/lcm/mod.rs +++ b/src/sessions/lcm/mod.rs @@ -30,10 +30,11 @@ pub use types::{ LcmGrepSort, LcmLifecycleState, LcmLifecycleUpdate, LcmLoadSessionMessage, LcmLoadSessionPage, LcmLoadSessionRequest, LcmMaintenanceDebt, LcmPayloadExpansion, LcmPayloadGcStatus, LcmPayloadRef, LcmPreflightRequest, LcmPreflightResponse, LcmRawMessage, LcmRawMessageOverview, - LcmScope, LcmSessionBoundaryRequest, LcmSessionBoundaryResponse, LcmSourceRef, LcmStatus, - LcmStorageKind, LcmStoreStatus, LcmSummaryExpansion, LcmSummaryNode, LcmSummaryNodeDraft, - LcmSummaryNodeOverview, LcmSummaryRequest, LcmSummarySourceMessage, LcmSummarySourceRange, - DERIVED_TRUNCATION_MARKER, LCM_COMPRESSION_BOUNDARY_COOLDOWN_SECONDS, + LcmRecentSession, LcmReplayMessage, LcmReplaySummaryNode, LcmScope, LcmSessionBoundaryRequest, + LcmSessionBoundaryResponse, LcmSessionReplayRequest, LcmSessionReplaySlice, LcmSourceRef, + LcmStatus, LcmStorageKind, LcmStoreStatus, LcmSummaryExpansion, LcmSummaryNode, + LcmSummaryNodeDraft, LcmSummaryNodeOverview, LcmSummaryRequest, LcmSummarySourceMessage, + LcmSummarySourceRange, DERIVED_TRUNCATION_MARKER, LCM_COMPRESSION_BOUNDARY_COOLDOWN_SECONDS, LCM_DEFAULT_FRESH_TAIL_COUNT, LCM_DEFAULT_SUMMARY_FAN_IN, MAX_DERIVED_SNIPPET_CHARS, MAX_DERIVED_TEXT_CHARS, }; diff --git a/src/sessions/lcm/query.rs b/src/sessions/lcm/query.rs index bd3ce0e7..169773d1 100644 --- a/src/sessions/lcm/query.rs +++ b/src/sessions/lcm/query.rs @@ -16,8 +16,9 @@ use super::{ LcmExpandQuerySynthesisPrompt, LcmExpandRequest, LcmExpandResponse, LcmExpandSourcePagination, LcmExpandTarget, LcmExpandedSummarySource, LcmGcConfig, LcmGrepHit, LcmGrepRequest, LcmGrepSort, LcmLoadSessionMessage, LcmLoadSessionPage, LcmLoadSessionRequest, LcmRawMessage, - LcmRawMessageOverview, LcmScope, LcmSourceRef, LcmStatus, LcmStorageKind, LcmStoreStatus, - LcmSummaryExpansion, LcmSummaryNode, LcmSummaryNodeOverview, + LcmRawMessageOverview, LcmRecentSession, LcmReplayMessage, LcmReplaySummaryNode, LcmScope, + LcmSessionReplayRequest, LcmSessionReplaySlice, LcmSourceRef, LcmStatus, LcmStorageKind, + LcmStoreStatus, LcmSummaryExpansion, LcmSummaryNode, LcmSummaryNodeOverview, LCM_COMPRESSION_BOUNDARY_COOLDOWN_SECONDS, LCM_DEFAULT_FRESH_TAIL_COUNT, LCM_DEFAULT_SUMMARY_FAN_IN, LCM_EXPAND_QUERY_SYNTHESIS_SYSTEM_PROMPT, LCM_SCHEMA_VERSION, }; @@ -156,6 +157,204 @@ pub(crate) async fn load_session( }) } +/// Lists sessions in the raw LCM store ordered by most recent activity +/// (latest message timestamp, falling back to insertion order). +pub(crate) async fn recent_sessions( + conn: &Connection, + provider: Option<&str>, + limit: usize, +) -> Result, LcmError> { + let limit = clamp_limit(limit); + let mut values = Vec::new(); + let provider_clause = match provider { + Some(provider) => { + values.push(Value::Text(provider.to_string())); + "WHERE provider = ?" + } + None => "", + }; + values.push(Value::Integer(limit as i64)); + let sql = format!( + "SELECT provider, session_id, COUNT(*), MIN(timestamp), MAX(timestamp), MAX(store_id) + FROM lcm_raw_messages + {provider_clause} + GROUP BY provider, session_id + ORDER BY COALESCE(MAX(timestamp), MAX(store_id)) DESC, MAX(store_id) DESC + LIMIT ?" + ); + let mut rows = conn.query(&sql, values).await?; + let mut sessions = Vec::new(); + while let Some(row) = rows.next().await? { + sessions.push(LcmRecentSession { + provider: row.get(0)?, + session_id: row.get(1)?, + message_count: row.get(2)?, + first_timestamp: row.get(3)?, + last_timestamp: row.get(4)?, + last_store_id: row.get(5)?, + }); + } + Ok(sessions) +} + +/// Lists providers that contain raw messages for an explicit session id, +/// ordered by most recent activity. +pub(crate) async fn session_providers( + conn: &Connection, + session_id: &str, +) -> Result, LcmError> { + let mut rows = conn + .query( + "SELECT provider + FROM lcm_raw_messages + WHERE session_id = ?1 + GROUP BY provider + ORDER BY COALESCE(MAX(timestamp), MAX(store_id)) DESC, MAX(store_id) DESC", + params![session_id], + ) + .await?; + let mut providers = Vec::new(); + while let Some(row) = rows.next().await? { + providers.push(row.get(0)?); + } + Ok(providers) +} + +/// Loads a bounded turn-ordered replay slice for one session: head turns, +/// tail turns (deduplicated against the head), and top summary-DAG nodes. +pub(crate) async fn session_replay_slice( + conn: &Connection, + request: &LcmSessionReplayRequest, +) -> Result { + let mut rows = conn + .query( + "SELECT COUNT(*) FROM lcm_raw_messages WHERE provider = ?1 AND session_id = ?2", + params![request.provider.as_str(), request.session_id.as_str()], + ) + .await?; + let total_messages: i64 = match rows.next().await? { + Some(row) => row.get(0)?, + None => 0, + }; + + let head = replay_slice_messages(conn, request, ReplayDirection::Head, None).await?; + let last_head_store_id = head.last().map(|message| message.store_id); + let mut tail = + replay_slice_messages(conn, request, ReplayDirection::Tail, last_head_store_id).await?; + tail.reverse(); + let included = (head.len() + tail.len()) as i64; + let summary_nodes = replay_slice_summary_nodes(conn, request).await?; + + Ok(LcmSessionReplaySlice { + provider: request.provider.clone(), + session_id: request.session_id.clone(), + total_messages, + omitted_messages: (total_messages - included).max(0), + head, + tail, + summary_nodes, + }) +} + +#[derive(Clone, Copy)] +enum ReplayDirection { + Head, + Tail, +} + +async fn replay_slice_messages( + conn: &Connection, + request: &LcmSessionReplayRequest, + direction: ReplayDirection, + after_store_id: Option, +) -> Result, LcmError> { + let limit = match direction { + ReplayDirection::Head => request.head_limit, + ReplayDirection::Tail => request.tail_limit, + }; + if limit == 0 { + return Ok(Vec::new()); + } + let order = match direction { + ReplayDirection::Head => "ASC", + ReplayDirection::Tail => "DESC", + }; + let values = vec![ + Value::Text(request.provider.clone()), + Value::Text(request.session_id.clone()), + Value::Integer(after_store_id.unwrap_or(0)), + Value::Integer(clamp_limit(limit) as i64), + ]; + let sql = format!( + "SELECT message_id, store_id, role, ordinal, timestamp, snippet_text + FROM lcm_raw_messages + WHERE provider = ? AND session_id = ? AND store_id > ? + ORDER BY store_id {order} + LIMIT ?" + ); + let mut rows = conn.query(&sql, values).await?; + let mut messages = Vec::new(); + while let Some(row) = rows.next().await? { + let snippet_text: String = row.get(5)?; + let (snippet, truncated) = bounded_replay_snippet(&snippet_text, request.max_snippet_chars); + messages.push(LcmReplayMessage { + message_id: row.get(0)?, + store_id: row.get(1)?, + role: row.get(2)?, + ordinal: row.get(3)?, + timestamp: row.get(4)?, + snippet, + truncated, + }); + } + Ok(messages) +} + +async fn replay_slice_summary_nodes( + conn: &Connection, + request: &LcmSessionReplayRequest, +) -> Result, LcmError> { + if request.summary_limit == 0 { + return Ok(Vec::new()); + } + let mut rows = conn + .query( + "SELECT node_id, depth, created_at, summary_text + FROM lcm_summary_nodes + WHERE provider = ?1 AND session_id = ?2 + ORDER BY depth DESC, created_at DESC, node_id + LIMIT ?3", + params![ + request.provider.as_str(), + request.session_id.as_str(), + clamp_limit(request.summary_limit) as i64, + ], + ) + .await?; + let mut nodes = Vec::new(); + while let Some(row) = rows.next().await? { + let summary_text: String = row.get(3)?; + let (snippet, truncated) = bounded_replay_snippet(&summary_text, request.max_summary_chars); + nodes.push(LcmReplaySummaryNode { + node_id: row.get(0)?, + depth: row.get(1)?, + created_at: row.get(2)?, + snippet, + truncated, + }); + } + Ok(nodes) +} + +fn bounded_replay_snippet(text: &str, max_chars: usize) -> (String, bool) { + let text = text.trim(); + if text.chars().nth(max_chars).is_none() { + (text.to_string(), false) + } else { + (text.chars().take(max_chars).collect(), true) + } +} + pub(crate) async fn grep( conn: &Connection, request: LcmGrepRequest, diff --git a/src/sessions/lcm/types.rs b/src/sessions/lcm/types.rs index 48341c27..c373b14c 100644 --- a/src/sessions/lcm/types.rs +++ b/src/sessions/lcm/types.rs @@ -148,6 +148,64 @@ pub struct LcmLoadSessionMessage { pub metadata_json: Option, } +/// Recency-ordered overview of one session in the LCM raw store, used to +/// select "recently active" sessions for automation replay evidence. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct LcmRecentSession { + pub provider: String, + pub session_id: String, + pub message_count: i64, + pub first_timestamp: Option, + pub last_timestamp: Option, + pub last_store_id: i64, +} + +/// Bounded turn-ordered replay slice request for one session: up to +/// `head_limit` opening turns, `tail_limit` closing turns, and +/// `summary_limit` summary-DAG nodes, each snippet capped by chars. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct LcmSessionReplayRequest { + pub provider: String, + pub session_id: String, + pub head_limit: usize, + pub tail_limit: usize, + pub max_snippet_chars: usize, + pub summary_limit: usize, + pub max_summary_chars: usize, +} + +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct LcmSessionReplaySlice { + pub provider: String, + pub session_id: String, + pub total_messages: i64, + /// Messages between the head and tail slices that were not included. + pub omitted_messages: i64, + pub head: Vec, + pub tail: Vec, + pub summary_nodes: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct LcmReplayMessage { + pub message_id: String, + pub store_id: i64, + pub role: String, + pub ordinal: i64, + pub timestamp: Option, + pub snippet: String, + pub truncated: bool, +} + +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct LcmReplaySummaryNode { + pub node_id: String, + pub depth: i64, + pub created_at: i64, + pub snippet: String, + pub truncated: bool, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "snake_case")] pub enum LcmScope { diff --git a/tests/automation_runner_test/session_reflector.rs b/tests/automation_runner_test/session_reflector.rs index 58ea8e55..6b867244 100644 --- a/tests/automation_runner_test/session_reflector.rs +++ b/tests/automation_runner_test/session_reflector.rs @@ -614,6 +614,7 @@ async fn session_reflector_runner_reads_hermes_profile_lcm_with_filters() { start_time: Some(1_715_100_000), end_time: Some(1_715_100_010), run_id: None, + ..SessionReflectorAutomationOptions::default() }, ) .await @@ -624,6 +625,193 @@ async fn session_reflector_runner_reads_hermes_profile_lcm_with_filters() { assert_eq!(run.ledger_record.rejected_count, 0); } +#[tokio::test] +async fn session_reflector_replays_recent_sessions_without_keyword_matches() { + let temp = tempdir().unwrap(); + let cg = init_project(temp.path()).await; + // Deliberately avoids every keyword in the default reflection query so + // the grep channel returns nothing and only session replay surfaces it. + seed_session_message_in_db( + &GlobalDb::open_at(&cg.store_layout().sessions_db_path) + .await + .expect("session db open"), + cg.project_root(), + SeedSessionMessage { + provider: "cursor", + session_id: "session-replay-1", + message_id: "session-replay-1-message-001", + role: "user", + timestamp: 1_715_000_050, + text: "Always pass the offline flag to cargo nextest on this machine.", + source: None, + }, + ) + .await; + let backend = SessionReplayEvidenceBackend::new( + json!({ + "facts": [ + { + "content": "Cargo nextest must run with the offline flag on this machine", + "category": "project", + "tags": ["testing"], + "entities": ["cargo-nextest"], + "trust": 0.7, + "source_span": { + "session_id": "session-replay-1", + "message_id": "session-replay-1-message-001" + }, + "reason": "Replayed session turn states the requirement directly" + } + ] + }), + "session-replay-1", + "session-replay-1-message-001", + ); + let config = AutomationConfig { + enabled: true, + backend: AutomationBackend::CodexAppServer, + host_mode: AutomationHostMode::Standalone, + tasks: AutomationTaskSet { + session_reflector: AutomationTaskConfig { + enabled: true, + schedule: Some("manual".to_string()), + ..AutomationTaskConfig::default() + }, + ..AutomationTaskSet::default() + }, + ..AutomationConfig::default() + }; + + let run = run_session_reflector_with_backend( + &cg, + &config, + &backend, + SessionReflectorAutomationOptions::default(), + ) + .await + .unwrap(); + + assert_eq!(run.ledger_record.status, AutomationRunStatus::Succeeded); + assert_eq!( + run.ledger_record.accepted_count, 1, + "a fact citing a replay-only turn should validate: {:?}", + run.report["rejected_facts"] + ); + assert_eq!(run.ledger_record.rejected_count, 0); +} + +#[tokio::test] +async fn session_reflector_suppresses_replay_for_filtered_runs() { + let temp = tempdir().unwrap(); + let cg = init_project(temp.path()).await; + seed_session_message_in_db( + &GlobalDb::open_at(&cg.store_layout().sessions_db_path) + .await + .expect("session db open"), + cg.project_root(), + SeedSessionMessage { + provider: "cursor", + session_id: "session-replay-filtered", + message_id: "session-replay-filtered-message-001", + role: "user", + timestamp: 1_715_000_070, + text: "Always pass the offline flag to cargo nextest on this machine.", + source: None, + }, + ) + .await; + let backend = SessionJsonBackend::new(json!({"facts": []})); + let config = AutomationConfig { + enabled: true, + backend: AutomationBackend::CodexAppServer, + host_mode: AutomationHostMode::Standalone, + tasks: AutomationTaskSet { + session_reflector: AutomationTaskConfig { + enabled: true, + schedule: Some("manual".to_string()), + ..AutomationTaskConfig::default() + }, + ..AutomationTaskSet::default() + }, + ..AutomationConfig::default() + }; + + let run = run_session_reflector_with_backend( + &cg, + &config, + &backend, + SessionReflectorAutomationOptions { + role: Some("assistant".to_string()), + ..SessionReflectorAutomationOptions::default() + }, + ) + .await + .unwrap(); + + assert_eq!(backend.calls(), 0); + assert_eq!(run.ledger_record.status, AutomationRunStatus::Skipped); + assert_eq!( + run.ledger_record.error.as_deref(), + Some("no_session_evidence") + ); +} + +#[tokio::test] +async fn session_reflector_skips_when_replay_disabled_and_no_grep_hits() { + let temp = tempdir().unwrap(); + let cg = init_project(temp.path()).await; + seed_session_message_in_db( + &GlobalDb::open_at(&cg.store_layout().sessions_db_path) + .await + .expect("session db open"), + cg.project_root(), + SeedSessionMessage { + provider: "cursor", + session_id: "session-replay-2", + message_id: "session-replay-2-message-001", + role: "user", + timestamp: 1_715_000_060, + text: "Always pass the offline flag to cargo nextest on this machine.", + source: None, + }, + ) + .await; + let backend = SessionJsonBackend::new(json!({"facts": []})); + let config = AutomationConfig { + enabled: true, + backend: AutomationBackend::CodexAppServer, + host_mode: AutomationHostMode::Standalone, + tasks: AutomationTaskSet { + session_reflector: AutomationTaskConfig { + enabled: true, + schedule: Some("manual".to_string()), + ..AutomationTaskConfig::default() + }, + ..AutomationTaskSet::default() + }, + ..AutomationConfig::default() + }; + + let run = run_session_reflector_with_backend( + &cg, + &config, + &backend, + SessionReflectorAutomationOptions { + include_recent_sessions: false, + ..SessionReflectorAutomationOptions::default() + }, + ) + .await + .unwrap(); + + assert_eq!(backend.calls(), 0); + assert_eq!(run.ledger_record.status, AutomationRunStatus::Skipped); + assert_eq!( + run.ledger_record.error.as_deref(), + Some("no_session_evidence") + ); +} + #[tokio::test] async fn session_reflector_runner_ledgers_malformed_backend_output() { let temp = tempdir().unwrap(); diff --git a/tests/automation_runner_test/skill_writer.rs b/tests/automation_runner_test/skill_writer.rs index 01dde018..da808bf7 100644 --- a/tests/automation_runner_test/skill_writer.rs +++ b/tests/automation_runner_test/skill_writer.rs @@ -79,6 +79,98 @@ async fn skill_writer_default_provider_searches_all_providers() { assert_eq!(run.ledger_record.status, AutomationRunStatus::Succeeded); } +#[tokio::test] +async fn skill_writer_replays_recent_sessions_without_keyword_matches() { + let _env_lock = ENV_LOCK.lock().await; + let temp = tempdir().unwrap(); + let profile_root = temp.path().join("profile"); + let cg = init_project(temp.path()).await; + let db = GlobalDb::open_at(&cg.store_layout().sessions_db_path) + .await + .expect("session db open"); + // Deliberately avoids every keyword in the default skill writer query so + // the grep channel returns nothing and only session replay surfaces it. + seed_session_message_in_db( + &db, + cg.project_root(), + SeedSessionMessage { + provider: "cursor", + session_id: "skill-writer-replay-1", + message_id: "skill-writer-replay-1-message-001", + role: "assistant", + timestamp: 1_715_000_070, + text: "Ran the dashboard build twice before every release cut.", + source: None, + }, + ) + .await; + let _global_db = isolate_global_db(&cg); + let backend = SkillWriterReplayEvidenceBackend::new("skill-writer-replay-1"); + let config = enabled_skill_writer_config(); + + let run = run_skill_writer_with_backend( + &cg, + &config, + &backend, + SkillWriterAutomationOptions { + profile_root: Some(profile_root), + ..SkillWriterAutomationOptions::default() + }, + ) + .await + .unwrap(); + + assert_eq!(run.ledger_record.status, AutomationRunStatus::Succeeded); +} + +#[tokio::test] +async fn skill_writer_skips_when_replay_disabled_and_no_grep_hits() { + let _env_lock = ENV_LOCK.lock().await; + let temp = tempdir().unwrap(); + let profile_root = temp.path().join("profile"); + let cg = init_project(temp.path()).await; + let db = GlobalDb::open_at(&cg.store_layout().sessions_db_path) + .await + .expect("session db open"); + seed_session_message_in_db( + &db, + cg.project_root(), + SeedSessionMessage { + provider: "cursor", + session_id: "skill-writer-replay-2", + message_id: "skill-writer-replay-2-message-001", + role: "assistant", + timestamp: 1_715_000_080, + text: "Ran the dashboard build twice before every release cut.", + source: None, + }, + ) + .await; + let _global_db = isolate_global_db(&cg); + let backend = SkillJsonBackend::new(json!({"skills": []})); + let config = enabled_skill_writer_config(); + + let run = run_skill_writer_with_backend( + &cg, + &config, + &backend, + SkillWriterAutomationOptions { + include_recent_sessions: false, + profile_root: Some(profile_root), + ..SkillWriterAutomationOptions::default() + }, + ) + .await + .unwrap(); + + assert_eq!(backend.calls(), 0); + assert_eq!(run.ledger_record.status, AutomationRunStatus::Skipped); + assert_eq!( + run.ledger_record.error.as_deref(), + Some("no_skill_writer_evidence") + ); +} + #[tokio::test] async fn skill_writer_runner_reads_hermes_profile_lcm() { let temp = tempdir().unwrap(); diff --git a/tests/automation_runner_test/support.rs b/tests/automation_runner_test/support.rs index accd60b7..3d176c51 100644 --- a/tests/automation_runner_test/support.rs +++ b/tests/automation_runner_test/support.rs @@ -545,6 +545,8 @@ impl AgentTaskBackend for InspectSessionEvidenceBackend { assert_eq!(evidence["role"], json!("assistant")); assert_eq!(evidence["start_time"], json!(1_715_100_000_i64)); assert_eq!(evidence["end_time"], json!(1_715_100_010_i64)); + assert_eq!(evidence["evidence_mode"], json!("grep_only")); + assert_eq!(evidence["recent_session_slices"], json!(null)); let hits = evidence["hits"].as_array().expect("hits array"); assert_eq!(hits.len(), 1); assert_eq!(hits[0]["session_id"], json!("hermes-reflect-1")); @@ -564,6 +566,120 @@ impl AgentTaskBackend for InspectSessionEvidenceBackend { } } +/// Asserts the session reflector received session-replay evidence for one +/// recently active session (with no keyword grep hits) and replies with the +/// configured facts output. +pub(crate) struct SessionReplayEvidenceBackend { + output: Value, + expected_session_id: &'static str, + expected_message_id: &'static str, +} + +impl SessionReplayEvidenceBackend { + pub(crate) fn new( + output: Value, + expected_session_id: &'static str, + expected_message_id: &'static str, + ) -> Self { + Self { + output, + expected_session_id, + expected_message_id, + } + } +} + +impl AgentTaskBackend for SessionReplayEvidenceBackend { + fn run_task( + &self, + request: &AgentTaskRequest, + ) -> tracedecay::errors::Result { + assert_eq!(request.task, AgentTaskKind::SessionReflector); + assert_request_contract( + request, + "session_reflector", + "session_reflector:v2", + "facts", + ); + let evidence = &request.context["session_reflection_evidence"]; + assert_eq!(evidence["evidence_mode"], json!("session_replay_with_grep")); + assert_eq!(evidence["hits"], json!([])); + let slices = &evidence["recent_session_slices"]; + assert_eq!(slices["mode"], json!("recent_sessions")); + assert_eq!(slices["session_selection"], json!("recent_activity")); + assert_eq!(slices["bounds"]["head_turns"], json!(4)); + assert_eq!(slices["bounds"]["tail_turns"], json!(4)); + let sessions = slices["sessions"].as_array().expect("replay sessions"); + assert_eq!(sessions.len(), 1); + assert_eq!( + sessions[0]["session_id"], + json!(self.expected_session_id), + "replay slice should target the recently active session" + ); + let head = sessions[0]["head"].as_array().expect("head turns"); + assert!( + head.iter() + .any(|message| message["message_id"] == json!(self.expected_message_id)), + "replayed head turns should include the seeded message" + ); + Ok(AgentTaskResponse { + run_id: request.run_id.clone(), + task: request.task, + output_text: self.output.to_string(), + output_json: Some(self.output.clone()), + model: Some("fixture-model".to_string()), + input_tokens: Some(10), + output_tokens: Some(20), + }) + } +} + +/// Asserts the skill writer received session-replay evidence (with no +/// keyword grep hits) and replies with an empty skills array. +pub(crate) struct SkillWriterReplayEvidenceBackend { + expected_session_id: &'static str, +} + +impl SkillWriterReplayEvidenceBackend { + pub(crate) fn new(expected_session_id: &'static str) -> Self { + Self { + expected_session_id, + } + } +} + +impl AgentTaskBackend for SkillWriterReplayEvidenceBackend { + fn run_task( + &self, + request: &AgentTaskRequest, + ) -> tracedecay::errors::Result { + assert_eq!(request.task, AgentTaskKind::SkillWriter); + assert_request_contract(request, "skill_writer", "skill_writer:v2", "skills"); + let evidence = &request.context["skill_writer_evidence"]; + assert_eq!(evidence["evidence_mode"], json!("session_replay_with_grep")); + assert_eq!(evidence["hits"], json!([])); + let slices = &evidence["recent_session_slices"]; + assert_eq!(slices["mode"], json!("recent_sessions")); + assert_eq!(slices["session_selection"], json!("recent_activity")); + let sessions = slices["sessions"].as_array().expect("replay sessions"); + assert!( + sessions + .iter() + .any(|session| session["session_id"] == json!(self.expected_session_id)), + "replay slices should include the recently active session" + ); + Ok(AgentTaskResponse { + run_id: request.run_id.clone(), + task: request.task, + output_text: json!({"skills": []}).to_string(), + output_json: Some(json!({"skills": []})), + model: Some("fixture-model".to_string()), + input_tokens: Some(10), + output_tokens: Some(20), + }) + } +} + pub(crate) fn assert_request_contract( request: &AgentTaskRequest, task_key: &str, diff --git a/tests/session_suite/lcm_query.rs b/tests/session_suite/lcm_query.rs index beeada5e..a574f442 100644 --- a/tests/session_suite/lcm_query.rs +++ b/tests/session_suite/lcm_query.rs @@ -3,8 +3,9 @@ use tracedecay::global_db::{GlobalDb, ParseOffset}; use tracedecay::sessions::lcm::{ LcmContentSlice, LcmDescribeRequest, LcmDescribeTarget, LcmError, LcmExpandQueryRequest, LcmExpandRequest, LcmExpandTarget, LcmGcConfig, LcmGrepRequest, LcmGrepSort, - LcmLifecycleUpdate, LcmLoadSessionRequest, LcmMaintenanceDebt, LcmScope, LcmSourceRef, - LcmStorageKind, LcmSummaryNodeDraft, LCM_SCHEMA_VERSION, MAX_DERIVED_SNIPPET_CHARS, + LcmLifecycleUpdate, LcmLoadSessionRequest, LcmMaintenanceDebt, LcmScope, + LcmSessionReplayRequest, LcmSourceRef, LcmStorageKind, LcmSummaryNodeDraft, LCM_SCHEMA_VERSION, + MAX_DERIVED_SNIPPET_CHARS, }; use tracedecay::sessions::{SessionMessageRecord, SessionRecord}; @@ -2243,3 +2244,215 @@ async fn status_counts_lossy_ingest_records_with_pinned_metadata_semantics() { assert!(status.redaction.enabled); assert_eq!(status.redaction.legacy_truncated_count, 0); } + +#[tokio::test] +async fn recent_sessions_orders_by_last_activity_with_provider_filter() { + let tmp = TempDir::new().unwrap(); + let db = open_lcm_db(&tmp).await; + let db_path = isolated_db_path(&tmp); + // raw_message assigns timestamp = base + ordinal, so the session with + // more messages has the most recent activity. + insert_raw_messages( + &db, + &db_path, + "cursor", + "session-older", + &["first turn".to_string(), "second turn".to_string()], + ) + .await; + insert_raw_messages( + &db, + &db_path, + "codex", + "session-newer", + &[ + "first turn".to_string(), + "second turn".to_string(), + "third turn".to_string(), + ], + ) + .await; + + let sessions = db + .lcm_recent_sessions(None, 10) + .await + .expect("recent sessions should load"); + assert_eq!( + sessions + .iter() + .map(|session| session.session_id.as_str()) + .collect::>(), + vec!["session-newer", "session-older"] + ); + assert_eq!(sessions[0].provider, "codex"); + assert_eq!(sessions[0].message_count, 3); + assert_eq!(sessions[0].last_timestamp, Some(1_715_000_003)); + assert_eq!(sessions[1].message_count, 2); + + let cursor_only = db + .lcm_recent_sessions(Some("cursor"), 10) + .await + .expect("provider-filtered recent sessions should load"); + assert_eq!(cursor_only.len(), 1); + assert_eq!(cursor_only[0].session_id, "session-older"); +} + +#[tokio::test] +async fn session_providers_finds_explicit_session_beyond_recent_limit() { + let tmp = TempDir::new().unwrap(); + let db = open_lcm_db(&tmp).await; + let db_path = isolated_db_path(&tmp); + insert_raw_messages( + &db, + &db_path, + "codex", + "explicit-session", + &["older explicit turn".to_string()], + ) + .await; + for idx in 0..105 { + insert_raw_messages( + &db, + &db_path, + "cursor", + &format!("newer-session-{idx:03}"), + &["newer turn".to_string()], + ) + .await; + } + + let recent = db + .lcm_recent_sessions(None, 100) + .await + .expect("recent sessions should load"); + assert!( + recent + .iter() + .all(|session| session.session_id != "explicit-session"), + "fixture must place explicit session outside the bounded recent scan" + ); + + let providers = db + .lcm_session_providers("explicit-session") + .await + .expect("explicit session providers should load without recency limit"); + assert_eq!(providers, vec!["codex"]); +} + +#[tokio::test] +async fn session_replay_slice_bounds_head_tail_and_summaries() { + let tmp = TempDir::new().unwrap(); + let db = open_lcm_db(&tmp).await; + let contents = (1..=12) + .map(|idx| format!("turn-{idx:02} with a deliberately verbose body")) + .collect::>(); + let store_ids = insert_raw_messages( + &db, + &isolated_db_path(&tmp), + "cursor", + "session-replay", + &contents, + ) + .await; + db.lcm_insert_summary_node(summary_draft( + "cursor", + "session-replay", + "summary of the replayed session", + vec![LcmSourceRef::RawMessage { + store_id: store_ids[0], + }], + )) + .await + .expect("summary should insert"); + + let slice = db + .lcm_session_replay_slice(&LcmSessionReplayRequest { + provider: "cursor".to_string(), + session_id: "session-replay".to_string(), + head_limit: 4, + tail_limit: 4, + max_snippet_chars: 12, + summary_limit: 3, + max_summary_chars: 500, + }) + .await + .expect("replay slice should load"); + + assert_eq!(slice.total_messages, 12); + assert_eq!(slice.omitted_messages, 4); + assert_eq!( + slice + .head + .iter() + .map(|message| message.message_id.as_str()) + .collect::>(), + vec![ + "session-replay-message-001", + "session-replay-message-002", + "session-replay-message-003", + "session-replay-message-004", + ] + ); + assert_eq!( + slice + .tail + .iter() + .map(|message| message.message_id.as_str()) + .collect::>(), + vec![ + "session-replay-message-009", + "session-replay-message-010", + "session-replay-message-011", + "session-replay-message-012", + ] + ); + for message in slice.head.iter().chain(slice.tail.iter()) { + assert!(message.truncated); + assert_eq!(message.snippet.chars().count(), 12); + } + assert_eq!(slice.summary_nodes.len(), 1); + assert_eq!( + slice.summary_nodes[0].snippet, + "summary of the replayed session" + ); + assert!(!slice.summary_nodes[0].truncated); +} + +#[tokio::test] +async fn session_replay_slice_short_session_has_no_tail_overlap() { + let tmp = TempDir::new().unwrap(); + let db = open_lcm_db(&tmp).await; + insert_raw_messages( + &db, + &isolated_db_path(&tmp), + "cursor", + "session-short", + &[ + "only turn one".to_string(), + "only turn two".to_string(), + "only turn three".to_string(), + ], + ) + .await; + + let slice = db + .lcm_session_replay_slice(&LcmSessionReplayRequest { + provider: "cursor".to_string(), + session_id: "session-short".to_string(), + head_limit: 4, + tail_limit: 4, + max_snippet_chars: 500, + summary_limit: 3, + max_summary_chars: 500, + }) + .await + .expect("replay slice should load"); + + assert_eq!(slice.total_messages, 3); + assert_eq!(slice.omitted_messages, 0); + assert_eq!(slice.head.len(), 3); + assert!(slice.tail.is_empty(), "tail must not repeat head turns"); + assert!(!slice.head[0].truncated); + assert_eq!(slice.head[0].snippet, "only turn one"); + assert!(slice.summary_nodes.is_empty()); +}