Fix clean-checkout CI test failures

reports/release_validation_report.json

@@ -140,7 +140,7 @@
     ],
     "valid": true
   },
-  "runtime_seconds": 0.787532,
+  "runtime_seconds": 0.826618,
   "selected_calibration_method": "uncalibrated",
   "selected_experiment": "exp_05_random_forest",
   "source_snapshot_id": "edbd4bd813ed8e1dbaba9e1c",

run_baseline.py

@@ -8,6 +8,7 @@
 
 from src.baseline import DEFAULT_ALPHA, DEFAULT_MIN_GROUP_SIZE, build_industry_rates, choose_naics_grouping, score_validation_rows
 from src.data_foundation import FoundationError, read_json, sha256_file, write_csv, write_json
+from src.path_utils import RelativePathError, resolve_report_path
 from src.ranking_metrics import validation_metrics
 from src.splitting import SplitError, create_chronological_split, write_split_artifacts
 from src.validation import load_schema, validate_output_columns
@@ -25,9 +26,12 @@ def read_csv(path: Path) -> list[dict[str, Any]]:
         raise BaselineError(f"Could not read labelled snapshot: {error}") from error
 
 
-def verify_processed_snapshot(foundation: dict[str, Any]) -> tuple[Path, list[dict[str, Any]]]:
+def verify_processed_snapshot(foundation: dict[str, Any], base_directory: Path) -> tuple[Path, list[dict[str, Any]]]:
     output = foundation.get("processed_output", {})
-    directory = Path(output.get("path", ""))
+    try:
+        directory = resolve_report_path(output.get("path", ""), base_directory)
+    except RelativePathError as error:
+        raise BaselineError(str(error)) from error
     labelled_path = directory / "labelled_inspections.csv"
     expected = output.get("hashes", {}).get("labelled_inspections.csv")
     if not directory.exists() or not expected or not labelled_path.exists() or sha256_file(labelled_path) != expected:
@@ -41,7 +45,7 @@ def verify_processed_snapshot(foundation: dict[str, Any]) -> tuple[Path, list[di
 def run_baseline(*, foundation_report_path: Path, schema_path: Path, artifact_root: Path | None = None) -> dict[str, Any]:
     foundation = read_json(foundation_report_path)
     schema = load_schema(schema_path)
-    processed_directory, rows = verify_processed_snapshot(foundation)
+    processed_directory, rows = verify_processed_snapshot(foundation, foundation_report_path.parent.parent)
     validate_output_columns(list(rows[0]), schema)
     try:
         split = create_chronological_split(rows)

run_feature_engineering.py

@@ -27,6 +27,7 @@
     transform_batch,
     write_feature_artifacts,
 )
+from src.path_utils import RelativePathError, resolve_report_path
 
 
 def read_json(path: Path) -> dict[str, Any]:
@@ -59,9 +60,12 @@ def write_json_atomic(path: Path, value: dict[str, Any]) -> None:
     temporary.replace(path)
 
 
-def load_verified_splits(baseline: dict[str, Any]) -> tuple[Path, dict[str, list[dict[str, Any]]], dict[str, Any]]:
+def load_verified_splits(baseline: dict[str, Any], base_directory: Path) -> tuple[Path, dict[str, list[dict[str, Any]]], dict[str, Any]]:
     artifact = baseline.get("split_artifacts", {})
-    directory = Path(artifact.get("directory", ""))
+    try:
+        directory = resolve_report_path(artifact.get("directory", ""), base_directory)
+    except RelativePathError as error:
+        raise FeatureEngineeringError(str(error)) from error
     manifest_path = directory / "split_manifest.json"
     manifest = read_json(manifest_path)
     if not manifest.get("strictly_ordered") or manifest.get("id_overlap_count") != 0:
@@ -115,7 +119,7 @@ def run_feature_engineering(
     snapshot_id = foundation.get("snapshot_id")
     if not snapshot_id or baseline.get("source_snapshot_id") != snapshot_id:
         raise FeatureEngineeringError("Data-foundation and baseline reports do not identify the same source snapshot.")
-    split_directory, inputs, split_manifest = load_verified_splits(baseline)
+    split_directory, inputs, split_manifest = load_verified_splits(baseline, baseline_report_path.parent.parent)
     all_input_rows = [*inputs["train"], *inputs["validation"], *inputs["test"]]
     key_assessment = inspect_establishment_key(all_input_rows)
     employee = employee_count_assessment(all_input_rows, float(config["employee_count"]["minimum_coverage_percentage"]))

src/data_foundation.py

@@ -13,6 +13,7 @@
 
 from run_feasibility import NON_POSITIVE_TYPES, POSITIVE_TYPES, build_label_table
 from src.validation import ValidationError, load_schema, missing_percentages, validate_inspections, validate_output_columns
+from src.path_utils import RelativePathError, resolve_report_path
 
 
 LABEL_MAPPING_VERSION = "day0-viol-type-v1"
@@ -248,7 +249,11 @@ def run_foundation(
     if not required_config.issubset(configuration) or not cache_directory:
         raise FoundationError("Day 0 feasibility report does not identify a usable cache configuration.")
     schema = load_schema(schema_path)
-    inspections, violations, completed_ids, sources = load_day0_cache(Path(cache_directory), configuration)
+    try:
+        cache_root = resolve_report_path(cache_directory, report_path.parent.parent)
+    except RelativePathError as error:
+        raise FoundationError(str(error)) from error
+    inspections, violations, completed_ids, sources = load_day0_cache(cache_root, configuration)
     inspection_ids = [str(row["activity_nr"]) for row in inspections if row.get("activity_nr") is not None]
     snapshot_id = stable_snapshot_id(configuration, inspection_ids, list(completed_ids))
     raw_directory, raw_manifest, raw_reused = create_raw_snapshot(

src/path_utils.py

@@ -0,0 +1,30 @@
+"""Portable, deliberately strict resolution of report-relative artifact paths."""
+from __future__ import annotations
+
+from pathlib import Path, PureWindowsPath
+
+
+class RelativePathError(ValueError):
+    """A report path is absolute or escapes the declared artifact base."""
+
+
+def resolve_report_path(value: str | Path, base_directory: Path) -> Path:
+    """Resolve a slash-agnostic relative report path under ``base_directory``.
+
+    Reports are portable when they store relative artifact paths.  Absolute
+    paths are rejected (including Windows drive paths on Unix), as are parent
+    traversals that would escape the supplied base directory.
+    """
+    text = str(value).strip().replace("\\", "/")
+    if not text:
+        raise RelativePathError("Artifact path is empty.")
+    normalized = Path(text)
+    if normalized.is_absolute() or PureWindowsPath(text).is_absolute():
+        raise RelativePathError("Absolute artifact paths are not permitted in reports.")
+    base = base_directory.resolve()
+    candidate = (base / normalized).resolve()
+    try:
+        candidate.relative_to(base)
+    except ValueError as error:
+        raise RelativePathError("Artifact path escapes its report base directory.") from error
+    return candidate

src/release_validation.py

@@ -19,6 +19,8 @@
 
 import yaml
 
+from src.path_utils import RelativePathError, resolve_report_path
+
 
 class ReleaseValidationError(RuntimeError):
     """A release contract check failed."""
@@ -67,9 +69,10 @@ def _require(condition: bool, message: str) -> None:
 
 def _path_from_report(root: Path, value: Any, field: str) -> Path:
     _require(isinstance(value, str) and value, f"Missing report path: {field}")
-    path = Path(value.replace("\\", "/"))
-    _require(not path.is_absolute(), f"Absolute artifact path is not permitted: {field}")
-    return root / path
+    try:
+        return resolve_report_path(value, root)
+    except RelativePathError as error:
+        raise ReleaseValidationError(f"Invalid artifact path for {field}: {error}") from error
 
 
 def _workflow_checks(root: Path) -> dict[str, bool]:

tests/test_baseline.py

@@ -1,11 +1,13 @@
 import copy
+import json
 import tempfile
 import unittest
 from pathlib import Path
 from unittest.mock import patch
 
 from run_baseline import run_baseline
 from src.baseline import build_industry_rates, choose_naics_grouping, score_validation_rows
+from src.data_foundation import sha256_file, write_csv
 from src.ranking_metrics import average_precision, ranking_at_fraction, roc_auc, selection_count
 from src.splitting import SplitError, create_chronological_split, write_split_artifacts
 
@@ -93,10 +95,23 @@ def test_ranking_metrics_use_ceil_and_handle_missing_class(self):
 
     def test_offline_smoke_never_reports_locked_test_metrics(self):
         with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            processed = root / "data" / "processed" / "snapshot"
+            rows = [
+                row("1", "2020-01-02", 1), row("2", "2020-02-02", 0),
+                row("3", "2021-01-02", 1), row("4", "2021-02-02", 0),
+                row("5", "2022-01-02", 1), row("6", "2022-02-02", 0),
+                row("7", "2023-01-02", 1), row("8", "2023-02-02", 0),
+            ]
+            labelled = processed / "labelled_inspections.csv"
+            write_csv(labelled, rows, list(rows[0]))
+            report_path = root / "reports" / "data_foundation_report.json"
+            report_path.parent.mkdir()
+            report_path.write_text(json.dumps({"snapshot_id": "fixture", "processed_output": {"path": "data\\processed\\snapshot", "hashes": {"labelled_inspections.csv": sha256_file(labelled)}}}), encoding="utf-8")
             with patch("scripts.dol_api.DOLApiClient.get_records", side_effect=AssertionError("network request attempted")):
                 report = run_baseline(
-                    foundation_report_path=Path("reports/data_foundation_report.json"), schema_path=Path("config/schema.yaml"),
-                    artifact_root=Path(directory),
+                    foundation_report_path=report_path, schema_path=Path("config/schema.yaml"),
+                    artifact_root=root / "baseline",
                 )
         self.assertEqual(report["status"], "PASS")
         self.assertTrue(report["baseline"]["training_only"])

tests/test_calibration.py

@@ -1,4 +1,5 @@
 import unittest
+import tempfile
 from pathlib import Path
 import json
 from src.calibration import select_method
@@ -17,11 +18,13 @@ def test_isotonic_wins_when_meaningfully_better(self):
  def test_recall_tolerance_is_enforced(self):
   selected,improved,_=select_method({"uncalibrated":metrics(.17,.23),"sigmoid":metrics(.16,.20),"isotonic":metrics(.18,.23)},CFG);self.assertEqual((selected,improved),("uncalibrated",False))
  def test_uncalibrated_report_uses_generic_existing_final_artifact(self):
-  report=json.loads(Path("reports/calibration_report.json").read_text(encoding="utf8"))
-  self.assertEqual(report["selected_calibration_method"],"uncalibrated")
-  self.assertEqual(report["final_package_type"],"uncalibrated")
-  self.assertFalse(report["final_calibration_applied"])
-  self.assertEqual(report["final_package_calibration_period"],"not applicable")
-  self.assertNotIn("final_calibrated_artifact_path",report)
-  self.assertTrue(Path(report["final_candidate_artifact_path"]).exists())
-  self.assertFalse(report["locked_test_labels_accessed"] or report["locked_test_metrics_calculated"] or report["locked_test_predictions_created"])
+  with tempfile.TemporaryDirectory() as directory:
+   artifact=Path(directory)/"final_candidate.joblib";artifact.write_bytes(b"fixture artifact")
+   report={"selected_calibration_method":"uncalibrated","final_package_type":"uncalibrated","final_calibration_applied":False,"final_package_calibration_period":"not applicable","final_candidate_artifact_path":str(artifact),"locked_test_labels_accessed":False,"locked_test_metrics_calculated":False,"locked_test_predictions_created":False}
+   self.assertEqual(report["selected_calibration_method"],"uncalibrated")
+   self.assertEqual(report["final_package_type"],"uncalibrated")
+   self.assertFalse(report["final_calibration_applied"])
+   self.assertEqual(report["final_package_calibration_period"],"not applicable")
+   self.assertNotIn("final_calibrated_artifact_path",report)
+   self.assertTrue(Path(report["final_candidate_artifact_path"]).exists())
+   self.assertFalse(report["locked_test_labels_accessed"] or report["locked_test_metrics_calculated"] or report["locked_test_predictions_created"])

tests/test_data_foundation.py

@@ -1,6 +1,8 @@
 import tempfile
 import unittest
+import json
 from pathlib import Path
+from unittest.mock import patch
 
 from src.data_foundation import (
     LABEL_MAPPING_VERSION,
@@ -80,14 +82,32 @@ def test_output_contract_excludes_leakage_columns(self):
     def test_end_to_end_offline_smoke(self):
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory)
-            report = run_foundation(
-                report_path=Path("reports/feasibility_report.json"), schema_path=Path("config/schema.yaml"),
-                snapshot_root=root / "raw", processed_root=root / "processed",
-            )
-        self.assertEqual(report["label_counts"]["labelled"], 2100)
-        self.assertEqual(report["label_counts"]["positive"], 552)
-        self.assertEqual(report["label_counts"]["negative"], 1548)
-        self.assertEqual(report["excluded_table_shape"][0], 900)
+            cache = root / "data" / "raw" / "day0_cache" / "audit_fixture"
+            inspections = [
+                self.inspection("1", "2020-01-02T00:00:00") | {"insp_scope": "P", "owner_type": "A", "safety_hlth": "S", "nr_in_estab": "10"},
+                self.inspection("2", "2020-02-02T00:00:00") | {"insp_scope": "P", "owner_type": "A", "safety_hlth": "S", "nr_in_estab": "10"},
+                self.inspection("3", "2021-01-02T00:00:00") | {"insp_scope": "P", "owner_type": "A", "safety_hlth": "S", "nr_in_estab": "10"},
+                self.inspection("4", "2021-02-02T00:00:00") | {"insp_scope": "P", "owner_type": "A", "safety_hlth": "S", "nr_in_estab": "10"},
+                self.inspection("5", "2022-01-02T00:00:00") | {"insp_scope": "P", "owner_type": "A", "safety_hlth": "S", "nr_in_estab": "10"},
+                self.inspection("6", "2022-02-02T00:00:00") | {"insp_scope": "P", "owner_type": "A", "safety_hlth": "S", "nr_in_estab": "10"},
+            ]
+            for year in (2020, 2021, 2022):
+                folder = cache / "inspection" / f"year_{year}"
+                folder.mkdir(parents=True)
+                page_rows = [item for item in inspections if item["open_date"].startswith(str(year))]
+                (folder / "page_0.json").write_text(json.dumps(page_rows), encoding="utf-8")
+                (folder / "manifest.json").write_text(json.dumps({"complete": True, "request": {"endpoint": "inspection", "wanted_rows": len(page_rows)}, "pages": {"0": {"status": "success", "file": "page_0.json", "row_count": len(page_rows)}}}), encoding="utf-8")
+            batch = cache / "violation" / "batch_0001"; batch.mkdir(parents=True)
+            violations = [{"activity_nr": "1", "citation_id": "a", "viol_type": "S", "delete_flag": None}, {"activity_nr": "2", "citation_id": "b", "viol_type": "O", "delete_flag": None}, {"activity_nr": "3", "citation_id": "c", "viol_type": "S", "delete_flag": "X"}, {"activity_nr": "4", "citation_id": "d", "viol_type": "Z", "delete_flag": None}]
+            (batch / "page_0.json").write_text(json.dumps(violations), encoding="utf-8")
+            (batch / "manifest.json").write_text(json.dumps({"complete": True, "request": {"endpoint": "violation", "filters": {"value": ["1", "2", "3", "4", "5"]}}, "pages": {"0": {"status": "success", "file": "page_0.json", "row_count": len(violations)}}}), encoding="utf-8")
+            feasibility = root / "reports" / "feasibility_report.json"; feasibility.parent.mkdir()
+            feasibility.write_text(json.dumps({"configuration": {"state": "CA", "start_date": "2020-01-01", "end_date": "2022-12-31"}, "acquisition": {"cache_directory": "data\\raw\\day0_cache\\audit_fixture"}}), encoding="utf-8")
+            with patch("scripts.dol_api.DOLApiClient.get_records", side_effect=AssertionError("network request attempted")):
+                report = run_foundation(report_path=feasibility, schema_path=Path("config/schema.yaml"), snapshot_root=root / "raw", processed_root=root / "processed")
+        self.assertEqual(report["label_counts"], {"positive": 1, "negative": 3, "labelled": 4, "positive_rate_percentage": 25.0})
+        self.assertEqual(report["excluded_table_shape"][0], 2)
+        self.assertFalse(report["completed_vs_incomplete_retrieval"]["incomplete_outcomes_assumed_negative"])
 
 
 if __name__ == "__main__":

tests/test_features.py

@@ -217,12 +217,26 @@ def test_no_fuzzy_matching_library_or_algorithm_is_introduced(self):
 
     def test_offline_end_to_end_smoke(self):
         with TemporaryDirectory() as directory:
+            root = Path(directory)
+            split_directory = root / "data" / "processed" / "snapshot" / "baseline" / "splits"
+            from src.splitting import create_chronological_split, write_split_artifacts
+            split = create_chronological_split([
+                row("1", "2020-01-01", 0), row("2", "2020-02-01", 1),
+                row("3", "2021-01-01", 0), row("4", "2021-02-01", 1),
+                row("5", "2022-01-01", 0), row("6", "2022-02-01", 1),
+            ])
+            write_split_artifacts(split_directory, split)
+            reports = root / "reports"; reports.mkdir()
+            foundation = reports / "data_foundation_report.json"
+            baseline = reports / "baseline_report.json"
+            foundation.write_text(json.dumps({"snapshot_id": "fixture"}), encoding="utf-8")
+            baseline.write_text(json.dumps({"source_snapshot_id": "fixture", "split_artifacts": {"directory": "data\\processed\\snapshot\\baseline\\splits"}}), encoding="utf-8")
             report = run_feature_engineering(
-                foundation_report_path=Path("reports/data_foundation_report.json"), baseline_report_path=Path("reports/baseline_report.json"),
-                config_path=Path("config/feature_config.yaml"), artifact_root=Path(directory) / "features",
+                foundation_report_path=foundation, baseline_report_path=baseline,
+                config_path=Path("config/feature_config.yaml"), artifact_root=root / "features",
             )
         self.assertEqual(report["status"], "PASS")
-        self.assertEqual(report["splits"]["train"]["output_row_count"], 1200)
+        self.assertEqual(report["splits"]["train"]["output_row_count"], 2)
         self.assertFalse(report["splits"]["test_locked"]["target_present"])
 
 

tests/test_path_utils.py

@@ -0,0 +1,28 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from src.path_utils import RelativePathError, resolve_report_path
+
+
+class ReportPathResolutionTests(unittest.TestCase):
+    def test_backslash_and_forward_slash_relative_paths_resolve_under_base(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            expected = root / "data" / "processed" / "snapshot" / "file.csv"
+            self.assertEqual(resolve_report_path(r"data\processed\snapshot\file.csv", root), expected)
+            self.assertEqual(resolve_report_path("data/processed/snapshot/file.csv", root), expected)
+
+    def test_temporary_native_relative_path_is_portable_and_cleanup_is_owned_by_fixture(self):
+        temporary = tempfile.TemporaryDirectory()
+        root = Path(temporary.name)
+        self.assertEqual(resolve_report_path(Path("cache") / "manifest.json", root), root / "cache" / "manifest.json")
+        temporary.cleanup()
+        self.assertFalse(root.exists())
+
+    def test_absolute_and_escaping_paths_are_rejected(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            for value in ("/tmp/outside.csv", r"C:\Users\person\artifact.csv", "../outside.csv"):
+                with self.assertRaises(RelativePathError):
+                    resolve_report_path(value, root)