Skills › Scientific Agent Skills › scholar-evaluation › check_process.py
check_process.py
231 lines · skills/scholar-evaluation/scripts/check_process.py at 5f2bc03
1#!/usr/bin/env python3
2"""Check low-stakes assessment governance and bias-process controls."""
4from __future__ import annotations
6import argparse
7from pathlib import Path
8from typing import Any
10import _common
12SECTION_FIELDS: dict[str, dict[str, tuple[str, ...]]] = {
13 "committee": {
14 "booleans": ("qualified_members_confirmed", "training_completed"),
15 "references": ("accountable_owner_role", "training_record_ref"),
16 },
17 "conflicts": {
18 "booleans": (
19 "disclosure_process_confirmed",
20 "recusal_process_confirmed",
21 ),
22 "references": ("record_ref",),
23 },
24 "appeals": {
25 "booleans": ("process_confirmed",),
26 "references": ("process_ref", "notice_ref"),
27 },
28 "accessibility": {
29 "booleans": (
30 "accommodations_process_confirmed",
31 "accessible_materials_confirmed",
32 ),
33 "references": ("process_ref",),
34 },
35 "data_protection": {
36 "booleans": (
37 "purpose_limitation_confirmed",
38 "data_minimisation_confirmed",
39 "access_controls_confirmed",
40 "retention_schedule_confirmed",
41 "raw_private_documents_excluded_from_tool_outputs",
42 ),
43 "references": ("review_ref",),
44 },
45 "rubric_quality": {
46 "booleans": (
47 "construct_documented",
48 "rubric_provenance_documented",
49 "content_validity_documented",
50 "rater_training_documented",
51 "agreement_reviewed",
52 "inter_rater_reliability_reviewed",
53 "uncertainty_recorded",
54 "missing_and_not_applicable_supported",
55 "evidence_traceability_checked",
56 "weight_sensitivity_reviewed",
57 ),
58 "references": ("quality_record_ref",),
59 },
60 "fairness": {
61 "booleans": (
62 "stakeholder_review_completed",
63 "disciplinary_context_reviewed",
64 "subgroup_bias_review_completed",
65 "protected_attributes_excluded_from_scoring",
66 ),
67 "references": ("bias_review_ref",),
68 },
69 "decision_controls": {
70 "booleans": (
71 "no_automated_decision",
72 "no_person_ranking",
73 "no_decision_recommendation",
74 "qualified_human_accountability",
75 ),
76 "references": (),
77 },
78 "monitoring": {
79 "booleans": (
80 "rater_drift_reviewed",
81 "unintended_consequences_reviewed",
82 "periodic_revision_scheduled",
83 ),
84 "references": ("review_record_ref", "revision_owner_role"),
85 },
86}
89def validate_process(record: Any) -> list[_common.Issue]:
90 issues: list[_common.Issue] = []
91 top_fields = {
92 "schema_version",
93 "process_id",
94 "purpose",
95 "unit_of_assessment",
96 "data_classification",
97 "high_impact_use",
98 *SECTION_FIELDS,
99 }
100 if not _common.exact_keys(record, top_fields, "$", issues):
101 return issues
102 if record.get("schema_version") != _common.SCHEMA_VERSION:
103 issues.append(_common.Issue("SCHEMA_VERSION_UNSUPPORTED", "$.schema_version"))
104 _common.validate_identifier(record.get("process_id"), "$.process_id", issues)
105 if record.get("purpose") != _common.ALLOWED_PURPOSE:
106 issues.append(_common.Issue("PURPOSE_NOT_ALLOWED", "$.purpose"))
107 if record.get("unit_of_assessment") != _common.ALLOWED_UNIT:
108 issues.append(
109 _common.Issue(
110 "UNIT_OF_ASSESSMENT_NOT_ALLOWED", "$.unit_of_assessment"
111 )
112 )
113 if record.get("data_classification") not in _common.ALLOWED_CLASSIFICATIONS:
114 issues.append(
115 _common.Issue(
116 "DATA_CLASSIFICATION_NOT_ALLOWED", "$.data_classification"
117 )
118 )
119 _common.validate_bool(record.get("high_impact_use"), "$.high_impact_use", issues)
120 for section_name, field_groups in SECTION_FIELDS.items():
121 section = record.get(section_name)
122 expected = set(field_groups["booleans"]) | set(field_groups["references"])
123 if not _common.exact_keys(section, expected, f"$.{section_name}", issues):
124 continue
125 for field in field_groups["booleans"]:
126 _common.validate_bool(
127 section.get(field), f"$.{section_name}.{field}", issues
128 )
129 for field in field_groups["references"]:
130 _common.validate_reference(
131 section.get(field),
132 f"$.{section_name}.{field}",
133 issues,
134 allow_empty=True,
135 )
136 return issues
139def check_process(record: dict[str, Any]) -> dict[str, Any]:
140 schema_issues = validate_process(record)
141 if _common.error_issues(schema_issues):
142 return {
143 "schema_version": _common.SCHEMA_VERSION,
144 "report_type": "bias_and_process_check",
145 "status": "invalid",
146 "process_id": record.get("process_id") if isinstance(record, dict) else None,
147 "issues": [issue.as_dict() for issue in schema_issues],
148 }
149 blockers: list[str] = []
150 missing_controls: list[str] = []
151 if record["high_impact_use"]:
152 blockers.append("PROHIBITED_HIGH_IMPACT_USE")
153 if record["purpose"] != _common.ALLOWED_PURPOSE:
154 blockers.append("PROHIBITED_PURPOSE")
155 if record["unit_of_assessment"] != _common.ALLOWED_UNIT:
156 blockers.append("PROHIBITED_UNIT_OF_ASSESSMENT")
157 for field in SECTION_FIELDS["decision_controls"]["booleans"]:
158 if record["decision_controls"][field] is not True:
159 blockers.append(f"DECISION_CONTROL_NOT_CONFIRMED:{field}")
160 for section_name, field_groups in SECTION_FIELDS.items():
161 for field in field_groups["booleans"]:
162 if record[section_name][field] is not True:
163 missing_controls.append(f"{section_name}.{field}")
164 for field in field_groups["references"]:
165 if not record[section_name][field]:
166 missing_controls.append(f"{section_name}.{field}")
167 status = (
168 "blocked"
169 if blockers
170 else "incomplete"
171 if missing_controls
172 else "complete_for_low_stakes_process"
173 )
174 confirmed_count = sum(
175 record[section_name][field] is True
176 for section_name, groups in SECTION_FIELDS.items()
177 for field in groups["booleans"]
178 )
179 boolean_count = sum(
180 len(groups["booleans"]) for groups in SECTION_FIELDS.values()
181 )
182 return {
183 "schema_version": _common.SCHEMA_VERSION,
184 "report_type": "bias_and_process_check",
185 "notice": _common.NOTICE,
186 "status": status,
187 "process_id": record["process_id"],
188 "purpose": record["purpose"],
189 "unit_of_assessment": record["unit_of_assessment"],
190 "confirmed_boolean_controls": confirmed_count,
191 "total_boolean_controls": boolean_count,
192 "blockers": blockers,
193 "missing_controls": sorted(set(missing_controls)),
194 "prohibited_uses": sorted(_common.PROHIBITED_USES),
195 "protected_attribute_data_processed": False,
196 "decision_recommendation_provided": False,
197 "issues": [],
198 }
201def build_parser() -> argparse.ArgumentParser:
202 parser = argparse.ArgumentParser(
203 description=(
204 "Check committee, conflict, appeal, accessibility, data-protection, "
205 "rubric-quality, fairness, and monitoring attestations for a "
206 "low-stakes scholarly-work review process."
207 )
208 )
209 parser.add_argument(
210 "--process", required=True, type=Path, help="Local process-checklist JSON"
211 )
212 parser.add_argument("--output", type=Path, help="Optional local JSON report")
213 parser.add_argument(
214 "--force", action="store_true", help="Replace an existing output JSON file"
215 )
216 return parser
219def main() -> int:
220 args = build_parser().parse_args()
221 try:
222 report = check_process(_common.read_json(args.process))
223 _common.write_json(report, args.output, force=args.force)
224 return 0 if report["status"] == "complete_for_low_stakes_process" else 2
225 except _common.ValidationError as error:
226 _common.write_json(_common.failure_report(error), None)
227 return 2
230if __name__ == "__main__":
231 raise SystemExit(main())