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

tj_core/dream/
mod.rs
