"""Quality gate for conductor/chronology.md. Checks: 1. Needs Review threshold: >30% of rows are Needs Review -> FAIL 2. Status distribution sanity: 0 rows are Completed -> FAIL 3. Summary quality: >20% of summaries contain metadata-field text -> FAIL 4. Per-row evidence: every row must have a non-empty reason -> FAIL Modes: default: informational (exits 0, prints report) --strict: CI gate (exits 1 on any violation) """ from __future__ import annotations from dataclasses import dataclass, field from pathlib import Path import argparse import sys _METADATA_FIELD_PREFIXES = ( "**Priority:**", "**Date:**", "**Initialized:**", "**Track:**", "**Track ID:**", "**Parent umbrella:**", "**Status:**", "**Confidence:**", "**Ancestors:**", ) @dataclass(frozen=True, slots=True) class QualityGateResult: passed: bool exit_code: int violations: list[str] = field(default_factory=list) total_rows: int = 0 needs_review_count: int = 0 completed_count: int = 0 metadata_field_count: int = 0 empty_reason_count: int = 0 def run_quality_gate(rows: list[dict]) -> QualityGateResult: violations: list[str] = [] total: int = len(rows) if total == 0: return QualityGateResult(passed=False, exit_code=1, violations=["0 rows provided"]) needs_review_count: int = sum(1 for r in rows if r.get("status") == "Needs Review") completed_count: int = sum(1 for r in rows if r.get("status") == "Completed") metadata_field_count: int = sum( 1 for r in rows if any(str(r.get("summary", "")).startswith(p) for p in _METADATA_FIELD_PREFIXES) ) empty_reason_count: int = sum(1 for r in rows if not r.get("reason", "").strip()) # Check 1: Needs Review threshold (>30%) if total > 0 and (needs_review_count / total) > 0.30: violations.append( f"Needs Review threshold exceeded: {needs_review_count}/{total} " f"({needs_review_count/total*100:.0f}%) > 30%" ) # Check 2: 0 rows are Completed if completed_count == 0: violations.append("0 rows are Completed (classifier may be misclassifying everything)") # Check 3: Summary quality (>20% contain metadata-field text) if total > 0 and (metadata_field_count / total) > 0.20: violations.append( f"Summary quality: {metadata_field_count}/{total} " f"({metadata_field_count/total*100:.0f}%) contain metadata-field text > 20%" ) # Check 4: Per-row evidence (every row must have non-empty reason) if empty_reason_count > 0: violations.append(f"{empty_reason_count} rows have empty reason (no evidence)") passed: bool = len(violations) == 0 return QualityGateResult( passed=passed, exit_code=0 if passed else 1, violations=violations, total_rows=total, needs_review_count=needs_review_count, completed_count=completed_count, metadata_field_count=metadata_field_count, empty_reason_count=empty_reason_count, ) def _load_rows_from_chronology(chronology_path: Path) -> list[dict]: from scripts.audit.generate_chronology import walk_track_folders, _repo_root # chronology_path is conductor/chronology.md; walk_track_folders expects the conductor dir conductor_dir: Path = chronology_path.parent.resolve() if not (conductor_dir / "tracks").is_dir(): # Fallback: resolve from repo root repo_root: Path = _repo_root(conductor_dir) conductor_dir = repo_root / "conductor" return walk_track_folders(conductor_dir) def main() -> None: parser = argparse.ArgumentParser(description="Quality gate for conductor/chronology.md") parser.add_argument("--chronology", type=Path, default=Path("conductor/chronology.md")) parser.add_argument("--rows-json", type=Path, default=None, help="Load rows from JSON file instead of re-walking") parser.add_argument("--strict", action="store_true", help="Exit 1 on any violation") args = parser.parse_args() if args.rows_json and args.rows_json.is_file(): import json rows = json.loads(args.rows_json.read_text(encoding="utf-8")) else: rows = _load_rows_from_chronology(args.chronology) result = run_quality_gate(rows) print(f"Quality gate: {'PASS' if result.passed else 'FAIL'}") print(f" Total rows: {result.total_rows}") print(f" Needs Review: {result.needs_review_count}") print(f" Completed: {result.completed_count}") print(f" Metadata-field summaries: {result.metadata_field_count}") print(f" Empty reasons: {result.empty_reason_count}") if result.violations: print(" Violations:") for v in result.violations: print(f" - {v}") if args.strict and result.exit_code != 0: sys.exit(result.exit_code) if __name__ == "__main__": main()