aion_server/assistant/sessions/

lifecycle.rs

//! Creating, turning, resuming, cancelling and ending a session.
//!
//! # A session is created, not spawned
//!
//! ACP has no way to open a conversation without asking something: `session/new`
//! is followed by `session/prompt` in one opening, and there is no promptless
//! handshake to run at create time. So creating a session RECORDS it — and the
//! FIRST TURN is what starts a process, carrying the operator's own words as the
//! opening prompt.
//!
//! That is also why "not logged in" is a turn's `auth_required` code and never a
//! session state: the `-32000` an unauthenticated agent answers with arrives on
//! a prompt, so it belongs to the turn that asked.
//!
//! # Resume is gated on what the agent ACTUALLY said
//!
//! A session whose process is gone is `dormant` only when its last opening
//! recorded `load_session: true` — what that agent advertised at `initialize`,
//! not what agents of its kind usually support. When it did not, the session is
//! settled `ended` by an appended record naming the capability, and the next
//! turn is refused rather than spawning a process that would be asked for a
//! `session/load` it cannot answer.

use std::sync::Arc;

use aion_core::{
    AssistantCommandInvocation, AssistantConfigOption, AssistantConfigValue, AssistantSessionEvent,
    AssistantSessionFrame, AssistantSessionId, AssistantSessionProjection, AssistantSessionState,
    AssistantSessionSummary, AssistantTurnContext, ContentType, Payload,
};
use aion_integration_acp::catalogue::{self, CatalogueHarness};
use aion_integration_acp::{AcpConfigValue, TurnHandle};
use aion_integrations::{ActivityId, AgentHarness, AgentRunSpec, RunId, WorkflowId};
use aion_store::assistant::AssistantSessionRecord;
use chrono::Utc;
use tokio::sync::broadcast;

use crate::config::ResolvedAssistantAccount;

use super::error::AssistantSessionError;
use super::frames;
use super::launch;
use super::live::LiveSession;
use super::prompt;
use super::registry::{AssistantSessions, Availability};
use super::turn_driver;

use crate::assistant::grounding;

impl AssistantSessions {
    /// Open a session on a catalogue harness.
    ///
    /// Records the session; the harness starts on the first turn. The returned
    /// summary therefore reports `dormant` — no process, and continuable — which
    /// is the truthful answer for a conversation nobody has said anything in
    /// yet.
    ///
    /// The pick is REMEMBERED, caller-scoped, so the next new-conversation form
    /// opens on the harness this operator last used. It is written from what
    /// they actually opened rather than from a preference they set, and it is
    /// written only after the session record exists — a remembered pick for a
    /// session that was refused would be a choice nobody made.
    ///
    /// # Errors
    ///
    /// [`AssistantSessionError::NotCommissioned`] when the store is unusable,
    /// [`AssistantSessionError::UnknownHarness`] / `UnknownAccount` when the
    /// caller names one that does not exist,
    /// [`AssistantSessionError::HarnessUnavailable`] when the chosen harness's
    /// launch program is not on this server's `PATH` — refused HERE rather than
    /// accepted and failed at the first message — or whatever the store reports.
    pub async fn create(
        &self,
        subject: &str,
        harness: Option<&str>,
        account: Option<&str>,
        title: Option<String>,
    ) -> Result<AssistantSessionSummary, AssistantSessionError> {
        if let Availability::Unavailable { reason } = self.availability() {
            tracing::warn!(%reason, "an assistant session was requested on a server with none");
            return Err(AssistantSessionError::NotCommissioned { reason });
        }
        let harness = self.resolve_harness(subject, harness).await?;
        let account = self.resolve_account(harness.id, account)?;
        let session_id = AssistantSessionId::new_v4();
        // The plan is built here to MEASURE the harness: an unavailable one is
        // refused with the same typed shape the spawn would produce, so a
        // session is never accepted on a harness this machine cannot run. Its
        // token is DISCARDED — `mint()` is random per call, so a digest stored
        // now would name a secret no child will ever hold. The ONE mint that
        // counts happens at spawn, which stores the digest of the secret it
        // actually hands the child; until then the record carries no digest and
        // the MCP route refuses every bearer with `NoTokenMinted`, which is the
        // truth of a session that has no process.
        let plan = launch::plan(session_id, harness, account, self.aion_endpoint())?;
        drop(plan);
        let now = Utc::now();
        let record = AssistantSessionRecord {
            session_id,
            subject: subject.to_owned(),
            harness: harness.id.to_owned(),
            account: account.map(|account| account.name.clone()),
            title,
            created_at: now,
            updated_at: now,
            turns: 0,
            mcp_token_digest: None,
            // Nothing is offered until the harness itself advertises something:
            // a command list this server composed would be a list the agent
            // never promised to serve.
            commands: Vec::new(),
            config_options: Vec::new(),
        };
        self.store().put_assistant_session(record.clone()).await?;
        // Written after the record and before the summary: the memory is of a
        // session that exists.
        self.store()
            .put_assistant_default_harness(subject, harness.id)
            .await?;
        // The session's FIRST record says what it is: no process, and waiting
        // for something to be said. Without it the projection would meet a
        // transcript with no settling record and report the fallback.
        self.settle(
            session_id,
            AssistantSessionState::Dormant,
            CREATED_AWAITING_FIRST_TURN,
        )
        .await?;
        Ok(record.summary(
            AssistantSessionState::Dormant,
            Some(CREATED_AWAITING_FIRST_TURN.to_owned()),
        ))
    }

    /// The caller's CURRENT session: the newest one that is not ended.
    ///
    /// One thread is the shape an operator actually holds — the dock panel and
    /// the inline editor bar are two views of ONE conversation, not two
    /// conversations — so "which session am I in" is a first-class read rather
    /// than something each surface derives from a list and could derive
    /// differently.
    ///
    /// `Ok(None)` when the caller has no continuable session. That is an
    /// absence, not a refusal: a caller with no session is the ordinary state
    /// of somebody who has not started one.
    ///
    /// # Errors
    ///
    /// Whatever the store reports.
    pub async fn current(
        &self,
        subject: &str,
    ) -> Result<Option<AssistantSessionSummary>, AssistantSessionError> {
        // `list` is already newest-first and already projects each state, so the
        // current session is the first row that may be current — continuable
        // and not put away by the caller. Deriving it here rather
        // than in each surface means two surfaces cannot disagree about which
        // conversation the operator is in.
        Ok(self
            .list(subject)
            .await?
            .into_iter()
            .find(|summary| summary.state.is_current_candidate()))
    }

    /// A session's summary and its whole transcript.
    ///
    /// # Errors
    ///
    /// Not found (including another subject's session), or whatever the store
    /// reports.
    pub async fn read(
        &self,
        subject: &str,
        session_id: AssistantSessionId,
    ) -> Result<(AssistantSessionSummary, Vec<AssistantSessionFrame>), AssistantSessionError> {
        let record = self.owned_record(subject, session_id).await?;
        let frames = self.transcript_from(session_id, None).await?;
        let live = self.is_live(session_id).await;
        let projection = AssistantSessionProjection::of(frames.iter().map(|frame| &frame.event));
        let (state, reason) = projection.state(live.then_some(AssistantSessionState::Live));
        Ok((record.summary(state, reason), frames))
    }

    /// The transcript so far, then every frame as it arrives.
    ///
    /// The replay is read BEFORE the subscription is taken so nothing can land
    /// between the two: a frame committed after the read and before the
    /// subscribe would be in neither, and a client would have a hole it could
    /// not see. Taking the receiver first means the worst case is a DUPLICATE
    /// frame, which a client can drop by index.
    ///
    /// # Errors
    ///
    /// Not found, or whatever the store reports.
    pub async fn watch(
        &self,
        subject: &str,
        session_id: AssistantSessionId,
        after: Option<u64>,
    ) -> Result<
        (
            Vec<AssistantSessionFrame>,
            broadcast::Receiver<AssistantSessionFrame>,
        ),
        AssistantSessionError,
    > {
        let _record = self.owned_record(subject, session_id).await?;
        let receiver = self.recorder(session_id).subscribe();
        let replay = self.transcript_from(session_id, after).await?;
        Ok((replay, receiver))
    }

    /// Share the operator's on-screen context without asking anything.
    ///
    /// Appended to the transcript, which IS the shared context: the harness's
    /// `assistant_context` tool reads the latest one, and both console surfaces
    /// read the same record. There is no second store to keep in step.
    ///
    /// # Errors
    ///
    /// Not found, or whatever the store reports.
    pub async fn push_context(
        &self,
        subject: &str,
        session_id: AssistantSessionId,
        context: AssistantTurnContext,
    ) -> Result<(), AssistantSessionError> {
        let _record = self.owned_record(subject, session_id).await?;
        self.recorder(session_id)
            .record(AssistantSessionEvent::ContextShared {
                context,
                source: CONTEXT_SOURCE_PUSH.to_owned(),
            })
            .await
            .map(drop)
    }

    /// Ask the agent something.
    ///
    /// Starts the harness when none is running — freshly for a session that has
    /// never opened, or with `session/load` for a dormant one whose agent can
    /// reload its own conversation.
    ///
    /// # Errors
    ///
    /// Not found; [`AssistantSessionError::Busy`] when a turn is already open;
    /// [`AssistantSessionError::Ended`] when the session cannot be continued;
    /// [`AssistantSessionError::HarnessFailed`] when the harness will not start.
    pub async fn turn(
        &self,
        subject: &str,
        session_id: AssistantSessionId,
        text: String,
        context: Option<AssistantTurnContext>,
        command: Option<AssistantCommandInvocation>,
    ) -> Result<String, AssistantSessionError> {
        let record = self.owned_record(subject, session_id).await?;
        if let Some(command) = command.as_ref() {
            self.require_advertised(&record, command).await?;
        }
        // A command is DELIVERED as prompt text — `/name` and whatever follows
        // it — because that is the delivery form the ACP schema describes: its
        // only input shape is "all text that was typed after the command name".
        // There is no second transport to take, so the composition is the whole
        // of it, and the structured invocation is recorded beside the line so
        // the transcript still says a command was pressed.
        let asked = command
            .as_ref()
            .map_or_else(|| text.clone(), AssistantCommandInvocation::prompt_line);
        let composed = prompt::compose(context.as_ref(), &asked);
        // The FIRST turn of a session opens with the grounding preamble: where
        // the agent is standing and where this server's reference pack lives.
        // Server-side grounding, not operator context — it goes around the
        // composed prompt rather than through the turn-context formatter,
        // whose byte-for-byte mirror contract with the console is about what
        // the operator's screen said and nothing else. Once per session: on
        // every turn it would be noise the agent learns to read past, and a
        // resumed conversation already carries it in the history the agent
        // loads. The spawn (which a first turn always performs) rewrites the
        // pack itself before the child starts.
        let composed = if record.turns == 0 {
            let dir = grounding::directory().map_err(AssistantSessionError::Internal)?;
            format!("{}\n\n{composed}", grounding::preamble(&dir))
        } else {
            composed
        };
        let turn_id = uuid::Uuid::new_v4().to_string();

        let (live, opening) = match self.ensure_live(&record, &composed).await {
            Ok(started) => started,
            Err(error) => {
                // The spawn is where a missing binary, an absent account
                // variable or a refused resume is actually met, and the operator
                // is watching the SOCKET. So the refusal is recorded as this
                // turn's own failure — carrying the typed code and the message
                // with the install hint in it — before it is returned. A turn
                // that never reached an agent still has to say so on the
                // transcript, or a reloaded conversation shows a question with
                // no answer and no reason.
                self.record_turn_failure(session_id, &turn_id, &error).await;
                return Err(error);
            }
        };
        // Claimed AFTER the spawn: a fresh spawn's opening prompt IS this turn,
        // so the slot has to be claimed against the session that now exists.
        live.claim_turn()?;
        let started = self
            .record_turn_start(&live, &turn_id, context, &asked, command, &composed)
            .await;
        if let Err(error) = started {
            live.release_turn();
            return Err(error);
        }

        // The spawn's own prompt IS this turn, so a fresh spawn already handed
        // back the handle; only a session that was already live has to submit
        // one.
        // A fresh spawn's opening prompt IS this turn, so it already handed back
        // the handle; only a session that was already live has to submit one.
        let handle = if let Some(handle) = opening {
            handle
        } else {
            let submitted = live
                .with_session(async |session| session.prompt(composed.clone()).await)
                .await;
            match submitted {
                Some(Ok(handle)) => handle,
                Some(Err(error)) => {
                    live.release_turn();
                    return Err(AssistantSessionError::HarnessFailed {
                        harness: record.harness.clone(),
                        reason: error.to_string(),
                    });
                }
                None => {
                    live.release_turn();
                    return Err(AssistantSessionError::Ended {
                        session_id,
                        reason: "the harness process has been shut down".to_owned(),
                    });
                }
            }
        };

        self.touch(session_id, Some(&asked)).await?;
        // No per-turn timeout: a turn ends when the harness ends it or the
        // caller cancels it (RULED 2026-08-29). A number here would cancel
        // somebody's real work mid-thought, and there is no honest value for
        // "how long may an agent think".
        let driver_turn = turn_id.clone();
        tokio::spawn(turn_driver::drive(live, handle, driver_turn));
        Ok(turn_id)
    }

    /// Record a refusal that stopped a turn before it reached the agent.
    ///
    /// Best-effort by construction: the caller is already returning the typed
    /// error, and a store that cannot take this frame is a store that could not
    /// have taken the turn either. It is logged rather than swallowed, and it
    /// never replaces the refusal the caller gets.
    async fn record_turn_failure(
        &self,
        session_id: AssistantSessionId,
        turn_id: &str,
        error: &AssistantSessionError,
    ) {
        if let Err(recording) = self
            .recorder(session_id)
            .record(AssistantSessionEvent::TurnFailed {
                turn_id: turn_id.to_owned(),
                code: error.code().to_owned(),
                message: error.to_string(),
            })
            .await
        {
            tracing::warn!(
                session = %session_id,
                %recording,
                original = %error,
                "an assistant turn failed before it reached the agent, and the failure could not \
                 be recorded on its transcript"
            );
        }
    }

    /// Stop the open turn.
    ///
    /// # Errors
    ///
    /// Not found, or [`AssistantSessionError::Ended`] when no process is
    /// running to cancel.
    pub async fn cancel(
        &self,
        subject: &str,
        session_id: AssistantSessionId,
    ) -> Result<(), AssistantSessionError> {
        let _record = self.owned_record(subject, session_id).await?;
        let Some(live) = self.live(session_id) else {
            return Err(AssistantSessionError::Ended {
                session_id,
                reason: "no harness process is running for this session".to_owned(),
            });
        };
        turn_driver::cancel(&live).await
    }

    /// Set one advertised configuration option on a live session — the model
    /// picker's write side.
    ///
    /// Refused BY NAME when the harness has not advertised the option, or (for
    /// a select) the value: a client offers what the session's
    /// `config_options` say, and anything else is a control that should never
    /// have been on the screen — the exact rule turns follow for commands. The
    /// agent's answer, the full option set as it now stands, is recorded on
    /// the transcript exactly as an advertisement is (which also refreshes the
    /// record's cache), and returned.
    ///
    /// # Errors
    ///
    /// Not found / not yours; [`AssistantSessionError::UnknownConfigOption`]
    /// for an unadvertised option or value, or a value of the wrong shape;
    /// [`AssistantSessionError::Ended`] when no process is running;
    /// [`AssistantSessionError::HarnessFailed`] when the agent refuses, the
    /// transport fails, or the answer cannot be read as an option set.
    pub async fn set_config_option(
        &self,
        subject: &str,
        session_id: AssistantSessionId,
        option_id: &str,
        value: &serde_json::Value,
    ) -> Result<Vec<AssistantConfigOption>, AssistantSessionError> {
        let record = self.owned_record(subject, session_id).await?;
        let chosen = chosen_config_value(&record.config_options, session_id, option_id, value)?;
        let Some(live) = self.live(session_id) else {
            return Err(AssistantSessionError::Ended {
                session_id,
                reason: "no harness process is running for this session".to_owned(),
            });
        };
        let delivered = live
            .with_session(async |session| session.set_config_option(option_id, chosen).await)
            .await;
        let answer = match delivered {
            None => {
                return Err(AssistantSessionError::Ended {
                    session_id,
                    reason: "the harness process has already been shut down".to_owned(),
                });
            }
            Some(Err(error)) => {
                return Err(AssistantSessionError::HarnessFailed {
                    harness: record.harness.clone(),
                    reason: error.to_string(),
                });
            }
            Some(Ok(answer)) => answer,
        };
        let Some(options) = frames::config_options(&answer) else {
            // The change may have applied — the agent answered — but the
            // answer cannot be read as an option set. Recorded verbatim so
            // the record is complete, and reported loudly rather than
            // returned as an empty set (an empty set is a withdrawal).
            live.recorder()
                .record(AssistantSessionEvent::Raw {
                    turn_id: None,
                    source: "session/set_config_option/response".to_owned(),
                    value: answer,
                })
                .await?;
            return Err(AssistantSessionError::HarnessFailed {
                harness: record.harness.clone(),
                reason: "the agent answered `session/set_config_option` with an unreadable \
                         option set; the transcript holds the answer verbatim"
                    .to_owned(),
            });
        };
        live.recorder()
            .record(AssistantSessionEvent::ConfigOptions {
                options: options.clone(),
            })
            .await?;
        Ok(options)
    }

    /// Say whether a session can be continued, settling it when it cannot.
    ///
    /// There is no promptless spawn to perform here — ACP opens a conversation
    /// by asking something — so this does not start a process. What it does is
    /// make the resume decision READABLE before an operator types: a session
    /// whose agent never advertised `loadSession` is settled `ended` by an
    /// appended record naming the capability, so the panel can say so instead of
    /// offering a box that would refuse.
    ///
    /// # Errors
    ///
    /// Not found, or whatever the store reports.
    pub async fn resume(
        &self,
        subject: &str,
        session_id: AssistantSessionId,
    ) -> Result<AssistantSessionSummary, AssistantSessionError> {
        let record = self.owned_record(subject, session_id).await?;
        if self.is_live(session_id).await {
            return Ok(record.summary(AssistantSessionState::Live, None));
        }
        let projection = self.projection(session_id).await?;
        if projection.acp_session_ref.is_none() || projection.is_resumable() {
            let (state, reason) = projection.state(None);
            return Ok(record.summary(state, reason));
        }
        self.settle(session_id, AssistantSessionState::Ended, RESUME_REFUSED)
            .await?;
        Ok(record.summary(
            AssistantSessionState::Ended,
            Some(RESUME_REFUSED.to_owned()),
        ))
    }

    /// Put a session away: shut its harness down and settle it.
    ///
    /// The transcript is KEPT. Deleting a session deletes a process, not a
    /// record — an operator reading back what an agent did a week ago is
    /// exactly who this surface exists for.
    ///
    /// What it settles to follows the same fact every resume decision follows:
    /// whether the agent can reload the conversation. A session whose agent
    /// advertised `loadSession` and left a handle settles
    /// [`AssistantSessionState::Closed`] — out of the operator's way, never
    /// current, but reopened by the next turn taken on it from history. One
    /// whose agent cannot reload settles [`AssistantSessionState::Ended`], and
    /// the transcript is all that is left of it.
    ///
    /// # Errors
    ///
    /// Not found, or whatever the store reports.
    pub async fn delete(
        &self,
        subject: &str,
        session_id: AssistantSessionId,
    ) -> Result<(), AssistantSessionError> {
        let _record = self.owned_record(subject, session_id).await?;
        if let Some(live) = self.forget(session_id) {
            live.close().await;
        }
        // The per-session edit lock dies with the process. A call still holding
        // its `Arc` finishes safely; nothing new is minted until a turn spawns
        // a fresh child, and a fresh child is handed a fresh bearer.
        self.document_edit_locks().remove(&session_id);
        self.fail_open_turn(session_id, "process_exited").await?;
        let projection = self.projection(session_id).await?;
        if projection.is_resumable() {
            return self
                .settle(session_id, AssistantSessionState::Closed, CLOSED_RESUMABLE)
                .await;
        }
        self.settle(session_id, AssistantSessionState::Ended, DELETED)
            .await?;
        self.recorder(session_id)
            .record(AssistantSessionEvent::Ended {
                reason: DELETED.to_owned(),
            })
            .await
            .map(drop)
    }

    /// Settle every session whose process is gone, at boot.
    ///
    /// WRITTEN BACK, not merely displayed: the settlement is an appended record
    /// with its cause, so the next reader projects it rather than recomputing
    /// the same decision — and so the decision itself is auditable.
    ///
    /// Returns how many sessions were settled.
    ///
    /// # Errors
    ///
    /// Whatever the store reports.
    pub async fn sweep_orphans(&self) -> Result<usize, AssistantSessionError> {
        match self.sweep_orphans_inner().await {
            Ok(settled) => {
                // The sweep is this server's first end-to-end use of the
                // assistant store, so it is where "can this server hold a
                // conversation at all" is actually answered — and answering it
                // is what puts a real sentence in `sessions_disabled_reason`
                // instead of leaving the field for a configuration question that
                // no longer exists.
                self.clear_store_fault();
                Ok(settled)
            }
            Err(error) => {
                self.report_store_fault(&error);
                Err(error)
            }
        }
    }

    /// Fail the turn a dying process leaves open, BEFORE the session settles.
    ///
    /// A `Request` is appended at acceptance; its answer arrives only from the
    /// process. When the process is gone, no `TurnCompleted`/`TurnFailed` will
    /// ever come, and a transcript that settles around an open turn folds as
    /// busy forever — the console refuses new turns, and a later edit batch
    /// would be attributed to a question that ended with the process. So every
    /// settle path for a dead process closes the record's open question first,
    /// with a `TurnFailed` that says exactly what happened.
    async fn fail_open_turn(
        &self,
        session_id: AssistantSessionId,
        code: &str,
    ) -> Result<(), AssistantSessionError> {
        let projection = self.projection(session_id).await?;
        let Some(turn_id) = projection.open_turn_id else {
            return Ok(());
        };
        self.recorder(session_id)
            .record(AssistantSessionEvent::TurnFailed {
                turn_id,
                code: code.to_owned(),
                message: OPEN_TURN_ORPHANED.to_owned(),
            })
            .await
            .map(drop)
    }

    /// The sweep itself; [`Self::sweep_orphans`] records what it found out about
    /// the store.
    async fn sweep_orphans_inner(&self) -> Result<usize, AssistantSessionError> {
        let listing = self.store().list_assistant_sessions().await?;
        let mut settled = 0_usize;
        for record in listing.sessions {
            let session_id = record.session_id;
            if self.is_live(session_id).await {
                continue;
            }
            let projection = self.projection(session_id).await?;
            if projection.settled.is_some() {
                continue;
            }
            let (state, reason) = if projection.is_resumable() {
                (
                    AssistantSessionState::Dormant,
                    PROCESS_EXITED_RESUMABLE.to_owned(),
                )
            } else {
                (
                    AssistantSessionState::Ended,
                    PROCESS_EXITED_ENDED.to_owned(),
                )
            };
            self.fail_open_turn(session_id, "process_exited").await?;
            self.settle(session_id, state, reason).await?;
            settled = settled.saturating_add(1);
        }
        if settled > 0 {
            tracing::info!(
                settled,
                "assistant sessions whose harness process is gone were settled at boot"
            );
        }
        Ok(settled)
    }

    /// Shut every live harness down — the server is stopping.
    pub async fn shutdown(&self) {
        let ids: Vec<AssistantSessionId> = self.live_ids();
        for session_id in ids {
            if let Some(live) = self.forget(session_id) {
                live.close().await;
            }
            if let Err(error) = self.fail_open_turn(session_id, "server_stopped").await {
                tracing::warn!(
                    session = %session_id,
                    %error,
                    "an assistant session's open turn could not be failed while the server \
                     stopped; the transcript keeps an unanswered question until the boot sweep"
                );
            }
            if let Err(error) = self
                .settle(session_id, AssistantSessionState::Dormant, SERVER_STOPPED)
                .await
            {
                tracing::warn!(
                    session = %session_id,
                    %error,
                    "an assistant session could not be settled while the server stopped; the boot \
                     sweep will settle it on the next start"
                );
            }
        }
    }

    /// Start a harness for this session if none is running, returning it and —
    /// for a fresh spawn — the opening turn's handle.
    async fn ensure_live(
        &self,
        record: &AssistantSessionRecord,
        composed_prompt: &str,
    ) -> Result<(Arc<LiveSession>, Option<TurnHandle>), AssistantSessionError> {
        let session_id = record.session_id;
        // Serialised per session: two turns arriving at once on a session with
        // no process must not both spawn one.
        let spawn_lock = self.spawn_lock(session_id);
        let _held = spawn_lock.lock().await;
        if let Some(live) = self.live(session_id)
            && live.is_alive().await
        {
            return Ok((live, None));
        }
        let projection = self.projection(session_id).await?;
        // Ended is terminal, whichever door a turn arrives through. The resume
        // path below refuses by name when the agent cannot reload; this refusal
        // covers the OTHER door — a session that never opened (or whose ended
        // record predates any open) would otherwise fall through to a fresh
        // spawn and come back `live` under the id the caller deleted.
        if matches!(projection.settled, Some((AssistantSessionState::Ended, _))) {
            tracing::warn!(
                %session_id,
                harness = %record.harness,
                "a turn on an ended assistant session was refused: ended is terminal"
            );
            return Err(AssistantSessionError::Ended {
                session_id,
                reason: ENDED_IS_TERMINAL.to_owned(),
            });
        }
        let prior = if projection.acp_session_ref.is_none() {
            None
        } else if projection.is_resumable() {
            projection.acp_session_ref.clone()
        } else {
            self.settle(session_id, AssistantSessionState::Ended, RESUME_REFUSED)
                .await?;
            return Err(AssistantSessionError::Ended {
                session_id,
                reason: RESUME_REFUSED.to_owned(),
            });
        };
        let (live, handle) = self.spawn(record, composed_prompt, prior).await?;
        Ok((live, Some(handle)))
    }

    /// Spawn the harness and adopt it.
    async fn spawn(
        &self,
        record: &AssistantSessionRecord,
        composed_prompt: &str,
        prior: Option<String>,
    ) -> Result<(Arc<LiveSession>, TurnHandle), AssistantSessionError> {
        let session_id = record.session_id;
        // The RECORD's harness, never the caller's last pick: a conversation
        // runs on the harness it was opened on, whatever the operator has picked
        // since.
        let harness = self.harness_entry(&record.harness).ok_or_else(|| {
            AssistantSessionError::UnknownHarness {
                requested: record.harness.clone(),
                declared: self.declared_harnesses(),
            }
        })?;
        let account = match record.account.as_deref() {
            Some(name) => Some(self.resolve_account_by_name(harness.id, name)?),
            None => None,
        };
        // Availability is RE-MEASURED inside the plan: `npx` may have been
        // uninstalled since the session was created, and a stale "available"
        // would surface as an unexplained spawn failure instead of the
        // catalogue's install hint.
        let plan = launch::plan(session_id, harness, account, self.aion_endpoint())?;
        // THE one mint. The child is handed `plan.token`'s secret, so the
        // record's digest is set to that token's digest, unconditionally — the
        // digest follows the secret most recently handed to a live child, and
        // nothing else ever writes it. The record is RE-READ from the store
        // first: the caller's `record` was read before this await point, and
        // writing a stale copy back is how a freshly stored digest gets
        // clobbered by an older one (the defect this replaces: `touch()` used
        // to re-put its caller's pre-spawn copy, restoring the pre-spawn digest
        // and 401ing every MCP call the child made).
        if let Some(minted) = plan.token.as_ref() {
            let mut updated = self
                .store()
                .get_assistant_session(&session_id)
                .await?
                .ok_or(AssistantSessionError::NotFound { session_id })?;
            updated.mcp_token_digest = Some(minted.digest().to_owned());
            updated.updated_at = Utc::now();
            self.store().put_assistant_session(updated).await?;
        }
        // The grounding pack is rewritten before the child starts, so the
        // directory the first turn's preamble names holds THIS binary's
        // documents by the time the agent can read it.
        let grounding_dir = grounding::directory().map_err(AssistantSessionError::Internal)?;
        grounding::materialize(&grounding_dir).map_err(AssistantSessionError::Internal)?;
        let mut built = plan.harness;
        if prior.is_some() {
            built = built.with_session_parameter(RESUME_PARAMETER);
        }
        let input = spawn_input(composed_prompt, prior.as_deref())?;
        let spec = AgentRunSpec::new(
            WorkflowId::new(session_id.as_uuid()),
            RunId::new_v4(),
            ActivityId::from_sequence_position(1),
            1,
            ACTIVITY_TYPE.to_owned(),
            input,
        );
        // No spawn timeout: the spawn either yields `initialize` or the child
        // exits, and both are events (RULED 2026-08-29). A child that starts and
        // then says nothing holds this turn open until it exits or the caller
        // goes away — at which point the future is dropped and the agent's whole
        // process group goes with it.
        let started =
            built
                .start(spec)
                .await
                .map_err(|error| AssistantSessionError::HarnessFailed {
                    harness: record.harness.clone(),
                    reason: error.to_string(),
                })?;

        // The registry's own channel, never a fresh one: a socket opened on this
        // session before the spawn must receive what the spawn produces.
        let recorder = self.recorder(session_id);
        let mut started = started;
        let handle = started.take_first_turn().ok_or_else(|| {
            AssistantSessionError::Internal(
                "a freshly started assistant harness had no opening turn to drive".to_owned(),
            )
        })?;
        let live = Arc::new(LiveSession::new(session_id, started, recorder));
        self.adopt(Arc::clone(&live));
        // Recorded through the LIVE session's recorder, so the frames a watcher
        // is already subscribed to include the opening.
        live.recorder()
            .record(AssistantSessionEvent::SessionOpened {
                acp_session_ref: live.acp_session_ref().to_owned(),
                load_session: live.supports_load_session(),
                at: Utc::now(),
                resumed: prior.is_some(),
            })
            .await?;
        live.recorder()
            .record(AssistantSessionEvent::State {
                state: AssistantSessionState::Live,
                reason: None,
            })
            .await?;
        // The agent may advertise its configuration options — the model picker
        // among them — on the OPENING response rather than as an update. The
        // pump only streams updates into open turns, so this is the one moment
        // that advertisement exists; recorded here it reaches the transcript,
        // the record cache and every watcher exactly as a mid-turn update
        // would. An unreadable advertisement is recorded verbatim instead —
        // the same nothing-is-dropped rule the translator applies.
        if let Some(advertised) = live.take_initial_config_options() {
            let frame = match frames::config_options(&advertised) {
                Some(options) => AssistantSessionEvent::ConfigOptions { options },
                None => AssistantSessionEvent::Raw {
                    turn_id: None,
                    source: "session/new/config_options".to_owned(),
                    value: advertised,
                },
            };
            live.recorder().record(frame).await?;
        }
        Ok((live, handle))
    }

    /// Record the frames that open a turn.
    ///
    /// The REQUEST comes first and carries what the operator asked and the
    /// screen they asked it from — one record, appended before anything the
    /// harness produces for this turn, so a reloaded conversation is questions
    /// and answers rather than answers alone.
    async fn record_turn_start(
        &self,
        live: &Arc<LiveSession>,
        turn_id: &str,
        context: Option<AssistantTurnContext>,
        text: &str,
        command: Option<AssistantCommandInvocation>,
        composed: &str,
    ) -> Result<(), AssistantSessionError> {
        live.recorder()
            .record(AssistantSessionEvent::Request {
                turn_id: turn_id.to_owned(),
                text: text.to_owned(),
                // An EMPTY context is no context: recording one would put a
                // blank screen on the record and make the shared context the
                // harness reads emptier than the last real one.
                context: context.filter(|context| !prompt::is_empty(context)),
                command,
            })
            .await?;
        live.recorder()
            .record(AssistantSessionEvent::TurnStarted {
                turn_id: turn_id.to_owned(),
                at: Utc::now(),
                prompt: composed.to_owned(),
            })
            .await
            .map(drop)
    }

    /// Refuse a command the harness never advertised, naming what it did.
    ///
    /// The record's cache answers first because it is already loaded; only a
    /// name that is NOT in it costs a full transcript read, and that read is
    /// what decides. A cache that had fallen behind would otherwise refuse a
    /// command the agent really does serve — a refusal is the one answer that
    /// must not be given on stale information.
    async fn require_advertised(
        &self,
        record: &AssistantSessionRecord,
        command: &AssistantCommandInvocation,
    ) -> Result<(), AssistantSessionError> {
        if record
            .commands
            .iter()
            .any(|advertised| advertised.name == command.name)
        {
            return Ok(());
        }
        let advertised = self.projection(record.session_id).await?.commands;
        if advertised.iter().any(|entry| entry.name == command.name) {
            return Ok(());
        }
        Err(AssistantSessionError::UnknownCommand {
            session_id: record.session_id,
            requested: command.name.clone(),
            advertised: declared_names(advertised.iter().map(|entry| entry.name.as_str())),
        })
    }

    /// The catalogue harness the caller named, or the one they last used.
    ///
    /// A caller who names none gets their LAST PICK, and a caller who has never
    /// picked gets the first catalogue entry that can actually run on this
    /// machine — availability measured now, because offering an entry this box
    /// cannot start is how a first message turns into a refusal. If nothing can
    /// run, the first entry is chosen anyway so the refusal that follows names
    /// a harness and carries its install hint, rather than being a bare "no".
    async fn resolve_harness(
        &self,
        subject: &str,
        requested: Option<&str>,
    ) -> Result<&'static CatalogueHarness, AssistantSessionError> {
        if let Some(name) = requested {
            return self
                .harness_entry(name)
                .ok_or_else(|| AssistantSessionError::UnknownHarness {
                    requested: name.to_owned(),
                    declared: self.declared_harnesses(),
                });
        }
        if let Some(remembered) = self.last_harness_pick(subject).await?
            && let Some(harness) = self.harness_entry(&remembered)
        {
            return Ok(harness);
        }
        self.catalogue()
            .iter()
            .find(|entry| entry.available())
            .or_else(|| self.catalogue().first())
            .ok_or_else(|| AssistantSessionError::UnknownHarness {
                requested: "(none named)".to_owned(),
                declared: self.declared_harnesses(),
            })
    }

    /// The catalogue entry `id` names in THIS registry's catalogue, or none.
    fn harness_entry(&self, id: &str) -> Option<&'static CatalogueHarness> {
        catalogue::harness_in(self.catalogue(), id)
    }

    /// Every harness this registry offers, for a refusal that has to say so.
    fn declared_harnesses(&self) -> String {
        catalogue::ids_of(self.catalogue())
    }

    /// The account the caller named, or none when they named none.
    fn resolve_account(
        &self,
        harness: &str,
        requested: Option<&str>,
    ) -> Result<Option<&ResolvedAssistantAccount>, AssistantSessionError> {
        match requested {
            None => Ok(None),
            Some(name) => self.resolve_account_by_name(harness, name).map(Some),
        }
    }

    /// The named account, or a refusal listing what the harness declares.
    fn resolve_account_by_name(
        &self,
        harness: &str,
        name: &str,
    ) -> Result<&ResolvedAssistantAccount, AssistantSessionError> {
        self.config()
            .account(harness, name)
            .ok_or_else(|| AssistantSessionError::UnknownAccount {
                harness: harness.to_owned(),
                requested: name.to_owned(),
                declared: declared_names(
                    self.config()
                        .harness(harness)
                        .into_iter()
                        .flat_map(|declared| declared.accounts.iter())
                        .map(|account| account.name.as_str()),
                ),
            })
    }
}

/// The job input one spawn carries: the prompt, and the prior session when this
/// opening is a resume.
///
/// A JSON OBJECT rather than a bare string because that is how the adapter's
/// workspace reads a resume parameter: the prompt is the one field that is
/// neither the directory parameter nor the declared session parameter.
fn spawn_input(prompt: &str, prior: Option<&str>) -> Result<Payload, AssistantSessionError> {
    let mut input = serde_json::Map::new();
    input.insert(
        PROMPT_PARAMETER.to_owned(),
        serde_json::Value::String(prompt.to_owned()),
    );
    if let Some(prior) = prior {
        input.insert(
            RESUME_PARAMETER.to_owned(),
            serde_json::Value::String(prior.to_owned()),
        );
    }
    let bytes = serde_json::to_vec(&serde_json::Value::Object(input)).map_err(|error| {
        AssistantSessionError::Internal(format!("the harness job input is not encodable: {error}"))
    })?;
    Ok(Payload::new(ContentType::Json, bytes))
}

/// The declared names, rendered for a refusal.
fn declared_names<'name>(names: impl Iterator<Item = &'name str>) -> String {
    let rendered: Vec<String> = names.map(|name| format!("`{name}`")).collect();
    if rendered.is_empty() {
        return "none".to_owned();
    }
    rendered.join(", ")
}

/// The job-input field carrying the opening prompt.
const PROMPT_PARAMETER: &str = "prompt";
/// The job-input field naming the prior ACP session to reload.
const RESUME_PARAMETER: &str = "acp_session";
/// The activity type stamped on a session's transcript events.
const ACTIVITY_TYPE: &str = "assistant.session";
/// `source` on a context pushed without a question.
const CONTEXT_SOURCE_PUSH: &str = "push";

/// Why a freshly created session is dormant.
pub const CREATED_AWAITING_FIRST_TURN: &str = "created; the harness starts with the first turn, because ACP opens a conversation by asking \
     something";
/// Why a session with a gone process that CAN be reloaded is dormant.
pub const PROCESS_EXITED_RESUMABLE: &str = "process_exited; the agent advertised `loadSession`, so the next turn reopens this \
     conversation";
/// Why a session with a gone process that cannot be reloaded is ended.
pub const PROCESS_EXITED_ENDED: &str = "process_exited; the agent did not advertise `loadSession`, so its conversation cannot be \
     reopened";
/// Why a resume was refused.
pub const RESUME_REFUSED: &str = "resume_refused: loadSession not advertised — this agent cannot reload a prior conversation, \
     and starting a fresh one would silently discard it";
/// Why a deleted session ended.
pub const DELETED: &str = "deleted by the caller";
/// Why a turn on an ended session is refused: ended is terminal. A session the
/// caller deleted, or whose agent could not reload it, has no next turn — a
/// turn that spawned a fresh agent for it would resurrect a conversation the
/// caller had already put down for good, under the same id.
pub const ENDED_IS_TERMINAL: &str =
    "this session has ended and cannot be turned again; open a new session";
/// Why a session is closed rather than ended: the caller put it away, and the
/// agent can reload it.
pub const CLOSED_RESUMABLE: &str = "closed by the caller; the agent advertised `loadSession`, so a turn taken on this \
     conversation reopens it";
/// Why a session went dormant when the server stopped.
pub const SERVER_STOPPED: &str = "process_exited; the server that held this session stopped";
/// The `TurnFailed` message for a turn whose process died before answering.
pub const OPEN_TURN_ORPHANED: &str =
    "the agent's process ended before this turn finished; nothing further arrives for it";

/// Validate a config change against the session's advertisement and produce
/// the wire value — the model picker's refusal-by-advertisement, carved out of
/// the facade so the facade stays readable.
///
/// # Errors
///
/// [`AssistantSessionError::UnknownConfigOption`] for an unadvertised option,
/// an unadvertised select choice, or a value of the wrong shape for the
/// advertised kind — in every case naming what IS advertised, because the fix
/// is always "send what the advertisement says".
fn chosen_config_value(
    config_options: &[AssistantConfigOption],
    session_id: AssistantSessionId,
    option_id: &str,
    value: &serde_json::Value,
) -> Result<AcpConfigValue, AssistantSessionError> {
    let advertised_ids = || {
        if config_options.is_empty() {
            "none".to_owned()
        } else {
            config_options
                .iter()
                .map(|option| option.id.as_str())
                .collect::<Vec<_>>()
                .join(", ")
        }
    };
    let Some(advertised) = config_options.iter().find(|option| option.id == option_id) else {
        return Err(AssistantSessionError::UnknownConfigOption {
            session_id,
            requested: option_id.to_owned(),
            advertised: advertised_ids(),
        });
    };
    match (&advertised.value, value) {
        (AssistantConfigValue::Select { choices, .. }, serde_json::Value::String(id)) => {
            if choices.iter().any(|choice| choice.id == *id) {
                Ok(AcpConfigValue::Choice(id.clone()))
            } else {
                Err(AssistantSessionError::UnknownConfigOption {
                    session_id,
                    requested: format!("{option_id} = {id}"),
                    advertised: choices
                        .iter()
                        .map(|choice| choice.id.as_str())
                        .collect::<Vec<_>>()
                        .join(", "),
                })
            }
        }
        (AssistantConfigValue::Toggle { .. }, serde_json::Value::Bool(flag)) => {
            Ok(AcpConfigValue::Toggle(*flag))
        }
        // A value of the wrong SHAPE for the advertised kind: refused with the
        // same variant, because the fix is the same — send what the
        // advertisement says this option takes.
        (AssistantConfigValue::Select { .. }, _) => {
            Err(AssistantSessionError::UnknownConfigOption {
                session_id,
                requested: format!("{option_id} (a select takes a choice id string)"),
                advertised: advertised_ids(),
            })
        }
        (AssistantConfigValue::Toggle { .. }, _) => {
            Err(AssistantSessionError::UnknownConfigOption {
                session_id,
                requested: format!("{option_id} (a toggle takes true or false)"),
                advertised: advertised_ids(),
            })
        }
    }
}