meerkat-mob 0.8.10

//! MobStorage bundle.
//!
//! Groups the event store with a session store for a mob's isolated storage.

#[cfg(not(target_arch = "wasm32"))]
use crate::store::SqliteMobStores;
use crate::store::{
    InMemoryMobEventStore, InMemoryMobIdentityStatusStore, InMemoryMobIdentityStore,
    InMemoryMobRunStore, InMemoryMobRuntimeMetadataStore, InMemoryMobSpecStore,
    InMemoryRealmProfileStore, MobEventStore, MobIdentityMemberStore, MobIdentityStatusStore,
    MobIdentityStore, MobRunStore, MobRuntimeMetadataStore, MobSpecStore, RealmProfileStore,
    authority_validating_mob_run_store,
};
#[cfg(not(target_arch = "wasm32"))]
use std::path::Path;
use std::sync::Arc;

/// Storage bundle for a mob.
///
/// Contains both the mob event store (structural state) and session
/// store (for meerkat sessions). Each mob has its own isolated storage.
#[derive(Clone)]
pub struct MobStorage {
    /// Event store for mob structural events.
    pub(crate) events: Arc<dyn MobEventStore>,
    /// Flow run persistence store.
    pub(crate) runs: Arc<dyn MobRunStore>,
    /// Flow spec persistence store.
    pub(crate) specs: Arc<dyn MobSpecStore>,
    /// Runtime metadata store for supervisor authority and compatibility projections.
    pub(crate) runtime_metadata: Arc<dyn MobRuntimeMetadataStore>,
    /// Sole desired-state authority for identity intent, leases, and immutable custody.
    pub(crate) identity: Arc<dyn MobIdentityStore>,
    /// Optional built-in-only transaction joining identity authority to the
    /// structural member and identity-local wiring event targets. Custom store
    /// compositions cannot provide this implicitly.
    pub(crate) identity_member: Option<Arc<dyn MobIdentityMemberStore>>,
    /// Replaceable output-only identity convergence diagnostics.
    pub(crate) identity_status: Arc<dyn MobIdentityStatusStore>,
    /// Realm-scoped reusable profile store.
    pub(crate) realm_profiles: Option<Arc<dyn RealmProfileStore>>,
}

impl MobStorage {
    /// Create a storage bundle with in-memory stores (for tests and ephemeral mobs).
    pub fn in_memory() -> Self {
        let (runs, specs) = Self::in_memory_flow_stores();
        let events = InMemoryMobEventStore::new();
        let identity = InMemoryMobIdentityStore::paired_with_event_store(
            &events,
            Arc::new(crate::store::SystemMobIdentityStoreClock),
        );
        let identity_member: Arc<dyn MobIdentityMemberStore> = Arc::new(identity.clone());
        Self {
            events: Arc::new(events),
            runs,
            specs,
            runtime_metadata: Arc::new(InMemoryMobRuntimeMetadataStore::new()),
            identity: Arc::new(identity),
            identity_member: Some(identity_member),
            identity_status: Arc::new(InMemoryMobIdentityStatusStore::new()),
            realm_profiles: Some(Arc::new(InMemoryRealmProfileStore::new())),
        }
    }

    /// Create in-memory run/spec stores for flow persistence.
    pub fn in_memory_flow_stores() -> (Arc<dyn MobRunStore>, Arc<dyn MobSpecStore>) {
        (
            authority_validating_mob_run_store(Arc::new(InMemoryMobRunStore::new())),
            Arc::new(InMemoryMobSpecStore::new()),
        )
    }

    /// Build a full storage bundle from a custom event store and in-memory flow stores.
    pub fn with_events(events: Arc<dyn MobEventStore>) -> Self {
        let (runs, specs) = Self::in_memory_flow_stores();
        Self {
            events,
            runs,
            specs,
            runtime_metadata: Arc::new(InMemoryMobRuntimeMetadataStore::new()),
            identity: Arc::new(InMemoryMobIdentityStore::new()),
            identity_member: None,
            identity_status: Arc::new(InMemoryMobIdentityStatusStore::new()),
            realm_profiles: Some(Arc::new(InMemoryRealmProfileStore::new())),
        }
    }

    /// Test-only convenience with explicitly ephemeral identity authority.
    #[cfg(test)]
    pub(crate) fn with_events_and_runtime_metadata(
        events: Arc<dyn MobEventStore>,
        runtime_metadata: Arc<dyn MobRuntimeMetadataStore>,
    ) -> Self {
        let (runs, specs) = Self::in_memory_flow_stores();
        Self {
            events,
            runs,
            specs,
            runtime_metadata,
            identity: Arc::new(InMemoryMobIdentityStore::new()),
            identity_member: None,
            identity_status: Arc::new(InMemoryMobIdentityStatusStore::new()),
            realm_profiles: Some(Arc::new(InMemoryRealmProfileStore::new())),
        }
    }

    /// Build a storage bundle from custom store implementations.
    pub fn custom(
        events: Arc<dyn MobEventStore>,
        runs: Arc<dyn MobRunStore>,
        specs: Arc<dyn MobSpecStore>,
        identity: Arc<dyn MobIdentityStore>,
        identity_status: Arc<dyn MobIdentityStatusStore>,
    ) -> Self {
        Self::custom_with_runtime_metadata(
            events,
            runs,
            specs,
            Arc::new(InMemoryMobRuntimeMetadataStore::new()),
            identity,
            identity_status,
        )
    }

    /// Build a storage bundle from custom store implementations, including runtime metadata.
    pub fn custom_with_runtime_metadata(
        events: Arc<dyn MobEventStore>,
        runs: Arc<dyn MobRunStore>,
        specs: Arc<dyn MobSpecStore>,
        runtime_metadata: Arc<dyn MobRuntimeMetadataStore>,
        identity: Arc<dyn MobIdentityStore>,
        identity_status: Arc<dyn MobIdentityStatusStore>,
    ) -> Self {
        Self {
            events,
            runs: authority_validating_mob_run_store(runs),
            specs,
            runtime_metadata,
            identity,
            identity_member: None,
            identity_status,
            realm_profiles: None,
        }
    }

    /// Attach the realm-scoped reusable profile store used by mob runtimes.
    pub fn with_realm_profile_store(
        mut self,
        realm_profiles: Option<Arc<dyn RealmProfileStore>>,
    ) -> Self {
        self.realm_profiles = realm_profiles;
        self
    }

    /// Return whether the structural event log is empty.
    pub async fn is_event_log_empty(&self) -> Result<bool, crate::store::MobStoreError> {
        Ok(self.events.latest_cursor().await? == 0)
    }

    /// Create a storage bundle backed by a single SQLite database file.
    ///
    /// Uses WAL mode — no exclusive file lock is held, so the same path
    /// can be reopened after drop within the same process.
    #[cfg(not(target_arch = "wasm32"))]
    pub fn persistent(path: impl AsRef<Path>) -> Result<Self, crate::MobError> {
        let stores = SqliteMobStores::open(path)?;
        let identity = stores.identity_store();
        let identity_member: Arc<dyn MobIdentityMemberStore> = Arc::new(identity.clone());
        Ok(Self {
            events: Arc::new(stores.event_store()),
            runs: authority_validating_mob_run_store(Arc::new(stores.run_store())),
            specs: Arc::new(stores.spec_store()),
            runtime_metadata: Arc::new(stores.runtime_metadata_store()),
            identity: Arc::new(identity),
            identity_member: Some(identity_member),
            identity_status: Arc::new(stores.identity_status_store()),
            realm_profiles: Some(Arc::new(stores.realm_profile_store())),
        })
    }
}

impl std::fmt::Debug for MobStorage {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MobStorage")
            .field("events", &"<dyn MobEventStore>")
            .field("runs", &"<dyn MobRunStore>")
            .field("specs", &"<dyn MobSpecStore>")
            .field("runtime_metadata", &"<dyn MobRuntimeMetadataStore>")
            .field("identity", &"<dyn MobIdentityStore>")
            .field(
                "identity_member",
                &self
                    .identity_member
                    .as_ref()
                    .map(|_| "<dyn MobIdentityMemberStore>"),
            )
            .field("identity_status", &"<dyn MobIdentityStatusStore>")
            .field(
                "realm_profiles",
                &self
                    .realm_profiles
                    .as_ref()
                    .map(|_| "<dyn RealmProfileStore>"),
            )
            .finish()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::event::{MobEventKind, NewMobEvent};
    use crate::ids::{FlowId, FrameId, LoopId, LoopInstanceId, MobId, RunId, StepId};
    use crate::run::{
        FailureLedgerEntry, FrameSnapshot, LoopIterationLedgerEntry, LoopSnapshot, MobRun,
        MobRunProvenanceAuthority, MobRunStatus, StepLedgerEntry,
    };
    use crate::store::MobStoreError;
    use std::sync::Mutex;

    struct ForgedRunStore {
        run: Mutex<Option<MobRun>>,
    }

    impl ForgedRunStore {
        fn new(run: Option<MobRun>) -> Self {
            Self {
                run: Mutex::new(run),
            }
        }
    }

    #[async_trait::async_trait]
    impl MobRunStore for ForgedRunStore {
        async fn create_run(&self, run: MobRun) -> Result<(), MobStoreError> {
            *self.run.lock().expect("forged run mutex") = Some(run);
            Ok(())
        }

        async fn get_run(&self, run_id: &RunId) -> Result<Option<MobRun>, MobStoreError> {
            Ok(self
                .run
                .lock()
                .expect("forged run mutex")
                .as_ref()
                .filter(|run| &run.run_id == run_id)
                .cloned())
        }

        async fn list_runs(
            &self,
            mob_id: &MobId,
            flow_id: Option<&FlowId>,
        ) -> Result<Vec<MobRun>, MobStoreError> {
            Ok(self
                .run
                .lock()
                .expect("forged run mutex")
                .as_ref()
                .filter(|run| &run.mob_id == mob_id)
                .filter(|run| flow_id.is_none_or(|flow_id| &run.flow_id == flow_id))
                .cloned()
                .into_iter()
                .collect())
        }

        async fn cas_flow_state_with_authority(
            &self,
            _run_id: &RunId,
            _expected: &crate::run::flow_run::State,
            _next: &crate::run::flow_run::State,
            _authority_inputs: Vec<crate::machines::mob_machine::MobMachineInput>,
        ) -> Result<bool, MobStoreError> {
            Err(MobStoreError::Internal(
                "not implemented in forged store".into(),
            ))
        }

        async fn cas_run_snapshot_with_authority(
            &self,
            _run_id: &RunId,
            _expected_status: MobRunStatus,
            _expected_flow_state: &crate::run::flow_run::State,
            _next_status: MobRunStatus,
            _next_flow_state: &crate::run::flow_run::State,
            _authority_inputs: Vec<crate::machines::mob_machine::MobMachineInput>,
        ) -> Result<bool, MobStoreError> {
            Err(MobStoreError::Internal(
                "not implemented in forged store".into(),
            ))
        }

        async fn append_step_entry_with_authority(
            &self,
            _run_id: &RunId,
            _entry: StepLedgerEntry,
            _authority: MobRunProvenanceAuthority,
        ) -> Result<(), MobStoreError> {
            Err(MobStoreError::Internal(
                "not implemented in forged store".into(),
            ))
        }

        async fn append_step_entry_if_absent_with_authority(
            &self,
            _run_id: &RunId,
            _entry: StepLedgerEntry,
            _authority: MobRunProvenanceAuthority,
        ) -> Result<bool, MobStoreError> {
            Err(MobStoreError::Internal(
                "not implemented in forged store".into(),
            ))
        }

        async fn append_failure_entry_with_authority(
            &self,
            _run_id: &RunId,
            _entry: FailureLedgerEntry,
            _authority: MobRunProvenanceAuthority,
        ) -> Result<(), MobStoreError> {
            Err(MobStoreError::Internal(
                "not implemented in forged store".into(),
            ))
        }

        async fn cas_frame_state_with_authority(
            &self,
            _run_id: &RunId,
            _frame_id: &FrameId,
            _expected: Option<&FrameSnapshot>,
            _next: FrameSnapshot,
            _authority_inputs: Vec<crate::machines::mob_machine::MobMachineInput>,
        ) -> Result<bool, MobStoreError> {
            Err(MobStoreError::Internal(
                "not implemented in forged store".into(),
            ))
        }

        #[allow(clippy::too_many_arguments)]
        async fn cas_grant_node_slot_with_authority(
            &self,
            _run_id: &RunId,
            _expected_run_state: &crate::run::flow_run::State,
            _next_run_state: crate::run::flow_run::State,
            _frame_id: &FrameId,
            _expected_frame: &FrameSnapshot,
            _next_frame: FrameSnapshot,
            _authority_inputs: Vec<crate::machines::mob_machine::MobMachineInput>,
        ) -> Result<bool, MobStoreError> {
            Err(MobStoreError::Internal(
                "not implemented in forged store".into(),
            ))
        }

        #[allow(clippy::too_many_arguments)]
        async fn cas_complete_step_and_record_output_with_authority(
            &self,
            _run_id: &RunId,
            _frame_id: &FrameId,
            _expected_frame: &FrameSnapshot,
            _next_frame: FrameSnapshot,
            _step_output_key: String,
            _step_output: serde_json::Value,
            _loop_context: Option<(&LoopId, u64)>,
            _authority_inputs: Vec<crate::machines::mob_machine::MobMachineInput>,
        ) -> Result<bool, MobStoreError> {
            Err(MobStoreError::Internal(
                "not implemented in forged store".into(),
            ))
        }

        #[allow(clippy::too_many_arguments)]
        async fn cas_start_loop_with_authority(
            &self,
            _run_id: &RunId,
            _loop_instance_id: &LoopInstanceId,
            _expected_run_state: &crate::run::flow_run::State,
            _next_run_state: crate::run::flow_run::State,
            _frame_id: &FrameId,
            _expected_frame: &FrameSnapshot,
            _next_frame: FrameSnapshot,
            _initial_loop: LoopSnapshot,
            _authority_inputs: Vec<crate::machines::mob_machine::MobMachineInput>,
        ) -> Result<bool, MobStoreError> {
            Err(MobStoreError::Internal(
                "not implemented in forged store".into(),
            ))
        }

        #[allow(clippy::too_many_arguments)]
        async fn cas_loop_request_body_frame_with_authority(
            &self,
            _run_id: &RunId,
            _loop_instance_id: &LoopInstanceId,
            _expected_loop: &LoopSnapshot,
            _next_loop: LoopSnapshot,
            _expected_run_state: &crate::run::flow_run::State,
            _next_run_state: crate::run::flow_run::State,
            _authority_inputs: Vec<crate::machines::mob_machine::MobMachineInput>,
        ) -> Result<bool, MobStoreError> {
            Err(MobStoreError::Internal(
                "not implemented in forged store".into(),
            ))
        }

        #[allow(clippy::too_many_arguments)]
        async fn cas_grant_body_frame_start_with_authority(
            &self,
            _run_id: &RunId,
            _loop_instance_id: &LoopInstanceId,
            _expected_loop: &LoopSnapshot,
            _next_loop: LoopSnapshot,
            _frame_id: &FrameId,
            _initial_frame: FrameSnapshot,
            _ledger_entry: LoopIterationLedgerEntry,
            _expected_run_state: &crate::run::flow_run::State,
            _next_run_state: crate::run::flow_run::State,
            _authority_inputs: Vec<crate::machines::mob_machine::MobMachineInput>,
        ) -> Result<bool, MobStoreError> {
            Err(MobStoreError::Internal(
                "not implemented in forged store".into(),
            ))
        }

        #[allow(clippy::too_many_arguments)]
        async fn cas_complete_body_frame_with_authority(
            &self,
            _run_id: &RunId,
            _loop_instance_id: &LoopInstanceId,
            _expected_loop: &LoopSnapshot,
            _next_loop: LoopSnapshot,
            _frame_id: &FrameId,
            _expected_frame: &FrameSnapshot,
            _next_frame: FrameSnapshot,
            _expected_run_state: &crate::run::flow_run::State,
            _next_run_state: crate::run::flow_run::State,
            _authority_inputs: Vec<crate::machines::mob_machine::MobMachineInput>,
        ) -> Result<bool, MobStoreError> {
            Err(MobStoreError::Internal(
                "not implemented in forged store".into(),
            ))
        }

        #[allow(clippy::too_many_arguments)]
        async fn cas_complete_loop_with_authority(
            &self,
            _run_id: &RunId,
            _loop_instance_id: &LoopInstanceId,
            _expected_loop: &LoopSnapshot,
            _next_loop: LoopSnapshot,
            _frame_id: &FrameId,
            _expected_frame: &FrameSnapshot,
            _next_frame: FrameSnapshot,
            _expected_run_state: &crate::run::flow_run::State,
            _next_run_state: crate::run::flow_run::State,
            _authority_inputs: Vec<crate::machines::mob_machine::MobMachineInput>,
        ) -> Result<bool, MobStoreError> {
            Err(MobStoreError::Internal(
                "not implemented in forged store".into(),
            ))
        }
    }

    fn forged_status_run() -> MobRun {
        let mut run = MobRun::authority_backed_for_steps(
            RunId::new(),
            MobId::from("mob"),
            crate::FlowId::from("flow"),
            [StepId::from("step-1")],
            MobRunStatus::Pending,
            serde_json::json!({}),
        )
        .expect("authority-backed run");
        run.status = MobRunStatus::Completed;
        run
    }

    #[tokio::test]
    async fn test_in_memory_storage_creates_working_stores() {
        let storage = MobStorage::in_memory();

        // Event store works
        let event = NewMobEvent {
            mob_id: MobId::from("test"),
            timestamp: None,
            kind: MobEventKind::MobCompleted,
        };
        let stored = storage.events.append(event).await.unwrap();
        assert_eq!(stored.cursor, 1);

        let all = storage.events.replay_all().await.unwrap();
        assert_eq!(all.len(), 1);

        let mob_id = MobId::from("test");
        let identity = crate::AgentIdentity::from("member-a");
        assert!(matches!(
            storage
                .identity
                .observe_identity_intent(&mob_id, &identity)
                .await
                .unwrap(),
            crate::IdentityStoredObservation::Missing
        ));
        assert!(matches!(
            storage
                .identity
                .claim_or_renew_identity_lease(
                    &mob_id,
                    &identity,
                    "controller",
                    "incarnation-a",
                    1_000,
                )
                .await
                .unwrap(),
            crate::IdentityLeaseClaimOutcome::Acquired(_)
        ));

        let status = crate::IdentityConvergenceStatus {
            identity: identity.clone(),
            intent_revision: None,
            lease_epoch: Some(1),
            decision: Some(crate::IdentityReconcileDecision::AcquireLease),
            observed_at_ms: 1,
            detail: None,
        };
        storage
            .identity_status
            .replace_identity_convergence_status(&mob_id, &status)
            .await
            .unwrap();
        assert_eq!(
            storage
                .identity_status
                .load_identity_convergence_status(&mob_id, &identity)
                .await
                .unwrap(),
            crate::IdentityStoredObservation::Valid(status)
        );
    }

    #[tokio::test]
    async fn test_in_memory_flow_stores_create_working_run_and_spec_stores() {
        let (runs, specs) = MobStorage::in_memory_flow_stores();
        let run = MobRun::authority_backed_for_steps(
            RunId::new(),
            MobId::from("mob"),
            crate::FlowId::from("flow"),
            [StepId::from("step-1")],
            MobRunStatus::Pending,
            serde_json::json!({}),
        )
        .expect("authority-backed run");
        runs.create_run(run.clone()).await.unwrap();
        assert!(runs.get_run(&run.run_id).await.unwrap().is_some());

        let definition = crate::definition::MobDefinition::from_toml(
            r#"
[mob]
id = "mob"
[profiles.worker]
model = "test"
"#,
        )
        .unwrap();
        let revision = specs
            .put_spec(&MobId::from("mob"), &definition, None)
            .await
            .unwrap();
        assert_eq!(revision, 1);
    }

    #[tokio::test]
    async fn custom_run_store_rejects_forged_lifecycle_projection() {
        let forged = forged_status_run();
        let storage = MobStorage::custom(
            Arc::new(InMemoryMobEventStore::new()),
            Arc::new(ForgedRunStore::new(Some(forged.clone()))),
            Arc::new(InMemoryMobSpecStore::new()),
            Arc::new(InMemoryMobIdentityStore::new()),
            Arc::new(InMemoryMobIdentityStatusStore::new()),
        );

        let read_error = storage
            .runs
            .get_run(&forged.run_id)
            .await
            .expect_err("custom store read must reject forged run projection");
        assert!(
            read_error
                .to_string()
                .contains("not authorized by MobMachine"),
            "unexpected custom read error: {read_error}"
        );

        let list_error = storage
            .runs
            .list_runs(&forged.mob_id, None)
            .await
            .expect_err("custom store list must reject forged run projection");
        assert!(
            list_error
                .to_string()
                .contains("not authorized by MobMachine"),
            "unexpected custom list error: {list_error}"
        );

        let write_error = storage
            .runs
            .create_run(forged)
            .await
            .expect_err("custom store create must reject forged run projection");
        assert!(
            write_error
                .to_string()
                .contains("not authorized by MobMachine"),
            "unexpected custom create error: {write_error}"
        );
    }

    #[tokio::test]
    async fn test_persistent_storage_uses_shared_database_for_all_stores() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("mob.db");
        let storage = MobStorage::persistent(&db_path).unwrap();

        let event = NewMobEvent {
            mob_id: MobId::from("mob"),
            timestamp: None,
            kind: MobEventKind::MobCompleted,
        };
        storage.events.append(event).await.unwrap();

        let run = MobRun::authority_backed_for_steps(
            RunId::new(),
            MobId::from("mob"),
            crate::FlowId::from("flow"),
            [StepId::from("step-1")],
            MobRunStatus::Pending,
            serde_json::json!({}),
        )
        .expect("authority-backed run");
        storage.runs.create_run(run.clone()).await.unwrap();
        assert!(storage.runs.get_run(&run.run_id).await.unwrap().is_some());

        let definition = crate::definition::MobDefinition::from_toml(
            r#"
[mob]
id = "mob"
[profiles.worker]
model = "test"
"#,
        )
        .unwrap();
        let revision = storage
            .specs
            .put_spec(&MobId::from("mob"), &definition, None)
            .await
            .unwrap();
        assert_eq!(revision, 1);

        let identity = crate::AgentIdentity::from("persistent-member");
        let lease = storage
            .identity
            .claim_or_renew_identity_lease(
                &MobId::from("mob"),
                &identity,
                "controller",
                "incarnation-a",
                30_000,
            )
            .await
            .unwrap();
        let expected_epoch = match lease {
            crate::IdentityLeaseClaimOutcome::Acquired(claim) => claim.epoch,
            other => panic!("expected initial persistent lease, got {other:?}"),
        };
        drop(storage);

        let reopened = MobStorage::persistent(&db_path).unwrap();
        assert!(matches!(
            reopened
                .identity
                .observe_identity_lease(&MobId::from("mob"), &identity)
                .await
                .unwrap(),
            crate::IdentityStoredObservation::Valid(crate::IdentityLeaseRecord {
                epoch_highwater,
                ..
            }) if epoch_highwater == expected_epoch
        ));
    }
}