aion-server 0.29.0

//! 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, AssistantSessionEvent, AssistantSessionFrame, AssistantSessionId,
    AssistantSessionProjection, AssistantSessionState, AssistantSessionSummary,
    AssistantTurnContext, ContentType, Payload,
};
use aion_integration_acp::TurnHandle;
use aion_integration_acp::catalogue::{self, CatalogueHarness};
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::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(),
        };
        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 continuable row. 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_continuable()))
    }

    /// 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
    }

    /// 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()),
        ))
    }

    /// End a session: 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.
    ///
    /// # 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 session. A call still holding
        // its `Arc` finishes safely; nothing new is minted for a deleted
        // session, because the bearer check refuses an ended session's tools.
        self.document_edit_locks().remove(&session_id);
        self.fail_open_turn(session_id, "process_exited").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?;
        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 = catalogue::harness(&record.harness).ok_or_else(|| {
            AssistantSessionError::UnknownHarness {
                requested: record.harness.clone(),
                declared: catalogue::ids(),
            }
        })?;
        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?;
        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 catalogue::harness(name).ok_or_else(|| AssistantSessionError::UnknownHarness {
                requested: name.to_owned(),
                declared: catalogue::ids(),
            });
        }
        if let Some(remembered) = self.last_harness_pick(subject).await?
            && let Some(harness) = catalogue::harness(&remembered)
        {
            return Ok(harness);
        }
        catalogue::CATALOGUE
            .iter()
            .find(|entry| entry.available())
            .or_else(|| catalogue::CATALOGUE.first())
            .ok_or_else(|| AssistantSessionError::UnknownHarness {
                requested: "(none named)".to_owned(),
                declared: catalogue::ids(),
            })
    }

    /// 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 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";