supercode_runtime/

background.rs

//! P5-6 (COMPOSABLE-HARNESS-DESIGN.md §2 module 4 `tools.background`: "D1
//! background exec + monitor/event feed; D10 bg-manager; D3 self-paced/
//! scheduled loops"; §2.1 "tools.background → permissions.approvals
//! (auto-policy) [C6 as dep]"; §2.2 C6): the pure, agent-independent data
//! shapes and bounded-buffer arithmetic a runtime agent's
//! `background_exec`/`background_status`/`background_list`/`background_kill`
//! intrinsics build on — kept separate from `agent.rs` so the bounded-
//! capture truncation logic and job-id shape are unit-testable without a
//! full `Agent`/mock-`Provider`/real-subprocess harness, the same
//! "pure config → set, testable without the loop" precedent
//! subagent runtime documents for itself (P5-3).
//!
//! **Activation.** Everything here is inert until `Agent` actually consults
//! it, which only happens when `Config::tools_background_enabled` is `true`
//! (`capabilities.tools_background.enabled`, default `false`) — importing
//! this module changes nothing for an agent that never turns the module on.
//!
//! **Process-kill reuse.** The actual OS-process spawn/kill machinery lives
//! in the agent loop (it needs `tokio::process::Command`/`Child`, which this
//! module deliberately does not depend on, keeping it synchronous and
//! trivially unit-testable). The concurrency bound reuses
//! the subagent concurrency guard verbatim — the same
//! generic `Arc<AtomicUsize>` gauge machinery, just a second, independent
//! gauge instance scoped to background JOBS rather than subagent SPAWNS
//! (`Agent::background_concurrency_gauge`, distinct from
//! `Agent::subagent_concurrency_gauge`).

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Mutex;

/// P5-6 (build brief "cap the buffer like P5-2's 16MiB caps"): the default
/// per-job bounded-capture ceiling, matching `crate::mcp::MCP_MAX_RESPONSE_BYTES`'s
/// hardening precedent — generous for real command output while bounding
/// how much memory one background job (let alone `max_concurrent` of them
/// at once) can force this process to hold.
pub const DEFAULT_MAX_OUTPUT_BYTES: usize = 16 * 1024 * 1024;

/// P5-6 (resource bound, mirroring `crate::subagents`'s "max concurrent...
/// cap, fail-closed... configurable" precedent): the default maximum number
/// of background jobs this agent may have in flight at once.
pub const DEFAULT_MAX_CONCURRENT: usize = 4;

/// A background job's run state, as observed by `background_status`/
/// `background_list`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JobStatus {
    /// Still running (no exit observed yet).
    Running,
    /// Exited on its own; `Some(code)` when the platform reported one
    /// (`None` covers a signal-terminated exit with no portable code, same
    /// convention `BashTool::execute` already uses via
    /// `status.code().unwrap_or(-1)` — this type keeps the `Option`
    /// instead of collapsing it, so callers can tell "exit code 0" from
    /// "no code available" if they care to).
    Exited(Option<i32>),
    /// Killed via `background_kill` (or reclaimed on agent drop) before it
    /// exited on its own.
    Killed,
}

impl JobStatus {
    /// The `"status"` string a tool result JSON reports.
    pub fn as_str(self) -> &'static str {
        match self {
            JobStatus::Running => "running",
            JobStatus::Exited(_) => "exited",
            JobStatus::Killed => "killed",
        }
    }
}

/// Bounded, incrementally-appended output capture shared (via `Arc`)
/// between a job's stdout/stderr reader tasks and whatever later polls it
/// (`background_status`/`background_list`). Thread-safe; every method is
/// fail-soft on a poisoned lock (treats it as "temporarily unavailable",
/// the same posture `crate::permissions::approval::ApprovalCache` already
/// documents for itself) rather than panicking a reader task or a tool
/// call.
#[derive(Debug, Default)]
pub struct CapturedOutput {
    inner: Mutex<CaptureState>,
}

#[derive(Debug, Default)]
struct CaptureState {
    buf: String,
    truncated: bool,
    /// Byte offset into `buf` already handed back by a previous
    /// `drain_new` call — the event-feed cursor.
    drained: usize,
}

impl CapturedOutput {
    /// A fresh, empty capture.
    pub fn new() -> Self {
        CapturedOutput::default()
    }

    /// Append `chunk`, never growing the retained buffer past `cap` bytes —
    /// bytes beyond the cap are DROPPED (fail-closed, never buffered) and
    /// `truncated` latches `true` the first time that happens and stays
    /// true thereafter. A caller must keep reading the underlying pipe past
    /// this point regardless (to avoid blocking the child on a full,
    /// undrained pipe) — this method only bounds what's RETAINED in
    /// memory, not what's read off the pipe.
    pub fn append(&self, chunk: &str, cap: usize) {
        if chunk.is_empty() {
            return;
        }
        let Ok(mut st) = self.inner.lock() else {
            return;
        };
        if st.buf.len() >= cap {
            st.truncated = true;
            return;
        }
        let remaining = cap - st.buf.len();
        if chunk.len() <= remaining {
            st.buf.push_str(chunk);
        } else {
            // Largest char boundary <= remaining, same approach
            // `Agent::cap_tool_output` already uses.
            let mut end = remaining;
            while end > 0 && !chunk.is_char_boundary(end) {
                end -= 1;
            }
            st.buf.push_str(&chunk[..end]);
            st.truncated = true;
        }
    }

    /// The full captured text so far, and whether it was ever truncated.
    pub fn snapshot(&self) -> (String, bool) {
        self.inner
            .lock()
            .map(|st| (st.buf.clone(), st.truncated))
            .unwrap_or_default()
    }

    /// Text appended since the last `drain_new` call (or since creation, on
    /// the first call) — the event-feed's per-poll delta. Advances the
    /// drain cursor even on an empty result, so polling twice in a row with
    /// no new output between them returns `""` the second time, never a
    /// repeat of the first drain.
    pub fn drain_new(&self) -> String {
        let Ok(mut st) = self.inner.lock() else {
            return String::new();
        };
        let new = st.buf[st.drained..].to_string();
        st.drained = st.buf.len();
        new
    }
}

/// P5-6: process-wide sequence number backing [`next_job_id`] —
/// disambiguates two jobs spawned in the same millisecond, mirroring
/// `crate::agent`'s own `SUBAGENT_ID_SEQ` precedent (kept as a second,
/// independent counter rather than sharing that one, since a job id and a
/// subagent id are never compared against each other).
static JOB_ID_SEQ: AtomicU64 = AtomicU64::new(0);

/// A fresh, process-unique background job id (`"bg-<hex-ts>-<hex-seq>"`),
/// given the caller's own millisecond timestamp (kept as a parameter rather
/// than reading the clock in here, so this stays a pure function for the
/// unit tests below).
pub fn next_job_id(now_ms: i64) -> String {
    let seq = JOB_ID_SEQ.fetch_add(1, Ordering::Relaxed);
    format!("bg-{now_ms:x}-{seq:x}")
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn job_status_as_str_matches_the_documented_schema_values() {
        assert_eq!(JobStatus::Running.as_str(), "running");
        assert_eq!(JobStatus::Exited(Some(0)).as_str(), "exited");
        assert_eq!(JobStatus::Exited(None).as_str(), "exited");
        assert_eq!(JobStatus::Killed.as_str(), "killed");
    }

    #[test]
    fn next_job_id_is_unique_across_calls_even_at_the_same_timestamp() {
        let a = next_job_id(1000);
        let b = next_job_id(1000);
        assert_ne!(a, b);
        assert!(a.starts_with("bg-"));
    }

    #[test]
    fn captured_output_appends_under_the_cap_without_truncation() {
        let out = CapturedOutput::new();
        out.append("hello ", 100);
        out.append("world", 100);
        let (buf, truncated) = out.snapshot();
        assert_eq!(buf, "hello world");
        assert!(!truncated);
    }

    #[test]
    fn captured_output_never_grows_past_the_cap_and_latches_truncated() {
        let out = CapturedOutput::new();
        out.append("0123456789", 5); // only "01234" fits
        let (buf, truncated) = out.snapshot();
        assert_eq!(buf, "01234");
        assert!(truncated);
        assert_eq!(buf.len(), 5);

        // Further appends past an already-full cap change nothing except
        // (already-true) truncated — never grow past the cap.
        out.append("more data that must be dropped entirely", 5);
        let (buf2, truncated2) = out.snapshot();
        assert_eq!(buf2, "01234");
        assert!(truncated2);
    }

    #[test]
    fn captured_output_respects_char_boundaries_when_truncating() {
        let out = CapturedOutput::new();
        // "héllo" — 'é' is 2 bytes; cap=2 lands mid-character at byte 2.
        out.append("héllo", 2);
        let (buf, truncated) = out.snapshot();
        assert!(truncated);
        assert!(buf.is_char_boundary(buf.len()));
        assert!(std::str::from_utf8(buf.as_bytes()).is_ok());
    }

    #[test]
    fn drain_new_returns_only_the_delta_since_the_last_call() {
        let out = CapturedOutput::new();
        out.append("first ", 1000);
        assert_eq!(out.drain_new(), "first ");
        assert_eq!(out.drain_new(), "", "no new output since the last drain");
        out.append("second", 1000);
        assert_eq!(out.drain_new(), "second");
    }

    #[test]
    fn ten_mb_of_appends_never_retains_past_the_configured_cap() {
        // Regression proof for the "must not OOM" requirement: a job
        // producing far more output than the cap must leave the retained
        // buffer bounded throughout, not merely "eventually" bounded.
        let out = CapturedOutput::new();
        let cap = 1024;
        let chunk = "x".repeat(4096);
        for _ in 0..2560 {
            // 2560 * 4096 ~= 10 MiB fed in, cap = 1 KiB.
            out.append(&chunk, cap);
            let (buf, _) = out.snapshot();
            assert!(buf.len() <= cap, "buffer must never exceed the cap");
        }
        let (buf, truncated) = out.snapshot();
        assert_eq!(buf.len(), cap);
        assert!(truncated);
    }
}