feat(sessions): git-anchored session/LCM correlation

Query sessions and messages by branch, worktree, or commit: span-based
attribution tables (session_git_spans, commit_sessions) in the per-
project session store; live span recording from hook-route metadata
(incl. linked-worktree resolution via git-common-dir identity); commit
attribution on ingest; historical backfill CLI (sessions git-backfill,
reflog-segmented, idempotent); tracedecay_sessions_for MCP tool plus
branch/worktree/commit filters on message_search and lcm_grep.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: ScriptedAlchemy

plugin/README-cursor.md

@@ -132,6 +132,7 @@ per-call review, add the snippet below to `~/.cursor/permissions.json`
     "tracedecay:tracedecay_retrieve",
     "tracedecay:tracedecay_runtime",
     "tracedecay:tracedecay_search",
+    "tracedecay:tracedecay_sessions_for",
     "tracedecay:tracedecay_signature",
     "tracedecay:tracedecay_signature_search",
     "tracedecay:tracedecay_similar",

plugin/skills/recalling-session-context/SKILL.md

@@ -16,10 +16,11 @@ This skill owns the **FTS → LCM** lane of `tracedecay_message_search`: `messag
 3. **Lossless replay → `tracedecay_lcm_load_session`** (`session_id`, `after_store_id` + `limit` for stable pagination, `roles`, `content_offset`/`content_limit`): ordered raw messages of one session; page with `next_cursor` instead of asking for everything at once.
 4. **Summary-DAG drill-down:** `tracedecay_lcm_describe` (`session_id`) for the session's raw/summary shape; `tracedecay_lcm_expand` (`target.kind`: `raw_message`|`summary_node`|`external_payload`) to open one node, paging sources via `source_offset`/`source_limit`; `tracedecay_lcm_expand_query` (`query`) to assemble bounded retrieval context for a prompt in one call.
 5. **Store inspection → `tracedecay_lcm_status`** (counts, token estimates, DAG depth/compression ratio) when you need to know what the store contains before searching it.
+6. **Git-scoped session lookup → `tracedecay_sessions_for`** (`git_ref`: `branch`|`worktree`|`commit`, `value`, optional `since`/`until`, `limit`): which sessions were active on a branch or in a worktree, or which conversations produced a commit; feed the returned session ids back into rungs 2–4.
 
 ## Guardrails
 
-- Steps 1–5 are read-only. `tracedecay_lcm_compress`, `tracedecay_lcm_preflight`, and `tracedecay_lcm_session_boundary` are **lifecycle-integration tools for host agents** — never invoke them casually during recall.
+- Steps 1–6 are read-only. `tracedecay_lcm_compress`, `tracedecay_lcm_preflight`, and `tracedecay_lcm_session_boundary` are **lifecycle-integration tools for host agents** — never invoke them casually during recall.
 - For multi-step recall, dispatch scoped read-only subagents by session id, time window, provider, role, or query variant. Subagents must not call lifecycle or repair tools; the parent agent validates cited messages/summaries and produces the final timeline.
 - If the LCM store itself looks wrong (missing sessions, broken FTS, stale counts) → `tracedecay_lcm_doctor` (`mode: "diagnose"` first; `repair`/`clean` mutate and need explicit user intent).
 - All LCM tools default to `storage_scope: "project_local"`; only pass `hermes_profile` (with an absolute `hermes_home`) when the user asks about a Hermes profile store.

src/analytics_bridge.rs

@@ -289,6 +289,7 @@ pub async fn run_analytics_diagnostics(
             project_id: project_filter.clone(),
             session_id: None,
             event_kind: None,
+            since: None,
             limit: 10_000,
         })
         .await

src/automation/runner.rs

@@ -713,6 +713,7 @@ async fn build_session_reflector_evidence(
             role: role.clone(),
             start_time: options.start_time,
             end_time: options.end_time,
+            git_filter: crate::sessions::git_correlation::GitScopeFilter::default(),
         })
         .await
         .map_err(|e| TraceDecayError::Config {
@@ -828,6 +829,7 @@ async fn build_skill_writer_evidence(
             role: None,
             start_time: None,
             end_time: None,
+            git_filter: crate::sessions::git_correlation::GitScopeFilter::default(),
         })
         .await
         .map_err(|e| TraceDecayError::Config {

src/automation/skill_usage/analytics.rs

@@ -68,6 +68,7 @@ pub async fn ingest_project_analytics_events(
             project_id: Some(GlobalDb::canonical_project_key(project_root)),
             session_id: None,
             event_kind: None,
+            since: None,
             limit,
         })
         .await

src/cli.rs

@@ -636,6 +636,35 @@ pub enum SessionsAction {
         /// Registered project root path or alias whose session store should be searched
         #[arg(long, conflicts_with = "project_id")]
         project_path: Option<String>,
+        /// Only sessions correlated with this git branch
+        #[arg(long)]
+        branch: Option<String>,
+        /// Only sessions correlated with this worktree path
+        #[arg(long)]
+        worktree: Option<String>,
+        /// Only sessions that produced this commit (full or >=6-char prefix)
+        #[arg(long)]
+        commit: Option<String>,
+    },
+    /// Backfill the session↔git correlation index from historical session,
+    /// analytics, and reflog signals
+    GitBackfill {
+        /// Registered project id whose session store should be backfilled
+        #[arg(long)]
+        project_id: Option<String>,
+        /// Registered project root path or alias whose session store should be backfilled
+        #[arg(long, conflicts_with = "project_id")]
+        project_path: Option<String>,
+        /// Lower bound on session activity and commit times (ISO-8601 or unix
+        /// seconds); defaults to 90 days ago
+        #[arg(long)]
+        since: Option<String>,
+        /// Maximum number of sessions to scan
+        #[arg(long, default_value_t = 500)]
+        limit_sessions: usize,
+        /// Derive and report counts without writing to the session store
+        #[arg(long)]
+        dry_run: bool,
     },
 }
 

src/cli/parse_tests.rs

@@ -1552,6 +1552,9 @@ fn parses_sessions_ingest_and_search_commands() {
                     project_path,
                     since,
                     until,
+                    branch,
+                    worktree,
+                    commit,
                 },
         }) => {
             assert_eq!(query, "needle");
@@ -1561,6 +1564,9 @@ fn parses_sessions_ingest_and_search_commands() {
             assert!(project_path.is_none());
             assert!(since.is_none());
             assert!(until.is_none());
+            assert!(branch.is_none());
+            assert!(worktree.is_none());
+            assert!(commit.is_none());
         }
         _ => panic!("expected sessions search command"),
     }

src/dashboard/analytics_api.rs

@@ -123,6 +123,7 @@ async fn durable_analytics_rows(
                 project_id: Some(project_id.to_string()),
                 session_id: None,
                 event_kind: None,
+                since: None,
                 limit: ANALYTICS_EVENT_LIMIT,
             })
             .await

src/global_db.rs

@@ -95,6 +95,8 @@ pub struct AnalyticsEventQuery {
     pub project_id: Option<String>,
     pub session_id: Option<String>,
     pub event_kind: Option<String>,
+    /// Inclusive lower bound on `timestamp` (unix seconds). `None` = unbounded.
+    pub since: Option<i64>,
     pub limit: usize,
 }
 
@@ -1018,6 +1020,9 @@ impl GlobalDb {
         crate::sessions::lcm::schema::ensure_lcm_schema(&db.conn)
             .await
             .ok()?;
+        crate::sessions::git_correlation::ensure_git_correlation_schema(&db.conn)
+            .await
+            .ok()?;
         // One-off self-heal: re-derive timestamps and token-usage counters
         // for legacy messages ingested before extraction existed.
         // Marker-guarded (runs once per store) and fail-open, like the LCM
@@ -1880,11 +1885,9 @@ impl GlobalDb {
     /// filter by canonical project path. Returns zeros on any DB error.
     pub async fn sum_savings(&self, project: Option<&str>, since: i64) -> SavingsTotal {
         let project = project.map(|p| Self::canonical_project_key(Path::new(p)));
-        let sql_with_project =
-            "SELECT COALESCE(SUM(CASE WHEN before_tokens > after_tokens THEN before_tokens - after_tokens ELSE 0 END), 0), COUNT(*) \
+        let sql_with_project = "SELECT COALESCE(SUM(CASE WHEN before_tokens > after_tokens THEN before_tokens - after_tokens ELSE 0 END), 0), COUNT(*) \
              FROM savings_ledger WHERE project_path = ?1 AND ts >= ?2";
-        let sql_all =
-            "SELECT COALESCE(SUM(CASE WHEN before_tokens > after_tokens THEN before_tokens - after_tokens ELSE 0 END), 0), COUNT(*) \
+        let sql_all = "SELECT COALESCE(SUM(CASE WHEN before_tokens > after_tokens THEN before_tokens - after_tokens ELSE 0 END), 0), COUNT(*) \
              FROM savings_ledger WHERE ts >= ?1";
 
         let rows = match project.as_deref() {
@@ -1912,14 +1915,12 @@ impl GlobalDb {
     /// Group ledger entries by UTC calendar day. Newest-first.
     pub async fn savings_history(&self, project: Option<&str>, since: i64) -> Vec<SavingsDay> {
         let project = project.map(|p| Self::canonical_project_key(Path::new(p)));
-        let sql_with_project =
-            "SELECT (ts/86400)*86400 AS day, \
+        let sql_with_project = "SELECT (ts/86400)*86400 AS day, \
                     COALESCE(SUM(CASE WHEN before_tokens > after_tokens THEN before_tokens - after_tokens ELSE 0 END), 0), \
                     COUNT(*) \
              FROM savings_ledger WHERE project_path = ?1 AND ts >= ?2 \
              GROUP BY day ORDER BY day DESC";
-        let sql_all =
-            "SELECT (ts/86400)*86400 AS day, \
+        let sql_all = "SELECT (ts/86400)*86400 AS day, \
                     COALESCE(SUM(CASE WHEN before_tokens > after_tokens THEN before_tokens - after_tokens ELSE 0 END), 0), \
                     COUNT(*) \
              FROM savings_ledger WHERE ts >= ?1 \
@@ -2241,6 +2242,10 @@ impl GlobalDb {
         ] {
             push_optional_analytics_filter(&mut clauses, &mut values, column, value);
         }
+        if let Some(since) = query.since {
+            values.push(Value::Integer(since));
+            clauses.push(format!("timestamp >= ?{}", values.len()));
+        }
         if !clauses.is_empty() {
             sql.push_str(" WHERE ");
             sql.push_str(&clauses.join(" AND "));
@@ -3165,6 +3170,83 @@ impl GlobalDb {
             .await
     }
 
+    // ── Session ↔ git correlation ────────────────────────────────────
+
+    /// Folds one live/backfilled git observation into the span table.
+    /// See [`crate::sessions::git_correlation::record_span_observation`].
+    pub async fn git_record_span_observation(
+        &self,
+        observation: &crate::sessions::git_correlation::SpanObservation,
+        merge_gap_secs: i64,
+    ) -> Result<i64, crate::sessions::git_correlation::GitCorrelationError> {
+        crate::sessions::git_correlation::record_span_observation(
+            &self.conn,
+            observation,
+            merge_gap_secs,
+        )
+        .await
+    }
+
+    /// Attributes one commit to one session (idempotent).
+    /// See [`crate::sessions::git_correlation::upsert_commit_session`].
+    pub async fn git_upsert_commit_session(
+        &self,
+        record: &crate::sessions::git_correlation::CommitSessionRecord,
+    ) -> Result<bool, crate::sessions::git_correlation::GitCorrelationError> {
+        crate::sessions::git_correlation::upsert_commit_session(&self.conn, record).await
+    }
+
+    /// Runs the commit-attribution sweep, delegating branch-scoped git log
+    /// reads to `scan`. See
+    /// [`crate::sessions::git_correlation::run_commit_attribution_sweep`].
+    pub async fn git_run_commit_attribution_sweep<F>(
+        &self,
+        gap_secs: i64,
+        scan: F,
+    ) -> Result<usize, crate::sessions::git_correlation::GitCorrelationError>
+    where
+        F: FnMut(
+            &crate::sessions::git_correlation::SpanScanTarget,
+        ) -> Vec<crate::sessions::git_correlation::ScannedCommit>,
+    {
+        crate::sessions::git_correlation::run_commit_attribution_sweep(&self.conn, gap_secs, scan)
+            .await
+    }
+
+    /// Returns sessions correlated with a branch, worktree, or commit.
+    /// See [`crate::sessions::git_correlation::sessions_for`].
+    pub async fn git_sessions_for(
+        &self,
+        query: &crate::sessions::git_correlation::SessionsForQuery,
+    ) -> Result<
+        Vec<crate::sessions::git_correlation::SessionGitCorrelationHit>,
+        crate::sessions::git_correlation::GitCorrelationError,
+    > {
+        crate::sessions::git_correlation::sessions_for(&self.conn, query).await
+    }
+
+    /// Resolves the `(provider, session_id)` pairs matching a git-scope filter.
+    /// See [`crate::sessions::git_correlation::session_ids_for_scope`].
+    pub async fn git_session_ids_for_scope(
+        &self,
+        filter: &crate::sessions::git_correlation::GitScopeFilter,
+    ) -> Result<Option<Vec<(String, String)>>, crate::sessions::git_correlation::GitCorrelationError>
+    {
+        crate::sessions::git_correlation::session_ids_for_scope(&self.conn, filter).await
+    }
+
+    /// Lists per-session activity windows for the historical git-correlation
+    /// backfill: each row carries the session's declared `started_at`/`ended_at`
+    /// plus the min/max `session_messages.timestamp`, so the caller can derive
+    /// coarse activity windows without a second query per session. Ordered
+    /// newest-first (by the latest known activity), capped at `limit`.
+    pub async fn session_activity_rows(
+        &self,
+        limit: usize,
+    ) -> Result<Vec<crate::sessions::git_correlation::SessionActivityRow>, String> {
+        crate::sessions::git_correlation::session_activity_rows(&self.conn, limit).await
+    }
+
     /// Searches message text for a provider, optionally constrained to one project.
     pub async fn search_session_messages(
         &self,
@@ -3198,6 +3280,32 @@ impl GlobalDb {
             query,
             limit,
             filters,
+            None,
+        )
+        .await
+    }
+
+    /// Like [`Self::search_session_messages_filtered`], additionally scoping
+    /// hits to sessions correlated with a git branch/worktree/commit via
+    /// EXISTS pushdown against the git-correlation tables. Pass `provider =
+    /// None` to search all providers. A git-scoped call against a store
+    /// predating the correlation schema returns no hits.
+    pub async fn search_session_messages_git_scoped(
+        &self,
+        provider: Option<&str>,
+        project_key: Option<&str>,
+        query: &str,
+        limit: usize,
+        filters: SessionSearchFilters<'_>,
+        git_filter: &crate::sessions::git_correlation::GitScopeFilter,
+    ) -> Vec<SessionMessageSearchResult> {
+        self.search_session_messages_filtered_inner(
+            provider,
+            project_key,
+            query,
+            limit,
+            filters,
+            Some(git_filter),
         )
         .await
     }
@@ -3210,7 +3318,7 @@ impl GlobalDb {
         limit: usize,
         filters: SessionSearchFilters<'_>,
     ) -> Vec<SessionMessageSearchResult> {
-        self.search_session_messages_filtered_inner(None, project_key, query, limit, filters)
+        self.search_session_messages_filtered_inner(None, project_key, query, limit, filters, None)
             .await
     }
 
@@ -3221,7 +3329,20 @@ impl GlobalDb {
         query: &str,
         limit: usize,
         filters: SessionSearchFilters<'_>,
+        git_filter: Option<&crate::sessions::git_correlation::GitScopeFilter>,
     ) -> Vec<SessionMessageSearchResult> {
+        // A git-scoped search against a store written before the correlation
+        // schema existed can never match; report empty rather than issuing a
+        // `no such table` EXISTS subquery.
+        if let Some(filter) = git_filter {
+            if !filter.is_empty()
+                && !crate::sessions::git_correlation::tables_present(&self.conn)
+                    .await
+                    .unwrap_or(false)
+            {
+                return Vec::new();
+            }
+        }
         let fts_query = session_fts_query(query);
         if fts_query.is_empty() || limit == 0 {
             return Vec::new();
@@ -3285,6 +3406,19 @@ impl GlobalDb {
         ) {
             sql.push_str(" AND s.is_subagent = 1");
         }
+        // Reuse the shared scoping SQL (also used by the lcm/grep path) so the
+        // branch/worktree/commit EXISTS semantics stay in one place. Its
+        // anonymous `?` placeholders bind to the next sequential positions,
+        // which — since the predicate and its values are appended together in
+        // order — line up with the numbered placeholders that follow.
+        if let Some(filter) = git_filter {
+            if let Some((predicate, predicate_values)) =
+                crate::sessions::git_correlation::git_scope_exists_predicate(filter, "m.session_id")
+            {
+                let _ = write!(sql, " AND {predicate}");
+                query_params.extend(predicate_values);
+            }
+        }
         for term in &literal_terms {
             query_params.push(Value::Text(term.clone()));
             let _ = write!(