fix(sessions): scope every session-store refusal to sessions

crates/tracedecay-global-db/src/observation/schema.rs

@@ -18,7 +18,7 @@ const OBSERVATION_UNIFIED_IDENTITY_MIGRATION: &str = "observations-unified-ident
 /// observation identity. The store's other authorities stay admissible; its
 /// session features are refused until the store is reset.
 pub(crate) const OBSERVATIONS_PREDATE_UNIFIED_IDENTITY: crate::registered::RefusedAuthorityV1 =
-    crate::registered::RefusedAuthorityV1 {
+    crate::registered::RefusedAuthorityV1::Shape {
         authority: "observations",
         reason: "observation rows predate the unified observation identity and cannot be read; reset the profile so ingestion can rebuild them from host transcripts",
     };

crates/tracedecay-global-db/src/registered.rs

@@ -2,6 +2,7 @@ use std::future::Future;
 use std::path::Path;
 use std::sync::{Arc, OnceLock, RwLock, Weak};
 
+use crate::schema_stages::RegisteredSchemaAttachmentV1;
 use tracedecay_domain::errors::TraceDecayError;
 use tracedecay_runtime_core::{
     db::{
@@ -32,18 +33,38 @@ type SessionRelationGraphStateV1 = RwLock<
     )>,
 >;
 
-/// An authority inside an admitted store whose persisted rows this binary
-/// refuses to read. The store serves its other authorities; every feature
-/// that reads the refused one gets [`Self::error`] until the store is reset.
+/// A session authority inside an admitted store whose persisted shape this
+/// binary refuses to read. The store serves its other authorities; every
+/// session feature gets [`Self::error`] until the store is reset.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub(crate) struct RefusedAuthorityV1 {
-    pub(crate) authority: &'static str,
-    pub(crate) reason: &'static str,
+pub(crate) enum RefusedAuthorityV1 {
+    /// Rows or tables whose shape nothing converts.
+    Shape {
+        authority: &'static str,
+        reason: &'static str,
+    },
+    /// A recorded schema version other than the one this binary writes.
+    Version {
+        component: &'static str,
+        found_version: Option<i64>,
+        required_version: i64,
+    },
 }
 
 impl RefusedAuthorityV1 {
     pub(crate) fn error(self) -> TraceDecayError {
-        TraceDecayError::reset_required(self.authority, self.reason)
+        match self {
+            Self::Shape { authority, reason } => TraceDecayError::reset_required(authority, reason),
+            Self::Version {
+                component,
+                found_version,
+                required_version,
+            } => TraceDecayError::ProfileResetRequired {
+                component,
+                found_version,
+                required_version,
+            },
+        }
     }
 }
 
@@ -98,12 +119,17 @@ impl RegisteredGlobalDbOwnerV1 {
     ) -> tracedecay_domain::errors::Result<Self> {
         let temporary = database.issue_lease().map_err(registered_owner_error)?;
         let registered = RegisteredGlobalDb::from_owned_database(temporary);
-        let (_, refused_authority) =
-            super::schema_stages::ensure_attached_registered_schema(&registered.database).await?;
-        // A store refused for reset is never converged: its reset deletes it.
-        if refused_authority.is_none() {
-            super::schema_stages::converge_attached_registered_schema(&registered.database).await?;
-        }
+        let refused_authority =
+            match super::schema_stages::ensure_attached_registered_schema(&registered.database)
+                .await?
+            {
+                RegisteredSchemaAttachmentV1::Admitted(_) => {
+                    super::schema_stages::converge_attached_registered_schema(&registered.database)
+                        .await?;
+                    None
+                }
+                RegisteredSchemaAttachmentV1::SessionsRefused(refused) => Some(refused),
+            };
         drop(registered);
         Ok(Self {
             database,
@@ -114,16 +140,24 @@ impl RegisteredGlobalDbOwnerV1 {
     }
 
     /// Returns the resumable convergence plan for an already admitted schema
-    /// without retaining an unowned client lease.
+    /// without retaining an unowned client lease. A store admitted in its
+    /// typed reset-required state has no plan: its reset deletes it.
     #[hotpath::measure(future = true, label = "global_db.registered.admit_daemon")]
     pub async fn admit_and_attach_for_daemon(
         database: DatabaseOwnerV1,
-    ) -> tracedecay_domain::errors::Result<(Self, super::schema_stages::RegisteredSchemaConvergence)>
-    {
+    ) -> tracedecay_domain::errors::Result<(
+        Self,
+        Option<super::schema_stages::RegisteredSchemaConvergence>,
+    )> {
         let temporary = database.issue_lease().map_err(registered_owner_error)?;
         let registered = RegisteredGlobalDb::from_owned_database(temporary);
         let (convergence, refused_authority) =
-            super::schema_stages::ensure_attached_registered_schema(&registered.database).await?;
+            match super::schema_stages::ensure_attached_registered_schema(&registered.database)
+                .await?
+            {
+                RegisteredSchemaAttachmentV1::Admitted(convergence) => (Some(convergence), None),
+                RegisteredSchemaAttachmentV1::SessionsRefused(refused) => (None, Some(refused)),
+            };
         drop(registered);
         Ok((
             Self {

crates/tracedecay-global-db/src/registered/git_correlation_schema_tests.rs

@@ -8,8 +8,8 @@ use tracedecay_runtime_core::db::TestDatabaseRuntimeScope;
 use crate::tests::harness::open_registered_test_database_fixture;
 
 /// A store recorded at the whole-projection Git evidence schema (version 5)
-/// is refused with the typed reset and left byte-for-byte untouched: nothing
-/// converts or copies the older shape.
+/// is admitted in its typed reset-required state and left byte-for-byte
+/// untouched: nothing converts or copies the older shape.
 #[tokio::test]
 async fn an_older_git_correlation_schema_is_refused_without_mutation() {
     crate::register_registered_schema_installer();
@@ -33,19 +33,23 @@ async fn an_older_git_correlation_schema_is_refused_without_mutation() {
         .unwrap();
     let before = fs::read(&database_path).unwrap();
 
-    let error = match open_registered_test_database_fixture(&database_path, scope()).await {
-        Ok(_) => panic!("an older Git correlation schema must not be admitted"),
-        Err(error) => error,
-    };
-
-    assert!(matches!(
-        error,
-        TraceDecayError::ProfileResetRequired {
-            component: "git correlation",
-            found_version: Some(5),
-            required_version: 6,
-        }
-    ));
+    let (lease, owner) = open_registered_test_database_fixture(&database_path, scope())
+        .await
+        .expect("the store's other authorities stay admissible");
+    for refusal in [lease.reset_required(), owner.reset_required()] {
+        assert!(
+            matches!(
+                refusal,
+                Some(TraceDecayError::ProfileResetRequired {
+                    component: "git correlation",
+                    found_version: Some(5),
+                    required_version: 6,
+                })
+            ),
+            "an older Git correlation schema refuses session features: {refusal:?}"
+        );
+    }
+    drop((lease, owner));
     assert_eq!(
         fs::read(&database_path).unwrap(),
         before,

crates/tracedecay-global-db/src/registered/workflow_schema_tests.rs

@@ -56,25 +56,27 @@ async fn assert_workflow_schema_reset_without_mutation(malformed_schema: String)
     drop(connection);
     drop(database);
 
-    let error = match open_registered_test_database_fixture(
+    let (lease, owner) = open_registered_test_database_fixture(
         &database_path,
         TestDatabaseRuntimeScope::ProjectSessions {
             project_id: ProjectId::new("project.workflow-schema").unwrap(),
         },
     )
     .await
-    {
-        Ok(_) => panic!("malformed workflow schema must not be completed"),
-        Err(error) => error,
-    };
-
-    assert!(matches!(
-        error,
-        TraceDecayError::ResetRequired {
-            ref authority,
-            ..
-        } if authority == "workflow"
-    ));
+    .expect("the store's other authorities stay admissible");
+    for refusal in [lease.reset_required(), owner.reset_required()] {
+        assert!(
+            matches!(
+                refusal,
+                Some(TraceDecayError::ResetRequired {
+                    ref authority,
+                    ..
+                }) if authority == "workflow"
+            ),
+            "a malformed workflow schema refuses session features: {refusal:?}"
+        );
+    }
+    drop((lease, owner));
     assert_eq!(
         fs::read(&database_path).unwrap(),
         before_bytes,