fix(pchronicle): generate ids for compact JSONL records

crates/persisting-pchronicle-cli/src/lib.rs

@@ -207,7 +207,7 @@ enum Command {
     ///
     /// Compact JSONL stores each JSON object as one record without trajectory semantics.
     /// Select it with either --input-format compact-jsonl or
-    /// --output-format compact-jsonl; it accepts local .jsonl input and supports
+    /// --output-format compact-jsonl; it accepts local JSON/JSONL input and supports
     /// create or replace, not stdin or append.
     Import(ImportArgs),
     /// Permanently delete a Dataset directory or object-store prefix.
@@ -702,7 +702,7 @@ struct ImportArgs {
     output: Option<String>,
 
     /// Input exchange format. Auto detects run data; compact-jsonl is explicit.
-    /// Compact JSONL recursively reads local .jsonl files, one object per line.
+    /// Compact JSONL recursively reads local .json, .jsonl, or .ndjson files.
     #[arg(short = 'i', long = "input-format", alias = "format", value_enum, default_value_t = ExchangeFormat::Auto)]
     format: ExchangeFormat,
 
@@ -730,7 +730,8 @@ struct ImportArgs {
     #[arg(long, value_parser = parse_byte_size, default_value = "256MiB")]
     max_input_bytes: Option<usize>,
 
-    /// Compact JSONL mapping. id/timestamp override $.id/$.timestamp; other names add JSONB columns.
+    /// Compact JSONL mapping. id/timestamp override $.id/$.timestamp; missing or invalid id values
+    /// use source_filename#line_number; other names add JSONB columns.
     /// Example: --column id=$.event.id --column model=$.payload.model.
     #[arg(long = "column", value_name = "NAME=JSON_PATH", action = clap::ArgAction::Append)]
     columns: Vec<String>,

crates/persisting-pchronicle-cli/src/server/explorer.rs

@@ -52,9 +52,12 @@ pub(crate) struct CatalogTree {
 pub(crate) struct CatalogTreeChild {
     pub(crate) name: String,
     pub(crate) kind: String,
+    pub(crate) data_type: String,
     pub(crate) path: String,
     pub(crate) run_count: usize,
     pub(crate) failed_count: usize,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) total_tokens: Option<u64>,
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub(crate) entries: Vec<CatalogTreeChild>,
 }
@@ -105,6 +108,27 @@ struct ChildAcc {
     run_count: usize,
     failed_count: usize,
     has_deeper: bool,
+    data_types: BTreeSet<String>,
+}
+
+fn data_type(format: Option<&str>) -> &'static str {
+    match format {
+        Some("compact-jsonl/v1") => "compact-jsonl",
+        Some("storyline-lance") | None => "storyline",
+        Some(_) => "other",
+    }
+}
+
+fn combined_data_type(data_types: &BTreeSet<String>) -> String {
+    match data_types.len() {
+        0 => "unknown".into(),
+        1 => data_types
+            .iter()
+            .next()
+            .cloned()
+            .unwrap_or_else(|| "unknown".into()),
+        _ => "mixed".into(),
+    }
 }
 
 fn dataset_children(runs: &[&RunSummary]) -> Vec<CatalogTreeChild> {
@@ -114,8 +138,12 @@ fn dataset_children(runs: &[&RunSummary]) -> Vec<CatalogTreeChild> {
             run_count: 0,
             failed_count: 0,
             has_deeper: false,
+            data_types: BTreeSet::new(),
         });
         entry.run_count += 1;
+        entry
+            .data_types
+            .insert(data_type(run.format.as_deref()).into());
         if is_failed_status(&run.status) {
             entry.failed_count += 1;
         }
@@ -125,9 +153,11 @@ fn dataset_children(runs: &[&RunSummary]) -> Vec<CatalogTreeChild> {
         .map(|(name, acc)| CatalogTreeChild {
             name: name.clone(),
             kind: "dataset".into(),
+            data_type: combined_data_type(&acc.data_types),
             path: name,
             run_count: acc.run_count,
             failed_count: acc.failed_count,
+            total_tokens: None,
             entries: Vec::new(),
         })
         .collect()
@@ -161,8 +191,12 @@ fn file_children(runs: &[&RunSummary], prefix: &str) -> Vec<CatalogTreeChild> {
             run_count: 0,
             failed_count: 0,
             has_deeper: false,
+            data_types: BTreeSet::new(),
         });
         entry.run_count += 1;
+        entry
+            .data_types
+            .insert(data_type(run.format.as_deref()).into());
         if is_failed_status(&run.status) {
             entry.failed_count += 1;
         }
@@ -184,6 +218,8 @@ fn file_children(runs: &[&RunSummary], prefix: &str) -> Vec<CatalogTreeChild> {
             name,
             run_count: acc.run_count,
             failed_count: acc.failed_count,
+            data_type: combined_data_type(&acc.data_types),
+            total_tokens: None,
             entries: Vec::new(),
         })
         .collect()
@@ -208,14 +244,40 @@ fn fold_tree_children(
     children.push(CatalogTreeChild {
         name: "other".into(),
         kind: "other".into(),
+        data_type: "mixed".into(),
         path: prefix.to_string(),
         run_count: rest.iter().map(|child| child.run_count).sum(),
         failed_count: rest.iter().map(|child| child.failed_count).sum(),
+        total_tokens: None,
         entries: rest,
     });
     children
 }
 
+pub(crate) fn apply_total_tokens(
+    children: &mut [CatalogTreeChild],
+    file_tokens: &BTreeMap<String, u64>,
+) {
+    for child in children {
+        apply_total_tokens(&mut child.entries, file_tokens);
+        child.total_tokens = if child.entries.is_empty() {
+            file_tokens
+                .iter()
+                .filter(|(file, _)| {
+                    *file == &child.path || file.starts_with(&format!("{}/", child.path))
+                })
+                .map(|(_, tokens)| *tokens)
+                .reduce(u64::saturating_add)
+        } else {
+            child
+                .entries
+                .iter()
+                .filter_map(|entry| entry.total_tokens)
+                .reduce(u64::saturating_add)
+        };
+    }
+}
+
 #[derive(Clone, Debug, Serialize)]
 pub(crate) struct ExplorerPage<T> {
     pub(crate) snapshot: PageSnapshot,

crates/persisting-pchronicle-cli/src/server/mod.rs

@@ -843,6 +843,8 @@ async fn explorer_tree(
         let (duration_ms, total_tokens) = tree_prefix_metrics(&runtime, &name, &tree.prefix).await;
         tree.duration_ms = duration_ms;
         tree.total_tokens = total_tokens;
+        let file_tokens = tree_file_tokens(&runtime, &name, &tree.prefix).await;
+        explorer::apply_total_tokens(&mut tree.children, &file_tokens);
     }
     Ok(Json(tree))
 }
@@ -870,7 +872,11 @@ async fn tree_prefix_metrics(
         format!(" WHERE _file_ = '{escaped}' OR _file_ LIKE '{escaped}/%'")
     };
     let sql = format!(
-        "SELECT MIN(timestamp) AS start_ts, MAX(timestamp) AS end_ts FROM {ident}.steps{file_clause}"
+        "SELECT MIN(timestamp) AS start_ts, MAX(timestamp) AS end_ts, SUM(COALESCE(\
+            json_get_int(metrics, 'total_tokens'),\
+            json_get_int(metrics, 'prompt_tokens') + json_get_int(metrics, 'completion_tokens'),\
+            json_get_int(metrics, 'prompt_tokens_len') + json_get_int(metrics, 'completion_tokens_len')\
+        )) AS total_tokens FROM {ident}.steps{file_clause}"
     );
     let mut buffer = Vec::new();
     let write = tokio::time::timeout(
@@ -894,6 +900,56 @@ async fn tree_prefix_metrics(
     )
 }
 
+async fn tree_file_tokens(
+    runtime: &CatalogRuntime,
+    dataset: &str,
+    prefix: &str,
+) -> BTreeMap<String, u64> {
+    let Some(ident) = sql_ident(dataset) else {
+        return BTreeMap::new();
+    };
+    let file_clause = if prefix.is_empty() {
+        String::new()
+    } else {
+        let escaped = prefix.replace('\'', "''");
+        format!(" WHERE _file_ = '{escaped}' OR _file_ LIKE '{escaped}/%'")
+    };
+    let sql = format!(
+        "SELECT _file_, SUM(COALESCE(\
+            json_get_int(metrics, 'total_tokens'),\
+            json_get_int(metrics, 'prompt_tokens') + json_get_int(metrics, 'completion_tokens'),\
+            json_get_int(metrics, 'prompt_tokens_len') + json_get_int(metrics, 'completion_tokens_len')\
+        )) AS total_tokens FROM {ident}.steps{file_clause} GROUP BY _file_"
+    );
+    let mut buffer = Vec::new();
+    let write = tokio::time::timeout(
+        Duration::from_secs(3),
+        runtime
+            .engine
+            .write_query_jsonl_with_max_rows(&sql, &mut buffer, None),
+    )
+    .await;
+    let Ok(Ok(())) = write else {
+        return BTreeMap::new();
+    };
+    buffer
+        .split(|byte| *byte == b'\n')
+        .filter_map(|line| {
+            let row: Value = serde_json::from_slice(line).ok()?;
+            let file = row.get("_file_")?.as_str()?.to_owned();
+            let tokens = row
+                .get("total_tokens")
+                .and_then(Value::as_u64)
+                .or_else(|| {
+                    row.get("total_tokens")
+                        .and_then(Value::as_i64)
+                        .map(|value| value.max(0) as u64)
+                })?;
+            Some((file, tokens))
+        })
+        .collect()
+}
+
 fn timestamp_span_ms(start: Option<&Value>, end: Option<&Value>) -> Option<i64> {
     let start = json_timestamp_ms(start?)?;
     let end = json_timestamp_ms(end?)?;

crates/persisting-pchronicle/src/store/catalog/mod.rs

@@ -836,6 +836,7 @@ fn is_lance_directory(path: &Path) -> bool {
     path.extension()
         .and_then(|extension| extension.to_str())
         .is_some_and(|extension| extension.eq_ignore_ascii_case("lance"))
+        || path.join("_versions").is_dir()
 }
 
 fn path_is_inside_lance_directory(path: &str) -> bool {

crates/persisting-pchronicle/src/store/catalog/tests.rs

@@ -219,6 +219,33 @@ async fn ignores_derived_lance_sidecars_during_discovery() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn discovers_extensionless_compact_lance_dataset() -> Result<()> {
+    let temp = tempfile::tempdir()?;
+    let input = temp.path().join("input.jsonl");
+    let compact = temp.path().join("compact");
+    fs::write(&input, b"{\"timestamp\":1,\"value\":\"ok\"}\n")?;
+    crate::storage::CompactJsonlStore::import_path(
+        &input,
+        &compact,
+        &crate::storage::CompactJsonlOptions::default(),
+    )
+    .await?;
+    fs::remove_file(input)?;
+
+    let snapshot = DatasetCatalogSnapshot::discover(
+        vec![DatasetMount::default(temp.path().to_string_lossy())?],
+        Some(DEFAULT_DATASET_NAME.into()),
+        CatalogSnapshotOptions::default(),
+    )
+    .await?;
+    let sources = &snapshot.datasets()[0].sources;
+    assert_eq!(sources.len(), 1);
+    assert_eq!(sources[0].file, "compact");
+    assert_eq!(sources[0].format.as_deref(), Some("compact-jsonl/v1"));
+    Ok(())
+}
+
 #[tokio::test]
 async fn report_mode_skips_oversized_files_when_querying_all_runs() -> Result<()> {
     let temp = tempfile::tempdir()?;