//! Session orchestration for `Archivist`.
//!
//! Covers registration (with alias detection), resolution, mandatory-primitive
//! dispatch, read-modify-write metadata wrappers, and move/copy semantics.
//! Ported from the `FileBasedArchivist` bodies in `archivist.rs`; persistence
//! routes through backend sub-traits (`SessionMappingBackend`,
//! `ConnectorRegistryBackend`) plus the mandatory `ArchiveBackend` surface.

use chrono::Utc;
use uuid::Uuid;

use crate::backend::ArchiveCapability;
use crate::coordinator::Archivist;
use crate::error::{ArchivistError, Result};
use crate::types::{
    MessageCursor, MessagePage, MessageRecord, MoveReport, RegisterSessionRequest,
    RegisterSessionResponse, RegisterStatus, SessionCompleteness, SessionListQuery,
    SessionMetadata, SessionPage, MAX_PAGE_LIMIT,
};

impl Archivist {
    // ------------------------------------------------------------------
    // Filter helpers (Task 20)
    // ------------------------------------------------------------------

    /// Evaluate a registration's filter against optional session metadata.
    ///
    /// - If the filter is unrestricted, always allows.
    /// - If the filter is restricted and metadata is present, delegates to
    ///   `ArchiveFilter::allows`.
    /// - If the filter is restricted but metadata is unavailable, rejects
    ///   (the safe default: without metadata we cannot prove the session
    ///   should be replicated).
    pub(crate) fn filter_allows(
        reg: &crate::registry::ArchiveRegistration,
        session: Option<&SessionMetadata>,
    ) -> bool {
        if reg.filter.is_unrestricted() {
            return true;
        }
        match session {
            Some(s) => reg.filter.allows(s, &s.connector_uid),
            None => false,
        }
    }

    /// Lazily load session metadata for filter evaluation.
    ///
    /// Returns `None` when none of the non-primary registrations carry a
    /// non-unrestricted filter (no load needed) or when the load itself
    /// failed (e.g. the session isn't present anywhere yet).
    pub(crate) async fn load_metadata_for_filter(
        &self,
        scroll_id: Uuid,
        regs: &[std::sync::Arc<crate::registry::ArchiveRegistration>],
        primary_name: &str,
    ) -> Option<SessionMetadata> {
        let any_restricted = regs.iter().any(|r| {
            r.name != primary_name
                && r.enabled
                && r.write_active
                && !r.filter.is_unrestricted()
        });
        if !any_restricted {
            return None;
        }
        self.get_session_metadata(scroll_id, None).await.ok()
    }

    // ------------------------------------------------------------------
    // Registration & alias detection
    // ------------------------------------------------------------------

    /// Register a session with alias detection.
    ///
    /// Ported from `FileBasedArchivist::register_session`. Persistence routes
    /// through the backend's `SessionMappingBackend::put_mapping` and
    /// `ArchiveBackend::put_session` methods; alias detection
    /// (Accepted vs. Aliased) stays at the coordinator.
    pub async fn register_session(
        &self,
        req: RegisterSessionRequest,
        archive: Option<String>,
    ) -> Result<RegisterSessionResponse> {
        let primary = self.resolve_primary(archive.clone()).await?;
        let regs: Vec<std::sync::Arc<crate::registry::ArchiveRegistration>> =
            self.registrations.read().await.clone();

        let primary_mapping = primary.backend.as_session_mapping().ok_or_else(|| {
            ArchivistError::PrimaryUnavailable {
                name: primary.name.clone(),
                reason: "backend lacks SessionMapping capability".into(),
            }
        })?;
        let primary_registry = primary.backend.as_connector_registry().ok_or_else(|| {
            ArchivistError::PrimaryUnavailable {
                name: primary.name.clone(),
                reason: "backend lacks ConnectorRegistry capability".into(),
            }
        })?;

        // Generate scroll_id (use custom_scroll_id only if it's UUID7, otherwise generate new)
        // This validation prevents UUID4 leaks into folder names.
        let scroll_id = match req.custom_scroll_id {
            Some(uuid) if uuid.get_version_num() == 7 => {
                tracing::debug!("Using provided UUID7 as scroll_id: {}", uuid);
                uuid
            }
            Some(uuid) => {
                tracing::warn!(
                    "Rejected non-UUID7 custom_scroll_id: {} (version {}), generating fresh UUID7",
                    uuid,
                    uuid.get_version_num()
                );
                Uuid::now_v7()
            }
            None => Uuid::now_v7(),
        };

        // Validate that the connector exists on the PRIMARY.
        if primary_registry
            .get_connector(req.connector_uid)
            .await?
            .is_none()
        {
            return Err(ArchivistError::ConnectorUnknown(req.connector_uid));
        }

        // Alias detection: if a mapping already exists for this
        // (connector_uid, native_session_id) on the PRIMARY, return ALIASED.
        // Alias detection stays on the primary only — the answer is canonical
        // and must not be per-backend.
        if let Some(existing_scroll) = primary_mapping
            .get_mapping(req.connector_uid, &req.native_session_id)
            .await?
        {
            return Ok(RegisterSessionResponse {
                status: RegisterStatus::Aliased,
                scroll_id: existing_scroll,
                alias_of: Some(existing_scroll),
            });
        }

        // ACCEPTED: write mapping first, then the session metadata. This order
        // matches the original `FileBasedArchivist` sequence — if metadata
        // creation fails, the mapping still lets `resolve_session` work, and
        // the next `append_messages` reconstructs a minimal session.json.
        let now = Utc::now();

        // Write mapping on PRIMARY first. Any failure here propagates — the
        // canonical mapping must succeed for the operation to be meaningful.
        if let Err(e) = primary_mapping
            .put_mapping(req.connector_uid, &req.native_session_id, scroll_id)
            .await
        {
            self.record_write_failure(&primary, &format!("{e}")).await;
            return Err(e);
        }
        self.record_write_success(&primary).await;

        let session_metadata = SessionMetadata {
            version: 1,
            scroll_id,
            created_at: now,
            updated_at: now,
            title: req.title.clone(),
            connector_uid: req.connector_uid,
            native_session_id: Some(req.native_session_id.clone()),
            agent_id: req.agent_id.clone(),
            parent_scroll_id: req.parent_scroll_id,
            continuation: req.continuation,
            tags: Vec::new(),
            metadata: req.metadata.clone(),
            no_update: false,
            kind: crate::types::SessionKind::Chat,
            acp_client_id: None,
            is_connected: None,
            current_session_id: None,
            models: None,
            modes: None,
            config_options: None,
            completeness: req.completeness,
            matrix_room_id: None,
            matrix_sharing_active: false,
            matrix_shared_at: None,
            is_subagent: req.is_subagent,
            subagent_type: req.subagent_type.clone(),
            spawning_tool_use_id: req.spawning_tool_use_id.clone(),
        };

        // Mirror the original best-effort semantics: if writing session.json
        // fails, log and proceed — the mapping is already durable.
        match &primary.write_policy {
            crate::registry::WritePolicy::Inline => {
                if let Err(e) = primary.backend.put_session(session_metadata.clone()).await {
                    tracing::warn!(
                        scroll_id = %scroll_id,
                        native_session_id = %req.native_session_id,
                        connector_uid = %req.connector_uid,
                        error = %e,
                        "Failed to write session metadata after mapping write. \
                         Session is registered but metadata will be created on first message write."
                    );
                    self.record_write_failure(&primary, &format!("{e}")).await;
                } else {
                    self.record_write_success(&primary).await;
                }
            }
            crate::registry::WritePolicy::Queued { .. } => {
                let writer = primary
                    .writer
                    .as_ref()
                    .expect("queued policy implies writer handle present");
                // Queued put_session is fire-and-forget; errors drift health
                // on the writer task side.
                let _ = writer
                    .enqueue(crate::registry::writer::WriteOp::PutSession(
                        session_metadata.clone(),
                    ))
                    .await;
            }
        }

        // Secondaries: fan out mapping + session with capability filter.
        for reg in regs.iter() {
            if reg.name == primary.name {
                continue;
            }
            if !reg.enabled || !reg.write_active {
                continue;
            }

            // Per-archive include/exclude filter. Checked before capability so
            // cheaper checks come first.
            if !reg.filter.allows(&session_metadata, &session_metadata.connector_uid) {
                tracing::debug!(
                    archive = %reg.name,
                    scroll_id = %scroll_id,
                    op = "register_session",
                    "filter_skip"
                );
                continue;
            }

            // Mapping on secondary (capability check). Mapping writes have no
            // `WriteOp` variant (SessionMapping is a sub-trait, not part of the
            // queued dispatch surface), so they remain inline in both policies.
            let Some(sec_mapping) = reg.backend.as_session_mapping() else {
                tracing::debug!(
                    backend = reg.name.as_str(),
                    type_name = reg.type_name,
                    op = "register_session:mapping",
                    "capability_skip"
                );
                continue;
            };
            if let Err(e) = sec_mapping
                .put_mapping(req.connector_uid, &req.native_session_id, scroll_id)
                .await
            {
                self.record_write_failure(reg, &format!("{e}")).await;
                if reg.failure_mode == crate::registry::FailureMode::Required {
                    return Err(e);
                }
                continue;
            }

            // put_session on secondary — honours the secondary's write_policy.
            match &reg.write_policy {
                crate::registry::WritePolicy::Inline => {
                    if let Err(e) = reg.backend.put_session(session_metadata.clone()).await {
                        self.record_write_failure(reg, &format!("{e}")).await;
                        if reg.failure_mode == crate::registry::FailureMode::Required {
                            return Err(e);
                        }
                    } else {
                        self.record_write_success(reg).await;
                    }
                }
                crate::registry::WritePolicy::Queued { .. } => {
                    let writer = reg
                        .writer
                        .as_ref()
                        .expect("queued policy implies writer handle present");
                    if let Err(e) = writer
                        .enqueue(crate::registry::writer::WriteOp::PutSession(
                            session_metadata.clone(),
                        ))
                        .await
                    {
                        if reg.failure_mode == crate::registry::FailureMode::Required {
                            return Err(e);
                        }
                    }
                }
            }
        }

        Ok(RegisterSessionResponse {
            status: RegisterStatus::Accepted,
            scroll_id,
            alias_of: None,
        })
    }

    /// Resolve a native session ID into a scroll_id via the mapping table.
    ///
    /// A missing mapping is surfaced as `SessionUnknown(Uuid::nil())` to keep
    /// the legacy `Archivist::resolve_session` contract (caller expects an
    /// error, not `Ok(None)`).
    pub async fn resolve_session(
        &self,
        connector_uid: Uuid,
        native_session_id: &str,
        archive: Option<String>,
    ) -> Result<Uuid> {
        let backend = self.resolve_backend(archive).await?;
        let mapping = backend.as_session_mapping().ok_or_else(|| {
            ArchivistError::CapabilityNotSupported {
                capability: ArchiveCapability::SessionMapping,
                backend: "selected".into(),
            }
        })?;

        match mapping.get_mapping(connector_uid, native_session_id).await? {
            Some(scroll_id) => Ok(scroll_id),
            None => Err(ArchivistError::SessionUnknown(Uuid::nil())),
        }
    }

    /// Locate the `(connector_uid, scroll_id)` that owns a native session id.
    pub async fn find_session_owner(
        &self,
        native_session_id: &str,
        archive: Option<String>,
    ) -> Result<Option<(Uuid, Uuid)>> {
        let backend = self.resolve_backend(archive).await?;
        let mapping = backend.as_session_mapping().ok_or_else(|| {
            ArchivistError::CapabilityNotSupported {
                capability: ArchiveCapability::SessionMapping,
                backend: "selected".into(),
            }
        })?;
        mapping.find_owner(native_session_id).await
    }

    // ------------------------------------------------------------------
    // Mandatory primitive dispatch
    // ------------------------------------------------------------------

    /// Fetch session metadata. `None` from the backend becomes
    /// `SessionUnknown` so callers can treat a missing session as an error.
    pub async fn get_session_metadata(
        &self,
        scroll_id: Uuid,
        _archive: Option<String>,
    ) -> Result<SessionMetadata> {
        // `archive` is now ignored for reads; routing picks the highest-priority
        // backend that has the session.
        self.read_walk_per_session(
            scroll_id,
            |_reg| true,
            |backend| async move { backend.get_session(scroll_id).await },
        )
        .await?
        .ok_or(ArchivistError::SessionUnknown(scroll_id))
    }

    /// List sessions with cursor pagination. The `archive` field on the query
    /// selects the backend; it is consumed by the coordinator and not passed
    /// on to the backend implementation.
    pub async fn list_sessions_paged(
        &self,
        mut query: SessionListQuery,
    ) -> Result<SessionPage> {
        let archive = query.archive.take();
        let backend = self.resolve_backend(archive).await?;
        backend.list_sessions_paged(query).await
    }

    pub async fn append_messages(
        &self,
        scroll_id: Uuid,
        messages: Vec<MessageRecord>,
        archive: Option<String>,
    ) -> Result<()> {
        let primary = self.resolve_primary(archive.clone()).await?;
        let regs: Vec<std::sync::Arc<crate::registry::ArchiveRegistration>> =
            self.registrations.read().await.clone();

        match &primary.write_policy {
            crate::registry::WritePolicy::Inline => {
                if let Err(e) = primary
                    .backend
                    .append_messages(scroll_id, messages.clone())
                    .await
                {
                    self.record_write_failure(&primary, &format!("{e}")).await;
                    return Err(e);
                }
                self.record_write_success(&primary).await;
            }
            crate::registry::WritePolicy::Queued { .. } => {
                let writer = primary
                    .writer
                    .as_ref()
                    .expect("queued policy implies writer handle present");
                writer
                    .enqueue(crate::registry::writer::WriteOp::AppendMessages {
                        scroll_id,
                        msgs: messages.clone(),
                    })
                    .await?;
            }
        }

        // Load metadata once for filter checks. Only needed if at least one
        // non-primary secondary has a non-unrestricted filter.
        let session_metadata_for_filter = self
            .load_metadata_for_filter(scroll_id, &regs, &primary.name)
            .await;

        for reg in regs.iter() {
            if reg.name == primary.name {
                continue;
            }
            if !reg.enabled || !reg.write_active {
                continue;
            }
            if !Self::filter_allows(reg, session_metadata_for_filter.as_ref()) {
                tracing::debug!(
                    archive = %reg.name,
                    scroll_id = %scroll_id,
                    op = "append_messages",
                    "filter_skip"
                );
                continue;
            }
            match &reg.write_policy {
                crate::registry::WritePolicy::Inline => {
                    if let Err(e) = reg.backend.append_messages(scroll_id, messages.clone()).await
                    {
                        self.record_write_failure(reg, &format!("{e}")).await;
                        if reg.failure_mode == crate::registry::FailureMode::Required {
                            return Err(e);
                        }
                    } else {
                        self.record_write_success(reg).await;
                    }
                }
                crate::registry::WritePolicy::Queued { .. } => {
                    let writer = reg
                        .writer
                        .as_ref()
                        .expect("queued policy implies writer handle present");
                    if let Err(e) = writer
                        .enqueue(crate::registry::writer::WriteOp::AppendMessages {
                            scroll_id,
                            msgs: messages.clone(),
                        })
                        .await
                    {
                        if reg.failure_mode == crate::registry::FailureMode::Required {
                            return Err(e);
                        }
                    }
                }
            }
        }
        Ok(())
    }

    pub async fn get_messages_paged(
        &self,
        scroll_id: Uuid,
        cursor: Option<MessageCursor>,
        limit: usize,
        _archive: Option<String>,
    ) -> Result<MessagePage> {
        // `archive` is now ignored for reads; routing picks the highest-priority
        // backend that has the session.
        Ok(self
            .read_walk_per_session(
                scroll_id,
                |_reg| true,
                |backend| {
                    let cursor = cursor.clone();
                    async move {
                        backend
                            .get_messages_paged(scroll_id, cursor, limit)
                            .await
                            .map(Some)
                    }
                },
            )
            .await?
            .unwrap_or(MessagePage {
                items: Vec::new(),
                next_cursor: None,
            }))
    }

    /// Fetch all messages for a session by draining every cursor page.
    ///
    /// This is a convenience wrapper over `get_messages_paged` that reconstructs
    /// the flat `Vec` shape the pre-Phase-2 `Archivist::get_messages` trait
    /// method promised. Consumers that know they want to stream large histories
    /// should prefer `get_messages_paged` directly.
    pub async fn get_messages(
        &self,
        scroll_id: Uuid,
        archive: Option<String>,
    ) -> Result<Vec<MessageRecord>> {
        let mut out = Vec::new();
        let mut cursor = None;
        loop {
            let page = self
                .get_messages_paged(
                    scroll_id,
                    cursor,
                    crate::types::MAX_PAGE_LIMIT,
                    archive.clone(),
                )
                .await?;
            out.extend(page.items);
            match page.next_cursor {
                Some(c) => cursor = Some(c),
                None => return Ok(out),
            }
        }
    }

    /// Return a slice of messages by offset/limit (sorted chronologically).
    ///
    /// Ported from `FileBasedArchivist::get_messages_range`. This loads the
    /// full message set (via `get_messages`) and slices it — it is NOT cursor-
    /// based and should only be used for small offsets / tests. For anything
    /// production-sized, use `get_messages_paged`.
    pub async fn get_messages_range(
        &self,
        scroll_id: Uuid,
        offset: usize,
        limit: usize,
        archive: Option<String>,
    ) -> Result<Vec<MessageRecord>> {
        let all = self.get_messages(scroll_id, archive).await?;
        Ok(all.into_iter().skip(offset).take(limit).collect())
    }

    pub async fn count_messages(
        &self,
        scroll_id: Uuid,
        _archive: Option<String>,
    ) -> Result<usize> {
        // `archive` is now ignored for reads; routing picks the highest-priority
        // backend that has the session.
        Ok(self
            .read_walk_per_session(
                scroll_id,
                |_reg| true,
                |backend| async move { backend.count_messages(scroll_id).await.map(Some) },
            )
            .await?
            .unwrap_or(0))
    }

    pub async fn delete_session(
        &self,
        scroll_id: Uuid,
        _archive: Option<String>, // ignored: delete spans every backend that has the session
    ) -> Result<()> {
        let regs: Vec<std::sync::Arc<crate::registry::ArchiveRegistration>> =
            self.registrations.read().await.clone();

        let mut read_only_violations: Vec<String> = Vec::new();
        let mut last_required_error: Option<ArchivistError> = None;

        for reg in regs.iter() {
            if !reg.enabled {
                continue;
            }

            // Cheap existence check before attempting delete. We treat read failures
            // here as "backend doesn't have it" to avoid cascading errors.
            let exists = reg
                .backend
                .get_session(scroll_id)
                .await
                .ok()
                .flatten()
                .is_some();
            if !exists {
                continue;
            }

            if !reg.write_active {
                read_only_violations.push(reg.name.clone());
                continue;
            }

            match &reg.write_policy {
                crate::registry::WritePolicy::Inline => {
                    match reg.backend.delete_session(scroll_id).await {
                        Ok(()) => self.record_write_success(reg).await,
                        Err(e) => {
                            self.record_write_failure(reg, &format!("{e}")).await;
                            if reg.failure_mode == crate::registry::FailureMode::Required
                                && last_required_error.is_none()
                            {
                                last_required_error = Some(e);
                            }
                        }
                    }
                }
                crate::registry::WritePolicy::Queued { .. } => {
                    if let Some(writer) = reg.writer.as_ref() {
                        let _ = writer
                            .enqueue(crate::registry::writer::WriteOp::DeleteSession { scroll_id })
                            .await;
                    }
                }
            }
        }

        // Cache invalidation regardless of outcome — the session no longer has a stable home.
        self.read_cache.invalidate(scroll_id).await;

        if let Some(e) = last_required_error {
            return Err(e);
        }
        if let Some(name) = read_only_violations.into_iter().next() {
            return Err(ArchivistError::DeleteOnReadOnlyBackend {
                backend: name,
                scroll_id,
            });
        }
        Ok(())
    }

    pub async fn clear_session_messages(
        &self,
        scroll_id: Uuid,
        archive: Option<String>,
    ) -> Result<()> {
        let primary = self.resolve_primary(archive.clone()).await?;
        let regs: Vec<std::sync::Arc<crate::registry::ArchiveRegistration>> =
            self.registrations.read().await.clone();

        match &primary.write_policy {
            crate::registry::WritePolicy::Inline => {
                if let Err(e) = primary.backend.clear_session_messages(scroll_id).await {
                    self.record_write_failure(&primary, &format!("{e}")).await;
                    return Err(e);
                }
                self.record_write_success(&primary).await;
            }
            crate::registry::WritePolicy::Queued { .. } => {
                let writer = primary
                    .writer
                    .as_ref()
                    .expect("queued policy implies writer handle present");
                writer
                    .enqueue(crate::registry::writer::WriteOp::ClearSessionMessages {
                        scroll_id,
                    })
                    .await?;
            }
        }

        let session_metadata_for_filter = self
            .load_metadata_for_filter(scroll_id, &regs, &primary.name)
            .await;

        for reg in regs.iter() {
            if reg.name == primary.name {
                continue;
            }
            if !reg.enabled || !reg.write_active {
                continue;
            }
            if !Self::filter_allows(reg, session_metadata_for_filter.as_ref()) {
                tracing::debug!(
                    archive = %reg.name,
                    scroll_id = %scroll_id,
                    op = "clear_session_messages",
                    "filter_skip"
                );
                continue;
            }
            match &reg.write_policy {
                crate::registry::WritePolicy::Inline => {
                    if let Err(e) = reg.backend.clear_session_messages(scroll_id).await {
                        self.record_write_failure(reg, &format!("{e}")).await;
                        if reg.failure_mode == crate::registry::FailureMode::Required {
                            return Err(e);
                        }
                    } else {
                        self.record_write_success(reg).await;
                    }
                }
                crate::registry::WritePolicy::Queued { .. } => {
                    let writer = reg
                        .writer
                        .as_ref()
                        .expect("queued policy implies writer handle present");
                    if let Err(e) = writer
                        .enqueue(crate::registry::writer::WriteOp::ClearSessionMessages {
                            scroll_id,
                        })
                        .await
                    {
                        if reg.failure_mode == crate::registry::FailureMode::Required {
                            return Err(e);
                        }
                    }
                }
            }
        }
        Ok(())
    }

    // ------------------------------------------------------------------
    // Read-modify-write metadata wrappers
    // ------------------------------------------------------------------

    /// Update `title` and/or `model` on an existing session.
    ///
    /// Ported from `FileBasedArchivist::update_session_metadata`. `title` is
    /// set directly; `model` is written into `metadata.model` (JSON object).
    /// `updated_at` is always bumped.
    ///
    /// NOTE: Read-mutate-write operations silently fall through to inline
    /// even under `WritePolicy::Queued`. There is no `WriteOp::UpdateSessionMetadata`
    /// variant because RMW doesn't compose with batching — the read-side of the
    /// RMW must see a consistent, already-persisted session.
    pub async fn update_session_metadata(
        &self,
        scroll_id: Uuid,
        title: Option<String>,
        model: Option<String>,
        archive: Option<String>,
    ) -> Result<()> {
        let primary = self.resolve_primary(archive.clone()).await?;
        let regs: Vec<std::sync::Arc<crate::registry::ArchiveRegistration>> =
            self.registrations.read().await.clone();

        let apply = |backend: std::sync::Arc<dyn crate::backend::ArchiveBackend>,
                     title: Option<String>,
                     model: Option<String>|
              -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<Option<String>>> + Send>> {
            Box::pin(async move {
                let mut session_metadata = backend
                    .get_session(scroll_id)
                    .await?
                    .ok_or(ArchivistError::SessionUnknown(scroll_id))?;

                if let Some(new_title) = title {
                    session_metadata.title = Some(new_title);
                }
                if let Some(new_model) = model {
                    if let Some(obj) = session_metadata.metadata.as_object_mut() {
                        obj.insert("model".to_string(), serde_json::Value::String(new_model));
                    }
                }

                session_metadata.updated_at = Utc::now();
                let title_dbg = session_metadata.title.clone();
                backend.put_session(session_metadata).await?;
                Ok(title_dbg)
            })
        };

        let title_dbg = match apply(primary.backend.clone(), title.clone(), model.clone()).await {
            Ok(t) => {
                self.record_write_success(&primary).await;
                t
            }
            Err(e) => {
                self.record_write_failure(&primary, &format!("{e}")).await;
                return Err(e);
            }
        };

        let session_metadata_for_filter = self
            .load_metadata_for_filter(scroll_id, &regs, &primary.name)
            .await;

        for reg in regs.iter() {
            if reg.name == primary.name {
                continue;
            }
            if !reg.enabled || !reg.write_active {
                continue;
            }
            if !Self::filter_allows(reg, session_metadata_for_filter.as_ref()) {
                tracing::debug!(
                    archive = %reg.name,
                    scroll_id = %scroll_id,
                    op = "update_session_metadata",
                    "filter_skip"
                );
                continue;
            }
            if let Err(e) = apply(reg.backend.clone(), title.clone(), model.clone()).await {
                self.record_write_failure(reg, &format!("{e}")).await;
                if reg.failure_mode == crate::registry::FailureMode::Required {
                    return Err(e);
                }
            } else {
                self.record_write_success(reg).await;
            }
        }

        tracing::info!(
            scroll_id = %scroll_id,
            title = ?title_dbg,
            "Updated session metadata"
        );

        Ok(())
    }

    /// Update ACP-specific metadata (`models`, `modes`, `config_options`).
    ///
    /// NOTE: read-mutate-write; falls through to inline under
    /// `WritePolicy::Queued` (no `WriteOp` variant).
    pub async fn update_session_acp_metadata(
        &self,
        scroll_id: Uuid,
        models: Option<serde_json::Value>,
        modes: Option<serde_json::Value>,
        config_options: Option<serde_json::Value>,
        archive: Option<String>,
    ) -> Result<()> {
        let primary = self.resolve_primary(archive.clone()).await?;
        let regs: Vec<std::sync::Arc<crate::registry::ArchiveRegistration>> =
            self.registrations.read().await.clone();

        let apply = |backend: std::sync::Arc<dyn crate::backend::ArchiveBackend>,
                     models: Option<serde_json::Value>,
                     modes: Option<serde_json::Value>,
                     config_options: Option<serde_json::Value>|
              -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<(bool, bool, bool)>> + Send>> {
            Box::pin(async move {
                let mut session_metadata = backend
                    .get_session(scroll_id)
                    .await?
                    .ok_or(ArchivistError::SessionUnknown(scroll_id))?;

                if let Some(new_models) = models {
                    session_metadata.models = Some(new_models);
                }
                if let Some(new_modes) = modes {
                    session_metadata.modes = Some(new_modes);
                }
                if let Some(new_config_options) = config_options {
                    session_metadata.config_options = Some(new_config_options);
                }

                session_metadata.updated_at = Utc::now();

                let has_models = session_metadata.models.is_some();
                let has_modes = session_metadata.modes.is_some();
                let has_config_options = session_metadata.config_options.is_some();

                backend.put_session(session_metadata).await?;
                Ok((has_models, has_modes, has_config_options))
            })
        };

        let (has_models, has_modes, has_config_options) = match apply(
            primary.backend.clone(),
            models.clone(),
            modes.clone(),
            config_options.clone(),
        )
        .await
        {
            Ok(flags) => {
                self.record_write_success(&primary).await;
                flags
            }
            Err(e) => {
                self.record_write_failure(&primary, &format!("{e}")).await;
                return Err(e);
            }
        };

        let session_metadata_for_filter = self
            .load_metadata_for_filter(scroll_id, &regs, &primary.name)
            .await;

        for reg in regs.iter() {
            if reg.name == primary.name {
                continue;
            }
            if !reg.enabled || !reg.write_active {
                continue;
            }
            if !Self::filter_allows(reg, session_metadata_for_filter.as_ref()) {
                tracing::debug!(
                    archive = %reg.name,
                    scroll_id = %scroll_id,
                    op = "update_session_acp_metadata",
                    "filter_skip"
                );
                continue;
            }
            if let Err(e) = apply(
                reg.backend.clone(),
                models.clone(),
                modes.clone(),
                config_options.clone(),
            )
            .await
            {
                self.record_write_failure(reg, &format!("{e}")).await;
                if reg.failure_mode == crate::registry::FailureMode::Required {
                    return Err(e);
                }
            } else {
                self.record_write_success(reg).await;
            }
        }

        tracing::info!(
            scroll_id = %scroll_id,
            has_models = has_models,
            has_modes = has_modes,
            has_config_options = has_config_options,
            "Updated session agent metadata"
        );

        Ok(())
    }

    /// Update the Matrix sharing flags on a session. `matrix_shared_at` is
    /// stamped on the first transition to `sharing_active=true`.
    ///
    /// NOTE: read-mutate-write; falls through to inline under
    /// `WritePolicy::Queued` (no `WriteOp` variant).
    pub async fn update_session_sharing(
        &self,
        scroll_id: Uuid,
        matrix_room_id: Option<String>,
        sharing_active: bool,
        archive: Option<String>,
    ) -> Result<()> {
        let primary = self.resolve_primary(archive.clone()).await?;
        let regs: Vec<std::sync::Arc<crate::registry::ArchiveRegistration>> =
            self.registrations.read().await.clone();

        let apply = |backend: std::sync::Arc<dyn crate::backend::ArchiveBackend>,
                     matrix_room_id: Option<String>,
                     sharing_active: bool|
              -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send>> {
            Box::pin(async move {
                let mut session_metadata = backend
                    .get_session(scroll_id)
                    .await?
                    .ok_or(ArchivistError::SessionUnknown(scroll_id))?;

                session_metadata.matrix_room_id = matrix_room_id;
                session_metadata.matrix_sharing_active = sharing_active;
                if sharing_active && session_metadata.matrix_shared_at.is_none() {
                    session_metadata.matrix_shared_at = Some(Utc::now());
                }
                session_metadata.updated_at = Utc::now();

                backend.put_session(session_metadata).await
            })
        };

        if let Err(e) = apply(primary.backend.clone(), matrix_room_id.clone(), sharing_active).await
        {
            self.record_write_failure(&primary, &format!("{e}")).await;
            return Err(e);
        }
        self.record_write_success(&primary).await;

        let session_metadata_for_filter = self
            .load_metadata_for_filter(scroll_id, &regs, &primary.name)
            .await;

        for reg in regs.iter() {
            if reg.name == primary.name {
                continue;
            }
            if !reg.enabled || !reg.write_active {
                continue;
            }
            if !Self::filter_allows(reg, session_metadata_for_filter.as_ref()) {
                tracing::debug!(
                    archive = %reg.name,
                    scroll_id = %scroll_id,
                    op = "update_session_sharing",
                    "filter_skip"
                );
                continue;
            }
            if let Err(e) =
                apply(reg.backend.clone(), matrix_room_id.clone(), sharing_active).await
            {
                self.record_write_failure(reg, &format!("{e}")).await;
                if reg.failure_mode == crate::registry::FailureMode::Required {
                    return Err(e);
                }
            } else {
                self.record_write_success(reg).await;
            }
        }

        tracing::debug!(
            scroll_id = %scroll_id,
            sharing_active = sharing_active,
            "Updated session sharing metadata"
        );

        Ok(())
    }

    /// Set the `completeness` level on a session.
    ///
    /// NOTE: read-mutate-write; falls through to inline under
    /// `WritePolicy::Queued` (no `WriteOp` variant).
    pub async fn update_session_completeness(
        &self,
        scroll_id: Uuid,
        completeness: SessionCompleteness,
        archive: Option<String>,
    ) -> Result<()> {
        let primary = self.resolve_primary(archive.clone()).await?;
        let regs: Vec<std::sync::Arc<crate::registry::ArchiveRegistration>> =
            self.registrations.read().await.clone();

        let apply = |backend: std::sync::Arc<dyn crate::backend::ArchiveBackend>,
                     completeness: SessionCompleteness|
              -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send>> {
            Box::pin(async move {
                let mut session_metadata = backend
                    .get_session(scroll_id)
                    .await?
                    .ok_or(ArchivistError::SessionUnknown(scroll_id))?;

                session_metadata.completeness = completeness;
                session_metadata.updated_at = Utc::now();

                backend.put_session(session_metadata).await
            })
        };

        if let Err(e) = apply(primary.backend.clone(), completeness).await {
            self.record_write_failure(&primary, &format!("{e}")).await;
            return Err(e);
        }
        self.record_write_success(&primary).await;

        let session_metadata_for_filter = self
            .load_metadata_for_filter(scroll_id, &regs, &primary.name)
            .await;

        for reg in regs.iter() {
            if reg.name == primary.name {
                continue;
            }
            if !reg.enabled || !reg.write_active {
                continue;
            }
            if !Self::filter_allows(reg, session_metadata_for_filter.as_ref()) {
                tracing::debug!(
                    archive = %reg.name,
                    scroll_id = %scroll_id,
                    op = "update_session_completeness",
                    "filter_skip"
                );
                continue;
            }
            if let Err(e) = apply(reg.backend.clone(), completeness).await {
                self.record_write_failure(reg, &format!("{e}")).await;
                if reg.failure_mode == crate::registry::FailureMode::Required {
                    return Err(e);
                }
            } else {
                self.record_write_success(reg).await;
            }
        }

        tracing::debug!(
            scroll_id = %scroll_id,
            completeness = ?completeness,
            "Updated session completeness"
        );

        Ok(())
    }

    // ------------------------------------------------------------------
    // Listing helpers
    // ------------------------------------------------------------------

    /// Return every session in the selected archive whose
    /// `matrix_sharing_active` flag is set.
    ///
    /// `SessionListQuery` does not yet expose a `matrix_sharing_active`
    /// filter, so we page through every session (including hidden ones) and
    /// filter client-side. Included-hidden matches the original
    /// `FileBasedArchivist::list_sessions_with_active_sharing`, which scanned
    /// `.contexts/` without respecting `no_update` / `is_subagent`.
    pub async fn list_sessions_with_active_sharing(
        &self,
        archive: Option<String>,
    ) -> Result<Vec<SessionMetadata>> {
        let backend = self.resolve_backend(archive).await?;

        let mut results = Vec::new();
        let mut q = SessionListQuery {
            include_hidden: true,
            limit: MAX_PAGE_LIMIT,
            ..SessionListQuery::default()
        };

        loop {
            let page = backend.list_sessions_paged(q.clone()).await?;
            for session in page.items {
                if session.matrix_sharing_active {
                    results.push(session);
                }
            }
            match page.next_cursor {
                Some(c) => q.cursor = Some(c),
                None => break,
            }
        }

        Ok(results)
    }

    // ------------------------------------------------------------------
    // Move & copy
    // ------------------------------------------------------------------

    /// Move a session to a different connector.
    ///
    /// Ported from `FileBasedArchivist::move_session`: updates the session's
    /// `connector_uid`, removes the mapping from the source connector's
    /// table (via `rewrite_connector_mappings`), and appends a new mapping
    /// to the target.
    pub async fn move_session_to_connector(
        &self,
        scroll_id: Uuid,
        target_connector_uid: Uuid,
        archive: Option<String>,
    ) -> Result<()> {
        let backend = self.resolve_backend(archive).await?;
        let mapping = backend.as_session_mapping().ok_or_else(|| {
            ArchivistError::CapabilityNotSupported {
                capability: ArchiveCapability::SessionMapping,
                backend: "selected".into(),
            }
        })?;
        let registry = backend.as_connector_registry().ok_or_else(|| {
            ArchivistError::CapabilityNotSupported {
                capability: ArchiveCapability::ConnectorRegistry,
                backend: "selected".into(),
            }
        })?;

        // 1. Read session metadata.
        let mut metadata = backend
            .get_session(scroll_id)
            .await?
            .ok_or(ArchivistError::SessionUnknown(scroll_id))?;
        let old_connector_uid = metadata.connector_uid;
        if old_connector_uid == target_connector_uid {
            return Ok(());
        }

        // 2. Verify target connector exists.
        if registry.get_connector(target_connector_uid).await?.is_none() {
            return Err(ArchivistError::ConnectorUnknown(target_connector_uid));
        }

        // 3. Update session metadata with new connector_uid.
        metadata.connector_uid = target_connector_uid;
        let native_session_id = metadata
            .native_session_id
            .clone()
            .unwrap_or_else(|| scroll_id.to_string());
        backend.put_session(metadata).await?;

        // 4. Remove mapping from old connector by rewriting its sessions
        //    table without the moved row. The trait's
        //    `rewrite_connector_mappings` also handles cache invalidation.
        let old_mappings = mapping.list_mappings_for_connector(old_connector_uid).await?;
        let filtered: Vec<_> = old_mappings
            .into_iter()
            .filter(|m| m.scroll_id != scroll_id)
            .collect();
        mapping
            .rewrite_connector_mappings(old_connector_uid, filtered)
            .await?;

        // 5. Append mapping to target connector.
        mapping
            .put_mapping(target_connector_uid, &native_session_id, scroll_id)
            .await?;

        tracing::info!(
            scroll_id = %scroll_id,
            from = %old_connector_uid,
            to = %target_connector_uid,
            "Moved session to new connector"
        );

        Ok(())
    }

    /// Bulk-move many sessions. Collects per-session errors into a
    /// `MoveReport` without aborting the whole batch.
    pub async fn move_sessions_to_connector(
        &self,
        scroll_ids: Vec<Uuid>,
        target_connector_uid: Uuid,
        archive: Option<String>,
    ) -> Result<MoveReport> {
        let mut report = MoveReport::default();

        for scroll_id in scroll_ids {
            match self
                .move_session_to_connector(scroll_id, target_connector_uid, archive.clone())
                .await
            {
                Ok(()) => {
                    report.moved += 1;
                }
                Err(e) => {
                    report.failed += 1;
                    report
                        .errors
                        .push(format!("Failed to move session {}: {}", scroll_id, e));
                    tracing::warn!(
                        scroll_id = %scroll_id,
                        error = %e,
                        "Failed to move session during bulk move"
                    );
                }
            }
        }

        tracing::info!(
            moved = report.moved,
            failed = report.failed,
            target = %target_connector_uid,
            "Completed bulk session move"
        );

        Ok(report)
    }

    /// Copy a session to a new connector under a fresh scroll_id.
    ///
    /// Copies the session metadata (with updated `scroll_id`, `connector_uid`,
    /// `created_at`, `updated_at`) and all messages, page by page. The target
    /// connector gets a new mapping entry.
    pub async fn copy_session_to_connector(
        &self,
        scroll_id: Uuid,
        target_connector_uid: Uuid,
        archive: Option<String>,
    ) -> Result<Uuid> {
        let backend = self.resolve_backend(archive).await?;
        let mapping = backend.as_session_mapping().ok_or_else(|| {
            ArchivistError::CapabilityNotSupported {
                capability: ArchiveCapability::SessionMapping,
                backend: "selected".into(),
            }
        })?;
        let registry = backend.as_connector_registry().ok_or_else(|| {
            ArchivistError::CapabilityNotSupported {
                capability: ArchiveCapability::ConnectorRegistry,
                backend: "selected".into(),
            }
        })?;

        // 1. Read source session metadata.
        let source_metadata = backend
            .get_session(scroll_id)
            .await?
            .ok_or(ArchivistError::SessionUnknown(scroll_id))?;

        // 2. Verify target connector exists.
        if registry.get_connector(target_connector_uid).await?.is_none() {
            return Err(ArchivistError::ConnectorUnknown(target_connector_uid));
        }

        // 3. Create new scroll_id + timestamp.
        let new_scroll_id = Uuid::now_v7();
        let now = Utc::now();

        // 4. Persist new session metadata with updated identity.
        let new_metadata = SessionMetadata {
            scroll_id: new_scroll_id,
            connector_uid: target_connector_uid,
            created_at: now,
            updated_at: now,
            ..source_metadata
        };
        let native_session_id = new_metadata
            .native_session_id
            .clone()
            .unwrap_or_else(|| new_scroll_id.to_string());
        backend.put_session(new_metadata).await?;

        // 5. Copy messages via the paged primitive. Uses MAX_PAGE_LIMIT
        //    per batch to stream large sessions without loading everything.
        let mut cursor: Option<MessageCursor> = None;
        loop {
            let page = backend
                .get_messages_paged(scroll_id, cursor.clone(), MAX_PAGE_LIMIT)
                .await?;
            if page.items.is_empty() {
                break;
            }

            // Rewrite each message's session reference to the new scroll_id
            // so the copied session owns its own records.
            let rewritten: Vec<MessageRecord> = page
                .items
                .into_iter()
                .map(|mut m| {
                    m.session = new_scroll_id;
                    m
                })
                .collect();

            backend.append_messages(new_scroll_id, rewritten).await?;

            match page.next_cursor {
                Some(c) => cursor = Some(c),
                None => break,
            }
        }

        // 6. Register mapping for the new (connector, native_id) → new_scroll_id.
        mapping
            .put_mapping(target_connector_uid, &native_session_id, new_scroll_id)
            .await?;

        tracing::info!(
            source_scroll_id = %scroll_id,
            new_scroll_id = %new_scroll_id,
            target_connector = %target_connector_uid,
            "Copied session to new connector"
        );

        Ok(new_scroll_id)
    }

    // ------------------------------------------------------------------
    // Cross-archive move/copy (Phase 3)
    // ------------------------------------------------------------------

    /// Copy a session from one archive to another, preserving `scroll_id`.
    ///
    /// Copies session metadata, message pages, DAG edges, and meta events
    /// where both sides have the required capability. Leaves the source
    /// intact.
    pub async fn copy_session(
        &self,
        scroll_id: Uuid,
        from: &str,
        to: &str,
    ) -> Result<()> {
        let from_reg = self
            .find_registration(from)
            .await
            .ok_or_else(|| ArchivistError::ArchiveNameUnknown(from.into()))?;
        let to_reg = self
            .find_registration(to)
            .await
            .ok_or_else(|| ArchivistError::ArchiveNameUnknown(to.into()))?;

        if !to_reg.enabled || !to_reg.write_active {
            return Err(ArchivistError::PrimaryUnavailable {
                name: to_reg.name.clone(),
                reason: "target backend is disabled or not write-active".into(),
            });
        }

        // 1. Session metadata
        let meta = from_reg
            .backend
            .get_session(scroll_id)
            .await?
            .ok_or(ArchivistError::SessionUnknown(scroll_id))?;
        to_reg.backend.put_session(meta).await?;

        // 2. Messages — page through.
        let mut cursor: Option<MessageCursor> = None;
        loop {
            let page = from_reg
                .backend
                .get_messages_paged(scroll_id, cursor.clone(), MAX_PAGE_LIMIT)
                .await?;
            if page.items.is_empty() {
                break;
            }
            to_reg.backend.append_messages(scroll_id, page.items).await?;
            match page.next_cursor {
                Some(c) => cursor = Some(c),
                None => break,
            }
        }

        // 3. DAG edges (both sides must support Dag).
        if let (Some(src_dag), Some(dst_dag)) =
            (from_reg.backend.as_dag(), to_reg.backend.as_dag())
        {
            for edge in src_dag.get_dag_edges(scroll_id).await? {
                dst_dag.append_dag_edge(edge).await?;
            }
        }

        // 4. Meta events (both sides must support MetaEvents).
        if let (Some(src_me), Some(dst_me)) =
            (from_reg.backend.as_meta_events(), to_reg.backend.as_meta_events())
        {
            let events = src_me.get_meta_events(scroll_id).await?;
            if !events.is_empty() {
                dst_me.append_meta_events(scroll_id, events).await?;
            }
        }

        // Cache: leave pointing at `from` (source remains canonical).
        Ok(())
    }

    /// Move a session from one archive to another: `copy_session` followed by
    /// source-side delete. If the copy fails, the source is intact. If the
    /// source-side delete fails AFTER a successful copy, returns
    /// `ArchivistError::PartialMove`.
    pub async fn move_session(
        &self,
        scroll_id: Uuid,
        from: &str,
        to: &str,
    ) -> Result<()> {
        // 1. Copy.
        self.copy_session(scroll_id, from, to).await?;

        // 2. Delete from source only.
        let from_reg = self
            .find_registration(from)
            .await
            .ok_or_else(|| ArchivistError::ArchiveNameUnknown(from.into()))?;

        if let Err(e) = from_reg.backend.delete_session(scroll_id).await {
            self.record_write_failure(&from_reg, &format!("{e}")).await;
            return Err(ArchivistError::PartialMove {
                copied_to: to.into(),
                delete_error: Box::new(e),
            });
        }

        // 3. Cache: rewrite to `to`.
        self.read_cache.rewrite(scroll_id, to.into()).await;

        Ok(())
    }
}
