From 1a0a8c34e1bc53e777d6ffdb93736632257777ab Mon Sep 17 00:00:00 2001 From: Jake Magar Date: Thu, 17 Sep 2026 20:49:32 -0400 Subject: [PATCH 1/2] fix: repair transcript skill evidence and search --- README.md | 6 +- docs/architecture.md | 4 +- packages/cortex-rmcp/README.md | 6 +- src/app/services/skill_backfill.rs | 5 +- src/app/services/skill_backfill_tests.rs | 37 +++++++ src/app/skill_signal_detectors.rs | 16 +++ src/cli/complete_tests.rs | 1 + src/db/maintenance.rs | 48 +++++---- src/db/pool.rs | 110 ++++++++++++++++++- src/db/pool_tests.rs | 121 ++++++++++++++++++++- src/db/queries.rs | 8 +- src/db/queries_tests.rs | 11 +- src/db/skill_incident_evidence.rs | 9 +- src/db/skill_incident_evidence_tests.rs | 2 +- src/db/skill_incidents.rs | 13 +-- src/db/skill_incidents_tests.rs | 64 ++++++++++- src/mcp/action_flags.rs | 1 + src/mcp/schemas.rs | 63 ++++++++++- src/mcp/schemas_tests.rs | 29 +++++ src/receiver/writer_tests.rs | 6 +- src/scanner.rs | 32 ++++-- src/scanner/checkpoint.rs | 32 +++++- src/scanner/checkpoint_tests.rs | 61 +++++++++++ src/scanner/claude.rs | 3 +- src/scanner/codex_tests.rs | 1 + src/scanner/skill_events.rs | 129 +++++++++++++++++++---- src/scanner/skill_events_tests.rs | 60 +++++++++++ src/scanner_tests.rs | 31 ++++++ 28 files changed, 823 insertions(+), 86 deletions(-) diff --git a/README.md b/README.md index 4900cb4c7..f8d666343 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 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,7 +685,7 @@ 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 @@ -693,7 +693,7 @@ Cortex uses SQLite with: - 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 diff --git a/docs/architecture.md b/docs/architecture.md index a452a83f4..f9b3c9adf 100644 --- a/docs/architecture.md +++ b/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 | diff --git a/packages/cortex-rmcp/README.md b/packages/cortex-rmcp/README.md index 4900cb4c7..f8d666343 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 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,7 +685,7 @@ 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 @@ -693,7 +693,7 @@ Cortex uses SQLite with: - 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 diff --git a/src/app/services/skill_backfill.rs b/src/app/services/skill_backfill.rs index ce5ebd81b..d971de91d 100644 --- a/src/app/services/skill_backfill.rs +++ b/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::(line_text) { diff --git a/src/app/services/skill_backfill_tests.rs b/src/app/services/skill_backfill_tests.rs index 845121344..1ee3bef4c 100644 --- a/src/app/services/skill_backfill_tests.rs +++ b/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":"vibin:repo-status /vibin:repo-status"}]}}"#, + ); + + 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) = 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() { diff --git a/src/app/skill_signal_detectors.rs b/src/app/skill_signal_detectors.rs index 7d35efdc3..92ea145f0 100644 --- a/src/app/skill_signal_detectors.rs +++ b/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::(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) diff --git a/src/cli/complete_tests.rs b/src/cli/complete_tests.rs index f3c25f63e..36e299260 100644 --- a/src/cli/complete_tests.rs +++ b/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", diff --git a/src/db/maintenance.rs b/src/db/maintenance.rs index 7c7f367e3..ec2a0fcd3 100644 --- a/src/db/maintenance.rs +++ b/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; + } } } } diff --git a/src/db/pool.rs b/src/db/pool.rs index 01d2a6fa4..5f644ea19 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 = 58; +pub const KNOWN_SCHEMA_VERSION: i64 = 60; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct SchemaVersionInfo { @@ -396,6 +396,15 @@ pub fn init_pool(config: &StorageConfig) -> Result { tokenize='porter unicode61' ); + -- Transcript-only FTS keeps AI/session searches out of the much larger + -- syslog/Docker corpus while preserving the same message tokenizer. + CREATE VIRTUAL TABLE IF NOT EXISTS ai_logs_fts USING fts5( + message, + content='logs', + content_rowid='id', + tokenize='porter unicode61' + ); + -- Trigger to keep FTS in sync on INSERT only. -- DELETE and UPDATE triggers are intentionally absent: bulk DELETEs during -- retention purge and storage-budget enforcement fire the trigger for every @@ -407,6 +416,20 @@ pub fn init_pool(config: &StorageConfig) -> Result { INSERT INTO logs_fts(rowid, message) VALUES (new.id, new.message); END; + 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); + END; + + -- Unlike the global FTS index, transcript volume is small enough to + -- keep deletions exact. This also makes extractor-revision replay remove + -- 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); + END; + -- Hostname registry for quick lookups CREATE TABLE IF NOT EXISTS hosts ( hostname TEXT PRIMARY KEY, @@ -2379,6 +2402,10 @@ pub fn init_pool(config: &StorageConfig) -> Result { CREATE INDEX IF NOT EXISTS idx_ai_mcp_events_error_time ON ai_mcp_events(is_error, timestamp) WHERE is_error = 1; + CREATE INDEX IF NOT EXISTS idx_ai_mcp_events_call_log_id + ON ai_mcp_events(call_log_id) WHERE call_log_id IS NOT NULL; + CREATE INDEX IF NOT EXISTS idx_ai_mcp_events_result_log_id + ON ai_mcp_events(result_log_id) WHERE result_log_id IS NOT NULL; INSERT OR IGNORE INTO schema_migrations (version) VALUES (39); COMMIT;", @@ -2458,6 +2485,8 @@ pub fn init_pool(config: &StorageConfig) -> Result { ON ai_hook_events(ai_tool, ai_project, ai_session_id, timestamp); CREATE INDEX IF NOT EXISTS idx_ai_hook_events_evidence_time ON ai_hook_events(evidence_kind, timestamp); + CREATE INDEX IF NOT EXISTS idx_ai_hook_events_log_id + ON ai_hook_events(log_id) WHERE log_id IS NOT NULL; INSERT OR IGNORE INTO schema_migrations (version) VALUES (40); COMMIT;", @@ -3476,6 +3505,85 @@ pub fn init_pool(config: &StorageConfig) -> Result { tracing::info!("Migration 58: indexed recurring-error graph evidence lookup"); } + // Migration 59: transcript extraction semantics now persist speaker roles + // and recognize modern Claude skill command envelopes. Existing completed + // transcript checkpoints are revision 0 and will be atomically reset and + // replayed on their next scan. + if !migration_applied(&conn, 59)? { + let tx = conn.transaction()?; + if table_exists(&tx, "transcript_sources")? { + add_column_if_missing( + &tx, + "transcript_sources", + "extractor_revision", + "INTEGER NOT NULL DEFAULT 0", + )?; + } + tx.execute( + "INSERT OR IGNORE INTO schema_migrations (version) VALUES (59)", + [], + )?; + tx.commit()?; + tracing::info!("Migration 59: versioned transcript extraction checkpoints"); + } + + // Migration 60: AI/session full-text search no longer competes with the + // global syslog/Docker FTS corpus. Rebuild this derived index from only + // transcript rows, then keep it current with transcript INSERT/DELETE triggers. If an + // interrupted upgrade left the table without the migration marker, drop and + // rebuild it deterministically before marking the migration complete. + if !migration_applied(&conn, 60)? { + tracing::info!( + "Migration 60: building transcript-only FTS index; this may take time on large transcript histories" + ); + let started = std::time::Instant::now(); + let mcp_indexes = if table_exists(&conn, "ai_mcp_events")? { + "CREATE INDEX IF NOT EXISTS idx_ai_mcp_events_call_log_id + ON ai_mcp_events(call_log_id) WHERE call_log_id IS NOT NULL; + CREATE INDEX IF NOT EXISTS idx_ai_mcp_events_result_log_id + ON ai_mcp_events(result_log_id) WHERE result_log_id IS NOT NULL;" + } else { + "" + }; + let hook_index = if table_exists(&conn, "ai_hook_events")? { + "CREATE INDEX IF NOT EXISTS idx_ai_hook_events_log_id + ON ai_hook_events(log_id) WHERE log_id IS NOT NULL;" + } else { + "" + }; + conn.execute_batch(&format!( + "BEGIN IMMEDIATE; + 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, + content='logs', + content_rowid='id', + tokenize='porter unicode61' + ); + INSERT INTO ai_logs_fts(rowid, message) + SELECT id, message 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) VALUES (new.id, new.message); + 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) + VALUES ('delete', old.id, old.message); + END; + {mcp_indexes} + {hook_index} + INSERT OR IGNORE INTO schema_migrations (version) VALUES (60); + COMMIT;", + ))?; + tracing::info!( + elapsed_ms = started.elapsed().as_millis(), + "Migration 60: transcript-only 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 29d2bc576..df36e681d 100644 --- a/src/db/pool_tests.rs +++ b/src/db/pool_tests.rs @@ -10,9 +10,124 @@ use rusqlite::OptionalExtension; #[test] fn documented_schema_count_matches_known_version() { - assert_eq!(KNOWN_SCHEMA_VERSION, 58); - assert!(include_str!("../../README.md").contains("58 sequential schema migrations")); - assert!(include_str!("../../docs/architecture.md").contains("58 sequential migrations")); + 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")); +} + +#[test] +fn migration_59_versions_transcript_extractor_checkpoints() { + let dir = tempfile::tempdir().unwrap(); + let pool = init_pool(&test_storage_config(dir.path().join("migration-59.db"))).unwrap(); + let conn = pool.get().unwrap(); + let columns: Vec = conn + .prepare("PRAGMA table_info(transcript_sources)") + .unwrap() + .query_map([], |row| row.get(1)) + .unwrap() + .collect::>() + .unwrap(); + assert!(columns.iter().any(|column| column == "extractor_revision")); + let default_revision: i64 = conn + .query_row( + "SELECT dflt_value FROM pragma_table_info('transcript_sources') WHERE name='extractor_revision'", + [], + |row| { + let raw: String = row.get(0)?; + Ok(raw.trim_matches('\'').parse::().unwrap()) + }, + ) + .unwrap(); + assert_eq!(default_revision, 0); + let marker_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM schema_migrations WHERE version = 59", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(marker_count, 1); +} + +#[test] +fn migration_60_creates_transcript_only_fts_and_insert_trigger() { + let dir = tempfile::tempdir().unwrap(); + let pool = init_pool(&test_storage_config(dir.path().join("migration-60.db"))).unwrap(); + + let mut ai = migration_test_log(); + ai.message = "transcriptneedle".into(); + ai.raw = ai.message.clone(); + let mut ordinary = migration_test_log(); + ordinary.message = "systemneedle".into(); + ordinary.raw = ordinary.message.clone(); + ordinary.ai_tool = None; + ordinary.ai_project = None; + ordinary.ai_session_id = None; + insert_logs_batch(&pool, &[ai, ordinary]).unwrap(); + + let conn = pool.get().unwrap(); + let transcript_matches: i64 = conn + .query_row( + "SELECT COUNT(*) FROM ai_logs_fts WHERE ai_logs_fts MATCH 'transcriptneedle'", + [], + |row| row.get(0), + ) + .unwrap(); + let system_matches: i64 = conn + .query_row( + "SELECT COUNT(*) FROM ai_logs_fts WHERE ai_logs_fts MATCH 'systemneedle'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(transcript_matches, 1); + assert_eq!(system_matches, 0); + + let ai_log_id: i64 = conn + .query_row( + "SELECT id FROM logs WHERE message = 'transcriptneedle'", + [], + |row| row.get(0), + ) + .unwrap(); + conn.execute("DELETE FROM logs WHERE id = ?1", [ai_log_id]) + .unwrap(); + let after_delete: i64 = conn + .query_row( + "SELECT COUNT(*) FROM ai_logs_fts WHERE ai_logs_fts MATCH 'transcriptneedle'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!( + after_delete, 0, + "AI FTS must remove deleted transcript rows" + ); + + let marker_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM schema_migrations WHERE version = 60", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(marker_count, 1); + + let indexes = conn + .prepare("SELECT name FROM sqlite_master WHERE type = 'index' AND name IN ('idx_ai_mcp_events_call_log_id', 'idx_ai_mcp_events_result_log_id', 'idx_ai_hook_events_log_id') ORDER BY name") + .unwrap() + .query_map([], |row| row.get::<_, String>(0)) + .unwrap() + .collect::>>() + .unwrap(); + assert_eq!( + indexes, + vec![ + "idx_ai_hook_events_log_id".to_string(), + "idx_ai_mcp_events_call_log_id".to_string(), + "idx_ai_mcp_events_result_log_id".to_string(), + ] + ); } #[test] diff --git a/src/db/queries.rs b/src/db/queries.rs index f49d7ad55..b40ee6b71 100644 --- a/src/db/queries.rs +++ b/src/db/queries.rs @@ -1408,10 +1408,10 @@ fn search_ai_sessions_sql( l.hostname, l.timestamp, l.message - FROM logs_fts - JOIN logs l ON l.id = logs_fts.rowid - WHERE logs_fts MATCH ?1{filters} - ORDER BY logs_fts.rowid DESC + FROM ai_logs_fts + JOIN logs l ON l.id = ai_logs_fts.rowid + WHERE ai_logs_fts MATCH ?1{filters} + ORDER BY ai_logs_fts.rowid DESC LIMIT {} ), bounded_candidates AS MATERIALIZED ( diff --git a/src/db/queries_tests.rs b/src/db/queries_tests.rs index 29347e96e..ba727198d 100644 --- a/src/db/queries_tests.rs +++ b/src/db/queries_tests.rs @@ -1166,8 +1166,15 @@ fn search_ai_sessions_query_plan_uses_session_host_time_index() { let (sql, bindings) = search_ai_sessions_sql(¶ms, 10); assert!(sql.contains("candidates AS MATERIALIZED")); assert!( - sql.contains("FROM logs_fts") && sql.contains("WHERE logs_fts MATCH ?1"), - "session search candidates must be FTS-first" + !sql.contains( + "FROM logs_fts +" + ), + "session search must not scan the global syslog FTS index" + ); + assert!( + sql.contains("FROM ai_logs_fts") && sql.contains("WHERE ai_logs_fts MATCH ?1"), + "session search candidates must use the transcript-only FTS index" ); assert!( sql.contains("LIMIT 5000"), diff --git a/src/db/skill_incident_evidence.rs b/src/db/skill_incident_evidence.rs index d4536d50c..a99af4d85 100644 --- a/src/db/skill_incident_evidence.rs +++ b/src/db/skill_incident_evidence.rs @@ -10,7 +10,9 @@ use anyhow::Result; use serde::{Deserialize, Serialize}; -use crate::app::skill_signal_detectors::{detect_tool_failure, detect_user_correction}; +use crate::app::skill_signal_detectors::{ + detect_tool_failure, detect_user_correction, transcript_event_is_user, +}; use super::models::LogEntry; use super::pool::DbPool; @@ -309,7 +311,10 @@ pub fn investigate_ai_skill_incidents( let mut nearby_user_corrections: Vec = nearby_logs .iter() - .filter(|e| detect_user_correction(&e.message)) + .filter(|e| { + transcript_event_is_user(e.metadata_json.as_deref()) + && detect_user_correction(&e.message) + }) .cloned() .collect(); let nearby_user_corrections_truncated = nearby_user_corrections.len() > NEARBY_SUBSET_CAP; diff --git a/src/db/skill_incident_evidence_tests.rs b/src/db/skill_incident_evidence_tests.rs index 488edcceb..b6d9069f1 100644 --- a/src/db/skill_incident_evidence_tests.rs +++ b/src/db/skill_incident_evidence_tests.rs @@ -34,7 +34,7 @@ fn make_ai_entry( ai_project: Some(project.to_string()), ai_session_id: Some(session_id.to_string()), ai_transcript_path: Some(format!("{project}/{session_id}.jsonl")), - metadata_json: None, + metadata_json: Some(serde_json::json!({"event_kind": "user"}).to_string()), http_status: None, auth_outcome: None, dns_blocked: None, diff --git a/src/db/skill_incidents.rs b/src/db/skill_incidents.rs index 6141bfc48..d40aaec4b 100644 --- a/src/db/skill_incidents.rs +++ b/src/db/skill_incidents.rs @@ -15,7 +15,7 @@ use std::collections::HashMap; use crate::app::skill_signal_detectors::{ detect_ignored_instruction, detect_overlong_loop, detect_scope_or_source_confusion, - detect_tool_failure, detect_user_correction, + detect_tool_failure, detect_user_correction, transcript_event_is_user, }; use super::pool::DbPool; @@ -252,16 +252,16 @@ pub fn search_ai_skill_incidents( .unwrap_or_else(|_| last_seen.clone()); let mut anchor_stmt = conn.prepare( - "SELECT id, message FROM logs + "SELECT id, message, metadata_json FROM logs WHERE ai_session_id = ?1 AND ai_project = ?2 AND ai_tool = ?3 AND timestamp >= ?4 AND timestamp <= ?5 ORDER BY timestamp ASC LIMIT 500", )?; - let anchor_rows: Vec<(i64, String)> = anchor_stmt + let anchor_rows: Vec<(i64, String, Option)> = anchor_stmt .query_map( rusqlite::params![session_id, project, tool, win_from, win_to], - |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)), + |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?, row.get(2)?)), )? .collect::>>()?; @@ -270,9 +270,10 @@ pub fn search_ai_skill_incidents( let tool_call_rows = anchor_rows.len(); let mut has_correction_or_frustration = false; - for (id, message) in &anchor_rows { + for (id, message, metadata_json) in &anchor_rows { let mut hit = false; - if detect_user_correction(message) { + if transcript_event_is_user(metadata_json.as_deref()) && detect_user_correction(message) + { counts.user_correction_after_skill += 1; has_correction_or_frustration = true; hit = true; diff --git a/src/db/skill_incidents_tests.rs b/src/db/skill_incidents_tests.rs index 348800b6a..4b2fe7e50 100644 --- a/src/db/skill_incidents_tests.rs +++ b/src/db/skill_incidents_tests.rs @@ -33,7 +33,7 @@ fn make_ai_entry( ai_project: Some(project.to_string()), ai_session_id: Some(session_id.to_string()), ai_transcript_path: Some(format!("{project}/{session_id}.jsonl")), - metadata_json: None, + metadata_json: Some(serde_json::json!({"event_kind": "user"}).to_string()), http_status: None, auth_outcome: None, dns_blocked: None, @@ -155,6 +155,68 @@ fn search_ai_skill_incidents_groups_by_skill_session_window_and_scores() { assert!(incident.incident_id.starts_with("skill-inc-")); } +#[test] +fn assistant_acknowledgment_is_not_counted_as_user_correction() { + let (pool, _dir) = test_pool(); + let skill_log = make_ai_entry( + "2026-01-01T00:00:00Z", + "devhost", + "codex", + "/tmp/project-role", + "sess-role", + "loaded skill review-skill", + ); + let mut assistant = make_ai_entry( + "2026-01-01T00:01:00Z", + "devhost", + "codex", + "/tmp/project-role", + "sess-role", + "You're right, that is wrong and I should correct it.", + ); + assistant.metadata_json = Some(serde_json::json!({"event_kind": "assistant"}).to_string()); + insert_logs_batch(&pool, &[skill_log, assistant]).unwrap(); + let conn = pool.get().unwrap(); + let log_id: i64 = conn + .query_row("SELECT id FROM logs ORDER BY id LIMIT 1", [], |row| { + row.get(0) + }) + .unwrap(); + drop(conn); + insert_skill_event( + &pool, + log_id, + "codex", + "/tmp/project-role", + "sess-role", + "devhost", + "2026-01-01T00:00:00Z", + "review-skill", + None, + ); + let result = search_ai_skill_incidents( + &pool, + &AiSkillIncidentParams { + skill: Some("review-skill".into()), + ..Default::default() + }, + ) + .unwrap(); + assert_eq!(result.incidents.len(), 1); + assert_eq!( + result.incidents[0] + .signal_counts + .user_correction_after_skill, + 0 + ); + assert!( + !result.incidents[0] + .signals_present + .iter() + .any(|signal| signal == "user_correction_after_skill") + ); +} + #[test] fn search_ai_skill_incidents_sorts_by_score_with_total_cmp() { let (pool, _dir) = test_pool(); diff --git a/src/mcp/action_flags.rs b/src/mcp/action_flags.rs index d00400b3f..ab2d9f006 100644 --- a/src/mcp/action_flags.rs +++ b/src/mcp/action_flags.rs @@ -374,6 +374,7 @@ pub(super) const SKILL_INCIDENT_FLAGS: &[FlagSpec] = &[ pub(super) const SKILL_INVESTIGATE_FLAGS: &[FlagSpec] = &[ AI_INCIDENT_ID_FLAG, + AI_SKILL_FLAG, AI_PLUGIN_FLAG, AI_TOOL_FLAG, AI_PROJECT_FLAG, diff --git a/src/mcp/schemas.rs b/src/mcp/schemas.rs index 6b7ebb373..f69309fdf 100644 --- a/src/mcp/schemas.rs +++ b/src/mcp/schemas.rs @@ -127,7 +127,64 @@ pub(super) fn tool_definitions() -> Vec { "tool": { "type": "string", "enum": ["claude", "codex", "gemini"], - "description": "For action=filter, sessions, search_sessions, abuse, ai_correlate, usage_blocks, project_context, or list_ai_projects: AI tool filter." + "description": "AI tool filter for transcript/session correlation and skill/MCP/hook event or incident actions." + }, + "skill": { + "type": "string", + "description": "For skill_events, skill_incidents, or skill_investigate: exact skill name filter." + }, + "plugin": { + "type": "string", + "description": "For skill_events, skill_incidents, or skill_investigate: exact plugin name filter." + }, + "tool_name": { + "type": "string", + "description": "For mcp_events, mcp_incidents, or mcp_investigate: raw MCP tool-call name filter." + }, + "mcp_server": { + "type": "string", + "description": "For mcp_events, mcp_incidents, or mcp_investigate: MCP server filter." + }, + "mcp_tool": { + "type": "string", + "description": "For mcp_events, mcp_incidents, or mcp_investigate: MCP tool filter." + }, + "hook_event": { + "type": "string", + "description": "For hook_events, hook_incidents, or hook_investigate: hook event filter." + }, + "hook_name": { + "type": "string", + "description": "For hook_events, hook_incidents, or hook_investigate: hook name filter." + }, + "hook_source": { + "type": "string", + "description": "For hook_events, hook_incidents, or hook_investigate: hook source filter." + }, + "evidence_kind": { + "type": "string", + "description": "For hook_events or hook_incidents: exact evidence-kind filter." + }, + "incident_id": { + "type": "string", + "description": "For skill_investigate, mcp_investigate, or hook_investigate: exact incident id." + }, + "signals": { + "type": "array", + "items": {"type": "string"}, + "description": "For skill_incidents, mcp_incidents, or hook_incidents: require one or more signal types." + }, + "min_score": { + "type": "number", + "description": "For skill_incidents, mcp_incidents, or hook_incidents: minimum incident priority score." + }, + "is_error": { + "type": "boolean", + "description": "For mcp_events: when true, return only failed MCP calls." + }, + "hostname": { + "type": "string", + "description": "Hostname filter for skill, MCP, or hook event/incident actions." }, "source": { "type": "string", @@ -170,7 +227,7 @@ pub(super) fn tool_definitions() -> Vec { }, "session_id": { "type": "string", - "description": "For action=filter or ai_correlate: exact AI session id filter." + "description": "Exact AI session id filter for transcript correlation and skill/MCP/hook event or incident actions." }, "branch": {"type":"string","maxLength":512,"description":"For action=evidence_scope: exact Git branch projected by Agent Observatory."}, "worktree": {"type":"string","maxLength":4096,"description":"For action=evidence_scope: exact absolute worktree path projected by Agent Observatory."}, @@ -290,7 +347,7 @@ pub(super) fn tool_definitions() -> Vec { }, "correlation_window_minutes": { "type": "integer", - "description": "For action=abuse_investigate: minutes before first and after last anchor for nearby non-AI log correlation, default 5, max 120." + "description": "Evidence-correlation window for abuse_investigate, skill_investigate, mcp_investigate, or hook_investigate; default 5, max 120." }, "group_by": { "type": "string", diff --git a/src/mcp/schemas_tests.rs b/src/mcp/schemas_tests.rs index b1d20e98d..f7cca4239 100644 --- a/src/mcp/schemas_tests.rs +++ b/src/mcp/schemas_tests.rs @@ -191,6 +191,35 @@ fn event_actions_advertise_canonical_mcp_time_bounds() { } } +#[test] +fn incident_and_event_actions_expose_specialized_filters() { + let properties = &tool_definitions()[0]["inputSchema"]["properties"]; + for field in [ + "skill", + "plugin", + "tool_name", + "mcp_server", + "mcp_tool", + "hook_event", + "hook_name", + "hook_source", + "evidence_kind", + "incident_id", + "signals", + "min_score", + "is_error", + "hostname", + ] { + assert!( + properties.get(field).is_some(), + "missing MCP schema field {field}" + ); + } + assert_eq!(properties["signals"]["type"], "array"); + assert_eq!(properties["min_score"]["type"], "number"); + assert_eq!(properties["is_error"]["type"], "boolean"); +} + #[test] fn tool_definition_exposes_agent_cost_metadata() { let tools = tool_definitions(); diff --git a/src/receiver/writer_tests.rs b/src/receiver/writer_tests.rs index a478be70c..bb860b99e 100644 --- a/src/receiver/writer_tests.rs +++ b/src/receiver/writer_tests.rs @@ -479,7 +479,11 @@ async fn shutdown_signal_flushes_queued_rows_before_long_deadline() { shutdown_rx, )); shutdown_tx.send(true).unwrap(); - tokio::time::timeout(std::time::Duration::from_secs(2), writer) + // Keep this bound orders of magnitude below the 3,600s flush interval, but + // leave enough scheduler/SQLite headroom for the full 3k-test suite running + // concurrently on loaded CI hosts. The isolated path normally completes in + // well under a second; this asserts prompt shutdown, not machine idleness. + tokio::time::timeout(std::time::Duration::from_secs(10), writer) .await .expect("shutdown must not wait for the flush deadline") .unwrap(); diff --git a/src/scanner.rs b/src/scanner.rs index bb077f60b..68cb6b7e1 100644 --- a/src/scanner.rs +++ b/src/scanner.rs @@ -26,7 +26,8 @@ use crate::scanner::mcp_events::{ extract_antigravity_mcp_events, extract_claude_mcp_events, extract_codex_mcp_events, }; 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, }; pub(crate) mod antigravity; @@ -50,6 +51,7 @@ const MAX_INDEX_CHUNK_RECORDS: usize = 500; const MAX_INDEX_CHUNK_BYTES: usize = 4 * 1024 * 1024; const MAX_AI_PROJECT_CHARS: usize = 512; const MAX_AI_SESSION_ID_CHARS: usize = 128; +pub(crate) const TRANSCRIPT_EXTRACTOR_REVISION: i64 = 2; const MAX_TRANSCRIPT_PATH_CHARS: usize = 1024; const MAX_SESSION_METADATA_CHARS: usize = 512; const MAX_ABANDONED_SNAPSHOT_WORKERS: usize = 2; @@ -936,6 +938,15 @@ pub fn index_file_with_options( } else { None }; + if stored_metadata.as_ref().is_some_and(|metadata| { + metadata.file_size.is_some() && metadata.extractor_revision != TRANSCRIPT_EXTRACTOR_REVISION + }) { + // Extractor semantics changed. Preserve neither old import receipts nor + // derived log/event rows: reset atomically and replay this source using + // the current extractor revision. + checkpoint_store.reset_source(source_id, &canonical)?; + stored_metadata = None; + } let bounded_stream = options.scan_budget.is_some_and(|budget| { snapshot.is_none() && current_metadata.size > budget.per_source_max_bytes }); @@ -1237,7 +1248,7 @@ pub fn index_file_with_options( // line as `None`. match &parsed.raw_value { Some(value) - if line_text.contains("attributionSkill") + if claude_line_may_contain_skill_event(line_text) || line_text.contains("hook_") => { ChunkSkillSource::Claude(value.clone()) @@ -2712,13 +2723,22 @@ impl TranscriptSessionMetadata { pub(crate) fn transcript_event_kind(value: &serde_json::Value) -> String { let payload = value.get("payload").unwrap_or(value); - let kind = payload + let payload_type = payload .get("type") .or_else(|| value.get("type")) - .or_else(|| payload.get("role")) + .and_then(serde_json::Value::as_str); + let role = payload + .get("role") .or_else(|| value.get("role")) - .and_then(serde_json::Value::as_str) - .unwrap_or("unknown"); + .and_then(serde_json::Value::as_str); + // Codex wraps conversational messages as payload.type="message" and stores + // the actual speaker in payload.role. Prefer that role or every message + // becomes "unknown", which poisons speaker-sensitive incident signals. + let kind = if payload_type == Some("message") { + role.or(payload_type).unwrap_or("unknown") + } else { + payload_type.or(role).unwrap_or("unknown") + }; match kind { "user" | "human" => "user", "assistant" => "assistant", diff --git a/src/scanner/checkpoint.rs b/src/scanner/checkpoint.rs index c1f12c43d..290605abb 100644 --- a/src/scanner/checkpoint.rs +++ b/src/scanner/checkpoint.rs @@ -27,6 +27,7 @@ pub struct SourceMetadata { pub source_revision: Option, /// The recovery point represented by `last_offset`. pub scan_state: SourceScanState, + pub extractor_revision: i64, } /// The only scanner recovery states persisted for a transcript source. @@ -134,9 +135,9 @@ impl<'a> CheckpointStore<'a> { file_mtime: Option, ) -> Result { let conn = self.pool.get()?; - let Some((stored_size, stored_mtime, last_error, scan_state)) = conn + let Some((stored_size, stored_mtime, last_error, scan_state, extractor_revision)) = conn .query_row( - "SELECT file_size, file_mtime, last_error, scan_state + "SELECT file_size, file_mtime, last_error, scan_state, extractor_revision FROM transcript_sources WHERE id = ?1", [source_id], @@ -146,6 +147,7 @@ impl<'a> CheckpointStore<'a> { row.get::<_, Option>(1)?, row.get::<_, Option>(2)?, row.get::<_, Option>(3)?, + row.get::<_, i64>(4)?, )) }, ) @@ -157,14 +159,15 @@ impl<'a> CheckpointStore<'a> { Ok(last_error.is_none() && SourceScanState::from_db_value(scan_state) == SourceScanState::Complete && stored_size == Some(file_size as i64) - && stored_mtime == file_mtime) + && stored_mtime == file_mtime + && extractor_revision == crate::scanner::TRANSCRIPT_EXTRACTOR_REVISION) } pub fn source_metadata(&self, source_id: i64) -> Result> { let conn = self.pool.get()?; conn.query_row( "SELECT file_size, file_mtime, content_hash, last_offset, last_error, - source_revision, scan_state + source_revision, scan_state, extractor_revision FROM transcript_sources WHERE id = ?1", [source_id], @@ -177,6 +180,7 @@ impl<'a> CheckpointStore<'a> { last_error: row.get(4)?, source_revision: row.get(5)?, scan_state: SourceScanState::from_db_value(row.get(6)?), + extractor_revision: row.get(7)?, }) }, ) @@ -195,6 +199,21 @@ impl<'a> CheckpointStore<'a> { "DELETE FROM transcript_parse_errors WHERE source_id = ?1", [source_id], )?; + // Source replay replaces every transcript-derived projection. Some MCP + // result-only and hook rows intentionally use SET NULL for ordinary log + // retention, so remove those projections explicitly before deleting the + // source logs or stale events could survive a parser/extractor replay. + tx.execute( + "DELETE FROM ai_mcp_events + WHERE call_log_id IN (SELECT id FROM logs WHERE ai_transcript_path = ?1) + OR result_log_id IN (SELECT id FROM logs WHERE ai_transcript_path = ?1)", + [canonical_path], + )?; + tx.execute( + "DELETE FROM ai_hook_events + WHERE log_id IN (SELECT id FROM logs WHERE ai_transcript_path = ?1)", + [canonical_path], + )?; tx.execute( "DELETE FROM logs WHERE ai_transcript_path = ?1", [canonical_path], @@ -219,6 +238,7 @@ impl<'a> CheckpointStore<'a> { last_offset = 0, source_revision = NULL, scan_state = 'restart', + extractor_revision = 0, last_indexed_at = NULL, last_error = NULL WHERE id = ?1", @@ -655,6 +675,7 @@ pub fn update_complete_source_metadata_in_tx( last_offset = ?5, source_revision = ?6, scan_state = 'complete', + extractor_revision = ?7, last_indexed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'), last_error = NULL WHERE id = ?1", @@ -665,6 +686,7 @@ pub fn update_complete_source_metadata_in_tx( file_metadata.content_hash, file_metadata.size as i64, source_revision, + crate::scanner::TRANSCRIPT_EXTRACTOR_REVISION, ], )?; Ok(()) @@ -695,6 +717,7 @@ pub fn update_partial_source_metadata_in_tx( last_offset = ?5, source_revision = ?6, scan_state = ?7, + extractor_revision = ?8, last_indexed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now'), last_error = NULL WHERE id = ?1", @@ -706,6 +729,7 @@ pub fn update_partial_source_metadata_in_tx( last_offset, source_revision, scan_state.as_db_value(), + crate::scanner::TRANSCRIPT_EXTRACTOR_REVISION, ], )?; Ok(()) diff --git a/src/scanner/checkpoint_tests.rs b/src/scanner/checkpoint_tests.rs index 64585aff2..89712aef0 100644 --- a/src/scanner/checkpoint_tests.rs +++ b/src/scanner/checkpoint_tests.rs @@ -38,10 +38,70 @@ fn ensure_source_reuses_existing_source_id() { last_error: None, source_revision: None, scan_state: SourceScanState::Complete, + extractor_revision: 0, } ); } +#[test] +fn reset_source_removes_all_transcript_derived_projections() { + let (pool, _dir) = test_pool(); + let store = CheckpointStore::new(&pool); + let path = "/tmp/replay.jsonl"; + let source_id = store.ensure_source(path, "codex_session").unwrap(); + let conn = pool.get().unwrap(); + 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-01-01T00:00:00Z', 'host-a', 'info', 'transcript row', 'transcript row', 'transcript://codex_session', 'codex', '/tmp/project', 'sess-1', ?1, '{\"event_kind\":\"user\"}')", + [path], + ).unwrap(); + let log_id = conn.last_insert_rowid(); + conn.execute( + "INSERT INTO ai_skill_events (log_id, ai_tool, hostname, timestamp, skill_name, event_kind, evidence_kind) + VALUES (?1, 'codex', 'host-a', '2026-01-01T00:00:00Z', 'review-skill', 'codex_skill_block', 'transcript_content')", + [log_id], + ).unwrap(); + conn.execute( + "INSERT INTO ai_mcp_events (call_log_id, result_log_id, ai_tool, hostname, timestamp, call_id, tool_name, event_kind) + VALUES (NULL, ?1, 'codex', 'host-a', '2026-01-01T00:00:00Z', 'call-1', 'tool-result', 'result')", + [log_id], + ).unwrap(); + conn.execute( + "INSERT INTO ai_hook_events (log_id, ai_tool, hostname, timestamp, hook_event, status, evidence_kind) + VALUES (?1, 'codex', 'host-a', '2026-01-01T00:00:00Z', 'PostToolUse', 'ok', 'runtime_transcript')", + [log_id], + ).unwrap(); + conn.execute( + "INSERT INTO transcript_import_records (source_id, record_key) VALUES (?1, 'record-1')", + [source_id], + ) + .unwrap(); + drop(conn); + + store.reset_source(source_id, path).unwrap(); + + let conn = pool.get().unwrap(); + for table in [ + "logs", + "ai_skill_events", + "ai_mcp_events", + "ai_hook_events", + "transcript_import_records", + ] { + let count: i64 = conn + .query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(count, 0, "{table} must be replaced during source replay"); + } + drop(conn); + let metadata = store.source_metadata(source_id).unwrap().unwrap(); + assert_eq!(metadata.scan_state, SourceScanState::Restart); + assert_eq!(metadata.extractor_revision, 0); + assert_eq!(metadata.file_size, None); +} + #[test] fn ensure_source_updates_source_kind_for_existing_path() { let (pool, _dir) = test_pool(); @@ -283,6 +343,7 @@ fn partial_checkpoint_persists_full_source_metadata_and_recovery_state() { last_error: None, source_revision: Some("revision-1".to_string()), scan_state: SourceScanState::Boundary, + extractor_revision: crate::scanner::TRANSCRIPT_EXTRACTOR_REVISION, }) ); diff --git a/src/scanner/claude.rs b/src/scanner/claude.rs index 2dd854ecc..5c8bf2a74 100644 --- a/src/scanner/claude.rs +++ b/src/scanner/claude.rs @@ -15,7 +15,8 @@ pub fn parse_line( let value: Value = serde_json::from_str(line)?; let message = extract_message(&value); let mut session_metadata = extract_session_metadata(&value); - let has_structured_evidence = line.contains("attributionSkill") || line.contains("hook_"); + let has_structured_evidence = + super::skill_events::claude_line_may_contain_skill_event(line) || line.contains("hook_"); if message.is_empty() && session_metadata == TranscriptSessionMetadata::default() && !has_structured_evidence diff --git a/src/scanner/codex_tests.rs b/src/scanner/codex_tests.rs index 7a0d5ff6e..ed2f4a465 100644 --- a/src/scanner/codex_tests.rs +++ b/src/scanner/codex_tests.rs @@ -132,6 +132,7 @@ fn codex_parser_does_not_invent_a_title_from_transcript_content() { .unwrap(); assert!(parsed.session_metadata.title.is_none()); assert!(parsed.session_metadata.title_provenance.is_none()); + assert_eq!(parsed.event_kind, "user"); } #[test] diff --git a/src/scanner/skill_events.rs b/src/scanner/skill_events.rs index b30ba8069..12f461086 100644 --- a/src/scanner/skill_events.rs +++ b/src/scanner/skill_events.rs @@ -2,9 +2,9 @@ //! //! Two independent extractors feed the same [`ExtractedSkillEvent`] shape: //! - Claude: structured `attributionSkill` / `attributionPlugin` JSON fields -//! (top-level or `message.*` nesting — a third `payload.*` candidate was -//! deliberately NOT added: no observed transcript sample confirms that -//! shape, so it would be speculative). +//! plus observed package-qualified skill command envelopes such as +//! `vibin:repo-status` and +//! `/vibin:repo-status`. //! - Codex: native skill headers in transcript content, including truncated //! bodies, and separately typed successful command-read evidence. A JSON //! marker in ordinary message text is never command-execution evidence. @@ -29,6 +29,7 @@ const MAX_SKILL_FIELD_CHARS: usize = 256; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum SkillEventKind { ClaudeAttribution, + ClaudeSkillCommand, CodexSkillBlock, CodexSkillRead, } @@ -37,6 +38,7 @@ impl SkillEventKind { pub fn as_str(self) -> &'static str { match self { Self::ClaudeAttribution => "claude_attribution", + Self::ClaudeSkillCommand => "claude_skill_command", Self::CodexSkillBlock => "codex_skill_block", Self::CodexSkillRead => "codex_skill_read", } @@ -115,20 +117,79 @@ fn clamp_chars(value: &str, max_chars: usize) -> String { } } -/// Extract Claude skill-attribution events from a raw transcript JSON value. -/// Checks top-level and `message.*` nesting for `attributionSkill` / -/// `attributionPlugin` string fields (Claude transcripts use flat top-level -/// fields on user-facing records and nested `message.*` fields on some -/// tool-result records). Returns one event per candidate location that has a -/// non-empty `attributionSkill`; at most one event in practice since a single -/// transcript line only has one of the two shapes. +/// Cheap raw-line guard shared by live ingest and historical backfill. +/// Modern Claude desktop sessions do not consistently emit `attributionSkill`; +/// package-qualified skill invocations are instead represented by command +/// envelopes in the user message. +pub(crate) fn claude_line_may_contain_skill_event(line: &str) -> bool { + line.contains("attributionSkill") + || line.contains("") + || line.contains("") +} + +static CLAUDE_SKILL_COMMAND_TAG: LazyLock = LazyLock::new(|| { + // Only package-qualified commands are strong enough evidence to call a + // Claude command a skill invocation. Unqualified slash commands can be + // ordinary built-ins or plugin commands and remain intentionally ignored. + Regex::new( + r"(?s)\s*/?([A-Za-z0-9_.-]+:[A-Za-z0-9_.:-]+)\s*|\s*/?([A-Za-z0-9_.-]+:[A-Za-z0-9_.:-]+)\s*", + ) + .expect("static regex") +}); + +fn collect_claude_text(value: &serde_json::Value, out: &mut Vec) { + match value { + serde_json::Value::String(text) => out.push(text.clone()), + serde_json::Value::Array(items) => { + for item in items { + if let Some(text) = item.as_str() { + out.push(text.to_string()); + } else { + for field in ["text", "content"] { + if let Some(text) = item.get(field).and_then(serde_json::Value::as_str) { + out.push(text.to_string()); + } + } + } + } + } + _ => {} + } +} + +fn claude_record_is_user(value: &serde_json::Value) -> bool { + value + .pointer("/message/role") + .or_else(|| value.get("role")) + .or_else(|| value.get("type")) + .and_then(serde_json::Value::as_str) + .is_some_and(|role| matches!(role, "user" | "human")) +} + +fn claude_event_identity(event: &ExtractedSkillEvent) -> String { + if event.skill_name.contains(':') { + event.skill_name.clone() + } else if let Some(plugin) = event.skill_plugin.as_deref() { + format!("{plugin}:{}", event.skill_name) + } else { + event.skill_name.clone() + } +} + +/// Extract Claude skill events from a raw transcript JSON value. +/// +/// Supported evidence shapes: +/// - structured `attributionSkill` / `attributionPlugin` fields at top-level +/// or under `message.*`; +/// - observed package-qualified command envelopes such as +/// `vibin:repo-status` and +/// `/vibin:repo-status` in message content. /// -/// Eng review Fix 1: callers should already have skipped calling this -/// function at all when the source text doesn't contain `"attributionSkill"` -/// as a substring — this function itself has nothing further to -/// short-circuit on since it operates on an already-parsed `Value`, not raw -/// text. +/// Duplicate evidence for the same skill on one transcript row is collapsed, +/// preferring the structured attribution when both forms are present. pub fn extract_claude_skill_events(value: &serde_json::Value) -> Vec { + let mut events = Vec::new(); + let mut seen = std::collections::HashSet::new(); let candidates = [ value, value.get("message").unwrap_or(&serde_json::Value::Null), @@ -151,11 +212,43 @@ pub fn extract_claude_skill_events(value: &serde_json::Value) -> Vec vibin:repo-status /vibin:repo-status " + }] + } + }); + let events = extract_claude_skill_events(&value); + assert_eq!(events.len(), 1); + assert_eq!(events[0].skill_name, "vibin:repo-status"); + assert_eq!(events[0].skill_plugin.as_deref(), Some("vibin")); + assert_eq!(events[0].event_kind, SkillEventKind::ClaudeSkillCommand); + assert_eq!( + events[0].evidence_kind, + SkillEvidenceKind::TranscriptContent + ); +} + +#[test] +fn ignores_unqualified_claude_command_envelopes() { + let value = json!({ + "type": "user", + "content": "help/help" + }); + assert!(extract_claude_skill_events(&value).is_empty()); +} + +#[test] +fn ignores_assistant_quoted_claude_command_envelope() { + let value = json!({ + "message": { + "role": "assistant", + "content": [{ + "type": "text", + "text": "Example: vibin:repo-status /vibin:repo-status" + }] + } + }); + assert!(extract_claude_skill_events(&value).is_empty()); +} + +#[test] +fn structured_claude_attribution_wins_over_duplicate_command_envelope() { + let value = json!({ + "attributionSkill": "repo-status", + "attributionPlugin": "vibin", + "content": "vibin:repo-status" + }); + let events = extract_claude_skill_events(&value); + assert_eq!(events.len(), 1); + assert_eq!(events[0].skill_name, "repo-status"); + assert_eq!(events[0].skill_plugin.as_deref(), Some("vibin")); + assert_eq!(events[0].event_kind, SkillEventKind::ClaudeAttribution); +} + #[test] fn emits_nothing_when_attribution_fields_absent() { let value = json!({"sessionId": "sess-1", "content": "just chatting"}); diff --git a/src/scanner_tests.rs b/src/scanner_tests.rs index 6ed47fb45..26f902d7c 100644 --- a/src/scanner_tests.rs +++ b/src/scanner_tests.rs @@ -802,6 +802,7 @@ fn append_start_requires_stored_content_hash() { last_error: None, source_revision: None, scan_state: checkpoint::SourceScanState::Complete, + extractor_revision: TRANSCRIPT_EXTRACTOR_REVISION, }; let current = FileMetadata { size: 20, @@ -2530,6 +2531,36 @@ fn indexing_claude_transcript_extracts_skill_events() { assert_eq!(event_kind, "claude_attribution"); } +#[test] +fn indexing_claude_command_envelope_extracts_skill_events() { + let (pool, dir) = test_pool(); + let file = dir.path().join("claude-command-skill.jsonl"); + std::fs::write( + &file, + concat!( + r#"{"sessionId":"sess-command","message":{"role":"user","content":[{"type":"text","text":"vibin:repo-status /vibin:repo-status"}]}}"#, + " +" + ), + ) + .unwrap(); + + let result = index_file(&pool, &file, "explicit_file").unwrap(); + assert_eq!(result.ingested, 1); + + let conn = pool.get().unwrap(); + let (skill_name, plugin, event_kind): (String, Option, 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_name, "vibin:repo-status"); + assert_eq!(plugin.as_deref(), Some("vibin")); + assert_eq!(event_kind, "claude_skill_command"); +} + #[test] fn indexing_codex_transcript_extracts_skill_events() { let (pool, dir) = test_pool(); From 05f06ad7f89f1b410d9fd5734bee66ccb6166de3 Mon Sep 17 00:00:00 2001 From: Jake Magar Date: Thu, 17 Sep 2026 21:02:07 -0400 Subject: [PATCH 2/2] fix: update rustls for RUSTSEC-2026-0285 --- Cargo.lock | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a007ee50e..6c891735c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -157,9 +157,9 @@ checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" [[package]] name = "aws-lc-rs" -version = "1.17.0" +version = "1.18.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5ec2f1fc3ec205783a5da9a7e6c1509cc69dedf09a1949e412c1e18469326d00" +checksum = "b281d307588d634de920874890732659e2e7672f72b5e10e81badc1a8a83621e" dependencies = [ "aws-lc-sys", "zeroize", @@ -167,14 +167,15 @@ dependencies = [ [[package]] name = "aws-lc-sys" -version = "0.41.0" +version = "0.45.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1a2f9779ce85b93ab6170dd940ad0169b5766ff848247aff13bb788b832fe3f4" +checksum = "9bff6c3b54fad79a2e60b8102caf565819711497c1f5f092f49508e2f5c31b27" dependencies = [ "cc", "cmake", "dunce", "fs_extra", + "pkg-config", ] [[package]] @@ -2777,9 +2778,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.40" +version = "0.23.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef86cd5876211988985292b91c96a8f2d298df24e75989a43a3c73f2d4d8168b" +checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634" dependencies = [ "aws-lc-rs", "once_cell", @@ -2840,9 +2841,9 @@ checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" [[package]] name = "rustls-webpki" -version = "0.103.13" +version = "0.103.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" +checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" dependencies = [ "aws-lc-rs", "ring", @@ -3414,7 +3415,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.2", "once_cell", "rustix", "windows-sys 0.61.2",