aion-server 0.30.0

//! Driving one turn from the harness's stream onto the record.
//!
//! The task that owns a turn does four things in order and nothing else:
//! translate every event the harness produced, record it (durably first, live
//! second), interpret the turn's terminal result, and release the session's
//! turn slot. It runs to completion on every path — including the failing
//! ones — because a turn that ends without a terminal frame leaves a console
//! spinning on a conversation that is over.

use std::sync::Arc;

use aion_core::AssistantSessionEvent;
use aion_integration_acp::{PromptOutcome, TurnHandle};
use aion_integrations::HarnessError;
use aion_integrations::contract::AgentSession as _;
use futures::StreamExt;

use super::error::AssistantSessionError;
use super::frames::TurnFrames;
use super::live::{LiveSession, Recorder};

/// The `code` a turn carries when the harness has no login on the server host.
///
/// Named once, and named here, because it is the ONE failure a console renders
/// as instructions rather than as text.
pub(crate) const AUTH_REQUIRED_CODE: &str = "auth_required";

/// The ACP error code an agent answers with to demand authentication.
const ACP_AUTH_REQUIRED: i64 = -32000;

/// Run one turn to its terminal frame.
///
/// UNBOUNDED, and deliberately: a turn ends when the harness ends it or when the
/// caller cancels it (RULED 2026-08-29, the out-of-the-box amendment). The
/// per-turn timeout this used to carry was retired rather than defaulted,
/// because there is no honest answer to "how long may an agent think" that is
/// not somebody's real work cancelled mid-thought — and with the knob gone, an
/// invented number would be one nobody could raise. What bounds a turn instead
/// is the operator: `POST /assistant/sessions/{id}/cancel` delivers ACP's own
/// cancel, and ending the session takes the agent's process group with it.
pub(crate) async fn drive(live: Arc<LiveSession>, handle: TurnHandle, turn_id: String) {
    let recorder = live.recorder().clone();
    if let Err(error) = pump(&recorder, handle, &turn_id).await {
        record_failure(&recorder, &turn_id, "record_failed", &error).await;
    }
    live.release_turn();
}

/// Drain the turn's events onto the record, then interpret its result.
async fn pump(
    recorder: &Recorder,
    mut handle: TurnHandle,
    turn_id: &str,
) -> Result<(), AssistantSessionError> {
    let mut frames = TurnFrames::new(turn_id);
    let mut events = handle.events();
    while let Some(event) = events.next().await {
        for frame in frames.translate(event) {
            recorder.record(frame).await?;
        }
    }
    // Anything the translator was still holding — an ask whose decision never
    // came — reaches the record before the terminal frame rather than dying
    // with the translator.
    for frame in frames.flush() {
        recorder.record(frame).await?;
    }
    let terminal = match handle.wait().await {
        Ok(outcome) => completed_or_failed(turn_id, &outcome),
        Err(error) => AssistantSessionEvent::TurnFailed {
            turn_id: turn_id.to_owned(),
            code: harness_error_code(&error).to_owned(),
            message: error.to_string(),
        },
    };
    recorder.record(terminal).await.map(drop)
}

/// The terminal frame for a turn the agent answered.
///
/// An answered turn is not automatically a completed one: an agent may answer
/// `session/prompt` with a JSON-RPC error, and `-32000` in particular is the
/// authentication demand only the operator can satisfy, out of band, on the
/// server host.
fn completed_or_failed(turn_id: &str, outcome: &PromptOutcome) -> AssistantSessionEvent {
    if let Some(error) = outcome.response.error.as_ref() {
        let code = if error.code == ACP_AUTH_REQUIRED {
            AUTH_REQUIRED_CODE.to_owned()
        } else {
            format!("agent_error_{}", error.code.abs())
        };
        return AssistantSessionEvent::TurnFailed {
            turn_id: turn_id.to_owned(),
            code,
            message: error.message.clone(),
        };
    }
    let stop_reason = outcome
        .response
        .result
        .as_ref()
        .and_then(|result| result.get("stopReason"))
        .and_then(serde_json::Value::as_str)
        .unwrap_or(UNSTATED_STOP_REASON)
        .to_owned();
    AssistantSessionEvent::TurnCompleted {
        turn_id: turn_id.to_owned(),
        final_message: outcome.final_message.clone(),
        stop_reason,
        session_ref: None,
    }
}

/// The stop reason reported for a response that carried none.
///
/// A conforming agent always states one; naming the absence keeps it
/// distinguishable from an agent that really said `end_turn`.
pub(crate) const UNSTATED_STOP_REASON: &str = "unstated";

/// The `code` for a harness failure: the typed variant's own name, in
/// `snake_case`.
///
/// The variant name rather than a message, because a console branches on the
/// code and reads the message. Exhaustive on purpose: a new harness error must
/// be given a code rather than silently joining another's.
fn harness_error_code(error: &HarnessError) -> &'static str {
    match error {
        HarnessError::CapabilityNotSupported { .. } => "capability_not_supported",
        HarnessError::StaleTarget { .. } => "stale_target",
        HarnessError::Occupied { .. } => "occupied",
        HarnessError::Transport { .. } => "transport",
        HarnessError::Protocol { .. } => "protocol",
        HarnessError::Harness { .. } => "harness",
        HarnessError::PolicyRefused { .. } => "policy_refused",
        HarnessError::Configuration { .. } => "configuration",
        HarnessError::Contract { .. } => "contract",
        // `HarnessError` is `#[non_exhaustive]`: a variant added upstream must
        // still reach a console as SOMETHING branchable rather than as an
        // unhandled shape, and the message beside it carries the detail.
        _ => "harness_error",
    }
}

/// Cancel the open turn on a live session — the one intervention ACP
/// advertises.
pub(crate) async fn cancel(live: &Arc<LiveSession>) -> Result<(), AssistantSessionError> {
    let session_id = live.session_id();
    let delivered = live
        .with_session(async |session| {
            session
                .intervene(aion_core::InterventionCommand {
                    workflow_id: aion_core::WorkflowId::new(session_id.as_uuid()),
                    run_id: aion_core::RunId::new(session_id.as_uuid()),
                    activity_id: aion_core::ActivityId::from_sequence_position(1),
                    attempt: 1,
                    issued_by: None,
                    issued_at: chrono::Utc::now(),
                    kind: aion_core::InterventionKind::Cancel {
                        reason: "the operator stopped this turn".to_owned(),
                    },
                })
                .await
        })
        .await;
    match delivered {
        None => Err(AssistantSessionError::Ended {
            session_id,
            reason: "the harness process has already been shut down".to_owned(),
        }),
        Some(Err(error)) => Err(AssistantSessionError::HarnessFailed {
            harness: session_id.to_string(),
            reason: error.to_string(),
        }),
        Some(Ok(())) => Ok(()),
    }
}

/// Record a terminal frame, logging rather than propagating a store failure.
///
/// This is the LAST write of a turn and there is nobody left to return an error
/// to; a failure here means the record is incomplete, which must be loud in the
/// log even though it can no longer be answered to a caller.
async fn record_terminal(recorder: &Recorder, event: AssistantSessionEvent) {
    if let Err(error) = recorder.record(event).await {
        tracing::error!(
            session = %recorder.session_id(),
            %error,
            "an assistant turn's terminal frame could not be recorded; the transcript is \
             incomplete for this turn"
        );
    }
}

/// Report a failure that happened while RECORDING, which cannot itself be
/// recorded through the same path that just failed.
async fn record_failure(
    recorder: &Recorder,
    turn_id: &str,
    code: &str,
    error: &AssistantSessionError,
) {
    tracing::error!(
        session = %recorder.session_id(),
        turn = %turn_id,
        %error,
        "an assistant turn could not be recorded"
    );
    record_terminal(
        recorder,
        AssistantSessionEvent::TurnFailed {
            turn_id: turn_id.to_owned(),
            code: code.to_owned(),
            message: error.to_string(),
        },
    )
    .await;
}