remem/ingest/
sessions.rs

1//! `remem ingest-sessions` — batch, incremental, idempotent ingestion of
2//! Claude Code / Codex session transcripts into `raw_messages` (issue #722).
3//!
4//! Discovery walks each scan root for `*.jsonl` files (skipping `subagents/`
5//! directories), a per-file cursor in `ingest_cursors` skips files whose
6//! mtime and size are unchanged, and each hit is drained through the existing
7//! `drain_transcript` path so the `raw_messages` UNIQUE constraint dedupes
8//! against the Stop-hook ingestion running concurrently.
9
10use std::collections::{BTreeMap, BTreeSet};
11use std::path::{Path, PathBuf};
12
13use anyhow::{bail, Result};
14use rusqlite::{params, Connection, OptionalExtension};
15use serde::Serialize;
16
17use crate::identity::InstallHost;
18use crate::memory::raw_archive::{self, TranscriptDrainOptions, SOURCE_ROOT_LOCAL};
19
20/// A file whose mtime is within this many seconds of now is treated as an
21/// actively-appended session: a JSON parse failure on its last line is a
22/// partial tail, not a file failure, and the cursor does not advance.
23const ACTIVE_TAIL_WINDOW_SECS: i64 = 60;
24const RESERVED_CURSOR_OUTCOME_SOURCE_ROOT: &str = "cursor-outcome";
25
26/// One scan root: a label recorded as `raw_messages.source_root` plus the
27/// directory to walk.
28#[derive(Debug, Clone)]
29pub struct ScanRoot {
30    pub host: InstallHost,
31    pub label: String,
32    pub path: PathBuf,
33    /// Default local roots are optional because many users only have one host
34    /// installed. User-supplied `--root HOST:LABEL=PATH` entries are required
35    /// and must not fail silently.
36    pub required: bool,
37}
38
39impl ScanRoot {
40    /// Parse a `--root HOST:LABEL=PATH` argument.
41    pub fn parse(spec: &str) -> Result<Self> {
42        let Some((host, root)) = spec.split_once(':') else {
43            bail!("invalid --root {spec:?}: expected HOST:LABEL=PATH");
44        };
45        let host = InstallHost::parse(host)
46            .map_err(|error| anyhow::anyhow!("invalid --root {spec:?}: {error}"))?;
47        validate_batch_host(host)?;
48        let Some((label, path)) = root.split_once('=') else {
49            bail!("invalid --root {spec:?}: expected HOST:LABEL=PATH");
50        };
51        let label = label.trim();
52        let path = path.trim();
53        if label.is_empty() || path.is_empty() {
54            bail!("invalid --root {spec:?}: label and path must be non-empty");
55        }
56        if label == RESERVED_CURSOR_OUTCOME_SOURCE_ROOT {
57            bail!("invalid --root {spec:?}: source-root label {label:?} is reserved");
58        }
59        Ok(Self {
60            host,
61            label: label.to_string(),
62            path: PathBuf::from(shellexpand_home(path)),
63            required: true,
64        })
65    }
66
67    pub(crate) fn validate_for_batch(&self) -> Result<()> {
68        validate_batch_host(self.host)
69    }
70}
71
72fn validate_batch_host(host: InstallHost) -> Result<()> {
73    if host == InstallHost::Cursor {
74        bail!(
75            "Cursor filesystem roots are unsupported; Cursor transcripts must enter through the Stop snapshot contract"
76        );
77    }
78    Ok(())
79}
80
81fn shellexpand_home(path: &str) -> String {
82    if let Some(rest) = path.strip_prefix("~/") {
83        if let Some(home) = dirs::home_dir() {
84            return home.join(rest).to_string_lossy().to_string();
85        }
86    }
87    path.to_string()
88}
89
90/// Default local scan roots: `~/.claude/projects` and `~/.codex/sessions`.
91/// Both are labeled `local` to match the hook-path `source_root` default.
92pub fn default_scan_roots() -> Vec<ScanRoot> {
93    let Some(home) = dirs::home_dir() else {
94        crate::log::warn("ingest-sessions", "home directory unavailable");
95        return Vec::new();
96    };
97    vec![
98        ScanRoot {
99            host: InstallHost::ClaudeCode,
100            label: SOURCE_ROOT_LOCAL.to_string(),
101            path: home.join(".claude").join("projects"),
102            required: false,
103        },
104        ScanRoot {
105            host: InstallHost::CodexCli,
106            label: SOURCE_ROOT_LOCAL.to_string(),
107            path: home.join(".codex").join("sessions"),
108            required: false,
109        },
110    ]
111}
112
113/// Machine-readable batch summary (product invariant 6).
114#[derive(Debug, Clone, Default, Serialize, PartialEq, Eq)]
115pub struct IngestSummary {
116    pub scanned: usize,
117    pub skipped: usize,
118    pub ingested_messages: usize,
119    pub failed_files: usize,
120    pub partial_files: usize,
121}
122
123impl IngestSummary {
124    pub fn exit_code(&self) -> i32 {
125        if self.failed_files > 0 {
126            1
127        } else {
128            0
129        }
130    }
131}
132
133#[derive(Debug, Clone, Default)]
134pub struct IngestOptions {
135    /// Skip files whose mtime is older than this lower bound (backfill bound;
136    /// window semantics on message timestamps belong to the query side).
137    pub since_epoch: Option<i64>,
138}
139
140/// Run one batch ingestion pass over the given scan roots (callers build the
141/// list from `default_scan_roots()` plus any `--root HOST:LABEL=PATH` extras).
142pub fn run_ingest_sessions(
143    conn: &Connection,
144    roots: &[ScanRoot],
145    options: &IngestOptions,
146) -> Result<IngestSummary> {
147    for root in roots {
148        root.validate_for_batch()?;
149    }
150    if roots
151        .iter()
152        .any(|root| root.label == RESERVED_CURSOR_OUTCOME_SOURCE_ROOT)
153    {
154        bail!(
155            "source-root label {:?} is reserved and cannot be ingested",
156            RESERVED_CURSOR_OUTCOME_SOURCE_ROOT
157        );
158    }
159    let mut summary = IngestSummary::default();
160    let now = chrono::Utc::now().timestamp();
161    let mut project_cache = BTreeMap::new();
162    let mut discovered = Vec::new();
163    for root in roots {
164        let (files, discovery_failures) = discover_transcript_files(root);
165        for failure in discovery_failures {
166            summary.failed_files += 1;
167            crate::log::error("ingest-sessions", &failure);
168        }
169        for file in files {
170            summary.scanned += 1;
171            let plan = match super::session_identity::probe_with_project_cache(
172                root.host,
173                &root.label,
174                &root.path,
175                &file,
176                None,
177                &mut project_cache,
178            ) {
179                Ok(plan) => plan,
180                Err(error) => {
181                    summary.failed_files += 1;
182                    crate::log::error(
183                        "ingest-sessions",
184                        &format!("identity probe {} failed: {error}", file.display()),
185                    );
186                    continue;
187                }
188            };
189            let mtime_epoch = plan.observed_mtime_ns / 1_000_000_000;
190            let phase_b_eligible = options.since_epoch.is_none_or(|since| mtime_epoch >= since);
191            discovered.push((root.clone(), plan, phase_b_eligible));
192        }
193    }
194    if summary.failed_files > 0 {
195        crate::log::error(
196            "ingest-sessions",
197            "Phase A discovery/probe was incomplete; Phase B mutation is blocked",
198        );
199        return Ok(summary);
200    }
201    conn.execute_batch("SAVEPOINT gh871_identity_phase_a")?;
202    let phase_a =
203        (|| -> Result<Vec<(ScanRoot, super::session_identity::TranscriptPlan, i64, bool)>> {
204            let mut prepared = Vec::with_capacity(discovered.len());
205            let mut groups = BTreeSet::new();
206            for (root, plan, phase_b_eligible) in discovered {
207                let identity_id = super::session_identity::upsert_claim(conn, &plan, now)?;
208                let host = plan
209                    .host
210                    .expect("hostless batch plans are rejected before persistence")
211                    .as_db_value()
212                    .to_string();
213                groups.insert((
214                    host,
215                    plan.source_root.clone(),
216                    plan.fallback_session_id.clone(),
217                ));
218                prepared.push((root, plan, identity_id, phase_b_eligible));
219            }
220            for (host, source_root, fallback_session_id) in groups {
221                super::session_identity::resolve_fallback_group(
222                    conn,
223                    Some(&host),
224                    &source_root,
225                    &fallback_session_id,
226                )?;
227            }
228            Ok(prepared)
229        })();
230    let prepared = match phase_a {
231        Ok(prepared) => {
232            conn.execute_batch("RELEASE gh871_identity_phase_a")?;
233            prepared
234        }
235        Err(error) => {
236            conn.execute_batch(
237                "ROLLBACK TO gh871_identity_phase_a; RELEASE gh871_identity_phase_a",
238            )?;
239            return Err(error.context("persist complete transcript identity claim set"));
240        }
241    };
242
243    let mut prepared_groups = BTreeMap::new();
244    for prepared_file in prepared {
245        let host = prepared_file
246            .1
247            .host
248            .expect("hostless batch plans are rejected before persistence")
249            .as_db_value()
250            .to_string();
251        let key = (
252            host,
253            prepared_file.1.source_root.clone(),
254            prepared_file.1.fallback_session_id.clone(),
255        );
256        prepared_groups
257            .entry(key)
258            .or_insert_with(Vec::new)
259            .push(prepared_file);
260    }
261    for ((host, source_root, fallback_session_id), group) in prepared_groups {
262        conn.execute_batch("SAVEPOINT gh871_identity_phase_b_group")?;
263        let ingested_before = summary.ingested_messages;
264        let partial_before = summary.partial_files;
265        let mut identity_conflict = false;
266        for (root, plan, identity_id, phase_b_eligible) in &group {
267            if !phase_b_eligible {
268                let indexed = super::session_identity::index_events(
269                    &plan.transcript_path,
270                    u64::try_from(plan.observed_size_bytes).unwrap_or(u64::MAX),
271                )
272                .and_then(|index| {
273                    super::session_identity::record_since_skipped_event_index(
274                        conn,
275                        *identity_id,
276                        index,
277                        now,
278                    )
279                });
280                match indexed {
281                    Ok(()) => summary.skipped += 1,
282                    Err(error) => {
283                        summary.failed_files += 1;
284                        crate::log::error(
285                            "ingest-sessions",
286                            &format!("index skipped {} failed: {error}", plan.path.display()),
287                        );
288                    }
289                }
290                continue;
291            }
292            conn.execute_batch("SAVEPOINT gh871_identity_phase_b_file")?;
293            let inserted_before = summary.ingested_messages;
294            let result = ingest_prepared_file(conn, root, plan, *identity_id, now, &mut summary);
295            match result {
296                PreparedFileResult::Commit => {
297                    conn.execute_batch("RELEASE gh871_identity_phase_b_file")?;
298                }
299                PreparedFileResult::Rollback {
300                    identity_conflict: file_identity_conflict,
301                } => {
302                    conn.execute_batch(
303                        "ROLLBACK TO gh871_identity_phase_b_file;
304                         RELEASE gh871_identity_phase_b_file",
305                    )?;
306                    summary.ingested_messages = inserted_before;
307                    if file_identity_conflict {
308                        identity_conflict = true;
309                        break;
310                    }
311                }
312            }
313        }
314        if identity_conflict {
315            conn.execute_batch(
316                "ROLLBACK TO gh871_identity_phase_b_group;
317                 RELEASE gh871_identity_phase_b_group",
318            )?;
319            summary.ingested_messages = ingested_before;
320            summary.partial_files = partial_before;
321            super::session_identity::mark_fallback_group_conflict(
322                conn,
323                &host,
324                &source_root,
325                &fallback_session_id,
326                "stable_occurrence_mismatch",
327            )?;
328        } else {
329            conn.execute_batch("RELEASE gh871_identity_phase_b_group")?;
330        }
331    }
332
333    crate::log::info(
334        "ingest-sessions",
335        &format!(
336            "batch done scanned={} skipped={} ingested_messages={} failed_files={} partial_files={}",
337            summary.scanned,
338            summary.skipped,
339            summary.ingested_messages,
340            summary.failed_files,
341            summary.partial_files
342        ),
343    );
344    Ok(summary)
345}
346
347pub(crate) fn discover_transcript_files(root: &ScanRoot) -> (Vec<PathBuf>, Vec<String>) {
348    if !root.path.is_dir() {
349        let failures = if root.required {
350            vec![format!(
351                "required scan root {}={} is missing or not a directory",
352                root.label,
353                root.path.display()
354            )]
355        } else {
356            Vec::new()
357        };
358        return (Vec::new(), failures);
359    }
360    let mut files = Vec::new();
361    let mut failures = Vec::new();
362    collect_jsonl_files(&root.path, &mut files, &mut failures);
363    files.sort();
364    (files, failures)
365}
366
367/// Recursively collect `*.jsonl` files, excluding `subagents/` directories.
368fn collect_jsonl_files(dir: &Path, out: &mut Vec<PathBuf>, failures: &mut Vec<String>) {
369    let entries = match std::fs::read_dir(dir) {
370        Ok(entries) => entries,
371        Err(error) => {
372            failures.push(format!("read scan dir {} failed: {}", dir.display(), error));
373            return;
374        }
375    };
376    for entry in entries {
377        let entry = match entry {
378            Ok(entry) => entry,
379            Err(error) => {
380                failures.push(format!(
381                    "read scan dir entry in {} failed: {}",
382                    dir.display(),
383                    error
384                ));
385                continue;
386            }
387        };
388        let path = entry.path();
389        let file_type = match entry.file_type() {
390            Ok(file_type) => file_type,
391            Err(error) => {
392                failures.push(format!("stat {} failed: {}", path.display(), error));
393                continue;
394            }
395        };
396        if file_type.is_dir() {
397            if entry.file_name() == "subagents" {
398                continue;
399            }
400            collect_jsonl_files(&path, out, failures);
401        } else if file_type.is_file() && path.extension().is_some_and(|ext| ext == "jsonl") {
402            out.push(path);
403        }
404    }
405}
406
407enum PreparedFileResult {
408    Commit,
409    Rollback { identity_conflict: bool },
410}
411
412fn ingest_prepared_file(
413    conn: &Connection,
414    root: &ScanRoot,
415    plan: &super::session_identity::TranscriptPlan,
416    identity_id: i64,
417    now: i64,
418    summary: &mut IngestSummary,
419) -> PreparedFileResult {
420    let identity = match super::session_identity::load(conn, identity_id) {
421        Ok(identity) => identity,
422        Err(error) => {
423            summary.failed_files += 1;
424            crate::log::error(
425                "ingest-sessions",
426                &format!("load identity {} failed: {error}", plan.path.display()),
427            );
428            return PreparedFileResult::Commit;
429        }
430    };
431    if identity.status == "conflict" {
432        summary.failed_files += 1;
433        crate::log::error(
434            "ingest-sessions",
435            &format!(
436                "identity conflict for transcript {}; raw rows remain unchanged",
437                plan.path.display()
438            ),
439        );
440        return PreparedFileResult::Commit;
441    }
442    let mtime_epoch = plan.observed_mtime_ns / 1_000_000_000;
443    let size_bytes = plan.observed_size_bytes;
444    match cursor_unchanged(conn, root, &plan.path, mtime_epoch, size_bytes) {
445        Ok(true) if identity.contract_version >= 1 => {
446            summary.skipped += 1;
447            return PreparedFileResult::Commit;
448        }
449        Ok(true) | Ok(false) => {}
450        Err(error) => {
451            summary.failed_files += 1;
452            crate::log::error(
453                "ingest-sessions",
454                &format!("cursor lookup {} failed: {}", plan.path.display(), error),
455            );
456            return PreparedFileResult::Commit;
457        }
458    }
459
460    let event_index = match super::session_identity::index_events(
461        &plan.transcript_path,
462        u64::try_from(size_bytes).unwrap_or(u64::MAX),
463    ) {
464        Ok(index) => index,
465        Err(error) => {
466            summary.failed_files += 1;
467            crate::log::error(
468                "ingest-sessions",
469                &format!("index {} failed: {error}", plan.path.display()),
470            );
471            return PreparedFileResult::Commit;
472        }
473    };
474    let drain_options = TranscriptDrainOptions {
475        source_root: &root.label,
476        tolerate_partial_tail: now - mtime_epoch <= ACTIVE_TAIL_WINDOW_SECS,
477        transcript_identity_id: Some(identity.id),
478    };
479
480    match raw_archive::drain_transcript_with_capture_limit(
481        conn,
482        &plan.transcript_path,
483        &identity.canonical_session_id,
484        &identity.project,
485        plan.branch.as_deref(),
486        plan.cwd.as_deref(),
487        &drain_options,
488        Some(u64::try_from(size_bytes).unwrap_or(u64::MAX)),
489    ) {
490        Ok(report) => {
491            summary.ingested_messages += report.inserted;
492            if report.has_failures() {
493                // drain_transcript_with_options already recorded the failure
494                // in raw_ingest_failures; keep the cursor behind so the file
495                // is retried on the next run.
496                summary.failed_files += 1;
497                crate::log::error(
498                    "ingest-sessions",
499                    &format!(
500                        "file {} failed: kind={} parse_errors={} insert_errors={} read_error={}",
501                        plan.path.display(),
502                        report.failure_kind().unwrap_or("unknown"),
503                        report.parse_errors,
504                        report.insert_errors,
505                        report.read_error.is_some()
506                    ),
507                );
508                if report.identity_conflicts > 0 {
509                    return PreparedFileResult::Rollback {
510                        identity_conflict: true,
511                    };
512                }
513            } else if report.partial_tail {
514                summary.partial_files += 1;
515            } else {
516                let completion = (|| -> Result<super::session_identity::RekeyReport> {
517                    conn.execute_batch("SAVEPOINT gh871_identity_complete")?;
518                    let rekey = super::session_identity::rekey_legacy_rows(conn, &identity)?;
519                    super::session_identity::mark_complete(conn, identity.id, event_index, now)?;
520                    advance_cursor(conn, root, &plan.path, mtime_epoch, size_bytes, now)?;
521                    conn.execute_batch("RELEASE gh871_identity_complete")?;
522                    Ok(rekey)
523                })();
524                match completion {
525                    Ok(rekey) => {
526                        summary.ingested_messages =
527                            summary.ingested_messages.saturating_sub(rekey.merged);
528                    }
529                    Err(error) => {
530                        if let Err(rollback_error) = conn.execute_batch(
531                            "ROLLBACK TO gh871_identity_complete; RELEASE gh871_identity_complete",
532                        ) {
533                            crate::log::error(
534                                "ingest-sessions",
535                                &format!(
536                                    "identity completion rollback {} failed: {rollback_error}",
537                                    plan.path.display()
538                                ),
539                            );
540                        }
541                        summary.failed_files += 1;
542                        crate::log::error(
543                            "ingest-sessions",
544                            &format!(
545                                "identity completion {} failed: {error}",
546                                plan.path.display()
547                            ),
548                        );
549                        return PreparedFileResult::Rollback {
550                            identity_conflict: error
551                                .downcast_ref::<crate::memory::raw_occurrence::RawIdentityConflict>(
552                                )
553                                .is_some(),
554                        };
555                    }
556                }
557            }
558        }
559        Err(error) => {
560            summary.failed_files += 1;
561            crate::log::error(
562                "ingest-sessions",
563                &format!("drain {} failed: {}", plan.path.display(), error),
564            );
565        }
566    }
567    PreparedFileResult::Commit
568}
569
570fn cursor_unchanged(
571    conn: &Connection,
572    root: &ScanRoot,
573    file: &Path,
574    mtime_epoch: i64,
575    size_bytes: i64,
576) -> Result<bool> {
577    let key = cursor_key(root, file);
578    let row: Option<(i64, i64)> = conn
579        .query_row(
580            "SELECT mtime_epoch, size_bytes FROM ingest_cursors WHERE file_path = ?1",
581            params![key],
582            |row| Ok((row.get(0)?, row.get(1)?)),
583        )
584        .optional()?;
585    Ok(row == Some((mtime_epoch, size_bytes)))
586}
587
588pub(crate) fn cursor_matches_identity(
589    conn: &Connection,
590    root: &ScanRoot,
591    file: &Path,
592    observed_mtime_ns: i64,
593    observed_size_bytes: i64,
594) -> Result<bool> {
595    cursor_unchanged(
596        conn,
597        root,
598        file,
599        observed_mtime_ns / 1_000_000_000,
600        observed_size_bytes,
601    )
602}
603
604fn advance_cursor(
605    conn: &Connection,
606    root: &ScanRoot,
607    file: &Path,
608    mtime_epoch: i64,
609    size_bytes: i64,
610    now: i64,
611) -> Result<()> {
612    let key = cursor_key(root, file);
613    conn.execute(
614        "INSERT INTO ingest_cursors (file_path, mtime_epoch, size_bytes, last_ingested_at)
615         VALUES (?1, ?2, ?3, ?4)
616         ON CONFLICT(file_path) DO UPDATE SET
617             mtime_epoch = excluded.mtime_epoch,
618             size_bytes = excluded.size_bytes,
619             last_ingested_at = excluded.last_ingested_at",
620        params![key, mtime_epoch, size_bytes, now],
621    )?;
622    Ok(())
623}
624
625fn cursor_key(root: &ScanRoot, file: &Path) -> String {
626    format!("{}\0{}", root.label, file.to_string_lossy())
627}
628
629#[cfg(test)]
630#[path = "sessions/tests.rs"]
631mod tests;
remem/ingest/sessions.rs

remem/ingest/
sessions.rs
