diff --git a/src/daemon.rs b/src/daemon.rs index aef688932f..22b97e9aa4 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -2630,7 +2630,9 @@ fn read_checkpoint_body( #[derive(Debug)] enum FamilySequencerEntry { - PendingRoot, + PendingRoot { + root_sid: String, + }, ReadyCommand(Box), Checkpoint { request: Box, @@ -2715,6 +2717,14 @@ struct TraceIngressState { /// Roots whose start event was identified as definitely read-only. All /// subsequent events for these roots (including exit) take the fast path. root_definitely_read_only: HashSet, + /// Commit roots currently blocked inside their commit-message editor. + /// `git commit` cannot update refs until this child returns, so completed + /// commands may safely pass this root in the family sequencer meanwhile. + root_commit_editor_child_ids: HashMap, + /// Roots that have run a nested mutating Git command. Their reflog ranges + /// may already contain mutations, so they must remain ordering barriers + /// even if the parent later opens a commit-message editor. + root_saw_nested_mutation: HashSet, root_open_connections: HashMap, unidentified_open_connections: usize, root_close_markers_enqueued: HashSet, @@ -3281,9 +3291,12 @@ impl ActorDaemonCoordinator { ordinal: state.next_ordinal, }; state.next_ordinal = state.next_ordinal.saturating_add(1); - state - .entries - .insert(order, FamilySequencerEntry::PendingRoot); + state.entries.insert( + order, + FamilySequencerEntry::PendingRoot { + root_sid: root_sid.to_string(), + }, + ); order }; @@ -3457,7 +3470,7 @@ impl ActorDaemonCoordinator { ))); }; match entry { - FamilySequencerEntry::PendingRoot => { + FamilySequencerEntry::PendingRoot { .. } => { *entry = replacement; } _ => { @@ -3491,6 +3504,9 @@ impl ActorDaemonCoordinator { if ingress.root_definitely_read_only.contains(root_sid) { continue; } + if ingress.root_commit_editor_child_ids.contains_key(root_sid) { + continue; + } if !ingress.root_mutating.get(root_sid).copied().unwrap_or(true) { continue; } @@ -3663,6 +3679,8 @@ impl ActorDaemonCoordinator { ingress.root_target_repo_only.remove(root_sid); ingress.root_last_activity_ns.remove(root_sid); ingress.root_definitely_read_only.remove(root_sid); + ingress.root_commit_editor_child_ids.remove(root_sid); + ingress.root_saw_nested_mutation.remove(root_sid); ingress.root_open_connections.remove(root_sid); ingress.root_close_markers_enqueued.remove(root_sid); } @@ -3792,6 +3810,7 @@ impl ActorDaemonCoordinator { ingress.root_open_connections.iter().any(|(root, count)| { *count > 0 && !ingress.root_definitely_read_only.contains(root) + && !ingress.root_commit_editor_child_ids.contains_key(root) && ingress.root_mutating.get(root).copied().unwrap_or(true) }) } @@ -3807,6 +3826,7 @@ impl ActorDaemonCoordinator { ingress.root_open_connections.iter().any(|(root, count)| { *count > 0 && !ingress.root_definitely_read_only.contains(root) + && !ingress.root_commit_editor_child_ids.contains_key(root) && ingress.root_mutating.get(root).copied().unwrap_or(true) && ingress .root_families @@ -4596,6 +4616,14 @@ impl ActorDaemonCoordinator { .clone()) } + fn trace_root_is_waiting_for_commit_editor(&self, root_sid: &str) -> Result { + let ingress = self + .trace_ingress_state + .lock() + .map_err(|_| GitAiError::Generic("trace ingress state lock poisoned".to_string()))?; + Ok(ingress.root_commit_editor_child_ids.contains_key(root_sid)) + } + async fn drain_ready_family_sequencer_entries_locked( &self, family: &str, @@ -4612,24 +4640,34 @@ impl ActorDaemonCoordinator { let Some(state) = map.get_mut(family) else { return Ok(()); }; - while let Some(first_entry) = state.entries.first_entry() { - if matches!(first_entry.get(), FamilySequencerEntry::PendingRoot) { + let mut removable_orders = Vec::new(); + for (order, entry) in &state.entries { + if let FamilySequencerEntry::PendingRoot { root_sid } = entry { + if self.trace_root_is_waiting_for_commit_editor(root_sid)? { + continue; + } break; } - let entry_root_sid = match first_entry.get() { + let entry_root_sid = match entry { FamilySequencerEntry::ReadyCommand(command) => Some(command.root_sid.as_str()), _ => None, }; if self.family_entry_blocked_by_prior_open_trace_root( family, - first_entry.key().started_at_ns, + order.started_at_ns, entry_root_sid, )? { break; } - let (order, entry) = first_entry.remove_entry(); + removable_orders.push(*order); + } + for order in removable_orders { + let entry = state + .entries + .remove(&order) + .expect("selected family sequencer entry should still exist"); match entry { - FamilySequencerEntry::PendingRoot => { + FamilySequencerEntry::PendingRoot { .. } => { unreachable!("pending root should not be removed from sequencer front"); } other => { @@ -5019,7 +5057,7 @@ impl ActorDaemonCoordinator { ); } FamilySequencerEntry::Canceled => {} - FamilySequencerEntry::PendingRoot => {} + FamilySequencerEntry::PendingRoot { .. } => {} } } let _ = self.end_family_effect(family); @@ -6423,6 +6461,102 @@ impl ActorDaemonCoordinator { Ok(()) } + /// Applies the only trace2 wait state that is safe to pass in a family: + /// a `git commit` parent blocked in its commit-message editor. Git cannot + /// update the target ref until that child exits. Other editor-using + /// commands (notably rebase) remain barriers because they may already have + /// moved refs before opening an editor. + /// + /// Returns the root's family when it newly starts yielding so any entries + /// already queued behind it can be drained immediately. + fn update_commit_editor_wait_state( + &self, + payload: &Value, + ) -> Result, GitAiError> { + let event = payload + .get("event") + .and_then(Value::as_str) + .unwrap_or_default(); + let Some(sid) = payload.get("sid").and_then(Value::as_str) else { + return Ok(None); + }; + let root_sid = trace_root_sid(sid); + let child_id = payload.get("child_id").and_then(Value::as_u64); + + if event == "start" && sid != root_sid { + let argv = trace_payload_argv(payload); + let primary = trace_argv_primary_command(&argv); + if trace_invocation_may_mutate_refs(primary.as_deref(), &argv) { + let mut ingress = self.trace_ingress_state.lock().map_err(|_| { + GitAiError::Generic("trace ingress state lock poisoned".to_string()) + })?; + ingress + .root_saw_nested_mutation + .insert(root_sid.to_string()); + ingress.root_commit_editor_child_ids.remove(root_sid); + self.trace_ingest_progress_notify.notify_waiters(); + } + return Ok(None); + } + + if event == "child_start" + && sid == root_sid + && payload.get("child_class").and_then(Value::as_str) == Some("editor") + { + // A child_start event's own argv belongs to the editor, not Git. + // Classify the parent from the root argv retained at its start. + let mut ingress = self.trace_ingress_state.lock().map_err(|_| { + GitAiError::Generic("trace ingress state lock poisoned".to_string()) + })?; + let primary = ingress + .root_argv + .get(root_sid) + .and_then(|argv| trace_argv_primary_command(argv)); + if primary.as_deref() != Some("commit") + || ingress.root_saw_nested_mutation.contains(root_sid) + { + return Ok(None); + } + let Some(child_id) = child_id else { + return Ok(None); + }; + ingress + .root_commit_editor_child_ids + .insert(root_sid.to_string(), child_id); + drop(ingress); + let family = self + .pending_root_slots_by_root + .lock() + .map_err(|_| { + GitAiError::Generic("pending root slots map lock poisoned".to_string()) + })? + .get(root_sid) + .map(|slot| slot.family.clone()); + self.trace_ingest_progress_notify.notify_waiters(); + return Ok(family); + } + + if event == "child_exit" && sid == root_sid { + let mut ingress = self.trace_ingress_state.lock().map_err(|_| { + GitAiError::Generic("trace ingress state lock poisoned".to_string()) + })?; + if child_id.is_some_and(|child_id| { + ingress.root_commit_editor_child_ids.get(root_sid).copied() == Some(child_id) + }) { + ingress.root_commit_editor_child_ids.remove(root_sid); + self.trace_ingest_progress_notify.notify_waiters(); + } + } else if is_terminal_root_trace_event(event, sid, root_sid) { + let mut ingress = self.trace_ingress_state.lock().map_err(|_| { + GitAiError::Generic("trace ingress state lock poisoned".to_string()) + })?; + ingress.root_commit_editor_child_ids.remove(root_sid); + ingress.root_saw_nested_mutation.remove(root_sid); + } + + Ok(None) + } + async fn apply_trace_payload_to_state( &self, payload: Value, @@ -6455,6 +6589,7 @@ impl ActorDaemonCoordinator { return Ok(outcome); } + let family_yielded_by_commit_editor = self.update_commit_editor_wait_state(&payload)?; self.maybe_append_pending_root_from_trace_payload(&payload)?; let emitted = { let mut normalizer = self.normalizer.lock().await; @@ -6478,6 +6613,10 @@ impl ActorDaemonCoordinator { .await?; return Ok(TracePayloadApplyOutcome::QueuedFamily); } + if let Some(family) = family_yielded_by_commit_editor { + self.drain_ready_family_sequencer_entries(&family).await?; + return Ok(TracePayloadApplyOutcome::QueuedFamily); + } return Ok(TracePayloadApplyOutcome::None); }; let root_sid = command.root_sid.clone(); diff --git a/tests/daemon_mode.rs b/tests/daemon_mode.rs index 6573722937..00f15667cb 100644 --- a/tests/daemon_mode.rs +++ b/tests/daemon_mode.rs @@ -2485,6 +2485,255 @@ fn daemon_sync_family_ignores_open_mutating_root_from_other_family() { own_sync.join().unwrap(); } +#[test] +#[cfg(not(windows))] +fn daemon_commit_waiting_for_editor_does_not_block_later_commit() { + let repo = TestRepo::new_dedicated_daemon(); + let trace_socket = daemon_trace_socket_path(&repo); + let control_socket = daemon_control_socket_path(&repo); + let worktree = repo_workdir_string(&repo); + let git_dir = repo.path().join(".git").to_string_lossy().to_string(); + let waiting_sid = "commit-waiting-for-editor"; + + let mut waiting_trace = + open_local_socket_stream_with_timeout(&trace_socket, DAEMON_TEST_PROBE_TIMEOUT) + .expect("failed to connect waiting commit trace socket"); + write_trace_frames_to_stream( + &mut waiting_trace, + &[ + json!({ + "event": "start", + "sid": waiting_sid, + "argv": ["git", "commit"], + "time_ns": 1_000u64, + }), + json!({ + "event": "def_repo", + "sid": waiting_sid, + "worktree": worktree, + "repo": git_dir, + "time_ns": 1_001u64, + }), + json!({ + "event": "cmd_name", + "sid": waiting_sid, + "name": "commit", + "hierarchy": "commit", + "time_ns": 1_002u64, + }), + json!({ + "event": "child_start", + "sid": waiting_sid, + "child_id": 0, + "child_class": "editor", + "argv": ["vim", ".git/COMMIT_EDITMSG"], + "time_ns": 1_003u64, + }), + json!({ + "event": "start", + "sid": format!("{waiting_sid}/nested"), + "argv": ["git", "status"], + "time_ns": 1_004u64, + }), + json!({ + "event": "child_start", + "sid": format!("{waiting_sid}/nested"), + "child_id": 0, + "child_class": "pager", + "argv": ["less"], + "time_ns": 1_005u64, + }), + json!({ + "event": "child_exit", + "sid": format!("{waiting_sid}/nested"), + "child_id": 0, + "code": 0, + "time_ns": 1_006u64, + }), + ], + ); + + let later_session = repos::test_repo::new_daemon_test_sync_session_id(); + let session_arg = format!("git-ai.testSyncSession={later_session}"); + send_trace_frames( + &trace_socket, + &[ + json!({ + "event": "start", + "sid": "later-completed-commit", + "argv": ["git", "-c", session_arg, "commit", "-m", "later commit"], + "time_ns": 2_000u64, + }), + json!({ + "event": "def_repo", + "sid": "later-completed-commit", + "worktree": worktree, + "repo": git_dir, + "time_ns": 2_001u64, + }), + json!({ + "event": "cmd_name", + "sid": "later-completed-commit", + "name": "commit", + "hierarchy": "commit", + "time_ns": 2_002u64, + }), + json!({ + "event": "exit", + "sid": "later-completed-commit", + "code": 0, + "time_ns": 2_100u64, + }), + trace_atexit_frame("later-completed-commit", 0, 2_101u64), + ], + ); + + let started = std::time::Instant::now(); + while started.elapsed() < Duration::from_secs(2) { + if repo + .daemon_completion_entries() + .iter() + .any(|entry| entry.test_sync_session.as_deref() == Some(later_session.as_str())) + { + let sync = send_control_request_with_timeout( + &control_socket, + &ControlRequest::SyncFamily { + repo_working_dir: repo_workdir_string(&repo), + }, + Duration::from_secs(1), + ) + .expect("sync.family should ignore a commit blocked in its editor"); + assert!(sync.ok, "sync.family failed: {sync:?}"); + + write_trace_frames_to_stream( + &mut waiting_trace, + &[ + json!({ + "event": "child_exit", + "sid": waiting_sid, + "child_id": 0, + "code": 1, + "time_ns": 3_000u64, + }), + trace_atexit_frame(waiting_sid, 1, 3_001u64), + ], + ); + return; + } + thread::sleep(Duration::from_millis(10)); + } + + panic!( + "daemon did not process a later commit while an earlier commit waited for its editor; entries={:?}; logs={}", + repo.daemon_completion_entries(), + repo.daemon_stderr_contents() + ); +} + +#[test] +#[cfg(not(windows))] +fn daemon_rebase_waiting_for_editor_still_blocks_later_commit() { + let repo = TestRepo::new_dedicated_daemon(); + let trace_socket = daemon_trace_socket_path(&repo); + let worktree = repo_workdir_string(&repo); + let git_dir = repo.path().join(".git").to_string_lossy().to_string(); + let rebase_sid = "rebase-waiting-for-editor"; + + let mut rebase_trace = + open_local_socket_stream_with_timeout(&trace_socket, DAEMON_TEST_PROBE_TIMEOUT) + .expect("failed to connect waiting rebase trace socket"); + write_trace_frames_to_stream( + &mut rebase_trace, + &[ + json!({ + "event": "start", + "sid": rebase_sid, + "argv": ["git", "rebase", "--interactive", "HEAD~2"], + "time_ns": 1_000u64, + }), + json!({ + "event": "def_repo", + "sid": rebase_sid, + "worktree": worktree, + "repo": git_dir, + "time_ns": 1_001u64, + }), + json!({ + "event": "child_start", + "sid": rebase_sid, + "child_id": 0, + "child_class": "editor", + "argv": ["vim", ".git/rebase-merge/git-rebase-todo"], + "time_ns": 1_002u64, + }), + ], + ); + + let later_session = repos::test_repo::new_daemon_test_sync_session_id(); + let session_arg = format!("git-ai.testSyncSession={later_session}"); + send_trace_frames( + &trace_socket, + &[ + json!({ + "event": "start", + "sid": "commit-behind-open-rebase", + "argv": ["git", "-c", session_arg, "commit", "-m", "later commit"], + "time_ns": 2_000u64, + }), + json!({ + "event": "def_repo", + "sid": "commit-behind-open-rebase", + "worktree": worktree, + "repo": git_dir, + "time_ns": 2_001u64, + }), + json!({ + "event": "exit", + "sid": "commit-behind-open-rebase", + "code": 0, + "time_ns": 2_100u64, + }), + trace_atexit_frame("commit-behind-open-rebase", 0, 2_101u64), + ], + ); + + thread::sleep(Duration::from_millis(250)); + assert!( + !repo + .daemon_completion_entries() + .iter() + .any(|entry| entry.test_sync_session.as_deref() == Some(later_session.as_str())), + "a rebase editor may open after refs have moved and must remain an ordering barrier" + ); + + write_trace_frames_to_stream( + &mut rebase_trace, + &[ + json!({ + "event": "child_exit", + "sid": rebase_sid, + "child_id": 0, + "code": 1, + "time_ns": 3_000u64, + }), + trace_atexit_frame(rebase_sid, 1, 3_001u64), + ], + ); + + let started = std::time::Instant::now(); + while started.elapsed() < Duration::from_secs(2) { + if repo + .daemon_completion_entries() + .iter() + .any(|entry| entry.test_sync_session.as_deref() == Some(later_session.as_str())) + { + return; + } + thread::sleep(Duration::from_millis(10)); + } + panic!("later commit did not drain after the rebase root completed"); +} + #[test] #[cfg(not(windows))] fn daemon_partial_trace_line_does_not_block_checkpoint_control_request() {