diff --git a/docs/REFERENCE.md b/docs/REFERENCE.md index 4ca5cb2..b73e0ad 100644 --- a/docs/REFERENCE.md +++ b/docs/REFERENCE.md @@ -743,6 +743,11 @@ The supervisor also creates its ordinary server child with `CREATE_NO_WINDOW`. T prevents the otherwise empty console window without changing the Interactive logon principal required for tunnel and network access. +The supervisor holds an operating-system lock for its lifetime. A second +`service run` exits before launching another server or opening the shared log. +The lock is released when the process exits, including after a crash; a leftover +lock file does not prevent recovery. + The Windows supervisor begins liveness monitoring after the server first answers `/health` successfully. Six consecutive failed probes (with a five-second pause between them and a three-second request timeout) restart the contained server process tree. A slow diff --git a/src/diff.rs b/src/diff.rs index b48fc3e..1857943 100644 --- a/src/diff.rs +++ b/src/diff.rs @@ -634,9 +634,20 @@ async fn load_or_initialize_persistent( legacy_owner_key: &str, ) -> Result { let refs = checkpoint_refs(&workspace.key, owner_key); - migrate_legacy_checkpoint_refs(config, workspace, owner_key, legacy_owner_key, &refs).await?; - let project_open = read_ref(config, &workspace.git_root, &refs.project_open).await?; - let last_diff = read_ref(config, &workspace.git_root, &refs.last_diff).await?; + // Most calls already have both baselines. Do not repeat the legacy namespace + // probes before every edit or command; Windows pays for each Git process. + let (mut project_open, mut last_diff) = tokio::try_join!( + read_ref(config, &workspace.git_root, &refs.project_open), + read_ref(config, &workspace.git_root, &refs.last_diff), + )?; + if project_open.is_none() || last_diff.is_none() { + migrate_legacy_checkpoint_refs(config, workspace, owner_key, legacy_owner_key, &refs) + .await?; + (project_open, last_diff) = tokio::try_join!( + read_ref(config, &workspace.git_root, &refs.project_open), + read_ref(config, &workspace.git_root, &refs.last_diff), + )?; + } match (project_open, last_diff) { (Some(project_open), Some(last_diff)) => Ok(CheckpointPair { @@ -1379,12 +1390,19 @@ fn strings(values: &[&str]) -> Vec { values.iter().map(OsString::from).collect() } +#[cfg(test)] +tokio::task_local! { + static TEST_GIT_COMMAND_COUNT: std::cell::Cell; +} + fn git_command( config: &AppConfig, cwd: &Path, args: &[OsString], environment: &[(OsString, OsString)], ) -> Command { + #[cfg(test)] + let _ = TEST_GIT_COMMAND_COUNT.try_with(|count| count.set(count.get() + 1)); let mut command = Command::new("git"); command.args(args).current_dir(cwd); scrub_untrusted_child_env(&mut command, config); @@ -1474,19 +1492,7 @@ async fn run_git( } Ok::<_, std::io::Error>(data) }; - let err = async { - let mut data = Vec::new(); - stderr - .take(MAX_GIT_ERROR_BYTES as u64 + 1) - .read_to_end(&mut data) - .await?; - if data.len() > MAX_GIT_ERROR_BYTES { - return Err(std::io::Error::other( - "Git checkpoint diagnostics exceeded 64 KiB.", - )); - } - Ok::<_, std::io::Error>(data) - }; + let err = read_git_diagnostics(stderr); let (_, stdout, stderr, status) = tokio::try_join!(send, out, err, child.wait())?; Ok::<_, std::io::Error>(Output { status, @@ -1517,6 +1523,26 @@ async fn run_git( } } +// Git may emit a warning for every path (for example CRLF conversion). Drain +// stderr under run_git's deadline so pipe backpressure cannot stall a checkpoint, +// while retaining bounded diagnostics and the command's actual exit status. +async fn read_git_diagnostics( + mut reader: R, +) -> std::io::Result> { + let mut data = Vec::new(); + (&mut reader) + .take(MAX_GIT_ERROR_BYTES as u64 + 1) + .read_to_end(&mut data) + .await?; + if data.len() > MAX_GIT_ERROR_BYTES { + tokio::io::copy(&mut reader, &mut tokio::io::sink()).await?; + const TRUNCATED: &[u8] = b"\n[Additional Git diagnostics omitted.]\n"; + data.truncate(MAX_GIT_ERROR_BYTES - TRUNCATED.len()); + data.extend_from_slice(TRUNCATED); + } + Ok(data) +} + async fn git_checked_input( config: &AppConfig, cwd: &Path, @@ -1582,6 +1608,55 @@ mod tests { use super::*; + #[tokio::test] + async fn checkpoint_diagnostics_are_bounded_but_fully_drained() { + for length in [0, 128, MAX_GIT_ERROR_BYTES, MAX_GIT_ERROR_BYTES * 16] { + let input = vec![b'w'; length]; + let mut reader = input.as_slice(); + let output = read_git_diagnostics(&mut reader).await.unwrap(); + assert!(reader.is_empty(), "all stderr must be consumed"); + assert!(output.len() <= MAX_GIT_ERROR_BYTES); + if length <= MAX_GIT_ERROR_BYTES { + assert_eq!(output, input); + } else { + assert!(output.ends_with(b"[Additional Git diagnostics omitted.]\n")); + } + } + // A producer which never closes remains subject to the command deadline. + let (_writer, reader) = tokio::io::duplex(16); + assert!( + tokio::time::timeout(Duration::from_millis(30), read_git_diagnostics(reader)) + .await + .is_err() + ); + } + + #[cfg(windows)] + #[tokio::test] + async fn noisy_checkpoint_preserves_success_and_failure_exit_statuses() { + for code in [0, 7] { + let mut command = Command::new("powershell.exe"); + command.args(["-NoProfile", "-NonInteractive", "-Command", &format!( + "[Console]::Error.Write(('warning' * 20000)); [Console]::Out.Write('checkpoint'); exit {code}" + )]).kill_on_drop(true); + let output = super::run_git(command, None, 10_000).await.unwrap(); + assert_eq!(output.status.code(), Some(code)); + assert_eq!(output.stdout, b"checkpoint"); + assert_eq!(output.stderr.len(), MAX_GIT_ERROR_BYTES); + assert!( + output + .stderr + .ends_with(b"[Additional Git diagnostics omitted.]\n") + ); + if code != 0 { + assert!( + git_failure("git checkpoint", &output.stderr, output.status.code()) + .contains("failed") + ); + } + } + } + #[cfg(unix)] #[tokio::test] async fn checkpoint_deadlines_terminate_the_owned_process_group() { @@ -1628,6 +1703,52 @@ mod tests { .collect() } + #[tokio::test] + async fn warm_mutation_validates_refs_without_repeating_legacy_probes() { + let repo = tempfile::tempdir().unwrap(); + run_git(repo.path(), &["init", "--quiet"]); + std::fs::write(repo.path().join("file.txt"), "before\n").unwrap(); + let config = crate::config::default_config(repo.path().to_path_buf()); + let owner = DiffOwner::conversation( + &ConversationIdentity::from_openai_session("warm-checkpoint").unwrap(), + ); + let manager = DiffCheckpointManager::new(); + manager + .ensure_initialized(&config, owner.clone()) + .await + .unwrap(); + std::fs::write(repo.path().join("file.txt"), "after\n").unwrap(); + + TEST_GIT_COMMAND_COUNT + .scope(std::cell::Cell::new(0), async { + let (availability, guard) = manager + .begin_mutation(&config, owner.clone()) + .await + .unwrap(); + assert_eq!(availability, DiffAvailability::Ready); + // Resolve the workspace and validate the two current commit refs. + // Missing/corrupt baselines still take the initialization path. + assert_eq!(TEST_GIT_COMMAND_COUNT.with(|count| count.get()), 3); + drop(guard); + }) + .await; + + let diff = manager + .show_diff( + &config, + owner, + DiffRequest { + since: DiffBaseline::ProjectOpen, + advance: false, + include_patch: true, + }, + ) + .await + .unwrap(); + assert!(diff.patch.contains("-before")); + assert!(diff.patch.contains("+after")); + } + #[test] fn every_generated_diff_uses_histogram_and_rename_detection() { let summary_args = rendered_args(diff_args( diff --git a/src/exec_sessions.rs b/src/exec_sessions.rs index 9d79502..7b05cfa 100644 --- a/src/exec_sessions.rs +++ b/src/exec_sessions.rs @@ -386,6 +386,7 @@ impl PendingBuffer { } } + #[cfg(test)] fn is_empty(&self) -> bool { self.head.is_empty() && self.tail.is_empty() && self.omitted == 0 } @@ -488,7 +489,8 @@ impl ExecSession { tokio::time::sleep(remaining.min(Duration::from_millis(25))).await; } - if self.exit_code().is_some() { + let exited = self.exit_code().is_some(); + if exited { // Let the readers finish so the final bytes are not lost to the race // between process exit and stream EOF. let _ = @@ -496,7 +498,9 @@ impl ExecSession { } let output = self.take_pending(); - (output.text, self.exit_code().is_some(), output.truncated) + // If the child exits after the snapshot, keep its session addressable + // for the next poll's final drain instead of discarding late stdout. + (output.text, exited, output.truncated) } } @@ -536,6 +540,7 @@ impl ExecSessionState { self.sessions.lock().unwrap().remove(&id) } + #[cfg(test)] fn len(&self) -> usize { self.sessions.lock().unwrap().len() } @@ -556,6 +561,36 @@ impl ExecSessionState { reap_sessions(&mut self.sessions.lock().unwrap(), idle_timeout) } + /// Makes room for one more session. Only running processes count against + /// `max`; when the map is full of finished-but-unread results, the oldest + /// of those (by last client activity) are dropped to admit the new command. + fn admit(&self, max: usize) -> bool { + let mut sessions = self.sessions.lock().unwrap(); + if sessions.len() < max { + return true; + } + let mut finished = sessions + .values() + .filter(|session| session.exit_code().is_some()) + .map(|session| (*session.last_activity.lock().unwrap(), session.id)) + .collect::>(); + finished.sort_unstable(); + let excess = sessions.len() + 1 - max; + if finished.len() < excess { + return false; + } + for (_, id) in finished.into_iter().take(excess) { + if let Some(evicted) = sessions.remove(&id) { + tracing::info!( + exec_session_id = evicted.id, + exit_code = ?evicted.exit_code(), + "unread finished exec session evicted to admit a new command" + ); + } + } + true + } + /// Starts a transport-owned reaper. It holds only a weak reference, so it /// exits when the transport's last [`SessionState`] view is dropped. fn spawn_idle_reaper(self: &Arc, idle_timeout: Duration) { @@ -1104,10 +1139,12 @@ impl SessionState { self.exec.remove(id) } + #[cfg(test)] fn exec_session_count(&self) -> usize { self.exec.len() } + #[cfg(test)] fn reap_exec_sessions(&self, idle_timeout: Duration) -> Vec> { self.exec.reap(idle_timeout) } @@ -1457,21 +1494,26 @@ impl SessionState { } } -/// Removes finished-and-empty sessions and, when `idle_timeout` is non-zero, -/// removes every session idle beyond it. Running idle sessions are returned so -/// the caller can kill them without holding the map lock; completed sessions -/// whose final output was never polled are simply expired. +/// Expires sessions only after their configured non-zero idle timeout. +/// A quiet exit still has an unread exit status. The tools remove a session +/// after returning that final result; an empty output buffer is not an ack. +/// Running expired sessions are returned for termination outside the map lock. fn reap_sessions( map: &mut HashMap>, idle_timeout: Duration, ) -> Vec> { let mut killed = Vec::new(); map.retain(|_, s| { - if s.exit_code().is_some() && s.pending.lock().unwrap().is_empty() { - return false; - } if !idle_timeout.is_zero() && s.idle_for() >= idle_timeout { - if s.exit_code().is_none() { + let exit_code = s.exit_code(); + tracing::info!( + exec_session_id = s.id, + process_id = ?s.pid, + exit_code = ?exit_code, + idle_timeout_ms = idle_timeout.as_millis() as u64, + "exec session expired by configured idle timeout" + ); + if exit_code.is_none() { killed.push(s.clone()); } return false; @@ -1481,12 +1523,10 @@ fn reap_sessions( killed } -/// Removes sessions whose process already exited and left nothing buffered. -/// Idle enforcement is the background reaper's job (see `spawn_idle_reaper`), -/// so this opportunistic pass never kills a running session. -pub fn reap_finished_sessions(state: &SessionState) { - let _ = state.reap_exec_sessions(Duration::ZERO); -} +/// Compatibility entry point; final results are now retained until polled. +/// Configured idle expiry is enforced by the background reaper. +#[deprecated(note = "Poll final results; configured idle cleanup runs in the background")] +pub fn reap_finished_sessions(_state: &SessionState) {} /// Spawns `cmd` through a shell and registers it as a resident session. The /// caller is responsible for having validated `cmd` against policy first. @@ -1497,15 +1537,13 @@ pub fn start_exec_session( cwd: &std::path::Path, shell: Option<&str>, ) -> Result, String> { - reap_finished_sessions(state); - { - let live = state.exec_session_count(); - if live >= config.exec.max_sessions { - return Err(format!( - "Too many live exec sessions ({}). Finish or terminate an existing session before starting another.", - config.exec.max_sessions - )); - } + // Finished sessions keep their final output for the client, but they must + // not hold admission slots that only running processes need. + if !state.exec.admit(config.exec.max_sessions) { + return Err(format!( + "Too many live exec sessions ({}). Finish or terminate an existing session before starting another.", + config.exec.max_sessions + )); } let parts = match shell.filter(|value| !value.is_empty()) { @@ -1679,6 +1717,38 @@ mod tests { use crate::tools::write_stdin::WriteStdin; use serde_json::json; + #[tokio::test] + async fn owner_message_wake_zero_wait_completed_snapshot_keeps_final_drain() { + let pending = Arc::new(StdMutex::new(PendingBuffer::new())); + let drain_done = Arc::new(Notify::new()); + let session = ExecSession { + id: 1, + command: "final-drain-fixture".into(), + pid: None, + started_at: Instant::now(), + stdin: TokioMutex::new(None), + pending: pending.clone(), + exit_code: Arc::new(StdMutex::new(Some(0))), + drain_done: drain_done.clone(), + last_activity: StdMutex::new(Instant::now()), + }; + // Spawn is not polled until yield_output reaches its await, so the + // final reader completes after the zero-wait snapshot begins. + let reader = tokio::spawn(async move { + pending + .lock() + .unwrap() + .push_str("final output after process exit"); + drain_done.notify_waiters(); + }); + let (output, exited, truncated) = session.yield_output_with_metadata(0).await; + reader.await.unwrap(); + assert!(exited); + assert!(!truncated); + assert_eq!(output, "final output after process exit"); + assert!(session.take_pending().text.is_empty()); + } + #[test] fn classifies_shells() { assert_eq!(shell_type_of("powershell.exe"), ShellType::PowerShell); @@ -2095,6 +2165,69 @@ exit $exit kill_exec_session(&resident); } + #[tokio::test] + async fn quiet_exit_status_survives_cleanup_reconnect_and_another_command() { + for expected_code in [0, 23] { + let root = tempfile::tempdir().unwrap(); + let config = crate::config::default_config(root.path().to_owned()); + let store = ConversationExecSessionStore::new(); + let identity = ConversationIdentity::from_openai_session(&format!( + "quiet-exit-receipt-{expected_code}" + )) + .unwrap(); + let first_transport = SessionState::new(); + let first_call = store.session_for(&identity, &first_transport); + #[cfg(windows)] + let (shell, command) = ( + "powershell", + format!( + "while (-not (Test-Path -LiteralPath 'release')) {{ Start-Sleep -Milliseconds 10 }}; exit {expected_code}" + ), + ); + #[cfg(not(windows))] + let (shell, command) = ( + "sh", + format!("while [ ! -f release ]; do sleep 0.01; done; exit {expected_code}"), + ); + let resident = + start_exec_session(&first_call, &config, &command, root.path(), Some(shell)) + .unwrap(); + let (output, exited, _) = resident.yield_output_with_metadata(0).await; + assert!(output.is_empty()); + assert!(!exited, "fixture must first return a resident session"); + drop(first_call); + drop(first_transport); + + std::fs::write(root.path().join("release"), b"").unwrap(); + for _ in 0..500 { + if resident.exit_code().is_some() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert_eq!(resident.exit_code(), Some(expected_code)); + assert!(resident.pending.lock().unwrap().is_empty()); + // Simulate both background cleanup and a fresh transport while the + // caller has yet to receive even this silent process's exit code. + assert!(reap_conversation_states(&store.states, Duration::from_secs(3600)).is_empty()); + let next_transport = SessionState::new(); + let next_call = store.session_for(&identity, &next_transport); + let other = ExecCommand + .call(json!({"cmd":"exit 0", "shell":shell}), &config, &next_call) + .await; + assert!(!other.is_error, "{}", other.joined_text()); + let final_result = WriteStdin + .call(json!({"session_id":resident.id}), &config, &next_call) + .await; + assert_eq!(final_result.is_error, expected_code != 0); + let result = final_result.structured_content.unwrap(); + assert_eq!(result["exit_code"], expected_code); + assert!(result.get("session_id").is_none()); + assert_eq!(result["output"], ""); + assert!(next_call.exec_session(resident.id).is_none()); + } + } + #[test] fn conversation_exec_states_are_isolated_and_empty_states_are_reclaimed() { let store = ConversationExecSessionStore::new(); @@ -2147,6 +2280,46 @@ exit $exit assert_eq!(state.exec_session_count(), 0); } + #[tokio::test] + async fn unread_finished_sessions_do_not_block_admission_but_running_ones_do() { + let dir = std::env::temp_dir(); + let mut config = crate::config::default_config(dir.clone()); + config.exec.max_sessions = 2; + let state = SessionState::new(); + let finished_first = start_exec_session(&state, &config, "exit 3", &dir, None).unwrap(); + let finished_second = start_exec_session(&state, &config, "exit 4", &dir, None).unwrap(); + for session in [&finished_first, &finished_second] { + for _ in 0..400 { + if session.exit_code().is_some() { + break; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + assert!(session.exit_code().is_some(), "fixture must finish"); + } + finished_second.touch(); + assert_eq!(state.exec_session_count(), 2); + + // The map is full of unread results: the least recently used one gives + // way and the other stays readable. + let running = + start_exec_session(&state, &config, &resident_sleep_command(), &dir, None).unwrap(); + assert!(state.exec_session(finished_first.id).is_none()); + assert!(state.exec_session(finished_second.id).is_some()); + let second_running = + start_exec_session(&state, &config, &resident_sleep_command(), &dir, None).unwrap(); + assert!(state.exec_session(finished_second.id).is_none()); + + // Every slot now holds a live process, so admission is refused. + let refused = start_exec_session(&state, &config, &resident_sleep_command(), &dir, None); + let Err(error) = refused else { + panic!("a third running command must be refused") + }; + assert!(error.contains("Too many live exec sessions")); + kill_exec_session(&running); + kill_exec_session(&second_running); + } + #[tokio::test] async fn touch_resets_the_idle_clock() { let dir = std::env::temp_dir(); diff --git a/src/markdown_chat/wait.rs b/src/markdown_chat/wait.rs index 776b6d5..f97eff4 100644 --- a/src/markdown_chat/wait.rs +++ b/src/markdown_chat/wait.rs @@ -59,6 +59,89 @@ pub enum WaitOutcome { } impl ChatFile { + /// Wake a running tool for unread owner/team text without acknowledging it, + /// entering input-required mode, or releasing queued messages. + pub(crate) async fn wait_for_unread( + self: &Arc, + cancellation: CancellationToken, + ) -> Result<(), String> { + self.wait_for_unread_with_watcher(cancellation, true).await + } + + async fn wait_for_unread_with_watcher( + self: &Arc, + cancellation: CancellationToken, + native: bool, + ) -> Result<(), String> { + if cancellation.is_cancelled() { + return Err("Owner message observation cancelled".into()); + } + let unread = self.unread_snapshot().await?; + if cancellation.is_cancelled() { + return Err("Owner message observation cancelled".into()); + } + if !unread.messages.is_empty() { + return Ok(()); + } + let (sender, mut events) = mpsc::channel(1); + let path = std::fs::canonicalize(self.path()).unwrap_or_else(|_| self.path().to_path_buf()); + let mut watcher = if native { + notify::recommended_watcher(move |result: notify::Result| match result { + Ok(event) + if !matches!(event.kind, EventKind::Access(_)) + && (event.need_rescan() + || event.paths.iter().any(|changed| changed == &path)) => + { + let _ = sender.try_send(false); + } + Err(_) => { + let _ = sender.try_send(true); + } + _ => {} + }) + .ok() + } else { + None + }; + if let Some(watching) = watcher.as_mut() + && watching + .watch( + self.path().parent().ok_or("CHAT.md has no parent")?, + RecursiveMode::NonRecursive, + ) + .is_err() + { + watcher = None; + } + loop { + if cancellation.is_cancelled() { + return Err("Owner message observation cancelled".into()); + } + // The post-registration check closes the append/watch race. This + // snapshot also leaves the cursor and agent-wait state untouched. + let unread = self.unread_snapshot().await?; + if cancellation.is_cancelled() { + return Err("Owner message observation cancelled".into()); + } + if !unread.messages.is_empty() { + return Ok(()); + } + let poll = if watcher.is_some() { + Duration::from_secs(2) + } else { + Duration::from_millis(500) + }; + tokio::select! { + biased; + _ = cancellation.cancelled() => return Err("Owner message observation cancelled".into()), + event = events.recv(), if watcher.is_some() => { + if event != Some(false) { watcher = None; } + } + _ = tokio::time::sleep(poll) => {} + } + } + } + #[cfg(test)] fn begin_agent_wait(&self, remaining: Duration) -> WaitGuard<'_> { self.begin_wait(remaining, true) @@ -237,6 +320,84 @@ mod tests { use super::*; use std::io::Write; + #[tokio::test] + async fn owner_message_wake_observation_preserves_cursor_and_wait_state() { + let root = tempfile::tempdir().unwrap(); + let chat = Arc::new(ChatFile::new(root.path().join("chat/CHAT.md"), true)); + chat.ensure().await.unwrap(); + chat.append_user("pending-owner".into(), "Do this next".into()) + .await + .unwrap(); + let before = chat.unread_snapshot().await.unwrap(); + let _waiting = chat.begin_agent_wait(Duration::from_secs(30)); + let wait_until = chat.agent_waiting_until_ms(); + tokio::time::timeout( + Duration::from_secs(1), + chat.wait_for_unread(CancellationToken::new()), + ) + .await + .unwrap() + .unwrap(); + assert_eq!(chat.unread_snapshot().await.unwrap(), before); + assert_eq!( + chat.agent_waiting_until_ms(), + wait_until, + "observation must not enter or clear an input wait" + ); + } + + #[tokio::test] + async fn owner_message_wake_fallback_observes_external_append_and_cancels_without_consuming() { + let root = tempfile::tempdir().unwrap(); + let chat = Arc::new(ChatFile::new(root.path().join("chat/CHAT.md"), true)); + chat.ensure().await.unwrap(); + let waiting = tokio::spawn({ + let chat = chat.clone(); + async move { + chat.wait_for_unread_with_watcher(CancellationToken::new(), false) + .await + } + }); + tokio::time::sleep(Duration::from_millis(50)).await; + std::fs::OpenOptions::new() + .append(true) + .open(chat.path()) + .unwrap() + .write_all(b"Fallback owner message\n") + .unwrap(); + tokio::time::timeout(Duration::from_secs(2), waiting) + .await + .unwrap() + .unwrap() + .unwrap(); + assert!(chat.agent_waiting_until_ms().is_none()); + assert!( + chat.read(true) + .await + .unwrap() + .text + .contains("Fallback owner message") + ); + let before = chat.unread_snapshot().await.unwrap(); + let cancellation = CancellationToken::new(); + let observing = tokio::spawn({ + let chat = chat.clone(); + let cancellation = cancellation.clone(); + async move { chat.wait_for_unread(cancellation).await } + }); + tokio::time::sleep(Duration::from_millis(50)).await; + cancellation.cancel(); + assert!( + tokio::time::timeout(Duration::from_secs(1), observing) + .await + .unwrap() + .unwrap() + .is_err() + ); + assert_eq!(chat.unread_snapshot().await.unwrap(), before); + assert!(chat.agent_waiting_until_ms().is_none()); + } + #[test] fn waiting_deadline_tracks_the_actual_wait_duration() { let root = tempfile::tempdir().unwrap(); diff --git a/src/project_clone.rs b/src/project_clone.rs index a2813dc..c300d57 100644 --- a/src/project_clone.rs +++ b/src/project_clone.rs @@ -1002,8 +1002,10 @@ async fn clone_repository_from( command .arg("--") .arg(clone_source) - .arg(&temporary) - .current_dir(clone_dir) + .arg(crate::worktrees::strip_verbatim_prefix(temporary.clone())) + .current_dir(crate::worktrees::strip_verbatim_prefix( + clone_dir.to_path_buf(), + )) .env("GIT_TERMINAL_PROMPT", "0") .env("GCM_INTERACTIVE", "Never") .env("GIT_OPTIONAL_LOCKS", "0") @@ -1498,7 +1500,7 @@ async fn git_output( command.args(args); } command - .current_dir(cwd) + .current_dir(crate::worktrees::strip_verbatim_prefix(cwd.to_path_buf())) .env("GIT_TERMINAL_PROMPT", "0") .env("GCM_INTERACTIVE", "Never") .env("GIT_OPTIONAL_LOCKS", "0") diff --git a/src/service.rs b/src/service.rs index 4ad709c..fbea5d2 100644 --- a/src/service.rs +++ b/src/service.rs @@ -1026,7 +1026,7 @@ fn windows_install_script(spec: &ServiceSpec) -> String { let working_dir = powershell_quote(&spec.working_dir.to_string_lossy()); let task = powershell_quote(WINDOWS_TASK); format!( - "$ErrorActionPreference = 'Stop'\n$taskName = {task}\n$existing = Get-ScheduledTask -TaskName $taskName -ErrorAction SilentlyContinue\nif ($existing) {{ Stop-ScheduledTask -InputObject $existing -ErrorAction SilentlyContinue }}\n$identity = [System.Security.Principal.WindowsIdentity]::GetCurrent().Name\n$engine = (Get-Process -Id $PID).Path\n$action = New-ScheduledTaskAction -Execute $engine -Argument {arguments} -WorkingDirectory {working_dir}\n$trigger = New-ScheduledTaskTrigger -AtLogOn -User $identity\n$settings = New-ScheduledTaskSettingsSet -AllowStartIfOnBatteries -DontStopIfGoingOnBatteries -StartWhenAvailable -RestartCount 999 -RestartInterval (New-TimeSpan -Minutes 1) -ExecutionTimeLimit ([TimeSpan]::Zero) -MultipleInstances IgnoreNew\n$principal = New-ScheduledTaskPrincipal -UserId $identity -LogonType Interactive -RunLevel Limited\nRegister-ScheduledTask -TaskName $taskName -Action $action -Trigger $trigger -Settings $settings -Principal $principal -Force | Out-Null\nEnable-ScheduledTask -TaskName $taskName | Out-Null\nStart-ScheduledTask -TaskName $taskName\n" + "$ErrorActionPreference = 'Stop'\n$taskName = {task}\n$existing = Get-ScheduledTask -TaskName $taskName -ErrorAction SilentlyContinue\nif ($existing) {{ Stop-ScheduledTask -InputObject $existing -ErrorAction SilentlyContinue }}\n$identity = [System.Security.Principal.WindowsIdentity]::GetCurrent().Name\n$engine = (Get-Process -Id $PID).Path\n$action = New-ScheduledTaskAction -Execute $engine -Argument {arguments} -WorkingDirectory {working_dir}\n$trigger = New-ScheduledTaskTrigger -AtLogOn -User $identity\n$settings = New-ScheduledTaskSettingsSet -Priority 4 -AllowStartIfOnBatteries -DontStopIfGoingOnBatteries -StartWhenAvailable -RestartCount 999 -RestartInterval (New-TimeSpan -Minutes 1) -ExecutionTimeLimit ([TimeSpan]::Zero) -MultipleInstances IgnoreNew\n$principal = New-ScheduledTaskPrincipal -UserId $identity -LogonType Interactive -RunLevel Limited\nRegister-ScheduledTask -TaskName $taskName -Action $action -Trigger $trigger -Settings $settings -Principal $principal -Force | Out-Null\nEnable-ScheduledTask -TaskName $taskName | Out-Null\nStart-ScheduledTask -TaskName $taskName\n" ) } @@ -1903,12 +1903,15 @@ fn taskkill(pid: u32, force: bool) { #[cfg(windows)] fn windows_server_creation_flags() -> u32 { - windows_sys::Win32::System::Threading::CREATE_NO_WINDOW + // Task Scheduler can start the supervisor BelowNormal. Give the native + // server Normal priority explicitly instead of inheriting that class. + use windows_sys::Win32::System::Threading::{CREATE_NO_WINDOW, NORMAL_PRIORITY_CLASS}; + CREATE_NO_WINDOW | NORMAL_PRIORITY_CLASS } #[cfg(all(test, not(windows)))] fn windows_server_creation_flags() -> u32 { - 0x0800_0000 + 0x0800_0020 } #[cfg(windows)] @@ -2026,19 +2029,44 @@ async fn shutdown_signal() { std::future::pending::<()>().await; } +fn acquire_supervisor_lock(path: &Path) -> anyhow::Result { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent).context("create the service lock directory")?; + } + let mut options = OpenOptions::new(); + options.create(true).truncate(false).read(true).write(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + let lock = options + .open(path) + .context("open the service supervisor lock")?; + lock.try_lock() + .context("another Codexify service supervisor is already running")?; + Ok(lock) +} + pub async fn run_supervisor(config: PathBuf) -> anyhow::Result<()> { if !config.is_absolute() { bail!("service supervisor requires an absolute --config path"); } let config = normalize_native_path(config); validate_native_path(&config)?; + // Keep the OS lock for this process's lifetime. Native task restarts can + // leave an older supervisor alive; it must never race the new supervisor + // to launch children against the same ports or rotate the shared log. + let service_log_path = log_path()?; + let _supervisor_lock = + acquire_supervisor_lock(&service_log_path.with_extension("supervisor.lock"))?; let executable = std::env::current_exe() .context("locate the running codexify executable")? .canonicalize() .context("canonicalize the running codexify executable")?; let executable = normalize_native_path(executable); let log = Arc::new(StdMutex::new(RotatingLog::open( - log_path()?, + service_log_path, LOG_MAX_BYTES, LOG_GENERATIONS, )?)); @@ -2362,6 +2390,22 @@ pub async fn print_logs(follow: bool) -> anyhow::Result<()> { mod tests { use super::*; + #[test] + fn supervisor_lock_excludes_duplicates_and_survives_stale_files() { + let root = tempfile::tempdir().unwrap(); + let path = root.path().join("logs/codexify.supervisor.lock"); + let first = acquire_supervisor_lock(&path).unwrap(); + let error = acquire_supervisor_lock(&path).unwrap_err(); + assert!(error.to_string().contains("already running")); + drop(first); + // An exited or killed supervisor leaves the file but releases the OS + // lock. Recovery must not require deleting a stale PID/lock file. + assert!(path.exists()); + let recovered = acquire_supervisor_lock(&path).unwrap(); + assert!(acquire_supervisor_lock(&path).is_err()); + drop(recovered); + } + fn spec(root: &Path) -> ServiceSpec { ServiceSpec { executable: root.join("bin/codexify"), @@ -2522,6 +2566,7 @@ mod tests { let script = windows_install_script(&spec); assert!(script.contains("New-ScheduledTaskTrigger -AtLogOn")); + assert!(script.contains("New-ScheduledTaskSettingsSet -Priority 4 ")); assert!(script.contains("-RestartCount 999")); assert!(script.contains("New-TimeSpan -Minutes 1")); assert!(script.contains("-WindowStyle Hidden")); @@ -2535,7 +2580,7 @@ mod tests { let launcher = windows_hidden_service_script(&spec); assert!(launcher.contains(&powershell_quote(&spec.executable.to_string_lossy()))); assert!(launcher.contains(&powershell_quote(&spec.config.to_string_lossy()))); - assert_eq!(windows_server_creation_flags(), 0x0800_0000); + assert_eq!(windows_server_creation_flags(), 0x0800_0020); } #[test] diff --git a/src/tools/exec_command.rs b/src/tools/exec_command.rs index 2df50b3..909f418 100644 --- a/src/tools/exec_command.rs +++ b/src/tools/exec_command.rs @@ -1,15 +1,17 @@ use async_trait::async_trait; use serde::Deserialize; use serde_json::{Value, json}; +use std::sync::Arc; +use tokio_util::sync::CancellationToken; use crate::exec_sessions::{ DEFAULT_MAX_OUTPUT_TOKENS, EXEC_DEFAULT_YIELD_MS, EXEC_MAX_YIELD_MS, EXEC_MIN_YIELD_MS, - SessionState, ShellType, UnifiedExecOutput, clamp, generate_chunk_id, resolve_shell, - shell_type_of, start_exec_session, truncate_output, + ExecSession, SessionState, ShellType, UnifiedExecOutput, clamp, generate_chunk_id, + resolve_shell, shell_type_of, start_exec_session, truncate_output, }; use crate::output_budget::resolve_requested_output_tokens; use crate::safe_path::resolve_safe_path; -use crate::tool::{Tool, ToolBehavior, parse_tool_args}; +use crate::tool::{Tool, ToolBehavior, ToolRequestContext, parse_tool_args}; use crate::types::{AppConfig, ToolAuditMetadata, ToolResult}; /// The output schema shared by `exec_command` and `write_stdin`. @@ -84,6 +86,58 @@ pub fn describe_exec_command(config: &AppConfig) -> String { pub struct ExecCommand; +pub(super) fn command_chat( + config: &AppConfig, + session: &SessionState, + context: Option<&ToolRequestContext>, +) -> Option> { + if !config.markdown_chat.enabled { + return None; + } + let context = context?; + match context + .markdown_chat + .chat(config, context.task_conversation.as_ref(), session) + { + Ok(chat) => Some(chat), + Err(error) => { + tracing::debug!(%error, "owner message observation unavailable during command output wait"); + None + } + } +} + +pub(super) async fn wait_for_command_output( + exec_session: &ExecSession, + yield_ms: u64, + chat: Option<&Arc>, + cancellation: &CancellationToken, +) -> Result<(String, bool, bool), String> { + let output = crate::audit::stage( + "command_output_wait", + exec_session.yield_output_with_metadata(yield_ms), + ); + tokio::pin!(output); + if let Some(chat) = chat { + tokio::select! { + biased; + _ = cancellation.cancelled() => return Err("Output wait cancelled; the command session remains available.".into()), + message = chat.wait_for_unread(cancellation.clone()) => match message { + // End only the output wait. Draining does not signal the child, + // resend stdin, consume owner text, or change the session ID. + Ok(()) => return Ok(exec_session.yield_output_with_metadata(0).await), + Err(error) => tracing::debug!(%error, "owner message observation failed; retaining command output wait"), + }, + result = &mut output => return Ok(result), + } + } + tokio::select! { + biased; + _ = cancellation.cancelled() => Err("Output wait cancelled; the command session remains available.".into()), + result = &mut output => Ok(result), + } +} + #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] struct ExecCommandArgs { @@ -156,6 +210,31 @@ impl Tool for ExecCommand { } async fn call(&self, args: Value, config: &AppConfig, session: &SessionState) -> ToolResult { + self.run(args, config, session, None).await + } + + async fn call_with_context( + &self, + args: Value, + config: &AppConfig, + session: &SessionState, + context: &ToolRequestContext, + ) -> ToolResult { + self.run(args, config, session, Some(context)).await + } +} + +impl ExecCommand { + async fn run( + &self, + args: Value, + config: &AppConfig, + session: &SessionState, + context: Option<&ToolRequestContext>, + ) -> ToolResult { + if context.is_some_and(|context| context.cancellation.is_cancelled()) { + return ToolResult::error("Call cancelled before the command was started."); + } let ExecCommandArgs { cmd, workdir, @@ -204,8 +283,19 @@ impl Tool for ExecCommand { Err(e) => return ToolResult::error(e), }; + let chat = command_chat(config, session, context); + let cancellation = context + .map(|context| context.cancellation.clone()) + .unwrap_or_default(); let (output, exited, buffer_truncated) = - exec_session.yield_output_with_metadata(yield_ms).await; + match wait_for_command_output(&exec_session, yield_ms, chat.as_ref(), &cancellation) + .await + { + Ok(output) => output, + Err(error) => { + return ToolResult::error(format!("{error} Session id: {}", exec_session.id)); + } + }; let (text, original_token_count, truncated) = truncate_output(&output, max_output_tokens); let mut result = UnifiedExecOutput { @@ -253,7 +343,365 @@ mod tests { use super::*; use crate::config::default_config; use crate::output_budget::approx_token_count; + use crate::tools::write_stdin::WriteStdin; use serde_json::json; + use std::time::Duration; + + struct OwnerWakeFixture { + root: tempfile::TempDir, + config: AppConfig, + session: Arc, + context: ToolRequestContext, + chat: Arc, + } + + async fn owner_wake_fixture() -> OwnerWakeFixture { + let root = tempfile::tempdir().unwrap(); + let mut config = default_config(root.path().canonicalize().unwrap()); + config.memory.dir = Some(root.path().join("metadata").display().to_string()); + config.markdown_chat.enabled = true; + config.exec.default_shell = Some(if cfg!(windows) { "powershell" } else { "sh" }.into()); + let identity = crate::project_bindings::ConversationIdentity::from_openai_session( + "command-owner-wake", + ); + let context = ToolRequestContext { + conversation: identity.clone(), + task_conversation: identity, + task_owner_generation: Some(0), + conversation_retired: false, + connector_schema_version: None, + conversation_schema_version: None, + conversation_authorizations: Arc::new( + crate::conversation_auth::ConversationAuthorizationStore::new(), + ), + project_bindings: Arc::new(crate::project_bindings::ProjectBindingStore::new( + root.path().join("bindings"), + )), + diff_checkpoints: Arc::new(crate::diff::DiffCheckpointManager::new()), + artifact_egress: Arc::new(crate::artifact_egress::ArtifactEgressStore::new_at( + config.artifact_egress.clone(), + root.path().join("artifacts"), + )), + markdown_chat: Arc::default(), + agent_automation: None, + cancellation: CancellationToken::new(), + }; + let session = Arc::new(SessionState::new()); + let chat = context + .markdown_chat + .chat(&config, context.task_conversation.as_ref(), &session) + .unwrap(); + chat.ensure().await.unwrap(); + OwnerWakeFixture { + root, + config, + session, + context, + chat, + } + } + + fn finish_on_input_command() -> &'static str { + if cfg!(windows) { + "$line=[Console]::ReadLine(); [Console]::WriteLine('finished:'+ $line)" + } else { + "IFS= read -r line; printf 'finished:%s\\n' \"$line\"" + } + } + + async fn wait_for_test_file(path: &std::path::Path) { + tokio::time::timeout(Duration::from_secs(8), async { + while !path.is_file() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + } + + fn spawn_poll( + fixture: &OwnerWakeFixture, + id: u64, + chars: &str, + ) -> tokio::task::JoinHandle { + let config = fixture.config.clone(); + let session = fixture.session.clone(); + let context = fixture.context.clone(); + let chars = chars.to_string(); + tokio::spawn(async move { + WriteStdin + .call_with_context( + json!({"session_id":id,"chars":chars,"yield_time_ms":300_000}), + &config, + &session, + &context, + ) + .await + }) + } + + #[tokio::test] + async fn owner_message_wake_long_poll_preserves_process_cursor_and_task_isolation() { + let fixture = owner_wake_fixture().await; + fixture + .chat + .append_user("already-read".into(), "Old acknowledged input".into()) + .await + .unwrap(); + fixture.chat.read(true).await.unwrap(); + let before = fixture.chat.unread_snapshot().await.unwrap(); + let exec = start_exec_session( + &fixture.session, + &fixture.config, + finish_on_input_command(), + fixture.root.path(), + None, + ) + .unwrap(); + let mut polling = spawn_poll(&fixture, exec.id, ""); + assert!( + tokio::time::timeout(Duration::from_millis(100), &mut polling) + .await + .is_err(), + "already-read text must not end the wait" + ); + let other = crate::project_bindings::ConversationIdentity::from_openai_session( + "unrelated-command-owner", + ) + .unwrap(); + let other_chat = fixture + .context + .markdown_chat + .chat(&fixture.config, Some(&other), &fixture.session) + .unwrap(); + other_chat + .append_user("other-owner".into(), "Do not wake a different task".into()) + .await + .unwrap(); + assert!( + tokio::time::timeout(Duration::from_millis(150), &mut polling) + .await + .is_err(), + "another task's message must not wake this command" + ); + fixture + .chat + .append_user("new-owner".into(), "Read this instruction now".into()) + .await + .unwrap(); + let result = tokio::time::timeout(Duration::from_secs(4), polling) + .await + .unwrap() + .unwrap(); + assert!(!result.is_error, "{}", result.joined_text()); + assert_eq!( + result.structured_content.as_ref().unwrap()["session_id"], + exec.id + ); + assert!(exec.exit_code().is_none()); + assert!(fixture.session.exec_session(exec.id).is_some()); + let pending = fixture.chat.unread_snapshot().await.unwrap(); + assert_eq!(pending.read_through, before.read_through); + assert_eq!(pending.messages.len(), 1); + assert_eq!( + other_chat.unread_snapshot().await.unwrap().messages.len(), + 1 + ); + fixture.chat.read(true).await.unwrap(); + let finished = WriteStdin + .call_with_context( + json!({"session_id":exec.id,"chars":"finish\n","yield_time_ms":1500}), + &fixture.config, + &fixture.session, + &fixture.context, + ) + .await; + assert_eq!( + finished.structured_content.as_ref().unwrap()["exit_code"], + 0 + ); + assert!(finished.joined_text().contains("finished:finish")); + } + + #[tokio::test] + async fn owner_message_wake_after_stdin_never_resends_input_or_loses_late_output() { + let fixture = owner_wake_fixture().await; + let command = if cfg!(windows) { + "$first=[Console]::ReadLine(); [IO.File]::AppendAllText('stdin-events.txt',$first+\"`n\"); [Console]::WriteLine('first:'+ $first); $second=[Console]::ReadLine(); [IO.File]::AppendAllText('stdin-events.txt',$second+\"`n\"); [Console]::WriteLine('second:'+ $second)" + } else { + "IFS= read -r first; printf '%s\\n' \"$first\" >> stdin-events.txt; printf 'first:%s\\n' \"$first\"; IFS= read -r second; printf '%s\\n' \"$second\" >> stdin-events.txt; printf 'second:%s\\n' \"$second\"" + }; + let exec = start_exec_session( + &fixture.session, + &fixture.config, + command, + fixture.root.path(), + None, + ) + .unwrap(); + let input = spawn_poll(&fixture, exec.id, "only-once\n"); + let events = fixture.root.path().join("stdin-events.txt"); + wait_for_test_file(&events).await; + fixture + .chat + .append_user( + "stdin-owner".into(), + "New work while input waits for output".into(), + ) + .await + .unwrap(); + let result = tokio::time::timeout(Duration::from_secs(4), input) + .await + .unwrap() + .unwrap(); + assert!(!result.is_error, "{}", result.joined_text()); + assert_eq!( + result.structured_content.as_ref().unwrap()["session_id"], + exec.id + ); + assert_eq!( + std::fs::read_to_string(&events) + .unwrap() + .lines() + .collect::>(), + ["only-once"] + ); + assert!(exec.exit_code().is_none()); + fixture.chat.read(true).await.unwrap(); + let finished = WriteStdin + .call_with_context( + json!({"session_id":exec.id,"chars":"finish\n","yield_time_ms":1500}), + &fixture.config, + &fixture.session, + &fixture.context, + ) + .await; + assert_eq!( + finished.structured_content.as_ref().unwrap()["exit_code"], + 0 + ); + assert!(finished.joined_text().contains("second:finish")); + let combined = format!("{}{}", result.joined_text(), finished.joined_text()); + assert_eq!( + combined.matches("first:only-once").count(), + 1, + "early and late output must be delivered exactly once" + ); + assert_eq!( + std::fs::read_to_string(&events) + .unwrap() + .lines() + .collect::>(), + ["only-once", "finish"] + ); + } + + #[tokio::test] + async fn owner_message_wake_initial_exec_returns_its_registered_session() { + let fixture = owner_wake_fixture().await; + let config = fixture.config.clone(); + let session = fixture.session.clone(); + let context = fixture.context.clone(); + let running = tokio::spawn(async move { + ExecCommand + .call_with_context( + json!({"cmd":finish_on_input_command(),"yield_time_ms":30_000}), + &config, + &session, + &context, + ) + .await + }); + let id = tokio::time::timeout(Duration::from_secs(8), async { + loop { + if let Some(id) = fixture.session.exec_session_ids().first() { + break *id; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + fixture + .chat + .append_user( + "initial-owner".into(), + "Handle this before waiting further".into(), + ) + .await + .unwrap(); + let result = tokio::time::timeout(Duration::from_secs(4), running) + .await + .unwrap() + .unwrap(); + assert!(!result.is_error, "{}", result.joined_text()); + assert_eq!( + result.structured_content.as_ref().unwrap()["session_id"], + id + ); + assert!( + fixture + .session + .exec_session(id) + .unwrap() + .exit_code() + .is_none() + ); + assert_eq!( + fixture.chat.unread_snapshot().await.unwrap().messages.len(), + 1 + ); + fixture.chat.read(true).await.unwrap(); + let finished = WriteStdin + .call_with_context( + json!({"session_id":id,"chars":"finish\n","yield_time_ms":5000}), + &fixture.config, + &fixture.session, + &fixture.context, + ) + .await; + assert_eq!( + finished.structured_content.as_ref().unwrap()["exit_code"], + 0 + ); + } + + #[tokio::test] + async fn owner_message_wake_real_cancellation_keeps_session_and_buffer() { + let fixture = owner_wake_fixture().await; + let command = if cfg!(windows) { + "[Console]::WriteLine('buffered-before-cancel'); [IO.File]::WriteAllText('cancellation-ready','ready'); $line=[Console]::ReadLine(); [Console]::WriteLine('after-cancel')" + } else { + "printf 'buffered-before-cancel\\n'; printf ready > cancellation-ready; IFS= read -r line; printf 'after-cancel\\n'" + }; + let exec = start_exec_session( + &fixture.session, + &fixture.config, + command, + fixture.root.path(), + None, + ) + .unwrap(); + wait_for_test_file(&fixture.root.path().join("cancellation-ready")).await; + let polling = spawn_poll(&fixture, exec.id, ""); + tokio::time::sleep(Duration::from_millis(50)).await; + fixture.context.cancellation.cancel(); + let result = tokio::time::timeout(Duration::from_secs(2), polling) + .await + .unwrap() + .unwrap(); + assert!(result.is_error); + assert!(result.joined_text().contains("Output wait cancelled")); + assert!(exec.exit_code().is_none()); + assert!(fixture.session.exec_session(exec.id).is_some()); + let (buffer, exited, _) = exec.yield_output_with_metadata(0).await; + assert!(!exited); + assert!(buffer.contains("buffered-before-cancel")); + exec.write_stdin("finish\n").await.unwrap(); + let (late, exited, _) = exec.yield_output_with_metadata(1500).await; + assert!(exited); + assert!(late.contains("after-cancel")); + } fn large_output_command() -> &'static str { if cfg!(windows) { diff --git a/src/tools/write_stdin.rs b/src/tools/write_stdin.rs index cba510d..96b58e9 100644 --- a/src/tools/write_stdin.rs +++ b/src/tools/write_stdin.rs @@ -10,7 +10,9 @@ use crate::exec_sessions::{ }; use crate::output_budget::resolve_requested_output_tokens; use crate::tool::{Tool, ToolBehavior, ToolRequestContext, parse_tool_args}; -use crate::tools::exec_command::{render_unified_exec_output, unified_exec_output_schema}; +use crate::tools::exec_command::{ + command_chat, render_unified_exec_output, unified_exec_output_schema, wait_for_command_output, +}; use crate::types::{AppConfig, ToolAuditMetadata, ToolContent, ToolResult}; pub struct WriteStdin; @@ -203,11 +205,14 @@ impl WriteStdin { // Checkpoint the input action, not the following wait for output. } - let (output, exited, buffer_truncated) = tokio::select! { - biased; - _ = cancellation.cancelled() => return ToolResult::error("Output wait cancelled; the command session remains available."), - output = crate::audit::stage("command_output_wait", exec_session.yield_output_with_metadata(yield_ms)) => output, - }; + let chat = command_chat(config, session, context); + let (output, exited, buffer_truncated) = + match wait_for_command_output(&exec_session, yield_ms, chat.as_ref(), &cancellation) + .await + { + Ok(output) => output, + Err(error) => return ToolResult::error(error), + }; let (text, original_token_count, truncated) = truncate_output(&output, max_output_tokens); let mut result = UnifiedExecOutput { diff --git a/src/worktrees.rs b/src/worktrees.rs index 001fbaa..2f1ebfc 100644 --- a/src/worktrees.rs +++ b/src/worktrees.rs @@ -523,7 +523,7 @@ fn prepare_worktrees_root(configured: &Path) -> Result { /// plain `C:\…` before any of these paths reach git. Only the drive-letter /// form is unwrapped; UNC and other prefixes, and every non-Windows path, are /// returned unchanged. -fn strip_verbatim_prefix(path: PathBuf) -> PathBuf { +pub(crate) fn strip_verbatim_prefix(path: PathBuf) -> PathBuf { #[cfg(windows)] { use std::path::{Component, Prefix}; diff --git a/tests/worktree_selection.rs b/tests/worktree_selection.rs index 0cca879..b579da1 100644 --- a/tests/worktree_selection.rs +++ b/tests/worktree_selection.rs @@ -262,6 +262,7 @@ async fn targeted_github_urls_use_isolated_worktrees_without_moving_the_source() fs::create_dir_all(&access_root).unwrap(); git(&upstream, &["init", "--quiet"]); + git(&upstream, &["config", "core.autocrlf", "false"]); git(&upstream, &["config", "user.name", "Codexify Tests"]); git( &upstream, @@ -312,6 +313,7 @@ async fn targeted_github_urls_use_isolated_worktrees_without_moving_the_source() ], ); git(&project_root, &["config", "protocol.file.allow", "always"]); + git(&project_root, &["config", "core.autocrlf", "false"]); let mut auto = config(&root, access_root.clone(), WorktreeMode::Auto); auto.project_catalog.codex_config.enabled = false; @@ -352,7 +354,10 @@ async fn targeted_github_urls_use_isolated_worktrees_without_moving_the_source() .await .unwrap(); assert!(!repeated.newly_selected); - assert_eq!(repeated.project_root, selection.project_root); + assert_eq!( + repeated.project_root, + fs::canonicalize(&selection.project_root).unwrap() + ); let fetched_refs_before_switch = git( &project_root, @@ -431,7 +436,10 @@ async fn targeted_github_urls_use_isolated_worktrees_without_moving_the_source() .await .unwrap(); assert!(!repeated_commit.newly_selected); - assert_eq!(repeated_commit.project_root, commit_selection.project_root); + assert_eq!( + repeated_commit.project_root, + fs::canonicalize(&commit_selection.project_root).unwrap() + ); let mut never = auto.clone(); never.worktrees.mode = WorktreeMode::Never; @@ -646,7 +654,7 @@ async fn new_worktrees_use_the_selected_base_commit_and_keep_authorized_source_m let reopened = ProjectBindingStore::new(root.path().join("bindings")); assert_eq!( reopened.effective_config(&config, &task).unwrap().work_dir, - selected.project_root + fs::canonicalize(&selected.project_root).unwrap() ); let retry = if explicit { reopened @@ -664,7 +672,8 @@ async fn new_worktrees_use_the_selected_base_commit_and_keep_authorized_source_m } .unwrap(); assert_eq!( - retry.project_root, selected.project_root, + retry.project_root, + fs::canonicalize(&selected.project_root).unwrap(), "retrying creation must reuse its saved workspace" ); }