aion-integration-acp 0.31.0

//! The reader pump: the one task that owns the connection's reader and handles every inbound frame.
//!
//! - a `session/update` **notification** is translated ([`crate::translate::update_to_kinds`]) and
//!   forwarded to the session's event stream — it has no `id`, so it can NEVER be captured as the
//!   result,
//! - the **Response** whose `id` matches the outstanding `session/prompt` is the terminal result, and
//!   nothing else can be,
//! - a `session/request_permission` **request** is recorded, answered from the
//!   [`PermissionPolicy`], and the decision recorded — a permission request left unanswered would
//!   block the agent until the activity timed out,
//! - any other peer **request** (`fs/*`, `terminal/*`, an unknown method) is answered with `-32601`
//!   and recorded. **Every** peer request gets an answer; silence wedges the agent.
//!
//! # The event stream ending is the end-of-run signal
//!
//! The platform's agent driver (`aion-worker`'s `spawn_agent`) pumps `events.next()` and takes the
//! terminal result only when the stream yields `None`. So the transcript sender is **dropped the
//! moment the terminal Response has been handled** — after the accumulator is flushed, the frame is
//! recorded, the `Stop` event is emitted and the outcome is sent. A real ACP agent is a persistent
//! stdio server: it answers `session/prompt` and keeps running, so waiting for the child to exit
//! before closing the stream would hang every successful turn forever.
//!
//! The loop itself keeps reading after that, so a peer request that was already on the wire when the
//! Response arrived still gets an answer. Be honest about how wide that window is: in the driver
//! path it is very narrow — the driver breaks out as soon as the stream ends, `wait_result` consumes
//! the session, and closing the session ends the child. It is a courtesy for a frame already in
//! flight, not a guaranteed window; a frame arriving after the stream closed is logged (it can no
//! longer reach the transcript) rather than silently discarded.
//!
//! # Where the loop stops, and where it does not
//!
//! It stops at **end-of-stream** or on a **transport** error, and reports which one to
//! `wait_result` through `PumpEnd` (see [`crate::outcome`]). A **protocol** error — a line that is not valid JSON-RPC — is
//! recoverable and does NOT stop it: the reader consumed exactly one line, so the framing is still
//! aligned, and a `node`-launched agent printing `Debugger listening on ws://…` between two frames
//! must not be able to destroy a turn whose answer has already been received.
//!
//! [`PermissionPolicy`]: crate::permission::PermissionPolicy

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

use agent_client_protocol_schema::v1::{
    PromptResponse, RequestPermissionOutcome, RequestPermissionRequest, SessionId,
    SessionNotification, SessionUpdate,
};
use aion_core::{ActivityEvent, ActivityEventKind, StopKind};
use aion_integrations::HarnessError;
use aion_integrations::jsonrpc::{
    IncomingMessage, JsonRpcConnection, JsonRpcError, JsonRpcId, JsonRpcNotification,
    JsonRpcRequest, JsonRpcResponse,
};
use serde_json::{Value, json};
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::sync::{mpsc, oneshot};
use tokio::task::JoinSet;

use crate::messages::MessageAccumulator;
use crate::outcome::{PromptOutcome, PumpEnd};
use crate::permission::PermissionPolicy;
use crate::protocol;
use crate::translate::{self, EventIdentity, EventKind};

mod output;

use output::{emit, emit_for, queue_response};

/// The `Raw` source label for a line the transport could not read as a JSON-RPC frame.
///
/// Not a `session/update` source: the frame never got as far as being classified, so labelling it
/// under the update namespace would claim more about it than is known.
const UNREADABLE_FRAME_SOURCE: &str = "acp/unreadable_frame";

/// One turn the pump is streaming: the outstanding `session/prompt` and the two channels its
/// result and its transcript leave by.
///
/// A turn owns its own accumulator because a message buffer is per-conversation-turn: chunks from
/// the second turn must never be joined onto the first turn's completed text.
pub(crate) struct OpenTurn {
    /// The id of the outstanding `session/prompt`: its Response is this turn's terminal result,
    /// and no other id can be.
    pub(crate) prompt_id: JsonRpcId,
    /// The transcript channel. Dropped when the turn ends: the driver's end-of-run signal is this
    /// stream ending, so holding it past the answer hangs the run.
    pub(crate) event_tx: mpsc::UnboundedSender<ActivityEvent>,
    /// The terminal-result channel, consumed exactly once however the turn ends.
    pub(crate) outcome_tx: oneshot::Sender<PumpEnd>,
    /// Streamed-message buffer: chunks in, deltas and completed messages out.
    pub(crate) accumulator: MessageAccumulator,
}

/// The slot a second turn is handed to the pump through.
///
/// A [`tokio::sync::Mutex`] rather than a channel because the submitting side must be able to take
/// its turn BACK: [`crate::AcpSession::prompt`] fills the slot before it writes the request (so the
/// pump can never read the answer to a turn it has not adopted), and if that write fails it
/// retracts the turn rather than leaving the pump holding one that will never be answered.
pub(crate) type PendingTurn = Arc<tokio::sync::Mutex<Option<OpenTurn>>>;

/// Answers this client is awaiting for requests OTHER than the open prompt —
/// `session/set_config_option` today. An entry is registered BEFORE its
/// request is written (the same slot-before-write rule the prompt follows), so
/// the pump can never read an answer nobody is waiting on; it is drained by
/// the pump when the matching response arrives, retracted by the submitter
/// when its write fails, and emptied when the pump ends so no caller waits on
/// an answer that cannot come.
///
/// A `std::sync::Mutex` deliberately: every hold is a few instructions with no
/// await inside, and the pump's response path is synchronous.
pub(crate) type PendingRequests =
    Arc<std::sync::Mutex<Vec<(JsonRpcId, oneshot::Sender<JsonRpcResponse>)>>>;

/// Register the answer channel for one outbound request, before it is written.
pub(crate) fn register_pending_request(
    requests: &PendingRequests,
    id: JsonRpcId,
    sender: oneshot::Sender<JsonRpcResponse>,
) {
    let mut held = match requests.lock() {
        Ok(guard) => guard,
        Err(poisoned) => poisoned.into_inner(),
    };
    held.push((id, sender));
}

/// Take back the entry for a request whose write failed.
pub(crate) fn retract_pending_request(requests: &PendingRequests, id: &JsonRpcId) {
    let mut held = match requests.lock() {
        Ok(guard) => guard,
        Err(poisoned) => poisoned.into_inner(),
    };
    held.retain(|(pending, _)| pending != id);
}

/// The waiter for `id`, when one is registered.
fn take_pending_request(
    requests: &PendingRequests,
    id: &JsonRpcId,
) -> Option<oneshot::Sender<JsonRpcResponse>> {
    let mut held = match requests.lock() {
        Ok(guard) => guard,
        Err(poisoned) => poisoned.into_inner(),
    };
    let position = held.iter().position(|(pending, _)| pending == id)?;
    Some(held.remove(position).1)
}

/// Drop every waiting answer channel, so each waiter reads a closed channel —
/// a stated ending — instead of waiting forever on a pump that has stopped.
fn fail_pending_requests(requests: &PendingRequests) {
    let mut held = match requests.lock() {
        Ok(guard) => guard,
        Err(poisoned) => poisoned.into_inner(),
    };
    held.clear();
}

/// The immutable context one pump task holds for the life of a session.
///
/// Grouped into a struct rather than positional arguments so a future field cannot be wired into the
/// wrong parameter slot. The per-TURN channels are deliberately NOT here: they are consumed when a
/// turn ends, so they live in [`OpenTurn`] inside the mutable [`PumpState`], where dropping them is
/// expressible.
pub(crate) struct PumpContext<R, W> {
    pub(crate) connection: Arc<JsonRpcConnection<R, W>>,
    pub(crate) session_id: SessionId,
    pub(crate) policy: PermissionPolicy,
    pub(crate) cancelled: Arc<AtomicBool>,
    pub(crate) identity: EventIdentity,
    /// The slot a newly submitted turn arrives in.
    pub(crate) pending: PendingTurn,
    /// The answer channels for outstanding non-prompt requests.
    pub(crate) requests: PendingRequests,
    /// Whether a turn is open. Set by the submitter before it writes the prompt; cleared by this
    /// pump when the turn ends. It is what makes "one turn at a time" a refusal rather than a race.
    pub(crate) busy: Arc<AtomicBool>,
    /// Raised when this pump stops, so a later `prompt` refuses at once instead of queueing a turn
    /// nothing will ever answer.
    pub(crate) ended: Arc<AtomicBool>,
}

/// The mutable per-session state the pump threads through every frame.
struct PumpState {
    /// Best-effort worker-local event sequence, continuous ACROSS turns: one session is one
    /// transcript, and restarting the sequence per turn would make two events collide.
    worker_seq: u64,
    /// The turn currently streaming, if any. `None` between turns — a frame that arrives then has
    /// nowhere durable to go and is logged rather than dropped silently.
    open: Option<OpenTurn>,
    /// Peer-request answers currently being written, each on a task of its own.
    ///
    /// Owned here so they die with the pump: a session that is dropped must not leave a task
    /// writing into a child that is being killed. See [`queue_response`] for why the write cannot
    /// happen inline.
    writers: JoinSet<()>,
}

/// The reader pump: handles every inbound frame until end-of-stream.
///
/// `deferred` carries the notifications that arrived during the pre-prompt handshake, before this
/// pump existed; they are replayed first so the transcript starts with the frames the agent actually
/// sent first.
///
/// Every exit reports a [`PumpEnd`] to `wait_result` if the prompt is still unanswered, so a caller
/// never waits on a channel nobody will write to and never reads a cause that did not happen.
pub(crate) async fn pump_loop<R, W>(
    context: PumpContext<R, W>,
    deferred: Vec<JsonRpcNotification>,
    load_replay: std::ops::Range<usize>,
    first_turn: OpenTurn,
) where
    R: AsyncRead + Unpin + Send + 'static,
    W: AsyncWrite + Unpin + Send + 'static,
{
    let mut state = PumpState {
        worker_seq: 0,
        open: Some(first_turn),
        writers: JoinSet::new(),
    };
    for (index, note) in deferred.into_iter().enumerate() {
        if load_replay.contains(&index) {
            replay_notification(&context, &mut state, note);
        } else {
            handle_notification(&context, &mut state, note);
        }
    }
    loop {
        let message = context.connection.recv().await;
        // Adopted AFTER the read and BEFORE the frame is dispatched. A turn is placed in the
        // pending slot strictly before its `session/prompt` is written, so by the time this pump
        // can read that turn's first frame the slot is already filled: the response can never
        // arrive for a turn this loop has not adopted.
        adopt_pending_turn(&context, &mut state).await;
        match message {
            Ok(Some(IncomingMessage::Notification(note))) => {
                handle_notification(&context, &mut state, note);
            }
            Ok(Some(IncomingMessage::Response(response))) => {
                handle_response(&context, &mut state, response);
            }
            Ok(Some(IncomingMessage::Request(request))) => {
                handle_peer_request(&context, &mut state, request);
            }
            Ok(None) => {
                end_pump(&context, &mut state, UnansweredEnd::StreamEnded).await;
                return;
            }
            Err(HarnessError::Protocol { detail }) => {
                record_unreadable_frame(&context, &mut state, &detail);
            }
            Err(error) => {
                let detail = error.to_string();
                tracing::warn!(%detail, "acp adapter: reader pump ended on a transport error");
                end_pump(&context, &mut state, UnansweredEnd::TransportFailed(detail)).await;
                return;
            }
        }
    }
}

/// Adopts a turn waiting in the pending slot, if any.
///
/// A turn can only be waiting while none is open — [`crate::AcpSession::prompt`] refuses a second
/// one — so this never overwrites a live turn. The assertion is defensive rather than decorative:
/// if it ever failed, the open turn would be dropped with its result channel unconsumed, and its
/// caller would read "the pump ended without reporting" for a turn that was simply displaced.
async fn adopt_pending_turn<R, W>(context: &PumpContext<R, W>, state: &mut PumpState) {
    if state.open.is_some() {
        return;
    }
    let Some(turn) = context.pending.lock().await.take() else {
        return;
    };
    tracing::debug!(
        session = %context.session_id,
        id = ?turn.prompt_id,
        "acp adapter: adopted a new prompt turn on the live session"
    );
    state.open = Some(turn);
}

/// Reports how the session ended to whoever is still waiting, and closes it to new turns.
///
/// The order matters: the `ended` flag is raised FIRST, so a `prompt` racing this shutdown refuses
/// immediately rather than filling a slot nobody will read; then the pending slot is drained (a
/// turn already in it gets the same end the open turn does, rather than a dropped sender the caller
/// would have to interpret); then the open turn is answered.
async fn end_pump<R, W>(context: &PumpContext<R, W>, state: &mut PumpState, end: UnansweredEnd) {
    context.ended.store(true, Ordering::SeqCst);
    fail_pending_requests(&context.requests);
    if let Some(stranded) = context.pending.lock().await.take() {
        let _ = stranded.outcome_tx.send(end.clone().into_pump_end());
    }
    report_end(context, state, end.into_pump_end());
}

/// How the pump stopped with a prompt still outstanding.
///
/// [`PumpEnd`] cannot be cloned — its answered arm carries the agent's own response frame — but a
/// pump that stops with BOTH an open turn and a turn waiting in the pending slot owes the same
/// ending to each. This is that ending, in the only two shapes it can take, so neither waiter is
/// left reading a dropped sender it would have to guess the cause of.
#[derive(Clone)]
enum UnansweredEnd {
    /// The reader reached end-of-stream: the agent closed its stdout.
    StreamEnded,
    /// The reader hit an unrecoverable transport failure.
    TransportFailed(String),
}

impl UnansweredEnd {
    /// The terminal report one waiter receives.
    fn into_pump_end(self) -> PumpEnd {
        match self {
            Self::StreamEnded => PumpEnd::StreamEnded,
            Self::TransportFailed(detail) => PumpEnd::TransportFailed { detail },
        }
    }
}

/// Records a line the transport could not read as a JSON-RPC frame, and keeps pumping.
///
/// This is a **recoverable** condition, not the end of the run. The reader is line-oriented and has
/// already consumed the offending line, so the framing is still aligned; an agent launched through
/// `node` or `npx` that prints `Debugger listening on ws://127.0.0.1:9229` to stdout must not be
/// able to throw away an answer that has already arrived.
///
/// **Known seam limitation:** the offending line's own bytes are not recoverable here. The shared
/// [`JsonRpcConnection::recv`] reports the rejection as a [`HarnessError::Protocol`] carrying serde's
/// message (which names a position, not the content) and does not surface the line itself, so what
/// reaches the transcript is the rejection verbatim rather than the text. Recording the rejection is
/// what is available; inventing a line would be worse than recording none.
fn record_unreadable_frame<R, W>(context: &PumpContext<R, W>, state: &mut PumpState, detail: &str) {
    tracing::warn!(
        %detail,
        "acp adapter: a line on the agent's stdout is not a JSON-RPC frame; recorded and skipped"
    );
    let kind = translate::raw_kind(
        UNREADABLE_FRAME_SOURCE,
        &json!({
            "error": detail,
            "recovered": true,
            "note": "the line was skipped and the turn continued; the transport does not surface \
                     the line's own bytes",
        }),
    );
    emit(context, state, vec![kind]);
}

/// Reports how the turn ended, if it has not already been reported.
///
/// A dropped receiver means the caller abandoned `wait_result`; that is not an error worth
/// surfacing, and it is the only reason the send can fail.
fn report_end<R, W>(context: &PumpContext<R, W>, state: &mut PumpState, end: PumpEnd) {
    if let Some(open) = state.open.take() {
        let _ = open.outcome_tx.send(end);
    }
    context.busy.store(false, Ordering::SeqCst);
}

/// Handles one inbound notification: `session/update` becomes transcript events, anything else is
/// recorded verbatim.
/// The source label on a frame the agent replayed during `session/load`.
pub(crate) const LOAD_REPLAY_SOURCE: &str = "session/load-replay";

/// Handles one notification that arrived while `session/load` was outstanding.
///
/// What an agent streams during a load is the PRIOR conversation, replayed — with one exception:
/// advertisements (`available_commands_update`, `current_mode_update`, `config_option_update`,
/// `session_info_update`) describe the session as it stands NOW, and are handled exactly like
/// live frames so a resumed conversation keeps its command list. Everything else — message and
/// thought chunks, tool calls, plans, anything undecodable or unrecognised — is recorded verbatim
/// under [`LOAD_REPLAY_SOURCE`] and never as the open turn's own frames: the transcript already
/// holds the prior conversation durably, and a resumed turn's answer must fold to its own words
/// alone. (Replayed chunks used to reach the new turn as fresh deltas, so its reading opened with
/// the whole prior conversation before the actual answer.)
fn replay_notification<R, W>(
    context: &PumpContext<R, W>,
    state: &mut PumpState,
    note: JsonRpcNotification,
) {
    if note.method != protocol::METHOD_SESSION_UPDATE {
        handle_notification(context, state, note);
        return;
    }
    let params = note.params.clone().unwrap_or(Value::Null);
    if let Ok(notification) = serde_json::from_value::<SessionNotification>(params.clone())
        && is_standing_update(&notification.update)
    {
        handle_notification(context, state, note);
        return;
    }
    emit(
        context,
        state,
        vec![translate::raw_kind(LOAD_REPLAY_SOURCE, &params)],
    );
}

/// Whether a `session/update` variant states the session's CURRENT standing rather than replaying
/// its history.
const fn is_standing_update(update: &SessionUpdate) -> bool {
    matches!(
        update,
        SessionUpdate::AvailableCommandsUpdate(_)
            | SessionUpdate::CurrentModeUpdate(_)
            | SessionUpdate::ConfigOptionUpdate(_)
            | SessionUpdate::SessionInfoUpdate(_)
    )
}

fn handle_notification<R, W>(
    context: &PumpContext<R, W>,
    state: &mut PumpState,
    note: JsonRpcNotification,
) {
    let params = note.params.unwrap_or(Value::Null);
    if note.method == protocol::METHOD_SESSION_UPDATE {
        emit_session_update(context, state, &params);
        return;
    }
    if note.method == protocol::METHOD_CANCEL_REQUEST {
        // Every inbound peer request is answered by this pump, so there is never one pending for the
        // agent to cancel. It is recorded rather than acted on.
        tracing::debug!(
            method = %note.method,
            "acp adapter: agent cancelled a request; this client holds none pending"
        );
    } else {
        tracing::warn!(method = %note.method, "acp adapter: unrecognised notification method");
    }
    emit(
        context,
        state,
        vec![translate::raw_kind(note.method.as_str(), &params)],
    );
}

/// Decodes a `session/update` notification and forwards the events it produces.
///
/// A frame that does not decode as a [`SessionNotification`] is a protocol violation on the agent's
/// side; it is recorded verbatim rather than dropped, because an operator watching a stalled run
/// needs to see the malformed frame.
fn emit_session_update<R, W>(context: &PumpContext<R, W>, state: &mut PumpState, params: &Value) {
    let notification: SessionNotification = match serde_json::from_value(params.clone()) {
        Ok(notification) => notification,
        Err(source) => {
            tracing::warn!(%source, "acp adapter: undecodable session/update params");
            let kind = translate::raw_kind(
                format!("{}/undecodable", translate::UPDATE_SOURCE_PREFIX),
                &json!({ "error": source.to_string(), "params": params }),
            );
            emit(context, state, vec![kind]);
            return;
        }
    };
    if notification.session_id != context.session_id {
        // One session per run: an update for another session cannot be attributed to this attempt's
        // transcript, but it is still evidence of an agent bug, so it is recorded verbatim.
        tracing::warn!(
            received = %notification.session_id,
            expected = %context.session_id,
            "acp adapter: session/update for an unexpected session id"
        );
        let kind = translate::raw_kind(
            format!("{}/foreign_session", translate::UPDATE_SOURCE_PREFIX),
            &notification,
        );
        emit(context, state, vec![kind]);
        return;
    }
    let Some(open) = state.open.as_mut() else {
        // Between turns there is no accumulator to feed and no transcript to write to. The frame
        // is recorded in the log rather than dropped silently: an agent streaming updates for a
        // turn that is over is a conformance problem an operator needs a trace of.
        tracing::warn!(
            session = %context.session_id,
            "acp adapter: a session/update arrived with no turn open; it has no transcript to \
             reach"
        );
        return;
    };
    let kinds = translate::update_to_kinds(&notification.update, &mut open.accumulator);
    emit(context, state, kinds);
}

/// Handles one inbound Response: the id-matched `session/prompt` reply is the terminal result.
///
/// In order: flush the open messages (so the last durable message precedes the boundary), record the
/// frame verbatim, emit the terminal [`ActivityEventKind::Stop`], hand the response plus the
/// accumulated text to `wait_result`, and then **drop the transcript sender** so the events stream
/// ends — which is the driver's end-of-run signal.
///
/// A Response with any other id answers a request this adapter never sent.
fn handle_response<R, W>(
    context: &PumpContext<R, W>,
    state: &mut PumpState,
    response: JsonRpcResponse,
) {
    // A non-prompt request's answer first — `session/set_config_option` — by
    // exact id. The open prompt's id can never be here: the two are allocated
    // from one counter and registered on different paths.
    if let Some(sender) = take_pending_request(&context.requests, &response.id) {
        if sender.send(response).is_err() {
            tracing::warn!(
                "acp adapter: a configuration request's answer arrived after its caller stopped \
                 waiting; the option's state is whatever the agent next advertises"
            );
        }
        return;
    }
    match state.open.as_ref() {
        Some(open) if open.prompt_id == response.id => {}
        Some(open) => {
            tracing::warn!(
                received = ?response.id,
                expected = ?open.prompt_id,
                "acp adapter: response for an id this client never requested; ignoring the frame"
            );
            return;
        }
        None => {
            tracing::warn!(
                id = ?response.id,
                "acp adapter: a response arrived with no turn open (a second answer for a turn \
                 already reported); ignoring the frame"
            );
            return;
        }
    }
    // TAKEN, not borrowed: the turn's transcript sender must be dropped at the end of this
    // function — that is the driver's end-of-run signal — and the turn's result channel is
    // consumed exactly once here.
    let Some(mut open) = state.open.take() else {
        return;
    };
    let flushed = open.accumulator.finish();
    emit_for(context, &mut state.worker_seq, &open, flushed);
    // Every other frame gets a verbatim twin, and this one decides the activity's result.
    emit_for(
        context,
        &mut state.worker_seq,
        &open,
        vec![translate::prompt_response_kind(&response)],
    );
    emit_for(
        context,
        &mut state.worker_seq,
        &open,
        vec![stop_event_kind(&response)],
    );
    let outcome = PromptOutcome {
        response,
        text: open.accumulator.completed_text().to_owned(),
        final_message: open.accumulator.final_message().to_owned(),
    };
    let OpenTurn {
        outcome_tx,
        event_tx,
        ..
    } = open;
    // A dropped receiver means the caller abandoned the turn; not an error worth surfacing.
    let _ = outcome_tx.send(PumpEnd::Answered(Box::new(outcome)));
    // The session is free to take another turn only once this one's result has been reported.
    context.busy.store(false, Ordering::SeqCst);
    // The transcript sender is dropped here, which is what ends this turn's events stream — the
    // driver's end-of-run signal, unchanged from the one-prompt path.
    drop(event_tx);
}

/// The terminal [`ActivityEventKind::Stop`] for a `session/prompt` Response.
///
/// **Always emitted.** A transcript that just stops mid-stream with no terminal boundary is not an
/// honest record of a failed turn, and there is nothing to invent: `aion_core`'s
/// [`StopKind::Error`] exists for exactly the case where a run stopped because of an error. The
/// three non-success shapes each get one carrying the reason, and the frame itself is on the
/// transcript verbatim beside it ([`translate::prompt_response_kind`]).
///
/// `error` is checked before `result` so the boundary agrees with the terminal interpretation in
/// [`crate::outcome`], which reports an error Response as the failure it is regardless of what else
/// the frame carries.
fn stop_event_kind(response: &JsonRpcResponse) -> EventKind {
    let reason = if let Some(error) = response.error.as_ref() {
        StopKind::Error {
            message: format!(
                "session/prompt failed (code {}): {}",
                error.code, error.message
            ),
        }
    } else if let Some(result) = response.result.as_ref() {
        match serde_json::from_value::<PromptResponse>(result.clone()) {
            Ok(prompt) => translate::stop_kind(prompt.stop_reason),
            Err(source) => {
                tracing::warn!(%source, "acp adapter: undecodable session/prompt result");
                StopKind::Error {
                    message: format!(
                        "the session/prompt result is not an ACP PromptResponse ({source}): {result}"
                    ),
                }
            }
        }
    } else {
        StopKind::Error {
            message: "the session/prompt response carried neither result nor error".to_owned(),
        }
    };
    (ActivityEventKind::Stop { reason }, false)
}

/// Handles one peer-initiated request. Every path queues exactly one response.
fn handle_peer_request<R, W>(
    context: &PumpContext<R, W>,
    state: &mut PumpState,
    request: JsonRpcRequest,
) where
    R: AsyncRead + Unpin + Send + 'static,
    W: AsyncWrite + Unpin + Send + 'static,
{
    if request.method == protocol::METHOD_SESSION_REQUEST_PERMISSION {
        answer_permission_request(context, state, request);
        return;
    }
    // Anything else is a method this client did not advertise (`fs/*`, `terminal/*`) or does not
    // know. It is refused with an answer, never with silence.
    tracing::warn!(
        method = %request.method,
        "acp adapter: refusing a peer request for an unadvertised capability"
    );
    emit(
        context,
        state,
        vec![translate::refused_request_kind(
            &request.method,
            request.params.as_ref(),
        )],
    );
    let refusal = protocol::unadvertised_method_response(request.id, &request.method);
    queue_response(context, state, refusal);
}

/// Records, decides, records again, and answers one `session/request_permission`.
///
/// The order is deliberate: the ask reaches the transcript before the answer is written, so a run
/// that dies mid-decision still shows what was asked. A request that cannot be decoded is refused
/// with `-32602` rather than left unanswered — an undecodable request is still a request the agent is
/// blocked on.
fn answer_permission_request<R, W>(
    context: &PumpContext<R, W>,
    state: &mut PumpState,
    request: JsonRpcRequest,
) where
    R: AsyncRead + Unpin + Send + 'static,
    W: AsyncWrite + Unpin + Send + 'static,
{
    let params = request.params.clone().unwrap_or(Value::Null);
    let decoded: RequestPermissionRequest = match serde_json::from_value(params.clone()) {
        Ok(decoded) => decoded,
        Err(source) => {
            tracing::warn!(%source, "acp adapter: undecodable session/request_permission params");
            emit(
                context,
                state,
                vec![translate::raw_kind(
                    format!(
                        "{}/undecodable",
                        protocol::METHOD_SESSION_REQUEST_PERMISSION
                    ),
                    &json!({ "error": source.to_string(), "params": params }),
                )],
            );
            let refusal = JsonRpcResponse::failure(
                request.id,
                JsonRpcError::new(
                    aion_integrations::jsonrpc::error_codes::INVALID_PARAMS,
                    format!("session/request_permission params are not decodable: {source}"),
                ),
            );
            queue_response(context, state, refusal);
            return;
        }
    };

    emit(
        context,
        state,
        vec![translate::permission_request_kind(&decoded)],
    );

    // A client that has cancelled MUST answer every permission request `cancelled`, whatever the
    // configured policy would otherwise have chosen.
    let outcome = if context.cancelled.load(Ordering::SeqCst) {
        RequestPermissionOutcome::Cancelled
    } else {
        context.policy.decide(&decoded.options)
    };
    emit(
        context,
        state,
        vec![translate::permission_decision_kind(
            context.policy,
            &decoded.tool_call.tool_call_id.to_string(),
            &decoded.options,
            &outcome,
        )],
    );

    let answer = match serde_json::to_value(RequestPermissionResponseShim::of(&outcome)) {
        Ok(result) => JsonRpcResponse::success(request.id, result),
        Err(source) => {
            // Unreachable for the two outcome shapes, but an unanswered request is unacceptable, so
            // the failure is still an ANSWER rather than a dropped frame.
            tracing::error!(%source, "acp adapter: could not encode a permission outcome");
            JsonRpcResponse::failure(
                request.id,
                JsonRpcError::new(
                    aion_integrations::jsonrpc::error_codes::INTERNAL_ERROR,
                    format!("could not encode the permission outcome: {source}"),
                ),
            )
        }
    };
    queue_response(context, state, answer);
}

/// The `session/request_permission` result envelope: `{ "outcome": <outcome> }`.
///
/// ACP wraps the outcome in an object because a JSON-RPC result must be one, and the schema's own
/// [`agent_client_protocol_schema::v1::RequestPermissionResponse`] is `#[non_exhaustive]` — it cannot
/// be constructed field-by-field from outside its crate, and its `new` takes ownership. This borrows
/// the outcome instead so the same value can be recorded on the transcript and sent on the wire
/// without a clone diverging between the two.
#[derive(serde::Serialize)]
struct RequestPermissionResponseShim<'a> {
    outcome: &'a RequestPermissionOutcome,
}

impl<'a> RequestPermissionResponseShim<'a> {
    /// Wraps one outcome for the wire.
    const fn of(outcome: &'a RequestPermissionOutcome) -> Self {
        Self { outcome }
    }
}