aion-server 0.31.0

//! The lifecycle decisions, measured against a real store.
//!
//! # Why no process is spawned
//!
//! Every decision pinned here is made BEFORE a harness would be started: whether
//! a dormant session can be reopened, whether an append that fails takes the
//! broadcast with it, whether a command the agent never advertised is refused,
//! which session is the caller's current one. Spawning an agent to measure them
//! would put a process boundary between the assertion and the thing it is about.
//!
//! Where a decision can only be told apart by what happens NEXT — the resumable
//! arm — the discriminator is the failure's own variant: a session that must not
//! be reopened is refused as [`AssistantSessionError::Ended`] and never reaches
//! a spawn, while a resumable one reaches one and fails as
//! [`AssistantSessionError::HarnessFailed`] against a binary that is not an ACP
//! agent. Two different variants is the proof that two different branches ran.

use std::sync::Arc;

use aion_core::{
    AssistantCommand, AssistantCommandInvocation, AssistantSessionEvent, AssistantSessionState,
};
use aion_store::assistant::AssistantSessionStore;

use super::error::AssistantSessionError;
use super::fixture::{
    OPERATOR, RefusingTranscriptStore, open_session, opened, registry, registry_over, went_dormant,
};
use super::lifecycle::{CLOSED_RESUMABLE, DELETED, ENDED_IS_TERMINAL, RESUME_REFUSED};

type TestResult = Result<(), String>;

/// The bound every "must not hang" assertion runs under.
///
/// Generous relative to the work (all of it is in-memory) and short relative to
/// a hang: a refusal that took this long would be a refusal nobody waits for.
const BOUNDED_WAIT: std::time::Duration = std::time::Duration::from_secs(10);

/// Run `future` under [`BOUNDED_WAIT`], failing by name if it does not finish.
///
/// The requirement is "a typed refusal WITHIN a bounded wait", and a refusal
/// that arrives is only half of it — a hang and a very slow answer look
/// identical to anyone watching, so the bound is asserted rather than assumed.
async fn within_bound<T>(what: &str, future: impl Future<Output = T>) -> Result<T, String> {
    tokio::time::timeout(BOUNDED_WAIT, future)
        .await
        .map_err(|_elapsed| format!("{what} did not answer within {BOUNDED_WAIT:?}; it hung"))
}

/// Row (a), the refusing arm: a dormant session whose agent never advertised
/// `loadSession` refuses the next turn, by a TYPED variant, within a bounded
/// wait — and never spawns a process that would then hang on a `session/load`
/// the agent cannot answer.
#[tokio::test]
async fn a_dormant_session_whose_agent_cannot_reload_refuses_the_next_turn() -> TestResult {
    let (sessions, _store) = registry();
    let summary = open_session(&sessions).await?;
    opened(&sessions, summary.session_id, false).await?;
    went_dormant(&sessions, summary.session_id).await?;

    let refusal = within_bound(
        "a turn on an unresumable dormant session",
        sessions.turn(
            OPERATOR,
            summary.session_id,
            "carry on".to_owned(),
            None,
            None,
        ),
    )
    .await?;
    match refusal {
        Err(AssistantSessionError::Ended { reason, .. }) => {
            assert_eq!(
                reason, RESUME_REFUSED,
                "the refusal must name the capability the agent did not advertise"
            );
            assert!(
                reason.contains("loadSession"),
                "the reason names the capability: {reason}"
            );
        }
        other => {
            return Err(format!(
                "an unresumable dormant session must refuse as Ended, got {other:?}"
            ));
        }
    }

    // And the refusal is WRITTEN BACK: the session is now settled ended by an
    // appended record, so the next reader projects the decision rather than
    // recomputing it — and the panel can say why instead of offering a box that
    // would refuse again.
    let (state, reason) = sessions
        .state_of_session(summary.session_id)
        .await
        .map_err(|error| error.to_string())?;
    assert_eq!(state, AssistantSessionState::Ended);
    assert_eq!(reason.as_deref(), Some(RESUME_REFUSED));
    Ok(())
}

/// Row (a), the resuming arm, and the density claim.
///
/// A dormant session whose agent DID advertise `loadSession` takes the other
/// branch: it reaches a spawn, and the spawn asks the agent to load the prior
/// conversation. The fixture's stub agent refuses every `session/load` by name,
/// so the spawn fails — as `HarnessFailed`, which is a different variant from
/// the refusing arm above and is therefore the proof that a different branch
/// ran; and it fails in milliseconds on any box under any load, which is what
/// keeps [`BOUNDED_WAIT`] an assertion about this code rather than about the
/// host. It is NOT settled ended, because nothing decided the conversation was
/// over.
///
/// The transcript's indices continue DENSE across the whole episode: the resume
/// attempt appends where the transcript left off rather than restarting, which
/// is what makes a client's `?after=` cursor survive a dormancy.
#[tokio::test]
async fn a_dormant_session_whose_agent_can_reload_reaches_a_resume_and_keeps_its_indices()
-> TestResult {
    let (sessions, store) = registry();
    let summary = open_session(&sessions).await?;
    opened(&sessions, summary.session_id, true).await?;
    went_dormant(&sessions, summary.session_id).await?;
    let before = store
        .assistant_transcript_head(&summary.session_id)
        .await
        .map_err(|error| error.to_string())?;

    let outcome = within_bound(
        "a turn on a resumable dormant session",
        sessions.turn(
            OPERATOR,
            summary.session_id,
            "carry on".to_owned(),
            None,
            None,
        ),
    )
    .await?;
    match outcome {
        Err(AssistantSessionError::HarnessFailed { .. }) => {}
        Err(AssistantSessionError::Ended { reason, .. }) => {
            return Err(format!(
                "a resumable session must NOT be settled ended before it is tried: {reason}"
            ));
        }
        other => {
            return Err(format!(
                "the fixture's command is not an ACP agent, so the spawn must fail by name, got \
                 {other:?}"
            ));
        }
    }

    // Nothing settled it: a failed spawn is a failed attempt, not the end of a
    // conversation the agent still holds.
    let (state, _reason) = sessions
        .state_of_session(summary.session_id)
        .await
        .map_err(|error| error.to_string())?;
    assert_ne!(
        state,
        AssistantSessionState::Ended,
        "a spawn that failed must not end a conversation the agent can still reload"
    );

    // Dense, and CONTINUING: every index from zero, with nothing restarted.
    let frames = store
        .assistant_transcript(&summary.session_id, None)
        .await
        .map_err(|error| error.to_string())?;
    let indices: Vec<u64> = frames.iter().map(|frame| frame.index).collect();
    let expected: Vec<u64> = (0..indices.len() as u64).collect();
    assert_eq!(
        indices, expected,
        "the transcript's indices must stay dense from zero across a resume"
    );
    assert!(
        indices.len() as u64 >= before,
        "the resume must append at the head, never behind it"
    );
    Ok(())
}

/// Putting a conversation away is not ending it — when the agent can reload
/// it.
///
/// Two arms on the same delete, told apart by the one fact every resume
/// decision turns on. The agent that advertised `loadSession` leaves a session
/// that is CLOSED: settled by name, out of the operator's current slot, and
/// still reaching a spawn on the next turn (the fixture's binary is not an ACP
/// agent, so that spawn fails as `HarnessFailed` — the variant that proves the
/// resume arm ran, not the refusing one). The agent that did not leaves a
/// session that is ENDED, refused before any spawn.
#[tokio::test]
async fn closing_a_reloadable_session_keeps_it_reopenable_but_never_current() -> TestResult {
    let (sessions, _store) = registry();

    // The reloadable arm.
    let reloadable = open_session(&sessions).await?;
    opened(&sessions, reloadable.session_id, true).await?;
    went_dormant(&sessions, reloadable.session_id).await?;
    let current = sessions
        .current(OPERATOR)
        .await
        .map_err(|error| error.to_string())?
        .map(|summary| summary.session_id);
    assert_eq!(
        current,
        Some(reloadable.session_id),
        "before it is put away, the dormant session is the operator's current one"
    );
    sessions
        .delete(OPERATOR, reloadable.session_id)
        .await
        .map_err(|error| error.to_string())?;
    let (state, reason) = sessions
        .state_of_session(reloadable.session_id)
        .await
        .map_err(|error| error.to_string())?;
    assert_eq!(
        state,
        AssistantSessionState::Closed,
        "a reloadable session the caller shut is closed, not ended"
    );
    assert_eq!(reason.as_deref(), Some(CLOSED_RESUMABLE));
    let current = sessions
        .current(OPERATOR)
        .await
        .map_err(|error| error.to_string())?;
    assert!(
        current.is_none(),
        "a closed session is never the current one, however new it is: {current:?}"
    );
    let outcome = within_bound(
        "a turn on a closed session",
        sessions.turn(
            OPERATOR,
            reloadable.session_id,
            "let us pick this back up".to_owned(),
            None,
            None,
        ),
    )
    .await?;
    match outcome {
        Err(AssistantSessionError::HarnessFailed { .. }) => {}
        other => {
            return Err(format!(
                "a turn on a closed session must reach a resume (and fail by name against a \
                 binary that is not an agent), got {other:?}"
            ));
        }
    }

    // The arm that cannot reload.
    let unreloadable = open_session(&sessions).await?;
    opened(&sessions, unreloadable.session_id, false).await?;
    went_dormant(&sessions, unreloadable.session_id).await?;
    sessions
        .delete(OPERATOR, unreloadable.session_id)
        .await
        .map_err(|error| error.to_string())?;
    let (state, reason) = sessions
        .state_of_session(unreloadable.session_id)
        .await
        .map_err(|error| error.to_string())?;
    assert_eq!(
        state,
        AssistantSessionState::Ended,
        "a session whose agent cannot reload it is ended when the caller shuts it"
    );
    assert_eq!(reason.as_deref(), Some(DELETED));
    match within_bound(
        "a turn on an ended session",
        sessions.turn(
            OPERATOR,
            unreloadable.session_id,
            "carry on".to_owned(),
            None,
            None,
        ),
    )
    .await?
    {
        Err(AssistantSessionError::Ended { .. }) => Ok(()),
        other => Err(format!(
            "a turn on an ended session is refused before any spawn, got {other:?}"
        )),
    }
}

/// Ended is terminal through the OTHER door: a session that was created and
/// never opened, then deleted, has no prior conversation to resume — so the
/// spawn path, not the resume path, would serve its next turn. Before this
/// cell it did: the turn spawned a fresh agent under the deleted id and the
/// transcript read `live` again (a real transcript, 2026-08-31, a console
/// deleting a session moments before its first turn arrived). The turn is
/// refused by name, nothing spawns, and no settling record can leave `ended`.
#[tokio::test]
async fn a_turn_on_a_never_opened_session_the_caller_deleted_is_refused_and_spawns_nothing()
-> TestResult {
    let (sessions, store) = registry();
    let created = open_session(&sessions).await?;
    sessions
        .delete(OPERATOR, created.session_id)
        .await
        .map_err(|error| error.to_string())?;
    let (state, reason) = sessions
        .state_of_session(created.session_id)
        .await
        .map_err(|error| error.to_string())?;
    assert_eq!(state, AssistantSessionState::Ended);
    assert_eq!(reason.as_deref(), Some(DELETED));
    let frames_before = store
        .assistant_transcript(&created.session_id, None)
        .await
        .map_err(|error| error.to_string())?
        .len();

    match within_bound(
        "a turn on a never-opened, deleted session",
        sessions.turn(
            OPERATOR,
            created.session_id,
            "hello?".to_owned(),
            None,
            None,
        ),
    )
    .await?
    {
        Err(AssistantSessionError::Ended { reason, .. }) => {
            assert_eq!(reason, ENDED_IS_TERMINAL);
        }
        other => {
            return Err(format!(
                "a turn on an ended session is refused by name before any spawn, got {other:?}"
            ));
        }
    }
    // The refusal is the turn's own failure frame — one row, so a reloaded
    // conversation shows the question with its reason — and nothing else: no
    // opening, because nothing spawned.
    let rows_after = store
        .assistant_transcript(&created.session_id, None)
        .await
        .map_err(|error| error.to_string())?
        .len();
    assert_eq!(
        rows_after,
        frames_before + 1,
        "a refused turn on an ended session records exactly its own failure"
    );
    let frames_after = sessions
        .transcript_from(created.session_id, None)
        .await
        .map_err(|error| error.to_string())?;
    match frames_after.last().map(|frame| &frame.event) {
        Some(AssistantSessionEvent::TurnFailed { message, .. }) => {
            assert!(
                message.contains(ENDED_IS_TERMINAL),
                "the failure frame names the reason: {message}"
            );
        }
        other => {
            return Err(format!(
                "the last frame is the refused turn's failure, got {other:?}"
            ));
        }
    }
    assert!(
        !frames_after
            .iter()
            .any(|frame| matches!(frame.event, AssistantSessionEvent::SessionOpened { .. })),
        "a refused turn on an ended session spawns nothing: {frames_after:?}"
    );

    // The other direction: a settling record that would leave `ended` is refused.
    match sessions
        .settle(created.session_id, AssistantSessionState::Live, "resurrect")
        .await
    {
        Err(AssistantSessionError::Ended { reason, .. }) => assert_eq!(reason, DELETED),
        other => {
            return Err(format!(
                "settling out of ended must be refused, got {other:?}"
            ));
        }
    }
    let (state, _) = sessions
        .state_of_session(created.session_id)
        .await
        .map_err(|error| error.to_string())?;
    assert_eq!(state, AssistantSessionState::Ended, "still ended");
    Ok(())
}

/// T1: append-before-broadcast. A store that refuses the append fails the turn
/// and broadcasts NOTHING.
///
/// The subscriber is taken first and is the instrument: a frame the client saw
/// and the store never took would make the live stream and the record disagree
/// in the one direction no later read can repair.
#[tokio::test]
async fn a_store_that_refuses_the_append_fails_the_turn_and_broadcasts_nothing() -> TestResult {
    let store = Arc::new(RefusingTranscriptStore::new());
    let sessions = registry_over(Arc::clone(&store) as Arc<dyn AssistantSessionStore>);
    // The record write still works on this store, so a session exists to turn
    // on — and `create` itself fails at its first APPEND, which is the same
    // discipline seen one call earlier.
    let created = open_session(&sessions).await;
    match created {
        Err(message) => {
            assert!(
                message.contains(RefusingTranscriptStore::REFUSAL),
                "the refusal must reach the caller verbatim: {message}"
            );
        }
        Ok(summary) => {
            return Err(format!(
                "a store that refuses every append must fail the opening it cannot record, got \
                 {summary:?}"
            ));
        }
    }

    // The record was written before the refused append, so a session id exists
    // to subscribe to and drive. This is the arm the requirement names.
    let listing = store
        .list_assistant_sessions()
        .await
        .map_err(|error| error.to_string())?;
    let record = listing
        .sessions
        .first()
        .ok_or_else(|| {
            "the create wrote its record before the append it could not make".to_owned()
        })?
        .clone();

    let mut watcher = sessions.recorder(record.session_id).subscribe();
    let refusal = within_bound(
        "a turn on a store that refuses every append",
        sessions.turn(OPERATOR, record.session_id, "hello".to_owned(), None, None),
    )
    .await?;
    assert!(
        refusal.is_err(),
        "a turn whose frames cannot be recorded must fail rather than stream unrecorded frames"
    );
    assert!(
        matches!(
            watcher.try_recv(),
            Err(tokio::sync::broadcast::error::TryRecvError::Empty)
        ),
        "NOTHING may be broadcast when the append that would have justified it failed"
    );
    Ok(())
}

/// The positive control for the cell above: on a store that ACCEPTS the append,
/// the same subscriber does receive frames. Without this, a broadcast channel
/// that never delivered anything would pass the pin above for the wrong reason.
#[tokio::test]
async fn a_store_that_accepts_the_append_does_broadcast_what_it_took() -> TestResult {
    let (sessions, _store) = registry();
    let summary = open_session(&sessions).await?;
    let mut watcher = sessions.recorder(summary.session_id).subscribe();
    sessions
        .recorder(summary.session_id)
        .record(AssistantSessionEvent::Delta {
            turn_id: "t-1".to_owned(),
            text: "hello".to_owned(),
        })
        .await
        .map_err(|error| error.to_string())?;
    let frame = watcher
        .try_recv()
        .map_err(|error| format!("a committed frame must reach a subscriber: {error}"))?;
    assert!(matches!(frame.event, AssistantSessionEvent::Delta { .. }));
    Ok(())
}

/// C4: the REQUEST frame is appended at acceptance, before anything the harness
/// produces for that turn — so a reloaded conversation is questions and answers
/// rather than answers alone.
///
/// Measured against the fixture's stub agent, which starts and answers, so the
/// ordering is a real one between a request and the frames an agent produced.
/// (A turn refused at the spawn records only its failure — O5's second half,
/// pinned below — because no turn frame may precede the opening.)
#[tokio::test]
async fn the_operators_request_reaches_the_transcript_before_any_harness_frame() -> TestResult {
    let (sessions, store) = registry();
    let summary = open_session(&sessions).await?;
    // The spawn will fail, which is fine: the request is recorded by the turn
    // path, and nothing the harness would say can precede it.
    drop(
        within_bound(
            "a turn whose harness cannot start",
            sessions.turn(
                OPERATOR,
                summary.session_id,
                "fix the check".to_owned(),
                None,
                None,
            ),
        )
        .await?,
    );
    let frames = store
        .assistant_transcript(&summary.session_id, None)
        .await
        .map_err(|error| error.to_string())?;
    let decoded: Vec<AssistantSessionEvent> = frames
        .iter()
        .map(|frame| serde_json::from_slice(frame.payload.bytes()))
        .collect::<Result<_, _>>()
        .map_err(|error| error.to_string())?;
    assert!(
        decoded.iter().any(|event| matches!(
            event,
            AssistantSessionEvent::Request { text, .. } if text == "fix the check"
        )),
        "the operator's own words must be on the transcript: {decoded:?}"
    );
    Ok(())
}

/// T4's refusal arm: a command the harness never advertised is refused by name,
/// and the refusal states what the harness DOES advertise.
#[tokio::test]
async fn a_command_the_harness_never_advertised_is_refused_naming_what_it_does_offer() -> TestResult
{
    let (sessions, _store) = registry();
    let summary = open_session(&sessions).await?;
    sessions
        .recorder(summary.session_id)
        .record(AssistantSessionEvent::AvailableCommands {
            commands: vec![AssistantCommand {
                name: "compact".to_owned(),
                description: "compact the conversation".to_owned(),
                input_hint: None,
            }],
        })
        .await
        .map_err(|error| error.to_string())?;

    let refusal = within_bound(
        "a turn invoking an unadvertised command",
        sessions.turn(
            OPERATOR,
            summary.session_id,
            String::new(),
            None,
            Some(AssistantCommandInvocation {
                name: "rewrite-everything".to_owned(),
                input: None,
            }),
        ),
    )
    .await?;
    match refusal {
        Err(AssistantSessionError::UnknownCommand {
            requested,
            advertised,
            ..
        }) => {
            assert_eq!(requested, "rewrite-everything");
            assert!(
                advertised.contains("compact"),
                "the refusal must say what IS offered: {advertised}"
            );
        }
        other => {
            return Err(format!(
                "an unadvertised command must be refused by name, got {other:?}"
            ));
        }
    }
    Ok(())
}

/// The positive control: the ADVERTISED command is not refused here. It reaches
/// the spawn and fails there, which is a different variant — so the refusal
/// above is about the command and not about commands in general.
#[tokio::test]
async fn an_advertised_command_passes_the_check_and_reaches_the_harness() -> TestResult {
    let (sessions, store) = registry();
    let summary = open_session(&sessions).await?;
    sessions
        .recorder(summary.session_id)
        .record(AssistantSessionEvent::AvailableCommands {
            commands: vec![AssistantCommand {
                name: "compact".to_owned(),
                description: "compact the conversation".to_owned(),
                input_hint: Some("what to keep".to_owned()),
            }],
        })
        .await
        .map_err(|error| error.to_string())?;

    let outcome = within_bound(
        "a turn invoking an advertised command",
        sessions.turn(
            OPERATOR,
            summary.session_id,
            String::new(),
            None,
            Some(AssistantCommandInvocation {
                name: "compact".to_owned(),
                input: Some("keep the plan".to_owned()),
            }),
        ),
    )
    .await?;
    assert!(
        !matches!(outcome, Err(AssistantSessionError::UnknownCommand { .. })),
        "an advertised command must not be refused: {outcome:?}"
    );

    // And the DELIVERY form is on the record: `/compact keep the plan`, which is
    // exactly what the ACP schema's `unstructured` input describes — the command
    // name followed by everything typed after it.
    let frames = store
        .assistant_transcript(&summary.session_id, None)
        .await
        .map_err(|error| error.to_string())?;
    let decoded: Vec<AssistantSessionEvent> = frames
        .iter()
        .map(|frame| serde_json::from_slice(frame.payload.bytes()))
        .collect::<Result<_, _>>()
        .map_err(|error| error.to_string())?;
    assert!(
        decoded.iter().any(|event| matches!(
            event,
            AssistantSessionEvent::Request { text, command: Some(command), .. }
                if text == "/compact keep the plan" && command.name == "compact"
        )),
        "the composed line AND the structured invocation must both be recorded: {decoded:?}"
    );
    Ok(())
}

/// The listing's command cache is exactly what the transcript projects. The
/// record holds a copy so a listing need not read whole conversations; this is
/// what stops that copy becoming a second, drifting answer.
#[tokio::test]
async fn the_records_command_cache_is_what_the_transcript_projects() -> TestResult {
    let (sessions, store) = registry();
    let summary = open_session(&sessions).await?;
    let advertised = vec![
        AssistantCommand {
            name: "compact".to_owned(),
            description: "compact the conversation".to_owned(),
            input_hint: None,
        },
        AssistantCommand {
            name: "plan".to_owned(),
            description: "draft a plan".to_owned(),
            input_hint: Some("what to plan".to_owned()),
        },
    ];
    sessions
        .recorder(summary.session_id)
        .record(AssistantSessionEvent::AvailableCommands {
            commands: advertised.clone(),
        })
        .await
        .map_err(|error| error.to_string())?;

    let projected = sessions
        .projection(summary.session_id)
        .await
        .map_err(|error| error.to_string())?
        .commands;
    let cached = store
        .get_assistant_session(&summary.session_id)
        .await
        .map_err(|error| error.to_string())?
        .ok_or_else(|| "the session record is present".to_owned())?
        .commands;
    assert_eq!(projected, advertised);
    assert_eq!(
        cached, projected,
        "the record's cache and the transcript's projection are ONE answer"
    );
    Ok(())
}

/// T5, both arms: `current` is the caller's newest non-ended session, and an
/// absence when they hold none.
#[tokio::test]
async fn the_current_session_is_the_newest_continuable_one_or_an_absence() -> TestResult {
    let (sessions, _store) = registry();
    assert_eq!(
        sessions
            .current(OPERATOR)
            .await
            .map_err(|error| error.to_string())?,
        None,
        "a caller who has started nothing holds no current session"
    );

    let first = open_session(&sessions).await?;
    let second = open_session(&sessions).await?;
    let current = sessions
        .current(OPERATOR)
        .await
        .map_err(|error| error.to_string())?
        .ok_or_else(|| "a caller with two open sessions holds a current one".to_owned())?;
    assert_eq!(
        current.session_id, second.session_id,
        "the NEWEST continuable session is the current one"
    );

    // Ending the newest falls back to the one before it, rather than to nothing:
    // a closed conversation does not close the operator's thread.
    sessions
        .delete(OPERATOR, second.session_id)
        .await
        .map_err(|error| error.to_string())?;
    let after = sessions
        .current(OPERATOR)
        .await
        .map_err(|error| error.to_string())?
        .ok_or_else(|| "the earlier session is still continuable".to_owned())?;
    assert_eq!(after.session_id, first.session_id);

    // And another caller's sessions are not this caller's: `current` is
    // caller-scoped like every other read here.
    assert_eq!(
        sessions
            .current("somebody-else")
            .await
            .map_err(|error| error.to_string())?,
        None,
        "one caller's conversation must never be another's current session"
    );
    Ok(())
}

/// O5, the first half: NOTHING pre-warms.
///
/// Creating a session records a conversation and starts no process — there is no
/// opening on its transcript and no live session in the registry — so the first
/// message is what spawns the harness. A server that warmed a child at create
/// time would be running an agent on the operator's box for a conversation
/// nobody had started.
#[tokio::test]
async fn creating_a_session_starts_nothing_and_records_no_opening() -> TestResult {
    let (sessions, _store) = registry();
    let summary = open_session(&sessions).await?;
    assert_eq!(
        summary.state,
        AssistantSessionState::Dormant,
        "a conversation nobody has spoken in has no process"
    );
    assert!(
        !sessions.is_live(summary.session_id).await,
        "no harness process may exist before the first message"
    );
    let frames = sessions
        .transcript_from(summary.session_id, None)
        .await
        .map_err(|error| error.to_string())?;
    assert!(
        !frames
            .iter()
            .any(|frame| matches!(frame.event, AssistantSessionEvent::SessionOpened { .. })),
        "an opening on the transcript of a session nobody has spoken in means a process was \
         started for it: {frames:?}"
    );
    Ok(())
}

/// O5, the second half: no turn frame can precede the opening.
///
/// The first turn goes to the SPAWN before it records anything of its own, so a
/// turn that never reached an agent leaves no `request` and no `turn_started` on
/// the transcript — only its own typed failure. That ordering is what makes
/// "the `initialize` event is recorded before any turn event" true by
/// construction rather than by luck: there is no path on which a turn frame is
/// written first.
///
/// Measured against a session whose harness the build no longer ships, because
/// that refusal is met at exactly the same point a missing binary is — inside
/// `ensure_live`, before any frame — and needs no process to reach.
#[tokio::test]
async fn a_turn_that_never_reached_an_agent_records_its_failure_and_no_turn_frames() -> TestResult {
    let (sessions, store) = registry();
    let summary = open_session(&sessions).await?;
    // The record names a harness this build does not carry — a session opened
    // by an older build, which is the honest shape of "the spawn cannot be
    // planned".
    let mut record = store
        .get_assistant_session(&summary.session_id)
        .await
        .map_err(|error| error.to_string())?
        .ok_or_else(|| "the opened session has a record".to_owned())?;
    record.harness = "a-harness-this-build-does-not-ship".to_owned();
    store
        .put_assistant_session(record)
        .await
        .map_err(|error| error.to_string())?;

    let refusal = within_bound(
        "a turn on a session whose harness cannot be planned",
        sessions.turn(OPERATOR, summary.session_id, "hello".to_owned(), None, None),
    )
    .await?;
    match refusal {
        Err(AssistantSessionError::UnknownHarness { requested, .. }) => {
            assert_eq!(requested, "a-harness-this-build-does-not-ship");
        }
        other => {
            return Err(format!(
                "a turn that cannot plan a spawn must refuse by name, got {other:?}"
            ));
        }
    }

    let frames = sessions
        .transcript_from(summary.session_id, None)
        .await
        .map_err(|error| error.to_string())?;
    assert!(
        !frames.iter().any(|frame| matches!(
            frame.event,
            AssistantSessionEvent::Request { .. } | AssistantSessionEvent::TurnStarted { .. }
        )),
        "a turn that never reached an agent must not have written a request or a start: {frames:?}"
    );
    // But it must not be silent either: the operator is watching the socket, so
    // the refusal is on the transcript with its typed code.
    let failed = frames
        .iter()
        .find_map(|frame| match &frame.event {
            AssistantSessionEvent::TurnFailed { code, message, .. } => {
                Some((code.clone(), message.clone()))
            }
            _ => None,
        })
        .ok_or_else(|| format!("the failure must reach the transcript: {frames:?}"))?;
    assert_eq!(failed.0, "unknown_harness");
    assert!(
        failed.1.contains("a-harness-this-build-does-not-ship"),
        "the recorded failure names what could not be run: {}",
        failed.1
    );
    Ok(())
}

/// O1: `sessions_disabled_reason` is for a fault the product can NAME.
///
/// A stock server serves the assistant, so the only thing that can take the
/// surface down is the store the conversation would live in — and when that
/// happens the reason carries the store's own error, not a sentence about a
/// configuration section that no longer exists.
#[tokio::test]
async fn a_store_the_boot_sweep_cannot_read_takes_the_surface_down_by_name() -> TestResult {
    let (sessions, _store) = registry();
    assert!(
        sessions.availability().is_available(),
        "a stock server serves the assistant before anything has gone wrong"
    );
    let refusing = registry_over(Arc::new(RefusingTranscriptStore::new()));
    // The boot sweep is where the store is first exercised end to end, so it is
    // where the answer comes from. A sweep that could not settle its orphans is
    // a server that cannot record a conversation.
    let settled = open_session(&refusing).await;
    assert!(
        settled.is_err(),
        "this instrument's store refuses every append"
    );
    refusing.report_store_fault(&AssistantSessionError::Store(
        aion_store::StoreError::Backend(RefusingTranscriptStore::REFUSAL.to_owned()),
    ));
    let reason = refusing
        .availability()
        .reason()
        .ok_or_else(|| "a server with an unusable store must say so".to_owned())?
        .to_owned();
    assert!(
        reason.contains(RefusingTranscriptStore::REFUSAL),
        "the reason carries the store's own error: {reason}"
    );
    assert!(
        !reason.contains("[assistant]"),
        "`not configured` is never the reason any more: {reason}"
    );
    // And a create on it is refused with that same sentence, so the panel and
    // the refusal say one thing.
    match refusing.create(OPERATOR, None, None, None).await {
        Err(AssistantSessionError::NotCommissioned { reason: refused }) => {
            assert_eq!(refused, reason);
            Ok(())
        }
        other => Err(format!(
            "a server that cannot record a conversation must refuse to open one, got {other:?}"
        )),
    }
}

/// The record's token digest follows the secret most recently handed to a live
/// child, and nothing that runs later in the same turn may restore an older
/// one.
///
/// This is the defect's own interleave, without a process. A turn used to read
/// its copy of the record, the spawn stored the digest of the secret it handed
/// the child, and then `touch` — taking the caller's pre-spawn COPY by value —
/// put that copy back whole, restoring the pre-spawn digest. The child held the
/// new secret, the store held the old digest, and every MCP call the agent made
/// was refused as `WrongToken` from that write on. The pin drives the same
/// interleave through the store: read-copy, spawn-write, bookkeeping — and
/// asserts the spawn's digest survives the bookkeeping, which also proves the
/// bookkeeping ran against what the store held NOW rather than a copy carried
/// across an await.
#[tokio::test]
async fn the_turn_bookkeeping_never_restores_a_pre_spawn_token_digest() -> TestResult {
    let (sessions, store) = registry();
    let summary = open_session(&sessions).await?;
    let session_id = summary.session_id;

    // The spawn's write: the digest of the token the child was actually handed.
    // (`open_session` mints no bearer, so the pre-spawn copy a stale writer
    // would restore is `None` — the exact value the live defect restored.)
    let handed = "the-digest-of-the-secret-the-child-holds".to_owned();
    let mut spawned = store
        .get_assistant_session(&session_id)
        .await
        .map_err(|error| error.to_string())?
        .ok_or("the opened session's record is missing before the spawn write")?;
    spawned.mcp_token_digest = Some(handed.clone());
    store
        .put_assistant_session(spawned)
        .await
        .map_err(|error| error.to_string())?;

    // The turn's bookkeeping, handed only the session id.
    sessions
        .touch(session_id, Some("first words of the conversation"))
        .await
        .map_err(|error| error.to_string())?;

    let held = store
        .get_assistant_session(&session_id)
        .await
        .map_err(|error| error.to_string())?
        .ok_or("the session's record is missing after the bookkeeping write")?;
    if held.mcp_token_digest.as_deref() != Some(handed.as_str()) {
        return Err(format!(
            "the bookkeeping write restored a pre-spawn token digest: the child holds a secret \
             for {handed:?} and the store now says {:?}",
            held.mcp_token_digest
        ));
    }
    if held.turns != 1 {
        return Err(format!(
            "the bookkeeping did not run: expected 1 recorded turn, found {}",
            held.turns
        ));
    }
    if held.title.as_deref() != Some("first words of the conversation") {
        return Err(format!(
            "the first turn's words did not become the title: found {:?}",
            held.title
        ));
    }
    Ok(())
}