diff --git a/README.md b/README.md index f8d66634..f3991c8f 100644 --- a/README.md +++ b/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,7 +685,7 @@ 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 @@ -693,7 +693,7 @@ Cortex uses SQLite with: - 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 diff --git a/docs/architecture.md b/docs/architecture.md index f9b3c9ad..d05f7949 100644 --- a/docs/architecture.md +++ b/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 | diff --git a/packages/cortex-rmcp/README.md b/packages/cortex-rmcp/README.md index f8d66634..f3991c8f 100644 --- a/packages/cortex-rmcp/README.md +++ b/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,7 +685,7 @@ 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 @@ -693,7 +693,7 @@ Cortex uses SQLite with: - 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 diff --git a/src/app/services/skill_backfill.rs b/src/app/services/skill_backfill.rs index d971de91..5dfd901c 100644 --- a/src/app/services/skill_backfill.rs +++ b/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 -//! `` 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 +//! `` / `` 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, } +fn extract_forwarded_claude_skill_events(row: &CandidateRow) -> Vec { + if !row.message.contains("") && !row.message.contains("") { + return Vec::new(); + } + let is_user = row + .metadata_json + .as_deref() + .and_then(|json| serde_json::from_str::(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 = HashMap::new(); let mut wanted_by_file: HashMap<&str, HashSet> = 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::(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::(line_text) { + Ok(value) => extract_claude_skill_events(&value), + Err(_) => { + result.parse_errors += 1; + continue; + } } } } diff --git a/src/app/services/skill_backfill_tests.rs b/src/app/services/skill_backfill_tests.rs index 1ee3bef4..92830fec 100644 --- a/src/app/services/skill_backfill_tests.rs +++ b/src/app/services/skill_backfill_tests.rs @@ -103,6 +103,28 @@ fn insert_legacy_claude_log_row_without_source(pool: &DbPool) -> i64 { insert_claude_row(pool, None) } +fn insert_forwarded_claude_command_row(pool: &DbPool, message: &str) -> i64 { + let conn = pool.get().unwrap(); + let metadata_json = serde_json::json!({ + "content_scrubbed": true, + "event_kind": "user", + "source": { + "adapter_version": "cortex-ai-forwarder-v1", + "locator": "sha256:forwarded-transcript", + "provider": "claude" + }, + "source_type": "transcript" + }) + .to_string(); + conn.execute( + "INSERT INTO logs (timestamp, hostname, severity, message, raw, source_ip, ai_tool, ai_project, ai_session_id, ai_transcript_path, metadata_json) + VALUES ('2026-09-04T02:10:01.475Z', 'devhost', 'info', ?1, '', 'transcript://claude_project', 'claude', 'cortex', 'sess-forwarded', 'sha256:forwarded-transcript', ?2)", + rusqlite::params![message, metadata_json], + ) + .unwrap(); + conn.last_insert_rowid() +} + #[tokio::test] #[serial(skill_backfill_guard)] async fn dry_run_reports_counts_without_inserting() { @@ -211,6 +233,43 @@ async fn backfill_recovers_modern_claude_skill_command_envelope() { assert_eq!(kind, "claude_skill_command"); } +#[tokio::test] +#[serial(skill_backfill_guard)] +async fn backfill_recovers_forwarded_scrubbed_claude_command_envelope() { + let (service, _dir) = test_service(); + let pool = service.pool_for_test(); + insert_forwarded_claude_command_row( + &pool, + "vibin:review-pr /vibin:review-pr Labby 568 + Depot 79", + ); + + let result = service + .backfill_skill_events(SkillBackfillRequest { + since: Some("2026-09-01T00:00:00Z".to_string()), + 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, evidence): (String, Option, String, String) = conn + .query_row( + "SELECT skill_name, skill_plugin, event_kind, evidence_kind FROM ai_skill_events", + [], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + ) + .unwrap(); + assert_eq!(skill, "vibin:review-pr"); + assert_eq!(plugin.as_deref(), Some("vibin")); + assert_eq!(kind, "claude_skill_command"); + assert_eq!(evidence, "transcript_content"); +} + #[tokio::test] #[serial(skill_backfill_guard)] async fn claude_row_without_transcript_path_counts_as_source_unavailable() { diff --git a/src/db/pool.rs b/src/db/pool.rs index 5f644ea1..cbb677ac 100644 --- a/src/db/pool.rs +++ b/src/db/pool.rs @@ -258,7 +258,7 @@ pub(crate) fn try_write_conn_for( } } -pub const KNOWN_SCHEMA_VERSION: i64 = 60; +pub const KNOWN_SCHEMA_VERSION: i64 = 61; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct SchemaVersionInfo { @@ -400,6 +400,7 @@ pub fn init_pool(config: &StorageConfig) -> Result { -- syslog/Docker corpus while preserving the same message tokenizer. CREATE VIRTUAL TABLE IF NOT EXISTS ai_logs_fts USING fts5( message, + ai_tool, content='logs', content_rowid='id', tokenize='porter unicode61' @@ -418,7 +419,8 @@ pub fn init_pool(config: &StorageConfig) -> Result { CREATE TRIGGER IF NOT EXISTS logs_ai_transcript AFTER INSERT ON logs WHEN new.ai_tool IS NOT NULL BEGIN - INSERT INTO ai_logs_fts(rowid, message) VALUES (new.id, new.message); + INSERT INTO ai_logs_fts(rowid, message, ai_tool) + VALUES (new.id, new.message, new.ai_tool); END; -- Unlike the global FTS index, transcript volume is small enough to @@ -426,8 +428,8 @@ pub fn init_pool(config: &StorageConfig) -> Result { -- stale transcript terms instead of leaving phantom AI search entries. CREATE TRIGGER IF NOT EXISTS logs_ad_transcript AFTER DELETE ON logs WHEN old.ai_tool IS NOT NULL BEGIN - INSERT INTO ai_logs_fts(ai_logs_fts, rowid, message) - VALUES ('delete', old.id, old.message); + INSERT INTO ai_logs_fts(ai_logs_fts, rowid, message, ai_tool) + VALUES ('delete', old.id, old.message, old.ai_tool); END; -- Hostname registry for quick lookups @@ -3584,6 +3586,52 @@ pub fn init_pool(config: &StorageConfig) -> Result { ); } + // Migration 61: production dogfood showed that a common FTS term can + // still walk a large fraction of transcript history before post-FTS tool + // and time filters are applied. Index the small provider value inside the + // transcript FTS itself and add an AI-only timestamp index used to derive + // an exact-safe rowid floor for bounded searches. Rebuild is derived and + // atomic, matching migration 60's crash-recovery behavior. + if !migration_applied(&conn, 61)? { + tracing::info!( + "Migration 61: rebuilding transcript FTS with provider scope; this may take time on large transcript histories" + ); + let started = std::time::Instant::now(); + conn.execute_batch( + "BEGIN IMMEDIATE; + CREATE INDEX IF NOT EXISTS idx_logs_ai_timestamp_tool + ON logs(timestamp, ai_tool) WHERE ai_tool IS NOT NULL; + DROP TRIGGER IF EXISTS logs_ai_transcript; + DROP TRIGGER IF EXISTS logs_ad_transcript; + DROP TABLE IF EXISTS ai_logs_fts; + CREATE VIRTUAL TABLE ai_logs_fts USING fts5( + message, + ai_tool, + content='logs', + content_rowid='id', + tokenize='porter unicode61' + ); + INSERT INTO ai_logs_fts(rowid, message, ai_tool) + SELECT id, message, ai_tool FROM logs WHERE ai_tool IS NOT NULL; + CREATE TRIGGER logs_ai_transcript AFTER INSERT ON logs + WHEN new.ai_tool IS NOT NULL BEGIN + INSERT INTO ai_logs_fts(rowid, message, ai_tool) + VALUES (new.id, new.message, new.ai_tool); + END; + CREATE TRIGGER logs_ad_transcript AFTER DELETE ON logs + WHEN old.ai_tool IS NOT NULL BEGIN + INSERT INTO ai_logs_fts(ai_logs_fts, rowid, message, ai_tool) + VALUES ('delete', old.id, old.message, old.ai_tool); + END; + INSERT OR IGNORE INTO schema_migrations (version) VALUES (61); + COMMIT;", + )?; + tracing::info!( + elapsed_ms = started.elapsed().as_millis(), + "Migration 61: provider-scoped transcript FTS index ready" + ); + } + if table_exists(&conn, "host_heartbeats")? && table_exists(&conn, "host_heartbeats_latest")? { let deleted_heartbeat_latest = conn.execute( "DELETE FROM host_heartbeats_latest diff --git a/src/db/pool_tests.rs b/src/db/pool_tests.rs index df36e681..9615f927 100644 --- a/src/db/pool_tests.rs +++ b/src/db/pool_tests.rs @@ -10,9 +10,9 @@ use rusqlite::OptionalExtension; #[test] fn documented_schema_count_matches_known_version() { - assert_eq!(KNOWN_SCHEMA_VERSION, 60); - assert!(include_str!("../../README.md").contains("60 sequential schema migrations")); - assert!(include_str!("../../docs/architecture.md").contains("60 sequential migrations")); + assert_eq!(KNOWN_SCHEMA_VERSION, 61); + assert!(include_str!("../../README.md").contains("61 sequential schema migrations")); + assert!(include_str!("../../docs/architecture.md").contains("61 sequential migrations")); } #[test] @@ -130,6 +130,65 @@ fn migration_60_creates_transcript_only_fts_and_insert_trigger() { ); } +#[test] +fn migration_61_scopes_transcript_fts_by_tool_and_indexes_ai_time() { + let dir = tempfile::tempdir().unwrap(); + let pool = init_pool(&test_storage_config(dir.path().join("migration-61.db"))).unwrap(); + let conn = pool.get().unwrap(); + + let columns: Vec = conn + .prepare("PRAGMA table_info(ai_logs_fts)") + .unwrap() + .query_map([], |row| row.get(1)) + .unwrap() + .collect::>() + .unwrap(); + assert!(columns.iter().any(|column| column == "ai_tool")); + + let index_exists: i64 = conn + .query_row( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'index' AND name = 'idx_logs_ai_timestamp_tool'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(index_exists, 1); + drop(conn); + + let mut claude = migration_test_log(); + claude.ai_tool = Some("claude".into()); + claude.ai_session_id = Some("migration-claude".into()); + claude.message = "sharedneedle from claude".into(); + claude.raw = claude.message.clone(); + + let mut codex = migration_test_log(); + codex.ai_tool = Some("codex".into()); + codex.ai_session_id = Some("migration-codex".into()); + codex.message = "sharedneedle from codex".into(); + codex.raw = codex.message.clone(); + insert_logs_batch(&pool, &[claude, codex]).unwrap(); + + let conn = pool.get().unwrap(); + let claude_matches: i64 = conn + .query_row( + r#"SELECT COUNT(*) FROM ai_logs_fts + WHERE ai_logs_fts MATCH 'message : (sharedneedle) AND ai_tool : "claude"'"#, + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(claude_matches, 1); + + let marker_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM schema_migrations WHERE version = 61", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(marker_count, 1); +} + #[test] fn migration_57_adds_transcript_receipt_fingerprints() { let dir = tempfile::tempdir().unwrap(); diff --git a/src/db/queries.rs b/src/db/queries.rs index b40ee6b7..3837c1e1 100644 --- a/src/db/queries.rs +++ b/src/db/queries.rs @@ -1373,6 +1373,19 @@ pub fn search_ai_sessions( const CANDIDATE_CAP: usize = 5_000; +fn ai_session_fts_query(query: &str, tool: Option<&str>) -> String { + let mut scoped = format!("message : ({query})"); + if let Some(tool) = tool.filter(|tool| { + !tool.is_empty() + && tool + .chars() + .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '_' | '-' | '.')) + }) { + scoped.push_str(&format!(" AND ai_tool : \"{tool}\"")); + } + scoped +} + fn search_ai_sessions_sql( params: &SearchAiSessionsParams, limit: usize, @@ -1382,7 +1395,21 @@ fn search_ai_sessions_sql( let mut query_params = SqlParams::new(2); query_params .bindings - .push(rusqlite::types::Value::Text(params.query.clone())); + .push(rusqlite::types::Value::Text(ai_session_fts_query( + ¶ms.query, + params.ai_tool.as_deref(), + ))); + let mut fts_rowid_floor = String::new(); + if let Some(since) = ¶ms.since { + let idx = query_params.push_text(since.clone()); + fts_rowid_floor = format!( + " AND ai_logs_fts.rowid >= COALESCE( + (SELECT MIN(id) + FROM logs INDEXED BY idx_logs_ai_timestamp_tool + WHERE ai_tool IS NOT NULL AND timestamp >= ?{idx}), + 9223372036854775807)" + ); + } push_ai_scope_filters( &mut filters, &mut query_params, @@ -1410,7 +1437,7 @@ fn search_ai_sessions_sql( l.message FROM ai_logs_fts JOIN logs l ON l.id = ai_logs_fts.rowid - WHERE ai_logs_fts MATCH ?1{filters} + WHERE ai_logs_fts MATCH ?1{fts_rowid_floor}{filters} ORDER BY ai_logs_fts.rowid DESC LIMIT {} ), diff --git a/src/db/queries_tests.rs b/src/db/queries_tests.rs index ba727198..051545cd 100644 --- a/src/db/queries_tests.rs +++ b/src/db/queries_tests.rs @@ -1206,6 +1206,90 @@ fn search_ai_sessions_query_plan_uses_session_host_time_index() { ); } +#[test] +fn search_ai_sessions_pushes_tool_and_time_scope_into_fts_candidates() { + let (pool, _dir) = test_pool(); + insert_logs_batch( + &pool, + &[ + make_ai_entry( + "2026-09-10T12:00:00Z", + "host-a", + "claude", + "/tmp/project", + "old-claude", + "error before the requested window", + ), + make_ai_entry( + "2026-09-12T12:00:00Z", + "host-a", + "claude", + "/tmp/project", + "recent-claude", + "error inside the requested window", + ), + make_ai_entry( + "2026-09-12T12:01:00Z", + "host-a", + "codex", + "/tmp/project", + "recent-codex", + "error from a different provider", + ), + ], + ) + .unwrap(); + + let params = SearchAiSessionsParams { + query: "error".into(), + ai_tool: Some("claude".into()), + since: Some("2026-09-11T00:00:00Z".into()), + limit: Some(5), + ..Default::default() + }; + let result = search_ai_sessions(&pool, ¶ms).unwrap(); + assert_eq!(result.sessions.len(), 1); + assert_eq!(result.sessions[0].ai_session_id, "recent-claude"); + + let (sql, bindings) = search_ai_sessions_sql(¶ms, 5); + assert!( + sql.contains("INDEXED BY idx_logs_ai_timestamp_tool"), + "time-bounded session search must derive a safe FTS rowid floor from the AI-only time index:\n{sql}" + ); + assert!( + sql.contains("ai_logs_fts.rowid >="), + "time-bounded session search must push its safe rowid floor into FTS:\n{sql}" + ); + + let Some(rusqlite::types::Value::Text(fts_query)) = bindings.first() else { + panic!("first search_sessions binding must be the FTS query"); + }; + assert!( + fts_query.contains("message : (error)") && fts_query.contains("ai_tool : \"claude\""), + "tool scope must be pushed into the transcript FTS query itself; got {fts_query:?}" + ); +} + +#[test] +fn search_ai_sessions_does_not_interpolate_unsafe_tool_scope_into_fts() { + let params = SearchAiSessionsParams { + query: "needle".into(), + ai_tool: Some("claude\" OR message:error".into()), + limit: Some(5), + ..Default::default() + }; + + let (sql, bindings) = search_ai_sessions_sql(¶ms, 5); + let Some(rusqlite::types::Value::Text(fts_query)) = bindings.first() else { + panic!("first search_sessions binding must be the FTS query"); + }; + assert_eq!(fts_query, "message : (needle)"); + assert!( + sql.contains("l.ai_tool = ?"), + "unsafe provider values must fall back to the parameterized relational equality filter" + ); +} + #[test] fn search_ai_sessions_finds_rows_inserted_after_rollup_refresh() { let (pool, _dir) = test_pool();