Improve pChronicle cache reuse and tolerant batch imports

crates/persisting-pchronicle-cli/src/exchange/decode.rs

@@ -438,7 +438,8 @@ pub(crate) fn decode_import_source(
             format!("{} is not UTF-8: {error}", diagnostic_path.display()),
         )
     })?;
-    let allow_skip = requested_format == ExchangeFormat::Auto && logical_source_path.is_some();
+    // Batch imports keep readable sources regardless of the requested format.
+    let allow_skip = logical_source_path.is_some();
     let format = match resolve_import_format(
         requested_format,
         suggested_format,
@@ -840,6 +841,79 @@ mod tests {
     use super::*;
     use std::path::Path;
 
+    #[test]
+    fn directory_import_skips_unreadable_sources_for_every_format() {
+        for format in [
+            ExchangeFormat::Atif,
+            ExchangeFormat::Actf,
+            ExchangeFormat::OpenaiMessages,
+            ExchangeFormat::Storyline,
+            ExchangeFormat::Codex,
+            ExchangeFormat::ClaudeCode,
+        ] {
+            let outcome = decode_import_source(
+                format,
+                None,
+                ImportOutputFormat::Preserve,
+                Some(Path::new("input")),
+                Some(Path::new("bad.json")),
+                Some(Path::new("bad.json")),
+                b"not json",
+                &mut Default::default(),
+            )
+            .unwrap();
+            assert!(
+                matches!(outcome, DecodeImportOutcome::Skipped { .. }),
+                "{format:?}"
+            );
+        }
+        for (format, input) in [
+            (
+                ExchangeFormat::Atif,
+                include_bytes!("../../assets/onboard/support-ticket.json").as_slice(),
+            ),
+            (
+                ExchangeFormat::Actf,
+                include_bytes!("../../assets/onboard/code-repair.actf.json").as_slice(),
+            ),
+            (
+                ExchangeFormat::OpenaiMessages,
+                include_bytes!("../../assets/onboard/training.json").as_slice(),
+            ),
+        ] {
+            let outcome = decode_import_source(
+                format,
+                None,
+                ImportOutputFormat::Preserve,
+                Some(Path::new("input")),
+                Some(Path::new("source.json")),
+                Some(Path::new("source.json")),
+                input,
+                &mut Default::default(),
+            )
+            .unwrap();
+            match outcome {
+                DecodeImportOutcome::Imported(decoded) => {
+                    assert!(!decoded.storylines.is_empty(), "{format:?}");
+                }
+                DecodeImportOutcome::Skipped { reason, .. } => panic!("{format:?}: {reason}"),
+            }
+        }
+        assert!(
+            decode_import_source(
+                ExchangeFormat::Actf,
+                None,
+                ImportOutputFormat::Preserve,
+                None,
+                None,
+                None,
+                b"{}",
+                &mut Default::default(),
+            )
+            .is_err()
+        );
+    }
+
     #[test]
     fn visible_json_extensions() {
         assert!(is_visible_json_file(Path::new("a.json")));

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

@@ -314,6 +314,7 @@ fn command(
         "SSL_CERT_DIR",
         "PCHRONICLE_QUERY_MEMORY_LIMIT",
         "PCHRONICLE_LANCE_CACHE_CAPACITY_BYTES",
+        "PCHRONICLE_LANCE_CACHE_BLOCK_SIZE_BYTES",
         // Workers issue the reads, so admission tuning that never reaches them
         // tunes nothing.
         "PCHRONICLE_OBJECT_STORE_CONCURRENCY",
@@ -339,7 +340,11 @@ impl Worker {
         // and every scope can share them. Keeping them beside the per-scope
         // caches rather than inside one means a reader does not refetch what
         // another worker already paid for.
-        let blocks = root.join("blocks");
+        let blocks = std::path::absolute(
+            std::env::var_os("PCHRONICLE_LANCE_CACHE_DIR")
+                .map(PathBuf::from)
+                .unwrap_or_else(|| root.join("blocks")),
+        )?;
         let mut builder = std::fs::DirBuilder::new();
         builder.recursive(true);
         #[cfg(unix)]

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

@@ -3121,7 +3121,8 @@ async fn directory_import_dedupes_unknown_warnings_across_sources() -> Result<()
 }
 
 #[tokio::test]
-async fn directory_import_skips_invalid_json_and_publishes_valid_sources() -> Result<()> {
+async fn directory_import_explicit_format_skips_invalid_json_and_publishes_valid_sources()
+-> Result<()> {
     let temp = tempfile::tempdir()?;
     let input = temp.path().join("input");
     fs::create_dir_all(&input)?;
@@ -3136,6 +3137,8 @@ async fn directory_import_skips_invalid_json_and_publishes_valid_sources() -> Re
         let mut argv = vec![
             "pchronicle".to_owned(),
             "import".to_owned(),
+            "--format".to_owned(),
+            "atif".to_owned(),
             "--from".to_owned(),
             input.to_string_lossy().into_owned(),
             "--output".to_owned(),

crates/persisting-pchronicle/src/formats/actf/mod.rs

@@ -176,8 +176,7 @@ fn decode_json(
     let envelope = take_unknown_fields_envelope(&mut value)?;
     let mut document: ActfDocument =
         serde_json::from_value(value).map_err(|error| InputIssue::invalid(error.to_string()))?;
-    normalize_solved_at(&mut document.solved_at);
-    reconcile_document_tool_lists(&mut document);
+    document.normalize_for_import();
     document.validate()?;
     let mut stories =
         actf_to_storylines(&document).map_err(|error| InputIssue::invalid(error.to_string()))?;
@@ -218,11 +217,15 @@ fn decode_json(
 pub struct ActfDocument {
     pub task_id: String,
     /// Some error dumps emit numeric categories (`2`) instead of strings.
-    #[serde(deserialize_with = "stringish")]
+    #[serde(default, deserialize_with = "stringish")]
     pub category: String,
+    #[serde(default, deserialize_with = "null_as_default")]
     pub k: u64,
+    #[serde(default, deserialize_with = "null_as_default")]
     pub correct: bool,
+    #[serde(default, deserialize_with = "null_as_default")]
     pub attempts_tried: u64,
+    #[serde(default, deserialize_with = "null_as_default")]
     pub solved_at: Value,
     pub attempts: BTreeMap<String, ActfAttempt>,
     #[serde(flatten)]
@@ -231,11 +234,13 @@ pub struct ActfDocument {
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct ActfAttempt {
+    #[serde(default, deserialize_with = "null_as_default")]
     pub correct: bool,
     #[serde(default)]
     pub final_answer: Value,
     #[serde(default)]
     pub ground_truth: Value,
+    #[serde(default)]
     pub trajectory: ActfTrajectory,
     #[serde(default, deserialize_with = "null_as_empty_string")]
     pub status: String,
@@ -269,6 +274,16 @@ pub struct ActfTrajectory {
     pub extra: Map<String, Value>,
 }
 
+impl Default for ActfTrajectory {
+    fn default() -> Self {
+        Self::from_event_log(Vec::new())
+    }
+}
+
+fn default_schema_version() -> String {
+    ACTF_SCHEMA_VERSION.into()
+}
+
 impl ActfTrajectory {
     pub fn from_event_log(events: Vec<Value>) -> Self {
         let started_at = events
@@ -334,6 +349,7 @@ enum ActfTrajectoryWire {
     /// object shape, not this array.
     Events(Vec<Value>),
     Canonical {
+        #[serde(default = "default_schema_version")]
         schema_version: String,
         #[serde(default)]
         steps: Vec<ActfStep>,
@@ -396,7 +412,9 @@ impl<'de> Deserialize<'de> for ActfTrajectory {
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct ActfStep {
     pub step_id: i64,
+    #[serde(default, deserialize_with = "null_as_default")]
     pub assistant_content: ActfAssistantContent,
+    #[serde(default, deserialize_with = "null_as_default")]
     pub metric: ActfMetric,
     #[serde(default, deserialize_with = "null_as_empty_string")]
     pub system_prompt: String,
@@ -424,7 +442,7 @@ impl ActfStep {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Default, Clone, PartialEq, Serialize, Deserialize)]
 pub struct ActfAssistantContent {
     #[serde(default, deserialize_with = "null_as_empty_string")]
     pub content: String,
@@ -436,7 +454,7 @@ pub struct ActfAssistantContent {
     pub extra: Map<String, Value>,
 }
 
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Default, Clone, PartialEq, Serialize, Deserialize)]
 pub struct ActfMetric {
     #[serde(default)]
     pub prompt_tokens_len: Value,
@@ -559,13 +577,23 @@ fn reconcile_document_tool_lists(document: &mut ActfDocument) {
 }
 
 impl ActfDocument {
+    /// Recover optional metadata from the attempts that are actually present.
+    pub(crate) fn normalize_for_import(&mut self) {
+        if self.category.trim().is_empty() {
+            self.category = "unknown".into();
+        }
+        self.attempts_tried = self.attempts.len() as u64;
+        self.k = self.k.max(self.attempts_tried);
+        normalize_solved_at(&mut self.solved_at);
+        reconcile_document_tool_lists(self);
+    }
+
     #[cfg(any(test, feature = "lance-store"))]
     pub fn from_json_str(input: &str) -> InputResult<Self> {
         let sanitized = super::common::sanitize_json_nonfinite(input);
         let mut document: Self = serde_json::from_str(sanitized.as_ref())
             .map_err(|error| InputIssue::invalid(error.to_string()))?;
-        normalize_solved_at(&mut document.solved_at);
-        reconcile_document_tool_lists(&mut document);
+        document.normalize_for_import();
         document.validate()?;
         Ok(document)
     }
@@ -762,6 +790,38 @@ mod tests {
         .unwrap()
     }
 
+    #[test]
+    fn imports_incomplete_actf_metadata_without_losing_steps() {
+        let mut value = serde_json::to_value(fixture()).unwrap();
+        for key in ["category", "correct", "solved_at"] {
+            value.as_object_mut().unwrap().remove(key);
+        }
+        value["k"] = json!(0);
+        value["attempts_tried"] = json!(99);
+        let attempt = value["attempts"]["1"].as_object_mut().unwrap();
+        attempt.remove("correct");
+        let trajectory = attempt["trajectory"].as_object_mut().unwrap();
+        trajectory.remove("schema_version");
+        let step = trajectory["steps"][0].as_object_mut().unwrap();
+        step.remove("metric");
+        let stories = crate::document::decode_json_storylines(
+            DocumentFormat::Actf,
+            &value.to_string(),
+            "incomplete.actf.json",
+        )
+        .unwrap();
+        assert_eq!(stories.len(), 1);
+        assert_eq!(stories[0].turns.len(), 1);
+        assert_eq!(
+            stories[0].turns[0].reasoning_content.as_deref(),
+            Some("inspect")
+        );
+        assert_eq!(stories[0].turns[0].tool_calls.as_ref().unwrap().len(), 1);
+        let document = ActfDocument::from_json_str(&value.to_string()).unwrap();
+        assert_eq!(document.category, "unknown");
+        assert_eq!((document.k, document.attempts_tried), (1, 1));
+    }
+
     #[test]
     fn accepts_null_or_empty_object_trajectory_as_empty_event_log() {
         for trajectory in [json!(null), json!({})] {