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, TSIDX_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    // Replace `output` and drop both sidecars in one transaction (5a-2b):
145    // they describe the bytes being replaced (offsets, receive times), and a
146    // stale output.idx could pass the covered_size check by coincidence.
147    global
148        .replace_file(zone, OUTPUT_FILE, bytes, &[TSIDX_FILE, "output.idx"])
149        .map_err(|e| format!("replace_file: {e}"))?;
150    let hwm = count_nonblank_lines(bytes);
151    let overlay = serde_json::json!({
152        "schemaVersion": 2,
153        "savedAt": "",
154        "highWaterMark": hwm,
155        "sourceBlockId": "",
156        "documentState": {
157            "collapsedNodeIds": [],
158            "pinnedNodeIds": [],
159            "scrollPosition": 0,
160            "filter": {
161                "showThinking": false,
162                "showSuccessfulTools": true,
163                "showFailedTools": true,
164                "showIncoming": true,
165                "showOutgoing": true
166            }
167        },
168        "paneState": { "detailsOpen": false }
169    });
170    let overlay_bytes = serde_json::to_vec(&overlay).map_err(|e| format!("overlay json: {e}"))?;
171    overwrite_zone_file(global, zone, SNAPSHOT_FILE, &overlay_bytes)?;
172    Ok(())
173}
174
175/// Create-or-replace a zone file with exactly `bytes` (FileStore `write_file`
176/// requires the file to exist first).
177fn overwrite_zone_file(fs: &FileStore, zone: &str, name: &str, bytes: &[u8]) -> Result<(), String> {
178    use crate::backend::storage::error::StoreError;
179    if fs.stat(zone, name).map_err(|e| format!("stat: {e}"))?.is_none() {
180        match fs.make_file(zone, name, FileMeta::default(), FileOpts::default()) {
181            Ok(()) => {}
182            Err(StoreError::AlreadyExists) => {}
183            Err(e) => return Err(format!("make_file: {e}")),
184        }
185    }
186    fs.write_file(zone, name, bytes).map_err(|e| format!("write_file: {e}"))
187}
188
189fn count_nonblank_lines(bytes: &[u8]) -> u64 {
190    String::from_utf8_lossy(bytes)
191        .lines()
192        .filter(|l| !l.trim().is_empty())
193        .count() as u64
194}
195
196/// Read all blocks from `objdb` whose `meta.agentId` is in `want`, returning
197/// `(def_id, block_oid)` pairs. Read-only.
198fn read_agent_blocks(
199    objdb: &Path,
200    want: &HashSet<&str>,
201) -> Result<Vec<(String, String)>, rusqlite::Error> {
202    let conn = Connection::open_with_flags(objdb, OpenFlags::SQLITE_OPEN_READ_ONLY)?;
203    let mut stmt = conn.prepare("SELECT data FROM db_block")?;
204    // `db_block.data` is declared TEXT but the mstore stores the JSON as a BLOB
205    // (`typeof = 'blob'`), so read raw bytes — `get::<String>` would error on
206    // every row and silently skip the whole DB. `from_slice` parses either.
207    let rows = stmt.query_map([], |row| row.get::<_, Vec<u8>>(0))?;
208    let mut out = Vec::new();
209    for r in rows {
210        let Ok(bytes) = r else { continue };
211        let Ok(v) = serde_json::from_slice::<serde_json::Value>(&bytes) else { continue };
212        let meta = v.get("meta");
213        // The agent definition id is stored under either `agentId` (current
214        // shape) or the legacy `agent:id` — match both, exactly like the
215        // canonical block scan in `agent_session/migrations/v1_blocks.rs`.
216        // Legacy blocks are a core part of the pre-existing population this
217        // backfill targets, so missing them would strand those conversations
218        // permanently (the marker is written after the scan). (#1403)
219        let agent_id = meta
220            .and_then(|m| m.get("agentId"))
221            .and_then(|x| x.as_str())
222            .or_else(|| meta.and_then(|m| m.get("agent:id")).and_then(|x| x.as_str()));
223        let oid = v.get("oid").and_then(|x| x.as_str());
224        if let (Some(aid), Some(oid)) = (agent_id, oid) {
225            if want.contains(aid) {
226                out.push((aid.to_string(), oid.to_string()));
227            }
228        }
229    }
230    Ok(out)
231}
232
233/// Reassemble a zone's `output` file bytes directly from `db_file_data`
234/// (read-only, avoids opening a second writable FileStore on another channel).
235/// Returns `None` when absent or empty. Agent `output` is non-circular, so
236/// concatenating parts in `partidx` order matches `FileStore::read_file`.
237fn read_zone_output(conn: &Connection, block_oid: &str) -> Option<Vec<u8>> {
238    let size: Option<i64> = conn
239        .query_row(
240            "SELECT size FROM db_wave_file WHERE zoneid = ?1 AND name = ?2",
241            rusqlite::params![block_oid, OUTPUT_FILE],
242            |r| r.get(0),
243        )
244        .ok();
245    if size.unwrap_or(0) <= 0 {
246        return None;
247    }
248    let mut stmt = conn
249        .prepare("SELECT data FROM db_file_data WHERE zoneid = ?1 AND name = ?2 ORDER BY partidx")
250        .ok()?;
251    let rows = stmt
252        .query_map(rusqlite::params![block_oid, OUTPUT_FILE], |r| {
253            r.get::<_, Vec<u8>>(0)
254        })
255        .ok()?;
256    let mut buf = Vec::new();
257    for r in rows {
258        if let Ok(part) = r {
259            buf.extend_from_slice(&part);
260        }
261    }
262    if buf.is_empty() {
263        None
264    } else {
265        Some(buf)
266    }
267}
268
269/// Every `<home>/channels/*/versions/*/data` and `<home>/dev/<branch>[/<sub>]/data`
270/// (mirrors `registry::def_migrate::collect_scan_dbs`, returning the data dir).
271fn collect_data_dirs(home: &Path) -> Vec<PathBuf> {
272    let mut dirs = Vec::new();
273    let mut add = |d: PathBuf| {
274        if d.join("db").join("objects.db").is_file() {
275            dirs.push(d);
276        }
277    };
278    for ch in dir_subdirs(&home.join("channels")) {
279        for v in dir_subdirs(&ch.join("versions")) {
280            add(v.join("data"));
281        }
282    }
283    for br in dir_subdirs(&home.join("dev")) {
284        add(br.join("data"));
285        for sub in dir_subdirs(&br) {
286            add(sub.join("data"));
287        }
288    }
289    dirs
290}
291
292fn dir_subdirs(p: &Path) -> Vec<PathBuf> {
293    let mut out = Vec::new();
294    if let Ok(rd) = std::fs::read_dir(p) {
295        for e in rd.flatten() {
296            let path = e.path();
297            if path.is_dir() {
298                out.push(path);
299            }
300        }
301    }
302    out
303}
304
305fn marker_version(marker: &Path) -> u32 {
306    std::fs::read_to_string(marker)
307        .ok()
308        .and_then(|s| s.trim().parse::<u32>().ok())
309        .unwrap_or(0)
310}
311
312fn write_marker(marker: &Path) -> std::io::Result<()> {
313    if let Some(parent) = marker.parent() {
314        let _ = std::fs::create_dir_all(parent);
315    }
316    std::fs::write(marker, BACKFILL_VERSION.to_string())
317}
318
319#[cfg(test)]
320mod tests {
321    use super::*;
322    use crate::backend::agent_session::OUTPUT_FILE as OUT;
323    use std::sync::Arc;
324
325    // Build a channel data dir at <home>/channels/<ch>/versions/<v>/data with a
326    // db_block row (agentId=def) and a filestore.db holding that block's output.
327    fn seed_channel(home: &Path, ch: &str, def_id: &str, block_oid: &str, output: &[u8]) {
328        seed_channel_keyed(home, ch, "agentId", def_id, block_oid, output);
329    }
330
331    fn seed_channel_keyed(
332        home: &Path,
333        ch: &str,
334        meta_key: &str,
335        def_id: &str,
336        block_oid: &str,
337        output: &[u8],
338    ) {
339        let data = home
340            .join("channels")
341            .join(ch)
342            .join("versions")
343            .join("0.44.1")
344            .join("data");
345        let db = data.join("db");
346        std::fs::create_dir_all(&db).unwrap();
347        // objects.db with a minimal db_block(data) row.
348        let oc = Connection::open(db.join("objects.db")).unwrap();
349        oc.execute("CREATE TABLE db_block (data TEXT)", []).unwrap();
350        let block_json = serde_json::json!({"oid": block_oid, "meta": {"view":"agent", meta_key: def_id}});
351        // Store as a BLOB to mirror the mstore (db_block.data is declared TEXT
352        // but holds blob-typed JSON) — the reader reads raw bytes.
353        oc.execute(
354            "INSERT INTO db_block (data) VALUES (?1)",
355            rusqlite::params![block_json.to_string().into_bytes()],
356        )
357        .unwrap();
358        drop(oc);
359        // filestore.db with the block's output written via FileStore.
360        let fs = FileStore::open(&db.join("filestore.db")).unwrap();
361        fs.make_file(block_oid, OUT, FileMeta::default(), FileOpts::default()).unwrap();
362        fs.append_data(block_oid, OUT, output).unwrap();
363    }
364
365    #[test]
366    fn backfill_seeds_global_zone_from_richest_channel_block() {
367        let tmp = tempfile::tempdir().unwrap();
368        let home = tmp.path();
369        let tdir = home.join("shared").join("agents").join("transcripts");
370        std::fs::create_dir_all(&tdir).unwrap();
371        let def_id = "11111111-2222-3333-4444-555555555555";
372
373        // Two channels for the same agent; the richer (bigger) output wins.
374        let small = b"{\"type\":\"system\"}\n";
375        let big = b"{\"type\":\"system\"}\n{\"type\":\"assistant\"}\n{\"type\":\"result\"}\n";
376        seed_channel(home, "verify-a", def_id, "blk-small", small);
377        seed_channel(home, "verify-b", def_id, "blk-big", big);
378
379        let global = Arc::new(FileStore::open(&tdir.join("filestore.db")).unwrap());
380        let stats = backfill_transcripts_once(home, &tdir, &[def_id.to_string()], &global);
381
382        assert_eq!(stats.seeded, 1, "should seed the agent zone");
383        let zone = agent_current_zone(def_id);
384        let out = global.read_file(&zone, OUT).unwrap().unwrap();
385        assert_eq!(out, big, "global zone seeded with the RICHEST channel output");
386        // Snapshot overlay: v2, hwm = non-blank line count, sourceBlockId empty.
387        let snap = global.read_file(&zone, SNAPSHOT_FILE).unwrap().unwrap();
388        let o: serde_json::Value = serde_json::from_slice(&snap).unwrap();
389        assert_eq!(o["schemaVersion"], 2);
390        assert_eq!(o["highWaterMark"], 3);
391        assert_eq!(o["sourceBlockId"], "");
392
393        // Idempotent: second run is a no-op (marker written).
394        let stats2 = backfill_transcripts_once(home, &tdir, &[def_id.to_string()], &global);
395        assert_eq!(stats2.seeded, 0);
396        assert_eq!(stats2.data_dirs_scanned, 0, "marker short-circuits the re-run");
397    }
398
399    #[test]
400    fn backfill_matches_legacy_agent_id_key() {
401        // Blocks from older builds stamp the legacy `agent:id` meta key instead
402        // of `agentId`; the backfill must still recover them. (reagent P1 #1403.)
403        let tmp = tempfile::tempdir().unwrap();
404        let home = tmp.path();
405        let tdir = home.join("shared").join("agents").join("transcripts");
406        std::fs::create_dir_all(&tdir).unwrap();
407        let def_id = "deadbeef-0000-1111-2222-333344445555";
408        seed_channel_keyed(home, "legacy-ch", "agent:id", def_id, "blk-legacy", b"{\"type\":\"system\"}\n{\"type\":\"result\"}\n");
409
410        let global = Arc::new(FileStore::open(&tdir.join("filestore.db")).unwrap());
411        let stats = backfill_transcripts_once(home, &tdir, &[def_id.to_string()], &global);
412        assert_eq!(stats.seeded, 1, "legacy agent:id block must be recovered");
413        let out = global.read_file(&agent_current_zone(def_id), OUT).unwrap().unwrap();
414        assert!(out.starts_with(b"{\"type\":\"system\"}"));
415    }
416
417    #[test]
418    fn backfill_does_not_clobber_richer_global_zone() {
419        let tmp = tempfile::tempdir().unwrap();
420        let home = tmp.path();
421        let tdir = home.join("shared").join("agents").join("transcripts");
422        std::fs::create_dir_all(&tdir).unwrap();
423        let def_id = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee";
424        seed_channel(home, "verify-a", def_id, "blk1", b"{\"type\":\"system\"}\n");
425
426        // Global already holds MORE than the channel source → must not clobber.
427        let global = Arc::new(FileStore::open(&tdir.join("filestore.db")).unwrap());
428        let zone = agent_current_zone(def_id);
429        global.make_file(&zone, OUT, FileMeta::default(), FileOpts::default()).unwrap();
430        global.write_file(&zone, OUT, b"{\"a\":1}\n{\"b\":2}\n{\"c\":3}\n{\"d\":4}\n").unwrap();
431
432        let stats = backfill_transcripts_once(home, &tdir, &[def_id.to_string()], &global);
433        assert_eq!(stats.seeded, 0);
434        assert_eq!(stats.skipped_global_richer, 1);
435        let out = global.read_file(&zone, OUT).unwrap().unwrap();
436        assert!(out.starts_with(b"{\"a\":1}"), "richer global content preserved");
437    }
438}
agentmux_srv/backend/transcript_backfill.rs

agentmux_srv/backend/
transcript_backfill.rs
