feat(hotpath): fill sessions ingest/transcript/backfill gaps

crates/tracedecay-sessions/src/runtime/git_correlation/backfill/bounded.rs

@@ -122,6 +122,7 @@ pub struct BoundedBackfillOutcome {
     pub interruption: Option<BoundedBackfillInterruption>,
 }
 
+#[hotpath::measure(label = "sessions.git_correlation.backfill.bounded_page", future = true)]
 pub async fn run_bounded_history_index_page<S>(
     session_store: &S,
     opts: &BackfillOptions,
@@ -409,6 +410,7 @@ pub(super) enum StreamGitEvidenceOutcome {
     Skip(BackfillSkipReason),
 }
 
+#[hotpath::measure(label = "sessions.git_correlation.backfill.stream_evidence", future = true)]
 async fn stream_git_evidence<S: GitCorrelationSessionStore>(
     session_store: &S,
     row: &SessionActivityRow,
@@ -519,6 +521,7 @@ async fn stream_git_evidence<S: GitCorrelationSessionStore>(
     resume_git_evidence(session_store, key, opts, control, stats, committed).await
 }
 
+#[hotpath::measure(label = "sessions.git_correlation.backfill.resume_evidence", future = true)]
 async fn resume_git_evidence<S: GitCorrelationSessionStore>(
     session_store: &S,
     key: GitHistoryProgressKey,
@@ -626,6 +629,7 @@ async fn resume_git_evidence<S: GitCorrelationSessionStore>(
     }
 }
 
+#[hotpath::measure(label = "sessions.git_correlation.backfill.dry_run", future = true)]
 async fn dry_run_native_history(
     project_path: &std::path::Path,
     window_start: i64,
@@ -807,6 +811,7 @@ async fn dry_run_segment(
     Ok(emitted)
 }
 
+#[hotpath::measure(label = "sessions.git_correlation.backfill.frontier_persist", future = true)]
 async fn persist_frontier<S: GitCorrelationSessionStore>(
     session_store: &S,
     candidate: GitHistoryIndexFrontier,

crates/tracedecay-sessions/src/runtime/git_correlation/backfill/history_progress.rs

@@ -21,6 +21,7 @@ pub(super) const fn initial_reflog_content_chain() -> &'static str {
     INITIAL_REFLOG_CONTENT_CHAIN
 }
 
+#[hotpath::measure(label = "sessions.git_correlation.history_schema", future = true)]
 pub(in super::super) async fn install_final_schema(
     conn: &(impl Executor + ?Sized),
 ) -> Result<(), GitCorrelationError> {
@@ -273,6 +274,7 @@ pub(super) struct GitHistorySeenRow {
     pub oid: String,
 }
 
+#[hotpath::measure(label = "sessions.git_correlation.history_progress_read", future = true)]
 pub(super) async fn read_progress(
     conn: &(impl QueryExecutor + ?Sized),
     key: GitHistoryProgressKey,
@@ -298,6 +300,7 @@ pub(super) async fn read_progress(
         .transpose()
 }
 
+#[hotpath::measure(label = "sessions.git_correlation.history_progress_oldest", future = true)]
 pub(super) async fn read_oldest_progress(
     conn: &(impl QueryExecutor + ?Sized),
 ) -> Result<Option<GitHistoryProgressRow>, GitCorrelationError> {
@@ -324,6 +327,7 @@ pub(super) async fn read_oldest_progress(
 }
 
 /// Inserts a new exact progress row without replacing an existing source seal.
+#[hotpath::measure(label = "sessions.git_correlation.history_progress_insert", future = true)]
 pub(super) async fn insert_progress(
     conn: &(impl Executor + ?Sized),
     progress: &GitHistoryProgressRow,
@@ -355,6 +359,7 @@ pub(super) async fn insert_progress(
 }
 
 /// Advances only mutable cursor fields when both generation and source seal match.
+#[hotpath::measure(label = "sessions.git_correlation.history_progress_cas", future = true)]
 pub(super) async fn compare_and_swap_progress(
     conn: &(impl Executor + ?Sized),
     expected_generation: u64,
@@ -546,6 +551,7 @@ pub(super) async fn read_segment(
 }
 
 /// Inserts a segment or updates only its mutable flags when its sealed shape matches.
+#[hotpath::measure(label = "sessions.git_correlation.history_segment_upsert", future = true)]
 pub(super) async fn upsert_segment(
     conn: &(impl Executor + ?Sized),
     segment: &GitHistorySegmentRow,
@@ -579,6 +585,7 @@ pub(super) async fn upsert_segment(
     Ok(changed == 1)
 }
 
+#[hotpath::measure(label = "sessions.git_correlation.history_pending_page", future = true)]
 pub(super) async fn read_pending_page(
     conn: &(impl QueryExecutor + ?Sized),
     key: GitHistoryProgressKey,

crates/tracedecay-sessions/src/runtime/hosts/cursor.rs

@@ -664,6 +664,7 @@ pub async fn try_ingest_cursor_user_transcript_event_capped_with_registered_root
     .await
 }
 
+#[hotpath::measure(label = "sessions.hosts.cursor.ingest_user_event_capped", future = true)]
 pub async fn try_ingest_cursor_user_transcript_event_capped_with_admission(
     event_json: &str,
     admission: &dyn HostAdmission,
@@ -807,6 +808,7 @@ pub(in crate::runtime) async fn try_ingest_cursor_user_sweep_capped_with_session
     .await
 }
 
+#[hotpath::measure(label = "sessions.hosts.cursor.sweep_admit", future = true)]
 async fn admit_cursor_sweep_observations_with_session_ids(
     source: &CursorSweepSource,
     project_root: &Path,
@@ -995,6 +997,7 @@ impl TranscriptSource for CursorSweepSource {
         "cursor"
     }
 
+    #[hotpath::measure(label = "sessions.hosts.cursor.sweep_discover")]
     fn transcript_paths(&self, project_root: &Path) -> Vec<PathBuf> {
         if let Some(registered_slugs) = &self.user_registered_slugs {
             let Ok(entries) = std::fs::read_dir(&self.cursor_projects_dir) else {
@@ -1314,6 +1317,7 @@ fn parent_dispatch_model_for_subagent(
     None
 }
 
+#[hotpath::measure(label = "sessions.hosts.cursor.dispatch_model_scan")]
 fn dispatch_model_for_agent(path: &Path, agent_id: &str) -> Option<String> {
     let file = File::open(path).ok()?;
     let mut frames = RawJsonlFrameReader::new(BufReader::new(file), MAX_JSONL_RECORD_BYTES);

crates/tracedecay-sessions/src/runtime/hosts/cursor_composer/ingest.rs

@@ -126,6 +126,7 @@ impl ComposerIngestContext<'_, '_> {
     }
 }
 
+#[hotpath::measure(label = "sessions.hosts.cursor.composer_projection_drain", future = true)]
 async fn drain_composer_projection_queue(
     context: &ComposerIngestContext<'_, '_>,
 ) -> TranscriptIngestResult<crate::runtime::cursor::projection::CursorProjectionDrainStats> {
@@ -163,6 +164,7 @@ struct ComposerCoverageContext<'facade> {
     cancellation: &'facade ObservationCancellation,
 }
 
+#[hotpath::measure(label = "sessions.hosts.cursor.composer_coverage_advance", future = true)]
 async fn advance_composer_coverage(
     context: ComposerCoverageContext<'_>,
     source: ObservationSourceIdentityV1,
@@ -263,6 +265,7 @@ impl CursorComposerSource {
             .await
     }
 
+    #[hotpath::measure(label = "sessions.hosts.cursor.composer_sweep", future = true)]
     async fn ingest_with_context(
         &self,
         context: &ComposerIngestContext<'_, '_>,
@@ -337,6 +340,7 @@ impl CursorComposerSource {
         Ok(outcome.finished(byte_budget.consumed(), byte_budget.deferred()))
     }
 
+    #[hotpath::measure(label = "sessions.hosts.cursor.composer_state_vscdb", future = true)]
     async fn ingest_state_vscdb(
         &self,
         context: &ComposerIngestContext<'_, '_>,
@@ -783,6 +787,7 @@ impl CursorComposerSource {
         }
     }
 
+    #[hotpath::measure(label = "sessions.hosts.cursor.composer_chat_stores", future = true)]
     async fn ingest_chat_store_dbs(
         &self,
         context: &ComposerIngestContext<'_, '_>,
@@ -810,6 +815,7 @@ impl CursorComposerSource {
         }
     }
 
+    #[hotpath::measure(label = "sessions.hosts.cursor.composer_store_db", future = true)]
     async fn ingest_one_store_db(
         &self,
         context: &ComposerIngestContext<'_, '_>,

crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs

@@ -487,6 +487,7 @@ pub(super) struct HermesProfileSource {
     pub profile: Option<String>,
 }
 
+#[hotpath::measure(label = "sessions.hosts.hermes.discover_profiles")]
 fn all_profile_sources(hermes_homes: &[PathBuf]) -> Vec<HermesProfileSource> {
     let mut out = Vec::new();
     let mut seen = BTreeSet::new();
@@ -518,6 +519,7 @@ fn all_profile_sources(hermes_homes: &[PathBuf]) -> Vec<HermesProfileSource> {
     out
 }
 
+#[hotpath::measure(label = "sessions.hosts.hermes.discover_candidates")]
 fn candidate_state_dbs(hermes_homes: &[PathBuf], project_root: &Path) -> Vec<HermesProfileSource> {
     let mut out = Vec::new();
     let mut seen = BTreeSet::new();

crates/tracedecay-sessions/src/runtime/ingest/scheduler.rs

@@ -133,6 +133,7 @@ pub(super) fn finish_user_provider_coverage(
     }
 }
 
+#[hotpath::measure(label = "sessions.ingest.frontier_read", future = true)]
 pub(super) async fn read_ingest_frontier<S: TranscriptIngestStore>(
     store: &S,
     key: &str,
@@ -143,6 +144,7 @@ pub(super) async fn read_ingest_frontier<S: TranscriptIngestStore>(
     }
 }
 
+#[hotpath::measure(label = "sessions.ingest.frontier_write", future = true)]
 pub(super) async fn write_ingest_frontier<S: TranscriptIngestStore>(
     store: &S,
     key: &str,
@@ -166,6 +168,7 @@ pub(super) async fn write_ingest_frontier<S: TranscriptIngestStore>(
         .is_ok()
 }
 
+#[hotpath::measure(label = "sessions.ingest.codex_frontier_read", future = true)]
 pub(super) async fn read_codex_discovery_frontier<S: TranscriptIngestStore>(
     store: &S,
 ) -> TranscriptIngestResult<CodexDiscoveryFrontier> {
@@ -178,6 +181,7 @@ pub(super) async fn read_codex_discovery_frontier<S: TranscriptIngestStore>(
     CodexDiscoveryFrontier::from_parse_offsets(stored, epoch)
 }
 
+#[hotpath::measure(label = "sessions.ingest.codex_frontier_write", future = true)]
 pub(super) async fn write_codex_discovery_frontier<S: TranscriptIngestStore>(
     store: &S,
     expected: CodexDiscoveryFrontier,

crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs

@@ -839,6 +839,7 @@ struct SharedJsonlBuildOptions {
     cancellation: Option<Arc<std::sync::atomic::AtomicBool>>,
 }
 
+#[hotpath::measure(label = "sessions.observation.jsonl_page_build")]
 fn build_shared_jsonl_page(
     path: PathBuf,
     previous: StoredCursor,
@@ -1043,6 +1044,7 @@ async fn prepare_shared_jsonl_window(
     .await
 }
 
+#[hotpath::measure(label = "sessions.observation.jsonl_window_prepare", future = true)]
 async fn prepare_shared_jsonl_window_with_background_cpu(
     page: &SharedJsonlPage,
     start: usize,
@@ -1223,6 +1225,7 @@ async fn shared_jsonl_page(
     .await
 }
 
+#[hotpath::measure(label = "sessions.observation.jsonl_page_lookup", future = true)]
 async fn shared_jsonl_page_with_cancellation(
     path: &Path,
     previous: StoredCursor,
@@ -1568,6 +1571,7 @@ impl ActiveAdmission<'_> {
         .with_resume_checkpoint(self.file_identity, resume_fingerprint))
     }
 
+    #[hotpath::measure(label = "sessions.observation.jsonl_coverage_advance", future = true)]
     async fn advance_coverage(
         &self,
         expected_cursor: &mut Option<ObservationSourceCursorV1>,
@@ -1760,6 +1764,7 @@ impl ActiveAdmission<'_> {
         }
     }
 
+    #[hotpath::measure(label = "sessions.observation.jsonl_capture", future = true)]
     async fn capture(
         &self,
         expected_cursor: &mut Option<ObservationSourceCursorV1>,
@@ -1778,6 +1783,7 @@ impl ActiveAdmission<'_> {
             .await
     }
 
+    #[hotpath::measure(label = "sessions.observation.jsonl_capture_window", future = true)]
     async fn capture_window(
         &self,
         expected_cursor: &mut Option<ObservationSourceCursorV1>,

crates/tracedecay-sessions/src/runtime/source.rs

@@ -82,6 +82,7 @@ impl HostProviderCoverage {
 pub(super) const CODEX_HISTORY_FRONTIER_KEY: &str = "tracedecay-internal:codex-history-frontier:v2";
 pub(super) const CODEX_HISTORY_EPOCH_KEY: &str = "tracedecay-internal:codex-history-epoch:v2";
 
+#[hotpath::measure(label = "sessions.source.coverage_read", future = true)]
 pub(super) async fn read_host_provider_coverage(
     admission: &dyn HostAdmission,
     scope: &ObservationScopeV1,
@@ -142,6 +143,7 @@ pub(super) async fn persist_codex_history_frontier(
         })
 }
 
+#[hotpath::measure(label = "sessions.source.coverage_persist", future = true)]
 pub(super) async fn persist_host_provider_coverage(
     admission: &(impl HostAdmission + ?Sized),
     scope: &ObservationScopeV1,
@@ -357,6 +359,7 @@ pub struct LoadedTranscriptCursor {
     durable_offset: ParseOffset,
 }
 
+#[hotpath::measure(label = "sessions.source.cursor_load", future = true)]
 pub async fn load_transcript_cursor<S: TranscriptIngestStore>(
     store: &S,
     key: TranscriptCursorKey,
@@ -819,6 +822,7 @@ use jsonl::{
     stream_new_jsonl_raw_strict, stream_new_jsonl_strict, stream_new_jsonl_with_policy,
 };
 
+#[hotpath::measure(label = "sessions.source.preflight_jsonl")]
 pub fn preflight_strict_jsonl(
     provider: &'static str,
     path: &Path,
@@ -859,6 +863,7 @@ pub struct ChangedFile {
 /// with deterministic ids. Idempotent upserts make re-adding unchanged messages
 /// a no-op. Returns `None` when the file cannot be read or is unchanged since
 /// the last run.
+#[hotpath::measure(label = "sessions.source.read_changed")]
 pub fn read_changed_file(path: &Path, prev: StoredCursor, max_bytes: u64) -> Option<ChangedFile> {
     let meta = match std::fs::metadata(path) {
         Ok(meta) => meta,
@@ -892,6 +897,7 @@ pub fn read_changed_file(path: &Path, prev: StoredCursor, max_bytes: u64) -> Opt
 /// own content hash moves or a companion sidecar file's hash moves. The stored
 /// cursor's `position` is a combined hash of both files so a sidecar-only
 /// update (e.g. Cline `ui_messages.json` usage counters) triggers a re-ingest.
+#[hotpath::measure(label = "sessions.source.read_changed_companion")]
 pub fn read_changed_with_companion(
     primary: &Path,
     companion: &Path,

crates/tracedecay-sessions/src/runtime/store_access/transcript.rs

@@ -14,6 +14,7 @@ enum TranscriptWritePolicy {
     ProjectionOnly,
 }
 
+#[hotpath::measure(label = "sessions.transcript.offset_read", future = true)]
 pub async fn get_parse_offset(
     conn: &impl QueryExecutor,
     path: &str,
@@ -123,6 +124,7 @@ pub async fn require_expected_offset(
     }
 }
 
+#[hotpath::measure(label = "sessions.transcript.offset_write", future = true)]
 pub async fn set_parse_offset(
     conn: &impl Executor,
     path: &str,
@@ -156,6 +158,7 @@ impl<D: SessionRegisteredDb + Sync> SessionStoreAccess<'_, D> {
             .map_err(|error| TranscriptPersistenceError::storage("begin transcript batch", error))
     }
 
+    #[hotpath::measure(label = "sessions.transcript.session_upsert", future = true)]
     pub async fn upsert_session(&self, session: &SessionRecord) -> bool {
         let Ok(transaction) = self.begin_transcript_transaction().await else {
             return false;
@@ -212,6 +215,7 @@ impl<D: SessionRegisteredDb + Sync> SessionStoreAccess<'_, D> {
             .flatten()
     }
 
+    #[hotpath::measure(label = "sessions.transcript.session_read", future = true)]
     pub async fn get_session_result(
         &self,
         provider: &str,
@@ -294,6 +298,7 @@ impl<D: SessionRegisteredDb + Sync> SessionStoreAccess<'_, D> {
         })
     }
 
+    #[hotpath::measure(label = "sessions.transcript.message_upsert", future = true)]
     async fn upsert_session_message_in_existing_tx(
         &self,
         conn: &impl Executor,
@@ -418,6 +423,7 @@ impl<D: SessionRegisteredDb + Sync> SessionStoreAccess<'_, D> {
         .await
     }
 
+    #[hotpath::measure(label = "sessions.transcript.offset_commit", future = true)]
     pub async fn persist_transcript_offset_result(
         &self,
         parse_offset_path: &str,
@@ -452,6 +458,7 @@ impl<D: SessionRegisteredDb + Sync> SessionStoreAccess<'_, D> {
         .map_err(|error| error.to_string())
     }
 
+    #[hotpath::measure(label = "sessions.transcript.batch_persist", future = true)]
     async fn upsert_transcript_batches_inner(
         &self,
         batches: &[TranscriptBatch],
@@ -565,6 +572,7 @@ impl<D: SessionRegisteredDb + Sync> SessionStoreAccess<'_, D> {
             .map_err(|error| format!("commit transcript parse offset: {error}"))
     }
 
+    #[hotpath::measure(label = "sessions.transcript.offset_advance", future = true)]
     pub async fn advance_parse_offset_result(
         &self,
         path: &str,
@@ -583,6 +591,7 @@ impl<D: SessionRegisteredDb + Sync> SessionStoreAccess<'_, D> {
 
     /// Exact compare-and-set for versioned parse-offset authorities whose
     /// numeric fields are not monotonic transcript positions.
+    #[hotpath::measure(label = "sessions.transcript.offset_replace", future = true)]
     pub async fn replace_parse_offset_result(
         &self,
         path: &str,
@@ -600,6 +609,7 @@ impl<D: SessionRegisteredDb + Sync> SessionStoreAccess<'_, D> {
     /// Atomically compare-and-replace two parse-offset keys. Both expected
     /// values are checked before either write and one transaction owns the
     /// pair through commit.
+    #[hotpath::measure(label = "sessions.transcript.offset_replace_pair", future = true)]
     pub async fn replace_parse_offset_pair_result(
         &self,
         first: (&str, ParseOffset, ParseOffset),