supercode-harness 0.4.6

//! P5-6 (COMPOSABLE-HARNESS-DESIGN.md §2 module 4 `tools.background`):
//! background exec + monitor/event feed + bg-manager + §2.2 C6 auto-policy,
//! driven end to end through `Agent::send`.
//!
//! **LIVE-AGENT-TEST SAFETY (non-negotiable, per the build brief).** Every
//! test in this file drives `Agent` through a scripted MOCK `Provider` (the
//! same `ScriptedProvider` idiom `tests/subagents.rs`/`tests/agent_loop.rs`
//! already use) — never a real model endpoint. `background_exec` itself
//! spawns a REAL OS process, but only ever harmless commands (`echo`,
//! `sleep`, `yes | head`) — never a real coding-CLI or termfleet provider,
//! exactly the "a background `sleep 100` killed on drop is a fine test"
//! carve-out the build brief names.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Mutex;
use std::time::Duration;

use async_trait::async_trait;
use supercode_harness::permissions::{
    ApprovalOutcome, ApprovalRequest, PermissionsApprovalHandler,
};
use supercode_harness::{
    Agent, ChatMessage, ChatRequest, Config, Error, FunctionCall, Provider, Role, ToolCall, Usage,
};

fn tool_call(id: &str, name: &str, args: serde_json::Value) -> ToolCall {
    ToolCall {
        id: id.to_string(),
        kind: "function".to_string(),
        function: FunctionCall {
            name: name.to_string(),
            arguments: args.to_string(),
        },
    }
}

fn assistant_with_calls(calls: Vec<ToolCall>) -> ChatMessage {
    ChatMessage {
        role: Role::Assistant,
        content: None,
        content_parts: None,
        tool_calls: Some(calls),
        tool_call_id: None,
        name: None,
        metadata: Default::default(),
    }
}

fn last_tool_content(req: &ChatRequest) -> String {
    let last = req.messages.last().expect("at least one message");
    assert_eq!(
        last.role,
        Role::Tool,
        "expected the last message to be a tool result"
    );
    last.content.clone().unwrap_or_default()
}

async fn yield_many(n: usize) {
    for _ in 0..n {
        tokio::task::yield_now().await;
    }
}

/// Linux-only real-process liveness check (the CI/dev environment this
/// build targets) — `/proc/<pid>` existing IS the process still being alive
/// at the OS level, independent of anything this crate's own bookkeeping
/// claims. Used to PROVE a killed/dropped background job's process is
/// genuinely gone, not merely marked so in our own state.
#[cfg(target_os = "linux")]
fn process_alive(pid: u32) -> bool {
    std::path::Path::new(&format!("/proc/{pid}")).exists()
}

/// Poll (bounded) until `process_alive(pid)` is false, or panic — proves
/// termination actually happened rather than merely "probably will soon".
#[cfg(target_os = "linux")]
async fn wait_for_process_death(pid: u32) {
    tokio::time::timeout(Duration::from_secs(10), async move {
        while process_alive(pid) {
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
    })
    .await
    .expect("process must actually terminate within the timeout");
}

/// Parse the `ppid` field out of a raw `/proc/<pid>/stat` file's contents.
/// The `comm` field (2nd, parenthesized) can itself contain spaces/parens,
/// so this skips to the LAST `)` before splitting on whitespace — the
/// standard-safe way to parse this file (matches `man proc`'s own advice).
#[cfg(target_os = "linux")]
fn parse_ppid(stat_contents: &str) -> Option<u32> {
    let rparen = stat_contents.rfind(')')?;
    let rest = &stat_contents[rparen + 1..];
    let mut fields = rest.split_whitespace();
    let _state = fields.next()?;
    fields.next()?.parse().ok()
}

/// Every OS pid whose PARENT is `pid`, discovered by scanning `/proc` —
/// i.e. `pid`'s DIRECT children, independent of anything this crate's own
/// `BackgroundJob` bookkeeping claims. Used to find a background job's
/// GRANDCHILD (relative to the tracked job leader) for the process-group
/// kill proofs below: a process the OLD per-child-only kill
/// (`Child::start_kill` on just the tracked pid) would leave orphaned.
#[cfg(target_os = "linux")]
fn direct_children_of(pid: u32) -> Vec<u32> {
    let mut out = Vec::new();
    let Ok(entries) = std::fs::read_dir("/proc") else {
        return out;
    };
    for entry in entries.flatten() {
        let Some(candidate) = entry
            .file_name()
            .to_str()
            .and_then(|s| s.parse::<u32>().ok())
        else {
            continue;
        };
        if let Ok(contents) = std::fs::read_to_string(format!("/proc/{candidate}/stat")) {
            if parse_ppid(&contents) == Some(pid) {
                out.push(candidate);
            }
        }
    }
    out
}

/// Poll (bounded) until `pid` has at least `want` direct children visible in
/// `/proc`, returning them — proves a background job's subprocess (e.g. the
/// `sleep` backgrounded by `sleep 987654 & ... ; wait`, or either side of a
/// `a | b` pipeline) has actually been forked before this test goes on to
/// kill/drop the job and check it.
#[cfg(target_os = "linux")]
async fn wait_for_children(pid: u32, want: usize) -> Vec<u32> {
    tokio::time::timeout(Duration::from_secs(10), async move {
        loop {
            let kids = direct_children_of(pid);
            if kids.len() >= want {
                return kids;
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
    })
    .await
    .expect("the background job's grandchild(ren) must appear within the timeout")
}

// ---------------------------------------------------------------------------
// 1. Default-off: byte-identical to today's no-tools_background behavior.
// ---------------------------------------------------------------------------

struct PlainProvider;
#[async_trait]
impl Provider for PlainProvider {
    async fn complete(
        &self,
        _req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        Ok((ChatMessage::assistant("ok"), Usage::default()))
    }
}

#[tokio::test]
async fn default_off_never_advertises_any_background_tool() {
    let config = Config::builder().build();
    let agent = Agent::with_provider(config, Box::new(PlainProvider));
    let schemas = agent.tool_schemas();
    let names: Vec<&str> = schemas.iter().map(|s| s.name.as_str()).collect();
    for n in [
        "background_exec",
        "background_status",
        "background_list",
        "background_kill",
    ] {
        assert!(!names.contains(&n), "{n} must not be advertised when off");
    }
}

struct HallucinatedCallProvider {
    calls: AtomicUsize,
}
#[async_trait]
impl Provider for HallucinatedCallProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        match self.calls.fetch_add(1, Ordering::SeqCst) {
            0 => Ok((
                assistant_with_calls(vec![tool_call(
                    "call_1",
                    "background_exec",
                    serde_json::json!({"command": "echo hi"}),
                )]),
                Usage::default(),
            )),
            1 => {
                let content = last_tool_content(req);
                assert_eq!(
                    content, "Error: model requested unknown tool: background_exec",
                    "a disabled call must fall through to the plain unknown-tool error, \
                     never a background-specific one: {content}"
                );
                Ok((ChatMessage::assistant("handled"), Usage::default()))
            }
            n => panic!("unexpected call {n}"),
        }
    }
}

#[tokio::test]
async fn default_off_hallucinated_call_is_plain_unknown_tool() {
    let config = Config::builder().build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(HallucinatedCallProvider {
            calls: AtomicUsize::new(0),
        }),
    );
    let reply = agent.send("start").await.unwrap();
    assert_eq!(reply, "handled");
}

// ---------------------------------------------------------------------------
// 2. Happy path: exec returns immediately with a job id (never the command's
//    output); a later status poll (same agent, second `send()`) reports the
//    REAL captured output and reaps the job; a poll after that is refused
//    as unknown (proving the reap actually removed it).
// ---------------------------------------------------------------------------

struct HappyPathProvider {
    calls: AtomicUsize,
    captured_id: Mutex<Option<String>>,
}
#[async_trait]
impl Provider for HappyPathProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        match self.calls.fetch_add(1, Ordering::SeqCst) {
            0 => {
                assert!(req.tools.iter().any(|t| t.name == "background_exec"));
                assert!(req.tools.iter().any(|t| t.name == "background_status"));
                assert!(req.tools.iter().any(|t| t.name == "background_list"));
                assert!(req.tools.iter().any(|t| t.name == "background_kill"));
                Ok((
                    assistant_with_calls(vec![tool_call(
                        "call_1",
                        "background_exec",
                        serde_json::json!({"command": "echo hello-from-bg"}),
                    )]),
                    Usage::default(),
                ))
            }
            1 => {
                let content = last_tool_content(req);
                let v: serde_json::Value = serde_json::from_str(&content).unwrap();
                assert_eq!(v["status"], "running", "content: {content}");
                assert!(v["job_id"].as_str().is_some(), "content: {content}");
                assert!(v["pid"].as_u64().is_some(), "content: {content}");
                *self.captured_id.lock().unwrap() = Some(v["job_id"].as_str().unwrap().to_string());
                Ok((ChatMessage::assistant("spawned"), Usage::default()))
            }
            2 => {
                let id = self.captured_id.lock().unwrap().clone().unwrap();
                Ok((
                    assistant_with_calls(vec![tool_call(
                        "call_2",
                        "background_status",
                        serde_json::json!({"job_id": id}),
                    )]),
                    Usage::default(),
                ))
            }
            3 => {
                let content = last_tool_content(req);
                let v: serde_json::Value = serde_json::from_str(&content).unwrap();
                assert_eq!(v["status"], "exited", "content: {content}");
                assert_eq!(v["exit_code"], 0, "content: {content}");
                assert!(
                    v["output"].as_str().unwrap().contains("hello-from-bg"),
                    "content: {content}"
                );
                assert_eq!(v["output_truncated"], false);
                Ok((ChatMessage::assistant("polled"), Usage::default()))
            }
            4 => {
                // Second poll of the SAME (already-reaped) id — must be
                // refused as unknown, proving the terminal poll above
                // actually removed it from the tracked set.
                let id = self.captured_id.lock().unwrap().clone().unwrap();
                Ok((
                    assistant_with_calls(vec![tool_call(
                        "call_3",
                        "background_status",
                        serde_json::json!({"job_id": id}),
                    )]),
                    Usage::default(),
                ))
            }
            5 => {
                let content = last_tool_content(req);
                assert!(
                    content.contains("unknown background job id"),
                    "content: {content}"
                );
                Ok((ChatMessage::assistant("done"), Usage::default()))
            }
            n => panic!("unexpected call {n}"),
        }
    }
}

#[tokio::test]
async fn background_exec_returns_immediately_then_status_reports_output_and_reaps() {
    let config = Config::builder().tools_background_enabled(true).build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(HappyPathProvider {
            calls: AtomicUsize::new(0),
            captured_id: Mutex::new(None),
        }),
    );
    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("start"))
        .await
        .expect("must not hang")
        .unwrap();
    assert_eq!(reply, "spawned");

    // Give the real (near-instant) `echo` process every chance to finish
    // and its reader tasks to drain it before polling status.
    tokio::time::sleep(Duration::from_millis(100)).await;
    yield_many(50).await;

    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("check"))
        .await
        .expect("must not hang")
        .unwrap();
    assert_eq!(reply, "polled");

    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("check again"))
        .await
        .expect("must not hang")
        .unwrap();
    assert_eq!(reply, "done");
}

// ---------------------------------------------------------------------------
// 3. Security: a background command goes through the SAME sandbox/
//    permissions as a foreground `bash` call — a `bash(...)`-authored deny
//    rule catches it identically.
// ---------------------------------------------------------------------------

struct DenyRuleProvider {
    calls: AtomicUsize,
}
#[async_trait]
impl Provider for DenyRuleProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        match self.calls.fetch_add(1, Ordering::SeqCst) {
            0 => Ok((
                assistant_with_calls(vec![tool_call(
                    "call_1",
                    "background_exec",
                    serde_json::json!({"command": "rm -rf /tmp/should-never-run"}),
                )]),
                Usage::default(),
            )),
            1 => {
                let content = last_tool_content(req);
                assert!(
                    content.contains("was not approved"),
                    "a bash(...) deny rule must refuse a background exec identically to a \
                     foreground bash call: {content}"
                );
                Ok((
                    ChatMessage::assistant("denied as expected"),
                    Usage::default(),
                ))
            }
            n => panic!("unexpected call {n}"),
        }
    }
}

#[tokio::test]
async fn background_exec_is_refused_by_the_same_bash_deny_rule_as_a_foreground_call() {
    let mut config = Config::builder().tools_background_enabled(true).build();
    config.permissions_enabled = true;
    config.tool_deny_patterns = vec!["bash(rm -rf*)".to_string()];
    let mut agent = Agent::with_provider(
        config,
        Box::new(DenyRuleProvider {
            calls: AtomicUsize::new(0),
        }),
    );
    let reply = agent.send("start").await.unwrap();
    assert_eq!(reply, "denied as expected");
}

// ---------------------------------------------------------------------------
// 4. §2.2 C6: an Ask-tier command is DENIED for a background job, never
//    hangs, and NEVER consults an interactive handler — even one that IS
//    installed and would allow the same command run in the foreground.
// ---------------------------------------------------------------------------

struct AlwaysAllowHandler {
    invoked: std::sync::Arc<std::sync::atomic::AtomicBool>,
}
impl PermissionsApprovalHandler for AlwaysAllowHandler {
    fn ask(&self, _req: &ApprovalRequest) -> ApprovalOutcome {
        self.invoked.store(true, Ordering::SeqCst);
        ApprovalOutcome::Allow
    }
}

struct C6Provider {
    calls: AtomicUsize,
}
#[async_trait]
impl Provider for C6Provider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        match self.calls.fetch_add(1, Ordering::SeqCst) {
            0 => Ok((
                assistant_with_calls(vec![tool_call(
                    "call_1",
                    "background_exec",
                    serde_json::json!({"command": "echo should-be-denied"}),
                )]),
                Usage::default(),
            )),
            1 => {
                let content = last_tool_content(req);
                assert!(
                    content.contains("was not approved") || content.contains("auto-policy"),
                    "an Ask-tier background command must be denied, never hang or ask \
                     interactively: {content}"
                );
                Ok((
                    ChatMessage::assistant("denied, never hung"),
                    Usage::default(),
                ))
            }
            n => panic!("unexpected call {n}"),
        }
    }
}

#[tokio::test]
async fn background_exec_ask_tier_command_is_denied_without_ever_hanging_or_asking() {
    let mut config = Config::builder().tools_background_enabled(true).build();
    config.permissions_enabled = true;
    // Force `bash` to the engine's Ask tier.
    config.permissions_ask_patterns = vec!["bash".to_string()];
    let mut agent = Agent::with_provider(
        config,
        Box::new(C6Provider {
            calls: AtomicUsize::new(0),
        }),
    );
    let invoked = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    agent.set_permissions_approval_handler(AlwaysAllowHandler {
        invoked: invoked.clone(),
    });

    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("start"))
        .await
        .expect("must never hang, even under an Ask-tier policy with no way to resolve")
        .unwrap();
    assert_eq!(reply, "denied, never hung");
    assert!(
        !invoked.load(Ordering::SeqCst),
        "a background_exec call must NEVER consult an interactive handler, even one that's \
         installed and would have allowed the call"
    );
}

/// Contrast case: the SAME config/handler, but the identical command run in
/// the (non-background) `bash` tool IS allowed — proving §4's denial above
/// is specific to the background path (C6), not a blanket deny of `bash`
/// under this policy.
struct ForegroundContrastProvider {
    calls: AtomicUsize,
}
#[async_trait]
impl Provider for ForegroundContrastProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        match self.calls.fetch_add(1, Ordering::SeqCst) {
            0 => Ok((
                assistant_with_calls(vec![tool_call(
                    "call_1",
                    "bash",
                    serde_json::json!({"command": "echo fine-in-foreground"}),
                )]),
                Usage::default(),
            )),
            1 => {
                let content = last_tool_content(req);
                assert!(
                    content.contains("fine-in-foreground"),
                    "the interactive handler must still allow a FOREGROUND ask-tier call: \
                     {content}"
                );
                Ok((
                    ChatMessage::assistant("foreground allowed"),
                    Usage::default(),
                ))
            }
            n => panic!("unexpected call {n}"),
        }
    }
}

#[tokio::test]
async fn the_same_ask_tier_command_is_allowed_in_the_foreground_via_the_interactive_handler() {
    let mut config = Config::builder().tools_background_enabled(true).build();
    config.permissions_enabled = true;
    config.permissions_ask_patterns = vec!["bash".to_string()];
    let mut agent = Agent::with_provider(
        config,
        Box::new(ForegroundContrastProvider {
            calls: AtomicUsize::new(0),
        }),
    );
    agent.set_permissions_approval_handler(AlwaysAllowHandler {
        invoked: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
    });
    let reply = agent.send("start").await.unwrap();
    assert_eq!(reply, "foreground allowed");
}

// ---------------------------------------------------------------------------
// 5. Concurrency cap: fail-closed, named error, never a stray held slot.
// ---------------------------------------------------------------------------

struct ConcurrencyCapProvider {
    calls: AtomicUsize,
}
#[async_trait]
impl Provider for ConcurrencyCapProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        match self.calls.fetch_add(1, Ordering::SeqCst) {
            0 => Ok((
                assistant_with_calls(vec![
                    tool_call(
                        "call_1",
                        "background_exec",
                        serde_json::json!({"command": "sleep 5"}),
                    ),
                    tool_call(
                        "call_2",
                        "background_exec",
                        serde_json::json!({"command": "sleep 5"}),
                    ),
                ]),
                Usage::default(),
            )),
            1 => {
                let n = req.messages.len();
                let first = req.messages[n - 2].content.clone().unwrap_or_default();
                let second = req.messages[n - 1].content.clone().unwrap_or_default();
                assert!(first.contains("\"status\":\"running\""), "first: {first}");
                assert!(
                    second.contains("BackgroundJobConcurrencyExceeded")
                        || second.contains("already running"),
                    "second: {second}"
                );
                Ok((ChatMessage::assistant("both handled"), Usage::default()))
            }
            n => panic!("unexpected call {n}"),
        }
    }
}

#[tokio::test]
async fn concurrency_cap_refuses_a_second_job_past_max_concurrent() {
    let config = Config::builder()
        .tools_background_enabled(true)
        .tools_background_max_concurrent(1)
        .build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(ConcurrencyCapProvider {
            calls: AtomicUsize::new(0),
        }),
    );
    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("start"))
        .await
        .expect("must not hang")
        .unwrap();
    assert_eq!(reply, "both handled");
}

// ---------------------------------------------------------------------------
// 6. Bounded output: a job producing far more output than the configured
//    cap is truncated-with-marker, never buffered unboundedly.
// ---------------------------------------------------------------------------

struct BoundedOutputProvider {
    calls: AtomicUsize,
    captured_id: Mutex<Option<String>>,
}
#[async_trait]
impl Provider for BoundedOutputProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        match self.calls.fetch_add(1, Ordering::SeqCst) {
            0 => Ok((
                assistant_with_calls(vec![tool_call(
                    "call_1",
                    "background_exec",
                    // Far more than the 256-byte cap configured below.
                    serde_json::json!({"command": "yes x | head -c 100000"}),
                )]),
                Usage::default(),
            )),
            1 => {
                let content = last_tool_content(req);
                let v: serde_json::Value = serde_json::from_str(&content).unwrap();
                *self.captured_id.lock().unwrap() = Some(v["job_id"].as_str().unwrap().to_string());
                Ok((ChatMessage::assistant("spawned"), Usage::default()))
            }
            2 => {
                let id = self.captured_id.lock().unwrap().clone().unwrap();
                Ok((
                    assistant_with_calls(vec![tool_call(
                        "call_2",
                        "background_status",
                        serde_json::json!({"job_id": id}),
                    )]),
                    Usage::default(),
                ))
            }
            3 => {
                let content = last_tool_content(req);
                let v: serde_json::Value = serde_json::from_str(&content).unwrap();
                let output = v["output"].as_str().unwrap();
                assert!(
                    output.len() <= 256,
                    "retained output must never exceed the configured cap: {} bytes",
                    output.len()
                );
                assert_eq!(v["output_truncated"], true, "content: {content}");
                Ok((ChatMessage::assistant("bounded ok"), Usage::default()))
            }
            n => panic!("unexpected call {n}"),
        }
    }
}

#[tokio::test]
async fn a_verbose_job_is_truncated_with_a_marker_and_never_exceeds_the_cap() {
    let config = Config::builder()
        .tools_background_enabled(true)
        .tools_background_max_output_bytes(256)
        .build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(BoundedOutputProvider {
            calls: AtomicUsize::new(0),
            captured_id: Mutex::new(None),
        }),
    );
    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("start"))
        .await
        .expect("must not hang")
        .unwrap();
    assert_eq!(reply, "spawned");

    // Let the (bounded-output, but real) process finish before polling.
    tokio::time::sleep(Duration::from_millis(300)).await;
    yield_many(50).await;

    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("check"))
        .await
        .expect("must not hang")
        .unwrap();
    assert_eq!(reply, "bounded ok");
}

// ---------------------------------------------------------------------------
// 7. Real process termination: `background_kill` actually kills the OS
//    process (proven via /proc), not just marks it killed in-memory.
// ---------------------------------------------------------------------------

#[cfg(target_os = "linux")]
struct KillProvider {
    calls: AtomicUsize,
    captured_pid: std::sync::Arc<Mutex<Option<u32>>>,
    captured_id: Mutex<Option<String>>,
}
#[cfg(target_os = "linux")]
#[async_trait]
impl Provider for KillProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        match self.calls.fetch_add(1, Ordering::SeqCst) {
            0 => Ok((
                assistant_with_calls(vec![tool_call(
                    "call_1",
                    "background_exec",
                    serde_json::json!({"command": "sleep 30"}),
                )]),
                Usage::default(),
            )),
            1 => {
                let content = last_tool_content(req);
                let v: serde_json::Value = serde_json::from_str(&content).unwrap();
                *self.captured_id.lock().unwrap() = Some(v["job_id"].as_str().unwrap().to_string());
                *self.captured_pid.lock().unwrap() = v["pid"].as_u64().map(|p| p as u32);
                Ok((ChatMessage::assistant("spawned"), Usage::default()))
            }
            2 => {
                let id = self.captured_id.lock().unwrap().clone().unwrap();
                Ok((
                    assistant_with_calls(vec![tool_call(
                        "call_2",
                        "background_kill",
                        serde_json::json!({"job_id": id}),
                    )]),
                    Usage::default(),
                ))
            }
            3 => {
                let content = last_tool_content(req);
                assert!(
                    content.contains("\"status\":\"killed\""),
                    "content: {content}"
                );
                Ok((ChatMessage::assistant("killed"), Usage::default()))
            }
            n => panic!("unexpected call {n}"),
        }
    }
}

#[cfg(target_os = "linux")]
#[tokio::test]
async fn background_kill_actually_terminates_the_real_os_process() {
    let config = Config::builder().tools_background_enabled(true).build();
    let captured_pid = std::sync::Arc::new(Mutex::new(None));
    let mut agent = Agent::with_provider(
        config,
        Box::new(KillProvider {
            calls: AtomicUsize::new(0),
            captured_pid: captured_pid.clone(),
            captured_id: Mutex::new(None),
        }),
    );
    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("start"))
        .await
        .expect("must not hang")
        .unwrap();
    assert_eq!(reply, "spawned");

    let pid = captured_pid.lock().unwrap().expect("pid captured at spawn");
    assert!(
        process_alive(pid),
        "the sleep process must be alive before kill"
    );

    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("kill it"))
        .await
        .expect("must not hang")
        .unwrap();
    assert_eq!(reply, "killed");

    wait_for_process_death(pid).await;
}

// ---------------------------------------------------------------------------
// 8. Real process termination on agent DROP: a still-running background
//    job's process is genuinely killed when the parent `Agent` drops.
// ---------------------------------------------------------------------------

#[cfg(target_os = "linux")]
struct DropKillProvider {
    calls: AtomicUsize,
    pid_tx: Mutex<Option<tokio::sync::oneshot::Sender<u32>>>,
}
#[cfg(target_os = "linux")]
#[async_trait]
impl Provider for DropKillProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        match self.calls.fetch_add(1, Ordering::SeqCst) {
            0 => Ok((
                assistant_with_calls(vec![tool_call(
                    "call_1",
                    "background_exec",
                    serde_json::json!({"command": "sleep 30"}),
                )]),
                Usage::default(),
            )),
            1 => {
                let content = last_tool_content(req);
                let v: serde_json::Value = serde_json::from_str(&content).unwrap();
                let pid = v["pid"].as_u64().unwrap() as u32;
                if let Some(tx) = self.pid_tx.lock().unwrap().take() {
                    let _ = tx.send(pid);
                }
                Ok((ChatMessage::assistant("spawned"), Usage::default()))
            }
            n => panic!("unexpected call {n}"),
        }
    }
}

#[cfg(target_os = "linux")]
#[tokio::test]
async fn dropping_the_agent_kills_the_real_process_of_a_still_running_job() {
    let (tx, rx) = tokio::sync::oneshot::channel();
    let config = Config::builder().tools_background_enabled(true).build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(DropKillProvider {
            calls: AtomicUsize::new(0),
            pid_tx: Mutex::new(Some(tx)),
        }),
    );
    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("start"))
        .await
        .expect("must not hang")
        .unwrap();
    assert_eq!(reply, "spawned");

    let pid = tokio::time::timeout(Duration::from_secs(5), rx)
        .await
        .expect("must not hang")
        .expect("pid must have been sent");
    assert!(
        process_alive(pid),
        "the sleep process must be alive before drop"
    );

    drop(agent);

    wait_for_process_death(pid).await;
}

// ---------------------------------------------------------------------------
// 9. `background_list` surfaces a running job without reaping it.
// ---------------------------------------------------------------------------

struct ListProvider {
    calls: AtomicUsize,
}
#[async_trait]
impl Provider for ListProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        match self.calls.fetch_add(1, Ordering::SeqCst) {
            0 => Ok((
                assistant_with_calls(vec![tool_call(
                    "call_1",
                    "background_exec",
                    serde_json::json!({"command": "sleep 5"}),
                )]),
                Usage::default(),
            )),
            1 => Ok((
                assistant_with_calls(vec![tool_call(
                    "call_2",
                    "background_list",
                    serde_json::json!({}),
                )]),
                Usage::default(),
            )),
            2 => {
                let content = last_tool_content(req);
                let v: serde_json::Value = serde_json::from_str(&content).unwrap();
                let jobs = v["jobs"].as_array().unwrap();
                assert_eq!(jobs.len(), 1, "content: {content}");
                assert_eq!(jobs[0]["command"], "sleep 5");
                assert_eq!(jobs[0]["status"], "running");
                Ok((ChatMessage::assistant("listed"), Usage::default()))
            }
            n => panic!("unexpected call {n}"),
        }
    }
}

#[tokio::test]
async fn background_list_surfaces_a_running_job_without_reaping_it() {
    let config = Config::builder().tools_background_enabled(true).build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(ListProvider {
            calls: AtomicUsize::new(0),
        }),
    );
    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("start"))
        .await
        .expect("must not hang")
        .unwrap();
    assert_eq!(reply, "listed");
}

// ---------------------------------------------------------------------------
// 10. `Error` variants this module introduces render the text tests above
// grep for.
// ---------------------------------------------------------------------------

#[test]
fn background_error_variants_render_the_documented_text() {
    let e = Error::BackgroundJobConcurrencyExceeded { max_concurrent: 4 };
    assert!(e
        .to_string()
        .contains("4 background job(s) already running"));

    let e = Error::BackgroundJobNotFound("bg-x".to_string());
    assert!(e.to_string().contains("unknown background job id `bg-x`"));
}

// ---------------------------------------------------------------------------
// 11. Fable-5 review (HIGH, "grandchildren orphaned on kill AND agent-drop"):
//     `background_kill`/agent-drop must terminate the WHOLE process group a
//     background job spawned, not just the one directly-tracked pid. Proven
//     with the review's own probe: `sleep 987654 & echo $!; wait` — the
//     tracked pid is the `sh`, `sleep` is a GRANDCHILD it forks and
//     backgrounds. Bounded-timeout `wait_for_process_death` below is
//     fail-on-revert: with only the old per-child `Child::start_kill`, the
//     orphaned `sleep` (reparented to init) would never die and this test
//     would time out and panic.
// ---------------------------------------------------------------------------

#[cfg(target_os = "linux")]
struct GrandchildKillProvider {
    calls: AtomicUsize,
    captured_tracked_pid: std::sync::Arc<Mutex<Option<u32>>>,
    captured_id: Mutex<Option<String>>,
}
#[cfg(target_os = "linux")]
#[async_trait]
impl Provider for GrandchildKillProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        match self.calls.fetch_add(1, Ordering::SeqCst) {
            0 => Ok((
                assistant_with_calls(vec![tool_call(
                    "call_1",
                    "background_exec",
                    serde_json::json!({"command": "sleep 987654 & echo $!; wait"}),
                )]),
                Usage::default(),
            )),
            1 => {
                let content = last_tool_content(req);
                let v: serde_json::Value = serde_json::from_str(&content).unwrap();
                *self.captured_id.lock().unwrap() = Some(v["job_id"].as_str().unwrap().to_string());
                *self.captured_tracked_pid.lock().unwrap() = v["pid"].as_u64().map(|p| p as u32);
                Ok((ChatMessage::assistant("spawned"), Usage::default()))
            }
            2 => {
                let id = self.captured_id.lock().unwrap().clone().unwrap();
                Ok((
                    assistant_with_calls(vec![tool_call(
                        "call_2",
                        "background_kill",
                        serde_json::json!({"job_id": id}),
                    )]),
                    Usage::default(),
                ))
            }
            3 => {
                let content = last_tool_content(req);
                assert!(
                    content.contains("\"status\":\"killed\""),
                    "content: {content}"
                );
                Ok((ChatMessage::assistant("killed"), Usage::default()))
            }
            n => panic!("unexpected call {n}"),
        }
    }
}

#[cfg(target_os = "linux")]
#[tokio::test]
async fn background_kill_also_terminates_a_grandchild_process_left_behind_by_a_background_job() {
    let config = Config::builder().tools_background_enabled(true).build();
    let captured_tracked_pid = std::sync::Arc::new(Mutex::new(None));
    let mut agent = Agent::with_provider(
        config,
        Box::new(GrandchildKillProvider {
            calls: AtomicUsize::new(0),
            captured_tracked_pid: captured_tracked_pid.clone(),
            captured_id: Mutex::new(None),
        }),
    );
    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("start"))
        .await
        .expect("must not hang")
        .unwrap();
    assert_eq!(reply, "spawned");

    let tracked_pid = captured_tracked_pid
        .lock()
        .unwrap()
        .expect("tracked pid captured at spawn");
    assert!(
        process_alive(tracked_pid),
        "the tracked `sh` process must be alive before kill"
    );

    // The `sleep 987654` grandchild the tracked `sh` forks and backgrounds
    // — a DIRECT child of the tracked pid, invisible to a per-child-only
    // kill.
    let grandchildren = wait_for_children(tracked_pid, 1).await;
    let grandchild_pid = grandchildren[0];
    assert!(
        process_alive(grandchild_pid),
        "the grandchild `sleep` process must be alive before kill"
    );

    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("kill it"))
        .await
        .expect("must not hang")
        .unwrap();
    assert_eq!(reply, "killed");

    // Both the tracked leader AND the grandchild must actually die — the
    // grandchild check is what a per-child-only kill would fail.
    wait_for_process_death(tracked_pid).await;
    wait_for_process_death(grandchild_pid).await;
}

// ---------------------------------------------------------------------------
// 12. Same grandchild-orphaning defect, but on the agent-DROP path rather
//     than an explicit `background_kill` — `impl Drop for Agent` must also
//     kill the whole process group, not just the tracked leader.
// ---------------------------------------------------------------------------

#[cfg(target_os = "linux")]
struct GrandchildDropProvider {
    calls: AtomicUsize,
    pid_tx: Mutex<Option<tokio::sync::oneshot::Sender<u32>>>,
}
#[cfg(target_os = "linux")]
#[async_trait]
impl Provider for GrandchildDropProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        match self.calls.fetch_add(1, Ordering::SeqCst) {
            0 => Ok((
                assistant_with_calls(vec![tool_call(
                    "call_1",
                    "background_exec",
                    serde_json::json!({"command": "sleep 987654 & echo $!; wait"}),
                )]),
                Usage::default(),
            )),
            1 => {
                let content = last_tool_content(req);
                let v: serde_json::Value = serde_json::from_str(&content).unwrap();
                let pid = v["pid"].as_u64().unwrap() as u32;
                if let Some(tx) = self.pid_tx.lock().unwrap().take() {
                    let _ = tx.send(pid);
                }
                Ok((ChatMessage::assistant("spawned"), Usage::default()))
            }
            n => panic!("unexpected call {n}"),
        }
    }
}

#[cfg(target_os = "linux")]
#[tokio::test]
async fn dropping_the_agent_also_terminates_a_grandchild_process_left_behind_by_a_background_job() {
    let (tx, rx) = tokio::sync::oneshot::channel();
    let config = Config::builder().tools_background_enabled(true).build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(GrandchildDropProvider {
            calls: AtomicUsize::new(0),
            pid_tx: Mutex::new(Some(tx)),
        }),
    );
    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("start"))
        .await
        .expect("must not hang")
        .unwrap();
    assert_eq!(reply, "spawned");

    let tracked_pid = tokio::time::timeout(Duration::from_secs(5), rx)
        .await
        .expect("must not hang")
        .expect("pid must have been sent");
    assert!(
        process_alive(tracked_pid),
        "the tracked `sh` process must be alive before drop"
    );

    let grandchildren = wait_for_children(tracked_pid, 1).await;
    let grandchild_pid = grandchildren[0];
    assert!(
        process_alive(grandchild_pid),
        "the grandchild `sleep` process must be alive before drop"
    );

    drop(agent);

    wait_for_process_death(tracked_pid).await;
    wait_for_process_death(grandchild_pid).await;
}

// ---------------------------------------------------------------------------
// 13. Pipeline case (`a | b`): both stages are direct children of the
//     tracked `sh` (siblings, not chained) — `background_kill` must
//     terminate BOTH, not just whichever the (nonexistent) direct-child
//     tracking would have picked.
// ---------------------------------------------------------------------------

#[cfg(target_os = "linux")]
struct PipelineKillProvider {
    calls: AtomicUsize,
    captured_tracked_pid: std::sync::Arc<Mutex<Option<u32>>>,
    captured_id: Mutex<Option<String>>,
}
#[cfg(target_os = "linux")]
#[async_trait]
impl Provider for PipelineKillProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        match self.calls.fetch_add(1, Ordering::SeqCst) {
            0 => Ok((
                assistant_with_calls(vec![tool_call(
                    "call_1",
                    "background_exec",
                    // Never terminates on its own within the test's
                    // lifetime — `cat` keeps reading forever — so any
                    // death observed below is provably caused by the kill,
                    // not a natural pipeline completion racing it.
                    serde_json::json!({"command": "yes | cat > /dev/null"}),
                )]),
                Usage::default(),
            )),
            1 => {
                let content = last_tool_content(req);
                let v: serde_json::Value = serde_json::from_str(&content).unwrap();
                *self.captured_id.lock().unwrap() = Some(v["job_id"].as_str().unwrap().to_string());
                *self.captured_tracked_pid.lock().unwrap() = v["pid"].as_u64().map(|p| p as u32);
                Ok((ChatMessage::assistant("spawned"), Usage::default()))
            }
            2 => {
                let id = self.captured_id.lock().unwrap().clone().unwrap();
                Ok((
                    assistant_with_calls(vec![tool_call(
                        "call_2",
                        "background_kill",
                        serde_json::json!({"job_id": id}),
                    )]),
                    Usage::default(),
                ))
            }
            3 => {
                let content = last_tool_content(req);
                assert!(
                    content.contains("\"status\":\"killed\""),
                    "content: {content}"
                );
                Ok((ChatMessage::assistant("killed"), Usage::default()))
            }
            n => panic!("unexpected call {n}"),
        }
    }
}

#[cfg(target_os = "linux")]
#[tokio::test]
async fn background_kill_terminates_every_stage_of_a_piped_background_job() {
    let config = Config::builder().tools_background_enabled(true).build();
    let captured_tracked_pid = std::sync::Arc::new(Mutex::new(None));
    let mut agent = Agent::with_provider(
        config,
        Box::new(PipelineKillProvider {
            calls: AtomicUsize::new(0),
            captured_tracked_pid: captured_tracked_pid.clone(),
            captured_id: Mutex::new(None),
        }),
    );
    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("start"))
        .await
        .expect("must not hang")
        .unwrap();
    assert_eq!(reply, "spawned");

    let tracked_pid = captured_tracked_pid
        .lock()
        .unwrap()
        .expect("tracked pid captured at spawn");

    // `yes` and `cat` — both direct children (pipeline siblings) of the
    // tracked `sh`.
    let stages = wait_for_children(tracked_pid, 2).await;
    assert_eq!(stages.len(), 2, "expected both pipeline stages: {stages:?}");
    for &pid in &stages {
        assert!(
            process_alive(pid),
            "pipeline stage {pid} must be alive before kill"
        );
    }

    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("kill it"))
        .await
        .expect("must not hang")
        .unwrap();
    assert_eq!(reply, "killed");

    wait_for_process_death(tracked_pid).await;
    for pid in stages {
        wait_for_process_death(pid).await;
    }
}
// ---------------------------------------------------------------------------
// 14. Fable-5 review (LOW, "pre_tool_hook + doom-loop don't cover
//     background_exec"): a `pre_tool_hook` that vetoes everything must deny
//     a `background_exec` call, matching a foreground `bash` call — proven
//     against `tests/agent_loop.rs::pre_and_post_tool_hooks`'s own
//     "blocked by pre-tool hook" wording. Fail-on-revert: before the fix,
//     `background_exec` was intercepted and fully executed (spawning a REAL
//     process) before `Self::finish_prepare` — where the hook lives — ever
//     ran, so this assertion would fail (the job would actually spawn and
//     report `"status":"running"` instead of being denied).
// ---------------------------------------------------------------------------

struct PreToolHookVetoProvider {
    calls: AtomicUsize,
}
#[async_trait]
impl Provider for PreToolHookVetoProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        match self.calls.fetch_add(1, Ordering::SeqCst) {
            0 => Ok((
                assistant_with_calls(vec![tool_call(
                    "call_1",
                    "background_exec",
                    serde_json::json!({"command": "echo should-never-spawn"}),
                )]),
                Usage::default(),
            )),
            1 => {
                let content = last_tool_content(req);
                assert!(
                    content.contains("blocked by pre-tool hook"),
                    "a vetoing pre_tool_hook must deny background_exec exactly like a \
                     foreground call: {content}"
                );
                assert!(
                    content.contains("vetoed by test hook"),
                    "the hook's own reason text must be surfaced: {content}"
                );
                Ok((
                    assistant_with_calls(vec![tool_call(
                        "call_2",
                        "background_list",
                        serde_json::json!({}),
                    )]),
                    Usage::default(),
                ))
            }
            2 => {
                let content = last_tool_content(req);
                let v: serde_json::Value = serde_json::from_str(&content).unwrap();
                assert_eq!(
                    v["jobs"].as_array().unwrap().len(),
                    0,
                    "a vetoed background_exec must never have reached the spawn — no job \
                     should be tracked: {content}"
                );
                Ok((
                    ChatMessage::assistant("denied, no job spawned"),
                    Usage::default(),
                ))
            }
            n => panic!("unexpected call {n}"),
        }
    }
}

#[tokio::test]
async fn background_exec_is_denied_by_a_pre_tool_hook_like_a_foreground_call() {
    let config = Config::builder()
        .tools_background_enabled(true)
        .pre_tool_hook(Box::new(|name, _args| {
            if name == "background_exec" {
                Some("vetoed by test hook".to_string())
            } else {
                None
            }
        }))
        .build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(PreToolHookVetoProvider {
            calls: AtomicUsize::new(0),
        }),
    );
    let reply = tokio::time::timeout(Duration::from_secs(10), agent.send("start"))
        .await
        .expect("must not hang")
        .unwrap();
    assert_eq!(reply, "denied, no job spawned");
}