Setting the file. One moment. resources/verify_step6_completion.py
resources/verify_step6_completion.py
Python·139 lines·4 KB
15
ENTRY_REQUIRED_FILES
=
(
"evaluations.jsonl"
,)
16DATASET_ANALYSIS_FILES = ("analysis.md", "analysis-summary.md")
17ROOT_ANALYSIS_FILES = ("action-plan.md", "action-plan-summary.md", "meta.json")
18
19
20def _dataset_dirs(results_dir: Path) -> list[Path]:
21 return sorted(
22 path
23 for path in results_dir.iterdir()
24 if path.is_dir() and path.name.startswith("dataset-")
25 )
26
27
28def _entry_dirs(dataset_dir: Path) -> list[Path]:
29 return sorted(
30 path
31 for path in dataset_dir.iterdir()
32 if path.is_dir() and path.name.startswith("entry-")
33 )
34
35
36def _read_jsonl(path: Path, errors: list[str]) -> list[dict[str, object]]:
37 rows: list[dict[str, object]] = []
38 try:
39 for index, line in enumerate(
40 path.read_text(encoding="utf-8").splitlines(), start=1
41 ):
42 if not line.strip():
43 continue
44 obj = json.loads(line)
45 if not isinstance(obj, dict):
46 errors.append(f"{path}: line {index} is not a JSON object")
47 continue
48 rows.append(obj)
49 except OSError as exc:
50 errors.append(f"{path}: could not read file ({exc})")
51 except json.JSONDecodeError as exc:
52 errors.append(f"{path}: invalid JSONL ({exc})")
53 return rows
54
55
56def validate_results_dir(results_dir: Path) -> list[str]:
57 """Return a list of validation errors for a pixie results directory."""
58 errors: list[str] = []
59
60 if not results_dir.is_dir():
61 return [f"{results_dir}: results directory not found"]
62
63 for file_name in ROOT_ANALYSIS_FILES:
64 if not (results_dir / file_name).is_file():
65 errors.append(f"Missing root artifact: {results_dir / file_name}")
66
67 datasets = _dataset_dirs(results_dir)
68 if not datasets:
69 errors.append(f"{results_dir}: no dataset-* directories found")
70 return errors
71
72 for dataset_dir in datasets:
73 for file_name in DATASET_ANALYSIS_FILES:
74 if not (dataset_dir / file_name).is_file():
75 errors.append(f"Missing dataset artifact: {dataset_dir / file_name}")
76
77 entry_dirs = _entry_dirs(dataset_dir)
78 if not entry_dirs:
79 errors.append(f"{dataset_dir}: no entry-* directories found")
80 continue
81
82 for entry_dir in entry_dirs:
83 for file_name in ENTRY_REQUIRED_FILES:
84 if not (entry_dir / file_name).is_file():
85 errors.append(f"Missing entry artifact: {entry_dir / file_name}")
86
87 evaluations_path = entry_dir / "evaluations.jsonl"
88 if not evaluations_path.is_file():
89 continue
90
91 evaluations = _read_jsonl(evaluations_path, errors)
92 for row in evaluations:
93 status = row.get("status")
94 if status == "pending":
95 errors.append(
96 "Pending evaluation remains: "
97 f"{evaluations_path} ({row.get('evaluator', 'unknown evaluator')})"
98 )
99 continue
100
101 if "score" not in row:
102 errors.append(
103 "Missing score in scored evaluation: "
104 f"{evaluations_path} ({row.get('evaluator', 'unknown evaluator')})"
105 )
106 if "reasoning" not in row:
107 errors.append(
108 "Missing reasoning in scored evaluation: "
109 f"{evaluations_path} ({row.get('evaluator', 'unknown evaluator')})"
110 )
111
112 return errors
113
114
115def main(argv: list[str] | None = None) -> int:
116 """CLI entry point."""
117 parser = argparse.ArgumentParser(
118 description="Validate Step 6 completion for a pixie results directory"
119 )
120 parser.add_argument(
121 "results_dir",
122 type=Path,
123 help="Path to pixie_qa/results/<test_id>",
124 )
125 args = parser.parse_args(argv)
126
127 errors = validate_results_dir(args.results_dir)
128 if errors:
129 print("Step 6 completion check failed:")
130 for error in errors:
131 print(f"- {error}")
132 return 1
133
134 print("Step 6 completion check passed.")
135 return 0
136
137
138if __name__ == "__main__":
139 sys.exit(main())