//! Background sub-agent registry.
//!
//! Tracks sub-agents spawned via `InvokeAgent`. Post-#1163 (Lean A)
//! every `InvokeAgent` dispatch is a background spawn — the foreground
//! / blocking variant (and its `background:bool` parameter) was
//! deleted to match Codex's `spawn_agent` shape and Claude Code's
//! `TaskCreate` shape. The inference loop drains completed results
//! and injects them as user-role messages so the model sees them on
//! the next iteration.
//!
//! ## Lifecycle
//!
//! 1. **Spawn**: every `InvokeAgent` call creates a tokio task and
//!    returns its `task_id` immediately (no blocking variant)
//! 2. **Track**: the task handle + metadata are stored in `ChildAgentRegistry`
//! 3. **Poll**: before each inference call, the loop calls `drain_completed()`
//! 4. **Inject**: completed results are appended as user messages
//! 5. **Cleanup**: on registry drop, all pending task handles are aborted —
//!    no orphan futures, no leaked worktrees. (Phase 1 of #1022, B3.)
//!
//! ## Cancellation cascade
//!
//! Bg-agent tasks receive a `CancellationToken` derived from the parent's
//! token via `child_token()` (wired in `crate::sub_agent_dispatch`). When
//! the parent is cancelled, every bg child sees it; when the registry
//! drops without cancellation, [`tokio_util::task::AbortOnDropHandle`]
//! still aborts the futures so we never leak. Both paths are covered.
//! (Phase 1 of #1022, B2+B3.)
//!
//! ## Thread safety
//!
//! The registry is wrapped in `Arc<Mutex<>>` and shared between the main
//! inference loop and the background task spawner.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
// **#1022 B16**: was `std::sync::Mutex`. Switched to `parking_lot::Mutex`
// for three reasons:
//   1. **No poisoning** — if a thread panics while holding the lock,
//      subsequent calls don't get a `PoisonError`. The bg registry
//      is shared between the main inference loop and N spawned tasks;
//      a panic in one critical section bricking every subsequent
//      drain would be a particularly bad failure mode.
//   2. **Faster on contention** — no atomic check for poison flag,
//      no `Result` allocation. The contention is real: `drain_completed`
//      runs on every loop iteration.
//   3. **Cleaner API** — `.lock()` returns a guard directly, no
//      `.unwrap()` boilerplate at every call site.
// We deliberately keep this *sync* (not `tokio::sync::Mutex`) because
// the critical sections are short HashMap ops with no awaits inside.
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use tokio::sync::{mpsc, oneshot, watch};
use tokio_util::sync::CancellationToken;
use tokio_util::task::AbortOnDropHandle;

use crate::engine::EngineEvent;

// ── Layer 0 of #996 ──────────────────────────────────────────────────────
//
// Status enum + watch-channel plumbing + per-task cancel + snapshot API.
// Pure infrastructure: no slash commands, no LLM tools, no UI changes.
// Layers 1+ (slash commands, tools, status-bar pill) consume this surface.
//
// Modeled on Codex's `tokio::sync::watch::Receiver<AgentStatus>` pattern
// (codex-rs/core/src/session/mod.rs). The bg-agent task drives the
// `watch::Sender`; the registry stores the matching `Receiver` and exposes
// snapshots to whoever asks (slash command, LLM tool, status-bar pill).

/// Lifecycle of a single background sub-agent task.
///
/// The bg-agent future drives transitions through `watch::Sender<AgentStatus>`.
/// Initial value is [`AgentStatus::Pending`]; the future flips to `Running`
/// when execution actually starts and to one of the terminal variants
/// (`Completed`, `Errored`, `Cancelled`) when it finishes.
///
/// `Running.iter` reflects the current inference iteration. Background
/// agents emit live updates via Layer 4 (#1058); `0` is the entry-point
/// placeholder before the first iteration fires. There is no upper bound
/// (#1110 removed the sub-agent cap) — the counter widens to `u32` so
/// long-running agents can't silently wrap.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
#[non_exhaustive]
pub enum AgentStatus {
    /// Reserved but the spawned future hasn't started yet.
    Pending,
    /// Actively executing. `iter` is the current inference iteration
    /// (1..); `0` means "started, no iter info yet" (Layer 0 default).
    /// There is no upper bound — see #1110 / `LoopDetector` for termination.
    Running {
        /// Current inference iteration (1..). `0` is the entry-point
        /// placeholder emitted before the first iteration in
        /// `run_bg_agent`; background agents update this live
        /// (Layer 4, #1058). `u32` because #1110 removed the sub-agent
        /// iteration cap and a meandering weak model could plausibly
        /// exceed `u8::MAX` before context exhaustion.
        iter: u32,
    },
    /// User or parent fired the cancel token. Terminal.
    Cancelled,
    /// Sub-agent returned a final answer. Terminal.
    Completed {
        /// The agent's final output. Truncation for display is the
        /// renderer's job (see Codex's `COLLAB_AGENT_RESPONSE_PREVIEW_GRAPHEMES`).
        summary: String,
    },
    /// Sub-agent returned an error. Terminal.
    Errored {
        /// Error message as produced by `execute_sub_agent`. Same
        /// truncation note as `Completed.summary`.
        error: String,
    },
}

/// Snapshot of a pending bg-agent task — what `/agents` and the
/// `ListBackgroundTasks` LLM tool will render.
///
/// Cloned out of the registry under the lock so callers can format/display
/// without holding it. `age` is computed from `started_at` at snapshot time.
///
/// `#[non_exhaustive]`: external crates may add fields like `parent_task_id`,
/// `cancel_token`, or daemon-side metadata in future minor releases without
/// it being a breaking change. Constructed only inside `koda-core`, so the
/// attribute imposes zero friction on consumers (they only read fields).
///
/// `#[must_use]`: ignoring a snapshot is almost always a bug — the only
/// reason to call `snapshot()` is to inspect or render the contents.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
#[must_use = "a snapshot is only useful if you read its fields"]
pub struct ChildTaskSnapshot {
    /// Monotonic id assigned at `reserve()` time. Stable for the
    /// lifetime of the task; reused across snapshots.
    pub task_id: u32,
    /// Configured agent name (`explore`, `verify`, ...).
    pub agent_name: String,
    /// The prompt the parent delegated. Surfaced verbatim by
    /// `/agents -v`; truncation is the renderer's job.
    pub prompt: String,
    /// Wall-clock duration since the task was attached. Computed at
    /// snapshot time, so successive snapshots of the same task
    /// report different ages.
    pub age: Duration,
    /// Latest value from the task's `watch::Receiver<AgentStatus>`.
    pub status: AgentStatus,
    /// `true` if this entry represents a background sub-agent
    /// (auto-drains its result on a future iteration), `false` for
    /// foreground sub-agents (parent awaits inline).
    ///
    /// **PR-A0.5 of #1232**: added when the registry was generalized
    /// from bg-only to host both fg and bg child agents. Today every
    /// constructor still passes `true`; PR-A wires up the fg path
    /// that registers entries with `false`.
    ///
    /// Consumers that want a bg-only view (e.g. the `/agents`
    /// overlay's "truly backgrounded" filter) should check this
    /// field rather than assuming presence in the registry implies
    /// background.
    pub is_background: bool,
    /// Sub-agent task that spawned this bg-agent. `None` = top-level
    /// (the user's main conversation).
    ///
    /// **#996 Layer 2 / Model D**: tracked so that when a sub-agent
    /// exits, [`ChildAgentRegistry::cancel_for_spawner`] can fire the
    /// cancel token on every bg-agent it left behind. Mirrors
    /// Claude Code's `agentId` field on `LocalShellTaskState`
    /// (`prevents 10-day fake-logs.sh zombies`).
    ///
    /// We don't use this for permission scoping — any caller with a
    /// `task_id` can manage any task. Spawner is *only* a cleanup
    /// hook (matching Claude Code's flat-permissions design).
    pub spawner: Option<u32>,
}

impl ChildTaskSnapshot {
    /// Test-only constructor for downstream crates (e.g. `koda-cli`
    /// renderer tests). The `#[non_exhaustive]` attribute on the
    /// struct forbids brace-init from outside this crate; this
    /// helper preserves that invariant for production code while
    /// still letting renderer-layer tests build fixtures without
    /// spinning up a full `ChildAgentRegistry` + `tokio::spawn`.
    ///
    /// Hidden from rustdoc (`#[doc(hidden)]`) so it doesn't pollute
    /// the public surface; intentional opt-out of API stability.
    #[doc(hidden)]
    pub fn for_testing(
        task_id: u32,
        agent_name: String,
        prompt: String,
        age: std::time::Duration,
        status: AgentStatus,
        spawner: Option<u32>,
    ) -> Self {
        Self {
            task_id,
            agent_name,
            prompt,
            age,
            status,
            // PR-A0.5 of #1232: legacy test fixtures predate the fg/bg
            // split, so default to `true` to match historical behavior
            // (every test agent today is a bg agent). Tests that need
            // to assert fg behavior should use `for_testing_fg`.
            is_background: true,
            spawner,
        }
    }

    /// Test-only constructor for foreground child-agent fixtures.
    /// Same shape as [`Self::for_testing`] but with `is_background`
    /// flipped to `false`. Added in PR-A0.5 of #1232 so PR-A's
    /// fg-routing tests can assert overlay behavior under the new
    /// `is_background == false` path without resurrecting the
    /// brace-init bypass that `#[non_exhaustive]` forbids.
    #[doc(hidden)]
    pub fn for_testing_fg(
        task_id: u32,
        agent_name: String,
        prompt: String,
        age: std::time::Duration,
        status: AgentStatus,
        spawner: Option<u32>,
    ) -> Self {
        Self {
            task_id,
            agent_name,
            prompt,
            age,
            status,
            is_background: false,
            spawner,
        }
    }
}

/// Payload sent over the bg-agent oneshot.
///
/// Pre-#1022-B9 this was just `String` (the model's final output).
/// Now also carries the trace lines collected by
/// [`crate::engine::sink::BufferingSink`] so the inference loop
/// can surface them to the user when injecting the result.
///
/// The `Result<BgPayload, BgPayload>` shape preserves the prior
/// success/failure discrimination: `Ok` means `execute_sub_agent`
/// returned text, `Err` means it returned an error (the trace is
/// useful in *both* cases — the bg agent may have done several
/// steps before erroring).
pub type BgPayload = (String, Vec<String>);

/// A completed background agent result.
///
/// `#[non_exhaustive]`: future fields (e.g. token usage, latency, model
/// version) can be added without a breaking change. Constructed only
/// inside `koda-core`; external callers only read fields.
///
/// `#[must_use]`: an agent result that's never inspected is dead work.
#[derive(Debug)]
#[non_exhaustive]
#[must_use = "the whole point of running a bg agent is to inspect its result"]
pub struct BgAgentResult {
    /// The agent name that produced this result.
    pub agent_name: String,
    /// The original prompt that was delegated.
    pub prompt: String,
    /// The agent's output (or error message).
    pub output: String,
    /// Whether the agent succeeded.
    pub success: bool,
    /// **#1022 B9**: narrative trace lines captured by
    /// [`crate::engine::sink::BufferingSink`] inside the bg agent.
    /// Pre-fix this was implicitly always empty (bg agents ran with
    /// `NullSink`). Now populated with one line per significant
    /// event (tool start, info, auto-rejected approval) so the user
    /// can see what the bg agent did at result-injection time.
    /// Empty for the cancelled / panicked case (`output` carries the
    /// failure detail in those paths).
    pub events: Vec<String>,
    /// **#1108 P2a**: the `tool_call_id` of the parent's `InvokeAgent`
    /// call that started this bg agent. `Some` for normal spawns,
    /// `None` only for tests that bypass the dispatch path. Used by
    /// the inference loop's drain handler to persist `events` to
    /// `session_events` with this id as `parent_tool_call_id`, so
    /// the transcript renderer can fold the trace under the parent's
    /// `InvokeAgent` tool result.
    pub parent_tool_call_id: Option<String>,
}

/// Handle returned when a background agent is spawned.
///
/// Holds the task's [`tokio_util::task::AbortOnDropHandle`] so the
/// future is aborted if the registry is dropped before the task
/// completes (B3 of #1022). Also holds the per-task
/// [`CancellationToken`] so future per-task cancel commands
/// (`/cancel <id>` — see #996) have a hook to fire.
struct ChildAgentEntry {
    agent_name: String,
    prompt: String,
    rx: oneshot::Receiver<Result<BgPayload, BgPayload>>,
    /// Per-task cancel — derived as a `child_token()` of the parent
    /// session's token at spawn time. Firing this token (via
    /// [`ChildAgentRegistry::cancel`] for #996, or via the registry-drop
    /// path) causes the in-flight bg agent to observe `is_cancelled()`
    /// on its next loop iteration.
    cancel: CancellationToken,
    /// Live status channel — the spawned future writes; the registry
    /// reads at snapshot time. See [`AgentStatus`] for the lifecycle.
    status_rx: watch::Receiver<AgentStatus>,
    /// When the task was attached. Used to compute `age` in snapshots.
    started_at: Instant,
    /// Sub-agent task that spawned this bg-agent. `None` = top-level.
    /// **#996 Layer 2 / Model D** — see [`ChildTaskSnapshot::spawner`].
    spawner: Option<u32>,
    /// `true` for background agents, `false` for foreground. Surfaced
    /// via the snapshot's `is_background` field. PR-A0.5 of #1232.
    is_background: bool,
    /// **#1108 P2a**: parent's `InvokeAgent` tool_call_id. Stored at
    /// reserve time, surfaced via [`BgAgentResult::parent_tool_call_id`]
    /// at drain time. `None` only for legacy `attach()`-based tests.
    parent_tool_call_id: Option<String>,
    /// Aborts the spawned task on drop. The bg path uses
    /// `tokio::spawn` on the multi-thread runtime (#1022 B5):
    /// `execute_sub_agent` returns an explicitly `Send`-bounded
    /// future, so abort works promptly at any await point. The
    /// cancel-token cascade is still the primary stop signal
    /// (so the bg task can run any cleanup it owns).
    _handle: AbortOnDropHandle<()>,
}

/// Registry of running background sub-agents.
///
/// Shared via `Arc` between the inference loop (which drains results)
/// and the tool dispatch (which spawns agents).
pub struct ChildAgentRegistry {
    pending: Mutex<HashMap<u32, ChildAgentEntry>>,
    next_id: Mutex<u32>,
    /// Sender half of an unbounded mpsc that producers (every
    /// [`ChildStatusEmitter`] clone) call to deliver status events.
    ///
    /// **#1321 architecture**: pre-#1321 koda used a
    /// `Mutex<VecDeque<EngineEvent>>` polled once per parent
    /// inference iteration. That left bg-agent activity invisible for
    /// the entire duration of any long tool call (notably `WaitTask`,
    /// up to 300s) — events sat in the queue, then flushed in a
    /// stale flood when the wait returned.
    ///
    /// The mpsc shape mirrors Codex's `tx_event: Sender<Event>` (see
    /// `codex-rs/core/src/session/session.rs:17`) and Claude Code's
    /// `async function* runAgent` yield-based event flow: producers
    /// push, a single consumer pulls continuously, no shared queue
    /// to drain on a timer.
    ///
    /// **Unbounded** because bg-agent emissions are at human-readable
    /// rates (one per tool call, not per token), the consumer is
    /// always-on (the session-spawned forwarder), and back-pressuring
    /// a producer would block sub-agent inference loops mid-turn for
    /// reasons the model can't reason about.
    event_tx: mpsc::UnboundedSender<EngineEvent>,
    /// Receiver half. `Some` until [`Self::take_event_receiver`] hands
    /// it to the session forwarder, then `None` forever.
    ///
    /// Held in `Mutex<Option<_>>` so we can move it out exactly once.
    /// While the receiver still lives here (no forwarder attached —
    /// the headless / test path) callers can still drain events via
    /// [`Self::drain_status_events`] using `try_recv`. Once moved out,
    /// `drain_status_events` returns empty because the forwarder owns
    /// the rx and emits each event the moment it lands.
    event_rx: Mutex<Option<mpsc::UnboundedReceiver<EngineEvent>>>,
}

/// Fan-out helper for bg-agent status transitions.
///
/// Bundles the per-task `watch::Sender<AgentStatus>` (read by
/// `/agents` and the status-bar pill via the registry's snapshot
/// API) with a back-reference to the registry's event queue (drained
/// by the inference loop and forwarded to the engine sink — which is
/// what closes the #1076 boundary leak).
///
/// `Clone` is intentional so Layer 4 (`#1058`, live `iter` heartbeat)
/// can hold its own copy inside `execute_sub_agent` while
/// `run_bg_agent` keeps another for the entry / terminal transitions.
/// Both clones share the same `watch::Sender` and `Arc<registry>`,
/// so every `.send()` reaches both fan-out targets.
#[derive(Clone)]
pub struct ChildStatusEmitter {
    task_id: u32,
    spawner: Option<u32>,
    is_background: bool,
    status_tx: watch::Sender<AgentStatus>,
    registry: Arc<ChildAgentRegistry>,
}

impl ChildStatusEmitter {
    /// Construct from the parts handed back by [`ChildAgentRegistry::reserve`].
    ///
    /// The registry `Arc` is held for the lifetime of the bg agent,
    /// which is fine: the inference loop already keeps an `Arc` on
    /// the same registry, and registry drop is what aborts every
    /// in-flight bg task (B3 of #1022) — so an emitter outliving its
    /// registry is impossible by construction.
    pub fn new(
        task_id: u32,
        spawner: Option<u32>,
        is_background: bool,
        status_tx: watch::Sender<AgentStatus>,
        registry: Arc<ChildAgentRegistry>,
    ) -> Self {
        Self {
            task_id,
            spawner,
            is_background,
            status_tx,
            registry,
        }
    }

    /// Drive a status transition.
    ///
    /// Fans out to:
    /// 1. The per-task `watch::Sender` (so `snapshot()` / `/agents`
    ///    see the new state on the next read — no behavior change).
    /// 2. The registry's event queue, drained by the inference loop
    ///    and forwarded to the active `EngineSink` (so the TUI / ACP
    ///    / headless clients all see the same `ChildTaskUpdate` event).
    ///
    /// `watch::Sender::send` only fails if every receiver was dropped,
    /// which means the registry entry is gone — in that case the queue
    /// push is harmless (it'll be drained and ignored by clients that
    /// don't recognize the task id). We deliberately don't gate the
    /// queue push on the watch send result so a racing reap doesn't
    /// swallow the terminal `ChildTaskUpdate`.
    pub fn send(&self, status: AgentStatus) {
        let _ = self.status_tx.send(status.clone());
        self.registry
            .push_status_event(EngineEvent::ChildTaskUpdate {
                task_id: self.task_id,
                spawner: self.spawner,
                is_background: self.is_background,
                status,
            });
    }

    /// Forward a live activity event from inside the bg agent up to
    /// the parent's sink.
    ///
    /// **#1201 B**: paired with [`crate::engine::sink::ForwardingChildSink`],
    /// which decorates the bg agent's `BufferingSink` and calls this
    /// method whenever an interesting child event happens (tool
    /// start/end, info line). The event lands on the same registry
    /// queue as `ChildTaskUpdate`, so the inference loop's existing
    /// drain in [`crate::inference`] forwards it to whatever sink
    /// is active (TUI / headless / ACP) without further plumbing.
    ///
    /// Cheap: a single mutex acquisition + `VecDeque::push_back`,
    /// the same cost as [`Self::send`]. The activity feed runs at
    /// roughly tool-call frequency — not hot enough for any of this
    /// to register in a profile.
    pub fn send_activity(&self, kind: crate::engine::event::ChildAgentActivityKind) {
        self.registry
            .push_status_event(EngineEvent::ChildAgentActivity {
                task_id: self.task_id,
                spawner: self.spawner,
                // PR-A of #1232 §1: thread the emitter's stored
                // `is_background` flag through. Bg path passes `true`,
                // fg path passes `false`. Replaces PR-A0's hardcoded
                // `true` placeholder — the emitter has held the real
                // value since PR-A0.5.
                is_background: self.is_background,
                kind,
            });
    }

    /// Current status (read from the watch channel). Useful for
    /// terminal-disambiguation logic (e.g. "was this a cancel or a
    /// real error?") without taking the registry lock.
    pub fn current(&self) -> AgentStatus {
        self.status_tx.borrow().clone()
    }

    /// Test helper: peek the underlying watch sender. Production
    /// code should always go through [`Self::send`].
    #[cfg(test)]
    pub fn status_sender(&self) -> watch::Sender<AgentStatus> {
        self.status_tx.clone()
    }
}

/// Reservation slot returned by [`ChildAgentRegistry::reserve`].
///
/// The two-phase pattern (`reserve` → spawn → `attach`) lets the
/// dispatcher hand the oneshot sender into the spawned future
/// *before* the future exists, so the spawned closure can `move` it
/// without referencing the registry. The `cancel` token is a
/// `child_token()` of the parent's cancel — fires either when the
/// parent fires (cascade) or when this slot is individually cancelled
/// (future per-task `/cancel <id>` UX, #996).
pub struct ChildAgentReservation {
    /// Monotonically-assigned task ID. Surfaces in user-facing
    /// messages (`Background agent 'foo' started (agent:7)`) and
    /// keys the per-task `/cancel <id>` UX (#996).
    pub task_id: u32,
    /// Sender half of the result oneshot. Move into the spawned
    /// future so it can deliver `Ok(output)` / `Err(message)`.
    pub tx: oneshot::Sender<Result<BgPayload, BgPayload>>,
    /// Receiver half. Move back into the registry via [`ChildAgentRegistry::attach`]
    /// so `drain_completed()` can poll it.
    pub rx: oneshot::Receiver<Result<BgPayload, BgPayload>>,
    /// Per-task cancel token. Cloned for the spawned future
    /// (`bg_cancel`) and re-stored on the registry entry
    /// (`entry_cancel`); both halves observe parent cancellation
    /// because this is a `child_token()` of the parent.
    pub cancel: CancellationToken,
    /// Status sender — move into the spawned future. The future is
    /// the sole writer; it transitions through
    /// [`AgentStatus::Pending`] → `Running` → terminal.
    pub status_tx: watch::Sender<AgentStatus>,
    /// Status receiver — hand back to the registry via [`ChildAgentRegistry::attach`]
    /// so `snapshot()` and `/agents` can read the current state without
    /// touching the spawn site.
    pub status_rx: watch::Receiver<AgentStatus>,
    /// Sub-agent task id of the spawner, or `None` for the top-level
    /// loop. Carried verbatim to [`ChildAgentRegistry::attach`] so the
    /// entry knows who spawned it (Model D cleanup-on-exit).
    pub spawner: Option<u32>,
    /// `true` if this reservation is for a background sub-agent
    /// (auto-drains its result on a future iteration), `false` for
    /// foreground sub-agents (parent awaits inline).
    ///
    /// **PR-A0.5 of #1232**: added when the registry was generalized
    /// from bg-only to host both fg and bg child agents. The flag is
    /// carried into [`ChildAgentRegistry::attach`] so the registry
    /// entry's snapshots and the [`EngineEvent::ChildTaskUpdate`]
    /// events emitted by [`ChildStatusEmitter::send`] both reflect
    /// the same source-of-truth value.
    pub is_background: bool,
}

/// RAII handle returned from [`ChildAgentRegistry::register_fg_with_emitter`].
///
/// **PR-A of #1232 §1**: foreground sub-agents are owned by the
/// inline call that's awaiting them — there's no oneshot drain to
/// hook into for cleanup. Instead the parent holds this guard for
/// the duration of the await; on drop the registry entry is removed
/// so the `/agents` overlay stops showing the row.
///
/// Drop semantics:
/// * Successful return — guard dropped at scope exit, entry pulled.
/// * `?`-bubbled error — guard dropped during stack unwind, entry
///   pulled. Without RAII this path leaks the registry slot, which
///   the overlay would then render as a phantom "running" row.
/// * Panic — same as `?`. Mirrors `AbortOnDropHandle`'s drop-safety
///   story for the bg path.
///
/// The guard intentionally does *not* cancel the child token — fg
/// sub-agents return through the inline await, so the cancel-token
/// cascade isn't needed for cleanup. Cancellation comes from the
/// usual Ctrl-C path on the parent's loop.
pub struct FgRegistrationGuard {
    registry: Arc<ChildAgentRegistry>,
    task_id: u32,
}

impl FgRegistrationGuard {
    /// The registry id assigned at registration time. Stable for the
    /// lifetime of the guard. Useful for tests asserting that the
    /// emitter and the guard refer to the same entry.
    pub fn task_id(&self) -> u32 {
        self.task_id
    }
}

impl Drop for FgRegistrationGuard {
    fn drop(&mut self) {
        // Best-effort removal. If the registry was already drained
        // (shouldn't happen for fg — `drain_completed` skips us),
        // the `remove` is a no-op.
        self.registry.pending.lock().remove(&self.task_id);
    }
}

impl ChildAgentRegistry {
    /// Create an empty registry.
    pub fn new() -> Self {
        let (event_tx, event_rx) = mpsc::unbounded_channel();
        Self {
            pending: Mutex::new(HashMap::new()),
            next_id: Mutex::new(1),
            event_tx,
            event_rx: Mutex::new(Some(event_rx)),
        }
    }

    /// Push an event onto the status channel. Called by
    /// [`ChildStatusEmitter::send`]; not part of the public API.
    ///
    /// Direct mpsc send — the consumer (either the session-spawned
    /// forwarder or, in headless / test paths, [`Self::drain_status_events`])
    /// observes the event the moment it lands. No queue, no pump,
    /// no per-iteration drain latency.
    ///
    /// **Send failure** (channel closed) is logged at `debug` and
    /// dropped on the floor. The only way to close the unbounded
    /// channel is to drop the registry itself, in which case the
    /// process is shutting down and no consumer cares about late
    /// events anyway.
    pub(crate) fn push_status_event(&self, event: EngineEvent) {
        if let Err(e) = self.event_tx.send(event) {
            tracing::debug!("ChildAgentRegistry status channel closed; dropping event: {e}");
        }
    }

    /// Move the receiver out of the registry so the session can spawn
    /// a long-lived forwarder task that drives the channel directly.
    ///
    /// Returns `Some(rx)` exactly once over the registry's lifetime;
    /// every subsequent call returns `None`. Mirrors Codex's
    /// `forward_events` pattern (`codex-rs/core/src/codex_delegate.rs`):
    /// one consumer per channel, owned by a dedicated task that runs
    /// for as long as the registry / session lives.
    ///
    /// **Headless / test paths** that never call this can keep using
    /// [`Self::drain_status_events`] — the receiver stays in the
    /// registry's `Mutex<Option<_>>` and `try_recv`-drain works as
    /// before. Production (TUI / ACP) calls this exactly once at
    /// session startup so events flow continuously to the user's
    /// `EngineSink` regardless of what tool is currently in flight.
    pub fn take_event_receiver(&self) -> Option<mpsc::UnboundedReceiver<EngineEvent>> {
        self.event_rx.lock().take()
    }

    /// Drain queued status events synchronously.
    ///
    /// **Two modes**, transparent to the caller:
    ///
    /// 1. **Receiver still resident** (no forwarder attached yet —
    ///    headless, tests, the gap between session construction and
    ///    [`Self::take_event_receiver`]): pulls every queued event via
    ///    `try_recv` until the channel is empty, returns them in FIFO
    ///    order. Cheap: one mutex acquisition + a `try_recv` loop.
    ///
    /// 2. **Receiver moved to the forwarder**: returns an empty `Vec`
    ///    — the forwarder owns the rx and is emitting each event the
    ///    moment it lands; there is nothing left to drain here.
    ///
    /// Kept in the public API so existing callers (test suites, the
    /// pre-#1321 inference-loop drain we are about to delete) keep
    /// compiling without a rewrite. New code should rely on the
    /// session forwarder; only call this if you know the receiver
    /// hasn't been taken.
    pub fn drain_status_events(&self) -> Vec<EngineEvent> {
        let mut guard = self.event_rx.lock();
        let Some(rx) = guard.as_mut() else {
            return Vec::new();
        };
        let mut out = Vec::new();
        while let Ok(ev) = rx.try_recv() {
            out.push(ev);
        }
        out
    }

    /// Reserve a task ID and produce a oneshot sender + child cancel
    /// token for the spawn site to consume. Call [`Self::attach`] with
    /// the resulting `JoinHandle` to complete registration.
    ///
    /// `spawner` records who is reserving this slot. `None` means
    /// the top-level inference loop; `Some(invocation_id)` means a
    /// sub-agent invocation. Used by [`Self::cancel_for_spawner`] to
    /// reap children when a sub-agent exits (Model E cleanup-on-exit)
    /// and by [`Self::snapshot_for_caller`] to scope LLM-tool views.
    ///
    /// The two-phase shape (`reserve` → spawn → `attach`) exists
    /// because `tokio::spawn` produces the `JoinHandle` *after* the
    /// future is built, but the future needs to own the `tx` to deliver
    /// its result. Reservation gives us `tx` early; attach binds the
    /// handle once it exists.
    pub fn reserve(
        &self,
        parent_cancel: &CancellationToken,
        spawner: Option<u32>,
        is_background: bool,
    ) -> ChildAgentReservation {
        let (tx, rx) = oneshot::channel();
        let (status_tx, status_rx) = watch::channel(AgentStatus::Pending);
        let mut id = self.next_id.lock();
        let task_id = *id;
        *id += 1;
        ChildAgentReservation {
            task_id,
            tx,
            rx,
            cancel: parent_cancel.child_token(),
            status_tx,
            status_rx,
            spawner,
            is_background,
        }
    }

    /// Bind a spawned task's metadata to a previously [`reserve`]d slot.
    ///
    /// `rx` must be the receiver paired with the `tx` handed out by
    /// `reserve`. Holding `handle` as `AbortOnDropHandle` ensures the
    /// task is aborted on registry drop (B3 of #1022). `status_rx`
    /// is the read half of the watch channel whose write half
    /// (`status_tx`) was moved into the spawned future.
    ///
    /// [`reserve`]: Self::reserve
    //
    // 9 args trips `clippy::too_many_arguments` (limit 7). Each one
    // is load-bearing: id + name + prompt are display metadata;
    // rx/cancel/status_rx are the three channels we own; spawner is
    // the cleanup-routing key (Model E); handle is the
    // AbortOnDropHandle. Bundling into a struct just to satisfy
    // a heuristic would add a one-use type for no readability win
    // — "practicality beats purity".
    #[allow(clippy::too_many_arguments)]
    pub fn attach(
        &self,
        reservation_id: u32,
        agent_name: &str,
        prompt: &str,
        rx: oneshot::Receiver<Result<BgPayload, BgPayload>>,
        cancel: CancellationToken,
        status_rx: watch::Receiver<AgentStatus>,
        spawner: Option<u32>,
        is_background: bool,
        parent_tool_call_id: Option<String>,
        handle: tokio::task::JoinHandle<()>,
    ) {
        self.pending.lock().insert(
            reservation_id,
            ChildAgentEntry {
                agent_name: agent_name.to_string(),
                prompt: prompt.to_string(),
                rx,
                cancel,
                status_rx,
                started_at: Instant::now(),
                spawner,
                is_background,
                parent_tool_call_id,
                _handle: AbortOnDropHandle::new(handle),
            },
        );
    }

    /// Register a **foreground** sub-agent and hand back the live
    /// signal lines for it.
    ///
    /// **PR-A of #1232 §1**: the bg path uses `reserve` + `attach`,
    /// which provisions a oneshot result channel and an
    /// `AbortOnDropHandle` so the inference loop can drain results
    /// asynchronously. Foreground sub-agents don't need any of that
    /// — the parent is *already blocked* awaiting the inline call,
    /// so the result returns through the normal function return path.
    /// What the parent *does* need is the registry membership so the
    /// `/agents` overlay can render an activity row, plus an emitter
    /// so child events fan out to the same `ChildAgentActivity`
    /// stream that bg uses.
    ///
    /// Returns `(emitter, guard)`:
    /// * `emitter` — use `.send(...)` to push status transitions
    ///   (`Pending` → `Running { iter }` → terminal) and
    ///   `.send_activity(...)` to fan out tool/info events. The
    ///   emitter's `is_background` is wired to `false` so every
    ///   `ChildTaskUpdate`/`ChildAgentActivity` event the fg path
    ///   emits is correctly tagged.
    /// * `guard` — a [`FgRegistrationGuard`] that removes the entry
    ///   from the registry when dropped. Hold it for the duration of
    ///   the inline sub-agent call. RAII over manual unregister so a
    ///   panic / `?`-bubble doesn't leak the entry. (Mirrors
    ///   `AbortOnDropHandle`'s drop-safety story for the bg path.)
    ///
    /// The entry is inserted with a noop oneshot sender that never
    /// fires + a noop spawn handle. `drain_completed` skips fg
    /// entries unconditionally (see its doc), so neither the dummy
    /// rx nor the dummy handle ever surface to a consumer.
    pub fn register_fg_with_emitter(
        self: &Arc<Self>,
        agent_name: &str,
        prompt: &str,
        parent_cancel: &CancellationToken,
        spawner: Option<u32>,
        parent_tool_call_id: Option<String>,
    ) -> (ChildStatusEmitter, FgRegistrationGuard) {
        let (_dummy_tx, dummy_rx) = oneshot::channel::<Result<BgPayload, BgPayload>>();
        // Leak the sender into the entry's environment via a parked
        // task so `try_recv` always returns `Empty`, not `Closed`.
        // We can't actually keep `_dummy_tx` alive that way without a
        // task — simpler: just drop it here, accept that try_recv
        // would return `Closed`, and rely on `drain_completed`'s
        // `is_background` early-skip (the whole reason that skip
        // exists). Belt + suspenders against a future refactor that
        // accidentally drops the skip.
        drop(_dummy_tx);

        let (status_tx, status_rx) = watch::channel(AgentStatus::Pending);
        let cancel = parent_cancel.child_token();
        let task_id = {
            let mut id = self.next_id.lock();
            let next = *id;
            *id += 1;
            next
        };
        let noop = tokio::spawn(async {});
        self.pending.lock().insert(
            task_id,
            ChildAgentEntry {
                agent_name: agent_name.to_string(),
                prompt: prompt.to_string(),
                rx: dummy_rx,
                cancel,
                status_rx,
                started_at: Instant::now(),
                spawner,
                is_background: false,
                parent_tool_call_id,
                _handle: AbortOnDropHandle::new(noop),
            },
        );

        let emitter = ChildStatusEmitter::new(task_id, spawner, false, status_tx, self.clone());
        let guard = FgRegistrationGuard {
            registry: self.clone(),
            task_id,
        };
        (emitter, guard)
    }

    /// Convenience for tests: register a synthetic entry without a
    /// real spawned task. The provided `tx` can be used to fire the
    /// result manually. The handle is a noop spawned task that
    /// returns immediately, so `_handle` has something to abort.
    #[cfg(test)]
    pub fn register_test(
        &self,
        agent_name: &str,
        prompt: &str,
    ) -> (u32, oneshot::Sender<Result<BgPayload, BgPayload>>) {
        let (id, tx, _status_tx, _cancel) =
            self.register_test_with_status(agent_name, prompt, None);
        (id, tx)
    }

    /// Test-only sibling of [`register_test`] that returns the status
    /// sender so a test can manually drive transitions without
    /// needing a real spawned `run_bg_agent`. The cancel token also
    /// comes back so cancel-cascade tests can verify the channel.
    ///
    /// `spawner` is recorded on the entry so scope-filtering /
    /// kill-on-exit tests can exercise both the top-level (`None`)
    /// and sub-agent (`Some(id)`) paths.
    #[cfg(test)]
    pub fn register_test_with_status(
        &self,
        agent_name: &str,
        prompt: &str,
        spawner: Option<u32>,
    ) -> (
        u32,
        oneshot::Sender<Result<BgPayload, BgPayload>>,
        watch::Sender<AgentStatus>,
        CancellationToken,
    ) {
        let (tx, rx) = oneshot::channel();
        let (status_tx, status_rx) = watch::channel(AgentStatus::Pending);
        let mut id = self.next_id.lock();
        let task_id = *id;
        *id += 1;
        drop(id);
        let cancel = CancellationToken::new();
        let cancel_observer = cancel.clone();
        let noop = tokio::spawn(async {});
        self.pending.lock().insert(
            task_id,
            ChildAgentEntry {
                agent_name: agent_name.to_string(),
                prompt: prompt.to_string(),
                rx,
                cancel,
                status_rx,
                started_at: Instant::now(),
                spawner,
                // Test-only entries default to bg-true (legacy behavior).
                // PR-A0.5 of #1232.
                is_background: true,
                parent_tool_call_id: None,
                _handle: AbortOnDropHandle::new(noop),
            },
        );
        (task_id, tx, status_tx, cancel_observer)
    }

    /// Drain all completed background agents. Non-blocking — only takes
    /// entries whose oneshot has already resolved.
    ///
    /// **PR-A of #1232 §1**: foreground entries (`is_background == false`)
    /// are skipped. The fg dispatch path returns the sub-agent's output
    /// inline through its own `await`, so injecting a duplicate result
    /// via the bg drain channel would corrupt the parent's transcript.
    /// Defense-in-depth: today the fg registration path doesn't even
    /// hold a sender that can fire, so the `try_recv` would only ever
    /// return `Empty` or `Closed` — but `Closed` would manufacture a
    /// fake "cancelled" `BgAgentResult` (see the `Closed` arm below),
    /// which the inference loop would then inject into the parent. The
    /// `is_background` skip here makes that whole class of bug
    /// impossible by construction rather than relying on the fg path's
    /// discipline to never close its sender.
    pub fn drain_completed(&self) -> Vec<BgAgentResult> {
        let mut guard = self.pending.lock();
        let mut completed = Vec::new();
        let mut done_ids = Vec::new();

        for (id, entry) in guard.iter_mut() {
            if !entry.is_background {
                // PR-A of #1232 §1: see fn-level doc.
                continue;
            }
            match entry.rx.try_recv() {
                Ok(Ok((output, events))) => {
                    done_ids.push(*id);
                    completed.push(BgAgentResult {
                        agent_name: entry.agent_name.clone(),
                        prompt: entry.prompt.clone(),
                        output,
                        success: true,
                        events,
                        parent_tool_call_id: entry.parent_tool_call_id.clone(),
                    });
                }
                Ok(Err((err, events))) => {
                    done_ids.push(*id);
                    completed.push(BgAgentResult {
                        agent_name: entry.agent_name.clone(),
                        prompt: entry.prompt.clone(),
                        output: err,
                        success: false,
                        events,
                        parent_tool_call_id: entry.parent_tool_call_id.clone(),
                    });
                }
                Err(oneshot::error::TryRecvError::Empty) => {
                    // Still running
                }
                Err(oneshot::error::TryRecvError::Closed) => {
                    // Sender dropped without sending — task panicked or was cancelled.
                    // No events available (the buffering sink died with the task).
                    done_ids.push(*id);
                    completed.push(BgAgentResult {
                        agent_name: entry.agent_name.clone(),
                        prompt: entry.prompt.clone(),
                        output: "[background agent task was cancelled]".to_string(),
                        success: false,
                        events: Vec::new(),
                        parent_tool_call_id: entry.parent_tool_call_id.clone(),
                    });
                }
            }
        }

        for id in done_ids {
            guard.remove(&id);
        }

        completed
    }

    /// How many background agents are still running.
    pub fn pending_count(&self) -> usize {
        self.pending.lock().len()
    }

    // ── Layer 0 of #996: per-task cancel + snapshot ───────────────────────────────

    /// Fire the cancel token for a single task.
    ///
    /// Returns `true` if a pending task with that id existed and was
    /// signalled, `false` if the id is unknown (already drained,
    /// completed, or never registered). Idempotent: calling twice on
    /// the same id is safe — [`CancellationToken::cancel`] is itself
    /// idempotent.
    ///
    /// The entry stays in `pending` until the spawned future actually
    /// observes the token and finishes. `drain_completed()` then
    /// reaps it via the closed-sender path (or the future's terminal
    /// `tx.send` if it noticed and shut down cleanly).
    pub fn cancel(&self, task_id: u32) -> bool {
        let guard = self.pending.lock();
        match guard.get(&task_id) {
            Some(entry) => {
                entry.cancel.cancel();
                true
            }
            None => false,
        }
    }

    /// Snapshot every pending task's metadata for `/agents` and the
    /// `ListBackgroundTasks` LLM tool.
    ///
    /// `age` is computed against `Instant::now()` at call time, so two
    /// snapshots of the same task report different ages. Status is read
    /// from each entry's `watch::Receiver` (no blocking, no waiting).
    /// Sorted by ascending `task_id` so the output is stable across calls.
    ///
    /// **Unscoped**: returns every task regardless of spawner. Used by
    /// the TUI `/agents` command (humans want the global view) and as
    /// the engine of [`Self::snapshot_for_caller`] (which filters).
    pub fn snapshot(&self) -> Vec<ChildTaskSnapshot> {
        let guard = self.pending.lock();
        let now = Instant::now();
        let mut out: Vec<_> = guard
            .iter()
            .map(|(id, entry)| ChildTaskSnapshot {
                task_id: *id,
                agent_name: entry.agent_name.clone(),
                prompt: entry.prompt.clone(),
                age: now.saturating_duration_since(entry.started_at),
                status: entry.status_rx.borrow().clone(),
                is_background: entry.is_background,
                spawner: entry.spawner,
            })
            .collect();
        out.sort_by_key(|s| s.task_id);
        out
    }

    /// Scoped snapshot for the `ListBackgroundTasks` LLM tool.
    ///
    /// **Model E scoping**: a caller only sees tasks whose `spawner`
    /// matches its own `caller_spawner`. Top-level callers pass `None`
    /// and see only top-level-spawned tasks; sub-agent callers pass
    /// `Some(invocation_id)` and see only their own.
    ///
    /// Strict equality — a sub-agent does NOT see sibling sub-agents'
    /// tasks, and the top-level does NOT see sub-agents' tasks via the
    /// LLM (the TUI's `/agents` command remains the global view).
    pub fn snapshot_for_caller(&self, caller_spawner: Option<u32>) -> Vec<ChildTaskSnapshot> {
        self.snapshot()
            .into_iter()
            .filter(|s| s.spawner == caller_spawner)
            .collect()
    }

    /// Clone the status receiver for a task so callers can await
    /// [`watch::Receiver::changed`] without holding the lock.
    ///
    /// Returns `None` if the task has already been drained from the registry.
    /// Primarily used by tests to observe live iteration-counter updates
    /// without polling `snapshot()` in a tight loop.
    pub fn subscribe(&self, task_id: u32) -> Option<watch::Receiver<AgentStatus>> {
        let guard = self.pending.lock();
        guard.get(&task_id).map(|e| e.status_rx.clone())
    }

    // ── Layer 2 of #996 ───────────────────────────────────────────────
    //
    // Scoped cancel + cleanup-on-exit + WaitTask machinery.
    // The unscoped [`Self::cancel`] above stays for the TUI (humans get
    // the global view); LLM tools route through [`Self::cancel_as_caller`]
    // so a sub-agent can't reach across into a sibling's task.

    /// Scoped per-task cancel for the `CancelTask` LLM tool.
    ///
    /// **Model E permission rule** — `caller_spawner` must equal the
    /// task's `spawner`, with `None == None` (top-level can cancel
    /// top-level tasks; sub-agent invocation N can cancel only its
    /// own tasks). Returns [`CancelOutcome::Forbidden`] otherwise.
    ///
    /// The unscoped [`Self::cancel`] is the TUI's contract — humans
    /// at the keyboard implicitly have full authority. This method is
    /// the LLM's contract.
    pub fn cancel_as_caller(&self, task_id: u32, caller_spawner: Option<u32>) -> CancelOutcome {
        let guard = self.pending.lock();
        match guard.get(&task_id) {
            None => CancelOutcome::NotFound,
            Some(entry) if entry.spawner != caller_spawner => CancelOutcome::Forbidden,
            Some(entry) => {
                entry.cancel.cancel();
                CancelOutcome::Cancelled
            }
        }
    }

    /// Fire the cancel token on every task whose `spawner` matches.
    ///
    /// Called from the sub-agent dispatch path when an invocation
    /// exits (Model E cleanup-on-exit). Returns the number of tasks
    /// signalled, purely for tracing — the actual reaping happens
    /// later via [`Self::drain_completed`] once the futures observe
    /// their cancel tokens and finish.
    ///
    /// Idempotent: re-calling on the same `spawner` after all tasks
    /// have been reaped returns `0`.
    pub fn cancel_for_spawner(&self, spawner: u32) -> usize {
        let guard = self.pending.lock();
        let mut count = 0;
        for entry in guard.values() {
            if entry.spawner == Some(spawner) {
                entry.cancel.cancel();
                count += 1;
            }
        }
        count
    }
}

/// Outcome of [`ChildAgentRegistry::cancel_as_caller`].
///
/// Mirrors HTTP-ish status codes so the LLM-tool layer can produce
/// useful error messages without inspecting registry internals.
///
/// `#[must_use]`: silently dropping the outcome of a cancel call is
/// always a bug — the dispatch layer translates each variant into a
/// distinct LLM-tool response payload.
///
/// Note: deliberately NOT `#[non_exhaustive]`. The `koda-cli` rendering
/// layer matches every variant exhaustively, and the compiler-driven
/// missed-variant warning we get from that is a feature, not a wart.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[must_use = "every CancelOutcome maps to a distinct dispatch-layer response"]
pub enum CancelOutcome {
    /// Task existed, caller owned it, cancel token fired.
    Cancelled,
    /// No task with that id (already drained, never registered, or
    /// completed and reaped).
    NotFound,
    /// Task exists but caller's `spawner` doesn't match. The LLM
    /// surface translates this into a permission-style error.
    Forbidden,
}

impl Default for ChildAgentRegistry {
    fn default() -> Self {
        Self::new()
    }
}

impl Drop for ChildAgentRegistry {
    /// Abort every still-pending bg task on registry drop.
    ///
    /// `AbortOnDropHandle::drop` does the work — this impl exists
    /// only to make the lifecycle explicit and to give a single
    /// place to add telemetry later.
    fn drop(&mut self) {
        // **#1022 B16**: simplified post-parking_lot. The pre-fix
        // version had to handle `PoisonError` (via
        // `match get_mut() { Ok | Err(into_inner()) }`) because a
        // panic-while-held would poison `std::sync::Mutex`.
        // `parking_lot::Mutex` doesn't poison, so the cleanup path
        // is now the obvious one: take the map, log if non-empty,
        // let `AbortOnDropHandle::drop` do the actual abort work.
        let map = std::mem::take(&mut *self.pending.lock());
        if !map.is_empty() {
            tracing::debug!(
                count = map.len(),
                "ChildAgentRegistry dropped with pending tasks; aborting"
            );
        }
        // Map drops here → each entry's `AbortOnDropHandle` aborts
        // its task. No orphans. No leaked worktrees.
    }
}

/// Wrap in Arc for sharing between inference loop and tool dispatch.
pub fn new_shared() -> Arc<ChildAgentRegistry> {
    Arc::new(ChildAgentRegistry::new())
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::time::Duration;

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn register_and_complete() {
        let reg = ChildAgentRegistry::new();
        let (task_id, tx) = reg.register_test("explore", "find all tests");
        assert_eq!(task_id, 1);
        assert_eq!(reg.pending_count(), 1);

        // Not yet complete
        assert!(reg.drain_completed().is_empty());

        // Complete it
        tx.send(Ok(("found 42 tests".to_string(), Vec::new())))
            .unwrap();
        let results = reg.drain_completed();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].agent_name, "explore");
        assert_eq!(results[0].output, "found 42 tests");
        assert!(results[0].success);
        assert_eq!(reg.pending_count(), 0);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn drain_only_completed() {
        let reg = ChildAgentRegistry::new();
        let (_id1, tx1) = reg.register_test("task", "build");
        let (_id2, _tx2) = reg.register_test("explore", "search");

        tx1.send(Ok(("done".to_string(), Vec::new()))).unwrap();

        let results = reg.drain_completed();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].agent_name, "task");
        assert_eq!(reg.pending_count(), 1); // explore still pending
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn dropped_sender_reports_cancelled() {
        let reg = ChildAgentRegistry::new();
        let (_id, tx) = reg.register_test("task", "build");
        drop(tx); // simulate task panic/cancel

        let results = reg.drain_completed();
        assert_eq!(results.len(), 1);
        assert!(!results[0].success);
        assert!(results[0].output.contains("cancelled"));
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn error_result() {
        let reg = ChildAgentRegistry::new();
        let (_id, tx) = reg.register_test("verify", "check");
        tx.send(Err(("test failures".to_string(), Vec::new())))
            .unwrap();

        let results = reg.drain_completed();
        assert_eq!(results.len(), 1);
        assert!(!results[0].success);
        assert_eq!(results[0].output, "test failures");
    }

    /// #1022 B9 regression: the narrative trace captured by
    /// `BufferingSink` inside the bg agent must propagate through
    /// the oneshot → registry → `BgAgentResult.events`. Pre-fix
    /// this field didn't exist; bg agents ran with `NullSink` and
    /// the user only saw spawn + completion lines. The fix is
    /// useless if the trace gets dropped at any of the three hops,
    /// so this test pins the round-trip end-to-end.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn events_propagate_through_drain_for_success() {
        let reg = ChildAgentRegistry::new();
        let (_id, tx) = reg.register_test("explore", "map repo");
        let trace = vec![
            "  \u{1f527} Read".to_string(),
            "  \u{1f527} Grep".to_string(),
            "  \u{26a1} cache hit".to_string(),
        ];
        tx.send(Ok(("map result".to_string(), trace.clone())))
            .unwrap();

        let results = reg.drain_completed();
        assert_eq!(results.len(), 1);
        assert!(results[0].success);
        assert_eq!(
            results[0].events, trace,
            "trace lost between sender and BgAgentResult"
        );
    }

    /// #1022 B9 regression: trace must propagate even when the bg
    /// agent failed. The trace is *most* useful in the failure case
    /// — "the agent tried Read, Bash, Edit, then errored" is the
    /// kind of breadcrumb that turns a black-box failure into a
    /// debuggable one.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn events_propagate_through_drain_for_failure() {
        let reg = ChildAgentRegistry::new();
        let (_id, tx) = reg.register_test("build", "compile");
        let trace = vec![
            "  \u{1f527} Bash".to_string(),
            "  \u{2398} approval auto-rejected for Delete (no user channel)".to_string(),
        ];
        tx.send(Err(("compile failed".to_string(), trace.clone())))
            .unwrap();

        let results = reg.drain_completed();
        assert_eq!(results.len(), 1);
        assert!(!results[0].success);
        assert_eq!(results[0].events, trace);
    }

    /// #1022 B9 corollary: cancelled / panicked tasks have *no*
    /// trace available (the buffering sink died with the task), and
    /// that's an explicitly-empty Vec rather than uninitialized.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn cancelled_task_has_empty_event_trace() {
        let reg = ChildAgentRegistry::new();
        let (_id, tx) = reg.register_test("flaky", "x");
        drop(tx); // simulate panic / abort
        let results = reg.drain_completed();
        assert_eq!(results.len(), 1);
        assert!(!results[0].success);
        assert!(
            results[0].events.is_empty(),
            "cancel path must yield empty trace"
        );
    }

    /// Phase 1 of #1022, B3 regression test: dropping the registry
    /// must abort still-running spawned tasks. Without
    /// `AbortOnDropHandle` (or an explicit `JoinHandle::abort` in
    /// `Drop`), the spawned future would keep running after the
    /// registry — and any worktrees / API tokens / writes it owns —
    /// were dropped. That's the leak we're fixing.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn registry_drop_aborts_pending_tasks() {
        let reg = ChildAgentRegistry::new();
        let parent = CancellationToken::new();
        let reservation = reg.reserve(&parent, None, true);
        let task_id = reservation.task_id;
        let cancel_for_task = reservation.cancel.clone();
        let tx = reservation.tx;
        let rx = reservation.rx;
        let cancel_for_entry = reservation.cancel;
        let status_rx = reservation.status_rx;

        // Use a flag the task sets only if it ever finishes a full
        // sleep. If abort works, the flag stays false even though
        // we wait long enough for a non-aborted task to finish.
        let ran_to_completion = Arc::new(AtomicBool::new(false));
        let flag = ran_to_completion.clone();

        // Deterministic abort signal (#1312): a drop guard inside
        // the spawned task fires `aborted_tx` whenever the task
        // unwinds — cooperative completion, parent cancel, or hard
        // abort. The receiver lets the test wait for *the actual
        // abort* instead of guessing a settle time.
        //
        // Why a drop guard (and not just a `tx.send()` after the
        // `select!`): `JoinHandle::abort()` does **not** run the
        // task's normal continuation — it polls the future, sees
        // it cancelled, and drops it. Code after the await never
        // runs. The drop guard fires from within the task's
        // destructor, which DOES run on abort. This is the only
        // shape that detects abort vs. completion.
        struct AbortSignal(Option<oneshot::Sender<()>>);
        impl Drop for AbortSignal {
            fn drop(&mut self) {
                if let Some(tx) = self.0.take() {
                    let _ = tx.send(());
                }
            }
        }
        let (aborted_tx, aborted_rx) = oneshot::channel::<()>();
        let abort_signal = AbortSignal(Some(aborted_tx));

        let handle = tokio::spawn(async move {
            // Move the guard into the task so its Drop fires when
            // the task unwinds (whether via normal completion,
            // cooperative cancel, or JoinHandle::abort()).
            let _abort_signal = abort_signal;
            // Either the cancel token fires (parent cascade) or we
            // get aborted (drop cascade). The slow sleep just gives
            // the test time to drop the registry before we'd
            // naturally finish.
            tokio::select! {
                _ = cancel_for_task.cancelled() => {}
                _ = tokio::time::sleep(Duration::from_secs(60)) => {
                    flag.store(true, Ordering::SeqCst);
                }
            }
            let _ = tx.send(Ok(("done".to_string(), Vec::new())));
        });
        reg.attach(
            task_id,
            "explore",
            "long task",
            rx,
            cancel_for_entry,
            status_rx,
            None,
            true,
            None,
            handle,
        );

        // No "give the task a tick to start" sleep needed: `attach`
        // calls `self.pending.lock().insert(...)` synchronously, so
        // `pending_count` is incremented by the time `attach` returns.
        // (Pre-#1312 this was a 20ms blind sleep based on the (false)
        // assumption that the spawned task had to start before the
        // count moved.)
        assert_eq!(reg.pending_count(), 1);

        // Drop the registry — this must abort the spawned task.
        drop(reg);

        // Deterministic readiness (#1312 was: blind `sleep(100ms)`).
        // Wait for the AbortSignal drop guard to fire. If abort
        // landed, the guard runs from the task's destructor and
        // resolves `aborted_rx` immediately. If abort did NOT land,
        // the receiver hangs (because the task is still running its
        // 60s sleep and the guard hasn't dropped) and the timeout
        // bails with a loud message — a real regression signal, not
        // a flake.
        tokio::time::timeout(Duration::from_secs(2), aborted_rx)
            .await
            .expect(
                "AbortOnDropHandle did not abort the spawned task within 2s of \
                 dropping the registry — task destructor never ran",
            )
            .expect("AbortSignal sender dropped without sending; task body panicked?");
        assert!(
            !ran_to_completion.load(Ordering::SeqCst),
            "task slept to completion — AbortOnDropHandle did not abort it"
        );
    }

    /// Phase 1 of #1022, B2 regression test: cancelling the parent
    /// token must cascade to bg-agent child tokens handed out by
    /// `reserve`.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn parent_cancel_cascades_to_reserved_child() {
        let reg = ChildAgentRegistry::new();
        let parent = CancellationToken::new();
        let r1 = reg.reserve(&parent, None, true);
        let r2 = reg.reserve(&parent, None, true);

        assert!(!r1.cancel.is_cancelled());
        assert!(!r2.cancel.is_cancelled());

        parent.cancel();

        assert!(
            r1.cancel.is_cancelled(),
            "child 1 token should observe parent cancel"
        );
        assert!(
            r2.cancel.is_cancelled(),
            "child 2 token should observe parent cancel"
        );
    }

    // ── Layer 0 of #996 ──────────────────────────────────────────────────────
    //
    // Status channel + per-task cancel + snapshot.

    /// `cancel(task_id)` must fire that task's cancel token.
    /// This is the hook the future `/cancel <id>` slash command and
    /// `CancelAgent` LLM tool will call. Verifies a known id returns
    /// true *and* the underlying token actually fires.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn cancel_known_task_fires_token() {
        let reg = ChildAgentRegistry::new();
        let (task_id, _tx, _status_tx, observer) =
            reg.register_test_with_status("explore", "map repo", None);

        assert!(!observer.is_cancelled(), "precondition");
        let fired = reg.cancel(task_id);
        assert!(fired, "cancel(known_id) should report success");
        assert!(
            observer.is_cancelled(),
            "the task's cancel token should observe the cancellation"
        );
    }

    /// `cancel` on an unknown / already-drained id must return false
    /// instead of panicking. The slash command and LLM tool will
    /// surface this to the user as "no such task".
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn cancel_unknown_task_returns_false() {
        let reg = ChildAgentRegistry::new();
        assert!(
            !reg.cancel(999),
            "cancel of an unknown id should be a no-op returning false"
        );
    }

    /// `cancel` is idempotent — calling twice on the same id is safe
    /// (the underlying [`CancellationToken::cancel`] is itself
    /// idempotent). Both calls return true while the entry is still
    /// in `pending`; a third call after drain returns false.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn cancel_is_idempotent_while_pending() {
        let reg = ChildAgentRegistry::new();
        let (task_id, _tx, _status_tx, _observer) =
            reg.register_test_with_status("explore", "x", None);

        assert!(reg.cancel(task_id));
        assert!(
            reg.cancel(task_id),
            "second cancel should still find the entry and report success"
        );
    }

    /// `snapshot()` must return one entry per pending task with
    /// stable ordering by `task_id`. Status defaults to `Pending`
    /// because no spawned future has flipped it yet.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn snapshot_lists_pending_tasks_in_id_order() {
        let reg = ChildAgentRegistry::new();
        let (id_a, _tx_a) = reg.register_test("explore", "map");
        let (id_b, _tx_b) = reg.register_test("verify", "check");

        let snap = reg.snapshot();
        assert_eq!(snap.len(), 2);
        // Ordering is by ascending task_id, regardless of HashMap
        // iteration order — this is the contract `/agents` relies on.
        assert_eq!(snap[0].task_id, id_a);
        assert_eq!(snap[0].agent_name, "explore");
        assert_eq!(snap[0].prompt, "map");
        assert_eq!(snap[0].status, AgentStatus::Pending);
        assert_eq!(snap[1].task_id, id_b);
        assert_eq!(snap[1].agent_name, "verify");
        assert_eq!(snap[1].status, AgentStatus::Pending);
    }

    /// `snapshot()` reads the live status channel — a `status_tx.send`
    /// must be observable on the very next snapshot, with no polling
    /// or yielding required (`watch::Receiver::borrow` is sync).
    /// This is the contract that lets the status-bar pill (Layer 3)
    /// and live `/agents -v` (Layer 1) reflect transitions immediately.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn snapshot_reflects_status_writes() {
        let reg = ChildAgentRegistry::new();
        let (task_id, _tx, status_tx, _cancel) =
            reg.register_test_with_status("explore", "map", None);

        // Default is Pending.
        assert_eq!(reg.snapshot()[0].status, AgentStatus::Pending);

        // Flip to Running and observe.
        status_tx.send(AgentStatus::Running { iter: 3 }).unwrap();
        let snap = reg.snapshot();
        assert_eq!(snap.len(), 1);
        assert_eq!(snap[0].task_id, task_id);
        assert_eq!(snap[0].status, AgentStatus::Running { iter: 3 });

        // Flip to Completed and observe.
        status_tx
            .send(AgentStatus::Completed {
                summary: "42 files".to_string(),
            })
            .unwrap();
        assert_eq!(
            reg.snapshot()[0].status,
            AgentStatus::Completed {
                summary: "42 files".to_string()
            }
        );
    }

    /// `snapshot()` reports a sane `age` that grows monotonically.
    /// We don't assert exact values (CI clocks are jittery) — just
    /// that two successive snapshots show a non-decreasing age and
    /// that the value is non-negative (saturating subtraction
    /// prevents underflow if the system clock jumps backwards).
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn snapshot_age_is_monotonic() {
        let reg = ChildAgentRegistry::new();
        let (_id, _tx) = reg.register_test("explore", "x");

        let age1 = reg.snapshot()[0].age;
        tokio::time::sleep(Duration::from_millis(15)).await;
        let age2 = reg.snapshot()[0].age;
        assert!(
            age2 >= age1,
            "age should be monotonic non-decreasing across snapshots"
        );
    }

    /// `snapshot()` on an empty registry returns an empty Vec, not a
    /// panic and not None. `/agents` will use this to render "No
    /// background agents."
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn snapshot_empty_registry_is_empty_vec() {
        let reg = ChildAgentRegistry::new();
        assert!(reg.snapshot().is_empty());
    }

    /// Once a task is drained (completed and removed from `pending`),
    /// it disappears from `snapshot()` immediately. This pins the
    /// contract that `/agents` reflects the *currently-pending* set,
    /// not historical tasks. The Layer 1 "recently-completed lingers
    /// 30 s" UX is implemented at the *display* layer, not here.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn snapshot_drops_drained_tasks() {
        let reg = ChildAgentRegistry::new();
        let (_id, tx) = reg.register_test("explore", "x");
        assert_eq!(reg.snapshot().len(), 1);

        tx.send(Ok(("done".to_string(), Vec::new()))).unwrap();
        let _ = reg.drain_completed();

        assert!(
            reg.snapshot().is_empty(),
            "drained tasks must not appear in snapshots"
        );
    }

    // ── Layer 2 of #996: scoped APIs + WaitOutcome ────────────────────────

    /// `snapshot_for_caller(None)` returns only top-level tasks;
    /// `snapshot_for_caller(Some(N))` returns only N's tasks. Cross-spawner
    /// visibility is exactly zero — the Model E isolation guarantee.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn snapshot_for_caller_filters_by_spawner() {
        let reg = ChildAgentRegistry::new();
        let (top_id, _tx, _, _) = reg.register_test_with_status("a", "top", None);
        let (sub_a_id, _tx, _, _) = reg.register_test_with_status("b", "sub-a", Some(7));
        let (_sub_b_id, _tx, _, _) = reg.register_test_with_status("c", "sub-b", Some(9));

        let top = reg.snapshot_for_caller(None);
        assert_eq!(top.len(), 1);
        assert_eq!(top[0].task_id, top_id);

        let sub_a = reg.snapshot_for_caller(Some(7));
        assert_eq!(sub_a.len(), 1);
        assert_eq!(sub_a[0].task_id, sub_a_id);

        // Sub-agent 7 sees nothing of sub-agent 9's, of top-level's, etc.
        assert!(reg.snapshot_for_caller(Some(42)).is_empty());
    }

    /// `cancel_as_caller` enforces the Model E permission rule.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn cancel_as_caller_returns_forbidden_for_other_spawner() {
        let reg = ChildAgentRegistry::new();
        let (id, _tx, _, observer) = reg.register_test_with_status("x", "y", Some(7));

        // Wrong caller — not the top-level (None != Some(7)) and not a peer.
        assert_eq!(
            reg.cancel_as_caller(id, None),
            CancelOutcome::Forbidden,
            "top-level must not be able to cancel sub-agent's task"
        );
        assert_eq!(
            reg.cancel_as_caller(id, Some(99)),
            CancelOutcome::Forbidden,
            "sibling sub-agent must not be able to cancel"
        );
        assert!(
            !observer.is_cancelled(),
            "forbidden calls must NOT fire the cancel token"
        );

        // Correct caller — the original spawner.
        assert_eq!(reg.cancel_as_caller(id, Some(7)), CancelOutcome::Cancelled);
        assert!(observer.is_cancelled());
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn cancel_as_caller_returns_not_found_for_unknown_id() {
        let reg = ChildAgentRegistry::new();
        assert_eq!(reg.cancel_as_caller(999, None), CancelOutcome::NotFound);
    }

    /// `cancel_for_spawner` cleans up exactly one sub-agent's children
    /// and leaves siblings + top-level alone. The cleanup-on-exit hook.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn cancel_for_spawner_kills_only_matching_children() {
        let reg = ChildAgentRegistry::new();
        let (_top, _, _, top_obs) = reg.register_test_with_status("top", "t", None);
        let (_a1, _, _, a1_obs) = reg.register_test_with_status("a1", "x", Some(7));
        let (_a2, _, _, a2_obs) = reg.register_test_with_status("a2", "y", Some(7));
        let (_b, _, _, b_obs) = reg.register_test_with_status("b", "z", Some(9));

        let count = reg.cancel_for_spawner(7);
        assert_eq!(count, 2, "both of spawner 7's children must be signalled");

        assert!(a1_obs.is_cancelled());
        assert!(a2_obs.is_cancelled());
        assert!(!top_obs.is_cancelled(), "top-level must be untouched");
        assert!(!b_obs.is_cancelled(), "sibling spawner's task untouched");

        // Idempotent — calling again after entries are still alive
        // re-fires the (already-cancelled, no-op) tokens.
        assert_eq!(reg.cancel_for_spawner(7), 2);
        // Calling for an unknown spawner returns 0.
        assert_eq!(reg.cancel_for_spawner(99), 0);
    }

    // ── #1076: ChildStatusEmitter → sink fan-out ───────────────────────────────

    /// Helper: build an emitter wired to the given registry, mirroring
    /// the production construction in `sub_agent_dispatch::execute_sub_agent`.
    fn emitter_for(
        reg: &Arc<ChildAgentRegistry>,
        task_id: u32,
        spawner: Option<u32>,
    ) -> ChildStatusEmitter {
        let (tx, _rx) = watch::channel(AgentStatus::Pending);
        ChildStatusEmitter::new(task_id, spawner, true, tx, reg.clone())
    }

    fn extract(event: &EngineEvent) -> (u32, Option<u32>, bool, &AgentStatus) {
        match event {
            EngineEvent::ChildTaskUpdate {
                task_id,
                spawner,
                is_background,
                status,
            } => (*task_id, *spawner, *is_background, status),
            other => panic!("expected ChildTaskUpdate, got {other:?}"),
        }
    }

    #[test]
    fn emitter_send_queues_engine_event_on_registry() {
        let reg = Arc::new(ChildAgentRegistry::new());
        let emitter = emitter_for(&reg, 7, Some(42));

        // Initial: queue is empty.
        assert!(
            reg.drain_status_events().is_empty(),
            "fresh registry must have an empty event queue"
        );

        emitter.send(AgentStatus::Running { iter: 0 });
        let drained = reg.drain_status_events();
        assert_eq!(drained.len(), 1, "single send must produce one event");
        let (id, spawner, _is_bg, status) = extract(&drained[0]);
        assert_eq!(id, 7);
        assert_eq!(spawner, Some(42));
        assert!(matches!(status, AgentStatus::Running { iter: 0 }));
    }

    #[test]
    fn emitter_drain_is_fifo_and_clears_queue() {
        let reg = Arc::new(ChildAgentRegistry::new());
        let emitter = emitter_for(&reg, 1, None);

        emitter.send(AgentStatus::Running { iter: 0 });
        emitter.send(AgentStatus::Running { iter: 1 });
        emitter.send(AgentStatus::Running { iter: 2 });
        emitter.send(AgentStatus::Completed {
            summary: "done".into(),
        });

        let drained = reg.drain_status_events();
        assert_eq!(drained.len(), 4, "all four sends must surface");

        // FIFO: transition order is preserved across batches.  This
        // matters for clients that render "iter N" progress — a
        // reorder would show the count moving backwards.
        let iters: Vec<_> = drained
            .iter()
            .filter_map(|e| match e {
                EngineEvent::ChildTaskUpdate {
                    status: AgentStatus::Running { iter },
                    ..
                } => Some(*iter),
                _ => None,
            })
            .collect();
        assert_eq!(iters, vec![0, 1, 2]);

        // Last event is the terminal Completed.
        assert!(matches!(
            extract(&drained[3]).3,
            AgentStatus::Completed { .. }
        ));

        // Drain consumes — second drain is empty.
        assert!(
            reg.drain_status_events().is_empty(),
            "drain must clear the queue"
        );
    }

    #[test]
    fn emitter_send_also_updates_watch_channel() {
        // The watch fan-out is what `/agents` and `snapshot()` read.
        // Sink fan-out (queue) is for the inference-loop → EngineSink
        // path.  Both targets must see every transition or `/agents`
        // and the TUI/ACP/headless clients will disagree on state.
        let reg = Arc::new(ChildAgentRegistry::new());
        let (tx, mut rx) = watch::channel(AgentStatus::Pending);
        let emitter = ChildStatusEmitter::new(3, None, true, tx, reg.clone());

        emitter.send(AgentStatus::Running { iter: 5 });

        // Watch channel observed.
        assert!(matches!(
            *rx.borrow_and_update(),
            AgentStatus::Running { iter: 5 }
        ));
        // Queue observed.
        let drained = reg.drain_status_events();
        assert_eq!(drained.len(), 1);
        assert!(matches!(
            extract(&drained[0]).3,
            AgentStatus::Running { iter: 5 }
        ));
    }

    #[test]
    fn emitter_clones_share_queue_and_watch() {
        // Layer 4 holds one clone for live `iter` heartbeats;
        // `run_bg_agent` keeps another for entry / terminal sends.
        // Both clones must funnel into the same registry queue and
        // the same per-task watch channel — otherwise terminal
        // states could land on a different queue than the heartbeats
        // and clients would see Running forever.
        let reg = Arc::new(ChildAgentRegistry::new());
        let (tx, _rx) = watch::channel(AgentStatus::Pending);
        let a = ChildStatusEmitter::new(11, Some(2), true, tx, reg.clone());
        let b = a.clone();

        a.send(AgentStatus::Running { iter: 1 });
        b.send(AgentStatus::Completed {
            summary: "ok".into(),
        });

        let drained = reg.drain_status_events();
        assert_eq!(drained.len(), 2, "clones must share the registry queue");
        // Watch channel reflects the LATEST send, regardless of which
        // clone made it (watch is overwriting by design).
        assert!(matches!(a.current(), AgentStatus::Completed { .. }));
    }

    #[test]
    fn agent_status_round_trips_through_serde() {
        // `EngineEvent::ChildTaskUpdate` is the wire format for ACP /
        // headless / future transports.  All `AgentStatus` variants
        // must survive a serde round-trip or the boundary leak fix
        // creates a new boundary leak (engine emits, transport drops
        // it on the floor).
        for status in [
            AgentStatus::Pending,
            AgentStatus::Running { iter: 0 },
            AgentStatus::Running { iter: 17 },
            AgentStatus::Cancelled,
            AgentStatus::Completed {
                summary: "hello".into(),
            },
            AgentStatus::Errored {
                error: "boom".into(),
            },
        ] {
            let event = EngineEvent::ChildTaskUpdate {
                task_id: 1,
                spawner: Some(2),
                is_background: true,
                status: status.clone(),
            };
            let json = serde_json::to_string(&event).expect("serialize");
            let back: EngineEvent = serde_json::from_str(&json).expect("deserialize");
            match back {
                EngineEvent::ChildTaskUpdate {
                    task_id,
                    spawner,
                    is_background,
                    status: round_tripped,
                } => {
                    assert_eq!(task_id, 1);
                    assert_eq!(spawner, Some(2));
                    assert!(is_background, "is_background must round-trip true");
                    assert_eq!(round_tripped, status, "json round-trip lost data: {json}");
                }
                other => panic!("round-trip changed variant: {other:?}"),
            }
        }
    }

    #[test]
    fn drain_status_events_is_empty_on_fresh_registry() {
        // The inference loop calls `drain_status_events` every
        // iteration; the no-bg-task case must be cheap and yield
        // an empty Vec without any allocations forced by mistakes
        // in the queue type (e.g. `Some(VecDeque::new())`).
        let reg = ChildAgentRegistry::new();
        let drained = reg.drain_status_events();
        assert!(drained.is_empty());
    }

    /// #1201 B: `ChildStatusEmitter::send_activity` must push events
    /// onto the *same* registry queue that `ChildStatusEmitter::send`
    /// uses, with the emitter's `task_id` and `spawner` baked in.
    /// This pins the wire-up so the inference loop's existing drain
    /// in `inference.rs` picks up activity events without any extra
    /// plumbing.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn send_activity_queues_bg_child_activity_with_task_id() {
        let registry = new_shared();
        let (status_tx, _status_rx) = tokio::sync::watch::channel(AgentStatus::Pending);
        let emitter = ChildStatusEmitter::new(42, Some(7), true, status_tx, registry.clone());

        emitter.send_activity(crate::engine::event::ChildAgentActivityKind::ToolStart {
            tool_name: "Read".into(),
            summary: "Read foo.txt".into(),
        });
        emitter.send_activity(crate::engine::event::ChildAgentActivityKind::Info {
            message: "\u{26a1} cache hit".into(),
        });

        let drained = registry.drain_status_events();
        assert_eq!(drained.len(), 2);
        assert!(matches!(
            &drained[0],
            EngineEvent::ChildAgentActivity {
                task_id: 42,
                spawner: Some(7),
                is_background: true,
                kind: crate::engine::event::ChildAgentActivityKind::ToolStart { tool_name, .. }
            } if tool_name == "Read"
        ));
        assert!(matches!(
            &drained[1],
            EngineEvent::ChildAgentActivity {
                task_id: 42,
                spawner: Some(7),
                is_background: true,
                kind: crate::engine::event::ChildAgentActivityKind::Info { message }
            } if message.contains("cache hit")
        ));
    }

    // ── PR-A of #1232 §1: foreground sub-agent registry path ────────────

    /// `register_fg_with_emitter` inserts an entry whose snapshot
    /// reports `is_background == false`. Sanity-checks that the fg
    /// path is wired to the same registry the `/agents` overlay reads.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn register_fg_inserts_entry_with_is_background_false() {
        let reg = Arc::new(ChildAgentRegistry::new());
        let parent = CancellationToken::new();

        let (_emitter, _guard) = reg.register_fg_with_emitter(
            "explore",
            "audit the storage layer",
            &parent,
            Some(7),
            Some("call_42".into()),
        );

        let snap = reg.snapshot();
        assert_eq!(snap.len(), 1, "fg registration must add exactly one entry");
        let entry = &snap[0];
        assert_eq!(entry.agent_name, "explore");
        assert_eq!(entry.prompt, "audit the storage layer");
        assert_eq!(entry.spawner, Some(7));
        assert!(
            !entry.is_background,
            "fg entries must be tagged is_background=false so the overlay \
             and any future is_background-aware filter agree"
        );
    }

    /// Dropping the `FgRegistrationGuard` removes the entry.
    /// RAII covers `?`-bubbled errors and panics; without it the
    /// overlay would render a phantom "running" row indefinitely.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn fg_guard_drop_removes_entry_from_registry() {
        let reg = Arc::new(ChildAgentRegistry::new());
        let parent = CancellationToken::new();

        {
            let (_emitter, _guard) =
                reg.register_fg_with_emitter("explore", "do thing", &parent, None, None);
            assert_eq!(reg.snapshot().len(), 1, "entry present while guard alive");
        }

        assert_eq!(
            reg.snapshot().len(),
            0,
            "guard's Drop impl must remove the registry entry on scope exit"
        );
    }

    /// `drain_completed` must skip foreground entries unconditionally.
    /// Defense-in-depth: today the fg path doesn't fire its (dropped)
    /// oneshot sender, so try_recv would yield `Closed` and the
    /// `Closed` arm would manufacture a fake "cancelled"
    /// `BgAgentResult` — which the inference loop would then inject
    /// as a duplicate of the result the parent already returned
    /// inline. The `is_background` skip prevents that whole bug class
    /// by construction.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn drain_completed_skips_foreground_entries() {
        let reg = Arc::new(ChildAgentRegistry::new());
        let parent = CancellationToken::new();

        let (_emitter, _guard) =
            reg.register_fg_with_emitter("explore", "prompt", &parent, None, None);

        let drained = reg.drain_completed();
        assert!(
            drained.is_empty(),
            "fg entries must never be drained as bg results; got {drained:?}"
        );
        assert_eq!(
            reg.snapshot().len(),
            1,
            "drain must not pop fg entries either"
        );
    }

    /// A bg entry whose oneshot sender has been dropped DOES get
    /// drained (yielding the synthetic "cancelled" result). This is
    /// the existing behavior the fg-skip is defending against.
    /// Pinning it so a future refactor that changes the bg semantics
    /// can't silently regress the fg case.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn drain_completed_still_synthesizes_cancelled_for_bg_with_closed_rx() {
        let reg = Arc::new(ChildAgentRegistry::new());
        let (_id, tx) = reg.register_test("explore", "prompt");
        // Drop the sender to simulate a panicked/cancelled bg task.
        drop(tx);

        let drained = reg.drain_completed();
        assert_eq!(drained.len(), 1, "bg with closed rx still drains");
        assert!(!drained[0].success, "closed-rx drain marks failure");
        assert!(
            drained[0].output.contains("cancelled"),
            "closed-rx synthetic result mentions cancellation"
        );
    }

    /// Fg emitter's `send_activity` must tag events with
    /// `is_background: false`. Pre-PR-A the emitter hardcoded `true`,
    /// which would have made every fg activity event mis-tagged on
    /// the wire — ACP / headless filters keying on that flag would
    /// silently drop fg events.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn fg_emitter_tags_activity_events_with_is_background_false() {
        let reg = Arc::new(ChildAgentRegistry::new());
        let parent = CancellationToken::new();

        let (emitter, _guard) =
            reg.register_fg_with_emitter("explore", "prompt", &parent, None, None);
        emitter.send_activity(crate::engine::event::ChildAgentActivityKind::ToolStart {
            tool_name: "Read".into(),
            summary: "Read foo.rs".into(),
        });

        let drained = reg.drain_status_events();
        let activity = drained
            .iter()
            .find(|e| matches!(e, EngineEvent::ChildAgentActivity { .. }))
            .expect("activity event present");
        let EngineEvent::ChildAgentActivity { is_background, .. } = activity else {
            unreachable!("matched above");
        };
        assert!(
            !*is_background,
            "fg emitter must produce ChildAgentActivity {{ is_background: false }}; \
             pre-PR-A this was hardcoded true and would silently mis-tag the wire"
        );
    }

    // ─── #1321: mpsc-channel receiver semantics ──────────────────

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn take_event_receiver_returns_some_then_none() {
        // Contract: the receiver moves out of the registry exactly
        // once. Production calls this from `KodaSession::attach_event_sink`
        // at TUI startup; a second call must be a safe no-op (returns
        // `None`) so accidental double-attach doesn't deadlock or
        // panic. Mirrors Codex's `forward_events` lifecycle: spawn
        // once, abort once.
        let reg = ChildAgentRegistry::new();
        let first = reg.take_event_receiver();
        assert!(first.is_some(), "first call must yield the receiver");
        let second = reg.take_event_receiver();
        assert!(
            second.is_none(),
            "second call must return None — receiver is single-consumer"
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn drain_status_events_returns_empty_after_take() {
        // After the session forwarder takes the receiver, the legacy
        // synchronous drain path must be a benign no-op rather than
        // panicking or returning stale events. This guarantees the
        // headless-shaped helpers and the (now-removed) per-iteration
        // drain in `inference_loop` keep compiling for any external
        // caller that hasn't migrated yet.
        let reg = ChildAgentRegistry::new();
        // Push something *first*, then take the receiver: that event
        // belongs to the new owner, not to a late `drain_status_events`
        // caller.
        reg.push_status_event(EngineEvent::ChildTaskUpdate {
            task_id: 1,
            spawner: None,
            is_background: true,
            status: AgentStatus::Pending,
        });
        let mut rx = reg.take_event_receiver().expect("take must succeed");
        assert!(
            reg.drain_status_events().is_empty(),
            "drain after take must yield empty — events route through the receiver"
        );
        // The pre-take event reached the new receiver, not the void.
        let ev = rx
            .try_recv()
            .expect("pre-take event must reach the new owner");
        assert!(matches!(
            ev,
            EngineEvent::ChildTaskUpdate { task_id: 1, .. }
        ));
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn forwarder_emits_during_blocking_consumer() {
        // **The actual bug fix from #1321.** Pre-#1321 a long-running
        // tool call (e.g. `WaitTask` blocking up to 300 s) starved the
        // bg-status drain because the drain only ran between
        // inference-loop iterations. With the mpsc receiver owned by a
        // dedicated forwarder task, status events reach the sink the
        // moment `push_status_event` fires — even while the would-be
        // drainer is parked inside an arbitrarily-long `await`.
        //
        // Simulates the failure shape: a fake "foreground" future that
        // sleeps for a long time, in parallel with bg-agent status
        // emissions. Asserts that each emission reaches the sink
        // *before* the foreground future returns.
        use std::sync::Arc;
        use std::sync::atomic::{AtomicUsize, Ordering};
        use std::time::Duration;

        struct CountingSink(Arc<AtomicUsize>);
        impl crate::engine::EngineSink for CountingSink {
            fn emit(&self, _ev: EngineEvent) {
                self.0.fetch_add(1, Ordering::SeqCst);
            }
        }

        let reg = Arc::new(ChildAgentRegistry::new());
        let count = Arc::new(AtomicUsize::new(0));
        let sink: Arc<dyn crate::engine::EngineSink> = Arc::new(CountingSink(count.clone()));

        // Stand up the forwarder by hand (mirrors `attach_event_sink`).
        let mut rx = reg.take_event_receiver().unwrap();
        let forwarder = tokio::spawn(async move {
            while let Some(ev) = rx.recv().await {
                sink.emit(ev);
            }
        });

        // Foreground "tool call" that holds the loop hostage.
        let foreground = tokio::spawn(async {
            tokio::time::sleep(Duration::from_millis(300)).await;
        });

        // Producer: emits status events at intervals while the
        // foreground task is still in its `await`. Pre-#1321 these
        // would queue silently — invisible until the foreground
        // returned. Post-#1321 they reach the sink immediately.
        let reg_p = reg.clone();
        let producer = tokio::spawn(async move {
            for iter in 1..=5u32 {
                tokio::time::sleep(Duration::from_millis(40)).await;
                reg_p.push_status_event(EngineEvent::ChildTaskUpdate {
                    task_id: 7,
                    spawner: None,
                    is_background: true,
                    status: AgentStatus::Running { iter },
                });
            }
        });

        producer.await.unwrap();
        // Wait for the forwarder to drain in-flight events. Cap at a
        // value well below the foreground sleep (300 ms) so a
        // regression — events queueing instead of streaming — fails
        // here, not by accident from the foreground completing first.
        let deadline = std::time::Instant::now() + Duration::from_millis(150);
        while count.load(Ordering::SeqCst) < 5 && std::time::Instant::now() < deadline {
            tokio::task::yield_now().await;
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
        let observed_before_foreground = count.load(Ordering::SeqCst);
        foreground.await.unwrap();

        assert_eq!(
            observed_before_foreground, 5,
            "all 5 bg-status events must reach the sink BEFORE the foreground \
             tool call returns; pre-#1321 this was 0 (queued, not flowing)"
        );

        // Drop registry so the channel closes and the forwarder exits.
        drop(reg);
        let _ = forwarder.await;
    }
}
