scv-tools 0.4.11

//! [`NativeAgentTool`]: one agent CLI process per turn, its stdout parsed
//! as it arrives, continuing a conversation through the CLI's own resume
//! flags when the adapter supports it.

use std::{
    ffi::OsString,
    io::Read as _,
    path::{Path, PathBuf},
    sync::Arc,
    time::Duration,
};

use async_trait::async_trait;
use scv_core::{ToolContext, ToolError, ToolOutput, ToolRisk};
use serde_json::Value;
use tokio::sync::Mutex;

use crate::{
    AgentAdapterConfig, AgentAuthPlan, DelegationContext,
    args::{Timeouts, bounded, parse_args, validate_process_args},
    delegate::{
        activity::{RunHeader, RunTransport},
        adapters::{self, ModeReport, OutputFormat, Resume},
        agent::{Accepts, Backend},
        conversation::{self, ConversationStore},
        mode,
        output::{self, AgentStream, RunExit, STDERR_TAIL_BYTES, TailBuffer, add_sign_in_hint},
        records::{self, DelegationGuard, DelegationRegistry},
        request::{
            AgentArgs, resolve_call_cwd, valid_effort, valid_model_name, validate_agent_cwd,
        },
        takeover::{self, TakeoverLaunch},
    },
    process::{ProcessSpec, child_pid, drain_output, spawn_process, supervise},
};

/// How long after a Codex thread starts its rollout may take to record the
/// turn's mode before the run stops unverified.
const ROLLOUT_WAIT: Duration = Duration::from_secs(30);
/// How often the rollout is read meanwhile.
const ROLLOUT_POLL: Duration = Duration::from_millis(200);

/// An agent run as one CLI process per turn.
pub(crate) struct NativeAgentTool {
    /// The adapter name, such as `codex`.
    name: String,
    command: String,
    /// The executable found on `PATH` or in the install directories; an
    /// agent that is not installed is not offered.
    pub(crate) resolved: Option<PathBuf>,
    args: Vec<String>,
    prompt_args: Vec<String>,
    full_permission_args: Option<Vec<String>>,
    /// The policy mode in effect for this agent, such as `auto`.
    policy: String,
    mode_report: ModeReport,
    /// It has a full-autonomy switch, so a mode to verify.
    mode_switch: bool,
    model_args: Vec<String>,
    effort_args: Vec<String>,
    environment: Vec<(OsString, OsString)>,
    timeouts: Timeouts,
    output_limit: usize,
    output: OutputFormat,
    resume: Resume,
    home: Option<PathBuf>,
    delegation: Option<DelegationContext>,
    conversations: Arc<ConversationStore>,
    auth: AgentAuthPlan,
    /// Where worktrees SCV created live, which a call may name as `cwd`.
    worktrees: Option<PathBuf>,
    /// How the owner takes a conversation over in its terminal, or why not.
    takeover: Result<TakeoverLaunch, String>,
}

impl NativeAgentTool {
    /// What this agent takes: `model` and `effort` when its adapter maps
    /// them to arguments, and `session` when its CLI can resume.
    pub(crate) fn accepts(&self) -> Accepts {
        Accepts {
            model: !self.model_args.is_empty(),
            effort: !self.effort_args.is_empty(),
            session: self.resume.is_supported(),
        }
    }

    /// The fixed arguments, validated model and effort selections, and the
    /// prompt arguments; the prompt is appended separately as the final argument.
    fn command_args(&self, args: &AgentArgs) -> Result<Vec<String>, ToolError> {
        validate_process_args(&args.prompt)?;
        self.timeouts.resolve(args.timeout_seconds)?;
        if let Some(cwd) = &args.cwd {
            validate_agent_cwd(cwd)?;
        }
        if let Some(session) = &args.session {
            if !self.resume.is_supported() {
                return Err(ToolError::invalid_arguments(format!(
                    "{} cannot continue a conversation; omit session to start a new one",
                    self.name
                )));
            }
            if !conversation::is_handle(session) {
                return Err(ToolError::invalid_arguments(format!(
                    "session {:?} is not a conversation handle; pass the `session` value an \
                     earlier {} call returned, or omit it to start a new conversation",
                    bounded(session, 80),
                    self.name
                )));
            }
        }
        // The prompt follows the flags as a positional argument, so it must
        // not be readable as one.
        if args.prompt.starts_with('-') {
            return Err(ToolError::invalid_arguments(
                "agent prompt must not start with '-'",
            ));
        }
        let mut command = self.args.clone();
        if let AgentAuthPlan::Claude { settings, .. } = &self.auth {
            command.push("--settings".into());
            command.push(serde_json::to_string(settings).map_err(|error| {
                ToolError::failed(format!("encode Claude Code settings: {error}"))
            })?);
            // The output format's arguments always follow, so the rule list
            // ends before the prompt.
            command.push("--disallowedTools".into());
            command.extend(
                adapters::CLAUDE_DENY_RULES
                    .iter()
                    .map(|rule| (*rule).to_owned()),
            );
        }
        command.extend(self.full_permission_args.iter().flatten().cloned());
        command.extend(self.output.args().iter().map(|arg| (*arg).to_owned()));
        for (field, value, template, placeholder) in [
            ("model", &args.model, &self.model_args, "{model}"),
            ("effort", &args.effort, &self.effort_args, "{effort}"),
        ] {
            let Some(value) = value else {
                continue;
            };
            if template.is_empty() {
                return Err(ToolError::invalid_arguments(format!(
                    "{} does not support selecting a {field}",
                    self.name
                )));
            }
            let valid = if field == "model" {
                valid_model_name(value)
            } else {
                valid_effort(value)
            };
            if !valid {
                return Err(ToolError::invalid_arguments(format!(
                    "invalid {field} {value:?}"
                )));
            }
            command.extend(template.iter().map(|part| part.replace(placeholder, value)));
        }
        command.extend(self.prompt_args.iter().cloned());
        Ok(command)
    }
    pub(crate) fn new(
        name: String,
        config: AgentAdapterConfig,
        timeouts: Timeouts,
        output_limit: usize,
        delegation: Option<DelegationContext>,
        conversations: Arc<ConversationStore>,
    ) -> Self {
        let resolved = adapters::resolve_agent_executable(&config.command, &config.search_dirs);
        let takeover = takeover::launch(
            &name,
            &config.command,
            &config.search_dirs,
            config.full_permission_args.is_some(),
            &config.environment,
            delegation.as_ref(),
        );
        Self {
            name,
            command: config.command,
            resolved,
            args: config.args,
            prompt_args: config.prompt_args,
            full_permission_args: config.full_permission_args,
            policy: config.policy,
            mode_report: config.mode_report,
            mode_switch: config.mode_switch,
            model_args: config.model_args,
            effort_args: config.effort_args,
            environment: config.environment,
            timeouts,
            output_limit,
            output: config.output,
            resume: config.resume,
            home: config.home,
            delegation,
            conversations,
            auth: config.auth,
            worktrees: None,
            takeover,
        }
    }

    /// Let calls name directories in the worktrees SCV created under `root`.
    pub(crate) fn with_worktrees(mut self, root: Option<PathBuf>) -> Self {
        self.worktrees = root;
        self
    }

    /// When SCV verifies this agent's mode.
    fn mode_check(&self) -> mode::Check {
        match self.mode_report {
            ModeReport::ClaudeInit => mode::Check::BeforeTools,
            ModeReport::CodexRollout => mode::Check::AfterStart,
            ModeReport::None if self.mode_switch => mode::Check::Unverified,
            ModeReport::None => mode::Check::NotApplicable,
        }
    }

    /// What the policy expects of the mode this agent's run reports, when
    /// it reports one.
    fn expected_mode(&self) -> Option<mode::Expected> {
        let full_mode = match self.mode_report {
            ModeReport::ClaudeInit => mode::CLAUDE_FULL_MODE,
            ModeReport::CodexRollout => mode::CODEX_FULL_MODE,
            ModeReport::None => return None,
        };
        Some(mode::Expected {
            agent: self.name.clone(),
            policy: self.policy.clone(),
            full_mode: full_mode.to_owned(),
            full: self.full_permission_args.is_some(),
        })
    }

    /// Arguments that name per-run files or IDs, placed after the fixed and
    /// format arguments. Returns the Codex last-message file to read and
    /// remove afterwards.
    fn run_args(&self, id: &str) -> (Vec<OsString>, Option<PathBuf>) {
        match self.output {
            OutputFormat::CodexJsonl => {
                let Some(dir) = self.home.as_ref().map(|home| home.join("tmp")) else {
                    return (Vec::new(), None);
                };
                if private_dir(&dir).is_err() {
                    return (Vec::new(), None);
                }
                let file = dir.join(format!("scv-{id}.last-message"));
                (vec!["-o".into(), file.clone().into()], Some(file))
            }
            OutputFormat::Text
            | OutputFormat::ClaudeStreamJson
            | OutputFormat::PiJson
            | OutputFormat::GeminiStreamJson
            | OutputFormat::CursorStreamJson
            | OutputFormat::AmpStreamJson => (Vec::new(), None),
        }
    }
}

/// `template` with `{session}` replaced by `session`.
fn session_args(template: &[&str], session: &str) -> Vec<OsString> {
    template
        .iter()
        .map(|part| OsString::from(part.replace("{session}", session)))
        .collect()
}

fn private_dir(dir: &Path) -> std::io::Result<()> {
    use std::os::unix::fs::PermissionsExt as _;
    std::fs::create_dir_all(dir)?;
    std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700))
}

/// Read and delete a file the CLI wrote for SCV, bounded to `limit` bytes.
fn take_file(path: &Path, limit: usize) -> Option<String> {
    let file = std::fs::File::open(path).ok();
    let _ = std::fs::remove_file(path);
    let mut bytes = Vec::new();
    std::io::Read::take(file?, u64::try_from(limit).unwrap_or(u64::MAX))
        .read_to_end(&mut bytes)
        .ok()?;
    let text = String::from_utf8_lossy(&bytes).trim().to_owned();
    (!text.is_empty()).then_some(text)
}

#[async_trait]
impl Backend for NativeAgentTool {
    fn risk(&self, arguments: &Value) -> Result<ToolRisk, ToolError> {
        let args: AgentArgs = parse_args(arguments)?;
        self.command_args(&args)?;
        Ok(ToolRisk::Delegate)
    }

    fn approval_summary(&self, arguments: &Value) -> Result<String, ToolError> {
        let args: AgentArgs = parse_args(arguments)?;
        let command_args = self.command_args(&args)?;
        let executable = self.resolved.as_ref().map_or_else(
            || self.command.as_str().into(),
            |path| path.display().to_string(),
        );
        let directory = args.cwd.as_deref().map_or_else(
            || "the workspace root".to_owned(),
            |cwd| format!("{:?} (inside the workspace)", bounded(cwd, 200)),
        );
        let conversation = args.session.as_deref().map_or_else(
            || " in a new conversation".to_owned(),
            |session| format!(", continuing conversation {session},"),
        );
        let timeout = self.timeouts.resolve(args.timeout_seconds)?;
        let duration = if args.no_timeout {
            "with no time limit".to_owned()
        } else {
            format!("for up to {} seconds", timeout.as_secs())
        };
        let permissions = mode::summary(
            &self.policy,
            self.full_permission_args.is_some(),
            self.mode_check(),
            false,
        );
        Ok(format!(
            "Launch {executable} with args {command_args:?} and prompt {:?}{conversation} in {directory} {duration}. The nested agent has your user permissions.{permissions}",
            bounded(&args.prompt, 2000),
        ))
    }

    fn busy(&self, arguments: &Value) -> Result<bool, ToolError> {
        let args: AgentArgs = parse_args(arguments)?;
        Ok(args
            .session
            .as_deref()
            .is_some_and(|h| self.conversations.is_busy(h)))
    }

    fn adoptable(&self) -> bool {
        self.resume.is_supported()
    }

    fn takeover(&self) -> Result<TakeoverLaunch, String> {
        self.takeover.clone()
    }

    async fn wait_idle(
        &self,
        arguments: &Value,
        cancellation: &tokio_util::sync::CancellationToken,
    ) -> Result<(), ToolError> {
        let args: AgentArgs = parse_args(arguments)?;
        if let Some(handle) = args.session.as_deref() {
            self.conversations.wait_idle(handle, cancellation).await?;
        }
        Ok(())
    }

    async fn execute(
        &self,
        arguments: Value,
        context: ToolContext,
    ) -> Result<ToolOutput, ToolError> {
        let args: AgentArgs = parse_args(&arguments)?;
        let command_args = self.command_args(&args)?;
        let deadline = self.timeouts.deadline(
            args.timeout_seconds,
            args.no_timeout,
            context.background_job,
        )?;
        let cwd = resolve_call_cwd(
            &context.workspace,
            self.worktrees.as_deref(),
            args.cwd.as_deref(),
            args.session
                .as_deref()
                .and_then(|handle| self.conversations.cwd(handle)),
        )?;
        let executable = self.resolved.as_ref().ok_or_else(|| {
            ToolError::unavailable(format!(
                "{} executable {:?} was not found on PATH or in the user's install directories",
                self.name, self.command
            ))
        })?;
        if let AgentAuthPlan::Claude { mode, home, .. } = &self.auth {
            let status = crate::stores::claude_status(*mode, home)
                .map_err(|error| ToolError::unavailable(format!("{error:#}")))?;
            if !status.ready {
                let billing = match mode {
                    adapters::AgentAuth::ClaudeApiKey => "API key billing",
                    adapters::AgentAuth::ClaudeSubscription => "Claude subscription",
                    adapters::AgentAuth::None => "unknown billing",
                };
                return Err(ToolError::unavailable(format!(
                    "{billing} identity for {} is unavailable: {}",
                    self.name,
                    status.lines.join("; ")
                )));
            }
        }
        let agent = self.name.as_str();
        let turn = if self.resume.is_supported() {
            Some(self.conversations.begin(
                agent,
                args.session.as_deref(),
                &cwd,
                self.resume.assigns_id(),
            )?)
        } else {
            None
        };
        let pending = self.delegation.as_ref().map(|delegation| {
            delegation.registry.begin_at(
                delegation.owner_depth(),
                agent,
                &delegation.session,
                &cwd,
                turn.as_ref().map(|turn| (turn.handle.as_str(), turn.turn)),
            )
        });
        let run_id = pending.as_ref().map_or_else(
            || uuid::Uuid::new_v4().simple().to_string(),
            |pending| pending.handle.clone(),
        );
        let activity =
            self.delegation
                .as_ref()
                .zip(pending.as_ref())
                .and_then(|(delegation, pending)| {
                    delegation.registry.open_activity(RunHeader {
                        handle: pending.handle.clone(),
                        agent: agent.to_owned(),
                        session: delegation.session.clone(),
                        transport: RunTransport::Cli,
                        cwd: cwd.clone(),
                        conversation: turn.as_ref().map(|turn| (turn.handle.clone(), turn.turn)),
                        streams: self.output != OutputFormat::Text,
                    })
                });
        if let Some(activity) = &activity {
            activity.turn_started(
                turn.as_ref().map(|turn| turn.handle.as_str()),
                turn.as_ref().map_or(1, |turn| turn.turn),
                &args.prompt,
                context.job.as_deref(),
            );
            // A session ID SCV chose, or the one a continued turn resumes.
            if let Some(vendor) = turn.as_ref().and_then(|turn| turn.vendor.as_deref()) {
                activity.set_vendor(vendor);
            }
        }
        let fixed_len = self.args.len()
            + if matches!(self.auth, AgentAuthPlan::Claude { .. }) {
                3 + adapters::CLAUDE_DENY_RULES.len()
            } else {
                0
            }
            + self.full_permission_args.as_ref().map_or(0, Vec::len)
            + self.output.args().len();
        let (fixed, rest) = command_args.split_at(fixed_len);
        let (selections, prompt_args) = rest.split_at(rest.len() - self.prompt_args.len());
        let (base, modes) = fixed.split_at(self.args.len());
        let continuing = args.session.is_some();
        let (run_args, last_message) = self.run_args(&run_id);
        let resume = match (self.resume, &turn) {
            (
                Resume::Supported {
                    start,
                    subcommand,
                    options,
                    positional,
                },
                Some(turn),
            ) => {
                let vendor = turn.vendor.as_deref().unwrap_or_default();
                if continuing {
                    (
                        subcommand.iter().map(OsString::from).collect(),
                        session_args(options, vendor),
                        session_args(positional, vendor),
                    )
                } else if turn.vendor.is_some() {
                    (Vec::new(), session_args(start, vendor), Vec::new())
                } else {
                    Default::default()
                }
            }
            _ => Default::default(),
        };
        let (subcommand, session_options, positional): (
            Vec<OsString>,
            Vec<OsString>,
            Vec<OsString>,
        ) = resume;
        let mut process_args: Vec<OsString> = base.iter().map(OsString::from).collect();
        process_args.extend(subcommand);
        process_args.extend(modes.iter().map(OsString::from));
        process_args.extend(run_args);
        process_args.extend(session_options);
        process_args.extend(selections.iter().map(OsString::from));
        process_args.extend(positional);
        process_args.extend(prompt_args.iter().map(OsString::from));
        process_args.push(OsString::from(args.prompt));
        let mut environment = self.environment.clone();
        match &pending {
            Some(pending) => environment.extend(pending.environment.iter().cloned()),
            None => environment.push((
                records::DEPTH_VARIABLE.into(),
                (records::current_depth() + 1).to_string().into(),
            )),
        }
        let registration = self
            .delegation
            .as_ref()
            .map(|delegation| Arc::clone(&delegation.registry))
            .zip(pending);
        let expected = self.expected_mode();
        let rollout = match (&self.home, self.mode_report) {
            (Some(home), ModeReport::CodexRollout) => Some(Rollout::new(
                home,
                turn.as_ref()
                    .and_then(|turn| turn.vendor.as_deref())
                    .filter(|_| continuing),
            )),
            _ => None,
        };
        let run = execute_agent_process(
            ProcessSpec {
                executable: executable.as_os_str().to_owned(),
                args: process_args,
                cwd,
                environment,
                clear_environment: false,
                sanitize_scv_environment: true,
                deadline,
                output_limit: self.output_limit,
            },
            AgentStream::new(self.output, self.output_limit)
                .with_auth(match &self.auth {
                    AgentAuthPlan::Claude { mode, .. } => *mode,
                    AgentAuthPlan::None => adapters::AgentAuth::None,
                })
                .with_progress(context.progress)
                .with_activity(activity.clone())
                .with_mode(expected),
            registration,
            rollout,
            context.cancellation,
        )
        .await;
        let fallback = last_message
            .as_deref()
            .and_then(|path| take_file(path, self.output_limit));
        let run = match run {
            Ok(run) => run,
            Err(error) => {
                if let Some(activity) = &activity {
                    let status = if error.kind == scv_core::ToolFailure::Cancelled {
                        "cancelled"
                    } else {
                        "failed"
                    };
                    activity.turn_ended(status, Some(&error.message));
                    activity.run_ended("lost");
                }
                return Err(error);
            }
        };
        let killed = run.exit == RunExit::Killed;
        let mut result = run.stream.finish(run.exit, fallback);
        if self.mode_report == ModeReport::None {
            result.mode = mode::unverified(self.mode_check(), None);
        }
        // A failed CLI's stderr is its own diagnostics, never the reply.
        let stderr = run.stderr_tail.trim();
        if result.status == output::RunStatus::Failed
            && !stderr.is_empty()
            && !result
                .error
                .as_deref()
                .is_some_and(|error| error.contains(stderr))
        {
            result.error = Some(match result.error.take() {
                Some(error) => format!("{error}\n{stderr}"),
                None => stderr.to_owned(),
            });
        }
        if let Some(activity) = &activity {
            if let Some(vendor) = &result.session {
                activity.set_vendor(vendor);
            }
            activity.turn_ended(result.status.as_str(), result.error.as_deref());
            activity.run_ended(if killed { "killed" } else { "exited" });
        }
        let conversation = turn.and_then(|turn| {
            let number = turn.turn;
            turn.finish(
                result.session.clone(),
                result.status == output::RunStatus::Completed,
            )
            .map(|handle| (handle, number))
        });
        let (content, truncated) = result.to_json(
            agent,
            conversation
                .as_ref()
                .map(|(handle, turn)| (handle.as_str(), *turn)),
            run.exit_code,
            &run.stderr_tail,
            self.output_limit,
        );
        let mut output = ToolOutput {
            content,
            failure: result.status.failure(),
            truncated,
        };
        if output.is_error() {
            add_sign_in_hint(&mut output, agent);
        }
        Ok(output)
    }
}

/// A delegated run's stdout reader, stderr tail, and how it ended.
struct AgentRun {
    stream: AgentStream,
    exit: RunExit,
    exit_code: Option<i32>,
    stderr_tail: String,
}

/// Where a Codex run's rollout records the mode of its turn.
#[derive(Clone)]
struct Rollout {
    home: PathBuf,
    /// A continued thread and its rollout's length before this turn: only
    /// what the turn appends after it counts.
    resumed: Option<(String, u64)>,
}

impl Rollout {
    fn new(home: &Path, resumed: Option<&str>) -> Self {
        Self {
            home: home.to_owned(),
            resumed: resumed.map(|thread| {
                let length = mode::codex_rollout(home, thread)
                    .and_then(|file| std::fs::metadata(file).ok())
                    .map_or(0, |metadata| metadata.len());
                (thread.to_owned(), length)
            }),
        }
    }

    /// The thread whose rollout records this turn: the one the run named,
    /// else the one it resumes, which a resumed turn need not name again.
    fn thread(&self, stream: &AgentStream) -> Option<String> {
        stream
            .session()
            .or(self.resumed.as_ref().map(|(thread, _)| thread.as_str()))
            .map(str::to_owned)
    }

    /// The mode this turn of thread `thread` recorded, once it has.
    fn read(&self, thread: &str) -> Option<String> {
        let file = mode::codex_rollout(&self.home, thread)?;
        let offset = match &self.resumed {
            Some((resumed, length)) if resumed == thread => *length,
            _ => 0,
        };
        mode::codex_turn_mode(&file, offset)
    }

    /// Check the turn's mode once the thread has started, stopping the run
    /// on a mismatch or when nothing is recorded in [`ROLLOUT_WAIT`].
    async fn watch(self, stream: Arc<Mutex<AgentStream>>) {
        let mut deadline = None;
        loop {
            tokio::time::sleep(ROLLOUT_POLL).await;
            let thread = {
                let stream = stream.lock().await;
                if stream.mode_settled() {
                    return;
                }
                self.thread(&stream)
            };
            let Some(thread) = thread else {
                continue;
            };
            let deadline =
                *deadline.get_or_insert_with(|| tokio::time::Instant::now() + ROLLOUT_WAIT);
            let reported = self.read(&thread);
            let mut stream = stream.lock().await;
            match reported {
                Some(reported) => return stream.check_mode(&reported),
                None if tokio::time::Instant::now() >= deadline => {
                    return stream.reject_missing_mode();
                }
                None => {}
            }
        }
    }

    /// Read the turn's mode once more after the run ended, for a turn too
    /// short for [`watch`](Self::watch).
    async fn settle(&self, stream: &Mutex<AgentStream>) {
        let mut stream = stream.lock().await;
        if stream.mode_settled() {
            return;
        }
        if let Some(reported) = self.thread(&stream).and_then(|thread| self.read(&thread)) {
            stream.check_mode(&reported);
        }
    }
}

/// Run a native agent: stdout is parsed as it arrives rather than buffered,
/// stderr keeps only its tail, and the run is recorded in the delegation
/// registry while it lasts. `rollout` checks a Codex turn's mode meanwhile.
async fn execute_agent_process(
    mut spec: ProcessSpec,
    stream: AgentStream,
    registration: Option<(Arc<DelegationRegistry>, records::PendingDelegation)>,
    rollout: Option<Rollout>,
    cancellation: tokio_util::sync::CancellationToken,
) -> Result<AgentRun, ToolError> {
    let project = crate::project_environment::resolve(
        &spec.cwd,
        &spec.environment,
        cancellation.clone().cancelled_owned(),
    )
    .await
    .map_err(|error| {
        if cancellation.is_cancelled() {
            ToolError::cancelled("Rust preflight cancelled")
        } else {
            ToolError::unavailable(format!("{error:#}"))
        }
    })?;
    if cancellation.is_cancelled() {
        return Err(ToolError::cancelled("agent launch cancelled"));
    }
    spec.environment.extend(project.environment);
    let mut child = spawn_process(&spec)?;
    let pid = child_pid(&child)?;
    // Bookkeeping must not fail the delegation: an unrecorded run is still
    // tagged, so a later sweep can find what it leaves behind.
    let guard: Option<DelegationGuard> =
        registration.and_then(|(registry, pending)| registry.register(pending, pid as u32).ok());
    // Cancelled from the stdout parser when Claude's init event names a
    // credential source other than the pinned one, or when the run reports
    // a permission mode the policy does not expect.
    let halt = stream.halt_token();
    let stdout = Arc::new(Mutex::new(stream));
    let watch = rollout
        .clone()
        .map(|rollout| tokio::spawn(rollout.watch(Arc::clone(&stdout))));
    let stderr = Arc::new(Mutex::new(TailBuffer::new(STDERR_TAIL_BYTES)));
    let stdout_task = child
        .stdout
        .take()
        .map(|reader| tokio::spawn(drain_output(reader, Arc::clone(&stdout))));
    let stderr_task = child
        .stderr
        .take()
        .map(|reader| tokio::spawn(drain_output(reader, Arc::clone(&stderr))));
    let finished = supervise(
        &mut child,
        pid,
        spec.deadline,
        cancellation,
        stdout_task,
        stderr_task,
        Some(&halt),
    )
    .await;
    records::untrack_spawned(pid as u32);
    if let Some(watch) = watch {
        watch.abort();
        let _ = watch.await;
    }
    if let Some(rollout) = &rollout {
        rollout.settle(&stdout).await;
    }
    let killed = guard.as_ref().is_some_and(DelegationGuard::was_killed);
    if let Some(guard) = guard {
        guard.finish().await;
    }
    let finished = finished?;
    let exit = if finished.timed_out {
        RunExit::TimedOut
    } else if killed {
        RunExit::Killed
    } else {
        RunExit::Exited {
            success: finished.status.success(),
        }
    };
    let stderr_tail = stderr.lock().await.text();
    let stream = Arc::try_unwrap(stdout)
        .map_err(|_| ToolError::failed("agent output reader is still running"))?
        .into_inner();
    Ok(AgentRun {
        stream,
        exit,
        exit_code: finished.status.code(),
        stderr_tail,
    })
}

#[cfg(test)]
mod tests;