aion-server 0.31.0

//! A registry over a real in-memory store, for the pins that need one.
//!
//! Everything here builds the SAME `AssistantSessions` the server runs, over the
//! reference store the conformance suite gates. No harness process is spawned:
//! the decisions these pins are about — whether a session is resumable, whether
//! a bearer is honoured, whether an append that fails takes the broadcast with
//! it — are all made before any process would be, which is exactly why they can
//! be measured without one.

use std::sync::Arc;

use aion_core::{
    AssistantSessionEvent, AssistantSessionId, AssistantSessionState, AssistantTurnContext,
};
use aion_store::assistant::{
    AssistantSessionListing, AssistantSessionRecord, AssistantSessionStore,
    AssistantTranscriptEvent,
};
use aion_store::{InMemoryStore, StoreError};
use async_trait::async_trait;
use chrono::{DateTime, Utc};

use crate::config::{
    AssistantAccountConfig, AssistantConfig, AssistantHarnessConfig, ResolvedAssistantConfig,
};

use aion_integration_acp::catalogue::CatalogueHarness;

use super::launch::AssistantEndpoints;
use super::registry::AssistantSessions;

/// The subject every fixture session belongs to unless a test says otherwise.
pub(crate) const OPERATOR: &str = "operator";

/// The catalogue harness the fixtures open sessions on.
///
/// A real catalogue id, because there is no other kind now: the launch line, the
/// availability and the install hint all come from the catalogue, and a made-up
/// name would be refused before any of the decisions these pins are about.
pub(crate) const HARNESS: &str = "claude-code";

/// The account the account-shaped fixtures name.
pub(crate) const ACCOUNT: &str = "work";

/// The server-environment variable the fixture account reads its value from.
pub(crate) const ACCOUNT_SOURCE: &str = "AION_FIXTURE_CLAUDE_DIR";

/// A resolved `[assistant]` section built through the REAL resolver, so what
/// these pins run against is what an operator's file would produce.
///
/// One catalogue harness with one account, which is the only thing the section
/// can still say. No command, no cwd, no timeouts: those are the catalogue's or
/// they are gone.
pub(crate) fn config() -> ResolvedAssistantConfig {
    AssistantConfig {
        harnesses: vec![AssistantHarnessConfig {
            name: Some(HARNESS.to_owned()),
            accounts: vec![AssistantAccountConfig {
                name: Some(ACCOUNT.to_owned()),
                env: [("CLAUDE_CONFIG_DIR".to_owned(), ACCOUNT_SOURCE.to_owned())]
                    .into_iter()
                    .collect(),
            }],
        }],
    }
    .resolved()
}

/// The endpoints a fixture server states.
pub(crate) fn endpoints() -> AssistantEndpoints {
    AssistantEndpoints {
        base: "http://127.0.0.1:9999".to_owned(),
        aion_mcp_enabled: false,
    }
}

/// A registry over a fresh in-memory store.
pub(crate) fn registry() -> (AssistantSessions, Arc<InMemoryStore>) {
    let store = Arc::new(InMemoryStore::default());
    let sessions = AssistantSessions::new(
        Arc::clone(&store) as Arc<dyn AssistantSessionStore>,
        config(),
        Some(endpoints()),
        CATALOGUE,
    );
    (sessions, store)
}

/// The catalogue every fixture registry resolves [`HARNESS`] through.
///
/// Its one entry keeps the shipped id, so records and refusals read as they do
/// in production, but it launches [`STUB_AGENT`]: a conforming ACP agent this
/// crate carries, which starts in milliseconds on any box with `python3`,
/// advertises `loadSession`, opens a fresh session, refuses every
/// `session/load` by name, and answers a prompt with one chunk and `end_turn`.
/// So a first turn reaches an agent and its frames land in the order the cells
/// pin, and a resume ATTEMPTED on a dormant session fails its spawn by name —
/// on every box, under any load. The shipped entry launches `npx`, and a unit
/// test with a Node launcher on its clock was red whenever the box was
/// compiling (2026-08-31, two batteries of three) and would refuse outright on
/// a box without `npx`. Availability itself is pinned against the shipped
/// catalogue in `launch_tests`, where the measurement belongs.
pub(crate) const CATALOGUE: &[CatalogueHarness] = &[CatalogueHarness {
    id: HARNESS,
    display_name: "Claude Code (fixture)",
    program: "python3",
    args: &[STUB_AGENT],
    install_hint: "the fixture harness needs `python3` on PATH",
}];

/// The stub agent's path, fixed at compile time from this crate's manifest
/// directory so the fixture resolves from whatever directory the tests run in.
const STUB_AGENT: &str = concat!(
    env!("CARGO_MANIFEST_DIR"),
    "/src/assistant/sessions/fixtures/stub_agent.py"
);

/// A registry over a store handed in, so a test can substitute a refusing one.
pub(crate) fn registry_over(store: Arc<dyn AssistantSessionStore>) -> AssistantSessions {
    AssistantSessions::new(store, config(), Some(endpoints()), CATALOGUE)
}

/// Open a session by writing its record, WITHOUT measuring the harness.
///
/// `create` refuses a harness whose launch program is not on this machine's
/// `PATH` — that is O4, and it is pinned where it belongs, at the create route,
/// with both arms decided by the measurement. Every OTHER pin in this suite is
/// about something that happens after a session exists (whether a dormant one
/// can be reopened, whether a failed append is broadcast, which session is
/// current), and none of them should turn red on a box that has no `npx`
/// installed. So the fixture opens a session the way the store sees one: a
/// record, and the same first settling frame `create` appends.
///
/// It mints no bearer, which is the honest reading of a session no spawn has
/// planned for yet — the cells that need one put a known digest on the record
/// themselves.
pub(crate) async fn open_session(
    sessions: &AssistantSessions,
) -> Result<aion_core::AssistantSessionSummary, String> {
    let session_id = AssistantSessionId::new_v4();
    let now = Utc::now();
    let record = AssistantSessionRecord {
        session_id,
        subject: OPERATOR.to_owned(),
        harness: HARNESS.to_owned(),
        account: None,
        title: None,
        created_at: now,
        updated_at: now,
        turns: 0,
        mcp_token_digest: None,
        commands: Vec::new(),
        config_options: Vec::new(),
    };
    sessions
        .store()
        .put_assistant_session(record.clone())
        .await
        .map_err(|error| error.to_string())?;
    sessions
        .settle(
            session_id,
            AssistantSessionState::Dormant,
            super::lifecycle::CREATED_AWAITING_FIRST_TURN,
        )
        .await
        .map_err(|error| error.to_string())?;
    Ok(record.summary(
        AssistantSessionState::Dormant,
        Some(super::lifecycle::CREATED_AWAITING_FIRST_TURN.to_owned()),
    ))
}

/// Record one session's opening, as a spawn would.
///
/// `load_session` is what the agent ACTUALLY advertised — the fact the whole
/// resume decision is gated on — so a fixture states it explicitly rather than
/// letting a default decide.
pub(crate) async fn opened(
    sessions: &AssistantSessions,
    session_id: AssistantSessionId,
    load_session: bool,
) -> Result<(), String> {
    sessions
        .recorder(session_id)
        .record(AssistantSessionEvent::SessionOpened {
            acp_session_ref: format!("acp-{session_id}"),
            load_session,
            at: Utc::now(),
            resumed: false,
        })
        .await
        .map(drop)
        .map_err(|error| error.to_string())
}

/// Settle a session dormant, as the boot sweep would after its process went.
pub(crate) async fn went_dormant(
    sessions: &AssistantSessions,
    session_id: AssistantSessionId,
) -> Result<(), String> {
    sessions
        .settle(
            session_id,
            AssistantSessionState::Dormant,
            super::lifecycle::PROCESS_EXITED_RESUMABLE,
        )
        .await
        .map_err(|error| error.to_string())
}

/// Share a context on a session, as a push would.
pub(crate) async fn shared_context(
    sessions: &AssistantSessions,
    session_id: AssistantSessionId,
    url: &str,
) -> Result<(), String> {
    sessions
        .push_context(
            OPERATOR,
            session_id,
            AssistantTurnContext {
                url: Some(url.to_owned()),
                concepts: Vec::new(),
                document: None,
            },
        )
        .await
        .map_err(|error| error.to_string())
}

/// A store whose transcript appends all REFUSE, and whose record reads work.
///
/// The instrument for the append-before-broadcast pin: a turn on this store must
/// fail and broadcast nothing, and the only way to see "nothing" is to have a
/// subscriber that would have seen something.
pub(crate) struct RefusingTranscriptStore {
    inner: InMemoryStore,
}

impl RefusingTranscriptStore {
    /// The message every refused append carries, so the pin can name it.
    pub(crate) const REFUSAL: &'static str =
        "this transcript store refuses every append (test instrument)";

    /// Build one over a fresh in-memory store.
    pub(crate) fn new() -> Self {
        Self {
            inner: InMemoryStore::default(),
        }
    }
}

#[async_trait]
impl AssistantSessionStore for RefusingTranscriptStore {
    async fn put_assistant_session(
        &self,
        record: AssistantSessionRecord,
    ) -> Result<(), StoreError> {
        self.inner.put_assistant_session(record).await
    }

    async fn get_assistant_session(
        &self,
        session_id: &AssistantSessionId,
    ) -> Result<Option<AssistantSessionRecord>, StoreError> {
        self.inner.get_assistant_session(session_id).await
    }

    async fn list_assistant_sessions(&self) -> Result<AssistantSessionListing, StoreError> {
        self.inner.list_assistant_sessions().await
    }

    /// The whole instrument: the durable append fails, so nothing may be
    /// broadcast and the caller must be told.
    async fn append_assistant_transcript_event(
        &self,
        _session_id: &AssistantSessionId,
        _recorded_at: DateTime<Utc>,
        _payload: aion_core::Payload,
    ) -> Result<u64, StoreError> {
        Err(StoreError::Backend(Self::REFUSAL.to_owned()))
    }

    async fn assistant_transcript_head(
        &self,
        session_id: &AssistantSessionId,
    ) -> Result<u64, StoreError> {
        self.inner.assistant_transcript_head(session_id).await
    }

    async fn assistant_transcript(
        &self,
        session_id: &AssistantSessionId,
        after: Option<u64>,
    ) -> Result<Vec<AssistantTranscriptEvent>, StoreError> {
        self.inner.assistant_transcript(session_id, after).await
    }

    async fn put_assistant_default_harness(
        &self,
        subject: &str,
        harness: &str,
    ) -> Result<(), StoreError> {
        self.inner
            .put_assistant_default_harness(subject, harness)
            .await
    }

    async fn assistant_default_harness(&self, subject: &str) -> Result<Option<String>, StoreError> {
        self.inner.assistant_default_harness(subject).await
    }
}