aion-integration-acp 0.31.0

//! [`AcpSession`] — a live ACP prompt turn behind the [`AgentSession`] seam.
//!
//! The session owns three things and delegates the rest:
//!
//! - the **negotiated capability set** ([`AgentSession::capabilities`]),
//! - the **event stream** out ([`AgentSession::events`]) and the **command channel** in
//!   ([`AgentSession::intervene`]),
//! - the **terminal result** ([`AgentSession::wait_result`]), and the orderly shutdown of the agent
//!   process that follows it.
//!
//! The inbound half of the conversation lives in [`crate::pump`] — one background task that owns the
//! connection's reader, translates `session/update` into transcript events, answers every peer
//! request, and captures the `session/prompt` Response as the result. The interpretation of that
//! Response lives in [`crate::outcome`].
//!
//! # The events stream ends at the answer, not at the process
//!
//! The platform's agent driver treats the event stream ending as end-of-run and only then takes the
//! result. A real ACP agent is a persistent stdio server that keeps running after it answers
//! `session/prompt`, so the pump drops its transcript sender as soon as the terminal Response has
//! been handled rather than waiting for the child's stdout to close. See [`crate::pump`].
//!
//! # Cancellation
//!
//! [`AgentSession::intervene`] with [`aion_core::InterventionKind::Cancel`] sends the
//! `session/cancel` notification. The spec also requires that a client which cancels answers every
//! pending `session/request_permission` with the `cancelled` outcome, so the session raises a
//! cancelled flag BEFORE writing the notification: from that moment the pump answers every
//! permission request `cancelled` rather than consulting the policy. A request the pump had already
//! answered from the policy before the flag was raised stays answered as it was — it happened before
//! the cancellation, and rewriting history would be the dishonest option.
//!
//! [`AcpSession`]: crate::session::AcpSession
//! [`AgentSession`]: aion_integrations::AgentSession
//! [`AgentSession::capabilities`]: aion_integrations::AgentSession::capabilities
//! [`AgentSession::events`]: aion_integrations::AgentSession::events
//! [`AgentSession::intervene`]: aion_integrations::AgentSession::intervene
//! [`AgentSession::wait_result`]: aion_integrations::AgentSession::wait_result

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use agent_client_protocol_schema::v1::{
    ContentBlock, PromptRequest, SessionConfigOptionValue, SessionId,
    SetSessionConfigOptionRequest, TextContent,
};
use aion_core::{
    ActivityEvent, InterventionCapabilities, InterventionCommand, InterventionKind, Payload,
};
use aion_integrations::HarnessError;
use aion_integrations::contract::AgentSession;
use aion_integrations::jsonrpc::{
    JsonRpcConnection, JsonRpcId, JsonRpcNotification, JsonRpcRequest,
};
use async_trait::async_trait;
use futures::stream::BoxStream;
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::process::Child;
use tokio::sync::{mpsc, oneshot};

use crate::containment::ProcessGroup;
use crate::messages::MessageAccumulator;
use crate::outcome::{PromptOutcome, PumpEnd};
use crate::permission::PermissionPolicy;
use crate::protocol;
use crate::pump::{
    OpenTurn, PendingRequests, PendingTurn, PumpContext, pump_loop, register_pending_request,
    retract_pending_request,
};
use crate::translate::{self, EventIdentity};

/// Everything one live session needs at construction.
///
/// Grouped into a struct rather than eight positional arguments so a future field cannot be wired
/// into the wrong parameter slot — the same reason [`PumpContext`] is a struct.
pub(crate) struct SessionStart<R, W> {
    /// The connection whose handshake is done and whose `session/prompt` is already in flight.
    pub(crate) connection: Arc<JsonRpcConnection<R, W>>,
    /// The session `session/new` minted.
    pub(crate) session_id: SessionId,
    /// The id of the outstanding `session/prompt`: its Response is the terminal result.
    pub(crate) prompt_id: JsonRpcId,
    /// How `session/request_permission` is answered.
    pub(crate) policy: PermissionPolicy,
    /// The `(workflow, activity, attempt)` identity stamped on every event.
    pub(crate) identity: EventIdentity,
    /// Notifications that arrived during the pre-prompt handshake, replayed onto the transcript
    /// first so nothing read before the pump existed is lost.
    pub(crate) deferred: Vec<JsonRpcNotification>,
    /// The index range of `deferred` that arrived while `session/load` was outstanding — the prior
    /// conversation being replayed, which the pump records as replay rather than as the open
    /// turn's own frames. `0..0` when the session was minted fresh.
    pub(crate) load_replay: std::ops::Range<usize>,
    /// The spawned agent. `None` for the in-memory test path, which drives a duplex.
    pub(crate) child: Option<Child>,
    /// How long the agent may take to exit on its own after its stdin closes. See
    /// [`crate::AcpHarness::new`].
    pub(crate) exit_grace: Duration,
    /// Whether this agent advertised `loadSession` at `initialize`.
    pub(crate) load_session: bool,
    /// The raw `configOptions` advertisement the opening response carried, in
    /// the update wire shape, when the agent sent one.
    pub(crate) config_options: Option<serde_json::Value>,
    /// The process group the agent leads, when the harness asked for
    /// containment. `None` means only the direct child can be signalled.
    pub(crate) process_group: Option<ProcessGroup>,
}

/// A live ACP prompt turn for one activity attempt.
///
/// Generic over the child's read/write halves so tests can drive a real [`JsonRpcConnection`] over an
/// in-memory duplex; in production the halves are the agent subprocess's stdout/stdin.
pub struct AcpSession<R, W> {
    /// The connection, held in an [`Option`] because closing the agent's stdin means dropping the
    /// LAST reference to it — that is the only in-band way to tell an ACP agent the conversation is
    /// over, and it cannot be done while the field still owns one.
    connection: Option<Arc<JsonRpcConnection<R, W>>>,
    capabilities: InterventionCapabilities,
    session_id: SessionId,
    events: Option<mpsc::UnboundedReceiver<ActivityEvent>>,
    outcome: Option<oneshot::Receiver<PumpEnd>>,
    /// Raised before `session/cancel` is written; makes the pump answer later permission requests
    /// with the `cancelled` outcome, as the spec requires of a client that cancels.
    cancelled: Arc<AtomicBool>,
    pump: Option<tokio::task::JoinHandle<()>>,
    /// The spawned agent, shut down when the result is taken or the session is dropped. `None` for
    /// the in-memory test path.
    child: Option<Child>,
    /// The grace the agent gets to exit on its own once its stdin has closed.
    exit_grace: Duration,
    /// The slot [`Self::prompt`] hands a new turn to the pump through.
    pending: PendingTurn,
    /// Whether a turn is open. `true` from construction: the harness opens the session with a
    /// prompt already in flight.
    busy: Arc<AtomicBool>,
    /// Raised by the pump when it stops, so a later [`Self::prompt`] refuses at once rather than
    /// queueing a turn that nothing will ever answer.
    ended: Arc<AtomicBool>,
    /// Whether this agent advertised `loadSession` at `initialize`.
    load_session: bool,
    /// The raw `configOptions` advertisement the opening response carried, if
    /// any — held for the session's owner to read once and record.
    initial_config_options: Option<serde_json::Value>,
    /// The answer channels for outstanding non-prompt requests, shared with
    /// the pump.
    requests: PendingRequests,
    /// The process group the agent leads, when the harness asked for
    /// containment.
    process_group: Option<ProcessGroup>,
}

/// The value one configuration option is set to.
///
/// The neutral half of ACP's `SessionConfigOptionValue`: a select choice's id,
/// or a toggle's flag. The caller knows which shape an option takes from the
/// agent's own advertisement — this enum states the caller's intent without
/// re-deriving it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum AcpConfigValue {
    /// The id of the select choice to pick.
    Choice(String),
    /// The value a toggle is set to.
    Toggle(bool),
}

impl<R, W> AcpSession<R, W>
where
    R: AsyncRead + Unpin + Send + 'static,
    W: AsyncWrite + Unpin + Send + 'static,
{
    /// The raw `configOptions` advertisement the opening response carried, if
    /// the agent sent one — in the same wire shape `config_option_update`
    /// uses, so the owner's one reader serves both. `None` is "the agent
    /// advertised nothing at open", never an empty list.
    #[must_use]
    pub const fn initial_config_options(&self) -> Option<&serde_json::Value> {
        self.initial_config_options.as_ref()
    }

    /// Set one advertised configuration option — the model picker's write.
    ///
    /// A REQUEST with a response, unlike every intervention this adapter
    /// delivers as a notification: the agent answers with the full option set
    /// as it now stands, and that answer is returned verbatim for the caller
    /// to record. The pending entry is registered BEFORE the request is
    /// written — the same slot-before-write rule `prompt` follows — so the
    /// pump can never read an answer nobody is waiting on; a failed write
    /// takes the entry back.
    ///
    /// # Errors
    ///
    /// - [`HarnessError::StaleTarget`] when the session is shut down or its
    ///   reader has stopped — immediately, or while awaiting the answer.
    /// - [`HarnessError::Harness`] when the agent refuses the change.
    /// - [`HarnessError::Protocol`] when the request cannot be encoded, or the
    ///   answer carries neither result nor error.
    /// - [`HarnessError::Transport`] when the request cannot be written.
    pub async fn set_config_option(
        &self,
        config_id: &str,
        value: AcpConfigValue,
    ) -> Result<serde_json::Value, HarnessError> {
        if self.ended.load(Ordering::SeqCst) {
            return Err(HarnessError::stale_target(
                "this ACP session's reader has stopped — the agent exited or its transport failed \
                 — so there is no live conversation left to configure; start a new session",
            ));
        }
        let connection = self.connection.as_ref().ok_or_else(|| {
            HarnessError::stale_target(
                "this ACP session has already been shut down; there is no live conversation left \
                 to configure",
            )
        })?;
        let wire_value = match value {
            AcpConfigValue::Choice(id) => SessionConfigOptionValue::value_id(id),
            AcpConfigValue::Toggle(flag) => SessionConfigOptionValue::boolean(flag),
        };
        let request = SetSessionConfigOptionRequest::new(
            self.session_id.clone(),
            config_id.to_owned(),
            wire_value,
        );
        let params = serde_json::to_value(&request).map_err(|source| {
            HarnessError::protocol(format!(
                "session/set_config_option params are not encodable: {source}"
            ))
        })?;
        let request_id = connection.next_request_id();
        let (answer_tx, answer_rx) = tokio::sync::oneshot::channel();
        register_pending_request(&self.requests, request_id.clone(), answer_tx);
        let frame = JsonRpcRequest::new(
            request_id.clone(),
            protocol::METHOD_SESSION_SET_CONFIG_OPTION,
            Some(params),
        );
        if let Err(error) = connection.send_request(&frame).await {
            retract_pending_request(&self.requests, &request_id);
            return Err(error);
        }
        let response = answer_rx.await.map_err(|_| {
            HarnessError::stale_target(
                "the agent's reader stopped before it answered this configuration change; the \
                 option's state is whatever the agent last advertised",
            )
        })?;
        if let Some(error) = response.error {
            return Err(HarnessError::harness(format!(
                "the agent refused `session/set_config_option`: {} (code {})",
                error.message, error.code
            )));
        }
        response.result.ok_or_else(|| {
            HarnessError::protocol(
                "the agent answered `session/set_config_option` with neither result nor error"
                    .to_owned(),
            )
        })
    }

    /// Builds a session over a connection whose handshake is already done and whose
    /// `session/prompt` request is already in flight, and starts the reader pump.
    ///
    /// The pump starts here so events stream from the moment the session exists.
    pub(crate) fn start(start: SessionStart<R, W>) -> Self {
        let SessionStart {
            connection,
            session_id,
            prompt_id,
            policy,
            identity,
            deferred,
            load_replay,
            child,
            exit_grace,
            load_session,
            config_options,
            process_group,
        } = start;
        let (event_tx, event_rx) = mpsc::unbounded_channel();
        let (outcome_tx, outcome_rx) = oneshot::channel();
        let cancelled = Arc::new(AtomicBool::new(false));
        let pending: PendingTurn = Arc::new(tokio::sync::Mutex::new(None));
        let requests: PendingRequests = Arc::new(std::sync::Mutex::new(Vec::new()));
        // The session is born BUSY: the harness has already written the opening `session/prompt`,
        // so the first turn is open before this constructor runs.
        let busy = Arc::new(AtomicBool::new(true));
        let ended = Arc::new(AtomicBool::new(false));

        let pump = tokio::spawn(pump_loop(
            PumpContext {
                connection: Arc::clone(&connection),
                session_id: session_id.clone(),
                policy,
                cancelled: Arc::clone(&cancelled),
                identity,
                pending: Arc::clone(&pending),
                requests: Arc::clone(&requests),
                busy: Arc::clone(&busy),
                ended: Arc::clone(&ended),
            },
            deferred,
            load_replay,
            OpenTurn {
                prompt_id,
                event_tx,
                outcome_tx,
                accumulator: MessageAccumulator::new(),
            },
        ));

        Self {
            connection: Some(connection),
            capabilities: translate::negotiated_capabilities(),
            session_id,
            events: Some(event_rx),
            outcome: Some(outcome_rx),
            cancelled,
            pump: Some(pump),
            child,
            exit_grace,
            pending,
            busy,
            ended,
            load_session,
            initial_config_options: config_options,
            requests,
            process_group,
        }
    }

    /// The process group this agent leads, when the harness asked for
    /// containment.
    ///
    /// Published so a composition root — and the pin that proves the tree is
    /// gone — can probe the group rather than take the adapter's word for it.
    #[must_use]
    pub const fn process_group(&self) -> Option<ProcessGroup> {
        self.process_group
    }

    /// Whether this agent advertised the `loadSession` capability at
    /// `initialize`.
    ///
    /// The ONLY honest basis for deciding whether a conversation can be reopened
    /// after its process is gone: it is what the agent said about itself, on
    /// this launch, rather than an assumption about what an agent of its kind
    /// supports. A caller that respawned and issued `session/load` against an
    /// agent that never advertised it would be refused by the adapter — after
    /// spawning a process for nothing.
    #[must_use]
    pub const fn supports_load_session(&self) -> bool {
        self.load_session
    }

    /// The ACP session this conversation runs in — the handle `session/load` would resume.
    #[must_use]
    pub fn session_id(&self) -> &SessionId {
        &self.session_id
    }

    /// Whether the reader pump is still running: `false` once the agent's stdout has ended or the
    /// transport has failed.
    ///
    /// The honest answer to "can this session still take a turn". A caller that holds a session
    /// across turns reads this to decide whether to retire it, instead of discovering the death
    /// by submitting a turn.
    #[must_use]
    pub fn is_live(&self) -> bool {
        !self.ended.load(Ordering::SeqCst)
    }

    /// Takes the FIRST turn — the one the harness opened with — as a [`TurnHandle`].
    ///
    /// A caller that drives a session across several turns needs the opening turn in the same
    /// shape as every later one; a caller that runs a single activity keeps using
    /// [`AgentSession::wait_result`], which consumes the session and shuts the agent down after.
    /// The two are exclusive by construction: both take the same receivers, so whichever runs
    /// first wins and the other reports that the result was already taken.
    ///
    /// Returns `None` when the first turn has already been taken (by either route).
    pub fn take_first_turn(&mut self) -> Option<TurnHandle> {
        let outcome = self.outcome.take()?;
        Some(TurnHandle {
            session_id: self.session_id.clone(),
            events: self.events.take(),
            outcome,
        })
    }

    /// Submits a SECOND (or later) turn on this live session.
    ///
    /// ACP models a conversation as one session carrying many `session/prompt` turns; the harness
    /// opens the first, and this is how the rest are asked. The agent keeps everything it learned
    /// in the earlier turns — that is the whole point of a session — so this is continuity without
    /// `session/load` and without a second process.
    ///
    /// **One turn at a time.** A prompt submitted while another is open is refused with
    /// [`HarnessError::Occupied`] rather than queued: the agent has one conversation, updates on
    /// the wire carry no turn id, and a second in-flight prompt would make every streamed chunk
    /// ambiguous about which turn it belongs to.
    ///
    /// The turn is handed to the pump BEFORE the request is written, so the pump can never read an
    /// answer for a turn it has not adopted. If the write fails, the turn is taken back and the
    /// session is free again.
    ///
    /// # Errors
    ///
    /// - [`HarnessError::StaleTarget`] when the session has already been shut down, or its reader
    ///   pump has stopped (the agent exited, or the transport failed) — an immediate refusal, never
    ///   a wait on an answer that cannot come.
    /// - [`HarnessError::Occupied`] when a turn is already open.
    /// - [`HarnessError::Protocol`] when the prompt cannot be encoded.
    /// - [`HarnessError::Transport`] when the request cannot be written.
    pub async fn prompt(&self, text: String) -> Result<TurnHandle, HarnessError> {
        if self.ended.load(Ordering::SeqCst) {
            return Err(HarnessError::stale_target(
                "this ACP session's reader has stopped — the agent exited or its transport failed \
                 — so there is no live conversation left to prompt; start a new session",
            ));
        }
        let connection = self.connection.as_ref().ok_or_else(|| {
            HarnessError::stale_target(
                "this ACP session has already been shut down; there is no live conversation left \
                 to prompt",
            )
        })?;
        // `text` is one of the two content-block types every ACP agent MUST support, so a text
        // block needs no capability check — the same reasoning the opening prompt is built on.
        let request = PromptRequest::new(
            self.session_id.clone(),
            vec![ContentBlock::Text(TextContent::new(text))],
        );
        // Encoded BEFORE the session is marked busy: an unencodable prompt must not leave the
        // session looking occupied by a turn that was never opened.
        let params = serde_json::to_value(&request).map_err(|source| {
            HarnessError::protocol(format!("session/prompt params are not encodable: {source}"))
        })?;
        if self
            .busy
            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
            .is_err()
        {
            return Err(HarnessError::occupied(
                "this ACP session already has a prompt turn open — an agent holds one conversation \
                 and ACP's streamed updates carry no turn id, so a second concurrent prompt could \
                 not be attributed; cancel the open turn or wait for its result",
            ));
        }
        let prompt_id = connection.next_request_id();
        let (event_tx, event_rx) = mpsc::unbounded_channel();
        let (outcome_tx, outcome_rx) = oneshot::channel();
        // Filled BEFORE the write: the pump adopts from this slot before it dispatches any frame,
        // so the answer can never arrive for a turn the pump does not yet know about.
        *self.pending.lock().await = Some(OpenTurn {
            prompt_id: prompt_id.clone(),
            event_tx,
            outcome_tx,
            accumulator: MessageAccumulator::new(),
        });
        let frame = JsonRpcRequest::new(prompt_id, protocol::METHOD_SESSION_PROMPT, Some(params));
        if let Err(error) = connection.send_request(&frame).await {
            // Retract the turn if the pump has not already adopted it. If it HAS, the session
            // stays busy on purpose: the pump owns that turn and will report its end when the
            // (now broken) connection closes, and clearing the flag here would let a second turn
            // queue behind one nothing will answer.
            if self.pending.lock().await.take().is_some() {
                self.busy.store(false, Ordering::SeqCst);
            }
            return Err(error);
        }
        Ok(TurnHandle {
            session_id: self.session_id.clone(),
            events: Some(event_rx),
            outcome: outcome_rx,
        })
    }

    /// Shuts the agent down without taking a result: the orderly close for a caller that drove its
    /// turns through [`Self::prompt`] and has no outstanding turn to await.
    ///
    /// Identical to the shutdown [`AgentSession::wait_result`] performs — the pump is stopped, the
    /// agent's stdin is closed, and it is given [`AcpHarness::new`]'s configured grace to exit
    /// before it is killed and reaped.
    ///
    /// [`AcpHarness::new`]: crate::AcpHarness::new
    pub async fn close(mut self) {
        self.shut_down_agent().await;
    }

    /// Awaits the pump's single terminal report and interprets it.
    ///
    /// Split from [`AgentSession::wait_result`] so the agent process is shut down on **every** path,
    /// including the failing ones.
    async fn terminal_result(&mut self) -> Result<Payload, HarnessError> {
        let receiver = self
            .outcome
            .take()
            .ok_or_else(|| HarnessError::protocol("the prompt result was already awaited"))?;
        // A dropped sender means the pump task itself ended without reporting — it panicked, or it
        // was aborted. The message says exactly that and asserts nothing about the channel, which
        // may well still be open with the agent's answer unread.
        let end = receiver.await.map_err(|_recv| {
            HarnessError::transport(
                "the ACP reader pump ended without reporting a terminal outcome: the pump task did \
                 not complete (it panicked, or the session was shut down), so no session/prompt \
                 answer was ever interpreted",
            )
        })?;
        end.into_payload(&self.session_id)
    }

    /// Stops the pump, closes the agent's stdin, and gives it the configured grace to exit before
    /// killing it.
    ///
    /// A SUCCESSFUL turn ends here too, which is why the grace exists: an agent that is
    /// `SIGKILL`ed the instant it answers gets no chance to flush a log, close a file it wrote, or
    /// release a lock. The order is the only one that works:
    ///
    /// 1. **stop the pump** — it holds a second reference to the connection, and the in-flight
    ///    peer-request answers it owns die with it (the turn is over and the process is going),
    /// 2. **drop the last connection reference** — that drops the child's stdin, and stdin EOF is
    ///    the only in-band signal ACP has for "the conversation is over",
    /// 3. **wait, bounded by the configured grace** — a worker must not be hung by an agent that
    ///    ignores EOF,
    /// 4. **kill and reap** whatever is still alive.
    ///
    /// One honest caveat on step 2: the pump's answer-writer tasks each hold a reference too, and
    /// aborting them releases those references when the scheduler drops the tasks rather than
    /// synchronously here. In the ordinary case there are none in flight and the stdin close is
    /// immediate; where one is, the close lands a scheduler tick later, which the grace absorbs
    /// because the wait is on the child's exit and not on a deadline for the close.
    async fn shut_down_agent(&mut self) {
        if let Some(pump) = self.pump.take() {
            pump.abort();
            match pump.await {
                Ok(()) => {}
                // The expected outcome of the abort above.
                Err(error) if error.is_cancelled() => {}
                Err(error) => {
                    tracing::error!(%error, "acp adapter: the reader pump task panicked");
                }
            }
        }
        drop(self.connection.take());
        let Some(mut child) = self.child.take() else {
            return;
        };
        // Sent while the leader is UNREAPED, and to the whole GROUP: an ACP
        // agent is usually a launcher, so the process this handle holds is not
        // the process doing the work. Everything below reaps, so this is the
        // last moment the group id is provably ours (see `crate::containment`).
        if let Some(group) = self.process_group {
            group.terminate();
        }
        if self.exit_grace.is_zero() {
            if let Some(group) = self.process_group {
                group.kill();
            }
            kill_and_reap(&mut child, "no exit grace is configured for this harness").await;
            return;
        }
        match tokio::time::timeout(self.exit_grace, child.wait()).await {
            Ok(Ok(status)) => {
                tracing::debug!(
                    ?status,
                    "acp adapter: the agent exited on its own after its stdin closed"
                );
                // The leader has been reaped here, so its group id is no longer
                // ours to signal. What the tree got is the SIGTERM above, which
                // reached every member; a descendant that ignored it is
                // reported, never signalled on a recycled id.
                if let Some(group) = self.process_group
                    && !group.is_gone()
                {
                    tracing::warn!(
                        process_group = group.as_raw(),
                        "acp adapter: the agent exited but its process group is still observable; \
                         a descendant outlived it and cannot be signalled now that the leader has \
                         been reaped"
                    );
                }
            }
            Ok(Err(source)) => tracing::warn!(
                %source,
                "acp adapter: could not wait for the agent to exit; it may still be running"
            ),
            Err(_elapsed) => {
                // Nothing has been reaped on this arm, so the group is still
                // ours to aim at: the whole tree is killed and only then is the
                // leader reaped.
                if let Some(group) = self.process_group {
                    group.kill();
                }
                kill_and_reap(
                    &mut child,
                    "the agent did not exit within the configured grace after its stdin closed",
                )
                .await;
            }
        }
    }
}

#[async_trait]
impl<R, W> AgentSession for AcpSession<R, W>
where
    R: AsyncRead + Unpin + Send + 'static,
    W: AsyncWrite + Unpin + Send + 'static,
{
    fn capabilities(&self) -> &InterventionCapabilities {
        &self.capabilities
    }

    fn events(&mut self) -> BoxStream<'static, ActivityEvent> {
        // Owned by the returned stream, so taking it never borrows the session: the driver pumps
        // events while also calling `intervene` and `wait_result`. Taken once; a second call yields
        // an empty stream rather than panicking.
        match self.events.take() {
            Some(receiver) => Box::pin(futures::stream::unfold(
                receiver,
                |mut receiver| async move { receiver.recv().await.map(|event| (event, receiver)) },
            )),
            None => Box::pin(futures::stream::empty()),
        }
    }

    async fn intervene(&self, cmd: InterventionCommand) -> Result<(), HarnessError> {
        // Capability-gate FIRST, against the set this session ADVERTISED: an unadvertised primitive
        // is rejected before anything is written, so the four primitives ACP cannot deliver never
        // reach the wire even if a caller ignores the advertisement.
        if !self.capabilities.supports(&cmd.kind) {
            return Err(HarnessError::capability_not_supported(
                translate::primitive_label(cmd.kind.primitive()),
            ));
        }
        // `Cancel` carries an obligation beyond its frame: from the moment a client cancels it must
        // answer every pending `session/request_permission` with `cancelled`, so the flag is raised
        // BEFORE the notification is written. It is done here, at the decision, rather than inside a
        // send helper, so the ordering the spec requires is visible where the choice is made.
        if let InterventionKind::Cancel { reason } = &cmd.kind {
            self.cancelled.store(true, Ordering::SeqCst);
            tracing::info!(
                session = %self.session_id,
                reason,
                "acp adapter: cancelling the prompt turn"
            );
        }
        // The translation is the single authority on what this adapter can put on the wire, so a
        // primitive with no ACP mechanism is refused BY it, by name — there is no second list here
        // that could drift from the one the published capability set is pinned against.
        let notification = translate::intervention_to_notification(&self.session_id, &cmd.kind)?;
        self.send(notification).await
    }

    async fn wait_result(mut self) -> Result<Payload, HarnessError> {
        let result = self.terminal_result().await;
        // On EVERY path, including a failure: the agent gets its orderly shutdown, and the process
        // is reaped rather than left for `Drop` to signal blindly.
        self.shut_down_agent().await;
        result
    }
}

impl<R, W> AcpSession<R, W>
where
    R: AsyncRead + Unpin + Send,
    W: AsyncWrite + Unpin + Send,
{
    /// Writes one translated intervention frame to the agent.
    ///
    /// Intervening on a session whose result has already been taken is a stale target, not a
    /// transport fault: the turn is over and the connection was deliberately released.
    async fn send(&self, notification: translate::AcpNotification) -> Result<(), HarnessError> {
        let connection = self.connection.as_ref().ok_or_else(|| {
            HarnessError::stale_target(
                "the ACP session has already been shut down; there is no turn left to intervene in",
            )
        })?;
        let frame = JsonRpcNotification::new(notification.method, Some(notification.params));
        connection.send_notification(&frame).await
    }
}

/// One prompt turn on a live [`AcpSession`]: its transcript out, and its terminal result.
///
/// Returned by [`AcpSession::prompt`] and by [`AcpSession::take_first_turn`], so the opening turn
/// and every later one have the same shape. It borrows nothing from the session, which is what lets
/// a caller stream a turn's events while cancelling it through the session at the same time.
///
/// # The events stream ends at the turn's answer
///
/// Exactly as for the one-prompt path: the pump drops this turn's transcript sender the moment its
/// `session/prompt` Response has been handled, so draining [`Self::events`] to `None` is the
/// end-of-turn signal. The agent process is untouched by that — it stays alive for the next turn.
pub struct TurnHandle {
    session_id: SessionId,
    events: Option<mpsc::UnboundedReceiver<ActivityEvent>>,
    outcome: oneshot::Receiver<PumpEnd>,
}

impl TurnHandle {
    /// The ACP session this turn ran in.
    #[must_use]
    pub fn session_id(&self) -> &SessionId {
        &self.session_id
    }

    /// This turn's transcript, in order.
    ///
    /// Taken once; a second call yields an empty stream rather than panicking — the same contract
    /// [`aion_integrations::AgentSession::events`] carries.
    pub fn events(&mut self) -> BoxStream<'static, ActivityEvent> {
        match self.events.take() {
            Some(receiver) => Box::pin(futures::stream::unfold(
                receiver,
                |mut receiver| async move { receiver.recv().await.map(|event| (event, receiver)) },
            )),
            None => Box::pin(futures::stream::empty()),
        }
    }

    /// Awaits this turn's terminal outcome.
    ///
    /// # Errors
    ///
    /// [`HarnessError::Transport`] when the turn never got an answer — the agent closed its stdout
    /// or the transport failed — naming which; or when the pump task itself ended without
    /// reporting.
    pub async fn wait(self) -> Result<PromptOutcome, HarnessError> {
        let end = self.outcome.await.map_err(|_recv| {
            HarnessError::transport(
                "the ACP reader pump ended without reporting a terminal outcome for this turn: the \
                 pump task did not complete (it panicked, or the session was shut down), so no \
                 session/prompt answer was ever interpreted",
            )
        })?;
        end.into_outcome()
    }
}

/// Kills the agent and reaps it, saying why.
///
/// [`Child::kill`] is `start_kill` **plus** the wait that reaps the zombie, which is what makes this
/// a shutdown rather than a signal. A failure here is the only evidence that the kill did not land,
/// so it is logged rather than discarded.
async fn kill_and_reap(child: &mut Child, why: &str) {
    if let Err(source) = child.kill().await {
        tracing::warn!(
            %source,
            why,
            "acp adapter: could not kill and reap the agent process"
        );
        return;
    }
    tracing::debug!(why, "acp adapter: the agent process was killed and reaped");
}

impl<R, W> Drop for AcpSession<R, W> {
    fn drop(&mut self) {
        // The orderly shutdown lives in `shut_down_agent`, which needs to await and so cannot run
        // here. This is the abandoned path — a session dropped without its result being taken — and
        // its only job is to leave no task and no process running.
        if let Some(pump) = self.pump.take() {
            pump.abort();
        }
        if let Some(mut child) = self.child.take() {
            // The GROUP first, while nothing has been reaped: an abandoned
            // session must not leave the agent's tool processes running against
            // the operator's files.
            if let Some(group) = self.process_group {
                group.kill();
            }
            // `start_kill` INITIATES a kill; it does not reap. The reap is tokio's own: the command
            // was built with `kill_on_drop(true)`, so dropping `child` at the end of this scope
            // performs it. A failure here means the signal did not reach the process at all (a
            // child that has already exited reports success), which must not be silent.
            if let Err(source) = child.start_kill() {
                tracing::warn!(
                    %source,
                    "acp adapter: could not signal the agent process while dropping the session"
                );
            }
        }
    }
}

/// Answers one peer request received OUTSIDE the pump — during the pre-prompt handshake — with the
/// same `-32601` refusal the pump uses.
///
/// The handshake reads frames before a session exists, and a conforming agent sends no request
/// there. A non-conforming one must still get an answer: an unanswered request would block it before
/// the run had even started.
///
/// # Errors
///
/// Returns the transport error when the refusal cannot be written.
pub(crate) async fn refuse_peer_request<R, W>(
    connection: &JsonRpcConnection<R, W>,
    request: JsonRpcRequest,
) -> Result<(), HarnessError>
where
    R: AsyncRead + Unpin + Send,
    W: AsyncWrite + Unpin + Send,
{
    tracing::warn!(
        method = %request.method,
        "acp adapter: refusing a peer request received during the handshake"
    );
    let refusal = protocol::unadvertised_method_response(request.id, &request.method);
    connection.send_response(&refusal).await
}