fix: idle workers stop polling the disk; web search, PowerShell policy, linked state and file-mode fixes

- Task workers (#6728, #6573): an idle worker with nothing claimable naps 1 s instead of stat-ing the queue every 200 ms (about 10 stats a second with two workers, which 0.9.12 did not do). Pending work and in-process notifies are unchanged. A settled queue now notices another process's write within about 1 s rather than 200 ms. - Task store lock (#6573): the holder records its pid and acquisition time in the lock file, so "busy" names the holder and how long it has held the lock when that can be read. The repeated "Task claim unavailable" error is logged once per episode, then at debug. - Web search (#6746): when the Bing fallback is allowed, DuckDuckGo gets 60% of the budget and Bing the rest, so a hanging DuckDuckGo no longer starves the fallback. A custom search_base_url still has no public fallback and keeps the full budget. - PowerShell (#6745): CODEWHALE_POWERSHELL_EXECUTION_POLICY=inherit omits the process-scope -ExecutionPolicy Bypass. The default is unchanged. Documented in ENVIRONMENTS (en, zh_hans). Not run on Windows. - Linked .codewhale: the workflow journal creates nothing until a record is appended and appends through the confined helpers; the coordination lock rejects a linked state path before creating directories. - File modes: the runtime log is created 0600 without following symlinks and an existing log is tightened; .reconcile.lock and current.json.lock reuse the private lock-file opener. fleet.lock was already 0600; a test pins it. Verified on macOS, Rust 1.99, together with the UI fixes in the next commit: cargo fmt --all --check clean; CI-policy clippy on codewhale-tui and codewhale-localization clean; focused tui lib tests 518 passed, 0 failed, 3 ignored; blocking-call budget within budget (one ratchet entry tightened for journal.rs); dead-code budget at budget. Full suite and other platforms not run locally. Signed-off-by: CodeWhale Bot <[email protected]> Co-Authored-By: Claude Fable 5.1 <[email protected]>
2026-11-08 08:04:25 -10:00 · 2026-10-01 17:47:51 -07:00
1 parent 9626357142
commit d2abc92da7
@@ -2628,6 +2628,19 @@ mod tests {
        );
    }

+    /// The workspace lock sidecar is created through the confined, private
+    /// file helper: owner-only, never umask-widened.
+    #[cfg(unix)]
+    #[test]
+    fn ledger_lock_file_is_owner_only() {
+        use std::os::unix::fs::PermissionsExt as _;
+        let workspace = TempDir::new().unwrap();
+        let ledger = FleetLedger::open(workspace.path()).unwrap();
+        let mode = |path: &Path| std::fs::metadata(path).unwrap().permissions().mode() & 0o777;
+        assert_eq!(mode(&ledger.lock_path), 0o600);
+        assert_eq!(mode(ledger.path()), 0o600);
+    }
+
    #[test]
    fn ledger_rejects_replaced_lock_identity() {
        let workspace = TempDir::new().unwrap();
@@ -1057,12 +1057,7 @@ impl OperationStore {
    }

    fn open_lock_file(&self) -> Result<fs::File> {
-        fs::OpenOptions::new()
-            .read(true)
-            .write(true)
-            .create(true)
-            .truncate(false)
-            .open(&self.lock_path)
+        crate::session_manager::open_private_lock_file(&self.lock_path)
            .with_context(|| format!("Failed to open {}", self.lock_path.display()))
    }

@@ -1665,6 +1660,19 @@ api_key_env = "CW_OPERATE_MISSING_TEST_KEY"
        Ok(())
    }

+    /// The cross-process lock sidecar is owner-only like the operation file.
+    #[cfg(unix)]
+    #[test]
+    fn operation_lock_file_is_owner_only() -> Result<()> {
+        use std::os::unix::fs::PermissionsExt as _;
+        let root = TempDir::new()?;
+        let store = OperationStore::open(root.path())?;
+        start_operation(&store, root.path(), None, None, true, "auto")?;
+        let lock = root.path().join("current.json.lock");
+        assert_eq!(fs::metadata(&lock)?.permissions().mode() & 0o777, 0o600);
+        Ok(())
+    }
+
    #[test]
    fn create_without_credentials_fails_closed() {
        let op = Operation::new("Keep shipping honest slices", None);
@@ -123,10 +123,7 @@ pub fn init() -> Result<TuiLogGuard> {
    let date = chrono::Local::now().format("%Y-%m-%d").to_string();
    let log_path = log_dir.join(log_file_name(&date, std::process::id()));

-    let file = OpenOptions::new()
-        .create(true)
-        .append(true)
-        .open(&log_path)
+    let file = open_log_file(&log_path)
        .with_context(|| format!("failed to open {}", log_path.display()))?;

    // The tracing-subscriber consumes a clone of the file handle for its
@@ -151,11 +148,7 @@ pub fn init() -> Result<TuiLogGuard> {
                    Ok(f) => Box::new(f),
                    Err(e) => {
                        tracing::warn!("Failed to clone log file handle: {e}, reopening");
-                        match std::fs::OpenOptions::new()
-                            .create(true)
-                            .append(true)
-                            .open(&log_path_clone)
-                        {
+                        match open_log_file(&log_path_clone) {
                            Ok(f) => Box::new(f),
                            Err(_) => Box::new(std::io::stderr()),
                        }
@@ -194,6 +187,29 @@ pub fn init() -> Result<TuiLogGuard> {
    })
 }

+/// Open (creating if needed) the per-process log for append. Logs can hold
+/// prompts and paths, so on Unix the file is owner-only: created 0600, an
+/// existing file is tightened to 0600, and a symlink at the log path is
+/// refused (`O_NOFOLLOW`) instead of followed to wherever it points.
+fn open_log_file(path: &Path) -> std::io::Result<std::fs::File> {
+    let mut options = OpenOptions::new();
+    options.create(true).append(true);
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::{OpenOptionsExt as _, PermissionsExt as _};
+        options
+            .mode(0o600)
+            .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC);
+        let file = options.open(path)?;
+        file.set_permissions(std::fs::Permissions::from_mode(0o600))?;
+        Ok(file)
+    }
+    #[cfg(not(unix))]
+    {
+        options.open(path)
+    }
+}
+
 pub(crate) fn log_directory() -> Option<PathBuf> {
    // $CODEWHALE_HOME is a hard override of the base data directory
    // (docs/CONFIGURATION.md): when SET, logs live under it and we do NOT fall
@@ -348,6 +364,52 @@ mod tests {
    use super::*;
    use std::fs::FileTimes;

+    #[cfg(unix)]
+    #[test]
+    fn log_file_is_owner_only_and_refuses_symlinks() {
+        use std::io::Write as _;
+        use std::os::unix::fs::{PermissionsExt as _, symlink};
+        let tmp = tempfile::TempDir::new().expect("temporary dir");
+        let mode = |path: &Path| {
+            std::fs::metadata(path)
+                .expect("metadata")
+                .permissions()
+                .mode()
+                & 0o777
+        };
+
+        let fresh = tmp.path().join("fresh.log");
+        open_log_file(&fresh)
+            .expect("create")
+            .write_all(b"one\n")
+            .expect("write");
+        assert_eq!(mode(&fresh), 0o600);
+
+        // An existing world-readable log is tightened, and stays appended to.
+        let old = tmp.path().join("old.log");
+        std::fs::write(&old, "before\n").expect("seed");
+        std::fs::set_permissions(&old, std::fs::Permissions::from_mode(0o644)).expect("chmod");
+        open_log_file(&old)
+            .expect("reopen")
+            .write_all(b"after\n")
+            .expect("write");
+        assert_eq!(mode(&old), 0o600);
+        assert_eq!(
+            std::fs::read_to_string(&old).expect("read"),
+            "before\nafter\n"
+        );
+
+        // A symlink planted at the log path is not followed.
+        let target = tmp.path().join("target.txt");
+        std::fs::write(&target, "keep").expect("target");
+        std::fs::set_permissions(&target, std::fs::Permissions::from_mode(0o644)).expect("chmod");
+        let link = tmp.path().join("link.log");
+        symlink(&target, &link).expect("symlink");
+        assert!(open_log_file(&link).is_err());
+        assert_eq!(std::fs::read_to_string(&target).expect("read"), "keep");
+        assert_eq!(mode(&target), 0o644);
+    }
+
    #[test]
    fn whitespace_home_override_is_consistent_across_tui_state_entry_points() {
        let _lock = crate::test_support::lock_test_env();
@@ -6381,8 +6381,14 @@ fn run_owner_lock_test_hook(point: OwnerLockTestPoint) {
 #[derive(Debug)]
 pub(crate) struct RuntimeProcessOwnerLock {
    _file: File,
+    /// Set once [`Self::record_holder`] wrote a holder record, so `Drop` clears
+    /// it before the lock is released instead of leaving a stale pid behind.
+    holder_recorded: bool,
 }

+/// Upper bound on the bytes read back from a lock file's holder record.
+const OWNER_LOCK_HOLDER_MAX_BYTES: u64 = 128;
+
 impl RuntimeProcessOwnerLock {
    /// Reuse the Runtime's protected OS lease for task-store ownership. A
    /// missing existing lease is uncertainty, never proof that an owner died.
@@ -6409,7 +6415,10 @@ impl RuntimeProcessOwnerLock {
                }
            }
            if owner_lock_is_at_path(&file, path)? {
-                return Ok(Some(Self { _file: file }));
+                return Ok(Some(Self {
+                    _file: file,
+                    holder_recorded: false,
+                }));
            }
            // Moved out of this store between open and lock: reopen.
        }
@@ -6453,7 +6462,10 @@ impl RuntimeProcessOwnerLock {
                }
            }
            if owner_lock_is_at_path(&file, &path)? {
-                return Ok(Self { _file: file });
+                return Ok(Self {
+                    _file: file,
+                    holder_recorded: false,
+                });
            }
            // A maintenance move renamed the file out of this store between
            // our open and our lock; the lock we hold is on the moved store.
@@ -6461,6 +6473,53 @@ impl RuntimeProcessOwnerLock {
        Err(Self::held_error())
    }

+    /// Write this process's pid and the acquisition time into the lock file so
+    /// a contender that finds the lock busy can say who holds it (#6573).
+    /// Best effort: diagnostics must never fail the lock holder.
+    pub(crate) fn record_holder(&mut self) {
+        let since_ms = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map_or(0, |elapsed| elapsed.as_millis());
+        let record = format!("pid={} since_ms={since_ms}\n", std::process::id());
+        let mut file = &self._file;
+        let written = file
+            .set_len(0)
+            .and_then(|()| file.seek(SeekFrom::Start(0)).map(|_| ()))
+            .and_then(|()| file.write_all(record.as_bytes()));
+        match written {
+            Ok(()) => self.holder_recorded = true,
+            Err(error) => tracing::debug!(%error, "Could not record lock holder"),
+        }
+    }
+
+    /// Read the holder recorded by [`Self::record_holder`]: its pid and how
+    /// long it has held the lock. Never blocks (advisory locks do not stop
+    /// reads) and never fails: anything unreadable or malformed is `None`.
+    pub(crate) fn read_holder(path: &Path) -> Option<(u32, Duration)> {
+        let file = open_runtime_store_file(path, "Owner lock holder record", |options| {
+            options.read(true);
+        })
+        .ok()?;
+        let mut text = String::new();
+        file.take(OWNER_LOCK_HOLDER_MAX_BYTES)
+            .read_to_string(&mut text)
+            .ok()?;
+        let mut pid = None;
+        let mut since_ms = None;
+        for field in text.split_whitespace() {
+            match field.split_once('=') {
+                Some(("pid", value)) => pid = value.parse::<u32>().ok(),
+                Some(("since_ms", value)) => since_ms = value.parse::<u64>().ok(),
+                _ => {}
+            }
+        }
+        let since = std::time::UNIX_EPOCH + Duration::from_millis(since_ms?);
+        let held_for = std::time::SystemTime::now()
+            .duration_since(since)
+            .unwrap_or_default();
+        Some((pid?, held_for))
+    }
+
    /// A held lock is typed `WouldBlock` so callers (the credential scrub,
    /// #6601) can tell a busy owner from a real lock failure.
    fn held_error() -> anyhow::Error {
@@ -6507,6 +6566,10 @@ impl RuntimeProcessOwnerLock {

 impl Drop for RuntimeProcessOwnerLock {
    fn drop(&mut self) {
+        if self.holder_recorded {
+            // Best effort: a stale pid must not outlive the lock it described.
+            let _ = self._file.set_len(0);
+        }
        // close() also releases, but unlocking first lets a same-process
        // reopen proceed without racing the previous fd's teardown (#5735).
        #[cfg(unix)]
@@ -134,7 +134,9 @@ fn normalize_managed_dir(path: PathBuf) -> std::io::Result<PathBuf> {
    std::env::current_dir().map(|cwd| cwd.join(path))
 }

-fn open_private_lock_file(path: &Path) -> io::Result<fs::File> {
+/// Open (creating if needed) an advisory-lock sidecar that is owner-only
+/// (0600), one regular link, and never followed through a symlink.
+pub(crate) fn open_private_lock_file(path: &Path) -> io::Result<fs::File> {
    let mut options = OpenOptions::new();
    options.create(true).read(true).write(true);
    #[cfg(unix)]
@@ -251,12 +251,8 @@ pub fn reconcile(
        dry_run: options.dry_run,
        ..ReconcileSummary::default()
    };
-    let lock_file = fs::OpenOptions::new()
-        .create(true)
-        .truncate(false)
-        .read(true)
-        .write(true)
-        .open(sessions_dir.join(RECONCILE_LOCK_FILE))?;
+    let lock_file =
+        crate::session_manager::open_private_lock_file(&sessions_dir.join(RECONCILE_LOCK_FILE))?;
    if !crate::runtime_threads::try_lock_file_exclusive(&lock_file)? {
        summary.skipped_concurrent = true;
        return Ok(summary);
@@ -153,6 +153,25 @@ fn manifests(root: &Path) -> String {
 /// The measured shape (#6144): documents are bound to a store under
 /// *another* id's directory, and that directory has no document of its own.
 /// Only the store nothing binds may move.
+/// The reconcile lock sidecar is owner-only, even when an older release left
+/// it world-readable.
+#[cfg(unix)]
+#[test]
+fn reconcile_lock_file_is_owner_only() {
+    use std::os::unix::fs::PermissionsExt as _;
+    let _lock = lock_test_env();
+    let fixture = Fixture::new();
+    let lock = fixture.dir().join(RECONCILE_LOCK_FILE);
+    let mode = || fs::metadata(&lock).unwrap().permissions().mode() & 0o777;
+
+    fixture.run();
+    assert_eq!(mode(), 0o600);
+
+    fs::set_permissions(&lock, fs::Permissions::from_mode(0o644)).unwrap();
+    fixture.run();
+    assert_eq!(mode(), 0o600);
+}
+
 #[test]
 fn reconcile_sets_aside_only_stores_no_document_binds() {
    let _env = lock_test_env();
@@ -165,14 +165,34 @@ fn powershell_prefers_script_file(shell_command: &str) -> bool {
 /// command behaves the same whether it travels as `-Command` or `-File`.
 /// Group Policy scopes still take precedence; see the module's known
 /// limitations.
+///
+/// `CODEWHALE_POWERSHELL_EXECUTION_POLICY=inherit` omits the flag so the
+/// machine/user policy applies instead (#6745). Unset, `bypass`, or any other
+/// value keeps the default.
 fn powershell_base_args() -> Vec<String> {
-    vec![
+    powershell_base_args_for_policy(
+        std::env::var(POWERSHELL_EXECUTION_POLICY_ENV)
+            .ok()
+            .as_deref(),
+    )
+}
+
+/// Environment variable that opts out of the process-scope policy bypass.
+const POWERSHELL_EXECUTION_POLICY_ENV: &str = "CODEWHALE_POWERSHELL_EXECUTION_POLICY";
+
+/// [`powershell_base_args`] with the opt-out value passed in, so the decision
+/// is testable without touching the process environment.
+fn powershell_base_args_for_policy(setting: Option<&str>) -> Vec<String> {
+    let mut args = vec![
        "-NoLogo".to_string(),
        "-NoProfile".to_string(),
        "-NonInteractive".to_string(),
-        "-ExecutionPolicy".to_string(),
-        "Bypass".to_string(),
-    ]
+    ];
+    if !setting.is_some_and(|value| value.trim().eq_ignore_ascii_case("inherit")) {
+        args.push("-ExecutionPolicy".to_string());
+        args.push("Bypass".to_string());
+    }
+    args
 }

 /// Wrap a model/user PowerShell command so native program failures surface
@@ -720,6 +740,34 @@ mod tests {
        }
    }

+    /// #6745: `inherit` is the only value that drops the process-scope bypass;
+    /// unset, `bypass`, and anything unrecognised keep it. Tested on the pure
+    /// decision so no test mutates the process environment.
+    #[test]
+    fn powershell_execution_policy_opt_out_only_honours_inherit() {
+        let bypass = |args: &[String]| {
+            args.iter()
+                .position(|a| a == "-ExecutionPolicy")
+                .map(|index| args[index + 1].clone())
+        };
+        for setting in [
+            None,
+            Some("bypass"),
+            Some("Bypass"),
+            Some(""),
+            Some("restricted"),
+        ] {
+            let args = powershell_base_args_for_policy(setting);
+            assert_eq!(bypass(&args).as_deref(), Some("Bypass"), "{setting:?}");
+            assert!(args.contains(&"-NonInteractive".to_string()), "{setting:?}");
+        }
+        for setting in ["inherit", "INHERIT", " inherit "] {
+            let args = powershell_base_args_for_policy(Some(setting));
+            assert_eq!(bypass(&args), None, "{setting:?}: {args:?}");
+            assert_eq!(args, ["-NoLogo", "-NoProfile", "-NonInteractive"]);
+        }
+    }
+
    #[test]
    fn powershell_trailing_comment_cannot_swallow_exit_capture() {
        // An unquoted `#` in a single-line payload comments to end-of-line;
@@ -43,6 +43,11 @@ const EVENT_CATCHUP_POLL: Duration = Duration::from_millis(200);
 // ignore its eligibility or generation fence.
 const CURRENT_TASK_SCHEMA_VERSION: u32 = 4;
 const STORE_REFRESH_INTERVAL: Duration = Duration::from_millis(200);
+/// How long a worker with nothing claimable naps between queue-fingerprint
+/// checks. An in-process notify still wakes it at once; this only bounds how
+/// soon it notices another process's queue write, so an idle session stats the
+/// queue about once a second per worker instead of five times (#6728, #6573).
+const STORE_SETTLED_NAP: Duration = Duration::from_secs(1);
 /// Retry an empty claim while queue metadata is missing or too recent to
 /// distinguish writes on a coarse-mtime filesystem. Once settled, idle workers
 /// wait for admission notifications or queue changes instead of reloading every
@@ -1472,6 +1477,9 @@ pub struct TaskManager {
    /// Full store loads performed by this manager (tests only, #6573).
    #[cfg(test)]
    store_loads: std::sync::atomic::AtomicUsize,
+    /// Queue fingerprint reads performed by worker loops (tests only, #6728).
+    #[cfg(test)]
+    fingerprint_reads: std::sync::atomic::AtomicUsize,
 }

 /// Cheap stat-only view of the shared queue file. Everything that makes work
@@ -1561,6 +1569,18 @@ impl ClaimSchedule {
            || self.next_claim.is_some_and(|deadline| now >= deadline)
    }

+    /// How long the worker may sleep before its next pass. Nothing claimable
+    /// (a settled empty queue, no retry deadline) only needs to notice another
+    /// process's queue write; pending work or a retry deadline keeps the
+    /// short tick.
+    fn nap(&self) -> Duration {
+        if self.next_claim.is_none() {
+            STORE_SETTLED_NAP
+        } else {
+            STORE_REFRESH_INTERVAL
+        }
+    }
+
    fn claimed_task(&mut self, now: Instant) {
        self.seen = None;
        self.next_claim = Some(now);
@@ -1590,6 +1610,18 @@ impl ClaimSchedule {
    }
 }

+/// The "store is busy" error, naming the lock holder when its record can be
+/// read. Reading the record is best effort and never fails the caller (#6573).
+fn store_busy_error(lock_path: &Path) -> anyhow::Error {
+    match RuntimeProcessOwnerLock::read_holder(lock_path) {
+        Some((pid, held_for)) => anyhow!(
+            "Task store is busy; state is unavailable (lock held by pid {pid} for {}s)",
+            held_for.as_secs()
+        ),
+        None => anyhow!("Task store is busy; state is unavailable"),
+    }
+}
+
 struct ManagerState {
    tasks: HashMap<String, TaskRecord>,
    queue: VecDeque<String>,
@@ -1726,6 +1758,8 @@ impl TaskManager {
            shutdown_drain: Mutex::new(()),
            #[cfg(test)]
            store_loads: std::sync::atomic::AtomicUsize::new(0),
+            #[cfg(test)]
+            fingerprint_reads: std::sync::atomic::AtomicUsize::new(0),
        });

        {
@@ -2483,29 +2517,49 @@ impl TaskManager {
    async fn worker_loop(self: Arc<Self>) {
        let mut schedule = ClaimSchedule::new(Instant::now());
        let mut woken = true;
+        // True while consecutive claims are failing: the first failure of an
+        // episode is an error, repeats are debug until a claim succeeds (#6573).
+        let mut claim_failing = false;
        loop {
            if self.cancel_token.is_cancelled() {
                break;
            }
            // Read before claiming so a write racing the claim changes the
            // fingerprint and is picked up on the next pass.
+            #[cfg(test)]
+            self.fingerprint_reads
+                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            let fingerprint = StoreFingerprint::read(&self.queue_path);
            if schedule.should_claim(&fingerprint, Instant::now(), woken) {
                match self.claim_next_task().await {
                    Ok(Some((id, request, cancel))) => {
+                        claim_failing = false;
                        schedule.claimed_task(Instant::now());
                        self.run_task(id, request, cancel).await;
                        woken = true;
                        continue;
                    }
-                    Ok(None) => schedule.found_nothing(fingerprint, Instant::now()),
+                    Ok(None) => {
+                        claim_failing = false;
+                        schedule.found_nothing(fingerprint, Instant::now());
+                    }
                    Err(error) => {
                        let retry = schedule.claim_failed(fingerprint, Instant::now());
-                        tracing::error!(
-                            %error,
-                            retry_ms = retry.as_millis() as u64,
-                            "Task claim unavailable; executor was not polled"
-                        );
+                        let retry_ms = retry.as_millis() as u64;
+                        if claim_failing {
+                            tracing::debug!(
+                                %error,
+                                retry_ms,
+                                "Task claim still unavailable; executor was not polled"
+                            );
+                        } else {
+                            claim_failing = true;
+                            tracing::error!(
+                                %error,
+                                retry_ms,
+                                "Task claim unavailable; executor was not polled"
+                            );
+                        }
                    }
                }
            }
@@ -2513,7 +2567,7 @@ impl TaskManager {
            tokio::select! {
                _ = self.cancel_token.cancelled() => break,
                _ = self.notify.notified() => woken = true,
-                _ = sleep(STORE_REFRESH_INTERVAL) => {},
+                _ = sleep(schedule.nap()) => {},
            }
        }
    }
@@ -3221,11 +3275,13 @@ impl TaskManager {
        let deadline = Instant::now() + Duration::from_secs(5);
        let mut wait = Duration::from_millis(5);
        loop {
-            if let Some(owner) = RuntimeProcessOwnerLock::try_acquire_file(&path, true)? {
+            if let Some(mut owner) = RuntimeProcessOwnerLock::try_acquire_file(&path, true)? {
+                // Lets a contender that times out name this process (#6573).
+                owner.record_holder();
                return Ok(owner);
            }
            if Instant::now() >= deadline {
-                bail!("Task store is busy; state is unavailable");
+                return Err(store_busy_error(&path));
            }
            sleep(wait).await;
            wait = (wait * 2).min(Duration::from_millis(50));
@@ -4215,7 +4271,10 @@ mod tests {
            manager.persist_queue_locked(&VecDeque::from([task.id.clone(), second.id.clone()]))?;
        }
        for id in [&task.id, &second.id] {
-            let done = wait_for_terminal_state(&manager, id, Duration::from_secs(2)).await?;
+            // Settled idle workers recheck the queue every STORE_SETTLED_NAP.
+            let done =
+                wait_for_terminal_state(&manager, id, STORE_SETTLED_NAP + Duration::from_secs(2))
+                    .await?;
            assert_eq!(done.status, TaskStatus::Completed);
        }
        assert_eq!(executions.load(Ordering::SeqCst), 2);
@@ -4223,6 +4282,91 @@ mod tests {
        Ok(())
    }

+    /// #6728: with nothing claimable, each worker rechecks the queue about once
+    /// per STORE_SETTLED_NAP rather than every STORE_REFRESH_INTERVAL. The
+    /// bound is an upper limit, so a slow machine only lowers the count.
+    #[tokio::test]
+    async fn settled_idle_workers_check_the_queue_at_the_settled_nap() -> Result<()> {
+        let root = tempfile::tempdir()?;
+        let workers = 2;
+        let manager = TaskManager::start_with_executor(
+            TaskManagerConfig {
+                worker_count: workers,
+                ..test_config(root.path().to_path_buf())
+            },
+            Arc::new(MockExecutor),
+        )
+        .await?;
+        sleep(STORE_IDLE_POLL_INTERVAL + Duration::from_millis(300)).await;
+        manager.fingerprint_reads.store(0, Ordering::Relaxed);
+        manager.store_loads.store(0, Ordering::Relaxed);
+
+        let window = Duration::from_secs(3);
+        sleep(window).await;
+        let reads = manager.fingerprint_reads.load(Ordering::Relaxed);
+        let loads = manager.store_loads.load(Ordering::Relaxed);
+        let ticks = (window.as_millis() / STORE_SETTLED_NAP.as_millis()) as usize + 1;
+        assert!(
+            reads <= workers * ticks,
+            "{workers} idle workers read the queue fingerprint {reads} times in {window:?}; \
+             expected at most {}",
+            workers * ticks
+        );
+        assert_eq!(loads, 0, "idle workers reloaded a settled store");
+        manager.shutdown_and_wait().await?;
+        Ok(())
+    }
+
+    #[test]
+    fn claim_schedule_naps_long_only_when_nothing_is_claimable() {
+        let now = Instant::now();
+        let settled = StoreFingerprint(Some((std::time::SystemTime::UNIX_EPOCH, 1, 1)));
+        let mut schedule = ClaimSchedule::new(now);
+        // A fresh schedule has a claim pending.
+        assert_eq!(schedule.nap(), STORE_REFRESH_INTERVAL);
+        schedule.found_nothing(settled.clone(), now);
+        assert_eq!(schedule.nap(), STORE_SETTLED_NAP);
+        // A failed claim keeps the short tick so a queue change is seen early.
+        schedule.claim_failed(settled.clone(), now);
+        assert_eq!(schedule.nap(), STORE_REFRESH_INTERVAL);
+        // Recent or missing metadata retries on a deadline, so it stays short.
+        schedule.found_nothing(StoreFingerprint(None), now);
+        assert_eq!(schedule.nap(), STORE_REFRESH_INTERVAL);
+        schedule.claimed_task(now);
+        assert_eq!(schedule.nap(), STORE_REFRESH_INTERVAL);
+    }
+
+    /// #6573: a busy store names the holder; an unreadable record falls back
+    /// to the bare message without failing.
+    #[test]
+    fn busy_store_error_names_the_lock_holder_when_known() -> Result<()> {
+        let root = tempfile::tempdir()?;
+        let path = root.path().join("task-store.lock");
+        let mut holder =
+            RuntimeProcessOwnerLock::try_acquire_file(&path, true)?.context("first acquire")?;
+        holder.record_holder();
+        assert!(RuntimeProcessOwnerLock::try_acquire_file(&path, true)?.is_none());
+
+        let message = store_busy_error(&path).to_string();
+        assert!(
+            message.contains(&format!("held by pid {}", std::process::id())),
+            "{message}"
+        );
+
+        // Releasing clears the record, so a later error cannot name a stale pid.
+        drop(holder);
+        assert_eq!(
+            store_busy_error(&path).to_string(),
+            "Task store is busy; state is unavailable"
+        );
+        // Garbage and a missing file also fall back.
+        fs::write(&path, "pid=oops since_ms=\n")?;
+        assert!(RuntimeProcessOwnerLock::read_holder(&path).is_none());
+        fs::remove_file(&path)?;
+        assert!(RuntimeProcessOwnerLock::read_holder(&path).is_none());
+        Ok(())
+    }
+
    #[tokio::test]
    async fn running_task_observes_durable_cancel_without_a_queue_change() -> Result<()> {
        let root = tempfile::tempdir()?;
@@ -3570,6 +3570,9 @@ impl CoordinationProcessLock {
    fn acquire(state_root: &Path) -> Result<Self> {
        let requested_root = normalize_subagent_workspace(state_root);
        let lock_dir = requested_root.join(".codewhale").join("state");
+        // Refuse a linked `.codewhale` or `state` before creating anything:
+        // `create_dir_all` follows links and would populate the target.
+        reject_root_relative_symlinks(&requested_root, &lock_dir.join(SUBAGENT_STATE_LOCK_FILE))?;
        fs::create_dir_all(&lock_dir)?;
        // Creating a missing root can change its canonical spelling on
        // Windows (for example by adding a `\\?\` prefix). Re-resolve both
@@ -11516,6 +11516,34 @@ fn persist_state_rejects_symlinked_state_directory() {
    );
 }

+/// A linked `.codewhale` must be refused before anything is created: the
+/// coordination lock used to `create_dir_all` through the link first.
+#[cfg(unix)]
+#[test]
+fn coordination_lock_refuses_linked_codewhale_without_writing_through_it() {
+    let tmp = tempdir().expect("tempdir");
+    let workspace = tmp.path().join("workspace");
+    let outside = tmp.path().join("outside");
+    std::fs::create_dir_all(&workspace).expect("mkdir workspace");
+    std::fs::create_dir_all(&outside).expect("mkdir outside");
+    std::os::unix::fs::symlink(&outside, workspace.join(".codewhale")).expect("link .codewhale");
+
+    let Err(err) = CoordinationProcessLock::acquire(&workspace) else {
+        panic!("a linked .codewhale must be refused");
+    };
+    let message = format!("{err:#}");
+    assert!(
+        message.contains("must not traverse symlinks")
+            || message.contains("must stay within state root"),
+        "{message}"
+    );
+    assert_eq!(
+        std::fs::read_dir(&outside).expect("read outside").count(),
+        0,
+        "nothing may be created through the link"
+    );
+}
+
 #[test]
 fn test_interrupted_status_name_and_summary() {
    let snapshot = make_snapshot(SubAgentStatus::Interrupted(
@@ -1502,6 +1502,10 @@ pub(crate) async fn run_backend_search(
    }
 }

+/// Share of the search budget a DuckDuckGo request may use when Bing is allowed
+/// to answer after it (#6746).
+const DUCKDUCKGO_BUDGET_SHARE: f64 = 0.6;
+
 #[derive(Clone, Copy)]
 struct ScrapeEndpoints<'a> {
    bing: &'a str,
@@ -1548,9 +1552,11 @@ async fn run_scrape_search_with_endpoints(
    context: &ToolContext,
    endpoints: ScrapeEndpoints<'_>,
 ) -> Result<BackendSearch, ToolError> {
+    let started = Instant::now();
+    let budget = Duration::from_millis(timeout_ms);
    let decider = context.network_policy.as_ref();
    let client = crate::tls::reqwest_client_builder()
-        .timeout(Duration::from_millis(timeout_ms))
+        .timeout(budget)
        .user_agent(USER_AGENT)
        .build()
        .map_err(|error| {
@@ -1561,7 +1567,8 @@ async fn run_scrape_search_with_endpoints(

    if provider == SearchProvider::Bing {
        check_policy(decider, BING_HOST)?;
-        let results = run_bing_search(&client, &query.query, max_results, endpoints.bing).await?;
+        let results =
+            run_bing_search(&client, &query.query, max_results, endpoints.bing, budget).await?;
        return Ok(BackendSearch {
            backend: BackendId::Bing,
            source: "bing".to_string(),
@@ -1578,7 +1585,20 @@ async fn run_scrape_search_with_endpoints(
        .allow_bing_fallback
        .unwrap_or_else(|| duckduckgo_allows_bing_fallback(context.search_base_url.as_deref()));
    check_policy(decider, &duckduckgo_host)?;
-    let fetched = fetch_duckduckgo_html(&client, &url).await;
+    // #6746: a hanging DuckDuckGo must leave the Bing fallback time to answer.
+    // When Bing may follow, DuckDuckGo gets a share of the budget and Bing
+    // the remainder; without the fallback DuckDuckGo keeps the whole budget.
+    let duckduckgo_budget = if allow_bing_fallback {
+        budget.mul_f64(DUCKDUCKGO_BUDGET_SHARE)
+    } else {
+        budget
+    };
+    let fetched = fetch_duckduckgo_html(&client, &url, duckduckgo_budget).await;
+    let bing_budget = || {
+        budget
+            .saturating_sub(started.elapsed())
+            .max(Duration::from_millis(1))
+    };
    let body = match fetched {
        Ok(body) => body,
        // #6746: an unreachable DuckDuckGo (connection error, timeout, or a
@@ -1587,7 +1607,15 @@ async fn run_scrape_search_with_endpoints(
        // error; otherwise the original failure is reported.
        Err(error) if allow_bing_fallback => {
            check_policy(decider, BING_HOST)?;
-            return match run_bing_search(&client, &query.query, max_results, endpoints.bing).await {
+            return match run_bing_search(
+                &client,
+                &query.query,
+                max_results,
+                endpoints.bing,
+                bing_budget(),
+            )
+            .await
+            {
                Ok(results) if !results.is_empty() => {
                    degraded.push(DegradedReason::BackendUnavailable {
                        backend: BackendId::DuckDuckGo,
@@ -1658,7 +1686,15 @@ async fn run_scrape_search_with_endpoints(
    }

    check_policy(decider, BING_HOST)?;
-    match run_bing_search(&client, &query.query, max_results, endpoints.bing).await {
+    match run_bing_search(
+        &client,
+        &query.query,
+        max_results,
+        endpoints.bing,
+        bing_budget(),
+    )
+    .await
+    {
        Ok(results) if !results.is_empty() => {
            degraded.push(DegradedReason::ScrapeFallback {
                from: BackendId::DuckDuckGo,
@@ -1696,9 +1732,14 @@ async fn run_scrape_search_with_endpoints(

 /// Fetch the DuckDuckGo HTML results page. A transport failure or a non-2xx
 /// status is an error so the caller can decide whether Bing may answer.
-async fn fetch_duckduckgo_html(client: &reqwest::Client, url: &str) -> Result<String, String> {
+async fn fetch_duckduckgo_html(
+    client: &reqwest::Client,
+    url: &str,
+    timeout: Duration,
+) -> Result<String, String> {
    let resp = client
        .get(url)
+        .timeout(timeout)
        .header(
            "Accept",
            "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
@@ -2372,12 +2413,14 @@ async fn run_bing_search(
    query: &str,
    max_results: usize,
    endpoint: &str,
+    timeout: Duration,
 ) -> Result<Vec<WebSearchEntry>, ToolError> {
    let mut url = reqwest::Url::parse(endpoint)
        .map_err(|error| ToolError::invalid_input(format!("Invalid Bing endpoint: {error}")))?;
    url.query_pairs_mut().append_pair("q", query);
    let resp = client
        .get(url)
+        .timeout(timeout)
        .header(
            "Accept",
            "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
@@ -4678,6 +4721,139 @@ mod tests {
        assert!(error.to_string().contains("HTTP 503"), "{error}");
    }

+    /// #6746: a DuckDuckGo that accepts the connection and then hangs must not
+    /// eat the whole budget; Bing answers inside the same total.
+    #[tokio::test]
+    async fn hanging_duckduckgo_leaves_bing_time_inside_the_total_budget() {
+        use crate::config::SearchProvider;
+        use crate::tools::spec::ToolContext;
+        use wiremock::matchers::{method, path};
+        use wiremock::{Mock, MockServer, ResponseTemplate};
+
+        let server = MockServer::start().await;
+        Mock::given(method("GET"))
+            .and(path("/html/"))
+            .respond_with(ResponseTemplate::new(200).set_delay(Duration::from_secs(30)))
+            .mount(&server)
+            .await;
+        Mock::given(method("GET"))
+            .and(path("/bing"))
+            .respond_with(ResponseTemplate::new(200).set_body_string(
+                r#"
+                <ol><li class="b_algo">
+                  <h2><a href="https://example.com/after-hang">After the hang</a></h2>
+                  <div class="b_caption"><p>Bing answered after DuckDuckGo hung.</p></div>
+                </li></ol>
+                "#,
+            ))
+            .mount(&server)
+            .await;
+
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let mut context = ToolContext::new(tmp.path().to_path_buf());
+        context.search_provider = SearchProvider::DuckDuckGo;
+        context.search_base_url = Some(format!("{}/html/", server.uri()));
+        let query = SearchQuery::new("hang".to_string(), 5, None, Vec::new(), None);
+        let budget = Duration::from_millis(3_000);
+        let started = Instant::now();
+        let outcome = tokio::time::timeout(
+            budget,
+            run_scrape_search_with_endpoints(
+                SearchProvider::DuckDuckGo,
+                &query,
+                3_000,
+                &context,
+                ScrapeEndpoints {
+                    bing: &format!("{}/bing", server.uri()),
+                    allow_bing_fallback: Some(true),
+                },
+            ),
+        )
+        .await
+        .expect("the whole search must finish inside its budget");
+        let raw = outcome.expect("Bing fallback should answer after DuckDuckGo hangs");
+
+        assert_eq!(raw.backend, BackendId::Bing);
+        assert_eq!(raw.results.len(), 1);
+        assert!(started.elapsed() < budget, "{:?}", started.elapsed());
+        assert!(
+            raw.degraded.contains(&DegradedReason::BackendUnavailable {
+                backend: BackendId::DuckDuckGo
+            }),
+            "{:?}",
+            raw.degraded
+        );
+    }
+
+    /// #6746: a custom `search_base_url` keeps no public fallback and keeps the
+    /// whole budget: a slow private endpoint is waited for, and Bing is never
+    /// contacted, whether the endpoint answers late or hangs.
+    #[tokio::test]
+    async fn custom_base_url_keeps_full_budget_and_never_reaches_bing() {
+        use crate::config::SearchProvider;
+        use crate::tools::spec::ToolContext;
+        use wiremock::matchers::{method, path};
+        use wiremock::{Mock, MockServer, ResponseTemplate};
+
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let query = SearchQuery::new("private".to_string(), 5, None, Vec::new(), None);
+        for (delay, expect_results) in [
+            (Duration::from_millis(1_500), true),
+            (Duration::from_secs(30), false),
+        ] {
+            let server = MockServer::start().await;
+            Mock::given(method("GET"))
+                .and(path("/html/"))
+                .respond_with(ResponseTemplate::new(200).set_delay(delay).set_body_string(
+                    r#"
+                            <html><body>
+                              <a class="result__a" href="https://example.com/private">Private</a>
+                              <div class="result__snippet">Private result</div>
+                            </body></html>
+                            "#,
+                ))
+                .mount(&server)
+                .await;
+            Mock::given(method("GET"))
+                .and(path("/bing"))
+                .respond_with(ResponseTemplate::new(200))
+                .mount(&server)
+                .await;
+
+            let mut context = ToolContext::new(tmp.path().to_path_buf());
+            context.search_provider = SearchProvider::DuckDuckGo;
+            context.search_base_url = Some(format!("{}/html/", server.uri()));
+            // 60% of 2s would be 1.2s: a 1.5s answer only arrives if the
+            // private endpoint kept the whole budget.
+            let outcome = run_scrape_search_with_endpoints(
+                SearchProvider::DuckDuckGo,
+                &query,
+                2_000,
+                &context,
+                ScrapeEndpoints {
+                    bing: &format!("{}/bing", server.uri()),
+                    allow_bing_fallback: None,
+                },
+            )
+            .await;
+            if expect_results {
+                let raw = outcome.expect("a slow private endpoint keeps its full budget");
+                assert_eq!(raw.backend, BackendId::DuckDuckGo);
+                assert_eq!(raw.results.len(), 1);
+            } else {
+                assert!(outcome.is_err(), "a hanging private endpoint is an error");
+            }
+            let bing_requests = server
+                .received_requests()
+                .await
+                .expect("recorded requests")
+                .iter()
+                .filter(|request| request.url.path() == "/bing")
+                .count();
+            assert_eq!(bing_requests, 0, "custom base URL must not reach Bing");
+        }
+    }
+
    #[tokio::test]
    async fn search_base_url_with_non_duckduckgo_provider_is_explicit_error() {
        use crate::config::SearchProvider;
@@ -5,7 +5,6 @@ use super::{
 };
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
-use std::fs::OpenOptions;
 use std::io::{BufRead, Write};
 use std::path::{Path, PathBuf};
 use std::sync::{Arc, Mutex, OnceLock};
@@ -178,34 +177,34 @@ enum WorkflowJournalRecord {

 #[derive(Debug)]
 struct WorkflowRunJournal {
+    /// Workspace the ledger must stay inside; links below it are refused.
+    root: PathBuf,
    ledger_path: PathBuf,
 }

 impl WorkflowRunJournal {
+    /// Opening has no side effects: `.codewhale/` and the ledger appear with
+    /// the first appended record, and only through the confined helpers, so a
+    /// linked `.codewhale` never receives an empty `state/` or ledger.
    fn open(workspace: &Path) -> Self {
-        let dir = workspace.join(CODEWHALE_DIR);
-        if let Err(err) = std::fs::create_dir_all(&dir) {
-            warn!(
-                "workflow journal dir create failed ({}): {err}",
-                dir.display()
-            );
+        Self {
+            root: workspace.to_path_buf(),
+            ledger_path: workspace.join(CODEWHALE_DIR).join(WORKFLOW_RUNS_FILE),
        }
-        let ledger_path = dir.join(WORKFLOW_RUNS_FILE);
-        if !ledger_path.exists()
-            && let Err(err) = std::fs::write(&ledger_path, "")
-        {
-            warn!(
-                "workflow journal create failed ({}): {err}",
-                ledger_path.display()
-            );
-        }
-        Self { ledger_path }
    }

    fn hydrate_runs(&self, recover_orphans: bool) -> HashMap<String, WorkflowRunRecord> {
-        let file = match std::fs::File::open(&self.ledger_path) {
+        let file = match crate::fs_confined::open_read(&self.root, &self.ledger_path) {
            Ok(file) => file,
-            Err(_) => return HashMap::new(),
+            Err(err) => {
+                if err.kind() != std::io::ErrorKind::NotFound {
+                    warn!(
+                        "workflow journal unreadable ({}): {err}",
+                        self.ledger_path.display()
+                    );
+                }
+                return HashMap::new();
+            }
        };
        let mut runs = HashMap::new();
        for line in std::io::BufReader::new(file).lines() {
@@ -297,10 +296,7 @@ impl WorkflowRunJournal {
        let mut line =
            serde_json::to_string(record).map_err(|err| std::io::Error::other(err.to_string()))?;
        line.push('\n');
-        let mut file = OpenOptions::new()
-            .create(true)
-            .append(true)
-            .open(&self.ledger_path)?;
+        let mut file = crate::fs_confined::open_append(&self.root, &self.ledger_path)?;
        file.write_all(line.as_bytes())?;
        file.flush()?;
        Ok(())
@@ -381,6 +377,45 @@ mod tests {
        }
    }

+    /// Opening the journal creates nothing; the first record creates the
+    /// ledger under a private `.codewhale/`.
+    #[test]
+    fn opening_the_journal_creates_nothing_until_a_record_is_appended() {
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let state = WorkflowWorkspaceState::open(tmp.path());
+        assert!(!tmp.path().join(CODEWHALE_DIR).exists());
+        state.record_snapshot(&sample_record("workflow_lazy", WorkflowRunStatus::Running));
+        assert!(state.journal_path().is_file());
+    }
+
+    /// A linked `.codewhale` gets no empty `state/` or ledger, and appends and
+    /// reads are refused instead of writing through the link.
+    #[cfg(unix)]
+    #[test]
+    fn journal_never_writes_through_a_linked_codewhale_dir() {
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let workspace = tmp.path().join("workspace");
+        let outside = tmp.path().join("outside");
+        std::fs::create_dir_all(&workspace).expect("mkdir workspace");
+        std::fs::create_dir_all(&outside).expect("mkdir outside");
+        std::os::unix::fs::symlink(&outside, workspace.join(CODEWHALE_DIR)).expect("link");
+
+        let journal = WorkflowRunJournal::open(&workspace);
+        assert!(journal.hydrate_runs(true).is_empty());
+        let err = journal
+            .append_snapshot(&sample_record("workflow_link", WorkflowRunStatus::Running))
+            .expect_err("append through a linked .codewhale must be refused");
+        assert!(!err.to_string().is_empty());
+
+        // The state-level entry points degrade to a warning, not a write.
+        let state = WorkflowWorkspaceState::open(&workspace);
+        state.record_progress("workflow_link", "phase: scan");
+        assert_eq!(
+            std::fs::read_dir(&outside).expect("read outside").count(),
+            0
+        );
+    }
+
    #[test]
    fn workflow_journal_hydrates_snapshots_and_progress() {
        let tmp = tempfile::tempdir().expect("tempdir");
@@ -80,6 +80,12 @@ policy does not govern. Scripts that a command itself calls run under the same
 process scope; the shell tool's approval and sandbox settings, not the
 execution policy, decide what may run.

+To let the machine or user policy apply instead, set
+`CODEWHALE_POWERSHELL_EXECUTION_POLICY=inherit` before starting Codewhale: the
+shell tool then omits `-ExecutionPolicy` entirely, so a `Restricted` or
+`AllSigned` policy refuses multiline commands that need the temporary `.ps1`
+script. Unset, `bypass`, or any other value keeps the default `Bypass`.
+
 ## Consolidated runtime commands

 The current `codewhale` binary runs the TUI in-process. Release installers copy
@@ -72,6 +72,12 @@ PowerShell 的拒绝信息会作为该命令的错误返回；Codewhale 不会
 同一进程范围内运行；决定什么可以运行的是 shell 工具的审批与沙箱设置，而不是
 执行策略。

+若希望改由机器或用户策略生效，请在启动 Codewhale 之前设置
+`CODEWHALE_POWERSHELL_EXECUTION_POLICY=inherit`：此时 shell 工具会完全省略
+`-ExecutionPolicy`，因此策略为 `Restricted` 或 `AllSigned` 时，需要临时 `.ps1`
+脚本的多行命令会被拒绝。未设置、设置为 `bypass` 或任何其他值时，仍保持默认的
+`Bypass`。
+
 ## 统一的运行时命令

 当前的 `codewhale` 二进制在进程内运行 TUI。发布安装器会把同样的字节复制到
@@ -467,8 +467,7 @@
    "std_fs": 1
  },
  "crates/tui/src/tools/workflow/journal.rs": {
-    "path_canonicalize": 2,
-    "std_fs": 4
+    "path_canonicalize": 2
  },
  "crates/tui/src/tools/workflow/mod.rs": {
    "path_canonicalize": 4,