1use 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#[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
33pub 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 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 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 let _ = runner.run(
555 &ProcessRequest::new("git")
556 .args(["rm", "-rf", "."]) .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
710fn 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 let _ = fs::remove_dir_all(&self.worktree_path);
832 Ok(())
833 }
834}
835
836impl Drop for WorktreeCleanup {
837 fn drop(&mut self) {
838 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}