diff --git a/crates/lore-core/migrations/0014_ingest_resume.sql b/crates/lore-core/migrations/0014_ingest_resume.sql new file mode 100644 index 0000000..f7a75ec --- /dev/null +++ b/crates/lore-core/migrations/0014_ingest_resume.sql @@ -0,0 +1,5 @@ +-- Migration 0014: the parser checkpoint each source's last ingest ended at +-- (`ParseCheckpoint` JSON), so an append is parsed and persisted from there +-- instead of from the start of the file. NULL means no checkpoint: the next +-- change parses the whole file, as every change did before this migration. +ALTER TABLE ingest_state ADD COLUMN resume_json TEXT; diff --git a/crates/lore-core/src/adapters.rs b/crates/lore-core/src/adapters.rs index 47bd8c6..c88b4dc 100644 --- a/crates/lore-core/src/adapters.rs +++ b/crates/lore-core/src/adapters.rs @@ -9,7 +9,7 @@ use std::path::PathBuf; use std::time::SystemTime; -use crate::model::ParsedSession; +use crate::model::{ParseCheckpoint, ParsedAppend, ParsedSession}; pub(crate) mod common; @@ -87,6 +87,25 @@ pub enum AdapterError { Unreadable, } +/// The rows a session already has in the archive, for appended lines that refer +/// back past a [`ParseCheckpoint`]. +pub trait ArchivedSession { + fn has_message(&self, native_uuid: &str) -> bool; + fn has_tool_call(&self, native_call_id: &str) -> bool; +} + +/// A session with nothing archived yet: every reference is to the parsed lines. +pub(crate) struct NothingArchived; + +impl ArchivedSession for NothingArchived { + fn has_message(&self, _: &str) -> bool { + false + } + fn has_tool_call(&self, _: &str) -> bool { + false + } +} + /// The read-only interface every agent adapter implements. pub trait AgentAdapter: Send + Sync { fn id(&self) -> AgentId; @@ -121,6 +140,32 @@ pub trait AgentAdapter: Send + Sync { /// source carries no native session id. fn parse_content(&self, content: &str, fallback_dedupe: &str) -> ParsedSession; + /// [`AgentAdapter::parse_content`] plus the parser state after the last + /// line, so a later append can resume there. `None` when this adapter + /// cannot resume, which keeps every change on a whole-file parse. + fn parse_checkpointed( + &self, + content: &str, + fallback_dedupe: &str, + ) -> (ParsedSession, Option) { + (self.parse_content(content, fallback_dedupe), None) + } + + /// Parse `tail`, the complete lines appended after `from`, as if the whole + /// file had been parsed. `archived` answers for rows parsed before `from`. + /// `None` when the tail changes what earlier lines produced; the caller + /// then parses the whole file. + fn parse_append( + &self, + from: &ParseCheckpoint, + tail: &str, + fallback_dedupe: &str, + archived: &dyn ArchivedSession, + ) -> Option { + let _ = (from, tail, fallback_dedupe, archived); + None + } + /// Parse one session file into the normalized model. The default reads the /// file once and delegates to [`AgentAdapter::parse_content`]; adapters need /// not override it. diff --git a/crates/lore-core/src/adapters/claude_code.rs b/crates/lore-core/src/adapters/claude_code.rs index f2f8887..656885f 100644 --- a/crates/lore-core/src/adapters/claude_code.rs +++ b/crates/lore-core/src/adapters/claude_code.rs @@ -12,13 +12,18 @@ use std::path::PathBuf; use serde_json::Value; -use super::common::{bounded, epoch_ms, fallback_title, json_field, sanitize_path, str_field}; +use super::common::{ + bounded, epoch_ms, finish_parse, json_field, resume_session, sanitize_path, str_field, + ParseEnd, ParseRun, +}; use super::{ - AgentAdapter, AgentId, AgentMetadata, Capabilities, Detection, DiscoveryRoots, SessionRef, + AgentAdapter, AgentId, AgentMetadata, ArchivedSession, Capabilities, Detection, DiscoveryRoots, + NothingArchived, SessionRef, }; use crate::model::{ - EventKind, FileChangeKind, FileEventSource, ParsedFileEvent, ParsedMessage, ParsedPart, - ParsedSegment, ParsedSession, ParsedToolCall, PartKind, Role, Tokens, + EventKind, FileChangeKind, FileEventSource, ParseCheckpoint, ParsedAppend, ParsedFileEvent, + ParsedMessage, ParsedPart, ParsedSegment, ParsedSession, ParsedToolCall, ParsedToolResult, + PartKind, Role, Tokens, }; const AGENT_ID: AgentId = AgentId("claude-code"); @@ -53,89 +58,109 @@ impl ClaudeCodeAdapter { /// Parse already-read file content. Separated for deterministic testing. #[must_use] pub fn parse_str(&self, content: &str, fallback_dedupe: &str) -> ParsedSession { - let mut session = ParsedSession::new(fallback_dedupe); - // Per-message context (cwd, branch) used to build segments afterward. - let mut contexts: Vec<(Option, Option)> = Vec::new(); - let mut offset: i64 = 0; - let mut seq: i64 = 0; - // native_call_id -> index into session.tool_calls, for result pairing. - let mut id_map: HashMap = HashMap::new(); - - for line in content.split_inclusive('\n') { - let line_start = offset; - offset += line.len() as i64; - let trimmed = line.trim(); - if trimmed.is_empty() { - continue; - } - let Ok(obj) = serde_json::from_str::(trimmed) else { - session.note_partial("unparseable line (possibly truncated)"); - continue; - }; - let event_type = obj.get("type").and_then(Value::as_str).unwrap_or(""); + self.parse_checkpointed(content, fallback_dedupe).0 + } +} - // Session-level metadata. - if session.native_session_id.is_none() { - session.native_session_id = str_field(&obj, "sessionId"); - } - if session.agent_version.is_none() { - session.agent_version = str_field(&obj, "version"); - } +/// Parse `content` as the lines that follow `from`. +fn parse_from( + from: &ParseCheckpoint, + content: &str, + fallback_dedupe: &str, + archived: &dyn ArchivedSession, +) -> ParseRun { + let mut session = resume_session(from, fallback_dedupe); + // Per-message context (cwd, branch) used to build segments afterward. + let mut contexts: Vec<(Option, Option)> = Vec::new(); + let mut offset = from.offset; + let mut seq = from.next_seq; + let mut calls = Calls { + id_map: HashMap::new(), + archived, + earlier_results: Vec::new(), + }; - match event_type { - "custom-title" => { - session.title = str_field(&obj, "customTitle").or(session.title.take()); - continue; - } - "ai-title" => { - if session.title.is_none() { - session.title = str_field(&obj, "aiTitle"); - } - continue; - } - // Internal bookkeeping events carry no conversation content. - "queue-operation" | "last-prompt" => continue, - "user" | "assistant" | "system" | "attachment" | "summary" | "mode" | "pr-link" => { - } - other => { - session.note_partial(format!("unhandled event type: {}", bounded(other))); - continue; - } - } + for line in content.split_inclusive('\n') { + let line_start = offset; + offset += line.len() as i64; + let trimmed = line.trim(); + if trimmed.is_empty() { + continue; + } + let Ok(obj) = serde_json::from_str::(trimmed) else { + session.note_partial("unparseable line (possibly truncated)"); + continue; + }; + let event_type = obj.get("type").and_then(Value::as_str).unwrap_or(""); + + // Session-level metadata. + if session.native_session_id.is_none() { + session.native_session_id = str_field(&obj, "sessionId"); + } + if session.agent_version.is_none() { + session.agent_version = str_field(&obj, "version"); + } - let msg = build_message(&obj, event_type, seq, line_start, &mut session); - extract_tool_calls(&obj, seq, &mut session, &mut id_map); - contexts.push((str_field(&obj, "cwd"), str_field(&obj, "gitBranch"))); - if let Some(ts) = msg.ts { - session.started_at = Some(session.started_at.map_or(ts, |s| s.min(ts))); - session.ended_at = Some(session.ended_at.map_or(ts, |e| e.max(ts))); + match event_type { + "custom-title" => { + session.title = str_field(&obj, "customTitle").or(session.title.take()); + continue; + } + "ai-title" => { + if session.title.is_none() { + session.title = str_field(&obj, "aiTitle"); + } + continue; } - if session.primary_model.is_none() { - session.primary_model = msg.model.clone(); + // Internal bookkeeping events carry no conversation content. + "queue-operation" | "last-prompt" => continue, + "user" | "assistant" | "system" | "attachment" | "summary" | "mode" | "pr-link" => {} + other => { + session.note_partial(format!("unhandled event type: {}", bounded(other))); + continue; } - session.messages.push(msg); - seq += 1; } - assign_segments(&mut session, &contexts); - super::common::resolve_file_event_segments(&mut session); - if session.title.is_none() { - session.title = fallback_title(&session.messages); - session.title_is_synthetic = session.title.is_some(); + let msg = build_message(&obj, event_type, seq, line_start, &mut session); + extract_tool_calls(&obj, seq, &mut session, &mut calls); + contexts.push((str_field(&obj, "cwd"), str_field(&obj, "gitBranch"))); + if let Some(ts) = msg.ts { + session.started_at = Some(session.started_at.map_or(ts, |s| s.min(ts))); + session.ended_at = Some(session.ended_at.map_or(ts, |e| e.max(ts))); } - session - } + if session.primary_model.is_none() { + session.primary_model = msg.model.clone(); + } + session.messages.push(msg); + seq += 1; + } + + assign_segments(&mut session, &contexts, from); + super::common::resolve_file_event_segments(&mut session); + let end = ParseEnd { + offset, + next_seq: seq, + earlier_results: calls.earlier_results, + cwd: None, + model: None, + restates_session: false, + }; + finish_parse(from, session, end, archived) +} + +/// Tool-call pairing state: calls parsed in this run, and results for calls +/// parsed before it. +struct Calls<'a> { + /// native_call_id -> index into session.tool_calls. + id_map: HashMap, + archived: &'a dyn ArchivedSession, + earlier_results: Vec, } /// Extract tool calls and results from a message's content blocks, pairing /// results to calls by `tool_use_id`, and derive file events from file-mutating /// tool invocations. -fn extract_tool_calls( - obj: &Value, - seq: i64, - session: &mut ParsedSession, - id_map: &mut HashMap, -) { +fn extract_tool_calls(obj: &Value, seq: i64, session: &mut ParsedSession, calls: &mut Calls<'_>) { let Some(blocks) = obj .get("message") .and_then(|m| m.get("content")) @@ -163,7 +188,7 @@ fn extract_tool_calls( is_error: None, duration_ms: None, }); - id_map.insert(id.clone(), idx); + calls.id_map.insert(id.clone(), idx); if let Some(fe) = file_event_from_tool(&name, block, id) { session.file_events.push(fe); } @@ -173,11 +198,20 @@ fn extract_tool_calls( session.note_partial("tool_result without tool_use_id"); continue; }; - if let Some(&idx) = id_map.get(&tid) { + let output_text = tool_result_text(block); + let is_error = block.get("is_error").and_then(Value::as_bool); + if let Some(&idx) = calls.id_map.get(&tid) { let tc = &mut session.tool_calls[idx]; tc.result_ref = Some((seq, ordinal)); - tc.output_text = tool_result_text(block); - tc.is_error = block.get("is_error").and_then(Value::as_bool); + tc.output_text = output_text; + tc.is_error = is_error; + } else if calls.archived.has_tool_call(&tid) { + calls.earlier_results.push(ParsedToolResult { + native_call_id: tid, + result_ref: (seq, ordinal), + output_text, + is_error, + }); } else { session.note_partial("tool_result without matching tool_use"); } @@ -395,16 +429,21 @@ fn raw_part(ordinal: i64, kind: PartKind, block: &Value) -> ParsedPart { } /// Group consecutive messages sharing (cwd, branch) into segments and assign -/// each message its `segment_ix`. -fn assign_segments(session: &mut ParsedSession, contexts: &[(Option, Option)]) { - session.segments.clear(); - let mut current: Option<(Option, Option)> = None; +/// each message its `segment_ix`, continuing the segment `from` left open. +fn assign_segments( + session: &mut ParsedSession, + contexts: &[(Option, Option)], + from: &ParseCheckpoint, +) { + session.segments = from.open_segment.iter().cloned().collect(); + let base = from.segment_count.saturating_sub(session.segments.len()); + let mut current = from + .open_segment + .as_ref() + .map(|s| (s.cwd.clone(), s.git_branch.clone())); for (i, ctx) in contexts.iter().enumerate() { let seq = session.messages[i].seq; if current.as_ref() != Some(ctx) { - if let Some(last) = session.segments.last_mut() { - last.seq_end = session.messages[i.saturating_sub(1)].seq; - } session.segments.push(ParsedSegment { seq_start: seq, seq_end: seq, @@ -417,8 +456,7 @@ fn assign_segments(session: &mut ParsedSession, contexts: &[(Option, Opt }); current = Some(ctx.clone()); } - let ix = session.segments.len() - 1; - session.messages[i].segment_ix = ix; + session.messages[i].segment_ix = base + session.segments.len() - 1; if let Some(last) = session.segments.last_mut() { last.seq_end = seq; } @@ -522,6 +560,30 @@ impl AgentAdapter for ClaudeCodeAdapter { fn parse_content(&self, content: &str, fallback_dedupe: &str) -> ParsedSession { self.parse_str(content, fallback_dedupe) } + + fn parse_checkpointed( + &self, + content: &str, + fallback_dedupe: &str, + ) -> (ParsedSession, Option) { + parse_from( + &ParseCheckpoint::default(), + content, + fallback_dedupe, + &NothingArchived, + ) + .checkpointed() + } + + fn parse_append( + &self, + from: &ParseCheckpoint, + tail: &str, + fallback_dedupe: &str, + archived: &dyn ArchivedSession, + ) -> Option { + parse_from(from, tail, fallback_dedupe, archived).append() + } } /// Recursively collect `*.jsonl` transcripts below `dir`. diff --git a/crates/lore-core/src/adapters/codex.rs b/crates/lore-core/src/adapters/codex.rs index 57e75c6..7e98391 100644 --- a/crates/lore-core/src/adapters/codex.rs +++ b/crates/lore-core/src/adapters/codex.rs @@ -14,15 +14,17 @@ use std::path::{Path, PathBuf}; use serde_json::Value; use super::common::{ - bounded, epoch_ms, fallback_title, non_negative_int_field, resolve_file_event_segments, - sanitize_path, str_field, unified_diff_line_counts, + bounded, epoch_ms, finish_parse, non_negative_int_field, resolve_file_event_segments, + resume_session, sanitize_path, str_field, unified_diff_line_counts, ParseEnd, ParseRun, }; use super::{ - AgentAdapter, AgentId, AgentMetadata, Capabilities, Detection, DiscoveryRoots, SessionRef, + AgentAdapter, AgentId, AgentMetadata, ArchivedSession, Capabilities, Detection, DiscoveryRoots, + NothingArchived, SessionRef, }; use crate::model::{ - EventKind, FileChangeKind, FileEventSource, ParsedFileEvent, ParsedMessage, ParsedPart, - ParsedSegment, ParsedSession, ParsedToolCall, PartKind, Role, Tokens, + EventKind, FileChangeKind, FileEventSource, ParseCheckpoint, ParsedAppend, ParsedFileEvent, + ParsedMessage, ParsedPart, ParsedSegment, ParsedSession, ParsedToolCall, ParsedToolResult, + PartKind, Role, Tokens, }; const AGENT_ID: AgentId = AgentId("codex"); @@ -61,176 +63,199 @@ impl CodexAdapter { /// Parse rollout content. Separated for deterministic testing. #[must_use] pub fn parse_str(&self, content: &str, fallback_dedupe: &str) -> ParsedSession { - let mut session = ParsedSession::new(fallback_dedupe); - let mut cwd: Option = None; - let mut model: Option = None; - let mut provider: Option = None; - let mut git: RecordedGit = (None, None, None); - // (cwd, model) at each emitted message, for segmentation. - let mut contexts: Vec<(Option, Option)> = Vec::new(); - let mut offset: i64 = 0; - let mut seq: i64 = 0; - // native call_id -> index into session.tool_calls, for result pairing. - let mut id_map: HashMap = HashMap::new(); - - for line in content.split_inclusive('\n') { - let line_start = offset; - offset += line.len() as i64; - let trimmed = line.trim(); - if trimmed.is_empty() { - continue; - } - let Ok(obj) = serde_json::from_str::(trimmed) else { - session.note_partial("unparseable line (possibly truncated)"); - continue; - }; - let typ = obj.get("type").and_then(Value::as_str).unwrap_or(""); - let ts = obj - .get("timestamp") - .and_then(Value::as_str) - .and_then(epoch_ms); - if let Some(ts) = ts { - session.started_at = Some(session.started_at.map_or(ts, |s| s.min(ts))); - session.ended_at = Some(session.ended_at.map_or(ts, |e| e.max(ts))); - } - let payload = obj.get("payload"); - - match typ { - "session_meta" => { - if let Some(p) = payload { - session.native_session_id = str_field(p, "id"); - session.agent_version = str_field(p, "cli_version"); - provider = str_field(p, "model_provider"); - if let Some(c) = str_field(p, "cwd") { - cwd = Some(c); - } - git = recorded_git(p); + self.parse_checkpointed(content, fallback_dedupe).0 + } +} + +/// Parse rollout `content` as the lines that follow `from`. +fn parse_from( + from: &ParseCheckpoint, + content: &str, + fallback_dedupe: &str, + archived: &dyn ArchivedSession, +) -> ParseRun { + let mut session = resume_session(from, fallback_dedupe); + let mut cwd = from.cwd.clone(); + let mut model = from.model.clone(); + // Every segment carries the provider. Recorded git reaches only the + // first segment, which a checkpoint has already emitted. + let mut provider = from.open_segment.as_ref().and_then(|s| s.provider.clone()); + let mut git: RecordedGit = (None, None, None); + let mut restates_session = false; + // (cwd, model) at each emitted message, for segmentation. + let mut contexts: Vec<(Option, Option)> = Vec::new(); + let mut offset = from.offset; + let mut seq = from.next_seq; + // native call_id -> index into session.tool_calls, for result pairing. + let mut id_map: HashMap = HashMap::new(); + let mut earlier_results = Vec::new(); + + for line in content.split_inclusive('\n') { + let line_start = offset; + offset += line.len() as i64; + let trimmed = line.trim(); + if trimmed.is_empty() { + continue; + } + let Ok(obj) = serde_json::from_str::(trimmed) else { + session.note_partial("unparseable line (possibly truncated)"); + continue; + }; + let typ = obj.get("type").and_then(Value::as_str).unwrap_or(""); + let ts = obj + .get("timestamp") + .and_then(Value::as_str) + .and_then(epoch_ms); + if let Some(ts) = ts { + session.started_at = Some(session.started_at.map_or(ts, |s| s.min(ts))); + session.ended_at = Some(session.ended_at.map_or(ts, |e| e.max(ts))); + } + let payload = obj.get("payload"); + + match typ { + "session_meta" => { + restates_session = true; + if let Some(p) = payload { + session.native_session_id = str_field(p, "id"); + session.agent_version = str_field(p, "cli_version"); + provider = str_field(p, "model_provider"); + if let Some(c) = str_field(p, "cwd") { + cwd = Some(c); } + git = recorded_git(p); } - "turn_context" => { - if let Some(p) = payload { - if let Some(c) = str_field(p, "cwd") { - cwd = Some(c); - } - if let Some(m) = str_field(p, "model") { - model = Some(m); - } + } + "turn_context" => { + if let Some(p) = payload { + if let Some(c) = str_field(p, "cwd") { + cwd = Some(c); + } + if let Some(m) = str_field(p, "model") { + model = Some(m); } } - "response_item" => { - let Some(p) = payload else { continue }; - match p.get("type").and_then(Value::as_str).unwrap_or("") { - "message" => { - let msg = message_item(p, seq, ts, model.clone(), line_start); - session.messages.push(msg); - contexts.push((cwd.clone(), model.clone())); - seq += 1; - } - "reasoning" => { - let msg = reasoning_item(p, seq, ts, model.clone(), line_start); - session.messages.push(msg); - contexts.push((cwd.clone(), model.clone())); - seq += 1; - } - "function_call" | "custom_tool_call" => { - let part = tool_part(PartKind::ToolUse, None, p); - session.messages.push(tool_message( - Role::Assistant, - part, - seq, - ts, - line_start, - )); - contexts.push((cwd.clone(), model.clone())); - register_tool_call(p, seq, &mut session, &mut id_map); - seq += 1; - } - "function_call_output" | "custom_tool_call_output" => { - let output_text = json_text(p.get("output")); - let part = tool_part(PartKind::ToolResult, output_text.clone(), p); - session.messages.push(tool_message( - Role::Tool, - part, - seq, - ts, - line_start, - )); - contexts.push((cwd.clone(), model.clone())); - attach_tool_output(p, seq, output_text, &mut session, &id_map); - seq += 1; - } - // tool_search_* / web_search_call are recorded but not yet modeled. - "tool_search_call" | "tool_search_output" | "web_search_call" => {} - other => { - session - .note_partial(format!("unknown response_item: {}", bounded(other))); - } + } + "response_item" => { + let Some(p) = payload else { continue }; + match p.get("type").and_then(Value::as_str).unwrap_or("") { + "message" => { + let msg = message_item(p, seq, ts, model.clone(), line_start); + session.messages.push(msg); + contexts.push((cwd.clone(), model.clone())); + seq += 1; + } + "reasoning" => { + let msg = reasoning_item(p, seq, ts, model.clone(), line_start); + session.messages.push(msg); + contexts.push((cwd.clone(), model.clone())); + seq += 1; + } + "function_call" | "custom_tool_call" => { + let part = tool_part(PartKind::ToolUse, None, p); + session.messages.push(tool_message( + Role::Assistant, + part, + seq, + ts, + line_start, + )); + contexts.push((cwd.clone(), model.clone())); + register_tool_call(p, seq, &mut session, &mut id_map); + seq += 1; + } + "function_call_output" | "custom_tool_call_output" => { + let output_text = json_text(p.get("output")); + let part = tool_part(PartKind::ToolResult, output_text.clone(), p); + session + .messages + .push(tool_message(Role::Tool, part, seq, ts, line_start)); + contexts.push((cwd.clone(), model.clone())); + let paired = attach_tool_output( + p, + seq, + output_text, + &mut session, + &id_map, + archived, + ); + earlier_results.extend(paired); + seq += 1; + } + // tool_search_* / web_search_call are recorded but not yet modeled. + "tool_search_call" | "tool_search_output" | "web_search_call" => {} + other => { + session.note_partial(format!("unknown response_item: {}", bounded(other))); } } - "event_msg" => { - let Some(p) = payload else { continue }; - // Conversation comes from response_item; most event_msg entries are - // telemetry. context_compacted leaves a marker; patch_apply_end yields - // file events attached to the matching tool call by call_id. - match p.get("type").and_then(Value::as_str).unwrap_or("") { - "context_compacted" => { - session - .messages - .push(compaction_marker(None, seq, ts, line_start)); - contexts.push((cwd.clone(), model.clone())); - seq += 1; - } - "patch_apply_end" => extract_patch_changes(p, ts, &mut session), - "token_count" => { - if let Some(tokens) = token_totals(p) { - session.total_tokens = tokens; - } - } - // Known telemetry that carries no conversation content of - // its own (the conversation is reconstructed from - // response_item). Intentionally ignored, not flagged - // (CODEX.md §3). - "task_started" - | "user_message" - | "agent_message" - | "agent_reasoning" - | "thread_settings_applied" - | "mcp_tool_call_start" - | "mcp_tool_call_end" - | "web_search_start" - | "web_search_end" - | "task_complete" - | "turn_aborted" => {} - // A genuinely new event_msg subtype: degrade to a bounded, - // content-free note rather than dropping it silently. - other => { - session.note_partial(format!("unknown event_msg: {}", bounded(other))); + } + "event_msg" => { + let Some(p) = payload else { continue }; + // Conversation comes from response_item; most event_msg entries are + // telemetry. context_compacted leaves a marker; patch_apply_end yields + // file events attached to the matching tool call by call_id. + match p.get("type").and_then(Value::as_str).unwrap_or("") { + "context_compacted" => { + session + .messages + .push(compaction_marker(None, seq, ts, line_start)); + contexts.push((cwd.clone(), model.clone())); + seq += 1; + } + "patch_apply_end" => extract_patch_changes(p, ts, &mut session), + "token_count" => { + if let Some(tokens) = token_totals(p) { + session.total_tokens = tokens; } } + // Known telemetry that carries no conversation content of + // its own (the conversation is reconstructed from + // response_item). Intentionally ignored, not flagged + // (CODEX.md §3). + "task_started" + | "user_message" + | "agent_message" + | "agent_reasoning" + | "thread_settings_applied" + | "mcp_tool_call_start" + | "mcp_tool_call_end" + | "web_search_start" + | "web_search_end" + | "task_complete" + | "turn_aborted" => {} + // A genuinely new event_msg subtype: degrade to a bounded, + // content-free note rather than dropping it silently. + other => { + session.note_partial(format!("unknown event_msg: {}", bounded(other))); + } } - "compacted" => { - let text = payload.and_then(|p| str_field(p, "message")); - session - .messages - .push(compaction_marker(text, seq, ts, line_start)); - contexts.push((cwd.clone(), model.clone())); - seq += 1; - } - // Known non-conversation record types. - "world_state" | "inter_agent_communication_metadata" => {} - other => { - session.note_partial(format!("unknown record type: {}", bounded(other))); - } + } + "compacted" => { + let text = payload.and_then(|p| str_field(p, "message")); + session + .messages + .push(compaction_marker(text, seq, ts, line_start)); + contexts.push((cwd.clone(), model.clone())); + seq += 1; + } + // Known non-conversation record types. + "world_state" | "inter_agent_communication_metadata" => {} + other => { + session.note_partial(format!("unknown record type: {}", bounded(other))); } } - - assign_segments(&mut session, &contexts, provider.as_ref(), &git); - resolve_file_event_segments(&mut session); - // Codex has no native title event, so any derived title is synthetic. - session.title = fallback_title(&session.messages); - session.title_is_synthetic = session.title.is_some(); - session } + + assign_segments(&mut session, &contexts, from, provider.as_ref(), &git); + resolve_file_event_segments(&mut session); + let end = ParseEnd { + offset, + next_seq: seq, + earlier_results, + cwd, + model, + restates_session, + }; + finish_parse(from, session, end, archived) } /// Derive file events from a `patch_apply_end.changes` object (keyed by path). @@ -517,22 +542,25 @@ fn push_reasoning_text(value: Option<&Value>, parts: &mut Vec, ord: } } -/// Group consecutive messages by (cwd, model). Recorded git attaches to the -/// first segment only (it is stamped once at session start), and provider to all. +/// Group consecutive messages by (cwd, model), continuing the segment `from` +/// left open. Recorded git attaches to the first segment only (it is stamped +/// once at session start), and provider to all. fn assign_segments( session: &mut ParsedSession, contexts: &[(Option, Option)], + from: &ParseCheckpoint, provider: Option<&String>, git: &RecordedGit, ) { - session.segments.clear(); - let mut current: Option<(Option, Option)> = None; + session.segments = from.open_segment.iter().cloned().collect(); + let base = from.segment_count.saturating_sub(session.segments.len()); + let mut current = from + .open_segment + .as_ref() + .map(|s| (s.cwd.clone(), s.model.clone())); for (i, ctx) in contexts.iter().enumerate() { let seq = session.messages[i].seq; if current.as_ref() != Some(ctx) { - if let Some(last) = session.segments.last_mut() { - last.seq_end = session.messages[i.saturating_sub(1)].seq; - } let first = session.segments.is_empty(); session.segments.push(ParsedSegment { seq_start: seq, @@ -546,8 +574,7 @@ fn assign_segments( }); current = Some(ctx.clone()); } - let ix = session.segments.len() - 1; - session.messages[i].segment_ix = ix; + session.messages[i].segment_ix = base + session.segments.len() - 1; if let Some(last) = session.segments.last_mut() { last.seq_end = seq; } @@ -620,22 +647,25 @@ fn register_tool_call( id_map.insert(call_id, idx); } -/// Attach a tool output to its call by `call_id`. +/// Attach a tool output to its call by `call_id`. A call parsed before this run +/// gets the output back as a result for the caller to record. fn attach_tool_output( p: &Value, seq: i64, output_text: Option, session: &mut ParsedSession, id_map: &HashMap, -) { + archived: &dyn ArchivedSession, +) -> Option { let Some(call_id) = str_field(p, "call_id") else { session.note_partial("tool output without call_id"); - return; + return None; }; - let Some(&idx) = id_map.get(&call_id) else { + let call = id_map.get(&call_id).copied(); + if call.is_none() && !archived.has_tool_call(&call_id) { session.note_partial("tool output without matching call"); - return; - }; + return None; + } let is_error = str_field(p, "status") .as_deref() .map(|s| !matches!(s, "completed" | "success" | "ok")) @@ -652,10 +682,19 @@ fn attach_tool_output( } }) }); + let Some(idx) = call else { + return Some(ParsedToolResult { + native_call_id: call_id, + result_ref: (seq, 0), + output_text, + is_error, + }); + }; let tc = &mut session.tool_calls[idx]; tc.result_ref = Some((seq, 0)); tc.output_text = output_text; tc.is_error = is_error; + None } /// Extract text from a JSON value that may be a string or a structured object. @@ -740,6 +779,30 @@ impl AgentAdapter for CodexAdapter { fn parse_content(&self, content: &str, fallback_dedupe: &str) -> ParsedSession { self.parse_str(content, fallback_dedupe) } + + fn parse_checkpointed( + &self, + content: &str, + fallback_dedupe: &str, + ) -> (ParsedSession, Option) { + parse_from( + &ParseCheckpoint::default(), + content, + fallback_dedupe, + &NothingArchived, + ) + .checkpointed() + } + + fn parse_append( + &self, + from: &ParseCheckpoint, + tail: &str, + fallback_dedupe: &str, + archived: &dyn ArchivedSession, + ) -> Option { + parse_from(from, tail, fallback_dedupe, archived).append() + } } /// Recursively collect `rollout-*.jsonl` files up to `depth` levels below `dir`. diff --git a/crates/lore-core/src/adapters/common.rs b/crates/lore-core/src/adapters/common.rs index 6f08a2a..af478ea 100644 --- a/crates/lore-core/src/adapters/common.rs +++ b/crates/lore-core/src/adapters/common.rs @@ -1,9 +1,179 @@ //! Small parsing helpers shared by adapters. -use crate::model::{ParsedMessage, PartKind, Role}; +use std::collections::{BTreeSet, HashSet}; + +use super::ArchivedSession; +use crate::model::{ + ParseCheckpoint, ParseStatus, ParsedAppend, ParsedMessage, ParsedSession, ParsedToolResult, + PartKind, Role, Tokens, +}; const TITLE_MAX_CHARS: usize = 80; +/// A session holding the session-level values `from` recorded and no rows, for +/// a parse that continues there. Its `title` is the native title so far. +pub(crate) fn resume_session(from: &ParseCheckpoint, fallback_dedupe: &str) -> ParsedSession { + let mut session = ParsedSession::new(fallback_dedupe); + session.native_session_id = from.native_session_id.clone(); + session.agent_version = from.agent_version.clone(); + session.primary_model = from.primary_model.clone(); + session.title = from.native_title.clone(); + session.started_at = from.started_at; + session.ended_at = from.ended_at; + session.total_tokens = from.total_tokens; + if from.note_count > 0 { + session.status = ParseStatus::Partial; + } + session +} + +/// Where a parse that started at `from` ended, beyond the session rows. +pub(crate) struct ParseEnd { + pub offset: i64, + pub next_seq: i64, + pub earlier_results: Vec, + pub cwd: Option, + pub model: Option, + /// The lines restate session-level context that earlier rows carry. + pub restates_session: bool, +} + +/// A parse from a checkpoint. `appendable` is false when the lines change +/// rows that earlier lines produced, so they cannot be persisted on their own. +pub(crate) struct ParseRun { + pub parsed: ParsedAppend, + pub appendable: bool, +} + +impl ParseRun { + /// The checkpoint a later append resumes from. A session without messages + /// has no segment to carry context forward, so it is parsed whole again. + pub(crate) fn checkpointed(self) -> (ParsedSession, Option) { + let checkpoint = self.parsed.checkpoint; + let resumable = checkpoint.open_segment.is_some().then_some(checkpoint); + (self.parsed.session, resumable) + } + + pub(crate) fn append(self) -> Option { + self.appendable.then_some(self.parsed) + } +} + +/// Settle the title and record where the next append resumes. `session.title` +/// holds the native title on entry. +pub(crate) fn finish_parse( + from: &ParseCheckpoint, + mut session: ParsedSession, + end: ParseEnd, + archived: &dyn ArchivedSession, +) -> ParseRun { + let uuids: HashSet<&str> = session + .messages + .iter() + .filter_map(|m| m.native_uuid.as_deref()) + .collect(); + let call_ids: HashSet<&str> = session + .tool_calls + .iter() + .map(|t| t.native_call_id.as_str()) + .collect(); + // A line that resolves what an earlier line left open changes that + // earlier line's row. + let resolves_earlier = from + .unresolved_parents + .iter() + .any(|u| uuids.contains(u.as_str())) + || from + .unresolved_calls + .iter() + .any(|c| call_ids.contains(c.as_str())); + let appendable = !resolves_earlier && !end.restates_session; + + let mut unresolved_parents: BTreeSet = + from.unresolved_parents.iter().cloned().collect(); + for parent in session + .messages + .iter() + .filter_map(|m| m.parent_native_uuid.as_deref()) + { + if !uuids.contains(parent) && !archived.has_message(parent) { + unresolved_parents.insert(parent.to_string()); + } + } + let mut unresolved_calls: BTreeSet = from.unresolved_calls.iter().cloned().collect(); + for call in session + .file_events + .iter() + .filter_map(|f| f.tool_native_call_id.as_deref()) + { + if !call_ids.contains(call) && !archived.has_tool_call(call) { + unresolved_calls.insert(call.to_string()); + } + } + + let native_title = session.title.take(); + let derived_title = from + .derived_title + .clone() + .or_else(|| fallback_title(&session.messages)); + session.title_is_synthetic = native_title.is_none() && derived_title.is_some(); + session.title = native_title.clone().or_else(|| derived_title.clone()); + + let base = from + .segment_count + .saturating_sub(usize::from(from.open_segment.is_some())); + let checkpoint = ParseCheckpoint { + offset: end.offset, + next_seq: end.next_seq, + segment_count: base + session.segments.len(), + tool_call_count: from.tool_call_count + session.tool_calls.len(), + file_event_count: from.file_event_count + session.file_events.len(), + native_session_id: session.native_session_id.clone(), + agent_version: session.agent_version.clone(), + primary_model: session.primary_model.clone(), + native_title, + derived_title, + started_at: session.started_at, + ended_at: session.ended_at, + total_tokens: session.total_tokens, + message_tokens: session + .messages + .iter() + .fold(from.message_tokens, |sum, m| add_tokens(sum, m.tokens)), + note_count: from.note_count + session.notes.len(), + first_note: from + .first_note + .clone() + .or_else(|| session.notes.first().map(|n| n.message.clone())), + open_segment: session.segments.last().cloned(), + cwd: end.cwd, + model: end.model, + unresolved_parents: unresolved_parents.into_iter().collect(), + unresolved_calls: unresolved_calls.into_iter().collect(), + }; + ParseRun { + parsed: ParsedAppend { + session, + earlier_results: end.earlier_results, + checkpoint, + }, + appendable, + } +} + +/// Per-field saturating sum; a field stays absent until some message has it. +fn add_tokens(sum: Tokens, more: Tokens) -> Tokens { + let add = |a: Option, b: Option| match (a, b) { + (None, None) => None, + (a, b) => Some(a.unwrap_or(0).saturating_add(b.unwrap_or(0))), + }; + Tokens { + input: add(sum.input, more.input), + output: add(sum.output, more.output), + cache: add(sum.cache, more.cache), + } +} + /// Derive a compact title from the first meaningful user request when the /// native log does not provide one. Agent/runtime bootstrap messages are /// deliberately skipped so permissions and injected repository instructions diff --git a/crates/lore-core/src/ingest.rs b/crates/lore-core/src/ingest.rs index 2f36328..5ae3613 100644 --- a/crates/lore-core/src/ingest.rs +++ b/crates/lore-core/src/ingest.rs @@ -13,13 +13,14 @@ //! (`SECRET_SCANNING.md`, `SEARCH.md`). Recorded-patch blobs finalize their //! `scan_state` here. -use std::collections::{HashMap, HashSet}; +use std::cell::Cell; +use std::collections::HashMap; use std::path::{Path, PathBuf}; use rusqlite::{params, Connection, OptionalExtension}; -use crate::adapters::{AdapterRegistry, AgentAdapter, SessionRef}; -use crate::model::ParsedSession; +use crate::adapters::{AdapterRegistry, AgentAdapter, ArchivedSession, SessionRef}; +use crate::model::{ParseCheckpoint, ParsedAppend, ParsedSession}; use crate::storage::blob::{BlobStore, StagedBlob}; use crate::storage::{Result, StorageError}; @@ -153,19 +154,38 @@ fn persist_rows( ], )?; - // A native title (custom/ai) is distinct content, so it is scanned and - // indexed like any cleartext. A synthetic fallback title merely echoes the - // first user message — already scanned and indexed — so re-scanning or - // re-indexing it would duplicate both secret findings and search hits - // (`SEARCH.md` §6). It is kept only in `agent_session.title` for display. + project_title(tx, &session_id, agent_id, parsed)?; + insert_rows( + tx, + &session_id, + agent_id, + parsed, + source_artifact_id, + patch_blobs, + RowBase::default(), + )?; + Ok(session_id) +} + +/// A native title (custom/ai) is distinct content, so it is scanned and +/// indexed like any cleartext. A synthetic fallback title merely echoes the +/// first user message — already scanned and indexed — so re-scanning or +/// re-indexing it would duplicate both secret findings and search hits +/// (`SEARCH.md` §6). It is kept only in `agent_session.title` for display. +fn project_title( + tx: &Connection, + session_id: &str, + agent_id: &str, + parsed: &ParsedSession, +) -> Result<()> { if let Some(title) = &parsed.title { if !parsed.title_is_synthetic { scan_and_project( tx, - &session_id, + session_id, None, "session", - &session_id, + session_id, "title", title, true, @@ -174,18 +194,43 @@ fn persist_rows( )?; } } + Ok(()) +} + +/// How many segments and file events the session already holds below the +/// rows being inserted. Zero for a whole session. +#[derive(Debug, Clone, Copy, Default)] +struct RowBase { + segments: usize, + file_events: usize, +} + +/// Insert the rows `parsed` carries. `base` says where its segments and file +/// events start among the session's. A reference to a message or tool call +/// that `parsed` lacks resolves against rows already in the archive. +fn insert_rows( + tx: &Connection, + session_id: &str, + agent_id: &str, + parsed: &ParsedSession, + source_artifact_id: Option<&str>, + patch_blobs: &[Option], + base: RowBase, +) -> Result<()> { + let segment_count = base.segments + parsed.segments.len(); + let segment_id = + |ix: usize| (ix < segment_count).then(|| det_id("seg", &[session_id, &ix.to_string()])); - // Segments. - let mut segment_ids = Vec::with_capacity(parsed.segments.len()); + // Segments. The first may be the open segment an append extends. for (i, seg) in parsed.segments.iter().enumerate() { - let sid = det_id("seg", &[&session_id, &i.to_string()]); tx.execute( "INSERT INTO session_segment (id, session_id, seq_start, seq_end, cwd, model, provider, context_source, resolution_confidence) - VALUES (?1,?2,?3,?4,?5,?6,?7,'event','unresolved')", + VALUES (?1,?2,?3,?4,?5,?6,?7,'event','unresolved') + ON CONFLICT(id) DO UPDATE SET seq_end = excluded.seq_end", params![ - sid, + segment_id(base.segments + i), session_id, seg.seq_start, seg.seq_end, @@ -194,14 +239,13 @@ fn persist_rows( seg.provider ], )?; - segment_ids.push(sid); } // Pre-assign message ids and build the uuid->id map for parent resolution. let mut msg_ids = Vec::with_capacity(parsed.messages.len()); let mut uuid_to_id: HashMap = HashMap::new(); for m in &parsed.messages { - let mid = det_id("m", &[&session_id, &m.seq.to_string()]); + let mid = det_id("m", &[session_id, &m.seq.to_string()]); if let Some(u) = &m.native_uuid { uuid_to_id.insert(u.clone(), mid.clone()); } @@ -212,12 +256,14 @@ fn persist_rows( let mut part_ids: HashMap<(i64, i64), String> = HashMap::new(); for (mi, m) in parsed.messages.iter().enumerate() { let mid = &msg_ids[mi]; - let segment_id = segment_ids.get(m.segment_ix); - let parent_id = m - .parent_native_uuid - .as_ref() - .and_then(|u| uuid_to_id.get(u)) - .cloned(); + let segment_id = segment_id(m.segment_ix); + let parent_id = match &m.parent_native_uuid { + Some(u) => match uuid_to_id.get(u) { + Some(id) => Some(id.clone()), + None => archived_message_id(tx, session_id, u)?, + }, + None => None, + }; tx.execute( "INSERT INTO message (id, session_id, segment_id, native_uuid, parent_id, parent_native_uuid, seq, @@ -266,11 +312,11 @@ fn persist_rows( // Scan every cleartext part for secrets. Index the text projection // only when the part is searchable (thinking is scanned, not // indexed); tool payload JSON is scanned but never indexed. - let seg = segment_id.map(String::as_str); + let seg = segment_id.as_deref(); if let Some(text) = &p.text { scan_and_project( tx, - &session_id, + session_id, seg, "message_part", &pid, @@ -284,7 +330,7 @@ fn persist_rows( if let Some(content) = &p.content_json { scan_and_project( tx, - &session_id, + session_id, seg, "message_part", &pid, @@ -300,13 +346,12 @@ fn persist_rows( } // Tool calls (the invocation part must exist to satisfy the NOT NULL FK). - let mut persisted_tool_call_ids = HashSet::new(); for tc in &parsed.tool_calls { let Some(call_part_id) = part_ids.get(&tc.call_ref) else { continue; }; let result_part_id = tc.result_ref.as_ref().and_then(|r| part_ids.get(r)); - let tid = det_id("t", &[&session_id, &tc.native_call_id]); + let tid = det_id("t", &[session_id, &tc.native_call_id]); tx.execute( "INSERT INTO tool_call (id, session_id, call_part_id, result_part_id, native_call_id, name, @@ -325,19 +370,22 @@ fn persist_rows( tc.duration_ms, ], )?; - persisted_tool_call_ids.insert(tc.native_call_id.as_str()); } // File events. A recorded patch payload is referenced from its staged blob // (byte-faithful; the diff itself lives in the content-addressed store). for (i, fe) in parsed.file_events.iter().enumerate() { - let fid = det_id("f", &[&session_id, &i.to_string()]); - let segment_id = segment_ids.get(fe.segment_ix); - let tool_call_id = fe - .tool_native_call_id - .as_ref() - .filter(|call_id| persisted_tool_call_ids.contains(call_id.as_str())) - .map(|c| det_id("t", &[&session_id, c])); + let fid = det_id("f", &[session_id, &(base.file_events + i).to_string()]); + // An event belongs to its call's segment. The call is in the archive + // whether this pass inserted it or an earlier ingest did. + let call = match &fe.tool_native_call_id { + Some(native) => archived_call(tx, &det_id("t", &[session_id, native]))?, + None => None, + }; + let (tool_call_id, segment_id) = match call { + Some((id, segment_id)) => (Some(id), segment_id), + None => (None, segment_id(fe.segment_ix)), + }; let patch_blob_id = match patch_blobs.get(i).and_then(Option::as_ref) { Some(staged) => Some(BlobStore::reference(tx, staged, PATCH_MEDIA_TYPE)?), None => None, @@ -393,11 +441,10 @@ fn persist_rows( // failure quarantines the patch: no projection, blob marked // `failed_quarantined` (SECRET_SCANNING.md §6). if let Some(patch) = &fe.patch_text { - let seg = segment_id.map(String::as_str); let outcome = scan_and_project( tx, - &session_id, - seg, + session_id, + segment_id.as_deref(), "file_event", &fid, "patch", @@ -422,11 +469,13 @@ fn persist_rows( // Agent-recorded git observations from segment context (branch/commit as the // agent stamped them; never labeled a captured session-time snapshot). + // An open segment an append extends already has its observation. for (i, seg) in parsed.segments.iter().enumerate() { if seg.git_branch.is_none() && seg.git_commit_sha.is_none() { continue; } - let gid = det_id("g", &[&session_id, &i.to_string()]); + let ix = (base.segments + i).to_string(); + let gid = det_id("g", &[session_id, &ix]); // Store only the credential-free normalized remote; a recorded URL may // carry an embedded token and must never be persisted verbatim. let remote_url_norm = seg @@ -437,11 +486,12 @@ fn persist_rows( "INSERT INTO git_observation (id, session_id, segment_id, source, observed_at, temporal_confidence, branch, commit_sha, remote_url_norm) - VALUES (?1,?2,?3,'agent_recorded', unixepoch('now')*1000, 'near_event', ?4,?5,?6)", + VALUES (?1,?2,?3,'agent_recorded', unixepoch('now')*1000, 'near_event', ?4,?5,?6) + ON CONFLICT(id) DO NOTHING", params![ gid, session_id, - segment_ids[i], + det_id("seg", &[session_id, &ix]), seg.git_branch, seg.git_commit_sha, remote_url_norm, @@ -453,9 +503,151 @@ fn persist_rows( // projects (migration 0011). Enrichment refreshes it again once repository // identity is resolved; this pass makes agent-recorded branches and commits // searchable even for a session Lore never manages to enrich. - crate::search::project_session_git(tx, &session_id)?; + crate::search::project_session_git(tx, session_id)?; + Ok(()) +} - Ok(session_id) +/// The id of a message already in the archive, by its native uuid. +fn archived_message_id(tx: &Connection, session_id: &str, uuid: &str) -> Result> { + Ok(tx + .prepare_cached("SELECT id FROM message WHERE session_id = ?1 AND native_uuid = ?2")? + .query_row(params![session_id, uuid], |r| r.get(0)) + .optional()?) +} + +/// Persist lines appended after `from`: add their rows, fill in results for +/// calls parsed before them, and bring the session's aggregates to the new end. +fn append_rows( + tx: &Connection, + session_id: &str, + agent_id: &str, + from: &ParseCheckpoint, + append: &ParsedAppend, + source_artifact_id: &str, + patch_blobs: &[Option], +) -> Result<()> { + tx.execute_batch("PRAGMA defer_foreign_keys = ON;")?; + let parsed = &append.session; + let end = &append.checkpoint; + tx.execute( + "UPDATE agent_session SET + title = ?2, ended_at = ?3, primary_model = ?4, message_count = ?5, + tool_call_count = ?6, total_input_tokens = ?7, total_output_tokens = ?8, + total_cache_tokens = ?9, parse_status = ?10, parse_note = ?11, agent_version = ?12 + WHERE id = ?1", + params![ + session_id, + parsed.title, + parsed.ended_at, + parsed.primary_model, + end.next_seq, + i64::try_from(end.tool_call_count).unwrap_or(i64::MAX), + end.total_tokens.input.or(end.message_tokens.input), + end.total_tokens.output.or(end.message_tokens.output), + end.total_tokens.cache.or(end.message_tokens.cache), + parsed.status.as_str(), + note_summary(end.note_count, end.first_note.as_deref()), + parsed.agent_version, + ], + )?; + + // Only a native title is projected, so only a new one changes the index. + if from.native_title != end.native_title { + tx.execute( + "DELETE FROM search_document + WHERE session_id = ?1 AND source_kind = 'session' AND source_id = ?1 + AND field = 'title'", + [session_id], + )?; + tx.execute( + "DELETE FROM secret_finding + WHERE source_kind = 'session' AND source_id = ?1 AND field = 'title'", + [session_id], + )?; + project_title(tx, session_id, agent_id, parsed)?; + } + + insert_rows( + tx, + session_id, + agent_id, + parsed, + Some(source_artifact_id), + patch_blobs, + RowBase { + segments: end.segment_count.saturating_sub(parsed.segments.len()), + file_events: from.file_event_count, + }, + )?; + + for result in &append.earlier_results { + let (seq, ordinal) = result.result_ref; + let message_id = det_id("m", &[session_id, &seq.to_string()]); + tx.execute( + "UPDATE tool_call SET result_part_id = ?2, output_text = ?3, is_error = ?4 + WHERE id = ?1", + params![ + det_id("t", &[session_id, &result.native_call_id]), + det_id("p", &[&message_id, &ordinal.to_string()]), + result.output_text, + result.is_error, + ], + )?; + } + + tx.execute( + "UPDATE session_source SET last_seq = COALESCE(?3, last_seq) + WHERE session_id = ?1 AND source_artifact_id = ?2", + params![ + session_id, + source_artifact_id, + parsed.messages.last().map(|m| m.seq) + ], + )?; + Ok(()) +} + +/// Rows a session already has, answered from the archive. A query error marks +/// the answers unreliable so the caller parses the whole file instead. +struct ArchivedRows<'a> { + conn: &'a Connection, + session_id: &'a str, + failed: Cell, +} + +impl ArchivedRows<'_> { + fn found(&self, lookup: Result>) -> bool { + lookup + .unwrap_or_else(|_| { + self.failed.set(true); + None + }) + .is_some() + } +} + +impl ArchivedSession for ArchivedRows<'_> { + fn has_message(&self, native_uuid: &str) -> bool { + self.found(archived_message_id(self.conn, self.session_id, native_uuid)) + } + + fn has_tool_call(&self, native_call_id: &str) -> bool { + let id = det_id("t", &[self.session_id, native_call_id]); + self.found(archived_call(self.conn, &id)) + } +} + +/// An archived tool call's id and the segment of the message that invoked it. +fn archived_call(tx: &Connection, call_id: &str) -> Result)>> { + Ok(tx + .prepare_cached( + "SELECT t.id, m.segment_id FROM tool_call t + JOIN message_part p ON p.id = t.call_part_id + JOIN message m ON m.id = p.message_id + WHERE t.id = ?1", + )? + .query_row([call_id], |r| Ok((r.get(0)?, r.get(1)?))) + .optional()?) } fn session_token_totals(parsed: &ParsedSession) -> crate::model::Tokens { @@ -495,13 +687,15 @@ fn parse_note(parsed: &ParsedSession) -> Option { if let Some(n) = &parsed.note { return Some(bounded(n)); } - match parsed.notes.len() { - 0 => None, - n => Some(bounded(&format!( - "{n} parser note(s); first: {}", - parsed.notes[0].message - ))), - } + note_summary( + parsed.notes.len(), + parsed.notes.first().map(|n| n.message.as_str()), + ) +} + +fn note_summary(count: usize, first: Option<&str>) -> Option { + let first = first.filter(|_| count > 0)?; + Some(bounded(&format!("{count} parser note(s); first: {first}"))) } fn bounded(s: &str) -> String { @@ -695,6 +889,7 @@ struct SourceRow { generation: i64, parser_version: Option, ingest_state: Option, + resume_json: Option, } struct SourceSnapshot { @@ -706,6 +901,9 @@ struct SourceSnapshot { prefix_hash: String, last_offset: i64, last_line: i64, + /// `full_hash` of the bytes the last ingest saw, when the file still has + /// that many. Equal to that ingest's `full_hash` only if they are unchanged. + hash_through_previous_size: Option, } struct SessionSourceRow { @@ -722,10 +920,13 @@ pub const MAX_SOURCE_BYTES: u64 = 64 * 1024 * 1024; /// /// Idempotent by construction: a changed file is fully re-parsed and its session /// rows are replaced, so append / truncate / rewrite all converge to the correct -/// state. Classification (via size + a prefix hash) drives generation bumps and -/// lets an unchanged file be skipped without re-parsing. Source-artifact -/// tracking, the session rows, the session↔source link, and the checkpoint all -/// commit in one transaction. +/// state. The exception is an append whose earlier bytes are provably the ones +/// the last ingest parsed: only the new lines are parsed and only their rows +/// are added, from the parser checkpoint that ingest stored. Classification +/// (via size + a prefix hash) drives generation bumps and lets an unchanged +/// file be skipped without re-parsing. Source-artifact tracking, the session +/// rows, the session↔source link, and the checkpoint all commit in one +/// transaction. pub fn ingest_file( conn: &Connection, adapter: &dyn AgentAdapter, @@ -734,6 +935,9 @@ pub fn ingest_file( ) -> Result { let meta = std::fs::metadata(path).map_err(|_| StorageError::Io)?; let is_oversized = meta.len() > MAX_SOURCE_BYTES; + let agent_id = adapter.id().0; + let path_str = path.to_string_lossy(); + let by_path = find_source_by_path(conn, agent_id, &path_str)?; let (content, snapshot, oversized_note) = if is_oversized { // Parse a bounded prefix (cut at a complete line) so a multi-hundred-MiB @@ -746,19 +950,20 @@ pub fn ingest_file( (content, snap, Some(note)) } else { let content = std::fs::read_to_string(path).map_err(|_| StorageError::Io)?; - let snap = source_snapshot(path, &meta, &content); + let previous_size = by_path.as_ref().and_then(|r| usize::try_from(r.size).ok()); + let snap = source_snapshot(path, &meta, &content, previous_size); (content, snap, None) }; - let agent_id = adapter.id().0; - - let prev = find_source( - conn, - agent_id, - snapshot.native_file_id.as_deref(), - &snapshot.path, - &snapshot.full_hash, - )?; + let prev = match by_path { + Some(row) => Some(row), + None => find_source_by_native_id( + conn, + agent_id, + snapshot.native_file_id.as_deref(), + &snapshot.full_hash, + )?, + }; let (artifact_id, generation, change) = match &prev { None => ( @@ -812,10 +1017,28 @@ pub fn ingest_file( .map(|s| s.to_string_lossy().into_owned()) .unwrap_or_else(|| snapshot.path.clone()); - let mut parsed = adapter.parse_content(&content, &fallback); + if let (Some(row), ChangeClass::Appended, None) = (&prev, change, &oversized_note) { + let source = AppendSource { + row, + content: &content, + snapshot: &snapshot, + fallback: &fallback, + }; + if let Some(session_id) = ingest_append(conn, adapter, &source, blobs)? { + return Ok(IngestOutcome::Ingested { + session_id, + source_artifact_id: artifact_id, + generation, + change, + }); + } + } + + let (mut parsed, checkpoint) = adapter.parse_checkpointed(&content, &fallback); if let Some(note) = oversized_note { parsed.note_partial(note); } + let resume = checkpoint.filter(|_| ends_on_line(&snapshot)); // Stage blob files before opening the transaction (DATA_MODEL.md §3). let patch_blobs = stage_patch_blobs(blobs, &parsed)?; @@ -857,33 +1080,154 @@ pub fn ingest_file( parsed.messages.last().map(|m| m.seq), ], )?; + write_ingest_state(&tx, &artifact_id, generation, &snapshot, resume.as_ref())?; + tx.commit()?; + + Ok(IngestOutcome::Ingested { + session_id, + source_artifact_id: artifact_id, + generation, + change, + }) +} + +/// An appended source and what the last ingest recorded for it. +struct AppendSource<'a> { + row: &'a SourceRow, + content: &'a str, + snapshot: &'a SourceSnapshot, + fallback: &'a str, +} + +/// Parse and persist only the lines appended since the last ingest. Returns +/// the session id, or `None` when the archive does not provably hold what that +/// ingest's checkpoint describes, or the new lines change earlier rows; the +/// caller then parses the whole file. +fn ingest_append( + conn: &Connection, + adapter: &dyn AgentAdapter, + source: &AppendSource<'_>, + blobs: &BlobStore, +) -> Result> { + let row = source.row; + let Some(from) = row + .resume_json + .as_deref() + .and_then(ParseCheckpoint::from_json) + else { + return Ok(None); + }; + // The checkpoint must sit at the end of exactly the bytes it parsed, and + // those bytes must be unchanged, not just their 4 KiB prefix. + let parsed_bytes_unchanged = row.parser_version.as_deref() == Some(PARSER_VERSION) + && row.ingest_state.as_deref() == Some("ok") + && from.offset == row.size + && row.full_hash.is_some() + && row.full_hash == source.snapshot.hash_through_previous_size; + let tail = usize::try_from(row.size) + .ok() + .and_then(|n| source.content.get(n..)); + let (true, Some(tail)) = (parsed_bytes_unchanged, tail) else { + return Ok(None); + }; + let Some(session_id) = sole_session(conn, &row.id, from.next_seq)? else { + return Ok(None); + }; + + let archived = ArchivedRows { + conn, + session_id: &session_id, + failed: Cell::new(false), + }; + let Some(append) = adapter.parse_append(&from, tail, source.fallback, &archived) else { + return Ok(None); + }; + // Search rows carry the session start, and the session id names the rows. + let agent_id = adapter.id().0; + if archived.failed.get() + || session_id_for(agent_id, &append.session) != session_id + || append.session.started_at != from.started_at + { + return Ok(None); + } + + let patch_blobs = stage_patch_blobs(blobs, &append.session)?; + let _write = crate::storage::write_lock(); + let tx = conn.unchecked_transaction()?; + append_rows( + &tx, + &session_id, + agent_id, + &from, + &append, + &row.id, + &patch_blobs, + )?; + update_source(&tx, &row.id, source.snapshot, row.generation)?; + add_path_alias(&tx, &row.id, &source.snapshot.path)?; + let resume = ends_on_line(source.snapshot).then_some(&append.checkpoint); + write_ingest_state(&tx, &row.id, row.generation, source.snapshot, resume)?; + tx.commit()?; + Ok(Some(session_id)) +} + +/// The session `source_artifact_id` alone feeds, when it holds `message_count` +/// messages. A session another source also feeds may hold that source's rows. +fn sole_session( + conn: &Connection, + source_artifact_id: &str, + message_count: i64, +) -> Result> { + let mut statement = conn.prepare( + "SELECT s.id FROM session_source ss + JOIN agent_session s ON s.id = ss.session_id + WHERE ss.source_artifact_id = ?1 AND s.message_count = ?2 + AND (SELECT count(*) FROM session_source o WHERE o.session_id = s.id) = 1", + )?; + let mut sessions = statement + .query_map(params![source_artifact_id, message_count], |r| r.get(0))? + .collect::>>()?; + Ok(if sessions.len() == 1 { + sessions.pop() + } else { + None + }) +} + +/// A checkpoint is only resumable at a line boundary: a partial last line is +/// parsed again whole once the agent finishes writing it. +fn ends_on_line(snapshot: &SourceSnapshot) -> bool { + snapshot.last_offset == snapshot.size +} + +fn write_ingest_state( + tx: &Connection, + source_artifact_id: &str, + generation: i64, + snapshot: &SourceSnapshot, + resume: Option<&ParseCheckpoint>, +) -> Result<()> { tx.execute( "INSERT INTO ingest_state (source_artifact_id, generation, last_offset, last_line, prefix_hash, - parser_version, state, updated_at) - VALUES (?1,?2,?3,?4,?5,?6,'ok', unixepoch('now')*1000) + parser_version, state, updated_at, resume_json) + VALUES (?1,?2,?3,?4,?5,?6,'ok', unixepoch('now')*1000, ?7) ON CONFLICT(source_artifact_id) DO UPDATE SET generation = excluded.generation, last_offset = excluded.last_offset, last_line = excluded.last_line, prefix_hash = excluded.prefix_hash, parser_version = excluded.parser_version, state = 'ok', - updated_at = excluded.updated_at", + updated_at = excluded.updated_at, resume_json = excluded.resume_json", params![ - artifact_id, + source_artifact_id, generation, snapshot.last_offset, snapshot.last_line, snapshot.prefix_hash, PARSER_VERSION, + resume.map(ParseCheckpoint::to_json), ], )?; - tx.commit()?; - - Ok(IngestOutcome::Ingested { - session_id, - source_artifact_id: artifact_id, - generation, - change, - }) + Ok(()) } fn session_id_for(agent_id: &str, parsed: &ParsedSession) -> String { @@ -949,51 +1293,52 @@ fn detach_reassigned_source( Ok(()) } -fn find_source( +fn source_row(r: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(SourceRow { + id: r.get(0)?, + size: r.get(1)?, + full_hash: r.get(2)?, + prefix_hash: r.get(3)?, + generation: r.get(4)?, + parser_version: r.get(5)?, + ingest_state: r.get(6)?, + resume_json: r.get(7)?, + }) +} + +fn find_source_by_path(conn: &Connection, agent_id: &str, path: &str) -> Result> { + conn.query_row( + "SELECT sa.id, sa.size, sa.full_hash, sa.prefix_hash, sa.generation, + ist.parser_version, ist.state, ist.resume_json + FROM source_artifact sa + LEFT JOIN ingest_state ist ON ist.source_artifact_id = sa.id + WHERE sa.agent_id = ?1 AND sa.current_path = ?2", + params![agent_id, path], + source_row, + ) + .optional() + .map_err(StorageError::from) +} + +/// A moved source: same file identity and content, new path. +fn find_source_by_native_id( conn: &Connection, agent_id: &str, native_file_id: Option<&str>, - path: &str, full_hash: &str, ) -> Result> { - let map = |r: &rusqlite::Row<'_>| { - Ok(SourceRow { - id: r.get(0)?, - size: r.get(1)?, - full_hash: r.get(2)?, - prefix_hash: r.get(3)?, - generation: r.get(4)?, - parser_version: r.get(5)?, - ingest_state: r.get(6)?, - }) - }; - let by_path = conn - .query_row( - "SELECT sa.id, sa.size, sa.full_hash, sa.prefix_hash, sa.generation, - ist.parser_version, ist.state - FROM source_artifact sa - LEFT JOIN ingest_state ist ON ist.source_artifact_id = sa.id - WHERE sa.agent_id = ?1 AND sa.current_path = ?2", - params![agent_id, path], - map, - ) - .optional()?; - if by_path.is_some() { - return Ok(by_path); - } - let Some(fid) = native_file_id else { return Ok(None); }; conn.query_row( "SELECT sa.id, sa.size, sa.full_hash, sa.prefix_hash, sa.generation, - ist.parser_version, ist.state + ist.parser_version, ist.state, ist.resume_json FROM source_artifact sa LEFT JOIN ingest_state ist ON ist.source_artifact_id = sa.id WHERE sa.agent_id = ?1 AND sa.native_file_id = ?2 AND sa.full_hash = ?3 ORDER BY sa.last_seen_at DESC LIMIT 1", params![agent_id, fid, full_hash], - map, + source_row, ) .optional() .map_err(StorageError::from) @@ -1131,34 +1476,54 @@ fn oversized_source_snapshot( prefix_hash, last_offset: 0, last_line: 0, + hash_through_previous_size: None, } } -fn source_snapshot(path: &Path, meta: &std::fs::Metadata, content: &str) -> SourceSnapshot { +fn source_snapshot( + path: &Path, + meta: &std::fs::Metadata, + content: &str, + previous_size: Option, +) -> SourceSnapshot { let bytes = content.as_bytes(); - let (full_hash, prefix_hash, last_offset, last_line) = fingerprint_and_checkpoint(bytes); + let (full_hash, prefix_hash, last_offset, last_line, hash_through_previous_size) = + fingerprint_and_checkpoint(bytes, previous_size); SourceSnapshot { path: path.to_string_lossy().into_owned(), native_file_id: file_id(meta), - size: i64::try_from(meta.len()).unwrap_or(i64::MAX), + // The bytes read, not the earlier stat: an agent can append between + // the two, and the size must describe what the hashes and the + // checkpoint cover. + size: i64::try_from(bytes.len()).unwrap_or(i64::MAX), mtime: mtime_ms(meta), full_hash, prefix_hash, last_offset, last_line, + hash_through_previous_size, } } /// Compute both source fingerprints and the complete-record checkpoint in one /// traversal. Changed multi-megabyte sessions are commonly append-only; avoid /// walking their full history separately for hashing and newline accounting. -fn fingerprint_and_checkpoint(bytes: &[u8]) -> (String, String, i64, i64) { +/// The same pass yields the full hash of the first `mark` bytes, which proves +/// an append left the earlier bytes alone. +fn fingerprint_and_checkpoint( + bytes: &[u8], + mark: Option, +) -> (String, String, i64, i64, Option) { let mut full = crate::hash::Fnv1a64::new(); let mut prefix = crate::hash::Fnv1a64::new(); let mut last_offset = 0usize; let mut complete_lines = 0usize; + let mut at_mark = None; for (index, byte) in bytes.iter().copied().enumerate() { + if Some(index) == mark { + at_mark = Some(full); + } full.update_byte(byte); if index < PREFIX_BYTES { prefix.update_byte(byte); @@ -1168,12 +1533,16 @@ fn fingerprint_and_checkpoint(bytes: &[u8]) -> (String, String, i64, i64) { complete_lines = complete_lines.saturating_add(1); } } + if mark == Some(bytes.len()) { + at_mark = Some(full); + } ( full.finish_hex(), prefix.finish_hex(), i64::try_from(last_offset).unwrap_or(i64::MAX), i64::try_from(complete_lines).unwrap_or(i64::MAX), + at_mark.map(crate::hash::Fnv1a64::finish_hex), ) } @@ -1218,11 +1587,32 @@ mod tests { #[test] fn combined_fingerprint_preserves_hashes_and_complete_line_checkpoint() { let bytes = b"first\nsecond\npartial"; - let (full, prefix, offset, lines) = fingerprint_and_checkpoint(bytes); + let (full, prefix, offset, lines, at_mark) = fingerprint_and_checkpoint(bytes, Some(6)); assert_eq!(full, fnv1a_hex(bytes)); assert_eq!(prefix, fnv1a_hex(&bytes[..bytes.len().min(PREFIX_BYTES)])); assert_eq!((offset, lines), (13, 2)); + assert_eq!(at_mark, Some(fnv1a_hex(b"first\n"))); + let whole = fingerprint_and_checkpoint(bytes, Some(bytes.len())).4; + assert_eq!(whole, Some(fnv1a_hex(bytes))); + assert_eq!(fingerprint_and_checkpoint(bytes, Some(99)).4, None); + } + // An agent appending between the stat and the read must not leave a size + // that disagrees with the hashed bytes: the checkpoint would be dropped + // and the next append would parse the whole file. + #[test] + fn snapshot_size_is_the_bytes_read_even_when_the_file_grew_after_stat() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("growing.jsonl"); + std::fs::write(&path, "{\"a\":1}\n").unwrap(); + let meta = std::fs::metadata(&path).unwrap(); + std::fs::write(&path, "{\"a\":1}\n{\"b\":2}\n").unwrap(); + let content = std::fs::read_to_string(&path).unwrap(); + + let snapshot = source_snapshot(&path, &meta, &content, None); + assert_eq!((snapshot.size, snapshot.last_offset), (16, 16)); + assert!(ends_on_line(&snapshot)); } + use crate::adapters::codex::CodexAdapter; use crate::model::{EventKind, ParsedMessage, Role, Tokens}; use crate::storage::blob::BlobStore; diff --git a/crates/lore-core/src/model.rs b/crates/lore-core/src/model.rs index f598250..1bcf281 100644 --- a/crates/lore-core/src/model.rs +++ b/crates/lore-core/src/model.rs @@ -353,10 +353,239 @@ impl ParsedSession { } } +/// A tool result whose call was parsed before the checkpoint an append resumed +/// from. Persisting it fills in that earlier call's result. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ParsedToolResult { + pub native_call_id: String, + pub result_ref: (i64, i64), + pub output_text: Option, + pub is_error: Option, +} + +/// A parser's state after the last line it consumed. Parsing appended lines +/// from a checkpoint yields what parsing the whole file would, so an append +/// costs the new lines rather than the history. +/// +/// Rows are not kept here; lines that refer back to earlier rows ask the +/// archive (`crate::adapters::ArchivedSession`). What no earlier line could +/// resolve is kept, because a later line that resolves it changes an earlier +/// row and the append cannot be parsed on its own. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct ParseCheckpoint { + /// Byte offset of the next line. + pub offset: i64, + pub next_seq: i64, + pub segment_count: usize, + pub tool_call_count: usize, + pub file_event_count: usize, + pub native_session_id: Option, + pub agent_version: Option, + pub primary_model: Option, + /// A title the agent recorded. Wins over `derived_title`. + pub native_title: Option, + /// The title derived from the first meaningful user request. + pub derived_title: Option, + pub started_at: Option, + pub ended_at: Option, + /// Cumulative totals the source recorded (`ParsedSession::total_tokens`). + pub total_tokens: Tokens, + /// Per-message usage summed so far, the fallback for absent totals. + pub message_tokens: Tokens, + pub note_count: usize, + pub first_note: Option, + /// The last segment. The next message extends it when its context matches. + pub open_segment: Option, + /// Turn context the next message inherits when the agent records it apart + /// from messages (Codex `turn_context`). + pub cwd: Option, + pub model: Option, + /// Parent uuids no parsed message carries. + pub unresolved_parents: Vec, + /// Tool call ids that file events name but no parsed call carries. + pub unresolved_calls: Vec, +} + +/// Lines parsed from a [`ParseCheckpoint`]. `session` carries the +/// session-level values as of the new end, and only the rows these lines add. +/// Its `segments` start at index `checkpoint.segment_count - +/// session.segments.len()`, re-emitting the open segment, and message +/// `segment_ix` values index the whole session. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ParsedAppend { + pub session: ParsedSession, + pub earlier_results: Vec, + pub checkpoint: ParseCheckpoint, +} + +const CHECKPOINT_FORMAT: i64 = 1; + +impl ParseCheckpoint { + #[must_use] + pub fn to_json(&self) -> String { + use serde_json::json; + let tokens = |t: Tokens| json!([t.input, t.output, t.cache]); + let segment = self.open_segment.as_ref().map(|s| { + json!({ + "seq_start": s.seq_start, "seq_end": s.seq_end, "cwd": s.cwd, + "model": s.model, "provider": s.provider, "git_branch": s.git_branch, + "git_commit_sha": s.git_commit_sha, "git_remote_url": s.git_remote_url, + }) + }); + json!({ + "format": CHECKPOINT_FORMAT, + "offset": self.offset, + "next_seq": self.next_seq, + "segment_count": self.segment_count, + "tool_call_count": self.tool_call_count, + "file_event_count": self.file_event_count, + "native_session_id": self.native_session_id, + "agent_version": self.agent_version, + "primary_model": self.primary_model, + "native_title": self.native_title, + "derived_title": self.derived_title, + "started_at": self.started_at, + "ended_at": self.ended_at, + "total_tokens": tokens(self.total_tokens), + "message_tokens": tokens(self.message_tokens), + "note_count": self.note_count, + "first_note": self.first_note, + "open_segment": segment, + "cwd": self.cwd, + "model": self.model, + "unresolved_parents": self.unresolved_parents, + "unresolved_calls": self.unresolved_calls, + }) + .to_string() + } + + /// `None` for anything [`Self::to_json`] of this build did not write; the + /// caller then parses the whole file. + #[must_use] + pub fn from_json(text: &str) -> Option { + use serde_json::Value; + let v: Value = serde_json::from_str(text).ok()?; + if v.get("format")?.as_i64()? != CHECKPOINT_FORMAT { + return None; + } + let int = |v: &Value, k: &str| v.get(k)?.as_i64(); + let count = |k: &str| usize::try_from(v.get(k)?.as_u64()?).ok(); + let opt_int = |v: &Value, k: &str| match v.get(k)? { + Value::Null => Some(None), + n => n.as_i64().map(Some), + }; + let opt_str = |v: &Value, k: &str| match v.get(k)? { + Value::Null => Some(None), + s => s.as_str().map(|s| Some(s.to_string())), + }; + let tokens = |k: &str| { + let t = v.get(k)?.as_array()?; + let field = |i: usize| match t.get(i)? { + Value::Null => Some(None), + n => n.as_i64().map(Some), + }; + Some(Tokens { + input: field(0)?, + output: field(1)?, + cache: field(2)?, + }) + }; + let strings = |k: &str| { + v.get(k)? + .as_array()? + .iter() + .map(|s| s.as_str().map(str::to_string)) + .collect::>>() + }; + let open_segment = match v.get("open_segment")? { + Value::Null => None, + s => Some(ParsedSegment { + seq_start: int(s, "seq_start")?, + seq_end: int(s, "seq_end")?, + cwd: opt_str(s, "cwd")?, + model: opt_str(s, "model")?, + provider: opt_str(s, "provider")?, + git_branch: opt_str(s, "git_branch")?, + git_commit_sha: opt_str(s, "git_commit_sha")?, + git_remote_url: opt_str(s, "git_remote_url")?, + }), + }; + Some(ParseCheckpoint { + offset: int(&v, "offset")?, + next_seq: int(&v, "next_seq")?, + segment_count: count("segment_count")?, + tool_call_count: count("tool_call_count")?, + file_event_count: count("file_event_count")?, + native_session_id: opt_str(&v, "native_session_id")?, + agent_version: opt_str(&v, "agent_version")?, + primary_model: opt_str(&v, "primary_model")?, + native_title: opt_str(&v, "native_title")?, + derived_title: opt_str(&v, "derived_title")?, + started_at: opt_int(&v, "started_at")?, + ended_at: opt_int(&v, "ended_at")?, + total_tokens: tokens("total_tokens")?, + message_tokens: tokens("message_tokens")?, + note_count: count("note_count")?, + first_note: opt_str(&v, "first_note")?, + open_segment, + cwd: opt_str(&v, "cwd")?, + model: opt_str(&v, "model")?, + unresolved_parents: strings("unresolved_parents")?, + unresolved_calls: strings("unresolved_calls")?, + }) + } +} + #[cfg(test)] mod tests { use super::*; + #[test] + fn checkpoint_json_round_trips() { + let checkpoint = ParseCheckpoint { + offset: 120, + next_seq: 3, + segment_count: 1, + tool_call_count: 2, + file_event_count: 1, + native_session_id: Some("s".into()), + agent_version: None, + primary_model: Some("m".into()), + native_title: None, + derived_title: Some("Fix the cache".into()), + started_at: Some(1), + ended_at: Some(9), + total_tokens: Tokens::default(), + message_tokens: Tokens { + input: Some(5), + output: None, + cache: Some(0), + }, + note_count: 1, + first_note: Some("unknown event".into()), + open_segment: Some(ParsedSegment { + seq_start: 0, + seq_end: 2, + cwd: Some("/repo".into()), + model: None, + provider: None, + git_branch: Some("main".into()), + git_commit_sha: None, + git_remote_url: None, + }), + cwd: None, + model: Some("gpt".into()), + unresolved_parents: vec!["p".into()], + unresolved_calls: vec![], + }; + let json = checkpoint.to_json(); + assert_eq!(ParseCheckpoint::from_json(&json), Some(checkpoint)); + assert_eq!( + ParseCheckpoint::from_json(&json.replace("\"format\":1", "\"format\":2")), + None + ); + } + #[test] fn enum_string_forms_match_schema() { assert_eq!(Role::Assistant.as_str(), "assistant"); diff --git a/crates/lore-core/src/storage/migrations.rs b/crates/lore-core/src/storage/migrations.rs index 2e65037..ff0b52b 100644 --- a/crates/lore-core/src/storage/migrations.rs +++ b/crates/lore-core/src/storage/migrations.rs @@ -83,6 +83,11 @@ const MIGRATIONS: &[Migration] = &[ name: "file_content_identity", sql: include_str!("../../migrations/0013_file_content_identity.sql"), }, + Migration { + version: 14, + name: "ingest_resume", + sql: include_str!("../../migrations/0014_ingest_resume.sql"), + }, ]; /// How many migrations exist. Tests assert against this rather than a literal diff --git a/crates/lore-core/tests/incremental_ingest.rs b/crates/lore-core/tests/incremental_ingest.rs new file mode 100644 index 0000000..4d8b649 --- /dev/null +++ b/crates/lore-core/tests/incremental_ingest.rs @@ -0,0 +1,562 @@ +//! An append is parsed and persisted from the last ingest's checkpoint. Every +//! row it leaves must equal what a fresh ingest of the same file writes, +//! except for row ids and timestamps. +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use std::collections::BTreeMap; + +use lore_core::adapters::claude_code::ClaudeCodeAdapter; +use lore_core::adapters::codex::CodexAdapter; +use lore_core::adapters::AgentAdapter; +use lore_core::ingest::{ingest_file, IngestOutcome}; +use lore_core::storage::blob::BlobStore; +use rusqlite::types::Value; +use rusqlite::Connection; +use serde_json::json; + +fn token() -> String { + format!("ghp{}", "_0123456789abcdefghijklmnopqrstuvwxyz") +} + +const TABLES: &[(&str, &str)] = &[ + ( + "agent_session", + "SELECT id, agent_id, native_session_id, dedupe_key, title, started_at, ended_at, + primary_model, message_count, tool_call_count, total_input_tokens, + total_output_tokens, total_cache_tokens, parse_status, parse_note, agent_version + FROM agent_session WHERE id = ?1", + ), + ( + "session_segment", + "SELECT id, seq_start, seq_end, cwd, model, provider, repository_id + FROM session_segment WHERE session_id = ?1 ORDER BY seq_start", + ), + ( + "message", + "SELECT id, segment_id, native_uuid, parent_id, parent_native_uuid, seq, role, + event_kind, is_sidechain, ts, model, token_input, token_output, token_cache, + stop_reason, source_artifact_id, source_offset, metadata_json + FROM message WHERE session_id = ?1 ORDER BY seq", + ), + ( + "message_part", + "SELECT p.id, p.message_id, p.ordinal, p.kind, p.text, p.content_json, p.searchable, + p.metadata_json + FROM message_part p JOIN message m ON m.id = p.message_id + WHERE m.session_id = ?1 ORDER BY m.seq, p.ordinal", + ), + ( + "tool_call", + "SELECT id, call_part_id, result_part_id, native_call_id, name, input_json, + output_text, is_error, duration_ms + FROM tool_call WHERE session_id = ?1 ORDER BY id", + ), + ( + "file_event", + "SELECT f.id, f.segment_id, f.tool_call_id, f.path, f.change_kind, f.old_path, + f.lines_added, f.lines_removed, f.patch_blob_id, b.scan_state, f.source, + f.event_ts, f.content_oid + FROM file_event f LEFT JOIN blob b ON b.id = f.patch_blob_id + WHERE f.session_id = ?1 ORDER BY f.id", + ), + ( + "secret_finding", + "SELECT id, source_kind, source_id, field, rule, span_start, span_end, severity, + value_fingerprint, disposition + FROM secret_finding WHERE session_id = ?1 ORDER BY id", + ), + ( + "search_document", + "SELECT segment_id, source_kind, source_id, field, ordinal, redacted_text, started_at, + agent_id + FROM search_document WHERE session_id = ?1 ORDER BY source_kind, source_id, field", + ), + ( + "git_observation", + "SELECT id, segment_id, source, branch, commit_sha, remote_url_norm + FROM git_observation WHERE session_id = ?1 ORDER BY id", + ), + ( + "search_git", + "SELECT segment_id, repository_id, source_class, branch, commit_sha + FROM search_git WHERE session_id = ?1 ORDER BY segment_id, source_class, branch", + ), + ( + "session_source", + "SELECT source_artifact_id, first_seq, last_seq FROM session_source + WHERE session_id = ?1", + ), + ( + "source", + "SELECT sa.size, sa.full_hash, sa.prefix_hash, sa.generation, i.last_offset, + i.last_line, i.prefix_hash, i.parser_version, i.state, i.resume_json + FROM session_source ss + JOIN source_artifact sa ON sa.id = ss.source_artifact_id + JOIN ingest_state i ON i.source_artifact_id = sa.id + WHERE ss.session_id = ?1", + ), +]; + +type Dump = BTreeMap<&'static str, Vec>>; + +fn dump(conn: &Connection, session_id: &str) -> Dump { + TABLES + .iter() + .map(|(name, sql)| { + let mut stmt = conn.prepare(sql).unwrap(); + let width = stmt.column_count(); + let rows = stmt + .query_map([session_id], |r| (0..width).map(|i| r.get(i)).collect()) + .unwrap() + .collect::>>>() + .unwrap(); + (*name, rows) + }) + .collect() +} + +fn message_rowids(conn: &Connection, session_id: &str) -> Vec<(i64, String)> { + let mut stmt = conn + .prepare("SELECT rowid, id FROM message WHERE session_id = ?1 ORDER BY seq") + .unwrap(); + let rows = stmt + .query_map([session_id], |r| Ok((r.get(0)?, r.get(1)?))) + .unwrap() + .collect::>>() + .unwrap(); + rows +} + +fn session_id(outcome: IngestOutcome) -> String { + match outcome { + IngestOutcome::Ingested { session_id, .. } => Some(session_id), + IngestOutcome::Skipped => None, + } + .expect("a changed file must be ingested") +} + +/// Grow a file through `cuts` (byte offsets into `content`), ingesting each +/// prefix into one archive and, separately, into a fresh one. Every step must +/// leave the same rows. Returns, per step after the first, whether the ingest +/// kept the session row (an append) rather than replacing the session, and the +/// grown archive with the session's id. +fn grow_and_compare( + adapter: &dyn AgentAdapter, + content: &str, + cuts: &[usize], +) -> (Vec, Connection, String) { + let dir = tempfile::tempdir().unwrap(); + let blobs = BlobStore::open(dir.path().join("blobs")).unwrap(); + let path = dir.path().join("session.jsonl"); + let live = lore_core::storage::open_in_memory().unwrap(); + live.execute_batch( + "CREATE TEMP TABLE replaced (session_id TEXT); + CREATE TEMP TRIGGER log_replace AFTER DELETE ON agent_session + BEGIN INSERT INTO replaced VALUES (old.id); END;", + ) + .unwrap(); + + let mut kept = Vec::new(); + let mut session = String::new(); + for (step, &cut) in cuts.iter().enumerate() { + std::fs::write(&path, &content[..cut]).unwrap(); + live.execute_batch("DELETE FROM replaced").unwrap(); + let id = session_id(ingest_file(&live, adapter, &path, &blobs).unwrap()); + if step > 0 { + let replaced: i64 = live + .query_row("SELECT count(*) FROM replaced", [], |r| r.get(0)) + .unwrap(); + kept.push(replaced == 0); + } + live.execute_batch("INSERT INTO search_fts(search_fts) VALUES('integrity-check');") + .unwrap(); + + let fresh = lore_core::storage::open_in_memory().unwrap(); + let fresh_id = session_id(ingest_file(&fresh, adapter, &path, &blobs).unwrap()); + assert_eq!(fresh_id, id); + assert_eq!( + dump(&live, &id), + dump(&fresh, &id), + "rows after step {step} (cut at byte {cut}) differ from a fresh ingest" + ); + session = id; + } + (kept, live, session) +} + +/// (title, message_count, tool_call_count, input, output, cache, parse_note) +type Summary = ( + String, + i64, + i64, + Option, + Option, + Option, + Option, +); + +fn summary(conn: &Connection, session_id: &str) -> Summary { + conn.query_row( + "SELECT title, message_count, tool_call_count, total_input_tokens, + total_output_tokens, total_cache_tokens, parse_note + FROM agent_session WHERE id = ?1", + [session_id], + |r| { + Ok(( + r.get(0)?, + r.get(1)?, + r.get(2)?, + r.get(3)?, + r.get(4)?, + r.get(5)?, + r.get(6)?, + )) + }, + ) + .unwrap() +} + +fn jsonl(lines: &[serde_json::Value]) -> String { + lines.iter().map(|l| format!("{l}\n")).collect() +} + +fn line_ends(content: &str) -> Vec { + content.match_indices('\n').map(|(i, _)| i + 1).collect() +} + +fn claude_session() -> String { + let sid = "cc-eq"; + let ctx = + |cwd: &str| json!({"sessionId": sid, "cwd": cwd, "gitBranch": "main", "version": "1.2.3"}); + let with = |base: serde_json::Value, extra: serde_json::Value| { + let mut v = base; + v.as_object_mut() + .unwrap() + .extend(extra.as_object().unwrap().clone()); + v + }; + jsonl(&[ + with( + ctx("/repo/app"), + json!({"type": "user", "uuid": "u1", "parentUuid": null, + "timestamp": "2026-08-10T10:00:00.000Z", + "message": {"role": "user", "content": "Fix the flaky cache test"}}), + ), + with( + ctx("/repo/app"), + json!({"type": "assistant", "uuid": "a1", "parentUuid": "u1", + "timestamp": "2026-08-10T10:00:05.000Z", + "message": {"role": "assistant", "model": "claude-x", "stop_reason": "tool_use", + "usage": {"input_tokens": 100, "output_tokens": 20, "cache_read_input_tokens": 50}, + "content": [ + {"type": "thinking", "thinking": "look at the cache", "signature": "sig"}, + {"type": "text", "text": "Editing the cache."}, + {"type": "tool_use", "id": "t1", "name": "Edit", + "input": {"file_path": "/repo/app/src/cache.rs", "old_string": "a", "new_string": "b"}} + ]}}), + ), + json!({"type": "file-history-snapshot", "messageId": "a1"}), + with( + ctx("/repo/app"), + json!({"type": "user", "uuid": "u2", "parentUuid": "a1", + "timestamp": "2026-08-10T10:00:06.000Z", + "message": {"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "t1", "content": "edited", "is_error": false} + ]}}), + ), + with( + ctx("/repo/app"), + json!({"type": "assistant", "uuid": "a2", "parentUuid": "u2", + "timestamp": "2026-08-10T10:00:07.000Z", + "message": {"role": "assistant", "model": "claude-x", + "usage": {"input_tokens": 7, "output_tokens": 3}, + "content": [{"type": "tool_use", "id": "t2", "name": "Write", + "input": {"file_path": "/repo/app/NOTES.md", "content": format!("key {}", token())}}]}}), + ), + json!({"type": "ai-title", "aiTitle": "Cache test fix", "sessionId": sid}), + with( + ctx("/repo/lib"), + json!({"type": "user", "uuid": "u3", "parentUuid": "a2", + "timestamp": "2026-08-10T10:00:08.000Z", + "message": {"role": "user", "content": format!("deploy with {}", token())}}), + ), + with( + ctx("/repo/lib"), + json!({"type": "assistant", "uuid": "a3", "parentUuid": "u3", + "isSidechain": true, "timestamp": "2026-08-10T10:00:09.000Z", + "message": {"role": "assistant", "content": [{"type": "text", "text": "sidechain note"}]}}), + ), + json!({"type": "custom-title", "customTitle": "Renamed", "sessionId": sid}), + with( + ctx("/repo/lib"), + json!({"type": "user", "uuid": "u4", "parentUuid": "u9", + "timestamp": "2026-08-10T10:00:10.000Z", + "message": {"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "t-missing", "content": "lost"} + ]}}), + ), + with( + ctx("/repo/lib"), + json!({"type": "assistant", "uuid": "u9", "parentUuid": "a3", + "timestamp": "2026-08-10T10:00:11.000Z", + "message": {"role": "assistant", "content": "late parent"}}), + ), + with( + ctx("/repo/app"), + json!({"type": "user", "uuid": "u5", "parentUuid": "u9", + "timestamp": "2026-08-10T10:00:12.000Z", + "message": {"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "t2", "content": [{"type": "text", "text": "written"}]} + ]}}), + ), + ]) +} + +#[test] +fn claude_code_appends_match_a_fresh_ingest_line_by_line() { + let content = claude_session(); + let (kept, live, id) = + grow_and_compare(&ClaudeCodeAdapter::new(), &content, &line_ends(&content)); + // Line 11 is the message an earlier line named as its parent: the earlier + // row changes, so that one append replaces the session. + assert_eq!( + kept, + [true, true, true, true, true, true, true, true, true, false, true] + ); + assert_eq!(summary(&live, &id), claude_summary()); +} + +fn claude_summary() -> Summary { + ( + "Renamed".to_string(), + 9, + 2, + Some(107), + Some(23), + Some(50), + Some("2 parser note(s); first: unhandled event type: file-history-snapshot".to_string()), + ) +} + +#[test] +fn claude_code_appends_match_a_fresh_ingest_in_chunks() { + let content = claude_session(); + let ends = line_ends(&content); + let cuts = [ends[1], ends[4], ends[8], ends[11]]; + let (kept, live, id) = grow_and_compare(&ClaudeCodeAdapter::new(), &content, &cuts); + // The late parent arrives in the same append as the message naming it. + assert_eq!(kept, [true, true, true]); + assert_eq!(summary(&live, &id), claude_summary()); +} + +fn codex_session() -> String { + let at = |s: u32| format!("2026-08-11T10:00:{s:02}.000Z"); + let item = |s: u32, payload: serde_json::Value| json!({"type": "response_item", "timestamp": at(s), "payload": payload}); + let event = |s: u32, payload: serde_json::Value| json!({"type": "event_msg", "timestamp": at(s), "payload": payload}); + jsonl(&[ + json!({"type": "session_meta", "timestamp": at(0), "payload": { + "id": "cx-eq", "cli_version": "0.9", "cwd": "/repo", "model_provider": "openai", + "git": {"branch": "main", "commit_hash": "abc123", + "repository_url": "https://github.com/o/r.git"}}}), + json!({"type": "turn_context", "timestamp": at(1), + "payload": {"cwd": "/repo", "model": "gpt-5"}}), + item( + 2, + json!({"type": "message", "role": "user", + "content": [{"type": "input_text", "text": "Add a config loader"}]}), + ), + item( + 3, + json!({"type": "reasoning", "summary": [{"type": "summary_text", "text": "plan"}], + "encrypted_content": "opaque"}), + ), + item( + 4, + json!({"type": "function_call", "name": "apply_patch", "arguments": "{}", + "call_id": "c1"}), + ), + event( + 5, + json!({"type": "token_count", "info": {"total_token_usage": + {"input_tokens": 10, "output_tokens": 5, "cached_input_tokens": 2}}}), + ), + item( + 6, + json!({"type": "function_call_output", "call_id": "c1", "output": "Done"}), + ), + event( + 7, + json!({"type": "patch_apply_end", "call_id": "c1", "success": true, + "changes": {"src/config.ts": {"type": "add", + "content": format!("const KEY = '{}'", token())}}}), + ), + json!({"type": "turn_context", "timestamp": at(8), + "payload": {"cwd": "/repo/pkg", "model": "gpt-5"}}), + item( + 9, + json!({"type": "message", "role": "assistant", + "content": [{"type": "output_text", "text": "Loader added"}]}), + ), + event(10, json!({"type": "brand_new_event"})), + json!({"type": "compacted", "timestamp": at(11), "payload": {"message": "summary"}}), + event( + 12, + json!({"type": "patch_apply_end", "call_id": "c2", "success": true, + "changes": {"a.ts": {"type": "update", "unified_diff": "@@ -1 +1 @@\n-x\n+y\n"}}}), + ), + item( + 13, + json!({"type": "function_call", "name": "shell", "arguments": "{}", + "call_id": "c2"}), + ), + event( + 14, + json!({"type": "token_count", "info": {"total_token_usage": + {"input_tokens": 40, "output_tokens": 9}}}), + ), + json!({"type": "session_meta", "timestamp": at(15), "payload": { + "id": "cx-eq", "cli_version": "0.9", "cwd": "/repo", "model_provider": "azure"}}), + ]) +} + +#[test] +fn codex_appends_match_a_fresh_ingest_line_by_line() { + let content = codex_session(); + let (kept, live, id) = grow_and_compare(&CodexAdapter::new(), &content, &line_ends(&content)); + // The session has no message, and so no checkpoint, until line 3. Line 14 + // carries the call an earlier patch named, and line 16 restates the + // session meta that every segment carries: both replace the session. + assert_eq!( + kept, + [ + false, false, true, true, true, true, true, true, true, true, true, true, false, true, + false + ] + ); + assert_eq!(summary(&live, &id), codex_summary()); +} + +fn codex_summary() -> Summary { + ( + "Add a config loader".to_string(), + 7, + 2, + Some(40), + Some(9), + None, + Some("1 parser note(s); first: unknown event_msg: brand_new_event".to_string()), + ) +} + +#[test] +fn codex_appends_match_a_fresh_ingest_in_chunks() { + let content = codex_session(); + let ends = line_ends(&content); + let cuts = [ends[3], ends[5], ends[9], ends[12], ends[14]]; + let (kept, live, id) = grow_and_compare(&CodexAdapter::new(), &content, &cuts); + assert_eq!(kept, [true, true, true, false]); + assert_eq!(summary(&live, &id), codex_summary()); +} + +// A watcher can fire while the agent is mid-way through writing a line. The +// half line parses as a partial session, and once the line is complete the +// session is whole again, exactly as a fresh ingest of each state reads. +#[test] +fn a_line_split_across_appends_matches_a_fresh_ingest() { + let content = claude_session(); + let ends = line_ends(&content); + let mid = |i: usize| ends[i - 1] + (ends[i] - ends[i - 1]) / 2; + let cuts = [ + ends[0], + mid(1), + ends[1], + mid(3), + mid(3) + 7, + ends[3], + ends[4], + ]; + let (kept, _, _) = grow_and_compare(&ClaudeCodeAdapter::new(), &content, &cuts); + // A half line leaves no checkpoint, so the step after it parses the whole + // file again. + assert_eq!(kept, [true, false, true, false, false, true]); +} + +#[test] +fn a_split_line_reads_partial_then_whole() { + let dir = tempfile::tempdir().unwrap(); + let blobs = BlobStore::open(dir.path().join("blobs")).unwrap(); + let path = dir.path().join("session.jsonl"); + let conn = lore_core::storage::open_in_memory().unwrap(); + let adapter = ClaudeCodeAdapter::new(); + let content = claude_session(); + let ends = line_ends(&content); + + let status = |conn: &Connection| -> (String, Option, i64) { + conn.query_row( + "SELECT parse_status, parse_note, message_count FROM agent_session", + [], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)), + ) + .unwrap() + }; + + std::fs::write(&path, &content[..ends[0]]).unwrap(); + ingest_file(&conn, &adapter, &path, &blobs).unwrap(); + std::fs::write(&path, &content[..ends[0] + 40]).unwrap(); + ingest_file(&conn, &adapter, &path, &blobs).unwrap(); + assert_eq!( + status(&conn), + ( + "partial".to_string(), + Some("1 parser note(s); first: unparseable line (possibly truncated)".to_string()), + 1 + ) + ); + + std::fs::write(&path, &content[..ends[1]]).unwrap(); + ingest_file(&conn, &adapter, &path, &blobs).unwrap(); + assert_eq!(status(&conn), ("ok".to_string(), None, 2)); +} + +#[test] +fn a_secret_in_an_appended_line_is_redacted_and_recorded() { + let dir = tempfile::tempdir().unwrap(); + let blobs = BlobStore::open(dir.path().join("blobs")).unwrap(); + let path = dir.path().join("session.jsonl"); + let conn = lore_core::storage::open_in_memory().unwrap(); + let adapter = ClaudeCodeAdapter::new(); + let content = claude_session(); + let ends = line_ends(&content); + + std::fs::write(&path, &content[..ends[5]]).unwrap(); + let sid = session_id(ingest_file(&conn, &adapter, &path, &blobs).unwrap()); + let before = message_rowids(&conn, &sid); + std::fs::write(&path, &content[..ends[6]]).unwrap(); + ingest_file(&conn, &adapter, &path, &blobs).unwrap(); + assert_eq!(&message_rowids(&conn, &sid)[..before.len()], &before[..]); + + let (text, rule, span): (String, String, (i64, i64)) = conn + .query_row( + "SELECT d.redacted_text, f.rule, f.span_start, f.span_end + FROM search_document d + JOIN message_part p ON p.id = d.source_id + JOIN message m ON m.id = p.message_id + JOIN secret_finding f ON f.source_id = p.id AND f.field = 'text' + WHERE m.seq = 4", + [], + |r| Ok((r.get(0)?, r.get(1)?, (r.get(2)?, r.get(3)?))), + ) + .unwrap(); + assert_eq!(text, "deploy with «redacted:github-token»"); + assert_eq!(rule, "github-token"); + assert_eq!(span, (12, 52)); + + let hits: i64 = conn + .query_row( + "SELECT count(*) FROM search_fts WHERE search_fts MATCH 'deploy'", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(hits, 1, "the appended line is searchable"); +} diff --git a/crates/lore-core/tests/persist.rs b/crates/lore-core/tests/persist.rs index 988b024..98d4af9 100644 --- a/crates/lore-core/tests/persist.rs +++ b/crates/lore-core/tests/persist.rs @@ -387,3 +387,87 @@ fn codex_recorded_patch_payloads_persist_as_faithful_blobs() { // Move: the recorded move diff round-trips. assert!(read_patch("src/renamed.ts").contains("moved")); } + +fn claude_line(session: &str, uuid: &str, parent: Option<&str>, role: &str, text: &str) -> String { + let line = serde_json::json!({ + "type": role, "uuid": uuid, "parentUuid": parent, "sessionId": session, + "timestamp": "2026-08-10T10:01:00.000Z", "cwd": "/repo/app", "gitBranch": "main", + "message": {"role": role, "content": text} + }); + format!("{line}\n") +} + +fn search_ids(conn: &Connection, session_id: &str) -> Vec { + let mut stmt = conn + .prepare("SELECT id FROM search_document WHERE session_id = ?1 ORDER BY id") + .unwrap(); + let ids = stmt + .query_map([session_id], |r| r.get(0)) + .unwrap() + .collect::>>() + .unwrap(); + ids +} + +// An append must cost the new rows, not the whole session: re-inserting every +// search projection is what made a live agent's session burn a core. Session B +// is ingested between A's two passes so a delete-and-reinsert of A cannot hide +// behind reused rowids. +#[test] +fn an_append_keeps_existing_search_rows_and_adds_only_the_new_one() { + use lore_core::ingest::{ingest_file, ChangeClass, IngestOutcome}; + + let dir = tempfile::tempdir().unwrap(); + let conn = lore_core::storage::open_in_memory().unwrap(); + let (_bd, blobs) = blob_store(); + let adapter = ClaudeCodeAdapter::new(); + + let a_path = dir.path().join("a.jsonl"); + let a_prefix = [ + claude_line("sess-a", "a1", None, "user", "alpha one"), + claude_line("sess-a", "a2", Some("a1"), "assistant", "alpha two"), + claude_line("sess-a", "a3", Some("a2"), "user", "alpha three"), + claude_line("sess-a", "a4", Some("a3"), "assistant", "alpha four"), + ] + .concat(); + std::fs::write(&a_path, &a_prefix).unwrap(); + let IngestOutcome::Ingested { session_id: a, .. } = + ingest_file(&conn, &adapter, &a_path, &blobs).unwrap() + else { + panic!("session A must ingest"); + }; + + let b_path = dir.path().join("b.jsonl"); + std::fs::write( + &b_path, + claude_line("sess-b", "b1", None, "user", "bravo one"), + ) + .unwrap(); + ingest_file(&conn, &adapter, &b_path, &blobs).unwrap(); + + let before = search_ids(&conn, &a); + assert_eq!(before.len(), 4); + + let appended = claude_line("sess-a", "a5", Some("a4"), "user", "alpha five"); + std::fs::write(&a_path, format!("{a_prefix}{appended}")).unwrap(); + let outcome = ingest_file(&conn, &adapter, &a_path, &blobs).unwrap(); + assert!(matches!( + outcome, + IngestOutcome::Ingested { + change: ChangeClass::Appended, + .. + } + )); + + let after = search_ids(&conn, &a); + assert_eq!(&after[..4], &before[..], "existing search rows are kept"); + assert_eq!(after.len(), 5, "exactly one search row is added"); + let added: String = conn + .query_row( + "SELECT redacted_text FROM search_document WHERE id = ?1", + [after[4]], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(added, "alpha five"); +}