agentmux_srv\backend/
transcript_backfill.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! One-shot backfill of pre-existing agent CONVERSATIONS into the GLOBAL
5//! transcript store, so the 9 (and any) agents created before cross-channel
6//! transcripts shipped load their history when opened from a fresh channel.
7//!
8//! The merged feature (#1399) mirrors *new* agent output into the global
9//! `agent:<defId>:current` zone, but pre-existing conversations were never
10//! mirrored — they live only in the channel they ran in. This scans every
11//! channel/dev `objects.db` for blocks belonging to each agent definition,
12//! reads the richest `output` from that channel's `filestore.db`, and seeds the
13//! global zone with it.
14//!
15//! **Source = the channel block's `output`, NOT the provider `.jsonl`.** The
16//! block `output` is the exact stdout stream-json the renderer already parses
17//! (`parseHistoryLines` + the `claude-stream-json` translator); the provider
18//! `~/.claude/projects/<slug>/<sid>.jsonl` transcript is a *different* envelope
19//! (`queue-operation` / `user`+`parentUuid`/`sessionId`) the renderer can't
20//! consume, so copying it would render empty. Using the block output needs zero
21//! translation.
22//!
23//! The seeded snapshot overlay uses `sourceBlockId: ""`, which the frontend
24//! treats as "matches any block" — so opening the agent (a fresh local block
25//! with no local `output`) takes the v2 fast-path and the
26//! `blockfile:read_range` global fallback (#1399) serves the backfilled zone.
27//!
28//! Marker-gated (`.transcripts_backfilled`, version-stamped) like the
29//! definition backfill (`registry::def_migrate`); read-only on every scanned
30//! SQLite (the global store is the only thing written).
31
32use std::collections::{HashMap, HashSet};
33use std::path::{Path, PathBuf};
34
35use rusqlite::{Connection, OpenFlags};
36
37use crate::backend::agent_session::{
38    agent_current_zone, is_valid_definition_id, OUTPUT_FILE, SNAPSHOT_FILE,
39};
40use crate::backend::storage::filestore::{FileMeta, FileOpts, FileStore};
41
42const MARKER: &str = ".transcripts_backfilled";
43
44/// Bump to re-run the backfill once for everyone (e.g. when the source-selection
45/// logic improves).
46const BACKFILL_VERSION: u32 = 1;
47
48#[derive(Debug, Default, Clone, Copy)]
49pub struct BackfillStats {
50    pub agents_seen: usize,
51    pub data_dirs_scanned: usize,
52    pub seeded: usize,
53    pub skipped_no_source: usize,
54    pub skipped_global_richer: usize,
55}
56
57/// Seed the global transcript zone for each agent in `def_ids` from its richest
58/// channel/dev block `output`. Marker-gated under `transcripts_dir`. Best-effort
59/// — any per-DB failure is skipped, never fatal.
60pub fn backfill_transcripts_once(
61    home: &Path,
62    transcripts_dir: &Path,
63    def_ids: &[String],
64    global: &FileStore,
65) -> BackfillStats {
66    let mut stats = BackfillStats::default();
67    let marker = transcripts_dir.join(MARKER);
68    if marker_version(&marker) >= BACKFILL_VERSION {
69        return stats;
70    }
71
72    let want: HashSet<&str> = def_ids
73        .iter()
74        .map(|s| s.as_str())
75        .filter(|s| is_valid_definition_id(s))
76        .collect();
77    if want.is_empty() {
78        let _ = write_marker(&marker);
79        return stats;
80    }
81
82    // def_id -> richest output bytes found across all channels/dev DBs.
83    let mut best: HashMap<String, Vec<u8>> = HashMap::new();
84    for data in collect_data_dirs(home) {
85        let objdb = data.join("db").join("objects.db");
86        let fdb = data.join("db").join("filestore.db");
87        if !objdb.is_file() || !fdb.is_file() {
88            continue;
89        }
90        stats.data_dirs_scanned += 1;
91        let blocks = match read_agent_blocks(&objdb, &want) {
92            Ok(b) if !b.is_empty() => b,
93            _ => continue,
94        };
95        let conn = match Connection::open_with_flags(&fdb, OpenFlags::SQLITE_OPEN_READ_ONLY) {
96            Ok(c) => c,
97            Err(_) => continue,
98        };
99        for (def_id, block_oid) in blocks {
100            if let Some(bytes) = read_zone_output(&conn, &block_oid) {
101                let better = best.get(&def_id).map(|b| bytes.len() > b.len()).unwrap_or(true);
102                if better {
103                    best.insert(def_id, bytes);
104                }
105            }
106        }
107    }
108
109    for def_id in def_ids {
110        if !is_valid_definition_id(def_id) {
111            continue;
112        }
113        stats.agents_seen += 1;
114        let Some(bytes) = best.get(def_id) else {
115            stats.skipped_no_source += 1;
116            continue;
117        };
118        let zone = agent_current_zone(def_id);
119        // Don't clobber a global zone that already has equal/more content (the
120        // live mirror, or a richer prior backfill, may already own it).
121        let cur = global
122            .stat(&zone, OUTPUT_FILE)
123            .ok()
124            .flatten()
125            .map(|f| f.size)
126            .unwrap_or(0);
127        if (bytes.len() as i64) <= cur {
128            stats.skipped_global_richer += 1;
129            continue;
130        }
131        match seed_global_zone(global, &zone, bytes) {
132            Ok(()) => stats.seeded += 1,
133            Err(e) => tracing::warn!(zone = %zone, error = %e, "transcript backfill: seed failed"),
134        }
135    }
136
137    let _ = write_marker(&marker);
138    stats
139}
140
141/// Write `bytes` as the zone's `output` and a v2 snapshot overlay
142/// (`sourceBlockId: ""`) so the open path restores it cross-channel.
143fn seed_global_zone(global: &FileStore, zone: &str, bytes: &[u8]) -> Result<(), String> {
144    overwrite_zone_file(global, zone, OUTPUT_FILE, bytes)?;
145    let hwm = count_nonblank_lines(bytes);
146    let overlay = serde_json::json!({
147        "schemaVersion": 2,
148        "savedAt": "",
149        "highWaterMark": hwm,
150        "sourceBlockId": "",
151        "documentState": {
152            "collapsedNodeIds": [],
153            "pinnedNodeIds": [],
154            "scrollPosition": 0,
155            "filter": {
156                "showThinking": false,
157                "showSuccessfulTools": true,
158                "showFailedTools": true,
159                "showIncoming": true,
160                "showOutgoing": true
161            }
162        },
163        "paneState": { "detailsOpen": false }
164    });
165    let overlay_bytes = serde_json::to_vec(&overlay).map_err(|e| format!("overlay json: {e}"))?;
166    overwrite_zone_file(global, zone, SNAPSHOT_FILE, &overlay_bytes)?;
167    Ok(())
168}
169
170/// Create-or-replace a zone file with exactly `bytes` (FileStore `write_file`
171/// requires the file to exist first).
172fn overwrite_zone_file(fs: &FileStore, zone: &str, name: &str, bytes: &[u8]) -> Result<(), String> {
173    use crate::backend::storage::error::StoreError;
174    if fs.stat(zone, name).map_err(|e| format!("stat: {e}"))?.is_none() {
175        match fs.make_file(zone, name, FileMeta::default(), FileOpts::default()) {
176            Ok(()) => {}
177            Err(StoreError::AlreadyExists) => {}
178            Err(e) => return Err(format!("make_file: {e}")),
179        }
180    }
181    fs.write_file(zone, name, bytes).map_err(|e| format!("write_file: {e}"))
182}
183
184fn count_nonblank_lines(bytes: &[u8]) -> u64 {
185    String::from_utf8_lossy(bytes)
186        .lines()
187        .filter(|l| !l.trim().is_empty())
188        .count() as u64
189}
190
191/// Read all blocks from `objdb` whose `meta.agentId` is in `want`, returning
192/// `(def_id, block_oid)` pairs. Read-only.
193fn read_agent_blocks(
194    objdb: &Path,
195    want: &HashSet<&str>,
196) -> Result<Vec<(String, String)>, rusqlite::Error> {
197    let conn = Connection::open_with_flags(objdb, OpenFlags::SQLITE_OPEN_READ_ONLY)?;
198    let mut stmt = conn.prepare("SELECT data FROM db_block")?;
199    // `db_block.data` is declared TEXT but the wstore stores the JSON as a BLOB
200    // (`typeof = 'blob'`), so read raw bytes — `get::<String>` would error on
201    // every row and silently skip the whole DB. `from_slice` parses either.
202    let rows = stmt.query_map([], |row| row.get::<_, Vec<u8>>(0))?;
203    let mut out = Vec::new();
204    for r in rows {
205        let Ok(bytes) = r else { continue };
206        let Ok(v) = serde_json::from_slice::<serde_json::Value>(&bytes) else { continue };
207        let meta = v.get("meta");
208        // The agent definition id is stored under either `agentId` (current
209        // shape) or the legacy `agent:id` — match both, exactly like the
210        // canonical block scan in `agent_session.rs`. Legacy blocks are a core
211        // part of the pre-existing population this backfill targets, so missing
212        // them would strand those conversations permanently (the marker is
213        // written after the scan). (reagent P1 / codex P2 #1403.)
214        let agent_id = meta
215            .and_then(|m| m.get("agentId"))
216            .and_then(|x| x.as_str())
217            .or_else(|| meta.and_then(|m| m.get("agent:id")).and_then(|x| x.as_str()));
218        let oid = v.get("oid").and_then(|x| x.as_str());
219        if let (Some(aid), Some(oid)) = (agent_id, oid) {
220            if want.contains(aid) {
221                out.push((aid.to_string(), oid.to_string()));
222            }
223        }
224    }
225    Ok(out)
226}
227
228/// Reassemble a zone's `output` file bytes directly from `db_file_data`
229/// (read-only, avoids opening a second writable FileStore on another channel).
230/// Returns `None` when absent or empty. Agent `output` is non-circular, so
231/// concatenating parts in `partidx` order matches `FileStore::read_file`.
232fn read_zone_output(conn: &Connection, block_oid: &str) -> Option<Vec<u8>> {
233    let size: Option<i64> = conn
234        .query_row(
235            "SELECT size FROM db_wave_file WHERE zoneid = ?1 AND name = ?2",
236            rusqlite::params![block_oid, OUTPUT_FILE],
237            |r| r.get(0),
238        )
239        .ok();
240    if size.unwrap_or(0) <= 0 {
241        return None;
242    }
243    let mut stmt = conn
244        .prepare("SELECT data FROM db_file_data WHERE zoneid = ?1 AND name = ?2 ORDER BY partidx")
245        .ok()?;
246    let rows = stmt
247        .query_map(rusqlite::params![block_oid, OUTPUT_FILE], |r| {
248            r.get::<_, Vec<u8>>(0)
249        })
250        .ok()?;
251    let mut buf = Vec::new();
252    for r in rows {
253        if let Ok(part) = r {
254            buf.extend_from_slice(&part);
255        }
256    }
257    if buf.is_empty() {
258        None
259    } else {
260        Some(buf)
261    }
262}
263
264/// Every `<home>/channels/*/versions/*/data` and `<home>/dev/<branch>[/<sub>]/data`
265/// (mirrors `registry::def_migrate::collect_scan_dbs`, returning the data dir).
266fn collect_data_dirs(home: &Path) -> Vec<PathBuf> {
267    let mut dirs = Vec::new();
268    let mut add = |d: PathBuf| {
269        if d.join("db").join("objects.db").is_file() {
270            dirs.push(d);
271        }
272    };
273    for ch in dir_subdirs(&home.join("channels")) {
274        for v in dir_subdirs(&ch.join("versions")) {
275            add(v.join("data"));
276        }
277    }
278    for br in dir_subdirs(&home.join("dev")) {
279        add(br.join("data"));
280        for sub in dir_subdirs(&br) {
281            add(sub.join("data"));
282        }
283    }
284    dirs
285}
286
287fn dir_subdirs(p: &Path) -> Vec<PathBuf> {
288    let mut out = Vec::new();
289    if let Ok(rd) = std::fs::read_dir(p) {
290        for e in rd.flatten() {
291            let path = e.path();
292            if path.is_dir() {
293                out.push(path);
294            }
295        }
296    }
297    out
298}
299
300fn marker_version(marker: &Path) -> u32 {
301    std::fs::read_to_string(marker)
302        .ok()
303        .and_then(|s| s.trim().parse::<u32>().ok())
304        .unwrap_or(0)
305}
306
307fn write_marker(marker: &Path) -> std::io::Result<()> {
308    if let Some(parent) = marker.parent() {
309        let _ = std::fs::create_dir_all(parent);
310    }
311    std::fs::write(marker, BACKFILL_VERSION.to_string())
312}
313
314#[cfg(test)]
315mod tests {
316    use super::*;
317    use crate::backend::agent_session::OUTPUT_FILE as OUT;
318    use std::sync::Arc;
319
320    // Build a channel data dir at <home>/channels/<ch>/versions/<v>/data with a
321    // db_block row (agentId=def) and a filestore.db holding that block's output.
322    fn seed_channel(home: &Path, ch: &str, def_id: &str, block_oid: &str, output: &[u8]) {
323        seed_channel_keyed(home, ch, "agentId", def_id, block_oid, output);
324    }
325
326    fn seed_channel_keyed(
327        home: &Path,
328        ch: &str,
329        meta_key: &str,
330        def_id: &str,
331        block_oid: &str,
332        output: &[u8],
333    ) {
334        let data = home
335            .join("channels")
336            .join(ch)
337            .join("versions")
338            .join("0.44.1")
339            .join("data");
340        let db = data.join("db");
341        std::fs::create_dir_all(&db).unwrap();
342        // objects.db with a minimal db_block(data) row.
343        let oc = Connection::open(db.join("objects.db")).unwrap();
344        oc.execute("CREATE TABLE db_block (data TEXT)", []).unwrap();
345        let block_json = serde_json::json!({"oid": block_oid, "meta": {"view":"agent", meta_key: def_id}});
346        // Store as a BLOB to mirror the wstore (db_block.data is declared TEXT
347        // but holds blob-typed JSON) — the reader reads raw bytes.
348        oc.execute(
349            "INSERT INTO db_block (data) VALUES (?1)",
350            rusqlite::params![block_json.to_string().into_bytes()],
351        )
352        .unwrap();
353        drop(oc);
354        // filestore.db with the block's output written via FileStore.
355        let fs = FileStore::open(&db.join("filestore.db")).unwrap();
356        fs.make_file(block_oid, OUT, FileMeta::default(), FileOpts::default()).unwrap();
357        fs.append_data(block_oid, OUT, output).unwrap();
358    }
359
360    #[test]
361    fn backfill_seeds_global_zone_from_richest_channel_block() {
362        let tmp = tempfile::tempdir().unwrap();
363        let home = tmp.path();
364        let tdir = home.join("shared").join("agents").join("transcripts");
365        std::fs::create_dir_all(&tdir).unwrap();
366        let def_id = "11111111-2222-3333-4444-555555555555";
367
368        // Two channels for the same agent; the richer (bigger) output wins.
369        let small = b"{\"type\":\"system\"}\n";
370        let big = b"{\"type\":\"system\"}\n{\"type\":\"assistant\"}\n{\"type\":\"result\"}\n";
371        seed_channel(home, "verify-a", def_id, "blk-small", small);
372        seed_channel(home, "verify-b", def_id, "blk-big", big);
373
374        let global = Arc::new(FileStore::open(&tdir.join("filestore.db")).unwrap());
375        let stats = backfill_transcripts_once(home, &tdir, &[def_id.to_string()], &global);
376
377        assert_eq!(stats.seeded, 1, "should seed the agent zone");
378        let zone = agent_current_zone(def_id);
379        let out = global.read_file(&zone, OUT).unwrap().unwrap();
380        assert_eq!(out, big, "global zone seeded with the RICHEST channel output");
381        // Snapshot overlay: v2, hwm = non-blank line count, sourceBlockId empty.
382        let snap = global.read_file(&zone, SNAPSHOT_FILE).unwrap().unwrap();
383        let o: serde_json::Value = serde_json::from_slice(&snap).unwrap();
384        assert_eq!(o["schemaVersion"], 2);
385        assert_eq!(o["highWaterMark"], 3);
386        assert_eq!(o["sourceBlockId"], "");
387
388        // Idempotent: second run is a no-op (marker written).
389        let stats2 = backfill_transcripts_once(home, &tdir, &[def_id.to_string()], &global);
390        assert_eq!(stats2.seeded, 0);
391        assert_eq!(stats2.data_dirs_scanned, 0, "marker short-circuits the re-run");
392    }
393
394    #[test]
395    fn backfill_matches_legacy_agent_id_key() {
396        // Blocks from older builds stamp the legacy `agent:id` meta key instead
397        // of `agentId`; the backfill must still recover them. (reagent P1 #1403.)
398        let tmp = tempfile::tempdir().unwrap();
399        let home = tmp.path();
400        let tdir = home.join("shared").join("agents").join("transcripts");
401        std::fs::create_dir_all(&tdir).unwrap();
402        let def_id = "deadbeef-0000-1111-2222-333344445555";
403        seed_channel_keyed(home, "legacy-ch", "agent:id", def_id, "blk-legacy", b"{\"type\":\"system\"}\n{\"type\":\"result\"}\n");
404
405        let global = Arc::new(FileStore::open(&tdir.join("filestore.db")).unwrap());
406        let stats = backfill_transcripts_once(home, &tdir, &[def_id.to_string()], &global);
407        assert_eq!(stats.seeded, 1, "legacy agent:id block must be recovered");
408        let out = global.read_file(&agent_current_zone(def_id), OUT).unwrap().unwrap();
409        assert!(out.starts_with(b"{\"type\":\"system\"}"));
410    }
411
412    #[test]
413    fn backfill_does_not_clobber_richer_global_zone() {
414        let tmp = tempfile::tempdir().unwrap();
415        let home = tmp.path();
416        let tdir = home.join("shared").join("agents").join("transcripts");
417        std::fs::create_dir_all(&tdir).unwrap();
418        let def_id = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee";
419        seed_channel(home, "verify-a", def_id, "blk1", b"{\"type\":\"system\"}\n");
420
421        // Global already holds MORE than the channel source → must not clobber.
422        let global = Arc::new(FileStore::open(&tdir.join("filestore.db")).unwrap());
423        let zone = agent_current_zone(def_id);
424        global.make_file(&zone, OUT, FileMeta::default(), FileOpts::default()).unwrap();
425        global.write_file(&zone, OUT, b"{\"a\":1}\n{\"b\":2}\n{\"c\":3}\n{\"d\":4}\n").unwrap();
426
427        let stats = backfill_transcripts_once(home, &tdir, &[def_id.to_string()], &global);
428        assert_eq!(stats.seeded, 0);
429        assert_eq!(stats.skipped_global_richer, 1);
430        let out = global.read_file(&zone, OUT).unwrap().unwrap();
431        assert!(out.starts_with(b"{\"a\":1}"), "richer global content preserved");
432    }
433}
agentmux_srv\backend/transcript_backfill.rs

agentmux_srv\backend/
transcript_backfill.rs
