use aether_core::events::{AgentEvent, ContextEvent, ModelEvent, TurnEvent, TurnOutcome};
use aether_sessions::analytics::{IngestOptions, QueryOptions, SessionIndexError, ingest_sessions, run_query};
use aether_sessions::testing::{
DEFAULT_CREATED_AT, session_meta, tool_call, tool_error, tool_result, turn_ended, user_message,
};
use aether_sessions::{SessionEvent, discover_session_files};
use llm::testing::session_usage_event;
use llm::{
ContextUsage, LlmCallPurpose, ModelIdentity, SessionUsageEvent, SessionUsageTotals, TokenUsage, Tokens, UsageCost,
UsageSource, Usd,
};
use serde_json::json;
use std::fs;
use std::path::Path;
use tempfile::TempDir;
#[test]
fn shared_discovery_filters_and_sorts_session_files_with_fingerprints() {
let temp = TempDir::new().unwrap();
fs::write(temp.path().join("b.jsonl"), "b").unwrap();
fs::write(temp.path().join("a.jsonl"), "aa").unwrap();
fs::write(temp.path().join("prompt-history.jsonl"), "history").unwrap();
fs::write(temp.path().join("notes.txt"), "notes").unwrap();
let files = discover_session_files(temp.path()).unwrap();
assert_eq!(
files.iter().map(|file| file.path.file_name().unwrap().to_string_lossy().into_owned()).collect::<Vec<_>>(),
["a.jsonl", "b.jsonl"],
);
assert_eq!(files[0].fingerprint.file_size, 2);
assert_eq!(files[1].fingerprint.file_size, 1);
}
#[tokio::test]
async fn typed_event_contract_populates_every_documented_view() {
let fixture = Fixture::new();
fixture.write_session(
"s1.jsonl",
&[
user_message("hello"),
SessionEvent::Agent(AgentEvent::Turn(TurnEvent::RetryScheduled {
purpose: LlmCallPurpose::Chat,
attempt: 1,
max_attempts: 3,
delay_ms: 10,
})),
turn_ended(TurnOutcome::Cancelled),
SessionEvent::Agent(AgentEvent::Model(ModelEvent::Switched {
previous: "old".to_string(),
new: "new".to_string(),
})),
SessionEvent::Agent(AgentEvent::Context(ContextEvent::UsageUpdated {
usage: ContextUsage { usage_ratio: Some(0.9), ..ContextUsage::default() },
})),
SessionEvent::Agent(AgentEvent::SessionUsage(root_usage(&UsageSource::new("root")))),
],
);
fixture.ingest().await;
assert_eq!(fixture.query("select count(*) from session_usage").await.rows, vec![vec![json!(1)]]);
assert_eq!(fixture.query("select count(*) from user_messages").await.rows, vec![vec![json!(1)]]);
assert_eq!(fixture.query("select count(*) from retries").await.rows, vec![vec![json!(1)]]);
assert_eq!(fixture.query("select count(*) from cancellations").await.rows, vec![vec![json!(1)]]);
assert_eq!(
fixture.query("select model_name from events where event_type = 'model_switched'").await.rows,
vec![vec![json!("new")]]
);
assert_eq!(fixture.query("select count(*) from agent_messages").await.rows, vec![vec![json!(0)]]);
}
#[tokio::test]
async fn one_database_failure_does_not_abort_other_files() {
let fixture = Fixture::new();
fixture.write_session_with_id("a.jsonl", "duplicate", &[user_message("first")]);
fixture.write_session_with_id("b.jsonl", "duplicate", &[user_message("second")]);
fixture.write_session("c.jsonl", &[user_message("third")]);
let summary = fixture.ingest().await;
assert_eq!(summary.files_indexed, 2);
assert_eq!(summary.files_failed, 1);
let sessions = fixture.query("select session_id from sessions order by session_id").await;
assert_eq!(sessions.rows, vec![vec![json!("c")], vec![json!("duplicate")]]);
}
#[tokio::test]
async fn end_to_end_ingest_query() {
let fixture = Fixture::new();
fixture.write_session(
"s1.jsonl",
&[
user_message("hello"),
tool_call("call-read", "read", "{}"),
tool_error("call-read", "read", "failed"),
context_usage(0.9),
],
);
let summary = fixture.ingest().await;
assert_eq!(summary.files_seen, 1);
assert_eq!(summary.files_indexed, 1);
assert_eq!(summary.events_indexed, 4);
let output = fixture.query("select tool_name, count(*) as failures from tool_errors group by tool_name").await;
assert_eq!(output.rows, vec![vec![json!("read"), json!(1)]]);
}
#[tokio::test]
async fn typed_projection_exposes_event_fields() {
let fixture = Fixture::new();
fixture
.write_session("s1.jsonl", &[user_message("hello"), tool_call("call-read", "read", "{}"), context_usage(0.9)]);
fixture.ingest().await;
let usage = fixture.query("select usage_ratio, input_tokens from context_usage").await;
assert_eq!(usage.rows, vec![vec![json!(0.9), json!(1)]]);
let model = fixture.query("select model_name from tool_calls").await;
assert_eq!(model.rows, vec![vec![serde_json::Value::Null]]);
}
#[tokio::test]
async fn session_usage_columns_are_projected_for_root_and_sub_agent_samples() {
let fixture = Fixture::new();
let root = UsageSource::new("root");
let mut child = UsageSource::new("explorer");
child.parent_agent_id = Some(root.agent_id.clone());
child.task_id = Some("task_0".to_string());
let child_usage = SessionUsageEvent {
source: child,
purpose: LlmCallPurpose::Compaction,
totals: SessionUsageTotals {
tokens: TokenUsage { cache_read_tokens: Some(3.into()), ..TokenUsage::new(14, 7) },
estimated_usd: Usd::new(0.25),
unpriced_calls: 1,
..SessionUsageTotals::default()
},
..session_usage_event(2, TokenUsage::new(4, 2))
};
fixture.write_session(
"s1.jsonl",
&[
SessionEvent::Agent(AgentEvent::SessionUsage(root_usage(&root))),
SessionEvent::Agent(AgentEvent::SessionUsage(child_usage)),
],
);
fixture.ingest().await;
let rows = fixture
.query(
"select usage_sequence, agent_name, parent_agent_id, task_id, call_purpose, provider, model_name, input_tokens, cache_read_tokens, total_input_tokens, estimated_cost_usd, total_estimated_cost_usd, unpriced_calls from session_usage order by usage_sequence",
)
.await;
assert_eq!(
rows.rows,
vec![
vec![
json!(1),
json!("root"),
serde_json::Value::Null,
serde_json::Value::Null,
json!("chat"),
json!("anthropic"),
json!("claude"),
json!(10),
json!(3),
json!(10),
json!(0.25),
json!(0.25),
json!(0),
],
vec![
json!(2),
json!("explorer"),
json!(root.agent_id),
json!("task_0"),
json!("compaction"),
serde_json::Value::Null,
serde_json::Value::Null,
json!(4),
serde_json::Value::Null,
json!(14),
serde_json::Value::Null,
json!(0.25),
json!(1),
],
]
);
}
#[tokio::test]
async fn tool_columns_are_projected_from_typed_events() {
let fixture = Fixture::new();
fixture.write_session(
"s1.jsonl",
&[
user_message("hello"),
tool_call("call-read", "read", "{}"),
tool_result("call-read", "read", "ok"),
tool_error("call-write", "write", "failed"),
],
);
fixture.ingest().await;
let schema = fixture.query("select sql from sqlite_master where type = 'table' and name = 'events'").await;
let schema_sql = schema.rows[0][0].as_str().unwrap();
assert!(schema_sql.contains("tool_call_id text"));
assert!(schema_sql.contains("tool_name text"));
assert!(schema_sql.contains("tool_arguments text"));
assert!(!schema_sql.contains("generated always"));
let tools = fixture
.query("select event_type, tool_call_id, tool_name, tool_arguments from events where tool_call_id is not null order by event_index")
.await;
assert_eq!(
tools.rows,
vec![
vec![json!("tool_call"), json!("call-read"), json!("read"), json!("{}")],
vec![json!("tool_result"), json!("call-read"), json!("read"), json!("{}")],
vec![json!("tool_error"), json!("call-write"), json!("write"), json!("{}")],
]
);
}
#[tokio::test]
async fn concurrent_ingest_indexes_multiple_changed_files_deterministically() {
let fixture = Fixture::new();
fixture.write_session("b.jsonl", &[user_message("second")]);
fixture.write_session("a.jsonl", &[user_message("first")]);
let summary = fixture.ingest().await;
assert_eq!(summary.files_seen, 2);
assert_eq!(summary.files_indexed, 2);
let output = fixture.query("select session_id from sessions order by source_path").await;
assert_eq!(output.rows, vec![vec![json!("a")], vec![json!("b")]]);
}
#[tokio::test]
async fn idempotent_rerun_skips_unchanged_files() {
let fixture = Fixture::new();
fixture.write_session("s1.jsonl", &[user_message("hello"), tool_error("call-read", "read", "failed")]);
fixture.ingest().await;
let second = fixture.ingest().await;
assert_eq!(second.files_skipped_unchanged, 1);
let output = fixture.query("select count(*) from events").await;
assert_eq!(output.rows, vec![vec![json!(2)]]);
}
#[tokio::test]
async fn changed_file_replaces_old_rows() {
let fixture = Fixture::new();
fixture.write_session("s1.jsonl", &[user_message("hello")]);
fixture.ingest().await;
fixture.write_session("s1.jsonl", &[user_message("hello"), tool_error("call-read", "read", "failed")]);
fixture.ingest().await;
let output = fixture.query("select count(*) from events").await;
assert_eq!(output.rows, vec![vec![json!(2)]]);
}
#[tokio::test]
async fn deleted_file_pruning_removes_rows() {
let fixture = Fixture::new();
fixture.write_session("s1.jsonl", &[user_message("hello"), tool_error("call-read", "read", "failed")]);
fixture.ingest().await;
fs::remove_file(fixture.sessions_dir.join("s1.jsonl")).unwrap();
let summary = fixture.ingest().await;
assert_eq!(summary.stale_files_pruned, 1);
let output = fixture.query("select count(*) from events").await;
assert_eq!(output.rows, vec![vec![json!(0)]]);
}
#[tokio::test]
async fn malformed_event_line_is_recorded() {
let fixture = Fixture::new();
let meta = serde_json::to_string(&session_meta("s1", DEFAULT_CREATED_AT)).unwrap();
let hello = serde_json::to_string(&user_message("hello")).unwrap();
let failed = serde_json::to_string(&tool_error("call-read", "read", "failed")).unwrap();
fs::write(fixture.sessions_dir.join("s1.jsonl"), format!("{meta}\n{hello}\nnot-json\n{failed}\n")).unwrap();
let summary = fixture.ingest().await;
assert_eq!(summary.parse_errors, 1);
let output = fixture.query("select count(*) from events").await;
assert_eq!(output.rows, vec![vec![json!(2)]]);
let errors = fixture.query("select count(*) from parse_errors").await;
assert_eq!(errors.rows, vec![vec![json!(1)]]);
}
#[tokio::test]
async fn malformed_metadata_records_file_failure() {
let fixture = Fixture::new();
fs::write(fixture.sessions_dir.join("bad.jsonl"), "not-json\n{}").unwrap();
let summary = fixture.ingest().await;
assert_eq!(summary.files_failed, 1);
let output = fixture.query("select status, count(*) from session_files group by status").await;
assert_eq!(output.rows, vec![vec![json!("error"), json!(1)]]);
}
#[tokio::test]
async fn valid_file_replaces_previous_metadata_failure() {
let fixture = Fixture::new();
fs::write(fixture.sessions_dir.join("s1.jsonl"), "not-json\n{}").unwrap();
let first = fixture.ingest().await;
assert_eq!(first.files_failed, 1);
fixture.write_session("s1.jsonl", &[user_message("hello")]);
let second = fixture.ingest().await;
assert_eq!(second.files_indexed, 1);
let output = fixture.query("select status, session_id, event_count from session_files").await;
assert_eq!(output.rows, vec![vec![json!("indexed"), json!("s1"), json!(1)]]);
}
#[tokio::test]
async fn metadata_failure_can_be_recorded_repeatedly() {
let fixture = Fixture::new();
fs::write(fixture.sessions_dir.join("bad.jsonl"), "not-json\n{}").unwrap();
fixture.ingest().await;
fs::write(fixture.sessions_dir.join("bad.jsonl"), "still-not-json\n{}").unwrap();
let second = fixture.ingest().await;
assert_eq!(second.files_failed, 1);
let output = fixture.query("select status, session_id from session_files").await;
assert_eq!(output.rows, vec![vec![json!("error"), serde_json::Value::Null]]);
}
#[tokio::test]
async fn read_only_query_safety_keeps_rows() {
let fixture = Fixture::new();
fixture.write_session("s1.jsonl", &[user_message("hello")]);
fixture.ingest().await;
let result = run_query(&QueryOptions {
db_path: fixture.db_path.clone(),
sql: "delete from events".to_string(),
max_rows: 100,
max_cell_chars: 1000,
timeout_ms: 2000,
})
.await;
assert!(matches!(result, Err(SessionIndexError::Sqlx(_))));
let output = fixture.query("select count(*) from events").await;
assert_eq!(output.rows, vec![vec![json!(1)]]);
}
struct Fixture {
_temp: TempDir,
sessions_dir: std::path::PathBuf,
db_path: std::path::PathBuf,
}
impl Fixture {
fn new() -> Self {
let temp = TempDir::new().unwrap();
let sessions_dir = temp.path().join("sessions");
fs::create_dir_all(&sessions_dir).unwrap();
let db_path = temp.path().join("index.sqlite");
Self { _temp: temp, sessions_dir, db_path }
}
fn write_session(&self, name: &str, events: &[SessionEvent]) {
let session_id = Path::new(name).file_stem().unwrap().to_string_lossy();
self.write_session_with_id(name, &session_id, events);
}
fn write_session_with_id(&self, name: &str, session_id: &str, events: &[SessionEvent]) {
let mut content = serde_json::to_string(&session_meta(session_id, DEFAULT_CREATED_AT)).unwrap();
content.push('\n');
for event in events {
content.push_str(&serde_json::to_string(event).unwrap());
content.push('\n');
}
fs::write(self.sessions_dir.join(name), content).unwrap();
}
async fn ingest(&self) -> aether_sessions::analytics::IngestSummary {
ingest_sessions(self.ingest_options()).await.unwrap()
}
async fn query(&self, sql: &str) -> aether_sessions::analytics::QueryOutput {
run_query(&self.query_options(sql)).await.unwrap()
}
fn ingest_options(&self) -> IngestOptions {
IngestOptions {
sessions_dir: self.sessions_dir.clone(),
db_path: self.db_path.clone(),
prune: true,
parse_concurrency: 2,
}
}
fn query_options(&self, sql: &str) -> QueryOptions {
QueryOptions {
db_path: self.db_path.clone(),
sql: sql.to_string(),
max_rows: 100,
max_cell_chars: 1000,
timeout_ms: 2000,
}
}
}
fn root_usage(source: &UsageSource) -> SessionUsageEvent {
let tokens = TokenUsage { cache_read_tokens: Some(3.into()), ..TokenUsage::new(10, 5) };
SessionUsageEvent {
source: source.clone(),
model: ModelIdentity { provider: Some("anthropic".into()), model_id: Some("claude".into()), pricing: None },
estimated_cost: Some(UsageCost { total_usd: Usd::new(0.25), ..UsageCost::default() }),
totals: SessionUsageTotals { tokens, estimated_usd: Usd::new(0.25), ..SessionUsageTotals::default() },
..session_usage_event(1, tokens)
}
}
fn context_usage(ratio: f64) -> SessionEvent {
SessionEvent::Agent(AgentEvent::Context(ContextEvent::UsageUpdated {
usage: ContextUsage {
usage_ratio: Some(ratio),
context_limit: Some(Tokens::new(100)),
input_tokens: Tokens::new(1),
},
}))
}