diff --git a/crates/tracedecay-global-db/src/observation/schema.rs b/crates/tracedecay-global-db/src/observation/schema.rs index 7323d641e0..a88a74150f 100644 --- a/crates/tracedecay-global-db/src/observation/schema.rs +++ b/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", }; diff --git a/crates/tracedecay-global-db/src/registered.rs b/crates/tracedecay-global-db/src/registered.rs index bd1d423823..292e8e032b 100644 --- a/crates/tracedecay-global-db/src/registered.rs +++ b/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, + 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 { 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(®istered.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(®istered.database).await?; - } + let refused_authority = + match super::schema_stages::ensure_attached_registered_schema(®istered.database) + .await? + { + RegisteredSchemaAttachmentV1::Admitted(_) => { + super::schema_stages::converge_attached_registered_schema(®istered.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, + )> { 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(®istered.database).await?; + match super::schema_stages::ensure_attached_registered_schema(®istered.database) + .await? + { + RegisteredSchemaAttachmentV1::Admitted(convergence) => (Some(convergence), None), + RegisteredSchemaAttachmentV1::SessionsRefused(refused) => (None, Some(refused)), + }; drop(registered); Ok(( Self { diff --git a/crates/tracedecay-global-db/src/registered/git_correlation_schema_tests.rs b/crates/tracedecay-global-db/src/registered/git_correlation_schema_tests.rs index 304aed367e..c7b0cc928c 100644 --- a/crates/tracedecay-global-db/src/registered/git_correlation_schema_tests.rs +++ b/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, diff --git a/crates/tracedecay-global-db/src/registered/workflow_schema_tests.rs b/crates/tracedecay-global-db/src/registered/workflow_schema_tests.rs index 4dce6eb8aa..2e82ec9dd9 100644 --- a/crates/tracedecay-global-db/src/registered/workflow_schema_tests.rs +++ b/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, diff --git a/crates/tracedecay-global-db/src/schema_stages.rs b/crates/tracedecay-global-db/src/schema_stages.rs index b0b25f5561..908a26b684 100644 --- a/crates/tracedecay-global-db/src/schema_stages.rs +++ b/crates/tracedecay-global-db/src/schema_stages.rs @@ -398,6 +398,15 @@ struct RegisteredSchemaAdmissionClassification { workflow_admission: WorkflowSchemaAdmission, } +/// A store's admission verdict. A store whose other authorities admit but +/// whose LCM, workflow, or git correlation shape this binary refuses is +/// admitted untouched for those other authorities: it is never installed or +/// converged, and its session features refuse until the store is reset. +enum RegisteredSchemaAdmission { + Admissible(RegisteredSchemaAdmissionClassification), + SessionAuthorityRefused(RefusedAuthorityV1), +} + /// Read-only classification of every schema authority's admission state, /// shared by initialization admission and existing-store attach. Each /// authority surfaces its own typed reset state; nothing here mutates the @@ -412,58 +421,47 @@ struct RegisteredSchemaAdmissionClassification { #[hotpath::measure(future = true, label = "global_db.schema.query.classify")] async fn classify_registered_schema_admission( connection: &impl QueryExecutor, -) -> tracedecay_domain::errors::Result { +) -> tracedecay_domain::errors::Result { Box::pin(classify_registered_schema_authorities(connection)).await } async fn classify_registered_schema_authorities( connection: &impl QueryExecutor, -) -> tracedecay_domain::errors::Result { - // The LCM authority classifies profile content first: a legacy or - // version-skewed session store must surface its own ProfileResetRequired - // state instead of being masked by the coarser workflow/configuration +) -> tracedecay_domain::errors::Result { + // The LCM authority classifies profile content first. Whenever a later + // authority also fails, the earliest session refusal is the store's hard + // verdict: a legacy or version-skewed session store surfaces its own + // reset identity instead of being masked by the coarser configuration // schema resets, which would also flag a store those features were simply // never installed in. - tracedecay_lcm::schema::require_admissible_lcm_schema(connection) - .await - .map_err(|error| match error { - tracedecay_lcm::LcmError::ProfileResetRequired { - found_version, - required_version, - } => tracedecay_domain::errors::TraceDecayError::ProfileResetRequired { - component: "LCM", - found_version, - required_version, - }, - error => global_db_operation_error("classify LCM schema admission", error), - })?; + let refused_lcm = lcm_schema_refusal(connection).await?; + let surface = |refused: Option| { + move |error| refused.map_or(error, RefusedAuthorityV1::error) + }; let configuration_fresh = configuration::fresh_configuration_store_evidence(connection) .await - .map_err(|error| match error { - configuration::ConfigurationSchemaError::ResetRequired { reason } => { - tracedecay_domain::errors::TraceDecayError::reset_required("configuration", reason) - } - configuration::ConfigurationSchemaError::Storage(error) => { - global_db_operation_error("inspect configuration schema freshness", error) - } - })?; + .map_err(configuration_schema_error( + "inspect configuration schema freshness", + )) + .map_err(surface(refused_lcm))?; let temporal_admission = session_temporal_schema::require_admissible_session_temporal_schema( connection, configuration_fresh.as_ref(), ) - .await?; - let workflow_admission = inspect_workflow_schema_for_admission(connection).await?; - require_admissible_git_correlation_schema(connection).await?; + .await + .map_err(surface(refused_lcm))?; + let workflow_admission = inspect_workflow_schema_for_admission(connection) + .await + .map_err(surface(refused_lcm))?; + let refused = refused_lcm.or(workflow_admission.err()); + let refused_git_correlation = git_correlation_schema_refusal(connection) + .await + .map_err(surface(refused))?; + let refused = refused.or(refused_git_correlation); configuration::admit_configuration_schema(connection, configuration_fresh.as_ref()) .await - .map_err(|error| match error { - configuration::ConfigurationSchemaError::ResetRequired { reason } => { - tracedecay_domain::errors::TraceDecayError::reset_required("configuration", reason) - } - configuration::ConfigurationSchemaError::Storage(error) => { - global_db_operation_error("admit configuration schema", error) - } - })?; + .map_err(configuration_schema_error("admit configuration schema")) + .map_err(surface(refused))?; // An existing catalog whose remote-deletion tombstone table drifted from the // contract cannot be trusted to gate replay or admission, so admission fails // closed with the tip's typed reset authority rather than silently @@ -471,37 +469,80 @@ async fn classify_registered_schema_authorities( if configuration_fresh.is_none() && let Err(error) = validate_remote_deletion_schema_contract(connection).await { - return Err(tracedecay_domain::errors::TraceDecayError::reset_required( - "remote deletion tombstones", - error.to_string(), + return Err(surface(refused)( + tracedecay_domain::errors::TraceDecayError::reset_required( + "remote deletion tombstones", + error.to_string(), + ), )); } - Ok(RegisteredSchemaAdmissionClassification { - configuration_fresh, - temporal_admission, - workflow_admission, + Ok(match (refused, workflow_admission) { + (None, Ok(workflow_admission)) => { + RegisteredSchemaAdmission::Admissible(RegisteredSchemaAdmissionClassification { + configuration_fresh, + temporal_admission, + workflow_admission, + }) + } + (Some(refused), _) | (None, Err(refused)) => { + RegisteredSchemaAdmission::SessionAuthorityRefused(refused) + } }) } +fn configuration_schema_error( + operation: &'static str, +) -> impl Fn(configuration::ConfigurationSchemaError) -> tracedecay_domain::errors::TraceDecayError +{ + move |error| match error { + configuration::ConfigurationSchemaError::ResetRequired { reason } => { + tracedecay_domain::errors::TraceDecayError::reset_required("configuration", reason) + } + configuration::ConfigurationSchemaError::Storage(error) => { + global_db_operation_error(operation, error) + } + } +} + +async fn lcm_schema_refusal( + connection: &impl QueryExecutor, +) -> tracedecay_domain::errors::Result> { + match tracedecay_lcm::schema::require_admissible_lcm_schema(connection).await { + Ok(_) => Ok(None), + Err(tracedecay_lcm::LcmError::ProfileResetRequired { + found_version, + required_version, + }) => Ok(Some(RefusedAuthorityV1::Version { + component: "LCM", + found_version, + required_version, + })), + Err(error) => Err(global_db_operation_error( + "classify LCM schema admission", + error, + )), + } +} + /// Git evidence is stored as per-session rows since schema version 6. A store /// recorded at any other version holds a shape nothing converts, so it keeps /// its data untouched behind the typed, versioned reset. -async fn require_admissible_git_correlation_schema( +async fn git_correlation_schema_refusal( connection: &impl QueryExecutor, -) -> tracedecay_domain::errors::Result<()> { +) -> tracedecay_domain::errors::Result> { let recorded = recorded_git_correlation_schema_version(connection) .await .map_err(|error| global_db_operation_error("inspect git correlation schema", error))?; - match recorded { - Some(found) if found != GIT_CORRELATION_SCHEMA_VERSION => Err( - tracedecay_domain::errors::TraceDecayError::ProfileResetRequired { + Ok(match recorded { + Some(found) if found != GIT_CORRELATION_SCHEMA_VERSION => { + Some(RefusedAuthorityV1::Version { component: "git correlation", found_version: Some(found), required_version: GIT_CORRELATION_SCHEMA_VERSION, - }, - ), - _ => Ok(()), - } + }) + } + _ => None, + }) } /// Authority named by the typed reset an existing store receives when the @@ -542,7 +583,12 @@ pub async fn ensure_registered_schema_for_admission( configuration_fresh, temporal_admission, workflow_admission, - } = classify_registered_schema_admission(installation).await?; + } = match classify_registered_schema_admission(installation).await? { + RegisteredSchemaAdmission::Admissible(classification) => classification, + RegisteredSchemaAdmission::SessionAuthorityRefused(refused) => { + return Err(refused.error()); + } + }; let is_fresh = configuration_fresh.is_some(); let force_exhaustive = !authority_invariant_triggers_intact(installation).await?; let transaction = installation @@ -693,14 +739,9 @@ async fn install_registered_schema_stage_sequence( let is_fresh = configuration_fresh.is_some(); configuration::ensure_configuration_schema(transaction, configuration_fresh) .await - .map_err(|error| match error { - configuration::ConfigurationSchemaError::ResetRequired { reason } => { - tracedecay_domain::errors::TraceDecayError::reset_required("configuration", reason) - } - configuration::ConfigurationSchemaError::Storage(error) => { - global_db_operation_error("initialize configuration schema", error) - } - })?; + .map_err(configuration_schema_error( + "initialize configuration schema", + ))?; ensure_authority_audit_checkpoint_schema(transaction).await?; if force_exhaustive && !is_fresh { // Persist the requirement before later schema work repairs the @@ -1002,18 +1043,25 @@ pub async fn converge_attached_registered_schema( /// initialization. The returned convergence plan carries the LCM status-index /// work for lifecycle-owned daemon maintenance; short-lived callers run that /// same work synchronously through [`converge_attached_registered_schema`]. -/// A store whose observation rows this binary refuses is still admitted for -/// its other authorities and returns that refused authority beside the plan. +/// A store whose LCM, workflow, or git correlation shape this binary refuses +/// is admitted untouched for its other authorities; one whose observation +/// rows it refuses is installed without them. Either returns the refused +/// authority instead of a plan. #[hotpath::measure(future = true, label = "global_db.schema.persist.attach")] pub(crate) async fn ensure_attached_registered_schema( database: &Database, -) -> tracedecay_domain::errors::Result<(RegisteredSchemaConvergence, Option)> { +) -> tracedecay_domain::errors::Result { let read_connection = database.read_connection(); let RegisteredSchemaAdmissionClassification { configuration_fresh, temporal_admission, workflow_admission, - } = classify_registered_schema_admission(&read_connection).await?; + } = match classify_registered_schema_admission(&read_connection).await? { + RegisteredSchemaAdmission::Admissible(classification) => classification, + RegisteredSchemaAdmission::SessionAuthorityRefused(refused) => { + return Ok(RegisteredSchemaAttachmentV1::SessionsRefused(refused)); + } + }; let force_exhaustive = !authority_invariant_triggers_intact(&read_connection).await?; let transaction = database .begin_bulk_write_transaction("install attached registered global database schema") @@ -1041,14 +1089,24 @@ pub(crate) async fn ensure_attached_registered_schema( transaction.commit().await?; } validate_admitted_authority_schema(&read_connection, configuration_fresh.is_some()).await?; - Ok(( - RegisteredSchemaConvergence { + Ok(match refused_authority { + Some(refused) => RegisteredSchemaAttachmentV1::SessionsRefused(refused), + None => RegisteredSchemaAttachmentV1::Admitted(RegisteredSchemaConvergence { force_exhaustive, is_fresh: configuration_fresh.is_some(), lcm_status_performance_indexes: true, - }, - refused_authority, - )) + }), + }) +} + +/// An attached store's admission outcome. +pub(crate) enum RegisteredSchemaAttachmentV1 { + /// Every authority admits; the plan completes historical convergence. + Admitted(RegisteredSchemaConvergence), + /// The store serves its other authorities and refuses every session + /// feature until it is reset. It is never converged: its reset deletes + /// it. + SessionsRefused(RefusedAuthorityV1), } #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -1059,7 +1117,7 @@ enum WorkflowSchemaAdmission { async fn inspect_workflow_schema_for_admission( conn: &impl QueryExecutor, -) -> tracedecay_domain::errors::Result { +) -> tracedecay_domain::errors::Result> { let mut rows = conn .query( "SELECT type, name, sql FROM sqlite_master @@ -1088,7 +1146,7 @@ async fn inspect_workflow_schema_for_admission( )); } if tables.is_empty() { - return Ok(WorkflowSchemaAdmission::Create); + return Ok(Ok(WorkflowSchemaAdmission::Create)); } let actual_workflow_tables = tables @@ -1106,9 +1164,9 @@ async fn inspect_workflow_schema_for_admission( .map(|contract| (contract.name, Some(contract.sql))) .collect::>(); if actual_workflow_tables != expected_workflow_tables { - return Err(workflow_schema_reset_required( + return Ok(Err(workflow_schema_refusal( "workflow tables are absent, incomplete, or not exact", - )); + ))); } let mut schema = conn @@ -1124,9 +1182,9 @@ async fn inspect_workflow_schema_for_admission( .await .map_err(|error| global_db_operation_error("read workflow schema identity", error))? else { - return Err(workflow_schema_reset_required( + return Ok(Err(workflow_schema_refusal( "workflow schema identity is missing", - )); + ))); }; let singleton = identity .get::(0) @@ -1147,9 +1205,9 @@ async fn inspect_workflow_schema_for_admission( || definition_digest != WORKFLOW_SCHEMA_DEFINITION_DIGEST_V1 || extra_identity { - return Err(workflow_schema_reset_required( + return Ok(Err(workflow_schema_refusal( "workflow schema identity does not match the final contract", - )); + ))); } for table in WORKFLOW_TABLE_CONTRACTS_V1 { @@ -1189,17 +1247,20 @@ async fn inspect_workflow_schema_for_admission( && actual.3 == expected.primary_key }); if !exact { - return Err(workflow_schema_reset_required( + return Ok(Err(workflow_schema_refusal( "workflow table columns do not match the final contract", - )); + ))); } } - Ok(WorkflowSchemaAdmission::Complete) + Ok(Ok(WorkflowSchemaAdmission::Complete)) } -fn workflow_schema_reset_required(reason: &str) -> tracedecay_domain::errors::TraceDecayError { - tracedecay_domain::errors::TraceDecayError::reset_required("workflow", reason) +fn workflow_schema_refusal(reason: &'static str) -> RefusedAuthorityV1 { + RefusedAuthorityV1::Shape { + authority: "workflow", + reason, + } } pub async fn validate_observation_authority_connection( diff --git a/crates/tracedecay-global-db/src/tests/harness.rs b/crates/tracedecay-global-db/src/tests/harness.rs index cb1ad44b5c..74baddb5c8 100644 --- a/crates/tracedecay-global-db/src/tests/harness.rs +++ b/crates/tracedecay-global-db/src/tests/harness.rs @@ -504,7 +504,7 @@ impl RegisteredGlobalDbHarness { _directory, _scope, }, - convergence, + convergence.expect("an admissible store returns its convergence plan"), ) } } diff --git a/crates/tracedecay-global-db/src/tests/lcm_schema/lcm_schema_contract.rs b/crates/tracedecay-global-db/src/tests/lcm_schema/lcm_schema_contract.rs index e0c053fca0..04b8694cea 100644 --- a/crates/tracedecay-global-db/src/tests/lcm_schema/lcm_schema_contract.rs +++ b/crates/tracedecay-global-db/src/tests/lcm_schema/lcm_schema_contract.rs @@ -35,19 +35,26 @@ async fn stale_or_future_lcm_marker_requires_reset_without_rewriting_marker() { drop(db); set_migration_version(&db_path, found_version).await; - let error = match open_global_db(&db_path).await { - Err(error) => error, - Ok(_) => panic!("incompatible LCM marker must require a reset"), - }; - assert!(matches!( - error, - TraceDecayError::ProfileResetRequired { - component: "LCM", - found_version: Some(actual), - required_version: - tracedecay_lcm::LCM_SCHEMA_VERSION, - } if actual == found_version - )); + let (lease, owner) = open_registered_test_database_fixture( + &db_path, + TestDatabaseRuntimeScope::ProfileSessions, + ) + .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: "LCM", + found_version: Some(actual), + required_version: tracedecay_lcm::LCM_SCHEMA_VERSION, + }) if actual == found_version + ), + "an incompatible LCM marker refuses session features: {refusal:?}" + ); + } + drop((lease, owner)); assert_eq!(schema_version(&db_path).await, found_version); } } diff --git a/crates/tracedecay-store-runtime/src/session_registry/maintenance.rs b/crates/tracedecay-store-runtime/src/session_registry/maintenance.rs index 2340b8285b..90f6bdd794 100644 --- a/crates/tracedecay-store-runtime/src/session_registry/maintenance.rs +++ b/crates/tracedecay-store-runtime/src/session_registry/maintenance.rs @@ -639,16 +639,14 @@ impl DaemonSessionRuntimeRegistryV1 { // reset deletes it. let long_lived = self.long_lived_session_maintenance; let (database, convergence) = if long_lived { - let (database, convergence) = - RegisteredGlobalDbOwnerV1::admit_and_attach_for_daemon(database).await?; - (database, Some(convergence)) + RegisteredGlobalDbOwnerV1::admit_and_attach_for_daemon(database).await? } else { ( RegisteredGlobalDbOwnerV1::admit_and_attach(database).await?, None, ) }; - if long_lived && database.reset_required().is_none() { + if convergence.is_some() { let lease = database.issue_lease().map_err(|error| { session_registry_error( "issue registered schema convergence client", diff --git a/crates/tracedecay/tests/transport_acceptance_suite/typed_terminal_restart_acceptance/stale_sessions_store_reset.rs b/crates/tracedecay/tests/transport_acceptance_suite/typed_terminal_restart_acceptance/stale_sessions_store_reset.rs index d5a30bbae7..138fc9fc50 100644 --- a/crates/tracedecay/tests/transport_acceptance_suite/typed_terminal_restart_acceptance/stale_sessions_store_reset.rs +++ b/crates/tracedecay/tests/transport_acceptance_suite/typed_terminal_restart_acceptance/stale_sessions_store_reset.rs @@ -1,18 +1,19 @@ -//! Session stores whose observation rows predate the unified observation -//! identity are refused as typed reset states without taking code -//! intelligence down with them, and their scoped reset deletes exactly those -//! stores. +//! Session stores whose persisted shape this binary refuses are held as typed +//! reset states without taking code intelligence down with them, and their +//! scoped reset deletes exactly those stores. //! //! A physically spawned `tracedecay daemon run` first writes a real profile -//! and project session store. Both are then given the shape a released binary -//! left behind: observation rows written before the unified identity, with no -//! unified-identity marker. Over that profile the project must still open, the -//! MCP host must initialize and list tools, code search must answer, session -//! reads must return the typed `reset_required` refusal naming -//! `tracedecay wipe --stale --yes`, and `tracedecay doctor` must name that -//! command instead of `tracedecay install`. The scoped reset then deletes both -//! session stores and nothing else, and the restarted daemon serves sessions -//! from empty stores. +//! and project session store. Stores are then given a shape a released binary +//! left behind: observation rows written before the unified identity, an LCM +//! schema version, a git correlation schema version, or a workflow schema +//! identity other than the one this binary writes. Over that profile the +//! project must still open, the MCP host must initialize and list tools, code +//! search and callers must answer, session reads against a refused store must +//! return the typed `reset_required` refusal naming +//! `tracedecay wipe --stale --yes`, and `tracedecay doctor` must count it as a +//! pending operator action naming that command. The scoped reset then deletes +//! exactly the refused stores, every other profile file stays byte-identical, +//! and the restarted daemon serves sessions from empty stores. use std::collections::BTreeMap; use std::io::Write; @@ -116,8 +117,13 @@ fn file_digests(root: &Path) -> BTreeMap { digests } -/// Whether `relative` belongs to one of the two refused session stores. -fn is_session_store_member(relative: &Path, project_store: &Path) -> bool { +/// Whether `relative` belongs to the project session store, or to the profile +/// session store when `with_profile_store`. +fn is_session_store_member( + relative: &Path, + project_store: &Path, + with_profile_store: bool, +) -> bool { let name = relative.to_string_lossy(); let in_project_sessions = relative.starts_with(project_store) && relative @@ -129,8 +135,9 @@ fn is_session_store_member(relative: &Path, project_store: &Path) -> bool { first.starts_with("sessions.") || first == ".sessions.db.host-admission" }); in_project_sessions - || name.starts_with("user-sessions.") - || name.starts_with(".user-sessions.db.host-admission") + || (with_profile_store + && (name.starts_with("user-sessions.") + || name.starts_with(".user-sessions.db.host-admission"))) } fn status(home: &Path, project: &Path) -> Value { @@ -298,8 +305,39 @@ fn session_status(home: &Path, project: &Path, storage_scope: &str) -> Value { ) } -#[test] -fn stale_session_stores_refuse_sessions_only_until_their_scoped_reset() { +/// Rewrites exactly one recorded shape row of a stopped session store. +fn execute_once(db_path: &Path, sql: &str) { + let connection = rusqlite::Connection::open(db_path).expect("open the session store"); + assert_eq!( + connection.execute(sql, []).expect("age the session store"), + 1, + "`{sql}` rewrites exactly one recorded shape row of {}", + db_path.display() + ); +} + +/// One way a released binary's session store differs from the shape this +/// binary writes, and the exact census entry the daemon reports for it. +struct SessionStoreRefusal { + /// Gives one stopped session store the released shape. + age: fn(&Path), + /// Also ages the profile session store, not only the project's. + ages_profile_store: bool, + authority: &'static str, + found_version: Value, + required_version: Value, + reason: &'static str, + /// A session tool reading the refused store. + session_tool: &'static str, + session_tool_args: fn() -> Value, +} + +/// Ages the session stores `refusal` names on a stopped daemon, then proves +/// code intelligence and MCP serve over them, session reads against them +/// refuse typed while every admissible session store keeps serving, doctor +/// counts each as a pending operator action, and `wipe --stale` deletes +/// exactly those stores and leaves every other profile file byte-identical. +fn refused_session_stores_serve_code_until_their_scoped_reset(refusal: &SessionStoreRefusal) { let home = tempfile::TempDir::new().expect("isolated home"); let home_path = canonical_existing_path(home.path()); let project = tempfile::TempDir::new().expect("project"); @@ -307,9 +345,17 @@ fn stale_session_stores_refuse_sessions_only_until_their_scoped_reset() { let profile_root = home_path.join(".tracedecay"); let project_id = tracedecay_runtime_core::storage::default_profile_project_id(&project_path); let project_store = PathBuf::from("projects").join(&project_id); + let mut aged = vec![("project", format!("project sessions {project_id}"))]; + if refusal.ages_profile_store { + aged.insert(0, ("user", "profile sessions".to_owned())); + } + let admissible_scopes: Vec<&str> = ["project", "user"] + .into_iter() + .filter(|scope| aged.iter().all(|(aged_scope, _)| aged_scope != scope)) + .collect(); let mut daemon = spawn_tracedecay_daemon_with(&home_path, |_| {}); - super::initialize_project(&home_path, &project_path, "stale-sessions-store-reset"); + super::initialize_project(&home_path, &project_path, "refused-session-stores"); wait_for_code_index_hit(&home_path, &project_path, "probe"); for scope in ["project", "user"] { let served = session_status(&home_path, &project_path, scope); @@ -322,16 +368,17 @@ fn stale_session_stores_refuse_sessions_only_until_their_scoped_reset() { .kill_and_wait() .expect("stop the daemon that wrote the profile"); - seed_pre_unified_observation_rows(&profile_root.join("user-sessions.db")); - seed_pre_unified_observation_rows(&profile_root.join(&project_store).join("sessions.db")); + (refusal.age)(&profile_root.join(&project_store).join("sessions.db")); + if refusal.ages_profile_store { + (refusal.age)(&profile_root.join("user-sessions.db")); + } let mut daemon = spawn_tracedecay_daemon_with(&home_path, |_| {}); - let (initialize, tools) = mcp_initialize_and_list_tools(&home_path, &project_path); assert_eq!( initialize.get("error"), None, - "MCP initialize must serve over stale session stores: {initialize}" + "MCP initialize must serve over refused session stores: {initialize}" ); let tool_names: Vec<&str> = tools["result"]["tools"] .as_array() @@ -360,7 +407,7 @@ fn stale_session_stores_refuse_sessions_only_until_their_scoped_reset() { &json!({ "node_id": probe_id, "format": "json" }), ), json!("evidence"), - "callers must serve over stale session stores" + "callers must serve over refused session stores" ); assert_eq!( code_read_outcome( @@ -370,27 +417,22 @@ fn stale_session_stores_refuse_sessions_only_until_their_scoped_reset() { &json!({ "file": "src/lib.rs", "format": "json" }), ), json!("evidence"), - "file dependents must serve over stale session stores" + "file dependents must serve over refused session stores" ); - let expected_stale = vec![ - json!({ - "store": "profile sessions", - "authority": "observations", - "found_version": null, - "required_version": null, - "reason": OBSERVATIONS_RESET_REASON, - "remedy": STALE_STORE_RESET, - }), - json!({ - "store": format!("project sessions {project_id}"), - "authority": "observations", - "found_version": null, - "required_version": null, - "reason": OBSERVATIONS_RESET_REASON, - "remedy": STALE_STORE_RESET, - }), - ]; - wait_for_reset_required_stores(&home_path, &project_path, &expected_stale); + let expected_census: Vec = aged + .iter() + .map(|(_, store)| { + json!({ + "store": store, + "authority": refusal.authority, + "found_version": refusal.found_version, + "required_version": refusal.required_version, + "reason": refusal.reason, + "remedy": STALE_STORE_RESET, + }) + }) + .collect(); + wait_for_reset_required_stores(&home_path, &project_path, &expected_census); let project_open = find_key(&status(&home_path, &project_path), "project_open"); assert!( project_open @@ -399,21 +441,47 @@ fn stale_session_stores_refuse_sessions_only_until_their_scoped_reset() { "project open must not stall on a session-store verdict: {project_open:?}" ); - for scope in ["project", "user"] { + for (scope, _) in &aged { let refused = super::cli_problem_envelope( &session_status(&home_path, &project_path, scope), - &format!("{scope} session read over a stale store"), + &format!("{scope} session read over a refused store"), ); super::assert_reset_required(&refused, &format!("{scope} session read")); assert_eq!( - refused["problem"]["detail"]["authority"], "observations", - "{scope} session read names the refused authority: {refused}" + ( + &refused["problem"]["detail"]["authority"], + &refused["problem"]["detail"]["remedy"] + ), + (&json!(refusal.authority), &json!(STALE_STORE_RESET)), + "{scope} session read names the refused authority and the scoped reset: {refused}" ); - assert_eq!( - refused["problem"]["detail"]["remedy"], STALE_STORE_RESET, - "{scope} session read names the scoped reset: {refused}" + } + for scope in &admissible_scopes { + let served = session_status(&home_path, &project_path, scope); + assert!( + find_key(&served, "problem").is_none_or(|problem| problem.is_null()), + "the admissible {scope} session store keeps serving: {served}" ); } + let session_tool_problem = || { + find_key( + &super::tool_call( + &home_path, + &project_path, + refusal.session_tool, + &(refusal.session_tool_args)(), + ), + "problem", + ) + .filter(|problem| !problem.is_null()) + .map(|problem| problem["kind"].clone()) + }; + assert_eq!( + session_tool_problem(), + Some(json!("reset_required")), + "{} must refuse typed over the refused store", + refusal.session_tool + ); let doctor = tracedecay_command_with_home(&home_path) .arg("doctor") @@ -426,13 +494,11 @@ fn stale_session_stores_refuse_sessions_only_until_their_scoped_reset() { String::from_utf8_lossy(&doctor.stdout), String::from_utf8_lossy(&doctor.stderr) ); - for store in [ - "profile sessions".to_owned(), - format!("project sessions {project_id}"), - ] { + for (_, store) in &aged { let line = format!( - "Store {store} requires reset ({OBSERVATIONS_RESET_REASON}). Pending operator \ - action: run `{STALE_STORE_RESET}`" + "Store {store} requires reset ({}). Pending operator action: run \ + `{STALE_STORE_RESET}`", + refusal.reason ); assert!( doctor_text.contains(&line), @@ -441,11 +507,11 @@ fn stale_session_stores_refuse_sessions_only_until_their_scoped_reset() { } assert!( doctor_text.contains("pending operator action(s)") && doctor_text.contains("no issues."), - "stale session stores are pending operator actions, not issues:\n{doctor_text}" + "refused session stores are pending operator actions, not issues:\n{doctor_text}" ); assert!( !doctor_text.contains("to fix most issues"), - "doctor must not send a stale session store to `tracedecay install`:\n{doctor_text}" + "doctor must not send a refused session store to `tracedecay install`:\n{doctor_text}" ); assert!( !doctor_text.contains("Stalled"), @@ -462,15 +528,17 @@ fn stale_session_stores_refuse_sessions_only_until_their_scoped_reset() { "the scoped reset failed:\n{reset_output}" ); let after_reset = file_digests(&profile_root); - let (stale_members, kept): (BTreeMap<_, _>, BTreeMap<_, _>) = before_reset - .into_iter() - .partition(|(relative, _)| is_session_store_member(relative, &project_store)); + let (refused_members, kept): (BTreeMap<_, _>, BTreeMap<_, _>) = + before_reset.into_iter().partition(|(relative, _)| { + is_session_store_member(relative, &project_store, refusal.ages_profile_store) + }); assert!( - stale_members.contains_key(Path::new("user-sessions.db")) - && stale_members.contains_key(&project_store.join("sessions.db")), - "both refused stores existed before the reset: {stale_members:#?}" + refused_members.contains_key(&project_store.join("sessions.db")) + && (refused_members.contains_key(Path::new("user-sessions.db")) + || kept.contains_key(Path::new("user-sessions.db"))), + "every session store existed before the reset: {refused_members:#?}" ); - for relative in stale_members.keys() { + for relative in refused_members.keys() { assert!( !after_reset.contains_key(relative), "the scoped reset left refused store member {} behind", @@ -490,15 +558,16 @@ fn stale_session_stores_refuse_sessions_only_until_their_scoped_reset() { Vec::::new(), "the scoped reset touched files outside the refused stores:\n{reset_output}" ); - for store in [ - "profile sessions".to_owned(), - format!("project sessions {project_id}"), - ] { + for (_, store) in &aged { assert!( reset_output.contains(&format!("reset {store}")), "the scoped reset did not report resetting {store}:\n{reset_output}" ); } + assert!( + refusal.ages_profile_store || !reset_output.contains("reset profile sessions"), + "the scoped reset must leave the admissible profile session store:\n{reset_output}" + ); let mut daemon = spawn_tracedecay_daemon_with(&home_path, |_| {}); wait_for_code_index_hit(&home_path, &project_path, "probe"); @@ -509,6 +578,12 @@ fn stale_session_stores_refuse_sessions_only_until_their_scoped_reset() { "the {scope} session store must serve after the scoped reset: {served}" ); } + assert_eq!( + session_tool_problem(), + None, + "{} must serve after the scoped reset", + refusal.session_tool + ); assert_eq!( reset_required_stores(&home_path, &project_path), Vec::::new(), @@ -517,3 +592,78 @@ fn stale_session_stores_refuse_sessions_only_until_their_scoped_reset() { let _ = daemon.kill_and_wait(); } + +#[test] +fn stale_session_stores_refuse_sessions_only_until_their_scoped_reset() { + refused_session_stores_serve_code_until_their_scoped_reset(&SessionStoreRefusal { + age: seed_pre_unified_observation_rows, + ages_profile_store: true, + authority: "observations", + found_version: Value::Null, + required_version: Value::Null, + reason: OBSERVATIONS_RESET_REASON, + session_tool: "tracedecay_lcm_grep", + session_tool_args: || json!({ "query": "probe", "format": "json" }), + }); +} + +#[test] +fn project_session_store_at_another_lcm_schema_version_refuses_sessions_only() { + refused_session_stores_serve_code_until_their_scoped_reset(&SessionStoreRefusal { + age: |db| { + execute_once( + db, + "UPDATE session_schema_migrations SET version = 12 WHERE name = 'lcm'", + ); + }, + ages_profile_store: false, + authority: "LCM", + found_version: json!(12), + required_version: json!(13), + reason: "LCM profile schema 12 is incompatible with required schema 13; reset the profile", + session_tool: "tracedecay_lcm_grep", + session_tool_args: || json!({ "query": "probe", "format": "json" }), + }); +} + +#[test] +fn project_session_store_at_another_git_correlation_version_refuses_sessions_only() { + refused_session_stores_serve_code_until_their_scoped_reset(&SessionStoreRefusal { + age: |db| { + execute_once( + db, + "UPDATE session_schema_migrations SET version = 5 WHERE name = 'git_correlation'", + ); + }, + ages_profile_store: false, + authority: "git correlation", + found_version: json!(5), + required_version: json!(6), + reason: "git correlation profile schema 5 is incompatible with required schema 6; reset \ + the profile", + session_tool: "tracedecay_sessions_for", + session_tool_args: || json!({ "git_ref": "branch", "value": "main", "format": "json" }), + }); +} + +#[test] +fn project_session_store_with_another_workflow_schema_identity_refuses_sessions_only() { + refused_session_stores_serve_code_until_their_scoped_reset(&SessionStoreRefusal { + // A workflow schema written from another table contract. + age: |db| { + execute_once( + db, + "UPDATE workflow_schema SET definition_digest = \ + 'sha256:0000000000000000000000000000000000000000000000000000000000000000'", + ); + }, + ages_profile_store: false, + authority: "workflow", + found_version: Value::Null, + required_version: Value::Null, + reason: "workflow persisted shape requires reset: workflow schema identity does not \ + match the final contract", + session_tool: "tracedecay_workflow_list_definitions", + session_tool_args: || json!({}), + }); +}