lean-ctx-server 3.12.3

// SPDX-License-Identifier: Apache-2.0
//! Explicit, in-process background jobs for `ctx_shell`.
//!
//! The MCP request may complete immediately, but the child keeps the exact
//! timeout, allow-list, path-jail and process-group policy of foreground shell
//! execution. Jobs intentionally live in the daemon: restarting it invalidates
//! outstanding jobs rather than silently orphaning subprocesses.

use std::collections::HashMap;
use std::path::Path;
use std::sync::{
    Arc, LazyLock, Mutex,
    atomic::{AtomicBool, AtomicU64, Ordering},
    mpsc,
};
use std::time::{Duration, Instant};

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum JobState {
    Running { output: String },
    Completed { output: String, exit_code: i32 },
    Cancelled { output: String },
}

const MAX_RETAINED_COMPLETED_JOBS: usize = 64;
const MAX_RETAINED_COMPLETED_BYTES: usize = 16 * 1024 * 1024;
const MAX_FOREGROUND_OUTPUT_RECEIVERS: usize = 64;
const COMPLETED_JOB_TTL: Duration = Duration::from_mins(5);

struct Job {
    cancel: Arc<AtomicBool>,
    state: JobState,
    finished_at: Option<Instant>,
    // Terminal text is storage-admitted only. Running jobs keep no output.
    generation: u64,
    actual_exit_code: i32,
    completeness: lean_ctx::core::context_admission::OutputCompleteness,
    output_withheld: bool,
    // Q13: the call's fixed, content-free notice. Kept beside the retained
    // text, never in it: storage admission, scrubbing and replacement do not
    // touch it, and every delivery puts it in front once.
    notice: Option<&'static str>,
    /// The class the command's output is admitted with (pz21 §1.8).
    class: lean_ctx::core::context_admission::origin::SourceClass,
    // One-shot raw results go only to foreground receivers that are still alive;
    // they are never included in JobState or a pollable snapshot.
    foreground_senders: Vec<mpsc::SyncSender<Arc<crate::server::execute::ShellExecution>>>,
    // #1876: identical launches coalesce onto one entry, so several foreground
    // calls can wait on it at once. A detached/background owner keeps it pollable.
    foreground_waiters: usize,
    background_owned: bool,
    telemetry_recorded: bool,
}

static JOBS: LazyLock<Mutex<HashMap<String, Job>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
static NEXT_GENERATION: AtomicU64 = AtomicU64::new(1);

/// How often a foreground run reports progress to the MCP client (#1173).
const TICK: Duration = Duration::from_secs(5);

fn prune_finished_jobs(jobs: &mut HashMap<String, Job>, now: Instant) {
    prune_finished_jobs_with_limits(
        jobs,
        now,
        MAX_RETAINED_COMPLETED_JOBS,
        MAX_RETAINED_COMPLETED_BYTES,
    );
}

fn prune_finished_jobs_with_limits(
    jobs: &mut HashMap<String, Job>,
    now: Instant,
    max_completed_jobs: usize,
    max_completed_bytes: usize,
) {
    jobs.retain(|_, job| {
        job.foreground_waiters > 0
            || job
                .finished_at
                .is_none_or(|finished_at| now.duration_since(finished_at) < COMPLETED_JOB_TTL)
    });

    let mut completed: Vec<_> = jobs
        .iter()
        .filter(|(_, job)| job.foreground_waiters == 0)
        .filter_map(|(id, job)| {
            let finished_at = job.finished_at?;
            let output_bytes = match &job.state {
                JobState::Completed { output, .. } | JobState::Cancelled { output } => output.len(),
                JobState::Running { .. } => 0,
            };
            Some((finished_at, id.clone(), output_bytes))
        })
        .collect();
    completed.sort_unstable_by_key(|(finished_at, _, _)| *finished_at);

    let mut retained_bytes = completed.iter().map(|(_, _, bytes)| bytes).sum::<usize>();
    let mut retained_jobs = completed.len();
    for (_, id, output_bytes) in completed {
        if retained_jobs <= max_completed_jobs && retained_bytes <= max_completed_bytes {
            break;
        }
        if retained_jobs == 1 {
            break;
        }
        jobs.remove(&id);
        retained_jobs -= 1;
        retained_bytes = retained_bytes.saturating_sub(output_bytes);
    }
}

pub fn start(
    command: String,
    cwd: String,
    extra_env: std::collections::HashMap<String, String>,
    timeout_ms: Option<u64>,
) -> String {
    start_inner(command, cwd, extra_env, timeout_ms, false)
}

fn start_inner(
    command: String,
    cwd: String,
    extra_env: std::collections::HashMap<String, String>,
    timeout_ms: Option<u64>,
    foreground: bool,
) -> String {
    start_inner_with_sender(command, cwd, extra_env, timeout_ms, foreground, None)
}

fn start_inner_with_sender(
    command: String,
    cwd: String,
    extra_env: std::collections::HashMap<String, String>,
    timeout_ms: Option<u64>,
    foreground: bool,
    output_sender: Option<mpsc::SyncSender<Arc<crate::server::execute::ShellExecution>>>,
) -> String {
    // IDs are content-addressed so tool responses stay deterministic (#498).
    // An identical in-flight launch coalesces onto the same job instead of
    // creating duplicate expensive builds/tests.
    let mut env_entries: Vec<_> = extra_env.iter().collect();
    env_entries.sort_unstable_by(|a, b| a.0.cmp(b.0));
    let env_key = env_entries
        .into_iter()
        .map(|(key, value)| format!("{key}={value}"))
        .collect::<Vec<_>>()
        .join("\0");
    let material = format!(
        "{command}\0{cwd}\0{}\0{env_key}",
        timeout_ms.unwrap_or_default()
    );
    let id = format!(
        "shell_{}",
        &blake3::hash(material.as_bytes()).to_hex()[..16]
    );
    let cancel = Arc::new(AtomicBool::new(false));
    let worker_cancel = Arc::clone(&cancel);
    {
        let mut jobs = JOBS
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        prune_finished_jobs(&mut jobs, Instant::now());
        // A finished entry that a foreground waiter has not read yet is joined
        // too (#1876): replacing it would hand that waiter's release to the new
        // job, which then disappears under its own waiter.
        if let Some(job) = jobs.get_mut(&id)
            && (matches!(job.state, JobState::Running { .. }) || job.foreground_waiters > 0)
        {
            if foreground {
                job.foreground_waiters += 1;
                if matches!(job.state, JobState::Running { .. })
                    && job.foreground_senders.len() < MAX_FOREGROUND_OUTPUT_RECEIVERS
                    && let Some(sender) = output_sender
                {
                    job.foreground_senders.push(sender);
                }
            } else {
                job.background_owned = true;
            }
            return id;
        }
        jobs.insert(
            id.clone(),
            Job {
                cancel,
                state: JobState::Running {
                    output: String::new(),
                },
                finished_at: None,
                generation: NEXT_GENERATION.fetch_add(1, Ordering::Relaxed),
                actual_exit_code: 0,
                completeness: lean_ctx::core::context_admission::OutputCompleteness::complete(),
                output_withheld: false,
                notice: None,
                class: lean_ctx::core::patterns::output_class(&command),
                foreground_senders: output_sender.into_iter().collect(),
                foreground_waiters: usize::from(foreground),
                background_owned: !foreground,
                telemetry_recorded: false,
            },
        );
    }

    let worker_id = id.clone();
    let project_root = lean_ctx::core::policy::runtime::detached_project_root(Path::new(&cwd)).ok();
    std::thread::spawn(move || {
        let execution = crate::server::execute::execute_command_with_env_cancellable_captured(
            &command,
            &cwd,
            &extra_env,
            timeout_ms,
            Some(&worker_cancel),
            // #1113/#1173: bounded by output, not wall clock — a monitor loop
            // emitting a line every 45s must survive. See `idle_keyed`.
            true,
        );
        let store = lean_ctx::core::context_admission::stores::StoreAdmission::current();
        let stored_output = project_root.as_deref().and_then(|root| {
            let root = root.to_str()?;
            lean_ctx::core::policy::runtime::with_project_source_view(root, || {
                // Background output is project-policy filtered before it is
                // retained. Preserve the existing audit of the original text
                // at this boundary: later tool-pipeline auditing sees only the
                // admitted text (or a content-free withheld marker).
                let policy_outcome = lean_ctx::core::policy::runtime::active().map(|policy| {
                    lean_ctx::core::policy::content::evaluate_text(&execution.output, &policy)
                });
                // #2164: a pipe still held by a process the command left
                // running does not cost the job its output.
                let stored =
                    store.admit_background_output(&execution.output, execution.completeness);
                if let Some(outcome) = policy_outcome
                    && !outcome.audit.is_empty()
                    && (outcome.blocked
                        || outcome.text != execution.output
                        || stored.as_deref() != Some(execution.output.as_str()))
                {
                    crate::server::policy_guard::audit_filter(
                        "ctx_shell",
                        &outcome.audit,
                        outcome.blocked,
                    );
                }
                stored
            })
            .ok()
            .flatten()
        });
        let mut jobs = JOBS
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let Some(job) = jobs.get_mut(&worker_id) else {
            return;
        };
        // Check epoch while the job insertion lock is held, so a policy change
        // while waiting for JOBS cannot repopulate terminal state.
        let stored_output = stored_output.filter(|_| store.is_current());
        let cancelled = worker_cancel.load(Ordering::Acquire);
        let foreground_result = Arc::new(execution);
        for sender in job.foreground_senders.drain(..) {
            let _ = sender.send(Arc::clone(&foreground_result));
        }
        let output_withheld = stored_output.is_none();
        let output = stored_output.unwrap_or_default();
        let exit_code = foreground_result.exit_code;
        job.actual_exit_code = exit_code;
        job.completeness = foreground_result.completeness;
        job.output_withheld = output_withheld;
        job.notice = foreground_result.notice;
        job.state = if cancelled {
            JobState::Cancelled { output }
        } else {
            JobState::Completed { output, exit_code }
        };
        job.finished_at = Some(Instant::now());
        // Only a background-owned job is counted here, so a failure nobody
        // polls still reaches telemetry. A foreground run is classified once by
        // the call pipeline, which knows that exit 1 with output (grep, diff,
        // test) is not an error; counting it here as well filed every failing
        // inline command twice, the second time as a LeanCTX-internal error.
        if !cancelled && exit_code != 0 && job.background_owned {
            let category = if exit_code == 124 {
                lean_ctx::core::telemetry_v2::ErrorCategory::Timeout
            } else {
                lean_ctx::core::telemetry_v2::ErrorCategory::Command
            };
            if lean_ctx::core::telemetry_aggregate::record_error_category(category).is_ok() {
                job.telemetry_recorded = true;
            }
        }
        prune_finished_jobs(&mut jobs, Instant::now());
    });
    id
}

/// Outcome of a foreground run that is allowed to detach on a soft cap.
pub enum ForegroundResult {
    /// The command finished within the soft cap; output is returned inline.
    /// Other coalesced foreground/background callers retain their result.
    Finished { output: String, exit_code: i32 },
    /// The command was still running at the soft cap and was left running as a
    /// pollable background job (#1106).
    Detached { job_id: String },
}

/// Legacy tuple-shaped foreground result for existing callers.
pub fn run_foreground_or_detach(
    command: String,
    cwd: String,
    extra_env: std::collections::HashMap<String, String>,
    timeout_ms: Option<u64>,
    soft_cap: Duration,
    on_tick: Option<&dyn Fn(Duration)>,
    cancel: Option<&tokio_util::sync::CancellationToken>,
) -> ForegroundResult {
    match run_foreground_or_detach_captured(
        command, cwd, extra_env, timeout_ms, soft_cap, on_tick, cancel,
    ) {
        CapturedForegroundResult::Finished {
            output,
            exit_code,
            notice,
            ..
        } => ForegroundResult::Finished {
            // Nothing shortens the text on this path: the notice goes in
            // front here.
            output: crate::server::execute::with_notice(notice, output),
            exit_code,
        },
        CapturedForegroundResult::Detached { job_id } => ForegroundResult::Detached { job_id },
    }
}

/// Process output returned only while the original foreground call remains alive.
pub(crate) enum CapturedForegroundResult {
    Finished {
        output: String,
        exit_code: i32,
        completeness: lean_ctx::core::context_admission::OutputCompleteness,
        retained_withheld: bool,
        /// Not part of `output`; see `ShellExecution::notice`.
        notice: Option<&'static str>,
    },
    Detached {
        job_id: String,
    },
}

/// Run a managed job with a one-shot channel for live foreground delivery.
/// Detached callers drop that receiver; later polls can see only stored text.
pub(crate) fn run_foreground_or_detach_captured(
    command: String,
    cwd: String,
    extra_env: std::collections::HashMap<String, String>,
    timeout_ms: Option<u64>,
    soft_cap: Duration,
    on_tick: Option<&dyn Fn(Duration)>,
    cancel: Option<&tokio_util::sync::CancellationToken>,
) -> CapturedForegroundResult {
    let (sender, receiver) = mpsc::sync_channel(1);
    let id = start_inner_with_sender(command, cwd, extra_env, timeout_ms, true, Some(sender));
    let started = Instant::now();
    let deadline = started + soft_cap;
    let mut next_tick = started + TICK;
    loop {
        if let Ok(execution) = receiver.try_recv() {
            release_foreground(&id, false);
            return CapturedForegroundResult::Finished {
                output: execution.output.clone(),
                exit_code: execution.exit_code,
                completeness: execution.completeness,
                retained_withheld: false,
                notice: execution.notice,
            };
        }
        if let Some(snapshot) = snapshot(&id) {
            match snapshot.state {
                JobState::Completed { output, exit_code } => {
                    release_foreground(&id, false);
                    return CapturedForegroundResult::Finished {
                        output,
                        exit_code,
                        completeness: snapshot.completeness,
                        retained_withheld: snapshot.output_withheld,
                        notice: snapshot.notice,
                    };
                }
                JobState::Cancelled { output } => {
                    release_foreground(&id, false);
                    return CapturedForegroundResult::Finished {
                        output,
                        exit_code: snapshot.actual_exit_code,
                        completeness: snapshot.completeness,
                        retained_withheld: snapshot.output_withheld,
                        notice: snapshot.notice,
                    };
                }
                JobState::Running { .. } => {}
            }
        }
        if cancel.is_some_and(tokio_util::sync::CancellationToken::is_cancelled) {
            release_foreground(&id, true);
            return CapturedForegroundResult::Detached { job_id: id };
        }
        let now = Instant::now();
        if now >= deadline {
            release_foreground(&id, true);
            return CapturedForegroundResult::Detached { job_id: id };
        }
        if let Some(tick) = on_tick
            && now >= next_tick
        {
            tick(started.elapsed());
            next_tick = now + TICK;
        }
        std::thread::sleep(Duration::from_millis(50));
    }
}

/// One foreground waiter is done with `id` (#1876). A detached waiter hands the
/// id to its caller, so the entry becomes background-owned and stays pollable.
/// Otherwise the entry is removed once no waiter and no background caller is
/// left; a waiter that still polls it must not find it gone.
fn release_foreground(id: &str, detached: bool) {
    let mut jobs = JOBS
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let Some(job) = jobs.get_mut(id) else {
        return;
    };
    job.foreground_waiters = job.foreground_waiters.saturating_sub(1);
    if detached {
        job.background_owned = true;
    } else if job.foreground_waiters == 0 && !job.background_owned {
        jobs.remove(id);
    }
}

#[cfg(test)]
pub(crate) fn remove_for_test(id: &str) {
    JOBS.lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .remove(id);
}

/// Jobs whose command is still running. An idle MCP server with one of these
/// is not idle: it holds the child and the output the agent will poll for.
pub fn running_count() -> usize {
    JOBS.lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .values()
        .filter(|job| matches!(job.state, JobState::Running { .. }))
        .count()
}

#[derive(Debug, Clone)]
pub(crate) struct JobSnapshot {
    pub(crate) state: JobState,
    pub(crate) generation: u64,
    pub(crate) completeness: lean_ctx::core::context_admission::OutputCompleteness,
    pub(crate) output_withheld: bool,
    pub(crate) actual_exit_code: i32,
    /// The job's fixed notice, if it has one; not part of the state's text.
    pub(crate) notice: Option<&'static str>,
    /// The class the command's output is admitted with (pz21 §1.8).
    pub(crate) class: lean_ctx::core::context_admission::origin::SourceClass,
}

/// A running job exposes state only; terminal text was storage-admitted before
/// it entered the pollable job record.
pub(crate) fn snapshot(id: &str) -> Option<JobSnapshot> {
    let jobs = JOBS
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let job = jobs.get(id)?;
    Some(JobSnapshot {
        state: job.state.clone(),
        generation: job.generation,
        completeness: job.completeness,
        output_withheld: job.output_withheld,
        actual_exit_code: job.actual_exit_code,
        notice: job.notice,
        class: job.class.clone(),
    })
}

pub fn status(id: &str) -> Option<JobState> {
    snapshot(id).map(|snapshot| snapshot.state)
}

/// Replace retained terminal text only if the caller still holds the exact
/// generation and text it inspected. A policy change can therefore scrub a job
/// without racing a reused content-addressed id or a newer poll.
pub(crate) fn replace_retained_output_if_generation(
    id: &str,
    generation: u64,
    expected: &str,
    expected_epoch: u64,
    replacement: Option<&str>,
) -> bool {
    let mut jobs = JOBS
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let Some(job) = jobs.get_mut(id) else {
        return false;
    };
    if job.generation != generation
        || (replacement.is_some()
            && lean_ctx::core::context_admission::stores::current_epoch() != expected_epoch)
    {
        return false;
    }
    let current = match &job.state {
        JobState::Completed { output, .. } | JobState::Cancelled { output } => output,
        JobState::Running { .. } => return false,
    };
    if current != expected {
        return false;
    }
    match &mut job.state {
        JobState::Completed { output, .. } | JobState::Cancelled { output } => {
            output.clear();
            if let Some(replacement) = replacement {
                output.push_str(replacement);
            }
        }
        JobState::Running { .. } => return false,
    }
    job.output_withheld = replacement.is_none();
    true
}

/// Persist one aggregate for a terminal failed job, then mark that concrete
/// retained job instance. A later execution may reuse the content-addressed ID,
/// but replaces the `Job` and therefore starts with a fresh marker.
pub fn record_error_telemetry_once(
    id: &str,
    category: lean_ctx::core::telemetry_v2::ErrorCategory,
) -> Result<bool, String> {
    let mut jobs = JOBS
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let Some(job) = jobs.get_mut(id) else {
        return Ok(false);
    };
    if job.telemetry_recorded
        || !matches!(job.state, JobState::Completed { exit_code, .. } if exit_code != 0)
    {
        return Ok(false);
    }
    lean_ctx::core::telemetry_aggregate::record_error_category(category)?;
    job.telemetry_recorded = true;
    Ok(true)
}

pub(crate) fn cancel_snapshot(id: &str) -> Option<JobSnapshot> {
    {
        let mut jobs = JOBS
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let job = jobs.get_mut(id)?;
        if matches!(job.state, JobState::Running { .. }) {
            job.cancel.store(true, Ordering::Release);
        }
    }
    snapshot(id)
}

pub fn cancel(id: &str) -> Option<JobState> {
    cancel_snapshot(id).map(|snapshot| snapshot.state)
}

#[cfg(test)]
mod tests {
    use super::{
        ForegroundResult, JobState, TICK, cancel, run_foreground_or_detach, start, status,
    };
    use std::time::Duration;

    /// A finished job's text is storage-admitted, on the job's own thread,
    /// under the process's gateway configuration and project policy, and is
    /// dropped when that configuration changes meanwhile. A test that reads
    /// the retained text holds both still: its own data directory (which
    /// serializes every test that changes the environment) and no project
    /// policy — environment first, policy second, as everywhere.
    fn retained_text_held_still() -> (
        lean_ctx::core::data_dir::IsolatedDataDir,
        lean_ctx::core::policy::runtime::TestPolicyOverride,
    ) {
        let data = lean_ctx::core::data_dir::isolated_data_dir();
        let community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
        (data, community)
    }

    fn completed_job(exit_code: i32) -> super::Job {
        super::Job {
            cancel: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
            state: JobState::Completed {
                output: "opaque output".into(),
                exit_code,
            },
            finished_at: Some(std::time::Instant::now()),
            generation: 0,
            actual_exit_code: 0,
            completeness: lean_ctx::core::context_admission::OutputCompleteness::complete(),
            output_withheld: false,
            notice: None,
            class: lean_ctx::core::context_admission::carrier::INSTANCE,
            foreground_senders: Vec::new(),
            foreground_waiters: 0,
            background_owned: true,
            telemetry_recorded: false,
        }
    }

    #[test]
    fn retained_output_replacement_checks_epoch_under_the_job_lock() {
        let id = "shell_epoch_replacement_test";
        let mut job = completed_job(0);
        job.generation = 41;
        super::JOBS
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .insert(id.into(), job);
        let epoch = lean_ctx::core::context_admission::stores::current_epoch();

        assert!(
            !super::replace_retained_output_if_generation(
                id,
                41,
                "opaque output",
                epoch.wrapping_add(1),
                Some("stale output"),
            ),
            "an old policy snapshot cannot repopulate a terminal job"
        );
        assert!(matches!(
            super::status(id),
            Some(JobState::Completed { output, .. }) if output == "opaque output"
        ));

        assert!(
            super::replace_retained_output_if_generation(
                id,
                41,
                "opaque output",
                epoch.wrapping_add(1),
                None,
            ),
            "scrubbing remains allowed after an epoch change"
        );
        assert!(matches!(
            super::status(id),
            Some(JobState::Completed { output, .. }) if output.is_empty()
        ));
        super::remove_for_test(id);
    }

    #[test]
    #[serial_test::serial]
    fn failed_job_telemetry_is_once_per_execution_even_when_id_is_reused() {
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();
        let id = "shell_same_content";
        super::JOBS
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .insert(id.into(), completed_job(2));

        assert!(
            super::record_error_telemetry_once(
                id,
                lean_ctx::core::telemetry_v2::ErrorCategory::Internal
            )
            .expect("record first execution")
        );
        assert!(
            !super::record_error_telemetry_once(
                id,
                lean_ctx::core::telemetry_v2::ErrorCategory::Internal
            )
            .expect("deduplicate repeated poll")
        );

        super::JOBS
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .insert(id.into(), completed_job(2));
        assert!(
            super::record_error_telemetry_once(
                id,
                lean_ctx::core::telemetry_v2::ErrorCategory::Internal
            )
            .expect("record replacement execution")
        );

        let batch = lean_ctx::core::telemetry_aggregate::preview_daily_batch().expect("preview");
        let count = batch
            .events
            .iter()
            .find_map(|envelope| match &envelope.event {
                lean_ctx::core::telemetry_v2::TelemetryEventV2::ErrorCategoryAggregate(metrics)
                    if metrics.category
                        == lean_ctx::core::telemetry_v2::ErrorCategory::Internal =>
                {
                    Some(metrics.count)
                }
                _ => None,
            });
        assert_eq!(count, Some(2));
        super::remove_for_test(id);
    }

    fn queued_error_count(category: lean_ctx::core::telemetry_v2::ErrorCategory) -> Option<u64> {
        // A preview fails fast while another test's send holds the state lock;
        // that is transient, so wait it out instead of failing this test.
        let mut attempt = 0;
        let batch = loop {
            match lean_ctx::core::telemetry_aggregate::preview_daily_batch() {
                Ok(batch) => break batch,
                Err(_) if attempt < 100 => {
                    attempt += 1;
                    std::thread::sleep(Duration::from_millis(20));
                }
                Err(error) => panic!("preview: {error}"),
            }
        };
        batch
            .events
            .iter()
            .find_map(|envelope| match &envelope.event {
                lean_ctx::core::telemetry_v2::TelemetryEventV2::ErrorCategoryAggregate(metrics)
                    if metrics.category == category =>
                {
                    Some(metrics.count)
                }
                _ => None,
            })
    }

    /// A failing inline command is classified once, by the call pipeline; the
    /// worker counts only background-owned jobs, and as a failed command, never
    /// as a LeanCTX-internal error.
    #[test]
    #[serial_test::serial]
    #[cfg_attr(windows, ignore)]
    fn worker_counts_only_background_failures_and_as_commands() {
        use lean_ctx::core::telemetry_v2::ErrorCategory;
        // The daily error counters are process-global: any parallel test whose
        // failing command or detached job lands while this one holds the
        // isolated data dir is counted here too (CI saw 2 instead of 1). Run
        // the exact scenario alone in a fresh test process so the counts are
        // this worker's own, like the MCP cancellation regression does.
        if std::env::var_os("LEAN_CTX_BACKGROUND_TELEMETRY_SUBPROCESS").is_none() {
            let environment = tempfile::tempdir().unwrap();
            let mut command = std::process::Command::new(std::env::current_exe().unwrap());
            command
                .args([
                    "--exact",
                    "server::background_shell::tests::worker_counts_only_background_failures_and_as_commands",
                    "--nocapture",
                    "--test-threads=1",
                ])
                .env("LEAN_CTX_BACKGROUND_TELEMETRY_SUBPROCESS", "1");
            for name in [
                "LEAN_CTX_DATA_DIR",
                "LEAN_CTX_CONFIG_DIR",
                "LEAN_CTX_STATE_DIR",
                "LEAN_CTX_CACHE_DIR",
            ] {
                command.env(name, environment.path());
            }
            let output = command.output().unwrap();
            let stdout = String::from_utf8_lossy(&output.stdout);
            assert!(
                output.status.success() && stdout.contains("1 passed"),
                "{stdout}\n{}",
                String::from_utf8_lossy(&output.stderr)
            );
            return;
        }
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();

        let result = run_foreground_or_detach(
            "exit 3".to_string(),
            ".".to_string(),
            std::collections::HashMap::default(),
            Some(10_000),
            Duration::from_secs(10),
            None,
            None,
        );
        assert!(matches!(
            result,
            ForegroundResult::Finished { exit_code: 3, .. }
        ));
        assert_eq!(queued_error_count(ErrorCategory::Command), None);
        assert_eq!(queued_error_count(ErrorCategory::Internal), None);

        let id = start(
            "exit 4".to_string(),
            ".".to_string(),
            std::collections::HashMap::default(),
            Some(10_000),
        );
        for _ in 0..200 {
            if matches!(status(&id), Some(JobState::Completed { .. })) {
                break;
            }
            std::thread::sleep(Duration::from_millis(25));
        }
        assert!(matches!(
            status(&id),
            Some(JobState::Completed { exit_code: 4, .. })
        ));
        assert_eq!(queued_error_count(ErrorCategory::Command), Some(1));
        assert_eq!(queued_error_count(ErrorCategory::Internal), None);
        super::remove_for_test(&id);
    }

    #[test]
    fn completed_job_retention_is_bounded() {
        let now = std::time::Instant::now();
        let mut jobs = std::collections::HashMap::new();
        for index in 0..=super::MAX_RETAINED_COMPLETED_JOBS {
            jobs.insert(
                format!("job_{index}"),
                super::Job {
                    cancel: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
                    state: JobState::Completed {
                        output: "x".repeat(1024),
                        exit_code: 0,
                    },
                    finished_at: Some(
                        now.checked_sub(Duration::from_secs((index + 1) as u64))
                            .unwrap(),
                    ),
                    generation: 0,
                    actual_exit_code: 0,
                    completeness: lean_ctx::core::context_admission::OutputCompleteness::complete(),
                    output_withheld: false,
                    notice: None,
                    class: lean_ctx::core::context_admission::carrier::INSTANCE,
                    foreground_senders: Vec::new(),
                    foreground_waiters: 0,
                    background_owned: true,
                    telemetry_recorded: false,
                },
            );
        }
        jobs.insert(
            "expired".to_string(),
            super::Job {
                cancel: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
                state: JobState::Completed {
                    output: "expired".to_string(),
                    exit_code: 0,
                },
                finished_at: Some(
                    now.checked_sub(super::COMPLETED_JOB_TTL + Duration::from_secs(1))
                        .unwrap(),
                ),
                generation: 0,
                actual_exit_code: 0,
                completeness: lean_ctx::core::context_admission::OutputCompleteness::complete(),
                output_withheld: false,
                notice: None,
                class: lean_ctx::core::context_admission::carrier::INSTANCE,
                foreground_senders: Vec::new(),
                foreground_waiters: 0,
                background_owned: true,
                telemetry_recorded: false,
            },
        );

        super::prune_finished_jobs(&mut jobs, now);

        assert_eq!(jobs.len(), super::MAX_RETAINED_COMPLETED_JOBS);
        assert!(!jobs.contains_key("expired"));
        assert!(!jobs.contains_key(&format!("job_{}", super::MAX_RETAINED_COMPLETED_JOBS)));
    }

    #[test]
    fn completed_job_output_bytes_are_bounded() {
        let now = std::time::Instant::now();
        let mut jobs = std::collections::HashMap::new();
        for index in 0..3 {
            jobs.insert(
                format!("job_{index}"),
                super::Job {
                    cancel: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
                    state: JobState::Completed {
                        output: "x".repeat(8),
                        exit_code: 0,
                    },
                    finished_at: Some(
                        now.checked_sub(Duration::from_secs((3 - index) as u64))
                            .unwrap(),
                    ),
                    generation: 0,
                    actual_exit_code: 0,
                    completeness: lean_ctx::core::context_admission::OutputCompleteness::complete(),
                    output_withheld: false,
                    notice: None,
                    class: lean_ctx::core::context_admission::carrier::INSTANCE,
                    foreground_senders: Vec::new(),
                    foreground_waiters: 0,
                    background_owned: true,
                    telemetry_recorded: false,
                },
            );
        }

        super::prune_finished_jobs_with_limits(&mut jobs, now, 10, 16);

        assert_eq!(jobs.len(), 2);
        assert!(!jobs.contains_key("job_0"));
    }

    #[test]
    #[cfg_attr(windows, ignore)]
    fn foreground_run_finishing_within_cap_returns_inline() {
        let (_data, _community) = retained_text_held_still();
        let result = run_foreground_or_detach(
            "printf FG_OK".to_string(),
            ".".to_string(),
            std::collections::HashMap::default(),
            Some(10_000),
            Duration::from_secs(10),
            None,
            None,
        );
        match result {
            ForegroundResult::Finished { output, exit_code } => {
                assert_eq!(exit_code, 0);
                assert!(output.contains("FG_OK"), "got: {output}");
            }
            ForegroundResult::Detached { .. } => panic!("fast command should not detach"),
        }
    }

    #[test]
    #[cfg_attr(windows, ignore)]
    fn foreground_run_exceeding_cap_detaches_to_pollable_job() {
        let result = run_foreground_or_detach(
            "sleep 5; printf SLOW_OK".to_string(),
            ".".to_string(),
            std::collections::HashMap::default(),
            Some(10_000),
            Duration::from_millis(100),
            None,
            None,
        );
        let ForegroundResult::Detached { job_id } = result else {
            panic!("slow command should detach");
        };
        assert!(job_id.starts_with("shell_"));
        // The returned id must resolve via status — the core #1106 guarantee.
        assert!(status(&job_id).is_some());
        cancel(&job_id);
    }

    /// #1173: a `timeout_ms` far beyond the soft cap must NOT keep the command
    /// in the foreground. Raising the cap past the MCP host's ~120s abort is
    /// what stranded results behind an unresolvable task id; the caller must
    /// still get a real `shell_*` job id at the cap.
    #[test]
    #[cfg_attr(windows, ignore)]
    fn large_timeout_ms_still_detaches_at_the_soft_cap() {
        let result = run_foreground_or_detach(
            "sleep 5; printf NEVER_INLINE".to_string(),
            ".".to_string(),
            std::collections::HashMap::default(),
            Some(600_000),
            Duration::from_millis(100),
            None,
            None,
        );
        let ForegroundResult::Detached { job_id } = result else {
            panic!("timeout_ms must not extend the foreground wait");
        };
        assert!(status(&job_id).is_some());
        cancel(&job_id);
    }

    /// #1173: a foreground run reports progress while it waits, so a slow
    /// command is distinguishable from a hang. Deliberately slower than the
    /// other tests here — it has to outlive one real [`TICK`].
    #[test]
    #[cfg_attr(windows, ignore)]
    fn foreground_run_reports_progress_while_waiting() {
        let ticks = std::sync::atomic::AtomicUsize::new(0);
        let tick = |elapsed: Duration| {
            assert!(elapsed >= TICK, "tick must report real elapsed time");
            ticks.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        };
        let result = run_foreground_or_detach(
            "sleep 30".to_string(),
            ".".to_string(),
            std::collections::HashMap::default(),
            Some(60_000),
            TICK + Duration::from_millis(500),
            Some(&tick),
            None,
        );
        let ForegroundResult::Detached { job_id } = result else {
            panic!("slow command should detach");
        };
        cancel(&job_id);
        assert!(
            ticks.load(std::sync::atomic::Ordering::Relaxed) >= 1,
            "no progress reported during a {}s+ foreground wait",
            TICK.as_secs()
        );
    }

    /// #1781: a client-side cancellation must free the waiting thread long
    /// before the soft cap. The assertion is on *elapsed time*, not just on the
    /// `Detached` shape — detaching at the cap returns the same variant, so the
    /// shape alone would pass even with the check removed.
    #[test]
    #[cfg_attr(windows, ignore)]
    fn cancelled_token_detaches_without_waiting_for_the_soft_cap() {
        let token = tokio_util::sync::CancellationToken::new();
        token.cancel();
        let started = std::time::Instant::now();
        let result = run_foreground_or_detach(
            // Distinct from the progress test's `sleep 30`: identical commands
            // coalesce onto one job (#1876), so a shared one would be cancelled
            // under the other test's feet.
            "sleep 31".to_string(),
            ".".to_string(),
            std::collections::HashMap::default(),
            Some(60_000),
            Duration::from_mins(1),
            None,
            Some(&token),
        );
        let elapsed = started.elapsed();
        let ForegroundResult::Detached { job_id } = result else {
            panic!("a cancelled request must detach instead of waiting it out");
        };
        cancel(&job_id);
        assert!(
            elapsed < Duration::from_secs(5),
            "cancelled wait took {elapsed:?} against a one-minute cap — token not observed"
        );
    }

    #[test]
    #[cfg_attr(windows, ignore)]
    fn background_job_runs_past_request_and_can_be_observed() {
        let (_data, _community) = retained_text_held_still();
        let id = start(
            "sleep 0.1; printf BG_JOB_OK".to_string(),
            ".".to_string(),
            std::collections::HashMap::default(),
            Some(10_000),
        );
        assert!(matches!(status(&id), Some(JobState::Running { .. })));
        for _ in 0..40 {
            if let Some(JobState::Completed { output, exit_code }) = status(&id) {
                assert_eq!(exit_code, 0);
                assert!(output.contains("BG_JOB_OK"), "got: {output}");
                return;
            }
            std::thread::sleep(Duration::from_millis(25));
        }
        panic!("background job did not complete");
    }

    #[test]
    #[cfg_attr(windows, ignore)]
    fn cancelling_background_job_returns_cancelled_state() {
        let id = start(
            "sleep 5".to_string(),
            ".".to_string(),
            std::collections::HashMap::default(),
            Some(10_000),
        );
        assert!(matches!(cancel(&id), Some(JobState::Running { .. })));
        for _ in 0..40 {
            if let Some(JobState::Cancelled { output }) = status(&id) {
                assert!(
                    !output.contains("[cancelled: command stopped on request]"),
                    "cancellation lifecycle belongs to the structured state, not retained output"
                );
                return;
            }
            std::thread::sleep(Duration::from_millis(25));
        }
        panic!("background job was not cancelled");
    }

    /// Running jobs must never retain or expose raw output; only terminal
    /// StoreAdmission-approved text can appear in a later poll.
    #[test]
    #[cfg_attr(windows, ignore)]
    fn running_background_job_status_never_exposes_live_output() {
        let id = start(
            "printf EARLY_LINE; sleep 3".to_string(),
            ".".to_string(),
            std::collections::HashMap::default(),
            Some(10_000),
        );
        let mut saw_running = false;
        for _ in 0..20 {
            if let Some(JobState::Running { output }) = status(&id) {
                saw_running = true;
                assert!(output.is_empty(), "running state retained: {output:?}");
                break;
            }
            std::thread::sleep(Duration::from_millis(10));
        }
        cancel(&id);
        assert!(
            saw_running,
            "job completed before running state was observed"
        );
    }

    /// #1876: identical foreground calls coalesce onto one job. The first to
    /// see it finish used to remove the shared entry, so every other caller
    /// polled a missing id until the host timed out. Each must get the output.
    #[test]
    #[cfg_attr(windows, ignore)]
    fn identical_concurrent_foreground_calls_all_observe_completion() {
        let (_data, _community) = retained_text_held_still();
        const CALLERS: usize = 3;
        let barrier = std::sync::Arc::new(std::sync::Barrier::new(CALLERS));
        let handles: Vec<_> = (0..CALLERS)
            .map(|_| {
                let barrier = std::sync::Arc::clone(&barrier);
                std::thread::spawn(move || {
                    barrier.wait();
                    run_foreground_or_detach(
                        "sleep 0.2; printf GH1876_ALL_OK".to_string(),
                        ".".to_string(),
                        std::collections::HashMap::default(),
                        Some(10_000),
                        Duration::from_secs(10),
                        None,
                        None,
                    )
                })
            })
            .collect();
        for handle in handles {
            match handle.join().expect("foreground caller panicked") {
                ForegroundResult::Finished { output, exit_code } => {
                    assert_eq!(exit_code, 0);
                    assert!(output.contains("GH1876_ALL_OK"), "got: {output}");
                }
                ForegroundResult::Detached { job_id } => {
                    cancel(&job_id);
                    panic!("a coalesced caller lost the shared job and detached");
                }
            }
        }
    }

    /// #1876: a foreground call that joins an explicit background job must
    /// not remove the entry the background caller still polls by id.
    #[test]
    #[cfg_attr(windows, ignore)]
    fn foreground_join_keeps_a_background_job_pollable() {
        let (_data, _community) = retained_text_held_still();
        let command = "sleep 0.2; printf GH1876_BG_OK".to_string();
        let id = start(
            command.clone(),
            ".".to_string(),
            std::collections::HashMap::default(),
            Some(10_000),
        );
        let result = run_foreground_or_detach(
            command,
            ".".to_string(),
            std::collections::HashMap::default(),
            Some(10_000),
            Duration::from_secs(10),
            None,
            None,
        );
        assert!(matches!(result, ForegroundResult::Finished { .. }));
        match status(&id) {
            Some(JobState::Completed { output, .. }) => assert!(output.contains("GH1876_BG_OK")),
            other => panic!("background job vanished after a foreground join: {other:?}"),
        }
        super::remove_for_test(&id);
    }

    /// #1876: a call that arrives after the job finished but before its waiter
    /// read the result must join that entry. Replacing it let the first
    /// waiter's release delete the second caller's job.
    #[test]
    #[cfg_attr(windows, ignore)]
    fn a_caller_joins_a_finished_entry_its_waiter_has_not_read() {
        // The retained text is storage-admitted under the process's gateway
        // configuration and dropped when that configuration changes while the
        // job finishes (`StoreAdmission::is_current`). This test reads the
        // retained text, so it holds the configuration still: its own data
        // directory (which serializes every test that changes one) and no
        // project policy.
        let _data = lean_ctx::core::data_dir::isolated_data_dir();
        let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
        let launch = || {
            super::start_inner(
                "printf GH1876_WINDOW".to_string(),
                ".".to_string(),
                std::collections::HashMap::default(),
                Some(10_000),
                true,
            )
        };
        let id = launch();
        let deadline = std::time::Instant::now() + Duration::from_secs(10);
        while !matches!(status(&id), Some(JobState::Completed { .. })) {
            assert!(std::time::Instant::now() < deadline, "job never completed");
            std::thread::sleep(Duration::from_millis(10));
        }

        assert_eq!(launch(), id);
        super::release_foreground(&id, false);
        match status(&id) {
            Some(JobState::Completed { output, .. }) => assert!(output.contains("GH1876_WINDOW")),
            other => panic!("the joined caller lost its result: {other:?}"),
        }
        super::release_foreground(&id, false);
        assert!(status(&id).is_none(), "the last waiter removes the entry");
    }

    /// #1876: fast identical commands finish inside one poll interval, which
    /// is where callers used to replace an unread entry.
    #[test]
    #[cfg_attr(windows, ignore)]
    fn fast_identical_concurrent_foreground_calls_all_return() {
        let (_data, _community) = retained_text_held_still();
        const CALLERS: usize = 8;
        for _ in 0..5 {
            let barrier = std::sync::Arc::new(std::sync::Barrier::new(CALLERS));
            let handles: Vec<_> = (0..CALLERS)
                .map(|_| {
                    let barrier = std::sync::Arc::clone(&barrier);
                    std::thread::spawn(move || {
                        barrier.wait();
                        run_foreground_or_detach(
                            "printf GH1876_FAST".to_string(),
                            ".".to_string(),
                            std::collections::HashMap::default(),
                            Some(10_000),
                            Duration::from_secs(5),
                            None,
                            None,
                        )
                    })
                })
                .collect();
            for handle in handles {
                match handle.join().expect("foreground caller panicked") {
                    ForegroundResult::Finished { output, .. } => {
                        assert!(output.contains("GH1876_FAST"), "got: {output}");
                    }
                    ForegroundResult::Detached { job_id } => {
                        cancel(&job_id);
                        panic!("a fast identical caller lost its job and detached");
                    }
                }
            }
        }
    }

    #[test]
    fn pruning_keeps_finished_jobs_that_a_waiter_has_not_read() {
        let now = std::time::Instant::now();
        let mut jobs = std::collections::HashMap::new();
        jobs.insert(
            "waited".to_string(),
            super::Job {
                cancel: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
                state: JobState::Completed {
                    output: "x".repeat(64),
                    exit_code: 0,
                },
                finished_at: Some(
                    now.checked_sub(super::COMPLETED_JOB_TTL + Duration::from_secs(1))
                        .unwrap(),
                ),
                generation: 0,
                actual_exit_code: 0,
                completeness: lean_ctx::core::context_admission::OutputCompleteness::complete(),
                output_withheld: false,
                notice: None,
                class: lean_ctx::core::context_admission::carrier::INSTANCE,
                foreground_senders: Vec::new(),
                foreground_waiters: 1,
                background_owned: false,
                telemetry_recorded: false,
            },
        );

        super::prune_finished_jobs_with_limits(&mut jobs, now, 0, 0);

        assert!(jobs.contains_key("waited"));
    }
}