tj_core/dream/
mod.rs

1//! Dream — offline memory passes over session transcripts.
2//!
3//! Pass A (backfill): re-read a session transcript and append the
4//! significant typed events the realtime classifier missed. Additive —
5//! the JSONL source of truth is never mutated.
6
7pub mod backend;
8pub mod backfill;
9pub mod http;
10pub mod prompt;
11pub mod scope;
12pub mod state;
13
14use crate::dream::backend::{BackfillInput, DreamBackend};
15
16pub struct DreamOptions {
17    pub project_hash: String,
18    /// If true, do not call the backend or write anything; report scope only.
19    pub dry_run: bool,
20}
21
22#[derive(Debug, Default, PartialEq)]
23pub struct DreamReport {
24    pub sessions_processed: usize,
25    pub events_backfilled: usize,
26}
27
28/// Run one dream Pass A over the given sessions, using the supplied
29/// backend. `sessions` is a list of (session_id, BackfillInput) the
30/// caller has already assembled from transcripts + existing events.
31pub fn run_dream(
32    conn: &rusqlite::Connection,
33    events_path: &std::path::Path,
34    opts: &DreamOptions,
35    backend: &dyn DreamBackend,
36    sessions: Vec<(String, BackfillInput)>,
37    run_id: &str,
38) -> anyhow::Result<DreamReport> {
39    let mut report = DreamReport::default();
40    for (session_id, input) in sessions {
41        report.sessions_processed += 1;
42        if opts.dry_run {
43            continue;
44        }
45        let proposed = backend.backfill(&input)?;
46        // Flatten existing texts across candidate tasks for the guard.
47        let existing: Vec<String> = input
48            .tasks
49            .iter()
50            .flat_map(|t| t.existing_events.clone())
51            .collect();
52        let kept = crate::dream::backfill::dedup_guard(proposed, &existing);
53        let mut writer = crate::storage::JsonlWriter::open(events_path)?;
54        for b in &kept {
55            let e = crate::dream::backfill::to_event(b, run_id, &session_id);
56            writer.append(&e)?;
57            crate::db::upsert_task_from_event(conn, &e, &opts.project_hash)?;
58            crate::db::index_event(conn, &e)?;
59            report.events_backfilled += 1;
60        }
61        writer.flush_durable()?;
62    }
63    Ok(report)
64}
65
66#[cfg(test)]
67mod tests {
68    use super::*;
69    use crate::dream::backend::{BackfillEvent, BackfillTaskContext, MockDreamBackend};
70    use crate::event::{Author, Event, EventType, Source};
71    use tempfile::TempDir;
72
73    fn task_input() -> (String, BackfillInput) {
74        (
75            "sess-1".to_string(),
76            BackfillInput {
77                tasks: vec![BackfillTaskContext {
78                    task_id: "tj-1".into(),
79                    title: "Demo".into(),
80                    existing_events: vec!["Already known fact.".into()],
81                }],
82                transcript: "user: ...\nassistant: ...".into(),
83            },
84        )
85    }
86
87    #[test]
88    fn run_dream_appends_novel_events_and_indexes() {
89        let d = TempDir::new().unwrap();
90        let conn = crate::db::open(d.path().join("s.sqlite")).unwrap();
91        let events_path = d.path().join("events.jsonl");
92
93        // Seed the task so upsert/index has a home (open event).
94        let open = Event::new(
95            "tj-1",
96            EventType::Open,
97            Author::User,
98            Source::Cli,
99            "Demo".into(),
100        );
101        crate::db::upsert_task_from_event(&conn, &open, "ph").unwrap();
102
103        let backend = MockDreamBackend {
104            events: vec![
105                BackfillEvent {
106                    event_type: EventType::Finding,
107                    task_id: "tj-1".into(),
108                    text: "A brand new finding.".into(),
109                    timestamp: "2026-06-08T10:00:00Z".into(),
110                },
111                BackfillEvent {
112                    event_type: EventType::Finding,
113                    task_id: "tj-1".into(),
114                    text: "Already known fact.".into(), // dup → dropped
115                    timestamp: "2026-06-08T10:01:00Z".into(),
116                },
117            ],
118        };
119        let opts = DreamOptions {
120            project_hash: "ph".into(),
121            dry_run: false,
122        };
123        let report = run_dream(
124            &conn,
125            &events_path,
126            &opts,
127            &backend,
128            vec![task_input()],
129            "run-1",
130        )
131        .unwrap();
132
133        assert_eq!(report.sessions_processed, 1);
134        assert_eq!(report.events_backfilled, 1); // dup dropped
135        let body = std::fs::read_to_string(&events_path).unwrap();
136        assert!(body.contains("A brand new finding."));
137        assert!(body.contains("\"source\":\"dream\""));
138        assert!(!body.contains("\"text\":\"Already known fact.\",\"refs\""));
139    }
140
141    #[test]
142    fn dry_run_writes_nothing_and_skips_backend() {
143        let d = TempDir::new().unwrap();
144        let conn = crate::db::open(d.path().join("s.sqlite")).unwrap();
145        let events_path = d.path().join("events.jsonl");
146        let backend = MockDreamBackend { events: vec![] };
147        let opts = DreamOptions {
148            project_hash: "ph".into(),
149            dry_run: true,
150        };
151        let report = run_dream(
152            &conn,
153            &events_path,
154            &opts,
155            &backend,
156            vec![task_input()],
157            "run-1",
158        )
159        .unwrap();
160        assert_eq!(report.sessions_processed, 1);
161        assert_eq!(report.events_backfilled, 0);
162        assert!(!events_path.exists());
163    }
164}
tj_core/dream/mod.rs

tj_core/dream/
mod.rs
