1use crate::git::shell_fetch;
2use crate::git::shell_push::PushFailureKind;
3use crate::git::shell_push::push_current_branch_with_result;
4use crate::git::utils::ensure_repo_ready_for_sync;
5use crate::git::utils::get_sync_branch;
6use crate::git::utils::is_worktree_dirty;
7use anyhow::Context;
8use anyhow::Result;
9use anyhow::bail;
10use chrono::DateTime;
11use chrono::Utc;
12use colored::Colorize;
13use git2::Commit;
14use git2::ErrorCode;
15use git2::Index;
16use git2::IndexAddOption;
17use git2::Oid;
18use git2::Repository;
19use git2::Signature;
20use git2::Tree;
21use serde::Deserialize;
22use serde::Serialize;
23use std::collections::HashMap;
24use std::path::Path;
25use std::time::Duration;
26use tokio::time::sleep;
27
28#[derive(Debug, Deserialize, Serialize)]
31struct LogEntryForMerge {
32 call_id: String,
33 started_at: DateTime<Utc>,
34 #[serde(flatten)]
35 rest: serde_json::Value,
36}
37
38fn is_tool_log_file(path: &str) -> bool {
44 if let Some(logs_idx) = path.find("/logs/") {
45 let after_logs = &path[logs_idx + 6..]; after_logs.starts_with("tool_logs_")
47 && std::path::Path::new(path)
48 .extension()
49 .is_some_and(|ext| ext.eq_ignore_ascii_case("jsonl"))
50 } else {
51 false
52 }
53}
54
55fn merge_jsonl_logs(ours_content: &[u8], theirs_content: &[u8]) -> Vec<u8> {
61 let mut records: HashMap<String, (DateTime<Utc>, String)> = HashMap::new();
62 let mut unparseable_lines: Vec<String> = Vec::new();
63
64 for line in String::from_utf8_lossy(ours_content).lines() {
66 if line.trim().is_empty() {
67 continue;
68 }
69 match serde_json::from_str::<LogEntryForMerge>(line) {
70 Ok(entry) => {
71 records.insert(entry.call_id.clone(), (entry.started_at, line.to_string()));
72 }
73 Err(_) => {
74 unparseable_lines.push(line.to_string());
75 }
76 }
77 }
78
79 for line in String::from_utf8_lossy(theirs_content).lines() {
81 if line.trim().is_empty() {
82 continue;
83 }
84 match serde_json::from_str::<LogEntryForMerge>(line) {
85 Ok(entry) => {
86 records.insert(entry.call_id.clone(), (entry.started_at, line.to_string()));
88 }
89 Err(_) => {
90 if !unparseable_lines.contains(&line.to_string()) {
92 unparseable_lines.push(line.to_string());
93 }
94 }
95 }
96 }
97
98 let mut sorted: Vec<_> = records.into_values().collect();
100 sorted.sort_by_key(|(ts, _)| *ts);
101
102 let mut output = sorted
104 .into_iter()
105 .map(|(_, line)| line)
106 .collect::<Vec<_>>()
107 .join("\n");
108
109 if !unparseable_lines.is_empty() {
110 if !output.is_empty() {
111 output.push('\n');
112 }
113 output.push_str(&unparseable_lines.join("\n"));
114 }
115
116 if !output.is_empty() {
117 output.push('\n');
118 }
119
120 output.into_bytes()
121}
122
123pub(crate) struct DivergenceAnalysis {
125 pub(crate) is_diverged: bool,
127 pub(crate) is_ahead: bool,
129 pub(crate) is_behind: bool,
131}
132
133const MAX_PUSH_RETRIES: u32 = 3;
134const RETRY_BASE_MS: u64 = 500;
135
136#[derive(Debug, Clone, Copy, PartialEq, Eq)]
137enum SyncRelation {
138 NoUpstream,
139 UpToDate,
140 AheadOnly,
141 BehindOnly,
142 Diverged,
143}
144
145#[derive(Debug, Clone, Copy, PartialEq, Eq)]
146enum SyncAttemptOutcome {
147 NoHeadChange,
148 FastForwarded,
149 Committed,
150}
151
152#[derive(Debug, Clone, Copy, PartialEq, Eq)]
153enum PushRaceResetMode {
154 Mixed,
155 Hard,
156}
157
158#[derive(Debug, Clone, Copy, PartialEq, Eq)]
159enum CommitParentPlan {
160 None,
161 HeadOnly,
162 UpstreamOnly,
163 HeadAndUpstream,
164}
165
166pub struct GitSync {
167 repo: Repository,
168 repo_path: std::path::PathBuf,
169 subpath: Option<String>,
170}
171
172impl GitSync {
173 pub fn new(repo_path: &Path, subpath: Option<String>) -> Result<Self> {
174 let repo = Repository::open(repo_path)?;
175 Ok(Self {
176 repo,
177 repo_path: repo_path.to_path_buf(),
178 subpath,
179 })
180 }
181
182 #[expect(
183 clippy::future_not_send,
184 reason = "git2::Repository is Send but not Sync; this is a known limitation"
185 )]
186 pub async fn sync(&self, mount_name: &str) -> Result<()> {
187 println!(" {} {}", "Syncing".cyan(), mount_name);
188
189 ensure_repo_ready_for_sync(&self.repo_path)?;
190
191 if self.repo.find_remote("origin").is_err() {
194 println!(
195 " {} No remote 'origin' configured (local-only)",
196 "Info".dimmed()
197 );
198 self.sync_without_remote(mount_name)?;
199 return Ok(());
200 }
201
202 let branch_name = get_sync_branch(&self.repo_path)?;
205
206 for attempt in 0..MAX_PUSH_RETRIES {
207 let attempt_head = self.head_commit_oid()?;
208 let sync_outcome = self.sync_once(mount_name, &branch_name)?;
209
210 let push_result =
211 push_current_branch_with_result(&self.repo_path, "origin", &branch_name)?;
212 if push_result.success {
213 println!(" {} Pushed to remote", "✓".green());
214 return Ok(());
215 }
216
217 let failure_kind = push_result.failure_kind.unwrap_or(PushFailureKind::Other);
218 if failure_kind == PushFailureKind::Race && attempt < MAX_PUSH_RETRIES - 1 {
219 println!(
220 " {} Push race detected; retrying after re-fetch",
221 "Info".dimmed()
222 );
223 let reset_mode = match sync_outcome {
224 SyncAttemptOutcome::FastForwarded => PushRaceResetMode::Hard,
225 SyncAttemptOutcome::NoHeadChange | SyncAttemptOutcome::Committed => {
226 PushRaceResetMode::Mixed
227 }
228 };
229 self.reset_after_push_race(attempt_head, reset_mode)?;
230 sleep(Duration::from_millis(RETRY_BASE_MS * 2u64.pow(attempt))).await;
231 continue;
232 }
233
234 let stderr = push_result.stderr.trim();
235 if stderr.is_empty() {
236 bail!("git push failed ({failure_kind:?})");
237 }
238 bail!("git push failed ({failure_kind:?}): {stderr}");
239 }
240
241 bail!("git push race retry budget exhausted after {MAX_PUSH_RETRIES} attempts")
242 }
243
244 fn sync_without_remote(&self, mount_name: &str) -> Result<()> {
245 let changes_staged = self.stage_changes()?;
246 if !changes_staged {
247 println!(" {} No changes to commit", "○".dimmed());
248 return Ok(());
249 }
250
251 let head_commit = self.head_commit()?;
252 let local_tree = self.local_tree_from_index()?;
253 let commit_oid = self.create_commit_from_relation(
254 mount_name,
255 &local_tree,
256 head_commit.as_ref(),
257 None,
258 SyncRelation::NoUpstream,
259 )?;
260 self.refresh_worktree_after_commit(commit_oid)?;
261 println!(" {} Committed changes", "✓".green());
262 Ok(())
263 }
264
265 fn sync_once(&self, mount_name: &str, branch_name: &str) -> Result<SyncAttemptOutcome> {
266 shell_fetch::fetch(&self.repo_path, "origin").with_context(|| {
267 format!(
268 "Fetch from origin failed for repo '{}'",
269 self.repo_path.display()
270 )
271 })?;
272
273 let head_commit = self.head_commit()?;
274 let upstream_commit = self.find_upstream_commit(branch_name)?;
275 let relation =
276 self.sync_relation(head_commit.as_ref(), upstream_commit.as_ref(), branch_name)?;
277
278 if let Some(upstream_commit) = upstream_commit.as_ref()
279 && self.should_premerge_before_staging(relation)?
280 {
281 self.premerge_jsonl_files(&upstream_commit.tree()?)?;
282 }
283
284 let changes_staged = self.stage_changes()?;
285 let local_tree = self.local_tree_from_index()?;
286
287 match relation {
288 SyncRelation::NoUpstream => {
289 if changes_staged {
290 let commit_oid = self.create_commit_from_relation(
291 mount_name,
292 &local_tree,
293 head_commit.as_ref(),
294 None,
295 relation,
296 )?;
297 self.refresh_worktree_after_commit(commit_oid)?;
298 println!(" {} Committed changes", "✓".green());
299 return Ok(SyncAttemptOutcome::Committed);
300 }
301 return Ok(SyncAttemptOutcome::NoHeadChange);
302 }
303 SyncRelation::UpToDate | SyncRelation::AheadOnly => {
304 if changes_staged {
305 let commit_oid = self.create_commit_from_relation(
306 mount_name,
307 &local_tree,
308 head_commit.as_ref(),
309 upstream_commit.as_ref(),
310 relation,
311 )?;
312 self.refresh_worktree_after_commit(commit_oid)?;
313 println!(" {} Committed changes", "✓".green());
314 return Ok(SyncAttemptOutcome::Committed);
315 }
316 println!(" {} No changes to commit", "○".dimmed());
317 return Ok(SyncAttemptOutcome::NoHeadChange);
318 }
319 SyncRelation::BehindOnly => {
320 let upstream_commit = upstream_commit.as_ref().ok_or_else(|| {
321 anyhow::anyhow!("Missing upstream commit for behind-only sync")
322 })?;
323 if !changes_staged {
324 self.fast_forward_to_commit(branch_name, upstream_commit)?;
325 println!(" {} Pulled remote changes", "✓".green());
326 return Ok(SyncAttemptOutcome::FastForwarded);
327 }
328 }
329 SyncRelation::Diverged => {
330 println!(
331 " {} Detected divergence from remote - merging before commit",
332 "Info".dimmed()
333 );
334 }
335 }
336
337 let upstream_commit = upstream_commit
338 .as_ref()
339 .ok_or_else(|| anyhow::anyhow!("Missing upstream commit for merge integration"))?;
340 let merged_tree = self.integrate_local_tree(
341 head_commit.as_ref(),
342 &local_tree,
343 upstream_commit,
344 relation,
345 )?;
346 let commit_oid = self.create_commit_from_relation(
347 mount_name,
348 &merged_tree,
349 head_commit.as_ref(),
350 Some(upstream_commit),
351 relation,
352 )?;
353 self.refresh_worktree_after_commit(commit_oid)?;
354 println!(" {} Integrated remote changes", "✓".green());
355
356 Ok(SyncAttemptOutcome::Committed)
357 }
358
359 fn should_premerge_before_staging(&self, relation: SyncRelation) -> Result<bool> {
360 Ok(match relation {
361 SyncRelation::Diverged => true,
362 SyncRelation::BehindOnly => is_worktree_dirty(&self.repo)?,
363 SyncRelation::NoUpstream | SyncRelation::UpToDate | SyncRelation::AheadOnly => false,
364 })
365 }
366
367 pub(crate) fn check_divergence(&self, branch_name: &str) -> Result<DivergenceAnalysis> {
369 let head = self.repo.head()?;
370 let upstream_ref = format!("refs/remotes/origin/{branch_name}");
371
372 let local_oid = head
373 .target()
374 .ok_or_else(|| anyhow::anyhow!("No HEAD target"))?;
375
376 let Ok(upstream_oid) = self.repo.refname_to_id(&upstream_ref) else {
377 return Ok(DivergenceAnalysis {
379 is_diverged: false,
380 is_ahead: true,
381 is_behind: false,
382 });
383 };
384
385 let (ahead, behind) = self.repo.graph_ahead_behind(local_oid, upstream_oid)?;
388
389 Ok(DivergenceAnalysis {
390 is_diverged: ahead > 0 && behind > 0,
391 is_ahead: ahead > 0,
392 is_behind: behind > 0,
393 })
394 }
395
396 fn sync_relation(
397 &self,
398 head_commit: Option<&Commit<'_>>,
399 upstream_commit: Option<&Commit<'_>>,
400 branch_name: &str,
401 ) -> Result<SyncRelation> {
402 match (head_commit, upstream_commit) {
403 (_, None) => Ok(SyncRelation::NoUpstream),
404 (None, Some(_)) => Ok(SyncRelation::BehindOnly),
405 (Some(_), Some(_)) => {
406 let analysis = self.check_divergence(branch_name)?;
407 Ok(
408 match (analysis.is_diverged, analysis.is_ahead, analysis.is_behind) {
409 (false, true, false) => SyncRelation::AheadOnly,
410 (false, false, true) => SyncRelation::BehindOnly,
411 (false, false, false) => SyncRelation::UpToDate,
412 _ => SyncRelation::Diverged,
414 },
415 )
416 }
417 }
418 }
419
420 fn head_commit(&self) -> Result<Option<Commit<'_>>> {
421 match self.repo.head() {
422 Ok(head) => {
423 let target = head
424 .target()
425 .ok_or_else(|| anyhow::anyhow!("No HEAD target"))?;
426 Ok(Some(self.repo.find_commit(target)?))
427 }
428 Err(e) if e.code() == ErrorCode::UnbornBranch => Ok(None),
429 Err(e) => Err(e.into()),
430 }
431 }
432
433 fn head_commit_oid(&self) -> Result<Option<Oid>> {
434 Ok(self.head_commit()?.map(|commit| commit.id()))
435 }
436
437 fn find_upstream_commit(&self, branch_name: &str) -> Result<Option<Commit<'_>>> {
438 match self
439 .repo
440 .refname_to_id(&format!("refs/remotes/origin/{branch_name}"))
441 {
442 Ok(oid) => Ok(Some(self.repo.find_commit(oid)?)),
443 Err(_) => Ok(None),
444 }
445 }
446
447 fn local_tree_from_index(&self) -> Result<Tree<'_>> {
448 let mut index = self.repo.index()?;
449 let tree_id = index.write_tree()?;
450 self.repo.find_tree(tree_id).map_err(Into::into)
451 }
452
453 fn integrate_local_tree(
454 &self,
455 head_commit: Option<&Commit<'_>>,
456 local_tree: &Tree<'_>,
457 upstream_commit: &Commit<'_>,
458 relation: SyncRelation,
459 ) -> Result<Tree<'_>> {
460 let ancestor_tree_id =
461 self.ancestor_tree_for_merge(head_commit, upstream_commit, relation)?;
462 let ancestor_tree = self.repo.find_tree(ancestor_tree_id)?;
463 let upstream_tree = upstream_commit.tree()?;
464 let mut merged_index =
465 self.repo
466 .merge_trees(&ancestor_tree, local_tree, &upstream_tree, None)?;
467
468 if merged_index.has_conflicts() {
469 self.resolve_merge_conflicts(&mut merged_index)?;
470 }
471 if merged_index.has_conflicts() {
472 bail!("Failed to resolve merge conflicts before final commit");
473 }
474
475 let tree_id = merged_index.write_tree_to(&self.repo)?;
476 self.repo.find_tree(tree_id).map_err(Into::into)
477 }
478
479 fn ancestor_tree_for_merge(
480 &self,
481 head_commit: Option<&Commit<'_>>,
482 upstream_commit: &Commit<'_>,
483 relation: SyncRelation,
484 ) -> Result<Oid> {
485 match relation {
486 SyncRelation::BehindOnly => match head_commit {
487 Some(head_commit) => Ok(head_commit.tree_id()),
488 None => self.empty_tree().map(|tree| tree.id()),
489 },
490 SyncRelation::Diverged => {
491 let head_commit = head_commit
492 .ok_or_else(|| anyhow::anyhow!("Missing HEAD commit for diverged merge"))?;
493 match self.repo.merge_base(head_commit.id(), upstream_commit.id()) {
494 Ok(merge_base_oid) => Ok(self.repo.find_commit(merge_base_oid)?.tree_id()),
495 Err(_) => self.empty_tree().map(|tree| tree.id()),
496 }
497 }
498 _ => self.empty_tree().map(|tree| tree.id()),
499 }
500 }
501
502 fn empty_tree(&self) -> Result<Tree<'_>> {
503 let mut index = Index::new()?;
504 let tree_id = index.write_tree_to(&self.repo)?;
505 self.repo.find_tree(tree_id).map_err(Into::into)
506 }
507
508 fn resolve_merge_conflicts(&self, index: &mut Index) -> Result<()> {
509 const GIT_INDEX_ENTRY_STAGEMASK: u16 = 0x3000;
511
512 let conflicts: Vec<_> = index
513 .conflicts()?
514 .collect::<std::result::Result<Vec<_>, _>>()?;
515
516 for conflict in conflicts {
517 let path = conflict
518 .our
519 .as_ref()
520 .or(conflict.their.as_ref())
521 .or(conflict.ancestor.as_ref())
522 .map(|entry| String::from_utf8_lossy(&entry.path).to_string())
523 .unwrap_or_default();
524
525 if is_tool_log_file(&path)
526 && let (Some(local), Some(remote)) = (&conflict.our, &conflict.their)
527 {
528 let local_blob = self.repo.find_blob(local.id)?;
529 let remote_blob = self.repo.find_blob(remote.id)?;
530 let merged = merge_jsonl_logs(remote_blob.content(), local_blob.content());
531
532 let file_path = self.repo_path.join(&path);
534 if let Some(parent) = file_path.parent() {
535 std::fs::create_dir_all(parent)?;
536 }
537 std::fs::write(&file_path, &merged)?;
538
539 let blob_oid = self.repo.blob(&merged)?;
543
544 index.conflict_remove(Path::new(&path))?;
548
549 let entry = git2::IndexEntry {
550 id: blob_oid,
551 file_size: u32::try_from(merged.len()).unwrap_or(u32::MAX),
552 ctime: local.ctime,
554 mtime: local.mtime,
555 dev: local.dev,
556 ino: local.ino,
557 mode: local.mode,
558 uid: local.uid,
559 gid: local.gid,
560 flags: local.flags & !GIT_INDEX_ENTRY_STAGEMASK,
561 flags_extended: local.flags_extended,
562 path: local.path.clone(),
563 };
564 index.add(&entry)?;
565 continue;
566 }
567
568 match (&conflict.our, &conflict.their) {
570 (_, Some(remote)) => {
571 index.conflict_remove(Path::new(&path))?;
573 let resolved = git2::IndexEntry {
574 ctime: remote.ctime,
575 mtime: remote.mtime,
576 dev: remote.dev,
577 ino: remote.ino,
578 mode: remote.mode,
579 uid: remote.uid,
580 gid: remote.gid,
581 file_size: remote.file_size,
582 id: remote.id,
583 flags: remote.flags & !GIT_INDEX_ENTRY_STAGEMASK,
584 flags_extended: remote.flags_extended,
585 path: remote.path.clone(),
586 };
587 index.add(&resolved)?;
588 }
589 (Some(_), None) => {
590 index.conflict_remove(Path::new(&path))?;
592 }
593 (None, None) => {}
594 }
595 }
596
597 Ok(())
600 }
601
602 fn create_commit_from_relation(
603 &self,
604 mount_name: &str,
605 tree: &Tree<'_>,
606 head_commit: Option<&Commit<'_>>,
607 upstream_commit: Option<&Commit<'_>>,
608 relation: SyncRelation,
609 ) -> Result<Oid> {
610 match commit_parent_plan(relation, head_commit.is_some(), upstream_commit.is_some())? {
611 CommitParentPlan::None => self.create_commit_for_tree(mount_name, tree, &[]),
612 CommitParentPlan::HeadOnly => {
613 let parents = head_commit.map(|commit| vec![commit]).unwrap_or_default();
614 self.create_commit_for_tree(mount_name, tree, &parents)
615 }
616 CommitParentPlan::UpstreamOnly => {
617 let upstream_commit = upstream_commit.ok_or_else(|| {
618 anyhow::anyhow!("Missing upstream commit for behind-only commit")
619 })?;
620 self.create_commit_for_tree(mount_name, tree, &[upstream_commit])
621 }
622 CommitParentPlan::HeadAndUpstream => {
623 let head_commit = head_commit
624 .ok_or_else(|| anyhow::anyhow!("Missing HEAD commit for diverged commit"))?;
625 let upstream_commit = upstream_commit.ok_or_else(|| {
626 anyhow::anyhow!("Missing upstream commit for diverged commit")
627 })?;
628 self.create_commit_for_tree(mount_name, tree, &[head_commit, upstream_commit])
629 }
630 }
631 }
632
633 fn create_commit_for_tree(
634 &self,
635 mount_name: &str,
636 tree: &Tree<'_>,
637 parents: &[&Commit<'_>],
638 ) -> Result<Oid> {
639 let sig = Signature::now("thoughts-sync", "thoughts@sync.local")?;
640 let message = if let Some(subpath) = &self.subpath {
641 format!("Auto-sync thoughts for {mount_name} (subpath: {subpath})")
642 } else {
643 format!("Auto-sync thoughts for {mount_name}")
644 };
645
646 let commit_oid = self
650 .repo
651 .commit(None, &sig, &sig, &message, tree, parents)?;
652
653 let (refname, is_branch) = match self.repo.head() {
657 Ok(head_ref) => {
658 let name = head_ref.name()?;
659 (name.to_string(), head_ref.is_branch())
660 }
661 Err(e) if e.code() == git2::ErrorCode::UnbornBranch => {
662 let head_ref = self.repo.find_reference("HEAD")?;
665 let symbolic_target = head_ref
666 .symbolic_target()?
667 .ok_or_else(|| anyhow::anyhow!("HEAD has no symbolic target"))?;
668 (symbolic_target.to_string(), true)
669 }
670 Err(e) => return Err(e.into()),
671 };
672
673 if is_branch {
676 self.repo.reference(
677 &refname,
678 commit_oid,
679 true, &format!("thoughts-sync: {message}"),
681 )?;
682 } else {
683 self.repo.set_head_detached(commit_oid)?;
684 }
685
686 Ok(commit_oid)
687 }
688
689 fn refresh_worktree_after_commit(&self, commit_oid: Oid) -> Result<()> {
690 if self.subpath.is_some() {
691 let commit = self.repo.find_commit(commit_oid)?;
692 self.refresh_subpath_after_commit(&commit)?;
693 return Ok(());
694 }
695
696 let obj = self.repo.find_object(commit_oid, None)?;
697 self.repo.reset(
698 &obj,
699 git2::ResetType::Hard,
700 Some(git2::build::CheckoutBuilder::default().force()),
701 )?;
702 Ok(())
703 }
704
705 fn refresh_subpath_after_commit(&self, commit: &Commit<'_>) -> Result<()> {
706 let subpath = self
707 .subpath
708 .as_deref()
709 .ok_or_else(|| anyhow::anyhow!("Missing subpath for subpath refresh"))?;
710 let tree = commit.tree()?;
711 let mut checkout = git2::build::CheckoutBuilder::default();
712 checkout.force().path(subpath);
713 self.repo
714 .checkout_tree(tree.as_object(), Some(&mut checkout))?;
715 self.refresh_index_in_scope()
716 }
717
718 fn fast_forward_to_commit(
719 &self,
720 branch_name: &str,
721 upstream_commit: &Commit<'_>,
722 ) -> Result<()> {
723 if is_worktree_dirty(&self.repo)? {
724 bail!(
725 "Cannot fast-forward: working tree has uncommitted changes. Please commit or stash before syncing."
726 );
727 }
728
729 self.repo.set_head(&format!("refs/heads/{branch_name}"))?;
730 let obj = self.repo.find_object(upstream_commit.id(), None)?;
731 self.repo.reset(
732 &obj,
733 git2::ResetType::Hard,
734 Some(git2::build::CheckoutBuilder::default().force()),
735 )?;
736 Ok(())
737 }
738
739 fn reset_after_push_race(
740 &self,
741 original_head: Option<Oid>,
742 reset_mode: PushRaceResetMode,
743 ) -> Result<()> {
744 if let Some(original_head) = original_head {
745 let obj = self.repo.find_object(original_head, None)?;
746 match reset_mode {
747 PushRaceResetMode::Mixed => {
748 self.repo.reset(&obj, git2::ResetType::Mixed, None)?;
749 }
750 PushRaceResetMode::Hard => {
751 self.repo.reset(
752 &obj,
753 git2::ResetType::Hard,
754 Some(git2::build::CheckoutBuilder::default().force()),
755 )?;
756 }
757 }
758 } else {
759 let branch_name = get_sync_branch(&self.repo_path)?;
760 self.repo.set_head(&format!("refs/heads/{branch_name}"))?;
761 self.repo.cleanup_state()?;
762 }
763 Ok(())
764 }
765
766 fn premerge_jsonl_files(&self, upstream_tree: &Tree<'_>) -> Result<()> {
767 for rel_path in self.tool_log_files_in_scope()? {
768 let Some(upstream_content) = self.read_tree_blob(upstream_tree, &rel_path)? else {
769 continue;
770 };
771
772 let local_path = self.repo_path.join(&rel_path);
773 let local_content = std::fs::read(&local_path)?;
774 let merged = merge_jsonl_logs(&upstream_content, &local_content);
775 if merged != local_content {
776 std::fs::write(local_path, merged)?;
777 }
778 }
779 Ok(())
780 }
781
782 fn tool_log_files_in_scope(&self) -> Result<Vec<String>> {
783 let root = self.subpath.as_ref().map_or_else(
784 || self.repo_path.clone(),
785 |subpath| self.repo_path.join(subpath),
786 );
787 let mut files = Vec::new();
788 self.collect_tool_log_files(&root, &mut files)?;
789 files.sort();
790 Ok(files)
791 }
792
793 fn collect_tool_log_files(&self, dir: &Path, files: &mut Vec<String>) -> Result<()> {
794 if !dir.exists() {
795 return Ok(());
796 }
797
798 for entry in std::fs::read_dir(dir)? {
799 let entry = entry?;
800 let path = entry.path();
801 if path.file_name().is_some_and(|name| name == ".git") {
802 continue;
803 }
804
805 if path.is_dir() {
806 self.collect_tool_log_files(&path, files)?;
807 continue;
808 }
809
810 let rel_path = path
811 .strip_prefix(&self.repo_path)
812 .with_context(|| format!("Failed to strip repo prefix from {}", path.display()))?;
813 let rel_path = rel_path.to_string_lossy().replace('\\', "/");
814 if is_tool_log_file(&rel_path) {
815 files.push(rel_path);
816 }
817 }
818
819 Ok(())
820 }
821
822 fn read_tree_blob(&self, tree: &Tree<'_>, rel_path: &str) -> Result<Option<Vec<u8>>> {
823 let entry = match tree.get_path(Path::new(rel_path)) {
824 Ok(entry) => entry,
825 Err(err) if err.code() == ErrorCode::NotFound => return Ok(None),
826 Err(err) => return Err(err.into()),
827 };
828
829 let blob = self.repo.find_blob(entry.id())?;
830 Ok(Some(blob.content().to_vec()))
831 }
832
833 fn stage_changes(&self) -> Result<bool> {
834 self.refresh_index_in_scope()?;
835
836 let index = self.repo.index()?;
837
838 let diff = match self.repo.head() {
841 Ok(head) => {
842 let head_oid = head
843 .target()
844 .ok_or_else(|| anyhow::anyhow!("HEAD reference has no target"))?;
845 let head_tree = self.repo.find_commit(head_oid)?.tree()?;
846 self.repo
847 .diff_tree_to_index(Some(&head_tree), Some(&index), None)?
848 }
849 Err(e) if e.code() == git2::ErrorCode::UnbornBranch => {
850 self.repo.diff_tree_to_index(None, Some(&index), None)?
852 }
853 Err(e) => return Err(e.into()),
854 };
855
856 Ok(diff.stats()?.files_changed() > 0)
857 }
858
859 fn refresh_index_in_scope(&self) -> Result<()> {
860 let mut index = self.repo.index()?;
861 let pathspecs = self.scoped_pathspecs();
862
863 index.add_all(pathspecs.iter(), IndexAddOption::DEFAULT, None)?;
864
865 index.update_all(pathspecs.iter(), None)?;
867
868 index.write()?;
869
870 Ok(())
871 }
872
873 fn scoped_pathspecs(&self) -> Vec<String> {
874 if let Some(subpath) = &self.subpath {
875 vec![format!("{}/*", subpath), format!("{}/**/*", subpath)]
876 } else {
877 vec![".".to_string()]
878 }
879 }
880}
881
882fn commit_parent_plan(
883 relation: SyncRelation,
884 has_head: bool,
885 has_upstream: bool,
886) -> Result<CommitParentPlan> {
887 Ok(match relation {
888 SyncRelation::NoUpstream | SyncRelation::UpToDate | SyncRelation::AheadOnly => {
889 if has_head {
890 CommitParentPlan::HeadOnly
891 } else {
892 CommitParentPlan::None
893 }
894 }
895 SyncRelation::BehindOnly => {
896 if !has_upstream {
897 bail!("Missing upstream commit for behind-only commit");
898 }
899 CommitParentPlan::UpstreamOnly
900 }
901 SyncRelation::Diverged => {
902 if !has_head || !has_upstream {
903 bail!("Missing head or upstream commit for diverged commit");
904 }
905 CommitParentPlan::HeadAndUpstream
906 }
907 })
908}
909
910#[cfg(test)]
911mod tests {
912 use super::*;
913
914 #[test]
915 fn test_merge_jsonl_deduplicates_by_call_id() {
916 let ours = br#"{"call_id":"abc","started_at":"2025-01-01T10:00:00Z","tool":"foo"}
917{"call_id":"def","started_at":"2025-01-01T11:00:00Z","tool":"bar"}"#;
918 let theirs = br#"{"call_id":"abc","started_at":"2025-01-01T10:00:00Z","tool":"foo_updated"}
919{"call_id":"ghi","started_at":"2025-01-01T12:00:00Z","tool":"baz"}"#;
920
921 let merged = merge_jsonl_logs(ours, theirs);
922 let merged_str = String::from_utf8_lossy(&merged);
923
924 assert!(merged_str.contains("foo_updated"));
926 assert!(!merged_str.contains(r#""tool":"foo""#)); assert!(merged_str.contains("def"));
928 assert!(merged_str.contains("ghi"));
929 }
930
931 #[test]
932 fn test_merge_jsonl_preserves_unparseable() {
933 let ours = b"not valid json\n";
934 let theirs = br#"{"call_id":"abc","started_at":"2025-01-01T10:00:00Z","tool":"foo"}"#;
935
936 let merged = merge_jsonl_logs(ours, theirs);
937 let merged_str = String::from_utf8_lossy(&merged);
938
939 assert!(merged_str.contains("not valid json"));
940 assert!(merged_str.contains("call_id"));
941 }
942
943 #[test]
944 fn test_merge_jsonl_sorts_by_timestamp() {
945 let ours = br#"{"call_id":"late","started_at":"2025-01-01T15:00:00Z","tool":"c"}"#;
946 let theirs = br#"{"call_id":"early","started_at":"2025-01-01T09:00:00Z","tool":"a"}
947{"call_id":"mid","started_at":"2025-01-01T12:00:00Z","tool":"b"}"#;
948
949 let merged = merge_jsonl_logs(ours, theirs);
950 let merged_str = String::from_utf8_lossy(&merged);
951 let lines: Vec<_> = merged_str.lines().collect();
952
953 assert!(lines[0].contains("early"));
954 assert!(lines[1].contains("mid"));
955 assert!(lines[2].contains("late"));
956 }
957
958 #[test]
959 fn test_merge_jsonl_empty_files() {
960 let merged = merge_jsonl_logs(b"", b"");
961 assert!(merged.is_empty());
962 }
963
964 #[test]
965 fn test_merge_jsonl_one_side_empty() {
966 let content = br#"{"call_id":"abc","started_at":"2025-01-01T10:00:00Z","tool":"foo"}"#;
967
968 let merged_ours_empty = merge_jsonl_logs(b"", content);
969 assert!(String::from_utf8_lossy(&merged_ours_empty).contains("abc"));
970
971 let merged_theirs_empty = merge_jsonl_logs(content, b"");
972 assert!(String::from_utf8_lossy(&merged_theirs_empty).contains("abc"));
973 }
974
975 #[test]
976 fn test_merge_context_jsonl_keeps_local_on_collision() {
977 let remote = br#"{"call_id":"same","started_at":"2025-01-01T10:00:00Z","tool":"remote"}"#;
978 let local = br#"{"call_id":"same","started_at":"2025-01-01T10:00:00Z","tool":"local"}"#;
979
980 let merged = merge_jsonl_logs(remote, local);
981 let merged_str = String::from_utf8_lossy(&merged);
982
983 assert!(merged_str.contains("local"));
984 assert!(!merged_str.contains("remote"));
985 }
986
987 #[test]
988 fn test_is_tool_log_file() {
989 assert!(is_tool_log_file("branch/logs/tool_logs_2025-01-01.jsonl"));
991 assert!(is_tool_log_file(
992 "foo/logs/tool_logs_2025-01-01_abc123.jsonl"
993 ));
994 assert!(is_tool_log_file("a/b/c/logs/tool_logs_whatever.jsonl"));
995
996 assert!(!is_tool_log_file("branch/logs/other.jsonl"));
998
999 assert!(!is_tool_log_file(
1001 "branch/research/tool_logs_2025-01-01.jsonl"
1002 ));
1003
1004 assert!(!is_tool_log_file("branch/logs/tool_logs_2025-01-01.json"));
1006
1007 assert!(!is_tool_log_file("tool_logs_config/logs/readme.jsonl"));
1009 assert!(!is_tool_log_file("tool_logs_foo/logs/bar.jsonl"));
1010
1011 assert!(!is_tool_log_file("tool_logs_2025-01-01.jsonl"));
1013 }
1014
1015 #[test]
1016 fn commit_parent_plan_selects_expected_parents() {
1017 assert_eq!(
1018 commit_parent_plan(SyncRelation::NoUpstream, false, false).unwrap(),
1019 CommitParentPlan::None
1020 );
1021 assert_eq!(
1022 commit_parent_plan(SyncRelation::UpToDate, true, true).unwrap(),
1023 CommitParentPlan::HeadOnly
1024 );
1025 assert_eq!(
1026 commit_parent_plan(SyncRelation::AheadOnly, true, false).unwrap(),
1027 CommitParentPlan::HeadOnly
1028 );
1029 assert_eq!(
1030 commit_parent_plan(SyncRelation::BehindOnly, true, true).unwrap(),
1031 CommitParentPlan::UpstreamOnly
1032 );
1033 assert_eq!(
1034 commit_parent_plan(SyncRelation::Diverged, true, true).unwrap(),
1035 CommitParentPlan::HeadAndUpstream
1036 );
1037 }
1038
1039 fn git_ok(dir: &std::path::Path, args: &[&str]) {
1046 let out = std::process::Command::new("git")
1047 .current_dir(dir)
1048 .args(args)
1049 .output()
1050 .expect("failed to spawn git");
1051 assert!(
1052 out.status.success(),
1053 "git {:?} failed: {}",
1054 args,
1055 String::from_utf8_lossy(&out.stderr)
1056 );
1057 }
1058
1059 fn git_stdout(dir: &std::path::Path, args: &[&str]) -> String {
1061 let out = std::process::Command::new("git")
1062 .current_dir(dir)
1063 .args(args)
1064 .output()
1065 .expect("failed to spawn git");
1066 assert!(out.status.success());
1067 String::from_utf8_lossy(&out.stdout).trim().to_string()
1068 }
1069
1070 #[test]
1073 fn divergence_no_upstream_ref() {
1074 let repo = tempfile::TempDir::new().unwrap();
1075 git_ok(repo.path(), &["init"]);
1076 std::fs::write(repo.path().join("a.txt"), "a").unwrap();
1077 git_ok(repo.path(), &["add", "."]);
1078 git_ok(
1079 repo.path(),
1080 &[
1081 "-c",
1082 "user.name=Test",
1083 "-c",
1084 "user.email=test@example.com",
1085 "commit",
1086 "-m",
1087 "initial",
1088 ],
1089 );
1090
1091 let sync = GitSync::new(repo.path(), None).unwrap();
1092 let analysis = sync.check_divergence("main").unwrap();
1093
1094 assert!(!analysis.is_diverged, "should not be diverged");
1095 assert!(analysis.is_ahead, "should be ahead (no upstream)");
1096 assert!(!analysis.is_behind, "should not be behind");
1097 }
1098
1099 #[test]
1102 fn divergence_up_to_date() {
1103 let repo = tempfile::TempDir::new().unwrap();
1104 git_ok(repo.path(), &["init"]);
1105 std::fs::write(repo.path().join("a.txt"), "a").unwrap();
1106 git_ok(repo.path(), &["add", "."]);
1107 git_ok(
1108 repo.path(),
1109 &[
1110 "-c",
1111 "user.name=Test",
1112 "-c",
1113 "user.email=test@example.com",
1114 "commit",
1115 "-m",
1116 "initial",
1117 ],
1118 );
1119 git_ok(repo.path(), &["branch", "-M", "main"]);
1121
1122 let head_oid = git_stdout(repo.path(), &["rev-parse", "HEAD"]);
1123 git_ok(
1124 repo.path(),
1125 &["update-ref", "refs/remotes/origin/main", &head_oid],
1126 );
1127
1128 let sync = GitSync::new(repo.path(), None).unwrap();
1129 let analysis = sync.check_divergence("main").unwrap();
1130
1131 assert!(!analysis.is_diverged, "should not be diverged");
1132 assert!(!analysis.is_ahead, "should not be ahead");
1133 assert!(!analysis.is_behind, "should not be behind");
1134 }
1135
1136 #[test]
1139 fn divergence_local_ahead_only() {
1140 let repo = tempfile::TempDir::new().unwrap();
1141 git_ok(repo.path(), &["init"]);
1142 std::fs::write(repo.path().join("a.txt"), "a").unwrap();
1143 git_ok(repo.path(), &["add", "."]);
1144 git_ok(
1145 repo.path(),
1146 &[
1147 "-c",
1148 "user.name=Test",
1149 "-c",
1150 "user.email=test@example.com",
1151 "commit",
1152 "-m",
1153 "C1",
1154 ],
1155 );
1156 git_ok(repo.path(), &["branch", "-M", "main"]);
1158
1159 let c1_oid = git_stdout(repo.path(), &["rev-parse", "HEAD"]);
1160 git_ok(
1161 repo.path(),
1162 &["update-ref", "refs/remotes/origin/main", &c1_oid],
1163 );
1164
1165 std::fs::write(repo.path().join("b.txt"), "b").unwrap();
1166 git_ok(repo.path(), &["add", "."]);
1167 git_ok(
1168 repo.path(),
1169 &[
1170 "-c",
1171 "user.name=Test",
1172 "-c",
1173 "user.email=test@example.com",
1174 "commit",
1175 "-m",
1176 "C2",
1177 ],
1178 );
1179
1180 let sync = GitSync::new(repo.path(), None).unwrap();
1181 let analysis = sync.check_divergence("main").unwrap();
1182
1183 assert!(!analysis.is_diverged, "should not be diverged");
1184 assert!(analysis.is_ahead, "should be ahead");
1185 assert!(!analysis.is_behind, "should not be behind");
1186 }
1187
1188 #[test]
1191 fn divergence_local_behind_only() {
1192 let repo = tempfile::TempDir::new().unwrap();
1193 git_ok(repo.path(), &["init"]);
1194 std::fs::write(repo.path().join("a.txt"), "a").unwrap();
1195 git_ok(repo.path(), &["add", "."]);
1196 git_ok(
1197 repo.path(),
1198 &[
1199 "-c",
1200 "user.name=Test",
1201 "-c",
1202 "user.email=test@example.com",
1203 "commit",
1204 "-m",
1205 "C1",
1206 ],
1207 );
1208 git_ok(repo.path(), &["branch", "-M", "main"]);
1210
1211 std::fs::write(repo.path().join("b.txt"), "b").unwrap();
1212 git_ok(repo.path(), &["add", "."]);
1213 git_ok(
1214 repo.path(),
1215 &[
1216 "-c",
1217 "user.name=Test",
1218 "-c",
1219 "user.email=test@example.com",
1220 "commit",
1221 "-m",
1222 "C2",
1223 ],
1224 );
1225
1226 let c2_oid = git_stdout(repo.path(), &["rev-parse", "HEAD"]);
1227 git_ok(repo.path(), &["reset", "--hard", "HEAD~1"]);
1228 git_ok(
1229 repo.path(),
1230 &["update-ref", "refs/remotes/origin/main", &c2_oid],
1231 );
1232
1233 let sync = GitSync::new(repo.path(), None).unwrap();
1234 let analysis = sync.check_divergence("main").unwrap();
1235
1236 assert!(!analysis.is_diverged, "should not be diverged");
1237 assert!(!analysis.is_ahead, "should not be ahead");
1238 assert!(analysis.is_behind, "should be behind");
1239 }
1240
1241 #[test]
1244 fn divergence_diverged() {
1245 let repo = tempfile::TempDir::new().unwrap();
1246 git_ok(repo.path(), &["init"]);
1247 std::fs::write(repo.path().join("a.txt"), "a").unwrap();
1248 git_ok(repo.path(), &["add", "."]);
1249 git_ok(
1250 repo.path(),
1251 &[
1252 "-c",
1253 "user.name=Test",
1254 "-c",
1255 "user.email=test@example.com",
1256 "commit",
1257 "-m",
1258 "C1",
1259 ],
1260 );
1261 git_ok(repo.path(), &["branch", "-M", "main"]);
1263
1264 let c1_oid = git_stdout(repo.path(), &["rev-parse", "HEAD"]);
1265
1266 std::fs::write(repo.path().join("b.txt"), "b").unwrap();
1267 git_ok(repo.path(), &["add", "."]);
1268 git_ok(
1269 repo.path(),
1270 &[
1271 "-c",
1272 "user.name=Test",
1273 "-c",
1274 "user.email=test@example.com",
1275 "commit",
1276 "-m",
1277 "C2-local",
1278 ],
1279 );
1280
1281 git_ok(repo.path(), &["branch", "remote-sim", &c1_oid]);
1282 git_ok(repo.path(), &["checkout", "remote-sim"]);
1283 std::fs::write(repo.path().join("c.txt"), "c").unwrap();
1284 git_ok(repo.path(), &["add", "."]);
1285 git_ok(
1286 repo.path(),
1287 &[
1288 "-c",
1289 "user.name=Test",
1290 "-c",
1291 "user.email=test@example.com",
1292 "commit",
1293 "-m",
1294 "C3-remote",
1295 ],
1296 );
1297
1298 let c3_oid = git_stdout(repo.path(), &["rev-parse", "HEAD"]);
1299 git_ok(repo.path(), &["checkout", "main"]);
1300 git_ok(
1301 repo.path(),
1302 &["update-ref", "refs/remotes/origin/main", &c3_oid],
1303 );
1304
1305 let sync = GitSync::new(repo.path(), None).unwrap();
1306 let analysis = sync.check_divergence("main").unwrap();
1307
1308 assert!(analysis.is_diverged, "should be diverged");
1309 assert!(analysis.is_ahead, "should be ahead");
1310 assert!(analysis.is_behind, "should be behind");
1311 }
1312
1313 #[test]
1314 fn refresh_worktree_after_commit_refreshes_only_subpath() {
1315 let repo = tempfile::TempDir::new().unwrap();
1316 git_ok(repo.path(), &["init"]);
1317 std::fs::create_dir_all(repo.path().join("branch")).unwrap();
1318 std::fs::write(repo.path().join("branch/data.txt"), "committed\n").unwrap();
1319 std::fs::write(repo.path().join("outside.txt"), "outside\n").unwrap();
1320 git_ok(repo.path(), &["add", "."]);
1321 git_ok(
1322 repo.path(),
1323 &[
1324 "-c",
1325 "user.name=Test",
1326 "-c",
1327 "user.email=test@example.com",
1328 "commit",
1329 "-m",
1330 "initial",
1331 ],
1332 );
1333 git_ok(repo.path(), &["branch", "-M", "main"]);
1334
1335 std::fs::write(repo.path().join("branch/data.txt"), "stale branch\n").unwrap();
1336 std::fs::write(repo.path().join("outside.txt"), "stale outside\n").unwrap();
1337 git_ok(repo.path(), &["add", "branch/data.txt", "outside.txt"]);
1338
1339 let sync = GitSync::new(repo.path(), Some("branch".to_string())).unwrap();
1340 let head_oid = Oid::from_str(&git_stdout(repo.path(), &["rev-parse", "HEAD"])).unwrap();
1341
1342 sync.refresh_worktree_after_commit(head_oid).unwrap();
1343
1344 assert_eq!(
1345 std::fs::read_to_string(repo.path().join("branch/data.txt")).unwrap(),
1346 "committed\n"
1347 );
1348 assert_eq!(
1349 std::fs::read_to_string(repo.path().join("outside.txt")).unwrap(),
1350 "stale outside\n"
1351 );
1352
1353 let status = git_stdout(repo.path(), &["status", "--short"]);
1354 assert!(!status.contains("branch/data.txt"), "status was: {status}");
1355 assert!(status.contains("outside.txt"), "status was: {status}");
1356 }
1357
1358 #[test]
1359 fn reset_after_push_race_hard_restores_fast_forwarded_worktree() {
1360 let repo = tempfile::TempDir::new().unwrap();
1361 git_ok(repo.path(), &["init"]);
1362 std::fs::write(repo.path().join("base.txt"), "one\n").unwrap();
1363 git_ok(repo.path(), &["add", "."]);
1364 git_ok(
1365 repo.path(),
1366 &[
1367 "-c",
1368 "user.name=Test",
1369 "-c",
1370 "user.email=test@example.com",
1371 "commit",
1372 "-m",
1373 "c1",
1374 ],
1375 );
1376 git_ok(repo.path(), &["branch", "-M", "main"]);
1377 let c1 = git_stdout(repo.path(), &["rev-parse", "HEAD"]);
1378
1379 std::fs::write(repo.path().join("base.txt"), "two\n").unwrap();
1380 git_ok(repo.path(), &["add", "."]);
1381 git_ok(
1382 repo.path(),
1383 &[
1384 "-c",
1385 "user.name=Test",
1386 "-c",
1387 "user.email=test@example.com",
1388 "commit",
1389 "-m",
1390 "c2",
1391 ],
1392 );
1393 let c2 = git_stdout(repo.path(), &["rev-parse", "HEAD"]);
1394 git_ok(repo.path(), &["reset", "--hard", &c1]);
1395
1396 let sync = GitSync::new(repo.path(), None).unwrap();
1397 let c2_commit = sync.repo.find_commit(Oid::from_str(&c2).unwrap()).unwrap();
1398
1399 sync.fast_forward_to_commit("main", &c2_commit).unwrap();
1400 assert_eq!(
1401 std::fs::read_to_string(repo.path().join("base.txt")).unwrap(),
1402 "two\n"
1403 );
1404
1405 sync.reset_after_push_race(Some(Oid::from_str(&c1).unwrap()), PushRaceResetMode::Hard)
1406 .unwrap();
1407
1408 assert_eq!(git_stdout(repo.path(), &["rev-parse", "HEAD"]), c1);
1409 assert_eq!(
1410 std::fs::read_to_string(repo.path().join("base.txt")).unwrap(),
1411 "one\n"
1412 );
1413 assert!(git_stdout(repo.path(), &["status", "--short"]).is_empty());
1414 }
1415}