From 44bd419a08a4a8898fe70711363238b90f538c67 Mon Sep 17 00:00:00 2001
From: Maxxx-step1121 <129008351+Maxxx-step1121@users.noreply.github.com>
Date: Fri, 13 Nov 2026 01:27:41 +0400
Subject: [PATCH 1/3] Keep command output and Git checkpoints reliable on
 Windows

exec_command keeps a session addressable until its final output and exit code
are polled, retains final results until read, and lets an owner message end
only the output wait. Git checkpoints skip repeated legacy-namespace probes and
drain stderr under the command deadline. Paths drop the \?\ verbatim prefix
before reaching Git, the native server runs at Normal priority, and the
supervisor holds an OS lock for its lifetime.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/REFERENCE.md           |   5 +
 src/diff.rs                 | 153 ++++++++++--
 src/exec_sessions.rs        | 180 ++++++++++++--
 src/markdown_chat/wait.rs   | 161 +++++++++++++
 src/project_clone.rs        |   8 +-
 src/service.rs              |  55 ++++-
 src/tools/exec_command.rs   | 456 +++++++++++++++++++++++++++++++++++-
 src/tools/write_stdin.rs    |  17 +-
 src/worktrees.rs            |   2 +-
 tests/worktree_selection.rs |  17 +-
 10 files changed, 997 insertions(+), 57 deletions(-)

diff --git a/docs/REFERENCE.md b/docs/REFERENCE.md
index 4ca5cb23..b73e0ad8 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 b48fc3eb..18579438 100644
--- a/src/diff.rs
+++ b/src/diff.rs
@@ -634,9 +634,20 @@ async fn load_or_initialize_persistent(
     legacy_owner_key: &str,
 ) -> Result<CheckpointPair, String> {
     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<OsString> {
     values.iter().map(OsString::from).collect()
 }
 
+#[cfg(test)]
+tokio::task_local! {
+    static TEST_GIT_COMMAND_COUNT: std::cell::Cell<usize>;
+}
+
 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<R: tokio::io::AsyncRead + Unpin>(
+    mut reader: R,
+) -> std::io::Result<Vec<u8>> {
+    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 9d795025..e8ab76d6 100644
--- a/src/exec_sessions.rs
+++ b/src/exec_sessions.rs
@@ -8,7 +8,7 @@
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
-use std::sync::{Arc, Mutex as StdMutex};
+use std::sync::{Arc, Mutex as StdMutex, OnceLock, Weak};
 use std::time::{Duration, Instant};
 
 use serde::Serialize;
@@ -488,7 +488,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 +497,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)
     }
 }
 
@@ -593,6 +596,33 @@ impl Drop for ExecSessionState {
 }
 
 type ConversationExecStates = HashMap<ConversationIdentity, Arc<ExecSessionState>>;
+static CONVERSATION_EXEC_ACTIVITY: OnceLock<
+    StdMutex<HashMap<String, Vec<Weak<ExecSessionState>>>>,
+> = OnceLock::new();
+
+pub(crate) fn conversation_has_running_commands(identity: &ConversationIdentity) -> bool {
+    let Some(registry) = CONVERSATION_EXEC_ACTIVITY.get() else {
+        return false;
+    };
+    let Ok(registry) = registry.try_lock() else {
+        return true;
+    };
+    let states = registry
+        .get(identity.stable_key())
+        .cloned()
+        .unwrap_or_default();
+    drop(registry);
+    states
+        .into_iter()
+        .filter_map(|state| state.upgrade())
+        .any(|state| {
+            state.sessions.try_lock().map_or(true, |sessions| {
+                sessions
+                    .values()
+                    .any(|command| command.exit_code().is_none())
+            })
+        })
+}
 
 /// Server-process ownership for ChatGPT resident commands. ChatGPT can replace
 /// its MCP transport between adjacent tool calls while retaining the same
@@ -622,6 +652,22 @@ impl ConversationExecSessionStore {
             .entry(identity.clone())
             .or_insert_with(|| Arc::new(ExecSessionState::new()))
             .clone();
+        if let Ok(mut registry) = CONVERSATION_EXEC_ACTIVITY
+            .get_or_init(Default::default)
+            .lock()
+        {
+            if registry.len() > 4096 {
+                registry.retain(|_, states| states.iter().any(|state| state.strong_count() > 0));
+            }
+            let states = registry
+                .entry(identity.stable_key().to_owned())
+                .or_default();
+            states.retain(|state| state.strong_count() > 0);
+            let weak = Arc::downgrade(&exec);
+            if !states.iter().any(|state| state.ptr_eq(&weak)) {
+                states.push(weak);
+            }
+        }
         transport_session.with_exec_state(exec)
     }
 
@@ -1108,6 +1154,7 @@ impl SessionState {
         self.exec.len()
     }
 
+    #[cfg(test)]
     fn reap_exec_sessions(&self, idle_timeout: Duration) -> Vec<Arc<ExecSession>> {
         self.exec.reap(idle_timeout)
     }
@@ -1457,21 +1504,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<u64, Arc<ExecSession>>,
     idle_timeout: Duration,
 ) -> Vec<Arc<ExecSession>> {
     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 +1533,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,7 +1547,6 @@ pub fn start_exec_session(
     cwd: &std::path::Path,
     shell: Option<&str>,
 ) -> Result<Arc<ExecSession>, String> {
-    reap_finished_sessions(state);
     {
         let live = state.exec_session_count();
         if live >= config.exec.max_sessions {
@@ -1679,6 +1728,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 +2176,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();
diff --git a/src/markdown_chat/wait.rs b/src/markdown_chat/wait.rs
index 776b6d5a..f97eff43 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<Self>,
+        cancellation: CancellationToken,
+    ) -> Result<(), String> {
+        self.wait_for_unread_with_watcher(cancellation, true).await
+    }
+
+    async fn wait_for_unread_with_watcher(
+        self: &Arc<Self>,
+        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<notify::Event>| 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 a2813dcc..c300d57b 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 4ad709cb..fbea5d2c 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<File> {
+    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 2df50b3c..909f418d 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<Arc<crate::markdown_chat::ChatFile>> {
+    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<crate::markdown_chat::ChatFile>>,
+    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<SessionState>,
+        context: ToolRequestContext,
+        chat: Arc<crate::markdown_chat::ChatFile>,
+    }
+
+    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<ToolResult> {
+        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::<Vec<_>>(),
+            ["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::<Vec<_>>(),
+            ["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 cba510dc..96b58e9f 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 001fbaab..2f1ebfce 100644
--- a/src/worktrees.rs
+++ b/src/worktrees.rs
@@ -523,7 +523,7 @@ fn prepare_worktrees_root(configured: &Path) -> Result<PathBuf, String> {
 /// 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 0cca8790..b579da17 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"
             );
         }

From c319e391b03e2bf303ce222e12fea1644170489d Mon Sep 17 00:00:00 2001
From: Maxxx-step1121 <129008351+Maxxx-step1121@users.noreply.github.com>
Date: Fri, 13 Nov 2026 13:33:09 +0400
Subject: [PATCH 2/3] Keep the standalone execution branch warning-free

Move the conversation command-activity registry to the agent-teams change
that consumes it, and compile PendingBuffer::is_empty only for tests now
that the reaper no longer treats an empty buffer as an acknowledgement.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/exec_sessions.rs | 46 ++------------------------------------------
 1 file changed, 2 insertions(+), 44 deletions(-)

diff --git a/src/exec_sessions.rs b/src/exec_sessions.rs
index e8ab76d6..dbf15711 100644
--- a/src/exec_sessions.rs
+++ b/src/exec_sessions.rs
@@ -8,7 +8,7 @@
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
-use std::sync::{Arc, Mutex as StdMutex, OnceLock, Weak};
+use std::sync::{Arc, Mutex as StdMutex};
 use std::time::{Duration, Instant};
 
 use serde::Serialize;
@@ -386,6 +386,7 @@ impl PendingBuffer {
         }
     }
 
+    #[cfg(test)]
     fn is_empty(&self) -> bool {
         self.head.is_empty() && self.tail.is_empty() && self.omitted == 0
     }
@@ -596,33 +597,6 @@ impl Drop for ExecSessionState {
 }
 
 type ConversationExecStates = HashMap<ConversationIdentity, Arc<ExecSessionState>>;
-static CONVERSATION_EXEC_ACTIVITY: OnceLock<
-    StdMutex<HashMap<String, Vec<Weak<ExecSessionState>>>>,
-> = OnceLock::new();
-
-pub(crate) fn conversation_has_running_commands(identity: &ConversationIdentity) -> bool {
-    let Some(registry) = CONVERSATION_EXEC_ACTIVITY.get() else {
-        return false;
-    };
-    let Ok(registry) = registry.try_lock() else {
-        return true;
-    };
-    let states = registry
-        .get(identity.stable_key())
-        .cloned()
-        .unwrap_or_default();
-    drop(registry);
-    states
-        .into_iter()
-        .filter_map(|state| state.upgrade())
-        .any(|state| {
-            state.sessions.try_lock().map_or(true, |sessions| {
-                sessions
-                    .values()
-                    .any(|command| command.exit_code().is_none())
-            })
-        })
-}
 
 /// Server-process ownership for ChatGPT resident commands. ChatGPT can replace
 /// its MCP transport between adjacent tool calls while retaining the same
@@ -652,22 +626,6 @@ impl ConversationExecSessionStore {
             .entry(identity.clone())
             .or_insert_with(|| Arc::new(ExecSessionState::new()))
             .clone();
-        if let Ok(mut registry) = CONVERSATION_EXEC_ACTIVITY
-            .get_or_init(Default::default)
-            .lock()
-        {
-            if registry.len() > 4096 {
-                registry.retain(|_, states| states.iter().any(|state| state.strong_count() > 0));
-            }
-            let states = registry
-                .entry(identity.stable_key().to_owned())
-                .or_default();
-            states.retain(|state| state.strong_count() > 0);
-            let weak = Arc::downgrade(&exec);
-            if !states.iter().any(|state| state.ptr_eq(&weak)) {
-                states.push(weak);
-            }
-        }
         transport_session.with_exec_state(exec)
     }
 

From 78ca93de05c134c75c4aff516e84a64ab1c4b66a Mon Sep 17 00:00:00 2001
From: Maxxx-step1121 <129008351+Maxxx-step1121@users.noreply.github.com>
Date: Fri, 13 Nov 2026 22:01:45 +0400
Subject: [PATCH 3/3] Let unread finished sessions give way to new commands

Admission counted every retained session against maxSessions, so a client
that never drained final results could hold all slots for the idle timeout.
Only running processes now block admission: when the map is full, the least
recently used finished-but-unread results are evicted (and logged) to admit
the new command.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/exec_sessions.rs | 87 ++++++++++++++++++++++++++++++++++++++++----
 1 file changed, 79 insertions(+), 8 deletions(-)

diff --git a/src/exec_sessions.rs b/src/exec_sessions.rs
index dbf15711..7b05cfa4 100644
--- a/src/exec_sessions.rs
+++ b/src/exec_sessions.rs
@@ -540,6 +540,7 @@ impl ExecSessionState {
         self.sessions.lock().unwrap().remove(&id)
     }
 
+    #[cfg(test)]
     fn len(&self) -> usize {
         self.sessions.lock().unwrap().len()
     }
@@ -560,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::<Vec<_>>();
+        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<Self>, idle_timeout: Duration) {
@@ -1108,6 +1139,7 @@ impl SessionState {
         self.exec.remove(id)
     }
 
+    #[cfg(test)]
     fn exec_session_count(&self) -> usize {
         self.exec.len()
     }
@@ -1505,14 +1537,13 @@ pub fn start_exec_session(
     cwd: &std::path::Path,
     shell: Option<&str>,
 ) -> Result<Arc<ExecSession>, String> {
-    {
-        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()) {
@@ -2249,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();