aion-server 0.29.0

//! `/assistant/sessions` — the operator's own conversations with an agent.
//!
//! A session is ONE live agent-harness process this server owns on one
//! caller's behalf. It is not a workflow run: there is no history to replay and
//! no determinism boundary, only a durable transcript and a process that dies
//! with the server.
//!
//! # No namespace
//!
//! Every route here is CALLER-scoped and takes no namespace. A session is one
//! operator's conversation and lives in no namespace, so the listing is
//! narrowed by the resolved subject and every addressed route is narrowed by
//! the record's own subject. A session belonging to ANOTHER subject is reported
//! as a plain `404`: answering `403` would confirm that a session with that id
//! exists, which is exactly the existence leak the namespace boundary refuses
//! everywhere else.
//!
//! # Authorization
//!
//! Behind the same [`HttpCaller`] extractor as every data route, and behind the
//! `assistant.sessions` grant on top of it. Spawning an agent process on the
//! server host is not a read: it is a deployment-wide capability, so it takes a
//! deployment-wide grant word rather than a namespace membership.
//!
//! # One status mapping
//!
//! [`session_refusal`] is the ONLY place an [`AssistantSessionError`] becomes an
//! HTTP status, and it is an exhaustive match — a variant added to the surface
//! cannot reach a client without a status being chosen for it here. The
//! WebSocket route reads the same function for its terminal error frame, so a
//! refusal reads identically whether it arrives over a request or a socket.

use aion_core::{
    AssistantCommandInvocation, AssistantSessionFrame, AssistantSessionId, AssistantSessionSummary,
    AssistantTurnContext,
};
use aion_proto::WireError;
use axum::{
    Json,
    extract::{Path, State},
    http::StatusCode,
    response::{IntoResponse, Response},
};
use serde::{Deserialize, Serialize};

use super::auth::HttpCaller;
use super::error::HttpWireError;
use crate::ServerState;
use crate::assistant::sessions::{AssistantSessionError, AssistantSessions};
use crate::error::ServerError;
use crate::namespace::CallerIdentity;
use crate::namespace::grants::{ASSISTANT_SESSIONS, require_grant};

/// `GET /assistant/sessions` — every session this caller owns, newest first.
pub(crate) async fn list_assistant_sessions(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
) -> Result<Json<AssistantSessionListBody>, AssistantHttpError> {
    let sessions = authorized(&state, &caller)?;
    let listed = sessions.list(caller.subject()).await?;
    Ok(Json(AssistantSessionListBody { sessions: listed }))
}

/// `POST /assistant/sessions` — open one.
pub(crate) async fn create_assistant_session(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
    Json(request): Json<CreateAssistantSessionRequest>,
) -> Result<(StatusCode, Json<AssistantSessionSummary>), AssistantHttpError> {
    let sessions = authorized(&state, &caller)?;
    // The one route that cannot possibly succeed on an uncommissioned server,
    // refused here with the availability reason rather than downstream: the
    // registry's own `NotCommissioned` covers the no-harness case only, while
    // this reason also names the `[mcp]`-dark case and states the remedy.
    if let Some(reason) = sessions.availability().reason() {
        return Err(AssistantHttpError::Session(
            StatusCode::SERVICE_UNAVAILABLE,
            WireError::backend(reason.to_owned()).with_error_type(NOT_COMMISSIONED_TYPE),
        ));
    }
    let summary = sessions
        .create(
            caller.subject(),
            request.harness.as_deref(),
            request.account.as_deref(),
            request.title,
        )
        .await?;
    Ok((StatusCode::CREATED, Json(summary)))
}

/// `GET /assistant/sessions/current` — the conversation this caller is in.
///
/// The newest session that can still take a turn, decided ONCE by the server.
/// An operator holds one conversation and the console shows it in more than one
/// place; each surface deriving "which one am I in" from the listing would be
/// two derivations that can disagree, and a dock panel and an editor bar
/// disagreeing about which conversation is open is not a rendering difference —
/// it is two conversations.
///
/// No current session is a `404` carrying the same body shape as any other
/// not-found here, never a `200` with a null: a client that must branch on the
/// answer branches on the status, as it does everywhere else on this surface.
pub(crate) async fn current_assistant_session(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
) -> Result<Json<AssistantSessionSummary>, AssistantHttpError> {
    let sessions = authorized(&state, &caller)?;
    match sessions.current(caller.subject()).await? {
        Some(summary) => Ok(Json(summary)),
        None => Err(AssistantHttpError::Session(
            StatusCode::NOT_FOUND,
            WireError::not_found_with_type(
                NOT_FOUND_TYPE,
                "this caller holds no assistant session that can take another turn; open one with \
                 POST /assistant/sessions",
            ),
        )),
    }
}

/// `GET /assistant/sessions/{id}` — the session and its whole transcript.
pub(crate) async fn read_assistant_session(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
    Path(id): Path<String>,
) -> Result<Json<AssistantSessionDetailBody>, AssistantHttpError> {
    let sessions = authorized(&state, &caller)?;
    let (summary, transcript) = sessions.read(caller.subject(), session_id(&id)?).await?;
    Ok(Json(AssistantSessionDetailBody {
        summary,
        transcript,
    }))
}

/// `POST /assistant/sessions/{id}/turns` — ask the agent something.
pub(crate) async fn submit_assistant_turn(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
    Path(id): Path<String>,
    Json(request): Json<AssistantTurnRequest>,
) -> Result<(StatusCode, Json<AssistantTurnAcceptedBody>), AssistantHttpError> {
    let sessions = authorized(&state, &caller)?;
    let session = session_id(&id)?;
    let command: Option<AssistantCommandInvocation> = request.command.map(Into::into);
    if command.is_none() && request.text.trim().is_empty() {
        // Refused at the boundary rather than sent: a prompt of whitespace
        // spends a turn, records a request nobody made, and comes back as an
        // agent's confusion instead of a refusal naming the field. A COMMAND
        // turn is exempt because the command IS the ask — a `compact` with no
        // words is a complete request.
        return Err(bad_request(
            "a turn carries the operator's prompt, and `text` is empty with no `command` to run; \
             send the text to ask, or name a command the harness advertised",
        ));
    }
    if let Some(command) = command.as_ref()
        && command.name.trim().is_empty()
    {
        return Err(bad_request(
            "a turn's `command.name` is empty; name the command the harness advertised, or omit \
             `command` and send `text`",
        ));
    }
    let turn_id = sessions
        .turn(
            caller.subject(),
            session,
            request.text,
            request.context,
            command,
        )
        .await?;
    Ok((
        StatusCode::ACCEPTED,
        Json(AssistantTurnAcceptedBody { turn_id }),
    ))
}

/// `PUT /assistant/sessions/{id}/context` — what is on the operator's screen
/// NOW, with no turn attached.
///
/// The same record a turn carries, pushed on its own so the agent's
/// `assistant_context` tool can read the current screen mid-turn rather than
/// the screen the turn started on.
pub(crate) async fn push_assistant_context(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
    Path(id): Path<String>,
    Json(context): Json<AssistantTurnContext>,
) -> Result<StatusCode, AssistantHttpError> {
    let sessions = authorized(&state, &caller)?;
    sessions
        .push_context(caller.subject(), session_id(&id)?, context)
        .await?;
    Ok(StatusCode::NO_CONTENT)
}

/// `POST /assistant/sessions/{id}/cancel` — stop the open turn.
pub(crate) async fn cancel_assistant_turn(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
    Path(id): Path<String>,
) -> Result<StatusCode, AssistantHttpError> {
    let sessions = authorized(&state, &caller)?;
    sessions.cancel(caller.subject(), session_id(&id)?).await?;
    Ok(StatusCode::ACCEPTED)
}

/// `POST /assistant/sessions/{id}/resume` — reopen a dormant session.
pub(crate) async fn resume_assistant_session(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
    Path(id): Path<String>,
) -> Result<(StatusCode, Json<AssistantSessionSummary>), AssistantHttpError> {
    let sessions = authorized(&state, &caller)?;
    let summary = sessions.resume(caller.subject(), session_id(&id)?).await?;
    Ok((StatusCode::ACCEPTED, Json(summary)))
}

/// `DELETE /assistant/sessions/{id}` — end the session and forget it.
pub(crate) async fn delete_assistant_session(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
    Path(id): Path<String>,
) -> Result<StatusCode, AssistantHttpError> {
    let sessions = authorized(&state, &caller)?;
    sessions.delete(caller.subject(), session_id(&id)?).await?;
    Ok(StatusCode::NO_CONTENT)
}

/// `GET /assistant/sessions` — the listing body.
#[derive(Debug, Serialize)]
pub(crate) struct AssistantSessionListBody {
    /// Every session this caller owns, newest first.
    sessions: Vec<AssistantSessionSummary>,
}

/// `GET /assistant/sessions/{id}` — the session and its whole transcript.
///
/// The transcript is served WHOLE rather than paged: the conversation is what a
/// client renders on open, and a page boundary in the middle of one would put a
/// client in the position of stitching a conversation together to display it.
#[derive(Debug, Serialize)]
pub(crate) struct AssistantSessionDetailBody {
    /// What the session is right now.
    summary: AssistantSessionSummary,
    /// Every frame, in transcript order.
    transcript: Vec<AssistantSessionFrame>,
}

/// `POST /assistant/sessions/{id}/turns` — the accepted turn's identity.
#[derive(Debug, Serialize)]
pub(crate) struct AssistantTurnAcceptedBody {
    /// The turn every frame it produces is tagged with, so a client can follow
    /// one turn on a socket that carries the whole session.
    turn_id: String,
}

/// `POST /assistant/sessions` — the request body.
///
/// Every field is optional and absence is a real answer: no harness named takes
/// the configured default, no account named takes the harness's own, and no
/// title leaves the server to derive one from the first prompt.
#[derive(Debug, Default, Deserialize)]
#[serde(default)]
pub(crate) struct CreateAssistantSessionRequest {
    /// The configured harness to run, or `None` for the default.
    harness: Option<String>,
    /// The configured account to run it under, or `None` for the harness's own.
    account: Option<String>,
    /// The operator's title for the conversation, or `None` to derive one.
    title: Option<String>,
}

/// `POST /assistant/sessions/{id}/turns` — the request body.
#[derive(Debug, Default, Deserialize)]
#[serde(default)]
pub(crate) struct AssistantTurnRequest {
    /// The operator's prompt. Required UNLESS `command` names one, because a
    /// command with no words after it is a complete request.
    text: String,
    /// What was on the operator's screen when they asked, composed into the
    /// prompt as prose by the server. Absent is a complete answer.
    context: Option<AssistantTurnContext>,
    /// The harness command this turn invokes, when it invokes one.
    ///
    /// Refused by name when the harness has not advertised it: a client offers
    /// what the session's `commands` say, and anything else is a control that
    /// should never have been on the screen.
    command: Option<AssistantCommandRequest>,
}

/// `{ command: { name, input } }` on a turn.
///
/// Its own request type rather than the core invocation, so the wire shape is
/// declared where the wire is read and a field added to the domain type does not
/// silently become accepted input.
#[derive(Debug, Deserialize)]
pub(crate) struct AssistantCommandRequest {
    /// The advertised command's name, without the leading `/`.
    name: String,
    /// The text that follows the command name, when the operator typed any.
    #[serde(default)]
    input: Option<String>,
}

impl From<AssistantCommandRequest> for AssistantCommandInvocation {
    fn from(request: AssistantCommandRequest) -> Self {
        Self {
            name: request.name,
            input: request.input,
        }
    }
}

/// A refusal from the assistant-session surface, as an HTTP response.
pub(crate) enum AssistantHttpError {
    /// A refusal whose status follows its wire code through the shared
    /// [`HttpWireError`] mapping — the grant denial and the malformed-id
    /// refusal both take it, so they read exactly as they do everywhere else.
    Wire(HttpWireError),
    /// A session-surface refusal, whose status [`session_refusal`] decided.
    ///
    /// Carried as an explicit status because three of them have no wire code
    /// that maps to the right one: `503` for an uncommissioned surface and
    /// `502` for a harness that failed or is not logged in are statements about
    /// a downstream process, and the shared mapping has no code for either.
    Session(StatusCode, WireError),
}

impl IntoResponse for AssistantHttpError {
    fn into_response(self) -> Response {
        match self {
            Self::Wire(error) => error.into_response(),
            Self::Session(status, wire) => (status, Json(wire)).into_response(),
        }
    }
}

impl From<ServerError> for AssistantHttpError {
    fn from(error: ServerError) -> Self {
        Self::Wire(HttpWireError(error.to_wire_error()))
    }
}

impl From<AssistantSessionError> for AssistantHttpError {
    fn from(error: AssistantSessionError) -> Self {
        let (status, wire) = session_refusal(&error);
        Self::Session(status, wire)
    }
}

/// The HTTP status and wire error for one session-surface refusal.
///
/// THE mapping — exhaustive, so a variant added to
/// [`AssistantSessionError`] fails to compile here until a status is chosen for
/// it, and shared with the WebSocket route so one refusal cannot mean two
/// different things on two transports.
///
/// [`AssistantSessionError::NotYours`] is deliberately not rendered from its own
/// `Display`: that message names the subject a session belongs to, and this
/// answer goes to a caller who must not learn the session exists at all. It is
/// therefore reported with the identical message a genuine not-found produces.
pub(crate) fn session_refusal(error: &AssistantSessionError) -> (StatusCode, WireError) {
    match error {
        AssistantSessionError::NotFound { session_id }
        | AssistantSessionError::NotYours { session_id, .. } => (
            StatusCode::NOT_FOUND,
            WireError::not_found_with_type(
                NOT_FOUND_TYPE,
                format!("assistant session {session_id} was not found"),
            ),
        ),
        AssistantSessionError::Busy { .. } => (
            StatusCode::CONFLICT,
            WireError::invalid_state_with_type(BUSY_TYPE, error.to_string()),
        ),
        AssistantSessionError::Ended { .. } => (
            StatusCode::CONFLICT,
            WireError::invalid_state_with_type(ENDED_TYPE, error.to_string()),
        ),
        AssistantSessionError::UnknownHarness { .. } => (
            StatusCode::BAD_REQUEST,
            WireError::invalid_input(error.to_string()).with_error_type(UNKNOWN_HARNESS_TYPE),
        ),
        AssistantSessionError::UnknownAccount { .. } => (
            StatusCode::BAD_REQUEST,
            WireError::invalid_input(error.to_string()).with_error_type(UNKNOWN_ACCOUNT_TYPE),
        ),
        // 400, not 409: the caller named a command that is not part of this
        // conversation's vocabulary. Nothing about the session's state has to
        // change for the request to become valid — a different name would do.
        AssistantSessionError::UnknownCommand { .. } => (
            StatusCode::BAD_REQUEST,
            WireError::invalid_input(error.to_string()).with_error_type(UNKNOWN_COMMAND_TYPE),
        ),
        AssistantSessionError::NotCommissioned { .. } => (
            StatusCode::SERVICE_UNAVAILABLE,
            WireError::backend(error.to_string()).with_error_type(NOT_COMMISSIONED_TYPE),
        ),
        // 503, not 400: the request named a real harness and the machine cannot
        // run it, which is a fact about this server that an operator fixes by
        // installing something. The message carries the launch line and the
        // catalogue's install hint, and it is the SAME message the first turn
        // would have produced — one refusal, met earlier.
        AssistantSessionError::HarnessUnavailable { .. } => (
            StatusCode::SERVICE_UNAVAILABLE,
            WireError::backend(error.to_string()).with_error_type(HARNESS_UNAVAILABLE_TYPE),
        ),
        // 503 for the same reason: the account is declared and the value it
        // needs is not on this server, so nothing the caller sends would help.
        AssistantSessionError::AccountEnvironmentAbsent { .. } => (
            StatusCode::SERVICE_UNAVAILABLE,
            WireError::backend(error.to_string()).with_error_type(ACCOUNT_ENVIRONMENT_TYPE),
        ),
        // 502, not 500: the agent is an upstream this server spoke to and got a
        // refusal from. Nothing in the console can enter a credential — the
        // operator logs the harness in on the server host — so the error's own
        // message, which says exactly that, is what is served.
        AssistantSessionError::AuthRequired { .. } => (
            StatusCode::BAD_GATEWAY,
            WireError::backend(error.to_string()).with_error_type(AUTH_REQUIRED_TYPE),
        ),
        AssistantSessionError::HarnessFailed { .. } => (
            StatusCode::BAD_GATEWAY,
            WireError::backend(error.to_string()).with_error_type(HARNESS_FAILED_TYPE),
        ),
        AssistantSessionError::Store(_) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            WireError::backend(error.to_string()).with_error_type(STORE_TYPE),
        ),
        AssistantSessionError::Internal(_) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            WireError::backend(error.to_string()).with_error_type(INTERNAL_TYPE),
        ),
    }
}

/// The `assistant.sessions` grant, checked before any session work.
///
/// Returns the registry rather than a unit so no route can reach it without
/// passing through the refusal first: there is one way to get an
/// [`AssistantSessions`] in this module, and it is authorized.
fn authorized<'state>(
    state: &'state ServerState,
    caller: &CallerIdentity,
) -> Result<&'state AssistantSessions, AssistantHttpError> {
    require_grant(caller, &ASSISTANT_SESSIONS)?;
    Ok(state.assistant_sessions())
}

/// Parse a `{id}` path segment, refusing with a `400` that names the text.
pub(crate) fn session_id(text: &str) -> Result<AssistantSessionId, AssistantHttpError> {
    AssistantSessionId::parse(text)
        .map_err(|error| AssistantHttpError::Wire(HttpWireError(invalid_id(&error.to_string()))))
}

/// A `400` naming what the caller sent.
fn bad_request(message: impl Into<String>) -> AssistantHttpError {
    AssistantHttpError::Wire(HttpWireError(
        WireError::invalid_input(message).with_error_type(INVALID_REQUEST_TYPE),
    ))
}

/// The wire error a malformed session id produces, shared with the socket
/// route so a bad id reads the same on both transports.
pub(crate) fn invalid_id(message: &str) -> WireError {
    WireError::invalid_input(message.to_owned()).with_error_type(INVALID_SESSION_ID_TYPE)
}

/// `error_type` for a session id that is not a session id.
const INVALID_SESSION_ID_TYPE: &str = "AssistantSessionIdInvalid";
/// `error_type` for a request body this surface refuses.
const INVALID_REQUEST_TYPE: &str = "AssistantRequestInvalid";
/// `error_type` for a session this caller cannot see.
const NOT_FOUND_TYPE: &str = "AssistantSessionNotFound";
/// `error_type` for a turn submitted while one is open.
const BUSY_TYPE: &str = "AssistantSessionBusy";
/// `error_type` for a session that cannot take another turn.
const ENDED_TYPE: &str = "AssistantSessionEnded";
/// `error_type` for a harness name the configuration does not declare.
const UNKNOWN_HARNESS_TYPE: &str = "AssistantHarnessUnknown";
/// `error_type` for an account name the harness does not declare.
const UNKNOWN_ACCOUNT_TYPE: &str = "AssistantAccountUnknown";
/// `error_type` for a command the harness has not advertised.
pub(crate) const UNKNOWN_COMMAND_TYPE: &str = "AssistantCommandUnknown";
/// `error_type` for a server that cannot open an assistant session at all.
pub(crate) const NOT_COMMISSIONED_TYPE: &str = "AssistantSessionsNotCommissioned";
/// `error_type` for a harness whose launch program is not on the server's PATH.
pub(crate) const HARNESS_UNAVAILABLE_TYPE: &str = "AssistantHarnessUnavailable";
/// `error_type` for an account whose environment this server does not carry.
const ACCOUNT_ENVIRONMENT_TYPE: &str = "AssistantAccountEnvironmentAbsent";
/// `error_type` for an agent that must be logged in on the server host.
const AUTH_REQUIRED_TYPE: &str = "AssistantHarnessAuthRequired";
/// `error_type` for a harness that could not be started.
const HARNESS_FAILED_TYPE: &str = "AssistantHarnessFailed";
/// `error_type` for a durable store that refused or failed.
const STORE_TYPE: &str = "AssistantSessionStoreFailed";
/// `error_type` for something this server itself could not do.
const INTERNAL_TYPE: &str = "AssistantSessionsInternal";

#[cfg(test)]
#[path = "assistant_sessions_tests.rs"]
mod assistant_sessions_tests;