aion_core/

assistant_session.rs

//! The assistant-session vocabulary: identity, state, the summary a listing
//! carries, the context a turn is asked with, and the frames a session streams.
//!
//! An assistant session is ONE live agent-harness process the server owns on a
//! caller's behalf — not a workflow run. It has no history to replay and no
//! determinism boundary; what it has is a durable transcript (every request and
//! every event, in order) and a process that dies with the server.
//!
//! These types live in `aion-core` rather than in `aion-server` for the reason
//! every other wire record does: the store persists them, the server serves
//! them, and the ops console's TypeScript types are generated from this crate.
//! One declaration, three consumers, no hand-kept copy.
//!
//! # The frames are the transcript
//!
//! [`AssistantSessionEvent`] is exactly what the WebSocket streams AND exactly
//! what is appended to the durable transcript, so replaying the transcript and
//! watching live are the same stream seen at two times. The index a frame
//! carries ([`AssistantSessionFrame`]) is assigned by the store, densely, and is
//! what a reconnecting client passes back as `?after=`.

use chrono::{DateTime, Utc};

use crate::assistant_document::{
    AssistantDocumentEditError, AssistantDocumentEditOp, apply_document_edits,
};
use serde::{Deserialize, Serialize};
use uuid::Uuid;

/// Identifier for one assistant session.
///
/// A UUID rather than an operator-chosen name: a session is minted by the
/// server, is never addressed by a name a human types, and a UUID keys a
/// fixed-width durable record without an escaping rule.
#[derive(
    Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord,
)]
pub struct AssistantSessionId(Uuid);

impl AssistantSessionId {
    /// Creates a session identifier from an existing UUID.
    #[must_use]
    pub const fn new(id: Uuid) -> Self {
        Self(id)
    }

    /// Mints a fresh session identifier.
    #[must_use]
    pub fn new_v4() -> Self {
        Self(Uuid::new_v4())
    }

    /// Returns the UUID backing this identifier.
    #[must_use]
    pub const fn as_uuid(&self) -> Uuid {
        self.0
    }

    /// Parses a session identifier from its canonical textual form.
    ///
    /// # Errors
    ///
    /// Returns [`AssistantSessionIdError`] naming the text that is not a UUID.
    /// A caller-supplied path segment is parsed here rather than pattern-matched
    /// somewhere downstream, so a malformed id is one refusal with one message.
    pub fn parse(text: &str) -> Result<Self, AssistantSessionIdError> {
        Uuid::parse_str(text)
            .map(Self)
            .map_err(|_source| AssistantSessionIdError {
                text: text.to_owned(),
            })
    }
}

impl std::fmt::Display for AssistantSessionId {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.0.fmt(formatter)
    }
}

/// A session identifier that could not be parsed from its textual form.
#[derive(thiserror::Error, Clone, Debug, PartialEq, Eq)]
#[error("`{text}` is not an assistant session id (session ids are UUIDs minted by the server)")]
pub struct AssistantSessionIdError {
    /// The text that was offered as a session id.
    pub text: String,
}

/// What a session is, as the wire reports it — three answers and no more.
///
/// A PROJECTION, never a stored field, exactly as `WorkflowStatus` is: it is
/// derived from the session's appended transcript records plus the live-process
/// registry. Nothing writes a state; things append records, and this is read off
/// them.
///
/// The states a UI might expect and will not find here are deliberate:
///
/// - **"busy" is not a state.** Whether a turn is open is what the transcript
///   already says, so a session cannot be busy on the wire and idle in its
///   frames. The refusal a caller needs — a second turn while one is open — is
///   a `409` on the turn, not a state to render.
/// - **"not logged in" is not a state.** It is the `auth_required` code on the
///   turn that hit it, which names the turn it failed on; a session-wide state
///   would claim the whole conversation was unusable when one turn was.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq, Hash)]
#[serde(rename_all = "snake_case")]
pub enum AssistantSessionState {
    /// A harness process is running for this session right now.
    Live,
    /// No process — but the harness advertised `loadSession`, so the
    /// conversation is the harness's OWN and can be reopened. The next turn
    /// respawns it and opens with `session/load`, so a dormant session is
    /// continuable and is never an error.
    Dormant,
    /// Terminal and read-only: it was deleted, or its harness cannot reload a
    /// prior conversation, so there is nothing left to return to.
    Ended,
}

impl AssistantSessionState {
    /// Whether a process is running for this session.
    #[must_use]
    pub const fn is_live(self) -> bool {
        matches!(self, Self::Live)
    }

    /// Whether another turn can be taken — now, or after a resume.
    #[must_use]
    pub const fn is_continuable(self) -> bool {
        matches!(self, Self::Live | Self::Dormant)
    }
}

/// One row of the session list.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct AssistantSessionSummary {
    /// The session's identity.
    pub session_id: AssistantSessionId,
    /// The configured harness name this session runs.
    pub harness: String,
    /// The configured account name, when the harness declares any.
    pub account: Option<String>,
    /// What the session is, right now.
    pub state: AssistantSessionState,
    /// Why it is in that state, in the server's own words, or `None`.
    ///
    /// An ended session carries the reason it ended; a dormant one carries why
    /// its process went away. A live one carries nothing, because nothing
    /// decided it.
    pub reason: Option<String>,
    /// When the session was created.
    pub created_at: DateTime<Utc>,
    /// The last time anything about the session changed.
    pub updated_at: DateTime<Utc>,
    /// How many turns have been submitted on it.
    pub turns: u64,
    /// The first 80 characters of the first prompt, or `None` before one.
    pub title: Option<String>,
    /// The commands the HARNESS most recently advertised, in the order it
    /// advertised them. Empty when it has advertised none.
    ///
    /// A projection over the transcript's
    /// [`AssistantSessionEvent::AvailableCommands`] records, never a stored
    /// field, and never a list this server composes: a command a client offers
    /// that the agent never advertised is a refusal waiting to happen, so the
    /// only honest source is what the agent itself said.
    pub commands: Vec<AssistantCommand>,
}

/// One command the harness advertised on this session.
///
/// The three fields ACP's `AvailableCommand` carries that a client needs to
/// OFFER one: what to send, what it does, and — when the command takes input —
/// the hint to show before any has been typed. ACP's only input form is
/// `unstructured` ("all text that was typed after the command name is provided
/// as input"), so a hint is the whole of what there is to publish about it.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct AssistantCommand {
    /// The command's name, as sent on a turn.
    pub name: String,
    /// What the agent says it does.
    pub description: String,
    /// The hint to show while the command's input is empty, when it takes one.
    pub input_hint: Option<String>,
}

/// A harness command a turn invokes.
///
/// ACP delivers a command as ORDINARY PROMPT TEXT — `/name` followed by
/// whatever input was typed after it, which is exactly what the schema's
/// `unstructured` input form says the agent receives. So this is not an
/// alternative transport; it is a statement of intent the server turns into the
/// spec's own delivery form, checks against what the agent advertised, and
/// records verbatim so the transcript says a command was invoked rather than
/// leaving a reader to infer it from a leading slash.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct AssistantCommandInvocation {
    /// The advertised command's name, without the leading `/`.
    pub name: String,
    /// The text typed after the command name, when any was.
    pub input: Option<String>,
}

impl AssistantCommandInvocation {
    /// The prompt line ACP delivers this invocation as.
    ///
    /// `/name` alone, or `/name input` — the one place the delivery form is
    /// spelled, so the wire tests and the turn path cannot disagree about it.
    #[must_use]
    pub fn prompt_line(&self) -> String {
        match self
            .input
            .as_deref()
            .map(str::trim)
            .filter(|input| !input.is_empty())
        {
            Some(input) => format!("/{} {input}", self.name),
            None => format!("/{}", self.name),
        }
    }
}

/// A position in a document, in editor coordinates.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq)]
pub struct AssistantDocumentPosition {
    /// Zero-based line number, as the editor's own model holds it.
    ///
    /// ZERO-based on the wire and ONE-based in the prose: the numbers a
    /// document model uses and the numbers an operator reads off a gutter are
    /// different numbers, and converting once — where the prose is composed —
    /// is what keeps them from drifting.
    pub line: u32,
    /// Zero-based column, counted in UTF-16 code units (the editor's own unit).
    pub column: u32,
}

/// A selected range in a document.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq)]
pub struct AssistantDocumentSelection {
    /// Where the selection starts.
    pub from: AssistantDocumentPosition,
    /// Where the selection ends.
    pub to: AssistantDocumentPosition,
}

/// The document a turn is asked about.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct AssistantDocumentContext {
    /// The document's workspace-relative path.
    pub path: String,
    /// The document's current text, as the editor holds it.
    pub text: String,
    /// The selection that scopes the request, when there is one.
    pub selection: Option<AssistantDocumentSelection>,
    /// Where the caret is, when the document came from a live editor.
    ///
    /// NOT a zero-width selection: a caret is where the operator IS and a
    /// selection is what they CHOSE, and an agent told a region was selected
    /// when none was edits the wrong thing. Carried as its own field so the
    /// prose can say "cursor on line 4" with no selection at all — and so the
    /// operator's own words never have to carry it (nothing is appended to
    /// what they typed; Tom, 2026-08-30).
    pub cursor: Option<AssistantDocumentPosition>,
}

/// What was on the operator's screen when they asked.
///
/// Composed into the prompt as prose by the server, and SHOWN to the operator
/// before it is sent — the console renders the same prefix the server will
/// build, and the operator may opt out of it.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, Default, PartialEq, Eq)]
pub struct AssistantTurnContext {
    /// The console URL the operator was on.
    pub url: Option<String>,
    /// The titles of the explain concepts declared on that screen.
    pub concepts: Vec<String>,
    /// The document under the editor's cursor, when the turn came from there.
    pub document: Option<AssistantDocumentContext>,
}

/// How a tool call is going.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum AssistantToolCallStatus {
    /// The agent has begun the call.
    Started,
    /// The call finished and its output is on the frame.
    Completed,
    /// The call failed; the failure is on the frame's output.
    Failed,
}

/// How a permission request was decided.
///
/// There is no console prompt in this cut: the configured policy decides, and
/// the decision is recorded so an operator can see what was asked and what was
/// answered.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum AssistantPermissionDecision {
    /// The policy allowed the call, once.
    AllowOnce,
    /// The policy denied the call.
    Deny,
}

/// One event on a session's transcript, and one frame on its WebSocket.
///
/// The SAME value is durably appended and streamed live, so replay and live are
/// one stream read at two times rather than two encodings that could disagree.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum AssistantSessionEvent {
    /// A harness process opened (or reopened) the conversation.
    ///
    /// Appended once per spawn — so a resumed session has two of these, and the
    /// LAST one carries the capabilities of the process that is running now. It
    /// is what makes the resume decision a fact read off the transcript rather
    /// than a guess: `load_session` is what the agent ACTUALLY advertised at
    /// `initialize`, not what its kind is assumed to support.
    SessionOpened {
        /// The harness's own session handle — what `session/load` resumes.
        acp_session_ref: String,
        /// Whether this agent advertised `loadSession` at `initialize`.
        load_session: bool,
        /// When it opened.
        at: DateTime<Utc>,
        /// Whether this open was a resume of a prior conversation.
        resumed: bool,
    },
    /// The operator's on-screen context, as it stood.
    ///
    /// Appended by every turn and by an explicit context push, so the LATEST one
    /// is the shared context both surfaces read — and the one the harness's
    /// `assistant_context` tool answers with. The transcript IS the shared
    /// context; there is no second store.
    ContextShared {
        /// What was on screen.
        context: AssistantTurnContext,
        /// What put it there: `turn` or `push`.
        source: String,
    },
    /// What the OPERATOR asked, appended the moment the turn was accepted and
    /// before any frame the harness produces for it.
    ///
    /// The record of the question. Without it a reloaded conversation is answers
    /// with no questions, and the transcript — which is the shared context both
    /// surfaces read — could not say what was asked. It carries the context the
    /// turn was sent with, so the request and the screen it was asked from are
    /// ONE record rather than two that could be appended out of order.
    Request {
        /// The turn this frame belongs to.
        turn_id: String,
        /// The operator's own words, without the composed context prefix.
        text: String,
        /// What was on the operator's screen, when they sent any.
        context: Option<AssistantTurnContext>,
        /// The harness command this turn invokes, when it invokes one.
        ///
        /// Recorded beside the text rather than folded into it, because a
        /// transcript that showed only the composed `/compact …` line could not
        /// say whether the operator pressed a command control or typed a line
        /// that happened to begin with a slash.
        command: Option<AssistantCommandInvocation>,
    },
    /// A turn was accepted and the prompt was sent.
    TurnStarted {
        /// The turn this frame belongs to.
        turn_id: String,
        /// When the turn started.
        at: DateTime<Utc>,
        /// The prompt exactly as it was sent, context prefix included.
        ///
        /// Recorded so the transcript answers "what was the agent actually
        /// asked" — a transcript that showed only the operator's typing would
        /// hide the composed context the answer was really shaped by. The
        /// operator's own words are on the [`Self::Request`] frame that precedes
        /// this one; they are not repeated here.
        prompt: String,
    },
    /// The harness advertised the commands it serves.
    ///
    /// A COMPLETE replacement of whatever it advertised before, exactly as ACP's
    /// `available_commands_update` is: an agent that drops a command stops
    /// listing it, and a merge would keep offering a command that would now be
    /// refused.
    AvailableCommands {
        /// Every command the harness serves, in the order it listed them.
        commands: Vec<AssistantCommand>,
    },
    /// An agent text chunk, in order.
    Delta {
        /// The turn this frame belongs to.
        turn_id: String,
        /// The chunk.
        text: String,
    },
    /// An agent thought chunk.
    Thought {
        /// The turn this frame belongs to.
        turn_id: String,
        /// The thought fragment.
        text: String,
    },
    /// A tool call the agent made.
    ToolCall {
        /// The turn this frame belongs to.
        turn_id: String,
        /// The agent's own id for the call, so updates join to one row.
        call_id: String,
        /// The tool's name.
        name: String,
        /// How the call is going.
        status: AssistantToolCallStatus,
        /// The call's input, when the agent reported one.
        #[ts(type = "unknown")]
        input: Option<serde_json::Value>,
        /// The call's output, when the agent reported one.
        #[ts(type = "unknown")]
        output: Option<serde_json::Value>,
    },
    /// A permission request, recorded with the decision the policy made.
    PermissionAsk {
        /// The turn this frame belongs to.
        turn_id: String,
        /// The agent's request, verbatim.
        #[ts(type = "unknown")]
        request: serde_json::Value,
        /// What the configured policy answered.
        decided: AssistantPermissionDecision,
    },
    /// The agent edited the shared document through its
    /// `assistant_document_edit` tool.
    ///
    /// Appended by the assistant MCP route AFTER validating the batch against
    /// the shared document as the transcript then held it, so a recorded batch
    /// always applied cleanly in full. The projection folds it into
    /// `latest_context` — which is how the agent's own next `assistant_context`
    /// read sees its edits — and the console applies the same operations to the
    /// live editor buffer, where the operator keeps or reverts them.
    DocumentEdit {
        /// The operations, in application order.
        edits: Vec<AssistantDocumentEditOp>,
        /// The shared document's revision AFTER this batch — monotonic per
        /// session and never reset, so a client replaying frames after a
        /// reconnect can skip batches it already applied.
        revision: u64,
    },
    /// The turn ended with an answer.
    TurnCompleted {
        /// The turn this frame belongs to.
        turn_id: String,
        /// The agent's last completed message.
        final_message: String,
        /// The canonical stop reason.
        stop_reason: String,
        /// The harness's own session handle, when it reported one.
        session_ref: Option<String>,
    },
    /// The turn ended without an answer.
    TurnFailed {
        /// The turn this frame belongs to.
        turn_id: String,
        /// `auth_required` for a `-32000`, otherwise the typed harness-error
        /// variant name in `snake_case`.
        code: String,
        /// The failure, in the harness's own words.
        message: String,
    },
    /// A frame the neutral vocabulary above cannot represent, passed through
    /// verbatim.
    ///
    /// The adapter's own rule — nothing is ever silently dropped — reaching this
    /// surface. A console renders what it knows and ignores the rest; an
    /// operator reconstructing what an agent actually did still has every frame.
    Raw {
        /// The turn it belongs to, when it belongs to one.
        turn_id: Option<String>,
        /// Where the frame came from, as the adapter labelled it.
        source: String,
        /// The frame, verbatim.
        #[ts(type = "unknown")]
        value: serde_json::Value,
    },
    /// The session's state changed.
    State {
        /// The state it changed to.
        state: AssistantSessionState,
        /// Why — `process_exited`, `deleted`,
        /// `resume_refused: loadSession not advertised`, `load_failed: …`.
        /// `None` when nothing decided it.
        reason: Option<String>,
    },
    /// The session is over; nothing further will arrive on this stream.
    Ended {
        /// Why it ended.
        reason: String,
    },
}

impl AssistantSessionEvent {
    /// The turn this event belongs to, when it belongs to one.
    ///
    /// `State` and `Ended` are session-wide and belong to no turn; saying so
    /// with `None` is what keeps a caller from inventing an attribution.
    #[must_use]
    pub fn turn_id(&self) -> Option<&str> {
        match self {
            Self::Request { turn_id, .. }
            | Self::TurnStarted { turn_id, .. }
            | Self::Delta { turn_id, .. }
            | Self::Thought { turn_id, .. }
            | Self::ToolCall { turn_id, .. }
            | Self::PermissionAsk { turn_id, .. }
            | Self::TurnCompleted { turn_id, .. }
            | Self::TurnFailed { turn_id, .. } => Some(turn_id),
            Self::Raw { turn_id, .. } => turn_id.as_deref(),
            Self::SessionOpened { .. }
            | Self::ContextShared { .. }
            | Self::AvailableCommands { .. }
            | Self::DocumentEdit { .. }
            | Self::State { .. }
            | Self::Ended { .. } => None,
        }
    }

    /// The state and cause this record SETTLES the session to, if it settles it
    /// at all.
    ///
    /// The lifecycle projection is exactly "the last record for which this
    /// returns `Some`, unless a process is running". A record that decides
    /// nothing returns `None`, which is what lets the scan walk backwards and
    /// stop at the first decision it meets.
    #[must_use]
    pub fn settles(&self) -> Option<(AssistantSessionState, Option<String>)> {
        match self {
            Self::State { state, reason } if !state.is_live() => Some((*state, reason.clone())),
            Self::Ended { reason } => Some((AssistantSessionState::Ended, Some(reason.clone()))),
            _ => None,
        }
    }
}

/// One event with the durable index the store assigned it.
///
/// The index is what a reconnecting client passes back as `?after=`, so it is on
/// every frame rather than only on replayed ones: a client that watched live and
/// then dropped its socket must be able to resume from what it last saw without
/// having read the transcript endpoint first.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct AssistantSessionFrame {
    /// The event's dense, store-assigned position in the session's transcript.
    pub index: u64,
    /// The event itself, flattened so the frame reads as one object.
    #[serde(flatten)]
    #[ts(flatten)]
    pub event: AssistantSessionEvent,
}

/// Everything a session's transcript says about it.
///
/// The one place the projection rules live, so the list surface, the detail
/// surface, the resume decision and the `assistant_context` tool all read the
/// same facts off the same records. Nothing here reads a stored status: there
/// is none.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct AssistantSessionProjection {
    /// The last state a record settled the session to, with its cause. `None`
    /// when no record has settled it — which, for a session with no running
    /// process, is a transcript the boot sweep has not yet written back to.
    pub settled: Option<(AssistantSessionState, Option<String>)>,
    /// The harness's own session handle from the most recent open.
    pub acp_session_ref: Option<String>,
    /// Whether the agent advertised `loadSession` at its most recent open.
    /// `false` until an open has been recorded: a session that never opened
    /// cannot be resumed either.
    pub load_session: bool,
    /// How many turns have started.
    pub turns: u64,
    /// What the operator typed on the first turn — the fallback title.
    pub first_turn_text: Option<String>,
    /// The turn a `Request` opened that no `TurnCompleted`/`TurnFailed` has
    /// closed yet. `None` when every asked turn has been answered.
    ///
    /// This is how a settle path knows there is a turn to close: a session
    /// whose process died mid-turn must have that turn FAILED on the record
    /// before the session settles, or every reader of the transcript folds an
    /// open turn forever — the console reads it as busy, and a later edit
    /// batch would be attributed to a turn that ended long ago.
    pub open_turn_id: Option<String>,
    /// The most recently shared on-screen context.
    pub latest_context: Option<AssistantTurnContext>,
    /// The shared document's revision: how many edit batches have folded into
    /// `latest_context`. Monotonic and never reset — a fresh context share
    /// rebases the text but does not rewind the count — so a client can use it
    /// as an idempotency guard when replaying frames.
    pub document_revision: u64,
    /// The commands the harness advertised most recently. A later advertisement
    /// REPLACES an earlier one whole, as ACP's own update does.
    pub commands: Vec<AssistantCommand>,
}

impl AssistantSessionProjection {
    /// Project a whole transcript, oldest event first.
    #[must_use]
    pub fn of<'events>(events: impl IntoIterator<Item = &'events AssistantSessionEvent>) -> Self {
        let mut projection = Self::default();
        for event in events {
            projection.apply(event);
        }
        projection
    }

    /// Fold one event in, in transcript order.
    fn apply(&mut self, event: &AssistantSessionEvent) {
        match event {
            AssistantSessionEvent::SessionOpened {
                acp_session_ref,
                load_session,
                ..
            } => {
                self.acp_session_ref = Some(acp_session_ref.clone());
                self.load_session = *load_session;
                // An open un-settles the session: a process is running again.
                self.settled = None;
            }
            AssistantSessionEvent::ContextShared { context, .. } => {
                self.latest_context = Some(context.clone());
            }
            AssistantSessionEvent::AvailableCommands { commands } => {
                // REPLACED, not merged: the agent's advertisement is complete,
                // and a merge would keep offering a command it has dropped.
                self.commands.clone_from(commands);
            }
            // The REQUEST is what counts a turn, not the start: it is appended
            // at acceptance and carries the operator's own words, so a turn that
            // was accepted and then failed to reach the agent is still a turn
            // that was asked.
            AssistantSessionEvent::Request {
                turn_id,
                text,
                context,
                ..
            } => {
                self.turns = self.turns.saturating_add(1);
                self.open_turn_id = Some(turn_id.clone());
                if self.first_turn_text.is_none() {
                    self.first_turn_text = Some(text.clone());
                }
                if let Some(context) = context {
                    self.latest_context = Some(context.clone());
                }
            }
            AssistantSessionEvent::TurnCompleted { turn_id, .. }
            | AssistantSessionEvent::TurnFailed { turn_id, .. } => {
                if self.open_turn_id.as_deref() == Some(turn_id.as_str()) {
                    self.open_turn_id = None;
                }
                if let Some(settled) = event.settles() {
                    self.settled = Some(settled);
                }
            }
            AssistantSessionEvent::DocumentEdit { edits, revision } => {
                // `max`, not assignment: revisions are minted monotonically at
                // append, so on an in-order transcript this IS assignment — the
                // `max` only guards the fold against a stream a caller hands it
                // out of order, where a rewind would break every client using
                // the revision as an idempotency cursor.
                self.document_revision = self.document_revision.max(*revision);
                if let Some(document) = self
                    .latest_context
                    .as_mut()
                    .and_then(|context| context.document.as_mut())
                {
                    match apply_document_edits(&document.text, edits) {
                        Ok(applied) => document.text = applied,
                        // Reachable, and benign. The append path validated this
                        // batch against the projection AS IT THEN STOOD, but a
                        // `ContextShared` recorded between validation and this
                        // fold can rebase `latest_context` to bytes the batch
                        // no longer matches. The projection stays TOTAL — an
                        // unappliable batch leaves the text as it stands rather
                        // than poisoning the fold — and the very next context
                        // share carries the console's own buffer, edits
                        // included, so the folded document converges on the
                        // operator's truth rather than drifting from it.
                        Err(
                            AssistantDocumentEditError::EmptyOldString { .. }
                            | AssistantDocumentEditError::Absent { .. }
                            | AssistantDocumentEditError::Ambiguous { .. },
                        ) => {}
                    }
                }
            }
            other => {
                if let Some(settled) = other.settles() {
                    self.settled = Some(settled);
                }
            }
        }
    }

    /// The state and cause to report, given whether a process is running.
    ///
    /// A running process ALWAYS wins: it is the one fact a transcript cannot
    /// contradict. With no process, the last settling record decides. With no
    /// process and no settling record — a state the boot sweep exists to
    /// prevent — the answer is [`AssistantSessionState::Ended`] with a cause
    /// that says exactly that, because a session with no process and no record
    /// of stopping must never be presented as running.
    #[must_use]
    pub fn state(
        &self,
        live: Option<AssistantSessionState>,
    ) -> (AssistantSessionState, Option<String>) {
        if let Some(state) = live {
            return (state, None);
        }
        match &self.settled {
            Some((state, cause)) => (*state, cause.clone()),
            None => (
                AssistantSessionState::Ended,
                Some(NO_SETTLING_RECORD.to_owned()),
            ),
        }
    }

    /// Whether a dormant session can be reopened: it has a handle to load and
    /// the agent said it can load one.
    #[must_use]
    pub fn is_resumable(&self) -> bool {
        self.load_session && self.acp_session_ref.is_some()
    }

    /// The title a listing shows: the first 80 characters of the first turn's
    /// own text, when the caller named no title of their own.
    #[must_use]
    pub fn derived_title(&self) -> Option<String> {
        self.first_turn_text.as_ref().map(|text| {
            let trimmed = text.trim();
            trimmed.chars().take(TITLE_CHARACTERS).collect()
        })
    }
}

/// How many characters of the first prompt become a session's title.
pub const TITLE_CHARACTERS: usize = 80;

/// The cause reported for a session with no process and no record of stopping.
///
/// Reachable only if the boot sweep did not run or could not write; naming it
/// keeps that state distinguishable from a session that was genuinely closed.
pub const NO_SETTLING_RECORD: &str =
    "no process is running and the transcript records no stop; treated as ended";

#[cfg(test)]
mod tests {
    use super::*;

    fn instant() -> Result<DateTime<Utc>, String> {
        use chrono::TimeZone;

        Utc.with_ymd_and_hms(2026, 8, 29, 6, 0, 0)
            .single()
            .ok_or_else(|| "the test instant must be valid".to_owned())
    }

    #[test]
    fn a_frame_carries_its_index_beside_the_event_discriminator() -> Result<(), String> {
        let frame = AssistantSessionFrame {
            index: 3,
            event: AssistantSessionEvent::Delta {
                turn_id: "t-1".to_owned(),
                text: "hello".to_owned(),
            },
        };
        let wire = serde_json::to_value(&frame).map_err(|error| error.to_string())?;
        assert_eq!(wire["index"], serde_json::json!(3));
        assert_eq!(wire["type"], serde_json::json!("delta"));
        assert_eq!(wire["turn_id"], serde_json::json!("t-1"));
        assert_eq!(wire["text"], serde_json::json!("hello"));
        Ok(())
    }

    #[test]
    fn every_event_shape_round_trips_through_its_wire_form() -> Result<(), String> {
        let events = vec![
            AssistantSessionEvent::SessionOpened {
                acp_session_ref: "sess-1".to_owned(),
                load_session: true,
                at: instant()?,
                resumed: false,
            },
            AssistantSessionEvent::ContextShared {
                context: AssistantTurnContext::default(),
                source: "turn".to_owned(),
            },
            AssistantSessionEvent::Request {
                turn_id: "t-1".to_owned(),
                text: "fix this".to_owned(),
                context: Some(AssistantTurnContext::default()),
                command: Some(AssistantCommandInvocation {
                    name: "compact".to_owned(),
                    input: Some("keep the plan".to_owned()),
                }),
            },
            AssistantSessionEvent::TurnStarted {
                turn_id: "t-1".to_owned(),
                at: instant()?,
                prompt: "On screen: /studio\n\nfix this".to_owned(),
            },
            AssistantSessionEvent::AvailableCommands {
                commands: vec![AssistantCommand {
                    name: "compact".to_owned(),
                    description: "compact the conversation".to_owned(),
                    input_hint: Some("what to keep".to_owned()),
                }],
            },
            AssistantSessionEvent::DocumentEdit {
                edits: vec![crate::assistant_document::AssistantDocumentEditOp {
                    old_string: "step one".to_owned(),
                    new_string: "step first".to_owned(),
                }],
                revision: 1,
            },
            AssistantSessionEvent::Delta {
                turn_id: "t-1".to_owned(),
                text: "part".to_owned(),
            },
            AssistantSessionEvent::Thought {
                turn_id: "t-1".to_owned(),
                text: "considering".to_owned(),
            },
            AssistantSessionEvent::ToolCall {
                turn_id: "t-1".to_owned(),
                call_id: "c-1".to_owned(),
                name: "check_document".to_owned(),
                status: AssistantToolCallStatus::Completed,
                input: Some(serde_json::json!({ "path": "a.awl" })),
                output: None,
            },
            AssistantSessionEvent::PermissionAsk {
                turn_id: "t-1".to_owned(),
                request: serde_json::json!({ "toolCall": { "toolCallId": "c-1" } }),
                decided: AssistantPermissionDecision::Deny,
            },
            AssistantSessionEvent::TurnCompleted {
                turn_id: "t-1".to_owned(),
                final_message: "done".to_owned(),
                stop_reason: "end_turn".to_owned(),
                session_ref: Some("sess-1".to_owned()),
            },
            AssistantSessionEvent::TurnFailed {
                turn_id: "t-2".to_owned(),
                code: "auth_required".to_owned(),
                message: "the agent requires authentication".to_owned(),
            },
            AssistantSessionEvent::State {
                state: AssistantSessionState::Dormant,
                reason: Some("process_exited".to_owned()),
            },
            AssistantSessionEvent::Ended {
                reason: "the operator closed the session".to_owned(),
            },
        ];
        for event in events {
            let bytes = serde_json::to_vec(&event).map_err(|error| error.to_string())?;
            let decoded: AssistantSessionEvent =
                serde_json::from_slice(&bytes).map_err(|error| error.to_string())?;
            assert_eq!(decoded, event);
        }
        Ok(())
    }

    #[test]
    fn a_session_id_parses_from_its_own_display_form() {
        let id = AssistantSessionId::new(Uuid::from_u128(7));
        assert_eq!(AssistantSessionId::parse(&id.to_string()), Ok(id));
        let error = AssistantSessionId::parse("not-a-uuid");
        assert!(
            matches!(&error, Err(failure) if failure.text == "not-a-uuid"),
            "a malformed id must be refused naming the text: {error:?}"
        );
    }

    #[test]
    fn only_a_running_process_is_live_and_a_dormant_session_is_still_continuable() {
        assert!(AssistantSessionState::Live.is_live());
        assert!(!AssistantSessionState::Dormant.is_live());
        assert!(!AssistantSessionState::Ended.is_live());
        // The distinction the whole resume path rests on: dormant is not
        // running and is not over.
        assert!(AssistantSessionState::Live.is_continuable());
        assert!(AssistantSessionState::Dormant.is_continuable());
        assert!(!AssistantSessionState::Ended.is_continuable());
    }

    #[test]
    fn the_three_wire_states_spell_exactly_what_the_console_parses() {
        // A parser on the other side refuses anything but these three words, so
        // a renamed variant must fail HERE rather than at a running console.
        for (state, spelling) in [
            (AssistantSessionState::Live, "\"live\""),
            (AssistantSessionState::Dormant, "\"dormant\""),
            (AssistantSessionState::Ended, "\"ended\""),
        ] {
            assert_eq!(
                serde_json::to_string(&state).unwrap_or_default(),
                spelling,
                "{state:?} must spell {spelling} on the wire"
            );
        }
    }

    #[test]
    fn session_wide_events_belong_to_no_turn() {
        assert_eq!(
            AssistantSessionEvent::State {
                state: AssistantSessionState::Live,
                reason: None,
            }
            .turn_id(),
            None
        );
        assert_eq!(
            AssistantSessionEvent::Ended {
                reason: "stopped".to_owned(),
            }
            .turn_id(),
            None
        );
        assert_eq!(
            AssistantSessionEvent::Delta {
                turn_id: "t-9".to_owned(),
                text: String::new(),
            }
            .turn_id(),
            Some("t-9")
        );
    }

    fn opened(load_session: bool, resumed: bool) -> Result<AssistantSessionEvent, String> {
        Ok(AssistantSessionEvent::SessionOpened {
            acp_session_ref: "sess-1".to_owned(),
            load_session,
            at: instant()?,
            resumed,
        })
    }

    /// The frame a turn is ACCEPTED with — the record of what was asked, which
    /// is also what the projection counts and titles from.
    fn turn(text: &str) -> AssistantSessionEvent {
        AssistantSessionEvent::Request {
            turn_id: "t-1".to_owned(),
            text: text.to_owned(),
            context: None,
            command: None,
        }
    }

    #[test]
    fn a_running_process_wins_over_every_settling_record() -> Result<(), String> {
        let events = vec![
            opened(true, false)?,
            AssistantSessionEvent::State {
                state: AssistantSessionState::Dormant,
                reason: Some("process_exited".to_owned()),
            },
            opened(true, true)?,
        ];
        let projection = AssistantSessionProjection::of(&events);
        // The second open cleared the settlement: a reopened session is not
        // dormant because it once was.
        assert_eq!(projection.settled, None);
        assert_eq!(
            projection.state(Some(AssistantSessionState::Live)),
            (AssistantSessionState::Live, None)
        );
        Ok(())
    }

    #[test]
    fn with_no_process_the_last_settling_record_decides() -> Result<(), String> {
        let events = vec![
            opened(true, false)?,
            turn("fix the check"),
            AssistantSessionEvent::State {
                state: AssistantSessionState::Dormant,
                reason: Some("process_exited".to_owned()),
            },
        ];
        let projection = AssistantSessionProjection::of(&events);
        assert_eq!(
            projection.state(None),
            (
                AssistantSessionState::Dormant,
                Some("process_exited".to_owned())
            )
        );
        assert!(projection.is_resumable());
        assert_eq!(projection.turns, 1);
        assert_eq!(projection.derived_title(), Some("fix the check".to_owned()));
        Ok(())
    }

    #[test]
    fn an_agent_that_never_advertised_load_session_is_not_resumable() -> Result<(), String> {
        let projection = AssistantSessionProjection::of(&[opened(false, false)?]);
        assert!(
            !projection.is_resumable(),
            "resume is gated on what the agent ACTUALLY advertised"
        );
        Ok(())
    }

    #[test]
    fn no_process_and_no_settling_record_reads_ended_not_running() -> Result<(), String> {
        let projection = AssistantSessionProjection::of(&[opened(true, false)?, turn("hello")]);
        let (state, cause) = projection.state(None);
        assert_eq!(state, AssistantSessionState::Ended);
        assert_eq!(cause.as_deref(), Some(NO_SETTLING_RECORD));
        Ok(())
    }

    #[test]
    fn the_latest_shared_context_is_the_one_the_tool_answers_with() {
        let first = AssistantTurnContext {
            url: Some("/studio".to_owned()),
            concepts: vec!["awl.step".to_owned()],
            document: None,
        };
        let second = AssistantTurnContext {
            url: Some("/runs".to_owned()),
            concepts: Vec::new(),
            document: None,
        };
        let events = vec![
            AssistantSessionEvent::ContextShared {
                context: first,
                source: "turn".to_owned(),
            },
            AssistantSessionEvent::ContextShared {
                context: second.clone(),
                source: "push".to_owned(),
            },
        ];
        assert_eq!(
            AssistantSessionProjection::of(&events).latest_context,
            Some(second)
        );
    }

    /// The console parses `request { turn_id, text, context }` by those exact
    /// names (`wire.ts:161`). A renamed field is a frame it drops, and a
    /// transcript with no request block claims nothing rather than being wrong —
    /// which is precisely why the drift has to fail HERE.
    #[test]
    fn a_request_frame_spells_what_the_console_parses() -> Result<(), String> {
        let wire = serde_json::to_value(AssistantSessionEvent::Request {
            turn_id: "t-7".to_owned(),
            text: "fix the check".to_owned(),
            context: Some(AssistantTurnContext {
                url: Some("/studio".to_owned()),
                concepts: vec!["awl.step".to_owned()],
                document: None,
            }),
            command: None,
        })
        .map_err(|error| error.to_string())?;
        assert_eq!(wire["type"], serde_json::json!("request"));
        assert_eq!(wire["turn_id"], serde_json::json!("t-7"));
        assert_eq!(wire["text"], serde_json::json!("fix the check"));
        assert_eq!(wire["context"]["url"], serde_json::json!("/studio"));
        Ok(())
    }

    /// A later advertisement REPLACES an earlier one whole. A merge would keep
    /// offering a command the agent has dropped, and the turn invoking it would
    /// be refused by the agent instead of by the surface that offered it.
    #[test]
    fn a_later_command_advertisement_replaces_the_earlier_one_whole() {
        let command = |name: &str| AssistantCommand {
            name: name.to_owned(),
            description: format!("{name} does something"),
            input_hint: None,
        };
        let events = vec![
            AssistantSessionEvent::AvailableCommands {
                commands: vec![command("compact"), command("plan")],
            },
            AssistantSessionEvent::AvailableCommands {
                commands: vec![command("compact")],
            },
        ];
        let projection = AssistantSessionProjection::of(&events);
        assert_eq!(
            projection
                .commands
                .iter()
                .map(|entry| entry.name.as_str())
                .collect::<Vec<_>>(),
            vec!["compact"],
            "`plan` was withdrawn and must not survive as an offer"
        );
    }

    /// A session nobody has advertised commands on offers none — the negative
    /// control for the cell above, which an always-populated list would pass.
    #[test]
    fn a_session_with_no_advertisement_offers_no_commands() {
        assert!(
            AssistantSessionProjection::default().commands.is_empty(),
            "a command list must come from the agent, never from this server"
        );
    }

    /// ACP delivers a command as ordinary prompt text: `/name` and the input
    /// after it. One spelling, here, so the turn path and the wire pins cannot
    /// disagree about the form the agent actually receives.
    #[test]
    fn a_command_is_delivered_as_the_slash_line_the_acp_spec_describes() {
        assert_eq!(
            AssistantCommandInvocation {
                name: "compact".to_owned(),
                input: Some("keep the plan".to_owned()),
            }
            .prompt_line(),
            "/compact keep the plan"
        );
        assert_eq!(
            AssistantCommandInvocation {
                name: "compact".to_owned(),
                input: None,
            }
            .prompt_line(),
            "/compact"
        );
        // Whitespace-only input is no input: a trailing space on the wire must
        // not become a trailing space in what the agent parses.
        assert_eq!(
            AssistantCommandInvocation {
                name: "compact".to_owned(),
                input: Some("   ".to_owned()),
            }
            .prompt_line(),
            "/compact"
        );
    }

    /// A turn's context reaches the shared-context projection through the
    /// REQUEST frame, so a turn asked from a screen updates what the
    /// `assistant_context` tool answers with without a second record.
    #[test]
    fn a_turns_own_context_becomes_the_shared_context() {
        let context = AssistantTurnContext {
            url: Some("/studio/a.awl".to_owned()),
            concepts: Vec::new(),
            document: None,
        };
        let projection = AssistantSessionProjection::of(&[AssistantSessionEvent::Request {
            turn_id: "t-1".to_owned(),
            text: "explain".to_owned(),
            context: Some(context.clone()),
            command: None,
        }]);
        assert_eq!(projection.latest_context, Some(context));
        assert_eq!(projection.turns, 1);
    }

    #[test]
    fn a_title_is_bounded_at_eighty_characters() -> Result<(), String> {
        let long = "x".repeat(200);
        let projection = AssistantSessionProjection::of(&[turn(&long)]);
        let title = projection
            .derived_title()
            .ok_or_else(|| "a started turn has a derivable title".to_owned())?;
        assert_eq!(title.chars().count(), TITLE_CHARACTERS);
        Ok(())
    }

    /// A `Request` opens a turn and only ITS `TurnCompleted`/`TurnFailed`
    /// closes it — this is what a settle path reads to know a dying process
    /// leaves a question on the record that will never be answered.
    #[test]
    fn a_request_opens_a_turn_and_its_ending_closes_it() {
        let mut projection = AssistantSessionProjection::of(&[turn("first ask")]);
        assert_eq!(projection.open_turn_id.as_deref(), Some("t-1"));
        // An ending for a DIFFERENT turn does not close this one.
        projection.apply(&AssistantSessionEvent::TurnFailed {
            turn_id: "t-0".to_owned(),
            code: "stale".to_owned(),
            message: "an earlier turn's ending arrives late".to_owned(),
        });
        assert_eq!(projection.open_turn_id.as_deref(), Some("t-1"));
        projection.apply(&AssistantSessionEvent::TurnCompleted {
            turn_id: "t-1".to_owned(),
            final_message: "answered".to_owned(),
            stop_reason: "end_turn".to_owned(),
            session_ref: None,
        });
        assert_eq!(projection.open_turn_id, None);
    }

    /// The projection folds a recorded edit batch into the shared document, so
    /// the agent's next `assistant_context` read sees its own edits — the
    /// transcript IS the shared document, with no second store to catch up.
    #[test]
    fn a_document_edit_folds_into_the_latest_context() {
        let events = vec![
            AssistantSessionEvent::ContextShared {
                context: AssistantTurnContext {
                    url: Some("/studio/pipeline.awl".to_owned()),
                    concepts: Vec::new(),
                    document: Some(AssistantDocumentContext {
                        path: "pipeline.awl".to_owned(),
                        text: "workflow demo\nstep one\n".to_owned(),
                        selection: None,
                        cursor: None,
                    }),
                },
                source: "turn".to_owned(),
            },
            AssistantSessionEvent::DocumentEdit {
                edits: vec![crate::assistant_document::AssistantDocumentEditOp {
                    old_string: "step one".to_owned(),
                    new_string: "step first".to_owned(),
                }],
                revision: 1,
            },
        ];
        let projection = AssistantSessionProjection::of(&events);
        assert_eq!(projection.document_revision, 1);
        let text = projection
            .latest_context
            .and_then(|context| context.document)
            .map(|document| document.text);
        assert_eq!(text, Some("workflow demo\nstep first\n".to_owned()));
    }

    /// An edit recorded with no shared document still advances the revision —
    /// the batch was appended, and a client's idempotency guard counts appended
    /// batches, not applicable ones — and the fold stays total rather than
    /// failing over a document that is not there.
    #[test]
    fn a_document_edit_with_no_shared_document_still_advances_the_revision() {
        let events = vec![AssistantSessionEvent::DocumentEdit {
            edits: vec![crate::assistant_document::AssistantDocumentEditOp {
                old_string: "anything".to_owned(),
                new_string: "else".to_owned(),
            }],
            revision: 1,
        }];
        let projection = AssistantSessionProjection::of(&events);
        assert_eq!(projection.document_revision, 1);
        assert_eq!(projection.latest_context, None);
    }

    /// The console parses `document_edit { edits, revision }` by those exact
    /// names. A renamed field is a frame the editor never applies, so the drift
    /// has to fail HERE.
    #[test]
    fn a_document_edit_frame_spells_what_the_console_parses() -> Result<(), String> {
        let wire = serde_json::to_value(AssistantSessionEvent::DocumentEdit {
            edits: vec![crate::assistant_document::AssistantDocumentEditOp {
                old_string: "a".to_owned(),
                new_string: "b".to_owned(),
            }],
            revision: 4,
        })
        .map_err(|error| error.to_string())?;
        assert_eq!(wire["type"], serde_json::json!("document_edit"));
        assert_eq!(wire["revision"], serde_json::json!(4));
        assert_eq!(wire["edits"][0]["old_string"], serde_json::json!("a"));
        assert_eq!(wire["edits"][0]["new_string"], serde_json::json!("b"));
        Ok(())
    }
}