remem/memory/
raw_archive.rs

1use anyhow::{Context, Result};
2use rusqlite::{params, Connection, OptionalExtension};
3
4mod session_labels;
5mod sessions;
6#[cfg(test)]
7pub(crate) use sessions::RawSessionListing;
8pub(crate) use sessions::{
9    build_session_listing_json, list_sessions_with_exclusions, SessionFingerprint,
10};
11pub use sessions::{
12    build_sessions_json, list_sessions, RawSessionQuery, RawSessionSummary, RawSessionsJson,
13};
14
15pub const ROLE_USER: &str = "user";
16pub const ROLE_ASSISTANT: &str = "assistant";
17
18pub const SOURCE_TRANSCRIPT: &str = "transcript";
19pub const SOURCE_HOOK: &str = "hook";
20pub const SOURCE_MANUAL: &str = "manual";
21
22pub const SOURCE_ROOT_LOCAL: &str = "local";
23
24#[derive(Debug, Clone)]
25pub struct RawMessage {
26    pub id: i64,
27    pub session_id: String,
28    pub project: String,
29    pub role: String,
30    pub content: String,
31    pub source: String,
32    pub branch: Option<String>,
33    pub cwd: Option<String>,
34    pub created_at_epoch: i64,
35}
36
37/// Exact byte-for-byte hash of the raw message content. Distinct from
38/// `memory::promote::slug::content_hash`, which normalizes whitespace/case for
39/// semantic dedup of curated memories.
40fn exact_content_hash(content: &str) -> String {
41    crate::db::content_identity_hash(content.as_bytes())
42}
43
44fn legacy_exact_content_hash(content: &str) -> String {
45    crate::db::legacy_content_identity_hash(content.as_bytes())
46}
47
48#[derive(Debug, Clone, Copy)]
49pub struct RawInsertOutcome {
50    pub id: i64,
51    pub inserted: bool,
52}
53
54#[derive(Debug, Clone, Default, PartialEq, Eq)]
55pub struct RawIngestReport {
56    pub inserted: usize,
57    pub duplicates: usize,
58    pub empty_messages: usize,
59    pub skipped_messages: usize,
60    pub parse_errors: usize,
61    pub insert_errors: usize,
62    pub identity_conflicts: usize,
63    pub read_error: Option<String>,
64    /// The last line failed JSON parse while the drain was told to tolerate an
65    /// actively-appended tail (issue #722). Not counted as a parse error; the
66    /// caller must not advance its ingest cursor so the tail is re-read later.
67    pub partial_tail: bool,
68}
69
70impl RawIngestReport {
71    pub fn has_failures(&self) -> bool {
72        self.read_error.is_some() || self.parse_errors > 0 || self.insert_errors > 0
73    }
74
75    pub fn failure_kind(&self) -> Option<&'static str> {
76        match (
77            self.read_error.is_some(),
78            self.parse_errors > 0,
79            self.insert_errors > 0,
80        ) {
81            (true, false, false) => Some("read_error"),
82            (false, true, false) => Some("parse_errors"),
83            (false, false, true) => Some("insert_errors"),
84            (true, _, _) | (_, true, true) => Some("mixed_errors"),
85            (false, false, false) => None,
86        }
87    }
88
89    fn failure_message(&self) -> String {
90        if let Some(error) = &self.read_error {
91            return error.clone();
92        }
93        format!(
94            "parse_errors={} insert_errors={}",
95            self.parse_errors, self.insert_errors
96        )
97    }
98}
99
100pub fn insert_raw_message(
101    conn: &Connection,
102    session_id: &str,
103    project: &str,
104    role: &str,
105    content: &str,
106    source: &str,
107    branch: Option<&str>,
108    cwd: Option<&str>,
109) -> Result<Option<RawInsertOutcome>> {
110    insert_raw_message_from_root(
111        conn,
112        session_id,
113        project,
114        role,
115        content,
116        source,
117        branch,
118        cwd,
119        SOURCE_ROOT_LOCAL,
120    )
121}
122
123#[allow(clippy::too_many_arguments)]
124pub fn insert_raw_message_from_root(
125    conn: &Connection,
126    session_id: &str,
127    project: &str,
128    role: &str,
129    content: &str,
130    source: &str,
131    branch: Option<&str>,
132    cwd: Option<&str>,
133    source_root: &str,
134) -> Result<Option<RawInsertOutcome>> {
135    insert_raw_message_from_root_at(
136        conn,
137        session_id,
138        project,
139        role,
140        content,
141        source,
142        branch,
143        cwd,
144        source_root,
145        None,
146    )
147}
148
149#[allow(clippy::too_many_arguments)]
150pub fn insert_raw_message_from_root_at(
151    conn: &Connection,
152    session_id: &str,
153    project: &str,
154    role: &str,
155    content: &str,
156    source: &str,
157    branch: Option<&str>,
158    cwd: Option<&str>,
159    source_root: &str,
160    created_at_epoch: Option<i64>,
161) -> Result<Option<RawInsertOutcome>> {
162    let trimmed = content.trim();
163    if trimmed.is_empty() {
164        return Ok(None);
165    }
166    let hash = exact_content_hash(trimmed);
167    if let Some(id) =
168        find_matching_legacy_raw_message(conn, session_id, project, role, trimmed, source_root)?
169    {
170        return Ok(Some(RawInsertOutcome {
171            id,
172            inserted: false,
173        }));
174    }
175    let inserted_at = created_at_epoch.unwrap_or_else(|| chrono::Utc::now().timestamp());
176
177    let event_time_source = if source == SOURCE_TRANSCRIPT {
178        if created_at_epoch.is_some() {
179            "transcript_event"
180        } else {
181            "ingest_fallback"
182        }
183    } else {
184        "legacy_unknown"
185    };
186    let inserted = conn.execute(
187        "INSERT OR IGNORE INTO raw_messages \
188         (session_id, project, role, content, content_hash, source, branch, cwd, \
189          created_at_epoch, source_root, event_time_source) \
190         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
191        params![
192            session_id,
193            project,
194            role,
195            trimmed,
196            hash,
197            source,
198            branch,
199            cwd,
200            inserted_at,
201            source_root,
202            event_time_source
203        ],
204    )?;
205
206    if inserted > 0 {
207        Ok(Some(RawInsertOutcome {
208            id: conn.last_insert_rowid(),
209            inserted: true,
210        }))
211    } else {
212        let existing: i64 = conn.query_row(
213            "SELECT id FROM raw_messages \
214             WHERE source_root = ?1 AND project = ?2 AND session_id = ?3 \
215               AND role = ?4 AND content_hash = ?5
216               AND transcript_identity_id IS NULL",
217            params![source_root, project, session_id, role, hash],
218            |row| row.get(0),
219        )?;
220        Ok(Some(RawInsertOutcome {
221            id: existing,
222            inserted: false,
223        }))
224    }
225}
226
227fn find_matching_legacy_raw_message(
228    conn: &Connection,
229    session_id: &str,
230    project: &str,
231    role: &str,
232    content: &str,
233    source_root: &str,
234) -> Result<Option<i64>> {
235    let legacy_hash = legacy_exact_content_hash(content);
236    let Some((id, stored_content)) = conn
237        .query_row(
238            "SELECT id, content FROM raw_messages
239             WHERE source_root = ?1 AND project = ?2 AND session_id = ?3 \
240               AND role = ?4 AND content_hash = ?5
241               AND transcript_identity_id IS NULL",
242            params![source_root, project, session_id, role, legacy_hash],
243            |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
244        )
245        .optional()?
246    else {
247        return Ok(None);
248    };
249
250    if stored_content == content {
251        Ok(Some(id))
252    } else {
253        Ok(None)
254    }
255}
256
257/// Options for a transcript drain beyond the hook-path defaults.
258#[derive(Debug, Clone)]
259pub struct TranscriptDrainOptions<'a> {
260    /// Label of the scan root the transcript came from (`local` for the hook
261    /// path and default `ingest-sessions` roots).
262    pub source_root: &'a str,
263    /// Treat a JSON parse failure on the final line as an actively-appended
264    /// partial tail instead of a parse error (issue #722). The caller decides
265    /// this from the file mtime; see `RawIngestReport::partial_tail`.
266    pub tolerate_partial_tail: bool,
267    /// Stable local transcript identity. When present, line ordinals preserve
268    /// repeated identical turns while making Stop/batch replay idempotent.
269    pub transcript_identity_id: Option<i64>,
270}
271
272impl Default for TranscriptDrainOptions<'_> {
273    fn default() -> Self {
274        Self {
275            source_root: SOURCE_ROOT_LOCAL,
276            tolerate_partial_tail: false,
277            transcript_identity_id: None,
278        }
279    }
280}
281
282/// Drain a Claude Code transcript JSONL file into raw_messages.
283pub fn drain_transcript(
284    conn: &Connection,
285    transcript_path: &str,
286    session_id: &str,
287    project: &str,
288    branch: Option<&str>,
289    cwd: Option<&str>,
290) -> Result<RawIngestReport> {
291    drain_transcript_with_options(
292        conn,
293        transcript_path,
294        session_id,
295        project,
296        branch,
297        cwd,
298        &TranscriptDrainOptions::default(),
299    )
300}
301
302pub fn raw_ingest_status(report: &RawIngestReport) -> &'static str {
303    if report.read_error.is_some() {
304        "read_failed"
305    } else if report.parse_errors > 0 || report.insert_errors > 0 {
306        "partial"
307    } else if report.inserted == 0 && report.duplicates > 0 {
308        "duplicate_only"
309    } else {
310        "ok"
311    }
312}
313
314/// Drain a transcript with an explicit source root and partial-tail policy.
315#[allow(clippy::too_many_arguments)]
316pub fn drain_transcript_with_options(
317    conn: &Connection,
318    transcript_path: &str,
319    session_id: &str,
320    project: &str,
321    branch: Option<&str>,
322    cwd: Option<&str>,
323    options: &TranscriptDrainOptions<'_>,
324) -> Result<RawIngestReport> {
325    drain_transcript_with_capture_limit(
326        conn,
327        transcript_path,
328        session_id,
329        project,
330        branch,
331        cwd,
332        options,
333        None,
334    )
335}
336
337#[allow(clippy::too_many_arguments)]
338pub(crate) fn drain_transcript_with_capture_limit(
339    conn: &Connection,
340    transcript_path: &str,
341    session_id: &str,
342    project: &str,
343    branch: Option<&str>,
344    cwd: Option<&str>,
345    options: &TranscriptDrainOptions<'_>,
346    byte_limit: Option<u64>,
347) -> Result<RawIngestReport> {
348    let mut report = RawIngestReport::default();
349    let mut record_ordinal = 0_i64;
350    let stream_result = with_raw_archive_drain_savepoint(conn, || {
351        crate::memory::raw_transcript::stream_transcript_lines(
352            transcript_path,
353            byte_limit,
354            |line, is_final| {
355                let ordinal = record_ordinal;
356                record_ordinal += 1;
357                use crate::memory::raw_transcript::TranscriptRecordClass;
358                let message =
359                    match crate::memory::raw_transcript::classify_transcript_line(line, None) {
360                        TranscriptRecordClass::Conversation(message)
361                        | TranscriptRecordClass::MetaUser(message)
362                        | TranscriptRecordClass::XmlControlUser(message)
363                        | TranscriptRecordClass::MissingEventTime(message) => message,
364                        TranscriptRecordClass::MalformedRecord => {
365                            if options.tolerate_partial_tail && is_final {
366                                report.partial_tail = true;
367                            } else {
368                                report.parse_errors += 1;
369                            }
370                            return;
371                        }
372                        TranscriptRecordClass::EmptyText => {
373                            report.empty_messages += 1;
374                            return;
375                        }
376                        TranscriptRecordClass::UnsupportedRecord
377                        | TranscriptRecordClass::OutsideWindow => {
378                            report.skipped_messages += 1;
379                            return;
380                        }
381                    };
382                let insert_result = if let Some(identity_id) = options.transcript_identity_id {
383                    crate::memory::raw_occurrence::insert_transcript_occurrence(
384                        conn,
385                        session_id,
386                        project,
387                        message.role,
388                        &message.text,
389                        branch,
390                        cwd,
391                        options.source_root,
392                        message.created_at_epoch,
393                        identity_id,
394                        ordinal,
395                    )
396                } else {
397                    insert_raw_message_from_root_at(
398                        conn,
399                        session_id,
400                        project,
401                        message.role,
402                        &message.text,
403                        SOURCE_TRANSCRIPT,
404                        branch,
405                        cwd,
406                        options.source_root,
407                        message.created_at_epoch,
408                    )
409                };
410                match insert_result {
411                    Ok(Some(outcome)) if outcome.inserted => report.inserted += 1,
412                    Ok(Some(_)) => report.duplicates += 1,
413                    Ok(None) => report.empty_messages += 1,
414                    Err(error) => {
415                        report.insert_errors += 1;
416                        if error
417                            .downcast_ref::<crate::memory::raw_occurrence::RawIdentityConflict>()
418                            .is_some()
419                        {
420                            report.identity_conflicts += 1;
421                        }
422                        crate::log::warn(
423                            "raw-archive",
424                            &format!("insert raw message failed: {}", error),
425                        );
426                    }
427                }
428            },
429        )
430        .map_err(anyhow::Error::from)?;
431        if report.parse_errors > 0 || report.insert_errors > 0 {
432            anyhow::bail!("raw archive drain validation failed");
433        }
434        Ok(())
435    })?;
436    if let Err(error) = stream_result {
437        report.inserted = 0;
438        report.duplicates = 0;
439        if report.parse_errors == 0 && report.insert_errors == 0 {
440            report.read_error = Some(format!("read transcript {transcript_path} failed: {error}"));
441            crate::log::warn(
442                "raw-archive",
443                report
444                    .read_error
445                    .as_deref()
446                    .unwrap_or("read transcript failed"),
447            );
448        }
449    }
450    if report.has_failures() {
451        record_raw_ingest_failure(
452            conn,
453            session_id,
454            project,
455            SOURCE_TRANSCRIPT,
456            Some(transcript_path),
457            &report,
458        )?;
459    }
460    Ok(report)
461}
462
463fn with_raw_archive_drain_savepoint<T>(
464    conn: &Connection,
465    f: impl FnOnce() -> Result<T>,
466) -> Result<std::result::Result<T, anyhow::Error>> {
467    conn.execute_batch("SAVEPOINT remem_raw_archive_drain;")
468        .context("start raw archive drain savepoint")?;
469    match f() {
470        Ok(value) => {
471            conn.execute_batch("RELEASE SAVEPOINT remem_raw_archive_drain;")
472                .context("release raw archive drain savepoint")?;
473            Ok(Ok(value))
474        }
475        Err(error) => {
476            let rollback = conn.execute_batch(
477                "ROLLBACK TO SAVEPOINT remem_raw_archive_drain;
478                 RELEASE SAVEPOINT remem_raw_archive_drain;",
479            );
480            match rollback {
481                Ok(()) => Ok(Err(error)),
482                Err(rollback_error) => Err(error).context(format!(
483                    "raw archive drain rollback also failed: {rollback_error}"
484                )),
485            }
486        }
487    }
488}
489
490pub fn record_raw_ingest_failure(
491    conn: &Connection,
492    session_id: &str,
493    project: &str,
494    source: &str,
495    transcript_path: Option<&str>,
496    report: &RawIngestReport,
497) -> Result<()> {
498    let Some(kind) = report.failure_kind() else {
499        return Ok(());
500    };
501    conn.execute(
502        "INSERT INTO raw_ingest_failures
503         (project, session_id, source, transcript_path, error_kind, error_message,
504          inserted, duplicates, parse_errors, insert_errors, created_at_epoch)
505         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
506        params![
507            project,
508            session_id,
509            source,
510            transcript_path,
511            kind,
512            crate::db::truncate_str(&report.failure_message(), 1000),
513            report.inserted as i64,
514            report.duplicates as i64,
515            report.parse_errors as i64,
516            report.insert_errors as i64,
517            chrono::Utc::now().timestamp()
518        ],
519    )?;
520    Ok(())
521}
522
523#[derive(Debug, Clone)]
524pub struct RawSearchRequest {
525    pub query: String,
526    pub project: Option<String>,
527    pub branch: Option<String>,
528    pub role: Option<String>,
529    pub limit: i64,
530    pub offset: i64,
531    /// Inclusive lower bound on `created_at_epoch`. None keeps the
532    /// pre-window behavior (issue #723).
533    pub since_epoch: Option<i64>,
534    /// Inclusive upper bound on `created_at_epoch`. None keeps the
535    /// pre-window behavior (issue #723).
536    pub until_epoch: Option<i64>,
537}
538
539pub fn search_raw_messages(conn: &Connection, req: &RawSearchRequest) -> Result<Vec<RawMessage>> {
540    let limit = req.limit.max(1);
541    let offset = req.offset.max(0);
542    let query = req.query.trim();
543    if query.is_empty() {
544        return Ok(vec![]);
545    }
546
547    let mut sql = String::from(
548        "SELECT r.id, r.session_id, r.project, r.role, r.content, r.source, \
549                r.branch, r.cwd, r.created_at_epoch \
550         FROM raw_messages r \
551         JOIN raw_messages_fts f ON f.rowid = r.id \
552         WHERE raw_messages_fts MATCH ?1",
553    );
554    let mut binds: Vec<Box<dyn rusqlite::types::ToSql>> = vec![Box::new(fts_query(query))];
555
556    if let Some(project) = req.project.as_deref() {
557        sql.push_str(" AND r.project = ?");
558        sql.push_str(&(binds.len() + 1).to_string());
559        binds.push(Box::new(project.to_string()));
560    }
561    if let Some(branch) = req.branch.as_deref() {
562        let idx = binds.len() + 1;
563        sql.push_str(&format!(" AND (r.branch = ?{idx} OR r.branch IS NULL)"));
564        binds.push(Box::new(branch.to_string()));
565    }
566    if let Some(role) = req.role.as_deref() {
567        sql.push_str(" AND r.role = ?");
568        sql.push_str(&(binds.len() + 1).to_string());
569        binds.push(Box::new(role.to_string()));
570    }
571    if let Some(since) = req.since_epoch {
572        sql.push_str(" AND r.created_at_epoch >= ?");
573        sql.push_str(&(binds.len() + 1).to_string());
574        binds.push(Box::new(since));
575    }
576    if let Some(until) = req.until_epoch {
577        sql.push_str(" AND r.created_at_epoch <= ?");
578        sql.push_str(&(binds.len() + 1).to_string());
579        binds.push(Box::new(until));
580    }
581
582    sql.push_str(&format!(
583        " ORDER BY r.created_at_epoch DESC LIMIT {} OFFSET {}",
584        limit, offset
585    ));
586
587    let mut stmt = conn.prepare(&sql)?;
588    let rows = stmt.query_map(
589        rusqlite::params_from_iter(crate::db::to_sql_refs(&binds)),
590        |row| {
591            Ok(RawMessage {
592                id: row.get(0)?,
593                session_id: row.get(1)?,
594                project: row.get(2)?,
595                role: row.get(3)?,
596                content: row.get(4)?,
597                source: row.get(5)?,
598                branch: row.get(6)?,
599                cwd: row.get(7)?,
600                created_at_epoch: row.get(8)?,
601            })
602        },
603    )?;
604
605    let mut out = Vec::new();
606    for row in rows {
607        out.push(row?);
608    }
609    Ok(out)
610}
611
612/// Parse a time bound given as Unix epoch seconds, an ISO8601 datetime, or a
613/// plain `YYYY-MM-DD` date interpreted as UTC midnight.
614///
615/// This public compatibility entry point retains its original date semantics.
616/// Query surfaces that need an inclusive date-only upper bound use the
617/// transport-neutral upper-bound parser instead.
618pub fn parse_time_bound(value: &str) -> Result<i64> {
619    super::raw_query::parse_time_lower_bound(value)
620}
621
622fn fts_query(query: &str) -> String {
623    // Wrap each token in quotes so we use phrase matching (robust against
624    // punctuation that trigram tokenizer would otherwise choke on).
625    let cleaned: Vec<String> = query
626        .split_whitespace()
627        .filter(|token| !token.is_empty())
628        .map(|token| format!("\"{}\"", token.replace('\"', "\"\"")))
629        .collect();
630    if cleaned.is_empty() {
631        format!("\"{}\"", query.replace('\"', "\"\""))
632    } else {
633        cleaned.join(" ")
634    }
635}
636
637#[cfg(test)]
638mod tests;
639#[cfg(test)]
640mod tests_session_labels;
remem/memory/raw_archive.rs

remem/memory/
raw_archive.rs
