diff --git a/crates/tracedecay-sessions/src/runtime/git_correlation/backfill/bounded.rs b/crates/tracedecay-sessions/src/runtime/git_correlation/backfill/bounded.rs index 98f22a30b3..23f660592f 100644 --- a/crates/tracedecay-sessions/src/runtime/git_correlation/backfill/bounded.rs +++ b/crates/tracedecay-sessions/src/runtime/git_correlation/backfill/bounded.rs @@ -122,6 +122,7 @@ pub struct BoundedBackfillOutcome { pub interruption: Option, } +#[hotpath::measure(label = "sessions.git_correlation.backfill.bounded_page", future = true)] pub async fn run_bounded_history_index_page( 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( session_store: &S, row: &SessionActivityRow, @@ -519,6 +521,7 @@ async fn stream_git_evidence( 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( session_store: &S, key: GitHistoryProgressKey, @@ -626,6 +629,7 @@ async fn resume_git_evidence( } } +#[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( session_store: &S, candidate: GitHistoryIndexFrontier, diff --git a/crates/tracedecay-sessions/src/runtime/git_correlation/backfill/history_progress.rs b/crates/tracedecay-sessions/src/runtime/git_correlation/backfill/history_progress.rs index 22d5a67319..9f19883a68 100644 --- a/crates/tracedecay-sessions/src/runtime/git_correlation/backfill/history_progress.rs +++ b/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, 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, diff --git a/crates/tracedecay-sessions/src/runtime/hosts/cursor.rs b/crates/tracedecay-sessions/src/runtime/hosts/cursor.rs index 64bf2704d7..d3425757a2 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/cursor.rs +++ b/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 { 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 { let file = File::open(path).ok()?; let mut frames = RawJsonlFrameReader::new(BufReader::new(file), MAX_JSONL_RECORD_BYTES); diff --git a/crates/tracedecay-sessions/src/runtime/hosts/cursor_composer/ingest.rs b/crates/tracedecay-sessions/src/runtime/hosts/cursor_composer/ingest.rs index 4300764e2e..a510b48a29 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/cursor_composer/ingest.rs +++ b/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 { @@ -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<'_, '_>, diff --git a/crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs b/crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs index 8f6d044305..63e613a2e7 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs @@ -487,6 +487,7 @@ pub(super) struct HermesProfileSource { pub profile: Option, } +#[hotpath::measure(label = "sessions.hosts.hermes.discover_profiles")] fn all_profile_sources(hermes_homes: &[PathBuf]) -> Vec { let mut out = Vec::new(); let mut seen = BTreeSet::new(); @@ -518,6 +519,7 @@ fn all_profile_sources(hermes_homes: &[PathBuf]) -> Vec { out } +#[hotpath::measure(label = "sessions.hosts.hermes.discover_candidates")] fn candidate_state_dbs(hermes_homes: &[PathBuf], project_root: &Path) -> Vec { let mut out = Vec::new(); let mut seen = BTreeSet::new(); diff --git a/crates/tracedecay-sessions/src/runtime/ingest/scheduler.rs b/crates/tracedecay-sessions/src/runtime/ingest/scheduler.rs index 1c0da3652f..b2aba23454 100644 --- a/crates/tracedecay-sessions/src/runtime/ingest/scheduler.rs +++ b/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( store: &S, key: &str, @@ -143,6 +144,7 @@ pub(super) async fn read_ingest_frontier( } } +#[hotpath::measure(label = "sessions.ingest.frontier_write", future = true)] pub(super) async fn write_ingest_frontier( store: &S, key: &str, @@ -166,6 +168,7 @@ pub(super) async fn write_ingest_frontier( .is_ok() } +#[hotpath::measure(label = "sessions.ingest.codex_frontier_read", future = true)] pub(super) async fn read_codex_discovery_frontier( store: &S, ) -> TranscriptIngestResult { @@ -178,6 +181,7 @@ pub(super) async fn read_codex_discovery_frontier( CodexDiscoveryFrontier::from_parse_offsets(stored, epoch) } +#[hotpath::measure(label = "sessions.ingest.codex_frontier_write", future = true)] pub(super) async fn write_codex_discovery_frontier( store: &S, expected: CodexDiscoveryFrontier, diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs index ea92e7acce..c837055228 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs @@ -839,6 +839,7 @@ struct SharedJsonlBuildOptions { cancellation: Option>, } +#[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, @@ -1760,6 +1764,7 @@ impl ActiveAdmission<'_> { } } + #[hotpath::measure(label = "sessions.observation.jsonl_capture", future = true)] async fn capture( &self, expected_cursor: &mut Option, @@ -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, diff --git a/crates/tracedecay-sessions/src/runtime/source.rs b/crates/tracedecay-sessions/src/runtime/source.rs index cafa374fc2..c93c532553 100644 --- a/crates/tracedecay-sessions/src/runtime/source.rs +++ b/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( 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 { 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, diff --git a/crates/tracedecay-sessions/src/runtime/store_access/transcript.rs b/crates/tracedecay-sessions/src/runtime/store_access/transcript.rs index 1567d43ca7..3fa68a7718 100644 --- a/crates/tracedecay-sessions/src/runtime/store_access/transcript.rs +++ b/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 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 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 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 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 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 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 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 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), diff --git a/crates/tracedecay-sessions/src/runtime/workflow/workflow_index.rs b/crates/tracedecay-sessions/src/runtime/workflow/workflow_index.rs index 86ac9fac14..b614937f54 100644 --- a/crates/tracedecay-sessions/src/runtime/workflow/workflow_index.rs +++ b/crates/tracedecay-sessions/src/runtime/workflow/workflow_index.rs @@ -101,6 +101,7 @@ impl From for WorkflowInde /// through the shared `session_schema_migrations` table exactly like /// [`crate::runtime::git_correlation::ensure_git_correlation_receipt_schema_in_transaction`], /// so both stores register under their own migration name in one table. +#[hotpath::measure(label = "sessions.workflow_index.ensure_schema", future = true)] pub async fn ensure_workflow_index_schema(conn: &impl Executor) -> Result<(), WorkflowIndexError> { if schema_version(conn) .await @@ -205,6 +206,7 @@ pub const INGEST_WATERMARK_KEY: &str = "ingest_watermark_mtime"; /// Reads the ingest watermark (max processed run-file mtime, unix seconds), or /// `0` when unset / the schema predates this table. Never errors: a store /// without the meta table simply reports no watermark, forcing a full sweep. +#[hotpath::measure(label = "sessions.workflow_index.watermark_read", future = true)] pub async fn read_ingest_watermark(conn: &impl QueryExecutor, key: &str) -> i64 { let Ok(mut rows) = conn .query( @@ -225,6 +227,7 @@ pub async fn read_ingest_watermark(conn: &impl QueryExecutor, key: &str) -> i64 /// whose transcripts grew (e.g. a `running` run that later `completed`) /// overwrites the mutable columns and refreshes `updated_at`. `created_at` is /// preserved. +#[hotpath::measure(label = "sessions.workflow_index.run_upsert", future = true)] pub async fn upsert_run(conn: &impl Executor, run: &WorkflowRun) -> Result<(), WorkflowIndexError> { if run.run_id.trim().is_empty() { return Err(WorkflowIndexError::InvalidArgument( @@ -266,6 +269,7 @@ pub async fn upsert_run(conn: &impl Executor, run: &WorkflowRun) -> Result<(), W /// Inserts or updates one agent row (idempotent on `(run_id, agent_label, /// agent_id)`). +#[hotpath::measure(label = "sessions.workflow_index.agent_upsert", future = true)] pub async fn upsert_agent( conn: &impl Executor, agent: &WorkflowAgent, @@ -414,6 +418,7 @@ where self.has_tables(&["workflow_runs", "workflow_agents"]).await } + #[hotpath::measure(label = "sessions.workflow_index.runs_for_session", future = true)] pub async fn runs_for_session( &self, parent_session_id: &str, @@ -443,6 +448,7 @@ where Ok(runs) } + #[hotpath::measure(label = "sessions.workflow_index.run_read", future = true)] pub async fn run_for_id( &self, run_id: &str, @@ -458,6 +464,7 @@ where rows.next().await?.map(|row| row_to_run(&row)).transpose() } + #[hotpath::measure(label = "sessions.workflow_index.agents_for_run", future = true)] pub async fn agents_for_run( &self, run_id: &str, @@ -532,6 +539,7 @@ where rows.next().await?.map(|row| row_to_agent(&row)).transpose() } + #[hotpath::measure(label = "sessions.workflow_index.runs_for_git_scope", future = true)] pub async fn runs_for_git_scope( &self, session_ids: Option<&[(String, String)]>,