fix: recover forwarded Claude skills and scope session search

README.md

@@ -17,7 +17,7 @@ Cortex began as a syslog receiver. It now covers network logs, Docker, managed f
 | Area | What Cortex provides |
 | --- | --- |
 | Ingest | UDP/TCP syslog, OTLP/HTTP logs, metrics, and traces, Docker logs and events, managed file tails, host heartbeats, AI transcripts, shell history, agent command records, and fleet inventory |
-| Storage | SQLite in WAL mode, FTS5 full-text search, bounded metadata, retention, storage budgets, maintenance jobs, checkpoints, and 60 sequential schema migrations |
+| Storage | SQLite in WAL mode, FTS5 full-text search, bounded metadata, retention, storage budgets, maintenance jobs, checkpoints, and 61 sequential schema migrations |
 | Investigation | Search, filtering, context, timelines, patterns, anomaly comparison, cross-source correlation, recurring error signatures, deterministic incident bundles, and graph explanations |
 | Fleet intelligence | SSH and API inventory collectors, host state, service topology, container and route relationships, redacted evidence, and rebuildable graph projections |
 | AI operations | Claude, Codex, Gemini CLI, and Antigravity session indexing; skill, MCP, and hook event extraction where each provider exposes them; incident clustering; and guarded local LLM assessments |
@@ -685,15 +685,15 @@ Cortex uses SQLite with:
 
 - WAL mode
 - A bounded r2d2 connection pool
-- FTS5 external-content indexing with `logs_fts` for the complete log corpus and a transcript-only `ai_logs_fts` projection for AI session search
+- FTS5 external-content indexing with `logs_fts` for the complete log corpus and a transcript-only `ai_logs_fts` projection that indexes message text plus provider scope for AI session search
 - Covering and composite indexes for common filters and timelines
 - Transactional batch writes
 - Durable source checkpoints and parse errors
 - Maintenance job tracking
 - Online backup support
 - Integrity checks, checkpoints, and vacuum workflows
 
-The current schema history contains 60 sequential migrations. CI derives this denominator from `KNOWN_SCHEMA_VERSION` and the migration registry. Migration 60 builds the transcript-only `ai_logs_fts` index once from rows with `ai_tool IS NOT NULL`; on large transcript histories this can hold the startup write transaction while that smaller derived index is populated. Subsequent transcript inserts and deletes maintain it incrementally, while global `logs_fts` remains the full-corpus search index. Server forwarding receipts have a seven-day replay horizon and are removed when their canonical evidence is deleted. The sender spool is intentionally shorter and bounded: an individual source retains at most 1,024 records or 1 MiB and evicts records older than one day; the aggregate spool retains at most 4,096 records or 4 MiB. Eviction removes the original payload and retains a pending gap marker so evidence loss remains visible. A retry after the server receipt horizon is a new ingestion attempt.
+The current schema history contains 61 sequential migrations. CI derives this denominator from `KNOWN_SCHEMA_VERSION` and the migration registry. Migration 60 introduced the transcript-only `ai_logs_fts` projection. Migration 61 rebuilds that derived index with provider scope and adds an AI-only `(timestamp, ai_tool)` index so time-bounded session searches can derive a safe rowid floor before FTS walks common terms. On large transcript histories these one-time derived-index rebuilds can hold the startup write transaction while they populate. Subsequent transcript inserts and deletes maintain `ai_logs_fts` incrementally, while global `logs_fts` remains the full-corpus search index. Server forwarding receipts have a seven-day replay horizon and are removed when their canonical evidence is deleted. The sender spool is intentionally shorter and bounded: an individual source retains at most 1,024 records or 1 MiB and evicts records older than one day; the aggregate spool retains at most 4,096 records or 4 MiB. Eviction removes the original payload and retains a pending gap marker so evidence loss remains visible. A retry after the server receipt horizon is a new ingestion attempt.
 
 ### Authoritative and derived data
 

docs/architecture.md

@@ -32,7 +32,7 @@ SQLite database and a service layer:
 | `config.rs` | all | Layered config: defaults → `config.toml` → `~/.cortex/.env` → process env; startup validation (non-loopback auth gate) |
 | `runtime.rs` + `runtime/` | all | `RuntimeCore`: wires pool, ingest, auth policy; spawns the maintenance tasks below |
 | `app/` | core | `CortexService` service layer — shared limits/validation for MCP, REST, and CLI |
-| `db/` | core | SQLite pool + 60 sequential migrations, FTS5 queries, retention and storage-budget maintenance |
+| `db/` | core | SQLite pool + 61 sequential migrations, FTS5 queries, retention and storage-budget maintenance |
 | `receiver/` + `receiver.rs` | core | UDP + TCP listeners (supervised with restart + backoff), RFC 3164/5424 + CEF parsing |
 | `ingest.rs` | core | mpsc channel + batch writer (one pool connection reserved for this writer) |
 | `otlp.rs` + `otlp/` | core | OTLP/HTTP protobuf ingest: `POST /v1/logs` (4 MiB cap), `POST /v1/metrics` and `POST /v1/traces` (8 MiB cap); all use `CORTEX_TOKEN` auth |

packages/cortex-rmcp/README.md

@@ -17,7 +17,7 @@ Cortex began as a syslog receiver. It now covers network logs, Docker, managed f
 | Area | What Cortex provides |
 | --- | --- |
 | Ingest | UDP/TCP syslog, OTLP/HTTP logs, metrics, and traces, Docker logs and events, managed file tails, host heartbeats, AI transcripts, shell history, agent command records, and fleet inventory |
-| Storage | SQLite in WAL mode, FTS5 full-text search, bounded metadata, retention, storage budgets, maintenance jobs, checkpoints, and 60 sequential schema migrations |
+| Storage | SQLite in WAL mode, FTS5 full-text search, bounded metadata, retention, storage budgets, maintenance jobs, checkpoints, and 61 sequential schema migrations |
 | Investigation | Search, filtering, context, timelines, patterns, anomaly comparison, cross-source correlation, recurring error signatures, deterministic incident bundles, and graph explanations |
 | Fleet intelligence | SSH and API inventory collectors, host state, service topology, container and route relationships, redacted evidence, and rebuildable graph projections |
 | AI operations | Claude, Codex, Gemini CLI, and Antigravity session indexing; skill, MCP, and hook event extraction where each provider exposes them; incident clustering; and guarded local LLM assessments |
@@ -685,15 +685,15 @@ Cortex uses SQLite with:
 
 - WAL mode
 - A bounded r2d2 connection pool
-- FTS5 external-content indexing with `logs_fts` for the complete log corpus and a transcript-only `ai_logs_fts` projection for AI session search
+- FTS5 external-content indexing with `logs_fts` for the complete log corpus and a transcript-only `ai_logs_fts` projection that indexes message text plus provider scope for AI session search
 - Covering and composite indexes for common filters and timelines
 - Transactional batch writes
 - Durable source checkpoints and parse errors
 - Maintenance job tracking
 - Online backup support
 - Integrity checks, checkpoints, and vacuum workflows
 
-The current schema history contains 60 sequential migrations. CI derives this denominator from `KNOWN_SCHEMA_VERSION` and the migration registry. Migration 60 builds the transcript-only `ai_logs_fts` index once from rows with `ai_tool IS NOT NULL`; on large transcript histories this can hold the startup write transaction while that smaller derived index is populated. Subsequent transcript inserts and deletes maintain it incrementally, while global `logs_fts` remains the full-corpus search index. Server forwarding receipts have a seven-day replay horizon and are removed when their canonical evidence is deleted. The sender spool is intentionally shorter and bounded: an individual source retains at most 1,024 records or 1 MiB and evicts records older than one day; the aggregate spool retains at most 4,096 records or 4 MiB. Eviction removes the original payload and retains a pending gap marker so evidence loss remains visible. A retry after the server receipt horizon is a new ingestion attempt.
+The current schema history contains 61 sequential migrations. CI derives this denominator from `KNOWN_SCHEMA_VERSION` and the migration registry. Migration 60 introduced the transcript-only `ai_logs_fts` projection. Migration 61 rebuilds that derived index with provider scope and adds an AI-only `(timestamp, ai_tool)` index so time-bounded session searches can derive a safe rowid floor before FTS walks common terms. On large transcript histories these one-time derived-index rebuilds can hold the startup write transaction while they populate. Subsequent transcript inserts and deletes maintain `ai_logs_fts` incrementally, while global `logs_fts` remains the full-corpus search index. Server forwarding receipts have a seven-day replay horizon and are removed when their canonical evidence is deleted. The sender spool is intentionally shorter and bounded: an individual source retains at most 1,024 records or 1 MiB and evicts records older than one day; the aggregate spool retains at most 4,096 records or 4 MiB. Eviction removes the original payload and retains a pending gap marker so evidence loss remains visible. A retry after the server receipt horizon is a new ingestion attempt.
 
 ### Authoritative and derived data
 

src/app/services/skill_backfill.rs

@@ -26,19 +26,17 @@
 //! through `api.rs`.
 //!
 //! **Claude row recovery**: `logs.message` for a Claude row is
-//! `claude::extract_message()`'s plain-text `content` extraction (e.g. "hi")
-//! — it never carries the raw `attributionSkill`/`attributionPlugin` JSON
-//! fields, unlike Codex where the transcript text itself (including the
-//! `<skill><name>` tag) survives `scrub_ai_message` intact. The only place
-//! that data still exists is the original JSONL file on disk, so Claude rows
-//! are recovered by re-reading the specific source line (via the shared
-//! `scanner::read_transcript_lines` helper, which applies the same bounded,
-//! newline-delimited record semantics as the ingest path) located by the
-//! persisted `ai_transcript_path` column and the `line_no` scanner.rs recorded
-//! in `metadata_json` at ingest time. Rows whose source file or line can no
-//! longer be located (deleted/rotated/legacy metadata predating `line_no`, or
-//! a line now exceeding the record-size bound) are counted in
-//! `source_unavailable` rather than treated as an error.
+//! `claude::extract_message()`'s plain-text `content` extraction. It does not
+//! retain raw `attributionSkill`/`attributionPlugin` JSON fields, so those
+//! still require the original JSONL source line. Modern package-qualified
+//! `<command-message>` / `<command-name>` envelopes are different: they are
+//! user-message content and survive both normalization and privacy-preserving
+//! transcript forwarding. The backfill therefore extracts those envelopes
+//! directly from a persisted user `logs.message` first, then falls back to
+//! `scanner::read_transcript_lines` for structured attribution when the source
+//! path plus `metadata_json.line_no` are locally recoverable. Rows with neither
+//! persisted command evidence nor a recoverable source are counted in
+//! `source_unavailable` rather than silently treated as complete.
 //!
 //! **Idempotency caveat**: re-running the backfill is a no-op *only while the
 //! source transcript files are unchanged*. Because the recovered `skill_name`
@@ -74,7 +72,7 @@ use tokio::sync::Semaphore;
 use crate::db::{DbPool, SkillEventInsert, insert_skill_events};
 use crate::scanner::read_transcript_lines;
 use crate::scanner::skill_events::{
-    claude_line_may_contain_skill_event, extract_claude_skill_events,
+    ExtractedSkillEvent, claude_line_may_contain_skill_event, extract_claude_skill_events,
     extract_codex_skill_events_with_kind,
 };
 
@@ -116,6 +114,38 @@ struct CandidateRow {
     metadata_json: Option<String>,
 }
 
+fn extract_forwarded_claude_skill_events(row: &CandidateRow) -> Vec<ExtractedSkillEvent> {
+    if !row.message.contains("<command-message>") && !row.message.contains("<command-name>") {
+        return Vec::new();
+    }
+    let is_user = row
+        .metadata_json
+        .as_deref()
+        .and_then(|json| serde_json::from_str::<serde_json::Value>(json).ok())
+        .and_then(|metadata| {
+            metadata
+                .get("event_kind")
+                .and_then(serde_json::Value::as_str)
+                .map(str::to_owned)
+        })
+        .is_some_and(|event_kind| event_kind == "user");
+    if !is_user {
+        return Vec::new();
+    }
+
+    extract_claude_skill_events(&serde_json::json!({
+        "message": {
+            "role": "user",
+            "content": [
+                {
+                    "type": "text",
+                    "text": row.message.as_str()
+                }
+            ]
+        }
+    }))
+}
+
 impl CortexService {
     pub async fn backfill_skill_events(
         &self,
@@ -174,19 +204,22 @@ fn run_backfill(
         result.scanned += rows.len() as u64;
         remaining = remaining.saturating_sub(rows.len() as u64);
 
-        // Resolve every Claude row's source line up front, grouped by file so
-        // rows sharing a transcript file open and scan it once per chunk. Two
-        // borrowed maps over `rows` (no owned-String clones): `row_source` maps
-        // each row id to its `(path, line_no)`, and `wanted_by_file` collects
-        // the distinct line numbers to pull from each file. See the "Claude row
-        // recovery" note at the top of this file for why `row.message` can't be
-        // used directly.
+        // Resolve source lines only for Claude rows that cannot be recovered
+        // from a persisted command envelope. Rows sharing a transcript file
+        // open and scan it once per chunk. Two borrowed maps over `rows` (no
+        // owned-String clones): `row_source` maps each row id to its
+        // `(path, line_no)`, and `wanted_by_file` collects the distinct line
+        // numbers to pull from each file. See the "Claude row recovery" note at
+        // the top of this file for the normalized-message/source split.
         let mut row_source: HashMap<i64, (&str, usize)> = HashMap::new();
         let mut wanted_by_file: HashMap<&str, HashSet<usize>> = HashMap::new();
         for row in &rows {
             if row.ai_tool != "claude" {
                 continue;
             }
+            if !extract_forwarded_claude_skill_events(row).is_empty() {
+                continue;
+            }
             match (
                 row.ai_transcript_path.as_deref(),
                 row.metadata_json.as_deref().and_then(line_no_from_metadata),
@@ -228,30 +261,35 @@ fn run_backfill(
         for row in &rows {
             let extracted = match row.ai_tool.as_str() {
                 "claude" => {
-                    let Some(&(path, line_no)) = row_source.get(&row.id) else {
-                        // Already counted in `source_unavailable` above.
-                        continue;
-                    };
-                    let Some(line_text) = resolved.get(&(path, line_no)) else {
-                        result.source_unavailable += 1;
-                        tracing::debug!(
-                            log_id = row.id,
-                            path,
-                            line_no,
-                            "skill backfill: transcript line unavailable (missing file or line out of range)"
-                        );
-                        continue;
-                    };
-                    // Cheap short-circuit on the actual raw JSON line (not
-                    // the scrubbed `row.message`) before parsing.
-                    if !claude_line_may_contain_skill_event(line_text) {
-                        continue;
-                    }
-                    match serde_json::from_str::<serde_json::Value>(line_text) {
-                        Ok(value) => extract_claude_skill_events(&value),
-                        Err(_) => {
-                            result.parse_errors += 1;
+                    let forwarded = extract_forwarded_claude_skill_events(row);
+                    if !forwarded.is_empty() {
+                        forwarded
+                    } else {
+                        let Some(&(path, line_no)) = row_source.get(&row.id) else {
+                            // Already counted in `source_unavailable` above.
                             continue;
+                        };
+                        let Some(line_text) = resolved.get(&(path, line_no)) else {
+                            result.source_unavailable += 1;
+                            tracing::debug!(
+                                log_id = row.id,
+                                path,
+                                line_no,
+                                "skill backfill: transcript line unavailable (missing file or line out of range)"
+                            );
+                            continue;
+                        };
+                        // Cheap short-circuit on the actual raw JSON line (not
+                        // the scrubbed `row.message`) before parsing.
+                        if !claude_line_may_contain_skill_event(line_text) {
+                            continue;
+                        }
+                        match serde_json::from_str::<serde_json::Value>(line_text) {
+                            Ok(value) => extract_claude_skill_events(&value),
+                            Err(_) => {
+                                result.parse_errors += 1;
+                                continue;
+                            }
                         }
                     }
                 }