Merge repository-authority runner reuse into apply-filter guard

codex-rs/core/src/session/review.rs

@@ -159,13 +159,16 @@ pub(super) async fn spawn_review_thread(
     if tc.environments.single_local_environment_cwd().is_some() {
         tc.turn_metadata_state.spawn_git_enrichment_task();
     }
+    // TODO(ccunningham): Review turns currently rely on `spawn_task` for TurnComplete but do not
+    // emit a parent TurnStarted. Consider giving review a full parent turn lifecycle
+    // (TurnStarted + TurnComplete) for consistency with other standalone tasks.
+    sess.spawn_task(tc.clone(), input, ReviewTask::new()).await;
+
+    // Announce entering review mode so UIs can switch modes.
     let review_request = ReviewRequest {
         target: resolved.target,
         user_facing_hint: Some(resolved.user_facing_hint),
     };
-    // TODO(ccunningham): Review turns currently rely on `spawn_task` for TurnComplete but do not
-    // emit a parent TurnStarted. Consider giving review a full parent turn lifecycle
-    // (TurnStarted + TurnComplete) for consistency with other standalone tasks.
-    sess.spawn_task(tc, input, ReviewTask::new(review_request))
+    sess.send_event(&tc, EventMsg::EnteredReviewMode(review_request))
         .await;
 }

codex-rs/core/src/session/tests.rs

@@ -131,8 +131,6 @@ use codex_protocol::protocol::RealtimeConversationListVoicesResponseEvent;
 use codex_protocol::protocol::RealtimeVoice;
 use codex_protocol::protocol::RealtimeVoicesList;
 use codex_protocol::protocol::ResumedHistory;
-use codex_protocol::protocol::ReviewRequest;
-use codex_protocol::protocol::ReviewTarget;
 use codex_protocol::protocol::RolloutItem;
 use codex_protocol::protocol::SessionMeta;
 use codex_protocol::protocol::SessionMetaLine;
@@ -10388,24 +10386,14 @@ async fn abort_review_task_emits_exited_then_aborted_and_records_history() {
         }],
         client_id: None,
     }];
-    sess.spawn_task(
-        Arc::clone(&tc),
-        input,
-        ReviewTask::new(ReviewRequest {
-            target: ReviewTarget::Custom {
-                instructions: "review".to_string(),
-            },
-            user_facing_hint: None,
-        }),
-    )
-    .await;
+    sess.spawn_task(Arc::clone(&tc), input, ReviewTask::new())
+        .await;
 
     sess.abort_all_tasks(TurnAbortReason::Interrupted).await;
 
-    // Aborting a review task should enter and exit review mode before surfacing the abort.
+    // Aborting a review task should exit review mode before surfacing the abort to the client.
     // We scan for these events (rather than relying on fixed ordering) since unrelated events
     // may interleave.
-    let mut entered_review_mode_idx = None;
     let mut exited_review_mode_idx = None;
     let mut turn_aborted_idx = None;
     let mut idx = 0usize;
@@ -10419,12 +10407,6 @@ async fn abort_review_task_emits_exited_then_aborted_and_records_history() {
         let event_idx = idx;
         idx = idx.saturating_add(1);
         match evt.msg {
-            EventMsg::EnteredReviewMode(_) => {
-                assert!(
-                    entered_review_mode_idx.replace(event_idx).is_none(),
-                    "expected exactly one EnteredReviewMode"
-                );
-            }
             EventMsg::ExitedReviewMode(ev) => {
                 assert!(ev.review_output.is_none());
                 exited_review_mode_idx = Some(event_idx);
@@ -10437,10 +10419,6 @@ async fn abort_review_task_emits_exited_then_aborted_and_records_history() {
             _ => {}
         }
     }
-    assert!(
-        entered_review_mode_idx.is_some(),
-        "expected EnteredReviewMode before abort"
-    );
     assert!(
         exited_review_mode_idx.is_some(),
         "expected ExitedReviewMode after abort"
@@ -10449,10 +10427,6 @@ async fn abort_review_task_emits_exited_then_aborted_and_records_history() {
         turn_aborted_idx.is_some(),
         "expected TurnAborted after abort"
     );
-    assert!(
-        entered_review_mode_idx.unwrap() < exited_review_mode_idx.unwrap(),
-        "expected EnteredReviewMode before ExitedReviewMode"
-    );
     assert!(
         exited_review_mode_idx.unwrap() < turn_aborted_idx.unwrap(),
         "expected ExitedReviewMode before TurnAborted"

codex-rs/core/src/tasks/review.rs

@@ -13,12 +13,7 @@ use codex_protocol::protocol::EventMsg;
 use codex_protocol::protocol::ExitedReviewModeEvent;
 use codex_protocol::protocol::ItemCompletedEvent;
 use codex_protocol::protocol::ReviewOutputEvent;
-use codex_protocol::protocol::ReviewRequest;
 use codex_protocol::protocol::SubAgentSource;
-use futures::FutureExt;
-use futures::future::BoxFuture;
-use futures::future::Shared;
-use tokio::sync::OnceCell;
 use tokio_util::sync::CancellationToken;
 
 use crate::codex_delegate::run_codex_thread_one_shot;
@@ -36,47 +31,12 @@ use super::SessionTask;
 use super::SessionTaskContext;
 use super::SessionTaskResult;
 
-type EnterReviewModeFuture = Shared<BoxFuture<'static, ()>>;
-
-pub(crate) struct ReviewTask {
-    review_request: ReviewRequest,
-    enter_review_mode: OnceCell<EnterReviewModeFuture>,
-}
+#[derive(Clone, Copy)]
+pub(crate) struct ReviewTask;
 
 impl ReviewTask {
-    pub(crate) fn new(review_request: ReviewRequest) -> Self {
-        Self {
-            review_request,
-            enter_review_mode: OnceCell::new(),
-        }
-    }
-
-    async fn enter_review_mode_once(
-        &self,
-        session: &Arc<SessionTaskContext>,
-        ctx: &Arc<TurnContext>,
-    ) {
-        // Normal completion and abort can race. Both paths await the same shared send future so
-        // neither ExitedReviewMode nor a terminal turn event can overtake EnteredReviewMode.
-        let enter_review_mode = self
-            .enter_review_mode
-            .get_or_init(|| {
-                let session = session.clone_session();
-                let ctx = Arc::clone(ctx);
-                let review_request = self.review_request.clone();
-                std::future::ready(
-                    async move {
-                        session
-                            .send_event(ctx.as_ref(), EventMsg::EnteredReviewMode(review_request))
-                            .await;
-                    }
-                    .boxed()
-                    .shared(),
-                )
-            })
-            .await
-            .clone();
-        enter_review_mode.await;
+    pub(crate) fn new() -> Self {
+        Self
     }
 }
 
@@ -96,7 +56,6 @@ impl SessionTask for ReviewTask {
         input: Vec<TurnInput>,
         cancellation_token: CancellationToken,
     ) -> SessionTaskResult {
-        self.enter_review_mode_once(&session, &ctx).await;
         session.session.services.session_telemetry.counter(
             "codex.task.review",
             /*inc*/ 1,
@@ -130,7 +89,6 @@ impl SessionTask for ReviewTask {
     }
 
     async fn abort(&self, session: Arc<SessionTaskContext>, ctx: Arc<TurnContext>) {
-        self.enter_review_mode_once(&session, &ctx).await;
         exit_review_mode(session.clone_session(), /*review_output*/ None, ctx).await;
     }
 }

codex-rs/core/tests/suite/review.rs

@@ -31,7 +31,6 @@ use std::path::PathBuf;
 use std::sync::Arc;
 use tempfile::TempDir;
 use tokio::io::AsyncWriteExt as _;
-use tokio::time::Duration;
 use uuid::Uuid;
 use wiremock::MockServer;
 
@@ -844,71 +843,8 @@ async fn review_uses_overridden_cwd_for_base_branch_merge_base() {
         .await
         .unwrap();
 
-    let entered_review_mode = tokio::time::timeout(Duration::from_secs(10), async {
-        loop {
-            let event = codex.next_event().await.expect("stream ended unexpectedly");
-            match event.msg {
-                EventMsg::EnteredReviewMode(_) => return Ok(()),
-                EventMsg::Error(error) => {
-                    return Err(format!(
-                        "review failed before entering review mode: {error:#?}"
-                    ));
-                }
-                EventMsg::TurnAborted(aborted) => {
-                    return Err(format!(
-                        "review aborted before entering review mode: {aborted:#?}"
-                    ));
-                }
-                EventMsg::ExitedReviewMode(exited) => {
-                    return Err(format!(
-                        "review exited before entering review mode: {exited:#?}"
-                    ));
-                }
-                EventMsg::TurnComplete(completed) => {
-                    return Err(format!(
-                        "review completed before entering review mode: {completed:#?}"
-                    ));
-                }
-                _ => {}
-            }
-        }
-    })
-    .await
-    .unwrap_or_else(|_| {
-        panic!(
-            "timeout before entering review mode; mock response requests observed: {}",
-            request_log.requests().len()
-        )
-    });
-    entered_review_mode.unwrap_or_else(|error| panic!("{error}"));
-
-    let completed_turn = tokio::time::timeout(Duration::from_secs(10), async {
-        loop {
-            let event = codex.next_event().await.expect("stream ended unexpectedly");
-            match event.msg {
-                EventMsg::TurnComplete(_) => return Ok(()),
-                EventMsg::Error(error) => {
-                    return Err(format!(
-                        "review failed before completing the turn: {error:#?}"
-                    ));
-                }
-                EventMsg::TurnAborted(aborted) => {
-                    return Err(format!(
-                        "review aborted before completing the turn: {aborted:#?}"
-                    ));
-                }
-                _ => {}
-            }
-        }
-    })
-    .await
-    .unwrap_or_else(|_| {
-        panic!(
-            "timeout after entering review mode; mock response requests observed: {}",
-            request_log.requests().len()
-        )
-    });
-    completed_turn.unwrap_or_else(|error| panic!("{error}"));
+    let _entered = wait_for_event(&codex, |ev| matches!(ev, EventMsg::EnteredReviewMode(_))).await;
+    let _complete = wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
 
     let requests = request_log.requests();
     assert_eq!(requests.len(), 1);

codex-rs/git-utils/src/branch.rs

@@ -2,10 +2,11 @@ use std::ffi::OsString;
 use std::path::Path;
 
 use crate::GitToolingError;
+use crate::git_command::GitRunner;
 use crate::operations::ensure_git_repository;
 use crate::operations::resolve_head;
 use crate::operations::resolve_repository_root;
-use crate::operations::run_git_for_stdout;
+use crate::operations::run_git_for_stdout_with_runner;
 
 /// Returns the merge-base commit between `HEAD` and the latest version between local
 /// and remote of the provided branch, if both exist.
@@ -16,25 +17,28 @@ pub fn merge_base_with_head(
     repo_path: &Path,
     branch: &str,
 ) -> Result<Option<String>, GitToolingError> {
-    ensure_git_repository(repo_path)?;
-    let repo_root = resolve_repository_root(repo_path)?;
-    let head = match resolve_head(repo_root.as_path())? {
+    let git = GitRunner::for_cwd_io(repo_path)?;
+    ensure_git_repository(&git, repo_path)?;
+    let repo_root = resolve_repository_root(&git, repo_path)?;
+    let head = match resolve_head(&git, repo_root.as_path())? {
         Some(head) => head,
         None => return Ok(None),
     };
 
-    let Some(branch_ref) = resolve_branch_ref(repo_root.as_path(), branch)? else {
+    let Some(branch_ref) = resolve_branch_ref(&git, repo_root.as_path(), branch)? else {
         return Ok(None);
     };
 
-    let preferred_ref =
-        if let Some(upstream) = resolve_upstream_if_remote_ahead(repo_root.as_path(), branch)? {
-            resolve_branch_ref(repo_root.as_path(), &upstream)?.unwrap_or(branch_ref)
-        } else {
-            branch_ref
-        };
+    let preferred_ref = if let Some(upstream) =
+        resolve_upstream_if_remote_ahead(&git, repo_root.as_path(), branch)?
+    {
+        resolve_branch_ref(&git, repo_root.as_path(), &upstream)?.unwrap_or(branch_ref)
+    } else {
+        branch_ref
+    };
 
-    let merge_base = run_git_for_stdout(
+    let merge_base = run_git_for_stdout_with_runner(
+        &git,
         repo_root.as_path(),
         vec![
             OsString::from("merge-base"),
@@ -47,8 +51,13 @@ pub fn merge_base_with_head(
     Ok(Some(merge_base))
 }
 
-fn resolve_branch_ref(repo_root: &Path, branch: &str) -> Result<Option<String>, GitToolingError> {
-    let rev = run_git_for_stdout(
+fn resolve_branch_ref(
+    git: &GitRunner,
+    repo_root: &Path,
+    branch: &str,
+) -> Result<Option<String>, GitToolingError> {
+    let rev = run_git_for_stdout_with_runner(
+        git,
         repo_root,
         vec![
             OsString::from("rev-parse"),
@@ -66,10 +75,12 @@ fn resolve_branch_ref(repo_root: &Path, branch: &str) -> Result<Option<String>,
 }
 
 fn resolve_upstream_if_remote_ahead(
+    git: &GitRunner,
     repo_root: &Path,
     branch: &str,
 ) -> Result<Option<String>, GitToolingError> {
-    let upstream = match run_git_for_stdout(
+    let upstream = match run_git_for_stdout_with_runner(
+        git,
         repo_root,
         vec![
             OsString::from("rev-parse"),
@@ -90,7 +101,8 @@ fn resolve_upstream_if_remote_ahead(
         Err(other) => return Err(other),
     };
 
-    let counts = match run_git_for_stdout(
+    let counts = match run_git_for_stdout_with_runner(
+        git,
         repo_root,
         vec![
             OsString::from("rev-list"),
@@ -120,7 +132,17 @@ fn resolve_upstream_if_remote_ahead(
 mod tests {
     use super::merge_base_with_head;
     use crate::GitToolingError;
+    #[cfg(unix)]
+    use crate::git_command::GitRunner;
+    use crate::git_command::git_runner_construction_count;
+    use crate::git_command::reset_git_runner_construction_count;
+    #[cfg(unix)]
+    use crate::operations::ensure_git_repository;
+    #[cfg(unix)]
+    use crate::operations::resolve_repository_root;
     use pretty_assertions::assert_eq;
+    #[cfg(unix)]
+    use std::io;
     use std::path::Path;
     use std::process::Command;
     use tempfile::tempdir;
@@ -187,8 +209,14 @@ mod tests {
         run_git_in(repo, &["checkout", "feature"]);
 
         let expected = run_git_stdout(repo, &["merge-base", "HEAD", "main"]);
+        reset_git_runner_construction_count();
         let merge_base = merge_base_with_head(repo, "main")?;
         assert_eq!(merge_base, Some(expected));
+        assert_eq!(
+            git_runner_construction_count(),
+            1,
+            "one high-level merge-base call must retain one Git runner"
+        );
 
         Ok(())
     }
@@ -253,4 +281,67 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn merge_base_returns_none_when_head_is_unborn() -> Result<(), GitToolingError> {
+        let temp = tempdir()?;
+        let repo = temp.path();
+        init_test_repo(repo);
+
+        let merge_base = merge_base_with_head(repo, "main")?;
+        assert_eq!(merge_base, None);
+
+        Ok(())
+    }
+
+    #[test]
+    fn merge_base_reports_non_repository() {
+        let temp = tempdir().expect("tempdir");
+        let path = temp.path();
+
+        let error = merge_base_with_head(path, "main").expect_err("non-repository must fail");
+        let GitToolingError::NotAGitRepository { path: error_path } = error else {
+            panic!("expected non-repository error, got {error:?}");
+        };
+        assert_eq!(error_path, path);
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn retained_runner_refuses_gitdir_retarget_between_subcommands() {
+        use std::os::unix::fs::symlink;
+
+        let fixture = tempdir().expect("fixture");
+        let repo = fixture.path().join("repo");
+        let admin = fixture.path().join("external-admin");
+        std::fs::create_dir_all(&repo).expect("create repository");
+        run_git_in(
+            fixture.path(),
+            &[
+                "init",
+                "--separate-git-dir",
+                admin.to_str().expect("UTF-8 admin path"),
+                repo.to_str().expect("UTF-8 repository path"),
+            ],
+        );
+
+        reset_git_runner_construction_count();
+        let git = GitRunner::for_cwd(&repo).expect("operation-scoped Git runner");
+        ensure_git_repository(&git, &repo).expect("first merge-base subcommand");
+
+        symlink(&admin, repo.join("switch")).expect("Git-dir route switch");
+        std::fs::write(repo.join(".git"), "gitdir: switch\n").expect("retarget Git-dir marker");
+
+        let error = resolve_repository_root(&git, &repo)
+            .expect_err("retained authority must refuse the second subcommand");
+        let GitToolingError::Io(error) = error else {
+            panic!("expected metadata revalidation error, got {error:?}");
+        };
+        assert_eq!(error.kind(), io::ErrorKind::PermissionDenied, "{error}");
+        assert_eq!(
+            git_runner_construction_count(),
+            1,
+            "subcommands must not rediscover authority after metadata retargeting"
+        );
+    }
 }

codex-rs/git-utils/src/git_command.rs

@@ -72,6 +72,8 @@ impl GitCommand {
 
 impl GitRunner {
     pub(crate) fn for_cwd(cwd: &Path) -> Result<Self, GitReadError> {
+        #[cfg(test)]
+        GIT_RUNNER_CONSTRUCTION_COUNT.with(|count| count.set(count.get() + 1));
         let authority = repository_authority_for_cwd(cwd)?;
         let search_path = std::env::var_os("PATH").ok_or(GitReadError::NoTrustedGit)?;
         Self::from_search_path(authority, &search_path)
@@ -145,6 +147,21 @@ impl GitRunner {
     }
 }
 
+#[cfg(test)]
+thread_local! {
+    static GIT_RUNNER_CONSTRUCTION_COUNT: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
+}
+
+#[cfg(test)]
+pub(crate) fn reset_git_runner_construction_count() {
+    GIT_RUNNER_CONSTRUCTION_COUNT.with(|count| count.set(0));
+}
+
+#[cfg(test)]
+pub(crate) fn git_runner_construction_count() -> usize {
+    GIT_RUNNER_CONSTRUCTION_COUNT.with(std::cell::Cell::get)
+}
+
 pub(crate) fn repository_authority_for_cwd(
     cwd: &Path,
 ) -> Result<RepositoryAuthority, GitReadError> {

codex-rs/git-utils/src/operations.rs

@@ -7,8 +7,9 @@ use crate::GitToolingError;
 use crate::git_command::GitRunner;
 use crate::safe_git::DISABLED_HOOKS_PATH;
 
-pub(crate) fn ensure_git_repository(path: &Path) -> Result<(), GitToolingError> {
-    match run_git_for_stdout(
+pub(crate) fn ensure_git_repository(git: &GitRunner, path: &Path) -> Result<(), GitToolingError> {
+    match run_git_for_stdout_with_runner(
+        git,
         path,
         vec![
             OsString::from("rev-parse"),
@@ -29,8 +30,12 @@ pub(crate) fn ensure_git_repository(path: &Path) -> Result<(), GitToolingError>
     }
 }
 
-pub(crate) fn resolve_head(path: &Path) -> Result<Option<String>, GitToolingError> {
-    match run_git_for_stdout(
+pub(crate) fn resolve_head(
+    git: &GitRunner,
+    path: &Path,
+) -> Result<Option<String>, GitToolingError> {
+    match run_git_for_stdout_with_runner(
+        git,
         path,
         vec![
             OsString::from("rev-parse"),
@@ -45,8 +50,12 @@ pub(crate) fn resolve_head(path: &Path) -> Result<Option<String>, GitToolingErro
     }
 }
 
-pub(crate) fn resolve_repository_root(path: &Path) -> Result<PathBuf, GitToolingError> {
-    let root = run_git_for_stdout(
+pub(crate) fn resolve_repository_root(
+    git: &GitRunner,
+    path: &Path,
+) -> Result<PathBuf, GitToolingError> {
+    let root = run_git_for_stdout_with_runner(
+        git,
         path,
         vec![
             OsString::from("rev-parse"),
@@ -70,6 +79,7 @@ where
     Ok(())
 }
 
+#[cfg(test)]
 pub(crate) fn run_git_for_stdout<I, S>(
     dir: &Path,
     args: I,
@@ -79,7 +89,21 @@ where
     I: IntoIterator<Item = S>,
     S: AsRef<OsStr>,
 {
-    let run = run_git(dir, args, env)?;
+    let git = GitRunner::for_cwd_io(dir)?;
+    run_git_for_stdout_with_runner(&git, dir, args, env)
+}
+
+pub(crate) fn run_git_for_stdout_with_runner<I, S>(
+    git: &GitRunner,
+    dir: &Path,
+    args: I,
+    env: Option<&[(OsString, OsString)]>,
+) -> Result<String, GitToolingError>
+where
+    I: IntoIterator<Item = S>,
+    S: AsRef<OsStr>,
+{
+    let run = run_git_with_runner(git, dir, args, env)?;
     String::from_utf8(run.output.stdout)
         .map(|value| value.trim().to_string())
         .map_err(|source| GitToolingError::GitOutputUtf8 {
@@ -93,6 +117,20 @@ fn run_git<I, S>(
     args: I,
     env: Option<&[(OsString, OsString)]>,
 ) -> Result<GitRun, GitToolingError>
+where
+    I: IntoIterator<Item = S>,
+    S: AsRef<OsStr>,
+{
+    let git = GitRunner::for_cwd_io(dir)?;
+    run_git_with_runner(&git, dir, args, env)
+}
+
+fn run_git_with_runner<I, S>(
+    git: &GitRunner,
+    dir: &Path,
+    args: I,
+    env: Option<&[(OsString, OsString)]>,
+) -> Result<GitRun, GitToolingError>
 where
     I: IntoIterator<Item = S>,
     S: AsRef<OsStr>,
@@ -109,7 +147,6 @@ where
         args_vec.push(OsString::from(arg.as_ref()));
     }
     let command_string = build_command_string(&args_vec);
-    let git = GitRunner::for_cwd_io(dir)?;
     let mut command = git.command_for_cwd(dir)?;
     if let Some(envs) = env {
         for (key, value) in envs {