codex_sync/sync/
codex.rs

1//! Transport-independent merge engine for Codex rollouts, SQLite state, and index data.
2
3use super::core;
4use crate::history;
5use anyhow::{Context, Result, bail};
6use rusqlite::{Connection, OptionalExtension, params};
7use serde::{Deserialize, Serialize};
8use serde_json::Value;
9use std::{
10    collections::HashSet,
11    fs,
12    path::Path,
13    time::{SystemTime, UNIX_EPOCH},
14};
15use walkdir::WalkDir;
16
17const MAX_BUNDLE_FILE_SIZE: u64 = 1024 * 1024 * 1024;
18const MAX_BUNDLE_FILES: usize = 100_000;
19
20#[derive(Debug, Default, Serialize, Deserialize)]
21pub struct MergeReport {
22    pub copied: usize,
23    pub skipped: usize,
24    pub conflicts: usize,
25    pub removed: usize,
26    pub database_rows: usize,
27    #[serde(default)]
28    pub index_entries_added: usize,
29    pub backup: String,
30}
31
32pub fn create_bundle(codex_home: &Path, bundle: &Path) -> Result<()> {
33    let files = bundle.join("files");
34    fs::create_dir_all(&files)?;
35    for directory in ["sessions", "archived_sessions", "attachments"] {
36        let source = codex_home.join(directory);
37        if source.is_dir() {
38            copy_tree(&source, &files.join(directory))?;
39        }
40    }
41    for name in ["session_index.jsonl", "history.jsonl"] {
42        let source = codex_home.join(name);
43        if source.is_file() {
44            core::atomic_copy(&source, &files.join(name))?;
45        }
46    }
47    let state = codex_home.join("state_5.sqlite");
48    if state.is_file() {
49        let target = files.join("state_5.sqlite");
50        let connection =
51            Connection::open_with_flags(state, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY)?;
52        connection.execute("VACUUM INTO ?1", params![target.to_string_lossy().as_ref()])?;
53    }
54    let node = hostname::get()?.to_string_lossy().into_owned();
55    let manifest = core::manifest_tree(
56        &files,
57        node.clone(),
58        node,
59        MAX_BUNDLE_FILE_SIZE,
60        MAX_BUNDLE_FILES,
61        |relative| allowed(&relative.to_string_lossy().replace('\\', "/")),
62    )?;
63    core::atomic_write(
64        &bundle.join(core::MANIFEST_FILE),
65        &serde_json::to_vec_pretty(&manifest)?,
66    )
67}
68
69pub fn apply_bundle(bundle: &Path, codex_home: &Path, mirror: bool) -> Result<MergeReport> {
70    fs::create_dir_all(codex_home)?;
71    let manifest = core::read_manifest(bundle).context("接收包缺少或无法读取 manifest.json")?;
72    core::validate_manifest(
73        codex_home,
74        &manifest,
75        MAX_BUNDLE_FILE_SIZE,
76        MAX_BUNDLE_FILES,
77    )?;
78    if let Some(file) = manifest.files.iter().find(|file| !allowed(&file.path)) {
79        bail!("拒绝不允许的同步文件：{}", file.path);
80    }
81    core::verify_snapshot(bundle, &manifest, MAX_BUNDLE_FILE_SIZE)?;
82
83    let backup = codex_home
84        .join("codex-sync-backups")
85        .join(format!("sync-import-{}", now()));
86    fs::create_dir_all(&backup)?;
87    let provider = active_model_provider(codex_home);
88    let mut report = MergeReport {
89        backup: backup.to_string_lossy().into_owned(),
90        ..Default::default()
91    };
92    if mirror {
93        report.removed = remove_target_only_files(codex_home, &manifest, &backup)?;
94    }
95
96    for file in &manifest.files {
97        if is_structured(file.path.as_str()) {
98            continue;
99        }
100        let source = core::safe_path(&bundle.join("files"), &file.path)?;
101        let target = core::safe_path(codex_home, &file.path)?;
102        if target.exists() {
103            let target_hash = blake3::hash(&fs::read(&target)?).to_hex();
104            if target_hash.as_str() == file.hash {
105                report.skipped += 1;
106                continue;
107            }
108            if !mirror {
109                if is_rollout(&file.path) {
110                    core::atomic_copy(&target, &backup.join("merged").join(&file.path))?;
111                    merge_rollout(&source, &target, provider.as_deref())?;
112                    report.copied += 1;
113                    continue;
114                }
115                report.conflicts += 1;
116                continue;
117            }
118            core::atomic_copy(&target, &backup.join("replaced").join(&file.path))?;
119        }
120        if is_rollout(&file.path) {
121            copy_rollout(&source, &target, provider.as_deref())?;
122        } else {
123            core::atomic_copy(&source, &target)?;
124        }
125        report.copied += 1;
126    }
127
128    for name in ["session_index.jsonl", "history.jsonl"] {
129        let source = bundle.join("files").join(name);
130        if !source.is_file() {
131            continue;
132        }
133        let target = codex_home.join(name);
134        if mirror {
135            if target.is_file() {
136                core::atomic_copy(&target, &backup.join(name))?;
137            }
138            core::atomic_copy(&source, &target)?;
139        } else {
140            merge_jsonl(&source, &target, &backup.join(name))?;
141        }
142    }
143
144    let source_database = bundle.join("files/state_5.sqlite");
145    let target_database = codex_home.join("state_5.sqlite");
146    if source_database.is_file() {
147        if target_database.is_file() {
148            report.database_rows = merge_database(
149                &source_database,
150                &target_database,
151                &backup.join("state_5.sqlite"),
152                provider.as_deref(),
153                mirror,
154            )?;
155        } else {
156            core::atomic_copy(&source_database, &target_database)?;
157        }
158    }
159    report.index_entries_added = history::reconcile_session_index(codex_home)?;
160    Ok(report)
161}
162
163fn is_structured(path: &str) -> bool {
164    matches!(
165        path,
166        "state_5.sqlite" | "session_index.jsonl" | "history.jsonl"
167    )
168}
169
170fn is_rollout(path: &str) -> bool {
171    path.starts_with("sessions/") || path.starts_with("archived_sessions/")
172}
173
174fn allowed(path: &str) -> bool {
175    is_structured(path) || is_rollout(path) || path.starts_with("attachments/")
176}
177
178fn remove_target_only_files(
179    codex_home: &Path,
180    manifest: &core::Manifest,
181    backup: &Path,
182) -> Result<usize> {
183    let wanted: HashSet<&str> = manifest
184        .files
185        .iter()
186        .map(|file| file.path.as_str())
187        .collect();
188    let mut removed = 0;
189    for root in ["sessions", "archived_sessions", "attachments"] {
190        let directory = codex_home.join(root);
191        if !directory.is_dir() {
192            continue;
193        }
194        for entry in WalkDir::new(&directory)
195            .follow_links(false)
196            .into_iter()
197            .filter_map(Result::ok)
198        {
199            if !entry.file_type().is_file() {
200                continue;
201            }
202            let relative = entry
203                .path()
204                .strip_prefix(codex_home)?
205                .to_string_lossy()
206                .replace('\\', "/");
207            if wanted.contains(relative.as_str()) {
208                continue;
209            }
210            core::atomic_copy(entry.path(), &backup.join("removed").join(&relative))?;
211            fs::remove_file(entry.path())?;
212            removed += 1;
213        }
214    }
215    Ok(removed)
216}
217
218fn copy_tree(source: &Path, target: &Path) -> Result<()> {
219    for entry in WalkDir::new(source)
220        .follow_links(false)
221        .into_iter()
222        .filter_map(Result::ok)
223    {
224        if !entry.file_type().is_file() {
225            continue;
226        }
227        let relative = entry.path().strip_prefix(source)?;
228        core::atomic_copy(entry.path(), &target.join(relative))?;
229    }
230    Ok(())
231}
232
233fn copy_rollout(source: &Path, target: &Path, provider: Option<&str>) -> Result<()> {
234    write_rollout(&[source], target, provider)
235}
236
237fn merge_rollout(source: &Path, target: &Path, provider: Option<&str>) -> Result<()> {
238    write_rollout(&[target, source], target, provider)
239}
240
241fn write_rollout(sources: &[&Path], target: &Path, provider: Option<&str>) -> Result<()> {
242    let mut seen = HashSet::new();
243    let mut saw_session_meta = false;
244    let mut output = String::new();
245    for source in sources {
246        append_rollout(
247            &fs::read_to_string(source)?,
248            provider,
249            &mut seen,
250            &mut saw_session_meta,
251            &mut output,
252        )?;
253    }
254    core::atomic_write(target, output.as_bytes())
255}
256
257fn append_rollout(
258    text: &str,
259    provider: Option<&str>,
260    seen: &mut HashSet<String>,
261    saw_session_meta: &mut bool,
262    output: &mut String,
263) -> Result<()> {
264    for line in text.lines() {
265        let mut value: Value = serde_json::from_str(line)?;
266        if value.get("type").and_then(Value::as_str) == Some("session_meta") {
267            if *saw_session_meta {
268                continue;
269            }
270            *saw_session_meta = true;
271            if let Some(provider) = provider
272                && let Some(payload) = value.get_mut("payload").and_then(Value::as_object_mut)
273            {
274                payload.insert("model_provider".into(), Value::String(provider.into()));
275            }
276        }
277        let rendered = serde_json::to_string(&value)?;
278        if seen.insert(rendered.clone()) {
279            output.push_str(&rendered);
280            output.push('\n');
281        }
282    }
283    Ok(())
284}
285
286fn merge_jsonl(source: &Path, target: &Path, backup: &Path) -> Result<()> {
287    let existing = fs::read_to_string(target).unwrap_or_default();
288    if target.is_file() {
289        core::atomic_copy(target, backup)?;
290    }
291    let mut seen: HashSet<String> = existing.lines().map(str::to_owned).collect();
292    let mut output = existing;
293    if !output.is_empty() && !output.ends_with('\n') {
294        output.push('\n');
295    }
296    for line in fs::read_to_string(source)?.lines() {
297        if seen.insert(line.to_owned()) {
298            output.push_str(line);
299            output.push('\n');
300        }
301    }
302    core::atomic_write(target, output.as_bytes())
303}
304
305fn merge_database(
306    source: &Path,
307    target: &Path,
308    backup: &Path,
309    provider: Option<&str>,
310    mirror: bool,
311) -> Result<usize> {
312    let mut connection = Connection::open(target)?;
313    connection.execute("VACUUM INTO ?1", params![backup.to_string_lossy().as_ref()])?;
314    connection.execute(
315        "ATTACH DATABASE ?1 AS src",
316        params![source.to_string_lossy().as_ref()],
317    )?;
318    let transaction = connection.transaction()?;
319    let mut changed = 0;
320    if mirror {
321        for table in ["thread_dynamic_tools", "thread_spawn_edges"] {
322            if table_exists(&transaction, "main", table)? {
323                changed += transaction.execute(&format!("DELETE FROM main.\"{table}\""), [])?;
324            }
325        }
326        if table_exists(&transaction, "main", "threads")?
327            && table_exists(&transaction, "src", "threads")?
328        {
329            changed += transaction.execute(
330                "DELETE FROM main.threads WHERE id NOT IN (SELECT id FROM src.threads)",
331                [],
332            )?;
333        }
334    }
335    for table in ["threads", "thread_dynamic_tools", "thread_spawn_edges"] {
336        if !table_exists(&transaction, "main", table)? || !table_exists(&transaction, "src", table)?
337        {
338            continue;
339        }
340        let destination = table_columns(&transaction, "main", table)?;
341        let source_columns: HashSet<_> = table_columns(&transaction, "src", table)?
342            .into_iter()
343            .collect();
344        let columns: Vec<_> = destination
345            .into_iter()
346            .filter(|column| source_columns.contains(column))
347            .collect();
348        if columns.is_empty() {
349            continue;
350        }
351        let quoted = columns
352            .iter()
353            .map(|column| quote(column))
354            .collect::<Vec<_>>();
355        let selected = columns
356            .iter()
357            .map(|column| {
358                if table == "threads" && column == "model_provider" && provider.is_some() {
359                    "?1".to_owned()
360                } else {
361                    quote(column)
362                }
363            })
364            .collect::<Vec<_>>();
365        let verb = if mirror {
366            "INSERT OR REPLACE"
367        } else {
368            "INSERT OR IGNORE"
369        };
370        let sql = format!(
371            "{verb} INTO main.\"{table}\" ({}) SELECT {} FROM src.\"{table}\"",
372            quoted.join(","),
373            selected.join(",")
374        );
375        changed += if let ("threads", Some(provider)) = (table, provider) {
376            transaction.execute(&sql, [provider])?
377        } else {
378            transaction.execute(&sql, [])?
379        };
380    }
381    transaction.commit()?;
382    Ok(changed)
383}
384
385fn quote(identifier: &str) -> String {
386    format!("\"{}\"", identifier.replace('"', "\"\""))
387}
388
389fn table_exists(connection: &Connection, schema: &str, table: &str) -> Result<bool> {
390    Ok(connection
391        .query_row(
392            &format!("SELECT 1 FROM {schema}.sqlite_master WHERE type='table' AND name=?1"),
393            [table],
394            |_| Ok(()),
395        )
396        .optional()?
397        .is_some())
398}
399
400fn table_columns(connection: &Connection, schema: &str, table: &str) -> Result<Vec<String>> {
401    let mut statement = connection.prepare(&format!("PRAGMA {schema}.table_info('{table}')"))?;
402    Ok(statement
403        .query_map([], |row| row.get(1))?
404        .collect::<rusqlite::Result<Vec<_>>>()?)
405}
406
407fn active_model_provider(codex_home: &Path) -> Option<String> {
408    let text = fs::read_to_string(codex_home.join("config.toml")).ok()?;
409    let value: toml::Value = toml::from_str(&text).ok()?;
410    Some(
411        value
412            .get("model_provider")
413            .and_then(toml::Value::as_str)
414            .unwrap_or("openai")
415            .to_owned(),
416    )
417}
418
419fn now() -> u64 {
420    SystemTime::now()
421        .duration_since(UNIX_EPOCH)
422        .unwrap_or_default()
423        .as_secs()
424}
425
426#[cfg(test)]
427mod tests {
428    use super::*;
429
430    fn temp_dir() -> std::path::PathBuf {
431        std::env::temp_dir().join(format!(
432            "codex-sync-merge-test-{}-{}",
433            std::process::id(),
434            SystemTime::now()
435                .duration_since(UNIX_EPOCH)
436                .unwrap_or_default()
437                .as_nanos()
438        ))
439    }
440
441    #[test]
442    fn bundle_merge_uses_target_provider_and_rebuilds_index() -> Result<()> {
443        let root = temp_dir();
444        let source = root.join("source");
445        let target = root.join("target");
446        let bundle = root.join("bundle");
447        fs::create_dir_all(source.join("sessions/2026/07/19"))?;
448        fs::create_dir_all(&target)?;
449        fs::write(
450            source.join("sessions/2026/07/19/rollout-local.jsonl"),
451            "{\"type\":\"session_meta\",\"payload\":{\"id\":\"local\",\"model_provider\":\"local\"}}\n",
452        )?;
453        fs::write(target.join("config.toml"), "model_provider = \"remote\"\n")?;
454        create_thread_database(&source.join("state_5.sqlite"), "local", "local")?;
455        create_thread_database(&target.join("state_5.sqlite"), "remote", "remote")?;
456
457        create_bundle(&source, &bundle)?;
458        let report = apply_bundle(&bundle, &target, false)?;
459        assert_eq!(report.copied, 1);
460        assert_eq!(report.database_rows, 1);
461        assert_eq!(report.index_entries_added, 2);
462        let rollout = fs::read_to_string(target.join("sessions/2026/07/19/rollout-local.jsonl"))?;
463        assert!(rollout.contains("\"model_provider\":\"remote\""));
464        assert_eq!(
465            fs::read_to_string(target.join("session_index.jsonl"))?
466                .lines()
467                .count(),
468            2
469        );
470        fs::remove_dir_all(root)?;
471        Ok(())
472    }
473
474    #[test]
475    fn rollout_merge_keeps_one_meta_and_both_event_sets() -> Result<()> {
476        let root = temp_dir();
477        fs::create_dir_all(&root)?;
478        let source = root.join("source.jsonl");
479        let target = root.join("target.jsonl");
480        fs::write(
481            &source,
482            "{\"type\":\"session_meta\",\"payload\":{\"id\":\"one\",\"model_provider\":\"local\"}}\n{\"type\":\"event\",\"id\":\"local\"}\n",
483        )?;
484        fs::write(
485            &target,
486            "{\"type\":\"session_meta\",\"payload\":{\"id\":\"one\",\"model_provider\":\"old\"}}\n{\"type\":\"session_meta\",\"payload\":{\"id\":\"one\"}}\n{\"type\":\"event\",\"id\":\"remote\"}\n",
487        )?;
488        merge_rollout(&source, &target, Some("openai"))?;
489        let merged = fs::read_to_string(&target)?;
490        assert_eq!(merged.matches("session_meta").count(), 1);
491        assert!(merged.contains("\"model_provider\":\"openai\""));
492        assert!(merged.contains("\"id\":\"local\""));
493        assert!(merged.contains("\"id\":\"remote\""));
494        fs::remove_dir_all(root)?;
495        Ok(())
496    }
497
498    fn create_thread_database(path: &Path, id: &str, provider: &str) -> Result<()> {
499        let connection = Connection::open(path)?;
500        connection.execute(
501            "CREATE TABLE threads (id TEXT PRIMARY KEY, model_provider TEXT NOT NULL)",
502            [],
503        )?;
504        connection.execute("INSERT INTO threads VALUES (?1, ?2)", params![id, provider])?;
505        Ok(())
506    }
507}
codex_sync/sync/codex.rs

codex_sync/sync/
codex.rs
