bamboo-storage 2026.10.21

//! Trusted singleton identity. Canonical Session files are the only authority;
//! the global index and unpublished staging directories never grant a role.

use super::root_lifetime::RootPublicationFault;
use super::*;
use bamboo_domain::{
    SessionAuthorityConflict, SessionAuthorityIdentity, SupervisorBootstrapReceipt,
    SupervisorManagementState, DEFAULT_SUPERVISOR_SESSION_ID,
};

fn invalid(message: &str) -> io::Error {
    io::Error::new(
        io::ErrorKind::InvalidData,
        format!("session_authority: {message}"),
    )
}

fn writer_conflict(error: io::Error) -> io::Error {
    io::Error::new(error.kind(), SessionAuthorityConflict(error.to_string()))
}

pub(super) fn validate_identity(session: &Session) -> io::Result<()> {
    if let SessionAuthorityIdentity::Supervisor { incarnation_id } = &session.authority_identity {
        if incarnation_id.is_nil()
            || session.id != DEFAULT_SUPERVISOR_SESSION_ID
            || session.kind != SessionKind::Root
            || session.root_session_id != session.id
            || session.parent_session_id.is_some()
            || session.spawn_depth != 0
        {
            return Err(invalid("invalid Supervisor identity or Root lineage"));
        }
    }
    validate_management(
        &session.authority_identity,
        session.supervisor_management.as_ref(),
    )?;
    Ok(())
}

fn validate_management(
    identity: &SessionAuthorityIdentity,
    state: Option<&SupervisorManagementState>,
) -> io::Result<()> {
    if let Some(state) = state {
        let SessionAuthorityIdentity::Supervisor { incarnation_id } = identity else {
            return Err(invalid(
                "Ordinary Sessions cannot hold Supervisor management state",
            ));
        };
        state.validate(*incarnation_id).map_err(invalid)?;
    }
    Ok(())
}

fn validate_management_overlay(
    main: Option<&SupervisorManagementState>,
    side: Option<&SupervisorManagementState>,
) -> io::Result<()> {
    if let Some(main) = main {
        let side = side.ok_or_else(|| invalid("runtime lost Supervisor management state"))?;
        if side.revision < main.revision || (side.revision == main.revision && side != main) {
            return Err(invalid(
                "Supervisor management overlay regressed or diverged",
            ));
        }
        if side.schema_version < main.schema_version {
            return Err(invalid("Supervisor management schema cannot downgrade"));
        }
        for (id, previous) in &main.links {
            let current = side
                .links
                .get(id)
                .ok_or_else(|| invalid("runtime lost a Supervisor link tombstone"))?;
            if current.revision < previous.revision
                || (current.revision == previous.revision && current != previous)
            {
                return Err(invalid("Supervisor link overlay regressed or diverged"));
            }
        }
        for (id, previous) in &main.observers {
            let current = side
                .observers
                .get(id)
                .ok_or_else(|| invalid("runtime lost an Observer tombstone"))?;
            if current.operation_id != previous.operation_id
                || current.parent_created_at != previous.parent_created_at
                || current.child_created_at != previous.child_created_at
                || current.target_links != previous.target_links
                || current.revision < previous.revision
                || (current.revision == previous.revision && current != previous)
                || (!previous.enabled && current.enabled)
            {
                return Err(invalid(
                    "Observer registration overlay regressed or diverged",
                ));
            }
        }
    }
    Ok(())
}

/// All overlay/recovery paths must reject a missing or divergent authority.
/// Ordinary legacy Sessions keep their existing compatibility semantics.
pub(super) fn validate_overlay(main: &Session, side: Option<&Session>) -> io::Result<()> {
    validate_identity(main)?;
    if let Some(side) = side {
        validate_identity(side)?;
    }
    let has_authority = !matches!(main.authority_identity, SessionAuthorityIdentity::Ordinary)
        || side.is_some_and(|side| {
            !matches!(side.authority_identity, SessionAuthorityIdentity::Ordinary)
        });
    if has_authority {
        let side =
            side.ok_or_else(|| invalid("published Supervisor is missing runtime authority"))?;
        if main.id != side.id || main.authority_identity != side.authority_identity {
            return Err(invalid(
                "runtime authority does not match the published Session",
            ));
        }
        validate_management_overlay(
            main.supervisor_management.as_ref(),
            side.supervisor_management.as_ref(),
        )?;
    }
    Ok(())
}

async fn real_directory(path: &Path) -> io::Result<bool> {
    match fs::symlink_metadata(path).await {
        Ok(metadata) if metadata.file_type().is_dir() => Ok(true),
        Ok(_) => Err(invalid("expected a real authority directory")),
        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(false),
        Err(error) => Err(error),
    }
}

async fn read_regular(path: &Path) -> io::Result<Vec<u8>> {
    let metadata = fs::symlink_metadata(path).await.map_err(|error| {
        if error.kind() == io::ErrorKind::NotFound {
            invalid("published Root has missing canonical files")
        } else {
            error
        }
    })?;
    if !metadata.file_type().is_file() {
        return Err(invalid("canonical authority is not a regular file"));
    }
    fs::read(path).await
}

// The strict generic authority port retains the legacy main-pair check for
// Ordinary Roots. Only the fixed default Supervisor has a separate bounded
// management proof that can replace that transcript-sized read.
#[derive(Deserialize)]
struct MainIdentity {
    id: String,
    created_at: DateTime<Utc>,
    #[serde(default)]
    kind: SessionKind,
    #[serde(default)]
    root_session_id: String,
    #[serde(default)]
    parent_session_id: Option<String>,
    #[serde(default)]
    spawn_depth: u32,
    #[serde(default)]
    authority_identity: SessionAuthorityIdentity,
    #[serde(default)]
    supervisor_management: Option<SupervisorManagementState>,
    #[serde(default)]
    root_orchestration_only: bool,
    #[serde(default)]
    root_tool_authority_revision: u64,
    #[serde(default)]
    root_mode_transition_epoch: u64,
    #[serde(default)]
    root_mode_operations: Vec<RootModeOperationReceipt>,
    #[serde(default)]
    conversation_reset: Option<bamboo_domain::ConversationResetState>,
}

impl SessionStoreV2 {
    /// Caller holds lifecycle/task/session read guards. Admission needs the
    /// strict canonical Root proof and full history; never repair its index.
    pub(super) async fn load_execution_admission_unchecked(
        &self,
        id: &str,
    ) -> io::Result<Option<Session>> {
        if !real_directory(&self.sessions_dir).await? {
            return Err(invalid("sessions directory is missing"));
        }
        let directory = self.sessions_dir.join(id);
        if !real_directory(&directory).await? {
            // Children retain the indexed, full-history reader. A genuinely
            // absent ordinary Session must still allow its first execution.
            return self.load_session_unlocked(id).await;
        }
        let Some(side) = self.load_root_authority_unchecked(id).await? else {
            return Ok(None);
        };
        let main_bytes = read_regular(&directory.join("session.json")).await?;
        compact_main::validate_full_main(&main_bytes)?;
        let main: Session = serde_json::from_slice(&main_bytes)
            .map_err(|_| invalid("invalid canonical session.json"))?;
        main.check_conversation_generation(&side)
            .map_err(|error| invalid(&error.to_string()))?;
        let main_root = if main.root_session_id.is_empty() {
            id
        } else {
            &main.root_session_id
        };
        if main.id != id
            || main.kind != SessionKind::Root
            || main_root != id
            || main.parent_session_id.is_some()
            || main.spawn_depth != 0
            || main.created_at != side.created_at
            || main.authority_identity != side.authority_identity
            || (main.root_orchestration_only && main.root_tool_authority_revision == 0)
            || side.root_tool_authority_revision < main.root_tool_authority_revision
            || (side.root_tool_authority_revision == main.root_tool_authority_revision
                && side.root_orchestration_only != main.root_orchestration_only)
            || side.root_mode_transition_epoch != main.root_mode_transition_epoch
            || side.root_mode_operations != main.root_mode_operations
        {
            return Err(invalid("canonical Root identity mismatch"));
        }
        validate_overlay(&main, Some(&side))?;
        self.validate_root_tool_authority_overlay(id, &main, Some(&side))
            .await?;
        // overlay_runtime_sidecar preserves main messages and provider history,
        // while taking current control state from the proven runtime sidecar.
        let mut full = overlay_runtime_sidecar(main, Some(side));
        full.clear_stale_root_token_budget();
        Ok(Some(full))
    }

    /// Caller owns the guards for its storage operation. Never consult the
    /// rebuildable index, and never recover a runtime authority from main.
    pub(super) async fn load_root_authority_unchecked(
        &self,
        id: &str,
    ) -> io::Result<Option<Session>> {
        validate_session_id(id)?;
        // The default Supervisor can reuse its fixed ID after revocation.
        // Its birth check must use the bounded committed proof below; the
        // ordinary lifetime helper scans the transcript-sized main file.
        if id != DEFAULT_SUPERVISOR_SESSION_ID && self.root_directory_is_revoked(id).await? {
            return Ok(None);
        }
        if !real_directory(&self.sessions_dir).await? {
            return Err(invalid("sessions directory is missing"));
        }
        let directory = self.sessions_dir.join(id);
        if !real_directory(&directory).await? {
            return Ok(None);
        }
        let mut side: Session =
            serde_json::from_slice(&read_regular(&directory.join(RUNTIME_SIDECAR_FILE)).await?)
                .map_err(|_| invalid("invalid canonical runtime.json"))?;
        validate_identity(&side)?;
        side.validate_conversation_state()
            .map_err(|error| invalid(&error.to_string()))?;
        let side_root = if matches!(side.authority_identity, SessionAuthorityIdentity::Ordinary)
            && side.root_session_id.is_empty()
        {
            id
        } else {
            &side.root_session_id
        };
        if side.id != id
            || side.kind != SessionKind::Root
            || side_root != id
            || side.parent_session_id.is_some()
            || side.spawn_depth != 0
            || (side.root_orchestration_only && side.root_tool_authority_revision == 0)
        {
            return Err(invalid("canonical Root identity mismatch"));
        }
        self.validate_root_tool_authority_against_proof(id, &side)
            .await?;
        self.validate_supervisor_proof(&side).await?;
        if !matches!(
            side.authority_identity,
            SessionAuthorityIdentity::Supervisor { .. }
        ) {
            let main_bytes = read_regular(&directory.join("session.json")).await?;
            compact_main::validate_full_main(&main_bytes)?;
            let main: MainIdentity = serde_json::from_slice(&main_bytes)
                .map_err(|_| invalid("invalid canonical session.json"))?;
            validate_management(
                &main.authority_identity,
                main.supervisor_management.as_ref(),
            )?;
            let main_root = if main.root_session_id.is_empty() {
                id
            } else {
                &main.root_session_id
            };
            if main.id != id
                || main.kind != SessionKind::Root
                || main_root != id
                || main.parent_session_id.is_some()
                || main.spawn_depth != 0
                || main.authority_identity != side.authority_identity
                || main.created_at != side.created_at
                || (main.root_orchestration_only && main.root_tool_authority_revision == 0)
                || side.root_tool_authority_revision < main.root_tool_authority_revision
                || (side.root_tool_authority_revision == main.root_tool_authority_revision
                    && side.root_orchestration_only != main.root_orchestration_only)
                || side.root_mode_transition_epoch != main.root_mode_transition_epoch
                || side.root_mode_operations != main.root_mode_operations
                || side.conversation_reset != main.conversation_reset
            {
                return Err(invalid("canonical Root identity mismatch"));
            }
        }
        if id == DEFAULT_SUPERVISOR_SESSION_ID
            && self
                .root_revocation(id)
                .await?
                .is_some_and(|cutoff| side.created_at <= cutoff)
        {
            return Ok(None);
        }
        side.root_session_id = id.to_string();
        side.messages.clear();
        side.clear_stale_root_token_budget();
        Ok(Some(side))
    }

    /// The only code path allowed to assign Supervisor authority. The fixed ID
    /// is protected by the same per-session file lock as ordinary writers.
    pub(super) async fn bootstrap_default_supervisor(
        &self,
        initial_model: &str,
    ) -> io::Result<SupervisorBootstrapReceipt> {
        let _lifecycle = self.lock_session_lifecycle_exclusive().await?;
        let _task = self.lock_runtime_task_transaction_exclusive().await?;
        self.recover_all_runtime_task_transactions_locked(&_task)
            .await?;
        self.recover_all_session_copy_transactions_locked().await?;
        let _session = self
            .acquire_session_maintenance_lock(DEFAULT_SUPERVISOR_SESSION_ID)
            .await?;
        let existing = self
            .load_root_authority_unchecked(DEFAULT_SUPERVISOR_SESSION_ID)
            .await;
        if existing.is_err() {
            // A legacy Ordinary owner is a conflict even without a sidecar.
            // This classification never repairs or grants authority, and the
            // strict authority port still rejects the incomplete file pair.
            let directory = self.sessions_dir.join(DEFAULT_SUPERVISOR_SESSION_ID);
            if matches!(fs::symlink_metadata(directory.join(RUNTIME_SIDECAR_FILE)).await,
                Err(error) if error.kind() == io::ErrorKind::NotFound)
            {
                if let Ok(bytes) = read_regular(&directory.join("session.json")).await {
                    compact_main::validate_full_main(&bytes)?;
                    if let Ok(main) = serde_json::from_slice::<Session>(&bytes) {
                        if main.id == DEFAULT_SUPERVISOR_SESSION_ID
                            && main.kind == SessionKind::Root
                            && main.parent_session_id.is_none()
                            && main.spawn_depth == 0
                            && (main.root_session_id.is_empty() || main.root_session_id == main.id)
                            && main.authority_identity.is_ordinary()
                        {
                            return Err(io::Error::new(
                                io::ErrorKind::AlreadyExists,
                                "default Supervisor ID is occupied by a legacy Ordinary Session",
                            ));
                        }
                    }
                }
            }
        }
        if let Some(existing) = existing? {
            let SessionAuthorityIdentity::Supervisor { incarnation_id } =
                existing.authority_identity
            else {
                return Err(io::Error::new(
                    io::ErrorKind::AlreadyExists,
                    "default Supervisor ID is occupied by an Ordinary Session",
                ));
            };
            // Authority is already strictly verified. The rebuildable index
            // still needs the real transcript count, never the empty CP view.
            let full = self
                .load_authoritative_root_session(DEFAULT_SUPERVISOR_SESSION_ID)
                .await?
                .ok_or_else(|| invalid("published Supervisor disappeared"))?;
            self.repair_index_from_authoritative_session(
                &full,
                Self::root_rel_path(DEFAULT_SUPERVISOR_SESSION_ID),
            )
            .await?;
            return Ok(SupervisorBootstrapReceipt {
                session_id: existing.id,
                incarnation_id,
                created: false,
            });
        }
        // The ID namespace also includes children. Only cold bootstrap scans
        // their canonical placements; the rebuildable index cannot prove absence.
        let mut roots = fs::read_dir(&self.sessions_dir).await?;
        while let Some(root) = roots.next_entry().await? {
            let kind = root.file_type().await?;
            if kind.is_file() {
                continue;
            }
            if !kind.is_dir() {
                return Err(invalid("non-directory canonical Root placement"));
            }
            let children = root.path().join("children");
            if real_directory(&children).await?
                && real_directory(&children.join(DEFAULT_SUPERVISOR_SESSION_ID)).await?
            {
                return Err(io::Error::new(
                    io::ErrorKind::AlreadyExists,
                    "default Supervisor ID is occupied by a Child placement",
                ));
            }
        }
        // Identity exists before model setup or the first message. An empty
        // model leaves this idle Session unbound; normal chat admission still
        // resolves/validates the configured model before it can execute.
        let incarnation_id = Uuid::new_v4();
        let mut session = Session::new(DEFAULT_SUPERVISOR_SESSION_ID, initial_model.trim());
        session.created_at = self.fresh_root_birth(DEFAULT_SUPERVISOR_SESSION_ID).await?;
        session.updated_at = session.created_at;
        session.title = "Supervisor".to_string();
        session.title_generated = true;
        session.authority_identity = SessionAuthorityIdentity::Supervisor { incarnation_id };
        let main_bytes = compact_main::serialize_main(&session)?;
        let runtime_bytes = serde_json::to_vec_pretty(&session).map_err(io::Error::other)?;
        let staging = self
            .bamboo_home_dir
            .join(format!(".supervisor-bootstrap-{}", Uuid::new_v4()));
        let destination = self.sessions_dir.join(DEFAULT_SUPERVISOR_SESSION_ID);
        self.remove_revoked_root_directory(DEFAULT_SUPERVISOR_SESSION_ID)
            .await?;
        fs::create_dir(&staging).await?;
        let result = async {
            fs::create_dir(staging.join("children")).await?;
            fs::create_dir(staging.join("attachments")).await?;
            durable_atomic_write(&staging.join("session.json"), &main_bytes).await?;
            durable_atomic_write(&staging.join(RUNTIME_SIDECAR_FILE), &runtime_bytes).await?;
            Self::stage_actor_tree_revision(&staging, &session).await?;
            Self::write_staged_root_tool_proof(&staging, &session).await?;
            Self::write_staged_supervisor_proof(&staging, &session).await?;
            sync_directory(&staging).await?;
            self.maybe_fail_root_publication(RootPublicationFault::BeforePublish)?;
            atomic_rename(&staging, &destination).await?;
            sync_parent_directory_entry(&staging).await?;
            sync_parent_directory_entry(&destination).await?;
            self.maybe_fail_root_publication(RootPublicationFault::BeforeIndex)?;
            self.repair_index_from_authoritative_session(
                &session,
                Self::root_rel_path(DEFAULT_SUPERVISOR_SESSION_ID),
            )
            .await?;
            Ok(SupervisorBootstrapReceipt {
                session_id: session.id.clone(),
                incarnation_id,
                created: true,
            })
        }
        .await;
        if result.is_err() {
            // Only this call's unpublished directory is removable. A complete
            // published Root survives index failures and is repaired on retry.
            let _ = fs::remove_dir_all(&staging).await;
        }
        result
    }

    /// Called inside the final cross-process write lock. Merging callers must
    /// adopt durable authority in their own snapshot before entering this writer;
    /// silently repairing only the serialized copy would leave their cache stale.
    pub(super) async fn validate_authority_for_save(&self, incoming: &Session) -> io::Result<()> {
        validate_identity(incoming).map_err(writer_conflict)?;
        if incoming.id != DEFAULT_SUPERVISOR_SESSION_ID {
            return Ok(());
        }
        let current = self
            .load_root_authority_unchecked(&incoming.id)
            .await
            .map_err(writer_conflict)?;
        match current {
            Some(current) if current.authority_identity == incoming.authority_identity => Ok(()),
            Some(_) => Err(writer_conflict(io::Error::new(
                io::ErrorKind::PermissionDenied,
                "writer identity does not match the durable Session; reload before saving",
            ))),
            None if matches!(
                incoming.authority_identity,
                SessionAuthorityIdentity::Ordinary
            ) =>
            {
                Ok(())
            }
            None => Err(writer_conflict(io::Error::new(
                io::ErrorKind::PermissionDenied,
                "only trusted bootstrap can create Supervisor authority",
            ))),
        }
    }
}