Skip to main content

ito_core/audit/
mirror.rs

1//! Best-effort synchronization of the local audit log to a dedicated remote branch.
2
3use std::collections::HashSet;
4use std::fs;
5use std::path::{Path, PathBuf};
6use std::time::{SystemTime, UNIX_EPOCH};
7
8use chrono::{DateTime, Duration, Utc};
9use ito_domain::audit::event::{AuditEvent, ops};
10
11use crate::process::{ProcessRequest, ProcessRunner, SystemProcessRunner};
12
13use super::writer::audit_log_path;
14
15const MAX_GIT_AUDIT_EVENTS: usize = 1000;
16const MAX_GIT_AUDIT_AGE_DAYS: i64 = 30;
17
18/// Failure details from an audit mirror sync attempt.
19#[derive(Debug, thiserror::Error)]
20#[error("{message}")]
21pub struct AuditMirrorError {
22    message: String,
23}
24
25impl AuditMirrorError {
26    fn new(message: impl Into<String>) -> Self {
27        Self {
28            message: message.into(),
29        }
30    }
31}
32
33/// Attempt to sync the local audit log to `origin/<branch>`.
34///
35/// This function returns an error when a sync attempt fails; callers that
36/// require best-effort semantics should log/print the error and continue.
37pub fn sync_audit_mirror(
38    repo_root: &Path,
39    ito_path: &Path,
40    branch: &str,
41) -> Result<(), AuditMirrorError> {
42    let runner = SystemProcessRunner;
43    sync_audit_mirror_with_runner(&runner, repo_root, ito_path, branch)
44}
45
46pub(crate) fn sync_audit_mirror_with_runner(
47    runner: &dyn ProcessRunner,
48    repo_root: &Path,
49    ito_path: &Path,
50    branch: &str,
51) -> Result<(), AuditMirrorError> {
52    if !is_git_worktree(runner, repo_root) {
53        return Ok(());
54    }
55
56    let local_log = audit_log_path(ito_path);
57    if !local_log.exists() {
58        return Ok(());
59    }
60
61    let worktree_path = unique_temp_worktree_path();
62    add_detached_worktree(runner, repo_root, &worktree_path)?;
63    let cleanup = WorktreeCleanup {
64        repo_root: repo_root.to_path_buf(),
65        worktree_path: worktree_path.clone(),
66    };
67
68    let result = (|| -> Result<(), AuditMirrorError> {
69        let fetched = fetch_branch(runner, repo_root, branch);
70        match fetched {
71            Ok(()) => checkout_detached_remote_branch(runner, &worktree_path, branch)?,
72            Err(FetchError::RemoteMissing) => checkout_orphan_branch(runner, &worktree_path)?,
73            Err(FetchError::Other(msg)) => return Err(AuditMirrorError::new(msg)),
74        }
75
76        write_merged_audit_log(
77            &local_log,
78            &worktree_path.join(".ito/.state/audit/events.jsonl"),
79        )?;
80        stage_audit_log(runner, &worktree_path)?;
81
82        if !has_staged_changes(runner, &worktree_path)? {
83            return Ok(());
84        }
85
86        commit_audit_log(runner, &worktree_path)?;
87        if push_branch(runner, &worktree_path, branch)? {
88            return Ok(());
89        }
90
91        // Retry once on non-fast-forward by refetching and re-merging.
92        let fetched = fetch_branch(runner, repo_root, branch);
93        match fetched {
94            Ok(()) => checkout_detached_remote_branch(runner, &worktree_path, branch)?,
95            Err(FetchError::RemoteMissing) => checkout_orphan_branch(runner, &worktree_path)?,
96            Err(FetchError::Other(msg)) => return Err(AuditMirrorError::new(msg)),
97        }
98        write_merged_audit_log(
99            &local_log,
100            &worktree_path.join(".ito/.state/audit/events.jsonl"),
101        )?;
102        stage_audit_log(runner, &worktree_path)?;
103        if has_staged_changes(runner, &worktree_path)? {
104            commit_audit_log(runner, &worktree_path)?;
105        }
106        if push_branch(runner, &worktree_path, branch)? {
107            return Ok(());
108        }
109
110        Err(AuditMirrorError::new(format!(
111            "audit mirror push to '{branch}' failed due to a remote conflict; try 'git fetch origin {branch}' and re-run, or disable mirroring with 'ito config set audit.mirror.enabled false'"
112        )))
113    })();
114
115    let cleanup_err = cleanup.cleanup_with_runner(runner);
116    if let Err(err) = cleanup_err {
117        eprintln!(
118            "Warning: failed to remove temporary audit mirror worktree '{}': {}",
119            cleanup.worktree_path.display(),
120            err
121        );
122    }
123    result
124}
125
126pub(crate) fn append_jsonl_to_internal_branch(
127    repo_root: &Path,
128    branch: &str,
129    jsonl: &str,
130) -> Result<(), AuditMirrorError> {
131    let runner = SystemProcessRunner;
132    append_jsonl_to_internal_branch_with_runner(&runner, repo_root, branch, jsonl)
133}
134
135pub(crate) fn append_jsonl_to_internal_branch_with_runner(
136    runner: &dyn ProcessRunner,
137    repo_root: &Path,
138    branch: &str,
139    jsonl: &str,
140) -> Result<(), AuditMirrorError> {
141    if !is_git_worktree(runner, repo_root) {
142        return Err(AuditMirrorError::new(
143            "internal audit branch unavailable outside a git worktree",
144        ));
145    }
146
147    let mut allow_retry = true;
148    loop {
149        match append_jsonl_to_internal_branch_attempt(runner, repo_root, branch, jsonl)? {
150            AppendBranchResult::Appended => return Ok(()),
151            AppendBranchResult::Conflict if allow_retry => {
152                allow_retry = false;
153            }
154            AppendBranchResult::Conflict => {
155                return Err(AuditMirrorError::new(format!(
156                    "failed to update internal audit branch '{branch}' due to concurrent writes; retry the command"
157                )));
158            }
159        }
160    }
161}
162
163enum AppendBranchResult {
164    Appended,
165    Conflict,
166}
167
168fn append_jsonl_to_internal_branch_attempt(
169    runner: &dyn ProcessRunner,
170    repo_root: &Path,
171    branch: &str,
172    jsonl: &str,
173) -> Result<AppendBranchResult, AuditMirrorError> {
174    let expected_old = current_branch_oid(runner, repo_root, branch)?;
175
176    let worktree_path = unique_temp_worktree_path();
177    add_detached_worktree(runner, repo_root, &worktree_path)?;
178    let cleanup = WorktreeCleanup {
179        repo_root: repo_root.to_path_buf(),
180        worktree_path: worktree_path.clone(),
181    };
182
183    let result = (|| -> Result<AppendBranchResult, AuditMirrorError> {
184        if expected_old.is_some() {
185            checkout_detached_local_branch(runner, &worktree_path, branch)?;
186        } else {
187            checkout_orphan_branch(runner, &worktree_path)?;
188        }
189
190        write_merged_jsonl(&worktree_path.join(".ito/.state/audit/events.jsonl"), jsonl)?;
191        stage_audit_log(runner, &worktree_path)?;
192
193        if !has_staged_changes(runner, &worktree_path)? {
194            return Ok(AppendBranchResult::Appended);
195        }
196
197        commit_internal_audit_log(runner, &worktree_path)?;
198        match update_branch_ref(runner, &worktree_path, branch, expected_old.as_deref())? {
199            UpdateRefResult::Updated => Ok(AppendBranchResult::Appended),
200            UpdateRefResult::Conflict => Ok(AppendBranchResult::Conflict),
201        }
202    })();
203
204    let cleanup_err = cleanup.cleanup_with_runner(runner);
205    if let Err(err) = cleanup_err {
206        eprintln!(
207            "Warning: failed to remove temporary audit worktree '{}': {}",
208            cleanup.worktree_path.display(),
209            err
210        );
211    }
212    result
213}
214
215fn current_branch_oid(
216    runner: &dyn ProcessRunner,
217    repo_root: &Path,
218    branch: &str,
219) -> Result<Option<String>, AuditMirrorError> {
220    let out = runner
221        .run(
222            &ProcessRequest::new("git")
223                .args(["rev-parse", "--verify", &format!("refs/heads/{branch}")])
224                .current_dir(repo_root),
225        )
226        .map_err(|e| AuditMirrorError::new(format!("git rev-parse failed: {e}")))?;
227    if out.success {
228        let oid = out.stdout.trim();
229        return Ok((!oid.is_empty()).then(|| oid.to_string()));
230    }
231    let detail = render_output(&out).to_ascii_lowercase();
232    if detail.contains("unknown revision") || detail.contains("needed a single revision") {
233        return Ok(None);
234    }
235    Err(AuditMirrorError::new(format!(
236        "failed to inspect internal audit branch '{branch}' ({})",
237        render_output(&out)
238    )))
239}
240
241pub(crate) fn read_internal_branch_log(
242    repo_root: &Path,
243    branch: &str,
244) -> Result<InternalBranchLogRead, AuditMirrorError> {
245    let runner = SystemProcessRunner;
246    read_internal_branch_log_with_runner(&runner, repo_root, branch)
247}
248
249#[derive(Debug, Clone, PartialEq, Eq)]
250pub(crate) enum InternalBranchLogRead {
251    BranchMissing,
252    LogMissing,
253    Contents(String),
254}
255
256pub(crate) fn read_internal_branch_log_with_runner(
257    runner: &dyn ProcessRunner,
258    repo_root: &Path,
259    branch: &str,
260) -> Result<InternalBranchLogRead, AuditMirrorError> {
261    if !is_git_worktree(runner, repo_root) {
262        return Err(AuditMirrorError::new(
263            "internal audit branch unavailable outside a git worktree",
264        ));
265    }
266
267    if !local_branch_exists(runner, repo_root, branch)? {
268        return Ok(InternalBranchLogRead::BranchMissing);
269    }
270
271    let pathspec = format!("refs/heads/{branch}:.ito/.state/audit/events.jsonl");
272    let out = runner
273        .run(
274            &ProcessRequest::new("git")
275                .args(["show", &pathspec])
276                .current_dir(repo_root),
277        )
278        .map_err(|e| AuditMirrorError::new(format!("git show failed: {e}")))?;
279    if out.success {
280        return Ok(InternalBranchLogRead::Contents(out.stdout));
281    }
282
283    let detail = render_output(&out).to_ascii_lowercase();
284    if detail.contains("does not exist in")
285        || detail.contains("path '.ito/.state/audit/events.jsonl' does not exist")
286    {
287        return Ok(InternalBranchLogRead::LogMissing);
288    }
289
290    Err(AuditMirrorError::new(format!(
291        "failed to read internal audit branch '{branch}' ({})",
292        render_output(&out)
293    )))
294}
295
296fn write_merged_audit_log(local_log: &Path, target_log: &Path) -> Result<(), AuditMirrorError> {
297    let local = fs::read_to_string(local_log)
298        .map_err(|e| AuditMirrorError::new(format!("failed to read local audit log: {e}")))?;
299    let remote = fs::read_to_string(target_log).unwrap_or_default();
300
301    let merged = merge_jsonl_lines(&remote, &local);
302
303    write_jsonl(target_log, &merged)
304}
305
306fn write_merged_jsonl(target_log: &Path, jsonl: &str) -> Result<(), AuditMirrorError> {
307    let existing = fs::read_to_string(target_log).unwrap_or_default();
308    let merged = merge_jsonl_lines(&existing, jsonl);
309
310    write_jsonl(target_log, &merged)
311}
312
313fn write_jsonl(target_log: &Path, contents: &str) -> Result<(), AuditMirrorError> {
314    if let Some(parent) = target_log.parent() {
315        fs::create_dir_all(parent).map_err(|e| {
316            AuditMirrorError::new(format!("failed to create audit mirror dir: {e}"))
317        })?;
318    }
319    fs::write(target_log, contents)
320        .map_err(|e| AuditMirrorError::new(format!("failed to write audit mirror log: {e}")))?;
321    Ok(())
322}
323
324fn merge_jsonl_lines(remote: &str, local: &str) -> String {
325    let mut out: Vec<String> = Vec::new();
326    let mut seen: HashSet<String> = HashSet::new();
327    let mut last_reconcile_key: Option<String> = None;
328
329    for line in remote.lines() {
330        if line.trim().is_empty() {
331            continue;
332        }
333        let reconcile_key = stable_reconcile_key(line);
334        if reconcile_key.is_some() && reconcile_key == last_reconcile_key {
335            increment_last_reconcile_count(&mut out);
336            continue;
337        }
338        out.push(line.to_string());
339        seen.insert(line.to_string());
340        last_reconcile_key = reconcile_key;
341    }
342
343    for line in local.lines() {
344        if line.trim().is_empty() {
345            continue;
346        }
347        let reconcile_key = stable_reconcile_key(line);
348        if reconcile_key.is_some() && reconcile_key == last_reconcile_key {
349            increment_last_reconcile_count(&mut out);
350            continue;
351        }
352        if seen.contains(line) {
353            continue;
354        }
355        out.push(line.to_string());
356        seen.insert(line.to_string());
357        last_reconcile_key = reconcile_key;
358    }
359
360    let out = truncate_git_audit_events(out);
361
362    if out.is_empty() {
363        return String::new();
364    }
365    // Ensure trailing newline so subsequent appends are line-oriented.
366    format!("{}\n", out.join("\n"))
367}
368
369fn truncate_git_audit_events(lines: Vec<String>) -> Vec<String> {
370    let mut events: Vec<(usize, DateTime<Utc>, String)> = Vec::new();
371    for (index, line) in lines.into_iter().enumerate() {
372        let Some(ts) = audit_event_timestamp(&line) else {
373            continue;
374        };
375        events.push((index, ts, line));
376    }
377    let Some((_, newest, _)) = events.iter().max_by_key(|(_, ts, _)| *ts) else {
378        return Vec::new();
379    };
380    let cutoff = *newest - Duration::days(MAX_GIT_AUDIT_AGE_DAYS);
381    let mut events: Vec<(usize, DateTime<Utc>, String)> = events
382        .into_iter()
383        .filter(|(_, ts, _)| *ts >= cutoff)
384        .collect();
385    if events.len() > MAX_GIT_AUDIT_EVENTS {
386        events.sort_by_key(|(index, ts, _)| (*ts, *index));
387        events.drain(0..events.len() - MAX_GIT_AUDIT_EVENTS);
388    }
389    events.sort_by_key(|(index, _, _)| *index);
390    events.into_iter().map(|(_, _, line)| line).collect()
391}
392
393fn audit_event_timestamp(line: &str) -> Option<DateTime<Utc>> {
394    let Ok(event) = serde_json::from_str::<AuditEvent>(line) else {
395        return None;
396    };
397    let Ok(ts) = DateTime::parse_from_rfc3339(&event.ts) else {
398        return None;
399    };
400    Some(ts.with_timezone(&Utc))
401}
402
403fn increment_last_reconcile_count(out: &mut [String]) {
404    let Some(last) = out.last_mut() else {
405        return;
406    };
407    let Ok(mut event) = serde_json::from_str::<AuditEvent>(last) else {
408        return;
409    };
410    if event.op != ops::RECONCILED {
411        return;
412    }
413    event.count = event.count.saturating_add(1);
414    let Ok(serialized) = serde_json::to_string(&event) else {
415        return;
416    };
417    *last = serialized;
418}
419
420fn stable_reconcile_key(line: &str) -> Option<String> {
421    let Ok(event) = serde_json::from_str::<AuditEvent>(line) else {
422        return None;
423    };
424    if event.op != ops::RECONCILED {
425        return None;
426    }
427
428    Some(format!(
429        "{}\u{1f}{}\u{1f}{:?}\u{1f}{:?}\u{1f}{:?}\u{1f}{}\u{1f}{}",
430        event.entity, event.entity_id, event.scope, event.from, event.to, event.actor, event.by
431    ))
432}
433
434fn add_detached_worktree(
435    runner: &dyn ProcessRunner,
436    repo_root: &Path,
437    worktree_path: &Path,
438) -> Result<(), AuditMirrorError> {
439    let out = runner
440        .run(
441            &ProcessRequest::new("git")
442                .args([
443                    "worktree",
444                    "add",
445                    "--detach",
446                    worktree_path.to_string_lossy().as_ref(),
447                ])
448                .current_dir(repo_root),
449        )
450        .map_err(|e| AuditMirrorError::new(format!("git worktree add failed: {e}")))?;
451    if out.success {
452        return Ok(());
453    }
454    Err(AuditMirrorError::new(format!(
455        "git worktree add failed ({})",
456        render_output(&out)
457    )))
458}
459
460#[derive(Debug)]
461enum FetchError {
462    RemoteMissing,
463    Other(String),
464}
465
466fn fetch_branch(
467    runner: &dyn ProcessRunner,
468    repo_root: &Path,
469    branch: &str,
470) -> Result<(), FetchError> {
471    let out = runner
472        .run(
473            &ProcessRequest::new("git")
474                .args(["fetch", "origin", branch])
475                .current_dir(repo_root),
476        )
477        .map_err(|e| FetchError::Other(format!("git fetch origin {branch} failed to run: {e}")))?;
478    if out.success {
479        return Ok(());
480    }
481    let detail = render_output(&out);
482    if detail.contains("couldn't find remote ref") {
483        return Err(FetchError::RemoteMissing);
484    }
485    Err(FetchError::Other(format!(
486        "git fetch origin {branch} failed ({detail})"
487    )))
488}
489
490fn checkout_detached_remote_branch(
491    runner: &dyn ProcessRunner,
492    worktree_path: &Path,
493    branch: &str,
494) -> Result<(), AuditMirrorError> {
495    let target = format!("origin/{branch}");
496    let out = runner
497        .run(
498            &ProcessRequest::new("git")
499                .args(["checkout", "--detach", &target])
500                .current_dir(worktree_path),
501        )
502        .map_err(|e| AuditMirrorError::new(format!("git checkout failed: {e}")))?;
503    if out.success {
504        return Ok(());
505    }
506    Err(AuditMirrorError::new(format!(
507        "failed to checkout audit mirror branch '{branch}' ({})",
508        render_output(&out)
509    )))
510}
511
512fn checkout_detached_local_branch(
513    runner: &dyn ProcessRunner,
514    worktree_path: &Path,
515    branch: &str,
516) -> Result<(), AuditMirrorError> {
517    let target = format!("refs/heads/{branch}");
518    let out = runner
519        .run(
520            &ProcessRequest::new("git")
521                .args(["checkout", "--detach", &target])
522                .current_dir(worktree_path),
523        )
524        .map_err(|e| AuditMirrorError::new(format!("git checkout failed: {e}")))?;
525    if out.success {
526        return Ok(());
527    }
528    Err(AuditMirrorError::new(format!(
529        "failed to checkout internal audit branch '{branch}' ({})",
530        render_output(&out)
531    )))
532}
533
534fn checkout_orphan_branch(
535    runner: &dyn ProcessRunner,
536    worktree_path: &Path,
537) -> Result<(), AuditMirrorError> {
538    let orphan = unique_orphan_branch_name();
539    let out = runner
540        .run(
541            &ProcessRequest::new("git")
542                .args(["checkout", "--orphan", orphan.as_str()])
543                .current_dir(worktree_path),
544        )
545        .map_err(|e| AuditMirrorError::new(format!("git checkout --orphan failed: {e}")))?;
546    if !out.success {
547        return Err(AuditMirrorError::new(format!(
548            "failed to create orphan audit mirror worktree ({})",
549            render_output(&out)
550        )));
551    }
552
553    // Remove tracked files from the index to keep the mirror branch focused.
554    let _ = runner.run(
555        &ProcessRequest::new("git")
556            .args(["rm", "-rf", "."]) // best-effort; may fail on empty trees
557            .current_dir(worktree_path),
558    );
559    Ok(())
560}
561
562fn stage_audit_log(
563    runner: &dyn ProcessRunner,
564    worktree_path: &Path,
565) -> Result<(), AuditMirrorError> {
566    let relative = ".ito/.state/audit/events.jsonl";
567    let out = runner
568        .run(
569            &ProcessRequest::new("git")
570                .args(["add", "-f", relative])
571                .current_dir(worktree_path),
572        )
573        .map_err(|e| AuditMirrorError::new(format!("git add failed: {e}")))?;
574    if out.success {
575        return Ok(());
576    }
577    Err(AuditMirrorError::new(format!(
578        "failed to stage audit mirror log ({})",
579        render_output(&out)
580    )))
581}
582
583fn has_staged_changes(
584    runner: &dyn ProcessRunner,
585    worktree_path: &Path,
586) -> Result<bool, AuditMirrorError> {
587    let relative = ".ito/.state/audit/events.jsonl";
588    let out = runner
589        .run(
590            &ProcessRequest::new("git")
591                .args(["diff", "--cached", "--quiet", "--", relative])
592                .current_dir(worktree_path),
593        )
594        .map_err(|e| AuditMirrorError::new(format!("git diff --cached failed: {e}")))?;
595    if out.success {
596        return Ok(false);
597    }
598    if out.exit_code == 1 {
599        return Ok(true);
600    }
601    Err(AuditMirrorError::new(format!(
602        "failed to inspect staged audit mirror changes ({})",
603        render_output(&out)
604    )))
605}
606
607fn commit_audit_log(
608    runner: &dyn ProcessRunner,
609    worktree_path: &Path,
610) -> Result<(), AuditMirrorError> {
611    let message = "chore(audit): mirror events";
612    let out = runner
613        .run(
614            &ProcessRequest::new("git")
615                .args(["commit", "-m", message])
616                .current_dir(worktree_path),
617        )
618        .map_err(|e| AuditMirrorError::new(format!("git commit failed: {e}")))?;
619    if out.success {
620        return Ok(());
621    }
622    Err(AuditMirrorError::new(format!(
623        "failed to commit audit mirror update ({})",
624        render_output(&out)
625    )))
626}
627
628fn commit_internal_audit_log(
629    runner: &dyn ProcessRunner,
630    worktree_path: &Path,
631) -> Result<(), AuditMirrorError> {
632    let message = "chore(audit): update internal log";
633    let out = runner
634        .run(
635            &ProcessRequest::new("git")
636                .args(["commit", "-m", message])
637                .current_dir(worktree_path),
638        )
639        .map_err(|e| AuditMirrorError::new(format!("git commit failed: {e}")))?;
640    if out.success {
641        return Ok(());
642    }
643    Err(AuditMirrorError::new(format!(
644        "failed to commit internal audit log ({})",
645        render_output(&out)
646    )))
647}
648
649enum UpdateRefResult {
650    Updated,
651    Conflict,
652}
653
654fn update_branch_ref(
655    runner: &dyn ProcessRunner,
656    worktree_path: &Path,
657    branch: &str,
658    expected_old: Option<&str>,
659) -> Result<UpdateRefResult, AuditMirrorError> {
660    let target = format!("refs/heads/{branch}");
661    let new_oid = branch_head_oid(runner, worktree_path)?;
662    let expected = expected_old.unwrap_or("0000000000000000000000000000000000000000");
663    let out = runner
664        .run(
665            &ProcessRequest::new("git")
666                .args(["update-ref", &target, &new_oid, expected])
667                .current_dir(worktree_path),
668        )
669        .map_err(|e| AuditMirrorError::new(format!("git update-ref failed: {e}")))?;
670    if out.success {
671        return Ok(UpdateRefResult::Updated);
672    }
673    let detail = render_output(&out);
674    let lower = detail.to_ascii_lowercase();
675    if lower.contains("cannot lock ref")
676        || lower.contains("is at ")
677        || lower.contains("reference already exists")
678    {
679        return Ok(UpdateRefResult::Conflict);
680    }
681    Err(AuditMirrorError::new(format!(
682        "failed to update internal audit branch '{branch}' ({})",
683        detail
684    )))
685}
686
687fn branch_head_oid(
688    runner: &dyn ProcessRunner,
689    worktree_path: &Path,
690) -> Result<String, AuditMirrorError> {
691    let out = runner
692        .run(
693            &ProcessRequest::new("git")
694                .args(["rev-parse", "HEAD"])
695                .current_dir(worktree_path),
696        )
697        .map_err(|e| AuditMirrorError::new(format!("git rev-parse HEAD failed: {e}")))?;
698    if out.success {
699        let oid = out.stdout.trim();
700        if !oid.is_empty() {
701            return Ok(oid.to_string());
702        }
703    }
704    Err(AuditMirrorError::new(format!(
705        "failed to resolve internal audit commit ({})",
706        render_output(&out)
707    )))
708}
709
710/// Push `HEAD` to `origin/<branch>`.
711///
712/// Returns `Ok(true)` when push succeeded, `Ok(false)` when the push was rejected due to non-fast-forward.
713fn push_branch(
714    runner: &dyn ProcessRunner,
715    worktree_path: &Path,
716    branch: &str,
717) -> Result<bool, AuditMirrorError> {
718    let refspec = format!("HEAD:refs/heads/{branch}");
719    let out = runner
720        .run(
721            &ProcessRequest::new("git")
722                .args(["push", "origin", &refspec])
723                .current_dir(worktree_path),
724        )
725        .map_err(|e| AuditMirrorError::new(format!("git push failed to run: {e}")))?;
726    if out.success {
727        return Ok(true);
728    }
729
730    let detail = render_output(&out);
731    if detail.contains("non-fast-forward") {
732        return Ok(false);
733    }
734
735    Err(AuditMirrorError::new(format!(
736        "audit mirror push failed ({detail})"
737    )))
738}
739
740fn local_branch_exists(
741    runner: &dyn ProcessRunner,
742    repo_root: &Path,
743    branch: &str,
744) -> Result<bool, AuditMirrorError> {
745    let target = format!("refs/heads/{branch}");
746    let out = runner
747        .run(
748            &ProcessRequest::new("git")
749                .args(["show-ref", "--verify", "--quiet", &target])
750                .current_dir(repo_root),
751        )
752        .map_err(|e| AuditMirrorError::new(format!("git show-ref failed: {e}")))?;
753    if out.success {
754        return Ok(true);
755    }
756    if out.exit_code == 1 {
757        return Ok(false);
758    }
759
760    Err(AuditMirrorError::new(format!(
761        "failed to inspect internal audit branch '{branch}' ({})",
762        render_output(&out)
763    )))
764}
765
766fn is_git_worktree(runner: &dyn ProcessRunner, repo_root: &Path) -> bool {
767    let out = runner.run(
768        &ProcessRequest::new("git")
769            .args(["rev-parse", "--is-inside-work-tree"])
770            .current_dir(repo_root),
771    );
772    let Ok(out) = out else {
773        return false;
774    };
775    out.success && out.stdout.trim() == "true"
776}
777
778fn render_output(out: &crate::process::ProcessOutput) -> String {
779    let stdout = out.stdout.trim();
780    let stderr = out.stderr.trim();
781
782    if !stderr.is_empty() {
783        return stderr.to_string();
784    }
785    if !stdout.is_empty() {
786        return stdout.to_string();
787    }
788    "no command output".to_string()
789}
790
791fn unique_temp_worktree_path() -> PathBuf {
792    let pid = std::process::id();
793    let nanos = match SystemTime::now().duration_since(UNIX_EPOCH) {
794        Ok(duration) => duration.as_nanos(),
795        Err(_) => 0,
796    };
797    std::env::temp_dir().join(format!("ito-audit-mirror-{pid}-{nanos}"))
798}
799
800fn unique_orphan_branch_name() -> String {
801    let pid = std::process::id();
802    let nanos = match SystemTime::now().duration_since(UNIX_EPOCH) {
803        Ok(duration) => duration.as_nanos(),
804        Err(_) => 0,
805    };
806    format!("ito-audit-mirror-orphan-{pid}-{nanos}")
807}
808
809struct WorktreeCleanup {
810    repo_root: PathBuf,
811    worktree_path: PathBuf,
812}
813
814impl WorktreeCleanup {
815    fn cleanup_with_runner(&self, runner: &dyn ProcessRunner) -> Result<(), String> {
816        let out = runner.run(
817            &ProcessRequest::new("git")
818                .args([
819                    "worktree",
820                    "remove",
821                    "--force",
822                    self.worktree_path.to_string_lossy().as_ref(),
823                ])
824                .current_dir(&self.repo_root),
825        );
826        if let Err(err) = out {
827            return Err(format!("git worktree remove failed: {err}"));
828        }
829
830        // Ensure the directory is gone even if git left it behind.
831        let _ = fs::remove_dir_all(&self.worktree_path);
832        Ok(())
833    }
834}
835
836impl Drop for WorktreeCleanup {
837    fn drop(&mut self) {
838        // Best-effort panic-safety: if callers unwind before `git worktree remove`
839        // runs, still remove the directory to avoid littering temp.
840        let _ = fs::remove_dir_all(&self.worktree_path);
841    }
842}
843
844#[cfg(test)]
845mod tests {
846    use super::*;
847
848    #[test]
849    fn merge_jsonl_dedupes_and_appends_local_lines() {
850        let a = task_event_id("2026-04-26T11:28:00.000Z", "1", "create", "pending");
851        let b = task_event_id("2026-04-26T11:28:01.000Z", "2", "create", "pending");
852        let c = task_event_id("2026-04-26T11:28:02.000Z", "3", "create", "pending");
853        let remote = format!("{a}\n{b}\n");
854        let local = format!("{b}\n{c}\n");
855        let merged = merge_jsonl_lines(&remote, &local);
856        assert_eq!(merged, format!("{a}\n{b}\n{c}\n"));
857    }
858
859    #[test]
860    fn merge_jsonl_ignores_blank_lines() {
861        let event = task_event("2026-04-26T11:28:00.000Z", "create", "pending");
862        let remote = format!("\n{event}\n\n");
863        let local = "\n\n";
864        let merged = merge_jsonl_lines(&remote, local);
865        assert_eq!(merged, format!("{event}\n"));
866    }
867
868    #[test]
869    fn merge_jsonl_aggregates_adjacent_equivalent_reconciled_events() {
870        let remote = format!(
871            "{}\n",
872            reconcile_event("2026-04-26T11:28:01.873Z", "pending")
873        );
874        let local = format!(
875            "{}\n",
876            reconcile_event("2026-04-26T11:28:21.643Z", "pending")
877        );
878
879        let merged = merge_jsonl_lines(&remote, &local);
880        let mut lines = merged.lines();
881        let event: serde_json::Value = serde_json::from_str(lines.next().unwrap()).unwrap();
882        assert_eq!(event["count"], serde_json::json!(2));
883        assert_eq!(lines.next(), None);
884    }
885
886    #[test]
887    fn merge_jsonl_keeps_reconciled_events_after_different_event() {
888        let remote = format!(
889            "{}\n{}\n",
890            reconcile_event("2026-04-26T11:28:01.873Z", "pending"),
891            task_event("2026-04-26T11:28:10.000Z", "status_change", "complete")
892        );
893        let local = format!(
894            "{}\n",
895            reconcile_event("2026-04-26T11:28:21.643Z", "pending")
896        );
897
898        let merged = merge_jsonl_lines(&remote, &local);
899        assert_eq!(merged, format!("{}{}", remote, local));
900    }
901
902    #[test]
903    fn merge_jsonl_drops_events_older_than_one_month_from_newest_event() {
904        let old = task_event_id("2026-03-01T00:00:00.000Z", "old", "create", "pending");
905        let recent = task_event_id("2026-03-28T00:00:00.000Z", "recent", "create", "pending");
906        let newest = task_event_id("2026-04-26T00:00:00.000Z", "newest", "create", "pending");
907        let remote = format!("{old}\n{recent}\n");
908        let local = format!("{newest}\n");
909
910        let merged = merge_jsonl_lines(&remote, &local);
911        assert_eq!(merged, format!("{recent}\n{newest}\n"));
912    }
913
914    #[test]
915    fn merge_jsonl_caps_git_log_to_newest_1000_events() {
916        let mut local = String::new();
917        for i in 0..1005 {
918            local.push_str(&task_event_id(
919                "2026-04-26T00:00:00.000Z",
920                &i.to_string(),
921                "create",
922                "pending",
923            ));
924            local.push('\n');
925        }
926
927        let merged = merge_jsonl_lines("", &local);
928        let lines: Vec<&str> = merged.lines().collect();
929        assert_eq!(lines.len(), 1000);
930        assert!(lines[0].contains("\"entity_id\":\"5\""));
931        assert!(lines[999].contains("\"entity_id\":\"1004\""));
932    }
933
934    #[test]
935    fn merge_jsonl_count_cap_uses_timestamp_not_input_position() {
936        let old = task_event_id("2026-04-01T00:00:00.000Z", "old", "create", "pending");
937        let mut local = format!("{old}\n");
938        for i in 0..1000 {
939            local.push_str(&task_event_id(
940                "2026-04-26T00:00:00.000Z",
941                &i.to_string(),
942                "create",
943                "pending",
944            ));
945            local.push('\n');
946        }
947
948        let merged = merge_jsonl_lines("", &local);
949        let lines: Vec<&str> = merged.lines().collect();
950        assert_eq!(lines.len(), 1000);
951        assert!(!merged.contains("\"entity_id\":\"old\""));
952    }
953
954    fn reconcile_event(ts: &str, from: &str) -> String {
955        serde_json::json!({
956            "v": 1,
957            "ts": ts,
958            "entity": "task",
959            "entity_id": "3.2",
960            "scope": "001-33_enhance-spec-driven-workflow-validation",
961            "op": "reconciled",
962            "from": from,
963            "actor": "reconcile",
964            "by": "@reconcile",
965            "meta": {
966                "reason": "task '3.2' has audit status 'pending' but no file entry"
967            },
968            "ctx": {
969                "session_id": "test",
970                "branch": "main",
971                "worktree": "main",
972                "commit": "abc123"
973            }
974        })
975        .to_string()
976    }
977
978    fn task_event(ts: &str, op: &str, to: &str) -> String {
979        task_event_id(ts, "3.2", op, to)
980    }
981
982    fn task_event_id(ts: &str, entity_id: &str, op: &str, to: &str) -> String {
983        serde_json::json!({
984            "v": 1,
985            "ts": ts,
986            "entity": "task",
987            "entity_id": entity_id,
988            "scope": "001-33_enhance-spec-driven-workflow-validation",
989            "op": op,
990            "to": to,
991            "actor": "cli",
992            "by": "@test",
993            "ctx": {
994                "session_id": "test"
995            }
996        })
997        .to_string()
998    }
999}