feat(hotpath): fill session-temporal-store query/persist gaps

crates/tracedecay-session-temporal-store/src/cursor_keys.rs

@@ -59,6 +59,7 @@ pub struct GlobalDbCursorKeyProvider {
     authenticators: Vec<(SignedCursorKeyRefV1, InMemoryCursorAuthenticator)>,
 }
 
+#[hotpath::measure(future = true, label = "session_temporal.cursor_key.provision")]
 pub(super) async fn ensure_active_session_cursor_key_in_transaction(
     transaction: &impl crate::handle::SessionTemporalExec,
 ) -> SessionStoreResult<SignedCursorKeyRefV1> {
@@ -195,6 +196,7 @@ pub(super) async fn ensure_active_session_cursor_key_in_transaction(
 }
 
 impl GlobalDbCursorKeyProvider {
+    #[hotpath::measure(future = true, label = "session_temporal.cursor_key.load_active")]
     pub async fn from_registered_active(
         read: &DatabaseEngineReadSnapshot,
     ) -> Result<Self, GlobalDbCursorKeyProviderError> {
@@ -252,6 +254,7 @@ impl GlobalDbCursorKeyProvider {
         Self::from_registered_key_ref_at(read, expected, now_micros().0).await
     }
 
+    #[hotpath::measure(future = true, label = "session_temporal.cursor_key.load")]
     async fn from_registered_key_ref_at(
         read: &DatabaseEngineReadSnapshot,
         expected: SignedCursorKeyRefV1,
@@ -307,6 +310,7 @@ impl GlobalDbCursorKeyProvider {
             authenticators,
         })
     }
+    #[hotpath::measure(label = "session_temporal.cursor_key.keyring")]
     pub fn retrieval_keyring(
         &self,
         privacy_domain: PrivacyDomainId,

crates/tracedecay-session-temporal-store/src/execution.rs

@@ -131,6 +131,7 @@ impl AuthorizedTemporalExecutionRequest {
         }
     }
 
+    #[hotpath::measure(label = "session_temporal.execution.validate_report")]
     pub fn validates_report(&self, report: &SessionTemporalExecutionReport) -> bool {
         let snapshot = &report.result().snapshot;
         let actual = snapshot.request();
@@ -319,6 +320,7 @@ pub struct AuthorizedTaskSessionExecutionRequestV1 {
 
 impl AuthorizedTaskSessionExecutionRequestV1 {
     #[allow(clippy::too_many_arguments)]
+    #[hotpath::measure(label = "session_temporal.execution.authorize_task_session")]
     pub fn new(
         temporal: AuthorizedTemporalExecutionRequest,
         retrieval: RetrievalRequest,

crates/tracedecay-session-temporal-store/src/hydration.rs

@@ -126,6 +126,7 @@ impl<B> SessionTemporalHydrationAdapter<B> {
 }
 
 impl<B: TemporalHydrationBackend> SessionTemporalHydrationAdapter<B> {
+    #[hotpath::measure(future = true, label = "session_temporal.hydrate.authorize")]
     async fn authorize(
         &self,
         snapshot: &TemporalExecutionSnapshot,
@@ -142,6 +143,7 @@ impl<B: TemporalHydrationBackend> SessionTemporalHydrationAdapter<B> {
         }
     }
 
+    #[hotpath::measure(future = true, label = "session_temporal.hydrate.read_authorized")]
     async fn read_after_recheck(
         &self,
         snapshot: &TemporalExecutionSnapshot,
@@ -345,6 +347,7 @@ impl<'snapshot> SessionTemporalHydrationAdapter<GlobalDbHydrationBackend<'snapsh
     }
 }
 
+#[hotpath::measure(future = true, label = "session_temporal.hydrate.decode_message")]
 pub(super) async fn session_message_from_hydrated_bytes(
     read: &TemporalSqlRead<'_>,
     snapshot: &TemporalExecutionSnapshot,
@@ -621,6 +624,7 @@ impl TemporalHydrationBackend for GlobalDbHydrationBackend<'_> {
 }
 
 #[allow(clippy::too_many_arguments)]
+#[hotpath::measure(future = true, label = "session_temporal.hydrate.read_occurrence")]
 async fn read_occurrence_content(
     conn: &TemporalSqlRead<'_>,
     storage_root: &Path,
@@ -815,6 +819,7 @@ async fn resolve_current(
     ))
 }
 
+#[hotpath::measure(future = true, label = "session_temporal.hydrate.resolve_occurrence")]
 async fn resolve_occurrence(
     conn: &TemporalSqlRead<'_>,
     snapshot: &TemporalExecutionSnapshot,
@@ -933,6 +938,7 @@ async fn resolve_occurrence(
     })))
 }
 
+#[hotpath::measure(future = true, label = "session_temporal.hydrate.resolve_summary")]
 async fn resolve_summary(
     conn: &TemporalSqlRead<'_>,
     relation_authority: Option<&SessionHydrationRelationAuthority<'_>>,
@@ -1081,6 +1087,7 @@ async fn resolve_summary(
 }
 
 #[allow(clippy::too_many_arguments)]
+#[hotpath::measure(future = true, label = "session_temporal.hydrate.summary_evidence")]
 async fn summary_has_provider_evidence(
     conn: &TemporalSqlRead<'_>,
     relation_store: &SessionRelationGraphStore,

crates/tracedecay-session-temporal-store/src/operations/message_anchor.rs

@@ -31,6 +31,7 @@ pub(super) type ResolvedMessageAnchor = (String, bool, i64);
 /// `Ok(None)` means the message has no canonical anchor in this store at all —
 /// the only case in which the publication falls back to a legacy compatibility
 /// anchor.
+#[hotpath::measure(future = true, label = "session_temporal.publication.resolve_anchor")]
 pub(super) async fn resolve_message_anchor(
     conn: &impl crate::handle::SessionTemporalExec,
     provider: &str,
@@ -47,6 +48,7 @@ pub(super) async fn resolve_message_anchor(
 }
 
 /// Resolves through the message's occurrence in the active temporal generation.
+#[hotpath::measure(future = true, label = "session_temporal.publication.anchor_occurrence")]
 async fn resolve_materialized_occurrence(
     conn: &impl crate::handle::SessionTemporalExec,
     provider: &str,
@@ -124,6 +126,7 @@ async fn resolve_materialized_occurrence(
 
 /// Resolves through the durable observation authority, which retains the
 /// exact-observation anchor before any generation materializes the occurrence.
+#[hotpath::measure(future = true, label = "session_temporal.publication.anchor_observation")]
 async fn resolve_canonical_observation(
     conn: &impl crate::handle::SessionTemporalExec,
     provider: &str,

crates/tracedecay-session-temporal-store/src/projection/persist.rs

@@ -41,6 +41,7 @@ fn observation_envelope(
         .map_err(|error| storage(PERSIST_OPERATION, error))
 }
 
+#[hotpath::measure(future = true, label = "session_temporal.projection.record_count")]
 pub async fn session_temporal_projection_record_count(
     conn: &impl crate::handle::SessionTemporalQuery,
     session_id: &SessionId,
@@ -266,6 +267,7 @@ async fn canonical_occurrence_projection(
     })
 }
 
+#[hotpath::measure(future = true, label = "session_temporal.persist.occurrences")]
 pub(super) async fn persist_occurrences(
     conn: &impl crate::handle::SessionTemporalExec,
     batch: &SessionTemporalProjectionBatchV1,
@@ -509,6 +511,7 @@ async fn persist_occurrence(
 }
 
 /// Derives the canonical occurrence for one already-resolved projection output.
+#[hotpath::measure(future = true, label = "session_temporal.projection.canonical_occurrence")]
 pub(super) async fn canonical_occurrence(
     conn: &impl crate::handle::SessionTemporalQuery,
     observation: &tracedecay_domain::DurableObservationV1,
@@ -703,6 +706,7 @@ pub(super) async fn ensure_agent(
     Ok(())
 }
 
+#[hotpath::measure(future = true, label = "session_temporal.projection.require_occurrence")]
 pub(super) async fn require_exact_occurrence(
     conn: &impl crate::handle::SessionTemporalExec,
     batch: &SessionTemporalProjectionBatchV1,
@@ -791,6 +795,7 @@ pub(super) async fn require_exact_occurrence(
     Ok(())
 }
 
+#[hotpath::measure(future = true, label = "session_temporal.projection.validate_copy")]
 pub(super) async fn validate_copy(
     conn: &impl crate::handle::SessionTemporalExec,
     batch: &SessionTemporalProjectionBatchV1,
@@ -908,6 +913,7 @@ pub(super) async fn occurrence_observation_and_anchor(
     Ok((observation, envelope, anchor_id))
 }
 
+#[hotpath::measure(future = true, label = "session_temporal.projection.validate_copy_proof")]
 pub(super) async fn validate_copy_proof(
     conn: &impl crate::handle::SessionTemporalExec,
     batch: &SessionTemporalProjectionBatchV1,
@@ -1005,6 +1011,7 @@ pub(super) async fn validate_copy_proof(
     Ok(())
 }
 
+#[hotpath::measure(future = true, label = "session_temporal.persist.assertion")]
 pub(super) async fn persist_assertion(
     conn: &impl crate::handle::SessionTemporalExec,
     batch: &SessionTemporalProjectionBatchV1,
@@ -1094,6 +1101,7 @@ pub(super) async fn persist_assertion(
     Ok(inserted)
 }
 
+#[hotpath::measure(future = true, label = "session_temporal.projection.validate_assertion")]
 pub(super) async fn validate_assertion(
     conn: &impl crate::handle::SessionTemporalExec,
     batch: &SessionTemporalProjectionBatchV1,

crates/tracedecay-session-temporal-store/src/query.rs

@@ -77,6 +77,7 @@ pub(super) fn encode_watermarks(
     .map_err(|error| storage(operation, error))
 }
 
+#[hotpath::measure(future = true, label = "session_temporal.query.read_generation")]
 pub(super) async fn read_generation(
     conn: &impl crate::handle::SessionTemporalQuery,
     session_id: &SessionId,
@@ -106,6 +107,7 @@ fn decode_generation(row: &Row, operation: &'static str) -> SessionStoreResult<G
     })
 }
 
+#[hotpath::measure(future = true, label = "session_temporal.query.read_active_generation")]
 pub(super) async fn read_active_generation(
     conn: &impl crate::handle::SessionTemporalQuery,
     session_id: &SessionId,
@@ -150,6 +152,7 @@ pub(super) async fn require_active_generation(
     }
 }
 
+#[hotpath::measure(future = true, label = "session_temporal.query.read_observation")]
 pub(super) async fn read_observation(
     conn: &impl crate::handle::SessionTemporalQuery,
     observation_id: &CanonicalObservationIdV1,
@@ -196,6 +199,7 @@ const OBSERVATION_READ_BATCH: usize = 500;
 /// would have returned. Ids that are absent are simply missing from the map:
 /// reporting a missing observation stays with the caller so it still surfaces in
 /// the caller's own iteration order, with the same message.
+#[hotpath::measure(future = true, label = "session_temporal.query.read_observations")]
 pub(super) async fn read_observations(
     conn: &impl crate::handle::SessionTemporalQuery,
     observation_ids: &[CanonicalObservationIdV1],

crates/tracedecay-session-temporal-store/src/retrieval/candidates.rs

@@ -230,6 +230,7 @@ impl RootAuthorityDecisions {
 ///
 /// Candidates are denied unless the window query names their ordinal, so a
 /// query that cannot see a candidate never widens the authorized root.
+#[hotpath::measure(future = true, label = "session_temporal.query.root_authority")]
 pub(super) async fn resolve_root_authority(
     conn: &TemporalSqlRead<'_>,
     candidates: &[RankingCandidate],
@@ -343,6 +344,7 @@ pub(super) async fn require_candidate_root_authority(
 }
 
 #[allow(clippy::too_many_arguments)]
+#[hotpath::measure(future = true, label = "session_temporal.query.candidate_clause")]
 pub(super) async fn query_candidate_clause(
     conn: &TemporalSqlRead<'_>,
     scope: &TemporalRetrievalScope,

crates/tracedecay-session-temporal-store/src/sql.rs

@@ -102,6 +102,7 @@ impl<'a> TemporalSqlRead<'a> {
         Self::Registered(read)
     }
 
+    #[hotpath::measure(future = true, label = "session_temporal.sql.query")]
     pub(super) async fn query<P>(&self, sql: &str, params: P) -> engine::Result<TemporalSqlRows>
     where
         P: engine::IntoParams,

crates/tracedecay-session-temporal-store/src/store.rs

@@ -68,6 +68,7 @@ impl<'a, D: SessionTemporalRegisteredDb + Sync> GlobalDbSessionTemporalStore<'a,
         SessionTemporalAccess::new(self.db)
     }
 
+    #[hotpath::measure(future = true, label = "session_temporal.store.persist_refresh_batch")]
     pub async fn persist_session_refresh_projection_batch(
         &self,
         progress: SessionRefreshProgressV1,
@@ -81,6 +82,10 @@ impl<'a, D: SessionTemporalRegisteredDb + Sync> GlobalDbSessionTemporalStore<'a,
             .await
     }
 
+    #[hotpath::measure(
+        future = true,
+        label = "session_temporal.store.persist_refresh_batch_controlled"
+    )]
     pub async fn persist_session_refresh_projection_batch_controlled(
         &self,
         progress: SessionRefreshProgressV1,
@@ -99,6 +104,7 @@ impl<'a, D: SessionTemporalRegisteredDb + Sync> GlobalDbSessionTemporalStore<'a,
             .await
     }
 
+    #[hotpath::measure(future = true, label = "session_temporal.store.refresh_recovery")]
     pub async fn session_refresh_recovery(
         &self,
         session_id: &tracedecay_domain::SessionId,
@@ -108,6 +114,7 @@ impl<'a, D: SessionTemporalRegisteredDb + Sync> GlobalDbSessionTemporalStore<'a,
             .await
     }
 
+    #[hotpath::measure(future = true, label = "session_temporal.store.refresh_running")]
     pub async fn running_session_refreshes(
         &self,
     ) -> SessionStoreResult<Vec<SessionRefreshRecoveryV1>> {