feat(hotpath): fill global-db schema/session/transcript gaps

crates/tracedecay-global-db/src/configuration/store/audit.rs

@@ -29,6 +29,7 @@ pub(super) fn encode_audit_payload(
 
 pub(super) const CONFIGURATION_AUDIT_REDACTION_KEY_BYTES: usize = 32;
 
+#[hotpath::measure(future = true, label = "global_db.configuration.audit.read_key")]
 pub(super) async fn read_audit_redaction_key(
     transaction: &impl QueryExecutor,
 ) -> ConfigurationStoreResult<Option<Zeroizing<Vec<u8>>>> {
@@ -55,6 +56,7 @@ pub(super) async fn read_audit_redaction_key(
     Ok(Some(material))
 }
 
+#[hotpath::measure(future = true, label = "global_db.configuration.audit.ensure_key")]
 pub(super) async fn ensure_audit_redaction_key(
     transaction: &impl Executor,
     created_at: UtcMicros,
@@ -93,6 +95,7 @@ pub(super) fn audit_target_commitment(
         .map_err(ConfigurationStoreError::from)
 }
 
+#[hotpath::measure(future = true, label = "global_db.configuration.audit.seal_target")]
 pub(super) async fn seal_audit_target<T: Serialize>(
     transaction: &impl Executor,
     event_id: &ConfigurationAuditEventId,
@@ -109,6 +112,7 @@ pub(super) async fn seal_audit_target<T: Serialize>(
     Ok((sealed, commitment))
 }
 
+#[hotpath::measure(future = true, label = "global_db.configuration.audit.validate_seal")]
 pub(super) async fn validate_sealed_audit_target(
     transaction: &impl QueryExecutor,
     event: &ConfigurationAuditEvent,
@@ -129,6 +133,7 @@ pub(super) async fn validate_sealed_audit_target(
     Ok(())
 }
 
+#[hotpath::measure(future = true, label = "global_db.configuration.audit.insert_dry_run")]
 pub(super) async fn insert_dry_run_audit_event(
     transaction: &impl Executor,
     record: &ConfigurationProtectedPlanRecordV1,
@@ -267,6 +272,7 @@ pub(super) fn decode_audit_row(
     Ok((event, sealed_target_reference))
 }
 
+#[hotpath::measure(future = true, label = "global_db.configuration.audit.read_event")]
 pub(super) async fn read_audit_event_from_transaction(
     transaction: &impl QueryExecutor,
     event_id: &ConfigurationAuditEventId,
@@ -295,6 +301,7 @@ pub(super) async fn read_audit_event_from_transaction(
     Ok(Some(event))
 }
 
+#[hotpath::measure(future = true, label = "global_db.configuration.audit.insert_event")]
 pub(super) async fn insert_audit_event_with_receipt_digest(
     transaction: &impl Executor,
     event: &ConfigurationAuditEvent,
@@ -354,6 +361,7 @@ pub(super) fn is_terminal_plan_event(event_kind: &str) -> bool {
     matches!(event_kind, "applied" | "rollback_applied")
 }
 
+#[hotpath::measure(future = true, label = "global_db.configuration.audit.append_terminal")]
 pub(super) async fn append_terminal_plan_event(
     transaction: &impl Executor,
     plan: &ProtectedChangePlan,
@@ -423,6 +431,7 @@ pub(super) async fn append_terminal_plan_event(
     Ok(())
 }
 
+#[hotpath::measure(future = true, label = "global_db.configuration.audit.match_terminal")]
 pub(super) async fn has_matching_terminal_plan_event(
     transaction: &impl QueryExecutor,
     plan: &ProtectedChangePlan,
@@ -454,6 +463,7 @@ pub(super) async fn has_matching_terminal_plan_event(
     Ok(terminal_count == 1 && matched)
 }
 
+#[hotpath::measure(future = true, label = "global_db.configuration.audit.read_page")]
 pub(super) async fn audit_from_transaction(
     transaction: &impl QueryExecutor,
     after: Option<&ConfigurationAuditEventId>,

crates/tracedecay-global-db/src/configuration/store/mutation.rs

@@ -144,6 +144,10 @@ pub(super) fn decode_stored_mutation_receipt(
     })
 }
 
+#[hotpath::measure(
+    future = true,
+    label = "global_db.configuration.query.idempotency_receipt"
+)]
 pub(super) async fn receipt_for_idempotency_from_transaction(
     transaction: &impl QueryExecutor,
     actor_id: &ActorId,
@@ -174,6 +178,10 @@ pub(super) async fn receipt_for_idempotency_from_transaction(
     Ok(Some(receipt))
 }
 
+#[hotpath::measure(
+    future = true,
+    label = "global_db.configuration.persist.mutation_receipt"
+)]
 pub(super) async fn insert_mutation_receipt(
     transaction: &impl Executor,
     commit: &ConfigurationCommitV1,
@@ -257,6 +265,7 @@ pub(super) fn validate_commit_bindings(
     Ok(())
 }
 
+#[hotpath::measure(future = true, label = "global_db.configuration.query.replay_match")]
 pub(super) async fn replay_matches_commit(
     transaction: &impl QueryExecutor,
     stored: &StoredMutationReceipt,
@@ -292,6 +301,7 @@ pub(super) async fn replay_matches_commit(
     Ok(true)
 }
 
+#[hotpath::measure(future = true, label = "global_db.configuration.persist.commit")]
 pub(super) async fn commit_configuration_transaction(
     transaction: &impl Executor,
     commit: &ConfigurationCommitV1,
@@ -310,6 +320,10 @@ pub(super) async fn commit_configuration_transaction(
     .await
 }
 
+#[hotpath::measure(
+    future = true,
+    label = "global_db.configuration.persist.commit_with_registry"
+)]
 pub(super) async fn commit_configuration_transaction_with_registry(
     transaction: &impl Executor,
     commit: &ConfigurationCommitV1,
@@ -596,6 +610,7 @@ pub(super) struct ConfigurationCommitDraft<'a, T> {
     pub(super) target: &'a T,
 }
 
+#[hotpath::measure(future = true, label = "global_db.configuration.persist.build_commit")]
 pub(super) async fn build_configuration_commit<T: Serialize>(
     transaction: &impl Executor,
     draft: ConfigurationCommitDraft<'_, T>,
@@ -780,6 +795,7 @@ pub(super) fn rollback_redacted_changes(
         .collect()
 }
 
+#[hotpath::measure(future = true, label = "global_db.configuration.query.current_state")]
 pub(super) async fn current_state_from_transaction(
     transaction: &impl QueryExecutor,
 ) -> Result<ConfigurationCurrentStateV1, ConfigurationError> {
@@ -798,6 +814,7 @@ pub(super) async fn current_state_from_transaction(
     })
 }
 
+#[hotpath::measure(future = true, label = "global_db.configuration.query.replay_control")]
 pub(super) async fn replay_control_receipt(
     transaction: &impl QueryExecutor,
     actor_id: &ActorId,
@@ -848,6 +865,7 @@ pub struct ConfigurationDirectCommitOutcomeV1 {
     pub current: ConfigurationCurrentStateV1,
 }
 
+#[hotpath::measure(future = true, label = "global_db.configuration.persist.commit_direct")]
 pub async fn commit_direct_in_transaction<E>(
     transaction: &E,
     authority: &ConfigurationMutationAuthority,
@@ -867,6 +885,10 @@ where
     .await
 }
 
+#[hotpath::measure(
+    future = true,
+    label = "global_db.configuration.persist.commit_direct_registry"
+)]
 pub(super) async fn commit_direct_in_transaction_with_registry<E>(
     transaction: &E,
     authority: &ConfigurationMutationAuthority,

crates/tracedecay-global-db/src/observation_projection/rebuild.rs

@@ -263,6 +263,10 @@ fn projection_retry_delay_micros(attempt_count: u32) -> i64 {
         .min(PROJECTION_RETRY_MAX_MICROS)
 }
 
+#[hotpath::measure(
+    future = true,
+    label = "global_db.observation_projection.persist.retry"
+)]
 async fn persist_projection_retry_on_database(
     database: &Database,
     observation_id: &CanonicalObservationIdV1,
@@ -293,6 +297,10 @@ async fn persist_projection_retry_on_database(
         .map_err(|error| storage("commit projection retry transaction", error))
 }
 
+#[hotpath::measure(
+    future = true,
+    label = "global_db.observation_projection.persist.rejection"
+)]
 async fn persist_projection_rejection_on_database(
     database: &Database,
     observation_id: &CanonicalObservationIdV1,

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

@@ -54,6 +54,7 @@ impl RegisteredGlobalDb {
     /// bound to `work_attempt`. The query is authorization-scoped by project,
     /// keyed by a canonical identity digest, and capped before any data leaves
     /// the reader. It never derives a Work identity from `owner_event_id`.
+    #[hotpath::measure(future = true, label = "global_db.delivery_settlement.query.censuses")]
     pub async fn work_attempt_delivery_censuses(
         &self,
         project_id: &str,
@@ -122,6 +123,7 @@ impl RegisteredGlobalDb {
     /// Durably records one concrete recipient at the completion boundary that
     /// observed it. Callers choose whether that boundary is pre-write or
     /// post-hoc and must describe their timing truthfully.
+    #[hotpath::measure(future = true, label = "global_db.delivery_settlement.persist.begin")]
     pub async fn begin_delivery_attempt(
         &self,
         project_id: &str,
@@ -133,6 +135,10 @@ impl RegisteredGlobalDb {
 
     /// Durably binds an opaque source acknowledgement token to the exact
     /// admitted recipient in the same transaction as attempt admission.
+    #[hotpath::measure(
+        future = true,
+        label = "global_db.delivery_settlement.persist.begin_receipted"
+    )]
     pub async fn begin_receipted_delivery_attempt(
         &self,
         project_id: &str,
@@ -208,6 +214,10 @@ impl RegisteredGlobalDb {
 
     /// Resolves a project-scoped source acknowledgement token to the exact
     /// durable attempt admitted for it. Unknown tokens remain a typed absence.
+    #[hotpath::measure(
+        future = true,
+        label = "global_db.delivery_settlement.query.source_receipt"
+    )]
     pub async fn delivery_attempt_for_source_receipt(
         &self,
         project_id: &str,
@@ -283,6 +293,10 @@ impl RegisteredGlobalDb {
     /// Reads one bounded due page of pending opaque source receipts. This is
     /// an indexed deadline scan; callers advance durable state by settling the
     /// returned exact attempts and may repeat until the page is empty.
+    #[hotpath::measure(
+        future = true,
+        label = "global_db.delivery_settlement.query.pending_due"
+    )]
     pub async fn pending_receipted_delivery_attempts_due(
         &self,
         project_id: &str,
@@ -348,6 +362,7 @@ impl RegisteredGlobalDb {
 
     /// CASes an admitted recipient to one immutable terminal outcome and
     /// returns the exact bounded surface census from the same transaction.
+    #[hotpath::measure(future = true, label = "global_db.delivery_settlement.persist.settle")]
     pub async fn settle_delivery_attempt(
         &self,
         project_id: &str,
@@ -496,6 +511,10 @@ fn validate_source_receipt_ref(source_receipt_ref: &str) -> Result<(), String> {
     Ok(())
 }
 
+#[hotpath::measure(
+    future = true,
+    label = "global_db.delivery_settlement.persist.bind_receipt"
+)]
 async fn bind_source_receipt(
     transaction: &RegisteredGlobalDbWriteTransaction<'_>,
     project_id: &str,
@@ -586,6 +605,10 @@ async fn bind_source_receipt(
     Ok(())
 }
 
+#[hotpath::measure(
+    future = true,
+    label = "global_db.delivery_settlement.persist.fanout_identity"
+)]
 async fn ensure_fanout_identity(
     transaction: &RegisteredGlobalDbWriteTransaction<'_>,
     project_id: &str,
@@ -696,6 +719,10 @@ impl StoredDeliveryAttempt {
     }
 }
 
+#[hotpath::measure(
+    future = true,
+    label = "global_db.delivery_settlement.query.read_attempt"
+)]
 async fn read_delivery_attempt(
     transaction: &RegisteredGlobalDbWriteTransaction<'_>,
     project_id: &str,
@@ -755,6 +782,10 @@ async fn read_delivery_attempt(
     }))
 }
 
+#[hotpath::measure(
+    future = true,
+    label = "global_db.delivery_settlement.query.count_attempts"
+)]
 async fn count_attempts(
     transaction: &RegisteredGlobalDbWriteTransaction<'_>,
     project_id: &str,
@@ -784,6 +815,10 @@ async fn count_attempts(
         })
 }
 
+#[hotpath::measure(
+    future = true,
+    label = "global_db.delivery_settlement.query.pending_census"
+)]
 async fn read_pending_delivery_census(
     transaction: &RegisteredGlobalDbWriteTransaction<'_>,
     project_id: &str,

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

@@ -13,12 +13,14 @@ pub(crate) use super::{SessionProviderCoverage, SessionProviderCoverageState};
 pub(crate) use tracedecay_sessions::runtime::store_access::SESSION_MESSAGES_AFTER_SQL;
 
 impl RegisteredGlobalDb {
+    #[hotpath::measure(future = true, label = "global_db.sessions.ingest_health")]
     pub async fn cursor_session_ingest_health(&self) -> Result<SessionIngestHealth, String> {
         SessionStoreAccess::new(self)
             .cursor_session_ingest_health()
             .await
     }
 
+    #[hotpath::measure(future = true, label = "global_db.sessions.ingest_health_provider")]
     pub async fn session_ingest_health_for_provider(
         &self,
         provider: Option<&str>,
@@ -28,6 +30,7 @@ impl RegisteredGlobalDb {
             .await
     }
 
+    #[hotpath::measure(future = true, label = "global_db.sessions.has_message")]
     pub async fn has_session_message(
         &self,
         provider: &str,
@@ -38,10 +41,12 @@ impl RegisteredGlobalDb {
             .await
     }
 
+    #[hotpath::measure(future = true, label = "global_db.sessions.message_count")]
     pub async fn session_message_count(&self) -> Result<i64, String> {
         SessionStoreAccess::new(self).session_message_count().await
     }
 
+    #[hotpath::measure(future = true, label = "global_db.sessions.project_message_count")]
     pub async fn session_message_count_for_project(
         &self,
         project_key: &str,
@@ -51,6 +56,7 @@ impl RegisteredGlobalDb {
             .await
     }
 
+    #[hotpath::measure(future = true, label = "global_db.sessions.messages_after")]
     pub async fn session_messages_after(
         &self,
         provider: &str,
@@ -68,6 +74,7 @@ impl RegisteredGlobalDb {
     /// `Ok(None)` is the truthful "this store holds no timestamped messages";
     /// a failed query or an unreadable timestamp stays an error rather than
     /// masquerading as an idle store.
+    #[hotpath::measure(future = true, label = "global_db.sessions.latest_activity")]
     pub async fn latest_session_activity_secs(
         &self,
     ) -> tracedecay_domain::errors::Result<Option<i64>> {
@@ -78,6 +85,7 @@ impl RegisteredGlobalDb {
 
     /// Reads one message by provider and id. `Ok(None)` is truthful absence;
     /// snapshot, query, and row-decode failures stay typed errors.
+    #[hotpath::measure(future = true, label = "global_db.sessions.get_message")]
     pub async fn get_session_message(
         &self,
         provider: &str,
@@ -92,6 +100,7 @@ impl RegisteredGlobalDb {
     ///
     /// `Ok(vec![])` is the truthful "nothing matched"; snapshot, query, and
     /// row-decode failures are typed errors instead of an empty result page.
+    #[hotpath::measure(future = true, label = "global_db.sessions.search_messages")]
     pub async fn search_session_messages(
         &self,
         provider: &str,
@@ -107,6 +116,7 @@ impl RegisteredGlobalDb {
     /// Lists each session's latest canonical goal state, newest first.
     /// Goals with no native timestamp rank after all timestamped goals
     /// instead of being assigned a fabricated epoch-zero time.
+    #[hotpath::measure(future = true, label = "global_db.sessions.recent_goals")]
     pub async fn recent_session_goals(
         &self,
         project_key: Option<&str>,
@@ -117,6 +127,7 @@ impl RegisteredGlobalDb {
             .await
     }
 
+    #[hotpath::measure(future = true, label = "global_db.sessions.workflow_facts")]
     pub async fn workflow_fact_rows(
         &self,
     ) -> Result<Vec<(String, Option<String>, Option<String>)>, TraceDecayError> {