126 lines
4.4 KiB
Python
126 lines
4.4 KiB
Python
"""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() |