fix: repair transcript skill evidence and 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 58 sequential schema migrations |
+| Storage | SQLite in WAL mode, FTS5 full-text search, bounded metadata, retention, storage budgets, maintenance jobs, checkpoints, and 60 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 for log messages
+- FTS5 external-content indexing with `logs_fts` for the complete log corpus and a transcript-only `ai_logs_fts` projection 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 58 sequential migrations. CI derives this denominator from `KNOWN_SCHEMA_VERSION` and the migration registry. 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 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.
 
 ### Authoritative and derived data
 

docs/architecture.md

@@ -12,7 +12,7 @@ SQLite database and a service layer:
 1. **Log intelligence core** — syslog UDP/TCP ingest, OTLP HTTP/protobuf
    `/v1/logs`, `/v1/metrics`, and `/v1/traces`,
    host-local agent Docker log ingest, legacy central pull Docker compatibility,
-   AI transcript indexing, FTS5 search, and the
+   AI transcript indexing, global FTS5 log search, transcript-only FTS5 session search, and the
    56-action `cortex` MCP tool plus the `/api/*` REST mirror. Source: `src/receiver/`,
    `src/ingest.rs`, `src/otlp.rs`, `src/agent/`, `src/docker_ingest/`, `src/db/`,
    `src/mcp/`, `src/api.rs`, `src/app/`.
@@ -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 + 58 sequential migrations, FTS5 queries, retention and storage-budget maintenance |
+| `db/` | core | SQLite pool + 60 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 58 sequential schema migrations |
+| Storage | SQLite in WAL mode, FTS5 full-text search, bounded metadata, retention, storage budgets, maintenance jobs, checkpoints, and 60 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 for log messages
+- FTS5 external-content indexing with `logs_fts` for the complete log corpus and a transcript-only `ai_logs_fts` projection 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 58 sequential migrations. CI derives this denominator from `KNOWN_SCHEMA_VERSION` and the migration registry. 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 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.
 
 ### Authoritative and derived data
 

src/app/services/skill_backfill.rs

@@ -74,7 +74,8 @@ use tokio::sync::Semaphore;
 use crate::db::{DbPool, SkillEventInsert, insert_skill_events};
 use crate::scanner::read_transcript_lines;
 use crate::scanner::skill_events::{
-    extract_claude_skill_events, extract_codex_skill_events_with_kind,
+    claude_line_may_contain_skill_event, extract_claude_skill_events,
+    extract_codex_skill_events_with_kind,
 };
 
 use super::super::models::{SkillBackfillRequest, SkillBackfillResult};
@@ -243,7 +244,7 @@ fn run_backfill(
                     };
                     // Cheap short-circuit on the actual raw JSON line (not
                     // the scrubbed `row.message`) before parsing.
-                    if !line_text.contains("attributionSkill") {
+                    if !claude_line_may_contain_skill_event(line_text) {
                         continue;
                     }
                     match serde_json::from_str::<serde_json::Value>(line_text) {

src/app/services/skill_backfill_tests.rs

@@ -174,6 +174,43 @@ async fn real_run_inserts_events_and_is_idempotent() {
     assert_eq!(second.skipped_duplicates, 1);
 }
 
+#[tokio::test]
+#[serial(skill_backfill_guard)]
+async fn backfill_recovers_modern_claude_skill_command_envelope() {
+    let (service, dir) = test_service();
+    let pool = service.pool_for_test();
+    insert_claude_log_row(
+        &pool,
+        dir.path(),
+        "session-command.jsonl",
+        r#"{"sessionId":"sess-command","message":{"role":"user","content":[{"type":"text","text":"<command-message>vibin:repo-status</command-message> <command-name>/vibin:repo-status</command-name>"}]}}"#,
+    );
+
+    let result = service
+        .backfill_skill_events(SkillBackfillRequest {
+            since: None,
+            limit: Some(100),
+            dry_run: false,
+        })
+        .await
+        .unwrap();
+    assert_eq!(result.scanned, 1);
+    assert_eq!(result.inserted, 1);
+    assert_eq!(result.source_unavailable, 0);
+
+    let conn = pool.get().unwrap();
+    let (skill, plugin, kind): (String, Option<String>, String) = conn
+        .query_row(
+            "SELECT skill_name, skill_plugin, event_kind FROM ai_skill_events",
+            [],
+            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
+        )
+        .unwrap();
+    assert_eq!(skill, "vibin:repo-status");
+    assert_eq!(plugin.as_deref(), Some("vibin"));
+    assert_eq!(kind, "claude_skill_command");
+}
+
 #[tokio::test]
 #[serial(skill_backfill_guard)]
 async fn claude_row_without_transcript_path_counts_as_source_unavailable() {

src/app/skill_signal_detectors.rs

@@ -62,6 +62,22 @@ fn contains_any_phrase(haystack_lower: &str, phrases: &[&str]) -> bool {
     phrases.iter().any(|p| haystack_lower.contains(p))
 }
 
+/// Return true only for transcript rows explicitly persisted as user-authored.
+/// Unknown/legacy speaker metadata is intentionally not guessed; extractor
+/// revision replay repairs historical rows before incident scoring.
+pub fn transcript_event_is_user(metadata_json: Option<&str>) -> bool {
+    metadata_json
+        .and_then(|raw| serde_json::from_str::<serde_json::Value>(raw).ok())
+        .and_then(|value| {
+            value
+                .get("event_kind")
+                .and_then(serde_json::Value::as_str)
+                .map(str::to_owned)
+        })
+        .as_deref()
+        == Some("user")
+}
+
 pub fn detect_user_correction(message: &str) -> bool {
     let lower = message.to_ascii_lowercase();
     contains_any_phrase(&lower, USER_CORRECTION_PHRASES)

src/cli/complete_tests.rs

@@ -160,6 +160,7 @@ fn refreshed_event_leaves_expose_exact_parser_flags() {
             "sessions skillinvestigate",
             &[
                 "--incident-id",
+                "--skill",
                 "--plugin",
                 "--tool",
                 "--project",

src/db/maintenance.rs

@@ -610,29 +610,31 @@ fn fts_incremental_merge(pool: &DbPool, deleted_rows: usize, merge_pages: u32) {
     for i in 0..iterations {
         match crate::db::write_conn(pool) {
             Ok(conn) => {
-                match conn.execute(
-                    "INSERT INTO logs_fts(logs_fts, rank) VALUES('merge', ?1)",
-                    [pages],
-                ) {
-                    Ok(_) => {
-                        tracing::trace!(
-                            iteration = i + 1,
-                            total_iterations = iterations,
-                            pages,
-                            "FTS incremental merge iteration"
-                        );
-                    }
-                    Err(e) => {
-                        // A correctly-formed merge only errors on a genuine
-                        // operational problem (busy/locked) or real corruption.
-                        // Log and stop — never auto-escalate to optimize/rebuild,
-                        // which rewrite the entire index under the write lock.
-                        tracing::warn!(
-                            error = %e,
-                            iteration = i + 1,
-                            "FTS incremental merge failed; stopping (no auto optimize/rebuild)"
-                        );
-                        return;
+                for index in ["logs_fts", "ai_logs_fts"] {
+                    let sql = format!("INSERT INTO {index}({index}, rank) VALUES('merge', ?1)");
+                    match conn.execute(&sql, [pages]) {
+                        Ok(_) => {
+                            tracing::trace!(
+                                index,
+                                iteration = i + 1,
+                                total_iterations = iterations,
+                                pages,
+                                "FTS incremental merge iteration"
+                            );
+                        }
+                        Err(e) => {
+                            // A correctly-formed merge only errors on a genuine
+                            // operational problem (busy/locked) or real corruption.
+                            // Log and stop — never auto-escalate to optimize/rebuild,
+                            // which rewrite the entire index under the write lock.
+                            tracing::warn!(
+                                error = %e,
+                                index,
+                                iteration = i + 1,
+                                "FTS incremental merge failed; stopping (no auto optimize/rebuild)"
+                            );
+                            return;
+                        }
                     }
                 }
             }